@@ -24,21 +24,15 @@ import (
2424// synctest gives us a fake clock, so the one-second warning timeout fires
2525// instantly and deterministically instead of relying on real sleeps.
2626func TestWildcardSlowConsumerNoDeadlock (t * testing.T ) {
27- // Capture log output to confirm the slow-consumer path actually ran.
28- defer func (orig * slog.Logger ) { log = orig }(log )
29- logs := & mockLogger {}
30- log = slog .New (slog .NewTextHandler (logs , nil ))
31-
3227 synctest .Test (t , func (t * testing.T ) {
3328 const emitters = 3
3429
35- bus := NewBus ()
30+ // Capture log output to confirm the slow-consumer path actually ran.
31+ logs := & mockLogger {}
32+ bus := NewBus (withLogger (slog .New (slog .NewTextHandler (logs , nil ))))
3633 sub , err := bus .Subscribe (event .WildcardSubscription , BufSize (1 ))
3734 require .NoError (t , err )
38- // The wildcard subscription is intentionally left open: wildcardSub.Close
39- // starts a drain goroutine that only returns once the subscription channel
40- // closes, which never happens for wildcard subs. Closing it here would
41- // leave that goroutine running and trip synctest's end-of-test check.
35+ defer sub .Close ()
4236
4337 em , err := bus .Emitter (new (EventB ))
4438 require .NoError (t , err )
@@ -95,3 +89,105 @@ func TestWildcardSlowConsumerNoDeadlock(t *testing.T) {
9589 }
9690 })
9791}
92+
93+ // TestWildcardSubCloseReleasesDrainGoroutine checks that closing a wildcard
94+ // subscription does not leak the drain goroutine started by removeSink. If the
95+ // drainer is still blocked when the bubble ends, synctest reports a deadlock.
96+ func TestWildcardSubCloseReleasesDrainGoroutine (t * testing.T ) {
97+ synctest .Test (t , func (t * testing.T ) {
98+ bus := NewBus ()
99+ sub , err := bus .Subscribe (event .WildcardSubscription )
100+ require .NoError (t , err )
101+ require .NoError (t , sub .Close ())
102+ })
103+ }
104+
105+ // TestWildcardCloseUnblocksStalledEmit covers the slow-consumer safety valve: an
106+ // emit stalled on a full wildcard subscriber, holding the node read lock, must
107+ // be released when the subscription is closed. wildcardNode.removeSink starts
108+ // draining the channel before it takes the write lock; were the order reversed,
109+ // Close would deadlock against the read lock the stalled emit still holds.
110+ func TestWildcardCloseUnblocksStalledEmit (t * testing.T ) {
111+ synctest .Test (t , func (t * testing.T ) {
112+ // The stalled emit warns as part of this test; keep it out of the output.
113+ bus := NewBus ()
114+
115+ // An unbuffered subscriber has nowhere to put the event, so the emit below
116+ // stalls immediately without a fill step.
117+ sub , err := bus .Subscribe (event .WildcardSubscription , BufSize (0 ))
118+ require .NoError (t , err )
119+
120+ em , err := bus .Emitter (new (EventB ))
121+ require .NoError (t , err )
122+ defer em .Close ()
123+
124+ emitReturned := make (chan struct {})
125+ go func () {
126+ em .Emit (EventB (0 ))
127+ close (emitReturned )
128+ }()
129+
130+ // Sleep past the warning timeout: the emit stalls, warns, and is now in
131+ // the deepest stall state — a bare send with no timeout escape, still
132+ // holding the node read lock.
133+ time .Sleep (slowConsumerWarningTimeout + time .Millisecond )
134+ synctest .Wait ()
135+
136+ // Close must free the stalled emit; synctest reports a deadlock if not.
137+ require .NoError (t , sub .Close ())
138+ <- emitReturned
139+ })
140+ }
141+
142+ func TestEmitLogsErrorOnStall (t * testing.T ) {
143+ synctest .Test (t , func (t * testing.T ) {
144+ ml := mockLogger {}
145+ logger := slog .New (slog .NewTextHandler (& ml , nil ))
146+
147+ bus1 := NewBus (withLogger (logger ))
148+ bus2 := NewBus (withLogger (logger ))
149+
150+ eventSub , err := bus1 .Subscribe (new (EventA ))
151+ if err != nil {
152+ t .Fatal (err )
153+ }
154+
155+ wildcardSub , err := bus2 .Subscribe (event .WildcardSubscription )
156+ if err != nil {
157+ t .Fatal (err )
158+ }
159+
160+ testCases := []event.Subscription {eventSub , wildcardSub }
161+ eventBuses := []event.Bus {bus1 , bus2 }
162+
163+ for i , sub := range testCases {
164+ bus := eventBuses [i ]
165+ em , err := bus .Emitter (new (EventA ))
166+ if err != nil {
167+ t .Fatal (err )
168+ }
169+ defer em .Close ()
170+
171+ go func () {
172+ for i := 0 ; i < subSettingsDefault .buffer + 2 ; i ++ {
173+ em .Emit (EventA {})
174+ }
175+ }()
176+
177+ time .Sleep (3 * time .Second )
178+ logs := ml .Logs ()
179+ found := false
180+ for _ , log := range logs {
181+ if strings .Contains (log , "slow consumer" ) {
182+ found = true
183+ break
184+ }
185+ }
186+ require .True (t , found , "expected to find slow consumer log" )
187+ ml .Clear ()
188+
189+ // Close the subscriber so the worker can finish.
190+ sub .Close ()
191+ }
192+ })
193+ }
0 commit comments