Retry¶
taskq.retry is the retry engine: the RetryPolicy model actors declare,
the classifier hooks that decide per-failure outcomes, the backoff
computations, and the lifecycle hooks. The Retries guide
walks the behavior; this page is the module index.
The names the guides don't spell out¶
Eight public names in this module are easy to miss because the guides talk about their EFFECTS without naming them. All eight render under the module directive below:
| Name | Role |
|---|---|
compute_backoff |
The single delay computation every retry path shares: curve (exponential/linear/fixed) → effective cap (min(policy.cap, max_retry_backoff)) → jitter drawn from a band fitted under the cap (deliberately NOT Full Jitter, which collapses toward zero on attempt 1). |
invoke_on_success / invoke_on_cancel / invoke_on_retry_exhausted |
The three @actor lifecycle-hook invokers: sync or async hooks, run under a timeout, a hook failure is logged and never changes the already-decided outcome. |
safe_mark_failed_or_retry |
Wraps the backend's mark_failed_or_retry, catching WorkerOwnershipMismatch: returns the persisted JobRow on success, None on ownership mismatch or a fenced-out epoch, which signals the caller to skip the on_retry_exhausted hook. |
time_budget_as_interval |
Renders an indefinite policy's time_budget as the Postgres interval the enqueue path passes so schedule_to_close = clock_timestamp() + time_budget is computed at insert time. |
MAX_ATTEMPTS_SMALLINT_CEILING |
32767 — the jobs.max_attempts column's smallint domain ceiling. |
MAX_ENQUEUABLE_MAX_ATTEMPTS |
32766 — one below the ceiling, the largest max_attempts a fresh enqueue or policy may carry (the defensive headroom a reclaim needs to add one). |
Module reference¶
retry ¶
Retry policy data carriers, decision types, backoff computation, and consumer-loop adapter wiring.
The data-model layer (RetryPolicy, Retry, Fail, RetryDecision,
JobRetryState, compute_backoff, RetryClassifier) is pure: no I/O, no
clock reads, no backend imports. The adapter layer (OnRetryExhausted,
OnSuccess, OnCancel, ActorConfigLike, decide_after_failure,
invoke_on_retry_exhausted, invoke_on_success, invoke_on_cancel,
safe_mark_failed_or_retry) wires the classifier to the consumer loop
and is permitted backend imports. The built-in classifier hooks
(rate_limit_aware_classifier, failure_taxonomy_classifier) sit between
the two: they are pure functions of (exception, attempt) except for
one bounded clock read — HTTP-date Retry-After parsing needs now
to turn a date into a delay; the clock is injectable (now=) and
never touched on a claimed-without-header or miss path.
MAX_ATTEMPTS_SMALLINT_CEILING
module-attribute
¶
The jobs.max_attempts column's domain ceiling.
MAX_ENQUEUABLE_MAX_ATTEMPTS
module-attribute
¶
Largest max_attempts a fresh enqueue or policy may carry.
One below the column ceiling, retained as a defensive margin: a row parked at exactly the ceiling has no headroom for any statement that needs to add one to a max_attempts-derived value. Rows can still legally REACH the ceiling, a snooze arm's saturating increment parks a snoozed job there, which is why the retry layer clamps row-stored values back into this bound before reconstructing a policy.
__all__
module-attribute
¶
__all__ = [
"MAX_ATTEMPTS_SMALLINT_CEILING",
"MAX_ENQUEUABLE_MAX_ATTEMPTS",
"ActorConfigLike",
"Fail",
"JobRetryState",
"OnCancel",
"OnRetryExhausted",
"OnSuccess",
"Retry",
"RetryClassifier",
"RetryClassifierHook",
"RetryDecision",
"RetryKind",
"RetryOverride",
"RetryPolicy",
"apply_jitter",
"compose_retry_classifiers",
"compute_backoff",
"decide_after_failure",
"failure_taxonomy_classifier",
"invoke_on_cancel",
"invoke_on_retry_exhausted",
"invoke_on_success",
"make_rate_limit_aware_classifier",
"rate_limit_aware_classifier",
"safe_mark_failed_or_retry",
"time_budget_as_interval",
]
DEFAULT_TRANSIENT_STATUSES
module-attribute
¶
DEFAULT_NON_RETRYABLE_STATUSES
module-attribute
¶
DEFAULT_NON_RETRYABLE_STATUSES: Final[frozenset[int]] = (
frozenset(
status
for status in range(400, 500)
if status not in {408, 425, 429}
)
)
DEFAULT_TRANSIENT_EXCEPTION_NAMES
module-attribute
¶
DEFAULT_TRANSIENT_EXCEPTION_NAMES: Final[frozenset[str]] = (
frozenset(
{
"Timeout",
"TimeoutError",
"ConnectError",
"ConnectTimeout",
"ReadTimeout",
"WriteTimeout",
"PoolTimeout",
"ServerTimeoutError",
"ReadTimeoutError",
"ConnectTimeoutError",
"ConnectionError",
"ConnectionResetError",
"ConnectionRefusedError",
"ConnectionAbortedError",
}
)
)
DEFAULT_EXCLUDED_EXCEPTION_NAMES
module-attribute
¶
RetryKind ¶
Closed set of retry tiers.
Why Literal and not an Enum: serialization round-trips through
model_dump(mode="json") produce plain strings without
use_enum_values configuration; pyright exhaustive matching works
identically for either; no .value access required at call sites.
RetryClassifierHook ¶
Optional per-actor hook for exception-instance-level retry classification.
non_retryable_exceptions and the built-in :class:PayloadValidationError
check classify by exception type alone. Some integrations need finer
granularity, a single exception type (e.g. an HTTP client's status-code
error) that should retry indefinitely on a 429, fail immediately on a 404,
and use a bounded transient budget on a 5xx, or a server-provided
Retry-After value that should drive the actual backoff delay. Register
one via @actor(retry_classifier=...).
Invoked with (exception, attempt) for every exception that survives
the adapter's unconditional-Fail checks: the actor's
non_retryable_exceptions and the built-in PayloadValidationError,
pydantic ValidationError, ResultTooLarge, and
UnencodableValue classes. Return
None to fall back to the actor's static RetryPolicy unchanged, or a
:class:RetryOverride to refine kind and/or delay for this specific
occurrence. Exceptions raised by the hook itself are caught and logged by
:func:decide_after_failure; classification falls back to the static
policy in that case, a broken hook can never crash the retry pipeline.
The returned override's runtime shape is verified too: an honestly
constructed :class:RetryOverride always passes, but a
model_construct bypass of pydantic validation (a delay that is
not a timedelta, a kind outside the :data:RetryKind
vocabulary) is logged under retry-classifier-hook-invalid-return and
falls back to the declared policy like any other broken hook — a
malformed override must never raise inside the decision arithmetic or
silently govern as indefinite.
OnRetryExhausted ¶
Hook fired when a job exhausts its retry budget.
Why JobRow (not generic JobRow[P]): the hook is dispatched from
the consumer loop, which knows only the raw JobRow with
payload: dict[str, object]. Making the hook generic over P
would require the consumer to track the original ActorRef for every
in-flight job, possible, but it propagates type parameters into the
registry for negligible benefit. Hooks that need a typed payload
re-validate via actor_ref.payload_type.model_validate(job_row.payload).
This is the documented payload-erasure boundary; see
the documented payload-erasure boundary.
OnSuccess ¶
Hook fired when a job succeeds. Receives (job_row, result).
Why object for the result type (not Any or generic R): the
hook is dispatched from the consumer loop, which erases the actor's
return type to object. This mirrors the non-generic
:data:OnRetryExhausted at the same payload-erasure boundary. Hooks
that need a typed result re-validate via the actor's
result_adapter.
OnCancel ¶
Hook fired when a job ends cancelled. Receives the terminal JobRow.
No result argument, unlike :data:OnSuccess: an actor that abandoned
its unit of work produced none. The row is the terminal one, so a hook
reading status sees cancelled.
The hook fires only for a job that reached a worker and was cancelled
while running. A job cancelled while still pending or scheduled
never enters a worker, so no hook of any kind can run for it, that
bookkeeping stays with whoever issued the cancel.
RetryPolicy ¶
Bases: BaseModel
Policy controlling retry behaviour for an actor.
backoff
class-attribute
instance-attribute
¶
Retry ¶
Bases: BaseModel
Retry decision: reschedule the job after retry_delay.
The delay, not a computed timestamp, is the decision payload: the
backend derives scheduled_at = now() + retry_delay and the
scheduled/pending status from its own clock (single arbiter, immune to
app↔DB clock skew). The delay never falls below
:data:~taskq.constants.MIN_DEFERRAL_INTERVAL, the requeue-rate
floor the deferral arms already apply at their writes, so a
degenerate curve (base=timedelta(0), where 0 * 2**k == 0 at
every rung) or a zero Retry-After override cannot turn the
failure cycle into a claim/run/fail round trip monopolising a worker
slot with no period and, for an indefinite kind, no attempt
ceiling.
Fail ¶
JobRetryState ¶
Bases: NamedTuple
Projection of JobRow columns consumed by the retry classifier.
schedule_to_close is not an input to classification, the
SQL deadline guard in mark_failed_or_retry is the single deadline
arbiter (one arbiter per predicate; see docs/architecture.md §Clock
Domains). The field is retained on the projection for observability
and hook consumers. start_to_close is reserved for per-attempt
timeout enforcement at the consumer level (asyncio.wait_for); not used
by the classifier.
RetryOverride ¶
Bases: BaseModel
Per-exception override returned by an actor's retry_classifier hook.
Both fields are optional; None means "use the actor's static
RetryPolicy/computed backoff for this field." Returning a
RetryOverride with only kind set lets one exception type
branch into different retry behaviour per occurrence, e.g. an HTTP
429 response goes indefinite while a 404 response on the same
exception type goes non_retryable. Returning one with only
delay set lets the actor honour a server-provided retry-after
duration instead of the policy's computed exponential/linear
backoff, while max_retry_backoff still applies as a safety
ceiling so a malicious or malformed header cannot strand a job. The
delay is honored EXACTLY — no jitter draw: an explicit override
delay is an explicit direction, and the library never mutates a
value the classifier specified (jitter spreads only the computed
curve). A fleet that wants spread on a hint applies it in its own
classifier (:func:apply_jitter).
A delay schedules the next attempt; it does not extend the job's
budget, in either dimension. It does not spare the attempt, the
retry still counts against max_attempts unless kind is also
set, or the actor raises RetryAfter(consume_budget=False). And it
does not move the job's schedule_to_close: an upstream under
pressure will happily hand back an hour, and if the delay puts the
next attempt past that deadline the deadline sweep fails the job
terminally before any worker looks at it. max_retry_backoff does
not protect against this, the two bounds mean different things, one
stopping a single absurd delay and the other stating how long the
caller still wants the result, and where they disagree
schedule-to-close wins.
delay
class-attribute
instance-attribute
¶
delay: timedelta | None = Field(
default=None,
description="When to retry, not whether the attempt is charged: a delay alone still spends one attempt of the job's budget, so a classifier returning only a delay against a sustained outage exhausts max_attempts on schedule. Pair it with kind='indefinite' to keep retrying, or raise RetryAfter(delay, consume_budget=False) from the actor body for a known-duration wait that spends no budget. Honored EXACTLY — the library never mutates a value the classifier specified (jitter spreads only the computed curve); to spread a hint across a fleet, apply apply_jitter() to it in your own classifier. Clamped by max_retry_backoff and floored at MIN_DEFERRAL_INTERVAL — an explicit delay=timedelta(0) is honored as 'as fast as the deferral floor allows' (a 1s scheduled requeue), never a pending-immediate one — but NOT reconciled with schedule_to_close, a delay landing past that deadline fails the job terminally through the deadline path.",
)
RetryClassifier ¶
Pure classifier that maps an exception + policy to a RetryDecision.
The classifier decides retry-kind and backoff only, it is
deliberately NOT a deadline arbiter. schedule_to_close is
arbitrated by the SQL guard in mark_failed_or_retry (single
arbiter, the backend's clock); a Python-side pre-check computed from
the worker's clock would disagree with it under app↔DB skew and kill
jobs early (or rubber-stamp them).
classify
staticmethod
¶
classify(
policy: RetryPolicy,
non_retryable_exceptions: tuple[
type[BaseException], ...
],
exception: BaseException,
attempt: int,
*,
max_retry_backoff: timedelta = DEFAULT_MAX_RETRY_BACKOFF,
override: RetryOverride | None = None,
) -> RetryDecision
Source code in src/taskq/retry.py
1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 | |
ActorConfigLike ¶
Bases: Protocol
Structural shape the adapter needs from the per-actor registration record. The eventual concrete ActorConfig class will satisfy this protocol structurally.
Attributes are declared as read-only properties because the concrete ActorConfig will be a frozen Pydantic model; writable Protocol attributes would not be satisfiable by any immutable class.
apply_jitter ¶
Spread an externally supplied advisory delay with jitter.
Same formula as :func:compute_backoff (see it for why
multiplicative-symmetric, not Full Jitter), for delays whose raw value
comes from outside the policy's own backoff curve, an admission
denial's retry_after. Every fielder of the same raw hint in one
round would otherwise re-attempt in lockstep (same token deficit, same
lease horizon), so the hint is spread across the band exactly as
failure backoff is; the knob stays the policy's own jitter, so
jitter=0.0 (uniform(1, 1)) is the identity and deterministic
suites stay deterministic.
The result is advisory timing only: any downstream floor (the snooze
arm's MIN_DEFERRAL_INTERVAL) still applies to the returned value.
Source code in src/taskq/retry.py
compute_backoff ¶
compute_backoff(
policy: RetryPolicy,
attempt: int,
rng: Random | None = None,
*,
max_retry_backoff: timedelta = DEFAULT_MAX_RETRY_BACKOFF,
) -> timedelta
Compute the backoff delay for a given attempt (1-indexed).
multiplicative-symmetric jitter ,
delay = raw * rng.uniform(1 - jitter, 1 + jitter)
with the band fitted under the cap before the draw (see
:func:_capped_jitter_band), so a saturated row spreads over
[cap·(1-j), cap] instead of stacking on the cap.
This is NOT Full Jitter (uniform(0, raw)) because Full Jitter
collapses toward zero on attempt 1, causing thundering-herd
retries. See Marc Brooker, "Exponential Backoff And Jitter",
AWS Architecture Blog, and the AWS SDKs' published jitter debate.
max_retry_backoff is the global ceiling applied after
policy.cap, i.e. effective_cap = min(policy.cap, max_retry_backoff).
The global cap prevents a misconfigured per-actor
RetryPolicy(cap=timedelta(days=365)) from stranding jobs for a year
with no operator visibility, a defensive layer beyond the policy's own cap.
Callers that hold WorkerSettings should pass
settings.max_retry_backoff; the default 24 h matches
WorkerSettings.max_retry_backoff.
Source code in src/taskq/retry.py
compose_retry_classifiers ¶
Compose classifier hooks into one, first-override-wins.
Each classifier is invoked in registration order with
(exception, attempt). The first classifier that returns a
:class:RetryOverride decides the outcome and later classifiers are
not consulted for that exception; a None falls through to the
next classifier; when every classifier returns None the
composition returns None so the actor's declared RetryPolicy
governs, exactly as a single hook returning None does.
Order matters: put the most specific classifier first. A composed
classifier registered via @actor(retry_classifier=...) sits at
the same seam as a single hook, so the adapter's own precedence
contract is unchanged, non_retryable_exceptions and the built-in
unconditional-Fail classes still win over the whole composition.
Per-classifier isolation (the single-classifier contract, composed):
a classifier that raises is logged at WARNING and skipped, and
composition continues with the next classifier, never propagating;
likewise a classifier returning something that is not a
:class:RetryOverride nor None — or a RetryOverride whose
fields bypassed pydantic validation (model_construct), which the
decision path could not use safely — is logged and skipped. The
adapter's carve-out applies per classifier: KeyboardInterrupt and
asyncio.CancelledError are never a classifier outcome and
propagate raw. A classifier that raises is skipped rather than
aborting the composition because the classifiers are independent
observers of the same exception, one of them being broken says
nothing about the others, and falling back to the declared policy on
the first broken one would silently discard the overrides the healthy
classifiers were registered to provide.
compose_retry_classifiers() with no arguments returns a hook that
always returns None (the declared policy governs), so call sites
can compose a possibly-empty list without a special case.
Source code in src/taskq/retry.py
613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 | |
rate_limit_aware_classifier ¶
rate_limit_aware_classifier(
exception: BaseException,
attempt: int,
*,
now: datetime | None = None,
) -> RetryOverride | None
Built-in classifier for application-level rate-limit raises.
TaskQ's own rate limiting denies at claim time (the job is
rescheduled, no attempt is charged) and never reaches this
classifier; what this classifier recognizes is the moment an
application's outgoing call is rate-limited and the actor lets the
signal propagate, the shape the cbre-pfc migration audit (W3.H/F1)
found consumers hand-rolling: without it a declared transient
policy burns one attempt of max_attempts per 429 and the job dies
after a few tens of seconds (base=5s exponential, max_attempts=3: two
delays, 5s + 10s — the 20s rung is never reached, the third failure is
terminal and schedules no delay), when the operator's intent for a rate
limit is "wait it out", the unbounded-in-attempts behaviour only an
indefinite kind provides.
Recognized signals, in check order:
- TaskQ's own :class:
~taskq.exceptions.ReservationUnavailableraised withsource="rate_limit": a shared limiter's denial surfacing in the actor's frame. Asource="reservation"denial is a concurrency-slot condition, not a rate limit, and is NOT recognized. - Common HTTP 429 duck-types, no import required: an exception carrying
.response.status_code == 429(httpx / httpx2 / requests'HTTPStatusError/HTTPErrorshapes),.response.status == 429or a bare.status == 429on the exception (aiohttp'sClientResponseError), or a bare.status_code == 429. - An exception whose class is literally named
RateLimitError(the openai/anthropic-style SDK shape), even when it carries no HTTP status attribute.
On a claimed signal the classifier sniffs the server's retry hint —
the retry-after and x-retry-after headers, case-insensitive,
on exception.response.headers (httpx/requests) or a bare
exception.headers (aiohttp) — and parses it as a decimal-fraction
seconds value ("120", "0.5") or an HTTP-date (see
:func:_parse_retry_after).
Everything else returns None: the declared policy governs, so a
500 keeps its bounded transient budget and a 404 its non-retryable
verdict even when this classifier is composed in. Header sniffing
runs only on a claimed signal: the miss path (the path every ordinary
exception takes) does no getattr chains beyond the status
duck-typing and allocates nothing (pinned by
tests/perf/test_retry_classifier_miss_benchmark.py and the
tracemalloc pin in tests/test_retry_after_header_sniffing.py).
The override and what bounds it (the full table lives in
docs/guides/retries.md §5):
- a parsed hint →
RetryOverride(kind="indefinite", delay=hint): the hint flows through as the override delay VERBATIM (honored EXACTLY — no jitter draw: an explicit delay is an explicit direction, the library never mutates a value the classifier specified) and meets the documented bounds on the decision path — clamped tomax_retry_backoffby :meth:RetryClassifier._retry_decision(verified, not assumed — the same path anyRetryOverride.delaytakes; the clamp reads the verbatim raw value), then floored atMIN_DEFERRAL_INTERVAL. A finite hint beyond the ceiling CLAMPS to it — the ceiling's documented job ("a malicious or malformed header cannot strand a job"), the operator's knob doing the bounding. The parse treats only garbage shapes as garbage (zero, negative, or no grammar match — the closed decimal-fraction grammar is a deliberate tightening of the formerint()parse's accidents); an oversized-but-finite hint is the ceiling's input, not garbage. The classifier adds no ceiling of its own. - a claimed signal with no usable hint (no header, or the
curve-fallback garbage cases: zero / negative / unparsable) →
RetryOverride(kind="indefinite")with nodelay: the declared policy's backoff curve (with jitter, cap, andmax_retry_backoff) keeps computing when to retry, and the job'sschedule_to_close(time_budgetfor an indefinite policy) stays the single stopping condition, the same bound everyindefinitejob has. Kind only, because the trap being fixed is the attempt budget, not the curve.
Hazard: that stopping condition must exist. A transient actor is
never stamped with a schedule_to_close (time_budget is only
honored for an indefinite-declared policy; see
:func:time_budget_as_interval), so composing this classifier into a
transient actor makes a sustained 429 storm retry the job forever
— no attempt ceiling and no deadline. The parsed delay does not
change this: it schedules when the next attempt lands, it does not
move the job's schedule_to_close, and a delay landing past that
deadline still fails the job terminally in the deadline path. Give
the actor a stopping condition: declare it kind="indefinite" with
a time_budget, put a domain classifier before this one that
bounds the 429s, pass a per-enqueue schedule_to_close, or use the
factory's bounded mode —
:func:make_rate_limit_aware_classifier(claim_kind="transient")
keeps max_attempts as the stopper while still honoring the
server's hint.
This built-in is the claim_kind="indefinite" instance of
:func:make_rate_limit_aware_classifier, the factory for consumers
whose intent differs: claim_kind="transient" for 429 → BOUNDED
retry (the server's hint honored, max_attempts the stopper), and
claim_kind=None for a never-claiming identity that slots into a
composition unchanged.
Compose it after your domain-specific classifiers:
compose_retry_classifiers(my_domain_classifier, rate_limit_aware_classifier);
a more specific classifier registered before it wins by composition
order, and one registered after it never sees a 429 this classifier
claimed.
now injects the clock for HTTP-date parsing (tests, or a caller
with its own clock domain); the default reads
datetime.now(UTC) — this is the module's one clock read,
never touched on the miss path.
Source code in src/taskq/retry.py
874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 | |
make_rate_limit_aware_classifier ¶
make_rate_limit_aware_classifier(
claim_kind: Literal["indefinite", "transient"]
| None = "indefinite",
) -> RetryClassifierHook
Build a rate-limit classifier whose CLAIM is configurable — the
factory behind :func:rate_limit_aware_classifier (which is exactly
make_rate_limit_aware_classifier() at the default, kept as a
module-level name so existing registrations and pins are untouched).
The recognition surface is identical for every mode — TaskQ's own
:class:~taskq.exceptions.ReservationUnavailable with
source="rate_limit", the HTTP 429 duck-types, and the
RateLimitError-named shape — and so are the hint parsing, the
garbage rules, and the bounds (the operator's max_retry_backoff
ceiling and the MIN_DEFERRAL_INTERVAL floor, applied by the
decision path to every override delay). Only the override's KIND
changes; see the mode table:
================ ============================================= ============================================= =========================================
claim_kind Override on a claimed signal What bounds it The haunt hazard (per mode)
================ ============================================= ============================================= =========================================
"indefinite" RetryOverride(kind="indefinite"), schedule_to_close — the delay schedules A transient actor has no
delay=hint when the server sent one when, never whether: a deadline is schedule_to_close (time_budget is
REQUIRED. Declare the actor only honored for an indefinite-declared
kind="indefinite" with a time_budget, policy), so a sustained 429 storm retries
or pass schedule_to_close= per enqueue. the job FOREVER — no attempt ceiling, no
deadline. This mode's intent is "wait it
out"; a stopping condition must exist.
"transient" RetryOverride(kind="transient"), max_attempts — the budget STAYS the None: the attempt budget terminates the
delay=hint when the server sent one stopper; the hint (when present) sets storm even with no deadline. The intent
when within it. The bounded AND this mode exists for: "429 → bounded
server-honoring shape. retry, and honor the server's when".
None never claims — returns None for every the declared policy governs everything None: the classifier contributes no claims
input (the identity for composition) to compose over; it only reserves the
recognition surface for future modes.
================ ============================================= ============================================= =========================================
Validation is at construction: a claim_kind outside the three
modes raises ValueError here — fail loud at build time, never a
silently-misclaiming classifier at override time. "non_retryable"
is deliberately NOT a mode: a rate limit means "retry later" by
definition (§4), so stamping one non-retryable is the misclaim this
factory exists to make impossible — build that intent with a domain
classifier registered before this one (composition order), where
the decision is visible in your own code.
The returned hook carries the built-in's injectable now
(keyword-only, defaulting to the real clock) so tests and custom
clock domains pin HTTP-date parsing exactly as they pin the
built-in's. The attempt argument is the
:data:RetryClassifierHook protocol's required arity and is
deliberately unread — the verdict never depends on the attempt
count (pinned by the identity battery's attempt sweep).
Why the transient mode exists: the built-in's indefinite claim is
the haunt class for a transient actor with no time_budget
(the §5 danger block). A consumer who wants 429 → BOUNDED retry must
not have to hand-roll a classifier to say so — build this mode, keep
the shared parsing and bounds, and max_attempts does the
stopping.
Example::
bounded_rate_limit = make_rate_limit_aware_classifier(claim_kind="transient")
compose_retry_classifiers(my_domain_classifier, bounded_rate_limit)
Source code in src/taskq/retry.py
1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 | |
failure_taxonomy_classifier ¶
failure_taxonomy_classifier(
*,
transient_status: Iterable[int] | None = None,
non_retryable_status: Iterable[int] | None = None,
include_names: Iterable[str] | None = None,
exclude_names: Iterable[str] | None = None,
infer_timeout_by_suffix: bool = False,
) -> RetryClassifierHook
Configurable classifier for the common failure shapes, the conservative default the §4 table describes.
Where :func:rate_limit_aware_classifier owns the one signal with
dedicated semantics (429/rate-limit → indefinite), this classifier
owns the mundane taxonomy: the shapes a declared transient policy
should govern but a single exception type cannot express
per-instance.
Claims, in check order:
exclude_names— an exception whose class name is listed returnsNonewhatever else matches (an explicit exclusion outranks every signal, status included).transient→ :class:RetryOverride(kind="transient")— claimed by any of:- connection-class:
isinstanceof the builtinConnectionErrorfamily (ConnectionResetError,ConnectionRefusedError, …); - TimeoutError-shaped:
isinstanceof the builtinTimeoutError(coverssocket.timeoutand, since 3.11,asyncio.TimeoutError) — a TYPE-based signal, always on — or an exact class name ininclude_namesor theDEFAULT_TRANSIENT_EXCEPTION_NAMEScurated set; infer_timeout_by_suffix=True→ additionally, a class name ENDING inTimeout/TimeoutErrorclaims transient. This is the opt-in convenience flag — OFF by default, and the default is a contract: defaults never override explicit direction, so the library will not widen a consumer's narrowly pinned semantics. The cost the flag buys: it is a name heuristic, not a semantics check, and its proven counterexample ispymongo.errors.ExecutionTimeout— the server killed the operation for exceedingmaxTimeMS, and a re-run of the same query re-fails deterministically — whose name ends inTimeoutbut whose meaning is "this work can never succeed". Under the default exact-name matching the taxonomy returnsNonefor it (the declared policy governs); with the flag on it is claimedtransientand every such retry burns budget on unwinnable work. Consumers who route these shapes narrowly (static-policy, pinned per class) keep the default; consumers who want the httpx/requests/aiohttp-style convenience without enumerating names take the flag knowingly;- status in
transient_status(defaults:DEFAULT_TRANSIENT_STATUSES= 408, 425, and the 5xx band), duck-typed through the same status shapes the built-in reads. non_retryable→RetryOverride(kind="non_retryable")— status innon_retryable_status(defaults:DEFAULT_NON_RETRYABLE_STATUSES= 4xx minus {408, 425, 429}). A status present in both sets is transient — check order decides, and "retry later" is the safer wrong answer than killing a retryable job.- Anything else →
None. The conservative default: over-claiming is the haunt class.
Configuration replaces, never merges: passing
transient_status=frozenset({429}) claims ONLY 429 (the 5xx band
and 408/425 stop being transient); include_names replaces the
default name set (extend DEFAULT_TRANSIENT_EXCEPTION_NAMES | {"Yours"}
to keep the defaults). Build the classifier once at registration and
reuse it — the sets are frozen at build time, per-call cost is a few
frozenset lookups.
Why the taxonomy never returns indefinite: an indefinite
override has no attempt ceiling, and its single stopping condition is
the job's schedule_to_close — which a transient actor never
has stamped (time_budget is only honored for an
indefinite-declared policy), so composing an
indefinite-claiming classifier into a transient actor retries
the job forever: no attempt ceiling, no deadline, no warning at
registration or override time. That is the haunt this module's §5
danger block documents for the one built-in that genuinely needs the
kind (a rate limit means "wait it out"). The taxonomy's signals carry
no such semantics: a connection reset or a 503 is what
max_attempts-bounded retrying is FOR, so every claim here is
transient (bounded by the declared budget) or non_retryable
(terminal), and the declared policy keeps governing everything the
taxonomy is unsure about. Pin:
tests/test_failure_taxonomy_classifier.py::test_taxonomy_never_returns_indefinite.
Source code in src/taskq/retry.py
1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 | |
time_budget_as_interval ¶
Return retry.time_budget when kind=='indefinite' and time_budget
is set; otherwise None. Used by the enqueue path to pass
time_budget as a $N::interval parameter so PG can compute
schedule_to_close = clock_timestamp() + $N::interval.
Source code in src/taskq/retry.py
decide_after_failure ¶
decide_after_failure(
actor_config: ActorConfigLike,
exception: BaseException,
job_state: JobRetryState,
*,
max_retry_backoff: timedelta = DEFAULT_MAX_RETRY_BACKOFF,
log: BoundLogger | None = None,
) -> RetryDecision
Adapter between the pure classifier and the consumer loop.
Reconstructs a RetryPolicy from row-stored max_attempts / retry_kind
(authoritative) combined with live-registration scalars
(backoff, base, cap, jitter, time_budget) that are not stored on the
row, reusing the registered policy object directly when the row
agrees with it and the registered policy satisfies the cap>=base
invariant, so the common no-drift path skips per-failure pydantic
validation. Any row/registration mismatch falls through to the full
constructor, which fails loud. If the actor registered a
retry_classifier hook, invokes it to get a per-exception
:class:RetryOverride, then delegates to RetryClassifier.classify.
max_retry_backoff is the global ceiling forwarded to
compute_backoff. The consumer passes
settings.max_retry_backoff so the knob is operator-controlled.
No clock input: the classifier decides retry-kind and backoff only ,
the schedule_to_close deadline is arbitrated server-side by
mark_failed_or_retry (see RetryClassifier's docstring).
Source code in src/taskq/retry.py
1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 | |
invoke_on_retry_exhausted
async
¶
invoke_on_retry_exhausted(
hook: OnRetryExhausted | None,
job_row: JobRow,
exception: BaseException,
timeout: float,
*,
log: BoundLogger | None = None,
) -> None
Invoke the on_retry_exhausted hook, best-effort and timeout-bounded.
If the hook is None, returns immediately. If the hook returns a coroutine, wraps the await in asyncio.wait_for with the given timeout. TimeoutError and other exceptions are caught and logged at WARNING; they never propagate to the caller.
Source code in src/taskq/retry.py
invoke_on_success
async
¶
invoke_on_success(
hook: OnSuccess | None,
job_row: JobRow,
result: object,
timeout: float,
*,
log: BoundLogger | None = None,
) -> None
Invoke the on_success hook, best-effort and timeout-bounded.
If the hook is None, returns immediately. If the hook returns an awaitable, wraps the await in asyncio.wait_for with the given timeout. TimeoutError and other exceptions are caught and logged at WARNING; they never propagate to the caller.
Source code in src/taskq/retry.py
invoke_on_cancel
async
¶
invoke_on_cancel(
hook: OnCancel | None,
job_row: JobRow,
timeout: float,
*,
log: BoundLogger | None = None,
) -> None
Invoke the on_cancel hook, best-effort and timeout-bounded.
Runs beside the terminal write that moved the job to cancelled,
so it can neither block that write nor undo it: work cut short still
has to release whatever it held, and a hook that could raise into
this path would leave the row running behind a lease only the
reclaim sweep clears.
Source code in src/taskq/retry.py
safe_mark_failed_or_retry
async
¶
safe_mark_failed_or_retry(
backend: Backend,
job_id: JobId,
worker_id: UUID,
error_info: ErrorInfo,
retry_delay: timedelta | None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
*,
log: BoundLogger | None = None,
attempt: int | None = None,
claim_epoch: int | None = None,
) -> JobRow | None
Wrap mark_failed_or_retry, catching WorkerOwnershipMismatch .
Returns the persisted JobRow on success, or None on ownership mismatch
(signals the caller to skip the on_retry_exhausted hook). attempt is
the attempt-identity epoch threaded from the handler's job-row
snapshot, and claim_epoch the claim-identity epoch, see
Backend.mark_failed_or_retry; a fenced-out epoch surfaces here as
the same None a worker-fence miss produces.