1---
2title: Queues
3description: Cloudflare Queues give you reliable, at-least-once message delivery between Workers — a WriteQueue producer binding on one side and an Effect-style consumeQueueMessages handler with automatic ack/retry on the other.
4---
5
6A Cloudflare Queue decouples work from the request that triggered
7it: a Worker *produces* messages onto the queue, Cloudflare buffers
8them, and a consumer Worker receives them in batches with
9**at-least-once** delivery — failed batches are retried and
10eventually dead-lettered.
11
12Reach for a Queue whenever work shouldn't block a response: sending
13emails, processing uploads, fanning out webhooks, or smoothing
14bursty traffic. Both halves live in the same alchemy program: the
15producer side is the `Cloudflare.Queues.WriteQueue(...)` binding
16([reference](/providers/cloudflare/queues/queue)), and the consumer
17side is the Effect-style
18`Cloudflare.Queues.consumeQueueMessages(queue, handler)` API this
19page walks through.
20
21`consumeQueueMessages(...)` does both halves in one call: it registers a
22runtime queue listener on the Worker, and it auto-creates the
23`Cloudflare.Queues.Consumer` resource that tells Cloudflare to
24dispatch messages from the queue to this Worker. No separate
25deploy-time wiring is needed.
26
27By the end you'll have a Worker that:
28
29- Sends a JSON message via `POST /queue/send`.
30- Receives the message in a queue handler registered on the same
31 Worker, persists the body to R2, and acks the batch.
32- Reads the persisted body via `GET /queue/result/:id`.
33
34## Create the Queue and Bucket
35
36Both resources are plain `yield*` calls — no special config.
37
38```typescript
39// alchemy.run.ts
40import * as Alchemy from "alchemy";
41import * as Cloudflare from "alchemy/Cloudflare";
42import * as Effect from "effect/Effect";
43
44export const Queue = Cloudflare.Queues.Queue("Queue");
45export const Bucket = Cloudflare.R2.Bucket("Bucket");
46```
47
48The Queue's name is generated from the stack/stage/id. The Bucket
49will store each consumed message at `/queue/<id>` so the integ
50test can read it back.
51
52## Bind the Queue producer
53
54In the Worker's Construction phase, yield the queue resource and ask
55`Cloudflare.Queues.WriteQueue` for a typed sender. The sender exposes `send` /
56`sendBatch` that round-trip through Cloudflare's runtime.
57
58```typescript
59// src/Api.ts
60import * as Cloudflare from "alchemy/Cloudflare";
61import * as Effect from "effect/Effect";
62import { Bucket } from "./Bucket.ts";
63import { Queue } from "./Queue.ts";
64
65export default Cloudflare.Worker(
66 "Api",
67 { main: import.meta.url },
68 Effect.gen(function* () {
69 const bucket = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
70 const queueResource = yield* Queue;
71 const queue = yield* Cloudflare.Queues.WriteQueue(queueResource);
72
73 return { fetch: Effect.gen(function* () { return new Response("ok"); }) };
74 }).pipe(
75 Effect.provide(Cloudflare.Queues.WriteQueueBinding),
76 Effect.provide(Cloudflare.R2.ReadWriteBucketBinding),
77 ),
78);
79```
80
81`queueResource` (the resolved `Queue` resource) is what you pass to
82`consumeQueueMessages(...)` next — it's the same handle, not a new one.
83
84## Subscribe to incoming messages
85
86`Cloudflare.Queues.consumeQueueMessages(queue, handler)` registers a queue
87event listener on the Worker. The handler receives a
88`Stream.Stream<Message<Body>>` — one stream per batch — and is
89expected to return `Effect.Effect<void>`.
90
91```diff lang="typescript"
92+import * as Stream from "effect/Stream";
93+
94+interface QueueMessageBody {
95+ id: string;
96+ text: string;
97+ sentAt: number;
98+}
99
100 Effect.gen(function* () {
101 const bucket = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
102 const queueResource = yield* Queue;
103 const queue = yield* Cloudflare.Queues.WriteQueue(queueResource);
104
105+ yield* Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(
106+ queueResource,
107+ (stream) =>
108+ Stream.runForEach(stream, (msg) =>
109+ bucket
110+ .put(`/queue/${msg.body.id}`, JSON.stringify(msg.body), {
111+ httpMetadata: { contentType: "application/json" },
112+ })
113+ .pipe(Effect.asVoid),
114+ ),
115+ );
116```
117
118Acking is automatic: if the handler succeeds, every message in the
119batch is `ack()`ed; if it fails, every message is `retry()`ed and
120Cloudflare applies the consumer's `maxRetries` and `retryDelay`
121before dead-lettering. For finer control, call `msg.ack()` /
122`msg.retry()` per message inside the handler.
123
124## Tune batching and backoff
125
126`consumeQueueMessages(queue, props)` accepts a settings object. Time fields are
127`Duration.Input` — any of these are accepted:
128
129```typescript
130maxWaitTime: Duration.seconds(5) // a Duration value
131maxWaitTime: 5_000 // a number of milliseconds
132maxWaitTime: "5 seconds" // a "<number> <unit>" string
133retryDelay: "1 second" // singular and plural units both work
134```
135
136```diff lang="typescript"
137+import * as Duration from "effect/Duration";
138
139- Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(queueResource, (stream) =>
140+ Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(
141+ queueResource,
142+ {
143+ batchSize: 25,
144+ maxRetries: 3,
145+ maxWaitTime: "5 seconds",
146+ retryDelay: Duration.seconds(30),
147+ },
148+ (stream) =>
149```
150
151`maxWaitTime` is rounded up to whole milliseconds and `retryDelay`
152to whole seconds when forwarded to Cloudflare.
153
154## Provide the runtime layer
155
156`consumeQueueMessages(...)` is a `Context.Service` call —
157`EventSourceLive` is the layer that registers the listener
158with the Worker's runtime context. Add it to the layer stack
159alongside the other binding lives.
160
161```diff lang="typescript"
162 }).pipe(
163+ Effect.provide(Cloudflare.Queues.EventSourceLive),
164 Effect.provide(Cloudflare.Queues.WriteQueueBinding),
165 Effect.provide(Cloudflare.R2.ReadWriteBucketBinding),
166 ),
167```
168
169Without the live layer, the `consumeQueueMessages` call fails at deploy with
170`Service not found: Cloudflare.Queues.EventSource`.
171
172## Add the producer route
173
174`POST /queue/send` enqueues a message and returns the generated id.
175The integ test uses the id to poll for the consumed result.
176
177```typescript
178return {
179 fetch: Effect.gen(function* () {
180 const request = yield* HttpServerRequest;
181 if (request.url === "/queue/send" && request.method === "POST") {
182 const text = yield* request.text;
183 const msg: QueueMessageBody = {
184 id: crypto.randomUUID(),
185 text,
186 sentAt: Date.now(),
187 };
188 yield* queue.send(msg).pipe(Effect.orDie);
189 return yield* HttpServerResponse.json({ sent: msg }, { status: 202 });
190 }
191 return HttpServerResponse.text("Not Found", { status: 404 });
192 }),
193};
194```
195
196## Add the result-read route
197
198`GET /queue/result/:id` reads `/queue/<id>` from the bucket. The
199consumer runs asynchronously, so the test polls this route with a
200short backoff until the object appears.
201
202```typescript
203if (request.url.startsWith("/queue/result/") && request.method === "GET") {
204 const id = request.url.split("/queue/result/")[1];
205 return yield* bucket.get(`/queue/${id}`).pipe(
206 Effect.flatMap((object) =>
207 object === null
208 ? Effect.succeed(HttpServerResponse.text("not yet", { status: 404 }))
209 : object.text().pipe(
210 Effect.map((body) =>
211 HttpServerResponse.text(body, {
212 headers: { "content-type": "application/json" },
213 }),
214 ),
215 ),
216 ),
217 Effect.catchTag("R2Error", (e) =>
218 Effect.succeed(HttpServerResponse.text(e.message, { status: 500 })),
219 ),
220 );
221}
222```
223
224## What's the difference vs. a native `queue()` handler?
225
226Cloudflare's runtime delivers queue events to a `queue(batch, env)`
227export on the worker module. You can write that directly — see
228[`examples/cloudflare-worker-async`](https://github.com/alchemy-run/alchemy/blob/main/examples/cloudflare-worker-async)
229for the plain async-handler shape:
230
231```typescript
232export default {
233 async queue(batch, env) {
234 for (const msg of batch.messages) {
235 await env.Bucket.put(`/queue/${msg.body.id}`, ...);
236 msg.ack();
237 }
238 },
239};
240```
241
242`Cloudflare.Queues.consumeQueueMessages(queue, handler)` is the same primitive
243on the Effect side — the Worker bundle's runtime context routes
244the dispatch to the registered listener — but you get
245`Effect.gen` composition, typed errors, automatic batch
246ack/retry, and the same surface as `AWS.SQS.consumeQueueMessages(queue, handler)`.
247
248## Receive Workflow lifecycle events
249
250A subscription delivers platform events into a Queue. For a Workflow
251bound to an async Worker, pass the binding directly as the source:
252
253```typescript
254const worker = yield* Cloudflare.Worker("IngestionWorker", {
255 main: "./src/worker.ts",
256 env: {
257 INGESTION: Cloudflare.Workflow("Ingestion", {
258 className: "IngestionWorkflow",
259 }),
260 },
261});
262const queue = yield* Cloudflare.Queues.Queue("WorkflowEventsQueue");
263
264yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
265 source: worker.env.INGESTION,
266 events: ["instance.completed", "instance.errored"],
267 queueId: queue.queueId,
268});
269```
270
271The subscription uses the binding's deferred physical name, including on
272the first deployment. Renaming the Workflow replaces the subscription.
273The Queue still needs a consumer to handle these messages, as above.
274
275A `WorkflowResource` or a reference to a deployed Workflow is also a source:
276
277```typescript
278yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
279 source: yield* Cloudflare.Workflow.ref("Ingestion", {
280 stack: "workflow-host",
281 stage: "production",
282 }),
283 events: ["instance.completed", "instance.errored"],
284 queueId: queue.queueId,
285});
286```
287
288Use the Workflow's logical ID, including any namespace. Omit the options for
289the current stack and stage. References read persisted state, so deploy the
290host first; deleting the consumer stack does not delete the referenced Workflow.
291
292Explicit sources remain supported, including
293`{ type: "workflows.workflow", workflowName: "existing-ingestion" }` for a
294Workflow referenced by name and `{ type: "r2" }` for R2 events. Cloudflare
295allows at most one subscription per source per account.
296
297## Use resources as event sources
298
299A subscription also accepts these resource values, or their yielded `.ref(...)` references:
300
301| Resource | Cloudflare source | Event scope |
302| --- | --- | --- |
303| `Cloudflare.Images.Variant` | `images` | All Images events in the account |
304| `Cloudflare.KV.Namespace` | `kv` | All namespace events in the account |
305| `Cloudflare.R2.Bucket` | `r2` | All bucket events in the account |
306| `Cloudflare.R2.SuperSlurperJob` | `superSlurper` | All migration job events in the account |
307| `Cloudflare.Vectorize.Index` | `vectorize` | All index events in the account |
308| `Cloudflare.AI.Model` | `workersAi.model` | Batch events for the selected model |
309| `Cloudflare.Worker` | `workersBuilds.worker` | Builds for the selected Worker |
310| `Cloudflare.Workflows.WorkflowResource` | `workflows.workflow` | Instances of the selected Workflow |
311
312The resource's account must match the subscription's Cloudflare account. Account metadata remains in Alchemy state and is not sent as an extra Cloudflare source field. A collision with an existing subscription fails rather than changing another stack's destination Queue.
313
314Event delivery can lag subscription creation or replacement even after the destination Queue accepts messages. Deployment confirms configuration, not delivery readiness. Verify delivery before emitting events that must be observed.
315
316During Vectorize subscription replacement or a destination Queue update, Cloudflare can still route new events to the previous Queue. Replacement events can carry the deleted subscription's ID; updates retain the same subscription ID. A single early event does not prove that routing has fully propagated. Keep the previous destination available during the transition and verify the receiving Queue, `metadata.eventSubscriptionId`, and the event's resource identity.
317
318### Account-wide resource events
319
320```typescript
321const cache = yield* Cloudflare.KV.Namespace("Cache");
322
323yield* Cloudflare.Queues.Subscription("NamespaceEvents", {
324 source: cache,
325 events: ["namespace.created", "namespace.deleted"],
326 queueId: queue.queueId,
327});
328```
329
330This receives namespace events for the **entire account**, not just `Cache`. R2, Vectorize, Images, and Super Slurper have the same account-wide behavior. Passing a resource creates a deployment dependency, so its initial creation can precede the subscription and its final deletion can follow subscription removal. Use the explicit `{ type: "kv" }` form if the subscription must be deployed before a namespace is created.
331
332### References from another stack
333
334```typescript
335yield* Cloudflare.Queues.Subscription("BucketEvents", {
336 source: yield* Cloudflare.R2.Bucket.ref("Uploads", {
337 stack: "storage",
338 stage: "production",
339 }),
340 events: ["bucket.created", "bucket.deleted"],
341 queueId: queue.queueId,
342});
343```
344
345All the resource types above support `.ref`. The source must already exist in Alchemy state. Removing this subscription preserves the referenced resource and its source stack.
346
347### Images upload events
348
349```typescript
350yield* Cloudflare.Queues.Subscription("ImageEvents", {
351 source: yield* Cloudflare.Images.Variant.ref("Thumbnail"),
352 events: ["image.uploaded"],
353 queueId: queue.queueId,
354});
355```
356
357The variant selects its Images account. The subscription receives uploads across that account, not only images served with `Thumbnail`.
358
359### Vectorize events
360
361```typescript
362yield* Cloudflare.Queues.Subscription("IndexEvents", {
363 source: yield* Cloudflare.Vectorize.Index.ref("Search"),
364 events: ["index.created", "index.deleted"],
365 queueId: queue.queueId,
366});
367```
368
369These events describe index lifecycle changes, not individual vector writes.
370
371### Super Slurper events
372
373```typescript
374yield* Cloudflare.Queues.Subscription("MigrationEvents", {
375 source: yield* Cloudflare.R2.SuperSlurperJob.ref("Migration"),
376 events: ["job.started", "job.completed", "job.aborted"],
377 queueId: queue.queueId,
378});
379```
380
381`R2.SuperSlurperJob` manages a one-off migration. Destroying an active job cancels it; it does not delete migrated objects or terminal job history. This subscription selects all jobs in the account, not object-level events from one job.
382
383### Workers AI batch events
384
385```typescript
386const model = yield* Cloudflare.AI.Model("Embeddings", {
387 modelName: "@cf/baai/bge-m3",
388});
389
390yield* Cloudflare.Queues.Subscription("BatchEvents", {
391 source: model,
392 events: ["batch.queued", "batch.succeeded", "batch.failed"],
393 queueId: queue.queueId,
394});
395```
396
397`AI.Model` validates a Cloudflare-managed catalog model and persists a non-owning handle. It neither deploys nor invokes the model. `yield* Cloudflare.AI.Model.ref("Embeddings")` can be used in place of `model` after deployment. Batch events require asynchronous batch inference; ordinary synchronous inference does not emit them.
398
399### Workers Builds events
400
401```typescript
402yield* Cloudflare.Queues.Subscription("BuildEvents", {
403 source: yield* Cloudflare.Worker.ref("Website"),
404 events: ["build.started", "build.succeeded", "build.failed"],
405 queueId: queue.queueId,
406});
407```
408
409A direct Worker resource is also accepted. The source uses the Worker's physical name and follows replacements. The Worker needs a Workers Builds integration to emit these events; an ordinary Alchemy upload is not a Workers Builds run.
410
411## Where next
412
413Related:
414
415- [Workers](/cloudflare/compute/workers) — the runtime that produces and
416 consumes queue messages.
417- [Durable Objects](/cloudflare/compute/durable-objects) — for stateful
418 coordination instead of fire-and-forget work.
419- [Workflows](/cloudflare/compute/workflows) — multi-step orchestration with
420 checkpointed steps when fire-and-forget isn't enough.
421
422Reference:
423
424- [Queue API reference](/providers/cloudflare/queues/queue)
425- [Consumer API reference](/providers/cloudflare/queues/consumer)
426- [Subscription API reference](/providers/cloudflare/queues/subscription)
427- [Workers AI model handle API reference](/providers/cloudflare/ai/model)
428- [Super Slurper migration API reference](/providers/cloudflare/r2/superslurperjob)