diff --git a/packages/ai/src/task/base/AiTask.ts b/packages/ai/src/task/base/AiTask.ts index 8e9a086c7..2ce34c954 100644 --- a/packages/ai/src/task/base/AiTask.ts +++ b/packages/ai/src/task/base/AiTask.ts @@ -370,7 +370,10 @@ export class AiTask< /** * Validates that model inputs are valid ModelConfig objects. */ - public override async validateInput(input: Input): Promise { + public override async validateInput( + input: Input, + skipPorts?: ReadonlySet + ): Promise { const inputSchema = this.inputSchema(); if (typeof inputSchema === "boolean") { if (inputSchema === false) { @@ -422,7 +425,7 @@ export class AiTask< } } - return super.validateInput(input); + return super.validateInput(input, skipPorts); } public override async narrowInput(input: Input, registry: ServiceRegistry): Promise { diff --git a/packages/ai/src/task/generation/ImageEditTask.ts b/packages/ai/src/task/generation/ImageEditTask.ts index b0b4cf454..7c57de209 100644 --- a/packages/ai/src/task/generation/ImageEditTask.ts +++ b/packages/ai/src/task/generation/ImageEditTask.ts @@ -93,8 +93,11 @@ export class ImageEditTask extends AiImageOutputTask { - const ok = await super.validateInput(input); + public override async validateInput( + input: ImageEditTaskInput, + skipPorts?: ReadonlySet + ): Promise { + const ok = await super.validateInput(input, skipPorts); if (!ok) return false; await this.validateProviderImageInput(input); return true; diff --git a/packages/ai/src/task/generation/ImageGenerateTask.ts b/packages/ai/src/task/generation/ImageGenerateTask.ts index 13f452fc1..091be6e8e 100644 --- a/packages/ai/src/task/generation/ImageGenerateTask.ts +++ b/packages/ai/src/task/generation/ImageGenerateTask.ts @@ -66,8 +66,11 @@ export class ImageGenerateTask extends AiImageOutputTask< return ImageGenerateOutputSchema as DataPortSchema; } - public override async validateInput(input: ImageGenerateTaskInput): Promise { - const ok = await super.validateInput(input); + public override async validateInput( + input: ImageGenerateTaskInput, + skipPorts?: ReadonlySet + ): Promise { + const ok = await super.validateInput(input, skipPorts); if (!ok) return false; await this.validateProviderImageInput(input); return true; diff --git a/packages/indexeddb/src/job-queue/IndexedDbQueueStorage.ts b/packages/indexeddb/src/job-queue/IndexedDbQueueStorage.ts index 16011b3ea..7ff225309 100644 --- a/packages/indexeddb/src/job-queue/IndexedDbQueueStorage.ts +++ b/packages/indexeddb/src/job-queue/IndexedDbQueueStorage.ts @@ -159,8 +159,7 @@ export class IndexedDbQueueStorage implements IQueueStorage { request.onsuccess = () => { const job = request.result as - | (JobStorageFormat & Record) - | undefined; + (JobStorageFormat & Record) | undefined; if (job && job.queue === this.queueName && this.matchesPrefixes(job)) { resolve(job); } else { @@ -416,8 +415,7 @@ export class IndexedDbQueueStorage implements IQueueStorage { const existing = getReq.result as - | (JobStorageFormat & Record) - | undefined; + (JobStorageFormat & Record) | undefined; if (!existing || existing.queue !== this.queueName || !this.matchesPrefixes(existing)) { // Contract: complete() must silently no-op (not throw) when the row is // missing or belongs to another queue. Lease-expiry races can legitimately @@ -654,8 +652,7 @@ export class IndexedDbQueueStorage implements IQueueStorage { request.onsuccess = () => { const job = request.result as - | (JobStorageFormat & Record) - | undefined; + (JobStorageFormat & Record) | undefined; if (job && this.matchesPrefixes(job)) { resolve(job.output ?? null); } else { diff --git a/packages/job-queue/src/job/JobQueueServer.ts b/packages/job-queue/src/job/JobQueueServer.ts index ea8eccbd7..3d793ed00 100644 --- a/packages/job-queue/src/job/JobQueueServer.ts +++ b/packages/job-queue/src/job/JobQueueServer.ts @@ -13,8 +13,8 @@ import type { JobStorageFormat, QueueChangePayload } from "../queue-storage/IQue import { JobStatus } from "../queue-storage/IQueueStorage"; import type { DeadLetter } from "./DeadLetter"; import { Job, JobClass } from "./Job"; -import type { StreamEventLike } from "./JobQueueEventListeners"; import { JobQueueClient } from "./JobQueueClient"; +import type { StreamEventLike } from "./JobQueueEventListeners"; import { JobQueueWorker } from "./JobQueueWorker"; import { storageToClass } from "./JobStorageConverters"; diff --git a/packages/storage/src/tabular/HttpTabularProxyStorage.ts b/packages/storage/src/tabular/HttpTabularProxyStorage.ts index 855cb81a9..9dd51fbd8 100644 --- a/packages/storage/src/tabular/HttpTabularProxyStorage.ts +++ b/packages/storage/src/tabular/HttpTabularProxyStorage.ts @@ -44,8 +44,7 @@ export interface HttpTabularProxyOptions< readonly schema: Schema; readonly primaryKey: PrimaryKeyNames; readonly indexes?: readonly ( - | keyof Schema["properties"] - | readonly (keyof Schema["properties"])[] + keyof Schema["properties"] | readonly (keyof Schema["properties"])[] )[]; /** Optional base path. Defaults to `/api/storage`. Trailing slashes are stripped. */ readonly basePath?: string; diff --git a/packages/storage/src/tabular/HuggingFaceTabularStorage.ts b/packages/storage/src/tabular/HuggingFaceTabularStorage.ts index 0f704f256..a2893ae40 100644 --- a/packages/storage/src/tabular/HuggingFaceTabularStorage.ts +++ b/packages/storage/src/tabular/HuggingFaceTabularStorage.ts @@ -145,8 +145,7 @@ export class HuggingFaceTabularStorage< schema, primaryKeyNames, (options?.indexes ?? []) as readonly ( - | keyof NoInfer - | readonly (keyof NoInfer)[] + keyof NoInfer | readonly (keyof NoInfer)[] )[], "never", // HF datasets don't support client-provided keys. tabularMigrations, diff --git a/packages/storage/src/tabular/ITabularStorage.ts b/packages/storage/src/tabular/ITabularStorage.ts index 8d77f4fe7..22de0323d 100644 --- a/packages/storage/src/tabular/ITabularStorage.ts +++ b/packages/storage/src/tabular/ITabularStorage.ts @@ -96,12 +96,7 @@ export interface TabularSubscribeOptions { } export type JSONValue = - | string - | number - | boolean - | null - | JSONValue[] - | { [key: string]: JSONValue }; + string | number | boolean | null | JSONValue[] | { [key: string]: JSONValue }; export type SearchOperator = "=" | "<" | "<=" | ">" | ">="; diff --git a/packages/task-graph/src/EXECUTION_MODEL.md b/packages/task-graph/src/EXECUTION_MODEL.md index 48dee1ba2..33b1ba479 100644 --- a/packages/task-graph/src/EXECUTION_MODEL.md +++ b/packages/task-graph/src/EXECUTION_MODEL.md @@ -287,12 +287,12 @@ Failed tasks are never cached — only `Ok` results reach `saveOutput`. `saveOut Binary output ports whose bytes were piped into a stream-capable cache carry a branded `CacheRef` in the cached row. On a **cache hit**, the runner mirrors the fresh-run event contract, driven by two graph-computed consumer hints (`IRunConfig.hasStreamingConsumers` / `hasMaterializingConsumers`): - **Stream-capable consumer** (`x-stream: "binary"` on both ends of an edge): the cached bytes replay as chunked `binary-delta` events, pull-paced from the repository's streaming reader (`getOutputStreamByRef`), so memory stays bounded by the read chunk size. The finish event keeps the ref at the port. -- **Materializing consumer** (target port cannot consume the stream): the ref hydrates into the **enriched finish event** as a `Blob`/`ArrayBuffer` (per the port's `format`), exactly what a fresh run's accumulator would have delivered. The *returned* output still carries the small ref. +- **Materializing consumer** (target port cannot consume the stream): the ref hydrates into the **enriched finish event** as a `Blob`/`ArrayBuffer` (per the port's `format`), exactly what a fresh run's accumulator would have delivered. The _returned_ output still carries the small ref. - **No consumers**: no reads are performed; the synthetic finish carries the ref unchanged (callers resolve via `resolveOutput` / `resolveJobOutputStream`). **Rows store the wire form**: the cached row always carries the `CacheRef`, never inline bytes — JSON-row backings would destroy an inline `Blob`/`ArrayBuffer` (`JSON.stringify(Blob)` is `{}`). Below-threshold hydration to inline bytes applies to the value **returned to the caller**, identically on fresh runs and cache hits. -**Single binary port**: the cache sink keys bytes by `(taskType, inputs)` with no port axis, so cache-streaming supports exactly one binary output port. Tasks with multiple binary ports take the accumulation path (enforced in both `StreamPump.canStreamBinaryToCache` and `CacheCoordinator.getBinaryRefSinksByPolicy`); their inline outputs are only safely cacheable by non-JSON-row backings until per-port refs land. +**Single binary port**: the legacy cache sink keys bytes by `(taskType, inputs)` with no port axis, so this path supports exactly one binary output port. Tasks with multiple binary ports take the accumulation path (enforced in both `StreamPump.canStreamBinaryToCache` and `CacheCoordinator.getBinaryRefSinksByPolicy`) — unless the run opts into the per-port sink path below, which has a port axis and no such limit. **Self-healing dangling refs**: when a ref needed for replay or hydration no longer resolves (blob evicted, cache cleared), the hit converts into a **miss** — the task re-executes and rewrites both the row and the bytes. No events are emitted before all refs are validated. @@ -302,6 +302,99 @@ Binary output ports whose bytes were piped into a stream-capable cache carry a b `FsFolderTaskOutputRepository` (node/bun) is the production streaming backing: JSON rows via `FsFolderTabularStorage`, bytes as sidecar files under `/blobs/` written incrementally and published by atomic rename — `_.bin`, so a re-run overwrites rather than leaks. Two instances over one folder interoperate (the cross-process read story). +### Per-port stream sinks and the no-accumulation passthrough + +Everything above generalizes from "one binary port" to **every delta stream mode** +(`append`, `object`, `binary`) behind an opt-in run flag, +`TaskGraphRunConfig.noAccumulation` (default off — off is byte-identical to the +accumulation path). + +**Per-port sinks (cache as tee).** When the flag is on, the task is cacheable, +and the cache backing implements the port-aware stream writer +(`saveOutputStreamPort`, advertised via `supportsStreamingPorts()`), the runner +builds one `StreamSink` per streaming output port. `StreamProcessor` encodes +each port's deltas through that mode's codec (`append` → UTF-8 text, `object` → +NDJSON delta log, `binary` → identity bytes) and routes the bytes to the sink +instead of buffering them into an enriched finish event. Each port's slot in the +returned output carries its own `CacheRef`; the cached row stores those refs +(wire form), and below-threshold hydration to inline values applies to the value +returned to the caller, as on the binary path. Backings without +`saveOutputStreamPort` — inline-only backings — never see this path: the task +falls back to accumulation and its outputs are cached inline as before. + +**Skippable edge materialization.** An edge qualifies as a _passthrough edge_ +when the flag is on, the edge carries a live stream with no transforms, source +and target ports declare the **same** delta stream mode, the target is itself a +streamable task (it implements `executeStream`; only streamable tasks receive +`ctx.inputStreams`) without a subgraph, the source port has exactly **one** +consumer, and the target port does not set `x-validate-stream: true`. Such an +edge skips the full-speed materialize drain entirely: the consumer takes its +data from the live event stream (handed to its `executeStream` via +`ctx.inputStreams`, without a tee — nothing else will read the edge), and the +edge's settled value is set from the producer's result when it finishes (the +per-port `CacheRef`, or the inline value a below-threshold ref was rehydrated +to). Every non-qualifying edge — transforms, mode mismatch, fan-out, `*` +edges, non-streamable or subgraph targets — falls back to today's drain, which +is correct, just without the memory and pacing win. + +**Caching a stream-fed consumer.** A consumer reading a live stream computes +its cache key while the streamed port is unsettled (`CacheRef` or nothing in +the slot), so the streamed content cannot contribute to the key. Rather than +let two runs that differ only in stream payload collide on one cache entry, +the runner disables caching (`kind: "none"`) for any run consuming a live +stream at an unsettled port. Drained edges settle the value before key +computation and keep caching as usual. + +**Validation of stream-wired inputs.** Whole-value input validation is a +settled-value concept, and a stream-wired port has no settled value while the +stream is live — its static slot holds a `CacheRef` (or nothing) until finish. +The runner therefore exempts such a port from validation when (and only when) +its slot is a ref or `undefined`; a port that already carries a materialized +value is validated as usual. A target port can opt back in with +`x-validate-stream: true`, which also disqualifies its edge from the +passthrough so the drain materializes a validatable value. + +**All-mode backpressure.** On a passthrough edge the producer is paced to the +consumer's read rate by a per-port `BackpressureGate` owned by the graph +runner: each event enqueued onto the edge stream charges the gate with its +`streamEventCost` (UTF-8 bytes for text deltas, JSON-encoded length for object +deltas, raw byte length for binary), and each event the consumer reads credits +it back. After emitting a delta, `StreamProcessor` awaits the gate for that +port (threaded down as `IRunConfig.edgeBackpressure`, so the task layer stays +edge-agnostic); once buffered cost reaches the high-water mark — +`TaskGraphRunConfig.streamHighWaterBytes`, defaulting to the binary router's +8 MiB — the producer parks until the consumer drains below the mark. Producer +completion, abort, error, and consumer termination all close the gate, so a +parked producer can never be orphaned. A gate is built only when the consumer +can make read progress while the producer is parked: if any OTHER edge into +the consumer is sourced from the producer or one of its descendants (a drained +edge, a mode-mismatched edge, a static-value edge), that edge settles only +after the producer finishes, so gating would deadlock the pair — such +consumers keep the ungated passthrough (correct, just unpaced). Tasks that +emit through a side channel can cooperate explicitly via `ctx.backpressure()`, +which awaits both the cache-sink routers and every edge gate. + +**Producer failure mid-stream.** A producer FAILURE (not abort) enqueues an +in-stream error event on every attached edge stream before closing it, so a +drained edge materializes the failure — a consumer already dispatched +(unblocked at STREAMING) fails with the producer's error instead of +completing, and caching, an output derived from truncated input. Abort keeps +the graceful close: the run-level abort cascade is already tearing everything +down. + +**Fan-out limitation.** Precise pacing is single-consumer by design. A source +port feeding two or more consumers keeps the tee'd drain: every consumer still +receives all events in order, but pacing is best-effort — no gate bounds the +producer's lead. + +**Cache hit ≡ fresh run.** On a cache hit for a task with per-port refs and a +same-mode streaming consumer, the runner replays each port's cached bytes +through the mode's codec as delta events, so downstream tasks observe the same +event sequence as a fresh run. Replay honors the same consumer-edge gate as a +fresh run (each emitted delta awaits `edgeBackpressure`); ungated consumers +replay at read speed. Materializing consumers receive hydrated values in the +enriched finish event, exactly as on the binary path. + ### Durable execution model A run is an atomic unit on a single worker. When the worker crashes: diff --git a/packages/task-graph/src/cache/BinaryPortCodec.ts b/packages/task-graph/src/cache/BinaryPortCodec.ts index de09d8a07..8ee2b6c74 100644 --- a/packages/task-graph/src/cache/BinaryPortCodec.ts +++ b/packages/task-graph/src/cache/BinaryPortCodec.ts @@ -35,8 +35,14 @@ function bytesToBase64(bytes: Uint8Array): string { if (typeof Buffer !== "undefined") { return Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength).toString("base64"); } + // Block-wise String.fromCharCode: one call per 32 KiB instead of per byte — + // a per-byte loop is minutes of main-thread jank on multi-MB payloads, and + // spreading the whole array in one call overflows the argument stack. + const BLOCK = 0x8000; let bin = ""; - for (let i = 0; i < bytes.length; i++) bin += String.fromCharCode(bytes[i] ?? 0); + for (let i = 0; i < bytes.length; i += BLOCK) { + bin += String.fromCharCode(...bytes.subarray(i, i + BLOCK)); + } return btoa(bin); } diff --git a/packages/task-graph/src/cache/resolveJobOutput.ts b/packages/task-graph/src/cache/resolveJobOutput.ts index 23880e979..a10c08534 100644 --- a/packages/task-graph/src/cache/resolveJobOutput.ts +++ b/packages/task-graph/src/cache/resolveJobOutput.ts @@ -24,8 +24,7 @@ export interface JobHandleLike { * `TaskOutputRepository` exposes). */ export type RefBacking = - | CacheRefResolver - | { readonly getOutputByRef?: (ref: CacheRef) => Promise }; + CacheRefResolver | { readonly getOutputByRef?: (ref: CacheRef) => Promise }; /** * Await a job's completion and hydrate every {@link CacheRef} inside its @@ -123,6 +122,28 @@ async function outputValueToStream( yield bytes; })(); } + // Inline values of the non-binary streamable modes: a below-threshold + // append port hydrates to a string, an object port to a plain object/array. + // Adapt them to the same byte encoding the ref form streams (UTF-8 text; + // one NDJSON line that folds back to the same value) so caller behavior + // does not flip on payload size. + if (typeof candidate === "string") { + const bytes = new TextEncoder().encode(candidate); + return (async function* () { + yield bytes; + })(); + } + if ( + port !== undefined && + candidate !== null && + typeof candidate === "object" && + (Array.isArray(candidate) || Object.getPrototypeOf(candidate) === Object.prototype) + ) { + const bytes = new TextEncoder().encode(JSON.stringify(candidate) + "\n"); + return (async function* () { + yield bytes; + })(); + } return undefined; } @@ -131,9 +152,11 @@ async function outputValueToStream( * output cache without materializing it. `port` selects the output port; * when omitted, the single branded {@link CacheRef} reachable in the output * is used (two or more refs without a port is an error; zero resolves - * `undefined`). Inline `Blob` / `ArrayBuffer` / `Uint8Array` values at a - * named port are adapted to a stream so callers don't branch on whether the - * reference threshold kept the value inline. + * `undefined`). Inline values at a named port — `Blob` / `ArrayBuffer` / + * `Uint8Array`, plus the `string` and plain object/array forms a + * below-threshold append/object ref hydrates to — are adapted to a stream so + * callers don't branch on whether the reference threshold kept the value + * inline. * * Portless discovery walks the ENTIRE output, including fields whose content * the job may have copied from untrusted input — a crafted branded ref shape diff --git a/packages/task-graph/src/cache/resolveRef.ts b/packages/task-graph/src/cache/resolveRef.ts index 1c167e4f7..5fbeec4fd 100644 --- a/packages/task-graph/src/cache/resolveRef.ts +++ b/packages/task-graph/src/cache/resolveRef.ts @@ -114,7 +114,43 @@ export async function resolveOutput( ): Promise { if (!hasMatchingRef(output, options?.filter, new WeakSet())) return output; const limit = createLimiter(options?.concurrency); - return (await walk(output, resolver, limit, options?.filter, new WeakSet())) as T; + // Acyclic values (the norm) memoize each container's resolution promise so a + // subtree shared between two slots resolves ONCE and both slots receive the + // same resolved copy — a plain visited-set would hand the second slot the + // original, unresolved object. Cyclic values keep the conservative + // visited-set behavior: cycles are returned by reference, unrewritten. + const memo = containsCycle(output) ? undefined : new WeakMap>(); + return (await walk(output, resolver, limit, options?.filter, new WeakSet(), memo)) as T; +} + +/** + * Depth-first cycle probe over the same container vocabulary as the walker + * (plain objects, Array, Map, Set; leaves are opaque). Gray/black coloring: + * an object seen again while still on the current path is a back-edge. + */ +function containsCycle( + value: unknown, + gray: WeakSet = new WeakSet(), + black: WeakSet = new WeakSet() +): boolean { + if (value === null || typeof value !== "object" || isLeaf(value)) return false; + const obj = value as object; + if (black.has(obj)) return false; + if (gray.has(obj)) return true; + gray.add(obj); + const children: Iterable = Array.isArray(value) + ? value + : value instanceof Map + ? value.values() + : value instanceof Set + ? value + : Object.values(value); + for (const child of children) { + if (containsCycle(child, gray, black)) return true; + } + gray.delete(obj); + black.add(obj); + return false; } /** @@ -173,7 +209,8 @@ async function walk( resolver: CacheRefResolver, limit: Limiter, filter: ((ref: CacheRef) => boolean) | undefined, - visited: WeakSet + visited: WeakSet, + memo: WeakMap> | undefined ): Promise { if (isCacheRef(value)) { if (filter && !filter(value)) return value; @@ -181,20 +218,46 @@ async function walk( } if (value === null || value === undefined) return value; if (isLeaf(value)) return value; - // Cycle / shared-subtree guard: a revisited object is returned by reference - // unchanged. Cycles are not rewritten — the caller keeps their original - // graph topology, including any unresolved refs the cycle contains. - if (typeof value === "object" && visited.has(value as object)) return value; + const obj = value as object; + if (memo) { + // Acyclic path: a shared subtree resolves once; every referencing slot + // awaits the same promise and receives the same resolved copy. + const pending = memo.get(obj); + if (pending !== undefined) return pending; + } else if (visited.has(obj)) { + // Cyclic fallback: a revisited object is returned by reference unchanged. + // Cycles are not rewritten — the caller keeps their original graph + // topology, including any unresolved refs the cycle contains. + return value; + } if (!hasMatchingRef(value, filter, new WeakSet())) return value; - if (typeof value === "object") visited.add(value as object); + if (memo) { + const promise = walkContainer(value, resolver, limit, filter, visited, memo); + memo.set(obj, promise); + return promise; + } + visited.add(obj); + return walkContainer(value, resolver, limit, filter, visited, memo); +} + +async function walkContainer( + value: unknown, + resolver: CacheRefResolver, + limit: Limiter, + filter: ((ref: CacheRef) => boolean) | undefined, + visited: WeakSet, + memo: WeakMap> | undefined +): Promise { if (Array.isArray(value)) { - return Promise.all(value.map((v) => walk(v, resolver, limit, filter, visited))); + return Promise.all(value.map((v) => walk(v, resolver, limit, filter, visited, memo))); } if (value instanceof Map) { const out = new Map(); const entries = Array.from(value.entries()); const resolved = await Promise.all( - entries.map(async ([k, v]) => [k, await walk(v, resolver, limit, filter, visited)] as const) + entries.map( + async ([k, v]) => [k, await walk(v, resolver, limit, filter, visited, memo)] as const + ) ); for (const [k, v] of resolved) out.set(k, v); return out; @@ -202,12 +265,12 @@ async function walk( if (value instanceof Set) { const out = new Set(); const resolved = await Promise.all( - Array.from(value).map((v) => walk(v, resolver, limit, filter, visited)) + Array.from(value).map((v) => walk(v, resolver, limit, filter, visited, memo)) ); for (const v of resolved) out.add(v); return out; } - if (typeof value === "object") { + if (typeof value === "object" && value !== null) { const source = value as Record; // Only plain objects (Object.prototype / null prototype) reach this // branch; class instances are screened out by isLeaf above and returned @@ -217,7 +280,7 @@ async function walk( // matches the input even though resolutions race. const keys = Object.keys(source); const resolvedValues = await Promise.all( - keys.map((k) => walk(source[k], resolver, limit, filter, visited)) + keys.map((k) => walk(source[k], resolver, limit, filter, visited, memo)) ); for (let i = 0; i < keys.length; i++) out[keys[i]!] = resolvedValues[i]; return out; diff --git a/packages/task-graph/src/cache/streamCodec.ts b/packages/task-graph/src/cache/streamCodec.ts index a451098fe..29216af4c 100644 --- a/packages/task-graph/src/cache/streamCodec.ts +++ b/packages/task-graph/src/cache/streamCodec.ts @@ -5,7 +5,7 @@ */ import type { StreamEvent, StreamMode } from "../task/StreamTypes"; -import { foldObjectDelta } from "../task/StreamTypes"; +import { foldObjectDelta, materializeBinary } from "../task/StreamTypes"; /** * Translates a single output port's streaming deltas to and from the ordered @@ -36,19 +36,32 @@ export interface StreamPortCodec { materialize(bytes: AsyncIterable, port: string): Promise; } +// TextEncoder is stateless — one shared instance serves every event on the +// delta hot path instead of a fresh allocation per encodeEvent call. +const utf8 = new TextEncoder(); + +/** Shared `encode` body: {@link StreamPortCodec.encodeEvent} applied over a stream. */ +async function* encodeEvents( + codec: StreamPortCodec, + events: AsyncIterable, + port: string +): AsyncIterable { + for await (const e of events) { + const bytes = codec.encodeEvent(e, port); + if (bytes) yield bytes; + } +} + const appendCodec: StreamPortCodec = { mode: "append", encodeEvent(event, port) { if (event.type === "text-delta" && event.port === port && event.textDelta) { - return new TextEncoder().encode(event.textDelta); + return utf8.encode(event.textDelta); } return undefined; }, - async *encode(events, port) { - for await (const e of events) { - const bytes = this.encodeEvent(e, port); - if (bytes) yield bytes; - } + encode(events, port) { + return encodeEvents(this, events, port); }, async *decode(bytes, port) { const dec = new TextDecoder(); @@ -72,15 +85,12 @@ const objectCodec: StreamPortCodec = { mode: "object", encodeEvent(event, port) { if (event.type === "object-delta" && event.port === port) { - return new TextEncoder().encode(JSON.stringify(event.objectDelta) + "\n"); + return utf8.encode(JSON.stringify(event.objectDelta) + "\n"); } return undefined; }, - async *encode(events, port) { - for await (const e of events) { - const bytes = this.encodeEvent(e, port); - if (bytes) yield bytes; - } + encode(events, port) { + return encodeEvents(this, events, port); }, decode: decodeObject, async materialize(bytes, port) { @@ -120,11 +130,8 @@ const binaryCodec: StreamPortCodec = { if (event.type === "binary-delta" && event.port === port) return event.binaryDelta; return undefined; }, - async *encode(events, port) { - for await (const e of events) { - const bytes = this.encodeEvent(e, port); - if (bytes) yield bytes; - } + encode(events, port) { + return encodeEvents(this, events, port); }, async *decode(bytes, port) { for await (const chunk of bytes) yield { type: "binary-delta", port, binaryDelta: chunk }; @@ -132,7 +139,7 @@ const binaryCodec: StreamPortCodec = { async materialize(bytes) { const parts: Uint8Array[] = []; for await (const chunk of bytes) parts.push(chunk); - return new Blob(parts as unknown as BlobPart[]); + return materializeBinary(parts, "blob"); }, }; diff --git a/packages/task-graph/src/storage/FsFolderTaskOutputRepository.ts b/packages/task-graph/src/storage/FsFolderTaskOutputRepository.ts index 413e1d149..ef1e0206e 100644 --- a/packages/task-graph/src/storage/FsFolderTaskOutputRepository.ts +++ b/packages/task-graph/src/storage/FsFolderTaskOutputRepository.ts @@ -7,7 +7,7 @@ import { FsFolderTabularStorage } from "@workglow/storage"; import { makeFingerprint } from "@workglow/util"; import { randomUUID } from "node:crypto"; -import { createReadStream, existsSync } from "node:fs"; +import { createReadStream, openSync } from "node:fs"; import { mkdir, open, readdir, readFile, rename, rm, stat } from "node:fs/promises"; import { join } from "node:path"; import type { CacheRef } from "../cache/CacheRef"; @@ -135,7 +135,17 @@ export class FsFolderTaskOutputRepository extends TaskOutputTabularRepository { try { try { for await (const chunk of chunks) { - await handle.write(chunk); + // write(2) may write fewer bytes than requested (quota/NFS/FUSE + // edges); loop until the whole chunk is on disk so the returned + // size — stamped into the CacheRef — never overstates the file. + let offset = 0; + while (offset < chunk.byteLength) { + const { bytesWritten } = await handle.write(chunk, offset, chunk.byteLength - offset); + if (bytesWritten <= 0) { + throw new Error(`Short write persisting blob ${name} (0 bytes accepted)`); + } + offset += bytesWritten; + } size += chunk.byteLength; } await handle.sync(); @@ -182,12 +192,23 @@ export class FsFolderTaskOutputRepository extends TaskOutputTabularRepository { override getOutputStreamByRef(ref: CacheRef): AsyncIterable | undefined { const path = this.blobPath(ref); - if (path === undefined || !existsSync(path)) return undefined; - return (async function* () { - for await (const chunk of createReadStream(path)) { - yield chunk as Uint8Array; - } - })(); + if (path === undefined) return undefined; + // Open the file descriptor NOW so a missing blob reports `undefined` (the + // clean cache-miss contract, matching getOutputByRef) instead of an ENOENT + // thrown mid-iteration; once open, a concurrent prune can no longer break + // the read (POSIX keeps the inode alive until the handle closes). + let fd: number; + try { + fd = openSync(path, "r"); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === "ENOENT") return undefined; + throw err; + } + // Return the ReadStream itself (it is an AsyncIterable): its async + // iterator's `return()` destroys the stream and releases the fd, so a + // caller that resolves several port streams and then abandons them (e.g. a + // replay that converts to a miss) can close without reading. + return createReadStream(path, { fd, autoClose: true }) as unknown as AsyncIterable; } override async deleteOutputByRef(ref: CacheRef): Promise { diff --git a/packages/task-graph/src/task-graph/EdgeMaterializer.ts b/packages/task-graph/src/task-graph/EdgeMaterializer.ts index 2b883c1c4..c6dfa3356 100644 --- a/packages/task-graph/src/task-graph/EdgeMaterializer.ts +++ b/packages/task-graph/src/task-graph/EdgeMaterializer.ts @@ -7,7 +7,6 @@ import { getLogger } from "@workglow/util"; import type { ImageValue } from "@workglow/util/media"; import { previewSource } from "@workglow/util/media"; -import { isCacheRef } from "../cache/CacheRef"; import type { ITask } from "../task/ITask"; import type { TaskInput, TaskOutput } from "../task/TaskTypes"; import { TaskStatus } from "../task/TaskTypes"; @@ -119,14 +118,16 @@ export class EdgeMaterializer { // non-idempotent transforms, so skip the whole post-materialisation step. if (dataflow.stream !== undefined) { // Exception: a no-accumulation passthrough edge is NOT drained - // downstream, so nothing else populates its value. When the source - // produced a per-port CacheRef, set it as the edge's durable value (no - // transforms by definition) so a consumer that reads the static slot - // resolves it via input hydration. A non-ref value is left alone. + // downstream, so nothing else populates its value. Set the settled + // slot from the producer's result (no transforms by definition): + // either the per-port CacheRef (large output — a static-slot reader + // resolves it via input hydration) or the inline value a + // below-threshold ref was rehydrated to. Without this, the edge's + // settled value would depend on output size. const noAccumulation = this.runner["noAccumulation"] === true; if (StreamPump.isNoAccumulationPassthroughEdge(this.graph, dataflow, noAccumulation)) { - const ref = (results as Record)[dataflow.sourceTaskPortId]; - if (isCacheRef(ref)) dataflow.value = ref; + const value = (results as Record)[dataflow.sourceTaskPortId]; + if (value !== undefined) dataflow.value = value; } continue; } diff --git a/packages/task-graph/src/task-graph/StreamPump.ts b/packages/task-graph/src/task-graph/StreamPump.ts index 5d9b9e92c..41621cc33 100644 --- a/packages/task-graph/src/task-graph/StreamPump.ts +++ b/packages/task-graph/src/task-graph/StreamPump.ts @@ -7,15 +7,21 @@ import type { ResourceScope, ServiceRegistry } from "@workglow/util"; import { getLogger } from "@workglow/util"; import type { TaskOutputRepository } from "../storage/TaskOutputRepository"; +import { BackpressureGate } from "../task/BackpressureGate"; import type { ITask } from "../task/ITask"; import type { StreamEvent, StreamMode } from "../task/StreamTypes"; import { + DEFAULT_BINARY_HIGH_WATER_BYTES, edgeNeedsAccumulation, getOutputStreamMode, getPortStreamMode, getStreamingPorts, + isDeltaStreamMode, + isTaskStreamable, + portForcesStreamValidation, + streamEventCost, } from "../task/StreamTypes"; -import type { TaskInput } from "../task/TaskTypes"; +import type { TaskIdType, TaskInput } from "../task/TaskTypes"; import { TaskStatus } from "../task/TaskTypes"; import { Dataflow, DATAFLOW_ALL_PORTS } from "./Dataflow"; import type { EdgeMaterializer } from "./EdgeMaterializer"; @@ -64,6 +70,18 @@ export interface StreamingRunOptions { readonly legacyCacheExplicitlyDisabled?: boolean; } +/** + * Per-port pacing state for a no-accumulation passthrough edge: the gate that + * parks the producer, plus a FIFO of the costs charged for enqueued events so + * the consumer-side credit reuses the charge instead of recomputing it. + * + * @internal + */ +interface EdgeGateState { + readonly gate: BackpressureGate; + readonly pendingCosts: number[]; +} + /** * @internal * Streaming bridge. Awaits upstream streaming inputs, runs streaming tasks, @@ -102,14 +120,25 @@ export class StreamPump { * Tees streaming inputs for a streamable task — one copy goes to the task's * executeStream() (via inputStreams), one stays on the edge for materialization * by awaitStreamInputs. + * + * A no-accumulation passthrough edge is never drained downstream, so its + * materialize copy would only pile every event up in the unread tee branch — + * the accumulation the passthrough exists to avoid. Such an edge hands its + * stream to the consumer directly (no tee); the edge keeps the same stream + * reference so "this edge is streaming" checks still hold, and its settled + * value arrives as the per-port {@link CacheRef} at producer finish. */ - prepareStreamingInputs(task: ITask): void { + prepareStreamingInputs(task: ITask, noAccumulation: boolean = false): void { const dataflows = this.graph.getSourceDataflows(task.id); const streamingEdges = dataflows.filter((df) => df.stream !== undefined); if (streamingEdges.length === 0) return; const inputStreams = new Map>(); for (const df of streamingEdges) { const stream = df.stream!; + if (StreamPump.isNoAccumulationPassthroughEdge(this.graph, df, noAccumulation)) { + inputStreams.set(df.targetTaskPortId, stream); + continue; + } const [forwardCopy, materializeCopy] = stream.tee(); inputStreams.set(df.targetTaskPortId, forwardCopy); df.setStream(materializeCopy); @@ -183,13 +212,53 @@ export class StreamPump { options.noAccumulation === true ); + // One gate per source port that feeds a no-accumulation passthrough edge. + // The edge stream charges the gate as events are enqueued and credits it + // as the consumer reads; the producer's StreamProcessor parks on the + // `edgeBackpressure` thunk after each delta, pacing it to the consumer. + const edgeGates = this.buildPassthroughEdgeGates(task, options); + const edgeBackpressure = edgeGates + ? async (port?: string): Promise => { + if (port !== undefined) { + await edgeGates.get(port)?.gate.awaitBelowMark(); + return; + } + await Promise.all(Array.from(edgeGates.values(), (s) => s.gate.awaitBelowMark())); + } + : undefined; + // Safety release: a passthrough consumer that reaches a terminal state + // without reading its stream to completion (throws mid-read, gets + // disabled, or simply ignores ctx.inputStreams) would otherwise leave the + // producer parked at the gate forever. + const gateCleanups: Array<() => void> = []; + if (edgeGates) { + for (const df of this.graph.getTargetDataflows(task.id)) { + const state = edgeGates.get(df.sourceTaskPortId); + if (!state) continue; + const target = this.graph.getTask(df.targetTaskId); + if (!target) continue; + const onTargetStatus = (status: TaskStatus) => { + if ( + status === TaskStatus.COMPLETED || + status === TaskStatus.FAILED || + status === TaskStatus.ABORTING || + status === TaskStatus.DISABLED + ) { + state.gate.close(); + } + }; + target.on("status", onTargetStatus); + gateCleanups.push(() => target.off("status", onTargetStatus)); + } + } + let streamingNotified = false; const onStatus = (status: TaskStatus) => { if (status === TaskStatus.STREAMING && !streamingNotified) { streamingNotified = true; this.runScheduler.pushStatusFromNodeToEdges(task, ctx, TaskStatus.STREAMING); - this.pushStreamToEdges(task, streamMode); + this.pushStreamToEdges(task, streamMode, edgeGates); this.processScheduler.onTaskStreaming(task.id); } }; @@ -241,6 +310,7 @@ export class StreamPump { runId: options.runId, noAccumulation: options.noAccumulation, streamHighWaterBytes: options.streamHighWaterBytes, + edgeBackpressure, // Sinks are installed regardless of downstream needs: when both an // accumulator and a router exist (downstream needs materialized + cache // can stream), StreamProcessor tees — accumulator drives the enriched @@ -260,6 +330,10 @@ export class StreamPump { task.off("stream_start", onStreamStart); task.off("stream_chunk", onStreamChunk); task.off("stream_end", onStreamEnd); + for (const cleanup of gateCleanups) cleanup(); + // Idempotent: normally already closed by the edge stream's end/terminate + // handlers; this covers runs that never flipped to STREAMING at all. + if (edgeGates) for (const state of edgeGates.values()) state.gate.close(); } } @@ -352,6 +426,11 @@ export class StreamPump { // that cannot affirmatively report streaming support as non-streaming. if (typeof outputCache?.supportsStreaming !== "function") return false; if (!outputCache.supportsStreaming()) return false; + // The sink is only built for cacheable tasks (getBinaryRefSinksByPolicy + // refuses otherwise). Skipping accumulation for a non-cacheable task would + // leave its binary deltas with neither an accumulator nor a sink — the + // output would silently drop to the finish payload ({}). + if (!task.cacheable) return false; const outSchema = task.outputSchema(); const streamingPorts = getStreamingPorts(outSchema); @@ -387,10 +466,7 @@ export class StreamPump { if (!outputCache.supportsStreamingPorts()) return false; const streamingPorts = getStreamingPorts(task.outputSchema()); if (streamingPorts.length === 0) return false; - const allDelta = streamingPorts.every( - (p) => p.mode === "append" || p.mode === "object" || p.mode === "binary" - ); - if (!allDelta) return false; + if (!streamingPorts.every((p) => isDeltaStreamMode(p.mode))) return false; return !StreamPump.anyConsumerNeedsMaterialized(graph, task); } @@ -415,10 +491,17 @@ export class StreamPump { const source = graph.getTask(df.sourceTaskId); const target = graph.getTask(df.targetTaskId); if (!source || !target) return false; + // The consumer must actually take its data from the live stream: only + // streamable tasks receive ctx.inputStreams (prepareStreamingInputs is + // gated on isTaskStreamable), so a non-streamable target with a matching + // input mode still needs the drain to materialize its value. Subgraph + // hosts (GraphAsTask etc.) also need the drain — their inner tasks read + // the settled input slot, which the passthrough leaves unmaterialized. + if (!isTaskStreamable(target) || target.hasChildren()) return false; const srcMode = getPortStreamMode(source.outputSchema(), df.sourceTaskPortId); - if (srcMode !== "append" && srcMode !== "object" && srcMode !== "binary") return false; + if (!isDeltaStreamMode(srcMode)) return false; if (getPortStreamMode(target.inputSchema(), df.targetTaskPortId) !== srcMode) return false; - if (StreamPump.portForcesStreamValidation(target.inputSchema(), df.targetTaskPortId)) { + if (portForcesStreamValidation(target.inputSchema(), df.targetTaskPortId)) { return false; } // Single consumer of this source port: a fan-out source port must keep the @@ -429,17 +512,6 @@ export class StreamPump { return fanout.length === 1; } - /** Reads the per-port `x-validate-stream` opt-in from an input schema. */ - private static portForcesStreamValidation( - schema: ReturnType, - port: string - ): boolean { - if (typeof schema === "boolean") return false; - const prop = (schema.properties as Record)?.[port]; - if (!prop || typeof prop === "boolean") return false; - return prop["x-validate-stream"] === true; - } - /** * Returns `true` when any outgoing dataflow edge from {@link task} has a * target task whose input port can't consume the source's stream mode @@ -467,49 +539,132 @@ export class StreamPump { } /** - * Returns `true` when any outgoing dataflow edge targets an input port that - * consumes the source port's binary stream mode directly (`x-stream: - * "binary"` on both ends). Used to decide whether a cache hit should replay - * cached bytes as `binary-delta` events: with no stream-capable consumer - * the replay would be wasted reads. `*` fan-out edges don't count — their - * consumers receive materialized values, not streams. + * Returns `true` when any outgoing edge targets an input port that consumes + * the source port's delta stream mode directly (same `append` / `object` / + * `binary` mode on both ends). Drives whether a cache hit replays the cached + * bytes as delta events (via the per-mode codec) for a stream-capable + * consumer; `*` fan-out edges don't count (their consumers receive + * materialized values). */ - static anyConsumerAcceptsBinaryStream(graph: TaskGraph, task: ITask): boolean { + static anyConsumerAcceptsStream(graph: TaskGraph, task: ITask): boolean { const outSchema = task.outputSchema(); return graph.getTargetDataflows(task.id).some((df) => { if (df.sourceTaskPortId === DATAFLOW_ALL_PORTS) return false; - if (getPortStreamMode(outSchema, df.sourceTaskPortId) !== "binary") return false; + const srcMode = getPortStreamMode(outSchema, df.sourceTaskPortId); + if (!isDeltaStreamMode(srcMode)) return false; const targetTask = graph.getTask(df.targetTaskId); if (!targetTask) return false; - return getPortStreamMode(targetTask.inputSchema(), df.targetTaskPortId) === "binary"; + return getPortStreamMode(targetTask.inputSchema(), df.targetTaskPortId) === srcMode; }); } /** - * All-mode analogue of {@link anyConsumerAcceptsBinaryStream}: `true` when any - * outgoing edge targets an input port that consumes the source port's delta - * stream mode directly (same `append` / `object` / `binary` mode on both - * ends). Drives whether a cache hit replays the cached bytes as delta events - * (via the per-mode codec) for a stream-capable consumer; `*` fan-out edges - * don't count (their consumers receive materialized values). + * Builds one {@link EdgeGateState} per source port whose (single) outgoing + * edge qualifies as a no-accumulation passthrough AND whose consumer can + * make read progress while this producer is parked. The gate's high-water + * mark is the run's `streamHighWaterBytes` (falling back to + * {@link DEFAULT_BINARY_HIGH_WATER_BYTES}). Returns `undefined` when the flag + * is off or no edge qualifies, so the entire pacing path stays dormant. + * + * Liveness guard: the producer may park on a gate long before it finishes, + * so the consumer must be able to reach its stream reads without waiting on + * anything that itself waits on this producer. Any OTHER edge into the + * consumer whose source is this producer or one of its descendants — a + * drained streaming edge, a mode-mismatched edge, a static-value edge — + * settles only after this producer finishes, so gating would deadlock the + * pair. Such consumers keep the ungated passthrough (correct, just unpaced). */ - static anyConsumerAcceptsStream(graph: TaskGraph, task: ITask): boolean { - const outSchema = task.outputSchema(); - return graph.getTargetDataflows(task.id).some((df) => { - if (df.sourceTaskPortId === DATAFLOW_ALL_PORTS) return false; - const srcMode = getPortStreamMode(outSchema, df.sourceTaskPortId); - if (srcMode !== "append" && srcMode !== "object" && srcMode !== "binary") return false; - const targetTask = graph.getTask(df.targetTaskId); - if (!targetTask) return false; - return getPortStreamMode(targetTask.inputSchema(), df.targetTaskPortId) === srcMode; + private buildPassthroughEdgeGates( + task: ITask, + options: StreamingRunOptions + ): Map | undefined { + if (options.noAccumulation !== true) return undefined; + const highWaterMark = + options.streamHighWaterBytes !== undefined && options.streamHighWaterBytes > 0 + ? options.streamHighWaterBytes + : DEFAULT_BINARY_HIGH_WATER_BYTES; + let reachable: ReadonlySet | undefined; + let gates: Map | undefined; + for (const df of this.graph.getTargetDataflows(task.id)) { + if (!StreamPump.isNoAccumulationPassthroughEdge(this.graph, df, true)) continue; + reachable ??= this.tasksReachableFrom(task.id); + const consumerWaitsOnProducer = this.graph + .getSourceDataflows(df.targetTaskId) + .some((e) => e !== df && reachable!.has(e.sourceTaskId)); + if (consumerWaitsOnProducer) continue; + // The passthrough predicate guarantees a single consumer per source + // port, so one gate per port is exactly one gate per edge. + gates ??= new Map(); + gates.set(df.sourceTaskPortId, { + gate: new BackpressureGate(highWaterMark), + pendingCosts: [], + }); + } + return gates; + } + + /** Task ids reachable from `taskId` via outgoing dataflows, including itself. */ + private tasksReachableFrom(taskId: TaskIdType): ReadonlySet { + const seen = new Set([taskId]); + const queue: TaskIdType[] = [taskId]; + while (queue.length > 0) { + const id = queue.pop()!; + for (const df of this.graph.getTargetDataflows(id)) { + if (!seen.has(df.targetTaskId)) { + seen.add(df.targetTaskId); + queue.push(df.targetTaskId); + } + } + } + return seen; + } + + /** + * Wraps a per-port edge stream so every event read by the consumer credits + * the port's gate with the cost charged when that event was enqueued (a + * FIFO — events cross the gate in order, so each read pops the oldest + * charge), waking a producer parked at the high-water mark. Close (end, + * cancel, or an upstream read failure) also closes the gate so an abandoned + * consumer can never orphan a parked producer. + */ + private static wrapStreamWithGateCredit( + stream: ReadableStream, + state: EdgeGateState + ): ReadableStream { + const reader = stream.getReader(); + return new ReadableStream({ + async pull(controller) { + let done: boolean; + let value: StreamEvent | undefined; + try { + ({ done, value } = await reader.read()); + } catch (err) { + state.gate.close(); + throw err; + } + if (done || value === undefined) { + state.gate.close(); + controller.close(); + return; + } + state.gate.credit(state.pendingCosts.shift() ?? streamEventCost(value)); + controller.enqueue(value); + }, + cancel(reason) { + state.gate.close(); + return reader.cancel(reason); + }, }); } /** - * Returns true if an event carries a port-specific delta (text-delta or object-delta). + * Returns true if an event carries a port-specific delta (text-delta, + * object-delta, or binary-delta). */ private static isPortDelta(event: StreamEvent): event is StreamEvent & { port: string } { - return event.type === "text-delta" || event.type === "object-delta"; + return ( + event.type === "text-delta" || event.type === "object-delta" || event.type === "binary-delta" + ); } /** @@ -520,11 +675,18 @@ export class StreamPump { * * Also taps snapshot events to write per-port data into each edge's * `latestSnapshot` slot for downstream peek-during-streaming. + * + * When a `gate` is supplied (no-accumulation passthrough), every enqueued + * event charges the gate with its {@link streamEventCost}; the consumer-side + * wrapper ({@link wrapStreamWithGateCredit}) credits it back on each read. + * Stream end and producer abort/error close the gate so a parked producer + * is always released. */ private createStreamFromTaskEvents( task: ITask, portId: string | undefined, - edgesForGroup: ReadonlyArray + edgesForGroup: ReadonlyArray, + gate?: EdgeGateState ): ReadableStream { // Shared teardown closure — hoisted out of start() so cancel() (which the // ReadableStream invokes on reader.cancel()) can call it too. Without this, @@ -542,6 +704,10 @@ export class StreamPump { cleanup = () => { if (closed) return; closed = true; + // Release any producer parked on this port's passthrough gate — + // every teardown path (end, terminal status, reader cancel) must + // wake it or the run hangs. + gate?.gate.close(); try { controller.close(); } catch { @@ -576,6 +742,11 @@ export class StreamPump { } } } + if (gate) { + const cost = streamEventCost(event); + gate.gate.account(cost); + gate.pendingCosts.push(cost); + } controller.enqueue(event); } catch { // Stream may be closed @@ -587,7 +758,23 @@ export class StreamPump { const onStatus = (status: TaskStatus) => { // Terminal statuses with no stream_end (error/abort -> FAILED, or a // completion that bypassed stream_end) must still release the stream. - if (status === TaskStatus.FAILED || status === TaskStatus.COMPLETED) { + // A producer FAILURE first surfaces in-stream so a drained edge + // materializes the error instead of quietly settling on whatever + // partial data had arrived — a consumer already dispatched + // (unblocked at STREAMING) must not complete, and cache, an output + // derived from truncated input. Abort closes gracefully (the + // run-level abort cascade is already tearing everything down). + if (status === TaskStatus.FAILED) { + try { + controller.enqueue({ + type: "error", + error: task.error ?? new Error(`Task ${task.type} failed during streaming`), + } as StreamEvent); + } catch { + // Stream may already be closed + } + cleanup(); + } else if (status === TaskStatus.COMPLETED || status === TaskStatus.ABORTING) { cleanup(); } }; @@ -613,8 +800,18 @@ export class StreamPump { * Creates per-port filtered ReadableStreams for specific-port edges and * unfiltered streams for DATAFLOW_ALL_PORTS edges. Within each port group, * uses tee() for fan-out to multiple consumers. + * + * A port with an entry in `edgeGates` (single passthrough consumer by + * construction) gets a gate-instrumented stream: events charge the gate as + * they are enqueued and credit it as the consumer reads, so the producer can + * park against the consumer's read rate. Fan-out groups never have a gate — + * multi-consumer pacing stays best-effort. */ - private pushStreamToEdges(task: ITask, _streamMode: StreamMode): void { + private pushStreamToEdges( + task: ITask, + _streamMode: StreamMode, + edgeGates?: ReadonlyMap + ): void { const targetDataflows = this.graph.getTargetDataflows(task.id); if (targetDataflows.length === 0) return; @@ -632,10 +829,15 @@ export class StreamPump { for (const [portKey, edges] of groups) { const filterPort = portKey === DATAFLOW_ALL_PORTS ? undefined : portKey; - const stream = this.createStreamFromTaskEvents(task, filterPort, edges); + // Gates exist only for single-consumer passthrough ports; a gate on a + // multi-edge group cannot happen by construction, but guard anyway so a + // tee'd fan-out is never double-credited. + const gate = + filterPort !== undefined && edges.length === 1 ? edgeGates?.get(filterPort) : undefined; + const stream = this.createStreamFromTaskEvents(task, filterPort, edges, gate); if (edges.length === 1) { - edges[0].setStream(stream); + edges[0].setStream(gate ? StreamPump.wrapStreamWithGateCredit(stream, gate) : stream); } else { let currentStream = stream; for (let i = 0; i < edges.length; i++) { diff --git a/packages/task-graph/src/task-graph/TaskGraphRunner.ts b/packages/task-graph/src/task-graph/TaskGraphRunner.ts index 853e03d4d..c3f2d071d 100644 --- a/packages/task-graph/src/task-graph/TaskGraphRunner.ts +++ b/packages/task-graph/src/task-graph/TaskGraphRunner.ts @@ -495,7 +495,7 @@ export class TaskGraphRunner { // the task's executeStream() (via inputStreams) while the other stays // on the edge for materialization by awaitStreamInputs. if (isStreamable) { - this.streamPump.prepareStreamingInputs(task); + this.streamPump.prepareStreamingInputs(task, this.noAccumulation); } // Await any active streams on input dataflow edges so their values diff --git a/packages/task-graph/src/task/CacheCoordinator.ts b/packages/task-graph/src/task/CacheCoordinator.ts index deb502b4d..38e8eb492 100644 --- a/packages/task-graph/src/task/CacheCoordinator.ts +++ b/packages/task-graph/src/task/CacheCoordinator.ts @@ -17,8 +17,13 @@ import type { TaskOutputRepository } from "../storage/TaskOutputRepository"; import type { ITask } from "./ITask"; import type { BinaryRefSink, StreamSink } from "./StreamProcessor"; import type { StreamEvent } from "./StreamTypes"; -import { assertBinaryFormat, getStreamingPorts, materializeBinary } from "./StreamTypes"; -import { Task } from "./Task"; +import { + assertBinaryFormat, + foldObjectDelta, + getStreamingPorts, + isDeltaStreamMode, + materializeBinary, +} from "./StreamTypes"; import type { TaskRunContext } from "./TaskRunContext"; import type { TaskInput, TaskOutput } from "./TaskTypes"; import { TaskStatus } from "./TaskTypes"; @@ -36,6 +41,13 @@ interface SchemaProperties { export interface CacheReplayContext { readonly hasMaterializingConsumers: boolean; readonly hasStreamingConsumers: boolean; + /** + * Graph-installed producer park for gated passthrough edges (see + * {@link IRunConfig.edgeBackpressure}). When present, cache-hit replay + * awaits it after each emitted delta so a slow consumer paces the replay + * exactly as it paces a fresh run; absent, replay is read-speed. + */ + readonly edgeBackpressure?: (port?: string) => Promise; } /** @@ -72,7 +84,10 @@ export class CacheCoordinator { if (!outputCache) return inputs; - const inputSchema = (this.task.constructor as typeof Task).inputSchema(); + // Instance schema, not the static one: dynamic-schema tasks add ports at + // runtime, and the key must normalize the same ports the save/lookup + // paths serialize. + const inputSchema = this.task.inputSchema(); const normalized = await CacheCoordinator.normalizeInputsForCacheKey( inputs as Record, inputSchema as unknown as SchemaProperties @@ -121,7 +136,11 @@ export class CacheCoordinator; const refPorts = getStreamingPorts(outputSchema).filter( - (p) => - (p.mode === "append" || p.mode === "object" || p.mode === "binary") && - isCacheRef(source[p.port]) + (p) => isDeltaStreamMode(p.mode) && isCacheRef(source[p.port]) ); if (refPorts.length === 0) return "none"; // Resolve every ref before emitting any event so a dangling ref becomes a // clean miss with zero observable side effects. + const resolved = await Promise.all( + refPorts.map(async ({ port }) => ({ + port, + stream: await streamRefViaBacking(source[port] as CacheRef, outputCache), + })) + ); const streams = new Map>(); - for (const { port } of refPorts) { - const stream = await streamRefViaBacking(source[port] as CacheRef, outputCache); - if (stream === undefined) return "miss"; + for (const { port, stream } of resolved) { + if (stream === undefined) { + // Release any streams already opened for other ports before reporting + // the miss — a backing may hold a file handle per resolved stream. + for (const s of streams.values()) { + void s[Symbol.asyncIterator]().return?.(undefined); + } + return "miss"; + } streams.set(port, stream); } @@ -219,12 +248,17 @@ export class CacheCoordinator = { ...source }; for (const { port, mode } of refPorts) { const stream = streams.get(port)!; if (mode === "binary") { - // Stream chunk-by-chunk so large binary stays memory-bounded when no - // materializing consumer needs the whole artifact. + // Stream chunk-by-chunk; whole-artifact buffering happens only when a + // materializing consumer needs the settled value. const chunks: Uint8Array[] | undefined = needBytes ? [] : undefined; for await (const chunk of stream) { chunks?.push(chunk); @@ -234,27 +268,31 @@ export class CacheCoordinator { - for (const c of buf) yield c; - }; - if (replayDeltas) { - for await (const ev of codec.decode(bytesOf(), port)) { + let text = ""; + let folded: Record | unknown[] | undefined; + for await (const ev of codec.decode(stream, port)) { + if (replayDeltas) { this.task.emit("stream_chunk", ev as StreamEvent); + if (pace) await pace(port); + } + if (needBytes) { + if (ev.type === "text-delta") text += ev.textDelta; + else if (ev.type === "object-delta") folded = foldObjectDelta(folded, ev.objectDelta); } } if (needBytes) { - finishData[port] = await codec.materialize(bytesOf(), port); + finishData[port] = mode === "append" ? text : folded; } } } @@ -274,7 +312,7 @@ export class CacheCoordinator { if (!outputCache || !this.task.cacheable || output === undefined) return; - const outputSchema = (this.task.constructor as typeof Task).outputSchema(); + const outputSchema = this.task.outputSchema(); const wireOutputs = await CacheCoordinator.serializeOutputPorts( output as Record, outputSchema as unknown as SchemaProperties @@ -286,23 +324,6 @@ export class CacheCoordinator, - metadata: Record, - outputCache: TaskOutputRepository | undefined - ): Promise { - if (!outputCache || !this.task.cacheable) return undefined; - if (!outputCache.supportsStreaming()) return undefined; - return outputCache.saveOutputStream!(this.task.type, keyInputs, chunks, metadata); - } - // ======================================================================== // Policy-aware routing methods // ======================================================================== @@ -358,14 +379,15 @@ export class CacheCoordinator; - const binaryPorts = getStreamingPorts(outputSchema) - .filter((p) => p.mode === "binary") + const refPorts = getStreamingPorts(outputSchema) + .filter((p) => isDeltaStreamMode(p.mode)) .map((p) => p.port); await Promise.all( - binaryPorts.map(async (port) => { + refPorts.map(async (port) => { const value = source[port]; if (!isCacheRef(value)) return; try { diff --git a/packages/task-graph/src/task/ITask.ts b/packages/task-graph/src/task/ITask.ts index bdaca95a3..04a235bb1 100644 --- a/packages/task-graph/src/task/ITask.ts +++ b/packages/task-graph/src/task/ITask.ts @@ -69,15 +69,17 @@ export interface IExecuteContext { */ resourceScope?: ResourceScope; /** - * Optional cooperative backpressure hook for streaming tasks that emit very - * large binary outputs by direct event emission (rather than through the - * StreamProcessor's `await router.push(...)` path). Tasks may `await` this - * between yields/emits to give downstream sinks a chance to drain. + * Optional cooperative backpressure hook for streaming tasks that emit + * large outputs by direct event emission (rather than through the + * StreamProcessor's awaited per-event path). Tasks may `await` this between + * yields/emits to give downstream sinks and consumers a chance to drain: it + * resolves once every active cache-sink router AND any consumer-edge gate + * is back below its high-water mark. * * Defaults to a no-op when the runtime does not install a real backpressure * source — tasks can call it unconditionally without paying a cost. */ - binaryBackpressure?: () => Promise; + backpressure?: () => Promise; } export type IExecutePreviewContext = Pick; @@ -189,6 +191,18 @@ export interface IRunConfig { */ streamHighWaterBytes?: number; + /** + * Graph-installed producer park for the no-accumulation passthrough path. + * The graph runner owns the consumer-edge gates (it is the only layer that + * knows the edges); this thunk closes over them so the task-level streaming + * runtime can pace the producer without any edge knowledge. Called with a + * port name it awaits that port's gate (used after each delta); called with + * no argument it awaits every gate (the cooperative + * {@link IExecuteContext.backpressure} hook). Absent on standalone runs and + * whenever no outgoing edge qualifies for the passthrough. + */ + edgeBackpressure?: (port?: string) => Promise; + /** * Optional callback invoked whenever a task's progress changes during execution. * @param task - The task whose progress changed. diff --git a/packages/task-graph/src/task/StreamProcessor.ts b/packages/task-graph/src/task/StreamProcessor.ts index 5e81da217..94563bc37 100644 --- a/packages/task-graph/src/task/StreamProcessor.ts +++ b/packages/task-graph/src/task/StreamProcessor.ts @@ -101,6 +101,16 @@ export interface StreamProcessorDeps { * {@link DEFAULT_BINARY_HIGH_WATER_BYTES} when omitted. */ readonly binaryHighWaterBytes?: number; + /** + * Consumer-edge backpressure for the no-accumulation passthrough path, + * threaded down from the graph runner (see `IRunConfig.edgeBackpressure`). + * After emitting a delta the processor awaits this with the event's port so + * the producer is paced to that port's consumer read rate; the cooperative + * `IExecuteContext.backpressure` hook awaits it with no argument (all + * ports). Absent on standalone runs — the processor then paces only against + * its own cache-sink routers. + */ + readonly edgeBackpressure?: (port?: string) => Promise; } /** @@ -203,17 +213,16 @@ export class StreamProcessor this.task.emit("stream_start"); // Cooperative backpressure hook for executeStream() implementations that - // emit through a side channel (not StreamProcessor's awaited `push`). When - // any port has a router (we'd be applying byte-bounded backpressure on the - // direct `binary-delta` path anyway), `await ctx.binaryBackpressure()` - // waits until ALL active routers are at-or-below their high-water mark. - // Without a router this is a cheap no-op. - const binaryBackpressure = async (): Promise => { - if (routers.size === 0) return; + // emit through a side channel (not StreamProcessor's awaited per-event + // path). `await ctx.backpressure()` waits until ALL active cache-sink + // routers AND any consumer-edge gate are back below their high-water + // marks. With no router and no edge gate this is a cheap no-op. + const backpressure = async (): Promise => { const waits: Promise[] = []; for (const { router } of routers.values()) { if (router._bufferedBytes >= router._highWaterMarkBytes) waits.push(router._awaitDrain()); } + if (deps.edgeBackpressure) waits.push(deps.edgeBackpressure()); if (waits.length === 0) return; await Promise.all(waits); }; @@ -225,9 +234,10 @@ export class StreamProcessor registry: deps.registry, resourceScope: deps.resourceScope, inputStreams: deps.inputStreams, - binaryBackpressure, + backpressure, }); + let sawFinish = false; try { for await (const event of stream) { // For snapshot events, update runOutputData BEFORE emitting stream_chunk @@ -258,6 +268,10 @@ export class StreamProcessor // accumulator (if any) still drives the enriched finish event. await routeDelta(event.port, event); this.task.emit("stream_chunk", event as StreamEvent); + // Pace the producer to this port's consumer read rate on a + // passthrough edge (no-op elsewhere): the emit above charged the + // edge gate; park here until the consumer drains below the mark. + if (deps.edgeBackpressure) await deps.edgeBackpressure(event.port); break; } case "object-delta": { @@ -280,6 +294,7 @@ export class StreamProcessor // Tee to the port's sink (encoded as NDJSON) when one exists. await routeDelta(event.port, event); this.task.emit("stream_chunk", event as StreamEvent); + if (deps.edgeBackpressure) await deps.edgeBackpressure(event.port); break; } case "binary-delta": { @@ -305,6 +320,7 @@ export class StreamProcessor accumulatedBinary.set(event.port, arr); } this.task.emit("stream_chunk", event as StreamEvent); + if (deps.edgeBackpressure) await deps.edgeBackpressure(event.port); break; } case "snapshot": { @@ -317,6 +333,7 @@ export class StreamProcessor break; } case "finish": { + sawFinish = true; const hasEnrichment = accumulated !== undefined || accumulatedObjects !== undefined || @@ -438,10 +455,20 @@ export class StreamProcessor for (const { router } of routers.values()) router.fail(failure); throw err; } finally { - // Defensive: if the loop exited without seeing a `finish` event - // (e.g. abort, generator return without yield), close routers so their - // sinks see end-of-stream rather than blocking on the next chunk. - for (const { router } of routers.values()) router.end(); + // If the loop exited without a `finish` event (abort via cooperative + // generator return, or a generator ending early), the routed bytes are + // incomplete: FAIL the routers so their sinks reject and discard the + // partial write, instead of committing a truncated blob to the cache as + // a finished artifact. After a normal finish, `end()` is an idempotent + // no-op (the finish handler already ended the routers). + if (sawFinish) { + for (const { router } of routers.values()) router.end(); + } else { + const incomplete = new TaskError( + `Task ${this.task.type} stream ended without a finish event; discarding partial output.` + ); + for (const { router } of routers.values()) router.fail(incomplete); + } } // Check if the task was aborted during streaming @@ -535,7 +562,7 @@ export class BinaryStreamRouter { } /** - * @internal Used by {@link IExecuteContext.binaryBackpressure} so a task + * @internal Used by {@link IExecuteContext.backpressure} so a task * emitting via a side channel can park until the consumer drains. Resolves * immediately when the buffer is already under the mark or the router has * been closed. diff --git a/packages/task-graph/src/task/StreamTypes.ts b/packages/task-graph/src/task/StreamTypes.ts index 1e707dc39..14de02e30 100644 --- a/packages/task-graph/src/task/StreamTypes.ts +++ b/packages/task-graph/src/task/StreamTypes.ts @@ -174,6 +174,34 @@ export function getStreamingPorts( return result; } +/** + * Delta stream modes: the modes whose events are incremental per-port deltas + * that can be encoded to (and replayed from) a single-port byte stream via + * {@link getStreamPortCodec}. `replace` (snapshot-driven), `none`, and `mixed` + * are not delta modes. + */ +export function isDeltaStreamMode( + mode: StreamMode +): mode is Extract { + return mode === "append" || mode === "object" || mode === "binary"; +} + +/** + * Reads the per-port `x-validate-stream` opt-in from an input schema: a port + * that sets it wants its stream materialized and validated as a whole value, + * opting out of both the validation exemption for stream-wired ports and the + * no-accumulation passthrough for its edge. + */ +export function portForcesStreamValidation( + schema: DataPortSchema | JsonSchema, + port: string +): boolean { + if (typeof schema === "boolean") return false; + const prop = (schema.properties as Record)?.[port]; + if (!prop || typeof prop === "boolean") return false; + return prop["x-validate-stream"] === true; +} + /** * Returns the dominant output stream mode for a task by inspecting its output schema. * Returns `"mixed"` when ports use different modes (e.g., append + object). @@ -271,25 +299,6 @@ export function getObjectPortId(schema: DataPortSchema): string | undefined { return undefined; } -/** - * Returns the port ID (property name) of the first output port that declares - * `x-stream: "binary"`, or `undefined` if no such port exists. - * - * @param schema - The task's output DataPortSchema - * @returns The port name with binary streaming, or undefined - */ -export function getBinaryPortId(schema: DataPortSchema): string | undefined { - if (typeof schema === "boolean") return undefined; - const props = schema.properties; - if (!props) return undefined; - - for (const [name, prop] of Object.entries(props)) { - if (!prop || typeof prop === "boolean") continue; - if ((prop as any)["x-stream"] === "binary") return name; - } - return undefined; -} - /** * Canonical vocabulary for the `format` annotation on a binary streaming output * port. `"blob"` materializes chunks into a `Blob` (the default); `"binary"` @@ -311,6 +320,28 @@ export type BinaryFormat = "blob" | "binary"; */ export const DEFAULT_BINARY_HIGH_WATER_BYTES = 8 * 1024 * 1024; +const streamCostEncoder = new TextEncoder(); + +/** + * Buffered cost of a single stream event, in bytes, for backpressure + * accounting. Delta events cost their payload size (UTF-8 bytes for + * `text-delta`, JSON-encoded length for `object-delta`, raw byte length for + * `binary-delta`); control events (`finish`, `snapshot`, `phase`, `error`) + * cost nothing — they are not what a slow consumer buffers up on. + */ +export function streamEventCost(event: StreamEvent): number { + switch (event.type) { + case "text-delta": + return streamCostEncoder.encode(event.textDelta).byteLength; + case "object-delta": + return JSON.stringify(event.objectDelta).length; + case "binary-delta": + return event.binaryDelta.byteLength; + default: + return 0; + } +} + /** * Reads the `format` annotation of a single output port from the task's output * schema. Returns the raw string (or `undefined`) — callers needing the diff --git a/packages/task-graph/src/task/TaskRegistry.ts b/packages/task-graph/src/task/TaskRegistry.ts index 3a68f55ea..2941f5b84 100644 --- a/packages/task-graph/src/task/TaskRegistry.ts +++ b/packages/task-graph/src/task/TaskRegistry.ts @@ -42,22 +42,26 @@ function registerTask(baseClass: AnyTaskConstructor): void { ); } - // Validate every binary streaming output port's `format` against the - // canonical {@link BinaryFormat} vocabulary BEFORE adding to the registry. - // A typo like `format: "Blob"` would otherwise silently coerce to the - // ArrayBuffer branch in `materializeBinary`, producing the wrong runtime - // type for every consumer of that port. Fail at registration so the + // Validate every binary streaming port's `format` against the canonical + // {@link BinaryFormat} vocabulary BEFORE adding to the registry — output + // ports (materializeBinary picks the runtime type) AND input ports (input + // hydration picks Blob vs ArrayBuffer the same way). A typo like + // `format: "Blob"` would otherwise silently coerce to the wrong branch, + // producing the wrong runtime type. Fail at registration so the // misconfiguration surfaces near the task definition site. const outputSchema = baseClass.outputSchema(); - for (const { port, mode } of getStreamingPorts(outputSchema)) { - if (mode !== "binary") continue; - try { - assertBinaryFormat(outputSchema, port); - } catch (err) { - const message = err instanceof Error ? err.message : String(err); - throw new Error( - `Cannot register task "${baseClass.type}": invalid binary stream port. ${message}` - ); + const inputSchema = baseClass.inputSchema(); + for (const schema of [outputSchema, inputSchema]) { + for (const { port, mode } of getStreamingPorts(schema)) { + if (mode !== "binary") continue; + try { + assertBinaryFormat(schema, port); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + throw new Error( + `Cannot register task "${baseClass.type}": invalid binary stream port. ${message}` + ); + } } } @@ -65,7 +69,7 @@ function registerTask(baseClass: AnyTaskConstructor): void { // Validate schemas at registration time (soft — warn only, don't throw) const schemas = [ - { name: "inputSchema", schema: baseClass.inputSchema() }, + { name: "inputSchema", schema: inputSchema }, { name: "outputSchema", schema: outputSchema }, ] as const; diff --git a/packages/task-graph/src/task/TaskRunner.ts b/packages/task-graph/src/task/TaskRunner.ts index 4cbd8598f..d0a9b5679 100644 --- a/packages/task-graph/src/task/TaskRunner.ts +++ b/packages/task-graph/src/task/TaskRunner.ts @@ -12,11 +12,12 @@ import { ServiceRegistry, SpanStatusCode, } from "@workglow/util"; -import type { DataPortSchema } from "@workglow/util/schema"; import { isCacheRef, resolveReferenceThreshold } from "../cache/CacheRef"; import type { CacheRegistry } from "../cache/CacheRegistry"; import { CACHE_REGISTRY, DefaultCacheRegistry } from "../cache/CacheRegistry"; +import { streamRefViaBacking } from "../cache/resolveRef"; import { RunPrivateCacheRepo } from "../cache/RunPrivateCacheRepo"; +import { getStreamPortCodec } from "../cache/streamCodec"; import { TASK_OUTPUT_REPOSITORY, TaskOutputRepository } from "../storage/TaskOutputRepository"; import type { Taskish } from "../task-graph/Conversions"; import { ensureTask } from "../task-graph/Conversions"; @@ -30,7 +31,9 @@ import { getBinaryPortFormat, getOutputStreamMode, getPortStreamMode, + isDeltaStreamMode, isTaskStreamable, + portForcesStreamValidation, } from "./StreamTypes"; import { Task } from "./Task"; import { @@ -196,10 +199,8 @@ export class TaskRunner< const inputs: Input = await this.hydrateInputRefs(this.task.runInputData as Input); this.task.runInputData = inputs; - const isValid = await this.task.validateInput( - inputs, - this.streamWiredValidationSkips(inputs) - ); + const streamWiredSkips = this.streamWiredValidationSkips(inputs); + const isValid = await this.task.validateInput(inputs, streamWiredSkips); if (!isValid) { throw new TaskInvalidInputError("Invalid input data"); } @@ -224,6 +225,16 @@ export class TaskRunner< let policy = this.task.getCachePolicy(inputs); + // A port fed by a live event stream has no settled value when the + // cache key is computed — the streamed content cannot contribute to + // the key, so two runs differing only in stream payload would collide + // on one entry (stale hits, poisoned rows). Disable caching for any + // run consuming a live stream at an unsettled port; a drained edge + // settles the value before this point and keeps caching as usual. + if (streamWiredSkips !== undefined && streamWiredSkips.size > 0) { + policy = { kind: "none" }; + } + // Standalone TaskRunner cannot namespace private cache writes without a // runId — TaskGraphRunner owns the wrap. If a standalone caller routes // to the private slot with no runId, downgrade to `kind: "none"` so the @@ -270,6 +281,7 @@ export class TaskRunner< { hasMaterializingConsumers: config.hasMaterializingConsumers === true, hasStreamingConsumers: config.hasStreamingConsumers === true, + edgeBackpressure: config.edgeBackpressure, } ); @@ -315,6 +327,7 @@ export class TaskRunner< binaryRefSinks, refSinks, binaryHighWaterBytes, + edgeBackpressure: config.edgeBackpressure, }) : await this.executeTask(inputs, ctx); @@ -336,7 +349,7 @@ export class TaskRunner< // unreferenced. Best-effort delete it so the cache directory // does not accumulate orphans on every save failure. if ((refSinks ?? binaryRefSinks) !== undefined && outputs !== undefined) { - await this.cacheCoordinator.cleanupOrphanBlobsForBinaryPorts( + await this.cacheCoordinator.cleanupOrphanBlobsForStreamPorts( outputs as Output, this.cacheRegistry, policy, @@ -436,7 +449,7 @@ export class TaskRunner< let skip: Set | undefined; for (const port of this.inputStreams.keys()) { if (getPortStreamMode(schema, port) === "none") continue; - if (TaskRunner.portForcesStreamValidation(schema, port)) continue; + if (portForcesStreamValidation(schema, port)) continue; const value = source[port]; if (value !== undefined && !isCacheRef(value)) continue; (skip ??= new Set()).add(port); @@ -444,24 +457,17 @@ export class TaskRunner< return skip; } - /** Reads the per-port `x-validate-stream` opt-in from an input schema. */ - private static portForcesStreamValidation(schema: DataPortSchema, port: string): boolean { - if (typeof schema === "boolean") return false; - const prop = (schema.properties as Record)?.[port]; - if (!prop || typeof prop === "boolean") return false; - return prop["x-validate-stream"] === true; - } - /** * Hydrate branded {@link CacheRef} values in resolved inputs to inline - * bytes before `execute()` runs, resolving against the run's cache registry - * (private repo first, then deterministic). Materialization type follows the - * input port's `format` annotation (`"binary"` → `ArrayBuffer`, anything - * else → `Blob`). + * values before `execute()` runs, resolving against the run's cache registry + * (private repo first, then deterministic). Materialization is mode-aware: + * an `append` / `object` ref decodes through its stream codec back to the + * string / folded object the port expects; anything else follows the input + * port's `format` annotation (`"binary"` → `ArrayBuffer`, else → `Blob`). * - * Binary-streaming input ports with a live input stream are skipped: those - * consumers take bytes from the stream and the ref at the port remains the - * durable pointer — hydrating it would re-materialize what the stream + * Stream-wired input ports with a live input stream are skipped: those + * consumers take their data from the stream and the ref at the port remains + * the durable pointer — hydrating it would re-materialize what the stream * already delivers. * * Hydration runs before cache-key computation so a ref-bearing input @@ -481,31 +487,55 @@ export class TaskRunner< const schema = this.task.inputSchema(); const source = inputs as Record; + const hydrations = await Promise.all( + Object.entries(source).map(async ([port, value]) => { + if (!isCacheRef(value)) return undefined; + // A stream-wired input port (any mode) with a live input stream keeps + // its ref as the durable pointer — the consumer takes its data from + // the stream, so hydrating the ref would re-materialize what the + // stream already delivers. + if (getPortStreamMode(schema, port) !== "none" && this.inputStreams?.has(port)) { + return undefined; + } + // append / object refs persist codec-encoded delta bytes; decode them + // back to the settled value instead of handing a byte Blob to a + // string/object port. + if (value.mode !== undefined && value.mode !== "binary" && isDeltaStreamMode(value.mode)) { + for (const repo of repos) { + const stream = await streamRefViaBacking(value, repo); + if (stream === undefined) continue; + const inlined = await getStreamPortCodec(value.mode).materialize(stream, port); + return { port, inlined }; + } + throw this.unresolvableInputRefError(port); + } + let blob: Blob | undefined; + for (const repo of repos) { + blob = await repo.getOutputByRef!(value); + if (blob !== undefined) break; + } + if (blob === undefined) throw this.unresolvableInputRefError(port); + const inlined = + getBinaryPortFormat(schema, port) === "binary" ? await blob.arrayBuffer() : blob; + return { port, inlined }; + }) + ); let out: Record | undefined; - for (const [port, value] of Object.entries(source)) { - if (!isCacheRef(value)) continue; - // A stream-wired input port (any mode) with a live input stream keeps its - // ref as the durable pointer — the consumer takes its data from the - // stream, so hydrating the ref would re-materialize what the stream - // already delivers. - if (getPortStreamMode(schema, port) !== "none" && this.inputStreams?.has(port)) continue; - let blob: Blob | undefined; - for (const repo of repos) { - blob = await repo.getOutputByRef!(value); - if (blob !== undefined) break; - } - if (blob === undefined) { - throw new TaskFailedError( - `Task "${this.task.type}" input port "${port}" holds a cache ref that no configured ` + - `cache backing can resolve (entry evicted?).` - ); - } + for (const h of hydrations) { + if (!h) continue; out ??= { ...source }; - out[port] = getBinaryPortFormat(schema, port) === "binary" ? await blob.arrayBuffer() : blob; + out[h.port] = h.inlined; } return (out ?? source) as Input; } + private unresolvableInputRefError(port: string): TaskFailedError { + return new TaskFailedError( + `Task "${this.task.type}" input port "${port}" holds a cache ref that no configured ` + + `cache backing can resolve (entry evicted?).` + ); + } + public async runPreview(overrides: Partial = {}): Promise { if (this.task.status === TaskStatus.PROCESSING) { return this.task.runOutputData as Output; diff --git a/packages/test/src/test/task-graph/Spec2QueueRowAndRehydrate.test.ts b/packages/test/src/test/task-graph/QueueRowAndRehydrate.test.ts similarity index 96% rename from packages/test/src/test/task-graph/Spec2QueueRowAndRehydrate.test.ts rename to packages/test/src/test/task-graph/QueueRowAndRehydrate.test.ts index febcd460c..0b70920af 100644 --- a/packages/test/src/test/task-graph/Spec2QueueRowAndRehydrate.test.ts +++ b/packages/test/src/test/task-graph/QueueRowAndRehydrate.test.ts @@ -29,8 +29,8 @@ import { beforeAll, beforeEach, describe, expect, it } from "vitest"; type BinOut = { bytes: Blob }; /** - * Streaming memory cache that exposes both `saveOutputStream` (Spec 2 path, - * returns CacheRef + stores bytes in a side map) and `getOutputByRef` so the + * Streaming memory cache that exposes both `saveOutputStream` (the ref sink + * path: returns CacheRef + stores bytes in a side map) and `getOutputByRef` so the * cross-process resolution test below can hydrate refs without touching the * main `saveOutput` row. */ @@ -173,13 +173,13 @@ beforeEach(() => { }); /** - * The principal user value of Spec 2: the SAVED ROW in the cache (the same - * value job-queue would carry through `JobStorageFormat.output`) stays small + * The principal user value of result-as-reference: the SAVED ROW in the cache + * (the same value job-queue would carry through `JobStorageFormat.output`) stays small * regardless of payload size when the ref path is taken. These tests measure * the wire size by JSON-serializing the saved output the way a real storage * backend (Postgres/SQLite) would. */ -describe("Spec 2 — saved-row size & cross-process rehydration", () => { +describe("saved-row size & cross-process rehydration", () => { it("force-ref keeps the saved row tiny (CacheRef envelope only); bytes live in the streaming cache", async () => { const task = new BigBlobStreamTask(); const output = await task.run({}, { registry: services, referenceThresholdBytes: 0 }); @@ -299,7 +299,7 @@ describe("Spec 2 — saved-row size & cross-process rehydration", () => { waitFor: async () => savedOutput as unknown as { bytes: Blob }, }; const resolved = await resolveJobOutput(handle, repo); - // Per Spec 2 §2: best-effort, returns undefined on cache miss. + // Resolution is best-effort: a dangling ref resolves to undefined. expect(resolved.bytes).toBeUndefined(); }); }); diff --git a/packages/test/src/test/task-graph/StreamBackpressureEngaged.test.ts b/packages/test/src/test/task-graph/StreamBackpressureEngaged.test.ts new file mode 100644 index 000000000..8a249a64b --- /dev/null +++ b/packages/test/src/test/task-graph/StreamBackpressureEngaged.test.ts @@ -0,0 +1,278 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Backpressure on the no-accumulation passthrough edge. A fast append-mode + * producer feeds a deliberately slow same-mode consumer over a passthrough + * edge with a small `streamHighWaterBytes`. The consumer-edge gate must park + * the producer so its lead over the consumer (bytes emitted but not yet read) + * stays within a small multiple of the high-water mark — instead of the + * producer racing to completion and the whole stream buffering in memory. + * + * With `noAccumulation` off, the same graph takes the accumulation drain: the + * producer runs free (no pacing) and the edge materializes the full string — + * the pre-change behavior. + */ + +import type { CacheRef, StreamEvent, TaskInput, TaskOutput } from "@workglow/task-graph"; +import { + Dataflow, + IExecuteContext, + isCacheRef, + makeCacheRef, + Task, + TaskGraph, + TaskGraphRunner, + TaskOutputRepository, +} from "@workglow/task-graph"; +import { sleep } from "@workglow/util"; +import { DataPortSchema } from "@workglow/util/schema"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +const CHUNK_BYTES = 64; +const CHUNKS = 40; +const TOTAL_BYTES = CHUNK_BYTES * CHUNKS; +const HIGH_WATER = 4 * CHUNK_BYTES; + +// -------------------------------------------------------------------------- +// Streaming-port in-memory cache (per-port byte streams keyed by task+inputs). +// The sink drains immediately, so the only thing that can pace the producer is +// the consumer-edge gate under test. +// -------------------------------------------------------------------------- +class StreamPortMemoryRepo extends TaskOutputRepository { + private rows = new Map(); + private blobs = new Map(); + + constructor() { + super({ outputCompression: false }); + } + override async saveOutput(taskType: string, inputs: TaskInput, output: TaskOutput) { + this.rows.set(taskType + JSON.stringify(inputs), output); + } + override async getOutput(taskType: string, inputs: TaskInput) { + return this.rows.get(taskType + JSON.stringify(inputs)); + } + override async clear() { + this.rows.clear(); + this.blobs.clear(); + } + override async size() { + return this.rows.size; + } + override async clearOlderThan() {} + override isDurable() { + return false; + } + + override async saveOutputStreamPort( + taskType: string, + inputs: TaskInput, + port: string, + mode: string, + chunks: AsyncIterable, + _metadata: Record + ): Promise { + const parts: number[] = []; + for await (const c of chunks) for (const b of c) parts.push(b); + const bytes = Uint8Array.from(parts); + const key = `inmem://${taskType}::${JSON.stringify(inputs)}::${port}`; + this.blobs.set(key, bytes); + return makeCacheRef({ + $ref: key, + port, + mode: mode as CacheRef["mode"], + size: bytes.byteLength, + }); + } + override async getOutputByRef(ref: CacheRef): Promise { + const bytes = this.blobs.get(ref.$ref); + return bytes === undefined ? undefined : new Blob([bytes as Uint8Array]); + } + override getOutputStreamByRef(ref: CacheRef): AsyncIterable | undefined { + const bytes = this.blobs.get(ref.$ref); + if (bytes === undefined) return undefined; + return (async function* () { + yield bytes; + })(); + } +} + +type Out = { text: string }; + +/** Shared per-run counters so the producer can observe the consumer's progress. */ +class RunMeter { + producedBytes = 0; + consumedBytes = 0; + peakLeadBytes = 0; + noteProduced(bytes: number): void { + this.producedBytes += bytes; + this.peakLeadBytes = Math.max(this.peakLeadBytes, this.producedBytes - this.consumedBytes); + } + noteConsumed(bytes: number): void { + this.consumedBytes += bytes; + } +} + +/** + * Fast cacheable append-mode source: emits `CHUNKS` ASCII chunks of + * `CHUNK_BYTES` each with no delay. Each yield resumes only after the + * streaming runtime's per-event awaits, so `noteProduced` (called before the + * next yield) reflects any pacing applied to the producer. + */ +class FastAppendSource extends Task, Out> { + public static override type = "StreamBackpressureEngaged_Source"; + public static override category = "Test"; + public static override cacheable = true; + public meter: RunMeter | undefined; + + public static override inputSchema(): DataPortSchema { + return { type: "object", properties: {}, additionalProperties: false } as const; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string", "x-stream": "append" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + async *executeStream(): AsyncIterable> { + for (let i = 0; i < CHUNKS; i++) { + const chunk = String.fromCharCode(97 + (i % 26)).repeat(CHUNK_BYTES); + yield { type: "text-delta", port: "text", textDelta: chunk }; + this.meter?.noteProduced(CHUNK_BYTES); + } + yield { type: "finish", data: {} as Out }; + } +} + +/** Slow same-mode passthrough consumer: sleeps between reads of the live stream. */ +class SlowAppendConsumer extends Task<{ text: string }, Out> { + public static override type = "StreamBackpressureEngaged_Consumer"; + public static override category = "Test"; + public static override cacheable = false; + public meter: RunMeter | undefined; + + public static override inputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string", "x-stream": "append" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string", "x-stream": "append" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + async *executeStream( + input: { text: string }, + ctx: IExecuteContext + ): AsyncIterable> { + const stream = ctx.inputStreams?.get("text"); + if (!stream) { + yield { type: "text-delta", port: "text", textDelta: input.text ?? "" }; + yield { type: "finish", data: {} as Out }; + return; + } + const reader = stream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (value.type === "text-delta") { + await sleep(2); + this.meter?.noteConsumed(value.textDelta.length); + yield { type: "text-delta", port: "text", textDelta: value.textDelta }; + } + } + } finally { + reader.releaseLock(); + } + yield { type: "finish", data: {} as Out }; + } +} + +function buildGraph(meter: RunMeter): { + graph: TaskGraph; + edge: Dataflow; + consumer: SlowAppendConsumer; +} { + const graph = new TaskGraph(); + const source = new FastAppendSource({ id: "source" }); + source.meter = meter; + // threshold 0 keeps the per-port ref in Output so the edge observably + // carries a CacheRef rather than a rehydrated inline string. + source.runConfig = { ...source.runConfig, referenceThresholdBytes: 0 }; + const consumer = new SlowAppendConsumer({ id: "consumer" }); + consumer.meter = meter; + graph.addTasks([source, consumer]); + const edge = new Dataflow("source", "text", "consumer", "text"); + graph.addDataflow(edge); + return { graph, edge, consumer }; +} + +const expectedText = (): string => { + let s = ""; + for (let i = 0; i < CHUNKS; i++) s += String.fromCharCode(97 + (i % 26)).repeat(CHUNK_BYTES); + return s; +}; + +describe("no-accumulation passthrough backpressure", () => { + let cache: StreamPortMemoryRepo; + beforeEach(() => { + cache = new StreamPortMemoryRepo(); + }); + afterEach(async () => { + await cache.clear(); + }); + + it("parks a fast producer to the slow consumer's read rate (peak lead bounded)", async () => { + const meter = new RunMeter(); + const { graph, edge, consumer } = buildGraph(meter); + const runner = new TaskGraphRunner(graph); + const results = await runner.runGraph( + {}, + { outputCache: cache, noAccumulation: true, streamHighWaterBytes: HIGH_WATER } + ); + + // The producer's lead over the consumer stayed within the gate's bound: + // the high-water mark plus a small pipeline allowance (the chunk that + // crossed the mark, the edge wrapper's one-event prefetch, and the event + // in flight between charge and the consumer's counter). + expect(meter.peakLeadBytes).toBeLessThanOrEqual(HIGH_WATER + 3 * CHUNK_BYTES); + // Sanity: the bound is meaningful — far below free-running distance. + expect(TOTAL_BYTES).toBeGreaterThan(HIGH_WATER * 4); + + // Passthrough semantics held: the edge carries the per-port ref, and the + // consumer's paced read still produced the complete output. + expect(isCacheRef(edge.value)).toBe(true); + const consumerResult = results.find((r) => r.id === "consumer"); + expect((consumerResult!.data as Out).text).toBe(expectedText()); + expect(meter.consumedBytes).toBe(TOTAL_BYTES); + void consumer; + }, 20_000); + + it("with noAccumulation off, the producer runs free and the edge materializes", async () => { + const meter = new RunMeter(); + const { graph, edge } = buildGraph(meter); + const runner = new TaskGraphRunner(graph); + const results = await runner.runGraph( + {}, + { outputCache: cache, streamHighWaterBytes: HIGH_WATER } + ); + + // Off-path: no gate — the producer's lead is unbounded by the mark (it + // races well past it while the slow consumer lags behind). + expect(meter.peakLeadBytes).toBeGreaterThan(HIGH_WATER * 4); + + // Pre-change behavior: the edge drains to the materialized string. + expect(edge.value).toBe(expectedText()); + const consumerResult = results.find((r) => r.id === "consumer"); + expect((consumerResult!.data as Out).text).toBe(expectedText()); + }, 20_000); +}); diff --git a/packages/test/src/test/task-graph/StreamBinaryPump.test.ts b/packages/test/src/test/task-graph/StreamBinaryPump.test.ts index 006cdd936..8694189b4 100644 --- a/packages/test/src/test/task-graph/StreamBinaryPump.test.ts +++ b/packages/test/src/test/task-graph/StreamBinaryPump.test.ts @@ -259,9 +259,10 @@ describe("StreamBinaryPump — C1 binary source → non-binary consumer", () => // // These tests assert the DECISION in isolation, not a real-run outcome. We // deliberately do NOT run a streaming-cache graph and assert "binary port absent -// from finish" as correct: in the reduced scope nothing drives saveOutputStream -// on a real run, so absent bytes there means SILENT DATA LOSS, not success. The -// live pipe (cache actually receiving the bytes on a real run) lands in Spec 2. +// from finish" as correct: with no live sink driving saveOutputStream on a real +// run, absent bytes there means SILENT DATA LOSS, not success. The live pipe +// (cache actually receiving the bytes on a real run) is covered by the +// per-port sink and cache stream-out suites. // ============================================================================ describe("StreamBinaryPump.canStreamBinaryToCache — decision", () => { @@ -309,7 +310,7 @@ describe("StreamBinaryPump.canStreamBinaryToCache — decision", () => { }); // ============================================================================ -// Stream-out decision: anyConsumerAcceptsBinaryStream +// Stream-out decision: anyConsumerAcceptsStream (binary schemas) // ============================================================================ /** Streaming consumer: its `bytes` input port accepts the binary stream mode. */ @@ -340,7 +341,7 @@ class BinaryStreamConsumer extends Task { } } -describe("StreamPump.anyConsumerAcceptsBinaryStream", () => { +describe("StreamPump.anyConsumerAcceptsStream (binary)", () => { it("returns true when an out-edge targets a binary-streaming input port", () => { const graph = new TaskGraph(); const source = new CacheableBinaryStreamSource({ id: "source" }); @@ -348,7 +349,7 @@ describe("StreamPump.anyConsumerAcceptsBinaryStream", () => { graph.addTasks([source, consumer]); graph.addDataflow(new Dataflow("source", "bytes", "consumer", "bytes")); - expect(StreamPump.anyConsumerAcceptsBinaryStream(graph, source)).toBe(true); + expect(StreamPump.anyConsumerAcceptsStream(graph, source)).toBe(true); }); it("returns false with no consumers", () => { @@ -356,7 +357,7 @@ describe("StreamPump.anyConsumerAcceptsBinaryStream", () => { const source = new CacheableBinaryStreamSource({ id: "source" }); graph.addTask(source); - expect(StreamPump.anyConsumerAcceptsBinaryStream(graph, source)).toBe(false); + expect(StreamPump.anyConsumerAcceptsStream(graph, source)).toBe(false); }); it("returns false when the only consumer needs a materialized value", () => { @@ -366,7 +367,7 @@ describe("StreamPump.anyConsumerAcceptsBinaryStream", () => { graph.addTasks([source, sink]); graph.addDataflow(new Dataflow("source", "bytes", "sink", "bytes")); - expect(StreamPump.anyConsumerAcceptsBinaryStream(graph, source)).toBe(false); + expect(StreamPump.anyConsumerAcceptsStream(graph, source)).toBe(false); }); it("returns false for * fan-out edges (consumers receive materialized values)", () => { @@ -376,7 +377,7 @@ describe("StreamPump.anyConsumerAcceptsBinaryStream", () => { graph.addTasks([source, consumer]); graph.addDataflow(new Dataflow("source", "*", "consumer", "*")); - expect(StreamPump.anyConsumerAcceptsBinaryStream(graph, source)).toBe(false); + expect(StreamPump.anyConsumerAcceptsStream(graph, source)).toBe(false); }); it("returns true with mixed consumers (one streams, one materializes)", () => { @@ -388,7 +389,7 @@ describe("StreamPump.anyConsumerAcceptsBinaryStream", () => { graph.addDataflow(new Dataflow("source", "bytes", "consumer", "bytes")); graph.addDataflow(new Dataflow("source", "bytes", "sink", "bytes")); - expect(StreamPump.anyConsumerAcceptsBinaryStream(graph, source)).toBe(true); + expect(StreamPump.anyConsumerAcceptsStream(graph, source)).toBe(true); }); }); @@ -428,8 +429,8 @@ describe("StreamBinaryPump — C2 accumulation materializes bytes on a real run" }); it("tees when a downstream edge needs materialized AND the cache can stream", async () => { - // Spec 2 Phase E: cache-can-stream + downstream-needs-materialized used to - // inhibit refs entirely. Now both paths fire — accumulator drives the + // cache-can-stream + downstream-needs-materialized used to inhibit refs + // entirely. Now both paths fire — accumulator drives the // enriched finish event (Blob for the edge consumer) and the router // writes to the cache so the queue/cache row stays small. const graph = new TaskGraph(); diff --git a/packages/test/src/test/task-graph/StreamBinaryTypes.test.ts b/packages/test/src/test/task-graph/StreamBinaryTypes.test.ts index df2ca1fe9..bcb29da94 100644 --- a/packages/test/src/test/task-graph/StreamBinaryTypes.test.ts +++ b/packages/test/src/test/task-graph/StreamBinaryTypes.test.ts @@ -7,7 +7,6 @@ import type { StreamBinaryDelta, StreamEvent, StreamMode } from "@workglow/task- import { assertBinaryFormat, edgeNeedsAccumulation, - getBinaryPortId, getOutputStreamMode, getPortStreamMode, getStreamingPorts, @@ -94,19 +93,6 @@ describe("binary-aware port helpers", () => { expect(getOutputStreamMode(mixedSchema)).toBe("mixed"); }); - it("getBinaryPortId finds the first binary port", () => { - expect(getBinaryPortId(binarySchema)).toBe("bytes"); - expect(getBinaryPortId(mixedSchema)).toBe("bytes"); - }); - - it("getBinaryPortId returns undefined when no binary port", () => { - const noBinary = { - type: "object", - properties: { text: { type: "string", "x-stream": "append" } }, - } as const satisfies DataPortSchema; - expect(getBinaryPortId(noBinary)).toBeUndefined(); - }); - it("edgeNeedsAccumulation: binary source → non-binary target accumulates", () => { const target = { type: "object", diff --git a/packages/test/src/test/task-graph/StreamMixedModeFanout.test.ts b/packages/test/src/test/task-graph/StreamMixedModeFanout.test.ts new file mode 100644 index 000000000..8f2ae43fe --- /dev/null +++ b/packages/test/src/test/task-graph/StreamMixedModeFanout.test.ts @@ -0,0 +1,414 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * No-accumulation passthrough coverage beyond the single-port case: + * + * - **Mixed-mode**: one source with an `append` port and an `object` port, + * each feeding its own same-mode passthrough consumer. Each port gets its + * own gate: a slow object consumer bounds the producer's lead on the object + * port while the fast text consumer's port flows freely — pacing is + * per-port, not per-task. + * + * - **Fan-out**: a single source port feeding TWO same-mode consumers is NOT + * a passthrough edge (the predicate requires a single consumer of the source + * port), so it falls back to the tee'd drain: correct, in-order delivery to + * both consumers, but only best-effort pacing (no precise gate). + */ + +import type { CacheRef, StreamEvent, TaskInput, TaskOutput } from "@workglow/task-graph"; +import { + Dataflow, + IExecuteContext, + isCacheRef, + makeCacheRef, + StreamPump, + Task, + TaskGraph, + TaskGraphRunner, + TaskOutputRepository, +} from "@workglow/task-graph"; +import { sleep } from "@workglow/util"; +import { DataPortSchema } from "@workglow/util/schema"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +const CHUNKS = 30; +const TEXT_CHUNK = "abcdefgh"; // 8 bytes per text delta +const HIGH_WATER = 256; + +// -------------------------------------------------------------------------- +// Streaming-port in-memory cache (per-port byte streams keyed by task+inputs). +// -------------------------------------------------------------------------- +class StreamPortMemoryRepo extends TaskOutputRepository { + public savePortCalls: Array<{ port: string; mode: string }> = []; + private rows = new Map(); + private blobs = new Map(); + + constructor() { + super({ outputCompression: false }); + } + override async saveOutput(taskType: string, inputs: TaskInput, output: TaskOutput) { + this.rows.set(taskType + JSON.stringify(inputs), output); + } + override async getOutput(taskType: string, inputs: TaskInput) { + return this.rows.get(taskType + JSON.stringify(inputs)); + } + override async clear() { + this.rows.clear(); + this.blobs.clear(); + } + override async size() { + return this.rows.size; + } + override async clearOlderThan() {} + override isDurable() { + return false; + } + + override async saveOutputStreamPort( + taskType: string, + inputs: TaskInput, + port: string, + mode: string, + chunks: AsyncIterable, + _metadata: Record + ): Promise { + this.savePortCalls.push({ port, mode }); + const parts: number[] = []; + for await (const c of chunks) for (const b of c) parts.push(b); + const bytes = Uint8Array.from(parts); + const key = `inmem://${taskType}::${JSON.stringify(inputs)}::${port}`; + this.blobs.set(key, bytes); + return makeCacheRef({ + $ref: key, + port, + mode: mode as CacheRef["mode"], + size: bytes.byteLength, + }); + } + override async getOutputByRef(ref: CacheRef): Promise { + const bytes = this.blobs.get(ref.$ref); + return bytes === undefined ? undefined : new Blob([bytes as Uint8Array]); + } + override getOutputStreamByRef(ref: CacheRef): AsyncIterable | undefined { + const bytes = this.blobs.get(ref.$ref); + if (bytes === undefined) return undefined; + return (async function* () { + yield bytes; + })(); + } +} + +// ========================================================================== +// Mixed-mode: append + object ports, each with its own passthrough consumer. +// ========================================================================== + +type MixedOut = { text: string; items: unknown[] }; + +const objDelta = (i: number): unknown[] => [{ id: i, pad: "x".repeat(32) }]; + +/** Per-port producer/consumer byte accounting, using streamEventCost's units. */ +class PortMeter { + produced = 0; + consumed = 0; + peakLead = 0; + noteProduced(cost: number): void { + this.produced += cost; + this.peakLead = Math.max(this.peakLead, this.produced - this.consumed); + } + noteConsumed(cost: number): void { + this.consumed += cost; + } +} + +/** Interleaves text deltas (fast port) with object deltas (slow port). */ +class MixedSource extends Task, MixedOut> { + public static override type = "StreamMixedMode_Source"; + public static override category = "Test"; + public static override cacheable = true; + public textMeter: PortMeter | undefined; + public objMeter: PortMeter | undefined; + + public static override inputSchema(): DataPortSchema { + return { type: "object", properties: {}, additionalProperties: false } as const; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { + text: { type: "string", "x-stream": "append" }, + items: { type: "array", "x-stream": "object" }, + }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + async *executeStream(): AsyncIterable> { + for (let i = 0; i < CHUNKS; i++) { + yield { type: "text-delta", port: "text", textDelta: TEXT_CHUNK }; + this.textMeter?.noteProduced(TEXT_CHUNK.length); + const delta = objDelta(i); + yield { type: "object-delta", port: "items", objectDelta: delta }; + this.objMeter?.noteProduced(JSON.stringify(delta).length); + } + yield { type: "finish", data: {} as MixedOut }; + } +} + +/** Fast append passthrough consumer for the `text` port. */ +class FastTextConsumer extends Task<{ text: string }, { text: string }> { + public static override type = "StreamMixedMode_TextConsumer"; + public static override category = "Test"; + public static override cacheable = false; + public meter: PortMeter | undefined; + + public static override inputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string", "x-stream": "append" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string", "x-stream": "append" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + async *executeStream( + _input: { text: string }, + ctx: IExecuteContext + ): AsyncIterable> { + const stream = ctx.inputStreams?.get("text"); + if (stream) { + const reader = stream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (value.type === "text-delta") { + this.meter?.noteConsumed(value.textDelta.length); + yield { type: "text-delta", port: "text", textDelta: value.textDelta }; + } + } + } finally { + reader.releaseLock(); + } + } + yield { type: "finish", data: {} as { text: string } }; + } +} + +/** Slow object passthrough consumer for the `items` port. */ +class SlowObjectConsumer extends Task<{ items: unknown[] }, { items: unknown[] }> { + public static override type = "StreamMixedMode_ObjectConsumer"; + public static override category = "Test"; + public static override cacheable = false; + public meter: PortMeter | undefined; + + public static override inputSchema(): DataPortSchema { + return { + type: "object", + properties: { items: { type: "array", "x-stream": "object" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { items: { type: "array", "x-stream": "object" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + async *executeStream( + _input: { items: unknown[] }, + ctx: IExecuteContext + ): AsyncIterable> { + const stream = ctx.inputStreams?.get("items"); + if (stream) { + const reader = stream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (value.type === "object-delta") { + await sleep(2); + this.meter?.noteConsumed(JSON.stringify(value.objectDelta).length); + yield { type: "object-delta", port: "items", objectDelta: value.objectDelta }; + } + } + } finally { + reader.releaseLock(); + } + } + yield { type: "finish", data: {} as { items: unknown[] } }; + } +} + +// ========================================================================== +// Fan-out: one append port feeding two same-mode consumers. +// ========================================================================== + +/** Append passthrough reader that records every delta it sees, in order. */ +class RecordingTextConsumer extends Task<{ text: string }, { text: string }> { + public static override type = "StreamFanout_RecordingConsumer"; + public static override category = "Test"; + public static override cacheable = false; + public received: string[] = []; + + public static override inputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string", "x-stream": "append" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string", "x-stream": "append" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + async *executeStream( + _input: { text: string }, + ctx: IExecuteContext + ): AsyncIterable> { + const stream = ctx.inputStreams?.get("text"); + if (stream) { + const reader = stream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (value.type === "text-delta") { + this.received.push(value.textDelta); + yield { type: "text-delta", port: "text", textDelta: value.textDelta }; + } + } + } finally { + reader.releaseLock(); + } + } + yield { type: "finish", data: {} as { text: string } }; + } +} + +/** Append source emitting ordered, distinguishable chunks. */ +class OrderedAppendSource extends Task, { text: string }> { + public static override type = "StreamFanout_Source"; + public static override category = "Test"; + public static override cacheable = true; + + public static override inputSchema(): DataPortSchema { + return { type: "object", properties: {}, additionalProperties: false } as const; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string", "x-stream": "append" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + async *executeStream(): AsyncIterable> { + for (let i = 0; i < CHUNKS; i++) { + yield { type: "text-delta", port: "text", textDelta: `[${i}]` }; + } + yield { type: "finish", data: {} as { text: string } }; + } +} + +describe("no-accumulation mixed-mode and fan-out", () => { + let cache: StreamPortMemoryRepo; + beforeEach(() => { + cache = new StreamPortMemoryRepo(); + }); + afterEach(async () => { + await cache.clear(); + }); + + it("mixed-mode: each port sinks and paces independently against its own consumer", async () => { + const graph = new TaskGraph(); + const source = new MixedSource({ id: "source" }); + source.runConfig = { ...source.runConfig, referenceThresholdBytes: 0 }; + const textMeter = new PortMeter(); + const objMeter = new PortMeter(); + source.textMeter = textMeter; + source.objMeter = objMeter; + const textConsumer = new FastTextConsumer({ id: "textConsumer" }); + textConsumer.meter = textMeter; + const objectConsumer = new SlowObjectConsumer({ id: "objectConsumer" }); + objectConsumer.meter = objMeter; + graph.addTasks([source, textConsumer, objectConsumer]); + const textEdge = new Dataflow("source", "text", "textConsumer", "text"); + const itemsEdge = new Dataflow("source", "items", "objectConsumer", "items"); + graph.addDataflow(textEdge); + graph.addDataflow(itemsEdge); + + const runner = new TaskGraphRunner(graph); + const results = await runner.runGraph( + {}, + { outputCache: cache, noAccumulation: true, streamHighWaterBytes: HIGH_WATER } + ); + + // Both ports were sunk per-port with their own codec mode. + expect(cache.savePortCalls).toContainEqual({ port: "text", mode: "append" }); + expect(cache.savePortCalls).toContainEqual({ port: "items", mode: "object" }); + + // Both edges skipped the drain and carry per-port refs. + expect(isCacheRef(textEdge.value)).toBe(true); + expect(isCacheRef(itemsEdge.value)).toBe(true); + + // The slow object consumer bounded the producer's lead on ITS port. + const maxObjCost = JSON.stringify(objDelta(CHUNKS - 1)).length; + expect(objMeter.peakLead).toBeLessThanOrEqual(HIGH_WATER + 3 * maxObjCost); + // The fast text port was never the bottleneck: its lead stays within its + // own small pipeline allowance, not the slow port's backlog. + expect(textMeter.peakLead).toBeLessThanOrEqual(HIGH_WATER + 3 * TEXT_CHUNK.length); + + // Complete, correct outputs on both branches. + const textResult = results.find((r) => r.id === "textConsumer"); + expect((textResult!.data as { text: string }).text).toBe(TEXT_CHUNK.repeat(CHUNKS)); + const objResult = results.find((r) => r.id === "objectConsumer"); + const items = (objResult!.data as { items: unknown[] }).items; + expect(items).toHaveLength(CHUNKS); + expect(items[0]).toEqual({ id: 0, pad: "x".repeat(32) }); + expect(items[CHUNKS - 1]).toEqual({ id: CHUNKS - 1, pad: "x".repeat(32) }); + }, 20_000); + + it("fan-out: two consumers of one port fall back to the drain and both receive every event in order", async () => { + const graph = new TaskGraph(); + const source = new OrderedAppendSource({ id: "source" }); + const consumerA = new RecordingTextConsumer({ id: "consumerA" }); + const consumerB = new RecordingTextConsumer({ id: "consumerB" }); + graph.addTasks([source, consumerA, consumerB]); + const edgeA = new Dataflow("source", "text", "consumerA", "text"); + const edgeB = new Dataflow("source", "text", "consumerB", "text"); + graph.addDataflow(edgeA); + graph.addDataflow(edgeB); + + // A fanned-out source port is not a passthrough edge — pacing is + // best-effort by design (the precise gate is single-consumer only). + expect(StreamPump.isNoAccumulationPassthroughEdge(graph, edgeA, true)).toBe(false); + expect(StreamPump.isNoAccumulationPassthroughEdge(graph, edgeB, true)).toBe(false); + + const runner = new TaskGraphRunner(graph); + const results = await runner.runGraph( + {}, + { outputCache: cache, noAccumulation: true, streamHighWaterBytes: HIGH_WATER } + ); + + // Every consumer saw the identical, in-order delta sequence. + const expected = Array.from({ length: CHUNKS }, (_, i) => `[${i}]`); + expect(consumerA.received).toEqual(expected); + expect(consumerB.received).toEqual(expected); + + for (const id of ["consumerA", "consumerB"]) { + const result = results.find((r) => r.id === id); + expect((result!.data as { text: string }).text).toBe(expected.join("")); + } + }, 20_000); +}); diff --git a/packages/test/src/test/task-graph/resolveJobOutputStream.test.ts b/packages/test/src/test/task-graph/resolveJobOutputStream.test.ts index d2e34738c..984e1dffa 100644 --- a/packages/test/src/test/task-graph/resolveJobOutputStream.test.ts +++ b/packages/test/src/test/task-graph/resolveJobOutputStream.test.ts @@ -91,6 +91,19 @@ describe("resolveJobOutputStream", () => { const resolver = makeJobOutputStreamResolver(repo); const stream = await resolver({ file: ref }, "file"); expect(await collect(stream!)).toEqual([7, 7]); - expect(await resolver({ file: "not-binary" }, "file")).toBeUndefined(); + }); + + it("adapts an inline string at a named port to its UTF-8 bytes", async () => { + // A below-threshold append-mode ref hydrates to an inline string; the + // stream form must match what the surviving-ref form would have yielded, + // so caller behavior does not flip on payload size. + const repo = new StreamingMemoryRepo({}); + const stream = await resolveJobOutputStream(handleFor({ text: "hi" }), repo, "text"); + expect(await collect(stream!)).toEqual([104, 105]); + }); + + it("resolves undefined for an unsupported inline value", async () => { + const repo = new StreamingMemoryRepo({}); + expect(await resolveJobOutputStream(handleFor({ n: 42 }), repo, "n")).toBeUndefined(); }); }); diff --git a/providers/postgres/src/storage/PostgresTabularStorage.ts b/providers/postgres/src/storage/PostgresTabularStorage.ts index 7e0f35c78..7c00287f0 100644 --- a/providers/postgres/src/storage/PostgresTabularStorage.ts +++ b/providers/postgres/src/storage/PostgresTabularStorage.ts @@ -373,8 +373,7 @@ export class PostgresTabularStorage< */ protected override sqlToJsValue(column: string, value: ValueOptionType): Entity[keyof Entity] { const typeDef = this.schema.properties[column as keyof typeof this.schema.properties] as - | JsonSchema - | undefined; + JsonSchema | undefined; if (typeDef) { if (value === null && this.isNullable(typeDef)) { return null as Entity[keyof Entity]; diff --git a/providers/sqlite/src/storage/SqliteTabularStorage.ts b/providers/sqlite/src/storage/SqliteTabularStorage.ts index 372ec1ece..ca8b37b1f 100644 --- a/providers/sqlite/src/storage/SqliteTabularStorage.ts +++ b/providers/sqlite/src/storage/SqliteTabularStorage.ts @@ -272,8 +272,7 @@ export class SqliteTabularStorage< // Handle null values if (value === null) { const typeDef = this.schema.properties[column as keyof typeof this.schema.properties] as - | JsonSchema - | undefined; + JsonSchema | undefined; if (typeDef && this.isNullable(typeDef)) { return null; } @@ -282,8 +281,7 @@ export class SqliteTabularStorage< // Schema-based type handling for non-object/array values const typeDef = this.schema.properties[column as keyof typeof this.schema.properties] as - | JsonSchema - | undefined; + JsonSchema | undefined; if (typeDef) { const actualType = this.getNonNullType(typeDef); const isObject = @@ -338,8 +336,7 @@ export class SqliteTabularStorage< */ protected override sqlToJsValue(column: string, value: ValueOptionType): Entity[keyof Entity] { const typeDef = this.schema.properties[column as keyof typeof this.schema.properties] as - | JsonSchema - | undefined; + JsonSchema | undefined; if (typeDef) { if (value === null && this.isNullable(typeDef)) { return null as Entity[keyof Entity]; diff --git a/providers/supabase/src/storage/SupabaseTabularStorage.ts b/providers/supabase/src/storage/SupabaseTabularStorage.ts index ae6618217..2a419cab3 100644 --- a/providers/supabase/src/storage/SupabaseTabularStorage.ts +++ b/providers/supabase/src/storage/SupabaseTabularStorage.ts @@ -318,8 +318,7 @@ export class SupabaseTabularStorage< */ protected override sqlToJsValue(column: string, value: ValueOptionType): Entity[keyof Entity] { const typeDef = this.schema.properties[column as keyof typeof this.schema.properties] as - | JsonSchema - | undefined; + JsonSchema | undefined; if (typeDef) { if (value === null && this.isNullable(typeDef)) { return null as Entity[keyof Entity];