Cloudflare Queues

Cloudflare Queues

Status: Production Ready ✅ Last Updated: 2025-10-21 Dependencies: cloudflare-worker-base (for Worker setup) Latest Versions: wrangler@4.43.0, @cloudflare/workers-types@4.20251014.0

---

Quick Start (10 Minutes)

1. Create a Queue

# Create a new queue
npx wrangler queues create my-queue

# Output:
# ✅ Successfully created queue my-queue

# List all queues
npx wrangler queues list

# Get queue info
npx wrangler queues info my-queue

2. Set Up Producer (Send Messages)

wrangler.jsonc:

{
  "name": "my-producer",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-11",
  "queues": {
    "producers": [
      {
        "binding": "MY_QUEUE",           // Available as env.MY_QUEUE
        "queue": "my-queue"               // Queue name from step 1
      }
    ]
  }
}

src/index.ts (Producer):

import { Hono } from 'hono';

type Bindings = {
  MY_QUEUE: Queue;
};

const app = new Hono<{ Bindings: Bindings }>();

// Send single message
app.post('/send', async (c) => {
  const body = await c.req.json();

  await c.env.MY_QUEUE.send({
    userId: body.userId,
    action: 'process-order',
    timestamp: Date.now(),
  });

  return c.json({ status: 'queued' });
});

// Send batch of messages
app.post('/send-batch', async (c) => {
  const items = await c.req.json();

  await c.env.MY_QUEUE.sendBatch(
    items.map((item) => ({
      body: { userId: item.userId, action: item.action },
    }))
  );

  return c.json({ status: 'queued', count: items.length });
});

export default app;

3. Set Up Consumer (Process Messages)

Create consumer Worker:

# In a new directory
npm create cloudflare@latest my-consumer -- --type hello-world --ts
cd my-consumer

wrangler.jsonc:

{
  "name": "my-consumer",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-11",
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",              // Queue to consume from
        "max_batch_size": 10,             // Process up to 10 messages at once
        "max_batch_timeout": 5            // Or wait max 5 seconds
      }
    ]
  }
}

src/index.ts (Consumer):

export default {
  async queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    console.log(`Processing batch of ${batch.messages.length} messages`);

    for (const message of batch.messages) {
      console.log('Message:', message.id, message.body, `Attempt: ${message.attempts}`);

      // Your processing logic here
      await processMessage(message.body);
    }

    // Implicit acknowledgement: if this function returns without error,
    // all messages are automatically acknowledged
  },
};

async function processMessage(body: any) {
  // Process the message
  console.log('Processing:', body);
}

4. Deploy and Test

# Deploy producer
cd my-producer
npm run deploy

# Deploy consumer
cd my-consumer
npm run deploy

# Test by sending a message
curl -X POST https://my-producer.<your-subdomain>.workers.dev/send \
  -H "Content-Type: application/json" \
  -d '{"userId": "123", "action": "welcome-email"}'

# Watch consumer logs
npx wrangler tail my-consumer

---

Complete Producer API

send() - Send Single Message

interface QueueSendOptions {
  delaySeconds?: number;  // Delay delivery (0-43200 seconds / 12 hours)
}

await env.MY_QUEUE.send(body: any, options?: QueueSendOptions);

Examples:

// Simple send
await env.MY_QUEUE.send({ userId: '123', action: 'send-email' });

// Send with delay (10 minutes)
await env.MY_QUEUE.send(
  { userId: '123', action: 'reminder' },
  { delaySeconds: 600 }
);

// Send structured data
await env.MY_QUEUE.send({
  type: 'order-confirmation',
  orderId: 'ORD-123',
  email: 'user@example.com',
  items: [{ sku: 'ITEM-1', quantity: 2 }],
  total: 49.99,
  timestamp: Date.now(),
});

CRITICAL:

• Message body must be JSON serializable (structured clone algorithm)
• Maximum message size: 128 KB (including ~100 bytes internal metadata)
• Messages >128 KB will fail - split them or store in R2 and send reference

---

sendBatch() - Send Multiple Messages

interface MessageSendRequest<Body = any> {
  body: Body;
  delaySeconds?: number;
}

interface QueueSendBatchOptions {
  delaySeconds?: number;  // Default delay for all messages
}

await env.MY_QUEUE.sendBatch(
  messages: Iterable<MessageSendRequest>,
  options?: QueueSendBatchOptions
);

Examples:

// Send batch of messages
await env.MY_QUEUE.sendBatch([
  { body: { userId: '1', action: 'email' } },
  { body: { userId: '2', action: 'email' } },
  { body: { userId: '3', action: 'email' } },
]);

// Send batch with individual delays
await env.MY_QUEUE.sendBatch([
  { body: { task: 'task1' }, delaySeconds: 60 },   // 1 min
  { body: { task: 'task2' }, delaySeconds: 300 },  // 5 min
  { body: { task: 'task3' }, delaySeconds: 600 },  // 10 min
]);

// Send batch with default delay
await env.MY_QUEUE.sendBatch(
  [
    { body: { task: 'task1' } },
    { body: { task: 'task2' } },
  ],
  { delaySeconds: 3600 } // All delayed by 1 hour
);

// Dynamic batch from array
const tasks = await getTasks();
await env.MY_QUEUE.sendBatch(
  tasks.map((task) => ({
    body: {
      taskId: task.id,
      userId: task.userId,
      priority: task.priority,
    },
  }))
);

Limits:

• Maximum 100 messages per batch
• Maximum 256 KB total batch size
• Each message still limited to 128 KB individually

---

Complete Consumer API

Queue Handler Function

export default {
  async queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    // Process messages
  },
};

Parameters:

• batch - MessageBatch object containing messages
• env - Environment bindings (KV, D1, R2, etc.)
• ctx - Execution context for waitUntil(), passThroughOnException()

---

MessageBatch Interface

interface MessageBatch<Body = unknown> {
  readonly queue: string;              // Queue name
  readonly messages: Message<Body>[];  // Array of messages
  ackAll(): void;                      // Acknowledge all messages
  retryAll(options?: QueueRetryOptions): void;  // Retry all messages
}

Properties:

• `queue` - Name of the queue this batch came from
• Useful when one consumer handles multiple queues

• `messages` - Array of Message objects
• Ordering is best effort, not guaranteed
• Process order should not be assumed

Methods:

• `ackAll()` - Mark all messages as successfully delivered
• Even if handler throws error, these messages won't retry
• Use when you've safely processed all messages

• `retryAll(options?)` - Mark all messages for retry
• Messages re-queued immediately (or after delay)
• Counts towards max_retries limit

---

Message Interface

interface Message<Body = unknown> {
  readonly id: string;          // Unique message ID
  readonly timestamp: Date;     // When message was sent
  readonly body: Body;          // Message content
  readonly attempts: number;    // Retry count (starts at 1)
  ack(): void;                  // Acknowledge this message
  retry(options?: QueueRetryOptions): void;  // Retry this message
}

Properties:

• `id` - System-generated unique ID (UUID)
• `timestamp` - Date object when message was sent to queue
• `body` - Your message content (any JSON serializable type)
• `attempts` - Number of times consumer has processed this message
• Starts at 1 on first delivery
• Increments on each retry
• Use for exponential backoff: delaySeconds: 60 * message.attempts

Methods:

• `ack()` - Mark message as successfully delivered
• Message won't be retried even if handler fails later
• Critical for non-idempotent operations (DB writes, API calls)

• `retry(options?)` - Mark message for retry
• Re-queued immediately or after delaySeconds
• Counts towards max_retries (default 3)

---

QueueRetryOptions

interface QueueRetryOptions {
  delaySeconds?: number;  // Delay retry (0-43200 seconds / 12 hours)
}

Example:

// Retry immediately
message.retry();

// Retry after 5 minutes
message.retry({ delaySeconds: 300 });

// Exponential backoff based on attempts
message.retry({
  delaySeconds: Math.min(60 * Math.pow(2, message.attempts - 1), 3600),
});

---

Consumer Patterns

1. Basic Consumer (Implicit Acknowledgement)

Best for: Idempotent operations where retries are safe

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // Process message
      await sendEmail(message.body.email, message.body.content);
    }

    // Implicit ack: returning successfully acknowledges ALL messages
  },
};

Behavior:

• If function returns successfully → all messages acknowledged
• If function throws error → all messages retried
• Simple but can cause duplicate processing on partial failures

---

2. Explicit Acknowledgement (Non-Idempotent Operations)

Best for: Database writes, API calls, financial transactions

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        // Non-idempotent operation
        await env.DB.prepare(
          'INSERT INTO orders (id, user_id, amount) VALUES (?, ?, ?)'
        ).bind(message.body.orderId, message.body.userId, message.body.amount).run();

        // Explicitly acknowledge success
        message.ack();
      } catch (error) {
        console.error(`Failed to process ${message.id}:`, error);

        // Don't ack - will retry (or let it fail)
        // Optionally: message.retry() for explicit retry
      }
    }
  },
};

Why explicit ack?

• Prevents duplicate writes if one message in batch fails
• Only successfully processed messages are acknowledged
• Failed messages retry independently

---

3. Retry with Exponential Backoff

Best for: Rate-limited APIs, temporary failures

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        // Call rate-limited API
        await fetch('https://api.example.com/process', {
          method: 'POST',
          body: JSON.stringify(message.body),
        });

        message.ack();
      } catch (error) {
        if (error.status === 429) {
          // Rate limited - retry with exponential backoff
          const delaySeconds = Math.min(
            60 * Math.pow(2, message.attempts - 1), // 60s, 120s, 240s, ...
            3600 // Max 1 hour
          );

          console.log(`Rate limited. Retrying in ${delaySeconds}s (attempt ${message.attempts})`);
          message.retry({ delaySeconds });
        } else {
          // Other error - retry immediately
          message.retry();
        }
      }
    }
  },
};

---

4. Dead Letter Queue (DLQ) Pattern

Best for: Handling permanently failed messages

Setup DLQ:

# Create DLQ
npx wrangler queues create my-dlq

# Configure consumer with DLQ

wrangler.jsonc:

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10,
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"  // Failed messages go here
      }
    ]
  }
}

DLQ Consumer:

// Consumer for dead letter queue
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // Log failed message
      console.error('PERMANENTLY FAILED MESSAGE:', {
        id: message.id,
        attempts: message.attempts,
        body: message.body,
        timestamp: message.timestamp,
      });

      // Store in database for manual review
      await env.DB.prepare(
        'INSERT INTO failed_messages (id, body, attempts, failed_at) VALUES (?, ?, ?, ?)'
      ).bind(
        message.id,
        JSON.stringify(message.body),
        message.attempts,
        new Date().toISOString()
      ).run();

      // Optionally: send alert to ops team
      await sendAlert({
        type: 'queue-dlq',
        messageId: message.id,
        queue: batch.queue,
      });

      // Acknowledge to remove from DLQ
      message.ack();
    }
  },
};

How it works: 1. Message fails in main queue 2. Retries up to max_retries (default 3) 3. After max retries, sent to DLQ 4. DLQ consumer processes failed messages 5. Without DLQ, messages are deleted permanently

---

5. Multiple Queues, Single Consumer

Best for: Centralized processing logic

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // Switch based on queue name
    switch (batch.queue) {
      case 'high-priority-queue':
        await processHighPriority(batch.messages, env);
        break;

      case 'low-priority-queue':
        await processLowPriority(batch.messages, env);
        break;

      case 'email-queue':
        await processEmails(batch.messages, env);
        break;

      default:
        console.warn(`Unknown queue: ${batch.queue}`);
        // Log to DLQ or monitoring
    }
  },
};

async function processHighPriority(messages: Message[], env: Env) {
  for (const message of messages) {
    // Process with urgency
    await fastProcess(message.body);
    message.ack();
  }
}

async function processLowPriority(messages: Message[], env: Env) {
  for (const message of messages) {
    // Can take longer
    await slowProcess(message.body);
    message.ack();
  }
}

wrangler.jsonc:

{
  "queues": {
    "consumers": [
      { "queue": "high-priority-queue" },
      { "queue": "low-priority-queue" },
      { "queue": "email-queue" }
    ]
  }
}

---

Consumer Configuration

Batch Settings

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 100,        // 1-100 messages (default: 10)
        "max_batch_timeout": 30       // 0-60 seconds (default: 5)
      }
    ]
  }
}

How batching works:

• Consumer called when either condition met:
• max_batch_size messages accumulated
• max_batch_timeout seconds elapsed (whichever comes first)

Example:

• max_batch_size: 100, max_batch_timeout: 10
• If 100 messages arrive in 3 seconds → batch delivered immediately
• If only 50 messages arrive in 10 seconds → batch of 50 delivered

Tuning guidelines:

• High volume, low latency → max_batch_size: 100, max_batch_timeout: 1
• Low volume, batch writes → max_batch_size: 50, max_batch_timeout: 30
• Cost optimization → Larger batches = fewer invocations

---

Retry Settings

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_retries": 5,             // 0-100 (default: 3)
        "retry_delay": 300            // Seconds (default: 0)
      }
    ]
  }
}

`max_retries`:

• Number of times to retry failed message
• After max retries:
• With DLQ → message sent to DLQ
• Without DLQ → message deleted permanently

`retry_delay`:

• Default delay for all retried messages (seconds)
• Can be overridden with message.retry({ delaySeconds })
• Maximum delay: 43200 seconds (12 hours)

---

Concurrency Settings

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_concurrency": 10         // 1-250 (default: auto-scale)
      }
    ]
  }
}

How concurrency works:

• Queues auto-scales consumers based on backlog
• Default: Scales up to 250 concurrent invocations
• Setting max_concurrency limits scaling

When to set max_concurrency:

• ✅ Upstream API has rate limits
• ✅ Database connection limits
• ✅ Want to control costs
• ❌ Most cases - leave unset for best performance

Auto-scaling triggers:

• Growing backlog (messages accumulating)
• High error rate
• Processing speed vs. incoming rate

---

Dead Letter Queue

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"  // Name of DLQ
      }
    ]
  }
}

CRITICAL:

• DLQ must be created separately: npx wrangler queues create my-dlq
• Without DLQ, failed messages are deleted permanently
• Messages in DLQ persist for 4 days without consumer
• Always configure DLQ for production queues

---

Wrangler Commands

Create Queue

# Create queue with defaults
npx wrangler queues create my-queue

# Create with custom retention (4 days default, max 14 days)
npx wrangler queues create my-queue --message-retention-period-secs 1209600

# Create with delivery delay
npx wrangler queues create my-queue --delivery-delay-secs 60

---

List Queues

npx wrangler queues list

---

Get Queue Info

npx wrangler queues info my-queue

# Output shows:
# - Queue name
# - Message retention period
# - Delivery delay
# - Consumers
# - Backlog size

---

Update Queue

# Update message retention (60 seconds to 14 days)
npx wrangler queues update my-queue --message-retention-period-secs 604800

# Update delivery delay (0 to 12 hours)
npx wrangler queues update my-queue --delivery-delay-secs 3600

---

Delete Queue

npx wrangler queues delete my-queue

# Warning: This deletes ALL messages in the queue!
# Use with caution in production

---

Consumer Management

# Add consumer to queue
npx wrangler queues consumer add my-queue my-consumer-worker \
  --batch-size 50 \
  --batch-timeout 10 \
  --message-retries 5 \
  --max-concurrency 20 \
  --retry-delay-secs 300

# Remove consumer from queue
npx wrangler queues consumer remove my-queue my-consumer-worker

---

Purge Queue

# Delete ALL messages in queue (use with caution!)
npx wrangler queues purge my-queue

---

Pause/Resume Delivery

# Pause message delivery to consumers
npx wrangler queues pause-delivery my-queue

# Resume message delivery
npx wrangler queues resume-delivery my-queue

Use cases:

• Maintenance on consumer Workers
• Temporarily stop processing
• Debug issues without losing messages

---

Limits & Quotas

---

Pricing

Requires Workers Paid plan ($5/month)

Operations Pricing:

• First 1,000,000 operations/month: FREE
• After that: $0.40 per million operations

What counts as an operation:

• Each 64 KB chunk written, read, or deleted
• Messages >64 KB count as multiple operations:
• 65 KB message = 2 operations
• 127 KB message = 2 operations
• 128 KB message = 2 operations

Typical message lifecycle:

• 1 write + 1 read + 1 delete = 3 operations

Retries:

• Each retry = additional read operation
• Message retried 3 times = 1 write + 4 reads + 1 delete = 6 operations

Dead Letter Queue:

• Writing to DLQ = additional write operation

Cost examples:

• 1M messages/month (no retries): ((1M × 3) - 1M) / 1M × $0.40 = $0.80
• 10M messages/month: ((10M × 3) - 1M) / 1M × $0.40 = $11.60
• 100M messages/month: ((100M × 3) - 1M) / 1M × $0.40 = $119.60

---

TypeScript Types

// Queue binding (producer)
interface Queue<Body = any> {
  send(body: Body, options?: QueueSendOptions): Promise<void>;
  sendBatch(
    messages: Iterable<MessageSendRequest<Body>>,
    options?: QueueSendBatchOptions
  ): Promise<void>;
}

interface QueueSendOptions {
  delaySeconds?: number;
}

interface MessageSendRequest<Body = any> {
  body: Body;
  delaySeconds?: number;
}

interface QueueSendBatchOptions {
  delaySeconds?: number;
}

// Consumer handler
export default {
  queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void>;
}

interface MessageBatch<Body = unknown> {
  readonly queue: string;
  readonly messages: Message<Body>[];
  ackAll(): void;
  retryAll(options?: QueueRetryOptions): void;
}

interface Message<Body = unknown> {
  readonly id: string;
  readonly timestamp: Date;
  readonly body: Body;
  readonly attempts: number;
  ack(): void;
  retry(options?: QueueRetryOptions): void;
}

interface QueueRetryOptions {
  delaySeconds?: number;
}

---

Error Handling

Common Errors

1. Message Too Large

// ❌ Bad: Message >128 KB
await env.MY_QUEUE.send({
  data: largeArray, // >128 KB
});

// ✅ Good: Check size before sending
const message = { data: largeArray };
const size = new TextEncoder().encode(JSON.stringify(message)).length;

if (size > 128000) {
  // Store in R2, send reference
  const key = `messages/${crypto.randomUUID()}.json`;
  await env.MY_BUCKET.put(key, JSON.stringify(message));
  await env.MY_QUEUE.send({ type: 'large-message', r2Key: key });
} else {
  await env.MY_QUEUE.send(message);
}

---

2. Throughput Exceeded

// ❌ Bad: Exceeding 5000 msg/s per queue
for (let i = 0; i < 10000; i++) {
  await env.MY_QUEUE.send({ id: i }); // Too fast!
}

// ✅ Good: Use sendBatch
const messages = Array.from({ length: 10000 }, (_, i) => ({
  body: { id: i },
}));

// Send in batches of 100
for (let i = 0; i < messages.length; i += 100) {
  await env.MY_QUEUE.sendBatch(messages.slice(i, i + 100));
}

// ✅ Even better: Rate limit with delay
for (let i = 0; i < messages.length; i += 100) {
  await env.MY_QUEUE.sendBatch(messages.slice(i, i + 100));
  if (i + 100 < messages.length) {
    await new Promise(resolve => setTimeout(resolve, 100)); // 100ms delay
  }
}

---

3. Consumer Timeout

// ❌ Bad: Long processing without CPU limit increase
export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      await processForMinutes(message.body); // CPU timeout!
    }
  },
};

// ✅ Good: Increase CPU limit in wrangler.jsonc

wrangler.jsonc:

{
  "limits": {
    "cpu_ms": 300000  // 5 minutes (max allowed)
  }
}

---

4. Backlog Growing

// Issue: Consumer too slow, backlog growing

// ✅ Solution 1: Increase batch size
{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100  // Process more per invocation
    }]
  }
}

// ✅ Solution 2: Let concurrency auto-scale (don't set max_concurrency)

// ✅ Solution 3: Optimize consumer code
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // Process in parallel
    await Promise.all(
      batch.messages.map(async (message) => {
        await process(message.body);
        message.ack();
      })
    );
  },
};

---

Always Do ✅

1. Configure Dead Letter Queue for production queues 2. Use explicit ack() for non-idempotent operations (DB writes, API calls) 3. Validate message size before sending (<128 KB) 4. Use sendBatch() for multiple messages (more efficient) 5. Implement exponential backoff for retries 6. Set appropriate batch settings based on workload 7. Monitor queue backlog and consumer errors 8. Use ctx.waitUntil() for async cleanup in consumers 9. Handle errors gracefully - log, alert, retry 10. Let concurrency auto-scale (don't set max_concurrency unless needed)

---

Never Do ❌

1. Never assume message ordering - not guaranteed FIFO 2. Never rely on implicit ack for non-idempotent ops - use explicit ack() 3. Never send messages >128 KB - will fail 4. Never delete queues with active messages - data loss 5. Never skip DLQ configuration in production 6. Never exceed 5000 msg/s per queue - rate limit error 7. Never process messages synchronously in loop - use Promise.all() 8. Never ignore message.attempts - use for backoff logic 9. Never set max_concurrency=1 unless you have a very specific reason 10. Never forget to ack() in explicit acknowledgement patterns

---

Troubleshooting

Issue: Messages not being delivered to consumer

Possible causes: 1. Consumer not deployed 2. Wrong queue name in wrangler.jsonc 3. Delivery paused 4. Consumer throwing errors

Solution:

# Check queue info
npx wrangler queues info my-queue

# Check if delivery paused
npx wrangler queues resume-delivery my-queue

# Check consumer logs
npx wrangler tail my-consumer

---

Issue: Entire batch retried when one message fails

Cause: Using implicit acknowledgement with non-idempotent operations

Solution: Use explicit ack()

// ✅ Explicit ack
for (const message of batch.messages) {
  try {
    await dbWrite(message.body);
    message.ack(); // Only ack on success
  } catch (error) {
    console.error(`Failed: ${message.id}`);
    // Don't ack - will retry
  }
}

---

Issue: Messages deleted without processing

Cause: No Dead Letter Queue configured

Solution:

# Create DLQ
npx wrangler queues create my-dlq

# Add to consumer config

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "dead_letter_queue": "my-dlq"
    }]
  }
}

---

Issue: Consumer not auto-scaling

Possible causes: 1. max_concurrency set to 1 2. Consumer returning errors (not processing) 3. Batch processing too fast (no backlog)

Solution:

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      // Don't set max_concurrency - let it auto-scale
      "max_batch_size": 50  // Increase batch size instead
    }]
  }
}

---

Production Checklist

Before deploying to production:

• [ ] Dead Letter Queue created and configured
• [ ] Explicit ack() used for non-idempotent operations
• [ ] Message size validation (<128 KB)
• [ ] Batch settings optimized for workload
• [ ] Retry settings configured (max_retries, retry_delay)
• [ ] Consumer concurrency settings appropriate
• [ ] Error handling and logging implemented
• [ ] Monitoring and alerting set up
• [ ] DLQ consumer deployed
• [ ] Exponential backoff for rate-limited APIs
• [ ] CPU limits increased if needed (limits.cpu_ms)
• [ ] Message retention period appropriate
• [ ] Tested under load

---

Related Documentation

• Cloudflare Queues Docs
• How Queues Works
• JavaScript APIs
• Batching & Retries
• Consumer Concurrency
• Dead Letter Queues
• Wrangler Commands
• Limits
• Pricing

---

Last Updated: 2025-10-21 Version: 1.0.0 Maintainer: Jeremy Dawes | jeremy@jezweb.net

Cloudflare Queues

Complete knowledge domain for Cloudflare Queues - flexible message queue for asynchronous processing and background tasks on Workers.

---

Auto-Trigger Keywords

Primary Keywords

• cloudflare queues
• queues workers
• message queue
• queue bindings
• async processing
• background jobs
• queue consumer
• queue producer
• workers queue
• message broker

Secondary Keywords

• batch processing
• send queue
• sendBatch
• queue handler
• message retry
• queue ack
• acknowledge message
• dead letter queue
• dlq
• queue delay
• delay message
• consumer concurrency
• queue backlog
• wrangler queues

Error-Based Keywords

• queue timeout
• batch retry
• message lost
• queue throughput exceeded
• consumer not scaling
• queue storage limit
• Too Many Requests queue
• message retention exceeded
• visibility timeout
• max retries reached
• queue delivery failed
• consumer error

Framework Integration Keywords

• hono queues
• queues hono
• cloudflare workers queues
• wrangler queue
• queue api workers

---

What This Skill Does

This skill provides complete Cloudflare Queues knowledge including:

• ✅ Queue Creation - Create, configure, and manage queues with wrangler
• ✅ Producer Patterns - send(), sendBatch() with delays
• ✅ Consumer Patterns - Basic, explicit ack, DLQ, retry strategies
• ✅ Batching Configuration - max_batch_size, max_batch_timeout
• ✅ Retry Strategies - Implicit, explicit, with exponential backoff
• ✅ Dead Letter Queues - Handle permanently failed messages
• ✅ Explicit Acknowledgement - ack(), retry(), ackAll(), retryAll()
• ✅ Message Delays - Delay delivery up to 12 hours
• ✅ Consumer Concurrency - Auto-scaling up to 250 concurrent invocations
• ✅ Error Handling - Timeouts, retries, rate limits
• ✅ Pull Consumers - HTTP-based consumers (not just Workers)
• ✅ Performance Optimization - Batching, concurrency, cost reduction

---

Known Issues Prevented

---

When to Use This Skill

✅ Use this skill when:

• Decoupling application components
• Processing tasks asynchronously
• Handling background jobs (emails, notifications, data processing)
• Buffering calls to external APIs
• Implementing task queues
• Managing workload spikes
• Rate limiting upstream services
• Building event-driven architectures
• Processing webhooks asynchronously
• Scheduling delayed tasks

❌ When NOT to use:

• You need real-time processing (<1 second latency)
• You need guaranteed FIFO ordering (use Durable Objects)
• You need strict message ordering
• Messages >128 KB (split them or use R2 references)
• You need synchronous responses
• You need transactional guarantees across operations

---

Quick Example

import { Hono } from 'hono';

type Bindings = {
  MY_QUEUE: Queue;
};

const app = new Hono<{ Bindings: Bindings }>();

// Producer: Send message to queue
app.post('/tasks', async (c) => {
  const task = await c.req.json();

  // Send single message
  await c.env.MY_QUEUE.send({
    type: 'process-order',
    orderId: task.orderId,
    userId: task.userId,
  });

  return c.json({ status: 'queued' });
});

// Producer: Send batch of messages
app.post('/tasks/batch', async (c) => {
  const tasks = await c.req.json();

  // Send up to 100 messages at once
  await c.env.MY_QUEUE.sendBatch(
    tasks.map((task) => ({
      body: {
        type: 'process-order',
        orderId: task.orderId,
      },
    }))
  );

  return c.json({ status: 'queued', count: tasks.length });
});

export default app;

// Consumer: Process messages from queue
export default {
  async queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    // Process each message
    for (const message of batch.messages) {
      try {
        // Your processing logic
        await processOrder(message.body);

        // Explicitly acknowledge success
        message.ack();
      } catch (error) {
        console.error(`Failed to process message ${message.id}:`, error);

        // Retry with exponential backoff
        message.retry({
          delaySeconds: Math.min(60 * message.attempts, 3600),
        });
      }
    }
  },
};

---

Token Efficiency

• Manual Setup: 8,000-12,000 tokens
• With This Skill: 3,500-5,500 tokens
• Savings: ~50-55%

---

Files Included

• SKILL.md - Complete Queues knowledge domain
• templates/wrangler-queues-config.jsonc - Producer + Consumer bindings
• templates/queues-producer.ts - Send messages (single + batch)
• templates/queues-consumer-basic.ts - Basic consumer (implicit ack)
• templates/queues-consumer-explicit-ack.ts - Explicit ack pattern
• templates/queues-dlq-pattern.ts - Dead letter queue setup
• templates/queues-retry-with-delay.ts - Retry with exponential backoff
• reference/wrangler-commands.md - Complete CLI reference
• reference/producer-api.md - send/sendBatch API details
• reference/consumer-api.md - queue handler + batch operations
• reference/best-practices.md - Patterns, concurrency, optimization

---

Dependencies

• cloudflare-worker-base - For Hono + Vite + Worker setup
• wrangler - For queue management CLI

---

Production Status

✅ Production Ready

This skill is based on:

• Official Cloudflare Queues documentation
• Cloudflare Workers SDK examples
• Production-tested patterns
• Latest package versions (verified 2025-10-21)

---

Related Skills

• cloudflare-worker-base - Base Worker setup with Hono
• cloudflare-d1 - Serverless SQLite database
• cloudflare-r2 - Object storage
• cloudflare-kv - Key-value storage
• cloudflare-workers-ai - AI inference on Workers

---

Last Updated: 2025-10-21 Status: Production Ready ✅ Maintainer: Jeremy Dawes | jeremy@jezweb.net

Cloudflare Queues Best Practices

Production patterns, optimization strategies, and common pitfalls.

---

Consumer Design Patterns

1. Explicit Acknowledgement for Non-Idempotent Operations

Problem: Database writes or API calls get duplicated when batch retries

Solution: Use explicit ack() for each message

// ❌ Bad: Entire batch retried if one operation fails
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      await env.DB.prepare(
        'INSERT INTO orders (id, data) VALUES (?, ?)'
      ).bind(message.body.id, JSON.stringify(message.body)).run();
    }
    // If last insert fails, ALL inserts are retried → duplicates!
  },
};

// ✅ Good: Each message acknowledged individually
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        await env.DB.prepare(
          'INSERT INTO orders (id, data) VALUES (?, ?)'
        ).bind(message.body.id, JSON.stringify(message.body)).run();

        message.ack(); // Only ack on success
      } catch (error) {
        console.error(`Failed: ${message.id}`, error);
        // Don't ack - will retry this message only
      }
    }
  },
};

---

2. Exponential Backoff for Rate Limits

Problem: Retrying immediately hits same rate limit

Solution: Use exponential backoff based on attempts

// ❌ Bad: Retry immediately
try {
  await callRateLimitedAPI();
  message.ack();
} catch (error) {
  message.retry(); // Immediately hits rate limit again
}

// ✅ Good: Exponential backoff
try {
  await callRateLimitedAPI();
  message.ack();
} catch (error) {
  if (error.status === 429) {
    const delaySeconds = Math.min(
      60 * Math.pow(2, message.attempts - 1), // 1m, 2m, 4m, 8m, ...
      3600 // Max 1 hour
    );

    console.log(`Rate limited. Retrying in ${delaySeconds}s`);
    message.retry({ delaySeconds });
  }
}

---

3. Always Configure Dead Letter Queue

Problem: Messages deleted permanently after max retries

Solution: Always configure DLQ in production

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"  // ✅ Always configure
      }
    ]
  }
}

DLQ Consumer:

// Monitor and alert on DLQ messages
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // Log failure
      console.error('PERMANENT FAILURE:', message.id, message.body);

      // Store for manual review
      await env.DB.prepare(
        'INSERT INTO failed_messages (id, body, attempts) VALUES (?, ?, ?)'
      ).bind(message.id, JSON.stringify(message.body), message.attempts).run();

      // Send alert
      await sendAlert(`Message ${message.id} failed permanently`);

      message.ack();
    }
  },
};

---

Batch Configuration

Optimizing Batch Size

High volume, low latency:

{
  "queues": {
    "consumers": [{
      "queue": "high-volume-queue",
      "max_batch_size": 100,      // Max messages per batch
      "max_batch_timeout": 1      // Process ASAP
    }]
  }
}

Low volume, batch efficiency:

{
  "queues": {
    "consumers": [{
      "queue": "low-volume-queue",
      "max_batch_size": 50,       // Medium batch
      "max_batch_timeout": 30     // Wait for batch to fill
    }]
  }
}

Cost optimization:

{
  "queues": {
    "consumers": [{
      "queue": "cost-optimized",
      "max_batch_size": 100,      // Largest batches
      "max_batch_timeout": 60     // Max wait time
    }]
  }
}

---

Concurrency Management

Let It Auto-Scale (Default)

{
  "queues": {
    "consumers": [{
      "queue": "my-queue"
      // No max_concurrency - auto-scales to 250
    }]
  }
}

✅ Use when:

• Default case
• Want best performance
• No upstream rate limits

---

Limit Concurrency

{
  "queues": {
    "consumers": [{
      "queue": "rate-limited-api-queue",
      "max_concurrency": 10       // Limit to 10 concurrent consumers
    }]
  }
}

✅ Use when:

• Calling rate-limited APIs
• Database connection limits
• Want to control costs
• Protecting upstream services

---

Message Design

Include Metadata

// ✅ Good: Include helpful metadata
await env.MY_QUEUE.send({
  // Message type for routing
  type: 'order-confirmation',

  // Idempotency key
  idempotencyKey: crypto.randomUUID(),

  // Correlation ID for tracing
  correlationId: requestId,

  // Timestamps
  createdAt: Date.now(),
  scheduledFor: Date.now() + 3600000,

  // Version for schema evolution
  _version: 1,

  // Actual payload
  payload: {
    orderId: 'ORD-123',
    userId: 'USER-456',
    total: 99.99,
  },
});

---

Message Versioning

// Handle multiple message versions
export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      switch (message.body._version) {
        case 1:
          await processV1(message.body);
          break;
        case 2:
          await processV2(message.body);
          break;
        default:
          console.warn(`Unknown version: ${message.body._version}`);
      }

      message.ack();
    }
  },
};

---

Large Messages

Problem: Messages >128 KB fail

Solution: Store in R2, send reference

// Producer
const message = { largeData: ... };
const size = new TextEncoder().encode(JSON.stringify(message)).length;

if (size > 128 * 1024) {
  // Store in R2
  const key = `messages/${crypto.randomUUID()}.json`;
  await env.MY_BUCKET.put(key, JSON.stringify(message));

  // Send reference
  await env.MY_QUEUE.send({
    type: 'large-message',
    r2Key: key,
    size,
    timestamp: Date.now(),
  });
} else {
  await env.MY_QUEUE.send(message);
}

// Consumer
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      if (message.body.type === 'large-message') {
        // Fetch from R2
        const obj = await env.MY_BUCKET.get(message.body.r2Key);
        const data = await obj.json();

        await processLargeMessage(data);

        // Clean up R2
        await env.MY_BUCKET.delete(message.body.r2Key);
      } else {
        await processMessage(message.body);
      }

      message.ack();
    }
  },
};

---

Error Handling

Different Retry Strategies by Error Type

try {
  await processMessage(message.body);
  message.ack();
} catch (error) {
  // Rate limit - exponential backoff
  if (error.status === 429) {
    message.retry({
      delaySeconds: Math.min(60 * Math.pow(2, message.attempts - 1), 3600),
    });
  }
  // Server error - shorter backoff
  else if (error.status >= 500) {
    message.retry({ delaySeconds: 60 });
  }
  // Client error - don't retry
  else if (error.status >= 400) {
    console.error('Client error, not retrying:', error);
    // Don't ack or retry - goes to DLQ
  }
  // Unknown error - retry immediately
  else {
    message.retry();
  }
}

---

Circuit Breaker Pattern

class CircuitBreaker {
  private failures = 0;
  private lastFailure = 0;
  private state: 'closed' | 'open' | 'half-open' = 'closed';

  async call<T>(fn: () => Promise<T>): Promise<T> {
    if (this.state === 'open') {
      // Check if we should try again
      if (Date.now() - this.lastFailure > 60000) { // 1 minute
        this.state = 'half-open';
      } else {
        throw new Error('Circuit breaker is open');
      }
    }

    try {
      const result = await fn();

      // Success - reset
      if (this.state === 'half-open') {
        this.state = 'closed';
        this.failures = 0;
      }

      return result;
    } catch (error) {
      this.failures++;
      this.lastFailure = Date.now();

      // Open circuit after 3 failures
      if (this.failures >= 3) {
        this.state = 'open';
      }

      throw error;
    }
  }
}

const breaker = new CircuitBreaker();

export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      try {
        await breaker.call(() => callUpstreamAPI(message.body));
        message.ack();
      } catch (error) {
        if (error.message === 'Circuit breaker is open') {
          // Retry later when circuit might be closed
          message.retry({ delaySeconds: 120 });
        } else {
          message.retry({ delaySeconds: 60 });
        }
      }
    }
  },
};

---

Cost Optimization

Batch Operations

// ❌ Bad: 100 operations (3 per message)
for (let i = 0; i < 100; i++) {
  await env.MY_QUEUE.send({ id: i });
}

// ✅ Good: 3 operations total (write batch, read batch, delete batch)
await env.MY_QUEUE.sendBatch(
  Array.from({ length: 100 }, (_, i) => ({
    body: { id: i },
  }))
);

Larger Batches

// Process more messages per invocation
{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100  // ✅ Max batch size = fewer invocations
    }]
  }
}

---

Monitoring & Observability

Structured Logging

export default {
  async queue(batch: MessageBatch): Promise<void> {
    console.log(JSON.stringify({
      event: 'batch_start',
      queue: batch.queue,
      messageCount: batch.messages.length,
      timestamp: Date.now(),
    }));

    let processed = 0;
    let failed = 0;

    for (const message of batch.messages) {
      try {
        await processMessage(message.body);
        message.ack();
        processed++;
      } catch (error) {
        console.error(JSON.stringify({
          event: 'message_failed',
          messageId: message.id,
          attempts: message.attempts,
          error: error.message,
        }));
        failed++;
      }
    }

    console.log(JSON.stringify({
      event: 'batch_complete',
      processed,
      failed,
      duration: Date.now() - batch.messages[0].timestamp.getTime(),
    }));
  },
};

Metrics Tracking

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    const startTime = Date.now();

    for (const message of batch.messages) {
      const msgStartTime = Date.now();

      try {
        await processMessage(message.body);
        message.ack();

        // Track processing time
        await env.METRICS.put(
          `processing_time:${Date.now()}`,
          String(Date.now() - msgStartTime)
        );
      } catch (error) {
        await env.METRICS.put(
          `errors:${Date.now()}`,
          JSON.stringify({
            messageId: message.id,
            error: error.message,
          })
        );
      }
    }

    // Track batch metrics
    await env.METRICS.put(
      `batch_size:${Date.now()}`,
      String(batch.messages.length)
    );
  },
};

---

Testing

Local Development

# Start local dev server
npm run dev

# In another terminal, send test messages
curl -X POST http://localhost:8787/send \
  -H "Content-Type: application/json" \
  -d '{"test": "message"}'

# Watch consumer logs
npx wrangler tail my-consumer --local

Unit Tests

import { describe, it, expect } from 'vitest';

describe('Queue Consumer', () => {
  it('processes messages correctly', async () => {
    const batch: MessageBatch = {
      queue: 'test-queue',
      messages: [
        {
          id: '123',
          timestamp: new Date(),
          body: { type: 'test', data: 'hello' },
          attempts: 1,
          ack: () => {},
          retry: () => {},
        },
      ],
      ackAll: () => {},
      retryAll: () => {},
    };

    const env = {
      // Mock bindings
    };

    const ctx = {
      waitUntil: () => {},
      passThroughOnException: () => {},
    };

    await worker.queue(batch, env, ctx);

    // Assert expectations
  });
});

---

Last Updated: 2025-10-21

Consumer API Reference

Complete reference for consuming messages from Cloudflare Queues.

---

Queue Handler

Consumer Workers must export a queue() handler:

export default {
  async queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    // Process messages
  },
};

Parameters

• `batch` - MessageBatch object containing messages
• `env` - Environment bindings (KV, D1, R2, etc.)
• `ctx` - Execution context
• waitUntil(promise) - Extend Worker lifetime
• passThroughOnException() - Continue on error

Return Value

• Must return Promise<void> or void
• Throwing error = all unacknowledged messages retried
• Returning successfully = implicit ack for messages without explicit ack()

---

MessageBatch Interface

interface MessageBatch<Body = unknown> {
  readonly queue: string;
  readonly messages: Message<Body>[];
  ackAll(): void;
  retryAll(options?: QueueRetryOptions): void;
}

Properties

queue (string)

Name of the queue this batch came from.

Use case: One consumer handling multiple queues

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    switch (batch.queue) {
      case 'high-priority':
        await processUrgent(batch.messages);
        break;
      case 'low-priority':
        await processNormal(batch.messages);
        break;
      default:
        console.warn(`Unknown queue: ${batch.queue}`);
    }
  },
};

---

messages (Message[])

Array of Message objects.

Important:

• Ordering is best effort, not guaranteed
• Don't rely on message order
• Use timestamps for ordering if needed

// Sort by timestamp if order matters
const sortedMessages = batch.messages.sort(
  (a, b) => a.timestamp.getTime() - b.timestamp.getTime()
);

---

Methods

ackAll() - Acknowledge All Messages

Mark all messages as successfully delivered, even if handler throws error.

export default {
  async queue(batch: MessageBatch): Promise<void> {
    // Acknowledge all messages upfront
    batch.ackAll();

    // Even if this fails, messages won't retry
    await processMessages(batch.messages);
  },
};

Use cases:

• Idempotent operations where retries are safe
• Already processed messages (deduplication)
• Want to prevent retries regardless of outcome

---

retryAll(options?) - Retry All Messages

Mark all messages for retry.

interface QueueRetryOptions {
  delaySeconds?: number;  // 0-43200 (12 hours)
}

batch.retryAll();
batch.retryAll({ delaySeconds: 300 }); // Retry in 5 minutes

Use cases:

• Rate limiting (retry after backoff)
• Temporary system failure
• Upstream service unavailable

export default {
  async queue(batch: MessageBatch): Promise<void> {
    try {
      await callUpstreamAPI(batch.messages);
    } catch (error) {
      if (error.status === 503) {
        // Service unavailable - retry in 5 minutes
        batch.retryAll({ delaySeconds: 300 });
      } else {
        // Other error - retry immediately
        batch.retryAll();
      }
    }
  },
};

---

Message Interface

interface Message<Body = unknown> {
  readonly id: string;
  readonly timestamp: Date;
  readonly body: Body;
  readonly attempts: number;
  ack(): void;
  retry(options?: QueueRetryOptions): void;
}

Properties

id (string)

Unique system-generated message ID (UUID format).

console.log(message.id); // "550e8400-e29b-41d4-a716-446655440000"

---

timestamp (Date)

When message was sent to queue.

console.log(message.timestamp); // Date object
console.log(message.timestamp.toISOString()); // "2025-10-21T12:34:56.789Z"

// Check message age
const ageMs = Date.now() - message.timestamp.getTime();
console.log(`Message age: ${ageMs}ms`);

---

body (any)

Your message content.

interface MyMessage {
  type: string;
  userId: string;
  data: any;
}

const message: Message<MyMessage> = ...;
console.log(message.body.type);   // TypeScript knows the type
console.log(message.body.userId);
console.log(message.body.data);

---

attempts (number)

Number of times consumer has attempted to process this message. Starts at 1.

console.log(message.attempts); // 1 (first attempt)

// Use for exponential backoff
const delaySeconds = 60 * Math.pow(2, message.attempts - 1);
message.retry({ delaySeconds });

// Attempts: 1 → 60s, 2 → 120s, 3 → 240s, 4 → 480s, ...

---

Methods

ack() - Acknowledge Message

Mark message as successfully delivered. Won't retry even if handler fails.

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        // Non-idempotent operation
        await env.DB.prepare(
          'INSERT INTO orders (id, data) VALUES (?, ?)'
        ).bind(message.body.id, JSON.stringify(message.body)).run();

        // CRITICAL: Acknowledge success
        message.ack();
      } catch (error) {
        console.error(`Failed: ${message.id}`, error);
        // Don't ack - will retry
      }
    }
  },
};

Use cases:

• Database writes
• Payment processing
• Any non-idempotent operation
• Prevents duplicate processing

---

retry(options?) - Retry Message

Mark message for retry. Can optionally delay retry.

interface QueueRetryOptions {
  delaySeconds?: number;  // 0-43200 (12 hours)
}

message.retry();
message.retry({ delaySeconds: 600 }); // Retry in 10 minutes

Use cases:

• Rate limiting (429 errors)
• Temporary failures
• Exponential backoff

// Exponential backoff
message.retry({
  delaySeconds: Math.min(
    60 * Math.pow(2, message.attempts - 1),
    3600 // Max 1 hour
  ),
});

// Different delays for different errors
try {
  await processMessage(message.body);
  message.ack();
} catch (error) {
  if (error.status === 429) {
    // Rate limited - retry in 5 minutes
    message.retry({ delaySeconds: 300 });
  } else if (error.status >= 500) {
    // Server error - retry in 1 minute
    message.retry({ delaySeconds: 60 });
  } else {
    // Client error - don't retry
    console.error('Client error, not retrying');
  }
}

---

Acknowledgement Precedence Rules

When mixing ack/retry calls:

1. `ack()` or `retry()` wins - First call on a message takes precedence 2. Individual > Batch - Message-level call overrides batch-level call 3. Subsequent calls ignored - Second call on same message is silently ignored

// ack() takes precedence
message.ack();
message.retry(); // Ignored

// retry() takes precedence
message.retry();
message.ack(); // Ignored

// Individual overrides batch
message.ack();
batch.retryAll(); // Doesn't affect this message

// Batch doesn't affect individually handled messages
for (const msg of batch.messages) {
  msg.ack(); // These messages won't be affected by retryAll()
}
batch.retryAll(); // Only affects messages not explicitly ack'd

---

Processing Patterns

Sequential Processing

export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      await processMessage(message.body);
      message.ack();
    }
  },
};

Pros: Simple, ordered processing Cons: Slow for large batches

---

Parallel Processing

export default {
  async queue(batch: MessageBatch): Promise<void> {
    await Promise.all(
      batch.messages.map(async (message) => {
        try {
          await processMessage(message.body);
          message.ack();
        } catch (error) {
          console.error(`Failed: ${message.id}`, error);
        }
      })
    );
  },
};

Pros: Fast, efficient Cons: No ordering, higher memory usage

---

Batched Database Writes

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // Prepare all statements
    const statements = batch.messages.map((message) =>
      env.DB.prepare(
        'INSERT INTO events (id, data) VALUES (?, ?)'
      ).bind(message.id, JSON.stringify(message.body))
    );

    // Execute in batch
    const results = await env.DB.batch(statements);

    // Acknowledge based on results
    for (let i = 0; i < results.length; i++) {
      if (results[i].success) {
        batch.messages[i].ack();
      } else {
        console.error(`Failed: ${batch.messages[i].id}`);
      }
    }
  },
};

Pros: Efficient database usage Cons: More complex error handling

---

Message Type Routing

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        switch (message.body.type) {
          case 'email':
            await sendEmail(message.body, env);
            break;
          case 'sms':
            await sendSMS(message.body, env);
            break;
          case 'push':
            await sendPush(message.body, env);
            break;
          default:
            console.warn(`Unknown type: ${message.body.type}`);
        }

        message.ack();
      } catch (error) {
        console.error(`Failed: ${message.id}`, error);
        message.retry({ delaySeconds: 300 });
      }
    }
  },
};

---

ExecutionContext Methods

waitUntil(promise)

Extend Worker lifetime beyond handler return.

export default {
  async queue(batch: MessageBatch, env: Env, ctx: ExecutionContext): Promise<void> {
    for (const message of batch.messages) {
      await processMessage(message.body);
      message.ack();

      // Log asynchronously (doesn't block)
      ctx.waitUntil(
        env.LOGS.put(`log:${message.id}`, JSON.stringify({
          processedAt: Date.now(),
          message: message.body,
        }))
      );
    }
  },
};

---

passThroughOnException()

Continue processing even if handler throws.

export default {
  async queue(batch: MessageBatch, env: Env, ctx: ExecutionContext): Promise<void> {
    ctx.passThroughOnException();

    // If this throws, Worker doesn't fail
    // But unacknowledged messages will retry
    await processMessages(batch.messages);
  },
};

---

Last Updated: 2025-10-21

Producer API Reference

Complete reference for sending messages to Cloudflare Queues from Workers.

---

Queue Binding

Access queues via environment bindings configured in wrangler.jsonc:

{
  "queues": {
    "producers": [
      {
        "binding": "MY_QUEUE",
        "queue": "my-queue"
      }
    ]
  }
}

TypeScript:

type Bindings = {
  MY_QUEUE: Queue;
};

const app = new Hono<{ Bindings: Bindings }>();

app.post('/send', async (c) => {
  await c.env.MY_QUEUE.send({ data: 'hello' });
  return c.json({ sent: true });
});

---

send() - Send Single Message

Signature

interface Queue<Body = any> {
  send(body: Body, options?: QueueSendOptions): Promise<void>;
}

interface QueueSendOptions {
  delaySeconds?: number;  // 0-43200 (12 hours)
}

Parameters

• `body` - Any JSON serializable value
• Must be compatible with structured clone algorithm
• Max size: 128 KB (including ~100 bytes metadata)
• Types: primitives, objects, arrays, Date, Map, Set, etc.
• NOT supported: Functions, Symbols, DOM nodes

• `options.delaySeconds` (optional)
• Delay message delivery
• Range: 0-43200 seconds (0-12 hours)
• Default: 0 (immediate delivery)

Examples

// Send simple message
await env.MY_QUEUE.send({ userId: '123', action: 'welcome' });

// Send with delay (10 minutes)
await env.MY_QUEUE.send(
  { userId: '123', action: 'reminder' },
  { delaySeconds: 600 }
);

// Send complex object
await env.MY_QUEUE.send({
  type: 'order',
  order: {
    id: 'ORD-123',
    items: [
      { sku: 'ITEM-1', quantity: 2, price: 19.99 },
      { sku: 'ITEM-2', quantity: 1, price: 29.99 },
    ],
    total: 69.97,
    customer: {
      id: 'CUST-456',
      email: 'user@example.com',
    },
    metadata: {
      source: 'web',
      campaign: 'summer-sale',
    },
  },
  timestamp: Date.now(),
});

// Send with Date objects
await env.MY_QUEUE.send({
  scheduledFor: new Date('2025-12-25T00:00:00Z'),
  createdAt: new Date(),
});

---

sendBatch() - Send Multiple Messages

Signature

interface Queue<Body = any> {
  sendBatch(
    messages: Iterable<MessageSendRequest<Body>>,
    options?: QueueSendBatchOptions
  ): Promise<void>;
}

interface MessageSendRequest<Body = any> {
  body: Body;
  delaySeconds?: number;
}

interface QueueSendBatchOptions {
  delaySeconds?: number;  // Default delay for all messages
}

Parameters

• `messages` - Iterable of message objects
• Each message has body and optional delaySeconds
• Max 100 messages per batch
• Max 256 KB total batch size
• Can be Array, Set, Generator, etc.

• `options.delaySeconds` (optional)
• Default delay applied to all messages
• Overridden by individual message delaySeconds

Examples

// Send batch of messages
await env.MY_QUEUE.sendBatch([
  { body: { userId: '1', action: 'email' } },
  { body: { userId: '2', action: 'email' } },
  { body: { userId: '3', action: 'email' } },
]);

// Send batch with individual delays
await env.MY_QUEUE.sendBatch([
  { body: { task: 'immediate' }, delaySeconds: 0 },
  { body: { task: '5-min' }, delaySeconds: 300 },
  { body: { task: '1-hour' }, delaySeconds: 3600 },
]);

// Send batch with default delay (overridable per message)
await env.MY_QUEUE.sendBatch(
  [
    { body: { task: 'default-delay' } },
    { body: { task: 'custom-delay' }, delaySeconds: 600 },
  ],
  { delaySeconds: 300 } // Default 5 minutes
);

// Dynamic batch from database
const users = await getActiveUsers();
await env.MY_QUEUE.sendBatch(
  users.map(user => ({
    body: {
      type: 'send-notification',
      userId: user.id,
      email: user.email,
      message: 'You have a new message',
    },
  }))
);

// Generator pattern
async function* generateMessages() {
  for (let i = 0; i < 100; i++) {
    yield {
      body: { taskId: i, priority: i % 3 },
    };
  }
}

await env.MY_QUEUE.sendBatch(generateMessages());

---

Message Size Validation

Messages must be ≤128 KB. Check size before sending:

async function sendWithValidation(queue: Queue, message: any) {
  const messageStr = JSON.stringify(message);
  const size = new TextEncoder().encode(messageStr).length;
  const MAX_SIZE = 128 * 1024; // 128 KB

  if (size > MAX_SIZE) {
    throw new Error(
      `Message too large: ${size} bytes (max ${MAX_SIZE})`
    );
  }

  await queue.send(message);
}

Handling large messages:

// Store large data in R2, send reference
if (size > 128 * 1024) {
  const key = `messages/${crypto.randomUUID()}.json`;
  await env.MY_BUCKET.put(key, JSON.stringify(largeMessage));

  await env.MY_QUEUE.send({
    type: 'large-message',
    r2Key: key,
    metadata: {
      size,
      createdAt: Date.now(),
    },
  });
}

---

Throughput Management

Max throughput: 5,000 messages/second per queue.

Rate limiting:

// Batch sends for better throughput
const messages = Array.from({ length: 1000 }, (_, i) => ({
  body: { id: i },
}));

// Send in batches of 100 (10 sendBatch calls vs 1000 send calls)
for (let i = 0; i < messages.length; i += 100) {
  const batch = messages.slice(i, i + 100);
  await env.MY_QUEUE.sendBatch(batch);
}

// Add delay if needed
for (let i = 0; i < messages.length; i += 100) {
  const batch = messages.slice(i, i + 100);
  await env.MY_QUEUE.sendBatch(batch);

  if (i + 100 < messages.length) {
    await new Promise(resolve => setTimeout(resolve, 100)); // 100ms
  }
}

---

Error Handling

try {
  await env.MY_QUEUE.send(message);
} catch (error) {
  if (error.message.includes('Too Many Requests')) {
    // Throughput exceeded (>5000 msg/s)
    console.error('Rate limited');
  } else if (error.message.includes('too large')) {
    // Message >128 KB
    console.error('Message too large');
  } else {
    // Other error
    console.error('Queue send failed:', error);
  }
}

---

Production Patterns

Idempotency Keys

await env.MY_QUEUE.send({
  idempotencyKey: crypto.randomUUID(),
  orderId: 'ORD-123',
  action: 'process',
});

Message Versioning

await env.MY_QUEUE.send({
  _version: 1,
  _schema: 'order-v1',
  orderId: 'ORD-123',
  // ...
});

Correlation IDs

await env.MY_QUEUE.send({
  correlationId: requestId,
  parentSpanId: traceId,
  // ...
});

Priority Queues

// Use multiple queues for different priorities
if (priority === 'high') {
  await env.HIGH_PRIORITY_QUEUE.send(message);
} else {
  await env.LOW_PRIORITY_QUEUE.send(message);
}

---

Last Updated: 2025-10-21

Wrangler Commands for Cloudflare Queues

Complete reference for managing Cloudflare Queues via the wrangler CLI.

---

Queue Management

Create Queue

npx wrangler queues create <NAME> [OPTIONS]

Options:

• --delivery-delay-secs <SECONDS> - Default delay for all messages (0-43200)
• --message-retention-period-secs <SECONDS> - How long messages persist (60-1209600, default: 345600 / 4 days)

Examples:

# Create basic queue
npx wrangler queues create my-queue

# Create with custom retention (7 days)
npx wrangler queues create my-queue --message-retention-period-secs 604800

# Create with default delivery delay (5 minutes)
npx wrangler queues create delayed-queue --delivery-delay-secs 300

---

List Queues

npx wrangler queues list

Output:

┌──────────────────┬─────────────┬──────────┐
│ Name             │ Consumers   │ Messages │
├──────────────────┼─────────────┼──────────┤
│ my-queue         │ 1           │ 0        │
│ high-priority    │ 2           │ 142      │
│ my-dlq           │ 1           │ 5        │
└──────────────────┴─────────────┴──────────┘

---

Get Queue Info

npx wrangler queues info <NAME>

Example:

npx wrangler queues info my-queue

# Output:
# Queue: my-queue
# Message Retention: 345600 seconds (4 days)
# Delivery Delay: 0 seconds
# Consumers: 1
#   - my-consumer (batch_size: 10, batch_timeout: 5s, max_retries: 3)
# Backlog: 0 messages

---

Update Queue

npx wrangler queues update <NAME> [OPTIONS]

Options:

• --delivery-delay-secs <SECONDS> - Update default delay
• --message-retention-period-secs <SECONDS> - Update retention period

Examples:

# Update retention to 14 days (max)
npx wrangler queues update my-queue --message-retention-period-secs 1209600

# Update delivery delay to 10 minutes
npx wrangler queues update my-queue --delivery-delay-secs 600

---

Delete Queue

npx wrangler queues delete <NAME>

⚠️ WARNING:

• Deletes ALL messages in the queue
• Cannot be undone
• Use with extreme caution in production

Example:

npx wrangler queues delete old-queue

---

Consumer Management

Add Consumer

npx wrangler queues consumer add <QUEUE-NAME> <WORKER-SCRIPT-NAME> [OPTIONS]

Options:

• --batch-size <NUMBER> - Max messages per batch (1-100, default: 10)
• --batch-timeout <SECONDS> - Max wait time (0-60, default: 5)
• --message-retries <NUMBER> - Max retry attempts (0-100, default: 3)
• --max-concurrency <NUMBER> - Limit concurrent consumers (default: auto-scale to 250)
• --retry-delay-secs <SECONDS> - Default retry delay
• --dead-letter-queue <QUEUE-NAME> - DLQ for failed messages

Examples:

# Basic consumer
npx wrangler queues consumer add my-queue my-consumer

# Optimized for high throughput
npx wrangler queues consumer add my-queue my-consumer \
  --batch-size 100 \
  --batch-timeout 1

# With DLQ and retry settings
npx wrangler queues consumer add my-queue my-consumer \
  --batch-size 50 \
  --message-retries 5 \
  --retry-delay-secs 300 \
  --dead-letter-queue my-dlq

# Limit concurrency for rate-limited APIs
npx wrangler queues consumer add api-queue api-consumer \
  --max-concurrency 10

---

Remove Consumer

npx wrangler queues consumer remove <QUEUE-NAME> <WORKER-SCRIPT-NAME>

Example:

npx wrangler queues consumer remove my-queue my-consumer

---

Queue Operations

Purge Queue

npx wrangler queues purge <QUEUE-NAME>

⚠️ WARNING:

• Permanently deletes ALL messages
• Cannot be undone
• Use for clearing test data or stuck queues

Example:

npx wrangler queues purge test-queue

---

Pause Delivery

npx wrangler queues pause-delivery <QUEUE-NAME>

Use cases:

• Maintenance on consumer Workers
• Debugging consumer issues
• Temporarily stop processing without deleting messages

Example:

npx wrangler queues pause-delivery my-queue

---

Resume Delivery

npx wrangler queues resume-delivery <QUEUE-NAME>

Example:

npx wrangler queues resume-delivery my-queue

---

Event Subscriptions

Event subscriptions automatically send messages to a queue when events occur in other Cloudflare services.

Create Subscription

npx wrangler queues subscription create <QUEUE-NAME> [OPTIONS]

Options:

• --source <TYPE> - Event source (kv, r2, superSlurper, vectorize, workersAi.model, workersBuilds.worker, workflows.workflow)
• --events <EVENTS> - Comma-separated list of event types
• --name <NAME> - Subscription name (auto-generated if omitted)
• --enabled - Whether subscription is active (default: true)

Examples:

# Subscribe to R2 bucket events
npx wrangler queues subscription create my-queue \
  --source r2 \
  --events object-create,object-delete \
  --bucket-name my-bucket

# Subscribe to KV namespace events
npx wrangler queues subscription create my-queue \
  --source kv \
  --events key-write,key-delete \
  --namespace-id abc123

# Subscribe to Worker build events
npx wrangler queues subscription create build-queue \
  --source workersBuilds.worker \
  --events build-complete,build-failed \
  --worker-name my-worker

---

List Subscriptions

npx wrangler queues subscription list <QUEUE-NAME> [OPTIONS]

Options:

• --page <NUMBER> - Page number
• --per-page <NUMBER> - Results per page
• --json - Output as JSON

Example:

npx wrangler queues subscription list my-queue --json

---

Get Subscription

npx wrangler queues subscription get <QUEUE-NAME> --id <SUBSCRIPTION-ID> [--json]

Example:

npx wrangler queues subscription get my-queue --id sub_123 --json

---

Delete Subscription

npx wrangler queues subscription delete <QUEUE-NAME> --id <SUBSCRIPTION-ID>

Example:

npx wrangler queues subscription delete my-queue --id sub_123

---

Global Flags

These flags work on all commands:

• --help - Show help
• --config <PATH> - Path to wrangler.toml or wrangler.jsonc
• --cwd <PATH> - Run as if started in specified directory

---

Complete Workflow Example

# 1. Create queues
npx wrangler queues create my-queue
npx wrangler queues create my-dlq

# 2. Create and deploy producer Worker
cd my-producer
npm create cloudflare@latest -- --type hello-world --ts
# Add producer binding to wrangler.jsonc
npm run deploy

# 3. Create and deploy consumer Worker
cd ../my-consumer
npm create cloudflare@latest -- --type hello-world --ts
# Add consumer handler
npm run deploy

# 4. Add consumer to queue
npx wrangler queues consumer add my-queue my-consumer \
  --batch-size 50 \
  --message-retries 5 \
  --dead-letter-queue my-dlq

# 5. Monitor queue
npx wrangler queues info my-queue

# 6. Watch consumer logs
npx wrangler tail my-consumer

# 7. If needed, pause delivery
npx wrangler queues pause-delivery my-queue

# 8. Resume delivery
npx wrangler queues resume-delivery my-queue

---

Troubleshooting

Check queue backlog

npx wrangler queues info my-queue | grep "Backlog"

Clear stuck queue

npx wrangler queues purge my-queue

Verify consumer is attached

npx wrangler queues info my-queue | grep "Consumers"

Check for delivery paused

npx wrangler queues info my-queue
# Look for "Delivery: paused"

---

Last Updated: 2025-10-21 Wrangler Version: 4.43.0+