Respond to upstream core proto / api change: fail on sync start

Author: dandavison

temporalio/worker/_workflow_instance.py

@@ -857,23 +857,21 @@ def _apply_resolve_nexus_operation_start(
                 f"Failed to find nexus operation handle for job sequence number {job.seq}"
             )
         if job.HasField("operation_token"):
-            # The Nexus operation started asynchronously. A `ResolveNexusOperation` job
+            # The nexus operation started asynchronously. A `ResolveNexusOperation` job
             # will follow in a future activation.
             handle._resolve_start_success(job.operation_token)
         elif job.HasField("started_sync"):
-            # The Nexus operation 'started' in the sense that it's already resolved. A
+            # The nexus operation 'started' in the sense that it's already resolved. A
             # `ResolveNexusOperation` job will be in the same activation.
             handle._resolve_start_success(None)
-        elif job.HasField("cancelled_before_start"):
-            # From proto docs: the operation was cancelled before it was ever
-            # sent to server (same WFT). Note that core will still send a
-            # `ResolveNexusOperation` job in the same activation, so there does
-            # not need to be an exceptional case for this in lang.
-            #
-            # We do not resolve start here because it will be resolved in the by
-            # handle._resolve_failure when handling the follow-up
-            # ResolveNexusOperation from core.
-            pass
+        elif job.HasField("failed"):
+            # The nexus operation start failed; no ResolveNexusOperation will follow.
+            self._pending_nexus_operations.pop(job.seq, None)
+            handle._resolve_failure(
+                self._failure_converter.from_failure(
+                    job.failed, self._payload_converter
+                )
+            )
         else:
             raise ValueError(f"Unknown Nexus operation start status: {job}")
 

tests/nexus/test_workflow_caller.py

@@ -260,7 +260,7 @@ def __init__(
             }[input.op_input.caller_reference],
             endpoint=make_nexus_endpoint_name(task_queue),
         )
-        self._nexus_operation_started = False
+        self._nexus_operation_start_resolved = False
         self._proceed = False
 
     @workflow.run
@@ -271,12 +271,14 @@ async def run(
         task_queue: str,
     ) -> CallerWfOutput:
         op_input = input.op_input
-        op_handle = await self.nexus_client.start_operation(
-            self._get_operation(op_input),  # type: ignore[arg-type] # test uses non-public operation types
-            op_input,
-            headers=op_input.headers,
-        )
-        self._nexus_operation_started = True
+        try:
+            op_handle = await self.nexus_client.start_operation(
+                self._get_operation(op_input),  # type: ignore[arg-type] # test uses non-public operation types
+                op_input,
+                headers=op_input.headers,
+            )
+        finally:
+            self._nexus_operation_start_resolved = True
         if not input.op_input.response_type.exception_in_operation_start:
             if isinstance(input.op_input.response_type, SyncResponse):
                 assert (
@@ -295,8 +297,8 @@ async def run(
         return CallerWfOutput(op_output=OpOutput(value=op_output.value))
 
     @workflow.update
-    async def wait_nexus_operation_started(self) -> None:
-        await workflow.wait_condition(lambda: self._nexus_operation_started)
+    async def wait_nexus_operation_start_resolved(self) -> None:
+        await workflow.wait_condition(lambda: self._nexus_operation_start_resolved)
 
     @staticmethod
     def _get_operation(
@@ -720,7 +722,7 @@ async def _start_wf_and_nexus_op(
     )
 
     await client.execute_update_with_start_workflow(
-        CallerWorkflow.wait_nexus_operation_started,
+        CallerWorkflow.wait_nexus_operation_start_resolved,
         start_workflow_operation=start_op,
     )
     caller_wf_handle = await start_op.workflow_handle()