Package Overview¶
TaskQ public API. The taskq package re-exports the primary types used by application
code: the @actor decorator, JobsClient / TaskQ, JobHandle, exceptions,
RetryPolicy, cron scheduling, and batch helpers.
from taskq import actor, TaskQ, JobHandle, JobFailed, RetryPolicy
from taskq import cron, ScheduleHandle
from taskq.context import JobContext
from taskq.di import ProviderRegistry, Scope
taskq ¶
TaskQ: async-native, Postgres-backed background job library.
Canonical imports:
from taskq import actor, TaskQ, JobHandle, JobFailed, RetryPolicy
from taskq import cron, ScheduleHandle
from taskq.context import JobContext
from taskq.di import ProviderRegistry, Scope
TERMINAL_STATUSES
module-attribute
¶
TERMINAL_STATUSES: frozenset[JobStatus] = frozenset(
{
"succeeded",
"failed",
"cancelled",
"crashed",
"abandoned",
}
)
IdempotencyKey
module-attribute
¶
Distinguishes idempotency keys from identity keys at call sites.
IdentityKey
module-attribute
¶
Distinguishes identity keys from idempotency keys at call sites.
JobId
module-attribute
¶
Opaque job identifier, prevents UUID mixups across the API.
QueueName
module-attribute
¶
Validator alias for queue names, accepts plain str literals.
Why Annotated and not NewType: queue names are plain strings
requiring validation, no nominal type because no other str field
at any call site could be confused with queue. Annotated gives
runtime validation in Pydantic models without forcing every caller to
wrap literals in QueueName("default").
__all__
module-attribute
¶
__all__ = [
"TERMINAL_STATUSES",
"AbortBatchAfter",
"ActorConfigDriftError",
"ActorConfigDriftList",
"ActorConfigRow",
"ActorDeregistrationError",
"ActorFn",
"ActorFnWithCtx",
"ActorHandler",
"ActorHasActiveJobsError",
"ActorHasEnabledSchedulesError",
"ActorNotFoundError",
"ActorRef",
"ActorsClient",
"Backend",
"BackpressureError",
"BackpressureSnapshot",
"BatchAbortedError",
"BatchCompletionStatus",
"BatchCounts",
"BatchFailurePolicy",
"BatchFilter",
"BatchHandle",
"BatchMaxPendingExceededError",
"BatchRow",
"BatchSummary",
"BulkCancelResult",
"CancelPhase",
"CancelResult",
"Clock",
"ConnFactory",
"CronScheduleSpec",
"DIError",
"DependencyCycle",
"DeregisterResult",
"DstStrategy",
"DuplicateIdempotencyKeyError",
"EmptyBatchError",
"EmptyFilterError",
"EnqueueItem",
"ErrorReporter",
"EventRetentionGapError",
"EventRow",
"Fail",
"FakeClock",
"IdempotencyKey",
"IdempotencyKeyActorMismatchError",
"IdentityKey",
"IllegalStateTransition",
"JobContext",
"JobEvent",
"JobFailed",
"JobFilter",
"JobHandle",
"JobId",
"JobPage",
"JobRetryState",
"JobRow",
"JobSortField",
"JobStatus",
"JobsClient",
"MaxPendingExceededError",
"MaxPendingLockTimeoutError",
"MissingProvider",
"NullErrorReporter",
"OIDCSettings",
"OnCancel",
"OnSuccess",
"PartialBatchError",
"PayloadValidationError",
"PgCredential",
"PgCredentialProvider",
"PoolFactory",
"ProgressEvent",
"ProgressTooLarge",
"QueueBackpressure",
"QueueMode",
"QueueName",
"RateLimitBackend",
"RedisCredential",
"RedisCredentialProvider",
"RedisFactory",
"ReservationUnavailable",
"ResultTooLarge",
"ResultUnavailable",
"Retry",
"RetryAfter",
"RetryClassifier",
"RetryClassifierHook",
"RetryDecision",
"RetryKind",
"RetryOverride",
"RetryPolicy",
"SAMLSettings",
"ScheduleHandle",
"ScheduleRecord",
"SchemaNotMigratedError",
"ScopeViolation",
"ScopedIdempotencyMigrationPendingError",
"SingletonCollisionError",
"Snooze",
"StreamUnavailable",
"SubEnqueueError",
"SubJobEnqueuer",
"SystemClock",
"TaskQ",
"TaskQError",
"TaskQSettings",
"UniqueForLockTimeoutError",
"UnknownQueueError",
"WorkerConnections",
"WorkerOwnershipMismatch",
"WorkerSettings",
"__version__",
"actor",
"apply_batch_terminal_outcome",
"compose_retry_classifiers",
"cron",
"enrich_pg_dsn",
"ensure_sslmode_require",
"failure_taxonomy_classifier",
"make_dedicated_conn_factory",
"make_pg_pool_factory",
"make_rate_limit_aware_classifier",
"make_redis_client_factory",
"new_uuid",
"rate_limit_aware_classifier",
"register_cron",
"validate_actor_payload",
"wait_for_batch",
]
ActorFn ¶
Actor handler that takes only a payload.
The dispatcher injects nothing beyond the validated payload model. Use this shape for actors that don't need cancellation cooperation, attempt counters, or other context fields.
ActorFnWithCtx ¶
Actor handler that takes a payload and a typed :class:JobContext.
Declare ctx: JobContext[YourPayload] as the second parameter to
opt into context injection. The dispatcher constructs the context per
attempt, populates it with the validated payload and a fresh
:class:asyncio.Event for cooperative cancellation, then passes it
to the handler. Handlers that don't declare ctx skip this work.
JobStatus ¶
JobStatus = Literal[
"pending",
"scheduled",
"running",
"succeeded",
"failed",
"cancelled",
"crashed",
"abandoned",
]
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.
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.
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.
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.
ActorHandler ¶
Bases: Protocol
Most-general actor signature: payload first, then ctx and/or DI deps.
Pyright infers P_ from the first positional parameter (the
payload model) and R_ from the awaited return type, regardless
of how many additional ctx / DI parameters the handler
declares. This lets the :func:actor decorator preserve full
payload-and-result inference for FastAPI-style handlers like::
async def my_actor(
payload: OrderPayload,
ctx: JobContext[OrderPayload],
*,
db: DbSession,
http: HttpClient,
) -> OrderResult: ...
Without ActorHandler, callers would have to choose between
:data:ActorFn (payload only) and :data:ActorFnWithCtx (payload
+ ctx), neither of which describes the DI case.
ActorRef ¶
ActorRef(
*,
name: str,
queue: str,
fn: Callable[..., object],
is_sync: bool = False,
wants_ctx: bool,
dependencies: dict[str, type[object]],
payload_type: type[P],
result_adapter: TypeAdapter[R],
retry: RetryPolicy,
result_ttl: timedelta | None,
singleton: bool = False,
max_concurrent: int | None = None,
max_pending: int | None = None,
metadata: dict[str, object] | None = None,
unique_for: timedelta | None = None,
unique_states: tuple[
JobStatus, ...
] = DEFAULT_UNIQUE_STATES,
start_to_close: timedelta | None = None,
rate_limits: list[
str
| KeyedRateLimitRef
| TokenBucket
| SlidingWindow
]
| None = None,
reservations: list[
str | KeyedReservationRef | ConcurrencyReservation
]
| None = None,
non_retryable_exceptions: tuple[
type[BaseException], ...
] = (),
retry_classifier: RetryClassifierHook | None = None,
on_retry_exhausted: OnRetryExhausted | None = None,
on_retry_exhausted_timeout: float = 3.0,
on_success: OnSuccess | None = None,
on_success_timeout: float = 3.0,
on_cancel: OnCancel | None = None,
on_cancel_timeout: float = 3.0,
priority: int = 0,
)
Typed reference to a registered actor.
Created by the :func:actor decorator. Not callable directly via
the queue path, enqueue jobs by passing this ref to
:meth:JobsClient.enqueue(ref, payload) <taskq.client.JobsClient.enqueue>.
Direct in-process invocation (await my_actor(payload, ...)) is
available for tests and simulators.
The two type parameters carry the actor's payload and result types end-to-end:
payload_typere-validates raw dispatch-time payloads back toP(via :meth:pydantic.BaseModel.model_validate).result_adapterround-trips actor returns through the JSONBresultcolumn (dump_python(mode="json")on the worker side,validate_pythonon the client side).
Attributes:
| Name | Type | Description |
|---|---|---|
max_concurrent |
Fleet-wide concurrency cap for this actor.
max_concurrent may transiently exceed configured value by up to (num_active_producers - 1) * max_concurrent per actor under heavy contention (or (num_producers - 1) * limit_n when limit_n < max_concurrent). For strict correctness, use ConcurrencyReservation. |
|
metadata |
Arbitrary key-value metadata stored in
|
Both are stored as instance fields rather than class metadata so
pyright can infer P and R from a constructor call without
relying on phantom-type tricks.
wants_ctx records whether the handler declared a
:class:JobContext parameter; the dispatcher passes ctx only
when set. dependencies maps each DI parameter name to its
annotated type, the worker's DI pass resolves these at dispatch
time and passes them as keyword arguments to the handler. The DI
resolver itself is an erasure boundary (see
the resolver operates on the registered provider graph at runtime): user-declared annotations on
DI parameters are captured here as type[object] because the
resolver operates on the registered provider graph at runtime.
Source code in src/taskq/actor.py
__slots__
class-attribute
instance-attribute
¶
__slots__ = (
"_fn",
"dependencies",
"is_sync",
"max_concurrent",
"max_pending",
"metadata",
"name",
"non_retryable_exceptions",
"on_cancel",
"on_cancel_timeout",
"on_retry_exhausted",
"on_retry_exhausted_timeout",
"on_success",
"on_success_timeout",
"payload_type",
"priority",
"queue",
"rate_limits",
"reservations",
"result_adapter",
"result_ttl",
"retry",
"retry_classifier",
"singleton",
"start_to_close",
"unique_for",
"unique_states",
"wants_ctx",
)
on_retry_exhausted_timeout
instance-attribute
¶
config
property
¶
This ref, viewed as the per-actor config the consumer reads.
The registration record and the config are one object here, so the property is a named view rather than a second store, it exists so callers can say what they need (the retry policy and the lifecycle hooks) instead of reaching for the whole ref, and so the structural contract is asserted at one place.
fn
property
¶
Return the underlying handler.
May be sync (def) or async (async def). Direct
invocation through :meth:__call__ is the safer path because
it handles both shapes and enforces the declared signature.
__call__
async
¶
Direct invocation, bypasses enqueue, runs the handler in-process.
Pass a :class:JobContext only when the registered handler
declared one (when :attr:wants_ctx is True); calling with
a context against a no-ctx handler raises :class:TypeError,
and calling without a context against a ctx handler also
raises :class:TypeError.
deps mirrors the keyword arguments the worker's
dependency-injection pass supplies in production. Tests may
pass them explicitly; the call site is responsible for
matching the names recorded in :attr:dependencies. Missing
dependencies surface as a runtime TypeError from Python's
argument binding.
Production callers go through :meth:JobsClient.enqueue.
Source code in src/taskq/actor.py
ActorConfigRow
dataclass
¶
ActorConfigRow(
actor: str,
max_concurrent: int | None,
max_pending: int | None,
queue: str,
result_ttl: float | None,
metadata: dict[str, object],
updated_at: str,
)
Snapshot of one {schema}.actor_config row.
DeregisterResult
dataclass
¶
DeregisterResult(
actor: str,
queue: str,
actor_config_deleted: bool,
schedules_disabled: int,
jobs_cancelled: int,
terminal_jobs_remaining: int,
queue_purged: bool,
)
Outcome of a deregister_actor call.
actor_config_deleted is always True, if the row is not found,
deregister_actor raises :class:ActorNotFoundError instead of
returning a result with False. The field is retained for API
contract clarity and consumer assertions.
PgCredential
dataclass
¶
A Postgres credential issued by a rotating-credential provider.
password is always required (a token or dynamic password).
username, when set, overrides the DSN's userinfo user - needed by
providers that issue a fresh username alongside the password (e.g.
Vault dynamic DB creds). When None, the DSN's existing user is
preserved.
lease_duration is how long, in seconds, the issuer will honour this
credential - the Vault lease TTL. It is the bound a rebuild schedule
has to beat for a username-bearing pair (see :class:ReloadSchedule),
and None when the issuer does not say (token providers, whose
tokens refresh per connection and never need one).
PgCredentialProvider ¶
Bases: Protocol
Provides rotating Postgres credentials on demand.
Implementations fetch a fresh token / dynamic username+password each
call. Called by :func:make_pg_pool_factory /
:func:make_dedicated_conn_factory once at pool / connection
construction (to resolve user= and fail fast). A credential with
username unset is then re-fetched for every physical
connection asyncpg opens thereafter - not on each acquire(), which
hands back an already-authenticated connection from the pool - so
token providers are expected to cache and only hit the issuing service
when the cached token is near expiry. A credential that carries a
username is an issued pair used for the pool's life and is fetched
again only when the pool is rebuilt.
RedisCredential
dataclass
¶
RedisCredentialProvider ¶
Bases: Protocol
Provides rotating Redis credentials on demand.
Implementations fetch a fresh (username, token/password) each call.
Called by :func:make_redis_client_factory on every reconnect via
the redis-py CredentialProvider adapter.
Backend ¶
Bases: Protocol
Contract that both PostgresBackend and InMemoryBackend satisfy.
46 async methods plus two sync methods (subscribe_wake and
subscribe_cancel_wake) (48 methods total) covering enqueue,
dispatch, heartbeat, terminal writes, attempt history, cancel
signals, scheduling / sweeps, read, NOTIFY hook, schedule CRUD,
and batch operations. Method order grouped for review-grep
ergonomics.
Why monomorphic (no Generic[P, R]): the backend is the DB
adapter boundary. Payloads are stored as dict[str, object] (the
JSONB payload column) regardless of the actor's typed payload
model. Generic parameters here would propagate P and R into
every method (dispatch_batch, mark_succeeded, etc.) with no
safety benefit at the storage layer. The worker consumer
reconstructs the typed JobContext[P] at dispatch time using
ActorRef.payload_type.model_validate(row.payload).
supports_transactional_simulation
class-attribute
¶
Whether this backend simulates transactional sub-enqueue via a buffer (True) or relies on real database transactions (False).
PostgresBackend returns False, its real PG transaction provides
the atomicity guarantee directly: sub-job INSERTs run on the open
LOOP-scope connection and are rolled back along with the parent's
writes if the actor raises.
InMemoryBackend returns True, it has no real transaction
concept, so SubJobEnqueuer buffers EnqueueArgs and flushes on
actor success / discards on failure. A third-party Backend
implementation that wants transactional simulation in tests can opt
in by overriding this to True.
supports_progress_stream
class-attribute
¶
Whether JobHandle.progress_stream may run against this backend
(True), or must refuse up front with :class:NotImplementedError
(False).
PostgresBackend returns True, the PG poll fallback (and the Redis
pub/sub arm on top of it) is always available.
InMemoryBackend returns False, it has no pub/sub channel and no
event table to poll, so JobHandle.progress_stream raises
NotImplementedError ("progress_stream requires Redis; in-memory
backend does not support SSE.") before touching the transport. This
capability flag is the production-side replacement for a
client → testing import: the client layer asks the backend
instead of naming the test double. A third-party Backend that
cannot back a progress stream overrides this to False.
enqueue
async
¶
Source code in src/taskq/backend/_protocol.py
enqueue_batch
async
¶
enqueue_batch(
args_list: list[EnqueueArgs],
*,
connection: ConnLike | None = None,
enforce_max_pending: bool = True,
) -> list[JobRow]
Insert multiple jobs in a single batched operation.
All items in args_list must be validated before calling this method, the backend does not re-validate payloads. The list must be non-empty and contain at most 1000 items (enforced by the client layer).
When enforce_max_pending is true (the default), items carrying a
resolved max_pending cap are admission-checked as one
aggregate, existing pending+scheduled per actor plus this batch ,
and admission PARTITIONS per actor: an over-cap actor's items are
refused as a whole group, every other actor's items are inserted,
and :class:~taskq.exceptions.BatchMaxPendingExceededError raises
at the transaction boundary, after the admitted items commit when
this call owns the transaction, immediately after the insert on a
caller-supplied connection with an open transaction (that
transaction's commit/rollback decides their durability). Pass
enforce_max_pending=False only when the caller pre-admitted
every item against current capacity (the cron tick's suppression
preflight), where a re-check could abort an unrelated batch on a
race.
Returns one :class:JobRow per item in args_list, in the same
order. For idempotency-key collisions the existing row is
returned; its id will differ from the requested args.id.
On a cap refusal no rows are returned, the typed error carries
the refused item indices and the admitted count instead.
Source code in src/taskq/backend/_protocol.py
enqueue_batch_fast
async
¶
enqueue_batch_fast(
args_list: list[EnqueueArgs],
*,
connection: ConnLike | None = None,
enforce_max_pending: bool = True,
) -> int
Insert multiple jobs via the COPY FROM protocol for maximum throughput.
COPY cannot evaluate expressions or handle conflicts, so the write
is two statements inside one transaction: a bare COPY of the
domain-insensitive columns, then a corrective UPDATE
(enqueue_batch_fast_fixup) that stamps/decides the
clock-sensitive ones, status, scheduled_at,
schedule_to_close, result_expires_at, from the database
clock (clock_timestamp()); created_at takes its DDL
default (now()). Nothing is observable half-fixed: both
statements commit or abort together.
Consequences of the COPY-no-conflicts shape:
scheduled_at=Nonemeans immediate, the fixup's server-side CASE stamps it and decidespending/scheduled(the same single-arbiter contract as :meth:enqueue/:meth:enqueue_batch).schedule_to_close_interval/result_ttlare anchored to the server clock at ENQUEUE time by the fixup, a future-scheduled item with a short interval can therefore fail DeadlineExceeded before it is ever dispatched.- A duplicate
idempotency_key, within the batch or already stored, violates the unique index and aborts the ENTIRE batch (all-or-nothing atomicity; nothing is written), surfacing as :class:~taskq.exceptions.DuplicateIdempotencyKeyError. - Items carrying a resolved
max_pendingcap are admission-checked as one aggregate before the COPY, with the same per-actor partition as :meth:enqueue_batch, within-cap actors' rows are COPY'd, an over-cap actor's items are refused and raise :class:~taskq.exceptions.BatchMaxPendingExceededErrorafter the COPY commits; disable withenforce_max_pending=Falseonly for pre-admitted internal callers.
Returns the count of rows written. On success this is exactly
len(args_list), this path never deduplicates, so the count
never includes pre-existing rows. The in-memory mirror implements
the same contract: duplicates raise
:class:~taskq.exceptions.DuplicateIdempotencyKeyError before
any row is written, and the count is the number of items.
This is a performance-focused variant of :meth:enqueue_batch
(which DOES deduplicate idempotency-key collisions via ON
CONFLICT). Use for bulk import / backfill with 10K+ rows where
collision handling is not needed. Max batch size is 50 000
(client-enforced).
Source code in src/taskq/backend/_protocol.py
enqueue_with_conn
async
¶
Enqueue a job using the supplied connection.
The connection MUST already be in an open transaction managed by
the caller, this method does NOT issue BEGIN/COMMIT. The
autonomous variant enqueue(args) acquires its own connection
and opens a transaction internally.
Source code in src/taskq/backend/_protocol.py
dispatch_batch
async
¶
heartbeat_jobs
async
¶
heartbeat_jobs(
worker_id: UUID,
lock_lease: timedelta,
*,
disowned: Collection[UUID] = (),
) -> int
Renew the lock lease of every running job worker_id holds, except the disowned ids, rows the worker could not record an outcome for, whose leases must lapse for the reclaim sweep. Returns the number of rows renewed.
Source code in src/taskq/backend/_protocol.py
extend_reservation_leases
async
¶
extend_reservation_leases(
worker_id: UUID,
lock_lease: timedelta,
*,
disowned: Collection[UUID] = (),
) -> int
Renew the reservation-slot leases of every running job
worker_id holds, with the same disowned exclusion as
:meth:heartbeat_jobs. Returns the number of slots renewed.
Source code in src/taskq/backend/_protocol.py
mark_succeeded
async
¶
mark_succeeded(
job_id: JobId,
worker_id: UUID,
result: dict[str, object] | None = None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
fallback_result_ttl: timedelta | None = None,
*,
result_bytes: bytes | None = None,
attempt: int | None = None,
claim_epoch: int | None = None,
) -> bool
Mark a job succeeded, computing result_expires_at at completion.
attempt is the attempt-identity epoch: the handler's
dispatch-time job-row attempt snapshot. The terminal fence is
one epoch deeper than the worker fence, the write lands only on
a row whose current attempt matches, so a stale handler's
write after a same-worker reclaim/redispatch (the row re-dispatched
at attempt + 1 on the same worker) no-ops exactly like a
different worker's late write. None, a caller that cannot
present the epoch, also no-ops: a terminal write that cannot
prove which attempt it terminates must not terminate any attempt.
claim_epoch is the non-saturating claim-identity fence, the
claim_epoch value of the handler's own claim view. The
displayed attempt counter saturates at the smallint ceiling,
where a reclaim plus a redispatch would hand the stale handler
and the live one the same (worker, attempt) pair; the claim
epoch keeps advancing, so the stale write can only no-op.
None never satisfies the equality, the same
cannot-prove-it doctrine attempt applies.
The result reaches the backend in exactly one of two forms:
result, the actor's result dict, which the backend serializes
exactly once (orjson via :func:taskq._json.dumps_jsonb_str,
NUL-guarded), or result_bytes, the result already serialized
to orjson bytes (the exact output of :func:taskq._json.dumps),
which the backend reuses as-is: bound as
result_bytes.decode("utf-8") with result_size_bytes =
len(result_bytes) and no second serialization. The worker
consumer always passes result_bytes (it serialized the result
once already); direct callers normally pass result. Passing
both non-None raises ValueError; both None stores a NULL
result. The result_bytes form is NUL-guarded at the same
boundary, raising the same ValueError the dict form raises.
Expiry resolution, first match wins: a non-NULL stored
actor_config.result_ttl (operator-owned) applies; otherwise
fallback_result_ttl, the worker-side @actor(result_ttl=...)
literal, which the terminal-write SQL cannot see, applies;
otherwise the row's existing result_expires_at is kept. The
computed arms use clock_timestamp(), the wall-clock time the
write executes, not the transaction start, so neither a long
queue wait nor a long actor runtime can make a job complete
already expired and have its result reaped immediately.
Source code in src/taskq/backend/_protocol.py
mark_succeeded_with_conn
async
¶
mark_succeeded_with_conn(
conn: ConnLike,
job_id: JobId,
worker_id: UUID,
result: dict[str, object] | None = None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
fallback_result_ttl: timedelta | None = None,
*,
result_bytes: bytes | None = None,
attempt: int | None = None,
claim_epoch: int | None = None,
) -> bool
Mark a job succeeded using the supplied connection.
Used by the consumer when a LOOP-scope asyncpg.Connection is
available so the success status update commits atomically with
the actor's writes and sub-job INSERTs in the same transaction.
The connection MUST already be in an open transaction; this
method does NOT open or close one. The autonomous variant
mark_succeeded(...) acquires its own connection.
result / result_bytes follow the same two-form contract as
:meth:mark_succeeded, pass exactly one, or neither for a NULL
result. fallback_result_ttl follows the same resolution rule
as :meth:mark_succeeded. attempt and claim_epoch follow
the same fencing as :meth:mark_succeeded.
Source code in src/taskq/backend/_protocol.py
mark_failed_or_retry
async
¶
mark_failed_or_retry(
job_id: JobId,
worker_id: UUID,
error_info: ErrorInfo,
retry_delay: timedelta | None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
*,
attempt: int | None = None,
claim_epoch: int | None = None,
) -> JobRow
Mark a running job failed, or schedule a retry retry_delay later.
retry_delay=None is the terminal-fail arm (status='failed',
the original error_info persisted). A non-None delay is applied
by the backend's own clock, never the caller's: scheduled_at =
now() + delay, floored at
:data:taskq.constants.MIN_DEFERRAL_INTERVAL, the same bound the
deferral arms apply, so a failure-retry decision can never requeue
below the deferral floor, and the scheduled/pending status
derives from the effective delay alone (a sub-floor delay still
lands scheduled at least the floor out). The same statement
arbitrates the schedule_to_close deadline server-side, when
clock_timestamp() + effective delay would land past the
deadline, the row is failed with error_class='DeadlineExceeded'
instead of retried, so app↔DB clock skew can neither void the
retry backoff nor kill a job whose deadline has not actually
passed.
attempt is the attempt-identity epoch, see
:meth:mark_succeeded. A fenced-out write (stale epoch, wrong
worker, or a row that moved) raises
:class:~taskq.exceptions.WorkerOwnershipMismatch, which
:func:taskq.retry.safe_mark_failed_or_retry converts to the
None the handlers treat as a no-op.
Source code in src/taskq/backend/_protocol.py
mark_cancelled
async
¶
mark_cancelled(
job_id: JobId,
worker_id: UUID,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
*,
attempt: int | None = None,
claim_epoch: int | None = None,
) -> bool
Source code in src/taskq/backend/_protocol.py
write_cancel_escalation
async
¶
mark_abandoned
async
¶
mark_snoozed
async
¶
mark_snoozed(
job_id: JobId,
worker_id: UUID,
delay: timedelta,
*,
metadata_update: dict[str, object] | None = None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
outcome: SnoozeOutcome = "snoozed",
attempt: int | None = None,
claim_epoch: int | None = None,
denial_reason: DenialReason = "capacity",
) -> Literal["scheduled", "failed", "noop"]
Release a running job back to the queue without consuming retry budget.
attempt is the attempt-identity epoch, see
:meth:mark_succeeded; a fenced-out write returns "noop".
claim_epoch is the claim-identity fence, see
:meth:mark_succeeded; it applies to all three arms, the two
terminal deadline exits included.
A non-terminal snooze/denial writes NO job_attempts /
job_events rows, it is admission control or a voluntary
deferral, not an execution, and is counted on the job row
(snooze_count, or rate_limit_blocked_count when outcome
is a denial) plus OTEL. max_attempts is never raised: the
ceiling is a bound, not a counter.
outcome admits only the three deferral outcomes
(:data:SnoozeOutcome), the statement's arms key on exactly
those; an execution outcome has no arm (PG would leave the job
stranded running) and raises ValueError at the boundary
on both backends instead. delay is floored at
:data:taskq.constants.MIN_DEFERRAL_INTERVAL: a non-consuming
deferral reschedules at least that far out, so a zero delay
cannot park the job at the head of the dispatch order.
Every deferral shape refunds the claim's attempt increment
(floored at 0), so no deferral, actor-requested or admission
denial, spends retry budget. An admission denial carries HTTP
429 semantics: it reports that the fleet had no slot, which says
nothing about the work, so it can neither charge the budget nor
decide the outcome. A denied job reschedules until capacity
frees; its only terminal exit is its own schedule_to_close
("failed", DeadlineExceeded), and the counters on the row
are how sustained contention stays visible.
denial_reason names the cause of a denial-class outcome
(:data:DenialReason), "capacity" (the default) is a
saturation denial, the store answering "full"; "unavailable" is
the store failing to answer. Both take the identical non-consuming
path; the value is validated at the boundary so an undefined reason
is refused rather than silently accepted. It is persisted on
exactly one surface: the state_change event detail of the
mark_snoozed deadline arm, and only when that TERMINAL arm fires
on a denial-keyed outcome (reservation_denied /
rate_limit_denied whose schedule_to_close has lapsed),
where it names which starvation (saturation vs store outage)
killed the job. The discriminator is terminality, not
denial-ness: every non-terminal deferral, denial or not, writes
no event row, so grepping events for denial_reason finds
nothing on a live denial; the row's outcome counters are the
live-denial signal.
Source code in src/taskq/backend/_protocol.py
2225 2226 2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243 2244 2245 2246 2247 2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258 2259 2260 2261 2262 2263 2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 | |
mark_retry_after
async
¶
mark_retry_after(
job_id: JobId,
worker_id: UUID,
delay: timedelta,
*,
consume_budget: bool = True,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
attempt: int | None = None,
claim_epoch: int | None = None,
) -> Literal[
"scheduled",
"failed:DeadlineExceeded",
"failed:MaxAttemptsExceeded",
"noop",
]
Source code in src/taskq/backend/_protocol.py
mark_interrupted
async
¶
mark_interrupted(
job_id: JobId,
worker_id: UUID,
*,
attempt: int,
hold: timedelta,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
claim_epoch: int | None = None,
) -> Literal[
"pending",
"scheduled",
"failed:DeadlineExceeded",
"noop",
]
Release a running attempt this worker cannot finish because the process is going away.
The interruption releases a *started* attempt, so it does NOT
refund the claim's increment: the attempt did start executing, and
refunding it would re-create the exact attempt epoch the
interrupted handler still holds; that handler's later terminal
write would then pass the attempt fence and land on the
re-dispatched attempt, and the live execution's own terminal write
would no-op. No job_attempts row is written (an
interruption is not an execution outcome), one job_events
state_change with detail.reason = 'interrupted' records the
transition, and the row's interrupt_count is bumped. The
interrupted attempt counts against the retry budget; the
re-dispatch claims a fresh epoch at attempt + 1.
*hold* > 0 parks the row ``scheduled`` until the releasing process
is provably gone (a job released while its coroutine may still be
alive in this process must not be claimable elsewhere until then);
*hold* = 0 lands the row ``pending`` at the head of the order, the
row is genuinely free and the actor is gone, so no deferral floor
applies. A hold that would push the row past its
``schedule_to_close`` fails the job on the deadline instead
(``"failed:DeadlineExceeded"``), the same terminal exit every
deferral arm honours.
Fenced on ownership, the attempt epoch, and ``cancel_phase = 0``:
an operator cancel in flight wins and the call returns ``"noop"``
so the caller routes to the cancel ladder (the row carries the
operator's request; the deploy must not launder it into a release:
a row whose ``cancel_attempted_at`` is set terminalises as
cancelled, never as available).
*attempt* is the attempt-identity epoch, see
:meth:`mark_succeeded`. Here it is required, not optional: a
release that cannot prove which attempt it is handing back must
not touch the row (``"noop"``). *claim_epoch* is the
claim-identity fence, see :meth:`mark_succeeded`, applied to
both arms, the deadline exit included.
Source code in src/taskq/backend/_protocol.py
write_attempt
async
¶
Source code in src/taskq/backend/_protocol.py
get_attempts
async
¶
Source code in src/taskq/backend/_protocol.py
get_events
async
¶
Source code in src/taskq/backend/_protocol.py
poll_reclaim_events
async
¶
poll_reclaim_events(
after_id: int,
limit: int = DEFAULT_RECLAIM_POLL_LIMIT,
*,
visibility_delay: timedelta | None = None,
) -> list[EventRow]
Return up to limit crash-reclaim events with event_id >
after_id, ascending, the durable cursor behind
TaskQ.watch_reclaims.
The wall-stamp visibility margin is delay-only, never a skip: the
poll caps returned ids strictly below the lowest still-held-back
matching row above the cursor, so a clock step-back that inverts a
commit's (occurred_at, id) pair (the earlier-id row stamped
later than its sibling) delays the sibling instead of leaving it
below an advanced cursor. Rows the ceiling withholds are returned
in a later poll, id order preserved. The ceiling bounds delays
only by the held row's own stamp lie: a far-future-stamped row
blocks its higher-id tail for the size of that lie, a wait the
margin does not measure.
An event can still be silently missed if a job_events
writer transaction stays open longer than the visibility-delay
margin between its INSERT and its COMMIT, ids are allocated at
INSERT time but transactions commit out of order, so a
late-committing lower-id row can land behind an already-advanced
cursor. Rows
are therefore held back by a trailing-watermark filter
(visibility_delay; backend-configured default when None ,
see :data:taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY for
the exact assumption and its violation modes, and
PostgresBackend.check_reclaim_visibility_delay_risk for the
diagnostic that makes a violation operator-visible).
Source code in src/taskq/backend/_protocol.py
write_cancel_request
async
¶
cancel_where
async
¶
Cancel all jobs matching filter in a set-based operation.
Pending/scheduled jobs → terminal 'cancelled'. Running jobs → cancel_phase=1 (cooperative cancel + NOTIFY).
The filter's limit, cursor, and order_by fields are
ignored, this is a bulk write, not a paginated read. The write is
still not paginated in outcome, every matching row is cancelled ,
but it executes as internally bounded committed batches, so a
mid-operation failure leaves partial progress rather than rolling
back everything (re-running continues; already-cancelled rows are
skipped).
Guardrail: the client layer (:meth:JobsClient.cancel_where)
rejects empty filters (no predicates) with
:class:EmptyFilterError. Backend implementations receive a
filter that has already been validated. A direct backend call
with JobFilter() renders WHERE TRUE and cancels the
entire table, callers using the backend directly are
responsible for validating the filter.
Returns a :class:BulkCancelResult with counts and affected IDs.
Source code in src/taskq/backend/_protocol.py
poll_cancel_flags
async
¶
retry_job
async
¶
Re-run a job that has come to rest, by re-pending it.
An operator re-run is "run this again", so every terminal status
is a valid source: failed/crashed/cancelled, and also
succeeded (the replay path after a bad deploy, the status
records that the actor returned, never that its side effects were
right) and abandoned (an infrastructure interruption, not a
failure). running is excluded on correctness grounds:
re-pending a row while an attempt is live races that attempt's
terminal write and the job can execute twice concurrently.
pending/scheduled are excluded because the job is already
queued, there is nothing to put back, and re-pending would
discard its place in the dispatch order.
The attempt counter is NOT reset: an idempotent admin operation
must not restart the counter, so a re-run job climbs to fresh
attempt numbers and
no job_attempts write can collide on a spent epoch's primary
key. max_attempts rises to GREATEST(max_attempts,
attempt + 1) (capped at the smallint bound), which opens the
budget gates for at least one fresh execution while a
mid-budget re-run keeps its remaining budget.
Returns True if the job was retried, False if it was not
in a retryable state, or the smallint-bound ceiling cannot rise
past the spent attempt, in which case the row stays terminal
rather than re-pending a job whose next claim would overflow.
Source code in src/taskq/backend/_protocol.py
scheduled_to_pending
async
¶
Promote scheduled jobs whose scheduled_at has passed.
The backend's own clock is the arbiter (PG evaluates
scheduled_at <= clock_timestamp() server-side; InMemory
compares against its injected Clock). Returns the count of
promoted rows.
Source code in src/taskq/backend/_protocol.py
deadline_sweep
async
¶
Fail pending/scheduled jobs whose schedule_to_close has passed.
Transitions to failed with error_class='DeadlineExceeded',
arbitrated by the backend's own clock. Returns the count of swept
rows.
Source code in src/taskq/backend/_protocol.py
reclaim_expired_locks
async
¶
Reclaim running jobs whose lock has expired.
The expiry check is arbitrated by the backend's own clock; the grace parameters only widen the carve-out for jobs with an in-flight cancel request. Returns the count of reclaimed rows.
Source code in src/taskq/backend/_protocol.py
get
async
¶
One job row by id, None when it exists in neither tier.
Both backends read the hot jobs table first and fall back to
the archive tier (an archive hit is marked archived=True),
the same jobs-then-archive probe the CLI applies, so an id whose
row a prune archived still answers instead of reading as missing
(issue #314): PG probes jobs_archive, the twin probes its
prune simulation's _archive via get_archived's storage.
The result-TTL sweep touches only the hot tier on both backends,
so an archive hit reads the result columns it was archived with.
Source code in src/taskq/backend/_protocol.py
list_jobs
async
¶
List jobs matching filters, returning at most filters.limit
rows in keyset-pagination order.
filters.status accepts a single :data:JobStatus or a
sequence of statuses; filters.unfinished is a meta-filter for
non-terminal (True) or terminal (False) statuses ,
'unfinished' here means 'not yet finished' (pending, scheduled, or
running). See :class:JobFilter for details.
Source code in src/taskq/backend/_protocol.py
count_pending_jobs
async
¶
Return pending+scheduled job counts per actor.
Returns a dict mapping actor name to count. Only actors with
at least one pending or scheduled job appear in the result.
Actors not in the result have a count of zero. The actors
list is used as an IN/ANY filter, pass all distinct actor
names from a batch to fetch all counts in one round-trip.
Source code in src/taskq/backend/_protocol.py
count_active_jobs
async
¶
Count non-terminal jobs (pending, scheduled, running) in the given queues.
Returns the total count across all specified queues. Used by the drain monitor to detect when queues are empty. An empty queues list returns 0.
Source code in src/taskq/backend/_protocol.py
get_actor_max_pending
async
¶
Return the stored actor_config.max_pending for every actor
with a row.
Key present with an int value: the stored (operator-owned)
limit. Key present with None: a row exists but the column is
NULL (a cleared override). Key absent: no stored row. Client-side
capacity resolution
(:class:taskq.client._capacity.ActorCapacityCache) treats
"absent" and "NULL" identically, both fall back to the
@actor(...) literal; the distinction is preserved here only
so observability callers can tell them apart.
This is the enqueue-path analog of the dispatch CTE's per-cycle
actor_config join: one small whole-table read, consumed
through a TTL-bounded cache so the hot path pays no per-enqueue
query.
Source code in src/taskq/backend/_protocol.py
subscribe_wake ¶
subscribe_cancel_wake ¶
Return an async context manager yielding a fresh asyncio.Event
that is set whenever a cancel NOTIFY arrives for any job.
The heartbeat loop uses this to interrupt its sleep immediately on cancel, rather than waiting for the next scheduled tick.
Source code in src/taskq/backend/_protocol.py
subscribe_leader_wake ¶
Return an async context manager yielding a fresh asyncio.Event
that is set whenever a leader resign broadcast arrives on the
leadership wake channel (leader_wake_channel).
The election loop uses this to re-run the fenced elect immediately when a peer hands the lease back, rather than waiting out the heartbeat tick. The broadcast is a HINT: a backend with no LISTEN wire simply never sets the event, and the loop keeps today's tick cadence, which is the guarantee the crash bound rests on.
Source code in src/taskq/backend/_protocol.py
create_schedule
async
¶
Source code in src/taskq/backend/_protocol.py
list_schedules
async
¶
update_schedule
async
¶
delete_schedule
async
¶
Source code in src/taskq/backend/_protocol.py
enqueue_batch_atomic
async
¶
enqueue_batch_atomic(
items: Iterable[EnqueueArgs],
*,
batch_id: UUID,
queue: str,
batch_row: BatchRow | None,
finalizer_args: EnqueueArgs | None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> list[JobRow]
Source code in src/taskq/backend/_protocol.py
create_batch
async
¶
create_batch(
batch_id: UUID,
queue: str,
expected_size: int,
failure_threshold: int | None,
finalizer_job_id: UUID | None,
originating_actor: str | None,
*,
connection: ConnLike | None = None,
) -> None
Source code in src/taskq/backend/_protocol.py
increment_batch_failures
async
¶
reset_batch_failures
async
¶
abort_batch
async
¶
complete_batch
async
¶
Arbitrate the batch's completion.
Returns whether the caller owes the batch one post-commit
completion re-arbitration: True only on the
transactional-caller shape whose attempt gated out at the
tail-shape probe while connection was still inside its open
transaction (a gated attempt is not ordered against the caller's
own commit, so an all-terminal batch would otherwise be left
'active' with no attempt left to land). The caller honors it by
calling this method again with no connection= once its
transaction has ended; autonomous callers are never owed one.
Source code in src/taskq/backend/_protocol.py
get_batch
async
¶
list_batches
async
¶
count_batch_non_terminal
async
¶
BatchCounts
dataclass
¶
BatchCounts(
total: int,
pending: int,
succeeded: int,
failed: int,
cancelled: int,
crashed: int,
abandoned: int,
)
Live job-count aggregate for one batch.
Mirrors BatchCompletionStatus fields; defined at the protocol layer so backends do not import the client-side batch module.
BatchFilter
dataclass
¶
BatchFilter(
queue: str | None = None,
active: bool | None = None,
batch_id: UUID | None = None,
limit: int = 100,
cursor: str | None = None,
)
Filter parameters for Backend.list_batches.
Unlike JobFilter, this only carries fields relevant to batch queries: queue, active (status terminality), batch_id, limit, and cursor. Job-oriented fields (status, actor, tags, order_by, identity_key) are intentionally absent, using JobFilter for batch queries would silently ignore those fields, which is a type trap.
cursor is the same mechanism as :attr:JobFilter.cursor: an
opaque keyset token encoding the last row of the previous page, which
both backends must decode identically
(:func:~taskq.backend._cursor.encode_batch_cursor). It encodes
(created_at, id) where the job cursor encodes (priority,
scheduled_at, id), one field fewer, same |-delimited shape.
created_at alone is not a total order (the column defaults to
now(), the transaction timestamp, so one
enqueue_batch_atomic stamps every row it writes identically), so
id is the tiebreaker; it is UUIDv7 and therefore time-ordered,
which lets both columns sort DESC together and makes the seam a
single row-wise comparison.
There is no order_by: list_batches has exactly one ordering,
so the cursor cannot disagree with it. list_jobs has three, and
binds each to its own cursor shape through
:class:~taskq.backend._cursor.JobOrdering for the same reason.
__post_init__ ¶
Source code in src/taskq/backend/_protocol.py
BatchRow
dataclass
¶
BatchRow(
id: UUID,
queue: str,
status: BatchStatus,
expected_size: int,
consecutive_failures: int,
failure_threshold: int | None,
finalizer_job_id: UUID | None,
originating_actor: str | None,
created_at: datetime,
completed_at: datetime | None,
metadata: dict[str, object],
)
Read-model of a taskq.batches row.
CancelPhase ¶
Bases: IntEnum
Phases of cooperative-then-forced cancellation.
Why IntEnum and not Literal[0, 1, 2]: the cancel-poll loop
performs arithmetic comparisons (db_phase >= 1,
active.cancel_phase < 2) that Literal[int] does not narrow
correctly under pyright strict. IntEnum subclasses int, so
every existing comparison continues to work, while the typed enum
carries the OTel attribute semantics (cancel_phase attribute on
transition counters) and prevents bare-int values like 99 from
slipping past the type checker.
Values NONE, COOPERATIVE, and FORCED are persistable ,
they map directly to the PG cancel_phase column whose check
constraint is BETWEEN 0 AND 2. ABANDON_PENDING is an
in-process sentinel only: the cancel-poll loop sets it on
_ActiveJob to mark a job as queued for post-transaction
abandonment. It is never written to PG. Keeping it on the same
enum lets cancel_phase stay strongly typed end-to-end.
EventRow
dataclass
¶
EventRow(
event_id: int,
job_id: JobId,
occurred_at: datetime,
kind: Literal["state_change", "cancel_request"],
detail: dict[str, object],
)
Read-model of a taskq.job_events row.
Mirrors the job_events table shape: monotonic event_id,
the owning job, timestamp, event kind, and a detail payload.
JobFilter
dataclass
¶
JobFilter(
queue: str | None = None,
status: JobStatus | Sequence[JobStatus] | None = None,
actor: str | None = None,
identity_key: IdentityKey | None = None,
batch_id: UUID | None = None,
limit: int = 100,
cursor: str | None = None,
tags: tuple[str, ...] | None = None,
order_by: JobSortField | None = None,
active: bool | None = None,
created_before: datetime | None = None,
unfinished: bool | None = None,
)
Filter parameters for :meth:Backend.list_jobs and
:meth:Backend.cancel_where.
For cancel_where, the limit, cursor, and order_by
fields are ignored, a bulk cancel is not paginated. It is still not
paginated in outcome, every matching row is cancelled, but the
write executes as internally bounded committed batches, so a
mid-operation failure leaves partial progress rather than rolling
back everything (re-running continues; already-cancelled rows are
skipped). Use :meth:has_predicates to check whether the filter has
at least one predicate before passing it to cancel_where.
Heads-up: unfinished=True means 'not yet finished', a superset of
non-terminal statuses (pending + scheduled + running),
not just 'currently executing'. Read the unfinished section below
before relying on the name.
cursor is an opaque keyset-pagination token encoding the sort
columns of order_by's ordering from the last row of the previous
page -- (priority, scheduled_at, id) for the default,
(created_at, id) and (finished_at, id) for the others. Encode
it with :func:~taskq.backend._cursor.encode_job_cursor, passing the
same order_by: a cursor is only meaningful under the ordering that
produced it. Both backends must agree on cursor encoding and
comparison semantics.
batch_id is a :class:UUID. The PG backend converts it to its
canonical string form at the SQL boundary; the in-memory backend
compares the UUID directly. Keeping the typed shape here means
JobsClient.list(batch_id=UUID(...)) flows without an implicit
str(uuid) coercion.
status accepts either a single :data:JobStatus (backwards
compatible, e.g. JobFilter(status="pending")) or a sequence of
statuses (e.g. JobFilter(status=["pending", "running"])).
An empty sequence (status=[]) matches no jobs, it is not
treated as 'no filter'. Unknown status values raise
:class:ValueError in :meth:__post_init__, so untrusted input
fails identically on both backends instead of surfacing as a PG
enum-cast error or a silent empty result.
The PG backend renders a single status as status = $n and a
sequence as status = ANY($n); the in-memory backend performs a
membership check in both cases.
unfinished is a meta-filter that selects statuses by terminality,
use it to filter by whether a job is still running or has reached
a terminal state. The name states the predicate it applies:
'not yet finished' includes both work currently executing and work
not yet started:
unfinished=True→ non-terminal statuses (pending, scheduled, running)unfinished=False→ terminal statuses (succeeded, failed, cancelled, crashed, abandoned)unfinished=None(default) → no status-terminality filter
The non-terminal set is derived from
:data:~taskq.backend.statemachine.ACTIVE_STATUSES, which is itself
derived from the state machine, adding a new non-terminal state
updates this filter automatically.
status and unfinished are mutually exclusive; specifying both
raises :class:ValueError in :meth:__post_init__.
Deprecated alias: unfinished was named active, and that name
stays accepted as a constructor kwarg (positional or keyword), it
raises :class:DeprecationWarning and feeds the same predicate.
The rename exists because 'active' read as 'currently executing' to
every new reader, while the predicate is 'not yet finished'.
Usage examples::
JobsClient.list(JobFilter(status="pending"))
JobsClient.list(JobFilter(status=["pending", "running"]))
JobsClient.list(JobFilter(unfinished=True))
active
class-attribute
instance-attribute
¶
__post_init__ ¶
Source code in src/taskq/backend/_protocol.py
1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 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 | |
has_predicates ¶
Return True if at least one filter predicate is set.
Used by JobsClient.cancel_where to reject empty filters that
would match the entire table. New predicate fields added to
JobFilter MUST be added here and in
build_filter_conditions, the two are kept in sync manually.
Non-predicate fields (limit, cursor, order_by) are
excluded by design.
Source code in src/taskq/backend/_protocol.py
JobPage
dataclass
¶
JobRow
dataclass
¶
JobRow(
id: JobId,
actor: str,
queue: str,
payload: dict[str, object],
payload_schema_ver: int,
status: JobStatus,
priority: int,
attempt: int,
max_attempts: int,
retry_kind: RetryKind,
created_at: datetime,
scheduled_at: datetime,
identity_key: IdentityKey | None = None,
fairness_key: str | None = None,
schedule_to_close: datetime | None = None,
start_to_close: timedelta | None = None,
heartbeat_timeout: timedelta | None = None,
started_at: datetime | None = None,
finished_at: datetime | None = None,
last_heartbeat_at: datetime | None = None,
locked_by_worker: UUID | None = None,
lock_expires_at: datetime | None = None,
cancel_requested_at: datetime | None = None,
cancel_phase: CancelPhase = CancelPhase.NONE,
error_class: str | None = None,
error_message: str | None = None,
error_traceback: str | None = None,
progress_state: dict[str, object] = dict[str, object](),
progress_seq: int = 0,
result: dict[str, object] | None = None,
result_size_bytes: int | None = None,
result_expires_at: datetime | None = None,
idempotency_key: IdempotencyKey | None = None,
idempotency_scope: str = "",
trace_id: str | None = None,
span_id: str | None = None,
metadata: dict[str, object] = dict[str, object](),
tags: tuple[str, ...] = (),
snooze_count: int = 0,
rate_limit_blocked_count: int = 0,
interrupt_count: int = 0,
retry_base: timedelta = timedelta(seconds=5),
retry_cap: timedelta = timedelta(hours=1),
retry_backoff: Literal[
"exponential", "linear", "fixed"
] = "exponential",
retry_jitter: float = 0.2,
assignment_routed: bool = False,
claim_epoch: int = 0,
archived: bool = False,
parent_id: JobId | None = None,
)
Read-model of a taskq.jobs row. Every column the dispatch loop,
heartbeat, and terminal writes need appears as a typed field.
status uses a Literal union (8 values) matching the
job_status enum in 01.00.00_01_pre_initial.sql.
cancel_requested_at
class-attribute
instance-attribute
¶
progress_state
class-attribute
instance-attribute
¶
metadata
class-attribute
instance-attribute
¶
snooze_count
class-attribute
instance-attribute
¶
Coalesced count of non-consuming deferrals (Snooze and
RetryAfter(consume_budget=False)) since enqueue, the job-row
record of reschedules that consumed no retry budget. Trailing
default: rows materialised before the counters existed read 0.
rate_limit_blocked_count
class-attribute
instance-attribute
¶
Coalesced count of admission denials (reservation / rate-limit) since enqueue. Trailing default: rows materialised before the counters existed read 0.
interrupt_count
class-attribute
instance-attribute
¶
Coalesced count of infrastructure interruptions (a running attempt
released back to the queue by a worker shutdown) since enqueue, the
interruptions write no job_attempts rows and are not execution
outcomes, so attempt alone cannot count them. Trailing default:
rows materialised before the counter existed read 0.
retry_base
class-attribute
instance-attribute
¶
RetryPolicy.base stamped at enqueue time, the source crash
and heartbeat reclaim read to reschedule on this job's own curve.
Trailing default: rows materialised before the column existed read
RetryPolicy's own default.
retry_cap
class-attribute
instance-attribute
¶
RetryPolicy.cap stamped at enqueue time.
retry_backoff
class-attribute
instance-attribute
¶
RetryPolicy.backoff stamped at enqueue time.
retry_jitter
class-attribute
instance-attribute
¶
RetryPolicy.jitter stamped at enqueue time.
assignment_routed
class-attribute
instance-attribute
¶
Whether dispatch routes this row by its actor's stored assignment
rather than by its own queue label. Producer-placed rows are
False (the label governs, so an explicit enqueue(queue=...)
and a stale producer's post-move enqueue both stay where they were
put); every re-pend path sets it True, so a row handed back to
the fleet follows the actor's current queue instead of stranding on
one the operator has retired. See the routing contract in
taskq/backend/_dispatch_sql.py. Trailing default: rows
materialised before the marker existed read producer-placed.
claim_epoch
class-attribute
instance-attribute
¶
The row's non-saturating claim-epoch fence: bumped by exactly 1 on
every successful dispatch claim, left untouched by the reclaim sweeps
that clear locks, and fenced on by equality (against the writer's own
claim view) by every terminal/ownership write that fences on
attempt. Trailing default: rows materialised before the column
existed read 0, the one epoch no claim can ever stamp (the first
claim of any row stamps 1). The invariant lives in
01.00.18_02_pre_claim_epoch.sql.
archived
class-attribute
instance-attribute
¶
The row was read from the archive tier, not the hot jobs table.
Set only by a read that fell back to the archive tier (issue #314):
the hot table's rows always read False, and so does every row
from a read that did not probe the archive. Both backends' get
applies the fallback (PG probes jobs_archive, the twin probes
its prune simulation's archive), so an archived id answers with the
marker on both instead of reading as missing. A terminal job's row
outlives its prune only here, so without the marker a caller cannot
tell "this job finished and was archived" from a fabricated row.
Trailing default: rows materialised before the marker existed read
hot.
parent_id
class-attribute
instance-attribute
¶
The fan-out ledger (LIB-2): the enqueuing parent's job id when this row is a child enqueue made under a parent's context (the sibling stamp to tag inheritance, set at worker entry). Plain column, no foreign key — a purged parent's pending children are a defined, harmless state (see 01.00.23_01_pre_jobs_parent_id.sql). Trailing default: rows materialised before the column existed read None, as does every row enqueued outside a parent context.
JobSortField ¶
Bases: Enum
Sort ordering for :meth:Backend.list_jobs via :attr:JobFilter.order_by.
SCHEDULED_AT_ASC (and the default None) preserve the canonical
dispatch-friendly ordering, priority DESC, scheduled_at ASC, id ASC ,
so existing list_jobs callers see no behaviour change.
CREATED_AT_DESC and FINISHED_AT_DESC serve "latest run by business
key" queries: newest-created first and most-recently-finished first
(NULLS LAST) respectively.
Every ordering pages with a cursor. Each one orders id with its
primary column rather than against it -- created_at DESC, id DESC,
not created_at DESC, id ASC -- which is what makes the page seam a
single row-wise comparison. id is UUIDv7 and therefore
time-ordered, so running it with a timestamp column reorders nothing
in practice. The ordering and the cursor shape it pages with are one
object, :class:~taskq.backend._cursor.JobOrdering; a cursor encodes
the columns of the ordering it was produced under, so cursors are not
interchangeable between orderings.
ScheduleRecord ¶
Bases: BaseModel
Read-only snapshot of a cron schedule row from the database.
model_config = ConfigDict(frozen=True) enforces immutability per
public API discipline.
Clock ¶
Bases: Protocol
Time abstraction injected into Backends.
now() returns wall-clock UTC; monotonic() returns a
monotonically non-decreasing float suitable for local elapsed-time
deltas (e.g. cancel-phase tracking in).
SystemClock
dataclass
¶
Production clock delegating to the standard library.
now() calls datetime.now(UTC) (timezone-aware).
monotonic() calls time.monotonic().
BatchCompletionStatus ¶
Bases: BaseModel
Aggregated completion counts for a batch of jobs.
pending counts jobs still in flight (pending, scheduled,
or running status). is_complete is True when all jobs
have reached a terminal status.
BatchHandle ¶
Bases: BaseModel
Handle to a group of jobs inserted by a single
:meth:~taskq.client.JobsClient.enqueue_batch call.
Invariant: job_handles contains one
:class:~taskq.client.JobHandle per item in the original list
(including idempotency-key collisions that returned existing rows).
When a finalizer was enqueued, the finalizer handle is appended as
the last entry of job_handles AND set separately as
finalizer_handle. size is the number of non-finalizer items
(i.e. len(job_handles) - (1 if finalizer_handle is not None else 0)).
:meth:status queries the database for the current completion
counts of the batch.
job_handles
instance-attribute
¶
List of :class:~taskq.client.JobHandle instances, one per enqueued item.
finalizer_handle
class-attribute
instance-attribute
¶
The :class:~taskq.client.JobHandle for the finalizer job, or None
when no finalizer was enqueued. When set, the finalizer handle is also
appended as the last entry of :attr:job_handles for backward compat.
status
async
¶
Query live completion counts for all jobs in this batch.
Uses a JSONB containment query against the metadata column so
the jobs_metadata_gin_idx GIN index is used (@> is
supported by jsonb_path_ops). The query groups by status in a
single round-trip.
schema must match the schema used when the :class:PostgresBackend
was constructed (default "taskq").
Source code in src/taskq/batch.py
BatchSummary
dataclass
¶
BatchSummary(
batch_id: UUID,
queue: str,
status: BatchStatus,
expected_size: int,
consecutive_failures: int,
failure_threshold: int | None,
finalizer_job_id: UUID | None,
originating_actor: str | None,
created_at: datetime,
completed_at: datetime | None,
completion: BatchCompletionStatus,
)
One row from the batches table, augmented with live job counts.
EnqueueItem ¶
Bases: BaseModel
One item in a :meth:~taskq.client.JobsClient.enqueue_batch call.
actor_ref is an :class:~taskq.actor.ActorRef for any payload and
result type. payload is the Pydantic model that will be
serialized into the job row, it is validated by the actor's
payload_type inside :meth:~taskq.client.JobsClient.enqueue_batch
before any INSERT.
metadata is merged with the library-injected batch_id key
before the row is written; callers MUST NOT set metadata.batch_id
manually.
AbortBatchAfter
dataclass
¶
Bases: BatchFailurePolicy
Abort the batch after consecutive_failures consecutive failures.
should_abort(n) returns True when n >= consecutive_failures.
failure_threshold is set to consecutive_failures in
__post_init__ so the client can read it polymorphically via
policy.failure_threshold.
Running jobs are NOT cancelled by the abort, only pending and scheduled jobs are cancelled. Running jobs continue to completion. This matches the post-terminal-write hook design: the hook runs after the terminal write, so a job that was dispatched before the abort triggered will run to completion.
__post_init__ ¶
BatchFailurePolicy
dataclass
¶
Abstract base class for batch failure policies.
Subclasses implement :meth:should_abort to decide whether a batch
should be aborted given the current count of consecutive failures.
failure_threshold is the consecutive-failure count at which the
batch should be aborted. None disables abort (the default on the
base class). :class:AbortBatchAfter sets this to its
consecutive_failures value in __post_init__; custom policies
should set it to their computed threshold. The client reads
failure_policy.failure_threshold polymorphically, no
isinstance check is needed.
BackpressureSnapshot ¶
Bases: BaseModel
The typed read of the submit path's backpressure state.
Returned by JobsClient.backpressure() / TaskQ.backpressure().
One :class:QueueBackpressure entry per requested queue;
parent_id is the fan-out parent whose pending children the
verdicts include (explicit or ambient — the worker's parent
context), None when no parent is in play.
AS-OF semantics, carried explicitly (a snapshot that does not say
how old it is would be a lie by omission): :attr:as_of is the
wall-clock instant the read was taken; the depth half was EXACT at
that instant but is advisory from it on (the race between check and
later enqueue is acceptable, documented); the cap half is the
capacity snapshot named by :attr:cap_age_seconds, which is stale
by up to the cache's TTL (default 5s) by design.
ADVISORY semantics, documented and bounded: the read is a check and
the enqueue is a later act — the race between them is acceptable
(the signal predicts admission, it does not reserve it). The depth
half is one indexed aggregate per call; the cap half is the
ActorCapacityCache TTL snapshot.
as_of
instance-attribute
¶
UTC wall clock at the read. The whole snapshot is "as of" this
instant; the depth half was exact here, the cap half was already up
to :attr:cap_age_seconds old.
cap_age_seconds
class-attribute
instance-attribute
¶
How old the capacity snapshot the caps came from was, in seconds
(monotonic-clock delta at read time). Bounded by the cache's TTL
(default 5) when a snapshot exists; None when no snapshot
exists — then every cap half reads unknown with its reason.
BulkCancelResult ¶
Bases: BaseModel
Structured outcome of a bulk cancellation request.
Returned by JobsClient.cancel_where() so callers can inspect
how many jobs were cancelled directly (pending/scheduled → terminal
'cancelled') vs how many had cooperative cancel requested (running →
cancel_phase=1).
cancelled_directly
instance-attribute
¶
Count of pending/scheduled jobs moved straight to terminal 'cancelled'.
cancel_requested
instance-attribute
¶
Count of running jobs with cancel_phase=1 set (cooperative cancel).
cancelled_ids
instance-attribute
¶
IDs of jobs cancelled directly (pending/scheduled → cancelled).
cancel_requested_ids
instance-attribute
¶
IDs of running jobs with cancel requested.
CancelResult ¶
Bases: BaseModel
Structured outcome of a cancellation request.
Returned by JobsClient.cancel() so callers can inspect whether
the cancellation was initiated and what the status transition was.
JobEvent ¶
Bases: BaseModel
A single event yielded by :meth:TaskQ.stream.
Represents a point-in-time snapshot of a job's observable state.
Yielded on every status transition or progress update; the final
event always has terminal=True.
The progress_state and progress_seq fields reflect the last
values written by the worker. They are None / 0 until the
worker emits a progress update.
Serialises cleanly to JSON via model_dump() for SSE or WebSocket
fanout, fields are deliberately flat so the caller can forward the
event without transformation. Both transports below route through
orjson (:func:taskq._json.dumps), never stdlib json::
# HTTP: drop-in replacement for starlette's stdlib-json JSONResponse
OrjsonJSONResponse = orjson_response_class()
@app.get("/tasks/{job_id}/events")
async def events(job_id: UUID) -> Response:
event = latest_event(job_id)
return OrjsonJSONResponse(event.model_dump())
# WebSocket: send_json uses stdlib json.dumps internally, send the
# pre-rendered orjson text instead:
async for event in tq.stream(job_id):
await websocket.send_text(event.model_dump_json())
JobHandle ¶
JobHandle(
*,
row: JobRow,
result_adapter: TypeAdapter[R],
was_existing: bool,
client: JobsClient | None = None,
backend: Backend | None = None,
_redis_client: Redis | None = None,
_settings: TaskQSettings | None = None,
)
Typed handle to a single enqueued job.
Created by :class:JobsClient methods (:meth:~JobsClient.enqueue,
:meth:~JobsClient.get) or by :class:SubJobEnqueuer (with
backend= only). The type parameter R flows from the
actor's declared return type through :class:ActorRef into this
handle: JobHandle[OrderResult] for an actor returning
OrderResult, JobHandle[None] for fire-and-forget actors.
At least one of client or backend must be supplied. When
client is provided, _backend is filled from
client.backend. When only backend is provided, the four
read-back methods (:meth:status, :meth:refresh,
:meth:attempts, :meth:cancel) raise :class:RuntimeError
because they require the client's higher-level coordination.
:meth:wait always works (it reads through _backend directly).
Why result_adapter: TypeAdapter[R] is a constructor arg: pyright
only infers R for a generic class when the type parameter
appears in at least one field or method signature. The adapter is
that field, without it R would be phantom and inference would
silently fall back to Unknown.
Source code in src/taskq/client/_handle.py
deduplicated_onto_terminal
property
¶
This enqueue deduplicated onto a job that had already finished.
was_existing cannot carry this on its own: it is True for
every dedup, and the overwhelmingly common dedup, onto a live
pending or running job, is the mechanism working, the whole
point of an identity key. The case that needs a signal is the
rare one where the match was a job that already reached a
terminal state: no worker will pick that work up, and with a long
dedup horizon the caller can wait for the rest of the window
before anyone notices, because the strand looks exactly like the
success until the work was needed.
Decided from the row the creating call handed back, so learning
this costs no second round trip, and frozen there rather than
re-read from the live row, because it is a verdict about the
enqueue, not about the job's status now: a dedup onto a running
job that later finishes was never a strand. It pairs with the
WARN-level enqueue_deduplicated line the backend emits
for the same hit: that one is the operator's signal, this one is
the caller's.
archived
property
¶
The last observed row came from the archive tier.
Mirrors :attr:row's last-observed discipline: it reads
:attr:row's archived flag, so it is True exactly when
the job's row now lives in jobs_archive (the hot row was
pruned), and it advances with every :meth:refresh /
:meth:status / :meth:wait fetch like the row itself.
Constructing from an enqueue always starts False.
row
property
¶
The last :class:JobRow this handle observed.
Seeded at construction with the row the creating call fetched,
and advanced by every successful row fetch through the handle ,
:meth:refresh, :meth:status, and :meth:wait's polling loop
each record the row they just read. :meth:progress_stream
does not advance it, its Redis path fetches no rows, and
advancing only on the PG fallback would make the semantics
backend-dependent. Reading the property costs no backend round
trip: handle = await tq.get(id) followed by handle.row
is the single-read pattern for full row state, and a
long-lived handle's row stays current as its owner
refreshes or waits.
The row is returned by reference (it is frozen and backends hand the handle an isolated row); repeated reads between fetches return the same object.
status
async
¶
Return the current status of this job (live read).
Cheap, non-blocking: a single backend.get and a status
projection. No polling. Use this when you want to know the
state without waiting for a terminal transition. Advances
:attr:row to the fetched row. An id whose row the prune moved
to jobs_archive answers from there (its terminal status,
:attr:archived True); only an id that exists in neither
tier raises.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
this handle was constructed without a
:class: |
KeyError
|
the job exists in neither |
Source code in src/taskq/client/_handle.py
refresh
async
¶
Re-read the row from the backend and return the raw
:class:JobRow.
Useful for callers that want full row state (timestamps,
attempt counts, error metadata) without going through
:meth:wait. Does not block on terminal state, returns the
current row whatever its status. Advances :attr:row to the
fetched row, so after a refresh handle.row and the return
value are the same row. An id whose row the prune moved to
jobs_archive refreshes from there (archived=True on the
row); only an id that exists in neither tier raises.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
this handle was constructed without a
:class: |
KeyError
|
the job exists in neither |
Source code in src/taskq/client/_handle.py
attempts
async
¶
Return the attempt rows for this job, ordered by attempt number.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
this handle was constructed without a
:class: |
Source code in src/taskq/client/_handle.py
cancel
async
¶
Delegate to :meth:JobsClient.cancel.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
this handle was constructed without a
:class: |
KeyError
|
the job does not exist (never enqueued, or pruned past retention); the delegate's contract, a typo'd id and a pruned job are indistinguishable here. |
Source code in src/taskq/client/_handle.py
wait
async
¶
Block until the job reaches a terminal status, then return R.
Returns the actor's return value, validated through
:attr:result_adapter. The result type is R exactly ,
never R | None. Missing or failed results raise. Advances
:attr:row to each row the polling loop fetches, on return,
the terminal row the result was extracted from. A job whose row
the prune moved to jobs_archive waits on (and returns) the
archived terminal row (:attr:archived True); only an id
that exists in neither tier raises KeyError.
Raises:
| Type | Description |
|---|---|
ResultUnavailable
|
terminal state reached but no result was
stored (result TTL expired, actor returned |
JobFailed
|
the job ended in a non-success terminal state
( |
KeyError
|
the job exists in neither |
TimeoutError
|
|
Source code in src/taskq/client/_handle.py
progress_stream
async
¶
Stream live progress events for this job.
When Redis is configured, subscribes to the per-job Redis pub/sub
channel and yields :class:~taskq.progress.ProgressEvent objects in
real time. When Redis is not available, falls back to polling Postgres
at 500 ms intervals (jittered ±20%) and synthesising events from row
diffs; a pool or connection error on a poll is retried on the next
one.
Raises :class:NotImplementedError when the backend does not
support progress streaming (the in-memory backend sets its
supports_progress_stream capability to False, it has no
pub/sub), and :class:~taskq.exceptions.StreamUnavailable when
the row could not be re-read for 30 s straight — the Postgres
poll's own reads, and on the Redis path the safety-net re-reads
behind the subscription (an infra error out of a re-read is
retried inside a 30 s budget, one warning per failure; a
non-infra error propagates untouched).
The first event on the Redis path is the row as it stands when
the stream is opened — the same contract :meth:TaskQ.stream
documents: an already-terminal job yields its terminal event and
returns instead of waiting on a channel that will never speak
again, and a non-terminal row carrying content (progress has
moved, or the status moved past the row the handle was
constructed with) is replayed as the first event, so a fresh
consumer — a re-attach after a transport cut — does not starve
until the actor's next publish. A row with nothing to replay
(progress_seq == 0 and no status change) yields nothing
extra.
The stream terminates: every poll timeout re-reads the row as a
safety net (a terminal write whose pub/sub message was dropped —
Redis pub/sub is fire-and-forget — still ends the stream), and a
terminal=True event ends it as on the PG path. The one case
that never terminates on the Redis path is a job id that reads
back as missing: the stream cannot know the job's state, so it
falls through to the subscription (the PG fallback warns and ends
instead).
Does not advance :attr:row, the Redis path fetches no rows
through the subscription (the snapshot read and the on-timeout
safety net below read through _backend directly without
recording), and advancing only there would make the semantics
backend-dependent.
Source code in src/taskq/client/_handle.py
371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 | |
JobsClient ¶
JobsClient(
backend: Backend,
*,
clock: Clock | None = None,
settings: TaskQSettings | None = None,
capacity_cache_ttl: float = DEFAULT_CAPACITY_CACHE_TTL,
)
Public API for job operations.
Delegates to the injected :class:Backend and wraps results in
typed :class:JobHandle[R] instances. The client owns the
payload-serialization step that turns a typed P into the
dict[str, object] carried by :class:EnqueueArgs; the backend
sees only erased payloads.
Source code in src/taskq/client/_jobs.py
backend
property
¶
The underlying :class:Backend this client delegates to.
Exposed so :class:JobHandle can read the backend through the
client without accessing the private _backend attribute.
close
async
¶
invalidate_actor_capacity_cache ¶
Drop the cached actor_config.max_pending snapshot.
The next enqueue refreshes from the backend instead of waiting
out the TTL. Not needed in normal operation (staleness is
bounded by capacity_cache_ttl, default 5s); intended for
tests and for tooling that knows it just changed the table and
cannot wait out the TTL.
Source code in src/taskq/client/_jobs.py
enqueue
async
¶
enqueue(
ref: ActorRef[P, R],
payload: P,
*,
queue: QueueName | None = None,
allow_unregistered: bool = False,
scheduled_at: datetime | None = None,
priority: int | None = None,
schedule_to_close: datetime | None = None,
start_to_close: timedelta | None = None,
heartbeat_timeout: timedelta | None = None,
identity_key: IdentityKey | None = None,
fairness_key: str | None = None,
idempotency_key: IdempotencyKey | None = None,
idempotency_scope: str | None = None,
trace_id: str | None = None,
span_id: str | None = None,
metadata: dict[str, object] | None = None,
tags: list[str] | None = None,
) -> JobHandle[R]
Enqueue a job for the given actor and return a typed handle.
The payload is serialized through ref.payload_type so the
EnqueueArgs.payload carried over the backend boundary is a
plain dict[str, object] ready for the JSONB column. The
returned :class:JobHandle[R] carries ref.result_adapter so
:meth:JobHandle.wait can validate the stored result back to
R.
The actor's name is NOT validated against the registration table
here: any :class:~taskq.actor.ActorRef is accepted, and a ref
whose actor no worker fleet declares enqueues successfully and
then parks (every worker that claims it snoozes it back with
released_reason: "actor-not-found", budget-free; the
stranded-jobs detector surfaces it). If "my job never runs",
check the actor spelling against the worker's registry and
taskq job show's released reason before looking at queues or
capacity. This is the same defer-to-fire-time doctrine
:meth:create_schedule documents.
The metadata.singleton key is reserved by the library for
singleton enforcement. When ref.singleton is True the
library unconditionally writes metadata.singleton = True,
overriding any caller-supplied value. Callers MUST NOT set
metadata.singleton manually.
max_pending:
-
When the actor's effective
max_pendingis set, a pre-flight count ofpending+scheduledjobs for the actor is compared to the limit. Ifcount >= max_pending, :class:MaxPendingExceededErroris raised synchronously, the caller decides whether to retry, fail, or wait; the library does not block on capacity. -
The effective limit is operator-owned: a non-NULL stored
actor_config.max_pending(set viataskq actor-config set --max-pending) wins over the@actor(max_pending=...)literal; a cleared or absent stored value falls back to the literal. The client reads the stored value through a TTL-bounded cache (default 5s staleness; see :class:taskq.client._capacity.ActorCapacityCache), so an operator change takes effect fleet-wide within seconds without any redeploy or restart. -
Evaluation order at enqueue:
unique_fordedup → singleton pre-flight →max_pendingcount check →idempotency_keyINSERT → job INSERT. Aunique_forhit bypasses all remaining checks; a singleton collision fires beforemax_pendingto give the caller the more specificSingletonCollisionError. -
idempotency_keydoes not bypassmax_pending, the idempotency ON CONFLICT fires at step 5, after the max_pending check at step 3. Re-enqueuing with a duplicateidempotency_keywhen the queue is full raisesMaxPendingExceededError, not the deduplicated handle. Onlyunique_for(step 1) bypasses max_pending.
idempotency_key:
-
idempotency_keyis unique within itsidempotency_scope(composite(idempotency_scope, idempotency_key)uniqueness). The default scope (idempotency_scope=Noneor"") preserves the prior global-until-prune behavior exactly, so existing callers see zero behavior change. Passing an explicit scope (e.g. a run/batch/epoch id) lets two enqueues with the same business key in different scopes both succeed, decoupling the dedupe horizon fromprune_retention_*. -
Key length is bounded at
idempotency_key_max_bytes(TASKQ_IDEMPOTENCY_KEY_MAX_BYTES, default 1024 UTF-8 bytes) , the bound is the composite unique index's btree entry size, not a round number. Empty and whitespace-only keys raise :class:ValueErrorat the client boundary before any backend call. The same bound applies toidempotency_scope; an empty scope ("") is valid and equivalent toNone(the default/global scope). -
No time-based (TTL) dedupe window.
idempotency_scopedecouples the dedupe horizon fromprune_retention_*by namespace, not by time, there is noidempotency_ttlor equivalent "dedupe for the next N seconds" parameter. A key within a given scope still dedupes until pruned, exactly like the pre-scope global behavior, just scoped to that namespace. This is a deliberate scope decision, not an oversight: a real sliding-window TTL cannot be expressed as a single static unique index the way scope can, a sliding window would require either abandoning the atomicINSERT ... ON CONFLICTfor a check-then-insert lock (weaker concurrency guarantee) or encoding time-bucketing into the key itself (coarser, edge-artifact-prone semantics). If your use case genuinely needs "dedupe for the next hour, not forever," encode the window into the scope yourself (e.g. a time-bucketed scope string) until/unless a TTL parameter ships as a separate feature. -
Rolling-deploy note: if this schema is mid-upgrade (the
01.00.03_01_pre_idempotency_scope.sqlmigration applied but01.00.03_01_post_idempotency_scope_drop_old_index.sqlnot yet applied), reusing the sameidempotency_keyunder two differentidempotency_scopevalues raises :class:~taskq.exceptions.ScopedIdempotencyMigrationPendingErrorrather than silently dedupe against the wrong scope's job. The trigger is a key existing under a different scope, in either direction, an unscoped call reusing a key first written under a non-default scope raises it too. Only brand-new keys and same-scope repeats are unaffected. See that exception's docstring and the migration file's header comment for the full rationale.
unique_for:
-
unique_fordeduplication is serialized, not best-effort: the preflight and the INSERT run under one transaction-scoped advisory lock on(actor, identity_key), so concurrent enqueues of the same identity produce exactly one job and every caller is handed that job. This holds on pool connections and on a caller-supplied connection with no open transaction. On a caller-owned OPEN transaction the lock spans that transaction instead, so single-flight holds until its commit/rollback, a long-lived caller transaction can exhaust another same-identity enqueue's bounded wait (:class:~taskq.exceptions.UniqueForLockTimeoutError) rather than dedup against it. The dispatch CTE'srunning_identitiesfilter remains the execution-level guard behind the enqueue-level one. -
When either dedup mechanism matches an existing job,
JobHandle.was_existingisTrue. This field replaces the need for callers to inspect the row'screated_atto detect a dedup return.
allow_unregistered:
- The per-submit escape from
TASKQ_QUEUES_STRICT: with strict on, a queue no registered actor's assignment routes to raises :class:~taskq.exceptions.UnknownQueueErrorbefore any write;allow_unregistered=Trueskips that verdict for THIS call — for genuinely dynamic queue names (tenant-prefixed, sharded at runtime) that no registration can predict. The unserved-queue NOTE still fires (the diagnostic sits below the gate), so a typo'd name stays visible even on an escaped call. With strict off the flag changes nothing. The batch arms carry no per-item flag: their queues are the items' actors' own declared routes, the thing registration populates; use single enqueues for dynamic names.
Source code in src/taskq/client/_jobs.py
493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 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 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 | |
enqueue_batch
async
¶
enqueue_batch(
items: list[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
failure_policy: BatchFailurePolicy | None = None,
finalizer: EnqueueItem | None = None,
) -> BatchHandle
Enqueue multiple jobs in a single batched INSERT and return a
:class:~taskq.batch.BatchHandle.
All items share a single batch_id UUID written into each
job's metadata.batch_id field (as a string). When
batch_id is not supplied it is auto-generated as a UUIDv7 via
:func:~taskq._ids.new_job_id.
failure_policy:
When failure_policy is set (e.g.
:class:~taskq.batch_policy.AbortBatchAfter), a batches row
is created with the policy's failure threshold. After each child
job reaches a terminal state, the
:func:~taskq.batch.apply_batch_terminal_outcome hook inspects
the outcome: succeeded resets the consecutive-failure
counter; failed increments it and aborts the batch if the
threshold is reached. Aborting cancels all pending and scheduled
child jobs and sets the batch row to aborted.
finalizer:
When finalizer is set, a finalizer job is enqueued alongside
the batch. The finalizer is NOT stamped with batch_id
metadata (deadlock prevention, if it were, wait_for_batch
would count it as a child and the finalizer would wait for
itself). The batch row's finalizer_job_id column records the
link, and wait_for_batch automatically excludes that job
from counts. The finalizer is dispatched immediately; the
in-actor wait_for_batch snooze pattern gates on child-job
completion.
Transactional enqueue:
When failure_policy or finalizer is set and
connection is None, the entire operation (batch row +
all child jobs + finalizer) is inserted in a single transaction
via :meth:Backend.enqueue_batch_atomic. If any insert fails,
no rows are committed. When a connection is provided, the
caller controls the transaction boundary; the batch row and
finalizer are created as the last statements on that connection.
Validation rules:
len(items) == 0raises :class:ValueError.len(items) > MAX_BATCH_SIZEraises :class:ValueError.- ALL payloads are validated before any INSERT. A single failure
raises :class:
~taskq.exceptions.PayloadValidationErrorand leaves no rows inserted.
max_pending:
Admission is partitioned per actor. The backend issues one
aggregated SELECT actor, count(*) … WHERE actor = ANY($1)
GROUP BY actor for the entire batch, resolves per-actor
effective limits (operator-owned stored value when set, else the
@actor(...) literal, same resolution as :meth:enqueue),
and admits every within-cap actor's items; an over-cap actor's
items are refused as a whole group (never partially filled up to
the cap). When any actor is refused, the within-cap actors'
items are still enqueued and
:class:~taskq.exceptions.BatchMaxPendingExceededError raises
afterwards, it names each refused actor, the refused item
indices into items, and the admitted count, so a retry can
target only the refused items (or rely on idempotency_keys
to deduplicate a whole-batch retry). The exception is
deliberately not a :class:~taskq.exceptions.MaxPendingExceededError
subclass: handlers for that type assume nothing was enqueued,
and under this error part of the batch is already stored.
When failure_policy or finalizer is set and
connection is None (the atomic path), a cap violation
keeps the all-or-nothing contract instead: the whole single
transaction rolls back and plain
:class:~taskq.exceptions.MaxPendingExceededError raises with
nothing committed.
On a caller-supplied connection with an open transaction, the admitted items are inserted but their durability follows that transaction's commit/rollback.
unique_for (not applied on batch paths):
Actor-declared unique_for is a single-enqueue contract (see
:meth:enqueue). The batch INSERT writes every item without a
per-identity preflight, the single path's advisory-lock +
preflight round trips are exactly what bulk throughput exists to
avoid, so two batch items with the same
(actor, identity_key) both insert, and the dispatch CTE's
identity serialization ensures only one runs at a time.
unique_for items are conservatively fully counted toward
max_pending (a batch mixing unique_for retries near the
cap may refuse loudly rather than admit silently). Deduplicate
batch items with per-item idempotency_keys, that arbiter IS
applied, below. Whether the batch tier should honor
unique_for is an open design decision; this documents
current behavior.
idempotency_key collisions:
Items whose idempotency_key collides with an existing row
return the existing :class:~taskq.client.JobHandle (same
semantics as single-item :meth:enqueue).
Source code in src/taskq/client/_jobs.py
745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 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 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 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 | |
enqueue_batch_streaming
async
¶
enqueue_batch_streaming(
items: Iterable[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
failure_policy: BatchFailurePolicy | None = None,
finalizer: EnqueueItem | None = None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> BatchHandle
Enqueue jobs from a lazy iterable in chunks, returning a single
:class:~taskq.batch.BatchHandle.
Unlike :meth:enqueue_batch, this method accepts an
:class:~collections.abc.Iterable (including generators) and
inserts in chunks of chunk_size (1-MAX_BATCH_SIZE). All items share
the same batch_id. Payloads are validated on the fly as
each chunk is built.
When failure_policy or finalizer is set and
connection is None, the entire operation is delegated to
:meth:Backend.enqueue_batch_atomic for single-transaction
atomicity. Otherwise, chunks are inserted via
:meth:Backend.enqueue_batch: with a caller-supplied connection
all chunks share that connection's open transaction (one
transaction aggregate, the caller owns the boundary); with NO
connection each chunk is its own pool transaction, so progress
is committed incrementally, chunk by chunk.
No-connection failure surface: when chunk N fails (cap
refusal, payload validation, a driver error), chunks 1..N-1
are already durably committed and nothing is returned, the
BatchHandle is only constructed after the whole stream
drains. A blind retry of the full iterable would duplicate the
committed prefix. Retry safely by giving items
idempotency_keys (a retry deduplicates against the committed
rows) or by resuming from the point of failure.
Failure-policy counting limitation (caller-connection path):
the batch row is created AFTER all chunk inserts, it must carry
the final expected_size and create_batch is INSERT, not
upsert, so a child job enqueued on that connection that
reaches a terminal state BEFORE the row exists is not counted
toward failure_policy: increment_batch_failures finds
no row and returns (0, None, 0). The atomic
(no-connection) path is unaffected, its single transaction
makes the batch row and the child jobs visible together.
max_pending: enforced per chunk with the same per-actor
partition as :meth:enqueue_batch, each chunk's items carry
their actors' effective caps (operator-stored override when
set, else the @actor(...) literal, same resolution as
:meth:enqueue), the backend admits every within-cap actor's
items, and an over-cap actor's items are refused. On the
caller-connection path sequential chunks share one transaction,
so per-chunk admission accumulates to the true aggregate. On
the no-connection path each chunk counts against the rows the
previous committed chunks wrote, which also accumulates to the
true aggregate, but the refusal surfaces after the committed
prefix: :class:~taskq.exceptions.BatchMaxPendingExceededError
raises with STREAM-GLOBAL refused item indices and an
admitted_count covering every committed item; items after
the refusing chunk were never attempted. Retry only the
refused items (or rely on idempotency_keys). The atomic
path (failure_policy/finalizer, no connection) resolves
the same effective caps and keeps all-or-nothing: a cap
violation rolls back the entire single transaction and raises
plain :class:~taskq.exceptions.MaxPendingExceededError with
nothing committed.
unique_for (not applied on batch paths): actor-declared
unique_for is a single-enqueue contract (see
:meth:enqueue_batch's disclosure), every item writes, items
sharing an (actor, identity_key) are not deduplicated, and
unique_for items are fully counted toward max_pending.
This holds on both arms (the per-chunk INSERT and the atomic
path's chunked INSERT). The idempotency_key arbiter IS
applied on both arms: a colliding item returns the existing
row, exactly as :meth:enqueue_batch does.
Source code in src/taskq/client/_jobs.py
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 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 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 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 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 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 | |
get_batch
async
¶
Fetch a single batch row by ID.
Delegates to :meth:Backend.get_batch. Returns None when the
batch does not exist.
Source code in src/taskq/client/_jobs.py
list_batches
async
¶
List batches matching filter, returning :class:BatchSummary objects.
Delegates to :meth:Backend.list_batches and maps each
(BatchRow, BatchCounts) pair to a :class:BatchSummary
with a :class:BatchCompletionStatus derived from the live counts.
Source code in src/taskq/client/_jobs.py
enqueue_batch_fast
async
¶
enqueue_batch_fast(
items: list[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
) -> int
Enqueue jobs via COPY FROM protocol for maximum throughput.
WARNING, bulk-import semantics, not general-purpose enqueue:
this method does NOT detect or
reject idempotency-key collisions (a duplicate key aborts the
whole batch instead of being treated as "already enqueued"), and
returns a bare row count, not per-job handles, there is no
way to await, cancel, or otherwise reference an individual job
from the return value. max_pending IS enforced, with the same
per-actor partition admission and the same effective-cap
resolution (operator-stored override when set, else the
@actor(...) literal) as :meth:enqueue_batch (see its
docstring). Use :meth:enqueue_batch
unless you specifically need COPY-level throughput for a one-shot
bulk import/backfill and have already accounted for these gaps.
Returns the count of inserted rows, no :class:~taskq.batch.BatchHandle,
no :class:~taskq.client.JobHandle instances.
Validation rules:
len(items) == 0raises :class:ValueError.len(items) > 50_000raises :class:ValueError.- ALL payloads are validated before any INSERT, a single failure
raises :class:
~taskq.exceptions.PayloadValidationError.
Tradeoffs vs enqueue_batch:
- No idempotency-key collision handling. A duplicate key
aborts the entire batch with
:class:
~taskq.exceptions.DuplicateIdempotencyKeyError, a same-(idempotency_scope, idempotency_key)pair, whether repeated within the batch or already stored; nothing is written. Callers must pre-deduplicate. One carve-out: during the01.00.03pre→post migration window, a key reused across different scopes raises :class:~taskq.exceptions.ScopedIdempotencyMigrationPendingErrorinstead, matching the other enqueue paths. - No unique_for preflight. Actor-declared
unique_foris a single-enqueue contract (see :meth:enqueue_batch's disclosure): the COPY writes every item, items sharing an(actor, identity_key)are not deduplicated, andunique_foritems are fully counted towardmax_pending, same semantics as the unnest batch tier. - max_pending partition admission. One aggregated count runs
before the COPY: within-cap actors' rows are written, and an
over-cap actor's items are refused, the COPY of the admitted
rows commits first, then
:class:
~taskq.exceptions.BatchMaxPendingExceededErrorraises naming the refused actors and item indices (retry only those, or rely on idempotency keys). - No JobHandle instances. Only the inserted row count is
returned. Use
batch_idto query rows post-insert. - All-or-nothing on constraint violations. The entire COPY fails on any constraint violation (duplicate keys, singleton, CHECK), only cap admission partitions.
Use for bulk import / backfill with 1K-50K rows where throughput matters more than idempotency guarantees.
Source code in src/taskq/client/_jobs.py
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 1662 1663 1664 1665 1666 | |
get
async
¶
Look up a job by id.
Returns None when the job does not exist; otherwise wraps
the row in a :class:JobHandle[R]. The lookup reads the hot
jobs table first and falls back to jobs_archive (the
same jobs-then-archive probe taskq job show applies), so an
id whose row a prune moved to the archive returns that row's
handle with :attr:JobHandle.archived True instead of
reading as missing; None still means "in neither tier".
The caller may supply
result_adapter because lookups by id do not carry actor
identity, typical sources are
my_actor.result_adapter (when reuniting with an actor) or
TypeAdapter(type(None)) (when only row metadata is needed).
When result_adapter is None it defaults to
TypeAdapter(type(None)), which is suitable for status-only
lookups.
Source code in src/taskq/client/_jobs.py
get_row
async
¶
Look up a job by id and return the raw :class:JobRow.
Mirrors :meth:get's contract, one backend.get (hot table
first, jobs_archive fallback, an archive hit marked
archived=True), None when the job exists in neither
tier, without the handle machinery or
result adapter. For callers that never need a
:class:JobHandle, this is the direct form; for the fresh-read
case that does want a handle, prefer get plus the handle's
row property (still a single round trip).
Source code in src/taskq/client/_jobs.py
backpressure
async
¶
Read the submit path's backpressure state for queues.
A read-only typed read (the LIB-2 design, mirroring the
reader-method precedent of :meth:get_row / :meth:list) —
deliberately NOT an enqueue kwarg: enqueue is generic
JobHandle[R] and no flag may flip the reply type.
Per queue (see :class:~taskq.types.QueueBackpressure):
depth— jobs holding a pending slot IN THIS QUEUE (pending + scheduled), the queue-local ops view, ONE indexed aggregate for the whole call (count_pending_jobs_by_queue, thecount_active_jobspattern). The worker-side leader sampler's cache is in-process on the worker and not readable cross-process; the depth here is a real count, not a pretense. REPORTED, not verdict-bearing — the admission cap is per-ACTOR and governs the actor's count across ALL queues, so a queue-local count against an actor cap would false-OK exactly when an actor's traffic splits across queues.binding_actor/admission_load/effective_max_pending— the VERDICT's basis: the routing actor with the smallest headroom (its OWN pending+scheduled count, ALL queues, fromcount_pending_jobs— theenqueue_max_pending_countgrouping — against its stored cap from theActorCapacityCacheTTL snapshot). The data model has no per-queue depth threshold; the boundary is the actor's.children_depth— when a fan-out parent is in play (parent_id, or the ambient worker contextcurrent_parent_id()), the parent's pending children OUTSIDE the queue, exact via theparent_idledger stamp (01.00.23_01). REPORTED, not verdict-bearing: every pending child already counts toward its OWN actor's cap (wherever it sits), which is the number admission enforces — the fan-out pressure reaches the verdict through the routing actors' admission loads, and this field is the parent-attributed ledger view of the same rows. Children already in the queue are in itsdepthand never counted twice.state—overwhen any routing actor's OWN count is at or over its cap (the next enqueue for that actor is likely refused withMaxPendingExceededError),okonly when every routing actor resolves and none is over (an unresolvable actor may be over — claiming ok would be the false-ok),unknownwhen the verdict could not be computed (sick database, the actor-count read missing, no capacity snapshot, a cap that lives only in the@actorliteral).reasonnames the absent half. Fail-open, never a fabricated verdict.
ADVISORY semantics: the race between this check and the later
enqueue is acceptable (the signal predicts admission, it does
not reserve it); the snapshot carries as_of and
cap_age_seconds so its own staleness travels with it (the
cap age is bounded by the TTL, default 5s, and reports the last
SUCCESS, not the last attempt). Cost: one depth aggregate + one
actor-count aggregate per call, plus one more only when a
parent is in play; unknown on backend trouble — the read
never raises except for a missing schema
(:class:SchemaNotMigratedError, the actionable setup defect).
Source code in src/taskq/client/_jobs.py
1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 | |
list
async
¶
List jobs matching filter, returning a :class:JobPage.
filter.status accepts a single :data:JobStatus or a
sequence of statuses (e.g. JobFilter(status=["pending",
"running"])).
filter.unfinished is a meta-filter: unfinished=True selects
non-terminal statuses (pending, scheduled, running, 'not yet
finished', not 'currently executing') and unfinished=False selects
terminal ones. See :class:JobFilter for full semantics.
next_cursor is returned for every ordering, encoded from the
columns that ordering actually sorts by, and is None exactly on
the last page: the backend is asked for one row past the limit, so
a last page that happens to fill the limit is still known to be the
last, and a caller paging until the cursor runs out never fetches an
empty trailing page. filter.limit is capped at
:data:~taskq.backend._protocol.MAX_JOB_LIST_LIMIT (ValueError
above it); at the ceiling there is no room to look past the limit,
so a full page there carries a cursor that may lead to one empty
page.
Source code in src/taskq/client/_jobs.py
cancel
async
¶
Request cancellation of a job and return a :class:CancelResult.
Reads the row first via :meth:Backend.get. If the job does not
exist, raises :class:KeyError (carrying job_id), matching
Python's stdlib idiom for "asked for an entry by id; it isn't
there" — note this means a typo'd id and a pruned job are both
:class:KeyError, never a "not found" result; check existence
with :meth:get first when the distinction matters.
Then calls :meth:Backend.write_cancel_request and reads the
row again to capture the new status. The previous_status
reflects the row at the first read, not atomically at
write-time: the status can move between the two reads (a TOCTOU
window inherent to a request-write protocol), so a handler that
must not race the transition should re-check new_status
rather than assume previous_status.
Metrics: increments taskq.cancellation.requested exactly once
per call, regardless of cancellation_initiated outcome.
Source code in src/taskq/client/_jobs.py
cancel_where
async
¶
cancel_where(
filter: JobFilter,
reason: str | None = None,
*,
allow_empty_filter: bool = False,
) -> BulkCancelResult
Cancel all jobs matching filter in a single set-based operation.
Pending/scheduled jobs are moved straight to terminal 'cancelled'
(no running actor to cooperate with). Running jobs get
cancel_phase=1 set (cooperative cancel), the worker's
heartbeat-driven cancel controller observes the phase change and
sets the in-process cancel_event.
Guardrail: a filter with no predicates (no queue, status,
actor, identity_key, batch_id, tags, active, or created_before)
is rejected with :class:EmptyFilterError unless
allow_empty_filter=True is passed.
Filter fields used: queue, status, actor,
identity_key, batch_id, tags, active,
created_before. The limit, cursor, and order_by
fields are ignored.
Returns a :class:BulkCancelResult with counts and affected IDs.
Source code in src/taskq/client/_jobs.py
create_schedule
async
¶
create_schedule(
actor: str | ActorRef[P, R],
cron_expr: str,
*,
timezone: str = "UTC",
dst_strategy: DstStrategy = "skip",
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
name: str = "",
identity_key: IdentityKey | None = None,
enabled: bool = True,
owner: Literal["code", "operator"] = "operator",
) -> ScheduleHandle
Create a cron schedule. Raises :class:ValueError if both
payload_factory and static_payload are provided, or if
cron_expr is invalid.
owner declares who owns the schedule's enable/disable lifecycle.
"operator" (the default here: a client-created schedule is an
operator act) records operator ownership from birth, so a schedule
created disabled carries disabled_by='operator' and no worker
boot ever reverts it; "code" matches a @cron declaration,
whose stale auto-disable a boot reverts (see
:class:CronScheduleSpec).
The (actor, name) UNIQUE constraint means each (actor, name)
pair may have at most one schedule; a second create_schedule for
the same pair raises asyncpg.UniqueViolationError (PG) or
:class:ValueError (in-memory). Pass distinct name values to run
several cron schedules per actor (e.g. a per-property sync).
When identity_key is set, the cron loop propagates it to cron-fired jobs so they dedup against on-demand jobs for the same business key.
Does NOT validate actor existence at creation time, any string actor name is accepted (validation is deferred to fire time).
The first next_fire_at is seeded from the clock that
arbitrates its due-check: on a Postgres-backed client the PG
server clock is read first (one-row SELECT clock_timestamp()
via the backend's pool, mirroring the worker bootstrap), so
app↔DB clock skew cannot shift the fire chain; on a pool-less
(in-memory) client the seed comes from the client's injected
Clock. This matters permanently: the cron loop's normal path
recomputes every subsequent fire from the STORED fire time
(only a miss beyond cron_catch_up_window re-anchors on the
server clock), so the seed, not any per-tick correction ,
fixes the chain's phase for the schedule's life.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dst_strategy
|
DstStrategy
|
How to handle DST gaps and overlaps.
|
'skip'
|
Source code in src/taskq/client/_jobs.py
2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067 2068 2069 2070 2071 2072 2073 2074 2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088 2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114 | |
list_schedules
async
¶
List cron schedules, optionally filtered by actor or enabled status.
Source code in src/taskq/client/_jobs.py
update_schedule
async
¶
update_schedule(
schedule_id: UUID,
*,
cron_expr: str | None = None,
enabled: bool | None = None,
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
clear_payload_factory: bool = False,
) -> ScheduleRecord
Update a cron schedule. Setting enabled=True clears
last_fire_error and resets consecutive_failures to 0.
Raises :class:ValueError if both payload_factory and
static_payload are provided, or if cron_expr is invalid.
To explicitly clear payload_factory (set the column to NULL),
pass clear_payload_factory=True, None for payload_factory
means "don't change this field."
When cron_expr changes, the recomputed next_fire_at is
seeded from the same clock as create_schedule (the PG
server clock on Postgres-backed clients; the client's injected
Clock in-memory), the stored chain keeps its server-anchored
phase.
Source code in src/taskq/client/_jobs.py
2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 | |
delete_schedule
async
¶
QueueBackpressure ¶
Bases: BaseModel
One queue's backpressure verdict, the entry type of
:class:BackpressureSnapshot.
All depth/cap halves are None-able on purpose: None always
means "this half could not be seen", and any None half forces
the unknown state. A verdict is only ok/over when every
half it compares was actually read.
depth
class-attribute
instance-attribute
¶
Jobs holding a pending slot in THIS queue (pending + scheduled) —
the queue-local, ops view. NOT the verdict's basis: the admission
cap governs the ACTOR's count across ALL queues (see
:attr:admission_load); a queue-local count against an actor cap
would false-OK exactly when an actor's traffic splits across queues
(the F1 review finding). None: the depth read was unavailable.
children_depth
class-attribute
instance-attribute
¶
The fan-out parent's pending children OUTSIDE this queue — the
exact parent_id ledger, REPORTED not verdict-bearing: every
pending child already counts toward its OWN actor's cap, which is
the number the admission check enforces, so the fan-out pressure
reaches the verdict through the routing actors' admission loads
without a second copy of it here. None: no fan-out parent in
play, or the children read was unavailable.
binding_actor
class-attribute
instance-attribute
¶
The routing actor whose admission state binds the verdict: the
smallest headroom (stored cap minus its own pending+scheduled count)
among the actors whose stored assignment routes this queue.
None: no cap resolved (then state is unknown).
effective_max_pending
class-attribute
instance-attribute
¶
:attr:binding_actor's stored cap — the tightest effective
admission boundary for this queue. None: no cap resolvable from
this process (no snapshot, no routing row, or the routing actor
enforces only the @actor literal — code-side, invisible here).
admission_load
class-attribute
instance-attribute
¶
:attr:binding_actor's pending+scheduled count across ALL queues
— the number enqueue_max_pending_count actually compares against
its cap. The verdict is over exactly when this is at or above
:attr:effective_max_pending. None: the count read was
unavailable.
reason
class-attribute
instance-attribute
¶
Set exactly when state == 'unknown': what could not be seen.
TaskQ ¶
TaskQ(
*,
dsn: str | None = None,
pool: Pool | None = None,
pool_factory: PoolFactory | None = None,
pg_provider: PgCredentialProvider | None = None,
schema: str | None = None,
min_pool_size: int = 1,
max_pool_size: int = 5,
redis_url: str | None = None,
redis_client: Any | None = None,
pg_conn_factory: ConnFactory | None = None,
listen_conn: Connection | None = None,
poll_timeout: float = 30.0,
reclaim_event_visibility_delay: timedelta | None = None,
reload_interval: float | None = None,
)
Postgres-backed TaskQ client.
Manages a connection pool and exposes job operations directly. Supports
both the async context manager pattern and explicit open() / close()
for frameworks like FastAPI that manage their own lifecycle.
Parameters¶
dsn:
Postgres DSN string. Mutually exclusive with pool.
pool:
An already-open asyncpg.Pool. The caller retains ownership;
close() will not close it.
pool_factory:
A zero-arg async factory returning an asyncpg.Pool - the same
:data:~taskq.connections.PoolFactory the worker takes via
:class:~taskq.connections.WorkerConnections. TaskQ invokes it at
open() and owns the result (close() closes it), and
:meth:reload_credentials re-invokes it to rotate the pool in place.
This is the client-side equivalent of a <role>_pool_factory:
pair it with :func:taskq.auth.make_pg_pool_factory for a
rotating-credential deployment. Mutually exclusive with dsn and
pool.
pg_provider:
A :class:~taskq.auth.PgCredentialProvider. Sugar for
pool_factory=make_pg_pool_factory(dsn, pg_provider,
min_size=min_pool_size, max_size=max_pool_size) - it is exactly
that call and nothing more, so the two paths share one mechanism.
Requires dsn; use pool_factory directly when you need the
factory's other hooks (init, server_settings,
command_timeout).
schema:
TaskQ schema name. When omitted, resolved the way the worker and
CLI resolve it, TASKQ_SCHEMA_NAME from the process environment
or the .env cascade, with "taskq" as the final default, so
a client constructed with no opinion lands in the schema its worker
fleet is listening on. An explicit value always wins.
min_pool_size:
Minimum pool connections. Only used when dsn or pg_provider
is provided (both build the pool through the same factory path;
a caller-supplied pool / pool_factory is used as-is).
max_pool_size:
Maximum pool connections. Only used when dsn or
pg_provider is provided.
redis_url:
Redis URL string. Mutually exclusive with redis_client.
The library creates and owns the Redis client; close() will
close it.
redis_client:
An already-open redis.asyncio.Redis client. The caller retains
ownership; close() will not close it. Mutually exclusive with
redis_url.
pg_conn_factory:
A zero-arg async factory returning an asyncpg.Connection for the
LISTEN/NOTIFY transport used by :meth:watch_reclaims. Mutually
exclusive with listen_conn. Takes precedence over dsn when
set. Use this when you have no DSN (e.g. AAD-managed-identity auth)
but still want the reclaim wake-up. TaskQ owns and closes the
connection produced by the factory per watch_reclaims() call.
listen_conn:
A pre-constructed asyncpg.Connection for the LISTEN transport.
Caller-owned; TaskQ does not close it. Mutually exclusive with
pg_conn_factory. Takes precedence over dsn when set. Use
this to share a dedicated LISTEN conn across callers.
poll_timeout:
Maximum seconds to wait between transport wakeups before re-fetching
job state. Defaults to 30.0. :meth:stream on Postgres alone
has no wakeups and polls every min(poll_timeout, 0.5) seconds
(never below 0.1 s, jittered ±20%).
reload_interval:
Seconds between automatic :meth:reload_credentials rebuilds of a
factory-built pool (pool_factory / pg_provider); the
client-side TASKQ_RELOAD_INTERVAL. When None (the default)
the cadence is derived from the lease the provider grants - a
Vault dynamic credential is rebuilt at half its lease TTL (see
:class:taskq.auth.ReloadSchedule) - and a factory whose
credential carries no lease is never rebuilt automatically
(token providers refresh per connection and need no rebuild).
The rotation runs as a background task for the life of the client;
a rebuild that fails leaves the live pool serving and is retried
on the next tick. Rejected for dsn= and pool= (a
ValueError at construction), which have nothing to rebuild
from — it is not silently ignored, the docstring's earlier claim.
reclaim_event_visibility_delay:
Overrides the margin :meth:watch_reclaims assumes between a
job_events writer's INSERT and its COMMIT (see
:data:taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY, default
2s). It is passed through to the constructed backend's
poll_reclaim_events and bounds the NOTIFY catch-up retry.
Must match whatever margin the worker fleet's sweep-adjacent
backend uses: the margin's correctness depends on writer
transaction duration, not reader preference, so a client that
widens it alone only silences the gap detection on itself.
None (the default) uses the shipped constant.
Source code in src/taskq/client/_taskq.py
522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 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 | |
actors
property
¶
Actor configuration client, list, get, set capacity, deregister.
Raises RuntimeError if called before open() or outside an
async with block.
backend
property
¶
The Postgres :class:Backend this client built, read-only.
The seam a host that embeds the admin UI needs: create_router
takes the backend verbatim (backend=tq.backend) so the admin
portal's backend-mediated mutations (job cancel, job retry,
schedule run-now) serve from the SAME backend the trigger app
enqueues through, instead of the 503 "not configured" the router
answers when it has none. Raises RuntimeError if called before
open() or outside an async with block, like the other
accessors.
open
async
¶
Open the connection pool and prepare the client.
Called automatically by __aenter__. Safe to call explicitly
for frameworks that manage lifecycle outside an async with block.
Raises :class:RuntimeError if already open.
Source code in src/taskq/client/_taskq.py
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 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 | |
close
async
¶
Close the client and release the pool if owned.
Called automatically by __aexit__. Safe to call explicitly.
No-op if already closed.
Source code in src/taskq/client/_taskq.py
reload_credentials
async
¶
Rebuild the pool from pool_factory and swap it in, live.
The client-side counterpart of
:func:taskq.worker.deps.reload_credentials, and the supported
replacement for reaching into _pool / _client: it invokes the
factory (which fetches a fresh credential), atomically points every
subsystem that holds a pool - the backend deps behind
enqueue/get/list/cancel, the :attr:actors client, and
the stream() poll - at the new pool, then closes the old
one with the same bounded drain used at close().
A username-bearing pair (a Vault lease) is rotated automatically on
the client's :class:~taskq.auth.ReloadSchedule - reload_interval
when given, otherwise half the granted lease TTL - so this is the
on-demand path: call it on any signal your issuer gives you, or to
drop sessions opened under a revoked token. It is not needed for
the ordinary token refresh that :func:taskq.auth.make_pg_pool_factory
already does per physical connection. Concurrent calls (the
scheduled rotation and an explicit one) are serialized.
Raises :class:RuntimeError if the client is not open, or if the pool
is caller-owned (pool=) - TaskQ must never close a pool it does not
own, so the caller rotates that one itself.
If the factory fails, the exception propagates and the current pool is left untouched and serving: a transient token-endpoint outage must not turn into a client outage.
Source code in src/taskq/client/_taskq.py
__aenter__
async
¶
__aexit__
async
¶
enqueue
async
¶
enqueue(
ref: ActorRef[P, R],
payload: P,
*,
queue: QueueName | None = None,
scheduled_at: datetime | None = None,
priority: int | None = None,
schedule_to_close: datetime | None = None,
start_to_close: timedelta | None = None,
heartbeat_timeout: timedelta | None = None,
identity_key: IdentityKey | None = None,
fairness_key: str | None = None,
idempotency_key: IdempotencyKey | None = None,
idempotency_scope: str | None = None,
trace_id: str | None = None,
span_id: str | None = None,
metadata: dict[str, object] | None = None,
tags: list[str] | None = None,
) -> JobHandle[R]
Enqueue a job and return a typed handle.
The actor's name is NOT validated against the registration table
at enqueue time (any ActorRef is accepted, the same
defer-to-fire-time doctrine create_schedule applies): a
ref whose actor no worker declares enqueues successfully and the
job is parked at the snooze cadence by every worker that claims
it (released_reason: "actor-not-found", the stranded-jobs
detector surfaces it) rather than being refused here. See
:meth:JobsClient.enqueue for the full enqueue contract:
dedup, max_pending, and idempotency semantics.
schedule_to_close (absolute datetime) is deprecated, it crosses
clock domains (the app clock that produced it vs the database clock
that evaluates it). Declare retry.time_budget on the actor
instead; the interval form is anchored to the database clock.
Source code in src/taskq/client/_taskq.py
enqueue_batch
async
¶
enqueue_batch(
items: list[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
failure_policy: BatchFailurePolicy | None = None,
finalizer: EnqueueItem | None = None,
) -> BatchHandle
Enqueue multiple jobs in a single batched INSERT.
Delegates to :meth:JobsClient.enqueue_batch; see its docstring
for validation rules and idempotency-key collision semantics.
Source code in src/taskq/client/_taskq.py
enqueue_batch_streaming
async
¶
enqueue_batch_streaming(
items: Iterable[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
failure_policy: BatchFailurePolicy | None = None,
finalizer: EnqueueItem | None = None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> BatchHandle
Enqueue jobs from a lazy iterable in chunks.
Delegates to :meth:JobsClient.enqueue_batch_streaming; see its
docstring for chunk_size validation and streaming semantics.
Source code in src/taskq/client/_taskq.py
get_batch
async
¶
Fetch a single batch row by ID.
Delegates to :meth:JobsClient.get_batch. Returns None when
the batch does not exist.
Source code in src/taskq/client/_taskq.py
list_batches
async
¶
List batches matching filter, returning :class:BatchSummary objects.
Delegates to :meth:JobsClient.list_batches.
Source code in src/taskq/client/_taskq.py
enqueue_batch_fast
async
¶
enqueue_batch_fast(
items: list[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
) -> int
Enqueue jobs via COPY FROM protocol for maximum throughput.
Delegates to :meth:JobsClient.enqueue_batch_fast; see its
docstring for tradeoffs vs the regular :meth:enqueue_batch.
Source code in src/taskq/client/_taskq.py
get
async
¶
Look up a job by id. Returns None when the job does not exist.
Delegates to :meth:JobsClient.get (jobs table first, then the
archive tier; None means the id lives in neither). The
result_adapter default and its status-only-idiom rationale are
documented there.
Source code in src/taskq/client/_taskq.py
get_row
async
¶
Look up a job by id and return the raw JobRow, no handle.
Delegates to :meth:JobsClient.get_row: one backend read,
None when the job does not exist, mirroring :meth:get's
contract without the handle machinery.
Source code in src/taskq/client/_taskq.py
list
async
¶
List jobs matching filter, returning a :class:JobPage.
Delegates to :meth:JobsClient.list, note filter.unfinished
selects by terminality ('not yet finished'), not by execution
status ('currently executing'). See :class:JobFilter.
Source code in src/taskq/client/_taskq.py
backpressure
async
¶
Read the submit path's backpressure state for queues.
Delegates to :meth:JobsClient.backpressure; see it for the
full contract (the depth/cap/children halves, the fail-open
unknown state, the advisory semantics). Returns a
:class:~taskq.types.BackpressureSnapshot — a read-only typed
read mirroring the reader methods above, deliberately not an
enqueue kwarg.
Source code in src/taskq/client/_taskq.py
cancel
async
¶
Request cancellation of a job and return a :class:CancelResult.
Delegates to :meth:JobsClient.cancel. The result reports
previous_status (the row as first read) and new_status
(the row re-read after the cancel request was written; the two
are not atomically consistent, a job can move in between), and
cancellation_initiated is False when no cancellation was
needed (the job was already terminal, or had a cancel request
recorded).
Raises :class:KeyError carrying job_id when the job does not
exist (never enqueued, or pruned past retention) — the stdlib
"asked for an entry by id; it isn't there" idiom, matching
:meth:stream and :meth:JobHandle.wait.
Source code in src/taskq/client/_taskq.py
cancel_where
async
¶
cancel_where(
filter: JobFilter,
reason: str | None = None,
*,
allow_empty_filter: bool = False,
) -> BulkCancelResult
Cancel all jobs matching filter. See :meth:JobsClient.cancel_where.
Source code in src/taskq/client/_taskq.py
retry_job
async
¶
Re-pend a job that has come to rest (a terminal status).
The backend admin operation the admin UI's retry route calls:
every terminal status is a valid source (an operator re-run is
"run this again", including after a success), non-terminal
statuses are refused, and the attempt counter is not reset.
Returns False when the job was not in a retryable state, the
spent attempt sits at the smallint ceiling, or the job id does
not exist — the three cases are indistinguishable here by
design (the operation is a single conditional UPDATE); when a
False could mean a typo'd id rather than a busy job, check
existence first with :meth:get. The same conflict surfaces as
a 409 on the admin route.
Source code in src/taskq/client/_taskq.py
create_schedule
async
¶
create_schedule(
actor: str | ActorRef[P, R],
cron_expr: str,
*,
timezone: str = "UTC",
dst_strategy: DstStrategy = "skip",
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
name: str = "",
identity_key: IdentityKey | None = None,
enabled: bool = True,
) -> ScheduleHandle
Create a cron schedule. Delegates to :meth:JobsClient.create_schedule.
dst_strategy controls how DST gaps/overlaps are handled; see
:meth:JobsClient.create_schedule for the full semantics.
Source code in src/taskq/client/_taskq.py
list_schedules
async
¶
List cron schedules. Delegates to :meth:JobsClient.list_schedules.
Source code in src/taskq/client/_taskq.py
update_schedule
async
¶
update_schedule(
schedule_id: UUID,
*,
cron_expr: str | None = None,
enabled: bool | None = None,
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
clear_payload_factory: bool = False,
) -> ScheduleRecord
Update a cron schedule. Delegates to :meth:JobsClient.update_schedule.
Source code in src/taskq/client/_taskq.py
delete_schedule
async
¶
Delete a cron schedule. Delegates to :meth:JobsClient.delete_schedule.
stream
async
¶
Stream live state changes for a job as :class:JobEvent objects.
Yields one event per observable state transition (status change or
progress update), terminating automatically when the job reaches a
terminal state. The final event always has ``terminal=True``.
Latency contract
----------------
The first event is the row as it stands when ``stream()`` is
called. After that:
* **Redis configured** - the worker publishes every progress and
state-change event to the job's Redis channel, so a transition
is delivered as it happens; the row is re-read from Postgres
every ``poll_timeout`` seconds (default 30) as the safety net.
* **Postgres only** - nothing announces a job's transitions over
NOTIFY, so the row is re-read through the client's pool every
``min(poll_timeout, 0.5)`` seconds: a terminal write is observed
within half a second, at the cost of one primary-key read per
stream per half second. Lower ``poll_timeout`` to tighten it, to
a floor of 0.1 s; each wait is jittered ±20% so streams opened
together do not poll in lockstep. No connection is held per
stream, so this works on a pool-only client as well. A pool or
connection error on a read is retried on the next poll; reads
failing for 30 s straight end the stream with
:class:`~taskq.exceptions.StreamUnavailable`.
Usage::
async for event in tq.stream(job_id):
print(event.status, event.progress_state)
# loop exits automatically when event.terminal is True
# Or wire directly into a FastAPI SSE response:
async def event_generator():
async for event in tq.stream(job_id):
yield f"data: {event.model_dump_json()}
"
Raises
------
RuntimeError
Called before ``tq.open()`` or outside an ``async with`` block.
KeyError
The job does not exist — at the opening read, or on any later
poll whose row has been pruned in the meantime (the stream
cannot fabricate the promised terminal event, so it raises
rather than ending quietly).
StreamUnavailable
The Postgres poll could not re-read the row for 30 s straight.
Source code in src/taskq/client/_taskq.py
1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 | |
watch_reclaims
async
¶
watch_reclaims(
after_id: int = 0, *, poll_timeout: float | None = None
) -> AsyncIterator[EventRow]
Stream fleet-wide crash-reclaim events as :class:EventRow objects.
Delivery guarantee. At-least-once, and gap-free only under a
bounded writer-transaction assumption: an event is silently and
permanently missed if a job_events writer transaction stays
open longer than reclaim_event_visibility_delay (default 2s ,
see :data:taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY)
between its INSERT and its COMMIT. Ids are allocated at INSERT
time but transactions commit out of order, so a late-committing
lower-id row can land behind the cursor after the cursor has
already advanced past its position; no error is raised anywhere.
Sweep and terminal-write transactions are a handful of
single-round-trip statements, so 2s is a generous bound under
normal operation, but it is an assumption enforced by nothing in
the SQL, not a property the query guarantees. So this watcher
does not leave detection to chance: on a slow cadence
(_VISIBILITY_RISK_CHECK_INTERVAL, 60s) it runs the backend's
check_reclaim_visibility_delay_risk diagnostic (when the
backend implements it, PostgresBackend does) and logs a loud
watch-reclaims-visibility-delay-at-risk warning for every
long-open job_events writer it finds. That is a proxy
warning, not proof of an actual miss.
Yields job_events rows with kind='state_change' and
detail['reason']='lock_expired', ordered by the monotonic
event_id cursor ascending. to_state is 'pending' for
a retried reclaim, 'crashed' or 'cancelled' (cancel was
in-flight when the worker died) for a terminal one. cause
names which deadline fired, 'lock_expired' (the worker's
global lease) or 'heartbeat_timeout' (the per-job promise)
for a leader-sweep reclaim, 'isolate_self' for a reclaim a
worker performed on its own jobs when its heartbeat connection
died (isolate_self), the same channel so a consumer fanning
out on this feed sees heartbeat-loss self-recovery too.
Cursor and duplicate semantics¶
The caller persists the last-seen event_id and passes it back
as after_id on resumption. Persist the cursor after
processing each event: a crash between processing and persisting
re-delivers that event on the next run, delivery is
at-least-once, so consumers must dedupe on event_id. The
cursor is a watermark, not a reference: pruning job_events
rows at or below it is always safe, but rows pruned before the
consumer reads them are gone for good, only prune rows older
than your slowest consumer's cursor. A cursor far behind after
a long outage drains at query speed (full batches are re-polled
immediately, not one batch per poll_timeout).
Your own pruning is not the only deleter. The built-in event
retention sweep deletes these rows by age alone, cursor position
irrelevant: the lock_expired slice is carved out of the
ordinary event_retention_period window but only up to
:data:~taskq.constants.RECLAIM_OUTBOX_RETENTION_MULTIPLIER
times it (100x), after which it is deleted like any other event.
The terminal-job prune is a second deleter: its cascade removes
the archived jobs' events with them. Both deleters record their
highest deleted event id in the event-prune watermark
(job_events_prune_state, migration 01.00.20_02) in the same
transaction as the delete, and every delivery path of this
generator opens on the gap gate: a resumed cursor (a persisted
cursor, after_id > 0) strictly below the watermark ends the
stream with :class:~taskq.exceptions.EventRetentionGapError
before anything is delivered. The signal is conservative (the
watermark is a union bound across event kinds: it proves the feed
is incomplete, not which slice lost a row), and it closes the
silent-loss hole -- a consumer lagging past the horizon
would otherwise skip to live with no error on either side, free to
believe it saw everything. A fresh watcher (after_id=0) is a
new tail and never raises; a cursor at or above the watermark
never raises, every deleted id was at or below a position the
consumer had already been delivered. Size the retention period
against your slowest consumer's worst outage all the same: the
gate turns an over-long outage into a loud, actionable end, not
into a recovered feed.
Shutdown and backpressure¶
This is a pull-based async generator: events are fetched only as
fast as the consumer iterates, so a slow consumer polls
slower, no internal buffer grows. To stop, break out of the
async for (or cancel the consuming task); generator cleanup
removes the LISTEN registration and closes any owned connection.
Transport¶
On Postgres with a LISTEN transport source (dsn,
pg_conn_factory, or listen_conn), the method LISTENs on
wake_channel(schema) purely as a low-latency wakeup, but
always polls backend.poll_reclaim_events(after_id) as the
durable source of truth, NOTIFY is an optimisation, never the
only path, and a dropped LISTEN connection degrades to (and
recovers from) polling automatically. Without a LISTEN
transport or on Redis-configured backends, a plain poll loop
against poll_reclaim_events runs on poll_timeout.
Usage::
cursor = await load_cursor() # your own durable store
async for evt in tq.watch_reclaims(after_id=cursor):
outstanding -= 1 # fan-out completion tracking
cursor = evt.event_id
await save_cursor(cursor) # persist AFTER processing
Raises¶
RuntimeError
Called before tq.open() or outside an async with block.
EventRetentionGapError
The resumed cursor sits behind the event-prune watermark:
retention deleted undelivered events, the stream ends instead
of silently skipping to live.
Source code in src/taskq/client/_taskq.py
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 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 | |
ActorsClient ¶
Pool-wrapping facade for actor configuration operations.
Acquires a connection from the injected pool for each call, delegates
to taskq.actor_config_ops, and returns the result. The
caller must have opened the pool; this class does not manage its
lifecycle.
.. note::
This client is Postgres-only, it delegates to
:mod:taskq.actor_config_ops, which executes raw SQL against
the actor_config table. The :class:~taskq.backend._protocol.Backend
protocol does not include actor config operations, so
:class:~taskq.testing.InMemoryBackend does not support
ActorsClient.
Parameters¶
pool:
An open asyncpg.Pool. The caller retains ownership.
schema:
TaskQ schema name. Defaults to "taskq".
Source code in src/taskq/client/_actors.py
list
async
¶
List all stored actor_config rows, ordered by actor name.
get
async
¶
Get one actor_config row, or None if not found.
set_capacity
async
¶
set_capacity(
actor: str,
*,
max_concurrent: int | Unset | None = UNSET,
max_pending: int | Unset | None = UNSET,
result_ttl: float | Unset | None = UNSET,
) -> ActorConfigRow | None
Update capacity fields on an existing actor_config row.
Source code in src/taskq/client/_actors.py
deregister
async
¶
Deregister an actor with safety checks.
See :func:taskq.actor_config_ops.deregister_actor for
the full semantics.
Source code in src/taskq/client/_actors.py
SubJobEnqueuer ¶
SubJobEnqueuer(
loop_scope_resolved: Mapping[type, object] | None,
worker_pool: Pool | None,
backend: Backend,
*,
clock: Clock | None = None,
capacity_cache: ActorCapacityCache | None = None,
queues_strict: bool = False,
env_queues: Sequence[str] | None = None,
transaction_conn: ConnLike | None = None,
)
Enqueue sub-jobs from within an actor body.
Two construction shapes, deliberately distinct:
- Loop-level shared,
loop_scope_resolvedset, notransaction_conn. Serves the single-slot path and the autonomous fallback; transactional buffering is inferred from connection provenance (a LOOP-scope connection means "inside the consumer's transaction"). One instance per loop, so the per-100-enqueue re-warning fires on the loop-level counter, not per-job. Never shared across concurrent jobs that each own a transaction, its buffer state is unkeyed. - Per-job bound,
transaction_connset to the job's slot transaction connection (the shapedispatch_one_jobconstructs whenever the worker runs a dedicated slot pool). Every write joins that one transaction and the buffers are this job's alone, so a sibling slot's flush/discard/drain can never reach them.
Source code in src/taskq/client/_enqueuer.py
capacity_cache
property
¶
The TTL-bounded capacity snapshot this enqueuer reads.
Exposed read-only so a derived enqueuer (the per-job binding the dispatch path constructs) can share the loop-level cache instead of paying its own per-job snapshot refresh.
backpressure
async
¶
Read the submit path's backpressure state for queues.
The actor body's slice of the LIB-2 read (the fan-out decision
happens HERE, in the parent's execution): same contract as
:meth:JobsClient.backpressure — the actor-scoped verdict, the
fail-open unknown state, advisory semantics — reading this
enqueuer's own backend and capacity cache. parent_id defaults
to the ambient parent context: inside an actor body that is THIS
job, so the snapshot's fan-out ledger includes this parent's
pending children with no arguments. See
:class:~taskq.types.BackpressureSnapshot.
Source code in src/taskq/client/_enqueuer.py
enqueue
async
¶
enqueue(
actor_ref: ActorRef[P, R],
payload: P,
*,
connection: Connection | None = None,
allow_unregistered: bool = False,
scheduled_at: datetime | None = None,
priority: int | None = None,
fairness_key: str | None = None,
metadata: dict[str, object] | None = None,
identity_key: IdentityKey | None = None,
idempotency_key: IdempotencyKey | str | None = None,
idempotency_scope: str | None = None,
unique_for: timedelta | None = None,
unique_states: tuple[JobStatus, ...] | None = None,
max_pending: int | None = None,
_batch_id: str | None = None,
tags: list[str] | None = None,
inherit_tags: bool = True,
schedule_to_close: datetime | None = None,
start_to_close: timedelta | None = None,
heartbeat_timeout: timedelta | None = None,
) -> JobHandle[R]
Enqueue a sub-job. max_pending is a per-call limit resolved
against the operator-owned stored cap and the @actor(...)
literal: against a non-NULL stored max_pending the tighter of
the two wins (min(stored, per_call), an
explicit caller shedding load is never widened by an operator
override, and no code path can raise an operator's fleet cap);
with no stored value this parameter wins outright over the
literal (historical behavior, actor code may loosen its own
declaration).
allow_unregistered is the per-submit escape from
TASKQ_QUEUES_STRICT (whose hard branch this arm shares
through the capacity cache): a child queue no registered
assignment routes to normally raises
:class:~taskq.exceptions.UnknownQueueError before any write;
the flag skips that verdict for THIS call, for genuinely dynamic
child queue names. The unserved-queue NOTE still fires on an
escaped call. With strict off the flag changes nothing.
_batch_id is a library-internal parameter used by
:meth:enqueue_batch to stamp batch_id into metadata after
:func:build_enqueue_args has stripped any caller-supplied
batch_id (H5 security boundary). Callers MUST NOT pass it.
Source code in src/taskq/client/_enqueuer.py
279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 | |
enqueue_batch
async
¶
enqueue_batch(
items: Sequence[EnqueueItem[Any, Any]],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
) -> list[JobHandle[Any]]
Enqueue a batch of sub-jobs sharing a single batch_id.
All items share a single batch_id UUID written into each
job's metadata.batch_id field (as a string). When batch_id
is not supplied it is auto-generated as a UUIDv7 via
:func:~taskq._ids.new_job_id, mirrors
:meth:~taskq.client.JobsClient.enqueue_batch. Pass an explicit
batch_id to correlate this batch with a caller-constructed
identifier (e.g. a finalizer job enqueued separately that needs to
reference the same batch). An explicit batch_id naming an
existing TERMINAL batch row raises
:class:~taskq.exceptions.BatchIdExistsError, the same refusal
the create_batch arms give a collision, a terminal batch must not
gain a member. An ACTIVE row keeps accepting appends (resumption).
Raises ValueError when items is empty or exceeds
MAX_BATCH_SIZE, the same guardrails
:meth:~taskq.client.JobsClient.enqueue_batch applies to the
identical operation one layer up. Without the empty check the
no-connection fallback loop would iterate zero items and return
[] silently. The backend binds every item
as 26 parallel array parameters to a single unnest INSERT in one
transaction, so an uncapped batch enqueued from inside a job body is
unbounded fan-out that bypasses the client-side guardrail.
max_pending: the connection arm gets the backend's per-actor
partition admission, over-cap actors' items are refused after
the within-cap actors' items are inserted, and converts the
refusal to :class:~taskq.exceptions.PartialBatchError so this
method speaks one error contract across all connection modes
(the no-connection fallback already raises it per item). On a
LOOP-scope connection the inserted items ride the parent's
transaction: catching the error inside the actor body and
returning normally commits them; letting it propagate rolls
everything back.
Source code in src/taskq/client/_enqueuer.py
498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 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 | |
flush_buffer
async
¶
Flush buffered EnqueueArgs to the backend (in-memory simulation).
Called by the consumer on actor success, AFTER the LOOP-scope
transaction has committed. Per-item flush failures are collected
and re-raised as :class:~taskq.exceptions.SubEnqueueError after
the loop completes so callers can detect lost sub-jobs.
A batch row that went terminal while the parent's transaction was in flight refuses its members at the backend's single-enqueue write site (the terminal-batch guard, the twin of the PG single path's membership lock), so a terminal batch's buffered members surface HERE, per item, through that same SubEnqueueError collection - not as a wholesale preflight raise, which would speak no documented error contract, discard the buffered args with no handle to them, and block the flush of unrelated batches' buffered members. The refusal itself is not negotiable: nothing from a terminal batch lands, the bulk arms refuse it the same way.
Source code in src/taskq/client/_enqueuer.py
discard_buffer ¶
drain_for_re_enqueue ¶
Return and clear both loop-scope and pending buffers for re-enqueue.
Source code in src/taskq/client/_enqueuer.py
FakeClock ¶
Deterministic clock for tests.
Accepts a start datetime (typically
datetime(2025, 1, 1, tzinfo=UTC)). now() returns the current
internal time; move_to and advance let tests control the clock
explicitly. monotonic() returns elapsed seconds from _EPOCH
so that elapsed-time guards see a non-zero starting value.
Source code in src/taskq/clock.py
WorkerConnections
dataclass
¶
WorkerConnections(
dispatcher_pool: Pool | None = None,
dispatcher_pool_factory: PoolFactory | None = None,
heartbeat_pool: Pool | None = None,
heartbeat_pool_factory: PoolFactory | None = None,
worker_pool: Pool | None = None,
worker_pool_factory: PoolFactory | None = None,
notify_conn: Connection | None = None,
notify_conn_factory: ConnFactory | None = None,
leader_conn: Connection | None = None,
leader_conn_factory: ConnFactory | None = None,
redis_client: Redis | None = None,
redis_client_factory: RedisFactory | None = None,
)
Per-role connection overrides for the worker.
Each role has a <role> (pre-constructed, caller-owned) and a
<role>_factory (zero-arg async factory, TaskQ-owned) slot.
Leave both None for DSN-based construction (the default).
Example, AAD-managed-identity worker::
from azure.identity.aio import DefaultAzureCredential
from taskq.aad import EntraIdProvider
from taskq.auth import make_pg_pool_factory
from taskq.connections import WorkerConnections
cred = DefaultAzureCredential()
provider = EntraIdProvider(cred)
connections = WorkerConnections(
dispatcher_pool_factory=make_pg_pool_factory(
settings.pg_dsn_direct, provider, max_size=settings.dispatcher_pool_size,
),
heartbeat_pool_factory=make_pg_pool_factory(
settings.pg_dsn_direct, provider,
max_size=settings.heartbeat_pool_size,
command_timeout=settings.heartbeat_command_timeout,
),
worker_pool_factory=make_pg_pool_factory(
settings.pg_dsn_pooled, provider, max_size=settings.worker_pool_size,
),
)
Example, share an app-wide pool (caller-owned)::
connections = WorkerConnections(worker_pool=app_state.pg_pool)
dispatcher_pool
class-attribute
instance-attribute
¶
Dispatcher pool (pg_dsn_direct role). Caller-owned if set.
dispatcher_pool_factory
class-attribute
instance-attribute
¶
Factory for the dispatcher pool. TaskQ-owned.
heartbeat_pool
class-attribute
instance-attribute
¶
Heartbeat pool (pg_dsn_direct, heartbeat_command_timeout). Caller-owned.
heartbeat_pool_factory
class-attribute
instance-attribute
¶
Factory for the heartbeat pool. TaskQ-owned. command_timeout is
your responsibility when overriding, set it on create_pool.
worker_pool
class-attribute
instance-attribute
¶
Worker pool (pg_dsn_pooled role). Caller-owned.
worker_pool_factory
class-attribute
instance-attribute
¶
Factory for the worker pool. TaskQ-owned.
notify_conn
class-attribute
instance-attribute
¶
Dedicated LISTEN connection. Caller-owned. TaskQ still issues LISTEN.
notify_conn_factory
class-attribute
instance-attribute
¶
Factory for the LISTEN connection. TaskQ-owned.
leader_conn
class-attribute
instance-attribute
¶
Dedicated advisory-lock connection. Caller-owned.
leader_conn_factory
class-attribute
instance-attribute
¶
Factory for the advisory-lock connection. TaskQ-owned.
redis_client
class-attribute
instance-attribute
¶
Redis client for progress fanout / rate limiting. Caller-owned.
redis_client_factory
class-attribute
instance-attribute
¶
Factory for the Redis client. TaskQ-owned.
The client a factory returns replaces the one TaskQ builds from
TASKQ_REDIS_URL — including its resilience defaults. TaskQ-built
clients set health_check_interval=30, socket_keepalive=True,
a 5s socket_timeout (pinned to redis-py 8.x's asyncio default —
the bound the fail-closed wall stands on), and a 1-attempt
exponential-backoff Retry (0.02s base, 0.1s cap — see
taskq._redis_client's RETRY_RETRIES for why one) with
retry_on_error=[ConnectionError, TimeoutError] (see
taskq._redis_client). A factory-built
client must set these kwargs itself — without them an idle-closed
broker connection surfaces as a raw ConnectionError in the
progress-stream pubsub and as the limiter's fail-closed warning noise
(see docs/guides/managed-identities.md).
__post_init__ ¶
Reject concrete + factory for the same role (configuration error).
Source code in src/taskq/connections.py
has_any ¶
True if any override (concrete or factory) is set.
Source code in src/taskq/connections.py
JobContext
dataclass
¶
JobContext(
job_id: UUID,
actor: str,
queue: str,
attempt: int,
claim_epoch: int,
worker_id: UUID,
payload: P,
jobs: SubJobEnqueuer,
log: BoundLogger,
snooze_count: int = 0,
span: Span | None = None,
cancel_event: Event = asyncio.Event(),
_abort_requested: Event = threading.Event(),
_cancel_origin: CancelOrigin = CancelOrigin.NONE,
_progress_buffers: dict[UUID, _ProgressBuffer]
| None = None,
_redis_client: Redis | None = None,
_worker_settings: WorkerSettings | None = None,
_pending_publish_tasks: set[Task[None]] | None = None,
_progress_dropped_notice: Event = threading.Event(),
_sync_actor_task: Task[object] | None = None,
_tx_unwind_task: Task[object] | None = None,
_actor_body_task: Task[object] | None = None,
_tx_rollback_collision_window: bool = False,
)
Per-job execution context handed to worker actors.
The cancel_event field is a plain :class:asyncio.Event, never
wrapped in a cancel scope or :class:asyncio.TaskGroup (
PEP 789 mitigation). The consumer constructs a fresh event per
attempt; the cancel-poll hook sets it on phase 1; user actor code
polls :attr:cancellation_requested or awaits cancel_event.wait().
payload is typed as the actor's payload model P. The worker
consumer validates the raw dict[str, object] payload from the
JobRow against actor_ref.payload_type before constructing the
context, so handlers see a fully-validated Pydantic instance.
jobs provides :class:SubJobEnqueuer for enqueuing sub-jobs
from within the actor body. The enqueuer resolves the database
connection via LOOP-scope DI → worker-pool fallback.
snooze_count is the job row's count of completed non-consuming
deferrals (:class:~taskq.exceptions.Snooze /
:meth:RetryAfter<taskq.exceptions.RetryAfter> with
consume_budget=False) at dispatch time. Such a deferral refunds
the claim's attempt increment, so attempt alone cannot count
snooze cycles, an actor that wants to snooze N times and then
succeed keys off snooze_count, not off attempt.
cancel_event
class-attribute
instance-attribute
¶
cancel_origin
property
¶
Who asked for this attempt's cancellation.
Read it alongside :attr:cancellation_requested when the two
answers differ in what the actor should do: on a shutdown
(:attr:CancelOrigin.SHUTDOWN) the attempt is released back to the
fleet with its budget refunded, so an actor that can checkpoint
should prefer to stash progress and raise rather than return a
partial result; on an operator cancel
(:attr:CancelOrigin.OPERATOR) the job terminalises, so returning
the partial result is the only way to keep it.
check_cancelled ¶
should_abort ¶
Synchronous cancellation check for sync actors (thread-safe).
Sync actors cannot await the async :attr:cancel_event, so
they poll this method cooperatively. The cancel controller sets
the underlying :class:threading.Event during phase 1.
Returns:
| Type | Description |
|---|---|
bool
|
|
bool
|
actor should return or raise immediately. |
Source code in src/taskq/context.py
progress
async
¶
progress(
*,
step: int | None = None,
percent: float | None = None,
detail: str | None = None,
data: dict[str, object] | None = None,
) -> None
Report incremental progress for this job.
Updates the in-memory coalesce buffer synchronously, then schedules a
best-effort kind="progress" Redis publish as a background task
when a client is connected, this call never blocks on the network.
Raises :class:~taskq.exceptions.ProgressTooLarge if the serialised
data payload exceeds WorkerSettings.progress_data_max_bytes.
All arguments are optional and merged last-writer-wins into the
accumulated pending_state. Intermediate calls between periodic
flush ticks are coalesced: only the latest value for each field
reaches Postgres. seq is strictly monotone across calls.
data must have str dict keys (nested too): the size check
and the PG flush serialize it directly, and a non-str key
raises TypeError (JSON and jsonb cannot carry non-string
keys; silent coercion would corrupt the round-trip).
When no progress buffers are wired into this context (direct actor testing, a miswired context), the call, the Redis publish included, is a deliberate no-op; the first such dropped call emits one debug-level notice so the no-op is discoverable, and later calls stay silent.
The Redis publish is genuinely fire-and-forget: it may complete out
of order relative to other in-flight publishes for the same job.
Consumers reading the SSE/pub-sub stream already discard any event
whose seq is not greater than the last one seen (see
:mod:taskq.web.progress), so out-of-order or dropped publishes
never corrupt displayed state, the buffer mutation above (and the
eventual Postgres flush) is the durable source of truth. Failures
publishing to Redis are logged and recorded as a metric, never
raised here.
The publish is also COALESCED when a shared publish-task set is
wired: at most one publish per job is in flight at a time, and a
call landing while one is running only latches its event for the
in-flight task to re-publish when its round trip lands. A trailing
call is never lost (it publishes directly or through the latch);
intermediate calls while the gate is busy reach subscribers only as
the latched latest event. See
:func:taskq.progress._publish._publish_progress_event_coalesced
for the full residual.
Source code in src/taskq/context.py
263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 | |
CronScheduleSpec
dataclass
¶
CronScheduleSpec(
actor: str,
cron_expr: str,
timezone: str = "UTC",
dst_strategy: DstStrategy = "skip",
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
name: str = "",
identity_key: IdentityKey | None = None,
enabled: bool = True,
owner: Literal["code", "operator"] = "code",
)
Immutable specification for a cron schedule row.
Created by the :func:cron decorator or constructed directly for
register_cron(). payload_factory and static_payload are
mutually exclusive, setting both raises :class:ValueError at
construction time (via :func:cron).
dst_strategy controls how DST gaps and overlaps are handled.
See :data:DstStrategy for the semantics of each strategy.
owner controls who owns the schedule's enable/disable lifecycle:
"code"(default): the schedule is declared by a@crondecorator and the worker registration pass owns the enable state at startup. A schedule the cron loop auto-disabled on a transient failure blip is re-enabled when the code re-declares the schedule (the boot is proof the declaration is live again); an operator's deliberate disable is never reverted."operator": an operator owns the lifecycle (a schedule created throughJobsClient.create_schedule). The registration pass never re-enables it, whatever disabled it.
ScheduleHandle
dataclass
¶
ScheduleHandle(
schedule_id: UUID,
actor: str,
cron_expr: str,
timezone: str,
enabled: bool,
next_fire_at: datetime,
_backend: Backend = field(),
dst_strategy: DstStrategy = "skip",
name: str = "",
identity_key: IdentityKey | None = None,
)
Immutable handle for a cron schedule, returned by JobsClient methods.
The handle fields are a point-in-time snapshot of schedule state.
Async methods delegate to the Backend injected at construction time
(not part of the public __init__ signature) via ScheduleUpdateArgs.
enable() passes ScheduleUpdateArgs(enabled=True); the backend
resets consecutive_failures=0, last_fire_error=NULL and
disabled_by=NULL when enabled=True is set. disable() passes
enabled=False, which stamps disabled_by='operator': a deliberate
disable is operator intent, and the worker's startup registration pass
reverts only the cron loop's own 'auto' marker.
disable
async
¶
Disable this schedule.
Writes disabled_by='operator' alongside enabled=false: a
deliberate disable is operator intent, recorded so the worker's
startup registration pass never reverts it (only the cron loop's
own auto-disable is recoverable, see :class:CronScheduleSpec).
Source code in src/taskq/cron.py
enable
async
¶
ActorConfigDriftError ¶
ActorConfigDriftError(
actor: str,
field: Literal["metadata"],
registered: dict[str, object] | None,
stored: dict[str, object] | None,
)
Bases: TaskQError
One actor whose registered structural config differs from the stored row.
Only metadata is structural, no operator surface can move it, so a
mismatch there is always a correctness bug and refuses boot. The queue
assignment is operator-owned once a row exists (moved by
taskq actor-config move-queue): a differing literal never raises,
it logs actor-config-queue-override at boot and the stored
assignment wins. Capacity fields (max_concurrent,
max_pending, result_ttl) are likewise operator-owned and never
raise this error; see :func:taskq.worker.startup.sync_actor_config.
Source code in src/taskq/exceptions.py
ActorConfigDriftList ¶
Bases: TaskQError
Collected wrapper raised at worker startup when one or more actors have drift.
Source code in src/taskq/exceptions.py
ActorDeregistrationError ¶
ActorHasActiveJobsError ¶
ActorHasActiveJobsError(
actor: str,
active_count: int,
status_counts: dict[str, int],
*,
force: bool = False,
)
Bases: ActorDeregistrationError
Non-terminal jobs reference the actor.
Carries the count and per-status breakdown of the blocking jobs so the
caller can decide whether to cancel them first or use force=True.
When force is True, the message reflects that running jobs
cannot be cancelled by force=True, the caller must wait for
them to finish or cancel them individually first.
Source code in src/taskq/exceptions.py
ActorHasEnabledSchedulesError ¶
Bases: ActorDeregistrationError
Enabled cron schedules reference the actor.
Carries the schedule IDs so the caller can disable or delete them first.
Source code in src/taskq/exceptions.py
ActorNotFoundError ¶
Bases: ActorDeregistrationError
The actor_config row does not exist, nothing to deregister.
Currently raised only by :func:deregister_actor. Other ops
(get, set_capacity) return None for missing rows.
Source code in src/taskq/exceptions.py
BackpressureError ¶
Bases: TaskQError
Base class for synchronous enqueue-time backpressure signals.
Subclassed by SingletonCollisionError (singleton collision) and used directly by max_pending enforcement. The caller decides whether to retry, fail, or wait; the library does not block on capacity.
Source code in src/taskq/exceptions.py
BatchAbortedError ¶
Bases: TaskQError
A batch was aborted because consecutive failures exceeded the threshold.
Running jobs are NOT cancelled by the abort, only pending and scheduled jobs are cancelled. Running jobs continue to completion. This matches the post-terminal-write hook design: the hook runs after the terminal write, so a job that was dispatched before the abort triggered will run to completion.
Source code in src/taskq/exceptions.py
BatchMaxPendingExceededError ¶
BatchMaxPendingExceededError(
*,
refusals: list[MaxPendingExceededError],
refused_indices: dict[str, list[int]],
admitted_count: int,
)
Bases: BackpressureError
A bulk enqueue partitioned its admission per actor and refused some.
Raised by :meth:~taskq.backend._protocol.Backend.enqueue_batch /
:meth:~taskq.backend._protocol.Backend.enqueue_batch_fast (and every
client path riding them) when one or more actors' items exceed their
effective max_pending: the within-cap actors' items are inserted
FIRST, then this error raises naming the refusals, the bulk-tier
sibling of :class:PartialBatchError, which is the house shape for
partial batch admission (succeeded count + failed indices + typed
per-failure exceptions).
An over-cap actor's items are refused as a whole group, never partially filled up to the cap: the single-enqueue path refuses a capped enqueue outright, and a partial fill would admit an arbitrary prefix of the caller's items that the caller never chose.
Fields:
refusals, one :class:MaxPendingExceededErrorper over-cap actor (actor,current_countat the admission check, the effectivemax_pending).refused_indices, actor name -> indices into the caller's items list of that actor's refused items. For :meth:~taskq.client.JobsClient.enqueue_batch_streaming's chunked path the indices are stream-global and the stream stops at the refusing chunk: items after it were never attempted.admitted_count, how many items were admitted and inserted by the raising call.
Durability of the admitted items depends on the path: committed when
the call owned its transaction (no caller-supplied connection, one
pool transaction per call/chunk); inserted-but-uncommitted on a
caller-supplied connection with an open transaction, where that
transaction's commit/rollback decides. enqueue_batch /
enqueue_batch_streaming with failure_policy or finalizer
and no connection (the atomic path) never raises this error, its
single transaction keeps the legacy all-or-nothing contract and
raises plain :class:MaxPendingExceededError with nothing committed.
Deliberately NOT a :class:MaxPendingExceededError subclass: handlers
written for the pre-partition contract assume that a raised
MaxPendingExceededError left nothing enqueued. Under this error
part of the batch IS stored, a blind whole-batch retry would
duplicate the admitted items. Catch this type explicitly and retry
only the refused indices, or give items idempotency_keys so a
whole-batch retry deduplicates against the admitted rows.
The same hazard reaches handlers written against the shared base:
except BackpressureError catches this error too, and the
base's contract ("the caller decides whether to retry, fail, or
wait") predates partial admission. A generic backpressure handler
that retries the whole batch MUST first consult
admitted_count / refused_indices, retry only the refused
items, or rely on idempotency_keys, otherwise it duplicates
the admitted items on every retry.
Source code in src/taskq/exceptions.py
DependencyCycle ¶
Bases: TaskQError
A cycle was detected in the provider graph.
Source code in src/taskq/exceptions.py
DIError ¶
Bases: TaskQError
Base for DI engine errors not covered by startup-validation.
Raised by the solver at resolution time for malformed annotations (e.g. multiple Scope markers in one Annotated parameter) or unresolvable forward references in actor signatures. Distinct from MissingProvider / ScopeViolation / DependencyCycle, which are raised at startup validation.
DuplicateIdempotencyKeyError ¶
DuplicateIdempotencyKeyError(
*,
idempotency_key: str | None = None,
idempotency_scope: str | None = None,
detail: str | None = None,
)
Bases: TaskQError
enqueue_batch_fast aborted: an item's
(idempotency_scope, idempotency_key) pair is already enqueued by the
SAME actor. A pair spanning two actors raises
:class:IdempotencyKeyActorMismatchError instead, the same refusal the
single and batch tiers apply.
COPY has no ON CONFLICT arbiter, so a same-pair duplicate ,
repeated within the batch or raced against a row the composite
jobs_idempotency_scope_key_uniq index already covers, aborts the
ENTIRE batch before a single row is written (all-or-nothing; the
abort is deliberate bulk-import semantics, unchanged by the
classification this error introduced). The non-fast paths never
raise for this condition: their ON CONFLICT arbiter dedupes and
RETURNS the existing row, so a typed domain error for a
deduplication-constraint violation on the enqueue path does not exist
there, hence this class, expressing the same idempotency constraint
at the bulk-import boundary in both the SQL and in-memory backends.
Distinct from :class:ScopedIdempotencyMigrationPendingError, which is
the rolling-deploy window's cross-scope reuse signal.
idempotency_key / idempotency_scope carry the offending pair,
resolved exactly on both backends by one shared rule
(first_duplicate_idempotency_pair): the first item in batch order
whose pair repeats an earlier item or is already stored. The PG path
resolves the stored half with a targeted post-abort lookup inside the
caller's transaction scope, never by parsing the violation's detail
text, which renders values raw and unquoted (ambiguous under
positional reading for comma-bearing scopes, unusable when localized
or truncated). Both fields are None only when the conflicting
row could not be resolved at all, a committed-and-instantly-deleted
racer, where the detail-text match is the last word and still never
guesses. detail carries the postgres detail verbatim when
present.
Resolution: pre-deduplicate the items, or use
:meth:~taskq.client.JobsClient.enqueue_batch, which dedupes and
returns the existing rows.
Source code in src/taskq/exceptions.py
EmptyBatchError ¶
Bases: TaskQError
A batch has fewer jobs than the expected minimum, or is unknown.
This can happen when jobs were pruned before wait_for_batch ran,
when expected_size was set but jobs were never created, or when
wait_for_batch is given a batch_id this client never
enqueued (no jobs and no batches row: the expected=None arm,
whose message names the truth instead of a fabricated minimum).
Pass on_empty="ok" to wait_for_batch to suppress the
no-batch-row variant of this error.
Source code in src/taskq/exceptions.py
EmptyFilterError ¶
Bases: TaskQError
Raised when cancel_where is called with a filter that has no predicates.
A filter with no queue, status, actor, identity_key, batch_id, tags, or
active predicate would match every job in the table, almost certainly
a bug. The guardrail is intentionally loud: the caller must add at least
one predicate or explicitly bypass with allow_empty_filter=True.
Source code in src/taskq/exceptions.py
EventRetentionGapError ¶
Bases: TaskQError
A watch_reclaims stream resumed a cursor the retention horizon passed.
The event-prune watermark (job_events_prune_state.pruned_through_id,
migration 01.00.20_02) records the highest job_events id any retention
deleter has committed a delete below-or-at. A resumed cursor strictly
below that bound means at least one event id the consumer never received
was deleted: the hole can never be refilled, so the stream ends here
instead of silently skipping to live and letting the consumer believe it
saw everything.
This makes a loss loud that silent polling would hide:
no error is raised on either side, the poll simply returns the
surviving rows and the consumer's async for read like a quiet fleet.
The signal is conservative by design: the watermark is a union bound across event kinds, so it proves the feed is incomplete without proving which slice a consumer filters for lost a row. A cursor AT or above the watermark never raises: every deleted id was at or below a position the consumer had already been delivered.
Attributes:
| Name | Type | Description |
|---|---|---|
cursor |
the resumed cursor the poll rejected. |
|
pruned_through_id |
the watermark it sits behind. |
Remediation: reposition the consumer past the watermark (accepting the
recorded loss) or size event_retention_period against the slowest
consumer's worst outage so the horizon cannot pass a live cursor.
Source code in src/taskq/exceptions.py
IdempotencyKeyActorMismatchError ¶
IdempotencyKeyActorMismatchError(
*,
actor: str,
existing_actor: str,
existing_job_id: UUID | None,
idempotency_key: str,
idempotency_scope: str | None,
)
Bases: TaskQError
An idempotency hit resolved to a job of a DIFFERENT actor.
Uniqueness is (idempotency_scope, idempotency_key), schema-wide, so
two actors sharing a key collide. A same-actor hit is a dedup and returns
the existing row; a cross-actor hit cannot be one, the caller asked for
THIS actor's job and would receive a handle whose result is another
actor's, indistinguishable from a successful dedup, so it is refused.
The composite index here cannot include the actor without a migration,
so the hit is checked after the fact and refused instead of silently
resolved.
Nothing was inserted (single enqueue: the arbiter skipped the row; batch:
the whole batch is rolled back, all-or-nothing like a singleton
collision; batch fast: the COPY aborts the whole batch the same way).
Namespace keys per actor ("send_receipt:order_123") or
give the two actors different idempotency_scope values.
existing_job_id names the stored job the key matched; it is None
when the collision is between two items of the same fast-path batch,
where no row was stored for either.
Source code in src/taskq/exceptions.py
IllegalStateTransition ¶
Bases: TaskQError
Attempted to transition a job to a status not reachable from its current status.
Best-effort fast-path check only; the SQL WHERE clause is the authoritative serialization gate for concurrent writes.
Source code in src/taskq/exceptions.py
JobFailed ¶
Bases: TaskQError
:meth:JobHandle.wait saw a non-success terminal state.
Carries the row so callers can inspect status, error_class,
error_message, and error_traceback. Distinct from
:class:ResultUnavailable (which means terminal but no result
stored) and from the original actor exception (which is recorded on
the row, not raised).
Source code in src/taskq/exceptions.py
MaxPendingExceededError ¶
Bases: BackpressureError
Raised when an actor's max_pending queue-depth limit is reached.
current_count is the count of pending+scheduled jobs at the time
of the pre-flight check. max_pending is the configured limit.
The caller decides whether to retry, fail, or wait; the library does
not block on capacity.
Source code in src/taskq/exceptions.py
hint
class-attribute
instance-attribute
¶
hint = "Inside an actor body, prefer `raise Snooze(delay)` over letting this reach the retry classifier: a full queue is backpressure, not a failure, and a Snooze defers without spending the job's retry budget."
MaxPendingLockTimeoutError ¶
Bases: BackpressureError
Raised when the advisory-lock wait bounding a capped actor's exact count-then-insert exceeded its budget.
Distinct from :class:MaxPendingExceededError: the cap was never
observed -- this enqueue lost the race to even run the check, waiting
behind other producers on the same (schema, actor) lock until the
budget expired. The caller's correct response is the same as for a cap
rejection (retry later, or shed load), so this is raised from the same
:class:BackpressureError family and recorded against the same
taskq.backpressure.errors counter (kind="max_pending_lock_timeout").
timeout_ms is the budget that expired. pending is 0 and
max_pending is None -- no count was taken.
Deliberately NOT a subclass of :class:MaxPendingExceededError: that
class means "the cap is full" and carries the observed count; a lock
timeout means "too contended to check" and conflating the two would
mislead handlers that react to a full queue (e.g. by logging the
count). Catch :class:BackpressureError to treat both the same way.
Source code in src/taskq/exceptions.py
args
instance-attribute
¶
args = (
f"backpressure: enqueue for actor {actor!r} could not acquire the max_pending advisory lock within {timeout_ms:g} ms of contention (the exact cap check was not reached). Retry later or shed load, exactly as for MaxPendingExceededError.",
)
MissingProvider ¶
Bases: TaskQError
A type was injected but no provider is registered.
Source code in src/taskq/exceptions.py
PartialBatchError ¶
PartialBatchError(
*,
succeeded_count: int,
failed_items: list[tuple[int, Exception]],
total: int,
)
Bases: TaskQError
Raised when an autonomous enqueue_batch partially fails.
Items enqueued before the first failure are committed; remaining
items are not inserted. succeeded_count is the number of items
that were successfully enqueued. failed_items maps the index
of each failed item to its exception. total is the original
batch size.
Source code in src/taskq/exceptions.py
PayloadValidationError ¶
PayloadValidationError(
detail: str,
*,
actor: str | None = None,
payload_schema_ver: str | None = None,
validation_errors: list[dict[str, object]]
| None = None,
item_index: int | None = None,
)
Bases: TaskQError
Pydantic validation failed at enqueue or dispatch.
At enqueue: raised before the row is inserted ('fail at the door'). At dispatch: causes the job to transition to 'failed' with error_class='PayloadValidationError'. Non-retryable in both cases regardless of the actor's retry policy.
item_index is the failing item's position in the CALLER's
coordinate space (the batch list / the streaming caller's stream)
whenever the raise site knows one, batch per-item guards and their
remapping boundaries populate it, so a handler or retry tool reads
the position as a field instead of parsing the message. None
is the no-coordinate case: single-item enqueue, dispatch-time
validation, non-itemized configuration errors. The message embeds
the same index for humans; the field is the machine-readable copy.
Source code in src/taskq/exceptions.py
ProgressTooLarge ¶
Bases: TaskQError
Raised when progress data payload exceeds the configured size limit.
limit is the configured cap in bytes (WorkerSettings.progress_data_max_bytes).
actual is the serialised byte length of the data dict that was rejected.
Non-retryable: the caller must reduce the payload before retrying.
Source code in src/taskq/exceptions.py
ReservationUnavailable ¶
ReservationUnavailable(
bucket_name: str,
retry_after: timedelta,
*,
source: Literal[
"reservation", "rate_limit"
] = "reservation",
)
Bases: TaskQError
A ConcurrencyReservation slot could not be acquired.
When the upstream RateLimitDecision.retry_after is None, callers
MUST substitute DEFAULT_RESERVATION_BACKOFF. When it is
timedelta(0) (allowed decisions) callers MUST pass it through
unchanged, do NOT use a truthiness coalesce
(x or DEFAULT_RESERVATION_BACKOFF) because timedelta(0) is falsy
and would be wrongly replaced.
Source code in src/taskq/exceptions.py
ResultTooLarge ¶
Bases: TaskQError
Terminal result exceeded WorkerSettings.result_max_bytes.
Non-retryable: the actor already ran to completion and a re-run returns
the same oversized value, so retrying only burns the remaining attempts
(re-running the actor's side effects each time) before the job fails
anyway. Classified alongside PayloadValidationError in
:meth:taskq.retry.RetryClassifier.classify.
ResultUnavailable ¶
Bases: TaskQError
:meth:JobHandle.wait saw a terminal state but no usable result.
Causes:
- result TTL expired before the call;
- actor returned None while R is non-None
(treated as schema mismatch, not a value);
- row stored result=NULL for a non-success status.
reason distinguishes the two live-result causes at the instance:
"result_ttl_expired" when the row itself carries the expiry
evidence (result_expires_at in the past, which the retention
sweep leaves in place on the rows it nulls). The stamp proves the
TTL elapsed, no more: the completion write stamps
result_expires_at even when the result itself is NULL (an
actor that returned None) and the sweep only nulls rows whose
result was still stored, so a past stamp cannot distinguish a swept
result from a row that never stored one. The message asserts only
what the row proves. "not_stored" covers every other case.
Carries the row for inspection.
Source code in src/taskq/exceptions.py
RetryAfter ¶
Bases: TaskQError
Schedule retry at specific delay. Consumes retry budget by default.
Source code in src/taskq/exceptions.py
SchemaNotMigratedError ¶
Bases: TaskQError
Backend raised UndefinedTableError, the TaskQ schema is missing.
Translated by the client layer (:mod:taskq.client._jobs) from an
asyncpg.exceptions.UndefinedTableError on the enqueue/get/list/cancel
paths, so operators see an actionable message instead of a raw asyncpg
traceback. The original exception is chained via __cause__.
Source code in src/taskq/exceptions.py
ScopedIdempotencyMigrationPendingError ¶
ScopedIdempotencyMigrationPendingError(
*,
actor: str | None = None,
idempotency_key: str | None = None,
idempotency_scope: str | None = None,
detail: str | None = None,
)
Bases: TaskQError
idempotency_scope was used, but the schema has not yet had
01.00.03_01_post_idempotency_scope_drop_old_index.sql applied.
Between applying 01.00.03_01_pre_idempotency_scope.sql and its
post counterpart (the rolling-deploy window every worker's schema
passes through), BOTH the old global jobs_idempotency_key_uniq
index (on idempotency_key alone) and the new composite
jobs_idempotency_scope_key_uniq index (on (idempotency_scope,
idempotency_key)) exist simultaneously, this is deliberate, see the
"PHASE OBLIGATIONS" comment in the pre migration file, and is what
keeps pre-this-release code's unscoped ON CONFLICT (idempotency_key)
working unmodified during the window.
The cost of that safety: enqueuing the same idempotency_key under
two different idempotency_scope values satisfies the new
composite index's ON CONFLICT target (no conflict there, the
(scope, key) pair is new) but still violates the still-present old
global index, which is not covered by that ON CONFLICT target.
PostgreSQL raises UniqueViolationError for a conflict against a
non-arbiter unique index unconditionally, the library deliberately
does NOT catch that and silently fall back to a different scope's row,
because doing so would return the wrong job for the scope the caller
actually asked for, silently, which is a worse failure mode than a
loud, explicit error for a purely transitional migration-window
condition. Raised instead of letting the raw
asyncpg.UniqueViolationError propagate.
Any call, scoped or unscoped, is affected whenever its
idempotency_key already exists under a different scope: an
unscoped call that reuses a key first written under a non-default
scope raises this error just as a scoped call reusing an unscoped
key does (verified against live PostgreSQL). Only brand-new keys and
same-scope repeats are unaffected, a repeated key under the same
scope (including two unscoped calls, which share the default ''
scope) conflicts identically against both indexes for the exact same
row, which ON CONFLICT DO NOTHING on the composite index resolves
cleanly.
Resolution: confirm every worker is running the release that shipped
idempotency_scope, then apply
taskq migrate up --phase post (or a plain taskq migrate up) to
drop the old index and activate scoped dedupe, or avoid passing
idempotency_scope until that migration has run.
Source code in src/taskq/exceptions.py
ScopeViolation ¶
Bases: TaskQError
A provider depends on a shorter-lived scope than its own.
Source code in src/taskq/exceptions.py
SingletonCollisionError ¶
SingletonCollisionError(
actor: str,
blocking_job_id: UUID | None = None,
retry_after: timedelta | None = None,
)
Bases: BackpressureError
Raised when a singleton actor already has a job in pending/scheduled/running.
blocking_job_id is the UUID of the existing job from the Layer 1
pre-flight query; it is None when raised from the Layer 2
UniqueViolationError catch (the race path) because no pre-flight row
was fetched.
retry_after is computed from the blocking job's schedule_to_close
when available. It is None when the blocking job has no
schedule_to_close set, or when raised from the Layer 2 catch path.
The heartbeat_interval * 4 fallback is intentionally NOT
implemented, retry_after is computed from schedule_to_close only.
Callers who need a poll cadence when retry_after is None should poll on
their own schedule (research.md Gap 1, resolution path (a)). Reason:
heartbeat_interval is not available at the backend enqueue boundary;
propagating it would require enlarging the backend constructor surface and
is out of scope.
Source code in src/taskq/exceptions.py
Snooze ¶
Bases: TaskQError
Job returns control with new scheduled_at; does not consume retry budget.
Source code in src/taskq/exceptions.py
StreamUnavailable ¶
StreamUnavailable(
job_id: JobId,
*,
consecutive_failures: int,
elapsed: float,
last_error: BaseException,
)
Bases: TaskQError
A job stream could not re-read its job row for at least its failure budget.
Raised by either transport behind :meth:TaskQ.stream and
:meth:JobHandle.progress_stream — the Postgres poll transport's
fetches, and on the Redis path, the safety-net re-reads behind the
subscription — when every fetch has failed with a pool or connection
error for elapsed seconds (the transport's failure budget). A
blip shorter than the budget is retried silently beyond a warning
per failed poll; this is the end of the stream, so the caller's
async for cannot wait forever on a database that is not coming
back. The last failure is chained as __cause__.
Source code in src/taskq/exceptions.py
SubEnqueueError ¶
Bases: TaskQError
Raised by flush_buffer() when one or more buffered sub-job enqueues fail after parent commit.
failed_items carries each failed EnqueueArgs and the exception
that caused the enqueue to fail. The parent job has already been
marked succeeded, this exception signals that child jobs were lost.
Source code in src/taskq/exceptions.py
TaskQError ¶
Bases: Exception
Base for all library-raised exceptions.
UniqueForLockTimeoutError ¶
Bases: TaskQError
Raised when the advisory-lock wait bounding a unique_for
enqueue's preflight-then-insert exceeded its budget.
Distinct from :class:MaxPendingLockTimeoutError on purpose. That
error is a :class:BackpressureError: the caller's own load filled
the contention scope (every producer of a capped actor), the cap
check never ran, and the correct response is the same as for a cap
rejection, retry later or shed load. This error means the DEDUP
ANSWER for one (schema, actor, identity_key) could not be
determined in time: the contention scope is a single logical
entity's identity (a same-key stampede, or a black-holed holder the
server has not yet reaped), nothing about capacity is wrong, and the
correct response is to RETRY THE SAME ENQUEUE, by then the winner's
row is typically committed and the preflight returns it as a dedup
hit, which is the very outcome the wait existed to produce.
Deliberately NOT a :class:BackpressureError so handlers that react
to backpressure by shedding load or logging queue counts cannot
misreact, and deliberately NOT recorded against the
taskq.backpressure.errors counter (identity-key contention is
not a capacity signal; the unique-for-lock-timeout log event
carries the observability instead).
identity_key names the contended entity. timeout_ms is the
budget that expired. This enqueue wrote nothing: on a pool-owned
transaction the loser's transaction rolled back before any write; on
a caller-owned transaction (enqueue_with_conn /
TaskQ.with_conn) the savepoint the bounded acquire used rolled
back, the transaction remains usable, and durability of anything the
CALLER wrote alongside is the caller's decision, not this error's
claim to make.
Source code in src/taskq/exceptions.py
UnknownQueueError ¶
UnknownQueueError(
queues: Sequence[str],
known_queues: Sequence[str],
*,
actor: str | None = None,
source: Literal["submit", "worker_boot"] = "submit",
snapshot_is_ttl_bounded: bool = False,
env_queues: Sequence[str] | None = None,
)
Bases: TaskQError
Raised when strict queue-name validation refuses a queue name.
With TASKQ_QUEUES_STRICT=true the enqueue-time unserved-queue
check (ActorCapacityCache.maybe_warn_unserved_queue — the same
zero-I/O TTL-snapshot verdict every enqueue arm already consults)
becomes HARD: a queue no registered actor's assignment routes to
raises HERE, before any backend write, instead of storing an
orphaned row on a queue no worker claims. Silent data non-delivery
is the worst failure class a queueing library has, and a
one-character typo used to be enough to cause it.
Two raising sites, one class, both naming the offending queue AND the set it was judged against (the unserved-queue note's quality bar: the fix is obvious from the message alone):
- At submit (
source="submit"): the judged sets are TWO — the registered queue assignments (actor_configrows, fleet-wide DB truth the client can read without actor registration — the split-deployment constraint) and, when the process's settings carry one, its own declared set (TASKQ_QUEUES). A refusal requires the queue to miss BOTH: an env-declared but snapshot-absent queue is a new queue mid-deploy and is allowed through the registration lag; an EMPTY snapshot (no worker ever started, the first deploy) carries zero evidence of stranding and never refuses; an UNAVAILABLE snapshot (read failure, capability-less backend) never refuses — a database blip or migration window must not become a submit outage, and the check re-arms on the next refresh. Every ambiguous state resolves to ALLOW + the advisory note; over-rejection is strictly worse than over-permission. The snapshot is TTL-bounded, so a queue registered seconds ago may not be in it yet; when the process declares no queue set the snapshot is the only source and the message says so — deploy workers before clients for new queues, that deploy order is the operator's protection for the first submits. Nothing was stored when this raises — the refusal is at the door. - At worker boot (
source="worker_boot"): the judged set is the worker's own configured consume set (TASKQ_QUEUES), and the coverage is the fleet's stored assignments read back aftersync_actor_config. A configured queue nothing routes to is config drift between deploy and code; the worker refuses to boot. Deliberately worker-only: actors register only in the worker process, so a client-side boot check of this shape would brick split deployments (web submits, worker consumes).
Source code in src/taskq/exceptions.py
1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 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 | |
env_queues
instance-attribute
¶
WorkerOwnershipMismatch ¶
Bases: TaskQError
Terminal write predicate failed: job exists but is owned by a different worker.
Source code in src/taskq/exceptions.py
ErrorReporter ¶
Bases: Protocol
Vendor-neutral hook for routing terminal job failures to external systems.
Implementations capture the error and job row, then forward to a
vendor-specific backend (Sentry, Datadog, a DLQ, etc.). The library
calls :meth:report when a job reaches a terminal failure state ,
either because retries were exhausted or because the error was
non-retryable.
The call is wrapped in a try/except by :func:invoke_error_reporter;
a failing reporter never crashes the worker. Reporter failures are
counted on the taskq.error_reporter.failures counter with a
reporter_type attribute.
Register an :class:ErrorReporter instance as a DI provider
(registry.register_value(ErrorReporter, Scope.PROCESS, my_reporter))
or pass it directly to the worker bootstrap.
NullErrorReporter ¶
Default no-op :class:ErrorReporter, silently drops all reports.
Used when no vendor-specific error routing is configured. Instances are stateless and safe to share.
ProgressEvent ¶
Bases: BaseModel
Point-in-time progress snapshot published to Redis for SSE/stream fanout.
Covers both kind="progress" (incremental update) and
kind="state_change" (terminal or status transition) events. The
exclude_none=True flag on :meth:model_dump_json suppresses null
fields so the JSON payload stays compact on the wire.
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.
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.
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 | |
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.",
)
RetryPolicy ¶
Bases: BaseModel
Policy controlling retry behaviour for an actor.
backoff
class-attribute
instance-attribute
¶
OIDCSettings ¶
Bases: DotEnvConfig
OIDC SSO configuration (loaded from TASKQ_OIDC_* env vars).
issuer
class-attribute
instance-attribute
¶
issuer: str = Field(
default="",
description="OIDC discovery issuer URL (e.g. https://login.microsoftonline.com/{tenant}/v2.0).",
)
client_id
class-attribute
instance-attribute
¶
client_secret
class-attribute
instance-attribute
¶
redirect_uri
class-attribute
instance-attribute
¶
redirect_uri: str = Field(
default="",
description="Must match the app registration's configured redirect URI.",
)
session_secret
class-attribute
instance-attribute
¶
session_secret: SecretStr = Field(
default="",
description="Signing key for session cookies; use >=32 bytes of random data. Rotate to invalidate all sessions.",
)
session_max_age_seconds
class-attribute
instance-attribute
¶
session_max_age_seconds: int = Field(
default=28800,
ge=60,
description="Session lifetime (s). Default 8h.",
)
scope
class-attribute
instance-attribute
¶
scope: str = Field(
default="openid profile email",
description="OIDC scopes. Add 'Group.Read.All' for the Entra overage group_resolver (Graph API /me/memberOf).",
)
group_claim
class-attribute
instance-attribute
¶
group_claim: str | None = Field(
default=None,
description="ID token claim name for groups (e.g. 'groups', 'roles'). None = authentication-only authorization.",
)
allowed_groups
class-attribute
instance-attribute
¶
SAMLSettings ¶
Bases: DotEnvConfig
SAML SSO configuration (loaded from TASKQ_SAML_* env vars).
entity_id
class-attribute
instance-attribute
¶
acs_url
class-attribute
instance-attribute
¶
idp_entity_id
class-attribute
instance-attribute
¶
idp_sso_url
class-attribute
instance-attribute
¶
idp_x509_cert
class-attribute
instance-attribute
¶
sp_x509_cert
class-attribute
instance-attribute
¶
sp_x509_cert: str | None = Field(
default=None,
description="SP cert (signed requests / encrypted assertions).",
)
sp_private_key
class-attribute
instance-attribute
¶
session_secret
class-attribute
instance-attribute
¶
session_max_age_seconds
class-attribute
instance-attribute
¶
session_max_age_seconds: int = Field(
default=28800,
ge=60,
description="Session lifetime (s). Default 8h.",
)
group_attribute
class-attribute
instance-attribute
¶
allowed_groups
class-attribute
instance-attribute
¶
allow_cookieless_fallback
class-attribute
instance-attribute
¶
allow_cookieless_fallback: bool = Field(
default=False,
description="TASKQ_SAML_ALLOW_COOKIELESS_FALLBACK. Opt in to the SAML cookie-less ACS fallback: accept a callback with no usable correlation cookie when its validated InResponseTo names a pending AuthnRequest this process issued. Serves browsers that block the cross-site cookie; nothing ties the response to the browser posting it, so a captured signed response can be planted on a cookie-less victim (login CSRF) -- opt in only if that tradeoff is acceptable. Default off; see docs/guides/sso.md.",
)
TaskQSettings ¶
Bases: DotEnvConfig
Top-level TaskQ runtime configuration.
pg_dsn
class-attribute
instance-attribute
¶
pg_dsn: PostgresDsn = Field(
default=PostgresDsn(
"postgresql://taskq:taskq@localhost:5432/taskq"
),
description="Direct (non-PgBouncer) DSN. LISTEN/NOTIFY and advisory locks need a session.",
)
pg_is_pooled
class-attribute
instance-attribute
¶
pg_is_pooled: bool = Field(
default=False,
description="TASKQ_PG_IS_POOLED. Declare that the DSN(s) TaskQ builds pools from route through a transaction-mode pooler (PgBouncer pool_mode=transaction, RDS Proxy, Supavisor). asyncpg cannot detect a pooler from the wire protocol - PgBouncer speaks plain Postgres - so the operator declares the topology. When True every pool TaskQ builds disables the prepared-statement cache (statement_cache_size=0, max_cached_statement_lifetime=0, overriding TASKQ_STATEMENT_CACHE_SIZE / TASKQ_MAX_CACHED_STATEMENT_LIFETIME) so a prepared statement can never outlive the server connection a pooler remaps underneath it, and the worker treats pooler-remap statement errors (SQLSTATE 26000 unnamed_prepared_statement, 42P05 duplicate_prepared_statement) as transient instead of loud surprises. Applies only to pools TaskQ builds: bring-your-own pools (WorkerConnections factories, a caller-supplied pool) must set the same create_pool kwargs themselves. Leave False when every DSN reaches Postgres directly.",
)
schema_name
class-attribute
instance-attribute
¶
schema_name: str = Field(
default="taskq",
validator=_schema_name_validator,
description="Postgres schema for all TaskQ tables.",
)
queues_strict
class-attribute
instance-attribute
¶
queues_strict: bool = Field(
default=False,
description="TASKQ_QUEUES_STRICT. When True, the enqueue-time unserved-queue check becomes HARD: a submit whose queue no registered actor's assignment routes to raises UnknownQueueError before any backend write (covering direct enqueues, every batch arm, and sub-job fan-out), instead of storing an orphaned row on a queue no worker claims. The worker-side twin fails the boot when a configured queue (TASKQ_QUEUES) has no registered actor routing to it — worker-only, because actors register only there. Default False: the unserved-queue NOTE stays the only signal. Per-submit escape: allow_unregistered=True for genuinely dynamic queue names.",
)
redis_url
class-attribute
instance-attribute
¶
redis_url: RedisDsn | None = Field(
default=None,
description="Optional Redis URL. Required for real-time progress fanout.",
)
environment
class-attribute
instance-attribute
¶
environment: str | None = Field(
default=None,
description="TASKQ_ENVIRONMENT. Deployment environment label. The unauthenticated-admin WARNING ('admin-ui-no-auth') fires in EVERY environment whenever the admin UI is served without auth_dependency. 'dev' and 'development' additionally skip the fail-closed RuntimeError (the WARNING then notes the absence is the dev exemption); any other value (or None/empty) fails closed when admin_ui_require_auth is True.",
)
admin_max_sse_connections
class-attribute
instance-attribute
¶
admin_max_sse_connections: int = Field(
default=50,
ge=1,
description="TASKQ_ADMIN_MAX_SSE_CONNECTIONS. Maximum concurrent SSE connections the admin UI will serve. Used to size the connection-limit semaphore.",
)
progress_max_sse_connections
class-attribute
instance-attribute
¶
progress_max_sse_connections: int = Field(
default=50,
ge=1,
description="TASKQ_PROGRESS_MAX_SSE_CONNECTIONS. Maximum concurrent per-job progress SSE streams this process will serve. Each holds a Redis pubsub subscription and an asyncio task for as long as the client stays connected, so an uncapped endpoint is a resource-exhaustion surface on the app hosting the pipeline.",
)
progress_require_auth
class-attribute
instance-attribute
¶
progress_require_auth: bool = Field(
default=True,
description="TASKQ_PROGRESS_REQUIRE_AUTH. When True (the default), taskq.web.progress.create_router raises RuntimeError if auth_dependency is None in a non-dev environment, failing closed. Set to False to suppress the error and allow unauthenticated per-job progress/state endpoints in non-dev (not recommended - only for deployments that authenticate at the ingress).",
)
admin_host
class-attribute
instance-attribute
¶
admin_host: str = Field(
default="0.0.0.0",
description="TASKQ_ADMIN_HOST. Bind address for ``taskq ui serve``.",
)
admin_port
class-attribute
instance-attribute
¶
admin_port: int = Field(
default=8080,
ge=1,
le=65535,
description="TASKQ_ADMIN_PORT. Bind port for ``taskq ui serve``.",
)
admin_url
class-attribute
instance-attribute
¶
admin_url: str = Field(
default="http://localhost:8080",
description="TASKQ_ADMIN_URL. Public base URL of the admin UI as seen from a browser. Used by the example trigger app to construct redirect URLs after enqueueing. In a shared-container deployment this is the external address of the admin process (e.g. http://localhost:8001). Override when admin and trigger app are on different hosts or ports.",
)
admin_ui_polling_interval_seconds
class-attribute
instance-attribute
¶
admin_ui_polling_interval_seconds: float = Field(
default=2.0,
ge=0.1,
description="TASKQ_ADMIN_UI_POLLING_INTERVAL_SECONDS. How often the admin UI polls PG in polling/degraded mode. Injected as poll_interval_ms into every template.",
validator=_finite_float,
)
admin_worker_liveness_seconds
class-attribute
instance-attribute
¶
admin_worker_liveness_seconds: int = Field(
default=30,
ge=1,
description="TASKQ_ADMIN_WORKER_LIVENESS_SECONDS. How recently a worker must have written last_seen_at to count as alive: it drives the admin UI's 'queue has pending jobs but no alive worker' banner and the leader's watchdog_healthy verdict, and on the worker side the leader's taskq.queue.live_workers gauge and the stranded-jobs detector's unserved-queue arm. Must comfortably exceed TASKQ_HEARTBEAT_INTERVAL (default 10 s), so the default 30 s is three beats; a deployment that lengthens the heartbeat, or whose PG is cross-region, has to raise this or every healthy worker reads as dead. Measured by Postgres against clock_timestamp(), never by the admin process's own clock.",
)
admin_ui_allow_rate_limit_reset
class-attribute
instance-attribute
¶
admin_ui_allow_rate_limit_reset: bool = Field(
default=False,
description="TASKQ_ADMIN_UI_ALLOW_RATE_LIMIT_RESET. When True, the admin UI shows a reset button on the rate-limits page and serves the POST /rate-limits/{bucket_name}/reset endpoint. Default False for safety - prevents accidental resets in production.",
)
admin_ui_require_auth
class-attribute
instance-attribute
¶
admin_ui_require_auth: bool = Field(
default=True,
description="TASKQ_ADMIN_UI_REQUIRE_AUTH. When True (the default), create_router raises RuntimeError if auth_dependency is None in a non-dev environment, failing closed. Set to False to suppress the error and allow an unauthenticated admin UI in non-dev (not recommended - only for air-gapped or localhost-only deployments).",
)
admin_ui_frame_ancestors
class-attribute
instance-attribute
¶
admin_ui_frame_ancestors: str = Field(
default="none",
validator=_frame_ancestors_validator,
description="TASKQ_ADMIN_UI_FRAME_ANCESTORS. Who may frame admin pages: 'none' (the default, nobody) or 'self' (the admin UI's own origin, for a host app that embeds the admin UI in its own dashboard). Emitted as both 'Content-Security-Policy: frame-ancestors ...' and the legacy 'X-Frame-Options' (DENY / SAMEORIGIN). CSRF is no defence against UI redress: the framed page is the real, authenticated, same-origin page, so a tricked click carries a valid token.",
)
admin_ui_secure_cookies
class-attribute
instance-attribute
¶
admin_ui_secure_cookies: bool = Field(
default=True,
description="TASKQ_ADMIN_UI_SECURE_COOKIES. Sets the 'Secure' flag on the admin UI's CSRF cookie. A configured value, not one inferred from request.url.scheme: behind a TLS-terminating edge (Azure Application Gateway, App Service) the app sees plain http, so an inferred flag is silently dropped on a connection the browser reached over HTTPS. Set False only for local http dev, where a Secure cookie is rejected by the browser and the admin UI stops working.",
)
admin_acquire_timeout
class-attribute
instance-attribute
¶
admin_acquire_timeout: float = Field(
default=5.0,
gt=0,
description="TASKQ_ADMIN_ACQUIRE_TIMEOUT (seconds). Bounds every wait an admin UI or progress request makes for a backend resource before its own query runs: a Postgres pool checkout and a Redis read. A pool with every connection wedged, or a black-holed broker, answers the request with 503 (Retry-After: 2) after this long instead of hanging it - and every other request behind it - until the client gives up. The query itself is bounded by the pool's command_timeout.",
validator=_finite_float,
)
admin_actions_enabled
class-attribute
instance-attribute
¶
admin_actions_enabled: bool = Field(
default=False,
description="TASKQ_ADMIN_ACTIONS_ENABLED. When True, the admin UI permits state-changing actions: run schedule now, enable/disable/skip a schedule, retry job, cancel job. Default False - prevents on-demand triggering of registered business logic, and silent suppression of scheduled work, via the admin UI without explicit opt-in. Separate from auth_dependency, which controls read access to all admin routes.",
)
pg_credential_provider
class-attribute
instance-attribute
¶
pg_credential_provider: str | None = Field(
default=None,
description="TASKQ_PG_CREDENTIAL_PROVIDER. Module:attr reference to a PgCredentialProvider (e.g. myapp.auth:make_provider) - an instance, a zero-arg factory returning one, or the provider class. Every Postgres pool and dedicated connection is then built through it, so SIGHUP / TASKQ_RELOAD_INTERVAL rotate real credentials. The CLI options --pg-credential-provider (taskq worker / migrate / ui serve) override it. The ref is resolved, not validated, at load time: the import lives in the CLI so a bad ref exits 1 with a pointed message instead of a settings traceback. See docs/guides/managed-identities.md.",
)
redis_credential_provider
class-attribute
instance-attribute
¶
redis_credential_provider: str | None = Field(
default=None,
description="TASKQ_REDIS_CREDENTIAL_PROVIDER. Module:attr reference to a RedisCredentialProvider, in the same shapes as pg_credential_provider. Requires TASKQ_REDIS_URL. Overridden by --redis-credential-provider.",
)
reload_interval
class-attribute
instance-attribute
¶
reload_interval: float | None = Field(
default=None,
gt=0,
description="TASKQ_RELOAD_INTERVAL (seconds). Cadence of the credential hot-reload (the same path as SIGHUP) on the worker and on `taskq ui serve`: every provider-backed pool and connection is rebuilt on a fresh credential with no external signal required - the rotation path for platforms without SIGHUP (e.g. Windows) and for hands-off scheduled rotation (e.g. ~720s for AWS IAM's 15-minute tokens). Unset, the cadence is derived from the lease the provider grants when it reports one (a Vault dynamic credential is rebuilt at half its lease TTL - see taskq.auth.ReloadSchedule); a username-bearing provider that reports no lease then warns at startup, and only SIGHUP / deps.request_reload() rotate it. Only factory-backed resources are rebuilt; DSN/static credentials are unaffected.",
)
sso_backend
class-attribute
instance-attribute
¶
sso_backend: str = Field(
default="none",
validator=_sso_backend_validator,
description="TASKQ_SSO_BACKEND. Selects the SSO backend for the admin UI: 'none' (default, unauthenticated/BYO-auth), 'oidc' (taskq[oidc]), or 'saml' (taskq[saml]). See docs/guides/sso.md.",
)
health_token
class-attribute
instance-attribute
¶
health_token: SecretStr | None = Field(
default="",
description="TASKQ_HEALTH_TOKEN. Bearer token for machine-to-machine access to the health/metrics endpoints that `taskq ui serve` exposes: when set, those routes require a matching 'Authorization: Bearer <token>' header. The worker's own health listeners never check it - the unix socket is guarded by filesystem permissions, and the TASKQ_HEALTH_PORT TCP listener (/live, /ready) answers without a credential, so keep that port pod-network-only. Leave empty for unauthenticated access - but see health_require_token, which fails closed on an empty token outside dev.",
)
health_require_token
class-attribute
instance-attribute
¶
health_require_token: bool = Field(
default=True,
description="TASKQ_HEALTH_REQUIRE_TOKEN. When True (the default), taskq ui serve raises RuntimeError if health_token is empty in a non-dev environment, failing closed. Set to False to suppress the error and allow unauthenticated health/metrics endpoints in non-dev (e.g. when relying on network policy / cluster-internal-only access instead of a bearer token. Many k8s liveness/readiness probes don't send auth headers by default, so enabling the token may require updating the probe config too).",
)
migrate_on_start
class-attribute
instance-attribute
¶
migrate_on_start: bool = Field(
default=False,
description="TASKQ_MIGRATE_ON_START. When True, apply pending migrations before the admin UI accepts its first request. Aborts startup if migrations fail. Consumed ONLY by `taskq ui serve` -- the worker ignores it (and warns when it is set), because N worker replicas racing to migrate is the concurrent-migration hazard migrations are supposed to avoid. Migrate from a pre-deploy job or init container.",
)
example_host
class-attribute
instance-attribute
¶
example_host: str = Field(
default="0.0.0.0",
description="TASKQ_EXAMPLE_HOST. Bind address for the example trigger app (uvicorn). Only consumed by the example app; ignored by the worker and admin UI.",
)
example_port
class-attribute
instance-attribute
¶
example_port: int = Field(
default=8000,
ge=1,
le=65535,
description="TASKQ_EXAMPLE_PORT. Bind port for the example trigger app (uvicorn). Only consumed by the example app; ignored by the worker and admin UI.",
)
idempotency_key_max_bytes
class-attribute
instance-attribute
¶
idempotency_key_max_bytes: int = Field(
default=MAX_IDEMPOTENCY_KEY_BYTES,
ge=1,
le=IDEMPOTENCY_KEY_BYTES_CEILING,
description="TASKQ_IDEMPOTENCY_KEY_MAX_BYTES. Maximum UTF-8 byte length of idempotency_key, and of idempotency_scope, each. Bytes rather than characters because the real bound is the composite unique index jobs_idempotency_scope_key_uniq: a btree v4 entry cannot exceed 2704 bytes, counted encoded. The ceiling keeps scope + key + index-tuple overhead under that, so no value here can turn a valid enqueue into a raw Postgres 'index row size ... exceeds btree version 4 maximum'. Raise it when keys are derived from URLs, composite business keys or opaque vendor cursors.",
)
statement_cache_size
class-attribute
instance-attribute
¶
statement_cache_size: int = Field(
default=DEFAULT_STATEMENT_CACHE_SIZE,
ge=0,
description="TASKQ_STATEMENT_CACHE_SIZE. Size of the per-connection prepared-statement LRU asyncpg keeps on every pool TaskQ builds. asyncpg's default of 100 thrashes on TaskQ's read paths, list_jobs alone renders 384+ filter-combination SQL variants (a measured 90-96% steady-state miss rate, each miss re-paying the Parse/Describe round trips; see benchmarks/ab_stmt_cache.py and docs/guides/ops.md). 512 covers the variant space with headroom. 0 disables the statement cache. Applies only to pools TaskQ builds: bring-your-own pools (WorkerConnections factories, a caller-supplied pool) must pass the same create_pool kwargs themselves, TaskQ cannot resize a pool it did not build.",
)
max_cached_statement_lifetime
class-attribute
instance-attribute
¶
max_cached_statement_lifetime: int = Field(
default=DEFAULT_MAX_CACHED_STATEMENT_LIFETIME,
ge=0,
description="TASKQ_MAX_CACHED_STATEMENT_LIFETIME (seconds). How long a prepared statement may stay in asyncpg's per-connection cache on every pool TaskQ builds. asyncpg's default of 300 s re-prepares statements on workers that outlive it; 3600 s (1 h) keeps the textually-stable write hot loop (dispatch, enqueue, terminal updates, sweeps) prepared across ticks without pinning prepared plans for the process lifetime. 0 caches statements indefinitely. Same TaskQ-built-pools-only scope as statement_cache_size.",
)
max_pending_lock_timeout_ms
class-attribute
instance-attribute
¶
max_pending_lock_timeout_ms: float = Field(
default=5000.0,
description="TASKQ_MAX_PENDING_LOCK_TIMEOUT_MS (milliseconds). Bounded wait for the max_pending advisory lock on the single-enqueue path (the count-then-insert serialization per capped actor). Exhaustion raises MaxPendingLockTimeoutError, the same typed backpressure treatment as a cap rejection, and denials consume retry budget, so widen this during an outage that slows lock holders rather than letting the fixed ceiling convert slow holders into refused enqueues. Widening past the default re-derives the per-query bound of every pool the TaskQ client builds itself, so the wider budget is delivered end to end; at or below the default the client path delivers the budget clamped to 80% of that pool bound, a share of the 10 s shipped bound exceeds the 5 s default budget, so the defaults are delivered in full, and the margin is what the server-side lock_timeout needs to fire before the pool's own timer. 0 or less waits indefinitely server-side (the lock_timeout GUC convention shared with the sibling budgets), a pool TaskQ builds still applies its per-query bound, so set a large finite value there instead.",
validator=_finite_float,
)
unique_for_lock_timeout_ms
class-attribute
instance-attribute
¶
unique_for_lock_timeout_ms: float = Field(
default=5000.0,
description="TASKQ_UNIQUE_FOR_LOCK_TIMEOUT_MS (milliseconds). Bounded wait for the unique_for single-flight advisory lock on the single-enqueue path (the identity preflight-then-insert serialization). Exhaustion raises UniqueForLockTimeoutError with retry-yields-dedup guidance; the correct contention outcome is usually the dedup return, so a unique_for caller may want a longer wait than the max_pending budget before giving up on the answer. Separate knob from max_pending_lock_timeout_ms because the two budgets bound different semantics (identity dedup vs capacity admission). Widening past the default re-derives the per-query bound of every pool the TaskQ client builds itself (see TASKQ_MAX_PENDING_LOCK_TIMEOUT_MS for the delivery rule); at or below the default the client path delivers the budget clamped to 80% of that pool bound. 0 or less waits indefinitely server-side (the lock_timeout GUC convention shared with the sibling budgets) , a pool TaskQ builds still applies its per-query bound, so set a large finite value there instead.",
validator=_finite_float,
)
idempotency_lock_timeout_ms
class-attribute
instance-attribute
¶
idempotency_lock_timeout_ms: float = Field(
default=5000.0,
description="TASKQ_IDEMPOTENCY_LOCK_TIMEOUT_MS (milliseconds). Bounded wait for the idempotency token INSERT's speculative-lock conflict on the single-enqueue path, another transaction's UNCOMMITTED row with the same (idempotency_scope, idempotency_key) pair. On a transactional consumer the holder is the actor's own open transaction (unbounded by default), so this budget bounds the VICTIM; exhaustion raises IdempotencyKeyLockTimeoutError, meaning the dedup answer could not be determined in time, retry the same enqueue, which typically dedupes against the now-visible winner. Widening past the default re-derives the per-query bound of every pool the TaskQ client builds itself (see TASKQ_MAX_PENDING_LOCK_TIMEOUT_MS for the delivery rule); at or below the default the client path delivers the budget clamped to 80% of that pool bound. 0 or less waits indefinitely server-side (the lock_timeout GUC convention shared with the sibling budgets) , a pool TaskQ builds still applies its per-query bound, so set a large finite value there instead.",
validator=_finite_float,
)
timescaledb_hypertables
class-attribute
instance-attribute
¶
timescaledb_hypertables: bool = Field(
default=False,
description="TASKQ_TIMESCALEDB_HYPERTABLES. Opt in to TimescaleDB hypertables for the retention tables (jobs_archive, job_attempts_archive on their time columns, job_events on occurred_at) and Timescale-managed retention policies built from TASKQ_ARCHIVE_RETENTION_PERIOD and TASKQ_EVENT_RETENTION_PERIOD. Default false: with the flag off, setup issues ZERO new statements against the server and behavior is identical to plain Postgres, which remains fully supported. With the flag true, `taskq migrate up` requires the server to offer and load the timescaledb extension and fails loudly naming this setting when it does not (it never silently degrades); see docs/guides/timescaledb.md. The worker and client never touch this flag: the conversion is deploy-step DDL, and the sweeps keep running unchanged alongside the retention policies.",
)
oidc
property
¶
Lazily loaded OIDC sub-config (TASKQ_OIDC_* env vars).
Backed by dotenvmodel's cached() singleton: the environment is
read on first access and the same instance returned thereafter.
Reload (e.g. on SIGHUP): call OIDCSettings.cached().reload() to
re-read the environment and mutate the shared instance in place -
every holder observes the new values - or
OIDCSettings.reset_cached() to force the next access to
re-load. Tests that change TASKQ_OIDC_* mid-process must do
the same (or use cached_override()).
saml
property
¶
Lazily loaded SAML sub-config (TASKQ_SAML_* env vars).
See :attr:oidc for the singleton/caching semantics and the
SIGHUP-style reload recipe (SAMLSettings.cached().reload() /
SAMLSettings.reset_cached()).
is_dev_environment
property
¶
Whether this process is labeled a development environment.
The dev label is the single carve-out from the fail-closed auth
gates (admin UI, health/metrics token, progress router):
TASKQ_ENVIRONMENT set to dev or development lets those
surfaces start without auth so local development needs no token
machinery. Any other value - including unset - is not dev, so the
gates fail closed.
load
classmethod
¶
load(
env: str | None = None,
*,
override: bool | None = None,
env_dir: Path | str | None = None,
read_dotfiles: bool | None = None,
read_environ: bool | None = None,
load_local: bool | None = None,
) -> Self
Load settings via dotenvmodel's cascading .env discovery.
All parameters are forwarded to DotEnvConfig.load unchanged
(resolution: explicit argument > DOTENV_* env var > default).
override=None keeps dotenvmodel's default precedence, the
process environment beats .env files; pass override=True
or set DOTENV_OVERRIDE=true to make .env files win instead.
read_dotfiles=False / read_environ=False disable the
.env cascade / the process environment respectively, per
dotenvmodel's documented symmetry.
dotenvmodel logs a WARNING ("No .env files found in .env file is present - noisy on every CLI
invocation in projects that configure purely via real environment
variables. read_dotfiles=False is not the answer: it silences
the warning but disables the .env cascade entirely, a
documented core TaskQ feature. This override instead installs a
:class:_NoEnvFilesWarningFilter on the dotenvmodel logger
for the duration of the call, dropping only that one warning;
every other dotenvmodel warning (e.g. an invalid DOTENV_*
knob value) remains visible.
WorkerSettings.load inherits this override via MRO, so worker
startup gets the same quiet load.
Source code in src/taskq/settings.py
resolve_cascade_value
classmethod
¶
One variable's value through :meth:load's own layer resolution,
without loading (or validating) the model.
Resolution is EXACTLY load()'s, per dotenvmodel's own tier and
precedence rules: every behavior knob resolves through
resolve_load_params (explicit DOTENV_* env var > default),
the dotfile cascade is read with read_env_files (never written
into os.environ), read_dotfiles=False skips the file layer
entirely, read_environ=False excludes the process environment
as a value source, and the winner is process-env-vs-dotfile per
the resolved override (default: the process environment wins).
The per-field winner selection replicates dotenvmodel's private
config._resolve_raw_value: the function is not exported, so
its six-line policy is restated here and pinned by
tests/test_client_cli_first_use_bounded.py's precedence tests
in both directions.
Why this exists: load() validates EVERY field, so a caller
that needs one value (the client's schema default, its lock-budget
overlay) would fail on a malformed setting it never uses.
This method resolves the value; validation stays the caller's,
scoped to the fields it actually consumes.
The "No .env files found" warning is filtered for the duration of
the cascade read, exactly as :meth:load filters it.
Source code in src/taskq/settings.py
WorkerSettings ¶
Bases: TaskQSettings
Worker-specific configuration with three-pool sizing and dual-DSN support.
Extends :class:TaskQSettings with pool-size knobs, dual-DSN fields, and
the validated lock_lease >= max(heartbeat_interval,
heartbeat_command_timeout) + (max_heartbeat_failures + 1) *
(heartbeat_interval + heartbeat_command_timeout) invariant: the
lease must outlive the worst coherent failed-beat cascade to the
heartbeat's isolate decision - the last good beat's tail plus the
failed cycles - with the per-tick command budget (shared by each
failed beat's sequence AND teardown) making the cycle bound true
by enforcement.
pg_dsn_direct
class-attribute
instance-attribute
¶
pg_dsn_direct: PostgresDsn | None = Field(
default=None,
description="TASKQ_PG_DSN_DIRECT; falls back to pg_dsn when absent. Bypasses PgBouncer - used by dispatcher_pool, heartbeat_pool, notify_conn, and leader_conn.",
)
pg_dsn_pooled
class-attribute
instance-attribute
¶
pg_dsn_pooled: PostgresDsn | None = Field(
default=None,
description="TASKQ_PG_DSN_POOLED; falls back to pg_dsn when absent. May route through PgBouncer transaction mode - used by worker_pool only.",
)
dispatcher_pool_size
class-attribute
instance-attribute
¶
dispatcher_pool_size: int = Field(
default=4,
ge=1,
description="TASKQ_DISPATCHER_POOL_SIZE. Max connections for the dispatcher pool. Bypasses PgBouncer.",
)
dispatcher_command_timeout
class-attribute
instance-attribute
¶
dispatcher_command_timeout: float = Field(
default=5.0,
ge=1.0,
description="TASKQ_DISPATCHER_COMMAND_TIMEOUT (seconds). Per-query timeout for the dispatcher pool and the TaskQ-built leader connections (election, cron, monitor), and the single deadline wrapped around each period-1 leader-loop iteration (scheduled_wake, cron): a stalled PG errors the iteration instead of hanging the loop past its staleness budget. Checked at load time when the watchdog is enabled: timeout + the 1.0s leader-loop period must be < max(period x watchdog_tick_grace_factor, watchdog_stale_floor) for the period-1 leader loops (scheduled_wake, cron), so a timeout-capped iteration can never false-trip the stale-loop detector on a healthy worker. The producer loop is not checked (its multi-statement dispatch_batch is not wrapped in a single asyncio.timeout). For the dispatcher POOL this configured value is the FLOOR, not the applied bound: the admission-path rate-limit acquires run on that pool, so TaskQ re-derives the pool's command_timeout upward from a widened token_bucket_lock_timeout_ms or sliding_window_lock_timeout_ms (max(configured, widest budget / 0.8), connections.lock_budget_command_timeout_secs) the same way the client pool follows the enqueue lock budgets, a widened admission budget is honored end to end instead of being silently truncated by the pool's own client-side timer. The leader/notify dedicated connections keep the configured value: no admission acquire runs on them.",
validator=_finite_float,
)
dispatch_oversample
class-attribute
instance-attribute
¶
dispatch_oversample: int = Field(
default=2,
ge=1,
le=1000,
description="TASKQ_DISPATCH_OVERSAMPLE. Multiplier for per-actor candidate gathering in the dispatch SQL. Each LATERAL reads residual x oversample candidates. Higher values absorb more identity collisions and multi-producer contention. Default 2 (tolerates 50% dupe identities). Set 1 when no identity_key is used and single-producer. The window also bounds how far one dispatch round can slide past rows locked by concurrent dispatchers: oversample dispatchers polling the same (actor, queue) can hold the whole window at once, so size it at or above the number of dispatchers that routinely poll the same actor+queue. A round that finds its whole window locked expands it geometrically (up to 8x) while claimable rows remain, which covers transient oversubscription, the setting governs the steady state so the common case never pays the expansion round trip.",
)
claim_cursor_reset_seconds
class-attribute
instance-attribute
¶
claim_cursor_reset_seconds: float = Field(
default=0.0,
ge=0.0,
le=86400.0,
validator=_finite_float,
description="TASKQ_CLAIM_CURSOR_RESET_SECONDS. OPT-IN (default 0 = disabled): the claim cursor's jitter reset interval, in seconds - the MVCC-horizon hygiene pair's worker-side half (backend/_claim_cursor.py). When enabled, each worker's dispatch keeps a per-queue high-water mark of successfully-claimed job ids (UUIDv7, time-ordered) and bounds the claim query's candidate predicate at it, so the B-tree seek can land on live tuples instead of degenerating through the dead tuples a pinned MVCC horizon leaves behind. The bound is a selection predicate: any pending row whose id falls below it (a skewed producer's clock, a row the round's window never reached) is invisible to the claim query until the reset forgets the cursor - the reset is this interval, jittered +/-50% per entry so a fleet's resets never synchronize, armed at the entry's creating claim and never extended by later claims. Round-robin queues are exempt (cohort fairness is id-order-blind by design). Disabled by default on measured evidence: on this codebase's multi-surface claim CTE the bound's full-path win is plan-dependent (the spiral's planner blindness - dead rows are invisible to every fresh snapshot, so ANALYZE's estimates stay healthy while the index rots - keeps the probes on index choices the bound cannot position), so the bounded-stranding trade buys no robust end-to-end improvement; see tests/perf/test_mvcc_claim_degradation.py for the curves and the TaskQClaimLatencyDegraded runbook for the primary cure. Enable it for fleets whose shape matches the isolated-surface measurement (single-queue, monotone claim streams, fresh statistics).",
)
dispatch_scope_by_home_queue
class-attribute
instance-attribute
¶
dispatch_scope_by_home_queue: bool = Field(
default=False,
description="TASKQ_DISPATCH_SCOPE_BY_HOME_QUEUE. Deprecated no-op, accepted so configurations that set it keep loading: dispatch is assignment-routed now (the jobs row carries the routing decision the old per-actor-capacity scoping approximated), so the flag has nothing left to apply. The worker logs a deprecated-setting warning at startup when it is set; remove it from the environment.",
)
token_bucket_lock_timeout_ms
class-attribute
instance-attribute
¶
token_bucket_lock_timeout_ms: float = Field(
default=5000.0,
description="TASKQ_TOKEN_BUCKET_LOCK_TIMEOUT_MS (milliseconds). Bounded wait for the rate_limit_buckets row lock on the token-bucket Postgres acquire and refund. With the Postgres rate-limit fallback enabled a Redis outage funnels every admission through this lock, so the bound is what stops one black-holed holder stalling a bucket's admission. Exhaustion is an admission DENIAL, not a failure: the acquire fails closed and the denial's retry hint is one more budget, so shortening this tightens the re-check interval rather than refusing work. These acquires run on the dispatcher pool, whose client-side command_timeout would otherwise truncate a budget wider than its 5.0s floor before the server-side lock_timeout could fire: widening this past the default re-derives the TaskQ-built dispatcher pool's per-query bound upward (budget / 0.8), so the wider budget is delivered end to end, the same reconciliation the client pool applies to the enqueue lock budgets. A caller-supplied dispatcher pool keeps its own timeouts; size it above the budgets you set. 0 or less waits indefinitely server-side (the lock_timeout GUC convention shared with the sibling budgets), a TaskQ-built pool still applies its per-query bound, so set a large finite value there instead.",
validator=_finite_float,
)
sliding_window_lock_timeout_ms
class-attribute
instance-attribute
¶
sliding_window_lock_timeout_ms: float = Field(
default=5000.0,
description="TASKQ_SLIDING_WINDOW_LOCK_TIMEOUT_MS (milliseconds). Bounded wait for the sliding window's admission lock on Postgres, the per-bucket advisory lock on the log style and the rate_limit_buckets row lock on the GCRA style. Separate knob from token_bucket_lock_timeout_ms because the two limiter shapes hold their locks across different critical sections and a deployment may run only one of them. Exhaustion fails closed as a denial whose retry hint is one more budget. These acquires run on the dispatcher pool, whose client-side command_timeout would otherwise truncate a budget wider than its 5.0s floor before the server-side lock_timeout could fire: widening this past the default re-derives the TaskQ-built dispatcher pool's per-query bound upward (budget / 0.8), so the wider budget is delivered end to end, the same reconciliation the client pool applies to the enqueue lock budgets. A caller-supplied dispatcher pool keeps its own timeouts; size it above the budgets you set. 0 or less waits indefinitely server-side (the lock_timeout GUC convention shared with the sibling budgets), a TaskQ-built pool still applies its per-query bound, so set a large finite value there instead.",
validator=_finite_float,
)
heartbeat_pool_size
class-attribute
instance-attribute
¶
heartbeat_pool_size: int = Field(
default=4,
ge=1,
description="TASKQ_HEARTBEAT_POOL_SIZE. Max connections for the heartbeat pool. Bypasses PgBouncer.",
)
heartbeat_command_timeout
class-attribute
instance-attribute
¶
heartbeat_command_timeout: float = Field(
default=2.0,
gt=0.0,
description="TASKQ_HEARTBEAT_COMMAND_TIMEOUT (seconds). Per-query timeout for the heartbeat pool, and the SINGLE command budget the heartbeat tick wraps around its whole command sequence (BEGIN, the liveness write, the lease renewals, the still-held probe, the cancel hook's statements, COMMIT, each statement's own per-query timeout stays as the inner backstop). A tick whose statements legitimately need more than one of these in total now fails fast instead of dragging the beat out past the lease model's gap bound: raise this for a loaded or cross-region Postgres, and the lease-renewal threshold's safety floor absorbs the raised value automatically (see taskq.worker.heartbeat._lease_renewal_threshold). max_heartbeat_failures consecutive timeouts self-terminate the worker. Must be > 0: asyncpg reads 0 as 'no timeout', which turns a stalled beat into a hang.",
validator=_finite_float,
)
max_concurrency
class-attribute
instance-attribute
¶
max_concurrency: int = Field(
default=8,
ge=1,
description="TASKQ_MAX_CONCURRENCY. Upper bound on concurrent jobs. worker_pool max_size = int(max_concurrency * 1.5). When a LOOP-scope asyncpg.Connection is registered, this also sizes the worker's per-slot transaction pool (max_concurrency + 1 connections, fully warmed at boot), so a change requires a worker restart and moves the direct-connection budget, see docs/guides/deployment.md. Boot-only: no reload path re-reads settings.",
)
heartbeat_interval
class-attribute
instance-attribute
¶
heartbeat_interval: float = Field(
default=10.0,
ge=0.5,
description="TASKQ_HEARTBEAT_INTERVAL (seconds). Period between heartbeat ticks.",
validator=_finite_float,
)
lock_lease
class-attribute
instance-attribute
¶
lock_lease: float = Field(
default=60.0,
ge=1.0,
description="TASKQ_LOCK_LEASE (seconds). Time before a held lock is reclaimed by the recovery sweep. Must be >= max(heartbeat_interval, heartbeat_command_timeout) + (max_heartbeat_failures + 1) * (heartbeat_interval + heartbeat_command_timeout): the lease must outlive the worst coherent failed-beat cascade to the heartbeat's isolate decision - the last good beat's tail plus the failed cycles (at the defaults, 10 + 4 * (10 + 2) = 58).",
validator=_finite_float,
)
leader_lease
class-attribute
instance-attribute
¶
leader_lease: float = Field(
default=40.0,
ge=1.0,
description="TASKQ_LEADER_LEASE (seconds). How long the maintenance leader's lease is trusted without a renewal; another pod takes leadership once it lapses. Renewed every heartbeat_interval, and never held to less than 4 of them.",
validator=_finite_float,
)
leader_wake_jitter
class-attribute
instance-attribute
¶
leader_wake_jitter: float = Field(
default=0.25,
ge=0.0,
description="TASKQ_LEADER_WAKE_JITTER (seconds). Upper bound of the random delay a follower waits between hearing a leader's resign broadcast on the leadership wake channel and re-running its fenced elect, so a fleet of followers wakes in a spread and not in one millisecond. A literal 0 disables the damping delay (the wake then lands everyone on the row at once); the wake itself remains a hint either way - a follower that never hears it elects on the next heartbeat tick.",
validator=_finite_float,
)
max_heartbeat_failures
class-attribute
instance-attribute
¶
max_heartbeat_failures: int = Field(
default=3,
ge=1,
description="TASKQ_MAX_HEARTBEAT_FAILURES. Consecutive heartbeat failures before the worker self-terminates. Deliberate fail-fast: at the defaults (3 failures, 2 s command timeout) roughly six seconds of Postgres unavailability ends every worker at once, and the orchestrator restarts them into a recovered database while crash reclaim re-pends their leases, expect a restart herd on a Postgres failover, sized by your replica count.",
)
sweep_interval
class-attribute
instance-attribute
¶
sweep_interval: float = Field(
default=30.0,
ge=1.0,
description="TASKQ_SWEEP_INTERVAL (seconds). Period between leader sweep loop iterations, reclaim_expired_locks, sweep_expired_results, cleanup_stale_workers, and idle keyed-ref eviction. Lower values reduce recovery latency for crashed workers at the cost of more frequent PG queries.",
validator=_finite_float,
)
event_writer_batch_size
class-attribute
instance-attribute
¶
event_writer_batch_size: int = Field(
default=DEFAULT_EVENT_WRITER_BATCH_SIZE,
ge=1,
le=10000,
description="TASKQ_EVENT_WRITER_BATCH_SIZE. Rows per committed batch for every writer of job_events rows (the expired-lock, deadline and scheduled-to-pending sweeps, bulk cancel, actor deregistration). Keeps each batch transaction inside the reclaim_event_visibility_delay margin; the server-side statement_timeout remains the enforcement if a batch exceeds it. The loop drains the remainder across further batches/calls.",
)
event_writer_statement_timeout_ms
class-attribute
instance-attribute
¶
event_writer_statement_timeout_ms: float = Field(
default=DEFAULT_EVENT_WRITER_STATEMENT_TIMEOUT_MS,
ge=50.0,
description="TASKQ_EVENT_WRITER_STATEMENT_TIMEOUT_MS (milliseconds). Server-side statement_timeout applied to each event-writer batch transaction via SET LOCAL. Defaults to 7/8 of the 2 s reclaim_event_visibility_delay margin: a batch that cannot finish inside the watermark margin is aborted by the server rather than silently corrupting reclaim-event delivery.",
validator=_finite_float,
)
event_writer_reduced_batch_divisor
class-attribute
instance-attribute
¶
event_writer_reduced_batch_divisor: int = Field(
default=4,
ge=2,
le=1000,
description="TASKQ_EVENT_WRITER_REDUCED_BATCH_DIVISOR. Divisor for the reduced event-writer batch tier: once a worker's sweeps trip the batch-size breaker, batches shrink to max(1, event_writer_batch_size / this). Only a degradation ceiling , raising it makes the degraded tier closer to the normal one.",
)
sweep_breaker_failure_threshold
class-attribute
instance-attribute
¶
sweep_breaker_failure_threshold: int = Field(
default=3,
ge=1,
description="TASKQ_SWEEP_BREAKER_FAILURE_THRESHOLD. Consecutive sweep-batch cancellations (within sweep_breaker_window_secs) before the batch-size breaker latches to the reduced tier for the rest of the process lifetime. Any success between failures resets the consecutive count; a latched breaker does not unlatch.",
)
sweep_breaker_window_secs
class-attribute
instance-attribute
¶
sweep_breaker_window_secs: float = Field(
default=600.0,
ge=1.0,
description="TASKQ_SWEEP_BREAKER_WINDOW_SECS (seconds). Rolling window the sweep breaker counts consecutive failures within.",
validator=_finite_float,
)
sweep_drain_batches
class-attribute
instance-attribute
¶
sweep_drain_batches: int = Field(
default=8,
ge=1,
le=1000,
description="TASKQ_SWEEP_DRAIN_BATCHES. Maximum event-writer batches the leader's sweep loop executes per sweep per tick before leaving the remainder to the next tick. Bounded so one iteration cannot monopolise the loop; every batch commits, so a stopped drain keeps its progress.",
)
event_retention_period
class-attribute
instance-attribute
¶
event_retention_period: timedelta = Field(
default=DEFAULT_EVENT_RETENTION_PERIOD,
validator=_non_negative_timedelta,
description="TASKQ_EVENT_RETENTION_PERIOD. The age at which job_events rows are deleted regardless of parent-job status. timedelta(0) DISABLES the sweep, a deliberate inversion of the prune family's zero-means-archive-immediately: for a brand-new deletion loop the safe misconfiguration is off. The crash-reclaim outbox slice (kind='state_change' AND detail->>'reason'='lock_expired') is carved out of this window so an unread reclaim event survives it, but the carve-out is bounded: the same sweep deletes it at 100x this period (RECLAIM_OUTBOX_RETENTION_MULTIPLIER), so a short retention period bounds how far behind a lagging watch_reclaims consumer may run before it silently misses events. Negative values raise at settings load.",
)
event_retention_batch_size
class-attribute
instance-attribute
¶
event_retention_batch_size: int = Field(
default=DEFAULT_EVENT_RETENTION_BATCH_SIZE,
ge=1,
description="TASKQ_EVENT_RETENTION_BATCH_SIZE. job_events rows deleted per leader sweep tick, one committed batch.",
)
keyed_row_reclaim_period
class-attribute
instance-attribute
¶
keyed_row_reclaim_period: timedelta = Field(
default=DEFAULT_KEYED_ROW_RECLAIM_PERIOD,
validator=_non_negative_timedelta,
description="TASKQ_KEYED_ROW_RECLAIM_PERIOD. The idle age at which fleet-reclaimable keyed rows, keyed reservation_slots rows and PG-state-backed keyed rate_limit_buckets rows, marked by the keyed column, are deleted by the maintenance leader's sweep_idle_keyed_rows, one bounded committed batch per tick per table. Why a fleet sweep exists: keyed rows orphan when the worker that materialised them dies, because the in-process reclamation machinery (registry eviction + the pending-reclaim drain) dies with the process; the rows' own last_used_at stamp (refreshed by the acquire/release/upsert statements that already touch them) is the fleet-wide signal. Static buckets and redis-backend keyed rows are never deleted by this sweep at any setting. timedelta(0) DISABLES the sweep, the same zero-means-off sentinel event_retention_period uses, so a brand-new deletion loop's safe misconfiguration is off. Negative values raise at settings load.",
)
keyed_row_reclaim_batch_size
class-attribute
instance-attribute
¶
keyed_row_reclaim_batch_size: int = Field(
default=DEFAULT_KEYED_ROW_RECLAIM_BATCH_SIZE,
ge=1,
description="TASKQ_KEYED_ROW_RECLAIM_BATCH_SIZE. BUCKETS (rows, for rate_limit_buckets) the fleet reclaim sweep deletes per committed batch per tick, the constant-size bound that keeps one tick's DELETE independent of the dead-worker backlog it is recovering from, the same doctrine as event_retention_batch_size.",
)
queue_depth_interval
class-attribute
instance-attribute
¶
queue_depth_interval: float = Field(
default=15.0,
ge=1.0,
description="TASKQ_QUEUE_DEPTH_INTERVAL (seconds). Period between queue-depth metrics sampling iterations.",
validator=_finite_float,
)
reservation_slots_interval
class-attribute
instance-attribute
¶
reservation_slots_interval: float = Field(
default=15.0,
ge=1.0,
description="TASKQ_RESERVATION_SLOTS_INTERVAL (seconds). Period between reservation-slot metrics sampling iterations.",
validator=_finite_float,
)
stranded_jobs_interval
class-attribute
instance-attribute
¶
stranded_jobs_interval: float = Field(
default=60.0,
ge=1.0,
description="TASKQ_STRANDED_JOBS_INTERVAL (seconds). Period between stranded-jobs (pending jobs whose actor has no actor_config) warning checks.",
validator=_finite_float,
)
termination_grace_period
class-attribute
instance-attribute
¶
termination_grace_period: float = Field(
default=85.0,
ge=5.0,
description="TASKQ_TERMINATION_GRACE_PERIOD (seconds). Total wall-clock budget from SIGTERM to forced exit; the shutdown watchdog counts it down from the first shutdown signal. Must satisfy cancellation_grace + cleanup_grace < termination_grace - 5, and should cover the modelled worst case cancellation_grace + cleanup_grace + the ~42s bounded-close teardown tail (see WorkerSettings.worst_case_shutdown_seconds), the default does: 30 + 10 + 42 = 82s. The ~87s sibling-crash path (nine closes, including the conditional per-slot pool) exceeds the default by 2s on per-slot workers, that path is the documented caveat the model understates; raise this setting when per-slot workers need a tight crash budget. A value below the worst case still loads but logs shutdown-budget-exceeds-termination-grace at startup. Size the pod/container grace (terminationGracePeriodSeconds / stop_grace_period) from the same worst case, not from this setting alone.",
validator=_finite_float,
)
cancellation_grace_period
class-attribute
instance-attribute
¶
cancellation_grace_period: float = Field(
default=30.0,
ge=0.0,
description="TASKQ_CANCELLATION_GRACE_PERIOD (seconds). Cooperative cancel phase duration.",
validator=_finite_float,
)
cleanup_grace_period
class-attribute
instance-attribute
¶
cleanup_grace_period: float = Field(
default=10.0,
ge=0.0,
description="TASKQ_CLEANUP_GRACE_PERIOD (seconds). Force-cancel cleanup grace.",
validator=_finite_float,
)
reclaim_event_visibility_delay
class-attribute
instance-attribute
¶
reclaim_event_visibility_delay: float = Field(
default=RECLAIM_EVENT_VISIBILITY_DELAY.total_seconds(),
ge=0.0,
description="TASKQ_RECLAIM_EVENT_VISIBILITY_DELAY (seconds). Trailing-watermark margin poll_reclaim_events()/TaskQ.watch_reclaims() apply before returning a job_events row, so an out-of-commit-order sibling with a lower event_id has time to appear first (see docs/architecture.md's crash-reclaim section). Correctness assumes every job_events writer transaction commits within this margin of its INSERT; raise it if sweeps run under heavy lock contention or against very large batches, lower it if latency matters more and writes are known to be fast. A writer that exceeds the margin can cause a silently missed event - this is a real, not merely theoretical, risk under misconfiguration.",
validator=_finite_float,
)
max_retry_backoff
class-attribute
instance-attribute
¶
max_retry_backoff: timedelta = Field(
default=DEFAULT_MAX_RETRY_BACKOFF,
description="TASKQ_MAX_RETRY_BACKOFF (interval). Global ceiling on retry backoff per attempt - caps the per-actor RetryPolicy.cap so a misconfigured actor (e.g. cap=timedelta(days=365)) cannot strand jobs for an unreasonably long time. Default 24 h: conservative, aligns with a standard on-call rotation period",
)
default_start_to_close
class-attribute
instance-attribute
¶
default_start_to_close: timedelta | None = Field(
default=None,
validator=_positive_timedelta,
description="TASKQ_DEFAULT_START_TO_CLOSE (interval). Worker-side fallback per-attempt execution timeout, applied only when a job has no start_to_close of its own (neither passed at enqueue time nor declared as an @actor(start_to_close=...) default). None (the default) means unbounded - matches existing behaviour, opt-in only. Set this to give every actor on this worker a safety-net wall-clock budget per attempt, preventing a hung or infinite-looping actor from occupying a coroutine slot forever, without having to configure start_to_close on every individual actor. Precedence (highest wins): per-enqueue start_to_close > @actor(start_to_close=...) > this setting. This does not affect schedule_to_close, which is a separate, unrelated deadline for the job's *overall* retry budget across all attempts - start_to_close bounds a single attempt's wall-clock time.",
)
rate_limit_pg_fallback_enabled
class-attribute
instance-attribute
¶
rate_limit_pg_fallback_enabled: bool = Field(
default=True,
description="TASKQ_RATE_LIMIT_PG_FALLBACK_ENABLED. When False, Redis errors propagate instead of triggering PG fallback.",
)
max_keyed_reservations
class-attribute
instance-attribute
¶
max_keyed_reservations: int = Field(
default=DEFAULT_MAX_KEYED_RESERVATIONS,
ge=1,
description="TASKQ_MAX_KEYED_RESERVATIONS. Guardrail on the number of distinct keyed-reservation entries tracked in memory. When the limit is reached, new keyed reservations raise ReservationUnavailable. Tune to your workload's expected key cardinality: the guardrail is PER PROCESS, so adding replicas does not raise the effective tenant-key fleet capacity, a tenant fleet above the limit gets denials on every replica at once.",
)
max_keyed_rate_limits
class-attribute
instance-attribute
¶
max_keyed_rate_limits: int = Field(
default=10000,
ge=1,
description="TASKQ_MAX_KEYED_RATE_LIMITS. Guardrail on the number of distinct keyed-rate-limit entries tracked in memory. When the limit is reached, new keyed rate limits raise ReservationUnavailable. Independent from max_keyed_reservations, which governs keyed reservations only. Tune to your workload's expected key cardinality; like max_keyed_reservations the guardrail is per process and does not scale with the replica count.",
)
metrics_port
class-attribute
instance-attribute
¶
metrics_port: int | None = Field(
default=None,
ge=1,
le=65535,
description="TASKQ_METRICS_PORT. TCP port for the worker's standalone Prometheus scrape listener, bound on TASKQ_METRICS_HOST (falling back to TASKQ_HEALTH_HOST). Unset (the default) means no listener, so setting a port is the opt-in, the same shape as TASKQ_HEALTH_PORT. Needs the [prometheus] extra and TASKQ_OTEL_AUTOCONFIGURE=true: the `taskq worker` CLI then adds a PrometheusMetricReader to the SDK meter provider it installs, so every taskq_* / messaging_* series this worker records, the leader-sampled gauges and the dispatch/consume counters the admin process never sees, is served at http://<host>:<port>/metrics. See observability.md, Serving the metrics.",
)
health_enabled
class-attribute
instance-attribute
¶
health_enabled: bool = Field(
default=True,
description="TASKQ_HEALTH_ENABLED. Enable the health server: both the Unix socket and the optional TCP listener (health_port). False disables both, and a health_port set alongside it is not honoured, so probes against that port fail.",
)
health_socket_path
class-attribute
instance-attribute
¶
health_socket_path: str = Field(
default="/tmp/taskq_health.sock",
description="TASKQ_HEALTH_SOCKET_PATH. Unix socket path for the health server, serving /live, /ready, /metrics and the opt-in /tasks endpoint. Give each co-located process a unique path: a path whose live peer holds it fails to bind, and that collision is a WARNING with the boot continuing (a live peer owning the path is a rolling-restart shape, not a failure); see health_port for the TCP arm's different contract.",
)
health_pg_ping_timeout
class-attribute
instance-attribute
¶
health_pg_ping_timeout: float = Field(
default=0.2,
ge=0.0,
description="TASKQ_HEALTH_PG_PING_TIMEOUT. Seconds to wait for dispatcher_pool.acquire() in the readiness PG ping. Default 200ms .",
validator=_finite_float,
)
health_tasks_enabled
class-attribute
instance-attribute
¶
health_tasks_enabled: bool = Field(
default=False,
description="TASKQ_HEALTH_TASKS_ENABLED. Expose the privileged /tasks asyncio stack-dump endpoint on the Unix health socket. Off by default: the dump reveals code structure, file paths, and task names (never locals or payload values). Enabling it also tightens the socket to owner-only (no group/other access). Unix socket only, never mounted on the admin UI surface.",
)
health_host
class-attribute
instance-attribute
¶
health_host: str = Field(
default="0.0.0.0",
description="TASKQ_HEALTH_HOST. Bind address for the optional TCP health listener and the optional Prometheus scrape listener. Only used when health_port or metrics_port is set. Defaults to all interfaces because Azure Container Apps and Kubernetes probe and scrape the replica over the pod network; narrow it to 127.0.0.1 when a local sidecar is the only prober.",
)
metrics_host
class-attribute
instance-attribute
¶
metrics_host: str | None = Field(
default=None,
description="TASKQ_METRICS_HOST. Bind address for the optional Prometheus scrape listener, overriding TASKQ_HEALTH_HOST for that listener alone, so the scrape and the probes can sit on different interfaces (a loopback sidecar scraper next to a pod-network probe is the shape that needs this). Unset falls back to TASKQ_HEALTH_HOST. Only used when metrics_port is set.",
)
health_port
class-attribute
instance-attribute
¶
health_port: int | None = Field(
default=None,
ge=0,
le=65535,
description="TASKQ_HEALTH_PORT. TCP port for the HTTP health listener serving /live and /ready. Unset (the default) means no TCP listener at all, so setting a port is the opt-in. Required on Azure Container Apps, whose probes support only httpGet/tcpSocket and cannot reach a Unix socket (there is no exec probe type). The Unix socket keeps working either way. If the port cannot be bound the worker fails to start with HealthTcpBindError rather than run with probes silently dead: the orchestrator routes this replica's probes here, and a tcpSocket probe against a port some other process holds would pass while probing the wrong process). The unix socket's collision is deliberately softer: a live peer owns the path, so the boot warns and continues. 0 binds an ephemeral port (tests only).",
)
health_request_timeout
class-attribute
instance-attribute
¶
health_request_timeout: float = Field(
default=2.0,
gt=0.0,
description="TASKQ_HEALTH_REQUEST_TIMEOUT. Seconds allowed for a probe to send its whole request line and headers before the connection is dropped unanswered. Bounds a drip-feed client that would otherwise hold a connection open forever by staying just inside a per-line timeout. Keep it at or below the shortest probe timeoutSeconds you configure.",
validator=_finite_float,
)
health_max_header_bytes
class-attribute
instance-attribute
¶
health_max_header_bytes: int = Field(
default=16 * 1024,
gt=0,
description="TASKQ_HEALTH_MAX_HEADER_BYTES. Cap on a probe request's accumulated request line plus headers. Pairs with health_request_timeout to bound a peer that sends many small lines fast enough to stay inside the deadline. 16 KiB is far above any real probe request, which carries a path and a handful of headers.",
)
health_readiness_check_timeout
class-attribute
instance-attribute
¶
health_readiness_check_timeout: float = Field(
default=5.0,
gt=0.0,
description="TASKQ_HEALTH_READINESS_CHECK_TIMEOUT. Seconds each check registered via taskq.worker.health.register_readiness_check may take before it counts as a readiness failure. Defaults to 5s, matching the Azure Container Apps default readiness probe timeoutSeconds, so a wedged check fails the probe rather than outliving it.",
validator=_finite_float,
)
watchdog_enabled
class-attribute
instance-attribute
¶
watchdog_enabled: bool = Field(
default=True,
description="TASKQ_WATCHDOG_ENABLED. Master switch for the in-worker watchdog detectors (shutdown deadline, stale loop ticks, sibling contract, event-loop lag). A detector trip dumps the asyncio task stacks and force-exits non-zero so the supervisor restarts the worker instead of leaving it wedged.",
)
watchdog_loop_lag_budget
class-attribute
instance-attribute
¶
watchdog_loop_lag_budget: float = Field(
default=30.0,
gt=0.0,
description="TASKQ_WATCHDOG_LOOP_LAG_BUDGET (seconds). How long the event loop may go without scheduling before the lag watchdog trips. Deliberately far beyond any legitimate pause (GC, a slow tick) because the trip is terminal. Tier 2 of the lag detector; see watchdog_loop_lag_warn_budget for the non-terminal tier 1.",
validator=_finite_float,
)
watchdog_loop_lag_warn_budget
class-attribute
instance-attribute
¶
watchdog_loop_lag_warn_budget: float = Field(
default=5.0,
gt=0.0,
description="TASKQ_WATCHDOG_LOOP_LAG_WARN_BUDGET (seconds). Non-terminal tier-1 event-loop lag threshold: faulthandler thread dump + metric + deferred asyncio task-stack dump. Never exits; the terminal tier is watchdog_loop_lag_budget.",
validator=_finite_float,
)
watchdog_loop_lag_startup_grace
class-attribute
instance-attribute
¶
watchdog_loop_lag_startup_grace: float = Field(
default=30.0,
ge=0.0,
description="TASKQ_WATCHDOG_LOOP_LAG_STARTUP_GRACE (seconds). Grace before the lag watchdog arms, covering import-heavy startup, DI bootstrap, and first dispatch. Anchored to thread start; the lag detector also arms early once the first loop liveness tick lands.",
validator=_finite_float,
)
watchdog_tick_grace_factor
class-attribute
instance-attribute
¶
watchdog_tick_grace_factor: float = Field(
default=5.0,
gt=0.0,
description="TASKQ_WATCHDOG_TICK_GRACE_FACTOR. Multiplier on a loop's iteration period before its liveness tick is declared stale (floor 10s). Generous on purpose: a terminal detector must never fire on a merely loaded host.",
validator=_finite_float,
)
watchdog_dump_interval
class-attribute
instance-attribute
¶
watchdog_dump_interval: float = Field(
default=5.0,
gt=0.0,
description="TASKQ_WATCHDOG_DUMP_INTERVAL (seconds). Interval between straggler logs (names + await sites of still-alive siblings) while a shutdown is in progress.",
validator=_finite_float,
)
watchdog_dump_after_fraction
class-attribute
instance-attribute
¶
watchdog_dump_after_fraction: float = Field(
default=0.5,
gt=0.0,
lt=1.0,
description="TASKQ_WATCHDOG_DUMP_AFTER_FRACTION. Fraction of the shutdown deadline that must be consumed before straggler dumps begin (0.5 = only in the back half of the budget). A drain inside its front half is within expectations and stays quiet; one countdown-start record is always logged so the window is never blind. Must be < 1: at 1.0 the deadline trip would always fire first, silently disabling the dumps.",
validator=_finite_float,
)
watchdog_stale_floor
class-attribute
instance-attribute
¶
watchdog_stale_floor: float = Field(
default=10.0,
gt=0.0,
description="TASKQ_WATCHDOG_STALE_FLOOR (seconds). Minimum staleness budget for any loop (period x grace_factor, floored at this value). Guards tiny intervals against false trips under host starvation, a terminal detector must never fire on load.",
validator=_finite_float,
)
watchdog_check_interval
class-attribute
instance-attribute
¶
watchdog_check_interval: float = Field(
default=1.0,
gt=0.0,
description="TASKQ_WATCHDOG_CHECK_INTERVAL (seconds). Poll cadence for the stale-tick sweep and the loop-lag watchdog thread.",
validator=_finite_float,
)
poll_interval
class-attribute
instance-attribute
¶
poll_interval: float = Field(
default=1.0,
gt=0,
description="TASKQ_POLL_INTERVAL (seconds). Producer loop fallback polling cadence when the NOTIFY listener is unavailable.",
validator=_finite_float,
)
notify_health_check_interval
class-attribute
instance-attribute
¶
notify_health_check_interval: float = Field(
default=5.0,
gt=0,
description="TASKQ_NOTIFY_HEALTH_CHECK_INTERVAL (seconds). How often _health_check_loop issues SELECT 1 on notify_conn. Detection latency before reconnect is at most this interval.",
validator=_finite_float,
)
notify_reconnect_backoff_initial
class-attribute
instance-attribute
¶
notify_reconnect_backoff_initial: float = Field(
default=1.0,
gt=0,
description="TASKQ_NOTIFY_RECONNECT_BACKOFF_INITIAL (seconds). Initial exponential backoff delay before the first reconnect retry. Cap is 30 s (factor 2 per attempt). Backoff sequence: 1, 2, 4, 8, 16, 30.",
validator=_finite_float,
)
notify_listener_setup_timeout
class-attribute
instance-attribute
¶
notify_listener_setup_timeout: float = Field(
default=10.0,
validator=_positive_finite_float,
description="TASKQ_NOTIFY_LISTENER_SETUP_TIMEOUT (seconds). Bounds each ``LISTEN`` execute and ``add_listener`` call during NOTIFY listener setup and reconnect - a half-open PG connection that accepts TCP (or completes the reconnect factory handshake) but stalls on the LISTEN execute or registration would otherwise wedge the notify loop forever. On timeout the connection is closed (bounded) and the reconnect retry loop is entered (or the initial setup raises).",
)
notify_enabled
class-attribute
instance-attribute
¶
notify_enabled: bool = Field(
default=True,
description="TASKQ_NOTIFY_ENABLED. When True, the worker uses LISTEN/NOTIFY for near-zero-latency dispatch wakeups with poll interval as fallback. When False, the worker uses poll-only dispatch.",
)
notify_poll_interval
class-attribute
instance-attribute
¶
notify_poll_interval: float = Field(
default=5.0,
ge=0.5,
description="TASKQ_NOTIFY_POLL_INTERVAL (seconds). Fallback poll cadence when NOTIFY is enabled (rarely reached - NOTIFY handles the common case). Use poll_interval when NOTIFY is disabled.",
validator=_finite_float,
)
reload_factory_timeout
class-attribute
instance-attribute
¶
reload_factory_timeout: float = Field(
default=30.0,
gt=0,
description="TASKQ_RELOAD_FACTORY_TIMEOUT (seconds). Bounds each individual factory call - during a credential hot-reload (reload_credentials), at bootstrap when the worker opens its per-slot transaction pool (a fully-warmed open means one connection - and on a managed-identity deployment one credential fetch - per consumer slot), on the notify listener's health-check reconnect (reconnect_notify_conn), and at the DI scope bootstraps' first use of user-registered factories (the 'database pools, HTTP clients' provider class, resolved through ScopeContainer.get_or_create before any watchdog is armed). A hung token endpoint - or a black-holed DI factory - is marked failed for that resource - or logged as a reconnect attempt and retried - instead of wedging the reload coordinator, worker boot, or the reconnect loop.",
validator=_finite_float,
)
queues
class-attribute
instance-attribute
¶
queues: list[str] = Field(
default_factory=lambda: ["default"],
validator=_queue_names_validator,
description="TASKQ_QUEUES. Comma-separated list of queue names this worker will consume from.",
)
worker_label
class-attribute
instance-attribute
¶
worker_label: str | None = Field(
default=None,
validator=_worker_label_validator,
description="TASKQ_WORKER_LABEL. Human-readable label stored in the workers table for correlation with workgroup supervisors and external monitoring. When omitted the column is NULL; hostname and pid columns provide identification.",
)
workgroup_instance
class-attribute
instance-attribute
¶
workgroup_instance: str | None = Field(
default=None,
validator=_workgroup_instance_validator,
description="TASKQ_WORKGROUP_INSTANCE. UUIDv7 identifying the workgroup orchestrator that launched this worker. Used for cross-process correlation.",
)
pool_max_inactive_lifetime
class-attribute
instance-attribute
¶
pool_max_inactive_lifetime: float = Field(
default=300.0,
ge=0.0,
description="TASKQ_POOL_MAX_INACTIVE_LIFETIME (seconds). asyncpg max_inactive_connection_lifetime - closes connections idle longer than this threshold. Set to 3600.0 to match a typical SQLAlchemy pool_recycle=3600 setting when running alongside an SQLAlchemy-based service. Applied to dispatcher_pool, heartbeat_pool, worker_pool, and the conditional per-slot transaction pool.",
validator=_finite_float,
)
otel_enabled
class-attribute
instance-attribute
¶
otel_enabled: bool = Field(
default=True,
description="TASKQ_OTEL_ENABLED. When False, the library suppresses all span and metric creation but operations still succeed .",
)
otel_autoconfigure
class-attribute
instance-attribute
¶
otel_autoconfigure: bool = Field(
default=True,
description="TASKQ_OTEL_AUTOCONFIGURE. When True (the default), the `taskq worker` CLI installs SDK tracer and meter providers from the standard OTel environment variables (OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_TRACES_EXPORTER, OTEL_METRICS_EXPORTER, OTEL_LOGS_EXPORTER) and from TASKQ_METRICS_PORT, through the same configurator opentelemetry-instrument uses, whenever the [otel] extra is installed and no provider is set yet. Set False when the embedding application or a vendor distro configures the SDK itself and the worker must not touch the global providers. Honours OTEL_SDK_DISABLED either way.",
)
exception_message_max_chars
class-attribute
instance-attribute
¶
exception_message_max_chars: int = Field(
default=2000,
ge=100,
description="TASKQ_EXCEPTION_MESSAGE_MAX_CHARS. Bound on exception message text on spans and logs, after scrubbing. Matches the admin UI's traceback bound so there is one number for how much error text is kept, not two. Truncation appends the dropped character count, so an operator can see text was cut and raise this. Raise it when an actor formats large context into its messages; the stack trace is a separate field and is not bounded by this.",
)
exception_redaction_enabled
class-attribute
instance-attribute
¶
exception_redaction_enabled: bool = Field(
default=True,
description="TASKQ_EXCEPTION_REDACTION_ENABLED. When True (the default), Postgres 'DETAIL:' lines are dropped from exception text before it reaches spans and logs, because they quote caller-supplied row values (idempotency_key, identity_key, fairness_key routinely hold tenant or subject identifiers). Set to False for advanced debugging to ship the raw text, including those row values, to every configured telemetry backend; the worker logs a startup WARNING while it is off. URI credential masking (scheme://user:***@host) is NOT affected by this setting and is always applied - no debugging case justifies sending a password to a telemetry vendor.",
)
worker_group
class-attribute
instance-attribute
¶
worker_group: str = Field(
default="default",
description="TASKQ_WORKER_GROUP. Consumer group name emitted as messaging.consumer.group.name on CONSUMER spans .",
)
log_format
class-attribute
instance-attribute
¶
log_format: str = Field(
default="json",
validator=_log_format_validator,
description="TASKQ_LOG_FORMAT. json|console. Selects JSONRenderer or ConsoleRenderer in setup_logging.",
)
log_level
class-attribute
instance-attribute
¶
log_level: str = Field(
default="INFO",
validator=_log_level_validator,
description="TASKQ_LOG_LEVEL. Root logger level.",
)
log_events_level
class-attribute
instance-attribute
¶
log_events_level: str = Field(
default="info",
validator=_log_events_level_validator,
description="TASKQ_LOG_EVENTS_LEVEL. The operational-event-stream verbosity knob: info|warning|off|debug, case-insensitive. The job_events ledger is the audit trail - every state change is committed there in the same transaction as the state itself - so this knob suppresses the STREAMING DUPLICATE (the per-job JSON log lines) only, applied by a structlog processor before the serialization cost, never at the call sites. info (the default) is today's stream byte-identical. warning emits only the failure/anomaly stream: the failed ticks, the isolates, the watchdog trips, the backpressure refusals (anomaly events survive even when logged at INFO - the classification is by event name, not stdlib level), the happy path's per-job lines gone. off is for deployments consuming OTel spans or tailing job_events: the state-change lines gone entirely, the INFO-method lines of classified anomalies gone too, the WARNING-and-above anomalies STILL emitted - off never blinds the operator to failures, and unclassified events pass at every level there as well (fail open at every level, not just at warning). debug ADDS the per-tick internals (the loop-lag traces, the poll-cadence details). Unclassified events fail open: an event not in the classification table passes at every level, so a future event can never be silently suppressed. The full semantics table is obs/_structlog.py's docstring and configuration.md.",
)
prune_schedule_utc
class-attribute
instance-attribute
¶
prune_schedule_utc: str = Field(
default="03:00",
validator=_hh_mm_validator,
description="TASKQ_PRUNE_SCHEDULE_UTC. HH:MM (UTC) for the daily prune run. Ignored when prune_cron_expr is set.",
)
prune_cron_expr
class-attribute
instance-attribute
¶
prune_cron_expr: str | None = Field(
default=None,
validator=_cron_expr_validator,
description="TASKQ_PRUNE_CRON_EXPR. Full 5-field cron expression. When set, takes precedence over prune_schedule_utc.",
)
prune_batch_size
class-attribute
instance-attribute
¶
prune_batch_size: int = Field(
default=DEFAULT_PRUNE_BATCH_SIZE,
ge=1,
description="TASKQ_PRUNE_BATCH_SIZE. Rows to delete per batch.",
)
prune_retention_period
class-attribute
instance-attribute
¶
prune_retention_period: timedelta = Field(
default=DEFAULT_PRUNE_RETENTION,
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_PERIOD. Reserved as the global fallback retention for terminal statuses without a per-status knob. Currently INERT: the prune sweep reads the four per-status fields below, and together they cover every terminal status (succeeded, failed, cancelled, crashed, abandoned), so no status ever falls back here, setting this value changes nothing today. Size the per-status fields instead; see TASKQ_PRUNE_RETENTION_SUCCEEDED. Negative values raise ConstraintViolationError at settings load.",
)
prune_retention_succeeded
class-attribute
instance-attribute
¶
prune_retention_succeeded: timedelta = Field(
default=timedelta(days=30),
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_SUCCEEDED. How long a succeeded job stays in the hot jobs table before the daily prune sweep moves it to jobs_archive (where archive_retention_period then governs hard-deletion, 365 d by default, so history is not lost at prune time). Sizing is a hot-table trade: succeeded rows are usually the bulk of terminal volume, and every day of retention keeps roughly a day's terminal throughput in the hot table the admin /jobs list reads (at 100k jobs/day the default 30 d holds ~3M rows, see the storage-planning note in configuration.md). Lower it for high-volume actors whose successes nobody audits (the per-actor metadata retention_days override shortens it further for one actor); raise it when operators routinely inspect successful runs older than a month without querying the archive. timedelta(0) archives succeeded jobs at the next sweep, the prune family's zero-means-now polarity, deliberately opposite to the sweep family's zero-means-off (see the 0 convention in configuration.md). Negative values raise ConstraintViolationError at settings load.",
)
prune_retention_failed
class-attribute
instance-attribute
¶
prune_retention_failed: timedelta = Field(
default=timedelta(days=90),
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_FAILED. How long a failed job stays in the hot jobs table before the daily prune sweep moves it to jobs_archive. Failed rows are the first incident-audit trail, they carry error_class, error_message and the attempt history, so the default keeps them hot three times longer than succeeded rows (90 d vs 30 d). Size to how far back your on-call reads failures in the fast surfaces (admin /jobs) before the archive is acceptable; lower it only if failure volume makes the hot table's size the bigger incident risk. timedelta(0) archives failed jobs at the next sweep (zero-means-now, see the 0 convention in configuration.md). Negative values raise ConstraintViolationError at settings load.",
)
prune_retention_cancelled
class-attribute
instance-attribute
¶
prune_retention_cancelled: timedelta = Field(
default=timedelta(days=30),
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_CANCELLED. How long a cancelled job stays in the hot jobs table before the daily prune sweep moves it to jobs_archive. Cancelled rows are operator- or deadline-initiated and rarely revisited after the fact, so the default follows succeeded (30 d); raise it if cancellations are part of your audit story, lower it toward 0 for bulk-cancel workloads whose rows are pure churn. timedelta(0) archives cancelled jobs at the next sweep (zero-means-now, see the 0 convention in configuration.md). Negative values raise ConstraintViolationError at settings load.",
)
prune_retention_abandoned
class-attribute
instance-attribute
¶
prune_retention_abandoned: timedelta = Field(
default=timedelta(days=90),
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_ABANDONED. How long an abandoned job stays in the hot jobs table before the daily prune sweep moves it to jobs_archive. Also used for crashed jobs (no separate prune_retention_crashed field): both statuses mean the job outlived its execution budget or its worker, and both are the rows you reach for when reconstructing a fleet-level incident, so they share the longer 90 d default with failed. Same sizing trade and zero-means-now polarity as the sibling fields, see TASKQ_PRUNE_RETENTION_SUCCEEDED. Negative values raise ConstraintViolationError at settings load.",
)
archive_retention_period
class-attribute
instance-attribute
¶
archive_retention_period: timedelta = Field(
default=timedelta(days=365),
validator=_non_negative_timedelta,
description="TASKQ_ARCHIVE_RETENTION_PERIOD. How long archived jobs are retained in jobs_archive before hard-deletion. Default 1 year. timedelta(0) hard-deletes an archived row at the next archive-expiry sweep, the row's expire_at is stamped archive-time plus this period, so a zero period expires it on arrival; the prune family's zero-means-now polarity, not the deletion-sweep family's zero-means-off (see the 0 convention in configuration.md). Negative values raise ConstraintViolationError.",
)
archive_expiry_schedule_utc
class-attribute
instance-attribute
¶
archive_expiry_schedule_utc: str = Field(
default="04:00",
validator=_hh_mm_validator,
description="TASKQ_ARCHIVE_EXPIRY_SCHEDULE_UTC. HH:MM (UTC) for the daily archive expiry sweep. Default 04:00, 1 hour after the prune sweep.",
)
archive_expiry_cron_expr
class-attribute
instance-attribute
¶
archive_expiry_cron_expr: str | None = Field(
default=None,
validator=_cron_expr_validator,
description="TASKQ_ARCHIVE_EXPIRY_CRON_EXPR. Full 5-field cron expression. When set, takes precedence over archive_expiry_schedule_utc.",
)
force_update_actor_config
class-attribute
instance-attribute
¶
force_update_actor_config: bool = Field(
default=False,
description="When True, sync_actor_config silently overwrites a stored actor_config row whose metadata differs from the registered value. When False (the default), metadata drift raises ActorConfigDriftList and the worker refuses to start. The queue assignment and the capacity fields (max_concurrent, max_pending, result_ttl) are unaffected by this flag: once a row exists, the stored value is always authoritative and is never overwritten by the registered @actor(...) literal, regardless of force. Move an actor between queues with `taskq actor-config move-queue`; tune a stored capacity value with `taskq actor-config set`. Env var: TASKQ_FORCE_UPDATE_ACTOR_CONFIG.",
)
progress_coalesce_interval
class-attribute
instance-attribute
¶
progress_coalesce_interval: float = Field(
default=0.5,
ge=0.1,
description="TASKQ_PROGRESS_COALESCE_INTERVAL (seconds). How long the periodic flush loop waits between writing coalesced progress state to Postgres. Redis publishes are not throttled by this setting - each ctx.progress() call publishes immediately (fire-and-forget). Lower values increase PG write frequency; minimum 0.1 s.",
validator=_finite_float,
)
progress_data_max_bytes
class-attribute
instance-attribute
¶
progress_data_max_bytes: int = Field(
default=16384,
ge=1024,
le=1048576,
description="TASKQ_PROGRESS_DATA_MAX_BYTES. Maximum serialised byte length of the ``data`` dict in a single progress call. Payloads exceeding this limit raise ProgressTooLarge. Snapshot-style publishers (full-document snapshots as ``data``) must size the cap at or above their payload bound: once one snapshot exceeds it, every later snapshot raises too. Range: 1 KiB - 1 MiB; default 16 KiB.",
)
progress_publish_global
class-attribute
instance-attribute
¶
progress_publish_global: bool = Field(
default=True,
description="TASKQ_PROGRESS_PUBLISH_GLOBAL. When True (the default), progress events are additionally published to a schema-wide global fanout channel (in addition to the per-job channel). When False, events are only published to the per-job Redis channel. Does not affect Postgres flushing.",
)
result_max_bytes
class-attribute
instance-attribute
¶
result_max_bytes: int = Field(
default=MAX_RESULT_BYTES,
ge=1024,
le=1048576,
description="TASKQ_RESULT_MAX_BYTES. Maximum serialised byte length of a job's terminal result dict. A larger result raises ResultTooLarge, which is non-retryable, the actor already ran, so a re-run returns the same oversized value. Range: 1 KiB - 1 MiB (the same ceiling as progress_data_max_bytes, so the durable payload can be configured as large as the transient one); default 64 KiB. Raise it only with the row size in mind: unlike progress data, the result is stored for the job's result_ttl.",
)
cron_catch_up_window
class-attribute
instance-attribute
¶
cron_catch_up_window: timedelta = Field(
default=timedelta(hours=1),
validator=_non_negative_timedelta,
description="TASKQ_CRON_CATCH_UP_WINDOW. Missed firings within this window are caught up sequentially; older misses are skipped.",
)
cron_auto_disable_threshold
class-attribute
instance-attribute
¶
cron_auto_disable_threshold: int = Field(
default=3,
ge=1,
description="TASKQ_CRON_AUTO_DISABLE_THRESHOLD. Consecutive failures before a schedule is auto-disabled.",
)
cron_tick_limit
class-attribute
instance-attribute
¶
cron_tick_limit: int = Field(
default=DEFAULT_EVENT_WRITER_BATCH_SIZE,
ge=1,
le=10000,
description="TASKQ_CRON_TICK_LIMIT. Maximum schedules one cron tick selects, plans and fires. A catch-up burst larger than this drains across successive one-second ticks instead of one oversized transaction; the remainder stays due and untouched until its tick. Only the leader plans ticks, so this guardrail does not scale with the replica count: raise it when one tick's share of schedules genuinely exceeds it, or spread schedules off the second boundary.",
)
cron_payload_factory_timeout
class-attribute
instance-attribute
¶
cron_payload_factory_timeout: float = Field(
default=5.0,
validator=_positive_finite_float,
description="TASKQ_CRON_PAYLOAD_FACTORY_TIMEOUT. Per-call deadline for a cron schedule's payload factory (both the off-loop call and the coroutine a factory returns). Default 5.0s. The tick clamps it to stay strictly inside what is left of the leader's whole-tick deadline (dispatcher_command_timeout), so a factory that runs and outlives its granted deadline takes the named per-schedule failure this setting exists to record, never the whole-tick cancellation; a value at or above that deadline is therefore an upper bound, not the effective one. A factory is called only when the leftover can fund at least min(this value, a quarter of the tick's funded budget), a smaller leftover funds no call and the schedule is deferred (next_fire_at advances one tick cadence) rather than struck: the factory never ran, so there is no evidence against the schedule, only the schedule whose factory consumed the budget is failing. That also makes this setting the fairness lever when one slow-but-successful factory monopolizes the tick budget every tick (its peers defer indefinitely, watch taskq.cron.budget_deferrals and the cron-fire-budget-deferred log): set it BELOW the monopolizing factory's real duration and that factory takes the strike-and-auto-disable path instead, freeing its peers; raising dispatcher_command_timeout widens the funded budget the same resolution needs.",
)
idle_settle_window
class-attribute
instance-attribute
¶
idle_settle_window: float = Field(
default=2.0,
ge=0.0,
description="TASKQ_IDLE_SETTLE_WINDOW (seconds). Time the drain monitor waits after queues appear empty before declaring drained. Only used when --until-idle is active.",
validator=_finite_float,
)
idle_poll_interval
class-attribute
instance-attribute
¶
idle_poll_interval: float = Field(
default=1.0,
ge=0.1,
description="TASKQ_IDLE_POLL_INTERVAL (seconds). How often the drain monitor checks queue depth. Only used when --until-idle is active.",
validator=_finite_float,
)
idle_max_runtime
class-attribute
instance-attribute
¶
idle_max_runtime: float | None = Field(
default=None,
gt=0,
description="TASKQ_IDLE_MAX_RUNTIME (seconds). Maximum wall-clock time for until-idle mode. When exceeded, exit code 4. None = no limit. Only used when --until-idle is active.",
)
resolved_leader_lease
property
¶
The maintenance lease actually honoured, in seconds.
The lease is renewed once per heartbeat interval, so one shorter
than four of them would demote a leader whose renewal is merely a
tick behind, leadership would churn on every slow tick. Raising
heartbeat_interval alone must not produce that, and must not
refuse to boot either: a fleet that can no longer be told what to
do is worse than one running a longer lease than it asked for. The
configured value is therefore a floor that the same four-beat slack
the jobs' lock leases carry can raise, never a ceiling.
resolved_pg_dsn_direct
property
¶
Direct DSN guaranteed non-None after :meth:post_load.
Why a property: pg_dsn_direct: PostgresDsn | None carries the
environment-shape that distinguishes "user did not set
TASKQ_PG_DSN_DIRECT" (None, fallback to pg_dsn) from
"user set it explicitly". Once :meth:post_load has applied the
fallback, the field is always non-None - but pyright cannot
prove that across method boundaries. This property re-asserts the
invariant at every call site, eliminating the need for assert
or cast at call sites that read the DSN.
Raises :class:RuntimeError if accessed before :meth:post_load
ran (signals a programming error: WorkerSettings() constructor
must always go through :meth:load / :meth:load_from_dict).
resolved_pg_dsn_pooled
property
¶
Pooled DSN guaranteed non-None after :meth:post_load.
See :attr:resolved_pg_dsn_direct for the rationale.
worker_pool_size
property
¶
Derived pool size for worker_pool: int(max_concurrency * 1.5).
worst_case_shutdown_seconds
property
¶
Modelled worst-case wall clock from SIGTERM to process exit.
The shutdown phase graces plus the bounded-close tail that unwinds
after them (see :func:taskq._close.worst_case_teardown_tail).
This is deliberately NOT enforced by post_load. The cross-field
validator there rejects a config outright, and a budget below this
dead-backend worst case is a legitimate operator choice (a tight
dev deployment that would rather be SIGKILLed mid-unwind than wait
out a hung close). The shipped default covers the model, raising
the validator to reject sub-worst-case budgets would take that
choice away. The number is surfaced as a startup warning instead,
and docs/guides/deployment.md documents the pod-grace formula.
release_exit_tail_seconds
property
¶
Seconds past the termination deadline the process can still be alive: the tail every release hold pads its remaining share with.
The deadline trip is not instantaneous: the watchdog checks the
deadline once per watchdog_dump_interval (the check is clipped
to the deadline, so real lag is loop jitter and this term is
margin), and trip() then renders task stacks and joins the
bounded metrics flush before os._exit. The stack render and
critical log write have no bound of their own: the
RELEASE_EXIT_TAIL_SLACK_SECS heuristic covers them, and the
residual (a full stderr pipe under a slow docker logging driver)
is documented in docs/guides/workers.md. Kept as a settings
property because the lease arithmetic that guards the disown path
(see release_disown_lease_floor) needs the same number the
worker-layer hold pads with: one source, no drift.
release_park_lease_cap
property
¶
The lease-derived ceiling on the release-until-exited park.
The consumer's shutdown arm parks a still-running sync actor (or
transactional unwind) so a provable exit can earn the immediate
pending release, but the heartbeat, the row's only lease
renewer, stops when the orchestrator sets shutdown_event
(roughly cancellation_grace + cleanup_grace in). A park that
ran to the termination budget on a lease shorter than that budget
would let the row's lease expire mid-park, and the leader's
reclaim sweep (no carve-out for SHUTDOWN-origin rows:
cancel_phase stays 0) would re-pend a row whose actor thread
is still executing: the double-run, reachable from a single
infra-failed RELEASING write. The consumer therefore caps the
park at::
lock_lease - heartbeat_interval - TERMINAL_WRITE_BUDGET_SECS
which is exactly the bound that makes the parked consumer's
release write land before the earliest reclaim (last heartbeat,
up to one interval stale, plus the lease) for ANY config: the
write starts by cancel + cap and spends at most the terminal
write's budget, and cancel + (lock_lease - heartbeat -
write_budget) + write_budget <= (cancel + cleanup_grace -
heartbeat) + lock_lease holds identically. Safety by
construction, not by configuration, which is why there is no
cross-field rejection in post_load for this shape: the cap
makes every loadable config safe, and refusing e.g.
120/30/10/60 outright would reject a config the cap already
protects. The cost of a binding cap is latency, not safety: a
capped park releases the row earlier with a longer hold, and the
release-park-lease-capped startup warning names the configs
where that trade is being made.
release_park_budget_bound
property
¶
The budget-derived park bound at the graces: termination -
cancellation - cleanup - TERMINAL_WRITE_BUDGET_SECS.
The park's other ceiling (the remaining termination budget minus
the release write's own budget, evaluated at the cancel). The
lease cap binds first whenever
release_park_lease_cap < release_park_budget_bound: i.e.
whenever lock_lease < termination - cancellation - cleanup +
heartbeat, and that is the inequality the
release-park-lease-capped startup warning surfaces with its
arithmetic. At the shipped defaults (60 vs 55) the budget bound
binds and the park runs its full remaining budget.
release_disown_lease_floor
property
¶
The lock_lease the disown path needs (warning floor; the
shipped default is 63 vs lock_lease 60 and marginally fails).
When BOTH release writers fail their writes (the RELEASING phase's
and the consumer's: the consumer's exhaustion disowns the row),
the row stays running behind a lease the heartbeat has already
stopped renewing, and the leader's reclaim sweep becomes the only
exit: at the earliest last heartbeat + lock_lease. For the
reclaim to stay behind the process's true exit (the deadline trip
plus the exit tail, which is where an outlived actor thread dies),
the lease must cover:
lock_lease >= termination - cancellation - cleanup + heartbeat
+ release_exit_tail_seconds
At the shipped defaults that is 63 against lock_lease 60: a
residue of ~3s that requires the double write failure AND a sweep
tick landing inside it. Surfaced as a startup warning, not a hard
fail: the shipped default would not load otherwise, and whether to
spend 3 more seconds of lease on a double-failure residue is a
maintainer/operator call (see
_emit_startup_warnings in taskq.worker._bootstrap).
shutdown_budget_is_sufficient
property
¶
Whether termination_grace_period covers the modelled worst case.
post_load ¶
Apply DSN fallback and validate cross-field invariants after loading.
Runs automatically on every load path (load(),
load_from_dict(), reload(), and nested config loading),
including under validate=False - consistent with the per-field
validator hooks (transformation is part of loading, not
validation). No WorkerSettings.load / load_from_dict
override is needed; the base DotEnvConfig._load_fields invokes
this hook itself.
Returns list[ValidationError] so failures integrate with
dotenvmodel's uniform error hierarchy: a single returned error is
raised unchanged (its exact type preserved), several aggregate
into MultipleValidationErrors. Catch DotEnvModelError (the
common base) to cover both single and aggregate cases -
MultipleValidationErrors is a DotEnvModelError but not a
ValidationError, so except ValidationError alone misses the
multi-invariant case. ValidationError suffices only when at
most one invariant can fire (e.g. a single field constraint).
Source code in src/taskq/settings.py
2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443 2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464 2465 2466 2467 2468 2469 2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489 2490 2491 2492 2493 2494 2495 2496 2497 2498 2499 2500 2501 2502 2503 2504 2505 2506 2507 2508 2509 2510 2511 2512 2513 2514 2515 2516 2517 2518 2519 2520 2521 2522 2523 2524 2525 2526 2527 2528 2529 2530 2531 2532 2533 2534 2535 2536 2537 2538 2539 2540 2541 2542 2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581 2582 2583 2584 2585 2586 2587 2588 2589 2590 2591 2592 2593 2594 2595 2596 2597 2598 2599 2600 | |
new_uuid ¶
validate_actor_payload ¶
validate_actor_payload(
payload_type: type[BaseModel],
raw_payload: dict[str, object] | BaseModel,
actor: str | None = None,
*,
payload_schema_ver: str | None = None,
) -> BaseModel
Validate a raw payload dict (or existing BaseModel) against the actor's payload model.
Wraps pydantic.ValidationError as
:class:~taskq.exceptions.PayloadValidationError (non-retryable) so
the retry classifier fails the job immediately instead of retrying
a deterministic validation failure.
Error details are sanitized via include_url=False,
include_input=False to prevent attacker-controlled field values
from being persisted to the jobs row or surfaced in the web admin.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload_type
|
type[BaseModel]
|
The actor's Pydantic payload model class. |
required |
raw_payload
|
dict[str, object] | BaseModel
|
The raw |
required |
actor
|
str | None
|
The actor name, for error context. |
None
|
payload_schema_ver
|
str | None
|
The schema version attached to the payload being
validated, for a dispatch-time failure this is the job row's
stored |
None
|
Returns:
| Type | Description |
|---|---|
BaseModel
|
The validated |
Raises:
| Type | Description |
|---|---|
PayloadValidationError
|
If validation fails. |
Source code in src/taskq/_validation.py
enrich_pg_dsn ¶
Apply credential to dsn and return a self-contained DSN string.
The credential is written into the DSN userinfo (percent-encoded),
replacing any existing userinfo password - and replacing the userinfo
user when credential.username is set (Vault dynamic DB creds).
This is the only slot that is guaranteed to take effect: asyncpg's
resolver applies userinfo before query parameters (both behind
if user is None / if password is None guards), so a
query-string user= / password= is silently ignored whenever
the DSN already carries userinfo. A stale password= query
parameter is dropped (always shadowed by the userinfo password);
a user= query parameter is dropped only when the userinfo
carries a user to shadow it - a query-carried user with no userinfo
user is the effective principal and is preserved.
sslmode=require is added only when the DSN has no explicit
sslmode, so stronger modes (verify-full) are never downgraded.
Prefer the factory builders (:func:make_pg_pool_factory /
:func:make_dedicated_conn_factory) where possible - they pass the
credential as keyword arguments instead, keeping the token out of
the DSN string entirely.
Source code in src/taskq/auth.py
ensure_sslmode_require ¶
Add sslmode=require to dsn unless an sslmode is already set.
An explicit sslmode is never overridden - in particular stronger
modes (verify-ca / verify-full) must not be downgraded:
require skips certificate verification, which would expose the
very token this module injects to a MITM.
Public because anyone assembling a credential-bearing DSN by hand needs
exactly this rule and must not re-derive it: a token path that silently
connects without TLS puts the credential on the wire. The factory
builders in this module apply it for you; reach for it directly only on
the DSN paths they do not cover (a raw asyncpg.connect, a migration
connection, a DSN handed to another library).
sslmode=disable is an explicit choice and is preserved - that is how
a test container or a Unix-socket deployment opts out.
Because an explicit verify-ca/verify-full is passed through
untouched, it reaches asyncpg needing a CA bundle that this helper does
not supply: asyncpg and libpq do NOT fall back to the system trust store
for a verifying sslmode, they look for ~/.postgresql/root.crt and
raise ClientConfigurationError while parsing connection arguments
(before any socket is opened) when it is absent - which is the normal
case in a container. Such a DSN must carry its own sslrootcert=
(or set PGSSLROOTCERT); see the sslmode note in
docs/guides/managed-identities.md. Deliberately not defaulted here:
picking a trust root is a security decision and the correct bundle is
environment-specific.
Source code in src/taskq/auth.py
make_dedicated_conn_factory ¶
make_dedicated_conn_factory(
dsn: str,
provider: PgCredentialProvider,
*,
command_timeout: float | None = None,
setup: Callable[[Connection], Awaitable[None]]
| None = None,
server_settings: dict[str, str] | None = None,
connection_class: type[Connection] | None = None,
reload_schedule: ReloadSchedule | None = None,
) -> ConnFactory
Build a :data:~taskq.connections.ConnFactory backed by provider.
Used for the worker's notify_conn / leader_conn or
:class:taskq.TaskQ's pg_conn_factory. Like
:func:make_pg_pool_factory, the credential is passed as keyword
arguments (precedence over userinfo and query params; the token
never appears in the DSN string), and password= is an async
callable that asyncpg awaits per physical connection - a fresh token
for a token credential, the issued pair's password for a
username-bearing one.
A dedicated connection is opened once and then held for the life of
the worker, so the callable normally fires exactly once - but these
are precisely the long-lived connections a credential expiry kills,
and the callable is what makes a re-open by asyncpg (a LISTEN
connection reconnecting after the server drops it) authenticate with
the current token rather than the one captured when the factory was
first invoked. reload_credentials re-invokes the factory itself,
which is where a username-bearing pair is replaced.
command_timeout is forwarded to asyncpg.connect as the default
per-operation timeout. The worker's DSN-built notify_conn /
leader_conn carry dispatcher_command_timeout; pass it here too
so a credential-provider deployment does not silently drop the bound
that keeps a wedged query from stalling leader election.
setup runs once inside the factory, immediately after
asyncpg.connect establishes the connection (e.g. registering type
codecs, setting session GUCs) - asyncpg.connect takes no setup
parameter, so the factory applies the hook itself, and a hook failure
terminates the connection rather than returning one half-configured.
For a dedicated connection this is equivalent to
init on a pool - there is no acquire/reuse cycle. The hook is also
declared on the returned factory (see
:func:taskq.connections.with_connection_init), so when this factory
provides the worker's LOOP-scope asyncpg.Connection registration
the per-slot transaction pool inherits it - the codec family that a
bare set_type_codec on one live connection silently loses above
max_concurrency = 1.
server_settings is forwarded to asyncpg.connect and applied as
session-level GUCs at connect time (e.g.
{"statement_timeout": "30s", "search_path": "app"}).
connection_class is forwarded to asyncpg.connect and sets the
:class:asyncpg.Connection subclass for this connection. Use it to
install custom codecs or override connection methods.
reload_schedule is the :class:ReloadSchedule the connection is
rebuilt on, exactly as for :func:make_pg_pool_factory: pass the
consumer's shared schedule for a connection that lives as long as the
process (the worker's notify_conn / leader_conn), so the lease
it is issued counts toward the derived cadence. Omitted, the factory
is taken to be one-shot - a migration connection the caller opens,
uses and closes - and records its lease on a schedule of its own
without the pinned-pair warning, since there is no long-lived
connection to rotate.
Source code in src/taskq/auth.py
753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 | |
make_pg_pool_factory ¶
make_pg_pool_factory(
dsn: str,
provider: PgCredentialProvider,
*,
min_size: int = 1,
max_size: int = 4,
max_inactive_connection_lifetime: float = 300.0,
command_timeout: float | None = None,
statement_cache_size: int = DEFAULT_STATEMENT_CACHE_SIZE,
max_cached_statement_lifetime: int = DEFAULT_MAX_CACHED_STATEMENT_LIFETIME,
init: Callable[[Connection], Awaitable[None]]
| None = None,
setup: Callable[[Connection], Awaitable[None]]
| None = None,
server_settings: dict[str, str] | None = None,
connection_class: type[Connection] | None = None,
reload_schedule: ReloadSchedule | None = None,
) -> PoolFactory
Build a :data:~taskq.connections.PoolFactory backed by provider.
Each invocation fetches a fresh :class:PgCredential from provider
and calls asyncpg.create_pool with the credential as keyword
arguments - password= always, user= when the credential
carries a username. Keyword arguments take precedence over both DSN
userinfo and query parameters in asyncpg's resolver, so a stale
credential baked into dsn can never shadow the fresh one, and the
token never appears in the DSN string. The pool is owned by the
worker (entered on its AsyncExitStack).
Token refresh: password= is passed as an async callable, which
asyncpg invokes and awaits once per physical connection - the
connections opened at pool creation, those opened later by pool
growth, and the replacements opened after
max_inactive_connection_lifetime recycles an idle connection. For
a token credential (username unset: Entra ID, AWS IAM RDS) every
new connection therefore authenticates with a freshly fetched token,
and no external rotation is required. This matters because Postgres
authenticates at connect time only: a credential resolved once and
reused as a fixed string keeps working on already-open connections
while every new connection fails, roughly one token-lifetime after
deploy.
A username-bearing credential (Vault dynamic database credentials)
is one issued pair: asyncpg resolves user= once per pool, so the
pool is pinned to that username and every physical connection
authenticates with the pair's password - re-fetching would burn a
lease per connection for a password the pinned user cannot use. Such
a pool rotates when the factory is re-invoked: SIGHUP /
TASKQ_RELOAD_INTERVAL (taskq.worker.deps.reload_credentials)
rebuilds it on a fresh pair, on the cadence reload_schedule carries
(below). Reload is also the way to force a full pool rebuild for a
token credential (dropping sessions opened under a revoked token).
Per-connection setup: init is forwarded verbatim to
asyncpg.create_pool and runs once per new physical connection
- on the connections opened at pool creation, on connections opened
later by pool growth, and again on replacements opened after
max_inactive_connection_lifetime recycles an idle connection.
That lifecycle is exactly why this setup (registering type codecs -
e.g. pgvector.asyncpg.register_vector - preparing statements,
setting session GUCs) cannot be done correctly after pool creation:
a connection configured by hand is silently replaced under load or
after an idle period. The only per-connection work this factory does
of its own is the credential refresh described above (an asyncpg
password= callback, not an init hook), so a caller-supplied
init is the only hook of its kind: it is passed through unwrapped
and can never silently replace internal setup.
Per-acquire setup: setup is forwarded to asyncpg.create_pool
and runs every time a connection is acquired from the pool
(via pool.acquire()), not just on new-connection creation. Use
it for per-checkout work that must run even when a pooled connection
is reused - e.g. resetting search_path or verifying session
state. Unlike init, setup runs on every acquire, so keep it
lightweight. Both init and setup can be provided simultaneously.
server_settings is forwarded to asyncpg.create_pool and applied
as session-level GUCs on every new connection (e.g.
{"statement_timeout": "30s", "search_path": "app"}). Useful for
per-pool configuration that must be set at connection time.
statement_cache_size / max_cached_statement_lifetime are forwarded
to asyncpg.create_pool and default to TaskQ's tuning (the
taskq.connections module constants, 512 entries, 1 h lifetime ,
which are also the TaskQSettings field defaults). asyncpg's own
defaults (100 / 300 s) thrash on TaskQ's read paths, so a
provider-backed pool must get the same cache treatment as the
DSN-built ones. Call sites with a
:class:~taskq.settings.WorkerSettings in scope pass the values
resolved through :func:taskq.connections.statement_cache_kwargs so
the TASKQ_STATEMENT_CACHE_SIZE / TASKQ_MAX_CACHED_STATEMENT_LIFETIME
env vars apply to provider-backed pools too.
connection_class is forwarded to asyncpg.create_pool and sets
the :class:asyncpg.Connection subclass used by the pool. Use it to
install custom codecs or override connection methods across the
entire pool.
reload_schedule is the :class:ReloadSchedule this pool is rebuilt
on. Each build records the credential it was issued there - its
lease_duration when the issuer reports one - so a consumer with no
explicit interval rebuilds at :data:LEASE_RELOAD_FRACTION of the
granted TTL. Pass one schedule to every factory a consumer rebuilds
together (:func:build_worker_connections does) so they share the
shortest lease; omitted, the factory declares a schedule of its own,
readable with :func:reload_schedule_of. A username-bearing
credential built against a schedule that can derive nothing (no
configured interval, no reported lease) logs
pg-lease-pair-pinned-without-reload naming
TASKQ_RELOAD_INTERVAL, because reconnects will fail authentication
once the lease expires and nothing here can prevent it. It is a
warning, never a refusal: an operator rotating the lease externally
can ignore it.
Source code in src/taskq/auth.py
596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 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 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 | |
make_redis_client_factory ¶
make_redis_client_factory(
url: str | None,
provider: RedisCredentialProvider,
**client_kwargs: Any,
) -> RedisFactory
Build a :data:~taskq.connections.RedisFactory backed by provider.
url is the Redis URL without credentials. The factory attaches
a redis-py CredentialProvider that delegates to provider, so
reconnects re-fetch the credential automatically. Use a rediss://
(TLS) URL - with a plain redis:// URL the bearer token is sent
unencrypted, and the factory logs a warning.
If url is None the factory raises :class:RuntimeError when
called (matches the worker's "Redis not configured" contract).
Source code in src/taskq/auth.py
apply_batch_terminal_outcome
async
¶
apply_batch_terminal_outcome(
backend: Backend,
job: JobRow,
outcome: AttemptOutcome | Literal["noop"],
*,
transaction_conn: ConnLike | None = None,
) -> bool
Apply batch policy after a job reaches a terminal write.
Returns whether the CALLER owes the batch one post-commit completion
re-arbitration (backend.complete_batch(batch_id) with no
connection=): True only when the completion attempt gated out
at its tail-shape probe while riding transaction_conn's still-open
transaction -- a gated attempt is not ordered against the caller's
own commit, so without the reissue an all-terminal batch can be left
'active' with no attempt left to land. The reissue is one awaited
call after the transaction ends, best-effort (M7): dropping it
delays the batch's completion to the leader's stale-batch sweep.
Autonomous callers (no open transaction -- the consumer's hook runs
after consume_one_job's transaction has committed) always get
False and owe nothing.
Called after every terminal write by the consumer and the in-memory
runner. For non-batched jobs (no metadata.batch_id) this returns
immediately, zero overhead. outcome is the dispatch outcome the
caller reports: an attempt-row outcome, or the consumer's "noop"
(a terminal write that matched nothing, the job was never this
dispatch's to move, so no batch counter may budge).
"succeeded": resets the consecutive-failure counter and attempts completion. The reset carries aconsecutive_failures <> 0guard: a batch whose counter is already 0 - the healthy steady state every successful job of the batch pays - is not written, so the batches row is not locked and the member probe does not run; the failure-counter contract is unchanged, a non-zero counter resets exactly as before."failed": increments the consecutive-failure counter. If the threshold is reached, aborts the batch and logsbatch-aborted, abort wins, so no completion attempt runs on that path. The event reports the threshold DECISION (the abort was issued for this count); the flip itself can still be delayed by the bounded batches-row wait, a skip is separately disclosed by thebatch-abort-row-lock-timeoutdebug event, and the next threshold-triggered abort or the stale-batch sweep re-arbitrates. If the threshold is not reached, attempts completion."cancelled"/"crashed": attempts completion."snoozed"/"reservation_denied"/"rate_limit_denied"/"scheduled"/"noop": returns immediately, the job is rescheduled (or was never this dispatch's to move), not terminal.
Completion is self-arbitrating: every terminal outcome issues the
complete_batch attempt, and that statement's NOT EXISTS
guard decides against the live member set in its own snapshot. The
increment/reset count is advisory only, under READ COMMITTED two
members terminating concurrently can each read the other as
non-terminal, so a hook that gated the attempt on that count could
leave a fully-terminal batch for the leader sweep; the optimistic
attempt after the last terminal write is the one that lands, and a
premature one is a no-op.
Best-effort semantics (M7): the increment/reset/abort/complete writes are best-effort. A crash between the terminal job write and the counter increment loses that increment, the failure count is under-counted by one. The next terminal failure re-triggers the check and increments again, so a consistently failing batch still aborts (just one failure later than it would have). The same loss class covers a counter write that skipped instead of parking: the batches-row wait is bounded, and a streaming append holding the row past the budget skips that one increment or reset (the streak is frozen, never reset) while the failure itself stays recorded on the job row. The stale-batch sweep is the safety net for batch STATUS (it transitions stuck active/aborted rows to terminal) but it cannot recover lost failure counts, a crash gap means the consecutive-failure streak is permanently broken, potentially preventing an abort that should have fired.
Source code in src/taskq/batch.py
368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 | |
wait_for_batch
async
¶
wait_for_batch(
db: Connection | Pool,
batch_id: UUID,
*,
schema: str = "taskq",
snooze_interval: timedelta = timedelta(seconds=10),
snooze_via_exception: bool = True,
expect_at_least: int | None = None,
on_empty: Literal["error", "ok"] = "error",
exclude_job_id: UUID | None = None,
) -> BatchCompletionStatus
Convenience helper for the fan-out-then-finalize pattern.
Queries batch children by batch_id using the GIN-indexed
WHERE metadata @> $1::jsonb predicate, the containment form,
not the open-members partial index the completion probes use, because
the poll counts every member including the terminal ones. Each poll is
one round trip: the member counts and the batches row travel in a
single statement.
Inside an actor (snooze_via_exception=True, the default): - If any children are in-flight, raises Snooze(snooze_interval). The consumer transitions the job to scheduled; the actor is retried after snooze_interval without consuming retry budget. - If all children are terminal, returns BatchCompletionStatus.
Outside an actor (snooze_via_exception=False):
- Blocks via asyncio.sleep(snooze_interval) in a loop until all
children are terminal, then returns BatchCompletionStatus.
- Use this form from scripts and integration tests where no consumer
is present to translate a Snooze into a rescheduled job.
- Each poll's pool acquire is bounded by
:data:_POOL_ACQUIRE_TIMEOUT_S: a pool that cannot yield a
connection within the bound surfaces as a :class:TimeoutError
(the project's report idiom for an exhausted pool) rather than
parking the wait forever.
expect_at_least raises :class:~taskq.exceptions.EmptyBatchError
when fewer than the expected number of jobs are present and none are
in flight. on_empty controls the behaviour when the batch has
zero jobs and no batches row exists: "error" (default) raises
:class:~taskq.exceptions.EmptyBatchError; "ok" returns the
empty status. exclude_job_id omits a specific job from the
count, when not set, the batch row's finalizer_job_id is used
automatically.
If the batch row has status = 'aborted' and all jobs are
terminal, raises :class:~taskq.exceptions.BatchAbortedError.
snooze_interval is clamped to a minimum of 1 second.
schema must match the schema used when the PostgresBackend was
constructed (default "taskq").
Source code in src/taskq/batch.py
490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 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 | |
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 | |
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 | |
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 | |
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 | |
register_cron ¶
Add schedule to the module-level registry.
Validates the cron expression at call time. Raises :class:ValueError
on bad expression or mutually exclusive fields. Duplicate calls append
again, deduplication is the caller's responsibility (the DB
(actor, name) UNIQUE constraint is the authoritative gate at
startup time).
Source code in src/taskq/scheduler.py
cron()¶
taskq.cron is both a submodule (src/taskq/cron.py) and, via from taskq.cron import
cron, the name of a re-exported function on the taskq package. This name collision means
the cron() function does not render under the taskq package-level directive above
mkdocstrings resolves taskq.cron to the submodule. The explicit directive below documents
the function itself; see the Cron Scheduling guide for usage.
cron ¶
cron(
expression: str,
actor: str,
*,
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
name: str = "",
identity_key: IdentityKey | None = None,
timezone: str = "UTC",
dst_strategy: DstStrategy = "skip",
enabled: bool = True,
) -> CronScheduleSpec
Declare a cron schedule and auto-register it.
Validates expression via croniter.is_valid() and then a
landability walk: an expression that is syntactically valid but
matches no calendar date (day 30 in February, day 31 in a 30-day
month) also raises :class:ValueError. Raises
:class:ValueError if both payload_factory and static_payload
are provided.
The returned :class:CronScheduleSpec is registered via
:func:~taskq.scheduler.register_cron at decoration time so
decorated schedules are auto-discovered at worker startup without
any explicit register_cron() call.
Startup auto-discovery is create-only, skip-on-conflict. Existing
cron_schedules rows are never modified by the decorator
registration pass. If a @cron decorator's parameters change
after the schedule was first registered, the operator must manually
update or delete and recreate the schedule.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dst_strategy
|
DstStrategy
|
How to handle DST gaps and overlaps.
|
'skip'
|
Source code in src/taskq/cron.py
Error contracts at a glance¶
The surface answers "is this job there?" in three deliberately different ways. Mixing them up is the most common integration bug:
| Operation | Missing / nonexistent target |
|---|---|
TaskQ.get / get_row |
returns None (jobs table first, then the archive tier) |
TaskQ.cancel / JobHandle.cancel |
raises KeyError carrying the job id — a typo'd id and a pruned job are indistinguishable |
TaskQ.stream |
raises KeyError at the opening read and if the row is pruned mid-stream (it cannot fabricate the promised terminal event) |
TaskQ.retry_job |
returns False — a nonexistent id, a non-terminal status, and the attempt-ceiling conflict are all the same False by design |
wait_for_batch (foreign/typo'd batch id) |
raises EmptyBatchError ("is empty or unknown to this client") — pass on_empty="ok" when an empty batch is legitimate |
enqueue performs no actor-registration check: a ref whose actor no
worker declares is enqueued successfully and then parked at the snooze
cadence (released_reason: "actor-not-found", budget-free, surfaced by
the stranded-jobs detector) — check the actor spelling and
taskq job show before looking at queues or capacity.
The full exception taxonomy — every class, its fields, and its remedy — is indexed on the Exceptions API reference; the retry engine's module surface is on the Retry API reference.
Settings typos are silent
Unknown TASKQ_* environment variables are ignored by the
settings loader (dotenvmodel loads only its known fields and logs no
warning). A misspelled knob — TASKQ_MAX_PENDNG_LOCK_TIMEOUT_MS —
silently applies that knob's default instead of failing. Run
taskq doctor, whose environment
finding family names every unknown TASKQ_* variable in the
environment (with the closest real setting name as a remedy hint);
verify spellings against the field descriptions on
TaskQSettings when a setting
appears to have no effect.