Skip to content

Interactive Visual Architecture

This page provides an interactive visual breakdown of the six core architectural subsystems in litestar-queues. Use the tab switchers inside each diagram to inspect state transitions, wire contracts, fencing invariants, and deployment topologies.

Decoupled Storage vs ID-Only Execution Transports

BaseQueueBackend is the sole authority for task state (QueuedTaskRecord), uniqueness locks (TaskReservation), and results. External execution transports (CloudTasksExecutionBackend, CloudRunExecutionBackend, and message broker backends) carry only {"version": 1, "task_id": "<uuid>"} on the wire.

Decoupled Task Storage vs Execution Transport QueueService.enqueue writes the authoritative QueuedTaskRecord to BaseQueueBackend while BaseExecutionBackend carries only an ID-only version 1 task_id envelope. Litestar Route / Caller QueueService.enqueue() @task.enqueue(...) / enqueue_many() Validates TaskDefinition & options Computes unique_key & delay QueuedBackgroundTask 1. Persist Record 2. Dispatch ID Authoritative Task Storage BaseQueueBackend (QueuedTaskRecord) Arguments • Status • Leases • Retries • Results SQLSpec (11 DBs) • Advanced Alchemy • Redis • Valkey ID-Only Execution Transport {"version": 1, "task_id": "<uuid>"} Never carries task arguments or retry state on wire Local • Cloud Tasks • Cloud Run • Pub/Sub • SQS • Kafka Worker / Push Target claim_task(task_id) Claims authoritative record Runs WorkerHeartbeatManager Resolves DI & TaskExecutionContext complete_task / fail_task
Epoch-Fenced Task Claim, Heartbeat & Settlement Every heartbeat and terminal transition validates status='running', expected_retry_count, and expected_execution_ref to prevent zombie workers from overwriting reclaimed tasks. 1. Atomic Claim dequeue_tasks / claim_task Transitions pending/scheduled -> running Captures retry_count & execution_ref Stamps started_at & heartbeat_at FOR UPDATE SKIP LOCKED / Lua 2. Fenced Heartbeat Loop heartbeat_tasks(..., expected_*) Requires status == "running" Requires retry_count == expected Requires execution_ref == expected Lease lost -> cancel execution 3. Fenced Settlement complete_task / fail_task Conditional CAS on epoch + lease Persists result or schedules retry Updates TaskReservation uniqueness Emits QueueEvent via publisher
Invariant Authoritative Component Enforcement Mechanism
Single Source of Truth BaseQueueBackend All task arguments, status transitions, retry counters, and results live exclusively in storage.
ID-Only Wire Payload BaseExecutionBackend Wire payload is strictly {"version": 1, "task_id": "<uuid>"} with zero task arguments or secrets.
Epoch & Lease Fencing heartbeat_tasks(), complete_task(), fail_task() Every mutation checks status == "running", retry_count == expected_retry_count, and execution_ref == expected_execution_ref.

Worker Placement Topologies

Switch between the four worker execution topologies to see how WorkerConfig.placement adapts from single-command local development to horizontally scaled worker fleets and serverless scale-to-zero deployments.

Server Placement Topology (placement="server") Litestar CLI launches ServerWorkerSupervisor alongside web workers, sharing EphemeralQueueBackend or persistent backends across OS processes. Litestar CLI Parent ServerWorkerSupervisor Spawns alongside `litestar run` Propagates Ephemeral SQLite path Isolates web workers from queue loop Shared Queue Storage Ephemeral / SQL / Redis Default: EphemeralQueueBackend Cross-process WAL SQLite + events Requires shared backend if web > 1 Supervised Worker Child litestar queues run Single scheduler leader Auto-restarts on unexpected exit Clean SIGTERM drain on shutdown
ASGI Placement Topology (placement="asgi") Each ASGI lifespan starts an in-process Worker sharing the asyncio event loop and InMemoryQueueBackend or persistent storage. ASGI Lifespan Startup QueuePlugin.on_cli_init Starts inside ASGI lifespan Works with TestClient / Uvicorn Zero extra OS processes Shared Event Loop InMemoryQueueBackend Instant in-memory asyncio.Event wakeup Direct ChannelsPlugin memory fan-out Single-process development & tests In-Process Worker Task Worker.start() Executes coroutines & sync threads Shares app.state & DI container Graceful cancel on lifespan exit
External Placement Topology (placement="external") Web pods only enqueue tasks and serve SSE/WS streams while dedicated worker pods run litestar queues run or run-consumer with run_app_lifespan=True. Web Tier (Stateless Pods) placement="external" Zero background workers in web pods Enqueues records & serves HTTP/SSE Horizontal web autoscaling Durable Storage & Broker SQLSpec / Alchemy / Redis Native wakeups (LISTEN/NOTIFY, Pub/Sub) Or broker transport (Pub/Sub, SQS, Kafka) Cross-process QueueEventLog replay Dedicated Worker Fleet litestar queues run Enters ASGI lifespan (`run_app_lifespan`) Or `litestar queues run-consumer` Independent CPU/memory scaling
Serverless Scale-to-Zero Placement Zero idle workers; enqueue dispatches directly to Cloud Tasks or Cloud Run Jobs while Cloud Scheduler triggers /_litestar-queues/maintenance. Scale-to-Zero Web Service schedules_on_enqueue=True Dispatches future ETA at enqueue time Mounts `/_litestar-queues/cloud-tasks` Mounts `/_litestar-queues/maintenance` Managed Cloud Dispatch Cloud Tasks + Cloud Run Fast Push Lane (<= 30m HTTP OIDC) Heavy Batch Lane (Cloud Run Jobs) Task-level `execution_backend="..."` Periodic Maintenance QueueMaintenanceService Cloud Scheduler OIDC POST or Job Reconciles external refs & stale leases Fires cron schedules & prunes TTL
Placement Mode Configuration Process Boundary Best For
Server Supervised WorkerConfig(placement="server") Child OS process managed by ServerWorkerSupervisor during litestar run Local development and single-node deployments.
In-Process ASGI WorkerConfig(placement="asgi") Coroutine inside the web server's asyncio event loop Unit tests, TestClient, and lightweight single-process apps.
Dedicated External WorkerConfig(placement="external") Standalone litestar queues run or run-consumer pods Multi-replica production web tiers with independent worker scaling.
Serverless Scale-to-Zero placement="external" + Cloud Tasks / Cloud Run On-demand OIDC webhooks or one-shot Cloud Run Job containers Workloads that scale to zero compute cost when queues are idle.

Task State Machine, Clocks & Uniqueness Windows

Explore how tasks transition from pending or scheduled through fenced running leases into terminal states, how the three independent clocks operate without interfering, and how unique_until windows prevent duplicate work.

Task State Machine & Retry Budget Accounting Tasks transition from pending or scheduled to running, and settle into completed, failed, cancelled, or expired while separating retry_count from infrastructure interruptions. pending scheduled_at == None Ready for immediate claim scheduled scheduled_at > now Delayed or backoff retry running (Fenced Lease) attempts = retry_count - interruptions Active heartbeat_at lease refresh Exception & attempts < max_retries -> scheduled Stale lease reclaim -> retry_count++, interruptions++ completed Result stored • completed_at set failed Retries exhausted • error recorded cancelled cancel_task() • cooperative abort expired expires_at passed before claim
Independent Task Clocks Comparison of expires_in/expires_at (queued deadline), timeout (per-attempt execution ceiling), and terminal_ttl/terminal_retention (post-settlement storage cleanup). Queue Expiration Clock expires_in / expires_at Ticks while pending or scheduled Stops the moment a worker claims task Never interrupts a running task Outcome: status = "expired" Execution Timeout Clock timeout=30.0 | timedelta(...) Ticks per execution attempt (`started_at`) Enforced via asyncio deadline scope Consumes retry budget if max_retries > 0 Outcome: retry or status = "failed" Terminal Retention Clock terminal_ttl / terminal_retention Ticks after completed/failed/cancelled/expired Native Redis/Valkey PEXPIRE (`terminal_ttl`) Maintenance Phase (`cleanup_terminal`) Preserves unique_until="forever"
Uniqueness Windows & Permanent TaskReservation Comparison of unique_until='terminal', duplicate policies ('return_existing' vs 'raise'), and unique_until='forever' backed by TaskReservation. In-Flight Deduplication unique_until="terminal" Active while pending, scheduled, or running Duplicate enqueue returns existing task_id Releases lock as soon as task settles Ideal for coalescing webhook bursts Duplicate Conflict Policies on_duplicate="return_existing" | "raise" "return_existing" returns active TaskResult "raise" raises DuplicateTaskError Effective key hashed from args or unique_key Inspect via get_task_identity() Permanent Reservation unique_until="forever" Backed by compact TaskReservation record Survives terminal_ttl / cleanup_terminal Failed/cancelled/expired releases lock to retry Reset via reset_task_identity()
Clock / Mechanism Configured On When It Ticks Terminal Outcome
Queue Expiration (expires_in / expires_at) @task or enqueue() Only while status is pending or scheduled Transitions unstarted task to status="expired". Never interrupts running tasks.
Execution Timeout (timeout) @task or enqueue() Per execution attempt while status == "running" Cancels attempt and either schedules a retry or transitions to status="failed".
Terminal Retention (terminal_ttl / terminal_retention) *BackendConfig or QueueMaintenanceConfig After task reaches completed, failed, cancelled, or expired Deletes aged terminal task records while preserving unique_until="forever" reservations.

Serverless Scale-to-Zero & Maintenance

In serverless topologies, QueueConfig can register a default push lane (CloudTasksExecutionConfig for interactive tasks up to 30 minutes) alongside named batch lanes (execution_backends=(CloudRunExecutionConfig(...),) for multi-hour Cloud Run Jobs), backed by QueueMaintenanceService.

Serverless Scale-to-Zero Dual-Lane Dispatch QueueConfig routes interactive tasks (<= 30m) to CloudTasksExecutionConfig push webhooks and heavy batch tasks to CloudRunExecutionConfig one-shot Job containers. Multi-Backend Router QueueConfig execution_backend = CloudTasks execution_backends = (CloudRun,) @task(execution_backend="cloudrun") schedules_on_enqueue=True Fast Push Lane (<= 30 min) CloudTasksExecutionBackend Dispatches OIDC-signed HTTP POST Schedule Horizon up to 30 days ahead Heavy Batch Lane (Multi-Hour) CloudRunExecutionBackend Calls JobsClient.run_job(overrides) Injects QUEUES_TASK_ID env var Cloud Run Service Webhook /_litestar-queues/cloud-tasks CloudTasksOIDCGuard + claim_task Active lease -> 503; stale -> reclaim Cloud Run Job Execution litestar queues run-task Dedicated CPU/RAM container Exits 0 on complete, 1 on retry/fail
Six-Phase QueueMaintenanceService Sweep Ordered execution of expire, external, stale, schedules, terminal, and events phases via /_litestar-queues/maintenance or litestar queues run-maintenance. 1. expire Expire queued tasks past expires_at expire_pending 2. external Reconcile Cloud Run / Tasks refs reconcile_task 3. stale Reclaim stale running leases & redispatch requeue_stale 4. schedules Claim due cron/interval & enqueue claim_due 5. terminal Delete old terminal rows past retention cleanup_tasks 6. events Prune expired QueueEventLog history rows cleanup_events

Realtime Event Pipeline & Cursor Replay

Running tasks emit structured progress, log, and custom events through TaskExecutionContext. When EventHistoryConfig is configured on QueueEventsConfig.history, QueueEventPublisher commits durable QueueEventLogRecord rows before broadcasting live frames to ChannelsQueueEventSink, guaranteeing gap-free cursor replay across reconnects and planned max_stream_duration rollovers.

Dual-Path Task Event Publication Pipeline TaskExecutionContext and Worker publish QueueEvents to QueueEventPublisher, which persists durable history before broadcasting to ChannelsQueueEventSink SSE/WebSocket streams. Task & Worker Lifecycle TaskExecutionContext ctx.progress(percent, stage=...) ctx.log(message, level="info") ctx.event(event_type, data=...) enqueued • started • completed Event Coordinator QueueEventPublisher Commit-before-live ordering Assigns monotonic sequence ID delivery="background" | "strict" Path A Path B Durable History (QueueEventLog) EventHistoryConfig(batch_size=20) Batched record_events_batch() commits Query via query_events() & summarize_stages() Live Browser Fan-Out ChannelsQueueEventSink -> /queues/events WS `/queues/events` & SSE `/queues/events/sse` Queue / Task / Global topic channels
Gap-Free Cursor Replay & Planned Stream Rollover SSE and WebSocket subscribers replay missed history from Last-Event-ID or after_sequence, deduplicate live frames, and reconnect seamlessly on max_stream_duration. Subscribe & Replay Last-Event-ID / after_sequence Subscribes to live Channels topic first Replays QueueEventLog(after_sequence=N) Deduplicates live frames <= replayed seq Zero missed events on reconnect Planned Stream Rollover max_stream_duration=1500.0 Closes cleanly before Cloud Run timeout Emits `retry: <retry_ms>` SSE hint WebSocket close code 1001 (`stream_rollover`) Browser EventSource auto-resumes Shielded Teardown CancelScope(shield=True) Shields subscriber cleanup on disconnect Prevents leaked Channels subscriptions Flushes pending EventBuffer on shutdown Safe under high client churn

Transport Layer Matrix

litestar-queues separates storage, worker wakeup hints, execution dispatch, and browser event streaming into orthogonal layers that can be mixed and matched independently:

Transport Layers in Litestar Queues Architectural boundary between Authoritative Task Storage, Worker Wakeup Hints, Execution Dispatch Transports, and Browser Event Fan-Out. Task Storage BaseQueueBackend Sole source of truth for QueuedTaskRecord state Backends: • SQLSpec (11 dialects) • Advanced Alchemy • Redis 8.1+ / Valkey 6+ • Ephemeral / InMemory Worker Wakeups wakeup_transport Low-latency hint to wake idle polling workers Transports: • notify_queue / notify (PG) • aq / txeventq (Oracle) • poll_queue (SQLite/MySQL) • Redis/Valkey Pub/Sub Execution Dispatch BaseExecutionBackend Carries ID-only envelope {"version": 1, "task_id"} Backends: • Local / Immediate • Cloud Tasks (HTTP OIDC) • Cloud Run Jobs • Pub/Sub • SQS • RMQ • Kafka Browser Fan-Out QueueEventsConfig Streams live QueueEvents & replays QueueEventLog Delivery: • ChannelsQueueEventSink • /queues/events/sse (SSE) • /queues/events (WS) • QueueEventLog replay