Add capped exponential activity retry authoring

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `ActivityRetryPolicy.exponential(...)` creates generated activity retry schedules with a finite 60-second default maximum. It rejects `None` or nonfinite maxima and retains the existing wire shape and interval rounding. The legacy constructor, explicit backoff lists and policies already recorded in history keep their original values. (durable-workflow/.github#219)
+
 ## [2.17.2] - 2026-11-13
 
 ### Fixed

docs/sdk-reference.md

@@ -235,7 +235,7 @@ from durable_workflow import ActivityRetryPolicy
 result = yield ctx.schedule_activity(
     "charge-card",
     [order],
-    retry_policy=ActivityRetryPolicy(
+    retry_policy=ActivityRetryPolicy.exponential(
         max_attempts=4,
         initial_interval_seconds=1,
         backoff_coefficient=2,
@@ -248,6 +248,20 @@ result = yield ctx.schedule_activity(
 )
 ```
 
+Use `ActivityRetryPolicy.exponential(...)` for new generated retry schedules.
+Its maximum defaults to 60 seconds. For nine total attempts, the default
+intervals are `1, 2, 4, 8, 16, 32, 60, 60` seconds. `max_attempts` includes
+the first attempt; retries still share the activity's schedule-to-close
+deadline. An explicit `maximum_interval_seconds` selects a different finite,
+non-negative cap. Fractional intervals are capped before rounding up to whole
+command seconds. `None`, infinities and NaN are refused by the factory.
+
+The legacy `ActivityRetryPolicy(...)` constructor keeps its existing behavior:
+omitting `maximum_interval_seconds`, or supplying `None`, applies no cap.
+Existing explicit backoff lists and policies already recorded in history keep
+their original values. Adopt the factory in new workflow definitions without
+changing commands that an existing run must replay.
+
 Child workflow starts use the same retry policy shape and workflow-level
 execution/run timeout names:
 
@@ -386,6 +400,8 @@ entry. For example, five attempts with `[1, 2]` wait `1, 2, 2, 2` seconds before
 attempts two through five. An empty list adds no retry wait. A committed local
 result replays its recorded attempts without rerunning the handler, including
 older records made with a zero-delay tail.
+For a generated local policy, pass `ActivityRetryPolicy.exponential(...).to_dict()`
+as `retry_policy` to select the same finite cap and total-attempt rule.
 
 ## Worker sessions
 

examples/separate_workers.py

@@ -37,7 +37,7 @@ def run(self, ctx, name: str, retry_once: bool, activity_queue: str | None):
             ACTIVITY_TYPE,
             [name, retry_once],
             queue=activity_queue,
-            retry_policy=ActivityRetryPolicy(max_attempts=2, initial_interval_seconds=1),
+            retry_policy=ActivityRetryPolicy.exponential(max_attempts=2, initial_interval_seconds=1),
             start_to_close_timeout=5,
             schedule_to_close_timeout=20,
         ))

src/durable_workflow/workflow.py

@@ -366,6 +366,10 @@ class ActivityRetryPolicy:
     ``non_retryable_error_types`` names failure types that should bypass this
     retry budget. An activity worker can also report ``non_retryable=True`` on
     a failure to stop retrying that activity execution.
+
+    Use :meth:`exponential` for generated backoff with a finite default cap.
+    The legacy constructor preserves its original omission behavior: no
+    maximum is applied when ``maximum_interval_seconds`` is ``None``.
     """
 
     max_attempts: int = 3
@@ -375,6 +379,47 @@ class ActivityRetryPolicy:
     non_retryable_error_types: list[str] = field(default_factory=list)
     backoff_seconds: list[int] | None = None
 
+    @classmethod
+    def exponential(
+        cls,
+        *,
+        max_attempts: int = 3,
+        initial_interval_seconds: float = 1.0,
+        backoff_coefficient: float = 2.0,
+        maximum_interval_seconds: float = 60.0,
+        non_retryable_error_types: list[str] | None = None,
+    ) -> ActivityRetryPolicy:
+        """Generate capped backoff, counting the first attempt in ``max_attempts``.
+
+        The maximum defaults to 60 seconds and must be finite and non-negative.
+        Generated fractional intervals retain the constructor's rounding to
+        whole command seconds. This factory does not change existing policies
+        or the legacy constructor's defaults.
+        """
+        if type(max_attempts) is not int or max_attempts < 1:
+            raise ValueError("max_attempts must be an integer >= 1")
+        for name, value, minimum in (
+            ("initial_interval_seconds", initial_interval_seconds, 0),
+            ("backoff_coefficient", backoff_coefficient, 1),
+            ("maximum_interval_seconds", maximum_interval_seconds, 0),
+        ):
+            try:
+                valid = (
+                    isinstance(value, (int, float)) and not isinstance(value, bool)
+                    and math.isfinite(value) and value >= minimum
+                )
+            except OverflowError:
+                valid = False
+            if not valid:
+                raise ValueError(f"{name} must be a finite number >= {minimum}")
+        return cls(
+            max_attempts=max_attempts,
+            initial_interval_seconds=initial_interval_seconds,
+            backoff_coefficient=backoff_coefficient,
+            maximum_interval_seconds=maximum_interval_seconds,
+            non_retryable_error_types=list(non_retryable_error_types or []),
+        )
+
     def to_dict(self) -> dict[str, Any]:
         """Return the server command shape for this activity retry policy."""
         if self.max_attempts < 1:

tests/fixtures/activity_retry_legacy_2172.json

@@ -0,0 +1,32 @@
+{
+  "sdk_version": "2.17.2",
+  "wheel_sha256": "7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc",
+  "samples": [
+    {
+      "id": "omitted-maximum",
+      "options": {
+        "max_attempts": 20
+      },
+      "command_json": "{\"activity_type\":\"tests.retry\",\"arguments\":{\"blob\":\"wwHioz3/VYAiNwwCCgppbnB1dAA=\",\"codec\":\"avro\"},\"queue\":\"queue\",\"retry_policy\":{\"backoff_seconds\":[1,2,4,8,16,32,64,128,256,512,1024,2048,4096,8192,16384,32768,65536,131072,262144],\"max_attempts\":20,\"non_retryable_error_types\":[]},\"type\":\"schedule_activity\"}"
+    },
+    {
+      "id": "explicit-none",
+      "options": {
+        "max_attempts": 20,
+        "maximum_interval_seconds": null
+      },
+      "command_json": "{\"activity_type\":\"tests.retry\",\"arguments\":{\"blob\":\"wwHioz3/VYAiNwwCCgppbnB1dAA=\",\"codec\":\"avro\"},\"queue\":\"queue\",\"retry_policy\":{\"backoff_seconds\":[1,2,4,8,16,32,64,128,256,512,1024,2048,4096,8192,16384,32768,65536,131072,262144],\"max_attempts\":20,\"non_retryable_error_types\":[]},\"type\":\"schedule_activity\"}"
+    },
+    {
+      "id": "short-explicit-list",
+      "options": {
+        "max_attempts": 5,
+        "backoff_seconds": [
+          1,
+          2
+        ]
+      },
+      "command_json": "{\"activity_type\":\"tests.retry\",\"arguments\":{\"blob\":\"wwHioz3/VYAiNwwCCgppbnB1dAA=\",\"codec\":\"avro\"},\"queue\":\"queue\",\"retry_policy\":{\"backoff_seconds\":[1,2],\"max_attempts\":5,\"non_retryable_error_types\":[]},\"type\":\"schedule_activity\"}"
+    }
+  ]
+}

tests/test_activity_exponential_retry.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from durable_workflow import ActivityRetryPolicy, serializer, workflow
+from durable_workflow.workflow import CompleteWorkflow, RecordLocalActivity, ScheduleActivity, replay
+
+
+def test_generated_default_cap_is_finite_without_changing_total_attempts():
+    policy = ActivityRetryPolicy.exponential(max_attempts=9)
+    assert policy.maximum_interval_seconds == 60
+    assert policy.to_dict() == {
+        "max_attempts": 9,
+        "backoff_seconds": [1, 2, 4, 8, 16, 32, 60, 60],
+        "non_retryable_error_types": [],
+    }
+
+
+@pytest.mark.parametrize("maximum,expected", [
+    (4, [1, 2, 4, 4]), (2.5, [1, 2, 3, 3]), (0, [0, 0, 0, 0]),
+])
+def test_explicit_cap_retains_whole_second_command_resolution(maximum, expected):
+    policy = ActivityRetryPolicy.exponential(max_attempts=5, maximum_interval_seconds=maximum)
+    assert policy.to_dict()["backoff_seconds"] == expected
+
+
+def test_generated_policy_preserves_the_total_execution_deadline_on_the_command():
+    policy = ActivityRetryPolicy.exponential(max_attempts=9)
+    command = ScheduleActivity(
+        "tests.retry", [], retry_policy=policy, start_to_close_timeout=5, schedule_to_close_timeout=15,
+    ).to_server_command("queue")
+    assert command["retry_policy"] == policy.to_dict()
+    assert command["start_to_close_timeout"] == 5
+    assert command["schedule_to_close_timeout"] == 15
+
+
+@pytest.mark.parametrize("field", ["initial_interval_seconds", "maximum_interval_seconds", "backoff_coefficient"])
+@pytest.mark.parametrize("value", [
+    None, float("nan"), float("inf"), float("-inf"), -1, True, "60", pytest.param(10 ** 1000, id="too-large"),
+])
+def test_generated_parameters_refuse_nonfinite_or_unsupported_numbers(field, value: Any):
+    with pytest.raises(ValueError, match=field + " must be a finite number"):
+        ActivityRetryPolicy.exponential(**{field: value})
+
+
+def test_generated_coefficient_cannot_shrink_the_exponential_schedule():
+    with pytest.raises(ValueError, match="backoff_coefficient must be a finite number >= 1"):
+        ActivityRetryPolicy.exponential(backoff_coefficient=0.5)
+
+
+@pytest.mark.parametrize("attempts", [None, True, 0, -1, 3.0, "3"])
+def test_attempt_count_remains_a_positive_total(attempts: Any):
+    with pytest.raises(ValueError, match="max_attempts must be an integer"):
+        ActivityRetryPolicy.exponential(max_attempts=attempts)
+
+
+def test_one_total_attempt_has_no_retry_and_error_types_are_copied():
+    errors = ["ValidationError"]
+    policy = ActivityRetryPolicy.exponential(max_attempts=1, non_retryable_error_types=errors)
+    errors.append("LaterError")
+    assert policy.to_dict() == {
+        "max_attempts": 1, "backoff_seconds": [], "non_retryable_error_types": ["ValidationError"],
+    }
+
+
+def test_legacy_constructor_and_explicit_lists_keep_their_original_schedules():
+    assert ActivityRetryPolicy(max_attempts=20).to_dict()["backoff_seconds"] == [2 ** power for power in range(19)]
+    explicit_none = ActivityRetryPolicy(max_attempts=20, maximum_interval_seconds=None).to_dict()
+    assert explicit_none["backoff_seconds"][-1] == 262144
+    assert ActivityRetryPolicy(max_attempts=5, backoff_seconds=[1, 2]).to_dict() == {
+        "max_attempts": 5, "backoff_seconds": [1, 2], "non_retryable_error_types": [],
+    }
+
+
+LEGACY = json.loads((Path(__file__).parent / "fixtures/activity_retry_legacy_2172.json").read_text())
+
+
+def test_new_authoring_caps_the_published_long_schedule_without_rewriting_it():
+    legacy = json.loads(LEGACY["samples"][0]["command_json"])["retry_policy"]["backoff_seconds"]
+    current = ActivityRetryPolicy.exponential(max_attempts=20).to_dict()["backoff_seconds"]
+    assert legacy[-1] == 262144
+    assert current == [*legacy[:6], *([60] * 13)]
+
+
+@workflow.defn(name="tests.retry.legacy-remote")
+class LegacyRemote:
+    def run(self, ctx, options):
+        return (yield ctx.schedule_activity("tests.retry", ["input"], retry_policy=ActivityRetryPolicy(**options)))
+
+
+@workflow.defn(name="tests.retry.legacy-local")
+class LegacyLocal:
+    def run(self, ctx, policy):
+        return (yield ctx.local_activity("tests.retry", ["input"], retry_policy=policy))
+
+
+@pytest.mark.parametrize("sample", LEGACY["samples"], ids=lambda sample: sample["id"])
+def test_legacy_commands_match_published_sdk_bytes_and_local_history_replays_without_callback(sample):
+    pending = replay(LegacyRemote, [], [sample["options"]])
+    try:
+        actual = pending.commands[0].to_server_command("queue", payload_codec="avro")
+        assert json.dumps(actual, sort_keys=True, separators=(",", ":")) == sample["command_json"]
+    finally:
+        pending.close()
+    recorded_policy = json.loads(sample["command_json"])["retry_policy"]
+    backoff = recorded_policy["backoff_seconds"]
+    attempts = [
+        {"attempt_number": number, "outcome": "failed", "backoff_seconds": backoff[min(number - 1, len(backoff) - 1)]}
+        for number in range(1, recorded_policy["max_attempts"])
+    ]
+    attempts.append({"attempt_number": recorded_policy["max_attempts"], "outcome": "completed"})
+    history = [{"event_type": "ActivityCompleted", "payload": {
+        "workflow_sequence": 1, "activity_type": "tests.retry", "execution_mode": "local",
+        "retry_policy": recorded_policy, "attempts": attempts, "result": serializer.envelope("retained"),
+    }}]
+    retained = deepcopy(history)
+
+    def refuse_callback(_: RecordLocalActivity):
+        pytest.fail("committed legacy local result must not execute a callback")
+
+    outcome = replay(LegacyLocal, history, [recorded_policy], local_activity_executor=refuse_callback)
+    try:
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == "retained"
+        assert history == retained
+    finally:
+        outcome.close()

tests/test_local_activity_backoff.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from durable_workflow import Client, Worker, activity
+from durable_workflow import ActivityRetryPolicy, Client, Worker, activity
 from durable_workflow import worker as worker_module
 from durable_workflow.workflow import RecordLocalActivity
 
@@ -47,3 +47,10 @@ async def test_local_retries_without_a_backoff_list_do_not_wait(monkeypatch: pyt
     slept, attempts = await _run_failing_local_activity(monkeypatch, {"max_attempts": 3})
     assert slept == []
     assert len(attempts) == 3
+
+
+async def test_new_generated_cap_reaches_the_existing_local_executor(monkeypatch: pytest.MonkeyPatch) -> None:
+    policy = ActivityRetryPolicy.exponential(max_attempts=9).to_dict()
+    slept, attempts = await _run_failing_local_activity(monkeypatch, policy)
+    assert slept == [1, 2, 4, 8, 16, 32, 60, 60]
+    assert [attempt.get("backoff_seconds") for attempt in attempts] == [*slept, None]