Support activity retry options

Author: durable-workflow-ops

README.md

@@ -54,6 +54,29 @@ For a fuller deployable example, see
 [`examples/order_processing`](examples/order_processing), which runs a
 multi-activity order workflow against a local server with Docker Compose.
 
+## Activity retries and timeouts
+
+Configure per-call activity retries and deadlines from workflow code:
+
+```python
+from durable_workflow import ActivityRetryPolicy
+
+result = yield ctx.schedule_activity(
+    "charge-card",
+    [order],
+    retry_policy=ActivityRetryPolicy(
+        max_attempts=4,
+        initial_interval_seconds=1,
+        backoff_coefficient=2,
+        maximum_interval_seconds=30,
+        non_retryable_error_types=["ValidationError"],
+    ),
+    start_to_close_timeout=120,
+    schedule_to_close_timeout=300,
+    heartbeat_timeout=15,
+)
+```
+
 ## Features
 
 - **Async-first**: Built on `httpx` and `asyncio`

docs/index.md

@@ -18,10 +18,10 @@ pip install 'durable-workflow[prometheus]'
 
 - **[Client](reference/client.md)** — start workflows, signal, query, update, wait for results, manage schedules.
 - **[Worker](reference/worker.md)** — poll the server for workflow and activity tasks, dispatch to registered handlers.
-- **[Workflow](reference/workflow.md)** — workflow-side primitives: `ContinueAsNew`, `StartChildWorkflow`, and the workflow decorator.
+- **[Workflow](reference/workflow.md)** — workflow-side primitives: `ActivityRetryPolicy`, `ContinueAsNew`, `StartChildWorkflow`, and the workflow decorator.
 - **[Activity](reference/activity.md)** — activity decorator and execution context.
 - **[Errors](reference/errors.md)** — typed exceptions raised by the client and worker.
-- **[Retry policy](reference/retry_policy.md)** — retry configuration for activities and workflows.
+- **[Retry policy](reference/retry_policy.md)** — HTTP transport retry configuration for the client.
 - **[Metrics](reference/metrics.md)** — pluggable recorders, including a Prometheus adapter.
 - **[Serializer](reference/serializer.md)** — payload encoding and decoding helpers.
 - **[Sync helpers](reference/sync.md)** — blocking wrappers around the async client for scripts and tests.

src/durable_workflow/__init__.py

@@ -46,14 +46,16 @@
     NoopMetrics,
     PrometheusMetrics,
 )
+from .retry_policy import RetryPolicy, TransportRetryPolicy
 from .worker import Worker
-from .workflow import ContinueAsNew, StartChildWorkflow
+from .workflow import ActivityRetryPolicy, ContinueAsNew, StartChildWorkflow
 
 __all__ = [
     "__version__",
     "ActivityCancelled",
     "ActivityContext",
     "ActivityInfo",
+    "ActivityRetryPolicy",
     "ChildWorkflowFailed",
     "Client",
     "ContinueAsNew",
@@ -83,7 +85,9 @@
     "NoopMetrics",
     "QueryFailed",
     "PrometheusMetrics",
+    "RetryPolicy",
     "ServerError",
+    "TransportRetryPolicy",
     "Unauthorized",
     "UpdateRejected",
     "WorkflowAlreadyStarted",

src/durable_workflow/client.py

@@ -35,7 +35,7 @@
     _raise_for_status,
 )
 from .metrics import CLIENT_REQUEST_DURATION_SECONDS, CLIENT_REQUESTS, NOOP_METRICS, MetricsRecorder
-from .retry_policy import RetryPolicy
+from .retry_policy import TransportRetryPolicy
 
 PROTOCOL_VERSION = "1.0"
 CONTROL_PLANE_VERSION = "2"
@@ -330,15 +330,15 @@ def __init__(
         worker_token: str | None = None,
         namespace: str = "default",
         timeout: float = 60.0,
-        retry_policy: RetryPolicy | None = None,
+        retry_policy: TransportRetryPolicy | None = None,
         metrics: MetricsRecorder | None = None,
     ) -> None:
         self.base_url = base_url.rstrip("/")
         self.token = token
         self.control_token = control_token
         self.worker_token = worker_token
         self.namespace = namespace
-        self.retry_policy = retry_policy or RetryPolicy()
+        self.retry_policy = retry_policy or TransportRetryPolicy()
         self.metrics = metrics or NOOP_METRICS
         self._http = httpx.AsyncClient(base_url=self.base_url, timeout=timeout)
 

src/durable_workflow/retry_policy.py

@@ -2,11 +2,11 @@
 
 .. warning::
 
-   This :class:`RetryPolicy` covers **only client-side HTTP retries** for
+   :class:`TransportRetryPolicy` covers **only client-side HTTP retries** for
    transient transport errors (connection failures, timeouts, 5xx responses,
    429 rate-limiting). It is **not** the activity retry policy. Activity-level
-   retry and timeout configuration is tracked in
-   https://github.com/zorporation/durable-workflow/issues/392 and will land on
+   retry and timeout configuration lives on
+   :class:`durable_workflow.workflow.ActivityRetryPolicy` and is passed to
    ``ctx.schedule_activity(..., retry_policy=...)``.
 """
 
@@ -24,9 +24,9 @@
 
 
 @dataclass
-class RetryPolicy:
+class TransportRetryPolicy:
     """
-    Retry policy for transient server errors.
+    Retry policy for transient HTTP transport errors.
 
     Retries requests that fail with transient errors (connection errors,
     timeouts, 5xx server errors, 429 rate limit). Does not retry client
@@ -95,3 +95,9 @@ async def execute(self, fn: Callable[[], Awaitable[T]]) -> T:
         if last_exc:
             raise last_exc
         raise RuntimeError("retry loop exhausted with no exception")
+
+
+# Backward-compatible alias for earlier 0.x releases. Prefer
+# TransportRetryPolicy in new code so it is not confused with workflow-level
+# activity retry policy.
+RetryPolicy = TransportRetryPolicy

src/durable_workflow/sync.py

@@ -18,7 +18,7 @@
     WorkflowList,
 )
 from .metrics import MetricsRecorder
-from .retry_policy import RetryPolicy
+from .retry_policy import TransportRetryPolicy
 
 
 def _run(coro: Any) -> Any:
@@ -152,7 +152,7 @@ def __init__(
         worker_token: str | None = None,
         namespace: str = "default",
         timeout: float = 60.0,
-        retry_policy: RetryPolicy | None = None,
+        retry_policy: TransportRetryPolicy | None = None,
         metrics: MetricsRecorder | None = None,
     ) -> None:
         self._async = AsyncClient(

src/durable_workflow/workflow.py

@@ -19,9 +19,10 @@
 import contextlib
 import hashlib
 import logging
+import math
 import random
 import uuid
-from collections.abc import Callable, Iterable
+from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any
@@ -49,23 +50,97 @@ def registry() -> dict[str, type]:
 
 
 # ── Commands yielded from a workflow ──────────────────────────────────
+@dataclass
+class ActivityRetryPolicy:
+    """Retry policy applied to one scheduled activity call.
+
+    The policy is snapped onto the durable activity execution when the
+    workflow task completes, so later code deploys do not change the retry
+    budget for an already-scheduled activity.
+    """
+
+    max_attempts: int = 3
+    initial_interval_seconds: float = 1.0
+    backoff_coefficient: float = 2.0
+    maximum_interval_seconds: float | None = None
+    non_retryable_error_types: list[str] = field(default_factory=list)
+    backoff_seconds: list[int] | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return the server command shape for this activity retry policy."""
+        if self.max_attempts < 1:
+            raise ValueError("max_attempts must be >= 1")
+        if self.initial_interval_seconds < 0:
+            raise ValueError("initial_interval_seconds must be >= 0")
+        if self.backoff_coefficient < 1:
+            raise ValueError("backoff_coefficient must be >= 1")
+        if self.maximum_interval_seconds is not None and self.maximum_interval_seconds < 0:
+            raise ValueError("maximum_interval_seconds must be >= 0")
+
+        return {
+            "max_attempts": self.max_attempts,
+            "backoff_seconds": self._backoff_seconds(),
+            "non_retryable_error_types": [
+                value.strip()
+                for value in self.non_retryable_error_types
+                if isinstance(value, str) and value.strip()
+            ],
+        }
+
+    def _backoff_seconds(self) -> list[int]:
+        if self.backoff_seconds is not None:
+            return [max(0, int(seconds)) for seconds in self.backoff_seconds]
+
+        seconds: list[int] = []
+        current = self.initial_interval_seconds
+        maximum = self.maximum_interval_seconds
+        for _ in range(max(0, self.max_attempts - 1)):
+            value = current if maximum is None else min(current, maximum)
+            seconds.append(max(0, int(math.ceil(value))))
+            current *= self.backoff_coefficient
+        return seconds
+
+
+ActivityRetryPolicyInput = ActivityRetryPolicy | Mapping[str, Any]
+
+
 @dataclass
 class ScheduleActivity:
     """Command requesting an activity task."""
 
     activity_type: str
     arguments: list[Any]
     queue: str | None = None
+    retry_policy: ActivityRetryPolicyInput | None = None
+    start_to_close_timeout: int | None = None
+    schedule_to_start_timeout: int | None = None
+    schedule_to_close_timeout: int | None = None
+    heartbeat_timeout: int | None = None
 
     def to_server_command(
         self, task_queue: str, *, payload_codec: str = serializer.AVRO_CODEC
     ) -> dict[str, Any]:
-        return {
+        command: dict[str, Any] = {
             "type": "schedule_activity",
             "activity_type": self.activity_type,
             "arguments": serializer.envelope(self.arguments, codec=payload_codec),
             "queue": self.queue or task_queue,
         }
+        if self.retry_policy is not None:
+            command["retry_policy"] = (
+                self.retry_policy.to_dict()
+                if isinstance(self.retry_policy, ActivityRetryPolicy)
+                else dict(self.retry_policy)
+            )
+        if self.start_to_close_timeout is not None:
+            command["start_to_close_timeout"] = self.start_to_close_timeout
+        if self.schedule_to_start_timeout is not None:
+            command["schedule_to_start_timeout"] = self.schedule_to_start_timeout
+        if self.schedule_to_close_timeout is not None:
+            command["schedule_to_close_timeout"] = self.schedule_to_close_timeout
+        if self.heartbeat_timeout is not None:
+            command["heartbeat_timeout"] = self.heartbeat_timeout
+        return command
 
 
 @dataclass
@@ -266,9 +341,27 @@ def __init__(self, *, run_id: str = "", current_time: datetime | None = None) ->
         self.logger = _ReplayLogger(_REPLAY_LOGGER)
 
     def schedule_activity(
-        self, activity_type: str, arguments: list[Any], *, queue: str | None = None
+        self,
+        activity_type: str,
+        arguments: list[Any],
+        *,
+        queue: str | None = None,
+        retry_policy: ActivityRetryPolicyInput | None = None,
+        start_to_close_timeout: int | None = None,
+        schedule_to_start_timeout: int | None = None,
+        schedule_to_close_timeout: int | None = None,
+        heartbeat_timeout: int | None = None,
     ) -> ScheduleActivity:
-        return ScheduleActivity(activity_type=activity_type, arguments=list(arguments), queue=queue)
+        return ScheduleActivity(
+            activity_type=activity_type,
+            arguments=list(arguments),
+            queue=queue,
+            retry_policy=retry_policy,
+            start_to_close_timeout=start_to_close_timeout,
+            schedule_to_start_timeout=schedule_to_start_timeout,
+            schedule_to_close_timeout=schedule_to_close_timeout,
+            heartbeat_timeout=heartbeat_timeout,
+        )
 
     def start_timer(self, seconds: int) -> StartTimer:
         return StartTimer(delay_seconds=seconds)

tests/test_replay.py

@@ -3,6 +3,7 @@
 from durable_workflow import serializer, workflow
 from durable_workflow.errors import ChildWorkflowFailed
 from durable_workflow.workflow import (
+    ActivityRetryPolicy,
     CompleteWorkflow,
     ContinueAsNew,
     FailWorkflow,
@@ -111,6 +112,38 @@ def test_server_command_uses_payload_codec(self) -> None:
         assert server_cmd["arguments"]["codec"] == "json"
         assert serializer.decode(server_cmd["arguments"]["blob"], codec="json") == ["world"]
 
+    def test_schedule_activity_server_command_includes_retry_policy_and_timeouts(self) -> None:
+        cmd = ScheduleActivity(
+            activity_type="charge-card",
+            arguments=[{"order_id": "o-1"}],
+            queue="payments",
+            retry_policy=ActivityRetryPolicy(
+                max_attempts=4,
+                initial_interval_seconds=1,
+                backoff_coefficient=3,
+                maximum_interval_seconds=10,
+                non_retryable_error_types=["ValidationError"],
+            ),
+            start_to_close_timeout=120,
+            schedule_to_start_timeout=10,
+            schedule_to_close_timeout=300,
+            heartbeat_timeout=15,
+        )
+
+        server_cmd = cmd.to_server_command("default-queue")
+
+        assert server_cmd["type"] == "schedule_activity"
+        assert server_cmd["queue"] == "payments"
+        assert server_cmd["retry_policy"] == {
+            "max_attempts": 4,
+            "backoff_seconds": [1, 3, 9],
+            "non_retryable_error_types": ["ValidationError"],
+        }
+        assert server_cmd["start_to_close_timeout"] == 120
+        assert server_cmd["schedule_to_start_timeout"] == 10
+        assert server_cmd["schedule_to_close_timeout"] == 300
+        assert server_cmd["heartbeat_timeout"] == 15
+
 
 class TestTwoActivities:
     def test_first_schedules(self) -> None:
@@ -368,6 +401,26 @@ def test_logger_active_when_not_replaying(self) -> None:
         finally:
             logger.removeHandler(handler)
 
+    def test_schedule_activity_accepts_retry_policy_and_timeouts(self) -> None:
+        ctx = WorkflowContext(run_id="r1")
+        policy = ActivityRetryPolicy(max_attempts=2, backoff_seconds=[7])
+
+        cmd = ctx.schedule_activity(
+            "charge-card",
+            [{"order_id": "o-1"}],
+            retry_policy=policy,
+            start_to_close_timeout=120,
+            schedule_to_start_timeout=10,
+            schedule_to_close_timeout=300,
+            heartbeat_timeout=15,
+        )
+
+        assert cmd.retry_policy is policy
+        assert cmd.start_to_close_timeout == 120
+        assert cmd.schedule_to_start_timeout == 10
+        assert cmd.schedule_to_close_timeout == 300
+        assert cmd.heartbeat_timeout == 15
+
 
 class TestReplayWithRunId:
     def test_run_id_passed_to_context(self) -> None:

tests/test_retry_policy.py

@@ -3,10 +3,13 @@
 import httpx
 import pytest
 
-from durable_workflow.retry_policy import RetryPolicy
+from durable_workflow.retry_policy import RetryPolicy, TransportRetryPolicy
 
 
 class TestRetryPolicy:
+    def test_retry_policy_alias_kept_for_backward_compatibility(self) -> None:
+        assert RetryPolicy is TransportRetryPolicy
+
     def test_should_retry_connection_error(self) -> None:
         policy = RetryPolicy(max_attempts=3)
         exc = httpx.ConnectError("connection failed")