WAIT_REQUESTED

Author: dandavison

tests/nexus/test_workflow_caller_cancellation_types.py

@@ -175,6 +175,15 @@ async def run(self, input: Input) -> CancellationResult:
         #              closed as cancelled) before sending an activation job to Python resolving the
         #              nexus operation as cancelled
         op_handle.cancel()
+        if (
+            test_context.cancellation_type
+            == workflow.NexusOperationCancellationType.WAIT_REQUESTED
+        ):
+            # For WAIT_REQUESTED, we need core to receive the NexusOperationCancelRequestCompleted
+            # event. That event should trigger a workflow task, but does not currently due to
+            # https://github.com/temporalio/temporal/issues/8175. Force a new WFT, allowing time for
+            # the event hopefully to arrive.
+            await workflow.sleep(0.1, summary="Force new WFT")
         try:
             await op_handle
         except exceptions.NexusOperationError:
@@ -184,17 +193,22 @@ async def run(self, input: Input) -> CancellationResult:
             assert op_handle.operation_token
             # Block the thread for a bit to avoid flakiness in asserting that the future is
             # unblocked before the WFT completion is sent in the TryCancel case.
-            time.sleep(1e-2)
-
-            # Notify handler workflow that caller op future has been resolved
-            await workflow.get_external_workflow_handle_for(
-                HandlerWorkflow.run,
-                workflow_id=(
-                    nexus.WorkflowHandle[None]
-                    .from_token(self.operation_token)
-                    .workflow_id
-                ),
-            ).signal(HandlerWorkflow.set_caller_op_future_resolved)
+            time.sleep(0.01)
+
+            if (
+                input.cancellation_type
+                == workflow.NexusOperationCancellationType.WAIT_REQUESTED
+            ):
+                # We want to prove that the future can be unblocked before the handler workflow is
+                # cancelled. Send a signal, so that handler workflow can wait for it.
+                await workflow.get_external_workflow_handle_for(
+                    HandlerWorkflow.run,
+                    workflow_id=(
+                        nexus.WorkflowHandle[None]
+                        .from_token(self.operation_token)
+                        .workflow_id
+                    ),
+                ).signal(HandlerWorkflow.set_caller_op_future_resolved)
 
             await workflow.wait_condition(lambda: self.released)
             return CancellationResult(