fix(stream): mark flushed events _replayed, add Redis hot-path breaker

Critical: last round's fence-flush fix (keep the queue in the map
while draining) turned into an infinite loop because every shifted
event went back through chatEventHandler's fence-check branch and
was re-queued into the same queue we were trying to drain. Mark
each flushed event `_replayed = true` before dispatch so the
chatEventHandler fence-buffering branch short-circuits. Re-entrancy
ordering is still preserved: live frames arriving during an await
inside a flushed-event's handler still don't carry `_replayed` and
continue to buffer into the same queue, staying ordered after the
current flush batch.

Warning: under sustained Redis slowness every emit was paying
100ms on INCR and another 100ms on XADD+EXPIRE, so live token
delivery could stall 200ms/frame during an outage. Added a
module-level circuit breaker that trips after 3 consecutive
failures/timeouts and short-circuits the hot-path Redis calls for
a 10s cooldown window. First successful call after cooldown resets
the counter. Replay reads deliberately bypass the breaker because
they're user-initiated and worth the slower timeout.

backend/open_webui/socket/main.py

@@ -191,6 +191,32 @@ RESUME_STREAM_READ_TIMEOUT_SEC = float(
     os.environ.get('RESUME_STREAM_READ_TIMEOUT_SEC', '1.0')
 )
 
+# Module-level circuit breaker for the streaming hot path. After N
+# consecutive Redis failures/timeouts, short-circuit seq/log calls for
+# a cool-down window so every outgoing frame doesn't pay the timeout
+# wall-clock cost during a sustained outage. Global (not per-message)
+# because when Redis is the thing that's unhealthy, it's unhealthy for
+# everyone.
+_HOT_PATH_BREAKER_FAILURE_THRESHOLD = 3
+_HOT_PATH_BREAKER_COOLDOWN_SEC = 10.0
+_hot_path_breaker = {'failures': 0, 'open_until': 0.0}
+
+
+def _breaker_open() -> bool:
+    return time.time() < _hot_path_breaker['open_until']
+
+
+def _breaker_record_success() -> None:
+    if _hot_path_breaker['failures']:
+        _hot_path_breaker['failures'] = 0
+        _hot_path_breaker['open_until'] = 0.0
+
+
+def _breaker_record_failure() -> None:
+    _hot_path_breaker['failures'] += 1
+    if _hot_path_breaker['failures'] >= _HOT_PATH_BREAKER_FAILURE_THRESHOLD:
+        _hot_path_breaker['open_until'] = time.time() + _HOT_PATH_BREAKER_COOLDOWN_SEC
+
 
 def _stream_key(user_id: str, message_id: str) -> str:
     # user_id-scoped key — auth is implicit from the session's user.
@@ -202,21 +228,19 @@ def _stream_seq_key(user_id: str, message_id: str) -> str:
 
 
 async def _stream_seq_allocate(user_id: str, message_id: str):
-    """Allocate the next seq via atomic INCR, or None when resume is off.
-
-    Returns None when Redis is unavailable, times out, or
-    ENABLE_REALTIME_CHAT_SAVE is set. Callers emit without seq in that
-    case — live streaming continues, no dedupe/resume for that frame.
-    """
+    """Allocate the next seq via atomic INCR, or None when resume is off."""
     if ENABLE_REALTIME_CHAT_SAVE:
         return None
     if REDIS is None or not user_id or not message_id:
         return None
+    if _breaker_open():
+        return None
     try:
         key = _stream_seq_key(user_id, message_id)
         seq = await asyncio.wait_for(
             REDIS.incr(key), timeout=RESUME_STREAM_REDIS_TIMEOUT_SEC
         )
+        _breaker_record_success()
         if seq == 1:
             try:
                 await asyncio.wait_for(
@@ -227,9 +251,11 @@ async def _stream_seq_allocate(user_id: str, message_id: str):
                 pass
         return int(seq)
     except asyncio.TimeoutError:
+        _breaker_record_failure()
         log.warning(f'stream resume seq alloc timed out for {message_id}')
         return None
     except Exception as e:
+        _breaker_record_failure()
         log.warning(f'stream resume seq alloc failed for {message_id}: {e}')
         return None
 
@@ -238,6 +264,8 @@ async def _stream_log_append(user_id: str, message_id: str, envelope: dict, seq:
     """Append envelope to resume log. Timeout drops the entry, not the emit."""
     if REDIS is None or not user_id or not message_id:
         return
+    if _breaker_open():
+        return
     try:
         refresh_ttl = (seq == 1) or (seq % RESUME_STREAM_TTL_REFRESH_EVERY == 0)
         key = _stream_key(user_id, message_id)
@@ -258,9 +286,12 @@ async def _stream_log_append(user_id: str, message_id: str, envelope: dict, seq:
         await asyncio.wait_for(
             pipe.execute(), timeout=RESUME_STREAM_REDIS_TIMEOUT_SEC
         )
+        _breaker_record_success()
     except asyncio.TimeoutError:
+        _breaker_record_failure()
         log.warning(f'stream resume log append timed out for {message_id}')
     except Exception as e:
+        _breaker_record_failure()
         log.warning(f'stream resume log append failed for {message_id}: {e}')
 
 

src/lib/components/chat/Chat.svelte

@@ -648,12 +648,12 @@
 	};
 
 	// Drop fence AND flush buffered events (happy path + timeout).
-	// Drains via shift() and keeps the queue live in the map while
-	// draining: chatEventHandler awaits `tick()`, yielding control.
-	// A live frame arriving during that yield must keep buffering (so
-	// it stays ordered after still-queued earlier frames); otherwise
-	// it would bypass the queue, advance lastSeq, and cause the queued
-	// frames to be dropped by the dedupe guard when we get back to them.
+	// Drain via shift(), keeping the queue present in the map so live
+	// frames arriving during an await inside chatEventHandler keep
+	// buffering into the same queue instead of racing ahead. Flushed
+	// events are marked `_replayed` before dispatch so chatEventHandler's
+	// fence-buffering branch short-circuits and doesn't re-queue them
+	// (which would be an infinite loop).
 	const clearResumeFence = async (messageId) => {
 		const timer = resumeFenceTimerByMessageId.get(messageId);
 		if (timer) {
@@ -664,6 +664,9 @@
 		if (!queue) return;
 		while (queue.length > 0) {
 			const event = queue.shift();
+			if (event && typeof event === 'object') {
+				event._replayed = true;
+			}
 			try {
 				await chatEventHandler(event);
 			} catch (e) {