From 860bde88427e5d686e4dcbcd6cf4b4913c938f23 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 14 Apr 2026 21:34:44 +0000 Subject: [PATCH] fix(stream): close replay race, cursor-efficient resume, bounded client bookkeeping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review findings, all valid: 1. Race: log-then-emit instead of emit-then-log The previous ordering emitted the live WS frame first and appended to Redis afterward. A client disconnecting in that window and reconnecting before the append completed would issue resume-stream, see nothing newer, and never ask again — permanently losing that frame (most painfully the terminal done:True frame). Inverted the order: the log is now the source of truth for what was "sent", and the live emit follows. A reconnect that races the append now may see a replayed frame before the live frame arrives in the new session, but the seq idempotency guard in chatEventHandler drops the duplicate harmlessly. Duplicates are safe; losses are not. 2. Resume reads no longer full-scan XADD now uses an explicit stream ID of `0-{seq}` (the per-emitter seq counter guarantees strict monotonicity), so _stream_log_read can start XRANGE at `0-{after_seq+1}` and let Redis skip entries already delivered. Near-tail resumes (the common case) go from O(MAXLEN) to O(missed frames). Dropped the now-redundant seq field from the stream entries and the Python-side seq filter — the envelope still carries seq internally for client-side idempotency. 3. Client map is now pruned resumeSeqByMessageId was growing unboundedly across the session. Now: - cleared on loadChat (chat switch) - cleared on initNewChat (new chat) - individual entries deleted in chatCompletionEventHandler's done:true branch (most messages evict quickly this way) Long-lived sessions no longer accumulate dead keys for every completed message they've ever seen. --- backend/open_webui/socket/main.py | 48 +++++++++++++++++++++-------- src/lib/components/chat/Chat.svelte | 13 ++++++++ 2 files changed, 48 insertions(+), 13 deletions(-) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index 2488b45a17..fa81b84541 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -211,6 +211,12 @@ async def _stream_log_append(message_id: str, envelope: dict, seq: int) -> None: Uses a Redis pipeline to collapse XADD + (occasional) EXPIRE into a single network round-trip per call, keeping the added latency on the streaming hot path bounded to one Redis RTT. + + Uses an explicit stream ID of `0-{seq}` so that the resume read path + can start XRANGE at the caller's cursor (`0-{last_seq+1}`) instead of + scanning the whole stream. The `0-` prefix is arbitrary — we only + need the IDs to be strictly monotonic per stream, which our + per-emitter seq counter guarantees. """ if REDIS is None or not message_id: return @@ -220,7 +226,8 @@ async def _stream_log_append(message_id: str, envelope: dict, seq: int) -> None: pipe = REDIS.pipeline(transaction=False) pipe.xadd( key, - {'seq': str(seq), 'payload': json.dumps(envelope)}, + {'payload': json.dumps(envelope)}, + id=f'0-{seq}', maxlen=RESUME_STREAM_MAXLEN, approximate=True, ) @@ -242,11 +249,23 @@ async def _stream_log_truncate(message_id: str) -> None: async def _stream_log_read(message_id: str, after_seq: int): - """Return envelopes logged for message_id with seq > after_seq, in order.""" + """Return envelopes logged for message_id with seq > after_seq, in order. + + Stream IDs are `0-{seq}`, so we can start the XRANGE at + `0-{after_seq+1}` and let Redis skip everything already delivered + instead of scanning from the start every call. Near-tail resumes + (the common case for reconnects) become O(missed frames) instead of + O(MAXLEN). + """ if REDIS is None or not message_id: return [] try: - entries = await REDIS.xrange(_stream_key(message_id), min='-', max='+') + start_seq = max(0, after_seq) + 1 + entries = await REDIS.xrange( + _stream_key(message_id), + min=f'0-{start_seq}', + max='+', + ) except Exception as e: log.debug(f'stream resume log read failed for {message_id}: {e}') return [] @@ -263,12 +282,6 @@ async def _stream_log_read(message_id: str, after_seq: int): out = [] for _entry_id, fields in entries: - try: - seq = int(_field(fields, 'seq') or '0') - except (TypeError, ValueError): - continue - if seq <= after_seq: - continue payload = _field(fields, 'payload') if not payload: continue @@ -1012,11 +1025,20 @@ async def get_event_emitter(request_info, update_db=True): 'data': event_data, } - await sio.emit('events', envelope, room=f'user:{user_id}') - - # Append to the resume log AFTER the live emit so reconnecting - # clients can only ever see what live clients already received. + # Append to the resume log BEFORE the live emit. If we emitted + # first, a client that disconnects in the window between `sio.emit` + # and `_stream_log_append` could reconnect and issue resume-stream + # before the frame is logged, never see that frame in the replay, + # and then never ask again — permanently losing it (particularly + # painful for the terminal done:True frame). + # + # Logging first inverts the window: a reconnecting client MIGHT + # see a replayed frame before the live emit reaches their new + # session, but the seq idempotency guard in chatEventHandler drops + # the subsequent duplicate harmlessly. Duplicates are safe, losses + # are not. await _stream_log_append(message_id, envelope, seq) + await sio.emit('events', envelope, room=f'user:{user_id}') # If this event finalized the message, schedule log cleanup. Give # reconnecting clients a short grace window to pick up the final diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte index bac444d6c9..e7c319535c 100644 --- a/src/lib/components/chat/Chat.svelte +++ b/src/lib/components/chat/Chat.svelte @@ -1140,6 +1140,11 @@ const initNewChat = async () => { console.log('initNewChat'); + // Reset transport bookkeeping — the new chat has no in-flight + // resume state, and anything carried over would be for messages + // that no longer exist in this view. + resumeSeqByMessageId.clear(); + if ($user?.role !== 'admin' && $user?.permissions?.chat?.temporary_enforced) { await temporaryChatEnabled.set(true); } @@ -1377,6 +1382,10 @@ const loadChat = async () => { chatId.set(chatIdProp); + // Clear any seq bookkeeping carried over from a previous chat + // before populating history with this chat's messages. + resumeSeqByMessageId.clear(); + if ($temporaryChatEnabled) { temporaryChatEnabled.set(false); } @@ -1827,6 +1836,10 @@ if (done) { message.done = true; + // Resume log entry is no longer needed; drop it so the map + // doesn't accumulate dead keys over long-lived sessions. + resumeSeqByMessageId.delete(message.id); + if ($settings.responseAutoCopy) { copyToClipboard(message.content); }