From 7b855909776f1e530cd9f7d398bc89a1c741c43e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 11 Nov 2026 18:44:08 +0000 Subject: [PATCH 1/2] Repeat the final local-activity backoff entry Local activity retries used a zero delay once attempts outnumbered the backoff list, so a policy that backs off remotely retried in a tight loop locally. Repeat the final entry instead, matching Server-scheduled activities (durable-workflow/.github#219). Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 +++ src/durable_workflow/worker.py | 5 ++- tests/test_local_activity_backoff.py | 47 ++++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 1 deletion(-) create mode 100644 tests/test_local_activity_backoff.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d916538..1ccff56 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] +### Fixed + +- Local activities no longer retry with zero delay once their backoff list runs out. Each later retry repeats the final backoff entry, as Server-scheduled activities do (durable-workflow/.github#219). + ## [2.15.0] - 2026-11-11 ### Added diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index b3dffe1..c744650 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -2565,7 +2565,10 @@ async def heartbeat( or exception_type in non_retryable_types ) retry = not non_retryable and attempt_number < max_attempts - backoff_seconds = backoff[attempt_number - 1] if retry and len(backoff) >= attempt_number else 0 + # Once the list runs out, keep the final entry, as Server-scheduled + # activities do (.github#219). The list already carries the + # policy's maximum-interval cap. + backoff_seconds = backoff[min(attempt_number, len(backoff)) - 1] if retry and backoff else 0 if ( retry and command.schedule_to_close_timeout is not None and time.monotonic() - started_at + backoff_seconds >= command.schedule_to_close_timeout diff --git a/tests/test_local_activity_backoff.py b/tests/test_local_activity_backoff.py new file mode 100644 index 0000000..028473a --- /dev/null +++ b/tests/test_local_activity_backoff.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import Client, Worker, activity +from durable_workflow import worker as worker_module +from durable_workflow.workflow import RecordLocalActivity + + +async def _run_failing_local_activity(monkeypatch: pytest.MonkeyPatch, policy: dict[str, object]) -> tuple[list[float], list[dict[str, object]]]: + slept: list[float] = [] + + async def record_sleep(seconds: float) -> None: + slept.append(seconds) + + monkeypatch.setattr(worker_module.asyncio, "sleep", record_sleep) + + @activity.defn(name="tests.local.always-fails") + async def always_fails() -> None: + raise RuntimeError("still failing") + + async with Client("http://test", worker_token="worker") as client: + worker = Worker(client, task_queue="queue", activities=[always_fails]) + worker._renew_local_workflow_lease = AsyncMock() # type: ignore[method-assign] + command = RecordLocalActivity("tests.local.always-fails", [], retry_policy=policy) + await worker._execute_local_activity( + {"task_id": "task", "workflow_task_attempt": 1}, command, payload_codec="avro", + ) + assert command.outcome is not None + return slept, list(command.outcome["attempts"]) + + +async def test_local_retries_repeat_the_final_backoff_entry(monkeypatch: pytest.MonkeyPatch) -> None: + slept, attempts = await _run_failing_local_activity( + monkeypatch, {"max_attempts": 5, "backoff_seconds": [1, 2]}, + ) + # Before .github#219 the third and fourth retries used a zero delay. + assert slept == [1, 2, 2, 2] + assert [attempt.get("backoff_seconds") for attempt in attempts] == [1, 2, 2, 2, None] + + +async def test_local_retries_without_a_backoff_list_do_not_wait(monkeypatch: pytest.MonkeyPatch) -> None: + slept, attempts = await _run_failing_local_activity(monkeypatch, {"max_attempts": 3}) + assert slept == [] + assert len(attempts) == 3 From 2262901c59637cdf6e79ff23bd44b77a7be5b5dd Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 11 Nov 2026 18:56:14 +0000 Subject: [PATCH 2/2] Add replay-corpus evidence for the local backoff fix The corpus fixture runs the real local-activity executor and fails on the previous zero-delay tail. Also fix a test line over the length limit. Co-Authored-By: Claude Opus 5.5 --- ...-activity-backoff-repeats-final-entry.json | 27 ++++++++++ tests/test_local_activity_backoff.py | 4 +- tests/test_replay_regression_corpus.py | 50 ++++++++++++++++++- 3 files changed, 78 insertions(+), 3 deletions(-) create mode 100644 tests/fixtures/replay_regressions/local-activity-backoff-repeats-final-entry.json diff --git a/tests/fixtures/replay_regressions/local-activity-backoff-repeats-final-entry.json b/tests/fixtures/replay_regressions/local-activity-backoff-repeats-final-entry.json new file mode 100644 index 0000000..3f289cf --- /dev/null +++ b/tests/fixtures/replay_regressions/local-activity-backoff-repeats-final-entry.json @@ -0,0 +1,27 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "python-local-activity-backoff-repeats-final-entry", + "protocol_version": "1.20", + "bindings": ["python"], + "workflow": { + "type": "sdk.local-activity-backoff", + "input": [{"retry_policy": {"max_attempts": 5, "backoff_seconds": [1, 2]}}] + }, + "command_sequence": [ + { + "type": "local_activity_backoff", + "slept_seconds": [1, 2, 2, 2], + "attempt_backoff_seconds": [1, 2, 2, 2, null] + } + ], + "expected": { + "command_sequence": [ + { + "type": "local_activity_backoff", + "slept_seconds": [1, 2, 2, 2], + "attempt_backoff_seconds": [1, 2, 2, 2, null] + } + ] + } +} diff --git a/tests/test_local_activity_backoff.py b/tests/test_local_activity_backoff.py index 028473a..a24d089 100644 --- a/tests/test_local_activity_backoff.py +++ b/tests/test_local_activity_backoff.py @@ -9,7 +9,9 @@ from durable_workflow.workflow import RecordLocalActivity -async def _run_failing_local_activity(monkeypatch: pytest.MonkeyPatch, policy: dict[str, object]) -> tuple[list[float], list[dict[str, object]]]: +async def _run_failing_local_activity( + monkeypatch: pytest.MonkeyPatch, policy: dict[str, object], +) -> tuple[list[float], list[dict[str, object]]]: slept: list[float] = [] async def record_sleep(seconds: float) -> None: diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 87b50e3..078d1b4 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -10,7 +10,7 @@ import pytest -from durable_workflow import Replayer, Worker, serializer, workflow +from durable_workflow import Replayer, Worker, activity, serializer, workflow from durable_workflow.client import PROTOCOL_VERSION, Client, WorkflowExecution, WorkflowStreamAppendItem from durable_workflow.errors import ( DurableWorkflowError, @@ -502,6 +502,47 @@ def _assert_matches(expected: Any, actual: Any, context: str) -> None: assert actual == expected, f"{context} expected {expected!r}, observed {actual!r}" + +def _execute_local_activity_backoff(inputs: Any) -> dict[str, Any]: + """Run the real local-activity executor with an always-failing callback and + report the delays it waited between attempts (.github#219).""" + from durable_workflow import worker as worker_module + from durable_workflow.workflow import RecordLocalActivity + + assert isinstance(inputs, list) and len(inputs) == 1 and isinstance(inputs[0], dict) + policy = inputs[0]["retry_policy"] + + @activity.defn(name="tests.replay.local-always-fails") + async def always_fails() -> None: + raise RuntimeError("still failing") + + async def run() -> dict[str, Any]: + slept: list[float] = [] + original_sleep = worker_module.asyncio.sleep + + async def record_sleep(seconds: float) -> None: + slept.append(seconds) + + worker_module.asyncio.sleep = record_sleep # type: ignore[assignment] + try: + async with Client("http://regression-corpus", worker_token="worker") as client: + worker = Worker(client, task_queue="queue", activities=[always_fails]) + worker._renew_local_workflow_lease = AsyncMock() # type: ignore[method-assign] + command = RecordLocalActivity("tests.replay.local-always-fails", [], retry_policy=policy) + await worker._execute_local_activity( + {"task_id": "task", "workflow_task_attempt": 1}, command, payload_codec="avro", + ) + finally: + worker_module.asyncio.sleep = original_sleep # type: ignore[assignment] + assert command.outcome is not None + return { + "type": "local_activity_backoff", + "slept_seconds": slept, + "attempt_backoff_seconds": [attempt.get("backoff_seconds") for attempt in command.outcome["attempts"]], + } + + return asyncio.run(run()) + def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]: assert fixture.get("fixture_schema") == FIXTURE_SCHEMA assert "python" in fixture.get("bindings", []) @@ -512,6 +553,11 @@ def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]: assert isinstance(workflow_type, str) and workflow_type if workflow_type == "sdk.observed-worker-replay": return _execute_observed_worker_replay(fixture) + if workflow_type == "sdk.local-activity-backoff": + commands = [_execute_local_activity_backoff(workflow.get("input"))] + _assert_matches(fixture["command_sequence"], commands, f"{fixture['id']}.command_sequence") + _assert_matches(fixture["expected"], {"command_sequence": commands}, f"{fixture['id']}.expected") + return commands if workflow_type in { "sdk.query-task-dispatch", "sdk.generator-query-task-dispatch", "sdk.generator-side-effect-query-task-dispatch", }: @@ -735,7 +781,7 @@ def test_checked_in_replay_regression_corpus_uses_official_replayer( assert isinstance(expected, dict) and expected if workflow.get("type") in { "sdk.replay-conformance-report", "sdk.query-task-dispatch", "sdk.generator-query-task-dispatch", - "sdk.generator-side-effect-query-task-dispatch", "sdk.observed-worker-replay", + "sdk.generator-side-effect-query-task-dispatch", "sdk.observed-worker-replay", "sdk.local-activity-backoff", }: _execute_fixture(fixture) return