Cloudflare.Queues reference

Consumer

Source: src/Cloudflare/Queues/Consumer.ts

A Cloudflare Queue Consumer that processes messages from a Queue.

Register a Worker as a consumer of a Queue. The Worker’s queue() handler will be invoked with batches of messages.

Cloudflare allows at most one Worker consumer per queue (HTTP-pull consumers can coexist). The reconciler enforces this: if the queue already has a Worker consumer pointing at a different logical Worker’s script, the deploy fails with a clear error rather than silently adopting it. A stranded consumer from a prior generation of the same Worker (identified by the scripts’ ownership tags) is rebuilt in place.

Consumer: Registering a Consumer

Basic consumer

const queue = yield* Cloudflare.Queues.Queue("MyQueue");
const worker = yield* Cloudflare.Worker("Worker", { ... });

yield* Cloudflare.Queues.Consumer("MyConsumer", {
  queueId: queue.queueId,
  scriptName: worker.workerName,
});

Consumer with settings

yield* Cloudflare.Queues.Consumer("MyConsumer", {
  queueId: queue.queueId,
  scriptName: worker.workerName,
  settings: {
    batchSize: 50,
    maxRetries: 5,
    maxWaitTimeMs: 10000,
  },
});

Queue

Source: src/Cloudflare/Queues/Queue.ts

A Cloudflare Queue for reliable message passing between Workers.

Queues enable you to send and receive messages with guaranteed delivery. Create a queue as a resource, then bind it to a Worker to send messages at runtime. Register a consumer to process messages.

Queue: Creating a Queue

Basic queue

const queue = yield* Cloudflare.Queues.Queue("MyQueue");

Queue with explicit name

const queue = yield* Cloudflare.Queues.Queue("MyQueue", {
  name: "my-app-queue",
});

Queue: Binding to a Worker

In an Effect-style Worker, use Cloudflare.Queues.WriteQueue in the init phase and provide Cloudflare.Queues.WriteQueueBinding in the runtime layer. The returned WriteQueueClient exposes send and sendBatch.

import * as Cloudflare from "alchemy/Cloudflare";
import * as Effect from "effect/Effect";
import { HttpServerRequest } from "effect/http/HttpServerRequest";
import * as HttpServerResponse from "effect/http/HttpServerResponse";

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

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

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

QueueSink

Source: src/Cloudflare/Queues/QueueSink.ts

Binding service that exposes a Queue as an Effect Sink: run a Stream of message bodies into it and every element is enqueued.

Each stream chunk becomes one sendBatch call. Chunks larger than a Cloudflare batch (100 messages / 256 KB) are split into consecutive batches, preserving order. Bodies are any JSON-serializable value, the same as WriteQueueClient.send.

Provide QueueSinkBinding (native Worker binding), QueueSinkHttp (scoped HTTP token) or QueueSinkLocal (current credentials, for Actions). Each one is the sink layered over the matching WriteQueue implementation and wires the queue into the host exactly as that layer does.

QueueSink: Draining a Stream into a Queue

Run a Stream into a Queue

export default Cloudflare.Worker(
  "Api",
  { main: import.meta.url },
  Effect.gen(function* () {
    const sink = yield* Cloudflare.Queues.QueueSink(Clicks);

    return {
      fetch: Effect.gen(function* () {
        yield* Stream.fromIterable(events).pipe(Stream.run(sink));
        return HttpServerResponse.empty({ status: 202 });
      }),
    };
  }).pipe(Effect.provide(Cloudflare.Queues.QueueSinkBinding)),
);

Controlling batch size

// one sendBatch call per 25 messages
yield* Stream.fromIterable(events).pipe(
  Stream.rechunk(25),
  Stream.run(sink),
);

QueueSink: Source → Transform → Sink

Effect.gen(function* () {
  const enriched = yield* Cloudflare.Queues.QueueSink(EnrichedClicks);

  yield* Cloudflare.Queues.consumeQueueMessages<Click>(Clicks, (messages) =>
    messages.pipe(
      Stream.map((message) => message.body),
      Stream.filter((click) => click.button === "buy"),
      Stream.map((click) => ({ ...click, receivedAt: Date.now() })),
      Stream.run(enriched),
    ),
  );
}).pipe(
  Effect.provide(
    Layer.mergeAll(
      Cloudflare.Queues.EventSourceLive,
      Cloudflare.Queues.QueueSinkBinding,
    ),
  ),
);

QueueSinkBinding

Source: src/Cloudflare/Queues/QueueSinkBinding.ts Kind: Layer · Provides: Cloudflare.Queues.QueueSink

Implementation of the QueueSink service over the native Worker queue binding (WriteQueueBinding). Registers the same queue binding as WriteQueue, so a Worker can use both on one queue.

Works under alchemy dev: a local Worker’s binding targets the local queue broker.

QueueSinkBinding: Providing the Layer

export default Cloudflare.Worker(
  "Api",
  { main: import.meta.url },
  Effect.gen(function* () {
    const sink = yield* Cloudflare.Queues.QueueSink(Clicks);
    return {
      fetch: Effect.gen(function* () {
        yield* Stream.fromIterable(events).pipe(Stream.run(sink));
        return HttpServerResponse.empty({ status: 202 });
      }),
    };
  }).pipe(Effect.provide(Cloudflare.Queues.QueueSinkBinding)),
);

QueueSinkHttp

Source: src/Cloudflare/Queues/QueueSinkHttp.ts Kind: Layer · Provides: Cloudflare.Queues.QueueSink

Implementation of the QueueSink service over the Queues bulk-push HTTP API (WriteQueueHttp), authenticated with a scoped Queues Write API token bound into the host.

QueueSinkHttp: Providing the Layer

Effect.gen(function* () {
  const sink = yield* Cloudflare.Queues.QueueSink(Clicks);
  // ...
}).pipe(Effect.provide(Cloudflare.Queues.QueueSinkHttp));

QueueSinkLocal

Source: src/Cloudflare/Queues/QueueSinkLocal.ts Kind: Layer · Provides: Cloudflare.Queues.QueueSink

Implementation of the QueueSink service that pushes over the Queues bulk-push HTTP API with the current credentials (WriteQueueLocal) — for draining a Stream into a queue from an Action or other deploy-time Effect, with no Worker host.

QueueSinkLocal: Providing the Layer

const Seed = Alchemy.Action(
  "Seed",
  Effect.gen(function* () {
    const sink = yield* Cloudflare.Queues.QueueSink(queue);
    return Effect.fn(function* () {
      yield* Stream.range(1, 500).pipe(
        Stream.map((id) => ({ event: "seed", id })),
        Stream.run(sink),
      );
    });
  }).pipe(Effect.provide(Cloudflare.Queues.QueueSinkLocal)),
);

Subscription

Source: src/Cloudflare/Queues/Subscription.ts

A Cloudflare Queues event subscription — delivers platform events (R2 bucket events, KV namespace events, Workers Builds, Workflows, etc.) into a Queue as messages.

The source selects which product emits the events and is fixed at creation (changing it replaces the subscription). name, events, enabled, and the destination queueId are all mutable in place. Cloudflare allows at most one subscription per source per account.

Subscription: Creating a Subscription

R2 bucket events into a Queue

const queue = yield* Cloudflare.Queues.Queue("EventsQueue");

const subscription = yield* Cloudflare.Queues.Subscription("R2Events", {
  source: { type: "r2" },
  events: ["bucket.created", "bucket.deleted"],
  queueId: queue.queueId,
});

KV namespace events with an explicit name

const subscription = yield* Cloudflare.Queues.Subscription("KvEvents", {
  name: "kv-events",
  source: { type: "kv" },
  events: ["namespace.created"],
  queueId: queue.queueId,
});

Workers Builds events for one Worker

const subscription = yield* Cloudflare.Queues.Subscription("BuildEvents", {
  source: { type: "workersBuilds.worker", workerName: "my-worker" },
  events: ["build.started", "build.succeeded"],
  queueId: queue.queueId,
});

Workflow lifecycle events, from a Workflow bound in this stack

Pass the Workflow binding from the host Worker’s env directly. Its physical name remains an Output, so the subscription works on the Workflow’s first deployment and follows later renames. Only the source type and physical workflow name are persisted, not the binding metadata.

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

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

Workflow lifecycle events from a persisted resource reference

Use the logical resource ID, including any namespace. References read persisted state, so deploy the host first. Omitting the options uses the current stack and stage; the reference does not take ownership of the host.

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

Workflow lifecycle events by an existing physical name

const subscription = yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
  source: { type: "workflows.workflow", workflowName: "existing-ingestion" },
  events: ["instance.completed", "instance.errored"],
  queueId: queue.queueId,
});

Bounces and complaints from an Email Sending domain

const subscription = yield* Cloudflare.Queues.Subscription("MailEvents", {
  source: { type: "email.sending", zoneId: zone.zoneId, domain: "mail.example.com" },
  events: ["message.bounced", "message.complained"],
  queueId: queue.queueId,
});

Subscription: Resource and Reference Sources

Account-wide KV events from a namespace

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

KV, R2, Images, Vectorize, and Super Slurper sources are account-wide. Passing a resource retains its account and deployment dependency, but does not filter events to that resource. The subscription can miss the source’s initial creation or final deletion because it depends on that source. Use an explicit product descriptor when the subscription must exist first.

R2 bucket reference 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 resource forms accept yielded .ref references with optional stack and stage selectors. The referenced resource must already be deployed in the same Cloudflare account. Removing a subscription does not remove its referenced resource.

Images upload events from a variant reference

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

The variant selects its Images account; uploads are not filtered by variant.

Vectorize index events

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

Super Slurper migration events

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

This selects all migration jobs in the account, not one job’s objects.

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,
});

The model is a non-owning catalog handle. These events require asynchronous batch inference; ordinary synchronous inference does not emit them.

Workers Builds events from a Worker reference

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

The Worker must have a Workers Builds integration to emit build events. An ordinary Alchemy Worker upload is not a Workers Builds run.

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.

Subscription: Pausing delivery

const subscription = yield* Cloudflare.Queues.Subscription("R2Events", {
  source: { type: "r2" },
  events: ["bucket.created"],
  queueId: queue.queueId,
  enabled: false,
});

WriteQueue

Source: src/Cloudflare/Queues/WriteQueue.ts

Binding service that turns a Queue resource into a typed WriteQueueClient you can call from a Worker’s runtime Effect.

The Cloudflare Worker queue binding is producer-only — send for a single message and sendBatch for many in one call. Messages can be any JSON-serializable value.

WriteQueue: Sending Messages

Producer route

const queue = yield* Cloudflare.Queues.WriteQueue(Queue);

return {
  fetch: Effect.gen(function* () {
    yield* queue.send({ text: "hi", sentAt: Date.now() });
    return HttpServerResponse.empty({ status: 202 });
  }),
};

Sending a batch

yield* queue.sendBatch([
  { body: { event: "click", id: 1 } },
  { body: { event: "click", id: 2 } },
  { body: "raw text", contentType: "text" },
]);

Provide WriteQueueBinding (native Worker binding) or WriteQueueHttp (scoped HTTP token) in the worker’s runtime layer to resolve the underlying queue at request time.

WriteQueue is a single identifier that is simultaneously the binding’s Context tag, its type, and the callable — yield* Cloudflare.Queues.WriteQueue(queue).