@lunora/queue

@lunora/queue brings Cloudflare Queues to Lunora. Declare a queue with defineQueue in lunora/queues.ts; codegen wires a typed ctx.queues.<name> producer onto mutations and actions and generates the worker queue() push consumer. The config layer reconciles the wrangler queues.producers[] / queues.consumers[] entries from the same definition.

pnpm add @lunora/queue

Scaffold a queue:

vis generate lunora-queue --name=emailQueue

Declare a queue

// lunora/queues.ts
import { defineQueue } from "@lunora/queue";

import { api } from "./_generated/api";

export const emailQueue = defineQueue<{ to: string }>({
    handler: async (ctx, batch) => {
        for (const message of batch.messages) {
            try {
                await message.run(api.email.send, { to: message.body.to });
                message.ack();
            } catch (error) {
                ctx.log.error("send failed", message.id, error);
                message.retry();
            }
        }
    },
    // Push-consumer tuning (optional):
    // maxBatchSize: 10, maxBatchTimeout: 5, maxRetries: 3, deadLetterQueue: "dlq",
});

The handler runs inside a Lunora context: call any query/mutation/action with message.run(api.x.y, args) (the dispatch goes through the same path the scheduler and workflows use). Each message is ack/retry-able for at-least-once processing.

Poison messages

message.run(...) is ctx.run(...) pinned to that one message, and the pin is what keeps one bad message from taking its whole batch down.

Cloudflare delivers up to 100 messages per batch and redelivers the whole batch when the handler throws. So a message that can never succeed — a deleted row, a body that fails validation — burns every sibling's retry budget with it and eventually dead-letters the lot.

When a call made through message.run(...) fails deterministically (the dispatched function answered 400, 403, 404 or 422 — a retry would fail identically), the consumer attributes the failure to that message: it is acked and taken out of the queue, every message the handler had not yet decided is explicitly retried, and the batch as a whole resolves. Messages the handler already acked or retryed keep the decision it made. Transient failures (408, 429, 5xx, a timeout) are unchanged — the batch still throws and workerd retries it.

Attribution needs the pin. A bare ctx.run(api.x.y, args) inside the loop carries no message id, so its failure is unattributable and the whole batch retries — which is why the loop should call message.run. Isolation does not depend on the dev capture sink; it applies in production too.

Because an isolated message is acked, it never reaches the dead-letter queue. It is recorded with outcome error (and the failure message) in the Studio Queues log, which is where you see it — so keep dev capture on, or log it yourself in a catch.

Enqueue from a mutation or action

ctx.queues.<exportName> is a typed producer on Mutation and Action contexts (enqueue is a side effect, so it's excluded from the deterministic query context):

import { mutation } from "./_generated/server";
import { v } from "@lunora/values";

export const invite = mutation({
    args: { email: v.string() },
    handler: async (ctx, { email }) => {
        await ctx.queues.emailQueue.send({ to: email });
        // or in one call:
        // await ctx.queues.emailQueue.sendBatch([{ body: { to: a } }, { body: { to: b } }]);
    },
});

Push vs pull consumers

By default a queue is a push consumer: Cloudflare delivers batches to the worker's generated queue() handler. To expose the queue to an external HTTP pull consumer instead, omit the handler and set mode: "pull":

export const reports = defineQueue({ mode: "pull", name: "reports" });

lunora dev / lunora deploy reconcile the wrangler config for you: every queue gets a queues.producers[] entry; push queues add a worker consumer, pull queues add a type: "http_pull" consumer (with any batch/retry/dead-letter tuning carried through).

Studio

Declared queues show in the Studio Queues page (export name, deployed queue name, consumer mode, producer binding, and dead-letter queue), refreshed on every codegen run.