From 52cd64a5fe7ef01d01e218239fe3bb1965aa075f Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 23:13:59 +0000 Subject: [PATCH 1/2] Document helper-thread activity heartbeats and cancellation --- docs/reference/activity.md | 55 ++++++++++ examples/threaded_heartbeats.py | 56 ++++++++++ tests/test_threaded_heartbeat_example.py | 124 +++++++++++++++++++++++ 3 files changed, 235 insertions(+) create mode 100644 examples/threaded_heartbeats.py create mode 100644 tests/test_threaded_heartbeat_example.py diff --git a/docs/reference/activity.md b/docs/reference/activity.md index b7036f6..80460bc 100644 --- a/docs/reference/activity.md +++ b/docs/reference/activity.md @@ -7,4 +7,59 @@ default. They can call `context().heartbeat_sync(details)` and inspect See [activity execution and shutdown](worker.md#activity-execution) for thread capacity, cancellation and process-isolation limits. +## Heartbeats during one blocking call + +Capture `ctx = activity.context()` inside the activity before starting a helper +thread. A newly created thread does not automatically have the activity's +context. Calling `activity.context()` there can raise "called outside of an +activity execution". Pass the captured context instead. Its +`ctx.heartbeat_sync(details)` schedules the heartbeat on the owning worker loop +with the original task, attempt and worker identity. It can be called from a +helper thread, but must never run on that worker loop. Async activity code uses +`await ctx.heartbeat(details)`. + +The complete [threaded-heartbeat example](https://github.com/durable-workflow/sdk-python/blob/main/examples/threaded_heartbeats.py) +provides `call_with_heartbeats(work)`. Copy it into your application and use it +from an ordinary synchronous activity: + +```python +from durable_workflow import activity +from my_app.exports import export_service +from my_app.threaded_heartbeats import call_with_heartbeats + + +@activity.defn(name="wait_for_export") +def wait_for_export(export_id: str) -> dict: + return call_with_heartbeats( + lambda cancelled: export_service.wait( + export_id, timeout=120, cancel_event=cancelled, + ), + interval=5, + ) +``` + +`export_service` is your application's client, not a Durable Workflow API. It +must implement the finite timeout and observe `cancel_event` promptly, returning +or raising when it is set. Use the equivalent timeout and cancellation mechanism +from your own library. If it cannot cooperate, give it a finite I/O deadline; +the helper cannot force a blocked thread to exit. A worker process may need +separate isolation for a call that cannot be bounded. + +The helper records the original exception, sets the cancellation event, and +stops heartbeating when cancellation, lease-authority loss or a transport error +occurs. After the work unwinds, it joins the helper and re-raises that exception +before a result can be returned. In particular, do not swallow `ActivityCancelled` +or `ActivityLeaseAuthorityLost`, replace them with an application error, or +return a successful result after either. `ActivityCancelled` inherits from +`BaseException`, so `except Exception` alone does not catch it. Ordinary return +and work exceptions also stop and join the helper. + +Keep the client and its worker event loop alive until the helper has joined. +An in-flight heartbeat still needs its bounded transport request to finish. +Use finite client timeouts and bounded retry policy as well as a deadline on +the business operation. A helper reports liveness even if that operation hangs, +so its heartbeat is not proof of progress. Choose the interval using the +configured heartbeat timeout and leave room for request, scheduling and retry +delays. Five seconds is an example, not a guaranteed safe cadence. + ::: durable_workflow.activity diff --git a/examples/threaded_heartbeats.py b/examples/threaded_heartbeats.py new file mode 100644 index 0000000..abbc120 --- /dev/null +++ b/examples/threaded_heartbeats.py @@ -0,0 +1,56 @@ +"""Heartbeat a bounded, cooperatively cancellable call from a sync activity.""" +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from queue import SimpleQueue +from threading import Event, Thread +from typing import TypeVar + +from durable_workflow import activity +from durable_workflow.errors import ActivityCancelled + +T = TypeVar("T") + + +def call_with_heartbeats(work: Callable[[Event], T], *, interval: float = 5.0) -> T: + """Run work(cancelled), which must enforce its own finite deadline. + + Work must return or raise promptly when cancelled is set. This helper cannot + interrupt an arbitrary blocking call. Call it from a synchronous activity, + never from the worker event loop. + """ + try: + asyncio.get_running_loop() + except RuntimeError: + pass + else: + raise RuntimeError("call_with_heartbeats requires a synchronous activity thread") + if not 0 < interval < float("inf"): + raise ValueError("interval must be finite and positive") + + ctx = activity.context() + finished = Event() + cancelled = Event() + failures: SimpleQueue[BaseException] = SimpleQueue() + + def heartbeat() -> None: + while not finished.wait(interval): + try: + ctx.heartbeat_sync({"message": "Waiting for remote operation"}) + except (ActivityCancelled, Exception) as error: + # ActivityCancelled inherits BaseException, not Exception. + failures.put(error) + cancelled.set() + return + + helper = Thread(target=heartbeat, name="activity-heartbeat") + helper.start() + try: + return work(cancelled) + finally: + finished.set() + # Keep the captured context and client alive until the helper exits. + helper.join() + if not failures.empty(): + raise failures.get() diff --git a/tests/test_threaded_heartbeat_example.py b/tests/test_threaded_heartbeat_example.py new file mode 100644 index 0000000..e635f3f --- /dev/null +++ b/tests/test_threaded_heartbeat_example.py @@ -0,0 +1,124 @@ +"""Exercise the public example with real ActivityContext thread/loop behavior.""" +from __future__ import annotations + +import asyncio +import importlib.util +import unittest +from pathlib import Path +from threading import Event, Thread +from threading import enumerate as threads +from typing import Any + +from durable_workflow import activity +from durable_workflow.activity import ActivityContext, ActivityInfo, _set_context +from durable_workflow.errors import ActivityCancelled, ActivityLeaseAuthorityLost + +SPEC = importlib.util.spec_from_file_location( + "threaded_heartbeats_example", + Path(__file__).resolve().parents[1] / "examples" / "threaded_heartbeats.py", +) +assert SPEC is not None and SPEC.loader is not None +EXAMPLE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(EXAMPLE) + + +class HeartbeatClient: + def __init__(self, response: Any = None) -> None: + self.response = response if response is not None else {"can_continue": True} + self.calls: list[dict[str, Any]] = [] + self.observed = Event() + + async def heartbeat_activity_task(self, **request: Any) -> Any: + self.calls.append(request) + self.observed.set() + if isinstance(self.response, Exception): + raise self.response + return self.response + + +class ThreadedHeartbeatExampleTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.client = HeartbeatClient() + self.ctx = ActivityContext( + info=ActivityInfo( + "task-original", "export", "attempt-original", 2, "reports", "worker-original", + ), + client=self.client, # type: ignore[arg-type] + ) + _set_context(self.ctx) + + async def asyncTearDown(self) -> None: + _set_context(None) + self.assertFalse([thread for thread in threads() if thread.name == "activity-heartbeat"]) + + async def run_work(self, work: Any) -> Any: + return await asyncio.wait_for( + asyncio.to_thread(EXAMPLE.call_with_heartbeats, work, interval=0.01), + timeout=3, + ) + + async def test_new_thread_needs_the_captured_context(self) -> None: + errors: list[BaseException] = [] + + def look_up_context() -> None: + try: + activity.context() + except RuntimeError as error: + errors.append(error) + + helper = Thread(target=look_up_context) + helper.start() + helper.join() + self.assertEqual(len(errors), 1) + self.assertIn("outside of an activity", str(errors[0])) + + async def test_return_keeps_original_attempt_and_joins(self) -> None: + def work(cancelled: Event) -> dict[str, int]: + self.assertTrue(self.client.observed.wait(2)) + self.assertFalse(cancelled.is_set()) + return {"count": 7} + + self.assertEqual(await self.run_work(work), {"count": 7}) + self.assertTrue(self.client.calls) + for request in self.client.calls: + self.assertEqual(request["task_id"], "task-original") + self.assertEqual(request["activity_attempt_id"], "attempt-original") + self.assertEqual(request["lease_owner"], "worker-original") + + async def test_work_error_also_joins(self) -> None: + original = ValueError("business call failed") + + def work(cancelled: Event) -> None: + self.assertTrue(self.client.observed.wait(2)) + raise original + + with self.assertRaises(ValueError) as caught: + await self.run_work(work) + self.assertIs(caught.exception, original) + + def cancelled_work(self, cancelled: Event) -> str: + self.assertTrue(cancelled.wait(2)) + return "must not be returned" + + async def test_cancellation_stops_work_and_preserves_type(self) -> None: + self.client.response = {"cancel_requested": True} + with self.assertRaises(ActivityCancelled): + await self.run_work(self.cancelled_work) + + async def test_authority_loss_keeps_original_identity_and_reason(self) -> None: + self.client.response = {"can_continue": False, "reason": "activity_attempt_replaced"} + with self.assertRaises(ActivityLeaseAuthorityLost) as caught: + await self.run_work(self.cancelled_work) + self.assertEqual(caught.exception.task_id, "task-original") + self.assertEqual(caught.exception.reason, "activity_attempt_replaced") + + async def test_transport_error_is_not_silently_lost(self) -> None: + original = OSError("transport unavailable") + self.client.response = original + with self.assertRaises(OSError) as caught: + await self.run_work(self.cancelled_work) + self.assertIs(caught.exception, original) + + async def test_event_loop_call_refuses_before_work(self) -> None: + with self.assertRaisesRegex(RuntimeError, "synchronous activity thread"): + EXAMPLE.call_with_heartbeats(lambda cancelled: self.fail("must not run")) From f0e4cd23c23209760e9147d9d7afcd6ae8dd96e3 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 23:45:41 +0000 Subject: [PATCH 2/2] Separate heartbeat example import assertions --- tests/test_threaded_heartbeat_example.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_threaded_heartbeat_example.py b/tests/test_threaded_heartbeat_example.py index e635f3f..eb2d420 100644 --- a/tests/test_threaded_heartbeat_example.py +++ b/tests/test_threaded_heartbeat_example.py @@ -17,7 +17,8 @@ "threaded_heartbeats_example", Path(__file__).resolve().parents[1] / "examples" / "threaded_heartbeats.py", ) -assert SPEC is not None and SPEC.loader is not None +assert SPEC is not None +assert SPEC.loader is not None EXAMPLE = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(EXAMPLE)