Struct QueuesConfig 

pub struct QueuesConfig {
    pub enabled: bool,
    pub prefix: String,
    pub subscribe: BTreeMap<String, Vec<String>>,
    pub poll_secs: u64,
    pub batch: u32,
    pub max_attempts: u32,
    pub retry_backoff_secs: u64,
    pub lease_secs: u64,
    pub retain_hours: u64,
    pub publish: String,
}

Background work: a message published now, handled by a function shortly after, outside the request that caused it.

The transport is Postgres and nothing else — no broker to run, no second thing that can be down. A publish writes a row to queue_message and fires a NOTIFY; a subscriber wakes on that notification and claims the row with FOR UPDATE SKIP LOCKED. The two halves matter for different reasons: the row is what makes the message survive a restart and lets a failure be retried, and the notification is what makes it happen in milliseconds rather than on the next poll.

Because a message is a row, the guarantee is at-least-once: a handler that succeeds but crashes before its row is marked done runs again. Write handlers that can be run twice — the same reason billing_event exists.

[queues]
# Which function handles which topic. One name, or several.
[queues.subscribe]
"user.signed_up" = "send_welcome"
"order.paid"     = ["fulfil_order", "notify_ops"]

Fields

enabled: bool

Turn message handling off without deleting the subscriptions. Publishing still records rows, so nothing is lost while it is off — it is a pause, not a drain.

prefix: String

Prepended to the Postgres NOTIFY channel this app wakes on, so two apps sharing one database don’t wake each other for nothing.

subscribe: BTreeMap<String, Vec<String>>

Topic → the function(s) that handle it. Written as one name or a list:

[queues.subscribe]
"user.signed_up" = "send_welcome"
"order.paid"     = ["fulfil_order", "notify_ops"]

Each subscriber gets its own row and its own retries, so a failing notify_ops never re-runs fulfil_order.

poll_secs: u64

How often to sweep for work regardless of notifications. The NOTIFY is what makes delivery immediate; this is the safety net that picks up a message published while this process was starting, a retry whose backoff has expired, and anything a dropped connection lost the wakeup for.

batch: u32

Most messages claimed in one go. Larger batches trade latency on the last message for fewer round trips.

max_attempts: u32

How many times a message is tried before it is left failed for a person to look at. 1 means no retries at all.

retry_backoff_secs: u64

Base of the retry backoff, in seconds: attempt n waits retry_backoff_secs * 2^(n-1), so the default retries after 10s, 20s, 40s, 80s and then gives up.

lease_secs: u64

How long a claimed message may be worked on before another subscriber is allowed to take it.

This is what makes a killed process — an OOM, a rolling deploy, a lost node — recoverable rather than a message stuck forever in running. Set it comfortably above the slowest handler: expiring the lease early is what turns at-least-once into “twice, concurrently”.

retain_hours: u64

Delete handled messages after this many hours, on the same sweep. 0 keeps them forever, which is a reasonable choice for a low-volume app that wants the ledger.

publish: String

Who may publish over HTTP at POST <base>/queues/{topic}, in the same grammar a resource’s [permissions] uses.

private — the default — means there is no such endpoint at all. A topic is an internal name that triggers real work, so it is not something to expose without deciding to.

Implementations

impl QueuesConfig

pub fn channel(&self) -> String

The NOTIFY channel this app’s publishers and subscribers meet on.

One channel for the whole app rather than one per topic: the payload carries the topic, a listener has a single subscription to re-establish after a reconnect, and adding a topic needs no new LISTEN. It also sidesteps Postgres’s 63-byte limit on a channel name, which an app’s own topic names would otherwise have to live inside.

pub fn valid_topic(topic: &str) -> bool

Whether topic is a name this app will carry.

Deliberately narrow — letters, digits, and . _ - : — because a topic is an identifier that ends up in config keys, log lines and dashboard filters, and a topic with a space or a quote in it reads as a mistake everywhere it appears. Checked when publishing rather than trusted, since a topic can arrive from a function’s runtime string.

pub fn subscribers(&self, topic: &str) -> &[String]

The functions subscribed to a topic, in the order they were declared.

pub fn subscribed_functions(&self) -> BTreeSet<&str>

Every function name any topic points at, deduplicated. Used at boot to report a subscription whose function isn’t loaded.

pub fn is_active(&self) -> bool

Whether a subscriber loop should run: switched on and something to listen for. Publishing does not depend on this — a message published with no subscriber is still recorded, which is what makes “why didn’t my handler run?” answerable.

pub fn publish_access(&self) -> Access

The resolved policy for the HTTP publish endpoint. An unparseable publish closes the door, matching how every other access string here treats a typo — and so does owner, which names a column on a row and means nothing for a topic.

pub fn retry_delay_secs(&self, attempts: u32) -> u64

Seconds to wait before retrying a message that has failed attempts times, doubling each time and capped at an hour so a poisoned message doesn’t schedule itself past the retention sweep.

Trait Implementations

impl Clone for QueuesConfig

fn clone(&self) -> QueuesConfig

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for QueuesConfig

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

Formats the value using the given formatter.

impl Default for QueuesConfig

fn default() -> Self

Returns the “default value” for a type.

impl<'de> Deserialize<'de> for QueuesConfig

where QueuesConfig: Default,

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.