diff --git a/cloud/apps/relay/src/cell-reserve-book.test.ts b/cloud/apps/relay/src/cell-reserve-book.test.ts new file mode 100644 index 00000000000..c408b51467c --- /dev/null +++ b/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) + }) +}) diff --git a/cloud/apps/relay/src/cell-reserve-book.ts b/cloud/apps/relay/src/cell-reserve-book.ts new file mode 100644 index 00000000000..e67ed54161a --- /dev/null +++ b/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() + 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 + } + } +} diff --git a/cloud/apps/relay/src/cell-reserve-contract.ts b/cloud/apps/relay/src/cell-reserve-contract.ts new file mode 100644 index 00000000000..56d915f9500 --- /dev/null +++ b/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 + +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 + +// `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 diff --git a/cloud/apps/relay/src/reserve-placement.test.ts b/cloud/apps/relay/src/reserve-placement.test.ts new file mode 100644 index 00000000000..a79f3b642ea --- /dev/null +++ b/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 { + 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) + }) +}) diff --git a/cloud/apps/relay/src/reserve-placement.ts b/cloud/apps/relay/src/reserve-placement.ts new file mode 100644 index 00000000000..b3436f148c0 --- /dev/null +++ b/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() + + 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) { + 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 + }): Promise { + const tried = new Set() + 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 { + let highest = 0 + for (const epoch of knownEpochs) highest = Math.max(highest, epoch) + return highest + 1 +}