Skip to content

Extending Event History

You can extend durable event history in two ways:

  1. Declarative Extra Columns (EventHistoryExtraColumn): Extract domain keys (such as tenant_id, organization_id, or workflow_id) from QueueEvent.payload into first-class, indexed SQL columns on queue_task_event_history.
  2. Custom QueueEventLog Implementation: Implement the QueueEventLog protocol to store and query event history in an external analytics store (such as ClickHouse, BigQuery, or OpenSearch).

Declarative Extra Columns (EventHistoryExtraColumn)

docs/examples/events/custom_event_log.py
"""Configuring EventHistoryExtraColumn for indexed tenant-scoped event history queries."""

from pathlib import Path
from tempfile import TemporaryDirectory

from sqlspec.adapters.aiosqlite import AiosqliteConfig

from litestar_queues import QueueConfig, QueueService, TaskExecutionContext, WorkerConfig, task
from litestar_queues.backends.sqlspec import SQLSpecBackendConfig, SQLSpecQueueBackend, SQLSpecWorkerWakeupConfig
from litestar_queues.events import EventHistoryConfig, EventHistoryExtraColumn, QueueEventQuery, QueueEventsConfig


@task("billing.sync_tenant_ledger", queue="billing")
async def sync_tenant_ledger(tenant_id: str, _task_context: TaskExecutionContext) -> str:
    """Publish an event carrying tenant_id in the payload for extra-column extraction."""
    await _task_context.event(
        "ledger.synced", message="Synchronized tenant ledger", payload={"tenant_id": tenant_id, "entries": 42}
    )
    return f"synced:{tenant_id}"


async def main() -> None:
    """Verify EventHistoryExtraColumn extraction and extra filter querying on SQLSpec."""
    extra_columns = (EventHistoryExtraColumn(name="tenant_id", source="tenant_id", indexed=True),)
    with TemporaryDirectory() as tmp_dir:
        db_path = Path(tmp_dir) / "history_extra.sqlite3"
        backend_config = SQLSpecBackendConfig(
            sqlspec_config=AiosqliteConfig(connection_config={"database": str(db_path)}),
            event_history_extra_columns=extra_columns,
            worker_wakeups=SQLSpecWorkerWakeupConfig(transport="polling"),
        )
        config = QueueConfig(
            queue_backend=backend_config,
            execution_backend="immediate",
            worker=WorkerConfig(placement="external"),
            events=QueueEventsConfig(
                history=EventHistoryConfig(extra_columns=extra_columns, batch_size=5, flush_interval=0.1)
            ),
        )
        async with QueueService(config=config) as service:
            backend = service.get_queue_backend()
            assert isinstance(backend, SQLSpecQueueBackend)
            await backend.create_schema()

            await service.enqueue(sync_tenant_ledger, "tenant-alpha")
            await service.enqueue(sync_tenant_ledger, "tenant-beta")

            event_log = service.get_event_log()
            assert event_log is not None
            await event_log.flush_events()

            alpha_page = await event_log.query_events(
                QueueEventQuery(event_type="ledger.synced"), extra={"tenant_id": "tenant-alpha"}
            )
            assert alpha_page.total == 1
            assert alpha_page.items[0].extra["tenant_id"] == "tenant-alpha"

Column Validation Rules

When declaring EventHistoryExtraColumn(name=..., source=..., indexed=...) on EventHistoryConfig.extra_columns and SQLSpecBackendConfig.event_history_extra_columns:

  • Valid Unquoted SQL Identifiers: name must be a non-empty ASCII identifier starting with a letter or underscore and containing only [A-Za-z0-9_].
  • No Package Column Collisions: name (compared case-insensitively) cannot redeclare any of the 21 built-in queue_task_event_history columns (event_id, event_type, task_id, task_name, queue, worker_id, execution_backend, execution_profile, actor_type, actor_id, stage, level, message, detail, progress_current, progress_total, progress_percent, duration_ms, sequence, occurred_at, created_at).
  • Reserved Scoping Dimensions: name cannot use any of RESERVED_EVENT_HISTORY_COLUMNS ("actor", "entity", "scope", "scope_key").
  • Automatic Indexing: When indexed=True, create_schema() creates a composite (name, occurred_at) index on queue_task_event_history for fast tenant-scoped timeline queries.

Implementing a Custom QueueEventLog Provider

To plug in a custom event history store, implement the seven methods of the QueueEventLog protocol:

Method Signature Contract
publish_event async (self, event: QueueEvent) -> None Enqueue or persist a single QueueEvent
publish_event_after_commit async (self, event: QueueEvent, *, release: Callable[[], Awaitable[None]], barrier: bool = False) -> None Persist event (flushing immediately when barrier=True) and invoke await release() only after the record is durably committed
flush_events async (self) -> None Flush any buffered writes to durable storage
aclose async (self) -> None Drain pending writes and release connections during QueueService.close()
query_events async (self, query: QueueEventQuery \| None = None, *, extra: Mapping[str, str] \| None = None) -> OffsetPagination[QueueEventLogRecord] Filter, sort by stable order key, and paginate event records
summarize_stages async (self, query: QueueEventQuery \| None = None) -> list[QueueEventStageSummary] Call require_unpaginated_query(query) and return per-stage aggregations
cleanup_events async (self, *, before: datetime, match: QueueEventQuery \| None = None, exclude: Sequence[QueueEventQuery] = (), limit: int \| None = None) -> int Delete records older than before matching match and not matching any exclude query, up to limit

Commit-Aware Release Contract

Custom QueueEventLog implementations must implement publish_event_after_commit and aclose faithfully. Invoking await release() before the event is committed breaks the commit-before-live guarantee for SSE and WebSocket subscribers.