coder · ethanndickson · Aug 28, 2026
diff --git a/coderd/x/chatd/stream_subscribe.go b/coderd/x/chatd/stream_subscribe.go
@@ -58,7 +58,12 @@ func (p *Server) subscribeStreamLoop(
 		return subscribeWithInitialError(chatID, "failed to subscribe to chat updates")
 	}
 
-	pollerCh, unregisterPoller := p.streamSyncPoller.Register(chatID)
+	unregisterPoller := p.streamSyncPoller.Register(chatID, func(hint streamSyncHint) {
+		select {
+		case updateCh <- hint:
+		case <-streamCtx.Done():
+		}
+	})
 	loop := newStreamLoop(chat, p.db, logger, afterMessageID)
 	// The immediate sync builds the initial snapshot returned to the caller
 	// and the relay target for the forwarder. Hints only fire on state
@@ -97,13 +102,6 @@ func (p *Server) subscribeStreamLoop(
 				if !p.runStreamSync(streamCtx, loop, relay, events, hint) {
 					return
 				}
-			case hint, ok := <-pollerCh:
-				if !ok {
-					return
-				}
-				if !p.runStreamSync(streamCtx, loop, relay, events, hint) {
-					return
-				}
 			case part, ok := <-relay.Parts():
 				if !ok {
 					return

diff --git a/coderd/x/chatd/stream_sync_poller.go b/coderd/x/chatd/stream_sync_poller.go
@@ -27,8 +27,8 @@ type streamSyncPoller struct {
 }
 
 type streamSyncPollerSubscriber struct {
-	chatID uuid.UUID
-	hints  chan streamSyncHint
+	chatID  uuid.UUID
+	deliver func(streamSyncHint)
 }
 
 func newStreamSyncPoller(
@@ -68,15 +68,17 @@ func (p *streamSyncPoller) Close() {
 	p.cancel()
 }
 
-func (p *streamSyncPoller) Register(chatID uuid.UUID) (<-chan streamSyncHint, func()) {
+// Register subscribes deliver to poll hints for chatID until the returned
+// unregister func is called. deliver is invoked outside the poller's mutex and
+// may race with unregister, so it must remain safe to call after unregister
+// returns (e.g. by guarding on the subscriber's own context).
+func (p *streamSyncPoller) Register(chatID uuid.UUID, deliver func(streamSyncHint)) (unregister func()) {
 	if p == nil {
-		ch := make(chan streamSyncHint)
-		close(ch)
-		return ch, func() {}
+		return func() {}
 	}
 	subscriber := &streamSyncPollerSubscriber{
-		chatID: chatID,
-		hints:  make(chan streamSyncHint, 1),
+		chatID:  chatID,
+		deliver: deliver,
 	}
 	p.mu.Lock()
 	if p.subscribers[chatID] == nil {
@@ -85,7 +87,7 @@ func (p *streamSyncPoller) Register(chatID uuid.UUID) (<-chan streamSyncHint, fu
 	p.subscribers[chatID][subscriber] = struct{}{}
 	p.mu.Unlock()
 
-	return subscriber.hints, func() {
+	return func() {
 		p.unregister(subscriber)
 	}
 }
@@ -101,7 +103,6 @@ func (p *streamSyncPoller) unregister(subscriber *streamSyncPollerSubscriber) {
 	if len(chatSubscribers) == 0 {
 		delete(p.subscribers, subscriber.chatID)
 	}
-	close(subscriber.hints)
 }
 
 func (p *streamSyncPoller) loop() {
@@ -132,10 +133,7 @@ func (p *streamSyncPoller) pollOnce() {
 	for _, row := range rows {
 		hint := streamSyncHintFromPollRow(row)
 		for _, subscriber := range subscribers[row.ID] {
-			select {
-			case subscriber.hints <- hint:
-			default:
-			}
+			subscriber.deliver(hint)
 		}
 	}
 }

diff --git a/coderd/x/chatd/stream_sync_poller_internal_test.go b/coderd/x/chatd/stream_sync_poller_internal_test.go
@@ -0,0 +1,74 @@
+package chatd
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/google/uuid"
+	"go.uber.org/mock/gomock"
+
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	"github.com/coder/coder/v2/coderd/database"
+	"github.com/coder/coder/v2/coderd/database/dbmock"
+)
+
+// TestStreamSyncPollerConcurrentRegisterUnregister churns subscriber
+// registration while pollOnce delivers hints concurrently. Before the poller
+// became a callback fanout, unregister closed the subscriber's hint channel
+// while pollOnce could still be mid-send on its lock-free snapshot, panicking
+// with "send on closed channel" (GHSA-7x3x-59xg-4hrc). Run with -race.
+func TestStreamSyncPollerConcurrentRegisterUnregister(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+	db := dbmock.NewMockStore(ctrl)
+	db.EXPECT().GetChatStreamSyncRows(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
+		func(_ context.Context, ids []uuid.UUID) ([]database.GetChatStreamSyncRowsRow, error) {
+			rows := make([]database.GetChatStreamSyncRowsRow, 0, len(ids))
+			for _, id := range ids {
+				rows = append(rows, database.GetChatStreamSyncRowsRow{ID: id})
+			}
+			return rows, nil
+		},
+	)
+
+	poller := newStreamSyncPoller(context.Background(), db, nil, slogtest.Make(t, nil))
+	defer poller.Close()
+
+	chatID := uuid.New()
+	done := make(chan struct{})
+	var wg sync.WaitGroup
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-done:
+					return
+				default:
+				}
+				unregister := poller.Register(chatID, func(streamSyncHint) {})
+				unregister()
+			}
+		}()
+	}
+	for range 1000 {
+		poller.pollOnce()
+	}
+	close(done)
+	wg.Wait()
+}
+
+// TestStreamSyncPollerNilRegister verifies a nil poller degrades to a no-op
+// registration instead of terminating subscribers.
+func TestStreamSyncPollerNilRegister(t *testing.T) {
+	t.Parallel()
+
+	var poller *streamSyncPoller
+	unregister := poller.Register(uuid.New(), func(streamSyncHint) {
+		t.Fatal("nil poller must never deliver hints")
+	})
+	unregister()
+}