diff --git a/cloud/apps/relay/src/cell-reserve-admission.blackbox.test.ts b/cloud/apps/relay/src/cell-reserve-admission.blackbox.test.ts index a952679ff53..543d40040f5 100644 --- a/cloud/apps/relay/src/cell-reserve-admission.blackbox.test.ts +++ b/cloud/apps/relay/src/cell-reserve-admission.blackbox.test.ts @@ -330,6 +330,19 @@ describe('a reserve-mode cell admits from memory with its database wedged', () = expect(cell.book(identity, 9)).toEqual([{ outcome: 'ok' }]) }) + it('leaves a drained seat to its drain grace instead of the shorter demotion close', async () => { + const cell = await startCell({ admitMode: 'reserve' }) + const identity = await cell.host() + cell.book(identity, 9) + await cell.connect(identity, 9) + const page = cell.relay.sessions.seatFeed(null) + const joinedAt = 'full' in page ? page.full[0]!.joinedAt : -1 + cell.relay.sessions.drain(120_000) + const seat = { v: 1 as const, userId: 'user-1', relayHostId: identity.hostId, epoch: 9, joinedAt } + expect(cell.relay.sessions.demote(seat)).toBe('draining') + expect(cell.relay.sessions.demote(seat)).toBe('draining') + }) + it('gives every booking unit back when the cell starts to drain', async () => { const cell = await startCell({ admitMode: 'reserve' }) const identity = await cell.host() diff --git a/cloud/apps/relay/src/host-session-registry.ts b/cloud/apps/relay/src/host-session-registry.ts index 1df66ae5edb..4bbbf896830 100644 --- a/cloud/apps/relay/src/host-session-registry.ts +++ b/cloud/apps/relay/src/host-session-registry.ts @@ -719,7 +719,7 @@ export class HostSessionRegistry { // A director names this seat the loser of a duplicate: stop taking phones for it now, and // close it once its splices have had time to finish. Repeats are one demotion. - demote(request: DemoteRequest): 'demoted' | 'already-demoted' | 'not-seated' | 'off' { + demote(request: DemoteRequest): 'demoted' | 'already-demoted' | 'draining' | 'not-seated' | 'off' { // Only a reserve-mode cell takes demotions, so flipping admitMode back reverts the route. if (!this.reserveMode()) return 'off' const session = this.get(request) @@ -735,6 +735,8 @@ export class HostSessionRegistry { ) { return 'not-seated' } + // A drain already told this host to go: its own grace stands, never cut to the demotion's. + if (this.drainSentHosts.has(session.relayHostId) || session.regionalDrainAttemptId) return 'draining' const socket = session.socket if (this.demotedSeats.get(key) === socket) return 'already-demoted' this.demotedSeats.set(key, socket) diff --git a/cloud/apps/relay/src/relay-server.ts b/cloud/apps/relay/src/relay-server.ts index 078a799ab62..5d69a392772 100644 --- a/cloud/apps/relay/src/relay-server.ts +++ b/cloud/apps/relay/src/relay-server.ts @@ -255,6 +255,7 @@ export function createRelayServer( epoch: join.epoch }) } + directory.onOutranked = (seat) => reservePlacement?.supersedeOutranked(seat) reserveLedger.start() const reconcile = (): void => { const now = (options.now ?? Date.now)() diff --git a/cloud/apps/relay/src/reserve-assignment.test.ts b/cloud/apps/relay/src/reserve-assignment.test.ts index 56c465fd0d5..b385c3ab1a7 100644 --- a/cloud/apps/relay/src/reserve-assignment.test.ts +++ b/cloud/apps/relay/src/reserve-assignment.test.ts @@ -350,6 +350,17 @@ describe('reserve assignment on a director', () => { expect(demoted).toEqual([{ cellId: 'c1', epoch: 5, joinedAt: 1_000 }]) }) + // The booking map had not seen the c1 seat (half-open, joined within a poll lag): the feed + // demotes it once both joins are in the map, instead of the 75 s silence close. + it('tells an old reserve seat the booking missed to go once a newer join shows in the feed', () => { + const { assignment, directory, demoted, now } = setup() + directory.onOutranked = (seat) => assignment.supersedeOutranked(seat) + directory.apply('c2', feed('c2', { full: [seated(6, 3_000)] }), now.value, now.value) + expect(demoted).toEqual([]) + directory.apply('c1', feed('c1', { full: [seated(5, 1_000)] }), now.value, now.value) + expect(demoted).toEqual([{ cellId: 'c1', epoch: 5, joinedAt: 1_000 }]) + }) + it('answers a reconnect only at the newest epoch, newest join first', async () => { const { assignment, reserved } = setup({ feeds: [feed('c1', { full: [seated(6, 1_000)] }), feed('c2', { full: [seated(5, 3_000)] })] diff --git a/cloud/apps/relay/src/reserve-assignment.ts b/cloud/apps/relay/src/reserve-assignment.ts index 47fe8d66883..ac5da96678f 100644 --- a/cloud/apps/relay/src/reserve-assignment.ts +++ b/cloud/apps/relay/src/reserve-assignment.ts @@ -13,7 +13,7 @@ import { type PlacementCell, type ReservePlacer } from './reserve-placement.js' -import type { SeatFeedCell, ShadowSeat, ShadowSeatDirectory } from './shadow-seat-directory.js' +import type { OutrankedSeat, SeatFeedCell, ShadowSeat, ShadowSeatDirectory } from './shadow-seat-directory.js' // Step 5 on a director: answer a desktop from the in-memory map for cells that report // admitMode=reserve, and fall back to today's database path for everything else. The cell @@ -377,10 +377,10 @@ export class ReserveAssignment { 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 { + // Fire-and-forget; the cell checks the seat's epoch and join itself. False: no such cell listed. + demoteSeat(seat: { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }): boolean { const cell = this.input.cells().find((entry) => entry.cellId === seat.cellId) - if (!cell) return + if (!cell) return false void this.input.client.demote(cell, { v: 1, userId: seat.userId, @@ -388,6 +388,7 @@ export class ReserveAssignment { epoch: seat.epoch, joinedAt: seat.joinedAt }) + return true } // Supersede at placement: the old seats this director's map holds on reserve cells are told @@ -396,27 +397,34 @@ export class ReserveAssignment { for (const seat of seats) { if (seat.cellId === cellId || seat.epoch >= epoch) continue if (this.input.directory.admitModeOf(seat.cellId) !== 'reserve') continue - const cell = this.input.cells().find((entry) => entry.cellId === seat.cellId) - if (!cell) continue - this.summary.supersedes += 1 - void this.input.client.demote(cell, { - v: 1, - userId: identity.userId, - relayHostId: identity.relayHostId, - epoch: seat.epoch, - joinedAt: seat.joinedAt - }) - this.input.log?.( - JSON.stringify({ - event: 'orca_relay_reserve_supersede', - relayHostIdDigest: relayHostLogDigest(identity.relayHostId), - fromCellId: seat.cellId, - toCellId: cellId - }) - ) + this.demoteSuperseded({ ...identity, ...seat }, cellId, 'placement') } } + // The feed's half of supersede: a seat the booking map did not hold, found once both joins + // are in this map. Without it that seat holds its cell until the 75 s silence close. + supersedeOutranked(seat: OutrankedSeat): void { + this.demoteSuperseded(seat, seat.byCellId, 'feed') + } + + private demoteSuperseded( + seat: { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }, + toCellId: string, + via: 'placement' | 'feed' + ): void { + if (!this.demoteSeat(seat)) return + this.summary.supersedes += 1 + this.input.log?.( + JSON.stringify({ + event: 'orca_relay_reserve_supersede', + relayHostIdDigest: relayHostLogDigest(seat.relayHostId), + fromCellId: seat.cellId, + toCellId, + via + }) + ) + } + private flushSummary(): void { const now = this.now() if (now - this.summaryAt < SUMMARY_INTERVAL_MS) return diff --git a/cloud/apps/relay/src/reserve-placement-simulation.test.ts b/cloud/apps/relay/src/reserve-placement-simulation.test.ts index 6dae3349684..44f2cafa24e 100644 --- a/cloud/apps/relay/src/reserve-placement-simulation.test.ts +++ b/cloud/apps/relay/src/reserve-placement-simulation.test.ts @@ -294,6 +294,8 @@ describe('step 5 deterministic simulation', () => { [1, nearFull({ bookPastCap: true })], [2, drain({ intakeBurst: 200, directorShare: 0.05 })], [3, drainedWithoutSupersede()], + // Half-open seats the booking map had not seen hold their cell until the silence close. + [3, levelSeats(11, { skipFeedSupersede: true })], [4, levelSeats(11, { localTieBreak: true })], [5, mixed({ unguardedLedger: true })], // A director restart mid-queue loses its booked joins' rows until its boot reconcile. diff --git a/cloud/apps/relay/src/reserve-placement-simulation.ts b/cloud/apps/relay/src/reserve-placement-simulation.ts index 8eaa673d189..c6b9483ab61 100644 --- a/cloud/apps/relay/src/reserve-placement-simulation.ts +++ b/cloud/apps/relay/src/reserve-placement-simulation.ts @@ -9,6 +9,7 @@ import { ReservePlacer, type PlacementCell } from './reserve-placement.js' +import { seatOutranks } from './shadow-seat-directory.js' // A seeded discrete-event model of step 5 that drives the real CellReserveBook and // ReservePlacer. Everything else (sockets, the database, desktops) is modelled. The checks @@ -111,8 +112,10 @@ export type SimulationFaults = { intakeBurst?: number // Each director assumes this share of every cell's intake, instead of a fifth. directorShare?: number - // Placement does not supersede the old seat: a duplicate then outlives its bound. + // Nothing supersedes the old seat, at placement or from the feed: a duplicate outlives its bound. skipSupersede?: boolean + // Only placement supersedes: a half-open seat the booking map had not seen lasts to its silence close. + skipFeedSupersede?: boolean // A restarted director forgets its ledger queue and never reconciles. noBootReconcile?: boolean // The reconcile leaves a seat whose row is ahead of it alone. @@ -666,10 +669,15 @@ export async function runReservePlacementSimulation( placement: placementCell(cell, snapshot, requestedAt) } director.views.set(cell.cellId, current) + supersedeOutranked(director, seats!.keys()) } else if (current && current.incarnation === snapshot.incarnation) { for (const change of changes) applyChange(director, cell, current, change) current.cursor = snapshot.cursor current.placement = placementCell(cell, snapshot, requestedAt) + supersedeOutranked( + director, + changes.filter((change) => change.kind === 'join').map((change) => change.host) + ) } else { // A restarted director's (or a stale) delta: the next poll asks for a full snapshot. return @@ -758,6 +766,26 @@ export async function runReservePlacementSimulation( } } + // As shadow-seat-directory.ts onOutranked: once both joins are in this map, a reserve seat behind + // a later join at a newer epoch is told to go. A lower epoch that joined later is the row's. + function supersedeOutranked(director: SimDirector, hosts: Iterable): void { + if (faults.skipSupersede || faults.skipFeedSupersede) return + for (const host of new Set(hosts)) { + const seats = [...director.views].flatMap(([cellId, view]) => { + const seat = view.seats.get(host) + return seat ? [{ cellId, view, seat }] : [] + }) + if (seats.length < 2) continue + for (const { cellId, view, seat } of seats) { + if (seat.demoted || !view.placement.reserve) continue + const by = seats.find((other) => seatOutranks(other.seat, seat)) + if (!by) continue + report.supersedes += 1 + demoteFor(director, host, cellId, seat, { cellId: by.cellId, epoch: by.seat.epoch }) + } + } + } + // The seats the booking director's map held when it placed each booking: only those it supersedes. const heldAtBooking = new Map>() @@ -1258,9 +1286,9 @@ export async function runReservePlacementSimulation( report.maxOccupancyShare[cell.cellId] = Math.max(report.maxOccupancyShare[cell.cellId] ?? 0, share) } } - // A booking supersedes the old seats its director's map held within a round trip. Any other - // duplicate (a half-open seat today's path placed past, or one the booking map had not yet - // seen) has no lease to re-check and lasts until that seat's silence close. + // A booking supersedes the old seats its director's map held within a round trip, and the feed + // any seat a later join outranks. Any other duplicate (a half-open seat today's path placed past + // at a lower epoch than it) has no lease to re-check and lasts until that seat's silence close. const supersedeBound = Math.max(...directors.map((director) => config.pollLagMs(director.index))) + 2 * Math.max(...cells.map((cell) => config.rttMs(cell.region))) + @@ -1293,13 +1321,20 @@ export async function runReservePlacementSimulation( for (const { cell, seat } of seats) { if (cell.mode !== 'reserve' || seat.epoch === newest) continue // A leased seat goes by its renewal, as today. - const bound = !seat.leased && supersededBy(cell.cellId, seat.joinedAt) ? supersedeBound : duplicateBound + const tight = + !seat.leased && + (supersededBy(cell.cellId, seat.joinedAt) || seats.some((other) => seatOutranks(other.seat, seat))) + // The loose clock runs from the first outranking; the tight one from when supersede applies. const key = `${host}\u0000${cell.cellId}\u0000${seat.joinedAt}` - live.add(key) - const since = outrankedSince.get(key) ?? clock.now - outrankedSince.set(key, since) - if (clock.now - since > bound) { - violate(3, `${host}@${seat.epoch} on ${cell.cellId} outranked for ${clock.now - since} ms`) + const clocks: Array<[string, number]> = [[key, duplicateBound]] + if (tight) clocks.push([`${key}\u0000tight`, supersedeBound]) + for (const [clockKey, bound] of clocks) { + live.add(clockKey) + const since = outrankedSince.get(clockKey) ?? clock.now + outrankedSince.set(clockKey, since) + if (clock.now - since > bound) { + violate(3, `${host}@${seat.epoch} on ${cell.cellId} outranked for ${clock.now - since} ms`) + } } } } diff --git a/cloud/apps/relay/src/shadow-seat-directory.test.ts b/cloud/apps/relay/src/shadow-seat-directory.test.ts index 09cc2a352dd..fa0d4635d24 100644 --- a/cloud/apps/relay/src/shadow-seat-directory.test.ts +++ b/cloud/apps/relay/src/shadow-seat-directory.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest' import { RelayAssignmentStore } from './assignment-store.js' import { openInMemoryRelayDatabase } from './database.js' import { + SHADOW_SEAT_OUTRANK_SKEW_MS, SHADOW_SEAT_RECENTLY_LEFT_MAX_HOSTS, ShadowSeatDirectory, startShadowSeatPoller, @@ -577,3 +578,76 @@ describe('ShadowSeatDirectory.placementModeOf', () => { expect(directory.placementModeOf('cell-a')).toBe('reserve') }) }) + +describe('ShadowSeatDirectory.onOutranked', () => { + const reserve = { generation: 3, flags: { admitMode: 'reserve' } } + const db = { generation: 3, flags: { admitMode: 'db' } } + const full = (cellId: string, epoch: number, joinedAt: number, flags = reserve, state = 'active') => + feed({ + cellId, + seq: 1, + flagsApplied: flags, + full: [{ userId: 'user-a', relayHostId: 'host-1', epoch, generation: 1, state, joinedAt }] + }) + const directory = () => { + const map = new ShadowSeatDirectory() + map.setCells(['cell-a', 'cell-b']) + const outranked: unknown[] = [] + map.onOutranked = (seat) => outranked.push(seat) + return { map, outranked } + } + const halfOpen = { + cellId: 'cell-a', + userId: 'user-a', + relayHostId: 'host-1', + epoch: 3, + joinedAt: 10, + byCellId: 'cell-b' + } + + it('reports a reserve seat a later, newer join elsewhere outranks, in either feed order', () => { + const first = directory() + first.map.apply('cell-a', full('cell-a', 3, 10), 100) + first.map.apply('cell-b', full('cell-b', 4, 2_000), 101) + expect(first.outranked).toEqual([halfOpen]) + const reversed = directory() + reversed.map.apply('cell-b', full('cell-b', 4, 2_000), 100) + reversed.map.apply('cell-a', full('cell-a', 3, 10), 101) + expect(reversed.outranked).toEqual([halfOpen]) + const justPastSkew = directory() + justPastSkew.map.apply('cell-a', full('cell-a', 3, 10), 100) + justPastSkew.map.apply('cell-b', full('cell-b', 4, 11 + SHADOW_SEAT_OUTRANK_SKEW_MS), 101) + expect(justPastSkew.outranked).toEqual([halfOpen]) + }) + + it('reports a seat once, even when a full resync re-judges its host', () => { + const { map, outranked } = directory() + map.apply('cell-a', full('cell-a', 3, 10), 100) + map.apply('cell-b', full('cell-b', 4, 2_000), 101) + map.apply('cell-b', { ...full('cell-b', 4, 2_000), incarnation: '22222222-2222-4222-8222-222222222222' }, 102) + map.apply('cell-a', { ...full('cell-a', 3, 10), incarnation: '22222222-2222-4222-8222-222222222222' }, 103) + expect(outranked).toEqual([halfOpen]) + }) + + it('leaves a database-cell seat, a drain-only seat, a later lower epoch, and a join within the skew', () => { + const onDatabaseCell = directory() + onDatabaseCell.map.apply('cell-a', full('cell-a', 3, 10, db), 100) + onDatabaseCell.map.apply('cell-b', full('cell-b', 4, 2_000), 101) + const drainOnly = directory() + drainOnly.map.apply('cell-a', full('cell-a', 3, 10, reserve, 'drain-only'), 100) + drainOnly.map.apply('cell-b', full('cell-b', 4, 2_000), 101) + // Answered the old seat from a stale map after the newer join: the row decides that one. + const staleSticky = directory() + staleSticky.map.apply('cell-a', full('cell-a', 3, 3_000), 100) + staleSticky.map.apply('cell-b', full('cell-b', 4, 2_000), 101) + const withinSkew = directory() + withinSkew.map.apply('cell-a', full('cell-a', 3, 10), 100) + withinSkew.map.apply('cell-b', full('cell-b', 4, 10 + SHADOW_SEAT_OUTRANK_SKEW_MS), 101) + expect([onDatabaseCell, drainOnly, staleSticky, withinSkew].map(({ outranked }) => outranked)).toEqual([ + [], + [], + [], + [] + ]) + }) +}) diff --git a/cloud/apps/relay/src/shadow-seat-directory.ts b/cloud/apps/relay/src/shadow-seat-directory.ts index 144270bd8f3..7dacf6a21df 100644 --- a/cloud/apps/relay/src/shadow-seat-directory.ts +++ b/cloud/apps/relay/src/shadow-seat-directory.ts @@ -16,6 +16,16 @@ export const SHADOW_SEAT_POLL_TIMEOUT_MS = 2_000 export const SHADOW_SEAT_CELL_LIST_REFRESH_MS = 30_000 export const SHADOW_SEAT_RECENTLY_LEFT_TTL_MS = 24 * 60 * 60 * 1_000 export const SHADOW_SEAT_RECENTLY_LEFT_MAX_HOSTS = 100_000 +// Cells stamp joinedAt by their own clocks: a later join must clear this to outrank across cells. +export const SHADOW_SEAT_OUTRANK_SKEW_MS = 1_000 +const SHADOW_SEAT_OUTRANKED_REPORTS_MAX = 100_000 + +export function seatOutranks( + newer: { epoch: number; joinedAt: number }, + seat: { epoch: number; joinedAt: number } +): boolean { + return newer.epoch > seat.epoch && newer.joinedAt > seat.joinedAt + SHADOW_SEAT_OUTRANK_SKEW_MS +} const RECENTLY_LEFT_PER_HOST = 4 // Google identity tokens live an hour; one per poll would be 26 metadata reads a second. const IDENTITY_TOKEN_REUSE_MS = 10 * 60_000 @@ -172,6 +182,16 @@ export type BookedJoin = { reservedBy: string } +// A reserve seat behind a later, newer-epoch join of its host elsewhere that placement supersede missed. +export type OutrankedSeat = { + cellId: string + userId: string + relayHostId: string + epoch: number + joinedAt: number + byCellId: string +} + type CellCursor = SeatFeedCellState & { seats: Map } export type HeartbeatSnapshot = { readAt: number; expiresAt: ReadonlyMap } @@ -180,6 +200,11 @@ function hostKey(userId: string, relayHostId: string): string { return `${userId}\u0000${relayHostId}` } +function parseHostKey(key: string): { userId: string; relayHostId: string } { + const separator = key.indexOf('\u0000') + return { userId: key.slice(0, separator), relayHostId: key.slice(separator + 1) } +} + export class ShadowSeatDirectory { private readonly cells = new Map() private readonly hosts = new Map>() @@ -193,6 +218,10 @@ export class ShadowSeatDirectory { // Per host, the newest booking any cell reported (step 5 epoch mint). Age-ordered. private readonly bookings = new Map() onBookedJoin?: (join: BookedJoin) => void + onOutranked?: (seat: OutrankedSeat) => void + private readonly joinedSinceApply = new Set() + // Seats already reported, insertion-ordered: a full resync re-judges a host without a second demote. + private readonly reportedOutranked = new Set() // Cells no longer listed are forgotten with their seats; new ones start pending. // `required` defaults to every listed cell. @@ -225,6 +254,11 @@ export class ShadowSeatDirectory { // `polledAt` is when the request left; the reply holds every change before it. apply(cellId: string, response: SeatFeedResponse, now: number, polledAt = now): void { + this.applyResponse(cellId, response, now, polledAt) + this.reportOutranked() + } + + private applyResponse(cellId: string, response: SeatFeedResponse, now: number, polledAt: number): void { const cursor = this.cells.get(cellId) if (!cursor) return if (response.cellId !== cellId) { @@ -307,10 +341,7 @@ export class ShadowSeatDirectory { 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 } - }) + return [...cursor.seats].map(([key, seat]) => ({ ...parseHostKey(key), seat })) } // An old cell (404) has no feed; its last known seats stay, marked by status. @@ -558,6 +589,34 @@ export class ShadowSeatDirectory { this.hosts.set(key, seats) } seats.set(cursor.cellId, seat) + if (this.onOutranked && seats.size > 1) this.joinedSinceApply.add(key) + } + + // After the whole response, so a cell's first snapshot is judged by the mode it carries. Only a + // join triggers it, once per seat: a cell flipped to reserve later, or a lost demote, falls to + // the silence close or the ledger reconcile, as does a lower epoch that joined later. + private reportOutranked(): void { + const report = this.onOutranked + const keys = [...this.joinedSinceApply] + this.joinedSinceApply.clear() + if (!report) return + for (const key of keys) { + const seats = [...(this.hosts.get(key)?.values() ?? [])] + for (const seat of seats) { + if (seat.state !== 'active' || this.admitModeOf(seat.cellId) !== 'reserve') continue + const by = seats.find((other) => seatOutranks(other, seat)) + if (!by) continue + const { cellId, epoch, joinedAt } = seat + const reported = `${key}\u0000${cellId}\u0000${epoch}\u0000${joinedAt}` + if (this.reportedOutranked.has(reported)) continue + this.reportedOutranked.add(reported) + if (this.reportedOutranked.size > SHADOW_SEAT_OUTRANKED_REPORTS_MAX) { + const oldest = this.reportedOutranked.values().next().value + if (oldest !== undefined) this.reportedOutranked.delete(oldest) + } + report({ ...parseHostKey(key), cellId, epoch, joinedAt, byCellId: by.cellId }) + } + } } private unseat(cursor: CellCursor, key: string): void {