/** * Utils used by the bundler when transforming code */ import type { CryptoKey } from './encryption.js'; import type { EventsConsumer } from './events-consumer.js'; import type { QueueItem } from './global.js'; import type { Serializable } from './schemas.js'; import type { StepHydrationCache } from './step-hydration-cache.js'; export type StepFunction = ((...args: Args) => Promise) & { maxRetries?: number; stepId?: string; }; /** * Register a step function to be served in the server bundle. * Also sets the stepId property on the function for serialization support. */ export declare function registerStepFunction(stepId: string, stepFn: StepFunction): void; /** * Find a registered step function by name */ export declare function getStepFunction(stepId: string): StepFunction | undefined; /** * Get closure variables for the current step function * @internal */ export { __private_getClosureVars } from './step/get-closure-vars.js'; export interface WorkflowOrchestratorContext { runId: string; encryptionKey: CryptoKey | undefined; globalThis: typeof globalThis; eventsConsumer: EventsConsumer; /** * Map of pending invocations keyed by correlationId. * Using Map instead of Array for O(1) lookup/delete operations. */ invocationsQueue: Map; onWorkflowError: (error: Error) => void; generateUlid: () => string; generateNanoid: () => string; /** * Sequential promise queue that ensures all event-driven promise resolutions * (step results, hook payloads, failures, suspensions) happen in event log * order. Every resolve, reject, or workflow error is chained through this * queue so that even if individual operations take variable time (e.g., * async decryption), promises resolve deterministically. */ promiseQueue: Promise; /** * Counter of in-flight async data delivery operations (step result * hydration, hook payload hydration). Suspensions must wait for this * to reach 0 before firing, to avoid preempting data delivery. */ pendingDeliveries: number; /** * Ordered registry of in-flight "branch-deciding" deliveries — the * resolutions a workflow typically `Promise.race`s on, or awaits from * independent concurrent branches: hook payloads (`hook_received`), wait * completions (`wait_completed`), and step results (`step_completed` / * `step_failed`). Keyed by the delivery's position (index) in the consumed * event log. * * The problem: each of these resolutions reaches workflow code after a * different, workload-dependent number of microtask hops. A buffered hook * payload is observed via the async hook iterator (`yield await this`), * costing extra hops; a `wait_completed` resolves with fewer, and a reused * sleep can resolve in an entirely earlier loop iteration; a step result is * gated on hydration whose cost varies between replays of the SAME * invocation — the first replay pays the full decrypt/decompress/revive, * while later replays sharing the invocation's `stepHydrationCache` * memo-hit small primitive results and resolve in one or two hops. Either * way, the resolution that the committed event log ordered first can lose a * `Promise.race` (or a `useStep` ULID allocation) to a faster- or * already-resolved competitor, diverging from the log and surfacing as * `ReplayDivergenceError`. * * The fix is a strict, deterministic delivery order anchored on * event-log position: a delivery does not resolve to the workflow until * every relevant earlier-in-log delivery has been delivered. Because the * gate is "the earlier delivery resolved", not "won a timing race", the * outcome is independent of microtask hops, hydration/decryption time, * and `Promise.race` argument order. Which earlier kinds a delivery defers * behind is spelled out on {@link awaitEarlierDeliveries}. * * Index is used rather than the `eventId` string because `eventId` is an * opaque, world-assigned value not guaranteed to sort in creation order * (only the bundled ULID worlds happen to). * * Optional so older/out-of-tree contexts (and lightweight test harnesses) * that do not initialize it degrade gracefully to the previous behavior. */ pendingDeliveryBarriers?: Map; /** * Per-run memoization cache for hydrated step return values, keyed by the * `step_completed` event id. Owned by the inline replay loop in `runtime.ts` * and threaded through each `runWorkflow` call so it survives across replay * iterations of the SAME run (a fresh context is created each iteration) but * never leaks across unrelated runs. * * On replay K of a sequential N-step workflow, the step consumer would * otherwise re-decrypt and re-parse the results of all K already-completed * steps — O(N²) across an invocation. This cache makes a completed step's * result available in O(1) on subsequent replays. Only primitive results are * memoized, so a shared reference can never let one replay's mutation leak * into the next; see `step-hydration-cache.ts` for the full rationale. * * Optional so contexts that do not initialize it (test harnesses) degrade * gracefully to re-hydrating every replay — identical to previous behavior. */ stepHydrationCache?: StepHydrationCache; } /** The kind of branch-deciding delivery a barrier represents. */ export type DeliveryKind = 'hook' | 'wait' | 'step'; interface DeliveryBarrierEntry { kind: DeliveryKind; /** Resolves once this delivery has resolved to the workflow. */ delivered: Promise; /** * Whether this delivery is committed to reaching the workflow without any * further action by workflow code. True for wait completions and step * results, which always resolve from their own chain, and for a hook payload * that already had a waiting consumer when it was consumed. * * False for a BUFFERED hook payload no consumer has claimed yet: it is * delivered by `claim()`, i.e. whenever the workflow next reads the hook — * which may be causally *after* a later-in-log delivery. `arm()` flips it * once a consumer takes the payload. */ armed: boolean; /** * Retire this entry: resolve `delivered` and remove it from the registry, * exactly as `markDelivered` would. Called only by the context's safety-net * dispenser ({@link ensureBarrierSafetyNet}), and only on the lowest-index * entry at delivery idle. Idempotent. */ retire: () => void; } /** * Awaits, in strict event-log order, every still-registered delivery that is * earlier in the log than `eventIndex` and that a delivery of `kind` defers * behind (see {@link DEFER_BEHIND}), so that this resolution is handed to the * workflow only after all relevant earlier-in-log deliveries have been. This * is what keeps a `Promise.race` — or the ULID a follow-up `useStep` draws on * a concurrent branch — deterministic and aligned with the committed event * log, independent of microtask-hop counts, hydration time, or race-argument * order. When this delivery does have to wait, it also yields a macrotask * afterwards so the earlier delivery's consumer can run to its own next * suspension point first; see the comment at that `await` for why ordering the * `resolve()` calls alone is not enough. * * One asymmetry: a STEP result additionally skips any earlier delivery that * will not resolve on its own, i.e. one blocked (directly or transitively) on * a buffered hook payload no consumer has claimed. Such a payload is delivered * only when the workflow next reads the hook, and reaching that read very * commonly requires the step result itself (`await stepX()` before the read). * Gating the step on it would stall the workflow until the barrier's idle * safety net fires, which then releases every delivery queued behind that * payload at once — losing exactly the race this ordering exists to protect. * Waits and hooks keep gating on unclaimed payloads: for them, waiting for the * claim IS the ordering guarantee (a `wait_completed` must not preempt a * payload the log ordered first). * * A delivery that does gate on an unclaimed payload still delivers in log * order once that payload's barrier is retired, and that rests on the * PAYLOAD's barrier being retired before that of anything parked behind it. * That order is structural: safety-net retirements go through one per-context * dispenser that only ever retires the lowest-index entry at delivery idle, * and every retirement that wakes a chain flips * {@link hasParkedCommittedDelivery} back to true, re-blocking the dispenser * until the chain has drained — see {@link ensureBarrierSafetyNet}. (This * used to rest on the FIFO of one idle poll per barrier, which held for a * single parked segment but decayed to timing noise with several — the * release order, and therefore the ULIDs drawn by the woken branches, then * depended on how much log the replay had loaded. storm-log-replay.test.ts * replays a production log corrupted exactly that way.) */ export declare function awaitEarlierDeliveries(ctx: WorkflowOrchestratorContext, eventIndex: number | undefined, kind: DeliveryKind): Promise; /** Handle for a registered branch-deciding delivery barrier. */ export interface DeliveryBarrier { /** * Mark this delivery as delivered to the workflow. Resolves its * `delivered` promise so any later-in-log delivery gated on it (via * {@link awaitEarlierDeliveries}) may proceed, and removes it from the * registry. Idempotent. */ markDelivered: () => void; /** * Mark this delivery as committed to happening, for a barrier registered * unarmed (a buffered hook payload) once a consumer has claimed it. From * then on a later step result may be ordered behind it. Idempotent. */ arm: () => void; } /** * Register a branch-deciding delivery at its event-log index so that later * deliveries can be ordered strictly after it. Returns an inert handle when * `pendingDeliveryBarriers` is not initialized. * * Pass `armed: false` for a delivery whose resolution waits on workflow code * asking for it (a buffered hook payload); call `arm()` when it does. * * To guarantee a later delivery gated on this one can never hang when this * delivery is abandoned (the workflow took a different branch or is * suspending and never observes it), the barrier auto-resolves at idle. * * INVARIANT required of every call site: a barrier that is ever `armed` must * be paired with a delivery chain that runs unconditionally — attached when * the event is consumed (waits, step results, waiting-consumer hook * payloads), or by the `claim()` whose invocation is what arms it (buffered * hook payloads). The idle check ({@link scheduleWhenIdle}) refuses to * observe idle while an armed, self-resolving barrier is undelivered, and the * safety net below is itself idle-gated — so an armed barrier with no * unconditional chain would livelock every idle check in the run, including * its own retirement. */ export declare function registerDeliveryBarrier(ctx: WorkflowOrchestratorContext, eventIndex: number | undefined, kind: DeliveryKind, options?: { armed?: boolean; }): DeliveryBarrier; /** * Whether no data delivery (step result, hook payload) is in flight right now. * * "In flight" is two distinct windows, each with its own guard: * `pendingDeliveries > 0` covers hydration inside the serial queue slots, and * {@link hasParkedCommittedDelivery} covers the detached gap between a slot * releasing that counter and the delivery's `resolve()` actually running — * deliberately outside `pendingDeliveries` (see step.ts), and invisible to it. * * Anything that decides a replay is over, or that a replay went wrong, has to * consult this first: while it is false the workflow VM is mid-reaction, so * what it has and has not done yet says nothing about the run. Two callers * read it, for the two such decisions: {@link scheduleWhenIdle} for the * suspension, and the events consumer's unconsumed-event check for divergence. * * A non-empty barrier registry counts as in flight, even when every remaining * entry is parked behind an unclaimed buffered payload. Those entries only * move when the safety-net dispenser retires them (lowest-first, see * {@link ensureBarrierSafetyNet}), and the deliveries they release are real * workflow reactions — a suspension raised before they run would be computed * from a VM that has not seen them, scheduling none of their follow-up work * and leaving the run dormant (the vercel/workflow#3183 shape). The dispenser * itself is gated on {@link canRetireAbandonedBarriers}, the weaker predicate * without the registry term, precisely so it can do the draining that this * predicate waits for; registry size strictly decreases at each retirement, * so idle is always reached. */ export declare function isDeliveryIdle(ctx: WorkflowOrchestratorContext): boolean; /** * Schedule a callback to fire only after all pending data deliveries * (step results, hook payloads) and async deserialization have completed. * Uses a polling loop: setTimeout(0) → check pendingDeliveries and the * barrier registry → if anything is still in flight, wait for promiseQueue → * repeat. This handles the multi-round delivery pattern where each hook * payload delivery cycle appends new async work to the promiseQueue. What * counts as in flight is {@link isDeliveryIdle}. * * The initial `setTimeout(0)` macrotask is load-bearing and must NOT be * downgraded to a microtask (`queueMicrotask`/`Promise.resolve().then`). * `pendingDeliveries` only guards the host-side hydration window; between a * delivery's `resolve()` and the workflow VM body running its continuation to * register the next subscriber, `pendingDeliveries` is already 0 even though * the VM is mid-reaction. Node does not guarantee a microtask scheduled in * the host context settles after the cross-VM promise chain (resolve in host * → workflow code in VM → subscribe back in host); the macrotask boundary * gives that chain time to run, so the suspension does not preempt a sibling * delivery still in flight. Empirically, replacing it with `queueMicrotask` * breaks hook/sleep `Promise.race` ordering (CorruptedEventLogError). */ export declare function scheduleWhenIdle(ctx: WorkflowOrchestratorContext, fn: () => void): void; //# sourceMappingURL=private.d.ts.map