Threads 17: Stop newly provisioned thread runs - #246
Conversation
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Sorry @FranciscoMoretti, you have reached your weekly rate limit of 500000 diff characters.
Please try again later or upgrade to continue using Sourcery
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Greptile SummaryThis PR makes
Confidence Score: 3/5Not safe to merge: the credits-deduction gap and the shared deadline/stoppedAt timing issues flagged in prior rounds remain unresolved in the current diff. Three concrete correctness problems from earlier review rounds are still visible in the current code: (1) deductCredits runs unconditionally in finalizeMessageAndCredits even when messageSaved is false, so a user who stops a stream is still charged in full; (2) all three polling loops share a single 30-second deadline with a shared retryDelay, so a chat that takes time to provision consumes the budget meant for the message and response loops; (3) canceledAt is captured after the chat-provisioning loop rather than at click time, so fast models that finish during provisioning fall outside the getCancelableResponseMessage fallback window. The DB transaction and isNull(canceledAt) guard on updateMessage are solid improvements, and the client-side chatId forwarding is correct. Files Needing Attention: apps/chat/trpc/routers/chat.router.ts (deadline sharing, stoppedAt timing) and apps/chat/app/(chat)/api/chat/route.ts (unconditional credit deduction after canceled finalization) Important Files Changed
Sequence DiagramsequenceDiagram
participant U as User (Browser)
participant MI as MultimodalInput
participant TR as stopStream TRPC
participant DB as Database
U->>MI: Click Stop
MI->>TR: "mutate({ chatId, messageId })"
TR->>DB: getChatById(chatId)
alt chat not yet provisioned
loop up to 30s exp backoff
DB-->>TR: null
TR->>DB: getChatById(chatId)
end
end
DB-->>TR: chat row
TR->>TR: "capture canceledAt = new Date()"
TR->>DB: getMessageById(messageId)
alt message not yet provisioned
loop shared deadline/retryDelay
DB-->>TR: null
TR->>DB: getMessageById(messageId)
end
end
DB-->>TR: targetMessage
alt "targetMessage.role === assistant"
TR->>TR: "responseMessageId = messageId"
else "targetMessage.role === user"
loop shared deadline/retryDelay
TR->>DB: getCancelableResponseMessage
Note over DB: Q1 activeStreamId IS NOT NULL
Note over DB: Q2 fallback createdAt >= stoppedAt
DB-->>TR: response or null
end
end
TR->>DB: updateMessageCanceledAt(responseMessageId)
Note over DB: Tx: SET activeStreamId=null canceledAt=T
Note over DB: DELETE parts WHERE messageId=?
DB-->>TR: true
TR-->>MI: success
Note over MI: route.ts finalizeMessageAndCredits
MI->>DB: updateMessage skipped if canceledAt set
DB-->>MI: "messageSaved=false"
MI->>DB: deductCredits runs unconditionally
Reviews (37): Last reviewed commit: "fix(chat): make response cancellation te..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| while ( | ||
| !(await updateMessageCanceledAt({ | ||
| messageId: input.messageId, | ||
| canceledAt, | ||
| })) | ||
| ) { | ||
| if (Date.now() >= deadline) { | ||
| throw new TRPCError({ | ||
| code: "NOT_FOUND", | ||
| message: "Message not found", | ||
| }); | ||
| } | ||
|
|
||
| await new Promise((resolve) => setTimeout(resolve, retryDelay)); | ||
| retryDelay = Math.min(retryDelay * 2, 250); | ||
| } |
There was a problem hiding this comment.
The new flow validates that input.chatId belongs to the current user, but updateMessageCanceledAt only filters by eq(message.id, messageId) — it never checks that the target message actually belongs to input.chatId. A user who owns any valid chat can therefore craft a request with their own chatId and a messageId from a different user's chat and successfully set canceledAt on that foreign message, effectively stopping another user's stream without permission. The fix is to add an eq(message.chatId, chatId) predicate to the updateMessageCanceledAt query (or validate msg.chatId === input.chatId before updating).
| const canceledAt = new Date(); | ||
|
|
||
| while ( |
There was a problem hiding this comment.
retryDelay is shared across both polling loops, so the message loop starts at whatever value the chat loop left it at. If the chat loop ran several iterations and bumped retryDelay to 250 ms, the message polling loop starts at 250 ms instead of 25 ms. Resetting retryDelay between the two loops preserves the intended exponential-backoff shape for each phase.
| const canceledAt = new Date(); | |
| while ( | |
| const canceledAt = new Date(); | |
| retryDelay = 25; | |
| while ( |
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 26f76d0322
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| !(await updateMessageCanceledAt({ | ||
| messageId: input.messageId, | ||
| canceledAt, | ||
| })) |
There was a problem hiding this comment.
When a caller supplies an owned chatId together with a message ID from another chat, the ownership check validates only the supplied chat, while this update still filters solely by messageId. The mutation can therefore set canceledAt on another user's active message if its ID is known; include the authorized chat ID in the update predicate or verify the message belongs to that chat.
Useful? React with 👍 / 👎.
| while (!chat && Date.now() < deadline) { | ||
| await new Promise((resolve) => setTimeout(resolve, retryDelay)); | ||
| retryDelay = Math.min(retryDelay * 2, 250); | ||
| chat = await getChatById({ id: input.chatId }); |
There was a problem hiding this comment.
For any nonexistent UUID, every authenticated call now holds a request open for 30 seconds and performs roughly 120 database reads before returning NOT_FOUND; protectedProcedure only authenticates and adds no rate limit. A caller can cheaply amplify concurrent requests into sustained database and serverless-invocation load, so the provisioning race needs a bounded reservation mechanism that distinguishes legitimate pending chats rather than polling every supplied ID.
Useful? React with 👍 / 👎.
| !(await updateMessageCanceledAt({ | ||
| messageId: input.messageId, | ||
| canceledAt, | ||
| })) |
There was a problem hiding this comment.
When this update wins the provisioning race and stores canceledAt before the chat request starts its model stream, the request does not read that value until the onChunk callback in api/chat/route.ts. The provider request is therefore still launched and can run or incur cost until its first chunk arrives, which can be significantly delayed for slow or reasoning-heavy models; check the persisted flag and abort before creating the model stream.
Useful? React with 👍 / 👎.
26f76d0 to
dc8e43d
Compare
229aacd to
8a210bb
Compare
8a210bb to
38f495c
Compare
8fefd93 to
b001092
Compare
b001092 to
7f878e8
Compare
7f878e8 to
4ebf788
Compare
| const updatedMessages = await db.transaction(async (tx) => { | ||
| const updated = await tx | ||
| .update(message) | ||
| .set({ activeStreamId: null, canceledAt }) | ||
| .where(eq(message.id, messageId)) | ||
| .returning({ id: message.id }); | ||
|
|
||
| if (updated.length > 0 && canceledAt !== null) { | ||
| await tx.delete(part).where(eq(part.messageId, messageId)); | ||
| } |
There was a problem hiding this comment.
updateMessageCanceledAt has no guard on isNull(message.canceledAt) and no check on whether the message was already finalized. If finalizeMessageAndCredits → updateMessage commits parts (and clears activeStreamId) just before updateMessageCanceledAt runs, the transaction will match the existing row (since WHERE id = messageId always succeeds), delete the newly-saved parts, and return true. The user is charged the full model cost (since deductCredits executes regardless of messageSaved) but ends up with a canceled message that has no content.
The stop-stream flow lands here via the getCancelableResponseMessage fallback query (gte(createdAt, stoppedAt)), which matches already-finalized messages as long as they were created after stoppedAt. Since stoppedAt is recorded at the very start of the handler — before any DB work — any response message created while the stop request was in flight satisfies that predicate, including ones whose stream has already completed. Adding a guard such as isNotNull(message.activeStreamId) before deleting parts would limit destructive cancellation to messages that are still actively streaming.
| const canceledAt = new Date(); | ||
| let [targetMessage] = await getMessageById({ id: input.messageId }); |
There was a problem hiding this comment.
canceledAt is captured after the chat-provisioning loop completes, not when the user clicks Stop. It is then passed as stoppedAt to getCancelableResponseMessage, which uses gte(message.createdAt, stoppedAt) as the fallback to catch responses that finished streaming before activeStreamId could be observed. If provisioning takes, say, 3 seconds and a fast model finishes within that window, the response message will have activeStreamId = null (stream done) and createdAt < canceledAt — both queries miss it, the loop exits with responseMessageId = null, and the stop is silently dropped. Moving the timestamp capture to the very top of the handler makes stoppedAt anchor to when the user actually clicked Stop and eliminates the gap.
| const canceledAt = new Date(); | |
| let [targetMessage] = await getMessageById({ id: input.messageId }); | |
| let [targetMessage] = await getMessageById({ id: input.messageId }); |
| const updatedMessages = await db.transaction(async (tx) => { | ||
| const updated = await tx | ||
| .update(message) | ||
| .set({ activeStreamId: null, canceledAt }) | ||
| .where(eq(message.id, messageId)) | ||
| .returning({ id: message.id }); | ||
|
|
||
| if (updated.length > 0 && canceledAt !== null) { | ||
| await tx.delete(part).where(eq(part.messageId, messageId)); | ||
| } |
There was a problem hiding this comment.
updateMessageCanceledAt sets canceledAt and deletes parts for any row matching only id, with no isNull(message.canceledAt) guard. If this function is called after updateMessage has already committed the finalized content (parts saved, activeStreamId cleared), the transaction will still match, wipe the parts, and mark the row canceled. The user is then charged full credits by deductCredits (since messageSaved was true in the finalization path) but ends up with a canceled, empty message. Adding isNull(message.canceledAt) to the WHERE clause makes the update a no-op on already-finalized messages and prevents this double-write corruption.
1 participant
Summary
Behavior
Stop works during the provisioning window instead of being lost as a not-found race.
Verification
Review focus
Cancellation ownership, retry bounds, and database pressure of the short polling barrier.
Summary by cubic
Stops runs during new-thread provisioning and makes response cancellations terminal to prevent not-found races and wasted tokens.
Bug Fixes
stopStreamwaits (up to 30s) for the chat and target message, verifies ownership, scopes by{chatId, messageId}, resolves the response viagetCancelableResponseMessage(active or created after stop), and retriesupdateMessageCanceledAtuntil it succeeds.updateMessageCanceledAtclearsactiveStreamId, setscanceledAt, deletes parts in a transaction, and returns a boolean;updateMessageonly updates uncanceled rows and returns a boolean; finalization logs and skips canceled assistant messages.Migration
stopStreamnow requireschatId;multimodal-inputsends{chatId, messageId}. Update remaining callers.Written for commit 17dc851. Summary will update on new commits.
Stack