Queues

A Cloudflare Queue decouples work from the request that triggered it: a Worker produces messages onto the queue, Cloudflare buffers them, and a consumer Worker receives them in batches with at-least-once delivery — failed batches are retried and eventually dead-lettered.

Reach for a Queue whenever work shouldn’t block a response: sending emails, processing uploads, fanning out webhooks, or smoothing bursty traffic. Both halves live in the same alchemy program: the producer side is the Cloudflare.Queues.WriteQueue(...) binding (reference), and the consumer side is the Effect-style Cloudflare.Queues.consumeQueueMessages(queue, handler) API this page walks through.

consumeQueueMessages(...) does both halves in one call: it registers a runtime queue listener on the Worker, and it auto-creates the Cloudflare.Queues.Consumer resource that tells Cloudflare to dispatch messages from the queue to this Worker. No separate deploy-time wiring is needed.

By the end you’ll have a Worker that:

Sends a JSON message via POST /queue/send.
Receives the message in a queue handler registered on the same Worker, persists the body to R2, and acks the batch.
Reads the persisted body via GET /queue/result/:id.

Create the Queue and Bucket

Both resources are plain yield* calls — no special config.

import * as Alchemy from "alchemy";
import * as Cloudflare from "alchemy/Cloudflare";
import * as Effect from "effect/Effect";

export const Queue = Cloudflare.Queues.Queue("Queue");
export const Bucket = Cloudflare.R2.Bucket("Bucket");

The Queue’s name is generated from the stack/stage/id. The Bucket will store each consumed message at /queue/<id> so the integ test can read it back.

Bind the Queue producer

In the Worker’s Construction phase, yield the queue resource and ask Cloudflare.Queues.WriteQueue for a typed sender. The sender exposes send / sendBatch that round-trip through Cloudflare’s runtime.

import * as Cloudflare from "alchemy/Cloudflare";
import * as Effect from "effect/Effect";
import { Bucket } from "./Bucket.ts";
import { Queue } from "./Queue.ts";

export default Cloudflare.Worker(
  "Api",
  { main: import.meta.url },
  Effect.gen(function* () {
    const bucket = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
    const queueResource = yield* Queue;
    const queue = yield* Cloudflare.Queues.WriteQueue(queueResource);

    return { fetch: Effect.gen(function* () { return new Response("ok"); }) };
  }).pipe(
    Effect.provide(Cloudflare.Queues.WriteQueueBinding),
    Effect.provide(Cloudflare.R2.ReadWriteBucketBinding),
  ),
);

queueResource (the resolved Queue resource) is what you pass to consumeQueueMessages(...) next — it’s the same handle, not a new one.

Cloudflare.Queues.consumeQueueMessages(queue, handler) registers a queue event listener on the Worker. The handler receives a Stream.Stream<Message<Body>> — one stream per batch — and is expected to return Effect.Effect<void>.

import * as Stream from "effect/Stream";

interface QueueMessageBody {
  id: string;
  text: string;
  sentAt: number;
}

Effect.gen(function* () {
  const bucket = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
  const queueResource = yield* Queue;
  const queue = yield* Cloudflare.Queues.WriteQueue(queueResource);

  yield* Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(
    queueResource,
    (stream) =>
      Stream.runForEach(stream, (msg) =>
        bucket
          .put(`/queue/${msg.body.id}`, JSON.stringify(msg.body), {
            httpMetadata: { contentType: "application/json" },
          })
          .pipe(Effect.asVoid),
      ),
  );

Acking is automatic: if the handler succeeds, every message in the batch is ack()ed; if it fails, every message is retry()ed and Cloudflare applies the consumer’s maxRetries and retryDelay before dead-lettering. For finer control, call msg.ack() / msg.retry() per message inside the handler.

Tune batching and backoff

consumeQueueMessages(queue, props) accepts a settings object. Time fields are Duration.Input — any of these are accepted:

maxWaitTime: Duration.seconds(5)   // a Duration value
maxWaitTime: 5_000                 // a number of milliseconds
maxWaitTime: "5 seconds"           // a "<number> <unit>" string
retryDelay: "1 second"             // singular and plural units both work

import * as Duration from "effect/Duration";

  Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(queueResource, (stream) =>
  Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(
    queueResource,
    {
      batchSize: 25,
      maxRetries: 3,
      maxWaitTime: "5 seconds",
      retryDelay: Duration.seconds(30),
    },
    (stream) =>

maxWaitTime is rounded up to whole milliseconds and retryDelay to whole seconds when forwarded to Cloudflare.

Provide the runtime layer

consumeQueueMessages(...) is a Context.Service call — EventSourceLive is the layer that registers the listener with the Worker’s runtime context. Add it to the layer stack alongside the other binding lives.

}).pipe(
  Effect.provide(Cloudflare.Queues.EventSourceLive),
  Effect.provide(Cloudflare.Queues.WriteQueueBinding),
  Effect.provide(Cloudflare.R2.ReadWriteBucketBinding),
),

Without the live layer, the consumeQueueMessages call fails at deploy with Service not found: Cloudflare.Queues.EventSource.

Add the producer route

POST /queue/send enqueues a message and returns the generated id. The integ test uses the id to poll for the consumed result.

return {
  fetch: Effect.gen(function* () {
    const request = yield* HttpServerRequest;
    if (request.url === "/queue/send" && request.method === "POST") {
      const text = yield* request.text;
      const msg: QueueMessageBody = {
        id: crypto.randomUUID(),
        text,
        sentAt: Date.now(),
      };
      yield* queue.send(msg).pipe(Effect.orDie);
      return yield* HttpServerResponse.json({ sent: msg }, { status: 202 });
    }
    return HttpServerResponse.text("Not Found", { status: 404 });
  }),
};

Add the result-read route

GET /queue/result/:id reads /queue/<id> from the bucket. The consumer runs asynchronously, so the test polls this route with a short backoff until the object appears.

if (request.url.startsWith("/queue/result/") && request.method === "GET") {
  const id = request.url.split("/queue/result/")[1];
  return yield* bucket.get(`/queue/${id}`).pipe(
    Effect.flatMap((object) =>
      object === null
        ? Effect.succeed(HttpServerResponse.text("not yet", { status: 404 }))
        : object.text().pipe(
            Effect.map((body) =>
              HttpServerResponse.text(body, {
                headers: { "content-type": "application/json" },
              }),
            ),
          ),
    ),
    Effect.catchTag("R2Error", (e) =>
      Effect.succeed(HttpServerResponse.text(e.message, { status: 500 })),
    ),
  );
}

What’s the difference vs. a native `queue()` handler?

Cloudflare’s runtime delivers queue events to a queue(batch, env) export on the worker module. You can write that directly — see examples/cloudflare-worker-async for the plain async-handler shape:

export default {
  async queue(batch, env) {
    for (const msg of batch.messages) {
      await env.Bucket.put(`/queue/${msg.body.id}`, ...);
      msg.ack();
    }
  },
};

Cloudflare.Queues.consumeQueueMessages(queue, handler) is the same primitive on the Effect side — the Worker bundle’s runtime context routes the dispatch to the registered listener — but you get Effect.gen composition, typed errors, automatic batch ack/retry, and the same surface as AWS.SQS.consumeQueueMessages(queue, handler).

Receive Workflow lifecycle events

A subscription delivers platform events into a Queue. For a Workflow bound to an async Worker, pass the binding directly as the source:

const worker = yield* Cloudflare.Worker("IngestionWorker", {
  main: "./src/worker.ts",
  env: {
    INGESTION: Cloudflare.Workflow("Ingestion", {
      className: "IngestionWorkflow",
    }),
  },
});
const queue = yield* Cloudflare.Queues.Queue("WorkflowEventsQueue");

yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
  source: worker.env.INGESTION,
  events: ["instance.completed", "instance.errored"],
  queueId: queue.queueId,
});

The subscription uses the binding’s deferred physical name, including on the first deployment. Renaming the Workflow replaces the subscription. The Queue still needs a consumer to handle these messages, as above.

A WorkflowResource or a reference to a deployed Workflow is also a source:

yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
  source: yield* Cloudflare.Workflow.ref("Ingestion", {
    stack: "workflow-host",
    stage: "production",
  }),
  events: ["instance.completed", "instance.errored"],
  queueId: queue.queueId,
});

Use the Workflow’s logical ID, including any namespace. Omit the options for the current stack and stage. References read persisted state, so deploy the host first; deleting the consumer stack does not delete the referenced Workflow.

Explicit sources remain supported, including { type: "workflows.workflow", workflowName: "existing-ingestion" } for a Workflow referenced by name and { type: "r2" } for R2 events. Cloudflare allows at most one subscription per source per account.

Use resources as event sources

A subscription also accepts these resource values, or their yielded .ref(...) references:

Resource	Cloudflare source	Event scope
`Cloudflare.Images.Variant`	`images`	All Images events in the account
`Cloudflare.KV.Namespace`	`kv`	All namespace events in the account
`Cloudflare.R2.Bucket`	`r2`	All bucket events in the account
`Cloudflare.R2.SuperSlurperJob`	`superSlurper`	All migration job events in the account
`Cloudflare.Vectorize.Index`	`vectorize`	All index events in the account
`Cloudflare.AI.Model`	`workersAi.model`	Batch events for the selected model
`Cloudflare.Worker`	`workersBuilds.worker`	Builds for the selected Worker
`Cloudflare.Workflows.WorkflowResource`	`workflows.workflow`	Instances of the selected Workflow

The 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.

Event 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.

During 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.

Account-wide resource events

const cache = yield* Cloudflare.KV.Namespace("Cache");

yield* Cloudflare.Queues.Subscription("NamespaceEvents", {
  source: cache,
  events: ["namespace.created", "namespace.deleted"],
  queueId: queue.queueId,
});

This 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.

References from another stack

yield* Cloudflare.Queues.Subscription("BucketEvents", {
  source: yield* Cloudflare.R2.Bucket.ref("Uploads", {
    stack: "storage",
    stage: "production",
  }),
  events: ["bucket.created", "bucket.deleted"],
  queueId: queue.queueId,
});

All 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.

Images upload events

yield* Cloudflare.Queues.Subscription("ImageEvents", {
  source: yield* Cloudflare.Images.Variant.ref("Thumbnail"),
  events: ["image.uploaded"],
  queueId: queue.queueId,
});

The variant selects its Images account. The subscription receives uploads across that account, not only images served with Thumbnail.

Vectorize events

yield* Cloudflare.Queues.Subscription("IndexEvents", {
  source: yield* Cloudflare.Vectorize.Index.ref("Search"),
  events: ["index.created", "index.deleted"],
  queueId: queue.queueId,
});

These events describe index lifecycle changes, not individual vector writes.

Super Slurper events

yield* Cloudflare.Queues.Subscription("MigrationEvents", {
  source: yield* Cloudflare.R2.SuperSlurperJob.ref("Migration"),
  events: ["job.started", "job.completed", "job.aborted"],
  queueId: queue.queueId,
});

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.

Workers AI batch events

const model = yield* Cloudflare.AI.Model("Embeddings", {
  modelName: "@cf/baai/bge-m3",
});

yield* Cloudflare.Queues.Subscription("BatchEvents", {
  source: model,
  events: ["batch.queued", "batch.succeeded", "batch.failed"],
  queueId: queue.queueId,
});

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.

Workers Builds events

yield* Cloudflare.Queues.Subscription("BuildEvents", {
  source: yield* Cloudflare.Worker.ref("Website"),
  events: ["build.started", "build.succeeded", "build.failed"],
  queueId: queue.queueId,
});

A 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.

Where next

Workers — the runtime that produces and consumes queue messages.
Durable Objects — for stateful coordination instead of fire-and-forget work.
Workflows — multi-step orchestration with checkpointed steps when fire-and-forget isn’t enough.

Reference: