test(relay): check invariant 4 (one demotion winner) in the step-5 simulation (#28696)

* test(relay): check invariant 4 (one demotion winner) in the step-5 simulation Every demotion now names its winner (the newer booking, or the row). The model flags a director that names a different winner for a divergence another already named, and a host whose two seats are each demoted for the other. A fault where directors break level ties by their own cell order shows the check fires. Violations are capped per invariant so one flood cannot hide another class. Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * test(relay): bound unseen duplicates by the silence close and assert all invariants on contested ties A reserve seat the booking director's map had not seen (joined, then half-open within a poll) is never superseded and has no lease to re-check, so it lasts until the cell's control silence close (75 s + one 15 s ping). Invariant 3 now uses that bound for such seats, and the level-divergence config asserts every invariant. Stale maps now judge a divergence another director already demoted, so the config's "contested" counter (divergences judged by two or more directors) is the case invariant 4 guards. The silence-close constant now matches production (90 s, not 100 s). Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * test(relay): raise the floor on map-judged demotes and count only level contests A reconcile whose map has not seen another director's demotion still demotes in production, which raises its epoch floor; the model now does too. The contested counter counts only same-epoch pairs, the only ones invariant 4 can split. The keep-this-seat rule is one helper. Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * refactor(relay): name the simulation's fault-kept seat check for the fault Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * fix(relay): count only confirmed level demotions and every second judgment, and bound duplicates by placement records alone Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f
2026-11-05 19:47:27 +00:00 · 2026-10-14 07:26:27 -04:00
parent eb1ee8f908
commit fa5533e8cd
2 changed files with 166 additions and 21 deletions
@@ -42,8 +42,9 @@ function expectInvariants(report: SimulationReport): void {
 }

 describe('step 5 deterministic simulation', () => {
+  // Invariant 9 (dead-man) never trips here; its own tests below cover it.
  it.each(SEEDS)(
-    'holds invariants 1-3 and 5-8 across mixed versions, lagging polls, restarts and database stalls (seed %i)',
+    'holds invariants 1-8 across mixed versions, lagging polls, restarts and database stalls (seed %i)',
    async (seed) => {
      const report = await runReservePlacementSimulation({
        seed,
@@ -174,6 +175,48 @@ describe('step 5 deterministic simulation', () => {
    120_000
  )

+  // A desktop seated at the same epoch on two cells (a seat half-open within a poll lag, then
+  // placed again by a director that had not seen it): only the row breaks the tie. Short
+  // sessions and all five directors restarting each minute put boot reconciles on those seats.
+  const levelSeats = (seed: number, faults: SimulationConfig['faults']): SimulationConfig => ({
+    seed,
+    durationMs: 16 * 60_000,
+    directors: 5,
+    cells: Array.from({ length: 12 }, (_, index) => ({
+      cellId: `us${String(index).padStart(2, '0')}`,
+      region: 'us-central1' as const,
+      hardCap: 600,
+      ceiling: 480,
+      intakePerSec: 20,
+      mode: 'reserve' as const,
+      preseated: 100
+    })),
+    hosts: 3_000,
+    hostRegion: () => 'us-central1',
+    arrivalPerSec: 20,
+    meanSessionMs: 20_000,
+    reconnectDelayMs: [500, 2_000],
+    duplicateAssignShare: 0.05,
+    halfOpenShare: 0.5,
+    pollLagMs: (director) => [0, 200, 1_000, 2_000, 3_000][director]!,
+    rttMs: rtt,
+    directorRestarts: Array.from({ length: 75 }, (_, index) => ({
+      at: 60_000 + Math.floor(index / 5) * 60_000,
+      director: index % 5
+    })),
+    faults
+  })
+
+  it.each(SEEDS)(
+    'holds invariants 1-8 when directors contest level divergences (seed %i)',
+    async (seed) => {
+      const report = await runReservePlacementSimulation(levelSeats(seed, {}))
+      expect(report.contestedDivergences).toBeGreaterThan(0)
+      expectInvariants(report)
+    },
+    120_000
+  )
+
  // Each fault breaks one rule; the matching invariant must fire, or its check measures nothing.
  const mixed = (faults: SimulationConfig['faults']): SimulationConfig => ({
    seed: 11,
@@ -251,6 +294,7 @@ describe('step 5 deterministic simulation', () => {
    [1, nearFull({ bookPastCap: true })],
    [2, drain({ intakeBurst: 200, directorShare: 0.05 })],
    [3, drainedWithoutSupersede()],
+    [4, levelSeats(11, { localTieBreak: true })],
    [5, mixed({ unguardedLedger: true })],
    // A director restart mid-queue loses its booked joins' rows until its boot reconcile.
    [5, restartedMidQueue()],
@@ -1,5 +1,5 @@
 import { createHash } from 'node:crypto'
-import type { RelayRegion } from '@orca-cloud/relay-contract'
+import { RELAY_PROTOCOL_LIMITS, type RelayRegion } from '@orca-cloud/relay-contract'
 import { CellReserveBook } from './cell-reserve-book.js'
 import { CELL_INTAKE_BURST, CELL_RESERVE_TTL_MS, type ReserveOutcome } from './cell-reserve-contract.js'
 import {
@@ -12,7 +12,7 @@ import {

 // A seeded discrete-event model of step 5 that drives the real CellReserveBook and
 // ReservePlacer. Everything else (sockets, the database, desktops) is modelled. The checks
-// are the design's invariants 1-8; a violation is recorded, never thrown, so one run reports
+// are the design's invariants 1-9; a violation is recorded, never thrown, so one run reports
 // every class it hit.

 export function seededRandom(seed: number): () => number {
@@ -98,8 +98,8 @@ const CONTROL_RENEWAL_MS = 25_000
 const RECONCILE_SETTLE_MS = 30_000
 // A drained seat lingers this long (its splices finishing) unless superseded first.
 const DRAIN_GRACE_MS = 30_000
-// A cell closes a control silent for 75 s, checked once per 25 s ping.
-const CONTROL_SILENCE_CLOSE_MS = 100_000
+// A cell closes a control silent for 75 s, checked once per ping (host-session-registry.ts heartbeat).
+const CONTROL_SILENCE_CLOSE_MS = 75_000 + RELAY_PROTOCOL_LIMITS.controlPingIntervalMs
 // Longer than the slowest duplicate: a half-open seat's silence close.
 const QUIESCE_MS = 150_000

@@ -127,6 +127,8 @@ export type SimulationFaults = {
  noRawFeedMode?: boolean
  unguardedLedger?: boolean
  noEpochFloor?: boolean
+  // On a level divergence each director keeps the seat its own cell order prefers, not the row's.
+  localTieBreak?: boolean
 }

 // relay_cell_admit_effective: each cell writes its raw mode every 15 s (cell-admit-effective-writer.ts),
@@ -423,6 +425,11 @@ export type SimulationReport = {
  supersedes: number
  directorRestarts: number
  rowAheadDemotions: number
+  // Demotions whose winner is level with the loser (same epoch, another cell): the tie-break's cases.
+  levelDemotions: number
+  // Level divergences (same epoch, two cells) two or more directors judged: the only cases where
+  // invariant 4 can break, since a higher epoch wins for every director.
+  contestedDivergences: number
  // Row-ahead demotions per demoted host: a repair that converges needs one.
  rowAheadDemotionsPerHost: { max: number; p99: number }
  deadManTrips: number
@@ -503,8 +510,12 @@ export async function runReservePlacementSimulation(
  const random = seededRandom(config.seed)
  const clock = new Clock()
  const violations: SimulationViolation[] = []
+  // Capped per invariant, so a flood of one class cannot hide another.
+  const violationCounts = new Map<number, number>()
  const violate = (invariant: number, detail: string): void => {
-    if (violations.length < 50) violations.push({ invariant, at: clock.now, detail })
+    const count = violationCounts.get(invariant) ?? 0
+    violationCounts.set(invariant, count + 1)
+    if (count < 50) violations.push({ invariant, at: clock.now, detail })
  }
  const faults = config.faults ?? {}
  const database = new SimDatabase(violate, faults)
@@ -550,6 +561,8 @@ export async function runReservePlacementSimulation(
    supersedes: 0,
    directorRestarts: 0,
    rowAheadDemotions: 0,
+    levelDemotions: 0,
+    contestedDivergences: 0,
    rowAheadDemotionsPerHost: { max: 0, p99: 0 },
    deadManTrips: 0,
    reserveCallsOnTrippedCells: 0,
@@ -729,19 +742,75 @@ export async function runReservePlacementSimulation(
  // Supersede at placement: the director that books a host on `cellId` tells the old seats its
  // map holds on reserve cells to go. Each cell checks epoch and join itself.
  function supersede(director: SimDirector, host: string, cellId: string, epoch: number): void {
+    const held = new Set<string>()
+    for (const [seatCellId, view] of director.views) {
+      const seat = view.seats.get(host)
+      if (seat) held.add(`${seatCellId}\u0000${seat.joinedAt}`)
+    }
+    heldAtBooking.set(`${host}\u0000${cellId}\u0000${epoch}`, held)
    if (faults.skipSupersede) return
    for (const [seatCellId, view] of director.views) {
      const seat = view.seats.get(host)
      if (!seat || seat.demoted || seatCellId === cellId || seat.epoch >= epoch) continue
      if (!view.placement.reserve) continue
-      const loser = cellById.get(seatCellId)!
      report.supersedes += 1
-      clock.schedule(config.rttMs(loser.region), () =>
-        loser.demote(host, seat.epoch, seat.joinedAt, () => onSeatClosed(host, seatCellId, seat.epoch))
-      )
+      demoteFor(director, host, seatCellId, seat, { cellId, epoch })
    }
  }

+  // The seats the booking director's map held when it placed each booking: only those it supersedes.
+  const heldAtBooking = new Map<string, Set<string>>()
+
+  // Invariant 4: each demotion names its winner (the newer booking, or the row). Every director
+  // names the same one per divergence, so no two seats of a host are each demoted for the other.
+  const namedWinners = new Map<string, { won: string; by: string }>()
+  const contested = new Set<string>()
+  const demotedFor = new Map<string, string>()
+  type JudgedSeat = { cellId: string; epoch: number }
+  function judge(director: SimDirector, host: string, loserSeat: JudgedSeat, wonSeat: JudgedSeat): void {
+    const loser = `${loserSeat.cellId}@${loserSeat.epoch}`
+    const won = `${wonSeat.cellId}@${wonSeat.epoch}`
+    const divergence = `${host}\u0000${[loser, won].sort().join('\u0000')}`
+    const named = namedWinners.get(divergence)
+    if (named === undefined) {
+      namedWinners.set(divergence, { won, by: director.id })
+      return
+    }
+    // A second director judged it, agreeing or not: the tie-break was exercised.
+    if (named.by !== director.id && loserSeat.epoch === wonSeat.epoch && !contested.has(divergence)) {
+      contested.add(divergence)
+      report.contestedDivergences += 1
+    }
+    if (named.won !== won) {
+      violate(4, `${director.id} names ${won} over ${loser} for ${host}; ${named.by} named ${named.won}`)
+    }
+  }
+  function demoteFor(
+    director: SimDirector,
+    host: string,
+    cellId: string,
+    seat: { epoch: number; joinedAt: number },
+    winner: { cellId: string; epoch: number }
+  ): void {
+    const loser = `${cellId}@${seat.epoch}`
+    const won = `${winner.cellId}@${winner.epoch}`
+    judge(director, host, { cellId, epoch: seat.epoch }, winner)
+    const cell = cellById.get(cellId)!
+    clock.schedule(config.rttMs(cell.region), () => {
+      const live = cell.seats.get(host)
+      const wasLive = live !== undefined && !live.demoted
+      cell.demote(host, seat.epoch, seat.joinedAt, () => onSeatClosed(host, cellId, seat.epoch))
+      if (!wasLive || !live.demoted) return
+      // Only a live seat demoted for a live level winner is a same-epoch duplicate resolved.
+      const winnerSeat = cellById.get(winner.cellId)?.seats.get(host)
+      if (winner.epoch === seat.epoch && winnerSeat?.epoch === winner.epoch && !winnerSeat.demoted) {
+        report.levelDemotions += 1
+      }
+      demotedFor.set(`${host}\u0000${loser}`, won)
+      if (demotedFor.get(`${host}\u0000${won}`) === loser) violate(4, `${host}: ${loser} and ${won} each demoted for the other`)
+    })
+  }
+
  function onSeatClosed(hostId: string, cellId: string, epoch: number): void {
    const host = hostById.get(hostId)!
    if (host.current?.cellId === cellId && host.current.epoch === epoch) {
@@ -766,6 +835,18 @@ export async function runReservePlacementSimulation(
  const demotedBehindRow = new Map<string, number>()
  // A host demoted behind its row: its next answer, from any director, is that row or above it.
  const reassignAbove = new Map<string, { cellId: string; epoch: number }>()
+  // Only under localTieBreak: even directors keep the lower cell id of a level divergence, odd ones the higher.
+  function faultKeepsThisSeat(
+    director: SimDirector,
+    cellId: string,
+    epoch: number,
+    row: { cellId: string; epoch: number }
+  ): boolean {
+    return !!faults.localTieBreak && row.epoch === epoch && (cellId < row.cellId) === (director.index % 2 === 0)
+  }
+  function raiseRowFloor(director: SimDirector, host: string, epoch: number): void {
+    director.rowFloors.set(host, Math.max(epoch, director.rowFloors.get(host) ?? 0))
+  }
  function reconcileLedger(director: SimDirector): void {
    if (!database.up || director.old) return
    for (const [cellId, view] of director.views) {
@@ -775,13 +856,32 @@ export async function runReservePlacementSimulation(
        const row = database.rows.get(host)
        const live = cellById.get(cellId)!.seats.get(host)
        const diverged = row && (row.epoch > seat.epoch || (row.epoch === seat.epoch && row.cellId !== cellId))
+        // A map that has not yet seen another director's demotion still judges the seat; the
+        // cell drops the demote, but the winner it names must be the same.
+        // relay-server.ts raises the floor on every demote, so a dropped one raises it too.
+        if (row && diverged && live && live.joinedAt === seat.joinedAt && live.demoted && !faults.noRowAheadDemote) {
+          const thisSeat = { cellId, epoch: seat.epoch }
+          const rowSeat = { cellId: row.cellId, epoch: row.epoch }
+          if (faultKeepsThisSeat(director, cellId, seat.epoch, row)) judge(director, host, rowSeat, thisSeat)
+          else {
+            judge(director, host, thisSeat, rowSeat)
+            raiseRowFloor(director, host, row.epoch)
+          }
+        }
        if (row && diverged && live && live.joinedAt === seat.joinedAt && !live.demoted) {
          const key = `${director.id}\u0000${host}\u0000${cellId}\u0000${seat.joinedAt}`
          if (behindRow.has(key)) violate(8, `${host}@${seat.epoch} on ${cellId} still behind row @${row.epoch}`)
          behindRow.set(key, clock.now)
          if (faults.noRowAheadDemote) continue
+          if (faultKeepsThisSeat(director, cellId, seat.epoch, row)) {
+            const rowSeat = director.views.get(row.cellId)?.seats.get(host)
+            if (rowSeat && rowSeat.epoch === row.epoch && !rowSeat.demoted) {
+              demoteFor(director, host, row.cellId, rowSeat, { cellId, epoch: seat.epoch })
+            }
+            continue
+          }
          report.rowAheadDemotions += 1
-          director.rowFloors.set(host, Math.max(row.epoch, director.rowFloors.get(host) ?? 0))
+          raiseRowFloor(director, host, row.epoch)
          rowAheadDemotionsByHost.set(host, (rowAheadDemotionsByHost.get(host) ?? 0) + 1)
          // A director that had not seen the leave may re-place level with the row once; a later
          // divergence (a newer row) may demote the host again. One row never takes a third.
@@ -790,10 +890,7 @@ export async function runReservePlacementSimulation(
          demotedBehindRow.set(sameRow, behindThisRow)
          if (behindThisRow > 2) violate(8, `${host} demoted ${behindThisRow} times behind its row ${row.cellId}@${row.epoch}`)
          reassignAbove.set(host, { ...row })
-          const cell = cellById.get(cellId)!
-          clock.schedule(config.rttMs(cell.region), () =>
-            cell.demote(host, seat.epoch, seat.joinedAt, () => onSeatClosed(host, cellId, seat.epoch))
-          )
+          demoteFor(director, host, cellId, seat, row)
          continue
        }
        database.upsert(host, cellId, seat.epoch)
@@ -1161,8 +1258,9 @@ export async function runReservePlacementSimulation(
      report.maxOccupancyShare[cell.cellId] = Math.max(report.maxOccupancyShare[cell.cellId] ?? 0, share)
    }
  }
-  // A booking supersedes the old seat within a round trip; a duplicate nobody books past (a
-  // half-open seat while today's path places the host) lasts until that seat's silence close.
+  // A booking supersedes the old seats its director's map held within a round trip. Any other
+  // duplicate (a half-open seat today's path placed past, or one the booking map had not yet
+  // seen) has no lease to re-check and lasts until that seat's silence close.
  const supersedeBound =
    Math.max(...directors.map((director) => config.pollLagMs(director.index))) +
    2 * Math.max(...cells.map((cell) => config.rttMs(cell.region))) +
@@ -1186,13 +1284,16 @@ export async function runReservePlacementSimulation(
    for (const [host, seats] of seatsByHost) {
      if (seats.length < 2) continue
      const newest = Math.max(...seats.map((entry) => entry.seat.epoch))
-      // A seat outranked by a booking is superseded within the booking director's poll lag and
-      // a round trip; anything else (today's path placed the host) waits for the silence close.
-      const booked = seats.some((entry) => entry.seat.epoch === newest && entry.seat.reservedBy !== undefined)
+      // By the newest seats' placement records, whatever their join says: no record, no supersede.
+      const bookings = seats
+        .filter((entry) => entry.seat.epoch === newest)
+        .flatMap((entry) => heldAtBooking.get(`${host}\u0000${entry.cell.cellId}\u0000${newest}`) ?? [])
+      const supersededBy = (cellId: string, joinedAt: number): boolean =>
+        bookings.some((held) => held.has(`${cellId}\u0000${joinedAt}`))
      for (const { cell, seat } of seats) {
        if (cell.mode !== 'reserve' || seat.epoch === newest) continue
        // A leased seat goes by its renewal, as today.
-        const bound = booked && !seat.leased ? supersedeBound : duplicateBound
+        const bound = !seat.leased && supersededBy(cell.cellId, seat.joinedAt) ? supersedeBound : duplicateBound
        const key = `${host}\u0000${cell.cellId}\u0000${seat.joinedAt}`
        live.add(key)
        const since = outrankedSince.get(key) ?? clock.now