break up

Author: chris-olszewski

packages/interceptors-opentelemetry/src/instrumentation.ts

@@ -15,7 +15,7 @@ const payloadConverter = defaultPayloadConverter;
 /**
  * If found, return an otel Context deserialized from the provided headers
  */
-export async function extractContextFromHeaders(headers: Headers): Promise<otel.Context | undefined> {
+export function extractContextFromHeaders(headers: Headers): otel.Context | undefined {
   const encodedSpanContext = headers[TRACE_HEADER];
   if (encodedSpanContext === undefined) {
     return undefined;

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

@@ -34,7 +34,7 @@ export class OpenTelemetryActivityInboundInterceptor implements ActivityInboundC
   }
 
   async execute(input: ActivityExecuteInput, next: Next<ActivityInboundCallsInterceptor, 'execute'>): Promise<unknown> {
-    const context = await extractContextFromHeaders(input.headers);
+    const context = extractContextFromHeaders(input.headers);
     const spanName = `${SpanName.ACTIVITY_EXECUTE}${SPAN_DELIMITER}${this.ctx.info.activityType}`;
     return await instrument({ tracer: this.tracer, spanName, fn: () => next(input), context });
   }

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

@@ -56,7 +56,7 @@ export class OpenTelemetryInboundInterceptor implements WorkflowInboundCallsInte
     input: WorkflowExecuteInput,
     next: Next<WorkflowInboundCallsInterceptor, 'execute'>
   ): Promise<unknown> {
-    const context = await extractContextFromHeaders(input.headers);
+    const context = await Promise.resolve(extractContextFromHeaders(input.headers));
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.WORKFLOW_EXECUTE}${SPAN_DELIMITER}${workflowInfo().workflowType}`,
@@ -70,7 +70,7 @@ export class OpenTelemetryInboundInterceptor implements WorkflowInboundCallsInte
     input: SignalInput,
     next: Next<WorkflowInboundCallsInterceptor, 'handleSignal'>
   ): Promise<void> {
-    const context = await extractContextFromHeaders(input.headers);
+    const context = extractContextFromHeaders(input.headers);
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.WORKFLOW_SIGNAL}${SPAN_DELIMITER}${input.signalName}`,

packages/test/src/test-otel.ts

@@ -630,7 +630,7 @@ test('Can replay otel history from 1.13.1', async (t) => {
           workflowInterceptorModules: [require.resolve('./workflows/signal-start-otel')],
         }),
         interceptors: {
-          workflowModules: [require.resolve('./workflows/otel-interceptors')],
+          workflowModules: [require.resolve('./workflows/signal-start-otel')],
           activity: [
             (ctx) => ({
               inbound: new OpenTelemetryActivityInboundInterceptor(ctx),

packages/worker/src/workflow/vm-shared.ts

@@ -357,6 +357,7 @@ export abstract class BaseVMWorkflow implements Workflow {
         },
       }));
       this.activator.addKnownFlags(activation.availableInternalFlags ?? []);
+      if (activation.lastSdkVersion) this.activator.sdkVersion = activation.lastSdkVersion;
 
       // Initialization of the workflow must happen before anything else. Yet, keep the init job in
       // place in the list as we'll use it as a marker to know when to start the workflow function.

packages/workflow/src/flags.ts

@@ -44,6 +44,8 @@ export const SdkFlags = {
    *        to implicitely have this flag on.
    */
   ProcessWorkflowActivationJobsAsSingleBatch: defineFlag(2, true, [buildIdSdkVersionMatches(/1\.11\.[01]/)]),
+
+  OpenTelemetryInterceptorInsertYieldPoint: defineFlag(3, false, [({ info }) => false]),
 } as const;
 
 function defineFlag(id: number, def: boolean, alternativeConditions?: AltConditionFn[]): SdkFlag {

packages/workflow/src/internals.ts

@@ -434,6 +434,8 @@ export class Activator implements ActivationHandler {
 
   private readonly knownFlags = new Set<number>();
 
+  sdkVersion?: string;
+
   /**
    * Buffered sink calls per activation
    */
@@ -978,7 +980,22 @@ export class Activator implements ActivationHandler {
 
     const signalExecutionNum = this.signalHandlerExecutionSeq++;
     this.inProgressSignals.set(signalExecutionNum, { name: signalName, unfinishedPolicy });
-    const execute = composeInterceptors(interceptors, 'handleSignal', this.signalWorkflowNextHandler.bind(this));
+    const injectYield = shouldInjectYield(this.sdkVersion);
+    const addedInterceptor: WorkflowInterceptors['inbound'] = injectYield
+      ? [
+          {
+            handleSignal: async (input, next) => {
+              await Promise.resolve();
+              return next(input);
+            },
+          },
+        ]
+      : [];
+    const execute = composeInterceptors(
+      [...addedInterceptor, ...interceptors],
+      'handleSignal',
+      this.signalWorkflowNextHandler.bind(this)
+    );
     execute({
       args: arrayFromPayloads(this.payloadConverter, activation.input),
       signalName,
@@ -1305,3 +1322,23 @@ then you can disable this warning by passing an option when setting the handler:
     Array.from(names.entries()).map(([name, count]) => ({ name, count }))
   )}`;
 }
+
+function shouldInjectYield(version?: string): boolean {
+  if (!version) {
+    return false;
+  }
+  const [major, minor, patch] = version.split('.');
+  // 1.11.5 - 1.13.1: need to inject
+  if (major !== '1') return false;
+
+  switch (minor) {
+    case '11':
+      return patch === '5';
+    case '12':
+      return true;
+    case '13':
+      return patch === '1';
+    default:
+      return false;
+  }
+}