diff --git a/packages/storage/src/tabular/SharedInMemoryTabularStorage.ts b/packages/storage/src/tabular/SharedInMemoryTabularStorage.ts index 5c9b49e0c..2f9bc607b 100644 --- a/packages/storage/src/tabular/SharedInMemoryTabularStorage.ts +++ b/packages/storage/src/tabular/SharedInMemoryTabularStorage.ts @@ -60,6 +60,14 @@ export class SharedInMemoryTabularStorage< private isInitialized = false; private syncInProgress = false; private pendingMessages: BroadcastMessage[] = []; + /** + * Resolves when the current `syncFromOtherTabs()` settles, either because a + * peer tab sent a `SYNC_RESPONSE` (drained, then resolved) or because the + * sync timeout fired. `setupDatabase({ awaitTabSync: true })` awaits this so + * callers that immediately read after setup do not race the replay. + */ + private syncSettled: Promise | null = null; + private syncSettledResolver: (() => void) | null = null; constructor( channelName: string = "tabular_store", @@ -151,6 +159,7 @@ export class SharedInMemoryTabularStorage< } this.syncInProgress = false; await this.drainPendingMessages(); + this.resolveSyncSettled(); break; case "PUT": @@ -196,18 +205,33 @@ export class SharedInMemoryTabularStorage< if (!this.channel) return; this.syncInProgress = true; + this.syncSettled = new Promise((resolve) => { + this.syncSettledResolver = resolve; + }); this.channel.postMessage({ type: "SYNC_REQUEST" } as BroadcastMessage); setTimeout(() => { if (this.syncInProgress) { this.syncInProgress = false; - void this.drainPendingMessages().catch((error) => { - console.error("Failed to drain pending messages after sync timeout", error); - }); + void this.drainPendingMessages() + .catch((error) => { + console.error("Failed to drain pending messages after sync timeout", error); + }) + .finally(() => { + this.resolveSyncSettled(); + }); } }, SYNC_TIMEOUT); } + private resolveSyncSettled(): void { + const resolver = this.syncSettledResolver; + if (resolver) { + this.syncSettledResolver = null; + resolver(); + } + } + private async copyDataFromArray(entities: Entity[]): Promise { if (entities.length === 0) return; await this.inMemoryRepo.deleteAll(); @@ -220,12 +244,23 @@ export class SharedInMemoryTabularStorage< } } - public override async setupDatabase(): Promise { + /** + * Initialize the cross-tab sync barrier and apply any pending migrations. + * + * Defaults to `awaitTabSync: true` so that callers that immediately read or + * write after `setupDatabase()` resolves see other tabs' replayed rows; the + * fire-and-forget shape regressed in 64591e3 and caused first-call races. + * Pass `awaitTabSync: false` only for latency-sensitive startup paths that + * explicitly tolerate a stale initial view. + */ + public override async setupDatabase(options: { awaitTabSync?: boolean } = {}): Promise { if (this.isInitialized) return; this.isInitialized = true; - // Fire-and-forget: posts a SYNC_REQUEST and drains responses on a timeout - // (see syncFromOtherTabs); there is nothing to await. + const awaitTabSync = options.awaitTabSync ?? true; this.syncFromOtherTabs(); + if (awaitTabSync && this.syncSettled) { + await this.syncSettled; + } if (this.tabularMigrations && this.tabularMigrations.length > 0) { await this.applyTabularMigrations(); } diff --git a/packages/task-graph/src/cache/RunPrivateCacheRepo.ts b/packages/task-graph/src/cache/RunPrivateCacheRepo.ts index 20b9dd1ce..9b0e137b9 100644 --- a/packages/task-graph/src/cache/RunPrivateCacheRepo.ts +++ b/packages/task-graph/src/cache/RunPrivateCacheRepo.ts @@ -23,10 +23,21 @@ export interface RunPrivateCacheRepoOptions { * - Two wrappers with the same `runId` (e.g., a restart after a crash) see each * other's writes via the backing store — that's the restart-survival contract. * - Two wrappers with different `runId`s see only their own entries. + * + * Durable restart-resume requires deterministic task ids. When a task does not + * pin `config.id` to a stable value the default constructor mints a fresh v4 + * UUID per process, so the cache key would change across restarts and orphan + * every row. In that case {@link CacheCoordinator} keys by task type instead + * and signals the fallback via {@link noteFallbackKey}; this wrapper warns once + * per process to tell the operator that the cache is best-effort intra-process + * only until task ids are pinned. */ export class RunPrivateCacheRepo extends TaskOutputRepository { + private static fallbackWarned = false; + private readonly backing: TaskOutputRepository; private readonly runId: string; + private observedFallback: boolean = false; constructor({ backing, runId }: RunPrivateCacheRepoOptions) { super({ outputCompression: backing.outputCompression }); @@ -34,6 +45,27 @@ export class RunPrivateCacheRepo extends TaskOutputRepository { this.runId = runId; } + /** + * Called by {@link CacheCoordinator} the first time it has to key a private + * cache entry by task type because the task id is autogenerated. Emits a + * single `console.warn` per process so deployments without pinned ids learn + * that crash-resume is degraded; subsequent fallbacks are silent. + */ + public noteFallbackKey(): void { + this.observedFallback = true; + if (RunPrivateCacheRepo.fallbackWarned) return; + RunPrivateCacheRepo.fallbackWarned = true; + console.warn( + "[RunPrivateCacheRepo] Private cache fell back to taskType keying because " + + "task.id is autogenerated; pin task.id to enable crash-resume." + ); + } + + /** @internal Test-only hook for inspecting whether any fallback was observed. */ + public hasObservedFallback(): boolean { + return this.observedFallback; + } + private ns(cacheIdentity: string): string { return `__run:${this.runId}::${cacheIdentity}`; } diff --git a/packages/task-graph/src/task/CacheCoordinator.ts b/packages/task-graph/src/task/CacheCoordinator.ts index 304617c67..187c9ce63 100644 --- a/packages/task-graph/src/task/CacheCoordinator.ts +++ b/packages/task-graph/src/task/CacheCoordinator.ts @@ -7,6 +7,7 @@ import { getPortCodec } from "@workglow/util"; import { type CachePolicy, isPolicyCached, isPolicyPrivate } from "../cache/CachePolicy"; import type { CacheRegistry } from "../cache/CacheRegistry"; +import { RunPrivateCacheRepo } from "../cache/RunPrivateCacheRepo"; import type { TaskOutputRepository } from "../storage/TaskOutputRepository"; import type { ITask } from "./ITask"; import type { StreamEvent } from "./StreamTypes"; @@ -63,11 +64,20 @@ export class CacheCoordinator { if (!outputCache || !this.task.cacheable) return undefined; - const cached = await outputCache.getOutput(this.cacheIdentityKey(policy), keyInputs); + const cached = await outputCache.getOutput( + this.cacheIdentityKey(policy, outputCache), + keyInputs + ); if (cached === undefined) return undefined; const outputSchema = (this.task.constructor as typeof Task).outputSchema(); @@ -118,7 +131,11 @@ export class CacheCoordinator, outputSchema as unknown as SchemaProperties ); - await outputCache.saveOutput(this.cacheIdentityKey(policy), keyInputs, wireOutputs as Output); + await outputCache.saveOutput( + this.cacheIdentityKey(policy, outputCache), + keyInputs, + wireOutputs as Output + ); } // ======================================================================== diff --git a/packages/task-graph/src/task/ITask.ts b/packages/task-graph/src/task/ITask.ts index 51daa6321..b05c2386c 100644 --- a/packages/task-graph/src/task/ITask.ts +++ b/packages/task-graph/src/task/ITask.ts @@ -283,6 +283,13 @@ export interface ITaskSerialization { export interface ITaskState { readonly config: Config; get id(): unknown; + /** + * Returns true when `id` is a stable, caller-pinned value (not the + * autogenerated v4 UUID default). The run-private cache uses this to + * decide whether to key by instance id (deterministic) or fall back to + * keying by task type so a crash-restart can still find prior rows. + */ + hasDeterministicId(): boolean; status: TaskStatus; /** 0..100 for measured progress, or `undefined` for indeterminate (in-progress, percentage unknown). */ progress: number | undefined; diff --git a/packages/task-graph/src/task/Task.ts b/packages/task-graph/src/task/Task.ts index a4b55839d..60ba195bd 100644 --- a/packages/task-graph/src/task/Task.ts +++ b/packages/task-graph/src/task/Task.ts @@ -31,6 +31,13 @@ import { TaskRunner } from "./TaskRunner"; import type { TaskConfig, TaskIdType, TaskInput, TaskOutput, TaskTypeName } from "./TaskTypes"; import { TaskConfigSchema, TaskStatus } from "./TaskTypes"; +/** + * Matches a canonical lowercase RFC 4122 v4 UUID. Used to detect the + * autogenerated default id produced when no `config.id` was supplied so the + * cache layer can fall back to a deterministic per-type key. + */ +const UUID_V4_REGEX = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; + /** * Base class for all tasks that implements the ITask interface. * This abstract class provides common functionality for both simple and compound tasks. @@ -393,6 +400,20 @@ export class Task< return this.config.id; } + /** + * Whether this task instance has a caller-pinned (deterministic) id. The + * default constructor mints a fresh v4 UUID when no id is supplied, so a + * v4-UUID-shaped id is treated as autogenerated — not deterministic — and + * the run-private cache will fall back to keying by task type to preserve + * a best-effort crash-resume across restarts. Override (or pin `config.id` + * to a stable string) to enable exact row reuse across restarts. + */ + public hasDeterministicId(): boolean { + const id = this.config.id; + if (typeof id !== "string" || id.length === 0) return false; + return !UUID_V4_REGEX.test(id); + } + /** * Runtime configuration (not serialized with the task). * Set via the constructor's third argument or mutated by the graph runner. diff --git a/packages/test/src/test/storage-tabular/SharedInMemoryTabularStorageSyncBarrier.test.ts b/packages/test/src/test/storage-tabular/SharedInMemoryTabularStorageSyncBarrier.test.ts new file mode 100644 index 000000000..3c0b81df8 --- /dev/null +++ b/packages/test/src/test/storage-tabular/SharedInMemoryTabularStorageSyncBarrier.test.ts @@ -0,0 +1,102 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import { SharedInMemoryTabularStorage } from "@workglow/storage"; +import type { DataPortSchemaObject } from "@workglow/util/schema"; +import { describe, expect, it } from "vitest"; + +const ItemSchema = { + type: "object", + properties: { + id: { type: "string" }, + value: { type: "string" }, + }, + required: ["id", "value"], + additionalProperties: false, +} as const satisfies DataPortSchemaObject; + +const ItemPK = ["id"] as const; + +/** + * Builds a fresh, isolated channel name so parallel test runs cannot collide. + */ +function uniqueChannel(label: string): string { + return `sync_barrier_${label}_${Date.now()}_${Math.random().toString(36).slice(2)}`; +} + +describe("SharedInMemoryTabularStorage.setupDatabase awaits tab-sync by default", () => { + it("first tab's first get sees rows another tab wrote before setupDatabase() returns", async () => { + const channelName = uniqueChannel("default_await"); + + // Simulate a peer tab that already has rows. `peer` constructed first so it + // owns the canonical row set before the first tab calls setupDatabase(). + const peer = new SharedInMemoryTabularStorage( + channelName, + ItemSchema, + ItemPK + ); + await peer.setupDatabase(); + await peer.put({ id: "preexisting", value: "from-peer" }); + + // Fresh tab joins the channel after the peer has populated rows. + const fresh = new SharedInMemoryTabularStorage( + channelName, + ItemSchema, + ItemPK + ); + // Default is awaitTabSync: true — should not resolve until SYNC_RESPONSE + // has been applied to the inner store. + await fresh.setupDatabase(); + + const row = await fresh.get({ id: "preexisting" } as any); + expect(row).toBeDefined(); + expect(row).toMatchObject({ id: "preexisting", value: "from-peer" }); + + fresh.destroy(); + peer.destroy(); + }); + + it("explicit awaitTabSync: false matches the prior fire-and-forget behavior (opt-out)", async () => { + const channelName = uniqueChannel("opt_out"); + + const fresh = new SharedInMemoryTabularStorage( + channelName, + ItemSchema, + ItemPK + ); + // No peer is on the channel, so there is nothing to wait for; the opt-out + // path must still resolve promptly and the storage must still work. + await fresh.setupDatabase({ awaitTabSync: false }); + + await fresh.put({ id: "self", value: "from-self" }); + const row = await fresh.get({ id: "self" } as any); + expect(row).toMatchObject({ id: "self", value: "from-self" }); + + fresh.destroy(); + }); + + it("setupDatabase() resolves even when no peer responds (timeout path)", async () => { + const channelName = uniqueChannel("no_peer"); + + const solo = new SharedInMemoryTabularStorage( + channelName, + ItemSchema, + ItemPK + ); + + // With no peer broadcaster on the channel SYNC_RESPONSE never arrives; + // setupDatabase must still resolve via the SYNC_TIMEOUT fallback rather + // than hang. Bounded by a generous outer timeout in this test. + await Promise.race([ + solo.setupDatabase(), + new Promise((_, reject) => + setTimeout(() => reject(new Error("setupDatabase never resolved")), 5000) + ), + ]); + + solo.destroy(); + }); +}); diff --git a/packages/test/src/test/task-graph-cache/RunPrivateCacheKeyFallback.test.ts b/packages/test/src/test/task-graph-cache/RunPrivateCacheKeyFallback.test.ts new file mode 100644 index 000000000..30f181091 --- /dev/null +++ b/packages/test/src/test/task-graph-cache/RunPrivateCacheKeyFallback.test.ts @@ -0,0 +1,180 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + CACHE_REGISTRY, + DefaultCacheRegistry, + RunPrivateCacheRepo, + Task, + TaskGraph, + type CachePolicy, +} from "@workglow/task-graph"; +import { Container, ServiceRegistry } from "@workglow/util"; +import type { DataPortSchema } from "@workglow/util/schema"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { InMemoryTaskOutputRepository } from "../../binding/InMemoryTaskOutputRepository"; + +class CountingPrivateTask extends Task<{ q: string }, { r: string }> { + public static override type = "CountingPrivateTask"; + public static override cachePolicy: CachePolicy = { kind: "private" }; + public static runs: { q: string }[] = []; + + public static override inputSchema(): DataPortSchema { + return { + type: "object", + properties: { + q: { type: "string" }, + }, + required: ["q"], + additionalProperties: false, + } as const satisfies DataPortSchema; + } + + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { + r: { type: "string" }, + }, + required: ["r"], + additionalProperties: false, + } as const satisfies DataPortSchema; + } + + public override async execute(input: { q: string }): Promise<{ r: string }> { + CountingPrivateTask.runs.push({ q: input.q }); + return { r: `done:${input.q}` }; + } +} + +async function freshRepo(): Promise { + const r = new InMemoryTaskOutputRepository(); + await (r as any).setupDatabase?.(); + return r; +} + +function freshServices( + privateRepo: InMemoryTaskOutputRepository | RunPrivateCacheRepo +): ServiceRegistry { + const services = new ServiceRegistry(new Container()); + services.registerInstance( + CACHE_REGISTRY, + new DefaultCacheRegistry({ private: privateRepo as any }) + ); + return services; +} + +/** + * The original autogenerated-uuid bug: two consecutive runs of the same graph + * (modeling a crash + restart) mint fresh task ids on every construction, so a + * private cache keyed strictly on `task.id` could never reuse prior rows. With + * the fallback in place the cache keys on `task.type` and the second run hits. + */ +describe("RunPrivateCacheRepo cache-key fallback for autogenerated task ids", () => { + let warnSpy: ReturnType; + + beforeEach(() => { + CountingPrivateTask.runs = []; + // Reset the warn-once gate so each test sees fresh behavior. + (RunPrivateCacheRepo as any).fallbackWarned = false; + warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + }); + + afterEach(() => { + warnSpy.mockRestore(); + }); + + it("Task.hasDeterministicId(): true for pinned ids, false for autogen v4 UUIDs", () => { + const pinned = new CountingPrivateTask({ + id: "pinned-node-1", + defaults: { q: "hello" }, + } as any); + expect(pinned.hasDeterministicId()).toBe(true); + + const autogen = new CountingPrivateTask({ defaults: { q: "hello" } } as any); + expect(autogen.hasDeterministicId()).toBe(false); + }); + + it("crash-restart: a fresh graph with autogenerated ids reuses prior run-private rows via the type-key fallback", async () => { + // Models a real crash: a prior partial run left rows in the run-private + // namespace. The restart constructs new task instances (so new uuid ids) + // and expects to find those rows. Without the fallback, the new uuids + // never match. With the fallback, both rows are keyed by task type and + // the restart hits cache for both ports. + const backing = await freshRepo(); + const runId = "crash-restart-run"; + + // Seed: write what the prior run would have left under the type-keyed + // fallback. `__cv` is the cache version sentinel injected by CacheCoordinator. + const seeder = new RunPrivateCacheRepo({ backing, runId }); + await seeder.saveOutput( + CountingPrivateTask.type, + { q: "alpha", __cv: "1" }, + { r: "done:alpha" } + ); + + const services = freshServices(backing); + const graph = new TaskGraph(); + // Single task instance with an autogenerated id — same shape that the + // crashed run would have had on restart. + graph.addTask(new CountingPrivateTask({ defaults: { q: "alpha" } } as any)); + + await graph.run({}, { runId, registry: services }); + + // The cached row was reused — execute() was never called. + expect(CountingPrivateTask.runs.length).toBe(0); + }); + + it("pinned-id tasks still key by id (not type) and never trigger the fallback warn", async () => { + const backing = await freshRepo(); + const runId = "pinned-run"; + + // Seed the row under the pinned task-id key; the runtime should find it + // there and skip execution. (The type-key path would NOT have this row.) + const seeder = new RunPrivateCacheRepo({ backing, runId }); + await seeder.saveOutput("pinned-node", { q: "stable", __cv: "1" }, { r: "cached" }); + + const services = freshServices(backing); + const graph = new TaskGraph(); + graph.addTask(new CountingPrivateTask({ id: "pinned-node", defaults: { q: "stable" } } as any)); + + await graph.run({}, { runId, registry: services }); + + // Pinned id matched the seeded row exactly — no re-execution. + expect(CountingPrivateTask.runs.length).toBe(0); + // No fallback warn from RunPrivateCacheRepo should appear in console.warn. + const fallbackWarns = warnSpy.mock.calls.filter((args: unknown[]) => + String(args[0] ?? "").includes("[RunPrivateCacheRepo]") + ); + expect(fallbackWarns.length).toBe(0); + }); + + it("warns exactly once per process across many fallback writes", async () => { + const backing = await freshRepo(); + const services = freshServices(backing); + + const graph = new TaskGraph(); + // Three autogen-id tasks, none pinned — all should trigger the fallback path + // on first execution (the cache-save path). + graph.addTask(new CountingPrivateTask({ defaults: { q: "a" } } as any)); + graph.addTask(new CountingPrivateTask({ defaults: { q: "b" } } as any)); + graph.addTask(new CountingPrivateTask({ defaults: { q: "c" } } as any)); + + await graph.run({}, { runId: "warn-once-run-1", registry: services }); + + // A second graph + run keeps producing fallbacks — warn must still be once total. + const graph2 = new TaskGraph(); + graph2.addTask(new CountingPrivateTask({ defaults: { q: "d" } } as any)); + await graph2.run({}, { runId: "warn-once-run-2", registry: services }); + + // Filter for our specific fallback warning to ignore unrelated runner warns. + const fallbackWarns = warnSpy.mock.calls.filter((args: unknown[]) => + String(args[0] ?? "").includes("[RunPrivateCacheRepo]") + ); + expect(fallbackWarns.length).toBe(1); + expect(String(fallbackWarns[0]?.[0])).toContain("pin task.id to enable crash-resume"); + }); +});