@@ -217,25 +217,31 @@ func (b *Broker[T]) Publish(t EventType, payload T) {
217217// after timeout — recovery is the subscriber's responsibility (e.g. a
218218// re-fetch on the next session-visible event).
219219func (b *Broker[T]) PublishMustDeliver(ctx context.Context, t EventType, payload T) {
220+ // Hold the read lock across the bounded-blocking sends below. Both
221+ // Shutdown and per-subscriber teardown close subscriber channels under
222+ // the write lock, so holding RLock here is what guarantees a send can
223+ // never race a close. The earlier lock-free snapshot released b.mu
224+ // before sending, which let Shutdown/unsubscribe close a channel
225+ // mid-send — a data race the race detector flags (the recover() that
226+ // used to sit in mustDeliver only masked the resulting panic).
227+ //
228+ // The cost is bounded and paid only under saturation: a stalled
229+ // subscriber can delay Subscribe/Shutdown by at most (number of
230+ // subscribers × mustDeliverTimeout). With this app's handful of
231+ // subscribers that ceiling is a few hundred milliseconds, and only
232+ // when a subscriber's buffer is already full.
220233 b.mu.RLock()
234+ defer b.mu.RUnlock()
235+
221236 select {
222237 case <-b.done:
223− b.mu.RUnlock()
224238 return
225239 default:
226240 }
241+
227242 event := Event[T]{Type: t, Payload: payload}
228243 timeout := b.mustDeliverTimeout
229− // Snapshot the subscribers so the bounded-blocking sends below run
230− // without holding b.mu. Holding the read lock across a blocking send
231− // would stall Subscribe/Shutdown for up to N×timeout under saturation.
232− subs := make([]chan Event[T], 0, len(b.subs))
233244 for sub := range b.subs {
234− subs = append(subs, sub)
235− }
236− b.mu.RUnlock()
237−
238− for _, sub := range subs {
239245 if b.mustDeliver(ctx, sub, event, timeout) {
240246 // Context cancelled; stop delivering to the remaining subs.
241247 return
@@ -245,12 +251,10 @@ func (b *Broker[T]) PublishMustDeliver(ctx context.Context, t EventType, payload
245251
246252// mustDeliver performs one bounded-blocking send. It reports whether the
247253// context was cancelled (so the caller stops delivering to remaining
248−// subscribers). Because sends happen without holding b.mu, the channel may
249−// be closed concurrently by Shutdown or an unsubscribe; sending on a closed
250−// channel panics, so we recover and treat that subscriber as gone.
254+// subscribers). The caller holds b.mu.RLock, and every subscriber-channel
255+// close happens under b.mu's write lock, so the channel cannot be closed
256+// concurrently with this send.
251257func (b *Broker[T]) mustDeliver(ctx context.Context, sub chan Event[T], event Event[T], timeout time.Duration) (ctxDone bool) {
252− defer func() { _ = recover() }()
253−
254258 // Fast path: non-blocking send.
255259 select {
256260 case sub <- event: