fix(stream): sort replay by seq, skip log in REALTIME mode, strict flush order

Critical: INCR and XADD are separate calls, so with overlapping
emitters the stream's append order can diverge from seq order (a
concurrent emitter can win INCR for seq N+1 yet lose XADD and land
ahead of seq N in the stream). Replay returned entries in append
order, which meant the later-seq-but-earlier-in-stream frame would
advance lastSeq, permanently dropping the earlier-seq-but-later-in-
stream frame via the client dedupe guard. Fix: _stream_log_read now
sorts envelopes by seq before returning, making replay order
independent of append order.

Warning: loadChat now unconditionally triggers resume for unfinished
assistants with last_seq=0, which in ENABLE_REALTIME_CHAT_SAVE mode
(DB is kept up-to-date per token) would replay content already
loaded from the DB and produce duplicates. Gate _stream_seq_allocate
on `not ENABLE_REALTIME_CHAT_SAVE` — in realtime mode frames emit
without a seq, nothing is logged, client bypasses dedupe/resume, DB
stays authoritative for refresh recovery.

Suggestion: fence flush sort comparator returned 0 for mixed
seq/no-seq pairs, which is not strict ordering. Replaced with a
partition (seq-bearing vs. seq-less) + sort the seq-bearing subset +
concat. Deterministic regardless of engine sort stability.
This commit is contained in:
Claude 2026-04-14 22:31:02 +00:00
parent 572a5407bb
commit 06f4656f5e
No known key found for this signature in database
2 changed files with 32 additions and 12 deletions

View file

@ -25,6 +25,7 @@ from open_webui.config import (
from open_webui.env import (
VERSION,
ENABLE_REALTIME_CHAT_SAVE,
ENABLE_WEBSOCKET_SUPPORT,
WEBSOCKET_MANAGER,
WEBSOCKET_REDIS_URL,
@ -206,7 +207,15 @@ async def _stream_seq_allocate(user_id: str, message_id: str):
Returns the allocated seq, or None if Redis is unavailable or the
call times out. Callers treat None as "emit without seq" live
streaming continues but that frame isn't eligible for resume dedupe.
Also returns None when ENABLE_REALTIME_CHAT_SAVE is on: in that
mode the DB is already the authoritative per-token store, so the
frontend loads the up-to-date content on refresh. Logging the
resume stream in parallel would cause duplicate content on refresh
(resume replays from seq=0 on top of DB-loaded content).
"""
if ENABLE_REALTIME_CHAT_SAVE:
return None
if REDIS is None or not user_id or not message_id:
return None
try:
@ -306,10 +315,19 @@ async def _stream_log_read(user_id: str, message_id: str, after_seq: int):
if not payload:
continue
try:
out.append(json.loads(payload))
envelope = json.loads(payload)
except Exception:
continue
return out
out.append((entry_seq, envelope))
# Sort by seq before returning envelopes. Stream append order can
# diverge from seq order because INCR + XADD aren't atomic: a
# concurrent emitter can win the INCR race for a later seq yet lose
# the XADD race and land in the stream ahead of an earlier seq. If
# we handed envelopes to the client in append order, the client's
# `incomingSeq <= lastSeq` guard would permanently drop the later-
# arriving-but-earlier-numbered frame.
out.sort(key=lambda pair: pair[0])
return [envelope for _seq, envelope in out]
async def periodic_session_pool_cleanup():

View file

@ -673,17 +673,19 @@
const queue = resumeQueueByMessageId.get(messageId);
if (!queue) return;
resumeQueueByMessageId.delete(messageId);
// Stable sort: only reorder events that both carry a numeric seq.
// Seq-less events (Redis-down graceful-degradation case) keep
// their insertion order instead of being forced to the front,
// which would misorder state transitions like replace/done.
queue.sort((a, b) => {
const ah = typeof a?.seq === 'number';
const bh = typeof b?.seq === 'number';
if (!ah || !bh) return 0;
return a.seq - b.seq;
});
// Partition into seq-bearing and seq-less events so the sort
// comparator is a strict ordering on its domain (avoids the
// engine-dependent behavior of returning 0 for mixed pairs).
// Apply seq-ordered events first, then seq-less (graceful
// degradation frames) in insertion order.
const withSeq = [];
const withoutSeq = [];
for (const event of queue) {
if (typeof event?.seq === 'number') withSeq.push(event);
else withoutSeq.push(event);
}
withSeq.sort((a, b) => a.seq - b.seq);
for (const event of [...withSeq, ...withoutSeq]) {
try {
await chatEventHandler(event);
} catch (e) {