Building a Production Claude API Pipeline on Cloudflare Queues: Fault Tolerance, Backpressure, and Cost Control

wrangler.toml — the contract

Get this file right once and you won't fight it again. The shape below is what's actually deployed for the system this article is based on.

name = "claude-pipeline"
main = "src/index.ts"
compatibility_date = "2026-04-01"
 
[[queues.producers]]
queue = "claude-jobs-high"
binding = "QUEUE_HIGH"
 
[[queues.producers]]
queue = "claude-jobs-bulk"
binding = "QUEUE_BULK"
 
[[queues.consumers]]
queue = "claude-jobs-high"
max_batch_size = 5
max_batch_timeout = 2
max_retries = 5
max_concurrency = 10
dead_letter_queue = "claude-jobs-dlq"
 
[[queues.consumers]]
queue = "claude-jobs-bulk"
max_batch_size = 10
max_batch_timeout = 30
max_retries = 3
max_concurrency = 3
dead_letter_queue = "claude-jobs-dlq"
 
[[queues.consumers]]
queue = "claude-jobs-dlq"
max_batch_size = 1
max_concurrency = 1
 
[[d1_databases]]
binding = "DB"
database_name = "claude-jobs"
database_id = "YOUR_D1_DATABASE_ID"
 
[[r2_buckets]]
binding = "RESULTS"
bucket_name = "claude-results"

The max_concurrency asymmetry (10 vs 3) is the load-bearing setting here. Claude API rate limits depend on your tier, but for most production accounts a combined ceiling of ~13 concurrent requests stays comfortably below the RPM cap. If you set both to 10, a bulk upload will bury the high-priority lane the moment it hits.

Producer — keep it boring

The producer is where you should be the least clever. Validate, persist, enqueue, return — that's it.

// src/producer.ts
import type { Env, Job } from './types';
 
export async function enqueueJob(
  request: Request,
  env: Env
): Promise<Response> {
  const body = await request.json<{ prompt: string; priority?: 'high' | 'bulk' }>();
 
  // Reject bad input before we ever pay for a Claude call
  if (!body.prompt || body.prompt.length > 200_000) {
    return Response.json(
      { error: 'prompt missing or exceeds 200k chars' },
      { status: 400 }
    );
  }
 
  const jobId = crypto.randomUUID();
  const job: Job = {
    jobId,
    prompt: body.prompt,
    createdAt: Date.now(),
    attempts: 0,
    userId: request.headers.get('x-user-id') ?? 'anonymous',
  };
 
  // D1 first so we always have an audit row, even if Queue write fails
  await env.DB.prepare(
    'INSERT INTO jobs (job_id, status, created_at, user_id) VALUES (?, ?, ?, ?)'
  )
    .bind(jobId, 'pending', job.createdAt, job.userId)
    .run();
 
  const queue = body.priority === 'bulk' ? env.QUEUE_BULK : env.QUEUE_HIGH;
  await queue.send(job, {
    contentType: 'json',
    delaySeconds: body.priority === 'bulk' ? 5 : 0,
  });
 
  // Expected response: { jobId: "...", status: "pending", pollUrl: "/jobs/..." }
  return Response.json(
    { jobId, status: 'pending', pollUrl: `/jobs/${jobId}` },
    { status: 202 }
  );
}

The order matters: write to D1 first, then enqueue. If you reverse it and the D1 write fails after a successful enqueue, you have a job nobody knows about — completely opaque to support and reconciliation.

Consumer — the part that actually gets ack/retry/DLQ right

This is where I spent the most time getting the details right. My first version blindly called message.retry() on every error, which is exactly wrong: Claude API errors split cleanly into "retry will help" and "retry is a waste of money," and you have to encode that distinction yourself. Get it wrong and a permanently broken job will keep firing every few hours, billing you each time, until someone notices.

// src/consumer.ts
import Anthropic from '@anthropic-ai/sdk';
import type { Env, Job } from './types';
 
const RETRYABLE_STATUS = new Set([408, 429, 500, 502, 503, 504, 529]);
 
export async function consumeJobs(
  batch: MessageBatch<Job>,
  env: Env,
  ctx: ExecutionContext
) {
  const client = new Anthropic({ apiKey: env.ANTHROPIC_API_KEY });
 
  await Promise.all(
    batch.messages.map((msg) => processOne(msg, client, env, ctx))
  );
}
 
async function processOne(
  msg: Message<Job>,
  client: Anthropic,
  env: Env,
  ctx: ExecutionContext
) {
  const job = msg.body;
 
  try {
    // Always set an explicit timeout so it never collides with the Worker limit
    const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), 25_000);
 
    const response = await client.messages.create(
      {
        model: 'claude-sonnet-4-6',
        max_tokens: 4096,
        messages: [{ role: 'user', content: job.prompt }],
      },
      { signal: controller.signal }
    );
 
    clearTimeout(timeoutId);
 
    const text = response.content
      .filter((b) => b.type === 'text')
      .map((b) => (b as { text: string }).text)
      .join('\n');
 
    // R2 for the body so we don't blow past D1's row size
    await env.RESULTS.put(`results/${job.jobId}.txt`, text, {
      httpMetadata: { contentType: 'text/plain; charset=utf-8' },
    });
 
    await env.DB.prepare(
      'UPDATE jobs SET status = ?, completed_at = ?, input_tokens = ?, output_tokens = ? WHERE job_id = ?'
    )
      .bind(
        'completed',
        Date.now(),
        response.usage.input_tokens,
        response.usage.output_tokens,
        job.jobId
      )
      .run();
 
    msg.ack(); // success — remove from queue
  } catch (err: unknown) {
    const apiErr = err as { status?: number; message?: string; name?: string };
 
    // Permanent failures — don't retry, don't poison the DLQ
    if (apiErr.status === 400 || apiErr.status === 401 || apiErr.status === 403) {
      await env.DB.prepare(
        'UPDATE jobs SET status = ?, error = ? WHERE job_id = ?'
      )
        .bind('failed', `${apiErr.status}: ${apiErr.message}`, job.jobId)
        .run();
      msg.ack();
      return;
    }
 
    // Transient failures — exponential backoff
    if (
      apiErr.name === 'AbortError' ||
      RETRYABLE_STATUS.has(apiErr.status ?? 0)
    ) {
      const attempt = msg.attempts ?? 0;
      const delaySeconds = Math.min(60 * Math.pow(2, attempt), 600);
      msg.retry({ delaySeconds });
      return;
    }
 
    // Anything else — let max_retries push it to the DLQ
    msg.retry();
  }
}

The ack vs retry distinction is the part I want you to internalize. ack removes the message from the queue (success). retry puts it back, counted toward max_retries, after which it's auto-forwarded to the DLQ. Permanent failures should be ack'd with a failed row written to D1 — that way the DLQ stays clean and reserved for jobs that genuinely deserve human attention.

Dead-letter handling

The DLQ collects jobs that exhausted all retries. The temptation is to retry Claude one more time from here. Don't. The DLQ consumer's job is to record the failure and notify someone — not to keep paying for the same broken request.

// src/dlq.ts
import type { Env, Job } from './types';
 
export async function consumeDeadLetters(
  batch: MessageBatch<Job>,
  env: Env
) {
  for (const msg of batch.messages) {
    const job = msg.body;
 
    await env.DB.prepare(
      'UPDATE jobs SET status = ?, error = ? WHERE job_id = ?'
    )
      .bind('dead-lettered', 'exceeded retry budget', job.jobId)
      .run();
 
    if (env.NOTIFY_WEBHOOK_URL) {
      await fetch(env.NOTIFY_WEBHOOK_URL, {
        method: 'POST',
        headers: { 'content-type': 'application/json' },
        body: JSON.stringify({
          event: 'job.failed',
          jobId: job.jobId,
          userId: job.userId,
          createdAt: job.createdAt,
        }),
      }).catch(() => {}); // intentionally swallow — never let DLQ retry itself
    }
 
    msg.ack();
  }
}

That .catch(() => {}) looks sloppy at first glance, but it's deliberate. If the DLQ consumer can throw, you'll create a DLQ-of-the-DLQ situation. Push observability into Logpush or an external collector instead, and let the DLQ consumer be the dumb-but-reliable last resort.

Token budgeting with a Durable Object

Claude API has two independent rate limits: requests per minute (RPM) and tokens per minute (TPM). Cloudflare Queues' max_concurrency only governs request count, so a stream of long-prompt jobs can starve TPM long before RPM becomes a concern. A Durable Object gives us a process-global counter without any external infrastructure.

// src/budget.ts
export class TokenBudget {
  state: DurableObjectState;
  private readonly perMinute = 400_000;
 
  constructor(state: DurableObjectState) {
    this.state = state;
  }
 
  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    if (url.pathname !== '/reserve') return new Response('not found', { status: 404 });
 
    const requested = Number(url.searchParams.get('tokens') ?? '0');
    const now = Math.floor(Date.now() / 60_000);
 
    const stored = await this.state.storage.get<{ minute: number; used: number }>(
      'bucket'
    );
    const bucket = !stored || stored.minute !== now
      ? { minute: now, used: 0 }
      : stored;
 
    if (bucket.used + requested > this.perMinute) {
      return Response.json({ allowed: false, retryAfterSeconds: 60 - (Date.now() % 60_000) / 1000 });
    }
 
    bucket.used += requested;
    await this.state.storage.put('bucket', bucket);
    return Response.json({ allowed: true });
  }
}

Before each messages.create call, the consumer hits this DO with an estimated token cost (computed from prompt length). If it gets allowed: false, it just calls msg.retry() with an appropriate delay. A back-of-envelope estimate is enough to absorb 90%+ of TPM-driven 429s. Pair this with the practices in the dedicated rate limit and 429 handling guide for the long tail.

Five mistakes I've made (or watched others make)

1. Forgetting to call ack after a try/catch swallow. Cloudflare Queues default to "exception means retry." If you catch an error, write to D1, and don't ack, the job will get retried — and you'll pay Claude again for work you already finished. Make msg.ack() on the success path explicit and unmissable.

2. Cranking max_batch_size too high. Setting it to 100 and Promise.all-ing the batch creates a 100-wide concurrent burst against Claude's API on every Worker invocation. I've seen this nuke entire orgs into rate-limit jail. Keep max_batch_size between 5–10 and use p-limit or a similar concurrency limiter inside the consumer.

3. Trusting max_retries blindly. Five exponential-backoff retries don't help when the API is genuinely down for an hour — they just burn the budget early. For jobs with a meaningful "stale by" window, attach expiresAt at enqueue time and short-circuit-ack in the consumer when it's exceeded.

4. Putting the prompt body directly on the queue. Cloudflare Queues caps message size at 128 KiB. Long prompts will break this. Store the prompt in R2 or D1 and only put the ID on the queue — the claim-check pattern is the standard answer here.

5. Forgetting the DLQ exists until it's full. DLQs accumulate quietly. Wire up SELECT COUNT(*) FROM jobs WHERE status='dead-lettered' as a recurring check on day one, with a Slack or email alert above some threshold. It's the lowest-effort piece of observability you can add and the one I most regret postponing.

A real workload: parallel PDF analysis

Tying it together, here's how the same architecture handles a workload I actually run: users upload 10–100 PDFs at a time, each gets summarized and structured-extracted by Claude.

// src/pdf-pipeline.ts
import type { Env } from './types';
 
export async function uploadAndDispatch(
  request: Request,
  env: Env
): Promise<Response> {
  const form = await request.formData();
  const files = form.getAll('files') as File[];
 
  if (files.length === 0 || files.length > 100) {
    return Response.json(
      { error: 'send between 1 and 100 files' },
      { status: 400 }
    );
  }
 
  const batchId = crypto.randomUUID();
  const userId = request.headers.get('x-user-id') ?? 'anonymous';
 
  const jobs = await Promise.all(
    files.map(async (file, idx) => {
      const jobId = crypto.randomUUID();
      const key = `uploads/${batchId}/${jobId}.pdf`;
      await env.RESULTS.put(key, file.stream());
      return {
        jobId,
        batchId,
        userId,
        r2Key: key,
        idx,
        promptTemplate: 'pdf-summarize-v1',
        createdAt: Date.now(),
      };
    })
  );
 
  // Bulk lane — even a 100-file upload won't disturb the high-priority lane
  await env.QUEUE_BULK.sendBatch(
    jobs.map((j) => ({ body: j, contentType: 'json' }))
  );
 
  return Response.json(
    {
      batchId,
      total: files.length,
      pollUrl: `/batches/${batchId}`,
    },
    { status: 202 }
  );
}

The consumer adds an R2 GET and a Claude document block, but the skeleton is the same. With max_concurrency: 3 on the bulk consumer, even a user dropping 100 PDFs at once translates to a steady 3-wide stream against Claude — interactive users on the high-priority lane never feel it. That asymmetry is the whole point of the architecture.

For the deeper observability pattern referenced in the rules of thumb, Practical Monitoring by Mike Julian remains the cleanest treatment of alert-threshold design I've seen, and I came back to it repeatedly when calibrating DLQ alerts.

A D1 schema you won't regret on day 90

Status tracking alone isn't enough; you want enough operational shape baked into the schema that a 3 AM debugging session is bearable. The shape I converged on after several iterations is below.

-- migrations/0001_jobs.sql
CREATE TABLE jobs (
  job_id TEXT PRIMARY KEY,
  batch_id TEXT,
  user_id TEXT NOT NULL,
  status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'completed', 'failed', 'dead-lettered')),
  priority TEXT NOT NULL DEFAULT 'high',
  created_at INTEGER NOT NULL,
  started_at INTEGER,
  completed_at INTEGER,
  attempts INTEGER NOT NULL DEFAULT 0,
  input_tokens INTEGER,
  output_tokens INTEGER,
  model TEXT,
  error TEXT
);
 
CREATE INDEX idx_jobs_status_created ON jobs (status, created_at);
CREATE INDEX idx_jobs_user_status ON jobs (user_id, status);
CREATE INDEX idx_jobs_batch ON jobs (batch_id);

The CHECK on status exists for the same reason TypeScript's strict flag exists: a typo somewhere in your codebase will silently insert pendigg and you'll spend an hour wondering why the dashboard counts don't add up. Catching that at the database boundary is much cheaper than discovering it on a Friday afternoon.

Four metrics are worth surfacing on a dashboard from day one. Latency percentiles (completed_at - started_at) bucketed at p50/p95/p99 — for Sonnet 4.6 on long summarization workloads, p95 hovers around 40 seconds and p99 can spike past 80. A realistic SLO is "p95 under 60 seconds." Failure rate, separated into failed and dead-lettered. They mean different things: one is a permanent error you should fix in code, the other is a retry budget exhaustion that often points to upstream incidents. Queue depth, expressed as the age of the oldest pending job. Anything older than ten minutes means either consumer starvation or upstream Claude API problems. Token consumption, sliced by user and model, which becomes your single source of truth for cost.

All four can be computed with one SELECT each, so you can run them on a Cron-triggered Worker and ship the results to whatever observability stack you use. My current setup writes them into a metrics table on D1 every minute and exports to Grafana Cloud, but Logpush directly into your existing stack works fine.

Local development without burning real Claude calls

wrangler dev simulates Queues locally well enough for happy-path development, but the moment you start exercising error branches you don't want every test run charging your Anthropic account. Mocking the Anthropic client is the only sustainable approach.

// src/__tests__/consumer.test.ts
import { describe, it, expect, vi } from 'vitest';
import { processOne } from '../consumer';
 
describe('processOne', () => {
  it('writes a completed row on success', async () => {
    const fakeClient = {
      messages: {
        create: vi.fn().mockResolvedValue({
          content: [{ type: 'text', text: 'OK' }],
          usage: { input_tokens: 100, output_tokens: 20 },
        }),
      },
    };
 
    const ackSpy = vi.fn();
    const retrySpy = vi.fn();
    const msg = {
      body: { jobId: 'job-1', prompt: 'hello', userId: 'u1', createdAt: 0, attempts: 0 },
      ack: ackSpy,
      retry: retrySpy,
      attempts: 0,
    };
 
    const env = {
      RESULTS: { put: vi.fn().mockResolvedValue(undefined) },
      DB: {
        prepare: () => ({ bind: () => ({ run: vi.fn().mockResolvedValue({}) }) }),
      },
      ANTHROPIC_API_KEY: 'sk-test',
    };
 
    await processOne(msg as any, fakeClient as any, env as any, {} as any);
 
    expect(ackSpy).toHaveBeenCalledOnce();
    expect(retrySpy).not.toHaveBeenCalled();
  });
 
  it('retries on 429, acks on 400', async () => {
    const fakeClient = {
      messages: {
        create: vi.fn()
          .mockRejectedValueOnce(Object.assign(new Error('rate limited'), { status: 429 }))
          .mockRejectedValueOnce(Object.assign(new Error('bad request'), { status: 400 })),
      },
    };
 
    const ack1 = vi.fn(), retry1 = vi.fn();
    await processOne(
      { body: { jobId: 'j', prompt: 'p', userId: 'u', createdAt: 0, attempts: 0 }, ack: ack1, retry: retry1, attempts: 0 } as any,
      fakeClient as any,
      { /* env stub */ } as any,
      {} as any
    );
    expect(retry1).toHaveBeenCalled();
    expect(ack1).not.toHaveBeenCalled();
 
    const ack2 = vi.fn(), retry2 = vi.fn();
    await processOne(
      { body: { jobId: 'j', prompt: 'p', userId: 'u', createdAt: 0, attempts: 0 }, ack: ack2, retry: retry2, attempts: 0 } as any,
      fakeClient as any,
      { DB: { prepare: () => ({ bind: () => ({ run: vi.fn() }) }) } } as any,
      {} as any
    );
    expect(ack2).toHaveBeenCalled();
  });
});

What this guards is the most expensive regression possible: misclassifying a permanent error as retryable. If both of these tests pass on every PR, you'll never accidentally deploy a build that turns a 400 into an infinite retry loop. That's the only piece of test coverage in this entire system that I consider non-negotiable.

Rolling out to existing traffic without burning the building

Migrating an existing service from synchronous Claude calls to a queued pipeline is more delicate than building greenfield. Cutting over all traffic at once is rarely worth it. The four-step rollout I've used:

Step one is "low-stakes endpoints only." Pick endpoints where the user can wait and where a transient failure won't break anything important — generating a profile bio blurb, summarizing a long-tail document the user opened. This gives you confidence in the shape without staking real revenue on it.

Step two introduces deterministic feature flagging. You want 5% of users to flow through the new pipeline while 95% remain on the old path. A SHA-1 hash of userId modulo 100 is enough; it's deterministic per user, which means support tickets are reproducible.

// src/feature-flag.ts
async function isInRollout(userId: string, percent: number): Promise<boolean> {
  const buf = new TextEncoder().encode(userId);
  const hash = await crypto.subtle.digest('SHA-1', buf);
  const view = new DataView(hash);
  const value = view.getUint32(0) % 100;
  return value < percent;
}
 
export async function callClaude(prompt: string, userId: string, env: Env) {
  if (await isInRollout(userId, env.QUEUE_ROLLOUT_PERCENT ?? 5)) {
    return enqueueAndPoll(prompt, userId, env); // new path
  }
  return callDirect(prompt, env); // legacy path
}

Step three takes the rollout to 50%. Watch p95 latency, failure rate, and cost on both paths in parallel for at least 48 hours before deciding to proceed. Step four is 100%. The crucial property of this design is that QUEUE_ROLLOUT_PERCENT is a wrangler secret, so rolling back to 0% takes one command, no redeploy. The failure mode of "panicked midnight rollback" is not just survivable — it's boring.

Real-world cost breakdown for 10,000 jobs

The justification for any extra architectural piece is, ultimately, money. Here's the actual breakdown for processing 10,000 PDF analysis jobs through this pipeline. Average prompt: 3,000 input tokens, 800 output tokens, on Sonnet 4.6.

Claude API itself dominates: 30M input tokens and 8M output tokens at Sonnet 4.6's standard pricing comes out to roughly $135 ($90 input + $45 output). Adding prompt caching on the system prompt or recurring document scaffolding can shave 20–50% off that, depending on hit rate.

The Cloudflare side is comparatively trivial. Workers requests for the producer (10k) plus consumer invocations (about 10.5k including retries) sit comfortably inside the bundled Workers Paid plan, costing under $0.50 of variable usage on top of the base $5/month. Queues operations — a send, a receive, and an ack per message — total about 30,000 operations. The free tier covers the first 1M, so this is effectively zero. D1 row writes and reads land inside the bundled allotment as well. R2 holds both the input PDFs and the output text; for around 10 GB of storage, that's about $0.15/month.

In other words, the entire Cloudflare-side overhead of running this pipeline is around $5–6/month flat, regardless of whether you process 100 jobs or 10,000. The variable cost is essentially all Claude API. What the queue infrastructure buys you is much harder to put a number on but worth far more than the $5: it's the elimination of "we just paid Claude twice for the same job because the Worker timed out and the client retried." For a deeper treatment of the optimization side, see the Anthropic API cost optimization guide.

Queues vs. the Messages Batches endpoint

After reading this far you might wonder if you should just put everything on Queues. The honest answer is no — the Messages Batches API and Queues are complements, and the dividing line is clear.

If your user expects a result within a minute, Queues is the only sensible choice. The Batches API only guarantees completion within 24 hours, which is fundamentally incompatible with any interactive UX. But if your workload is "summarize last night's logs by tomorrow morning," the Batches API's 50% pricing discount is hard to ignore. You're paying half for compute that runs on Anthropic's schedule rather than yours, and that trade is great for genuinely background work.

The hybrid pattern I run in production uses both. The high-priority queue calls Claude API directly (latency-critical). The bulk queue's consumer doesn't actually call Claude — it accumulates jobs and submits them as a batch to the Messages Batches API every hour. Result polling and storage stay in the same D1/R2 plumbing. This keeps job lifecycle uniform across the system while capturing the discount where it applies.

Idempotency — "safe to run twice" is a feature, not an accident

Any system with retries — Cloudflare Queues or otherwise — must accept that the same message will sometimes be processed more than once. The textbook scenario: the Consumer finishes writing a Claude response to R2, and the Worker dies a millisecond before msg.ack() reaches the queue manager. There is no clean way to avoid this without distributed transactions, so the only honest answer is to make the Consumer idempotent — running it twice produces the same observable result as running it once.

The simplest and most reliable mechanism is to use jobId as both the R2 key and the write predicate.

// src/idempotent-write.ts
async function writeResultIfFirst(
  env: Env,
  jobId: string,
  text: string
): Promise<'wrote' | 'already-existed'> {
  const key = `results/${jobId}.txt`;
  const head = await env.RESULTS.head(key);
  if (head) {
    return 'already-existed'; // another consumer beat us to it
  }
  await env.RESULTS.put(key, text, {
    httpMetadata: { contentType: 'text/plain; charset=utf-8' },
    onlyIf: { etagDoesNotMatch: '*' }, // fails if anything is there
  });
  return 'wrote';
}

The conditional PUT ensures that even if two consumers process the same message simultaneously, only one write succeeds. When the consumer detects already-existed, it concludes "I'm the duplicate" and short-circuits to msg.ack() without calling Claude again. That single check structurally prevents double-charging.

The same pattern applies to D1. Update with a guard: UPDATE jobs SET ... WHERE job_id = ? AND status != 'completed'. The first-completed write wins; subsequent attempts become no-ops. Token counts from the first successful run are protected.

Push the idempotency contract one layer up — into the Producer — and you become resilient to client-side retries too. Accept an Idempotency-Key header in enqueueJob, and if the same key shows up twice, return the original jobId rather than creating a new one. Anyone who's worked with Stripe or other payment APIs will recognize the pattern; it's just as valuable in an AI pipeline. A flaky network on the user's side no longer translates to two Claude calls.

Where to start tomorrow

Don't try to migrate everything in one shot. Pick a single Claude API call that users could reasonably wait on — the most "background-y" thing in your app — and route just that one through QUEUE_BULK. The producer change is under ten lines. The consumer is the processOne function above. From there, every subsequent migration is mechanical: fan-out, observe, ack, retry, DLQ.

When you're ready to layer in higher availability, combine this with multi-model fallback so a single provider outage no longer takes the pipeline with it.