diff --git a/.changeset/world-local-hook-cache-rebuild.md b/.changeset/world-local-hook-cache-rebuild.md new file mode 100644 index 0000000000..382ac111d9 --- /dev/null +++ b/.changeset/world-local-hook-cache-rebuild.md @@ -0,0 +1,6 @@ +--- +'@workflow/world-local': patch +'@workflow/world': patch +--- + +Keep local hooks reachable after a crash or restart by rebuilding lost hook cache files from committed hook creation events, preventing active hook tokens from being reused. diff --git a/packages/world-local/src/storage.test.ts b/packages/world-local/src/storage.test.ts index 3a90579444..3b391b2639 100644 --- a/packages/world-local/src/storage.test.ts +++ b/packages/world-local/src/storage.test.ts @@ -2381,6 +2381,131 @@ describe('Storage', () => { ).toHaveLength(1); }); + it('rebuilds missing hook caches from a committed hook_created event', async () => { + // Regression for #2339: once hook_created is committed to the event log, + // the hook entity and token claim are cache files. If both are missing + // after a crash or upgrade, a normal hook read should rebuild them from + // the persisted event instead of treating the hook/token as gone. + const metadata = new Uint8Array([0xee]); + const hookId = 'hook_event_log_rebuild'; + const token = 'event-log-rebuild-token'; + + const created = await storage.events.create(testRunId, { + eventType: 'hook_created', + correlationId: hookId, + eventData: { token, metadata, isWebhook: true }, + }); + expect(created.event.eventType).toBe('hook_created'); + + const hookPath = path.join(testDir, 'hooks', `${hookId}.json`); + const tokenClaimPath = path.join( + testDir, + 'hooks', + 'tokens', + `${hashToken(token)}.json` + ); + await fs.unlink(hookPath); + await fs.unlink(tokenClaimPath); + await fs.writeFile( + path.join(testDir, 'events', 'wrun_malformed-event.json'), + '{' + ); + + const conflict = await storage.events.create(testRunId, { + eventType: 'hook_created', + correlationId: 'hook_event_log_rebuild_conflict', + eventData: { token }, + }); + expect(conflict.event.eventType).toBe('hook_conflict'); + expect((conflict.event as any).eventData.conflictingRunId).toBe( + testRunId + ); + + await fs.unlink(hookPath); + await fs.unlink(tokenClaimPath); + + await expect(storage.hooks.get(hookId)).resolves.toMatchObject({ + runId: testRunId, + hookId, + token, + metadata, + isWebhook: true, + }); + + const claim = JSON.parse(await fs.readFile(tokenClaimPath, 'utf8')); + expect(claim).toMatchObject({ + runId: testRunId, + hookId, + eventId: created.event.eventId, + }); + }); + + it('preserves legacy webhook default when rebuilding a hook without isWebhook', async () => { + const metadata = new Uint8Array([0xab]); + const hookId = 'hook_legacy_webhook_default'; + const token = 'legacy-webhook-default-token'; + const created = await storage.events.create(testRunId, { + eventType: 'hook_created', + correlationId: hookId, + eventData: { token, metadata }, + }); + expect(created.event.eventType).toBe('hook_created'); + + await fs.unlink(path.join(testDir, 'hooks', `${hookId}.json`)); + await fs.unlink( + path.join(testDir, 'hooks', 'tokens', `${hashToken(token)}.json`) + ); + + await expect(storage.hooks.get(hookId)).resolves.toMatchObject({ + hookId, + token, + metadata, + isWebhook: true, + }); + }); + + it('does not rebuild a hook for a run already marked terminal', async () => { + const hookId = 'hook_terminal_run_cache'; + const token = 'terminal-run-cache-token'; + await createHook(storage, testRunId, { hookId, token }); + + const hookPath = path.join(testDir, 'hooks', `${hookId}.json`); + const tokenClaimPath = path.join( + testDir, + 'hooks', + 'tokens', + `${hashToken(token)}.json` + ); + await fs.unlink(hookPath); + await fs.unlink(tokenClaimPath); + + const run = await storage.runs.get(testRunId); + await writeJSON( + path.join(testDir, 'runs', `${testRunId}.json`), + { + ...run, + status: 'cancelled', + completedAt: new Date(), + updatedAt: new Date(), + }, + { overwrite: true } + ); + + const nextRun = await createRun(storage, { + deploymentId: 'deployment-next', + workflowName: 'next-workflow', + input: new Uint8Array(), + }); + + const created = await storage.events.create(nextRun.runId, { + eventType: 'hook_created', + correlationId: 'hook_terminal_run_cache_next', + eventData: { token }, + }); + expect(created.event.eventType).toBe('hook_created'); + expect(created.hook?.runId).toBe(nextRun.runId); + }); + it('repairs an event-first orphan via the legacy-claim probe path', async () => { // Same crash window as the test above, but exercised through // the legacy-claim branch: the claim file lacks `eventId` (as diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index b441417221..bf02386464 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -22,6 +22,8 @@ import { EventSchema, HookSchema, isLegacySpecVersion, + isTerminalStepStatus, + isTerminalWorkflowRunStatus, requiresNewerWorld, SPEC_VERSION_CURRENT, StepSchema, @@ -52,7 +54,10 @@ import { hookRecoveryMarkerPath, monotonicUlid, } from './helpers.js'; -import { deleteAllHooksForRun } from './hooks-storage.js'; +import { + deleteAllHooksForRun, + rebuildLiveHookByTokenFromEventLog, +} from './hooks-storage.js'; import { handleLegacyEvent } from './legacy.js'; const HookTokenClaimSchema = z.object({ @@ -229,7 +234,7 @@ async function repairHookEntityFromPersistedEvent( environment: 'local', createdAt: persistedEvent.createdAt, specVersion: persistedEvent.specVersion, - isWebhook: eventData.isWebhook ?? false, + isWebhook: eventData.isWebhook ?? true, }; await writeExclusive( taggedPath(basedir, 'hooks', hookId, tag), @@ -418,14 +423,6 @@ export function createEventsStorage( // specVersion is always sent by the runtime, but we provide a fallback for safety const effectiveSpecVersion = data.specVersion ?? SPEC_VERSION_CURRENT; - // Helper to check if run is in terminal state - const isRunTerminal = (status: string) => - ['completed', 'failed', 'cancelled'].includes(status); - - // Helper to check if step is in terminal state - const isStepTerminal = (status: string) => - ['completed', 'failed', 'cancelled'].includes(status); - // Get current run state for validation (if not creating a new run) // Skip run validation for step_completed and step_retrying - they only operate // on running steps, and running steps are always allowed to modify regardless @@ -557,7 +554,7 @@ export function createEventsStorage( // ============================================================ // Run terminal state validation - if (currentRun && isRunTerminal(currentRun.status)) { + if (currentRun && isTerminalWorkflowRunStatus(currentRun.status)) { const runTerminalEvents = [ 'run_started', 'run_completed', @@ -647,14 +644,14 @@ export function createEventsStorage( } // Step terminal state validation - if (isStepTerminal(validatedStep.status)) { + if (isTerminalStepStatus(validatedStep.status)) { throw new EntityConflictError( `Cannot modify step in terminal state "${validatedStep.status}"` ); } // On terminal runs: only allow completing/failing in-progress steps - if (currentRun && isRunTerminal(currentRun.status)) { + if (currentRun && isTerminalWorkflowRunStatus(currentRun.status)) { if (validatedStep.status !== 'running') { throw new RunExpiredError( `Cannot modify non-running step on run in terminal state "${currentRun.status}"` @@ -949,7 +946,7 @@ export function createEventsStorage( StepSchema, tag ); - if (freshStep && isStepTerminal(freshStep.status)) { + if (freshStep && isTerminalStepStatus(freshStep.status)) { throw new EntityConflictError( `Cannot modify step in terminal state "${freshStep.status}"` ); @@ -1100,6 +1097,15 @@ export function createEventsStorage( 'tokens', `${hashToken(hookData.token)}.json` ); + // When the claim is absent, the event log is the only durable source + // that can distinguish a first hook from a crash-lost token cache. + if (!(await readHookTokenClaim(constraintPath))) { + await rebuildLiveHookByTokenFromEventLog( + basedir, + hookData.token, + tag + ); + } // Persist `eventId` in the claim so concurrent / cross- // process retries can converge on a single canonical // `hook_created` event path. See the recovery comment diff --git a/packages/world-local/src/storage/hooks-storage.ts b/packages/world-local/src/storage/hooks-storage.ts index e4c8f3c742..983977fec2 100644 --- a/packages/world-local/src/storage/hooks-storage.ts +++ b/packages/world-local/src/storage/hooks-storage.ts @@ -1,25 +1,210 @@ import path from 'node:path'; import { HookNotFoundError } from '@workflow/errors'; import type { + Event, GetHookParams, Hook, + HookCreatedEvent, ListHooksParams, PaginatedResponse, Storage, } from '@workflow/world'; -import { HookSchema } from '@workflow/world'; +import { + EventSchema, + HookSchema, + isTerminalRunEventType, + isTerminalWorkflowRunStatus, + WorkflowRunSchema, +} from '@workflow/world'; +import { z } from 'zod'; import { DEFAULT_RESOLVE_DATA_OPTION } from '../config.js'; import { assertSafeEntityId, deleteJSON, + hasTag, + isUntagged, + jsonReplacer, listJSONFiles, paginatedFileSystemQuery, readJSON, readJSONWithFallback, + taggedPath, + writeExclusive, } from '../fs.js'; import { filterHookData } from './filters.js'; import { hashToken, hookRecoveryMarkerPath } from './helpers.js'; +function isVisibleToTag(fileId: string, tag: string | undefined): boolean { + return tag ? isUntagged(fileId) || hasTag(fileId, tag) : isUntagged(fileId); +} + +function getHookCreatedToken(event: Event): string | undefined { + if (event.eventType !== 'hook_created') return undefined; + const token = (event.eventData as { token?: unknown }).token; + return typeof token === 'string' ? token : undefined; +} + +function hookFromCreatedEvent(event: Event & HookCreatedEvent): Hook { + const { token, metadata, isWebhook } = event.eventData; + return { + runId: event.runId, + hookId: event.correlationId, + token, + metadata, + ownerId: 'local-owner', + projectId: 'local-project', + environment: 'local', + createdAt: event.createdAt, + specVersion: event.specVersion, + isWebhook: isWebhook ?? true, + }; +} + +function isMatchingHookCreatedEvent( + event: Event, + matches: (event: Event) => boolean +): event is Event & HookCreatedEvent { + return ( + event.eventType === 'hook_created' && + typeof event.correlationId === 'string' && + matches(event) + ); +} + +function closesLiveHook( + event: Event, + liveEvent: Event & HookCreatedEvent +): boolean { + if (event.runId !== liveEvent.runId) return false; + return ( + (event.eventType === 'hook_disposed' && + event.correlationId === liveEvent.correlationId) || + isTerminalRunEventType(event.eventType) + ); +} + +async function readEventForHookScan(filePath: string): Promise { + try { + return await readJSON(filePath, EventSchema); + } catch (error) { + if (error instanceof SyntaxError || error instanceof z.ZodError) { + return null; + } + throw error; + } +} + +async function isTerminalRunCache( + basedir: string, + runId: string, + tag?: string +): Promise { + const run = await readJSONWithFallback( + basedir, + 'runs', + runId, + WorkflowRunSchema, + tag + ); + return run ? isTerminalWorkflowRunStatus(run.status) : false; +} + +async function findLiveHookCreatedEvent( + basedir: string, + matches: (event: Event) => boolean, + tag?: string +): Promise<(Event & HookCreatedEvent) | null> { + const eventsDir = path.join(basedir, 'events'); + const events: Event[] = []; + + for (const fileId of await listJSONFiles(eventsDir)) { + if (!isVisibleToTag(fileId, tag)) continue; + const event = await readEventForHookScan( + path.join(eventsDir, `${fileId}.json`) + ); + if (event) events.push(event); + } + + events.sort((a, b) => { + const byTime = a.createdAt.getTime() - b.createdAt.getTime(); + return byTime === 0 ? a.eventId.localeCompare(b.eventId) : byTime; + }); + + let liveEvent: (Event & HookCreatedEvent) | null = null; + for (const event of events) { + if (isMatchingHookCreatedEvent(event, matches)) { + liveEvent = event; + continue; + } + + if (liveEvent && closesLiveHook(event, liveEvent)) { + liveEvent = null; + } + } + + if (liveEvent && (await isTerminalRunCache(basedir, liveEvent.runId, tag))) { + return null; + } + + return liveEvent; +} + +async function restoreHookCachesFromEvent( + basedir: string, + event: Event & HookCreatedEvent, + tag?: string +): Promise { + const hook = hookFromCreatedEvent(event); + + const claimPath = path.join( + basedir, + 'hooks', + 'tokens', + `${hashToken(hook.token)}.json` + ); + await writeExclusive( + claimPath, + JSON.stringify({ + token: hook.token, + hookId: hook.hookId, + runId: hook.runId, + eventId: event.eventId, + }) + ); + await writeExclusive( + taggedPath(basedir, 'hooks', hook.hookId, tag), + JSON.stringify(hook, jsonReplacer, 2) + ); + + return hook; +} + +export async function rebuildLiveHookByTokenFromEventLog( + basedir: string, + token: string, + tag?: string +): Promise { + const event = await findLiveHookCreatedEvent( + basedir, + (candidate) => getHookCreatedToken(candidate) === token, + tag + ); + return event ? restoreHookCachesFromEvent(basedir, event, tag) : null; +} + +async function rebuildLiveHookByIdFromEventLog( + basedir: string, + hookId: string, + tag?: string +): Promise { + const event = await findLiveHookCreatedEvent( + basedir, + (candidate) => candidate.correlationId === hookId, + tag + ); + return event ? restoreHookCachesFromEvent(basedir, event, tag) : null; +} + /** * Creates a hooks storage implementation using the filesystem. * Implements the Storage['hooks'] interface with hook CRUD operations. @@ -54,7 +239,19 @@ export function createHooksStorage( tag ); if (!hook) { - throw new HookNotFoundError(hookId); + const rebuilt = await rebuildLiveHookByIdFromEventLog( + basedir, + hookId, + tag + ); + if (!rebuilt) { + throw new HookNotFoundError(hookId); + } + const resolveData = params?.resolveData || DEFAULT_RESOLVE_DATA_OPTION; + return filterHookData( + { ...rebuilt, isWebhook: rebuilt.isWebhook ?? true }, + resolveData + ); } const resolveData = params?.resolveData || DEFAULT_RESOLVE_DATA_OPTION; return filterHookData( @@ -64,7 +261,9 @@ export function createHooksStorage( } async function getByToken(token: string): Promise { - const hook = await findHookByToken(token); + const hook = + (await findHookByToken(token)) ?? + (await rebuildLiveHookByTokenFromEventLog(basedir, token, tag)); if (!hook) { throw new HookNotFoundError(token); } diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 48bd70d22e..f35ad85e1c 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -75,6 +75,20 @@ export const EventTypeSchema = z.enum([ 'wait_created', 'wait_completed', ]); +export type EventType = z.infer; +export const TerminalRunEventTypeSchema = EventTypeSchema.extract([ + 'run_completed', + 'run_failed', + 'run_cancelled', +] as const); +export type TerminalRunEventType = z.infer; +export const TERMINAL_RUN_EVENT_TYPES = TerminalRunEventTypeSchema.options; + +export function isTerminalRunEventType( + eventType: string +): eventType is TerminalRunEventType { + return TERMINAL_RUN_EVENT_TYPES.includes(eventType as TerminalRunEventType); +} // Base event schema with common properties // TODO: Event data on all specific event schemas can actually be undefined, @@ -164,7 +178,7 @@ const StepCreatedEventSchema = BaseEventSchema.extend({ * Event created when a hook is first invoked. The World implementation * atomically creates both the event and the hook entity. */ -const HookCreatedEventSchema = BaseEventSchema.extend({ +export const HookCreatedEventSchema = BaseEventSchema.extend({ eventType: z.literal('hook_created'), correlationId: z.string(), eventData: z.object({ @@ -369,6 +383,7 @@ export const EventSchema = AllEventsSchema.and( // Inferred types export type Event = z.infer; +export type HookCreatedEvent = z.infer; export type HookReceivedEvent = z.infer; export type HookConflictEvent = z.infer; diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index 50fd7ba678..e596735074 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -5,7 +5,11 @@ export { EVENT_DATA_REF_FIELDS, EventSchema, EventTypeSchema, + HookCreatedEventSchema, + isTerminalRunEventType, stripEventDataRefs, + TERMINAL_RUN_EVENT_TYPES, + TerminalRunEventTypeSchema, } from './events.js'; export type * from './hooks.js'; export { HookSchema } from './hooks.js'; @@ -28,6 +32,9 @@ export { export { reenqueueActiveRuns } from './recovery.js'; export type * from './runs.js'; export { + isTerminalWorkflowRunStatus, + TERMINAL_WORKFLOW_RUN_STATUSES, + TerminalWorkflowRunStatusSchema, WorkflowRunBaseSchema, WorkflowRunSchema, WorkflowRunStatusSchema, @@ -58,7 +65,13 @@ export { SPEC_VERSION_SUPPORTS_EVENT_SOURCING, } from './spec-version.js'; export type * from './steps.js'; -export { StepSchema, StepStatusSchema } from './steps.js'; +export { + isTerminalStepStatus, + StepSchema, + StepStatusSchema, + TERMINAL_STEP_STATUSES, + TerminalStepStatusSchema, +} from './steps.js'; export { DEFAULT_TIMESTAMP_THRESHOLD_FUTURE_MS, DEFAULT_TIMESTAMP_THRESHOLD_MS, diff --git a/packages/world/src/runs.ts b/packages/world/src/runs.ts index 3c4a6c1eb3..248bbe1176 100644 --- a/packages/world/src/runs.ts +++ b/packages/world/src/runs.ts @@ -14,6 +14,25 @@ export const WorkflowRunStatusSchema = z.enum([ 'failed', 'cancelled', ]); +export type WorkflowRunStatus = z.infer; +export const TerminalWorkflowRunStatusSchema = WorkflowRunStatusSchema.extract([ + 'completed', + 'failed', + 'cancelled', +] as const); +export type TerminalWorkflowRunStatus = z.infer< + typeof TerminalWorkflowRunStatusSchema +>; +export const TERMINAL_WORKFLOW_RUN_STATUSES = + TerminalWorkflowRunStatusSchema.options; + +export function isTerminalWorkflowRunStatus( + status: string +): status is TerminalWorkflowRunStatus { + return TERMINAL_WORKFLOW_RUN_STATUSES.includes( + status as TerminalWorkflowRunStatus + ); +} /** * Base schema for the Workflow runs. Prefer using WorkflowRunSchema @@ -94,7 +113,6 @@ export const WorkflowRunSchema = z.discriminatedUnion('status', [ ]); // Inferred types -export type WorkflowRunStatus = z.infer; export type WorkflowRun = z.infer; /** diff --git a/packages/world/src/steps.ts b/packages/world/src/steps.ts index ba0dbac282..4844d6e7aa 100644 --- a/packages/world/src/steps.ts +++ b/packages/world/src/steps.ts @@ -73,6 +73,20 @@ export const StepSchema = z.object({ // Inferred types export type StepStatus = z.infer; +export const TerminalStepStatusSchema = StepStatusSchema.extract([ + 'completed', + 'failed', + 'cancelled', +] as const); +export type TerminalStepStatus = z.infer; +export const TERMINAL_STEP_STATUSES = TerminalStepStatusSchema.options; + +export function isTerminalStepStatus( + status: string +): status is TerminalStepStatus { + return TERMINAL_STEP_STATUSES.includes(status as TerminalStepStatus); +} + export type Step = z.infer; /**