diff --git a/.changeset/silly-pears-jam.md b/.changeset/silly-pears-jam.md new file mode 100644 index 0000000000..055dfa8ec0 --- /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, 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..f93d0bcd65 --- /dev/null +++ b/packages/core/src/duplicate-event-fixtures.test.ts @@ -0,0 +1,120 @@ +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', +]); + +/** + * 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 38ba75e6cb..aefa6b63da 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,24 @@ 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 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. 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 { +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 // 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 +843,7 @@ export function EventRow({ onEncryptedDataDetected, suppressGroupDimming = false, showSeparateEventOccurrenceTimestamps = false, + isDuplicate = false, }: { event: Event; index: number; @@ -856,6 +868,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 +1109,49 @@ export function EventRow({ {/* Event Type */}
- + - {isPulsing && ( - - )} + > + {isPulsing && ( + + )} + + + {formatEventType(event.eventType)} - {formatEventType(event.eventType)} - +
{/* Name */} @@ -1310,6 +1330,23 @@ function EventListViewInner({ ); }, [events, effectiveSortOrder, isExactSearchActive, searchResults]); + // 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(events ?? [], { + isCompleteHistory: !hasMoreEvents && !isExactSearchActive, + }), + [events, hasMoreEvents, isExactSearchActive] + ); + // Detect encrypted fields across all loaded events (inline eventData). const hasEncryptedInlineData = useMemo(() => { const sourceEvents = isExactSearchActive ? searchResults : events; @@ -1342,8 +1379,8 @@ function EventListViewInner({ ); const durationMap = useMemo( - () => buildDurationMap(sortedEvents), - [sortedEvents] + () => buildDurationMap(sortedEvents, duplicateEventIds), + [sortedEvents, duplicateEventIds] ); const [selectedGroupKey, setSelectedGroupKey] = useState( @@ -1804,6 +1841,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/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 5674158d52..7fd65bc10d 100644 --- a/packages/web-shared/src/components/sidebar/events-list.tsx +++ b/packages/web-shared/src/components/sidebar/events-list.tsx @@ -10,6 +10,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 +48,7 @@ function EventItem({ onLoadEventData, encryptionKey, showSeparateEventOccurrenceTimestamps = false, + isDuplicate = false, }: { event: Event; onLoadEventData?: (event: Event) => Promise; @@ -54,6 +56,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 +139,15 @@ function EventItem({ >
- - {event.eventType} - + + + {event.eventType} + + {displayedCreatedAtTime} @@ -294,6 +304,7 @@ export function EventsList({ onRunClick, encryptionKey, showSeparateEventOccurrenceTimestamps = false, + duplicateEventIds, }: { events: Event[]; isLoading?: boolean; @@ -305,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( @@ -352,6 +369,7 @@ export function EventsList({ showSeparateEventOccurrenceTimestamps={ showSeparateEventOccurrenceTimestamps } + 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 ( - +
+ + {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..7111957ea2 --- /dev/null +++ b/packages/web-shared/src/lib/duplicate-events.test.ts @@ -0,0 +1,339 @@ +import type { Event, EventType } from '@workflow/world'; +import { DUPLICATE_EVENT_FIXTURES } from '@workflow/world/test-support/duplicate-event-fixtures.js'; +import { describe, expect, it } from 'vitest'; +import { findDuplicateEventIds } from './duplicate-events'; + +const BASE_TIME = Date.parse('2026-01-01T00:00:00.000Z'); + +const COMPLETE = { isCompleteHistory: true }; + +let nextSlot = 0; + +function event( + eventType: EventType, + options: { + correlationId?: string; + /** Log position. Defaults to the order the fixture created the event in. */ + slot?: number; + /** `createdAt`/`occurredAt`, in seconds after the fixture epoch. */ + at?: number; + occurredAt?: number; + } = {} +): Event { + nextSlot += 1; + const slot = options.slot ?? nextSlot; + const createdAt = new Date(BASE_TIME + (options.at ?? slot) * 1000); + return { + eventId: `evnt_${String(slot).padStart(26, '0')}`, + runId: 'run_1', + eventType, + correlationId: options.correlationId, + createdAt, + occurredAt: + options.occurredAt === undefined + ? createdAt + : new Date(BASE_TIME + options.occurredAt * 1000), + eventData: {}, + } 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 = [ + 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, COMPLETE)).toEqual(new Set()); + }); + + 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( + [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', () => { + // 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], COMPLETE)).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, 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. 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' }), + 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, 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', () => { + 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, COMPLETE)).toEqual(new Set()); + }); + + 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([completed, cancelled], COMPLETE)).toEqual( + new Set() + ); + }); + + 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(events, COMPLETE); + const descending = findDuplicateEventIds([...events].reverse(), COMPLETE); + + 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 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. + 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 = (eventType: EventType) => + ({ + ...event(eventType, { correlationId: 'wait_a' }), + eventId: undefined, + }) as unknown as Event; + + expect( + findDuplicateEventIds( + [ + anonymous('wait_created'), + anonymous('wait_completed'), + anonymous('wait_created'), + ], + COMPLETE + ) + ).toEqual(new Set()); + }); +}); + +/** + * 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(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 new file mode 100644 index 0000000000..3e3d29cccd --- /dev/null +++ b/packages/web-shared/src/lib/duplicate-events.ts @@ -0,0 +1,188 @@ +import { + type EntityEventClass, + type Event, + entityEventClass, + isSlotEventId, +} from '@workflow/world'; + +/** + * Identifies events a replay reads past. + * + * 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. + */ + +/** + * 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', +]); + +/** 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.'; + +/** + * Candidate log order, by event ID. + * + * Event IDs are fixed-width and monotonic within a run under both the ULID and + * 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 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(); +} + +/** + * Whether the order the run consumed its log in can be recovered from the + * events alone. + * + * 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. + */ +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(); + + 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) { + // 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)) { + 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/web-shared/src/lib/event-materialization.test.ts b/packages/web-shared/src/lib/event-materialization.test.ts new file mode 100644 index 0000000000..d26ae41d0c --- /dev/null +++ b/packages/web-shared/src/lib/event-materialization.test.ts @@ -0,0 +1,117 @@ +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, { isCompleteHistory: true }); + + 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_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, { 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 a605e9c624..d2f7a56c8b 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,24 @@ 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 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, { + isCompleteHistory, + }); const steps: MaterializedStep[] = []; for (const [correlationId, stepEvents] of groups) { @@ -121,6 +137,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'; @@ -248,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 new file mode 100644 index 0000000000..70ad41339a --- /dev/null +++ b/packages/web-shared/src/lib/trace-builder.test.ts @@ -0,0 +1,74 @@ +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), { + 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 6b01e5b042..b8d1186efd 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'; /** @@ -188,15 +189,40 @@ 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 { - 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 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 + : events.filter((event) => !duplicateEventIds.has(event.eventId)); + + const groupedEvents = groupEventsByCorrelation(actedOnEvents); + const latestKnownTime = computeLatestKnownTime(actedOnEvents, run); const { runSpan, spans } = buildSpans( run, groupedEvents, @@ -222,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..dbe6947b16 --- /dev/null +++ b/packages/world/src/test-support/duplicate-event-fixtures.ts @@ -0,0 +1,159 @@ +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: '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.', + 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], + }, +];