Fix replay determinism for retry timeouts

CHANGELOG.md

@@ -7,6 +7,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+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.
+
 ## v1.13.0
 
 ADDED

azure-functions-durable/CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+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.
+
 ## v2.0.0rc4
 
 ADDED

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+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.
+
 ## v1.13.0
 
 ADDED

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.utcoffset() is not None:
+                current_time = current_time.astimezone(UTC).replace(tzinfo=None)
+            start_time = self._start_time
+            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
 
         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]):

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(

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():

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():

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():

tests/durabletask/test_retry_timeout.py

@@ -0,0 +1,206 @@
+# 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("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])
+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"'