From fde116558e151b13dca39364682a0b8c7c57c6be Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Sat, 22 Aug 2026 17:03:45 +0000 Subject: [PATCH] [core] Complete a due wait even when its run is already terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A pending wait parks a delayed continuation message on the workflow queue; when it fires, the delivery's job is to record the wait's `wait_completed`. If the run finished first — a `Promise.race([sleep('1h'), hook])` whose hook won, a cancellation, a failure — every path that discovers the terminal state returned and acked without writing it, so the wait stayed open in the log forever: replay, `inspect`, and the dashboard timeline all read a run that terminated while still sleeping. The completion is now unconditional on run status. On a terminal run it is all the delivery does — no replay, no suspension dispatch — and the message is acked only after the write settles, so a transient World failure redelivers instead of dropping the completion. Not-yet-due waits are left to their own continuations. The Worlds refused the write too: a terminal transition reaps the run's waits (`run.resources-released`), and `wait_completed` against a reaped wait was `Wait "…" not found`. world-local, world-postgres, and world-sim now accept it and record the event without resurrecting the entity, deduplicating against the surviving `.completed` claim (local) or the event log (postgres, sim). Older backends that still refuse are logged and acked rather than nacked — redelivery cannot change that verdict. Co-Authored-By: Claude Opus 5 (1M context) Co-Authored-By: Pranay Prakash <1797812+pranaygp@users.noreply.github.com> --- .../complete-due-waits-on-terminal-runs.md | 5 + ...ait-completed-after-terminal-reap-local.md | 5 + ...-completed-after-terminal-reap-postgres.md | 5 + .../docs/v5/how-it-works/event-sourcing.mdx | 2 + packages/core/src/runtime.ts | 53 ++++ packages/core/src/runtime/due-waits.test.ts | 272 +++++++++++++++++ packages/core/src/runtime/due-waits.ts | 204 +++++++++++++ .../runtime/wait-completion-terminal.test.ts | 277 ++++++++++++++++++ packages/world-local/src/storage.test.ts | 103 +++++++ .../world-local/src/storage/events-storage.ts | 134 ++++++--- packages/world-postgres/src/storage.ts | 77 ++++- packages/world-postgres/test/storage.test.ts | 59 ++++ packages/world-sim/src/store.test.ts | 46 +++ packages/world-sim/src/store.ts | 32 +- 14 files changed, 1232 insertions(+), 42 deletions(-) create mode 100644 .changeset/complete-due-waits-on-terminal-runs.md create mode 100644 .changeset/wait-completed-after-terminal-reap-local.md create mode 100644 .changeset/wait-completed-after-terminal-reap-postgres.md create mode 100644 packages/core/src/runtime/due-waits.test.ts create mode 100644 packages/core/src/runtime/due-waits.ts create mode 100644 packages/core/src/runtime/wait-completion-terminal.test.ts diff --git a/.changeset/complete-due-waits-on-terminal-runs.md b/.changeset/complete-due-waits-on-terminal-runs.md new file mode 100644 index 0000000000..0f4f932919 --- /dev/null +++ b/.changeset/complete-due-waits-on-terminal-runs.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Record a wait's `wait_completed` when its queue message comes due even if the run has already finished, instead of acknowledging the delivery with the wait left open in the log diff --git a/.changeset/wait-completed-after-terminal-reap-local.md b/.changeset/wait-completed-after-terminal-reap-local.md new file mode 100644 index 0000000000..82aae6ce79 --- /dev/null +++ b/.changeset/wait-completed-after-terminal-reap-local.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-local': patch +--- + +Accept `wait_completed` for a wait whose entity the run's terminal transition reaped, recording the event without resurrecting the entity diff --git a/.changeset/wait-completed-after-terminal-reap-postgres.md b/.changeset/wait-completed-after-terminal-reap-postgres.md new file mode 100644 index 0000000000..bd21083a59 --- /dev/null +++ b/.changeset/wait-completed-after-terminal-reap-postgres.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-postgres': patch +--- + +Accept `wait_completed` for a wait whose row the run's terminal transition deleted, recording the event without recreating the row and deduplicating against the event log diff --git a/docs/content/docs/v5/how-it-works/event-sourcing.mdx b/docs/content/docs/v5/how-it-works/event-sourcing.mdx index 14c3bcf321..880d48188c 100644 --- a/docs/content/docs/v5/how-it-works/event-sourcing.mdx +++ b/docs/content/docs/v5/how-it-works/event-sourcing.mdx @@ -199,6 +199,8 @@ Events are categorized by the entity type they affect. Each event contains metad | `wait_created` | Creates a new wait in `waiting` state. Contains the timestamp when the wait should complete. | | `wait_completed` | Transitions the wait to `completed` state when the delay period has elapsed. | +A wait's `wait_completed` is recorded when the wait comes due, even if the run finished first — a `Promise.race([sleep('1h'), hook])` whose hook won still records the sleep's completion an hour later. Only the log entry is written in that case: the run is terminal, so nothing is replayed and the wait entity itself stays released. Without it a `wait_created` with no matching `wait_completed` would read as a run that ended while still sleeping. + ## Terminal States Terminal states represent the end of an entity's lifecycle. Once an entity reaches a terminal state, no further events can transition it to another state. diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 0c9bc472d2..77535085ac 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -27,6 +27,7 @@ import { getQueueTopicPrefix, isLegacySpecVersion, isTerminalRunEventType, + isTerminalWorkflowRunStatus, ROOT_RUN_ID_ATTRIBUTE, type RunInput, resolveQueueNamespace, @@ -67,6 +68,7 @@ import { guardDeploymentAffinity, type ReenqueueArgs, } from './runtime/deployment-guard.js'; +import { completeDueWaits } from './runtime/due-waits.js'; import { absorbSkippedSlotReport, appendUniqueEvents, @@ -2070,6 +2072,17 @@ export function workflowEntrypoint( eventType: terminalEvent.eventType, } ); + // The preload is the complete log (that is what + // `usableReplayPreload` attests), so a due wait is + // visible right here — record its completion before + // consuming the delivery. See runtime/due-waits.ts. + await completeDueWaits({ + world, + runId, + events: result.events, + specVersion: result.run.specVersion, + requestId, + }); return; } workflowRun = result.run; @@ -2300,6 +2313,22 @@ export function workflowEntrypoint( // so that we actually exit here without replaying the workflow at all, in the case // the replaying the workflow is itself failing. + // Terminal: no replay, no suspension dispatch — but + // a wait this delivery may have been scheduled for + // still gets its completion recorded before the + // message is acknowledged. See runtime/due-waits.ts. + if (isTerminalWorkflowRunStatus(result.run.status)) { + await completeDueWaits({ + world, + runId, + events: + eventLog.type === 'ready' + ? eventLog.events + : undefined, + specVersion: result.run.specVersion, + requestId, + }); + } return; } } catch (err) { @@ -2316,6 +2345,17 @@ export function workflowEntrypoint( 'Run already finished during setup, skipping', { workflowRunId: runId, message: err.message } ); + // Same as the non-running branch above: the run is + // done, but a due wait's completion is this + // delivery's job and is not skipped with the rest. + // The log has to be read here — the rejection + // carried no preload. + await completeDueWaits({ + world, + runId, + specVersion: workflowRun?.specVersion, + requestId, + }); return; } else { const errorCode = getWorkflowSetupErrorCode(err); @@ -2735,7 +2775,20 @@ export function workflowEntrypoint( // derived from these events, so checking the log here // gives us the same signal as a runs.get() round-trip // without the extra request per loop iteration. + // + // A wait that has come due is still completed first: the + // delivery may BE that wait's continuation, and the + // completion belongs in the log whether or not there is + // anything left to replay (see runtime/due-waits.ts). + // Only then is the message acknowledged. if (hasRecordedTerminalRunEvent(eventLog.events, runId)) { + await completeDueWaits({ + world, + runId, + events: eventLog.events, + specVersion: workflowRun?.specVersion, + requestId, + }); return; } diff --git a/packages/core/src/runtime/due-waits.test.ts b/packages/core/src/runtime/due-waits.test.ts new file mode 100644 index 0000000000..9076febd6a --- /dev/null +++ b/packages/core/src/runtime/due-waits.test.ts @@ -0,0 +1,272 @@ +import { EntityConflictError, WorkflowWorldError } from '@workflow/errors'; +import { + type CreateEventRequest, + type Event, + SPEC_VERSION_CURRENT, + SPEC_VERSION_LEGACY, + slotToEventId, + type World, +} from '@workflow/world'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { completeDueWaits, findDueWaits } from './due-waits.js'; +import { setWorld } from './world.js'; + +const runId = 'wrun_due_waits'; +const now = new Date('2026-05-19T12:00:00.000Z'); + +let eventIndex = 0; +function event(data: CreateEventRequest): Event { + eventIndex += 1; + return { + ...data, + specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, + runId, + eventId: slotToEventId(eventIndex), + createdAt: now, + } as Event; +} + +function waitCreated(correlationId: string, resumeAt: Date): Event { + return event({ + eventType: 'wait_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId, + eventData: { resumeAt }, + }); +} + +function waitCompleted(correlationId: string): Event { + return event({ + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId, + }); +} + +/** + * A World that records `events.create` calls, optionally rejecting + * `wait_completed` writes with `reject`. + */ +function fakeWorld(reject?: (correlationId: string) => unknown) { + const create = vi.fn(async (_runId: string, data: CreateEventRequest) => { + if (data.eventType === 'wait_completed' && reject) { + const err = reject(data.correlationId); + if (err) throw err; + } + return { event: event(data) }; + }); + const list = vi.fn(); + const world = { + specVersion: SPEC_VERSION_CURRENT, + events: { create, list }, + } as unknown as World; + return { world, create, list }; +} + +describe('findDueWaits', () => { + it('returns waits whose resumeAt has passed and that have no completion', () => { + const due = findDueWaits( + [ + waitCreated('wait_past', new Date(+now - 1_000)), + waitCreated('wait_exactly_now', now), + waitCreated('wait_future', new Date(+now + 1_000)), + waitCreated('wait_done', new Date(+now - 1_000)), + waitCompleted('wait_done'), + ], + +now + ); + + expect(due.map((w) => w.correlationId)).toEqual([ + 'wait_past', + 'wait_exactly_now', + ]); + expect(due[0]?.resumeAt).toEqual(new Date(+now - 1_000)); + }); + + it('accepts a resumeAt that arrived as a string', () => { + const created = waitCreated('wait_str', new Date(+now - 1_000)); + (created as { eventData: { resumeAt: unknown } }).eventData.resumeAt = + new Date(+now - 1_000).toISOString(); + + expect(findDueWaits([created], +now).map((w) => w.correlationId)).toEqual([ + 'wait_str', + ]); + }); + + it('leaves a wait alone when its deadline cannot be read', () => { + const created = waitCreated('wait_bad', new Date(+now - 1_000)); + (created as { eventData: { resumeAt: unknown } }).eventData.resumeAt = + 'not-a-date'; + + expect(findDueWaits([created], +now)).toEqual([]); + }); +}); + +describe('completeDueWaits', () => { + afterEach(() => { + setWorld(undefined); + vi.restoreAllMocks(); + }); + + it('records a wait_completed carrying the wait_created resumeAt', async () => { + const { world, create } = fakeWorld(); + const resumeAt = new Date(+now - 5_000); + + const summary = await completeDueWaits({ + world, + runId, + events: [waitCreated('wait_1', resumeAt)], + requestId: 'req_1', + now: +now, + }); + + expect(summary).toEqual({ + completed: ['wait_1'], + alreadyCompleted: [], + unrecordable: [], + }); + expect(create).toHaveBeenCalledTimes(1); + expect(create).toHaveBeenCalledWith( + runId, + { + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: 'wait_1', + eventData: { resumeAt }, + }, + { requestId: 'req_1' } + ); + }); + + it('writes nothing when no wait is due', async () => { + const { world, create } = fakeWorld(); + + const summary = await completeDueWaits({ + world, + runId, + events: [ + waitCreated('wait_future', new Date(+now + 60_000)), + waitCreated('wait_done', new Date(+now - 60_000)), + waitCompleted('wait_done'), + ], + now: +now, + }); + + expect(summary.completed).toEqual([]); + expect(create).not.toHaveBeenCalled(); + }); + + it('loads the event log when the caller has none', async () => { + const { world, create, list } = fakeWorld(); + list.mockResolvedValue({ + data: [waitCreated('wait_loaded', new Date(+now - 1_000))], + hasMore: false, + cursor: null, + }); + // loadWorkflowRunEvents reads the ambient World, not the argument. + setWorld(world); + + const summary = await completeDueWaits({ world, runId, now: +now }); + + expect(list).toHaveBeenCalledTimes(1); + expect(summary.completed).toEqual(['wait_loaded']); + expect(create).toHaveBeenCalledTimes(1); + }); + + it('treats a concurrent completion as done, not as a failure', async () => { + const { world } = fakeWorld( + () => new EntityConflictError('Wait "wait_1" already completed') + ); + + const summary = await completeDueWaits({ + world, + runId, + events: [waitCreated('wait_1', new Date(+now - 1_000))], + now: +now, + }); + + expect(summary).toEqual({ + completed: [], + alreadyCompleted: ['wait_1'], + unrecordable: [], + }); + }); + + it('reports a wait an older World cannot record, without failing the delivery', async () => { + // A backend that drops a terminal run's waits outright has nowhere to put + // the completion. Redelivery would reach the same verdict, so the caller + // must still be free to acknowledge. + const { world } = fakeWorld( + (correlationId) => + new WorkflowWorldError(`Wait "${correlationId}" not found`) + ); + + const summary = await completeDueWaits({ + world, + runId, + events: [waitCreated('wait_1', new Date(+now - 1_000))], + now: +now, + }); + + expect(summary).toEqual({ + completed: [], + alreadyCompleted: [], + unrecordable: ['wait_1'], + }); + }); + + it('rethrows a retryable World failure so the message is redelivered', async () => { + const { world } = fakeWorld( + () => new WorkflowWorldError('upstream unavailable', { status: 503 }) + ); + + await expect( + completeDueWaits({ + world, + runId, + events: [waitCreated('wait_1', new Date(+now - 1_000))], + now: +now, + }) + ).rejects.toThrow('upstream unavailable'); + }); + + it('completes every due wait even when one of them cannot be recorded', async () => { + const { world, create } = fakeWorld((correlationId) => + correlationId === 'wait_1' + ? new WorkflowWorldError('Wait "wait_1" not found') + : undefined + ); + + const summary = await completeDueWaits({ + world, + runId, + events: [ + waitCreated('wait_1', new Date(+now - 2_000)), + waitCreated('wait_2', new Date(+now - 1_000)), + ], + now: +now, + }); + + expect(summary.unrecordable).toEqual(['wait_1']); + expect(summary.completed).toEqual(['wait_2']); + expect(create).toHaveBeenCalledTimes(2); + }); + + it('uses the legacy write shape for a legacy run', async () => { + const { world, create } = fakeWorld(); + + await completeDueWaits({ + world, + runId, + events: [waitCreated('wait_1', new Date(+now - 1_000))], + specVersion: SPEC_VERSION_LEGACY, + now: +now, + }); + + expect(create).toHaveBeenCalledWith( + runId, + { eventType: 'wait_completed', correlationId: 'wait_1' }, + { requestId: undefined, v1Compat: true } + ); + }); +}); diff --git a/packages/core/src/runtime/due-waits.ts b/packages/core/src/runtime/due-waits.ts new file mode 100644 index 0000000000..ee15bcb0f2 --- /dev/null +++ b/packages/core/src/runtime/due-waits.ts @@ -0,0 +1,204 @@ +/** + * Completing a run's due waits independently of its status. + * + * A pending `wait` (`sleep()`, a `Promise.race` timer) parks a delayed + * continuation message on the workflow queue that fires when the wait comes + * due; the delivery's job is to record the wait's `wait_completed` and let the + * next replay advance past it (see runtime/wait-continuation.ts). + * + * That delivery can arrive at a run that has already finished — a raced + * `Promise.race([sleep('1h'), hook])` whose hook won, a cancellation, a + * failure. The handler has nothing left to replay in that case, but the wait's + * completion still has to be recorded: the event log is the run's whole + * history, and a `wait_created` with no `wait_completed` in it is an open wait + * forever. Every reader of the log — replay, `inspect`, the dashboard's + * timeline — then shows a run that terminated while still sleeping, and + * anything deriving open-wait state from the log (see `openHookAndWaitState`) + * counts a wait that can never resolve. + * + * So the completion is unconditional on run status, and on a terminal run it + * is ALL the delivery does: no replay, no suspension dispatch, no new entities + * — a terminal run rejects those anyway. The queue message is acknowledged + * only after the write settles, so a transient World failure redelivers rather + * than dropping the completion on the floor. + * + * Not-yet-due waits are left alone. Each one has (or will get) its own + * continuation for its own deadline, and that delivery takes the same path + * through here when it fires. + */ + +import { EntityConflictError } from '@workflow/errors'; +import { + type Event, + isLegacySpecVersion, + SPEC_VERSION_CURRENT, + type World, +} from '@workflow/world'; +import { isRetryableWorldError } from '../classify-error.js'; +import { runtimeLogger } from '../logger.js'; +import { loadWorkflowRunEvents } from './helpers.js'; + +/** A `wait_created` whose `resumeAt` has passed with no `wait_completed`. */ +interface DueWait { + correlationId: string; + resumeAt: Date; +} + +export interface DueWaitCompletionSummary { + /** Waits this call recorded a `wait_completed` for. */ + completed: string[]; + /** Waits another writer had already completed. */ + alreadyCompleted: string[]; + /** + * Waits the World refused to complete for a reason redelivery cannot fix — + * an older backend that drops a terminal run's waits outright, so the + * completion has nowhere to land. Logged and moved past: nacking the + * delivery would burn every redelivery on a verdict that cannot change. + */ + unrecordable: string[]; +} + +/** + * `resumeAt` as a Date. Worlds hand back parsed events, but a `resumeAt` that + * survived a JSON hop arrives as a string — a wait whose deadline cannot be + * read is left alone rather than treated as due. + */ +function readResumeAt(value: unknown): Date | undefined { + const date = + value instanceof Date + ? value + : typeof value === 'string' || typeof value === 'number' + ? new Date(value) + : undefined; + return date && !Number.isNaN(date.getTime()) ? date : undefined; +} + +/** The run's waits that have come due and are still open in `events`. */ +export function findDueWaits(events: Event[], now: number): DueWait[] { + const completed = new Set(); + for (const event of events) { + if (event.eventType === 'wait_completed') + completed.add(event.correlationId); + } + const due: DueWait[] = []; + for (const event of events) { + if (event.eventType !== 'wait_created') continue; + if (event.correlationId === undefined) continue; + if (completed.has(event.correlationId)) continue; + const resumeAt = readResumeAt(event.eventData?.resumeAt); + if (!resumeAt || now < resumeAt.getTime()) continue; + due.push({ correlationId: event.correlationId, resumeAt }); + } + return due; +} + +/** What one `wait_completed` write settled as. */ +type WriteOutcome = 'completed' | 'alreadyCompleted' | 'unrecordable'; + +/** + * Write one wait's `wait_completed`. + * + * Legacy runs take the shape `wakeUpRun` uses for them: no `specVersion`, no + * `eventData`, and the `v1Compat` flag that routes the write to the World's + * legacy handler. + */ +async function writeWaitCompleted( + world: World, + runId: string, + wait: DueWait, + v1Compat: boolean, + requestId: string | undefined +): Promise { + try { + await world.events.create( + runId, + v1Compat + ? { + eventType: 'wait_completed', + correlationId: wait.correlationId, + } + : { + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: wait.correlationId, + eventData: { resumeAt: wait.resumeAt }, + }, + { requestId, ...(v1Compat ? { v1Compat: true } : {}) } + ); + return 'completed'; + } catch (err) { + if (EntityConflictError.is(err)) return 'alreadyCompleted'; + // Retryable means the next delivery of this message can still land the + // write, so let it nack. Anything else is a verdict redelivery would only + // reproduce. + if (isRetryableWorldError(err)) throw err; + runtimeLogger.warn( + 'Wait came due but its completion could not be recorded', + { + workflowRunId: runId, + correlationId: wait.correlationId, + errorName: err instanceof Error ? err.name : 'UnknownError', + errorMessage: err instanceof Error ? err.message : String(err), + } + ); + return 'unrecordable'; + } +} + +/** + * Record `wait_completed` for every wait of `runId` that has come due. + * + * Resolves only once every due wait is either recorded, already recorded, or + * provably unrecordable, so a caller can treat the resolution as permission to + * acknowledge the delivery. Retryable World failures are rethrown for exactly + * that reason. + * + * `events` is the caller's already-loaded log when it has one; otherwise the + * log is read here. Nothing is written when no wait is due, which is the + * overwhelmingly common case for a delivery that reaches a terminal run. + */ +export async function completeDueWaits({ + world, + runId, + events, + specVersion, + requestId, + now = Date.now(), +}: { + world: World; + runId: string; + events?: Event[]; + specVersion?: number; + requestId?: string; + now?: number; +}): Promise { + const summary: DueWaitCompletionSummary = { + completed: [], + alreadyCompleted: [], + unrecordable: [], + }; + + const log = events ?? (await loadWorkflowRunEvents(runId)).events; + const due = findDueWaits(log, now); + if (due.length === 0) return summary; + + const v1Compat = isLegacySpecVersion(specVersion ?? SPEC_VERSION_CURRENT); + for (const wait of due) { + const outcome = await writeWaitCompleted( + world, + runId, + wait, + v1Compat, + requestId + ); + summary[outcome].push(wait.correlationId); + } + + if (summary.completed.length > 0) { + runtimeLogger.debug('Completed due waits', { + workflowRunId: runId, + completed: summary.completed, + }); + } + return summary; +} diff --git a/packages/core/src/runtime/wait-completion-terminal.test.ts b/packages/core/src/runtime/wait-completion-terminal.test.ts new file mode 100644 index 0000000000..2492e0f64b --- /dev/null +++ b/packages/core/src/runtime/wait-completion-terminal.test.ts @@ -0,0 +1,277 @@ +import { + EntityConflictError, + RunExpiredError, + WorkflowWorldError, +} from '@workflow/errors'; +import { + type CreateEventRequest, + type Event, + type EventResult, + SPEC_VERSION_CURRENT, + slotToEventId, + type WorkflowRun, + type World, +} from '@workflow/world'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { workflowEntrypoint } from '../runtime.js'; +import { dehydrateWorkflowArguments } from '../serialization.js'; +import { setWorld } from './world.js'; + +vi.mock('@vercel/functions', () => ({ + waitUntil: vi.fn(), +})); + +vi.mock('@workflow/utils/get-port', () => ({ + getPort: vi.fn().mockResolvedValue(3000), +})); + +const runId = 'wrun_terminal_wait'; +const workflowName = 'workflow'; +const deploymentId = 'dpl_terminal_wait'; +const waitCorrelationId = 'wait_terminal'; +const startedAt = new Date('2026-05-19T12:00:00.000Z'); +const fixedNow = new Date('2026-05-19T13:00:00.000Z'); + +/** + * The wait's continuation message arriving at a run that already finished. + * + * `run_started` is the runtime's first write on every delivery, and how a + * terminal run is discovered: a World that reaps the run rejects it + * (`runStartedOutcome: 'expired'` / `'conflict'`), one that keeps it answers + * with the terminal row (`'terminal-run'`), and one that accepts the write + * leaves the terminal event to be found in the log + * (`'terminal-in-log'` — the node replay loop's own check). + * + * All four have the same job: record the due wait's `wait_completed`, do no + * replay, and acknowledge. + */ +async function runTerminalDeliveryScenario(options: { + runStartedOutcome: + | 'expired' + | 'conflict' + | 'terminal-run' + | 'terminal-in-log'; + /** `resumeAt` of the run's single open wait. Defaults to already elapsed. */ + waitResumeAt?: Date; + /** Rejection for the `wait_completed` write, if any. */ + rejectWaitCompletion?: () => unknown; +}) { + vi.spyOn(Date, 'now').mockReturnValue(+fixedNow); + + const workflowArgs = await dehydrateWorkflowArguments([], runId, undefined); + const terminalRunStatus = + options.runStartedOutcome === 'terminal-run' ? 'completed' : 'running'; + const workflowRun = { + runId, + workflowName, + status: terminalRunStatus, + input: workflowArgs, + deploymentId, + specVersion: SPEC_VERSION_CURRENT, + startedAt, + createdAt: startedAt, + updatedAt: startedAt, + ...(terminalRunStatus === 'completed' ? { completedAt: startedAt } : {}), + } as WorkflowRun; + + let eventIndex = 0; + const event = (data: CreateEventRequest): Event => { + eventIndex += 1; + return { + ...data, + specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, + runId, + eventId: slotToEventId(eventIndex), + createdAt: new Date(+startedAt + eventIndex * 100), + } as Event; + }; + + const durableEvents: Event[] = [ + event({ + eventType: 'run_created', + specVersion: SPEC_VERSION_CURRENT, + eventData: { deploymentId, workflowName, input: workflowArgs }, + } as unknown as CreateEventRequest), + event({ eventType: 'run_started', specVersion: SPEC_VERSION_CURRENT }), + event({ + eventType: 'wait_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: waitCorrelationId, + eventData: { + resumeAt: options.waitResumeAt ?? new Date(+fixedNow - 1_000), + }, + }), + event({ + eventType: 'run_completed', + specVersion: SPEC_VERSION_CURRENT, + eventData: { output: undefined }, + } as unknown as CreateEventRequest), + ]; + + const createdEvents: Event[] = []; + const listEvents = vi.fn(async () => ({ + data: [...durableEvents], + hasMore: false, + cursor: durableEvents.at(-1)?.eventId ?? null, + })); + + const createEvent = vi.fn( + async (_runId: string, request: CreateEventRequest) => { + if (request.eventType === 'run_started') { + switch (options.runStartedOutcome) { + case 'expired': + throw new RunExpiredError( + `Workflow run "${runId}" is already in terminal state "completed"` + ); + case 'conflict': + throw new EntityConflictError( + `Cannot transition run from terminal state "completed"` + ); + case 'terminal-run': + case 'terminal-in-log': + return { + run: workflowRun, + events: [...durableEvents], + cursor: durableEvents.at(-1)?.eventId ?? null, + hasMore: false, + maxEvents: 10_000, + } satisfies EventResult; + } + } + if ( + request.eventType === 'wait_completed' && + options.rejectWaitCompletion + ) { + const err = options.rejectWaitCompletion(); + if (err) throw err; + } + const created = event(request); + durableEvents.push(created); + createdEvents.push(created); + return { event: created }; + } + ); + + const queue = vi.fn().mockResolvedValue({ messageId: 'msg_out' }); + let capturedHandler: + | (( + message: unknown, + metadata: { queueName: string; messageId: string; attempt: number } + ) => Promise) + | undefined; + + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + createQueueHandler: vi.fn((_prefix, handler) => { + capturedHandler = handler; + return vi.fn(); + }), + events: { list: listEvents, create: createEvent }, + queue, + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + } as unknown as World); + + const workflowCode = ` + const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")]; + async function workflow() { + await sleep("1h"); + } + ;globalThis.__private_workflows = new Map([[${JSON.stringify(workflowName)}, workflow]]); + `; + + const handler = workflowEntrypoint(workflowCode); + await handler(new Request('http://localhost', { method: 'POST' })); + expect(capturedHandler).toBeDefined(); + + const deliver = () => + capturedHandler?.( + { runId }, + { + queueName: `__wkf_workflow_${workflowName}`, + messageId: 'msg_wait_continuation', + attempt: 1, + } + ); + + return { deliver, createdEvents, createEvent, listEvents, queue }; +} + +describe('due wait completion on a terminal run', () => { + afterEach(() => { + setWorld(undefined); + vi.restoreAllMocks(); + }); + + it.each([ + ['run_started is rejected as expired', 'expired'], + ['run_started conflicts with the terminal transition', 'conflict'], + ['run_started reports the terminal run', 'terminal-run'], + ['the terminal event is found in the event log', 'terminal-in-log'], + ] as const)('records the due wait_completed when %s', async (_label, outcome) => { + const scenario = await runTerminalDeliveryScenario({ + runStartedOutcome: outcome, + }); + + await expect(scenario.deliver()).resolves.toBeUndefined(); + + expect(scenario.createdEvents).toEqual([ + expect.objectContaining({ + eventType: 'wait_completed', + correlationId: waitCorrelationId, + }), + ]); + // Terminal: no replay-derived writes and no follow-up message. The + // acknowledgement above is the delivery's only other effect. + expect(scenario.queue).not.toHaveBeenCalled(); + }); + + it('leaves a wait that is not due yet alone', async () => { + const scenario = await runTerminalDeliveryScenario({ + runStartedOutcome: 'expired', + waitResumeAt: new Date(+fixedNow + 60_000), + }); + + await scenario.deliver(); + + expect(scenario.createdEvents).toEqual([]); + }); + + it('does not complete a wait twice across redeliveries', async () => { + const scenario = await runTerminalDeliveryScenario({ + runStartedOutcome: 'expired', + }); + + await scenario.deliver(); + await scenario.deliver(); + + expect( + scenario.createdEvents.filter((e) => e.eventType === 'wait_completed') + ).toHaveLength(1); + }); + + it('acknowledges when the World refuses to record the completion', async () => { + // An older backend that drops a terminal run's waits has nowhere to put + // the completion, and every redelivery would reach the same verdict. + const scenario = await runTerminalDeliveryScenario({ + runStartedOutcome: 'expired', + rejectWaitCompletion: () => + new WorkflowWorldError(`Wait "${waitCorrelationId}" not found`), + }); + + await expect(scenario.deliver()).resolves.toBeUndefined(); + expect(scenario.createdEvents).toEqual([]); + }); + + it('does not acknowledge when the completion fails transiently', async () => { + const scenario = await runTerminalDeliveryScenario({ + runStartedOutcome: 'expired', + rejectWaitCompletion: () => + new WorkflowWorldError('backend unavailable', { status: 503 }), + }); + + // Nack, so the next delivery retries the completion. The failure is the + // World's, not the run's — nothing about the run is rewritten. + await expect(scenario.deliver()).rejects.toThrow('backend unavailable'); + expect(scenario.createdEvents).toEqual([]); + }); +}); diff --git a/packages/world-local/src/storage.test.ts b/packages/world-local/src/storage.test.ts index 8758edab5f..8d5883ea4d 100644 --- a/packages/world-local/src/storage.test.ts +++ b/packages/world-local/src/storage.test.ts @@ -4856,6 +4856,109 @@ describe('Storage', () => { /not found/i ); }); + + it('should allow wait_completed for a wait the terminal transition reaped', async () => { + // The wait's continuation fires after the run ended (a raced + // `Promise.race([sleep, hook])` whose hook won). Its completion still + // belongs in the log: a wait_created with no wait_completed reads as an + // open wait forever. The reaped entity stays reaped. + const run = await createRun(storage, { + deploymentId: 'deployment-123', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); + await createWait(storage, run.runId, { + waitId: 'wait_reaped', + resumeAt: new Date('2099-01-01'), + }); + await updateRun(storage, run.runId, 'run_completed', { + output: new Uint8Array([3]), + }); + + const result = await storage.events.create(run.runId, { + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: 'wait_reaped', + }); + + expect(result.event?.eventType).toBe('wait_completed'); + expect(result.event?.correlationId).toBe('wait_reaped'); + // No entity is resurrected — a terminal run holds no waits. + expect(result.wait).toBeUndefined(); + await expect( + fs.access(path.join(testDir, 'waits', `${run.runId}-wait_reaped.json`)) + ).rejects.toMatchObject({ code: 'ENOENT' }); + + const events = await storage.events.list({ runId: run.runId }); + expect( + events.data.filter((e) => e.eventType === 'wait_completed') + ).toHaveLength(1); + }); + + it('should reject a duplicate wait_completed after the terminal reap', async () => { + // The `.completed` claim lives under `.locks/`, which the reap does not + // touch, so redelivery of the wait's continuation still conflicts. + const run = await createRun(storage, { + deploymentId: 'deployment-123', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); + await createWait(storage, run.runId, { + waitId: 'wait_reaped_twice', + resumeAt: new Date('2099-01-01'), + }); + await updateRun(storage, run.runId, 'run_completed', { + output: new Uint8Array([3]), + }); + + const waitCompleted = { + eventType: 'wait_completed' as const, + specVersion: SPEC_VERSION_CURRENT, + correlationId: 'wait_reaped_twice', + }; + await storage.events.create(run.runId, waitCompleted); + + await expect( + storage.events.create(run.runId, waitCompleted) + ).rejects.toMatchObject({ name: 'EntityConflictError' }); + }); + + it('should reject wait_completed on a terminal run for a wait that never existed', async () => { + // Without a `wait_created` claim there is nothing the reap could have + // removed — this is a completion for a wait that never was. + const run = await createRun(storage, { + deploymentId: 'deployment-123', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); + await updateRun(storage, run.runId, 'run_completed', { + output: new Uint8Array([3]), + }); + + await expect( + storage.events.create(run.runId, { + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: 'wait_never_created', + }) + ).rejects.toThrow(/not found/i); + }); + + it('should still reject wait_completed for a missing wait on a live run', async () => { + const run = await createRun(storage, { + deploymentId: 'deployment-123', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); + + await expect( + storage.events.create(run.runId, { + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: 'wait_absent', + }) + ).rejects.toThrow(/not found/i); + }); }); describe('disallowed operations on terminal runs', () => { diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index d9371dcab4..dfde4f6e95 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -508,6 +508,60 @@ function withInProcessLock( return task; } +/** + * Path of the exclusive-create claim that records one half of a wait's + * lifecycle. These live under `.locks/`, which the terminal reap + * (`deleteAllWaitsForRun`) does not touch, so they outlive the wait entity + * and remain the durable record of what already happened to the wait. + */ +function waitClaimPath( + basedir: string, + waitCompositeKey: string, + claim: 'created' | 'completed', + tag?: string +): string { + return resolveWithinBase( + basedir, + '.locks', + 'waits', + tag ? `${waitCompositeKey}.${claim}.${tag}` : `${waitCompositeKey}.${claim}` + ); +} + +/** + * Whether a `wait_created` claim exists for the wait. Used after the terminal + * reap has removed the entity, where the claim is the only remaining proof + * that the wait was ever created. Both the tagged and untagged paths are + * checked, mirroring `readJSONWithFallback`'s lookup of the entity itself. + */ +async function waitCreatedClaimExists( + basedir: string, + waitCompositeKey: string, + tag?: string +): Promise { + const candidates = tag + ? [ + waitClaimPath(basedir, waitCompositeKey, 'created', tag), + waitClaimPath(basedir, waitCompositeKey, 'created'), + ] + : [waitClaimPath(basedir, waitCompositeKey, 'created')]; + for (const claimPath of candidates) { + try { + await fs.access(claimPath); + return true; + } catch (error) { + // Only ENOENT proves the claim is absent. Anything else (EACCES, + // EMFILE) would make a wait that WAS created look like one that never + // was, turning a legitimate completion into a permanent rejection — + // propagate instead. + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + throw error; + } + } + } + return false; +} + /** * Helper function to delete all waits associated with a workflow run. * Called when a run reaches a terminal state. @@ -2502,17 +2556,8 @@ export function createEventsStorage( // surfaces as EntityConflictError (replaces a prior TOCTOU // read-then-check that could let both writers through). const waitCompositeKey = `${effectiveRunId}-${data.correlationId}`; - const waitCreatedLockName = tag - ? `${waitCompositeKey}.created.${tag}` - : `${waitCompositeKey}.created`; - const waitCreatedLockPath = resolveWithinBase( - basedir, - '.locks', - 'waits', - waitCreatedLockName - ); const waitCreatedClaimed = await writeExclusive( - waitCreatedLockPath, + waitClaimPath(basedir, waitCompositeKey, 'created', tag), '' ); if (!waitCreatedClaimed) { @@ -2542,14 +2587,11 @@ export function createEventsStorage( // Uses writeExclusive on a lock file to atomically prevent concurrent // invocations from both completing the same wait (TOCTOU race). const waitCompositeKey = `${effectiveRunId}-${data.correlationId}`; - const waitLockName = tag - ? `${waitCompositeKey}.completed.${tag}` - : `${waitCompositeKey}.completed`; - const lockPath = resolveWithinBase( + const lockPath = waitClaimPath( basedir, - '.locks', - 'waits', - waitLockName + waitCompositeKey, + 'completed', + tag ); const claimed = await writeExclusive(lockPath, ''); if (!claimed) { @@ -2565,25 +2607,49 @@ export function createEventsStorage( tag ); if (!existingWait) { - // Clean up the lock file we just claimed — the wait doesn't exist - await fs.unlink(lockPath).catch(() => {}); - throw new WorkflowWorldError( - `Wait "${data.correlationId}" not found` + // A terminal run has already had its waits reaped + // (`deleteAllWaitsForRun`), and the `run.resources-released` + // invariant says they must stay reaped. The wait's completion + // still has to reach the LOG though: a `wait_created` with no + // `wait_completed` reads as an open wait forever, so replay and + // observability both see a run that terminated while still + // sleeping. Record the event and mutate nothing — the run is + // terminal, so there is no entity left to agree with it. + // + // Dedup survives the reap: the `.completed` claim above lives + // under `.locks/`, which `deleteAllWaitsForRun` does not touch, + // so a duplicate completion still loses the claim and conflicts. + // Deliberately keep the claim (no unlink) on this path for the + // same reason. The sibling `.created` claim survives the reap too, + // so it — not the reaped entity — is what proves the wait ever + // existed; without it this is a completion for a wait that was + // never created and stays a rejection. + const reapedWaitCreated = + currentRun !== null && + isTerminalWorkflowRunStatus(currentRun.status) && + (await waitCreatedClaimExists(basedir, waitCompositeKey, tag)); + if (!reapedWaitCreated) { + // Clean up the lock file we just claimed — the wait doesn't exist + await fs.unlink(lockPath).catch(() => {}); + throw new WorkflowWorldError( + `Wait "${data.correlationId}" not found` + ); + } + } else { + // The lock file (writeExclusive above) already prevents concurrent + // completions — no additional status check needed. + wait = { + ...existingWait, + status: 'completed', + completedAt: now, + updatedAt: now, + }; + await writeJSON( + taggedPath(basedir, 'waits', waitCompositeKey, tag), + wait, + { overwrite: true } ); } - // The lock file (writeExclusive above) already prevents concurrent - // completions — no additional status check needed. - wait = { - ...existingWait, - status: 'completed', - completedAt: now, - updatedAt: now, - }; - await writeJSON( - taggedPath(basedir, 'waits', waitCompositeKey, tag), - wait, - { overwrite: true } - ); } // Note: hook_received events are stored in the event log but don't // modify the Hook entity (which doesn't have a payload field) diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts index 7d6c10f86d..9e55c7da72 100644 --- a/packages/world-postgres/src/storage.ts +++ b/packages/world-postgres/src/storage.ts @@ -319,6 +319,34 @@ async function reportSkippedSlots( }; } +/** + * What the event log says about a wait whose row has been reaped by the run's + * terminal transition. The log is the only surviving record at that point, so + * it is what a post-terminal `wait_completed` is validated against: `created` + * is the proof the wait existed, `completed` is the duplicate check the + * (deleted) row would otherwise have made. + */ +async function readReapedWaitLogState( + db: DrizzleLike, + runId: string, + correlationId: string +): Promise<{ created: boolean; completed: boolean }> { + const rows = await db + .select({ eventType: Schema.events.eventType }) + .from(Schema.events) + .where( + and( + eq(Schema.events.runId, runId), + eq(Schema.events.correlationId, correlationId), + inArray(Schema.events.eventType, ['wait_created', 'wait_completed']) + ) + ); + return { + created: rows.some((row) => row.eventType === 'wait_created'), + completed: rows.some((row) => row.eventType === 'wait_completed'), + }; +} + function getHookRetentionLimitMs(): number { const days = Number( process.env.WORKFLOW_POSTGRES_HOOK_RETENTION_LIMIT_DAYS ?? 30 @@ -2050,11 +2078,50 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] { waitId, }); if (!existing) { - throw new WorkflowWorldError( - `Wait "${data.correlationId}" not found` - ); - } - if (existing.status === 'completed') { + // A terminal run has already had its waits deleted (see the + // `isTerminalRunEventType` reap above), and they must stay + // deleted. The wait's completion still has to reach the LOG + // though: a `wait_created` with no `wait_completed` reads as an + // open wait forever, so replay and observability both see a run + // that terminated while still sleeping. Record the event and + // mutate nothing — the run is terminal, so there is no entity + // left to agree with it. + // + // Dedup moves to the log for the same reason: the row that + // normally rejects a duplicate completion is gone. Both halves + // are read — without a `wait_created` this is a completion for a + // wait that never existed, and stays a rejection. + // + // This read-then-write is not atomic (the partial unique index on + // `workflow_events` does not cover `wait_completed`), so two + // writers landing inside the same window could both record one. + // The window needs two deliveries of the same wait's continuation + // — one idempotency key, one message — to be in flight at the same + // instant, and the cost is a duplicate event on a run that will + // never replay for a decision again. Widening the index would mean + // rebuilding it as unique over existing history, which is a much + // worse trade than this. + const reaped = + currentRun && + isTerminalWorkflowRunStatus(currentRun.status) && + data.correlationId !== undefined + ? await readReapedWaitLogState( + drizzle, + effectiveRunId, + data.correlationId + ) + : { created: false, completed: false }; + if (reaped.completed) { + throw new EntityConflictError( + `Wait "${data.correlationId}" already completed` + ); + } + if (!reaped.created) { + throw new WorkflowWorldError( + `Wait "${data.correlationId}" not found` + ); + } + } else if (existing.status === 'completed') { throw new EntityConflictError( `Wait "${data.correlationId}" already completed` ); diff --git a/packages/world-postgres/test/storage.test.ts b/packages/world-postgres/test/storage.test.ts index 4882e17c00..13deaae55d 100644 --- a/packages/world-postgres/test/storage.test.ts +++ b/packages/world-postgres/test/storage.test.ts @@ -2876,6 +2876,65 @@ describe('Storage (Postgres integration)', () => { expect(result.status).toBe('failed'); }); + it('should allow wait_completed for a wait the terminal transition deleted', async () => { + // The wait's continuation fires after the run ended (a raced + // `Promise.race([sleep, hook])` whose hook won). Its completion still + // belongs in the log: a wait_created with no wait_completed reads as an + // open wait forever. The deleted row stays deleted, and a redelivery of + // the same continuation still conflicts. + const run = await createRun(events, { + deploymentId: 'deployment-123', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); + await events.create(run.runId, { + eventType: 'wait_created', + correlationId: 'wait_deleted', + eventData: { resumeAt: new Date('2099-01-01') }, + }); + await updateRun(events, run.runId, 'run_completed', { + output: new Uint8Array([1]), + }); + + const result = await events.create(run.runId, { + eventType: 'wait_completed', + correlationId: 'wait_deleted', + }); + expect(result.event.eventType).toBe('wait_completed'); + expect(result.wait).toBeUndefined(); + + const waitRows = await drizzle + .select() + .from(DrizzleSchema.waits) + .where(eq(DrizzleSchema.waits.runId, run.runId)); + expect(waitRows).toEqual([]); + + await expect( + events.create(run.runId, { + eventType: 'wait_completed', + correlationId: 'wait_deleted', + }) + ).rejects.toMatchObject({ name: 'EntityConflictError' }); + }); + + it('should reject wait_completed on a terminal run for a wait that never existed', async () => { + const run = await createRun(events, { + deploymentId: 'deployment-123', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); + await updateRun(events, run.runId, 'run_completed', { + output: new Uint8Array([1]), + }); + + await expect( + events.create(run.runId, { + eventType: 'wait_completed', + correlationId: 'wait_never_created', + }) + ).rejects.toThrow(/not found/i); + }); + it('should auto-delete hooks when run completes (postgres-specific behavior)', async () => { const run = await createRun(events, { deploymentId: 'deployment-123', diff --git a/packages/world-sim/src/store.test.ts b/packages/world-sim/src/store.test.ts index 944dea5435..4051d45799 100644 --- a/packages/world-sim/src/store.test.ts +++ b/packages/world-sim/src/store.test.ts @@ -389,6 +389,52 @@ describe('sim store', () => { }) ).resolves.toMatchObject({ run: { status: 'cancelled' } }); }); + + it('accepts the completion of a wait the run released, once', async () => { + // The wait's continuation fires after the run ended. Its completion + // still belongs in the log — a `wait_created` with no `wait_completed` + // reads as an open wait forever — but the released row stays released. + await store.events.create(RUN, { + eventType: 'wait_created', + specVersion: SPEC, + correlationId: 'wait_1', + eventData: { resumeAt: new Date('2099-01-01') }, + }); + await store.events.create(RUN, { + eventType: 'run_completed', + specVersion: SPEC, + eventData: {}, + }); + + const waitCompleted = { + eventType: 'wait_completed' as const, + specVersion: SPEC, + correlationId: 'wait_1', + }; + const result = await store.events.create(RUN, waitCompleted); + expect(result.event?.eventType).toBe('wait_completed'); + expect(result.wait).toBeUndefined(); + + await expect( + store.events.create(RUN, waitCompleted) + ).rejects.toBeInstanceOf(EntityConflictError); + }); + + it('rejects the completion of a wait that never existed', async () => { + await store.events.create(RUN, { + eventType: 'run_completed', + specVersion: SPEC, + eventData: {}, + }); + + await expect( + store.events.create(RUN, { + eventType: 'wait_completed', + specVersion: SPEC, + correlationId: 'wait_never', + }) + ).rejects.toThrow(/not found/i); + }); }); describe('pagination', () => { diff --git a/packages/world-sim/src/store.ts b/packages/world-sim/src/store.ts index 5244555b42..d5a850b422 100644 --- a/packages/world-sim/src/store.ts +++ b/packages/world-sim/src/store.ts @@ -1220,9 +1220,35 @@ export function createSimStore(options: SimStoreOptions): SimStore { case 'wait_completed': { const existing = waits.get(waitKey(runId, data.correlationId)); if (!existing) { - throw new WorkflowWorldError( - `Wait "${data.correlationId}" not found` - ); + // A terminal run has had its waits released + // (`releaseRunResources`), and the `run.resources-released` + // invariant says they stay released. The wait's completion still + // has to reach the LOG though: a `wait_created` with no + // `wait_completed` reads as an open wait forever, so replay and + // observability both see a run that terminated while still + // sleeping. Append the event; `applyEvent` is total and touches no + // row for a wait that isn't there. + // + // With the row gone the log is what both checks read: a + // `wait_created` is the proof the wait existed, a `wait_completed` + // is the duplicate the row would otherwise have caught. + const released = + currentRun && isTerminalWorkflowRunStatus(currentRun.status) + ? eventsForRun(runId).filter( + (e) => e.correlationId === data.correlationId + ) + : []; + if (released.some((e) => e.eventType === 'wait_completed')) { + throw new EntityConflictError( + `Wait "${data.correlationId}" already completed` + ); + } + if (!released.some((e) => e.eventType === 'wait_created')) { + throw new WorkflowWorldError( + `Wait "${data.correlationId}" not found` + ); + } + break; } if (existing.status === 'completed') { throw new EntityConflictError(