From 7b3071494196d3eef5b7800bf910e99e6be49b7d Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 14 Apr 2026 22:44:59 +0000 Subject: [PATCH] fix(stream): refresh seq TTL with stream TTL, tighten hot-path timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - seq key now gets EXPIRE on the same periodic cadence as the stream key, not just on the first INCR. Prevents INCR restarting at 1 on responses longer than RESUME_STREAM_TTL_SEC (1h), which would have corrupted replay ordering. - Hot-path Redis timeout trimmed 200ms → 100ms. Halves worst-case stall under degraded Redis (now ~200ms max per frame if both INCR and append time out); healthy Redis (<1ms) is unaffected. Not addressed: the fundamental 2-RTT-per-frame cost of separate INCR and XADD. Collapsing them would require either a Lua script (cluster hash-tag concerns) or switching to Redis stream IDs as the seq cursor (invasive protocol change). Leaving the structure intact — if profiling shows real stalls in prod, stream-ID-as-seq is the cleaner follow-up. --- backend/open_webui/socket/main.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index 519be492ed..28d0c7558d 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -181,7 +181,7 @@ RESUME_STREAM_DONE_TTL_SEC = 30 RESUME_STREAM_TTL_REFRESH_EVERY = 64 # Tight upper bound on any Redis call in the streaming hot path so a # slow Redis degrades resume but can't stall live token delivery. -RESUME_STREAM_REDIS_TIMEOUT_SEC = 0.2 +RESUME_STREAM_REDIS_TIMEOUT_SEC = 0.1 def _stream_key(user_id: str, message_id: str) -> str: @@ -233,6 +233,7 @@ async def _stream_log_append(user_id: str, message_id: str, envelope: dict, seq: try: refresh_ttl = (seq == 1) or (seq % RESUME_STREAM_TTL_REFRESH_EVERY == 0) key = _stream_key(user_id, message_id) + seq_key = _stream_seq_key(user_id, message_id) pipe = REDIS.pipeline(transaction=False) pipe.xadd( key, @@ -241,7 +242,11 @@ async def _stream_log_append(user_id: str, message_id: str, envelope: dict, seq: approximate=True, ) if refresh_ttl: + # Refresh both keys together — without this the seq counter + # can expire mid-stream on responses longer than the TTL and + # INCR restarts at 1, corrupting replay ordering. pipe.expire(key, RESUME_STREAM_TTL_SEC) + pipe.expire(seq_key, RESUME_STREAM_TTL_SEC) await asyncio.wait_for( pipe.execute(), timeout=RESUME_STREAM_REDIS_TIMEOUT_SEC )