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 543d40040f5..709f7e21987 100644 --- a/cloud/apps/relay/src/cell-reserve-admission.blackbox.test.ts +++ b/cloud/apps/relay/src/cell-reserve-admission.blackbox.test.ts @@ -8,10 +8,13 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import WebSocket from 'ws' import type { RawData } from 'ws' import { CELL_FLAG_DEFAULTS, type CellFlags } from './cell-flags.js' +import { CELL_SEAT_DRIFT_SWEEP_MS } from './cell-seat-drift.js' +import type { CellSeatLog } from './cell-seat-log.js' import type { RelayConfig } from './config.js' import { PostgresDatabase } from './database.js' import type { AppliedControlFlags } from './relay-control-flag-channel.js' import { createRelayServer, HOST_HELLO_SHED_OLDEST_WAIT_MS } from './relay-server.js' +import { RELAY_WEBSOCKET_FORCE_CLOSE_MS } from './relay-websocket-close.js' // A step-5 cell whose database pool is wedged: every query waits forever and the hello shed // is armed. Booked and recently seated hosts must still connect; everything else must not. @@ -71,6 +74,20 @@ function cellConfig(port: number, issuer: string): RelayConfig { } } +// The registry's private feed log, to plant the drift the cell must repair. +function seatLogOf(sessions: object): CellSeatLog { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: HostSessionRegistry holds its feed log as `seatLog`. + return (sessions as { seatLog: CellSeatLog }).seatLog +} + +// The JSON lines a test's console.warn spy saw for one event. +function warned(event: string): Array> { + return vi + .mocked(console.warn) + .mock.calls.map(([line]) => (typeof line === 'string' && line.startsWith('{') ? JSON.parse(line) : null)) + .filter((entry) => entry?.event === event) +} + function nextMessage(socket: WebSocket): Promise> { return new Promise((resolve, reject) => { socket.once('message', (data: RawData) => { @@ -249,6 +266,9 @@ describe('a reserve-mode cell admits from memory with its database wedged', () = book, setFlags: (next: Partial) => { applied = { generation: applied.generation + 1, flags: { ...CELL_FLAG_DEFAULTS, ...next } } + }, + advance: (ms: number) => { + elapsedMs += ms } } } @@ -330,6 +350,150 @@ describe('a reserve-mode cell admits from memory with its database wedged', () = expect(cell.book(identity, 9)).toEqual([{ outcome: 'ok' }]) }) + describe('keeps its seat feed on its live sessions', () => { + const seatEpochs = (cell: { relay: { sessions: { seatFeed(since: number | null): unknown } } }) => { + const page = cell.relay.sessions.seatFeed(null) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a null cursor always answers a full page. + return (page as { full: Array<{ relayHostId: string; epoch: number }> }).full.map((entry) => entry.epoch) + } + + it('repairs at once the seat a director names when no session backs it', async () => { + const cell = await startCell({ admitMode: 'reserve' }) + const identity = await cell.host() + cell.book(identity, 9) + await cell.connect(identity, 9) + // A stale seat at a generation above the live session's: its changes could never replace it. + const log = seatLogOf(cell.relay.sessions) + log.append({ kind: 'join', userId: 'user-1', relayHostId: identity.hostId, epoch: 8, generation: 5, at: 4_242 }) + const before = cell.relay.sessions.seatFeed(null) + const seq = 'full' in before ? before.seq : 0 + const seat = { v: 1 as const, userId: 'user-1', relayHostId: identity.hostId, epoch: 8, joinedAt: 4_242 } + expect(cell.relay.sessions.demote(seat)).toBe('not-seated') + expect(cell.relay.sessions.seatFeed(seq)).toMatchObject({ + changes: [ + { kind: 'leave', epoch: 8, generation: 5, closeCode: 4409 }, + { kind: 'join', epoch: 9, generation: 1 } + ] + }) + expect(seatEpochs(cell)).toEqual([9]) + expect(warned('orca_relay_cell_seat_log_drift')).toMatchObject([ + { kind: 'unbacked-seat', via: 'demote', reason: 'epoch', seatEpoch: 8, sessionEpoch: 9, relisted: true } + ]) + expect(JSON.stringify(warned('orca_relay_cell_seat_log_drift'))).not.toContain(identity.hostId) + // Asked again, there is nothing left to repair. + expect(cell.relay.sessions.demote(seat)).toBe('not-seated') + expect(cell.relay.sessions.seatFeed(seq + 2)).toMatchObject({ changes: [] }) + }) + + it('leaves a seat no session backs once two sweeps in a row find it, with no director asking', async () => { + const cell = await startCell({ admitMode: 'db' }) + const ghost = await cell.host() + seatLogOf(cell.relay.sessions).append({ + kind: 'join', + userId: 'user-1', + relayHostId: ghost.hostId, + epoch: 10, + generation: 3, + at: 1_000 + }) + expect(seatEpochs(cell)).toEqual([10]) + cell.advance(CELL_SEAT_DRIFT_SWEEP_MS) + expect(seatEpochs(cell)).toEqual([]) + expect(warned('orca_relay_cell_seat_log_drift')).toMatchObject([ + { kind: 'unbacked-seat', via: 'sweep', reason: 'no-session', seatEpoch: 10, seatGeneration: 3, relisted: false } + ]) + // Left as moved: the cell will not seat it again at 10 from memory. + expect(cell.relay.sessions.seatFeed(0)).toMatchObject({ changes: [{}, { kind: 'leave', closeCode: 4409 }] }) + }) + + it('lists again a live session the feed lost, once two sweeps in a row find it', async () => { + const cell = await startCell({ admitMode: 'reserve' }) + const identity = await cell.host() + cell.book(identity, 9) + await cell.connect(identity, 9) + seatLogOf(cell.relay.sessions).append({ + kind: 'leave', + userId: 'user-1', + relayHostId: identity.hostId, + epoch: 9, + generation: 1, + at: 5_000 + }) + expect(seatEpochs(cell)).toEqual([]) + cell.advance(CELL_SEAT_DRIFT_SWEEP_MS) + expect(seatEpochs(cell)).toEqual([9]) + expect(warned('orca_relay_cell_seat_log_drift')).toMatchObject([{ kind: 'unlisted-session', sessionEpoch: 9 }]) + }) + + // A drained session's seat stays listed, with no session, until its socket's close lands. + it('repairs nothing for a seat that one sweep finds unbacked and is gone by the next', async () => { + const cell = await startCell({ admitMode: 'reserve' }) + const identity = await cell.host() + cell.book(identity, 9) + const live = await cell.connect(identity, 9) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: ws keeps its net.Socket as `_socket`. + ;(live.socket as unknown as { _socket: { pause(): void } })._socket.pause() + cell.relay.sessions.drainHost({ + attemptId: 'attempt-1', + userId: 'user-1', + relayHostId: identity.hostId, + sourceAssignmentEpoch: 9, + graceMs: 0 + }) + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(seatEpochs(cell)).toEqual([9]) + await new Promise((resolve) => setTimeout(resolve, RELAY_WEBSOCKET_FORCE_CLOSE_MS + 300)) + cell.advance(CELL_SEAT_DRIFT_SWEEP_MS) + expect(seatEpochs(cell)).toEqual([]) + expect(warned('orca_relay_cell_seat_log_drift')).toEqual([]) + }, 10_000) + + // A drained session leaves the map at once, but its socket closes only after its release (up + // to 3 s) and the peer's close reply (up to the 1 s force-close). A desktop of the same + // identity dialling back in that window restarted at generation 1, so the old socket's late + // leave, at generation 1 too, removed the live seat from the feed. + it('keeps a session that reconnected while its drained predecessor was still closing', async () => { + const cell = await startCell({ admitMode: 'reserve' }) + const identity = await cell.host() + cell.book(identity, 9) + const first = await cell.connect(identity, 9) + expect(first.ack).toMatchObject({ generation: 1 }) + // A peer slow to answer the close frame: the cell's side closes only at its force-close. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: ws keeps its net.Socket as `_socket`. + ;(first.socket as unknown as { _socket: { pause(): void } })._socket.pause() + expect( + cell.relay.sessions.drainHost({ + attemptId: 'attempt-1', + userId: 'user-1', + relayHostId: identity.hostId, + sourceAssignmentEpoch: 9, + graceMs: 0 + }) + ).toBe('accepted') + await new Promise((resolve) => setTimeout(resolve, 20)) + cell.book(identity, 10) + expect((await cell.connect(identity, 10)).ack).toMatchObject({ type: 'host-hello-ack', generation: 2 }) + // Past the force-close, so the old socket's leave has landed. + await new Promise((resolve) => setTimeout(resolve, RELAY_WEBSOCKET_FORCE_CLOSE_MS + 300)) + expect(seatEpochs(cell)).toEqual([10]) + }, 10_000) + + // The same identity connecting afresh at each turn, as desktops sharing one identity do: each + // turn replaces the session and its seat together, so no sweep finds anything to repair. + it('stays on the live session when the same identity connects afresh each turn', async () => { + const cell = await startCell({ admitMode: 'reserve' }) + const identity = await cell.host() + for (const epoch of [9, 10, 11]) { + cell.book(identity, epoch) + expect((await cell.connect(identity, epoch)).ack).toMatchObject({ type: 'host-hello-ack' }) + expect(seatEpochs(cell)).toEqual([epoch]) + cell.advance(CELL_SEAT_DRIFT_SWEEP_MS) + } + expect(seatEpochs(cell)).toEqual([11]) + expect(warned('orca_relay_cell_seat_log_drift')).toEqual([]) + }) + }) + 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() @@ -341,6 +505,8 @@ describe('a reserve-mode cell admits from memory with its database wedged', () = 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') + const after = cell.relay.sessions.seatFeed(null) + expect('full' in after && after.full.map((entry) => entry.epoch)).toEqual([9]) }) it('gives every booking unit back when the cell starts to drain', async () => { diff --git a/cloud/apps/relay/src/cell-reserve-client.ts b/cloud/apps/relay/src/cell-reserve-client.ts index 4daaa0beece..4a52d5e34af 100644 --- a/cloud/apps/relay/src/cell-reserve-client.ts +++ b/cloud/apps/relay/src/cell-reserve-client.ts @@ -2,6 +2,7 @@ import { CELL_DEMOTE_PATH, CELL_RESERVE_MAX_ITEMS, CELL_RESERVE_PATH, + DemoteResponseSchema, ReserveResponseSchema, type DemoteRequest, type ReserveItem @@ -46,14 +47,18 @@ export class CellReserveClient { }) } - // The cell treats repeats as one; a failure is only logged, and the next poll retries. - async demote(target: ReserveTarget, request: DemoteRequest): Promise { + // The cell treats repeats as one. Null: no answer the director can read; the next poll retries. + async demote(target: ReserveTarget, request: DemoteRequest): Promise { try { const response = await this.post(target, CELL_DEMOTE_PATH, request) - await response.body?.cancel().catch(() => undefined) - return response.ok + if (!response.ok) { + await response.body?.cancel().catch(() => undefined) + return null + } + const body = DemoteResponseSchema.safeParse(await response.json().catch(() => null)) + return body.success ? body.data.outcome : null } catch { - return false + return null } } diff --git a/cloud/apps/relay/src/cell-reserve-contract.ts b/cloud/apps/relay/src/cell-reserve-contract.ts index 56d915f9500..5a66381996f 100644 --- a/cloud/apps/relay/src/cell-reserve-contract.ts +++ b/cloud/apps/relay/src/cell-reserve-contract.ts @@ -56,3 +56,8 @@ export const DemoteRequestSchema = z.object({ joinedAt: z.number().int().nonnegative() }) export type DemoteRequest = z.infer + +// `not-seated`: the cell holds no live session for that seat. Lenient for the same reason as +// ReserveResponseSchema; every image since the route shipped answers `outcome`. +export const DemoteResponseSchema = z.object({ v: z.literal(1), outcome: z.string() }).passthrough() +export type DemoteOutcome = 'demoted' | 'already-demoted' | 'draining' | 'not-seated' | 'off' diff --git a/cloud/apps/relay/src/cell-seat-drift.test.ts b/cloud/apps/relay/src/cell-seat-drift.test.ts new file mode 100644 index 00000000000..960400871ae --- /dev/null +++ b/cloud/apps/relay/src/cell-seat-drift.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from 'vitest' +import { CELL_SEAT_DRIFT_SWEEP_MS, SeatDriftConfirmation, unbackedSeatReason } from './cell-seat-drift.js' + +describe('cell seat drift', () => { + const seat = { epoch: 10, generation: 3 } + const live = { state: 'active', socket: {}, assignmentEpoch: 10, generation: 3 } + + it('names why no live session backs a seat, and nothing when one does', () => { + expect(unbackedSeatReason(seat, live)).toBeNull() + expect(unbackedSeatReason(seat, { ...live, state: 'drain-only' })).toBeNull() + expect(unbackedSeatReason(seat, null)).toBe('no-session') + expect(unbackedSeatReason(seat, { ...live, state: 'closed' })).toBe('closed') + expect(unbackedSeatReason(seat, { ...live, socket: null })).toBe('no-socket') + expect(unbackedSeatReason(seat, { ...live, assignmentEpoch: 11 })).toBe('epoch') + expect(unbackedSeatReason(seat, { ...live, generation: 1 })).toBe('generation') + }) + + it('confirms only what two sweeps in a row found, at most once per sweep interval', () => { + const drift = new SeatDriftConfirmation() + expect(drift.due(0)).toBe(true) + expect(drift.confirm(['a', 'b'])).toEqual(new Set()) + expect(drift.due(CELL_SEAT_DRIFT_SWEEP_MS - 1)).toBe(false) + expect(drift.due(CELL_SEAT_DRIFT_SWEEP_MS)).toBe(true) + expect(drift.confirm(['b', 'c'])).toEqual(new Set(['b'])) + expect(drift.confirm(['c'])).toEqual(new Set(['c'])) + }) +}) diff --git a/cloud/apps/relay/src/cell-seat-drift.ts b/cloud/apps/relay/src/cell-seat-drift.ts new file mode 100644 index 00000000000..b0432e107ac --- /dev/null +++ b/cloud/apps/relay/src/cell-seat-drift.ts @@ -0,0 +1,47 @@ +import type { CellSeat } from './cell-seat-log.js' + +// How often the cell checks its seat log against its live sessions, on a director's feed poll. +export const CELL_SEAT_DRIFT_SWEEP_MS = 30_000 + +export type SeatBackingSession = { + state: string + socket: unknown + assignmentEpoch: number + generation: number +} + +export type UnbackedSeatReason = 'no-session' | 'closed' | 'no-socket' | 'epoch' | 'generation' + +// Why no live session backs this seat, or null when one does. A session backs its seat only at +// the epoch and generation it joined with; anything else is a seat the feed should not list. +export function unbackedSeatReason( + seat: Pick, + session: SeatBackingSession | null +): UnbackedSeatReason | null { + if (!session) return 'no-session' + if (session.state === 'closed') return 'closed' + if (!session.socket) return 'no-socket' + if (session.assignmentEpoch !== seat.epoch) return 'epoch' + if (session.generation !== seat.generation) return 'generation' + return null +} + +// Drift found on two sweeps in a row is repaired: a drained session's seat stays listed for the +// few seconds until its socket's close lands, and that is not drift. +export class SeatDriftConfirmation { + private seen = new Set() + private sweptAt = Number.NEGATIVE_INFINITY + + due(now: number): boolean { + if (now - this.sweptAt < CELL_SEAT_DRIFT_SWEEP_MS) return false + this.sweptAt = now + return true + } + + // Returns the ids seen on the previous sweep too, and remembers this sweep's for the next. + confirm(ids: readonly string[]): Set { + const confirmed = new Set(ids.filter((id) => this.seen.has(id))) + this.seen = new Set(ids) + return confirmed + } +} diff --git a/cloud/apps/relay/src/cell-seat-log.ts b/cloud/apps/relay/src/cell-seat-log.ts index 6524976911c..0c3c891f07b 100644 --- a/cloud/apps/relay/src/cell-seat-log.ts +++ b/cloud/apps/relay/src/cell-seat-log.ts @@ -101,6 +101,10 @@ export class CellSeatLog { return { seq: lastSeq, changes, more: lastSeq < this.headSeq } } + seatList(): CellSeat[] { + return [...this.seats.values()] + } + seatOf(userId: string, relayHostId: string): CellSeat | undefined { return this.seats.get(JSON.stringify([userId, relayHostId])) } diff --git a/cloud/apps/relay/src/host-session-registry.test.ts b/cloud/apps/relay/src/host-session-registry.test.ts index 50fcd6aedb5..f0384ca41a7 100644 --- a/cloud/apps/relay/src/host-session-registry.test.ts +++ b/cloud/apps/relay/src/host-session-registry.test.ts @@ -584,6 +584,63 @@ describe('host session cleanup races', () => { ) }) + // The root of a seat c34's feed listed for hours: a rebind whose database activation outlived + // the orphan grace revived the removed session outside the map, still listed and unreachable. + it('refuses a rebind whose session left the map while its activation was in flight', async () => { + const blocked = deferred() + const activateControl = vi + .fn() + .mockResolvedValueOnce('control:production-gce-c3:1') + .mockReturnValueOnce(blocked.promise) + const { registry, activate, releaseActivity } = createRegistry(activateControl) + const key = { userId: identity.sub, relayHostId: identity.relayHostId } + const originalSocket = new FakeSocket() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface. + await activate(originalSocket as unknown as WebSocket, identity, null, 1, false, 1) + const original = registry.get(key) + originalSocket.close(1006, 'network reset') + expect(original?.state).toBe('orphaned') + + const rebindSocket = new FakeSocket() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface. + const rebinding = activate(rebindSocket as unknown as WebSocket, identity, original, 1, true, 1) + await vi.advanceTimersByTimeAsync(0) + expect(activateControl).toHaveBeenCalledTimes(2) + // The grace runs out while the database answers slowly. + await vi.advanceTimersByTimeAsync(CONTROL_CONTINUITY_LIMITS.orphanGraceMs) + expect(registry.get(key)).toBeNull() + blocked.resolve('control:production-gce-c3:1') + await rebinding + + expect(rebindSocket.close).toHaveBeenCalledWith(RELAY_CLOSE_CODE.PEER_DROPPED, 'control activation superseded') + expect(registry.get(key)).toBeNull() + expect(original?.state).toBe('closed') + const page = registry.seatFeed(null) + expect('full' in page && page.full).toEqual([]) + expect(releaseActivity).toHaveBeenCalledWith(key, 'control:production-gce-c3:1') + }) + + // Two hellos proven against the same empty map could take the same generation; an activation at + // a generation this host already had here is refused, so a late leave cannot share it. + it('refuses a fresh activation at a generation this host already had here', async () => { + const activateControl = vi + .fn() + .mockResolvedValue('control:production-gce-c3:2') + const { registry, activate } = createRegistry(activateControl) + const key = { userId: identity.sub, relayHostId: identity.relayHostId } + const first = new FakeSocket() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface. + await activate(first as unknown as WebSocket, identity, null, 2, false, 1) + first.close(1006, 'network reset') + await vi.advanceTimersByTimeAsync(CONTROL_CONTINUITY_LIMITS.orphanGraceMs) + expect(registry.get(key)).toBeNull() + const reused = new FakeSocket() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface. + await activate(reused as unknown as WebSocket, identity, null, 2, false, 1) + expect(reused.close).toHaveBeenCalledWith(RELAY_CLOSE_CODE.PEER_DROPPED, 'control activation superseded') + expect(registry.get(key)).toBeNull() + }) + it('skips a closed queued control so its live retry avoids abandoned database work', async () => { const stalled = deferred() const activateControl = vi diff --git a/cloud/apps/relay/src/host-session-registry.ts b/cloud/apps/relay/src/host-session-registry.ts index fb42d7a1a9e..1821cf65320 100644 --- a/cloud/apps/relay/src/host-session-registry.ts +++ b/cloud/apps/relay/src/host-session-registry.ts @@ -35,7 +35,8 @@ import type { AssignmentLeaseShadow } from './assignment/assignment-lease-shadow import type { RelayAssignmentStore } from './assignment-store.js' import type { CellAdmitMode, CellReserveBook } from './cell-reserve-book.js' import type { DemoteRequest, ReserveOutcome, ReserveRequest } from './cell-reserve-contract.js' -import { CellSeatLog, type CellSeatChange, type CellSeatFeedPage } from './cell-seat-log.js' +import { SeatDriftConfirmation, unbackedSeatReason, type UnbackedSeatReason } from './cell-seat-drift.js' +import { CellSeatLog, type CellSeat, type CellSeatChange, type CellSeatFeedPage } from './cell-seat-log.js' import { isRelayDatabaseTransientError } from './database.js' import { printableCloseReason } from './control-close-reason-text.js' import { ControlRenewalBatch } from './control-renewal-batch.js' @@ -296,6 +297,9 @@ const NO_REJOIN_CLOSE_CODES: ReadonlySet = new Set([ ]) // A demoted duplicate keeps its splices this long, then closes so the desktop re-assigns. export const DEMOTION_CLOSE_MS = 60_000 +// Hosts whose highest generation is remembered; insertion order is age order. A host evicted +// here is long gone, so its successor restarting low meets no late leave. +const GENERATION_FLOOR_MAX_HOSTS = 100_000 // A flip back registers memory-admitted controls with at most this share of the pool in // flight (US 10 -> 3, Asia 16 -> 5), so DB-path hellos and renewals keep the rest. export const REREGISTER_POOL_SHARE = 1 / 3 @@ -334,6 +338,9 @@ export class HostSessionRegistry { // Seat key -> the demoted control socket. Its own set, not drainSentHosts: it ends with // that socket (close, rebind or a new join), so a later booking here is a fresh seat. private readonly demotedSeats = new Map() + private readonly seatDrift = new SeatDriftConfirmation() + // The highest generation each host has had on this cell, kept after its session leaves the map. + private readonly generationFloors = new Map() private reregistration: { queue: ReregistrationEntry[] // Sessions already queued, given up on, or done in this pass: the rescan skips them. @@ -776,6 +783,13 @@ export class HostSessionRegistry { session.assignmentEpoch !== request.epoch || seat?.joinedAt !== request.joinedAt ) { + // The director named exactly the seat this feed lists: if no live session backs it, the + // feed is what is wrong, and every director mirrors it, so correct the feed now. + const unbacked = + seat?.epoch === request.epoch && seat.joinedAt === request.joinedAt + ? unbackedSeatReason(seat, session) + : null + if (seat && unbacked) this.repairSeatDrift(seat, session, unbacked, 'demote') return 'not-seated' } // A drain already told this host to go: its own grace stands, never cut to the demotion's. @@ -1682,7 +1696,11 @@ export class HostSessionRegistry { (existing.state === 'orphaned' || existing.state === 'active') && sameHostIdentity(existing.identity, identity) ) - const generation = rebind ? existing!.generation : (existing?.generation ?? 0) + 1 + // Above every generation this host has had here, not only the one in the map: a session + // dropped from the map while its socket is still closing (drain, regional deadline, idle + // rehome) leaves later, and a successor reusing its generation had its seat removed by that leave. + const floor = this.generationFloors.get(key) ?? 0 + const generation = rebind ? existing!.generation : Math.max(existing?.generation ?? 0, floor) + 1 const ephemeral = nacl.box.keyPair() const challengeNonce = randomBytes(nacl.box.nonceLength) const challengeSecret = randomBytes(32) @@ -1912,6 +1930,24 @@ export class HostSessionRegistry { } return } + // The awaits above can outlive what this hello was proven against: an orphan grace expiring + // or a drain removes `existing` from the map, and another activation may take this generation. + // Rebinding a removed session revived it outside the map: connected, still in the feed, and + // unreachable by phones or demotion for as long as the host stayed up. Redial instead. + const key = this.key(identity.sub, identity.relayHostId) + if ( + (this.sessions.get(key) ?? null) !== existing || + (!rebind && generation <= (this.generationFloors.get(key) ?? 0)) + ) { + if (controlActivityId) { + this.releaseActivityBestEffort( + { userId: identity.sub, relayHostId: identity.relayHostId }, + controlActivityId + ) + } + this.refuseProvenControl(socket, refused, RELAY_CLOSE_CODE.PEER_DROPPED, 'control activation superseded') + return + } if (existing) this.observer.recordReconnect() if (rebind && existing) { const previousSocket = existing.socket @@ -2005,6 +2041,7 @@ export class HostSessionRegistry { // A host that proved itself again is not signed out, whatever it said last. this.hostCloseReasons.forget(sessionKey) this.sessions.set(sessionKey, session) + this.recordGenerationFloor(sessionKey, generation) this.demotedSeats.delete(sessionKey) this.wireActiveControl(session) this.sendHelloAck(session) @@ -2028,9 +2065,87 @@ export class HostSessionRegistry { } seatFeed(sinceSeq: number | null): CellSeatFeedPage { + this.sweepSeatDrift() return { ...this.seatLog.read(sinceSeq), seats: this.seatLog.seatCount() } } + // Every director mirrors this feed, so it must list exactly the live sessions. Swept on a + // director's poll, since only directors read it; drift seen twice in a row is repaired. + private sweepSeatDrift(): void { + if (!this.seatDrift.due(this.now())) return + const found = new Map void>() + for (const seat of this.seatLog.seatList()) { + const session = this.get(seat) + const reason = unbackedSeatReason(seat, session) + if (!reason) continue + const id = JSON.stringify(['seat', seat.userId, seat.relayHostId, seat.epoch, seat.generation, seat.joinedAt]) + found.set(id, () => this.repairSeatDrift(seat, session, reason, 'sweep')) + } + for (const session of this.sessions.values()) { + if (!liveSeatSession(session) || this.seatLog.seatOf(session.identity.sub, session.relayHostId)) continue + const id = JSON.stringify(['session', session.identity.sub, session.relayHostId, session.generation]) + found.set(id, () => { + this.appendSeatChange(session, 'join') + this.logSeatDrift(session.relayHostId, { kind: 'unlisted-session', via: 'sweep', sessionEpoch: session.assignmentEpoch }) + }) + } + for (const id of this.seatDrift.confirm([...found.keys()])) found.get(id)!() + } + + // Leaves the stale seat as moved (no rejoin from memory at its epoch), then lists the live + // session, if any, as it really is. The leave carries the stale seat's own generation, so the + // generation rule cannot drop it. + private repairSeatDrift( + seat: CellSeat, + session: HostSession | null, + reason: UnbackedSeatReason, + via: 'sweep' | 'demote' + ): void { + this.seatLog.append({ + kind: 'leave', + userId: seat.userId, + relayHostId: seat.relayHostId, + epoch: seat.epoch, + generation: seat.generation, + closeCode: RELAY_CLOSE_CODE.WRONG_CELL, + at: this.now() + }) + const relisted = session !== null && liveSeatSession(session) + if (relisted) this.appendSeatChange(session, 'join') + this.logSeatDrift(seat.relayHostId, { + kind: 'unbacked-seat', + via, + reason, + seatEpoch: seat.epoch, + seatGeneration: seat.generation, + seatAgeMs: this.now() - seat.joinedAt, + ...(session + ? { sessionEpoch: session.assignmentEpoch, sessionGeneration: session.generation, sessionState: session.state } + : {}), + relisted + }) + } + + private recordGenerationFloor(key: string, generation: number): void { + this.generationFloors.delete(key) + this.generationFloors.set(key, generation) + if (this.generationFloors.size > GENERATION_FLOOR_MAX_HOSTS) { + const oldest = this.generationFloors.keys().next().value + if (oldest !== undefined) this.generationFloors.delete(oldest) + } + } + + private logSeatDrift(relayHostId: string, detail: Record): void { + console.warn( + JSON.stringify({ + event: 'orca_relay_cell_seat_log_drift', + ...this.logIdentity(), + relayHostIdDigest: relayHostLogDigest(relayHostId), + ...detail + }) + ) + } + private appendSeatChange( session: HostSession, kind: CellSeatChange['kind'], @@ -2719,3 +2834,8 @@ function controlErrorCode(type: string, error: unknown): string { function databaseCause(error: unknown): PhoneDatabaseCause { return isRelayDatabaseTransientError(error) ? 'database_transient' : 'error' } + +// A session the feed must list: it holds its socket and is seated, active or draining. +function liveSeatSession(session: HostSession): boolean { + return session.socket !== null && (session.state === 'active' || session.state === 'drain-only') +} diff --git a/cloud/apps/relay/src/relay-server.ts b/cloud/apps/relay/src/relay-server.ts index 7175ca08dfc..dc9f49fa60c 100644 --- a/cloud/apps/relay/src/relay-server.ts +++ b/cloud/apps/relay/src/relay-server.ts @@ -275,7 +275,7 @@ export function createRelayServer( })) ), demote: (seat, row) => { - reservePlacement?.demoteSeat(seat) + reservePlacement?.demoteSeat(seat, 'reconcile') // The re-assign mints above the row, so the mirror can take it. reservePlacement?.raiseEpochFloor(seat, row.epoch) console.warn( diff --git a/cloud/apps/relay/src/relay-websocket-close.ts b/cloud/apps/relay/src/relay-websocket-close.ts index 4b0517c3e53..74554253885 100644 --- a/cloud/apps/relay/src/relay-websocket-close.ts +++ b/cloud/apps/relay/src/relay-websocket-close.ts @@ -1,6 +1,6 @@ import type WebSocket from 'ws' -const RELAY_WEBSOCKET_FORCE_CLOSE_MS = 1_000 +export const RELAY_WEBSOCKET_FORCE_CLOSE_MS = 1_000 const forceCloseTimers = new WeakMap>() // Delivering the rejection is best effort, and deliberately so. The close frame carrying the diff --git a/cloud/apps/relay/src/reserve-assignment.test.ts b/cloud/apps/relay/src/reserve-assignment.test.ts index fe424f412db..ec1bb9bcbc3 100644 --- a/cloud/apps/relay/src/reserve-assignment.test.ts +++ b/cloud/apps/relay/src/reserve-assignment.test.ts @@ -54,6 +54,7 @@ function setup( mode?: 'off' | 'dry-run' | 'on' startedAt?: number log?: (line: string) => void + demote?: (cellId: string) => string | null } = {} ) { const now = { value: 100_000 } @@ -72,7 +73,7 @@ function setup( }), demote: vi.fn(async (target: { cellId: string }, request: { epoch: number; joinedAt: number }) => { demoted.push({ cellId: target.cellId, epoch: request.epoch, joinedAt: request.joinedAt }) - return true + return options.demote ? options.demote(target.cellId) : 'demoted' }) } satisfies Pick // Each director's estimate starts empty; a second of refill gives it its share. @@ -503,4 +504,42 @@ describe('reserve assignment on a director', () => { { outcomes: { ok: 2 }, cells: { c1: { outcomes: { ok: 2 }, fullWithRoom: 0, minFullFreeShare: null } } } ]) }) + + describe('when a cell answers a demotion', () => { + const settle = () => new Promise((resolve) => setTimeout(resolve, 0)) + const outcomes = (lines: string[]) => + lines + .map((line) => JSON.parse(line)) + .filter((entry) => entry.event === 'orca_relay_reserve_demote_outcome') + + // The map follows the cell's feed: a refusing cell repairs that feed itself. + it('logs the answer by host digest and leaves the seat to the feed', async () => { + const lines: string[] = [] + const { assignment, directory } = setup({ + feeds: [feed('c1', { full: [seated(10, 1_000)] }), feed('c2')], + demote: () => 'not-seated', + log: (line) => lines.push(line) + }) + assignment.demoteSeat({ ...HOST, cellId: 'c1', epoch: 10, joinedAt: 1_000 }, 'reconcile') + await settle() + expect(outcomes(lines)).toMatchObject([ + { via: 'reconcile', cellId: 'c1', epoch: 10, outcome: 'not-seated' } + ]) + expect(outcomes(lines)[0].relayHostIdDigest).toMatch(/^[0-9a-f]{12}$/) + expect(lines.join('\n')).not.toContain(HOST.relayHostId) + expect(directory.seatsOnCell('c1')).toHaveLength(1) + }) + + it('records a cell it could not reach as unreachable', async () => { + const lines: string[] = [] + const { assignment } = setup({ + feeds: [feed('c1', { full: [seated(10, 1_000)] }), feed('c2')], + demote: () => null, + log: (line) => lines.push(line) + }) + assignment.demoteSeat({ ...HOST, cellId: 'c1', epoch: 10, joinedAt: 1_000 }, 'reconcile') + await settle() + expect(outcomes(lines)).toMatchObject([{ outcome: 'unreachable' }]) + }) + }) }) diff --git a/cloud/apps/relay/src/reserve-assignment.ts b/cloud/apps/relay/src/reserve-assignment.ts index eb5d0bc42ae..0ccb389cd7e 100644 --- a/cloud/apps/relay/src/reserve-assignment.ts +++ b/cloud/apps/relay/src/reserve-assignment.ts @@ -422,19 +422,36 @@ export class ReserveAssignment { } // 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 { + demoteSeat(seat: DemotedSeat, via: DemoteVia): boolean { const cell = this.input.cells().find((entry) => entry.cellId === seat.cellId) if (!cell) return false - void this.input.client.demote(cell, { - v: 1, - userId: seat.userId, - relayHostId: seat.relayHostId, - epoch: seat.epoch, - joinedAt: seat.joinedAt - }) + void this.input.client + .demote(cell, { + v: 1, + userId: seat.userId, + relayHostId: seat.relayHostId, + epoch: seat.epoch, + joinedAt: seat.joinedAt + }) + .then((outcome) => this.settleDemote(seat, via, outcome)) + .catch(() => console.warn('[orca-relay] reserve demote outcome not recorded')) return true } + // The cell repairs its own feed when it refuses a seat it still lists; this only records the answer. + private settleDemote(seat: DemotedSeat, via: DemoteVia, outcome: string | null): void { + this.log( + JSON.stringify({ + event: 'orca_relay_reserve_demote_outcome', + via, + cellId: seat.cellId, + epoch: seat.epoch, + relayHostIdDigest: relayHostLogDigest(seat.relayHostId), + outcome: outcome ?? 'unreachable' + }) + ) + } + // 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 { @@ -456,9 +473,9 @@ export class ReserveAssignment { toCellId: string, via: 'placement' | 'feed' ): void { - if (!this.demoteSeat(seat)) return + if (!this.demoteSeat(seat, via)) return this.summary.supersedes += 1 - this.input.log?.( + this.log( JSON.stringify({ event: 'orca_relay_reserve_supersede', relayHostIdDigest: relayHostLogDigest(seat.relayHostId), @@ -479,7 +496,7 @@ export class ReserveAssignment { latenciesMs.length === 0 ? null : Math.round(latenciesMs[Math.min(latenciesMs.length - 1, Math.floor(share * latenciesMs.length))]!) - ;(this.input.log ?? ((line: string) => console.log(line)))( + this.log( JSON.stringify({ event: 'orca_relay_reserve_placement_summary', mode: this.input.mode, @@ -489,8 +506,15 @@ export class ReserveAssignment { ) this.summary = emptySummary() } + + private log(line: string): void { + ;(this.input.log ?? ((entry: string) => console.log(entry)))(line) + } } +type DemotedSeat = { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number } +type DemoteVia = 'reconcile' | 'placement' | 'feed' + type PlacementSummary = { sticky: number stickyReserves: number