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) 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)

View file

@ -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(