From 15b4e29b3f16fb560843e44d2cd77597c1a971ce Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 2 Nov 2026 12:19:26 -0700 Subject: [PATCH 1/3] Fix replay determinism for retry timeouts Use orchestration time for activity and sub-orchestration retry expiration, normalize UTC timestamps, and cover replay and deadline boundaries. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 5 + azure-functions-durable/CHANGELOG.md | 4 + durabletask-azuremanaged/CHANGELOG.md | 4 + durabletask/task.py | 31 +++-- durabletask/worker.py | 4 +- tests/durabletask/test_retry_timeout.py | 173 ++++++++++++++++++++++++ 6 files changed, 208 insertions(+), 13 deletions(-) create mode 100644 tests/durabletask/test_retry_timeout.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d44516bf..59eb0afe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,11 @@ A single callback failure is still raised directly. FIXED +- Activity and sub-orchestration `retry_timeout` decisions now use replay-safe +orchestration time, preventing retries from expiring solely because history is +replayed later. Retry calculations also accept timezone-aware timestamps. +Histories that recorded timeout failures based on wall-clock time can replay +differently; drain affected instances on the previous SDK before upgrading. - Activity and sub-orchestration retry policies now honor `failureDetails.isNonRetriable`, surfacing the failure immediately without scheduling a retry timer or another attempt. Existing histories that already diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 90636829..a0c76103 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -29,6 +29,10 @@ can change during replay; see the [core upgrade warning](../CHANGELOG.md#unrelea FIXED +- With the corresponding core SDK update, activity and sub-orchestration +`retry_timeout` decisions use replay-safe orchestration time rather than +wall-clock time. See the [core upgrade guidance](../CHANGELOG.md#unreleased) +for histories that recorded timeout failures under the previous behavior. - With the corresponding core SDK update, activity and sub-orchestration retry policies, including compatibility `RetryOptions`, honor `failureDetails.isNonRetriable` and surface the failure immediately without retrying. diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 43812433..573d6c96 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -20,6 +20,10 @@ can change during replay; see the [core upgrade warning](../CHANGELOG.md#unrelea FIXED +- With the corresponding core SDK update, activity and sub-orchestration +`retry_timeout` decisions use replay-safe orchestration time rather than +wall-clock time. See the [core upgrade guidance](../CHANGELOG.md#unreleased) +for histories that recorded timeout failures under the previous behavior. - With the corresponding core SDK update, activity and sub-orchestration retry policies honor `failureDetails.isNonRetriable` and surface the failure immediately without retrying. Histories that already retried such failures can replay diff --git a/durabletask/task.py b/durabletask/task.py index 17cc7f3d..83eb3833 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -826,28 +826,37 @@ def __init__(self, retry_policy: RetryPolicy, action: pb.OrchestratorAction, def increment_attempt_count(self) -> None: self._attempt_count += 1 - def compute_next_delay(self) -> timedelta | None: + def compute_next_delay(self, current_time: datetime | None = None) -> timedelta | None: + """Compute the retry delay using the supplied orchestration time. + + Naive timestamps are treated as UTC. If omitted, current_time defaults + to wall-clock UTC for compatibility with callers outside the executor. + """ if self._attempt_count >= self._retry_policy.max_number_of_attempts: return None - retry_expiration: datetime = datetime.max - if self._retry_policy.retry_timeout is not None and self._retry_policy.retry_timeout != datetime.max: - retry_expiration = self._start_time + self._retry_policy.retry_timeout + if self._retry_policy.retry_timeout is not None: + if current_time is None: + current_time = datetime.now(tz=UTC) + if current_time.tzinfo is not None: + current_time = current_time.astimezone(UTC).replace(tzinfo=None) + start_time = self._start_time + if start_time.tzinfo is not None: + start_time = start_time.astimezone(UTC).replace(tzinfo=None) + if current_time - start_time >= self._retry_policy.retry_timeout: + return None if self._retry_policy.backoff_coefficient is None: backoff_coefficient = 1.0 else: backoff_coefficient = self._retry_policy.backoff_coefficient - if datetime.now(tz=UTC).replace(tzinfo=None) < retry_expiration: - next_delay_f = math.pow(backoff_coefficient, self._attempt_count - 1) * self._retry_policy.first_retry_interval.total_seconds() + next_delay_f = math.pow(backoff_coefficient, self._attempt_count - 1) * self._retry_policy.first_retry_interval.total_seconds() - if self._retry_policy.max_retry_interval is not None: - next_delay_f = min(next_delay_f, self._retry_policy.max_retry_interval.total_seconds()) + if self._retry_policy.max_retry_interval is not None: + next_delay_f = min(next_delay_f, self._retry_policy.max_retry_interval.total_seconds()) - return timedelta(seconds=next_delay_f) - - return None + return timedelta(seconds=next_delay_f) class TimerTask(CancellableTask[None]): diff --git a/durabletask/worker.py b/durabletask/worker.py index e4099042..2786c75e 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -2874,7 +2874,7 @@ def _cancel_timer() -> None: retryable_activity_task = cast(task.RetryableTask[Any], activity_task_obj) next_delay = ( None if event.taskFailed.failureDetails.isNonRetriable - else retryable_activity_task.compute_next_delay() + else retryable_activity_task.compute_next_delay(ctx.current_utc_datetime) ) if next_delay is None: retryable_activity_task.fail( @@ -2988,7 +2988,7 @@ def _cancel_timer() -> None: retryable_sub_orch_task = cast(task.RetryableTask[Any], sub_orch_task_obj) next_delay = ( None if failedEvent.failureDetails.isNonRetriable - else retryable_sub_orch_task.compute_next_delay() + else retryable_sub_orch_task.compute_next_delay(ctx.current_utc_datetime) ) if next_delay is None: retryable_sub_orch_task.fail( diff --git a/tests/durabletask/test_retry_timeout.py b/tests/durabletask/test_retry_timeout.py new file mode 100644 index 00000000..46da0056 --- /dev/null +++ b/tests/durabletask/test_retry_timeout.py @@ -0,0 +1,173 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +# pyright: reportPrivateUsage=false + +import logging +from collections.abc import Generator +from datetime import UTC, datetime, timedelta, timezone, tzinfo +from typing import Any + +import pytest + +from durabletask import task, worker +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb +from durabletask.serialization import JsonDataConverter + + +def _set_wall_clock(monkeypatch: pytest.MonkeyPatch, instant: datetime) -> None: + class FrozenDateTime(datetime): + @classmethod + def now(cls, tz: tzinfo | None = None) -> datetime: + return instant.astimezone(tz) if tz is not None else instant.replace(tzinfo=None) + + monkeypatch.setattr(task, "datetime", FrozenDateTime) + + +@pytest.mark.parametrize("start_zone", [None, UTC, timezone(timedelta(hours=2))]) +@pytest.mark.parametrize("current_zone", [None, UTC, timezone(timedelta(hours=-3))]) +@pytest.mark.parametrize("elapsed", [9, 10, 11]) +def test_retry_timeout_uses_supplied_time( + start_zone: tzinfo | None, current_zone: tzinfo | None, elapsed: int, +) -> None: + start = datetime(2020, 1, 1, tzinfo=UTC) + current = start + timedelta(seconds=elapsed) + start = start.astimezone(start_zone) if start_zone is not None else start.replace(tzinfo=None) + current = current.astimezone(current_zone) if current_zone is not None else current.replace(tzinfo=None) + retry = task.RetryableTask[Any]( + task.RetryPolicy( + first_retry_interval=timedelta(seconds=1), + max_number_of_attempts=3, + retry_timeout=timedelta(seconds=10), + ), + pb.OrchestratorAction(), + start, + is_sub_orch=False, + ) + + assert retry.compute_next_delay(current) == (timedelta(seconds=1) if elapsed < 10 else None) + + +@pytest.mark.parametrize("timeout", [None, timedelta(seconds=10)]) +@pytest.mark.parametrize("aware_start", [False, True]) +@pytest.mark.parametrize("elapsed", [1, 10]) +def test_retry_timeout_default_clock( + monkeypatch: pytest.MonkeyPatch, timeout: timedelta | None, aware_start: bool, elapsed: int, +) -> None: + start = datetime(2020, 1, 1, tzinfo=UTC) + _set_wall_clock(monkeypatch, start + timedelta(seconds=elapsed)) + retry = task.RetryableTask[Any]( + task.RetryPolicy( + first_retry_interval=timedelta(seconds=1), + max_number_of_attempts=3, + retry_timeout=timeout, + ), + pb.OrchestratorAction(), + start if aware_start else start.replace(tzinfo=None), + is_sub_orch=False, + ) + + expected = None if timeout is not None and elapsed >= 10 else timedelta(seconds=1) + assert retry.compute_next_delay() == expected + assert retry.compute_next_delay(start + timedelta(seconds=elapsed)) == expected + + +@pytest.mark.parametrize("timeout", [timedelta(0), timedelta.max]) +def test_retry_timeout_extreme_durations(timeout: timedelta) -> None: + start = datetime(2020, 1, 1) + retry = task.RetryableTask[Any]( + task.RetryPolicy( + first_retry_interval=timedelta(seconds=1), + max_number_of_attempts=3, + retry_timeout=timeout, + ), + pb.OrchestratorAction(), + start, + is_sub_orch=False, + ) + + assert retry.compute_next_delay(start) == (None if timeout == timedelta(0) else timedelta(seconds=1)) + + +@pytest.mark.parametrize("is_sub_orch", [False, True], ids=["activity", "sub-orchestration"]) +@pytest.mark.parametrize("elapsed", [1, 9, 10, 11]) +@pytest.mark.parametrize("wall_clock_year", [2019, 2030]) +def test_retry_timeout_replays_history( + monkeypatch: pytest.MonkeyPatch, is_sub_orch: bool, elapsed: int, wall_clock_year: int, +) -> None: + _set_wall_clock(monkeypatch, datetime(wall_clock_year, 1, 1, tzinfo=UTC)) + start = datetime(2020, 1, 1) + failed_at = start + timedelta(seconds=elapsed) + + def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Generator[task.Task[Any], Any, Any]: + retry_policy = task.RetryPolicy( + first_retry_interval=timedelta(seconds=1), + max_number_of_attempts=3, + retry_timeout=timedelta(seconds=10), + ) + if is_sub_orch: + return (yield ctx.call_sub_orchestrator("child", retry_policy=retry_policy)) + return (yield ctx.call_activity("child", retry_policy=retry_policy)) + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + history = [ + helpers.new_orchestrator_started_event(start), + helpers.new_execution_started_event(name, "instance", encoded_input=None), + helpers.new_sub_orchestration_created_event(1, "child", "child-instance") + if is_sub_orch else helpers.new_task_scheduled_event(1, "child"), + ] + failure_events = [ + helpers.new_orchestrator_started_event(failed_at), + helpers.new_sub_orchestration_failed_event(1, ValueError("transient")) + if is_sub_orch else helpers.new_task_failed_event(1, ValueError("transient")), + ] + + def execute(old: list[pb.HistoryEvent], new: list[pb.HistoryEvent]) -> worker.ExecutionResults: + executor = worker._OrchestrationExecutor(registry, logging.getLogger(__name__), JsonDataConverter()) + return executor.execute("instance", old, new) + + result = execute(history, failure_events) + replayed = execute(history + failure_events, [helpers.new_orchestrator_started_event(failed_at)]) + assert len(result.actions) == 1 + assert len(replayed.actions) == 1 + + if elapsed >= 10: + for action in (result.actions[0], replayed.actions[0]): + assert action.HasField("completeOrchestration") + completion = action.completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert completion.failureDetails.innerFailure.errorMessage == "transient" + return + + assert replayed.actions == result.actions + timer = result.actions[0] + assert timer.HasField("createTimer") + fire_at = failed_at + timedelta(seconds=1) + assert timer.createTimer.fireAt.ToDatetime() == fire_at + history += failure_events + [helpers.new_timer_created_event(timer.id, fire_at)] + timer_events = [ + helpers.new_orchestrator_started_event(fire_at), + helpers.new_timer_fired_event(timer.id, fire_at), + ] + result = execute(history, timer_events) + retries = [ + action for action in result.actions + if action.HasField("createSubOrchestration" if is_sub_orch else "scheduleTask") + ] + assert len(retries) == 1 + assert retries[0].id == 1 + + history += timer_events + [ + helpers.new_sub_orchestration_created_event(1, "child", "child-instance") + if is_sub_orch else helpers.new_task_scheduled_event(1, "child"), + ] + result = execute(history, [ + helpers.new_orchestrator_started_event(fire_at), + helpers.new_sub_orchestration_completed_event(1, '"success"') + if is_sub_orch else helpers.new_task_completed_event(1, '"success"'), + ]) + assert len(result.actions) == 1 + assert result.actions[0].completeOrchestration.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert result.actions[0].completeOrchestration.result.value == '"success"' From 4b246630448f7b6ee3240a5935ca0d50ca4ad541 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 2 Nov 2026 12:40:15 -0700 Subject: [PATCH 2/3] Align timeout E2E expectations with replay-safe retry decisions Verify expiration on the fifth failure after the 14-second deadline, with a higher attempt limit so timeout exhaustion is exercised independently. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../test_dts_async_orchestration_e2e.py | 6 ++++-- .../test_dts_orchestration_e2e.py | 10 ++++------ tests/durabletask/test_orchestration_e2e.py | 10 ++++------ 3 files changed, 12 insertions(+), 14 deletions(-) diff --git a/tests/durabletask-azuremanaged/test_dts_async_orchestration_e2e.py b/tests/durabletask-azuremanaged/test_dts_async_orchestration_e2e.py index 5f244382..d3c91120 100644 --- a/tests/durabletask-azuremanaged/test_dts_async_orchestration_e2e.py +++ b/tests/durabletask-azuremanaged/test_dts_async_orchestration_e2e.py @@ -420,10 +420,12 @@ def throw_activity_with_retry(ctx: task.ActivityContext, _): async def test_retry_timeout(): + # Failures at 0, 1, 3, and 7 seconds schedule retries; the fifth attempt + # at 15 seconds exhausts the 14-second timeout, not the 10-attempt limit. throw_activity_counter = 0 retry_policy = task.RetryPolicy( first_retry_interval=timedelta(seconds=1), - max_number_of_attempts=5, + max_number_of_attempts=10, backoff_coefficient=2, max_retry_interval=timedelta(seconds=10), retry_timeout=timedelta(seconds=14)) @@ -452,7 +454,7 @@ def throw_activity(ctx: task.ActivityContext, _): assert state.failure_details.error_type == "durabletask.task.TaskFailedError" assert state.failure_details.message.endswith("Activity task #1 failed: Kah-BOOOOM!!!") assert state.failure_details.stack_trace is not None - assert throw_activity_counter == 4 + assert throw_activity_counter == 5 async def test_custom_status(): diff --git a/tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py b/tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py index 14f9ce05..a4119db6 100644 --- a/tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py +++ b/tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py @@ -547,14 +547,12 @@ def throw_activity_with_retry(ctx: task.ActivityContext, _): def test_retry_timeout(): - # This test verifies that the retry timeout is working as expected. - # Max number of attempts is 5 and retry timeout is 14 seconds. - # Total seconds consumed till 4th attempt is 1 + 2 + 4 + 8 = 15 seconds. - # So, the 5th attempt should not be made and the orchestration should fail. + # Failures at 0, 1, 3, and 7 seconds schedule retries; the fifth attempt + # at 15 seconds exhausts the 14-second timeout, not the 10-attempt limit. throw_activity_counter = 0 retry_policy = task.RetryPolicy( first_retry_interval=timedelta(seconds=1), - max_number_of_attempts=5, + max_number_of_attempts=10, backoff_coefficient=2, max_retry_interval=timedelta(seconds=10), retry_timeout=timedelta(seconds=14)) @@ -583,7 +581,7 @@ def throw_activity(ctx: task.ActivityContext, _): assert state.failure_details.error_type == "durabletask.task.TaskFailedError" assert state.failure_details.message.endswith("Activity task #1 failed: Kah-BOOOOM!!!") assert state.failure_details.stack_trace is not None - assert throw_activity_counter == 4 + assert throw_activity_counter == 5 def test_custom_status(): diff --git a/tests/durabletask/test_orchestration_e2e.py b/tests/durabletask/test_orchestration_e2e.py index 1f898fdd..27c100c3 100644 --- a/tests/durabletask/test_orchestration_e2e.py +++ b/tests/durabletask/test_orchestration_e2e.py @@ -585,14 +585,12 @@ def throw_activity_with_retry(ctx: task.ActivityContext, _): def test_retry_timeout(): - # This test verifies that the retry timeout is working as expected. - # Max number of attempts is 5 and retry timeout is 14 seconds. - # Total seconds consumed till 4th attempt is 1 + 2 + 4 + 8 = 15 seconds. - # So, the 5th attempt should not be made and the orchestration should fail. + # Failures at 0, 1, 3, and 7 seconds schedule retries; the fifth attempt + # at 15 seconds exhausts the 14-second timeout, not the 10-attempt limit. throw_activity_counter = 0 retry_policy = task.RetryPolicy( first_retry_interval=timedelta(seconds=1), - max_number_of_attempts=5, + max_number_of_attempts=10, backoff_coefficient=2, max_retry_interval=timedelta(seconds=10), retry_timeout=timedelta(seconds=14)) @@ -619,7 +617,7 @@ def throw_activity(ctx: task.ActivityContext, _): assert state.failure_details.error_type == "durabletask.task.TaskFailedError" assert state.failure_details.message.endswith("Activity task #1 failed: Kah-BOOOOM!!!") assert state.failure_details.stack_trace is not None - assert throw_activity_counter == 4 + assert throw_activity_counter == 5 def test_custom_status(): From 9ac49b4e46c34f21666c0495a162f10ee8d305e0 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 3 Nov 2026 10:20:24 -0700 Subject: [PATCH 3/3] Treat undefined datetime offsets as naive in retry calculations Guard UTC conversion with utcoffset and cover undefined offsets on both retry timestamps without relying on the host timezone. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- durabletask/task.py | 4 +-- tests/durabletask/test_retry_timeout.py | 33 +++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/durabletask/task.py b/durabletask/task.py index 83eb3833..df2e03a2 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -838,10 +838,10 @@ def compute_next_delay(self, current_time: datetime | None = None) -> timedelta if self._retry_policy.retry_timeout is not None: if current_time is None: current_time = datetime.now(tz=UTC) - if current_time.tzinfo is not None: + if current_time.utcoffset() is not None: current_time = current_time.astimezone(UTC).replace(tzinfo=None) start_time = self._start_time - if start_time.tzinfo is not None: + if start_time.utcoffset() is not None: start_time = start_time.astimezone(UTC).replace(tzinfo=None) if current_time - start_time >= self._retry_policy.retry_timeout: return None diff --git a/tests/durabletask/test_retry_timeout.py b/tests/durabletask/test_retry_timeout.py index 46da0056..daa515ae 100644 --- a/tests/durabletask/test_retry_timeout.py +++ b/tests/durabletask/test_retry_timeout.py @@ -49,6 +49,39 @@ def test_retry_timeout_uses_supplied_time( assert retry.compute_next_delay(current) == (timedelta(seconds=1) if elapsed < 10 else None) +@pytest.mark.parametrize("undefined_start", [False, True]) +@pytest.mark.parametrize("undefined_current", [False, True]) +@pytest.mark.parametrize("elapsed", [1, 10, 11]) +def test_retry_timeout_treats_undefined_offsets_as_naive( + undefined_start: bool, undefined_current: bool, elapsed: int, +) -> None: + class UndefinedOffset(tzinfo): + def utcoffset(self, dt: datetime | None) -> None: + return None + + def dst(self, dt: datetime | None) -> None: + return None + + class NaiveDateTime(datetime): + def astimezone(self, tz: tzinfo | None = None) -> datetime: + raise AssertionError("Naive timestamps must not be converted using the host timezone") + + start = NaiveDateTime(2020, 1, 1, tzinfo=UndefinedOffset() if undefined_start else None) + current = NaiveDateTime(2020, 1, 1, 0, 0, elapsed, tzinfo=UndefinedOffset() if undefined_current else None) + retry = task.RetryableTask[Any]( + task.RetryPolicy( + first_retry_interval=timedelta(seconds=1), + max_number_of_attempts=3, + retry_timeout=timedelta(seconds=10), + ), + pb.OrchestratorAction(), + start, + is_sub_orch=False, + ) + + assert retry.compute_next_delay(current) == (timedelta(seconds=1) if elapsed < 10 else None) + + @pytest.mark.parametrize("timeout", [None, timedelta(seconds=10)]) @pytest.mark.parametrize("aware_start", [False, True]) @pytest.mark.parametrize("elapsed", [1, 10])