From 9a5e0fe48c710fbc5fee831d94660aa07ec26163 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 10:50:18 -0700 Subject: [PATCH 1/6] Ignore duplicate events per event class instead of failing the run --- .changeset/dull-hats-melt.md | 6 + packages/core/src/duplicate-events.test.ts | 153 +++++++++ packages/core/src/events-consumer.test.ts | 297 ++++++++++++++++++ packages/core/src/events-consumer.ts | 145 ++++++++- .../core/src/hook-sleep-interaction.test.ts | 96 +----- .../src/test-support/orchestrator-context.ts | 107 +++++++ packages/core/src/workflow.test.ts | 52 +-- packages/core/src/workflow.ts | 26 +- packages/core/src/workflow/sleep.test.ts | 28 +- packages/world/src/events.ts | 52 +++ packages/world/src/index.ts | 1 + 11 files changed, 825 insertions(+), 138 deletions(-) create mode 100644 .changeset/dull-hats-melt.md create mode 100644 packages/core/src/duplicate-events.test.ts create mode 100644 packages/core/src/test-support/orchestrator-context.ts 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..1ee3225578 --- /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]); + }); + + 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]); + }); + + 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..1434b717b1 100644 --- a/packages/core/src/events-consumer.test.ts +++ b/packages/core/src/events-consumer.test.ts @@ -734,4 +734,301 @@ describe('EventsConsumer', () => { expect(await unconsumedReceived.promise).toEqual(event); }); }); + + describe('duplicate event classes', () => { + // Waits past the deferred unconsumed-event window so a check that was not + // cancelled has definitely fired. + function waitPastDeferredCheck(): Promise { + return new Promise((resolve) => + setTimeout(resolve, DEFERRED_CHECK_DELAY_MS * 2) + ); + } + + // 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) => 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]); + }); + + 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]); + }); + + 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]); + }); + + 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]); + 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(); + 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(); + 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('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..47d0de47b5 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,12 @@ 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 log already records for the same entity. Diagnostics only: + * skipping is a normal outcome, not an error. + */ + onDuplicateEvent?: (event: Event) => 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 +194,14 @@ 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. See {@link EventsConsumer.isDuplicateEvent}. + */ + private readonly seenEventClasses = new Set(); private onConsumedEvent?: (event: Event) => void; private onUnconsumedEvent: (event: Event) => void; + private onDuplicateEvent?: (event: Event) => void; private getPromiseQueue: () => Promise; private isDeliveryIdle: () => boolean; private pendingUnconsumedCheck: Promise | null = null; @@ -204,6 +216,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; } @@ -361,6 +374,7 @@ export class EventsConsumer { resolutionKey(currentEvent.eventType, currentEvent.correlationId) ); } + this.recordEventClass(currentEvent); this.notifyConsumedEvent(currentEvent); } // remove the callback if it has finished @@ -432,6 +446,93 @@ 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 the class `event` belongs to, if it belongs to one. */ + private recordEventClass(event: Event) { + const key = this.eventClassKey(event); + if (key !== undefined) { + this.seenEventClasses.add(key); + } + } + + /** + * Whether `event` repeats a class the walk already consumed for the same + * entity: 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 from the deferred check only, once every registered callback has + * declined the event and the delivery gate says the VM is not mid-reaction, + * so it can never take an event a consumer wanted or one a consumer is still + * on its way to wanting. A retry's `step_started` is claimed by the step's + * live consumer and counts as an attempt exactly as before; only the copies + * nobody claims are skipped. It cannot starve a consumer that has yet to + * register either: a consumer claims its entity's events in log order from + * the moment the body creates it, so a class the walk already consumed was + * consumed with that consumer registered, and correlation ids are minted + * monotonically, so no future consumer claims this id. + * + * Checked ahead of {@link park} rather than after it. The two mapped types + * that are also parkable would otherwise be handled inconsistently: a + * repeated `wait_completed` is declined by park anyway (its {@link resolved} + * guard asks the same question for one-shot types), but a repeated + * `run_cancelled` would be parked, and a parked event nothing claims strands + * the walk into the divergence this skip exists to avoid. + */ + private isDuplicateEvent(event: Event): boolean { + const key = this.eventClassKey(event); + return key !== undefined && this.seenEventClasses.has(key); + } + + /** Steps the walk over a repeat of an already-consumed class. */ + private skipDuplicateEvent(event: Event) { + 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, + correlationId: event.correlationId, + } + ); + try { + this.onDuplicateEvent?.(event); + } 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 +596,41 @@ 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 neither the identity guard nor a skip + * (which steps the cursor) is meaningful, and it is parked already. + */ + 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.isDuplicateEvent(currentEvent)) { + this.skipDuplicateEvent(currentEvent); + this.consume(); + 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..244648df2e 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -429,6 +429,21 @@ async function createWorkflowSession({ ) ); }, + onDuplicateEvent: (event) => { + // 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', + { + workflowRunId: workflowRun.runId, + eventId: event.eventId, + eventType: event.eventType, + correlationId: event.correlationId, + } + ); + }, getPromiseQueue: () => promiseQueueHolder.current, isDeliveryIdle: () => deliveryIdleHolder.current(), }); @@ -462,6 +477,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 +488,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..290885ddfd 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -89,6 +89,58 @@ 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, a run either completes, + * fails, or is cancelled. + * + * 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. `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. + */ +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', + run_completed: 'run_terminal', + run_failed: 'run_terminal', + run_cancelled: 'run_terminal', +} 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, From b5b60921b7587f8eab75b99866f0dd99d86cc247 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 12:43:44 -0700 Subject: [PATCH 2/6] Mark ignored duplicate events in the observability UI Concurrent replays of one run share an event log, so a replay working from a stale prefix can commit a write the log already records. The runtime passes over those. The UI showed them as ordinary progress and let them move derived state. Derive the set of passed-over events from the log with `entityEventClass`, restricted to the classes a run records at most once per entity. A retried step legitimately repeats `step_started` and `step_retrying`, one per attempt, so those two are excluded and never marked. Marked events read greyed out in the sidebar event list and the events table, with a tooltip saying the event was ignored. They are also kept out of the derived step status, the queued/ran durations, and trace span geometry, where a second terminal event would otherwise stretch a step to whenever the losing replay happened to commit. --- .changeset/silly-pears-jam.md | 5 + .../src/components/event-list-view.tsx | 87 ++++++++---- .../src/components/sidebar/events-list.tsx | 23 ++- .../components/ui/duplicate-event-tooltip.tsx | 44 ++++++ packages/web-shared/src/index.ts | 4 + .../src/lib/duplicate-events.test.ts | 134 ++++++++++++++++++ .../web-shared/src/lib/duplicate-events.ts | 89 ++++++++++++ .../src/lib/event-materialization.test.ts | 87 ++++++++++++ .../src/lib/event-materialization.ts | 9 ++ .../web-shared/src/lib/trace-builder.test.ts | 53 +++++++ packages/web-shared/src/lib/trace-builder.ts | 15 +- 11 files changed, 516 insertions(+), 34 deletions(-) create mode 100644 .changeset/silly-pears-jam.md create mode 100644 packages/web-shared/src/components/ui/duplicate-event-tooltip.tsx create mode 100644 packages/web-shared/src/lib/duplicate-events.test.ts create mode 100644 packages/web-shared/src/lib/duplicate-events.ts create mode 100644 packages/web-shared/src/lib/event-materialization.test.ts create mode 100644 packages/web-shared/src/lib/trace-builder.test.ts diff --git a/.changeset/silly-pears-jam.md b/.changeset/silly-pears-jam.md new file mode 100644 index 0000000000..96d1369efa --- /dev/null +++ b/.changeset/silly-pears-jam.md @@ -0,0 +1,5 @@ +--- +'@workflow/web-shared': patch +--- + +Mark ignored duplicate events in the observability UI and exclude them from derived step status, durations, and trace spans diff --git a/packages/web-shared/src/components/event-list-view.tsx b/packages/web-shared/src/components/event-list-view.tsx index 38ba75e6cb..87685620e8 100644 --- a/packages/web-shared/src/components/event-list-view.tsx +++ b/packages/web-shared/src/components/event-list-view.tsx @@ -10,6 +10,7 @@ import type { } from 'react'; import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { Virtuoso, type VirtuosoHandle } from 'react-virtuoso'; +import { findDuplicateEventIds } from '../lib/duplicate-events'; import { type ExactIdSearchResult, type ExactWorkflowSearchIdKind, @@ -23,6 +24,7 @@ import { AttrSetEventBlock } from './sidebar/attributes-block'; import { ContextCardProvider } from './ui/context-card'; import { DataInspector, DecryptClickContext } from './ui/data-inspector'; import { DecryptButton } from './ui/decrypt-button'; +import { DuplicateEventTooltip } from './ui/duplicate-event-tooltip'; import { ErrorStackBlock, isStructuredError, @@ -190,15 +192,21 @@ export interface DurationInfo { * Build a map from correlationId → duration info by diffing * created ↔ started (queued) and started ↔ completed/failed/cancelled (ran). * Also computes run-level durations under the key '__run__'. + * + * Events the runtime passed over as repeats are excluded: a second + * `step_completed` written by a concurrent replay would otherwise stretch the + * step's measured runtime to whenever that replay happened to commit. */ export function buildDurationMap(events: Event[]): Map { + const duplicateEventIds = findDuplicateEventIds(events); + // Process events in chronological order so the result doesn't depend on // the caller's sort direction. Retried steps emit multiple `step_started` // events for the same correlationId; the queued duration must be measured // against the first one, not the last. - const chronological = [...events].sort( - (a, b) => getEffectiveEventTime(a) - getEffectiveEventTime(b) - ); + const chronological = [...events] + .filter((event) => !duplicateEventIds.has(event.eventId)) + .sort((a, b) => getEffectiveEventTime(a) - getEffectiveEventTime(b)); const createdTimes = new Map(); const firstStartedTimes = new Map(); @@ -832,6 +840,7 @@ export function EventRow({ onEncryptedDataDetected, suppressGroupDimming = false, showSeparateEventOccurrenceTimestamps = false, + isDuplicate = false, }: { event: Event; index: number; @@ -856,6 +865,8 @@ export function EventRow({ suppressGroupDimming?: boolean; /** Show occurredAt separately instead of folding it into the Created timestamp. */ showSeparateEventOccurrenceTimestamps?: boolean; + /** The event repeats a class already in the log, so the runtime ignored it. */ + isDuplicate?: boolean; }) { const [isLoading, setIsLoading] = useState(false); const [loadedEventData, setLoadedEventData] = useState( @@ -1095,43 +1106,49 @@ export function EventRow({ {/* Event Type */}
- + - {isPulsing && ( - - )} + > + {isPulsing && ( + + )} + + + {formatEventType(event.eventType)} - {formatEventType(event.eventType)} - +
{/* Name */} @@ -1310,6 +1327,17 @@ function EventListViewInner({ ); }, [events, effectiveSortOrder, isExactSearchActive, searchResults]); + // Events the runtime passed over as repeats. Computed from the source list + // rather than `sortedEvents` because which occurrence counted is a property + // of the log, not of the direction the table happens to be sorted in. + const duplicateEventIds = useMemo( + () => + findDuplicateEventIds( + isExactSearchActive ? (searchResults ?? []) : (events ?? []) + ), + [events, isExactSearchActive, searchResults] + ); + // Detect encrypted fields across all loaded events (inline eventData). const hasEncryptedInlineData = useMemo(() => { const sourceEvents = isExactSearchActive ? searchResults : events; @@ -1804,6 +1832,7 @@ function EventListViewInner({ encryptionKey={encryptionKey} onEncryptedDataDetected={handleEncryptedDataDetected} suppressGroupDimming={isExactSearchActive} + isDuplicate={duplicateEventIds.has(ev.eventId)} showSeparateEventOccurrenceTimestamps={ showSeparateEventOccurrenceTimestamps } diff --git a/packages/web-shared/src/components/sidebar/events-list.tsx b/packages/web-shared/src/components/sidebar/events-list.tsx index 5674158d52..c1ae4b0030 100644 --- a/packages/web-shared/src/components/sidebar/events-list.tsx +++ b/packages/web-shared/src/components/sidebar/events-list.tsx @@ -2,6 +2,7 @@ import { type Event, getEventDataRefFields } from '@workflow/world'; import { useCallback, useLayoutEffect, useMemo, useRef, useState } from 'react'; +import { findDuplicateEventIds } from '../../lib/duplicate-events'; import { hasEncryptedFields, isExpiredMarker } from '../../lib/hydration'; import { Collapsible, @@ -10,6 +11,7 @@ import { CollapsibleTrigger, } from '../ui/collapsible'; import { RunClickContext, StreamClickContext } from '../ui/data-inspector'; +import { DuplicateEventTooltip } from '../ui/duplicate-event-tooltip'; import { ErrorCard } from '../ui/error-card'; import { ErrorStackBlock, isStructuredError } from '../ui/error-stack-block'; import { Skeleton } from '../ui/skeleton'; @@ -47,6 +49,7 @@ function EventItem({ onLoadEventData, encryptionKey, showSeparateEventOccurrenceTimestamps = false, + isDuplicate = false, }: { event: Event; onLoadEventData?: (event: Event) => Promise; @@ -54,6 +57,8 @@ function EventItem({ encryptionKey?: Uint8Array; /** Show occurredAt separately instead of folding it into the Created timestamp. */ showSeparateEventOccurrenceTimestamps?: boolean; + /** The event repeats a class already in the log, so the runtime ignored it. */ + isDuplicate?: boolean; }) { const [loadedData, setLoadedData] = useState(null); const [isLoading, setIsLoading] = useState(false); @@ -135,9 +140,15 @@ function EventItem({ >
- - {event.eventType} - + + + {event.eventType} + + {displayedCreatedAtTime} @@ -317,6 +328,11 @@ export function EventsList({ [events] ); + const duplicateEventIds = useMemo( + () => findDuplicateEventIds(events), + [events] + ); + const hasEvents = sortedEvents.length > 0 && !error; if (!hasEvents && !isLoading) { @@ -352,6 +368,7 @@ export function EventsList({ showSeparateEventOccurrenceTimestamps={ showSeparateEventOccurrenceTimestamps } + isDuplicate={duplicateEventIds.has(event.eventId)} /> ))}
diff --git a/packages/web-shared/src/components/ui/duplicate-event-tooltip.tsx b/packages/web-shared/src/components/ui/duplicate-event-tooltip.tsx new file mode 100644 index 0000000000..59e454a000 --- /dev/null +++ b/packages/web-shared/src/components/ui/duplicate-event-tooltip.tsx @@ -0,0 +1,44 @@ +'use client'; + +import type { ReactNode } from 'react'; +import { DUPLICATE_EVENT_MESSAGE } from '../../lib/duplicate-events'; +import { + Tooltip, + TooltipContent, + TooltipProvider, + TooltipTrigger, +} from './tooltip'; + +/** + * Explains why an event is shown greyed out: it repeats a class the log + * already records for the same entity, so the runtime passed over it. + * + * Renders `children` untouched when `isDuplicate` is false, so a call site can + * wrap an event label unconditionally. Mounts its own {@link TooltipProvider} + * so it works in the sidebar and the events table alike; nesting one inside an + * existing provider is harmless. + */ +export function DuplicateEventTooltip({ + isDuplicate, + children, +}: { + isDuplicate: boolean; + children: ReactNode; +}): ReactNode { + if (!isDuplicate) return children; + + return ( + + + {children} + + {DUPLICATE_EVENT_MESSAGE} + + + + ); +} diff --git a/packages/web-shared/src/index.ts b/packages/web-shared/src/index.ts index 3d65e872ee..cbfd776f5b 100644 --- a/packages/web-shared/src/index.ts +++ b/packages/web-shared/src/index.ts @@ -10,6 +10,10 @@ export { stepEventsToStepEntity, waitEventsToWaitEntity, } from './components/workflow-traces/trace-span-construction'; +export { + DUPLICATE_EVENT_MESSAGE, + findDuplicateEventIds, +} from './lib/duplicate-events'; export type { EventAnalysis } from './lib/event-analysis'; export { analyzeEvents, diff --git a/packages/web-shared/src/lib/duplicate-events.test.ts b/packages/web-shared/src/lib/duplicate-events.test.ts new file mode 100644 index 0000000000..484875beee --- /dev/null +++ b/packages/web-shared/src/lib/duplicate-events.test.ts @@ -0,0 +1,134 @@ +import type { Event, EventType } from '@workflow/world'; +import { describe, expect, it } from 'vitest'; +import { findDuplicateEventIds } from './duplicate-events'; + +const BASE_TIME = Date.parse('2026-01-01T00:00:00.000Z'); + +let nextId = 0; + +function event( + eventType: EventType, + options: { + correlationId?: string; + /** Seconds after the fixture epoch. Defaults to insertion order. */ + at?: number; + eventId?: string; + } = {} +): Event { + nextId += 1; + const offsetSeconds = options.at ?? nextId; + return { + eventId: options.eventId ?? `evt_${nextId}`, + runId: 'run_1', + eventType, + correlationId: options.correlationId, + createdAt: new Date(BASE_TIME + offsetSeconds * 1000), + occurredAt: new Date(BASE_TIME + offsetSeconds * 1000), + eventData: {}, + } as unknown as Event; +} + +describe('findDuplicateEventIds', () => { + it('returns nothing for a log with no repeats', () => { + const events = [ + event('run_created'), + event('run_started'), + event('step_created', { correlationId: 'step_a' }), + event('step_started', { correlationId: 'step_a' }), + event('step_completed', { correlationId: 'step_a' }), + event('run_completed'), + ]; + + expect(findDuplicateEventIds(events)).toEqual(new Set()); + }); + + it('flags every repeat after the first of a once-per-entity class', () => { + const first = event('step_created', { correlationId: 'step_a' }); + const second = event('step_created', { correlationId: 'step_a' }); + const third = event('step_created', { correlationId: 'step_a' }); + + expect(findDuplicateEventIds([first, second, third])).toEqual( + new Set([second.eventId, third.eventId]) + ); + }); + + it('treats completed and failed as one terminal class', () => { + // A concurrent replay writing the other outcome does not move the step off + // the outcome the run acted on. + const failed = event('step_failed', { correlationId: 'step_a' }); + const completed = event('step_completed', { correlationId: 'step_a' }); + + expect(findDuplicateEventIds([failed, completed])).toEqual( + new Set([completed.eventId]) + ); + }); + + it('keys on the correlation id, so sibling entities never collide', () => { + const events = [ + event('step_created', { correlationId: 'step_a' }), + event('step_created', { correlationId: 'step_b' }), + event('step_completed', { correlationId: 'step_a' }), + event('step_completed', { correlationId: 'step_b' }), + ]; + + expect(findDuplicateEventIds(events)).toEqual(new Set()); + }); + + it('does not flag the repeated events of a retried step', () => { + // Each attempt legitimately records its own start, and each retryable + // failure its own step_retrying. Both are consumed, not passed over. + const events = [ + event('step_created', { correlationId: 'step_a' }), + event('step_started', { correlationId: 'step_a' }), + event('step_retrying', { correlationId: 'step_a' }), + event('step_started', { correlationId: 'step_a' }), + event('step_retrying', { correlationId: 'step_a' }), + event('step_started', { correlationId: 'step_a' }), + event('step_completed', { correlationId: 'step_a' }), + ]; + + expect(findDuplicateEventIds(events)).toEqual(new Set()); + }); + + it('does not flag repeated hook deliveries', () => { + const events = [ + event('hook_created', { correlationId: 'hook_a' }), + event('hook_received', { correlationId: 'hook_a' }), + event('hook_received', { correlationId: 'hook_a' }), + event('hook_disposed', { correlationId: 'hook_a' }), + ]; + + expect(findDuplicateEventIds(events)).toEqual(new Set()); + }); + + it('flags run-level repeats, which carry no correlation id', () => { + const started = event('run_started'); + const startedAgain = event('run_started'); + const completed = event('run_completed'); + const cancelled = event('run_cancelled'); + + expect( + findDuplicateEventIds([started, startedAgain, completed, cancelled]) + ).toEqual(new Set([startedAgain.eventId, cancelled.eventId])); + }); + + it('picks the earliest occurrence whichever way the caller sorted', () => { + const first = event('wait_created', { at: 1, correlationId: 'wait_a' }); + const second = event('wait_created', { at: 2, correlationId: 'wait_a' }); + + const ascending = findDuplicateEventIds([first, second]); + const descending = findDuplicateEventIds([second, first]); + + expect(ascending).toEqual(new Set([second.eventId])); + expect(descending).toEqual(ascending); + }); + + it('breaks timestamp ties on the order the log lists them in', () => { + const first = event('wait_completed', { at: 5, correlationId: 'wait_a' }); + const second = event('wait_completed', { at: 5, correlationId: 'wait_a' }); + + expect(findDuplicateEventIds([first, second])).toEqual( + new Set([second.eventId]) + ); + }); +}); diff --git a/packages/web-shared/src/lib/duplicate-events.ts b/packages/web-shared/src/lib/duplicate-events.ts new file mode 100644 index 0000000000..d9e647d390 --- /dev/null +++ b/packages/web-shared/src/lib/duplicate-events.ts @@ -0,0 +1,89 @@ +import { + type EntityEventClass, + type Event, + entityEventClass, +} from '@workflow/world'; + +/** + * Copy shown on an event the runtime passed over as a repeat. + */ +export const DUPLICATE_EVENT_MESSAGE = + 'Multiple processes resuming may result in duplicate events. This event was ignored and had no effect.'; + +/** + * Event classes a run records at most once per entity. A second event of one + * of these classes for the same entity is a repeat: the first one is what the + * replay acted on, and the runtime passes over the rest. + * + * `step_started` and `step_retrying` are the two classes deliberately left + * out. A retried step records one of each per attempt, and the runtime + * consumes them, so a repeat there is an extra attempt rather than an ignored + * event. + */ +const ONCE_PER_ENTITY_CLASSES: ReadonlySet = new Set([ + 'step_created', + 'step_terminal', + 'wait_created', + 'wait_completed', + 'hook_created', + 'hook_disposed', + 'run_started', + 'run_terminal', +]); + +function entityKey(event: Event): string | undefined { + const eventClass = entityEventClass(event.eventType); + if (eventClass === undefined || !ONCE_PER_ENTITY_CLASSES.has(eventClass)) { + return undefined; + } + // Run events carry no correlation id, so they key on the class alone. + return `${eventClass}:${event.correlationId ?? ''}`; +} + +function effectiveTime(event: Event): number { + const occurredAt = event.occurredAt; + if (occurredAt != null) { + const parsed = + occurredAt instanceof Date ? occurredAt : new Date(String(occurredAt)); + if (!Number.isNaN(parsed.getTime())) return parsed.getTime(); + } + return new Date(event.createdAt).getTime(); +} + +/** + * Event ids that repeat a class the log already records for the same entity. + * + * Concurrent replays of one run share an event log, so a replay working from + * a stale prefix can commit a write the log already has. The runtime passes + * over those (see `EventsConsumer` in `@workflow/core`); this reports them so + * the UI can say so rather than showing them as ordinary progress. + * + * Order is decided here, not by the caller: the earliest event of a class is + * the one that counted, whichever direction the view happens to sort in. The + * answer is exact only over a complete event list. A partial list (pagination, + * an exact-id search) can miss a repeat whose first occurrence is absent, + * which under-reports rather than mislabels. + */ +export function findDuplicateEventIds(events: readonly Event[]): Set { + const duplicates = new Set(); + if (events.length < 2) return duplicates; + + const ordered = events + .map((event, index) => ({ event, index })) + .sort( + (a, b) => + effectiveTime(a.event) - effectiveTime(b.event) || a.index - b.index + ); + + const seen = new Set(); + for (const { event } of ordered) { + const key = entityKey(event); + if (key === undefined) continue; + if (seen.has(key)) { + duplicates.add(event.eventId); + continue; + } + seen.add(key); + } + return duplicates; +} diff --git a/packages/web-shared/src/lib/event-materialization.test.ts b/packages/web-shared/src/lib/event-materialization.test.ts new file mode 100644 index 0000000000..4ed98a17f9 --- /dev/null +++ b/packages/web-shared/src/lib/event-materialization.test.ts @@ -0,0 +1,87 @@ +import type { Event, EventType } from '@workflow/world'; +import { describe, expect, it } from 'vitest'; +import { materializeSteps } from './event-materialization'; + +const BASE_TIME = Date.parse('2026-01-01T00:00:00.000Z'); + +let nextId = 0; + +function event( + eventType: EventType, + options: { correlationId?: string; at?: number } = {} +): Event { + nextId += 1; + const offsetSeconds = options.at ?? nextId; + return { + eventId: `evt_${nextId}`, + runId: 'run_1', + eventType, + correlationId: options.correlationId, + createdAt: new Date(BASE_TIME + offsetSeconds * 1000), + occurredAt: new Date(BASE_TIME + offsetSeconds * 1000), + eventData: eventType === 'step_created' ? { stepName: 'doWork' } : {}, + } as unknown as Event; +} + +describe('materializeSteps', () => { + it('derives status and timings from the run of a well-formed step', () => { + const events = [ + event('step_created', { correlationId: 'step_a', at: 1 }), + event('step_started', { correlationId: 'step_a', at: 2 }), + event('step_completed', { correlationId: 'step_a', at: 5 }), + ]; + + const [step] = materializeSteps(events); + + expect(step.status).toBe('completed'); + expect(step.attempt).toBe(1); + expect(step.startedAt?.getTime()).toBe(BASE_TIME + 2000); + expect(step.completedAt?.getTime()).toBe(BASE_TIME + 5000); + }); + + it('counts one attempt per start across retries', () => { + const events = [ + event('step_created', { correlationId: 'step_a', at: 1 }), + event('step_started', { correlationId: 'step_a', at: 2 }), + event('step_retrying', { correlationId: 'step_a', at: 3 }), + event('step_started', { correlationId: 'step_a', at: 4 }), + event('step_completed', { correlationId: 'step_a', at: 6 }), + ]; + + const [step] = materializeSteps(events); + + expect(step.attempt).toBe(2); + expect(step.status).toBe('completed'); + // The first start is when the step went from queued to running. + expect(step.startedAt?.getTime()).toBe(BASE_TIME + 2000); + }); + + it('keeps the outcome the run acted on when a replay writes another one', () => { + const events = [ + event('step_created', { correlationId: 'step_a', at: 1 }), + event('step_started', { correlationId: 'step_a', at: 2 }), + event('step_failed', { correlationId: 'step_a', at: 3 }), + // Written by a concurrent replay working from a stale prefix. + event('step_completed', { correlationId: 'step_a', at: 9 }), + ]; + + const [step] = materializeSteps(events); + + expect(step.status).toBe('failed'); + expect(step.completedAt?.getTime()).toBe(BASE_TIME + 3000); + expect(step.updatedAt.getTime()).toBe(BASE_TIME + 3000); + }); + + it('still lists the passed-over event on the entity', () => { + const events = [ + event('step_created', { correlationId: 'step_a', at: 1 }), + event('step_created', { correlationId: 'step_a', at: 2 }), + event('step_started', { correlationId: 'step_a', at: 3 }), + ]; + + const [step] = materializeSteps(events); + + expect(step.events).toHaveLength(3); + expect(step.status).toBe('running'); + }); +}); diff --git a/packages/web-shared/src/lib/event-materialization.ts b/packages/web-shared/src/lib/event-materialization.ts index a605e9c624..82324e5de9 100644 --- a/packages/web-shared/src/lib/event-materialization.ts +++ b/packages/web-shared/src/lib/event-materialization.ts @@ -17,6 +17,7 @@ import { isWaitEventType, type StepStatus, } from '@workflow/world'; +import { findDuplicateEventIds } from './duplicate-events'; // --------------------------------------------------------------------------- // Materialized entity types @@ -105,9 +106,16 @@ function getEventTimestamp(event: Event | undefined): Date | undefined { * * Handles partial event lists gracefully: a step may only have a * step_created event with no completion yet. + * + * The derived status and timestamps come from the events the run acted on. A + * repeat of a class the log already records was passed over by the runtime, + * so a second terminal event written by a concurrent replay does not move a + * step off the outcome the first one recorded. Every event stays on the + * entity's `events` list. */ export function materializeSteps(events: Event[]): MaterializedStep[] { const groups = groupByCorrelationId(events, isStepEventType); + const duplicateEventIds = findDuplicateEventIds(events); const steps: MaterializedStep[] = []; for (const [correlationId, stepEvents] of groups) { @@ -121,6 +129,7 @@ export function materializeSteps(events: Event[]): MaterializedStep[] { let updatedAt = getEventTimestamp(created) ?? created.createdAt; for (const e of stepEvents) { + if (duplicateEventIds.has(e.eventId)) continue; switch (e.eventType) { case 'step_started': status = 'running'; diff --git a/packages/web-shared/src/lib/trace-builder.test.ts b/packages/web-shared/src/lib/trace-builder.test.ts new file mode 100644 index 0000000000..bf0954e65a --- /dev/null +++ b/packages/web-shared/src/lib/trace-builder.test.ts @@ -0,0 +1,53 @@ +import type { Event, EventType, WorkflowRun } from '@workflow/world'; +import { describe, expect, it } from 'vitest'; +import { otelTimeToMs } from '../components/workflow-traces/trace-time-utils'; +import { buildTrace } from './trace-builder'; + +const BASE_TIME = Date.parse('2026-01-01T00:00:00.000Z'); + +let nextId = 0; + +function event( + eventType: EventType, + options: { correlationId?: string; at: number } +): Event { + nextId += 1; + return { + eventId: `evt_${nextId}`, + runId: 'run_1', + eventType, + correlationId: options.correlationId, + createdAt: new Date(BASE_TIME + options.at * 1000), + occurredAt: new Date(BASE_TIME + options.at * 1000), + eventData: eventType === 'step_created' ? { stepName: 'doWork' } : {}, + } as unknown as Event; +} + +const run = { + runId: 'run_1', + workflowName: 'demo', + status: 'running', + createdAt: new Date(BASE_TIME), +} as unknown as WorkflowRun; + +describe('buildTrace', () => { + it('ends a step span on the terminal event the run acted on', () => { + const events = [ + event('run_created', { at: 0 }), + event('run_started', { at: 0 }), + event('step_created', { correlationId: 'step_a', at: 1 }), + event('step_started', { correlationId: 'step_a', at: 1 }), + event('step_completed', { correlationId: 'step_a', at: 4 }), + // A concurrent replay commits the same outcome much later. Measuring the + // span against it would report a 20s step that ran for 3s. + event('step_completed', { correlationId: 'step_a', at: 20 }), + ]; + + const trace = buildTrace(run, events, new Date(BASE_TIME + 30_000)); + const stepSpan = trace.spans.find((span) => span.resource === 'step'); + + expect(stepSpan).toBeDefined(); + expect(otelTimeToMs(stepSpan?.endTime ?? [0, 0])).toBe(BASE_TIME + 4000); + expect(trace.knownDurationMs).toBe(4000); + }); +}); diff --git a/packages/web-shared/src/lib/trace-builder.ts b/packages/web-shared/src/lib/trace-builder.ts index 6b01e5b042..053a49f799 100644 --- a/packages/web-shared/src/lib/trace-builder.ts +++ b/packages/web-shared/src/lib/trace-builder.ts @@ -22,6 +22,7 @@ import { waitToSpan, } from '../components/workflow-traces/trace-span-construction'; import { otelTimeToMs } from '../components/workflow-traces/trace-time-utils'; +import { findDuplicateEventIds } from './duplicate-events'; import type { Span } from './trace-types'; /** @@ -195,8 +196,18 @@ export function buildTrace( events: Event[], now: Date ): TraceWithMeta { - const groupedEvents = groupEventsByCorrelation(events); - const latestKnownTime = computeLatestKnownTime(events, run); + // Span geometry comes from what the run acted on. A repeat of a class the + // log already records was passed over by the runtime, and letting one + // through here would stretch a span to whenever a concurrent replay + // committed it. The event lists still show them, marked as ignored. + const duplicateEventIds = findDuplicateEventIds(events); + const actedOnEvents = + duplicateEventIds.size === 0 + ? events + : events.filter((event) => !duplicateEventIds.has(event.eventId)); + + const groupedEvents = groupEventsByCorrelation(actedOnEvents); + const latestKnownTime = computeLatestKnownTime(actedOnEvents, run); const { runSpan, spans } = buildSpans( run, groupedEvents, From 9d71ab91f1d05e559f652402343c4dccc8a3f311 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 12:50:11 -0700 Subject: [PATCH 3/6] Skip events with no id when detecting duplicates An event the caller cannot identify cannot be marked: callers match on the id, so reporting a missing one tarred every other id-less event with it. --- packages/web-shared/src/lib/duplicate-events.test.ts | 12 ++++++++++++ packages/web-shared/src/lib/duplicate-events.ts | 4 ++++ 2 files changed, 16 insertions(+) diff --git a/packages/web-shared/src/lib/duplicate-events.test.ts b/packages/web-shared/src/lib/duplicate-events.test.ts index 484875beee..f59f134357 100644 --- a/packages/web-shared/src/lib/duplicate-events.test.ts +++ b/packages/web-shared/src/lib/duplicate-events.test.ts @@ -123,6 +123,18 @@ describe('findDuplicateEventIds', () => { expect(descending).toEqual(ascending); }); + it('skips events with no id, which callers cannot match on', () => { + const anonymous = (at: number) => + ({ + ...event('step_created', { correlationId: 'step_a', at }), + eventId: undefined, + }) as unknown as Event; + + expect(findDuplicateEventIds([anonymous(1), anonymous(2)])).toEqual( + new Set() + ); + }); + it('breaks timestamp ties on the order the log lists them in', () => { const first = event('wait_completed', { at: 5, correlationId: 'wait_a' }); const second = event('wait_completed', { at: 5, correlationId: 'wait_a' }); diff --git a/packages/web-shared/src/lib/duplicate-events.ts b/packages/web-shared/src/lib/duplicate-events.ts index d9e647d390..c4e73d1906 100644 --- a/packages/web-shared/src/lib/duplicate-events.ts +++ b/packages/web-shared/src/lib/duplicate-events.ts @@ -77,6 +77,10 @@ export function findDuplicateEventIds(events: readonly Event[]): Set { const seen = new Set(); for (const { event } of ordered) { + // An event the caller cannot identify cannot be marked: callers match on + // the id, so reporting a missing one would tar every other such event with + // it. + if (!event.eventId) continue; const key = entityKey(event); if (key === undefined) continue; if (seen.has(key)) { From 7ec31ac31f3b1e883523171d795a60f7f89de505 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Wed, 12 Aug 2026 14:22:40 -0700 Subject: [PATCH 4/6] Classify duplicate events from consumer lifetime, not event type A repeat only reaches nobody once its entity's consumer is gone, so an event is classified as read past only when a terminal event for the same entity sits earlier in the log. A retried step's repeated starts and a second creation of an open step are claimed by a live consumer and stay untouched. Order the fold on log position rather than on timestamps: a writer stamps createdAt on entry but takes its slot at publish time, and occurredAt is measured on the client. Classify nothing on a subset of the log. Which occurrence of a class came first is a property of the whole log, so callers now say whether they hold it, and derived state (span geometry, step status, durations) keeps every event when they do not. The fixtures both sides agree on live in @workflow/world, driven through EventsConsumer on the runtime side and findDuplicateEventIds on the UI side. --- .changeset/silly-pears-jam.md | 2 +- .../core/src/duplicate-event-fixtures.test.ts | 121 +++++++++++ .../src/components/event-list-view.tsx | 33 +-- .../sidebar/entity-detail-panel.tsx | 6 + .../src/components/sidebar/events-list.tsx | 15 +- .../sidebar/sidebar-data-context.tsx | 6 + .../src/components/trace-viewer.tsx | 18 +- .../trace-viewer/components/detail-panel.tsx | 1 + .../components/ui/duplicate-event-tooltip.tsx | 6 +- .../src/lib/duplicate-events.test.ts | 203 ++++++++++++++---- .../web-shared/src/lib/duplicate-events.ts | 156 ++++++++------ .../src/lib/event-materialization.test.ts | 34 ++- .../src/lib/event-materialization.ts | 27 ++- .../web-shared/src/lib/trace-builder.test.ts | 23 +- packages/web-shared/src/lib/trace-builder.ts | 26 ++- .../test-support/duplicate-event-fixtures.ts | 148 +++++++++++++ 16 files changed, 680 insertions(+), 145 deletions(-) create mode 100644 packages/core/src/duplicate-event-fixtures.test.ts create mode 100644 packages/world/src/test-support/duplicate-event-fixtures.ts diff --git a/.changeset/silly-pears-jam.md b/.changeset/silly-pears-jam.md index 96d1369efa..055dfa8ec0 100644 --- a/.changeset/silly-pears-jam.md +++ b/.changeset/silly-pears-jam.md @@ -2,4 +2,4 @@ '@workflow/web-shared': patch --- -Mark ignored duplicate events in the observability UI and exclude them from derived step status, durations, and trace spans +Mark ignored duplicate events in the observability UI and, when the whole event log is loaded, exclude them from derived step status, durations, and trace spans diff --git a/packages/core/src/duplicate-event-fixtures.test.ts b/packages/core/src/duplicate-event-fixtures.test.ts new file mode 100644 index 0000000000..ae2dca7b33 --- /dev/null +++ b/packages/core/src/duplicate-event-fixtures.test.ts @@ -0,0 +1,121 @@ +import { type Event, entityEventClass } from '@workflow/world'; +import { DUPLICATE_EVENT_FIXTURES } from '@workflow/world/test-support/duplicate-event-fixtures.js'; +import { describe, expect, it, vi } from 'vitest'; +import { EventConsumerResult, EventsConsumer } from './events-consumer.js'; + +/** + * The runtime's half of {@link DUPLICATE_EVENT_FIXTURES}. The observability + * UI's half runs the same fixtures through its own classifier, so a fixture + * whose expectation moves fails on both sides. + * + * What this drives is the walk in `EventsConsumer`: given consumers that claim + * their entity's events for as long as the entity is open, which events does + * it step over. The claim that the consumers behave that way is what + * `duplicate-events.test.ts` checks, against the real step and sleep + * primitives. + */ + +/** No deliveries are modeled here, so the delivery gate is always open. */ +const OPEN_GATE = { + getPromiseQueue: () => Promise.resolve(), + isDeliveryIdle: () => true, +}; + +/** Classes whose event deregisters its entity's consumer. */ +const TERMINAL_CLASSES = new Set([ + 'step_terminal', + 'wait_completed', + 'hook_disposed', + 'run_terminal', +]); + +/** + * One callback standing in for every consumer a replay registers. + * + * `step.ts` keeps a step's consumer alive from `step_created` until the step's + * outcome, claiming each attempt's `step_started` and `step_retrying` on the + * way; `sleep.ts` does the same for a wait. Both delete the queue item on the + * entity's terminal event, after which nothing claims that correlation id. + * `workflow.ts` declines a second `run_started` outright. + */ +function replayConsumers(): (event: Event | null) => EventConsumerResult { + const claimed = new Set(); + const closed = new Set(); + + return (event) => { + if (event === null) return EventConsumerResult.NotConsumed; + + const eventClass = entityEventClass(event.eventType); + // Belongs to no class: a delivery, or an event that precedes every replay. + // Their consumers subscribe lazily and take every copy. + if (eventClass === undefined) return EventConsumerResult.Consumed; + + const entity = event.correlationId ?? ''; + if (closed.has(entity)) return EventConsumerResult.NotConsumed; + + const classKey = `${eventClass}:${entity}`; + if (eventClass === 'run_started' && claimed.has(classKey)) { + return EventConsumerResult.NotConsumed; + } + + claimed.add(classKey); + if (TERMINAL_CLASSES.has(eventClass)) closed.add(entity); + return EventConsumerResult.Consumed; + }; +} + +function buildLog(fixture: (typeof DUPLICATE_EVENT_FIXTURES)[number]): Event[] { + return fixture.events.map( + (spec, index) => + ({ + eventId: `evnt_${String(index).padStart(26, '0')}`, + runId: 'wrun_test', + eventType: spec.eventType, + correlationId: spec.entity, + eventData: {}, + createdAt: new Date(), + }) as unknown as Event + ); +} + +/** + * Resolve once the walk has decided every event: it reached the end of the + * log, or it stopped on one nothing can claim, which is the divergence the + * skip exists to tell apart from a repeat. + */ +async function settle( + consumer: EventsConsumer, + length: number, + onUnconsumedEvent: { mock: { calls: unknown[] } } +) { + const deadline = Date.now() + 5000; + while ( + consumer.eventIndex < length && + onUnconsumedEvent.mock.calls.length === 0 && + Date.now() < deadline + ) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} + +describe('shared duplicate-event fixtures', () => { + for (const fixture of DUPLICATE_EVENT_FIXTURES) { + it(`steps over the right events: ${fixture.name}`, async () => { + const events = buildLog(fixture); + const onDuplicateEvent = vi.fn(); + const onUnconsumedEvent = vi.fn(); + const consumer = new EventsConsumer(events, { + ...OPEN_GATE, + onUnconsumedEvent, + onDuplicateEvent, + }); + + consumer.subscribe(replayConsumers()); + await settle(consumer, events.length, onUnconsumedEvent); + + expect(onDuplicateEvent.mock.calls.map(([event]) => event)).toEqual( + fixture.ignoredIndices.map((index) => events[index]) + ); + }); + } +}); diff --git a/packages/web-shared/src/components/event-list-view.tsx b/packages/web-shared/src/components/event-list-view.tsx index 87685620e8..aefa6b63da 100644 --- a/packages/web-shared/src/components/event-list-view.tsx +++ b/packages/web-shared/src/components/event-list-view.tsx @@ -193,13 +193,16 @@ export interface DurationInfo { * created ↔ started (queued) and started ↔ completed/failed/cancelled (ran). * Also computes run-level durations under the key '__run__'. * - * Events the runtime passed over as repeats are excluded: a second + * Events every replay reads past as repeats are excluded: a second * `step_completed` written by a concurrent replay would otherwise stretch the - * step's measured runtime to whenever that replay happened to commit. + * step's measured runtime to whenever that replay happened to commit. The + * caller supplies them, because whether an event is a repeat is a property of + * the whole log and this function may be handed a page of it. */ -export function buildDurationMap(events: Event[]): Map { - const duplicateEventIds = findDuplicateEventIds(events); - +export function buildDurationMap( + events: Event[], + duplicateEventIds: ReadonlySet = new Set() +): Map { // Process events in chronological order so the result doesn't depend on // the caller's sort direction. Retried steps emit multiple `step_started` // events for the same correlationId; the queued duration must be measured @@ -1327,15 +1330,21 @@ function EventListViewInner({ ); }, [events, effectiveSortOrder, isExactSearchActive, searchResults]); - // Events the runtime passed over as repeats. Computed from the source list + // Events every replay reads past as repeats. Computed from the source list // rather than `sortedEvents` because which occurrence counted is a property // of the log, not of the direction the table happens to be sorted in. + // + // A page short of the whole log, or an exact-ID search that returns one + // event, cannot answer that question: the event a repeat lost to may be + // outside the window, and reading it the other way round would mark the + // event the run acted on and drop it from the durations. Both cases classify + // nothing. const duplicateEventIds = useMemo( () => - findDuplicateEventIds( - isExactSearchActive ? (searchResults ?? []) : (events ?? []) - ), - [events, isExactSearchActive, searchResults] + findDuplicateEventIds(events ?? [], { + isCompleteHistory: !hasMoreEvents && !isExactSearchActive, + }), + [events, hasMoreEvents, isExactSearchActive] ); // Detect encrypted fields across all loaded events (inline eventData). @@ -1370,8 +1379,8 @@ function EventListViewInner({ ); const durationMap = useMemo( - () => buildDurationMap(sortedEvents), - [sortedEvents] + () => buildDurationMap(sortedEvents, duplicateEventIds), + [sortedEvents, duplicateEventIds] ); const [selectedGroupKey, setSelectedGroupKey] = useState( diff --git a/packages/web-shared/src/components/sidebar/entity-detail-panel.tsx b/packages/web-shared/src/components/sidebar/entity-detail-panel.tsx index 902fa01cf6..771b3dfd0e 100644 --- a/packages/web-shared/src/components/sidebar/entity-detail-panel.tsx +++ b/packages/web-shared/src/components/sidebar/entity-detail-panel.tsx @@ -47,6 +47,11 @@ export interface SelectedSpanInfo { spanId?: string; /** Raw correlated events from the store (NOT from the trace worker pipeline) */ rawEvents?: Event[]; + /** + * Events every replay reads past as repeats, computed from the whole log. + * `rawEvents` is one entity's slice, which cannot answer that on its own. + */ + duplicateEventIds?: ReadonlySet; } /** @@ -402,6 +407,7 @@ export function EntityDetailPanel({ showSeparateEventOccurrenceTimestamps={ showSeparateEventOccurrenceTimestamps } + duplicateEventIds={selectedSpan?.duplicateEventIds} /> )} diff --git a/packages/web-shared/src/components/sidebar/events-list.tsx b/packages/web-shared/src/components/sidebar/events-list.tsx index c1ae4b0030..7fd65bc10d 100644 --- a/packages/web-shared/src/components/sidebar/events-list.tsx +++ b/packages/web-shared/src/components/sidebar/events-list.tsx @@ -2,7 +2,6 @@ import { type Event, getEventDataRefFields } from '@workflow/world'; import { useCallback, useLayoutEffect, useMemo, useRef, useState } from 'react'; -import { findDuplicateEventIds } from '../../lib/duplicate-events'; import { hasEncryptedFields, isExpiredMarker } from '../../lib/hydration'; import { Collapsible, @@ -305,6 +304,7 @@ export function EventsList({ onRunClick, encryptionKey, showSeparateEventOccurrenceTimestamps = false, + duplicateEventIds, }: { events: Event[]; isLoading?: boolean; @@ -316,6 +316,12 @@ export function EventsList({ encryptionKey?: Uint8Array; /** Show occurredAt separately instead of folding it into the Created timestamp. */ showSeparateEventOccurrenceTimestamps?: boolean; + /** + * Events every replay reads past as repeats, from the caller that holds the + * whole log. `events` here is one entity's slice of it, which cannot answer + * the question on its own. + */ + duplicateEventIds?: ReadonlySet; }) { // Sort by the timestamp shown as Created by default. const sortedEvents = useMemo( @@ -328,11 +334,6 @@ export function EventsList({ [events] ); - const duplicateEventIds = useMemo( - () => findDuplicateEventIds(events), - [events] - ); - const hasEvents = sortedEvents.length > 0 && !error; if (!hasEvents && !isLoading) { @@ -368,7 +369,7 @@ export function EventsList({ showSeparateEventOccurrenceTimestamps={ showSeparateEventOccurrenceTimestamps } - isDuplicate={duplicateEventIds.has(event.eventId)} + isDuplicate={duplicateEventIds?.has(event.eventId)} /> ))} diff --git a/packages/web-shared/src/components/sidebar/sidebar-data-context.tsx b/packages/web-shared/src/components/sidebar/sidebar-data-context.tsx index dd26e768d8..eac0a2d65f 100644 --- a/packages/web-shared/src/components/sidebar/sidebar-data-context.tsx +++ b/packages/web-shared/src/components/sidebar/sidebar-data-context.tsx @@ -7,6 +7,12 @@ import type { FetchSpanDetail } from './use-selected-span-detail'; export interface SidebarDataContextValue { run: WorkflowRun; events: Event[]; + /** + * Events every replay reads past as repeats, computed once from the whole + * of `events` by whoever knows the list is complete. Absent when nobody + * could vouch for that, in which case nothing is marked. + */ + duplicateEventIds?: ReadonlySet; fetchSpanDetail: FetchSpanDetail; onStreamClick?: (streamId: string) => void; onRunClick?: (runId: string) => void; diff --git a/packages/web-shared/src/components/trace-viewer.tsx b/packages/web-shared/src/components/trace-viewer.tsx index 566d5ec8c6..b64da197ec 100644 --- a/packages/web-shared/src/components/trace-viewer.tsx +++ b/packages/web-shared/src/components/trace-viewer.tsx @@ -30,16 +30,28 @@ const TraceViewer = ({ if (!run?.runId) { return undefined; } - return buildTrace(run, events, new Date()); + // `hasMore` is the only place that knows whether more of the log is still + // to be fetched, and a repeat can only be told apart from the event it + // repeats with the whole log in hand. + return buildTrace(run, events, new Date(), { + isCompleteHistory: !hasMore, + }); // eslint-disable-next-line react-hooks/exhaustive-deps -- `new Date()` is intentionally not a dep - }, [run, events]); + }, [run, events, hasMore]); + + // The sidebar shows one entity's slice of the log, so it takes the trace's + // answer rather than recomputing one from the slice. + const sidebarValue = useMemo( + () => ({ ...sidebarData, duplicateEventIds: trace?.duplicateEventIds }), + [sidebarData, trace] + ); if (!trace || (loading && events.length === 0)) { return ; } return ( - +
{ event('run_completed'), ]; - expect(findDuplicateEventIds(events)).toEqual(new Set()); + expect(findDuplicateEventIds(events, COMPLETE)).toEqual(new Set()); }); - it('flags every repeat after the first of a once-per-entity class', () => { - const first = event('step_created', { correlationId: 'step_a' }); - const second = event('step_created', { correlationId: 'step_a' }); - const third = event('step_created', { correlationId: 'step_a' }); + it('flags every repeat a finished entity collects', () => { + const created = event('step_created', { correlationId: 'step_a' }); + const started = event('step_started', { correlationId: 'step_a' }); + const completed = event('step_completed', { correlationId: 'step_a' }); + const createdAgain = event('step_created', { correlationId: 'step_a' }); + const startedAgain = event('step_started', { correlationId: 'step_a' }); - expect(findDuplicateEventIds([first, second, third])).toEqual( - new Set([second.eventId, third.eventId]) - ); + expect( + findDuplicateEventIds( + [created, started, completed, createdAgain, startedAgain], + COMPLETE + ) + ).toEqual(new Set([createdAgain.eventId, startedAgain.eventId])); + }); + + it('leaves a class the log has not recorded for the entity yet', () => { + // The step finished without a step_started in the log, so this one repeats + // nothing. The runtime reports that as divergence rather than passing it + // over, and the UI must not present it as a settled repeat. + const events = [ + event('step_created', { correlationId: 'step_a' }), + event('step_completed', { correlationId: 'step_a' }), + event('step_started', { correlationId: 'step_a' }), + ]; + + expect(findDuplicateEventIds(events, COMPLETE)).toEqual(new Set()); }); it('treats completed and failed as one terminal class', () => { @@ -58,7 +85,7 @@ describe('findDuplicateEventIds', () => { const failed = event('step_failed', { correlationId: 'step_a' }); const completed = event('step_completed', { correlationId: 'step_a' }); - expect(findDuplicateEventIds([failed, completed])).toEqual( + expect(findDuplicateEventIds([failed, completed], COMPLETE)).toEqual( new Set([completed.eventId]) ); }); @@ -71,12 +98,13 @@ describe('findDuplicateEventIds', () => { event('step_completed', { correlationId: 'step_b' }), ]; - expect(findDuplicateEventIds(events)).toEqual(new Set()); + expect(findDuplicateEventIds(events, COMPLETE)).toEqual(new Set()); }); it('does not flag the repeated events of a retried step', () => { // Each attempt legitimately records its own start, and each retryable - // failure its own step_retrying. Both are consumed, not passed over. + // failure its own step_retrying. The step's consumer is registered for the + // whole sequence and takes all of them. const events = [ event('step_created', { correlationId: 'step_a' }), event('step_started', { correlationId: 'step_a' }), @@ -87,7 +115,19 @@ describe('findDuplicateEventIds', () => { event('step_completed', { correlationId: 'step_a' }), ]; - expect(findDuplicateEventIds(events)).toEqual(new Set()); + expect(findDuplicateEventIds(events, COMPLETE)).toEqual(new Set()); + }); + + it('does not flag a second step_created while the step is still open', () => { + // The step's consumer is registered and absorbs it, so the run does not + // read past this event. + const events = [ + event('step_created', { correlationId: 'step_a' }), + event('step_created', { correlationId: 'step_a' }), + event('step_started', { correlationId: 'step_a' }), + ]; + + expect(findDuplicateEventIds(events, COMPLETE)).toEqual(new Set()); }); it('does not flag repeated hook deliveries', () => { @@ -98,7 +138,7 @@ describe('findDuplicateEventIds', () => { event('hook_disposed', { correlationId: 'hook_a' }), ]; - expect(findDuplicateEventIds(events)).toEqual(new Set()); + expect(findDuplicateEventIds(events, COMPLETE)).toEqual(new Set()); }); it('flags run-level repeats, which carry no correlation id', () => { @@ -108,39 +148,122 @@ describe('findDuplicateEventIds', () => { const cancelled = event('run_cancelled'); expect( - findDuplicateEventIds([started, startedAgain, completed, cancelled]) + findDuplicateEventIds( + [started, startedAgain, completed, cancelled], + COMPLETE + ) ).toEqual(new Set([startedAgain.eventId, cancelled.eventId])); }); - it('picks the earliest occurrence whichever way the caller sorted', () => { - const first = event('wait_created', { at: 1, correlationId: 'wait_a' }); - const second = event('wait_created', { at: 2, correlationId: 'wait_a' }); + it('folds in log order, not in createdAt order', () => { + const created = event('wait_created', { correlationId: 'wait_a', at: 1 }); + const completed = event('wait_completed', { + correlationId: 'wait_a', + at: 3, + }); + // The repeat entered before the completion it lost to and only took its + // log position afterwards, so its createdAt is the earliest of the three. + const createdAgain = event('wait_created', { + correlationId: 'wait_a', + at: 0, + }); + + expect( + findDuplicateEventIds([created, completed, createdAgain], COMPLETE) + ).toEqual(new Set([createdAgain.eventId])); + }); + + it('folds in log order, not in occurredAt order', () => { + const created = event('wait_created', { correlationId: 'wait_a' }); + const completed = event('wait_completed', { correlationId: 'wait_a' }); + // occurredAt is measured on the writer's clock, which can run behind. + const createdAgain = event('wait_created', { + correlationId: 'wait_a', + occurredAt: -60, + }); + + expect( + findDuplicateEventIds([created, completed, createdAgain], COMPLETE) + ).toEqual(new Set([createdAgain.eventId])); + }); + + it('gives the same answer whichever way the caller sorted', () => { + const events = [ + event('wait_created', { correlationId: 'wait_a' }), + event('wait_completed', { correlationId: 'wait_a' }), + event('wait_created', { correlationId: 'wait_a' }), + ]; - const ascending = findDuplicateEventIds([first, second]); - const descending = findDuplicateEventIds([second, first]); + const ascending = findDuplicateEventIds(events, COMPLETE); + const descending = findDuplicateEventIds([...events].reverse(), COMPLETE); - expect(ascending).toEqual(new Set([second.eventId])); + expect(ascending).toEqual(new Set([events[2].eventId])); expect(descending).toEqual(ascending); }); + it('gives the same answer on tied timestamps whichever way the caller sorted', () => { + // Two replays that stamped the same millisecond. Only the log position + // separates them, so the answer must not depend on the caller's order. + const events = [ + event('wait_created', { correlationId: 'wait_a', at: 5 }), + event('wait_completed', { correlationId: 'wait_a', at: 5 }), + event('wait_created', { correlationId: 'wait_a', at: 5 }), + ]; + + const ascending = findDuplicateEventIds(events, COMPLETE); + const descending = findDuplicateEventIds([...events].reverse(), COMPLETE); + + expect(ascending).toEqual(new Set([events[2].eventId])); + expect(descending).toEqual(ascending); + }); + + it('classifies nothing when the caller holds part of the log', () => { + // A newest-first page can open on the repeat and omit the event it + // repeats, which would invert the answer. + const events = [ + event('wait_completed', { correlationId: 'wait_a' }), + event('wait_created', { correlationId: 'wait_a' }), + ]; + + expect(findDuplicateEventIds(events, { isCompleteHistory: false })).toEqual( + new Set() + ); + }); + it('skips events with no id, which callers cannot match on', () => { - const anonymous = (at: number) => + const anonymous = (eventType: EventType) => ({ - ...event('step_created', { correlationId: 'step_a', at }), + ...event(eventType, { correlationId: 'wait_a' }), eventId: undefined, }) as unknown as Event; - expect(findDuplicateEventIds([anonymous(1), anonymous(2)])).toEqual( - new Set() - ); + expect( + findDuplicateEventIds( + [ + anonymous('wait_created'), + anonymous('wait_completed'), + anonymous('wait_created'), + ], + COMPLETE + ) + ).toEqual(new Set()); }); +}); - it('breaks timestamp ties on the order the log lists them in', () => { - const first = event('wait_completed', { at: 5, correlationId: 'wait_a' }); - const second = event('wait_completed', { at: 5, correlationId: 'wait_a' }); +/** + * The other half of these runs against `EventsConsumer` in `@workflow/core`, + * so a fixture whose expectation moves fails on both sides. + */ +describe('shared duplicate-event fixtures', () => { + for (const fixture of DUPLICATE_EVENT_FIXTURES) { + it(`classifies the right events: ${fixture.name}`, () => { + const events = fixture.events.map((spec, index) => + event(spec.eventType, { correlationId: spec.entity, slot: index + 1 }) + ); - expect(findDuplicateEventIds([first, second])).toEqual( - new Set([second.eventId]) - ); - }); + expect(findDuplicateEventIds(events, COMPLETE)).toEqual( + new Set(fixture.ignoredIndices.map((index) => events[index].eventId)) + ); + }); + } }); diff --git a/packages/web-shared/src/lib/duplicate-events.ts b/packages/web-shared/src/lib/duplicate-events.ts index c4e73d1906..37ffbb0793 100644 --- a/packages/web-shared/src/lib/duplicate-events.ts +++ b/packages/web-shared/src/lib/duplicate-events.ts @@ -5,89 +5,119 @@ import { } from '@workflow/world'; /** - * Copy shown on an event the runtime passed over as a repeat. - */ -export const DUPLICATE_EVENT_MESSAGE = - 'Multiple processes resuming may result in duplicate events. This event was ignored and had no effect.'; - -/** - * Event classes a run records at most once per entity. A second event of one - * of these classes for the same entity is a repeat: the first one is what the - * replay acted on, and the runtime passes over the rest. + * Identifies events a replay reads past. * - * `step_started` and `step_retrying` are the two classes deliberately left - * out. A retried step records one of each per attempt, and the runtime - * consumes them, so a repeat there is an extra attempt rather than an ignored - * event. + * Concurrent replays of one run write to a shared log, 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. Every replay + * reads the first event of that class at the same position, so a later one + * cannot change what the workflow observes. + * + * The classification mirrors `entityEventClass` in `@workflow/world`, which is + * what the runtime keys its own duplicate detection on. What it cannot mirror + * is consumer state: the runtime passes over an event only after every + * registered callback has declined it, and a callback registered for a + * still-open entity legitimately claims a repeat (each retry of a step writes + * another `step_started`, and a live step consumer absorbs a second + * `step_created`). So a repeat counts here only once a terminal event for the + * same entity sits earlier in the log, which is the point past which no + * consumer remains. */ -const ONCE_PER_ENTITY_CLASSES: ReadonlySet = new Set([ - 'step_created', + +/** Classes whose event closes its entity: no consumer is left for it after. */ +const TERMINAL_EVENT_CLASSES: ReadonlySet = new Set([ 'step_terminal', - 'wait_created', 'wait_completed', - 'hook_created', 'hook_disposed', - 'run_started', 'run_terminal', ]); -function entityKey(event: Event): string | undefined { - const eventClass = entityEventClass(event.eventType); - if (eventClass === undefined || !ONCE_PER_ENTITY_CLASSES.has(eventClass)) { - return undefined; - } - // Run events carry no correlation id, so they key on the class alone. - return `${eventClass}:${event.correlationId ?? ''}`; -} +/** Classes with no entity to close first: the log records one per run. */ +const SINGLETON_EVENT_CLASSES: ReadonlySet = new Set([ + 'run_started', +]); + +/** Entity key for events that carry no correlation ID (the run itself). */ +const RUN_ENTITY_KEY = ''; + +/** + * Shown against an event this module reports. Deliberately says what the log + * shows rather than what the runtime did with it: tolerating these repeats is + * recent, and on a run recorded before it a repeat no consumer claimed failed + * the replay instead of being passed over. + */ +export const DUPLICATE_EVENT_MESSAGE = + 'Written by a concurrent replay after an event of the same kind was already recorded and acted on. The run follows the earlier one.'; -function effectiveTime(event: Event): number { - const occurredAt = event.occurredAt; - if (occurredAt != null) { - const parsed = - occurredAt instanceof Date ? occurredAt : new Date(String(occurredAt)); - if (!Number.isNaN(parsed.getTime())) return parsed.getTime(); +/** + * Log order. + * + * Event IDs are fixed-width and monotonic within a run under both the ULID and + * the slot scheme, so comparing them orders the log even where timestamps do + * not: a writer stamps `createdAt` on entry but takes its log position at + * publish time, and `occurredAt` is measured on the client. Length is compared + * first so a shorter ID never sorts after a longer one on a fixture or a log + * that mixes widths. + */ +function compareLogPosition(a: Event, b: Event): number { + if (a.eventId.length !== b.eventId.length) { + return a.eventId.length - b.eventId.length; } - return new Date(event.createdAt).getTime(); + return a.eventId < b.eventId ? -1 : a.eventId > b.eventId ? 1 : 0; } /** - * Event ids that repeat a class the log already records for the same entity. - * - * Concurrent replays of one run share an event log, so a replay working from - * a stale prefix can commit a write the log already has. The runtime passes - * over those (see `EventsConsumer` in `@workflow/core`); this reports them so - * the UI can say so rather than showing them as ordinary progress. + * The IDs of the events in `events` that repeat a class the log already + * records for the same entity, after that entity finished. * - * Order is decided here, not by the caller: the earliest event of a class is - * the one that counted, whichever direction the view happens to sort in. The - * answer is exact only over a complete event list. A partial list (pagination, - * an exact-id search) can miss a repeat whose first occurrence is absent, - * which under-reports rather than mislabels. + * `isCompleteHistory` must be false whenever the caller holds a subset of the + * run's log: one page of a paginated list, or the result of a search. Which + * occurrence of a class came first is a property of the whole log, so on a + * subset the earlier event may simply be missing, and the fold would report + * the surviving one. Nothing is classified in that case. */ -export function findDuplicateEventIds(events: readonly Event[]): Set { +export function findDuplicateEventIds( + events: readonly Event[], + { isCompleteHistory }: { isCompleteHistory: boolean } +): Set { const duplicates = new Set(); - if (events.length < 2) return duplicates; + if (!isCompleteHistory || events.length < 2) return duplicates; + const seenClasses = new Set(); + const closedEntities = new Set(); + + // Dropped before the sort, not during the fold: an event with no ID has no + // log position to order on, and the caller could not match it either. const ordered = events - .map((event, index) => ({ event, index })) - .sort( - (a, b) => - effectiveTime(a.event) - effectiveTime(b.event) || a.index - b.index - ); - - const seen = new Set(); - for (const { event } of ordered) { - // An event the caller cannot identify cannot be marked: callers match on - // the id, so reporting a missing one would tar every other such event with - // it. - if (!event.eventId) continue; - const key = entityKey(event); - if (key === undefined) continue; - if (seen.has(key)) { - duplicates.add(event.eventId); + .filter((event) => Boolean(event.eventId)) + .sort(compareLogPosition); + + for (const event of ordered) { + const eventClass = entityEventClass(event.eventType); + if (eventClass === undefined) continue; + + const entity = event.correlationId ?? RUN_ENTITY_KEY; + const classKey = `${eventClass}:${entity}`; + const repeatsClass = seenClasses.has(classKey); + const entityWasClosed = closedEntities.has(entity); + + if (TERMINAL_EVENT_CLASSES.has(eventClass)) { + closedEntities.add(entity); + } + + if (!repeatsClass) { + seenClasses.add(classKey); continue; } - seen.add(key); + + // The entity is still open, so a consumer is registered for it and takes + // this event: another attempt, not a repeat read past. + if (!entityWasClosed && !SINGLETON_EVENT_CLASSES.has(eventClass)) { + continue; + } + + duplicates.add(event.eventId); } + return duplicates; } diff --git a/packages/web-shared/src/lib/event-materialization.test.ts b/packages/web-shared/src/lib/event-materialization.test.ts index 4ed98a17f9..d26ae41d0c 100644 --- a/packages/web-shared/src/lib/event-materialization.test.ts +++ b/packages/web-shared/src/lib/event-materialization.test.ts @@ -65,7 +65,7 @@ describe('materializeSteps', () => { event('step_completed', { correlationId: 'step_a', at: 9 }), ]; - const [step] = materializeSteps(events); + const [step] = materializeSteps(events, { isCompleteHistory: true }); expect(step.status).toBe('failed'); expect(step.completedAt?.getTime()).toBe(BASE_TIME + 3000); @@ -75,13 +75,43 @@ describe('materializeSteps', () => { it('still lists the passed-over event on the entity', () => { const events = [ event('step_created', { correlationId: 'step_a', at: 1 }), + event('step_started', { correlationId: 'step_a', at: 2 }), + event('step_failed', { correlationId: 'step_a', at: 3 }), + event('step_completed', { correlationId: 'step_a', at: 9 }), + ]; + + const [step] = materializeSteps(events, { isCompleteHistory: true }); + + expect(step.events).toHaveLength(4); + }); + + it('takes the last outcome when the log may be incomplete', () => { + const events = [ + event('step_created', { correlationId: 'step_a', at: 1 }), + event('step_started', { correlationId: 'step_a', at: 2 }), + event('step_failed', { correlationId: 'step_a', at: 3 }), + event('step_completed', { correlationId: 'step_a', at: 9 }), + ]; + + // On a page of the log there is no telling which failure the run acted on, + // so nothing is passed over and the fold reports what it was given. + const [step] = materializeSteps(events); + + expect(step.status).toBe('completed'); + }); + + it('counts a repeated creation as one attempt while the step is open', () => { + const events = [ + event('step_created', { correlationId: 'step_a', at: 1 }), + // A live step consumer claims this, so it is an attempt, not a repeat. event('step_created', { correlationId: 'step_a', at: 2 }), event('step_started', { correlationId: 'step_a', at: 3 }), ]; - const [step] = materializeSteps(events); + const [step] = materializeSteps(events, { isCompleteHistory: true }); expect(step.events).toHaveLength(3); expect(step.status).toBe('running'); + expect(step.attempt).toBe(1); }); }); diff --git a/packages/web-shared/src/lib/event-materialization.ts b/packages/web-shared/src/lib/event-materialization.ts index 82324e5de9..d2f7a56c8b 100644 --- a/packages/web-shared/src/lib/event-materialization.ts +++ b/packages/web-shared/src/lib/event-materialization.ts @@ -108,14 +108,22 @@ function getEventTimestamp(event: Event | undefined): Date | undefined { * step_created event with no completion yet. * * The derived status and timestamps come from the events the run acted on. A - * repeat of a class the log already records was passed over by the runtime, - * so a second terminal event written by a concurrent replay does not move a - * step off the outcome the first one recorded. Every event stays on the - * entity's `events` list. + * repeat of a class the log already records is read past by every replay, so + * a second terminal event written by a concurrent replay does not move a step + * off the outcome the first one recorded. Every event stays on the entity's + * `events` list. + * + * That reduction needs the whole log to be sound, so it only runs when + * `isCompleteHistory` says `events` is it. See {@link findDuplicateEventIds}. */ -export function materializeSteps(events: Event[]): MaterializedStep[] { +export function materializeSteps( + events: Event[], + { isCompleteHistory = false }: { isCompleteHistory?: boolean } = {} +): MaterializedStep[] { const groups = groupByCorrelationId(events, isStepEventType); - const duplicateEventIds = findDuplicateEventIds(events); + const duplicateEventIds = findDuplicateEventIds(events, { + isCompleteHistory, + }); const steps: MaterializedStep[] = []; for (const [correlationId, stepEvents] of groups) { @@ -257,9 +265,12 @@ export function materializeWaits(events: Event[]): MaterializedWait[] { * Convenience function that materializes all entity types from a flat * event list. */ -export function materializeAll(events: Event[]): MaterializedEntities { +export function materializeAll( + events: Event[], + options: { isCompleteHistory?: boolean } = {} +): MaterializedEntities { return { - steps: materializeSteps(events), + steps: materializeSteps(events, options), hooks: materializeHooks(events), waits: materializeWaits(events), }; diff --git a/packages/web-shared/src/lib/trace-builder.test.ts b/packages/web-shared/src/lib/trace-builder.test.ts index bf0954e65a..70ad41339a 100644 --- a/packages/web-shared/src/lib/trace-builder.test.ts +++ b/packages/web-shared/src/lib/trace-builder.test.ts @@ -43,11 +43,32 @@ describe('buildTrace', () => { event('step_completed', { correlationId: 'step_a', at: 20 }), ]; - const trace = buildTrace(run, events, new Date(BASE_TIME + 30_000)); + const trace = buildTrace(run, events, new Date(BASE_TIME + 30_000), { + isCompleteHistory: true, + }); const stepSpan = trace.spans.find((span) => span.resource === 'step'); expect(stepSpan).toBeDefined(); expect(otelTimeToMs(stepSpan?.endTime ?? [0, 0])).toBe(BASE_TIME + 4000); expect(trace.knownDurationMs).toBe(4000); }); + + it('keeps every event in the geometry when the log may be incomplete', () => { + const events = [ + event('run_created', { at: 0 }), + event('run_started', { at: 0 }), + event('step_created', { correlationId: 'step_a', at: 1 }), + event('step_started', { correlationId: 'step_a', at: 1 }), + event('step_completed', { correlationId: 'step_a', at: 4 }), + event('step_completed', { correlationId: 'step_a', at: 20 }), + ]; + + // Without the whole log there is no telling which of the two completions + // the run acted on, so neither is dropped and the span covers both. + const trace = buildTrace(run, events, new Date(BASE_TIME + 30_000)); + const stepSpan = trace.spans.find((span) => span.resource === 'step'); + + expect(otelTimeToMs(stepSpan?.endTime ?? [0, 0])).toBe(BASE_TIME + 20_000); + expect(trace.duplicateEventIds.size).toBe(0); + }); }); diff --git a/packages/web-shared/src/lib/trace-builder.ts b/packages/web-shared/src/lib/trace-builder.ts index 053a49f799..b8d1186efd 100644 --- a/packages/web-shared/src/lib/trace-builder.ts +++ b/packages/web-shared/src/lib/trace-builder.ts @@ -189,18 +189,33 @@ export interface TraceWithMeta { resources: { name: string; attributes: Record }[]; /** Duration in ms from trace start to the latest known event. */ knownDurationMs: number; + /** + * The events left out of the span geometry as repeats. Empty unless the + * caller vouched for the log being complete. See + * {@link findDuplicateEventIds}. + */ + duplicateEventIds: ReadonlySet; } export function buildTrace( run: WorkflowRun, events: Event[], - now: Date + now: Date, + /** + * Whether `events` is the run's whole log. Defaults to false, which builds + * the trace from every event: on a subset there is no way to tell a repeat + * from the only copy the caller was given, and dropping the wrong one moves + * a span. See {@link findDuplicateEventIds}. + */ + { isCompleteHistory = false }: { isCompleteHistory?: boolean } = {} ): TraceWithMeta { // Span geometry comes from what the run acted on. A repeat of a class the - // log already records was passed over by the runtime, and letting one - // through here would stretch a span to whenever a concurrent replay - // committed it. The event lists still show them, marked as ignored. - const duplicateEventIds = findDuplicateEventIds(events); + // log already records is read past by every replay, and letting one through + // here would stretch a span to whenever a concurrent replay committed it. + // The event lists still show them, marked as repeats. + const duplicateEventIds = findDuplicateEventIds(events, { + isCompleteHistory, + }); const actedOnEvents = duplicateEventIds.size === 0 ? events @@ -233,5 +248,6 @@ export function buildTrace( }, ], knownDurationMs: Math.max(0, knownDurationMs), + duplicateEventIds, }; } diff --git a/packages/world/src/test-support/duplicate-event-fixtures.ts b/packages/world/src/test-support/duplicate-event-fixtures.ts new file mode 100644 index 0000000000..e3ebcc7b2a --- /dev/null +++ b/packages/world/src/test-support/duplicate-event-fixtures.ts @@ -0,0 +1,148 @@ +import type { EventType } from '../events.js'; + +/** + * Logs that mix events a replay reads past with events that only look like it. + * + * Two codebases answer the same question about a log and must answer it the + * same way. The runtime decides it while replaying: `EventsConsumer` passes + * over an event whose class it already consumed for that entity and which + * every registered callback declined. The observability UI decides it after + * the fact, from the log alone, with no consumers to ask. It stands in for + * them with the log's own record of when each entity finished, because that is + * the point past which the runtime has no consumer left for the entity. + * + * The two rules agree on every fixture here, and the interesting ones are the + * near misses: a retried step writes several `step_started` events that are + * all consumed, and a step that is still open absorbs a second `step_created`. + * Reading those as repeats would grey out attempts that ran. + * + * Each fixture is a whole run's log in log order, which is what both rules + * take as input. Entities are named, not correlation-id shaped, so each side + * can mint ids in whatever form it drives. + */ +export interface DuplicateEventFixtureEvent { + eventType: EventType; + /** The entity the event belongs to. Run-level events belong to none. */ + entity?: string; +} + +export interface DuplicateEventFixture { + name: string; + /** What makes this log worth pinning down. */ + why: string; + /** One run's whole log, in log order. */ + events: DuplicateEventFixtureEvent[]; + /** + * Indices into {@link events} of the events no consumer claims: the ones the + * runtime steps over and the UI greys out. Every other index is an event the + * run acted on. + */ + ignoredIndices: number[]; +} + +export const DUPLICATE_EVENT_FIXTURES: readonly DuplicateEventFixture[] = [ + { + name: 'start after the step completed', + why: 'A replay working from a prefix that predates the result re-invokes a step whose outcome is already recorded.', + events: [ + { eventType: 'run_created' }, + { eventType: 'run_started' }, + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + { eventType: 'step_completed', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + ], + ignoredIndices: [5], + }, + { + name: 'retry attempts', + why: 'Each attempt of a retried step writes its own start, and the live step consumer claims every one of them.', + events: [ + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + { eventType: 'step_retrying', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + { eventType: 'step_retrying', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + { eventType: 'step_completed', entity: 'step_a' }, + ], + ignoredIndices: [], + }, + { + name: 'second creation of an open step', + why: 'A step that has not finished still has a consumer, and it absorbs the repeat rather than leaving it unclaimed.', + events: [ + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + ], + ignoredIndices: [], + }, + { + name: 'second outcome for one step', + why: 'Completion and failure are one class, so the later outcome is a repeat of the earlier one whichever way round they land.', + events: [ + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + { eventType: 'step_failed', entity: 'step_a' }, + { eventType: 'step_completed', entity: 'step_a' }, + ], + ignoredIndices: [3], + }, + { + name: 'class the log has not recorded yet', + why: 'The trailing start repeats nothing, so nobody claiming it is divergence rather than a repeat, and neither side may hide it.', + events: [ + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_completed', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + ], + ignoredIndices: [], + }, + { + name: 'sleep recreated after it elapsed', + why: 'Waits close on completion the way steps close on their outcome.', + events: [ + { eventType: 'wait_created', entity: 'wait_a' }, + { eventType: 'wait_completed', entity: 'wait_a' }, + { eventType: 'wait_created', entity: 'wait_a' }, + ], + ignoredIndices: [2], + }, + { + name: 'repeated hook deliveries', + why: 'Deliveries belong to no class: a hook can be called any number of times, and every call is a call the run saw.', + events: [ + { eventType: 'hook_created', entity: 'hook_a' }, + { eventType: 'hook_received', entity: 'hook_a' }, + { eventType: 'hook_received', entity: 'hook_a' }, + { eventType: 'hook_disposed', entity: 'hook_a' }, + ], + ignoredIndices: [], + }, + { + name: 'two steps in flight', + why: 'Classes are tracked per entity, so sibling steps running the same shape never collide.', + events: [ + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_created', entity: 'step_b' }, + { eventType: 'step_started', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_b' }, + { eventType: 'step_completed', entity: 'step_b' }, + { eventType: 'step_completed', entity: 'step_a' }, + ], + ignoredIndices: [], + }, + { + name: 'second start of the run', + why: 'Run events carry no correlation id and share one bucket. Two replays can each write a start, and the run has one.', + events: [ + { eventType: 'run_created' }, + { eventType: 'run_started' }, + { eventType: 'run_started' }, + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + ], + ignoredIndices: [2], + }, +]; From 8955cce7f1e1b6262211301649618b47b3526d90 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Wed, 12 Aug 2026 14:27:19 -0700 Subject: [PATCH 5/6] Drop run_terminal from the closing classes after the merge Main's entityEventClass gives the run's own terminal events no class, so nothing consumes them and nothing can repeat them. --- .../core/src/duplicate-event-fixtures.test.ts | 1 - .../src/lib/duplicate-events.test.ts | 21 ++++++++++++------- .../web-shared/src/lib/duplicate-events.ts | 9 ++++++-- 3 files changed, 21 insertions(+), 10 deletions(-) diff --git a/packages/core/src/duplicate-event-fixtures.test.ts b/packages/core/src/duplicate-event-fixtures.test.ts index ae2dca7b33..f93d0bcd65 100644 --- a/packages/core/src/duplicate-event-fixtures.test.ts +++ b/packages/core/src/duplicate-event-fixtures.test.ts @@ -26,7 +26,6 @@ const TERMINAL_CLASSES = new Set([ 'step_terminal', 'wait_completed', 'hook_disposed', - 'run_terminal', ]); /** diff --git a/packages/web-shared/src/lib/duplicate-events.test.ts b/packages/web-shared/src/lib/duplicate-events.test.ts index 78f1fe7efd..6b1fc433d8 100644 --- a/packages/web-shared/src/lib/duplicate-events.test.ts +++ b/packages/web-shared/src/lib/duplicate-events.test.ts @@ -141,18 +141,25 @@ describe('findDuplicateEventIds', () => { expect(findDuplicateEventIds(events, COMPLETE)).toEqual(new Set()); }); - it('flags run-level repeats, which carry no correlation id', () => { + it('flags a second start of the run, which carries no correlation id', () => { const started = event('run_started'); const startedAgain = event('run_started'); + + expect(findDuplicateEventIds([started, startedAgain], COMPLETE)).toEqual( + new Set([startedAgain.eventId]) + ); + }); + + it('leaves a second outcome for the run alone', () => { + // Nothing consumes the run's own terminal events: the runtime exits rather + // than replaying the body once the log holds one. A second is a fault + // worth seeing, not a repeat the run passed over. const completed = event('run_completed'); const cancelled = event('run_cancelled'); - expect( - findDuplicateEventIds( - [started, startedAgain, completed, cancelled], - COMPLETE - ) - ).toEqual(new Set([startedAgain.eventId, cancelled.eventId])); + expect(findDuplicateEventIds([completed, cancelled], COMPLETE)).toEqual( + new Set() + ); }); it('folds in log order, not in createdAt order', () => { diff --git a/packages/web-shared/src/lib/duplicate-events.ts b/packages/web-shared/src/lib/duplicate-events.ts index 37ffbb0793..490f36a86f 100644 --- a/packages/web-shared/src/lib/duplicate-events.ts +++ b/packages/web-shared/src/lib/duplicate-events.ts @@ -24,12 +24,17 @@ import { * consumer remains. */ -/** Classes whose event closes its entity: no consumer is left for it after. */ +/** + * Classes whose event closes its entity: no consumer is left for it after. + * + * The run's own terminal events are absent because `entityEventClass` gives + * them no class. The runtime exits rather than replaying the body once the log + * holds one, so nothing ever consumes them and nothing can repeat them. + */ const TERMINAL_EVENT_CLASSES: ReadonlySet = new Set([ 'step_terminal', 'wait_completed', 'hook_disposed', - 'run_terminal', ]); /** Classes with no entity to close first: the log records one per run. */ From 94e05bd8804f70138d41ba2f369a6e33d636da95 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Thu, 13 Aug 2026 11:13:44 -0700 Subject: [PATCH 6/6] [web-shared] Only classify duplicates where the log order is knowable Two gaps in the fold, both from review on #3467. The event ID is a log position only for a slot-numbered run, whose slot is drawn at the publish. A ULID-numbered run is served by one backend in (createdAt, eventId) order and by another keyed on the ID, so concurrent writers can produce opposite orders and the ID alone does not fix the position. Such a log is now classified only where its timestamps corroborate its IDs. The fold also recorded a class first seen after its entity had closed, which is the point the runtime reports divergence and exits. A later event of that class then looked like a settled repeat although the run never read it. The fold stops there instead. --- .../src/lib/duplicate-events.test.ts | 63 +++++++++ .../web-shared/src/lib/duplicate-events.ts | 120 +++++++++++++----- .../test-support/duplicate-event-fixtures.ts | 11 ++ 3 files changed, 164 insertions(+), 30 deletions(-) diff --git a/packages/web-shared/src/lib/duplicate-events.test.ts b/packages/web-shared/src/lib/duplicate-events.test.ts index 6b1fc433d8..7111957ea2 100644 --- a/packages/web-shared/src/lib/duplicate-events.test.ts +++ b/packages/web-shared/src/lib/duplicate-events.test.ts @@ -37,6 +37,20 @@ function event( } as unknown as Event; } +/** + * The same event under the older ID scheme, whose IDs are ULIDs rather than + * slots. A backend serving such a log may order it by `(createdAt, eventId)` + * instead of by ID, so the ID alone does not fix the log position. + */ +function ulidEvent(...args: Parameters): Event { + const slotEvent = event(...args); + const slot = slotEvent.eventId.slice('evnt_'.length).replace(/^0+/, ''); + return { + ...slotEvent, + eventId: `evnt_01K${slot.padStart(23, '0')}`, + } as Event; +} + describe('findDuplicateEventIds', () => { it('returns nothing for a log with no repeats', () => { const events = [ @@ -224,6 +238,55 @@ describe('findDuplicateEventIds', () => { expect(descending).toEqual(ascending); }); + it('classifies a ULID log whose timestamps corroborate its ids', () => { + const created = ulidEvent('wait_created', { correlationId: 'wait_a' }); + const completed = ulidEvent('wait_completed', { correlationId: 'wait_a' }); + const createdAgain = ulidEvent('wait_created', { correlationId: 'wait_a' }); + + expect( + findDuplicateEventIds([created, completed, createdAgain], COMPLETE) + ).toEqual(new Set([createdAgain.eventId])); + }); + + it('classifies nothing on a ULID log whose timestamps contradict its ids', () => { + // A ULID carries no log position: one backend returns such a log in + // createdAt order and another in id order, and createdAt is stamped when + // the write arrives rather than when it commits. With the two orders + // disagreeing, which wait_created the run acted on depends on the backend, + // so naming either would be a guess. + const created = ulidEvent('wait_created', { correlationId: 'wait_a' }); + const completed = ulidEvent('wait_completed', { correlationId: 'wait_a' }); + const createdAgain = ulidEvent('wait_created', { + correlationId: 'wait_a', + at: -60, + }); + + expect( + findDuplicateEventIds([created, completed, createdAgain], COMPLETE) + ).toEqual(new Set()); + }); + + it('classifies nothing past the point the run diverged', () => { + // The step finished without a step_started, so the runtime reports + // divergence on the first trailing start and exits. The second start and + // the wait's repeat after it went unread, and neither is a repeat the run + // passed over. The wait's repeat before it still is. + const events = [ + event('wait_created', { correlationId: 'wait_a' }), + event('wait_completed', { correlationId: 'wait_a' }), + event('wait_created', { correlationId: 'wait_a' }), + event('step_created', { correlationId: 'step_a' }), + event('step_completed', { correlationId: 'step_a' }), + event('step_started', { correlationId: 'step_a' }), + event('step_started', { correlationId: 'step_a' }), + event('wait_created', { correlationId: 'wait_a' }), + ]; + + expect(findDuplicateEventIds(events, COMPLETE)).toEqual( + new Set([events[2].eventId]) + ); + }); + it('classifies nothing when the caller holds part of the log', () => { // A newest-first page can open on the repeat and omit the event it // repeats, which would invert the answer. diff --git a/packages/web-shared/src/lib/duplicate-events.ts b/packages/web-shared/src/lib/duplicate-events.ts index 490f36a86f..3e3d29cccd 100644 --- a/packages/web-shared/src/lib/duplicate-events.ts +++ b/packages/web-shared/src/lib/duplicate-events.ts @@ -2,6 +2,7 @@ import { type EntityEventClass, type Event, entityEventClass, + isSlotEventId, } from '@workflow/world'; /** @@ -55,48 +56,71 @@ export const DUPLICATE_EVENT_MESSAGE = 'Written by a concurrent replay after an event of the same kind was already recorded and acted on. The run follows the earlier one.'; /** - * Log order. + * Candidate log order, by event ID. * * Event IDs are fixed-width and monotonic within a run under both the ULID and - * the slot scheme, so comparing them orders the log even where timestamps do - * not: a writer stamps `createdAt` on entry but takes its log position at - * publish time, and `occurredAt` is measured on the client. Length is compared - * first so a shorter ID never sorts after a longer one on a fixture or a log - * that mixes widths. + * the slot scheme. Length is compared first so a shorter ID never sorts after + * a longer one on a log that mixes widths. + * + * Whether this *is* the log order depends on the ID scheme, which is why + * {@link hasKnowableLogOrder} gates the fold. See its doc. */ -function compareLogPosition(a: Event, b: Event): number { +function compareEventId(a: Event, b: Event): number { if (a.eventId.length !== b.eventId.length) { return a.eventId.length - b.eventId.length; } return a.eventId < b.eventId ? -1 : a.eventId > b.eventId ? 1 : 0; } +function createdAtMs(event: Event): number { + const createdAt = event.createdAt; + return createdAt instanceof Date + ? createdAt.getTime() + : new Date(createdAt as unknown as string).getTime(); +} + /** - * The IDs of the events in `events` that repeat a class the log already - * records for the same entity, after that entity finished. + * Whether the order the run consumed its log in can be recovered from the + * events alone. * - * `isCompleteHistory` must be false whenever the caller holds a subset of the - * run's log: one page of a paginated list, or the result of a search. Which - * occurrence of a class came first is a property of the whole log, so on a - * subset the earlier event may simply be missing, and the fold would report - * the surviving one. Nothing is classified in that case. + * Which occurrence of a class came first is the whole question here, so the + * fold needs the log's order, not an order. The backends do not agree on how + * to recover it for every ID scheme: + * + * - A **slot-numbered** run carries its position in the ID. The slot is drawn + * at the publish, which is the linearization point, so slot order is log + * order everywhere and the ID alone settles it. + * - A **ULID-numbered** run does not. One backend returns such a log in + * `(createdAt, eventId)` order while another returns it keyed on the ID, and + * `createdAt` is stamped when the write request arrives rather than when it + * commits. Concurrent writers can therefore produce opposite timestamp and + * ID orders, and the run consumed whichever its own backend served. + * + * So a ULID log is only knowable where the two orders agree. Where they + * contradict, the fold could name the surviving event and pass over the one + * the run acted on, which is worse than saying nothing. */ -export function findDuplicateEventIds( - events: readonly Event[], - { isCompleteHistory }: { isCompleteHistory: boolean } -): Set { - const duplicates = new Set(); - if (!isCompleteHistory || events.length < 2) return duplicates; +function hasKnowableLogOrder(orderedById: readonly Event[]): boolean { + if (orderedById.every((event) => isSlotEventId(event.eventId))) return true; + + for (let index = 1; index < orderedById.length; index++) { + const previous = createdAtMs(orderedById[index - 1]); + const current = createdAtMs(orderedById[index]); + // A missing or unparseable timestamp leaves nothing to corroborate the ID + // order with, which is the same position as a contradiction. + if (Number.isNaN(previous) || Number.isNaN(current)) return false; + if (current < previous) return false; + } + + return true; +} +/** The fold itself, over a log whose order is known. */ +function foldDuplicates(ordered: readonly Event[]): Set { + const duplicates = new Set(); const seenClasses = new Set(); const closedEntities = new Set(); - // Dropped before the sort, not during the fold: an event with no ID has no - // log position to order on, and the caller could not match it either. - const ordered = events - .filter((event) => Boolean(event.eventId)) - .sort(compareLogPosition); - for (const event of ordered) { const eventClass = entityEventClass(event.eventType); if (eventClass === undefined) continue; @@ -111,18 +135,54 @@ export function findDuplicateEventIds( } if (!repeatsClass) { + // First of its class, but the entity already finished: no consumer is + // left to take it and it repeats nothing, so the runtime reports + // divergence here and exits. Everything past this point went unread, so + // the fold stops with it rather than recording the class and presenting + // a later event of it as a repeat the run passed over. + if (entityWasClosed) break; seenClasses.add(classKey); continue; } // The entity is still open, so a consumer is registered for it and takes // this event: another attempt, not a repeat read past. - if (!entityWasClosed && !SINGLETON_EVENT_CLASSES.has(eventClass)) { - continue; + if (entityWasClosed || SINGLETON_EVENT_CLASSES.has(eventClass)) { + duplicates.add(event.eventId); } - - duplicates.add(event.eventId); } return duplicates; } + +/** + * The IDs of the events in `events` that repeat a class the log already + * records for the same entity, after that entity finished. + * + * `isCompleteHistory` must be false whenever the caller holds a subset of the + * run's log: one page of a paginated list, or the result of a search. Which + * occurrence of a class came first is a property of the whole log, so on a + * subset the earlier event may simply be missing, and the fold would report + * the surviving one. Nothing is classified in that case. + * + * Two other things make the answer unknowable and yield the same empty result: + * a log whose order cannot be recovered from the events (see + * {@link hasKnowableLogOrder}), and everything past the point the run + * diverged, since the run exited there and read no further. + */ +export function findDuplicateEventIds( + events: readonly Event[], + { isCompleteHistory }: { isCompleteHistory: boolean } +): Set { + if (!isCompleteHistory || events.length < 2) return new Set(); + + // Dropped before the sort, not during the fold: an event with no ID has no + // log position to order on, and the caller could not match it either. + const ordered = events + .filter((event) => Boolean(event.eventId)) + .sort(compareEventId); + + if (!hasKnowableLogOrder(ordered)) return new Set(); + + return foldDuplicates(ordered); +} diff --git a/packages/world/src/test-support/duplicate-event-fixtures.ts b/packages/world/src/test-support/duplicate-event-fixtures.ts index e3ebcc7b2a..dbe6947b16 100644 --- a/packages/world/src/test-support/duplicate-event-fixtures.ts +++ b/packages/world/src/test-support/duplicate-event-fixtures.ts @@ -99,6 +99,17 @@ export const DUPLICATE_EVENT_FIXTURES: readonly DuplicateEventFixture[] = [ ], ignoredIndices: [], }, + { + name: 'repeat of a class the log has not recorded yet', + why: 'The run stops on the first trailing start and never reads the second, so neither side may present it as a repeat the run passed over.', + events: [ + { eventType: 'step_created', entity: 'step_a' }, + { eventType: 'step_completed', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + { eventType: 'step_started', entity: 'step_a' }, + ], + ignoredIndices: [], + }, { name: 'sleep recreated after it elapsed', why: 'Waits close on completion the way steps close on their outcome.',