fix(otel): remove yield from execute interceptor

Author: chris-olszewski

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

@@ -59,12 +59,13 @@ export class OpenTelemetryInboundInterceptor implements WorkflowInboundCallsInte
     input: WorkflowExecuteInput,
     next: Next<WorkflowInboundCallsInterceptor, 'execute'>
   ): Promise<unknown> {
-    const context = await Promise.resolve(extractContextFromHeaders(input.headers));
+    const shouldInjectYield = getActivator().hasFlag(SdkFlags.OpenTelemetryInterceptorInsertYield);
+    const context = extractContextFromHeaders(input.headers);
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.WORKFLOW_EXECUTE}${SPAN_DELIMITER}${workflowInfo().workflowType}`,
       fn: () => next(input),
-      context,
+      context: shouldInjectYield ? await Promise.resolve(context) : context,
       acceptableErrors: (err) => err instanceof ContinueAsNew,
     });
   }

packages/test/src/test-flags.ts

@@ -1,7 +1,19 @@
 import test from 'ava';
-import { SdkFlags } from '@temporalio/workflow/lib/flags';
+import { SdkFlags, type SdkFlag } from '@temporalio/workflow/lib/flags';
 import type { WorkflowInfo } from '@temporalio/workflow';
 
+type Conditions = SdkFlag['alternativeConditions'];
+function composeConditions(conditions: Conditions): NonNullable<Conditions>[number] {
+  return (ctx) => {
+    for (const cond of conditions ?? []) {
+      if (cond(ctx)) {
+        return true;
+      }
+    }
+    return false;
+  };
+}
+
 test('OpenTelemetryHandleSignalInterceptorInsertYield enabled by version', (t) => {
   const cases = [
     { version: '1.0.0', expected: false },
@@ -14,15 +26,7 @@ test('OpenTelemetryHandleSignalInterceptorInsertYield enabled by version', (t) =
     { version: '1.14.0', expected: false },
   ];
   for (const { version, expected } of cases) {
-    const alternativeCondition = (ctx: { info: WorkflowInfo; sdkVersion: string | undefined }) => {
-      for (const cond of SdkFlags.OpenTelemetryHandleSignalInterceptorInsertYield.alternativeConditions!) {
-        if (cond(ctx)) {
-          return true;
-        }
-      }
-      return false;
-    };
-    const actual = alternativeCondition({
+    const actual = composeConditions(SdkFlags.OpenTelemetryHandleSignalInterceptorInsertYield.alternativeConditions)({
       info: {} as WorkflowInfo,
       sdkVersion: version,
     });
@@ -33,3 +37,23 @@ test('OpenTelemetryHandleSignalInterceptorInsertYield enabled by version', (t) =
     );
   }
 });
+
+test('OpenTelemetryInterceptorInsertYield enabled by version', (t) => {
+  const cases = [
+    { version: '0.1.0', expected: true },
+    { version: '1.0.0', expected: true },
+    { version: '1.9.0-rc.0', expected: true },
+    { version: '1.11.3', expected: true },
+    { version: '1.13.1', expected: true },
+    { version: '1.13.2', expected: false },
+    { version: '1.14.0', expected: false },
+    { version: '2.0.0', expected: false },
+  ];
+  for (const { version, expected } of cases) {
+    const actual = composeConditions(SdkFlags.OpenTelemetryInterceptorInsertYield.alternativeConditions)({
+      info: {} as WorkflowInfo,
+      sdkVersion: version,
+    });
+    t.is(actual, expected, `Expected OpenTelemetryInterceptorInsertYield on ${version} to evaluate as ${expected}`);
+  }
+});

packages/workflow/src/flags.ts

@@ -59,6 +59,13 @@ export const SdkFlags = {
   OpenTelemetryHandleSignalInterceptorInsertYield: defineFlag(3, false, [
     isBetween({ major: 1, minor: 11, patch: 3 }, { major: 1, minor: 13, patch: 2 }),
   ]),
+  /**
+   * The interceptors provided by @temporalio/interceptors-opentelemetry initially had unnecessary yield points.
+   * If replaying a workflow created from these versions a yield point is injected to prevent any NDE.
+   *
+   * @since Introduced in 1.13.2
+   */
+  OpenTelemetryInterceptorInsertYield: defineFlag(3, false, [isBefore({ major: 1, minor: 13, patch: 2 })]),
 } as const;
 
 function defineFlag(id: number, def: boolean, alternativeConditions?: AltConditionFn[]): SdkFlag {