Integration test for Standalone Activities delayed-start (#1520)

* Integration test for Standalone Activities delayed-start

* Bump server version, and de-flake delay assertion.

* CI debugging

* Remove debugging and warm server download in CI

* Fix `test_workflow_caller_cancellation_types_when_cancel_handler_fails`
flaking

* Remove server download warming

Author: eamsden

.github/workflows/ci.yml

@@ -46,7 +46,6 @@ jobs:
           components: "clippy"
       - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
         with:
-          cache-bin: false
           workspaces: temporalio/bridge -> target
       - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:
@@ -112,7 +111,6 @@ jobs:
       - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
       - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
         with:
-          cache-bin: false
           workspaces: temporalio/bridge -> target
       - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:
@@ -149,7 +147,6 @@ jobs:
           components: "clippy"
       - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
         with:
-          cache-bin: false
           workspaces: temporalio/bridge -> target
       - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:
@@ -188,7 +185,6 @@ jobs:
       - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
       - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
         with:
-          cache-bin: false
           workspaces: temporalio/bridge -> target
       - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:

tests/__init__.py

@@ -1 +1 @@
-DEV_SERVER_DOWNLOAD_VERSION = "v1.7.0"
+DEV_SERVER_DOWNLOAD_VERSION = "v1.7.1-standalone-nexus-operations"

tests/conftest.py

@@ -123,6 +123,8 @@ async def env(env_type: str) -> AsyncGenerator[WorkflowEnvironment, None]:
                 "--dynamic-config-value",
                 "activity.enableStandalone=true",
                 "--dynamic-config-value",
+                "activity.startDelayEnabled=true",
+                "--dynamic-config-value",
                 "history.enableChasm=true",
                 "--dynamic-config-value",
                 "history.enableTransitionHistory=true",

tests/nexus/test_workflow_caller_cancellation_types.py

@@ -20,6 +20,7 @@
 from temporalio.common import WorkflowIDConflictPolicy
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
+from tests.helpers import assert_eventually
 from tests.helpers.nexus import make_nexus_endpoint_name
 
 
@@ -40,25 +41,41 @@ class TestContext:
 class HandlerWorkflow:
     def __init__(self):
         self.caller_op_future_resolved = asyncio.Event()
+        self.cancel_requested = False
+        self.release_cancellation = asyncio.Event()
 
     @workflow.run
     async def run(self) -> None:
         try:
             await asyncio.Future()
         except asyncio.CancelledError:
+            self.cancel_requested = True
             if test_context.cancellation_type in [
                 workflow.NexusOperationCancellationType.TRY_CANCEL,
                 workflow.NexusOperationCancellationType.WAIT_REQUESTED,
             ]:
                 # We want to prove that the caller op future can be resolved before the operation
                 # (i.e. its backing workflow) is cancelled.
                 await self.caller_op_future_resolved.wait()
+            elif (
+                test_context.cancellation_type
+                == workflow.NexusOperationCancellationType.WAIT_COMPLETED
+            ):
+                await self.release_cancellation.wait()
             raise
 
     @workflow.signal
     def set_caller_op_future_resolved(self) -> None:
         self.caller_op_future_resolved.set()
 
+    @workflow.signal
+    def set_release_cancellation(self) -> None:
+        self.release_cancellation.set()
+
+    @workflow.query
+    def has_cancel_requested(self) -> bool:
+        return self.cancel_requested
+
 
 @nexusrpc.service
 class Service:
@@ -151,6 +168,10 @@ async def get_operation_token(self) -> str:
     async def wait_caller_op_future_resolved(self) -> None:
         await self.caller_op_future_resolved
 
+    @workflow.query
+    def has_caller_op_future_resolved(self) -> bool:
+        return self.caller_op_future_resolved.done()
+
     @workflow.run
     async def run(self, input: Input) -> CancellationResult:
         op_handle = await (
@@ -408,6 +429,17 @@ async def check_behavior_for_wait_cancellation_completed(
     Check that a cancellation request is sent and the caller workflow nexus operation future is
     unblocked after the operation is canceled.
     """
+
+    async def assert_handler_cancel_requested() -> None:
+        assert await handler_wf.query(HandlerWorkflow.has_cancel_requested)
+
+    await assert_eventually(assert_handler_cancel_requested)
+
+    handler_status = (await handler_wf.describe()).status
+    assert handler_status == WorkflowExecutionStatus.RUNNING
+    assert not await caller_wf.query(CallerWorkflow.has_caller_op_future_resolved)
+
+    await handler_wf.signal(HandlerWorkflow.set_release_cancellation)
     try:
         await handler_wf.result()
     except WorkflowFailureError as err:
@@ -418,8 +450,13 @@ async def check_behavior_for_wait_cancellation_completed(
     handler_status = (await handler_wf.describe()).status
     assert handler_status == WorkflowExecutionStatus.CANCELED
 
+    async def assert_caller_op_future_resolved() -> None:
+        assert await caller_wf.query(CallerWorkflow.has_caller_op_future_resolved)
+
+    await assert_eventually(assert_caller_op_future_resolved)
+
     await caller_wf.signal(CallerWorkflow.release)
-    result = await caller_wf.result()
+    await caller_wf.result()
 
     await assert_event_subsequence(
         caller_wf,
@@ -430,14 +467,6 @@ async def check_behavior_for_wait_cancellation_completed(
             EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED,
         ],
     )
-    handler_wf_canceled_event = await get_event_time(
-        handler_wf,
-        EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED,
-    )
-    assert handler_wf_canceled_event <= result.caller_op_future_resolved, (
-        "expected caller op future resolved after handler workflow canceled, but got "
-        f"{result.caller_op_future_resolved} before {handler_wf_canceled_event}"
-    )
 
 
 async def has_event(wf_handle: WorkflowHandle, event_type: EventType.ValueType):

tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py

@@ -23,6 +23,7 @@
 from temporalio.common import WorkflowIDConflictPolicy
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
+from tests.helpers import assert_eventually
 from tests.helpers.nexus import make_nexus_endpoint_name
 from tests.nexus.test_workflow_caller_cancellation_types import (
     assert_event_subsequence,
@@ -48,6 +49,7 @@ class HandlerWorkflow:
     def __init__(self):
         self.cancel_handler_released = asyncio.Event()
         self.caller_op_future_resolved = asyncio.Event()
+        self.release_completion = asyncio.Event()
 
     @workflow.run
     async def run(self) -> None:
@@ -61,6 +63,11 @@ async def run(self) -> None:
             # For WAIT_REQUESTED, we want to prove that the future can be unblocked before the
             # handler workflow completes.
             await self.caller_op_future_resolved.wait()
+        elif (
+            test_context.cancellation_type
+            == workflow.NexusOperationCancellationType.WAIT_COMPLETED
+        ):
+            await self.release_completion.wait()
 
     @workflow.signal
     def set_cancel_handler_released(self) -> None:
@@ -70,6 +77,14 @@ def set_cancel_handler_released(self) -> None:
     def set_caller_op_future_resolved(self) -> None:
         self.caller_op_future_resolved.set()
 
+    @workflow.signal
+    def set_release_completion(self) -> None:
+        self.release_completion.set()
+
+    @workflow.query
+    def has_cancel_handler_released(self) -> bool:
+        return self.cancel_handler_released.is_set()
+
 
 @nexusrpc.service
 class Service:
@@ -153,6 +168,10 @@ async def get_operation_token(self) -> str:
         assert self.operation_token
         return self.operation_token
 
+    @workflow.query
+    def has_caller_op_future_resolved(self) -> bool:
+        return self.caller_op_future_resolved.done()
+
     @workflow.run
     async def run(self, input: Input) -> CancellationResult:
         op_handle = await (
@@ -370,7 +389,23 @@ async def check_behavior_for_wait_cancellation_completed(
     caller_wf: WorkflowHandle[Any, CancellationResult],
     handler_wf: WorkflowHandle,
 ) -> None:
+    async def assert_cancel_handler_released() -> None:
+        assert await handler_wf.query(HandlerWorkflow.has_cancel_handler_released)
+
+    await assert_eventually(assert_cancel_handler_released)
+
+    handler_status = (await handler_wf.describe()).status
+    assert handler_status == WorkflowExecutionStatus.RUNNING
+    assert not await caller_wf.query(CallerWorkflow.has_caller_op_future_resolved)
+
+    await handler_wf.signal(HandlerWorkflow.set_release_completion)
     await handler_wf.result()
+
+    async def assert_caller_op_future_resolved() -> None:
+        assert await caller_wf.query(CallerWorkflow.has_caller_op_future_resolved)
+
+    await assert_eventually(assert_caller_op_future_resolved)
+
     await caller_wf.signal(CallerWorkflow.release)
     result = await caller_wf.result()
     assert not result.error_type
@@ -386,8 +421,3 @@ async def check_behavior_for_wait_cancellation_completed(
             EventType.EVENT_TYPE_NEXUS_OPERATION_COMPLETED,
         ],
     )
-    handler_wf_completed = await get_event_time(
-        handler_wf,
-        EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED,
-    )
-    assert handler_wf_completed <= result.caller_op_future_resolved

tests/test_activity.py

@@ -461,6 +461,38 @@ async def test_get_result(client: Client, env: WorkflowEnvironment):
         assert await result_via_execute_activity == 2
 
 
+async def test_start_activity_start_delay(client: Client, env: WorkflowEnvironment):
+    if env.supports_time_skipping:
+        pytest.skip(
+            "Java test server: https://github.com/temporalio/sdk-java/issues/2741"
+        )
+
+    activity_id = str(uuid.uuid4())
+    task_queue = str(uuid.uuid4())
+    start_delay = timedelta(seconds=2)
+
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        activities=[increment],
+    ):
+        activity_handle = await client.start_activity(
+            increment,
+            args=(1,),
+            id=activity_id,
+            task_queue=task_queue,
+            start_to_close_timeout=timedelta(seconds=5),
+            start_delay=start_delay,
+        )
+
+        assert await activity_handle.result() == 2
+        desc = await activity_handle.describe()
+        assert desc.last_started_time is not None
+        assert (
+            desc.last_started_time - desc.scheduled_time
+        ).total_seconds() >= start_delay.total_seconds() - 0.5
+
+
 async def test_get_activity_handle(client: Client, env: WorkflowEnvironment):
     if env.supports_time_skipping:
         pytest.skip(