Gemini API の自動パイプラインで、静かに消える失敗ジョブを取りこぼさない設計

RETRYABLE_STATUS = {429, 500, 502, 503, 504}
PERMANENT_STATUS = {400, 403, 404, 422}
 
 
def classify_failure(status_code: int, message: str) -> str:
    """失敗を 'retryable' / 'permanent' / 'unknown' に分類します。"""
    text = (message or "").upper()
 
    # 入力内容そのものが原因の失敗は、何度投げても同じ結果になります。
    if status_code in PERMANENT_STATUS:
        return "permanent"
    if "INVALID_ARGUMENT" in text or "FAILED_PRECONDITION" in text:
        return "permanent"
    if "SAFETY" in text or "BLOCKED" in text:
        return "permanent"
 
    # 一時的な負荷・ネットワーク起因は、時間を置けば回復し得ます。
    if status_code in RETRYABLE_STATUS:
        return "retryable"
    if "DEADLINE_EXCEEDED" in text or "UNAVAILABLE" in text:
        return "retryable"
 
    return "unknown"

import random
 
 
class ApiError(Exception):
    def __init__(self, status_code: int, message: str):
        super().__init__(message)
        self.status_code = status_code
        self.message = message
 
 
def run_with_dead_letter(job_id: str, payload: dict, model: str,
                         call_model, max_attempts: int = 4):
    """モデル呼び出しをリトライし、恒久失敗・リトライ尽きは退避します。
 
    call_model(payload) -> dict を呼ぶ前提です。失敗時は ApiError を送出します。
    """
    attempt = 0
    while True:
        attempt += 1
        try:
            return call_model(payload)
        except ApiError as exc:
            kind = classify_failure(exc.status_code, exc.message)
 
            if kind == "permanent":
                park(job_id, payload, model,
                     error_class="permanent:%d" % exc.status_code,
                     error_message=exc.message, attempts=attempt)
                return None
 
            if attempt >= max_attempts:
                park(job_id, payload, model,
                     error_class="exhausted:%d" % exc.status_code,
                     error_message=exc.message, attempts=attempt)
                return None
 
            # retryable / unknown は控えめなバックオフで様子を見ます。
            backoff = min(2 ** attempt, 30) + random.uniform(0, 1)
            time.sleep(backoff)

def replay(call_model, limit: int = 20, dry_run: bool = True,
           path: str = DLQ_PATH) -> dict:
    """parked のジョブを再投入します。原因修正後に手動で叩く前提です。"""
    con = sqlite3.connect(path)
    con.row_factory = sqlite3.Row
    rows = con.execute(
        "SELECT * FROM dead_letter WHERE status = 'parked' "
        "ORDER BY last_seen ASC LIMIT ?",
        (limit,),
    ).fetchall()
 
    recovered, still_failing = 0, 0
    for row in rows:
        payload = json.loads(row["payload_json"])
        if dry_run:
            print("[dry-run] would replay %s (%s)" % (row["job_id"], row["error_class"]))
            continue
        try:
            call_model(payload)
            con.execute(
                "UPDATE dead_letter SET status = 'recovered', last_seen = ? "
                "WHERE job_id = ?",
                (time.time(), row["job_id"]),
            )
            recovered += 1
        except ApiError:
            still_failing += 1
 
    con.commit()
    con.close()
    return {"scanned": len(rows), "recovered": recovered,
            "still_failing": still_failing, "dry_run": dry_run}

def dlq_summary(path: str = DLQ_PATH) -> None:
    """デッドレターの内訳を1日1回ログに出します。"""
    con = sqlite3.connect(path)
    rows = con.execute(
        "SELECT error_class, COUNT(*) AS n FROM dead_letter "
        "WHERE status = 'parked' GROUP BY error_class ORDER BY n DESC"
    ).fetchall()
    con.close()
 
    if not rows:
        print("dead-letter: clean (0 parked)")
        return
    total = sum(n for _, n in rows)
    print("dead-letter: %d parked" % total)
    for error_class, n in rows:
        print("  %4d  %s" % (n, error_class))