[core] Log pending consumers in divergence diagnostics

.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

.changeset/divergence-diagnostics-world.md

@@ -0,0 +1,2 @@
+---
+---

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 <divergenceCount> times after <maxRecoveryReplays> recovery replays; latest divergent event was <eventId>. Last divergence: <details>
+Workflow replay diverged <divergenceCount> times after <maxRecoveryReplays> recovery replays; latest divergent event was <eventId>; divergent event ids: <eventId>, <eventId>, ... Last divergence: <details>
 ```
 
+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.
+
+`<details>` 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_<id>, eventId=<eventId>. pending at this id: step <stepName> (step_<id>). consumer: index=<n>, length=<n>, parked=<n>, lastConsumed=<eventId>
+```
+
+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 `<stepName>`.
+
 For an unreadable stored payload:
 
 ```text

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<void> | null = null;
   private pendingUnconsumedTimeout: ReturnType<typeof setTimeout> | 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

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);
+  });
 });

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<string, unknown> | 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<string>();
   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)}`);

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",
           ],
         ]
       `);

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<string, unknown>
+): 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<unknown>[] = [];
+  return await dehydrateStepReturnValue(value, 'wrun_test', undefined, ops);
+}
+
+describe('findPendingItemAtOrdinal', () => {
+  const queue = new Map<string, QueueItem>([
+    [
+      `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.');
+  });
+});