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:
Claude 2026-04-15 07:53:35 +00:00
parent 418ef25dc4
commit 72429ea8c0
No known key found for this signature in database
2 changed files with 26 additions and 13 deletions

View file

@ -759,21 +759,30 @@ async def resume_stream(sid, data):
last_seq = int(data.get('last_seq') or 0)
except (TypeError, ValueError):
last_seq = 0
# Defense-in-depth: validate chat ownership even though the log
# key is already user-scoped. Fails CLOSED on DB errors so an
# infra hiccup can't skip the ownership check. When chat_id is
# absent we fall through to the user-scoped key guarantee so
# legacy clients that don't send chat_id still work. Either way
# we reply with (at worst) empty envelopes below so the client
# fence clears deterministically.
# Defense-in-depth: validate chat ownership AND message-in-chat
# binding even though the log key is already user-scoped. Fails
# CLOSED on DB errors so an infra hiccup can't skip the check.
# When chat_id is absent we fall through to the user-scoped key
# guarantee so legacy clients that don't send chat_id still work.
# Either way we reply with (at worst) empty envelopes below so
# the client fence clears deterministically.
chat_id = data.get('chat_id')
chat_ok = True
if chat_id:
try:
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:
log.warning(f'resume-stream chat ownership check failed: {e}')
log.warning(f'resume-stream chat/message check failed: {e}')
chat_ok = False
if chat_ok:
envelopes = await _stream_log_read(user_id, message_id, last_seq)

View file

@ -687,10 +687,14 @@
const requestResumeForMessage = (message) => {
if (!message || !message.id || message.done) return;
if (!$socket || !$socket.connected) return;
// Fence up before emit so racing live frames get buffered.
if (!resumeQueueByMessageId.has(message.id)) {
resumeQueueByMessageId.set(message.id, []);
}
// Don't stack resume requests. If a fence already exists for
// this message, an earlier request is in flight (or will be
// 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);
if (existingTimer) clearTimeout(existingTimer);
resumeFenceTimerByMessageId.set(