From 636cd65efa7fae56eebdf92286fb883f8b3ca100 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Thu, 8 Oct 2026 14:08:12 -0400 Subject: [PATCH 01/10] feat(endode-extstore-refs): external storage references are now encoded and decoded. --- temporalio/bridge/worker.py | 13 +++ temporalio/converter/_data_converter.py | 85 +++++++++++++- temporalio/converter/_extstore.py | 6 + temporalio/worker/_nexus.py | 19 ++- temporalio/worker/_workflow.py | 10 ++ tests/test_extstore.py | 147 ++++++++++++++++++++---- tests/worker/test_extstore.py | 59 +++++++++- 7 files changed, 314 insertions(+), 25 deletions(-) diff --git a/temporalio/bridge/worker.py b/temporalio/bridge/worker.py index 4b7f55d092..1a6622d1b0 100644 --- a/temporalio/bridge/worker.py +++ b/temporalio/bridge/worker.py @@ -315,6 +315,11 @@ async def decode_activation( Returns: Metrics from any external storage retrieval operations that occurred. """ + await CommandAwarePayloadVisitor( + skip_search_attributes=True, + skip_headers=not decode_headers, + ).visit(_Visitor(data_converter._decode_reference_payload_sequence), activation) + metrics = temporalio.converter._extstore.StorageOperationMetrics() with metrics.track(): await CommandAwarePayloadVisitor( @@ -363,4 +368,12 @@ async def encode_completion( completion, ) + await CommandAwarePayloadVisitor( + skip_search_attributes=True, + skip_headers=not encode_headers, + ).visit( + _Visitor(data_converter._encode_reference_payload_sequence), + completion, + ) + return metrics diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index f924ccf6c8..7b47e9ee84 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import dataclasses from collections.abc import Mapping, Sequence from dataclasses import dataclass @@ -15,6 +16,7 @@ import temporalio.common from temporalio.converter._extstore import ( _REFERENCE_ENCODING, + _REFERENCE_MARKER_KEY, _REFERENCE_MESSAGE_TYPE, ExternalStorage, StorageDriverStoreContext, @@ -44,6 +46,11 @@ def _is_reference_payload(p: temporalio.api.common.v1.Payload) -> bool: ) +def _has_reference_marker(p: temporalio.api.common.v1.Payload) -> bool: + """Return True if *p* carries the external-storage reference marker.""" + return p.metadata.get(_REFERENCE_MARKER_KEY) == b"true" + + # Import defaults from public API to avoid pydoctor cross-reference issues if TYPE_CHECKING: from temporalio.converter import DefaultFailureConverter, DefaultPayloadConverter @@ -118,6 +125,7 @@ async def encode( payloads = self.payload_converter.to_payloads(values) payloads = await self._encode_payload_sequence(payloads) payloads = await self._external_store_payload_sequence(payloads) + payloads = await self._encode_reference_payload_sequence(payloads) return payloads async def decode( @@ -135,6 +143,7 @@ async def decode( Returns: Decoded and converted values. """ + payloads = await self._decode_reference_payload_sequence(payloads) payloads = await self._external_retrieve_payload_sequence(payloads) payloads = await self._decode_payload_sequence(payloads) return self.payload_converter.from_payloads(payloads, type_hints) @@ -269,7 +278,7 @@ async def _transform_outbound_payload( payload = (await self.payload_codec.encode([payload]))[0] if self.external_storage: payload = await self.external_storage._store_payload(payload) - return payload + return (await self._encode_reference_payload_sequence([payload]))[0] async def _transform_outbound_payloads( self, payloads: temporalio.api.common.v1.Payloads @@ -278,10 +287,16 @@ async def _transform_outbound_payloads( await self.payload_codec.encode_wrapper(payloads) if self.external_storage: await self.external_storage._store_payloads(payloads) + encoded_payloads = await self._encode_reference_payload_sequence( + payloads.payloads + ) + for i, payload in enumerate(encoded_payloads): + payloads.payloads[i].CopyFrom(payload) async def _transform_inbound_payload( self, payload: temporalio.api.common.v1.Payload ) -> temporalio.api.common.v1.Payload: + payload = (await self._decode_reference_payload_sequence([payload]))[0] if self.external_storage: payload = await self.external_storage._retrieve_payload(payload) if self.payload_codec: @@ -291,6 +306,11 @@ async def _transform_inbound_payload( async def _transform_inbound_payloads( self, payloads: temporalio.api.common.v1.Payloads ): + decoded_payloads = await self._decode_reference_payload_sequence( + payloads.payloads + ) + for i, payload in enumerate(decoded_payloads): + payloads.payloads[i].CopyFrom(payload) if self.external_storage: await self.external_storage._retrieve_payloads(payloads) else: @@ -321,6 +341,69 @@ async def _external_store_payload_sequence( ) return stored_payloads + async def _encode_reference_payload_sequence( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + """Codec encode external storage reference payloads only.""" + encoded_payloads = list(payloads) + payload_codec = self.payload_codec + if not payload_codec: + return encoded_payloads + reference_indices = [ + i for i, p in enumerate(encoded_payloads) if _is_reference_payload(p) + ] + if not reference_indices: + return encoded_payloads + encoded_references = await asyncio.gather( + *(payload_codec.encode([encoded_payloads[i]]) for i in reference_indices) + ) + for i, [encoded_reference] in zip(reference_indices, encoded_references): + if not encoded_reference.external_payloads: + encoded_reference.external_payloads.extend( + encoded_payloads[i].external_payloads + ) + encoded_reference.metadata[_REFERENCE_MARKER_KEY] = b"true" + encoded_payloads[i] = encoded_reference + return encoded_payloads + + async def _decode_reference_payload_sequence( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + """Codec decode marked external storage reference payloads only.""" + decoded_payloads = list(payloads) + marked_indices = [ + i for i, p in enumerate(decoded_payloads) if _has_reference_marker(p) + ] + if not marked_indices: + return decoded_payloads + payload_codec = self.payload_codec + if payload_codec: + encoded_references = [] + for i in marked_indices: + encoded_reference = temporalio.api.common.v1.Payload() + encoded_reference.CopyFrom(decoded_payloads[i]) + del encoded_reference.metadata[_REFERENCE_MARKER_KEY] + del encoded_reference.external_payloads[:] + encoded_references.append(encoded_reference) + decoded = await asyncio.gather( + *(payload_codec.decode([r]) for r in encoded_references) + ) + references = [reference for [reference] in decoded] + else: + references = [decoded_payloads[i] for i in marked_indices] + for i, decoded_reference in zip(marked_indices, references): + if not _is_reference_payload(decoded_reference): + raise RuntimeError( + "[TMPRL1106] Encountered an encoded external storage reference that could not be decoded. " + "Check that the payload codec used to write it is configured." + ) + if not decoded_reference.external_payloads: + decoded_reference.external_payloads.extend( + decoded_payloads[i].external_payloads + ) + decoded_payloads[i] = decoded_reference + return decoded_payloads + async def _external_retrieve_payload_sequence( self, payloads: Sequence[temporalio.api.common.v1.Payload] ) -> list[temporalio.api.common.v1.Payload]: diff --git a/temporalio/converter/_extstore.py b/temporalio/converter/_extstore.py index c300448b92..5dc3fd03d4 100644 --- a/temporalio/converter/_extstore.py +++ b/temporalio/converter/_extstore.py @@ -28,6 +28,10 @@ _REFERENCE_ENCODING = b"json/external-storage-reference" _REFERENCE_MESSAGE_TYPE = ExternalStorageReference.DESCRIPTOR.full_name.encode() +# Indicates this payload represents an externally stored payload. If encoded, the +# _REFERENCE_ENCODING could be any value. The presence of the _REFERENCE_MARKER_KEY +# with a value of b"true" is used to reliably identify such payloads. +_REFERENCE_MARKER_KEY = "__temporal_external_storage_reference" @dataclass @@ -430,6 +434,7 @@ async def _store_payload(self, payload: Payload) -> Payload: f"Failed to serialize storage reference for driver '{driver.name()}'" ) reference_payload.external_payloads.add().size_bytes = external_size + reference_payload.metadata[_REFERENCE_MARKER_KEY] = b"true" ExternalStorage._record_metrics(1, external_size, start_time, {driver.name()}) @@ -494,6 +499,7 @@ async def _store_payload_sequence( f"Failed to serialize storage reference for driver '{driver.name()}'" ) reference_payload.external_payloads.add().size_bytes = sizes[i] + reference_payload.metadata[_REFERENCE_MARKER_KEY] = b"true" results[indices[i]] = reference_payload external_size += sizes[i] diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index a2f4b8ca75..621e43abbc 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -234,7 +234,9 @@ async def _encode_completion( completion: temporalio.bridge.proto.nexus.NexusTaskCompletion, data_converter: temporalio.converter.DataConverter, ) -> None: - """Apply the payload codec then external storage to the completion's payloads.""" + """Apply the payload codec, external storage, then the payload codec to + any storage references in the completion's payloads. + """ await PayloadVisitor(skip_search_attributes=True, skip_headers=True).visit( _PayloadTransformVisitor(data_converter._encode_payload_sequence), completion, @@ -243,6 +245,10 @@ async def _encode_completion( _PayloadTransformVisitor(data_converter._external_store_payload_sequence), completion, ) + await PayloadVisitor(skip_search_attributes=True, skip_headers=True).visit( + _PayloadTransformVisitor(data_converter._encode_reference_payload_sequence), + completion, + ) def _data_converter_for_nexus_task( self, endpoint: str, service: str, operation: str @@ -566,6 +572,17 @@ async def deserialize( # payload untouched. payload = temporalio.api.common.v1.Payload() payload.CopyFrom(self.payload) + try: + await PayloadVisitor(skip_search_attributes=True, skip_headers=True).visit( + _PayloadTransformVisitor(dc._decode_reference_payload_sequence), + payload, + ) + except Exception as err: + raise nexusrpc.HandlerError( + "Payload codec failed to decode Nexus operation input storage reference", + type=nexusrpc.HandlerErrorType.INTERNAL, + ) from err + try: await PayloadVisitor(skip_search_attributes=True).visit( _PayloadTransformVisitor(dc._external_retrieve_payload_sequence), diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index f50955c4cf..3e6ab1e6a8 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -956,6 +956,16 @@ async def _external_store_payload_sequence( dc = self._get_current_dc()._with_store_context(store_ctx) return await dc._external_store_payload_sequence(payloads) + async def _encode_reference_payload_sequence( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + return await self._get_current_dc()._encode_reference_payload_sequence(payloads) + + async def _decode_reference_payload_sequence( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + return await self._get_current_dc()._decode_reference_payload_sequence(payloads) + async def _external_retrieve_payload_sequence( self, payloads: Sequence[temporalio.api.common.v1.Payload] ) -> list[temporalio.api.common.v1.Payload]: diff --git a/tests/test_extstore.py b/tests/test_extstore.py index 943fc7be2a..2c8c173de2 100644 --- a/tests/test_extstore.py +++ b/tests/test_extstore.py @@ -7,6 +7,7 @@ import pytest from temporalio.api.common.v1 import Payload +from temporalio.api.failure.v1 import Failure from temporalio.api.sdk.v1.external_storage_pb2 import ExternalStorageReference from temporalio.converter import ( DataConverter, @@ -410,8 +411,7 @@ class TestPayloadCodecWithExternalStorage: async def test_dc_payload_codec_encodes_stored_bytes(self): """DataConverter.payload_codec encodes the bytes handed to the driver - for storage. The reference payload written to workflow history is NOT - encoded by the DataConverter codec.""" + for storage.""" driver = InMemoryTestDriver() dc_codec = RecordingPayloadCodec("binary/dc-encoded") @@ -428,10 +428,6 @@ async def test_dc_payload_codec_encodes_stored_bytes(self): assert len(encoded) == 1 assert driver._store_calls == 1 - # The reference payload written to history must NOT carry the dc_codec label. - assert dc_codec.encoded_count == 1 - assert encoded[0].metadata.get("encoding") != b"binary/dc-encoded" - # The bytes given to the driver must carry the dc_codec label. stored_payload = Payload() stored_payload.ParseFromString(next(iter(driver._storage.values()))) @@ -440,13 +436,11 @@ async def test_dc_payload_codec_encodes_stored_bytes(self): # Round-trip must recover the original value. decoded = await converter.decode(encoded, [str]) assert decoded[0] == large_value - assert dc_codec.decoded_count == 1 assert driver._retrieve_calls == 1 - async def test_dc_payload_codec_does_not_encode_reference_payload(self): - """The reference payload stored in workflow history is NOT encoded by - DataConverter.payload_codec – encoding is applied to the stored bytes - instead.""" + async def test_dc_payload_codec_encodes_reference_payload(self): + """The reference payload written to workflow history is encoded by + DataConverter.payload_codec and marked, keeping its size details.""" driver = InMemoryTestDriver() dc_codec = RecordingPayloadCodec("binary/dc-encoded") @@ -461,23 +455,132 @@ async def test_dc_payload_codec_does_not_encode_reference_payload(self): large_value = "x" * 200 encoded = await converter.encode([large_value]) assert len(encoded) == 1 - assert driver._store_calls == 1 - - # Reference payload in history is NOT encoded by DataConverter.payload_codec. - assert dc_codec.encoded_count == 1 - assert encoded[0].metadata.get("encoding") != b"binary/dc-encoded" - # Stored bytes ARE encoded by DataConverter.payload_codec. - stored_payload = Payload() - stored_payload.ParseFromString(next(iter(driver._storage.values()))) - assert stored_payload.metadata.get("encoding") == b"binary/dc-encoded" + # Once for the stored payload, once for the reference payload. + assert dc_codec.encoded_count == 2 + assert encoded[0].metadata.get("encoding") == b"binary/dc-encoded" + assert ( + encoded[0].metadata.get("__temporal_external_storage_reference") == b"true" + ) + stored_size = len(next(iter(driver._storage.values()))) + assert [d.size_bytes for d in encoded[0].external_payloads] == [stored_size] - # Round-trip. decoded = await converter.decode(encoded, [str]) assert decoded[0] == large_value - assert dc_codec.decoded_count == 1 + assert dc_codec.decoded_count == 2 assert driver._retrieve_calls == 1 + async def test_inline_payload_not_marked(self): + """Payloads that stay inline are encoded once and never marked.""" + dc_codec = RecordingPayloadCodec("binary/dc-encoded") + converter = DataConverter( + payload_codec=dc_codec, + external_storage=ExternalStorage( + drivers=[InMemoryTestDriver()], + payload_size_threshold=1000, + ), + ) + + encoded = await converter.encode(["small"]) + + assert dc_codec.encoded_count == 1 + assert "__temporal_external_storage_reference" not in encoded[0].metadata + assert await converter.decode(encoded, [str]) == ["small"] + + async def test_reference_payload_marked_without_codec(self): + """Without a codec, the reference payload is left unencoded but still + marked, and round-trips.""" + driver = InMemoryTestDriver() + converter = DataConverter( + external_storage=ExternalStorage( + drivers=[driver], payload_size_threshold=50 + ) + ) + + encoded = await converter.encode(["x" * 200]) + + assert encoded[0].metadata.get("encoding") == b"json/protobuf" + assert ( + encoded[0].metadata.get("__temporal_external_storage_reference") == b"true" + ) + assert await converter.decode(encoded, [str]) == ["x" * 200] + + async def test_unmarked_reference_payload_still_decodes(self): + """A reference payload written before references were marked and + encoded still decodes when a codec is configured.""" + driver = InMemoryTestDriver() + dc_codec = RecordingPayloadCodec("binary/dc-encoded") + external_storage = ExternalStorage(drivers=[driver], payload_size_threshold=50) + + # Produce an unencoded, unmarked reference that points at codec-encoded + # bytes, as older SDK versions did. + encoded_value = await DataConverter(payload_codec=dc_codec).encode(["x" * 200]) + old_reference = await external_storage._store_payload(encoded_value[0]) + del old_reference.metadata["__temporal_external_storage_reference"] + + converter = DataConverter( + payload_codec=dc_codec, external_storage=external_storage + ) + decoded = await converter.decode([old_reference], [str]) + assert decoded[0] == "x" * 200 + + async def test_encoded_reference_payload_without_codec_fails(self): + """An encoded reference payload cannot be read without a codec.""" + driver = InMemoryTestDriver() + external_storage = ExternalStorage(drivers=[driver], payload_size_threshold=50) + encoded = await DataConverter( + payload_codec=RecordingPayloadCodec("binary/dc-encoded"), + external_storage=external_storage, + ).encode(["x" * 200]) + + with pytest.raises( + RuntimeError, + match=r"\[TMPRL1106\] Encountered an encoded external storage reference that could not be decoded", + ): + await DataConverter(external_storage=external_storage).decode( + encoded, [str] + ) + + async def test_marked_payload_that_is_not_a_reference_fails(self): + """A marked payload that does not decode to a reference fails before + external storage retrieval.""" + driver = InMemoryTestDriver() + converter = DataConverter( + payload_codec=RecordingPayloadCodec("binary/dc-encoded"), + external_storage=ExternalStorage(drivers=[driver]), + ) + encoded = await converter.encode(["value"]) + encoded[0].metadata["__temporal_external_storage_reference"] = b"true" + + with pytest.raises( + RuntimeError, + match=r"\[TMPRL1106\] Encountered an encoded external storage reference that could not be decoded", + ): + await converter.decode(encoded, [str]) + assert driver._retrieve_calls == 0 + + async def test_failure_reference_payload_encoded(self): + """Reference payloads inside failures are encoded and marked too.""" + driver = InMemoryTestDriver() + converter = DataConverter( + payload_codec=RecordingPayloadCodec("binary/dc-encoded"), + external_storage=ExternalStorage( + drivers=[driver], + payload_size_threshold=50, + ), + ) + failure = Failure() + await converter.encode_failure(ApplicationError("boom", "x" * 200), failure) + + details = failure.application_failure_info.details.payloads + assert ( + details[0].metadata.get("__temporal_external_storage_reference") == b"true" + ) + + error = await converter.decode_failure(failure) + assert isinstance(error, ApplicationError) + assert error.details == ("x" * 200,) + class TestMultiDriver: """Tests for ExternalStorage with multiple drivers.""" diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index e8ef8edb2c..456df019d2 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -31,7 +31,7 @@ from temporalio.testing._workflow import WorkflowEnvironment from temporalio.worker import Replayer from tests.helpers import assert_task_fail_eventually, new_worker -from tests.test_extstore import InMemoryTestDriver +from tests.test_extstore import InMemoryTestDriver, RecordingPayloadCodec @dataclass(frozen=True) @@ -429,6 +429,63 @@ async def test_replay_extstore_history_succeeds_with_correct_extstore( ).replay_workflow(history) +async def test_extstore_reference_payloads_codec_encoded_in_history( + env: WorkflowEnvironment, +) -> None: + """Every reference payload written to history by the client, workflow + worker, and activity worker is codec-encoded and marked, and the history + replays.""" + data_converter = dataclasses.replace( + temporalio.converter.default(), + payload_codec=RecordingPayloadCodec("binary/test-encoded"), + external_storage=ExternalStorage( + drivers=[InMemoryTestDriver()], + payload_size_threshold=512, + ), + ) + client = await env.connect_client(data_converter=data_converter) + async with new_worker( + client, ExtStoreWorkflow, activities=[ext_store_activity] + ) as worker: + handle = await client.start_workflow( + ExtStoreWorkflow.run, + ExtStoreWorkflowInput( + input_data="wi" * 512, + activity_input_size=1024, + activity_output_size=1024, + output_size=1024, + ), + id=f"workflow-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + assert await handle.result() == "wo" * 512 + + history = await handle.fetch_history() + reference_payloads = [] + for event in history.events: + if event.HasField("workflow_execution_started_event_attributes"): + attrs = event.workflow_execution_started_event_attributes + reference_payloads.append(attrs.input.payloads[0]) + elif event.HasField("activity_task_scheduled_event_attributes"): + attrs = event.activity_task_scheduled_event_attributes + reference_payloads.append(attrs.input.payloads[0]) + elif event.HasField("activity_task_completed_event_attributes"): + attrs = event.activity_task_completed_event_attributes + reference_payloads.append(attrs.result.payloads[0]) + elif event.HasField("workflow_execution_completed_event_attributes"): + attrs = event.workflow_execution_completed_event_attributes + reference_payloads.append(attrs.result.payloads[0]) + assert len(reference_payloads) == 4 + for payload in reference_payloads: + assert payload.metadata["encoding"] == b"binary/test-encoded" + assert payload.metadata["__temporal_external_storage_reference"] == b"true" + assert payload.external_payloads[0].size_bytes > 0 + + await Replayer( + workflows=[ExtStoreWorkflow], data_converter=data_converter + ).replay_workflow(history) + + async def test_replay_extstore_history_fails_with_empty_driver( env: WorkflowEnvironment, ) -> None: From 3cf3c81b377263ba733dc38b9b79a991e852ee53 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Tue, 13 Oct 2026 13:24:29 -0400 Subject: [PATCH 02/10] add new integration tests for encoded extstore refs. --- tests/test_extstore_client.py | 336 ++++++++++++++++++++++++++++++++++ tests/worker/test_extstore.py | 85 +++++++++ 2 files changed, 421 insertions(+) create mode 100644 tests/test_extstore_client.py diff --git a/tests/test_extstore_client.py b/tests/test_extstore_client.py new file mode 100644 index 0000000000..ebdb6e8fa6 --- /dev/null +++ b/tests/test_extstore_client.py @@ -0,0 +1,336 @@ +"""Client paths encode and decode external storage reference payloads. + +These use a fake workflow service, so they check exactly what the client sends +and how it reads responses without depending on a server. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest + +import temporalio.api.common.v1 +import temporalio.api.enums.v1 +import temporalio.api.failure.v1 +import temporalio.api.history.v1 +import temporalio.api.update.v1 +import temporalio.api.workflowservice.v1 +import temporalio.client +import temporalio.converter +from temporalio.api.common.v1 import Payload +from temporalio.client import ( + Client, + Schedule, + ScheduleActionStartWorkflow, + ScheduleSpec, + WorkflowFailureError, +) +from temporalio.common import HeaderCodecBehavior +from temporalio.converter import DataConverter, ExternalStorage +from temporalio.exceptions import ApplicationError +from temporalio.service import ServiceClient +from tests.test_extstore import InMemoryTestDriver, RecordingPayloadCodec + +# Larger than the storage threshold below, so it is always stored externally. +LARGE = "x" * 200 +CODEC_ENCODING = b"binary/test-encoded" + + +class _FakeWorkflowService: + """Records each request and returns the configured response.""" + + def __init__(self, responses: Mapping[str, Any]) -> None: + self.requests: dict[str, Any] = {} + for name, response in responses.items(): + setattr(self, name, self._recorder(name, response)) + + def _recorder(self, name: str, response: Any) -> AsyncMock: + async def record(req: Any, **_kwargs: Any) -> Any: + self.requests[name] = req + return response + + return AsyncMock(side_effect=record) + + +def _data_converter() -> DataConverter: + return DataConverter( + payload_codec=RecordingPayloadCodec(CODEC_ENCODING.decode()), + external_storage=ExternalStorage( + drivers=[InMemoryTestDriver()], payload_size_threshold=50 + ), + ) + + +def _client( + service: _FakeWorkflowService, + data_converter: DataConverter, + **kwargs: Any, +) -> Client: + service_client = Mock(spec=ServiceClient) + service_client.workflow_service = service + service_client.config = Mock(identity="test-identity") + return Client( + service_client=service_client, + namespace="test-namespace", + data_converter=data_converter, + **kwargs, + ) + + +def _assert_encoded_reference(payload: Payload) -> None: + assert payload.metadata["encoding"] == CODEC_ENCODING + assert payload.metadata["__temporal_external_storage_reference"] == b"true" + assert payload.external_payloads + + +async def _encoded_reference(data_converter: DataConverter, value: Any) -> Payload: + [payload] = await data_converter.encode([value]) + _assert_encoded_reference(payload) + return payload + + +def _history_response( + event: temporalio.api.history.v1.HistoryEvent, +) -> temporalio.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse: + return temporalio.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse( + history=temporalio.api.history.v1.History(events=[event]) + ) + + +async def test_start_workflow_encodes_input_and_memo_references() -> None: + service = _FakeWorkflowService( + { + "start_workflow_execution": temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse( + run_id="run" + ) + } + ) + client = _client(service, _data_converter()) + + await client.start_workflow( + "Workflow", LARGE, id="wf", task_queue="tq", memo={"key": LARGE} + ) + + req = service.requests["start_workflow_execution"] + _assert_encoded_reference(req.input.payloads[0]) + _assert_encoded_reference(req.memo.fields["key"]) + + +async def test_signal_encodes_input_reference() -> None: + service = _FakeWorkflowService( + { + "signal_workflow_execution": temporalio.api.workflowservice.v1.SignalWorkflowExecutionResponse() + } + ) + client = _client(service, _data_converter()) + + await client.get_workflow_handle("wf").signal("signal", LARGE) + + _assert_encoded_reference( + service.requests["signal_workflow_execution"].input.payloads[0] + ) + + +async def test_signal_with_start_encodes_input_references() -> None: + service = _FakeWorkflowService( + { + "signal_with_start_workflow_execution": temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse( + run_id="run" + ) + } + ) + client = _client(service, _data_converter()) + + await client.start_workflow( + "Workflow", + LARGE, + id="wf", + task_queue="tq", + start_signal="signal", + start_signal_args=[LARGE], + ) + + req = service.requests["signal_with_start_workflow_execution"] + _assert_encoded_reference(req.input.payloads[0]) + _assert_encoded_reference(req.signal_input.payloads[0]) + + +async def test_update_encodes_args_and_decodes_result_references() -> None: + data_converter = _data_converter() + result = await _encoded_reference(data_converter, "update-result" * 20) + service = _FakeWorkflowService( + { + "update_workflow_execution": temporalio.api.workflowservice.v1.UpdateWorkflowExecutionResponse( + stage=temporalio.api.enums.v1.UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED, + outcome=temporalio.api.update.v1.Outcome( + success=temporalio.api.common.v1.Payloads(payloads=[result]) + ), + ) + } + ) + client = _client(service, data_converter) + + value = await client.get_workflow_handle("wf").execute_update( + "update", LARGE, result_type=str + ) + + assert value == "update-result" * 20 + req = service.requests["update_workflow_execution"] + _assert_encoded_reference(req.request.input.args.payloads[0]) + + +async def test_query_encodes_args_and_decodes_result_references() -> None: + data_converter = _data_converter() + result = await _encoded_reference(data_converter, "query-result" * 20) + service = _FakeWorkflowService( + { + "query_workflow": temporalio.api.workflowservice.v1.QueryWorkflowResponse( + query_result=temporalio.api.common.v1.Payloads(payloads=[result]) + ) + } + ) + client = _client(service, data_converter) + + value = await client.get_workflow_handle("wf").query( + "query", LARGE, result_type=str + ) + + assert value == "query-result" * 20 + _assert_encoded_reference( + service.requests["query_workflow"].query.query_args.payloads[0] + ) + + +async def test_create_schedule_encodes_action_input_reference() -> None: + service = _FakeWorkflowService( + {"create_schedule": temporalio.api.workflowservice.v1.CreateScheduleResponse()} + ) + client = _client(service, _data_converter()) + + await client.create_schedule( + "schedule", + Schedule( + action=ScheduleActionStartWorkflow( + "Workflow", LARGE, id="wf", task_queue="tq" + ), + spec=ScheduleSpec(), + ), + ) + + req = service.requests["create_schedule"] + _assert_encoded_reference(req.schedule.action.start_workflow.input.payloads[0]) + + +async def test_async_activity_completion_encodes_references() -> None: + service = _FakeWorkflowService( + { + "respond_activity_task_completed_by_id": temporalio.api.workflowservice.v1.RespondActivityTaskCompletedByIdResponse(), + "respond_activity_task_failed_by_id": temporalio.api.workflowservice.v1.RespondActivityTaskFailedByIdResponse(), + "record_activity_task_heartbeat_by_id": temporalio.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdResponse(), + } + ) + client = _client(service, _data_converter()) + handle = client.get_async_activity_handle( + workflow_id="wf", run_id=None, activity_id="activity" + ) + + await handle.heartbeat(LARGE) + await handle.complete(LARGE) + await handle.fail(ApplicationError("failed", LARGE)) + + _assert_encoded_reference( + service.requests["record_activity_task_heartbeat_by_id"].details.payloads[0] + ) + _assert_encoded_reference( + service.requests["respond_activity_task_completed_by_id"].result.payloads[0] + ) + failure = service.requests["respond_activity_task_failed_by_id"].failure + _assert_encoded_reference(failure.application_failure_info.details.payloads[0]) + + +class _LargeHeaderInterceptor(temporalio.client.Interceptor): + def intercept_client( + self, next: temporalio.client.OutboundInterceptor + ) -> temporalio.client.OutboundInterceptor: + return _LargeHeaderOutbound(next) + + +class _LargeHeaderOutbound(temporalio.client.OutboundInterceptor): + async def start_workflow( + self, input: temporalio.client.StartWorkflowInput + ) -> temporalio.client.WorkflowHandle[Any, Any]: + [header] = temporalio.converter.default().payload_converter.to_payloads([LARGE]) + input.headers = {**input.headers, "large": header} + return await super().start_workflow(input) + + +async def test_start_workflow_encodes_header_reference_when_headers_use_codec() -> None: + service = _FakeWorkflowService( + { + "start_workflow_execution": temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse( + run_id="run" + ) + } + ) + client = _client( + service, + _data_converter(), + interceptors=[_LargeHeaderInterceptor()], + header_codec_behavior=HeaderCodecBehavior.CODEC, + ) + + await client.start_workflow("Workflow", id="wf", task_queue="tq") + + req = service.requests["start_workflow_execution"] + _assert_encoded_reference(req.header.fields["large"]) + + +async def test_workflow_result_decodes_reference() -> None: + data_converter = _data_converter() + result = await _encoded_reference(data_converter, "workflow-result" * 20) + service = _FakeWorkflowService( + { + "get_workflow_execution_history": _history_response( + temporalio.api.history.v1.HistoryEvent( + event_type=temporalio.api.enums.v1.EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED, + workflow_execution_completed_event_attributes=temporalio.api.history.v1.WorkflowExecutionCompletedEventAttributes( + result=temporalio.api.common.v1.Payloads(payloads=[result]) + ), + ) + ) + } + ) + client = _client(service, data_converter) + + value = await client.get_workflow_handle("wf", result_type=str).result() + + assert value == "workflow-result" * 20 + + +async def test_workflow_failure_decodes_detail_reference() -> None: + data_converter = _data_converter() + failure = temporalio.api.failure.v1.Failure() + await data_converter.encode_failure(ApplicationError("failed", LARGE), failure) + _assert_encoded_reference(failure.application_failure_info.details.payloads[0]) + service = _FakeWorkflowService( + { + "get_workflow_execution_history": _history_response( + temporalio.api.history.v1.HistoryEvent( + event_type=temporalio.api.enums.v1.EventType.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED, + workflow_execution_failed_event_attributes=temporalio.api.history.v1.WorkflowExecutionFailedEventAttributes( + failure=failure + ), + ) + ) + } + ) + client = _client(service, data_converter) + + with pytest.raises(WorkflowFailureError) as err: + await client.get_workflow_handle("wf").result() + + assert isinstance(err.value.cause, ApplicationError) + assert err.value.cause.details == (LARGE,) diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index 456df019d2..c2476cdf7f 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -19,13 +19,18 @@ from temporalio.client import Client, WorkflowFailureError, WorkflowHandle from temporalio.common import RetryPolicy from temporalio.converter import ( + ActivitySerializationContext, ExternalStorage, + PayloadCodec, + SerializationContext, StorageDriver, StorageDriverActivityInfo, StorageDriverClaim, StorageDriverRetrieveContext, StorageDriverStoreContext, StorageDriverWorkflowInfo, + WithSerializationContext, + WorkflowSerializationContext, ) from temporalio.exceptions import ActivityError, ApplicationError from temporalio.testing._workflow import WorkflowEnvironment @@ -486,6 +491,86 @@ async def test_extstore_reference_payloads_codec_encoded_in_history( ).replay_workflow(history) +class ReferenceContextRecordingCodec(PayloadCodec, WithSerializationContext): + """Encodes like RecordingPayloadCodec and records the serialization context + each storage reference payload is encoded with.""" + + def __init__( + self, + reference_contexts: list[SerializationContext | None], + context: SerializationContext | None = None, + ) -> None: + self._reference_contexts = reference_contexts + self._context = context + self._codec = RecordingPayloadCodec("binary/test-encoded") + + def with_context( + self, context: SerializationContext + ) -> "ReferenceContextRecordingCodec": + return ReferenceContextRecordingCodec(self._reference_contexts, context) + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + for p in payloads: + if ( + p.metadata.get("messageType") + == b"temporal.api.sdk.v1.ExternalStorageReference" + ): + self._reference_contexts.append(self._context) + return await self._codec.encode(payloads) + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + return await self._codec.decode(payloads) + + +async def test_extstore_reference_payloads_use_payload_serialization_context( + env: WorkflowEnvironment, +) -> None: + """Each reference payload is encoded with the serialization context of the + value it replaces: workflow context for workflow input and result, activity + context for activity input and result.""" + reference_contexts: list[SerializationContext | None] = [] + data_converter = dataclasses.replace( + temporalio.converter.default(), + payload_codec=ReferenceContextRecordingCodec(reference_contexts), + external_storage=ExternalStorage( + drivers=[InMemoryTestDriver()], + payload_size_threshold=512, + ), + ) + client = await env.connect_client(data_converter=data_converter) + workflow_id = f"workflow-{uuid.uuid4()}" + async with new_worker( + client, ExtStoreWorkflow, activities=[ext_store_activity] + ) as worker: + await client.execute_workflow( + ExtStoreWorkflow.run, + ExtStoreWorkflowInput( + input_data="wi" * 512, + activity_input_size=1024, + activity_output_size=1024, + output_size=1024, + ), + id=workflow_id, + task_queue=worker.task_queue, + ) + + workflow_contexts = [ + c for c in reference_contexts if isinstance(c, WorkflowSerializationContext) + ] + activity_contexts = [ + c for c in reference_contexts if isinstance(c, ActivitySerializationContext) + ] + # Workflow input (client) and workflow result (workflow worker). + assert len(workflow_contexts) == 2 + assert all(c.workflow_id == workflow_id for c in workflow_contexts) + # Activity input (workflow worker) and activity result (activity worker). + assert len(activity_contexts) == 2 + assert all( + c.workflow_id == workflow_id and not c.is_local for c in activity_contexts + ) + assert len(reference_contexts) == 4 + + async def test_replay_extstore_history_fails_with_empty_driver( env: WorkflowEnvironment, ) -> None: From 9206f6276b5c6f005576113d4c936aac033b41a8 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Tue, 13 Oct 2026 15:10:58 -0400 Subject: [PATCH 03/10] fix failing tests --- tests/worker/test_extstore.py | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index c2476cdf7f..8a3ace4cc2 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -469,17 +469,21 @@ async def test_extstore_reference_payloads_codec_encoded_in_history( reference_payloads = [] for event in history.events: if event.HasField("workflow_execution_started_event_attributes"): - attrs = event.workflow_execution_started_event_attributes - reference_payloads.append(attrs.input.payloads[0]) + reference_payloads.append( + event.workflow_execution_started_event_attributes.input.payloads[0] + ) elif event.HasField("activity_task_scheduled_event_attributes"): - attrs = event.activity_task_scheduled_event_attributes - reference_payloads.append(attrs.input.payloads[0]) + reference_payloads.append( + event.activity_task_scheduled_event_attributes.input.payloads[0] + ) elif event.HasField("activity_task_completed_event_attributes"): - attrs = event.activity_task_completed_event_attributes - reference_payloads.append(attrs.result.payloads[0]) + reference_payloads.append( + event.activity_task_completed_event_attributes.result.payloads[0] + ) elif event.HasField("workflow_execution_completed_event_attributes"): - attrs = event.workflow_execution_completed_event_attributes - reference_payloads.append(attrs.result.payloads[0]) + reference_payloads.append( + event.workflow_execution_completed_event_attributes.result.payloads[0] + ) assert len(reference_payloads) == 4 for payload in reference_payloads: assert payload.metadata["encoding"] == b"binary/test-encoded" From 3597327ca5d13f91804cacd848fdbeec6d7e6fcb Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Thu, 15 Oct 2026 12:38:54 -0400 Subject: [PATCH 04/10] make codec calls sequential and update tests. --- CHANGELOG.md | 79 +++++++++++++++++++++++++ temporalio/converter/_data_converter.py | 60 +++++++------------ tests/nexus/test_temporal_extstore.py | 76 ++++++++++++++++++++++-- tests/test_extstore.py | 37 ++++++++++++ tests/test_extstore_client.py | 64 +++++++++++++------- tests/worker/test_extstore.py | 57 ++++++++++++++++++ 6 files changed, 310 insertions(+), 63 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a7ef493d8..f7853f49c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,85 @@ This file contains assembled releases only. # Changelog +## [Unreleased] + +### Added + +- Worker now warns when an activity method is registered from the class + instead of from an instance (i.e. its first parameter is an unbound `self`). +- `workflow.new_random()` accepts an optional `name` that is mixed into the seed, so differently + named generators, and `workflow.random()`, produce different sequences. + +### Changed + +- Payload converters exposed by data converters and workflow/activity accessors + retain transfer type conversion, so direct use behaves consistently with SDK + serialization. + +### Deprecated + +### :boom: Breaking Changes + +- External storage reference payloads are now encoded with the configured payload codec and + marked with `__temporal_external_storage_reference` metadata, so claim data such as bucket + names and object keys no longer reaches workflow history in plain text. References written by + earlier versions are still read. Workers on earlier SDK versions cannot read encoded + references, so if a codec is configured, a workflow processed by a worker on this version + must not then be processed by a worker on an earlier version (for example, after a rollback or + during a mixed-version deployment). Without a codec, references are not encoded and remain + readable by earlier versions. +- `temporalio.contrib.google_adk_agents`: ADK-generated ids and retry jitter now draw from a + workflow-private deterministic stream (a `workflow.new_random()` per run) instead of + `workflow.random()`. A workflow started under 1.34.0 that generated + ADK ids or jitter (for example one waiting on a HITL response) may not replay + deterministically across this upgrade; drain such workflows or use worker versioning. +- The OpenAI Agents integration has moved to the independently versioned + [`temporalio-openai-agents`](https://pypi.org/project/temporalio-openai-agents/) + package. The existing `temporalio[openai-agents]` extra now installs that + package, and the old public `temporalio.contrib.openai_agents` imports + remain available at runtime and retain their static type information. + New code should depend on `temporalio-openai-agents` directly and import + `temporalio.openai_agents`. +- **Experimental**: Nexus Workflow Updates now require `wait_for_stage` to be explicitly set to `ACCEPTED`. +- `temporalio.contrib.opentelemetry`: removed `TemporalIdGenerator.seed_span_id()` and + `seed_trace_id()`. + +### Fixed + +- Experimental Workflow Streams background publishing now retries transient + signal delivery failures without delaying payload conversion errors. +- Fixed Pydantic constraints being silently ignored inside or outside workflow + sandboxes by passing through `annotated_types` by default. + +- `GoogleAdkPlugin`'s deterministic providers now work in read-only contexts (query handlers, + update validators), returning the workflow's deterministic time and fresh entropy without + touching the workflow's random state. +- `contrib.google_adk_agents`: agents with an `output_schema` no longer fail every workflow task + when calling the model. The schema type is now sent to the model activity as its JSON schema. + Custom Pydantic schema generation is preserved. + Integer-valued output enums are normalized to strings to match Google GenAI. + +- `temporalio.contrib.strands` activity and MCP tools now give the model the + Activity's failure message and expose its exception to after-tool hooks. +- Reject unsupported `InterpreterPoolExecutor` activity executors when creating + a worker instead of failing during activity execution. + +- Encoding a datetime search attribute without a timezone now raises + `ValueError("Timezone must be present on all search attribute dates")` on + the typed path, matching the deprecated untyped encoder, instead of sending + a naive ISO string that the server rejects with `BadSearchAttributes`. + +- `temporalio.contrib.opentelemetry`: `TracingInterceptor` and `OpenTelemetryInterceptor` no longer + log `Failed to detach context` when a context is torn down on a different thread while + OpenTelemetry's threading instrumentation (enabled by strands, among others) is active; a + context is now detached exactly when its token is still valid in the current + `contextvars.Context`, which it stays when a workflow resumes on another pool thread. + +- Decode enums that mix in `str` or `int` (for example `class Color(str, Enum)`) to the enum + instead of a list of characters or an error, and decode `dict` keys typed as an `int` enum. + +### Security + ## [1.34.0] - 2026-09-30 ### Added diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 7b47e9ee84..e99cd77517 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -2,7 +2,6 @@ from __future__ import annotations -import asyncio import dataclasses from collections.abc import Mapping, Sequence from dataclasses import dataclass @@ -346,22 +345,14 @@ async def _encode_reference_payload_sequence( ) -> list[temporalio.api.common.v1.Payload]: """Codec encode external storage reference payloads only.""" encoded_payloads = list(payloads) - payload_codec = self.payload_codec - if not payload_codec: - return encoded_payloads - reference_indices = [ - i for i, p in enumerate(encoded_payloads) if _is_reference_payload(p) - ] - if not reference_indices: + if not self.payload_codec: return encoded_payloads - encoded_references = await asyncio.gather( - *(payload_codec.encode([encoded_payloads[i]]) for i in reference_indices) - ) - for i, [encoded_reference] in zip(reference_indices, encoded_references): + for i, payload in enumerate(encoded_payloads): + if not _is_reference_payload(payload): + continue + [encoded_reference] = await self.payload_codec.encode([payload]) if not encoded_reference.external_payloads: - encoded_reference.external_payloads.extend( - encoded_payloads[i].external_payloads - ) + encoded_reference.external_payloads.extend(payload.external_payloads) encoded_reference.metadata[_REFERENCE_MARKER_KEY] = b"true" encoded_payloads[i] = encoded_reference return encoded_payloads @@ -371,37 +362,28 @@ async def _decode_reference_payload_sequence( ) -> list[temporalio.api.common.v1.Payload]: """Codec decode marked external storage reference payloads only.""" decoded_payloads = list(payloads) - marked_indices = [ - i for i, p in enumerate(decoded_payloads) if _has_reference_marker(p) - ] - if not marked_indices: - return decoded_payloads - payload_codec = self.payload_codec - if payload_codec: - encoded_references = [] - for i in marked_indices: + for i, payload in enumerate(decoded_payloads): + if not _has_reference_marker(payload): + continue + reference = payload + if self.payload_codec: encoded_reference = temporalio.api.common.v1.Payload() - encoded_reference.CopyFrom(decoded_payloads[i]) + encoded_reference.CopyFrom(payload) del encoded_reference.metadata[_REFERENCE_MARKER_KEY] del encoded_reference.external_payloads[:] - encoded_references.append(encoded_reference) - decoded = await asyncio.gather( - *(payload_codec.decode([r]) for r in encoded_references) - ) - references = [reference for [reference] in decoded] - else: - references = [decoded_payloads[i] for i in marked_indices] - for i, decoded_reference in zip(marked_indices, references): - if not _is_reference_payload(decoded_reference): + [reference] = await self.payload_codec.decode([encoded_reference]) + if not _is_reference_payload(reference): + if not self.external_storage: + raise RuntimeError( + "[TMPRL1105] Detected externally stored payload(s) but external storage is not configured." + ) raise RuntimeError( "[TMPRL1106] Encountered an encoded external storage reference that could not be decoded. " "Check that the payload codec used to write it is configured." ) - if not decoded_reference.external_payloads: - decoded_reference.external_payloads.extend( - decoded_payloads[i].external_payloads - ) - decoded_payloads[i] = decoded_reference + if not reference.external_payloads: + reference.external_payloads.extend(payload.external_payloads) + decoded_payloads[i] = reference return decoded_payloads async def _external_retrieve_payload_sequence( diff --git a/tests/nexus/test_temporal_extstore.py b/tests/nexus/test_temporal_extstore.py index f23a692597..ef2f22d153 100644 --- a/tests/nexus/test_temporal_extstore.py +++ b/tests/nexus/test_temporal_extstore.py @@ -30,6 +30,7 @@ from temporalio.client import Client, WorkflowFailureError from temporalio.converter import ( ExternalStorage, + PayloadCodec, StorageDriverClaim, StorageDriverRetrieveContext, StorageDriverStoreContext, @@ -39,7 +40,7 @@ from temporalio.types import MethodAsyncSingleParam from temporalio.worker import UnsandboxedWorkflowRunner, Worker from tests.helpers.nexus import make_nexus_endpoint_name -from tests.test_extstore import InMemoryTestDriver +from tests.test_extstore import InMemoryTestDriver, RecordingPayloadCodec # Cloud CI's namespace credentials cannot manage Nexus endpoints. # See https://github.com/temporalio/sdk-python/issues/1704. @@ -137,11 +138,14 @@ async def retrieve( def _client_with_extstore( - env: WorkflowEnvironment, driver: InMemoryTestDriver + env: WorkflowEnvironment, + driver: InMemoryTestDriver, + payload_codec: PayloadCodec | None = None, ) -> Client: config = env.client.config() config["data_converter"] = dataclasses.replace( temporalio.converter.default(), + payload_codec=payload_codec, external_storage=ExternalStorage( drivers=[driver], payload_size_threshold=PAYLOAD_SIZE_THRESHOLD, @@ -154,8 +158,11 @@ async def _run_caller( env: WorkflowEnvironment, driver: InMemoryTestDriver, workflow_run: MethodAsyncSingleParam[Any, str, int], + *, + payload_codec: PayloadCodec | None = None, + workflow_id: str | None = None, ) -> int: - client = _client_with_extstore(env, driver) + client = _client_with_extstore(env, driver, payload_codec) task_queue = str(uuid.uuid4()) async with Worker( client, @@ -170,7 +177,7 @@ async def _run_caller( return await client.execute_workflow( workflow_run, task_queue, - id=str(uuid.uuid4()), + id=workflow_id or str(uuid.uuid4()), task_queue=task_queue, execution_timeout=timedelta(seconds=30), ) @@ -213,6 +220,67 @@ async def test_nexus_operation_sync_result_offloaded_and_retrieved( assert driver._retrieve_calls >= 1 +def _assert_encoded_reference(payload: Payload) -> None: + assert payload.metadata["encoding"] == b"binary/test-encoded" + assert payload.metadata["__temporal_external_storage_reference"] == b"true" + + +async def test_nexus_operation_input_reference_encoded_with_codec( + env: WorkflowEnvironment, +): + """With a codec, the operation input reference is encoded, and the Nexus + worker decodes it before retrieving the input.""" + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + + driver = InMemoryTestDriver() + workflow_id = str(uuid.uuid4()) + result = await _run_caller( + env, + driver, + SizeOpCallerWorkflow.run, + payload_codec=RecordingPayloadCodec("binary/test-encoded"), + workflow_id=workflow_id, + ) + + assert result == PAYLOAD_SIZE + history = await env.client.get_workflow_handle(workflow_id).fetch_history() + [scheduled] = [ + e.nexus_operation_scheduled_event_attributes + for e in history.events + if e.HasField("nexus_operation_scheduled_event_attributes") + ] + _assert_encoded_reference(scheduled.input) + + +async def test_nexus_operation_result_reference_encoded_with_codec( + env: WorkflowEnvironment, +): + """With a codec, the Nexus worker encodes the reference to an offloaded + operation result, and the caller decodes it.""" + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + + driver = InMemoryTestDriver() + workflow_id = str(uuid.uuid4()) + result = await _run_caller( + env, + driver, + BigResultOpCallerWorkflow.run, + payload_codec=RecordingPayloadCodec("binary/test-encoded"), + workflow_id=workflow_id, + ) + + assert result == PAYLOAD_SIZE + history = await env.client.get_workflow_handle(workflow_id).fetch_history() + [completed] = [ + e.nexus_operation_completed_event_attributes + for e in history.events + if e.HasField("nexus_operation_completed_event_attributes") + ] + _assert_encoded_reference(completed.result) + + async def test_nexus_operation_transient_retrieve_failure_recovers( env: WorkflowEnvironment, ): diff --git a/tests/test_extstore.py b/tests/test_extstore.py index 2c8c173de2..b101e4bcfc 100644 --- a/tests/test_extstore.py +++ b/tests/test_extstore.py @@ -581,6 +581,43 @@ async def test_failure_reference_payload_encoded(self): assert isinstance(error, ApplicationError) assert error.details == ("x" * 200,) + async def test_multiple_stored_payloads_encode_each_reference(self): + """When several payloads are stored together, every reference payload + is encoded and marked, and all round-trip.""" + driver = InMemoryTestDriver() + converter = DataConverter( + payload_codec=RecordingPayloadCodec("binary/dc-encoded"), + external_storage=ExternalStorage( + drivers=[driver], + payload_size_threshold=50, + ), + ) + values = ["a" * 200, "b" * 200, "c" * 200] + + encoded = await converter.encode(values) + + assert driver._store_calls == 1 + for payload in encoded: + assert payload.metadata.get("encoding") == b"binary/dc-encoded" + assert ( + payload.metadata.get("__temporal_external_storage_reference") == b"true" + ) + assert payload.external_payloads + assert await converter.decode(encoded, [str, str, str]) == values + + async def test_marked_reference_without_external_storage_fails(self): + """A marked reference read by a converter without external storage + reports that external storage is not configured.""" + encoded = await DataConverter( + payload_codec=RecordingPayloadCodec("binary/dc-encoded"), + external_storage=ExternalStorage( + drivers=[InMemoryTestDriver()], payload_size_threshold=50 + ), + ).encode(["x" * 200]) + + with pytest.raises(RuntimeError, match=r"\[TMPRL1105\]"): + await DataConverter().decode(encoded, [str]) + class TestMultiDriver: """Tests for ExternalStorage with multiple drivers.""" diff --git a/tests/test_extstore_client.py b/tests/test_extstore_client.py index ebdb6e8fa6..8ea6a89eb0 100644 --- a/tests/test_extstore_client.py +++ b/tests/test_extstore_client.py @@ -21,6 +21,7 @@ import temporalio.client import temporalio.converter from temporalio.api.common.v1 import Payload +from temporalio.api.sdk.v1.external_storage_pb2 import ExternalStorageReference from temporalio.client import ( Client, Schedule, @@ -29,7 +30,11 @@ WorkflowFailureError, ) from temporalio.common import HeaderCodecBehavior -from temporalio.converter import DataConverter, ExternalStorage +from temporalio.converter import ( + DataConverter, + ExternalStorage, + JSONProtoPayloadConverter, +) from temporalio.exceptions import ApplicationError from temporalio.service import ServiceClient from tests.test_extstore import InMemoryTestDriver, RecordingPayloadCodec @@ -80,15 +85,22 @@ def _client( ) -def _assert_encoded_reference(payload: Payload) -> None: +async def _assert_encoded_reference( + data_converter: DataConverter, payload: Payload +) -> None: assert payload.metadata["encoding"] == CODEC_ENCODING assert payload.metadata["__temporal_external_storage_reference"] == b"true" assert payload.external_payloads + [reference] = await data_converter._decode_reference_payload_sequence([payload]) + assert isinstance( + JSONProtoPayloadConverter().from_payload(reference, ExternalStorageReference), + ExternalStorageReference, + ) async def _encoded_reference(data_converter: DataConverter, value: Any) -> Payload: [payload] = await data_converter.encode([value]) - _assert_encoded_reference(payload) + await _assert_encoded_reference(data_converter, payload) return payload @@ -115,8 +127,8 @@ async def test_start_workflow_encodes_input_and_memo_references() -> None: ) req = service.requests["start_workflow_execution"] - _assert_encoded_reference(req.input.payloads[0]) - _assert_encoded_reference(req.memo.fields["key"]) + await _assert_encoded_reference(client.data_converter, req.input.payloads[0]) + await _assert_encoded_reference(client.data_converter, req.memo.fields["key"]) async def test_signal_encodes_input_reference() -> None: @@ -129,8 +141,9 @@ async def test_signal_encodes_input_reference() -> None: await client.get_workflow_handle("wf").signal("signal", LARGE) - _assert_encoded_reference( - service.requests["signal_workflow_execution"].input.payloads[0] + await _assert_encoded_reference( + client.data_converter, + service.requests["signal_workflow_execution"].input.payloads[0], ) @@ -154,8 +167,8 @@ async def test_signal_with_start_encodes_input_references() -> None: ) req = service.requests["signal_with_start_workflow_execution"] - _assert_encoded_reference(req.input.payloads[0]) - _assert_encoded_reference(req.signal_input.payloads[0]) + await _assert_encoded_reference(client.data_converter, req.input.payloads[0]) + await _assert_encoded_reference(client.data_converter, req.signal_input.payloads[0]) async def test_update_encodes_args_and_decodes_result_references() -> None: @@ -179,7 +192,9 @@ async def test_update_encodes_args_and_decodes_result_references() -> None: assert value == "update-result" * 20 req = service.requests["update_workflow_execution"] - _assert_encoded_reference(req.request.input.args.payloads[0]) + await _assert_encoded_reference( + client.data_converter, req.request.input.args.payloads[0] + ) async def test_query_encodes_args_and_decodes_result_references() -> None: @@ -199,8 +214,9 @@ async def test_query_encodes_args_and_decodes_result_references() -> None: ) assert value == "query-result" * 20 - _assert_encoded_reference( - service.requests["query_workflow"].query.query_args.payloads[0] + await _assert_encoded_reference( + client.data_converter, + service.requests["query_workflow"].query.query_args.payloads[0], ) @@ -221,7 +237,9 @@ async def test_create_schedule_encodes_action_input_reference() -> None: ) req = service.requests["create_schedule"] - _assert_encoded_reference(req.schedule.action.start_workflow.input.payloads[0]) + await _assert_encoded_reference( + client.data_converter, req.schedule.action.start_workflow.input.payloads[0] + ) async def test_async_activity_completion_encodes_references() -> None: @@ -241,14 +259,18 @@ async def test_async_activity_completion_encodes_references() -> None: await handle.complete(LARGE) await handle.fail(ApplicationError("failed", LARGE)) - _assert_encoded_reference( - service.requests["record_activity_task_heartbeat_by_id"].details.payloads[0] + await _assert_encoded_reference( + client.data_converter, + service.requests["record_activity_task_heartbeat_by_id"].details.payloads[0], ) - _assert_encoded_reference( - service.requests["respond_activity_task_completed_by_id"].result.payloads[0] + await _assert_encoded_reference( + client.data_converter, + service.requests["respond_activity_task_completed_by_id"].result.payloads[0], ) failure = service.requests["respond_activity_task_failed_by_id"].failure - _assert_encoded_reference(failure.application_failure_info.details.payloads[0]) + await _assert_encoded_reference( + client.data_converter, failure.application_failure_info.details.payloads[0] + ) class _LargeHeaderInterceptor(temporalio.client.Interceptor): @@ -285,7 +307,7 @@ async def test_start_workflow_encodes_header_reference_when_headers_use_codec() await client.start_workflow("Workflow", id="wf", task_queue="tq") req = service.requests["start_workflow_execution"] - _assert_encoded_reference(req.header.fields["large"]) + await _assert_encoded_reference(client.data_converter, req.header.fields["large"]) async def test_workflow_result_decodes_reference() -> None: @@ -314,7 +336,9 @@ async def test_workflow_failure_decodes_detail_reference() -> None: data_converter = _data_converter() failure = temporalio.api.failure.v1.Failure() await data_converter.encode_failure(ApplicationError("failed", LARGE), failure) - _assert_encoded_reference(failure.application_failure_info.details.payloads[0]) + await _assert_encoded_reference( + data_converter, failure.application_failure_info.details.payloads[0] + ) service = _FakeWorkflowService( { "get_workflow_execution_history": _history_response( diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index 8a3ace4cc2..216f4aebc8 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -575,6 +575,63 @@ async def test_extstore_reference_payloads_use_payload_serialization_context( assert len(reference_contexts) == 4 +async def test_replay_history_with_references_written_before_encoding( + env: WorkflowEnvironment, +) -> None: + """A history whose references were written unencoded and unmarked, as by + earlier SDK versions, still replays with a codec configured.""" + data_converter = dataclasses.replace( + temporalio.converter.default(), + payload_codec=RecordingPayloadCodec("binary/test-encoded"), + external_storage=ExternalStorage( + drivers=[InMemoryTestDriver()], + payload_size_threshold=512, + ), + ) + client = await env.connect_client(data_converter=data_converter) + async with new_worker( + client, ExtStoreWorkflow, activities=[ext_store_activity] + ) as worker: + handle = await client.start_workflow( + ExtStoreWorkflow.run, + ExtStoreWorkflowInput( + input_data="wi" * 512, + activity_input_size=1024, + activity_output_size=1024, + output_size=1024, + ), + id=f"workflow-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + await handle.result() + history = await handle.fetch_history() + + # Rewrite each reference into the earlier format. The stored payloads are + # unchanged, so the rewritten references still resolve. + rewritten = 0 + for event in history.events: + for payloads in ( + event.workflow_execution_started_event_attributes.input.payloads, + event.activity_task_scheduled_event_attributes.input.payloads, + event.activity_task_completed_event_attributes.result.payloads, + event.workflow_execution_completed_event_attributes.result.payloads, + ): + for payload in payloads: + if "__temporal_external_storage_reference" not in payload.metadata: + continue + [reference] = await data_converter._decode_reference_payload_sequence( + [payload] + ) + del reference.metadata["__temporal_external_storage_reference"] + payload.CopyFrom(reference) + rewritten += 1 + assert rewritten == 4 + + await Replayer( + workflows=[ExtStoreWorkflow], data_converter=data_converter + ).replay_workflow(history) + + async def test_replay_extstore_history_fails_with_empty_driver( env: WorkflowEnvironment, ) -> None: From a6ade423681684a95a12c8be0bf98b011f72ec31 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Thu, 15 Oct 2026 12:53:28 -0400 Subject: [PATCH 05/10] update payload decoding logic to read more clearly --- temporalio/converter/_data_converter.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index e99cd77517..9a8ed41980 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -365,6 +365,10 @@ async def _decode_reference_payload_sequence( for i, payload in enumerate(decoded_payloads): if not _has_reference_marker(payload): continue + if not self.external_storage: + raise RuntimeError( + "[TMPRL1105] Detected externally stored payload(s) but external storage is not configured." + ) reference = payload if self.payload_codec: encoded_reference = temporalio.api.common.v1.Payload() @@ -373,10 +377,6 @@ async def _decode_reference_payload_sequence( del encoded_reference.external_payloads[:] [reference] = await self.payload_codec.decode([encoded_reference]) if not _is_reference_payload(reference): - if not self.external_storage: - raise RuntimeError( - "[TMPRL1105] Detected externally stored payload(s) but external storage is not configured." - ) raise RuntimeError( "[TMPRL1106] Encountered an encoded external storage reference that could not be decoded. " "Check that the payload codec used to write it is configured." From 2796eed09a96d96f0fe086047993c9875ba85e42 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Thu, 15 Oct 2026 17:27:22 -0400 Subject: [PATCH 06/10] added new replay tests that use committed history files. updated changelog --- CHANGELOG.md | 79 ------ .../breaking-changes/discerning_murlock.md | 1 + tests/test_extstore.py | 17 +- tests/worker/test_extstore.py | 124 +++++---- ...re_references_before_encoding_history.json | 254 ++++++++++++++++++ ...ences_before_encoding_stored_payloads.json | 26 ++ 6 files changed, 366 insertions(+), 135 deletions(-) create mode 100644 changelog/breaking-changes/discerning_murlock.md create mode 100644 tests/worker/test_extstore_references_before_encoding_history.json create mode 100644 tests/worker/test_extstore_references_before_encoding_stored_payloads.json diff --git a/CHANGELOG.md b/CHANGELOG.md index f7853f49c8..3a7ef493d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,85 +7,6 @@ This file contains assembled releases only. # Changelog -## [Unreleased] - -### Added - -- Worker now warns when an activity method is registered from the class - instead of from an instance (i.e. its first parameter is an unbound `self`). -- `workflow.new_random()` accepts an optional `name` that is mixed into the seed, so differently - named generators, and `workflow.random()`, produce different sequences. - -### Changed - -- Payload converters exposed by data converters and workflow/activity accessors - retain transfer type conversion, so direct use behaves consistently with SDK - serialization. - -### Deprecated - -### :boom: Breaking Changes - -- External storage reference payloads are now encoded with the configured payload codec and - marked with `__temporal_external_storage_reference` metadata, so claim data such as bucket - names and object keys no longer reaches workflow history in plain text. References written by - earlier versions are still read. Workers on earlier SDK versions cannot read encoded - references, so if a codec is configured, a workflow processed by a worker on this version - must not then be processed by a worker on an earlier version (for example, after a rollback or - during a mixed-version deployment). Without a codec, references are not encoded and remain - readable by earlier versions. -- `temporalio.contrib.google_adk_agents`: ADK-generated ids and retry jitter now draw from a - workflow-private deterministic stream (a `workflow.new_random()` per run) instead of - `workflow.random()`. A workflow started under 1.34.0 that generated - ADK ids or jitter (for example one waiting on a HITL response) may not replay - deterministically across this upgrade; drain such workflows or use worker versioning. -- The OpenAI Agents integration has moved to the independently versioned - [`temporalio-openai-agents`](https://pypi.org/project/temporalio-openai-agents/) - package. The existing `temporalio[openai-agents]` extra now installs that - package, and the old public `temporalio.contrib.openai_agents` imports - remain available at runtime and retain their static type information. - New code should depend on `temporalio-openai-agents` directly and import - `temporalio.openai_agents`. -- **Experimental**: Nexus Workflow Updates now require `wait_for_stage` to be explicitly set to `ACCEPTED`. -- `temporalio.contrib.opentelemetry`: removed `TemporalIdGenerator.seed_span_id()` and - `seed_trace_id()`. - -### Fixed - -- Experimental Workflow Streams background publishing now retries transient - signal delivery failures without delaying payload conversion errors. -- Fixed Pydantic constraints being silently ignored inside or outside workflow - sandboxes by passing through `annotated_types` by default. - -- `GoogleAdkPlugin`'s deterministic providers now work in read-only contexts (query handlers, - update validators), returning the workflow's deterministic time and fresh entropy without - touching the workflow's random state. -- `contrib.google_adk_agents`: agents with an `output_schema` no longer fail every workflow task - when calling the model. The schema type is now sent to the model activity as its JSON schema. - Custom Pydantic schema generation is preserved. - Integer-valued output enums are normalized to strings to match Google GenAI. - -- `temporalio.contrib.strands` activity and MCP tools now give the model the - Activity's failure message and expose its exception to after-tool hooks. -- Reject unsupported `InterpreterPoolExecutor` activity executors when creating - a worker instead of failing during activity execution. - -- Encoding a datetime search attribute without a timezone now raises - `ValueError("Timezone must be present on all search attribute dates")` on - the typed path, matching the deprecated untyped encoder, instead of sending - a naive ISO string that the server rejects with `BadSearchAttributes`. - -- `temporalio.contrib.opentelemetry`: `TracingInterceptor` and `OpenTelemetryInterceptor` no longer - log `Failed to detach context` when a context is torn down on a different thread while - OpenTelemetry's threading instrumentation (enabled by strands, among others) is active; a - context is now detached exactly when its token is still valid in the current - `contextvars.Context`, which it stays when a workflow resumes on another pool thread. - -- Decode enums that mix in `str` or `int` (for example `class Color(str, Enum)`) to the enum - instead of a list of characters or an error, and decode `dict` keys typed as an `int` enum. - -### Security - ## [1.34.0] - 2026-09-30 ### Added diff --git a/changelog/breaking-changes/discerning_murlock.md b/changelog/breaking-changes/discerning_murlock.md new file mode 100644 index 0000000000..1e93b03bd2 --- /dev/null +++ b/changelog/breaking-changes/discerning_murlock.md @@ -0,0 +1 @@ +External storage reference payloads are now encoded with the configured payload codec and marked with `__temporal_external_storage_reference` metadata, so claim data such as bucket names and object keys no longer reaches workflow history in plain text. References written by earlier versions are still read. Workers on earlier SDK versions cannot read encoded references, so if a codec is configured, a workflow processed by a worker on this version must not then be processed by a worker on an earlier version (for example, after a rollback or during a mixed-version deployment). Without a codec, references are not encoded and remain readable by earlier versions. diff --git a/tests/test_extstore.py b/tests/test_extstore.py index b101e4bcfc..95498030af 100644 --- a/tests/test_extstore.py +++ b/tests/test_extstore.py @@ -583,26 +583,33 @@ async def test_failure_reference_payload_encoded(self): async def test_multiple_stored_payloads_encode_each_reference(self): """When several payloads are stored together, every reference payload - is encoded and marked, and all round-trip.""" + is encoded and marked, an inline payload between them is not marked, + and all round-trip.""" driver = InMemoryTestDriver() converter = DataConverter( payload_codec=RecordingPayloadCodec("binary/dc-encoded"), external_storage=ExternalStorage( drivers=[driver], - payload_size_threshold=50, + # The codec-encoded "inline" value is ~67 bytes, below this. + payload_size_threshold=100, ), ) - values = ["a" * 200, "b" * 200, "c" * 200] + values = ["a" * 200, "inline", "c" * 200] encoded = await converter.encode(values) assert driver._store_calls == 1 for payload in encoded: assert payload.metadata.get("encoding") == b"binary/dc-encoded" + for reference in (encoded[0], encoded[2]): assert ( - payload.metadata.get("__temporal_external_storage_reference") == b"true" + reference.metadata.get("__temporal_external_storage_reference") + == b"true" ) - assert payload.external_payloads + assert reference.external_payloads + inline = encoded[1] + assert "__temporal_external_storage_reference" not in inline.metadata + assert not inline.external_payloads assert await converter.decode(encoded, [str, str, str]) == values async def test_marked_reference_without_external_storage_fails(self): diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index 216f4aebc8..b4268b490b 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -1,12 +1,15 @@ import contextlib import dataclasses +import json import uuid from collections.abc import Iterator, Sequence from dataclasses import dataclass from datetime import timedelta +from pathlib import Path from unittest import mock import pytest +from google.protobuf import json_format import temporalio import temporalio.bridge.client @@ -16,7 +19,12 @@ import temporalio.converter from temporalio import activity, workflow from temporalio.api.common.v1 import Payload -from temporalio.client import Client, WorkflowFailureError, WorkflowHandle +from temporalio.client import ( + Client, + WorkflowFailureError, + WorkflowHandle, + WorkflowHistory, +) from temporalio.common import RetryPolicy from temporalio.converter import ( ActivitySerializationContext, @@ -575,61 +583,75 @@ async def test_extstore_reference_payloads_use_payload_serialization_context( assert len(reference_contexts) == 4 -async def test_replay_history_with_references_written_before_encoding( - env: WorkflowEnvironment, -) -> None: - """A history whose references were written unencoded and unmarked, as by - earlier SDK versions, still replays with a codec configured.""" - data_converter = dataclasses.replace( - temporalio.converter.default(), - payload_codec=RecordingPayloadCodec("binary/test-encoded"), - external_storage=ExternalStorage( - drivers=[InMemoryTestDriver()], - payload_size_threshold=512, - ), - ) - client = await env.connect_client(data_converter=data_converter) - async with new_worker( - client, ExtStoreWorkflow, activities=[ext_store_activity] - ) as worker: - handle = await client.start_workflow( - ExtStoreWorkflow.run, - ExtStoreWorkflowInput( - input_data="wi" * 512, - activity_input_size=1024, - activity_output_size=1024, - output_size=1024, - ), - id=f"workflow-{uuid.uuid4()}", - task_queue=worker.task_queue, +@workflow.defn +class ReferencesBeforeEncodingWorkflow: + """Replays the committed history in + test_extstore_references_before_encoding_history.json.""" + + @workflow.run + async def run(self, input: str) -> str: + await workflow.execute_activity( + "references_before_encoding_activity", + "ai" * 512, + start_to_close_timeout=timedelta(seconds=30), ) - await handle.result() - history = await handle.fetch_history() + return "wo" * 512 - # Rewrite each reference into the earlier format. The stored payloads are - # unchanged, so the rewritten references still resolve. - rewritten = 0 - for event in history.events: - for payloads in ( - event.workflow_execution_started_event_attributes.input.payloads, - event.activity_task_scheduled_event_attributes.input.payloads, - event.activity_task_completed_event_attributes.result.payloads, - event.workflow_execution_completed_event_attributes.result.payloads, - ): - for payload in payloads: - if "__temporal_external_storage_reference" not in payload.metadata: - continue - [reference] = await data_converter._decode_reference_payload_sequence( - [payload] - ) - del reference.metadata["__temporal_external_storage_reference"] - payload.CopyFrom(reference) - rewritten += 1 - assert rewritten == 4 + +async def test_replay_history_with_references_written_before_encoding() -> None: + """A history recorded with temporalio 1.34.0, whose references were written + unencoded and unmarked, still replays with a codec configured. + + The history and the payloads its references point to were recorded with + ReferencesBeforeEncodingWorkflow, a payload codec matching + RecordingPayloadCodec("binary/test-encoded"), and a driver storing claims the + same way as InMemoryTestDriver, with a 512-byte storage threshold. + """ + fixtures = Path(__file__).parent + history = WorkflowHistory.from_json( + "references-before-encoding", + ( + fixtures / "test_extstore_references_before_encoding_history.json" + ).read_text(), + ) + stored_payloads = json.loads( + ( + fixtures / "test_extstore_references_before_encoding_stored_payloads.json" + ).read_text() + ) + driver = InMemoryTestDriver() + for key, payload_json in stored_payloads.items(): + driver._storage[key] = json_format.ParseDict( + payload_json, Payload() + ).SerializeToString() + + references = [ + payload + for event in history.events + for payload in ( + *event.workflow_execution_started_event_attributes.input.payloads, + *event.activity_task_scheduled_event_attributes.input.payloads, + *event.activity_task_completed_event_attributes.result.payloads, + *event.workflow_execution_completed_event_attributes.result.payloads, + ) + ] + assert len(references) == 4 + for reference in references: + assert reference.metadata["encoding"] == b"json/protobuf" + assert "__temporal_external_storage_reference" not in reference.metadata await Replayer( - workflows=[ExtStoreWorkflow], data_converter=data_converter + workflows=[ReferencesBeforeEncodingWorkflow], + data_converter=dataclasses.replace( + temporalio.converter.default(), + payload_codec=RecordingPayloadCodec("binary/test-encoded"), + external_storage=ExternalStorage( + drivers=[driver], + payload_size_threshold=512, + ), + ), ).replay_workflow(history) + assert driver._retrieve_calls > 0 async def test_replay_extstore_history_fails_with_empty_driver( diff --git a/tests/worker/test_extstore_references_before_encoding_history.json b/tests/worker/test_extstore_references_before_encoding_history.json new file mode 100644 index 0000000000..4b46f8d0e9 --- /dev/null +++ b/tests/worker/test_extstore_references_before_encoding_history.json @@ -0,0 +1,254 @@ +{ + "events": [ + { + "eventId": "1", + "eventTime": "2026-10-15T18:04:15.639732Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED", + "taskId": "1048587", + "workflowExecutionStartedEventAttributes": { + "workflowType": { + "name": "ReferencesBeforeEncodingWorkflow" + }, + "taskQueue": { + "name": "references-before-encoding", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "input": { + "payloads": [ + { + "metadata": { + "messageType": "dGVtcG9yYWwuYXBpLnNkay52MS5FeHRlcm5hbFN0b3JhZ2VSZWZlcmVuY2U=", + "encoding": "anNvbi9wcm90b2J1Zg==" + }, + "data": "eyJjbGFpbURhdGEiOnsia2V5IjoicGF5bG9hZC0wIn0sImRyaXZlck5hbWUiOiJ0ZXN0LWRyaXZlciJ9", + "externalPayloads": [ + { + "sizeBytes": "1089" + } + ] + } + ] + }, + "workflowTaskTimeout": "10s", + "originalExecutionRunId": "01a140bc-7797-7b27-9a71-bfe54f3a1f6a", + "identity": "test-identity", + "firstExecutionRunId": "01a140bc-7797-7b27-9a71-bfe54f3a1f6a", + "attempt": 1, + "firstWorkflowTaskBackoff": "0s", + "workflowId": "references-before-encoding", + "priority": {} + } + }, + { + "eventId": "2", + "eventTime": "2026-10-15T18:04:15.639789Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048588", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "references-before-encoding", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "3", + "eventTime": "2026-10-15T18:04:15.642928Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048593", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "2", + "identity": "test-identity", + "requestId": "7af88403-fe6d-4f34-9219-d84cd0e8c922", + "historySizeBytes": "474", + "workerVersion": { + "buildId": "719681aca6c15dba2948274d63ba289f" + } + } + }, + { + "eventId": "4", + "eventTime": "2026-10-15T18:04:15.648786Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048598", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "2", + "startedEventId": "3", + "identity": "test-identity", + "workerVersion": { + "buildId": "719681aca6c15dba2948274d63ba289f" + }, + "sdkMetadata": { + "coreUsedFlags": [ + 1, + 3, + 2 + ], + "sdkName": "temporal-python", + "sdkVersion": "1.34.0" + }, + "meteringMetadata": {} + } + }, + { + "eventId": "5", + "eventTime": "2026-10-15T18:04:15.648861Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_SCHEDULED", + "taskId": "1048599", + "activityTaskScheduledEventAttributes": { + "activityId": "1", + "activityType": { + "name": "references_before_encoding_activity" + }, + "taskQueue": { + "name": "references-before-encoding", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "header": {}, + "input": { + "payloads": [ + { + "metadata": { + "messageType": "dGVtcG9yYWwuYXBpLnNkay52MS5FeHRlcm5hbFN0b3JhZ2VSZWZlcmVuY2U=", + "encoding": "anNvbi9wcm90b2J1Zg==" + }, + "data": "eyJjbGFpbURhdGEiOnsia2V5IjoicGF5bG9hZC0xIn0sImRyaXZlck5hbWUiOiJ0ZXN0LWRyaXZlciJ9", + "externalPayloads": [ + { + "sizeBytes": "1089" + } + ] + } + ] + }, + "scheduleToCloseTimeout": "0s", + "scheduleToStartTimeout": "0s", + "startToCloseTimeout": "30s", + "heartbeatTimeout": "0s", + "workflowTaskCompletedEventId": "4", + "retryPolicy": { + "initialInterval": "1s", + "backoffCoefficient": 2.0, + "maximumInterval": "100s" + }, + "useWorkflowBuildId": true, + "priority": {} + } + }, + { + "eventId": "6", + "eventTime": "2026-10-15T18:04:15.649402Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_STARTED", + "taskId": "1048603", + "activityTaskStartedEventAttributes": { + "scheduledEventId": "5", + "identity": "test-identity", + "requestId": "1deae0f7-9ca9-47c1-b260-d5dccb9e24ee", + "attempt": 1, + "workerVersion": { + "buildId": "719681aca6c15dba2948274d63ba289f" + } + } + }, + { + "eventId": "7", + "eventTime": "2026-10-15T18:04:15.650894Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_COMPLETED", + "taskId": "1048604", + "activityTaskCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "messageType": "dGVtcG9yYWwuYXBpLnNkay52MS5FeHRlcm5hbFN0b3JhZ2VSZWZlcmVuY2U=", + "encoding": "anNvbi9wcm90b2J1Zg==" + }, + "data": "eyJjbGFpbURhdGEiOnsia2V5IjoicGF5bG9hZC0yIn0sImRyaXZlck5hbWUiOiJ0ZXN0LWRyaXZlciJ9", + "externalPayloads": [ + { + "sizeBytes": "1089" + } + ] + } + ] + }, + "scheduledEventId": "5", + "startedEventId": "6", + "identity": "test-identity" + } + }, + { + "eventId": "8", + "eventTime": "2026-10-15T18:04:15.650917Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048605", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "test-identity-sticky", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "references-before-encoding" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "9", + "eventTime": "2026-10-15T18:04:15.651608Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048609", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "8", + "identity": "test-identity", + "requestId": "b4cf3329-413e-4a7c-b07a-8c7f54660de2", + "historySizeBytes": "1549", + "workerVersion": { + "buildId": "719681aca6c15dba2948274d63ba289f" + } + } + }, + { + "eventId": "10", + "eventTime": "2026-10-15T18:04:15.654284Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048613", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "8", + "startedEventId": "9", + "identity": "test-identity", + "workerVersion": { + "buildId": "719681aca6c15dba2948274d63ba289f" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "11", + "eventTime": "2026-10-15T18:04:15.654332Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED", + "taskId": "1048614", + "workflowExecutionCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "messageType": "dGVtcG9yYWwuYXBpLnNkay52MS5FeHRlcm5hbFN0b3JhZ2VSZWZlcmVuY2U=", + "encoding": "anNvbi9wcm90b2J1Zg==" + }, + "data": "eyJjbGFpbURhdGEiOnsia2V5IjoicGF5bG9hZC0zIn0sImRyaXZlck5hbWUiOiJ0ZXN0LWRyaXZlciJ9", + "externalPayloads": [ + { + "sizeBytes": "1089" + } + ] + } + ] + }, + "workflowTaskCompletedEventId": "10" + } + } + ] +} diff --git a/tests/worker/test_extstore_references_before_encoding_stored_payloads.json b/tests/worker/test_extstore_references_before_encoding_stored_payloads.json new file mode 100644 index 0000000000..03768b539c --- /dev/null +++ b/tests/worker/test_extstore_references_before_encoding_stored_payloads.json @@ -0,0 +1,26 @@ +{ + "payload-0": { + "metadata": { + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChYKCGVuY29kaW5nEgpqc29uL3BsYWluEoIIIndpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2ki" + }, + "payload-1": { + "metadata": { + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChYKCGVuY29kaW5nEgpqc29uL3BsYWluEoIIImFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWki" + }, + "payload-2": { + "metadata": { + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChYKCGVuY29kaW5nEgpqc29uL3BsYWluEoIIImFvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW8i" + }, + "payload-3": { + "metadata": { + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChYKCGVuY29kaW5nEgpqc29uL3BsYWluEoIIIndvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd28i" + } +} From 78f2861b6a8321fa30b71081786e4536d45b9dd4 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Fri, 16 Oct 2026 14:07:59 -0400 Subject: [PATCH 07/10] Disable encoding external storage references. Decoding is enabled. Add new tests. --- changelog/added/discerning_murlock.md | 1 + .../breaking-changes/discerning_murlock.md | 1 - temporalio/converter/_data_converter.py | 6 ++- temporalio/converter/_extstore.py | 11 ++++- tests/conftest.py | 10 +++++ tests/nexus/test_temporal_extstore.py | 2 + tests/test_extstore.py | 43 +++++++++++++++++++ tests/test_extstore_client.py | 2 + tests/worker/test_extstore.py | 2 + 9 files changed, 74 insertions(+), 4 deletions(-) create mode 100644 changelog/added/discerning_murlock.md delete mode 100644 changelog/breaking-changes/discerning_murlock.md diff --git a/changelog/added/discerning_murlock.md b/changelog/added/discerning_murlock.md new file mode 100644 index 0000000000..7e66851aa0 --- /dev/null +++ b/changelog/added/discerning_murlock.md @@ -0,0 +1 @@ +Workers and clients can now read external storage reference payloads that were encoded with the configured payload codec and marked with `__temporal_external_storage_reference` metadata. This release does not write encoded references yet; a later release will, so upgrade all workers before then. diff --git a/changelog/breaking-changes/discerning_murlock.md b/changelog/breaking-changes/discerning_murlock.md deleted file mode 100644 index 1e93b03bd2..0000000000 --- a/changelog/breaking-changes/discerning_murlock.md +++ /dev/null @@ -1 +0,0 @@ -External storage reference payloads are now encoded with the configured payload codec and marked with `__temporal_external_storage_reference` metadata, so claim data such as bucket names and object keys no longer reaches workflow history in plain text. References written by earlier versions are still read. Workers on earlier SDK versions cannot read encoded references, so if a codec is configured, a workflow processed by a worker on this version must not then be processed by a worker on an earlier version (for example, after a rollback or during a mixed-version deployment). Without a codec, references are not encoded and remain readable by earlier versions. diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 9a8ed41980..ba42bfaa7e 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -13,6 +13,7 @@ import temporalio.api.common.v1 import temporalio.api.failure.v1 import temporalio.common +import temporalio.converter._extstore from temporalio.converter._extstore import ( _REFERENCE_ENCODING, _REFERENCE_MARKER_KEY, @@ -345,7 +346,10 @@ async def _encode_reference_payload_sequence( ) -> list[temporalio.api.common.v1.Payload]: """Codec encode external storage reference payloads only.""" encoded_payloads = list(payloads) - if not self.payload_codec: + if ( + not temporalio.converter._extstore._ENCODE_STORAGE_REFERENCES + or not self.payload_codec + ): return encoded_payloads for i, payload in enumerate(encoded_payloads): if not _is_reference_payload(payload): diff --git a/temporalio/converter/_extstore.py b/temporalio/converter/_extstore.py index 5dc3fd03d4..ddf6e7f509 100644 --- a/temporalio/converter/_extstore.py +++ b/temporalio/converter/_extstore.py @@ -32,6 +32,11 @@ # _REFERENCE_ENCODING could be any value. The presence of the _REFERENCE_MARKER_KEY # with a value of b"true" is used to reliably identify such payloads. _REFERENCE_MARKER_KEY = "__temporal_external_storage_reference" +# Writing marked, codec-encoded references is off until a release that can read +# them has been out long enough for workers to upgrade: workers on earlier +# versions cannot read encoded references. Reading is always on. Set to True in a +# later release to start writing them. +_ENCODE_STORAGE_REFERENCES = False @dataclass @@ -434,7 +439,8 @@ async def _store_payload(self, payload: Payload) -> Payload: f"Failed to serialize storage reference for driver '{driver.name()}'" ) reference_payload.external_payloads.add().size_bytes = external_size - reference_payload.metadata[_REFERENCE_MARKER_KEY] = b"true" + if _ENCODE_STORAGE_REFERENCES: + reference_payload.metadata[_REFERENCE_MARKER_KEY] = b"true" ExternalStorage._record_metrics(1, external_size, start_time, {driver.name()}) @@ -499,7 +505,8 @@ async def _store_payload_sequence( f"Failed to serialize storage reference for driver '{driver.name()}'" ) reference_payload.external_payloads.add().size_bytes = sizes[i] - reference_payload.metadata[_REFERENCE_MARKER_KEY] = b"true" + if _ENCODE_STORAGE_REFERENCES: + reference_payload.metadata[_REFERENCE_MARKER_KEY] = b"true" results[indices[i]] = reference_payload external_size += sizes[i] diff --git a/tests/conftest.py b/tests/conftest.py index 62ab5d5967..d51d180135 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -14,6 +14,7 @@ from opentelemetry.metrics import NoOpMeterProvider from opentelemetry.util._once import Once +import temporalio.converter._extstore from temporalio.client import Client from temporalio.envconfig import ClientConfigProfile from temporalio.testing import WorkflowEnvironment @@ -343,3 +344,12 @@ def reset_otel_logger_provider(): yield opentelemetry._logs._internal._LOGGER_PROVIDER_SET_ONCE = Once() opentelemetry._logs._internal._LOGGER_PROVIDER = None + + +@pytest.fixture +def encode_storage_references(monkeypatch: pytest.MonkeyPatch) -> None: + """Turn on writing marked, codec-encoded external storage references, which + is off by default until a later release.""" + monkeypatch.setattr( + temporalio.converter._extstore, "_ENCODE_STORAGE_REFERENCES", True + ) diff --git a/tests/nexus/test_temporal_extstore.py b/tests/nexus/test_temporal_extstore.py index ef2f22d153..760b56be04 100644 --- a/tests/nexus/test_temporal_extstore.py +++ b/tests/nexus/test_temporal_extstore.py @@ -225,6 +225,7 @@ def _assert_encoded_reference(payload: Payload) -> None: assert payload.metadata["__temporal_external_storage_reference"] == b"true" +@pytest.mark.usefixtures("encode_storage_references") async def test_nexus_operation_input_reference_encoded_with_codec( env: WorkflowEnvironment, ): @@ -253,6 +254,7 @@ async def test_nexus_operation_input_reference_encoded_with_codec( _assert_encoded_reference(scheduled.input) +@pytest.mark.usefixtures("encode_storage_references") async def test_nexus_operation_result_reference_encoded_with_codec( env: WorkflowEnvironment, ): diff --git a/tests/test_extstore.py b/tests/test_extstore.py index 95498030af..daa0547f3d 100644 --- a/tests/test_extstore.py +++ b/tests/test_extstore.py @@ -6,6 +6,7 @@ import pytest +import temporalio.converter._extstore from temporalio.api.common.v1 import Payload from temporalio.api.failure.v1 import Failure from temporalio.api.sdk.v1.external_storage_pb2 import ExternalStorageReference @@ -406,6 +407,7 @@ async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: return results +@pytest.mark.usefixtures("encode_storage_references") class TestPayloadCodecWithExternalStorage: """Tests for interaction between DataConverter.payload_codec and external storage.""" @@ -626,6 +628,47 @@ async def test_marked_reference_without_external_storage_fails(self): await DataConverter().decode(encoded, [str]) +async def test_references_written_unencoded_and_unmarked_by_default(): + """Until writing encoded references is turned on, references are written as + in earlier releases: unencoded and unmarked, even with a codec configured.""" + driver = InMemoryTestDriver() + converter = DataConverter( + payload_codec=RecordingPayloadCodec("binary/dc-encoded"), + external_storage=ExternalStorage(drivers=[driver], payload_size_threshold=50), + ) + + encoded = await converter.encode(["x" * 200]) + + assert encoded[0].metadata.get("encoding") == b"json/protobuf" + assert "__temporal_external_storage_reference" not in encoded[0].metadata + assert await converter.decode(encoded, [str]) == ["x" * 200] + + +async def test_encoded_references_read_while_writing_them_is_off( + monkeypatch: pytest.MonkeyPatch, +): + """Encoded references written by a later release can be read before writing + them is turned on.""" + converter = DataConverter( + payload_codec=RecordingPayloadCodec("binary/dc-encoded"), + external_storage=ExternalStorage( + drivers=[InMemoryTestDriver()], payload_size_threshold=50 + ), + ) + monkeypatch.setattr( + temporalio.converter._extstore, "_ENCODE_STORAGE_REFERENCES", True + ) + encoded = await converter.encode(["x" * 200]) + assert encoded[0].metadata.get("encoding") == b"binary/dc-encoded" + assert encoded[0].metadata["__temporal_external_storage_reference"] == b"true" + + monkeypatch.setattr( + temporalio.converter._extstore, "_ENCODE_STORAGE_REFERENCES", False + ) + + assert await converter.decode(encoded, [str]) == ["x" * 200] + + class TestMultiDriver: """Tests for ExternalStorage with multiple drivers.""" diff --git a/tests/test_extstore_client.py b/tests/test_extstore_client.py index 8ea6a89eb0..b8588b0649 100644 --- a/tests/test_extstore_client.py +++ b/tests/test_extstore_client.py @@ -39,6 +39,8 @@ from temporalio.service import ServiceClient from tests.test_extstore import InMemoryTestDriver, RecordingPayloadCodec +pytestmark = pytest.mark.usefixtures("encode_storage_references") + # Larger than the storage threshold below, so it is always stored externally. LARGE = "x" * 200 CODEC_ENCODING = b"binary/test-encoded" diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index b4268b490b..acc2efe042 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -442,6 +442,7 @@ async def test_replay_extstore_history_succeeds_with_correct_extstore( ).replay_workflow(history) +@pytest.mark.usefixtures("encode_storage_references") async def test_extstore_reference_payloads_codec_encoded_in_history( env: WorkflowEnvironment, ) -> None: @@ -534,6 +535,7 @@ async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: return await self._codec.decode(payloads) +@pytest.mark.usefixtures("encode_storage_references") async def test_extstore_reference_payloads_use_payload_serialization_context( env: WorkflowEnvironment, ) -> None: From f59146d7750af6b3cb1fe1799851f41c3c3517ab Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Fri, 16 Oct 2026 14:17:43 -0400 Subject: [PATCH 08/10] fix lint issue --- tests/worker/test_extstore.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index acc2efe042..2994f890d2 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -591,7 +591,7 @@ class ReferencesBeforeEncodingWorkflow: test_extstore_references_before_encoding_history.json.""" @workflow.run - async def run(self, input: str) -> str: + async def run(self, _input: str) -> str: await workflow.execute_activity( "references_before_encoding_activity", "ai" * 512, From 8c5b34cb9537b06ecd5da6bdd7a9c624f1a24e22 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Fri, 16 Oct 2026 18:29:59 -0400 Subject: [PATCH 09/10] Update tests to make sure previously encoded extstore references are decoded correctly. Make naming more clear. --- tests/worker/test_extstore.py | 68 +++-- ...t_extstore_encoded_references_history.json | 254 ++++++++++++++++++ ...e_encoded_references_stored_payloads.json} | 0 ...xtstore_unencoded_references_history.json} | 0 ..._unencoded_references_stored_payloads.json | 26 ++ 5 files changed, 326 insertions(+), 22 deletions(-) create mode 100644 tests/worker/test_extstore_encoded_references_history.json rename tests/worker/{test_extstore_references_before_encoding_stored_payloads.json => test_extstore_encoded_references_stored_payloads.json} (100%) rename tests/worker/{test_extstore_references_before_encoding_history.json => test_extstore_unencoded_references_history.json} (100%) create mode 100644 tests/worker/test_extstore_unencoded_references_stored_payloads.json diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index 2994f890d2..c08f67db9c 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -585,10 +585,11 @@ async def test_extstore_reference_payloads_use_payload_serialization_context( assert len(reference_contexts) == 4 -@workflow.defn -class ReferencesBeforeEncodingWorkflow: - """Replays the committed history in - test_extstore_references_before_encoding_history.json.""" +# The committed histories record these workflow and activity names, so they +# must not change. +@workflow.defn(name="ReferencesBeforeEncodingWorkflow") +class ExtstoreReplayWorkflow: + """Replays the committed histories loaded by _load_recorded_history.""" @workflow.run async def run(self, _input: str) -> str: @@ -600,26 +601,24 @@ async def run(self, _input: str) -> str: return "wo" * 512 -async def test_replay_history_with_references_written_before_encoding() -> None: - """A history recorded with temporalio 1.34.0, whose references were written - unencoded and unmarked, still replays with a codec configured. +def _load_recorded_history( + name: str, +) -> tuple[WorkflowHistory, InMemoryTestDriver, list[Payload]]: + """Load a committed history, a driver holding the payloads its references + point to, and its four reference payloads. - The history and the payloads its references point to were recorded with - ReferencesBeforeEncodingWorkflow, a payload codec matching - RecordingPayloadCodec("binary/test-encoded"), and a driver storing claims the - same way as InMemoryTestDriver, with a 512-byte storage threshold. + Each history was recorded with ExtstoreReplayWorkflow, a payload codec + matching RecordingPayloadCodec("binary/test-encoded"), and a driver storing + claims the same way as InMemoryTestDriver, with a 512-byte storage threshold. + The codec always encoded the stored payloads. Whether it also encoded the + references depends on the recording. """ fixtures = Path(__file__).parent history = WorkflowHistory.from_json( - "references-before-encoding", - ( - fixtures / "test_extstore_references_before_encoding_history.json" - ).read_text(), + name, (fixtures / f"test_extstore_{name}_history.json").read_text() ) stored_payloads = json.loads( - ( - fixtures / "test_extstore_references_before_encoding_stored_payloads.json" - ).read_text() + (fixtures / f"test_extstore_{name}_stored_payloads.json").read_text() ) driver = InMemoryTestDriver() for key, payload_json in stored_payloads.items(): @@ -638,12 +637,14 @@ async def test_replay_history_with_references_written_before_encoding() -> None: ) ] assert len(references) == 4 - for reference in references: - assert reference.metadata["encoding"] == b"json/protobuf" - assert "__temporal_external_storage_reference" not in reference.metadata + return history, driver, references + +async def _replay_recorded_history( + history: WorkflowHistory, driver: InMemoryTestDriver +) -> None: await Replayer( - workflows=[ReferencesBeforeEncodingWorkflow], + workflows=[ExtstoreReplayWorkflow], data_converter=dataclasses.replace( temporalio.converter.default(), payload_codec=RecordingPayloadCodec("binary/test-encoded"), @@ -656,6 +657,29 @@ async def test_replay_history_with_references_written_before_encoding() -> None: assert driver._retrieve_calls > 0 +async def test_replay_decodes_unencoded_references() -> None: + """A history recorded with temporalio 1.34.0 replays. That release wrote + references unencoded and unmarked, even with a codec configured.""" + history, driver, references = _load_recorded_history("unencoded_references") + for reference in references: + assert reference.metadata["encoding"] == b"json/protobuf" + assert "__temporal_external_storage_reference" not in reference.metadata + + await _replay_recorded_history(history, driver) + + +async def test_replay_decodes_encoded_references_while_writing_them_is_off() -> None: + """A history whose references were codec-encoded and marked replays while + writing encoded references is off. It was recorded from this branch with + writing turned on, as a later release will write them.""" + history, driver, references = _load_recorded_history("encoded_references") + for reference in references: + assert reference.metadata["encoding"] == b"binary/test-encoded" + assert reference.metadata["__temporal_external_storage_reference"] == b"true" + + await _replay_recorded_history(history, driver) + + async def test_replay_extstore_history_fails_with_empty_driver( env: WorkflowEnvironment, ) -> None: diff --git a/tests/worker/test_extstore_encoded_references_history.json b/tests/worker/test_extstore_encoded_references_history.json new file mode 100644 index 0000000000..1d566e54d5 --- /dev/null +++ b/tests/worker/test_extstore_encoded_references_history.json @@ -0,0 +1,254 @@ +{ + "events": [ + { + "eventId": "1", + "eventTime": "2026-10-16T22:21:24.234668Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED", + "taskId": "1048587", + "workflowExecutionStartedEventAttributes": { + "workflowType": { + "name": "ReferencesBeforeEncodingWorkflow" + }, + "taskQueue": { + "name": "references-before-encoding", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "input": { + "payloads": [ + { + "metadata": { + "__temporal_external_storage_reference": "dHJ1ZQ==", + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChkKCGVuY29kaW5nEg1qc29uL3Byb3RvYnVmCi0KJV9fdGVtcG9yYWxfZXh0ZXJuYWxfc3RvcmFnZV9yZWZlcmVuY2USBHRydWUKOwoLbWVzc2FnZVR5cGUSLHRlbXBvcmFsLmFwaS5zZGsudjEuRXh0ZXJuYWxTdG9yYWdlUmVmZXJlbmNlEjx7ImNsYWltRGF0YSI6eyJrZXkiOiJwYXlsb2FkLTAifSwiZHJpdmVyTmFtZSI6InRlc3QtZHJpdmVyIn0aAwjBCA==", + "externalPayloads": [ + { + "sizeBytes": "1089" + } + ] + } + ] + }, + "workflowTaskTimeout": "10s", + "originalExecutionRunId": "01a146ce-3f8a-7a31-9ace-21c7f1037392", + "identity": "test-identity", + "firstExecutionRunId": "01a146ce-3f8a-7a31-9ace-21c7f1037392", + "attempt": 1, + "firstWorkflowTaskBackoff": "0s", + "workflowId": "encoded-references", + "priority": {} + } + }, + { + "eventId": "2", + "eventTime": "2026-10-16T22:21:24.234703Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048588", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "references-before-encoding", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "3", + "eventTime": "2026-10-16T22:21:24.239003Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048593", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "2", + "identity": "test-identity", + "requestId": "cd517ff8-a3e7-42d4-99b3-28ee8f1851bd", + "historySizeBytes": "598", + "workerVersion": { + "buildId": "8c2a779eb0ea4f39243fe56c372758b2" + } + } + }, + { + "eventId": "4", + "eventTime": "2026-10-16T22:21:24.249060Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048598", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "2", + "startedEventId": "3", + "identity": "test-identity", + "workerVersion": { + "buildId": "8c2a779eb0ea4f39243fe56c372758b2" + }, + "sdkMetadata": { + "coreUsedFlags": [ + 3, + 2, + 1 + ], + "sdkName": "temporal-python", + "sdkVersion": "1.34.0" + }, + "meteringMetadata": {} + } + }, + { + "eventId": "5", + "eventTime": "2026-10-16T22:21:24.249226Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_SCHEDULED", + "taskId": "1048599", + "activityTaskScheduledEventAttributes": { + "activityId": "1", + "activityType": { + "name": "references_before_encoding_activity" + }, + "taskQueue": { + "name": "references-before-encoding", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "header": {}, + "input": { + "payloads": [ + { + "metadata": { + "__temporal_external_storage_reference": "dHJ1ZQ==", + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChkKCGVuY29kaW5nEg1qc29uL3Byb3RvYnVmCi0KJV9fdGVtcG9yYWxfZXh0ZXJuYWxfc3RvcmFnZV9yZWZlcmVuY2USBHRydWUKOwoLbWVzc2FnZVR5cGUSLHRlbXBvcmFsLmFwaS5zZGsudjEuRXh0ZXJuYWxTdG9yYWdlUmVmZXJlbmNlEjx7ImNsYWltRGF0YSI6eyJrZXkiOiJwYXlsb2FkLTEifSwiZHJpdmVyTmFtZSI6InRlc3QtZHJpdmVyIn0aAwjBCA==", + "externalPayloads": [ + { + "sizeBytes": "1089" + } + ] + } + ] + }, + "scheduleToCloseTimeout": "0s", + "scheduleToStartTimeout": "0s", + "startToCloseTimeout": "30s", + "heartbeatTimeout": "0s", + "workflowTaskCompletedEventId": "4", + "retryPolicy": { + "initialInterval": "1s", + "backoffCoefficient": 2.0, + "maximumInterval": "100s" + }, + "useWorkflowBuildId": true, + "priority": {} + } + }, + { + "eventId": "6", + "eventTime": "2026-10-16T22:21:24.249342Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_STARTED", + "taskId": "1048603", + "activityTaskStartedEventAttributes": { + "scheduledEventId": "5", + "identity": "test-identity", + "requestId": "52ee8ebc-8674-4a3c-b948-485c0b4728ac", + "attempt": 1, + "workerVersion": { + "buildId": "8c2a779eb0ea4f39243fe56c372758b2" + } + } + }, + { + "eventId": "7", + "eventTime": "2026-10-16T22:21:24.252491Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_COMPLETED", + "taskId": "1048604", + "activityTaskCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "__temporal_external_storage_reference": "dHJ1ZQ==", + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChkKCGVuY29kaW5nEg1qc29uL3Byb3RvYnVmCi0KJV9fdGVtcG9yYWxfZXh0ZXJuYWxfc3RvcmFnZV9yZWZlcmVuY2USBHRydWUKOwoLbWVzc2FnZVR5cGUSLHRlbXBvcmFsLmFwaS5zZGsudjEuRXh0ZXJuYWxTdG9yYWdlUmVmZXJlbmNlEjx7ImNsYWltRGF0YSI6eyJrZXkiOiJwYXlsb2FkLTIifSwiZHJpdmVyTmFtZSI6InRlc3QtZHJpdmVyIn0aAwjBCA==", + "externalPayloads": [ + { + "sizeBytes": "1089" + } + ] + } + ] + }, + "scheduledEventId": "5", + "startedEventId": "6", + "identity": "test-identity" + } + }, + { + "eventId": "8", + "eventTime": "2026-10-16T22:21:24.252513Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048605", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "test-identity-sticky", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "references-before-encoding" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "9", + "eventTime": "2026-10-16T22:21:24.253136Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048609", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "8", + "identity": "test-identity", + "requestId": "78b8b8e0-2f33-432c-a7d9-c2db06878105", + "historySizeBytes": "1931", + "workerVersion": { + "buildId": "8c2a779eb0ea4f39243fe56c372758b2" + } + } + }, + { + "eventId": "10", + "eventTime": "2026-10-16T22:21:24.255732Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048613", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "8", + "startedEventId": "9", + "identity": "test-identity", + "workerVersion": { + "buildId": "8c2a779eb0ea4f39243fe56c372758b2" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "11", + "eventTime": "2026-10-16T22:21:24.255773Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED", + "taskId": "1048614", + "workflowExecutionCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "__temporal_external_storage_reference": "dHJ1ZQ==", + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "CjsKC21lc3NhZ2VUeXBlEix0ZW1wb3JhbC5hcGkuc2RrLnYxLkV4dGVybmFsU3RvcmFnZVJlZmVyZW5jZQotCiVfX3RlbXBvcmFsX2V4dGVybmFsX3N0b3JhZ2VfcmVmZXJlbmNlEgR0cnVlChkKCGVuY29kaW5nEg1qc29uL3Byb3RvYnVmEjx7ImNsYWltRGF0YSI6eyJrZXkiOiJwYXlsb2FkLTMifSwiZHJpdmVyTmFtZSI6InRlc3QtZHJpdmVyIn0aAwjBCA==", + "externalPayloads": [ + { + "sizeBytes": "1089" + } + ] + } + ] + }, + "workflowTaskCompletedEventId": "10" + } + } + ] +} \ No newline at end of file diff --git a/tests/worker/test_extstore_references_before_encoding_stored_payloads.json b/tests/worker/test_extstore_encoded_references_stored_payloads.json similarity index 100% rename from tests/worker/test_extstore_references_before_encoding_stored_payloads.json rename to tests/worker/test_extstore_encoded_references_stored_payloads.json diff --git a/tests/worker/test_extstore_references_before_encoding_history.json b/tests/worker/test_extstore_unencoded_references_history.json similarity index 100% rename from tests/worker/test_extstore_references_before_encoding_history.json rename to tests/worker/test_extstore_unencoded_references_history.json diff --git a/tests/worker/test_extstore_unencoded_references_stored_payloads.json b/tests/worker/test_extstore_unencoded_references_stored_payloads.json new file mode 100644 index 0000000000..03768b539c --- /dev/null +++ b/tests/worker/test_extstore_unencoded_references_stored_payloads.json @@ -0,0 +1,26 @@ +{ + "payload-0": { + "metadata": { + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChYKCGVuY29kaW5nEgpqc29uL3BsYWluEoIIIndpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2l3aXdpd2ki" + }, + "payload-1": { + "metadata": { + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChYKCGVuY29kaW5nEgpqc29uL3BsYWluEoIIImFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWlhaWFpYWki" + }, + "payload-2": { + "metadata": { + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChYKCGVuY29kaW5nEgpqc29uL3BsYWluEoIIImFvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW9hb2FvYW8i" + }, + "payload-3": { + "metadata": { + "encoding": "YmluYXJ5L3Rlc3QtZW5jb2RlZA==" + }, + "data": "ChYKCGVuY29kaW5nEgpqc29uL3BsYWluEoIIIndvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd293b3dvd28i" + } +} From 404b3ee469afaf29c898fcb80eec4774ef87e535 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Fri, 16 Oct 2026 18:30:19 -0400 Subject: [PATCH 10/10] Update changelog name. --- changelog/added/{discerning_murlock.md => discerning-murlock.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename changelog/added/{discerning_murlock.md => discerning-murlock.md} (100%) diff --git a/changelog/added/discerning_murlock.md b/changelog/added/discerning-murlock.md similarity index 100% rename from changelog/added/discerning_murlock.md rename to changelog/added/discerning-murlock.md