Clarity on client thread safety and sync activity cancel (#189)

Fixes #147

Author: cretz

README.md

@@ -783,8 +783,9 @@ Synchronous activities, i.e. functions that do not have `async def`, can be used
 activities.
 
 Cancellation for synchronous activities is done in the background and the activity must choose to listen for it and
-react appropriately. An activity must heartbeat to receive cancellation and there are other ways to be notified about
-cancellation (see "Activity Context" and "Heartbeating and Cancellation" later).
+react appropriately. If after cancellation is obtained an unwrapped `temporalio.exceptions.CancelledError` is raised,
+the activity will be marked cancelled. An activity must heartbeat to receive cancellation and there are other ways to be
+notified about cancellation (see "Activity Context" and "Heartbeating and Cancellation" later).
 
 Note, all calls from an activity to functions in the `temporalio.activity` package are powered by
 [contextvars](https://docs.python.org/3/library/contextvars.html). Therefore, new threads starting _inside_ of

temporalio/client.py

@@ -65,6 +65,12 @@ class Client:
     :py:attr:`service` property provides access to a raw gRPC client. To create
     another client, like for a different namespace, :py:func:`Client` may be
     directly instantiated with a :py:attr:`service` of another.
+
+    Clients are not thread-safe and should only be used in the event loop they
+    are first connected in. If a client needs to be used from another thread
+    than where it was created, make sure the event loop where it was created is
+    captured, and then call :py:func:`asyncio.run_coroutine_threadsafe` with the
+    client call and that event loop.
     """
 
     @staticmethod

temporalio/worker/_activity.py

@@ -401,14 +401,18 @@ async def _run_activity(
         except (
             Exception,
             asyncio.CancelledError,
+            temporalio.exceptions.CancelledError,
             temporalio.activity._CompleteAsyncError,
         ) as err:
             try:
                 if isinstance(err, temporalio.activity._CompleteAsyncError):
                     temporalio.activity.logger.debug("Completing asynchronously")
                     completion.result.will_complete_async.SetInParent()
                 elif (
-                    isinstance(err, asyncio.CancelledError)
+                    isinstance(
+                        err,
+                        (asyncio.CancelledError, temporalio.exceptions.CancelledError),
+                    )
                     and running_activity.cancelled_due_to_heartbeat_error
                 ):
                     err = running_activity.cancelled_due_to_heartbeat_error
@@ -419,7 +423,10 @@ async def _run_activity(
                         err, completion.result.failed.failure
                     )
                 elif (
-                    isinstance(err, asyncio.CancelledError)
+                    isinstance(
+                        err,
+                        (asyncio.CancelledError, temporalio.exceptions.CancelledError),
+                    )
                     and running_activity.cancelled_by_request
                 ):
                     temporalio.activity.logger.debug("Completing as cancelled")

tests/worker/test_activity.py

@@ -283,6 +283,31 @@ def wait_cancel() -> str:
     assert result.result == "Cancelled"
 
 
+async def test_sync_activity_thread_cancel_uncaught(
+    client: Client, worker: ExternalWorker
+):
+    @activity.defn
+    def wait_cancel() -> str:
+        while not activity.is_cancelled():
+            time.sleep(1)
+            activity.heartbeat()
+        raise CancelledError("Cancelled")
+
+    with pytest.raises(WorkflowFailureError) as err:
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            await _execute_workflow_with_activity(
+                client,
+                worker,
+                wait_cancel,
+                cancel_after_ms=100,
+                wait_for_cancellation=True,
+                heartbeat_timeout_ms=3000,
+                worker_config={"activity_executor": executor},
+            )
+    assert isinstance(err.value.cause, ActivityError)
+    assert isinstance(err.value.cause.cause, CancelledError)
+
+
 @activity.defn
 def picklable_activity_wait_cancel() -> str:
     while not activity.is_cancelled():
@@ -305,6 +330,32 @@ async def test_sync_activity_process_cancel(client: Client, worker: ExternalWork
     assert result.result == "Cancelled"
 
 
+@activity.defn
+def picklable_activity_raise_cancel() -> str:
+    while not activity.is_cancelled():
+        time.sleep(1)
+        activity.heartbeat()
+    raise CancelledError("Cancelled")
+
+
+async def test_sync_activity_process_cancel_uncaught(
+    client: Client, worker: ExternalWorker
+):
+    with pytest.raises(WorkflowFailureError) as err:
+        with concurrent.futures.ProcessPoolExecutor() as executor:
+            result = await _execute_workflow_with_activity(
+                client,
+                worker,
+                picklable_activity_raise_cancel,
+                cancel_after_ms=100,
+                wait_for_cancellation=True,
+                heartbeat_timeout_ms=3000,
+                worker_config={"activity_executor": executor},
+            )
+    assert isinstance(err.value.cause, ActivityError)
+    assert isinstance(err.value.cause.cause, CancelledError)
+
+
 async def test_activity_does_not_exist(client: Client, worker: ExternalWorker):
     @activity.defn
     async def say_hello(name: str) -> str: