Struct Queue 

pub struct Queue { /* private fields */ }

The app’s queue: publish here, claim from here.

Cloning is cheap — the DatabaseConnection inside is a pooled handle — so every worker holds one.

Implementations

impl Queue

pub fn new(db: &Db, app: &App) -> Self

The queue for an app. Never fails and is never optional: publish works in an app whose main.toml says nothing about queues, because the table is a built-in and a message with no subscriber is still worth recording.

pub fn config(&self) -> &QueuesConfig

pub fn topics(&self) -> Vec<String>

Every topic this app subscribes to.

pub async fn prepare(&self) -> Result<(), QueueError>

Add the index the claim query needs, if it isn’t there.

Not left to the migrator: the migrator’s job is to make columns match the resource declarations, and this index is not a property of the schema but of one query — the (status, available_at) lookup every subscriber runs on every sweep, which is a sequential scan of the whole ledger without it. Retention keeps that table small in a healthy app and large in exactly the app that is having a bad day.

pub async fn publish( &self, topic: &str, message: &Value, published_by: &str, ) -> Result<Publication, QueueError>

Publish a message: one row per subscriber, then one notification.

The order matters and is not an accident. The rows are committed first, so a subscriber woken by the notification always finds them; notifying first would race, and the loser would be a wakeup for work that isn’t visible yet — which looks exactly like a queue that randomly adds 30 seconds of latency.

pub async fn claim(&self, worker: &str) -> Result<Vec<Delivery>, QueueError>

Take up to [queues] batch messages for this app’s topics.

One statement, and that is the point: the SELECT … FOR UPDATE SKIP LOCKED runs inside the UPDATE’s own transaction, so the rows are claimed and the locks released in a single commit. Holding a transaction open across the handler instead would mean one database connection tied up per in-flight message, and a long handler blocking VACUUM on the whole table.

The running rows a dead worker left behind are swept back in by Queue::reclaim rather than here, so a stuck message costs a lease rather than being invisible.

pub async fn next_due(&self) -> Result<Option<u64>, QueueError>

Seconds until the next scheduled message becomes claimable, if there is one waiting.

This is what makes a retry honour the backoff it was given rather than the poll interval. A failure schedules itself for now() + 10s, but nothing publishes when a backoff expires — there is no NOTIFY for “a timer went off” — so a subscriber that always waited the full poll_secs would round every retry up to the next 30-second boundary. Asking the database when to come back costs one indexed query per cycle and makes the configured number mean what it says.

None means nothing is scheduled, and the caller should wait its full interval. Some(0) means something is due now.

pub async fn complete(&self, id: &str) -> Result<(), QueueError>

Mark a message handled.

pub async fn fail( &self, delivery: &Delivery, error: &str, ) -> Result<bool, QueueError>

Record a failed attempt: schedule the retry, or give up.

Giving up leaves the row failed with its error rather than deleting it. A dead-letter you have to go and look at is the point — a queue that quietly discards what it could not handle is a queue that loses orders.

pub async fn reclaim(&self) -> Result<u64, QueueError>

Offer up messages whose handler never came back.

Returns how many were taken back. The attempt is not undone: a handler that reliably kills its process — the classic out-of-memory loop — has spent an attempt, and will run out of them and land in the dead-letter instead of retrying until somebody notices the restart count.

pub async fn prune(&self) -> Result<u64, QueueError>

Delete handled messages older than [queues] retain_hours. 0 keeps them forever.

Only done rows. A failed row is the whole reason the ledger exists and is never swept — if it were, the dead-letter would empty itself overnight and the evidence would go with it.

pub async fn execute( &self, request: &str, published_by: &str, ) -> Result<Value, QueueError>

Run one operation on behalf of a function. The JSON surface behind HostApi::publish.

Trait Implementations

impl Clone for Queue

fn clone(&self) -> Queue

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Queue

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.