API reference

Core

smallage.core

Transport, worker and scheduling core. Importing Litestar from here is forbidden.

ENVELOPE_VERSION = b'1' module-attribute

__all__ = ['ENVELOPE_VERSION', 'BrokerHandler', 'Codec', 'CollectingEnqueuer', 'ConfigurationError', 'CronJob', 'DeferredEnqueuer', 'EagerEnqueuer', 'Enqueuer', 'Envelope', 'FilePayloadStore', 'JsonCodec', 'MalformedEnvelope', 'PayloadMissing', 'PayloadStore', 'PayloadTooLarge', 'Pending', 'Record', 'RedisResultStore', 'RedisScheduler', 'RedisStreamsTransport', 'ResultStore', 'RetryPolicy', 'Scheduler', 'Sleeper', 'SmallageError', 'StreamTransport', 'TaskHandler', 'TaskResult', 'UnknownTask', 'WorkerConfig', 'WorkerStats', 'current_enqueuer', 'dlq_key', 'from_fields', 'run', 'run_with_signals', 'to_fields', 'worker_running'] module-attribute

current_enqueuer = ContextVar('smallage_enqueuer', default=None) module-attribute

The enqueuer in force for this unit of work, if one was bound.

Set it and every enqueue inside the block routes through it, which is what lets publication know about the transaction without every call site being told.

Sleeper = Callable[[float], Awaitable[None]]

BrokerHandler

Bases: Protocol

Handles an entry this library did not write.

Broker mode gets the raw record: the payload is in somebody else's format, so there is no envelope to decode and no result to store.

Source code in src/smallage/core/protocols.py

class BrokerHandler(Protocol):
    """Handles an entry this library did not write.

    Broker mode gets the raw record: the payload is in somebody else's format,
    so there is no envelope to decode and no result to store.
    """

    async def __call__(self, record: Record, /) -> None: ...

Codec

Bases: Protocol

Payload encoder/decoder. The plugin substitutes the app's msgspec pair.

Source code in src/smallage/core/protocols.py

class Codec(Protocol):
    """Payload encoder/decoder. The plugin substitutes the app's msgspec pair."""

    def encode(self, value: object, /) -> bytes: ...
    def decode(self, raw: bytes, /) -> object: ...

CollectingEnqueuer

Records what would have been enqueued, and runs nothing.

For asserting that a request scheduled the work it was supposed to, without dragging a worker or a Redis into the test.

Source code in src/smallage/core/testing.py

class CollectingEnqueuer:
    """Records what would have been enqueued, and runs nothing.

    For asserting that a request scheduled the work it was supposed to, without
    dragging a worker or a Redis into the test.
    """

    def __init__(self) -> None:
        self.enqueued: list[tuple[Envelope, str]] = []

    async def enqueue(self, envelope: Envelope, *, queue: str) -> bytes:
        self.enqueued.append((envelope, queue))
        return f"collected-{len(self.enqueued)}".encode()

    def tasks(self) -> list[str]:
        return [envelope.task for envelope, _ in self.enqueued]

    def assert_enqueued(self, task: str, *, times: int = 1) -> None:
        actual = self.tasks().count(task)
        if actual != times:
            raise AssertionError(
                f"expected {task!r} to be enqueued {times} time(s), "
                f"got {actual}; enqueued: {self.tasks()}"
            )

    def assert_not_enqueued(self, task: str) -> None:
        self.assert_enqueued(task, times=0)

ConfigurationError

Bases: SmallageError

Invalid configuration. Raised at construction, never from the worker loop.

Source code in src/smallage/core/errors.py

class ConfigurationError(SmallageError):
    """Invalid configuration. Raised at construction, never from the worker loop."""

CronJob dataclass

Source code in src/smallage/core/cron.py

@dataclass(frozen=True, slots=True)
class CronJob:
    name: str
    expression: str
    task: str
    payload: bytes = b"{}"
    """Encoded arguments. An empty object, not null: it decodes into the task's
    argument struct, and a task taking nothing still has one."""
    queue: str = "default"
    timezone: str = "UTC"

    def __post_init__(self) -> None:
        if not self.name or ":" in self.name:
            raise ConfigurationError(
                f"cron job name must be non-empty and free of ':', got {self.name!r}"
            )
        try:
            zone = ZoneInfo(self.timezone)
        except (ZoneInfoNotFoundError, ValueError) as exc:
            raise ConfigurationError(
                f"cron job {self.name!r} has an unknown timezone {self.timezone!r}"
            ) from exc
        try:
            CronSim(self.expression, datetime.now(zone))
        except CronSimError as exc:
            raise ConfigurationError(
                f"cron job {self.name!r} has an invalid expression "
                f"{self.expression!r}: {exc}"
            ) from exc

payload = b'{}' class-attribute instance-attribute

Encoded arguments. An empty object, not null: it decodes into the task's argument struct, and a task taking nothing still has one.

DeferredEnqueuer

Holds jobs until something says the transaction went through.

Wrap the real enqueuer per unit of work, call flush from the commit hook and discard from the rollback one.

Source code in src/smallage/core/deferred.py

class DeferredEnqueuer:
    """Holds jobs until something says the transaction went through.

    Wrap the real enqueuer per unit of work, call ``flush`` from the commit hook
    and ``discard`` from the rollback one.
    """

    def __init__(self, target: Enqueuer) -> None:
        self.target = target
        self._pending: list[tuple[Envelope, str]] = []

    @property
    def pending(self) -> tuple[tuple[Envelope, str], ...]:
        return tuple(self._pending)

    async def enqueue(self, envelope: Envelope, *, queue: str) -> bytes:
        """Record the job. Nothing reaches Redis until ``flush``."""
        self._pending.append((envelope, queue))
        return envelope.id.encode()

    async def flush(self) -> list[bytes]:
        """Publish everything buffered, oldest first, and forget it.

        The buffer is cleared before publishing: a failure part way through must
        not leave jobs that would be published twice by the next flush.
        """
        buffered, self._pending = self._pending, []
        return [
            await self.target.enqueue(envelope, queue=queue)
            for envelope, queue in buffered
        ]

    def discard(self) -> None:
        """Drop everything buffered. For the rollback path."""
        self._pending.clear()

    @asynccontextmanager
    async def active(self) -> AsyncGenerator[DeferredEnqueuer]:
        """Route every enqueue in this block through the buffer.

        Binding only. Whether the block ends in ``flush`` or ``discard`` is the
        caller's decision, because only the caller knows how the transaction
        ended.
        """
        token = current_enqueuer.set(self)
        try:
            yield self
        finally:
            current_enqueuer.reset(token)

active() async

Route every enqueue in this block through the buffer.

Binding only. Whether the block ends in flush or discard is the caller's decision, because only the caller knows how the transaction ended.

Source code in src/smallage/core/deferred.py

@asynccontextmanager
async def active(self) -> AsyncGenerator[DeferredEnqueuer]:
    """Route every enqueue in this block through the buffer.

    Binding only. Whether the block ends in ``flush`` or ``discard`` is the
    caller's decision, because only the caller knows how the transaction
    ended.
    """
    token = current_enqueuer.set(self)
    try:
        yield self
    finally:
        current_enqueuer.reset(token)

discard()

Drop everything buffered. For the rollback path.

Source code in src/smallage/core/deferred.py

def discard(self) -> None:
    """Drop everything buffered. For the rollback path."""
    self._pending.clear()

enqueue(envelope, *, queue) async

Record the job. Nothing reaches Redis until flush.

Source code in src/smallage/core/deferred.py

async def enqueue(self, envelope: Envelope, *, queue: str) -> bytes:
    """Record the job. Nothing reaches Redis until ``flush``."""
    self._pending.append((envelope, queue))
    return envelope.id.encode()

flush() async

Publish everything buffered, oldest first, and forget it.

The buffer is cleared before publishing: a failure part way through must not leave jobs that would be published twice by the next flush.

Source code in src/smallage/core/deferred.py

async def flush(self) -> list[bytes]:
    """Publish everything buffered, oldest first, and forget it.

    The buffer is cleared before publishing: a failure part way through must
    not leave jobs that would be published twice by the next flush.
    """
    buffered, self._pending = self._pending, []
    return [
        await self.target.enqueue(envelope, queue=queue)
        for envelope, queue in buffered
    ]

EagerEnqueuer

Runs each task inline, at the moment it is enqueued.

Nothing about retries, ordering or concurrency is reproduced -- that is the point. It keeps unit tests of application code free of infrastructure, and anything that depends on real queue behaviour belongs in an integration test.

Source code in src/smallage/core/testing.py

class EagerEnqueuer:
    """Runs each task inline, at the moment it is enqueued.

    Nothing about retries, ordering or concurrency is reproduced -- that is the
    point. It keeps unit tests of application code free of infrastructure, and
    anything that depends on real queue behaviour belongs in an integration test.
    """

    def __init__(self, registry: Mapping[str, TaskHandler]) -> None:
        self.registry = registry
        self.enqueued: list[tuple[Envelope, str]] = []

    async def enqueue(self, envelope: Envelope, *, queue: str) -> bytes:
        handler = self.registry.get(envelope.task)
        if handler is None:
            raise UnknownTask(f"no handler registered for task {envelope.task!r}")
        self.enqueued.append((envelope, queue))
        await handler(envelope)
        return f"eager-{len(self.enqueued)}".encode()

Enqueuer

Bases: Protocol

Publishing seam.

Deferred publication (flushing on the SQLAlchemy after_commit hook) and the eager mode used in application unit tests are implementations of this protocol, not forks of the transport. The scheduler binds here too, so that it never has to import a concrete transport.

Source code in src/smallage/core/protocols.py

class Enqueuer(Protocol):
    """Publishing seam.

    Deferred publication (flushing on the SQLAlchemy ``after_commit`` hook) and the
    eager mode used in application unit tests are implementations of this protocol,
    not forks of the transport. The scheduler binds here too, so that it never has
    to import a concrete transport.
    """

    async def enqueue(self, envelope: Envelope, *, queue: str) -> bytes: ...

Envelope

Bases: Struct

Source code in src/smallage/core/envelope.py

class Envelope(msgspec.Struct, frozen=True, kw_only=True):
    id: str
    task: str
    payload: bytes
    enqueued_at: int
    attempt: int = 0
    payload_ref: str | None = None
    traceparent: str | None = None
    tracestate: str | None = None
    dedup: str | None = None
    """Application-chosen key. At most one job per key runs within its window."""
    result_ttl_ms: int | None = None
    """Keep the outcome for this long. Absent means nobody is waiting for it."""
    history: tuple[str, ...] = ()
    """One compact line per failed attempt, oldest first.

    Bounded by the retry policy, and each line is truncated, so a job that keeps
    failing cannot grow its own entry without limit.
    """

dedup = None class-attribute instance-attribute

Application-chosen key. At most one job per key runs within its window.

history = () class-attribute instance-attribute

One compact line per failed attempt, oldest first.

Bounded by the retry policy, and each line is truncated, so a job that keeps failing cannot grow its own entry without limit.

result_ttl_ms = None class-attribute instance-attribute

Keep the outcome for this long. Absent means nobody is waiting for it.

FilePayloadStore

Payloads as files under one directory.

Every worker must see the same directory -- a network volume, or a single host. A path local to one pod means the job runs wherever the file happens to be and fails everywhere else, so the reference records the path and reading it back from the wrong place says exactly that.

Source code in src/smallage/core/payloads.py

class FilePayloadStore:
    """Payloads as files under one directory.

    Every worker must see the same directory -- a network volume, or a single
    host. A path local to one pod means the job runs wherever the file happens to
    be and fails everywhere else, so the reference records the path and reading
    it back from the wrong place says exactly that.
    """

    def __init__(self, root: Path | str) -> None:
        self.root = Path(root)
        if not self.root.is_absolute():
            # A relative path resolves against each process's working directory,
            # which is rarely the same one twice.
            raise ConfigurationError(f"root must be absolute, got {self.root}")
        self.root.mkdir(parents=True, exist_ok=True)

    def _path(self, job_id: str) -> Path:
        if "/" in job_id or job_id in {"", ".", ".."}:
            raise ConfigurationError(f"job id is not a usable file name: {job_id!r}")
        return self.root / f"{job_id}.payload"

    async def put(self, job_id: str, data: bytes) -> str:
        path = self._path(job_id)
        # Write beside it and rename: a reader must never see half a payload,
        # and rename within a directory is atomic.
        staging = path.with_suffix(".partial")
        await anyio.Path(staging).write_bytes(data)
        await anyio.Path(staging).rename(path)
        return f"{SCHEME}{path}"

    async def get(self, reference: str) -> bytes:
        path = Path(reference.removeprefix(SCHEME))
        try:
            return await anyio.Path(path).read_bytes()
        except FileNotFoundError as exc:
            raise PayloadMissing(
                f"no payload at {reference}; the store is not the one that wrote "
                "it, or it has been cleaned up"
            ) from exc

    async def discard(self, reference: str) -> None:
        """Remove a payload once its job is finished with it."""
        await anyio.Path(Path(reference.removeprefix(SCHEME))).unlink(missing_ok=True)

discard(reference) async

Remove a payload once its job is finished with it.

Source code in src/smallage/core/payloads.py

async def discard(self, reference: str) -> None:
    """Remove a payload once its job is finished with it."""
    await anyio.Path(Path(reference.removeprefix(SCHEME))).unlink(missing_ok=True)

JsonCodec

Default payload codec. The plugin layer substitutes the app's msgspec pair.

Source code in src/smallage/core/envelope.py

class JsonCodec:
    """Default payload codec. The plugin layer substitutes the app's msgspec pair."""

    def encode(self, value: object, /) -> bytes:
        return msgspec.json.encode(value)

    def decode(self, raw: bytes, /) -> object:
        return msgspec.json.decode(raw)

MalformedEnvelope

Bases: SmallageError

A stream entry is missing a field the envelope requires.

Source code in src/smallage/core/errors.py

class MalformedEnvelope(SmallageError):
    """A stream entry is missing a field the envelope requires."""

PayloadMissing

Bases: SmallageError

A record references a payload the store no longer has.

Source code in src/smallage/core/payloads.py

class PayloadMissing(SmallageError):
    """A record references a payload the store no longer has."""

PayloadStore

Bases: Protocol

Somewhere large arguments live instead of in Redis.

Redis holds the stream in memory, so a payload measured in megabytes is a direct route to an out-of-memory kill. Above a threshold the arguments go here and the entry carries a reference.

Source code in src/smallage/core/protocols.py

class PayloadStore(Protocol):
    """Somewhere large arguments live instead of in Redis.

    Redis holds the stream in memory, so a payload measured in megabytes is a
    direct route to an out-of-memory kill. Above a threshold the arguments go
    here and the entry carries a reference.
    """

    async def put(self, job_id: str, data: bytes) -> str: ...
    async def get(self, reference: str) -> bytes: ...

PayloadTooLarge

Bases: SmallageError

Encoded arguments exceed the configured limit for an inline payload.

Source code in src/smallage/core/errors.py

class PayloadTooLarge(SmallageError):
    """Encoded arguments exceed the configured limit for an inline payload."""

Pending

Bases: Struct

One unacknowledged entry, as XPENDING describes it.

Source code in src/smallage/core/envelope.py

class Pending(msgspec.Struct, frozen=True):
    """One unacknowledged entry, as ``XPENDING`` describes it."""

    stream: str
    entry_id: bytes
    consumer: str
    times_delivered: int

Record

Bases: Struct

A stream entry as Redis returns it.

Broker mode consumes foreign payloads and never builds an Envelope, so the transport hands back raw records and decoding is a separate step in the worker.

Source code in src/smallage/core/envelope.py

class Record(msgspec.Struct, frozen=True):
    """A stream entry as Redis returns it.

    Broker mode consumes foreign payloads and never builds an ``Envelope``, so the
    transport hands back raw records and decoding is a separate step in the worker.
    """

    stream: str
    entry_id: bytes
    fields: dict[bytes, bytes]

RedisResultStore

Results in plain keys, with a list alongside so a waiter can block.

The list exists because polling for a result is the thing every queue gets asked to stop doing. A waiter blocks on it and puts the token back, so several waiters on one job all wake.

Source code in src/smallage/core/results.py

class RedisResultStore:
    """Results in plain keys, with a list alongside so a waiter can block.

    The list exists because polling for a result is the thing every queue gets
    asked to stop doing. A waiter blocks on it and puts the token back, so
    several waiters on one job all wake.
    """

    def __init__(
        self, *, control: Redis, namespace: str = "lrs", default_ttl_ms: int = 3_600_000
    ) -> None:
        if connection_kwarg(control, "decode_responses"):
            raise ConfigurationError(
                "control client must be built with decode_responses=False; "
                "results are opaque bytes"
            )
        if default_ttl_ms < 1:
            raise ConfigurationError(
                f"default_ttl_ms must be positive, got {default_ttl_ms}"
            )
        self.control = control
        self.namespace = validate_namespace(namespace)
        self.default_ttl_ms = default_ttl_ms

    async def store(
        self, job_id: str, result: TaskResult, *, ttl_ms: int | None = None
    ) -> None:
        ttl = ttl_ms or self.default_ttl_ms
        async with self.control.pipeline(transaction=True) as pipe:
            pipe.set(
                result_key(self.namespace, job_id),
                msgspec.msgpack.encode(result),
                px=ttl,
            )
            pipe.lpush(result_wait_key(self.namespace, job_id), b"1")
            pipe.pexpire(result_wait_key(self.namespace, job_id), ttl)
            await pipe.execute()

    async def get(self, job_id: str) -> TaskResult | None:
        raw: Any = await self.control.get(result_key(self.namespace, job_id))
        if raw is None:
            return None
        return msgspec.msgpack.decode(raw, type=TaskResult)

    async def wait(self, job_id: str, *, timeout_s: float) -> TaskResult | None:
        """Block until the job finishes, or give up. Returns None on timeout."""
        stored = await self.get(job_id)
        if stored is not None:
            return stored
        woken = await self.control.blpop(
            [result_wait_key(self.namespace, job_id)], timeout=timeout_s
        )
        if woken is None:
            return None
        # Put the token back so a second waiter on the same job also wakes.
        await self.control.rpush(result_wait_key(self.namespace, job_id), b"1")
        return await self.get(job_id)

wait(job_id, *, timeout_s) async

Block until the job finishes, or give up. Returns None on timeout.

Source code in src/smallage/core/results.py

async def wait(self, job_id: str, *, timeout_s: float) -> TaskResult | None:
    """Block until the job finishes, or give up. Returns None on timeout."""
    stored = await self.get(job_id)
    if stored is not None:
        return stored
    woken = await self.control.blpop(
        [result_wait_key(self.namespace, job_id)], timeout=timeout_s
    )
    if woken is None:
        return None
    # Put the token back so a second waiter on the same job also wakes.
    await self.control.rpush(result_wait_key(self.namespace, job_id), b"1")
    return await self.get(job_id)

RedisScheduler

Scheduled jobs held in a ZSET, with the payload beside it in a hash.

The entry lives in a hash rather than inside the ZSET member so payloads stay opaque bytes -- packing them into a member would mean encoding them twice.

Source code in src/smallage/core/scheduler.py

class RedisScheduler:
    """Scheduled jobs held in a ZSET, with the payload beside it in a hash.

    The entry lives in a hash rather than inside the ZSET member so payloads stay
    opaque bytes -- packing them into a member would mean encoding them twice.
    """

    def __init__(
        self,
        *,
        control: Redis,
        namespace: str = "lrs",
        shards: int = 1,
    ) -> None:
        if connection_kwarg(control, "decode_responses"):
            raise ConfigurationError(
                "control client must be built with decode_responses=False; "
                "scheduled payloads are opaque bytes"
            )
        if shards < 1:
            raise ConfigurationError(f"shards must be at least 1, got {shards}")
        self.control = control
        self.namespace = validate_namespace(namespace)
        self.shards = shards
        self.zset = sched_key(self.namespace)
        self.leader = leader_key(self.namespace)
        self._scripts: SchedulerScripts = register_scheduler(control)

    async def now_ms(self) -> int:
        """Redis is the clock. Worker clocks drift apart; this one does not."""
        seconds, microseconds = await self.control.time()
        return seconds * 1000 + microseconds // 1000

    async def schedule_at(
        self,
        envelope: Envelope,
        *,
        queue: str,
        when_ms: int,
        scheduled_id: str | None = None,
    ) -> str:
        job_id = scheduled_id or envelope.id
        stream = stream_for(
            self.namespace, validate_queue(queue), self.shards, envelope.id
        )
        fields: dict[Any, Any] = dict(to_fields(envelope))
        fields[STREAM_FIELD] = stream.encode()
        async with self.control.pipeline(transaction=True) as pipe:
            pipe.hset(sched_job_key(self.namespace, job_id), mapping=fields)
            pipe.zadd(self.zset, {job_id: when_ms})
            await pipe.execute()
        return job_id

    async def schedule_in(
        self, envelope: Envelope, *, queue: str, delay_ms: int
    ) -> str:
        return await self.schedule_at(
            envelope, queue=queue, when_ms=await self.now_ms() + delay_ms
        )

    async def schedule_cron(self, jobs: Sequence[CronJob]) -> list[str]:
        """Put the next occurrence of every job in the ZSET.

        Idempotent by construction: the id encodes the instant, so a second
        leader doing the same work writes the same member.
        """
        now = await self.now_ms()
        scheduled = []
        for job in jobs:
            fire_ms = next_fire_ms(job, now)
            if fire_ms is None:
                continue
            envelope = occurrence_envelope(job, fire_ms)
            await self.schedule_at(
                envelope, queue=job.queue, when_ms=fire_ms, scheduled_id=envelope.id
            )
            scheduled.append(envelope.id)
        return scheduled

    async def promote(self, *, limit: int = 100) -> list[bytes]:
        moved = await self._scripts.promote(
            keys=[self.zset], args=[limit, sched_job_key(self.namespace, "")]
        )
        return [bytes(entry_id) for entry_id in moved]

    async def pending(self) -> int:
        return int(await self.control.zcard(self.zset))

    async def hold_leadership(self, token: str, *, ttl_ms: int) -> bool:
        """Take the lease, or extend it if it is already ours.

        Extending is a compare-and-set: a worker that has already lost the lease
        must not be able to prolong the new holder's key.
        """
        taken = await self.control.set(self.leader, token, nx=True, px=ttl_ms)
        if taken:
            return True
        renewed = await self._scripts.renew_leader(
            keys=[self.leader], args=[token, ttl_ms]
        )
        return bool(renewed)

    async def release_leadership(self, token: str) -> bool:
        released = await self._scripts.release_leader(keys=[self.leader], args=[token])
        return bool(released)

hold_leadership(token, *, ttl_ms) async

Take the lease, or extend it if it is already ours.

Extending is a compare-and-set: a worker that has already lost the lease must not be able to prolong the new holder's key.

Source code in src/smallage/core/scheduler.py

async def hold_leadership(self, token: str, *, ttl_ms: int) -> bool:
    """Take the lease, or extend it if it is already ours.

    Extending is a compare-and-set: a worker that has already lost the lease
    must not be able to prolong the new holder's key.
    """
    taken = await self.control.set(self.leader, token, nx=True, px=ttl_ms)
    if taken:
        return True
    renewed = await self._scripts.renew_leader(
        keys=[self.leader], args=[token, ttl_ms]
    )
    return bool(renewed)

now_ms() async

Redis is the clock. Worker clocks drift apart; this one does not.

Source code in src/smallage/core/scheduler.py

async def now_ms(self) -> int:
    """Redis is the clock. Worker clocks drift apart; this one does not."""
    seconds, microseconds = await self.control.time()
    return seconds * 1000 + microseconds // 1000

schedule_cron(jobs) async

Put the next occurrence of every job in the ZSET.

Idempotent by construction: the id encodes the instant, so a second leader doing the same work writes the same member.

Source code in src/smallage/core/scheduler.py

async def schedule_cron(self, jobs: Sequence[CronJob]) -> list[str]:
    """Put the next occurrence of every job in the ZSET.

    Idempotent by construction: the id encodes the instant, so a second
    leader doing the same work writes the same member.
    """
    now = await self.now_ms()
    scheduled = []
    for job in jobs:
        fire_ms = next_fire_ms(job, now)
        if fire_ms is None:
            continue
        envelope = occurrence_envelope(job, fire_ms)
        await self.schedule_at(
            envelope, queue=job.queue, when_ms=fire_ms, scheduled_id=envelope.id
        )
        scheduled.append(envelope.id)
    return scheduled

RedisStreamsTransport

Stream operations for one queue.

Takes two clients on purpose: a blocking XREADGROUP occupies its connection for the whole block window, and an ack or a liveness refresh queued behind it makes a healthy worker look dead to its peers.

consumer must be unique per running worker. Redis groups entries in the pending list by consumer name, and reclaim uses that name to tell its own work from a peer's.

Source code in src/smallage/core/transport.py

class RedisStreamsTransport:
    """Stream operations for one queue.

    Takes two clients on purpose: a blocking ``XREADGROUP`` occupies its connection
    for the whole block window, and an ack or a liveness refresh queued behind it
    makes a healthy worker look dead to its peers.

    ``consumer`` must be unique per running worker. Redis groups entries in the
    pending list by consumer name, and reclaim uses that name to tell its own
    work from a peer's.
    """

    def __init__(
        self,
        *,
        reader: Redis,
        control: Redis,
        consumer: str,
        namespace: str = DEFAULT_NAMESPACE,
        queues: Sequence[str] = DEFAULT_QUEUES,
        shards: int = 1,
        fairness_every: int = 10,
        external: Sequence[str] = (),
        group: str = DEFAULT_GROUP,
        block_ms: int = 5_000,
        max_payload_bytes: int = DEFAULT_MAX_PAYLOAD_BYTES,
    ) -> None:
        if reader is control:
            raise ConfigurationError(
                "reader and control must be separate clients: a blocking "
                "XREADGROUP would otherwise stall acks and liveness refreshes"
            )
        for role, client in (("reader", reader), ("control", control)):
            if connection_kwarg(client, "decode_responses"):
                raise ConfigurationError(
                    f"{role} client must be built with decode_responses=False; "
                    "stream payloads are opaque bytes"
                )
        if not consumer:
            raise ConfigurationError("consumer must not be empty")
        if max_payload_bytes < 1:
            raise ConfigurationError(
                f"max_payload_bytes must be at least 1, got {max_payload_bytes}"
            )
        if block_ms < 0:
            raise ConfigurationError(f"block_ms must not be negative, got {block_ms}")
        if not queues and not external:
            raise ConfigurationError(
                "a worker with neither queues nor external streams would read "
                "nothing; name at least one queue or one external stream"
            )
        if len(set(queues)) != len(queues):
            raise ConfigurationError(f"queues must not repeat, got {list(queues)}")
        if fairness_every < 0:
            raise ConfigurationError(
                f"fairness_every must not be negative, got {fairness_every}"
            )

        socket_timeout = connection_kwarg(reader, "socket_timeout")
        if socket_timeout is None:
            raise ConfigurationError(
                "reader client needs an explicit socket_timeout, otherwise a read "
                "hung by a failover never returns and the worker goes quiet"
            )
        if socket_timeout * 1000 <= block_ms:
            raise ConfigurationError(
                f"reader socket_timeout ({socket_timeout}s) must outlast block_ms "
                f"({block_ms}ms), or every healthy blocking read times out"
            )

        self.reader = reader
        self.control = control
        self.consumer = consumer
        self.namespace = validate_namespace(namespace)
        self.queues = tuple(validate_queue(queue) for queue in queues)
        self.group = group
        self.block_ms = block_ms
        self.shards = shards
        self.fairness_every = fairness_every
        self.max_payload_bytes = max_payload_bytes
        self.streams_by_queue = {
            queue: stream_keys(self.namespace, queue, shards) for queue in self.queues
        }
        self.streams = [
            stream for queue in self.queues for stream in self.streams_by_queue[queue]
        ]
        self._queue_of = {
            stream: queue
            for queue, streams in self.streams_by_queue.items()
            for stream in streams
        }
        self.external_streams = tuple(external)
        if set(self.external_streams) & set(self.streams):
            raise ConfigurationError(
                "external streams must not name this namespace's own queues"
            )
        self.dlq = dlq_key(self.namespace)
        self._passes = 0
        self._scripts: TransportScripts = register_transport(control)

    def alive_key(self, entry_id: bytes) -> str:
        return alive_key(self.namespace, entry_id)

    async def ensure_group(self) -> None:
        for stream in (*self.streams, *self.external_streams):
            try:
                await self.control.xgroup_create(
                    stream, self.group, id="0", mkstream=True
                )
            except ResponseError as exc:
                if "BUSYGROUP" not in str(exc):
                    raise

    async def enqueue(self, envelope: Envelope, *, queue: str) -> bytes:
        if len(envelope.payload) > self.max_payload_bytes:
            raise PayloadTooLarge(
                f"payload for task {envelope.task!r} is {len(envelope.payload)} bytes, "
                f"over the {self.max_payload_bytes} byte limit; "
                "store it out of band and enqueue a payload_ref instead"
            )
        stream = stream_for(
            self.namespace, validate_queue(queue), self.shards, envelope.id
        )
        fields: dict[Any, Any] = dict(to_fields(envelope))
        entry_id = await self.control.xadd(stream, fields)
        return _as_bytes(entry_id)

    def queue_of(self, stream: str) -> str:
        return self._queue_of[stream]

    def is_external(self, stream: str) -> bool:
        """A stream somebody else writes, in somebody else's format."""
        return stream in self.external_streams

    def _priority_order(self) -> list[str]:
        """Which queue to try first this pass.

        Strict priority would starve the low queue outright, so every
        ``fairness_every`` passes the order is inverted and the low queue gets
        first refusal. That bounds how long anything can wait behind a busy
        high-priority queue.
        """
        self._passes += 1
        if self.fairness_every and self._passes % self.fairness_every == 0:
            return list(reversed(self.queues))
        return list(self.queues)

    async def read(self, count: int) -> list[Record]:
        if count <= 0:
            # XREADGROUP COUNT 0 is not the same thing as not reading: a fixed
            # COUNT is what skews work between workers (taskiq-redis#91).
            return []
        if not self.streams:
            # Nothing of our own to order or to block on, so the foreign read is
            # the only one left and it is the one that waits. Without the block
            # this is a busy loop against Redis.
            return await self._read(self.external_streams, count, self.block_ms)
        if self.external_streams:
            # Read separately: foreign stream names carry no hash tag of ours, so
            # putting them in the same XREADGROUP would break in a cluster.
            records = await self._read(self.external_streams, count, None)
            if records:
                return records
        if len(self.queues) > 1:
            # XREADGROUP with BLOCK wakes on whichever stream has something, so
            # it cannot express priority. A non-blocking sweep from high to low
            # can, and only when everything is empty do we block on all of them.
            for queue in self._priority_order():
                records = await self._read(self.streams_by_queue[queue], count, None)
                if records:
                    return records
        return await self._read(self.streams, count, self.block_ms)

    async def _read(
        self, streams: Sequence[str], count: int, block_ms: int | None
    ) -> list[Record]:
        reply = await self.reader.xreadgroup(
            groupname=self.group,
            consumername=self.consumer,
            streams=dict.fromkeys(streams, ">"),
            count=count,
            # block=0 would wait forever; a non-blocking read wants no BLOCK at all.
            block=block_ms,
        )
        return _records_from_read(reply)

    async def mark_alive(self, entry_ids: Sequence[bytes], *, ttl_ms: int) -> None:
        if not entry_ids:
            return
        async with self.control.pipeline(transaction=False) as pipe:
            for entry_id in entry_ids:
                pipe.set(self.alive_key(entry_id), self.consumer, px=ttl_ms)
            await pipe.execute()

    async def refresh_alive(self, entry_ids: Iterable[bytes], *, ttl_ms: int) -> None:
        """PEXPIRE rather than SET: it must not resurrect an already acked entry."""
        keys = [self.alive_key(entry_id) for entry_id in entry_ids]
        if not keys:
            return
        async with self.control.pipeline(transaction=False) as pipe:
            for key in keys:
                pipe.pexpire(key, ttl_ms)
            await pipe.execute()

    async def clear_alive(self, entry_ids: Iterable[bytes]) -> None:
        """Hand an entry over immediately instead of waiting out its TTL."""
        keys = [self.alive_key(entry_id) for entry_id in entry_ids]
        if keys:
            await self.control.delete(*keys)

    async def claim_dedup(self, key: str, *, owner: str, ttl_ms: int) -> bool:
        """Win the right to run this key, or report that someone already has.

        The guarantee is at-least-once, so an application that must not repeat a
        side effect needs a gate it controls. This is that gate. The owner's own
        claim lets it back in, so a retry is not mistaken for a duplicate.
        """
        won = await self._scripts.dedup_claim(
            keys=[dedup_key(self.namespace, key)], args=[owner, ttl_ms]
        )
        return bool(won)

    async def complete_dedup(self, key: str, *, owner: str, ttl_ms: int) -> None:
        await self._scripts.dedup_complete(
            keys=[dedup_key(self.namespace, key)], args=[owner, ttl_ms]
        )

    async def release_dedup(self, key: str, *, owner: str) -> None:
        await self._scripts.dedup_release(
            keys=[dedup_key(self.namespace, key)], args=[owner]
        )

    async def ack(self, stream: str, entry_ids: Sequence[bytes]) -> int:
        if not entry_ids:
            return 0
        keys = [stream, *(self.alive_key(entry_id) for entry_id in entry_ids)]
        acked = await self._scripts.ack(keys=keys, args=[self.group, *entry_ids])
        return int(acked)

    async def pending(self, *, count: int, min_idle_ms: int) -> list[Pending]:
        if count <= 0:
            return []
        found: list[Pending] = []
        for stream in (*self.streams, *self.external_streams):
            entries = await self.control.xpending_range(
                stream,
                self.group,
                min="-",
                max="+",
                count=count - len(found),
                idle=min_idle_ms,
            )
            found.extend(
                Pending(
                    stream=stream,
                    entry_id=_as_bytes(entry["message_id"]),
                    consumer=_as_bytes(entry["consumer"]).decode(),
                    times_delivered=int(entry["times_delivered"]),
                )
                for entry in entries
            )
            if len(found) >= count:
                break
        return found

    async def dead_letter(
        self, record: Record, *, reason: str, detail: str, times_delivered: int
    ) -> bytes:
        """Park an entry that will never succeed, keeping everything about it.

        The original payload rides along untouched, so a fixed deployment can
        replay it without reconstructing anything.
        """
        seconds, microseconds = await self.control.time()
        fields: dict[Any, Any] = dict(record.fields)
        fields[b"dlq_reason"] = reason.encode()
        fields[b"dlq_detail"] = detail.encode()[:MAX_DETAIL_BYTES]
        fields[b"dlq_source"] = record.stream.encode()
        fields[b"dlq_deliveries"] = str(times_delivered).encode()
        fields[b"dlq_at"] = str(seconds * 1000 + microseconds // 1000).encode()
        return _as_bytes(await self.control.xadd(self.dlq, fields))

    async def reclaim(
        self, stream: str, entry_id: bytes, *, min_idle_ms: int, ttl_ms: int
    ) -> list[Record]:
        reply = await self._scripts.reclaim(
            keys=[stream, self.alive_key(entry_id)],
            args=[self.group, self.consumer, min_idle_ms, entry_id, ttl_ms],
        )
        return parse_xclaim_reply(stream, list(reply))

    async def sweep_consumers(self, *, min_idle_ms: int) -> int:
        """Remove consumers that hold nothing and have gone quiet.

        A name is derived per process start, so every deploy leaves one behind
        and Redis expires none of them. Deleting one that still owns pending
        entries would make them unclaimable, so the script checks and deletes in
        one step rather than trusting a reading taken a round trip earlier.

        Deleting a live worker's idle consumer costs nothing -- Redis recreates
        it on the next read -- which is why the only hard condition is that it
        holds nothing.
        """
        deleted = 0
        for stream in (*self.streams, *self.external_streams):
            deleted += int(
                await self._scripts.sweep_consumers(
                    keys=[stream], args=[self.group, min_idle_ms]
                )
            )
        return deleted

    async def trim(self, *, retention_ms: int) -> None:
        """Trim by time, floored at the oldest unacked entry.

        MINID has the same hazard MAXLEN does once a pending entry is older than
        the retention window -- trimming it away would drop unacknowledged work.
        """
        seconds, microseconds = await self.control.time()
        now_ms = seconds * 1000 + microseconds // 1000
        floor_ms = now_ms - retention_ms
        for stream in self.streams:
            summary = await self.control.xpending(stream, self.group)
            oldest = summary.get("min") if summary else None
            limit = floor_ms if oldest is None else min(floor_ms, _entry_ms(oldest))
            await self.control.xtrim(stream, minid=limit, approximate=True)

    async def lag(self) -> int | None:
        """Depth from XINFO GROUPS. Redis reports NULL when it cannot reconcile."""
        total = 0
        for stream in (*self.streams, *self.external_streams):
            groups = await self.control.xinfo_groups(stream)
            for group in groups:
                if _as_bytes(_get(group, "name")) != self.group.encode():
                    continue
                value = _get(group, "lag")
                if value is None:
                    return None
                total += int(value)
        return total

claim_dedup(key, *, owner, ttl_ms) async

Win the right to run this key, or report that someone already has.

The guarantee is at-least-once, so an application that must not repeat a side effect needs a gate it controls. This is that gate. The owner's own claim lets it back in, so a retry is not mistaken for a duplicate.

Source code in src/smallage/core/transport.py

async def claim_dedup(self, key: str, *, owner: str, ttl_ms: int) -> bool:
    """Win the right to run this key, or report that someone already has.

    The guarantee is at-least-once, so an application that must not repeat a
    side effect needs a gate it controls. This is that gate. The owner's own
    claim lets it back in, so a retry is not mistaken for a duplicate.
    """
    won = await self._scripts.dedup_claim(
        keys=[dedup_key(self.namespace, key)], args=[owner, ttl_ms]
    )
    return bool(won)

clear_alive(entry_ids) async

Hand an entry over immediately instead of waiting out its TTL.

Source code in src/smallage/core/transport.py

async def clear_alive(self, entry_ids: Iterable[bytes]) -> None:
    """Hand an entry over immediately instead of waiting out its TTL."""
    keys = [self.alive_key(entry_id) for entry_id in entry_ids]
    if keys:
        await self.control.delete(*keys)

dead_letter(record, *, reason, detail, times_delivered) async

Park an entry that will never succeed, keeping everything about it.

The original payload rides along untouched, so a fixed deployment can replay it without reconstructing anything.

Source code in src/smallage/core/transport.py

async def dead_letter(
    self, record: Record, *, reason: str, detail: str, times_delivered: int
) -> bytes:
    """Park an entry that will never succeed, keeping everything about it.

    The original payload rides along untouched, so a fixed deployment can
    replay it without reconstructing anything.
    """
    seconds, microseconds = await self.control.time()
    fields: dict[Any, Any] = dict(record.fields)
    fields[b"dlq_reason"] = reason.encode()
    fields[b"dlq_detail"] = detail.encode()[:MAX_DETAIL_BYTES]
    fields[b"dlq_source"] = record.stream.encode()
    fields[b"dlq_deliveries"] = str(times_delivered).encode()
    fields[b"dlq_at"] = str(seconds * 1000 + microseconds // 1000).encode()
    return _as_bytes(await self.control.xadd(self.dlq, fields))

is_external(stream)

A stream somebody else writes, in somebody else's format.

Source code in src/smallage/core/transport.py

def is_external(self, stream: str) -> bool:
    """A stream somebody else writes, in somebody else's format."""
    return stream in self.external_streams

lag() async

Depth from XINFO GROUPS. Redis reports NULL when it cannot reconcile.

Source code in src/smallage/core/transport.py

async def lag(self) -> int | None:
    """Depth from XINFO GROUPS. Redis reports NULL when it cannot reconcile."""
    total = 0
    for stream in (*self.streams, *self.external_streams):
        groups = await self.control.xinfo_groups(stream)
        for group in groups:
            if _as_bytes(_get(group, "name")) != self.group.encode():
                continue
            value = _get(group, "lag")
            if value is None:
                return None
            total += int(value)
    return total

refresh_alive(entry_ids, *, ttl_ms) async

PEXPIRE rather than SET: it must not resurrect an already acked entry.

Source code in src/smallage/core/transport.py

async def refresh_alive(self, entry_ids: Iterable[bytes], *, ttl_ms: int) -> None:
    """PEXPIRE rather than SET: it must not resurrect an already acked entry."""
    keys = [self.alive_key(entry_id) for entry_id in entry_ids]
    if not keys:
        return
    async with self.control.pipeline(transaction=False) as pipe:
        for key in keys:
            pipe.pexpire(key, ttl_ms)
        await pipe.execute()

sweep_consumers(*, min_idle_ms) async

Remove consumers that hold nothing and have gone quiet.

A name is derived per process start, so every deploy leaves one behind and Redis expires none of them. Deleting one that still owns pending entries would make them unclaimable, so the script checks and deletes in one step rather than trusting a reading taken a round trip earlier.

Deleting a live worker's idle consumer costs nothing -- Redis recreates it on the next read -- which is why the only hard condition is that it holds nothing.

Source code in src/smallage/core/transport.py

async def sweep_consumers(self, *, min_idle_ms: int) -> int:
    """Remove consumers that hold nothing and have gone quiet.

    A name is derived per process start, so every deploy leaves one behind
    and Redis expires none of them. Deleting one that still owns pending
    entries would make them unclaimable, so the script checks and deletes in
    one step rather than trusting a reading taken a round trip earlier.

    Deleting a live worker's idle consumer costs nothing -- Redis recreates
    it on the next read -- which is why the only hard condition is that it
    holds nothing.
    """
    deleted = 0
    for stream in (*self.streams, *self.external_streams):
        deleted += int(
            await self._scripts.sweep_consumers(
                keys=[stream], args=[self.group, min_idle_ms]
            )
        )
    return deleted

trim(*, retention_ms) async

Trim by time, floored at the oldest unacked entry.

MINID has the same hazard MAXLEN does once a pending entry is older than the retention window -- trimming it away would drop unacknowledged work.

Source code in src/smallage/core/transport.py

async def trim(self, *, retention_ms: int) -> None:
    """Trim by time, floored at the oldest unacked entry.

    MINID has the same hazard MAXLEN does once a pending entry is older than
    the retention window -- trimming it away would drop unacknowledged work.
    """
    seconds, microseconds = await self.control.time()
    now_ms = seconds * 1000 + microseconds // 1000
    floor_ms = now_ms - retention_ms
    for stream in self.streams:
        summary = await self.control.xpending(stream, self.group)
        oldest = summary.get("min") if summary else None
        limit = floor_ms if oldest is None else min(floor_ms, _entry_ms(oldest))
        await self.control.xtrim(stream, minid=limit, approximate=True)

ResultStore

Bases: Protocol

Where a job's outcome goes when somebody is waiting for one.

Source code in src/smallage/core/protocols.py

class ResultStore(Protocol):
    """Where a job's outcome goes when somebody is waiting for one."""

    async def store(
        self, job_id: str, result: TaskResult, *, ttl_ms: int | None = None
    ) -> None: ...
    async def get(self, job_id: str) -> TaskResult | None: ...

RetryPolicy dataclass

Source code in src/smallage/core/retry.py

@dataclass(frozen=True, slots=True)
class RetryPolicy:
    max_attempts: int = 3
    max_deliveries: int = 5
    initial_backoff_ms: int = 1_000
    max_backoff_ms: int = 300_000
    multiplier: float = 2.0
    jitter: float = 0.2
    unknown_task_timeout_ms: int = 6 * 60 * 60 * 1000
    unknown_task_backoff_ms: int = 30_000
    history_limit: int = 10
    history_line_bytes: int = 200

    def __post_init__(self) -> None:
        if self.max_attempts < 0:
            raise ConfigurationError(
                f"max_attempts must not be negative, got {self.max_attempts}"
            )
        if self.max_deliveries < 1:
            raise ConfigurationError(
                f"max_deliveries must be at least 1, got {self.max_deliveries}"
            )
        if self.initial_backoff_ms < 1:
            raise ConfigurationError(
                f"initial_backoff_ms must be positive, got {self.initial_backoff_ms}"
            )
        if self.max_backoff_ms < self.initial_backoff_ms:
            raise ConfigurationError(
                f"max_backoff_ms ({self.max_backoff_ms}) must not be below "
                f"initial_backoff_ms ({self.initial_backoff_ms})"
            )
        if self.multiplier < 1:
            raise ConfigurationError(
                f"multiplier must be at least 1, got {self.multiplier}"
            )
        if not 0 <= self.jitter <= 1:
            raise ConfigurationError(f"jitter must be within 0..1, got {self.jitter}")
        if self.unknown_task_timeout_ms < 0:
            raise ConfigurationError(
                "unknown_task_timeout_ms must not be negative, got "
                f"{self.unknown_task_timeout_ms}"
            )

    def delay_ms(self, attempt: int, *, rand: float = 0.5) -> int:
        """Backoff before retry number ``attempt``.

        ``rand`` is the caller's draw from [0, 1), kept as an argument so the
        arithmetic stays pure and the spread stays testable. Jitter matters:
        a batch that fails together would otherwise retry together forever.
        """
        base = min(
            self.initial_backoff_ms * self.multiplier**attempt,
            float(self.max_backoff_ms),
        )
        spread = base * self.jitter * (2 * rand - 1)
        return max(0, int(base + spread))

    def exhausted(self, attempt: int) -> bool:
        return attempt >= self.max_attempts

    def over_delivered(self, times_delivered: int) -> bool:
        return times_delivered > self.max_deliveries

    def record_failure(
        self, history: tuple[str, ...], attempt: int, error: BaseException
    ) -> tuple[str, ...]:
        """Append one line about this failure, keeping the tail bounded.

        ``attempt`` is the attempt that just failed, not the one about to run.

        Truncating and capping matters: without it a job that keeps failing grows
        its own stream entry on every attempt.
        """
        line = f"{attempt}: {type(error).__name__}: {error}".replace("\n", " ")
        return (*history, line[: self.history_line_bytes])[-self.history_limit :]

    def unknown_task_expired(self, *, age_ms: int) -> bool:
        """A rolling deploy is measured in minutes, so this threshold is time."""
        return age_ms > self.unknown_task_timeout_ms

delay_ms(attempt, *, rand=0.5)

Backoff before retry number attempt.

rand is the caller's draw from [0, 1), kept as an argument so the arithmetic stays pure and the spread stays testable. Jitter matters: a batch that fails together would otherwise retry together forever.

Source code in src/smallage/core/retry.py

def delay_ms(self, attempt: int, *, rand: float = 0.5) -> int:
    """Backoff before retry number ``attempt``.

    ``rand`` is the caller's draw from [0, 1), kept as an argument so the
    arithmetic stays pure and the spread stays testable. Jitter matters:
    a batch that fails together would otherwise retry together forever.
    """
    base = min(
        self.initial_backoff_ms * self.multiplier**attempt,
        float(self.max_backoff_ms),
    )
    spread = base * self.jitter * (2 * rand - 1)
    return max(0, int(base + spread))

record_failure(history, attempt, error)

Append one line about this failure, keeping the tail bounded.

attempt is the attempt that just failed, not the one about to run.

Truncating and capping matters: without it a job that keeps failing grows its own stream entry on every attempt.

Source code in src/smallage/core/retry.py

def record_failure(
    self, history: tuple[str, ...], attempt: int, error: BaseException
) -> tuple[str, ...]:
    """Append one line about this failure, keeping the tail bounded.

    ``attempt`` is the attempt that just failed, not the one about to run.

    Truncating and capping matters: without it a job that keeps failing grows
    its own stream entry on every attempt.
    """
    line = f"{attempt}: {type(error).__name__}: {error}".replace("\n", " ")
    return (*history, line[: self.history_line_bytes])[-self.history_limit :]

unknown_task_expired(*, age_ms)

A rolling deploy is measured in minutes, so this threshold is time.

Source code in src/smallage/core/retry.py

def unknown_task_expired(self, *, age_ms: int) -> bool:
    """A rolling deploy is measured in minutes, so this threshold is time."""
    return age_ms > self.unknown_task_timeout_ms

Scheduler

Bases: Protocol

What the worker needs from a scheduler, and nothing more.

Source code in src/smallage/core/protocols.py

class Scheduler(Protocol):
    """What the worker needs from a scheduler, and nothing more."""

    async def now_ms(self) -> int: ...
    async def schedule_at(
        self,
        envelope: Envelope,
        *,
        queue: str,
        when_ms: int,
        scheduled_id: str | None = None,
    ) -> str: ...
    async def hold_leadership(self, token: str, *, ttl_ms: int) -> bool: ...
    async def release_leadership(self, token: str) -> bool: ...
    async def schedule_cron(self, jobs: Sequence[CronJob]) -> list[str]: ...
    async def promote(self, *, limit: int = 100) -> list[bytes]: ...

SmallageError

Bases: Exception

Base of every error this library raises.

Source code in src/smallage/core/errors.py

class SmallageError(Exception):
    """Base of every error this library raises."""

StreamTransport

Bases: Protocol

Everything the worker needs from a transport.

The worker binds here rather than to a concrete transport, which is what lets the consume loop and the supervisor loops be tested without a Redis at all.

Source code in src/smallage/core/protocols.py

class StreamTransport(Protocol):
    """Everything the worker needs from a transport.

    The worker binds here rather than to a concrete transport, which is what lets
    the consume loop and the supervisor loops be tested without a Redis at all.
    """

    consumer: str
    namespace: str
    group: str
    queues: tuple[str, ...]

    def queue_of(self, stream: str) -> str: ...
    def is_external(self, stream: str) -> bool: ...
    async def ensure_group(self) -> None: ...
    async def read(self, count: int) -> list[Record]: ...
    async def mark_alive(self, entry_ids: Sequence[bytes], *, ttl_ms: int) -> None: ...
    async def refresh_alive(
        self, entry_ids: Iterable[bytes], *, ttl_ms: int
    ) -> None: ...
    async def clear_alive(self, entry_ids: Iterable[bytes]) -> None: ...
    async def claim_dedup(self, key: str, *, owner: str, ttl_ms: int) -> bool: ...
    async def complete_dedup(self, key: str, *, owner: str, ttl_ms: int) -> None: ...
    async def release_dedup(self, key: str, *, owner: str) -> None: ...
    async def ack(self, stream: str, entry_ids: Sequence[bytes]) -> int: ...
    async def pending(self, *, count: int, min_idle_ms: int) -> list[Pending]: ...
    async def dead_letter(
        self, record: Record, *, reason: str, detail: str, times_delivered: int
    ) -> bytes: ...
    async def reclaim(
        self, stream: str, entry_id: bytes, *, min_idle_ms: int, ttl_ms: int
    ) -> list[Record]: ...
    async def trim(self, *, retention_ms: int) -> None: ...
    async def sweep_consumers(self, *, min_idle_ms: int) -> int: ...
    async def lag(self) -> int | None: ...

TaskHandler

Bases: Protocol

Source code in src/smallage/core/protocols.py

class TaskHandler(Protocol):
    async def __call__(self, envelope: Envelope, /) -> object: ...

TaskResult

Bases: Struct

What a finished job left behind, for whoever asked to be told.

Source code in src/smallage/core/envelope.py

class TaskResult(msgspec.Struct, frozen=True):
    """What a finished job left behind, for whoever asked to be told."""

    ok: bool
    value: bytes = b""
    error: str = ""

UnknownTask

Bases: SmallageError

Eager mode was asked to run a task that is not registered.

Source code in src/smallage/core/testing.py

class UnknownTask(SmallageError):
    """Eager mode was asked to run a task that is not registered."""

WorkerConfig dataclass

Source code in src/smallage/core/worker.py

@dataclass(frozen=True, slots=True)
class WorkerConfig:
    concurrency: int = 10
    alive_ttl_ms: int = 30_000
    min_idle_ms: int = 60_000
    reclaim_interval_s: float = 5.0
    trim_interval_s: float = 60.0
    retention_ms: int = 24 * 60 * 60 * 1000
    consumer_idle_ms: int = 60 * 60 * 1000
    """How long a consumer may hold nothing and say nothing before it is presumed
    gone and removed from the group.

    A live worker touches the group at least every `block_ms`, so an hour is far
    above anything a running process produces. Deleting one early is harmless --
    Redis recreates it on the next read -- so the generous default costs nothing
    but a slower cleanup."""
    drain_timeout_s: float = 30.0
    retry: RetryPolicy = field(default_factory=RetryPolicy)
    dedup_ttl_ms: int = 24 * 60 * 60 * 1000
    leader_ttl_ms: int = 15_000
    scheduler_interval_s: float = 1.0
    promote_limit: int = 100
    recovery_interval_s: float = 1.0
    """Pause after a supervisor loop hits an error before it tries again."""

    def __post_init__(self) -> None:
        if self.concurrency < 1:
            raise ConfigurationError(
                f"concurrency must be at least 1, got {self.concurrency}"
            )
        if self.alive_ttl_ms <= 0:
            raise ConfigurationError(
                f"alive_ttl_ms must be positive, got {self.alive_ttl_ms}"
            )
        if self.drain_timeout_s < 0:
            raise ConfigurationError(
                f"drain_timeout_s must not be negative, got {self.drain_timeout_s}"
            )
        if self.leader_ttl_ms <= self.scheduler_interval_s * 1000:
            raise ConfigurationError(
                f"leader_ttl_ms ({self.leader_ttl_ms}) must outlast "
                f"scheduler_interval_s ({self.scheduler_interval_s}s), or the lease "
                "lapses between passes and leadership flaps"
            )

    @property
    def refresh_interval_s(self) -> float:
        return self.alive_ttl_ms / 3 / 1000

consumer_idle_ms = 60 * 60 * 1000 class-attribute instance-attribute

How long a consumer may hold nothing and say nothing before it is presumed gone and removed from the group.

A live worker touches the group at least every block_ms, so an hour is far above anything a running process produces. Deleting one early is harmless -- Redis recreates it on the next read -- so the generous default costs nothing but a slower cleanup.

recovery_interval_s = 1.0 class-attribute instance-attribute

Pause after a supervisor loop hits an error before it tries again.

WorkerStats dataclass

Live counters for one worker process.

Plain integers on one event loop: incremented between awaits, read whenever.

Source code in src/smallage/core/stats.py

@dataclass(slots=True)
class WorkerStats:
    """Live counters for one worker process.

    Plain integers on one event loop: incremented between awaits, read whenever.
    """

    handled: int = 0
    failed: int = 0
    retried: int = 0
    buried: int = 0
    unknown_task: int = 0
    reclaimed: int = 0
    skipped_duplicate: int = 0
    in_flight: int = 0
    buried_by_reason: dict[str, int] = field(default_factory=dict)

    def bury(self, reason: str) -> None:
        self.buried += 1
        self.buried_by_reason[reason] = self.buried_by_reason.get(reason, 0) + 1

    def snapshot(self) -> dict[str, int]:
        return {
            "handled": self.handled,
            "failed": self.failed,
            "retried": self.retried,
            "buried": self.buried,
            "unknown_task": self.unknown_task,
            "reclaimed": self.reclaimed,
            "skipped_duplicate": self.skipped_duplicate,
            "in_flight": self.in_flight,
        }

dlq_key(namespace)

Exported: reading the dead letter queue is an ordinary thing to want, and an application should not have to rebuild the key schema to do it.

Source code in src/smallage/core/keys.py

def dlq_key(namespace: str) -> str:
    """Exported: reading the dead letter queue is an ordinary thing to want, and
    an application should not have to rebuild the key schema to do it."""
    return f"{{{namespace}}}:dlq"

from_fields(fields)

Source code in src/smallage/core/envelope.py

def from_fields(fields: Mapping[bytes, bytes]) -> Envelope:
    for name in _REQUIRED:
        if name not in fields:
            raise MalformedEnvelope(f"stream entry is missing field {name.decode()!r}")
    if fields[b"v"] != ENVELOPE_VERSION:
        raise MalformedEnvelope(
            f"unsupported envelope version {fields[b'v']!r}, "
            f"expected {ENVELOPE_VERSION!r}"
        )
    optional = {
        name: fields[key].decode()
        for name, key in (
            ("payload_ref", b"payload_ref"),
            ("traceparent", b"traceparent"),
            ("tracestate", b"tracestate"),
            ("dedup", b"dedup"),
        )
        if key in fields
    }
    # Unknown fields are ignored on purpose: a newer producer must not break an
    # older worker during a rolling deploy.
    raw_ttl = fields.get(b"result_ttl_ms")
    raw_history = fields.get(b"history")
    return Envelope(
        result_ttl_ms=int(raw_ttl) if raw_ttl is not None else None,
        history=tuple(raw_history.decode().split("\n")) if raw_history else (),
        id=fields[b"id"].decode(),
        task=fields[b"task"].decode(),
        payload=fields[b"payload"],
        enqueued_at=int(fields[b"enqueued_at"]),
        attempt=int(fields[b"attempt"]),
        **optional,
    )

run(transport, registry, config=None, *, scheduler, results=None, brokers=None, stats=None, sleep=anyio.sleep, shutdown=None, cron=()) async

Source code in src/smallage/core/worker.py

async def run(
    transport: StreamTransport,
    registry: Mapping[str, TaskHandler],
    config: WorkerConfig | None = None,
    *,
    scheduler: Scheduler,
    results: ResultStore | None = None,
    brokers: Mapping[str, BrokerHandler] | None = None,
    stats: WorkerStats | None = None,
    sleep: Sleeper = anyio.sleep,
    shutdown: anyio.Event | None = None,
    cron: Sequence[CronJob] = (),
) -> None:
    cfg = config or WorkerConfig()
    stop = shutdown or anyio.Event()
    slots = Slots()
    counters = stats or WorkerStats()

    await transport.ensure_group()
    slots.recoverable = await _orphans_of_previous_run(transport, cfg)

    async with anyio.create_task_group() as supervisors:
        supervisors.start_soon(_heartbeat_loop, transport, slots, cfg, sleep)
        supervisors.start_soon(_trim_loop, transport, cfg, sleep)
        supervisors.start_soon(_scheduler_loop, scheduler, cron, cfg, sleep)

        reclaiming = anyio.CancelScope()

        async with anyio.create_task_group() as handlers:

            def spawn(record: Record) -> None:
                handlers.start_soon(
                    _run_one,
                    transport,
                    scheduler,
                    results,
                    registry,
                    brokers or {},
                    slots,
                    cfg,
                    counters,
                    record,
                )

            async def reclaim() -> None:
                with reclaiming:
                    await _reclaim_loop(transport, slots, cfg, sleep, spawn, counters)

            handlers.start_soon(reclaim)
            await _consume_loop(transport, slots, cfg, stop, spawn)

            # Shutdown: stop pulling in new work, let in-flight handlers finish,
            # cut them off when the drain budget runs out. The clock starts only
            # here, so the deadline covers draining and nothing else.
            reclaiming.cancel()
            handlers.cancel_scope.deadline = anyio.current_time() + cfg.drain_timeout_s

        supervisors.cancel_scope.cancel()

run_with_signals(transport, registry, config=None, *, scheduler, results=None, brokers=None, stats=None, cron=()) async

Run until SIGTERM or SIGINT; a second signal means now, not soon.

Source code in src/smallage/core/worker.py

async def run_with_signals(
    transport: StreamTransport,
    registry: Mapping[str, TaskHandler],
    config: WorkerConfig | None = None,
    *,
    scheduler: Scheduler,
    results: ResultStore | None = None,
    brokers: Mapping[str, BrokerHandler] | None = None,
    stats: WorkerStats | None = None,
    cron: Sequence[CronJob] = (),
) -> None:
    """Run until SIGTERM or SIGINT; a second signal means now, not soon."""
    stop = anyio.Event()
    async with anyio.create_task_group() as tg:
        tg.start_soon(_watch_signals, stop, tg.cancel_scope)
        await run(
            transport,
            registry,
            config,
            scheduler=scheduler,
            results=results,
            brokers=brokers,
            stats=stats,
            cron=cron,
            shutdown=stop,
        )
        tg.cancel_scope.cancel()

to_fields(envelope)

Source code in src/smallage/core/envelope.py

def to_fields(envelope: Envelope) -> dict[bytes, bytes]:
    fields = {
        b"v": ENVELOPE_VERSION,
        b"id": envelope.id.encode(),
        b"task": envelope.task.encode(),
        b"payload": envelope.payload,
        b"enqueued_at": str(envelope.enqueued_at).encode(),
        b"attempt": str(envelope.attempt).encode(),
    }
    optional = {
        b"payload_ref": envelope.payload_ref,
        b"traceparent": envelope.traceparent,
        b"tracestate": envelope.tracestate,
        b"dedup": envelope.dedup,
    }
    if envelope.result_ttl_ms is not None:
        fields[b"result_ttl_ms"] = str(envelope.result_ttl_ms).encode()
    # Redis has no null: an unset field is written as no field at all.
    fields.update({k: v.encode() for k, v in optional.items() if v is not None})
    if envelope.history:
        fields[b"history"] = "\n".join(envelope.history).encode()
    return fields

worker_running(transport, registry, config=None, *, scheduler, results=None, brokers=None, stats=None, cron=()) async

Run a real worker for the duration of the block, then stop it cleanly.

Wrap it in a fixture for integration tests: leaving the block asks the worker to drain, so in-flight work finishes instead of vanishing mid-assertion.

Source code in src/smallage/core/testing.py

@asynccontextmanager
async def worker_running(
    transport: StreamTransport,
    registry: Mapping[str, TaskHandler],
    config: WorkerConfig | None = None,
    *,
    scheduler: Scheduler,
    results: ResultStore | None = None,
    brokers: Mapping[str, BrokerHandler] | None = None,
    stats: WorkerStats | None = None,
    cron: Sequence[CronJob] = (),
) -> AsyncGenerator[None]:
    """Run a real worker for the duration of the block, then stop it cleanly.

    Wrap it in a fixture for integration tests: leaving the block asks the worker
    to drain, so in-flight work finishes instead of vanishing mid-assertion.
    """
    stop = anyio.Event()
    async with anyio.create_task_group() as tg:
        tg.start_soon(
            partial(
                run,
                transport,
                registry,
                config,
                scheduler=scheduler,
                results=results,
                brokers=brokers,
                stats=stats,
                cron=cron,
                shutdown=stop,
            )
        )
        try:
            yield
        finally:
            stop.set()

Litestar plugin

smallage.litestar

Litestar integration layer: DI, CLI, serialization, health.

__all__ = ['QueueConfig', 'QueueHealth', 'QueuePlugin', 'Task', 'TaskRegistry', 'current_traceparent', 'queue_health'] module-attribute

current_traceparent = ContextVar('smallage_traceparent', default=None) module-attribute

The traceparent of the job being executed, for the duration of the handler.

Read it to attach the task's span to the request that enqueued it. It is a context variable rather than an argument so that a task signature stays about the task.

QueueConfig dataclass

How this application talks to its queue.

There is deliberately no string path to the app object here. Requiring "module:app" in a constructor pushes an import-time ordering problem onto every user; the CLI already has the application it was invoked with.

Keyword-only: fields are grouped by what they configure rather than appended, so a new one lands mid-class and positional order was never a contract worth keeping.

Source code in src/smallage/litestar/config.py

@dataclass(frozen=True, slots=True, kw_only=True)
class QueueConfig:
    """How this application talks to its queue.

    There is deliberately no string path to the app object here. Requiring
    ``"module:app"`` in a constructor pushes an import-time ordering problem onto
    every user; the CLI already has the application it was invoked with.

    Keyword-only: fields are grouped by what they configure rather than appended,
    so a new one lands mid-class and positional order was never a contract worth
    keeping.
    """

    registry: TaskRegistry
    redis_url: str = "redis://localhost:6379/0"
    namespace: str = "lrs"
    queues: Sequence[str] = ("default",)
    shards: int = 1
    group: str = "workers"
    consumer_prefix: str = "worker"
    consumer: str | None = None
    """The worker's consumer name, or None to derive one from `consumer_prefix`.

    Must be unique per running worker: Redis groups pending entries by consumer
    name, and reclaim uses that name to tell a worker's own work from a peer's.
    A literal here is shared by every replica; take it from the environment.
    `--consumer` sets this, which is how the name survives being settled before
    the application's lifespan opens the queue.
    """
    block_ms: int = 5_000
    worker: WorkerConfig = field(default_factory=WorkerConfig)
    run_app_lifespan: bool = True
    """Whether a worker process enters the application's lifespan.

    On by default, so a dependency that closes over something the lifespan opens
    resolves to an opened one in a worker as well as in a web process. Decline it
    when the lifespan does work that belongs to a web process -- starting a
    scheduler, warming a cache, claiming a lease -- which would otherwise happen
    in every worker replica.
    """
    cron: Sequence[CronJob] = ()
    result_ttl_ms: int = 3_600_000
    fairness_every: int = 10
    """Passes between giving the lowest-priority queue first refusal. 0 for
    strict priority, and the starvation that comes with it."""
    max_payload_bytes: int = 128 * 1024
    """Refused above this when no payload store is configured."""
    payloads: PayloadStore | None = None
    """Where arguments above `offload_over_bytes` go. Without one, the transport
    refuses an oversized record rather than dropping it quietly."""
    thread_limit: int = 20
    """Threads for synchronous tasks. The asyncio default is a silent bottleneck."""
    offload_over_bytes: int = 128 * 1024
    brokers: Mapping[str, BrokerHandler] = field(default_factory=dict)
    """Streams somebody else writes, and what handles each of them.

    One field rather than a list of streams beside a dictionary of handlers: a
    stream with no handler is read and dropped, a handler for a stream nobody
    subscribes to never runs, and two fields that must agree eventually will not.
    """

    @property
    def external(self) -> tuple[str, ...]:
        return tuple(self.brokers)

    traceparent: TraceparentSource = no_traceparent
    """Supplies the current W3C traceparent, so a task can join the request's trace."""

    def __post_init__(self) -> None:
        if not self.redis_url:
            raise ConfigurationError("redis_url must not be empty")

brokers = field(default_factory=dict) class-attribute instance-attribute

Streams somebody else writes, and what handles each of them.

One field rather than a list of streams beside a dictionary of handlers: a stream with no handler is read and dropped, a handler for a stream nobody subscribes to never runs, and two fields that must agree eventually will not.

consumer = None class-attribute instance-attribute

The worker's consumer name, or None to derive one from consumer_prefix.

Must be unique per running worker: Redis groups pending entries by consumer name, and reclaim uses that name to tell a worker's own work from a peer's. A literal here is shared by every replica; take it from the environment. --consumer sets this, which is how the name survives being settled before the application's lifespan opens the queue.

fairness_every = 10 class-attribute instance-attribute

Passes between giving the lowest-priority queue first refusal. 0 for strict priority, and the starvation that comes with it.

max_payload_bytes = 128 * 1024 class-attribute instance-attribute

Refused above this when no payload store is configured.

payloads = None class-attribute instance-attribute

Where arguments above offload_over_bytes go. Without one, the transport refuses an oversized record rather than dropping it quietly.

run_app_lifespan = True class-attribute instance-attribute

Whether a worker process enters the application's lifespan.

On by default, so a dependency that closes over something the lifespan opens resolves to an opened one in a worker as well as in a web process. Decline it when the lifespan does work that belongs to a web process -- starting a scheduler, warming a cache, claiming a lease -- which would otherwise happen in every worker replica.

thread_limit = 20 class-attribute instance-attribute

Threads for synchronous tasks. The asyncio default is a silent bottleneck.

traceparent = no_traceparent class-attribute instance-attribute

Supplies the current W3C traceparent, so a task can join the request's trace.

QueueHealth dataclass

Source code in src/smallage/litestar/health.py

@dataclass(frozen=True, slots=True)
class QueueHealth:
    namespace: str
    group: str
    queues: tuple[str, ...]
    lag: int | None
    """Depth from XINFO GROUPS. None when Redis cannot reconcile its counters."""
    stats: dict[str, int]
    """This process's own counters. All zero in a web process, which runs no tasks."""
    healthy: bool
    """A real field, not a property: a probe reads the response, not the object."""

healthy instance-attribute

A real field, not a property: a probe reads the response, not the object.

lag instance-attribute

Depth from XINFO GROUPS. None when Redis cannot reconcile its counters.

stats instance-attribute

This process's own counters. All zero in a web process, which runs no tasks.

QueuePlugin

Bases: InitPlugin, CLIPlugin

Wires the queue into an application, and the application into the CLI.

The plugin is itself the enqueuer the registry binds to, which is how tasks can be declared at import time while their Redis connections are opened by the application lifespan.

Source code in src/smallage/litestar/plugin.py

class QueuePlugin(InitPlugin, CLIPlugin):
    """Wires the queue into an application, and the application into the CLI.

    The plugin is itself the enqueuer the registry binds to, which is how tasks
    can be declared at import time while their Redis connections are opened by
    the application lifespan.
    """

    def __init__(self, config: QueueConfig) -> None:
        self.config = config
        self._transport: RedisStreamsTransport | None = None
        self._scheduler: RedisScheduler | None = None
        self._results: RedisResultStore | None = None
        self.stats = WorkerStats()
        """Counters for this process, served by the health endpoint."""

    @property
    def transport(self) -> RedisStreamsTransport:
        if self._transport is None:
            raise ConfigurationError(
                "the queue is not connected yet; it opens with the application"
            )
        return self._transport

    @property
    def scheduler(self) -> RedisScheduler:
        if self._scheduler is None:
            raise ConfigurationError(
                "the queue is not connected yet; it opens with the application"
            )
        return self._scheduler

    @property
    def results(self) -> RedisResultStore:
        if self._results is None:
            raise ConfigurationError(
                "the queue is not connected yet; it opens with the application"
            )
        return self._results

    async def enqueue(self, envelope: Envelope, *, queue: str) -> bytes:
        return await self.transport.enqueue(envelope, queue=queue)

    async def store(
        self, job_id: str, result: TaskResult, *, ttl_ms: int | None = None
    ) -> None:
        await self.results.store(job_id, result, ttl_ms=ttl_ms)

    async def get(self, job_id: str) -> TaskResult | None:
        return await self.results.get(job_id)

    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        self.config.registry.bind(
            _providers(app_config.dependencies),
            enqueuer=self,
            type_encoders=app_config.type_encoders,
            type_decoders=app_config.type_decoders,
            traceparent=self.config.traceparent,
            results=self,
            payloads=self.config.payloads,
            thread_limit=self.config.thread_limit,
            offload_over_bytes=self.config.offload_over_bytes,
        )
        # First, so the queue is the outermost manager: everything the
        # application opens after it can enqueue, and can still enqueue while
        # unwinding. Appended, it would close before their exits ran.
        app_config.lifespan.insert(0, self._lifespan)
        return app_config

    def on_cli_init(self, cli: Group) -> None:
        from smallage.litestar.cli import workers

        cli.add_command(workers)

    @asynccontextmanager
    async def _lifespan(self, app: Litestar) -> AsyncGenerator[None]:
        async with self.connected():
            yield

    @asynccontextmanager
    async def connected(self, *, consumer: str | None = None) -> AsyncGenerator[None]:
        """Open the queue's connections for the duration of the block.

        Used by the application lifespan and by the worker command alike, so a
        worker and a web process reach Redis in exactly the same shape.

        The consumer name comes from `config.consumer` unless a caller opening
        the queue itself passes one: a worker that enters the application's
        lifespan never opens the queue, so it has nowhere to pass an argument.
        """
        config = self.config
        # The reader owns its connection for a whole block window; a socket
        # timeout below that would kill every healthy read, above it is the only
        # thing that surfaces a connection hung by a failover.
        reader: Redis = Redis.from_url(
            config.redis_url, socket_timeout=config.block_ms / 1000 + 30
        )
        control: Redis = Redis.from_url(config.redis_url)
        try:
            self._transport = RedisStreamsTransport(
                reader=reader,
                control=control,
                consumer=_consumer_name(config, consumer),
                namespace=config.namespace,
                queues=config.queues,
                shards=config.shards,
                group=config.group,
                block_ms=config.block_ms,
                fairness_every=config.fairness_every,
                max_payload_bytes=config.max_payload_bytes,
                external=config.external,
            )
            self._scheduler = RedisScheduler(
                control=control, namespace=config.namespace, shards=config.shards
            )
            self._results = RedisResultStore(
                control=control,
                namespace=config.namespace,
                default_ttl_ms=config.result_ttl_ms,
            )
            await self._transport.ensure_group()
            yield
        finally:
            self._transport = None
            self._scheduler = None
            self._results = None
            await reader.aclose()
            await control.aclose()

    async def health(self) -> QueueHealth:
        """Everything a readiness probe needs, for whoever is serving one.

        The plugin registers no route of its own: which path a probe lives on,
        whether it is public, and what else belongs in the same response are the
        application's to decide, and a worker has no server to answer on at all.
        `QueueHealth.healthy` is the field to turn into a status code -- a probe
        reads the code, not the body, and answering 200 while unhealthy is a
        probe that can never fail.
        """
        return await queue_health(self.transport, self.stats)

stats = WorkerStats() instance-attribute

Counters for this process, served by the health endpoint.

connected(*, consumer=None) async

Open the queue's connections for the duration of the block.

Used by the application lifespan and by the worker command alike, so a worker and a web process reach Redis in exactly the same shape.

The consumer name comes from config.consumer unless a caller opening the queue itself passes one: a worker that enters the application's lifespan never opens the queue, so it has nowhere to pass an argument.

Source code in src/smallage/litestar/plugin.py

@asynccontextmanager
async def connected(self, *, consumer: str | None = None) -> AsyncGenerator[None]:
    """Open the queue's connections for the duration of the block.

    Used by the application lifespan and by the worker command alike, so a
    worker and a web process reach Redis in exactly the same shape.

    The consumer name comes from `config.consumer` unless a caller opening
    the queue itself passes one: a worker that enters the application's
    lifespan never opens the queue, so it has nowhere to pass an argument.
    """
    config = self.config
    # The reader owns its connection for a whole block window; a socket
    # timeout below that would kill every healthy read, above it is the only
    # thing that surfaces a connection hung by a failover.
    reader: Redis = Redis.from_url(
        config.redis_url, socket_timeout=config.block_ms / 1000 + 30
    )
    control: Redis = Redis.from_url(config.redis_url)
    try:
        self._transport = RedisStreamsTransport(
            reader=reader,
            control=control,
            consumer=_consumer_name(config, consumer),
            namespace=config.namespace,
            queues=config.queues,
            shards=config.shards,
            group=config.group,
            block_ms=config.block_ms,
            fairness_every=config.fairness_every,
            max_payload_bytes=config.max_payload_bytes,
            external=config.external,
        )
        self._scheduler = RedisScheduler(
            control=control, namespace=config.namespace, shards=config.shards
        )
        self._results = RedisResultStore(
            control=control,
            namespace=config.namespace,
            default_ttl_ms=config.result_ttl_ms,
        )
        await self._transport.ensure_group()
        yield
    finally:
        self._transport = None
        self._scheduler = None
        self._results = None
        await reader.aclose()
        await control.aclose()

health() async

Everything a readiness probe needs, for whoever is serving one.

The plugin registers no route of its own: which path a probe lives on, whether it is public, and what else belongs in the same response are the application's to decide, and a worker has no server to answer on at all. QueueHealth.healthy is the field to turn into a status code -- a probe reads the code, not the body, and answering 200 while unhealthy is a probe that can never fail.

Source code in src/smallage/litestar/plugin.py

async def health(self) -> QueueHealth:
    """Everything a readiness probe needs, for whoever is serving one.

    The plugin registers no route of its own: which path a probe lives on,
    whether it is public, and what else belongs in the same response are the
    application's to decide, and a worker has no server to answer on at all.
    `QueueHealth.healthy` is the field to turn into a status code -- a probe
    reads the code, not the body, and answering 200 while unhealthy is a
    probe that can never fail.
    """
    return await queue_health(self.transport, self.stats)

Task

What @registry.task gives back.

Calling it runs the function directly, with its own signature intact, which is what keeps type checking useful. enqueue takes the payload parameters only -- the injected ones come from the application at execution time.

Source code in src/smallage/litestar/registry.py

class Task[**P]:
    """What ``@registry.task`` gives back.

    Calling it runs the function directly, with its own signature intact, which
    is what keeps type checking useful. ``enqueue`` takes the payload parameters
    only -- the injected ones come from the application at execution time.
    """

    def __init__(
        self,
        registry: TaskRegistry,
        name: str,
        queue: str,
        function: Any,
        timeout_s: float | None = None,
    ) -> None:
        self.registry = registry
        self.name = name
        self.queue = queue
        self.function = function
        self.timeout_s = timeout_s
        self.__doc__ = function.__doc__
        self.__name__ = getattr(function, "__name__", name)

    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        if inspect.iscoroutinefunction(self.function):
            return await self.function(*args, **kwargs)
        return self.function(*args, **kwargs)

    async def enqueue(
        self,
        *,
        result_ttl_ms: int | None = None,
        dedup: str | None = None,
        **payload: Any,
    ) -> str:
        """Queue the task and hand back its job id.

        The id is what a result is looked up by, so it is returned even when
        nobody asked for one.
        """
        return await self.registry.enqueue(
            self.name, payload, result_ttl_ms=result_ttl_ms, dedup=dedup
        )

enqueue(*, result_ttl_ms=None, dedup=None, **payload) async

Queue the task and hand back its job id.

The id is what a result is looked up by, so it is returned even when nobody asked for one.

Source code in src/smallage/litestar/registry.py

async def enqueue(
    self,
    *,
    result_ttl_ms: int | None = None,
    dedup: str | None = None,
    **payload: Any,
) -> str:
    """Queue the task and hand back its job id.

    The id is what a result is looked up by, so it is returned even when
    nobody asked for one.
    """
    return await self.registry.enqueue(
        self.name, payload, result_ttl_ms=result_ttl_ms, dedup=dedup
    )

TaskRegistry

Tasks an application declares, and their execution once bound to it.

Source code in src/smallage/litestar/registry.py

class TaskRegistry:
    """Tasks an application declares, and their execution once bound to it."""

    def __init__(self) -> None:
        self._declared: dict[str, Task[Any]] = {}
        self._bound: dict[str, BoundTask] = {}
        self._enqueuer: Enqueuer | None = None
        self._traceparent: TraceparentSource = no_traceparent
        self._results: ResultStore | None = None
        self._payloads: PayloadStore | None = None
        self._threads = anyio.CapacityLimiter(DEFAULT_THREAD_LIMIT)
        self._offload_over_bytes = DEFAULT_OFFLOAD_OVER_BYTES
        self._serializer: Callable[[Any], Any] | None = None
        self._decoders: TypeDecodersSequence | None = None

    @overload
    def task[**P](self, function: Callable[P, Any], /) -> Task[P]: ...

    @overload
    def task[**P](
        self,
        /,
        *,
        name: str | None = None,
        queue: str = DEFAULT_QUEUE,
        timeout_s: float | None = None,
    ) -> Callable[[Callable[P, Any]], Task[P]]: ...

    def task[**P](
        self,
        function: Callable[P, Any] | None = None,
        /,
        *,
        name: str | None = None,
        queue: str = DEFAULT_QUEUE,
        timeout_s: float | None = None,
    ) -> Task[P] | Callable[[Callable[P, Any]], Task[P]]:
        def register(fn: Callable[P, Any]) -> Task[P]:
            task_name = name or fn.__name__
            if task_name in self._declared:
                raise ConfigurationError(f"task {task_name!r} is registered twice")
            declared = Task[P](self, task_name, queue, fn, timeout_s)
            self._declared[task_name] = declared
            return declared

        return register if function is None else register(function)

    @property
    def names(self) -> tuple[str, ...]:
        return tuple(self._declared)

    def bind(
        self,
        dependencies: Mapping[str, Provide],
        *,
        enqueuer: Enqueuer,
        type_encoders: TypeEncodersMap | None = None,
        type_decoders: TypeDecodersSequence | None = None,
        traceparent: TraceparentSource = no_traceparent,
        results: ResultStore | None = None,
        payloads: PayloadStore | None = None,
        thread_limit: int = DEFAULT_THREAD_LIMIT,
        offload_over_bytes: int = DEFAULT_OFFLOAD_OVER_BYTES,
    ) -> None:
        """Settle every task against the application. Failures land here, on boot."""
        self._enqueuer = enqueuer
        self._traceparent = traceparent
        self._results = results
        self._payloads = payloads
        self._threads = anyio.CapacityLimiter(thread_limit)
        self._offload_over_bytes = offload_over_bytes
        self._serializer = get_serializer(type_encoders)
        self._decoders = type_decoders
        self._bound = {
            name: _bind(task, dependencies) for name, task in self._declared.items()
        }

    def bound(self, name: str) -> BoundTask:
        task = self._bound.get(name)
        if task is None:
            raise ConfigurationError(
                f"task {name!r} is not bound; the plugin binds tasks at app startup"
            )
        return task

    async def result(self, job_id: str) -> TaskResult | None:
        if self._results is None:
            raise ConfigurationError("this application has no result store configured")
        return await self._results.get(job_id)

    async def enqueue(
        self,
        name: str,
        payload: Mapping[str, Any],
        *,
        result_ttl_ms: int | None = None,
        dedup: str | None = None,
    ) -> str:
        task = self.bound(name)
        # A unit of work may have bound its own, so that publication can wait for
        # the transaction that justified it.
        enqueuer = current_enqueuer.get() or self._enqueuer
        if enqueuer is None:  # pragma: no cover - bind() always sets it
            raise ConfigurationError("registry has no enqueuer")
        # Building the payload struct here is what makes a bad argument a caller's
        # problem rather than a worker's.
        try:
            arguments = task.payload_type(**payload)
        except TypeError as exc:
            raise ConfigurationError(f"bad arguments for task {name!r}: {exc}") from exc
        job_id = uuid4().hex
        encoded = encode_json(arguments, self._serializer)
        reference: str | None = None
        if len(encoded) > self._offload_over_bytes and self._payloads is not None:
            # Redis keeps the stream in memory; a large payload belongs elsewhere.
            reference = await self._payloads.put(job_id, encoded)
            encoded = b""
        envelope = Envelope(
            id=job_id,
            task=name,
            payload=encoded,
            payload_ref=reference,
            enqueued_at=time.time_ns() // 1_000_000,
            # Beside the payload, never inside it: restoring the span must not
            # require decoding a payload the worker may not understand.
            traceparent=self._traceparent(),
            result_ttl_ms=result_ttl_ms,
            dedup=dedup,
        )
        await enqueuer.enqueue(envelope, queue=task.queue)
        return job_id

    async def execute(self, envelope: Envelope) -> Any:
        """Run a task from its stream entry: decode, inject, call."""
        task = self.bound(envelope.task)
        raw = envelope.payload
        if envelope.payload_ref is not None:
            if self._payloads is None:
                raise ConfigurationError(
                    f"job {envelope.id!r} carries a payload reference but this "
                    "application has no payload store"
                )
            raw = await self._payloads.get(envelope.payload_ref)
        arguments = decode_json(raw, task.payload_type, self._decoders)
        payload = {field: getattr(arguments, field) for field in task.payload_fields}
        token = current_traceparent.set(envelope.traceparent)
        try:
            async with resolved(task.plan) as resolved_values:
                given = {name: resolved_values[name] for name in task.injected}
                return await self._call(task, {**payload, **given})
        finally:
            current_traceparent.reset(token)

    async def _call(self, task: BoundTask, arguments: dict[str, Any]) -> Any:
        if not task.is_async:
            # Off the event loop, under a limiter this project sizes itself: the
            # asyncio default silently caps throughput and starves the heartbeat.
            return await anyio.to_thread.run_sync(
                partial(task.function, **arguments), limiter=self._threads
            )
        if task.timeout_s is None:
            return await task.function(**arguments)
        with anyio.fail_after(task.timeout_s):
            return await task.function(**arguments)

    def handlers(self) -> dict[str, TaskHandler]:
        """What the worker dispatches on: one entry per bound task."""
        return dict.fromkeys(self._bound, self.execute)

bind(dependencies, *, enqueuer, type_encoders=None, type_decoders=None, traceparent=no_traceparent, results=None, payloads=None, thread_limit=DEFAULT_THREAD_LIMIT, offload_over_bytes=DEFAULT_OFFLOAD_OVER_BYTES)

Settle every task against the application. Failures land here, on boot.

Source code in src/smallage/litestar/registry.py

def bind(
    self,
    dependencies: Mapping[str, Provide],
    *,
    enqueuer: Enqueuer,
    type_encoders: TypeEncodersMap | None = None,
    type_decoders: TypeDecodersSequence | None = None,
    traceparent: TraceparentSource = no_traceparent,
    results: ResultStore | None = None,
    payloads: PayloadStore | None = None,
    thread_limit: int = DEFAULT_THREAD_LIMIT,
    offload_over_bytes: int = DEFAULT_OFFLOAD_OVER_BYTES,
) -> None:
    """Settle every task against the application. Failures land here, on boot."""
    self._enqueuer = enqueuer
    self._traceparent = traceparent
    self._results = results
    self._payloads = payloads
    self._threads = anyio.CapacityLimiter(thread_limit)
    self._offload_over_bytes = offload_over_bytes
    self._serializer = get_serializer(type_encoders)
    self._decoders = type_decoders
    self._bound = {
        name: _bind(task, dependencies) for name, task in self._declared.items()
    }

execute(envelope) async

Run a task from its stream entry: decode, inject, call.

Source code in src/smallage/litestar/registry.py

async def execute(self, envelope: Envelope) -> Any:
    """Run a task from its stream entry: decode, inject, call."""
    task = self.bound(envelope.task)
    raw = envelope.payload
    if envelope.payload_ref is not None:
        if self._payloads is None:
            raise ConfigurationError(
                f"job {envelope.id!r} carries a payload reference but this "
                "application has no payload store"
            )
        raw = await self._payloads.get(envelope.payload_ref)
    arguments = decode_json(raw, task.payload_type, self._decoders)
    payload = {field: getattr(arguments, field) for field in task.payload_fields}
    token = current_traceparent.set(envelope.traceparent)
    try:
        async with resolved(task.plan) as resolved_values:
            given = {name: resolved_values[name] for name in task.injected}
            return await self._call(task, {**payload, **given})
    finally:
        current_traceparent.reset(token)

handlers()

What the worker dispatches on: one entry per bound task.

Source code in src/smallage/litestar/registry.py

def handlers(self) -> dict[str, TaskHandler]:
    """What the worker dispatches on: one entry per bound task."""
    return dict.fromkeys(self._bound, self.execute)

queue_health(transport, stats=None) async

Source code in src/smallage/litestar/health.py

async def queue_health(
    transport: StreamTransport, stats: WorkerStats | None = None
) -> QueueHealth:
    lag = await transport.lag()
    return QueueHealth(
        namespace=transport.namespace,
        group=transport.group,
        queues=tuple(transport.queues),
        lag=lag,
        stats=(stats or WorkerStats()).snapshot(),
        healthy=lag is not None,
    )