mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-16 23:43:03 +00:00
fix(stream): close re-entrancy race in fence flush
clearResumeFence previously deleted the queue from the map before iterating it via for-of over the detached reference. Because chatEventHandler awaits tick() internally and yields control, a live frame arriving during that yield would find no queue in the map, bypass buffering entirely, advance lastSeq via the dedupe guard, and cause any still-pending earlier-seq frame in the flush iteration to be dropped on its next chatEventHandler call. Switch to a while+shift drain that keeps the queue present in the map for the whole flush. Frames arriving during any yield inside chatEventHandler push into the same queue the while loop is draining, so they stay ordered after earlier frames. The map entry is deleted only after the queue has been fully drained — between the final length check and the delete there is no await, so no new frame can slip in and be lost. Also moves the resumeActiveRequestIdByMessageId cleanup here so the request-id map doesn't accumulate stale entries over long-lived sessions. Not addressed: suggestion to add explicit chat_id validation to the resume endpoint. The per-user key scoping already enforces that a session can only ever read its own logs, and adding a DB chat-ownership check would re-introduce the "stub not yet persisted" failure mode we removed earlier. Deferring as defense-in-depth that doesn't correspond to a real threat under the current key design.
This commit is contained in:
parent
9af2b478bb
commit
0782e53d43
1 changed files with 10 additions and 5 deletions
|
|
@ -648,9 +648,12 @@
|
|||
};
|
||||
|
||||
// Drop fence AND flush buffered events (happy path + timeout).
|
||||
// Flush in insertion order: live frames were emitted sequentially by
|
||||
// one emitter and preserved by socket.io, so push-order == seq-order
|
||||
// and no sort is needed (mixing with seq-less frames would reorder).
|
||||
// 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.
|
||||
const clearResumeFence = async (messageId) => {
|
||||
const timer = resumeFenceTimerByMessageId.get(messageId);
|
||||
if (timer) {
|
||||
|
|
@ -659,14 +662,16 @@
|
|||
}
|
||||
const queue = resumeQueueByMessageId.get(messageId);
|
||||
if (!queue) return;
|
||||
resumeQueueByMessageId.delete(messageId);
|
||||
for (const event of queue) {
|
||||
while (queue.length > 0) {
|
||||
const event = queue.shift();
|
||||
try {
|
||||
await chatEventHandler(event);
|
||||
} catch (e) {
|
||||
console.error('resume fence flush error', e);
|
||||
}
|
||||
}
|
||||
resumeQueueByMessageId.delete(messageId);
|
||||
resumeActiveRequestIdByMessageId.delete(messageId);
|
||||
};
|
||||
|
||||
const requestResumeForMessage = (message) => {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue