From fda01facc21ddd92e95fa0b8f90e9c7f40b1bc18 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Sat, 10 Oct 2026 19:11:59 -0400 Subject: [PATCH] feat(relay): the director that booked a join writes its row about a second later (off) (#27048) Rebased onto main as one commit after its parent squash-merged. --- .../relay/src/floating-promise-census.test.ts | 1 + cloud/apps/relay/src/relay-server.ts | 73 ++++++ .../apps/relay/src/reserve-assignment.test.ts | 83 ++++++- cloud/apps/relay/src/reserve-assignment.ts | 94 +++++++- .../reserve-ledger-writer-postgres.test.ts | 61 +++++ .../relay/src/reserve-ledger-writer.test.ts | 127 ++++++++++ cloud/apps/relay/src/reserve-ledger-writer.ts | 225 ++++++++++++++++++ cloud/apps/relay/src/shadow-seat-directory.ts | 9 + 8 files changed, 664 insertions(+), 9 deletions(-) create mode 100644 cloud/apps/relay/src/reserve-ledger-writer-postgres.test.ts create mode 100644 cloud/apps/relay/src/reserve-ledger-writer.test.ts create mode 100644 cloud/apps/relay/src/reserve-ledger-writer.ts diff --git a/cloud/apps/relay/src/floating-promise-census.test.ts b/cloud/apps/relay/src/floating-promise-census.test.ts index 19fd403a8db..fcc056a5446 100644 --- a/cloud/apps/relay/src/floating-promise-census.test.ts +++ b/cloud/apps/relay/src/floating-promise-census.test.ts @@ -14,6 +14,7 @@ const NEVER_REJECTS = new Map([ ['cell-heartbeat-client.ts#send', 'one try/catch around the whole send'], ['cell-reserve-client.ts#send', 'one try/catch around the request; resolves every item'], ['cell-reserve-client.ts#demote', 'one try/catch around the request'], + ['reserve-ledger-writer.ts#flush', 'one try/catch around the write; retries with backoff'], ['control-renewal-batch.ts#flush', 'rejects the waiters, never itself'], ['regional-rehome-worker.ts#run', 'one try/catch around the whole poll'], ['relay-control-flag-channel.ts#poll', 'one catch on the whole read'] diff --git a/cloud/apps/relay/src/relay-server.ts b/cloud/apps/relay/src/relay-server.ts index d2f8e71ab56..e2066bfd91e 100644 --- a/cloud/apps/relay/src/relay-server.ts +++ b/cloud/apps/relay/src/relay-server.ts @@ -51,6 +51,11 @@ import { CellReserveClient } from './cell-reserve-client.js' import { googleMetadataIdentityToken, reusedIdentityToken } from './google-metadata-identity-token.js' import { ReserveAssignment } from './reserve-assignment.js' import { ReservePlacer } from './reserve-placement.js' +import { + reconcileReserveLedger, + RESERVE_LEDGER_RECONCILE_MS, + ReserveLedgerWriter +} from './reserve-ledger-writer.js' import { ProcessQueuedByteBudget } from './splice-forwarder.js' // A malformed percent-escape in the request target must be a client error, never a URIError @@ -185,6 +190,11 @@ export function createRelayServer( // Step 5 on directors: needs the map and the rehome credential the cells verify. const directorId = randomUUID() const rehomeAudience = config.rehomeAudience + // Only a director that books writes rows, and only for the joins it booked. + const reserveLedger = + config.role === 'director' && shadowSeatPoller && config.reservePlacement === 'on' + ? new ReserveLedgerWriter(observedDatabase, { now: options.now }) + : undefined const reservePlacement = config.role === 'director' && shadowSeatPoller && @@ -204,6 +214,7 @@ export function createRelayServer( ) }), readRow: async (identity) => await assignments.hostWhereabouts(identity), + databaseBusy: () => readRelayDatabasePoolPressure(database).databasePoolWaiting > 0, now: options.now, random: options.random }) @@ -228,6 +239,68 @@ export function createRelayServer( }, 60_000) : null offWithReserveTimer?.unref() + if (reserveLedger && shadowSeatPoller) { + const directory = shadowSeatPoller.directory + directory.onBookedJoin = (join) => { + if (join.reservedBy !== directorId) return + reserveLedger.enqueue({ + userId: join.userId, + relayHostId: join.relayHostId, + cellId: join.cellId, + epoch: join.epoch + }) + } + reserveLedger.start() + const reconcile = (): void => { + const now = (options.now ?? Date.now)() + void reconcileReserveLedger({ + writer: reserveLedger, + seats: () => + directory + .cellIds() + .filter((cellId) => directory.admitModeOf(cellId) === 'reserve') + .flatMap((cellId) => + directory.seatsOnCell(cellId).map(({ userId, relayHostId, seat }) => ({ + userId, + relayHostId, + cellId, + epoch: seat.epoch, + joinedAt: seat.joinedAt + })) + ), + demote: (seat, row) => { + reservePlacement?.demoteSeat(seat) + // The re-assign mints above the row, so the mirror can take it. + reservePlacement?.raiseEpochFloor(seat, row.epoch) + console.warn( + JSON.stringify({ + event: 'orca_relay_reserve_ledger_seat_behind_row', + cellId: seat.cellId, + epoch: seat.epoch, + rowEpoch: row.epoch + }) + ) + }, + now + }) + .catch(() => console.warn('[orca-relay] reserve ledger reconcile deferred')) + .finally(schedule) + } + // Jittered, so five directors do not reconcile together. + const schedule = (): void => { + const random = options.random ?? Math.random + setTimeout(reconcile, RESERVE_LEDGER_RECONCILE_MS + random() * 60_000).unref?.() + } + // A restart loses the queue: the first run goes as soon as the map is complete. + const clock = options.now ?? Date.now + const bootedAt = clock() + const firstRun = setInterval(() => { + if (!directory.isComplete() && clock() - bootedAt < RESERVE_LEDGER_RECONCILE_MS) return + clearInterval(firstRun) + reconcile() + }, 1_000) + firstRun.unref?.() + } const configuredConnectionLimits = config.connectionHardCap === undefined ? null diff --git a/cloud/apps/relay/src/reserve-assignment.test.ts b/cloud/apps/relay/src/reserve-assignment.test.ts index c48d8f3a4ac..7af627d7aa8 100644 --- a/cloud/apps/relay/src/reserve-assignment.test.ts +++ b/cloud/apps/relay/src/reserve-assignment.test.ts @@ -1,6 +1,12 @@ import { describe, expect, it, vi } from 'vitest' import type { CellReserveClient } from './cell-reserve-client.js' -import { ReserveAssignment, ReservePaceError, RESERVE_STARTUP_GATE_MS } from './reserve-assignment.js' +import { + ReserveAssignment, + ReservePaceError, + RESERVE_ROW_FLOOR_TTL_MS, + RESERVE_WRONG_CELL_ROW_READ_MS, + RESERVE_STARTUP_GATE_MS +} from './reserve-assignment.js' import { ReservePlacer, type ReserveAttempt } from './reserve-placement.js' import { ShadowSeatDirectory, type SeatFeedCell, type SeatFeedResponse } from './shadow-seat-directory.js' @@ -43,6 +49,7 @@ function setup( feeds?: SeatFeedResponse[] reserve?: (cellId: string, sticky: boolean) => ReserveAttempt row?: (identity: { userId: string; relayHostId: string }) => Promise<{ cellId: string; assignmentEpoch: number } | null> + databaseBusy?: () => boolean mode?: 'off' | 'dry-run' | 'on' startedAt?: number } = {} @@ -78,6 +85,7 @@ function setup( // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: see above. client: client as unknown as CellReserveClient, readRow: options.row ?? (async () => null), + ...(options.databaseBusy ? { databaseBusy: options.databaseBusy } : {}), now: () => now.value, random: () => 0.25, log: () => undefined @@ -88,7 +96,7 @@ function setup( for (const response of options.feeds ?? cells.map((cell) => feed(cell.cellId))) { directory.apply(response.cellId, { ...response, full: response.full ?? [] }, now.value - 1, now.value - 1) } - return { assignment, directory, reserved, demoted, now } + return { assignment, directory, reserved, demoted, now, placer } } function seated(epoch: number, joinedAt = 0) { @@ -162,6 +170,21 @@ describe('reserve assignment on a director', () => { expect(plan).toMatchObject({ kind: 'answer', lane: 'sticky', assignment: { cellId: 'c1', assignmentEpoch: 7 } }) }) + it('after the reconcile demotes a seat behind its row, mints above the row and never answers that seat', async () => { + const { assignment, directory, reserved, now } = setup({ feeds: [feed('c1', { full: [seated(7)] }), feed('c2')] }) + // The row names another cell at epoch 8; the mirror would refuse a re-assign at 8. + assignment.raiseEpochFloor(HOST, 8) + const plan = await assignment.plan(HOST, { reconnect: true, region: US }) + expect(plan).toMatchObject({ kind: 'answer', lane: 'placement', assignment: { assignmentEpoch: 9 } }) + expect(reserved.every((call) => !call.sticky && call.epoch === 9)).toBe(true) + // The floor lapses, so a host it no longer protects answers from memory again. + now.value += RESERVE_ROW_FLOOR_TTL_MS + 1 + for (const response of [feed('c1', { seq: 2, full: [seated(7)] }), feed('c2', { seq: 2 })]) { + directory.apply(response.cellId, response, now.value, now.value) + } + expect(await assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({ lane: 'sticky' }) + }) + it('never sends a drained or demoted host back, and books it elsewhere above every known epoch', async () => { const { assignment, directory, reserved, now } = setup() directory.apply( @@ -231,6 +254,62 @@ describe('reserve assignment on a director', () => { }) }) + it('after a WRONG_CELL leave, any director reads the row and mints above it', async () => { + const leftWrongCell = ( + row: () => Promise<{ cellId: string; assignmentEpoch: number } | null>, + databaseBusy?: () => boolean + ) => { + const harness = setup({ row, ...(databaseBusy ? { databaseBusy } : {}) }) + // Another director demoted the seat: this one only sees the leave in c1's feed. + harness.directory.apply( + 'c1', + feed('c1', { + full: undefined, + seq: 3, + changes: [ + { seq: 2, kind: 'join', ...HOST, epoch: 7, generation: 1, at: 1 }, + { seq: 3, kind: 'leave', ...HOST, epoch: 7, generation: 1, closeCode: 4409, at: 2 } + ] + }), + harness.now.value, + harness.now.value + ) + return harness + } + const row = vi.fn(async () => ({ cellId: 'c2', assignmentEpoch: 8 })) + const read = leftWrongCell(row) + expect(await read.assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({ + kind: 'answer', + lane: 'placement', + assignment: { assignmentEpoch: 9 } + }) + expect(row).toHaveBeenCalledTimes(1) + // The database down: placed as before the read existed, above what the map knows. + const down = leftWrongCell(async () => Promise.reject(new Error('timeout'))) + expect(await down.assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({ + kind: 'answer', + assignment: { assignmentEpoch: 8 } + }) + // Pool waiters: the read is skipped outright, so a stall cannot stack timed-out reads. + const busyRow = vi.fn(async () => ({ cellId: 'c2', assignmentEpoch: 8 })) + const busy = leftWrongCell(busyRow, () => true) + expect(await busy.assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({ + kind: 'answer', + assignment: { assignmentEpoch: 8 } + }) + expect(busyRow).not.toHaveBeenCalled() + // A stalled database: the read gives up after its bound and places as before. + vi.useFakeTimers() + try { + const stalled = leftWrongCell(() => new Promise(() => undefined)) + const plan = stalled.assignment.plan(HOST, { reconnect: true, region: US }) + await vi.advanceTimersByTimeAsync(RESERVE_WRONG_CELL_ROW_READ_MS) + expect(await plan).toMatchObject({ kind: 'answer', assignment: { assignmentEpoch: 8 } }) + } finally { + vi.useRealTimers() + } + }) + 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 701bdc1e0d4..2fa46f230bf 100644 --- a/cloud/apps/relay/src/reserve-assignment.ts +++ b/cloud/apps/relay/src/reserve-assignment.ts @@ -49,9 +49,32 @@ export type ReservePlan = placeFresh?: () => Promise } +// 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). +export const RESERVE_ROW_FLOOR_TTL_MS = 2 * 60 * 60_000 +const RESERVE_ROW_FLOOR_MAX = 100_000 + +// The WRONG_CELL row read's bound: well under a database stall (5-7 s). +export const RESERVE_WRONG_CELL_ROW_READ_MS = 1_000 + +// Rejects after `ms`; the read itself still settles later, and its failure is swallowed. +async function withinMs(read: Promise, ms: number): Promise { + let timer: ReturnType | undefined + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error('row_read_timeout')), ms) + }) + read.catch(() => undefined) + try { + return await Promise.race([read, timeout]) + } finally { + clearTimeout(timer) + } +} + export class ReserveAssignment { private summary: PlacementSummary = emptySummary() private summaryAt: number + private readonly rowFloors = new Map() constructor( private readonly input: { @@ -63,6 +86,9 @@ export class ReserveAssignment { client: CellReserveClient // One primary-key read of the host's row: only for a host the map has never seen. readRow: (identity: Identity) => Promise<{ cellId: string; assignmentEpoch: number } | null> + // True while the pool has waiters: the optional WRONG_CELL read is skipped, so timed-out + // reads cannot stack up behind a stall. + databaseBusy?: () => boolean now?: () => number random?: () => number log?: (line: string) => void @@ -130,19 +156,34 @@ export class ReserveAssignment { .filter((entry) => now - entry.at <= RESERVE_STICKY_RECENT_MS) .at(-1) const booking = directory.bookingOf(identity.userId, identity.relayHostId) + const rowFloor = this.rowFloor(identity, now) const known = [ ...seats.map((seat) => seat.epoch), ...(left ? [left.epoch] : []), - ...(booking ? [booking.epoch] : []) + ...(booking ? [booking.epoch] : []), + ...(rowFloor === undefined ? [] : [rowFloor]) ] let row: { cellId: string; assignmentEpoch: number } | null = null - if (known.length === 0) { + // A seat closed with WRONG_CELL was demoted (or failed to re-register) because its row is + // ahead, which every director sees in that cell's feed: read the row, so the re-assign mints + // above it on any director, not just the one that demoted it. + const leftWrongCell = + left?.closeCode === RELAY_CLOSE_CODE.WRONG_CELL && !(this.input.databaseBusy?.() ?? false) + if (known.length === 0 || leftWrongCell) { try { - row = await this.input.readRow(identity) + // Only the never-seen host needs the row; for the WRONG_CELL one it is an optimisation, + // bounded so a database stall cannot hold the reconnect (the reconcile repairs a miss). + row = + known.length === 0 + ? await this.input.readRow(identity) + : await withinMs(this.input.readRow(identity), RESERVE_WRONG_CELL_ROW_READ_MS) } catch { // A host the map never saw, with the database down: its epoch cannot be minted safely. - this.summary.retries.database += 1 - return { kind: 'retry', retryAfterSeconds: 2, reason: 'database' } + // One the map knows places as it did before this read existed. + if (known.length === 0) { + this.summary.retries.database += 1 + return { kind: 'retry', retryAfterSeconds: 2, reason: 'database' } + } } if (row) known.push(row.assignmentEpoch) } @@ -158,7 +199,7 @@ export class ReserveAssignment { } if (request.reconnect) { - const sticky = await this.stickyFromMemory(identity, seats, left, now) + const sticky = await this.stickyFromMemory(identity, seats, left, now, rowFloor ?? 0) if (sticky) { this.summary.sticky += 1 return { kind: 'answer', assignment: sticky, lane: 'sticky' } @@ -183,11 +224,14 @@ export class ReserveAssignment { identity: Identity, seats: ShadowSeat[], left: { cellId: string; epoch: number; incarnation?: string; closeCode?: number } | undefined, - now: number + now: number, + rowFloor: number ): Promise { // Highest epoch first, then the newest join: only the newest grant may be answered from // memory, so a stale seat (or a leaver at a lower epoch) is never sent back. const newest = Math.max(0, ...seats.map((seat) => seat.epoch), left?.epoch ?? 0) + // A seat at or behind a row the reconcile demoted it for is never answered again. + if (newest <= rowFloor) return null const candidates: Array<{ cellId: string epoch: number @@ -310,6 +354,42 @@ export class ReserveAssignment { return null } + // The ledger reconcile demoted this host's seat behind (or level with) a row at `epoch`. + raiseEpochFloor(identity: Identity, epoch: number): void { + const key = `${identity.userId}\u0000${identity.relayHostId}` + const current = this.rowFloors.get(key) + this.rowFloors.delete(key) + this.rowFloors.set(key, { + epoch: Math.max(epoch, current?.epoch ?? 0), + expiresAt: this.now() + RESERVE_ROW_FLOOR_TTL_MS + }) + if (this.rowFloors.size > RESERVE_ROW_FLOOR_MAX) { + this.rowFloors.delete(this.rowFloors.keys().next().value!) + } + } + + private rowFloor(identity: Identity, now: number): number | undefined { + const key = `${identity.userId}\u0000${identity.relayHostId}` + const floor = this.rowFloors.get(key) + if (!floor) return undefined + if (floor.expiresAt > now) return floor.epoch + this.rowFloors.delete(key) + return undefined + } + + // Fire-and-forget; the cell checks the seat's epoch and join itself. + demoteSeat(seat: { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }): void { + const cell = this.input.cells().find((entry) => entry.cellId === seat.cellId) + if (!cell) return + void this.input.client.demote(cell, { + v: 1, + userId: seat.userId, + relayHostId: seat.relayHostId, + epoch: seat.epoch, + joinedAt: seat.joinedAt + }) + } + // Supersede at placement: the old seats this director's map holds on reserve cells are told // to go. Each cell checks the seat's epoch and join itself, so a stale map only misses. private supersede(identity: Identity, seats: readonly ShadowSeat[], cellId: string, epoch: number): void { diff --git a/cloud/apps/relay/src/reserve-ledger-writer-postgres.test.ts b/cloud/apps/relay/src/reserve-ledger-writer-postgres.test.ts new file mode 100644 index 00000000000..2276ed767d9 --- /dev/null +++ b/cloud/apps/relay/src/reserve-ledger-writer-postgres.test.ts @@ -0,0 +1,61 @@ +import { randomUUID } from 'node:crypto' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { openRelayDatabase, type RelayDatabase } from './database.js' +import { reconcileReserveLedger, ReserveLedgerWriter } from './reserve-ledger-writer.js' + +// The ledger's statements on real PostgreSQL: the guarded upsert, its counter reset on a +// move, and the batched row read. + +const databaseUrl = process.env.ORCA_RELAY_TEST_POSTGRES_URL +const describePostgres = databaseUrl ? describe : describe.skip +const USER = `ledger-pg-${randomUUID().slice(0, 8)}` + +describePostgres('reserve ledger writer on PostgreSQL', () => { + let database: RelayDatabase + + beforeAll(async () => { + database = await openRelayDatabase({ databaseUrl, dataDir: '/tmp' }) + }) + + afterAll(async () => { + await database.query(`DELETE FROM relay_assignments WHERE user_id = ?`, [USER]) + await database.close() + }) + + const row = async (relayHostId: string) => + ( + await database.query( + `SELECT cell_id, assignment_epoch, reserved_controls FROM relay_assignments + WHERE user_id = ? AND relay_host_id = ?`, + [USER, relayHostId] + ) + )[0] + + it('writes only a higher epoch, drops a moved row’s counters, and reconciles from the map', async () => { + const writer = new ReserveLedgerWriter(database, { log: () => undefined }) + const host = 'ledgerpghost0001' + writer.enqueue({ userId: USER, relayHostId: host, cellId: 'c1', epoch: 5 }) + await writer.flush() + expect(await row(host)).toMatchObject({ cell_id: 'c1', assignment_epoch: '5' }) + await database.query(`UPDATE relay_assignments SET reserved_controls = 2 WHERE user_id = ?`, [USER]) + writer.enqueue({ userId: USER, relayHostId: host, cellId: 'c2', epoch: 5 }) + await writer.flush() + expect(await row(host)).toMatchObject({ cell_id: 'c1', assignment_epoch: '5', reserved_controls: '2' }) + writer.enqueue({ userId: USER, relayHostId: host, cellId: 'c2', epoch: 6 }) + await writer.flush() + expect(await row(host)).toMatchObject({ cell_id: 'c2', assignment_epoch: '6', reserved_controls: '0' }) + const other = 'ledgerpghost0002' + const result = await reconcileReserveLedger({ + writer, + seats: () => [ + { userId: USER, relayHostId: host, cellId: 'c1', epoch: 4, joinedAt: 0 }, + { userId: USER, relayHostId: other, cellId: 'c1', epoch: 2, joinedAt: 0 } + ], + now: 1_000_000 + }) + await writer.flush() + expect(result).toEqual({ upserted: 1, demoted: 0 }) + expect(await row(host)).toMatchObject({ cell_id: 'c2', assignment_epoch: '6' }) + expect(await row(other)).toMatchObject({ cell_id: 'c1', assignment_epoch: '2' }) + }) +}) diff --git a/cloud/apps/relay/src/reserve-ledger-writer.test.ts b/cloud/apps/relay/src/reserve-ledger-writer.test.ts new file mode 100644 index 00000000000..ca51c6709c7 --- /dev/null +++ b/cloud/apps/relay/src/reserve-ledger-writer.test.ts @@ -0,0 +1,127 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { openInMemoryRelayDatabase, type RelayDatabase } from './database.js' +import { reconcileReserveLedger, ReserveLedgerWriter } from './reserve-ledger-writer.js' + +const HOST = { userId: 'user-1', relayHostId: 'abcdefghijklmnop' } + +describe('reserve ledger writer', () => { + let database: RelayDatabase | undefined + + afterEach(async () => { + await database?.close() + database = undefined + }) + + async function setup(now = () => 1_000) { + database = await openInMemoryRelayDatabase() + return { database, writer: new ReserveLedgerWriter(database, { now, log: () => undefined }) } + } + + const row = async (db: RelayDatabase, relayHostId = HOST.relayHostId) => + ( + await db.query( + `SELECT cell_id, assignment_epoch, reserved_controls FROM relay_assignments + WHERE user_id = ? AND relay_host_id = ?`, + [HOST.userId, relayHostId] + ) + )[0] + + it('writes a booked join, and only a higher epoch replaces it', async () => { + const { database: db, writer } = await setup() + writer.enqueue({ ...HOST, cellId: 'c1', epoch: 5 }) + await writer.flush() + expect(await row(db)).toMatchObject({ cell_id: 'c1', assignment_epoch: 5 }) + writer.enqueue({ ...HOST, cellId: 'c2', epoch: 4 }) + writer.enqueue({ ...HOST, cellId: 'c3', epoch: 5 }) + await writer.flush() + expect(await row(db)).toMatchObject({ cell_id: 'c1', assignment_epoch: 5 }) + writer.enqueue({ ...HOST, cellId: 'c2', epoch: 6 }) + await writer.flush() + expect(await row(db)).toMatchObject({ cell_id: 'c2', assignment_epoch: 6 }) + }) + + it('drops a moved row’s counters but keeps a same-cell row’s', async () => { + const { database: db, writer } = await setup() + await db.query( + `INSERT INTO relay_assignments + (user_id, relay_host_id, cell_id, assignment_epoch, lease_expires_at, last_activity_at, + reserved_controls, reserved_splices, reserved_invites, pending_installs, + pending_confirmations, migration_leases) + VALUES (?, ?, 'c1', 3, 0, 0, 2, 0, 0, 0, 0, 0), (?, ?, 'c1', 3, 0, 0, 2, 0, 0, 0, 0, 0)`, + [HOST.userId, HOST.relayHostId, HOST.userId, 'bbbbbbbbbbbbbbbb'] + ) + writer.enqueue({ ...HOST, cellId: 'c9', epoch: 4 }) + writer.enqueue({ userId: HOST.userId, relayHostId: 'bbbbbbbbbbbbbbbb', cellId: 'c1', epoch: 4 }) + await writer.flush() + expect(await row(db)).toMatchObject({ cell_id: 'c9', reserved_controls: 0 }) + expect(await row(db, 'bbbbbbbbbbbbbbbb')).toMatchObject({ cell_id: 'c1', reserved_controls: 2 }) + }) + + it('keeps its queue through a database stall and drops the oldest past the bound', async () => { + let now = 1_000 + const { database: db, writer } = await setup(() => now) + const query = vi.spyOn(db, 'query').mockRejectedValueOnce(new Error('timeout')) + writer.enqueue({ ...HOST, cellId: 'c1', epoch: 5 }) + await writer.flush() + expect(writer.pending()).toBe(1) + // Backoff: no retry before it is due. + await writer.flush() + expect(query).toHaveBeenCalledTimes(1) + now += 1_000 + await writer.flush() + expect(writer.pending()).toBe(0) + expect(await row(db)).toMatchObject({ cell_id: 'c1', assignment_epoch: 5 }) + + const bounded = new ReserveLedgerWriter(db, { queueMax: 2, log: () => undefined }) + for (const epoch of [1, 2, 3]) bounded.enqueue({ ...HOST, cellId: 'c1', epoch }) + expect(bounded.pending()).toBe(2) + }) + + it('reconciles the map’s seats and never lowers an epoch', async () => { + const { database: db, writer } = await setup() + // Another path wrote a higher epoch on c7: the mirror leaves it, whatever the map says. + writer.enqueue({ ...HOST, cellId: 'c7', epoch: 9 }) + await writer.flush() + writer.enqueue({ ...HOST, cellId: 'c1', epoch: 9 }) + await writer.flush() + const other = { userId: 'user-2', relayHostId: 'cccccccccccccccc' } + const demoted: unknown[] = [] + const result = await reconcileReserveLedger({ + writer, + seats: () => [ + { ...HOST, cellId: 'c1', epoch: 8, joinedAt: 0 }, + { ...other, cellId: 'c1', epoch: 2, joinedAt: 0 }, + // Too young: it may still be superseded. + { userId: 'user-3', relayHostId: 'dddddddddddddddd', cellId: 'c1', epoch: 2, joinedAt: 990_000 } + ], + demote: (seat) => demoted.push(seat), + now: 1_000_000 + }) + await writer.flush() + // The row is ahead of the c1 seat: the row stays, and that seat is told to go. + expect(result).toEqual({ upserted: 1, demoted: 1 }) + expect(demoted).toEqual([{ ...HOST, cellId: 'c1', epoch: 8, joinedAt: 0 }]) + expect(await row(db)).toMatchObject({ cell_id: 'c7', assignment_epoch: 9 }) + const written = await db.query(`SELECT cell_id FROM relay_assignments WHERE user_id = 'user-2'`) + expect(written).toEqual([{ cell_id: 'c1' }]) + }) + + it('demotes a seat level with its row on another cell, which the mirror would never replace', async () => { + const { database: db, writer } = await setup() + writer.enqueue({ ...HOST, cellId: 'c7', epoch: 9 }) + await writer.flush() + const demoted: unknown[] = [] + const result = await reconcileReserveLedger({ + writer, + seats: () => [ + { ...HOST, cellId: 'c1', epoch: 9, joinedAt: 0 }, + // Level with its own row: nothing to do. + { ...HOST, relayHostId: 'eeeeeeeeeeeeeeee', cellId: 'c7', epoch: 9, joinedAt: 0 } + ], + demote: (seat, found) => demoted.push({ seat: seat.cellId, row: found }), + now: 1_000_000 + }) + expect(result).toEqual({ upserted: 1, demoted: 1 }) + expect(demoted).toEqual([{ seat: 'c1', row: { cellId: 'c7', epoch: 9 } }]) + }) +}) diff --git a/cloud/apps/relay/src/reserve-ledger-writer.ts b/cloud/apps/relay/src/reserve-ledger-writer.ts new file mode 100644 index 00000000000..d10d5d954cf --- /dev/null +++ b/cloud/apps/relay/src/reserve-ledger-writer.ts @@ -0,0 +1,225 @@ +import type { RelayDatabase } from './database.js' + +// Step 5's after-the-fact record: a reserve-mode cell admits a booked host without reading +// Postgres, and the director that booked it writes the host's relay_assignments row about a +// second later. Phones still resolve through that row until step 6. +// +// A guarded mirror, which nothing reads for a desktop's correctness: only a higher epoch +// replaces a row, and nothing ever lowers one. No relay_cells row and no activity lease is +// touched. A row whose cell changes drops its counters: the leases that back them keep +// their own cell, and unbacked units stay counted on the old cell until its next reconcile, +// which only makes that cell look fuller. + +export const RESERVE_LEDGER_FLUSH_MS = 250 +export const RESERVE_LEDGER_QUEUE_MAX = 200_000 +const RESERVE_LEDGER_BATCH = 500 +const RETRY_BASE_MS = 250 +const RETRY_MAX_MS = 5_000 +const SUMMARY_INTERVAL_MS = 60_000 + +export type LedgerEntry = { + userId: string + relayHostId: string + cellId: string + epoch: number +} + +export class ReserveLedgerWriter { + private queue: LedgerEntry[] = [] + private flushing = false + private failures = 0 + private retryAt = 0 + private timer: ReturnType | null = null + private dropped = 0 + private written = 0 + private summaryAt: number + + constructor( + private readonly database: RelayDatabase, + private readonly input: { + now?: () => number + queueMax?: number + log?: (line: string) => void + } = {} + ) { + this.summaryAt = this.now() + } + + private now(): number { + return (this.input.now ?? Date.now)() + } + + start(): void { + this.timer ??= setInterval(() => void this.flush(), RESERVE_LEDGER_FLUSH_MS) + this.timer.unref?.() + } + + stop(): void { + if (this.timer) clearInterval(this.timer) + this.timer = null + } + + // During a database stall the queue grows; past its bound the oldest entries go first, + // since the 10-minute reconcile rewrites every seat anyway. + enqueue(entry: LedgerEntry): void { + this.queue.push(entry) + const max = this.input.queueMax ?? RESERVE_LEDGER_QUEUE_MAX + if (this.queue.length > max) { + const excess = this.queue.length - max + this.queue.splice(0, excess) + this.dropped += excess + } + } + + pending(): number { + return this.queue.length + } + + // Single-flight; resolves once the queued rows are written or the attempt failed. + async flush(): Promise { + if (this.flushing || this.queue.length === 0 || this.now() < this.retryAt) { + this.logSummary() + return + } + this.flushing = true + const batch = this.queue.slice(0, RESERVE_LEDGER_BATCH) + try { + await this.write(batch) + this.queue.splice(0, batch.length) + this.written += batch.length + this.failures = 0 + this.retryAt = 0 + } catch { + this.failures += 1 + this.retryAt = this.now() + Math.min(RETRY_MAX_MS, RETRY_BASE_MS * 2 ** (this.failures - 1)) + } finally { + this.flushing = false + this.logSummary() + } + } + + private async write(batch: LedgerEntry[]): Promise { + // The newest entry per host wins inside one batch. + const latest = new Map() + for (const entry of batch) { + const key = `${entry.userId}\u0000${entry.relayHostId}` + const known = latest.get(key) + if (!known || entry.epoch > known.epoch) latest.set(key, entry) + } + await this.upsert([...latest.values()]) + } + + private async upsert(entries: LedgerEntry[]): Promise { + const now = this.now() + const rows = entries.map(() => '(?, ?, ?, ?, ?, ?, 0, 0, 0, 0, 0, 0)').join(', ') + const zeroOnMove = (column: string) => + `${column} = CASE WHEN relay_assignments.cell_id = excluded.cell_id + THEN relay_assignments.${column} ELSE 0 END` + await this.database.query( + `INSERT INTO relay_assignments + (user_id, relay_host_id, cell_id, assignment_epoch, lease_expires_at, last_activity_at, + reserved_controls, reserved_splices, reserved_invites, pending_installs, + pending_confirmations, migration_leases) + VALUES ${rows} + ON CONFLICT (user_id, relay_host_id) DO UPDATE SET + ${[ + 'reserved_controls', + 'reserved_splices', + 'reserved_invites', + 'pending_installs', + 'pending_confirmations', + 'migration_leases' + ] + .map(zeroOnMove) + .join(',\n ')}, + cell_id = excluded.cell_id, + assignment_epoch = excluded.assignment_epoch, + last_activity_at = excluded.last_activity_at + WHERE relay_assignments.assignment_epoch < excluded.assignment_epoch`, + entries.flatMap((entry) => [ + entry.userId, + entry.relayHostId, + entry.cellId, + entry.epoch, + now, + now + ]) + ) + } + + // Rows for these hosts, in one read per batch. + async readRows( + hosts: ReadonlyArray<{ userId: string; relayHostId: string }> + ): Promise> { + const rows = new Map() + for (let start = 0; start < hosts.length; start += RESERVE_LEDGER_BATCH) { + const chunk = hosts.slice(start, start + RESERVE_LEDGER_BATCH) + const found = await this.database.query( + `SELECT user_id, relay_host_id, cell_id, assignment_epoch FROM relay_assignments + WHERE ${chunk.map(() => '(user_id = ? AND relay_host_id = ?)').join(' OR ')}`, + chunk.flatMap((host) => [host.userId, host.relayHostId]) + ) + for (const row of found) { + rows.set(`${String(row.user_id)}\u0000${String(row.relay_host_id)}`, { + cellId: String(row.cell_id), + epoch: Number(row.assignment_epoch) + }) + } + } + return rows + } + + private logSummary(): void { + const now = this.now() + if (now - this.summaryAt < SUMMARY_INTERVAL_MS) return + this.summaryAt = now + ;(this.input.log ?? ((line: string) => console.log(line)))( + JSON.stringify({ + event: 'orca_relay_reserve_ledger_summary', + written: this.written, + pending: this.queue.length, + dropped: this.dropped, + failures: this.failures + }) + ) + this.written = 0 + this.dropped = 0 + } +} + +export const RESERVE_LEDGER_RECONCILE_MS = 10 * 60_000 +// A seat this young may still be superseded; the next reconcile takes it. +const RECONCILE_SETTLE_MS = 30_000 + +type ReconcileSeat = { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number } + +// Every director upserts the map's seats on reserve-mode cells: guarded and idempotent, so +// concurrent runs are harmless. A row ahead of the seat (today's path answered a newer epoch, +// and the desktop was then answered the old seat from memory) is never rewritten: that seat is +// demoted (epoch-checked, closes with 4409), so the desktop re-assigns to what the row names. +// The demoting director then mints above the row's epoch for that host (see raiseEpochFloor). +export async function reconcileReserveLedger(input: { + writer: ReserveLedgerWriter + seats: () => ReconcileSeat[] + demote?: (seat: ReconcileSeat, row: { cellId: string; epoch: number }) => void + now: number +}): Promise<{ upserted: number; demoted: number }> { + const seats = input.seats().filter((seat) => input.now - seat.joinedAt >= RECONCILE_SETTLE_MS) + const rows = await input.writer.readRows(seats) + let upserted = 0 + let demoted = 0 + for (const seat of seats) { + const row = rows.get(`${seat.userId}\u0000${seat.relayHostId}`) + // Level with the row on another cell is the same divergence: the mirror never replaces it. + const behindRow = row && (row.epoch > seat.epoch || (row.epoch === seat.epoch && row.cellId !== seat.cellId)) + if (row && behindRow && input.demote) { + input.demote(seat, row) + demoted += 1 + continue + } + if (row && row.epoch >= seat.epoch) continue + input.writer.enqueue(seat) + upserted += 1 + } + return { upserted, demoted } +} diff --git a/cloud/apps/relay/src/shadow-seat-directory.ts b/cloud/apps/relay/src/shadow-seat-directory.ts index 302bbd2b777..7255284cdac 100644 --- a/cloud/apps/relay/src/shadow-seat-directory.ts +++ b/cloud/apps/relay/src/shadow-seat-directory.ts @@ -284,6 +284,15 @@ export class ShadowSeatDirectory { return [...this.cells.keys()] } + // One cell's seats as the map holds them. + seatsOnCell(cellId: string): Array<{ userId: string; relayHostId: string; seat: ShadowSeat }> { + const cursor = this.cells.get(cellId) + if (!cursor) return [] + return [...cursor.seats].map(([key, seat]) => { + const separator = key.indexOf('\u0000') + return { userId: key.slice(0, separator), relayHostId: key.slice(separator + 1), seat } + }) + } // An old cell (404) has no feed; its last known seats stay, marked by status. markNoFeed(cellId: string, now: number, runtimeIncarnation?: string): void {