fix(otel-v2): provide id generator

Author: chris-olszewski

packages/interceptors-opentelemetry-v2/src/__tests__/history_files/otel_v2_smorgasbord.json

@@ -1,7 +1,6 @@
 /**
  * Manual tests to inspect tracing output
  */
-import * as http from 'http';
 import * as http2 from 'http2';
 import * as path from 'path';
 import * as otelApi from '@opentelemetry/api';
@@ -14,7 +13,6 @@ import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';
 import test from 'ava';
 import { v4 as uuid4 } from 'uuid';
 import { WorkflowClient, WithStartWorkflowOperation, WorkflowClientInterceptor, Client } from '@temporalio/client';
-import { TestWorkflowEnvironment } from '@temporalio/testing';
 import {
   ActivityInboundCallsInterceptor,
   ActivityOutboundCallsInterceptor,
@@ -147,15 +145,15 @@ if (RUN_INTEGRATION_TESTS) {
       const worker = await Worker.create({
         workflowsPath: require.resolve('./workflows'),
         activities,
-        taskQueue: 'test-otel',
+        taskQueue: 'test-otel-v2',
         plugins: [plugin],
       });
 
       const client = new Client({
         plugins: [plugin],
       });
       await worker.runUntil(
-        client.workflow.execute(workflows.smorgasbord, { taskQueue: 'test-otel', workflowId: uuid4() })
+        client.workflow.execute(workflows.smorgasbord, { taskQueue: 'test-otel-v2', workflowId: uuid4() })
       );
       await otel.shutdown();
       const originalSpan = spans.find(({ name }) => name === `${SpanName.WORKFLOW_START}${SPAN_DELIMITER}smorgasbord`);
@@ -301,7 +299,7 @@ if (RUN_INTEGRATION_TESTS) {
         workflowsPath: require.resolve('./workflows'),
         activities,
         enableSDKTracing: true,
-        taskQueue: 'test-otel',
+        taskQueue: 'test-otel-v2',
         interceptors: {
           workflowModules: [require.resolve('./workflows/otel-interceptors')],
           activity: [(ctx) => ({ inbound: new OpenTelemetryActivityInboundInterceptor(ctx) })],
@@ -312,7 +310,7 @@ if (RUN_INTEGRATION_TESTS) {
       const client = new WorkflowClient({
         interceptors: [new OpenTelemetryWorkflowClientInterceptor()],
       });
-      await worker.runUntil(client.execute(workflows.smorgasbord, { taskQueue: 'test-otel', workflowId: uuid4() }));
+      await worker.runUntil(client.execute(workflows.smorgasbord, { taskQueue: 'test-otel-v2', workflowId: uuid4() }));
       // Allow some time to ensure spans are flushed out to collector
       await new Promise<void>((resolve) => setTimeout(resolve, 5000));
       t.pass();
@@ -327,7 +325,7 @@ if (RUN_INTEGRATION_TESTS) {
     t.falsy((globalThis as any).window);
   });
 
-  test('instrumentation: Error status includes message and records exception', async (t) => {
+  test.serial('instrumentation: Error status includes message and records exception', async (t) => {
     const memoryExporter = new InMemorySpanExporter();
     const provider = new BasicTracerProvider({
       spanProcessors: [new SimpleSpanProcessor(memoryExporter)],
@@ -360,7 +358,7 @@ if (RUN_INTEGRATION_TESTS) {
     t.is(exceptionEvents.length, 1);
   });
 
-  test('Otel workflow omits ApplicationError with BENIGN category', async (t) => {
+  test.serial('Otel workflow omits ApplicationError with BENIGN category', async (t) => {
     const memoryExporter = new InMemorySpanExporter();
     const provider = new BasicTracerProvider({
       spanProcessors: [new SimpleSpanProcessor(memoryExporter)],
@@ -370,7 +368,7 @@ if (RUN_INTEGRATION_TESTS) {
     const worker = await Worker.create({
       workflowsPath: require.resolve('./workflows'),
       activities,
-      taskQueue: 'test-otel-benign-err',
+      taskQueue: 'test-otel-v2-benign-err',
       interceptors: {
         activity: [
           (ctx) => {
@@ -384,7 +382,7 @@ if (RUN_INTEGRATION_TESTS) {
 
     await worker.runUntil(
       client.execute(workflows.throwMaybeBenignErr, {
-        taskQueue: 'test-otel-benign-err',
+        taskQueue: 'test-otel-v2-benign-err',
         workflowId: uuid4(),
         retry: { maximumAttempts: 3 },
       })
@@ -399,7 +397,7 @@ if (RUN_INTEGRATION_TESTS) {
     t.is(spans[2]!.status.code, SpanStatusCode.OK);
   });
 
-  test('executeUpdateWithStart works correctly with OTEL interceptors', async (t) => {
+  test.serial('executeUpdateWithStart works correctly with OTEL interceptors', async (t) => {
     const staticResource = opentelemetry.resources.resourceFromAttributes({
       [ATTR_SERVICE_NAME]: 'ts-test-otel-worker',
     });
@@ -420,15 +418,15 @@ if (RUN_INTEGRATION_TESTS) {
         plugins: [plugin],
       }),
       activities,
-      taskQueue: 'test-otel-update-start',
+      taskQueue: 'test-otel-v2-update-start',
       plugins: [plugin],
     });
 
     const client = new WorkflowClient();
 
     const startWorkflowOperation = new WithStartWorkflowOperation(workflows.updateStartOtel, {
       workflowId: uuid4(),
-      taskQueue: 'test-otel-update-start',
+      taskQueue: 'test-otel-v2-update-start',
       workflowIdConflictPolicy: 'FAIL',
     });
 
@@ -450,8 +448,8 @@ if (RUN_INTEGRATION_TESTS) {
 
   // These tests verify makeWorkflowExporter's handling of async resource attributes:
   // https://github.com/temporalio/sdk-typescript/issues/1779
-  test(`makeWorkflowExporter with SpanProcessor does await async resource attributes`, async (t) => {
-    const taskQueue = `test-otel-async-processor`;
+  test.serial(`makeWorkflowExporter with SpanProcessor does await async resource attributes`, async (t) => {
+    const taskQueue = `test-otel-v2-async-processor`;
     const serviceName = `ts-test-otel-async-attributes`;
 
     // In OTel v2, async resource attributes are created by passing Promise
@@ -523,7 +521,7 @@ if (RUN_INTEGRATION_TESTS) {
       const spans = Array<opentelemetry.tracing.ReadableSpan>();
 
       const staticResource = opentelemetry.resources.resourceFromAttributes({
-        [ATTR_SERVICE_NAME]: 'ts-test-otel-prebundled-worker',
+        [ATTR_SERVICE_NAME]: 'ts-test-otel-v2-prebundled-worker',
       });
       const traceExporter: opentelemetry.tracing.SpanExporter = {
         export(spans_, resultCallback) {
@@ -561,7 +559,7 @@ if (RUN_INTEGRATION_TESTS) {
       const worker = await Worker.create({
         workflowBundle,
         activities,
-        taskQueue: 'test-otel-prebundled',
+        taskQueue: 'test-otel-v2-prebundled',
         plugins: [plugin],
       });
 
@@ -570,7 +568,7 @@ if (RUN_INTEGRATION_TESTS) {
         plugins: [plugin],
       });
       await worker.runUntil(
-        client.workflow.execute(workflows.smorgasbord, { taskQueue: 'test-otel-prebundled', workflowId: uuid4() })
+        client.workflow.execute(workflows.smorgasbord, { taskQueue: 'test-otel-v2-prebundled', workflowId: uuid4() })
       );
       await provider.shutdown();
 
@@ -649,7 +647,7 @@ if (RUN_INTEGRATION_TESTS) {
       const worker = await Worker.create({
         workflowsPath: require.resolve('./workflows'),
         activities,
-        taskQueue: 'test-otel-tracestate',
+        taskQueue: 'test-otel-v2-tracestate',
         interceptors: {
           workflowModules: [require.resolve('./workflows/otel-interceptors')],
         },
@@ -682,7 +680,7 @@ if (RUN_INTEGRATION_TESTS) {
         // Execute the workflow within this context so the traceState is propagated
         await otelApi.context.with(contextWithTraceState, async () => {
           await client.execute(workflows.successString, {
-            taskQueue: 'test-otel-tracestate',
+            taskQueue: 'test-otel-v2-tracestate',
             workflowId: uuid4(),
           });
         });
@@ -707,6 +705,61 @@ if (RUN_INTEGRATION_TESTS) {
       t.is(traceState!.get('vendor2'), 'value2');
     }
   });
+
+  test.serial('OTel span creation does not affect workflow PRNG sequence', async (t) => {
+    const plugin = new OpenTelemetryPlugin({
+      resource: opentelemetry.resources.resourceFromAttributes({
+        [ATTR_SERVICE_NAME]: 'ts-test-prng-isolation',
+      }),
+      spanProcessor: new SimpleSpanProcessor({
+        export(_spans, resultCallback) {
+          resultCallback({ code: ExportResultCode.SUCCESS });
+        },
+        async shutdown() {},
+      }),
+    });
+
+    const otelBundle = await createOtelTestWorkflowBundle({
+      workflowsPath: require.resolve('./workflows'),
+      plugins: [plugin],
+    });
+
+    // Run the workflow with OTel interceptors and capture uuid4() values
+    const taskQueue1 = 'test-otel-v2-prng-isolation-1';
+    const workerWithOtel = await Worker.create({
+      workflowBundle: otelBundle,
+      activities,
+      taskQueue: taskQueue1,
+      plugins: [plugin],
+    });
+    const client = new WorkflowClient();
+    const workflowId = uuid4();
+    const idsWithOtel = await workerWithOtel.runUntil(
+      client.execute(workflows.prngIsolation, { taskQueue: taskQueue1, workflowId })
+    );
+
+    t.is(idsWithOtel.length, 10, 'should produce 10 uuid4 values');
+    t.is(new Set(idsWithOtel).size, 10, 'all uuid4 values should be unique');
+
+    // Replay the same workflow — if the DeterministicIdGenerator properly
+    // isolates OTel's PRNG from the workflow's Math.random(), uuid4()
+    // produces the same sequence on replay.
+    const handle = client.getHandle(workflowId);
+    const history = await handle.fetchHistory();
+    await t.notThrowsAsync(
+      Worker.runReplayHistory(
+        {
+          workflowBundle: otelBundle,
+          interceptors: {
+            workflowModules: [require.resolve('./workflows/otel-interceptors')],
+          },
+        },
+        history
+      )
+    );
+
+    t.log('uuid4() values with OTel:', idsWithOtel);
+  });
 }
 
 test('Can replay otel history from 1.11.3', async (t) => {
@@ -755,60 +808,8 @@ test('Can replay otel history from 1.13.1', async (t) => {
   });
 });
 
-// These histories were recorded with OTel v1 libraries. The v2 OTel library's
-// different module structure changes the webpack module initialization order in
-// the workflow sandbox, which shifts the deterministic Math.random() state and
-// produces different uuid4() values for child workflow IDs.
-// Workflows without auto-generated IDs (otel_1_11_3, otel_1_13_1) replay fine.
-test.skip('Can replay smorgasbord from 1.13.1', async (t) => {
-  // This test will trigger NDE if yield points for `scheduleActivity` and `startChildWorkflowExecution` are not inserted
-  const hist = await loadHistory('otel_smorgasbord_1_13_1.json');
-  await t.notThrowsAsync(async () => {
-    await Worker.runReplayHistory(
-      {
-        workflowBundle: await createOtelTestWorkflowBundle({
-          workflowsPath: require.resolve('./workflows'),
-          workflowInterceptorModules: [require.resolve('./workflows/otel-interceptors')],
-        }),
-        interceptors: {
-          workflowModules: [require.resolve('./workflows/otel-interceptors')],
-          activity: [
-            (ctx) => ({
-              inbound: new OpenTelemetryActivityInboundInterceptor(ctx),
-            }),
-          ],
-        },
-      },
-      hist
-    );
-  });
-});
-
-test.skip('Can replay signal workflow from 1.13.1', async (t) => {
-  const hist = await loadHistory('signal_workflow_1_13_1.json');
-  await t.notThrowsAsync(async () => {
-    await Worker.runReplayHistory(
-      {
-        workflowBundle: await createOtelTestWorkflowBundle({
-          workflowsPath: require.resolve('./workflows/signal-workflow'),
-          workflowInterceptorModules: [require.resolve('./workflows/otel-interceptors')],
-        }),
-        interceptors: {
-          workflowModules: [require.resolve('./workflows/otel-interceptors')],
-          activity: [
-            (ctx) => ({
-              inbound: new OpenTelemetryActivityInboundInterceptor(ctx),
-            }),
-          ],
-        },
-      },
-      hist
-    );
-  });
-});
-
-test.skip('Can replay smorgasbord from 1.13.2', async (t) => {
-  const hist = await loadHistory('otel_smorgasbord_1_13_2.json');
+test('Can replay smorgasbord history from 1.15.0', async (t) => {
+  const hist = await loadHistory('otel_v2_smorgasbord.json');
   await t.notThrowsAsync(async () => {
     await Worker.runReplayHistory(
       {

packages/interceptors-opentelemetry-v2/src/__tests__/test-otel.ts

@@ -4,3 +4,4 @@ export * from './smorgasbord';
 export * from './success-string';
 export * from './throw-maybe-benign';
 export * from './update-start-otel';
+export * from './prng-isolation';

packages/interceptors-opentelemetry-v2/src/__tests__/workflows/index.ts

@@ -0,0 +1,13 @@
+/**
+ * Workflow that captures uuid4() values to verify OTel span creation
+ * does not affect the workflow's Math.random() PRNG sequence.
+ */
+import { uuid4 } from '@temporalio/workflow';
+
+export async function prngIsolation(): Promise<string[]> {
+  const ids: string[] = [];
+  for (let i = 0; i < 10; i++) {
+    ids.push(uuid4());
+  }
+  return ids;
+}

packages/interceptors-opentelemetry-v2/src/__tests__/workflows/prng-isolation.ts

@@ -0,0 +1,54 @@
+import type { IdGenerator } from '@opentelemetry/sdk-trace-base';
+import { alea, type RNG } from './workflow-imports';
+
+// Seed the OTel ID generator's PRNG from the workflow's PRNG, then advance
+// the workflow PRNG once so the two sequences diverge immediately.
+let otelRandom: RNG | undefined;
+
+function getOtelRandom(): RNG {
+  if (otelRandom === undefined) {
+    // Use Math.random() (which is the workflow's deterministic PRNG) to
+    // produce a seed for a SEPARATE alea instance. After this, the OTel
+    // PRNG and the workflow PRNG are independent sequences.
+    const seed = [Math.random(), Math.random(), Math.random(), Math.random()];
+    otelRandom = alea(seed.map((v) => (v * 0x100000000) >>> 0));
+  }
+  return otelRandom;
+}
+
+function randomHex(length: number): string {
+  const rng = getOtelRandom();
+  const chars: string[] = [];
+  for (let i = 0; i < length; i++) {
+    const nibble = (rng() * 16) >>> 0;
+    chars.push(nibble.toString(16));
+  }
+  return chars.join('');
+}
+
+/**
+ * Reset the OTel PRNG so the next workflow gets a freshly seeded generator.
+ * Must be called during workflow dispose since modules are shared across
+ * workflow instances in the reusable VM.
+ */
+export function resetIdGenerator(): void {
+  otelRandom = undefined;
+}
+
+/**
+ * Generates span and trace IDs using a PRNG that is separate from the
+ * workflow's Math.random(), preventing OTel ID generation from affecting
+ * the deterministic sequence used by uuid4() for child workflow IDs.
+ */
+export class DeterministicIdGenerator implements IdGenerator {
+  generateTraceId(): string {
+    const id = randomHex(32);
+    // Ensure non-zero per W3C Trace Context spec
+    return id === '00000000000000000000000000000000' ? '00000000000000000000000000000001' : id;
+  }
+
+  generateSpanId(): string {
+    const id = randomHex(16);
+    return id === '0000000000000000' ? '0000000000000001' : id;
+  }
+}

packages/interceptors-opentelemetry-v2/src/workflow/id-generator.ts

@@ -38,6 +38,7 @@ import {
 import { ContextManager } from './context-manager';
 import { SpanName, SPAN_DELIMITER } from './definitions';
 import { SpanExporter } from './span-exporter';
+import { DeterministicIdGenerator, resetIdGenerator } from './id-generator';
 import { workflowInfo, ContinueAsNew, getActivator, SdkFlags } from './workflow-imports';
 
 export * from './definitions';
@@ -51,6 +52,7 @@ function getTracer(): otel.Tracer {
   }
   if (tracer === undefined) {
     const provider = new tracing.BasicTracerProvider({
+      idGenerator: new DeterministicIdGenerator(),
       spanProcessors: [new tracing.SimpleSpanProcessor(new SpanExporter())],
     });
     otel.propagation.setGlobalPropagator(new W3CTraceContextPropagator());
@@ -326,6 +328,7 @@ export class OpenTelemetryInternalsInterceptor implements WorkflowInternalsInter
     if (contextManager !== undefined) {
       contextManager.disable();
     }
+    resetIdGenerator();
     next(input);
   }
 }

packages/interceptors-opentelemetry-v2/src/workflow/index.ts

@@ -7,3 +7,4 @@
 export { inWorkflowContext, proxySinks, workflowInfo, AsyncLocalStorage, ContinueAsNew } from '@temporalio/workflow';
 export { SdkFlags } from '@temporalio/workflow/lib/flags';
 export { getActivator } from '@temporalio/workflow/lib/global-attributes';
+export { alea, type RNG } from '@temporalio/workflow/lib/alea';