From 5bf27b9cfb77f643a3a7236b56398d90395ef3ea Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Mon, 10 Aug 2026 22:06:31 -0700 Subject: [PATCH] feat(world-local): support atomic start hooks --- .changeset/atomic-start-hook-local.md | 5 + packages/core/e2e/e2e.test.ts | 31 + packages/world-local/src/index.ts | 43 +- packages/world-local/src/storage.test.ts | 292 ++++++++- .../world-local/src/storage/events-storage.ts | 566 +++++++++++++----- packages/world-local/src/storage/helpers.ts | 67 ++- .../world-local/src/storage/hooks-storage.ts | 33 +- packages/world-local/src/test-helpers.ts | 9 +- 8 files changed, 841 insertions(+), 205 deletions(-) create mode 100644 .changeset/atomic-start-hook-local.md diff --git a/.changeset/atomic-start-hook-local.md b/.changeset/atomic-start-hook-local.md new file mode 100644 index 0000000000..3dfe5fe915 --- /dev/null +++ b/.changeset/atomic-start-hook-local.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-local': minor +--- + +Support atomic workflow admission with `start({ hook })` in the Local World. diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 18b0650099..35c2df9fc9 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -2129,6 +2129,37 @@ describe('e2e', () => { } ); + test.skipIf( + !isLocalDeployment() || + process.env.WORKFLOW_TARGET_WORLD === '@workflow/world-postgres' + )( + 'atomic start Hooks admit one concurrent run', + { timeout: 60_000 }, + async () => { + const token = `atomic-start-${Math.random().toString(36).slice(2)}`; + const workflow = await e2e('sleepingWorkflow'); + const results = await Promise.allSettled( + Array.from({ length: 6 }, () => + start(workflow, [5_000], { hook: { token } }) + ) + ); + const winner = results.find((result) => result.status === 'fulfilled'); + assert(winner?.status === 'fulfilled'); + + expect( + results.filter((result) => result.status === 'fulfilled') + ).toHaveLength(1); + for (const result of results) { + if (result.status === 'rejected') { + expect(HookConflictError.is(result.reason)).toBe(true); + assert(HookConflictError.is(result.reason)); + expect(result.reason.conflictingRunId).toBe(winner.value.runId); + } + } + await winner.value.returnValue; + } + ); + test( 'hookAdoptOwnerResultWorkflow - duplicate adopts the owner result via conflict.returnValue', { timeout: 120_000 }, diff --git a/packages/world-local/src/index.ts b/packages/world-local/src/index.ts index b89fd67af5..ae2952ce69 100644 --- a/packages/world-local/src/index.ts +++ b/packages/world-local/src/index.ts @@ -18,7 +18,11 @@ import { import { initDataDir } from './init.js'; import { instrumentObject } from './instrumentObject.js'; import { createQueue, type DirectHandler } from './queue.js'; -import { hashToken, hookRecoveryMarkerPath } from './storage/helpers.js'; +import { + hookRecoveryMarkerPath, + releaseHookTokenClaimIfOwnedBy, + StartHookAdmissionSchema, +} from './storage/helpers.js'; import { resetHookIndexEnsureCache } from './storage/hook-index.js'; import { createStorage } from './storage.js'; import { createStreamer } from './streamer.js'; @@ -75,6 +79,7 @@ export function createWorld(args?: Partial): LocalWorld { specVersion: SPEC_VERSION_CURRENT, capabilities: { hookRetention: { active: true }, + atomicStartHook: { active: true }, // world-local deduplicates concurrent `hook_received` writes sharing a // `(runId, resumeId)` via a filesystem sidecar claim (see // events-storage.ts `claimHookResume`), so resumeHook()'s parallel fast @@ -128,12 +133,8 @@ export function createWorld(args?: Partial): LocalWorld { // Selectively delete only files matching this tag const basedir = mergedConfig.dataDir; - // Delete hook token constraint files (and recovery markers, - // for disk hygiene) BEFORE deleting the hooks, since we need - // to read each hook to extract its token hash. Constraint - // files and markers are untagged (`{sha256}.json` and - // `{sha256}.recovery.json`) so listTaggedFiles won't find - // them — we must resolve them via the hook data. + // Claims and recovery markers are untagged, so release them through + // their tagged Hook or admission before deleting tagged entities. const hooksDir = path.join(basedir, 'hooks'); const taggedHookFiles = await listTaggedFiles(hooksDir, tag); const { HookSchema } = await import('@workflow/world'); @@ -144,9 +145,11 @@ export function createWorld(args?: Partial): LocalWorld { HookSchema ); if (hook?.token) { - await deleteJSON( - path.join(hooksDir, 'tokens', `${hashToken(hook.token)}.json`) - ); + await releaseHookTokenClaimIfOwnedBy(basedir, hook.token, { + runId: hook.runId, + hookId: hook.hookId, + tag, + }); await deleteJSON( hookRecoveryMarkerPath( basedir, @@ -159,6 +162,26 @@ export function createWorld(args?: Partial): LocalWorld { }) ); + const admissionsDir = path.join(basedir, 'hooks', 'admissions'); + const taggedAdmissionFiles = await listTaggedFiles(admissionsDir, tag); + await Promise.all( + taggedAdmissionFiles.map(async (admissionFile) => { + const admissionPath = path.join(admissionsDir, admissionFile); + const admission = await readJSON( + admissionPath, + StartHookAdmissionSchema + ); + if (admission && !('redirectRunId' in admission)) { + await releaseHookTokenClaimIfOwnedBy(basedir, admission.token, { + runId: admission.runId, + eventId: admission.eventId, + tag, + }); + } + await deleteJSON(admissionPath); + }) + ); + // Delete tagged entity files across all directories const entityDirs = [ 'runs', diff --git a/packages/world-local/src/storage.test.ts b/packages/world-local/src/storage.test.ts index db89c5a2cf..5385714e2d 100644 --- a/packages/world-local/src/storage.test.ts +++ b/packages/world-local/src/storage.test.ts @@ -2,7 +2,7 @@ import assert from 'node:assert/strict'; import { promises as fs } from 'node:fs'; import os from 'node:os'; import path from 'node:path'; -import { WorkflowWorldError } from '@workflow/errors'; +import { HookConflictError, WorkflowWorldError } from '@workflow/errors'; import type { Event, Storage } from '@workflow/world'; import { SPEC_VERSION_CURRENT, stripEventDataRefs } from '@workflow/world'; import { monotonicFactory } from 'ulid'; @@ -14,6 +14,7 @@ import { hashToken, hookDisposeLockPath, hookTokenClaimPath, + readStartHookAdmission, runTerminalMarkerPath, withHookTokenClaimLock, } from './storage/helpers.js'; @@ -333,6 +334,254 @@ describe('Storage', () => { }); }); + describe('atomic start Hooks', () => { + const runId = () => `wrun_${monotonicFactory()()}`; + const admit = ( + { + candidate, + token, + tokenRetentionUntil, + }: { + candidate: string; + token: string; + tokenRetentionUntil?: Date; + }, + worker: Storage = storage + ) => + worker.events.create(candidate, { + eventType: 'run_created', + specVersion: SPEC_VERSION_CURRENT, + eventData: { + deploymentId: 'deployment-123', + workflowName: 'atomic-start-workflow', + input: new Uint8Array(), + startHook: { token, tokenRetentionUntil }, + }, + }); + + it('admits one run across storage instances', async () => { + const token = 'atomic-start-race'; + const candidates = Array.from({ length: 8 }, () => runId()); + const results = await Promise.allSettled( + candidates.map((candidate) => + admit({ candidate, token }, createStorage(testDir)) + ) + ); + const winner = results.find((result) => result.status === 'fulfilled'); + assert(winner?.status === 'fulfilled'); + const winnerRunId = winner.value.run?.runId; + assert(winnerRunId); + + expect( + results.filter((result) => result.status === 'fulfilled') + ).toHaveLength(1); + for (const [index, result] of results.entries()) { + const decision = await readStartHookAdmission( + testDir, + candidates[index] + ); + if (result.status === 'fulfilled') { + expect(decision?.redirectRunId).toBeUndefined(); + continue; + } + expect(HookConflictError.is(result.reason)).toBe(true); + expect(result.reason.conflictingRunId).toBe(winnerRunId); + expect(decision?.redirectRunId).toBe(winnerRunId); + await expect( + storage.runs.get(candidates[index]) + ).rejects.toMatchObject({ name: 'WorkflowRunNotFoundError' }); + expect( + (await storage.events.list({ runId: candidates[index] })).data + ).toHaveLength(0); + } + }); + + it('releases the losing token when a candidate is reused', async () => { + const candidate = runId(); + const tokens = ['atomic-start-token-a', 'atomic-start-token-b']; + const results = await Promise.allSettled( + tokens.map((token) => + admit({ candidate, token }, createStorage(testDir)) + ) + ); + const decision = await readStartHookAdmission(testDir, candidate); + assert(decision); + const winnerIndex = tokens.indexOf(decision.token); + const loserIndex = winnerIndex === 0 ? 1 : 0; + + expect(results[winnerIndex].status).toBe('fulfilled'); + expect(results[loserIndex]).toMatchObject({ + status: 'rejected', + reason: { name: 'EntityConflictError' }, + }); + await expect( + admit({ candidate: runId(), token: tokens[loserIndex] }) + ).resolves.toHaveProperty('run'); + }); + + it('reclaims a reservation abandoned before admission', async () => { + const token = 'atomic-start-abandoned-reservation'; + const claimPath = hookTokenClaimPath(testDir, token); + await fs.mkdir(path.dirname(claimPath), { recursive: true }); + await fs.writeFile( + claimPath, + JSON.stringify({ + token, + runId: runId(), + eventId: 'evnt_abandoned', + }) + ); + + await expect( + admit({ candidate: runId(), token }) + ).resolves.toHaveProperty('run'); + }); + + it('keeps a committed reservation while its run is unpublished', async () => { + const token = 'atomic-start-committed-reservation'; + const ownerRunId = runId(); + await admit({ candidate: ownerRunId, token }); + await fs.unlink(path.join(testDir, 'runs', `${ownerRunId}.json`)); + + await expect( + admit({ candidate: runId(), token }) + ).rejects.toMatchObject({ + name: 'HookConflictError', + conflictingRunId: ownerRunId, + }); + }); + + it('replays a rejected candidate after the token is reused', async () => { + const token = 'atomic-start-redirect'; + const ownerRunId = runId(); + const rejectedRunId = runId(); + await admit({ candidate: ownerRunId, token }); + await expect( + admit({ candidate: rejectedRunId, token }) + ).rejects.toMatchObject({ + name: 'HookConflictError', + conflictingRunId: ownerRunId, + }); + + await updateRun(storage, ownerRunId, 'run_started'); + await updateRun(storage, ownerRunId, 'run_completed', { + output: new Uint8Array(), + }); + await admit({ candidate: runId(), token }); + + await expect( + admit({ candidate: rejectedRunId, token }) + ).rejects.toMatchObject({ + name: 'HookConflictError', + conflictingRunId: ownerRunId, + }); + await expect(storage.runs.get(rejectedRunId)).rejects.toMatchObject({ + name: 'WorkflowRunNotFoundError', + }); + }); + + it('converges direct and resilient admission for one candidate', async () => { + const token = 'atomic-start-replay'; + const candidate = runId(); + const runCreationData = { + deploymentId: 'deployment-123', + workflowName: 'atomic-start-workflow', + input: new Uint8Array(), + startHook: { token }, + }; + const direct = createStorage(testDir).events.create(candidate, { + eventType: 'run_created', + specVersion: SPEC_VERSION_CURRENT, + eventData: runCreationData, + }); + const queued = createStorage(testDir).events.create(candidate, { + eventType: 'run_started', + specVersion: SPEC_VERSION_CURRENT, + eventData: runCreationData, + }); + + await expect(Promise.all([direct, queued])).resolves.toHaveLength(2); + expect((await storage.runs.get(candidate)).status).toBe('running'); + const events = await storage.events.list({ runId: candidate }); + expect( + events.data.filter((event) => event.eventType === 'run_created') + ).toHaveLength(1); + }); + + it('materializes a reservation and preserves its retention', async () => { + const token = 'atomic-start-materialize'; + const ownerRunId = runId(); + const tokenRetentionUntil = new Date(Date.now() + 60_000); + await admit({ candidate: ownerRunId, token, tokenRetentionUntil }); + + await expect(storage.hooks.getByToken(token)).rejects.toMatchObject({ + name: 'HookNotFoundError', + }); + const hook = await createHook(storage, ownerRunId, { + hookId: 'hook_materialized', + token, + tokenRetentionUntil: new Date(Date.now() + 1_000), + }); + expect(hook.tokenRetentionUntil).toEqual(tokenRetentionUntil); + + await disposeHook(storage, ownerRunId, hook.hookId); + await expect( + admit({ candidate: runId(), token }) + ).resolves.toHaveProperty('run'); + }); + + it('retains a reservation across tags until its deadline', async () => { + const token = 'atomic-start-retention'; + const owner = createStorage(testDir, 'vitest-0'); + const claimant = createStorage(testDir, 'vitest-1'); + const ownerRunId = runId(); + const tokenRetentionUntil = new Date(Date.now() + 100); + await admit( + { candidate: ownerRunId, token, tokenRetentionUntil }, + owner + ); + await updateRun(owner, ownerRunId, 'run_started'); + await updateRun(owner, ownerRunId, 'run_completed', { + output: new Uint8Array(), + }); + + await expect( + admit({ candidate: runId(), token }, claimant) + ).rejects.toMatchObject({ + name: 'HookConflictError', + conflictingRunId: ownerRunId, + }); + await new Promise((resolve) => + setTimeout( + resolve, + Math.max(0, tokenRetentionUntil.getTime() - Date.now() + 1) + ) + ); + await expect( + admit({ candidate: runId(), token }, claimant) + ).resolves.toHaveProperty('run'); + }); + + it('rejects start Hook retention beyond the Local limit', async () => { + const candidate = runId(); + await expect( + admit({ + candidate, + token: 'atomic-start-over-retention-limit', + tokenRetentionUntil: new Date( + Date.now() + 31 * 24 * 60 * 60 * 1000 + ), + }) + ).rejects.toMatchObject({ + name: 'WorkflowWorldError', + status: 400, + }); + await expect(storage.runs.get(candidate)).rejects.toMatchObject({ + name: 'WorkflowRunNotFoundError', + }); + }); + }); + describe('update via events', () => { it('should update run status to running via run_started event', async () => { const created = await createRun(storage, { @@ -2532,6 +2781,25 @@ describe('Storage', () => { expect(result.hook?.hookId).toBe('hook_new'); }); + it('releases an ownerless materialized Hook claim', async () => { + const token = 'ownerless-materialized-claim'; + const claimPath = hookTokenClaimPath(testDir, token); + await fs.mkdir(path.dirname(claimPath), { recursive: true }); + await fs.writeFile( + claimPath, + JSON.stringify({ + token, + hookId: 'hook_missing', + runId: 'wrun_missing', + eventId: 'evnt_missing', + }) + ); + + await expect( + createHook(storage, testRunId, { hookId: 'hook_new', token }) + ).resolves.toMatchObject({ hookId: 'hook_new' }); + }); + it('should allow multiple hooks with different tokens for the same run', async () => { const hook1 = await createHook(storage, testRunId, { hookId: 'hook_1', @@ -2728,36 +2996,42 @@ describe('Storage', () => { // instead of recording a hook_conflict against a finished run. The // claim file is rewritten manually to simulate the window where the // run-completion cleanup has not deleted it yet (or crashed). - it('should treat a token claim owned by a terminal run as vacant', async () => { + it('should release a legacy tagged claim after its run ends', async () => { const token = 'terminal-owner-token'; + const taggedStorage = createStorage(testDir, 'vitest-0'); + const owner = await createRun(taggedStorage, { + deploymentId: 'deployment-123', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); - await createHook(storage, testRunId, { + await createHook(taggedStorage, owner.runId, { hookId: 'hook_1', token, }); - await updateRun(storage, testRunId, 'run_started'); - await updateRun(storage, testRunId, 'run_completed', { + await updateRun(taggedStorage, owner.runId, 'run_started'); + await updateRun(taggedStorage, owner.runId, 'run_completed', { output: new Uint8Array(), }); - // Simulate the stale claim surviving the terminal-run cleanup + // Simulate a pre-tag claim surviving terminal cleanup. await fs.writeFile( path.join(testDir, 'hooks', 'tokens', `${hashToken(token)}.json`), JSON.stringify({ token, hookId: 'hook_1', - runId: testRunId, + runId: owner.runId, eventId: 'evnt_00000000000000000000000000', }) ); - const run2 = await createRun(storage, { + const run2 = await createRun(taggedStorage, { deploymentId: 'deployment-456', workflowName: 'another-workflow', input: new Uint8Array(), }); - const result = await storage.events.create(run2.runId, { + const result = await taggedStorage.events.create(run2.runId, { eventType: 'hook_created', correlationId: 'hook_2', eventData: { token }, diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index fae85024e8..8a5690cd21 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -3,6 +3,7 @@ import fs from 'node:fs/promises'; import path from 'node:path'; import { EntityConflictError, + HookConflictError, HookNotFoundError, RunExpiredError, RunNotSupportedError, @@ -21,7 +22,8 @@ import type { PaginatedResponse, PaginationOptions, ResolveData, - SerializedData, + RunCreationData, + StartHook, Step, Storage, Wait, @@ -87,10 +89,13 @@ import { monotonicUlid, pendingHookEventPath, readHookTokenClaim, + readStartHookAdmission, reapPendingHookEvents, releaseHookTokenClaimIfOwnedBy, runTerminalMarkerPath, + type StartHookAdmission, scanRunEventIds, + startHookAdmissionPath, withHookTokenClaimLock, } from './helpers.js'; import { @@ -237,27 +242,21 @@ async function findCommittedResumeEvent( return null; } /** - * Whether a token claim held by another `(runId, hookId)` can never become - * live again and may therefore be released by a new claimant: - * - * - the claimed hook's disposal is committed (its dispose lock exists — - * the durable release of the claim file just hasn't landed yet, or was - * lost to a crash between the lock write and the claim delete), or - * - the owning run is terminal and its minimum retention has ended, or - * - the owning run does not exist (a claim can only be written during a - * suspension of an existing run, so an ownerless claim is debris). - * - * A claim from a mid-creation writer is never releasable: its owning run - * exists and is non-terminal, and its dispose lock does not exist. + * A claim may be replaced after disposal, or after its run and retention end. + * A missing run makes a legacy claim debris, but can mean that a modern + * reservation is still admitting its run. */ async function isHookTokenClaimReleasable( basedir: string, claim: HookTokenClaim, tag?: string ): Promise { + // Claims written before `tag` was stored retain the previous caller-tag + // lookup behavior. + const ownerTag = claim.tag ?? tag; if ( claim.hookId && - (await isHookDisposalCommitted(basedir, claim.hookId, tag)) + (await isHookDisposalCommitted(basedir, claim.hookId, ownerTag)) ) { return true; } @@ -266,10 +265,25 @@ async function isHookTokenClaimReleasable( 'runs', claim.runId, WorkflowRunSchema, - tag + ownerTag ); if (!owningRun) { - return true; + if (claim.hookId !== undefined || claim.eventId === undefined) { + return true; + } + // An accepted admission can briefly exist before its run. Without one, + // the reservation was abandoned before its decision was committed. + const admission = await readStartHookAdmission( + basedir, + claim.runId, + ownerTag + ); + return ( + !admission || + 'redirectRunId' in admission || + admission.eventId !== claim.eventId || + (claim.token !== undefined && admission.token !== claim.token) + ); } if (!isTerminalWorkflowRunStatus(owningRun.status)) { return false; @@ -280,6 +294,182 @@ async function isHookTokenClaimReleasable( ); } +async function deleteReleasableHookClaim({ + basedir, + token, + claim, + signal, + tag, +}: { + basedir: string; + token: string; + claim: HookTokenClaim; + signal: AbortSignal; + tag?: string; +}): Promise { + const ownerTag = claim.tag ?? tag; + signal.throwIfAborted(); + await deleteJSON(hookTokenClaimPath(basedir, token)); + if (!claim.hookId) return; + + await deleteJSON( + hookRecoveryMarkerPath(basedir, token, claim.runId, claim.hookId) + ); + await deleteJSON(taggedPath(basedir, 'hooks', claim.hookId, ownerTag)); + // Keep the marker until the Hook is gone so terminal cleanup can retry. + await deleteHookByRunMarker(basedir, claim.runId, claim.hookId, ownerTag); +} + +function resolveStartHookAdmission( + admission: StartHookAdmission, + token: string +): string { + if (admission.token !== token) { + throw new EntityConflictError( + `Workflow run "${admission.runId}" already uses a different start Hook token` + ); + } + if ('redirectRunId' in admission) { + throw new HookConflictError(admission.token, admission.redirectRunId); + } + return admission.eventId; +} + +/** + * Records one immutable decision per candidate run. The token claim chooses + * the winner; the admission record makes direct and queued copies replay that + * same result even if the token later becomes available. + */ +async function admitStartHook({ + basedir, + runId, + eventId, + startHook, + tag, +}: { + basedir: string; + runId: string; + eventId: string; + startHook: StartHook; + tag?: string; +}): Promise { + const replay = await readStartHookAdmission(basedir, runId, tag); + if (replay) { + return resolveStartHookAdmission(replay, startHook.token); + } + + return withHookTokenClaimLock(basedir, startHook.token, async (signal) => { + const replay = await readStartHookAdmission(basedir, runId, tag); + if (replay) { + return resolveStartHookAdmission(replay, startHook.token); + } + + const claimPath = hookTokenClaimPath(basedir, startHook.token); + const admissionPath = startHookAdmissionPath(basedir, runId, tag); + const commitAdmission = async (admission: StartHookAdmission) => { + signal.throwIfAborted(); + const created = await writeExclusive( + admissionPath, + JSON.stringify(admission) + ); + const stored = created + ? admission + : await readStartHookAdmission(basedir, runId, tag); + assert(stored); + if ('eventId' in admission && stored.token !== startHook.token) { + await deleteJSON(claimPath); + } + return resolveStartHookAdmission(stored, startHook.token); + }; + let claim = await readHookTokenClaim(claimPath); + if (!claim) { + signal.throwIfAborted(); + await deleteJSON(claimPath); + await rebuildLiveHookByTokenFromEventLog(basedir, startHook.token, tag); + claim = await readHookTokenClaim(claimPath); + } + + if ( + claim?.runId === runId && + claim.tag === tag && + claim.hookId === undefined && + claim.eventId + ) { + const admission: StartHookAdmission = { + token: startHook.token, + runId, + eventId: claim.eventId, + tokenRetentionUntil: claim.tokenRetentionUntil, + }; + return commitAdmission(admission); + } + + if (claim && (await isHookTokenClaimReleasable(basedir, claim, tag))) { + await deleteReleasableHookClaim({ + basedir, + token: startHook.token, + claim, + signal, + tag, + }); + claim = null; + } + + const admission: StartHookAdmission = claim + ? { token: startHook.token, runId, redirectRunId: claim.runId } + : { + token: startHook.token, + runId, + eventId, + tokenRetentionUntil: startHook.tokenRetentionUntil, + }; + + if (!claim) { + signal.throwIfAborted(); + assert( + await writeExclusive( + claimPath, + JSON.stringify({ + token: startHook.token, + runId, + eventId, + tokenRetentionUntil: startHook.tokenRetentionUntil, + tag, + } satisfies HookTokenClaim) + ) + ); + } + + return commitAdmission(admission); + }); +} + +function createPendingRun({ + runId, + specVersion, + createdAt, + data, +}: { + runId: string; + specVersion: number; + createdAt: Date; + data: RunCreationData; +}): WorkflowRun { + return { + runId, + deploymentId: data.deploymentId, + status: 'pending', + workflowName: data.workflowName, + specVersion, + executionContext: data.executionContext, + input: data.input, + attributes: data.attributes ?? {}, + encryptionPublicKey: data.encryptionPublicKey, + createdAt, + updatedAt: createdAt, + }; +} + async function readHookRecoveryMarker( markerPath: string ): Promise | null> { @@ -904,11 +1094,15 @@ export function createEventsStorage( data: AnyEventRequest, params?: CreateEventParams ): Promise { + const tokenRetentionUntil = + data.eventType === 'hook_created' + ? data.eventData.tokenRetentionUntil + : data.eventType === 'run_created' || data.eventType === 'run_started' + ? data.eventData?.startHook?.tokenRetentionUntil + : undefined; if ( - data.eventType === 'hook_created' && - data.eventData.tokenRetentionUntil !== undefined && - data.eventData.tokenRetentionUntil.getTime() > - Date.now() + hookRetentionLimitMs + tokenRetentionUntil && + tokenRetentionUntil.getTime() > Date.now() + hookRetentionLimitMs ) { throw new WorkflowWorldError( `Hook minimum retention cannot exceed ${hookRetentionLimitMs / DAY_MS} days in the Local World.`, @@ -1045,97 +1239,97 @@ export function createEventsStorage( if ( data.eventType === 'run_started' && !currentRun && - 'eventData' in data && - data.eventData + data.eventData?.deploymentId && + data.eventData.workflowName && + data.eventData.input !== undefined ) { - const runInputData = data.eventData as { - deploymentId?: string; - workflowName?: string; - input?: any; - executionContext?: Record; - attributes?: Record; - allowReservedAttributes?: true; - encryptionPublicKey?: string; + const runCreationData: RunCreationData = { + ...data.eventData, + deploymentId: data.eventData.deploymentId, + workflowName: data.eventData.workflowName, + input: data.eventData.input, }; - if ( - runInputData.deploymentId && - runInputData.workflowName && - runInputData.input !== undefined - ) { - validateAttributeChanges( - Object.entries(runInputData.attributes ?? {}).map( - ([key, value]) => ({ key, value }) - ), - { - allowReservedAttributes: - runInputData.allowReservedAttributes === true, - } - ); - // Atomically try to publish the run entity so only the first - // writer wins, preventing a TOCTOU race where a concurrent - // run_created from start() could overwrite a run that was - // already transitioned to 'running'. - const createdRun: WorkflowRun = { + validateAttributeChanges( + Object.entries(runCreationData.attributes ?? {}).map( + ([key, value]) => ({ key, value }) + ), + { + allowReservedAttributes: + runCreationData.allowReservedAttributes === true, + } + ); + let runCreatedEventId = await mintEventId(effectiveRunId); + if (runCreationData.startHook) { + runCreatedEventId = await admitStartHook({ + basedir, runId: effectiveRunId, - deploymentId: runInputData.deploymentId, - status: 'pending', - workflowName: runInputData.workflowName, + eventId: runCreatedEventId, + startHook: runCreationData.startHook, + tag, + }); + } + const runCreatedAt = + ulidToDate(runCreatedEventId.replace(/^evnt_/, '')) ?? now; + const createdRun = createPendingRun({ + runId: effectiveRunId, + specVersion: effectiveSpecVersion, + createdAt: runCreatedAt, + data: runCreationData, + }); + const runPath = taggedPath(basedir, 'runs', effectiveRunId, tag); + const created = await writeExclusive( + runPath, + JSON.stringify(createdRun, jsonReplacer) + ); + + if (created) { + const runCreatedEvent: Event = { + eventType: 'run_created', + runId: effectiveRunId, + eventId: runCreatedEventId, + createdAt: runCreatedAt, specVersion: effectiveSpecVersion, - executionContext: runInputData.executionContext, - input: runInputData.input, - output: undefined, - error: undefined, - startedAt: undefined, - completedAt: undefined, - attributes: runInputData.attributes ?? {}, - // Must be mirrored here too: this is the path that recreates a - // run from the queued message, which is exactly when the key - // would otherwise be lost for the rest of the run's life. - encryptionPublicKey: runInputData.encryptionPublicKey, - createdAt: now, - updatedAt: now, + eventData: runCreationData, }; - const runPath = taggedPath(basedir, 'runs', effectiveRunId, tag); - const created = await writeExclusive( - runPath, - JSON.stringify(createdRun, jsonReplacer) + const runCreatedEventPath = taggedPath( + basedir, + 'events', + `${effectiveRunId}-${runCreatedEventId}`, + tag ); - - if (created) { - // We created the run — also write the run_created event. - // Drawn before this invocation's own id so it takes the - // earlier slot: it must replay first. - const runCreatedEventId = await mintEventId(effectiveRunId); - const runCreatedEvent: Event = { - eventType: 'run_created', - runId: effectiveRunId, - eventId: runCreatedEventId, - createdAt: now, - specVersion: effectiveSpecVersion, - eventData: { - deploymentId: runInputData.deploymentId, - workflowName: runInputData.workflowName, - input: runInputData.input, - executionContext: runInputData.executionContext, - attributes: runInputData.attributes, - allowReservedAttributes: - runInputData.allowReservedAttributes, - encryptionPublicKey: runInputData.encryptionPublicKey, - }, - }; - await storeEvent(runCreatedEvent); - currentRun = createdRun; + const serializedRunCreatedEvent = JSON.stringify( + runCreatedEvent, + jsonReplacer, + 2 + ); + const published = await writeExclusive( + runCreatedEventPath, + serializedRunCreatedEvent + ); + if (published) { + rememberStoredEvent( + runCreatedEvent, + runCreatedEventPath, + serializedRunCreatedEvent + ); } else { - // Run already exists (concurrent run_created won the - // race). Re-read it so downstream logic sees the real state. - currentRun = await readJSONWithFallback( - basedir, - 'runs', - effectiveRunId, - WorkflowRunSchema, - tag + const persistedEvent = await readJSON( + runCreatedEventPath, + EventSchema ); + assert(persistedEvent?.eventType === 'run_created'); } + notePublishedSlot(effectiveRunId, runCreatedEventId); + currentRun = createdRun; + } else { + // Concurrent direct admission created the same run. + currentRun = await readJSONWithFallback( + basedir, + 'runs', + effectiveRunId, + WorkflowRunSchema, + tag + ); } } } @@ -1640,15 +1834,7 @@ export function createEventsStorage( // Create/update entity based on event type (event-sourced architecture) // Run lifecycle events if (data.eventType === 'run_created' && 'eventData' in data) { - const runData = data.eventData as { - deploymentId: string; - workflowName: string; - input: SerializedData; - executionContext?: Record; - attributes?: Record; - allowReservedAttributes?: true; - encryptionPublicKey?: string; - }; + const runData = data.eventData; validateAttributeChanges( Object.entries(runData.attributes ?? {}).map(([key, value]) => ({ key, @@ -1658,24 +1844,27 @@ export function createEventsStorage( allowReservedAttributes: runData.allowReservedAttributes === true, } ); - run = { + if (runData.startHook) { + eventId = await admitStartHook({ + basedir, + runId: effectiveRunId, + eventId, + startHook: runData.startHook, + tag, + }); + eventIdPinned = true; + event = { + ...event, + eventId, + createdAt: ulidToDate(eventId.replace(/^evnt_/, '')) ?? now, + }; + } + run = createPendingRun({ runId: effectiveRunId, - deploymentId: runData.deploymentId, - status: 'pending', - workflowName: runData.workflowName, - // Propagate specVersion from the event to the run entity specVersion: effectiveSpecVersion, - executionContext: runData.executionContext, - input: runData.input, - output: undefined, - error: undefined, - startedAt: undefined, - completedAt: undefined, - attributes: runData.attributes ?? {}, - encryptionPublicKey: runData.encryptionPublicKey, - createdAt: now, - updatedAt: now, - }; + createdAt: event.createdAt, + data: runData, + }); // Atomically publish the run entity file without overwriting an // existing winner. This prevents a TOCTOU race with the resilient // start path (run_started on non-existent run) that could result in @@ -1686,9 +1875,21 @@ export function createEventsStorage( JSON.stringify(run, jsonReplacer, 2) ); if (!created) { - throw new EntityConflictError( - `Workflow run "${effectiveRunId}" already exists` - ); + const existingRun = runData.startHook + ? await readJSONWithFallback( + basedir, + 'runs', + effectiveRunId, + WorkflowRunSchema, + tag + ) + : null; + if (!existingRun) { + throw new EntityConflictError( + `Workflow run "${effectiveRunId}" already exists` + ); + } + run = existingRun; } } else if (data.eventType === 'run_started') { // Reuse currentRun from validation (already read above) @@ -1767,7 +1968,7 @@ export function createEventsStorage( } ); await Promise.all([ - deleteAllHooksForRun(basedir, effectiveRunId), + deleteAllHooksForRun(basedir, effectiveRunId, tag), deleteAllWaitsForRun(basedir, effectiveRunId), ]); } @@ -1806,7 +2007,7 @@ export function createEventsStorage( } ); await Promise.all([ - deleteAllHooksForRun(basedir, effectiveRunId), + deleteAllHooksForRun(basedir, effectiveRunId, tag), deleteAllWaitsForRun(basedir, effectiveRunId), ]); } @@ -1837,7 +2038,7 @@ export function createEventsStorage( } ); await Promise.all([ - deleteAllHooksForRun(basedir, effectiveRunId), + deleteAllHooksForRun(basedir, effectiveRunId, tag), deleteAllWaitsForRun(basedir, effectiveRunId), ]); } @@ -2218,7 +2419,8 @@ export function createEventsStorage( runId: effectiveRunId, eventId, tokenRetentionUntil: hookData.tokenRetentionUntil, - }); + tag, + } satisfies HookTokenClaim); // Serialize claim replacement so a committed disposal or terminal // run cannot race its successor and create a spurious conflict @@ -2248,10 +2450,36 @@ export function createEventsStorage( } if ( existingClaim.runId === effectiveRunId && - existingClaim.hookId === data.correlationId + existingClaim.hookId === data.correlationId && + (existingClaim.tag === undefined || existingClaim.tag === tag) ) { return { status: 'owned' as const, claim: existingClaim }; } + if ( + existingClaim.runId === effectiveRunId && + existingClaim.tag === tag && + existingClaim.hookId === undefined && + existingClaim.eventId + ) { + const tokenRetentionUntil = + existingClaim.tokenRetentionUntil && + (!hookData.tokenRetentionUntil || + existingClaim.tokenRetentionUntil > + hookData.tokenRetentionUntil) + ? existingClaim.tokenRetentionUntil + : hookData.tokenRetentionUntil; + const claim = { + token: hookData.token, + hookId: data.correlationId, + runId: effectiveRunId, + eventId, + tokenRetentionUntil, + tag: existingClaim.tag ?? tag, + } satisfies HookTokenClaim; + signal.throwIfAborted(); + await writeJSON(constraintPath, claim, { overwrite: true }); + return { status: 'materialized' as const, claim }; + } if ( !(await isHookTokenClaimReleasable(basedir, existingClaim, tag)) ) { @@ -2260,27 +2488,13 @@ export function createEventsStorage( // The previous owner committed its release but did not finish // cleanup. Remove that lifetime before admitting a successor. - signal.throwIfAborted(); - await deleteJSON(constraintPath); - if (existingClaim.hookId) { - await deleteJSON( - taggedPath(basedir, 'hooks', existingClaim.hookId, tag) - ); - await deleteJSON( - hookRecoveryMarkerPath( - basedir, - hookData.token, - existingClaim.runId, - existingClaim.hookId - ) - ); - await deleteHookByRunMarker( - basedir, - existingClaim.runId, - existingClaim.hookId, - tag - ); - } + await deleteReleasableHookClaim({ + basedir, + token: hookData.token, + claim: existingClaim, + signal, + tag, + }); signal.throwIfAborted(); assert(await writeExclusive(constraintPath, claimContent)); return { status: 'claimed' as const }; @@ -2359,6 +2573,20 @@ export function createEventsStorage( }; } + if (claimResult.status === 'materialized') { + event = { + ...data, + eventData: { + ...data.eventData, + tokenRetentionUntil: claimResult.claim.tokenRetentionUntil, + }, + runId: effectiveRunId, + eventId, + createdAt: event.createdAt, + specVersion: effectiveSpecVersion, + }; + } + if (claimResult.status === 'conflict') { const existingClaim = claimResult.claim; const conflictEvent: Event = { @@ -2465,12 +2693,11 @@ export function createEventsStorage( // includes `(token, runId, hookId)` so different // lifetimes never collide, but cleaning up reduces disk // leak for hooks that go through the recovery path. - await releaseHookTokenClaimIfOwnedBy( - basedir, - existingHook.token, - existingHook.runId, - existingHook.hookId - ); + await releaseHookTokenClaimIfOwnedBy(basedir, existingHook.token, { + runId: existingHook.runId, + hookId: existingHook.hookId, + tag, + }); await deleteJSON( hookRecoveryMarkerPath( basedir, @@ -2791,6 +3018,21 @@ export function createEventsStorage( } if (!eventPublished) { + if ( + data.eventType === 'run_created' && + data.eventData.startHook && + run + ) { + const persistedEvent = await readJSON(eventPath, EventSchema); + assert(persistedEvent?.eventType === 'run_created'); + const resolveData = + params?.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION; + return { + event: stripEventDataRefs(persistedEvent, resolveData), + run, + maxEvents: getMaxEventsPerRun(), + }; + } // For `hook_created`, losing the event publish means the // event was already committed at this exact (canonical) // path. The original publisher may have crashed between diff --git a/packages/world-local/src/storage/helpers.ts b/packages/world-local/src/storage/helpers.ts index 9d599dc670..7ecee6ee30 100644 --- a/packages/world-local/src/storage/helpers.ts +++ b/packages/world-local/src/storage/helpers.ts @@ -2,7 +2,7 @@ import { createHash } from 'node:crypto'; import fs from 'node:fs/promises'; import path from 'node:path'; import { WorkflowWorldError } from '@workflow/errors'; -import { eventIdToSlot } from '@workflow/world'; +import { eventIdToSlot, StartHookSchema } from '@workflow/world'; import { lock } from 'proper-lockfile'; import { decodeTime, monotonicFactory } from 'ulid'; import { z } from 'zod'; @@ -13,6 +13,7 @@ import { readJSON, resolveWithinBase, stripTag, + taggedPath, ulidToDate, withWindowsRetry, } from '../fs.js'; @@ -347,15 +348,53 @@ export function hookTokenClaimPath(basedir: string, token: string): string { return path.join(basedir, 'hooks', 'tokens', `${hashToken(token)}.json`); } +export function startHookAdmissionPath( + basedir: string, + runId: string, + tag?: string +): string { + return taggedPath(basedir, 'hooks/admissions', runId, tag); +} + +const StartHookAdmissionBaseSchema = StartHookSchema.extend({ + runId: z.string(), +}); + +export const StartHookAdmissionSchema = z.union([ + StartHookAdmissionBaseSchema.extend({ + eventId: z.string(), + }).strict(), + StartHookAdmissionBaseSchema.omit({ tokenRetentionUntil: true }) + .extend({ + redirectRunId: z.string(), + }) + .strict(), +]); + +export type StartHookAdmission = z.infer; + +export function readStartHookAdmission( + basedir: string, + runId: string, + tag?: string +): Promise { + return readJSON( + startHookAdmissionPath(basedir, runId, tag), + StartHookAdmissionSchema + ); +} + export const HookTokenClaimSchema = z.object({ - // Legacy claims omitted hookId. Keeping it optional preserves their - // existing cross-hook conflict behavior (see #2283). + // The path is hashed, so current claims retain the token for inspection. + token: z.string().optional(), + // Reservations and legacy claims omit hookId. Legacy claims also omit + // eventId; their recovery marker pins the canonical hook_created event. hookId: z.string().optional(), runId: z.string(), - // Legacy claims also omitted eventId. Their recovery marker pins the - // canonical hook_created event before concurrent retries publish it. eventId: z.string().optional(), tokenRetentionUntil: z.coerce.date().optional(), + // Token claims are global, while run and Hook files may be tagged. + tag: z.string().optional(), }); export type HookTokenClaim = z.infer; @@ -456,18 +495,28 @@ export function hookResumeClaimPath( return path.join(basedir, 'hooks', 'resumes', `${key}.json`); } -/** Deletes a claim only while it still belongs to this Hook. */ +type HookTokenClaimOwner = { runId: string; tag: string | undefined } & ( + | { hookId: string } + | { eventId: string } +); + +/** Deletes a claim only while it still belongs to this owner. */ export async function releaseHookTokenClaimIfOwnedBy( basedir: string, token: string, - runId: string, - hookId: string + owner: HookTokenClaimOwner ): Promise { await withHookTokenClaimLock(basedir, token, async (signal) => { const claimPath = hookTokenClaimPath(basedir, token); const claim = await readHookTokenClaim(claimPath); if (!claim) return; - if (claim.runId === runId && claim.hookId === hookId) { + const owned = + claim.runId === owner.runId && + (claim.tag === undefined || claim.tag === owner.tag) && + ('hookId' in owner + ? claim.hookId === owner.hookId + : claim.hookId === undefined && claim.eventId === owner.eventId); + if (owned) { signal.throwIfAborted(); await deleteJSON(claimPath); } diff --git a/packages/world-local/src/storage/hooks-storage.ts b/packages/world-local/src/storage/hooks-storage.ts index 621c72e3f3..fdf0731aa6 100644 --- a/packages/world-local/src/storage/hooks-storage.ts +++ b/packages/world-local/src/storage/hooks-storage.ts @@ -30,10 +30,12 @@ import { } from '../fs.js'; import { filterHookData } from './filters.js'; import { + type HookTokenClaim, hookRecoveryMarkerPath, hookTokenClaimPath, isHookDisposalCommitted, readHookTokenClaim, + readStartHookAdmission, releaseHookTokenClaimIfOwnedBy, } from './helpers.js'; import { @@ -156,7 +158,8 @@ async function restoreHookCachesFromEvent( runId: hook.runId, eventId: event.eventId, tokenRetentionUntil: event.eventData.tokenRetentionUntil, - }) + tag, + } satisfies HookTokenClaim) ); // Marker before entity (see hook-index.ts crash-ordering invariant). await writeHookByRunMarker(basedir, hook.runId, hook.hookId, tag); @@ -340,7 +343,8 @@ export function createHooksStorage( */ export async function deleteAllHooksForRun( basedir: string, - runId: string + runId: string, + tag?: string ): Promise { // Discover this run's hooks via by-run markers (a prefix readdir) // instead of reading every live hook entity in the world. @@ -371,12 +375,11 @@ export async function deleteAllHooksForRun( // Release the claim only if it still points at this hook — a // claimant may already hold a fresh claim for the token (see // `isHookTokenClaimReleasable`). - await releaseHookTokenClaimIfOwnedBy( - basedir, - hook.token, - hook.runId, - hook.hookId - ); + await releaseHookTokenClaimIfOwnedBy(basedir, hook.token, { + runId: hook.runId, + hookId: hook.hookId, + tag: marker.tag, + }); await deleteJSON( hookRecoveryMarkerPath(basedir, hook.token, hook.runId, hook.hookId) ); @@ -385,4 +388,18 @@ export async function deleteAllHooksForRun( } await deleteHookByRunMarkerFile(basedir, marker.fileId); } + + const admission = await readStartHookAdmission(basedir, runId, tag); + if ( + admission && + !('redirectRunId' in admission) && + (!admission.tokenRetentionUntil || + admission.tokenRetentionUntil.getTime() <= Date.now()) + ) { + await releaseHookTokenClaimIfOwnedBy(basedir, admission.token, { + runId, + eventId: admission.eventId, + tag, + }); + } } diff --git a/packages/world-local/src/test-helpers.ts b/packages/world-local/src/test-helpers.ts index fa4ada687f..bb18de34e4 100644 --- a/packages/world-local/src/test-helpers.ts +++ b/packages/world-local/src/test-helpers.ts @@ -1,5 +1,6 @@ import type { Hook, + RunCreationData, SerializedData, Step, Storage, @@ -18,13 +19,7 @@ import { SPEC_VERSION_CURRENT } from '@workflow/world'; */ export async function createRun( storage: Storage, - data: { - deploymentId: string; - workflowName: string; - input: SerializedData; - executionContext?: Record; - attributes?: Record; - } + data: RunCreationData ): Promise { const result = await storage.events.create(null, { eventType: 'run_created',