Bounded Queue Maintenance¶

QueueMaintenanceService (configured via QueueMaintenanceConfig) executes a finite, bounded pass of repair and retention work under token-fenced distributed ownership and a wall-clock time budget, then returns.

Maintenance never starts a polling worker loop, never executes due application tasks, and never auto-provisions external schedulers or database schema out of band.

Runnable Configuration Example¶

"""Configuring and running bounded QueueMaintenanceService passes."""

from pathlib import Path
from tempfile import TemporaryDirectory

from sqlspec.adapters.aiosqlite import AiosqliteConfig

from litestar_queues import QueueConfig, QueueMaintenanceConfig, QueueMaintenanceService, QueueService, WorkerConfig
from litestar_queues.backends.sqlspec import SQLSpecBackendConfig, SQLSpecQueueBackend, SQLSpecWorkerWakeupConfig
from litestar_queues.events import EventHistoryConfig, QueueEventQuery, QueueEventRetentionRule, QueueEventsConfig


def build_maintenance_config() -> QueueMaintenanceConfig:
    """Construct a bounded QueueMaintenanceConfig with explicit retention thresholds and OIDC route protection."""
    return QueueMaintenanceConfig(
        time_budget=120.0,
        coordination_timeout=180.0,
        expire_limit=500,
        external_limit=100,
        stale_after=90.0,
        stale_limit=100,
        schedules_limit=50,
        terminal_retention=86_400.0,
        terminal_limit=1000,
        event_retention_rules=(QueueEventRetentionRule(max_age=604_800.0, match=QueueEventQuery(scope="task")),),
        event_limit=1000,
        mount_route=True,
        verify_oidc=True,
        oidc_audience="https://worker-service-uc.a.run.app",
        oidc_service_account_email="scheduler-invoker@my-gcp-project.iam.gserviceaccount.com",
    )


async def main() -> None:
    """Execute a finite 6-phase maintenance pass against a persistent SQLite SQLSpec backend."""
    maintenance_config = build_maintenance_config()
    with TemporaryDirectory() as tmp_dir:
        db_path = Path(tmp_dir) / "maintenance.sqlite3"
        backend_config = SQLSpecBackendConfig(
            sqlspec_config=AiosqliteConfig(connection_config={"database": str(db_path)}),
            worker_wakeups=SQLSpecWorkerWakeupConfig(transport="polling"),
        )
        config = QueueConfig(
            queue_backend=backend_config,
            execution_backend="local",
            worker=WorkerConfig(placement="external"),
            maintenance=maintenance_config,
            events=QueueEventsConfig(history=EventHistoryConfig()),
        )
        async with QueueService(config=config) as service:
            backend = service.get_queue_backend()
            assert isinstance(backend, SQLSpecQueueBackend)
            await backend.create_schema()

            runner = QueueMaintenanceService(service, maintenance_config)
            summary = await runner.run()
            assert summary.outcome == "completed"
            assert summary.acquired is True
            assert tuple(p.phase for p in summary.phases) == (
                "expire",
                "external",
                "stale",
                "schedules",
                "terminal",
                "events",
            )

Maintenance Phase Execution Order¶

Every maintenance run evaluates bounded phases in the fixed PHASE_ORDER:

Non-Destructive Defaults¶

Destructive cleanup and stale-recovery thresholds (stale_after, terminal_retention, and event_retention_rules) default to None (or ()), leaving their phases skipped until explicitly configured.

Distributed Ownership & Coordination Storage¶

Before running any phase, QueueMaintenanceService acquires a token-fenced distributed lease (MAINTENANCE_NAME = "queue-maintenance") from the storage backend for coordination_timeout seconds (default 360.0s).

• Budget Invariant: coordination_timeout must strictly exceed time_budget (default 300.0s) so distributed ownership outlives the entire maintenance run; violating this raises QueueConfigurationError.
• Concurrent Invocations: If another process already holds active maintenance ownership, QueueMaintenanceService.run() returns immediately with outcome="already_running" and acquired=False.
• In-Memory Backend Rejection: Because InMemoryQueueBackend is process-local and cannot coordinate across processes, invoking QueueMaintenanceService against the in-memory backend raises QueueConfigurationError. Use a persistent backend (SQLSpecQueueBackend, SQLAlchemyBackend, RedisQueueBackend, ValkeyQueueBackend, or EphemeralQueueBackend).

Schema Ownership & Migration Requirements¶

Running Maintenance via CLI & Exit Codes¶

Run all enabled phases or narrow execution to specific phases with litestar queues run-maintenance:

litestar --app my_app.main:app queues run-maintenance --json
litestar --app my_app.main:app queues run-maintenance --phase expire --phase external --phase schedules

Passing --phase only narrows which phases are evaluated; a phase whose retention threshold (stale_after, terminal_retention, or event_retention_rules) is unconfigured remains "skipped".

Authenticated HTTP Maintenance Endpoint¶

For serverless deployments where triggering an authenticated HTTP request from an external scheduler is simpler than launching a container job, set QueueMaintenanceConfig(mount_route=True, ...). QueuePlugin mounts POST /_litestar-queues/maintenance (or /_<namespace>/maintenance) protected by at least one of:

• verify_oidc=True with oidc_audience="https://..." (and optional oidc_service_account_email) using CloudTasksOIDCGuard
• trust_platform_auth=True when platform IAM already authenticates callers
• guards=(...) with custom Litestar Guard callables

The HTTP endpoint returns 200 OK for "completed" / "already_running", 207 Multi-Status for "partial", and 503 Service Unavailable for "failed".

Recommended Scheduling Cadence & Launch Surfaces¶

Because workers and native backend TTL policies (terminal_ttl) already handle routine operations, external maintenance is designed for infrequent periodic sweeps (such as every six hours with 0 */6 * * *, or daily). Cloud Run Jobs, Kubernetes CronJobs, systemd timers, and Cloud Scheduler are supported external launch surfaces; container orchestration itself is not part of the core design of QueueMaintenanceService.

apiVersion: batch/v1
kind: CronJob
metadata:
  name: litestar-queues-maintenance
spec:
  schedule: "0 */6 * * *"
  concurrencyPolicy: Forbid
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: OnFailure
          containers:
            - name: maintenance
              image: my-org/my-app:latest
              command:
                - litestar
                - --app
                - my_app.main:app
                - queues
                - run-maintenance
                - --json

gcloud scheduler jobs create http litestar-queues-maintenance \
  --location=us-central1 \
  --schedule="0 */6 * * *" \
  --uri="https://worker-service-uc.a.run.app/_litestar-queues/maintenance" \
  --http-method=POST \
  --oidc-service-account-email="scheduler-invoker@my-gcp-project.iam.gserviceaccount.com" \
  --oidc-token-audience="https://worker-service-uc.a.run.app"

[Unit]
Description=Run bounded litestar-queues maintenance every 6 hours

[Timer]
OnCalendar=*-*-* 00/6:00:00
Persistent=true

[Install]
WantedBy=timers.target