Claude APIをCloudflare Queuesで非同期処理する：障害復旧と優先度制御を含めた本番設計

name = "claude-pipeline"
main = "src/index.ts"
compatibility_date = "2026-04-01"
 
# Producer + Consumerを同じWorkerで提供するパターン
[[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            # 1度のバッチで取り出す最大数
max_batch_timeout = 2         # 2秒待ってもバッチが埋まらなければ起動
max_retries = 5               # 一時エラーの最大リトライ
max_concurrency = 10          # 同時起動するConsumer数の上限
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"

// 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' }>();
 
  // 入力検証 — Claude APIに到達する前に弾く方が安い
  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に「pending」として記録 — ステータスポーリングのため
  await env.DB.prepare(
    'INSERT INTO jobs (job_id, status, created_at, user_id) VALUES (?, ?, ?, ?)'
  )
    .bind(jobId, 'pending', job.createdAt, job.userId)
    .run();
 
  // キューに積む — Highは即時、Bulkは遅延もOK
  const queue = body.priority === 'bulk' ? env.QUEUE_BULK : env.QUEUE_HIGH;
  await queue.send(job, {
    contentType: 'json',
    // bulkは少し遅延させてHighの邪魔をしない
    delaySeconds: body.priority === 'bulk' ? 5 : 0,
  });
 
  // 期待出力: { jobId: "...", status: "pending", pollUrl: "/jobs/..." }
  return Response.json(
    { jobId, status: 'pending', pollUrl: `/jobs/${jobId}` },
    { status: 202 }
  );
}

// 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 });
 
  // バッチ内の各メッセージを並列処理 — Cloudflare Queuesは元々バッチで届く
  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;
  const startedAt = Date.now();
 
  try {
    // タイムアウトを明示的に切る — Workerの30sと衝突させない
    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に保存 — D1のサイズ制限を踏まないため
    await env.RESULTS.put(`results/${job.jobId}.txt`, text, {
      httpMetadata: { contentType: 'text/plain; charset=utf-8' },
    });
 
    // D1のステータスを更新
    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(); // 正常完了 — キューから削除される
  } catch (err: unknown) {
    const apiErr = err as { status?: number; message?: string; name?: string };
 
    // 直らないエラー — 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(); // ackすることで「DLQには送らない」明示的失敗
      return;
    }
 
    // 一時エラー — 指数バックオフでリトライ
    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); // 最大10分
      msg.retry({ delaySeconds });
      return;
    }
 
    // 想定外 — DLQに任せる
    msg.retry();
  }
}

// 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;
 
    // ステータスをdead-lettered更新
    await env.DB.prepare(
      'UPDATE jobs SET status = ?, error = ? WHERE job_id = ?'
    )
      .bind('dead-lettered', 'exceeded retry budget', job.jobId)
      .run();
 
    // 重要: ユーザー通知（メール・Webhook等）
    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(() => {}); // 通知失敗は無視（DLQの再DLQは作らない）
    }
 
    msg.ack();
  }
}

// src/budget.ts — Durable Objectでトークン残量を管理
export class TokenBudget {
  state: DurableObjectState;
  // 1分あたりの予算（RPM/TPMはAnthropicコンソールで確認）
  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) {
      // 予算切れ — Consumerにretryさせる
      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 });
  }
}

// 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: '1〜100ファイルを送ってください' },
      { status: 400 }
    );
  }
 
  const batchId = crypto.randomUUID();
  const userId = request.headers.get('x-user-id') ?? 'anonymous';
 
  // 各ファイルをR2に保存し、jobIdごとにキューへ
  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(),
      };
    })
  );
 
  // バルクキューに送る — 1ユーザーが大量投入してもHighキューに影響しない
  await env.QUEUE_BULK.sendBatch(
    jobs.map((j) => ({ body: j, contentType: 'json' }))
  );
 
  // batchIdでステータス追跡できる
  return Response.json(
    {
      batchId,
      total: files.length,
      pollUrl: `/batches/${batchId}`,
    },
    { status: 202 }
  );
}

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

// src/__tests__/consumer.test.ts
import { describe, it, expect, vi } from 'vitest';
import { processOne } from '../consumer';
 
describe('processOne', () => {
  it('completed状態でD1に書く正常系', 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('429はretry、400はack', 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 })),
      },
    };
 
    // 1回目: 429 → retry呼ばれる
    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();
 
    // 2回目: 400 → ack呼ばれる（リトライ不可）
    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();
  });
});

// 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); // 新経路
  }
  return callDirect(prompt, env); // 旧経路
}

// 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'; // 別のConsumerが既に書いた
  }
  await env.RESULTS.put(key, text, {
    httpMetadata: { contentType: 'text/plain; charset=utf-8' },
    onlyIf: { etagDoesNotMatch: '*' }, // 既存があれば失敗
  });
  return 'wrote';
}