Durable Event History¶

Configuring EventHistoryConfig on QueueEventsConfig.history enables backend-owned durable event persistence (QueueEventLog), paginated history queries, per-stage duration summaries, and commit-before-live delivery coordination.

Runnable Example¶

"""Persisting and querying durable event history with stage summaries and actor filters."""

from litestar_queues import QueueConfig, QueueService, TaskExecutionContext, WorkerConfig, task
from litestar_queues.events import (
    EventHistoryConfig,
    QueueEventActor,
    QueueEventEntityRef,
    QueueEventQuery,
    QueueEventsConfig,
)

EXPECTED_STAGE_COUNT = 2


@task("etl.import_catalog", queue="etl", actor=QueueEventActor(type="service", id="catalog-sync"))
async def import_catalog(catalog_id: str, _task_context: TaskExecutionContext) -> str:
    """Emit staged progress events with entity and actor attribution."""
    _task_context.entity = QueueEventEntityRef(type="catalog", id=catalog_id)
    _task_context.scope_key = "tenant:acme"
    await _task_context.progress(
        current=1, total=2, message="Extracted feed", payload={"stage": "extract", "duration_ms": 42.5}
    )
    await _task_context.progress(
        current=2, total=2, message="Loaded catalog rows", payload={"stage": "load", "duration_ms": 85.0}
    )
    return f"imported:{catalog_id}"


async def main() -> None:
    """Verify durable event history queries and per-stage aggregation."""
    config = QueueConfig(
        queue_backend="memory",
        execution_backend="immediate",
        worker=WorkerConfig(placement="external"),
        events=QueueEventsConfig(history=EventHistoryConfig(batch_size=10, flush_interval=0.1)),
    )
    async with QueueService(config=config) as service:
        handle = await service.enqueue(import_catalog, "cat-900")
        event_log = service.get_event_log()
        assert event_log is not None
        await event_log.flush_events()

        page = await event_log.query_events(
            QueueEventQuery(task_id=str(handle.id), order="asc"), extra={"actor_id": "catalog-sync"}
        )
        assert page.total >= EXPECTED_STAGE_COUNT

        summaries = await event_log.summarize_stages(QueueEventQuery(task_id=str(handle.id)))
        stages = {summary.stage for summary in summaries if summary.stage is not None}
        assert stages == {"extract", "load"}

Commit-Before-Live Guarantee¶

When both QueueEventsConfig.history and QueueEventsConfig.delivery (or stream) are enabled, QueueEventPublisher routes events through event_log.publish_event_after_commit(...):

1. Buffered Batch Persistence: Events enter the history buffer (batch_size=20, flush_interval=1.0s, max_pending=2000) and are written to the storage backend in batches.
2. Terminal Lifecycle Barrier (barrier=True): Terminal lifecycle events (task.completed, task.failed, task.cancelled, task.expired, task.stale_failed) force an immediate history flush and hold live ChannelsPlugin release until the database commit succeeds.
3. Zero Read-After-Notification Races: When a browser client receives a task.completed frame over SSE or WebSocket and immediately fetches the task's event history or stage summary via HTTP, the complete event trail is guaranteed to be committed and queryable in storage.

Storage Backend Support Matrix¶

Querying Event History (QueueEventQuery)¶

Retrieve the active QueueEventLog via queue_service.get_event_log() and call await event_log.query_events(query, extra=...):

In addition to QueueEventQuery fields, pass extra={"actor_id": "...", "actor_type": "..."} (or any custom EventHistoryExtraColumn name) to query_events() to filter by actor attribution or adopter-defined columns. query_events() returns an OffsetPagination[QueueEventLogRecord] with .items, .total, .limit, and .offset.

Per-Stage Aggregation (summarize_stages)¶

When tasks include "stage" (and optional "duration_ms") keys inside payload, call await event_log.summarize_stages(QueueEventQuery(task_id=...)) to aggregate matching records into QueueEventStageSummary objects:

• stage: Stage name string (or None for unstaged events)
• event_count: Total events recorded in that stage
• total_duration_ms: Sum of duration_ms across events in that stage
• first_event_at / last_event_at: Earliest and latest occurred_at UTC timestamps
• latest_sequence / latest_message: Sequence number and message from the newest event in the stage
• worst_level: Highest-severity level observed in the stage (critical > error > warning > info > debug)

Retention Rules & Maintenance Cleanup¶

Configure ordered QueueEventRetentionRule entries on QueueMaintenanceConfig.event_retention_rules to prune old event history during the "events" maintenance phase:

from litestar_queues import QueueMaintenanceConfig
from litestar_queues.events import QueueEventQuery, QueueEventRetentionRule

maintenance = QueueMaintenanceConfig(
    event_retention_rules=(
        QueueEventRetentionRule(max_age=86_400.0, match=QueueEventQuery(event_type="task.progress")),
        QueueEventRetentionRule(max_age=604_800.0, match=QueueEventQuery()),
    ),
    event_limit=1000,
)

Rules are evaluated in tuple order; the first rule whose match query matches a record governs its retention window (max_age in seconds). Records matching no rule are retained indefinitely.