Queue Events Overview¶

litestar-queues decouples task execution state (QueuedTaskRecord) from realtime event publishing (QueueEventsConfig). Tasks and workers emit typed QueueEvent envelopes that can simultaneously fan out over Litestar ChannelsPlugin SSE/WebSocket streams and persist into backend-owned durable event history (QueueEventLog).

Dual-Path Commit-Before-Live Pipeline Cursor Replay & Planned Stream Rollover

Runnable Example¶

docs/examples/events/publish_from_task.py

"""Publishing progress, logs, custom events, and heartbeat details from a task."""

from litestar_queues import QueueConfig, QueueService, TaskExecutionContext, WorkerConfig, task
from litestar_queues.events import (
    EventDeliveryConfig,
    InMemoryQueueEventSink,
    QueueEventsConfig,
    publish_task_event,
    publish_task_log,
    publish_task_progress,
)

EXPECTED_CUSTOM_EVENTS = 3


async def emit_helper_signals() -> None:
    """Publish progress and logs via module-level helpers that safely no-op when unbound."""
    await publish_task_progress(current=3, total=6, message="3/6 pages", payload={"page": 3})
    await publish_task_log("Checkpoint reached", level="info", payload={"step": "checkpoint"})
    await publish_task_event("crawl.page_discovered", payload={"url": "https://litestar.dev"}, immediate=False)


@task("crawler.index_site", queue="crawler")
async def index_site(domain: str, _task_context: TaskExecutionContext) -> dict[str, str]:
    """Execute a task that receives TaskExecutionContext via _task_context injection."""
    _task_context.beat(f"starting:{domain}")
    await _task_context.progress(current=1, total=6, message="1/6 pages", payload={"page": 1})
    await emit_helper_signals()
    await _task_context.progress(current=6, total=6, message="6/6 pages", payload={"page": 6})
    return {"domain": domain, "status": "indexed"}


async def main() -> None:
    """Verify _task_context injection and module-level helper no-op behavior outside a task."""
    await emit_helper_signals()

    sink = InMemoryQueueEventSink()
    config = QueueConfig(
        queue_backend="memory",
        execution_backend="immediate",
        worker=WorkerConfig(placement="external"),
        events=QueueEventsConfig(delivery=EventDeliveryConfig(sinks=(sink,), buffer=None)),
    )
    async with QueueService(config=config) as service:
        handle = await service.enqueue(index_site, "litestar.dev")
        await handle.refresh()
        assert handle.status == "completed"
        progress_events = [e for e in sink.events if e.type == "task.progress"]
        assert len(progress_events) == EXPECTED_CUSTOM_EVENTS
        discovered = [e for e in sink.events if e.type == "crawl.page_discovered"]
        assert len(discovered) == 1

Execution Signals¶

Signal	Emitter API	Wire Event Type	Purpose
Heartbeat	Automatic `WorkerHeartbeatManager` + optional `ctx.beat(detail)`	Storage `heartbeat_at` + metadata	The worker updates timestamps automatically to extend the running task's lease fence in storage; `ctx.beat(detail)` records last-value-wins diagnostic text on the next heartbeat write, does not update progress or terminal task state, and should not also be copied into a generic custom event
Progress	`await ctx.progress(...)` / `await publish_task_progress(...)`	`"task.progress"`	Emits structured progress (`current`, `total`, `percent`, `message`, `payload`) to live subscribers and durable history
Log	`await ctx.log(...)` / `await publish_task_log(...)`	`"task.log"`	Emits task-scoped structured log entries (`level`, `message`, `payload`)
Custom event	`await ctx.event(name, ...)` / `await publish_task_event(name, ...)`	Caller-defined (e.g. `"crawl.page_discovered"`)	Emits domain-specific milestone events during task execution
Lifecycle event	Worker / `QueueService` runtime	`"task.started"`, `"task.completed"`, `"task.failed"`, `"task.retried"`, `"task.cancelled"`, `"task.expired"`, `"task.interrupted"`, `"task.claim_lost"`, `"task.stale_failed"`	Authoritative state-transition events emitted automatically by `QueueService` after storage commits (including `will_retry` metadata on failure transitions and claim loss / stale failure transitions)

Publishing from Inside a Task¶

Automatic Context Injection¶

Declare a parameter named _task_context: TaskExecutionContext on any @task function to receive the attempt's bound TaskExecutionContext:

from litestar_queues import TaskExecutionContext, task


@task("crawler.crawl_domain", queue="crawler")
async def crawl_domain(domain: str, _task_context: TaskExecutionContext) -> dict[str, str]:
    ctx = _task_context
    ctx.beat("fetching-sitemap")
    await ctx.progress(
        current=3,
        total=6,
        message="3/6 pages",
        payload={"page": 3},
    )
    await ctx.event(
        "crawl.page_discovered",
        payload={"url": f"https://{domain}/docs"},
        immediate=False,
    )
    return {"domain": domain, "status": "complete"}

Module-Level ContextVar Helpers¶

When helper functions called deep inside a task do not pass _task_context down the call stack, use the module-level helpers from litestar_queues.events:

await publish_task_progress(current=..., total=..., message=..., payload=..., strict=False)
await publish_task_log(message, level="info", payload=..., strict=False)
await publish_task_event(event_type, message=..., payload=..., immediate=False, strict=False)
beat(detail)

By default (strict=False), calling these helpers outside an active task execution (for example when unit testing a helper function directly) safely no-ops and returns None. Pass strict=True (or call require_current_task_context()) to raise RuntimeError if no TaskExecutionContext is bound.

Live Delivery vs Durable Event History¶

QueueEventsConfig composes three independent capabilities:

Live Delivery (delivery=EventDeliveryConfig(...)): Buffers events in an in-memory ring buffer (EventBufferConfig) and flushes them to Litestar ChannelsPlugin (ChannelsQueueEventSink) and any custom sinks. Setting EventDeliveryConfig(strict=True) propagates sink delivery exceptions to the caller instead of logging and isolating them.
Browser Stream Routes (stream=EventStreamConfig(...)): Mounts ready-to-use SSE and WebSocket endpoints under /queues/events (see Realtime SSE & WebSocket Streams).
Durable Event History (history=EventHistoryConfig(...)): Persists events into the storage backend's QueueEventLog (queue_task_event_history in SQLSpec/Advanced Alchemy or Redis/Valkey Streams + Hashes) and enforces a commit-before-live barrier so browser subscribers never receive a lifecycle event before its history record is durably committed (see Durable Event History).