ロックは効いているのに件数だけ二重になる。数えていたのは書き込みではなく「書くつもり」だった

結論

FOR UPDATEで直列化していても、サマリの件数を「分類した意図」の長さで数えていると、行は正しく1行のままサマリの合計だけが2になることがある。 TVの死活監視チェッカ（runTvLivenessCheck）は、2つの実行が同時に発火するとnewlyDownの合計が1のはずが2になるflakyなテストを抱えていた。DB側の未解決行は親行のFOR UPDATEロックで常に1行に保たれており、行数のアサート自体は安定してpassしていた。壊れていたのはサマリの数え方で、実際にINSERTした件数ではなく「down対象として分類した件数」を返していたため、ロックの結果INSERTを見送った側の分類まで1件として計上されていた。

発端

このチェッカは、TVごとの生死を判定する純関数classifyTvLivenessと、その判定結果をDBへ反映するapplyTransitionsに分かれている。呼び出し元のrunTvLivenessCheckは、判定結果の長さをそのままサマリとして返していた。

// 修正前相当
const classification = classifyTvLiveness(states, now, thresholds);
await applyTransitions(tx, classification);

return {
  scanned: states.length,
  newlyDown: classification.newlyDown.length,
  recovered: classification.recovered.length,
};

classification.newlyDownは「down状態に遷移すべきと判定されたTVの一覧」であり、実際にDBへ書き込まれた件数ではない。2接続が同時に走るテストで、r1.newlyDown + r2.newlyDown === 1というアサートが不安定にfailしていた。

原因

classifyTvLivenessによる分類は、loadDeviceStatesが事前に読み取ったスナップショットに基づく。2本のチェッカ実行が同時に発火すると、どちらも「このTVはまだ未解決のdowntime行を持っていない」という同じ状態を読み、両方が同一TVをnewlyDownと分類してしまう。

実際のDB書き込みはapplyTransitions側で直列化されている。down遷移のINSERT前に、親のtv_devices行をFOR UPDATEでロックする。

// FOR UPDATE による直列化点（applyTransitions内）
await tx
  .select({ deviceId: tvDevices.deviceId })
  .from(tvDevices)
  .where(eq(tvDevices.deviceId, down.deviceId))
  .for("update");

const open = await tx
  .select({ id: tvDeviceDowntime.id })
  .from(tvDeviceDowntime)
  .where(
    and(eq(tvDeviceDowntime.deviceId, down.deviceId), isNull(tvDeviceDowntime.recoveredAt)),
  );
if (open.length > 0) {
  // 別チェッカ実行が先に INSERT 済み → 状態フラグだけ揃えて二重計上しない。
  await tx
    .update(tvDevices)
    .set({ alertState: "down", updatedAt: new Date() })
    .where(eq(tvDevices.deviceId, down.deviceId));
  continue;
}

未解決のdowntime行を持つ親TVの行が常にFK経由で1行存在するため、初回down（未解決行がまだ0件の状態）でもロック対象が空にならず、2本目の実行はロック解放を待って再走査し、未解決行が既に存在することを見てINSERTをスキップする。ここまではDBの整合性としては正しい。

問題は、このスキップされた側（敗者）のdownもまだclassification.newlyDownという配列には含まれたままだったことだ。サマリをclassification.newlyDown.lengthで数えると、勝者・敗者の両方が1件ずつ、合計2件として計上されてしまう。DBに書き込まれた行は1行だけなのに、チェッカが返すサマリの数字だけが実態と食い違っていた。

直し方

サマリの件数を、分類の長さではなくapplyTransitionsが実際に書き込んだ件数に変えた。

// 修正後（applyTransitions）
async function applyTransitions(
  tx: TenantTx,
  classification: TvLivenessClassification,
): Promise<{ newlyDown: number; recovered: number }> {
  let newlyDown = 0;
  let recovered = 0;
  for (const down of classification.newlyDown) {
    // ...FOR UPDATE ロック、未解決行の再確認...
    if (open.length > 0) {
      // ...状態フラグだけ揃えて continue（非計上）
      continue;
    }
    await tx.insert(tvDeviceDowntime).values({ /* ... */ });
    // ...
    newlyDown += 1; // 実際に INSERT した時のみ計上（スキップ経路は continue 済みで非計上）
  }

  for (const rec of classification.recovered) {
    const closed = await tx
      .update(tvDeviceDowntime)
      .set({ /* recoveredAt, durationSec など */ })
      .where(and(eq(tvDeviceDowntime.deviceId, rec.deviceId), isNull(tvDeviceDowntime.recoveredAt)))
      .returning({ id: tvDeviceDowntime.id });
    // ...
    if (closed.length > 0) {
      recovered += 1; // 実際に未解決行を締めた時のみ計上
    }
  }

  return { newlyDown, recovered };
}

down側はINSERTを実際に実行できた時だけnewlyDownをインクリメントし、ロックの結果スキップした経路（continue）ではインクリメントしない。recover側も対称に直した。UPDATEの条件が「未解決行のみ」なので、別のチェッカ実行が先に同じ行を締めていれば実際の更新行数は0になる。ここで.returning()を使い、戻ってきた行が1件以上あった場合だけrecoveredを増やすようにした。呼び出し元のrunTvLivenessCheckは、この実書込件数をそのままサマリとして返すだけになった。

// 修正後（runTvLivenessCheck）
return {
  scanned: states.length,
  ...(await applyTransitions(tx, classification)),
};

これで勝者側が1・敗者側が0となり、タイミングに依存せず合計は常に1に決定論化された。スキーマやマイグレーションの変更は不要で、既存のFOR UPDATEによる直列化はそのままに、戻り値の数え方だけを直している。

まとめ

ロックが効いているかどうかと、サマリの数字が正しいかどうかは、別の層の問題だった。今回のロックはDBの行を1行に保つという仕事を最初から正しく果たしており、崩れていたのは「何件処理したことにするか」を判定結果の長さで数えるという、ロックとは別の集計コードだった。同時実行を伴う処理でサマリや戻り値の件数を返すときは、「何をしようとしたか（分類・意図）」の数ではなく、「実際に何が起きたか（書き込み・更新の実績）」の数を数えるという区別を、常にどちらの数を返しているのか意識して書く必要がある。