dc0d982 · crush · git.iko.soy

git.iko.soy/crush

on main help sign in

fix(pubsub): eliminate broker close-vs-send data race

dc0d982iko3 months ago

internal/pubsub/broker.go+19−15

compare with main

unifiedsplit

@@ -217,25 +217,31 @@ func (b *Broker[T]) Publish(t EventType, payload T) {

217// after timeout — recovery is the subscriber's responsibility (e.g. a

218// re-fetch on the next session-visible event).

219func (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.

220 b.mu.RLock()

233 b.mu.RLock()

234+ defer b.mu.RUnlock()

235+

221 select {

236 select {

222 case <-b.done:

237 case <-b.done:

223− b.mu.RUnlock()

224 return

238 return

225 default:

239 default:

226 }

240 }

241+

227 event := Event[T]{Type: t, Payload: payload}

242 event := Event[T]{Type: t, Payload: payload}

228 timeout := b.mustDeliverTimeout

243 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))

233 for sub := range b.subs {

244 for sub := range b.subs {

234− subs = append(subs, sub)

235− }

236− b.mu.RUnlock()

237−

238− for _, sub := range subs {

239 if b.mustDeliver(ctx, sub, event, timeout) {

245 if b.mustDeliver(ctx, sub, event, timeout) {

240 // Context cancelled; stop delivering to the remaining subs.

246 // Context cancelled; stop delivering to the remaining subs.

241 return

247 return

@@ -245,12 +251,10 @@ func (b *Broker[T]) PublishMustDeliver(ctx context.Context, t EventType, payload

245

251

246// mustDeliver performs one bounded-blocking send. It reports whether the

252// mustDeliver performs one bounded-blocking send. It reports whether the

247// context was cancelled (so the caller stops delivering to remaining

253// context was cancelled (so the caller stops delivering to remaining

248−// subscribers). Because sends happen without holding b.mu, the channel may

254+// subscribers). The caller holds b.mu.RLock, and every subscriber-channel

249−// be closed concurrently by Shutdown or an unsubscribe; sending on a closed

255+// close happens under b.mu's write lock, so the channel cannot be closed

250−// channel panics, so we recover and treat that subscriber as gone.

256+// concurrently with this send.

251func (b *Broker[T]) mustDeliver(ctx context.Context, sub chan Event[T], event Event[T], timeout time.Duration) (ctxDone bool) {

257func (b *Broker[T]) mustDeliver(ctx context.Context, sub chan Event[T], event Event[T], timeout time.Duration) (ctxDone bool) {

252− defer func() { _ = recover() }()

253−

254 // Fast path: non-blocking send.

258 // Fast path: non-blocking send.

255 select {

259 select {

256 case sub <- event:

260 case sub <- event:

internal/pubsub/broker_test.go+35

@@ -57,6 +57,41 @@ func TestPublishMustDeliverDuringShutdown(t *testing.T) {

57 wg.Wait()

58}

60+// TestPublishMustDeliverDuringUnsubscribe ensures a send can't race a

61+// per-subscriber teardown: each subscriber is cancelled individually while

62+// PublishMustDeliver is delivering to it. Run under -race, this fails if the

63+// broker ever sends on a channel a teardown goroutine is concurrently closing.

64+func TestPublishMustDeliverDuringUnsubscribe(t *testing.T) {

65+ t.Parallel()

66+ b := NewBrokerWithOptions[int](1)

67+ b.SetMustDeliverTimeout(time.Millisecond)

68+ defer b.Shutdown()

69+

70+ var wg sync.WaitGroup

71+ wg.Add(1)

72+ go func() {

73+ defer wg.Done()

74+ for range 200 {

75+ b.PublishMustDeliver(context.Background(), CreatedEvent, 1)

76+ }

77+ }()

78+

79+ for range 8 {

80+ wg.Add(1)

81+ go func() {

82+ defer wg.Done()

83+ // Per-subscriber context: cancelling it tears the subscriber

84+ // down (removing + closing its channel) concurrently with the

85+ // publisher above. Never drained, so its buffer saturates and

86+ // sends block, widening the race window.

87+ ctx, cancel := context.WithCancel(context.Background())

88+ b.Subscribe(ctx)

89+ cancel()

90+ }()

91+ }

92+ wg.Wait()

93+}

94+

60// TestSubscribeGoroutineExitsOnShutdown ensures the per-subscriber goroutine

95// TestSubscribeGoroutineExitsOnShutdown ensures the per-subscriber goroutine

61// unblocks on Shutdown even when its context outlives the broker.

96// unblocks on Shutdown even when its context outlives the broker.

62func TestSubscribeGoroutineExitsOnShutdown(t *testing.T) {

97func TestSubscribeGoroutineExitsOnShutdown(t *testing.T) {