Cloudflare Queues

Cloudflare Queues

Status: Production Ready ✅ | Last Verified: 2025-12-27

Dependencies: cloudflare-worker-base (for Worker setup)

Contents: Quick Start • Critical Rules • Top Errors • Use Cases • When to Load References • Limits

---

Quick Start (10 Minutes)

1. Create Queue

bunx wrangler queues create my-queue
bunx wrangler queues list

2. Producer (Send Messages)

wrangler.jsonc:

{
  "name": "my-producer",
  "main": "src/index.ts",
  "queues": {
    "producers": [
      {
        "binding": "MY_QUEUE",
        "queue": "my-queue"
      }
    ]
  }
}

src/index.ts:

import { Hono } from 'hono';

type Bindings = {
  MY_QUEUE: Queue;
};

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

app.post('/send', async (c) => {
  await c.env.MY_QUEUE.send({
    userId: '123',
    action: 'process-order',
    timestamp: Date.now(),
  });

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

export default app;

3. Consumer (Process Messages)

wrangler.jsonc:

{
  "name": "my-consumer",
  "main": "src/consumer.ts",
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10,
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"
      }
    ]
  }
}

src/consumer.ts:

import type { MessageBatch } from '@cloudflare/workers-types';

export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      console.log('Processing:', message.body);
      // Your logic here
    }
    // Implicit ack: returning successfully acknowledges all messages
  },
};

Deploy:

bunx wrangler deploy

Load: references/setup-guide.md for complete 6-step setup with DLQ configuration

---

Critical Rules

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

---

Top 3 Critical Errors

Error #1: Message Too Large

Problem: Message exceeds 128 KB limit

Solution: Store large data in R2, send reference

// ❌ Wrong
await env.MY_QUEUE.send({ data: largeArray }); // >128 KB fails

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

if (size > 128000) {
  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);
}

Error #2: Throughput Exceeded

Problem: Exceeding 5000 messages/second per queue

Solution: Use sendBatch() and rate limiting

// ❌ Wrong
for (let i = 0; i < 10000; i++) {
  await env.MY_QUEUE.send({ id: i }); // Too fast!
}

// ✅ Correct
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));
}

Error #3: Entire Batch Retried When One Message Fails

Problem: Single message failure causes all messages to retry

Solution: Use explicit acknowledgement

// ❌ Wrong - implicit ack
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      await env.DB.prepare('INSERT INTO orders VALUES (?, ?)').bind(
        message.body.id,
        message.body.amount
      ).run();
    }
    // If any fails, ALL retry!
  },
};

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

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

Load `references/error-catalog.md` for all 10 errors including DLQ configuration, auto-scaling issues, message deletion prevention, and detailed solutions.

---

Common Use Cases

Use Case 1: Basic Message Queue

When: Simple async job processing (emails, notifications)

Quick Pattern:

// Producer
await env.MY_QUEUE.send({ type: 'email', to: 'user@example.com' });

// Consumer (implicit ack - for idempotent operations)
export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      await sendEmail(message.body.to, message.body.content);
    }
  },
};

Load: templates/queues-producer.ts + templates/queues-consumer-basic.ts

Use Case 2: Database Writes (Non-Idempotent)

When: Writing to database, must avoid duplicates

Load: templates/queues-consumer-explicit-ack.ts + references/consumer-api.md

Use Case 3: Retry with Exponential Backoff

When: Calling rate-limited APIs, temporary failures

Load: templates/queues-retry-with-delay.ts + references/error-catalog.md (Error #2, #3)

Use Case 4: Dead Letter Queue Pattern

When: Production systems, need to capture permanently failed messages

Load: templates/queues-dlq-pattern.ts + references/setup-guide.md (Step 4)

Use Case 5: High Throughput Processing

When: Processing thousands of messages per second

Quick Pattern:

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100,        // Large batches
      "max_batch_timeout": 5,       // Fast processing
      "max_concurrency": null       // Auto-scale
    }]
  }
}

Load: references/best-practices.md → Optimizing Throughput

---

When to Load References

Load `references/setup-guide.md` when:

• User needs complete setup walkthrough (queue → producer → consumer → DLQ)
• First time setting up Cloudflare Queues
• Need production configuration examples
• Want complete end-to-end example

Load `references/error-catalog.md` when:

• Encountering any of the 10 documented errors
• Troubleshooting queue issues
• Messages not being delivered/processed
• Need prevention checklist

Load `references/producer-api.md` when:

• Need complete producer API reference
• Using send() or sendBatch() methods
• Need message format specifications
• Handling large messages or batches

Load `references/consumer-api.md` when:

• Need complete consumer API reference
• Using explicit ack(), retry(), or retryAll()
• Need batch processing patterns
• Implementing complex consumer logic

Load `references/best-practices.md` when:

• Optimizing queue performance
• Production deployment guidance
• Monitoring and observability setup
• Security and reliability patterns

Load `references/wrangler-commands.md` when:

• Need CLI commands reference
• Managing queues (create, delete, list)
• Controlling delivery (pause, resume)
• Debugging queue issues
• Real-time monitoring and performance analysis

Load `references/typescript-types.md` when:

• Need complete TypeScript type definitions
• Working with Queue, MessageBatch, or Message interfaces
• Implementing type-safe producers or consumers
• Using generic types for message bodies
• Need type guards for message validation

Load `references/production-checklist.md` when:

• Preparing for production deployment
• Need pre-deployment verification checklist
• Want detailed explanations of production best practices
• Setting up monitoring, DLQ, or error handling
• Planning load testing or security review

Load `references/pull-consumers.md` when:

• Need to consume messages from non-Workers environments
• Integrating with existing backend services (Node.js, Python, Go)
• Implementing pull-based polling instead of push-based consumption
• Working with containerized or legacy applications

Load `references/http-publishing.md` when:

• Publishing messages from external systems via HTTP
• Integrating webhooks from third-party services
• Need to send messages without deploying Workers
• Implementing CI/CD pipeline notifications

Load `references/r2-event-integration.md` when:

• Triggering queue messages on R2 bucket events
• Implementing automated image/document processing
• Setting up event-driven data pipelines
• Need R2 object upload/delete notifications

---

Agents & Commands

Available Agents:

• queue-debugger - 9-phase diagnostic analysis for queue issues (systematic troubleshooting)
• queue-optimizer - Performance tuning and cost optimization (batch size, concurrency, retries)

Available Commands:

• /queue-setup - Interactive wizard for complete queue setup
• /queue-troubleshoot - Quick diagnostic for common issues
• /queue-monitor - Real-time metrics and status display

---

Limits & Quotas

Critical limits:

• Message size: 128 KB max per message
• Throughput: 5,000 messages/second per queue
• Batch size: 100 messages max per sendBatch()
• Consumer CPU time: 30 seconds (default), 300 seconds (max with config)
• Max retries: Configurable (default 3)
• Queue name: Max 63 characters, lowercase/numbers/hyphens

Load `references/best-practices.md` for handling limits and optimization strategies.

---

Configuration Reference

Producer: Add queue binding to wrangler.jsonc queues.producers array with binding and queue fields.

Consumer: Configure in wrangler.jsonc queues.consumers array with queue, max_batch_size (1-100), max_batch_timeout (0-60s), max_retries, dead_letter_queue, and optionally max_concurrency (default: auto-scale).

CPU Limits: Increase limits.cpu_ms from default 30,000ms if processing takes longer.

Load `references/setup-guide.md` for complete configuration examples and `templates/wrangler-queues-config.jsonc` for production-ready config.

---

Using Bundled Resources

References (references/)

• setup-guide.md - Complete 6-step setup (queue → producer → consumer → DLQ → deploy → production config)
• error-catalog.md - All 10 errors with solutions + prevention checklist
• producer-api.md - Complete producer API (send, sendBatch, message format)
• consumer-api.md - Complete consumer API (ack, retry, batch processing)
• best-practices.md - Performance, monitoring, security, reliability patterns
• wrangler-commands.md - CLI reference (create, delete, list, pause, resume)
• typescript-types.md - Complete TypeScript type definitions for Queue, MessageBatch, Message
• production-checklist.md - Pre-deployment verification and best practices
• pull-consumers.md - Pull-based consumers for non-Workers environments (HTTP polling)
• http-publishing.md - Publishing messages via HTTP API from external systems
• r2-event-integration.md - R2 event notifications triggering queue messages

Templates (templates/)

• queues-producer.ts - Basic producer with single and batch sending
• queues-consumer-basic.ts - Implicit ack consumer (idempotent operations)
• queues-consumer-explicit-ack.ts - Explicit ack consumer (non-idempotent)
• queues-retry-with-delay.ts - Exponential backoff retry pattern
• queues-dlq-pattern.ts - Dead letter queue setup and consumer
• wrangler-queues-config.jsonc - Complete production configuration

---

TypeScript Types

Use @cloudflare/workers-types package for complete type definitions: Queue, MessageBatch<Body>, Message<Body>, QueueSendOptions.

Load `references/typescript-types.md` for complete type reference with interfaces, generics, type guards, and usage examples.

---

Monitoring & Debugging

Key Commands: wrangler queues info (status), wrangler tail (logs), wrangler queues pause-delivery/resume-delivery (control).

Load `references/wrangler-commands.md` for complete CLI reference with real-time monitoring, debugging workflows, and performance analysis commands.

---

Production Checklist

12-Point Pre-Deployment Checklist: DLQ configuration, message acknowledgment strategy, size validation, batch optimization, concurrency settings, CPU limits, error handling, monitoring, rate limiting, idempotency, load testing, and security review.

Load `references/production-checklist.md` for complete checklist with detailed explanations, code examples, and deployment workflow.

---

Related Skills

• cloudflare-worker-base - Worker setup and configuration
• cloudflare-d1 - Database integration for queue consumers
• cloudflare-r2 - Store large message payloads
• cloudflare-workflows - More complex orchestration needs

---

Official Documentation

• Cloudflare Queues: https://developers.cloudflare.com/queues/
• Configuration: https://developers.cloudflare.com/queues/configuration/
• Wrangler Commands: https://developers.cloudflare.com/workers/wrangler/commands/#queues
• Limits: https://developers.cloudflare.com/queues/platform/limits/
• Troubleshooting: https://developers.cloudflare.com/queues/observability/troubleshooting/
• Best Practices: https://developers.cloudflare.com/queues/configuration/best-practices/

---

Questions? Issues?

1. Check references/error-catalog.md for all 10 errors and solutions 2. Review references/setup-guide.md for complete setup walkthrough 3. See references/best-practices.md for production patterns 4. Check official docs: https://developers.cloudflare.com/queues/

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

Cloudflare Queues Error Catalog

Complete catalog of 10 documented errors with solutions and troubleshooting.

---

Error #1: Message Too Large

Error: Message exceeds 128 KB limit

Source: Cloudflare Queues Limits documentation

Why It Happens: Message body exceeds the 128 KB per-message limit

Solution: Store large data in R2, send reference

// ❌ 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);
}

In consumer - retrieve from R2:

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

        // Process large data
        await process(largeData);

        // Cleanup
        await env.MY_BUCKET.delete(message.body.r2Key);
        message.ack();
      } else {
        await process(message.body);
        message.ack();
      }
    }
  },
};

---

Error #2: Throughput Exceeded

Error: Exceeding 5000 messages/second per queue

Source: Cloudflare Queues Limits

Why It Happens: Sending too many messages too quickly

Solution: Use sendBatch() and rate limiting

// ❌ 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
  }
}

---

Error #3: Consumer Timeout (CPU Limit Exceeded)

Error: Consumer exceeds CPU time limit (default 30 seconds)

Source: Cloudflare Workers CPU limits

Why It Happens: Message processing takes longer than CPU limit

Solution: Increase CPU limit in wrangler.jsonc

// ❌ 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!
    }
  },
};

wrangler.jsonc:

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

Alternative: Break into smaller chunks

// ✅ Process in smaller batches
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // Process in parallel chunks
    const chunkSize = 5;
    for (let i = 0; i < batch.messages.length; i += chunkSize) {
      const chunk = batch.messages.slice(i, i + chunkSize);

      await Promise.all(
        chunk.map(async (message) => {
          await process(message.body);
          message.ack();
        })
      );
    }
  },
};

---

Error #4: Queue Backlog Growing

Error: Messages accumulating faster than consumer can process

Source: Queue metrics showing growing backlog

Why It Happens: Consumer too slow, not scaling, or errors blocking processing

Solution: Multiple approaches

Solution 1: Increase batch size

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100  // Process more per invocation (default 10)
    }]
  }
}

Solution 2: Let concurrency auto-scale

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      // Don't set max_concurrency - let it auto-scale
      "max_batch_size": 100
    }]
  }
}

Solution 3: Optimize consumer code (parallel processing)

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // ❌ Bad: Sequential processing
    // for (const message of batch.messages) {
    //   await process(message.body);
    // }

    // ✅ Good: Parallel processing
    await Promise.all(
      batch.messages.map(async (message) => {
        await process(message.body);
        message.ack();
      })
    );
  },
};

---

Error #5: Messages Not Being Delivered to Consumer

Error: Messages sent but never reach consumer

Source: Queue monitoring

Why It Happens: Consumer not deployed, wrong queue name, delivery paused, or consumer errors

Solution: Systematic debugging

# 1. Check queue info (verify messages exist)
npx wrangler queues info my-queue

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

# 3. Verify consumer is deployed
npx wrangler deployments list

# 4. Check consumer logs for errors
npx wrangler tail my-consumer

# 5. Verify queue name in wrangler.jsonc matches

Common misconfigurations:

// ❌ Wrong: Queue name mismatch
{
  "queues": {
    "consumers": [{
      "queue": "my-qeue"  // Typo!
    }]
  }
}

// ✅ Correct: Exact queue name
{
  "queues": {
    "consumers": [{
      "queue": "my-queue"  // Must match exactly
    }]
  }
}

---

Error #6: Entire Batch Retried When One Message Fails

Error: Single message failure causes all messages in batch to retry

Source: Implicit acknowledgement behavior

Why It Happens: Using implicit ack with non-idempotent operations

Solution: Use explicit acknowledgement

// ❌ Bad: Implicit ack with non-idempotent operations
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // DB write - non-idempotent!
      await env.DB.prepare(
        'INSERT INTO orders (id, amount) VALUES (?, ?)'
      ).bind(message.body.id, message.body.amount).run();
    }
    // If any message fails, ALL retry!
  },
};

// ✅ Good: Explicit ack per message
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, amount) VALUES (?, ?)'
        ).bind(message.body.id, message.body.amount).run();

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

---

Error #7: Messages Deleted Without Processing

Error: Messages disappear after max retries without going to DLQ

Source: Missing DLQ configuration

Why It Happens: No Dead Letter Queue configured, messages deleted after max_retries

Solution: Configure DLQ

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

# 2. Add to consumer config

wrangler.jsonc:

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_retries": 3,
      "dead_letter_queue": "my-dlq"  // CRITICAL: Add this!
    }]
  }
}

Create DLQ consumer:

// src/dlq-consumer.ts
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // Log failed message
      console.error('PERMANENTLY FAILED:', {
        id: message.id,
        attempts: message.attempts,
        body: message.body,
      });

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

      message.ack();
    }
  },
};

---

Error #8: Consumer Not Auto-Scaling

Error: Consumer stays at low concurrency despite growing backlog

Source: Consumer configuration

Why It Happens: max_concurrency set too low, consumer errors, or no backlog

Solution: Remove max_concurrency limit

// ❌ Bad: Limits scaling
{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_concurrency": 1  // Won't scale!
    }]
  }
}

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

When to set max_concurrency:

• Rate-limited external APIs
• Database connection limits
• Resource contention issues

Otherwise: let it auto-scale (don't set it).

---

Error #9: Queue Name Already Exists

Error: Error: Queue name already in use

Source: wrangler queues create

Why It Happens: Attempting to create queue with existing name

Solution: Use different name or check existing queues

# Check existing queues
npx wrangler queues list

# Delete queue if no longer needed
npx wrangler queues delete old-queue

# Or use different name
npx wrangler queues create my-queue-v2

---

Error #10: Message Lost / Not Retried

Error: Message fails but doesn't retry

Source: Explicit ack() called even on failure

Why It Happens: Accidentally calling ack() in catch block

Solution: Only ack() on success

// ❌ Bad: Ack on failure (message lost!)
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        await process(message.body);
      } catch (error) {
        console.error('Failed:', error);
      }

      message.ack(); // ❌ Called even on error!
    }
  },
};

// ✅ Good: Only ack on success
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        await process(message.body);
        message.ack(); // ✅ Only called on success
      } catch (error) {
        console.error('Failed:', error);
        // Don't ack - will retry
      }
    }
  },
};

---

Troubleshooting Guide

Problem: High DLQ message count

Solution: 1. Check DLQ consumer logs for error patterns 2. Review business logic - are messages valid? 3. Check external dependencies (APIs, databases) 4. Consider increasing max_retries if transient failures

# Check DLQ size
npx wrangler queues info my-dlq

# Monitor DLQ consumer
npx wrangler tail my-dlq-consumer

Problem: Messages processed multiple times (duplicates)

Solution: 1. Make operations idempotent (use upsert, not insert) 2. Use explicit ack() for non-idempotent operations 3. Track processed message IDs in database

// Idempotent operation
await env.DB.prepare(
  'INSERT INTO orders (id, amount) VALUES (?, ?) ON CONFLICT (id) DO UPDATE SET amount = ?'
).bind(messageId, amount, amount).run();

Problem: Consumer running but not processing messages

Solution: 1. Check consumer logs for errors 2. Verify consumer function signature (must be queue() export) 3. Check bindings are correctly configured

// ❌ Wrong: Missing export
const queue = async (batch: MessageBatch) => { /* ... */ };

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

---

Prevention Checklist

Use this to avoid all 10 errors:

• [ ] Message size validated before sending (<128 KB)
• [ ] Using sendBatch() for multiple messages (not individual sends)
• [ ] CPU limit increased if processing >30 seconds
• [ ] max_batch_size optimized for workload
• [ ] max_concurrency NOT set (let it auto-scale) unless specific reason
• [ ] Dead Letter Queue created and configured
• [ ] DLQ consumer deployed and monitored
• [ ] Explicit ack() used for non-idempotent operations
• [ ] Only ack() called on success (not in error handler)
• [ ] Queue names match exactly between producer and consumer configs
• [ ] Consumer deployed before sending messages
• [ ] Operations made idempotent where possible
• [ ] External dependencies have retry logic with backoff

---

Official Resources:

• Cloudflare Queues Docs: https://developers.cloudflare.com/queues/
• Limits & Quotas: https://developers.cloudflare.com/queues/platform/limits/
• Troubleshooting: https://developers.cloudflare.com/queues/observability/troubleshooting/

HTTP Publishing to Cloudflare Queues

Official Feature: Direct HTTP endpoints allow external systems to publish messages to queues without Workers.

When to Use: Load this reference when the user asks to "publish from external service", "send messages via HTTP", "integrate third-party system with queue", or needs to publish from non-Workers environments.

---

Overview

HTTP Publishing enables any system with HTTP capabilities to send messages directly to Cloudflare Queues using REST API endpoints. This allows queue integration from:

• External APIs and webhooks
• CI/CD pipelines
• Monitoring/alerting systems
• Legacy applications
• Third-party services

Key Benefits:

• No Workers deployment required for publishing
• Standard HTTP POST requests
• Works from any environment
• Simple authentication via API tokens

---

Use Cases

1. Webhook Integration

Receive webhooks from external services and queue for processing:

# Stripe webhook → Queue
curl -X POST \
  https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/{queue_name}/messages \
  -H "Authorization: Bearer {api_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{
      "body": {
        "type": "payment.succeeded",
        "stripe_event_id": "evt_123",
        "amount": 4999,
        "customer_id": "cus_xyz"
      }
    }]
  }'

2. CI/CD Pipeline Events

Queue deployment notifications from GitHub Actions, GitLab CI:

# .github/workflows/deploy.yml
- name: Notify Queue on Deploy
  run: |
    curl -X POST \
      https://api.cloudflare.com/client/v4/accounts/${{ secrets.CF_ACCOUNT_ID }}/queues/deployment-events/messages \
      -H "Authorization: Bearer ${{ secrets.CF_API_TOKEN }}" \
      -H "Content-Type: application/json" \
      -d '{
        "messages": [{
          "body": {
            "event": "deployment.completed",
            "repo": "${{ github.repository }}",
            "commit": "${{ github.sha }}",
            "environment": "production"
          }
        }]
      }'

3. Monitoring Alerts

Send alerts from monitoring systems (Datadog, New Relic, Prometheus):

# Datadog webhook handler
import requests

def send_alert_to_queue(alert_data):
    response = requests.post(
        f'https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/monitoring-alerts/messages',
        headers={
            'Authorization': f'Bearer {api_token}',
            'Content-Type': 'application/json'
        },
        json={
            'messages': [{
                'body': {
                    'alert_id': alert_data['id'],
                    'severity': alert_data['severity'],
                    'service': alert_data['service'],
                    'message': alert_data['message'],
                    'timestamp': alert_data['timestamp']
                }
            }]
        }
    )
    return response.status_code == 200

4. Legacy System Integration

Queue messages from systems that can't run Workers:

// Java application
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

public void publishToQueue(OrderEvent event) {
    String endpoint = String.format(
        "https://api.cloudflare.com/client/v4/accounts/%s/queues/%s/messages",
        accountId, queueName
    );

    HttpPost request = new HttpPost(endpoint);
    request.setHeader("Authorization", "Bearer " + apiToken);
    request.setHeader("Content-Type", "application/json");

    String json = String.format(
        "{\"messages\": [{\"body\": {\"orderId\": \"%s\", \"userId\": \"%s\"}}]}",
        event.getOrderId(), event.getUserId()
    );

    request.setEntity(new StringEntity(json));

    try (CloseableHttpClient client = HttpClients.createDefault()) {
        client.execute(request);
    }
}

---

HTTP Publish API Reference

Endpoint

POST https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/{queue_name}/messages

Authentication

Requires Cloudflare API Token with Queue: Edit permission.

Request Headers

Authorization: Bearer {api_token}
Content-Type: application/json

Request Body Format

Single Message:

{
  "messages": [
    {
      "body": {
        "type": "order-created",
        "orderId": "12345",
        "userId": "user_789",
        "amount": 99.99
      }
    }
  ]
}

Batch Messages (up to 100 per request):

{
  "messages": [
    {
      "body": {
        "type": "order-created",
        "orderId": "12345"
      }
    },
    {
      "body": {
        "type": "order-created",
        "orderId": "12346"
      }
    },
    {
      "body": {
        "type": "order-created",
        "orderId": "12347"
      }
    }
  ]
}

Message with Delay:

{
  "messages": [
    {
      "body": {
        "type": "send-reminder",
        "userId": "user_123"
      },
      "delay_seconds": 3600
    }
  ]
}

Parameters:

• body (required): Message payload as JSON object (max 128 KB)
• delay_seconds (optional): Delay delivery by 0-43,200 seconds (12 hours)

Response Format

Success (200 OK):

{
  "success": true,
  "result": {
    "success_count": 3,
    "messages": [
      {
        "id": "msg_abc123",
        "timestamp": "2025-12-27T10:30:00Z"
      },
      {
        "id": "msg_def456",
        "timestamp": "2025-12-27T10:30:00Z"
      },
      {
        "id": "msg_ghi789",
        "timestamp": "2025-12-27T10:30:00Z"
      }
    ]
  }
}

Error (400 Bad Request):

{
  "success": false,
  "errors": [
    {
      "code": 10004,
      "message": "Message body exceeds 128 KB limit"
    }
  ]
}

---

Code Examples by Language

cURL

# Single message
curl -X POST \
  "https://api.cloudflare.com/client/v4/accounts/abc123/queues/my-queue/messages" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{
      "body": {
        "event": "user.signup",
        "userId": "user_123",
        "timestamp": 1703692800
      }
    }]
  }'

# Batch publish
curl -X POST \
  "https://api.cloudflare.com/client/v4/accounts/abc123/queues/my-queue/messages" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"body": {"orderId": "1"}},
      {"body": {"orderId": "2"}},
      {"body": {"orderId": "3"}}
    ]
  }'

# Delayed message
curl -X POST \
  "https://api.cloudflare.com/client/v4/accounts/abc123/queues/my-queue/messages" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{
      "body": {"task": "send-reminder"},
      "delay_seconds": 3600
    }]
  }'

Python

import requests
import json

class CloudflareQueuePublisher:
    def __init__(self, account_id: str, api_token: str):
        self.account_id = account_id
        self.api_token = api_token
        self.base_url = f'https://api.cloudflare.com/client/v4/accounts/{account_id}/queues'

    def publish(self, queue_name: str, messages: list, delay_seconds: int = 0):
        """
        Publish messages to queue

        Args:
            queue_name: Name of the queue
            messages: List of message bodies (dicts)
            delay_seconds: Optional delay (0-43200 seconds)

        Returns:
            Response dict with success status and message IDs
        """
        url = f'{self.base_url}/{queue_name}/messages'

        payload = {
            'messages': [
                {
                    'body': msg,
                    **({"delay_seconds": delay_seconds} if delay_seconds > 0 else {})
                }
                for msg in messages
            ]
        }

        response = requests.post(
            url,
            headers={
                'Authorization': f'Bearer {self.api_token}',
                'Content-Type': 'application/json'
            },
            json=payload
        )

        response.raise_for_status()
        return response.json()

# Usage
publisher = CloudflareQueuePublisher(
    account_id='abc123',
    api_token='your_api_token'
)

# Single message
result = publisher.publish('my-queue', [
    {'type': 'order-created', 'orderId': '12345'}
])

# Batch publish
result = publisher.publish('my-queue', [
    {'type': 'email', 'to': 'user1@example.com'},
    {'type': 'email', 'to': 'user2@example.com'},
    {'type': 'email', 'to': 'user3@example.com'}
])

# Delayed message
result = publisher.publish(
    'my-queue',
    [{'task': 'send-reminder', 'userId': 'user_123'}],
    delay_seconds=3600
)

print(f"Published {result['result']['success_count']} messages")

Node.js

// Node.js with fetch
class CloudflareQueuePublisher {
    constructor(accountId, apiToken) {
        this.accountId = accountId;
        this.apiToken = apiToken;
        this.baseUrl = `https://api.cloudflare.com/client/v4/accounts/${accountId}/queues`;
    }

    async publish(queueName, messages, delaySeconds = 0) {
        const url = `${this.baseUrl}/${queueName}/messages`;

        const payload = {
            messages: messages.map(body => ({
                body,
                ...(delaySeconds > 0 && { delay_seconds: delaySeconds })
            }))
        };

        const response = await fetch(url, {
            method: 'POST',
            headers: {
                'Authorization': `Bearer ${this.apiToken}`,
                'Content-Type': 'application/json'
            },
            body: JSON.stringify(payload)
        });

        if (!response.ok) {
            throw new Error(`HTTP ${response.status}: ${await response.text()}`);
        }

        return response.json();
    }
}

// Usage
const publisher = new CloudflareQueuePublisher(
    process.env.CF_ACCOUNT_ID,
    process.env.CF_API_TOKEN
);

// Publish single message
await publisher.publish('my-queue', [
    { type: 'user.signup', userId: 'user_123' }
]);

// Batch publish
await publisher.publish('my-queue', [
    { orderId: '1' },
    { orderId: '2' },
    { orderId: '3' }
]);

// Delayed publish
await publisher.publish(
    'notifications',
    [{ type: 'reminder', userId: 'user_123' }],
    3600
);

Go

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
)

type QueuePublisher struct {
    AccountID string
    APIToken  string
    Client    *http.Client
}

type Message struct {
    Body         interface{} `json:"body"`
    DelaySeconds int         `json:"delay_seconds,omitempty"`
}

type PublishRequest struct {
    Messages []Message `json:"messages"`
}

func (p *QueuePublisher) Publish(queueName string, bodies []interface{}, delaySeconds int) error {
    url := fmt.Sprintf(
        "https://api.cloudflare.com/client/v4/accounts/%s/queues/%s/messages",
        p.AccountID, queueName,
    )

    messages := make([]Message, len(bodies))
    for i, body := range bodies {
        messages[i] = Message{
            Body:         body,
            DelaySeconds: delaySeconds,
        }
    }

    payload := PublishRequest{Messages: messages}
    jsonData, err := json.Marshal(payload)
    if err != nil {
        return err
    }

    req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
    if err != nil {
        return err
    }

    req.Header.Set("Authorization", "Bearer "+p.APIToken)
    req.Header.Set("Content-Type", "application/json")

    resp, err := p.Client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("HTTP %d: %s", resp.StatusCode, resp.Status)
    }

    return nil
}

// Usage
func main() {
    publisher := &QueuePublisher{
        AccountID: "abc123",
        APIToken:  "your_api_token",
        Client:    &http.Client{},
    }

    // Single message
    err := publisher.Publish("my-queue", []interface{}{
        map[string]interface{}{
            "type":    "order-created",
            "orderId": "12345",
        },
    }, 0)

    // Batch publish
    err = publisher.Publish("my-queue", []interface{}{
        map[string]interface{}{"orderId": "1"},
        map[string]interface{}{"orderId": "2"},
        map[string]interface{}{"orderId": "3"},
    }, 0)
}

---

Batch Publishing Best Practices

1. Batch Size Optimization

Minimize API calls by batching messages:

# ❌ Bad: 100 API calls for 100 messages
for message in messages:
    publish_single(queue, message)

# ✅ Good: 1 API call for 100 messages (max batch size)
publish_batch(queue, messages[:100])

2. Handle Large Message Sets

Split large sets into batches of 100:

def publish_large_batch(queue_name: str, messages: list):
    """Publish unlimited messages in batches of 100"""
    BATCH_SIZE = 100
    results = []

    for i in range(0, len(messages), BATCH_SIZE):
        batch = messages[i:i + BATCH_SIZE]
        result = publisher.publish(queue_name, batch)
        results.append(result)

    return results

# Publish 1000 messages (10 API calls)
publish_large_batch('my-queue', all_messages)

3. Retry Failed Batches

Implement exponential backoff for transient failures:

import time

def publish_with_retry(queue_name: str, messages: list, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return publisher.publish(queue_name, messages)
        except requests.HTTPError as e:
            if e.response.status_code == 429:  # Rate limit
                wait = (2 ** attempt) * 1000  # Exponential backoff
                print(f"Rate limited, waiting {wait}ms...")
                time.sleep(wait / 1000)
            elif e.response.status_code >= 500:  # Server error
                time.sleep(2 ** attempt)
            else:
                raise  # Don't retry client errors (400-499)

    raise Exception(f"Failed after {max_retries} retries")

---

Message Size Limits

Maximum message size: 128 KB per message

Validate Message Size

import json

def validate_message_size(message: dict) -> bool:
    """Check if message is under 128 KB"""
    message_bytes = json.dumps(message).encode('utf-8')
    size_kb = len(message_bytes) / 1024

    if size_kb > 128:
        print(f"Message too large: {size_kb:.2f} KB (max 128 KB)")
        return False

    return True

# Check before publishing
if validate_message_size(large_message):
    publisher.publish('my-queue', [large_message])
else:
    # Store large payload in R2, send reference
    url = upload_to_r2(large_message)
    publisher.publish('my-queue', [{'type': 'large-data', 'url': url}])

Handle Large Payloads

For messages >128 KB, store in R2 and send reference:

def publish_large_payload(queue_name: str, payload: dict):
    """Publish large payload via R2 storage"""
    size = len(json.dumps(payload).encode('utf-8')) / 1024

    if size <= 128:
        # Small enough - publish directly
        publisher.publish(queue_name, [payload])
    else:
        # Too large - store in R2
        object_key = f'payloads/{uuid.uuid4()}.json'
        r2_url = upload_to_r2(object_key, payload)

        # Publish reference
        publisher.publish(queue_name, [{
            'type': 'large-payload',
            'r2_key': object_key,
            'r2_url': r2_url,
            'size_kb': size
        }])

# Consumer retrieves from R2
async def process_message(message):
    if message.body.type == 'large-payload':
        payload = await fetch_from_r2(message.body.r2_key)
        await process_payload(payload)
    else:
        await process_payload(message.body)

---

Error Handling

Common Error Codes

Error Response Example

{
  "success": false,
  "errors": [
    {
      "code": 10004,
      "message": "Message body exceeds 128 KB limit",
      "details": {
        "message_index": 0,
        "size_kb": 156.8
      }
    }
  ]
}

Comprehensive Error Handler

def publish_with_error_handling(queue_name: str, messages: list):
    try:
        result = publisher.publish(queue_name, messages)
        print(f"Published {result['result']['success_count']} messages")
        return result
    except requests.HTTPError as e:
        if e.response.status_code == 400:
            # Client error - fix before retrying
            error_data = e.response.json()
            print(f"Bad request: {error_data['errors']}")
            # Handle specific errors
            for error in error_data['errors']:
                if error['code'] == 10004:
                    print("Message too large - consider R2 storage")
                elif error['code'] == 10006:
                    print("Batch too large - split into smaller batches")
        elif e.response.status_code == 429:
            # Rate limited - backoff and retry
            print("Rate limited - backing off...")
            time.sleep(60)
            return publish_with_error_handling(queue_name, messages)
        elif e.response.status_code >= 500:
            # Server error - retry
            print("Server error - retrying...")
            time.sleep(5)
            return publish_with_error_handling(queue_name, messages)
        raise

---

Security Best Practices

1. Protect API Tokens

Never expose tokens in client-side code or version control:

# ❌ Bad: Hardcoded token
api_token = "abc123_hardcoded_token"

# ✅ Good: Environment variable
api_token = os.environ['CLOUDFLARE_API_TOKEN']

# ✅ Good: Secrets manager
api_token = get_secret('cloudflare/api_token')

2. Least Privilege Tokens

Create dedicated tokens with minimal permissions:

• Permission: Queue: Edit only
• Scope: Specific queue (not account-wide)
• IP restrictions: Limit to known IPs if possible

3. Validate Input

Sanitize message payloads before publishing:

def sanitize_and_publish(queue_name: str, user_input: dict):
    """Validate user input before publishing"""
    # Whitelist allowed fields
    allowed_fields = {'orderId', 'userId', 'amount'}
    sanitized = {
        k: v for k, v in user_input.items()
        if k in allowed_fields
    }

    # Validate data types
    if 'amount' in sanitized:
        sanitized['amount'] = float(sanitized['amount'])

    publisher.publish(queue_name, [sanitized])

4. Rate Limiting

Implement client-side rate limiting to avoid 429 errors:

from time import time

class RateLimitedPublisher:
    def __init__(self, publisher, max_requests_per_minute=60):
        self.publisher = publisher
        self.max_requests = max_requests_per_minute
        self.requests = []

    def publish(self, queue_name, messages):
        now = time()

        # Remove requests older than 1 minute
        self.requests = [t for t in self.requests if now - t < 60]

        if len(self.requests) >= self.max_requests:
            wait = 60 - (now - self.requests[0])
            print(f"Rate limit approaching, waiting {wait:.1f}s...")
            time.sleep(wait)
            self.requests = []

        self.requests.append(now)
        return self.publisher.publish(queue_name, messages)

---

Monitoring & Observability

Track Publish Success Rate

from prometheus_client import Counter

publish_success = Counter('queue_publish_success', 'Successful publishes')
publish_failure = Counter('queue_publish_failure', 'Failed publishes')

def publish_with_metrics(queue_name, messages):
    try:
        result = publisher.publish(queue_name, messages)
        publish_success.inc(result['result']['success_count'])
        return result
    except Exception as e:
        publish_failure.inc(len(messages))
        raise

Log Message Metadata

import logging

def publish_with_logging(queue_name, messages):
    logger.info(
        f"Publishing {len(messages)} messages to {queue_name}",
        extra={
            'queue': queue_name,
            'message_count': len(messages),
            'total_size_kb': sum(
                len(json.dumps(m).encode('utf-8')) for m in messages
            ) / 1024
        }
    )

    result = publisher.publish(queue_name, messages)

    logger.info(
        f"Published {result['result']['success_count']} messages",
        extra={
            'queue': queue_name,
            'message_ids': [m['id'] for m in result['result']['messages']]
        }
    )

    return result

---

Migration from Worker Producers

Before (Worker Producer):

// Worker publishing to queue
export default {
    async fetch(request: Request, env: Env) {
        await env.MY_QUEUE.send({
            type: 'order-created',
            orderId: '12345'
        });

        return new Response('Queued');
    }
}

After (HTTP Publishing):

# External service publishing via HTTP
curl -X POST \
  "https://api.cloudflare.com/client/v4/accounts/abc123/queues/my-queue/messages" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{
      "body": {
        "type": "order-created",
        "orderId": "12345"
      }
    }]
  }'

When to Migrate:

• Publishing from non-Workers environments
• Need HTTP API integration
• External systems can't deploy Workers
• Want unified publishing interface

When to Keep Workers:

• Already using Workers for other logic
• Need lowest latency publishing
• Workers billing model is favorable

---

Additional Resources

• Official Docs: https://developers.cloudflare.com/queues/configuration/http-publishing/
• API Reference: https://developers.cloudflare.com/api/operations/queue-messages-publish
• Rate Limits: https://developers.cloudflare.com/fundamentals/api/reference/limits/
• Message Size: https://developers.cloudflare.com/queues/platform/limits/

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

Cloudflare Queues Production Deployment Checklist

Last Updated: 2025-11-26

Complete this checklist before deploying queue consumers to production to ensure reliability, scalability, and proper error handling.

---

Pre-Deployment Checklist

1. Dead Letter Queue Configuration

• [ ] DLQ Created: Separate queue created for failed messages
• [ ] DLQ Consumer Deployed: Worker deployed to monitor and process DLQ messages
• [ ] DLQ Monitoring: Alerts configured for DLQ message count thresholds
• [ ] DLQ Retention: Appropriate retention policy set (default: 4 days)

Why: Messages that fail after max retries need a recovery mechanism. Without a DLQ, failed messages are permanently lost.

Configuration Example:

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

---

2. Message Acknowledgment Strategy

• [ ] Explicit ack() for Non-Idempotent Operations: Database writes, API calls, payments use explicit message.ack()
• [ ] Implicit ack for Idempotent Operations: Read-only operations can rely on automatic ack
• [ ] Ack Timing Verified: ack() called AFTER successful processing, not before

Why: Incorrect ack patterns cause duplicate processing or message loss.

Decision Tree:

• Database write/update? → Explicit ack()
• External API call? → Explicit ack()
• Read-only/idempotent? → Implicit ack (no call needed)

---

3. Message Size Validation

• [ ] Size Check Implemented: Validate messages <128 KB before sending
• [ ] Fallback Strategy: Large payloads stored in R2/KV with reference in queue
• [ ] Error Handling: Clear error messages when size limit exceeded

Why: Messages >128 KB are rejected, causing silent failures.

Validation Pattern:

const messageSize = JSON.stringify(payload).length;
if (messageSize > 128 * 1024) {
  // Store in R2, send reference
  const key = `large-messages/${crypto.randomUUID()}`;
  await env.R2_BUCKET.put(key, JSON.stringify(payload));
  await env.MY_QUEUE.send({ type: 'large', key });
} else {
  await env.MY_QUEUE.send(payload);
}

---

4. Batch Size Optimization

• [ ] Batch Size Tested: Optimal batch size determined through load testing
• [ ] Processing Time Measured: Average time per message calculated
• [ ] Timeout Risk Assessed: Total batch processing time <30 seconds (CPU limit)

Why: Too small = inefficient. Too large = timeout risk.

Guidelines:

• Fast operations (<100ms): Batch 100 messages
• Medium operations (100ms-1s): Batch 10-50 messages
• Slow operations (>1s): Batch 1-10 messages or increase CPU limit

---

5. Concurrency Configuration

• [ ] max_concurrency NOT Set (unless specific reason): Let Workers auto-scale
• [ ] Auto-Scaling Verified: Queue automatically creates concurrent invocations under load
• [ ] Concurrency Override Justified: If set to 1, document why

Why: Setting max_concurrency: 1 disables auto-scaling, creating bottlenecks.

When to Set max_concurrency:

• ✅ Ordering guarantee required (set to 1)
• ✅ Rate limiting external API (set to match API limits)
• ❌ "Just to be safe" (this hurts performance)

---

6. CPU Limit for Long-Running Tasks

• [ ] CPU Limit Increased: If processing takes >30 seconds, increase cpu_ms
• [ ] Processing Time Measured: Actual processing time documented
• [ ] Cost Impact Calculated: Higher CPU limits = higher cost

Configuration:

{
  "queues": {
    "consumers": [
      {
        "queue": "long-running-tasks",
        "max_batch_timeout": 60,  // seconds
        "max_batch_size": 10,
        "max_retries": 2,
        "cpu_ms": 180000  // 3 minutes (default: 30s)
      }
    ]
  }
}

Guidelines:

• Default: 30,000 ms (30 seconds)
• Video processing: 180,000 ms (3 minutes)
• Image processing: 60,000 ms (1 minute)

---

7. Error Handling & Retry Logic

• [ ] Try-Catch Implemented: All message processing wrapped in error handling
• [ ] Retry Strategy Defined: Exponential backoff or fixed delay configured
• [ ] Max Attempts Respected: Consumer checks message.attempts before retry
• [ ] Unrecoverable Errors Identified: Validation failures ack immediately (no retry)

Retry Pattern:

for (const message of batch.messages) {
  try {
    await processMessage(message.body);
    message.ack();
  } catch (error) {
    console.error(`Failed (attempt ${message.attempts}):`, error);

    // Exponential backoff: 1min, 2min, 4min, 8min, 16min
    const delay = Math.min(Math.pow(2, message.attempts) * 60, 3600);

    if (message.attempts >= 5) {
      // After 5 attempts, let it go to DLQ
      message.ack();
    } else {
      message.retry({ delaySeconds: delay });
    }
  }
}

---

8. Monitoring & Alerting

• [ ] CloudWatch Metrics Configured: Message counts, processing time, error rates
• [ ] Alerts Set Up: Notifications for DLQ threshold, high error rate, consumer failures
• [ ] Dashboard Created: Real-time visibility into queue health
• [ ] Log Aggregation: Consumer logs centralized (e.g., Logpush)

Key Metrics to Monitor:

• Messages Published: Rate of incoming messages
• Messages Processed: Rate of successful processing
• DLQ Message Count: Should stay near zero
• Consumer Errors: Error rate percentage
• Processing Latency: Average time per message

---

9. Rate Limiting for External APIs

• [ ] Rate Limits Identified: External API limits documented (req/sec, req/day)
• [ ] Throttling Implemented: Queue consumer respects API limits
• [ ] Backpressure Handled: Retry with appropriate delay when rate limited

Pattern:

let requestCount = 0;
const MAX_REQUESTS_PER_SECOND = 100;

for (const message of batch.messages) {
  if (requestCount >= MAX_REQUESTS_PER_SECOND) {
    // Rate limit reached, retry remaining messages with delay
    message.retry({ delaySeconds: 60 });
    continue;
  }

  await callExternalAPI(message.body);
  requestCount++;
  message.ack();
}

---

10. Idempotency Where Possible

• [ ] Idempotent Operations Identified: Operations that can be safely repeated
• [ ] Deduplication Keys Used: For critical operations (payments, orders)
• [ ] Idempotency Tokens: Passed to external APIs

Why: Network failures or retry logic can cause duplicate processing. Idempotency prevents double-charging, duplicate records, etc.

Pattern:

// Use message ID as idempotency key
const idempotencyKey = message.id;

// Check if already processed
const existing = await env.DB.prepare(
  'SELECT * FROM processed_messages WHERE message_id = ?'
).bind(idempotencyKey).first();

if (existing) {
  message.ack(); // Already processed, skip
  continue;
}

// Process and record
await processPayment(message.body, idempotencyKey);
await env.DB.prepare(
  'INSERT INTO processed_messages (message_id, processed_at) VALUES (?, ?)'
).bind(idempotencyKey, new Date().toISOString()).run();

message.ack();

---

11. Load Testing

• [ ] Load Test Completed: Queue tested at expected production volume
• [ ] Peak Load Tested: 2-3x expected volume tested
• [ ] Failure Scenarios Tested: Network failures, timeouts, API errors
• [ ] Recovery Tested: Queue recovery after consumer deployment/restart

Load Test Checklist: 1. Send 1000 messages in 1 minute 2. Send 10,000 messages in 10 minutes 3. Introduce 50% error rate, verify retries 4. Pause consumer mid-batch, verify recovery 5. Deploy new consumer version, verify no message loss

---

12. Security Review

• [ ] Authentication Required: Producer endpoints require authentication
• [ ] Input Validation: All message bodies validated before processing
• [ ] Secret Management: API keys stored in environment variables (not in code)
• [ ] CORS Configured: If producer is browser-based

---

Deployment Workflow

Pre-Deployment

1. ✅ Complete all checklist items above 2. ✅ Review code changes in PR 3. ✅ Run integration tests 4. ✅ Document any configuration changes

Deployment

1. Deploy DLQ Consumer First (if new) 2. Deploy Producer (if changed) 3. Deploy Main Consumer (always last) 4. Verify Monitoring: Check dashboards/logs immediately

Post-Deployment

1. Monitor for 1 hour: Watch for errors, DLQ messages 2. Verify Message Flow: Send test messages, confirm processing 3. Check Latency: Ensure processing time within expected range 4. Document Issues: Any unexpected behavior noted for next deployment

---

Common Production Pitfalls

❌ Don't:

• Set max_concurrency: 1 without justification
• Use implicit ack for database writes
• Send messages >128 KB without validation
• Deploy without DLQ configured
• Ignore retry logic and exponential backoff

✅ Do:

• Let concurrency auto-scale (omit max_concurrency)
• Use explicit ack() for non-idempotent operations
• Validate message size before sending
• Always configure DLQ + monitoring
• Implement exponential backoff for retries

---

Post-Production Monitoring

Daily Checks

• DLQ message count (should be near zero)
• Error rate (should be <1%)
• Processing latency (should be consistent)

Weekly Reviews

• Capacity planning (message volume trends)
• Cost analysis (CPU usage vs throughput)
• Error pattern analysis (common failure causes)

Monthly Audits

• Load test with production volume
• Review retry patterns and adjust if needed
• Update documentation with lessons learned

---

Additional Resources

• Best Practices Guide: references/best-practices.md
• Error Catalog: references/error-catalog.md (all 10 errors with solutions)
• Wrangler Commands: references/wrangler-commands.md (CLI reference)
• Official Docs: https://developers.cloudflare.com/queues/

---

Remember: A well-configured queue is reliable, scalable, and predictable. Take time to complete this checklist—it prevents costly production issues!

Pull-Based Consumers

Official Feature: Enable applications to retrieve messages on-demand from Cloudflare Queues rather than through push mechanisms.

When to Use: Load this reference when the user asks to "pull messages from queue", "retrieve messages on-demand", "consume from non-Workers environment", or needs to integrate queues with external systems that can't use Workers consumers.

---

Overview

Pull-based consumers allow applications outside of Cloudflare Workers to retrieve messages from queues via HTTP. This enables queue integration with:

• Traditional backend services (Node.js, Python, Go servers)
• Containerized applications
• Existing microservices architectures
• Systems that can't deploy as Workers consumers

Key Difference from Push Consumers:

• Push: Worker consumer automatically receives batches (default Cloudflare Queues pattern)
• Pull: Application makes HTTP requests to retrieve message batches when ready

---

Use Cases

1. Legacy System Integration

Integrate queues with existing backend services without rewriting as Workers:

# Python service pulls messages
import requests

def process_queue_messages():
    response = requests.post(
        'https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/{queue_name}/messages/pull',
        headers={
            'Authorization': 'Bearer {api_token}',
            'Content-Type': 'application/json'
        },
        json={
            'batch_size': 10,
            'visibility_timeout': 30
        }
    )

    messages = response.json()['result']['messages']
    for msg in messages:
        process_message(msg['body'])
        # Acknowledge after processing
        ack_message(msg['id'])

2. Containerized Applications

Pull messages from Docker containers or Kubernetes pods:

// Node.js container service
const fetch = require('node-fetch');

async function pollQueue() {
    const response = await fetch(
        `https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/queues/${QUEUE_NAME}/messages/pull`,
        {
            method: 'POST',
            headers: {
                'Authorization': `Bearer ${API_TOKEN}`,
                'Content-Type': 'application/json'
            },
            body: JSON.stringify({
                batch_size: 25,
                visibility_timeout: 60
            })
        }
    );

    const { messages } = await response.json().result;
    return messages;
}

// Poll every 5 seconds
setInterval(async () => {
    const messages = await pollQueue();
    await processMessages(messages);
}, 5000);

3. On-Demand Processing

Trigger message retrieval based on external events or schedules:

// Go service - pull on webhook trigger
package main

import (
    "bytes"
    "encoding/json"
    "net/http"
)

func pullMessages(batchSize int) ([]Message, error) {
    payload := map[string]int{
        "batch_size": batchSize,
        "visibility_timeout": 45,
    }

    body, _ := json.Marshal(payload)

    req, _ := http.NewRequest(
        "POST",
        fmt.Sprintf("https://api.cloudflare.com/client/v4/accounts/%s/queues/%s/messages/pull", accountID, queueName),
        bytes.NewBuffer(body),
    )

    req.Header.Set("Authorization", "Bearer " + apiToken)
    req.Header.Set("Content-Type", "application/json")

    resp, err := client.Do(req)
    // Handle response...
}

---

HTTP Pull API Reference

Endpoint

POST https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/{queue_name}/messages/pull

Authentication

Requires Cloudflare API Token with Queue: Read and Queue: Edit permissions.

Request Headers

Authorization: Bearer {api_token}
Content-Type: application/json

Request Body

{
  "batch_size": 10,           // Number of messages to retrieve (1-100)
  "visibility_timeout": 30    // Seconds messages hidden from other consumers (1-43200)
}

Parameters:

• batch_size (required): 1-100 messages per pull request
• visibility_timeout (required): 1-43,200 seconds (12 hours max)
• Messages become invisible to other consumers for this duration
• Must ack/nack within timeout or messages return to queue

Response Format

{
  "success": true,
  "result": {
    "messages": [
      {
        "id": "msg_abc123",
        "body": {
          "type": "order-processed",
          "orderId": "12345",
          "userId": "user_789"
        },
        "timestamp": "2025-12-27T10:30:00Z",
        "attempts": 1,
        "lease_id": "lease_xyz"
      }
    ],
    "queue_name": "my-queue"
  }
}

Response Fields:

• id: Unique message identifier (for ack/nack)
• body: Message payload (JSON object)
• timestamp: When message was enqueued
• attempts: Number of delivery attempts
• lease_id: Required for acknowledging message

---

Message Acknowledgment

Explicit Ack (Success)

After successfully processing, acknowledge the message:

POST https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/{queue_name}/messages/ack

{
  "acks": [
    {
      "message_id": "msg_abc123",
      "lease_id": "lease_xyz"
    }
  ]
}

Batch Acks (up to 100 messages):

{
  "acks": [
    {"message_id": "msg_1", "lease_id": "lease_1"},
    {"message_id": "msg_2", "lease_id": "lease_2"},
    {"message_id": "msg_3", "lease_id": "lease_3"}
  ]
}

Negative Ack (Failure)

Return message to queue for retry:

POST https://api.cloudflare.com/client/v4/accounts/{account_id}/queues/{queue_name}/messages/nack

{
  "nacks": [
    {
      "message_id": "msg_abc123",
      "lease_id": "lease_xyz",
      "retry_delay": 10  // Optional: seconds before retry (default: 0)
    }
  ]
}

Visibility Timeout Expiry

If neither ack nor nack within visibility_timeout:

• Message automatically returns to queue
• Increments attempts counter
• Subject to max_retries limit

---

Polling Strategies

1. Short Polling (Recommended for Low Latency)

// Poll every 2 seconds
async function shortPoll() {
    while (true) {
        const messages = await pullMessages(10);
        if (messages.length > 0) {
            await processMessages(messages);
        }
        await sleep(2000);
    }
}

Pros: Low latency, simple implementation Cons: Higher API call volume (consider rate limits)

2. Long Polling (Reduce API Calls)

// Poll every 30 seconds, larger batches
async function longPoll() {
    while (true) {
        const messages = await pullMessages(100);
        await processMessages(messages);
        await sleep(30000);
    }
}

Pros: Fewer API calls, reduced costs Cons: Higher latency for message processing

3. Adaptive Polling (Best of Both)

// Adjust interval based on queue activity
let pollInterval = 5000; // Start at 5 seconds

async function adaptivePoll() {
    while (true) {
        const messages = await pullMessages(25);

        if (messages.length === 0) {
            // No messages - slow down polling
            pollInterval = Math.min(pollInterval * 1.5, 60000); // Max 60s
        } else {
            // Messages found - speed up polling
            pollInterval = Math.max(pollInterval * 0.7, 1000); // Min 1s
            await processMessages(messages);
        }

        await sleep(pollInterval);
    }
}

Pros: Balances latency and API efficiency Cons: More complex logic

---

Error Handling

Handle Empty Queue

const response = await pullMessages(10);
if (response.messages.length === 0) {
    console.log('Queue empty, waiting...');
    // Implement backoff strategy
}

Handle Rate Limiting

try {
    const messages = await pullMessages(10);
} catch (error) {
    if (error.status === 429) {
        console.log('Rate limited, backing off...');
        await sleep(60000); // Wait 1 minute
    }
}

Visibility Timeout Management

const VISIBILITY_TIMEOUT = 60; // 60 seconds

async function processWithTimeout(message) {
    const startTime = Date.now();

    try {
        await processMessage(message.body);

        const elapsed = (Date.now() - startTime) / 1000;
        if (elapsed < VISIBILITY_TIMEOUT - 5) {
            // Ack before timeout
            await ackMessage(message.id, message.lease_id);
        } else {
            console.error('Processing took too long, message may retry');
        }
    } catch (error) {
        // Nack to return to queue
        await nackMessage(message.id, message.lease_id, 30);
    }
}

---

Comparison: Pull vs Push Consumers

When to Use Pull:

• Existing backend services (non-Workers)
• Containerized/VM deployments
• Need explicit control over polling
• Integration with existing infrastructure

When to Use Push:

• New Workers-native applications
• Need lowest latency
• Want automatic scaling
• Prefer managed infrastructure

---

Best Practices

1. Set Appropriate Visibility Timeout

• Too short: Messages re-delivered while still processing
• Too long: Delays retries on genuine failures
• Recommendation: 2-3x average processing time

// If processing takes ~10s on average
const VISIBILITY_TIMEOUT = 30; // 3x average

2. Batch Processing

Pull multiple messages per request to reduce API overhead:

// Good: Pull 25-100 messages per request
const messages = await pullMessages(50);
await Promise.all(messages.map(processMessage));

3. Graceful Shutdown

Ensure in-flight messages are ack'd or nack'd before shutdown:

process.on('SIGTERM', async () => {
    console.log('Shutting down gracefully...');

    // Stop polling
    stopPolling();

    // Finish in-flight messages
    await Promise.all(inFlightMessages.map(async (msg) => {
        try {
            await finishProcessing(msg);
            await ackMessage(msg.id, msg.lease_id);
        } catch (error) {
            await nackMessage(msg.id, msg.lease_id);
        }
    }));

    process.exit(0);
});

4. Monitor Queue Depth

Track backlog to adjust polling behavior:

// Check queue info periodically
const queueInfo = await getQueueInfo();
if (queueInfo.backlog > 1000) {
    // Increase polling frequency or add more workers
    console.log('High backlog detected, scaling up...');
}

5. Implement Retry Logic

Handle transient failures with exponential backoff:

async function processWithRetry(message, maxRetries = 3) {
    for (let attempt = 0; attempt < maxRetries; attempt++) {
        try {
            await processMessage(message.body);
            await ackMessage(message.id, message.lease_id);
            return;
        } catch (error) {
            if (attempt === maxRetries - 1) {
                // Final failure - nack to DLQ
                await nackMessage(message.id, message.lease_id);
            } else {
                // Retry with backoff
                await sleep(Math.pow(2, attempt) * 1000);
            }
        }
    }
}

---

Security Considerations

1. Protect API Tokens

Never hardcode tokens in source code:

// ❌ Bad: Hardcoded token
const API_TOKEN = 'abc123...';

// ✅ Good: Environment variable
const API_TOKEN = process.env.CLOUDFLARE_API_TOKEN;

2. Least Privilege

Create dedicated API token with minimal permissions:

• Permission: Queue: Read, Queue: Edit
• Scope: Specific queue only (not account-wide)

3. Secure Message Storage

If messages contain sensitive data, ensure secure processing:

async function processMessage(message) {
    // Decrypt sensitive fields
    const decrypted = await decrypt(message.body.sensitive_data);

    // Process...

    // Don't log sensitive data
    console.log('Processed order', message.body.orderId); // Safe
    // console.log(decrypted); // ❌ Never log sensitive data
}

---

Rate Limits

Pull-based consumers are subject to Cloudflare API rate limits:

• Free Plan: 1,200 requests/5 minutes (4 req/s)
• Paid Plans: Higher limits (check dashboard)

Recommendation: Use adaptive polling to stay within limits while maintaining responsiveness.

---

Migration from Push to Pull

If migrating existing Worker consumer to pull-based:

Before (Worker Consumer):

// Worker with queue consumer
export default {
    async queue(batch: MessageBatch<any>, env: Env) {
        for (const message of batch.messages) {
            await processMessage(message.body);
        }
    }
}

After (Pull Consumer):

// Standalone service pulling messages
async function main() {
    while (true) {
        const response = await fetch(pullEndpoint, {
            method: 'POST',
            headers: authHeaders,
            body: JSON.stringify({ batch_size: 10, visibility_timeout: 30 })
        });

        const { messages } = await response.json().result;

        for (const msg of messages) {
            await processMessage(msg.body);
            await ackMessage(msg.id, msg.lease_id);
        }

        await sleep(5000);
    }
}

---

Troubleshooting

Problem: Messages not appearing in pull requests

• Check: Verify queue has messages (wrangler queues info)
• Check: Ensure visibility timeout hasn't hidden messages
• Solution: Wait for visibility timeout to expire or increase batch_size

Problem: Messages being processed multiple times

• Check: Verify ack is being called successfully
• Check: Processing time vs visibility timeout
• Solution: Increase visibility_timeout or optimize processing speed

Problem: Rate limit errors (429)

• Check: Polling frequency too high
• Solution: Increase poll interval or implement adaptive polling

Problem: Lease ID invalid when acking

• Check: Visibility timeout expired before ack
• Solution: Reduce processing time or increase visibility_timeout

---

Additional Resources

• Official Docs: https://developers.cloudflare.com/queues/configuration/pull-consumers/
• API Reference: https://developers.cloudflare.com/api/operations/queue-messages-pull
• Rate Limits: https://developers.cloudflare.com/fundamentals/api/reference/limits/

R2 Event Notifications Integration with Queues

Official Feature: Trigger queue messages automatically when R2 objects are created, updated, or deleted.

When to Use: Load this reference when the user asks to "trigger queue on file upload", "process R2 events", "react to R2 changes", "implement event-driven R2 workflow", or needs to automate actions based on R2 bucket events.

---

Overview

R2 Event Notifications automatically send messages to Cloudflare Queues when objects in an R2 bucket are created, updated, or deleted. This enables event-driven architectures for:

• Image processing: Resize/optimize images on upload
• Document workflows: Extract metadata, generate previews
• Data pipelines: Trigger ETL jobs when data files arrive
• Backup systems: Replicate files across buckets
• Audit logging: Track all bucket modifications

Key Benefits:

• Zero polling: No need to continuously check bucket for changes
• Near real-time: Events delivered within seconds
• Scalable: Handles millions of events automatically
• Reliable: Guaranteed delivery with retries and DLQ support

---

Architecture Pattern

┌─────────────┐
│ File Upload │
│   to R2     │
└──────┬──────┘
       │
       ▼
┌─────────────────┐      ┌───────────────┐
│   R2 Bucket     │─────▶│ Queue Message │
│  (e.g., uploads)│      │ (PutObject)   │
└─────────────────┘      └───────┬───────┘
                                 │
                                 ▼
                         ┌─────────────────┐
                         │ Queue Consumer  │
                         │    (Worker)     │
                         └────────┬────────┘
                                  │
                  ┌───────────────┼───────────────┐
                  ▼               ▼               ▼
           ┌─────────┐     ┌─────────┐    ┌─────────┐
           │  Resize │     │ Extract │    │ Update  │
           │  Image  │     │Metadata │    │Database │
           └─────────┘     └─────────┘    └─────────┘

Flow: 1. User uploads file to R2 bucket 2. R2 generates event notification 3. Event automatically published to queue 4. Queue consumer (Worker) processes event 5. Consumer performs actions (resize image, update DB, etc.)

---

Use Cases

1. Automated Image Processing

Resize and optimize images on upload:

// wrangler.jsonc - R2 bucket config
{
  "name": "image-processor",
  "r2_buckets": [
    {
      "binding": "UPLOADS",
      "bucket_name": "user-uploads",
      "event_notifications": {
        "queue": "image-processing-queue",
        "rules": [
          {
            "actions": ["PutObject"],
            "prefix": "images/",
            "suffix": [".jpg", ".png", ".webp"]
          }
        ]
      }
    }
  ],
  "queues": {
    "consumers": [
      {
        "queue": "image-processing-queue",
        "max_batch_size": 10
      }
    ]
  }
}

// Worker consumer - Process uploaded images
export default {
  async queue(batch: MessageBatch<R2Event>, env: Env) {
    for (const message of batch.messages) {
      const event = message.body;

      if (event.action === 'PutObject') {
        const objectKey = event.object.key; // e.g., "images/photo.jpg"

        // Download original image
        const object = await env.UPLOADS.get(objectKey);
        const imageData = await object.arrayBuffer();

        // Resize to thumbnails
        const thumbnail = await resizeImage(imageData, 200, 200);
        const medium = await resizeImage(imageData, 800, 800);

        // Upload resized versions
        await env.UPLOADS.put(`${objectKey}-thumb.jpg`, thumbnail);
        await env.UPLOADS.put(`${objectKey}-medium.jpg`, medium);

        // Update database
        await env.DB.prepare(
          'INSERT INTO images (original, thumbnail, medium) VALUES (?, ?, ?)'
        ).bind(objectKey, `${objectKey}-thumb.jpg`, `${objectKey}-medium.jpg`).run();
      }
    }
  }
}

2. Document Metadata Extraction

Extract metadata from uploaded PDFs/documents:

// wrangler.jsonc
{
  "r2_buckets": [
    {
      "binding": "DOCUMENTS",
      "bucket_name": "company-documents",
      "event_notifications": {
        "queue": "document-processing-queue",
        "rules": [
          {
            "actions": ["PutObject"],
            "prefix": "uploads/",
            "suffix": [".pdf", ".docx"]
          }
        ]
      }
    }
  ]
}

// Worker consumer
export default {
  async queue(batch: MessageBatch<R2Event>, env: Env) {
    for (const message of batch.messages) {
      const event = message.body;
      const objectKey = event.object.key;

      // Download document
      const object = await env.DOCUMENTS.get(objectKey);
      const documentData = await object.arrayBuffer();

      // Extract metadata (using AI Workers or external service)
      const metadata = await extractMetadata(documentData);

      // Store metadata in D1
      await env.DB.prepare(`
        INSERT INTO documents (key, filename, size, mime_type, upload_date, metadata)
        VALUES (?, ?, ?, ?, ?, ?)
      `).bind(
        objectKey,
        metadata.filename,
        event.object.size,
        metadata.mimeType,
        event.object.uploaded,
        JSON.stringify(metadata.extracted)
      ).run();

      // Index for search
      await env.VECTORIZE.insert([{
        id: objectKey,
        values: metadata.embedding,
        metadata: { filename: metadata.filename }
      }]);
    }
  }
}

3. Data Pipeline Trigger

Start ETL job when data files arrive:

// wrangler.jsonc
{
  "r2_buckets": [
    {
      "binding": "DATA_LAKE",
      "bucket_name": "analytics-data",
      "event_notifications": {
        "queue": "etl-jobs-queue",
        "rules": [
          {
            "actions": ["PutObject"],
            "prefix": "raw-data/",
            "suffix": ".csv"
          }
        ]
      }
    }
  ]
}

// Worker consumer
export default {
  async queue(batch: MessageBatch<R2Event>, env: Env) {
    for (const message of batch.messages) {
      const event = message.body;

      if (event.action === 'PutObject') {
        const objectKey = event.object.key; // e.g., "raw-data/sales-2025-12-27.csv"

        // Download CSV
        const object = await env.DATA_LAKE.get(objectKey);
        const csvData = await object.text();

        // Parse and transform
        const records = parseCSV(csvData);
        const transformed = transformRecords(records);

        // Load into database
        await batchInsert(env.DB, transformed);

        // Move to processed folder
        await env.DATA_LAKE.put(
          objectKey.replace('raw-data/', 'processed/'),
          csvData
        );

        // Delete original
        await env.DATA_LAKE.delete(objectKey);

        // Notify completion
        await env.NOTIFICATIONS.send({
          type: 'etl-complete',
          file: objectKey,
          recordCount: records.length
        });
      }
    }
  }
}

4. Cross-Region Replication

Replicate files to backup bucket on upload:

// wrangler.jsonc
{
  "r2_buckets": [
    {
      "binding": "PRIMARY",
      "bucket_name": "primary-storage",
      "event_notifications": {
        "queue": "replication-queue",
        "rules": [
          {
            "actions": ["PutObject", "DeleteObject"],
            "prefix": "critical/"
          }
        ]
      }
    },
    {
      "binding": "BACKUP",
      "bucket_name": "backup-storage"
    }
  ]
}

// Worker consumer
export default {
  async queue(batch: MessageBatch<R2Event>, env: Env) {
    for (const message of batch.messages) {
      const event = message.body;
      const objectKey = event.object.key;

      if (event.action === 'PutObject') {
        // Copy to backup bucket
        const object = await env.PRIMARY.get(objectKey);
        await env.BACKUP.put(objectKey, object.body, {
          customMetadata: {
            ...object.customMetadata,
            replicated_at: new Date().toISOString(),
            source: 'primary-storage'
          }
        });
      } else if (event.action === 'DeleteObject') {
        // Delete from backup bucket
        await env.BACKUP.delete(objectKey);
      }
    }
  }
}

---

R2 Event Notification Configuration

Enable Event Notifications

Via wrangler.jsonc:

{
  "r2_buckets": [
    {
      "binding": "MY_BUCKET",
      "bucket_name": "my-bucket",
      "event_notifications": {
        "queue": "my-events-queue",    // Required: Queue name
        "rules": [                      // Optional: Filter rules
          {
            "actions": ["PutObject"],   // Event types to trigger
            "prefix": "uploads/",       // Optional: Object key prefix
            "suffix": [".jpg", ".png"]  // Optional: Object key suffix(es)
          }
        ]
      }
    }
  ]
}

Via wrangler CLI:

# Enable event notifications for bucket
wrangler r2 bucket event-notifications create my-bucket \
  --queue my-events-queue \
  --event-type PutObject

# Add filter rule
wrangler r2 bucket event-notifications create my-bucket \
  --queue my-events-queue \
  --event-type PutObject,DeleteObject \
  --prefix uploads/ \
  --suffix .jpg,.png

Via Dashboard: 1. Go to R2 bucket settings 2. Navigate to "Event Notifications" 3. Click "Add notification" 4. Select queue and configure rules

---

Event Notification Rules

Available Event Types

Currently supported: PutObject and DeleteObject

Filter Rules

Prefix filter - Match objects by key prefix:

{
  "prefix": "images/"  // Only trigger for keys starting with "images/"
}

Suffix filter - Match objects by key suffix:

{
  "suffix": [".jpg", ".png", ".webp"]  // Only trigger for image files
}

Combined filters - Both prefix AND suffix must match:

{
  "prefix": "uploads/",
  "suffix": [".pdf", ".docx"]  // Only "uploads/*.{pdf,docx}"
}

Multiple rules - OR logic across rules:

{
  "rules": [
    {
      "actions": ["PutObject"],
      "prefix": "images/",
      "suffix": [".jpg", ".png"]
    },
    {
      "actions": ["PutObject"],
      "prefix": "documents/",
      "suffix": [".pdf"]
    }
  ]
}

---

Event Message Format

PutObject Event

{
  "account": "abc123def456",
  "bucket": "my-bucket",
  "action": "PutObject",
  "object": {
    "key": "uploads/image.jpg",
    "size": 1048576,
    "eTag": "686897696a7c876b7e",
    "uploaded": "2025-12-27T10:30:00.000Z"
  },
  "eventTime": "2025-12-27T10:30:01.234Z"
}

Fields:

• account: Cloudflare account ID
• bucket: R2 bucket name
• action: Event type (PutObject, DeleteObject)
• object.key: Full object key path
• object.size: Object size in bytes
• object.eTag: Object ETag (hash)
• object.uploaded: When object was uploaded
• eventTime: When event was generated

DeleteObject Event

{
  "account": "abc123def456",
  "bucket": "my-bucket",
  "action": "DeleteObject",
  "object": {
    "key": "uploads/old-image.jpg"
  },
  "eventTime": "2025-12-27T10:35:00.123Z"
}

Note: DeleteObject events only include key (no size, eTag, uploaded)

---

Processing Event Messages

TypeScript Types

// R2 event notification types
interface R2Event {
  account: string;
  bucket: string;
  action: 'PutObject' | 'DeleteObject';
  object: R2EventObject;
  eventTime: string;
}

interface R2EventObject {
  key: string;
  size?: number;        // Only for PutObject
  eTag?: string;        // Only for PutObject
  uploaded?: string;    // Only for PutObject
}

// Worker consumer
export default {
  async queue(batch: MessageBatch<R2Event>, env: Env): Promise<void> {
    for (const message of batch.messages) {
      const event = message.body;

      switch (event.action) {
        case 'PutObject':
          await handlePutObject(event, env);
          break;
        case 'DeleteObject':
          await handleDeleteObject(event, env);
          break;
      }
    }
  }
}

async function handlePutObject(event: R2Event, env: Env) {
  const { key, size, eTag } = event.object;

  console.log(`New object: ${key} (${size} bytes, eTag: ${eTag})`);

  // Download object
  const object = await env.MY_BUCKET.get(key);
  if (!object) {
    console.error(`Object ${key} not found`);
    return;
  }

  // Process object
  await processObject(object);
}

async function handleDeleteObject(event: R2Event, env: Env) {
  const { key } = event.object;

  console.log(`Deleted object: ${key}`);

  // Clean up related data
  await env.DB.prepare('DELETE FROM objects WHERE key = ?').bind(key).run();
}

Accessing Object Content

async function processUploadedObject(event: R2Event, env: Env) {
  // Get object from bucket
  const object = await env.MY_BUCKET.get(event.object.key);

  if (!object) {
    console.error(`Object ${event.object.key} not found`);
    return;
  }

  // Read content in different formats
  const arrayBuffer = await object.arrayBuffer(); // Binary data
  const text = await object.text();                // Text content
  const stream = object.body;                       // Stream

  // Access metadata
  const metadata = {
    httpMetadata: object.httpMetadata,      // Content-Type, Cache-Control, etc.
    customMetadata: object.customMetadata,   // User-defined metadata
    size: object.size,
    etag: object.etag,
    uploaded: object.uploaded
  };

  console.log(`Processing ${event.object.key}:`, metadata);
}

---

Best Practices

1. Filter Events at Source

Use prefix/suffix filters to reduce unnecessary queue messages:

// ❌ Bad: Process all events, filter in consumer
{
  "event_notifications": {
    "queue": "all-events-queue",
    "rules": [{ "actions": ["PutObject"] }]  // No filtering
  }
}

// Consumer must filter:
if (event.object.key.endsWith('.jpg')) {
  await processImage(event);
}

// ✅ Good: Filter at R2 level
{
  "event_notifications": {
    "queue": "image-events-queue",
    "rules": [
      {
        "actions": ["PutObject"],
        "suffix": [".jpg", ".png", ".webp"]  // Filter at source
      }
    ]
  }
}

// Consumer processes only relevant events
await processImage(event);

2. Handle Missing Objects

Objects may be deleted between event and processing:

async function processEvent(event: R2Event, env: Env) {
  const object = await env.MY_BUCKET.get(event.object.key);

  if (!object) {
    // Object was deleted - skip processing
    console.log(`Object ${event.object.key} already deleted, skipping`);
    return;
  }

  await processObject(object);
}

3. Idempotent Processing

Ensure processing can safely run multiple times:

async function processImageIdempotent(event: R2Event, env: Env) {
  const key = event.object.key;
  const thumbnailKey = `${key}-thumb.jpg`;

  // Check if already processed
  const existing = await env.MY_BUCKET.head(thumbnailKey);
  if (existing) {
    console.log(`Thumbnail ${thumbnailKey} already exists, skipping`);
    return;
  }

  // Process image
  const object = await env.MY_BUCKET.get(key);
  const thumbnail = await resizeImage(await object.arrayBuffer(), 200, 200);
  await env.MY_BUCKET.put(thumbnailKey, thumbnail);
}

4. Batch Operations

Process multiple events efficiently:

export default {
  async queue(batch: MessageBatch<R2Event>, env: Env) {
    const putEvents = batch.messages
      .map(m => m.body)
      .filter(e => e.action === 'PutObject');

    // Batch download objects
    const objects = await Promise.all(
      putEvents.map(e => env.MY_BUCKET.get(e.object.key))
    );

    // Batch process
    const processed = await Promise.all(
      objects.map(obj => processObject(obj))
    );

    // Batch upload results
    await Promise.all(
      processed.map((data, i) =>
        env.MY_BUCKET.put(`processed/${putEvents[i].object.key}`, data)
      )
    );
  }
}

5. Error Handling with DLQ

Configure Dead Letter Queue for failed processing:

{
  "queues": {
    "consumers": [
      {
        "queue": "image-processing-queue",
        "max_batch_size": 10,
        "max_retries": 3,
        "dead_letter_queue": "image-processing-dlq"
      }
    ]
  }
}

async function processWithErrorHandling(event: R2Event, env: Env) {
  try {
    const object = await env.MY_BUCKET.get(event.object.key);

    if (!object) {
      // Object deleted - not an error, skip
      return;
    }

    await processObject(object);
  } catch (error) {
    console.error(`Failed to process ${event.object.key}:`, error);

    // Log to analytics
    await env.ANALYTICS.put(`errors/${event.object.key}`, {
      error: error.message,
      event,
      timestamp: new Date().toISOString()
    });

    // Re-throw to trigger retry/DLQ
    throw error;
  }
}

---

Advanced Patterns

Pattern 1: Multi-Stage Processing Pipeline

// Stage 1: Initial upload triggers image queue
// R2 event → image-processing-queue → Resize & Upload thumbnails

// Stage 2: Thumbnail upload triggers metadata queue
// R2 event → metadata-extraction-queue → Extract metadata

// wrangler.jsonc
{
  "r2_buckets": [
    {
      "binding": "IMAGES",
      "bucket_name": "user-images",
      "event_notifications": {
        "queue": "image-processing-queue",
        "rules": [
          {
            "actions": ["PutObject"],
            "prefix": "originals/",
            "suffix": [".jpg", ".png"]
          }
        ]
      }
    },
    {
      "binding": "THUMBNAILS",
      "bucket_name": "user-images",
      "event_notifications": {
        "queue": "metadata-extraction-queue",
        "rules": [
          {
            "actions": ["PutObject"],
            "prefix": "thumbnails/"
          }
        ]
      }
    }
  ]
}

Pattern 2: Conditional Processing Based on Metadata

async function processConditionally(event: R2Event, env: Env) {
  const object = await env.MY_BUCKET.get(event.object.key);

  // Check custom metadata
  const processType = object.customMetadata?.processType;

  switch (processType) {
    case 'image':
      await processImage(object);
      break;
    case 'video':
      await processVideo(object);
      break;
    case 'document':
      await processDocument(object);
      break;
    default:
      console.log(`Unknown process type: ${processType}`);
  }
}

Pattern 3: Fan-Out to Multiple Queues

// Single R2 event triggers multiple processing pipelines

// wrangler.jsonc - Multiple event notification configs
{
  "r2_buckets": [
    {
      "binding": "UPLOADS",
      "bucket_name": "user-uploads",
      "event_notifications": [
        {
          "queue": "thumbnail-queue",
          "rules": [{ "actions": ["PutObject"], "prefix": "images/" }]
        },
        {
          "queue": "metadata-queue",
          "rules": [{ "actions": ["PutObject"], "prefix": "images/" }]
        },
        {
          "queue": "backup-queue",
          "rules": [{ "actions": ["PutObject"], "prefix": "images/" }]
        }
      ]
    }
  ]
}

// Each queue has dedicated consumer:
// - thumbnail-queue → Resize images
// - metadata-queue → Extract EXIF data
// - backup-queue → Replicate to backup bucket

---

Monitoring & Debugging

Log Event Details

export default {
  async queue(batch: MessageBatch<R2Event>, env: Env) {
    console.log(`Processing ${batch.messages.length} R2 events`);

    for (const message of batch.messages) {
      const event = message.body;

      console.log({
        action: event.action,
        bucket: event.bucket,
        key: event.object.key,
        size: event.object.size,
        eventTime: event.eventTime
      });

      await processEvent(event, env);
    }
  }
}

Track Processing Metrics

import { Analytics } from '@cloudflare/workers-types';

async function trackMetrics(event: R2Event, env: Env) {
  await env.ANALYTICS.writeDataPoint({
    indexes: [event.bucket, event.action],
    doubles: [event.object.size || 0],
    blobs: [event.object.key]
  });
}

Check Queue Backlog

# Monitor queue status
wrangler queues info image-processing-queue

# Output:
# Queue: image-processing-queue
# Messages: 125 (backlog)
# Consumers: 1

---

Limits & Quotas

Note: Event notifications count toward queue message limits (50 messages/invocation on free tier, 1000 on paid).

---

Troubleshooting

Problem: Events not appearing in queue

• Check: Verify event notification configuration in wrangler.jsonc
• Check: Ensure queue exists: wrangler queues list
• Check: Upload object matching prefix/suffix filters
• Solution: Test with simple rule (no filters) first

Problem: Duplicate event processing

• Check: Multiple consumers on same queue
• Check: Consumer not acking messages
• Solution: Implement idempotent processing (check if already processed)

Problem: Queue backlog growing

• Check: Consumer processing too slow
• Check: Increase max_batch_size or max_concurrency
• Solution: Optimize processing logic, add more consumers

Problem: Object not found when processing event

• Check: Object deleted between event and processing
• Solution: Handle gracefully (check if object exists before processing)

---

Migration Guide

From Polling to Event Notifications

Before (Polling):

// Worker polls R2 bucket every minute
export default {
  async scheduled(event: ScheduledEvent, env: Env) {
    // List new objects
    const listed = await env.MY_BUCKET.list({ prefix: 'uploads/' });

    for (const object of listed.objects) {
      // Check if already processed
      const processed = await env.DB.prepare(
        'SELECT 1 FROM processed WHERE key = ?'
      ).bind(object.key).first();

      if (!processed) {
        await processObject(object.key, env);
        await env.DB.prepare(
          'INSERT INTO processed (key) VALUES (?)'
        ).bind(object.key).run();
      }
    }
  }
}

After (Event Notifications):

// Automatic event-driven processing
export default {
  async queue(batch: MessageBatch<R2Event>, env: Env) {
    for (const message of batch.messages) {
      const event = message.body;

      if (event.action === 'PutObject') {
        await processObject(event.object.key, env);
      }
    }
  }
}

Benefits of migration:

• ✅ No polling overhead
• ✅ Near real-time processing
• ✅ Automatic scaling
• ✅ Lower costs (no scheduled invocations)

---

Additional Resources

• Official Docs: https://developers.cloudflare.com/r2/buckets/event-notifications/
• Queue Configuration: https://developers.cloudflare.com/queues/configuration/
• R2 API: https://developers.cloudflare.com/r2/api/workers/workers-api/
• Examples: https://developers.cloudflare.com/r2/examples/