scriptoriumadmin/open-webui
1
0
Fork 0
mirror of https://github.com/open-webui/open-webui.git synced 2026-09-16 23:43:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix(stream): commit seq after handler success, halve fence fallback

Browse source 
Warning: lastSeq was advanced immediately after the dedupe check,
before the event handler ran. If a handler threw (e.g., chat:tags
fetch failure), the seq was already marked consumed and a subsequent
replay would skip it — silent data loss for side-channel events.

Move the seq commit to after the handler body completes. If the
handler throws, control exits via the exception and the commit never
runs, so the next replay resends that seq. Use max() against the
current stored value so a concurrent higher-seq handler that
committed during our awaits isn't clobbered.

Warning: fence fallback timeout was 10s, which is a visible live-
stream stall when a peer backend silently drops the resume-stream
request. Halved to 5s — still longer than the 1s XRANGE read timeout
(legitimate slow replays complete), but short enough that a missing
ack from a mixed-version peer unfreezes the UI sooner.
This commit is contained in:
Claude 2026-04-15 09:18:45 +00:00
parent 0a084cbe0c
commit c1c971ee47
No known key found for this signature in database
1 changed files with 22 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
src/lib/components/chat/Chat.svelte
View file

@ -464,14 +464,17 @@
return;
}
// Dedupe.
// Dedupe check only — advance lastSeq at the end of the
// handler, not here. If a downstream handler throws (e.g.,
// chat:tags fetch fails), leaving lastSeq unchanged lets
// a subsequent replay retry that seq instead of silently
// skipping it.
const incomingSeq = typeof event?.seq === 'number' ? event.seq : null;
if (incomingSeq !== null) {
const lastSeq = resumeSeqByMessageId.get(event.message_id) ?? 0;
if (incomingSeq <= lastSeq) {
return;
}
resumeSeqByMessageId.set(event.message_id, incomingSeq);
}
const type = event?.data?.type ?? null;
@ -628,6 +631,17 @@
}
history.messages[event.message_id] = message;
// Handler completed without throwing — safe to commit the
// seq now. Use max() against the current value so a
// concurrent higher-seq handler that committed during our
// awaits doesn't get clobbered.
if (incomingSeq !== null) {
const current = resumeSeqByMessageId.get(event.message_id) ?? 0;
if (incomingSeq > current) {
resumeSeqByMessageId.set(event.message_id, incomingSeq);
}
}
}
} else {
// Non-active chat completion: queue stays in the global store.
@ -635,8 +649,12 @@
}
};
// Fallback timer in case the replay ack never arrives.
const RESUME_FENCE_TIMEOUT_MS = 10000;
// Fallback timer in case the replay ack never arrives. 5s balances:
// a legitimately slow Redis XRANGE at RESUME_STREAM_READ_TIMEOUT_SEC
// (1s default), versus how long the UI should hold live frames if
// a peer backend silently drops the request (mixed-version rollout,
// handler not registered, etc).
const RESUME_FENCE_TIMEOUT_MS = 5000;
const resumeFenceTimerByMessageId = new Map();
// Active request_id per message — lets us ignore stale replay
// responses from superseded requests (reconnect churn).