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/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..ba42bfaa7e 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -13,8 +13,10 @@ 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, _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,55 @@ 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) + 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): + continue + [encoded_reference] = await self.payload_codec.encode([payload]) + if not encoded_reference.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 + + 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) + 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() + encoded_reference.CopyFrom(payload) + del encoded_reference.metadata[_REFERENCE_MARKER_KEY] + del encoded_reference.external_payloads[:] + [reference] = await self.payload_codec.decode([encoded_reference]) + if not _is_reference_payload(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 reference.external_payloads: + reference.external_payloads.extend(payload.external_payloads) + decoded_payloads[i] = 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..ddf6e7f509 100644 --- a/temporalio/converter/_extstore.py +++ b/temporalio/converter/_extstore.py @@ -28,6 +28,15 @@ _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" +# 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 @@ -430,6 +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 + if _ENCODE_STORAGE_REFERENCES: + reference_payload.metadata[_REFERENCE_MARKER_KEY] = b"true" ExternalStorage._record_metrics(1, external_size, start_time, {driver.name()}) @@ -494,6 +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] + if _ENCODE_STORAGE_REFERENCES: + 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/conftest.py b/tests/conftest.py index 741b1be4df..0b2ce24dac 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 @@ -351,3 +352,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 ac38051e21..9ec1ff9d13 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,69 @@ 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" + + +@pytest.mark.usefixtures("encode_storage_references") +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) + + +@pytest.mark.usefixtures("encode_storage_references") +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 943fc7be2a..daa0547f3d 100644 --- a/tests/test_extstore.py +++ b/tests/test_extstore.py @@ -6,7 +6,9 @@ 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 from temporalio.converter import ( DataConverter, @@ -405,13 +407,13 @@ 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.""" 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 +430,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 +438,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 +457,217 @@ 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,) + + async def test_multiple_stored_payloads_encode_each_reference(self): + """When several payloads are stored together, every reference payload + 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], + # The codec-encoded "inline" value is ~67 bytes, below this. + payload_size_threshold=100, + ), + ) + 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 ( + reference.metadata.get("__temporal_external_storage_reference") + == b"true" + ) + 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): + """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]) + + +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 new file mode 100644 index 0000000000..b8588b0649 --- /dev/null +++ b/tests/test_extstore_client.py @@ -0,0 +1,362 @@ +"""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.api.sdk.v1.external_storage_pb2 import ExternalStorageReference +from temporalio.client import ( + Client, + Schedule, + ScheduleActionStartWorkflow, + ScheduleSpec, + WorkflowFailureError, +) +from temporalio.common import HeaderCodecBehavior +from temporalio.converter import ( + DataConverter, + ExternalStorage, + JSONProtoPayloadConverter, +) +from temporalio.exceptions import ApplicationError +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" + + +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, + ) + + +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]) + await _assert_encoded_reference(data_converter, 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"] + 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: + service = _FakeWorkflowService( + { + "signal_workflow_execution": temporalio.api.workflowservice.v1.SignalWorkflowExecutionResponse() + } + ) + client = _client(service, _data_converter()) + + await client.get_workflow_handle("wf").signal("signal", LARGE) + + await _assert_encoded_reference( + client.data_converter, + 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"] + 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: + 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"] + await _assert_encoded_reference( + client.data_converter, 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 + await _assert_encoded_reference( + client.data_converter, + 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"] + await _assert_encoded_reference( + client.data_converter, 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)) + + await _assert_encoded_reference( + client.data_converter, + service.requests["record_activity_task_heartbeat_by_id"].details.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 + await _assert_encoded_reference( + client.data_converter, 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"] + await _assert_encoded_reference(client.data_converter, 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) + await _assert_encoded_reference( + data_converter, 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 ff0bf75818..8b9b12fd1c 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,22 +19,32 @@ 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, ExternalStorage, + PayloadCodec, + SerializationContext, StorageDriver, StorageDriverActivityInfo, StorageDriverClaim, StorageDriverRetrieveContext, StorageDriverStoreContext, StorageDriverWorkflowInfo, + WithSerializationContext, + WorkflowSerializationContext, ) from temporalio.exceptions import ActivityError, ApplicationError 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 +442,244 @@ 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: + """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"): + reference_payloads.append( + event.workflow_execution_started_event_attributes.input.payloads[0] + ) + elif event.HasField("activity_task_scheduled_event_attributes"): + reference_payloads.append( + event.activity_task_scheduled_event_attributes.input.payloads[0] + ) + elif event.HasField("activity_task_completed_event_attributes"): + reference_payloads.append( + event.activity_task_completed_event_attributes.result.payloads[0] + ) + elif event.HasField("workflow_execution_completed_event_attributes"): + 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" + 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) + + +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) + + +@pytest.mark.usefixtures("encode_storage_references") +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 + + +# 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: + await workflow.execute_activity( + "references_before_encoding_activity", + "ai" * 512, + start_to_close_timeout=timedelta(seconds=30), + ) + return "wo" * 512 + + +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. + + 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( + name, (fixtures / f"test_extstore_{name}_history.json").read_text() + ) + stored_payloads = json.loads( + (fixtures / f"test_extstore_{name}_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 + return history, driver, references + + +async def _replay_recorded_history( + history: WorkflowHistory, driver: InMemoryTestDriver +) -> None: + await Replayer( + workflows=[ExtstoreReplayWorkflow], + 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_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_encoded_references_stored_payloads.json b/tests/worker/test_extstore_encoded_references_stored_payloads.json new file mode 100644 index 0000000000..03768b539c --- /dev/null +++ b/tests/worker/test_extstore_encoded_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" + } +} diff --git a/tests/worker/test_extstore_unencoded_references_history.json b/tests/worker/test_extstore_unencoded_references_history.json new file mode 100644 index 0000000000..4b46f8d0e9 --- /dev/null +++ b/tests/worker/test_extstore_unencoded_references_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_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" + } +}