Workflow Context gets wiped when waiting for a signal

Community Support
1

We have a long running workflow and we pass along a context that is tied to the tenant that started the workflow this includes things like tenantId, and a SHA that references specific configs that we can pull in.

This is passed in using headers at the start of the workflow and we built some basic interceptors that take these headers in and pass them along to the scheduled activities in the outbound interceptor.

I started testing and things go well till we get to a stop in the flow that is waiting for a signal, when we receive this signal it wipes out the current workflow context. I know this is because the Signal doesnt have any of the headers that we had when the workflow started.

The question here is whether i am missing something or just have some bad assumptions about how context propagation works.

import {
    ActivityInput, AsyncLocalStorage, Next, WorkflowExecuteInput, WorkflowInboundCallsInterceptor,
    WorkflowInterceptorsFactory, WorkflowOutboundCallsInterceptor
} from '@temporalio/workflow';

const asyncStorage = new AsyncLocalStorage<any>();

class InboundInterceptor implements WorkflowInboundCallsInterceptor {
  async execute(
    input: WorkflowExecuteInput,
    next: Next<InboundInterceptor, 'execute'>,
  ): Promise<void> {
    const context = asyncStorage.getStore();
    // check if there is already a context (useful for when we get signals)
    if (context) {
      await next(input);
    } else {
      return new Promise((res) => {
        asyncStorage.run(input.headers, async () => {
          await next(input);
          res();
        });
      });
    }
  }
}

class OutboundInterceptor implements WorkflowOutboundCallsInterceptor {
  async scheduleActivity(
    input: ActivityInput,
    next: Next<WorkflowOutboundCallsInterceptor, 'scheduleActivity'>,
  ): Promise<unknown> {
    const headers = asyncStorage.getStore();
    return await next({
      ...input,
      headers,
    });
  }
}

export const interceptors: WorkflowInterceptorsFactory = () => ({
  outbound: [new OutboundInterceptor()],
  inbound: [new InboundInterceptor()],
});

The idea of the above code is just to persist the headers along to any activities that are scheduled. But i need this persistence to exist even after we receive a signal.

2

Signal handlers are executed in a different async context from the main workflow function that might be the cause here.
If you show some sample workflow code, I would have more details.

3

Note that WorkflowInboundCallsInterceptor.execute only wraps the main function execution, not signal handlers.

4

For now you can use attributes to store the headers from workflow start.
Somethings like this (untested):

import {
    ActivityInput, Next, WorkflowExecuteInput, WorkflowInboundCallsInterceptor,
    WorkflowInterceptorsFactory, WorkflowOutboundCallsInterceptor
} from '@temporalio/workflow';

interface Context {
  headers?: Record<string, Payload>;
}

class InboundInterceptor implements WorkflowInboundCallsInterceptor {
  constructor(protected readonly context: Context) {}

  async execute(
    input: WorkflowExecuteInput,
    next: Next<InboundInterceptor, 'execute'>,
  ): Promise<void> {
    this.context.headers = input.headers;
    return await next(input); 
  }
}

class OutboundInterceptor implements WorkflowOutboundCallsInterceptor {
  constructor(protected readonly context: Context) {}

  async scheduleActivity(
    input: ActivityInput,
    next: Next<WorkflowOutboundCallsInterceptor, 'scheduleActivity'>,
  ): Promise<unknown> {
    const headers = this.context.headers ?? {};
    return await next({
      ...input,
      headers,
    });
  }
}

export function interceptors(): WorkflowInterceptorsFactory {
  const context = {};
  return {
    outbound: [new OutboundInterceptor(context)],
    inbound: [new InboundInterceptor(context)],
  };
}