[Bug] Update-with-start double-encodes the OTel tracing header with HeaderCodecBehavior.CODEC, failing every workflow task

[LilianBoulardAcolad]

What are you really trying to do?

Running a Temporal Python service with end-to-end payload encryption: a custom PayloadCodec plus header_codec_behavior=HeaderCodecBehavior.CODEC so that headers are sealed too, and contrib.opentelemetry.TracingInterceptor for trace propagation. Workflows are started with update-with-start (execute_update_with_start_workflow).

Describe the bug

With a PayloadCodec installed and HeaderCodecBehavior.CODEC, the OTel tracing header on the update half of an update-with-start is passed through the codec twice. The worker decodes headers exactly once, so the header is still codec-encoded when the inbound tracing interceptor reads it, and every workflow task fails:

KeyError: 'Unknown payload encoding <your codec's encoding>'

The update never becomes durable. Plain start_workflow followed by a separate update is unaffected; only update-with-start is.

Cause, as far as I can trace it:

1. _TracingClientOutboundInterceptor.start_update_with_start_workflow (temporalio/contrib/opentelemetry/_interceptor.py, lines 335-340 on 1.31.0) copies the header Payload object from the start input into the update input:

   otel_header = input.start_workflow_input.headers.get(self.root.header_key)
   if otel_header:
       input.update_workflow_input.headers = {
           **input.update_workflow_input.headers,
           self.root.header_key: otel_header,
       }

   The dict is new, but the Payload is the same object in both mappings.

2. _apply_headers (temporalio/client/_helpers.py) encodes in place, mutating the caller-visible payload:

   if encode_headers:
       for payload in source.values():
           payload.CopyFrom(await data_converter._transform_outbound_payload(payload))

3. _ClientImpl._start_workflow_update_with_start builds the start request and then the update request, and each build calls _apply_headers once. Both calls reach the same shared Payload, so it is encoded twice. The start request's header is correct (encoded once); the update request's header is encoded twice.

4. The worker decodes headers once (_WorkflowWorker passes decode_headers=self._encode_headers to bridge.worker.decode_activation, which runs _decode_payload_sequence a single time over the DoUpdate headers), so one layer remains when _TracingWorkflowInboundInterceptor.handle_update_handler calls self.payload_converter.from_payloads([link_context_header]).

Note that the while True durability loop in _start_workflow_update_with_start re-sends already-built requests, so it adds no further layers: the count is exactly two regardless of retries.

Confirming instrumentation: a PayloadCodec.encode that flags payloads already carrying its own encoding marker reports exactly one such payload per update-with-start call. An interceptor placed innermost (after the tracing interceptor) prints identical id() values for the start and update header payloads.

Minimal Reproduction

Requires temporalio and opentelemetry-sdk. The tracing interceptor only writes a header when its tracer produces a valid span context, so the set_tracer_provider call is required to trigger this; no explicit caller span is needed (the interceptor's own client span is enough).

import asyncio, base64, sys, uuid
from collections.abc import Sequence

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

from temporalio import workflow
from temporalio.api.common.v1 import Payload
from temporalio.client import Client, WithStartWorkflowOperation
from temporalio.common import HeaderCodecBehavior, WorkflowIDConflictPolicy
from temporalio.contrib.opentelemetry import TracingInterceptor
from temporalio.converter import DataConverter, PayloadCodec
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

MARKER = b"binary/b64test"
double_encodes: list[int] = []


class B64Codec(PayloadCodec):
    """Any codec shows this; base64 keeps the repro dependency-free."""

    async def encode(self, payloads: Sequence[Payload]) -> list[Payload]:
        for p in payloads:
            if p.metadata.get("encoding") == MARKER:
                double_encodes.append(id(p))
        return [
            Payload(metadata={"encoding": MARKER}, data=base64.b64encode(p.SerializeToString()))
            for p in payloads
        ]

    async def decode(self, payloads: Sequence[Payload]) -> list[Payload]:
        out = []
        for p in payloads:
            if p.metadata.get("encoding") != MARKER:
                out.append(p)
                continue
            decoded = Payload()
            decoded.ParseFromString(base64.b64decode(p.data))
            out.append(decoded)
        return out


@workflow.defn
class MyWorkflow:
    def __init__(self) -> None:
        self.done = False

    @workflow.run
    async def run(self) -> str:
        await workflow.wait_condition(lambda: self.done)
        return "workflow-done"

    @workflow.update
    async def my_update(self) -> str:
        self.done = True
        return "update-done"


async def main() -> None:
    trace.set_tracer_provider(TracerProvider())

    env = await WorkflowEnvironment.start_local()
    try:
        client = await Client.connect(
            env.client.service_client.config.target_host,
            namespace="default",
            data_converter=DataConverter(payload_codec=B64Codec()),
            header_codec_behavior=HeaderCodecBehavior.CODEC,
            interceptors=[TracingInterceptor()],
        )
        task_queue = f"tq-{uuid.uuid4()}"
        async with Worker(client, task_queue=task_queue, workflows=[MyWorkflow]):
            start_op = WithStartWorkflowOperation(
                MyWorkflow.run,
                id=f"wf-{uuid.uuid4()}",
                task_queue=task_queue,
                id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
            )
            result = await asyncio.wait_for(
                client.execute_update_with_start_workflow(
                    MyWorkflow.my_update, start_workflow_operation=start_op
                ),
                timeout=20,
            )
            print(f"RESULT: {result}")
    except BaseException as e:
        print(f"FAILED: {type(e).__name__}: {e}")
        sys.exit(1)
    finally:
        print(f"payloads handed to encode() already encoded: {len(double_encodes)}")
        await env.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

Observed:

FAILED: WorkflowUpdateRPCTimeoutOrCancelledError: Timeout or cancellation waiting for update
payloads handed to encode() already encoded: 1

with the worker repeatedly logging:

  File ".../site-packages/temporalio/worker/_workflow_instance.py", line 721, in run_update
    success = await self._inbound.handle_update_handler(handler_input)
  File ".../site-packages/temporalio/contrib/opentelemetry/_interceptor.py", line 602, in handle_update_handler
    link_context_carrier = self.payload_converter.from_payloads(
  File ".../site-packages/temporalio/converter/_payload_converter.py", line 634, in from_payloads
    return self._inner_payload_converter.from_payloads(payloads, None)
  File ".../site-packages/temporalio/converter/_payload_converter.py", line 374, in from_payloads
    raise KeyError(f"Unknown payload encoding {encoding.decode()}")
KeyError: 'Unknown payload encoding binary/b64test'

and the server logging workflow-task-fail-cause=WorkflowWorkerUnhandledFailure.

Controls, all behaving as the analysis predicts, and identical on 1.31.0 and on main:

Variation	Result
As above	fails; 1 already-encoded payload
start_workflow then a separate execute_update	passes; 0 already-encoded payloads
Same update-with-start with HeaderCodecBehavior.NO_CODEC	passes (the object is still shared, but nothing mutates it)

Environment/Versions

• OS and processor: Ubuntu 24.04, x86_64
• SDK version: temporalio 1.31.0, and main built from source (Python 3.12.3); server is WorkflowEnvironment.start_local()
• opentelemetry-sdk / opentelemetry-api 1.44.0
• Not Docker/Kubernetes; also reproduced from a source build of main, see below

First affected release is 1.19.0: commit 2efd9a7 (#1150), which added start_update_with_start_workflow to the OTel interceptor, is contained in 1.19.0 and not in 1.18.0. Affected through 1.31.0.

main is affected too. I built it from source and reproduced there: sdk-python at 1c30f89b1f7fd50117f138bed2f4e96d93cd2b0b, sdk-core submodule at 00677170aa7dc62d90bf3d6d3d9f97717a49ea77. The failure and all three controls behave exactly as on 1.31.0.

That build reports version string 1.31.0, so to be sure of what was running I compared file hashes against the released 1.31.0: client/_impl.py and worker/_workflow.py differ (confirming the build really is main), while the two implicated files, contrib/opentelemetry/_interceptor.py and client/_helpers.py, are byte-identical between the two. Of the call-path files that changed between 1.31.0 and main, none contains a header-related change: the only header hunk is an unrelated Nexus completion-callback refactor, and _start_workflow_update_with_start and both request builders are untouched.

Additional context

A user-side workaround, for anyone hitting this: a client interceptor ordered after TracingInterceptor that gives the update half its own header payloads.

class _UnsharedUpdateHeaders(OutboundInterceptor):
    async def start_update_with_start_workflow(self, input):
        input.update_workflow_input.headers = {
            key: Payload(metadata=dict(payload.metadata), data=payload.data)
            for key, payload in input.update_workflow_input.headers.items()
        }
        return await super().start_update_with_start_workflow(input)


class UnshareUpdateWithStartHeaders(Interceptor):
    def intercept_client(self, next):
        return _UnsharedUpdateHeaders(next)


# interceptors=[TracingInterceptor(), UnshareUpdateWithStartHeaders()]

The order matters: interceptors are chained with reversed(), so the first entry is outermost, and the fix must sit inside the tracing interceptor to see the header it wrote. Putting it first instead reproduces the original failure.

Two identified possible fix sites:

• Have the OTel interceptor place a copy rather than the same Payload object into update_workflow_input.headers.
• Have _apply_headers not mutate its source (build the encoded payload into dest instead). This is the more general fix: the same in-place mutation would double-encode for any caller that reuses one header payload object across two RPCs, for example the same ScheduleActionStartWorkflow passed to two schedule calls.