feat(download): @stitchapi/download — batch downloader, plus the core fixes it surfaced

README.md

@@ -443,6 +443,7 @@ This repository is a [pnpm](https://pnpm.io) workspace. The published library is
 <tr><td><a href="packages/pino"><code>@stitchapi/pino</code></a></td><td>The stitch event stream as structured Pino logs</td></tr>
 <tr><td><a href="packages/sentry"><code>@stitchapi/sentry</code></a></td><td>Stitch events as Sentry breadcrumbs, with error capture</td></tr>
 <tr><th colspan="2">Surfaces</th></tr>
+<tr><td><a href="packages/download"><code>@stitchapi/download</code></a></td><td>Batch file downloads with FIFO concurrency, cancel, and ETA</td></tr>
 <tr><td><a href="packages/shell"><code>@stitchapi/shell</code></a></td><td>Run a static local command as a stitch (injection-proof)</td></tr>
 <tr><th colspan="2">Cache fingerprint adapters</th></tr>
 <tr><td><a href="packages/fingerprint-arktype"><code>@stitchapi/fingerprint-arktype</code></a></td><td>Cache-fingerprint strategy for ArkType schemas</td></tr>

packages/core/README.md

@@ -936,6 +936,7 @@ StitchAPI ships thin, peer-dependency integration packages — server frameworks
 <tr><td><a href="packages/pino"><code>@stitchapi/pino</code></a></td><td>The stitch event stream as structured Pino logs</td></tr>
 <tr><td><a href="packages/sentry"><code>@stitchapi/sentry</code></a></td><td>Stitch events as Sentry breadcrumbs, with error capture</td></tr>
 <tr><th colspan="2">Surfaces</th></tr>
+<tr><td><a href="packages/download"><code>@stitchapi/download</code></a></td><td>Batch file downloads with FIFO concurrency, cancel, and ETA</td></tr>
 <tr><td><a href="packages/shell"><code>@stitchapi/shell</code></a></td><td>Run a static local command as a stitch (injection-proof)</td></tr>
 <tr><th colspan="2">Cache fingerprint adapters</th></tr>
 <tr><td><a href="packages/fingerprint-arktype"><code>@stitchapi/fingerprint-arktype</code></a></td><td>Cache-fingerprint strategy for ArkType schemas</td></tr>

packages/core/src/download.ts

@@ -9,6 +9,7 @@
 // Bundle-frugal (Decision 10): reached only through the `download` subpath; `import { stitch }`
 // pulls in none of it.
 import type { InputOf } from './infer';
+import { acceptsStatus } from './resilience';
 import { seam as makeSeam } from './seam';
 import { makeStitch } from './stitch';
 import { verdictOf } from './surface';
@@ -120,6 +121,47 @@ export const downloadSurface: Surface<StitchInput, DownloadResult> & {
     interpret: (res, cfg): SurfaceOutcome<DownloadResult> => {
         const failure = verdictOf(res, cfg);
         if (failure) return failure;
+        // Past the declarative verdict comes the surface's OWN rule, which the composed `verdictOf`
+        // cannot express: a buffered download resolves to a WHOLE Blob, so a success must be a
+        // status that DEFINITIONALLY carries the whole entity — `200 OK`, or `204 No Content` (a
+        // legitimately empty body). `verdictOf` rules on `>= 400`, so every other sub-400 status
+        // arrives here already deemed acceptable, and two of them would otherwise be handed back as
+        // a downloaded file:
+        //   • `206 Partial Content` — `download` never sends a `Range`, so a 206 is a partial body
+        //     the server volunteered (a range-serving proxy/CDN, a resumed-and-mismatched cache);
+        //     accepting it hands the caller a truncated file as if it were the whole thing.
+        //   • a `3xx` the adapter could not resolve — a chain that hit the hop cap, or one with no
+        //     `Location` (http-adapter.ts `followRedirects` hands the last 3xx back rather than
+        //     throwing). There is no entity at all behind it. A narrower `status === 206` rule would
+        //     let that through as an empty Blob; test/gaps/download-redirect-loop.spec.ts pins it.
+        // Knowingly OUTSIDE the allow-list: `203 Non-Authoritative Information` and `226 IM Used`
+        // do carry a complete body, and are rejected anyway — rare enough, and not distinguishable
+        // from a partial without trusting the transforming intermediary, that the strict reading
+        // wins with `accept` as the documented way back in.
+        //
+        // `verdict.accept` is that opt-out, and reading it HERE is what keeps the rule honest under
+        // ADR 0022: `classifyStatus` asks `accept` only at `>= 400`, so without this line a caller
+        // who declared `206` NORMAL would still be rejected — the surface overriding an explicit
+        // declaration rather than ruling where the caller made none. It cannot contradict the two
+        // statuses ahead of it either way: `accept` only ever WIDENS (surface.ts), so a match makes
+        // the whole conjunction false wherever it sits. The order below is evaluation COST — no
+        // predicate is allocated on the hot `200` path — not semantics.
+        if (
+            res.status !== 200 &&
+            res.status !== 204 &&
+            !acceptsStatus(cfg.verdict?.accept)(res.status)
+        )
+            return {
+                ok: false,
+                message: `download: expected a complete body (200/204) but got HTTP ${res.status}${res.status === 206 ? ' — a partial (206) response was not requested (no Range header is ever sent)' : ''}`,
+                status: res.status,
+            };
+        // No `blob.size` vs `content-length` cross-check here: a truncated body (Content-Length
+        // advertises more bytes than are sent) never reaches this hook. undici HANGS on a clean-FIN
+        // short read until the caller's `timeout` fires, and REJECTS on an abrupt socket close — so a
+        // truncation surfaces as a timeout / transport error, never as a short Blob delivered here. A
+        // length check would be dead code (and a gzip/br body legitimately has `size != content-length`
+        // once decoded — it would be a false positive). See test/gaps/download-truncation.spec.ts.
         const data: DownloadResult = { blob: res.body as Blob };
         const filename =
             filenameFromDisposition(res.headers['content-disposition']) ??

packages/core/src/engine.ts

@@ -280,6 +280,32 @@ const hostKey = (req: AdapterRequest, cfg: ResolvedStitchConfig): string => {
 // skip it), so the full `response` can't leak into a JSONL/console log. See `drain` in stitch.ts.
 export const ERROR_SOURCE = Symbol('stitch.errorSource');
 
+// The single writer of that channel: pin the live error behind an event so the awaited / `.safe()`
+// path can recover it. `defineProperty` defaults to non-configurable, so a second pin on the same
+// event throws — routing every producer through one helper is what makes "written once" structural,
+// rather than an accident of mutually-exclusive branches at the call sites.
+const pinSource = <E extends object>(evt: E, err: unknown): E => {
+    Object.defineProperty(evt, ERROR_SOURCE, { value: err, enumerable: false });
+    return evt;
+};
+
+// Must the live error ride that channel to the caller, or can the awaited path rebuild it from the
+// event alone? One question, asked once — the mirror of `rebuildError`'s `isOurs` in stitch.ts,
+// which decides what to do with whatever arrives. Three kinds ride:
+//   • an error carrying `.response` (issue #155) — the full error response (body + url) for
+//     `StitchError.body`/`.url`. Tested FIRST and untyped, because a BYO adapter may throw a
+//     non-Error bag that carries one.
+//   • a delegate-backoff RateLimitError (issue #145) — re-thrown as THAT instance, class identity
+//     and `retryAfter` intact. (#662 folds this arm into the next by making it a StitchError.)
+//   • any other foreign Error — a bare transport/internal failure (undici UND_ERR_SOCKET /
+//     ECONNRESET / DNS, an AbortError, …) whose `.cause` becomes `StitchError.cause`, so a caller
+//     can tell a socket reset from a generic "fetch failed". A StitchError the engine minted itself
+//     is excluded: it keeps being rebuilt from the event (unchanged behaviour).
+const ridesThrough = (err: unknown): boolean =>
+    (err as { response?: AdapterResponse }).response !== undefined ||
+    err instanceof RateLimitError ||
+    (err instanceof Error && !(err instanceof StitchError));
+
 // A non-enumerable channel for the retained pre-validation body (`.inspect()`, ADR 0016). Like
 // ERROR_SOURCE, non-enumerable means a trace sink (Object.entries / JSON.stringify) never sees it, so
 // the unredacted body can't leak into a JSONL/console log. Rides the `result` event on the success
@@ -335,10 +361,7 @@ function contractViolationEvt(
             attempts: state.attempts,
         });
         Object.defineProperty(err, RAW_BODY, { value: raw, enumerable: false });
-        Object.defineProperty(evt, ERROR_SOURCE, {
-            value: err,
-            enumerable: false,
-        });
+        pinSource(evt, err);
     }
     return evt;
 }
@@ -353,25 +376,13 @@ function errEvt(err: unknown, name: string, attempts: number): StitchEvent {
         at: now(),
     };
     if (e.status !== undefined) evt.status = e.status;
-    // Delegate-backoff signal: stamp the structured `retryAfter` onto the event (so `.stream()`
-    // consumers get it) and pin the live RateLimitError so the awaited path re-throws it intact.
-    if (err instanceof RateLimitError) {
-        if (err.retryAfter !== undefined) evt.retryAfter = err.retryAfter;
-        Object.defineProperty(evt, ERROR_SOURCE, {
-            value: err,
-            enumerable: false,
-        });
-    } else if ((err as { response?: AdapterResponse }).response !== undefined) {
-        // HTTP failure (issue #155): the thrown error carries the full `.response` (body + url).
-        // Pin it on the SAME non-enumerable channel so `drain`/`asStitchError` can populate
-        // `StitchError.body`/`.url`. Non-enumerable ⇒ the body never serialises into a trace sink
-        // (privacy preserved); the enumerable `status`/`message` are all a sink sees.
-        Object.defineProperty(evt, ERROR_SOURCE, {
-            value: err,
-            enumerable: false,
-        });
-    }
-    return evt;
+    // Delegate-backoff signal: the structured `retryAfter` is stamped onto the EVENT so `.stream()`
+    // consumers see it; the live instance itself rides ERROR_SOURCE below for the awaited path.
+    if (err instanceof RateLimitError && err.retryAfter !== undefined)
+        evt.retryAfter = err.retryAfter;
+    // Non-enumerable ⇒ nothing pinned here ever serialises into a trace sink (privacy preserved);
+    // the enumerable `status`/`message` are all a sink sees.
+    return ridesThrough(err) ? pinSource(evt, err) : evt;
 }
 const doneEvt = (ok: boolean, t0: number, attempts: number): StitchEvent => ({
     type: 'done',

packages/core/src/stitch.ts

@@ -493,11 +493,13 @@ function rebuildError(ev: Extract<StitchEvent, { type: 'error' }>): Error {
             ? undefined
             : (source as { response?: { body?: unknown; url?: string } })
                   .response;
-    if (
-        res !== undefined &&
-        !(source instanceof StitchError) &&
-        !(source instanceof RateLimitError)
-    ) {
+    // Did the engine mint this error itself, or catch a foreign one? The two arms below ask that
+    // one question negatively and then positively, so ask it once.
+    // (#662 makes RateLimitError a StitchError subclass, at which point the second disjunct is
+    // redundant and this collapses to a bare `source instanceof StitchError`.)
+    const isOurs =
+        source instanceof StitchError || source instanceof RateLimitError;
+    if (res !== undefined && !isOurs) {
         return new StitchError(ev.message, {
             status: ev.status,
             attempts: ev.attempts,
@@ -506,12 +508,17 @@ function rebuildError(ev: Extract<StitchEvent, { type: 'error' }>): Error {
             cause: source,
         });
     }
-    return (
-        source ??
-        new StitchError(ev.message, {
-            status: ev.status,
-            attempts: ev.attempts,
-        })
+    // A pass-through terminal — a delegate-backoff RateLimitError or a contract-violation StitchError
+    // (`.inspect()` retain path) — is re-surfaced UNCHANGED.
+    if (isOurs) return source;
+    // Otherwise a StitchError from the event, carrying any bare transport/internal `source` the engine
+    // pinned (an undici UND_ERR_SOCKET / ECONNRESET, a DNS failure, an AbortError, …) as `cause` — so
+    // callers can read `err.cause` (and its `.code`) to tell a socket reset from a generic "fetch
+    // failed". Absent a source, no cause (unchanged). `cause` is non-enumerable, so it never leaks into
+    // a trace sink.
+    return new StitchError(
+        ev.message,
+        compact({ status: ev.status, attempts: ev.attempts, cause: source }),
     );
 }
 

packages/core/test/gaps/download-concurrency-ceiling.spec.ts

@@ -0,0 +1,86 @@
+// Pins P1 (download-test-rig-spec §2.8): under a shared host throttle, no more than `k` download
+// requests are OPEN ON THE WIRE at any instant. Asserted from the mock server's own concurrency probe
+// (`maxOpen`) — ACTUAL on-the-wire overlap, not "we decided to be concurrent". This is the ceiling the
+// future @stitchapi/download batch layer stands on: it sequences a caller's LIST, but the per-request
+// admission cap is core `throttle`, and it must genuinely bound the wire.
+//
+// N INDEPENDENT download() stitches (a batch of distinct files) share ONE host-keyed concurrency
+// budget via `throttle: { concurrency: k, pool: 'host' }` — P2, the cross-instance pooling substrate
+// (resilience.ts `hostStates`, keyed by URL host). Each route holds its response with `ttfbDelay` so
+// the k admitted slots stay open together long enough to observe the peak.
+//
+// Real-timer, LOOSE bounds (a socket test): the hold (150ms) is wide vs. loopback dispatch jitter
+// (two near-simultaneous fetches land sub-ms apart), so the k concurrent slots reliably overlap; no
+// exact timing is asserted, only a COUNT. Held sockets are force-destroyed at teardown (the server
+// tracks live sockets), so the suite exits.
+//
+// Deferred (needs the unbuilt @stitchapi/download batch orchestrator, NOT raw download()+throttle):
+//   • FIFO admission ORDER — that queued items (k+1)…N start in enqueue order as slots free (P6). The
+//     limiter serves concurrency waiters FIFO, but PROVING per-item order needs the batch API's stable
+//     item identity + start events, which don't exist yet.
+//   • Aggregate progress / ETA across the N concurrent streams (P8) — a batch-level roll-up, not a
+//     property of a single download().
+// Here we pin only the wire CEILING + host pooling those features are built on.
+import { download } from '../../src/download';
+import { startMockServer } from '../support/mock-server';
+import type { MockServer } from '../support/mock-server';
+
+let server: MockServer;
+beforeAll(async () => {
+    server = await startMockServer();
+});
+afterAll(async () => {
+    await server.close();
+});
+beforeEach(() => {
+    server.reset();
+});
+
+const blobText = async (b: Blob): Promise<string> =>
+    new TextDecoder().decode(await b.arrayBuffer());
+
+test('≤ k download requests are open on the wire at once under a shared host throttle', async () => {
+    const K = 2;
+    const N = 6;
+    const paths = Array.from({ length: N }, (_, i) => `/file-${i}`);
+    for (const p of paths)
+        server.route('GET', p, {
+            statuses: [200],
+            rawBody: `contents-of${p}`,
+            ttfbDelay: 150, // hold each admitted slot open long enough to observe the overlap
+        });
+
+    // N INDEPENDENT download stitches (distinct files), all sharing one host-keyed budget of K.
+    const calls = paths.map((p) =>
+        download({
+            baseUrl: server.url,
+            path: p,
+            throttle: { concurrency: K, pool: 'host' },
+            retry: { attempts: 1 },
+        }),
+    );
+
+    // `Promise.all` subscribes to every cold StitchResult in the same tick → all N are launched
+    // concurrently; the throttle, not the test, decides how many actually reach the wire.
+    const results = await Promise.all(calls.map((c) => c()));
+
+    // THE CEILING: the server never saw more than K requests open on the wire at any instant.
+    expect(server.maxOpen()).toBeLessThanOrEqual(K);
+    // …and the budget was genuinely SATURATED (real concurrency, not accidental serialization) — else
+    // "≤ K" would pass vacuously at 1. With N > K and a 150ms hold, the peak must reach K.
+    expect(server.maxOpen()).toBe(K);
+    // Every file still downloaded correctly and completely — the cap PACES work, it never drops it.
+    expect(results).toHaveLength(N);
+    for (let i = 0; i < N; i++)
+        expect(await blobText(results[i]!.blob)).toBe(
+            `contents-of${paths[i]!}`,
+        );
+    // Sanity: all N requests actually reached the server (nothing was silently dropped by the gate).
+    expect(server.callCount()).toBe(N);
+    // The probe also records a per-request arrival timestamp, in receipt order (the observable the
+    // future batch layer's ETA/progress math will read); one per request, monotonically non-decreasing.
+    const ts = server.arrivals();
+    expect(ts).toHaveLength(N);
+    for (let i = 1; i < ts.length; i++)
+        expect(ts[i]!).toBeGreaterThanOrEqual(ts[i - 1]!);
+});