git.iko.soy/crush
on main help sign in
perf: timer leak in setupSubscriber (#2147)
0491b67Carlos Alexandro Becker
go.mod+1
compare with main
unifiedsplit
@@ -60,6 +60,7 @@ require (
60 github.com/tidwall/gjson v1.18.0
60 github.com/tidwall/gjson v1.18.0
61 github.com/tidwall/sjson v1.2.5
61 github.com/tidwall/sjson v1.2.5
62 github.com/zeebo/xxh3 v1.1.0
62 github.com/zeebo/xxh3 v1.1.0
63+ go.uber.org/goleak v1.3.0
63 golang.org/x/net v0.49.0
64 golang.org/x/net v0.49.0
64 golang.org/x/sync v0.19.0
65 golang.org/x/sync v0.19.0
65 golang.org/x/text v0.33.0
66 golang.org/x/text v0.33.0
go.sum+2
@@ -375,6 +375,8 @@ go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFh
375go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps=
375go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps=
376go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
376go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
377go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
377go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
378+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
379+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
378go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
380go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
379go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
381go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
380go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
382go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
internal/app/app.go+15−1
@@ -421,6 +421,8 @@ func (app *App) setupEvents() {
421 app.cleanupFuncs = append(app.cleanupFuncs, cleanupFunc)
421 app.cleanupFuncs = append(app.cleanupFuncs, cleanupFunc)
422}
422}
423
423
424+const subscriberSendTimeout = 2 * time.Second
425+
424func setupSubscriber[T any](
426func setupSubscriber[T any](
425 ctx context.Context,
427 ctx context.Context,
426 wg *sync.WaitGroup,
428 wg *sync.WaitGroup,
@@ -430,6 +432,10 @@ func setupSubscriber[T any](
430) {
432) {
431 wg.Go(func() {
433 wg.Go(func() {
432 subCh := subscriber(ctx)
434 subCh := subscriber(ctx)
435+ sendTimer := time.NewTimer(0)
436+ <-sendTimer.C
437+ defer sendTimer.Stop()
438+
433 for {
439 for {
434 select {
440 select {
435 case event, ok := <-subCh:
441 case event, ok := <-subCh:
@@ -438,9 +444,17 @@ func setupSubscriber[T any](
438 return
444 return
439 }
445 }
440 var msg tea.Msg = event
446 var msg tea.Msg = event
447+ if !sendTimer.Stop() {
448+ select {
449+ case <-sendTimer.C:
450+ default:
451+ }
452+ }
453+ sendTimer.Reset(subscriberSendTimeout)
454+
441 select {
455 select {
442 case outputCh <- msg:
456 case outputCh <- msg:
443− case <-time.After(2 * time.Second):
457+ case <-sendTimer.C:
444 slog.Debug("Message dropped due to slow consumer", "name", name)
458 slog.Debug("Message dropped due to slow consumer", "name", name)
445 case <-ctx.Done():
459 case <-ctx.Done():
446 slog.Debug("Subscription cancelled", "name", name)
460 slog.Debug("Subscription cancelled", "name", name)
internal/app/app_test.gonew file+157
@@ -0,0 +1,157 @@
1+package app
2+
3+import (
4+ "context"
5+ "sync"
6+ "testing"
7+ "testing/synctest"
8+ "time"
9+
10+ tea "charm.land/bubbletea/v2"
11+ "github.com/charmbracelet/crush/internal/pubsub"
12+ "github.com/stretchr/testify/require"
13+ "go.uber.org/goleak"
14+)
15+
16+func TestSetupSubscriber_NormalFlow(t *testing.T) {
17+ synctest.Test(t, func(t *testing.T) {
18+ f := newSubscriberFixture(t, 10)
19+
20+ time.Sleep(10 * time.Millisecond)
21+ synctest.Wait()
22+
23+ f.broker.Publish(pubsub.CreatedEvent, "event1")
24+ f.broker.Publish(pubsub.CreatedEvent, "event2")
25+
26+ for range 2 {
27+ select {
28+ case <-f.outputCh:
29+ case <-time.After(5 * time.Second):
30+ t.Fatal("Timed out waiting for messages")
31+ }
32+ }
33+
34+ f.cancel()
35+ f.wg.Wait()
36+ })
37+}
38+
39+func TestSetupSubscriber_SlowConsumer(t *testing.T) {
40+ synctest.Test(t, func(t *testing.T) {
41+ f := newSubscriberFixture(t, 0)
42+
43+ const numEvents = 5
44+
45+ var pubWg sync.WaitGroup
46+ pubWg.Go(func() {
47+ for range numEvents {
48+ f.broker.Publish(pubsub.CreatedEvent, "event")
49+ time.Sleep(10 * time.Millisecond)
50+ synctest.Wait()
51+ }
52+ })
53+
54+ time.Sleep(time.Duration(numEvents) * (subscriberSendTimeout + 20*time.Millisecond))
55+ synctest.Wait()
56+
57+ received := 0
58+ for {
59+ select {
60+ case <-f.outputCh:
61+ received++
62+ default:
63+ pubWg.Wait()
64+ f.cancel()
65+ f.wg.Wait()
66+ require.Less(t, received, numEvents, "Slow consumer should have dropped some messages")
67+ return
68+ }
69+ }
70+ })
71+}
72+
73+func TestSetupSubscriber_ContextCancellation(t *testing.T) {
74+ synctest.Test(t, func(t *testing.T) {
75+ f := newSubscriberFixture(t, 10)
76+
77+ f.broker.Publish(pubsub.CreatedEvent, "event1")
78+ time.Sleep(100 * time.Millisecond)
79+ synctest.Wait()
80+
81+ f.cancel()
82+ f.wg.Wait()
83+ })
84+}
85+
86+func TestSetupSubscriber_DrainAfterDrop(t *testing.T) {
87+ synctest.Test(t, func(t *testing.T) {
88+ f := newSubscriberFixture(t, 0)
89+
90+ time.Sleep(10 * time.Millisecond)
91+ synctest.Wait()
92+
93+ // First event: nobody reads outputCh so the timer fires (message dropped).
94+ f.broker.Publish(pubsub.CreatedEvent, "event1")
95+ time.Sleep(subscriberSendTimeout + 25*time.Millisecond)
96+ synctest.Wait()
97+
98+ // Second event: triggers Stop()==false path; without the fix this deadlocks.
99+ f.broker.Publish(pubsub.CreatedEvent, "event2")
100+
101+ // If the timer drain deadlocks, wg.Wait never returns.
102+ done := make(chan struct{})
103+ go func() {
104+ f.cancel()
105+ f.wg.Wait()
106+ close(done)
107+ }()
108+
109+ select {
110+ case <-done:
111+ case <-time.After(5 * time.Second):
112+ t.Fatal("setupSubscriber goroutine hung — likely timer drain deadlock")
113+ }
114+ })
115+}
116+
117+func TestSetupSubscriber_NoTimerLeak(t *testing.T) {
118+ defer goleak.VerifyNone(t)
119+ synctest.Test(t, func(t *testing.T) {
120+ f := newSubscriberFixture(t, 100)
121+
122+ for range 100 {
123+ f.broker.Publish(pubsub.CreatedEvent, "event")
124+ time.Sleep(5 * time.Millisecond)
125+ synctest.Wait()
126+ }
127+
128+ f.cancel()
129+ f.wg.Wait()
130+ })
131+}
132+
133+type subscriberFixture struct {
134+ broker *pubsub.Broker[string]
135+ wg sync.WaitGroup
136+ outputCh chan tea.Msg
137+ cancel context.CancelFunc
138+}
139+
140+func newSubscriberFixture(t *testing.T, bufSize int) *subscriberFixture {
141+ t.Helper()
142+ ctx, cancel := context.WithCancel(t.Context())
143+ t.Cleanup(cancel)
144+
145+ f := &subscriberFixture{
146+ broker: pubsub.NewBroker[string](),
147+ outputCh: make(chan tea.Msg, bufSize),
148+ cancel: cancel,
149+ }
150+ t.Cleanup(f.broker.Shutdown)
151+
152+ setupSubscriber(ctx, &f.wg, "test", func(ctx context.Context) <-chan pubsub.Event[string] {
153+ return f.broker.Subscribe(ctx)
154+ }, f.outputCh)
155+
156+ return f
157+}