Merge pull request #69 from durable-workflow/feat/python-local-activity-execution

Execute Python local activities with durable replay

Author: rmcdaniel

docker-compose.test.yml

@@ -29,6 +29,7 @@ services:
       CACHE_STORE: redis
       WORKFLOW_SERVER_AUTH_DRIVER: token
       WORKFLOW_SERVER_AUTH_TOKEN: "test-token"
+      DW_WORKFLOW_TASK_TIMEOUT: 10
     depends_on:
       mysql:
         condition: service_healthy

docs/sdk-reference.md

@@ -185,6 +185,72 @@ receipt = yield ctx.start_child_workflow(
 )
 ```
 
+## Local activities
+
+Use a local activity for short work that should run in the workflow worker
+process without a separate activity task. Register it as an activity on the
+same `Worker`, then yield `ctx.local_activity(...)` from the workflow:
+
+```python
+import asyncio
+import os
+from uuid import uuid4
+
+from durable_workflow import Client, Worker, activity, workflow
+
+
+@activity.defn(name="orders.normalize")
+def normalize(order_id: str) -> str:
+    return order_id.strip().upper()
+
+
+@workflow.defn(name="orders.prepare")
+class PrepareOrder:
+    def run(self, ctx, order_id):
+        normalized = yield ctx.local_activity(
+            "orders.normalize",
+            [order_id],
+            retry_policy={"max_attempts": 2, "backoff_seconds": [1]},
+            start_to_close_timeout=5,
+        )
+        return {"order_id": normalized}
+
+
+async def main():
+    async with Client(
+        os.getenv("DURABLE_WORKFLOW_RUNTIME_URL", "http://127.0.0.1:8080"),
+        token=os.getenv("DURABLE_WORKFLOW_TOKEN", "dev-token-123"),
+        namespace="default",
+    ) as client:
+        worker = Worker(
+            client,
+            task_queue="orders",
+            workflows=[PrepareOrder],
+            activities=[normalize],
+        )
+        handle = await client.start_workflow(
+            workflow_type="orders.prepare",
+            workflow_id=f"order-{uuid4().hex}",
+            task_queue="orders",
+            input=["  a-123  "],
+        )
+        await worker.run_until(workflow_id=handle.workflow_id, timeout=30.0)
+        print(await handle.result(timeout=10.0))
+
+
+asyncio.run(main())
+```
+
+Use the Server setup from the [quickstart](index.md#first-workflow), or replace
+the URL and credentials with those of a provisioned Cloud namespace. The worker
+records the local result with its workflow task. Once Server accepts that
+record, replay reads the result instead of running the handler again. If the
+worker loses its lease or crashes before the record is committed, the handler
+may run again on a replacement worker. Make side effects idempotent and keep
+local attempts short; use `ctx.schedule_activity(...)` for separately queued
+work. Local activities are yielded one at a time; they are not members of a
+parallel workflow group.
+
 ## Deterministic parallel groups
 
 Yield a list to schedule one durable parallel barrier. Lists can nest and mix

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.1.2"
+version = "2.2.0"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.1.2"
-registry-version = "2.1.2"
+product-train = "2.2.0"
+registry-version = "2.2.0"
 supported-server-versions = "2.4.0"
 worker-protocol-version = "1.19"
 control-plane-version = "2"

src/durable_workflow/activity.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import contextvars
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
@@ -38,9 +38,11 @@ def __init__(
         *,
         info: ActivityInfo,
         client: Client,
+        heartbeat_callback: Callable[[dict[str, Any] | None], Awaitable[None]] | None = None,
     ) -> None:
         self._info = info
         self._client = client
+        self._heartbeat_callback = heartbeat_callback
         self._cancel_requested = False
 
     @property
@@ -65,6 +67,13 @@ async def heartbeat(self, details: dict[str, Any] | None = None) -> None:
         owning workflow has requested cancellation, so the activity can exit
         cleanly at its next natural break point.
         """
+        if self._heartbeat_callback is not None:
+            try:
+                await self._heartbeat_callback(details)
+            except ActivityCancelled:
+                self._cancel_requested = True
+                raise
+            return
         resp = await self._client.heartbeat_activity_task(
             task_id=self._info.task_id,
             activity_attempt_id=self._info.activity_attempt_id,

src/durable_workflow/client.py

@@ -66,9 +66,9 @@
 CONTROL_PLANE_VERSION = "2"
 PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST: dict[str, dict[str, str | bool]] = {
     "local_activities": {
-        "supported": False,
+        "supported": True,
         "minimum_protocol_version": "1.18",
-        "reason": "python_worker_does_not_execute_record_local_activity",
+        "implementation": "record_local_activity",
     },
     "worker_sessions": {
         "supported": False,

src/durable_workflow/external_storage.py

@@ -12,7 +12,12 @@
 from typing import Any, Protocol
 from urllib.parse import quote, unquote, urlparse
 
-from .errors import ExternalPayloadIntegrityMismatch, ExternalPayloadUnsupported
+from .errors import (
+    ExternalPayloadError,
+    ExternalPayloadIntegrityMismatch,
+    ExternalPayloadUnavailable,
+    ExternalPayloadUnsupported,
+)
 
 EXTERNAL_PAYLOAD_REFERENCE_SCHEMA = "durable-workflow.v2.external-payload-reference.v1"
 RUNTIME_EXTERNAL_PAYLOAD_REFERENCE_SCHEMA = (
@@ -556,7 +561,12 @@ def store_external_payload(
     if expires_at is not None:
         _validate_rfc3339(expires_at)
     sha256 = hashlib.sha256(data).hexdigest()
-    uri = driver.put(data, sha256=sha256, codec=codec)
+    try:
+        uri = driver.put(data, sha256=sha256, codec=codec)
+    except ExternalPayloadError:
+        raise
+    except Exception as exc:
+        raise ExternalPayloadUnavailable("external payload storage upload failed") from exc
     return ExternalPayloadReference(
         uri=uri,
         sha256=sha256,