diff --git a/cloud/apps/relay/src/app.ts b/cloud/apps/relay/src/app.ts index 21910a8ca2a..babfe5b43b9 100644 --- a/cloud/apps/relay/src/app.ts +++ b/cloud/apps/relay/src/app.ts @@ -31,6 +31,7 @@ import { RelayAssignmentRowBusyError, RelayCellInReserveModeError, RelayHomeCellUnavailableError, + type AssignOptions, type CellFenceAttemptEvidence, type RelayAssignment, type RelayAssignmentStore, @@ -460,7 +461,7 @@ export function createRelayApp( .setExpirationTime('5m') .sign(config.assignmentSigningKey) // Step 5: switched-on cells answer from the map; anything else keeps today's path below. - let step5: { epochFloor?: number; placeFresh?: () => Promise } = {} + let step5: Pick = {} if (operations.reservePlacement && !body.data.regionCorrection) { let plan: ReservePlan try { @@ -492,7 +493,7 @@ export function createRelayApp( context.header('Retry-After', String(plan.retryAfterSeconds)) return context.json({ error: 'assignment_temporarily_unavailable' }, 503) } - step5 = { epochFloor: plan.epochFloor, placeFresh: plan.placeFresh } + step5 = { epochFloor: plan.epochFloor, placeFresh: plan.placeFresh, regionReserve: plan.regionReserve } } let admission: { release(): void } | null = null let lane: AssignmentAdmissionLane = 'placement' diff --git a/cloud/apps/relay/src/assignment-store.ts b/cloud/apps/relay/src/assignment-store.ts index 53dfcb5598f..adc70d3befe 100644 --- a/cloud/apps/relay/src/assignment-store.ts +++ b/cloud/apps/relay/src/assignment-store.ts @@ -145,6 +145,8 @@ type RelayAssignmentStoreOptions = { export type AssignOptions = { epochFloor?: number placeFresh?: () => Promise + // A reserve cell in the placement region with room: the database lane leaves the region only after it. + regionReserve?: () => Promise hostRequested?: boolean } @@ -534,6 +536,13 @@ class AssignmentInventoryScopeChanged extends Error { } } +// The placement region has no database cell with room; nothing was written. +class AssignmentRegionFull extends Error { + constructor() { + super('assignment_region_full') + } +} + // Why a reason of its own: a host whose home cell is fenced-but-unattested is // refused regardless of fleet headroom, so reporting it as capacity sends // operators after capacity that was never short. Every cell boot and every @@ -1008,18 +1017,28 @@ export class RelayAssignmentStore { if (booked) return booked // Only placement needs the global inventory critical section; queueing those // attempts locally avoids turning true placement bursts into NOWAIT storms. - return await this.serializeAssignment( - async () => - await this.assignWithLockRetry( - identity, - lockMode, - preferredRegion, - placementRegion, - step5.epochFloor ?? 0, - // Only through the step-5 plan: its floor holds every epoch the map knows for the host. - step5.hostRequested === true && step5.epochFloor !== undefined - ) - ) + const placeOnDatabase = async (keepRegion: boolean) => + await this.serializeAssignment( + async () => + await this.assignWithLockRetry( + identity, + lockMode, + preferredRegion, + placementRegion, + step5.epochFloor ?? 0, + // Only through the step-5 plan: its floor holds every epoch the map knows for the host. + step5.hostRequested === true && step5.epochFloor !== undefined, + keepRegion + ) + ) + const regionReserve = step5.regionReserve + if (!regionReserve) return await placeOnDatabase(false) + try { + return await placeOnDatabase(true) + } catch (error) { + if (!(error instanceof AssignmentRegionFull)) throw error + } + return (await regionReserve()) ?? (await placeOnDatabase(false)) } private async assignStickyWithLockRetry( @@ -1039,7 +1058,8 @@ export class RelayAssignmentStore { preferredRegion?: RelayRegion, placementRegion: RelayRegion = preferredRegion ?? RELAY_DEFAULT_REGION, epochFloor = 0, - hostRequested = false + hostRequested = false, + keepRegion = false ): Promise { const deadline = Date.now() + ASSIGNMENT_LOCK_RETRY_DEADLINE_MS let inventoryScope: AssignmentInventoryScope = 'none' @@ -1052,7 +1072,8 @@ export class RelayAssignmentStore { preferredRegion, placementRegion, epochFloor, - hostRequested + hostRequested, + keepRegion ) } catch (error) { if (error instanceof AssignmentInventoryScopeChanged) { @@ -1325,7 +1346,9 @@ export class RelayAssignmentStore { preferredRegion?: RelayRegion, placementRegion: RelayRegion = preferredRegion ?? RELAY_DEFAULT_REGION, epochFloor = 0, - hostRequested = false + hostRequested = false, + // Refuse to spill out of placementRegion: AssignmentRegionFull instead. + keepRegion = false ): Promise { const now = this.now() // An isolated retry takes its tier's rows before anything reassigns this, @@ -1587,8 +1610,15 @@ export class RelayAssignmentStore { // cross-region spill it exists to refuse. const target = isolatedTarget ?? - (await this.leastLoadedCell(transaction, lockedCells, placementRegion)) - if (!target) throw new Error('relay_capacity_exhausted') + (await this.leastLoadedCell( + transaction, + lockedCells, + placementRegion, + keepRegion ? 'require' : 'prefer' + )) + if (!target) { + throw keepRegion ? new AssignmentRegionFull() : new Error('relay_capacity_exhausted') + } if (deadReserveSource) { events.push( JSON.stringify({ diff --git a/cloud/apps/relay/src/relay-region-placement.test.ts b/cloud/apps/relay/src/relay-region-placement.test.ts index 76cf5479a41..8754b15b0d5 100644 --- a/cloud/apps/relay/src/relay-region-placement.test.ts +++ b/cloud/apps/relay/src/relay-region-placement.test.ts @@ -1,7 +1,8 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest' -import { RelayAssignmentStore } from './assignment-store.js' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { RelayAssignmentStore, type RelayAssignment } from './assignment-store.js' import type { RelayCellConfig } from './config.js' import { openInMemoryRelayDatabase, type RelayDatabase } from './database.js' +import { ReservePaceError } from './reserve-assignment.js' const CELLS: RelayCellConfig[] = [ { @@ -179,6 +180,68 @@ describe('Relay regional placement', () => { ).resolves.toBeUndefined() }) + describe('with a reserve cell in the region', () => { + const identity = { userId: 'spill-user', relayHostId: 'spillhost0000001' } + const onReserve: RelayAssignment = { + ...identity, + cellId: 'asia-reserve', + cellUrl: 'https://asia-reserve.relay.example.com', + assignmentEpoch: 1, + leaseExpiresAt: 1_000_000, + region: 'asia-east2' + } + const assignAsia = async (regionReserve?: () => Promise) => + await store.assign(identity, 'asia-east2', 'asia-east2', undefined, { + placeFresh: async () => null, + regionReserve + }) + const fillAsia = async () => { + for (const cell of CELLS.slice(1)) await store.configureCell(cell, 'migration-only') + } + + it('books the reserve cell rather than spill out of a full region', async () => { + await fillAsia() + const place = vi.fn(async () => onReserve) + await expect(assignAsia(place)).resolves.toEqual(onReserve) + expect(place).toHaveBeenCalledOnce() + // The refused in-region attempt wrote nothing. + expect(await database.query(`SELECT cell_id FROM relay_assignments`)).toEqual([]) + }) + + it('answers a throttled reserve cell with its pace, not a cell abroad', async () => { + await fillAsia() + await expect( + assignAsia(async () => { + throw new ReservePaceError(4) + }) + ).rejects.toBeInstanceOf(ReservePaceError) + expect(await database.query(`SELECT cell_id FROM relay_assignments`)).toEqual([]) + }) + + it('spills as before when the reserve cell has no room either', async () => { + await fillAsia() + const place = vi.fn(async () => null) + await expect(assignAsia(place)).resolves.toMatchObject({ + cellId: 'us-c1', + region: 'us-central1' + }) + expect(place).toHaveBeenCalledOnce() + // No in-region reserve room at plan time: today's single spill. + await expect( + store.assign({ ...identity, relayHostId: 'spillhost0000002' }, 'asia-east2', 'asia-east2') + ).resolves.toMatchObject({ cellId: 'us-c1' }) + }) + + it('leaves a region with database room to the database lane', async () => { + const place = vi.fn(async () => onReserve) + await expect(assignAsia(place)).resolves.toMatchObject({ + cellId: 'asia-c1', + region: 'asia-east2' + }) + expect(place).not.toHaveBeenCalled() + }) + }) + it('expires identity-linked region preferences after 30 days', async () => { const identity = { userId: 'expiry-user', relayHostId: 'expiryhost000001' } await store.assign(identity, 'asia-east2') diff --git a/cloud/apps/relay/src/reserve-assignment.test.ts b/cloud/apps/relay/src/reserve-assignment.test.ts index 38a6aa27e08..b87c76801c3 100644 --- a/cloud/apps/relay/src/reserve-assignment.test.ts +++ b/cloud/apps/relay/src/reserve-assignment.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it, vi } from 'vitest' +import type { RelayRegion } from '@orca-cloud/relay-contract' import type { CellReserveClient } from './cell-reserve-client.js' import { ReserveAssignment, @@ -13,7 +14,7 @@ import { ShadowSeatDirectory, type SeatFeedCell, type SeatFeedResponse } from '. const HOST = { userId: 'user-1', relayHostId: 'abcdefghijklmnop' } const US = 'us-central1' as const -function cellList(ids: string[], region = US): SeatFeedCell[] { +function cellList(ids: string[], region: RelayRegion = US): SeatFeedCell[] { return ids.map((cellId) => ({ cellId, cellUrl: `https://${cellId}.example`, @@ -322,6 +323,59 @@ describe('reserve assignment on a director', () => { } }) + describe('for a share-draw loser', () => { + const ASIA = 'asia-east2' as const + const OTHER = { ...HOST, relayHostId: 'qrstuvwxyzabcdef' } + const setupAsia = (reserveCell = feed('c1'), reserve?: (cellId: string) => ReserveAttempt) => + setup({ + ...(reserve ? { reserve } : {}), + cells: [...cellList(['c1', 'c2'], ASIA), ...cellList(['c3'])], + feeds: [ + reserveCell, + // Database room far above the reserve cell's, so the 0.25 draw loses. + feed('c2', { admitMode: 'db', counts: { seats: 0, controls: 0, bookings: 0, units: 0, ceiling: 5_000 } }), + feed('c3') + ] + }) + const regionReserveOf = async (assignment: ReserveAssignment, identity = HOST) => { + const plan = await assignment.plan(identity, { reconnect: false, region: ASIA }) + expect(plan).toMatchObject({ kind: 'database', epochFloor: 0 }) + return plan.kind === 'database' ? plan.regionReserve : undefined + } + + it('offers the reserve cell in its region only, only while it has a seat', async () => { + const { assignment, reserved, directory, now } = setupAsia() + const regionReserve = await regionReserveOf(assignment) + expect(reserved).toEqual([]) + await expect(regionReserve?.()).resolves.toMatchObject({ cellId: 'c1', region: ASIA }) + expect(reserved.map((call) => call.cellId)).toEqual(['c1']) + // A full reserve cell is not offered, and the US reserve cell never is. + const full = feed('c1', { counts: { seats: 500, controls: 500, bookings: 0, units: 500, ceiling: 500 } }) + directory.apply('c1', full, now.value, now.value) + expect(await regionReserveOf(assignment, OTHER)).toBeUndefined() + expect(reserved.map((call) => call.cellId)).toEqual(['c1']) + }) + + it('paces when the reserve cell has seats but no intake budget, rather than give up the region', async () => { + const { assignment, reserved } = setupAsia(feed('c1', { intake: { perSec: 1, burst: 20, tokens: 0 } })) + const regionReserve = await regionReserveOf(assignment) + await expect(regionReserve?.()).rejects.toBeInstanceOf(ReservePaceError) + expect(reserved).toEqual([]) + }) + + it('gives the region up when the reserve cell refuses with a seat still on the map', async () => { + for (const outcome of ['full', 'unreachable']) { + const { assignment, reserved } = setupAsia(feed('c1'), (cellId) => + cellId === 'c1' ? { outcome } : { outcome: 'ok' } + ) + const regionReserve = await regionReserveOf(assignment) + // Null sends the host to the database lane's cross-region fallback, not a 1 s retry. + await expect(regionReserve?.()).resolves.toBeNull() + expect(reserved.map((call) => call.cellId)).toEqual(['c1']) + } + }) + }) + 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( diff --git a/cloud/apps/relay/src/reserve-assignment.ts b/cloud/apps/relay/src/reserve-assignment.ts index a6417457110..de647665138 100644 --- a/cloud/apps/relay/src/reserve-assignment.ts +++ b/cloud/apps/relay/src/reserve-assignment.ts @@ -4,7 +4,7 @@ import { RELAY_DEFAULT_REGION, type RelayRegion } from '@orca-cloud/relay-contract' -import type { RelayAssignment } from './assignment-store.js' +import type { AssignOptions, RelayAssignment } from './assignment-store.js' import { CELL_RESERVE_TTL_MS } from './cell-reserve-contract.js' import type { CellReserveClient } from './cell-reserve-client.js' import { relayHostLogDigest } from './relay-host-log-digest.js' @@ -45,11 +45,7 @@ type Identity = { userId: string; relayHostId: string } export type ReservePlan = | { kind: 'answer'; assignment: RelayAssignment; lane: 'sticky' | 'placement' } | { kind: 'retry'; retryAfterSeconds: number; reason: 'map-incomplete' | 'database' | 'paced' } - | { - kind: 'database' - epochFloor?: number - placeFresh?: () => Promise - } + | ({ kind: 'database' } & Pick) // 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). @@ -209,15 +205,36 @@ export class ReserveAssignment { } const placeFresh = async (): Promise => await this.placeFresh(identity, request.region, epochFloor, seats) + const regionReserve = this.regionReserve(identity, request.region, epochFloor, seats, now) // Where the host was last seen decides who owns it: a database-mode cell keeps today's // path, sticky included; only a fresh placement may be booked on a reserve-mode cell. const lastOnReserve = lastCellId !== undefined && directory.placementModeOf(lastCellId) === 'reserve' if (!lastOnReserve && (seats.length > 0 || left !== undefined || row !== null)) { - return { kind: 'database', epochFloor, placeFresh } + return { kind: 'database', epochFloor, placeFresh, regionReserve } } const placed = await placeFresh() if (placed) return { kind: 'answer', assignment: placed, lane: 'placement' } - return { kind: 'database', epochFloor } + return { kind: 'database', epochFloor, regionReserve } + } + + // For the share draw's losers: the database lane tries the region first, then this, then spills. + // Undefined while no in-region reserve cell has a seat, so the database lane runs as today. + private regionReserve( + identity: Identity, + region: RelayRegion, + epochFloor: number, + seats: readonly ShadowSeat[], + now: number + ): AssignOptions['regionReserve'] { + const hasRoom = this.placementCells().some( + (cell) => this.input.placer.isCandidate(cell, region, now) && this.input.placer.hasSeat(cell) + ) + if (!hasRoom) return undefined + return async () => { + const placed = await this.placeFresh(identity, region, epochFloor, seats, true) + if (placed) this.summary.regionKept += 1 + return placed + } } // The same epoch on the same cell: no booking, no database, no row lock. A cell that @@ -288,16 +305,19 @@ export class ReserveAssignment { identity: Identity, region: RelayRegion, epochFloor: number, - seats: readonly ShadowSeat[] = [] + seats: readonly ShadowSeat[] = [], + // The database lane found the region full: no share draw, and no other region. + regionFull = false ): Promise { const placements = this.placementCells() - if (this.random() >= this.input.placer.reservePoolShare(placements, region)) return null - const regions = region === RELAY_DEFAULT_REGION ? [region] : [region, RELAY_DEFAULT_REGION] + if (!regionFull && this.random() >= this.input.placer.reservePoolShare(placements, region)) return null + const regions = regionFull || region === RELAY_DEFAULT_REGION ? [region] : [region, RELAY_DEFAULT_REGION] for (const target of regions) { const result = await this.input.placer.place({ cells: placements, region: target, epoch: mintEpoch([epochFloor]), + regionFull, reserve: async (cellId, epoch) => { const cell = this.input.cells().find((entry) => entry.cellId === cellId) if (!cell) return { outcome: 'unreachable' } @@ -463,6 +483,8 @@ type PlacementSummary = { sticky: number stickyReserves: number placed: number + // Placements the database lane would have sent out of the region. + regionKept: number calls: number database: number supersedes: number @@ -488,6 +510,7 @@ function emptySummary(): PlacementSummary { sticky: 0, stickyReserves: 0, placed: 0, + regionKept: 0, calls: 0, database: 0, supersedes: 0, diff --git a/cloud/apps/relay/src/reserve-placement.ts b/cloud/apps/relay/src/reserve-placement.ts index 3e677c054f5..2cba1baaf84 100644 --- a/cloud/apps/relay/src/reserve-placement.ts +++ b/cloud/apps/relay/src/reserve-placement.ts @@ -106,8 +106,12 @@ export class ReservePlacer { // Above every epoch this director knows for the host. epoch: number reserve: (cellId: string, epoch: number) => Promise + // The database lane found the region full: a throttled reserve cell paces instead. + regionFull?: boolean }): Promise { const tried = new Set() + // Refused for a reason other than intake: no seat to wait for this time. + const refused = new Set() let epoch = input.epoch let calls = 0 while (calls < RESERVE_MAX_TRIES) { @@ -120,23 +124,30 @@ export class ReservePlacer { if (answer.outcome === 'ok') return { kind: 'placed', cellId: cell.cellId, epoch, calls } this.uncountBooking(cell, countedAt) if (answer.outcome === 'intake') this.drain(cell) + else refused.add(cell.cellId) 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) + return this.fallback(input.cells, input.region, calls, { regionFull: input.regionFull, refused }) } - // 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 { + // Database cells in the region come first (unless known full), then pacing on a cell that + // only lacks budget; anything else falls to the database path, which spills as today. + fallback( + cells: readonly PlacementCell[], + region: RelayRegion, + calls: number, + options: { regionFull?: boolean; refused?: ReadonlySet } = {} + ): PlacementResult { const now = this.now() - const databaseCells = cells.some( - (cell) => !cell.reserve && cell.region === region && cell.general && !cell.draining - ) + const databaseCells = + !options.regionFull && + 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 (options.refused?.has(cell.cellId)) continue if (!this.isCandidate(cell, region, now) || !this.hasSeat(cell)) continue const estimate = this.refill(cell, now) const rate = this.ratePerSec(cell)