Queues

A queue lets a Worker say “do this later” and answer now. You send a message; a moment later, a consumer gets it in a batch and does the slow part. In forge-effect the queue is typed by an Effect Schema, end to end:

export const Signup = Schema.Struct({ email: Schema.String, plan: Schema.Literals(['free', 'pro']) });

export class Signups extends Queue.Make<Signups>()('signups', { schema: Signup }) {}

yield* (yield* Signups).send({ email, plan }); // checked against Signup, then sent

A message that isn’t a Signup never leaves your Worker, and one that arrives in a shape your consumer can’t decode never reaches its handler.

We’ll build the sign-up app from example/signups.ts: a Worker that takes sign-ups over HTTP, queues them, and consumes the queue, writing each one to D1.

Step 1: Declare the queue

A queue is a class with a schema, like any other resource:

import { Effect, Schema } from 'effect';

import { App, D1, Queue, Worker } from '@forge-effect/infra';

export const Signup = Schema.Struct({ email: Schema.String, plan: Schema.Literals(['free', 'pro']) });

export class Db extends D1.Database<Db>()('main', { migrations: ['CREATE TABLE signups (email TEXT, plan TEXT)'] }) {}
export class Signups extends Queue.Make<Signups>()('signups', { schema: Signup }) {}
export class Failed extends Queue.Make<Failed>()('signups-failed', { schema: Schema.Unknown }) {}

Two queues: signups for the work, and signups-failed for messages that couldn’t be handled however many times we tried. That one’s schema is Schema.Unknown, since what lands there is, by definition, whatever went wrong.

Step 2: Send

The fetch handler takes a sign-up and queues it:

const fetch = Effect.gen(function* () {
  const request = yield* Worker.Request;
  if (request.method !== 'POST') {
    return new Response('POST a sign-up\n', { status: 405 });
  }
  const body = yield* Effect.tryPromise(() => request.json());
  // A body that isn't a sign-up fails to encode, and answers 500.
  yield* (yield* Signups).send(body as typeof Signup.Type);
  return new Response('queued\n', { status: 202 });
});

send encodes the message with the queue’s schema, then sends it as JSON. If it doesn’t fit (a plan of "gold", say), send fails with a SchemaError and nothing is queued. Here that failure goes unhandled, so the Worker answers 500; in your app you’d likely turn it into a 400. send can also fail with a QueueError, if Cloudflare refuses it.

Two more ways to send:

const emails = yield* Emails;
// Delivered in a minute, not now.
yield* emails.send({ to: 'ada@example.com', subject: 'Welcome!' }, { delaySeconds: 60 });
// Several at once, in one call.
yield* emails.sendBatch([
  { to: 'ada@example.com', subject: 'Tip 1' },
  { to: 'ada@example.com', subject: 'Tip 2' },
]);

Step 3: Consume

A consumer is a handler for one message, already decoded:

const record = (signup: typeof Signup.Type) =>
  Effect.gen(function* () {
    const db = yield* Db;
    yield* db.run('INSERT INTO signups (email, plan) VALUES (?, ?)', signup.email, signup.plan);
  });

export const api = Worker.make('api', {
  fetch,
  queues: [Signups.consume(record, { batchSize: 10, maxRetries: 3, deadLetter: Failed })],
}).bind(Signups, Db);

export default App.make('signups', { workers: [api] });

Signups.consume(record, options) declares the consumer, and the Worker’s queues lists the queues it consumes. Look at what it binds: Signups, because fetch sends to it, and Db, because record writes to it. The compiler checks both, as ever. The queue it consumes and its dead-letter queue come with the consumer, so infra graph shows all three:

$ pnpm infra graph example/signups.ts
signups-api (fetch, queue)
  SIGNUPS: queue signups-signups
  MAIN: d1 signups-main
  SIGNUPS_FAILED: queue signups-signups-failed

What happens to each message

It’s decoded with the queue’s schema.
Your handler runs with it.
The handler succeeds: the message is acknowledged, and gone.
The handler fails, or the message didn’t decode: it’s retried (and the failure logged, as message <id> failed, attempt <n>).
Out of retries, it goes to the dead-letter queue, if you named one.

So a bad message doesn’t take its batch down with it: the others are acknowledged, and only the bad one comes back.

The handler gets a second argument too, the message’s metadata: its id, its timestamp, and how many attempts it’s had, this one included.

const deliver = (email: typeof Email.Type, meta: Queue.Meta) =>
  Effect.log(`sending ${email.subject} to ${email.to}, attempt ${meta.attempts}`);

The knobs

Emails.consume(deliver, {
  batchSize: 50, // up to 50 messages a batch
  maxWaitSeconds: 5, // or whatever has arrived after 5 seconds
  concurrency: 10, // ten of a batch at a time
  maxRetries: 5,
  retryDelay: 30, // a failed message waits 30 seconds before its next try
  deadLetter: Bounced,
}),

Option	What it does
`batchSize`	The most messages a batch holds.
`maxWaitSeconds`	How long a batch waits to fill.
`maxRetries`	How many times a message is retried before it’s dead-lettered.
`retryDelay`	How long a retried message waits, in seconds.
`deadLetter`	The queue a message goes to once it’s out of retries.
`concurrency`	How many messages of a batch your handler works on at once. One by default, so a batch goes in order.
`maxConcurrency`	How many batches Cloudflare runs at once. Unset, Cloudflare scales up to its maximum.

All but concurrency are Cloudflare’s consumer settings, by Cloudflare’s names. concurrency is forge-effect’s: it’s how many of a batch’s messages Effect runs at the same time.

Step 4: Test it

Two halves to test: what your Worker sends, and what its consumer does. Queue.memory(Signups) gives you a Layer and a list of everything sent. Here’s the real test, from test/queue.test.ts:

test('a send is encoded with the queue\'s schema; one that isn\'t a message fails, and answers 500', async () => {
  const queue = Queue.memory(Signups);
  const fakes = Layer.mergeAll(queue.layer, sqlite(Db));
  const post = (body: unknown) =>
    Worker.handle(api, new Request('https://x.dev/', { method: 'POST', body: JSON.stringify(body) })).pipe(Effect.provide(fakes));
  assert.equal((await Effect.runPromise(post({ email: 'a@b.dev', plan: 'pro' }))).status, 202);
  assert.equal((await Effect.runPromise(post({ email: 'a@b.dev', plan: 'gold' }))).status, 500);
  assert.deepEqual(queue.sent, [{ email: 'a@b.dev', plan: 'pro' }]); // the gold plan never left
});

For the consumer, messages(bodies) from @forge-effect/infra/testing makes a batch as workerd delivers one, and keeps track of what was acknowledged and retried. Worker.runQueue runs the Worker’s consumer of a queue (by its name in the app) on it:

import assert from 'node:assert/strict';
import { test } from 'node:test';

import { Effect, Layer } from 'effect';

import { Queue, Worker } from '@forge-effect/infra';
import { messages, sqlite } from '@forge-effect/infra/testing';
import { api, Db, Signups } from '../example/signups.ts';

test('a sign-up is recorded; one that does not decode is retried', async () => {
  const batch = messages([{ email: 'ada@example.com', plan: 'pro' }, { email: 'bob@example.com', plan: 'gold' }]);
  await Effect.runPromise(
    Worker.runQueue(api, 'signups', batch.messages).pipe(Effect.provide(Layer.mergeAll(Queue.memory(Signups).layer, sqlite(Db)))),
  );
  assert.deepEqual(batch.acked, ['m0']);
  assert.deepEqual(batch.retried, ['m1']); // gold isn't a plan
});

A neat trick: feed one Worker’s queue.sent to another’s consumer, with messages(queue.sent), and you’ve tested the whole trip, producer to consumer, in memory. The media example’s test does exactly that, across three apps.

Prefer the real runtime? infra dev runs your queues and their consumers locally in workerd. See Local dev.

Step 5: Ship it

Before you ship, infra check looks at your queues too:

$ pnpm infra check example/signups.ts
info queue-no-consumer  example/signups.ts:32:32
  the queue signups-failed is a dead-letter queue no Worker consumes: its messages wait there to be looked at

✓ signups: no problems found (app, types, bundle)

That’s only information: a dead-letter queue nobody consumes is often what you want, a place where failures wait for a human. (A queue with senders and no consumer at all is a warning: its messages would wait until they expire.)

When you ship, here’s what happens:

The queues are made under your app’s names: signups-signups and signups-signups-failed (and signups-pr-7-signups in a preview).
The Worker becomes the queue’s consumer once it’s live, with its settings (batch size, retries, dead-letter queue).
If a Worker stops consuming a queue, the next ship removes it as that queue’s consumer.
Teardown goes in the order Cloudflare needs: consumers first, then Workers, then queues. (Cloudflare won’t delete a Worker that consumes a queue, or a queue a Worker binds. A live teardown taught us that.)

This was checked live: a POST of a sign-up answered 202, and the consumer wrote it about a minute later.

Queues across apps

One app can own a queue, and another consume it. In the media example, media announces each upload on its queue uploaded, and catalog indexes them:

const index = Uploaded.from('media').consume(({ key, size, type, at }) =>
  Effect.gen(function* () {
    const file = yield* (yield* Uploads).get(key);
    // Deleted since it was announced: nothing to index.
    if (file._tag === 'None') {
      return;
    }
    // … fingerprint it and write it to the index
  }),
);

export default Worker.make('indexer', { fetch, queues: [index] }).bind(Index, Uploads.from('media'));

Both apps import Uploaded from one shared module, so they agree on the schema. Two checks keep it that way, before a ship rather than in the middle of one:

queue-schema-mismatch: each ship of the owner records the queue’s schema. If yours no longer fits it (you’d send messages its consumer won’t decode, or consume ones you won’t decode), your plan stops. Bring your copy in line, or ship the owner first.
queue-consumer-taken: Cloudflare allows one consumer per queue. If another Worker consumes it already (one of another app, or one of yours you stopped declaring, kept until prune), your plan names that Worker and its app, and stops. Within one app, two Workers consuming the same queue don’t even load.

How it works, briefly

Messages go over the wire as JSON, in the schema’s JSON form (Schema.toCodecJson), the same form calls between Workers use.
In your account, the queue is signups-signups, and that’s the name workerd gives a batch. The entry generated for your Worker maps it back to signups, so your code only ever sees the name you declared.
Each message of a batch is decoded and handled on its own, and acknowledged or retried on its own. That’s why one bad message costs only itself a retry.

Good to know

Any schema travels. A message goes in its schema’s JSON form, so a Schema.Date or a Schema.BigInt field leaves as a string and arrives as a Date or a bigint again. Queue.memory keeps what’s sent as JSON too, so a test sees what the consumer will.
Cloudflare delivers a message at least once. Retries mean your handler may see the same message twice. Write it so that’s harmless: the media indexer’s INSERT … ON CONFLICT DO UPDATE is one way.
A queue can stay in a jurisdiction, such as the EU: Queue.Make<Audit>()('audit', { schema, jurisdiction: 'eu' }). See where your data lives.
An R2 bucket can send to a queue when its objects change, with R2.Notification as the queue’s schema. That’s in More bindings.

Workflows: when the work is several steps, with sleeps and waits between them.
Durable Objects: state with an address.