Work Queue API

Brokerless work queues: a QueueRole owner (enqueue / claim / ack / nack + lease sweeper) and the competing-consumer worker loop. QueueStore holds the job state in memory and writes it through to the app's StoragePlugin.

Guide: Work Queues.

Work queues: jobs, not events. See docs/user-guide/work-queues.md.

CronError

Bases: ValueError

Raised for a malformed cron expression.

Source code in src/istos/queue/cron.py

class CronError(ValueError):
    """Raised for a malformed cron expression."""

CronSchedule

A parsed cron expression that can compute its next firing time.

Source code in src/istos/queue/cron.py

class CronSchedule:
    """A parsed cron expression that can compute its next firing time."""

    def __init__(self, expr: str) -> None:
        fields = expr.split()
        if len(fields) != 5:
            raise CronError(
                f"cron expression needs 5 fields (min hour dom month dow), got {len(fields)}"
            )
        self.expr = expr
        self.minute = _parse_field(fields[0], *_RANGES[0])
        self.hour = _parse_field(fields[1], *_RANGES[1])
        self.dom = _parse_field(fields[2], *_RANGES[2])
        self.month = _parse_field(fields[3], *_RANGES[3])
        dow = _parse_field(fields[4].replace("7", "0") if fields[4] == "7" else fields[4], *_RANGES[4])
        if 7 in dow:  # 7 is an alias for Sunday
            dow.discard(7)
            dow.add(0)
        self.dow = dow
        self._dom_restricted = fields[2] != "*"
        self._dow_restricted = fields[4] != "*"

    def _matches(self, t: _dt.datetime) -> bool:
        if t.minute not in self.minute or t.hour not in self.hour or t.month not in self.month:
            return False
        cron_dow = (t.weekday() + 1) % 7  # Python Mon=0..Sun=6 → cron Sun=0..Sat=6
        dom_ok = t.day in self.dom
        dow_ok = cron_dow in self.dow
        if self._dom_restricted and self._dow_restricted:
            return dom_ok or dow_ok
        return dom_ok and dow_ok

    def next_after(self, after: _dt.datetime) -> _dt.datetime:
        """The first firing strictly after ``after`` (minute resolution)."""
        t = after.replace(second=0, microsecond=0) + _dt.timedelta(minutes=1)
        # A year of minutes is a generous upper bound; any valid expression fires
        # within that window.
        for _ in range(366 * 24 * 60):
            if self._matches(t):
                return t
            t += _dt.timedelta(minutes=1)
        raise CronError(f"no matching time within a year for {self.expr!r}")

next_after(after)

The first firing strictly after after (minute resolution).

Source code in src/istos/queue/cron.py

def next_after(self, after: _dt.datetime) -> _dt.datetime:
    """The first firing strictly after ``after`` (minute resolution)."""
    t = after.replace(second=0, microsecond=0) + _dt.timedelta(minutes=1)
    # A year of minutes is a generous upper bound; any valid expression fires
    # within that window.
    for _ in range(366 * 24 * 60):
        if self._matches(t):
            return t
        t += _dt.timedelta(minutes=1)
    raise CronError(f"no matching time within a year for {self.expr!r}")

JobContext dataclass

What the queue knows about this delivery of a job.

A worker that names a ctx parameter is handed one, the same way a handler that names db is handed the app's storage::

@app.worker("jobs/email")
async def send(job, ctx: JobContext):
    if ctx.is_last_attempt:
        log.warning("final try for %s: %s", ctx.job_id, ctx.last_error)

The job body stays exactly what was enqueued — delivery metadata lives here rather than being mixed into the caller's data.

Source code in src/istos/queue/store.py

@dataclass(frozen=True)
class JobContext:
    """What the queue knows about *this* delivery of a job.

    A worker that names a ``ctx`` parameter is handed one, the same way a handler
    that names ``db`` is handed the app's storage::

        @app.worker("jobs/email")
        async def send(job, ctx: JobContext):
            if ctx.is_last_attempt:
                log.warning("final try for %s: %s", ctx.job_id, ctx.last_error)

    The job body stays exactly what was enqueued — delivery metadata lives here
    rather than being mixed into the caller's data.
    """

    job_id: str
    queue: str
    attempt: int                      # 1 on first delivery, 2 on first redelivery
    max_attempts: int
    last_error: Optional[str] = None  # why the previous attempt nacked, if it did
    correlation_id: Optional[str] = None
    traceparent: Optional[str] = None

    @property
    def is_retry(self) -> bool:
        """True when a previous attempt at this job failed."""
        return self.attempt > 1

    @property
    def is_last_attempt(self) -> bool:
        """True when raising will dead-letter the job instead of redelivering it."""
        return self.attempt >= self.max_attempts

is_last_attempt property

True when raising will dead-letter the job instead of redelivering it.

is_retry property

True when a previous attempt at this job failed.

QueueRole

The queue owner. Binds enqueue / claim / ack / nack / result queryables on the shared session, publishes a nudge when a job arrives, and reclaims expired leases on a timer. One owner per queue.

Source code in src/istos/queue/role.py

class QueueRole:
    """The queue owner. Binds enqueue / claim / ack / nack / result queryables on
    the shared session, publishes a nudge when a job arrives, and reclaims expired
    leases on a timer. One owner per queue."""

    def __init__(
        self,
        prefix: str,
        store: QueueStore,
        *,
        lease_s: float = 30.0,
        max_attempts: int = 5,
        sweep_interval_s: float = 5.0,
        retry_backoff_s: float = 0.0,
        retry_backoff_max_s: float = 600.0,
        retry_jitter: float = 0.1,
        ha: bool = False,
        authorizer: Optional[Authorizer] = None,
        logger: Optional[Any] = None,
    ) -> None:
        self.prefix = prefix.rstrip("/")
        self.store = store
        self.lease_s = lease_s
        self.max_attempts = max_attempts
        self.sweep_interval_s = sweep_interval_s
        self.retry_backoff_s = retry_backoff_s
        self.retry_backoff_max_s = retry_backoff_max_s
        self.retry_jitter = retry_jitter
        self.ha = ha
        self._authorizer = authorizer
        self._logger = logger or _logger
        self._session: Optional[Any] = None
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._queryables: List[Any] = []
        self._sweeper: Optional[asyncio.Task] = None
        # Leader election (HA mode).
        self._id = uuid.uuid4().hex
        self._members: set = set()
        self._active = False
        self._live_token: Optional[Any] = None
        self._liveliness_sub: Optional[Any] = None
        self._election_lock = asyncio.Lock()

    @property
    def is_active(self) -> bool:
        """Whether this node is the queue's live owner right now — always, without
        ``ha``; only the elected leader with it. Used to run a co-located
        :meth:`schedule` beat on exactly one node."""
        return self._active

    def _backoff(self, attempts: int) -> float:
        """Exponential backoff with jitter for the retry after ``attempts`` tries."""
        if self.retry_backoff_s <= 0:
            return 0.0
        delay: float = min(self.retry_backoff_s * (2 ** max(0, attempts - 1)), self.retry_backoff_max_s)
        if self.retry_jitter:
            delay *= 1.0 + random.uniform(-self.retry_jitter, self.retry_jitter)
        return max(0.0, delay)

    def _notify(self) -> None:
        """Nudge idle workers that a job is ready, so they claim without waiting
        for their next poll. Best-effort — a missed nudge is caught by the poll."""
        if self._session is not None:
            try:
                self._session.put(f"{self.prefix}/notify", b"1")
            except Exception:  # pragma: no cover - nudge is optional
                pass

    async def bind(self, session: "zenoh.Session", loop: asyncio.AbstractEventLoop) -> None:
        self._session = session
        self._loop = loop
        if not self.ha:
            await self._activate()
            return
        # HA: elect a single leader among the owner replicas via Zenoh liveliness.
        # Each replica declares a token; the lowest id among the live set leads and
        # binds the queryables. When it dies, its token drops and the next takes over.
        me = f"{self.prefix}/_owners/{self._id}"
        self._live_token = session.liveliness().declare_token(me)

        def _on_live(sample: "zenoh.Sample") -> None:
            if not loop.is_closed():
                asyncio.run_coroutine_threadsafe(self._on_membership(sample), loop)

        # history=True replays the currently-alive tokens so we see peers already up.
        self._liveliness_sub = session.liveliness().declare_subscriber(
            f"{self.prefix}/_owners/*", _on_live, history=True,
        )
        async with self._election_lock:
            self._members.add(self._id)
            await self._reconcile()

    async def _on_membership(self, sample: "zenoh.Sample") -> None:
        owner_id = str(sample.key_expr).rsplit("/", 1)[-1]
        async with self._election_lock:
            if sample.kind == zenoh.SampleKind.PUT:
                self._members.add(owner_id)
            else:
                self._members.discard(owner_id)
            await self._reconcile()

    async def _reconcile(self) -> None:
        leader = min(self._members) if self._members else None
        if leader == self._id and not self._active:
            self._logger.info("Queue %s: became owner (leader)", self.prefix, extra={"prefix": self.prefix})
            await self._activate()
        elif leader != self._id and self._active:
            self._logger.info("Queue %s: stepped down as owner", self.prefix, extra={"prefix": self.prefix})
            await self._deactivate()

    async def _activate(self) -> None:
        loop = self._loop
        session = self._session
        assert loop is not None and session is not None
        routes = {
            "enqueue": self._on_enqueue,
            "claim": self._on_claim,
            "ack": self._on_ack,
            "nack": self._on_nack,
            "result": self._on_result,
            "dead": self._on_dead,
            "chord": self._on_chord,
        }

        def _make(handler: Callable) -> Callable:
            def _cb(query: "zenoh.Query") -> None:
                if not loop.is_closed():
                    asyncio.run_coroutine_threadsafe(handler(query), loop)
            return _cb

        await self.store.load()
        for verb, handler in routes.items():
            q = session.declare_queryable(f"{self.prefix}/{verb}", _make(handler))
            self._queryables.append(q)
        self._sweeper = asyncio.ensure_future(self._sweep_loop())
        self._active = True
        self._logger.info("Bound queue owner for %s", self.prefix, extra={"prefix": self.prefix})

    async def _deactivate(self) -> None:
        if self._sweeper is not None:
            self._sweeper.cancel()
            self._sweeper = None
        for q in self._queryables:
            try:
                q.undeclare()
            except Exception:  # pragma: no cover
                pass
        self._queryables.clear()
        self._active = False

    async def _authorize(self, query: "zenoh.Query", params: dict) -> bool:
        if self._authorizer is None:
            return True
        raw = getattr(query, "attachment", None)
        attachment = bytes(raw) if raw is not None else None
        try:
            await check_authorized(
                self._authorizer,
                AuthContext(
                    prefix=self.prefix, key_expr=str(query.key_expr), params=params,
                    attachment=attachment, operation="queue",
                ),
            )
            return True
        except UnauthorizedError as e:
            _reply(query, ErrorResponse(
                error=e.code, code=e.code, message=e.message,
            ).to_dict())
            return False

    async def _on_enqueue(self, query: "zenoh.Query") -> None:
        params = _query_params(query)
        if not await self._authorize(query, params):
            return
        wf = params.get("wf")  # base64 JSON continuation, opaque to the owner
        raw = getattr(query, "attachment", None)
        env = RequestEnvelope.from_attachment(bytes(raw) if raw is not None else None)
        job_id = await self.store.add(
            _query_payload(query),
            max_attempts=self.max_attempts,
            priority=_as_int(params, "priority", 0),
            delay_s=_as_float(params, "delay", 0.0),
            wf=wf,
            correlation_id=env.correlation_id,
            traceparent=env.traceparent,
        )
        if _as_float(params, "delay", 0.0) <= 0:
            self._notify()
        _reply(query, {"job_id": job_id})

    async def _on_claim(self, query: "zenoh.Query") -> None:
        params = _query_params(query)
        if not await self._authorize(query, params):
            return
        rec = await self.store.claim(lease_s=self.lease_s)
        if rec is None:
            _reply(query, {"empty": True})
            return
        _reply(query, {
            "job_id": rec.id,
            "attempt": rec.attempts,
            "max_attempts": rec.max_attempts,
            "last_error": rec.last_error,   # why the previous attempt nacked, if any
            "data": base64.b64encode(rec.data).decode("ascii"),
            "wf": rec.wf,
            "correlation_id": rec.correlation_id,
            "traceparent": rec.traceparent,
        })

    async def _on_ack(self, query: "zenoh.Query") -> None:
        params = _query_params(query)
        if not await self._authorize(query, params):
            return
        result = _query_payload(query) or None
        ok = await self.store.ack(params.get("job_id", ""), result=result)
        _reply(query, {"ok": ok})

    async def _on_nack(self, query: "zenoh.Query") -> None:
        params = _query_params(query)
        if not await self._authorize(query, params):
            return
        error = _query_payload(query).decode("utf-8") or None
        attempt = _as_int(params, "attempt", 1)
        disposition = await self.store.nack(
            params.get("job_id", ""), error=error, retry_delay_s=self._backoff(attempt),
        )
        _reply(query, {"ok": disposition is not None, "disposition": disposition})

    async def _on_result(self, query: "zenoh.Query") -> None:
        params = _query_params(query)
        if not await self._authorize(query, params):
            return
        state, data = await self.store.result(params.get("job_id", ""))
        _reply(query, {
            "state": state,
            "result": base64.b64encode(data).decode("ascii") if data is not None else None,
        })

    async def _on_chord(self, query: "zenoh.Query") -> None:
        params = _query_params(query)
        if not await self._authorize(query, params):
            return
        result_b64 = _query_payload(query).decode("ascii") or None
        results = await self.store.chord_report(
            params.get("chord_id", ""), _as_int(params, "index", 0),
            _as_int(params, "size", 1), result_b64,
        )
        # `complete` is True (with results) for exactly one member; None otherwise.
        _reply(query, {"complete": results is not None, "results": results})

    async def _on_dead(self, query: "zenoh.Query") -> None:
        params = _query_params(query)
        if not await self._authorize(query, params):
            return
        dead = await self.store.dead_letters()
        _reply(query, {
            "jobs": [
                {
                    "job_id": r.id,
                    "attempts": r.attempts,
                    "last_error": r.last_error,
                    "data": base64.b64encode(r.data).decode("ascii"),
                }
                for r in dead
            ]
        })

    async def _sweep_loop(self) -> None:
        try:
            while True:
                await asyncio.sleep(self.sweep_interval_s)
                moved = await self.store.sweep()
                if moved:
                    self._notify()  # reclaimed jobs are ready again
                    self._logger.info(
                        "Queue %s reclaimed %d expired lease(s)", self.prefix, moved,
                        extra={"prefix": self.prefix},
                    )
        except asyncio.CancelledError:
            pass

    def unbind(self) -> None:
        if self._sweeper is not None:
            self._sweeper.cancel()
            self._sweeper = None
        for q in self._queryables:
            try:
                q.undeclare()
            except Exception:  # pragma: no cover - best-effort teardown
                pass
        self._queryables.clear()
        for handle in (self._liveliness_sub, self._live_token):
            if handle is not None:
                try:
                    handle.undeclare()
                except Exception:  # pragma: no cover
                    pass
        self._liveliness_sub = None
        self._live_token = None
        self._active = False
        self._session = None

    async def aclose(self) -> None:
        self.unbind()

is_active property

Whether this node is the queue's live owner right now — always, without ha; only the elected leader with it. Used to run a co-located :meth:schedule beat on exactly one node.

QueueStore

Authoritative job state for one queue, held in the owner's memory and (optionally) written through to a StoragePlugin so it survives a restart.

A single owner serializes every mutation through _lock, which is what makes claim/ack atomic without a database transaction — there is only ever one writer.

Source code in src/istos/queue/store.py

class QueueStore:
    """Authoritative job state for one queue, held in the owner's memory and
    (optionally) written through to a ``StoragePlugin`` so it survives a restart.

    A single owner serializes every mutation through ``_lock``, which is what
    makes claim/ack atomic without a database transaction — there is only ever
    one writer.
    """

    def __init__(
        self,
        name: str,
        storage: Any = None,
        *,
        keep_results: bool = False,
        result_ttl_s: float = 3600.0,
    ) -> None:
        self._name = name
        self._storage = storage
        self.keep_results = keep_results
        self.result_ttl_s = result_ttl_s
        self._jobs: Dict[str, JobRecord] = {}
        self._ids: set = set()                              # membership, mirrored to storage
        self._ready: List[Tuple[int, int, str]] = []        # (-priority, seq, id)
        self._delayed: List[Tuple[float, str]] = []         # (not_before, id)
        self._leases: List[Tuple[float, str]] = []          # (lease_until, id)
        self._done: List[Tuple[float, str]] = []            # (expiry, id)
        self._chords: Dict[str, dict] = {}                  # chord_id → barrier state
        self._seq = 0
        self._lock = asyncio.Lock()

    # --- heap bookkeeping (all O(log n); entries are validated lazily on pop) ---

    def _offer_ready(self, rec: JobRecord) -> None:
        if rec.not_before > time.time():
            heapq.heappush(self._delayed, (rec.not_before, rec.id))
        else:
            heapq.heappush(self._ready, (-rec.priority, rec.seq, rec.id))

    def _mature_delayed(self, now: float) -> None:
        while self._delayed and self._delayed[0][0] <= now:
            _, job_id = heapq.heappop(self._delayed)
            rec = self._jobs.get(job_id)
            if rec is not None and rec.state == JobState.READY and rec.not_before <= now:
                heapq.heappush(self._ready, (-rec.priority, rec.seq, rec.id))

    def _reclaim_leases(self, now: float) -> List[JobRecord]:
        """Move expired leases back to ready (or dead). Returns the changed records
        so the caller can persist them."""
        changed: List[JobRecord] = []
        while self._leases and self._leases[0][0] <= now:
            _, job_id = heapq.heappop(self._leases)
            rec = self._jobs.get(job_id)
            if rec is None or rec.state != JobState.LEASED or rec.lease_until > now:
                continue  # stale entry: job was acked, nacked, or re-leased
            if rec.attempts >= rec.max_attempts:
                rec.state = JobState.DEAD
            else:
                rec.state = JobState.READY
                rec.lease_until = 0.0
                rec.not_before = 0.0  # a lost lease is redelivered promptly
                heapq.heappush(self._ready, (-rec.priority, rec.seq, rec.id))
            changed.append(rec)
        return changed

    # --- persistence (per-job writes are O(1); the index only moves on membership) ---

    def _index_key(self) -> str:
        return f"queue:{self._name}:index"

    def _job_key(self, job_id: str) -> str:
        return f"queue:{self._name}:job:{job_id}"

    async def load(self) -> None:
        """Recover state from storage on bind. No-op for the in-memory default."""
        if self._storage is None:
            return
        try:
            index = await self._storage.get(self._index_key()) or []
            for job_id in index:
                raw = await self._storage.get(self._job_key(job_id))
                if raw is None:
                    continue
                rec = JobRecord.from_dict(raw if isinstance(raw, dict) else json.loads(raw))
                self._jobs[rec.id] = rec
                self._ids.add(rec.id)
                self._seq = max(self._seq, rec.seq)
                if rec.state == JobState.READY:
                    self._offer_ready(rec)
                elif rec.state == JobState.LEASED:
                    heapq.heappush(self._leases, (rec.lease_until, rec.id))
                elif rec.state == JobState.DONE:
                    heapq.heappush(self._done, (rec.completed_at + self.result_ttl_s, rec.id))
        except Exception:  # recovery is best-effort — never block startup
            _logger.exception("Queue %s failed to recover from storage", self._name)

    async def _write_job(self, rec: JobRecord) -> None:
        if self._storage is None:
            return
        try:
            await self._storage.put(self._job_key(rec.id), rec.to_dict())
        except Exception:
            _logger.exception("Queue %s failed to persist job %s", self._name, rec.id)

    async def _write_index(self) -> None:
        if self._storage is None:
            return
        try:
            await self._storage.put(self._index_key(), list(self._ids))
        except Exception:
            _logger.exception("Queue %s failed to persist index", self._name)

    async def _forget(self, job_id: str) -> None:
        self._jobs.pop(job_id, None)
        self._ids.discard(job_id)
        if self._storage is not None:
            try:
                await self._storage.delete(self._job_key(job_id))
            except Exception:
                _logger.exception("Queue %s failed to erase job %s", self._name, job_id)
        await self._write_index()

    # --- operations ---

    async def add(
        self, data: bytes, *, max_attempts: int, priority: int = 0, delay_s: float = 0.0,
        wf: Optional[str] = None,
        correlation_id: Optional[str] = None,
        traceparent: Optional[str] = None,
    ) -> str:
        async with self._lock:
            self._seq += 1
            rec = JobRecord(
                id=uuid.uuid4().hex, data=data, max_attempts=max_attempts,
                priority=priority, seq=self._seq, wf=wf,
                not_before=time.time() + delay_s if delay_s > 0 else 0.0,
                correlation_id=correlation_id,
                traceparent=traceparent,
            )
            self._jobs[rec.id] = rec
            self._ids.add(rec.id)
            self._offer_ready(rec)
            await self._write_job(rec)
            await self._write_index()
        return rec.id

    async def claim(self, *, lease_s: float) -> Optional[JobRecord]:
        """Lease the highest-priority eligible job (FIFO within a priority) in
        O(log n): pop the ready heap, skipping stale entries. Matures delayed jobs
        and reclaims expired leases first. Returns a copy so callers can't mutate
        state off-lock."""
        now = time.time()
        async with self._lock:
            self._mature_delayed(now)
            for reclaimed in self._reclaim_leases(now):
                await self._write_job(reclaimed)
            while self._ready:
                _, _, job_id = heapq.heappop(self._ready)
                rec = self._jobs.get(job_id)
                if rec is None or rec.state != JobState.READY or rec.not_before > now:
                    continue  # tombstone: job changed state since it was queued
                rec.state = JobState.LEASED
                rec.attempts += 1
                rec.lease_until = now + lease_s
                heapq.heappush(self._leases, (rec.lease_until, rec.id))
                await self._write_job(rec)
                return JobRecord.from_dict(rec.to_dict())
        return None

    async def ack(self, job_id: str, *, result: Optional[bytes] = None) -> bool:
        async with self._lock:
            rec = self._jobs.get(job_id)
            if rec is None:
                return False
            if self.keep_results:
                rec.state = JobState.DONE
                rec.result = base64.b64encode(result).decode("ascii") if result is not None else None
                rec.completed_at = time.time()
                rec.lease_until = 0.0
                heapq.heappush(self._done, (rec.completed_at + self.result_ttl_s, rec.id))
                await self._write_job(rec)
            else:
                await self._forget(job_id)
        return True

    async def nack(
        self, job_id: str, *, error: Optional[str] = None, retry_delay_s: float = 0.0,
    ) -> Optional[str]:
        """Fail a leased job. Redelivers (after ``retry_delay_s``) if attempts
        remain, else dead-letters. Returns "requeued" or "dead" (or None if
        the job is unknown)."""
        async with self._lock:
            rec = self._jobs.get(job_id)
            if rec is None:
                return None
            rec.last_error = error
            if rec.attempts >= rec.max_attempts:
                rec.state = JobState.DEAD
                disposition = "dead"
            else:
                rec.state = JobState.READY
                rec.lease_until = 0.0
                rec.not_before = time.time() + retry_delay_s if retry_delay_s > 0 else 0.0
                self._offer_ready(rec)
                disposition = "requeued"
            await self._write_job(rec)
        return disposition

    async def sweep(self) -> int:
        """Reclaim jobs whose lease expired (worker crashed without ack) and purge
        results past their TTL. Returns how many leases were reclaimed."""
        now = time.time()
        async with self._lock:
            changed = self._reclaim_leases(now)
            for reclaimed in changed:
                await self._write_job(reclaimed)
            while self._done and self._done[0][0] <= now:
                _, job_id = heapq.heappop(self._done)
                rec = self._jobs.get(job_id)
                if (
                    rec is not None and rec.state == JobState.DONE
                    and rec.completed_at + self.result_ttl_s <= now
                ):
                    await self._forget(job_id)
        return len(changed)

    async def result(self, job_id: str) -> Tuple[str, Optional[bytes]]:
        """Return ``(state, result_bytes)`` for a job. State is "unknown" once the
        job (and its retained result) is gone."""
        async with self._lock:
            rec = self._jobs.get(job_id)
            if rec is None:
                return "unknown", None
            data = base64.b64decode(rec.result) if rec.result is not None else None
            return rec.state.value, data

    async def dead_letters(self) -> List[JobRecord]:
        async with self._lock:
            return [
                JobRecord.from_dict(r.to_dict())
                for r in self._jobs.values() if r.state == JobState.DEAD
            ]

    async def stats(self) -> Dict[str, int]:
        async with self._lock:
            counts = {s.value: 0 for s in JobState}
            for rec in self._jobs.values():
                counts[rec.state.value] += 1
            return counts

    async def chord_report(
        self, chord_id: str, index: int, size: int, result: Optional[str],
    ) -> Optional[List[Optional[str]]]:
        """Record one chord member's completion. Returns the collected results (in
        member order) exactly once — to the caller that completes the barrier — and
        None for every earlier member. Results are base64 strings (or None)."""
        async with self._lock:
            chord = self._chords.get(chord_id)
            if chord is None:
                chord = {
                    "size": size, "results": [None] * size,
                    "reported": set(), "fired": False,
                }
                self._chords[chord_id] = chord
            if chord["fired"] or index in chord["reported"]:
                return None  # already fired, or this member was already counted (redelivery)
            chord["reported"].add(index)
            chord["results"][index] = result
            if len(chord["reported"]) >= chord["size"]:
                chord["fired"] = True
                results = list(chord["results"])
                del self._chords[chord_id]
                return results
            return None

chord_report(chord_id, index, size, result) async

Record one chord member's completion. Returns the collected results (in member order) exactly once — to the caller that completes the barrier — and None for every earlier member. Results are base64 strings (or None).

Source code in src/istos/queue/store.py

async def chord_report(
    self, chord_id: str, index: int, size: int, result: Optional[str],
) -> Optional[List[Optional[str]]]:
    """Record one chord member's completion. Returns the collected results (in
    member order) exactly once — to the caller that completes the barrier — and
    None for every earlier member. Results are base64 strings (or None)."""
    async with self._lock:
        chord = self._chords.get(chord_id)
        if chord is None:
            chord = {
                "size": size, "results": [None] * size,
                "reported": set(), "fired": False,
            }
            self._chords[chord_id] = chord
        if chord["fired"] or index in chord["reported"]:
            return None  # already fired, or this member was already counted (redelivery)
        chord["reported"].add(index)
        chord["results"][index] = result
        if len(chord["reported"]) >= chord["size"]:
            chord["fired"] = True
            results = list(chord["results"])
            del self._chords[chord_id]
            return results
        return None

claim(*, lease_s) async

Lease the highest-priority eligible job (FIFO within a priority) in O(log n): pop the ready heap, skipping stale entries. Matures delayed jobs and reclaims expired leases first. Returns a copy so callers can't mutate state off-lock.

Source code in src/istos/queue/store.py

async def claim(self, *, lease_s: float) -> Optional[JobRecord]:
    """Lease the highest-priority eligible job (FIFO within a priority) in
    O(log n): pop the ready heap, skipping stale entries. Matures delayed jobs
    and reclaims expired leases first. Returns a copy so callers can't mutate
    state off-lock."""
    now = time.time()
    async with self._lock:
        self._mature_delayed(now)
        for reclaimed in self._reclaim_leases(now):
            await self._write_job(reclaimed)
        while self._ready:
            _, _, job_id = heapq.heappop(self._ready)
            rec = self._jobs.get(job_id)
            if rec is None or rec.state != JobState.READY or rec.not_before > now:
                continue  # tombstone: job changed state since it was queued
            rec.state = JobState.LEASED
            rec.attempts += 1
            rec.lease_until = now + lease_s
            heapq.heappush(self._leases, (rec.lease_until, rec.id))
            await self._write_job(rec)
            return JobRecord.from_dict(rec.to_dict())
    return None

load() async

Recover state from storage on bind. No-op for the in-memory default.

Source code in src/istos/queue/store.py

async def load(self) -> None:
    """Recover state from storage on bind. No-op for the in-memory default."""
    if self._storage is None:
        return
    try:
        index = await self._storage.get(self._index_key()) or []
        for job_id in index:
            raw = await self._storage.get(self._job_key(job_id))
            if raw is None:
                continue
            rec = JobRecord.from_dict(raw if isinstance(raw, dict) else json.loads(raw))
            self._jobs[rec.id] = rec
            self._ids.add(rec.id)
            self._seq = max(self._seq, rec.seq)
            if rec.state == JobState.READY:
                self._offer_ready(rec)
            elif rec.state == JobState.LEASED:
                heapq.heappush(self._leases, (rec.lease_until, rec.id))
            elif rec.state == JobState.DONE:
                heapq.heappush(self._done, (rec.completed_at + self.result_ttl_s, rec.id))
    except Exception:  # recovery is best-effort — never block startup
        _logger.exception("Queue %s failed to recover from storage", self._name)

nack(job_id, *, error=None, retry_delay_s=0.0) async

Fail a leased job. Redelivers (after retry_delay_s) if attempts remain, else dead-letters. Returns "requeued" or "dead" (or None if the job is unknown).

Source code in src/istos/queue/store.py

async def nack(
    self, job_id: str, *, error: Optional[str] = None, retry_delay_s: float = 0.0,
) -> Optional[str]:
    """Fail a leased job. Redelivers (after ``retry_delay_s``) if attempts
    remain, else dead-letters. Returns "requeued" or "dead" (or None if
    the job is unknown)."""
    async with self._lock:
        rec = self._jobs.get(job_id)
        if rec is None:
            return None
        rec.last_error = error
        if rec.attempts >= rec.max_attempts:
            rec.state = JobState.DEAD
            disposition = "dead"
        else:
            rec.state = JobState.READY
            rec.lease_until = 0.0
            rec.not_before = time.time() + retry_delay_s if retry_delay_s > 0 else 0.0
            self._offer_ready(rec)
            disposition = "requeued"
        await self._write_job(rec)
    return disposition

result(job_id) async

Return (state, result_bytes) for a job. State is "unknown" once the job (and its retained result) is gone.

Source code in src/istos/queue/store.py

async def result(self, job_id: str) -> Tuple[str, Optional[bytes]]:
    """Return ``(state, result_bytes)`` for a job. State is "unknown" once the
    job (and its retained result) is gone."""
    async with self._lock:
        rec = self._jobs.get(job_id)
        if rec is None:
            return "unknown", None
        data = base64.b64decode(rec.result) if rec.result is not None else None
        return rec.state.value, data

sweep() async

Reclaim jobs whose lease expired (worker crashed without ack) and purge results past their TTL. Returns how many leases were reclaimed.

Source code in src/istos/queue/store.py

async def sweep(self) -> int:
    """Reclaim jobs whose lease expired (worker crashed without ack) and purge
    results past their TTL. Returns how many leases were reclaimed."""
    now = time.time()
    async with self._lock:
        changed = self._reclaim_leases(now)
        for reclaimed in changed:
            await self._write_job(reclaimed)
        while self._done and self._done[0][0] <= now:
            _, job_id = heapq.heappop(self._done)
            rec = self._jobs.get(job_id)
            if (
                rec is not None and rec.state == JobState.DONE
                and rec.completed_at + self.result_ttl_s <= now
            ):
                await self._forget(job_id)
    return len(changed)

worker_wrapper

A competing consumer. Runs concurrency claim→run→ack loops against a queue owner elsewhere on the mesh, waking on the owner's nudge instead of busy polling. The handler returning acks the job (and hands back its result); the handler raising nacks it (redeliver with backoff, then dead-letter).

Source code in src/istos/queue/worker.py

class worker_wrapper:
    """A competing consumer. Runs ``concurrency`` claim→run→ack loops against a
    queue owner elsewhere on the mesh, waking on the owner's nudge instead of busy
    polling. The handler returning acks the job (and hands back its result); the
    handler raising nacks it (redeliver with backoff, then dead-letter)."""

    def __init__(
        self,
        func: Callable,
        prefix: str,
        *,
        concurrency: int = 1,
        poll_interval_s: float = 1.0,
        serializer: Serialize,
        token: Optional[Any] = None,
        dependency_overrides: Optional[dict] = None,
    ) -> None:
        self.func = func
        self.prefix = prefix.rstrip("/")
        self.concurrency = max(1, concurrency)
        self.poll_interval_s = poll_interval_s
        self.serializer = serializer
        self._token = token
        self._has_depends = has_dependencies(func)
        self._skip_names = tuple(positional_param_names(func)[:1])  # the job param
        self._wants_ctx = _wants_ctx(func)
        self._overrides = dependency_overrides or {}
        self._app: Any = None
        self._tasks: List[asyncio.Task] = []
        self._subscriber: Optional[Any] = None
        self._wake: Optional[asyncio.Event] = None
        self._running = False

    def start(self, app: Any) -> None:
        self._app = app
        self._running = True
        self._wake = asyncio.Event()
        loop = asyncio.get_running_loop()
        session = app._session_manager.session
        if session is not None:
            def _on_notify(_sample: Any) -> None:
                if not loop.is_closed() and self._wake is not None:
                    loop.call_soon_threadsafe(self._wake.set)
            try:
                self._subscriber = session.declare_subscriber(f"{self.prefix}/notify", _on_notify)
            except Exception:  # pragma: no cover - fall back to polling only
                self._subscriber = None
        for _ in range(self.concurrency):
            self._tasks.append(asyncio.ensure_future(self._loop()))

    async def stop(self) -> None:
        self._running = False
        if self._subscriber is not None:
            try:
                self._subscriber.undeclare()
            except Exception:  # pragma: no cover
                pass
            self._subscriber = None
        for task in self._tasks:
            task.cancel()
        for task in self._tasks:
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass
        self._tasks.clear()

    async def _run_body(self, job: Any, ctx: Optional[JobContext] = None) -> Any:
        extra = {"ctx": ctx} if self._wants_ctx else {}
        if self._has_depends:
            return await invoke_with_dependencies(
                self.func, args=(job,), context=extra, skip_names=self._skip_names,
                overrides=self._overrides,
            )
        if inspect.iscoroutinefunction(self.func):
            return await self.func(job, **extra)
        return await asyncio.to_thread(lambda: self.func(job, **extra))

    async def _idle(self) -> None:
        """Wait for a nudge, or fall through after poll_interval to re-check for
        redelivered / newly-eligible jobs the nudge might not cover."""
        wake = self._wake
        if wake is None:
            await asyncio.sleep(self.poll_interval_s)
            return
        try:
            await asyncio.wait_for(wake.wait(), timeout=self.poll_interval_s)
        except asyncio.TimeoutError:
            pass
        wake.clear()

    async def _loop(self) -> None:
        while self._running:
            try:
                reply = await self._app._queue_claim(self.prefix, token=self._token)
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                # Without this a rejected token looks the same as an idle queue.
                _logger.warning(
                    "Worker for %s could not claim: %s", self.prefix, exc,
                    extra={"prefix": self.prefix},
                )
                await asyncio.sleep(self.poll_interval_s)
                continue

            if not reply or reply.get("empty"):
                await self._idle()
                continue

            job_id = reply["job_id"]
            attempt = reply.get("attempt", 1)
            wf = reply.get("wf")
            ctx = JobContext(
                job_id=job_id,
                queue=self.prefix,
                attempt=attempt,
                max_attempts=reply.get("max_attempts", 0),
                last_error=reply.get("last_error"),
                correlation_id=reply.get("correlation_id"),
                traceparent=reply.get("traceparent"),
            )
            try:
                job = self.serializer.deserialize(base64.b64decode(reply["data"]))
                result = await self._run_body(job, ctx)
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                _logger.exception(
                    "Worker for %s failed job %s (attempt %s)",
                    self.prefix, job_id, attempt, extra={"prefix": self.prefix},
                )
                try:
                    await self._app._queue_nack(
                        self.prefix, job_id, error=str(exc), attempt=attempt, token=self._token,
                    )
                except Exception:
                    pass
                continue

            try:
                payload = self.serializer.serialize(result)
                result_bytes = payload.encode("utf-8") if isinstance(payload, str) else payload
                # Advance any workflow (chain step / chord report) before acking, so
                # a crash redelivers the whole step rather than losing the follow-on.
                await self._advance_workflow(wf, result_bytes)
                await self._app._queue_ack(
                    self.prefix, job_id, result=result_bytes, token=self._token,
                )
            except Exception:
                # The lease will expire and the job redeliver — at-least-once.
                _logger.exception(
                    "Worker for %s could not finish job %s", self.prefix, job_id,
                    extra={"prefix": self.prefix},
                )

    async def _advance_workflow(self, wf_b64: Optional[str], result_bytes: bytes) -> None:
        if not wf_b64:
            return
        wf = _decode_wf(wf_b64)
        chain = wf.get("chain")
        if chain:
            nxt, rest = chain[0], chain[1:]
            next_wf = _encode_wf({"chain": rest}) if rest else None
            await self._app._queue_enqueue(
                nxt["prefix"], result_bytes, wf=next_wf, token=self._token,
            )
        chord = wf.get("chord")
        if chord:
            result_b64 = base64.b64encode(result_bytes).decode("ascii")
            reply = await self._app._queue_chord_report(
                self.prefix, chord["id"], chord["index"], chord["size"],
                result_b64, token=self._token,
            )
            if reply and reply.get("complete"):
                member_results = [
                    self.serializer.deserialize(base64.b64decode(r)) if r is not None else None
                    for r in (reply.get("results") or [])
                ]
                cb = chord["callback"]
                cb_input = self.serializer.deserialize(base64.b64decode(cb["data"]))
                body = self.serializer.serialize({"results": member_results, "input": cb_input})
                cb_bytes = body.encode("utf-8") if isinstance(body, str) else body
                await self._app._queue_enqueue(cb["prefix"], cb_bytes, token=self._token)