Skip to content

Commit cc49c01

Browse files
committed
fix(sdk): treat the read timeout as a total budget across skipped frames
chat.messages.next() and once() re-issued the wait with the caller's full timeout on every skipped invalid frame, so the timeout was per attempt rather than a total. A client sending invalid frames faster than the timeout kept the read blocked indefinitely and the documented return on timeout never fired. Both now compute a deadline once and pass the remaining time to each attempt, which also preserves the existing zero-timeout behaviour of draining only what is already buffered.
1 parent b70a880 commit cc49c01

1 file changed

Lines changed: 20 additions & 4 deletions

File tree

  • packages/trigger-sdk/src/v3

packages/trigger-sdk/src/v3/ai.ts

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1996,11 +1996,18 @@ const messagesInput: ChatMessages = {
19961996
},
19971997
once(options) {
19981998
return new InputStreamOncePromise<ChatTaskWirePayload>((resolve, reject) => {
1999-
// Same skip-and-wait rule as `waitWithIdleTimeout`: a payload that fails
2000-
// validation is reported and not surfaced.
1999+
/**
2000+
* Same skip-and-wait rule as `waitWithIdleTimeout`: a payload that fails
2001+
* validation is reported and not surfaced. The timeout is a total budget
2002+
* across retries, so skipping a frame cannot extend the wait forever.
2003+
*/
2004+
const deadline =
2005+
options?.timeoutMs === undefined ? undefined : Date.now() + options.timeoutMs;
20012006
const take = () => {
20022007
chatInputRouter()
2003-
.next(CHAT_ROUTE_MESSAGES, { timeoutMs: options?.timeoutMs })
2008+
.next(CHAT_ROUTE_MESSAGES, {
2009+
timeoutMs: deadline === undefined ? undefined : Math.max(0, deadline - Date.now()),
2010+
})
20042011
.then(async (record) => {
20052012
if (!record) {
20062013
resolve({
@@ -2055,9 +2062,18 @@ const messagesInput: ChatMessages = {
20552062
// Consuming read, so it takes the same claim-and-validate path as the other
20562063
// reads: a record the observer still owns is put back and awaited, and an
20572064
// invalid payload is reported and skipped rather than surfaced raw.
2065+
const totalMs = timeoutInSeconds === undefined ? undefined : timeoutInSeconds * 1000;
2066+
/**
2067+
* The caller's timeout is a total budget, not a per-attempt one. Skipping an
2068+
* invalid frame must not buy another full wait, or a client sending invalid
2069+
* frames faster than the timeout would keep the read blocked indefinitely
2070+
* and it would never return.
2071+
*/
2072+
const deadline = totalMs === undefined ? undefined : Date.now() + totalMs;
2073+
20582074
while (true) {
20592075
const record = await chatInputRouter().next(CHAT_ROUTE_MESSAGES, {
2060-
timeoutMs: timeoutInSeconds === undefined ? undefined : timeoutInSeconds * 1000,
2076+
timeoutMs: deadline === undefined ? undefined : Math.max(0, deadline - Date.now()),
20612077
});
20622078
if (!record) return undefined;
20632079

0 commit comments

Comments
 (0)