Cloudflare Queues in Blaze: Producing and Consuming Messages

Send messages to Cloudflare Queues from Blaze routes and process them in a batch consumer handler. Decouples background work from HTTP responses.

Cloudflare Queues is a durable, at-least-once message queue built for Workers. Your Blaze routes act as producers — they enqueue a message and immediately return a response to the client — while a separate consumer function processes those messages in batches on its own schedule. This decoupling keeps your HTTP responses fast and moves expensive or unreliable work (sending emails, calling third-party APIs, running data pipelines) out of the critical path.

Configure the binding

1. Declare producers and consumers in wrangler.toml

   wrangler.toml

   [[queues.producers]]
   binding    = "QUEUE"
   queue      = "job-queue"
   
   [[queues.consumers]]
   queue              = "job-queue"
   max_batch_size     = 10
   max_batch_timeout  = 5
   max_retries        = 3
   dead_letter_queue  = "job-queue-dlq"
   
2. Define the Env type

   src/types/env.ts

   type JobMessage = {
     type:    string;
     payload: unknown;
   };
   
   export type Env = {
     QUEUE: Queue<JobMessage>;
     // ...other bindings
   };
   
3. Pass Env to createApp

   src/index.ts

   import { createApp } from "blaze";
   import type { Env } from "./types/env";
   
   const app = createApp<Env>();
   // req.env.QUEUE is typed as Queue<JobMessage> in every handler
   

Producing messages

Call req.env.QUEUE.send(body) from any route handler to enqueue a message. The call is asynchronous but lightweight — it does not wait for the consumer to process the message. Return a 202 Accepted to signal to the client that the work has been queued, not completed.

app.post("/jobs", async (req, res) => {
  const body = await req.json<{ type: string; payload: unknown }>();

  await req.env.QUEUE.send({
    type:    body.type,
    payload: body.payload,
  });

  res.status(202).json({ status: "queued" });
});

Sending multiple messages at once

Use req.env.QUEUE.sendBatch(messages) to enqueue up to 100 messages in a single call. Each item in the array must be an object with a body property. Batching is more efficient than calling .send() in a loop because the entire batch is written to disk in one round-trip.

app.post("/jobs/batch", async (req, res) => {
  const items = await req.json<Array<{ type: string; payload: unknown }>>();

  await req.env.QUEUE.sendBatch(
    items.map((item) => ({ body: item }))
  );

  res.status(202).json({ status: "queued", count: items.length });
});

A single sendBatch call can contain up to 100 messages. Each individual message body must be under 128 KB, and the total size of the batch cannot exceed 256 KB. For larger workloads, split the array into chunks and call sendBatch once per chunk.

Consuming messages

Wire the consumer alongside app.fetch in your default export. The queue function receives a MessageBatch containing up to max_batch_size messages. Call msg.ack() after successful processing to prevent redelivery.

src/index.ts

import { createApp } from "blaze";
import type { Env } from "./types/env";

const app = createApp<Env>();

// ...route definitions

async function processMessage(
  body: { type: string; payload: unknown },
  env: Env
): Promise<void> {
  switch (body.type) {
    case "send-email":
      await sendEmail(body.payload, env);
      break;
    case "resize-image":
      await resizeImage(body.payload, env);
      break;
    default:
      console.warn("Unknown message type:", body.type);
  }
}

export default {
  fetch: app.fetch,

  async queue(batch: MessageBatch<{ type: string; payload: unknown }>, env: Env) {
    for (const msg of batch.messages) {
      try {
        await processMessage(msg.body, env);
        msg.ack(); // mark as successfully processed
      } catch (err) {
        console.error("Failed to process message:", err);
        msg.retry(); // requeue for another attempt
      }
    }
  },
};

The queue consumer runs in a separate invocation from your HTTP handlers — it is not part of the Blaze middleware chain. It receives the raw env object directly, not a BlazeRequest. You can still call any binding on env (D1, KV, R2, AI) from within the consumer.

Typed message bodies

The Queue<T> generic flows from your Env type into both req.env.QUEUE.send() and the MessageBatch<T> argument in the consumer, giving you end-to-end type safety without extra casting.

// Narrow the union to specific job types
type EmailJob   = { type: "send-email";   payload: { to: string; subject: string; body: string } };
type ImageJob   = { type: "resize-image"; payload: { key: string; width: number; height: number } };
type JobMessage = EmailJob | ImageJob;

export type Env = {
  QUEUE: Queue<JobMessage>;
};

// In a route handler — TypeScript enforces valid message shapes
app.post("/jobs/email", async (req, res) => {
  const { to, subject, body } = await req.json<EmailJob["payload"]>();

  await req.env.QUEUE.send({ type: "send-email", payload: { to, subject, body } });

  res.status(202).json({ status: "queued" });
});

// In the consumer — body is typed as JobMessage
async function queue(batch: MessageBatch<JobMessage>, env: Env) {
  for (const msg of batch.messages) {
    if (msg.body.type === "send-email") {
      // msg.body.payload is typed as { to, subject, body }
      await sendEmail(msg.body.payload, env);
    }
    msg.ack();
  }
}

Dead letter queues

When a message exceeds max_retries, Cloudflare automatically moves it to the dead letter queue specified in wrangler.toml. Inside your consumer, call msg.retry() to explicitly requeue a failed message for another attempt, or msg.ack() to discard it even when processing failed (useful for intentionally skipping malformed messages).

for (const msg of batch.messages) {
  try {
    await processMessage(msg.body, env);
    msg.ack();
  } catch (err) {
    if (isTransientError(err)) {
      msg.retry(); // will be redelivered up to max_retries times
    } else {
      // Permanent failure — ack to discard and avoid infinite retries
      console.error("Discarding unprocessable message:", msg.body, err);
      msg.ack();
    }
  }
}

Configure the dead letter queue in wrangler.toml so permanently failed messages are preserved for inspection:

wrangler.toml

[[queues.consumers]]
queue             = "job-queue"
max_retries       = 3
dead_letter_queue = "job-queue-dlq"

When a queue send does not need to complete before you return a response, use req.ctx.waitUntil() to fire it in the background. This is useful for low-priority analytics or audit events where you want to accept the HTTP response immediately and let the enqueue happen concurrently.

app.post("/track", async (req, res) => {
  const event = await req.json();

  // Respond at once — the send happens after the response is flushed
  res.status(202).json({ ok: true });

  req.ctx.waitUntil(
    req.env.QUEUE.send({ type: "analytics-event", payload: event })
  );
});