diff --git a/coderd/chatd/chatd.go b/coderd/chatd/chatd.go index f08b22a8de..1265166c12 100644 --- a/coderd/chatd/chatd.go +++ b/coderd/chatd/chatd.go @@ -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() } } diff --git a/coderd/chatd/chatd_internal_test.go b/coderd/chatd/chatd_internal_test.go index 27fd482b54..6cd6100579 100644 --- a/coderd/chatd/chatd_internal_test.go +++ b/coderd/chatd/chatd_internal_test.go @@ -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) +}