Serde and pagination: abortable reads, paginator close on malformed PageInfo, tristate null, SSE single report (#79)

docs/sdk-documentation/write-a-paging-strategy.md

@@ -95,7 +95,7 @@ for await (const page of paginator.pages()) { /* page by page */ }
 independent walks, not two views of one. That is also why `@dexpace/rx`'s `pageItems$`/`pages$` are
 cold and repeatable while its SSE observables are not.
 
-## The four rules
+## The five rules
 
 **1. Take everything you need from the response before your promise settles** (`PAGE-5`). The
 response you are handed is live and single-use; the engine may close it the moment `parse` resolves.
@@ -135,6 +135,20 @@ which is an erratum recorded in `docs/knowledge/notes/pagination.md` and `docs/w
 `maxPages` on `PaginatorInit` is the backstop, not the design. Loop detection is not the paginator's
 job.
 
+**5. Always return a well-formed `PageInfo`** (`PAGE-4`). `items` must be an array — an empty one is
+fine and is a perfectly valid non-terminal page — and `nextRequest === undefined` is the **single,
+exclusive** end-of-stream signal. A `PageInfo` that is itself `null` or `undefined`, or whose `items`
+is either, is a programmer error and the engine treats it as one: it closes the response and throws
+an assertion naming the invariant you broke. It does **not** end the walk quietly, because "the
+strategy forgot to `return`" and "the server ran out of pages" must not look the same from the
+outside. Use `pageInfo(items, next?)` and this cannot happen; the check exists because `parse`
+crosses a seam, where an `any`-typed decode or a trusted server field can produce a shape the types
+say is impossible.
+
+Terminating and failing are different acts. To *end* the walk, return `pageInfo(items)` with no next
+request. To *fail* it, throw — the engine closes the response and your error reaches the consumer
+unwrapped (`PAGE-13`, `PAGE-28`).
+
 `PaginationError` is reserved for engine misuse and precondition violations — not for "the server
 returned a page I did not understand", which is your `extract`'s error to raise.
 

docs/sdk-documentation/write-a-serde.md

@@ -61,6 +61,33 @@ import {
 
 const TEXT = new TextEncoder();
 
+/**
+ * Settle when `operation` settles, or as soon as `signal` aborts — whichever comes first, with the
+ * caller's own `reason` surfaced verbatim. `throwIfAborted()` alone cannot interrupt a `read()` or
+ * `write()` that never resolves, which is the case that leaves a caller's stream locked forever.
+ */
+const raceAbort = async <T>(
+  operation: Promise<T>,
+  signal: AbortSignal | undefined,
+): Promise<T> => {
+  if (signal === undefined) return operation;
+  signal.throwIfAborted();
+  let onAbort = (): void => undefined;
+  const aborted = new Promise<never>((_resolve, reject) => {
+    onAbort = (): void => {
+      reject(signal.reason as unknown);
+    };
+    signal.addEventListener('abort', onAbort, {once: true});
+  });
+  try {
+    // The loser of the race keeps `Promise.race`'s own handler, so a `read()` that rejects after
+    // the lock is released never becomes an unhandled rejection.
+    return await Promise.race([operation, aborted]);
+  } finally {
+    signal.removeEventListener('abort', onAbort);
+  }
+};
+
 export function csvSerde(): Serde {
   const encode = (value: unknown): string => {
     if (!Array.isArray(value)) {
@@ -106,7 +133,8 @@ export function csvSerde(): Serde {
         options?.signal?.throwIfAborted(); // before the lock: an aborted call leaves the sink free
         const writer = sink.getWriter(); // TypeError if contended — a programmer error, not re-typed
         try {
-          await writer.write(TEXT.encode(encode(value)));
+          // Raced, not just checked: a slow sink parks this write, and the abort must reach it.
+          await raceAbort(writer.write(TEXT.encode(encode(value))), options?.signal);
         } finally {
           writer.releaseLock(); // never close or abort: the caller owns the sink (SERDE-3)
         }
@@ -115,14 +143,15 @@ export function csvSerde(): Serde {
     deserializer: {
       deserialize: (data, target) => decode(new TextDecoder().decode(data), target),
       async deserializeFrom(source, target, options) {
-        options?.signal?.throwIfAborted(); // before the lock, and again after every read below
+        options?.signal?.throwIfAborted(); // before the lock: an aborted call never takes one
         const reader = source.getReader();
         const chunks: Uint8Array[] = [];
         try {
           for (;;) {
-            const {done, value} = await reader.read();
+            // Raced, not checked between chunks: a source that stalls mid-body parks the loop
+            // inside `read()`, where a between-chunks check never runs again.
+            const {done, value} = await raceAbort(reader.read(), options?.signal);
             if (done) break;
-            options?.signal?.throwIfAborted();
             chunks.push(value);
           }
         } finally {
@@ -141,7 +170,7 @@ export function csvSerde(): Serde {
 }
 ```
 
-Six rules, all visible above:
+Seven rules, all visible above:
 
 1. **Raise `SerializationError` / `DeserializationError`, never a raw error** — with one stated
    exception: `serializeInto`'s out-of-range or does-not-fit case is a plain `RangeError` with no
@@ -160,6 +189,17 @@ Six rules, all visible above:
    target, on **every** entry point (`SERDE-13`), never return a `null` that detonates at a later
    field access. The fallback label is the literal `'the target type'`; each codec repeats it,
    because `SEAM-1` leaves core with no exported constant to share.
+7. **An abort must race the pending operation, not sit between two of them** (`SERDE-3`). The seam
+   promises that "an aborted call never leaves the caller's source locked", and a
+   `throwIfAborted()` between chunks cannot keep it: a source that stalls mid-body parks the drain
+   inside `read()`, so the call never settles and the lock is never released. Race each pending
+   `read()`/`write()` against the signal, remove the listener in a `finally`, then release the lock
+   as usual. Releasing a reader with a read still outstanding is legal on every supported runtime
+   and does unlock the stream — the outstanding read rejects, differently per runtime
+   (`AbortError` on Bun 1.3.14, `TypeError: Invalid state: Releasing reader` on Node 20.3 and 26,
+   measured 2026-09-05), which is why the caller must see the signal's `reason` instead.
+   `@dexpace/codec-json` holds **one** listener for the whole drive rather than one per chunk; the
+   example above takes the simpler per-operation form.
 
 `mediaType` is the default `Content-Type` — `serdeBody(value, serde)` reads it, and a caller may
 override per body.

packages/codec-json/src/abort-race.ts

@@ -0,0 +1,84 @@
+// SPDX-License-Identifier: MIT
+// packages/codec-json/src/abort-race.ts
+
+/**
+ * One abort listener held for the length of a whole stream drive, plus the race that lets it settle
+ * an operation that is already *pending*.
+ *
+ * @internal
+ */
+export interface AbortRace {
+  /**
+   * Settle with `operation`, or reject with the signal's `reason` the moment it aborts — whichever
+   * happens first.
+   *
+   * Also rejects before `operation` is even consulted when the signal is already aborted, which is
+   * the between-chunks check the loop used to make for itself.
+   */
+  race<T>(operation: Promise<T>): Promise<T>;
+
+  /** Drop the abort listener. Call from the `finally` that releases the stream lock. */
+  release(): void;
+}
+
+/** The no-signal case: no listener to install, no race to run, no allocation per chunk. */
+const UNRACED: AbortRace = Object.freeze({
+  race: <T>(operation: Promise<T>): Promise<T> => operation,
+  release: (): void => undefined,
+});
+
+/**
+ * Bind `signal` to a single listener that can interrupt any number of pending operations
+ * (SERDE-3, audit #67 / #79).
+ *
+ * `throwIfAborted()` between chunks is not enough on its own: a `reader.read()` that never resolves
+ * is never raced against anything, so the drain parks inside it, the call never settles, and
+ * `source.locked` stays `true` for the rest of the process — the opposite of the seam's promise that
+ * "an aborted call never leaves the caller's source locked". Racing the pending operation is what
+ * makes that promise true rather than aspirational.
+ *
+ * The signal's `reason` is surfaced verbatim, never re-typed: a caller aborting with its own error
+ * gets that error back, and a bare `abort()` gets the platform's `AbortError` `DOMException`, which
+ * is exactly what `throwIfAborted()` would have thrown.
+ *
+ * One listener per call, not one per chunk — a 10 000-chunk body would otherwise register and remove
+ * 10 000 listeners on a signal the caller may hold for the life of a request.
+ *
+ * @param signal - the caller's signal, or `undefined` when the call took none.
+ * @returns a race bound to `signal`, whose `release()` removes the listener.
+ * @internal
+ */
+export function abortRace(signal: AbortSignal | undefined): AbortRace {
+  if (signal === undefined) return UNRACED;
+
+  let onAbort = (): void => undefined;
+  const aborted = new Promise<never>((_resolve, reject) => {
+    onAbort = (): void => {
+      // The seam documents that a caller sees its own abort `reason` verbatim, and a caller may
+      // abort with any value at all — `controller.abort('gone')` is legal. Re-typing it here would
+      // break that contract, and it is also exactly what `throwIfAborted()` throws.
+      // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors -- see above; re-enable if the seam ever narrows `reason` to an Error
+      reject(signal.reason as unknown);
+    };
+    signal.addEventListener('abort', onAbort, {once: true});
+  });
+  // An abort can land while nothing is racing this promise — between two reads, or after the last
+  // one and before `release()`. That rejection would be an unhandled one, which takes the process
+  // down under Node's default policy (`docs/knowledge/harvested/cancellation-and-timeouts.md:26`).
+  // A no-op handler marks it handled without stopping `Promise.race` below from seeing it.
+  void aborted.catch(() => undefined);
+
+  return Object.freeze({
+    async race<T>(operation: Promise<T>): Promise<T> {
+      signal.throwIfAborted();
+      // A pending `operation` that rejects after losing the race is still settled through
+      // `Promise.race`'s own handler, so it never becomes an unhandled rejection either — measured
+      // on Bun 1.3.14 and Node 20.3/26, where releasing a reader with a read outstanding rejects
+      // that read (`AbortError` on Bun, `TypeError` on Node).
+      return Promise.race([operation, aborted]);
+    },
+    release(): void {
+      signal.removeEventListener('abort', onAbort);
+    },
+  });
+}

packages/codec-json/src/json-serde.test.ts

@@ -32,6 +32,37 @@ async function rejection(promise: Promise<unknown>): Promise<unknown> {
   }
 }
 
+/** How long a raced abort is given to settle a parked read or write before the case fails. */
+const SETTLE_MS = 250;
+
+/**
+ * Fails with a named error instead of letting the runner time out, so a regression reads as "the
+ * abort never settled the call" rather than as a five-second stall with no diagnosis.
+ *
+ * `Promise.race` keeps a handler on `promise`, so a later rejection of the losing side is never an
+ * unhandled one.
+ */
+async function settleWithin<T>(promise: Promise<T>, ms: number): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const deadline = new Promise<never>((_resolve, reject) => {
+    timer = setTimeout(() => {
+      reject(new Error(`did not settle within ${String(ms)}ms`));
+    }, ms);
+  });
+  try {
+    return await Promise.race([promise, deadline]);
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+/** One macrotask, which is long enough for a drain loop to reach its second, parked read. */
+function untilParked(): Promise<void> {
+  return new Promise<void>(resolve => {
+    setTimeout(resolve, 5);
+  });
+}
+
 test('declares application/json as its wire media type', () => {
   expect(jsonSerde().mediaType).toBe('application/json');
 });
@@ -673,3 +704,140 @@ describe('the options argument stays optional on both stream methods', () => {
     ).toEqual({a: 1});
   });
 });
+
+// Module-scope, not describe-local: the pending-abort suite is split across sibling describes to
+// stay inside `max-lines-per-function`, and both halves need these.
+const PENDING_ABORT_SERDE = jsonSerde();
+const passthroughSchema: Schema<unknown> = {parse: (i: unknown) => i};
+
+/**
+ * Hands over one chunk and then never produces another, so the drain parks *inside*
+ * `reader.read()` — the state a between-chunks signal check structurally cannot observe.
+ */
+function stallingSource(
+  first: string,
+  onCancel?: () => void,
+): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(new TextEncoder().encode(first));
+    },
+    pull() {
+      return new Promise<never>(() => undefined);
+    },
+    cancel() {
+      onCancel?.();
+    },
+  });
+}
+
+describe('an abort that lands while a READ is pending (audit #67 / #79)', () => {
+  test('deserializeFrom settles with the caller reason and unlocks the source (SERDE-3)', async () => {
+    let cancelled = false;
+    const source = stallingSource('{"a":', () => {
+      cancelled = true;
+    });
+    const controller = new AbortController();
+    const reason = new Error('the caller gave up mid-drain');
+
+    const settled = rejection(
+      PENDING_ABORT_SERDE.deserializer.deserializeFrom(
+        source,
+        {schema: passthroughSchema},
+        {signal: controller.signal},
+      ),
+    );
+    await untilParked();
+    controller.abort(reason);
+
+    expect(await settleWithin(settled, SETTLE_MS)).toBe(reason);
+    // The whole point of the fix: the caller gets its stream back, still usable.
+    expect(source.locked).toBe(false);
+    expect(cancelled).toBe(false);
+  });
+
+  test('an abort with no reason surfaces the platform AbortError the seam documents', async () => {
+    const source = stallingSource('{"a":');
+    const controller = new AbortController();
+
+    const settled = rejection(
+      PENDING_ABORT_SERDE.deserializer.deserializeFrom(
+        source,
+        {schema: passthroughSchema},
+        {signal: controller.signal},
+      ),
+    );
+    await untilParked();
+    controller.abort();
+
+    expect(await settleWithin(settled, SETTLE_MS)).toMatchObject({
+      name: 'AbortError',
+    });
+    expect(source.locked).toBe(false);
+  });
+});
+
+describe('an abort that lands while a WRITE is pending (audit #67 / #79)', () => {
+  test('serializeTo settles with the caller reason and unlocks the sink (SERDE-3)', async () => {
+    let closed = false;
+    let aborted = false;
+    const sink = new WritableStream<Uint8Array>({
+      write() {
+        return new Promise<never>(() => undefined);
+      },
+      close() {
+        closed = true;
+      },
+      abort() {
+        aborted = true;
+      },
+    });
+    const controller = new AbortController();
+    const reason = new Error('the caller gave up mid-write');
+
+    const settled = rejection(
+      PENDING_ABORT_SERDE.serializer.serializeTo({a: 1}, sink, {
+        signal: controller.signal,
+      }),
+    );
+    await untilParked();
+    controller.abort(reason);
+
+    expect(await settleWithin(settled, SETTLE_MS)).toBe(reason);
+    expect(sink.locked).toBe(false);
+    expect(closed).toBe(false);
+    expect(aborted).toBe(false);
+  });
+
+  test('a signal that never fires leaves both directions unchanged', async () => {
+    const controller = new AbortController();
+    const source = new ReadableStream<Uint8Array>({
+      start(streamController) {
+        streamController.enqueue(new TextEncoder().encode('{"a":'));
+        streamController.enqueue(new TextEncoder().encode('1}'));
+        streamController.close();
+      },
+    });
+    const written: string[] = [];
+    const sink = new WritableStream<Uint8Array>({
+      write(chunk) {
+        written.push(new TextDecoder().decode(chunk));
+      },
+    });
+
+    expect(
+      await PENDING_ABORT_SERDE.deserializer.deserializeFrom(
+        source,
+        {schema: passthroughSchema},
+        {signal: controller.signal},
+      ),
+    ).toEqual({a: 1});
+    await PENDING_ABORT_SERDE.serializer.serializeTo({a: 1}, sink, {
+      signal: controller.signal,
+    });
+
+    expect(written.join('')).toBe('{"a":1}');
+    expect(source.locked).toBe(false);
+    expect(sink.locked).toBe(false);
+  });
+});