diff --git a/.changeset/divergence-diagnostics-core.md b/.changeset/divergence-diagnostics-core.md new file mode 100644 index 0000000000..2ab187a39e --- /dev/null +++ b/.changeset/divergence-diagnostics-core.md @@ -0,0 +1,6 @@ +--- +'@workflow/core': patch +'@workflow/world': patch +--- + +When logging corrupt event logs due to replay divergence, specify the divergent events, and carry error message through retries diff --git a/.changeset/divergence-diagnostics-world.md b/.changeset/divergence-diagnostics-world.md new file mode 100644 index 0000000000..a845151cc8 --- /dev/null +++ b/.changeset/divergence-diagnostics-world.md @@ -0,0 +1,2 @@ +--- +--- diff --git a/docs/content/docs/v5/errors/corrupted-event-log.mdx b/docs/content/docs/v5/errors/corrupted-event-log.mdx index ba82e48ea5..5ad3d13520 100644 --- a/docs/content/docs/v5/errors/corrupted-event-log.mdx +++ b/docs/content/docs/v5/errors/corrupted-event-log.mdx @@ -18,9 +18,19 @@ This is a **workflow-level fatal error**. It cannot be caught or handled inside For replay divergence: ```text -Workflow replay diverged times after recovery replays; latest divergent event was . Last divergence:
+Workflow replay diverged times after recovery replays; latest divergent event was ; divergent event ids: , , ... Last divergence:
``` +The `divergent event ids` list has one entry per divergence in the recovery chain, oldest first. Every recovery replay diverging at the same event points at a fixed disagreement between the log and the code; ids that wander point at a race with another writer. + +`
` is the last divergence's own message. When the replay could not place an event, it names the invocation that was pending under that event's position at the time, and where the replay's walk over the log stood: + +```text +Replay could not consume event: eventType=wait_created, correlationId=wait_, eventId=. pending at this id: step (step_). consumer: index=, length=, parked=, lastConsumed= +``` + +Steps, sleeps and hooks draw their ids from one sequence, so `pending at this id` reports the entity the replay put at that position, whatever its kind. In the example, the log recorded a sleep where this replay reached a step named ``. + For an unreadable stored payload: ```text diff --git a/packages/core/src/events-consumer.ts b/packages/core/src/events-consumer.ts index 3fb7d48b12..0a997ddd67 100644 --- a/packages/core/src/events-consumer.ts +++ b/packages/core/src/events-consumer.ts @@ -188,6 +188,14 @@ export interface EventsConsumerOptions { isDeliveryIdle: () => boolean; } +/** See {@link EventsConsumer.describe}. */ +export interface EventsConsumerSnapshot { + index: number; + length: number; + parked: number; + lastConsumedEventId: string | undefined; +} + export class EventsConsumer { eventIndex: number; readonly events: Event[]; @@ -228,6 +236,13 @@ export class EventsConsumer { private pendingUnconsumedCheck: Promise | null = null; private pendingUnconsumedTimeout: ReturnType | null = null; private unconsumedCheckVersion = 0; + /** + * The event a callback most recently claimed, for {@link describe}. Tracked + * separately from {@link eventIndex} because the walk also steps over events + * (parked, sealed no-ops, duplicates) without anyone consuming them, so + * `events[eventIndex - 1]` does not say what replay last acted on. + */ + private lastConsumed: Event | undefined; constructor(events: Event[], options: EventsConsumerOptions) { // Own copy: the runtime mutates its event array in place, and a retained @@ -277,6 +292,26 @@ export class EventsConsumer { }; } + /** + * Where the walk stands, for a divergence message. Everything here is + * already known to the consumer; the point of the method is that the + * orchestrator can print it without reaching into private state. + * + * `index` is the ordered walk's position (the offset of the event it is + * stuck on, when it is stuck), `length` the log as this consumer holds it, + * `parked` how many events the walk stepped over and still holds, and + * `lastConsumedEventId` the id of the event a callback most recently + * claimed, `undefined` before the first claim. + */ + describe(): EventsConsumerSnapshot { + return { + index: this.eventIndex, + length: this.events.length, + parked: this.parked.length, + lastConsumedEventId: this.lastConsumed?.eventId, + }; + } + append(events: Event[]): void { for (const event of events) this.events.push(event); process.nextTick(this.consume); @@ -409,6 +444,7 @@ export class EventsConsumer { ); } this.recordEventClass(currentEvent); + this.lastConsumed = currentEvent; this.notifyConsumedEvent(currentEvent); } // remove the callback if it has finished diff --git a/packages/core/src/log-format.test.ts b/packages/core/src/log-format.test.ts index c673893f06..f9d58686cb 100644 --- a/packages/core/src/log-format.test.ts +++ b/packages/core/src/log-format.test.ts @@ -156,7 +156,46 @@ describe('composeLogLine', () => { user error · Error run wrun_01ABC · myWorkflow (./workflows/x) step step_01XYZ · add (./workflows/x) - retry 4 attempts · 3 max retries" + retry 4 attempts · 3 max retries + error Transient failure" `); }); + + // A WARN whose framing is a fixed sentence and whose message has no stack + // body carries the underlying error's text only in `errorMessage`. It used + // to be dropped because the key was well-known but never rendered, which is + // how a production divergence WARN lost the text saying what diverged. + test('renders errorMessage when neither framing nor body carries it', () => { + const out = composeLogLine( + PREFIX, + 'Workflow replay diverged; queueing a recovery replay before declaring the event log corrupted', + { + errorCode: 'REPLAY_DIVERGENCE', + divergenceEventId: 'evnt_3', + divergenceCount: 1, + loopIteration: 2, + servedByRetainedSession: false, + errorMessage: + 'Replay could not consume event: eventType=wait_created, correlationId=wait_01K, eventId=evnt_3. pending at this id: step drainStep (step_01K).', + } + ); + expect(out).toMatchInlineSnapshot(` + "[workflow-sdk] Workflow replay diverged; queueing a recovery replay before declaring the event log corrupted + code REPLAY_DIVERGENCE + error Replay could not consume event: eventType=wait_created, correlationId=wait_01K, eventId=evnt_3. pending at this id: step drainStep (step_01K). + divergenceCount 1 + divergenceEventId evnt_3 + loopIteration 2 + servedByRetainedSession false" + `); + }); + + test('drops errorMessage when the stack body already carries it', () => { + const out = composeLogLine( + PREFIX, + 'Workflow failed\nError: boom\n at x (./workflows/x.ts:1:1)', + { errorName: 'Error', errorMessage: 'boom' } + ); + expect(out).not.toMatch(/^\s+error\s+boom/m); + }); }); diff --git a/packages/core/src/log-format.ts b/packages/core/src/log-format.ts index 5e95a3a7d0..a2c50bff99 100644 --- a/packages/core/src/log-format.ts +++ b/packages/core/src/log-format.ts @@ -36,7 +36,7 @@ export function composeLogLine( ): string { const [framing, ...rest] = message.split('\n'); const body = rest.join('\n'); - const fields = renderStructuredFields(framing ?? '', metadata); + const fields = renderStructuredFields(framing ?? '', body, metadata); const trimmedBody = trimStackBody(body); const lines: string[] = [`${prefix} ${framing ?? ''}`]; @@ -47,18 +47,23 @@ export function composeLogLine( function renderStructuredFields( framing: string, + body: string, metadata: Record | undefined ): string | null { if (!metadata || Object.keys(metadata).length === 0) return null; // Drop fields that the message already encodes. We render framings and // stacks into the message string itself in step executor / combined runtime, so - // repeating them here would be pure noise. + // repeating them here would be pure noise. A message with neither (a WARN + // whose framing is a fixed sentence and whose `errorMessage` is the only + // place the underlying error's text appears) keeps `errorMessage` and + // renders it as its own row below. const redundant = new Set(); redundant.add('errorStack'); + const errorMessage = pickString(metadata, 'errorMessage'); if ( - typeof metadata.errorMessage === 'string' && - framing.includes(metadata.errorMessage as string) + errorMessage && + (framing.includes(errorMessage) || body.includes(errorMessage)) ) { redundant.add('errorMessage'); } @@ -130,6 +135,10 @@ function renderStructuredFields( lines.push(` ${kvKey('code')} ${Ansi.dim(errorCode)}`); } + if (errorMessage && !redundant.has('errorMessage')) { + lines.push(` ${kvKey('error')} ${formatPassthroughValue(errorMessage)}`); + } + const hint = pickString(metadata, 'hint'); if (hint) { lines.push(` ${Ansi.hint(hint)}`); diff --git a/packages/core/src/logger.test.ts b/packages/core/src/logger.test.ts index 5560c9e213..78066923ca 100644 --- a/packages/core/src/logger.test.ts +++ b/packages/core/src/logger.test.ts @@ -148,6 +148,7 @@ describe('logger', () => { user error · FatalError run wrun_123 step step_456 + error boom hint: Move the call to a step function.", ], ] @@ -178,7 +179,8 @@ describe('logger', () => { user error · Error run wrun_abc step step_xyz - retry 4 attempts · 3 max retries", + retry 4 attempts · 3 max retries + error Transient failure", ], ] `); diff --git a/packages/core/src/replay-divergence.test.ts b/packages/core/src/replay-divergence.test.ts new file mode 100644 index 0000000000..42856dae5d --- /dev/null +++ b/packages/core/src/replay-divergence.test.ts @@ -0,0 +1,160 @@ +import type { Event } from '@workflow/world'; +import { describe, expect, it } from 'vitest'; +import { EventsConsumer } from './events-consumer.js'; +import type { QueueItem } from './global.js'; +import { WorkflowSuspension } from './global.js'; +import { + describeDivergenceContext, + findPendingItemAtOrdinal, +} from './replay-divergence.js'; +import { dehydrateStepReturnValue } from './serialization.js'; +import { createUseStep } from './step.js'; +import { + CORR_IDS, + runWithDiscontinuation, + setupWorkflowContext, +} from './test-support/orchestrator-context.js'; + +/** + * A replay that draws the same ordinal for a different kind of entity than + * the log recorded there leaves an event nobody can consume. The message the + * runtime raises for that names the pending invocation that holds the + * ordinal, because that is what says which branch of the workflow the replay + * took differently. In the production incident this was written for, the log + * held a `wait_created` where the replay was waiting on a step. + */ + +const RESUME_AT = new Date('2099-01-01T00:00:00.000Z'); + +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; +} + +async function dehydrate(value: unknown) { + const ops: Promise[] = []; + return await dehydrateStepReturnValue(value, 'wrun_test', undefined, ops); +} + +describe('findPendingItemAtOrdinal', () => { + const queue = new Map([ + [ + `step_${CORR_IDS[0]}`, + { + type: 'step', + correlationId: `step_${CORR_IDS[0]}`, + stepName: 'a', + args: [], + }, + ], + [ + `hook_${CORR_IDS[1]}`, + { type: 'hook', correlationId: `hook_${CORR_IDS[1]}`, token: 't' }, + ], + ]); + + it('finds an exact correlation id', () => { + expect(findPendingItemAtOrdinal(queue, `step_${CORR_IDS[0]}`)?.type).toBe( + 'step' + ); + }); + + it('finds a different kind of entity holding the same ULID body', () => { + expect(findPendingItemAtOrdinal(queue, `wait_${CORR_IDS[0]}`)?.type).toBe( + 'step' + ); + expect(findPendingItemAtOrdinal(queue, `step_${CORR_IDS[1]}`)?.type).toBe( + 'hook' + ); + }); + + it('returns undefined when nothing holds the ordinal', () => { + expect( + findPendingItemAtOrdinal(queue, `wait_${CORR_IDS[2]}`) + ).toBeUndefined(); + expect(findPendingItemAtOrdinal(queue, 'no-prefix')).toBeUndefined(); + }); +}); + +describe('describeDivergenceContext', () => { + it('reports the consumer walk position and last consumed event', () => { + const events = [ + event(0, 'run_started', undefined as unknown as string, {}), + event(1, 'wait_created', `wait_${CORR_IDS[0]}`, { resumeAt: RESUME_AT }), + ]; + const consumer = new EventsConsumer(events, { + onUnconsumedEvent: () => {}, + getPromiseQueue: () => Promise.resolve(), + isDeliveryIdle: () => true, + }); + const before = describeDivergenceContext(events[1], new Map(), consumer); + expect(before).toBe( + 'pending at this id: none. consumer: index=0, length=2, parked=0, lastConsumed=none' + ); + }); +}); + +describe('unconsumable event message', () => { + it('names the step holding the ordinal of an unconsumable wait_created', async () => { + const result = await dehydrate('launched'); + // Log written by a replay that reached a sleep at the second draw; this + // replay reaches a step there instead. + const events = [ + event(0, 'step_created', `step_${CORR_IDS[0]}`, { stepName: 'launch' }), + event(1, 'step_started', `step_${CORR_IDS[0]}`, { stepName: 'launch' }), + event(2, 'step_completed', `step_${CORR_IDS[0]}`, { + stepName: 'launch', + result, + }), + event(3, 'wait_created', `wait_${CORR_IDS[1]}`, { resumeAt: RESUME_AT }), + ]; + const ctx = setupWorkflowContext(events); + const useStep = createUseStep(ctx); + + const { error } = await runWithDiscontinuation(ctx, async () => { + await useStep('launch')(); + await useStep('drainStep')(); + return 'done'; + }); + + expect(WorkflowSuspension.is(error)).toBe(false); + const message = String(error); + expect(message).toContain( + `eventType=wait_created, correlationId=wait_${CORR_IDS[1]}, eventId=evnt_3` + ); + expect(message).toContain( + `pending at this id: step drainStep (step_${CORR_IDS[1]})` + ); + expect(message).toContain( + 'consumer: index=3, length=4, parked=0, lastConsumed=evnt_2' + ); + }); + + it('says none when nothing pending holds the ordinal', async () => { + const events = [ + event(0, 'wait_created', 'wait_01JZZZZZZZZZZZZZZZZZZZZZZZ', { + resumeAt: RESUME_AT, + }), + ]; + const ctx = setupWorkflowContext(events); + const useStep = createUseStep(ctx); + + const { error } = await runWithDiscontinuation(ctx, async () => { + await useStep('launch')(); + return 'done'; + }); + + expect(String(error)).toContain('pending at this id: none.'); + }); +}); diff --git a/packages/core/src/replay-divergence.ts b/packages/core/src/replay-divergence.ts new file mode 100644 index 0000000000..d8b38d88b1 --- /dev/null +++ b/packages/core/src/replay-divergence.ts @@ -0,0 +1,71 @@ +import type { Event } from '@workflow/world'; +import type { EventsConsumer } from './events-consumer.js'; +import type { QueueItem } from './global.js'; + +/** + * Step, wait, hook and attribute correlation ids share one ULID draw per + * entity, prefixed by kind (`step_`, `wait_`, `hook_`, + * `attr_`). A replay that drew the same ordinal for a different kind of + * entity than the one the log recorded there produces an event nobody can + * consume, and the decisive fact for diagnosing that is which pending + * invocation currently holds the ordinal. This looks it up by the shared ULID + * body, so a `wait_created` the log holds at a position where the replay is + * waiting on `step_` names that step. + */ +export function findPendingItemAtOrdinal( + invocationsQueue: Map, + correlationId: string +): QueueItem | undefined { + const exact = invocationsQueue.get(correlationId); + if (exact) return exact; + const body = ordinalBody(correlationId); + if (!body) return undefined; + for (const [id, item] of invocationsQueue) { + if (ordinalBody(id) === body) return item; + } + return undefined; +} + +function ordinalBody(correlationId: string): string | undefined { + const at = correlationId.indexOf('_'); + return at === -1 ? undefined : correlationId.slice(at + 1); +} + +function describeQueueItem(item: QueueItem): string { + switch (item.type) { + case 'step': + return `step ${item.stepName} (${item.correlationId})`; + case 'hook': + return `hook (${item.correlationId})`; + case 'wait': + return `wait (${item.correlationId})`; + case 'attribute': + return `attribute (${item.correlationId})`; + } + item satisfies never; + return 'unknown'; +} + +/** + * The detail appended to a `ReplayDivergenceError` raised because the replay + * could not place an event. Callers keep their own leading sentence (tests and + * users match on it) and append this after it. + * + * Example: + * `pending at this id: step drainStep (step_01K…). consumer: index=41, + * length=44, parked=0, lastConsumed=evnt_01K…` + */ +export function describeDivergenceContext( + event: Event, + invocationsQueue: Map, + eventsConsumer: EventsConsumer +): string { + const pending = event.correlationId + ? findPendingItemAtOrdinal(invocationsQueue, event.correlationId) + : undefined; + const snapshot = eventsConsumer.describe(); + return [ + `pending at this id: ${pending ? describeQueueItem(pending) : 'none'}.`, + `consumer: index=${snapshot.index}, length=${snapshot.length}, parked=${snapshot.parked}, lastConsumed=${snapshot.lastConsumedEventId ?? 'none'}`, + ].join(' '); +} diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 88bc7002b6..763a086bc1 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -70,7 +70,7 @@ async function runWorkflowHandlerWithEvents( createdEvents?: unknown[]; createdEventParams?: unknown[]; queueCalls?: QueueCall[]; - replayDivergence?: { eventId: string; count: number }; + replayDivergence?: { eventId: string; count: number; eventIds?: string[] }; /** * Make created events visible to subsequent events.list calls (appended * to `events`), like a real World. Needed for flows where the handler @@ -819,6 +819,11 @@ describe('workflowEntrypoint replay guards', () => { }, ]; + using warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + using error = vi + .spyOn(console, 'error') + .mockImplementation(() => undefined); + const initialAttemptEvents: unknown[] = []; const queueCalls: QueueCall[] = []; await runWorkflowHandlerWithEvents( @@ -838,17 +843,39 @@ describe('workflowEntrypoint replay guards', () => { expect(initialAttemptEvents).not.toContainEqual( expect.objectContaining({ eventType: 'run_failed' }) ); + // The recovery message starts the divergence history at this event. expect(queueCalls.map((c) => c.message)).toContainEqual( expect.objectContaining({ replayDivergence: { eventId: slotToEventId(1), count: 1, + eventIds: [slotToEventId(1)], }, }) ); + // The WARN says which pass of which invocation diverged and what that + // pass was working from, and the console renderer keeps the divergence + // text (`errorMessage`) rather than dropping it as a well-known field. + const warnOut = warn.mock.calls + .map((c) => String(c[0])) + .find((line) => line.includes('Workflow replay diverged')); + expect(warnOut).toBeDefined(); + expect(warnOut).toContain('loopIteration 1'); + expect(warnOut).toContain('servedByRetainedSession false'); + expect(warnOut).toContain('eventLogLength 2'); + expect(warnOut).toContain(`eventLogLastEventId ${slotToEventId(2)}`); + expect(warnOut).toContain('isRecoveryReplay false'); + expect(warnOut).toContain('hasHookInput false'); + expect(warnOut).toContain( + `Replay could not consume event: eventType=wait_created, correlationId=wait_01HK153X00VFKAJV9XFN9JXXRS, eventId=${slotToEventId(1)}. pending at this id: none.` + ); const terminalAttemptEvents: unknown[] = []; const terminalAttemptParams: unknown[] = []; + const priorEventIds = Array.from( + { length: REPLAY_DIVERGENCE_MAX_RETRIES }, + (_, i) => `evnt_prior_${i}` + ); await runWorkflowHandlerWithEvents( `const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")]; async function workflow() { @@ -863,10 +890,25 @@ describe('workflowEntrypoint replay guards', () => { replayDivergence: { eventId: 'different-event', count: REPLAY_DIVERGENCE_MAX_RETRIES, + eventIds: priorEventIds, }, } ); + // The terminal error lists the whole chain so a reader can tell whether + // every divergence hit one event or wandered, and the terminal console + // line carries that message and the same pass context as the WARN. + const errorOut = error.mock.calls + .map((c) => String(c[0])) + .find((line) => line.includes('Error while running workflow')); + expect(errorOut).toBeDefined(); + expect(errorOut).toContain('code CORRUPTED_EVENT_LOG'); + expect(errorOut).toContain( + `divergent event ids: ${[...priorEventIds, slotToEventId(1)].join(', ')}` + ); + expect(errorOut).toContain('isRecoveryReplay true'); + expect(errorOut).toContain('loopIteration 1'); + const failedIndex = terminalAttemptEvents.findIndex( (event) => (event as { eventType?: string }).eventType === 'run_failed' ); @@ -989,7 +1031,11 @@ describe('workflowEntrypoint replay guards', () => { ); expect(queueCalls.map((c) => c.message)).toContainEqual( expect.objectContaining({ - replayDivergence: { eventId: slotToEventId(1), count: 1 }, + replayDivergence: { + eventId: slotToEventId(1), + count: 1, + eventIds: [slotToEventId(1)], + }, }) ); }); diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index d9699fad35..53ab1871b5 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -4832,25 +4832,71 @@ export function workflowEntrypoint( } let replayDivergenceCountForFailure: number | undefined; + // Populated for a divergence so the terminal log below + // carries the same context the WARN does. + let divergenceFields: Record = {}; if (ReplayDivergenceError.is(err)) { const divergenceCount = (replayDivergence?.count ?? 0) + 1; const maxRecoveryReplays = getReplayDivergenceMaxRetries(); + // Every divergence in this recovery chain, oldest + // first, bounded by the chain's own length: a + // recovery replay is queued at most + // `maxRecoveryReplays` times, so the terminal one + // reads `maxRecoveryReplays + 1` ids. A producer that + // predates the field, or one whose value failed to + // parse, contributes nothing and the history restarts + // at the prior message's `eventId`. + const divergenceEventIds = [ + ...(replayDivergence?.eventIds ?? + (replayDivergence + ? [replayDivergence.eventId] + : [])), + err.eventId, + ].slice(-(maxRecoveryReplays + 1)); + // Which pass of which invocation diverged, and what + // that pass was working from. All of it is already + // in scope; none of it is another round trip. The + // point is that a divergence on a cold replay of a + // full log and one on a retained-VM resume of a + // delta are different bugs, and the message alone + // does not say which this was. + divergenceFields = { + errorCode: RUN_ERROR_CODES.REPLAY_DIVERGENCE, + divergenceEventId: err.eventId, + priorDivergenceEventId: replayDivergence?.eventId, + divergenceEventIds, + divergenceCount, + deliveryAttempt: metadata.attempt, + maxRecoveryReplays, + loopIteration, + servedByRetainedSession, + eventLogState: eventLog.type, + eventLogLength: + eventLog.type === 'loadAll' + ? undefined + : eventLog.events.length, + eventLogLastEventId: + eventLog.type === 'loadAll' + ? undefined + : eventLog.events.at(-1)?.eventId, + cursor: + eventLog.type === 'ready' + ? (eventLog.cursor ?? undefined) + : eventLog.type === 'loadAfter' + ? eventLog.cursor + : undefined, + hasHookInput: hookInput !== undefined, + hasWaitContinuation: waitContinuation !== undefined, + isRecoveryReplay: replayDivergence !== undefined, + setupSource: resumeTracking?.setupSource, + }; if (divergenceCount <= maxRecoveryReplays) { runLogger.warn( 'Workflow replay diverged; queueing a recovery replay before declaring the event log corrupted', - { - errorCode: RUN_ERROR_CODES.REPLAY_DIVERGENCE, - divergenceEventId: err.eventId, - priorDivergenceEventId: - replayDivergence?.eventId, - divergenceCount, - deliveryAttempt: metadata.attempt, - maxRecoveryReplays, - errorMessage: err.message, - } + { ...divergenceFields, errorMessage: err.message } ); await queueMessage( world, @@ -4862,6 +4908,7 @@ export function workflowEntrypoint( replayDivergence: { eventId: err.eventId, count: divergenceCount, + eventIds: divergenceEventIds, }, } ); @@ -4870,7 +4917,7 @@ export function workflowEntrypoint( replayDivergenceCountForFailure = divergenceCount; terminalError = new CorruptedEventLogError( - `Workflow replay diverged ${divergenceCount} times after ${maxRecoveryReplays} recovery replays; latest divergent event was ${err.eventId}. Last divergence: ${err.message}`, + `Workflow replay diverged ${divergenceCount} times after ${maxRecoveryReplays} recovery replays; latest divergent event was ${err.eventId}; divergent event ids: ${divergenceEventIds.join(', ')}. Last divergence: ${err.message}`, { cause: err } ); } else if (replayStart > 0) { @@ -4907,9 +4954,17 @@ export function workflowEntrypoint( const errorCode = classifyRunError(terminalError); runtimeLogger.error('Error while running workflow', { + // Divergence context first so the classified code + // and name of the terminal error win. + ...divergenceFields, workflowRunId: runId, errorCode, errorName, + // The console renderer drops `errorStack` on the + // assumption that the message body carries it, and + // this message has no body, so without this row the + // terminal error's text never reaches the console. + errorMessage, errorStack, }); diff --git a/packages/core/src/test-support/orchestrator-context.ts b/packages/core/src/test-support/orchestrator-context.ts index 0b68b46bb3..d3ef900202 100644 --- a/packages/core/src/test-support/orchestrator-context.ts +++ b/packages/core/src/test-support/orchestrator-context.ts @@ -6,6 +6,7 @@ import { monotonicFactory } from 'ulid'; import { vi } from 'vitest'; import { EventsConsumer } from '../events-consumer.js'; import type { WorkflowOrchestratorContext } from '../private.js'; +import { describeDivergenceContext } from '../replay-divergence.js'; import { ReplayPayloadCache } from '../replay-payload-cache.js'; import { createContext } from '../vm/index.js'; @@ -44,9 +45,13 @@ export function setupWorkflowContext( // Fake context: no deliveries are modeled, so the gate is a no-op here. isDeliveryIdle: () => true, onUnconsumedEvent: (event) => { - ctxRef.current?.onWorkflowError( + const current = ctxRef.current; + if (!current) return; + // Same detail the production path appends (see `onUnconsumedEvent` in + // workflow.ts), so a test can assert on what a user would read. + 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.` + `Unconsumed event in event log: eventType=${event.eventType}, correlationId=${event.correlationId}, eventId=${event.eventId}. This indicates a corrupted or invalid event log. ${describeDivergenceContext(event, current.invocationsQueue, current.eventsConsumer)}` ) ); }, diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index c7e3f6ccc0..faecee4c46 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -22,6 +22,7 @@ import { ENOTSUP, WorkflowSuspension } from './global.js'; import { runtimeLogger } from './logger.js'; import type { WorkflowOrchestratorContext } from './private.js'; import { isDeliveryIdle } from './private.js'; +import { describeDivergenceContext } from './replay-divergence.js'; import { ReplayPayloadCache } from './replay-payload-cache.js'; import { getPortLazy } from './runtime/get-port-lazy.js'; import { runIdCreatedAt } from './runtime/run-id-time.js'; @@ -492,9 +493,13 @@ async function createWorkflowSessionInner( } }, onUnconsumedEvent: (event) => { + // `workflowContext` is assigned below, before any event can be offered, + // so it is always populated by the time this fires. The appended detail + // names the pending invocation that holds this event's ordinal (the + // usual reason nobody can consume it) and where the walk stands. onWorkflowError( new ReplayDivergenceError( - `Replay could not consume event: eventType=${event.eventType}, correlationId=${event.correlationId}, eventId=${event.eventId}.`, + `Replay could not consume event: eventType=${event.eventType}, correlationId=${event.correlationId}, eventId=${event.eventId}. ${describeDivergenceContext(event, workflowContext.invocationsQueue, eventsConsumer)}`, { eventId: event.eventId } ) ); @@ -1228,7 +1233,7 @@ async function createWorkflowSessionInner( if (stranded) { return failWorkflow( new ReplayDivergenceError( - `Replay finished without consuming event: eventType=${stranded.eventType}, correlationId=${stranded.correlationId}, eventId=${stranded.eventId}.`, + `Replay finished without consuming event: eventType=${stranded.eventType}, correlationId=${stranded.correlationId}, eventId=${stranded.eventId}. ${describeDivergenceContext(stranded, workflowContext.invocationsQueue, eventsConsumer)}`, { eventId: stranded.eventId } ) ); diff --git a/packages/world/src/queue.test.ts b/packages/world/src/queue.test.ts index 3997465561..795e3f67ad 100644 --- a/packages/world/src/queue.test.ts +++ b/packages/world/src/queue.test.ts @@ -198,6 +198,58 @@ describe('QueuePayloadSchema', () => { }).success ).toBe(false); }); + + describe('replayDivergence', () => { + it('round-trips the divergence history', () => { + expect( + QueuePayloadSchema.parse({ + runId: 'wrun_01ABC', + replayDivergence: { + eventId: 'evnt_2', + count: 2, + eventIds: ['evnt_1', 'evnt_2'], + }, + }) + ).toEqual({ + runId: 'wrun_01ABC', + replayDivergence: { + eventId: 'evnt_2', + count: 2, + eventIds: ['evnt_1', 'evnt_2'], + }, + }); + }); + + // A producer that predates the history sends only the latest position. + it('accepts a message without a history', () => { + expect( + QueuePayloadSchema.parse({ + runId: 'wrun_01ABC', + replayDivergence: { eventId: 'evnt_1', count: 1 }, + }) + ).toEqual({ + runId: 'wrun_01ABC', + replayDivergence: { eventId: 'evnt_1', count: 1 }, + }); + }); + + // The history is diagnostic only. A malformed value must not fail the + // parse, which would fail every delivery of the recovery message. + it('drops a malformed history instead of failing the parse', () => { + const parsed = QueuePayloadSchema.parse({ + runId: 'wrun_01ABC', + replayDivergence: { + eventId: 'evnt_1', + count: 1, + eventIds: [1, { not: 'a string' }], + }, + }); + expect(parsed).toEqual({ + runId: 'wrun_01ABC', + replayDivergence: { eventId: 'evnt_1', count: 1 }, + }); + }); + }); }); describe('RunInputSchema environment', () => { diff --git a/packages/world/src/queue.ts b/packages/world/src/queue.ts index 1f795f8dae..d89bde853a 100644 --- a/packages/world/src/queue.ts +++ b/packages/world/src/queue.ts @@ -270,11 +270,23 @@ export const WorkflowInvokePayloadSchema = z.object({ runId: z.string(), traceCarrier: TraceCarrierSchema.optional(), requestedAt: z.coerce.date().optional(), - /** Consecutive replay divergences in this recovery chain and latest position. */ + /** + * Consecutive replay divergences in this recovery chain and latest position. + * + * `eventIds` is every divergent event id in the chain, oldest first, so the + * terminal failure can say whether each recovery replay diverged at the same + * event or wandered. The producer bounds it to the recovery budget plus one. + * `.catch(undefined)` on the field for the reason `hookResumeTiming` has it: + * a malformed history must degrade to "no history" rather than fail the + * parse of every delivery of this message. A consumer that predates the + * field ignores it; a producer that predates it sends none, and the consumer + * restarts the history from `eventId`. + */ replayDivergence: z .object({ eventId: z.string(), count: z.number().int().positive(), + eventIds: z.array(z.string()).optional().catch(undefined), }) .optional(), /**