mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-16 23:43:03 +00:00
fix(stream): message-in-chat check and don't stack resume requests
- Added message-in-chat binding check alongside the existing chat-
ownership check. Both use the same already-fetched chat object, so
this is still one DB round-trip. Defense-in-depth for the case
where a client supplies one of its own chat_ids but a different
chat's message_id — the user-scoped key would still serve the
request otherwise.
- requestResumeForMessage now early-returns if a fence already
exists for the same message. Fixes two operational concerns:
* Mixed-version deployments where an older backend doesn't echo
request_id: previously every reconnect reset the fence timer
AND set a new request_id, so no reply could ever match; now
the first fence runs to completion or timeout and only
subsequent requests after that get a fresh chance.
* Rapid reconnect churn: used to kick the timer forward
indefinitely, now one fence per message lifecycle.
This commit is contained in:
parent
418ef25dc4
commit
72429ea8c0
2 changed files with 26 additions and 13 deletions
|
|
@ -759,21 +759,30 @@ async def resume_stream(sid, data):
|
||||||
last_seq = int(data.get('last_seq') or 0)
|
last_seq = int(data.get('last_seq') or 0)
|
||||||
except (TypeError, ValueError):
|
except (TypeError, ValueError):
|
||||||
last_seq = 0
|
last_seq = 0
|
||||||
# Defense-in-depth: validate chat ownership even though the log
|
# Defense-in-depth: validate chat ownership AND message-in-chat
|
||||||
# key is already user-scoped. Fails CLOSED on DB errors so an
|
# binding even though the log key is already user-scoped. Fails
|
||||||
# infra hiccup can't skip the ownership check. When chat_id is
|
# CLOSED on DB errors so an infra hiccup can't skip the check.
|
||||||
# absent we fall through to the user-scoped key guarantee so
|
# When chat_id is absent we fall through to the user-scoped key
|
||||||
# legacy clients that don't send chat_id still work. Either way
|
# guarantee so legacy clients that don't send chat_id still work.
|
||||||
# we reply with (at worst) empty envelopes below so the client
|
# Either way we reply with (at worst) empty envelopes below so
|
||||||
# fence clears deterministically.
|
# the client fence clears deterministically.
|
||||||
chat_id = data.get('chat_id')
|
chat_id = data.get('chat_id')
|
||||||
chat_ok = True
|
chat_ok = True
|
||||||
if chat_id:
|
if chat_id:
|
||||||
try:
|
try:
|
||||||
chat = await Chats.get_chat_by_id_and_user_id(chat_id, user_id)
|
chat = await Chats.get_chat_by_id_and_user_id(chat_id, user_id)
|
||||||
chat_ok = chat is not None
|
if not chat:
|
||||||
|
chat_ok = False
|
||||||
|
else:
|
||||||
|
# Reuse the chat record for the message-in-chat check
|
||||||
|
# so we don't pay a second DB round-trip.
|
||||||
|
messages = (
|
||||||
|
getattr(chat, 'chat', {}) or {}
|
||||||
|
).get('history', {}).get('messages', {}) or {}
|
||||||
|
if message_id not in messages:
|
||||||
|
chat_ok = False
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
log.warning(f'resume-stream chat ownership check failed: {e}')
|
log.warning(f'resume-stream chat/message check failed: {e}')
|
||||||
chat_ok = False
|
chat_ok = False
|
||||||
if chat_ok:
|
if chat_ok:
|
||||||
envelopes = await _stream_log_read(user_id, message_id, last_seq)
|
envelopes = await _stream_log_read(user_id, message_id, last_seq)
|
||||||
|
|
|
||||||
|
|
@ -687,10 +687,14 @@
|
||||||
const requestResumeForMessage = (message) => {
|
const requestResumeForMessage = (message) => {
|
||||||
if (!message || !message.id || message.done) return;
|
if (!message || !message.id || message.done) return;
|
||||||
if (!$socket || !$socket.connected) return;
|
if (!$socket || !$socket.connected) return;
|
||||||
// Fence up before emit so racing live frames get buffered.
|
// Don't stack resume requests. If a fence already exists for
|
||||||
if (!resumeQueueByMessageId.has(message.id)) {
|
// this message, an earlier request is in flight (or will be
|
||||||
resumeQueueByMessageId.set(message.id, []);
|
// resolved by its fallback timer). Stacking would reset the
|
||||||
}
|
// timer on every reconnect and, under mixed-version deployments
|
||||||
|
// where an old backend doesn't echo request_id, leave the fence
|
||||||
|
// stuck with request_ids that never match.
|
||||||
|
if (resumeQueueByMessageId.has(message.id)) return;
|
||||||
|
resumeQueueByMessageId.set(message.id, []);
|
||||||
const existingTimer = resumeFenceTimerByMessageId.get(message.id);
|
const existingTimer = resumeFenceTimerByMessageId.get(message.id);
|
||||||
if (existingTimer) clearTimeout(existingTimer);
|
if (existingTimer) clearTimeout(existingTimer);
|
||||||
resumeFenceTimerByMessageId.set(
|
resumeFenceTimerByMessageId.set(
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue