From f7c7e190059a885ba94a3810ab0279bb0ab74caa Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:13:26 +0000 Subject: [PATCH 1/2] feat: add capped exponential activity retry authoring --- CHANGELOG.md | 4 + docs/sdk-reference.md | 18 ++- examples/separate_workers.py | 2 +- src/durable_workflow/workflow.py | 45 +++++++ .../fixtures/activity_retry_legacy_2172.json | 32 +++++ tests/test_activity_exponential_retry.py | 123 ++++++++++++++++++ tests/test_local_activity_backoff.py | 9 +- 7 files changed, 230 insertions(+), 3 deletions(-) create mode 100644 tests/fixtures/activity_retry_legacy_2172.json create mode 100644 tests/test_activity_exponential_retry.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 69a93c3f..7ca988f6 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index e7a2ce9e..913a9225 100644 --- a/docs/sdk-reference.md +++ b/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 diff --git a/examples/separate_workers.py b/examples/separate_workers.py index f21df468..942f63d6 100644 --- a/examples/separate_workers.py +++ b/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, )) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 47f01a01..5a3db4c3 100644 --- a/src/durable_workflow/workflow.py +++ b/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: diff --git a/tests/fixtures/activity_retry_legacy_2172.json b/tests/fixtures/activity_retry_legacy_2172.json new file mode 100644 index 00000000..74455595 --- /dev/null +++ b/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\"}" + } + ] +} diff --git a/tests/test_activity_exponential_retry.py b/tests/test_activity_exponential_retry.py new file mode 100644 index 00000000..2eff4fb3 --- /dev/null +++ b/tests/test_activity_exponential_retry.py @@ -0,0 +1,123 @@ +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, 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 + + +@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() diff --git a/tests/test_local_activity_backoff.py b/tests/test_local_activity_backoff.py index a24d089a..b004068f 100644 --- a/tests/test_local_activity_backoff.py +++ b/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] From 6c94f6a152ec5434c607b427a36b5af855403a70 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:15:18 +0000 Subject: [PATCH 2/2] test: retain total deadline with capped activity retries --- tests/test_activity_exponential_retry.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/tests/test_activity_exponential_retry.py b/tests/test_activity_exponential_retry.py index 2eff4fb3..b4060967 100644 --- a/tests/test_activity_exponential_retry.py +++ b/tests/test_activity_exponential_retry.py @@ -8,7 +8,7 @@ import pytest from durable_workflow import ActivityRetryPolicy, serializer, workflow -from durable_workflow.workflow import CompleteWorkflow, RecordLocalActivity, replay +from durable_workflow.workflow import CompleteWorkflow, RecordLocalActivity, ScheduleActivity, replay def test_generated_default_cap_is_finite_without_changing_total_attempts(): @@ -29,6 +29,16 @@ def test_explicit_cap_retains_whole_second_command_resolution(maximum, expected) 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"),