diff --git a/.changeset/run-status-long-poll.md b/.changeset/run-status-long-poll.md new file mode 100644 index 0000000000..a73768ca54 --- /dev/null +++ b/.changeset/run-status-long-poll.md @@ -0,0 +1,15 @@ +--- +'@workflow/core': minor +'@workflow/world': minor +'@workflow/world-local': minor +'@workflow/world-postgres': minor +'@workflow/world-vercel': minor +--- + +Resolve `await run.returnValue` as soon as a run finishes, via a new optional World long poll. + +`Storage['runs']` gains `waitForTerminalStatus(runId, { timeoutMs, signal, resolveData })`: one read the World holds open until the run reaches a terminal status, returning the same entity `runs.get()` returns (a budget that expires returns the latest snapshot, not an error). `Run#pollReturnValue` uses it when the World implements it, so a run that finishes mid-wait is reported immediately instead of at the next poll tick — previously up to a full `WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS` (1s) later. + +Implemented by `world-vercel` against workflow-server's new long-pollable `GET /v2/runs/:runId/status` route, by `world-postgres` with `LISTEN`/`NOTIFY` on run-terminal writes, and by `world-local` with an in-process signal over the run files. Every implementation re-reads the run before answering and backstops the wait with a periodic re-read, so a lost notification costs latency rather than correctness. + +The method is optional and the fast path is strictly additive: a World that omits it (`world-sim`, third-party adapters) keeps interval-polling `runs.get()` exactly as before, and `world-vercel` falls back to the plain read when the workflow-server it is talking to has no such route. `WORKFLOW_RETURN_VALUE_LONG_POLL=0` restores interval polling everywhere; `WORKFLOW_RETURN_VALUE_WAIT_MS` tunes the per-call budget (default 25s). diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx index 7dbf1957cb..ac2c6ab5da 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx @@ -151,6 +151,40 @@ const run = await world.runs.get(runId); // [!code highlight] **Returns:** `WorkflowRun` (or `WorkflowRunWithoutData` when `resolveData: 'none'`) +### runs.waitForTerminalStatus() + +Optional. Long poll for a run to reach a terminal status (`completed`, +`failed`, or `cancelled`) instead of re-reading it on an interval. This is what +`await run.returnValue` uses, so a run's result reaches the awaiting side as +soon as it finishes rather than at the next poll tick. + +```typescript lineNumbers +const run = await world.runs.waitForTerminalStatus?.(runId, { // [!code highlight] + timeoutMs: 25_000, // [!code highlight] +}); // [!code highlight] +``` + +| Parameter | Type | Description | +|-----------|------|-------------| +| `runId` | `string` | The workflow run ID | +| `params.timeoutMs` | `number` | Upper bound on the wait — the call returns earlier, the moment the run is terminal | +| `params.signal` | `AbortSignal` | Abandons the wait | +| `params.resolveData` | `'all' \| 'none'` | Whether to include input/output data. Default: `'all'` | + +**Returns:** the same `WorkflowRun` as `runs.get()` — terminal if the run +finished within the budget, otherwise the latest snapshot. An expired budget is +a normal return, not an error, and a missing run throws +`WorkflowRunNotFoundError` exactly as `runs.get()` does. + + + Not every backend can hold a read open, so this method is optional and may + also return a non-terminal snapshot before `timeoutMs` is up. Callers pace + their own retries — `await run.returnValue` keeps consecutive non-terminal + observations at least one `WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS` apart — and + worlds that omit the method are polled on that interval instead. Set + `WORKFLOW_RETURN_VALUE_LONG_POLL=0` to force interval polling everywhere. + + ### runs.list() ```typescript lineNumbers diff --git a/docs/content/worlds/v5/building-a-world.mdx b/docs/content/worlds/v5/building-a-world.mdx index 0d0bfef497..8660fd204c 100644 --- a/docs/content/worlds/v5/building-a-world.mdx +++ b/docs/content/worlds/v5/building-a-world.mdx @@ -88,6 +88,9 @@ interface Storage { runs: { get(id: string, params?: GetWorkflowRunParams): Promise; list(params?: ListWorkflowRunsParams): Promise>; + + // Optional: long poll for a terminal status (see below) + waitForTerminalStatus?(id: string, params?: WaitForTerminalRunStatusParams): Promise; }; steps: { @@ -129,6 +132,21 @@ Keep the owning Run available for at least as long as its token remains unavaila **Automatic Hook Cleanup:** When a run ends, remove its live Hooks. Make each token available unless its `tokenRetentionUntil` is still in the future. A `hook_disposed` event always makes the token available immediately. +### Optional: Waiting for a Terminal Run Status + +`await run.returnValue` has to find out when a run finished. Without help it re-reads the run every second, so a run that finishes just after a read is reported up to a second late. Implement `runs.waitForTerminalStatus(id, { timeoutMs, signal, resolveData })` and the runtime asks once and is answered the moment the run ends. + +The contract is deliberately forgiving, because "wait" means something different in every store: + +- Resolve as soon as the run's status is terminal, returning the same entity `get` returns. +- Resolve no later than roughly `timeoutMs` with the latest snapshot, whatever its status. **A timeout is a normal return, never an error** — a run that is still running is a legitimate answer, and the runtime simply asks again. +- `timeoutMs` is an upper bound, not a lower one: returning a non-terminal snapshot early is allowed, and the runtime paces its own retries. +- Fail exactly like `get` — a missing run throws `WorkflowRunNotFoundError`. + +How you wait is up to your store. The reference worlds use, respectively, a server-side long poll (`world-vercel` holds `GET /v2/runs/:runId/status` open), `LISTEN`/`NOTIFY` (`world-postgres`), and an in-process emitter over the run files (`world-local`). Whatever the mechanism, treat the notification as a *signal only* and re-read the run before answering, and back the wait with a periodic re-read so a lost notification costs latency rather than hanging until the budget expires. + +Omitting the method is a supported choice — a store with no change notification, or a deterministic simulator like `world-sim` where a real wait would stall a virtual clock, simply leaves it off and the runtime keeps interval-polling `get`. Nothing else degrades; there is no capability to declare. + ### Event ID Allocation diff --git a/packages/core/src/runtime/run-return-value-long-poll.test.ts b/packages/core/src/runtime/run-return-value-long-poll.test.ts new file mode 100644 index 0000000000..7a349e36fa --- /dev/null +++ b/packages/core/src/runtime/run-return-value-long-poll.test.ts @@ -0,0 +1,293 @@ +import { WorkflowRunCancelledError } from '@workflow/errors'; +import { SPEC_VERSION_CURRENT, type World } from '@workflow/world'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +// Mock version module to avoid missing generated file +vi.mock('../version.js', () => ({ version: '0.0.0-test' })); + +import { dehydrateWorkflowReturnValue } from '../serialization.js'; +import { + getReturnValueWaitTimeoutMs, + isReturnValueLongPollEnabled, + Run, +} from './run.js'; +import { setWorld } from './world.js'; + +/** + * `await run.returnValue` and the World's optional long poll + * (`runs.waitForTerminalStatus`). + * + * The behavior being pinned down here is the *pacing*, not the values: which + * read the runtime issues, and how long it waits between two non-terminal + * observations. All of it runs on fake timers so the assertions are about the + * scheduling itself rather than wall-clock luck. + */ + +const RUN_ID = 'wrun_01JB0000000000000000000000'; + +const baseRun = { + runId: RUN_ID, + workflowName: 'test-workflow', + specVersion: 2, + input: [], + createdAt: new Date(0), + updatedAt: new Date(0), + startedAt: new Date(0), + deploymentId: 'test-deployment', +}; + +const runningRun = { ...baseRun, status: 'running' as const }; +const cancelledRun = { + ...baseRun, + status: 'cancelled' as const, + completedAt: new Date(0), +}; + +function createWorld(runs: Partial): World { + return { + specVersion: SPEC_VERSION_CURRENT, + runs: { + get: vi.fn().mockResolvedValue(runningRun), + ...runs, + }, + events: { + list: vi + .fn() + .mockResolvedValue({ data: [], hasMore: false, cursor: null }), + create: vi.fn(), + }, + queue: vi.fn().mockResolvedValue(undefined), + } as unknown as World; +} + +/** Resolve after `ms` on the *fake* clock. */ +const sleepFake = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + +describe('Run.returnValue long poll', () => { + const envNames = [ + 'WORKFLOW_RETURN_VALUE_LONG_POLL', + 'WORKFLOW_RETURN_VALUE_WAIT_MS', + 'WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS', + ] as const; + const originalEnv = new Map( + envNames.map((name) => [name, process.env[name]]) + ); + + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + for (const [name, value] of originalEnv) { + if (value === undefined) delete process.env[name]; + else process.env[name] = value; + } + setWorld(undefined as unknown as World); + }); + + it('waits on the World long poll instead of polling runs.get', async () => { + const waitForTerminalStatus = vi.fn().mockResolvedValue(cancelledRun); + const world = createWorld({ waitForTerminalStatus }); + setWorld(world); + + await expect(new Run(RUN_ID).returnValue).rejects.toBeInstanceOf( + WorkflowRunCancelledError + ); + + expect(waitForTerminalStatus).toHaveBeenCalledTimes(1); + expect(waitForTerminalStatus).toHaveBeenCalledWith(RUN_ID, { + timeoutMs: getReturnValueWaitTimeoutMs(), + }); + // The status read went through the long poll — no interval poll happened. + expect(world.runs.get).not.toHaveBeenCalled(); + }); + + it('forwards the configured wait budget', async () => { + process.env.WORKFLOW_RETURN_VALUE_WAIT_MS = '3000'; + const waitForTerminalStatus = vi.fn().mockResolvedValue(cancelledRun); + setWorld(createWorld({ waitForTerminalStatus })); + + await expect(new Run(RUN_ID).returnValue).rejects.toBeInstanceOf( + WorkflowRunCancelledError + ); + + expect(waitForTerminalStatus).toHaveBeenCalledWith(RUN_ID, { + timeoutMs: 3_000, + }); + }); + + it('resolves the hydrated return value from a completed long poll', async () => { + vi.useRealTimers(); + const output = await dehydrateWorkflowReturnValue( + { ok: true }, + RUN_ID, + undefined + ); + const completedRun = { + ...baseRun, + status: 'completed' as const, + completedAt: new Date(0), + output, + }; + const waitForTerminalStatus = vi.fn().mockResolvedValue(completedRun); + setWorld( + createWorld({ + waitForTerminalStatus, + get: vi.fn().mockResolvedValue(completedRun), + }) + ); + + await expect(new Run(RUN_ID).returnValue).resolves.toEqual({ ok: true }); + }); + + it('paces a World whose wait returns a non-terminal run early', async () => { + // A World that cannot actually hold the wait open (e.g. world-vercel + // talking to a server without the long-poll route) answers immediately + // with a non-terminal run. The loop must fall back to interval polling + // rather than spinning on it. + process.env.WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS = '1000'; + const waitForTerminalStatus = vi + .fn() + .mockResolvedValueOnce(runningRun) + .mockResolvedValue(cancelledRun); + setWorld(createWorld({ waitForTerminalStatus })); + + const pending = new Run(RUN_ID).returnValue; + const assertion = expect(pending).rejects.toBeInstanceOf( + WorkflowRunCancelledError + ); + + await vi.advanceTimersByTimeAsync(0); + expect(waitForTerminalStatus).toHaveBeenCalledTimes(1); + + // Still inside the interval — no second attempt yet. + await vi.advanceTimersByTimeAsync(999); + expect(waitForTerminalStatus).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(1); + expect(waitForTerminalStatus).toHaveBeenCalledTimes(2); + + await assertion; + }); + + it('does not add an interval sleep when the wait already outlasted it', async () => { + // The whole point of the long poll: a wait that blocked for longer than + // the poll interval re-issues immediately instead of sleeping again. + process.env.WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS = '1000'; + const waitForTerminalStatus = vi + .fn() + .mockImplementationOnce(async () => { + await sleepFake(1_500); + return runningRun; + }) + .mockResolvedValue(cancelledRun); + setWorld(createWorld({ waitForTerminalStatus })); + + const pending = new Run(RUN_ID).returnValue; + const assertion = expect(pending).rejects.toBeInstanceOf( + WorkflowRunCancelledError + ); + + await vi.advanceTimersByTimeAsync(1_500); + expect(waitForTerminalStatus).toHaveBeenCalledTimes(2); + + await assertion; + }); + + it('interval-polls runs.get when the World has no long poll', async () => { + process.env.WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS = '1000'; + const get = vi + .fn() + .mockResolvedValueOnce(runningRun) + .mockResolvedValue(cancelledRun); + const world = createWorld({ get }); + setWorld(world); + + const pending = new Run(RUN_ID).returnValue; + const assertion = expect(pending).rejects.toBeInstanceOf( + WorkflowRunCancelledError + ); + + await vi.advanceTimersByTimeAsync(0); + expect(get).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1_000); + expect(get).toHaveBeenCalledTimes(2); + + await assertion; + }); + + it('kill switch restores the fixed-interval poll', async () => { + process.env.WORKFLOW_RETURN_VALUE_LONG_POLL = '0'; + process.env.WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS = '1000'; + const waitForTerminalStatus = vi.fn().mockResolvedValue(cancelledRun); + const get = vi + .fn() + .mockResolvedValueOnce(runningRun) + .mockResolvedValue(cancelledRun); + setWorld(createWorld({ get, waitForTerminalStatus })); + + const pending = new Run(RUN_ID).returnValue; + const assertion = expect(pending).rejects.toBeInstanceOf( + WorkflowRunCancelledError + ); + + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(1_000); + + await assertion; + expect(waitForTerminalStatus).not.toHaveBeenCalled(); + expect(get).toHaveBeenCalledTimes(2); + }); +}); + +describe('isReturnValueLongPollEnabled', () => { + const envName = 'WORKFLOW_RETURN_VALUE_LONG_POLL'; + const originalValue = process.env[envName]; + + afterEach(() => { + if (originalValue === undefined) delete process.env[envName]; + else process.env[envName] = originalValue; + }); + + it('defaults to on', () => { + delete process.env[envName]; + expect(isReturnValueLongPollEnabled()).toBe(true); + }); + + it('treats an empty value as unset', () => { + process.env[envName] = ''; + expect(isReturnValueLongPollEnabled()).toBe(true); + }); + + it.each(['0', 'false', 'FALSE'])('is disabled by %s', (value) => { + process.env[envName] = value; + expect(isReturnValueLongPollEnabled()).toBe(false); + }); + + it.each(['1', 'true'])('stays enabled for %s', (value) => { + process.env[envName] = value; + expect(isReturnValueLongPollEnabled()).toBe(true); + }); +}); + +describe('getReturnValueWaitTimeoutMs', () => { + const envName = 'WORKFLOW_RETURN_VALUE_WAIT_MS'; + const originalValue = process.env[envName]; + + afterEach(() => { + if (originalValue === undefined) delete process.env[envName]; + else process.env[envName] = originalValue; + }); + + it('defaults to a budget under the adapter request timeout', () => { + delete process.env[envName]; + expect(getReturnValueWaitTimeoutMs()).toBe(25_000); + }); + + it('accepts a runtime override', () => { + process.env[envName] = '5000'; + expect(getReturnValueWaitTimeoutMs()).toBe(5_000); + }); +}); diff --git a/packages/core/src/runtime/run.ts b/packages/core/src/runtime/run.ts index 1663df9b8d..86439273cf 100644 --- a/packages/core/src/runtime/run.ts +++ b/packages/core/src/runtime/run.ts @@ -40,6 +40,46 @@ export function getReturnValuePollIntervalMs(): number { ); } +/** + * How long a single `runs.waitForTerminalStatus` call may block while waiting + * for a run to finish. The wait is re-issued until the run is terminal, so + * this is a per-call budget rather than a limit on total wait time — its only + * job is to bound one request so a stalled connection cannot hold the awaiting + * side forever. + * + * 25s keeps a Vercel long poll comfortably inside `world-vercel`'s 60s + * per-request HTTP timeout, so the wait budget is always observed as a + * *response* (a non-terminal snapshot) rather than as a client-side timeout. + */ +const RETURN_VALUE_WAIT_TIMEOUT_MS = 25_000; + +/** @internal */ +export function getReturnValueWaitTimeoutMs(): number { + return envNumber( + 'WORKFLOW_RETURN_VALUE_WAIT_MS', + RETURN_VALUE_WAIT_TIMEOUT_MS, + { integer: true, min: 1 } + ); +} + +/** + * Whether `await run.returnValue` may use the World's long poll + * (`runs.waitForTerminalStatus`) instead of interval-polling `runs.get`. + * + * Default **ON** wherever the World implements the method — a World that does + * not is already on the interval path with nothing to switch off. Reads + * `process.env.WORKFLOW_RETURN_VALUE_LONG_POLL` lazily; an explicit `'0'` / + * `'false'` is the kill switch, restoring the fixed-interval poll exactly as + * it behaved before the fast path existed. + * + * @internal + */ +export function isReturnValueLongPollEnabled(): boolean { + const raw = process.env.WORKFLOW_RETURN_VALUE_LONG_POLL; + if (raw === undefined || raw === '') return true; + return !(raw === '0' || raw.toLowerCase() === 'false'); +} + /** * A `ReadableStream` extended with workflow-specific helpers. */ @@ -335,6 +375,17 @@ export class Run { const NOT_FOUND_MAX_RETRIES = this.#resilientStart ? 3 : 0; const NOT_FOUND_DELAYS = [1_000, 3_000, 6_000]; + // Prefer the World's long poll: one read that the backend holds open + // until the run finishes, instead of asking again every second and + // paying up to a full interval of quantization on a run that already + // ended. Worlds that cannot wait simply do not implement it (see + // `Storage['runs'].waitForTerminalStatus`) and this stays the exact + // fixed-interval poll it has always been — as does an operator who + // throws the `WORKFLOW_RETURN_VALUE_LONG_POLL=0` kill switch. + const waitForTerminalStatus = isReturnValueLongPollEnabled() + ? world.runs.waitForTerminalStatus?.bind(world.runs) + : undefined; + // NOTE: when this poll runs inside a step (e.g. the step that a parent // workflow uses to await a child workflow's `returnValue`), it blocks // a queue worker slot for as long as the child run takes to finish. @@ -343,8 +394,13 @@ export class Run { // default on the Postgres world and the notes in the eager-processing // changelog for details. while (true) { + const iterationStartedAt = Date.now(); try { - const run = await world.runs.get(this.runId); + const run = waitForTerminalStatus + ? await waitForTerminalStatus(this.runId, { + timeoutMs: getReturnValueWaitTimeoutMs(), + }) + : await world.runs.get(this.runId); if (run.status === 'completed') { const encryptionKey = await this.#getEncryptionKey(); @@ -386,9 +442,19 @@ export class Run { throw new WorkflowRunNotCompletedError(this.runId, run.status); } catch (error) { if (WorkflowRunNotCompletedError.is(error)) { - await new Promise((resolve) => - setTimeout(resolve, getReturnValuePollIntervalMs()) - ); + // Space consecutive non-terminal observations at least one poll + // interval apart. On the plain-poll path that is the familiar fixed + // sleep; on the long-poll path the wait has usually already + // outlasted the interval and this is a no-op — but it also means a + // World whose wait returns early (a backend with no long poll, a + // clamped budget) degrades to interval polling instead of spinning. + const remainingIntervalMs = + getReturnValuePollIntervalMs() - (Date.now() - iterationStartedAt); + if (remainingIntervalMs > 0) { + await new Promise((resolve) => + setTimeout(resolve, remainingIntervalMs) + ); + } continue; } if ( diff --git a/packages/docs-typecheck/src/docs-globals.d.ts b/packages/docs-typecheck/src/docs-globals.d.ts index ba5d020ade..547f71af6a 100644 --- a/packages/docs-typecheck/src/docs-globals.d.ts +++ b/packages/docs-typecheck/src/docs-globals.d.ts @@ -186,6 +186,8 @@ declare global { runs: { get: (...args: any[]) => Promise; list: (...args: any[]) => Promise; + // Optional on World, so reference snippets call it through `?.` + waitForTerminalStatus?: (...args: any[]) => Promise; }; steps: { get: (...args: any[]) => Promise; diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index d9371dcab4..3335fef1ab 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -104,6 +104,7 @@ import { rebuildLiveHookByTokenFromEventLog, } from './hooks-storage.js'; import { handleLegacyEvent } from './legacy.js'; +import { signalRunTerminal } from './run-status-signal.js'; import { withRunFileLock } from './runs-storage.js'; const DAY_MS = 24 * 60 * 60 * 1000; @@ -561,6 +562,12 @@ async function writeRunUnderLifecycleLock( await writeJSON(taggedPath(basedir, 'runs', runId, tag), next, { overwrite: true, }); + // Wake `runs.waitForTerminalStatus` waiters in this process. Emitted after + // the file is on disk so a woken waiter re-reads a terminal run, and from + // here because every run-lifecycle write funnels through this helper. + if (isTerminalWorkflowRunStatus(next.status)) { + signalRunTerminal(runId); + } return next; }); } diff --git a/packages/world-local/src/storage/run-status-signal.ts b/packages/world-local/src/storage/run-status-signal.ts new file mode 100644 index 0000000000..8a80bf6291 --- /dev/null +++ b/packages/world-local/src/storage/run-status-signal.ts @@ -0,0 +1,89 @@ +import { EventEmitter } from 'node:events'; +import { envNumber } from '@workflow/world'; + +/** + * In-process wakeups for `runs.waitForTerminalStatus`. + * + * world-local's store is the filesystem, which has no change notification a + * reader can subscribe to. So the wait is built from two halves: + * + * - **The emitter below**, signalled by the run-lifecycle writer + * (`writeRunUnderLifecycleLock` in `events-storage.ts`) whenever it commits + * a terminal run. In the ordinary local-dev topology the workflow and the + * caller awaiting its result live in the same process, so this is the path + * that actually fires, and it fires within a tick of the run finishing. + * + * - **A short backstop poll** of the run file (see + * {@link getRunStatusPollIntervalMs}), which covers everything the emitter + * cannot see: a second process (a `wf dev` server plus a separate CLI + * invocation, or several workers over one data dir), and the narrow window + * between a waiter's read and its subscribe. + * + * Neither half is trusted for the status itself — the waiter always re-reads + * the run file, so a missed or duplicated signal only ever costs latency. + */ + +/** Signals are edge-only: the run id is the whole message. */ +const emitter = new EventEmitter<{ [key: `run:${string}`]: [] }>(); +// A dev server can have many runs awaited at once; the default cap of 10 +// listeners per event would warn on legitimate fan-out. +emitter.setMaxListeners(0); + +const RUN_STATUS_POLL_INTERVAL_MS = 100; + +/** + * Backstop interval for the terminal-status wait, in ms. Override with + * `WORKFLOW_LOCAL_RUN_STATUS_POLL_INTERVAL_MS`. + * + * Deliberately far below the SDK's own 1s `runs.get` poll: re-reading one + * small JSON file is cheap, and this is the ceiling on how long a *cross + * process* local run takes to be noticed. + */ +export function getRunStatusPollIntervalMs(): number { + return envNumber( + 'WORKFLOW_LOCAL_RUN_STATUS_POLL_INTERVAL_MS', + RUN_STATUS_POLL_INTERVAL_MS, + { integer: true, min: 1 } + ); +} + +/** + * Announce that a run reached a terminal status in this process. Call it after + * the run file has been written, so a woken waiter re-reads a terminal run. + */ +export function signalRunTerminal(runId: string): void { + emitter.emit(`run:${runId}`); +} + +/** + * Wait for the next in-process terminal signal for `runId`, the timeout, or an + * abort — whichever comes first. Resolves either way; the caller decides what + * to do by re-reading the run. + */ +export function waitForRunTerminalSignal( + runId: string, + timeoutMs: number, + signal?: AbortSignal +): Promise { + if (timeoutMs <= 0 || signal?.aborted) return Promise.resolve(); + + return new Promise((resolve) => { + const key = `run:${runId}` as const; + let settled = false; + const settle = () => { + if (settled) return; + settled = true; + clearTimeout(timer); + emitter.off(key, settle); + signal?.removeEventListener('abort', settle); + resolve(); + }; + + const timer = setTimeout(settle, timeoutMs); + // Never keep a process alive just to observe a run it stopped caring + // about (e.g. a CLI command that finished while a wait was in flight). + timer.unref?.(); + emitter.once(key, settle); + signal?.addEventListener('abort', settle, { once: true }); + }); +} diff --git a/packages/world-local/src/storage/run-status-wait.test.ts b/packages/world-local/src/storage/run-status-wait.test.ts new file mode 100644 index 0000000000..49e61afc3a --- /dev/null +++ b/packages/world-local/src/storage/run-status-wait.test.ts @@ -0,0 +1,143 @@ +import { promises as fs } from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { WorkflowRunNotFoundError } from '@workflow/errors'; +import type { Storage } from '@workflow/world'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { createStorage } from '../storage.js'; +import { createRun, updateRun } from '../test-helpers.js'; + +/** + * `runs.waitForTerminalStatus` on world-local: the in-process terminal signal + * plus its filesystem backstop poll (see `run-status-signal.ts`). + */ +describe('runs.waitForTerminalStatus (world-local)', () => { + let testDir: string; + let storage: Storage; + + beforeEach(async () => { + testDir = await fs.mkdtemp(path.join(os.tmpdir(), 'run-status-wait-')); + storage = createStorage(testDir); + }); + + afterEach(async () => { + await fs.rm(testDir, { recursive: true, force: true }); + }); + + const newRun = () => + createRun(storage, { + deploymentId: 'dpl_test', + workflowName: 'test-workflow', + input: new Uint8Array([1]), + }); + + const waitForTerminalStatus = () => { + const wait = storage.runs.waitForTerminalStatus; + if (!wait) throw new Error('world-local should implement the long poll'); + return wait; + }; + + /** + * Run `fn` after `ms`, returning a promise for its completion — so a test + * can await the write it triggered and not race the temp-dir cleanup. + */ + const delayed = (ms: number, fn: () => Promise): Promise => + new Promise((resolve, reject) => { + setTimeout(() => fn().then(resolve, reject), ms); + }); + + it('returns an already-terminal run without waiting', async () => { + const run = await newRun(); + await updateRun(storage, run.runId, 'run_started'); + await updateRun(storage, run.runId, 'run_completed', { + output: new Uint8Array([2]), + }); + + const startedAt = Date.now(); + const waited = await waitForTerminalStatus()(run.runId, { + timeoutMs: 30_000, + }); + + expect(waited.status).toBe('completed'); + expect(Date.now() - startedAt).toBeLessThan(1_000); + }); + + it('resolves as soon as the run completes', async () => { + const run = await newRun(); + await updateRun(storage, run.runId, 'run_started'); + + const pending = waitForTerminalStatus()(run.runId, { timeoutMs: 30_000 }); + // Finish the run while the wait is parked, the way a workflow finishing in + // another part of the same dev server would. + const finishing = delayed(20, () => + updateRun(storage, run.runId, 'run_completed', { + output: new Uint8Array([2]), + }) + ); + + const startedAt = Date.now(); + const waited = await pending; + await finishing; + + expect(waited.status).toBe('completed'); + // Nowhere near the 30s budget: the signal (or its 100ms backstop) wakes + // the wait, not the budget expiring. + expect(Date.now() - startedAt).toBeLessThan(5_000); + }); + + it('picks up a cancellation', async () => { + const run = await newRun(); + await updateRun(storage, run.runId, 'run_started'); + + const pending = waitForTerminalStatus()(run.runId, { timeoutMs: 30_000 }); + const cancelling = delayed(20, () => + updateRun(storage, run.runId, 'run_cancelled') + ); + + expect((await pending).status).toBe('cancelled'); + await cancelling; + }); + + it('returns the latest non-terminal snapshot when the budget expires', async () => { + const run = await newRun(); + await updateRun(storage, run.runId, 'run_started'); + + const startedAt = Date.now(); + const waited = await waitForTerminalStatus()(run.runId, { timeoutMs: 150 }); + + // A timeout is a normal return, not an error. + expect(waited.status).toBe('running'); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(140); + }); + + it('returns immediately without a budget', async () => { + const run = await newRun(); + await updateRun(storage, run.runId, 'run_started'); + + const startedAt = Date.now(); + expect((await waitForTerminalStatus()(run.runId)).status).toBe('running'); + expect(Date.now() - startedAt).toBeLessThan(1_000); + }); + + it('stops early when the caller aborts', async () => { + const run = await newRun(); + await updateRun(storage, run.runId, 'run_started'); + const controller = new AbortController(); + + const pending = waitForTerminalStatus()(run.runId, { + timeoutMs: 30_000, + signal: controller.signal, + }); + setTimeout(() => controller.abort(), 20); + + expect((await pending).status).toBe('running'); + }); + + it('fails like get for an unknown run', async () => { + await expect( + waitForTerminalStatus()('wrun_01JB0000000000000000000000', { + timeoutMs: 30_000, + }) + ).rejects.toBeInstanceOf(WorkflowRunNotFoundError); + }); +}); diff --git a/packages/world-local/src/storage/runs-storage.ts b/packages/world-local/src/storage/runs-storage.ts index e124818530..0448b352db 100644 --- a/packages/world-local/src/storage/runs-storage.ts +++ b/packages/world-local/src/storage/runs-storage.ts @@ -12,6 +12,7 @@ import type { import { AttributeValidationError, applyAttributeChanges, + isTerminalWorkflowRunStatus, validateAttributeChanges, WorkflowRunSchema, } from '@workflow/world'; @@ -25,6 +26,10 @@ import { } from '../fs.js'; import { filterRunData } from './filters.js'; import { getObjectCreatedAt } from './helpers.js'; +import { + getRunStatusPollIntervalMs, + waitForRunTerminalSignal, +} from './run-status-signal.js'; /** * Internal extension of `ListWorkflowRunsParams` that adds a `fileIdFilter` @@ -38,6 +43,7 @@ export interface LocalListWorkflowRunsParams extends ListWorkflowRunsParams { export interface LocalRunsStorage { get: Storage['runs']['get']; + waitForTerminalStatus: NonNullable; getMany: NonNullable; list: { ( @@ -121,6 +127,33 @@ export function createRunsStorage( return { get, + /** + * Long poll for a terminal run status — see + * `Storage['runs'].waitForTerminalStatus`. + * + * Reads the run, and while it is non-terminal waits for either an + * in-process terminal signal or the short backstop interval before + * reading again (see `run-status-signal.ts` for why both). Returns the + * latest snapshot once `timeoutMs` is up, whatever its status, and + * propagates `WorkflowRunNotFoundError` exactly as `get` does. + */ + waitForTerminalStatus: (async (id: string, params?: any) => { + const deadline = Date.now() + (params?.timeoutMs ?? 0); + while (true) { + const run = await get(id, params); + if (isTerminalWorkflowRunStatus(run.status)) return run; + + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0 || params?.signal?.aborted) return run; + + await waitForRunTerminalSignal( + id, + Math.min(remainingMs, getRunStatusPollIntervalMs()), + params?.signal + ); + } + }) as NonNullable, + getMany: (async (ids: readonly string[], params?: any) => { const uniqueIds = [...new Set(ids)]; const runs = await Promise.all( diff --git a/packages/world-postgres/src/index.ts b/packages/world-postgres/src/index.ts index 84618fdbc2..4310f62864 100644 --- a/packages/world-postgres/src/index.ts +++ b/packages/world-postgres/src/index.ts @@ -4,6 +4,10 @@ import { Pool } from 'pg'; import type { PostgresWorldConfig } from './config.js'; import { createClient, type Drizzle } from './drizzle/index.js'; import { createQueue } from './queue.js'; +import { + createRunStatusListener, + type RunStatusListener, +} from './run-status.js'; import { createEventsStorage, createHooksStorage, @@ -12,9 +16,12 @@ import { } from './storage.js'; import { createStreamer } from './streamer.js'; -function createStorage(drizzle: Drizzle): Storage { +function createStorage( + drizzle: Drizzle, + runStatusListener: RunStatusListener +): Storage { return { - runs: createRunsStorage(drizzle), + runs: createRunsStorage(drizzle, runStatusListener), events: createEventsStorage(drizzle), hooks: createHooksStorage(drizzle), steps: createStepsStorage(drizzle), @@ -59,7 +66,10 @@ export function createWorld( const drizzle = createClient(pool); const queue = createQueue(config, pool); - const storage = createStorage(drizzle); + // Opens its `LISTEN` connection lazily, on the first `waitForTerminalStatus` + // call, so a deployment that never awaits a run never pays for it. + const runStatusListener = createRunStatusListener(pool); + const storage = createStorage(drizzle, runStatusListener); const streamer = createStreamer(pool, drizzle); return { @@ -85,6 +95,7 @@ export function createWorld( async close() { await queue.close(); await streamer.close(); + await runStatusListener.close(); if (pool !== config.pool) { await pool.end(); } diff --git a/packages/world-postgres/src/run-status.ts b/packages/world-postgres/src/run-status.ts new file mode 100644 index 0000000000..733a909a53 --- /dev/null +++ b/packages/world-postgres/src/run-status.ts @@ -0,0 +1,164 @@ +import { EventEmitter } from 'node:events'; +import { envNumber } from '@workflow/world'; +import { sql } from 'drizzle-orm'; +import type { Pool } from 'pg'; +import type { Drizzle } from './drizzle/index.js'; +import { listenChannel } from './streamer.js'; + +/** + * Terminal-run-status wakeups behind `runs.waitForTerminalStatus`. + * + * A caller awaiting a run's outcome (`await run.returnValue`) asks the World + * to hold the read until the run finishes, instead of re-reading it every + * second and paying up to a full interval of quantization. Postgres already + * has the primitive for that: the run-terminal write issues a `NOTIFY` (see + * {@link notifyRunTerminal}) and the waiter is parked on a `LISTEN` for it — + * the same mechanism `createStreamer` uses for stream chunks. + * + * The notification is a *signal only*: waiters re-read the run row, so a + * duplicate or lost message can never produce a wrong answer. Because it can + * be lost — a `NOTIFY` that fires between a waiter's read and its `LISTEN`, a + * dropped listener connection — the wait is also backstopped by a periodic + * re-read ({@link getRunStatusPollIntervalMs}), which bounds the damage of a + * miss to one interval. + * + * One `LISTEN` connection is shared by every waiter in the process and is + * opened lazily, so a deployment that never awaits a run never pays for it. + */ + +/** `NOTIFY` channel carrying terminal run ids. */ +export const RUN_STATUS_TOPIC = 'workflow_run_status'; + +/** + * How long to wait before re-attempting the shared `LISTEN` connection after a + * failed attempt. Long enough that a database that cannot host the listener at + * all costs one connection attempt every few seconds rather than one per + * waiting run per poll interval, short enough that a restart is picked back up + * well within a single run's wait. + */ +const LISTEN_RETRY_BACKOFF_MS = 5_000; + +const RUN_STATUS_POLL_INTERVAL_MS = 1_000; + +/** + * Backstop re-read interval for a terminal-status wait, in ms. Override with + * `WORKFLOW_POSTGRES_RUN_STATUS_POLL_INTERVAL_MS`. + * + * The `NOTIFY` is what makes the wait fast; this only bounds how long a *lost* + * notification can go unnoticed, so it is kept at the interval the SDK would + * have polled at anyway — the wait is then never slower than the poll it + * replaces, and normally three orders of magnitude faster. + */ +export function getRunStatusPollIntervalMs(): number { + return envNumber( + 'WORKFLOW_POSTGRES_RUN_STATUS_POLL_INTERVAL_MS', + RUN_STATUS_POLL_INTERVAL_MS, + { integer: true, min: 1 } + ); +} + +/** + * Announce that a run reached a terminal status. + * + * Best-effort: a failed `NOTIFY` costs a waiter its backstop interval and + * nothing else, so it must never fail the write that produced the status. + * Call it after the terminal `UPDATE` has committed. + */ +export async function notifyRunTerminal( + drizzle: Drizzle, + runId: string +): Promise { + try { + await drizzle.execute(sql`SELECT pg_notify(${RUN_STATUS_TOPIC}, ${runId})`); + } catch { + // Intentionally ignored — see above. + } +} + +export interface RunStatusListener { + /** + * Resolve when `runId` is announced terminal, when `timeoutMs` elapses, or + * when `signal` aborts — whichever is first. The caller decides what + * happened by re-reading the run. + */ + wait(runId: string, timeoutMs: number, signal?: AbortSignal): Promise; + /** Release the shared `LISTEN` connection. Safe to call more than once. */ + close(): Promise; +} + +export function createRunStatusListener(pool: Pool): RunStatusListener { + const emitter = new EventEmitter<{ [key: `run:${string}`]: [] }>(); + // Many runs can be awaited at once; the 10-listener default would warn on + // legitimate fan-out. + emitter.setMaxListeners(0); + + let subscription: + | Promise<{ close: () => Promise } | undefined> + | undefined; + let retrySubscribeAfter = 0; + + const ensureSubscribed = () => { + if (subscription) return subscription; + // A failed LISTEN must be re-attemptable — a database restart or a brief + // network blip at process start would otherwise degrade every wait to + // backstop polling for the lifetime of the process. Bounded by a backoff + // so that a genuinely unavailable listener does not turn every waiting + // run into a connection attempt per poll interval. + if (Date.now() < retrySubscribeAfter) return undefined; + + subscription = listenChannel(pool, RUN_STATUS_TOPIC, async (payload) => { + if (payload) emitter.emit(`run:${payload}`); + }).catch(() => { + // No listener connection available (pool options that don't permit a + // second client, a database without LISTEN, a restarting server). Waits + // degrade to the backstop re-read — the behavior of a plain poll — and + // the next wait past the backoff tries again. + subscription = undefined; + retrySubscribeAfter = Date.now() + LISTEN_RETRY_BACKOFF_MS; + return undefined; + }); + return subscription; + }; + + return { + async wait(runId, timeoutMs, signal) { + if (timeoutMs <= 0 || signal?.aborted) return; + + const key = `run:${runId}` as const; + // Kick off (or reuse) the shared subscription without awaiting it, so + // the listener below is registered in this same tick — a notification + // delivered while the connection is still coming up then lands on this + // waiter instead of slipping past it. + void ensureSubscribed(); + + await new Promise((resolve) => { + let settled = false; + const settle = () => { + if (settled) return; + settled = true; + clearTimeout(timer); + emitter.off(key, settle); + signal?.removeEventListener('abort', settle); + resolve(); + }; + + const timer = setTimeout(settle, timeoutMs); + // Never hold the process open for a run nobody is waiting on anymore. + timer.unref?.(); + emitter.once(key, settle); + signal?.addEventListener('abort', settle, { once: true }); + }); + }, + + async close() { + const pending = subscription; + subscription = undefined; + // Keep a closed listener closed: nothing should re-open it after the + // world has been shut down. + retrySubscribeAfter = Number.POSITIVE_INFINITY; + emitter.removeAllListeners(); + const active = await pending?.catch(() => undefined); + await active?.close().catch(() => undefined); + }, + }; +} diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts index 7d6c10f86d..a83c29a751 100644 --- a/packages/world-postgres/src/storage.ts +++ b/packages/world-postgres/src/storage.ts @@ -76,6 +76,11 @@ import { import { monotonicFactory } from 'ulid'; import { type Drizzle, Schema } from './drizzle/index.js'; import type { SerializedContent } from './drizzle/schema.js'; +import { + getRunStatusPollIntervalMs, + notifyRunTerminal, + type RunStatusListener, +} from './run-status.js'; import { compact } from './util.js'; const DAY_MS = 24 * 60 * 60 * 1000; @@ -373,7 +378,15 @@ function deserializeStepError(step: any): Step { } as Step; } -export function createRunsStorage(drizzle: Drizzle): Storage['runs'] { +export function createRunsStorage( + drizzle: Drizzle, + /** + * Shared `LISTEN` subscription used by `waitForTerminalStatus`. Omit it and + * the wait still works, purely on its backstop re-read — which is what a + * direct caller constructing storage without a pool gets. + */ + runStatusListener?: RunStatusListener +): Storage['runs'] { const { runs } = Schema; const get = drizzle .select() @@ -382,21 +395,50 @@ export function createRunsStorage(drizzle: Drizzle): Storage['runs'] { .limit(1) .prepare('workflow_runs_get'); + const getRun = (async (id, params) => { + const [value] = await get.execute({ id }); + if (!value) { + throw new WorkflowRunNotFoundError(id); + } + value.output ||= value.outputJson; + value.input ||= value.inputJson; + value.executionContext ||= value.executionContextJson; + value.error ||= parseErrorJson(value.errorJson); + const deserialized = deserializeRunError(compact(value)); + const parsed = WorkflowRunSchema.parse(deserialized); + const resolveData = params?.resolveData ?? 'all'; + return filterRunData(parsed, resolveData); + }) as Storage['runs']['get']; + return { - get: (async (id, params) => { - const [value] = await get.execute({ id }); - if (!value) { - throw new WorkflowRunNotFoundError(id); + get: getRun, + + /** + * Long poll for a terminal run status — see + * `Storage['runs'].waitForTerminalStatus`. + * + * Reads the run, and while it is non-terminal parks on the run-terminal + * `NOTIFY` (bounded by the backstop re-read interval) before reading + * again. Returns the latest snapshot once `timeoutMs` is up, whatever its + * status, and propagates `WorkflowRunNotFoundError` exactly as `get` does. + */ + waitForTerminalStatus: (async (id, params) => { + const deadline = Date.now() + (params?.timeoutMs ?? 0); + while (true) { + const run = await getRun(id, params); + if (isTerminalWorkflowRunStatus(run.status)) return run; + + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0 || params?.signal?.aborted) return run; + + const waitMs = Math.min(remainingMs, getRunStatusPollIntervalMs()); + if (runStatusListener) { + await runStatusListener.wait(id, waitMs, params?.signal); + } else { + await new Promise((resolve) => setTimeout(resolve, waitMs)); + } } - value.output ||= value.outputJson; - value.input ||= value.inputJson; - value.executionContext ||= value.executionContextJson; - value.error ||= parseErrorJson(value.errorJson); - const deserialized = deserializeRunError(compact(value)); - const parsed = WorkflowRunSchema.parse(deserialized); - const resolveData = params?.resolveData ?? 'all'; - return filterRunData(parsed, resolveData); - }) as Storage['runs']['get'], + }) as NonNullable, getMany: (async (ids, params) => { const uniqueIds = [...new Set(ids)]; if (uniqueIds.length === 0) { @@ -2212,6 +2254,16 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] { }; } + // Wake `runs.waitForTerminalStatus` waiters. Every run-terminal + // transition in this world happens on the way to here (run_completed / + // run_failed / run_cancelled all update the row above), and the update + // has committed by now, so a woken waiter re-reads a terminal run. The + // early-return paths above are the idempotent ones — a run that was + // *already* terminal, whose original transition announced itself. + if (run && isTerminalWorkflowRunStatus(run.status)) { + await notifyRunTerminal(drizzle, effectiveRunId); + } + const eventResult: EventResult = { event: stripEventDataRefs(parsed, resolveData), run, diff --git a/packages/world-postgres/test/run-status-wait.test.ts b/packages/world-postgres/test/run-status-wait.test.ts new file mode 100644 index 0000000000..49c3ecd00d --- /dev/null +++ b/packages/world-postgres/test/run-status-wait.test.ts @@ -0,0 +1,220 @@ +import { execSync } from 'node:child_process'; +import { PostgreSqlContainer } from '@testcontainers/postgresql'; +import { WorkflowRunNotFoundError } from '@workflow/errors'; +import type { WorkflowRun } from '@workflow/world'; +import { Pool } from 'pg'; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + test, +} from 'vitest'; +import { createClient } from '../src/drizzle/index.js'; +import { + createRunStatusListener, + type RunStatusListener, +} from '../src/run-status.js'; +import { createEventsStorage, createRunsStorage } from '../src/storage.js'; + +/** + * `runs.waitForTerminalStatus` on world-postgres. + * + * The point of these tests is to pin down that the wait is driven by the + * run-terminal `NOTIFY` and not by its backstop re-read: the backstop is + * dialed up to 10s here, so a wait that resolves in milliseconds can only have + * been woken by the notification. + */ +describe('runs.waitForTerminalStatus (Postgres integration)', () => { + if (process.platform === 'win32') { + test.skip('skipped on Windows since it relies on a docker container', () => {}); + return; + } + + const pollIntervalEnv = 'WORKFLOW_POSTGRES_RUN_STATUS_POLL_INTERVAL_MS'; + const originalPollInterval = process.env[pollIntervalEnv]; + + let container: Awaited>; + let pool: Pool; + let drizzle: ReturnType; + let listener: RunStatusListener; + let runs: ReturnType; + let events: ReturnType; + + beforeAll(async () => { + container = await new PostgreSqlContainer('postgres:15-alpine').start(); + const dbUrl = container.getConnectionUri(); + process.env.DATABASE_URL = dbUrl; + process.env.WORKFLOW_POSTGRES_URL = dbUrl; + + execSync('pnpm db:push', { + stdio: 'inherit', + cwd: process.cwd(), + env: process.env, + }); + + // >1 connection: the wait holds a read while the completing writer needs + // its own, and the LISTEN client is separate from the pool entirely. + pool = new Pool({ connectionString: dbUrl, max: 4 }); + drizzle = createClient(pool); + listener = createRunStatusListener(pool); + runs = createRunsStorage(drizzle, listener); + events = createEventsStorage(drizzle); + }, 120_000); + + beforeEach(async () => { + // Only the NOTIFY can wake a wait this quickly. + process.env[pollIntervalEnv] = '10000'; + await pool.query( + 'TRUNCATE TABLE workflow.workflow_events, workflow.workflow_event_slots, workflow.workflow_steps, workflow.workflow_hooks, workflow.workflow_runs RESTART IDENTITY CASCADE' + ); + }); + + afterEach(() => { + if (originalPollInterval === undefined) { + delete process.env[pollIntervalEnv]; + } else { + process.env[pollIntervalEnv] = originalPollInterval; + } + }); + + afterAll(async () => { + await listener?.close(); + await pool?.end(); + await container?.stop(); + }); + + async function startRun(): Promise { + const created = await events.create(null, { + eventType: 'run_created', + eventData: { + deploymentId: 'dpl_test', + workflowName: 'test-workflow', + input: new Uint8Array([1]), + }, + } as never); + if (!created.run) throw new Error('Expected run to be created'); + await events.create(created.run.runId, { + eventType: 'run_started', + } as never); + return created.run; + } + + const finish = ( + runId: string, + eventType: 'run_completed' | 'run_failed' | 'run_cancelled', + eventData?: Record + ) => events.create(runId, { eventType, eventData } as never); + + const waitForTerminalStatus = () => { + const wait = runs.waitForTerminalStatus; + if (!wait) throw new Error('world-postgres should implement the long poll'); + return wait; + }; + + it('returns an already-terminal run immediately', async () => { + const run = await startRun(); + await finish(run.runId, 'run_completed', { output: new Uint8Array([2]) }); + + const startedAt = Date.now(); + const waited = await waitForTerminalStatus()(run.runId, { + timeoutMs: 30_000, + }); + + expect(waited.status).toBe('completed'); + expect(Date.now() - startedAt).toBeLessThan(1_000); + }); + + it('is woken by the run-terminal NOTIFY, not by its backstop', async () => { + const run = await startRun(); + + const pending = waitForTerminalStatus()(run.runId, { timeoutMs: 30_000 }); + // Give the waiter time to park on the LISTEN, then finish the run. + await new Promise((resolve) => setTimeout(resolve, 250)); + const startedAt = Date.now(); + await finish(run.runId, 'run_completed', { output: new Uint8Array([2]) }); + + const waited = await pending; + + expect(waited.status).toBe('completed'); + // The backstop re-read is 10s away, so anything close to instant proves + // the notification did the waking. + expect(Date.now() - startedAt).toBeLessThan(2_000); + }); + + it('wakes on a failed run too', async () => { + const run = await startRun(); + + const pending = waitForTerminalStatus()(run.runId, { timeoutMs: 30_000 }); + await new Promise((resolve) => setTimeout(resolve, 250)); + await finish(run.runId, 'run_failed', { + error: new Uint8Array([3]), + errorCode: 'USER_ERROR', + }); + + const waited = await pending; + expect(waited.status).toBe('failed'); + }); + + it('wakes on a cancelled run', async () => { + const run = await startRun(); + + const pending = waitForTerminalStatus()(run.runId, { timeoutMs: 30_000 }); + await new Promise((resolve) => setTimeout(resolve, 250)); + await finish(run.runId, 'run_cancelled'); + + const waited = await pending; + expect(waited.status).toBe('cancelled'); + }); + + it('returns the latest non-terminal snapshot when the budget expires', async () => { + const run = await startRun(); + + const startedAt = Date.now(); + const waited = await waitForTerminalStatus()(run.runId, { timeoutMs: 200 }); + + expect(waited.status).toBe('running'); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(190); + }); + + it('falls back to the backstop re-read without a listener', async () => { + // A runs storage built without the shared LISTEN subscription (a direct + // caller, or a pool that cannot host one) must still resolve — just on its + // re-read interval instead of on the notification. + process.env[pollIntervalEnv] = '50'; + const pollingRuns = createRunsStorage(drizzle); + const run = await startRun(); + + const wait = pollingRuns.waitForTerminalStatus; + if (!wait) throw new Error('expected a long-poll implementation'); + const pending = wait(run.runId, { timeoutMs: 30_000 }); + await new Promise((resolve) => setTimeout(resolve, 100)); + await finish(run.runId, 'run_completed', { output: new Uint8Array([2]) }); + + expect((await pending).status).toBe('completed'); + }); + + it('stops early when the caller aborts', async () => { + const run = await startRun(); + const controller = new AbortController(); + + const pending = waitForTerminalStatus()(run.runId, { + timeoutMs: 30_000, + signal: controller.signal, + }); + setTimeout(() => controller.abort(), 100); + + expect((await pending).status).toBe('running'); + }); + + it('fails like get for an unknown run', async () => { + await expect( + waitForTerminalStatus()('wrun_01JB0000000000000000000000', { + timeoutMs: 30_000, + }) + ).rejects.toBeInstanceOf(WorkflowRunNotFoundError); + }); +}); diff --git a/packages/world-vercel/src/run-status-long-poll.test.ts b/packages/world-vercel/src/run-status-long-poll.test.ts new file mode 100644 index 0000000000..74337f1d14 --- /dev/null +++ b/packages/world-vercel/src/run-status-long-poll.test.ts @@ -0,0 +1,292 @@ +import { WorkflowRunNotFoundError, WorkflowWorldError } from '@workflow/errors'; +import { MockAgent } from 'undici'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { + _resetRunStatusLongPollSupportForTests, + waitForWorkflowRunTerminalStatus, +} from './runs.js'; +import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; + +const ORIGIN = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; +const RUN_ID = 'wrun_01JB0000000000000000000000'; + +const runBody = (status: string) => ({ + runId: RUN_ID, + status, + deploymentId: 'dpl_1', + workflowName: 'test-workflow', + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', +}); + +function mockAgent(): MockAgent { + const agent = new MockAgent(); + agent.disableNetConnect(); + return agent; +} + +/** + * `waitForWorkflowRunTerminalStatus` — the `world-vercel` half of + * `runs.waitForTerminalStatus`, backed by workflow-server's long-pollable + * `GET /v2/runs/:runId/status`. + * + * The interesting behavior is the degradation: this adapter can be talking to + * a workflow-server that does not have the route, and it must tell that apart + * from a run that does not exist. + */ +describe('waitForWorkflowRunTerminalStatus', () => { + const requestTimeoutEnv = 'WORKFLOW_REQUEST_TIMEOUT_MS'; + const originalRequestTimeout = process.env[requestTimeoutEnv]; + + beforeEach(() => { + _resetRunStatusLongPollSupportForTests(); + }); + + afterEach(() => { + if (originalRequestTimeout === undefined) { + delete process.env[requestTimeoutEnv]; + } else { + process.env[requestTimeoutEnv] = originalRequestTimeout; + } + }); + + it('long polls the status route with the requested budget', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}/status?remoteRefBehavior=resolve&waitMs=25000`, + method: 'GET', + }) + .reply(200, runBody('completed')); + + const run = await waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 25_000 }, + { token: 'test-token', dispatcher: agent } + ); + + expect(run.status).toBe('completed'); + agent.assertNoPendingInterceptors(); + }); + + it('returns a non-terminal snapshot when the server budget expires', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}/status?remoteRefBehavior=resolve&waitMs=25000`, + method: 'GET', + }) + .reply(200, runBody('running')); + + // A run that is still going is an answer, not an error. + const run = await waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 25_000 }, + { token: 'test-token', dispatcher: agent } + ); + + expect(run.status).toBe('running'); + agent.assertNoPendingInterceptors(); + }); + + it('asks for lazy refs when the caller does not want payloads', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}/status?remoteRefBehavior=lazy&waitMs=1000`, + method: 'GET', + }) + .reply(200, runBody('completed')); + + const run = await waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 1_000, resolveData: 'none' }, + { token: 'test-token', dispatcher: agent } + ); + + expect(run.output).toBeUndefined(); + agent.assertNoPendingInterceptors(); + }); + + it('reads plainly when there is no budget to wait with', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}?remoteRefBehavior=resolve`, + method: 'GET', + }) + .reply(200, runBody('running')); + + const run = await waitForWorkflowRunTerminalStatus( + RUN_ID, + {}, + { token: 'test-token', dispatcher: agent } + ); + + expect(run.status).toBe('running'); + agent.assertNoPendingInterceptors(); + }); + + it('clamps the budget under the adapter request timeout', async () => { + // Leaves 10s of headroom, so a 12s request timeout permits a 2s wait — + // the budget must always expire as a response, never as a client timeout. + process.env[requestTimeoutEnv] = '12000'; + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}/status?remoteRefBehavior=resolve&waitMs=2000`, + method: 'GET', + }) + .reply(200, runBody('completed')); + + await waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 25_000 }, + { token: 'test-token', dispatcher: agent } + ); + + agent.assertNoPendingInterceptors(); + }); + + it('degrades to the plain read when the server has no status route', async () => { + const agent = mockAgent(); + // Two waits: the first discovers the route is missing, the second must not + // even try it again. + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}/status?remoteRefBehavior=resolve&waitMs=25000`, + method: 'GET', + }) + .reply(404, { error: 'not-found', message: 'No route matches GET …' }); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}?remoteRefBehavior=resolve`, + method: 'GET', + }) + .reply(200, runBody('running')) + .times(2); + + const config = { token: 'test-token', dispatcher: agent }; + const first = await waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 25_000 }, + config + ); + const second = await waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 25_000 }, + config + ); + + expect(first.status).toBe('running'); + expect(second.status).toBe('running'); + // Only one status-route attempt was made across both calls. + agent.assertNoPendingInterceptors(); + }); + + it('suppresses the fast path per backend, not process-wide', async () => { + // One process can hold worlds pointed at different backends — the + // api.vercel.com proxy and workflow-server directly — which can be on + // different versions. A miss against one must not disable the other. + const agent = mockAgent(); + const direct = agent.get(ORIGIN); + direct + .intercept({ + path: `/api/v2/runs/${RUN_ID}/status?remoteRefBehavior=resolve&waitMs=25000`, + method: 'GET', + }) + .reply(404, { error: 'not-found', message: 'No route matches GET …' }); + direct + .intercept({ + path: `/api/v2/runs/${RUN_ID}?remoteRefBehavior=resolve`, + method: 'GET', + }) + .reply(200, runBody('running')); + agent + .get('https://api.vercel.com') + .intercept({ + path: `/v1/workflow/v2/runs/${RUN_ID}/status?remoteRefBehavior=resolve&waitMs=25000`, + method: 'GET', + }) + .reply(200, runBody('completed')); + + await waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 25_000 }, + { token: 'test-token', dispatcher: agent } + ); + const viaProxy = await waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 25_000 }, + { + token: 'test-token', + projectConfig: { projectId: 'prj_1', teamId: 'team_1' }, + dispatcher: agent, + } + ); + + expect(viaProxy.status).toBe('completed'); + agent.assertNoPendingInterceptors(); + }); + + it('reports a missing run as not found, and keeps long polling enabled', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}/status?remoteRefBehavior=resolve&waitMs=25000`, + method: 'GET', + }) + .reply(404, { error: 'not-found', message: 'workflow run not found' }) + .times(2); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}?remoteRefBehavior=resolve`, + method: 'GET', + }) + .reply(404, { error: 'not-found', message: 'workflow run not found' }) + .times(2); + + const config = { token: 'test-token', dispatcher: agent }; + await expect( + waitForWorkflowRunTerminalStatus(RUN_ID, { timeoutMs: 25_000 }, config) + ).rejects.toBeInstanceOf(WorkflowRunNotFoundError); + + // A bad run ID says nothing about the server's routes, so the next wait + // still tries the fast path (its own 404 pair is consumed here). + await expect( + waitForWorkflowRunTerminalStatus(RUN_ID, { timeoutMs: 25_000 }, config) + ).rejects.toBeInstanceOf(WorkflowRunNotFoundError); + + agent.assertNoPendingInterceptors(); + }); + + it('propagates a server error instead of masking it with a plain read', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v2/runs/${RUN_ID}/status?remoteRefBehavior=resolve&waitMs=25000`, + method: 'GET', + }) + .reply(500, { error: 'internal-server-error', message: 'boom' }); + + await expect( + waitForWorkflowRunTerminalStatus( + RUN_ID, + { timeoutMs: 25_000 }, + { token: 'test-token', dispatcher: agent } + ) + ).rejects.toBeInstanceOf(WorkflowWorldError); + + agent.assertNoPendingInterceptors(); + }); +}); diff --git a/packages/world-vercel/src/runs.ts b/packages/world-vercel/src/runs.ts index 33147f525b..e525d9d928 100644 --- a/packages/world-vercel/src/runs.ts +++ b/packages/world-vercel/src/runs.ts @@ -13,16 +13,19 @@ import { type PaginatedResponse, PaginatedResponseSchema, SerializedDataSchema, + type WaitForTerminalRunStatusParams, type WorkflowRun, WorkflowRunBaseSchema, type WorkflowRunWithoutData, } from '@workflow/world'; import { z } from 'zod'; +import { getRequestTimeoutMs } from './http-core.js'; import { normalizeWorkflowRunData } from './serialized-data.js'; import type { APIConfig } from './utils.js'; import { DEFAULT_RESOLVE_DATA_OPTION, deserializeError, + getHttpUrl, makeRequest, } from './utils.js'; @@ -198,31 +201,179 @@ export async function getWorkflowRun( config?: APIConfig ): Promise { const resolveData = params?.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION; + + try { + return await readRun(id, { resolveData }, config); + } catch (error) { + if (error instanceof WorkflowWorldError && error.status === 404) { + throw new WorkflowRunNotFoundError(id); + } + throw error; + } +} + +/** + * Issue one run read against workflow-server and normalize it into the World's + * `WorkflowRun` shape. + * + * `waitMs`, when set, targets the long-pollable `GET /v2/runs/:runId/status` + * route instead of the plain read: same entity, same errors, but the server + * holds the request open until the run reaches a terminal status or the budget + * expires. Shared by `getWorkflowRun` and `waitForWorkflowRunTerminalStatus` so + * the two can never drift in how they parse or filter a run. + */ +async function readRun( + id: string, + params: { + resolveData: 'none' | 'all'; + waitMs?: number; + signal?: AbortSignal; + }, + config?: APIConfig +): Promise { + const { resolveData, waitMs, signal } = params; const remoteRefBehavior = resolveData === 'none' ? 'lazy' : 'resolve'; const searchParams = new URLSearchParams(); searchParams.set('remoteRefBehavior', remoteRefBehavior); + if (waitMs !== undefined) searchParams.set('waitMs', String(waitMs)); + const path = waitMs === undefined ? '' : '/status'; const queryString = searchParams.toString(); - const endpoint = `/v2/runs/${encodeURIComponent(id)}${queryString ? `?${queryString}` : ''}`; + const endpoint = `/v2/runs/${encodeURIComponent(id)}${path}${queryString ? `?${queryString}` : ''}`; - try { - const run = await makeRequest({ - endpoint, - options: { method: 'GET' }, - config, - retryConnectTimeout: true, - schema: (remoteRefBehavior === 'lazy' - ? WorkflowRunWireWithRefsSchema - : WorkflowRunWireSchema) as any, - }); + const run = await makeRequest({ + endpoint, + options: { method: 'GET', ...(signal ? { signal } : {}) }, + config, + retryConnectTimeout: true, + schema: (remoteRefBehavior === 'lazy' + ? WorkflowRunWireWithRefsSchema + : WorkflowRunWireSchema) as any, + }); - return filterRunData(run, resolveData); + return filterRunData(run, resolveData); +} + +/** + * Headroom kept between the wait budget we ask the server to hold and the + * adapter's own per-request HTTP timeout. The budget must always expire as a + * *response* (a non-terminal run) rather than as a client-side timeout: a + * timeout is indistinguishable from a broken backend and would turn a healthy + * wait into retry noise. + */ +const WAIT_TIMEOUT_HEADROOM_MS = 10_000; + +/** + * How long a `404`/`405`/`501` on the long-poll route suppresses further + * attempts before the next one re-probes. + * + * A miss means the backend serving this base URL predates the route (or has it + * rolled back), which is a property of the *backend*, not of the run — so it is + * cached rather than re-learned per call. It expires so a client that outlives + * a server roll-forward picks the fast path back up on its own. + */ +const LONG_POLL_UNSUPPORTED_TTL_MS = 5 * 60 * 1000; + +/** + * Suppression deadline per base URL, because one process can hold worlds + * pointed at different backends — the api.vercel.com proxy (with + * `projectConfig`) and workflow-server directly resolve to different hosts, + * which can be on different versions. A miss against one must not disable the + * fast path for the other. Bounded by construction: the key is the resolved + * base URL, of which a process has a handful at most. + */ +const longPollUnsupportedUntilByBaseUrl = new Map(); + +/** Test-only: forget that the long-poll route was unavailable. @internal */ +export function _resetRunStatusLongPollSupportForTests(): void { + longPollUnsupportedUntilByBaseUrl.clear(); +} + +/** + * Wait for a run to reach a terminal status, using workflow-server's + * long-pollable `GET /v2/runs/:runId/status` route. + * + * Implements `Storage['runs'].waitForTerminalStatus`: resolves as soon as the + * run is terminal, and otherwise with the latest snapshot once the budget + * expires. Never throws on a timeout — a still-running run is an answer. + * + * Degrades in two places, because the adapter can outlive the server version + * it was built against: + * + * - **Budget.** Clamped to leave {@link WAIT_TIMEOUT_HEADROOM_MS} under the + * adapter's per-request timeout, and the server clamps again to its own + * ceiling. A budget that clamps to zero is just a plain read. + * - **Missing route.** A `404` is ambiguous — the run may not exist, or this + * server may not have the route — so it is resolved by falling back to the + * plain read, which is the answer we want either way: it raises + * `WorkflowRunNotFoundError` for a missing run, and returns the run when the + * *route* was what was missing. Only the latter (proof that the run exists + * and the route does not) suppresses the fast path, and only for the base URL + * that answered, so neither one bad run ID nor one lagging backend can + * disable long polling everywhere. + */ +export async function waitForWorkflowRunTerminalStatus( + id: string, + params: WaitForTerminalRunStatusParams & { resolveData: 'none' }, + config?: APIConfig +): Promise; +export async function waitForWorkflowRunTerminalStatus( + id: string, + params?: WaitForTerminalRunStatusParams & { resolveData?: 'all' }, + config?: APIConfig +): Promise; +export async function waitForWorkflowRunTerminalStatus( + id: string, + params?: WaitForTerminalRunStatusParams, + config?: APIConfig +): Promise; +export async function waitForWorkflowRunTerminalStatus( + id: string, + params?: WaitForTerminalRunStatusParams, + config?: APIConfig +): Promise { + const resolveData = params?.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION; + const waitMs = Math.max( + 0, + Math.min( + params?.timeoutMs ?? 0, + getRequestTimeoutMs() - WAIT_TIMEOUT_HEADROOM_MS + ) + ); + + const { baseUrl } = getHttpUrl(config); + const unsupportedUntil = longPollUnsupportedUntilByBaseUrl.get(baseUrl) ?? 0; + + if (waitMs === 0 || Date.now() < unsupportedUntil) { + return getWorkflowRun(id, { resolveData }, config); + } + + try { + return await readRun( + id, + { + resolveData, + waitMs, + ...(params?.signal ? { signal: params.signal } : {}), + }, + config + ); } catch (error) { - if (error instanceof WorkflowWorldError && error.status === 404) { - throw new WorkflowRunNotFoundError(id); + if ( + !(error instanceof WorkflowWorldError) || + !(error.status === 404 || error.status === 405 || error.status === 501) + ) { + throw error; } - throw error; + + // Throws WorkflowRunNotFoundError when the run is what was missing. + const run = await getWorkflowRun(id, { resolveData }, config); + longPollUnsupportedUntilByBaseUrl.set( + baseUrl, + Date.now() + LONG_POLL_UNSUPPORTED_TTL_MS + ); + return run; } } diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts index 46fced038c..14c858356b 100644 --- a/packages/world-vercel/src/storage.ts +++ b/packages/world-vercel/src/storage.ts @@ -16,6 +16,7 @@ import { getWorkflowRun, getWorkflowRuns, listWorkflowRuns, + waitForWorkflowRunTerminalStatus, } from './runs.js'; import { getStep, listWorkflowRunSteps } from './steps.js'; import type { APIConfig } from './utils.js'; @@ -26,6 +27,10 @@ export function createStorage(config?: APIConfig): Storage { runs: { get: ((id: string, params?: any) => getWorkflowRun(id, params, config)) as Storage['runs']['get'], + waitForTerminalStatus: ((id: string, params?: any) => + waitForWorkflowRunTerminalStatus(id, params, config)) as NonNullable< + Storage['runs']['waitForTerminalStatus'] + >, getMany: ((ids: readonly string[], params?: any) => getWorkflowRuns(ids, params, config)) as NonNullable< Storage['runs']['getMany'] diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index ea7bb66eea..44c6073129 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -20,6 +20,7 @@ import type { BulkCancelWorkflowRunsResult, GetWorkflowRunParams, ListWorkflowRunsParams, + WaitForTerminalRunStatusParams, WorkflowRun, WorkflowRunWithoutData, } from './runs.js'; @@ -154,6 +155,63 @@ export interface Storage { params?: GetWorkflowRunParams ): Promise; + /** + * Long poll for a run to reach a terminal status (`completed`, `failed`, + * or `cancelled`), returning the same entity `get` returns. + * + * This is how a caller awaiting a run's outcome — `await run.returnValue` + * — avoids paying interval-poll quantization for it: instead of asking + * "is it done yet?" every second, it asks once and the World answers the + * moment the run finishes. + * + * The contract: + * + * - **Resolve as soon as the run is terminal**, with the run entity in + * the shape `params.resolveData` asks for. + * - **Resolve no later than roughly `params.timeoutMs`** with the latest + * snapshot, whatever its status. A timeout is a normal return, never an + * error: a run that is still running is a legitimate answer. + * - **`timeoutMs` is an upper bound, not a lower one.** An + * implementation MAY resolve earlier with a non-terminal snapshot — + * e.g. `@workflow/world-vercel` does when the backend it is talking to + * has no long-poll route and it degrades to a plain read. Callers must + * therefore pace their own retries rather than assume one call per + * `timeoutMs` (the runtime's `Run#pollReturnValue` keeps consecutive + * non-terminal observations at least one poll interval apart). + * - **Fail exactly like `get`.** A missing run throws + * `WorkflowRunNotFoundError`; transport failures surface as they would + * on any other read. + * + * OPTIONAL. Omit it entirely when the World has no way to wait — a + * deterministic simulator, a store with no change notification — and the + * runtime keeps interval-polling `get` on + * `WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS`. There is nothing to declare + * beyond the method's presence, and no behavior degrades when it is + * absent: the fast path is strictly additive. + * + * Implementations are free to satisfy this however their backend allows — + * a server-side long poll (`world-vercel` holds + * `GET /v2/runs/:runId/status` open), a change notification + * (`world-postgres` uses `LISTEN`/`NOTIFY`, `world-local` an in-process + * emitter), or a tight internal poll — as long as a lost or missing + * notification degrades to returning a snapshot rather than hanging past + * the budget. + */ + waitForTerminalStatus?: { + ( + id: string, + params: WaitForTerminalRunStatusParams & { resolveData: 'none' } + ): Promise; + ( + id: string, + params?: WaitForTerminalRunStatusParams & { resolveData?: 'all' } + ): Promise; + ( + id: string, + params?: WaitForTerminalRunStatusParams + ): Promise; + }; + /** * Retrieves several runs as one snapshot. The result preserves the input * order and contains `null` for run IDs that do not exist. diff --git a/packages/world/src/runs.ts b/packages/world/src/runs.ts index 08f0931c44..3012dc31dc 100644 --- a/packages/world/src/runs.ts +++ b/packages/world/src/runs.ts @@ -188,6 +188,28 @@ export interface GetWorkflowRunParams { resolveData?: ResolveData; } +/** + * Params for the optional `runs.waitForTerminalStatus` long poll. + */ +export interface WaitForTerminalRunStatusParams extends GetWorkflowRunParams { + /** + * How long the caller is willing to wait, in milliseconds. Treat it as an + * upper bound on the wait, not a promise about the duration: the call + * resolves as soon as the run is terminal, and a World may also resolve + * early with a non-terminal snapshot (see + * {@link Storage.runs.waitForTerminalStatus}). + * + * Implementations clamp this to whatever their backend can hold open. + */ + timeoutMs?: number; + + /** + * Abandon the wait when this signal aborts. Implementations that cannot + * observe it may ignore it; callers must not rely on it to bound the call. + */ + signal?: AbortSignal; +} + export interface ListWorkflowRunsParams { workflowName?: string; status?: WorkflowRunStatus;