Worker

The TaskQ worker is a long-running asyncio process that polls a Postgres-backed job queue, dispatches jobs to registered actor handlers, and keeps the cluster healthy through heartbeating, leader election, and graceful shutdown. Every worker process runs a single asyncio.TaskGroup containing a fixed set of sibling coroutines that live for the lifetime of the process: a heartbeat loop, a NOTIFY listener, a maintenance-leader loop, a producer stub, and max_concurrency consumer loops. All siblings observe a shared shutdown_event; when it is set every sibling returns cleanly and the process exits.

Prerequisites

• Python 3.12+
• TaskQ installed (uv add taskq-py); core includes asyncpg, no extra needed
• A running Postgres instance with the TaskQ schema applied (taskq migrate up)

See ../getting-started/quick-start.md for initial setup and ../architecture.md for system-level context.

---

Internal components

Producer loop. Polls the jobs table for pending rows via the dispatch CTE. Each tick acquires a direct connection from dispatcher_pool, runs the strict-FIFO dispatch SQL (atomic FOR UPDATE SKIP LOCKED + UPDATE … SET status='running'), and pushes dispatched rows onto an in-process asyncio.Queue[JobRow] (local_queue). The local queue's maxsize is set to max_concurrency; back-pressure from a full queue naturally throttles the producer.

Consumer loops. max_concurrency concurrent coroutines drain local_queue. Each iteration pops one JobRow, resolves the actor's DI scope via build_actor_scope, validates the payload against actor_ref.payload_type, registers the job with ActiveJobRegistry, invokes the actor function, writes the terminal state to Postgres, and deregisters. All Postgres writes inside a consumer are wrapped in asyncio.shield so that cancellation during shutdown cannot strand a row in running status.

Heartbeat loop. On every heartbeat_interval tick, acquires one connection from heartbeat_pool, opens a single transaction, and atomically updates workers.last_seen_at, extends jobs.lock_expires_at for all running jobs owned by this worker, extends reservation_slots.lease_expires_at, and (if this worker is the leader) pings maintenance_leader.last_seen_at. After the transaction commits, runs the cancel-controller's run_post_tx to drain any phase-3 abandonment queue. Consecutive failures increment heartbeat_failures; exceeding max_heartbeat_failures triggers isolate_self.

NOTIFY listener. Holds a dedicated direct connection (notify_conn) subscribed to the schema's wake channel (wake_channel(schema), see architecture.md). When a NOTIFY arrives, the listener reads the payload (the inserted row's queue) and calls event.set() only for subscribers whose served queue set includes that queue; an empty payload (the COPY fixup's bulk wake, or an older trigger still live in a rolling deploy) wakes every subscriber. A wake that never arrives, filtered or dropped, costs one fallback poll interval, never work: the poll is authoritative. A health-check coroutine issues SELECT 1 every notify_health_check_interval seconds and reconnects with jittered bounded exponential backoff (±25% multiplicative jitter around each doubling delay) on failure.

External code (for example a bulk-enqueue script) can wake sleeping workers immediately without going through the normal enqueue path. The channel name embeds a hash of the schema rather than the schema itself (so it fits Postgres's 63-byte identifier limit for any schema name); derive it in SQL from the value of TASKQ_SCHEMA_NAME (default taskq):

SELECT pg_notify('taskq_wake_' || left(encode(sha224('taskq'::bytea), 'hex'), 10), '');
-- for the default schema this is the channel taskq_wake_124a200651

In Python, taskq.constants.wake_channel(schema) returns the same name.

Maintenance leader. One worker per schema wins a Postgres advisory lock (pg_try_advisory_lock) and becomes the maintenance leader. The leader runs eleven cooperative sub-loops inside a single asyncio.TaskGroup: _election_loop, _watchdog_loop, _scheduled_wake_loop, _cron_loop, _sweep_loop, _prune_loop, _archive_expiry_loop, _queue_depth_loop, _backlog_detection_loop, _reservation_slots_loop, and _stranded_jobs_loop. Non-leader workers re-attempt election each heartbeat_interval. See maintenance-sweeps.md for what the sweep loops do and why their database work is bounded.

Cancel controller. Runs inside the heartbeat transaction on each tick. Polls jobs.cancel_phase for all jobs owned by this worker. Drives three phases: cooperative observation (set cancel_event, record timestamp), forced escalation (write cancel_phase=2 to Postgres, then task.cancel()), and abandonment queuing (hand the job ID to _pending_abandons for post-transaction cleanup). Phase-3 terminal writes run in run_post_tx after the transaction releases its row locks to avoid deadlock.

Health server. An asyncio-based Unix-domain-socket HTTP server exposing /live, /ready, and /metrics. Started at worker boot when health_enabled=True; stopped gracefully during teardown.

The Unix socket is not reachable via Kubernetes httpGet probes. Use exec probes instead:

livenessProbe:
  exec:
    command: ["taskq", "health", "live"]

In-worker watchdog. Four detectors (TASKQ_WATCHDOG_ENABLED=true, the default) catch hangs and deadlocks that heartbeat and /live cannot see, and force the process down so the supervisor restarts it instead of leaving it wedged:

1. Shutdown deadline. A ShutdownWatchdog task lives outside the worker TaskGroup (as a group child it would be cancelled by the very sibling crash it exists to catch). It parks on shutdown_event, then counts down termination_grace_period, anchored on the first shutdown signal (deps.shutdown_started_at) so the cancel and cleanup graces are not double-counted against the budget. While counting it logs still-alive siblings every watchdog_dump_interval; at the deadline it trips. This is what makes termination_grace_period an enforced budget rather than a validation-only one.
2. Stale loop ticks. Interval-driven siblings opt in via liveness.tick(name, period=...) once per iteration; a loop is stale when its tick age exceeds max(period × watchdog_tick_grace_factor, watchdog_stale_floor). The sweep runs every watchdog_check_interval. Loops that legitimately park (the NOTIFY listener, consumers, the reload coordinator) have no cadence and are covered by the other detectors instead, and gated loops call liveness.forget(name) when their gate closes (for example the leader watchdog while not leader) so an idle gate cannot false-trip.
3. Sibling contract. A long-lived sibling that returns cleanly while no shutdown is in progress is a bug: the sibling spawner sets shutdown_event and raises RuntimeError, so the worker exits instead of running half-staffed. A sibling that raises also sets shutdown_event and increments taskq.worker.sibling_crashes_total. This check stays armed even when the watchdog is disabled.
4. Event-loop lag. A daemon thread measures loop scheduling lag (a fully blocked loop cannot run an in-loop detector). It arms after watchdog_loop_lag_startup_grace or the first liveness tick, whichever comes first, and trips when no beat has been scheduled within watchdog_loop_lag_budget, dumping faulthandler thread frames first.

A trip emits worker-watchdog-trip at CRITICAL (labelled by detector), dumps every live asyncio task (name, coroutine, await-site frames; never locals or payload values), flushes, and calls os._exit(2) with no further awaits: a wedged process cannot be trusted to unwind. In-flight jobs are reclaimed by the leader sweep on lock_lease expiry, and the non-zero exit code guarantees the supervisor restarts the worker.

Observability surfaces: send SIGUSR2 to dump the same task-stack payload on demand (not available on Windows); GET /tasks on the health socket serves it as JSON when TASKQ_HEALTH_TASKS_ENABLED=true; the /ready body reports loop_tick_ages and shutdown_elapsed_seconds, and a stale loop flips readiness to 503 with a stale_loops reason. The OTel instruments are taskq.worker.event_loop_lag_seconds (per-beat scheduling latency), taskq.worker.watchdog_loop_lag_warns_total and taskq.worker.watchdog_trips_total (by detector), taskq.worker.shutdown_duration_seconds, taskq.worker.loop_tick_age_seconds (by loop), and taskq.worker.sibling_crashes_total (by loop); see observability.md: The watchdog family.

Shutdown orchestrator. Handles SIGTERM/SIGINT. Drives the four-phase sequence: DRAINING → CANCELLING → FORCING → RELEASING, then sets shutdown_event so all TaskGroup siblings return.

DI scope chain. Three scope containers (ProcessScope, ThreadScope, LoopScope) are bootstrapped in sequence after open_worker_deps. They resolve declared dependencies for actors at dispatch time using build_actor_scope, which opens a per-invocation TRANSIENT scope. TRANSIENT teardown runs after each job regardless of outcome.

---

Starting a worker

Via the CLI

taskq worker --actors myapp.actors:registry

See cli.md for the full option reference. The --actors argument is required. All other settings load from environment variables or .env files.

Programmatically via worker_main()

worker_main is the production entry point. It sets up logging, starts an asyncio.Runner, and calls _main which wires the full TaskGroup.

from taskq.settings import WorkerSettings
from taskq.worker.run import worker_main
from myapp.actors import registry

settings = WorkerSettings.load()
exit_code = worker_main(settings, actor_registry=registry)

actor_registry must be a Mapping[str, ActorRef]. Passing actor_registry=None runs stub consumers (M0/internal use only). Production code must always pass a registry.

WorkerSettings.load() reads all TASKQ_* environment variables and applies DSN fallback and invariant validation. Always construct settings through load() or load_from_dict(), never via the constructor directly, because post_load() must run.

worker_main returns an int exit code (0 on clean shutdown). In a container entrypoint:

import contextlib
import signal
import sys
from taskq.settings import WorkerSettings
from taskq.worker.run import worker_main
from myapp.actors import registry

if __name__ == "__main__":
    code = worker_main(WorkerSettings.load(), actor_registry=registry)
    # Why: worker_main deliberately leaves the host's SIGTERM disposition
    # alone (an ignored disposition leaks into every process the host forks
    # afterwards); the entrypoint whose next act is its own death installs
    # the ignore here, so a redundant stop-signal landing in the teardown
    # window cannot erase the drain's exit status with -15.
    with contextlib.suppress(ValueError):  # not the main thread -> no window to guard
        signal.signal(signal.SIGTERM, signal.SIG_IGN)
    sys.exit(code)

The entrypoint-side drain-verdict guard. worker_main deliberately does not touch the host's SIGTERM disposition: an ignored disposition is process-global and inherited, so one that leaked from the worker loop would land in every process the host forks afterwards (a proven -9-class CI failure). The trade is that between worker_main returning and the entrypoint's own exit, a redundant stop-signal — from a supervisor that has not yet noticed the drain finished — is delivered by the default disposition and kills the process with -15, replacing the drain's exit code the same way a SIGKILL would. An entrypoint whose next act after worker_main returns is its own death should therefore install signal.signal(signal.SIGTERM, signal.SIG_IGN) right after the call, before exiting, as the example above and examples/worker.py's _run_worker_with_verdict_guard do; the contextlib.suppress(ValueError) guard covers the not-the-main-thread case, where the window no longer exists to guard.

---

Actor registry

The worker accepts actors as either a Mapping[str, ActorRef] (keys are actor names) or an Iterable[ActorRef] (names are read from ActorRef.name).

Defining actors in a module:

# myapp/actors.py
from pydantic import BaseModel
from taskq import actor, RetryPolicy


class SendEmailPayload(BaseModel):
    to: str
    subject: str
    body: str


@actor(queue="email", retry=RetryPolicy(kind="transient", max_attempts=5))
async def send_email(payload: SendEmailPayload) -> None: ...


# Iterable form: pass the ActorRef objects directly
registry = [send_email]

# Mapping form: keyed by actor name
registry_map: dict[str, object] = {"send_email": send_email}

CLI invocation using the iterable form:

taskq worker --actors myapp.actors:registry

CLI invocation using the mapping form:

taskq worker --actors myapp.actors:registry_map

The module:attr string must resolve to a Mapping[str, ActorRef] or an Iterable[ActorRef] at import time. If the attribute is neither, the CLI prints an error and exits with code 1.

Generator registries are unsafe. The attribute resolved from MODULE:ATTR must be a reusable Mapping or a list/tuple of ActorRef, not a generator or other one-shot iterable. The CLI iterates the resolved object twice during type-checking: the first pass exhausts a generator, and the second pass sees an empty sequence and silently builds an empty registry, causing all dispatched jobs to be dropped with dispatch-actor-not-found errors.

See actors.md for the full @actor decorator reference.

---

Queue selection

Each worker consumes from one or more named queues. The queue list can be set three ways (highest to lowest precedence):

1. --queues CLI flag (one flag per queue name)
2. TASKQ_QUEUES environment variable (comma-separated)
3. Default: ["default"]

Multiple queues per worker:

taskq worker --actors myapp.actors:registry --queues default --queues priority --queues email

TASKQ_QUEUES=default,priority taskq worker --actors myapp.actors:registry

The --queues flag is a multi-value Typer option: pass it once per queue name. Do not pass a single comma-separated string to --queues on the command line; use the environment variable form for comma-separated input.

The dispatch CTE filters jobs.queue = ANY($queues), so one worker process can consume from any subset of queues in a single polling round.

Actors declare which queue they target via @actor(queue="..."). A worker that does not include that queue in its TASKQ_QUEUES list will never pick up those jobs.

Actors on unconsumed queues

An actor registered on a queue this worker does not consume is stranded from this worker's perspective: enqueue keeps succeeding (any client can insert into any queue), but this worker's dispatch CTE filters jobs.queue = ANY($queues) and will never claim those jobs. If no other worker consumes the queue either, the jobs sit pending forever with no error anywhere: the silent pile-up failure mode.

This is deliberately a warning, not a fatal error, because a split-queue topology is legitimate: one process consuming ["default"] while another consumes ["cron"] is a normal deployment shape. The check is judged against the stored actor_config.queue when a stored row already exists, the assignment taskq queue migrate rewrites and the cron leader's fires follow, falling back to the @actor(queue=...) literal only for an actor with no stored row yet (this check runs before the row is seeded on a first-ever boot). A worker whose code literal has not yet caught up with a fleet-wide queue move is the normal, blessed state during a rolling deploy, so judging coverage by the literal alone would warn on that healthy window instead of on the actor whose routed work nothing can actually claim. At bootstrap, every worker whose registry contains an actor routed to a queue outside its own subscription logs one warning:

actors-on-unconsumed-queues  actors={"mailer": "email", "nightly": "cron"}  queues=["cron", "email"]  worker_queues=["default"]

The note field spells out the condition under which it is a real problem: only when no other worker consumes those queues.

How to detect and resolve stranding:

• At startup: the actors-on-unconsumed-queues warning names every affected actor, the queue that actually routes it (the stored assignment when one exists), and this worker's subscription, with the distinct unconsumed queue names in the queues field.
• After the fact: taskq actor-config list shows each registered actor's stored queue column; cross-check that every queue appearing there is consumed by some worker's TASKQ_QUEUES / --queues. The admin UI's Jobs page (taskq ui serve) shows pending jobs per queue and makes a growing backlog on one queue visible.
• Fix: either add the actor's queue to this worker (TASKQ_QUEUES=default,cron / repeat the --queues flag), or point the actor at a consumed queue via @actor(queue=...) and re-register (see troubleshooting.md, "Jobs stuck in pending").

Worker consumes no queues

A second bootstrap warning covers the blunter misconfiguration: TASKQ_QUEUES resolving to an empty list (typically the variable set to an empty value; unset, it defaults to ["default"]). The dispatch CTE then matches no queue at all, so this worker never claims a job, on any queue: every job meant for it sits pending forever unless another worker consumes the queue: the same silent pile-up as above, but for every queue at once. Unlike the warning above, no fleet legitimately runs a worker that consumes nothing, so there is no split-queue caveat to spell out. It fires once at bootstrap, with or without registered actors:

worker-consumes-no-queues  worker_queues=[]

Fix: set TASKQ_QUEUES (or pass --queues once per queue) to the queues this worker should consume.

Queue dispatch modes

Each queue has a mode column in the queues table that controls how the dispatch CTE orders candidates. The mode is resolved by querying the queues table at dispatch time (one indexed lookup per batch). Queues not present in the table default to strict_fifo.

Mode	Behaviour
strict_fifo (default)	Jobs are dispatched in priority-then-time order (priority DESC, scheduled_at, id). Every pending job competes freely: a deep queue of one actor can starve others if all candidates share high priority.
round_robin	Per-actor lateral dispatch. Jobs are interleaved by fairness_key cohort. Within a cohort, ordering is priority-then-time. Across cohorts, dispatch picks round-robin: one job from each fairness cohort per round. This prevents a deep queue of one tenant/actor from starving all others.

When to use round_robin: - Multi-tenant queues where one busy tenant's backlog must not block others. - Queues with distinct fairness cohorts (e.g. per-customer processing). - Any scenario where strict FIFO would cause head-of-line blocking across unrelated work streams.

Setting a queue's mode:

taskq queues set-mode multi round_robin

The change takes effect within the queue-mode cache's 5 s TTL: no worker restart needed (dispatch serves queue modes from a per-worker TTL cache; the process running set-mode invalidates its own caches immediately, and each worker picks the flip up on its next cache refill). Queues with no row default to strict_fifo, and nothing in TaskQ creates rows, so a queue is strict_fifo until you run this command.

round_robin alone does nothing: it needs fairness_key on the jobs

Setting a queue's mode is only half the change. Cohorts are formed by fairness_key, and the dispatch CTE partitions on COALESCE(fairness_key, '__null__'), so jobs enqueued without one all collapse into a single __null__ cohort. Round-robin across one cohort is exactly priority-then-time order: a round_robin queue where nothing sets a fairness_key behaves identically to strict_fifo.

Nothing warns you. taskq queues set-mode succeeds, the queue reports round_robin, and the starvation you set out to fix continues unchanged.

The key is an enqueue-time argument; there is no actor-level declaration, so fixing starvation means changing the producer, not just the queue:

# Without fairness_key, set-mode round_robin is inert.
await client.enqueue(process_doc, payload, fairness_key=payload.tenant_id)

Verify cohorts actually exist before trusting the mode:

SELECT coalesce(fairness_key, '__null__') AS cohort, count(*)
FROM {schema}.jobs
WHERE queue = 'multi' AND status = 'pending'
GROUP BY 1 ORDER BY 2 DESC;

One __null__ row is the fingerprint of an inert round_robin queue.

fairness_key and round_robin: fairness_key is an enqueue-time argument (JobsClient.enqueue(..., fairness_key=...)); there is no actor-level declaration, and the key travels on each job's row. The key only affects dispatch on a round_robin queue; on the default strict_fifo it is stored and ignored. See jobs-clients.md for the fairness_key enqueue parameter.

dispatch_oversample: The dispatch CTE gathers residual × oversample candidates per actor in the LATERAL subquery. residual is the actor's remaining concurrency capacity. The oversample multiplier (default 2, env TASKQ_DISPATCH_OVERSAMPLE) absorbs identity-key collisions and multi-producer contention without reducing dispatch yield. Set to 1 when no identity_key is used and single-producer deployment.

The candidate 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, and a round that finds its whole window locked expands the window geometrically (up to 8×) while claimable rows remain. The expansion covers transient oversubscription; size oversample at or above the number of dispatchers that routinely poll the same actor+queue so the steady state never pays the expansion round trip.

---

Concurrency model

TaskQ limits concurrency at three independent scopes. They compose: a job must satisfy all three, and they are not substitutes for one another. Reaching for the wrong one is the most common configuration mistake in production.

Layer	Knob	Scope	Strict?	Change takes effect
Process	TASKQ_MAX_CONCURRENCY	One worker process	Per-process only	Worker restart
Queue	queues.max_concurrent (taskq queues set-max-concurrent)	Fleet-wide, per queue	Yes (leased slots)	Worker restart (read at startup)
Actor	actor_config.max_concurrent (taskq actor-config set)	Fleet-wide, per actor	No (best-effort)	Next dispatch cycle
Resource	@actor(reservations=[...]) ([ConcurrencyReservation])(rate-limiting.md#concurrencyreservation)	Fleet-wide, per named slot bucket	Yes (leased slots)	Worker restart

Total fleet concurrency for an actor is bounded by the lowest of these that applies. Note the process layer multiplies: TASKQ_MAX_CONCURRENCY=8 across 5 replicas is up to 40 concurrent jobs before any fleet-wide cap applies.

Which one do I want?

• "This one actor must never run in parallel with itself.": a ConcurrencyReservation with slots=1, declared via @actor(reservations=[...]). This is the only real mutex. See Choosing a concurrency control.
• "This one actor should not hog the fleet.": taskq actor-config set <actor> --max-concurrent N. Best-effort is fine here; you want a damper, not a guarantee.
• "At most N jobs from this queue, fleet-wide, whatever publishes to it.": taskq queues set-max-concurrent <queue> --max-concurrent N.
• "How much work should one pod do at once?": TASKQ_MAX_CONCURRENCY, sized against the pod's CPU/memory and pool sizes.

Never size a queue for its narrowest actor

A queue cap bounds a whole tier and cannot distinguish a thread-unsafe native call from an HTTP round-trip. Setting a queue's cap to 1 to protect one unsafe actor serializes every other actor on that queue; a fleet that looks correctly configured, reports healthy, and produces almost nothing. The symptom is low throughput with idle workers and a growing pending backlog on one queue only.

Put the constraint where the constraint actually is: a reservation on the offending actor, or move it to its own queue.

Process-level bound

max_concurrency (default 8, env TASKQ_MAX_CONCURRENCY) is the upper bound on simultaneously executing jobs; and, since the producer sizes every claim by the slots genuinely free (local_queue capacity minus queued rows minus actively running jobs), also the upper bound on rows this worker holds locked at once. A worker whose consumers are all busy claims nothing further: its next claim fires the moment a job finishes (the consumer wakes the producer at the deregister that frees the slot), not on the next poll tick. The transient window between a consumer taking a row from the queue and registering it in the active-jobs registry is one scheduler step wide, so a claim can overshoot the free-slot count by at most the consumer count for that instant.

The trade this makes: local prefetch used to hide one dispatch round trip for sub-millisecond actors (the producer claimed a fresh batch while the last one was still executing). With exact slot accounting, a finished row's replacement claim waits for the accounting to settle, bounded by the same-tick deregister wake, not a poll interval. The win is that a busy worker no longer locks up to 2× max_concurrency rows (double the crash-reclaim exposure, and pending work locked behind long jobs while peer workers idle).

worker_pool_size is derived automatically:

worker_pool_size = int(max_concurrency * 1.5)

The 1.5 factor provides headroom for terminal writes that occur just after a job finishes while the slot is being recycled. This pool is used for worker-path Postgres writes (mark_succeeded, mark_failed_or_retry, mark_cancelled, mark_abandoned). It may route through PgBouncer in transaction mode; see PgBouncer compatibility.

When a LOOP-scope asyncpg.Connection is registered and max_concurrency > 1, the setting also sizes a fourth pool: the per-slot transaction pool (max_concurrency + 1 direct connections, fully warmed at boot) that carries every per-job transaction: the actor's own writes (the actor receives its slot connection by injection), the terminal write, and transactional sub-enqueues. It is always direct-DSN and worker-internal: there is deliberately no WorkerConnections slot for it, and a worker that cannot open it fails to boot.

Session state on slot connections

At max_concurrency = 1 the actor runs on the registered connection itself, so everything configured on it applies by construction. Above that the actor runs on a slot connection. Before it builds the pool, the worker reads the registered connection's live session state (its search_path and its role) and opens every slot connection with those values as startup settings. Unqualified table references and RLS-driving roles therefore resolve identically at every concurrency, whether the application set them with a connect keyword, from an init hook, or with a plain SET after connecting. The worker logs slot_pool_inherits_registered_session naming what it carried across; if a value cannot be read it logs slot-pool-registered-session-unreadable at WARN rather than falling back silently.

Reading before the build is what keeps the pool warm: the slot connections are opened during the build, so state applied afterwards would reach only connections re-established lazily on first acquire, putting connection establishment, and a managed-identity credential fetch, back into the dispatch path. It is also what makes the state survive a SIGHUP credential rotation, which rebuilds the pool from the same factory.

Type codecs and init hooks carry only through a hook-declaring registration. A codec registered on a live connection with set_type_codec is stored inside the driver's per-connection state, which exposes no way to read it back, so what the worker can replay onto slot connections depends on how the LOOP-scope connection was registered:

• Registered through a factory that declares its init hook, carried. Register the connection with register_factory(asyncpg.Connection, Scope.LOOP, ...), where the factory is wrapped in taskq.connections.with_connection_init(factory, init) (for a hand-rolled or plain-DSN factory) or built by taskq.auth.make_dedicated_conn_factory(..., setup=...) (for a credential-provider deployment; the factory applies setup to the fresh connection itself, asyncpg.connect takes no setup parameter). The worker reads the declared hook off the registration and installs it as the slot pool's init, so every slot connection runs the same per-connection setup and decodes identically at every concurrency. The boot log records it as slot_pool_inherits_registered_session with init_hook=true.
• Registered as a raw, already-built connection, not carried, and warned. register_value(asyncpg.Connection, Scope.LOOP, conn) hands the worker a finished connection whose codecs and hooks are sealed inside the driver; there is nothing to read back and nothing to replay. When the per-slot pool activates on such a registration the worker logs slot_pool_registered_setup_not_inherited at WARN, naming this boundary and the factory channel above. The slot connections still work, but a query relying on such a codec returns the driver's default representation (e.g. raw strings for json), not the application's decoded form, and only this warning says so. At max_concurrency = 1 the actor runs on the registered connection itself, so the codec is present there and absent above it.

Either way, the session-state read-back (search_path, role) applies independently of the codec channel.

dispatcher_pool_size (default 4) and heartbeat_pool_size (default 4) are independent pools; both always use the direct DSN.

The worker spawns exactly max_concurrency consumer loop coroutines. They are cooperatively concurrent (asyncio, not threads). CPU-bound work should be offloaded to a thread pool executor via asyncio.get_running_loop().run_in_executor.

For fleet-sizing math (connections per worker against max_connections, throughput per process) see ops.md: Sizing.

---

Dispatch sequence

Each consumer loop iteration follows this sequence:

1. Dequeue from local queue. The consumer races local_queue.get() against shutdown_event.wait(). On shutdown win, the consumer returns cleanly.

2. Actor lookup. The job's actor field is looked up in actor_registry. If not found, logs dispatch-actor-not-found and continues to the next job (the row remains running; the sweep will reclaim it after lock_lease expires).

3. Payload validation. actor_ref.payload_type.model_validate(job.payload) validates the raw JSONB dict against the declared Pydantic model. A PayloadValidationError is non-retryable and immediately fails the job.

4. DI scope resolution. build_actor_scope opens a TRANSIENT scope, resolves all Annotated[T, Scope.X] parameters declared by the actor handler, and returns them as resolved.di_kwargs.

5. JobContext construction. A JobContext[P] is built with the validated payload, a fresh cancel_event, the worker ID, attempt number, and an OTel consumer span linked to the producer span via trace_id / span_id from the job row.

6. Rate-limit / reservation acquire. If the actor declares rate_limits or reservations and a RateLimitRegistry is registered at LOOP scope, acquire_for_actor is called. On denial (ReservationUnavailable), the job is snoozed and the actor is not invoked.

7. Actor invocation. The actor function is called with (payload, ctx, **di_kwargs). If a LOOP-scope asyncpg.Connection is registered, the invocation and mark_succeeded_with_conn are wrapped in a single transaction, making the job status update and any sub-enqueues transactional. On a single-slot worker (TASKQ_MAX_CONCURRENCY=1) that transaction runs on the registered LOOP-scope connection; at higher concurrency the worker opens a per-slot transaction pool, each job transacts on its own slot connection, and the actor receives that same slot connection by injection, so the actor's own writes join the transaction and concurrent slots never share a connection.

8. Result / exception handling. See Retry and backoff. All terminal Postgres writes are wrapped in asyncio.shield.

9. TRANSIENT scope teardown. Runs unconditionally after each invocation regardless of outcome.

10. Rate-limit release. release_for_actor is called in the finally block (best-effort, not shielded).

---

Retry and backoff

Each actor carries a RetryPolicy. The default policy is:

RetryPolicy(
    kind="transient",
    max_attempts=3,
    backoff="exponential",
    base=timedelta(seconds=5),
    cap=timedelta(hours=1),
    jitter=0.2,
)

Policy fields:

Field	Type	Default	Description
kind	"transient" \| "indefinite" \| "non_retryable"	"transient"	Retry strategy
max_attempts	int	3	Maximum attempts before failing (used by transient)
time_budget	timedelta \| None	None	Optional wall-clock deadline for indefinite retries; see below
backoff	"exponential" \| "linear" \| "fixed"	"exponential"	Delay formula
base	timedelta	5s	Base delay
cap	timedelta	1h	Per-attempt backoff ceiling
jitter	float	0.2	Multiplicative jitter fraction in [0.0, 1.0]

time_budget on indefinite retries. When kind="indefinite" and time_budget is set, the enqueue path automatically computes schedule_to_close = enqueue_time + time_budget. Callers do not need to compute an absolute deadline themselves. See retries.md for details.

Policy kinds:

kind	Behavior
transient	Retries up to max_attempts. attempt >= max_attempts → failed.
indefinite	Retries until schedule_to_close deadline. max_attempts is ignored for the retry decision.
non_retryable	Never retried; any exception immediately fails the job.

Backoff formulas (all capped at min(policy.cap, max_retry_backoff)):

backoff	Formula
exponential	base * 2^(attempt-1)
linear	base * attempt
fixed	base

After computing the raw delay, multiplicative jitter is applied: delay = raw * uniform(1 - jitter, 1 + jitter). The default jitter=0.2 gives ±20% variation.

Global backoff ceiling: TASKQ_MAX_RETRY_BACKOFF (default 24h) is applied as effective_cap = min(policy.cap, max_retry_backoff). This prevents a misconfigured cap=timedelta(days=365) from stranding jobs silently.

Retry-delay floor: every failure-retry, snooze, and non-consuming retry-after delay is floored at 1 second (MIN_DEFERRAL_INTERVAL), applied after the cap and — on the computed-curve path, the only one jittered — the jitter draw, so a sub-second base (or a zero Retry-After) cannot produce a zero-period claim/run/fail loop monopolising a worker slot; for a cap below 1 second the floor, not the cap, is the binding bound on the stamped delay. The exception is a consuming RetryAfter (consume_budget=True), which keeps its raw delay: an immediate consuming retry is a real execution, bounded by the budget it spends.

Control-flow exceptions:

• Snooze(delay: timedelta): re-schedules the job with mark_snoozed and increments the row's snooze_count counter column. If schedule_to_close has passed, the job is failed with DeadlineExceeded.
• RetryAfter(delay: timedelta, consume_budget: bool): re-schedules the job immediately. When consume_budget=True, the re-schedule counts against max_attempts. When False, it does not. Fails the job if schedule_to_close has passed or max_attempts is exhausted.

Both exceptions are raised from inside the actor body; they are not errors.

---

Cancellation

Cancellation is a three-phase protocol coordinated between the API layer (which writes cancel_phase to Postgres) and the heartbeat loop (which polls it). The phases map to CancelPhase enum values:

Phase	Value	Trigger	Worker action
NONE	0	n/a	Job running normally
COOPERATIVE	1	Cancel flag written by API	Sets ctx.cancel_event; actor may observe and return cooperatively
FORCED	2	cancellation_grace_period elapsed	Writes cancel_phase=2 to Postgres, then calls task.cancel()
ABANDON_PENDING	3	cleanup_grace_period elapsed	Queues job for mark_abandoned after heartbeat transaction commits

Actor-side cooperative cancellation:

@actor
async def long_running(payload: MyPayload, ctx: JobContext[MyPayload]) -> None:
    for chunk in chunks:
        if ctx.cancel_event.is_set():
            return  # cooperative exit; job will be marked cancelled
        await process(chunk)

The CancelController protocol has two methods called by the heartbeat loop:

• run_in_tx(conn): phases 1-3 eligibility check, runs inside the heartbeat transaction.
• run_post_tx(): drains the phase-3 abandonment queue after the transaction commits.

run_post_tx must always be called after run_in_tx on the same tick, even if run_in_tx raises. The heartbeat loop calls it unconditionally.

The consumer skips mark_cancelled when cancel_phase >= 3 (ABANDON_PENDING), because run_post_tx owns that terminal write.

---

Heartbeat and liveness

Every heartbeat_interval seconds (default 10.0s, env TASKQ_HEARTBEAT_INTERVAL) the heartbeat loop:

1. Acquires a connection from heartbeat_pool with a timeout=interval.
2. Opens a transaction and executes four SQL statements atomically: update workers.last_seen_at, extend jobs.lock_expires_at by lock_lease, extend reservation_slots.lease_expires_at, and (if leader) ping maintenance_leader.last_seen_at.
3. Runs the cancel-controller run_in_tx inside the same transaction.
4. After the transaction commits, calls run_post_tx.

lock_lease (default 60.0s, env TASKQ_LOCK_LEASE) is the duration a job's lock remains valid without a heartbeat. The invariant lock_lease >= max(heartbeat_interval, heartbeat_command_timeout) + (max_heartbeat_failures + 1) * (heartbeat_interval + heartbeat_command_timeout) is enforced at startup and prevents the recovery sweep from reclaiming locks on a live worker that experienced transient heartbeat delays: the lease must outlive the worst coherent failed-beat cascade to the heartbeat's isolate decision, including the command timeout a contended beat can burn and the last good beat's tail. At the defaults that floor is 58s against the 60s lease.

If heartbeat_pool.acquire() times out, raises a connection error, run_in_tx raises an OSError, or the tick fails with any other unexpected (non-transient) error, heartbeat_failures is incremented. The transient and unexpected arms share one ledger: the same counter, the same heartbeat_consecutive_failures gauge, the same miss record, the same halfway early warning, and the same threshold. Only a fully successful tick resets the count. This is deliberate: a persistent non-transient fault (a revoked UPDATE grant, a driver contract violation) fails every tick exactly as a dead Postgres does, and an unexpected error that only logged while the counter stayed at zero would tick forever on a worker that looks healthy (green /ready, gauge pinned at 0) while its locks expire underneath every running job. The heartbeat loop deliberately does not use UnexpectedLoopErrorGuard, the leader loops' backstop: the guard's contract is deliberately fatal (re-raise into the TaskGroup, recover by lease expiry), while isolate_self proactively transitions the running rows before signalling shutdown, and the lease arithmetic (lock_lease >= max(heartbeat_interval, heartbeat_command_timeout) + (max_heartbeat_failures + 1) * (heartbeat_interval + heartbeat_command_timeout)) is sized against this one threshold. When heartbeat_failures > max_heartbeat_failures (default 3), isolate_self is called:

1. Opens a fresh direct connection (bypassing heartbeat_pool, which may be exhausted).
2. In a transaction, reads all running jobs owned by this worker.
3. For each job: transitions retryable jobs to pending (scheduled on the row's deterministic reclaim backoff, ≈5s at defaults) and non-retryable jobs to crashed. Writes an attempt record with error_class=HeartbeatLost.
4. Always sets shutdown_event so the process exits.

The isolate path writes the same reclaim event the leader's sweep writes (reason='lock_expired', cause='isolate_self'), so heartbeat-loss reclaims reach the Backend.poll_reclaim_events() / TaskQ.watch_reclaims() feed exactly like sweep reclaims.

One residual to know about: locally-running sync actors are cancelled and joined bounded (cancellation_grace_period + cleanup_grace_period plus close slack), but a sync actor's thread cannot be interrupted, and one that outlives the join keeps its row out of the re-pend; when that consumer's unwinding never completes, the interrupt write never lands, so the row's release falls to the same lock-lease expiry backstop the sweep's reclaims rest on. The operator-facing contract for sync actors at shutdown is in actors.md — Sync actors and shutdown.

---

Leader election

Every worker competes for the Postgres advisory lock taskq:maintenance_leader:<schema> (built by taskq.constants.schema_lock_name, e.g. taskq:maintenance_leader:taskq for the default schema) by calling pg_try_advisory_lock(hashtextextended($1, 0)) on a dedicated direct connection (leader_conn). Only one worker per schema can hold the lock; it is held for the lifetime of the connection (session-scoped advisory lock).

Advisory locks are session-scoped and are dropped when the connection is released. This is why leader_conn must use pg_dsn_direct and cannot route through PgBouncer in transaction mode: transaction-mode pooling releases the underlying session between transactions, which would silently drop the lock.

The elected leader:

• Upserts a row into maintenance_leader (worker ID, elected timestamp).
• Runs the maintenance sweep loop every TASKQ_SWEEP_INTERVAL (default 30 s): reclaim_expired_locks (sweep 1), deadline_sweep (sweep 2), and, when the backend supports them, sweep_leaked_reservation_slots (sweep 4), sweep_expired_results, cleanup_stale_workers, and complete_stale_batches (marks active batches with zero non-terminal child jobs as complete, the safety net for lost apply_batch_terminal_outcome hooks). Each event-writing sweep transitions at most one bounded batch per call (TASKQ_EVENT_WRITER_BATCH_SIZE rows, its own committed transaction with a server-side statement_timeout); a call that returns rows drains up to TASKQ_SWEEP_DRAIN_BATCHES batches per tick before leaving the remainder to the next tick; every batch commits, so a stopped drain keeps its progress. See maintenance-sweeps.md for the reasoning and the tuning knobs.
• Runs the scheduled-wake loop every 1 second, promoting due scheduled jobs to pending (scheduled_at <= statement_timestamp(): a STABLE bound so the partial index serves the scan) and issuing a pg_notify wake signal. One bounded batch per tick; a larger due backlog drains across the one-second ticks.
• Runs the cron loop every 1 second on a dedicated connection, firing due schedules under the schema-qualified cron advisory lock (taskq:cron:<schema>). One tick fires at most TASKQ_CRON_TICK_LIMIT schedules; a catch-up burst drains across successive ticks. Per-schedule failures (payload factory, missing actor) are isolated per schedule; an insert-level failure aborts the whole tick and the next one-second tick retries.
• Runs the prune loop (Sweep 5) once daily at TASKQ_PRUNE_SCHEDULE_UTC (default 03:00 UTC). Moves terminal jobs from jobs to jobs_archive after their per-status retention period has elapsed. After job pruning completes, prune_old_batches deletes completed batch rows past the same cutoff (see Batch Subsystem). Acquires the schema-qualified advisory lock taskq:prune:<schema> to prevent concurrent runs across a rolling deploy.
• Runs the archive expiry loop (Sweep 6) once daily at TASKQ_ARCHIVE_EXPIRY_SCHEDULE_UTC (default 04:00 UTC, 1 hour after the prune). Hard-deletes rows from jobs_archive once their expire_at has passed. Acquires the schema-qualified advisory lock taskq:archive_expiry:<schema>.
• Runs the stranded-jobs detector (_stranded_jobs_loop) every 60 seconds: warns (does not delete or reassign) when pending/scheduled jobs exist for an actor with no actor_config row, typically because the actor was removed from the registry but jobs referencing it are still enqueued.
• Samples queue depth and reservation slot counts every 15 seconds for OTel gauges. The backlog detector (_backlog_detection_loop, jobs-by-status and oldest-due-age gauges) is deliberately not leader-gated: every worker samples it, so the promotion-stall signature stays visible under the exact election failure it exists to expose.

A watchdog coroutine probes leader_monitor_conn (a second dedicated direct connection) every 5 seconds. On connection failure, the watchdog clears is_leader and closes both leader connections, which releases the advisory lock and allows another worker to win the next election.

Non-leader workers re-attempt election every heartbeat_interval seconds.

---

Graceful shutdown

SIGTERM (or SIGINT) triggers orchestrate_shutdown. A second signal fast-advances CANCELLING → FORCING. A third signal calls sys.exit(1) (Kubernetes SIGKILL is the hard backstop). A third signal sent while the tracked-actor reap gate is waiting exits through sys.exit(1)'s interpreter unwind, which joins the default executor like any join: a still-running actor thread is waited out, not killed. That shape is operator-forced, predates this design, and the two-legged promise below does not cover it.

Phase sequence:

Phase	shutdown_phase value	Action
DRAINING	1	Sets producer_stop_event; calls drain_local_queue_to_pending to re-pend locked-but-not-started rows (the claim's attempt increment is refunded)
CANCELLING	2	Sets cancel_event on all in-flight jobs and stamps their cancel origin as the shutdown; waits up to cancellation_grace_period for cooperative exit
FORCING	3	Calls task.cancel() on remaining jobs (even when a job's write_cancel_escalation PG write fails; the local cancel is the process-side half of the phase and is never skipped); issues write_cancel_escalation(phase=2) per job, which lands only on rows already carrying an operator's cancel request (the row says who asked)
RELEASING	4	Releases still-running jobs back to the fleet via mark_interrupted; the spent attempt stands (no refund), one interrupted event, held scheduled behind the remaining termination_grace_period budget plus the watchdog's exit tail (the dump-interval lag before the deadline trip is observed, and the bounded metrics flush before os._exit); or lock_lease when the shutdown watchdog is disabled; jobs under an operator cancel are terminalised here with the operator's OWN verdict (mark_cancelled, forced - the same fenced write the unwinding consumer races), never abandoned; closes leader_conn (releasing the advisory lock)

DRAINING phase detail. drain_local_queue_to_pending re-pends DB-level rows where status='running' AND locked_by_worker is this worker, excluding jobs with a live consumer. The claim stamps started_at at claim time, so the predicate cannot distinguish "claimed but unstarted" from "executing"; only this process's active-jobs registry can, which is what the exclusion reads. The producer loop repeats the same hand-back on its own exit, so a claim round that was in flight when the stop event landed is caught too. Rows the consumer loops took off the local queue before the stop signal are never dispatched past it.

RELEASING phase detail. The RELEASING phase releases rows whose actors never unwound, externally (not from the job's own task) via mark_interrupted: the row goes back to the fleet with the spent attempt standing (no refund; the attempt started executing), held unclaimable until this process is provably gone; so the fleet never has two live runners for one row. A job under an operator cancel is the exception: the release's cancel_phase = 0 fence declines it and the operator ladder's terminal (mark_abandoned) still runs for it. The job's asyncio task was already cancelled at FORCING; any pending shield calls in the consumer may or may not succeed.

The release is held back until the actor has provably exited. The consumer's own cancellation arm (not just the orchestrator's RELEASING pass) writes the release, and it no longer assumes the actor unwound with the await. An async actor has provably unwound by the time the CancelledError reaches the consumer, the row is released pending, claimable immediately. A sync def actor has not: task.cancel() cancels the await on the executor thread, never the thread, and the transactional path's tx_task unwinds its rollback asynchronously after the cancel. Both leave tracked handles on the job's context, and the consumer parks on them, bounded by the remaining termination budget minus the release write's own retry budget, then capped by the lease (lock_lease - heartbeat_interval - write budget; the heartbeat stops at shutdown_event, so the park must never outlive the lease the reclaim sweep reads), releasing pending (hold=0) only on a provable exit inside that window. An actor that outlives the window (an unbounded sync loop, say) is still released, never stranded: the hold is the rest of the process's exit window plus the watchdog's exit tail. A cancellation landing on the park (the TaskGroup teardown after RELEASING) ends the waiting, never the release.

The exit the hold models is enforced, not hoped for. "Until this process is provably gone" is true by construction, on both exits. The shutdown watchdog stays armed until every tracked actor handle is reaped: either the handle completes (the actor finished on its own; the remaining process lifetime is bounded pool closes, which touch no row), or the deadline trip fires and os._exit ends the process with the still-running thread inside it; the trip's reason says exactly that (tracked-actor-outlived-teardown, with the live handle count, so a dump is not misread as a loop stall). Without that gate the clean path disarmed the watchdog and then parked in the default executor's join (THREAD_JOIN_TIMEOUT, 300s) waiting for the very thread the hold assumed was gone: a released row became claimable while its actor still ran. The thread's post-release completion accomplishes nothing (its row was already released-with-hold; the fleet re-attempts the work), so ending an outlived attempt at the deadline loses nothing.

The exit tail exists because the deadline trip is not instantaneous: the watchdog checks the deadline once per watchdog_dump_interval (the final sleep is clipped to the deadline, so real check lag is loop jitter and the dump-interval term is margin) and then dumps stacks and joins a bounded (~2s) metrics flush before os._exit, so the process can be alive a few seconds past the deadline. A hold that ended at the bare deadline would leave exactly that window in which the released row is claimable while the dying process could still touch it. Residual: the stack render and the critical log write inside the trip have no bound of their own; the ~1s of slack in the tail is a heuristic, and a full stderr pipe under a slow Docker logging driver can outlive any fixed slack. Everything else in the tail is bounded; this one term is documented padding, and the cost of under-padding it is overlap only in the pathological-logger case.

The platform-grace window. What the platform does around SIGTERM must fit the worker's own timing, and it is a window, not a single number:

• Floor: the platform's SIGTERM→SIGKILL patience must exceed the shutdown phases or no release lands at all: at least cancellation_grace_period + cleanup_grace_period (40s at the defaults), and for a clean exit (not just the release write) worst_case_shutdown_seconds (82s at the defaults: the graces plus the bounded-close tail). Kubernetes terminationGracePeriodSeconds, Docker stop_grace_period, and the workgroup's supervisor.shutdown_grace (default 100s, sized from that 82s worst case plus margin, so a default workgroup now lets a child's held release and clean exit land; a startup warning names the computed numbers when a configured grace falls below the release floor (40s) or the clean-exit floor (82s)) are the three knobs that hold this end.
• Ceiling: matters only when watchdog_enabled=false: with no deadline trip to bound the process, the platform SIGKILL must land by termination_grace_period + exit tail (93s at the defaults) or a lingering executor thread can outlive a released row's hold. With the watchdog enabled there is no ceiling: the process polices its own exit, and a platform grace larger than the trip bound never engages.

The no-concurrent-run promise is scoped accordingly: it holds when the watchdog is enabled, or when the platform SIGKILL lands by termination_grace_period + exit tail.

For sync-actor fleets (plain def handlers, the only execution shape a deploy can never interrupt mid-body): give actors a cooperative stop by polling ctx.should_abort() in long loops so they exit inside the cancellation grace and get the immediate pending release; an actor that cannot check in still gets the held release, at the cost of the exit window's latency on every deploy it outlives, and its thread is ended at the deadline trip, by design, per the exit paragraph above. "Provably exited" is wrapper-level: threads or subprocesses the actor itself spawned are not proven gone (a subprocess outlives even os._exit); the guarantee is about TaskQ's own executor thread, not the actor's fan-out.

After RELEASING completes, shutdown_event is set and all TaskGroup siblings return.

Timing constraints (validated at startup):

cancellation_grace_period + cleanup_grace_period < termination_grace_period - 5.0
cancellation_grace_period + cleanup_grace_period < lock_lease

Defaults: cancellation_grace_period=30.0, cleanup_grace_period=10.0, termination_grace_period=85.0, lock_lease=60.0. The releasing hold additionally covers the watchdog's exit tail past the deadline (watchdog_dump_interval + ~2s flush + ~1s render slack ≈ 8s at the defaults): the process's own exit is enforced to that bound (see the exit paragraph above), so size the platform grace (terminationGracePeriodSeconds / stop_grace_period) from the same worst case, not from the hold. Two lease relationships are surfaced as startup warnings rather than validated: lock_lease below termination - cancellation - cleanup + heartbeat caps the release park short of the full budget (safe: the cap is what keeps the parked release write ahead of the lease reclaim, but interrupted sync actors release earlier with longer holds), and lock_lease below termination - cancellation - cleanup + heartbeat + exit tail (~63 vs 60 at the defaults) leaves a ~3s double-write-failure residue in which the reclaim sweep can take a disowned row before the deadline trip ends its actor.

---

Health server

When TASKQ_HEALTH_ENABLED=true (the default), the worker binds a Unix-domain socket at TASKQ_HEALTH_SOCKET_PATH and serves HTTP endpoints over it: /live, /ready, /metrics, and the opt-in /tasks stack-dump endpoint (see below).

A worker booted with no explicit path — no --health-socket-path, no TASKQ_HEALTH_SOCKET_PATH in the process environment or .env — binds a per-process default, /tmp/taskq_health_<pid>.sock, so two co-located workers cannot collide on one path (the second worker used to lose the bind and keep running with no socket). The setting's fallback value remains the shared /tmp/taskq_health.sock; it applies only when the variable IS set. Because the taskq health CLI runs in its own process and reads the same cascade, point it at an unconfigured worker explicitly: set TASKQ_HEALTH_SOCKET_PATH to the path the worker logs at boot (health-server-started's socket_path).

The Unix socket is not reachable via Kubernetes httpGet probes. Use exec probes:

livenessProbe:
  exec:
    command: ["taskq", "health", "live"]
  initialDelaySeconds: 5
  periodSeconds: 10
readinessProbe:
  exec:
    command: ["taskq", "health", "ready"]
  initialDelaySeconds: 5
  periodSeconds: 10

On platforms without exec probes (Azure Container Apps), or when an orchestrator or scraper should reach the worker over the network, set TASKQ_HEALTH_PORT to serve the same /live and /ready over TCP on TASKQ_HEALTH_HOST (default 0.0.0.0, the pod-network address container probes need). The TCP listener is off unless the port is set, answers /tasks with 404 (the socket's filesystem permissions do not apply to a TCP port), and carries no token check: keep it pod-network-only. The two transports bind independently: a TCP port that cannot be bound refuses the worker's startup (HealthTcpBindError: the probes target that port), while a unix-socket collision (a live peer owns the path) costs the Unix listener alone; the TCP listener is still bound, so port-routed probes keep answering (HealthUnixBindCollisionError, the same health-server-unavailable WARN as when no port is set). With no health_port configured at all, the collision warns and boots with no listener, as it always has. The Prometheus scrape listener (TASKQ_METRICS_PORT, host TASKQ_METRICS_HOST falling back to TASKQ_HEALTH_HOST) is a separate, unauthenticated surface documented in observability.md and refuses startup on a bind failure the same way. Full per-platform recipes: deployment.md: Listener deployment recipes.

Endpoints:

Path	Success condition	Success response	Failure response
GET /live	Event loop responsive within 1.0s	200 {"status":"ok"}	503 {"status":"unresponsive"}
GET /ready	shutdown_phase == NONE and PG ping succeeds within health_pg_ping_timeout (the per-slot transaction pool is pinged too, when it exists)	200 {"ready":true,...}	503 {"ready":false,...}
GET /metrics	Always	200 Prometheus text format	n/a

The /ready response body includes:

{
  "ready": true,
  "live": true,
  "reasons": [],
  "redis_configured": false,
  "active_jobs": 3,
  "is_leader": true,
  "maintenance": {
    "degraded": false,
    "reasons": ["no sweep has completed yet"]
  },
  "loop_tick_ages": {"heartbeat": 0.4, "producer": 0.1},
  "shutdown_elapsed_seconds": null,
  "shutdown_phase": null
}

maintenance is this process's view of its own sweep health: degraded: true when any sweep's last success is more than three sweep_intervals old or any sweep is running at the reduced batch tier (the batch-size breaker latched). It is deliberately not part of the 200/503 verdict: degraded means "up, but sweeps are unhealthy", an operator signal in the body, not a probe failure. "no sweep has completed yet" is the informational reason a process that has never completed a sweep (fresh, or never leader) reports instead of degraded. See maintenance-sweeps.md.

The /metrics response body:

# HELP taskq_active_jobs Currently in-flight jobs on this worker.
# TYPE taskq_active_jobs gauge
taskq_active_jobs 3
# HELP taskq_is_leader 1 if this worker holds the maintenance leader lock.
# TYPE taskq_is_leader gauge
taskq_is_leader 1
# HELP taskq_shutdown_phase Current shutdown phase enum value (0=NONE).
# TYPE taskq_shutdown_phase gauge
taskq_shutdown_phase 0

Probe these from a Kubernetes sidecar or taskq health live / taskq health ready. See cli.md for the CLI commands.

---

ActorConfig sync

At startup, after register_worker, the worker calls sync_actor_config for every registered actor. This writes (or updates) rows in {schema}.actor_config, and treats the row's fields as two different kinds of state:

• Capacity fields (max_concurrent, max_pending, result_ttl) are operator-owned. The @actor(...) literal only seeds the value the first time a row is created for that actor. On every subsequent startup, the stored value wins: a registered literal that differs from the stored value is expected (an operator tuned it), is never an error, and is logged at info level as actor-config-capacity-override. The UPSERT never writes these columns back on conflict, so nothing a worker does at startup can clobber an operator's change.
• The queue assignment is operator-owned the same way, moved by taskq actor-config move-queue, never by a boot. A registered literal that differs from the stored queue is the rolling-deploy window of a move (or a literal that has not followed the assignment yet): the worker logs actor-config-queue-override at WARNING and boots, and the UPSERT never writes the column back on conflict, so a stale-literal pod cannot undo the move. The warning is louder than the capacity one on purpose: the cron leader's fires follow the stored queue while producers enqueue by their own literal, so a persistent mismatch means the fleet's two routing halves disagree; reconcile with taskq actor-config move-queue or by deploying the matching literal.
• Metadata still guards against real bugs (no operator surface can move it, so any mismatch is one) and behaves exactly as before:
• force=False (default): a mismatch raises ActorConfigDriftList and the worker refuses to start. The CLI prints the drift details and instructs the operator to re-run with --force-update-actor-config.
• force=True: logs actor-config-drift-overwrite at ERROR for each drifted field and overwrites the stored value.

The sync uses a transactional SELECT-then-UPSERT to prevent races between concurrent worker startups.

Consequence: on an existing deployment, @actor(max_concurrent=...) is dead config

Operator-owned means the decorator literal is read only when the row is first created. Editing max_concurrent, max_pending or result_ttl in code and deploying it changes nothing on any deployment where the actor has already booted once: the rollout succeeds, the code review passes, and dispatch keeps using the stored value. A row seeded before the argument existed holds NULL, which for max_concurrent means uncapped, so a "we capped it" change can leave every actor running unbounded.

Nothing fails. The only signal is an info-level actor-config-capacity-override line per diverging actor at startup.

Reconcile deliberately:

taskq actor-config diff                                # what code says vs what is stored
taskq actor-config set my_actor --max-concurrent 4     # effective next dispatch cycle

taskq actor-config set is idempotent, so a deploy step or boot-time hook that applies your intended capacities is a safe way to make the code the source of truth. Verify with taskq actor-config list; an unexpected max_concurrent = NULL is the fingerprint of this trap.

ActorConfigDriftList wraps one ActorConfigDriftError per drifted structural field per actor. A single startup check can produce multiple ActorConfigDriftError instances, one for each combination of actor × structural field that differs.

Tuning capacity on a live deployment. Since a worker restart never changes a stored capacity value, use taskq actor-config set instead:

taskq actor-config set checkout_charge --max-concurrent 20
taskq actor-config set checkout_charge --max-pending 5000
taskq actor-config get checkout_charge
taskq actor-config list
taskq actor-config diff --actors myapp.actors:registry

All three capacity fields take effect without a worker restart, through three different mechanisms:

• max_concurrent: the dispatch query re-reads actor_config.max_concurrent on every dispatch cycle. Effective immediately.
• result_ttl: the terminal-write UPDATE re-reads actor_config.result_ttl on every job completion. Effective immediately for jobs completing after the change.
• max_pending: every enqueue-side process (web apps, workers' sub-job enqueues) holds a TTL-bounded in-process snapshot of the table (taskq/client/_capacity.py) and refreshes it at most once per TTL (default 5s). Effective fleet-wide within seconds. The cache fails open: if the refresh query fails, enqueue falls back to the last good snapshot (or the code literal) and retries no sooner than the TTL, so a sick database never turns into a per-enqueue failing query storm. A process that cannot wait out the TTL — ops tooling that just rewrote a stored cap and must admit against it on its very next enqueue — drops the snapshot immediately with JobsClient.invalidate_actor_capacity_cache(); see Client API.

Clearing a field (--clear-max-concurrent etc.) writes NULL. For max_concurrent that means unlimited: the dispatch SQL cannot see the code literal once a row exists. For max_pending and result_ttl NULL means revert to the @actor(...) literal; clearing undoes the operator override and restores the code default. The reverted result_ttl literal is supplied by the completing worker and applied from the job's completion timestamp: a job that sat in the queue longer than its TTL still gets its full result lifetime rather than completing already expired.

Upgrading note. Before this change, stored capacity always matched the last deployed literal (any drift blocked startup, or --force-update-actor-config rewrote it), so existing deployments upgrade in place: enforcement continues from the stored values. Do not run taskq actor-config set until every pod runs the new version; an old-version pod that restarts mid-rollout still crashes on capacity drift.

Moving an actor between queues. One operator action, two phases:

taskq actor-config move-queue mailer email_tier

The actor's pending/scheduled backlog is rewritten onto the target queue as bounded committed batches, then one final transaction locks the stored assignment, carries the old queue's queues row (mode + max_concurrent) to the target when the target has no row of its own, and flips the assignment, so old-queue strays drain through the target's consumers, and a crash mid-drain re-runs cleanly (the abort logs actor-queue-move-aborted with the count already committed). Running jobs finish on the workers that claimed them; any that re-pend instead (failure retry, crash reclaim, operator retry) keep their old queue label as an audit trail but are routed at dispatch by the actor's current assignment, so the running-job tail drains through the target queue's consumers; other actors on the old queue are untouched; cron fires follow the moved assignment from the flip on. Deploy the matching @actor(queue=...) literal before, during, or after the move, in any order, and keep workers consuming the old queue until every producer runs the new literal (stale producers keep enqueueing to it). The same rule governs a per-job enqueue(queue=...) override: it places the job's first attempt only, and any re-pend follows the actor's current assignment; see Queue overrides: first placement only.

The fields checked for structural drift are: metadata. Actor name changes require a migration; rename detection is not implemented.

---

Running multiple workers

Multiple worker processes against the same database are fully supported. Each process registers a separate row in {schema}.workers with its own UUID, hostname, and PID.

Dispatch safety. The dispatch CTE uses SELECT ... FOR UPDATE SKIP LOCKED, so two workers polling simultaneously cannot pick up the same job. Each job row is locked by exactly one worker at a time.

Single leader. Only one worker holds the taskq:maintenance_leader:<schema> advisory lock at a time. Other workers retry election on every heartbeat_interval tick. If the leader pod dies, the lock is released when the connection closes, and another worker wins the next election.

Rolling deploy gotcha. Old and new worker versions that declare different queue literals for the same actor name both boot: each logs actor-config-queue-override (the old side) or matches (the new side), and the stored assignment (the newest one, moved by taskq actor-config move-queue) stays authoritative throughout the rollout. Make sure workers consume both queues for the duration of the window and drop the old queue from TASKQ_QUEUES only after every producer runs the new literal and its backlog has drained.

A rolling deploy that changes metadata for the same actor name still trips the startup guard: new pods fail with ActorConfigDriftList. Best practice:

1. Deploy the first new pod with --force-update-actor-config (or TASKQ_FORCE_UPDATE_ACTOR_CONFIG=true). This overwrites the stored metadata and logs the change at ERROR with force=true.
2. Deploy all remaining pods without the flag. By the time they start, the stored config already matches the new registration, so no drift is detected.

Do not leave --force-update-actor-config set permanently. It allows any future metadata drift to be silently overwritten, removing the startup guard that protects against accidental mis-deploys.

A rolling deploy that only changes max_concurrent, max_pending, or result_ttl in the @actor(...) decorator needs none of this: those fields no longer participate in drift detection at all. Old and new pods can register different literals for the same actor simultaneously without either one failing to start; the stored row (whatever it currently is) stays authoritative throughout the rollout. Use taskq actor-config set if you actually want the new literal to take effect.

---

Workgroup supervisor

The workgroup supervisor (taskq workgroup start) manages multiple worker processes with per-worker configuration from a TOML file. It handles child health checks, crash restart, and graceful shutdown propagation.

Configuration

# workgroup.toml
actors = "myapp.actors:registry"

[defaults]
poll_interval = 1.0
max_concurrency = 4

[[workers]]
name = "api"
queues = ["default"]
max_concurrency = 8
poll_interval = 0.5

[workers.health]
enabled = true
check_interval = 15
stale_after = 60

[[workers]]
name = "media"
queues = ["media"]
max_concurrency = 2

Starting

taskq workgroup start workgroup.toml

Key behaviours

• Each child process is launched as taskq worker with per-worker CLI args derived from the TOML spec.
• The supervisor assigns a workgroup_instance UUIDv7 to correlate child workers in the workers table.
• Health checks (when health.enabled = true) poll the workers.last_seen_at column for the child's registered PID. Stale workers are SIGKILL'd.
• Crash restart with burst-limiting: the supervisor tracks restart counts and throttles rapid restarts.
• Graceful shutdown: sends SIGTERM to all children on supervisor shutdown. Children that don't exit within a timeout are SIGKILL'd.
• The supervisor label (worker_label) and workgroup instance (workgroup_instance) are stored in WorkerSettings and the workers table for cross-process correlation.

---

PgBouncer compatibility

The worker opens up to four asyncpg connection pools and two dedicated connections. Each targets a specific DSN for correctness reasons:

Connection	DSN used	Why
dispatcher_pool	pg_dsn_direct	Shares infrastructure with session-mode connections; direct connection avoids transaction-mode complications
heartbeat_pool	pg_dsn_direct	Same rationale as dispatcher_pool
worker_pool	pg_dsn_pooled	Terminal writes use short transactions; transaction-mode PgBouncer is safe here
slot pool (conditional)	pg_dsn_direct	Per-slot transaction connections: a transaction-mode pool would break every transaction boundary; opened only when a LOOP-scope asyncpg.Connection is registered and max_concurrency > 1
notify_conn	pg_dsn_direct	LISTEN state is session-scoped; a transaction-mode pool would drop the subscription between transactions
leader_conn	pg_dsn_direct	pg_try_advisory_lock produces a session-scoped lock; a transaction-mode pool would release the lock between transactions, allowing another worker to win the lock silently

Configuration:

TASKQ_PG_DSN_DIRECT=postgresql://user:pass@pg-primary:5432/mydb
TASKQ_PG_DSN_POOLED=postgresql://user:pass@pgbouncer:5432/mydb

When neither is set, both fall back to TASKQ_PG_DSN. When both DSNs are the same (no PgBouncer), the worker operates identically.

The dispatcher pool carries jit = off; leave it there

Every pool TaskQ builds for the dispatcher role is opened with server_settings={"jit": "off"}. Everything that pool runs: the dispatch claim statement, the leader sweeps, the admission lock acquires, is a short bounded OLTP round trip that LLVM compilation cannot help, and the dispatch statement's estimated cost at deep backlogs crosses Postgres's jit_above_cost, which would make Postgres compile the plan (~1 s of Optimization+Emission measured at a 30k-row due backlog, where the bounded scan itself takes ~2 ms) on top of the work itself. This is an operational guard: the statement's own estimate is fixed at the source, and the guard keeps any future estimate surprise from ever paying compile time on a hot dispatch loop. If you supply your own dispatcher pool (WorkerConnections.dispatcher_pool_factory), do not enable JIT on it for "more speed"; if a dispatch-path statement ever genuinely benefits from compilation, that is a measured statement-level change, not a pool knob.

If a LOOP-scope asyncpg.Connection provider is registered but the two DSNs differ, the worker emits a loop_scope_conn_dsn_mismatch warning at startup. PgBouncer in transaction mode breaks session semantics required by LOOP-scope connection providers. Either set both DSNs to the same direct endpoint for workers that use LOOP-scope connections, or omit the LOOP-scope connection provider and use the autonomous commit path.

---

Fork safety

A process that forks while TaskQ's connections are live - a job body calling os.fork(), a multiprocessing fork context, a prefork server such as gunicorn --preload that built its TaskQ in the master and forks its workers - inherits every socket TaskQ owns: the asyncpg pools' sockets, the NOTIFY LISTEN connection, the Redis clients' sockets. Two processes sharing one TCP stream corrupt both protocol state machines silently (a parent reading the server's answer to the CHILD's statement as the answer to its own, a terminal ledger writable by both processes). TaskQ installs an at-fork guard and refuses the inherited wire.

The fork contract: every wire resource TaskQ BUILDS records the process that created it. In a forked child, the first touch of one of those resources raises taskq._forkguard.ForkedInheritedProcessError - a typed refusal naming what was refused and what to do instead, BEFORE a byte reaches the shared socket. The parent keeps running; the worker's loops also consume an at-fork stamp and emit fork-detected-in-worker-process once per fork, so the log carries one honest signal that a fork happened and what it put at risk.

What is guarded (the resources TaskQ builds, with the guard riding the connection_class= TaskQ passes itself):

• every asyncpg pool and dedicated connection TaskQ opens - the query methods, and the COPY / custom-codec methods too (those reach asyncpg's protocol object directly and carry their own overrides);
• every redis.asyncio client TaskQ builds - plain commands (send_command) AND pipeline executes (redis-py's Pipeline packs all queued commands and writes them with one direct send_packed_command call, which is guarded separately);
• the worker's own loops (the parent-side report above).

What is NOT guarded:

• A pool or connection the APPLICATION built itself BEFORE TaskQ.open() / worker startup - plain asyncpg.create_pool() or redis.asyncio.from_url() called without TaskQ's guarded connection class - carries no guard, even in a process that also opened a TaskQ. The guard cannot see sockets it did not create.
• The OpenTelemetry exporters' background threads are not recoverable in a forked child (the child's copies of the exporter threads do not exist; its telemetry is lost or flushed through inherited pipes). Documented Python behavior; open telemetry after the fork instead.

Remediation:

• A child that needs its own TaskQ opens one AFTER the fork (fresh sockets, its own event loop).
• A child that only runs another program keeps subprocess.Popen's default close_fds=True, which drops the inherited descriptors.
• An application that pre-builds its own pools in a master process either moves that creation after the fork, or passes the guard explicitly: connection_class=taskq._forkguard.guarded_connection_class() for asyncpg (and guarded_redis_connection_class() for redis clients). A caller-supplied class of its own still wins everywhere TaskQ threads connection_class= - the guard never silently wraps a class the caller chose.

---

Until-idle mode

For CLI tools and batch processing scripts that enqueue a finite set of jobs and need to exit with a status code, pass --until-idle (or until_idle=True to worker_main()):

taskq worker --actors myapp.actors:registry --until-idle

exit_code = worker_main(settings, actor_registry=registry, until_idle=True)
# 0 = all jobs succeeded
# 3 = some jobs failed
# 4 = idle-max-runtime exceeded

The worker polls its subscribed queues every idle_poll_interval (default 1s). When no pending, scheduled, or running jobs remain for the settle window (idle_settle_window, default 2s), the worker triggers its normal graceful shutdown and exits.

An optional --idle-max-runtime cap (or TASKQ_IDLE_MAX_RUNTIME) forces exit with code 4 if the drain takes too long: useful for CI pipelines with time budgets.

Scheduled jobs: Jobs with a future scheduled_at count as "active": the worker waits for them to become due, dispatches them, and processes them before exiting. Use --idle-max-runtime if you don't want to wait for far-future scheduled jobs.

Cron: --until-idle is incompatible with cron-driven workloads, which enqueue jobs indefinitely. Do not combine --until-idle with @cron decorators. A startup WARNING is emitted when both until_idle and cron schedules are active.

Multi-worker: If multiple workers consume the same queue, --until-idle waits for the queue to drain across ALL workers, not just this one. This is correct for finite-batch scenarios but may cause unexpected waiting in shared-queue deployments.

idle_settle_window = 0: Setting the settle window to 0 disables the settle period, making drain detection near-instant (one poll interval). This is safe when the producer has finished enqueuing before the worker starts (typical for CI and batch workloads). For shared-queue or concurrent-producer scenarios, keep the default 2s settle window to handle the race where a producer enqueues between the drain monitor's check and the shutdown trigger.

Known limitations of drain_failures (worker-local counter):

The drain_failures counter tracks only jobs dispatched by this worker that returned "failed" from dispatch_one_job. It does NOT capture:

• Deadline-exceeded sweep failures: Jobs that reach failed via the maintenance leader's deadline sweep (pending/scheduled → failed without any consumer dispatch) are never seen by di_consumer_loop and do not increment drain_failures. Exit 0 is possible with deadline-exceeded failures in the batch.
• Other-worker failures: If multiple workers share the queue and another worker's consumer fails a job, this worker's drain_failures stays 0. Exit 0 is possible with failures on other workers.
• Jobs cancelled during shutdown: Jobs cancelled by the shutdown phase (CANCELLING → FORCING) propagate CancelledError out of dispatch_one_job without incrementing drain_failures; timeout-interrupted, not user-failed. This is correct: a cancelled job during drain is NOT a "failure".
• Actor-not-found jobs: A misconfigured job whose actor is not in this worker's registry is released via mark_snoozed(10s) and cycles scheduled→pending→claimed→snoozed indefinitely. Since scheduled counts as active, the queue never reads as drained, and the worker hangs until --idle-max-runtime is hit. Use --idle-max-runtime as the escape hatch.

For batch workflows requiring exact failure accounting, query the backend for terminal job statuses after the drain completes.

---

WorkerSettings reference

All variables use the TASKQ_ prefix. WorkerSettings extends TaskQSettings; variables from TaskQSettings are marked with a dagger (†).

Env var	Type	Default	Description
TASKQ_PG_DSN †	PostgresDsn	postgresql://taskq:taskq@localhost:5432/taskq	Direct DSN, used as fallback for both split DSNs
TASKQ_SCHEMA_NAME †	str	taskq	Postgres schema for all TaskQ tables
TASKQ_REDIS_URL †	RedisDsn \| None	None	Redis URL; required for rate-limiting and real-time progress
TASKQ_ENVIRONMENT †	str \| None	None	Deployment environment label
TASKQ_ADMIN_HOST †	str	0.0.0.0	Bind address for taskq ui serve
TASKQ_ADMIN_PORT †	int	8080	Bind port for taskq ui serve
TASKQ_ADMIN_URL †	str	http://localhost:8080	Public base URL of the admin UI
TASKQ_PG_DSN_DIRECT	PostgresDsn \| None	None (falls back to TASKQ_PG_DSN)	Direct Postgres DSN; bypasses PgBouncer
TASKQ_PG_DSN_POOLED	PostgresDsn \| None	None (falls back to TASKQ_PG_DSN)	Pooled DSN; may route through PgBouncer
TASKQ_DISPATCHER_POOL_SIZE	int	4	Max connections in dispatcher_pool
TASKQ_HEARTBEAT_POOL_SIZE	int	4	Max connections in heartbeat_pool
TASKQ_MAX_CONCURRENCY	int	8	Max concurrent jobs; worker_pool_size = int(max_concurrency * 1.5)
TASKQ_HEARTBEAT_INTERVAL	float	10.0	Seconds between heartbeat ticks
TASKQ_LOCK_LEASE	float	60.0	Seconds before a lock is reclaimed; must be >= max(heartbeat_interval, heartbeat_command_timeout) + (max_heartbeat_failures + 1) * (heartbeat_interval + heartbeat_command_timeout) (58 at the defaults)
TASKQ_MAX_HEARTBEAT_FAILURES	int	3	Consecutive heartbeat failures before isolate_self
TASKQ_TERMINATION_GRACE_PERIOD	float	85.0	Total seconds from SIGTERM to forced exit; sized to cover the default shutdown worst case (82s)
TASKQ_CANCELLATION_GRACE_PERIOD	float	30.0	Seconds for cooperative cancel phase
TASKQ_CLEANUP_GRACE_PERIOD	float	10.0	Seconds for force-cancel cleanup phase
TASKQ_MAX_RETRY_BACKOFF	timedelta	PT24H	Global ceiling on per-attempt retry backoff
TASKQ_DEFAULT_START_TO_CLOSE	timedelta \| None	None	Worker-wide fallback per-attempt execution timeout, applied only when neither the enqueue call nor the actor sets start_to_close. None = unbounded. See retries.md: start_to_close vs schedule_to_close
TASKQ_RATE_LIMIT_PG_FALLBACK_ENABLED	bool	True	Fall back to Postgres when Redis errors occur during rate limiting
TASKQ_HEALTH_ENABLED	bool	True	Enable the health server, both transports (Unix socket and the optional TCP listener)
TASKQ_HEALTH_SOCKET_PATH	str	/tmp/taskq_health.sock	Path for the health Unix socket (the field fallback; a worker booted with no explicit path binds the per-process default /tmp/taskq_health_<pid>.sock instead — see Health server)
TASKQ_HEALTH_PORT	int \| None	None	Optional TCP health listener (/live, /ready); off unless set, required on Azure Container Apps. See Health server
TASKQ_HEALTH_HOST	str	0.0.0.0	Bind host for both TCP listeners; pod-network default
TASKQ_HEALTH_TASKS_ENABLED	bool	False	Expose the privileged /tasks stack dump on the Unix socket (mode 0600); TCP answers 404
TASKQ_HEALTH_REQUEST_TIMEOUT	float	2.0	Seconds a probe gets to send its request line and headers before the connection is dropped
TASKQ_METRICS_PORT	int \| None	None	Optional Prometheus scrape listener; off unless set, needs taskq[prometheus]
TASKQ_METRICS_HOST	str \| None	None	Bind host override for the scrape listener alone; falls back to TASKQ_HEALTH_HOST
TASKQ_HEALTH_PG_PING_TIMEOUT	float	0.2	Seconds to wait for the readiness PG ping
TASKQ_POLL_INTERVAL	float	1.0	Fallback producer polling cadence (seconds) when NOTIFY is disabled
TASKQ_NOTIFY_ENABLED	bool	true	When true, the worker uses LISTEN/NOTIFY for near-zero-latency dispatch wakeups. When false, uses poll-only dispatch with poll_interval.
TASKQ_NOTIFY_POLL_INTERVAL	float	5.0	Fallback poll cadence when NOTIFY is enabled (rarely reached; NOTIFY handles the common case). Uses poll_interval when NOTIFY is disabled.
TASKQ_NOTIFY_HEALTH_CHECK_INTERVAL	float	5.0	How often the NOTIFY listener health-checks its connection
TASKQ_NOTIFY_RECONNECT_BACKOFF_INITIAL	float	1.0	Initial backoff before first NOTIFY reconnect attempt (doubles per attempt, capped at 30s; each delay carries ±25% multiplicative jitter so a fleet that loses PG simultaneously does not reconnect in lockstep)
TASKQ_QUEUES	list[str]	["default"]	Queue names this worker consumes; comma-separated
TASKQ_POOL_MAX_INACTIVE_LIFETIME	float	300.0	Seconds before an idle pool connection is closed
TASKQ_WORKER_LABEL	str \| None	None	Human-readable label for this worker, stored in workers.worker_label
TASKQ_WORKGROUP_INSTANCE	str \| None	None	UUIDv7 identifying the workgroup orchestrator that launched this worker
TASKQ_OTEL_ENABLED	bool	True	Enable OTel span and metric creation
TASKQ_WORKER_GROUP	str	default	Consumer group name on CONSUMER spans
TASKQ_LOG_FORMAT	str	json	json or console
TASKQ_LOG_LEVEL	str	INFO	Root logger level
TASKQ_FORCE_UPDATE_ACTOR_CONFIG	bool	False	Overwrite metadata-drifted actor-config rows without raising (the queue assignment and capacity fields are never boot-rewritten); see ActorConfig sync
TASKQ_PRUNE_SCHEDULE_UTC	str	03:00	Daily fire time for the prune sweep (Sweep 5) in HH:MM UTC. Ignored when TASKQ_PRUNE_CRON_EXPR is set.
TASKQ_PRUNE_CRON_EXPR	str \| None	None	Full 5-field cron for the prune sweep; overrides TASKQ_PRUNE_SCHEDULE_UTC.
TASKQ_PRUNE_BATCH_SIZE	int	10000	Rows per prune CTE batch.
TASKQ_PRUNE_RETENTION_PERIOD	timedelta	30d	Global fallback retention in jobs before archival.
TASKQ_PRUNE_RETENTION_SUCCEEDED	timedelta	30d	Per-status retention for succeeded jobs.
TASKQ_PRUNE_RETENTION_FAILED	timedelta	90d	Per-status retention for failed jobs.
TASKQ_PRUNE_RETENTION_CANCELLED	timedelta	30d	Per-status retention for cancelled jobs.
TASKQ_PRUNE_RETENTION_ABANDONED	timedelta	90d	Per-status retention for abandoned and crashed jobs.
TASKQ_ARCHIVE_RETENTION_PERIOD	timedelta	365d	How long a row stays in jobs_archive before hard-deletion by Sweep 6.
TASKQ_ARCHIVE_EXPIRY_SCHEDULE_UTC	str	04:00	Daily fire time for the archive expiry sweep (Sweep 6) in HH:MM UTC.
TASKQ_ARCHIVE_EXPIRY_CRON_EXPR	str \| None	None	Full 5-field cron for the archive expiry sweep; overrides TASKQ_ARCHIVE_EXPIRY_SCHEDULE_UTC.
TASKQ_IDLE_SETTLE_WINDOW	float	2.0	Seconds the drain monitor waits after queues appear empty before declaring drained. Only used with --until-idle.
TASKQ_IDLE_POLL_INTERVAL	float	1.0	How often the drain monitor checks queue depth. Only used with --until-idle.
TASKQ_IDLE_MAX_RUNTIME	float \| None	None	Maximum wall-clock seconds for until-idle mode. When exceeded, exit code 4. None = no limit. Only used with --until-idle.