Introduce NexusOperationError to wrap nexus op errors

Author: dandavison

temporalio/converter.py

@@ -901,13 +901,56 @@ def _error_to_failure(
             failure.child_workflow_execution_failure_info.retry_state = (
                 temporalio.api.enums.v1.RetryState.ValueType(error.retry_state or 0)
             )
+        elif isinstance(error, temporalio.exceptions.NexusOperationError):
+            failure.nexus_operation_execution_failure_info.SetInParent()
+            failure.nexus_operation_execution_failure_info.operation_token = (
+                error.operation_token
+            )
 
     def from_failure(
         self,
         failure: temporalio.api.failure.v1.Failure,
         payload_converter: PayloadConverter,
     ) -> BaseException:
         """See base class."""
+
+        # message Failure {
+        #     string message = 1;
+        #     // The source this Failure originated in, e.g. TypeScriptSDK / JavaSDK
+        #     // In some SDKs this is used to rehydrate the stack trace into an exception object.
+        #     string source = 2;
+        #     string stack_trace = 3;
+        #     // Alternative way to supply `message` and `stack_trace` and possibly other attributes, used for encryption of
+        #     // errors originating in user code which might contain sensitive information.
+        #     // The `encoded_attributes` Payload could represent any serializable object, e.g. JSON object or a `Failure` proto
+        #     // message.
+        #     //
+        #     // SDK authors:
+        #     // - The SDK should provide a default `encodeFailureAttributes` and `decodeFailureAttributes` implementation that:
+        #     //   - Uses a JSON object to represent `{ message, stack_trace }`.
+        #     //   - Overwrites the original message with "Encoded failure" to indicate that more information could be extracted.
+        #     //   - Overwrites the original stack_trace with an empty string.
+        #     //   - The resulting JSON object is converted to Payload using the default PayloadConverter and should be processed
+        #     //     by the user-provided PayloadCodec
+        #     //
+        #     // - If there's demand, we could allow overriding the default SDK implementation to encode other opaque Failure attributes.
+        #     // (-- api-linter: core::0203::optional=disabled --)
+        #     temporal.api.common.v1.Payload encoded_attributes = 20;
+        #     Failure cause = 4;
+        #     oneof failure_info {
+        #         ApplicationFailureInfo application_failure_info = 5;
+        #         TimeoutFailureInfo timeout_failure_info = 6;
+        #         CanceledFailureInfo canceled_failure_info = 7;
+        #         TerminatedFailureInfo terminated_failure_info = 8;
+        #         ServerFailureInfo server_failure_info = 9;
+        #         ResetWorkflowFailureInfo reset_workflow_failure_info = 10;
+        #         ActivityFailureInfo activity_failure_info = 11;
+        #         ChildWorkflowExecutionFailureInfo child_workflow_execution_failure_info = 12;
+        #         NexusOperationFailureInfo nexus_operation_execution_failure_info = 13;
+        #         NexusHandlerFailureInfo nexus_handler_failure_info = 14;
+        #     }
+        # }
+
         # If encoded attributes are present and have the fields we expect,
         # extract them
         if failure.HasField("encoded_attributes"):
@@ -978,6 +1021,15 @@ def from_failure(
                 else None,
             )
         elif failure.HasField("child_workflow_execution_failure_info"):
+            # message ChildWorkflowExecutionFailureInfo {
+            #     string namespace = 1;
+            #     temporal.api.common.v1.WorkflowExecution workflow_execution = 2;
+            #     temporal.api.common.v1.WorkflowType workflow_type = 3;
+            #     int64 initiated_event_id = 4;
+            #     int64 started_event_id = 5;
+            #     temporal.api.enums.v1.RetryState retry_state = 6;
+            # }
+
             child_info = failure.child_workflow_execution_failure_info
             err = temporalio.exceptions.ChildWorkflowError(
                 failure.message or "Child workflow error",
@@ -993,6 +1045,45 @@ def from_failure(
                 if child_info.retry_state
                 else None,
             )
+        elif failure.HasField("nexus_handler_failure_info"):
+            # message NexusHandlerFailureInfo {
+            #     // The Nexus error type as defined in the spec:
+            #     // https://github.com/nexus-rpc/api/blob/main/SPEC.md#predefined-handler-errors.
+            #     string type = 1;
+            #     // Retry behavior, defaults to the retry behavior of the error type as defined in the spec.
+            #     temporal.api.enums.v1.NexusHandlerErrorRetryBehavior retry_behavior = 2;
+            # }
+            # TODO(dan): core never sends this currently?
+            raise NotImplementedError(
+                "TODO: Nexus handler failure info not implemented"
+            )
+        elif failure.HasField("nexus_operation_execution_failure_info"):
+            # message NexusOperationFailureInfo {
+            #     // The NexusOperationScheduled event ID.
+            #     int64 scheduled_event_id = 1;
+            #     // Endpoint name.
+            #     string endpoint = 2;
+            #     // Service name.
+            #     string service = 3;
+            #     // Operation name.
+            #     string operation = 4;
+            #     // Operation ID - may be empty if the operation completed synchronously.
+            #     //
+            #     // Deprecated: Renamed to operation_token.
+            #     string operation_id = 5;
+            #     // Operation token - may be empty if the operation completed synchronously.
+            #     string operation_token = 6;
+            # }
+            # TODO(dan)
+            nexus_op_failure_info = failure.nexus_operation_execution_failure_info
+            err = temporalio.exceptions.NexusOperationError(
+                failure.message or "Nexus operation error",
+                scheduled_event_id=nexus_op_failure_info.scheduled_event_id,
+                endpoint=nexus_op_failure_info.endpoint,
+                service=nexus_op_failure_info.service,
+                operation=nexus_op_failure_info.operation,
+                operation_token=nexus_op_failure_info.operation_token,
+            )
         else:
             err = temporalio.exceptions.FailureError(failure.message or "Failure error")
         err._failure = failure

temporalio/exceptions.py

@@ -284,6 +284,15 @@ def retry_state(self) -> Optional[RetryState]:
 class ChildWorkflowError(FailureError):
     """Error raised on child workflow failure."""
 
+    # message ChildWorkflowExecutionFailureInfo {
+    #     string namespace = 1;
+    #     temporal.api.common.v1.WorkflowExecution workflow_execution = 2;
+    #     temporal.api.common.v1.WorkflowType workflow_type = 3;
+    #     int64 initiated_event_id = 4;
+    #     int64 started_event_id = 5;
+    #     temporal.api.enums.v1.RetryState retry_state = 6;
+    # }
+
     def __init__(
         self,
         message: str,
@@ -342,6 +351,70 @@ def retry_state(self) -> Optional[RetryState]:
         return self._retry_state
 
 
+class NexusOperationError(FailureError):
+    """Error raised on Nexus operation failure."""
+
+    # message NexusOperationFailureInfo {
+    #     // The NexusOperationScheduled event ID.
+    #     int64 scheduled_event_id = 1;
+    #     // Endpoint name.
+    #     string endpoint = 2;
+    #     // Service name.
+    #     string service = 3;
+    #     // Operation name.
+    #     string operation = 4;
+    #     // Operation ID - may be empty if the operation completed synchronously.
+    #     //
+    #     // Deprecated: Renamed to operation_token.
+    #     string operation_id = 5;
+    #     // Operation token - may be empty if the operation completed synchronously.
+    #     string operation_token = 6;
+    # }
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        scheduled_event_id: int,
+        endpoint: str,
+        service: str,
+        operation: str,
+        operation_token: str,
+    ):
+        """Initialize a Nexus operation error."""
+        super().__init__(message)
+        self._scheduled_event_id = scheduled_event_id
+        self._endpoint = endpoint
+        self._service = service
+        self._operation = operation
+        self._operation_token = operation_token
+
+    @property
+    def scheduled_event_id(self) -> int:
+        """The NexusOperationScheduled event ID for the failed operation."""
+        return self._scheduled_event_id
+
+    @property
+    def endpoint(self) -> str:
+        """The endpoint name for the failed operation."""
+        return self._endpoint
+
+    @property
+    def service(self) -> str:
+        """The service name for the failed operation."""
+        return self._service
+
+    @property
+    def operation(self) -> str:
+        """The name of the failed operation."""
+        return self._operation
+
+    @property
+    def operation_token(self) -> str:
+        """The operation token returned by the failed operation."""
+        return self._operation_token
+
+
 def is_cancelled_exception(exception: BaseException) -> bool:
     """Check whether the given exception is considered a cancellation exception
     according to Temporal.

temporalio/worker/_workflow_instance.py

@@ -796,7 +796,22 @@ def _apply_resolve_child_workflow_execution(
         self,
         job: temporalio.bridge.proto.workflow_activation.ResolveChildWorkflowExecution,
     ) -> None:
+        # message ResolveChildWorkflowExecution {
+        #     // Sequence number as provided by lang in the corresponding StartChildWorkflowExecution command
+        #     uint32 seq = 1;
+        #     child_workflow.ChildWorkflowResult result = 2;
+        # }
         # No matter the result, we know we want to pop
+
+        # // Used by core to resolve child workflow executions.
+        # message ChildWorkflowResult {
+        #     oneof status {
+        #         Success completed = 1;
+        #         Failure failed = 2;
+        #         Cancellation cancelled = 3;
+        #     }
+        # }
+
         handle = self._pending_child_workflows.pop(job.seq, None)
         if not handle:
             raise RuntimeError(
@@ -813,6 +828,11 @@ def _apply_resolve_child_workflow_execution(
                 ret = ret_vals[0]
             handle._resolve_success(ret)
         elif job.result.HasField("failed"):
+            # // Used in ChildWorkflowResult to report non successful outcomes such as
+            # // application failures, timeouts, terminations, and cancellations.
+            # message Failure {
+            #     temporal.api.failure.v1.Failure failure = 1;
+            # }
             handle._resolve_failure(
                 self._failure_converter.from_failure(
                     job.result.failed.failure, self._payload_converter
@@ -926,20 +946,18 @@ def _apply_resolve_nexus_operation(
         self,
         job: temporalio.bridge.proto.workflow_activation.ResolveNexusOperation,
     ) -> None:
+        # message ResolveNexusOperation {
+        #   // Sequence number as provided by lang in the corresponding ScheduleNexusOperation command
+        #   uint32 seq = 1;
+        #   nexus.NexusOperationResult result = 2;
+        # }
         handle = self._pending_nexus_operations.get(job.seq)
         if not handle:
             raise RuntimeError(
                 f"Failed to find nexus operation handle for job sequence number {job.seq}"
             )
 
         result = job.result
-        # Handle the four oneof variants of NexusOperationResult
-
-        # message ResolveNexusOperation {
-        #   // Sequence number as provided by lang in the corresponding ScheduleNexusOperation command
-        #   uint32 seq = 1;
-        #   nexus.NexusOperationResult result = 2;
-        # }
         # message NexusOperationResult {
         #   oneof status {
         #       temporal.api.common.v1.Payload completed = 1;
@@ -948,8 +966,13 @@ def _apply_resolve_nexus_operation(
         #       temporal.api.failure.v1.Failure timed_out = 4;
         #   }
         # }
+        # Handle the four oneof variants of NexusOperationResult
 
         if result.HasField("completed"):
+            # message Payload {
+            #     map<string,bytes> metadata = 1;
+            #     bytes data = 2;
+            # }
             [output] = self._convert_payloads(
                 [result.completed],
                 [handle._input.output_type] if handle._input.output_type else None,

tests/worker/test_nexus.py

@@ -19,7 +19,15 @@
 import temporalio.nexus
 import temporalio.nexus.handler
 from temporalio import workflow
-from temporalio.client import Client, WorkflowExecutionStatus, WorkflowHandle
+from temporalio.client import (
+    Client,
+    WithStartWorkflowOperation,
+    WorkflowExecutionStatus,
+    WorkflowFailureError,
+    WorkflowHandle,
+)
+from temporalio.common import WorkflowIDConflictPolicy
+from temporalio.exceptions import CancelledError, NexusOperationError
 from temporalio.worker import UnsandboxedWorkflowRunner, Worker
 
 
@@ -292,9 +300,9 @@ async def test_async_response(
         await create_nexus_endpoint(task_queue, client)
         operation_workflow_id = "default-workflow-id"
 
-        # Start the caller workflow.
+        # Start the caller workflow and wait until it confirms the Nexus operation has started.
         block_forever_waiting_for_cancellation = request_cancel
-        caller_wf_handle = await client.start_workflow(
+        start_op = WithStartWorkflowOperation(
             MyCallerWorkflow.run,
             args=[
                 MyInput(
@@ -312,12 +320,15 @@ async def test_async_response(
             ],
             id=str(uuid.uuid4()),
             task_queue=task_queue,
+            id_conflict_policy=WorkflowIDConflictPolicy.FAIL,
         )
 
-        # Wait until the caller wf knows that the Nexus operation has started.
-        await caller_wf_handle.execute_update(
-            MyCallerWorkflow.wait_nexus_operation_started
+        await client.execute_update_with_start_workflow(
+            MyCallerWorkflow.wait_nexus_operation_started,
+            start_workflow_operation=start_op,
         )
+        caller_wf_handle = await start_op.workflow_handle()
+
         # check that the operation-backing workflow now exists, and that (a) the handler
         # workflow accepted the link to the calling Nexus event, and that (b) the caller
         # workflow NexusOperationStarted event received in return a link to the
@@ -339,11 +350,25 @@ async def test_async_response(
         if request_cancel:
             # The operation response was asynchronous and so request_cancel is honored. See
             # explanation below.
-            # TODO: what error should be raised here?
-            with pytest.raises(BaseException) as ei:
+            with pytest.raises(WorkflowFailureError) as ei:
                 await caller_wf_handle.result()
             e = ei.value
-            print(f"🌈 workflow failed: {e.__class__.__name__}({e})")
+            assert isinstance(e, WorkflowFailureError)
+            assert isinstance(e.__cause__, NexusOperationError)
+            assert isinstance(e.__cause__.__cause__, CancelledError)
+            # ID of first command after update accepted
+            assert e.__cause__.scheduled_event_id == 6
+            assert e.__cause__.endpoint == make_nexus_endpoint_name(task_queue)
+            assert e.__cause__.service == "MyService"
+            assert (
+                e.__cause__.operation == "my_async_operation"
+                if use_shorthand_defined_operation
+                else "my_sync_or_async_operation"
+            )
+            assert temporalio.nexus.handler.StartWorkflowOperationResult._decode_token(
+                e.__cause__.operation_token
+            ) == (handler_wf_handle.id, handler_wf_info.run_id)
+            # Check that the handler workflow was canceled
             handler_wf_info = await handler_wf_handle.describe()
             assert handler_wf_info.status == WorkflowExecutionStatus.CANCELED
         else:
@@ -427,6 +452,22 @@ async def assert_handler_workflow_has_link_to_caller_workflow(
 
 # TODO(dan): test exceptions in Nexus worker are raised
 
+
+async def print_history(handle: WorkflowHandle):
+    print("\n\n")
+    history = await handle.fetch_history()
+    for event in history.events:
+        try:
+            event_type_name = temporalio.api.enums.v1.EventType.Name(
+                event.event_type
+            ).replace("EVENT_TYPE_", "")
+        except ValueError:
+            # Handle unknown event types
+            event_type_name = f"Unknown({event.event_type})"
+        print(f"{event.event_id}. {event_type_name}")
+    print("\n\n")
+
+
 # When request_cancel is True, the NexusOperationHandle in the workflow evolves
 # through the following states:
 #                           start_fut             result_fut            handle_task w/ fut_waiter                        (task._must_cancel)