Skip to content

Commit 34c2d69

Browse files
gtrempergtremper
authored
fix: stop retrying silent chat streams indefinitely (#4949)
When the server returns `200` without records, chat subscriptions can retry indefinitely. The client stall timer fires after 60 seconds, before S2 closes the response at its 120-second timeout, so the existing EOF budget never applies. Terminal failures also leave persisted `isStreaming` state active, so a reload starts another subscription. Normal chat subscriptions now permit five reconnects after stall timeouts. A separate stall counter preserves unlimited retries for retryable connection failures, fetch timeouts, and browser wakeups. Decoded records restore the stall budget. Terminal failures clear state only for the owning subscription, and watch subscriptions remain unlimited. The existing stall timer already ignores keepalives because the parser drops them before the timer reset. The comment correction does not change that behavior. ## Checklist - [x] The PR title follows the contribution convention. - [x] The changes include tests and a changeset. - [x] Local package builds, formatting, lint, and knip passed. - [ ] Maintainer CI and reference-project validation. ## Testing - Regression tests reproduce timeout exhaustion, indefinite stalls, stale terminal state, and recovery beyond five connection failures. - Full suites at `8149b96`: 695 SDK tests and 1,122 core tests passed. - Final error-message and constructor changes: all 43 stream tests passed. - Core and SDK builds passed. - Repository formatting, lint, and knip passed. - General and security review passes found no remaining defects. - The debug-marker check reports existing markers in `apps/webapp/app/services/previewAutoArchive.server.ts`; the changed files contain none. The tests use local HTTP servers. They cover separate stall limits, fetch timeouts, body failures, keepalives, progress resets, mixed failures, wakeups, cancellation, watch recovery, and replacement ownership. Core tests exercise stall exhaustion with short timers. The SDK's six-minute silence window needs reference-project validation. ## Changelog Silent chat subscriptions now report `Stream stalled: no records received` after five stall retries. Network failures retain automatic recovery, and watch subscriptions remain unlimited. ## Risk | Condition | Result | | --- | --- | | Retryable connection or fetch failure | No finite retry deadline; exponential backoff continues | | Repeated connected silence | The sixth 60-second stall ends the subscription | | Browser wake or `online` event | Reconnect without consuming or restoring the stall budget | With immediate response headers, six silent attempts take 368.5–377 seconds, about 6.1–6.3 minutes. Network delays extend this window. A healthy tool call with no records can also reach this limit: silence does not prove that the run is dead. The terminal error stops automatic client resumption but does not cancel the server-side run. Shared core consumers retain their configured retry limits. Repeated successful responses without records now increase backoff until a decoded record arrives. Caller cancellation and token-refresh limits remain unchanged.
1 parent 8067b1f commit 34c2d69

5 files changed

Lines changed: 380 additions & 16 deletions

File tree

‎.changeset/quiet-chat-stream-retries.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
"@trigger.dev/core": patch
3+
"@trigger.dev/sdk": patch
4+
---
5+
6+
Chat streams now report `Stream stalled: no records received` after five retries of a connected stream that sends no records. Network failures and browser wakeups retain automatic recovery. Healthy tool calls with no records for about six minutes also reach this silence limit. Watch subscriptions remain unlimited, and caller cancellation still closes cleanly.

‎packages/core/src/v3/apiClient/runStream-retries.test.ts‎

Lines changed: 188 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,188 @@
1+
import { createServer, type Server, type ServerResponse } from "node:http";
2+
import { afterEach, beforeEach, describe, expect, it } from "vitest";
3+
import { SSEStreamSubscription } from "./runStream.js";
4+
5+
describe("SSE retry exhaustion", () => {
6+
let server: Server;
7+
let url: string;
8+
let abort: AbortController;
9+
let attempts: number;
10+
let respond: (response: ServerResponse) => void;
11+
let subscription: SSEStreamSubscription;
12+
13+
beforeEach(async () => {
14+
attempts = 0;
15+
abort = new AbortController();
16+
server = createServer((_request, response) => {
17+
attempts++;
18+
respond(response);
19+
});
20+
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
21+
const address = server.address();
22+
if (!address || typeof address === "string") throw new Error("Expected a TCP address");
23+
url = `http://127.0.0.1:${address.port}`;
24+
});
25+
26+
afterEach(async () => {
27+
abort.abort();
28+
server.closeAllConnections();
29+
await new Promise<void>((resolve) => server.close(() => resolve()));
30+
});
31+
32+
async function open(
33+
options: {
34+
fetchTimeoutMs?: number;
35+
stallTimeoutMs?: number;
36+
maxRetries?: number;
37+
maxStallRetries?: number;
38+
} = {}
39+
) {
40+
subscription = new SSEStreamSubscription(url, {
41+
signal: abort.signal,
42+
maxRetries: 2,
43+
retryDelayMs: 1,
44+
retryJitter: 0,
45+
...options,
46+
});
47+
return (await subscription.subscribe()).getReader();
48+
}
49+
50+
it.each(["fetch", "stall"] as const)(
51+
"reports exhausted %s timeouts as failures",
52+
async (failure) => {
53+
respond = (response) => {
54+
if (failure === "stall") {
55+
response.writeHead(200, { "Content-Type": "text/event-stream" });
56+
response.flushHeaders();
57+
}
58+
};
59+
const reader = await open({
60+
fetchTimeoutMs: failure === "fetch" ? 100 : 1_000,
61+
stallTimeoutMs: 100,
62+
});
63+
64+
await expect(reader.read()).rejects.toMatchObject({
65+
name: "Error",
66+
message: "Stream connection retries exhausted",
67+
});
68+
expect(attempts).toBe(3);
69+
}
70+
);
71+
72+
it.each([
73+
["comment", ": keepalive\n\n"],
74+
["keepalive event", "event: keepalive\ndata: {}\n\n"],
75+
["empty batch", 'event: batch\ndata: {"records":[]}\n\n'],
76+
])("does not reset the retry budget after a %s", async (_name, payload) => {
77+
respond = (response) => {
78+
response.writeHead(200, {
79+
"Content-Type": "text/event-stream",
80+
"X-Stream-Version": "v2",
81+
});
82+
response.write(payload);
83+
};
84+
const reader = await open({ stallTimeoutMs: 100, maxRetries: Infinity, maxStallRetries: 2 });
85+
86+
await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
87+
expect(attempts).toBe(3);
88+
});
89+
90+
it("restores the retry budget after a decoded record", async () => {
91+
respond = (response) => {
92+
if (attempts !== 3) {
93+
response.writeHead(503).end();
94+
return;
95+
}
96+
response.writeHead(200, { "Content-Type": "text/event-stream" });
97+
response.write('id: 1\ndata: {"hello":1}\n\n');
98+
};
99+
const reader = await open({ stallTimeoutMs: 100 });
100+
101+
expect(await reader.read()).toMatchObject({ done: false, value: { chunk: { hello: 1 } } });
102+
await expect(reader.read()).rejects.toMatchObject({ status: 503 });
103+
expect(attempts).toBe(5);
104+
});
105+
106+
it("closes without retries when the caller cancels", async () => {
107+
respond = () => abort.abort();
108+
const reader = await open();
109+
110+
expect(await reader.read()).toEqual({ done: true, value: undefined });
111+
expect(attempts).toBe(1);
112+
});
113+
114+
it("limits silent stalls without a general retry limit", async () => {
115+
respond = (response) => {
116+
response.writeHead(200, { "Content-Type": "text/event-stream" });
117+
response.flushHeaders();
118+
};
119+
const reader = await open({ stallTimeoutMs: 100, maxRetries: Infinity, maxStallRetries: 2 });
120+
121+
await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
122+
expect(attempts).toBe(3);
123+
});
124+
125+
it("restores the stall budget only after a decoded record", async () => {
126+
respond = (response) => {
127+
response.writeHead(200, { "Content-Type": "text/event-stream" });
128+
response.flushHeaders();
129+
if (attempts === 3) response.write('id: 1\ndata: {"hello":1}\n\n');
130+
};
131+
const reader = await open({ stallTimeoutMs: 100, maxRetries: Infinity, maxStallRetries: 2 });
132+
133+
expect(await reader.read()).toMatchObject({ done: false, value: { chunk: { hello: 1 } } });
134+
await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
135+
expect(attempts).toBe(5);
136+
});
137+
138+
it.each(["http", "fetch", "body", "wake"] as const)(
139+
"does not charge %s failures to the stall budget",
140+
async (failure) => {
141+
respond = (response) => {
142+
if (attempts === 5) {
143+
response.writeHead(200, { "Content-Type": "text/event-stream" });
144+
response.end('id: 1\ndata: {"hello":1}\n\n');
145+
} else if (failure === "http") {
146+
response.writeHead(503).end();
147+
} else if (failure !== "fetch") {
148+
response.writeHead(200, { "Content-Type": "text/event-stream" });
149+
response.write(": keepalive\n\n");
150+
setTimeout(() => {
151+
if (failure === "wake") subscription.forceReconnect();
152+
else response.destroy();
153+
}, 10);
154+
}
155+
};
156+
const reader = await open({
157+
maxRetries: Infinity,
158+
maxStallRetries: 0,
159+
fetchTimeoutMs: 100,
160+
stallTimeoutMs: 1_000,
161+
});
162+
163+
expect(await reader.read()).toMatchObject({ done: false, value: { chunk: { hello: 1 } } });
164+
expect(attempts).toBe(5);
165+
}
166+
);
167+
168+
it("retains the stall budget across connection failures and wakeups", async () => {
169+
respond = (response) => {
170+
if (attempts === 2) {
171+
response.writeHead(503).end();
172+
} else if (attempts !== 4) {
173+
response.writeHead(200, { "Content-Type": "text/event-stream" });
174+
response.flushHeaders();
175+
if (attempts === 3) setTimeout(() => subscription.forceReconnect(), 10);
176+
}
177+
};
178+
const reader = await open({
179+
maxRetries: Infinity,
180+
maxStallRetries: 1,
181+
fetchTimeoutMs: 100,
182+
stallTimeoutMs: 100,
183+
});
184+
185+
await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
186+
expect(attempts).toBe(5);
187+
});
188+
});

‎packages/core/src/v3/apiClient/runStream.ts‎

Lines changed: 24 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -219,7 +219,9 @@ export class SSEStreamSubscription implements StreamSubscription {
219219
private lastEventId: string | undefined;
220220
private from: "beginning" | "latest";
221221
private retryCount = 0;
222+
private stallCount = 0;
222223
private maxRetries: number;
224+
private maxStallRetries: number;
223225
private retryDelayMs: number;
224226
private maxRetryDelayMs: number;
225227
private retryJitter: number;
@@ -273,9 +275,11 @@ export class SSEStreamSubscription implements StreamSubscription {
273275
// the connection is established, force a reconnect. Catches
274276
// silent-dead-socket cases (mobile OS killed the TCP socket but
275277
// the read just blocks). Disabled (`0`) by default; opt in
276-
// explicitly. Servers that emit periodic keepalive comments
277-
// reset the timer naturally.
278+
// explicitly. Only decoded records reset the timer.
278279
stallTimeoutMs?: number;
280+
// Reconnects after stall timeouts before the stream errors.
281+
// Only decoded records restore this budget. Defaults to Infinity.
282+
maxStallRetries?: number;
279283
// HTTP statuses that should NOT be retried — fail the stream
280284
// permanently. Defaults cover the permanent client-error set:
281285
// `400` (bad request), `404` (stream gone), `409` (conflict),
@@ -293,6 +297,7 @@ export class SSEStreamSubscription implements StreamSubscription {
293297
this.lastEventId = options.lastEventId;
294298
this.from = options.from ?? "beginning";
295299
this.maxRetries = options.maxRetries ?? Infinity;
300+
this.maxStallRetries = options.maxStallRetries ?? Infinity;
296301
this.retryDelayMs = options.retryDelayMs ?? 100;
297302
this.maxRetryDelayMs = options.maxRetryDelayMs ?? 5000;
298303
this.retryJitter = options.retryJitter ?? 0.5;
@@ -403,7 +408,11 @@ export class SSEStreamSubscription implements StreamSubscription {
403408
const armStall = () => {
404409
if (this.stallTimeoutMs <= 0) return;
405410
clearTimeout(stallTimer);
406-
stallTimer = setTimeout(() => this.internalAbort?.abort(), this.stallTimeoutMs);
411+
stallTimer = setTimeout(() => {
412+
if (!this.internalAbort || this.internalAbort.signal.aborted) return;
413+
this.stallCount++;
414+
this.internalAbort.abort();
415+
}, this.stallTimeoutMs);
407416
};
408417

409418
// Idempotent — both the catch (before recursion) and the finally
@@ -461,7 +470,6 @@ export class SSEStreamSubscription implements StreamSubscription {
461470

462471
const streamVersion = response.headers.get("X-Stream-Version") ?? "v1";
463472
this.sessionSettled = response.headers.get("X-Session-Settled") === "true";
464-
this.retryCount = 0; // reset on success
465473
armStall();
466474

467475
// Dedup window for record ids. Bounded with FIFO eviction so a
@@ -576,8 +584,11 @@ export class SSEStreamSubscription implements StreamSubscription {
576584
return;
577585
}
578586

579-
armStall(); // any chunk (including server keepalives) resets the silence timer
587+
armStall(); // Each decoded record resets the silence timer.
580588
this.authRefreshed = false;
589+
// Headers alone do not establish stream recovery.
590+
this.retryCount = 0;
591+
this.stallCount = 0;
581592
controller.enqueue(value);
582593
}
583594
} catch (error) {
@@ -644,8 +655,14 @@ export class SSEStreamSubscription implements StreamSubscription {
644655
return;
645656
}
646657

647-
if (this.retryCount >= this.maxRetries) {
648-
const finalError = error || new Error("Max retries reached");
658+
const stallsExhausted = this.stallCount > this.maxStallRetries;
659+
if (this.retryCount >= this.maxRetries || stallsExhausted) {
660+
// Internal timeouts are failures, not caller cancellation.
661+
const finalError = stallsExhausted
662+
? new Error("Stream stalled: no records received")
663+
: error?.name === "AbortError"
664+
? new Error("Stream connection retries exhausted")
665+
: error || new Error("Max retries reached");
649666
controller.error(finalError);
650667
this.options.onError?.(finalError);
651668
return;

0 commit comments

Comments
 (0)