Add test and error case

Author: tconley1428

temporalio/worker/_activity.py

@@ -334,6 +334,24 @@ async def _handle_start_activity_task(
                         ),
                         completion.result.failed.failure,
                     )
+                elif (
+                    isinstance(
+                        err,
+                        (asyncio.CancelledError, temporalio.exceptions.CancelledError),
+                    )
+                    and running_activity.cancellation_details.details
+                    and running_activity.cancellation_details.details.reset
+                ):
+                    temporalio.activity.logger.warning(
+                        "Completing as failure due to unhandled cancel error produced by activity reset",
+                    )
+                    await self._data_converter.encode_failure(
+                        temporalio.exceptions.ApplicationError(
+                            type="ActivityReset",
+                            message="Unhandled activity cancel error produced by activity reset",
+                        ),
+                        completion.result.failed.failure,
+                    )
                 elif (
                     isinstance(
                         err,

temporalio/worker/_workflow_instance.py

@@ -2791,6 +2791,7 @@ def _apply_schedule_command(
                 command.user_metadata.summary.CopyFrom(
                     self._instance._payload_converter.to_payload(self._input.summary)
                 )
+            print("Activity summary: ", command.user_metadata.summary)
             if self._input.priority:
                 command.schedule_activity.priority.CopyFrom(
                     self._input.priority._to_proto()

tests/worker/test_activity.py

@@ -9,10 +9,12 @@
 import threading
 import time
 import uuid
+from concurrent.futures import ThreadPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from contextvars import ContextVar
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
+from time import sleep
 from typing import Any, Callable, List, NoReturn, Optional, Sequence, Type
 
 import temporalio.api.workflowservice.v1
@@ -1500,7 +1502,6 @@ async def wait_cancel() -> str:
             ),
             id=activity.info().activity_id,
         )
-        activity.logger.info(f"Sending reset request: {req}")
         await client.workflow_service.reset_activity(req)
         try:
             while True:
@@ -1511,9 +1512,42 @@ async def wait_cancel() -> str:
             assert details is not None
             return "Got cancelled error, reset? " + str(details.reset)
 
+    @activity.defn
+    def sync_wait_cancel() -> str:
+        req = temporalio.api.workflowservice.v1.ResetActivityRequest(
+            namespace=client.namespace,
+            execution=temporalio.api.common.v1.WorkflowExecution(
+                workflow_id=activity.info().workflow_id,
+                run_id=activity.info().workflow_run_id,
+            ),
+            id=activity.info().activity_id,
+        )
+        asyncio.run(client.workflow_service.reset_activity(req))
+        try:
+            while True:
+                sleep(0.3)
+                activity.heartbeat()
+        except temporalio.exceptions.CancelledError:
+            details = activity.cancellation_details()
+            assert details is not None
+            return "Got cancelled error, reset? " + str(details.reset)
+        except Exception as e:
+            return str(type(e)) + str(e)
+
     result = await _execute_workflow_with_activity(
         client,
         worker,
         wait_cancel,
     )
     assert result.result == "Got cancelled error, reset? True"
+
+    config = WorkerConfig(
+        activity_executor=ThreadPoolExecutor(max_workers=1),
+    )
+    result = await _execute_workflow_with_activity(
+        client,
+        worker,
+        sync_wait_cancel,
+        worker_config=config,
+    )
+    assert result.result == "Got cancelled error, reset? True"

tests/worker/test_workflow.py

@@ -814,7 +814,10 @@ class SimpleActivityWorkflow:
     @workflow.run
     async def run(self, name: str) -> str:
         return await workflow.execute_activity(
-            say_hello, name, schedule_to_close_timeout=timedelta(seconds=5)
+            say_hello,
+            name,
+            schedule_to_close_timeout=timedelta(seconds=5),
+            summary="Do a thing",
         )
 
 
@@ -8327,3 +8330,28 @@ async def test_workflow_headers_with_codec(
             assert headers["foo"].data == b"bar"
         else:
             assert headers["foo"].data != b"bar"
+
+
+async def test_summary_with_codec(client: Client, env: WorkflowEnvironment):
+    if env.supports_time_skipping:
+        pytest.skip("Time skipping server doesn't persist headers.")
+
+    # Make client with this codec and run a couple of existing tests
+    config = client.config()
+    config["data_converter"] = DataConverter(payload_codec=SimpleCodec())
+    client = Client(**config)
+
+    async with new_worker(
+        client,
+        SimpleActivityWorkflow,
+        SignalAndQueryWorkflow,
+        activities=[say_hello],
+    ) as worker:
+        workflow_handle = await client.start_workflow(
+            SimpleActivityWorkflow.run,
+            "Temporal",
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        assert await workflow_handle.result() == "Hello, Temporal!"
+    assert False