WIP: complete Nexus task as failure

Author: dandavison

temporalio/converter.py

@@ -901,6 +901,7 @@ def _error_to_failure(
             failure.child_workflow_execution_failure_info.retry_state = (
                 temporalio.api.enums.v1.RetryState.ValueType(error.retry_state or 0)
             )
+        # TODO(dan): test coverage for this
         elif isinstance(error, temporalio.exceptions.NexusOperationError):
             failure.nexus_operation_execution_failure_info.SetInParent()
             failure.nexus_operation_execution_failure_info.operation_token = (
@@ -1073,6 +1074,7 @@ def from_failure(
             #     string operation_token = 6;
             # }
             # TODO(dan)
+            # This is covered by cancellation tests
             nexus_op_failure_info = failure.nexus_operation_execution_failure_info
             err = temporalio.exceptions.NexusOperationError(
                 failure.message or "Nexus operation error",

temporalio/worker/_nexus.py

@@ -245,50 +245,58 @@ async def _run_nexus_operation(
         #     bytes details = 3;
         # }
 
-        completion = temporalio.bridge.proto.nexus.NexusTaskCompletion(
-            task_token=task_token,
-        )
-        try:
-            result = await start(input, options)
-        except BaseException:
-            # await self._data_converter.encode_failure(err, completion.error.failure)
-            # TODO(dan): mirror appropriate aspects of _run_activity error handling
-            raise NotImplementedError(
-                "TODO: Nexus operation error handling not implemented"
-            )
-
-        try:
-            # Send task completion
-            if isinstance(result, nexusrpc.handler.StartOperationAsyncResult):
-                print(f"🟢 Nexus operation started with async response {result}")
-                op_resp = temporalio.api.nexus.v1.StartOperationResponse(
-                    async_success=temporalio.api.nexus.v1.StartOperationResponse.Async(
-                        operation_token=result.token,
-                        links=[
-                            temporalio.api.nexus.v1.Link(url=l.url, type=l.type)
-                            for l in result.links
-                        ],
-                    )
+        async def run() -> temporalio.bridge.proto.nexus.NexusTaskCompletion:
+            try:
+                result = await start(input, options)
+            except BaseException as err:
+                # TODO(dan): should encode_failure be called here?? (It accepts the
+                # api.Failure proto struct, not the Nexus one.)
+                # await self._data_converter.encode_failure(err, completion.error.failure)
+                return temporalio.bridge.proto.nexus.NexusTaskCompletion(
+                    task_token=task_token,
+                    error=temporalio.api.nexus.v1.HandlerError(
+                        # TODO(dan): what should error_type be? See `Predefined Handler Errors` in spec
+                        error_type="INTERNAL",
+                        failure=temporalio.api.nexus.v1.Failure(
+                            message=str(err),
+                            metadata={},
+                            details=b"",
+                        ),
+                    ),
                 )
             else:
-                # TODO(dan): are we going to use StartOperationSyncResult from nexusrpc?
-                # (contains links and headers in addition to result) IIRC Go does something
-                # like that.
-                [payload] = await self._data_converter.encode([result])
-                op_resp = temporalio.api.nexus.v1.StartOperationResponse(
-                    sync_success=temporalio.api.nexus.v1.StartOperationResponse.Sync(
-                        payload=payload
+                if isinstance(result, nexusrpc.handler.StartOperationAsyncResult):
+                    print(f"🟢 Nexus operation started with async response {result}")
+                    op_resp = temporalio.api.nexus.v1.StartOperationResponse(
+                        async_success=temporalio.api.nexus.v1.StartOperationResponse.Async(
+                            operation_token=result.token,
+                            links=[
+                                temporalio.api.nexus.v1.Link(url=l.url, type=l.type)
+                                for l in result.links
+                            ],
+                        )
+                    )
+                else:
+                    # TODO(dan): are we going to use StartOperationSyncResult from nexusrpc?
+                    # (contains links and headers in addition to result) IIRC Go does something
+                    # like that.
+                    [payload] = await self._data_converter.encode([result])
+                    op_resp = temporalio.api.nexus.v1.StartOperationResponse(
+                        sync_success=temporalio.api.nexus.v1.StartOperationResponse.Sync(
+                            payload=payload
+                        )
                     )
+                return temporalio.bridge.proto.nexus.NexusTaskCompletion(
+                    task_token=task_token,
+                    completed=temporalio.api.nexus.v1.Response(start_operation=op_resp),
                 )
-            completion.completed.CopyFrom(
-                temporalio.api.nexus.v1.Response(start_operation=op_resp)
-            )
+
+        try:
+            completion = await run()
             await self._bridge_worker().complete_nexus_task(completion)
             del self._running_operations[task_token]
         except Exception:
-            temporalio.nexus.logger.exception(
-                "Failed to send Nexus operation completion"
-            )
+            temporalio.nexus.logger.exception("Failed completing Nexus operation")
 
     async def _handle_cancel_operation(
         self, request: temporalio.api.nexus.v1.CancelOperationRequest, task_token: bytes