完了したはずの Gemini ジョブが『実行中』に戻った — Webhook の順序入れ替わりを止める単調適用の設計

// オペレーションのライフサイクルに単調増加する順位を与える。
// 数が大きいほど「進んだ」状態。終端(SUCCEEDED/FAILED/CANCELLED)は同位の最大値。
const STATE_RANK: Record<string, number> = {
  PENDING: 0,
  RUNNING: 1,
  SUCCEEDED: 2,
  FAILED: 2,
  CANCELLED: 2,
};
 
const TERMINAL = new Set(["SUCCEEDED", "FAILED", "CANCELLED"]);
 
function rankOf(state: string): number {
  const r = STATE_RANK[state];
  if (r === undefined) {
    // 未知の状態は最下位扱いにし、既存の進行を絶対に巻き戻させない
    return -1;
  }
  return r;
}

type OpState = {
  name: string;          // オペレーション名（正規化済み）
  state: string;         // 現在の状態
  updateTime: string;    // サーバー側の更新時刻(RFC3339)
  version: number;       // 楽観ロック用のバージョン
};
 
type OpEvent = {
  name: string;
  state: string;
  updateTime: string;    // イベントが表すサーバー側更新時刻
};
 
type ApplyResult =
  | { applied: true; next: OpState }
  | { applied: false; reason: string };
 
function applyEvent(cur: OpState, ev: OpEvent): ApplyResult {
  // 1) 既に終端なら、非終端イベントは常に無視する（巻き戻り防止の主役）
  if (TERMINAL.has(cur.state) && !TERMINAL.has(ev.state)) {
    return { applied: false, reason: "regress-from-terminal" };
  }
 
  const curRank = rankOf(cur.state);
  const evRank = rankOf(ev.state);
 
  // 2) 順位が後退するイベントは捨てる
  if (evRank < curRank) {
    return { applied: false, reason: "lower-rank" };
  }
 
  // 3) 同位のときは updateTime の新しい方だけを通す（フェンシング）
  if (evRank === curRank && ev.updateTime <= cur.updateTime) {
    return { applied: false, reason: "stale-or-equal-update-time" };
  }
 
  // 4) ここまで来たら前進。次の状態を作る
  return {
    applied: true,
    next: {
      ...cur,
      state: ev.state,
      updateTime: ev.updateTime,
      version: cur.version + 1,
    },
  };
}

import Database from "better-sqlite3";
 
const db = new Database("ops.db");
db.exec(`
  CREATE TABLE IF NOT EXISTS operations (
    name        TEXT PRIMARY KEY,
    state       TEXT NOT NULL,
    update_time TEXT NOT NULL,
    version     INTEGER NOT NULL
  );
`);
 
// 競合時はリトライする。リデューサーは純粋なので何度走らせても安全。
function ingest(ev: OpEvent, maxRetry = 4): ApplyResult {
  for (let attempt = 0; attempt < maxRetry; attempt++) {
    const cur = db
      .prepare("SELECT name, state, update_time AS updateTime, version FROM operations WHERE name = ?")
      .get(ev.name) as OpState | undefined;
 
    if (!cur) {
      // 初見のオペレーション。条件付き INSERT で競合を検出する
      const ins = db
        .prepare(
          "INSERT OR IGNORE INTO operations(name, state, update_time, version) VALUES(?, ?, ?, 1)"
        )
        .run(ev.name, ev.state, ev.updateTime);
      if (ins.changes === 1) {
        return { applied: true, next: { name: ev.name, state: ev.state, updateTime: ev.updateTime, version: 1 } };
      }
      continue; // 同時に誰かが INSERT した。読み直してリトライ
    }
 
    const res = applyEvent(cur, ev);
    if (!res.applied) return res; // 後退イベント。捨てて終わり（成功扱い）
 
    // version 一致を条件に更新。負ければ誰かが先に進めた → リトライ
    const upd = db
      .prepare(
        "UPDATE operations SET state = ?, update_time = ?, version = ? WHERE name = ? AND version = ?"
      )
      .run(res.next.state, res.next.updateTime, res.next.version, ev.name, cur.version);
 
    if (upd.changes === 1) return res;
    // 競合した。ループ先頭へ戻って読み直す
  }
  return { applied: false, reason: "max-retry-exceeded" };
}

// Webhook 受信ハンドラ（署名検証は別記事の方式を踏襲）
app.post("/gemini/webhook", verifySignature, (req, res) => {
  const ev = normalizeEvent(req.body); // name 正規化・updateTime 抽出
  const result = ingest(ev);
  if (result.applied) enqueueSideEffect(result.next); // 公開・課金などは冪等な受け口へ
  res.sendStatus(200); // 後退イベントを捨てた場合も 200。再配送を誘発しない
});
 
// 照合ポーラー（低頻度）。API の現在状態を「イベント」に変換して同じ口へ流す
async function reconcileOnce(name: string) {
  const op = await genaiClient.operations.get({ name });
  const ev: OpEvent = {
    name: normalizeName(op.name),
    state: op.done ? terminalStateOf(op) : "RUNNING",
    updateTime: op.metadata?.updateTime ?? new Date(0).toISOString(),
  };
  ingest(ev); // 単調適用を通るので、新鮮な Webhook を古い照合で塗り潰さない
}