Claude API 529 과부하 대응 - 재시도·큐잉 설계

import pLimit from "p-limit";

type ClaudeResponse = unknown;

type CallOptions = {
  maxRetries?: number;
  baseDelayMs?: number;
  capDelayMs?: number;
  timeoutMs?: number;
};

function sleep(ms: number) {
  return new Promise((r) => setTimeout(r, ms));
}

function parseRetryAfterSeconds(headers: Headers): number | null {
  const v = headers.get("retry-after");
  if (!v) return null;
  const n = Number(v);
  return Number.isFinite(n) ? n : null;
}

function fullJitterDelay(base: number, cap: number, attempt: number) {
  const exp = Math.min(cap, base * Math.pow(2, attempt));
  return Math.floor(Math.random() * exp);
}

async function withTimeout<T>(p: Promise<T>, timeoutMs: number): Promise<T> {
  const ac = new AbortController();
  const t = setTimeout(() => ac.abort(), timeoutMs);
  try {
    // 호출부에서 signal을 받아 사용하도록 구성하는 편이 더 좋지만,
    // 예시는 단순화를 위해 timeout 경쟁으로 처리합니다.
    return await Promise.race([
      p,
      new Promise<T>((_, rej) =>
        ac.signal.addEventListener("abort", () => rej(new Error("timeout")))
      ),
    ]);
  } finally {
    clearTimeout(t);
  }
}

async function callClaudeOnce(prompt: string): Promise<{ ok: boolean; status: number; headers: Headers; json?: ClaudeResponse }> {
  // 실제로는 Anthropic SDK를 쓰거나 fetch로 호출합니다.
  const res = await fetch("https://api.anthropic.com/v1/messages", {
    method: "POST",
    headers: {
      "content-type": "application/json",
      "x-api-key": process.env.ANTHROPIC_API_KEY ?? "",
      "anthropic-version": "2023-06-01",
    },
    body: JSON.stringify({
      model: "claude-3-5-sonnet-latest",
      max_tokens: 512,
      messages: [{ role: "user", content: prompt }],
    }),
  });

  const ok = res.ok;
  const status = res.status;
  const headers = res.headers;

  if (!ok) return { ok, status, headers };
  return { ok, status, headers, json: await res.json() };
}

export function createClaudeClient(concurrency: number) {
  const limit = pLimit(concurrency);

  return {
    call(prompt: string, opts: CallOptions = {}) {
      const {
        maxRetries = 4,
        baseDelayMs = 250,
        capDelayMs = 4000,
        timeoutMs = 30000,
      } = opts;

      return limit(async () => {
        let lastErr: unknown;

        for (let attempt = 0; attempt <= maxRetries; attempt++) {
          try {
            const res = await withTimeout(callClaudeOnce(prompt), timeoutMs);

            if (res.ok) return res.json;

            // 529: 과부하이므로 재시도 후보
            if (res.status === 529) {
              const ra = parseRetryAfterSeconds(res.headers);
              const delay = ra != null
                ? ra * 1000
                : fullJitterDelay(baseDelayMs, capDelayMs, attempt);

              await sleep(delay);
              continue;
            }

            // 기타 상태코드: 여기서는 단순화
            throw new Error(`claude error status=${res.status}`);
          } catch (e) {
            lastErr = e;
            // 네트워크 오류도 제한적으로 재시도 가능
            if (attempt < maxRetries) {
              const delay = fullJitterDelay(baseDelayMs, capDelayMs, attempt);
              await sleep(delay);
              continue;
            }
          }
        }

        throw lastErr ?? new Error("claude call failed");
      });
    },
  };
}

create table ai_job (
  id bigserial primary key,
  idempotency_key text not null,
  status text not null check (status in ('queued','running','succeeded','failed')),
  priority int not null default 0,
  payload jsonb not null,
  attempt int not null default 0,
  available_at timestamptz not null default now(),
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create unique index ai_job_idem_uk on ai_job (idempotency_key);
create index ai_job_pick_idx on ai_job (status, available_at, priority, created_at);

with next_job as (
  select id
  from ai_job
  where status = 'queued'
    and available_at <= now()
  order by priority desc, created_at asc
  for update skip locked
  limit 1
)
update ai_job
set status = 'running', updated_at = now()
where id in (select id from next_job)
returning *;

function computeBackoffMs(attempt: number) {
  const base = 500;
  const cap = 15000;
  const exp = Math.min(cap, base * Math.pow(2, attempt));
  return Math.floor(Math.random() * exp);
}

// 의사코드
async function handleJob(job: any, claude: any, db: any) {
  try {
    const result = await claude.call(job.payload.prompt, { maxRetries: 1 });

    await db.query(
      "update ai_job set status='succeeded', updated_at=now(), payload = payload || $2::jsonb where id=$1",
      [job.id, JSON.stringify({ result })]
    );
  } catch (e: any) {
    const isOverload = String(e?.message ?? "").includes("status=529");

    if (isOverload && job.attempt < 10) {
      const delay = computeBackoffMs(job.attempt);
      await db.query(
        "update ai_job set status='queued', attempt=attempt+1, available_at=now()+($2||' milliseconds')::interval, updated_at=now() where id=$1",
        [job.id, String(delay)]
      );
      return;
    }

    await db.query(
      "update ai_job set status='failed', updated_at=now() where id=$1",
      [job.id]
    );
  }
}