Instrument workflow replay scaling

.changeset/tidy-spans-measure.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Add detailed workflow replay phase spans and aggregate event-consumption and step-hydration diagnostics.

.github/workflows/benchmarks.yml

@@ -157,6 +157,10 @@ jobs:
           WORKFLOW_VERCEL_TEAM: "team_nO2mCG4W8IxPIeKoSsqwAxxB"
           WORKFLOW_VERCEL_PROJECT: ${{ matrix.target.project-id }}
           WORKFLOW_VERCEL_PROJECT_SLUG: ${{ matrix.target.project-slug }}
+          # Keep PR benchmarks entirely on preview infrastructure. This is the
+          # same protected workflow-server preview used by the Vercel E2E lane;
+          # main remains pointed at the public production server.
+          VERCEL_WORKFLOW_SERVER_URL: ${{ github.ref != 'refs/heads/main' && secrets.VERCEL_WORKFLOW_SERVER_URL || '' }}
           # Report the PR head SHA (not the synthetic merge commit) in results
           GITHUB_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
           # Trusted-sources OIDC token is minted on demand by

packages/core/bench/workflow-replay.bench.ts

@@ -0,0 +1,167 @@
+import type { Event, WorkflowRun } from '@workflow/world';
+import { monotonicFactory } from 'ulid';
+import { bench, describe } from 'vitest';
+import { WorkflowSuspension } from '../src/global.js';
+import {
+  dehydrateStepReturnValue,
+  dehydrateWorkflowArguments,
+} from '../src/serialization.js';
+import {
+  createStepHydrationCache,
+  type StepHydrationCache,
+} from '../src/step-hydration-cache.js';
+import { createContext } from '../src/vm/index.js';
+import { runWorkflow } from '../src/workflow.js';
+
+/**
+ * Isolated replay benchmark: no World, queue, workflow-server, or storage.
+ * Run from the repository root with:
+ * `pnpm vitest bench packages/core/bench/workflow-replay.bench.ts --run`
+ */
+
+type ResultMode = 'object-miss' | 'primitive-miss' | 'primitive-hit';
+
+interface ReplayFixture {
+  workflowRun: WorkflowRun;
+  events: Event[];
+  initialCache: StepHydrationCache;
+}
+
+const workflowCode = `
+  const timedNoopStep = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("timedNoopStep");
+  async function workflow(count) {
+    for (let i = 0; i < count; i++) await timedNoopStep(i);
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);
+`;
+
+async function createReplayFixture(
+  completedSteps: number,
+  resultMode: ResultMode
+): Promise<ReplayFixture> {
+  const runId = `wrun_bench_${completedSteps}_${resultMode}`;
+  const workflowName = 'workflow';
+  const deploymentId = 'local-benchmark';
+  const startedAt = new Date('2024-01-01T00:00:00.000Z');
+  const input = await dehydrateWorkflowArguments(
+    [completedSteps + 1],
+    runId,
+    undefined
+  );
+  const workflowRun: WorkflowRun = {
+    runId,
+    workflowName,
+    deploymentId,
+    status: 'running',
+    input,
+    attributes: {},
+    createdAt: startedAt,
+    updatedAt: startedAt,
+    startedAt,
+  };
+  const vm = createContext({
+    seed: `${runId}:${workflowName}:${deploymentId}`,
+    fixedTimestamp: +startedAt,
+  });
+  const ulid = monotonicFactory(() => vm.globalThis.Math.random());
+  const initialCache = createStepHydrationCache();
+  const events: Event[] = [
+    {
+      eventId: 'event-run-created',
+      runId,
+      eventType: 'run_created',
+      eventData: { deploymentId, workflowName, input },
+      createdAt: startedAt,
+    },
+    {
+      eventId: 'event-run-started',
+      runId,
+      eventType: 'run_started',
+      createdAt: startedAt,
+    },
+  ];
+
+  for (let i = 0; i < completedSteps; i++) {
+    const correlationId = `step_${ulid(+startedAt)}`;
+    const completedEventId = `event-${i}-completed`;
+    const result = resultMode === 'object-miss' ? { start: i, end: i } : i;
+    if (resultMode === 'primitive-hit') {
+      initialCache.set(completedEventId, result);
+    }
+    events.push(
+      {
+        eventId: `event-${i}-created`,
+        runId,
+        eventType: 'step_created',
+        correlationId,
+        eventData: { stepName: 'timedNoopStep', input: undefined },
+        createdAt: startedAt,
+      },
+      {
+        eventId: `event-${i}-started`,
+        runId,
+        eventType: 'step_started',
+        correlationId,
+        eventData: { stepName: 'timedNoopStep' },
+        createdAt: startedAt,
+      },
+      {
+        eventId: completedEventId,
+        runId,
+        eventType: 'step_completed',
+        correlationId,
+        eventData: {
+          stepName: 'timedNoopStep',
+          result: await dehydrateStepReturnValue(result, runId, undefined),
+        },
+        createdAt: startedAt,
+      }
+    );
+  }
+
+  return { workflowRun, events, initialCache };
+}
+
+process.env.VERCEL_URL = 'local-replay-benchmark.test';
+
+const fixtures = new Map<string, ReplayFixture>();
+for (const completedSteps of [20, 120, 1020]) {
+  for (const resultMode of [
+    'object-miss',
+    'primitive-miss',
+    'primitive-hit',
+  ] as const) {
+    fixtures.set(
+      `${completedSteps}:${resultMode}`,
+      await createReplayFixture(completedSteps, resultMode)
+    );
+  }
+}
+
+describe('runWorkflow sequential replay scaling', () => {
+  for (const completedSteps of [20, 120, 1020]) {
+    for (const resultMode of [
+      'object-miss',
+      'primitive-miss',
+      'primitive-hit',
+    ] as const) {
+      bench(`${completedSteps} completed steps / ${resultMode}`, async () => {
+        const fixture = fixtures.get(`${completedSteps}:${resultMode}`);
+        if (!fixture) throw new Error('Missing replay benchmark fixture');
+        const cache = new Map(fixture.initialCache);
+        try {
+          await runWorkflow(
+            workflowCode,
+            fixture.workflowRun,
+            fixture.events,
+            undefined,
+            cache
+          );
+          throw new Error('Expected workflow replay to suspend');
+        } catch (error) {
+          if (!WorkflowSuspension.is(error)) throw error;
+        }
+      });
+    }
+  }
+});

packages/core/src/events-consumer.ts

@@ -1,5 +1,6 @@
 import { type Event, envNumber } from '@workflow/world';
 import { eventsLogger } from './logger.js';
+import type { WorkflowReplayMetrics } from './replay-telemetry.js';
 
 /**
  * Delay before firing the deferred unconsumed-event check after the promise
@@ -65,6 +66,8 @@ export interface EventsConsumerOptions {
    * deserialization delays the resolve() that triggers the next subscribe().
    */
   getPromiseQueue: () => Promise<void>;
+  /** Aggregate diagnostics for this replay; omitted when tracing is inactive. */
+  replayMetrics?: WorkflowReplayMetrics;
 }
 
 export class EventsConsumer {
@@ -76,14 +79,17 @@ export class EventsConsumer {
   private getPromiseQueue: () => Promise<void>;
   private pendingUnconsumedCheck: Promise<void> | null = null;
   private pendingUnconsumedTimeout: ReturnType<typeof setTimeout> | null = null;
+  private pendingUnconsumedStartedAt: number | null = null;
   private unconsumedCheckVersion = 0;
+  private replayMetrics?: WorkflowReplayMetrics;
 
   constructor(events: Event[], options: EventsConsumerOptions) {
     this.events = events;
     this.eventIndex = 0;
     this.onConsumedEvent = options.onConsumedEvent;
     this.onUnconsumedEvent = options.onUnconsumedEvent;
     this.getPromiseQueue = options.getPromiseQueue;
+    this.replayMetrics = options.replayMetrics;
   }
 
   /**
@@ -101,6 +107,12 @@ export class EventsConsumer {
     // Incrementing the version causes any in-flight promise chain check to no-op.
     // Also clear the pending setTimeout if it hasn't fired yet.
     if (this.pendingUnconsumedCheck !== null) {
+      if (this.replayMetrics && this.pendingUnconsumedStartedAt !== null) {
+        this.replayMetrics.subscriptionWaitCount++;
+        this.replayMetrics.subscriptionWaitMs +=
+          performance.now() - this.pendingUnconsumedStartedAt;
+      }
+      this.pendingUnconsumedStartedAt = null;
       this.unconsumedCheckVersion++;
       this.pendingUnconsumedCheck = null;
       if (this.pendingUnconsumedTimeout !== null) {
@@ -160,8 +172,23 @@ export class EventsConsumer {
    * sentinel).
    */
   private consumeOne(currentEvent: Event | null): boolean {
+    const metrics = this.replayMetrics;
+    if (!metrics) return this.consumeOneCallbacks(currentEvent);
+    const startedAt = performance.now();
+    try {
+      return this.consumeOneCallbacks(currentEvent, metrics);
+    } finally {
+      metrics.eventConsumeMs += performance.now() - startedAt;
+    }
+  }
+
+  private consumeOneCallbacks(
+    currentEvent: Event | null,
+    metrics?: WorkflowReplayMetrics
+  ): boolean {
     for (let i = 0; i < this.callbacks.length; i++) {
       const callback = this.callbacks[i];
+      if (metrics) metrics.eventCallbackInvocations++;
       let handled = EventConsumerResult.NotConsumed;
       try {
         handled = callback(currentEvent);
@@ -175,6 +202,7 @@ export class EventsConsumer {
         continue;
       }
       if (currentEvent !== null) {
+        if (metrics) metrics.eventsConsumed++;
         this.notifyConsumedEvent(currentEvent);
       }
       // consumer handled this event, so increase the event index
@@ -199,6 +227,9 @@ export class EventsConsumer {
     // resolve() → user code → subscribe()) completes first. If the event
     // is still unconsumed after the queue drains, it's truly orphaned.
     if (currentEvent !== null) {
+      if (this.replayMetrics) {
+        this.pendingUnconsumedStartedAt = performance.now();
+      }
       const checkVersion = ++this.unconsumedCheckVersion;
       this.pendingUnconsumedCheck = this.getPromiseQueue()
         .then(
@@ -222,6 +253,7 @@ export class EventsConsumer {
             this.pendingUnconsumedTimeout = null;
             if (this.unconsumedCheckVersion === checkVersion) {
               this.pendingUnconsumedCheck = null;
+              this.pendingUnconsumedStartedAt = null;
               this.onUnconsumedEvent(currentEvent);
             }
           }, getDeferredCheckDelayMs());

packages/core/src/private.ts

@@ -6,6 +6,7 @@ import { withResolvers } from '@workflow/utils';
 import type { CryptoKey } from './encryption.js';
 import type { EventsConsumer } from './events-consumer.js';
 import type { QueueItem } from './global.js';
+import type { WorkflowReplayMetrics } from './replay-telemetry.js';
 import type { Serializable } from './schemas.js';
 import type { StepHydrationCache } from './step-hydration-cache.js';
 
@@ -207,6 +208,11 @@ export interface WorkflowOrchestratorContext {
    * gracefully to re-hydrating every replay — identical to previous behavior.
    */
   stepHydrationCache?: StepHydrationCache;
+  /**
+   * Low-overhead aggregate replay diagnostics for the active workflow.run
+   * span. Undefined when tracing is unavailable or the span is not recording.
+   */
+  replayMetrics?: WorkflowReplayMetrics;
 }
 
 /** The kind of branch-deciding delivery a barrier represents. */