diff --git a/.changeset/dull-hats-melt.md b/.changeset/dull-hats-melt.md new file mode 100644 index 0000000000..54cce04fbf --- /dev/null +++ b/.changeset/dull-hats-melt.md @@ -0,0 +1,6 @@ +--- +'@workflow/core': patch +'@workflow/world': patch +--- + +Ignore duplicate events that a concurrent replay wrote for an entity the event log already records, instead of failing the run with a corrupted event log diff --git a/packages/core/src/duplicate-events.test.ts b/packages/core/src/duplicate-events.test.ts new file mode 100644 index 0000000000..5466a386dc --- /dev/null +++ b/packages/core/src/duplicate-events.test.ts @@ -0,0 +1,153 @@ +import type { Event } from '@workflow/world'; +import { describe, expect, it, vi } from 'vitest'; +import { WorkflowSuspension } from './global.js'; +import { dehydrateStepReturnValue } from './serialization.js'; +import { createUseStep } from './step.js'; +import { + CORR_IDS, + runWithDiscontinuation, + setupWorkflowContext, +} from './test-support/orchestrator-context.js'; +import { createSleep } from './workflow/sleep.js'; + +/** + * Concurrent replays of one run share a single event log and write to it + * without a currency guard, so a replay working from a stale prefix can commit + * a second `step_created` / `step_started` / `wait_created` for an entity the + * log already records one of. Those writes are committed but inert: every + * replay reads the first event of that class at the same position, so the + * straggler cannot change what the workflow observes. + * + * Before this behavior existed the straggler had no consumer left to claim it + * (the entity's consumer deregistered when it took the step's result), which + * surfaced as `ReplayDivergenceError` and, after retries, a terminal + * `CORRUPTED_EVENT_LOG` on a run whose log was fine. + * + * These tests drive the real step and sleep primitives against hand-written + * logs. The unit-level behavior lives in `events-consumer.test.ts`. + */ + +const RESUME_AT = new Date('2099-01-01T00:00:00.000Z'); + +async function dehydrate(value: unknown) { + const ops: Promise[] = []; + return await dehydrateStepReturnValue(value, 'wrun_test', undefined, ops); +} + +function event( + index: number, + eventType: string, + correlationId: string, + eventData: Record +): Event { + return { + eventId: `evnt_${index}`, + runId: 'wrun_test', + eventType, + correlationId, + eventData, + createdAt: new Date(), + } as unknown as Event; +} + +function pendingStepNames(ctx: ReturnType) { + return [...ctx.invocationsQueue.values()] + .filter((i) => i.type === 'step') + .map((i) => (i.type === 'step' ? i.stepName : undefined)); +} + +describe('events repeating a class already in the log', () => { + it('ignores a step_started that lands after the step completed', async () => { + const result = await dehydrate('a-result'); + const onDuplicateEvent = vi.fn(); + const events = [ + event(0, 'step_created', `step_${CORR_IDS[0]}`, { stepName: 'stepA' }), + event(1, 'step_started', `step_${CORR_IDS[0]}`, { stepName: 'stepA' }), + event(2, 'step_completed', `step_${CORR_IDS[0]}`, { + stepName: 'stepA', + result, + }), + // A concurrent replay that had not yet seen evnt_2 re-invokes stepA. + event(3, 'step_started', `step_${CORR_IDS[0]}`, { stepName: 'stepA' }), + event(4, 'step_created', `step_${CORR_IDS[1]}`, { stepName: 'stepB' }), + ]; + const ctx = setupWorkflowContext(events, { onDuplicateEvent }); + const useStep = createUseStep(ctx); + + const observed: unknown[] = []; + const { error } = await runWithDiscontinuation(ctx, async () => { + const stepA = useStep('stepA'); + const stepB = useStep('stepB'); + observed.push(await stepA()); + observed.push(await stepB()); + return 'done'; + }); + + // Suspension, not divergence: the run is waiting on stepB. + expect(WorkflowSuspension.is(error)).toBe(true); + expect(observed).toEqual(['a-result']); + expect(pendingStepNames(ctx)).toEqual(['stepB']); + expect(onDuplicateEvent).toHaveBeenCalledTimes(1); + expect(onDuplicateEvent).toHaveBeenCalledWith(events[3], 'step_started'); + }); + + it('ignores a wait_created that lands after the wait completed', async () => { + const onDuplicateEvent = vi.fn(); + const events = [ + event(0, 'wait_created', `wait_${CORR_IDS[0]}`, { resumeAt: RESUME_AT }), + event(1, 'wait_completed', `wait_${CORR_IDS[0]}`, { + resumeAt: RESUME_AT, + }), + // A concurrent replay that had not yet seen evnt_1 re-created the sleep. + event(2, 'wait_created', `wait_${CORR_IDS[0]}`, { resumeAt: RESUME_AT }), + event(3, 'step_created', `step_${CORR_IDS[1]}`, { + stepName: 'afterSleep', + }), + ]; + const ctx = setupWorkflowContext(events, { onDuplicateEvent }); + const sleep = createSleep(ctx); + const useStep = createUseStep(ctx); + + const { error } = await runWithDiscontinuation(ctx, async () => { + await sleep(RESUME_AT); + await useStep('afterSleep')(); + return 'done'; + }); + + expect(WorkflowSuspension.is(error)).toBe(true); + expect(pendingStepNames(ctx)).toEqual(['afterSleep']); + expect(onDuplicateEvent).toHaveBeenCalledTimes(1); + expect(onDuplicateEvent).toHaveBeenCalledWith(events[2], 'wait_created'); + }); + + it('still reports divergence for an event repeating nothing in the log', async () => { + const result = await dehydrate('a-result'); + const onDuplicateEvent = vi.fn(); + const events = [ + event(0, 'step_created', `step_${CORR_IDS[0]}`, { stepName: 'stepA' }), + event(1, 'step_started', `step_${CORR_IDS[0]}`, { stepName: 'stepA' }), + event(2, 'step_completed', `step_${CORR_IDS[0]}`, { + stepName: 'stepA', + result, + }), + // Belongs to no entity this workflow ever creates: the sleep below mints + // CORR_IDS[1]. Nothing can consume it, and nothing should suppress it. + event(3, 'wait_created', 'wait_01JZZZZZZZZZZZZZZZZZZZZZZZ', { + resumeAt: RESUME_AT, + }), + ]; + const ctx = setupWorkflowContext(events, { onDuplicateEvent }); + const sleep = createSleep(ctx); + const useStep = createUseStep(ctx); + + const { error } = await runWithDiscontinuation(ctx, async () => { + await useStep('stepA')(); + await sleep(RESUME_AT); + return 'done'; + }); + + expect(WorkflowSuspension.is(error)).toBe(false); + expect(String(error)).toContain('Unconsumed event in event log'); + expect(onDuplicateEvent).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/core/src/events-consumer.test.ts b/packages/core/src/events-consumer.test.ts index 9bae822106..0406b181fd 100644 --- a/packages/core/src/events-consumer.test.ts +++ b/packages/core/src/events-consumer.test.ts @@ -1,10 +1,11 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; -import { describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { DEFERRED_CHECK_DELAY_MS, EventConsumerResult, EventsConsumer, + MIN_DEFERRED_CHECK_DELAY_MS, } from './events-consumer.js'; // Helper function to create mock events @@ -734,4 +735,412 @@ describe('EventsConsumer', () => { expect(await unconsumedReceived.promise).toEqual(event); }); }); + + describe('duplicate event classes', () => { + // Nothing here waits on the window for its result — a duplicate is stepped + // over in the pass that offers it — so run at the shortest legal delay and + // let the assertions that a check did NOT fire be cheap. + beforeEach(() => { + vi.stubEnv( + 'WORKFLOW_DEFERRED_CHECK_DELAY_MS', + String(MIN_DEFERRED_CHECK_DELAY_MS) + ); + }); + afterEach(() => { + vi.unstubAllEnvs(); + }); + + // Waits long enough for a deferred check that was not cancelled to have + // fired. Assertions that a check DID fire should poll instead: the window + // is a lower bound on when the timer is eligible to run, and a loaded + // runner with a coarse timer can take considerably longer to get there. + function waitPastDeferredCheck(): Promise { + return new Promise((resolve) => + setTimeout(resolve, MIN_DEFERRED_CHECK_DELAY_MS * 4) + ); + } + + // Unlike createMockEvent above, this builds the real `Event` shape, which + // the duplicate-class skip needs: it reads `eventType` and `correlationId`. + let realEventCounter = 0; + function realEvent( + eventType: string, + correlationId: string | undefined + ): Event { + realEventCounter++; + return { + eventId: `evnt_${realEventCounter}`, + runId: 'wrun_test', + eventType, + correlationId, + eventData: {}, + createdAt: new Date(), + } as unknown as Event; + } + + /** + * A consumer for one entity: takes every event carrying `correlationId` + * and deregisters once it has taken `terminalType`. This is the shape the + * runtime's step/wait consumers have, and the reason a straggler for that + * id has no callback left to claim it. + */ + function entityConsumer(correlationId: string, terminalType: string) { + return vi.fn((event: Event | null) => { + if (event === null || event.correlationId !== correlationId) { + return EventConsumerResult.NotConsumed; + } + return event.eventType === terminalType + ? EventConsumerResult.Finished + : EventConsumerResult.Consumed; + }); + } + + function consumerFor( + events: Event[], + overrides: Partial<{ + onUnconsumedEvent: (event: Event) => void; + onDuplicateEvent: ( + event: Event, + firstEventType: Event['eventType'] + ) => void; + onConsumedEvent: (event: Event) => void; + }> = {} + ) { + return new EventsConsumer(events, { + onUnconsumedEvent: vi.fn(), + getPromiseQueue: () => Promise.resolve(), + // No deliveries are modeled here, so the gate is always open. + isDeliveryIdle: () => true, + ...overrides, + }); + } + + it('skips a step_started that repeats a class already in the log', async () => { + const corr = 'step_A'; + const events = [ + realEvent('step_created', corr), + realEvent('step_started', corr), + realEvent('step_completed', corr), + // Written by a concurrent replay working from a prefix that predates + // the completion, so it lands after it. + realEvent('step_started', corr), + ]; + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + }); + + consumer.subscribe(entityConsumer(corr, 'step_completed')); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(events.length); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + expect(onDuplicateEvent).toHaveBeenCalledTimes(1); + expect(onDuplicateEvent).toHaveBeenCalledWith(events[3], 'step_started'); + }); + + it('skips a step_created that repeats a class already in the log', async () => { + // Classes are tracked independently, so a completed step still has a + // recorded step_created and a second one is ignorable. + const corr = 'step_A'; + const events = [ + realEvent('step_created', corr), + realEvent('step_completed', corr), + realEvent('step_created', corr), + ]; + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + }); + + consumer.subscribe(entityConsumer(corr, 'step_completed')); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(events.length); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + expect(onDuplicateEvent).toHaveBeenCalledWith(events[2], 'step_created'); + }); + + it('skips a duplicate wait_completed ahead of parking it', async () => { + // wait_completed is parkable, so without the class check this would be + // held for a consumer that can never come and strand the walk. + const corr = 'wait_A'; + const events = [ + realEvent('wait_created', corr), + realEvent('wait_completed', corr), + realEvent('wait_completed', corr), + ]; + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + }); + + consumer.subscribe(entityConsumer(corr, 'wait_completed')); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(events.length); + expect(consumer.parkedSummary).toBeUndefined(); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + expect(onDuplicateEvent).toHaveBeenCalledWith( + events[2], + 'wait_completed' + ); + }); + + it('skips a duplicate run_started, which carries no correlation id', async () => { + const events = [ + realEvent('run_started', undefined), + realEvent('run_started', undefined), + ]; + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const onConsumedEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + onConsumedEvent, + }); + + // The runtime's run-lifecycle callback takes the first run_started and + // declines the rest rather than deregistering, since it still handles + // other run events. + let consumedRunStarted = false; + consumer.subscribe((event: Event | null) => { + if (event?.eventType !== 'run_started' || consumedRunStarted) { + return EventConsumerResult.NotConsumed; + } + consumedRunStarted = true; + return EventConsumerResult.Consumed; + }); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(events.length); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + expect(onDuplicateEvent).toHaveBeenCalledWith(events[1], 'run_started'); + expect(onConsumedEvent).toHaveBeenCalledTimes(1); + }); + + it('still reports an unconsumed event for a correlation id the log has nothing for', async () => { + const events = [ + realEvent('step_created', 'step_A'), + realEvent('step_started', 'step_A'), + realEvent('step_completed', 'step_A'), + // A different entity that no callback ever claims. wait_created is not + // parkable: its position is this replay's own decision record. + realEvent('wait_created', 'wait_B'), + ]; + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + }); + + consumer.subscribe(entityConsumer('step_A', 'step_completed')); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(3); + expect(onDuplicateEvent).not.toHaveBeenCalled(); + await vi.waitFor(() => { + expect(onUnconsumedEvent).toHaveBeenCalledWith(events[3]); + }); + }); + + it('does not let one class suppress another for the same entity', async () => { + // The step's outcome is in the log but its first attempt never wrote a + // step_started, so this one is not a repeat of anything and divergence + // is the right answer. + const corr = 'step_A'; + const events = [ + realEvent('step_created', corr), + realEvent('step_completed', corr), + realEvent('step_started', corr), + ]; + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + }); + + consumer.subscribe(entityConsumer(corr, 'step_completed')); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(2); + expect(onDuplicateEvent).not.toHaveBeenCalled(); + await vi.waitFor(() => { + expect(onUnconsumedEvent).toHaveBeenCalledWith(events[2]); + }); + }); + + it('does not track hook deliveries, whose consumers subscribe lazily', async () => { + // A hook legitimately fires many times under one id, so a second + // hook_received is not a repeat of a decided outcome. It keeps the + // parking path rather than being skipped. + const corr = 'hook_A'; + const events = [ + realEvent('hook_created', corr), + realEvent('hook_received', corr), + realEvent('hook_received', corr), + ]; + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + }); + + // Takes the create and the first delivery, then deregisters. + consumer.subscribe(entityConsumer(corr, 'hook_received')); + await waitPastDeferredCheck(); + + expect(onDuplicateEvent).not.toHaveBeenCalled(); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + expect(consumer.strandedEvent).toEqual(events[2]); + }); + + it('never takes an event a registered callback still wants', async () => { + // The skip is a last resort, consulted only after every callback + // declined, which is what lets a retry's step_started reach the live + // consumer and count as an attempt. + const corr = 'step_A'; + const events = [ + realEvent('step_created', corr), + realEvent('step_started', corr), + realEvent('step_retrying', corr), + realEvent('step_started', corr), + ]; + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + }); + + const callback = vi.fn((event: Event | null) => + event === null + ? EventConsumerResult.NotConsumed + : EventConsumerResult.Consumed + ); + consumer.subscribe(callback); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(events.length); + expect(callback).toHaveBeenCalledWith(events[3]); + expect(onDuplicateEvent).not.toHaveBeenCalled(); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + }); + + it('steps over a straggler without waiting out the deferred window', async () => { + // The window buys time for a consumer that has yet to register, and no + // such consumer can want an event of a class this replay already + // consumed. Paying it anyway costs the delay per straggler per replay, + // which is what makes a stormed run's log expensive to walk. + const corr = 'step_A'; + const stragglers = 3; + const events = [ + realEvent('step_created', corr), + realEvent('step_started', corr), + realEvent('step_completed', corr), + ...Array.from({ length: stragglers }, () => + realEvent('step_started', corr) + ), + ]; + // Run at the real delay: the point of the test is the difference + // between paying it and not, so shortening it would erase the signal. + vi.unstubAllEnvs(); + const consumer = consumerFor(events); + + const start = Date.now(); + consumer.subscribe(entityConsumer(corr, 'step_completed')); + await vi.waitFor( + () => { + expect(consumer.eventIndex).toBe(events.length); + }, + { interval: 1 } + ); + + // Deferring each straggler would cost one window apiece, so the walk + // finishing inside a single window means none of them went through the + // deferred check. + expect(Date.now() - start).toBeLessThan(DEFERRED_CHECK_DELAY_MS); + }); + + it('reports the first outcome when a repeat decides the class differently', async () => { + // Two replays raced a nondeterministic step to opposite results. The + // first one is what the workflow observed, on every replay; the second + // is dropped, and the drop is worth surfacing. + const corr = 'step_A'; + const events = [ + realEvent('step_created', corr), + realEvent('step_completed', corr), + realEvent('step_failed', corr), + ]; + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { onDuplicateEvent }); + + consumer.subscribe(entityConsumer(corr, 'step_completed')); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(events.length); + expect(onDuplicateEvent).toHaveBeenCalledWith( + events[2], + 'step_completed' + ); + }); + + it('leaves a duplicate run_cancelled to the parking path', async () => { + // Terminal run events have no class: nothing consumes them, so no class + // could ever be recorded for one. In production the runtime exits before + // replaying a body whose log already holds one, so this path is only + // reachable in a test — the point is that the skip does not claim to + // handle it. + const events = [ + realEvent('run_started', undefined), + realEvent('run_cancelled', undefined), + realEvent('run_cancelled', undefined), + ]; + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { onDuplicateEvent }); + + let consumedRunStarted = false; + consumer.subscribe((event: Event | null) => { + if (event?.eventType !== 'run_started' || consumedRunStarted) { + return EventConsumerResult.NotConsumed; + } + consumedRunStarted = true; + return EventConsumerResult.Consumed; + }); + await waitPastDeferredCheck(); + + expect(onDuplicateEvent).not.toHaveBeenCalled(); + expect(consumer.parkedSummary?.eventType).toBe('run_cancelled'); + }); + + it('does not advance the deterministic clock for a skipped event', async () => { + const corr = 'step_A'; + const events = [ + realEvent('step_created', corr), + realEvent('step_completed', corr), + realEvent('step_created', corr), + ]; + const onConsumedEvent = vi.fn(); + const consumer = consumerFor(events, { onConsumedEvent }); + + consumer.subscribe(entityConsumer(corr, 'step_completed')); + await waitPastDeferredCheck(); + + expect(consumer.eventIndex).toBe(events.length); + // The workflow body never observed the straggler, so a log containing it + // must produce the same timestamps as a log that does not. + expect(onConsumedEvent).toHaveBeenCalledTimes(2); + expect(onConsumedEvent).not.toHaveBeenCalledWith(events[2]); + }); + }); }); diff --git a/packages/core/src/events-consumer.ts b/packages/core/src/events-consumer.ts index 36647ed96c..a6686006b9 100644 --- a/packages/core/src/events-consumer.ts +++ b/packages/core/src/events-consumer.ts @@ -1,4 +1,4 @@ -import { type Event, envNumber } from '@workflow/world'; +import { type Event, entityEventClass, envNumber } from '@workflow/world'; import { eventsLogger } from './logger.js'; /** @@ -145,6 +145,16 @@ export interface EventsConsumerOptions { * downstream subscribe() calls have a chance to cancel the check first. */ onUnconsumedEvent: (event: Event) => void; + /** + * Callback invoked when an event is skipped because it repeats an event + * class the walk already consumed for the same entity. `firstEventType` is + * the type that recorded the class, which is the one the workflow observed. + * Diagnostics only: skipping is a normal outcome, not an error — though a + * `firstEventType` differing from `event.eventType` says the two writers + * decided the entity's outcome differently, which is worth more than an + * info log. + */ + onDuplicateEvent?: (event: Event, firstEventType: Event['eventType']) => void; /** * Returns the current promise queue. The unconsumed event check is chained * onto this queue so it only fires after all pending async work (e.g., @@ -188,8 +198,20 @@ export class EventsConsumer { * rather than parked for a consumer that cannot exist. */ private readonly resolved = new Set(); + /** + * `:` for every event class the walk has already + * consumed, mapped to the event type that recorded it. The type is kept so a + * repeat that decided the same class *differently* (a `step_failed` behind a + * `step_completed`) can be reported as more than a re-commit. See + * {@link EventsConsumer.firstEventTypeOfClass}. + */ + private readonly seenEventClasses = new Map(); private onConsumedEvent?: (event: Event) => void; private onUnconsumedEvent: (event: Event) => void; + private onDuplicateEvent?: ( + event: Event, + firstType: Event['eventType'] + ) => void; private getPromiseQueue: () => Promise; private isDeliveryIdle: () => boolean; private pendingUnconsumedCheck: Promise | null = null; @@ -204,6 +226,7 @@ export class EventsConsumer { this.eventIndex = 0; this.onConsumedEvent = options.onConsumedEvent; this.onUnconsumedEvent = options.onUnconsumedEvent; + this.onDuplicateEvent = options.onDuplicateEvent; this.getPromiseQueue = options.getPromiseQueue; this.isDeliveryIdle = options.isDeliveryIdle; } @@ -325,6 +348,15 @@ export class EventsConsumer { return; } if (!consumed) { + // Nobody wanted it. If it repeats a class this walk already consumed + // for the same entity it is a straggler from a concurrent replay: + // step over it in this pass rather than paying the deferred window for + // a consumer that cannot come (see `firstEventTypeOfClass`). + const firstType = this.firstEventTypeOfClass(currentEvent); + if (firstType !== undefined) { + this.skipDuplicateEvent(currentEvent, firstType); + continue; + } this.scheduleUnconsumedCheck(currentEvent, true); return; } @@ -361,6 +393,7 @@ export class EventsConsumer { resolutionKey(currentEvent.eventType, currentEvent.correlationId) ); } + this.recordEventClass(currentEvent); this.notifyConsumedEvent(currentEvent); } // remove the callback if it has finished @@ -432,6 +465,106 @@ export class EventsConsumer { return true; } + /** + * The key `event`'s class is tracked under, or `undefined` for the event + * types that belong to no class (`hook_received`, `hook_conflict`, + * `attr_set`, `run_created`) and are therefore never skipped. + * + * Run events carry no correlation id. They are classes of the run itself, so + * they all key off the same bucket. + */ + private eventClassKey(event: Event): string | undefined { + const eventClass = entityEventClass(event.eventType); + return eventClass === undefined + ? undefined + : `${eventClass}:${event.correlationId}`; + } + + /** + * Remembers that `event`'s class is now decided for its entity, if the type + * belongs to a class. First writer wins: the recorded type is the one the + * workflow observed, and a later repeat is measured against it. + */ + private recordEventClass(event: Event) { + const key = this.eventClassKey(event); + if (key !== undefined && !this.seenEventClasses.has(key)) { + this.seenEventClasses.set(key, event.eventType); + } + } + + /** + * The type that already decided `event`'s class for the same entity, or + * `undefined` when nothing has: a second `step_created` for one step, a + * second terminal outcome, a second `step_started` after the step's result is + * already in the log. + * + * Such an event is committed but inert. Concurrent replays write into one + * log without a currency guard, so a replay working from a prefix that + * predates another replay's write can commit its own copy of work the log + * already records. That copy cannot change what the workflow observed: the + * outcome was decided by the first event of the class and every later replay + * reads that same event at the same log position, so ignoring the straggler + * is deterministic across replays. + * + * Classes are tracked separately, so passing one does not suppress another. + * A step whose result is in the log still reaches its `step_created` and + * `step_started` consumers if it has yet to see those classes. + * + * Consulted only after every registered callback has declined the event, so + * it can never take an event a consumer wanted. A retry's `step_started` is + * claimed by the step's live consumer and counts as an attempt exactly as + * before, and a second `step_created` reaching a step that has not finished + * is likewise consumed rather than skipped; only the copies nobody claims are + * skipped. + * + * Unlike the divergence report, this does *not* wait out the deferred window + * first, and it does not need to. The window buys time for a consumer that + * has yet to register, and no such consumer can want this event: the class + * was recorded by a consumption in this same replay, which means the entity's + * consumer was registered and took an event of this class, and correlation + * ids are minted from a monotonic ULID per body position, so nothing later in + * the body registers a second consumer under this id. Waiting would cost + * `getDeferredCheckDelayMs()` per straggler per replay for information that + * cannot arrive — 0.75% of production runs carry at least one straggler, and + * the p99 among those carries 155. + * + * The invariant to preserve if hook identity ever becomes caller-supplied + * (an idempotency key rather than a minted id): two `createHook` calls in one + * body could then share a correlation id, and the second consumer's + * `hook_created` would be a repeat of a class this replay already recorded. + * That would make skipping wrong for `hook_created`, and is the reason the + * class map lives next to the event types rather than being inferred. + */ + private firstEventTypeOfClass(event: Event): Event['eventType'] | undefined { + const key = this.eventClassKey(event); + return key === undefined ? undefined : this.seenEventClasses.get(key); + } + + /** Steps the walk over a repeat of an already-consumed class. */ + private skipDuplicateEvent(event: Event, firstType: Event['eventType']) { + this.eventIndex++; + // Deliberately not routed through `notifyConsumedEvent`: the deterministic + // clock advances only on events the workflow actually observed. A skipped + // event is invisible to the workflow body, and a log that happens to + // contain one must produce the same timestamps as a log that does not. + eventsLogger.debug( + 'Skipping event that repeats a class already in the log', + { + eventId: event.eventId, + eventType: event.eventType, + firstEventType: firstType, + correlationId: event.correlationId, + } + ); + try { + this.onDuplicateEvent?.(event, firstType); + } catch (error) { + eventsLogger.error('onDuplicateEvent callback threw an error', { + error, + }); + } + } + private handleEndOfLog() { // Everything still parked is waiting for a consumer some later replay will // register, which is the whole point of parking — except once the log @@ -495,23 +628,43 @@ export class EventsConsumer { return; } this.pendingUnconsumedCheck = null; - if (mayPark) { - if (this.events[this.eventIndex] !== currentEvent) { - // An append() drain claimed it while the check was in flight. - // Only subscribe() cancels the check, so this is reachable. - return; - } - if (this.park(currentEvent)) { - this.consume(); - return; - } - } - this.onUnconsumedEvent(currentEvent); + this.resolveUnconsumedEvent(currentEvent, mayPark); }, getDeferredCheckDelayMs()); }); }); } + /** + * Decide what a still-unconsumed event is, now that the promise queue has + * drained and the delivery gate says the VM is not mid-reaction. + * + * `mayPark` is false only for the end-of-log recheck of an event {@link park} + * already holds. Nothing in the first branch applies to one of those: it is + * not the event at the cursor, so the identity guard is not meaningful, and + * it is parked already. + * + * A duplicate class never arrives here. {@link consume} steps over one in the + * pass that offered it, before this check is ever scheduled, and a class + * recorded while the check was in flight can only have been recorded by a + * consumption inside {@link consume}, whose next pass re-offers this event + * and steps over it there — leaving the identity guard above to drop the + * in-flight check. + */ + private resolveUnconsumedEvent(currentEvent: Event, mayPark: boolean) { + if (mayPark) { + if (this.events[this.eventIndex] !== currentEvent) { + // An append() drain claimed it while the check was in flight. + // Only subscribe() cancels the check, so this is reachable. + return; + } + if (this.park(currentEvent)) { + this.consume(); + return; + } + } + this.onUnconsumedEvent(currentEvent); + } + /** * Run `fn` once no data delivery is in flight, polling the way * `scheduleWhenIdle` does: let the promise queue drain, re-check a timer diff --git a/packages/core/src/hook-sleep-interaction.test.ts b/packages/core/src/hook-sleep-interaction.test.ts index 9505cbb17c..ea55e7868a 100644 --- a/packages/core/src/hook-sleep-interaction.test.ts +++ b/packages/core/src/hook-sleep-interaction.test.ts @@ -1,16 +1,12 @@ -import { WorkflowRuntimeError } from '@workflow/errors'; -import { withResolvers } from '@workflow/utils'; -import type { Event } from '@workflow/world'; -import * as nanoid from 'nanoid'; -import { monotonicFactory } from 'ulid'; import { afterEach, describe, expect, it, vi } from 'vitest'; -import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; -import type { WorkflowOrchestratorContext } from './private.js'; -import { ReplayPayloadCache } from './replay-payload-cache.js'; import { dehydrateStepReturnValue } from './serialization.js'; import { createUseStep } from './step.js'; -import { createContext } from './vm/index.js'; +import { + CORR_IDS, + runWithDiscontinuation, + setupWorkflowContext, +} from './test-support/orchestrator-context.js'; import { createCreateHook } from './workflow/hook.js'; import { createSleep } from './workflow/sleep.js'; @@ -28,88 +24,6 @@ import { createSleep } from './workflow/sleep.js'; * so suspensions wait for both async deserialization AND microtask deliveries. */ -function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { - const context = createContext({ - seed: 'test', - fixedTimestamp: 1753481739458, - }); - const ulid = monotonicFactory(() => context.globalThis.Math.random()); - const workflowStartedAt = context.globalThis.Date.now(); - const promiseQueueHolder = { current: Promise.resolve() }; - // Forward onUnconsumedEvent through ctx.onWorkflowError so tests that wire - // onWorkflowError to a discontinuation promise (see runWithDiscontinuation) - // actually observe false-positive unconsumed-event detections instead of - // silently dropping them. - const ctxRef: { current?: WorkflowOrchestratorContext } = {}; - const ctx: WorkflowOrchestratorContext = { - suspensionGeneration: 0, - runId: 'wrun_test', - encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), - globalThis: context.globalThis, - eventsConsumer: new EventsConsumer(events, { - // Fake context: no deliveries are modeled, so the gate is a no-op here. - isDeliveryIdle: () => true, - onUnconsumedEvent: (event) => { - ctxRef.current?.onWorkflowError( - new WorkflowRuntimeError( - `Unconsumed event in event log: eventType=${event.eventType}, correlationId=${event.correlationId}, eventId=${event.eventId}. This indicates a corrupted or invalid event log.` - ) - ); - }, - getPromiseQueue: () => promiseQueueHolder.current, - }), - invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), - generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => - new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) - ), - onWorkflowError: vi.fn(), - get promiseQueue() { - return promiseQueueHolder.current; - }, - set promiseQueue(value: Promise) { - promiseQueueHolder.current = value; - }, - pendingDeliveries: 0, - pendingDeliveryBarriers: new Map(), - }; - ctxRef.current = ctx; - return ctx; -} - -// Deterministic correlation IDs from the ULID generator with seed 'test' -const CORR_IDS = [ - '01K11TFZ62YS0YYFDQ3E8B9YCV', - '01K11TFZ62YS0YYFDQ3E8B9YCW', - '01K11TFZ62YS0YYFDQ3E8B9YCX', - '01K11TFZ62YS0YYFDQ3E8B9YCY', - '01K11TFZ62YS0YYFDQ3E8B9YCZ', - '01K11TFZ62YS0YYFDQ3E8B9YD0', -]; - -// ─── Helpers ─────────────────────────────────────────── - -async function runWithDiscontinuation( - ctx: WorkflowOrchestratorContext, - workflowFn: () => Promise -): Promise<{ result?: any; error?: any }> { - const workflowDiscontinuation = withResolvers(); - ctx.onWorkflowError = workflowDiscontinuation.reject; - - let result: any; - let error: any; - try { - result = await Promise.race([ - workflowFn(), - workflowDiscontinuation.promise, - ]); - } catch (err) { - error = err; - } - return { result, error }; -} - /** * Defines the full test suite for a given deserialization mode. * In 'async' mode, hydrateStepReturnValue is mocked with a 10ms delay diff --git a/packages/core/src/test-support/orchestrator-context.ts b/packages/core/src/test-support/orchestrator-context.ts new file mode 100644 index 0000000000..1ae7a78204 --- /dev/null +++ b/packages/core/src/test-support/orchestrator-context.ts @@ -0,0 +1,107 @@ +import { WorkflowRuntimeError } from '@workflow/errors'; +import { withResolvers } from '@workflow/utils'; +import type { Event } from '@workflow/world'; +import * as nanoid from 'nanoid'; +import { monotonicFactory } from 'ulid'; +import { vi } from 'vitest'; +import { EventsConsumer } from '../events-consumer.js'; +import type { WorkflowOrchestratorContext } from '../private.js'; +import { ReplayPayloadCache } from '../replay-payload-cache.js'; +import { createContext } from '../vm/index.js'; + +/** + * Builds an orchestrator context that replays a hand-written event log through + * the real workflow primitives (`createUseStep`, `createSleep`, + * `createCreateHook`), without a World or a VM entrypoint. + */ +export function setupWorkflowContext( + events: Event[], + options: { onDuplicateEvent?: (event: Event) => void } = {} +): WorkflowOrchestratorContext { + const context = createContext({ + seed: 'test', + fixedTimestamp: 1753481739458, + }); + const ulid = monotonicFactory(() => context.globalThis.Math.random()); + const workflowStartedAt = context.globalThis.Date.now(); + const promiseQueueHolder = { current: Promise.resolve() }; + // Forward onUnconsumedEvent through ctx.onWorkflowError so tests that wire + // onWorkflowError to a discontinuation promise (see runWithDiscontinuation) + // actually observe false-positive unconsumed-event detections instead of + // silently dropping them. + const ctxRef: { current?: WorkflowOrchestratorContext } = {}; + const ctx: WorkflowOrchestratorContext = { + suspensionGeneration: 0, + runId: 'wrun_test', + encryptionKey: undefined, + replayPayloadCache: new ReplayPayloadCache(undefined), + globalThis: context.globalThis, + eventsConsumer: new EventsConsumer(events, { + // Fake context: no deliveries are modeled, so the gate is a no-op here. + isDeliveryIdle: () => true, + onUnconsumedEvent: (event) => { + ctxRef.current?.onWorkflowError( + new WorkflowRuntimeError( + `Unconsumed event in event log: eventType=${event.eventType}, correlationId=${event.correlationId}, eventId=${event.eventId}. This indicates a corrupted or invalid event log.` + ) + ); + }, + onDuplicateEvent: options.onDuplicateEvent, + getPromiseQueue: () => promiseQueueHolder.current, + }), + invocationsQueue: new Map(), + generateUlid: () => ulid(workflowStartedAt), + generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => + new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) + ), + onWorkflowError: vi.fn(), + get promiseQueue() { + return promiseQueueHolder.current; + }, + set promiseQueue(value: Promise) { + promiseQueueHolder.current = value; + }, + pendingDeliveries: 0, + pendingDeliveryBarriers: new Map(), + }; + ctxRef.current = ctx; + return ctx; +} + +/** + * Deterministic correlation IDs from the ULID generator with seed 'test', in + * the order {@link setupWorkflowContext}'s generator mints them. + */ +export const CORR_IDS = [ + '01K11TFZ62YS0YYFDQ3E8B9YCV', + '01K11TFZ62YS0YYFDQ3E8B9YCW', + '01K11TFZ62YS0YYFDQ3E8B9YCX', + '01K11TFZ62YS0YYFDQ3E8B9YCY', + '01K11TFZ62YS0YYFDQ3E8B9YCZ', + '01K11TFZ62YS0YYFDQ3E8B9YD0', +]; + +/** + * Runs `workflowFn` against `ctx`, racing it against the context's error + * channel so a `WorkflowSuspension` or a detected divergence surfaces as + * `error` rather than hanging. + */ +export async function runWithDiscontinuation( + ctx: WorkflowOrchestratorContext, + workflowFn: () => Promise +): Promise<{ result?: any; error?: any }> { + const workflowDiscontinuation = withResolvers(); + ctx.onWorkflowError = workflowDiscontinuation.reject; + + let result: any; + let error: any; + try { + result = await Promise.race([ + workflowFn(), + workflowDiscontinuation.promise, + ]); + } catch (err) { + error = err; + } + return { result, error }; +} diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts index f47e1d1d7e..1351c064bb 100644 --- a/packages/core/src/workflow.test.ts +++ b/packages/core/src/workflow.test.ts @@ -4383,7 +4383,7 @@ describe('runWorkflow', () => { ).toEqual('sleep with date completed'); }); - it('should reject with WorkflowRuntimeError for duplicate wait_completed in event log', async () => { + it('should ignore a duplicate wait_completed and keep replaying', async () => { const ops: Promise[] = []; const workflowRunId = 'test-run-123'; const workflowRun: WorkflowRun = { @@ -4424,11 +4424,9 @@ describe('runWorkflow', () => { createdAt: new Date('2024-01-01T00:00:05.000Z'), }, { - // Duplicate wait_completed — all worlds enforce one wait_completed - // per correlationId, so this shape indicates a corrupted event log. - // Its position between the sleep's completion and the subsequent - // step events means it blocks event consumption until onUnconsumedEvent - // fires. + // Duplicate wait_completed. The wait's outcome was already decided by + // event-1, so this one is committed but inert and replay skips past + // it to the step events below rather than stalling on it. eventId: 'event-2', runId: workflowRunId, eventType: 'wait_completed', @@ -4461,22 +4459,28 @@ describe('runWorkflow', () => { }, ]; - await expect( - runWorkflow( - `const doWork = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("doWork"); + const result = await runWorkflow( + `const doWork = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("doWork"); const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")]; async function workflow() { await sleep('5s'); const result = await doWork(); return result; }${getWorkflowTransformCode('workflow')}`, - workflowRun, - events + workflowRun, + events + ); + expect( + await hydrateWorkflowReturnValue( + result as any, + workflowRunId, + noEncryptionKey, + ops ) - ).rejects.toThrow('Replay could not consume event'); + ).toEqual('step done'); }); - it('should reject with WorkflowRuntimeError for duplicate step_completed blocking subsequent events', async () => { + it('should ignore a duplicate step_completed and keep replaying', async () => { const ops: Promise[] = []; const workflowRunId = 'test-run-123'; const workflowRun: WorkflowRun = { @@ -4518,7 +4522,9 @@ describe('runWorkflow', () => { createdAt: new Date('2024-01-01T00:00:01.000Z'), }, { - // Duplicate step_completed - orphaned, blocks events below + // Duplicate step_completed. doWork1's result was already decided by + // event-1, so the second result is never observed and replay skips + // past it to doWork2's events. eventId: 'event-2', runId: workflowRunId, eventType: 'step_completed', @@ -4552,18 +4558,24 @@ describe('runWorkflow', () => { }, ]; - await expect( - runWorkflow( - `const doWork1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("doWork1"); + const result = await runWorkflow( + `const doWork1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("doWork1"); const doWork2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("doWork2"); async function workflow() { await doWork1(); return await doWork2(); }${getWorkflowTransformCode('workflow')}`, - workflowRun, - events + workflowRun, + events + ); + expect( + await hydrateWorkflowReturnValue( + result as any, + workflowRunId, + noEncryptionKey, + ops ) - ).rejects.toThrow(WorkflowRuntimeError); + ).toEqual('second done'); }); it('should reject with WorkflowRuntimeError for orphaned step_completed blocking workflow step', async () => { diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index 17cf8ea855..1ceca96e0d 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -429,6 +429,36 @@ async function createWorkflowSession({ ) ); }, + onDuplicateEvent: (event, firstEventType) => { + const details = { + workflowRunId: workflowRun.runId, + eventId: event.eventId, + eventType: event.eventType, + firstEventType, + correlationId: event.correlationId, + }; + if (firstEventType !== event.eventType) { + // Two writers reached opposite conclusions about one entity: a + // `step_failed` behind a `step_completed`, or the reverse. Ignoring it + // is still correct and still deterministic (replay reads the first one + // at the same position every time), but unlike a re-commit of the same + // outcome there is no reading of this where both writers were right, + // so the discarded outcome is worth an error in the run's logs. + runtimeLogger.error( + 'Ignoring event that decides an already-decided outcome differently', + details + ); + return; + } + // Not an error: the first event of this class decided the outcome at a + // lower log position and replay reads that one. Logged because a + // straggler is still evidence of two replays writing for the same + // entity, which is worth seeing when diagnosing a run. + runtimeLogger.info( + 'Ignoring event that repeats a class already in the event log', + details + ); + }, getPromiseQueue: () => promiseQueueHolder.current, isDeliveryIdle: () => deliveryIdleHolder.current(), }); @@ -462,6 +492,7 @@ async function createWorkflowSession({ // Consume run lifecycle events - these are structural events that don't // need special handling in the workflow, but must be consumed to advance // past them in the event log + let consumedRunStarted = false; workflowContext.eventsConsumer.subscribe((event) => { if (!event) { return EventConsumerResult.NotConsumed; @@ -472,8 +503,16 @@ async function createWorkflowSession({ return EventConsumerResult.Consumed; } - // Consume run_started - every run has exactly one + // Consume the first run_started. Two replays can each write one (the + // create is idempotent for a run already running, which makes the write + // safe, not single-shot); the first is the one the replay observes, and + // the consumer skips the rest rather than advancing the workflow clock + // twice. if (event.eventType === 'run_started') { + if (consumedRunStarted) { + return EventConsumerResult.NotConsumed; + } + consumedRunStarted = true; return EventConsumerResult.Consumed; } diff --git a/packages/core/src/workflow/sleep.test.ts b/packages/core/src/workflow/sleep.test.ts index 1dd45b52ce..7780bc05f1 100644 --- a/packages/core/src/workflow/sleep.test.ts +++ b/packages/core/src/workflow/sleep.test.ts @@ -4,7 +4,7 @@ import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; import { describe, expect, it, vi } from 'vitest'; -import { EventsConsumer } from '../events-consumer.js'; +import { DEFERRED_CHECK_DELAY_MS, EventsConsumer } from '../events-consumer.js'; import { WorkflowSuspension } from '../global.js'; import type { WorkflowOrchestratorContext } from '../private.js'; import { ReplayPayloadCache } from '../replay-payload-cache.js'; @@ -368,11 +368,11 @@ describe('createSleep', () => { expect(ctx.onWorkflowError).not.toHaveBeenCalled(); }); - it('should raise ReplayDivergenceError when duplicate wait_completed events cannot be consumed', async () => { - // When the event log has 2 wait_completed for a single wait_created, - // the first wait_completed removes the callback (Finished), but the second - // wait_completed has no consumer. The onUnconsumedEvent callback should - // trigger a ReplayDivergenceError via onWorkflowError. + it('should ignore a duplicate wait_completed rather than reporting divergence', async () => { + // When the event log has 2 wait_completed for a single wait_created, the + // first one removes the callback (Finished) and the second has no consumer + // left. That is not divergence: the wait's outcome was already decided by + // the first, so the duplicate is committed but inert and replay skips it. const ctx = setupWorkflowContext([ { eventId: 'evnt_0', @@ -406,16 +406,18 @@ describe('createSleep', () => { }, ]); - const errorReceived = withResolvers(); - ctx.onWorkflowError = errorReceived.resolve; - const sleep = createSleep(ctx); await sleep('1s'); - // The duplicate wait_completed at index 2 is orphaned and triggers the error - const workflowError = await errorReceived.promise; - expect(workflowError).toBeInstanceOf(ReplayDivergenceError); - expect(workflowError?.message).toContain('evnt_2'); + // Wait past the deferred unconsumed-event window so a check that was not + // skipped would have fired by now. + await new Promise((resolve) => + setTimeout(resolve, DEFERRED_CHECK_DELAY_MS * 2) + ); + + expect(ctx.onWorkflowError).not.toHaveBeenCalled(); + // The cursor moved past the duplicate instead of stalling on it. + expect(ctx.eventsConsumer.eventIndex).toBe(3); }); it('should resolve with void when wait_completed', async () => { diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index ba37c947ee..fab49f591c 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -89,6 +89,59 @@ export function isTerminalStepEventType( return TERMINAL_STEP_EVENT_TYPES.includes(eventType as TerminalStepEventType); } +/** + * Groups event types into the classes a replay tracks per entity: the entity + * named by the event's `correlationId`, or the run itself for run events, + * which carry none. + * + * Types that share a class are the mutually exclusive outcomes of one + * decision, so the log records the class once and the first event of it is the + * one that counts: a step either completes or fails. + * + * Classes are independent of each other. A step whose result is in the log has + * still recorded exactly one `step_created`, and can still record another + * `step_started` if an attempt is running somewhere. What a class bounds is + * which events can be *ignored*: a replay may pass over an event whose class + * it already recorded for that entity and which no consumer wants (see + * `EventsConsumer`), and only then. + * + * Note the omissions, all of them types a mapping would be dead weight for. + * `hook_received` and `hook_conflict` are deliveries whose consumer subscribes + * lazily, `attr_set` is written on every attribute write, and `run_created` + * precedes every replay. The terminal run types are absent for a different + * reason: recording a class requires a consumer to take an event of it, and no + * consumer takes `run_completed` / `run_failed` / `run_cancelled` — the runtime + * exits before replaying the body once the log holds one, so they never reach a + * consumer at all. An entry for them could never match. + */ +const ENTITY_EVENT_CLASS_BY_TYPE = { + step_created: 'step_created', + step_started: 'step_started', + step_retrying: 'step_retrying', + step_completed: 'step_terminal', + step_failed: 'step_terminal', + wait_created: 'wait_created', + wait_completed: 'wait_completed', + hook_created: 'hook_created', + hook_disposed: 'hook_disposed', + run_started: 'run_started', +} as const satisfies Partial>; + +export type EntityEventClass = + (typeof ENTITY_EVENT_CLASS_BY_TYPE)[keyof typeof ENTITY_EVENT_CLASS_BY_TYPE]; + +/** + * The per-entity class `eventType` belongs to, or `undefined` when it belongs + * to none. See {@link ENTITY_EVENT_CLASS_BY_TYPE}. + */ +export function entityEventClass( + eventType: string +): EntityEventClass | undefined { + return ( + ENTITY_EVENT_CLASS_BY_TYPE as Record + )[eventType]; +} + const HookLifecycleEventTypeSchema = EventTypeSchema.extract([ 'hook_created', 'hook_received', diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index 50ce1b9a0a..0fb00afc91 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -38,6 +38,7 @@ export { EVENT_DATA_REF_FIELDS, EventSchema, EventTypeSchema, + entityEventClass, getEventDataPayloadField, getEventDataRefFields, HOOK_EVENTS_REQUIRING_EXISTENCE,