Claude API のジョブキューに優先度と公平性を組み込む設計 — 5,000万DL個人開発で実装したバックプレッシャーの実践

3層プライオリティキューの実装 — critical / standard / batch

私のワーカーが Redis 上に持っているキューは 3 本です。それぞれ別の論理 LIST で、優先度のラベルだけ違います。enqueue する側は、リクエストの性質に応じてどのキューに入れるかを決めます。

# enqueue.py — 呼び出し側の判定
def classify_priority(req) -> str:
    # 1. ユーザー操作起点 → critical
    if req.origin == "user_action":
        return "critical"
    # 2. 7日以内に Claude が応答すれば良いもの → batch
    if req.sla_seconds is None or req.sla_seconds >= 86400:
        return "batch"
    # 3. それ以外（レビュー返信、定期サマリ等）→ standard
    return "standard"
 
def enqueue(redis, req):
    priority = classify_priority(req)
    payload = json.dumps(req.to_dict())
    redis.lpush(f"queue:{priority}:{req.tenant_id}", payload)
    redis.zincrby("tenant:enqueued_count", 1, req.tenant_id)

ワーカー側は、毎ループの先頭で「どのキューから読むか」を決めます。素朴な実装は「critical があれば critical、なければ standard、なければ batch」ですが、これだと batch が永遠に動かない（飢餓）ので、最低保証を入れます。

# worker.py — 取り出し側の優先度判定
import random
 
# 重み: critical 70%, standard 25%, batch 5%
PRIORITY_WEIGHTS = [("critical", 70), ("standard", 25), ("batch", 5)]
 
def pick_priority(redis) -> str | None:
    # critical が空でなければ常に critical
    if redis.exists("queue:critical:*"):
        return "critical"
    # weighted random で standard と batch を選ぶ
    r = random.random() * 100
    cumulative = 0
    for name, w in PRIORITY_WEIGHTS:
        cumulative += w
        if r < cumulative:
            if has_any(redis, name):
                return name
    return None

ここで重要なのは、critical を「絶対優先」にしておく一方で、standard と batch には確率的な配分を与えていることです。「critical が空のときに standard を必ず選ぶ」では batch が永遠に動かないことがあり（standard に投入が続けばその時間は batch がゼロ）、本番で 1 度詰まらせました。確率配分にしておくと、最悪でも 5% のスループットは batch に保証されます。

DRR (Deficit Round Robin) でテナント間の公平性を保証する

優先度キューを 3 本に分けても、同じ優先度の中でテナント間の公平性は別問題です。例えば critical キューの中で、テナント A が 100 件、テナント B が 1 件のリクエストを持っているとき、素朴に FIFO すると B のリクエストが A の 100 件の後ろに並びます。

これを解くのが DRR（Deficit Round Robin）です。各テナントに「クォータ（赤字許容量）」を持たせ、毎ラウンドで一定量を補充し、自分のクォータの範囲内で取り出せる分だけ取り出す、というアルゴリズムです。

# drr.py — Deficit Round Robin の実装
class DRRScheduler:
    def __init__(self, redis, quantum: int = 5):
        self.redis = redis
        self.quantum = quantum  # 1ラウンドで補充するクォータ
        self.deficit: dict[str, int] = {}  # tenant_id -> 累積クォータ
 
    def pick(self, priority: str) -> tuple[str, bytes] | None:
        # active なテナント一覧（このpriorityで在庫があるテナント）
        active = self._active_tenants(priority)
        if not active:
            return None
 
        # Round-robin で全テナントを巡回
        for tenant_id in active:
            self.deficit.setdefault(tenant_id, 0)
            self.deficit[tenant_id] += self.quantum
 
            # クォータの範囲内で取り出せるだけ取り出す
            while self.deficit[tenant_id] >= 1:
                payload = self.redis.rpop(f"queue:{priority}:{tenant_id}")
                if payload is None:
                    break  # このテナントの在庫が尽きた
                cost = self._estimate_cost(payload)  # 推定トークン数 / 1000
                if cost > self.deficit[tenant_id]:
                    # クォータ不足。次のラウンドに持ち越し
                    self.redis.rpush(f"queue:{priority}:{tenant_id}", payload)
                    break
                self.deficit[tenant_id] -= cost
                return tenant_id, payload
        return None
 
    def _estimate_cost(self, payload: bytes) -> int:
        # 入力文字数 / 4000 を推定トークンコストとする（雑な見積もり）
        req = json.loads(payload)
        return max(1, len(req["input"]) // 4000)

このスケジューラは、テナント A が長文を多く持っていてもクォータ枯渇で頭が止まり、テナント B が短文 1 件持っているなら次のラウンドで取り出されます。実測では、テナント A が critical キューに 100 件投入したシナリオで、テナント B の 1 件の応答時間が p99 で 9.8 秒 → 1.7 秒に短縮されました。

quantum の値は本番でチューニングが要ります。私は最初 quantum=1 で始めましたが、短文が偏在するときに長文がいつまで経っても取り出されない（赤字が積み上がらない）症状が出て、最終的に quantum=5（= 入力トークン換算で約 20,000 トークン/ラウンド）に落ち着いています。

バックプレッシャーと 429 の作法 — Retry-After を尊重する設計

優先度と公平性が入っても、Anthropic 側のレート制限を踏むと全てが止まります。私のティアでは 1 分間に 50 リクエスト・40,000 入力トークン・8,000 出力トークンが上限で、これを超えると 429 が返ります。

ここで素朴に「即時リトライ」を入れると、429 が連続して悪化するだけです。Retry-After ヘッダを必ず読み、その値の 1.2倍 の jitter 付きでスリープするのが本番で安定しました。

# call_with_backpressure.py
import asyncio, random
 
async def call_claude_with_backpressure(client, scheduler, payload, max_retry=4):
    for attempt in range(max_retry + 1):
        try:
            return await client.messages.create(**payload)
        except APIStatusError as e:
            if e.status_code != 429:
                raise
            retry_after = float(e.response.headers.get("retry-after", "1.0"))
            # jitter は ±20%
            sleep_for = retry_after * 1.2 * (1.0 + (random.random() - 0.5) * 0.4)
            scheduler.record_throttle(payload["tenant_id"], retry_after)
            await asyncio.sleep(sleep_for)
    raise RuntimeError("max retry exceeded")

record_throttle がポイントです。私は、テナントごとに直近 60 秒間にどれだけ 429 を引き起こしたかを Redis の sorted set で記録し、ある閾値を超えたテナントを次のセクションで述べる「予算超過テナント」として降格させています。429 をワーカー全体ではなくテナント単位で原因帰属するのが大事です。あるテナントが暴走している間、他のテナントが巻き添えで遅くなる構図を断ち切れます。

実測では、jitter なしの素朴な指数バックオフから jitter 付きに変えたことで、429 の再発率が 38% → 7% に下がりました。jitter は気休めではなく、複数ワーカーが同時に同じタイミングで再試行するスタンピード（thundering herd）を確実に防いでいます。

「予算超過テナント」の自動降格メカニズム

DRR で公平にしても、あるテナントが許容量を超えて投入し続けると、結局 standard 全体のスループットを蝕みます。私は「直近 60 秒で 429 を 5 回以上引き起こした」または「日次の入力トークン量が予算の 150% を超えた」テナントを、自動で standard → batch に降格する仕組みを入れています。

# tenant_demotion.py
DEMOTION_THRESHOLDS = {
    "throttle_60s": 5,
    "daily_budget_ratio": 1.5,
}
 
def check_and_demote(redis, tenant_id: str) -> bool:
    # 1. 直近60秒の 429 回数
    now = time.time()
    redis.zremrangebyscore(f"throttle:{tenant_id}", 0, now - 60)
    throttle_count = redis.zcard(f"throttle:{tenant_id}")
    # 2. 日次トークン消費
    daily_tokens = int(redis.get(f"tokens:{tenant_id}:{today_key()}") or 0)
    budget = int(redis.get(f"budget:{tenant_id}") or 100000)
 
    if (throttle_count >= DEMOTION_THRESHOLDS["throttle_60s"] or
            daily_tokens > budget * DEMOTION_THRESHOLDS["daily_budget_ratio"]):
        redis.setex(f"demoted:{tenant_id}", 300, "1")  # 5分間 batch 降格
        return True
    return False

降格状態のテナントが投入するリクエストは、enqueue 側で classify_priority を上書きして batch に押し込みます。降格は 5 分で自動解除されます。これは「ペナルティ」というよりは「コンジェスチョンコントロール」に近い設計で、テナントの責任を問うのではなく、システム全体の健全性を保つための自己防衛です。

実体験として、私のアプリの一つで、サードパーティのレビュー集約サービスが API を叩きすぎてしまい、彼ら（私のテナント）が他の私のアプリの返信スループットを完全に止めた時期がありました。降格メカニズムを入れてから、彼らが暴走しても他のアプリへの影響は 5 分以内に収束するようになりました。広告 eCPM への影響も 12% 程度のスループット低下で済むようになり、収益への打撃が許容範囲に収まっています。

運用メトリクスと SLO — テールレイテンシで品質を測る

スケジューラとバックプレッシャーが入ったら、次は「ちゃんと効いているか」を測る必要があります。私が最終的に Prometheus / Grafana に出している主要メトリクスは次の 6 つです。

• claude_queue_depth{priority} — 各キューの在庫数（30 秒ごとに gauge）
• claude_request_latency_seconds{priority} — リクエスト完了までの時間（ヒストグラム）
• claude_throttle_count{tenant_id} — テナントごとの 429 回数（直近 60 秒）
• claude_demoted_tenants — 現時点で降格中のテナント数
• claude_drr_fairness_index — DRR の公平性スコア（Jain's fairness index、後述）
• claude_input_tokens_total{tenant_id} — テナントごとの累積入力トークン

特に重要なのは p99 レイテンシ と Jain's fairness index です。p50（中央値）はどんな設計でも見栄えがいいので、p99 を SLO にしないとテール側の問題を見逃します。私の場合、critical キューの p99 を 2.0 秒、standard の p99 を 6.0 秒、batch の p99 を 120 秒という SLO で運用しています。

Jain's fairness index は、(Σx_i)² / (n * Σx_i²) で計算される値で、全テナントが完全に公平なら 1.0、極端に偏ると 1/n に近づきます。私のスケジューラ導入前は 0.42（テナント 8 つでほぼ 1 つだけが占有）でしたが、DRR 導入後は 0.86〜0.91 で安定しています。テナント数が変動しても比較できるので、運用上のヘルスチェックに便利です。

ここで「公平すぎる」状態にもリスクがあります。例えば全テナントに完全に等量配分すると、ヘビーユーザーが想定する応答時間を満たせなくなり、課金プランの根拠が崩れます。私は有料テナント（Pro プラン以上）に DRR の quantum を 3 倍にする補正を入れ、公平性指数を意図的に 0.85 前後に保っています。

1年運用して気付いた4つの落とし穴

ここまでが設計の話で、最後に本番で踏んだ落とし穴を 4 つ共有します。

1 つ目: critical の絶対優先は「優先度反転」を起こす

critical を絶対優先にすると、standard キューで稼働中のワーカーが critical の到着を待ってアイドルする時間が増え、結果として standard の総スループットが落ちる現象がありました。私は「critical が空のときだけ standard を選ぶ」ではなく、ワーカーごとに critical 優先確率 = 0.9 を持たせる確率優先に変えました。これだけで standard の p50 が 4.1 秒 → 2.6 秒に改善しました。

2 つ目: クォータ補充タイミングのバイアス

DRR の quantum を全テナントに一度に補充するナイーブな実装だと、補充直後に全テナントが一斉にリクエストを取り出して Claude API への突発負荷を引き起こします。私は補充タイミングをテナントごとにずらす（hash で 0〜1000ms 散らす）ことで、API への秒間リクエスト数の標準偏差を 3.4 → 1.1 に減らしました。

3 つ目: enqueue 時の優先度分類は「呼び出し元」を信用しない

最初は呼び出し元が priority="critical" を指定する設計でしたが、現場のコードが増えるにつれて「とりあえず critical で投げる」スタイルが蔓延しました。今は呼び出し元が指定するのは origin（user_action / scheduled / batch_job）だけで、優先度はスケジューラ側でルールベースに決めています。設計上の責任を一箇所に寄せるべきポイントです。

4 つ目: モデル切り替えとキュー設計は連動する

Claude Haiku 4.5 を導入してから、長文ジョブの処理時間が中央値で 54% 短縮されました。これによって、それまで batch に降格させていた処理の一部を standard で十分捌けるようになり、降格しきい値を緩める必要が出ました。モデルが変わるとキューの最適パラメータも動きます。私は四半期に 1 度、過去 30 日のレイテンシ分布を見て quantum と降格しきい値を再チューニングしています。

本番で 1 年回した今、優先度キューと DRR を入れる前と後で「同じ Claude API ティア・同じワーカー台数」のままユーザー体験の体感がはっきり変わりました。最も大きいのは、私自身がアラートで叩き起こされる頻度が 週 3 回 → 月 1 回 に減ったことです。設計はユーザー向けの体験だけでなく、運用者の睡眠時間にも効きます。

次に進むなら、本記事のコードをそのまま流用して、まず quantum=5 の DRR と 5% batch 保証 から動かしてみてください。1 週間運用して p99 と Jain's fairness index のグラフを眺めると、自分のワークロードに合った quantum の感覚が掴めるはずです。同じ詰まりに悩んでいる方の参考になれば幸いです。