Honor non-retriable activity and sub-orchestration failures

CHANGELOG.md

@@ -34,6 +34,11 @@ A single callback failure is still raised directly.
 
 FIXED
 
+- 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
+retried such failures can replay differently; drain affected instances on the
+previous SDK before upgrading.
 - Grouped failures now retain nested exception messages, tracebacks, and notes
 in failure details, including `ExceptionGroup` failures from task callbacks.
 Traceback rendering is limited to 15 children per group and 10 group levels,

azure-functions-durable/CHANGELOG.md

@@ -29,6 +29,11 @@ can change during replay; see the [core upgrade warning](../CHANGELOG.md#unrelea
 
 FIXED
 
+- 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.
+Histories that already retried such failures can replay differently; drain
+affected instances on the previous SDK before upgrading.
 - With the corresponding core SDK update, grouped failures retain nested
 exception messages, tracebacks, and notes in failure details. Tracebacks are
 limited to 15 children per group and 10 group levels with truncation markers;

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
+policies honor `failureDetails.isNonRetriable` and surface the failure immediately
+without retrying. Histories that already retried such failures can replay
+differently; drain affected instances on the previous SDK before upgrading.
 - With the corresponding core SDK update, grouped failures retain nested
 exception messages, tracebacks, and notes in failure details. Tracebacks are
 limited to 15 children per group and 10 group levels with truncation markers;

durabletask/worker.py

@@ -2872,7 +2872,10 @@ def _cancel_timer() -> None:
                 activity_task_obj = cast(object, activity_task)
                 if isinstance(activity_task_obj, task.RetryableTask):
                     retryable_activity_task = cast(task.RetryableTask[Any], activity_task_obj)
-                    next_delay = retryable_activity_task.compute_next_delay()
+                    next_delay = (
+                        None if event.taskFailed.failureDetails.isNonRetriable
+                        else retryable_activity_task.compute_next_delay()
+                    )
                     if next_delay is None:
                         retryable_activity_task.fail(
                             f"{ctx.instance_id}: Activity task #{task_id} failed: {event.taskFailed.failureDetails.errorMessage}",
@@ -2983,7 +2986,10 @@ def _cancel_timer() -> None:
                 sub_orch_task_obj = cast(object, sub_orch_task)
                 if isinstance(sub_orch_task_obj, task.RetryableTask):
                     retryable_sub_orch_task = cast(task.RetryableTask[Any], sub_orch_task_obj)
-                    next_delay = retryable_sub_orch_task.compute_next_delay()
+                    next_delay = (
+                        None if failedEvent.failureDetails.isNonRetriable
+                        else retryable_sub_orch_task.compute_next_delay()
+                    )
                     if next_delay is None:
                         retryable_sub_orch_task.fail(
                             f"Sub-orchestration task #{task_id} failed: {failedEvent.failureDetails.errorMessage}",

tests/durabletask/test_orchestration_executor.py

@@ -3,7 +3,7 @@
 
 import json
 import logging
-from datetime import datetime, timedelta
+from datetime import UTC, datetime, timedelta
 from typing import Any
 
 import pytest
@@ -938,6 +938,91 @@ def get_exception_properties(self, exception: Exception):
     assert failure.innerFailure.properties["code"].string_value == "VALUE"
 
 
+@pytest.mark.parametrize("is_sub_orch", [False, True], ids=["activity", "sub-orchestration"])
+@pytest.mark.parametrize("is_non_retriable", [None, False, True], ids=["unset", "false", "true"])
+@pytest.mark.parametrize("is_replaying", [False, True], ids=["new-failure", "replayed-failure"])
+@pytest.mark.parametrize("catch_failure", [False, True], ids=["unhandled", "caught"])
+def test_retry_policy_respects_non_retriable_failure(
+    is_sub_orch: bool,
+    is_non_retriable: bool | None,
+    is_replaying: bool,
+    catch_failure: bool,
+):
+    """Non-retriable failures bypass retries and retain their failure details."""
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        retry_policy = task.RetryPolicy(
+            first_retry_interval=timedelta(seconds=1),
+            max_number_of_attempts=3,
+        )
+        try:
+            if is_sub_orch:
+                yield ctx.call_sub_orchestrator("child", retry_policy=retry_policy)
+            else:
+                yield ctx.call_activity("child", retry_policy=retry_policy)
+        except task.TaskFailedError as error:
+            if not catch_failure:
+                raise
+            assert error.details.error_type == "builtins.ValueError"
+            assert error.details.message == "Permanent failure"
+            return "compensated"
+        return None
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+    current_timestamp = datetime.now(UTC).replace(tzinfo=None)
+    old_events = [
+        helpers.new_orchestrator_started_event(current_timestamp),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None),
+    ]
+    if is_sub_orch:
+        old_events.append(helpers.new_sub_orchestration_created_event(1, "child", "child-instance"))
+        failed_event = helpers.new_sub_orchestration_failed_event(1, ValueError("Permanent failure"))
+        failure_details = failed_event.subOrchestrationInstanceFailed.failureDetails
+    else:
+        old_events.append(helpers.new_task_scheduled_event(1, "child"))
+        failed_event = helpers.new_task_failed_event(1, ValueError("Permanent failure"))
+        failure_details = failed_event.taskFailed.failureDetails
+    if is_non_retriable is not None:
+        failure_details.isNonRetriable = is_non_retriable
+
+    new_events = [helpers.new_orchestrator_started_event(current_timestamp)]
+    if is_replaying:
+        old_events.append(failed_event)
+    else:
+        new_events.append(failed_event)
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+
+    if is_non_retriable:
+        completion = get_and_validate_complete_orchestration_action_list(1, result.actions)
+        if catch_failure:
+            assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+            assert completion.result.value == json.dumps("compensated")
+        else:
+            assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+            assert completion.failureDetails.innerFailure == failure_details
+    else:
+        assert len(result.actions) == 1
+        timer = result.actions[0]
+        assert timer.HasField("createTimer")
+        expected_fire_at = current_timestamp + timedelta(seconds=1)
+        assert timer.createTimer.fireAt.ToDatetime() == expected_fire_at
+
+        old_events += new_events
+        executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+        result = executor.execute(TEST_INSTANCE_ID, old_events, [
+            helpers.new_orchestrator_started_event(expected_fire_at),
+            helpers.new_timer_fired_event(timer.id, expected_fire_at),
+        ])
+        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
+
+
 def test_activity_retry_policies():
     """Tests the retry policy logic for activity tasks"""