Queues and Durable Objects: async messaging and state

TL;DR

Queues: fire-and-forget async messaging. Producers send(), consumers process batches with ack() / retry(), a dead letter queue holds messages that fail after N attempts.

Durable Objects (DOs): single-writer actors. Each ID hashes to exactly one instance running in exactly one isolate, with its own SQL storage, strong consistency, and WebSocket hibernation.

The key claim:

Use Queues when processing can be delayed (seconds to minutes). Use DOs when you need state consistent across multiple requests or connections (counters, chat rooms, locks, rate limiters, sessions). Combine them when the workflow has both: the DO coordinates, the Queue runs heavy tasks.

This post walks through the Queues flow (producer, broker, consumer, retry, DLQ), the DO model (id, stub, instance, storage, hibernation), 6 real-world patterns, and gotchas from this blog.

---

Who this is for

• Developers building workflows with background tasks (email digests, image resize, webhook replay).
• Anyone who needs a rate limiter, counter, WebSocket server, or session manager.
• Lambda + SQS users looking for the Workers equivalent.

Read first: Part 2 (runtime + CPU limit), Part 3 (when to use which primitive), Part 6 (D1 for shared state).

After this post you’ll:

• Write a basic producer + consumer for Queues.
• Build your first DO for a counter or rate limiter.
• Decide between Queue, DO, or both.
• Know the limits and production gotchas.

---

What this post isn’t about

• WebSocket hibernation API details: Part 15 (Durable Objects for real-time).
• Advanced batch optimisation for Queues: basics here, depth in Part 17.
• DO migration versioning: advanced, rarely encountered.

---

Queues: async messaging

When to use Queues

• Fire-and-forget jobs: image resize, email send, index rebuild — the user doesn’t wait.
• Rate-limiting outgoing traffic: smoothing bursts to a steady rate.
• Retry with backoff: external API calls that fail often, where you don’t want to block the user.
• Fan-out processing: one event → N workers in parallel.
• Dead letter handling: failed messages → DLQ for debugging.

Binding

{
  "queues": {
    "producers": [
      { "binding": "MY_QUEUE", "queue": "my-queue" }
    ],
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10,
        "max_batch_timeout": 5,
        "max_retries": 3,
        "dead_letter_queue": "my-queue-dlq"
      }
    ]
  }
}

Producer

async fetch(request, env) {
  const body = await request.json();

  // Send one message
  await env.MY_QUEUE.send({ type: "image-resize", url: body.url });

  // Batch send
  await env.MY_QUEUE.sendBatch([
    { body: { type: "email", to: "a@x.com" } },
    { body: { type: "email", to: "b@x.com" } },
  ]);

  return Response.json({ queued: true });
}

send() is a subrequest, ~10ms. It doesn’t wait for the consumer.

Consumer

export default {
  async fetch(request, env) { /* ... */ },

  async queue(batch: MessageBatch<MyMsg>, env: Env, ctx: ExecutionContext) {
    for (const msg of batch.messages) {
      try {
        if (msg.body.type === "image-resize") {
          await resizeImage(env, msg.body.url);
        } else if (msg.body.type === "email") {
          await sendEmail(env, msg.body.to);
        }
        msg.ack();  // success, remove from queue
      } catch (err) {
        console.error(`Failed msg ${msg.id}:`, err);
        msg.retry({ delaySeconds: 60 });  // retry in 60s
      }
    }
  }
};

The queue handler gets a batch of up to max_batch_size messages, or waits up to max_batch_timeout seconds.

Retry and DLQ

• Default max_retries: 3. After 3 failures, the message auto-moves to the DLQ.
• dead_letter_queue configured in the consumer block.
• The DLQ is just another queue; a separate consumer reads it for debugging / alerting / replay.

// DLQ consumer
async queue(batch, env) {
  for (const msg of batch.messages) {
    await env.DB.prepare("INSERT INTO failed_jobs (msg_id, body, retry_count) VALUES (?, ?, ?)")
      .bind(msg.id, JSON.stringify(msg.body), 3)
      .run();
    msg.ack();  // persisted to DB, remove from DLQ
  }
}

Limits

• Max batch size: 100 messages per batch.
• Max retries: 100 (default 3).
• Message body: 128 KB default, up to 1 MB with config.
• Delivery: at-least-once. Messages can be delivered twice → the consumer must be idempotent.

At-least-once = consumers must be idempotent

// WRONG: if retried, the user gets the email twice
async function handleMsg(msg) {
  await sendEmail(msg.body.email);
  msg.ack();
}

// RIGHT: check if already processed
async function handleMsg(msg, env) {
  const processed = await env.DB.prepare("SELECT 1 FROM processed WHERE msg_id = ?")
    .bind(msg.id).first();
  if (processed) { msg.ack(); return; }

  await sendEmail(msg.body.email);
  await env.DB.prepare("INSERT INTO processed (msg_id) VALUES (?)")
    .bind(msg.id).run();
  msg.ack();
}

Or design the side effect to be idempotent itself (UPSERT, PUT instead of INSERT, flag-set instead of increment).

---

Durable Objects: stateful actors

Mental model

• Namespace: the Worker class registered with Cloudflare. Bound to env.
• ID: identifier for one instance. Got via idFromName(string) (stable hash) or newUniqueId().
• Instance: exactly one isolate on the whole Cloudflare network for that ID. Single-writer, strong consistency.
• Stub: the handle the client uses to call the instance (stub.fetch() or RPC methods).
• Storage: a SQL database local to each instance, transactional.

When to use DOs

• Counters, rate limiters: need atomic increment.
• WebSocket servers: chat rooms, multiplayer games, live collaboration.
• Locks / mutexes: coordinating against external resources.
• Sessions with in-memory state: shopping carts, form wizards.
• Transactions across keys: hard in D1, easy in DO storage.

Defining a DO

// src/durable-objects/counter.ts
export class Counter implements DurableObject {
  state: DurableObjectState;

  constructor(state: DurableObjectState, env: Env) {
    this.state = state;
  }

  async fetch(request: Request): Promise<Response> {
    let count = (await this.state.storage.get<number>("count")) ?? 0;

    const url = new URL(request.url);
    if (url.pathname === "/increment") {
      count++;
      await this.state.storage.put("count", count);
    }

    return Response.json({ count });
  }
}

// src/index.ts
export { Counter } from "./durable-objects/counter";

export default {
  async fetch(request, env) {
    const id = env.COUNTER.idFromName("global");
    const stub = env.COUNTER.get(id);
    return stub.fetch(request);
  }
};

Binding

{
  "durable_objects": {
    "bindings": [
      { "name": "COUNTER", "class_name": "Counter" }
    ]
  },
  "migrations": [
    { "tag": "v1", "new_sqlite_classes": ["Counter"] }
  ]
}

new_sqlite_classes (new-style) is for DOs with SQLite storage (the current default). The old style used new_classes with KV-like storage.

Storage API

// KV-like
await this.state.storage.get("key");
await this.state.storage.put("key", value);
await this.state.storage.delete("key");
await this.state.storage.list({ prefix: "user:" });

// SQL (SQLite-backed, strong consistency per instance)
this.state.storage.sql.exec(
  "CREATE TABLE IF NOT EXISTS rooms (id TEXT PRIMARY KEY, users TEXT)"
);
const rows = this.state.storage.sql.exec(
  "SELECT * FROM rooms WHERE id = ?", roomId
).toArray();

// Transaction
await this.state.storage.transaction(async (txn) => {
  const current = await txn.get<number>("balance");
  await txn.put("balance", current + 100);
});

The single-writer guarantee

One instance ID = one isolate running at a time. All requests to that ID queue inside the instance and are processed sequentially.

Consequences:

• Atomic: a get + put in the same request is atomic, no race.
• Location pinned: the instance runs at the PoP nearest the first request and stays there. Later requests from distant regions pay higher latency.
• Bottleneck: one ID is a single-threaded actor. Throughput is bounded.

→ Shard if you need high throughput:

// Instead of one global counter
env.COUNTER.idFromName("global");

// Shard into 16 counters
const shard = userId.charCodeAt(0) % 16;
env.COUNTER.idFromName(`shard-${shard}`);
// Sum when reading

Hibernation for WebSockets

WebSockets hold connections for a long time → the DO needs to stay alive → resources get tied up. The hibernation API: when idle, evict memory, keep storage, wake up on a new message.

export class ChatRoom implements DurableObject {
  async fetch(request: Request): Promise<Response> {
    if (request.headers.get("Upgrade") !== "websocket") {
      return new Response("Expected WebSocket", { status: 426 });
    }

    const [client, server] = Object.values(new WebSocketPair());

    // Accept with hibernation
    this.state.acceptWebSocket(server);

    return new Response(null, { status: 101, webSocket: client });
  }

  async webSocketMessage(ws: WebSocket, message: string) {
    // Broadcast to all connections
    for (const peer of this.state.getWebSockets()) {
      peer.send(message);
    }
  }

  async webSocketClose(ws: WebSocket, code: number, reason: string) {
    // Cleanup
  }
}

Don’t use the traditional server.addEventListener("message", ...) — that’s non-hibernating. acceptWebSocket() + webSocketMessage handler is the hibernation-enabled form.

---

6 real-world patterns

① Rate limiter (DO)

export class RateLimiter implements DurableObject {
  state: DurableObjectState;

  constructor(state, env) {
    this.state = state;
  }

  async fetch(request: Request) {
    const now = Date.now();
    const windowMs = 60_000;

    let timestamps = (await this.state.storage.get<number[]>("requests")) ?? [];
    timestamps = timestamps.filter(t => now - t < windowMs);

    if (timestamps.length >= 60) {
      return Response.json({ allowed: false, reset: timestamps[0] + windowMs }, { status: 429 });
    }

    timestamps.push(now);
    await this.state.storage.put("requests", timestamps);

    return Response.json({ allowed: true, remaining: 60 - timestamps.length });
  }
}

// Client Worker
async fetch(request, env) {
  const ip = request.headers.get("CF-Connecting-IP") ?? "unknown";
  const id = env.RATE_LIMITER.idFromName(`ip:${ip}`);
  const res = await env.RATE_LIMITER.get(id).fetch(request.url);
  const { allowed } = await res.json();
  if (!allowed) return new Response("Rate limited", { status: 429 });
  // ... handle request
}

One DO instance per IP. Atomic increment, no race.

② Job queue with retry (Queue)

// Producer
async fetch(request, env) {
  await env.JOBS.send({ type: "digest", userId: body.userId });
  return Response.json({ queued: true });
}

// Consumer
async queue(batch, env) {
  for (const msg of batch.messages) {
    try {
      if (msg.body.type === "digest") {
        await sendWeeklyDigest(env, msg.body.userId);
      }
      msg.ack();
    } catch (err) {
      msg.retry({ delaySeconds: Math.pow(2, msg.attempts) * 60 });
    }
  }
}

Exponential backoff: 1m, 2m, 4m… Dead-letter after max_retries.

③ WebSocket chat room (DO)

export class ChatRoom implements DurableObject {
  state: DurableObjectState;

  async fetch(request: Request) {
    if (request.headers.get("Upgrade") === "websocket") {
      const [client, server] = Object.values(new WebSocketPair());
      this.state.acceptWebSocket(server);
      return new Response(null, { status: 101, webSocket: client });
    }
    return new Response("WebSocket only", { status: 426 });
  }

  async webSocketMessage(ws: WebSocket, message: string) {
    const peers = this.state.getWebSockets();
    for (const peer of peers) peer.send(message);

    // Persist the last 50 messages
    this.state.storage.sql.exec(
      "INSERT INTO messages (ts, body) VALUES (?, ?)",
      Date.now(), message
    );
    this.state.storage.sql.exec(
      "DELETE FROM messages WHERE id NOT IN (SELECT id FROM messages ORDER BY ts DESC LIMIT 50)"
    );
  }
}

The client Worker routes /chat/:roomId → env.CHAT_ROOMS.idFromName(roomId).fetch().

④ Fan-out background task (Queue)

// Producer
async scheduled(event, env) {
  const users = await env.DB.prepare("SELECT id FROM subscribers WHERE active = 1").all();
  const messages = users.results.map(u => ({ body: { userId: u.id } }));
  await env.DIGEST_QUEUE.sendBatch(messages);
}

// Consumer (parallel workers)
async queue(batch, env) {
  await Promise.all(batch.messages.map(async msg => {
    try {
      await sendDigestEmail(env, msg.body.userId);
      msg.ack();
    } catch (err) {
      msg.retry();
    }
  }));
}

10k subscribers → 10k messages → consumer processes batches in parallel.

⑤ Counter with sharding (DO)

const SHARDS = 32;

async function incrementViewCount(env: Env, slug: string) {
  const shard = hash(slug + Math.random()) % SHARDS;
  const id = env.COUNTER.idFromName(`view:${slug}:shard:${shard}`);
  await env.COUNTER.get(id).fetch("https://do/increment");
}

async function getViewCount(env: Env, slug: string) {
  let total = 0;
  await Promise.all([...Array(SHARDS)].map(async (_, i) => {
    const id = env.COUNTER.idFromName(`view:${slug}:shard:${i}`);
    const res = await env.COUNTER.get(id).fetch("https://do/read");
    const { count } = await res.json();
    total += count;
  }));
  return total;
}

Why shard: one DO handles ~1k req/s. A popular page’s view count above 1k/s = bottleneck. 32 shards → 32k req/s.

⑥ Mutex lock (DO)

export class Lock implements DurableObject {
  async fetch(request: Request) {
    const url = new URL(request.url);
    if (url.pathname === "/acquire") {
      const held = await this.state.storage.get<number>("heldUntil");
      const now = Date.now();
      if (held && held > now) {
        return Response.json({ acquired: false, waitMs: held - now });
      }
      await this.state.storage.put("heldUntil", now + 30_000);  // 30s TTL
      return Response.json({ acquired: true, ttl: 30 });
    }
    if (url.pathname === "/release") {
      await this.state.storage.delete("heldUntil");
      return Response.json({ released: true });
    }
  }
}

// Client
const id = env.LOCK.idFromName("publish-cron");
const res = await env.LOCK.get(id).fetch("https://do/acquire");
const { acquired } = await res.json();
if (!acquired) return; // another Worker has it, skip
try {
  await runExpensivePublish(env);
} finally {
  await env.LOCK.get(id).fetch("https://do/release");
}

Use when multiple Workers might try to run the same job (cron-trigger spread, webhook retries) — prevents double execution.

---

Queues vs Durable Objects: picking one

Scenario	Primitive
Fire-and-forget background job	Queue
Per-user/IP rate limiter	DO
Retry-with-backoff against external APIs	Queue
WebSocket chat / real-time	DO
Atomic counter	DO
Batch email send	Queue
Session with in-memory state	DO
Webhook replay during downtime	Queue
Lock / mutex	DO
Cross-key transaction	DO
Scheduled fan-out (cron → workers)	Queue (producer in scheduled)

Combining them

A strong pattern: DO coordinator + Queue worker.

// DO receives a real-time event, validates, enqueues tasks
export class OrderRoom implements DurableObject {
  async fetch(request: Request) {
    const order = await request.json();
    // Validate + save state
    await this.state.storage.put(`order:${order.id}`, order);
    // Enqueue async tasks
    await this.env.EMAIL_QUEUE.send({ type: "order-confirm", orderId: order.id });
    await this.env.SHIPPING_QUEUE.send({ type: "ship", orderId: order.id });
    return Response.json({ status: "queued" });
  }
}

DO = source of truth + coordinator. Queue = task runner.

---

Gotchas

① Queue message size limit

Default is 128 KB. Larger payloads → store in R2/D1 and send a pointer:

// WRONG: sending raw image binary through the Queue
await env.QUEUE.send({ type: "resize", image: bigImage });  // overflow

// RIGHT
await env.BUCKET.put(`tmp/${id}`, bigImage);
await env.QUEUE.send({ type: "resize", bucketKey: `tmp/${id}` });

② Consumer timeout

Each batch handler has a wall-time limit (default 30s). A batch of 100 messages at 500ms each = 50s → timeout.

Either reduce max_batch_size or parallelise with Promise.all.

③ DO location lock-in

The instance pins to the PoP nearest the first request. A user in Vietnam creating a room → the DO runs in SIN. A US user joining later → 250ms RTT.

Fixes:

• Shard by region if the use case allows.
• Or accept the trade-off (cross-region real-time is intrinsically hard).
• Or pass locationHint on creation (beta feature).

④ DO migrations can wipe storage

// WRONG: deleting the class in a migration
"migrations": [
  { "tag": "v1", "new_classes": ["Counter"] },
  { "tag": "v2", "deleted_classes": ["Counter"] }  // DESTROYS ALL DATA
]

deleted_classes wipes storage. Only use when you’re absolutely sure no data matters.

⑤ At-least-once but the consumer throws → infinite retries?

No. Cloudflare tracks attempts. After max_retries, it auto-moves to the DLQ.

But if the consumer neither ack()s nor retry()s, the message is auto-retried when the batch times out. The safe pattern: always explicitly ack() or retry().

⑥ DO storage size limit

• KV-like storage: no hard cap (cost-bounded).
• SQLite storage (new-style): 10 GB per DO.
• Over that → shard.

⑦ RPC methods vs fetch

Old API: stub.fetch(request). New API: direct RPC methods.

// Old
await stub.fetch("https://do/increment", { method: "POST" });

// New (Workers 2024)
export class Counter extends DurableObject {
  async increment() {
    // ...
  }
}

// Client
await stub.increment();  // type-safe, no URL construction

RPC is simpler. Check Cloudflare docs for the compatibility date you need.

---

Production checklist

Queue

• Consumer is idempotent (at-least-once delivery).
• max_retries is sensible (3-5), with a DLQ.
• Message bodies < 128 KB; larger payloads via R2 pointers.
• Batch size + timeout fit within the 30s wall-time limit.
• DLQ consumer alerts / logs; doesn’t silently discard.

Durable Object

• ID strategy is clear (stable idFromName or newUniqueId).
• Sharding in place if throughput > 1k req/s.
• Long-lived WebSockets use the hibernation API.
• Storage size is monitored; sharded before hitting 10 GB.
• Migrations don’t delete in-production classes.

---

Wrap-up

Queues and DOs are the only “stateful” primitives Workers has. Queue = async; DO = sync coordination. Knowing both unlocks most of the remaining full-stack use cases.

The strongest pattern is DO-as-coordinator + Queue-as-task-runner. An e-commerce order, a chat app, a notification system — they all look like this.

Block Storage (Parts 5-8) is done. Block 3 (Frameworks) opens with Part 9: Router choice (Hono, Itty, or vanilla).

---

References

• Queues documentation
• Durable Objects overview
• DO Storage API
• WebSocket hibernation