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/packages/core/src/reconnecting-framed-stream.test.ts b/packages/core/src/reconnecting-framed-stream.test.ts index 56e3b1ac64..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 @@ -834,7 +868,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 +909,269 @@ 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 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') }], + ] 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) }, + held, + end, + ]), + 2: () => scriptedStream([held, 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.*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 + ); + }); + } + + 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. + 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) }, + held, + { 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([held, { 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. + 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([held, { 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); + }); + }); }); diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index 0b23a0ddd7..f588ec1935 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -7,7 +7,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'; @@ -955,6 +955,26 @@ const getFramedStreamMaxTotalReconnects = (): number => { integer: true, min: 1 } ); +/** + * 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 + * 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; + +/** + * 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 * transparent auto-reconnect. @@ -978,6 +998,14 @@ const getFramedStreamMaxTotalReconnects = (): number => * trusted (legacy behavior) rather than failing a read that may well be * complete. * + * 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 * a tail-index lookup; the returned stream behaves as a single-shot read. @@ -994,6 +1022,8 @@ export function createReconnectingFramedStream( let consumedFrames = 0; let reconnectCount = 0; let totalReconnectCount = 0; + let stallsBelowTail = 0; + let sessionStartedAt = 0; let reader: ReadableStreamDefaultReader | undefined; let canceled = false; let cancelReason: unknown; @@ -1040,10 +1070,21 @@ export function createReconnectingFramedStream( return false; } if (connectMs === undefined) connectMs = Date.now() - connectStart; + sessionStartedAt = Date.now(); reader = stream.getReader(); return true; } + /** The stream's metadata, or `undefined` if the lookup fails. */ + async function getStreamInfo(): Promise { + try { + const world = await getWorldLazy(); + return await world.streams.getInfo(runId, name); + } catch { + return undefined; + } + } + /** * Whether an upstream EOF represents genuine completion: the stream's * authoritative metadata says it is done AND the frames delivered so far @@ -1052,13 +1093,48 @@ 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 { - try { - const world = await getWorldLazy(); - const info = await world.streams.getInfo(runId, name); - return info.done && currentStartIndex + consumedFrames > info.tailIndex; - } catch { - return true; + 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 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 + ): Promise { + // 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; + } + // 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; + if (resolved.tailIndex < currentStartIndex) { + stallsBelowTail = 0; + return; + } + stallsBelowTail++; + 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}: ` + + `${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.` + ); } } @@ -1146,6 +1222,8 @@ export function createReconnectingFramedStream( return; } try { + await assertNotStalledAtHole(); + if (canceled) return; if (!(await reconnect())) return; } catch (reconnectErr) { controller.error(reconnectErr); @@ -1164,11 +1242,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);