feat(broadcast): replace utils/broadcast with broadcaster/broadcast

[EgeCaner]

User description

What

Rewrites utils/broadcast as broadcaster/broadcast, a bounded fan-out event stream over a
ring buffer, split into two layers: a reusable MPMC ring (broadcaster/broadcast/ring) and a
thin fan-out API over it. The old package is deleted in this PR; it had no callers outside its
own tests.

Overwrite-on-full semantics are unchanged: publishers never block on consumers, and a reader
that falls a full ring behind gets a LaggedError telling it where to resume instead of stale
data.

Nothing consumes the new package yet. Putting it and feed behind one interface, and
migrating the producers and RPC subscriptions, is the next PR.

What changes against utils/broadcast

Separate publisher and subscriber types. Broadcast mints Publisher and Subscribable
value handles instead of being one object that both sends and subscribes — the shape
feed.Feed already has, so the next PR can put both behind one interface.

No Close. Close, the done channel and ErrClosed / ErrFutureSeq /
ErrInvalidSequence are gone; only a Subscription has a lifecycle, ended by Unsubscribe.

No global lock. The old Send held a broadcast-wide mutex:

func (b *Broadcast[T]) Send(msg *T) error {
	// Acquire lock upfront to avoid races around Close.
	b.mu.Lock()

It covered two things. Close woke readers by broadcasting the slot at the current tail, which
is only correct if the tail cannot move under it. And sequence assignment was split — Send
read the tail, stored, and only then incremented — so without the mutex two publishers claim
the same tail and the sequence between them is never published. The package doc conceded the
cost: "producer-side throughput is effectively single-writer (MPMC-safe but not truly parallel
on Send)".

Close is gone, and the claim is now atomic: tail.Add(1) before the store, with Slot.Write
ignoring a store whose sequence is not newer, since a writer can be lapped in between.
Publishers proceed in parallel.

Reading is an iterator. The old Subscription.run did the whole protocol in one loop —
park on a slot, detect an overwrite, build the lag notification, push into the channel, watch
two stop signals. It now lives in its own package behind
RingBuffer.Iterator(ctx) iter.Seq[EventOrLag[T]], with the channel pump on top of it and
SubscribeWith for a transform in between.

Send takes T, not *T. The pointer parameter avoided copying a large T, but the
types we publish are already pointers, which makes it **core.Block: the caller passes the
address of its own pointer, and behind an interface — where the next PR takes it — that address
escapes, so every Send allocates. By value it is one word for a pointer, two for an
interface, and nothing escapes. Send also returns nothing, and EventOrLag accessors are
comma-ok (AsEvent() (T, bool)) instead of the ErrNoEvent / ErrNoLag sentinels.

Benchmarks

Apple M3 Max, GOMAXPROCS=14, Go 1.27.0. One publisher sending as fast as it can, payload
*int, ring capacity 1024, all subscribers draining for the whole run. Go's default duration
per configuration, -count=3, median. All four transports run under one harness, with the old
package restored from main to measure it.

feed is feed.Feed. old is utils/broadcast. new chan is Subscription.Recv().
new iter is ranging the ring's iter.Seq directly.

Each cell is publish rate / delivery rate per subscriber.

subscribers	feed	old	new chan	new iter
1	24.5 M/s / 504 k/s	56.2 M/s / 29 k/s	85.7 M/s / 45 k/s	21.9 M/s / 21.6 M/s
4	3.5 M/s / 1.3 M/s	14.2 M/s / 471 k/s	18.3 M/s / 620 k/s	7.2 M/s / 6.3 M/s
32	228 k/s / 213 k/s	1.8 M/s / 679 k/s	1.1 M/s / 563 k/s	1.2 M/s / 1.1 M/s
128	55 k/s / 55 k/s	1.3 M/s / 210 k/s	593 k/s / 205 k/s	2.0 M/s / 827 k/s
256	29 k/s / 29 k/s	1.1 M/s / 110 k/s	444 k/s / 112 k/s	2.3 M/s / 559 k/s
512	14 k/s / 14 k/s	924 k/s / 59 k/s	291 k/s / 62 k/s	2.4 M/s / 339 k/s
1024	7 k/s / 7 k/s	579 k/s / 32 k/s	175 k/s / 33 k/s	1.8 M/s / 189 k/s

Feed's publisher collapses with subscriber count: Send holds one mutex and touches every
subscriber's channel on every message, so it falls from 24.5 M/s at one subscriber to 7 k/s at
1024. Past 128 subscribers nothing is dropped only because the publisher is too slow for anyone
to fall behind.

Old publishes faster than new chan from 32 subscribers up. Per-subscriber delivery is within
a few percent of it from 128 up and ahead of it at 32. At 1024 subscribers the two deliver the
same in total, 32.6 M/s against 33.5 M/s, with old putting more of the budget into publishing,
so each of its subscribers sees 5.5% of the stream against 18.7%.

Ranging the iterator skips the channel hop and is fast at both ends: 1.8 M/s published, 189 k/s
per subscriber, 193 M/s delivered across all of them.

Multiple publishers

Same harness, publisher count varied. Cells are publish rate / delivery rate per subscriber.

publishers	subscribers	feed	old	new chan	new iter
1	32	220 k/s / 201 k/s	1.6 M/s / 752 k/s	1.1 M/s / 538 k/s	1.3 M/s / 1.1 M/s
2	32	173 k/s / 146 k/s	1.5 M/s / 751 k/s	1.7 M/s / 433 k/s	1.5 M/s / 1.1 M/s
4	32	186 k/s / 170 k/s	1.5 M/s / 759 k/s	6.7 M/s / 30 k/s	2.3 M/s / 1.3 M/s
8	32	169 k/s / 153 k/s	1.5 M/s / 738 k/s	9.1 M/s / 22 k/s	4.0 M/s / 1.1 M/s
1	1024	6 k/s / 6 k/s	1.0 M/s / 30 k/s	194 k/s / 32 k/s	1.6 M/s / 193 k/s
2	1024	5 k/s / 5 k/s	955 k/s / 29 k/s	255 k/s / 30 k/s	1.9 M/s / 175 k/s
4	1024	6 k/s / 6 k/s	894 k/s / 30 k/s	371 k/s / 30 k/s	2.2 M/s / 151 k/s
8	1024	6 k/s / 6 k/s	960 k/s / 29 k/s	567 k/s / 27 k/s	2.5 M/s / 120 k/s

Feed and old are flat in publisher count — 1.6 to 1.5 M/s at 32 subscribers, 1.0 M/s to 960 k/s
at 1024 — which is the serialised Send. The new ring scales with publishers instead, but only
overtakes old in absolute terms at 32 subscribers, and there the extra rate comes straight out
of delivery. At 1024 subscribers old still publishes more and the two deliver the same. This is
the shape of the serialisation, not a throughput win.

---

PR Type

Enhancement, Tests, Other

---

Description

• Move utils/broadcast to broadcaster/broadcast

• Remove global lock for per-slot locks

• Split API into Publisher and Subscribable handles

• Introduce iter.Seq and zero-allocation EventOrLag

---

File Walkthrough

	Relevant files
Enhancement	6 files broadcast.go Introduce high-level Broadcast API wrapping ring buffer    +33/-0    publisher.go Add Publisher type for sending to ring buffer                        +18/-0    event_or_lag.go Add zero-allocation EventOrLag tagged union and LaggedError +85/-0    ring_buffer.go Implement MPMC ring buffer with atomic sequences and iterators +150/-0  slot.go Implement fixed-position Slot with per-slot mutex and condition +80/-0    subscription.go Implement Subscribable and Subscription wrapping ring iterators +74/-0
Tests	6 files broadcast_benchmark_test.go Add benchmarks for high-level broadcast throughput              +123/-0  broadcast_test.go Add comprehensive tests for broadcast semantics                    +422/-0  ring_buffer_benchmark_test.go Add benchmarks for concurrent ring buffer reads and writes +253/-0  ring_buffer_test.go Add unit tests for ring buffer operations and sequences    +664/-0  broadcast_benchmark_test.go Remove old broadcast benchmarks                                                    +0/-240  broadcast_test.go Remove old broadcast tests                                                              +0/-427
Miscellaneous	2 files broadcast.go Remove old broadcast implementation                                            +0/-307  errors.go Remove old broadcast error definitions                                      +0/-34

---

[github-actions]

Claude Code is working…

I'll analyze this and get back to you.

View job run

[github-actions]

Reviewing PR #4085

• Gather context (read diff, CLAUDE.md)
• Review ring/slot.go — synchronization primitives
• Review ring/ring_buffer.go — MPMC logic, iterator, contexts
• Review ring/event_or_lag.go
• Review broadcast.go, publisher.go, subscription.go
• Review tests/benchmarks for coverage and flakiness
• Post findings

View job run

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪

⚠️ Risk level: Low

📂 Priority files broadcaster/broadcast/ring/ring_buffer.go broadcaster/broadcast/ring/slot.go broadcaster/broadcast/subscription.go

🏅 Score: 85

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Goroutine Leak Risk The pump goroutine started in iterToChan only exits when ctx is cancelled (via Unsubscribe) or the underlying ring iterator ends. If a caller stops draining Recv() without calling Unsubscribe, the goroutine blocks forever on the channel-send select waiting for either buffer space or ctx.Done(), leaking the goroutine along with the underlying ring Iterator registration (AfterFunc hook). This is an API-contract requirement (must always call Unsubscribe) rather than a code defect, but it's worth flagging since nothing enforces it and there's no finalizer/timeout safeguard. // iterToChan pumps seq into a channel of the given buffer size from a goroutine; the channel // is closed when seq ends or ctx is done, and a blocked send gives up when ctx is done. func iterToChan[T any](ctx context.Context, seq iter.Seq[T], bufferSize int) <-chan T { out := make(chan T, bufferSize) go func() { defer close(out) for value := range seq { select { case out <- value: case <-ctx.Done(): return } } }() return out }

[codecov]

Codecov Report

❌ Patch coverage is 87.03704% with 14 lines in your changes missing coverage. Please review.
✅ Project coverage is 79.61%. Comparing base (9cc601f) to head (2abc4f5).

Files with missing lines	Patch %	Lines
broadcaster/broadcast/ring/event_or_lag.go	72.00%	7 Missing ⚠️
broadcaster/broadcast/subscription.go	77.77%	4 Missing ⚠️
broadcaster/broadcast/ring/slot.go	88.00%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4085      +/-   ##
==========================================
- Coverage   79.78%   79.61%   -0.17%     
==========================================
  Files         477      481       +4     
  Lines       36663    36670       +7     
==========================================
- Hits        29250    29196      -54     
- Misses       7404     7465      +61     
  Partials        9        9              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Reviewing PR #4085

• Gather context (read diff, CLAUDE.md)
• Review ring/slot.go — synchronization primitives
• Review ring/ring_buffer.go — MPMC logic, iterator, contexts
• Review ring/event_or_lag.go
• Review broadcast.go, publisher.go, subscription.go
• Review tests/benchmarks for coverage and flakiness
• Post findings

View job run

[github-actions]

Standalone PR Review

PR-Agent could not safely update the persistent review. This standalone result will not replace the canonical review.

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

Estimated effort to review: 2 🔵🔵⚪⚪⚪

⚠️ Risk level: Low

📂 Priority files: None

🏅 Score: 95

🧪 PR contains tests

🔒 No security concerns identified

⚡ No major issues detected

[github-actions]

Claude finished @EgeCaner's task in 11m 19s —— View job

---

Reviewing PR #4085

• Gather context (read diff, CLAUDE.md)
• Review ring/slot.go — synchronization primitives
• Review ring/ring_buffer.go — MPMC logic, iterator, contexts
• Review ring/event_or_lag.go
• Review broadcast.go, publisher.go, subscription.go
• Review tests/benchmarks for coverage and flakiness
• Post findings

Summary

Solid design and unusually thorough test suite (races, cancellation, lag resync, zero-allocation guards, synctest-based determinism). The per-slot lock + atomic-tail claim scheme in ring_buffer.go/slot.go is correct: I traced the wakeWaiters proof (comment at ring_buffer.go:81-91) against the cancellation races and it holds — a lapped writer is safely dropped by Slot.Write's seq <= s.seq check, and a cancelled reader is always woken because it can only be parked on a slot that either the tail-broadcast or an in-flight Write will hit.

No blockers or important issues found. One nit posted inline:

• nit — broadcaster/broadcast/subscription.go:29: SubscribeWith has zero test coverage (only Subscribe() is exercised in broadcast_test.go), which lines up with the codecov gaps flagged on this file. Worth a test given it runs a caller-supplied transform inside the pump goroutine.

Also confirmed via grep that no code outside utils/broadcast's own tests referenced the deleted package, and nothing yet imports the new broadcaster/broadcast package, matching the PR description ("nothing consumes the new package yet").

Note: go test/go vet execution wasn't permitted in this sandbox, so this review is based on static reading of the diff plus go build (which succeeded); I did not independently re-run the race/synctest suite.

[github-actions]

Standalone PR Review

PR-Agent could not safely update the persistent review. This standalone result will not replace the canonical review.

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

⚠️ Risk level: Medium

📂 Priority files broadcaster/broadcast/ring/ring_buffer.go broadcaster/broadcast/ring/slot.go

🏅 Score: 95

🧪 PR contains tests

🔒 No security concerns identified

⚡ No major issues detected

[github-actions]

SubscribeWith has no test coverage anywhere in this PR — broadcast_test.go only exercises Subscribe(). It's public API with real behavioral nuance (the transform closure controls how/whether lag notifications survive, and it runs inside the pump goroutine, so a panicking or blocking transform takes the whole subscription down with it). Given this is called out in codecov as one of the coverage gaps in this file, a test subscribing with a transform (e.g. dropping lags, or mapping to a derived type) plus asserting Unsubscribe still terminates the pump goroutine would close the gap and pin the transform-composition behavior.

Fix this →