feat(relay): step-5 reserve core: the cell's booking book and the director's placer (inert) (#27038)

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

cloud/apps/relay/src/cell-reserve-book.test.ts

@@ -0,0 +1,113 @@
+import { describe, expect, it } from 'vitest'
+import { CellReserveBook, type BookingCapacity, type ReserveContext } from './cell-reserve-book.js'
+
+function units(limit: number): BookingCapacity & { held: () => number } {
+  let held = 0
+  return {
+    held: () => held,
+    canReserve: () => held < limit,
+    tryReserve: () => {
+      if (held >= limit) return null
+      held += 1
+      let released = false
+      return {
+        release: () => {
+          if (released) return
+          released = true
+          held -= 1
+        }
+      }
+    }
+  }
+}
+
+const reserveMode: ReserveContext = { mode: 'reserve', draining: false, seatedEpoch: () => undefined }
+
+function item(userId: string, epoch = 2, extra: { sticky?: boolean; ttlMs?: number } = {}) {
+  return { userId, relayHostId: 'abcdefghijklmnop', epoch, ttlMs: extra.ttlMs ?? 30_000, ...extra }
+}
+
+describe('cell reserve book', () => {
+  it('books up to the capacity and answers full past it', () => {
+    let now = 0
+    const capacity = units(2)
+    const book = new CellReserveBook(capacity, () => 100, () => now)
+    expect(book.reserve('d1', item('a'), reserveMode)).toEqual({ outcome: 'ok' })
+    expect(book.reserve('d1', item('b'), reserveMode)).toEqual({ outcome: 'ok' })
+    expect(book.reserve('d1', item('c'), reserveMode)).toEqual({ outcome: 'full' })
+    expect(capacity.held()).toBe(2)
+    now = 31_000
+    book.sweep()
+    expect(capacity.held()).toBe(0)
+    expect(book.count()).toBe(0)
+  })
+
+  it('refuses past the intake budget and refills at the configured rate', () => {
+    let now = 0
+    const book = new CellReserveBook(units(1_000), () => 2, () => now, 3)
+    const outcomes = ['a', 'b', 'c', 'd'].map((user) => book.reserve('d1', item(user), reserveMode))
+    expect(outcomes.map((outcome) => outcome.outcome)).toEqual(['ok', 'ok', 'ok', 'intake'])
+    now = 500
+    expect(book.reserve('d1', item('e'), reserveMode).outcome).toBe('ok')
+    expect(book.reserve('d1', item('f'), reserveMode).outcome).toBe('intake')
+  })
+
+  it('answers seated-newer when the host is seated or booked at an equal or higher epoch', () => {
+    const book = new CellReserveBook(units(10), () => 100)
+    const seatedAt5: ReserveContext = { ...reserveMode, seatedEpoch: () => 5 }
+    expect(book.reserve('d1', item('a', 5), seatedAt5)).toEqual({ outcome: 'seated-newer', epoch: 5 })
+    expect(book.reserve('d1', item('b', 3), reserveMode).outcome).toBe('ok')
+    expect(book.reserve('d2', item('b', 3), reserveMode)).toEqual({ outcome: 'seated-newer', epoch: 3 })
+    expect(book.reserve('d2', item('b', 4), reserveMode).outcome).toBe('ok')
+    expect(book.count()).toBe(1)
+  })
+
+  it('lets a sticky re-booking skip the intake bucket but never the cap', () => {
+    const capacity = units(1)
+    const book = new CellReserveBook(capacity, () => 0, () => 0, 1)
+    expect(book.reserve('d1', item('a'), reserveMode).outcome).toBe('ok')
+    expect(book.reserve('d1', item('b'), reserveMode).outcome).toBe('full')
+    expect(book.reserve('d1', item('c', 2, { sticky: true }), reserveMode).outcome).toBe('full')
+    expect(book.take('a', 'abcdefghijklmnop', 2)).not.toBeNull()
+    expect(book.reserve('d1', item('c', 2, { sticky: true }), reserveMode).outcome).toBe('ok')
+  })
+
+  it('answers ok for a sticky re-booking of a host already seated at that epoch, booking nothing', () => {
+    const capacity = units(1)
+    const book = new CellReserveBook(capacity, () => 0, () => 0, 0)
+    const seated: ReserveContext = { ...reserveMode, seatedEpoch: () => 4 }
+    expect(book.reserve('d1', item('a', 4, { sticky: true }), seated).outcome).toBe('ok')
+    expect(capacity.held()).toBe(0)
+  })
+
+  it('consumes a booking once, only at its own epoch, and releases its unit', () => {
+    const capacity = units(5)
+    const book = new CellReserveBook(capacity, () => 100)
+    book.reserve('d7', item('a', 9), reserveMode)
+    expect(book.take('a', 'abcdefghijklmnop', 8)).toBeNull()
+    expect(book.take('a', 'abcdefghijklmnop', 9)).toMatchObject({ epoch: 9, directorId: 'd7' })
+    expect(book.take('a', 'abcdefghijklmnop', 9)).toBeNull()
+    expect(capacity.held()).toBe(0)
+  })
+
+  it('answers off outside reserve mode, draining while draining, and dry runs book nothing', () => {
+    const capacity = units(5)
+    const book = new CellReserveBook(capacity, () => 100)
+    const databaseMode: ReserveContext = { ...reserveMode, mode: 'db' }
+    expect(book.reserve('d1', item('a'), databaseMode).outcome).toBe('off')
+    expect(book.reserve('d1', item('a'), databaseMode, true).outcome).toBe('ok')
+    expect(book.reserve('d1', item('a'), { ...reserveMode, draining: true }).outcome).toBe('draining')
+    expect(capacity.held()).toBe(0)
+    expect(book.count()).toBe(0)
+  })
+
+  it('hands a lower-epoch booking unit to its successor rather than taking a second', () => {
+    const capacity = units(1)
+    const book = new CellReserveBook(capacity, () => 100)
+    expect(book.reserve('d1', item('a', 2), reserveMode).outcome).toBe('ok')
+    expect(book.reserve('d2', item('a', 3), reserveMode).outcome).toBe('ok')
+    expect(capacity.held()).toBe(1)
+    book.clear()
+    expect(capacity.held()).toBe(0)
+  })
+})

cloud/apps/relay/src/cell-reserve-book.ts

@@ -0,0 +1,166 @@
+import {
+  CELL_INTAKE_BURST,
+  type ReserveItem,
+  type ReserveOutcome
+} from './cell-reserve-contract.js'
+
+// The cell's half of step 5: it alone says yes or no to each booking. One event loop makes
+// "check cap and budget, then count the booking" atomic with no lock.
+
+// A connection unit held for a booked host until its hello consumes it or the TTL ends.
+export type BookingUnit = { release(): void }
+
+export type BookingCapacity = {
+  // Null when one more unit would cross the cell's placement ceiling.
+  tryReserve(): BookingUnit | null
+  canReserve(): boolean
+}
+
+export type CellAdmitMode = 'db' | 'reserve'
+
+export type ReserveContext = {
+  mode: CellAdmitMode
+  draining: boolean
+  // The epoch this host is seated at here, if it is.
+  seatedEpoch: (userId: string, relayHostId: string) => number | undefined
+}
+
+export type CellBooking = {
+  userId: string
+  relayHostId: string
+  epoch: number
+  directorId: string
+  expiresAt: number
+}
+
+type HeldBooking = CellBooking & { unit: BookingUnit }
+
+function hostKey(userId: string, relayHostId: string): string {
+  return `${userId}\u0000${relayHostId}`
+}
+
+export class IntakeBucket {
+  private tokens: number
+  private refilledAt: number
+
+  constructor(
+    private readonly ratePerSec: () => number,
+    private readonly burst: number,
+    now: number
+  ) {
+    this.tokens = burst
+    this.refilledAt = now
+  }
+
+  available(now: number): number {
+    const elapsedMs = Math.max(0, now - this.refilledAt)
+    this.tokens = Math.min(this.burst, this.tokens + (elapsedMs / 1_000) * this.ratePerSec())
+    this.refilledAt = now
+    return this.tokens
+  }
+
+  take(now: number): boolean {
+    if (this.available(now) < 1) return false
+    this.tokens -= 1
+    return true
+  }
+}
+
+export class CellReserveBook {
+  private readonly bookings = new Map<string, HeldBooking>()
+  private readonly intake: IntakeBucket
+
+  constructor(
+    private readonly capacity: BookingCapacity,
+    private readonly intakePerSec: () => number,
+    private readonly now: () => number = Date.now,
+    private readonly burst = CELL_INTAKE_BURST
+  ) {
+    this.intake = new IntakeBucket(intakePerSec, burst, now())
+  }
+
+  reserve(
+    directorId: string,
+    item: ReserveItem,
+    context: ReserveContext,
+    dryRun = false
+  ): ReserveOutcome {
+    const now = this.now()
+    this.sweep(now)
+    if (context.mode !== 'reserve' && !dryRun) return { outcome: 'off' }
+    if (context.draining) return { outcome: 'draining' }
+    const key = hostKey(item.userId, item.relayHostId)
+    const seated = context.seatedEpoch(item.userId, item.relayHostId)
+    const booked = this.bookings.get(key)
+    const known = Math.max(seated ?? 0, booked?.epoch ?? 0)
+    if (item.sticky) {
+      if (known > item.epoch) return { outcome: 'seated-newer', epoch: known }
+      if (seated === item.epoch) return { outcome: 'ok' }
+      if (booked?.epoch === item.epoch) {
+        if (!dryRun) booked.expiresAt = Math.max(booked.expiresAt, now + item.ttlMs)
+        return { outcome: 'ok' }
+      }
+    } else if (known >= item.epoch) {
+      return { outcome: 'seated-newer', epoch: known }
+    }
+    // A lower-epoch booking for this host hands its unit to the new one.
+    if (!booked && !this.capacity.canReserve()) return { outcome: 'full' }
+    // The post-restart sticky re-booking books nothing new, so it takes no token.
+    if (!item.sticky && this.intake.available(now) < 1) return { outcome: 'intake' }
+    if (dryRun) return { outcome: 'ok' }
+    const unit = booked?.unit ?? this.capacity.tryReserve()
+    if (!unit) return { outcome: 'full' }
+    if (!item.sticky) this.intake.take(now)
+    this.bookings.set(key, {
+      userId: item.userId,
+      relayHostId: item.relayHostId,
+      epoch: item.epoch,
+      directorId,
+      expiresAt: now + item.ttlMs,
+      unit
+    })
+    return { outcome: 'ok' }
+  }
+
+  has(userId: string, relayHostId: string): boolean {
+    const booking = this.bookings.get(hostKey(userId, relayHostId))
+    return booking !== undefined && booking.expiresAt > this.now()
+  }
+
+  // The hello's rule 1: a booking for this host at exactly this epoch admits it once.
+  take(userId: string, relayHostId: string, epoch: number): CellBooking | null {
+    const key = hostKey(userId, relayHostId)
+    const booking = this.bookings.get(key)
+    if (!booking || booking.epoch !== epoch || booking.expiresAt <= this.now()) return null
+    this.bookings.delete(key)
+    booking.unit.release()
+    const { unit: _unit, ...held } = booking
+    return held
+  }
+
+  // Leaving reserve mode, or the cell starting to drain, voids every booking.
+  clear(): void {
+    for (const booking of this.bookings.values()) booking.unit.release()
+    this.bookings.clear()
+  }
+
+  sweep(now = this.now()): void {
+    for (const [key, booking] of this.bookings) {
+      if (booking.expiresAt > now) continue
+      this.bookings.delete(key)
+      booking.unit.release()
+    }
+  }
+
+  count(): number {
+    return this.bookings.size
+  }
+
+  intakeSnapshot(): { perSec: number; burst: number; tokens: number } {
+    return {
+      perSec: this.intakePerSec(),
+      burst: this.burst,
+      tokens: Math.floor(this.intake.available(this.now()) * 100) / 100
+    }
+  }
+}

cloud/apps/relay/src/cell-reserve-contract.ts

@@ -0,0 +1,58 @@
+import { z } from 'zod'
+
+// Step 5 wire between directors and cells (v1). Directors book a seat on a cell before
+// answering a desktop; the cell admits or refuses each booking itself. Readers ignore
+// unknown fields, so either side may add optional ones.
+
+export const CELL_RESERVE_PATH = '/internal/reserve'
+export const CELL_DEMOTE_PATH = '/internal/demote'
+
+export const CELL_RESERVE_MAX_ITEMS = 64
+export const CELL_RESERVE_TTL_MS = 30_000
+export const CELL_INTAKE_BURST = 20
+
+export const ReserveItemSchema = z.object({
+  userId: z.string().min(1).max(256),
+  relayHostId: z.string().regex(/^[A-Za-z0-9_-]{16}$/),
+  epoch: z.number().int().positive(),
+  ttlMs: z.number().int().positive().max(CELL_RESERVE_TTL_MS),
+  // The re-booking of a host seated here before this cell restarted: no intake token.
+  sticky: z.boolean().optional()
+})
+export type ReserveItem = z.infer<typeof ReserveItemSchema>
+
+export const ReserveRequestSchema = z.object({
+  v: z.literal(1),
+  directorId: z.string().regex(/^[A-Za-z0-9_-]{1,64}$/),
+  // Checks every rule and books nothing (the shadow gate measures this before any flip).
+  dryRun: z.boolean().optional(),
+  items: z.array(ReserveItemSchema).min(1).max(CELL_RESERVE_MAX_ITEMS)
+})
+export type ReserveRequest = z.infer<typeof ReserveRequestSchema>
+
+// `off`: the cell is not in reserve mode, so the director must use the database path.
+export type ReserveOutcome =
+  | { outcome: 'ok' }
+  | { outcome: 'full' }
+  | { outcome: 'intake' }
+  | { outcome: 'draining' }
+  | { outcome: 'off' }
+  | { outcome: 'seated-newer'; epoch: number }
+
+// Lenient: a newer cell may answer an outcome this director does not know; it counts as a no.
+export const ReserveResponseSchema = z.object({
+  v: z.literal(1),
+  results: z.array(
+    z.object({ outcome: z.string(), epoch: z.number().int().optional() }).passthrough()
+  )
+})
+
+// Names one seat exactly: the epoch alone would also match the host rejoining this cell later.
+export const DemoteRequestSchema = z.object({
+  v: z.literal(1),
+  userId: z.string().min(1).max(256),
+  relayHostId: z.string().regex(/^[A-Za-z0-9_-]{16}$/),
+  epoch: z.number().int().positive(),
+  joinedAt: z.number().int().nonnegative()
+})
+export type DemoteRequest = z.infer<typeof DemoteRequestSchema>

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

@@ -0,0 +1,174 @@
+import { describe, expect, it } from 'vitest'
+import {
+  mintEpoch,
+  ReservePlacer,
+  type PlacementCell
+} from './reserve-placement.js'
+
+function cell(cellId: string, overrides: Partial<PlacementCell> = {}): PlacementCell {
+  return {
+    cellId,
+    region: 'us-central1',
+    reserve: true,
+    general: true,
+    draining: false,
+    polledAt: 0,
+    seats: 0,
+    bookings: 0,
+    ceiling: 100,
+    intakePerSec: 50,
+    intakeTokens: 20,
+    ...overrides
+  }
+}
+
+// Estimates start empty after a restart; a second of refill gives each cell its share.
+function warmPlacer(now: { value: number }, random = () => 0.3) {
+  const placer = new ReservePlacer(() => now.value, random)
+  return placer
+}
+
+describe('reserve placer', () => {
+  it('places on the first cell that answers ok and counts one call', async () => {
+    const now = { value: 0 }
+    const placer = warmPlacer(now)
+    const cells = [cell('c1'), cell('c2')]
+    placer.pick(cells, 'us-central1', new Set())
+    now.value = 1_000
+    cells.forEach((entry) => (entry.polledAt = now.value))
+    const result = await placer.place({
+      cells,
+      region: 'us-central1',
+      epoch: 4,
+      reserve: async () => ({ outcome: 'ok' })
+    })
+    expect(result).toMatchObject({ kind: 'placed', epoch: 4, calls: 1 })
+  })
+
+  it('stops after three refusals and falls back to the database pool when it has cells', async () => {
+    const now = { value: 0 }
+    const placer = warmPlacer(now)
+    const cells = [cell('c1'), cell('c2'), cell('c3'), cell('c4'), cell('db1', { reserve: false })]
+    placer.pick(cells, 'us-central1', new Set())
+    now.value = 1_000
+    cells.forEach((entry) => (entry.polledAt = now.value))
+    const asked: string[] = []
+    const result = await placer.place({
+      cells,
+      region: 'us-central1',
+      epoch: 2,
+      reserve: async (cellId) => {
+        asked.push(cellId)
+        return { outcome: 'full' }
+      }
+    })
+    expect(asked).toHaveLength(3)
+    expect(new Set(asked).size).toBe(3)
+    expect(result).toEqual({ kind: 'database', calls: 3 })
+  })
+
+  it('re-mints above a seated-newer answer', async () => {
+    const now = { value: 0 }
+    const placer = warmPlacer(now)
+    const cells = [cell('c1'), cell('c2')]
+    placer.pick(cells, 'us-central1', new Set())
+    now.value = 1_000
+    cells.forEach((entry) => (entry.polledAt = now.value))
+    const epochs: number[] = []
+    const result = await placer.place({
+      cells,
+      region: 'us-central1',
+      epoch: 3,
+      reserve: async (_cellId, epoch) => {
+        epochs.push(epoch)
+        return epochs.length === 1 ? { outcome: 'seated-newer', epoch: 7 } : { outcome: 'ok' }
+      }
+    })
+    expect(epochs).toEqual([3, 8])
+    expect(result).toMatchObject({ kind: 'placed', epoch: 8 })
+  })
+
+  it('paces with a bounded Retry-After when reserve cells have seats but no budget', () => {
+    const now = { value: 0 }
+    const placer = warmPlacer(now)
+    const cells = [cell('c1', { intakePerSec: 2 })]
+    expect(placer.pick(cells, 'us-central1', new Set())).toBeNull()
+    expect(placer.fallback(cells, 'us-central1', 0)).toEqual({ kind: 'pace', retryAfterSeconds: 3, calls: 0 })
+  })
+
+  it('never offers a stale, draining, non-general, full or other-region cell', () => {
+    const now = { value: 10_000 }
+    const placer = warmPlacer(now)
+    const cells = [
+      cell('stale', { polledAt: 6_000 }),
+      cell('draining', { draining: true, polledAt: 10_000 }),
+      cell('isolated', { general: false, polledAt: 10_000 }),
+      cell('full', { seats: 99, bookings: 1, polledAt: 10_000 }),
+      cell('asia-east2', { region: 'asia-east2', polledAt: 10_000 }),
+      cell('db', { reserve: false, polledAt: 10_000 }),
+      cell('old', { ceiling: undefined, polledAt: 10_000 })
+    ]
+    placer.pick(cells, 'us-central1', new Set())
+    now.value = 11_000
+    expect(placer.pick(cells, 'us-central1', new Set())).toBeNull()
+  })
+
+  it('counts its own bookings against the seat estimate until the next poll', () => {
+    const now = { value: 0 }
+    const placer = warmPlacer(now)
+    const cells = [cell('c1', { ceiling: 2, intakePerSec: 1_000 })]
+    placer.pick(cells, 'us-central1', new Set())
+    now.value = 1_000
+    cells[0]!.polledAt = 1_000
+    // Two of this director's own bookings fill the estimated ceiling.
+    void placer.place({ cells, region: 'us-central1', epoch: 1, reserve: async () => ({ outcome: 'ok' }) })
+    void placer.place({ cells, region: 'us-central1', epoch: 1, reserve: async () => ({ outcome: 'ok' }) })
+    expect(placer.pick(cells, 'us-central1', new Set())).toBeNull()
+  })
+
+  it('weights the reserve pool by free seats and gives it nothing while a database cell is unknown', () => {
+    const now = { value: 0 }
+    const placer = warmPlacer(now)
+    const cells = [cell('r1', { seats: 50 }), cell('d1', { reserve: false, seats: 50 })]
+    expect(placer.reservePoolShare(cells, 'us-central1')).toBeCloseTo(0.5)
+    cells.push(cell('d2', { reserve: false, ceiling: undefined }))
+    expect(placer.reservePoolShare(cells, 'us-central1')).toBe(0)
+  })
+})
+
+describe('reserve placer under concurrent placements', () => {
+  it('takes back only its own booking when a refusal lands after a later call was counted', async () => {
+    const now = { value: 0 }
+    const placer = warmPlacer(now)
+    const target = cell('c1', { intakeTokens: 20 })
+    placer.pick([target], 'us-central1', new Set())
+    now.value = 1_000
+    target.polledAt = now.value
+    let refuseFirst: (() => void) | undefined
+    const first = placer.place({
+      cells: [target],
+      region: 'us-central1',
+      epoch: 4,
+      reserve: () => new Promise((resolve) => (refuseFirst = () => resolve({ outcome: 'full' })))
+    })
+    now.value = 1_005
+    const second = await placer.place({
+      cells: [target],
+      region: 'us-central1',
+      epoch: 7,
+      reserve: async () => ({ outcome: 'ok' })
+    })
+    expect(second).toMatchObject({ kind: 'placed' })
+    refuseFirst!()
+    await first
+    // The cell's poll sent at 1,003 covers neither; only the second (counted at 1,005) is left.
+    expect(placer.load({ ...target, polledAt: 1_003 })).toBe(1)
+  })
+})
+
+describe('epoch mint', () => {
+  it('mints one above every known epoch', () => {
+    expect(mintEpoch([])).toBe(1)
+    expect(mintEpoch([3, 9, 4])).toBe(10)
+  })
+})

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

@@ -0,0 +1,230 @@
+import type { RelayRegion } from '@orca-cloud/relay-contract'
+import { CELL_INTAKE_BURST, CELL_RESERVE_TTL_MS, type ReserveOutcome } from './cell-reserve-contract.js'
+
+// The director's half of step 5: every director picks a cell from its own map and asks the
+// cell to hold a seat. It only estimates; the cell decides. Pure, so the simulation drives
+// exactly this code.
+
+export const RESERVE_MAX_TRIES = 3
+// A cell not polled this recently is not a candidate (unverifiable, never gone).
+export const RESERVE_CELL_FRESH_MS = 3_000
+// Each director assumes it owns this share of a cell's intake (5 = the configured director
+// count). Fewer live directors only under-use the budget; more only cause refusals.
+export const RESERVE_DIRECTOR_SHARE = 5
+export const RESERVE_PACE_MIN_SECONDS = 1
+export const RESERVE_PACE_MAX_SECONDS = 10
+
+export type PlacementCell = {
+  cellId: string
+  region: RelayRegion
+  // The cell reports admitMode=reserve as applied.
+  reserve: boolean
+  general: boolean
+  draining: boolean
+  // When this director last sent a poll that the cell answered; null if never.
+  polledAt: number | null
+  seats: number
+  bookings: number
+  // Cell-reported placement ceiling; undefined from a cell that predates step 5.
+  ceiling?: number
+  intakePerSec: number
+  intakeTokens: number
+}
+
+export type ReserveAttempt = ReserveOutcome | { outcome: 'unreachable' } | { outcome: string }
+
+export type PlacementResult =
+  | { kind: 'placed'; cellId: string; epoch: number; calls: number }
+  // No reserve cell said yes: the caller runs today's database path.
+  | { kind: 'database'; calls: number }
+  // Reserve cells have seats but no intake budget: 503 with this Retry-After.
+  | { kind: 'pace'; retryAfterSeconds: number; calls: number }
+
+type Estimate = { tokens: number; refilledAt: number; ownBookings: number[] }
+
+export class ReservePlacer {
+  private readonly estimates = new Map<string, Estimate>()
+
+  constructor(
+    private readonly now: () => number = Date.now,
+    private readonly random: () => number = Math.random,
+    private readonly share = RESERVE_DIRECTOR_SHARE
+  ) {}
+
+  isCandidate(cell: PlacementCell, region: RelayRegion, now = this.now()): boolean {
+    return (
+      cell.reserve &&
+      cell.region === region &&
+      cell.general &&
+      !cell.draining &&
+      cell.ceiling !== undefined &&
+      cell.polledAt !== null &&
+      now - cell.polledAt <= RESERVE_CELL_FRESH_MS
+    )
+  }
+
+  // Seats, the cell's bookings and this director's own bookings since that poll.
+  load(cell: PlacementCell): number {
+    return cell.seats + cell.bookings + this.ownBookingsSince(cell)
+  }
+
+  hasSeat(cell: PlacementCell): boolean {
+    return cell.ceiling !== undefined && this.load(cell) < cell.ceiling
+  }
+
+  hasBudget(cell: PlacementCell, now = this.now()): boolean {
+    return this.refill(cell, now).tokens >= 1
+  }
+
+  // Power of two choices over cells with an estimated seat and budget.
+  pick(cells: readonly PlacementCell[], region: RelayRegion, tried: ReadonlySet<string>) {
+    const now = this.now()
+    const candidates = cells.filter(
+      (cell) =>
+        !tried.has(cell.cellId) &&
+        this.isCandidate(cell, region, now) &&
+        this.hasSeat(cell) &&
+        this.hasBudget(cell, now)
+    )
+    if (candidates.length === 0) return null
+    const first = candidates[Math.floor(this.random() * candidates.length)]!
+    if (candidates.length === 1) return first
+    let second = candidates[Math.floor(this.random() * (candidates.length - 1))]!
+    if (second === first) second = candidates[candidates.length - 1]!
+    const ratio = (cell: PlacementCell) => this.load(cell) / cell.ceiling!
+    return ratio(second) < ratio(first) ? second : first
+  }
+
+  async place(input: {
+    cells: readonly PlacementCell[]
+    region: RelayRegion
+    // Above every epoch this director knows for the host.
+    epoch: number
+    reserve: (cellId: string, epoch: number) => Promise<ReserveAttempt>
+  }): Promise<PlacementResult> {
+    const tried = new Set<string>()
+    let epoch = input.epoch
+    let calls = 0
+    while (calls < RESERVE_MAX_TRIES) {
+      const cell = this.pick(input.cells, input.region, tried)
+      if (!cell) break
+      tried.add(cell.cellId)
+      const countedAt = this.countBooking(cell)
+      calls += 1
+      const answer = await input.reserve(cell.cellId, epoch)
+      if (answer.outcome === 'ok') return { kind: 'placed', cellId: cell.cellId, epoch, calls }
+      this.uncountBooking(cell, countedAt)
+      if (answer.outcome === 'intake') this.drain(cell)
+      if (answer.outcome === 'seated-newer' && 'epoch' in answer && typeof answer.epoch === 'number') {
+        epoch = Math.max(epoch, answer.epoch + 1)
+      }
+    }
+    return this.fallback(input.cells, input.region, calls)
+  }
+
+  // Database cells in the region come first, then pacing; a region with neither falls to the
+  // database path too, which spills to another region as today.
+  fallback(cells: readonly PlacementCell[], region: RelayRegion, calls: number): PlacementResult {
+    const now = this.now()
+    const databaseCells = cells.some(
+      (cell) => !cell.reserve && cell.region === region && cell.general && !cell.draining
+    )
+    if (databaseCells) return { kind: 'database', calls }
+    let refillMs: number | null = null
+    for (const cell of cells) {
+      if (!this.isCandidate(cell, region, now) || !this.hasSeat(cell)) continue
+      const estimate = this.refill(cell, now)
+      const rate = this.ratePerSec(cell)
+      if (rate <= 0) continue
+      const waitMs = (Math.max(0, 1 - estimate.tokens) / rate) * 1_000
+      refillMs = refillMs === null ? waitMs : Math.min(refillMs, waitMs)
+    }
+    if (refillMs === null) return { kind: 'database', calls }
+    const seconds = Math.ceil(refillMs / 1_000)
+    return {
+      kind: 'pace',
+      retryAfterSeconds: Math.min(RESERVE_PACE_MAX_SECONDS, Math.max(RESERVE_PACE_MIN_SECONDS, seconds)),
+      calls
+    }
+  }
+
+  // Share of the region's free seats on reserve cells; 0 unless every database cell in the
+  // region reports its ceiling, so this never places only on the new pool by accident.
+  reservePoolShare(cells: readonly PlacementCell[], region: RelayRegion): number {
+    const now = this.now()
+    let reserveFree = 0
+    let databaseFree = 0
+    for (const cell of cells) {
+      if (cell.region !== region || !cell.general || cell.draining) continue
+      if (cell.reserve) {
+        if (this.isCandidate(cell, region, now)) {
+          reserveFree += Math.max(0, cell.ceiling! - this.load(cell))
+        }
+        continue
+      }
+      if (cell.ceiling === undefined) return 0
+      databaseFree += Math.max(0, cell.ceiling - cell.seats)
+    }
+    const total = reserveFree + databaseFree
+    return total === 0 ? 0 : reserveFree / total
+  }
+
+  private ratePerSec(cell: PlacementCell): number {
+    return cell.intakePerSec / this.share
+  }
+
+  private burst(): number {
+    return Math.max(1, CELL_INTAKE_BURST / this.share)
+  }
+
+  private refill(cell: PlacementCell, now: number): Estimate {
+    let estimate = this.estimates.get(cell.cellId)
+    if (!estimate) {
+      // A restarted director starts empty and refills, so it cannot burst on top of the others.
+      estimate = { tokens: 0, refilledAt: now, ownBookings: [] }
+      this.estimates.set(cell.cellId, estimate)
+    }
+    const elapsed = Math.max(0, now - estimate.refilledAt) / 1_000
+    estimate.tokens = Math.min(this.burst(), estimate.tokens + elapsed * this.ratePerSec(cell))
+    estimate.refilledAt = now
+    return estimate
+  }
+
+  private ownBookingsSince(cell: PlacementCell): number {
+    const estimate = this.estimates.get(cell.cellId)
+    if (!estimate) return 0
+    const now = this.now()
+    const since = cell.polledAt ?? -Infinity
+    return estimate.ownBookings.filter((at) => at >= since && at > now - CELL_RESERVE_TTL_MS)
+      .length
+  }
+
+  // Returns the entry it added, so a refusal removes that one and not a concurrent call's.
+  private countBooking(cell: PlacementCell): number {
+    const estimate = this.refill(cell, this.now())
+    estimate.tokens = Math.max(0, estimate.tokens - 1)
+    const at = this.now()
+    estimate.ownBookings.push(at)
+    if (estimate.ownBookings.length > 1_000) estimate.ownBookings.shift()
+    return at
+  }
+
+  // A refusal other than `intake` spent no token on the cell.
+  private uncountBooking(cell: PlacementCell, at: number): void {
+    const estimate = this.refill(cell, this.now())
+    // A poll may already have dropped it; then there is nothing of this call's to remove.
+    const index = estimate.ownBookings.lastIndexOf(at)
+    if (index >= 0) estimate.ownBookings.splice(index, 1)
+    estimate.tokens = Math.min(this.burst(), estimate.tokens + 1)
+  }
+
+  private drain(cell: PlacementCell): void {
+    this.refill(cell, this.now()).tokens = 0
+  }
+}
+
+export function mintEpoch(knownEpochs: Iterable<number>): number {
+  let highest = 0
+  for (const epoch of knownEpochs) highest = Math.max(highest, epoch)
+  return highest + 1
+}