From e10088618929cf90a555479cacca61ea469a9d7f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 18 Oct 2026 00:00:24 -0700 Subject: [PATCH 1/3] Took standalone operation callbacks on a Worker. --- .github/workflows/ci.yml | 7 +- scripts/gen_nexus_system_api.py | 89 +++- scripts/gen_payload_visitor.py | 24 +- temporalio/bridge/_visitor.py | 26 ++ temporalio/client/__init__.py | 2 + temporalio/client/_impl.py | 12 + temporalio/client/_interceptor.py | 2 + temporalio/client/_nexus.py | 53 +++ temporalio/nexus/system/_support.py | 19 + .../system/notification_service/__init__.py | 33 ++ .../system/notification_service/models.py | 416 ++++++++++++++++++ .../system/notification_service/services.py | 30 ++ .../nexus/system/workflow_service/__init__.py | 4 +- temporalio/worker/_nexus.py | 10 +- tests/helpers/nexus.py | 44 +- tests/nexus/test_notification_service.py | 208 +++++++++ tests/nexus/test_standalone_callbacks.py | 214 +++++++++ tests/nexus/test_temporal_system_nexus.py | 10 +- tests/worker/test_visitor.py | 49 +++ 19 files changed, 1233 insertions(+), 19 deletions(-) create mode 100644 temporalio/nexus/system/notification_service/__init__.py create mode 100644 temporalio/nexus/system/notification_service/models.py create mode 100644 temporalio/nexus/system/notification_service/services.py create mode 100644 tests/nexus/test_notification_service.py create mode 100644 tests/nexus/test_standalone_callbacks.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 30d17a80ed..376af0f41b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -233,8 +233,13 @@ jobs: - run: uv add --dev --python 3.10 "googleapis-common-protos==1.70.0" - run: uv add --python 3.10 "protobuf<4" - run: uv sync --all-extras - - run: cargo install --locked nexgen --version 0.2.8 --features advanced --force - run: poe build-develop + # The generators build their pinned nexgen revision on first use, a full + # Rust build, so it's cached by the pin. + - uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 + with: + path: .nexgen + key: nexgen-${{ hashFiles('scripts/gen_nexus_system_api.py') }} - run: poe gen-protos - name: Check generation unchanged run: | diff --git a/scripts/gen_nexus_system_api.py b/scripts/gen_nexus_system_api.py index 1133520700..d3304912c1 100644 --- a/scripts/gen_nexus_system_api.py +++ b/scripts/gen_nexus_system_api.py @@ -11,7 +11,8 @@ base_dir = Path(__file__).parent.parent sys.path.insert(0, str(base_dir)) -wit_input_dir = ( +# The upstream WIT dependencies come from Core's API copy. +wit_deps_dir = ( base_dir / "temporalio" / "bridge" @@ -21,10 +22,19 @@ / "protos" / "api_upstream" / "nexus" + / "deps" ) +# The workflow service WIT with the stream notifier operations, and the notification service WIT, +# come from the temporalio-streams crate, which carries the API they name. +wit_input_dir = gen_protos.streams_proto_dir.parent / "nexus" wit_path = wit_input_dir / "workflow-service.wit" -wit_deps_dir = wit_input_dir / "deps" output_dir = base_dir / "temporalio" / "nexus" / "system" / "workflow_service" +# The service a Worker implements to take NexusHandler callbacks. The server calls +# it, not a Workflow, so it gets plain definitions and no System Nexus caller. +notification_wit_path = wit_input_dir / "notification-service.wit" +notification_output_dir = ( + base_dir / "temporalio" / "nexus" / "system" / "notification_service" +) workflow_init_path = base_dir / "temporalio" / "workflow" / "__init__.py" workflowservice_request_response_proto = ( gen_protos.api_proto_dir @@ -34,28 +44,64 @@ / "v1" / "request_response.proto" ) -NEX_GEN_VERSION = "0.2.8" +notificationservice_request_response_proto = ( + gen_protos.api_proto_dir + / "temporal" + / "api" + / "notificationservice" + / "v1" + / "request_response.proto" +) +workflowservice_stream_request_response_proto = ( + gen_protos.streams_api_proto_dir + / "temporal" + / "api" + / "workflowservice" + / "v1" + / "stream_request_response.proto" +) +notificationservice_progress_proto = ( + gen_protos.nexus_api_proto_dir + / "temporal" + / "api" + / "notificationservice" + / "v1" + / "progress.proto" +) +# The stream bindings need nexgen fixes that no release carries yet, so both generators +# use this pinned revision of the nexgen fork. NEX_GEN_BIN points them at a binary built +# some other way. +NEX_GEN_REPOSITORY = "https://github.com/moetemp/nexgen" +NEX_GEN_REVISION = "3c60ecc9987007d70f8d798d09373445b0219d54" +nex_gen_root = base_dir / ".nexgen" / NEX_GEN_REVISION def nex_gen_command() -> list[str]: if bin_path := os.environ.get("NEX_GEN_BIN"): return [bin_path] - if shutil.which("nexgen") is None: + binary = nex_gen_root / "bin" / "nexgen" + if not binary.exists(): subprocess.check_call( [ "cargo", "install", "--locked", "nexgen", - "--version", - NEX_GEN_VERSION, + "--git", + NEX_GEN_REPOSITORY, + "--rev", + NEX_GEN_REVISION, "--features", "advanced", - "--force", - ] + "--root", + str(nex_gen_root), + ], + # The git CLI honours the user's URL rewrites and credentials, + # which cargo's built-in client does not. + env={**os.environ, "CARGO_NET_GIT_FETCH_WITH_CLI": "true"}, ) - return ["nexgen"] + return [str(binary)] def build_descriptor_set(descriptor_path: Path) -> None: @@ -64,10 +110,15 @@ def build_descriptor_set(descriptor_path: Path) -> None: sys.executable, "-mgrpc_tools.protoc", f"--proto_path={gen_protos.api_proto_dir}", + f"--proto_path={gen_protos.streams_api_proto_dir}", + f"--proto_path={gen_protos.nexus_api_proto_dir}", f"--proto_path={gen_protos.proto_dir}", "--include_imports", f"--descriptor_set_out={descriptor_path}", str(workflowservice_request_response_proto), + str(notificationservice_request_response_proto), + str(workflowservice_stream_request_response_proto), + str(notificationservice_progress_proto), ] ) @@ -113,6 +164,8 @@ def generate_nexus_system_api() -> None: raise RuntimeError(f"missing WIT source: {wit_path}") if not wit_deps_dir.exists(): raise RuntimeError(f"missing WIT dependency directory: {wit_deps_dir}") + if not notification_wit_path.exists(): + raise RuntimeError(f"missing WIT source: {notification_wit_path}") with tempfile.TemporaryDirectory(dir=base_dir) as temp_dir: descriptor_path = Path(temp_dir) / "temporal_api.bin" @@ -138,6 +191,22 @@ def generate_nexus_system_api() -> None: ] ) + shutil.rmtree(notification_output_dir, ignore_errors=True) + subprocess.check_call( + [ + *command, + "python", + str(notification_wit_path), + str(wit_deps_dir), + "--support-package", + "temporalio.nexus.system._support", + "--descriptors", + str(descriptor_path), + "--output", + str(notification_output_dir), + ] + ) + (output_dir.parent / "__init__.py").touch() generate_workflow_exports() subprocess.check_call( @@ -150,6 +219,7 @@ def generate_nexus_system_api() -> None: "I", "--fix", str(output_dir), + str(notification_output_dir), str(workflow_init_path), ] ) @@ -160,6 +230,7 @@ def generate_nexus_system_api() -> None: "ruff", "format", str(output_dir), + str(notification_output_dir), str(workflow_init_path), ] ) diff --git a/scripts/gen_payload_visitor.py b/scripts/gen_payload_visitor.py index 7fab85dcda..f1fb266f62 100644 --- a/scripts/gen_payload_visitor.py +++ b/scripts/gen_payload_visitor.py @@ -2,7 +2,7 @@ import sys from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path -from typing import cast, get_args, get_origin, get_type_hints +from typing import Any, cast, get_args, get_origin, get_type_hints import google.protobuf.message import nexusrpc @@ -22,10 +22,10 @@ from temporalio.converter._payload_converter import _get_transfer_type_converter -def discover_system_nexus_roots() -> list[Descriptor]: - module_path = base_dir / "temporalio" / "nexus" / "system" / "workflow_service" +def load_system_package(name: str) -> Any: + module_path = base_dir / "temporalio" / "nexus" / "system" / name spec = spec_from_file_location( - "temporalio_nexus_system_workflow_service", + f"temporalio_nexus_system_{name}", module_path / "__init__.py", submodule_search_locations=[str(module_path)], ) @@ -34,9 +34,23 @@ def discover_system_nexus_roots() -> list[Descriptor]: module = module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) + return module + +def discover_system_nexus_roots() -> list[Descriptor]: + # A Worker calls the workflow service, and the server calls the + # notification service on a Worker. Both carry system payloads. + services = [ + load_system_package("workflow_service")._services.WorkflowService, + load_system_package("notification_service").NotificationService, + ] roots: list[Descriptor] = [] - for annotation in get_type_hints(module._services.WorkflowService).values(): + annotations = [ + annotation + for service in services + for annotation in get_type_hints(service).values() + ] + for annotation in annotations: if get_origin(annotation) is not nexusrpc.Operation: continue for operation_type in get_args(annotation): diff --git a/temporalio/bridge/_visitor.py b/temporalio/bridge/_visitor.py index d8fa05edf3..d34a6b0559 100644 --- a/temporalio/bridge/_visitor.py +++ b/temporalio/bridge/_visitor.py @@ -650,3 +650,29 @@ async def _visit_temporal_api_workflowservice_v1_SignalWithStartWorkflowExecutio self, fs: VisitorFunctions, o: Any ) -> None: pass + + async def _visit_temporal_api_notificationservice_v1_OnCompleteRequest( + self, fs: VisitorFunctions, o: Any + ): + if o.HasField("source_context"): + await self._visit_temporal_api_common_v1_Payload(fs, o.source_context) + if o.HasField("success"): + await self._visit_temporal_api_common_v1_Payload(fs, o.success) + elif o.HasField("failure"): + await self._visit_temporal_api_failure_v1_Failure(fs, o.failure) + + async def _visit_temporal_api_notificationservice_v1_OnCompleteResponse( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + + async def _visit_temporal_api_notificationservice_v1_OnProgressRequest( + self, fs: VisitorFunctions, o: Any + ): + if o.HasField("source_context"): + await self._visit_temporal_api_common_v1_Payload(fs, o.source_context) + + async def _visit_temporal_api_notificationservice_v1_OnProgressResponse( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass diff --git a/temporalio/client/__init__.py b/temporalio/client/__init__.py index 2eef41a39d..ca687ccb72 100644 --- a/temporalio/client/__init__.py +++ b/temporalio/client/__init__.py @@ -145,6 +145,7 @@ ) from ._nexus import ( NexusClient, + NexusHandlerCallback, NexusOperationExecution, NexusOperationExecutionAsyncIterator, NexusOperationExecutionCancellationInfo, @@ -246,6 +247,7 @@ "AsyncActivityHandle", "ActivityHandle", "NexusClient", + "NexusHandlerCallback", "NexusOperationExecution", "NexusOperationExecutionAsyncIterator", "NexusOperationExecutionCancellationInfo", diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index 0c1e67b85e..df08c55002 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -1610,6 +1610,18 @@ async def start_nexus_operation( for k, v in input.headers.items(): req.nexus_header[k] = v + for callback in input.completion_callbacks: + handler = temporalio.api.common.v1.Callback.NexusHandler( + task_queue_name=callback.task_queue, + service=callback.service, + operation=callback.operation, + progress_operation=callback.progress_operation, + ) + if callback.source_context is not None: + [context] = await data_converter.encode([callback.source_context]) + handler.source_context.CopyFrom(context) + req.completion_callbacks.add(nexus_handler=handler) + resp: temporalio.api.workflowservice.v1.StartNexusOperationExecutionResponse try: resp = await self._client.workflow_service.start_nexus_operation_execution( diff --git a/temporalio/client/_interceptor.py b/temporalio/client/_interceptor.py index 68077ebc93..2d30dbc496 100644 --- a/temporalio/client/_interceptor.py +++ b/temporalio/client/_interceptor.py @@ -31,6 +31,7 @@ AsyncActivityIDReference, ) from ._nexus import ( + NexusHandlerCallback, NexusOperationExecutionAsyncIterator, NexusOperationExecutionCount, NexusOperationExecutionDescription, @@ -605,6 +606,7 @@ class StartNexusOperationInput: headers: Mapping[str, str] rpc_metadata: Mapping[str, str | bytes] rpc_timeout: timedelta | None + completion_callbacks: Sequence[NexusHandlerCallback] = () @dataclass diff --git a/temporalio/client/_nexus.py b/temporalio/client/_nexus.py index 1f0a7338ea..475119b973 100644 --- a/temporalio/client/_nexus.py +++ b/temporalio/client/_nexus.py @@ -464,6 +464,46 @@ def cause(self) -> BaseException: return self.__cause__ +@dataclass(frozen=True) +class NexusHandlerCallback: + """A Worker's Nexus operation that the server calls when an operation completes. + + .. warning:: + This API is experimental and unstable. + + Pass it to :py:meth:`NexusClient.start_operation`. When the operation + completes, the server starts ``operation`` of ``service`` on + ``task_queue``, in the operation's namespace, with an + :py:class:`temporalio.nexus.system.notification_service.OnCompleteRequest`. + With ``progress_operation`` set, it also starts that operation with each + :py:class:`temporalio.nexus.system.notification_service.OnProgressRequest` + the operation reports, any number of times and in any order relative to the + completion. + A Worker serves both with a handler of + :py:class:`temporalio.nexus.system.notification_service.NotificationService`, + and needs no Nexus endpoint for it. + + The server must take the NexusHandler callback kind on a standalone + operation (``nexusoperation.enabledCallbackKinds``). + """ + + task_queue: str + """The task queue of the Worker that takes the calls.""" + + service: str = "temporal.api.notificationservice.v1.NotificationService" + """The Nexus service of both operations.""" + + operation: str = "OnComplete" + """The operation that takes the completion.""" + + progress_operation: str = "" + """The operation that takes the progress. Empty for the completion only.""" + + source_context: Any = None + """A value the server sends back in each request, encoded with the client's + data converter. ``None`` sends none.""" + + class NexusClient(ABC, Generic[NexusServiceType]): """Client for starting standalone Nexus operations. @@ -490,6 +530,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[OutputT]: ... @@ -512,6 +553,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[OutputT]: ... @@ -536,6 +578,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[OutputT]: ... @@ -560,6 +603,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[OutputT]: ... @@ -584,6 +628,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[OutputT]: ... @@ -607,6 +652,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[OutputT]: ... @@ -636,6 +682,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[OutputT]: ... @@ -656,6 +703,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[Any]: @@ -686,6 +734,9 @@ async def start_operation( search_attributes: Search attributes for the operation. summary: Summary for the operation. headers: Headers to attach to the Nexus request. + completion_callbacks: Worker Nexus operations the server calls with + the operation's completion, and with its progress for a callback + that names a progress operation. rpc_metadata: Headers used on the RPC call. rpc_timeout: Optional RPC deadline to set for the RPC call. @@ -974,6 +1025,7 @@ async def start_operation( search_attributes: temporalio.common.TypedSearchAttributes | None = None, summary: str | None = None, headers: Mapping[str, str] | None = None, + completion_callbacks: Sequence[NexusHandlerCallback] = (), rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> NexusOperationHandle[Any]: @@ -1005,6 +1057,7 @@ async def start_operation( headers=dict(headers) if headers else {}, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout, + completion_callbacks=completion_callbacks, ) ) diff --git a/temporalio/nexus/system/_support.py b/temporalio/nexus/system/_support.py index 49ae624300..99ed771739 100644 --- a/temporalio/nexus/system/_support.py +++ b/temporalio/nexus/system/_support.py @@ -6,6 +6,7 @@ import temporalio.api.common.v1.message_pb2 as common_pb2 import temporalio.api.enums.v1.workflow_pb2 as workflow_enums_pb2 +import temporalio.api.failure.v1.message_pb2 as failure_pb2 import temporalio.api.taskqueue.v1.message_pb2 as taskqueue_pb2 import temporalio.api.workflow.v1 import temporalio.common @@ -152,6 +153,24 @@ def payload_to_proto( return _value_to_payload(payload) +def failure_from_proto( + proto: failure_pb2.Failure, +) -> BaseException: + return temporalio.nexus.system._current_user_failure_converter().from_failure( + proto, temporalio.nexus.system._current_user_payload_converter() + ) + + +def failure_to_proto( + error: BaseException, +) -> failure_pb2.Failure: + failure = failure_pb2.Failure() + temporalio.nexus.system._current_user_failure_converter().to_failure( + error, temporalio.nexus.system._current_user_payload_converter(), failure + ) + return failure + + def memo_from_proto( proto: common_pb2.Memo, ) -> collections.abc.Mapping[str, object]: diff --git a/temporalio/nexus/system/notification_service/__init__.py b/temporalio/nexus/system/notification_service/__init__.py new file mode 100644 index 0000000000..68ae254475 --- /dev/null +++ b/temporalio/nexus/system/notification_service/__init__.py @@ -0,0 +1,33 @@ +# Generated by nexgen v0.2.8. DO NOT EDIT! + +from __future__ import annotations + +from .models import ( + NexusOperationProgress, + NexusOperationProgressOperation, + NexusOperationProgressOperationKey, + NexusOperationProgressOperationOperationKey, + OnCompleteRequest, + OnCompleteRequestResult, + OnCompleteRequestResultFailure, + OnCompleteRequestResultSuccess, + OnCompleteResponse, + OnProgressRequest, + OnProgressResponse, +) +from .services import NotificationService + +__all__ = [ + "NexusOperationProgress", + "NexusOperationProgressOperation", + "NexusOperationProgressOperationOperationKey", + "NexusOperationProgressOperationKey", + "OnCompleteRequest", + "OnCompleteRequestResult", + "OnCompleteRequestResultSuccess", + "OnCompleteRequestResultFailure", + "OnCompleteResponse", + "OnProgressRequest", + "OnProgressResponse", + "NotificationService", +] diff --git a/temporalio/nexus/system/notification_service/models.py b/temporalio/nexus/system/notification_service/models.py new file mode 100644 index 0000000000..a184913a70 --- /dev/null +++ b/temporalio/nexus/system/notification_service/models.py @@ -0,0 +1,416 @@ +# Generated by nexgen v0.2.8. DO NOT EDIT! + +from __future__ import annotations + +import dataclasses +import typing + +import typing_extensions + +import temporalio.api.nexus.v1.progress_pb2 +import temporalio.api.notificationservice.v1.progress_pb2 +import temporalio.api.notificationservice.v1.request_response_pb2 +import temporalio.converter +import temporalio.nexus.system._support as _support + + +@dataclasses.dataclass(slots=True) +class OnCompleteRequest: + """ + .. warning:: + This API is experimental and subject to change. + """ + + result: OnCompleteRequestResult + source_context: typing.Any | None = None + receiver_state: bytes | None = None + + +class _OnCompleteRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + OnCompleteRequest, + temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest, + ] +): + transfer_type: ( + type[ + temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest + ] + | None + ) = temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest, + type_hint: type[OnCompleteRequest], + ) -> OnCompleteRequest: + _oneof_result: OnCompleteRequestResultSuccess | OnCompleteRequestResultFailure + match value.WhichOneof("result"): + case "success": + _oneof_result = OnCompleteRequestResultSuccess( + _support.payload_from_proto(value.success) + ) + case "failure": + _oneof_result = OnCompleteRequestResultFailure( + _support.failure_from_proto(value.failure) + ) + case None: + raise ValueError("missing required field OnCompleteRequest.result") + return OnCompleteRequest( + result=_oneof_result, + source_context=_support.payload_from_proto(value.source_context) + if value.HasField("source_context") + else None, + receiver_state=value.receiver_state if bool(value.receiver_state) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: OnCompleteRequest, + ) -> temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest: + message = temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest() + match value.result: + case OnCompleteRequestResultSuccess(): + message.success.CopyFrom(_support.payload_to_proto(value.result.value)) + case OnCompleteRequestResultFailure(): + message.failure.CopyFrom(_support.failure_to_proto(value.result.value)) + if value.source_context is not None: + message.source_context.CopyFrom( + _support.payload_to_proto(value.source_context) + ) + if value.receiver_state is not None: + message.receiver_state = value.receiver_state + return message + + +temporalio.converter.transfer_type_convertible(_OnCompleteRequestTransferTypeConverter)( + OnCompleteRequest +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class OnCompleteResponse: + """ + .. warning:: + This API is experimental and subject to change. + """ + + pass + + +class _OnCompleteResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + OnCompleteResponse, + temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse, + ] +): + transfer_type: ( + type[ + temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse + ] + | None + ) = temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse, + type_hint: type[OnCompleteResponse], + ) -> OnCompleteResponse: + return OnCompleteResponse() + + @typing_extensions.override + def to_transfer_type( + self, + value: OnCompleteResponse, + ) -> temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse: + message = temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse() + return message + + +temporalio.converter.transfer_type_convertible( + _OnCompleteResponseTransferTypeConverter +)(OnCompleteResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class OnProgressRequest: + """ + .. warning:: + This API is experimental and subject to change. + """ + + progress: NexusOperationProgress | None = None + source_context: typing.Any | None = None + operation_token: str | None = None + receiver_state: bytes | None = None + + +class _OnProgressRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + OnProgressRequest, + temporalio.api.notificationservice.v1.progress_pb2.OnProgressRequest, + ] +): + transfer_type: ( + type[temporalio.api.notificationservice.v1.progress_pb2.OnProgressRequest] + | None + ) = temporalio.api.notificationservice.v1.progress_pb2.OnProgressRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.notificationservice.v1.progress_pb2.OnProgressRequest, + type_hint: type[OnProgressRequest], + ) -> OnProgressRequest: + return OnProgressRequest( + progress=_NexusOperationProgressTransferTypeConverter().from_transfer_type( + value.progress, NexusOperationProgress + ) + if value.HasField("progress") + else None, + source_context=_support.payload_from_proto(value.source_context) + if value.HasField("source_context") + else None, + operation_token=value.operation_token + if bool(value.operation_token) + else None, + receiver_state=value.receiver_state if bool(value.receiver_state) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: OnProgressRequest, + ) -> temporalio.api.notificationservice.v1.progress_pb2.OnProgressRequest: + message = temporalio.api.notificationservice.v1.progress_pb2.OnProgressRequest() + if value.progress is not None: + message.progress.CopyFrom( + _NexusOperationProgressTransferTypeConverter().to_transfer_type( + value.progress + ) + ) + if value.source_context is not None: + message.source_context.CopyFrom( + _support.payload_to_proto(value.source_context) + ) + if value.operation_token is not None: + message.operation_token = value.operation_token + if value.receiver_state is not None: + message.receiver_state = value.receiver_state + return message + + +temporalio.converter.transfer_type_convertible(_OnProgressRequestTransferTypeConverter)( + OnProgressRequest +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class NexusOperationProgress: + """ + .. warning:: + This API is experimental and subject to change. + """ + + operation: NexusOperationProgressOperation | None = None + position: str | None = None + counter: int | None = None + metadata: dict[str, str] | None = dataclasses.field(default_factory=dict) + + +class _NexusOperationProgressTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + NexusOperationProgress, + temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress, + ] +): + transfer_type: ( + type[temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress] | None + ) = temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress, + type_hint: type[NexusOperationProgress], + ) -> NexusOperationProgress: + _oneof_operation: NexusOperationProgressOperationOperationKey | None + match value.WhichOneof("operation"): + case "operation_key": + _oneof_operation = NexusOperationProgressOperationOperationKey( + _NexusOperationProgressOperationKeyTransferTypeConverter().from_transfer_type( + value.operation_key, NexusOperationProgressOperationKey + ) + ) + case None: + _oneof_operation = None + return NexusOperationProgress( + operation=_oneof_operation, + position=value.position if bool(value.position) else None, + counter=value.counter if value.counter != 0 else None, + metadata={key: item for key, item in value.metadata.items()}, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: NexusOperationProgress, + ) -> temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress: + message = temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress() + if value.operation is not None: + match value.operation: + case NexusOperationProgressOperationOperationKey(): + message.operation_key.CopyFrom( + _NexusOperationProgressOperationKeyTransferTypeConverter().to_transfer_type( + value.operation.value + ) + ) + if value.position is not None: + message.position = value.position + if value.counter is not None: + message.counter = value.counter + if value.metadata: + message.metadata.update(value.metadata) + return message + + +temporalio.converter.transfer_type_convertible( + _NexusOperationProgressTransferTypeConverter +)(NexusOperationProgress) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class NexusOperationProgressOperationKey: + endpoint: str | None = None + service: str | None = None + operation: str | None = None + operation_token: str | None = None + + +class _NexusOperationProgressOperationKeyTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + NexusOperationProgressOperationKey, + temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress.OperationKey, + ] +): + transfer_type: ( + type[temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress.OperationKey] + | None + ) = temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress.OperationKey + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress.OperationKey, + type_hint: type[NexusOperationProgressOperationKey], + ) -> NexusOperationProgressOperationKey: + return NexusOperationProgressOperationKey( + endpoint=value.endpoint if bool(value.endpoint) else None, + service=value.service if bool(value.service) else None, + operation=value.operation if bool(value.operation) else None, + operation_token=value.operation_token + if bool(value.operation_token) + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: NexusOperationProgressOperationKey, + ) -> temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress.OperationKey: + message = ( + temporalio.api.nexus.v1.progress_pb2.NexusOperationProgress.OperationKey() + ) + if value.endpoint is not None: + message.endpoint = value.endpoint + if value.service is not None: + message.service = value.service + if value.operation is not None: + message.operation = value.operation + if value.operation_token is not None: + message.operation_token = value.operation_token + return message + + +temporalio.converter.transfer_type_convertible( + _NexusOperationProgressOperationKeyTransferTypeConverter +)(NexusOperationProgressOperationKey) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class OnProgressResponse: + """ + .. warning:: + This API is experimental and subject to change. + """ + + receiver_state: bytes | None = None + behind: bool | None = None + + +class _OnProgressResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + OnProgressResponse, + temporalio.api.notificationservice.v1.progress_pb2.OnProgressResponse, + ] +): + transfer_type: ( + type[temporalio.api.notificationservice.v1.progress_pb2.OnProgressResponse] + | None + ) = temporalio.api.notificationservice.v1.progress_pb2.OnProgressResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.notificationservice.v1.progress_pb2.OnProgressResponse, + type_hint: type[OnProgressResponse], + ) -> OnProgressResponse: + return OnProgressResponse( + receiver_state=value.receiver_state if bool(value.receiver_state) else None, + behind=value.behind if bool(value.behind) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: OnProgressResponse, + ) -> temporalio.api.notificationservice.v1.progress_pb2.OnProgressResponse: + message = ( + temporalio.api.notificationservice.v1.progress_pb2.OnProgressResponse() + ) + if value.receiver_state is not None: + message.receiver_state = value.receiver_state + if value.behind is not None: + message.behind = value.behind + return message + + +temporalio.converter.transfer_type_convertible( + _OnProgressResponseTransferTypeConverter +)(OnProgressResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class OnCompleteRequestResultSuccess: + value: typing.Any + + +@dataclasses.dataclass(slots=True) +class OnCompleteRequestResultFailure: + value: BaseException + + +OnCompleteRequestResult = ( + OnCompleteRequestResultSuccess | OnCompleteRequestResultFailure +) + + +@dataclasses.dataclass(slots=True) +class NexusOperationProgressOperationOperationKey: + value: NexusOperationProgressOperationKey + + +NexusOperationProgressOperation = NexusOperationProgressOperationOperationKey diff --git a/temporalio/nexus/system/notification_service/services.py b/temporalio/nexus/system/notification_service/services.py new file mode 100644 index 0000000000..d36eb077de --- /dev/null +++ b/temporalio/nexus/system/notification_service/services.py @@ -0,0 +1,30 @@ +# Generated by nexgen v0.2.8. DO NOT EDIT! + +from __future__ import annotations + +from nexusrpc import Operation, service + +from .models import ( + OnCompleteRequest, + OnCompleteResponse, + OnProgressRequest, + OnProgressResponse, +) + + +@service(name="temporal.api.notificationservice.v1.NotificationService") +class NotificationService: + """ + .. warning:: + This API is experimental and subject to change. + """ + + on_complete: Operation[ + OnCompleteRequest, + OnCompleteResponse, + ] = Operation(name="OnComplete") + + on_progress: Operation[ + OnProgressRequest, + OnProgressResponse, + ] = Operation(name="OnProgress") diff --git a/temporalio/nexus/system/workflow_service/__init__.py b/temporalio/nexus/system/workflow_service/__init__.py index f6a6f0c682..0695ad9b8e 100644 --- a/temporalio/nexus/system/workflow_service/__init__.py +++ b/temporalio/nexus/system/workflow_service/__init__.py @@ -40,7 +40,9 @@ def __init__( ) -__nexus_operation_registry__ = { +__nexus_operation_registry__: dict[ + tuple[str, str], _NexusOperationInfo[typing.Any, typing.Any] +] = { ( "temporal.api.workflowservice.v1.WorkflowService", "SignalWithStartWorkflowExecution", diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index a2f4b8ca75..3d494c444a 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -471,7 +471,15 @@ async def _start_operation( ) ) elif isinstance(result, nexusrpc.handler.StartOperationResultSync): - [payload] = data_converter.payload_converter.to_payloads([result.value]) + payload_converter = data_converter.payload_converter + if temporalio.nexus.system._is_system_payload(start_request.payload): + # The server sent a system payload, such as a NexusHandler + # callback's progress, and reads the answer without a codec. + payload_converter = temporalio.nexus.system._get_payload_converter( + data_converter.payload_converter, + data_converter.failure_converter, + ) + [payload] = payload_converter.to_payloads([result.value]) return temporalio.api.nexus.v1.StartOperationResponse( sync_success=temporalio.api.nexus.v1.StartOperationResponse.Sync( payload=payload, diff --git a/tests/helpers/nexus.py b/tests/helpers/nexus.py index 0af468b44f..834e1eaf62 100644 --- a/tests/helpers/nexus.py +++ b/tests/helpers/nexus.py @@ -1,8 +1,13 @@ -from collections.abc import Sequence +import asyncio +from collections.abc import Callable, Sequence +from typing import Any import temporalio.api.common.v1 import temporalio.api.enums.v1 import temporalio.api.history.v1 +import temporalio.api.workflowservice.v1 +import temporalio.client +import temporalio.service from temporalio.client import WorkflowHistory @@ -100,3 +105,40 @@ def assert_links_match( def _link_sort_key(link: temporalio.api.common.v1.Link) -> bytes: return link.SerializeToString(deterministic=True) + + +async def end_standalone_operation( + client: temporalio.client.Client, + handle: temporalio.client.NexusOperationHandle[Any], + served: Callable[[], bool], + wait: float = 10.0, +) -> None: + """Leave no standalone operation, nor a callback of one, behind a test. + + Call it while the Worker that takes the operation's callbacks still runs. + A callback whose Worker is gone is retried against a task queue no one + polls, which slows every later delivery on the server. So the operation is + terminated, and if ``served`` doesn't turn true within ``wait`` seconds, as + on a server that never calls back, the operation is deleted with its + callbacks. + """ + try: + await handle.terminate(reason="the test ended") + except temporalio.service.RPCError: + pass + loop = asyncio.get_running_loop() + deadline = loop.time() + wait + while not served() and loop.time() < deadline: + await asyncio.sleep(0.1) + if served(): + return + try: + await client.workflow_service.delete_nexus_operation_execution( + temporalio.api.workflowservice.v1.DeleteNexusOperationExecutionRequest( + namespace=client.namespace, + operation_id=handle.operation_id, + run_id=handle.run_id or "", + ) + ) + except temporalio.service.RPCError: + pass diff --git a/tests/nexus/test_notification_service.py b/tests/nexus/test_notification_service.py new file mode 100644 index 0000000000..5fc25165df --- /dev/null +++ b/tests/nexus/test_notification_service.py @@ -0,0 +1,208 @@ +"""A Worker that implements the generated notification service. + +The server delivers a NexusHandler callback's progress and completion as Nexus +tasks with no endpoint. Each input is a system payload, binary protobuf that +the Worker's codec must not touch, which holds payloads the client encoded +with that codec. The server reads each answer as a protobuf, so the answer +must not go through the codec either. +""" + +from __future__ import annotations + +from collections.abc import MutableSequence, Sequence +from unittest import mock + +import nexusrpc +from nexusrpc.handler import StartOperationContext, service_handler, sync_operation + +import temporalio.api.nexus.v1 +import temporalio.converter +from temporalio.api.common.v1 import Payload +from temporalio.api.notificationservice import v1 as notificationservice +from temporalio.bridge._visitor import PayloadVisitor +from temporalio.bridge._visitor_functions import VisitorFunctions +from temporalio.bridge.proto.nexus import NexusTaskCompletion +from temporalio.nexus.system.notification_service import ( + NotificationService, + OnCompleteRequest, + OnCompleteRequestResultSuccess, + OnCompleteResponse, + OnProgressRequest, + OnProgressResponse, +) +from temporalio.worker._nexus import _NexusTaskCancellation, _NexusWorker + +SERVICE = "temporal.api.notificationservice.v1.NotificationService" + + +class _TestCodec(temporalio.converter.PayloadCodec): + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload(metadata={"test-codec": b"true"}, data=p.SerializeToString()) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + decoded = [] + for payload in payloads: + if payload.metadata.get("test-codec") != b"true": + raise RuntimeError("not encoded with the test codec") + decoded.append(Payload.FromString(payload.data)) + return decoded + + +_CONVERTER = temporalio.converter.DataConverter(payload_codec=_TestCodec()) + + +@service_handler(service=NotificationService) +class _Receiver: + def __init__(self) -> None: + self.progress: list[OnProgressRequest] = [] + self.completions: list[OnCompleteRequest] = [] + + @sync_operation + async def on_progress( + self, _ctx: StartOperationContext, input: OnProgressRequest + ) -> OnProgressResponse: + self.progress.append(input) + return OnProgressResponse(receiver_state=b"next", behind=True) + + @sync_operation + async def on_complete( + self, _ctx: StartOperationContext, input: OnCompleteRequest + ) -> OnCompleteResponse: + self.completions.append(input) + return OnCompleteResponse() + + +def _worker(receiver: _Receiver) -> _NexusWorker: + return _NexusWorker( + bridge_worker=lambda: mock.MagicMock(), + client=mock.MagicMock(namespace="default"), + namespace="default", + task_queue="tq", + service_handlers=[receiver], + data_converter=_CONVERTER, + interceptors=[], + metric_meter=mock.MagicMock(), + executor=None, + ) + + +async def _client_payload(value: str) -> Payload: + # What a client puts in the callback's source context or a result. + [payload] = await _CONVERTER.encode([value]) + return payload + + +def _server_payload( + request: notificationservice.OnProgressRequest + | notificationservice.OnCompleteRequest, +) -> Payload: + # The server's EncodeSystemPayload. + return Payload( + metadata={ + "encoding": b"binary/protobuf", + "messageType": request.DESCRIPTOR.full_name.encode(), + "__temporal_system_payload": b"true", + }, + data=request.SerializeToString(), + ) + + +async def _deliver( + worker: _NexusWorker, operation: str, payload: Payload +) -> temporalio.api.nexus.v1.StartOperationResponse: + return await worker._start_operation( + temporalio.api.nexus.v1.StartOperationRequest( + service=SERVICE, operation=operation, payload=payload + ), + headers={}, + cancellation=_NexusTaskCancellation(), + request_deadline=None, + endpoint="", + data_converter=worker._data_converter_for_nexus_task("", SERVICE, operation), + ) + + +class _Codec(VisitorFunctions): + """Runs the codec over a task completion, as the Worker does before it sends it.""" + + async def visit_payload(self, payload: Payload) -> None: + [encoded] = await _CONVERTER.encode([payload]) + payload.CopyFrom(encoded) + + async def visit_payloads(self, payloads: MutableSequence[Payload]) -> None: + for payload in payloads: + await self.visit_payload(payload) + + +async def _as_sent(response: temporalio.api.nexus.v1.StartOperationResponse) -> Payload: + completion = NexusTaskCompletion(completed=temporalio.api.nexus.v1.Response()) + completion.completed.start_operation.CopyFrom(response) + await PayloadVisitor().visit(_Codec(), completion) + return completion.completed.start_operation.sync_success.payload + + +def test_the_generated_definition_names_the_wire_operations() -> None: + definition = nexusrpc.get_service_definition(NotificationService) + assert definition is not None + assert definition.name == SERVICE + progress = definition.operation_definitions["OnProgress"] + assert progress.input_type is OnProgressRequest + assert progress.output_type is OnProgressResponse + complete = definition.operation_definitions["OnComplete"] + assert complete.input_type is OnCompleteRequest + assert complete.output_type is OnCompleteResponse + + +async def test_a_receiver_decodes_progress_as_the_server_encodes_it() -> None: + receiver = _Receiver() + request = notificationservice.OnProgressRequest( + source_context=await _client_payload("orders"), + operation_token="token", + receiver_state=b"state", + ) + request.progress.position = "7" + request.progress.counter = 3 + + response = await _deliver(_worker(receiver), "OnProgress", _server_payload(request)) + + assert response.HasField("sync_success"), response + [got] = receiver.progress + assert got.operation_token == "token" + assert got.receiver_state == b"state" + assert got.progress is not None + assert got.progress.position == "7" + assert got.progress.counter == 3 + # The codec decoded the client's payload inside the system payload, and the + # payload converter decoded the value. + assert got.source_context == "orders" + + sent = await _as_sent(response) + # The server reads the answer as a protobuf, so the codec leaves it alone. + assert sent.metadata["encoding"] == b"binary/protobuf" + assert "test-codec" not in sent.metadata + assert notificationservice.OnProgressResponse.FromString( + sent.data + ) == notificationservice.OnProgressResponse(receiver_state=b"next", behind=True) + + +async def test_a_receiver_decodes_a_completion_as_the_server_encodes_it() -> None: + receiver = _Receiver() + request = notificationservice.OnCompleteRequest( + success=await _client_payload("done"), + source_context=await _client_payload("orders"), + receiver_state=b"state", + ) + + response = await _deliver(_worker(receiver), "OnComplete", _server_payload(request)) + + assert response.HasField("sync_success"), response + [got] = receiver.completions + assert got.result == OnCompleteRequestResultSuccess("done") + assert got.source_context == "orders" + assert got.receiver_state == b"state" + sent = await _as_sent(response) + assert sent.metadata["encoding"] == b"binary/protobuf" + assert "test-codec" not in sent.metadata diff --git a/tests/nexus/test_standalone_callbacks.py b/tests/nexus/test_standalone_callbacks.py new file mode 100644 index 0000000000..83276ddda1 --- /dev/null +++ b/tests/nexus/test_standalone_callbacks.py @@ -0,0 +1,214 @@ +"""NexusHandler completion callbacks on a standalone Nexus operation. + +The server takes callbacks of the NexusHandler kind on a standalone start only +with ``nexusoperation.enabledCallbackKinds`` set, so the live tests skip on a +server that refuses them. The streams live lane turns the setting on and fails +on that skip. +""" + +from __future__ import annotations + +import asyncio +import uuid +from collections.abc import Sequence +from datetime import timedelta +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest +from nexusrpc.handler import StartOperationContext, service_handler, sync_operation + +import temporalio.api.workflowservice.v1 +import temporalio.converter +from temporalio.api.common.v1 import Payload +from temporalio.client import ( + Client, + NexusClient, + NexusHandlerCallback, + NexusOperationHandle, +) +from temporalio.nexus.system.notification_service import ( + NotificationService, + OnCompleteRequest, + OnCompleteRequestResultSuccess, + OnCompleteResponse, + OnProgressRequest, + OnProgressResponse, +) +from temporalio.service import RPCError, RPCStatusCode, ServiceClient +from temporalio.testing import WorkflowEnvironment +from temporalio.worker import Worker +from tests.helpers.nexus import end_standalone_operation, make_nexus_endpoint_name + +# The message a live test skips with when the server refuses the callback or +# never delivers it. The streams live lane fails on it. +NO_HANDLER_CALLBACKS = "no NexusHandler callbacks on a standalone operation" + + +class _TestCodec(temporalio.converter.PayloadCodec): + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload(metadata={"test-codec": b"true"}, data=p.SerializeToString()) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [Payload.FromString(p.data) for p in payloads] + + +def _captured_client() -> tuple[Client, dict[str, Any]]: + captured: dict[str, Any] = {} + + async def start(req: Any, **_kwargs: Any) -> Any: + captured["start"] = req + return temporalio.api.workflowservice.v1.StartNexusOperationExecutionResponse( + run_id="run" + ) + + service_client = Mock(spec=ServiceClient) + service_client.workflow_service = Mock() + service_client.workflow_service.start_nexus_operation_execution = AsyncMock( + side_effect=start + ) + service_client.config = Mock(identity="test-identity") + client = Client( + service_client=service_client, + namespace="test-namespace", + data_converter=temporalio.converter.DataConverter(payload_codec=_TestCodec()), + ) + return client, captured + + +async def test_a_start_carries_each_callback_with_its_encoded_context() -> None: + client, captured = _captured_client() + nexus_client: NexusClient[Any] = client.create_nexus_client( + service="Orders", endpoint="orders" + ) + + await nexus_client.start_operation( + "watch", + "order-1", + id="op-1", + completion_callbacks=[ + NexusHandlerCallback( + task_queue="app", + progress_operation="OnProgress", + source_context={"follower": "orders"}, + ), + NexusHandlerCallback(task_queue="audit", service="Audit", operation="Done"), + ], + ) + + request = captured["start"] + first, second = request.completion_callbacks + handler = first.nexus_handler + assert handler.task_queue_name == "app" + assert handler.service == "temporal.api.notificationservice.v1.NotificationService" + assert handler.operation == "OnComplete" + assert handler.progress_operation == "OnProgress" + # The client's codec encodes the context, as it does the input. + assert handler.source_context.metadata["test-codec"] == b"true" + [context] = await client.data_converter.decode([handler.source_context]) + assert context == {"follower": "orders"} + other = second.nexus_handler + assert (other.task_queue_name, other.service, other.operation) == ( + "audit", + "Audit", + "Done", + ) + assert other.progress_operation == "" + assert not other.HasField("source_context") + + +async def test_a_start_without_callbacks_sends_none() -> None: + client, captured = _captured_client() + nexus_client: NexusClient[Any] = client.create_nexus_client( + service="Orders", endpoint="orders" + ) + + await nexus_client.start_operation("watch", "order-1", id="op-1") + + assert not captured["start"].completion_callbacks + + +@service_handler(name="StandaloneCallbackEcho") +class _Echo: + @sync_operation + async def echo(self, _ctx: StartOperationContext, input: str) -> str: + return input + + +@service_handler(service=NotificationService) +class _Receiver: + def __init__(self) -> None: + self.endpoints: list[str] = [] + self.completions: asyncio.Queue[OnCompleteRequest] = asyncio.Queue() + + @sync_operation + async def on_progress( + self, _ctx: StartOperationContext, input: OnProgressRequest + ) -> OnProgressResponse: + _ = input + return OnProgressResponse() + + @sync_operation + async def on_complete( + self, _ctx: StartOperationContext, input: OnCompleteRequest + ) -> OnCompleteResponse: + import temporalio.nexus + + self.endpoints.append(temporalio.nexus.info().endpoint) + await self.completions.put(input) + return OnCompleteResponse() + + +async def test_live_a_worker_gets_the_completion_of_a_standalone_operation( + client: Client, env: WorkflowEnvironment +) -> None: + if env.supports_time_skipping: + pytest.skip("Standalone Nexus Operation tests don't work with time-skipping") + task_queue = str(uuid.uuid4()) + endpoint = make_nexus_endpoint_name(task_queue) + receiver = _Receiver() + async with Worker( + client, + task_queue=task_queue, + nexus_service_handlers=[_Echo(), receiver], + ): + await env.create_nexus_endpoint(endpoint, task_queue) + nexus_client: NexusClient[Any] = client.create_nexus_client( + service="StandaloneCallbackEcho", endpoint=endpoint + ) + try: + handle: NexusOperationHandle[str] = await nexus_client.start_operation( + "echo", + "hello", + id=str(uuid.uuid4()), + schedule_to_close_timeout=timedelta(seconds=30), + completion_callbacks=[ + NexusHandlerCallback( + task_queue=task_queue, source_context="context" + ) + ], + ) + except RPCError as err: + if err.status == RPCStatusCode.INVALID_ARGUMENT and "allback" in str(err): + pytest.skip(f"{NO_HANDLER_CALLBACKS}: {err}") + raise + try: + assert await handle.result() == "hello" + try: + completion = await asyncio.wait_for(receiver.completions.get(), 30) + except asyncio.TimeoutError: + # A server without the callback kind may keep the callback and never + # call it. + pytest.skip(f"{NO_HANDLER_CALLBACKS}: no OnComplete within 30s") + finally: + await end_standalone_operation( + client, handle, lambda: bool(receiver.endpoints) + ) + + # The server sends the task with no endpoint, which the Worker serves. + assert receiver.endpoints == [""] + assert completion.result == OnCompleteRequestResultSuccess("hello") + assert completion.source_context == "context" diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index 7035a8461f..ef77000cdd 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -614,10 +614,16 @@ def _build_proto_sample(message_type: type[Message]) -> Message: return message -def _populate_proto_sample(message: Message, *, path: str = "value") -> None: +def _populate_proto_sample( + message: Message, *, path: str = "value", active: frozenset[str] = frozenset() +) -> None: + # A recursive type such as Failure.cause is filled once, so the sample ends. + active = active | {message.DESCRIPTOR.full_name} seen_oneofs: set[str] = set() for raw_field in message.DESCRIPTOR.fields: field = cast(FieldDescriptor, raw_field) + if field.message_type is not None and field.message_type.full_name in active: + continue if field.containing_oneof is not None: if field.containing_oneof.name in seen_oneofs: continue @@ -632,6 +638,7 @@ def _populate_proto_sample(message: Message, *, path: str = "value") -> None: _populate_proto_sample( getattr(message, field.name).add(), path=f"{path}.{field.name}[0]", + active=active, ) else: getattr(message, field.name).append( @@ -641,6 +648,7 @@ def _populate_proto_sample(message: Message, *, path: str = "value") -> None: _populate_proto_sample( getattr(message, field.name), path=f"{path}.{field.name}", + active=active, ) else: setattr( diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py index 0cf1ad3283..139760e8d8 100644 --- a/tests/worker/test_visitor.py +++ b/tests/worker/test_visitor.py @@ -6,6 +6,7 @@ import pytest from google.protobuf.duration_pb2 import Duration +import temporalio.api.notificationservice.v1.request_response_pb2 as notification_pb2 import temporalio.api.workflowservice.v1.request_response_pb2 as workflowservice_pb2 import temporalio.bridge.worker import temporalio.converter @@ -363,6 +364,54 @@ async def test_unknown_system_nexus_payload_raises_application_error(): assert not err.value.non_retryable +async def test_a_notification_progress_input_is_visited(): + # The server sends OnProgress to a Worker as a system payload. The source context + # inside it was encoded by the client, so the codec has to reach it. + data_converter = temporalio.converter.default() + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, + data_converter.failure_converter, + ) + request = notification_pb2.OnProgressRequest( + source_context=Payload(data=b"context"), + operation_token="token", + receiver_state=b"state", + ) + request.progress.position = "3" + request.progress.counter = 3 + payload = payload_converter.to_payload(request) + assert payload is not None + + await PayloadVisitor().visit(Visitor(), payload) + + decoded = payload_converter.from_payload(payload) + assert isinstance(decoded, notification_pb2.OnProgressRequest) + assert decoded.source_context.metadata["visited"] == b"True" + assert decoded.receiver_state == b"state" + + +async def test_a_notification_completion_input_is_visited(): + data_converter = temporalio.converter.default() + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, + data_converter.failure_converter, + ) + request = notification_pb2.OnCompleteRequest( + success=Payload(data=b"result"), + source_context=Payload(data=b"context"), + receiver_state=b"state", + ) + payload = payload_converter.to_payload(request) + assert payload is not None + + await PayloadVisitor().visit(Visitor(), payload) + + decoded = payload_converter.from_payload(payload) + assert isinstance(decoded, notification_pb2.OnCompleteRequest) + assert decoded.success.metadata["visited"] == b"True" + assert decoded.source_context.metadata["visited"] == b"True" + + async def test_concurrent_throughput(): """Demonstrate that concurrent visitation is faster than serialized for I/O-bound codecs.""" N_CMDS = 10 From 7d455e1416d51b8a12349d1082b13469679031b0 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 18 Oct 2026 00:57:23 -0700 Subject: [PATCH 2/3] Dropped the visitor test of the removed OnProgress op. The notifier ops left the system Nexus service, so OnProgressRequest isn't generated anymore. The completion test still covers the system payload path. --- tests/worker/test_visitor.py | 26 -------------------------- 1 file changed, 26 deletions(-) diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py index cb47d87346..14ec122689 100644 --- a/tests/worker/test_visitor.py +++ b/tests/worker/test_visitor.py @@ -337,32 +337,6 @@ async def test_unknown_system_nexus_payload_raises_application_error(): assert not err.value.non_retryable -async def test_a_notification_progress_input_is_visited(): - # The server sends OnProgress to a Worker as a system payload. The source context - # inside it was encoded by the client, so the codec has to reach it. - data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_payload_converter( - data_converter.payload_converter, - data_converter.failure_converter, - ) - request = notification_pb2.OnProgressRequest( - source_context=Payload(data=b"context"), - operation_token="token", - receiver_state=b"state", - ) - request.progress.position = "3" - request.progress.counter = 3 - payload = payload_converter.to_payload(request) - assert payload is not None - - await PayloadVisitor().visit(Visitor(), payload) - - decoded = payload_converter.from_payload(payload) - assert isinstance(decoded, notification_pb2.OnProgressRequest) - assert decoded.source_context.metadata["visited"] == b"True" - assert decoded.receiver_state == b"state" - - async def test_a_notification_completion_input_is_visited(): data_converter = temporalio.converter.default() payload_converter = nexus_system._get_payload_converter( From dd4d80de28bb4ecd843182210d93333a6c373779 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 18 Oct 2026 01:13:12 -0700 Subject: [PATCH 3/3] Restored the OnProgress visitor test with the progress module. OnProgressRequest moved to progress_pb2 with the crate's Nexus API copy. Only the notifier ops left. --- tests/worker/test_visitor.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py index 14ec122689..feb1149100 100644 --- a/tests/worker/test_visitor.py +++ b/tests/worker/test_visitor.py @@ -6,6 +6,7 @@ import pytest from google.protobuf.duration_pb2 import Duration +import temporalio.api.notificationservice.v1.progress_pb2 as progress_pb2 import temporalio.api.notificationservice.v1.request_response_pb2 as notification_pb2 import temporalio.api.workflowservice.v1.request_response_pb2 as workflowservice_pb2 import temporalio.bridge.worker @@ -337,6 +338,32 @@ async def test_unknown_system_nexus_payload_raises_application_error(): assert not err.value.non_retryable +async def test_a_notification_progress_input_is_visited(): + # The server sends OnProgress to a Worker as a system payload. The source context + # inside it was encoded by the client, so the codec has to reach it. + data_converter = temporalio.converter.default() + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, + data_converter.failure_converter, + ) + request = progress_pb2.OnProgressRequest( + source_context=Payload(data=b"context"), + operation_token="token", + receiver_state=b"state", + ) + request.progress.position = "3" + request.progress.counter = 3 + payload = payload_converter.to_payload(request) + assert payload is not None + + await PayloadVisitor().visit(Visitor(), payload) + + decoded = payload_converter.from_payload(payload) + assert isinstance(decoded, progress_pb2.OnProgressRequest) + assert decoded.source_context.metadata["visited"] == b"True" + assert decoded.receiver_state == b"state" + + async def test_a_notification_completion_input_is_visited(): data_converter = temporalio.converter.default() payload_converter = nexus_system._get_payload_converter(