fix(coderd/chatd): rate-limit stream drop WARN logs to avoid log spam (#23340)

- Rate-limit "chat stream buffer full" and "dropping chat stream event" WARN logs to at most once per 10s per chat. - Intermediate drops not logged; WARN includes `dropped_count`. - Per-chat tracking on `chatStreamState` using timestamp comparison against `quartz.Clock` — no global tickers, no new `Server` fields. - Subscriber and buffer drop counters reset at all lifecycle boundaries. > 🤖 This PR was created with the help of Coder Agents, and was reviewed by my human. 🧑‍💻
2026-09-24 15:04:27 +08:00 · 2026-03-20 12:16:39 +00:00
parent e08c3c1699
commit d8cad81ada
2 changed files with 162 additions and 7 deletions
@@ -64,6 +64,14 @@ const (
 	// of 5 means recovery runs at 1/5 of the stale-after duration.
 	staleRecoveryIntervalDivisor = 5

+	// streamDropWarnInterval controls how often WARN-level logs are
+	// emitted when stream events are dropped. Between intervals the
+	// drop is logged at DEBUG to avoid log spam. This uses a
+	// timestamp comparison rather than a quartz.Ticker because the
+	// state is per-chat — a ticker per chat would require extra
+	// goroutines and lifecycle management.
+	streamDropWarnInterval = 10 * time.Second
+
 	// DefaultMaxChatsPerAcquire is the maximum number of chats to
 	// acquire in a single processOnce call. Batching avoids
 	// waiting a full polling interval between acquisitions
@@ -326,6 +334,19 @@ type chatStreamState struct {
 	durableMessages      []codersdk.ChatStreamEvent
 	durableEvictedBefore int64 // highest message ID evicted from durable cache
 	subscribers          map[uuid.UUID]chan codersdk.ChatStreamEvent
+	bufferDropCount      int64
+	bufferLastWarnAt     time.Time
+	subscriberDropCount  int64
+	subscriberLastWarnAt time.Time
+}
+
+// resetDropCounters zeroes the rate-limiting state for both buffer
+// and subscriber drop warnings. The caller must hold s.mu.
+func (s *chatStreamState) resetDropCounters() {
+	s.bufferDropCount = 0
+	s.bufferLastWarnAt = time.Time{}
+	s.subscriberDropCount = 0
+	s.subscriberLastWarnAt = time.Time{}
 }

 // MaxQueueSize is the maximum number of queued user messages per chat.
@@ -1498,8 +1519,17 @@ func (p *Server) publishToStream(chatID uuid.UUID, event codersdk.ChatStreamEven
 			return
 		}
 		if len(state.buffer) >= maxStreamBufferSize {
-			p.logger.Warn(context.Background(), "chat stream buffer full, dropping oldest event",
-				slog.F("chat_id", chatID), slog.F("buffer_size", len(state.buffer)))
+			state.bufferDropCount++
+			now := p.clock.Now()
+			if now.Sub(state.bufferLastWarnAt) >= streamDropWarnInterval {
+				p.logger.Warn(context.Background(), "chat stream buffer full, dropping oldest event",
+					slog.F("chat_id", chatID),
+					slog.F("buffer_size", len(state.buffer)),
+					slog.F("dropped_count", state.bufferDropCount),
+				)
+				state.bufferDropCount = 0
+				state.bufferLastWarnAt = now
+			}
 			state.buffer = state.buffer[1:]
 		}
 		state.buffer = append(state.buffer, event)
@@ -1510,19 +1540,32 @@ func (p *Server) publishToStream(chatID uuid.UUID, event codersdk.ChatStreamEven
 	}
 	state.mu.Unlock()

+	var subDropped int64
 	for _, ch := range subscribers {
 		select {
 		case ch <- event:
 		default:
-			p.logger.Warn(context.Background(), "dropping chat stream event",
-				slog.F("chat_id", chatID), slog.F("type", event.Type))
+			subDropped++
 		}
 	}

-	// Clean up the stream entry if it was created by
-	// getOrCreateStreamState but has no subscribers and is not
-	// actively buffering (e.g. publish with no watchers).
+	// Re-acquire the lock once for both subscriber-drop logging and
+	// idle cleanup. Merging these avoids an unnecessary unlock/re-lock
+	// gap between the two sections.
 	state.mu.Lock()
+	if subDropped > 0 {
+		state.subscriberDropCount += subDropped
+		now := p.clock.Now()
+		if now.Sub(state.subscriberLastWarnAt) >= streamDropWarnInterval {
+			p.logger.Warn(context.Background(), "dropping chat stream event",
+				slog.F("chat_id", chatID),
+				slog.F("type", event.Type),
+				slog.F("dropped_count", state.subscriberDropCount),
+			)
+			state.subscriberDropCount = 0
+			state.subscriberLastWarnAt = now
+		}
+	}
 	p.cleanupStreamIfIdle(chatID, state)
 	state.mu.Unlock()
 }
@@ -2447,11 +2490,13 @@ func (p *Server) processChat(ctx context.Context, chat database.Chat) {
 	streamState := p.getOrCreateStreamState(chat.ID)
 	streamState.mu.Lock()
 	streamState.buffer = nil
+	streamState.resetDropCounters()
 	streamState.buffering = true
 	streamState.mu.Unlock()
 	defer func() {
 		streamState.mu.Lock()
 		streamState.buffer = nil
+		streamState.resetDropCounters()
 		streamState.buffering = false
 		p.cleanupStreamIfIdle(chat.ID, streamState)
 		streamState.mu.Unlock()
@@ -2997,6 +3042,7 @@ func (p *Server) runChat(
 			if ss, ok := val.(*chatStreamState); ok {
 				ss.mu.Lock()
 				ss.buffer = nil
+				ss.resetDropCounters()
 				ss.mu.Unlock()
 			}
 		}
@@ -3207,6 +3253,7 @@ func (p *Server) runChat(
 				if rs, ok := val.(*chatStreamState); ok {
 					rs.mu.Lock()
 					rs.buffer = nil
+					rs.resetDropCounters()
 					rs.mu.Unlock()
 				}
 			}
@@ -11,6 +11,7 @@ import (
 	"go.uber.org/mock/gomock"
 	"golang.org/x/xerrors"

+	"cdr.dev/slog/v3"
 	"cdr.dev/slog/v3/sloggers/slogtest"
 	"github.com/coder/coder/v2/coderd/database"
 	"github.com/coder/coder/v2/coderd/database/dbmock"
@@ -19,6 +20,8 @@ import (
 	"github.com/coder/coder/v2/codersdk"
 	"github.com/coder/coder/v2/codersdk/workspacesdk"
 	"github.com/coder/coder/v2/codersdk/workspacesdk/agentconnmock"
+	"github.com/coder/coder/v2/testutil"
+	"github.com/coder/quartz"
 )

 func TestRefreshChatWorkspaceSnapshot_NoReloadWhenWorkspacePresent(t *testing.T) {
@@ -454,3 +457,108 @@ func requireNoStreamEvent(t *testing.T, events <-chan codersdk.ChatStreamEvent,
 	case <-time.After(wait):
 	}
 }
+
+// TestPublishToStream_DropWarnRateLimiting walks through a
+// realistic lifecycle: buffer fills up, subscriber channel fills
+// up, counters get reset between steps. It verifies that WARN
+// logs are rate-limited to at most once per streamDropWarnInterval
+// and that counter resets re-enable an immediate WARN.
+func TestPublishToStream_DropWarnRateLimiting(t *testing.T) {
+	t.Parallel()
+
+	sink := testutil.NewFakeSink(t)
+	mClock := quartz.NewMock(t)
+
+	server := &Server{
+		logger: sink.Logger(),
+		clock:  mClock,
+	}
+
+	chatID := uuid.New()
+	subCh := make(chan codersdk.ChatStreamEvent, 1)
+	subCh <- codersdk.ChatStreamEvent{} // pre-fill so sends always drop
+
+	// Set up state that mirrors a running chat: buffer at capacity,
+	// buffering enabled, one saturated subscriber.
+	state := &chatStreamState{
+		buffering: true,
+		buffer:    make([]codersdk.ChatStreamEvent, maxStreamBufferSize),
+		subscribers: map[uuid.UUID]chan codersdk.ChatStreamEvent{
+			uuid.New(): subCh,
+		},
+	}
+	server.chatStreams.Store(chatID, state)
+
+	bufferMsg := "chat stream buffer full, dropping oldest event"
+	subMsg := "dropping chat stream event"
+
+	filter := func(level slog.Level, msg string) func(slog.SinkEntry) bool {
+		return func(e slog.SinkEntry) bool {
+			return e.Level == level && e.Message == msg
+		}
+	}
+
+	// --- Phase 1: buffer-full rate limiting ---
+	// message_part events hit both the buffer-full and subscriber-full
+	// paths. The first publish triggers a WARN for each; the rest
+	// within the window are DEBUG.
+	partEvent := codersdk.ChatStreamEvent{
+		Type:        codersdk.ChatStreamEventTypeMessagePart,
+		MessagePart: &codersdk.ChatStreamMessagePart{},
+	}
+	for i := 0; i < 50; i++ {
+		server.publishToStream(chatID, partEvent)
+	}
+
+	require.Len(t, sink.Entries(filter(slog.LevelWarn, bufferMsg)), 1)
+	require.Empty(t, sink.Entries(filter(slog.LevelDebug, bufferMsg)))
+	requireFieldValue(t, sink.Entries(filter(slog.LevelWarn, bufferMsg))[0], "dropped_count", int64(1))
+
+	// Subscriber also saw 50 drops (one per publish).
+	require.Len(t, sink.Entries(filter(slog.LevelWarn, subMsg)), 1)
+	require.Empty(t, sink.Entries(filter(slog.LevelDebug, subMsg)))
+	requireFieldValue(t, sink.Entries(filter(slog.LevelWarn, subMsg))[0], "dropped_count", int64(1))
+
+	// --- Phase 2: clock advance triggers second WARN with count ---
+	mClock.Advance(streamDropWarnInterval + time.Second)
+	server.publishToStream(chatID, partEvent)
+
+	bufWarn := sink.Entries(filter(slog.LevelWarn, bufferMsg))
+	require.Len(t, bufWarn, 2)
+	requireFieldValue(t, bufWarn[1], "dropped_count", int64(50))
+
+	subWarn := sink.Entries(filter(slog.LevelWarn, subMsg))
+	require.Len(t, subWarn, 2)
+	requireFieldValue(t, subWarn[1], "dropped_count", int64(50))
+
+	// --- Phase 3: counter reset (simulates step persist) ---
+	state.mu.Lock()
+	state.buffer = make([]codersdk.ChatStreamEvent, maxStreamBufferSize)
+	state.resetDropCounters()
+	state.mu.Unlock()
+
+	// The very next drop should WARN immediately — the reset zeroed
+	// lastWarnAt so the interval check passes.
+	server.publishToStream(chatID, partEvent)
+
+	bufWarn = sink.Entries(filter(slog.LevelWarn, bufferMsg))
+	require.Len(t, bufWarn, 3, "expected WARN immediately after counter reset")
+	requireFieldValue(t, bufWarn[2], "dropped_count", int64(1))
+
+	subWarn = sink.Entries(filter(slog.LevelWarn, subMsg))
+	require.Len(t, subWarn, 3, "expected subscriber WARN immediately after counter reset")
+	requireFieldValue(t, subWarn[2], "dropped_count", int64(1))
+}
+
+// requireFieldValue asserts that a SinkEntry contains a field with
+// the given name and value.
+func requireFieldValue(t *testing.T, entry slog.SinkEntry, name string, expected interface{}) {
+	t.Helper()
+	for _, f := range entry.Fields {
+		if f.Name == name {
+			require.Equal(t, expected, f.Value, "field %q value mismatch", name)
+			return
+		}
+	}
+	t.Fatalf("field %q not found in log entry", name)
+}