From c1a34f7eba661b89762379d378e4dbbf0ae880ed Mon Sep 17 00:00:00 2001 From: Tony Toth Date: Thu, 5 Nov 2026 17:19:49 +0200 Subject: [PATCH] Skip no-op payload passes without a codec or external storage --- changelog/fixed/sprinting-sandbox-sloth.md | 1 + temporalio/bridge/worker.py | 14 ++++-- temporalio/converter/_data_converter.py | 7 +-- temporalio/worker/_workflow.py | 7 ++- tests/test_serialization_context.py | 58 +++++++++++++++++++++- 5 files changed, 76 insertions(+), 11 deletions(-) create mode 100644 changelog/fixed/sprinting-sandbox-sloth.md diff --git a/changelog/fixed/sprinting-sandbox-sloth.md b/changelog/fixed/sprinting-sandbox-sloth.md new file mode 100644 index 000000000..47a59f023 --- /dev/null +++ b/changelog/fixed/sprinting-sandbox-sloth.md @@ -0,0 +1 @@ +Workflow workers without a payload codec or external storage skip the payload codec passes and the per-command serialization-context lookups on every activation and completion, restoring the workflow task and replay performance lost since 1.23. diff --git a/temporalio/bridge/worker.py b/temporalio/bridge/worker.py index 02cc7d03f..c16696944 100644 --- a/temporalio/bridge/worker.py +++ b/temporalio/bridge/worker.py @@ -330,10 +330,11 @@ async def decode_activation( _Visitor(data_converter._external_retrieve_payload_sequence), activation ) - await CommandAwarePayloadVisitor( - skip_search_attributes=True, - skip_headers=not decode_headers, - ).visit(_Visitor(data_converter._decode_payload_sequence), activation) + if data_converter.payload_codec: + await CommandAwarePayloadVisitor( + skip_search_attributes=True, + skip_headers=not decode_headers, + ).visit(_Visitor(data_converter._decode_payload_sequence), activation) return metrics @@ -349,6 +350,10 @@ async def encode_completion( Returns: Metrics from any external storage store operations that occurred. """ + metrics = temporalio.converter._extstore.StorageOperationMetrics() + if not data_converter._has_codec_or_external_storage: + return metrics + await CommandAwarePayloadVisitor( skip_search_attributes=True, skip_headers=not encode_headers, @@ -357,7 +362,6 @@ async def encode_completion( completion, ) - metrics = temporalio.converter._extstore.StorageOperationMetrics() with metrics.track(): await CommandAwarePayloadVisitor( skip_search_attributes=True, diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 663c4b079..4985c5b7d 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -423,10 +423,11 @@ async def _decode_payload_sequence( decoded_payloads = await self.payload_codec.decode(decoded_payloads) return decoded_payloads - # Temporary shortcircuit detection while the _decode_* methods may no-op if - # a payload codec is not configured. Remove once those paths have more to them. + # Without a codec or external storage, the payload sequence methods only reject + # external storage references, so workers skip the passes that cannot change a + # payload and the per-command context lookups. @property - def _decode_payload_has_effect(self) -> bool: + def _has_codec_or_external_storage(self) -> bool: return self.payload_codec is not None or self.external_storage is not None diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 64c8a323b..b0d97c9b6 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -399,7 +399,10 @@ async def _handle_activation( ), ), ) - if workflow: + # Resolving a command's context calls into the workflow instance (the + # sandbox, when sandboxed) for every payload sequence, and only a codec + # or external storage can use it. + if workflow and data_converter._has_codec_or_external_storage: data_converter = _CommandAwareDataConverter.create( instance=workflow.instance, context_free_dc=self._data_converter, @@ -516,7 +519,7 @@ async def _handle_activation( completion.run_id = act.run_id # Encode completion - if workflow: + if workflow and self._data_converter._has_codec_or_external_storage: workflow_context = temporalio.converter.WorkflowSerializationContext( namespace=self._namespace, workflow_id=workflow.workflow_id, diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py index 1c612dcee..def83f1bb 100644 --- a/tests/test_serialization_context.py +++ b/tests/test_serialization_context.py @@ -67,7 +67,10 @@ from temporalio.exceptions import ApplicationError, NexusOperationError from temporalio.testing import WorkflowEnvironment from temporalio.worker import Replayer, Worker -from temporalio.worker._workflow_instance import UnsandboxedWorkflowRunner +from temporalio.worker._workflow_instance import ( + UnsandboxedWorkflowRunner, + _WorkflowInstanceImpl, +) from tests.helpers import assert_eq_eventually from tests.helpers.nexus import make_nexus_endpoint_name from tests.test_extstore import InMemoryTestDriver @@ -1347,6 +1350,59 @@ async def test_codec_with_context(client: Client): del test_traces[workflow_id] +async def test_command_context_resolved_only_with_codec_or_external_storage( + client: Client, monkeypatch: pytest.MonkeyPatch +): + """Context resolution calls into the workflow instance, which is costly in the + sandbox, so workers skip it when no codec or external storage can use it.""" + lookups = 0 + get_serialization_context = _WorkflowInstanceImpl.get_serialization_context + + def counting_get_serialization_context(self, command_info): # type: ignore[reportMissingParameterType] + nonlocal lookups + lookups += 1 + return get_serialization_context(self, command_info) + + monkeypatch.setattr( + _WorkflowInstanceImpl, + "get_serialization_context", + counting_get_serialization_context, + ) + + async def run_workflow(data_converter: DataConverter) -> int: + nonlocal lookups + lookups = 0 + workflow_id = str(uuid.uuid4()) + task_queue = str(uuid.uuid4()) + client_config = client.config() + client_config["data_converter"] = data_converter + converter_client = Client(**client_config) + async with Worker( + converter_client, + task_queue=task_queue, + workflows=[CodecTestWorkflow], + workflow_runner=UnsandboxedWorkflowRunner(), + ): + await converter_client.execute_workflow( + CodecTestWorkflow.run, + "data", + id=workflow_id, + task_queue=task_queue, + ) + test_traces.pop(workflow_id, None) + return lookups + + assert await run_workflow(DataConverter.default) == 0 + assert ( + await run_workflow( + dataclasses.replace( + DataConverter.default, payload_codec=PayloadCodecWithContext() + ) + ) + > 0 + ) + + # Local activity codec test