From 9af2b478bbe3b80072c0bcc1c97800d9a9c9794f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 14 Apr 2026 22:56:14 +0000 Subject: [PATCH] fix(stream): keep seq state across terminal events, reply on auth fail, broaden terminal TTL - Stopped deleting resumeSeqByMessageId entries on done / cancel / error. The map now only clears on chat/navigation transitions (loadChat, initNewChat, component unmount). Fixes a continuation- reuses-message_id duplication hole: if a message completes and its id is immediately reused for a continued/extended response, the log's 30s grace window can still serve old envelopes; resetting the client's lastSeq to 0 on done made those old envelopes replay. Keeping the seq alive until the message is truly retired avoids re-applying them. - resume_stream now always replies when the payload has a message_id, even on auth failure (empty envelopes). Silent early-return branches could previously leave the client fence raised for the full 10s fallback timeout during reconnect/auth churn. Silent drops only remain for cases where the client couldn't have raised a fence. - Terminal TTL shortening now also triggers on chat:tasks:cancel and error-finalized completions, not just done:True. Cancelled/errored streams no longer leak log+seq keys for the full 1h TTL. --- backend/open_webui/socket/main.py | 66 +++++++++++++++-------------- src/lib/components/chat/Chat.svelte | 7 --- 2 files changed, 34 insertions(+), 39 deletions(-) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index 1703ec95e8..5655671228 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -671,36 +671,30 @@ async def chat_events(sid, data): @sio.on('resume-stream') async def resume_stream(sid, data): - """One `resume-stream:replay` batch emit (payload + completion signal). + """Batch replay emit; also serves as the client's fence-clear signal. - Sent whenever we have a valid (user, message_id); auth-rejection - branches drop silently since the client never raised a fence for - those. Key scoping by user_id means no DB auth check is needed. + Reply whenever we have a message_id to target so the client fence + clears deterministically — even on auth failure. Silent drops only + happen when the client couldn't have raised a fence in the first + place (malformed payload, no message_id). """ if not isinstance(data, dict): return - - user = SESSION_POOL.get(sid) - if not user: - return - - user_id = user.get('id') message_id = data.get('message_id') - request_id = data.get('request_id') - try: - last_seq = int(data.get('last_seq') or 0) - except (TypeError, ValueError): - last_seq = 0 - - if not user_id or not message_id: + if not message_id: return + request_id = data.get('request_id') envelopes = [] - if REDIS is not None: + user = SESSION_POOL.get(sid) + user_id = user.get('id') if user else None + if user_id and REDIS is not None: + try: + last_seq = int(data.get('last_seq') or 0) + except (TypeError, ValueError): + last_seq = 0 envelopes = await _stream_log_read(user_id, message_id, last_seq) - # Echo request_id so the client can ignore stale replies that - # correspond to a superseded request (reconnect churn). await sio.emit( 'resume-stream:replay', { @@ -1033,19 +1027,27 @@ async def get_event_emitter(request_info, update_db=True): await _stream_log_append(user_id, message_id, envelope, seq) await sio.emit('events', envelope, room=f'user:{user_id}') - # On done, shorten TTL so log + seq counter self-evict together. + # Any terminal event shortens TTL so log + seq self-evict together. + # Covers normal completion (done:True), explicit cancel, and + # errored completions — all end-of-stream flows that the client + # will no longer replay against. + outer_type = event_data.get('type') if isinstance(event_data, dict) else None inner = event_data.get('data') if isinstance(event_data, dict) else None - if isinstance(inner, dict) and inner.get('done') is True: - if REDIS is not None and user_id and message_id: - try: - pipe = REDIS.pipeline(transaction=False) - pipe.expire(_stream_key(user_id, message_id), RESUME_STREAM_DONE_TTL_SEC) - pipe.expire(_stream_seq_key(user_id, message_id), RESUME_STREAM_DONE_TTL_SEC) - await asyncio.wait_for( - pipe.execute(), timeout=RESUME_STREAM_REDIS_TIMEOUT_SEC - ) - except Exception as e: - log.warning(f'stream resume log done-TTL shorten failed for {message_id}: {e}') + is_terminal = ( + (isinstance(inner, dict) and inner.get('done') is True) + or outer_type == 'chat:tasks:cancel' + or (isinstance(inner, dict) and inner.get('error')) + ) + if is_terminal and REDIS is not None and user_id and message_id: + try: + pipe = REDIS.pipeline(transaction=False) + pipe.expire(_stream_key(user_id, message_id), RESUME_STREAM_DONE_TTL_SEC) + pipe.expire(_stream_seq_key(user_id, message_id), RESUME_STREAM_DONE_TTL_SEC) + await asyncio.wait_for( + pipe.execute(), timeout=RESUME_STREAM_REDIS_TIMEOUT_SEC + ) + except Exception as e: + log.warning(f'stream resume log terminal-TTL shorten failed for {message_id}: {e}') if update_db and message_id and not request_info.get('chat_id', '').startswith('local:'): event_type = event_data.get('type') diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte index ceabf2da95..f4d4f0f08e 100644 --- a/src/lib/components/chat/Chat.svelte +++ b/src/lib/components/chat/Chat.svelte @@ -485,12 +485,10 @@ // Set all response messages to done for (const messageId of history.messages[message.parentId].childrenIds) { history.messages[messageId].done = true; - resumeSeqByMessageId.delete(messageId); } await processNextInQueue($chatId); } else { message.done = true; - resumeSeqByMessageId.delete(message.id); } } else if (type === 'chat:message:delta' || type === 'message') { message.content += data.content; @@ -1908,8 +1906,6 @@ if (done) { message.done = true; - resumeSeqByMessageId.delete(message.id); - if ($settings.responseAutoCopy) { copyToClipboard(message.content); } @@ -2541,7 +2537,6 @@ }; responseMessage.done = true; - resumeSeqByMessageId.delete(responseMessageId); history.messages[responseMessageId] = responseMessage; history.currentId = responseMessageId; @@ -2609,7 +2604,6 @@ content: $i18n.t(`Uh-oh! There was an issue with the response.`) + '\n' + errorMessage }; responseMessage.done = true; - resumeSeqByMessageId.delete(responseMessage.id); if (responseMessage.statusHistory) { responseMessage.statusHistory = responseMessage.statusHistory.filter( @@ -2643,7 +2637,6 @@ if (responseMessage.parentId && history.messages[responseMessage.parentId]) { for (const messageId of history.messages[responseMessage.parentId].childrenIds) { history.messages[messageId].done = true; - resumeSeqByMessageId.delete(messageId); } }