feat(relay): the director that booked a join writes its row about a second later (off) (#27048)

Rebased onto main as one commit after its parent squash-merged.

cloud/apps/relay/src/floating-promise-census.test.ts

@@ -14,6 +14,7 @@ const NEVER_REJECTS = new Map([
   ['cell-heartbeat-client.ts#send', 'one try/catch around the whole send'],
   ['cell-reserve-client.ts#send', 'one try/catch around the request; resolves every item'],
   ['cell-reserve-client.ts#demote', 'one try/catch around the request'],
+  ['reserve-ledger-writer.ts#flush', 'one try/catch around the write; retries with backoff'],
   ['control-renewal-batch.ts#flush', 'rejects the waiters, never itself'],
   ['regional-rehome-worker.ts#run', 'one try/catch around the whole poll'],
   ['relay-control-flag-channel.ts#poll', 'one catch on the whole read']

cloud/apps/relay/src/relay-server.ts

@@ -51,6 +51,11 @@ import { CellReserveClient } from './cell-reserve-client.js'
 import { googleMetadataIdentityToken, reusedIdentityToken } from './google-metadata-identity-token.js'
 import { ReserveAssignment } from './reserve-assignment.js'
 import { ReservePlacer } from './reserve-placement.js'
+import {
+  reconcileReserveLedger,
+  RESERVE_LEDGER_RECONCILE_MS,
+  ReserveLedgerWriter
+} from './reserve-ledger-writer.js'
 import { ProcessQueuedByteBudget } from './splice-forwarder.js'
 
 // A malformed percent-escape in the request target must be a client error, never a URIError
@@ -185,6 +190,11 @@ export function createRelayServer(
   // Step 5 on directors: needs the map and the rehome credential the cells verify.
   const directorId = randomUUID()
   const rehomeAudience = config.rehomeAudience
+  // Only a director that books writes rows, and only for the joins it booked.
+  const reserveLedger =
+    config.role === 'director' && shadowSeatPoller && config.reservePlacement === 'on'
+      ? new ReserveLedgerWriter(observedDatabase, { now: options.now })
+      : undefined
   const reservePlacement =
     config.role === 'director' &&
     shadowSeatPoller &&
@@ -204,6 +214,7 @@ export function createRelayServer(
             )
           }),
           readRow: async (identity) => await assignments.hostWhereabouts(identity),
+          databaseBusy: () => readRelayDatabasePoolPressure(database).databasePoolWaiting > 0,
           now: options.now,
           random: options.random
         })
@@ -228,6 +239,68 @@ export function createRelayServer(
         }, 60_000)
       : null
   offWithReserveTimer?.unref()
+  if (reserveLedger && shadowSeatPoller) {
+    const directory = shadowSeatPoller.directory
+    directory.onBookedJoin = (join) => {
+      if (join.reservedBy !== directorId) return
+      reserveLedger.enqueue({
+        userId: join.userId,
+        relayHostId: join.relayHostId,
+        cellId: join.cellId,
+        epoch: join.epoch
+      })
+    }
+    reserveLedger.start()
+    const reconcile = (): void => {
+      const now = (options.now ?? Date.now)()
+      void reconcileReserveLedger({
+        writer: reserveLedger,
+        seats: () =>
+          directory
+            .cellIds()
+            .filter((cellId) => directory.admitModeOf(cellId) === 'reserve')
+            .flatMap((cellId) =>
+              directory.seatsOnCell(cellId).map(({ userId, relayHostId, seat }) => ({
+                userId,
+                relayHostId,
+                cellId,
+                epoch: seat.epoch,
+                joinedAt: seat.joinedAt
+              }))
+            ),
+        demote: (seat, row) => {
+          reservePlacement?.demoteSeat(seat)
+          // The re-assign mints above the row, so the mirror can take it.
+          reservePlacement?.raiseEpochFloor(seat, row.epoch)
+          console.warn(
+            JSON.stringify({
+              event: 'orca_relay_reserve_ledger_seat_behind_row',
+              cellId: seat.cellId,
+              epoch: seat.epoch,
+              rowEpoch: row.epoch
+            })
+          )
+        },
+        now
+      })
+        .catch(() => console.warn('[orca-relay] reserve ledger reconcile deferred'))
+        .finally(schedule)
+    }
+    // Jittered, so five directors do not reconcile together.
+    const schedule = (): void => {
+      const random = options.random ?? Math.random
+      setTimeout(reconcile, RESERVE_LEDGER_RECONCILE_MS + random() * 60_000).unref?.()
+    }
+    // A restart loses the queue: the first run goes as soon as the map is complete.
+    const clock = options.now ?? Date.now
+    const bootedAt = clock()
+    const firstRun = setInterval(() => {
+      if (!directory.isComplete() && clock() - bootedAt < RESERVE_LEDGER_RECONCILE_MS) return
+      clearInterval(firstRun)
+      reconcile()
+    }, 1_000)
+    firstRun.unref?.()
+  }
   const configuredConnectionLimits =
     config.connectionHardCap === undefined
       ? null

cloud/apps/relay/src/reserve-assignment.test.ts

@@ -1,6 +1,12 @@
 import { describe, expect, it, vi } from 'vitest'
 import type { CellReserveClient } from './cell-reserve-client.js'
-import { ReserveAssignment, ReservePaceError, RESERVE_STARTUP_GATE_MS } from './reserve-assignment.js'
+import {
+  ReserveAssignment,
+  ReservePaceError,
+  RESERVE_ROW_FLOOR_TTL_MS,
+  RESERVE_WRONG_CELL_ROW_READ_MS,
+  RESERVE_STARTUP_GATE_MS
+} from './reserve-assignment.js'
 import { ReservePlacer, type ReserveAttempt } from './reserve-placement.js'
 import { ShadowSeatDirectory, type SeatFeedCell, type SeatFeedResponse } from './shadow-seat-directory.js'
 
@@ -43,6 +49,7 @@ function setup(
     feeds?: SeatFeedResponse[]
     reserve?: (cellId: string, sticky: boolean) => ReserveAttempt
     row?: (identity: { userId: string; relayHostId: string }) => Promise<{ cellId: string; assignmentEpoch: number } | null>
+    databaseBusy?: () => boolean
     mode?: 'off' | 'dry-run' | 'on'
     startedAt?: number
   } = {}
@@ -78,6 +85,7 @@ function setup(
     // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: see above.
     client: client as unknown as CellReserveClient,
     readRow: options.row ?? (async () => null),
+    ...(options.databaseBusy ? { databaseBusy: options.databaseBusy } : {}),
     now: () => now.value,
     random: () => 0.25,
     log: () => undefined
@@ -88,7 +96,7 @@ function setup(
   for (const response of options.feeds ?? cells.map((cell) => feed(cell.cellId))) {
     directory.apply(response.cellId, { ...response, full: response.full ?? [] }, now.value - 1, now.value - 1)
   }
-  return { assignment, directory, reserved, demoted, now }
+  return { assignment, directory, reserved, demoted, now, placer }
 }
 
 function seated(epoch: number, joinedAt = 0) {
@@ -162,6 +170,21 @@ describe('reserve assignment on a director', () => {
     expect(plan).toMatchObject({ kind: 'answer', lane: 'sticky', assignment: { cellId: 'c1', assignmentEpoch: 7 } })
   })
 
+  it('after the reconcile demotes a seat behind its row, mints above the row and never answers that seat', async () => {
+    const { assignment, directory, reserved, now } = setup({ feeds: [feed('c1', { full: [seated(7)] }), feed('c2')] })
+    // The row names another cell at epoch 8; the mirror would refuse a re-assign at 8.
+    assignment.raiseEpochFloor(HOST, 8)
+    const plan = await assignment.plan(HOST, { reconnect: true, region: US })
+    expect(plan).toMatchObject({ kind: 'answer', lane: 'placement', assignment: { assignmentEpoch: 9 } })
+    expect(reserved.every((call) => !call.sticky && call.epoch === 9)).toBe(true)
+    // The floor lapses, so a host it no longer protects answers from memory again.
+    now.value += RESERVE_ROW_FLOOR_TTL_MS + 1
+    for (const response of [feed('c1', { seq: 2, full: [seated(7)] }), feed('c2', { seq: 2 })]) {
+      directory.apply(response.cellId, response, now.value, now.value)
+    }
+    expect(await assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({ lane: 'sticky' })
+  })
+
   it('never sends a drained or demoted host back, and books it elsewhere above every known epoch', async () => {
     const { assignment, directory, reserved, now } = setup()
     directory.apply(
@@ -231,6 +254,62 @@ describe('reserve assignment on a director', () => {
     })
   })
 
+  it('after a WRONG_CELL leave, any director reads the row and mints above it', async () => {
+    const leftWrongCell = (
+      row: () => Promise<{ cellId: string; assignmentEpoch: number } | null>,
+      databaseBusy?: () => boolean
+    ) => {
+      const harness = setup({ row, ...(databaseBusy ? { databaseBusy } : {}) })
+      // Another director demoted the seat: this one only sees the leave in c1's feed.
+      harness.directory.apply(
+        'c1',
+        feed('c1', {
+          full: undefined,
+          seq: 3,
+          changes: [
+            { seq: 2, kind: 'join', ...HOST, epoch: 7, generation: 1, at: 1 },
+            { seq: 3, kind: 'leave', ...HOST, epoch: 7, generation: 1, closeCode: 4409, at: 2 }
+          ]
+        }),
+        harness.now.value,
+        harness.now.value
+      )
+      return harness
+    }
+    const row = vi.fn(async () => ({ cellId: 'c2', assignmentEpoch: 8 }))
+    const read = leftWrongCell(row)
+    expect(await read.assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({
+      kind: 'answer',
+      lane: 'placement',
+      assignment: { assignmentEpoch: 9 }
+    })
+    expect(row).toHaveBeenCalledTimes(1)
+    // The database down: placed as before the read existed, above what the map knows.
+    const down = leftWrongCell(async () => Promise.reject(new Error('timeout')))
+    expect(await down.assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({
+      kind: 'answer',
+      assignment: { assignmentEpoch: 8 }
+    })
+    // Pool waiters: the read is skipped outright, so a stall cannot stack timed-out reads.
+    const busyRow = vi.fn(async () => ({ cellId: 'c2', assignmentEpoch: 8 }))
+    const busy = leftWrongCell(busyRow, () => true)
+    expect(await busy.assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({
+      kind: 'answer',
+      assignment: { assignmentEpoch: 8 }
+    })
+    expect(busyRow).not.toHaveBeenCalled()
+    // A stalled database: the read gives up after its bound and places as before.
+    vi.useFakeTimers()
+    try {
+      const stalled = leftWrongCell(() => new Promise(() => undefined))
+      const plan = stalled.assignment.plan(HOST, { reconnect: true, region: US })
+      await vi.advanceTimersByTimeAsync(RESERVE_WRONG_CELL_ROW_READ_MS)
+      expect(await plan).toMatchObject({ kind: 'answer', assignment: { assignmentEpoch: 8 } })
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('paces with Retry-After when every reserve cell is out of budget and no database cell exists', async () => {
     const { assignment } = setup({ reserve: () => ({ outcome: 'intake' }) })
     await expect(assignment.plan(HOST, { reconnect: false, region: US })).rejects.toBeInstanceOf(

cloud/apps/relay/src/reserve-assignment.ts

@@ -49,9 +49,32 @@ export type ReservePlan =
       placeFresh?: () => Promise<RelayAssignment | null>
     }
 
+// A host whose seat the ledger reconcile demoted mints above its row for this long, so the
+// re-assign does not collide with the row again (the mirror refuses an equal epoch).
+export const RESERVE_ROW_FLOOR_TTL_MS = 2 * 60 * 60_000
+const RESERVE_ROW_FLOOR_MAX = 100_000
+
+// The WRONG_CELL row read's bound: well under a database stall (5-7 s).
+export const RESERVE_WRONG_CELL_ROW_READ_MS = 1_000
+
+// Rejects after `ms`; the read itself still settles later, and its failure is swallowed.
+async function withinMs<T>(read: Promise<T>, ms: number): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined
+  const timeout = new Promise<never>((_resolve, reject) => {
+    timer = setTimeout(() => reject(new Error('row_read_timeout')), ms)
+  })
+  read.catch(() => undefined)
+  try {
+    return await Promise.race([read, timeout])
+  } finally {
+    clearTimeout(timer)
+  }
+}
+
 export class ReserveAssignment {
   private summary: PlacementSummary = emptySummary()
   private summaryAt: number
+  private readonly rowFloors = new Map<string, { epoch: number; expiresAt: number }>()
 
   constructor(
     private readonly input: {
@@ -63,6 +86,9 @@ export class ReserveAssignment {
       client: CellReserveClient
       // One primary-key read of the host's row: only for a host the map has never seen.
       readRow: (identity: Identity) => Promise<{ cellId: string; assignmentEpoch: number } | null>
+      // True while the pool has waiters: the optional WRONG_CELL read is skipped, so timed-out
+      // reads cannot stack up behind a stall.
+      databaseBusy?: () => boolean
       now?: () => number
       random?: () => number
       log?: (line: string) => void
@@ -130,19 +156,34 @@ export class ReserveAssignment {
       .filter((entry) => now - entry.at <= RESERVE_STICKY_RECENT_MS)
       .at(-1)
     const booking = directory.bookingOf(identity.userId, identity.relayHostId)
+    const rowFloor = this.rowFloor(identity, now)
     const known = [
       ...seats.map((seat) => seat.epoch),
       ...(left ? [left.epoch] : []),
-      ...(booking ? [booking.epoch] : [])
+      ...(booking ? [booking.epoch] : []),
+      ...(rowFloor === undefined ? [] : [rowFloor])
     ]
     let row: { cellId: string; assignmentEpoch: number } | null = null
-    if (known.length === 0) {
+    // A seat closed with WRONG_CELL was demoted (or failed to re-register) because its row is
+    // ahead, which every director sees in that cell's feed: read the row, so the re-assign mints
+    // above it on any director, not just the one that demoted it.
+    const leftWrongCell =
+      left?.closeCode === RELAY_CLOSE_CODE.WRONG_CELL && !(this.input.databaseBusy?.() ?? false)
+    if (known.length === 0 || leftWrongCell) {
       try {
-        row = await this.input.readRow(identity)
+        // Only the never-seen host needs the row; for the WRONG_CELL one it is an optimisation,
+        // bounded so a database stall cannot hold the reconnect (the reconcile repairs a miss).
+        row =
+          known.length === 0
+            ? await this.input.readRow(identity)
+            : await withinMs(this.input.readRow(identity), RESERVE_WRONG_CELL_ROW_READ_MS)
       } catch {
         // A host the map never saw, with the database down: its epoch cannot be minted safely.
-        this.summary.retries.database += 1
-        return { kind: 'retry', retryAfterSeconds: 2, reason: 'database' }
+        // One the map knows places as it did before this read existed.
+        if (known.length === 0) {
+          this.summary.retries.database += 1
+          return { kind: 'retry', retryAfterSeconds: 2, reason: 'database' }
+        }
       }
       if (row) known.push(row.assignmentEpoch)
     }
@@ -158,7 +199,7 @@ export class ReserveAssignment {
     }
 
     if (request.reconnect) {
-      const sticky = await this.stickyFromMemory(identity, seats, left, now)
+      const sticky = await this.stickyFromMemory(identity, seats, left, now, rowFloor ?? 0)
       if (sticky) {
         this.summary.sticky += 1
         return { kind: 'answer', assignment: sticky, lane: 'sticky' }
@@ -183,11 +224,14 @@ export class ReserveAssignment {
     identity: Identity,
     seats: ShadowSeat[],
     left: { cellId: string; epoch: number; incarnation?: string; closeCode?: number } | undefined,
-    now: number
+    now: number,
+    rowFloor: number
   ): Promise<RelayAssignment | null> {
     // Highest epoch first, then the newest join: only the newest grant may be answered from
     // memory, so a stale seat (or a leaver at a lower epoch) is never sent back.
     const newest = Math.max(0, ...seats.map((seat) => seat.epoch), left?.epoch ?? 0)
+    // A seat at or behind a row the reconcile demoted it for is never answered again.
+    if (newest <= rowFloor) return null
     const candidates: Array<{
       cellId: string
       epoch: number
@@ -310,6 +354,42 @@ export class ReserveAssignment {
     return null
   }
 
+  // The ledger reconcile demoted this host's seat behind (or level with) a row at `epoch`.
+  raiseEpochFloor(identity: Identity, epoch: number): void {
+    const key = `${identity.userId}\u0000${identity.relayHostId}`
+    const current = this.rowFloors.get(key)
+    this.rowFloors.delete(key)
+    this.rowFloors.set(key, {
+      epoch: Math.max(epoch, current?.epoch ?? 0),
+      expiresAt: this.now() + RESERVE_ROW_FLOOR_TTL_MS
+    })
+    if (this.rowFloors.size > RESERVE_ROW_FLOOR_MAX) {
+      this.rowFloors.delete(this.rowFloors.keys().next().value!)
+    }
+  }
+
+  private rowFloor(identity: Identity, now: number): number | undefined {
+    const key = `${identity.userId}\u0000${identity.relayHostId}`
+    const floor = this.rowFloors.get(key)
+    if (!floor) return undefined
+    if (floor.expiresAt > now) return floor.epoch
+    this.rowFloors.delete(key)
+    return undefined
+  }
+
+  // Fire-and-forget; the cell checks the seat's epoch and join itself.
+  demoteSeat(seat: { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }): void {
+    const cell = this.input.cells().find((entry) => entry.cellId === seat.cellId)
+    if (!cell) return
+    void this.input.client.demote(cell, {
+      v: 1,
+      userId: seat.userId,
+      relayHostId: seat.relayHostId,
+      epoch: seat.epoch,
+      joinedAt: seat.joinedAt
+    })
+  }
+
   // Supersede at placement: the old seats this director's map holds on reserve cells are told
   // to go. Each cell checks the seat's epoch and join itself, so a stale map only misses.
   private supersede(identity: Identity, seats: readonly ShadowSeat[], cellId: string, epoch: number): void {

cloud/apps/relay/src/reserve-ledger-writer-postgres.test.ts

@@ -0,0 +1,61 @@
+import { randomUUID } from 'node:crypto'
+import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { openRelayDatabase, type RelayDatabase } from './database.js'
+import { reconcileReserveLedger, ReserveLedgerWriter } from './reserve-ledger-writer.js'
+
+// The ledger's statements on real PostgreSQL: the guarded upsert, its counter reset on a
+// move, and the batched row read.
+
+const databaseUrl = process.env.ORCA_RELAY_TEST_POSTGRES_URL
+const describePostgres = databaseUrl ? describe : describe.skip
+const USER = `ledger-pg-${randomUUID().slice(0, 8)}`
+
+describePostgres('reserve ledger writer on PostgreSQL', () => {
+  let database: RelayDatabase
+
+  beforeAll(async () => {
+    database = await openRelayDatabase({ databaseUrl, dataDir: '/tmp' })
+  })
+
+  afterAll(async () => {
+    await database.query(`DELETE FROM relay_assignments WHERE user_id = ?`, [USER])
+    await database.close()
+  })
+
+  const row = async (relayHostId: string) =>
+    (
+      await database.query(
+        `SELECT cell_id, assignment_epoch, reserved_controls FROM relay_assignments
+         WHERE user_id = ? AND relay_host_id = ?`,
+        [USER, relayHostId]
+      )
+    )[0]
+
+  it('writes only a higher epoch, drops a moved row’s counters, and reconciles from the map', async () => {
+    const writer = new ReserveLedgerWriter(database, { log: () => undefined })
+    const host = 'ledgerpghost0001'
+    writer.enqueue({ userId: USER, relayHostId: host, cellId: 'c1', epoch: 5 })
+    await writer.flush()
+    expect(await row(host)).toMatchObject({ cell_id: 'c1', assignment_epoch: '5' })
+    await database.query(`UPDATE relay_assignments SET reserved_controls = 2 WHERE user_id = ?`, [USER])
+    writer.enqueue({ userId: USER, relayHostId: host, cellId: 'c2', epoch: 5 })
+    await writer.flush()
+    expect(await row(host)).toMatchObject({ cell_id: 'c1', assignment_epoch: '5', reserved_controls: '2' })
+    writer.enqueue({ userId: USER, relayHostId: host, cellId: 'c2', epoch: 6 })
+    await writer.flush()
+    expect(await row(host)).toMatchObject({ cell_id: 'c2', assignment_epoch: '6', reserved_controls: '0' })
+    const other = 'ledgerpghost0002'
+    const result = await reconcileReserveLedger({
+      writer,
+      seats: () => [
+        { userId: USER, relayHostId: host, cellId: 'c1', epoch: 4, joinedAt: 0 },
+        { userId: USER, relayHostId: other, cellId: 'c1', epoch: 2, joinedAt: 0 }
+      ],
+      now: 1_000_000
+    })
+    await writer.flush()
+    expect(result).toEqual({ upserted: 1, demoted: 0 })
+    expect(await row(host)).toMatchObject({ cell_id: 'c2', assignment_epoch: '6' })
+    expect(await row(other)).toMatchObject({ cell_id: 'c1', assignment_epoch: '2' })
+  })
+})

cloud/apps/relay/src/reserve-ledger-writer.test.ts

@@ -0,0 +1,127 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { openInMemoryRelayDatabase, type RelayDatabase } from './database.js'
+import { reconcileReserveLedger, ReserveLedgerWriter } from './reserve-ledger-writer.js'
+
+const HOST = { userId: 'user-1', relayHostId: 'abcdefghijklmnop' }
+
+describe('reserve ledger writer', () => {
+  let database: RelayDatabase | undefined
+
+  afterEach(async () => {
+    await database?.close()
+    database = undefined
+  })
+
+  async function setup(now = () => 1_000) {
+    database = await openInMemoryRelayDatabase()
+    return { database, writer: new ReserveLedgerWriter(database, { now, log: () => undefined }) }
+  }
+
+  const row = async (db: RelayDatabase, relayHostId = HOST.relayHostId) =>
+    (
+      await db.query(
+        `SELECT cell_id, assignment_epoch, reserved_controls FROM relay_assignments
+         WHERE user_id = ? AND relay_host_id = ?`,
+        [HOST.userId, relayHostId]
+      )
+    )[0]
+
+  it('writes a booked join, and only a higher epoch replaces it', async () => {
+    const { database: db, writer } = await setup()
+    writer.enqueue({ ...HOST, cellId: 'c1', epoch: 5 })
+    await writer.flush()
+    expect(await row(db)).toMatchObject({ cell_id: 'c1', assignment_epoch: 5 })
+    writer.enqueue({ ...HOST, cellId: 'c2', epoch: 4 })
+    writer.enqueue({ ...HOST, cellId: 'c3', epoch: 5 })
+    await writer.flush()
+    expect(await row(db)).toMatchObject({ cell_id: 'c1', assignment_epoch: 5 })
+    writer.enqueue({ ...HOST, cellId: 'c2', epoch: 6 })
+    await writer.flush()
+    expect(await row(db)).toMatchObject({ cell_id: 'c2', assignment_epoch: 6 })
+  })
+
+  it('drops a moved row’s counters but keeps a same-cell row’s', async () => {
+    const { database: db, writer } = await setup()
+    await db.query(
+      `INSERT INTO relay_assignments
+       (user_id, relay_host_id, cell_id, assignment_epoch, lease_expires_at, last_activity_at,
+        reserved_controls, reserved_splices, reserved_invites, pending_installs,
+        pending_confirmations, migration_leases)
+       VALUES (?, ?, 'c1', 3, 0, 0, 2, 0, 0, 0, 0, 0), (?, ?, 'c1', 3, 0, 0, 2, 0, 0, 0, 0, 0)`,
+      [HOST.userId, HOST.relayHostId, HOST.userId, 'bbbbbbbbbbbbbbbb']
+    )
+    writer.enqueue({ ...HOST, cellId: 'c9', epoch: 4 })
+    writer.enqueue({ userId: HOST.userId, relayHostId: 'bbbbbbbbbbbbbbbb', cellId: 'c1', epoch: 4 })
+    await writer.flush()
+    expect(await row(db)).toMatchObject({ cell_id: 'c9', reserved_controls: 0 })
+    expect(await row(db, 'bbbbbbbbbbbbbbbb')).toMatchObject({ cell_id: 'c1', reserved_controls: 2 })
+  })
+
+  it('keeps its queue through a database stall and drops the oldest past the bound', async () => {
+    let now = 1_000
+    const { database: db, writer } = await setup(() => now)
+    const query = vi.spyOn(db, 'query').mockRejectedValueOnce(new Error('timeout'))
+    writer.enqueue({ ...HOST, cellId: 'c1', epoch: 5 })
+    await writer.flush()
+    expect(writer.pending()).toBe(1)
+    // Backoff: no retry before it is due.
+    await writer.flush()
+    expect(query).toHaveBeenCalledTimes(1)
+    now += 1_000
+    await writer.flush()
+    expect(writer.pending()).toBe(0)
+    expect(await row(db)).toMatchObject({ cell_id: 'c1', assignment_epoch: 5 })
+
+    const bounded = new ReserveLedgerWriter(db, { queueMax: 2, log: () => undefined })
+    for (const epoch of [1, 2, 3]) bounded.enqueue({ ...HOST, cellId: 'c1', epoch })
+    expect(bounded.pending()).toBe(2)
+  })
+
+  it('reconciles the map’s seats and never lowers an epoch', async () => {
+    const { database: db, writer } = await setup()
+    // Another path wrote a higher epoch on c7: the mirror leaves it, whatever the map says.
+    writer.enqueue({ ...HOST, cellId: 'c7', epoch: 9 })
+    await writer.flush()
+    writer.enqueue({ ...HOST, cellId: 'c1', epoch: 9 })
+    await writer.flush()
+    const other = { userId: 'user-2', relayHostId: 'cccccccccccccccc' }
+    const demoted: unknown[] = []
+    const result = await reconcileReserveLedger({
+      writer,
+      seats: () => [
+        { ...HOST, cellId: 'c1', epoch: 8, joinedAt: 0 },
+        { ...other, cellId: 'c1', epoch: 2, joinedAt: 0 },
+        // Too young: it may still be superseded.
+        { userId: 'user-3', relayHostId: 'dddddddddddddddd', cellId: 'c1', epoch: 2, joinedAt: 990_000 }
+      ],
+      demote: (seat) => demoted.push(seat),
+      now: 1_000_000
+    })
+    await writer.flush()
+    // The row is ahead of the c1 seat: the row stays, and that seat is told to go.
+    expect(result).toEqual({ upserted: 1, demoted: 1 })
+    expect(demoted).toEqual([{ ...HOST, cellId: 'c1', epoch: 8, joinedAt: 0 }])
+    expect(await row(db)).toMatchObject({ cell_id: 'c7', assignment_epoch: 9 })
+    const written = await db.query(`SELECT cell_id FROM relay_assignments WHERE user_id = 'user-2'`)
+    expect(written).toEqual([{ cell_id: 'c1' }])
+  })
+
+  it('demotes a seat level with its row on another cell, which the mirror would never replace', async () => {
+    const { database: db, writer } = await setup()
+    writer.enqueue({ ...HOST, cellId: 'c7', epoch: 9 })
+    await writer.flush()
+    const demoted: unknown[] = []
+    const result = await reconcileReserveLedger({
+      writer,
+      seats: () => [
+        { ...HOST, cellId: 'c1', epoch: 9, joinedAt: 0 },
+        // Level with its own row: nothing to do.
+        { ...HOST, relayHostId: 'eeeeeeeeeeeeeeee', cellId: 'c7', epoch: 9, joinedAt: 0 }
+      ],
+      demote: (seat, found) => demoted.push({ seat: seat.cellId, row: found }),
+      now: 1_000_000
+    })
+    expect(result).toEqual({ upserted: 1, demoted: 1 })
+    expect(demoted).toEqual([{ seat: 'c1', row: { cellId: 'c7', epoch: 9 } }])
+  })
+})

cloud/apps/relay/src/reserve-ledger-writer.ts

@@ -0,0 +1,225 @@
+import type { RelayDatabase } from './database.js'
+
+// Step 5's after-the-fact record: a reserve-mode cell admits a booked host without reading
+// Postgres, and the director that booked it writes the host's relay_assignments row about a
+// second later. Phones still resolve through that row until step 6.
+//
+// A guarded mirror, which nothing reads for a desktop's correctness: only a higher epoch
+// replaces a row, and nothing ever lowers one. No relay_cells row and no activity lease is
+// touched. A row whose cell changes drops its counters: the leases that back them keep
+// their own cell, and unbacked units stay counted on the old cell until its next reconcile,
+// which only makes that cell look fuller.
+
+export const RESERVE_LEDGER_FLUSH_MS = 250
+export const RESERVE_LEDGER_QUEUE_MAX = 200_000
+const RESERVE_LEDGER_BATCH = 500
+const RETRY_BASE_MS = 250
+const RETRY_MAX_MS = 5_000
+const SUMMARY_INTERVAL_MS = 60_000
+
+export type LedgerEntry = {
+  userId: string
+  relayHostId: string
+  cellId: string
+  epoch: number
+}
+
+export class ReserveLedgerWriter {
+  private queue: LedgerEntry[] = []
+  private flushing = false
+  private failures = 0
+  private retryAt = 0
+  private timer: ReturnType<typeof setInterval> | null = null
+  private dropped = 0
+  private written = 0
+  private summaryAt: number
+
+  constructor(
+    private readonly database: RelayDatabase,
+    private readonly input: {
+      now?: () => number
+      queueMax?: number
+      log?: (line: string) => void
+    } = {}
+  ) {
+    this.summaryAt = this.now()
+  }
+
+  private now(): number {
+    return (this.input.now ?? Date.now)()
+  }
+
+  start(): void {
+    this.timer ??= setInterval(() => void this.flush(), RESERVE_LEDGER_FLUSH_MS)
+    this.timer.unref?.()
+  }
+
+  stop(): void {
+    if (this.timer) clearInterval(this.timer)
+    this.timer = null
+  }
+
+  // During a database stall the queue grows; past its bound the oldest entries go first,
+  // since the 10-minute reconcile rewrites every seat anyway.
+  enqueue(entry: LedgerEntry): void {
+    this.queue.push(entry)
+    const max = this.input.queueMax ?? RESERVE_LEDGER_QUEUE_MAX
+    if (this.queue.length > max) {
+      const excess = this.queue.length - max
+      this.queue.splice(0, excess)
+      this.dropped += excess
+    }
+  }
+
+  pending(): number {
+    return this.queue.length
+  }
+
+  // Single-flight; resolves once the queued rows are written or the attempt failed.
+  async flush(): Promise<void> {
+    if (this.flushing || this.queue.length === 0 || this.now() < this.retryAt) {
+      this.logSummary()
+      return
+    }
+    this.flushing = true
+    const batch = this.queue.slice(0, RESERVE_LEDGER_BATCH)
+    try {
+      await this.write(batch)
+      this.queue.splice(0, batch.length)
+      this.written += batch.length
+      this.failures = 0
+      this.retryAt = 0
+    } catch {
+      this.failures += 1
+      this.retryAt = this.now() + Math.min(RETRY_MAX_MS, RETRY_BASE_MS * 2 ** (this.failures - 1))
+    } finally {
+      this.flushing = false
+      this.logSummary()
+    }
+  }
+
+  private async write(batch: LedgerEntry[]): Promise<void> {
+    // The newest entry per host wins inside one batch.
+    const latest = new Map<string, LedgerEntry>()
+    for (const entry of batch) {
+      const key = `${entry.userId}\u0000${entry.relayHostId}`
+      const known = latest.get(key)
+      if (!known || entry.epoch > known.epoch) latest.set(key, entry)
+    }
+    await this.upsert([...latest.values()])
+  }
+
+  private async upsert(entries: LedgerEntry[]): Promise<void> {
+    const now = this.now()
+    const rows = entries.map(() => '(?, ?, ?, ?, ?, ?, 0, 0, 0, 0, 0, 0)').join(', ')
+    const zeroOnMove = (column: string) =>
+      `${column} = CASE WHEN relay_assignments.cell_id = excluded.cell_id
+         THEN relay_assignments.${column} ELSE 0 END`
+    await this.database.query(
+      `INSERT INTO relay_assignments
+       (user_id, relay_host_id, cell_id, assignment_epoch, lease_expires_at, last_activity_at,
+        reserved_controls, reserved_splices, reserved_invites, pending_installs,
+        pending_confirmations, migration_leases)
+       VALUES ${rows}
+       ON CONFLICT (user_id, relay_host_id) DO UPDATE SET
+         ${[
+           'reserved_controls',
+           'reserved_splices',
+           'reserved_invites',
+           'pending_installs',
+           'pending_confirmations',
+           'migration_leases'
+         ]
+           .map(zeroOnMove)
+           .join(',\n         ')},
+         cell_id = excluded.cell_id,
+         assignment_epoch = excluded.assignment_epoch,
+         last_activity_at = excluded.last_activity_at
+       WHERE relay_assignments.assignment_epoch < excluded.assignment_epoch`,
+      entries.flatMap((entry) => [
+        entry.userId,
+        entry.relayHostId,
+        entry.cellId,
+        entry.epoch,
+        now,
+        now
+      ])
+    )
+  }
+
+  // Rows for these hosts, in one read per batch.
+  async readRows(
+    hosts: ReadonlyArray<{ userId: string; relayHostId: string }>
+  ): Promise<Map<string, { cellId: string; epoch: number }>> {
+    const rows = new Map<string, { cellId: string; epoch: number }>()
+    for (let start = 0; start < hosts.length; start += RESERVE_LEDGER_BATCH) {
+      const chunk = hosts.slice(start, start + RESERVE_LEDGER_BATCH)
+      const found = await this.database.query(
+        `SELECT user_id, relay_host_id, cell_id, assignment_epoch FROM relay_assignments
+         WHERE ${chunk.map(() => '(user_id = ? AND relay_host_id = ?)').join(' OR ')}`,
+        chunk.flatMap((host) => [host.userId, host.relayHostId])
+      )
+      for (const row of found) {
+        rows.set(`${String(row.user_id)}\u0000${String(row.relay_host_id)}`, {
+          cellId: String(row.cell_id),
+          epoch: Number(row.assignment_epoch)
+        })
+      }
+    }
+    return rows
+  }
+
+  private logSummary(): void {
+    const now = this.now()
+    if (now - this.summaryAt < SUMMARY_INTERVAL_MS) return
+    this.summaryAt = now
+    ;(this.input.log ?? ((line: string) => console.log(line)))(
+      JSON.stringify({
+        event: 'orca_relay_reserve_ledger_summary',
+        written: this.written,
+        pending: this.queue.length,
+        dropped: this.dropped,
+        failures: this.failures
+      })
+    )
+    this.written = 0
+    this.dropped = 0
+  }
+}
+
+export const RESERVE_LEDGER_RECONCILE_MS = 10 * 60_000
+// A seat this young may still be superseded; the next reconcile takes it.
+const RECONCILE_SETTLE_MS = 30_000
+
+type ReconcileSeat = { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }
+
+// Every director upserts the map's seats on reserve-mode cells: guarded and idempotent, so
+// concurrent runs are harmless. A row ahead of the seat (today's path answered a newer epoch,
+// and the desktop was then answered the old seat from memory) is never rewritten: that seat is
+// demoted (epoch-checked, closes with 4409), so the desktop re-assigns to what the row names.
+// The demoting director then mints above the row's epoch for that host (see raiseEpochFloor).
+export async function reconcileReserveLedger(input: {
+  writer: ReserveLedgerWriter
+  seats: () => ReconcileSeat[]
+  demote?: (seat: ReconcileSeat, row: { cellId: string; epoch: number }) => void
+  now: number
+}): Promise<{ upserted: number; demoted: number }> {
+  const seats = input.seats().filter((seat) => input.now - seat.joinedAt >= RECONCILE_SETTLE_MS)
+  const rows = await input.writer.readRows(seats)
+  let upserted = 0
+  let demoted = 0
+  for (const seat of seats) {
+    const row = rows.get(`${seat.userId}\u0000${seat.relayHostId}`)
+    // Level with the row on another cell is the same divergence: the mirror never replaces it.
+    const behindRow = row && (row.epoch > seat.epoch || (row.epoch === seat.epoch && row.cellId !== seat.cellId))
+    if (row && behindRow && input.demote) {
+      input.demote(seat, row)
+      demoted += 1
+      continue
+    }
+    if (row && row.epoch >= seat.epoch) continue
+    input.writer.enqueue(seat)
+    upserted += 1
+  }
+  return { upserted, demoted }
+}

cloud/apps/relay/src/shadow-seat-directory.ts

@@ -284,6 +284,15 @@ export class ShadowSeatDirectory {
     return [...this.cells.keys()]
   }
 
+  // One cell's seats as the map holds them.
+  seatsOnCell(cellId: string): Array<{ userId: string; relayHostId: string; seat: ShadowSeat }> {
+    const cursor = this.cells.get(cellId)
+    if (!cursor) return []
+    return [...cursor.seats].map(([key, seat]) => {
+      const separator = key.indexOf('\u0000')
+      return { userId: key.slice(0, separator), relayHostId: key.slice(separator + 1), seat }
+    })
+  }
 
   // An old cell (404) has no feed; its last known seats stay, marked by status.
   markNoFeed(cellId: string, now: number, runtimeIncarnation?: string): void {