1# Queues API Reference
2 
3## Producer: Send Messages
4 
5```typescript
6// Basic send
7await env.MY_QUEUE.send({ url: request.url, timestamp: Date.now() });
8 
9// Options: delay (max 43200s), contentType (json|text|bytes|v8)
10await env.MY_QUEUE.send(message, { delaySeconds: 600 });
11await env.MY_QUEUE.send(message, { delaySeconds: 0 }); // Override queue default
12 
13// Batch (up to 100 msgs or 256 KB)
14await env.MY_QUEUE.sendBatch([
15  { body: 'msg1' },
16  { body: 'msg2' },
17  { body: 'msg3', options: { delaySeconds: 300 } }
18]);
19 
20// Non-blocking with ctx.waitUntil - send continues after response
21ctx.waitUntil(env.MY_QUEUE.send({ data: 'async' }));
22 
23// Background tasks in queue consumer
24export default {
25  async queue(batch: MessageBatch, env: Env, ctx: ExecutionContext): Promise<void> {
26    for (const msg of batch.messages) {
27      await processMessage(msg.body);
28      
29      // Fire-and-forget analytics (doesn't block ack)
30      ctx.waitUntil(
31        env.ANALYTICS_QUEUE.send({ messageId: msg.id, processedAt: Date.now() })
32      );
33      
34      msg.ack();
35    }
36  }
37};
38```
39 
40## Consumer: Push-based (Worker)
41 
42```typescript
43// Type-safe handler with ExportedHandler
44interface Env {
45  MY_QUEUE: Queue;
46  DB: D1Database;
47}
48 
49export default {
50  async queue(batch: MessageBatch<MessageBody>, env: Env, ctx: ExecutionContext): Promise<void> {
51    // batch.queue, batch.messages.length
52    for (const msg of batch.messages) {
53      // msg.id, msg.body, msg.timestamp, msg.attempts
54      try {
55        await processMessage(msg.body);
56        msg.ack();
57      } catch (error) {
58        msg.retry({ delaySeconds: 600 });
59      }
60    }
61  }
62} satisfies ExportedHandler<Env>;
63```
64 
65**CRITICAL WARNINGS:**
66 
671. **Messages not explicitly ack'd or retry'd will auto-retry indefinitely** until `max_retries` is reached. Always call `msg.ack()` or `msg.retry()` for each message.
68 
692. **Throwing uncaught errors retries the ENTIRE batch**, not just the failed message. Always wrap individual message processing in try/catch and call `msg.retry()` explicitly per message.
70 
71```typescript
72// ❌ BAD: Uncaught error retries entire batch
73async queue(batch: MessageBatch): Promise<void> {
74  for (const msg of batch.messages) {
75    await riskyOperation(msg.body); // If this throws, entire batch retries
76    msg.ack();
77  }
78}
79 
80// ✅ GOOD: Catch per message, handle individually
81async queue(batch: MessageBatch): Promise<void> {
82  for (const msg of batch.messages) {
83    try {
84      await riskyOperation(msg.body);
85      msg.ack();
86    } catch (error) {
87      msg.retry({ delaySeconds: 60 });
88    }
89  }
90}
91```
92 
93## Ack/Retry Precedence Rules
94 
951. **Per-message calls take precedence**: If you call both `msg.ack()` and `msg.retry()`, last call wins
962. **Batch calls don't override**: `batch.ackAll()` only affects messages without explicit ack/retry
973. **No action = automatic retry**: Messages with no explicit action retry with configured delay
98 
99```typescript
100async queue(batch: MessageBatch): Promise<void> {
101  for (const msg of batch.messages) {
102    msg.ack();        // Message marked for ack
103    msg.retry();      // Overrides ack - message will retry
104  }
105  
106  batch.ackAll();     // Only affects messages not explicitly handled above
107}
108```
109 
110## Batch Operations
111 
112```typescript
113// Acknowledge entire batch
114try {
115  await bulkProcess(batch.messages);
116  batch.ackAll();
117} catch (error) {
118  batch.retryAll({ delaySeconds: 300 });
119}
120```
121 
122## Exponential Backoff
123 
124```typescript
125async queue(batch: MessageBatch, env: Env): Promise<void> {
126  for (const msg of batch.messages) {
127    try {
128      await processMessage(msg.body);
129      msg.ack();
130    } catch (error) {
131      // 30s, 60s, 120s, 240s, 480s, ... up to 12h max
132      const delay = Math.min(30 * (2 ** msg.attempts), 43200);
133      msg.retry({ delaySeconds: delay });
134    }
135  }
136}
137```
138 
139## Multiple Queues, Single Consumer
140 
141```typescript
142export default {
143  async queue(batch: MessageBatch, env: Env): Promise<void> {
144    switch (batch.queue) {
145      case 'high-priority': await processUrgent(batch.messages); break;
146      case 'low-priority': await processDeferred(batch.messages); break;
147      case 'email': await sendEmails(batch.messages); break;
148      default: batch.retryAll();
149    }
150  }
151};
152```
153 
154## Consumer: Pull-based (HTTP)
155 
156```typescript
157// Pull messages
158const response = await fetch(
159  `https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull`,
160  {
161    method: 'POST',
162    headers: { 'authorization': `Bearer ${API_TOKEN}`, 'content-type': 'application/json' },
163    body: JSON.stringify({ visibility_timeout_ms: 6000, batch_size: 50 })
164  }
165);
166 
167const data = await response.json();
168 
169// Acknowledge
170await fetch(
171  `https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack`,
172  {
173    method: 'POST',
174    headers: { 'authorization': `Bearer ${API_TOKEN}`, 'content-type': 'application/json' },
175    body: JSON.stringify({
176      acks: [{ lease_id: msg.lease_id }],
177      retries: [{ lease_id: msg2.lease_id, delay_seconds: 600 }]
178    })
179  }
180);
181```
182 
183## Interfaces
184 
185```typescript
186interface MessageBatch<Body = unknown> {
187  readonly queue: string;
188  readonly messages: Message<Body>[];
189  ackAll(): void;
190  retryAll(options?: QueueRetryOptions): void;
191}
192 
193interface Message<Body = unknown> {
194  readonly id: string;
195  readonly timestamp: Date;
196  readonly body: Body;
197  readonly attempts: number;
198  ack(): void;
199  retry(options?: QueueRetryOptions): void;
200}
201 
202interface QueueSendOptions {
203  contentType?: 'text' | 'bytes' | 'json' | 'v8';
204  delaySeconds?: number; // 0-43200
205}
206```
207