Run default synchronous activities in bounded remote threads

docs/reference/activity.md

@@ -1,3 +1,10 @@
 # Activity
 
+Synchronous activities execute in the worker's bounded remote thread pool by
+default. They can call `context().heartbeat_sync(details)` and inspect
+`context().is_cancelled`; a heartbeat reporting cancellation raises
+`ActivityCancelled`. Async activities use `await context().heartbeat(details)`.
+See [activity execution and shutdown](worker.md#activity-execution) for thread
+capacity, cancellation and process-isolation limits.
+
 ::: durable_workflow.activity

docs/reference/worker.md

@@ -4,6 +4,31 @@ See [Separate workflow and activity processes](../guides/separate-workers.md)
 for a runnable deployment example with distinct registrations, independent
 slots, remote activity retries and signal-driven shutdown.
 
+## Activity execution
+
+By default, ordinary synchronous `def` remote activities run in a dedicated
+thread pool bounded by `max_concurrent_activity_tasks`. Each thread receives
+its own `activity.context()`. A blocked synchronous call does not occupy the
+worker event loop, so other activity slots, workflow tasks and worker
+heartbeats can progress. Async `async def` activities run on that event loop
+and must use nonblocking I/O. Blocking work inside an async function still
+stalls the loop.
+
+From a synchronous activity, call
+`activity.context().heartbeat_sync(details)` to report progress with the
+original task, attempt and worker identity. It waits for the owner's event
+loop to send the request using the normal client transport timeouts and
+raises `ActivityCancelled` when cancellation is reported. Async activities use
+`await activity.context().heartbeat(details)`. Calling the synchronous method
+on the owner event loop raises an error instead of deadlocking it.
+
+Threads suit blocking I/O; they do not provide hard isolation or parallel
+Python CPU execution. Use bounded downstream timeouts and cooperative
+heartbeat checks. For CPU-heavy or untrusted callbacks, use separately
+isolated worker processes and their resource limits. The separately
+negotiated cooperative-cancellation capability retains its supervised process
+execution; selecting a default thread does not enable that protocol.
+
 ## Lifecycle and shutdown
 
 `Worker.run()` and `Worker.run_until()` register through the worker protocol
@@ -13,6 +38,15 @@ removing that worker-plane registration. Work still running at the deadline is
 cancelled. Calling `stop()` more than once is safe and sends at most one
 deregistration request. No deregistration is sent if registration failed.
 
+Cancelling an await cannot stop a running synchronous activity thread. Its
+physical slot remains occupied until the callback returns, and a late return
+does not submit a result after its await was cancelled. If the callback has
+not stopped by the shared shutdown deadline, shutdown raises `RuntimeError`
+and retains the registration. Python can keep the process alive while that
+thread remains running. Configure callback I/O timeouts and a supervisor
+termination policy; a timeout is not proof that an external side effect was
+cancelled or that the worker stopped gracefully.
+
 An external `stop()` also interrupts `run_until()`; the pending `run_until()`
 call ends with `asyncio.CancelledError` after its accepted workflow or activity
 work has drained or been cancelled. If an async poller, the query-poller thread,

src/durable_workflow/activity.py

@@ -1,6 +1,7 @@
 """Activity decorator, registry, and execution context."""
 from __future__ import annotations
 
+import asyncio
 import contextvars
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
@@ -44,6 +45,10 @@ def __init__(
         self._client = client
         self._heartbeat_callback = heartbeat_callback
         self._cancel_requested = False
+        try:
+            self._event_loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop()
+        except RuntimeError:
+            self._event_loop = None
 
     @property
     def info(self) -> ActivityInfo:
@@ -55,6 +60,29 @@ def is_cancelled(self) -> bool:
         """``True`` once the server has signalled that the owning workflow cancelled this activity."""
         return self._cancel_requested
 
+    def heartbeat_sync(self, details: dict[str, Any] | None = None) -> None:
+        """Heartbeat from a synchronous activity thread and observe cancellation.
+
+        The request runs on the worker's event loop using the original attempt
+        and client. Async handlers must use ``await heartbeat()`` instead.
+        """
+        loop = self._event_loop
+        if loop is None or not loop.is_running():
+            raise RuntimeError("heartbeat_sync requires a running worker event loop")
+        try:
+            caller_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            caller_loop = None
+        if caller_loop is loop:
+            raise RuntimeError("use await activity.context().heartbeat() on the worker event loop")
+        heartbeat = self.heartbeat(details)
+        try:
+            pending = asyncio.run_coroutine_threadsafe(heartbeat, loop)
+        except RuntimeError:
+            heartbeat.close()
+            raise
+        pending.result()
+
     async def heartbeat(self, details: dict[str, Any] | None = None) -> None:
         """Report liveness to the server and check for a cancellation request.
 

src/durable_workflow/worker.py

@@ -3070,7 +3070,9 @@ async def _run_activity_task(self, task: dict[str, Any]) -> str:
             if self._cooperative_cancellation_supported:
                 result = await self._execute_cooperative_remote_callable(task, fn, tuple(args), act_ctx.info)
             else:
-                result = await self._execute_activity_callable(task, activity_type, tuple(args), fn)
+                result = await self._execute_activity_callable(
+                    task, activity_type, tuple(args), fn, run_sync_in_thread=True, remote=True,
+                )
         except _RemoteActivityExecutionAborted as error:
             log.warning("remote activity %s claim abandoned: %s", task_id, error)
             return "claim_aborted"
@@ -4511,6 +4513,17 @@ async def _shutdown(self) -> None:
                 )
             await asyncio.gather(*in_flight, return_exceptions=True)
 
+        # Cancelling the asyncio await cannot stop a running Python thread.
+        # Keep registration and capacity until the actual callback has returned.
+        while any(not future.done() for future in self._remote_activity_threads):
+            remaining = self._remaining_shutdown_time(deadline)
+            if remaining <= 0:
+                raise RuntimeError(
+                    "worker shutdown has unconfirmed activity thread stop(s); "
+                    "the worker registration remains active"
+                )
+            await asyncio.sleep(min(0.01, remaining))
+
         if self._local_activity_executor is not None:
             self._local_activity_executor.shutdown(wait=False, cancel_futures=True)
         if self._remote_activity_executor is not None:

tests/test_sync_activities.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import threading
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from durable_workflow import activity, serializer, workflow
+from durable_workflow.client import Client
+from durable_workflow.worker import Worker
+
+
+def activity_task(number: int = 1) -> dict[str, Any]:
+    return {
+        "task_id": f"sync-task-{number}",
+        "activity_attempt_id": f"sync-attempt-{number}",
+        "activity_type": "sync-probe",
+        "arguments": serializer.encode([], codec="avro"),
+        "payload_codec": "avro",
+    }
+
+
+async def wait_for_thread(event: threading.Event) -> None:
+    async def observe() -> None:
+        while not event.is_set():
+            await asyncio.sleep(0.001)
+
+    await asyncio.wait_for(observe(), timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_default_sync_handlers_run_concurrently_without_stalling_heartbeat() -> None:
+    release = threading.Event()
+    started: list[str] = []
+
+    @activity.defn(name="sync-probe")
+    def blocking() -> str:
+        started.append(threading.current_thread().name)
+        assert release.wait(1), "event loop did not release the activity"
+        return activity.context().info.task_id
+
+    @workflow.defn(name="sync-side-workflow")
+    class SideWorkflow:
+        def run(self, ctx: Any) -> str:
+            return "workflow-progress"
+
+    client = AsyncMock(spec=Client)
+    worker = Worker(client, task_queue="sync-queue", workflows=[SideWorkflow], activities=[blocking],
+                    max_concurrent_activity_tasks=2, heartbeat_interval=0.01)
+    heartbeat = asyncio.create_task(worker._heartbeat_loop())
+    tasks = [asyncio.create_task(worker._run_activity_task(activity_task(n))) for n in (1, 2)]
+    try:
+        await asyncio.sleep(0.05)
+        assert len(started) == 2
+        assert all(name.startswith("dw-remote-activity") for name in started)
+        assert not any(task.done() for task in tasks)
+        assert client.heartbeat_worker.await_count >= 1
+        assert worker._current_task_slots()["activity_available"] == 0
+        commands = await worker._run_workflow_task({
+            "task_id": "side-workflow-task", "workflow_type": "sync-side-workflow",
+            "workflow_task_attempt": 1, "history_events": [],
+            "arguments": serializer.encode([], codec="avro"), "payload_codec": "avro",
+        })
+        assert commands is not None
+        assert commands[0]["type"] == "complete_workflow"
+        client.complete_workflow_task.assert_awaited_once()
+        assert not any(task.done() for task in tasks)
+    finally:
+        release.set()
+        worker._stop.set()
+        await heartbeat
+        results = await asyncio.gather(*tasks)
+        await worker._shutdown()
+    assert results == ["completed", "completed"]
+    assert {call.kwargs["result"] for call in client.complete_activity_task.await_args_list} == {
+        "sync-task-1", "sync-task-2",
+    }
+
+
+@pytest.mark.asyncio
+async def test_sync_activity_heartbeat_retains_owner_and_observes_cancellation() -> None:
+    contexts: list[activity.ActivityContext] = []
+
+    @activity.defn(name="sync-probe")
+    def blocking() -> None:
+        ctx = activity.context()
+        contexts.append(ctx)
+        ctx.heartbeat_sync({"processed": 7})
+        pytest.fail("cancelled handler must not return a result")
+
+    client = AsyncMock(spec=Client)
+    client.heartbeat_activity_task.return_value = {"cancel_requested": True}
+    worker = Worker(client, task_queue="sync-queue", worker_id="sync-owner", activities=[blocking])
+    try:
+        assert await worker._run_activity_task(activity_task()) == "cancelled"
+        client.heartbeat_activity_task.assert_awaited_once_with(
+            task_id="sync-task-1", activity_attempt_id="sync-attempt-1",
+            lease_owner="sync-owner", details={"processed": 7},
+        )
+        assert contexts[0].is_cancelled
+        client.complete_activity_task.assert_not_awaited()
+        assert client.fail_activity_task.await_args.kwargs["failure_type"] == "ActivityCancelled"
+    finally:
+        await worker._shutdown()
+
+
+@pytest.mark.asyncio
+async def test_async_activity_remains_on_loop_and_sync_heartbeat_cannot_deadlock_it() -> None:
+    owner = threading.current_thread()
+
+    @activity.defn(name="sync-probe")
+    async def nonblocking() -> str:
+        assert threading.current_thread() is owner
+        ctx = activity.context()
+        with pytest.raises(RuntimeError, match="await.*heartbeat"):
+            ctx.heartbeat_sync()
+        await ctx.heartbeat({"async": True})
+        return "async-result"
+
+    client = AsyncMock(spec=Client)
+    client.heartbeat_activity_task.return_value = {"cancel_requested": False}
+    worker = Worker(client, task_queue="sync-queue", activities=[nonblocking])
+    assert await worker._run_activity_task(activity_task()) == "completed"
+    assert client.complete_activity_task.await_args.kwargs["result"] == "async-result"
+    assert worker._remote_activity_executor is None
+
+
+@pytest.mark.asyncio
+async def test_cancelled_awaiter_keeps_physical_thread_capacity_until_return() -> None:
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    @activity.defn(name="sync-probe")
+    def blocking() -> None:
+        started.set()
+        try:
+            release.wait(2)
+        finally:
+            finished.set()
+
+    client = AsyncMock(spec=Client)
+    worker = Worker(client, task_queue="sync-queue", activities=[blocking], max_concurrent_activity_tasks=1)
+    running = asyncio.create_task(worker._run_activity_task(activity_task()))
+    try:
+        await wait_for_thread(started)
+        running.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await running
+        assert not finished.is_set()
+        assert worker._current_task_slots()["activity_available"] == 0
+        client.complete_activity_task.assert_not_awaited()
+    finally:
+        release.set()
+        await wait_for_thread(finished)
+        await asyncio.sleep(0)
+        await worker._shutdown()
+    assert worker._current_task_slots()["activity_available"] == 1
+
+
+@pytest.mark.asyncio
+async def test_shutdown_retains_registration_if_activity_thread_has_not_stopped() -> None:
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    @activity.defn(name="sync-probe")
+    def blocking() -> None:
+        started.set()
+        try:
+            release.wait(2)
+        finally:
+            finished.set()
+
+    client = AsyncMock(spec=Client)
+    worker = Worker(client, task_queue="sync-queue", activities=[blocking], shutdown_timeout=0.01)
+    worker._registered = True
+    running = worker._track(worker._run_activity_task(activity_task()))
+    try:
+        await wait_for_thread(started)
+        with pytest.raises(RuntimeError, match="activity thread.*registration remains active"):
+            await worker._shutdown()
+        assert worker._registered
+        client.deregister_worker_registration.assert_not_awaited()
+        assert not finished.is_set()
+        client.complete_activity_task.assert_not_awaited()
+    finally:
+        release.set()
+        await wait_for_thread(finished)
+        with contextlib.suppress(asyncio.CancelledError):
+            await running
+        await asyncio.sleep(0)
+        await worker._shutdown()
+    client.deregister_worker_registration.assert_awaited_once_with(worker.worker_id)