From 80d7e2dded8938ab48e4f375a2c851e606ad4980 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Fri, 9 Oct 2026 16:58:07 -0400 Subject: [PATCH] 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 | 39 +-- .../relay/src/placement-load-band.test.ts | 253 ++++++++++++++++++ cloud/apps/relay/src/placement-load-band.ts | 120 +++++++++ cloud/apps/relay/src/relay-server.ts | 2 + 4 files changed, 397 insertions(+), 17 deletions(-) create mode 100644 cloud/apps/relay/src/placement-load-band.test.ts create mode 100644 cloud/apps/relay/src/placement-load-band.ts diff --git a/cloud/apps/relay/src/assignment-store.ts b/cloud/apps/relay/src/assignment-store.ts index 007b7108615..8a59931ee08 100644 --- a/cloud/apps/relay/src/assignment-store.ts +++ b/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 } diff --git a/cloud/apps/relay/src/placement-load-band.test.ts b/cloud/apps/relay/src/placement-load-band.test.ts new file mode 100644 index 00000000000..7920b90750f --- /dev/null +++ b/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[] = [] + 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() + 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() + 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)) + } + }) +}) diff --git a/cloud/apps/relay/src/placement-load-band.ts b/cloud/apps/relay/src/placement-load-band.ts new file mode 100644 index 00000000000..91cfecc6353 --- /dev/null +++ b/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 +} + +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() + 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(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]) + } +} diff --git a/cloud/apps/relay/src/relay-server.ts b/cloud/apps/relay/src/relay-server.ts index d61f2b36b2e..ed06da58723 100644 --- a/cloud/apps/relay/src/relay-server.ts +++ b/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: