Configuring fanout and retry in a single file with Cloudflare Workers + D1 + Queues

If you've ever tried to attach async work to a serverless function, you've probably hit this wall. You need to receive an HTTP request and return a response immediately, but behind the scenes you have to send an email, fire webhooks to multiple endpoints, or post-process a file. On a traditional server you'd spin up a background thread or a queue worker, but in serverless the execution context disappears the moment the request ends.

The Cloudflare Workers + Queues + D1 combination solves this problem quite elegantly. A single Worker file holds both the HTTP handler (producer) and the queue consumption handler (consumer), and the runtime executes the two roles independently. You keep a single-repository codebase while implementing async processing, fanout, retries, and a Dead Letter Queue all in one stack. Over 2024–2025, Queues was completely rebuilt on top of Durable Objects, dramatically improving throughput and concurrent consumer scalability, and it's now mature enough to use in production.

This article walks through real code covering everything from wrangler.toml configuration to fanout pattern implementation, retry logic and DLQ wiring, and job state tracking and idempotency with D1.

How the Three Services Fit Together

Let's start with the role each service plays.

Cloudflare Workers is a serverless function platform based on V8 Isolates. It runs at PoPs worldwide, and new Isolate initialization takes only a few milliseconds, making Cold Start overhead essentially negligible. There is, however, a CPU time ceiling — per the Workers limits documentation, the CPU time limit on the paid (Standard) plan is 30 seconds per request. If your design tries to handle a heavy 5-minute job in one shot, you're already past that boundary.

Cloudflare Queues is a managed message queue natively integrated with Workers. It guarantees At-Least-Once delivery and provides batching, retries, and a Dead Letter Queue out of the box. Consumer instances scale horizontally to a considerable degree as the queue backlog grows, though keep in mind that the upper limit is a soft limit that varies by plan and queue configuration.

Cloudflare D1 is a serverless SQLite database built into Workers. It connects directly via bindings with no network hops and supports ACID transactions. It's a great fit for tracking job state and ensuring idempotency.

Cloudflare's Workers Best Practices emphasizes the principle of offloading heavy work from the request path to Queues or Workflows, and highlights that bindings like D1 and Queues are referenced in-process with no network hops.

A 202 is returned the moment the request arrives, and the actual heavy work is handled asynchronously through the queue. The client receives a jobId and can poll the status separately. The ordering of the D1 write and the queue publish will be revisited later.

Placing Producer and Consumer Together in a Single File

The first step is declaring the bindings in wrangler.toml. Note that the code in this article assumes @cloudflare/workers-types is installed. If you're using only the standard lib.dom.d.ts, the generic syntax on Request.json() and global types like Queue and D1Database won't resolve correctly.

name = "my-worker"
main = "src/worker.ts"
compatibility_date = "2024-09-23"
 
[[queues.producers]]
binding = "MY_QUEUE"
queue = "my-job-queue"
 
[[queues.consumers]]
queue = "my-job-queue"
max_batch_size = 10
max_batch_timeout = 5
max_retries = 3
dead_letter_queue = "my-job-dlq"
 
[[queues.consumers]]
queue = "my-job-dlq"
max_batch_size = 5
max_batch_timeout = 30
 
[[d1_databases]]
binding = "DB"
database_name = "jobs-db"
database_id = "<your-database-id>"

max_batch_size = 10 delivers up to 10 messages at once to the consumer in a batch, and max_batch_timeout = 5 flushes the batch after 5 seconds even if it has fewer than 10 messages. This is useful for reducing D1 write round-trips and optimizing cost.

Next, create the D1 schema. The webhook_deliveries table used later is included here as well.

-- migrations/0001_create_jobs.sql
CREATE TABLE IF NOT EXISTS jobs (
  id TEXT PRIMARY KEY,
  type TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending',
  payload TEXT,
  error TEXT,
  created_at INTEGER NOT NULL DEFAULT (unixepoch()),
  updated_at INTEGER NOT NULL DEFAULT (unixepoch())
);
 
CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status);
CREATE INDEX IF NOT EXISTS idx_jobs_status_updated ON jobs(status, updated_at);
 
CREATE TABLE IF NOT EXISTS webhook_deliveries (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  endpoint_id TEXT NOT NULL,
  event_type TEXT NOT NULL,
  status TEXT NOT NULL,
  error TEXT,
  created_at INTEGER NOT NULL DEFAULT (unixepoch())
);

Now the Worker file. The producer and consumer live in a single file, the main queue and DLQ are branched by queue name, and idempotency handling is integrated into the main flow.

// src/worker.ts
export interface Env {
  MY_QUEUE: Queue<JobMessage>;
  DB: D1Database;
}
 
export interface JobMessage {
  id: string;
  type: 'send_email' | 'webhook_fanout' | 'process_file';
  payload: unknown;
}
 
export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    if (req.method !== 'POST') {
      return new Response('Method Not Allowed', { status: 405 });
    }
 
    const body = (await req.json()) as {
      type: JobMessage['type'];
      payload: unknown;
    };
    const job: JobMessage = {
      id: crypto.randomUUID(),
      type: body.type,
      payload: body.payload,
    };
 
    // Write to D1 first — D1 and Queues cannot be wrapped in a single transaction
    await env.DB.prepare(
      'INSERT INTO jobs (id, type, status, payload) VALUES (?, ?, ?, ?)'
    )
      .bind(job.id, job.type, 'pending', JSON.stringify(job.payload))
      .run();
 
    await env.MY_QUEUE.send(job);
 
    return Response.json({ jobId: job.id }, { status: 202 });
  },
 
  async queue(
    batch: MessageBatch<JobMessage>,
    env: Env
  ): Promise<void> {
    if (batch.queue === 'my-job-dlq') {
      await handleDlqBatch(batch, env);
      return;
    }
    await handleMainBatch(batch, env);
  },
};

I was initially unsure whether having both fetch and queue handlers in the same file was valid. The Cloudflare runtime calls fetch for HTTP requests and queue for queued messages, each independently. The codebase is unified, but the runtime executes the two roles separately. The batch.queue property tells you which queue the batch came from.

Consumer Body: Handling Idempotency and Status Update Failures

The main consumer must satisfy three things simultaneously: (1) the same message arriving twice must not be processed twice, (2) a failure in the job itself must be distinguished from a failure to record the status, and (3) retries must be delayed to avoid a Retry Storm.

async function handleMainBatch(
  batch: MessageBatch<JobMessage>,
  env: Env
): Promise<void> {
  for (const message of batch.messages) {
    const job = message.body;
 
    // Optimistic lock: transition to processing only when status is pending
    const claim = await env.DB.prepare(
      `UPDATE jobs
         SET status = 'processing', updated_at = unixepoch()
       WHERE id = ? AND status = 'pending'`
    )
      .bind(job.id)
      .run();
 
    // Another instance already claimed it or it's already done — skip to prevent duplicate processing
    if (claim.meta.changes === 0) {
      message.ack();
      continue;
    }
 
    let processed = false;
    try {
      await processJob(job, env);
      processed = true;
    } catch (err) {
      const errorMsg = err instanceof Error ? err.message : String(err);
      await safeUpdateStatus(env, job.id, 'pending', errorMsg);
      // Exponential backoff retry — attempts is provided in message metadata
      message.retry({ delaySeconds: backoffSeconds(message.attempts) });
      continue;
    }
 
    try {
      await env.DB.prepare(
        `UPDATE jobs
            SET status = 'done', error = NULL, updated_at = unixepoch()
          WHERE id = ?`
      )
        .bind(job.id)
        .run();
      message.ack();
    } catch (dbErr) {
      // Job succeeded but only the status write failed — ack the message to prevent re-execution;
      // state recovery is handled by the compensation scan below
      console.error('status update failed after success', job.id, dbErr);
      message.ack();
    }
 
    void processed;
  }
}
 
async function safeUpdateStatus(
  env: Env,
  id: string,
  status: string,
  error: string | null
): Promise<void> {
  try {
    await env.DB.prepare(
      `UPDATE jobs
          SET status = ?, error = ?, updated_at = unixepoch()
        WHERE id = ?`
    )
      .bind(status, error, id)
      .run();
  } catch (e) {
    console.error('status update failed', id, e);
  }
}
 
function backoffSeconds(attempts: number): number {
  // 10s, 20s, 40s ... max 300s
  return Math.min(300, 10 * 2 ** Math.max(0, attempts - 1));
}
 
async function processJob(job: JobMessage, env: Env): Promise<void> {
  switch (job.type) {
    case 'send_email':
      await sendEmail(job.payload, env);
      break;
    case 'webhook_fanout':
      await fanoutWebhooks(job.payload as WebhookFanoutPayload, env);
      break;
    case 'process_file':
      await processFile(job.payload, env);
      break;
    default:
      throw new Error(`Unknown job type: ${(job as JobMessage).type}`);
  }
}

Two things differ from an initial draft here.

First, the optimistic lock (WHERE id = ? AND status = 'pending') is placed in the main flow. Since UPDATE executes atomically, changes === 0 means another consumer instance already claimed it or it's already complete, so we simply ack(). This is the minimum defense against duplicate processing in an At-Least-Once environment.

Second, a D1 UPDATE failure after processJob succeeds is not handled with retry(). If a retry fires when the job has already affected the outside world (e.g., an email was sent), it causes duplicate execution. In that case, ack() the message and leave state recovery to the compensation logic below.

Applying exponential backoff via message.retry({ delaySeconds: ... }) is also something you must handle in production. If an external API is down and you retry immediately in a tight loop, you risk hammering a system that's trying to recover — a classic Retry Storm.

Fanout Pattern: Delivering One Event to Multiple Destinations Simultaneously

The word "fanout" is used in two contexts. One is platform-level fanout — when many messages pile up in a queue, the Cloudflare runtime automatically scales consumer Worker instances horizontally for parallel processing. The other is application-level fanout — delivering a single event to multiple endpoints concurrently from within a consumer.

Here's a structure that simultaneously delivers a payment-completed event to Slack, an internal service, and a third-party hook.

interface WebhookEndpoint {
  id: string;
  url: string;
}
 
interface WebhookFanoutPayload {
  eventType: string;
  eventData: unknown;
  endpoints: WebhookEndpoint[];
}
 
async function fanoutWebhooks(
  payload: WebhookFanoutPayload,
  env: Env
): Promise<void> {
  // Promise.allSettled: if one fails, the rest continue sending
  const results = await Promise.allSettled(
    payload.endpoints.map((endpoint) =>
      sendWebhook(endpoint, payload.eventType, payload.eventData)
    )
  );
 
  // D1 batch API: process multiple INSERTs in a single round-trip
  const stmt = env.DB.prepare(
    'INSERT INTO webhook_deliveries (endpoint_id, event_type, status, error) VALUES (?, ?, ?, ?)'
  );
 
  await env.DB.batch(
    results.map((result, i) =>
      stmt.bind(
        payload.endpoints[i].id,
        payload.eventType,
        result.status === 'fulfilled' ? 'delivered' : 'failed',
        result.status === 'rejected' ? String(result.reason) : null
      )
    )
  );
}
 
async function sendWebhook(
  endpoint: WebhookEndpoint,
  eventType: string,
  data: unknown
): Promise<void> {
  const res = await fetch(endpoint.url, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ event: eventType, data, timestamp: Date.now() }),
  });
 
  if (!res.ok) {
    throw new Error(`Webhook failed: ${res.status} ${res.statusText}`);
  }
}

The reason for using Promise.allSettled matters. With Promise.all, a single failure causes the entire fanout to fail, triggering a message retry. That can result in duplicate deliveries to endpoints that already succeeded. Using allSettled and recording per-endpoint results in D1 lets you build a flow that identifies only the failed endpoints and pushes them into a separate retry job.

The D1 batch() API handling multiple INSERTs in a single round-trip is also worth noting. In D1, each write round-trip is both latency and cost, so this optimization becomes increasingly noticeable as fanout scale grows.

Retries and DLQ: Turning Failures into Data

With max_retries = 3, every time an exception is thrown in the consumer or message.retry() is called, the retry count is decremented. Once exhausted, the message moves to the separate queue specified in dead_letter_queue.

The my-job-dlq consumer was already declared in wrangler.toml earlier, and the top-level queue handler branches on batch.queue === 'my-job-dlq'. DLQ batch processing continues as follows.

async function handleDlqBatch(
  batch: MessageBatch<JobMessage>,
  env: Env
): Promise<void> {
  for (const message of batch.messages) {
    await env.DB.prepare(
      `UPDATE jobs
          SET status = 'dead', updated_at = unixepoch()
        WHERE id = ?`
    )
      .bind(message.body.id)
      .run();
 
    await notifyOpsTeam(message.body); // Conceptual example: Slack, PagerDuty, etc.
 
    // DLQ messages must also be acked — skipping this causes reprocessing loops
    message.ack();
  }
}

Honestly, forgetting to ack() in the DLQ is a mistake you'll make at least once. The DLQ is still a queue, so if you don't consume its messages they keep accumulating, and the DLQ consumer itself can fall into a retry loop.

Two Fragile Points: Compensation Transactions and Stuck Processing

The code so far still has two holes. If you don't plug them, you will encounter them in production.

One. A job where the D1 INSERT succeeded but the queue publish failed. There's no way to atomically combine D1 and Queues in the producer, so it's possible to end up with a pending entry in the jobs table but no corresponding message in the queue.

Two. A job whose consumer crashed after being moved to processing. If the instance dies after the optimistic lock transitions the job to processing, no one picks it up again. A processing record with a stale updated_at lingers like a zombie.

Both cases are resolved with a compensation sweep that uses a Cron Trigger to periodically scan and re-enqueue.

# Add to wrangler.toml
[triggers]
crons = ["*/5 * * * *"]

// Add a scheduled handler to the same Worker file
export default {
  async fetch(/* ... */) { /* same as above */ },
  async queue(/* ... */) { /* same as above */ },
 
  async scheduled(_event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
    ctx.waitUntil(recoverStuckJobs(env));
  },
};
 
async function recoverStuckJobs(env: Env): Promise<void> {
  // 1) pending jobs where the queue publish was lost — pending for more than 2 minutes since creation
  const pendingStuck = await env.DB.prepare(
    `SELECT id, type, payload FROM jobs
      WHERE status = 'pending'
        AND created_at < unixepoch() - 120
      LIMIT 100`
  ).all<{ id: string; type: JobMessage['type']; payload: string }>();
 
  // 2) Jobs stuck in processing — reset to pending if not updated in over 10 minutes
  await env.DB.prepare(
    `UPDATE jobs
        SET status = 'pending', updated_at = unixepoch()
      WHERE status = 'processing'
        AND updated_at < unixepoch() - 600`
  ).run();
 
  const stuckProcessing = await env.DB.prepare(
    `SELECT id, type, payload FROM jobs
      WHERE status = 'pending'
        AND updated_at >= unixepoch() - 30
      LIMIT 100`
  ).all<{ id: string; type: JobMessage['type']; payload: string }>();
 
  const candidates = [
    ...(pendingStuck.results ?? []),
    ...(stuckProcessing.results ?? []),
  ];
 
  for (const row of candidates) {
    await env.MY_QUEUE.send({
      id: row.id,
      type: row.type,
      payload: JSON.parse(row.payload),
    });
  }
}

The same job might end up in the queue twice, but the consumer's optimistic lock will immediately discard the second execution. The thresholds (2 minutes, 10 minutes) can be tuned to match your service SLA.

Trade-offs: When This Stack Fits and When It Doesn't

Where it works well

Where it falls short

Cloudflare Workflows, now generally available since 2025, is also worth mentioning. Multi-step jobs that need to maintain state between steps or retry only a specific failed step are a better fit for Workflows than Queues. A simple rule of thumb: use Queues for fanout, buffering, and event consumption; use Workflows for multi-step dependent tasks.

Closing Thoughts

Keeping fetch and queue in a single file isn't just an aesthetic preference — the real benefit of this stack is that the job schema, handlers, and compensation logic are bundled in one repository and one deployment unit, meaningfully reducing operational overhead. That said, because everything runs on At-Least-Once delivery, optimistic locking and a Cron-based compensation scan should be treated as fundamentals, not optional extras. If your jobs are single-step and the state graph is simple, use this combination as-is; when inter-step dependencies and partial retries become necessary, that's the moment to migrate to Workflows.

References

• Cloudflare Workers Limits
• Cloudflare Queues — Batching, Retries and Delays
• Cloudflare Queues — Dead Letter Queues
• Cloudflare Queues — How Queues Works
• Cloudflare Queues JavaScript APIs
• Cloudflare D1 Documentation
• Cloudflare Workers Best Practices
• Cloudflare Workers Cron Triggers
• Cloudflare Workflows Documentation
• Cloudflare Blog — How We Built Cloudflare Queues