From 34d46732976a248ff65a6a4f92af86a680fb8830 Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Wed, 23 Sep 2026 19:23:31 +0000 Subject: [PATCH 1/4] [core] Add failing test: live read retries a hole below the known tail Co-Authored-By: Claude Co-Authored-By: Pranay Prakash <1797812+pranaygp@users.noreply.github.com> --- .../src/reconnecting-framed-stream.test.ts | 93 ++++++++++++++++++- 1 file changed, 92 insertions(+), 1 deletion(-) diff --git a/packages/core/src/reconnecting-framed-stream.test.ts b/packages/core/src/reconnecting-framed-stream.test.ts index 56e3b1ac64..f4cf8b7ace 100644 --- a/packages/core/src/reconnecting-framed-stream.test.ts +++ b/packages/core/src/reconnecting-framed-stream.test.ts @@ -834,7 +834,8 @@ describe('createReconnectingFramedStream', () => { { 0: () => scriptedStream([{ kind: 'close' }]), }, - async () => ({ tailIndex: 4, done: false }) + // Nothing written yet: an idle stream, not a hole below the tail. + async () => ({ tailIndex: -1, done: false }) ); setWorld(world); @@ -874,4 +875,94 @@ describe('createReconnectingFramedStream', () => { // reconnect throws before opening another stream. expect(calls).toBe(FRAMED_STREAM_MAX_TOTAL_RECONNECTS + 1); }); + describe('a hole below the known tail (#4306)', () => { + // The stream's metadata reports chunks through index 5, but chunk 2 is + // missing server-side. The live read delivers 0..1, and every session + // opened at index 2 ends (after the server's max duration in production, + // ~2 min each) without delivering a frame. Reconnecting from the same + // index can never make progress, so the reader must fail fast with an + // error naming the missing index instead of burning the whole + // consecutive-reconnect budget (~50 sessions, well over an hour). + const ends = [ + ['a clean EOF', { kind: 'close' }], + ['an errored body', { kind: 'error', err: new Error('max duration') }], + ] as const; + + for (const [label, end] of ends) { + it(`fails fast when every session at the hole ends with ${label}`, async () => { + const { world, calls } = makeWorldWithScriptedStreams( + { + 0: () => + scriptedStream([ + { kind: 'value', value: payloadFrame(0) }, + { kind: 'value', value: payloadFrame(1) }, + end, + ]), + 2: () => scriptedStream([end]), + }, + async () => ({ tailIndex: 5, done: false }) + ); + setWorld(world); + + const reader = createReconnectingFramedStream( + RUN_ID, + 's', + 0 + ).getReader(); + expect((await reader.read()).value).toEqual(payloadFrame(0)); + expect((await reader.read()).value).toEqual(payloadFrame(1)); + const error = await reader.read().catch((cause: unknown) => cause); + + expect(StreamError.is(error)).toBe(true); + expect(error).toHaveProperty( + 'message', + expect.stringMatching(/index 2.*tail index 5/) + ); + const atHole = calls.filter((i) => i === 2).length; + expect(atHole).toBeGreaterThan(0); + expect(atHole).toBeLessThan(FRAMED_STREAM_MAX_RECONNECTS); + }); + } + + it('keeps waiting when the session is idle at the tail', async () => { + // Position 2 with tail 1: the reader is caught up and waiting for new + // chunks, which is not a hole. The next session delivers. + let infoCalls = 0; + const { world, calls } = makeWorldWithScriptedStreams( + { + 0: () => + scriptedStream([ + { kind: 'value', value: payloadFrame(0) }, + { kind: 'value', value: payloadFrame(1) }, + { kind: 'close' }, + ]), + 2: () => + calls.filter((i) => i === 2).length < 5 + ? scriptedStream([{ kind: 'close' }]) + : scriptedStream([ + { kind: 'value', value: payloadFrame(2) }, + { kind: 'close' }, + ]), + }, + async () => { + infoCalls++; + return calls.filter((i) => i === 2).length < 5 + ? { tailIndex: 1, done: false } + : { tailIndex: 2, done: true }; + } + ); + setWorld(world); + + const chunks = await readAll( + createReconnectingFramedStream(RUN_ID, 's', 0) + ); + expect(chunks).toEqual([ + payloadFrame(0), + payloadFrame(1), + payloadFrame(2), + ]); + expect(calls.filter((i) => i === 2)).toHaveLength(5); + expect(infoCalls).toBeGreaterThan(0); + }); + }); }); From e3278e0b5019fa19489a6fd6e671f133211ce7fe Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Wed, 23 Sep 2026 19:29:01 +0000 Subject: [PATCH 2/4] [core] Fail a live stream read that stalls at a hole below the known tail Co-Authored-By: Claude Co-Authored-By: Pranay Prakash <1797812+pranaygp@users.noreply.github.com> --- .changeset/stream-read-hole-fail-fast.md | 5 ++ .../docs/v5/configuration/runtime-tuning.mdx | 5 ++ packages/core/src/serialization.ts | 80 +++++++++++++++++-- 3 files changed, 82 insertions(+), 8 deletions(-) create mode 100644 .changeset/stream-read-hole-fail-fast.md diff --git a/.changeset/stream-read-hole-fail-fast.md b/.changeset/stream-read-hole-fail-fast.md new file mode 100644 index 0000000000..de04edbe0f --- /dev/null +++ b/.changeset/stream-read-hole-fail-fast.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Fail a live stream read with a `StreamError` naming the index when consecutive reads stall at a chunk the stream reports but never serves, instead of reconnecting to it until the reconnect budget runs out diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 57e329b655..0cde022731 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -357,6 +357,11 @@ These variables are primarily for tests, debugging, or unusual deployments. - Default: `1000` - Total reconnect cap per stream session. +### `WORKFLOW_FRAMED_STREAM_MAX_STALLS_BELOW_TAIL` + +- Default: `3` +- Consecutive stream-reader sessions that may end without delivering a chunk while the stream reports a chunk at the resume index. Past this, the read fails with a `StreamError` naming the unreadable index, so the consumer can resume after it with `startIndex` instead of waiting indefinitely. + ### `WORKFLOW_WAIT_CONTINUATION_MAX_DELAY_SECONDS` - Default: `82800` (23 hours) diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index e2ebd02e0a..715441d9f1 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -6,7 +6,7 @@ import { WorkflowRuntimeError, } from '@workflow/errors'; import { once } from '@workflow/utils'; -import type { StreamWriteSession } from '@workflow/world'; +import type { StreamInfoResponse, StreamWriteSession } from '@workflow/world'; import { envNumber } from '@workflow/world/env-config'; import { parse, stringify, unflatten } from 'devalue'; import { monotonicFactory } from 'ulid'; @@ -888,6 +888,25 @@ const getFramedStreamMaxTotalReconnects = (): number => { integer: true, min: 1 } ); +/** + * Consecutive live-read sessions that may end without delivering a frame while + * the stream's metadata reports a chunk at the resume index, before the read + * fails. Such a session is not waiting for new data: the chunk is known to + * exist but was not served, so the index is a hole (the chunk is missing from + * storage) and reconnecting from it can never make progress. Each stalled + * session typically lasts the server's maximum response duration, so the + * general reconnect budget alone would stall a reader for over an hour. + */ +export const FRAMED_STREAM_MAX_STALLS_BELOW_TAIL = 3; + +/** Effective stall cap. Override: `WORKFLOW_FRAMED_STREAM_MAX_STALLS_BELOW_TAIL`. */ +const getFramedStreamMaxStallsBelowTail = (): number => + envNumber( + 'WORKFLOW_FRAMED_STREAM_MAX_STALLS_BELOW_TAIL', + FRAMED_STREAM_MAX_STALLS_BELOW_TAIL, + { integer: true, min: 1 } + ); + /** * Wraps the length-prefix-framed byte stream from `world.streams.get` with * transparent auto-reconnect. @@ -911,6 +930,12 @@ const getFramedStreamMaxTotalReconnects = (): number => * trusted (legacy behavior) rather than failing a read that may well be * complete. * + * A session that ends without delivering a frame while the metadata reports + * a chunk at the resume index is a stall at a hole (a chunk missing from + * storage). After `FRAMED_STREAM_MAX_STALLS_BELOW_TAIL` consecutive stalls the + * read errors with a `StreamError` naming the missing index, so the consumer + * can resume past it with `startIndex` instead of waiting indefinitely. + * * Negative `startIndex` values (last-N semantics) skip the reconnect * machinery because we cannot compute an absolute resume position without * a tail-index lookup; the returned stream behaves as a single-shot read. @@ -927,6 +952,7 @@ export function createReconnectingFramedStream( let consumedFrames = 0; let reconnectCount = 0; let totalReconnectCount = 0; + let stallsBelowTail = 0; let reader: ReadableStreamDefaultReader | undefined; let canceled = false; let cancelReason: unknown; @@ -985,13 +1011,49 @@ export function createReconnectingFramedStream( * into an error (or a reconnect loop) on a transient metadata blip would * be worse than the legacy behavior this check guards against. */ - async function isVerifiedComplete(): Promise { + async function getStreamInfo(): Promise { try { const world = await getWorldLazy(); - const info = await world.streams.getInfo(runId, name); - return info.done && currentStartIndex + consumedFrames > info.tailIndex; + return await world.streams.getInfo(runId, name); } catch { - return true; + return undefined; + } + } + + function isVerifiedComplete(info: StreamInfoResponse | undefined): boolean { + if (!info) return true; + return info.done && currentStartIndex + consumedFrames > info.tailIndex; + } + + /** + * Called when a session ends. Throws once consecutive sessions have ended + * without a frame while the metadata reports a chunk at the resume index. + * Metadata is only fetched for zero-frame sessions (unless the caller + * already has it), so sessions that deliver data pay nothing extra. + */ + async function assertNotStalledAtHole( + info?: StreamInfoResponse + ): Promise { + if (consumedFrames > 0) { + stallsBelowTail = 0; + return; + } + const resolved = info ?? (await getStreamInfo()); + // Unknown metadata neither confirms nor rules out a hole. + if (!resolved) return; + if (resolved.tailIndex < currentStartIndex) { + stallsBelowTail = 0; + return; + } + stallsBelowTail++; + if (stallsBelowTail >= getFramedStreamMaxStallsBelowTail()) { + throw new StreamError( + `Stream "${name}" of run "${runId}" has no readable chunk at index ` + + `${currentStartIndex}, below its tail index ${resolved.tailIndex}: ` + + `${stallsBelowTail} consecutive reads from that index delivered ` + + `nothing. The chunk is missing from storage; read from startIndex ` + + `${currentStartIndex + 1} or later to continue past it.` + ); } } @@ -1079,6 +1141,8 @@ export function createReconnectingFramedStream( return; } try { + await assertNotStalledAtHole(); + if (canceled) return; if (!(await reconnect())) return; } catch (reconnectErr) { controller.error(reconnectErr); @@ -1097,11 +1161,11 @@ export function createReconnectingFramedStream( // errored body, but on some paths it reaches the client as a clean // EOF), and a completed stream can still be cut mid-body; both // would otherwise be silently read as a shorter, complete stream. - const verifiedComplete = - !reconnectSupported || (await isVerifiedComplete()); + const info = reconnectSupported ? await getStreamInfo() : undefined; if (canceled) return; - if (!verifiedComplete) { + if (reconnectSupported && !isVerifiedComplete(info)) { try { + await assertNotStalledAtHole(info); if (!(await reconnect())) return; } catch (reconnectErr) { controller.error(reconnectErr); From a86b8229c51aa6a7ec90c40b4be22ba1b80908d3 Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Fri, 6 Nov 2026 01:06:22 +0000 Subject: [PATCH 3/4] [core] Don't count a session that received partial-frame bytes as a hole stall A session at the resume index that is cut partway through a large chunk received bytes but completed no frame, so it was counted as a stall below the tail. Three such cuts in a row failed a healthy read with "the chunk is missing from storage". Bytes arriving show the server is serving the chunk, so treat them as progress. Also cover the consecutive-stall reset. Co-Authored-By: Claude Co-Authored-By: Peter Wielander <29887157+VaguelySerious@users.noreply.github.com> --- .../docs/v5/configuration/runtime-tuning.mdx | 2 +- .../src/reconnecting-framed-stream.test.ts | 64 +++++++++++++++++++ packages/core/src/serialization.ts | 15 +++-- 3 files changed, 74 insertions(+), 7 deletions(-) diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index fcc0d1942a..f0f3c4d509 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -424,7 +424,7 @@ These variables are primarily for tests, debugging, or unusual deployments. ### `WORKFLOW_FRAMED_STREAM_MAX_STALLS_BELOW_TAIL` - Default: `3` -- Consecutive stream-reader sessions that may end without delivering a chunk while the stream reports a chunk at the resume index. Past this, the read fails with a `StreamError` naming the unreadable index, so the consumer can resume after it with `startIndex` instead of waiting indefinitely. +- Consecutive stream-reader sessions that may end without receiving any data while the stream reports a chunk at the resume index. Past this, the read fails with a `StreamError` naming the unreadable index, so the consumer can resume after it with `startIndex` instead of waiting indefinitely. ### `WORKFLOW_WAIT_CONTINUATION_MAX_DELAY_SECONDS` diff --git a/packages/core/src/reconnecting-framed-stream.test.ts b/packages/core/src/reconnecting-framed-stream.test.ts index f4cf8b7ace..2c1388ecdb 100644 --- a/packages/core/src/reconnecting-framed-stream.test.ts +++ b/packages/core/src/reconnecting-framed-stream.test.ts @@ -924,6 +924,70 @@ describe('createReconnectingFramedStream', () => { }); } + it('does not treat a chunk cut mid-frame as a hole', async () => { + // Chunk 2 is large and every early session is cut partway through it. + // The server is serving the chunk, so this is slow progress, not a hole. + const big = encodeFrame(new Uint8Array(64).fill(2)); + let atTwo = 0; + const { world } = makeWorldWithScriptedStreams( + { + 0: () => + scriptedStream([ + { kind: 'value', value: payloadFrame(0) }, + { kind: 'value', value: payloadFrame(1) }, + { kind: 'error', err: new Error('cut') }, + ]), + 2: () => + ++atTwo <= 5 + ? scriptedStream([ + { kind: 'value', value: big.slice(0, 20) }, + { kind: 'error', err: new Error('cut') }, + ]) + : scriptedStream([{ kind: 'value', value: big }]), + }, + async () => ({ tailIndex: 2, done: true }) + ); + setWorld(world); + + const chunks = await readAll( + createReconnectingFramedStream(RUN_ID, 's', 0) + ); + expect(chunks).toEqual([payloadFrame(0), payloadFrame(1), big]); + expect(atTwo).toBe(6); + }); + + it('counts only consecutive stalls, resetting when a session delivers', async () => { + // Two stalls at index 2, then chunk 2 arrives; two more stalls at + // index 3, then chunk 3 arrives. Four stalls in total, but never three + // in a row, so the read completes. + const seen: Record = {}; + const stallTwiceThen = (idx: number) => () => { + seen[idx] = (seen[idx] ?? 0) + 1; + return seen[idx] <= 2 + ? scriptedStream([{ kind: 'close' }]) + : scriptedStream([{ kind: 'value', value: payloadFrame(idx) }]); + }; + const { world, calls } = makeWorldWithScriptedStreams( + { + 0: () => + scriptedStream([ + { kind: 'value', value: payloadFrame(0) }, + { kind: 'value', value: payloadFrame(1) }, + ]), + 2: stallTwiceThen(2), + 3: stallTwiceThen(3), + }, + async () => ({ tailIndex: 3, done: true }) + ); + setWorld(world); + + const chunks = await readAll( + createReconnectingFramedStream(RUN_ID, 's', 0) + ); + expect(chunks).toEqual([0, 1, 2, 3].map(payloadFrame)); + expect(calls).toEqual([0, 2, 2, 2, 3, 3, 3]); + }); + it('keeps waiting when the session is idle at the tail', async () => { // Position 2 with tail 1: the reader is caught up and waiting for new // chunks, which is not a hole. The next session delivers. diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index 67d4338c4a..d3624aacb2 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -943,7 +943,7 @@ const getFramedStreamMaxTotalReconnects = (): number => ); /** - * Consecutive live-read sessions that may end without delivering a frame while + * Consecutive live-read sessions that may end without receiving any bytes while * the stream's metadata reports a chunk at the resume index, before the read * fails. Such a session is not waiting for new data: the chunk is known to * exist but was not served, so the index is a hole (the chunk is missing from @@ -984,7 +984,7 @@ const getFramedStreamMaxStallsBelowTail = (): number => * trusted (legacy behavior) rather than failing a read that may well be * complete. * - * A session that ends without delivering a frame while the metadata reports + * A session that ends without receiving any bytes while the metadata reports * a chunk at the resume index is a stall at a hole (a chunk missing from * storage). After `FRAMED_STREAM_MAX_STALLS_BELOW_TAIL` consecutive stalls the * read errors with a `StreamError` naming the missing index, so the consumer @@ -1081,14 +1081,17 @@ export function createReconnectingFramedStream( /** * Called when a session ends. Throws once consecutive sessions have ended - * without a frame while the metadata reports a chunk at the resume index. - * Metadata is only fetched for zero-frame sessions (unless the caller - * already has it), so sessions that deliver data pay nothing extra. + * without receiving any bytes while the metadata reports a chunk at the + * resume index. Metadata is only fetched for sessions that received nothing + * (unless the caller already has it), so sessions that deliver data pay + * nothing extra. */ async function assertNotStalledAtHole( info?: StreamInfoResponse ): Promise { - if (consumedFrames > 0) { + // Partial-frame bytes count as progress: the server is serving the chunk + // at the resume index (a large one cut mid-body), so it is not a hole. + if (consumedFrames > 0 || buffer.length > 0) { stallsBelowTail = 0; return; } From 2d04a32afb45d94ce1178da9d5870b20da3161a2 Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Fri, 13 Nov 2026 19:25:12 +0000 Subject: [PATCH 4/4] [core] Count only long zero-byte sessions as hole stalls; cover adjacent holes A zero-byte session below the tail now counts toward the hole stall cap only if it stayed open for at least 10 seconds. A session at a hole is held open until the server's max response duration (about two minutes), so a session that ends sooner is a brief blip: it is retried against the general reconnect budget as before, instead of three of them failing the read within a second. Drop the WORKFLOW_FRAMED_STREAM_MAX_STALLS_BELOW_TAIL override and its docs entry; the stall cap is a constant. Add a regression test for two adjacent holes: resuming at index + 1 lands on the second hole, fails the same bounded way naming that index, and resuming past it reads the rest of the stream. Co-Authored-By: Claude Co-Authored-By: Peter Wielander <29887157+VaguelySerious@users.noreply.github.com> --- .../docs/v5/configuration/runtime-tuning.mdx | 5 - .../src/reconnecting-framed-stream.test.ts | 199 +++++++++++++++--- packages/core/src/serialization.ts | 63 +++--- 3 files changed, 209 insertions(+), 58 deletions(-) diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index cec3f065f6..64b790c031 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -348,11 +348,6 @@ These variables are primarily for tests, debugging, or unusual deployments. - Default: `1000` - Total reconnect cap per stream session. -### `WORKFLOW_FRAMED_STREAM_MAX_STALLS_BELOW_TAIL` - -- Default: `3` -- Consecutive stream-reader sessions that may end without receiving any data while the stream reports a chunk at the resume index. Past this, the read fails with a `StreamError` naming the unreadable index, so the consumer can resume after it with `startIndex` instead of waiting indefinitely. - ### `WORKFLOW_WAIT_CONTINUATION_MAX_DELAY_SECONDS` - Default: `82800` (23 hours) diff --git a/packages/core/src/reconnecting-framed-stream.test.ts b/packages/core/src/reconnecting-framed-stream.test.ts index 2c1388ecdb..f819725a93 100644 --- a/packages/core/src/reconnecting-framed-stream.test.ts +++ b/packages/core/src/reconnecting-framed-stream.test.ts @@ -1,6 +1,6 @@ import { StreamError, StreamExpiredError } from '@workflow/errors'; import { SPEC_VERSION_CURRENT, type World } from '@workflow/world'; -import { afterEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; vi.mock('./version.js', () => ({ version: '0.0.0-test' })); @@ -8,7 +8,9 @@ import { setWorld } from './runtime/world.js'; import { createReconnectingFramedStream, FRAMED_STREAM_MAX_RECONNECTS, + FRAMED_STREAM_MAX_STALLS_BELOW_TAIL, FRAMED_STREAM_MAX_TOTAL_RECONNECTS, + FRAMED_STREAM_MIN_STALL_SESSION_MS, } from './serialization.js'; const FRAME_HEADER_SIZE = 4; @@ -29,32 +31,44 @@ function payloadFrame(n: number): Uint8Array { * Build a stream from a scripted pull sequence. Each entry either * enqueues a value or errors — this keeps the stream from transitioning * to the errored state before earlier values are actually read (which - * `start()`-time `controller.error` does immediately). + * `start()`-time `controller.error` does immediately). An `elapse` entry + * advances the (fake) system clock before the next entry, modelling a + * session the server holds open. */ function scriptedStream( steps: Array< | { kind: 'value'; value: Uint8Array } | { kind: 'error'; err: unknown } | { kind: 'close' } + | { kind: 'elapse'; ms: number } >, onCancel?: (reason?: unknown) => void ): ReadableStream { let i = 0; - return new ReadableStream({ - pull(controller) { - const step = steps[i++]; - if (!step) { - controller.close(); - return; - } - if (step.kind === 'value') controller.enqueue(step.value); - else if (step.kind === 'error') controller.error(step.err); - else controller.close(); - }, - cancel(reason) { - onCancel?.(reason); + return new ReadableStream( + { + pull(controller) { + let step = steps[i++]; + while (step?.kind === 'elapse') { + vi.setSystemTime(Date.now() + step.ms); + step = steps[i++]; + } + if (!step) { + controller.close(); + return; + } + if (step.kind === 'value') controller.enqueue(step.value); + else if (step.kind === 'error') controller.error(step.err); + else controller.close(); + }, + cancel(reason) { + onCancel?.(reason); + }, }, - }); + // Pull only when the reader reads, so an `elapse` entry runs during the + // session rather than while `streams.get` is still opening it. + { highWaterMark: 0 } + ); } async function readAll( @@ -70,6 +84,26 @@ async function readAll( return chunks; } +/** + * Reads a stream to its end, pushing each chunk into `out` as it arrives, and + * returns the error the stream failed with, if any. + */ +async function readInto( + stream: ReadableStream, + out: Uint8Array[] +): Promise { + const reader = stream.getReader(); + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) return undefined; + out.push(value); + } + } catch (error) { + return error; + } +} + /** * Builds a mock world whose `streams.get` returns a prepared sequence per * `startIndex`. Each call records the requested startIndex so assertions can @@ -878,11 +912,26 @@ describe('createReconnectingFramedStream', () => { describe('a hole below the known tail (#4306)', () => { // The stream's metadata reports chunks through index 5, but chunk 2 is // missing server-side. The live read delivers 0..1, and every session - // opened at index 2 ends (after the server's max duration in production, - // ~2 min each) without delivering a frame. Reconnecting from the same - // index can never make progress, so the reader must fail fast with an - // error naming the missing index instead of burning the whole + // opened at index 2 stays open until the server's max duration (~2 min in + // production) and ends without delivering a frame. Reconnecting from the + // same index can never make progress, so the reader must fail fast with + // an error naming the missing index instead of burning the whole // consecutive-reconnect budget (~50 sessions, well over an hour). + beforeEach(() => { + vi.useFakeTimers({ toFake: ['Date'] }); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + /** A session the server holds open for its max duration. */ + const held = { kind: 'elapse', ms: 120_000 } as const; + /** A session that ends right away, as during a brief server blip. */ + const quick = { + kind: 'elapse', + ms: FRAMED_STREAM_MIN_STALL_SESSION_MS - 1, + } as const; + const ends = [ ['a clean EOF', { kind: 'close' }], ['an errored body', { kind: 'error', err: new Error('max duration') }], @@ -896,9 +945,10 @@ describe('createReconnectingFramedStream', () => { scriptedStream([ { kind: 'value', value: payloadFrame(0) }, { kind: 'value', value: payloadFrame(1) }, + held, end, ]), - 2: () => scriptedStream([end]), + 2: () => scriptedStream([held, end]), }, async () => ({ tailIndex: 5, done: false }) ); @@ -916,14 +966,108 @@ describe('createReconnectingFramedStream', () => { expect(StreamError.is(error)).toBe(true); expect(error).toHaveProperty( 'message', - expect.stringMatching(/index 2.*tail index 5/) + expect.stringMatching(/index 2.*tail index 5.*startIndex 3/) + ); + // The session that delivered 0..1 then idled is not a stall; the + // sessions opened at index 2 are. + expect(calls.filter((i) => i === 2)).toHaveLength( + FRAMED_STREAM_MAX_STALLS_BELOW_TAIL ); - const atHole = calls.filter((i) => i === 2).length; - expect(atHole).toBeGreaterThan(0); - expect(atHole).toBeLessThan(FRAMED_STREAM_MAX_RECONNECTS); }); } + it('fails fast again at each of two adjacent holes, so a consumer can step past both', async () => { + // Chunks 2 and 3 are both missing (production has shown adjacent + // missing indices). Resuming at index + 1 lands on the second hole, + // which must fail the same bounded way, naming its own index, rather + // than wait out the general reconnect budget. Resuming past it reads + // the rest of the stream. + const { world, calls } = makeWorldWithScriptedStreams( + { + 0: () => + scriptedStream([ + { kind: 'value', value: payloadFrame(0) }, + { kind: 'value', value: payloadFrame(1) }, + held, + { kind: 'error', err: new Error('max duration') }, + ]), + 2: () => scriptedStream([held, { kind: 'close' }]), + 3: () => scriptedStream([held, { kind: 'close' }]), + 4: () => + scriptedStream([ + { kind: 'value', value: payloadFrame(4) }, + { kind: 'value', value: payloadFrame(5) }, + { kind: 'close' }, + ]), + }, + async () => ({ tailIndex: 5, done: true }) + ); + setWorld(world); + + // Consumer loop: on a hole error, resume after the index it names. + const resumeAfterHole = /read from startIndex (\d+)/; + const delivered: Uint8Array[] = []; + const holes: number[] = []; + let startIndex: number | undefined = 0; + while (startIndex !== undefined) { + const error = await readInto( + createReconnectingFramedStream(RUN_ID, 's', startIndex), + delivered + ); + if (error === undefined) break; + expect(StreamError.is(error)).toBe(true); + const next = resumeAfterHole.exec((error as Error).message); + expect(next).not.toBeNull(); + startIndex = Number(next?.[1]); + holes.push(startIndex - 1); + } + + expect(holes).toEqual([2, 3]); + expect(delivered).toEqual([0, 1, 4, 5].map(payloadFrame)); + expect(calls.filter((i) => i === 2)).toHaveLength( + FRAMED_STREAM_MAX_STALLS_BELOW_TAIL + ); + expect(calls.filter((i) => i === 3)).toHaveLength( + FRAMED_STREAM_MAX_STALLS_BELOW_TAIL + ); + }); + + it('does not count sessions that end quickly as stalls', async () => { + // A brief server blip below the tail: several sessions at index 2 end + // with no data almost immediately, then the chunk is served. A hole + // session stays open for the server's max duration, so these are + // retried against the general reconnect budget instead. + const blips = FRAMED_STREAM_MAX_STALLS_BELOW_TAIL * 3; + let atTwo = 0; + const { world } = makeWorldWithScriptedStreams( + { + 0: () => + scriptedStream([ + { kind: 'value', value: payloadFrame(0) }, + { kind: 'value', value: payloadFrame(1) }, + { kind: 'error', err: new Error('blip') }, + ]), + 2: () => + ++atTwo <= blips + ? scriptedStream([ + quick, + atTwo % 2 + ? { kind: 'close' } + : { kind: 'error', err: new Error('blip') }, + ]) + : scriptedStream([{ kind: 'value', value: payloadFrame(2) }]), + }, + async () => ({ tailIndex: 2, done: true }) + ); + setWorld(world); + + const chunks = await readAll( + createReconnectingFramedStream(RUN_ID, 's', 0) + ); + expect(chunks).toEqual([0, 1, 2].map(payloadFrame)); + expect(atTwo).toBe(blips + 1); + }); + it('does not treat a chunk cut mid-frame as a hole', async () => { // Chunk 2 is large and every early session is cut partway through it. // The server is serving the chunk, so this is slow progress, not a hole. @@ -941,6 +1085,7 @@ describe('createReconnectingFramedStream', () => { ++atTwo <= 5 ? scriptedStream([ { kind: 'value', value: big.slice(0, 20) }, + held, { kind: 'error', err: new Error('cut') }, ]) : scriptedStream([{ kind: 'value', value: big }]), @@ -964,7 +1109,7 @@ describe('createReconnectingFramedStream', () => { const stallTwiceThen = (idx: number) => () => { seen[idx] = (seen[idx] ?? 0) + 1; return seen[idx] <= 2 - ? scriptedStream([{ kind: 'close' }]) + ? scriptedStream([held, { kind: 'close' }]) : scriptedStream([{ kind: 'value', value: payloadFrame(idx) }]); }; const { world, calls } = makeWorldWithScriptedStreams( @@ -1002,7 +1147,7 @@ describe('createReconnectingFramedStream', () => { ]), 2: () => calls.filter((i) => i === 2).length < 5 - ? scriptedStream([{ kind: 'close' }]) + ? scriptedStream([held, { kind: 'close' }]) : scriptedStream([ { kind: 'value', value: payloadFrame(2) }, { kind: 'close' }, diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index 29fcfb62c4..f588ec1935 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -966,13 +966,14 @@ const getFramedStreamMaxTotalReconnects = (): number => */ export const FRAMED_STREAM_MAX_STALLS_BELOW_TAIL = 3; -/** Effective stall cap. Override: `WORKFLOW_FRAMED_STREAM_MAX_STALLS_BELOW_TAIL`. */ -const getFramedStreamMaxStallsBelowTail = (): number => - envNumber( - 'WORKFLOW_FRAMED_STREAM_MAX_STALLS_BELOW_TAIL', - FRAMED_STREAM_MAX_STALLS_BELOW_TAIL, - { integer: true, min: 1 } - ); +/** + * Shortest zero-byte session that counts as a stall below the tail. A session + * at a hole stays open until the server's maximum response duration (about two + * minutes on world-vercel). One that ends sooner is a transport blip: it is + * retried against the general reconnect budget as before, and neither counts + * toward nor resets {@link FRAMED_STREAM_MAX_STALLS_BELOW_TAIL}. + */ +export const FRAMED_STREAM_MIN_STALL_SESSION_MS = 10_000; /** * Wraps the length-prefix-framed byte stream from `world.streams.get` with @@ -997,11 +998,13 @@ const getFramedStreamMaxStallsBelowTail = (): number => * trusted (legacy behavior) rather than failing a read that may well be * complete. * - * A session that ends without receiving any bytes while the metadata reports - * a chunk at the resume index is a stall at a hole (a chunk missing from - * storage). After `FRAMED_STREAM_MAX_STALLS_BELOW_TAIL` consecutive stalls the - * read errors with a `StreamError` naming the missing index, so the consumer - * can resume past it with `startIndex` instead of waiting indefinitely. + * A session that stays open for at least `FRAMED_STREAM_MIN_STALL_SESSION_MS` + * and ends without receiving any bytes while the metadata reports a chunk at + * the resume index is a stall at a hole (a chunk missing from storage). After + * `FRAMED_STREAM_MAX_STALLS_BELOW_TAIL` consecutive stalls the read errors + * with a `StreamError` naming the missing index, so the consumer can resume + * past it with `startIndex` instead of waiting indefinitely. The reader never + * skips the hole itself: silently dropping a chunk can corrupt consumer state. * * Negative `startIndex` values (last-N semantics) skip the reconnect * machinery because we cannot compute an absolute resume position without @@ -1020,6 +1023,7 @@ export function createReconnectingFramedStream( let reconnectCount = 0; let totalReconnectCount = 0; let stallsBelowTail = 0; + let sessionStartedAt = 0; let reader: ReadableStreamDefaultReader | undefined; let canceled = false; let cancelReason: unknown; @@ -1066,18 +1070,12 @@ export function createReconnectingFramedStream( return false; } if (connectMs === undefined) connectMs = Date.now() - connectStart; + sessionStartedAt = Date.now(); reader = stream.getReader(); return true; } - /** - * Whether an upstream EOF represents genuine completion: the stream's - * authoritative metadata says it is done AND the frames delivered so far - * cover every chunk up to the tail (one outer frame == one server chunk). - * A metadata failure trusts the EOF, since turning a healthy completion - * into an error (or a reconnect loop) on a transient metadata blip would - * be worse than the legacy behavior this check guards against. - */ + /** The stream's metadata, or `undefined` if the lookup fails. */ async function getStreamInfo(): Promise { try { const world = await getWorldLazy(); @@ -1087,17 +1085,25 @@ export function createReconnectingFramedStream( } } + /** + * Whether an upstream EOF represents genuine completion: the stream's + * authoritative metadata says it is done AND the frames delivered so far + * cover every chunk up to the tail (one outer frame == one server chunk). + * A metadata failure trusts the EOF, since turning a healthy completion + * into an error (or a reconnect loop) on a transient metadata blip would + * be worse than the legacy behavior this check guards against. + */ function isVerifiedComplete(info: StreamInfoResponse | undefined): boolean { if (!info) return true; return info.done && currentStartIndex + consumedFrames > info.tailIndex; } /** - * Called when a session ends. Throws once consecutive sessions have ended - * without receiving any bytes while the metadata reports a chunk at the - * resume index. Metadata is only fetched for sessions that received nothing - * (unless the caller already has it), so sessions that deliver data pay - * nothing extra. + * Called when a session ends. Throws once consecutive sessions have stayed + * open without receiving any bytes while the metadata reports a chunk at + * the resume index. Metadata is only fetched for long sessions that + * received nothing (unless the caller already has it), so sessions that + * deliver data pay nothing extra. */ async function assertNotStalledAtHole( info?: StreamInfoResponse @@ -1108,6 +1114,11 @@ export function createReconnectingFramedStream( stallsBelowTail = 0; return; } + // A zero-byte session that ended quickly is a blip, not a session held + // open at a hole: leave it to the general reconnect budget. + if (Date.now() - sessionStartedAt < FRAMED_STREAM_MIN_STALL_SESSION_MS) { + return; + } const resolved = info ?? (await getStreamInfo()); // Unknown metadata neither confirms nor rules out a hole. if (!resolved) return; @@ -1116,7 +1127,7 @@ export function createReconnectingFramedStream( return; } stallsBelowTail++; - if (stallsBelowTail >= getFramedStreamMaxStallsBelowTail()) { + if (stallsBelowTail >= FRAMED_STREAM_MAX_STALLS_BELOW_TAIL) { throw new StreamError( `Stream "${name}" of run "${runId}" has no readable chunk at index ` + `${currentStartIndex}, below its tail index ${resolved.tailIndex}: ` +