diff --git a/docs/reference/activity.md b/docs/reference/activity.md index f4a66fe..b7036f6 100644 --- a/docs/reference/activity.md +++ b/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 diff --git a/docs/reference/worker.md b/docs/reference/worker.md index 5578f9c..029bfc2 100644 --- a/docs/reference/worker.md +++ b/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, diff --git a/src/durable_workflow/activity.py b/src/durable_workflow/activity.py index 42d4705..4982085 100644 --- a/src/durable_workflow/activity.py +++ b/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. diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 7880f21..577d89e 100644 --- a/src/durable_workflow/worker.py +++ b/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: diff --git a/tests/test_sync_activities.py b/tests/test_sync_activities.py new file mode 100644 index 0000000..58a7e9b --- /dev/null +++ b/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)