From fd1c295570a593e369fd36a5143bb50662a9a218 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 22 Jun 2026 21:53:23 +0000 Subject: [PATCH 1/2] [core] Retry stream reopen against the reconnect budget (#2334) Signed-off-by: Peter Wielander --- .../src/reconnecting-framed-stream.test.ts | 47 +++++++++++++++++++ packages/core/src/serialization.ts | 44 ++++++++++++----- 2 files changed, 78 insertions(+), 13 deletions(-) diff --git a/packages/core/src/reconnecting-framed-stream.test.ts b/packages/core/src/reconnecting-framed-stream.test.ts index 50cbda8ecd..6f76819016 100644 --- a/packages/core/src/reconnecting-framed-stream.test.ts +++ b/packages/core/src/reconnecting-framed-stream.test.ts @@ -172,6 +172,53 @@ describe('createReconnectingFramedStream', () => { expect(calls).toEqual([0, 2]); }); + it('retries the reopen itself against the budget instead of failing fatally', async () => { + // After 2 frames the connection drops (read error → reconnect at index 2). + // The first *reopen* attempt also fails — the server is briefly + // unavailable during the reconnect window. That transient failure of the + // reopen is the exact blip this wrapper exists to survive, so it must be + // counted against the budget and retried, not treated as fatal. The + // second reopen succeeds and the stream completes. + const calls: number[] = []; + let reopenAttempts = 0; + const world = { + streams: { + get: vi.fn( + async (_runId: string, _name: string, startIndex?: number) => { + const idx = startIndex ?? 0; + calls.push(idx); + if (idx === 0) { + return scriptedStream([ + { kind: 'value', value: payloadFrame(1) }, + { kind: 'value', value: payloadFrame(2) }, + { kind: 'error', err: new Error('max-duration abort') }, + ]); + } + // Reopen at index 2: throw on the first attempt, succeed on the next. + reopenAttempts++; + if (reopenAttempts === 1) { + throw new Error('reopen failed: server briefly unavailable'); + } + return scriptedStream([ + { kind: 'value', value: payloadFrame(3) }, + { kind: 'close' }, + ]); + } + ), + }, + } as unknown as World; + setWorld(world); + + const stream = createReconnectingFramedStream(RUN_ID, 's', 0); + const chunks = await readAll(stream); + + // The failed reopen did not surface to the consumer; the stream recovered. + expect(chunks).toEqual([payloadFrame(1), payloadFrame(2), payloadFrame(3)]); + // index 0 once, then index 2 twice — the failed reopen and the retry both + // resume from the same position. + expect(calls).toEqual([0, 2, 2]); + }); + it('respects an initial non-zero startIndex on reconnect', async () => { const { world, calls } = makeWorldWithScriptedStreams({ 10: () => diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index 352354cafa..12b133bfdc 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -710,26 +710,44 @@ export function createReconnectingFramedStream( } async function reconnect(): Promise { - reconnectCount++; - totalReconnectCount++; - if (reconnectCount > FRAMED_STREAM_MAX_RECONNECTS) { - throw new Error( - `Stream "${name}" exceeded maximum reconnection attempts (${FRAMED_STREAM_MAX_RECONNECTS})` - ); - } - if (totalReconnectCount > FRAMED_STREAM_MAX_TOTAL_RECONNECTS) { - throw new Error( - `Stream "${name}" exceeded maximum total reconnection attempts (${FRAMED_STREAM_MAX_TOTAL_RECONNECTS})` - ); - } if (reader) { await reader.cancel().catch(() => {}); reader = undefined; } + // Advance the resume position past the frames already delivered, then + // drop any partial-frame bytes — the reopened connection re-sends from a + // frame boundary at the new index. currentStartIndex += consumedFrames; consumedFrames = 0; buffer = new Uint8Array(0); - await connect(); + + // Retry the reopen itself against the reconnect budget. A transient + // failure of connect() — the server briefly unavailable during the + // reconnect window — is the exact blip this wrapper exists to survive, so + // count it against the budget and try again rather than treating it as + // fatal. Only budget exhaustion (a server that stays down) terminates the + // stream. + for (;;) { + reconnectCount++; + totalReconnectCount++; + if (reconnectCount > FRAMED_STREAM_MAX_RECONNECTS) { + throw new Error( + `Stream "${name}" exceeded maximum reconnection attempts (${FRAMED_STREAM_MAX_RECONNECTS})` + ); + } + if (totalReconnectCount > FRAMED_STREAM_MAX_TOTAL_RECONNECTS) { + throw new Error( + `Stream "${name}" exceeded maximum total reconnection attempts (${FRAMED_STREAM_MAX_TOTAL_RECONNECTS})` + ); + } + try { + await connect(); + return; + } catch { + // Reopen failed transiently; loop to retry, counting against the + // budget so a server that never recovers still terminates the stream. + } + } } return new ReadableStream({ From fa454c6a720cd658a6d2aa89ef8ee7d24b9835ed Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Mon, 22 Jun 2026 21:59:44 +0000 Subject: [PATCH 2/2] Fix: New test uses `main`-branch APIs (`RUN_ID` arg and `world.streams.get`) that don't exist on the stable branch, causing a compile error and a broken test. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit fixes the issue reported at packages/core/src/reconnecting-framed-stream.test.ts:212 ## Bug The test `"retries the reopen itself against the budget instead of failing fatally"` (added at `packages/core/src/reconnecting-framed-stream.test.ts:175`) was backported from `main` without adapting it to the `stable` branch API. Two concrete mismatches: 1. **Undeclared identifier `RUN_ID`** — The test called: ```ts const stream = createReconnectingFramedStream(RUN_ID, 's', 0); ``` `RUN_ID` is never imported or declared anywhere in the file. This is a hard TypeScript compile error (`Cannot find name 'RUN_ID'`) and would be a runtime `ReferenceError`. On the stable branch the signature is `createReconnectingFramedStream(name: string, startIndex?: number)` — no `runId` parameter. 2. **Wrong world mock shape** — The test built a world mock exposing `world.streams.get(runId, name, startIndex)`, but the stable `connect()` calls `world.readFromStream(name, effectiveStartIndex)` (verified in `serialization.ts`). So `world.readFromStream` would be `undefined` and `connect()` would throw, never exercising the intended reopen-retry path. ## Fix Rewrote the test to match the stable API, matching the existing `makeWorldWithScriptedStreams` pattern used by every other test in the file: - Mock `readFromStream(_name, startIndex)` instead of `streams.get(...)`. - Call `createReconnectingFramedStream('s', 0)` (2-arg form). The test's intent is preserved: index 0 delivers two frames then errors; the first reopen at index 2 throws transiently; the second reopen succeeds. Assertions remain `chunks === [1, 2, 3]` and `calls === [0, 2, 2]`. This matches the stable `reconnect()` implementation, which retries `connect()` against the reconnect budget — so `readFromStream` is called with `2` twice (failed reopen + successful retry), yielding `[0, 2, 2]`. Co-authored-by: Vercel Co-authored-by: VaguelySerious --- .../src/reconnecting-framed-stream.test.ts | 46 +++++++++---------- 1 file changed, 21 insertions(+), 25 deletions(-) diff --git a/packages/core/src/reconnecting-framed-stream.test.ts b/packages/core/src/reconnecting-framed-stream.test.ts index 6f76819016..3f9c2385ba 100644 --- a/packages/core/src/reconnecting-framed-stream.test.ts +++ b/packages/core/src/reconnecting-framed-stream.test.ts @@ -182,34 +182,30 @@ describe('createReconnectingFramedStream', () => { const calls: number[] = []; let reopenAttempts = 0; const world = { - streams: { - get: vi.fn( - async (_runId: string, _name: string, startIndex?: number) => { - const idx = startIndex ?? 0; - calls.push(idx); - if (idx === 0) { - return scriptedStream([ - { kind: 'value', value: payloadFrame(1) }, - { kind: 'value', value: payloadFrame(2) }, - { kind: 'error', err: new Error('max-duration abort') }, - ]); - } - // Reopen at index 2: throw on the first attempt, succeed on the next. - reopenAttempts++; - if (reopenAttempts === 1) { - throw new Error('reopen failed: server briefly unavailable'); - } - return scriptedStream([ - { kind: 'value', value: payloadFrame(3) }, - { kind: 'close' }, - ]); - } - ), - }, + readFromStream: vi.fn(async (_name: string, startIndex?: number) => { + const idx = startIndex ?? 0; + calls.push(idx); + if (idx === 0) { + return scriptedStream([ + { kind: 'value', value: payloadFrame(1) }, + { kind: 'value', value: payloadFrame(2) }, + { kind: 'error', err: new Error('max-duration abort') }, + ]); + } + // Reopen at index 2: throw on the first attempt, succeed on the next. + reopenAttempts++; + if (reopenAttempts === 1) { + throw new Error('reopen failed: server briefly unavailable'); + } + return scriptedStream([ + { kind: 'value', value: payloadFrame(3) }, + { kind: 'close' }, + ]); + }), } as unknown as World; setWorld(world); - const stream = createReconnectingFramedStream(RUN_ID, 's', 0); + const stream = createReconnectingFramedStream('s', 0); const chunks = await readAll(stream); // The failed reopen did not surface to the consumer; the stream recovered.