Task Options & Uniqueness¶

Every @task declaration sets default scheduling, timeout, thread-offloading, and deduplication behavior. Callers can override scheduling and identity options per invocation in QueueService.enqueue(...).

Runnable Example¶

"""Configuring task priority, delays, expiration, timeouts, and uniqueness reservations."""

from datetime import timedelta

from litestar_queues import QueueConfig, QueueService, WorkerConfig, task


@task(
    "billing.charge_invoice",
    queue="billing",
    priority=10,
    timeout=30.0,
    expires_in=timedelta(minutes=15),
    unique_by="arguments",
    unique_until="forever",
)
async def charge_invoice(invoice_id: str, amount_cents: int) -> dict[str, object]:
    """Charge an invoice at most once per unique argument signature."""
    return {"invoice_id": invoice_id, "amount_cents": amount_cents, "charged": True}


async def main() -> None:
    """Verify argument-based forever uniqueness and explicit reservation reset."""
    config = QueueConfig(
        queue_backend="memory", execution_backend="immediate", worker=WorkerConfig(placement="external")
    )
    async with QueueService(config=config) as service:
        first = await service.enqueue(charge_invoice, "inv-1001", 2500)
        duplicate = await service.enqueue(charge_invoice, "inv-1001", 2500)
        assert first.id == duplicate.id

        await first.refresh()
        assert first.record is not None
        assert first.record.key is not None

        reservation = await service.get_task_identity(first.record.key)
        assert reservation is not None
        assert reservation.task_id == first.id

        was_reset = await service.reset_task_identity(first.record.key)
        assert was_reset is True

        requeued = await service.enqueue(charge_invoice, "inv-1001", 2500)
        assert requeued.id != first.id

Priority & Delayed Execution¶

• Priority (priority: int = 0): Higher integers are claimed before lower integers within the same due window (priority=10 runs before priority=0).
• Relative Delay (run_after: float | timedelta | None): Defers initial eligibility by adding the duration to the enqueue timestamp and storing the record with status="scheduled".
• Absolute Schedule (scheduled_at: datetime | None): Passed at enqueue(..., scheduled_at=...) to schedule a task for a specific UTC timestamp. Naive datetime objects are interpreted as UTC.

Lifecycle Clocks¶

litestar-queues separates queue waiting deadlines, execution attempt limits, and post-terminal storage retention into independent clocks:

Synchronous Callables & Thread Offloading¶

litestar-queues runs on an asyncio event loop. Synchronous def task functions are offloaded to a managed ThreadPoolExecutor (QueueConfig.sync_thread_pool_size, defaulting to min(32, effective_cpu_count + 4)).

Task Identity & Deduplication¶

Deduplication prevents duplicate queue records from piling up when producers submit redundant work.

Identity Precedence Order¶

When QueueService.enqueue() resolves the effective identity key stored on QueuedTaskRecord.key, it evaluates sources in strict precedence order:

When unique_by="arguments" is active, litestar-queues binds positional and keyword arguments against the callable's inspect.Signature, applies default parameter values, normalizes collections, and hashes the serialized payload. Payloads exceeding QueueConfig.max_argument_identity_bytes (default 65536 bytes) raise TaskIdentityTooLargeError.

Identity Lifetime: Terminal vs Forever¶

unique_until controls how long an identity key blocks subsequent enqueue() calls:

• unique_until="terminal" (default): Blocks duplicate enqueues only while the existing record is non-terminal (pending, scheduled, or running). Matching enqueue() calls return a TaskResult pointing to the active record. Once the record reaches completed, failed, cancelled, or expired, the identity is automatically released.
• unique_until="forever": Persists a lightweight TaskReservation (key, task_id, task_name, created_at) that survives terminal completion and routine cleanup_completed / terminal_ttl purges. Subsequent enqueues with the same identity return the existing reservation's task handle until an operator explicitly resets the key.

Inspecting and Resetting Forever Reservations¶

Use QueueService.get_task_identity(key) and QueueService.reset_task_identity(key) with the exact effective key (record.key) to inspect or clear a permanent reservation:

reservation = await queue_service.get_task_identity(record.key)
if reservation is not None:
    await queue_service.reset_task_identity(reservation.key)