diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index 31fb6e774d..9fc8f51058 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -192,8 +192,11 @@ def _stream_key(user_id: str, message_id: str) -> str: async def _stream_log_append(user_id: str, message_id: str, envelope: dict, seq: int) -> None: """Append an envelope to the resume log. - Uses explicit stream ID `0-{seq}` so XRANGE can start at the client's - cursor on resume. Pipelined with the periodic EXPIRE. + Uses Redis-generated stream IDs (the default `*`) so overlapping + emitters for the same message_id — continuation, crash-retry, etc. — + cannot collide with each other's IDs. The seq lives in the entry + fields instead, and the read path filters by it. Pipelined with the + periodic EXPIRE. """ if REDIS is None or not user_id or not message_id: return @@ -203,8 +206,7 @@ async def _stream_log_append(user_id: str, message_id: str, envelope: dict, seq: pipe = REDIS.pipeline(transaction=False) pipe.xadd( key, - {'payload': json.dumps(envelope)}, - id=f'0-{seq}', + {'seq': str(seq), 'payload': json.dumps(envelope)}, maxlen=RESUME_STREAM_MAXLEN, approximate=True, ) @@ -225,14 +227,18 @@ async def _stream_log_truncate(user_id: str, message_id: str) -> None: async def _stream_log_read(user_id: str, message_id: str, after_seq: int): - """Return envelopes with seq > after_seq, in order.""" + """Return envelopes with seq > after_seq, in order. + + Full scan bounded by MAXLEN; Python filters by seq because auto IDs + don't encode it. With MAXLEN=2000 this is a few ms at worst and + resume is a rare, user-driven event so the cost is fine. + """ if REDIS is None or not user_id or not message_id: return [] try: - start_seq = max(0, after_seq) + 1 entries = await REDIS.xrange( _stream_key(user_id, message_id), - min=f'0-{start_seq}', + min='-', max='+', ) except Exception as e: @@ -250,6 +256,12 @@ async def _stream_log_read(user_id: str, message_id: str, after_seq: int): out = [] for _entry_id, fields in entries: + try: + entry_seq = int(_field(fields, 'seq') or '0') + except (TypeError, ValueError): + continue + if entry_seq <= after_seq: + continue payload = _field(fields, 'payload') if not payload: continue @@ -943,14 +955,9 @@ async def disconnect(sid): async def get_event_emitter(request_info, update_db=True): - # Per-emitter monotonic seq for the resume log (explicit ID 0-{seq}). + # Per-emitter monotonic seq for the resume log. Lives in the entry's + # `seq` field (Redis auto-generates the stream IDs). seq_counter = {'n': 0} - # Wipe any prior log so fresh XADDs (starting at 0-1) don't collide - # with a retry/continuation's leftover IDs. - ri_user_id = request_info.get('user_id') if isinstance(request_info, dict) else None - ri_message_id = request_info.get('message_id') if isinstance(request_info, dict) else None - if ri_user_id and ri_message_id and REDIS is not None: - await _stream_log_truncate(ri_user_id, ri_message_id) async def __event_emitter__(event_data): user_id = request_info['user_id'] diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte index aeaa961684..b50d72d4dd 100644 --- a/src/lib/components/chat/Chat.svelte +++ b/src/lib/components/chat/Chat.svelte @@ -650,6 +650,20 @@ const RESUME_FENCE_TIMEOUT_MS = 10000; const resumeFenceTimerByMessageId = new Map(); + // Drop the fence + timer without applying any buffered events. + // Used on disconnect / chat-change / init transitions where the + // buffered events are about state that is about to become stale. + const dropResumeFence = (messageId) => { + const timer = resumeFenceTimerByMessageId.get(messageId); + if (timer) { + clearTimeout(timer); + resumeFenceTimerByMessageId.delete(messageId); + } + resumeQueueByMessageId.delete(messageId); + }; + + // Drop fence AND flush buffered events through chatEventHandler. + // Used on the happy path (replay ack arrived) and on timeout fallback. const clearResumeFence = async (messageId) => { const timer = resumeFenceTimerByMessageId.get(messageId); if (timer) { @@ -712,9 +726,9 @@ } }; - const clearAllResumeFences = () => { + const dropAllResumeFences = () => { for (const messageId of [...resumeQueueByMessageId.keys()]) { - clearResumeFence(messageId); + dropResumeFence(messageId); } }; @@ -827,7 +841,7 @@ // 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); + $socket?.on('disconnect', dropAllResumeFences); $audioQueue?.destroy(); @@ -947,7 +961,7 @@ $socket?.off('events', chatEventHandler); $socket?.off('connect', requestResumeForAllInProgress); $socket?.off('resume-stream:replay', onResumeStreamReplay); - $socket?.off('disconnect', clearAllResumeFences); + $socket?.off('disconnect', dropAllResumeFences); audioQueueInstance?.destroy(); audioQueue.set(null); } catch (e) { @@ -1211,7 +1225,7 @@ const initNewChat = async () => { console.log('initNewChat'); resumeSeqByMessageId.clear(); - clearAllResumeFences(); + dropAllResumeFences(); if ($user?.role !== 'admin' && $user?.permissions?.chat?.temporary_enforced) { await temporaryChatEnabled.set(true); @@ -1451,7 +1465,7 @@ chatId.set(chatIdProp); resumeSeqByMessageId.clear(); - clearAllResumeFences(); + dropAllResumeFences(); if ($temporaryChatEnabled) { temporaryChatEnabled.set(false);