Add ChainRetryPolicy to compose retry policies in order

[kaxil]

Follow-up to AIP-105 (#65474). A task's retry_policy can now be a sequence of policies.

Summary

ChainRetryPolicy([...]) consults policies in order with the original task exception. The first RETRY or FAIL wins. A DEFAULT means the policy has nothing to add and the next one is asked; when every policy returns DEFAULT the task's own retries and retry_delay apply, as they do today.

Two orders become expressible that were not:

• Rules first. Known failures are settled by an ExceptionRetryPolicy before a slower or costlier policy (one that calls a model, for instance) is consulted, so it never sees them.
• A floor behind a fallible policy. A policy that can fail on its own gets deterministic rules after it, without carrying its own copy of a fallback mechanism.

from airflow.sdk import ChainRetryPolicy, ExceptionRetryPolicy, RetryAction, RetryRule

retry_policy = ChainRetryPolicy([
    ExceptionRetryPolicy(rules=[RetryRule(exception=PermissionError, action=RetryAction.FAIL)]),
    HTTPStatusRetryPolicy(),  # any RetryPolicy, from any package
    ExceptionRetryPolicy(rules=[RetryRule(exception=ConnectionError, retry_delay=timedelta(seconds=30))]),
])

Design rationale

DEFAULT means "next policy", not "stop". The SDK documents RetryAction.DEFAULT as "fall through to standard retry logic". Inside a chain that is what you get once the chain is exhausted, so DEFAULT is the natural abstention signal and no new enum value or wrapper type is needed. Two consequences are documented: a RetryRule with action=DEFAULT passes control on rather than ending the chain, and an ExceptionRetryPolicy with default=RetryAction.FAIL ends the chain wherever it sits.

A broken policy does not take the floor down with it. An ordinary exception from a policy, or a return value that is not a RetryDecision, is logged and treated as DEFAULT, so the rules after it still run. BaseException propagates, so a cancelled or terminated task behaves as it does now.

The reason survives truncation. The winning decision's reason names the deciding policy first and the earlier verdicts after it: HTTPStatusRetryPolicy: HTTP 404 (after ExceptionRetryPolicy: no decision). The worker stores retry_reason truncated to 500 characters, so the verdict comes first and the trail is what gets cut. The worker stores the reason only on a RETRY; on FAIL and DEFAULT it appears in the task log, and the docs say so.

Why the SDK. The class composes RetryPolicy, RetryDecision and RetryAction, all SDK types, and a chain of two rule policies or a user's own policy followed by rules is a use with no provider involved. Third-party RetryPolicy subclasses compose without depending on anything else. The constructor takes a sequence like ExceptionRetryPolicy(rules=[...]) rather than *policies, so a chain built from configuration is a plain list. Members are validated as RetryPolicy instances at construction, which is stricter than the operator's duck-typed retry_policy argument, so a typo fails at parse time rather than at the first task failure.

No serialization or migration. retry_policy is already a flag-only serialized field and the worker re-parses the Dag, so a chain rides along like any other policy object. ChainRetryPolicy does not implement serialize(); nothing in Airflow calls it, and the base docstring now says so.

Live run

Airflow from this branch under Breeze, with LLMRetryPolicy from the common.ai provider on main as the middle rung and anthropic:claude-sonnet-5 behind it. One chain shape for every task: a rule set that fails on PermissionError, then the model policy, then a rule set that retries ConnectionError after 15 seconds. Each task raises a different failure so a different rung decides. retries=1.

Task	Failure raised	Decided by	Decision and reason
rules_first__403	PermissionError	first rule set; the model is never called	FAIL, ExceptionRetryPolicy: never retry a 403
model_decides__429	RuntimeError("429 Too Many Requests ... Retry-After: 45")	the model	RETRY 60s, LLMRetryPolicy: category=rate_limit ... action=retry delay=60s (after ExceptionRetryPolicy: no decision)
model_decides__bad_data	ValueError("unexpected column ...")	the model	FAIL, LLMRetryPolicy: category=data ... action=fail (after ExceptionRetryPolicy: no decision)
floor__model_unreachable	ConnectionError, model connection id does not exist	the floor	RETRY 15s, ExceptionRetryPolicy: floor (after ExceptionRetryPolicy: no decision; LLMRetryPolicy: LLM classification not applied (model_error); task retry settings apply)

The last row is the case the chain exists for: the model policy abstains with a DEFAULT that carries its own explanation, and the rules after it still decide. Task logs, filtered to the decision line:

Docs

A "Chaining policies" section on the Tasks page, marked versionadded 3.4.0, with an example from example_retry_policy.py. While there, the custom-policy example on the same page called RetryDecision.retry(retry_delay=...); the parameter is delay, so copying the page raised TypeError on the first 429. Fixed.