fix(relay): spread drain placement over a load band instead of one empty cell (#26802)

* fix(relay): spread placement over a load band so a drain does not land on one empty cell

Placement took the single least-loaded cell, so a just-rolled cell took a
whole drain and every follow-up serialized on its relay_cells row. Directors
now pick at random among cells within 5% of capacity of the minimum, and skip
a cell that already took over 40% of its pool's placements in the last 10 s
(floor 4). It only reorders: the least-loaded cell is the fallback, the hard
cap and region preference are unchanged.

* fix(relay): size the load band's intake window for the 1,200 s drain pace

At ~0.5 picks/s per director, a 10 s window with a floor of 4 let the floor
decide and the empty cell took ~65% of a c27-sized drain. A 60 s window and
a floor of 2 keep it at 40% at both the 300 s and 1,200 s paces across five
directors. Adds a once-a-minute orca_relay_placement_load_band line (picks,
budget skips, fallbacks, band size, picks per cell).

* chore(relay): log the load-band summary at info, not warning

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

@@ -1,5 +1,6 @@
 import { createDrainMigrationRowLookup } from './drain-migration-row-lookup.js'
 import { HeapWindowReaper } from './heap-window-reaper.js'
+import { PlacementLoadBand } from './placement-load-band.js'
 import {
   selectIdleRegionalRehomes,
   type IdleRegionalRehomeCandidate,
@@ -130,6 +131,8 @@ type RelayAssignmentStoreOptions = {
   regionalRehomeDirectorRegion?: RelayRegion
   requireLiveCells?: boolean
   heartbeatTtlMs?: number
+  // Absent keeps the strict least-loaded pick (tests that assert exact cells).
+  placementLoadBand?: PlacementLoadBand
   recordControlRenewal?: (durationMs: number, outcome: ControlRenewalOutcome) => void
 }
 
@@ -565,6 +568,7 @@ export class RelayAssignmentStore {
     { failures: number; until: number }
   >()
   private readonly recordControlRenewal?: RelayAssignmentStoreOptions['recordControlRenewal']
+  private readonly placementLoadBand?: PlacementLoadBand
   private readonly admissionSelector: RelayCellAdmissionSelector
   private readonly migrationCellRegistrar: RelayMigrationCellRegistrar
   private readonly activityQueue = new AssignmentIdentityQueue()
@@ -591,6 +595,7 @@ export class RelayAssignmentStore {
     this.requireLiveCells = options.requireLiveCells ?? false
     this.heartbeatTtlMs = options.heartbeatTtlMs ?? 45_000
     this.recordControlRenewal = options.recordControlRenewal
+    this.placementLoadBand = options.placementLoadBand
     this.admissionSelector = new RelayCellAdmissionSelector(database, now)
     this.migrationCellRegistrar = new RelayMigrationCellRegistrar(database, now)
   }
@@ -7796,25 +7801,25 @@ export class RelayAssignmentStore {
         connectionHeadroom.get(cellId) !== false
       )
     })
-    candidates.sort((left, right) => {
-      const leftLoad =
-        integer(left, 'reserved_requests') +
-        (runtimeLoad?.get(text(left, 'cell_id')) ?? integer(left, 'observed_requests'))
-      const rightLoad =
-        integer(right, 'reserved_requests') +
-        (runtimeLoad?.get(text(right, 'cell_id')) ?? integer(right, 'observed_requests'))
-      const loadDifference =
-        leftLoad / integer(left, 'capacity_requests') -
-        rightLoad / integer(right, 'capacity_requests')
-      return loadDifference || text(left, 'cell_id').localeCompare(text(right, 'cell_id'))
-    })
-    const preferred = candidates.filter(
-      (candidate) =>
-        (regions.get(text(candidate, 'cell_id')) ?? RELAY_DEFAULT_REGION) === preferredRegion
+    const loadRatio = (row: SqlRow) =>
+      (integer(row, 'reserved_requests') +
+        (runtimeLoad?.get(text(row, 'cell_id')) ?? integer(row, 'observed_requests'))) /
+      integer(row, 'capacity_requests')
+    const ranked = candidates
+      .map((row) => ({ row, cellId: text(row, 'cell_id'), loadRatio: loadRatio(row) }))
+      .sort(
+        (left, right) =>
+          left.loadRatio - right.loadRatio || left.cellId.localeCompare(right.cellId)
+      )
+    const preferred = ranked.filter(
+      (candidate) => (regions.get(candidate.cellId) ?? RELAY_DEFAULT_REGION) === preferredRegion
     )
-    const selected = regionMode === 'require' ? preferred[0] : (preferred[0] ?? candidates[0])
+    const pool = regionMode === 'require' || preferred.length > 0 ? preferred : ranked
+    const selected = this.placementLoadBand
+      ? this.placementLoadBand.pick(pool, this.now())
+      : pool[0]
     return selected
-      ? cell(selected, regions.get(text(selected, 'cell_id')) ?? RELAY_DEFAULT_REGION)
+      ? cell(selected.row, regions.get(selected.cellId) ?? RELAY_DEFAULT_REGION)
       : null
   }
 

cloud/apps/relay/src/placement-load-band.test.ts

@@ -0,0 +1,253 @@
+import { describe, expect, it } from 'vitest'
+import {
+  PLACEMENT_INTAKE_FLOOR,
+  PLACEMENT_INTAKE_SHARE,
+  PLACEMENT_INTAKE_WINDOW_MS,
+  PlacementLoadBand
+} from './placement-load-band.js'
+import { RelayAssignmentStore } from './assignment-store.js'
+import { openInMemoryRelayDatabase } from './database.js'
+
+function seededRandom(seed: number): () => number {
+  let state = seed >>> 0
+  return () => {
+    state = (state + 0x6d2b79f5) >>> 0
+    let value = Math.imul(state ^ (state >>> 15), 1 | state)
+    value = (value + Math.imul(value ^ (value >>> 7), 61 | value)) ^ value
+    return ((value ^ (value >>> 14)) >>> 0) / 4_294_967_296
+  }
+}
+
+type SimulatedCell = { cellId: string; reserved: number; capacity: number }
+
+const quiet = () => undefined
+
+// Each pick goes to one of the directors at random, as the load balancer spreads requests.
+function simulate(
+  directors: PlacementLoadBand[],
+  cells: SimulatedCell[],
+  picks: number,
+  at: (index: number) => number,
+  route: () => number = seededRandom(99)
+) {
+  const counts = new Map(cells.map((cell) => [cell.cellId, 0]))
+  for (let index = 0; index < picks; index += 1) {
+    const ranked = cells
+      .filter((cell) => cell.reserved < cell.capacity)
+      .map((cell) => ({ cellId: cell.cellId, loadRatio: cell.reserved / cell.capacity, cell }))
+      .sort(
+        (left, right) =>
+          left.loadRatio - right.loadRatio || left.cellId.localeCompare(right.cellId)
+      )
+    const director = directors[Math.floor(route() * directors.length)]
+    const picked = director?.pick(ranked, at(index))
+    if (!picked) throw new Error('no candidate')
+    picked.cell.reserved += 1
+    counts.set(picked.cellId, (counts.get(picked.cellId) ?? 0) + 1)
+  }
+  return counts
+}
+
+// c27's drain on 10-08: 2,770 hosts onto one empty Asia cell (2,805 free)
+// and four fuller ones (2,311 free between them).
+function c27DrainCells(): SimulatedCell[] {
+  return [
+    { cellId: 'cell-empty', reserved: 0, capacity: 4000 },
+    ...['cell-b', 'cell-c', 'cell-d', 'cell-e'].map((cellId, index) => ({
+      cellId,
+      reserved: 4000 - 577 - (index === 0 ? 3 : 0),
+      capacity: 4000
+    }))
+  ]
+}
+
+function drainShare(paceMs: number, seed: number) {
+  const cells = c27DrainCells()
+  const directors = Array.from(
+    { length: 5 },
+    (_, index) => new PlacementLoadBand(seededRandom(seed + index), quiet)
+  )
+  const hosts = 2770
+  const counts = simulate(directors, cells, hosts, (index) => (index * paceMs) / hosts)
+  return { counts, cells, share: (counts.get('cell-empty') ?? 0) / hosts }
+}
+
+describe('PlacementLoadBand', () => {
+  it('spreads a drain over an empty cell and four fuller cells within one window', () => {
+    const cells: SimulatedCell[] = [
+      { cellId: 'cell-empty', reserved: 0, capacity: 6000 },
+      ...['cell-b', 'cell-c', 'cell-d', 'cell-e'].map((cellId) => ({
+        cellId,
+        reserved: 2100,
+        capacity: 6000
+      }))
+    ]
+    const director = new PlacementLoadBand(seededRandom(7), quiet)
+    const counts = simulate([director], cells, 2000, () => 1_000)
+
+    expect([...counts.values()].reduce((sum, count) => sum + count, 0)).toBe(2000)
+    expect(counts.get('cell-empty')).toBeLessThanOrEqual(PLACEMENT_INTAKE_SHARE * 2000 + 1)
+    for (const cellId of ['cell-b', 'cell-c', 'cell-d', 'cell-e']) {
+      expect(counts.get(cellId)).toBeGreaterThan(200)
+    }
+    for (const cell of cells) expect(cell.reserved).toBeLessThanOrEqual(cell.capacity)
+  })
+
+  it.each([
+    ['1,200 s', 1_200_000],
+    ['300 s', 300_000]
+  ])('keeps the empty cell near the share across five directors at the %s pace', (_, paceMs) => {
+    for (const seed of [1, 2, 3]) {
+      const { counts, cells, share } = drainShare(paceMs, seed * 10)
+      expect([...counts.values()].reduce((sum, count) => sum + count, 0)).toBe(2770)
+      expect(share).toBeGreaterThan(0.3)
+      expect(share).toBeLessThan(0.5)
+      for (const cell of cells) expect(cell.reserved).toBeLessThanOrEqual(cell.capacity)
+    }
+  })
+
+  it('gives the empty cell about 60% when only one other cell is eligible', () => {
+    const cells: SimulatedCell[] = [
+      { cellId: 'cell-empty', reserved: 0, capacity: 6000 },
+      { cellId: 'cell-b', reserved: 2100, capacity: 6000 }
+    ]
+    const counts = simulate([new PlacementLoadBand(seededRandom(4), quiet)], cells, 1000, () => 1)
+    const share = (counts.get('cell-empty') ?? 0) / 1000
+    expect(share).toBeGreaterThan(0.55)
+    expect(share).toBeLessThan(0.65)
+  })
+
+  it('logs one summary a minute with skips, fallbacks and band size', () => {
+    const lines: Record<string, unknown>[] = []
+    const band = new PlacementLoadBand(seededRandom(6), (line) => lines.push(JSON.parse(line)))
+    const ranked = [
+      { cellId: 'cell-a', loadRatio: 0 },
+      { cellId: 'cell-b', loadRatio: 0.5 }
+    ]
+    for (let second = 0; second <= 60; second += 1) band.pick(ranked, second * 1_000)
+    expect(lines).toHaveLength(1)
+    expect(lines[0]).toMatchObject({
+      event: 'orca_relay_placement_load_band',
+      windowMs: 60_000,
+      picks: 61,
+      bandSizeMax: 1
+    })
+    expect(lines[0]?.budgetSkips).toBeGreaterThan(0)
+    expect(lines[0]?.fallbacks).toBeGreaterThan(0)
+  })
+
+  it('never places past capacity while the fuller cells fill', () => {
+    const cells: SimulatedCell[] = [
+      { cellId: 'cell-empty', reserved: 0, capacity: 100 },
+      ...['cell-b', 'cell-c'].map((cellId) => ({ cellId, reserved: 90, capacity: 100 }))
+    ]
+    const director = new PlacementLoadBand(seededRandom(11), quiet)
+    const counts = simulate([director], cells, 120, () => 1_000)
+
+    expect(cells.map((cell) => cell.reserved)).toEqual([100, 100, 100])
+    expect(counts.get('cell-empty')).toBe(100)
+  })
+
+  it('falls back to the least-loaded cell when every candidate is over budget', () => {
+    const band = new PlacementLoadBand(seededRandom(3), quiet)
+    const only = [{ cellId: 'cell-a', loadRatio: 0.5 }]
+    for (let index = 0; index < 50; index += 1) {
+      expect(band.pick(only, 1_000)?.cellId).toBe('cell-a')
+    }
+    expect(band.pick([], 1_000)).toBeUndefined()
+  })
+
+  it('keeps quiet placement on the least-loaded cell outside the band', () => {
+    let now = 0
+    const band = new PlacementLoadBand(seededRandom(5), quiet)
+    const ranked = [
+      { cellId: 'cell-a', loadRatio: 0.1 },
+      { cellId: 'cell-b', loadRatio: 0.3 }
+    ]
+    for (let index = 0; index < 20; index += 1) {
+      now += PLACEMENT_INTAKE_WINDOW_MS / PLACEMENT_INTAKE_FLOOR
+      expect(band.pick(ranked, now)?.cellId).toBe('cell-a')
+    }
+  })
+
+  it('picks at random among cells inside the band', () => {
+    const band = new PlacementLoadBand(seededRandom(9), quiet)
+    const ranked = [
+      { cellId: 'cell-a', loadRatio: 0.3 },
+      { cellId: 'cell-b', loadRatio: 0.32 },
+      { cellId: 'cell-c', loadRatio: 0.4 }
+    ]
+    const picked = new Set<string>()
+    let now = 0
+    for (let index = 0; index < 40; index += 1) {
+      now += PLACEMENT_INTAKE_WINDOW_MS
+      picked.add(band.pick(ranked, now)?.cellId ?? '')
+    }
+    expect([...picked].sort()).toEqual(['cell-a', 'cell-b'])
+  })
+})
+
+describe('leastLoadedCell with the load band', () => {
+  async function setupRegions() {
+    const database = await openInMemoryRelayDatabase()
+    const store = new RelayAssignmentStore(database, () => 1_000, {
+      placementLoadBand: new PlacementLoadBand(seededRandom(13), quiet)
+    })
+    const asia = ['asia-a', 'asia-b', 'asia-c', 'asia-d', 'asia-e']
+    await store.reconcileCells([
+      ...asia.map((id) => ({
+        id,
+        url: `https://${id}.example.com`,
+        capacityRequests: 100,
+        region: 'asia-east2' as const
+      })),
+      {
+        id: 'us-a',
+        url: 'https://us-a.example.com',
+        capacityRequests: 1000,
+        region: 'us-central1' as const
+      }
+    ])
+    await database.query(
+      `UPDATE relay_cells SET observed_requests = 35 WHERE cell_id IN (?, ?, ?, ?)`,
+      asia.slice(1)
+    )
+    return { database, store }
+  }
+
+  async function placements(store: RelayAssignmentStore, count: number) {
+    const counts = new Map<string, number>()
+    for (let index = 0; index < count; index += 1) {
+      const assignment = await store.assign(
+        { userId: `user-${index}`, relayHostId: `host${String(index).padStart(12, '0')}` },
+        'asia-east2'
+      )
+      counts.set(assignment.cellId, (counts.get(assignment.cellId) ?? 0) + 1)
+    }
+    return counts
+  }
+
+  it('spreads within the preferred region and keeps an empty US cell out', async () => {
+    const { store } = await setupRegions()
+    const counts = await placements(store, 200)
+
+    expect(counts.get('us-a')).toBeUndefined()
+    expect(counts.get('asia-a')).toBeLessThanOrEqual(PLACEMENT_INTAKE_SHARE * 200 + 1)
+    const used = [...counts.values()].filter((count) => count >= 20)
+    expect(used.length).toBeGreaterThanOrEqual(3)
+  })
+
+  it('fills the preferred region to its hard cap before spilling, never past it', async () => {
+    const { database, store } = await setupRegions()
+    const counts = await placements(store, 520)
+    const rows = await database.query(
+      `SELECT reserved_requests, capacity_requests FROM relay_cells`
+    )
+
+    // The hard cap is reserved < capacity: five Asia cells hold 500, the rest spill to the US.
+    expect(counts.get('us-a')).toBe(20)
+    for (const row of rows) {
+      expect(Number(row.reserved_requests)).toBeLessThanOrEqual(Number(row.capacity_requests))
+    }
+  })
+})

cloud/apps/relay/src/placement-load-band.ts

@@ -0,0 +1,120 @@
+// Spreads placement over cells near the minimum load, so a freshly rolled
+// (empty) cell does not take a whole drain and serialize every follow-up on
+// its one relay_cells row. It only orders candidates: a candidate is never
+// refused, and the pool's least-loaded cell is the fallback.
+
+// Cells within this share of capacity of the least-loaded one are equals.
+export const PLACEMENT_LOAD_BAND = 0.05
+// A minute, so a director at the 1,200 s drain pace (~0.5 picks/s) still
+// holds enough picks for the share, not the floor, to decide.
+export const PLACEMENT_INTAKE_WINDOW_MS = 60_000
+// A cell over this share of its pool's recent placements yields to the rest.
+export const PLACEMENT_INTAKE_SHARE = 0.4
+// Below this many recent picks a cell is never skipped, so quiet traffic
+// still goes to the least-loaded cell.
+export const PLACEMENT_INTAKE_FLOOR = 2
+export const PLACEMENT_LOAD_BAND_SUMMARY_MS = 60_000
+
+export type LoadBandCandidate = { cellId: string; loadRatio: number }
+
+type LoadBandSummary = {
+  picks: number
+  // Picks where at least one candidate was over budget and skipped.
+  budgetSkips: number
+  // Picks where every candidate was over budget: the least-loaded cell took it.
+  fallbacks: number
+  bandSizeTotal: number
+  bandSizeMax: number
+  picksByCell: Record<string, number>
+}
+
+function emptySummary(): LoadBandSummary {
+  return {
+    picks: 0,
+    budgetSkips: 0,
+    fallbacks: 0,
+    bandSizeTotal: 0,
+    bandSizeMax: 0,
+    picksByCell: {}
+  }
+}
+
+export class PlacementLoadBand {
+  // Director-instance-local and in memory: it shapes order, never correctness.
+  private readonly intake = new Map<string, number[]>()
+  private summary = emptySummary()
+  private summaryStartedAt: number | undefined
+
+  constructor(
+    private readonly random: () => number = Math.random,
+    private readonly log: (line: string) => void = (line) => console.info(line)
+  ) {}
+
+  // `sorted` is ascending by load ratio. Records the pick as intake; a pick
+  // whose transaction retries counts again, which only spreads harder.
+  pick<T extends LoadBandCandidate>(sorted: readonly T[], now: number): T | undefined {
+    const least = sorted[0]
+    if (!least) return undefined
+    const recent = new Map(
+      sorted.map((candidate) => [candidate.cellId, this.recentIntake(candidate.cellId, now)])
+    )
+    let poolIntake = 0
+    for (const count of recent.values()) poolIntake += count
+    const budget = Math.max(PLACEMENT_INTAKE_FLOOR, PLACEMENT_INTAKE_SHARE * poolIntake)
+    const withinBudget = sorted.filter((candidate) => (recent.get(candidate.cellId) ?? 0) < budget)
+    const pool = withinBudget.length > 0 ? withinBudget : [least]
+    const ceiling = pool[0]!.loadRatio + PLACEMENT_LOAD_BAND
+    const band = pool.filter((candidate) => candidate.loadRatio <= ceiling)
+    const picked = band[Math.min(band.length - 1, Math.floor(this.random() * band.length))]!
+    this.record(picked.cellId, now)
+    this.observe(now, {
+      cellId: picked.cellId,
+      skipped: withinBudget.length < sorted.length,
+      fallback: withinBudget.length === 0,
+      bandSize: band.length
+    })
+    return picked
+  }
+
+  // One line a minute tells a spread drain from a floor- or fallback-dominated one.
+  private observe(
+    now: number,
+    pick: { cellId: string; skipped: boolean; fallback: boolean; bandSize: number }
+  ): void {
+    this.summaryStartedAt ??= now
+    const summary = this.summary
+    summary.picks += 1
+    if (pick.skipped) summary.budgetSkips += 1
+    if (pick.fallback) summary.fallbacks += 1
+    summary.bandSizeTotal += pick.bandSize
+    summary.bandSizeMax = Math.max(summary.bandSizeMax, pick.bandSize)
+    summary.picksByCell[pick.cellId] = (summary.picksByCell[pick.cellId] ?? 0) + 1
+    if (now - this.summaryStartedAt < PLACEMENT_LOAD_BAND_SUMMARY_MS) return
+    this.log(
+      JSON.stringify({
+        event: 'orca_relay_placement_load_band',
+        windowMs: now - this.summaryStartedAt,
+        ...summary
+      })
+    )
+    this.summary = emptySummary()
+    this.summaryStartedAt = now
+  }
+
+  private recentIntake(cellId: string, now: number): number {
+    const times = this.intake.get(cellId)
+    if (!times) return 0
+    const cutoff = now - PLACEMENT_INTAKE_WINDOW_MS
+    let expired = 0
+    while (expired < times.length && times[expired]! <= cutoff) expired += 1
+    if (expired > 0) times.splice(0, expired)
+    if (times.length === 0) this.intake.delete(cellId)
+    return times.length
+  }
+
+  private record(cellId: string, now: number): void {
+    const times = this.intake.get(cellId)
+    if (times) times.push(now)
+    else this.intake.set(cellId, [now])
+  }
+}

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

@@ -30,6 +30,7 @@ import { observeRelayDatabase } from './observed-relay-database.js'
 import { RelayObservability } from './relay-observability.js'
 import { combineRegionalRehomeSafety } from './regional-rehome-safety.js'
 import { RelayConnectionLedger, type RelayConnectionUpgrade } from './relay-connection-ledger.js'
+import { PlacementLoadBand } from './placement-load-band.js'
 import { createRelayReadiness } from './relay-readiness.js'
 import { createRelayTokenVerifier, readBearer } from './relay-token-verifier.js'
 import { closeRelayWebSocket } from './relay-websocket-close.js'
@@ -129,6 +130,7 @@ export function createRelayServer(
   const store = new RelayCredentialStore(observedDatabase, options.now)
   const assignments = new RelayAssignmentStore(observedDatabase, options.now, {
     requireLiveCells: config.role === 'director',
+    placementLoadBand: config.role === 'director' ? new PlacementLoadBand(options.random) : undefined,
     regionalRehomeCohortPercent: config.regionCorrectionCohortPercent ?? 0,
     // The director runs in the database's region; only its rehome readers use this.
     regionalRehomeDirectorRegion: