Document helper-thread heartbeats with cancellation and teardown

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

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()

tests/test_threaded_heartbeat_example.py

@@ -0,0 +1,125 @@
+"""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
+assert 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"))