From d655395b094558f320f9467a54a5f6318147d597 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 14 Apr 2026 22:05:22 +0000 Subject: [PATCH] fix(stream): unconditional completion, fence timeout, single-batch replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three correctness fixes from review: Critical: in non-Redis deployments the old resume_stream handler returned early without clearing the client fence, so any call to requestResumeForMessage would buffer live frames forever and freeze the UI. Server now always emits exactly one reply (with an empty envelope list when Redis is disabled or the log is empty), so the client's fence always clears regardless of backend configuration. Warning: a lost ack — whether from a transient disconnect between emit and reply, a server-side exception, a not-yet-registered client handler, or anything else — would deadlock a message's UI updates indefinitely. Added: - 10s per-message fence timeout that clears the fence and flushes buffered frames if no reply arrives - disconnect listener that drops all fences on socket loss so the reconnect-triggered fresh resume doesn't inherit a stale timer - try/finally around replay application so a malformed envelope can't skip the fence clear Suggestion: replay is now a single batch emit instead of N sequential emits. The server bundles all envelopes (possibly zero) into one `resume-stream:replay` message; the client applies them in order and immediately flushes the live-frame fence. Collapses "replay stream + completion ack" into a single round-trip, removes the per-envelope emit loop latency, and simplifies the state machine. --- backend/open_webui/socket/main.py | 29 +++++----- src/lib/components/chat/Chat.svelte | 84 +++++++++++++++++++++++------ 2 files changed, 82 insertions(+), 31 deletions(-) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index ac463890ec..31fb6e774d 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -614,13 +614,17 @@ async def chat_events(sid, data): @sio.on('resume-stream') async def resume_stream(sid, data): - """Replay log entries with seq > last_seq to the caller. + """Replay missed log entries in a single batch. + + One emit carries all envelopes with seq > last_seq (possibly zero) + and also serves as the completion signal — the client clears its + live-frame fence in the batch handler. Always emitting exactly once + (even when Redis is unavailable or the log is empty) keeps the + client from deadlocking on a fence that never clears. Auth is implicit: the stream key is scoped by user_id, so an authenticated session can only ever read its own logs. No DB lookup. """ - if REDIS is None: - return if not isinstance(data, dict): return @@ -638,19 +642,14 @@ async def resume_stream(sid, data): if not user_id or not message_id: return - envelopes = await _stream_log_read(user_id, message_id, last_seq) - for envelope in envelopes: - # Tag as replayed. The client uses this to distinguish replay - # from live frames so it can buffer live frames during replay - # and avoid advancing its seq high-water on them (which would - # otherwise cause later replay frames to be dropped by the dedupe - # guard). Replay is session-scoped; live listeners in the user - # room are already served via the normal emit path. - await sio.emit('events', {**envelope, '_replayed': True}, to=sid) - # Signal replay complete so the client flushes its buffered live - # frames in seq order. + envelopes = [] + if REDIS is not None: + envelopes = await _stream_log_read(user_id, message_id, last_seq) + await sio.emit( - 'resume-stream:complete', {'message_id': message_id}, to=sid + 'resume-stream:replay', + {'message_id': message_id, 'envelopes': envelopes}, + to=sid, ) diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte index 1e6cfbbe2b..aeaa961684 100644 --- a/src/lib/components/chat/Chat.svelte +++ b/src/lib/components/chat/Chat.svelte @@ -643,32 +643,78 @@ // Ask the server to replay any frames we missed for a message. // Idempotent — the seq guard in chatEventHandler drops duplicates. + // Safety net: if a resume-stream:replay ack never arrives (server + // crash mid-handler, network loss between emit and ack, handler not + // registered yet, etc.), clear the fence and flush anyway after this + // timeout so live UI updates don't freeze indefinitely. + const RESUME_FENCE_TIMEOUT_MS = 10000; + const resumeFenceTimerByMessageId = new Map(); + + const clearResumeFence = async (messageId) => { + const timer = resumeFenceTimerByMessageId.get(messageId); + if (timer) { + clearTimeout(timer); + resumeFenceTimerByMessageId.delete(messageId); + } + const queue = resumeQueueByMessageId.get(messageId); + if (!queue) return; + resumeQueueByMessageId.delete(messageId); + queue.sort((a, b) => (a?.seq ?? 0) - (b?.seq ?? 0)); + for (const event of queue) { + try { + await chatEventHandler(event); + } catch (e) { + console.error('resume fence flush error', e); + } + } + }; + const requestResumeForMessage = (message) => { if (!message || !message.id || message.done) return; if (!$socket || !$socket.connected) return; - // Raise the fence BEFORE emitting so any live frame that arrives - // while the server is still reading XRANGE gets buffered instead - // of racing the replay frames. + // Raise the fence BEFORE emitting so any live frame arriving + // while the server reads XRANGE is buffered, not raced past the + // replay frames. if (!resumeQueueByMessageId.has(message.id)) { resumeQueueByMessageId.set(message.id, []); } + const existingTimer = resumeFenceTimerByMessageId.get(message.id); + if (existingTimer) clearTimeout(existingTimer); + resumeFenceTimerByMessageId.set( + message.id, + setTimeout(() => { + console.warn('resume-stream fence timed out for', message.id); + clearResumeFence(message.id); + }, RESUME_FENCE_TIMEOUT_MS) + ); $socket.emit('resume-stream', { message_id: message.id, last_seq: resumeSeqByMessageId.get(message.id) ?? 0 }); }; - const onResumeStreamComplete = async (payload) => { + const onResumeStreamReplay = async (payload) => { const messageId = payload?.message_id; if (!messageId) return; - const queue = resumeQueueByMessageId.get(messageId); - if (!queue) return; - resumeQueueByMessageId.delete(messageId); - // Flush buffered live frames in seq order. chatEventHandler's - // dedupe guard will drop anything already covered by replay. - queue.sort((a, b) => (a?.seq ?? 0) - (b?.seq ?? 0)); - for (const event of queue) { - await chatEventHandler(event); + const envelopes = Array.isArray(payload?.envelopes) ? payload.envelopes : []; + try { + // Replay frames bypass the fence (they're what the fence is + // waiting for) but still pass through chatEventHandler's + // dedupe guard so anything already applied is a no-op. + for (const envelope of envelopes) { + envelope._replayed = true; + await chatEventHandler(envelope); + } + } finally { + // Always clear the fence, even if replay application threw + // partway through, so live updates aren't frozen forever. + await clearResumeFence(messageId); + } + }; + + const clearAllResumeFences = () => { + for (const messageId of [...resumeQueueByMessageId.keys()]) { + clearResumeFence(messageId); } }; @@ -776,7 +822,12 @@ // Resume any in-flight streams on reconnect. $socket?.on('connect', requestResumeForAllInProgress); - $socket?.on('resume-stream:complete', onResumeStreamComplete); + $socket?.on('resume-stream:replay', onResumeStreamReplay); + // Drop any stale fences on disconnect so the reconnect path + // starts from a clean slate instead of inheriting a timer that + // could fire after the new resume request has already raised a + // fresh fence. + $socket?.on('disconnect', clearAllResumeFences); $audioQueue?.destroy(); @@ -895,7 +946,8 @@ window.removeEventListener('message', onMessageHandler); $socket?.off('events', chatEventHandler); $socket?.off('connect', requestResumeForAllInProgress); - $socket?.off('resume-stream:complete', onResumeStreamComplete); + $socket?.off('resume-stream:replay', onResumeStreamReplay); + $socket?.off('disconnect', clearAllResumeFences); audioQueueInstance?.destroy(); audioQueue.set(null); } catch (e) { @@ -1159,7 +1211,7 @@ const initNewChat = async () => { console.log('initNewChat'); resumeSeqByMessageId.clear(); - resumeQueueByMessageId.clear(); + clearAllResumeFences(); if ($user?.role !== 'admin' && $user?.permissions?.chat?.temporary_enforced) { await temporaryChatEnabled.set(true); @@ -1399,7 +1451,7 @@ chatId.set(chatIdProp); resumeSeqByMessageId.clear(); - resumeQueueByMessageId.clear(); + clearAllResumeFences(); if ($temporaryChatEnabled) { temporaryChatEnabled.set(false);