From dd1999b8f1117f60a9812731ca2747b1b8f940e3 Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Wed, 1 Jul 2026 16:02:18 -0700 Subject: [PATCH] fix(world-local): rebuild hook caches from event log --- .changeset/world-local-hook-cache-rebuild.md | 6 + packages/world-local/src/storage.test.ts | 157 +++++++++++-- .../world-local/src/storage/events-storage.ts | 41 ++-- .../world-local/src/storage/hooks-storage.ts | 206 +++++++++++++++++- packages/world/src/events.ts | 17 +- packages/world/src/index.ts | 15 +- packages/world/src/runs.ts | 20 +- packages/world/src/steps.ts | 14 ++ 8 files changed, 436 insertions(+), 40 deletions(-) create mode 100644 .changeset/world-local-hook-cache-rebuild.md 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 dcf4f109c2..2126187baf 100644 --- a/packages/world-local/src/storage.test.ts +++ b/packages/world-local/src/storage.test.ts @@ -1304,24 +1304,24 @@ describe('Storage', () => { // never be mistaken for the complete delta. await updateRun(storage, testRunId, 'run_started'); - // Open many hooks BEFORE the cursor so they are not part of the delta; - // their in-band deliveries (below) are. - const HOOK_COUNT = 25; // > paginatedFileSystemQuery default limit (20) - for (let i = 0; i < HOOK_COUNT; i++) { - await createHook(storage, testRunId, { - hookId: `corr_hook_${i}`, - token: `tok_${i}`, - }); - } + await createHook(storage, testRunId, { + hookId: 'corr_delta_page_hook', + token: 'tok_delta_page_hook', + }); const sinceCursor = await currentCursor(); - // A burst of in-band hook deliveries lands while the step runs, then - // the sequential step itself — together far more than one page. - for (let i = 0; i < HOOK_COUNT; i++) { + // A burst of in-band hook deliveries lands while the step runs. One + // hook is enough here; the assertion is about delta pagination, not + // creating many distinct hook tokens. + const DELTA_FILLER_EVENT_COUNT = 21; // > default page limit (20) + for (let i = 0; i < DELTA_FILLER_EVENT_COUNT; i++) { await storage.events.create(testRunId, { eventType: 'hook_received' as const, - correlationId: `corr_hook_${i}`, - eventData: { token: `tok_${i}`, payload: new Uint8Array([i]) }, + correlationId: 'corr_delta_page_hook', + eventData: { + token: 'tok_delta_page_hook', + payload: new Uint8Array([i]), + }, }); } await createStep(storage, testRunId, { @@ -1351,7 +1351,9 @@ describe('Storage', () => { expect(result.hasMore).toBe(true); expect(firstPage.hasMore).toBe(true); - expect(result.events?.length).toBeLessThan(HOOK_COUNT + 3); + expect(result.events?.length).toBeLessThan( + DELTA_FILLER_EVENT_COUNT + 3 + ); expect(result.events?.map((e) => e.eventId)).toEqual( firstPage.data.map((e) => e.eventId) ); @@ -3323,6 +3325,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 0f82192dbd..d2130b4f2b 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -24,6 +24,9 @@ import { EventSchema, HookSchema, isLegacySpecVersion, + isTerminalRunEventType, + isTerminalStepStatus, + isTerminalWorkflowRunStatus, requiresNewerWorld, SPEC_VERSION_CURRENT, StepSchema, @@ -57,7 +60,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'; import { withRunFileLock } from './runs-storage.js'; @@ -259,7 +265,7 @@ async function repairHookEntityFromPersistedEvent( environment: 'local', createdAt: persistedEvent.createdAt, specVersion: persistedEvent.specVersion, - isWebhook: eventData.isWebhook ?? false, + isWebhook: eventData.isWebhook ?? true, isSystem: eventData.isSystem ?? false, }; await writeExclusive( @@ -485,11 +491,7 @@ export function createEventsStorage( ): void { // Terminal runs release their cached history so a long-lived dev // server doesn't retain completed runs forever. - if ( - event.eventType === 'run_completed' || - event.eventType === 'run_failed' || - event.eventType === 'run_cancelled' - ) { + if (isTerminalRunEventType(event.eventType)) { clearRunCache(event.runId); return; } @@ -626,14 +628,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 @@ -801,7 +795,7 @@ export function createEventsStorage( (data.eventData as { input?: unknown }).input !== undefined; // Run terminal state validation - if (currentRun && isRunTerminal(currentRun.status)) { + if (currentRun && isTerminalWorkflowRunStatus(currentRun.status)) { const runTerminalEvents = [ 'run_started', 'run_completed', @@ -916,14 +910,14 @@ export function createEventsStorage( // the lazy-start path (no step yet) — there is nothing terminal to // guard against in that case, so these checks are skipped. if (validatedStep) { - 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}"` @@ -1429,7 +1423,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}"` ); @@ -1569,6 +1563,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..ad94af25ae 100644 --- a/packages/world-local/src/storage/hooks-storage.ts +++ b/packages/world-local/src/storage/hooks-storage.ts @@ -1,25 +1,211 @@ 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, isSystem } = 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, + isSystem: isSystem ?? false, + }; +} + +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 +240,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 +262,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 00d7eb017d..35a7ea83ec 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -78,6 +78,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, @@ -184,7 +198,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({ @@ -421,6 +435,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 8052636320..cb07754508 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -27,7 +27,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'; @@ -50,6 +54,9 @@ export { export { reenqueueActiveRuns } from './recovery.js'; export type * from './runs.js'; export { + isTerminalWorkflowRunStatus, + TERMINAL_WORKFLOW_RUN_STATUSES, + TerminalWorkflowRunStatusSchema, WorkflowRunBaseSchema, WorkflowRunSchema, WorkflowRunStatusSchema, @@ -82,7 +89,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 822a0ac80d..f2cd1de462 100644 --- a/packages/world/src/runs.ts +++ b/packages/world/src/runs.ts @@ -10,6 +10,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 @@ -121,7 +140,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 749f5484e8..68b27dc7db 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; /**