git.iko.soy/crush
on main help sign in
fix(app): replace single events channel with pubsub.Broker for fan-out (#2663)
4d16e96Tai Groot
internal/app/app.go+16−39
compare with main
unifiedsplit
@@ -66,7 +66,7 @@ type App struct {
66
66
67 serviceEventsWG *sync.WaitGroup
67 serviceEventsWG *sync.WaitGroup
68 eventsCtx context.Context
68 eventsCtx context.Context
69− events chan tea.Msg
69+ events *pubsub.Broker[tea.Msg]
70 tuiWG *sync.WaitGroup
70 tuiWG *sync.WaitGroup
71
71
72 // global context and cleanup functions
72 // global context and cleanup functions
@@ -100,7 +100,7 @@ func New(ctx context.Context, conn *sql.DB, store *config.ConfigStore) (*App, er
100
100
101 config: store,
101 config: store,
102
102
103− events: make(chan tea.Msg, 100),
103+ events: pubsub.NewBroker[tea.Msg](),
104 serviceEventsWG: &sync.WaitGroup{},
104 serviceEventsWG: &sync.WaitGroup{},
105 tuiWG: &sync.WaitGroup{},
105 tuiWG: &sync.WaitGroup{},
106 agentNotifications: pubsub.NewBroker[notify.Notification](),
106 agentNotifications: pubsub.NewBroker[notify.Notification](),
@@ -153,18 +153,15 @@ func (app *App) Store() *config.ConfigStore {
153 return app.config
153 return app.config
154}
154}
155
155
156−// Events returns the events channel for the application.
156+// Events returns a per-caller subscription channel for application events.
157+// Each caller receives its own channel; all callers receive every event.
157−func (app *App) Events() <-chan tea.Msg {
158+func (app *App) Events(ctx context.Context) <-chan pubsub.Event[tea.Msg] {
158− return app.events
159+ return app.events.Subscribe(ctx)
159}
160}
160
161
161−// SendEvent pushes a message into the application's events channel.
162+// SendEvent publishes a message to all event subscribers.
162−// It is non-blocking; the message is dropped if the channel is full.
163func (app *App) SendEvent(msg tea.Msg) {
163func (app *App) SendEvent(msg tea.Msg) {
164− select {
165− case app.events <- msg:
164+ app.events.Publish(pubsub.UpdatedEvent, msg)
166− default:
167− }
168}
165}
169
166
170// AgentNotifications returns the broker for agent notification events.
167// AgentNotifications returns the broker for agent notification events.
@@ -486,26 +483,21 @@ func (app *App) setupEvents() {
486 cleanupFunc := func(context.Context) error {
483 cleanupFunc := func(context.Context) error {
487 cancel()
484 cancel()
488 app.serviceEventsWG.Wait()
485 app.serviceEventsWG.Wait()
486+ app.events.Shutdown()
489 return nil
487 return nil
490 }
488 }
491 app.cleanupFuncs = append(app.cleanupFuncs, cleanupFunc)
489 app.cleanupFuncs = append(app.cleanupFuncs, cleanupFunc)
492}
490}
493
491
494−const subscriberSendTimeout = 2 * time.Second
495−
496func setupSubscriber[T any](
492func setupSubscriber[T any](
497 ctx context.Context,
493 ctx context.Context,
498 wg *sync.WaitGroup,
494 wg *sync.WaitGroup,
499 name string,
495 name string,
500 subscriber func(context.Context) <-chan pubsub.Event[T],
496 subscriber func(context.Context) <-chan pubsub.Event[T],
501− outputCh chan<- tea.Msg,
497+ broker *pubsub.Broker[tea.Msg],
502) {
498) {
503 wg.Go(func() {
499 wg.Go(func() {
504 subCh := subscriber(ctx)
500 subCh := subscriber(ctx)
505− sendTimer := time.NewTimer(0)
506− <-sendTimer.C
507− defer sendTimer.Stop()
508−
509 for {
501 for {
510 select {
502 select {
511 case event, ok := <-subCh:
503 case event, ok := <-subCh:
@@ -513,23 +505,7 @@ func setupSubscriber[T any](
513 slog.Debug("Subscription channel closed", "name", name)
505 slog.Debug("Subscription channel closed", "name", name)
514 return
506 return
515 }
507 }
516− var msg tea.Msg = event
508+ broker.Publish(pubsub.UpdatedEvent, tea.Msg(event))
517− if !sendTimer.Stop() {
518− select {
519− case <-sendTimer.C:
520− default:
521− }
522− }
523− sendTimer.Reset(subscriberSendTimeout)
524−
525− select {
526− case outputCh <- msg:
527− case <-sendTimer.C:
528− slog.Debug("Message dropped due to slow consumer", "name", name)
529− case <-ctx.Done():
530− slog.Debug("Subscription cancelled", "name", name)
531− return
532− }
533 case <-ctx.Done():
509 case <-ctx.Done():
534 slog.Debug("Subscription cancelled", "name", name)
510 slog.Debug("Subscription cancelled", "name", name)
535 return
511 return
@@ -579,17 +555,18 @@ func (app *App) Subscribe(program *tea.Program) {
579 })
555 })
580 defer app.tuiWG.Done()
556 defer app.tuiWG.Done()
581
557
558+ events := app.events.Subscribe(tuiCtx)
582 for {
559 for {
583 select {
560 select {
584 case <-tuiCtx.Done():
561 case <-tuiCtx.Done():
585 slog.Debug("TUI message handler shutting down")
562 slog.Debug("TUI message handler shutting down")
586 return
563 return
587− case msg, ok := <-app.events:
564+ case ev, ok := <-events:
588 if !ok {
565 if !ok {
589 slog.Debug("TUI message channel closed")
566 slog.Debug("TUI message channel closed")
590 return
567 return
591 }
568 }
592− program.Send(msg)
569+ program.Send(ev.Payload)
593 }
570 }
594 }
571 }
595}
572}
@@ -649,9 +626,9 @@ func (app *App) checkForUpdates(ctx context.Context) {
649 if err != nil || !info.Available() {
626 if err != nil || !info.Available() {
650 return
627 return
651 }
628 }
652− app.events <- UpdateAvailableMsg{
629+ app.events.Publish(pubsub.UpdatedEvent, UpdateAvailableMsg{
653 CurrentVersion: info.Current,
630 CurrentVersion: info.Current,
654 LatestVersion: info.Latest,
631 LatestVersion: info.Latest,
655 IsDevelopment: info.IsDevelopment(),
632 IsDevelopment: info.IsDevelopment(),
656− }
633+ })
657}
634}
internal/app/app_test.go+133−107
@@ -2,156 +2,182 @@ package app
2
2
3import (
3import (
4 "context"
4 "context"
5+ "fmt"
5 "sync"
6 "sync"
6 "testing"
7 "testing"
7− "testing/synctest"
8 "time"
8 "time"
9
9
10 tea "charm.land/bubbletea/v2"
10 tea "charm.land/bubbletea/v2"
11 "github.com/charmbracelet/crush/internal/pubsub"
11 "github.com/charmbracelet/crush/internal/pubsub"
12 "github.com/stretchr/testify/require"
12 "github.com/stretchr/testify/require"
13− "go.uber.org/goleak"
14)
13)
15
14
15+// TestSetupSubscriber_NormalFlow verifies that events published to the source
16+// broker are forwarded to the output broker.
16func TestSetupSubscriber_NormalFlow(t *testing.T) {
17func TestSetupSubscriber_NormalFlow(t *testing.T) {
17− synctest.Test(t, func(t *testing.T) {
18+ t.Parallel()
18− f := newSubscriberFixture(t, 10)
19
19
20− time.Sleep(10 * time.Millisecond)
20+ ctx, cancel := context.WithCancel(t.Context())
21− synctest.Wait()
21+ defer cancel()
22
22
23− f.broker.Publish(pubsub.CreatedEvent, "event1")
23+ src := pubsub.NewBroker[string]()
24− f.broker.Publish(pubsub.CreatedEvent, "event2")
24+ defer src.Shutdown()
25+ out := pubsub.NewBroker[tea.Msg]()
26+ defer out.Shutdown()
25
27
28+ ch := out.Subscribe(ctx)
29+
30+ var wg sync.WaitGroup
31+ setupSubscriber(ctx, &wg, "test", src.Subscribe, out)
32+
33+ // Yield so the subscriber goroutine can call src.Subscribe before we publish.
34+ time.Sleep(10 * time.Millisecond)
35+
36+ src.Publish(pubsub.CreatedEvent, "hello")
37+ src.Publish(pubsub.CreatedEvent, "world")
38+
26− for range 2 {
39+ for range 2 {
27− select {
40+ select {
28− case <-f.outputCh:
41+ case <-ch:
29− case <-time.After(5 * time.Second):
42+ case <-time.After(5 * time.Second):
30− t.Fatal("Timed out waiting for messages")
43+ t.Fatal("timed out waiting for forwarded event")
31− }
32 }
44 }
45+ }
33
46
34− f.cancel()
47+ cancel()
35− f.wg.Wait()
48+ wg.Wait()
36− })
37}
49}
38
50
39−func TestSetupSubscriber_SlowConsumer(t *testing.T) {
51+// TestSetupSubscriber_ContextCancellation verifies the goroutine exits cleanly
40− synctest.Test(t, func(t *testing.T) {
52+// when the context is cancelled.
41− f := newSubscriberFixture(t, 0)
53+func TestSetupSubscriber_ContextCancellation(t *testing.T) {
54+ t.Parallel()
42
55
43− const numEvents = 5
56+ ctx, cancel := context.WithCancel(t.Context())
44
57
45− var pubWg sync.WaitGroup
58+ src := pubsub.NewBroker[string]()
46− pubWg.Go(func() {
59+ defer src.Shutdown()
47− for range numEvents {
60+ out := pubsub.NewBroker[tea.Msg]()
48− f.broker.Publish(pubsub.CreatedEvent, "event")
61+ defer out.Shutdown()
49− time.Sleep(10 * time.Millisecond)
50− synctest.Wait()
51− }
52− })
53
62
54− time.Sleep(time.Duration(numEvents) * (subscriberSendTimeout + 20*time.Millisecond))
63+ var wg sync.WaitGroup
55− synctest.Wait()
64+ setupSubscriber(ctx, &wg, "test", src.Subscribe, out)
56−
65+
57− received := 0
66+ src.Publish(pubsub.CreatedEvent, "event")
58− for {
67+ cancel()
68+
69+ done := make(chan struct{})
70+ go func() { wg.Wait(); close(done) }()
71+
59− select {
72+ select {
60− case <-f.outputCh:
73+ case <-done:
61− received++
74+ case <-time.After(5 * time.Second):
62− default:
75+ t.Fatal("setupSubscriber goroutine did not exit after context cancellation")
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− })
76+ }
71}
77}
72
78
73−func TestSetupSubscriber_ContextCancellation(t *testing.T) {
79+// TestEvents_ZeroConsumers verifies that publishing with no subscribers does
74− synctest.Test(t, func(t *testing.T) {
80+// not block or panic.
75− f := newSubscriberFixture(t, 10)
81+func TestEvents_ZeroConsumers(t *testing.T) {
82+ t.Parallel()
83+
84+ broker := pubsub.NewBroker[tea.Msg]()
85+ defer broker.Shutdown()
86+
87+ require.Equal(t, 0, broker.GetSubscriberCount())
76
88
89+ // Must not block.
90+ done := make(chan struct{})
91+ go func() {
77− f.broker.Publish(pubsub.CreatedEvent, "event1")
92+ broker.Publish(pubsub.UpdatedEvent, tea.Msg("msg1"))
78− time.Sleep(100 * time.Millisecond)
93+ broker.Publish(pubsub.UpdatedEvent, tea.Msg("msg2"))
79− synctest.Wait()
94+ close(done)
95+ }()
80
96
81− f.cancel()
97+ select {
82− f.wg.Wait()
98+ case <-done:
99+ case <-time.After(time.Second):
100+ t.Fatal("Publish with zero consumers blocked")
83− })
101+ }
84}
102}
85
103
86−func TestSetupSubscriber_DrainAfterDrop(t *testing.T) {
104+// TestEvents_OneConsumer verifies that a single subscriber receives every event
87− synctest.Test(t, func(t *testing.T) {
105+// exactly once.
88− f := newSubscriberFixture(t, 0)
106+func TestEvents_OneConsumer(t *testing.T) {
107+ t.Parallel()
89
108
90− time.Sleep(10 * time.Millisecond)
109+ ctx, cancel := context.WithCancel(t.Context())
91− synctest.Wait()
110+ defer cancel()
92
111
93− // First event: nobody reads outputCh so the timer fires (message dropped).
112+ broker := pubsub.NewBroker[tea.Msg]()
94− f.broker.Publish(pubsub.CreatedEvent, "event1")
113+ defer broker.Shutdown()
95− time.Sleep(subscriberSendTimeout + 25*time.Millisecond)
96− synctest.Wait()
97
114
98− // Second event: triggers Stop()==false path; without the fix this deadlocks.
115+ ch := broker.Subscribe(ctx)
99− f.broker.Publish(pubsub.CreatedEvent, "event2")
100
116
101− // If the timer drain deadlocks, wg.Wait never returns.
117+ const n = 10
102− done := make(chan struct{})
118+ for i := range n {
103− go func() {
119+ broker.Publish(pubsub.UpdatedEvent, tea.Msg(i))
104− f.cancel()
105− f.wg.Wait()
106− close(done)
107− }()
120+ }
108
121
122+ for i := range n {
109 select {
123 select {
110− case <-done:
124+ case ev := <-ch:
125+ require.Equal(t, tea.Msg(i), ev.Payload)
111 case <-time.After(5 * time.Second):
126 case <-time.After(5 * time.Second):
112− t.Fatal("setupSubscriber goroutine hung — likely timer drain deadlock")
127+ t.Fatalf("timed out waiting for event %d", i)
113 }
128 }
114− })
129+ }
115}
130}
116
131
117−func TestSetupSubscriber_NoTimerLeak(t *testing.T) {
132+// TestEvents_NConsumers verifies that every subscriber receives every event
118− defer goleak.VerifyNone(t)
133+// exactly once, regardless of how many concurrent consumers are attached.
119− synctest.Test(t, func(t *testing.T) {
134+func TestEvents_NConsumers(t *testing.T) {
120− f := newSubscriberFixture(t, 100)
135+ t.Parallel()
121
136
122− for range 100 {
137+ for _, n := range []int{2, 5, 10} {
123− f.broker.Publish(pubsub.CreatedEvent, "event")
138+ t.Run(fmt.Sprintf("consumers=%d", n), func(t *testing.T) {
124− time.Sleep(5 * time.Millisecond)
139+ t.Parallel()
125− synctest.Wait()
140+ testNConsumers(t, n)
126− }
141+ })
127−
142+ }
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}
143}
139
144
140−func newSubscriberFixture(t *testing.T, bufSize int) *subscriberFixture {
145+func testNConsumers(t *testing.T, n int) {
141 t.Helper()
146 t.Helper()
147+
142 ctx, cancel := context.WithCancel(t.Context())
148 ctx, cancel := context.WithCancel(t.Context())
143− t.Cleanup(cancel)
149+ defer cancel()
150+
151+ broker := pubsub.NewBroker[tea.Msg]()
152+ defer broker.Shutdown()
144
153
145− f := &subscriberFixture{
154+ // Subscribe all N consumers before publishing.
146− broker: pubsub.NewBroker[string](),
147− outputCh: make(chan tea.Msg, bufSize),
155+ channels := make([]<-chan pubsub.Event[tea.Msg], n)
148− cancel: cancel,
156+ for i := range n {
157+ channels[i] = broker.Subscribe(ctx)
149 }
158 }
150− t.Cleanup(f.broker.Shutdown)
159+ require.Equal(t, n, broker.GetSubscriberCount())
151
160
152− setupSubscriber(ctx, &f.wg, "test", func(ctx context.Context) <-chan pubsub.Event[string] {
161+ const numEvents = 20
153− return f.broker.Subscribe(ctx)
162+ for i := range numEvents {
154− }, f.outputCh)
163+ broker.Publish(pubsub.UpdatedEvent, tea.Msg(i))
164+ }
155
165
166+ // Each consumer must receive all numEvents messages.
167+ var wg sync.WaitGroup
168+ for i, ch := range channels {
169+ wg.Go(func() {
170+ for j := range numEvents {
171+ select {
172+ case ev := <-ch:
173+ require.Equal(t, tea.Msg(j), ev.Payload,
174+ "consumer %d: wrong payload for event %d", i, j)
175+ case <-time.After(5 * time.Second):
176+ t.Errorf("consumer %d: timed out waiting for event %d", i, j)
156− return f
177+ return
178+ }
179+ }
180+ })
181+ }
182+ wg.Wait()
157}
183}
internal/backend/events.go+5−3
@@ -8,16 +8,18 @@ import (
8 mcptools "github.com/charmbracelet/crush/internal/agent/tools/mcp"
8 mcptools "github.com/charmbracelet/crush/internal/agent/tools/mcp"
9 "github.com/charmbracelet/crush/internal/app"
9 "github.com/charmbracelet/crush/internal/app"
10 "github.com/charmbracelet/crush/internal/config"
10 "github.com/charmbracelet/crush/internal/config"
11+ "github.com/charmbracelet/crush/internal/pubsub"
11)
12)
12
13
13−// SubscribeEvents returns the event channel for a workspace's app.
14+// SubscribeEvents returns a per-caller event channel for a workspace.
15+// Each caller receives all events; multiple callers do not compete.
14−func (b *Backend) SubscribeEvents(workspaceID string) (<-chan tea.Msg, error) {
16+func (b *Backend) SubscribeEvents(ctx context.Context, workspaceID string) (<-chan pubsub.Event[tea.Msg], error) {
15 ws, err := b.GetWorkspace(workspaceID)
17 ws, err := b.GetWorkspace(workspaceID)
16 if err != nil {
18 if err != nil {
17 return nil, err
19 return nil, err
18 }
20 }
19
21
20− return ws.Events(), nil
22+ return ws.Events(ctx), nil
21}
23}
22
24
23// GetLSPStates returns the state of all LSP clients.
25// GetLSPStates returns the state of all LSP clients.
internal/server/proto.go+3−3
@@ -199,7 +199,7 @@ func (c *controllerV1) handleGetWorkspaceProviders(w http.ResponseWriter, r *htt
199func (c *controllerV1) handleGetWorkspaceEvents(w http.ResponseWriter, r *http.Request) {
199func (c *controllerV1) handleGetWorkspaceEvents(w http.ResponseWriter, r *http.Request) {
200 flusher := http.NewResponseController(w)
200 flusher := http.NewResponseController(w)
201 id := r.PathValue("id")
201 id := r.PathValue("id")
202− events, err := c.backend.SubscribeEvents(id)
202+ events, err := c.backend.SubscribeEvents(r.Context(), id)
203 if err != nil {
203 if err != nil {
204 c.handleError(w, r, err)
204 c.handleError(w, r, err)
205 return
205 return
@@ -218,8 +218,8 @@ func (c *controllerV1) handleGetWorkspaceEvents(w http.ResponseWriter, r *http.R
218 if !ok {
218 if !ok {
219 return
219 return
220 }
220 }
221− c.server.logDebug(r, "Sending event", "event", fmt.Sprintf("%T %+v", ev, ev))
221+ c.server.logDebug(r, "Sending event", "event", fmt.Sprintf("%T %+v", ev.Payload, ev.Payload))
222− wrapped := wrapEvent(ev)
222+ wrapped := wrapEvent(ev.Payload)
223 if wrapped == nil {
223 if wrapped == nil {
224 continue
224 continue
225 }
225 }