diff --git a/pyproject.toml b/pyproject.toml index 14dafe4..99804e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.5.4" +version = "2.5.5" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.5.4" -registry-version = "2.5.4" +product-train = "2.5.5" +registry-version = "2.5.5" supported-server-versions = "2.5.16" worker-protocol-version = "1.19" control-plane-version = "2" diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index d20b751..776abf4 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -19,6 +19,7 @@ from __future__ import annotations import asyncio +import copy import hashlib import json as json_module import math @@ -2672,6 +2673,26 @@ def get_workflow_handle( """ return WorkflowHandle(self, workflow_id=workflow_id, run_id=run_id, workflow_type=workflow_type) + def _nexus_target_client(self, target_namespace: str | None) -> Client: + if target_namespace is not None and not target_namespace.strip(): + raise ValueError("target_namespace must be nonempty") + if target_namespace is None or target_namespace == self.namespace: + return self + self._auth_token(worker=False) + scoped = copy.copy(self) + scoped.namespace = target_namespace + # Share the HTTP pool, never the caller's namespace discovery or + # provider credentials. The worker keeps using the original client. + scoped.worker_token = None + scoped._cluster_info = None + scoped._cluster_info_lock = asyncio.Lock() + scoped._runtime_external_payload_transport_cache = None + scoped._runtime_external_payload_transport_resolved = False + scoped.external_storage = None + scoped.external_storage_threshold_bytes = None + scoped.external_storage_cache = ExternalPayloadCache() + return scoped + async def execute_nexus_operation( self, endpoint_name: str, @@ -2685,6 +2706,7 @@ async def execute_nexus_operation( idempotency_key: str | None = None, payload_codec: str | None = None, caller_namespace: str | None = None, + target_namespace: str | None = None, caller_workflow_instance_id: str | None = None, caller_workflow_run_id: str | None = None, service_sdk_language: str | None = None, @@ -2708,7 +2730,11 @@ async def execute_nexus_operation( the worker uses this client method to make the single durable execute request and records the returned service-call surface into workflow history. + + ``target_namespace`` selects the service catalog and its payload + storage. The caller identity and worker namespace are preserved. """ + execution_client = self._nexus_target_client(target_namespace) request_body = nexus_request_payload( arguments=list(arguments) if arguments is not None else [], payload_codec=payload_codec, @@ -2736,8 +2762,21 @@ async def execute_nexus_operation( ) context = f"{endpoint_name}/{service_name}/{operation_name}" + wire_body = request_body + transport = await execution_client._runtime_external_payload_transport() + if transport is not None: + arguments_envelope = await execution_client._externalize_runtime_payloads( + {"codec": request_body["payload_codec"], "blob": request_body["arguments"]}, + worker=False, + transport=transport, + uploaded={}, + slot=("arguments",), + ) + if "external_payload" in arguments_envelope: + wire_body = {**request_body, "arguments": arguments_envelope} + try: - data = await self._request("POST", path, json=request_body, context=context) + data = await execution_client._request("POST", path, json=wire_body, context=context) except ServerError as exc: if exc.status != 409 or not isinstance(exc.body, dict): raise @@ -2747,6 +2786,7 @@ async def execute_nexus_operation( service_name=service_name, operation_name=operation_name, request_payload=request_body, + target_namespace=target_namespace, service_sdk_language=service_sdk_language, artifact_tuple=artifact_tuple, published_artifact_worker_execution=published_artifact_worker_execution, @@ -2770,6 +2810,7 @@ async def execute_nexus_operation( service_name=service_name, operation_name=operation_name, request_payload=request_body, + target_namespace=target_namespace, service_sdk_language=service_sdk_language, artifact_tuple=artifact_tuple, published_artifact_worker_execution=published_artifact_worker_execution, diff --git a/src/durable_workflow/nexus.py b/src/durable_workflow/nexus.py index 941e54d..a338441 100644 --- a/src/durable_workflow/nexus.py +++ b/src/durable_workflow/nexus.py @@ -8,6 +8,7 @@ from dataclasses import dataclass, field from typing import Any +from . import serializer from .errors import NexusOperationFailed NEXUS_OPERATION_RESULT_SCHEMA = "durable-workflow.v2.sdk-python.nexus-operation-result" @@ -49,6 +50,7 @@ def stable_nexus_idempotency_key( service_name: str, operation_name: str, arguments: Sequence[Any], + target_namespace: str | None = None, ) -> str: """Return a deterministic idempotency key for one workflow-side Nexus call.""" source = { @@ -61,6 +63,8 @@ def stable_nexus_idempotency_key( "operation_name": operation_name, "arguments": list(arguments), } + if target_namespace is not None: + source["target_namespace"] = target_namespace encoded = json.dumps(source, sort_keys=True, separators=(",", ":"), default=_json_default) return f"dw-py-nexus-{hashlib.sha256(encoded.encode()).hexdigest()}" @@ -87,9 +91,10 @@ def nexus_request_payload( duplicate_start_policy: str | None = None, ) -> dict[str, Any]: """Build the server execute-operation request body.""" + codec = payload_codec or serializer.AVRO_CODEC return _clean_dict({ - "arguments": list(arguments), - "payload_codec": payload_codec, + "arguments": serializer.encode(list(arguments), codec=codec), + "payload_codec": codec, "mode_override": mode, "wait_for": wait_for, "wait_timeout_seconds": wait_timeout_seconds, @@ -134,6 +139,7 @@ class NexusOperationResult: caller_observed_error_type: str | None = None typed_error_message: str | None = None result: Any = None + target_namespace: str | None = None @classmethod def from_response( @@ -147,6 +153,7 @@ def from_response( service_sdk_language: str | None = None, artifact_tuple: Mapping[str, Any] | None = None, published_artifact_worker_execution: bool | None = None, + target_namespace: str | None = None, ) -> NexusOperationResult: call = data.get("service_call") source = call if isinstance(call, Mapping) else data @@ -188,6 +195,7 @@ def from_response( typed_error_message=_optional_str(source.get("typed_error_message")) or _optional_str(data.get("typed_error_message")), result=response_value, + target_namespace=target_namespace, ) @classmethod @@ -199,6 +207,9 @@ def from_recorded_payload(cls, value: Any, *, expected: Mapping[str, Any]) -> Ne if value.get("version") != NEXUS_OPERATION_RESULT_VERSION: raise ValueError("recorded Nexus operation result version is not supported") + if value.get("target_namespace") != expected.get("target_namespace"): + raise ValueError("recorded Nexus target namespace does not match the current workflow command") + for key in ("endpoint_name", "service_name", "operation_name", "idempotency_key"): expected_value = expected.get(key) recorded_value = value.get(key) @@ -235,6 +246,7 @@ def from_recorded_payload(cls, value: Any, *, expected: Mapping[str, Any]) -> Ne caller_observed_error_type=_optional_str(value.get("caller_observed_error_type")), typed_error_message=_optional_str(value.get("typed_error_message")), result=value.get("result"), + target_namespace=_optional_str(value.get("target_namespace")), ) if result.is_failure: raise result.to_failure() @@ -287,6 +299,7 @@ def to_recorded_payload(self) -> dict[str, Any]: "endpoint_name": self.endpoint_name, "service_name": self.service_name, "operation_name": self.operation_name, + "target_namespace": self.target_namespace, "idempotency_key": self.request_payload.get("idempotency_key"), "caller_workflow_instance_id": self.caller_workflow_instance_id, "caller_workflow_run_id": self.caller_workflow_run_id, diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index d8c853c..c285675 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -1462,6 +1462,7 @@ async def _resolve_workflow_nexus_commands( idempotency_key=command.idempotency_key, payload_codec=command.payload_codec, caller_namespace=command.caller_namespace, + target_namespace=command.target_namespace, caller_workflow_instance_id=_string_or_none(task.get("workflow_id")), caller_workflow_run_id=_string_or_none(task.get("run_id")), service_sdk_language=command.service_sdk_language, diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index c6e47f7..85bd674 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -1196,6 +1196,7 @@ class NexusServiceCall: memo: dict[str, Any] | None = None search_attributes: dict[str, Any] | None = None duplicate_start_policy: str | None = None + target_namespace: str | None = None def to_server_command(self, *args: Any, **kwargs: Any) -> dict[str, Any]: raise RuntimeError( @@ -1209,6 +1210,7 @@ def expected_recorded_metadata(self) -> dict[str, Any]: "service_name": self.service_name, "operation_name": self.operation_name, "idempotency_key": self.idempotency_key, + "target_namespace": self.target_namespace, } def recorded_result(self, value: Any) -> NexusOperationResult: @@ -2402,6 +2404,7 @@ def call_nexus_service( wait_for: str | None = "completed", wait_timeout_seconds: int | None = None, caller_namespace: str | None = None, + target_namespace: str | None = None, service_sdk_language: str | None = None, artifact_tuple: Mapping[str, Any] | None = None, published_artifact_worker_execution: bool | None = None, @@ -2420,7 +2423,12 @@ def call_nexus_service( If the service reports a typed failure, replay raises :class:`~durable_workflow.errors.NexusOperationFailed` at the yield point so workflow code can compensate or let the workflow fail. + + ``target_namespace`` addresses a service in another namespace and + participates in the durable call identity and replay checks. """ + if target_namespace is not None and not target_namespace.strip(): + raise ValueError("target_namespace must be nonempty") args = list(arguments) if arguments is not None else [] if idempotency_key is None: idempotency_key = stable_nexus_idempotency_key( @@ -2431,6 +2439,7 @@ def call_nexus_service( service_name=service_name, operation_name=operation_name, arguments=args, + target_namespace=target_namespace, ) self._nexus_call_counter += 1 return NexusServiceCall( @@ -2444,6 +2453,7 @@ def call_nexus_service( wait_for=wait_for, wait_timeout_seconds=wait_timeout_seconds, caller_namespace=caller_namespace, + target_namespace=target_namespace, service_sdk_language=service_sdk_language, artifact_tuple=dict(artifact_tuple) if artifact_tuple is not None else None, published_artifact_worker_execution=published_artifact_worker_execution, @@ -6589,6 +6599,14 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if result_cursor < len(resolved_results): _assert_next_step_matches(cmd) recorded_value = resolved_results[result_cursor] + if ( + isinstance(recorded_value, Mapping) + and recorded_value.get("target_namespace") != cmd.target_namespace + ): + raise NonDeterministicReplayError( + current_call_sequence, "NexusServiceCall", ["SideEffectRecorded"], + detail="nexus_target_namespace_changed: recorded target differs from the workflow command", + ) result_cursor += 1 try: next_value = cmd.recorded_result(recorded_value) diff --git a/tests/fixtures/codec_regressions/nexus-world-arguments-avro.json b/tests/fixtures/codec_regressions/nexus-world-arguments-avro.json new file mode 100644 index 0000000..bf2b780 --- /dev/null +++ b/tests/fixtures/codec_regressions/nexus-world-arguments-avro.json @@ -0,0 +1,21 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.codec-regression/v1", + "id": "nexus-world-arguments-avro", + "protocol": { + "codec": "avro", + "schema": "durable_workflow.protocol.Value", + "version": "1", + "fingerprint": "e2a33dff55802237" + }, + "bindings": ["python"], + "value": { + "type": "array", + "items": [{"type": "string", "value": "world"}] + }, + "framing": { + "encoding": "avro-single-object", + "wire_base64": "wwHioz3/VYAiNwwCCgp3b3JsZAA=" + }, + "failure_policy": {"operation": "round_trip", "error": null} +} diff --git a/tests/fixtures/replay_regressions/nexus-target-namespace-changed.json b/tests/fixtures/replay_regressions/nexus-target-namespace-changed.json new file mode 100644 index 0000000..10d5883 --- /dev/null +++ b/tests/fixtures/replay_regressions/nexus-target-namespace-changed.json @@ -0,0 +1,19 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "nexus-target-namespace-changed", + "protocol_version": "1.19", + "bindings": ["python"], + "workflow": {"type": "tests.replay.nexus-target", "input": [], "payload_codec": "avro"}, + "history": [{"event_type": "SideEffectRecorded", "payload": { + "sequence": 1, + "payload_codec": "avro", + "result": "wwHioz3/VYAiNw4WDHNjaGVtYQpqZHVyYWJsZS13b3JrZmxvdy52Mi5zZGstcHl0aG9uLm5leHVzLW9wZXJhdGlvbi1yZXN1bHQOdmVyc2lvbgQCEGFjY2VwdGVkAgEec2VydmljZV9jYWxsX2lkCiBvcmlnaW5hbC1jYWxsLWlkGmVuZHBvaW50X25hbWUKDmdyZWV0ZXIYc2VydmljZV9uYW1lCgxzaGFyZWQcb3BlcmF0aW9uX25hbWUKCmdyZWV0HmlkZW1wb3RlbmN5X2tleQoab3JpZ2luYWwtY2FsbCB0YXJnZXRfbmFtZXNwYWNlCgxzaGFyZWQMc3RhdHVzChJjb21wbGV0ZWQMcmVzdWx0ChpzZXJ2aWNlLXZhbHVlAA==" + }}], + "expected": {"command_sequence": []}, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "nexus_target_namespace_changed", + "workflow_sequence": 1 + } +} diff --git a/tests/test_client.py b/tests/test_client.py index c2e67c6..2a77c36 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -289,7 +289,8 @@ async def test_execute_nexus_operation_records_request_and_evidence(self, client "/api/service-endpoints/greeter/services/shared/operations/greet/execute", ) body = mock.call_args.kwargs["json"] - assert body["arguments"] == ["Ada"] + assert body["payload_codec"] == "avro" + assert serializer.decode(body["arguments"], codec=body["payload_codec"]) == ["Ada"] assert body["mode_override"] == "sync" assert body["wait_for"] == "completed" assert body["idempotency_key"] == "idem-1" @@ -303,7 +304,7 @@ async def test_execute_nexus_operation_records_request_and_evidence(self, client assert recorded["caller_sdk_language"] == "sdk-python" assert recorded["service_sdk_language"] == "workflow-php" assert recorded["operation_name"] == "greet" - assert recorded["request_payload"]["arguments"] == ["Ada"] + assert serializer.decode(recorded["request_payload"]["arguments"], codec="avro") == ["Ada"] assert recorded["response_or_failure_surface"]["result"] == {"greeting": "hello Ada"} assert recorded["service_call_id"] == "svc-call-1" assert recorded["artifact_tuple"]["sdk-python"] == "0.4.95" diff --git a/tests/test_nexus_namespace.py b/tests/test_nexus_namespace.py new file mode 100644 index 0000000..4ff77ac --- /dev/null +++ b/tests/test_nexus_namespace.py @@ -0,0 +1,160 @@ +import asyncio +import json + +import httpx +import pytest + +from durable_workflow import Client, serializer, workflow +from durable_workflow.errors import NonDeterministicReplayError +from durable_workflow.nexus import NexusOperationResult, stable_nexus_idempotency_key +from durable_workflow.workflow import CompleteWorkflow, NexusServiceCall, WorkflowContext, replay +from tests.test_runtime_external_payload_transport import FakeRuntimePayloadServer + + +@pytest.mark.asyncio +async def test_target_requests_and_concurrent_worker_heartbeat_keep_separate_scopes(): + requests = [] + + async def handler(request): + requests.append(request) + await asyncio.sleep(0) + if request.url.path == "/api/cluster/info": + return httpx.Response(200, json={"namespace": {"name": request.headers["x-namespace"]}}) + return httpx.Response(200, json={"accepted": True, "id": "call-1", "status": "started"}) + + async with Client( + "https://runtime.example", namespace="tenant-b", control_token="control", worker_token="worker", + ) as client: + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handler)) + client._cluster_info = {"namespace": {"name": "tenant-b"}} + result, _ = await asyncio.gather( + client.execute_nexus_operation("greeter", "shared", "greet", ["Ada"], target_namespace="shared", + caller_workflow_instance_id="caller", caller_workflow_run_id="original-run"), + client.heartbeat_worker(worker_id="caller-worker"), + ) + assert client.namespace == "tenant-b" + assert client._cluster_info == {"namespace": {"name": "tenant-b"}} + + nexus = next(request for request in requests if request.url.path.endswith("/execute")) + heartbeat = next(request for request in requests if request.url.path == "/api/worker/heartbeat") + assert nexus.headers["x-namespace"] == "shared" + assert nexus.headers["authorization"] == "Bearer control" + assert heartbeat.headers["x-namespace"] == "tenant-b" + assert heartbeat.headers["authorization"] == "Bearer worker" + body = json.loads(nexus.content) + assert body["caller_namespace"] == "tenant-b" + assert body["caller_workflow_instance_id"] == "caller" + assert body["caller_workflow_run_id"] == "original-run" + assert result.target_namespace == "shared" + assert result.to_recorded_payload()["target_namespace"] == "shared" + + +@pytest.mark.asyncio +async def test_cross_namespace_payload_discovery_upload_and_admission_use_target_scope(): + server = FakeRuntimePayloadServer(threshold_bytes=1) + requests = [] + + def handler(request): + requests.append(request) + assert request.headers["x-namespace"] == "shared" + assert request.headers["authorization"] == "Bearer control" + if request.url.path == "/api/cluster/info": + info = server.cluster_info() + info["namespace"]["name"] = "shared" + return httpx.Response(200, json=info) + if request.url.path == "/api/external-payloads/v1": + reference = server.seed(request.content.decode()) + assert int(request.headers["x-durable-workflow-payload-size"]) == reference["size_bytes"] + assert request.headers["x-durable-workflow-payload-sha256"] == reference["sha256"] + return httpx.Response(201, json={ + "schema": "durable-workflow.v2.runtime-external-payload-upload.v1", + "transport_version": 1, "reference": reference, + }) + return httpx.Response(200, json={"accepted": True, "id": "call-1", "status": "started"}) + + async with Client( + "https://runtime.example", namespace="tenant-b", control_token="control", worker_token="worker", + ) as client: + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handler)) + client._cluster_info = {"namespace": {"name": "tenant-b"}} + client._runtime_external_payload_transport_resolved = True + result = await client.execute_nexus_operation( + "greeter", "shared", "greet", ["Ada" * 128], target_namespace="shared", + ) + assert client.namespace == "tenant-b" + assert client._runtime_external_payload_transport_cache is None + + assert [request.url.path for request in requests] == [ + "/api/cluster/info", "/api/external-payloads/v1", + "/api/service-endpoints/greeter/services/shared/operations/greet/execute", + ] + body = json.loads(requests[-1].content) + reference = body["arguments"]["external_payload"] + assert serializer.decode(server.payloads[reference["reference_id"]].decode(), codec="avro") == ["Ada" * 128] + assert body["caller_namespace"] == "tenant-b" + assert serializer.decode(result.request_payload["arguments"], codec="avro") == ["Ada" * 128] + + +@pytest.mark.asyncio +async def test_target_scope_discards_caller_discovery_payload_cache_and_provider(): + client = Client("https://runtime.example", namespace="tenant-b", token="shared-token") + client._cluster_info = {"namespace": {"name": "tenant-b"}} + client._runtime_external_payload_transport_resolved = True + target = client._nexus_target_client("shared") + assert target._http is client._http + assert target.namespace == "shared" + assert target._cluster_info is None + assert target._cluster_info_lock is not client._cluster_info_lock + assert not target._runtime_external_payload_transport_resolved + assert target.external_storage is None + assert target.external_storage_cache is not client.external_storage_cache + await client.aclose() + + +@workflow.defn(name="nexus-target-caller") +class NexusTargetCaller: + def run(self, ctx: WorkflowContext, target: str | None): + result = yield ctx.call_nexus_service("greeter", "shared", "greet", ["Ada"], + target_namespace=target, idempotency_key="original-call") + return result.service_call_id + + +@pytest.mark.parametrize("target", [None, "shared"]) +def test_original_target_replays_and_changed_target_is_refused(target): + command = replay(NexusTargetCaller, [], [target], workflow_id="caller", run_id="original-run").commands[0] + assert isinstance(command, NexusServiceCall) + result = NexusOperationResult(accepted=True, service_call_id="call-1", endpoint_name="greeter", + service_name="shared", operation_name="greet", target_namespace=target, + request_payload={"idempotency_key": command.idempotency_key}, status="completed") + history = [{"event_type": "SideEffectRecorded", "payload": { + "sequence": 1, "payload_codec": "avro", "result": serializer.encode(result.to_recorded_payload(), codec="avro"), + }}] + outcome = replay( + NexusTargetCaller, history, [target], workflow_id="caller", run_id="original-run", payload_codec="avro", + ) + assert isinstance(outcome.commands[0], CompleteWorkflow) + assert outcome.commands[0].result == "call-1" + changed = "other" if target else "shared" + with pytest.raises(NonDeterministicReplayError, match="nexus_target_namespace_changed"): + replay(NexusTargetCaller, history, [changed], workflow_id="caller", run_id="original-run", payload_codec="avro") + + +def test_explicit_target_participates_in_call_identity_and_default_stays_compatible(): + identity = dict(workflow_id="caller", run_id="original-run", ordinal=0, endpoint_name="greeter", + service_name="shared", operation_name="greet", arguments=["Ada"]) + legacy = stable_nexus_idempotency_key(**identity) + assert stable_nexus_idempotency_key(**identity, target_namespace=None) == legacy + assert stable_nexus_idempotency_key(**identity, target_namespace="shared") != legacy + assert stable_nexus_idempotency_key(**identity, target_namespace="shared") != stable_nexus_idempotency_key( + **identity, target_namespace="other") + + +@pytest.mark.asyncio +async def test_empty_target_is_refused_before_network_and_worker_credentials_are_not_reused(): + async with Client("https://runtime.example", namespace="tenant-b", worker_token="worker-only") as client: + with pytest.raises(ValueError, match="nonempty"): + await client.execute_nexus_operation("greeter", "shared", "greet", target_namespace=" ") + with pytest.raises(ValueError, match="control-plane requests require"): + await client.execute_nexus_operation("greeter", "shared", "greet", target_namespace="shared") diff --git a/tests/test_nexus_wire.py b/tests/test_nexus_wire.py new file mode 100644 index 0000000..72e5260 --- /dev/null +++ b/tests/test_nexus_wire.py @@ -0,0 +1,38 @@ +import json +from pathlib import Path + +import pytest + +from durable_workflow import serializer +from durable_workflow.nexus import nexus_request_payload, stable_nexus_idempotency_key + + +def test_nexus_request_matches_the_regression_corpus_wire(): + fixture = json.loads((Path(__file__).parent / "fixtures" / "codec_regressions" + / "nexus-world-arguments-avro.json").read_text()) + request = nexus_request_payload(arguments=["world"]) + assert request["arguments"] == fixture["framing"]["wire_base64"] + assert serializer.decode(request["arguments"], codec=request["payload_codec"]) == ["world"] + + +@pytest.mark.parametrize("arguments", [[], ["world"], [None, True, 42, {"nested": [b"\x00\xff", 1.5]}]]) +def test_nexus_arguments_use_the_portable_avro_payload(arguments): + request = nexus_request_payload(arguments=arguments) + assert request["payload_codec"] == "avro" + assert isinstance(request["arguments"], str) + assert serializer.decode(request["arguments"], codec="avro") == arguments + + +def test_nexus_refuses_json_as_a_workflow_payload_codec(): + with pytest.raises(ValueError, match="unsupported_payload_codec"): + nexus_request_payload(arguments=["world"], payload_codec="json") + + +def test_encoding_does_not_change_the_existing_workflow_call_identity(): + arguments = ["world", {"value": 42}] + identity = dict(workflow_id="caller", run_id="original-run", ordinal=0, + endpoint_name="endpoint", service_name="service", operation_name="greet") + key = stable_nexus_idempotency_key(**identity, arguments=arguments) + request = nexus_request_payload(arguments=arguments, idempotency_key=key) + decoded = serializer.decode(request["arguments"], codec=request["payload_codec"]) + assert request["idempotency_key"] == stable_nexus_idempotency_key(**identity, arguments=decoded) diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 98405b5..dffadaa 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -259,7 +259,18 @@ async def run(self, ctx: WorkflowContext) -> str: return "body was never executed" +@workflow.defn(name="tests.replay.nexus-target") +class NexusTargetWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + result = yield ctx.call_nexus_service( + "greeter", "shared", "greet", ["Ada"], + target_namespace="other", idempotency_key="original-call", + ) + return result.service_call_id + + WORKFLOWS = [ + NexusTargetWorkflow, UnsupportedAsyncWorkflow, ChildPolicyAuthorWorkflow, ColdReplacementSatisfiedConditionWorkflow, diff --git a/tests/test_runtime_external_payload_transport.py b/tests/test_runtime_external_payload_transport.py index 9983f7e..a2f2a44 100644 --- a/tests/test_runtime_external_payload_transport.py +++ b/tests/test_runtime_external_payload_transport.py @@ -309,6 +309,52 @@ def runtime_envelope(server: FakeRuntimePayloadServer, value: Any) -> dict[str, return {"codec": "avro", "external_payload": server.seed(blob)} +@pytest.mark.asyncio +@pytest.mark.parametrize("threshold_bytes", [1, 4096]) +async def test_nexus_arguments_offload_without_changing_recorded_arguments(threshold_bytes: int) -> None: + server = FakeRuntimePayloadServer(threshold_bytes=threshold_bytes) + path = "/api/service-endpoints/greeter/services/shared/operations/greet/execute" + server.responses[("POST", path)] = {"accepted": True, "id": "call-1", "status": "started"} + client = runtime_client(server) + arguments = ["hello" * 32, {"binary": b"\x00\xff"}] + try: + result = await client.execute_nexus_operation( + "greeter", "shared", "greet", arguments, caller_namespace="caller", + idempotency_key="original-call", + ) + finally: + await client.aclose() + + body = next(body for method, route, body in server.requests if method == "POST" and route == path) + if threshold_bytes == 1: + reference = body["arguments"]["external_payload"] + assert body["arguments"]["codec"] == "avro" + assert serializer.decode(server.payloads[reference["reference_id"]].decode(), codec="avro") == arguments + assert server.upload_count == 1 + assert "external_storage" not in json.dumps(body) + else: + assert serializer.decode(body["arguments"], codec="avro") == arguments + assert server.upload_count == 0 + assert body["caller_namespace"] == "caller" + assert body["idempotency_key"] == "original-call" + assert client.namespace == "billing" + assert serializer.decode(result.request_payload["arguments"], codec="avro") == arguments + assert result.request_payload["idempotency_key"] == "original-call" + + +@pytest.mark.asyncio +async def test_nexus_oversized_arguments_fail_before_upload_or_admission() -> None: + server = FakeRuntimePayloadServer(threshold_bytes=1, max_payload_bytes=32) + client = runtime_client(server) + try: + with pytest.raises(ExternalPayloadOversized): + await client.execute_nexus_operation("greeter", "shared", "greet", ["x" * 128]) + finally: + await client.aclose() + assert server.upload_count == 0 + assert not server.requests + + @pytest.mark.asyncio async def test_client_start_and_result_use_runtime_credentials_without_provider_setup() -> None: server = FakeRuntimePayloadServer() diff --git a/tests/test_worker.py b/tests/test_worker.py index 6c26752..13ead6d 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -124,12 +124,13 @@ def run(self, ctx, request): # type: ignore[no-untyped-def] @workflow.defn(name="nexus-worker-wf") class NexusWorkerWorkflow: - def run(self, ctx): # type: ignore[no-untyped-def] + def run(self, ctx, target=None): # type: ignore[no-untyped-def] result = yield ctx.call_nexus_service( "greeter", "shared", "greet", ["Ada"], + target_namespace=target, service_sdk_language="workflow-php", artifact_tuple={"sdk-python": "0.4.95"}, published_artifact_worker_execution=True, @@ -1823,8 +1824,9 @@ async def test_complete_on_resolved_activity(self, mock_client: AsyncMock) -> No assert serializer.decode(commands[0]["result"]["blob"], codec="avro") == "done" @pytest.mark.asyncio + @pytest.mark.parametrize("target", [None, "shared"]) async def test_workflow_nexus_service_call_records_side_effect_and_resumes( - self, mock_client: AsyncMock + self, mock_client: AsyncMock, target: str | None, ) -> None: async def fake_execute(endpoint_name: str, service_name: str, operation_name: str, arguments: list, **kwargs): return NexusOperationResult( @@ -1836,6 +1838,7 @@ async def fake_execute(endpoint_name: str, service_name: str, operation_name: st caller_workflow_instance_id=kwargs["caller_workflow_instance_id"], caller_workflow_run_id=kwargs["caller_workflow_run_id"], service_sdk_language=kwargs["service_sdk_language"], + target_namespace=kwargs["target_namespace"], request_payload={ "arguments": list(arguments), "idempotency_key": kwargs["idempotency_key"], @@ -1858,7 +1861,7 @@ async def fake_execute(endpoint_name: str, service_name: str, operation_name: st "workflow_type": "nexus-worker-wf", "workflow_task_attempt": 1, "history_events": [], - "arguments": serializer.encode([], codec="avro"), + "arguments": serializer.encode([target], codec="avro"), "payload_codec": "avro", } @@ -1869,6 +1872,7 @@ async def fake_execute(endpoint_name: str, service_name: str, operation_name: st assert execute_call.args[:4] == ("greeter", "shared", "greet", ["Ada"]) assert execute_call.kwargs["caller_workflow_instance_id"] == "wf-caller" assert execute_call.kwargs["caller_workflow_run_id"] == "run-caller" + assert execute_call.kwargs["target_namespace"] == target assert execute_call.kwargs["service_sdk_language"] == "workflow-php" assert execute_call.kwargs["raise_on_failure"] is False assert execute_call.kwargs["idempotency_key"].startswith("dw-py-nexus-")