[core] When live stream read waits forever at missing chunk, fail early instead

.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

packages/core/src/reconnecting-framed-stream.test.ts

@@ -1,14 +1,16 @@
 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' }));
 
 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<Uint8Array> {
   let i = 0;
-  return new ReadableStream<Uint8Array>({
-    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<Uint8Array>(
+    {
+      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<Uint8Array>,
+  out: Uint8Array[]
+): Promise<unknown> {
+  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<number, number> = {};
+      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);
+    });
+  });
 });