Queues and background jobs

A queue holds messages in the app's own database until a consumer function has handled them. Messages survive a restart and are delivered at least once, so a consumer must be safe to run twice for the same message (use the send's id to keep a sender from queuing the same job twice).

Declare, send, consume

A version declares its queues (the queues array of the version manifest, like cron). Sending to a name the live version does not declare throws an error that names the queue, so a typo never piles up messages nobody reads.

"queues": [
  { "name": "emails", "consumer": "mailer", "batch_size": 10, "max_retries": 5 },
  { "name": "images", "consumer": "resizer", "dead_letter_queue": "images-failed" },
  { "name": "images-failed" }
]

substrio push and substrio deploy build that array from substrio.toml: one [queues.<name>] table per queue, with the same fields (a queue name with a . cannot be a table header, so it cannot be declared there).

[queues.emails]
consumer = "mailer"
batch_size = 10
max_retries = 5

[queues.images]
consumer = "resizer"
dead_letter_queue = "images-failed"

[queues.images-failed]

Send from any function:

substrio.queue("emails").send({ to: "[email protected]" });
substrio.queue("emails").send({ to: "[email protected]" }, { delay: 60_000, id: "welcome-bo" });
substrio.queue("emails").sendBatch([{ to: "a" }, { body: { to: "b" }, delay: 5000 }]);

send returns { seq, duplicate }. A send whose id is already waiting in the queue (or being delivered) sends nothing and answers duplicate: true with the first message's seq. The id is forgotten once its message is acked or dead. delay is in milliseconds. A send is refused inside db.transaction.

sendBatch takes bodies, or { body, delay, id } objects (an item whose own body has a body key must use the object form).

The consumer is the module export queue(batch, env, ctx) of the function the declaration names:

export default {
  async queue(batch, env, ctx) {
    for (const msg of batch.messages) {   // msg.body, msg.attempts, msg.id, msg.timestamp
      try {
        await deliver(msg.body);
        msg.ack();
      } catch (e) {
        msg.retry({ delay: 30_000 });     // or msg.retry(): exponential backoff
      }
    }
  },
};

A message the handler does not settle is acked when the handler returns and retried when it throws or rejects (the throw is recorded in function.errors). batch.ackAll() and batch.retryAll({ delay }) settle every message.

Settings

Each queue takes these members (all optional, milliseconds where named _ms); a member left out takes the default below.

A message past max_retries goes to the dead_letter_queue (as a new message there), or, without one, stays in the queue's dead letters.

How it behaves

• A delivery hides the message for visibility_timeout_ms. If the handler has not settled it by then (it is still running, or the server stopped), the message is delivered again with attempts one higher; the first run's late ack() is ignored.
• A burst of messages waits in the database until a consumer takes it; none is refused or dropped. Each consumer call is a normal function call (same limits, same logs).
• The queues of a version replace the app's declared queues when the version goes live; a queue a later version drops keeps its messages (a send to it is refused until it is declared again).
• A sandbox's queues are its own, over its copy of the data.

Operations

queue.list, queue.stats, queue.purge, queue.dead_letters and queue.retry_dead are on the HTTP API (/v1/_queues), the CLI and MCP:

• queue.stats <name>: ready, in_flight, delayed and dead counts.
• queue.purge <name> [scope]: pending (default), dead or all.
• queue.dead_letters <name>: page the dead letters with limit and cursor.
• queue.retry_dead <name> [seq]: send one or every dead letter back with its attempts reset.