diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index d797a44b..f20bb310 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +ADDED + +- Added opt-in `HttpRetryOptions` on `context.call_http()` for durable initial +request retries with configurable timing and status selection. Empty or omitted +status lists select all non-2xx responses; exhausted retries fail the call. +Subsequent 202 Location polls do not inherit the policy. See +[usage and upgrade guidance](README.md#durable-http-failure-retries). +Release coordination requires the corrected core replay-safe retry-timeout +behavior before this provider feature is published; version and dependency +minimum updates belong in the dedicated release PR. + FIXED - With the corresponding core SDK update, activity and sub-orchestration diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 9bf0d77b..469b102b 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -40,6 +40,79 @@ or more task hubs. Filtering support depends on the backend. so they can be used as dictionary keys or set elements. Construct a new ID instead of assigning or deleting attributes on an existing ID. +## Durable HTTP failure retries + +Opt in to durable retries for the **initial** `context.call_http()` request with +`HttpRetryOptions`. It uses the core activity retry engine and durable timers, +not an in-process HTTP retry loop: + +```python +from datetime import timedelta + +import azure.durable_functions as df + + +def orchestrator(context: df.DurableOrchestrationContext): + response = yield context.call_http( + "GET", + "https://example.com/data", + retry_options=df.HttpRetryOptions( + first_retry_interval=timedelta(seconds=2), + max_number_of_attempts=4, + backoff_coefficient=2, + max_retry_interval=timedelta(seconds=30), + retry_timeout=timedelta(minutes=2), + status_codes_to_retry=[429, 503], + ), + ) + return response.content +``` + +Durations use `timedelta`, matching `RetryPolicy`, rather than the deprecated +millisecond-based `RetryOptions`. The attempt limit includes the initial +attempt. Backoff defaults to 1 (constant delay), the maximum retry interval +defaults to six days, and the retry timeout defaults to unbounded. Set +`max_retry_interval=None` to remove the interval cap. + +With a policy, an omitted, `None`, or empty `status_codes_to_retry` selects +**all non-2xx responses**, including unfollowed redirects, matching .NET +`EnsureSuccessStatusCode` semantics, not just 4xx/5xx. A nonempty list selects +exactly those statuses: selecting 202 retries it instead of starting polling. +Excluded statuses return the full `DurableHttpResponse` unchanged. Selected +responses are closed without reading their bodies and become activity +failures; exhaustion fails `call_http` with `durabletask.task.TaskFailedError` +rather than returning the last selected response. Transport and other activity +failures also use the policy regardless of the status list. + +The policy and status list are snapshotted when the request is scheduled. +Omitting `retry_options` preserves the existing request payload and behavior. +Automatic 202/Location polling remains separate: subsequent GET polls inherit +neither the failure retry policy nor its status classification. Poll HTTP +errors return normally, and a poll activity/transport failure fails the call +without these retries. `Retry-After` controls polling delays, not failure retry +delays. + +`retry_timeout` uses replay-safe orchestration time from the initial activity +scheduling. It stops retrying when a failure is handled at or beyond the +timeout. It does **not** cancel or time-limit an in-flight HTTP request, and +the next retry delay may cross the timeout. Unlike the JavaScript core engine, +Python does not reject a retry solely because its next delay would exceed the +remaining timeout. + +> [!WARNING] +> Repeating requests can duplicate side effects, including when a server +> processes a request but its response is lost. Enable retries only for safe +> operations or use server-enforced idempotency keys. Adding or changing retry +> options changes durable scheduling and failure handling. Keep policies +> deterministic; version new caller orchestrations or drain affected in-flight +> instances on the old code before changing their policy. + +This feature requires the corrected core replay-safe retry-timeout behavior in +[#326](https://github.com/microsoft/durabletask-python/pull/326). Release the core +fix before publishing this provider feature, then coordinate the provider +version and core dependency minimum in a dedicated release PR. This feature +does not itself change package versions or dependency minimums. + ## Unicode JSON serialization By default, durable payloads preserve the shared `azure-functions` codec's diff --git a/azure-functions-durable/azure/durable_functions/__init__.py b/azure-functions-durable/azure/durable_functions/__init__.py index df5059a5..de861727 100644 --- a/azure-functions-durable/azure/durable_functions/__init__.py +++ b/azure-functions-durable/azure/durable_functions/__init__.py @@ -8,7 +8,7 @@ from .decorators.durable_app import Blueprint, DFApp from .client import DurableFunctionsClient, SyncDurableFunctionsClient -from .http.models import DurableHttpRequest, DurableHttpResponse +from .http.models import DurableHttpRequest, DurableHttpResponse, HttpRetryOptions from .orchestrator import Orchestrator from .internal.compat.retry_options import RetryOptions from .internal.compat.orchestration_runtime_status import OrchestrationRuntimeStatus @@ -51,6 +51,7 @@ "Entity", "EntityId", "EntityStateResponse", + "HttpRetryOptions", "ManagedIdentityTokenSource", "Orchestrator", "OrchestrationRuntimeStatus", diff --git a/azure-functions-durable/azure/durable_functions/http/builtin.py b/azure-functions-durable/azure/durable_functions/http/builtin.py index 0fe4ff42..cc48058c 100644 --- a/azure-functions-durable/azure/durable_functions/http/builtin.py +++ b/azure-functions-durable/azure/durable_functions/http/builtin.py @@ -34,7 +34,7 @@ from typing import Any, Generator, Optional from urllib.parse import urljoin, urlparse -from .models import DurableHttpResponse +from .models import DurableHttpResponse, HttpRetryOptions # Reserved built-in function names. The v1 host used ``BuiltIn::HttpActivity``; # ``::`` is not a valid Azure Functions function name, so ``__`` is used here. @@ -163,8 +163,10 @@ def builtin_http_activity(input: dict[str, Any]) -> dict[str, Any]: ``input`` is the JSON form of a :class:`~azure.durable_functions.http.models.DurableHttpRequest` - (``method``, ``uri``, ``content``, ``headers``, ``tokenSource``). The - return value is the JSON form of a + (``method``, ``uri``, ``content``, ``headers``, ``tokenSource``, + ``retryOptions``). Selected response statuses raise an activity failure + before buffering the body, allowing the core durable retry policy to run. + The return value is the JSON form of a :class:`~azure.durable_functions.http.models.DurableHttpResponse`. The parameter and return are declared as ``dict[str, Any]`` for static type @@ -204,16 +206,24 @@ def builtin_http_activity(input: dict[str, Any]) -> dict[str, Any]: req = urllib.request.Request(url=uri, data=data, method=method, headers=headers) try: - with _open_http_request(req) as resp: - status = int(resp.status) - resp_headers = {k: v for k, v in resp.headers.items()} - body = resp.read().decode("utf-8", errors="replace") + response = _open_http_request(req) except urllib.error.HTTPError as e: - # Non-2xx responses (including 202) surface here; capture rather than raise - # so the orchestrator can inspect the status code and headers. - status = int(e.code) - resp_headers = {k: v for k, v in (e.headers or {}).items()} - body = e.read().decode("utf-8", errors="replace") + # urllib raises for HTTP errors and unfollowed redirects. Treat these + # like ordinary responses so excluded statuses still return their body. + response = e + + with response as resp: + status = int(resp.code if isinstance(resp, urllib.error.HTTPError) else resp.status) + retry_options = request.get("retryOptions") + if retry_options is not None: + codes = retry_options.get("statusCodesToRetry") + selected = status in codes if codes else not 200 <= status < 300 + if selected: + # Leaving the context closes the response without reading a + # potentially unbounded body before the durable retry begins. + raise RuntimeError(f"HTTP request failed with status code {status}.") + resp_headers = {k: v for k, v in (resp.headers or {}).items()} + body = resp.read().decode("utf-8", errors="replace") return DurableHttpResponse( status_code=status, headers=resp_headers, content=body).to_json() @@ -287,8 +297,13 @@ def builtin_http_poll_orchestrator(context: Any) -> Generator[Any, Any, DurableH "sub-orchestration.") request: dict[str, Any] = context.get_input() or {} - response: dict[str, Any] = yield context.call_activity( - BUILTIN_HTTP_ACTIVITY_NAME, request) + retry_options = request.get("retryOptions") + if retry_options is None: + initial_task = context.call_activity(BUILTIN_HTTP_ACTIVITY_NAME, request) + else: + initial_task = context.call_activity_with_retry( + BUILTIN_HTTP_ACTIVITY_NAME, HttpRetryOptions.from_json(retry_options), request) + response: dict[str, Any] = yield initial_task # Track the URI of the most recent request so a relative ``Location`` can be # resolved against it. diff --git a/azure-functions-durable/azure/durable_functions/http/models.py b/azure-functions-durable/azure/durable_functions/http/models.py index 15a1fbc9..59264e4c 100644 --- a/azure-functions-durable/azure/durable_functions/http/models.py +++ b/azure-functions-durable/azure/durable_functions/http/models.py @@ -12,11 +12,76 @@ from __future__ import annotations +from datetime import timedelta from typing import Any, Optional, cast +from durabletask.task import RetryPolicy + from ..internal.compat.token_source import TokenSource +class HttpRetryOptions(RetryPolicy): + """Durable failure retries for the initial HTTP request, not Location polls. + + Durations use ``timedelta``, as in :class:`durabletask.task.RetryPolicy`. + ``max_number_of_attempts`` includes the initial attempt. Backoff defaults to + 1, the maximum retry interval to six days, and the retry timeout to unbounded. + An omitted or empty ``status_codes_to_retry`` selects every non-2xx + response; a nonempty list selects exactly those codes (even 202). + Transport and other activity failures are retried regardless of this list. + Exhaustion fails ``call_http`` rather than returning a selected response. + """ + + def __init__(self, *, + first_retry_interval: timedelta, + max_number_of_attempts: int, + backoff_coefficient: float | None = 1.0, + max_retry_interval: timedelta | None = timedelta(days=6), + retry_timeout: timedelta | None = None, + status_codes_to_retry: Optional[list[int]] = None) -> None: + super().__init__( + first_retry_interval=first_retry_interval, + max_number_of_attempts=max_number_of_attempts, + backoff_coefficient=backoff_coefficient, + max_retry_interval=max_retry_interval, + retry_timeout=retry_timeout) + self._status_codes_to_retry = ( + list(status_codes_to_retry) if status_codes_to_retry is not None else None) + + @property + def status_codes_to_retry(self) -> Optional[list[int]]: + """Return a copy of the selected response codes, or ``None``.""" + return (list(self._status_codes_to_retry) + if self._status_codes_to_retry is not None else None) + + def to_json(self) -> dict[str, Any]: + """Snapshot the policy into a JSON-serializable wire payload.""" + millisecond = timedelta(milliseconds=1) + return { + "firstRetryIntervalInMilliseconds": self.first_retry_interval / millisecond, + "maxNumberOfAttempts": self.max_number_of_attempts, + "backoffCoefficient": self.backoff_coefficient, + "maxRetryIntervalInMilliseconds": ( + self.max_retry_interval / millisecond if self.max_retry_interval is not None else None), + "retryTimeoutInMilliseconds": ( + self.retry_timeout / millisecond if self.retry_timeout is not None else None), + "statusCodesToRetry": self.status_codes_to_retry, + } + + @classmethod + def from_json(cls, value: dict[str, Any]) -> "HttpRetryOptions": + """Reconstruct a snapshotted policy from its wire payload.""" + max_interval = value["maxRetryIntervalInMilliseconds"] + timeout = value["retryTimeoutInMilliseconds"] + return cls( + first_retry_interval=timedelta(milliseconds=value["firstRetryIntervalInMilliseconds"]), + max_number_of_attempts=value["maxNumberOfAttempts"], + backoff_coefficient=value["backoffCoefficient"], + max_retry_interval=timedelta(milliseconds=max_interval) if max_interval is not None else None, + retry_timeout=timedelta(milliseconds=timeout) if timeout is not None else None, + status_codes_to_retry=value.get("statusCodesToRetry")) + + class DurableHttpRequest: """Data structure representing a durable HTTP request.""" @@ -25,12 +90,14 @@ def __init__(self, uri: str, content: Optional[str] = None, headers: Optional[dict[str, str]] = None, - token_source: Optional[TokenSource] = None): + token_source: Optional[TokenSource] = None, + retry_options: Optional[HttpRetryOptions] = None): self._method = method self._uri = uri self._content = content self._headers = headers self._token_source = token_source + self._retry_options = retry_options.to_json() if retry_options is not None else None @property def method(self) -> str: @@ -57,6 +124,12 @@ def token_source(self) -> Optional[TokenSource]: """Get the source of the OAuth token to add to the request.""" return self._token_source + @property + def retry_options(self) -> Optional[HttpRetryOptions]: + """Get a copy of the policy captured when this request was created.""" + return (HttpRetryOptions.from_json(self._retry_options) + if self._retry_options is not None else None) + def to_json(self) -> dict[str, Any]: """Convert this request into a JSON-serializable dictionary.""" json_dict: dict[str, Any] = {"method": self._method, "uri": self._uri} @@ -66,6 +139,9 @@ def to_json(self) -> dict[str, Any]: json_dict["headers"] = dict(self._headers) if self._token_source is not None: json_dict["tokenSource"] = self._token_source.to_json() + retry_options = self.retry_options + if retry_options is not None: + json_dict["retryOptions"] = retry_options.to_json() return json_dict diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py b/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py index f93176dd..84561e39 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py @@ -4,7 +4,7 @@ import inspect import json from datetime import UTC, datetime -from typing import Any, Callable, Generator, Optional, cast +from typing import TYPE_CHECKING, Any, Callable, Generator, Optional, cast from uuid import UUID from durabletask import task @@ -15,6 +15,9 @@ from .function_context import FunctionContext from .token_source import TokenSource +if TYPE_CHECKING: + from ...http.models import HttpRetryOptions + class DurableOrchestrationContext: """Azure Functions-style orchestration context (v1-compatible). @@ -215,7 +218,8 @@ def call_http(self, content: Any = None, headers: Optional[dict[str, str]] = None, token_source: Optional[TokenSource] = None, - is_raw_str: bool = False) -> Task[Any]: + is_raw_str: bool = False, + retry_options: Optional["HttpRetryOptions"] = None) -> Task[Any]: """Schedule a durable HTTP call to the specified endpoint. The request is executed by a built-in activity and, if the endpoint @@ -240,6 +244,11 @@ def call_http(self, is_raw_str: bool When ``True``, send ``str`` content as-is instead of JSON-encoding it. Requires ``content`` to be a ``str``. + retry_options: Optional[HttpRetryOptions] + Opt-in durable failure retries for the initial request only. + Selected HTTP statuses and activity failures fail this task after + exhaustion. Location polls do not inherit the policy. Only enable + retries when repeating the request's side effects is safe. """ # Imported lazily to avoid a module import cycle # (http.models -> compat.token_source). @@ -260,7 +269,8 @@ def call_http(self, else: json_content = json.dumps(content) - request = DurableHttpRequest(method, uri, json_content, headers, token_source) + request = DurableHttpRequest( + method, uri, json_content, headers, token_source, retry_options) return self._ctx.call_sub_orchestrator( BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, input=request.to_json(), diff --git a/tests/azure-functions-durable/test_http_retry.py b/tests/azure-functions-durable/test_http_retry.py new file mode 100644 index 00000000..3273bef1 --- /dev/null +++ b/tests/azure-functions-durable/test_http_retry.py @@ -0,0 +1,395 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import base64 +import io +import json +import urllib.error +from datetime import datetime, timedelta, tzinfo +from unittest.mock import MagicMock, patch + +import pytest + +import azure.durable_functions as df +from azure.durable_functions.http.builtin import ( + BUILTIN_HTTP_ACTIVITY_NAME, + BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, + builtin_http_activity, + builtin_http_poll_orchestrator, +) +from azure.durable_functions.internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER +from azure.durable_functions.worker import DurableFunctionsWorker +from durabletask import task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb + + +def _options(**kwargs): + return df.HttpRetryOptions( + first_retry_interval=timedelta(seconds=2), + max_number_of_attempts=3, + **kwargs) + + +def test_http_retry_defaults_and_round_trip(): + policy = _options() + assert isinstance(policy, df.RetryPolicy) + assert policy.backoff_coefficient == 1 + assert policy.max_retry_interval == timedelta(days=6) + assert policy.retry_timeout is None + assert policy.status_codes_to_retry is None + assert policy.to_json() == { + "firstRetryIntervalInMilliseconds": 2000, + "maxNumberOfAttempts": 3, + "backoffCoefficient": 1, + "maxRetryIntervalInMilliseconds": 518400000, + "retryTimeoutInMilliseconds": None, + "statusCodesToRetry": None, + } + assert df.HttpRetryOptions.from_json(json.loads(json.dumps(policy.to_json()))).to_json() == policy.to_json() + + +def test_request_snapshots_policy_and_status_list(): + codes = [429, 503] + policy = _options( + backoff_coefficient=2, + max_retry_interval=timedelta(seconds=30), + retry_timeout=timedelta(minutes=1), + status_codes_to_retry=codes) + request = df.DurableHttpRequest("GET", "https://example.com", retry_options=policy) + expected = policy.to_json() + codes.append(500) + # Even changes to the caller's policy after constructing the request must + # not alter the wire snapshot. + policy._max_number_of_attempts = 100 + policy._status_codes_to_retry.append(404) + payload = request.to_json() + assert payload["retryOptions"] == expected + payload["retryOptions"]["statusCodesToRetry"].append(400) + assert request.to_json()["retryOptions"] == expected + assert request.retry_options.to_json() == expected + assert df.DurableHttpRequest("GET", "https://example.com").retry_options is None + + +@pytest.mark.parametrize("kwargs", [ + {"first_retry_interval": timedelta(seconds=-1), "max_number_of_attempts": 2}, + {"first_retry_interval": timedelta(seconds=1), "max_number_of_attempts": 0}, + {"first_retry_interval": timedelta(seconds=1), "max_number_of_attempts": 2, "backoff_coefficient": 0.5}, + {"first_retry_interval": timedelta(seconds=1), "max_number_of_attempts": 2, "max_retry_interval": timedelta(seconds=-1)}, + {"first_retry_interval": timedelta(seconds=1), "max_number_of_attempts": 2, "retry_timeout": timedelta(seconds=-1)}, +]) +def test_http_retry_policy_preserves_core_validation(kwargs): + with pytest.raises(ValueError): + df.HttpRetryOptions(**kwargs) + + +@pytest.mark.parametrize("codes", [None, [], [202, 429]]) +@pytest.mark.parametrize("status", [199, 200, 202, 204, 302, 400, 429, 503]) +@pytest.mark.parametrize("http_error", [False, True]) +def test_activity_classifies_before_reading_and_closes_response(codes, status, http_error): + selected = status in codes if codes else not 200 <= status < 300 + if http_error: + body = io.BytesIO(b"response") + response = urllib.error.HTTPError( + "https://example.com", status, "response", {"X-Test": "value"}, body) + read = MagicMock(wraps=response.read) + response.read = read + open_options = {"side_effect": response} + else: + response = MagicMock() + response.status = status + response.headers = {"X-Test": "value"} + response.__enter__.return_value = response + read = response.read + read.return_value = b"response" + open_options = {"return_value": response} + if selected: + read.side_effect = AssertionError("Selected error bodies must never be read") + + with patch("azure.durable_functions.http.builtin._open_http_request", **open_options): + request = df.DurableHttpRequest( + "GET", "https://example.com", + retry_options=_options(status_codes_to_retry=codes)).to_json() + if selected: + with pytest.raises(RuntimeError, match=f"status code {status}"): + builtin_http_activity(request) + read.assert_not_called() + else: + assert builtin_http_activity(request) == { + "status_code": status, "headers": {"X-Test": "value"}, "content": "response"} + read.assert_called_once_with() + if http_error: + assert body.closed + else: + response.__exit__.assert_called_once() + + +def test_no_options_preserves_http_error_response(): + body = io.BytesIO(b"error detail") + response = urllib.error.HTTPError( + "https://example.com", 503, "unavailable", {"Retry-After": "9"}, body) + with patch("azure.durable_functions.http.builtin._open_http_request", side_effect=response): + assert builtin_http_activity({"method": "GET", "uri": "https://example.com"}) == { + "status_code": 503, "headers": {"Retry-After": "9"}, "content": "error detail"} + assert body.closed + + +class _History: + """Dispatch through the Functions worker, recreating it for every replay.""" + + def __init__(self, func, input=None, parent=None): + self.func = func + self.now = datetime(2020, 1, 1) + self.events = [] + started = helpers.new_execution_started_event( + func.__name__, "instance", json.dumps(input)) + if parent is not None: + started.executionStarted.parentInstance.orchestrationInstance.instanceId = parent + self.actions = self.dispatch([started]) + + def execute(self, past, new): + request = pb.OrchestratorRequest( + instanceId="instance", pastEvents=past, newEvents=new) + encoded = base64.b64encode(request.SerializeToString()).decode() + output = DurableFunctionsWorker().execute_orchestration_request(self.func, encoded) + response = pb.OrchestratorResponse() + response.ParseFromString(base64.b64decode(output)) + return list(response.actions) + + def dispatch(self, events): + new = [helpers.new_orchestrator_started_event(self.now), *events] + actions = self.execute(self.events, new) + self.events.extend(new) + self.actions = actions + return actions + + def only_action(self, kind): + assert len(self.actions) == 1 + action = self.actions[0] + assert action.HasField(kind), action + return action + + def run_activity(self, *, status=200, headers=None, error=None): + action = self.only_action("scheduleTask") + assert action.scheduleTask.name == BUILTIN_HTTP_ACTIVITY_NAME + request = json.loads(action.scheduleTask.input.value) + self.events.append(helpers.new_task_scheduled_event( + action.id, BUILTIN_HTTP_ACTIVITY_NAME, encoded_input=action.scheduleTask.input.value)) + response = MagicMock() + response.status = status + response.headers = headers or {} + response.read.return_value = b"body" + response.__enter__.return_value = response + with patch("azure.durable_functions.http.builtin._open_http_request", + **({"side_effect": error} if error else {"return_value": response})): + try: + result = builtin_http_activity(request) + except Exception as exc: + completion = helpers.new_task_failed_event(action.id, exc) + else: + completion = helpers.new_task_completed_event(action.id, json.dumps(result)) + self.dispatch([completion]) + return request + + def fire_timer(self, expected_delay): + action = self.only_action("createTimer") + fire_at = action.createTimer.fireAt.ToDatetime() + assert fire_at == self.now + expected_delay + self.events.append(helpers.new_timer_created_event(action.id, fire_at)) + self.now = fire_at + self.dispatch([helpers.new_timer_fired_event(action.id, fire_at)]) + + def assert_replay(self): + # Replay old history with no new completions: the emitted action must + # match even though the real wall clock is years ahead of this history. + replayed = self.execute(self.events, [helpers.new_orchestrator_started_event(self.now)]) + expected = [pb.OrchestratorAction.FromString(action.SerializeToString()) for action in self.actions] + for action in replayed + expected: + if action.HasField("completeOrchestration"): + # The outer traceback includes a different executor line when + # the failure is processed as an old event rather than new. + action.completeOrchestration.failureDetails.ClearField("stackTrace") + assert replayed == expected + + def result(self): + completion = self.only_action("completeOrchestration").completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, completion + return DEFAULT_FUNCTIONS_DATA_CONVERTER.deserialize( + completion.result.value, df.DurableHttpResponse) + + +def _poll_history(options): + request = df.DurableHttpRequest( + "POST", "https://example.com/start", + content="request", headers={"X-Test": "value"}, retry_options=options) + return _History(builtin_http_poll_orchestrator, request.to_json(), parent="caller") + + +@pytest.mark.parametrize("transport_error", [False, True]) +def test_durable_retry_then_success(transport_error): + history = _poll_history(_options(status_codes_to_retry=[503])) + request = history.run_activity( + status=503, error=urllib.error.URLError("disconnected") if transport_error else None) + history.assert_replay() + history.fire_timer(timedelta(seconds=2)) + assert history.run_activity() == request + assert history.result().status_code == 200 + assert history.result().content == "body" + + +def test_default_backoff_is_constant_and_timeout_is_unbounded(): + history = _poll_history(_options()) + history.now += timedelta(days=365) + for _ in range(2): + history.run_activity(status=503) + history.fire_timer(timedelta(seconds=2)) + history.run_activity() + assert history.result().status_code == 200 + + +@pytest.mark.parametrize("remove_cap", [False, True]) +def test_default_six_day_cap_and_explicit_unbounded_interval(remove_cap): + kwargs = {"max_retry_interval": None} if remove_cap else {} + history = _poll_history(df.HttpRetryOptions( + first_retry_interval=timedelta(days=8), max_number_of_attempts=2, **kwargs)) + history.run_activity(status=503) + # Core splits long durable timers into three-day chunks. + history.fire_timer(timedelta(days=3)) + history.fire_timer(timedelta(days=3)) + if remove_cap: + history.fire_timer(timedelta(days=2)) + history.run_activity() + assert history.result().status_code == 200 + + +def test_durable_retries_backoff_cap_and_exhaustion(): + history = _poll_history(df.HttpRetryOptions( + first_retry_interval=timedelta(seconds=2), + max_number_of_attempts=4, backoff_coefficient=3, + max_retry_interval=timedelta(seconds=5))) + for delay in (2, 5, 5): + history.run_activity(status=503) + history.assert_replay() + history.fire_timer(timedelta(seconds=delay)) + history.run_activity(status=503) + history.assert_replay() + completion = history.only_action("completeOrchestration").completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "status code 503" in str(completion.failureDetails) + + +@pytest.mark.parametrize("wall_clock_year", [2019, 2030]) +@pytest.mark.parametrize("elapsed", [9, 10, 11]) +def test_http_timeout_uses_replayed_orchestration_clock(monkeypatch, wall_clock_year, elapsed): + class FrozenDateTime(datetime): + @classmethod + def now(cls, tz: tzinfo | None = None) -> datetime: + instant = datetime(wall_clock_year, 1, 1) + return instant.replace(tzinfo=tz) + + monkeypatch.setattr(task, "datetime", FrozenDateTime) + history = _poll_history(_options(retry_timeout=timedelta(seconds=10))) + history.now += timedelta(seconds=elapsed) + history.run_activity(status=503) + history.assert_replay() + if elapsed >= 10: + completion = history.only_action("completeOrchestration").completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + else: + # Core checks elapsed time when a failure arrives, not whether the next + # delay will cross the timeout. The retry at t=11 is still scheduled. + history.fire_timer(timedelta(seconds=2)) + history.run_activity() + assert history.result().status_code == 200 + + +@pytest.mark.parametrize("codes", [[429], []]) +def test_initial_202_polls_without_retrying_later_errors(codes): + history = _poll_history(_options(status_codes_to_retry=codes)) + history.run_activity(status=202, headers={"Location": "/status", "Retry-After": "3"}) + history.fire_timer(timedelta(seconds=3)) + poll = history.run_activity(status=503) + assert poll == {"method": "GET", "uri": "https://example.com/status", "headers": {"X-Test": "value"}} + assert history.result().status_code == 503 + + +def test_selected_202_retries_instead_of_polling(): + history = _poll_history(_options(status_codes_to_retry=[202])) + original = history.run_activity( + status=202, headers={"Location": "/status", "Retry-After": "9"}) + history.fire_timer(timedelta(seconds=2)) + assert history.run_activity() == original + assert history.result().status_code == 200 + + +def test_retry_then_multiple_location_polls_do_not_inherit_policy(): + history = _poll_history(_options(status_codes_to_retry=[503])) + original = history.run_activity(status=503) + history.fire_timer(timedelta(seconds=2)) + assert history.run_activity(status=202, headers={"Location": "/first"}) == original + for location in ("/second", "/third"): + history.fire_timer(timedelta(seconds=1)) + poll = history.run_activity(status=202, headers={"Location": location}) + assert "retryOptions" not in poll + history.fire_timer(timedelta(seconds=1)) + assert "retryOptions" not in history.run_activity(status=503) + assert history.result().status_code == 503 + + +def test_location_poll_transport_failure_is_not_retried(): + history = _poll_history(_options()) + history.run_activity(status=202, headers={"Location": "/status"}) + history.fire_timer(timedelta(seconds=1)) + history.run_activity(error=urllib.error.URLError("poll disconnected")) + completion = history.only_action("completeOrchestration").completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + + +def test_transport_failure_without_options_is_not_retried(): + history = _poll_history(None) + history.run_activity(error=urllib.error.URLError("disconnected")) + completion = history.only_action("completeOrchestration").completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + + +@pytest.mark.parametrize("with_options", [False, True]) +def test_excluded_error_and_no_options_return_full_response(with_options): + history = _poll_history(_options(status_codes_to_retry=[429]) if with_options else None) + history.run_activity(status=503, headers={"X-Error": "detail"}) + response = history.result() + assert response.to_json() == {"status_code": 503, "headers": {"X-Error": "detail"}, "content": "body"} + + +@pytest.mark.parametrize("succeed", [False, True]) +def test_call_http_schedules_snapshot_and_observes_child_result(succeed): + codes = [503] + policy = df.HttpRetryOptions( + first_retry_interval=timedelta(seconds=2), + max_number_of_attempts=1, status_codes_to_retry=codes) + + def caller(context): + pending = context.call_http("GET", "https://example.com", retry_options=policy) + codes.append(404) + return (yield pending) + + parent = _History(caller) + action = parent.only_action("createSubOrchestration") + assert action.createSubOrchestration.name == BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME + payload = json.loads(action.createSubOrchestration.input.value) + assert payload["retryOptions"]["statusCodesToRetry"] == [503] + parent.events.append(helpers.new_sub_orchestration_created_event( + action.id, BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, "child", + encoded_input=action.createSubOrchestration.input.value)) + child = _History(builtin_http_poll_orchestrator, payload, parent="instance") + child.run_activity(status=200 if succeed else 503) + completion = child.only_action("completeOrchestration").completeOrchestration + if succeed: + parent.dispatch([helpers.new_sub_orchestration_completed_event(action.id, completion.result.value)]) + assert parent.result().status_code == 200 + else: + failure = helpers.new_sub_orchestration_failed_event(action.id, RuntimeError("placeholder")) + failure.subOrchestrationInstanceFailed.failureDetails.CopyFrom(completion.failureDetails) + parent.dispatch([failure]) + completion = parent.only_action("completeOrchestration").completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "status code 503" in str(completion.failureDetails)