mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
fix(chatd): start stream buffering before publishing running status (#22571)
## Problem There is a race condition in the chat stream reconnect path. When a client connects (or reconnects) to `/stream`, sometimes they only see a `status: running` event but never receive any `message_part` events — the stream appears stuck. ## Root Cause In `processChat`, the sequence is: 1. `publishStatus(running)` — broadcasts `status: running` to all subscribers and via pubsub. 2. `runChat()` is called. 3. Inside `runChat`, there's significant setup work (model resolution, DB queries, title generation, prompt building, instruction resolution). 4. Only **after** all that setup does `runChat` set `buffering = true` on the stream state. If a client connects to `/stream` between steps 1 and 4: - `Subscribe()` reads `chat.Status == running` from the DB, so it includes `status: running` in the snapshot. - But `buffering` is still `false`, so `subscribeToStream` returns an **empty** local snapshot (no message_parts). - `publishToStream` **drops** all `message_part` events when `buffering` is false. - Result: client sees `running` but never gets any streaming content. ## Fix Move the `buffering = true` setup (and its deferred cleanup) from `runChat` into `processChat`, right before `publishStatus(running)`. This guarantees the buffer is active before any subscriber can observe `status: running`, so: - The snapshot always includes any in-flight `message_part` events. - `publishToStream` never drops parts because buffering is already on.
This commit is contained in:
+26
-16
@@ -1674,6 +1674,27 @@ func (p *Server) processChat(ctx context.Context, chat database.Chat) {
|
||||
}
|
||||
}()
|
||||
|
||||
// Start buffering stream events BEFORE publishing the running
|
||||
// status. This closes a race where a subscriber sees
|
||||
// status=running but misses message_part events because
|
||||
// buffering hasn't started yet — the subscriber gets an empty
|
||||
// snapshot and publishToStream drops message_parts while
|
||||
// buffering is false.
|
||||
p.streamMu.Lock()
|
||||
startState := p.streamStateLocked(chat.ID)
|
||||
startState.buffer = nil
|
||||
startState.buffering = true
|
||||
p.streamMu.Unlock()
|
||||
defer func() {
|
||||
p.streamMu.Lock()
|
||||
if stopState, ok := p.chatStreams[chat.ID]; ok {
|
||||
stopState.buffer = nil
|
||||
stopState.buffering = false
|
||||
p.cleanupStreamIfIdleLocked(chat.ID, stopState)
|
||||
}
|
||||
p.streamMu.Unlock()
|
||||
}()
|
||||
|
||||
p.publishStatus(chat.ID, database.ChatStatusRunning, uuid.NullUUID{
|
||||
UUID: p.workerID,
|
||||
Valid: true,
|
||||
@@ -1920,22 +1941,11 @@ func (p *Server) runChat(
|
||||
prompt = chatprompt.InsertSystem(prompt, defaultSubagentInstruction)
|
||||
}
|
||||
|
||||
// Start buffering stream events for this chat so that new
|
||||
// subscribers receive a snapshot of in-flight message parts.
|
||||
p.streamMu.Lock()
|
||||
startState := p.streamStateLocked(chat.ID)
|
||||
startState.buffer = nil
|
||||
startState.buffering = true
|
||||
p.streamMu.Unlock()
|
||||
defer func() {
|
||||
p.streamMu.Lock()
|
||||
if stopState, ok := p.chatStreams[chat.ID]; ok {
|
||||
stopState.buffer = nil
|
||||
stopState.buffering = false
|
||||
p.cleanupStreamIfIdleLocked(chat.ID, stopState)
|
||||
}
|
||||
p.streamMu.Unlock()
|
||||
}()
|
||||
// NOTE: Buffering was already started in processChat before
|
||||
// the running status was published, so message_part events
|
||||
// are captured from the moment subscribers can see
|
||||
// status=running. The deferred cleanup also lives in
|
||||
// processChat.
|
||||
|
||||
currentChat := chat
|
||||
loadChatSnapshot := func(
|
||||
|
||||
Reference in New Issue
Block a user