Merge main into codec deprecation review

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: andystaples

CHANGELOG.md

@@ -9,6 +9,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 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.
 - Type discovery now recognizes `dict` / `Mapping` and `tuple` containers wrapping
 reconstructable types (such as dataclasses and `from_json()`-capable types) via
 `JsonDataConverter.can_reconstruct()`. Inbound activity and orchestrator inputs as

azure-functions-durable/CHANGELOG.md

@@ -7,8 +7,23 @@ 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
+`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.
 - Async and synchronous Functions-bound clients now propagate the calling
 function's invocation ID in gRPC metadata, enabling host client-operation log
 correlation even when synchronous clients are cached across invocations.

azure-functions-durable/README.md

@@ -83,6 +83,79 @@ These compatibility paths remain supported.
 > Changing typing modes can change values returned during orchestration replay.
 > Validate existing payloads and drain affected orchestrations before switching.
 
+## 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

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",

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.

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
 
 

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

durabletask-azuremanaged/CHANGELOG.md

@@ -9,6 +9,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 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 `durabletask` SDK update, type discovery now recognizes
 `dict`, `Mapping`, and `tuple` container annotations wrapping reconstructable types,
 automatically reconstructing recorded inputs and inferred activity results into typed model

durabletask/internal/orchestration_entity_context.py

@@ -28,7 +28,8 @@ def is_inside_critical_section(self) -> bool:
 
     def get_available_entities(self) -> Generator[EntityInstanceId, None, None]:
         if self.is_inside_critical_section:
-            for available_lock in self.available_locks:
+            # Do not forward send, throw, or close to the lock iterator.
+            for available_lock in self.available_locks:  # noqa: UP028
                 yield available_lock
 
     def validate_suborchestration_transition(self) -> tuple[bool, str]: