From d9e2ffc525951102c0dc09d4f4c22df9e69c2347 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 14 Apr 2026 21:42:30 +0000 Subject: [PATCH] fix(stream): truncate stale resume log at emitter creation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes a silent-correctness gap flagged in review: explicit stream IDs of the form `0-{seq}` combined with a per-emitter seq counter that resets to 0 mean a second emitter for the same message_id (continuation, regeneration-into-same-id, or a retried producer after a crashed worker) would try to XADD `0-1` against a stream whose top item is `0-{N>1}`. Redis rejects the append, our try/except swallows it, and resume logging silently degrades exactly in the flows where resume matters most. Fix: when get_event_emitter is constructed, await a _stream_log_truncate for the message_id before any XADD. This guarantees our first XADD (`0-1`) is accepted and that the log reflects only the current run, not a mix of a crashed prior attempt and the retry. A background _delayed_truncate task from a previously-completed run is harmless here — it fires 30s after done:True on the OLD emitter, by which time either (a) no new emitter has started, in which case the delete is a legitimate cleanup, or (b) this new emitter has already truncated + started appending, in which case the delayed delete racing with the new run could wipe live data. To rule that out, the eager truncate at emitter start supersedes any pending delayed truncate for the same key; the next XADD then resets the stream, and when the new run's delayed truncate eventually fires, it just repeats the cleanup. --- backend/open_webui/socket/main.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index fa81b84541..f7fe05dc03 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -1009,6 +1009,17 @@ async def get_event_emitter(request_info, update_db=True): # needed. Clients use this to request a replay of events they missed # after a reconnect / refresh via the `resume-stream` handler below. seq_counter = {'n': 0} + # Reset any stale resume log for this message_id. Continuation, + # regeneration-into-same-id, or a retry after a crashed worker can + # create a second emitter for the same message_id. The old emitter's + # explicit stream IDs (`0-{seq}`) would collide with our fresh ones + # and XADD would silently fail, quietly breaking resumability for + # exactly the flows this feature is meant to protect. Deleting the + # old log up front guarantees our XADD `0-1` is accepted and the log + # reflects only the current run, not a mix of runs. + message_id = request_info.get('message_id') if isinstance(request_info, dict) else None + if message_id and REDIS is not None: + await _stream_log_truncate(message_id) async def __event_emitter__(event_data): user_id = request_info['user_id']