diff --git a/cloud/apps/relay/src/cell-phone-path.blackbox.test.ts b/cloud/apps/relay/src/cell-phone-path.blackbox.test.ts index 2cc3b27c294..8e39081babd 100644 --- a/cloud/apps/relay/src/cell-phone-path.blackbox.test.ts +++ b/cloud/apps/relay/src/cell-phone-path.blackbox.test.ts @@ -39,6 +39,29 @@ const DATABASE_PATH_ACCEPT = [ 'INSERT INTO relay_connection_bases (basis_conn_id, user_id, relay_host_id, relay_device_id, owning_control_generation, credential_kind, invite_token_hash, accepted_credential_version, accepted_as, deadline, active, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)' ] +const CONFIRM_DEVICE_READ = + 'SELECT * FROM relay_devices WHERE user_id = ? AND relay_host_id = ? AND relay_device_id = ?' + +// The flag-off confirm, as main issued it at 9ea54343710. +const DATABASE_PATH_CONFIRM = [ + 'BEGIN', + 'SELECT basis_conn_id, result_json FROM relay_confirm_results WHERE user_id = ? AND relay_host_id = ? AND req_id = ?', + 'SELECT * FROM relay_connection_bases WHERE basis_conn_id = ?', + 'UPDATE relay_devices SET updated_at = updated_at WHERE user_id = ? AND relay_host_id = ? AND relay_device_id = ?', + 'SELECT * FROM relay_devices WHERE user_id = ? AND relay_host_id = ? AND relay_device_id = ?', + 'UPDATE relay_devices SET current_expires_at = ?, updated_at = ? WHERE user_id = ? AND relay_host_id = ? AND relay_device_id = ?', + 'INSERT INTO relay_confirm_results (user_id, relay_host_id, req_id, basis_conn_id, tuple_json, result_json, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?)', + 'INSERT INTO relay_audit_events (id, at, type, user_id, relay_host_id, relay_device_id, detail_json) VALUES (?, ?, ?, ?, ?, ?, ?)', + 'COMMIT', + 'BEGIN', + 'SELECT * FROM relay_assignments WHERE user_id = ? AND relay_host_id = ? [locked]', + 'SELECT * FROM relay_assignment_activity_leases WHERE user_id = ? AND relay_host_id = ? ORDER BY activity_id ASC [locked]', + 'DELETE FROM relay_assignment_activity_leases WHERE user_id = ? AND relay_host_id = ? AND activity_id = ?', + 'UPDATE relay_assignments SET pending_confirmations = CASE WHEN pending_confirmations + ? < 0 THEN 0 ELSE pending_confirmations + ? END, lease_expires_at = CASE WHEN lease_expires_at > ? THEN lease_expires_at ELSE ? END, last_activity_at = ? WHERE user_id = ? AND relay_host_id = ?', + 'UPDATE relay_cells SET reserved_requests = CASE WHEN reserved_requests + ? < 0 THEN 0 ELSE reserved_requests + ? END, updated_at = ? WHERE cell_id = ? AND (? <= 0 OR reserved_requests + ? <= capacity_requests) RETURNING cell_id', + 'COMMIT' +] + describe('the memory phone path', () => { const cleanup: Array<() => Promise | void> = [] @@ -60,7 +83,8 @@ describe('the memory phone path', () => { // The admit-effective writer ticks on its own; every other statement is the phone's. const phoneStatements = () => database.statements.filter((sql) => !sql.includes('relay_cell_admit_effective')) - return { ...reserve, database, phoneStatements } + // The test's own reads and writes go around the recorder. + return { ...reserve, database, raw: inner, phoneStatements } } async function seatedPhoneHost(phonePath: 'db' | 'memory') { @@ -72,6 +96,33 @@ describe('the memory phone path', () => { return { reserve, seated } } + // A phone resumed and spliced to its desktop, waiting to confirm. + async function splicedPhone(phonePath: 'db' | 'memory', device: { expiresAt?: number } = {}) { + const reserve = await cell() + reserve.setAdmitMode('reserve', { phonePath }) + const seated = await reserve.bookedHost(3) + await reserve.installDevice(seated.identity.hostId, TOKEN, device) + const connOpen = nextFrame(seated.socket, 'conn-open') + const phone = await reserve.dialPhone(seated.identity.hostId, TOKEN) + const basisConnId = String((await connOpen).connId) + await reserve.attachData(await connOpen, seated.ack.generation) + expect(await phone.hello).toMatchObject({ ok: true }) + reserve.database.statements.length = 0 + // The desktop's confirm, sent in reply to the phone over the end-to-end channel. + const confirm = async (reqId: string, basis = basisConnId) => { + const answer = Promise.race([ + nextFrame(seated.socket, 'device-resume-confirmed'), + nextFrame(seated.socket, 'control-error') + ]) + seated.socket.send(JSON.stringify({ type: 'device-resume-confirm', v: 1, reqId, basisConnId: basis })) + return await answer + } + return { reserve, seated, phone, basisConnId, confirm } + } + + const deviceRow = async (reserve: Awaited>) => + (await reserve.raw.query(`SELECT current_expires_at FROM relay_devices`))[0] + const warnings = (event: string) => vi .mocked(console.warn) @@ -102,7 +153,7 @@ describe('the memory phone path', () => { expect(reserve.phoneStatements()).toEqual([DATABASE_PATH_ACCEPT[0]]) await reserve.attachData(await connOpen, seated.ack.generation) expect(await phone.hello).toMatchObject({ type: 'relay-hello', ok: true, credentialKind: 'resume' }) - expect(reserve.phoneStatements()).toEqual([DATABASE_PATH_ACCEPT[0], DATABASE_PATH_ACCEPT.at(-1)]) + expect(reserve.phoneStatements()).toEqual([DATABASE_PATH_ACCEPT[0]]) expect(await reserve.activityLeases()).toEqual([]) }) @@ -204,4 +255,121 @@ describe('the memory phone path', () => { expect(await connOpen).toMatchObject({ kind: 'resume' }) expect((await reserve.activityLeases()).map((lease) => lease.activity_kind)).toEqual(['confirmation']) }) + + it('confirms from the memory basis with one fresh read, renewing a device due for it', async () => { + const { reserve, confirm } = await splicedPhone('memory') + const before = Number((await deviceRow(reserve))?.current_expires_at) + const answer = await confirm('confirm-1') + expect(answer).toMatchObject({ type: 'device-resume-confirmed', reqId: 'confirm-1', renewed: true, acceptedAs: 'current' }) + const after = Number((await deviceRow(reserve))?.current_expires_at) + expect(after).toBeGreaterThan(before) + expect(answer.resumeExpiresAt).toBe(after) + expect(reserve.phoneStatements()).toEqual([ + CONFIRM_DEVICE_READ, + expect.stringMatching(/^UPDATE relay_devices SET current_expires_at = \?.*RETURNING current_expires_at$/), + expect.stringMatching(/^INSERT INTO relay_audit_events/) + ]) + expect(await reserve.activityLeases()).toEqual([]) + // A retry of the same request answers from memory; another basis cannot reuse its reqId. + reserve.database.statements.length = 0 + expect(await confirm('confirm-1')).toEqual(answer) + expect(reserve.phoneStatements()).toEqual([]) + expect(await confirm('confirm-1', 'another-basis')).toMatchObject({ + type: 'control-error', + code: 'confirmation_tuple_mismatch' + }) + }) + + it('answers renewed:true with the stored expiry when renewal is not due, or its write fails', async () => { + const fresh = Date.now() + 29.5 * 24 * 60 * 60 * 1000 + const notDue = await splicedPhone('memory', { expiresAt: fresh }) + expect(await notDue.confirm('confirm-1')).toMatchObject({ renewed: true, resumeExpiresAt: fresh }) + expect(notDue.reserve.phoneStatements()).toEqual([CONFIRM_DEVICE_READ, expect.stringMatching(/^INSERT INTO relay_audit_events/)]) + + const due = Date.now() + 20 * 24 * 60 * 60 * 1000 + const failing = await splicedPhone('memory', { expiresAt: due }) + failing.reserve.database.failing = (sql) => sql.startsWith('UPDATE relay_devices') + expect(await failing.confirm('confirm-1')).toMatchObject({ renewed: true, resumeExpiresAt: due }) + expect(Number((await deviceRow(failing.reserve))?.current_expires_at)).toBe(due) + // Never silent: a renewal left failing ages phones toward re-pairing. + expect(warnings('orca_relay_resume_renewal_failed')).toEqual([ + { event: 'orca_relay_resume_renewal_failed', cause: 'error', cellId: failing.reserve.config.cellId } + ]) + }) + + it('confirms from memory after the switch turned off between accept and confirm, taking no lease', async () => { + const { reserve, confirm } = await splicedPhone('memory') + reserve.setAdmitMode('reserve', { phonePath: 'db' }) + expect(await confirm('confirm-1')).toMatchObject({ type: 'device-resume-confirmed', renewed: true }) + expect(reserve.phoneStatements()[0]).toBe(CONFIRM_DEVICE_READ) + expect(await reserve.activityLeases()).toEqual([]) + }) + + it('replays a stored answer after its splice closed, and drops the basis itself', async () => { + const { reserve, phone, confirm } = await splicedPhone('memory') + const answer = await confirm('confirm-1') + phone.socket.close() + await phone.closed + await new Promise((resolve) => setTimeout(resolve, 50)) + reserve.database.statements.length = 0 + expect(await confirm('confirm-1')).toEqual(answer) + expect(await confirm('confirm-1', 'another-basis')).toMatchObject({ code: 'confirmation_tuple_mismatch' }) + expect(reserve.phoneStatements()).toEqual([]) + // A new request on the closed basis finds none in memory, so it goes to the database. + expect(await confirm('confirm-2')).toMatchObject({ code: 'confirmation_not_active' }) + expect(reserve.phoneStatements()[0]).toBe('BEGIN') + }) + + it('answers a grace-version confirm renewed:false with its expiry unchanged, as today', async () => { + const { reserve, confirm } = await splicedPhone('memory') + const stored = Number((await deviceRow(reserve))?.current_expires_at) + // An install rotated the credential after the accept read, so the accepted version is grace. + await reserve.raw.query( + `UPDATE relay_devices SET grace_hash = current_hash, grace_version = 1, grace_expires_at = ?, + current_hash = 'rotated', current_version = 2`, + [Date.now() + 60_000] + ) + expect(await confirm('confirm-1')).toMatchObject({ + type: 'device-resume-confirmed', + currentVersion: 2, + acceptedAs: 'current', + renewed: false, + resumeExpiresAt: stored + }) + expect(reserve.phoneStatements()).toEqual([CONFIRM_DEVICE_READ, expect.stringMatching(/^INSERT INTO relay_audit_events/)]) + }) + + it('answers every device change since the accept read as the database confirm does', async () => { + const changes: Record = { + revoked: ['UPDATE relay_devices SET revoked_at = ?', [Date.now() - 1]], + expired: ['UPDATE relay_devices SET current_expires_at = ?', [Date.now() - 1]], + retired: ["UPDATE relay_devices SET current_hash = 'rotated-twice', current_version = 3, grace_version = 2", []], + unchanged: ['UPDATE relay_devices SET updated_at = updated_at', []] + } + const outcome = async (phonePath: 'db' | 'memory', [sql, params]: [string, unknown[]]) => { + const { reserve, confirm } = await splicedPhone(phonePath) + await reserve.raw.query(sql, params) + const answer = await confirm('confirm-1') + return answer.type === 'control-error' ? answer.code : `${answer.type}:${answer.acceptedAs}:${answer.renewed}` + } + for (const [name, change] of Object.entries(changes)) { + expect(await outcome('memory', change), name).toBe(await outcome('db', change)) + } + expect(await outcome('memory', changes.revoked!)).toBe('reject-revoked') + }) + + it('sends a confirm with no memory basis to the database, which holds none either', async () => { + const { reserve, phone, confirm } = await splicedPhone('memory') + phone.socket.close() + await phone.closed + await new Promise((resolve) => setTimeout(resolve, 50)) + expect(await confirm('confirm-1')).toMatchObject({ type: 'control-error', code: 'confirmation_not_active' }) + expect(reserve.phoneStatements()[0]).toBe('BEGIN') + }) + + it('keeps the database confirm statement for statement while phonePath is db', async () => { + const { reserve, confirm } = await splicedPhone('db') + expect(await confirm('confirm-1')).toMatchObject({ type: 'device-resume-confirmed', renewed: true }) + expect(reserve.phoneStatements()).toEqual(DATABASE_PATH_CONFIRM) + }) }) diff --git a/cloud/apps/relay/src/credential-store.ts b/cloud/apps/relay/src/credential-store.ts index 476b2652e8f..b72daadd0b9 100644 --- a/cloud/apps/relay/src/credential-store.ts +++ b/cloud/apps/relay/src/credential-store.ts @@ -95,6 +95,21 @@ function string(row: SqlRow, field: string): string { return value } +// Memory confirms renew a device at most this often (design decision 3: at most once a day of use). +export const RESUME_RENEWAL_INTERVAL_MS = 24 * 60 * 60 * 1000 + +function tokenFamily(device: SqlRow) { + return { + currentVersion: number(device, 'current_version'), + currentHash: string(device, 'current_hash'), + currentExpiresAt: number(device, 'current_expires_at'), + graceVersion: optionalNumber(device, 'grace_version'), + graceHash: typeof device.grace_hash === 'string' ? device.grace_hash : undefined, + graceExpiresAt: optionalNumber(device, 'grace_expires_at'), + revokedAt: optionalNumber(device, 'revoked_at') + } +} + // A device row's answer to a resume token: null when revoked, or neither hash is still valid. function resumeReservation( device: SqlRow, @@ -517,19 +532,7 @@ export class RelayCredentialStore { const device = devices[0] if (!device) throw new RelayStoreError('credential_not_found') const acceptedVersion = number(basis, 'accepted_credential_version') - const decision = decideResumeCommit( - { - currentVersion: number(device, 'current_version'), - currentHash: string(device, 'current_hash'), - currentExpiresAt: number(device, 'current_expires_at'), - graceVersion: optionalNumber(device, 'grace_version'), - graceHash: typeof device.grace_hash === 'string' ? device.grace_hash : undefined, - graceExpiresAt: optionalNumber(device, 'grace_expires_at'), - revokedAt: optionalNumber(device, 'revoked_at') - }, - acceptedVersion, - now - ) + const decision = decideResumeCommit(tokenFamily(device), acceptedVersion, now) if (decision.startsWith('reject-')) throw new RelayStoreError(decision) const renewed = decision === 'renew-current' const resumeExpiresAt = renewed @@ -581,6 +584,84 @@ export class RelayCredentialStore { }) } + // Step 6's confirm for a basis the cell holds in memory: one fresh device read, and a renewal + // write at most once a day per device. A current-version confirm always answers renewed:true + // with the expiry the database now holds, so a phone keeps scheduling its renewals. + async confirmMemoryResume( + input: RelayIdentity & { + reqId: string + basisConnId: string + relayDeviceId: string + acceptedCredentialVersion: number + acceptedAs: 'current' | 'grace' + }, + options: { onRenewalFailure?: (error: unknown) => void } = {} + ): Promise { + const now = this.now() + const devices = await this.database.query( + `SELECT * FROM relay_devices + WHERE user_id = ? AND relay_host_id = ? AND relay_device_id = ?`, + [input.userId, input.relayHostId, input.relayDeviceId] + ) + const device = devices[0] + if (!device) throw new RelayStoreError('credential_not_found') + const decision = decideResumeCommit(tokenFamily(device), input.acceptedCredentialVersion, now) + if (decision.startsWith('reject-')) throw new RelayStoreError(decision) + const renewed = decision === 'renew-current' + let resumeExpiresAt = number(device, 'current_expires_at') + const renewalDue = resumeExpiresAt < now + RELAY_PROTOCOL_LIMITS.resumeTtlMs - RESUME_RENEWAL_INTERVAL_MS + if (renewed && renewalDue) { + // A failing renewal still answers the expiry the database holds, but must stay visible: + // left failing, phones age toward re-pairing. + resumeExpiresAt = await this.renewResume(input, device, now).catch((error: unknown) => { + options.onRenewalFailure?.(error) + return resumeExpiresAt + }) + } + // Off the answer's critical outcome: an audit row the database refuses is not a failed confirm. + await this.auditWith(this.database, { + type: 'resume-confirmed', + userId: input.userId, + relayHostId: input.relayHostId, + relayDeviceId: input.relayDeviceId, + detail: { reqId: input.reqId, basisConnId: input.basisConnId, renewed } + }).catch(() => console.warn('[orca-relay] resume-confirmed audit row not written')) + return { + v: 1, + reqId: input.reqId, + currentVersion: number(device, 'current_version'), + acceptedAs: input.acceptedAs, + renewed, + resumeExpiresAt, + ...(optionalNumber(device, 'grace_expires_at') === undefined + ? {} + : { graceExpiresAt: optionalNumber(device, 'grace_expires_at') }) + } + } + + // Guarded by the row just read, so a revoke or install that committed since leaves it alone; + // the answer then keeps the expiry that was read, which the database still holds. + private async renewResume( + identity: RelayIdentity & { relayDeviceId: string }, + device: SqlRow, + now: number + ): Promise { + const renewedTo = now + RELAY_PROTOCOL_LIMITS.resumeTtlMs + const rows = await this.database.query( + `UPDATE relay_devices SET current_expires_at = ?, updated_at = ? + WHERE user_id = ? AND relay_host_id = ? AND relay_device_id = ? + AND current_version = ? AND current_hash = ? AND revoked_at IS NULL + AND current_expires_at < ? + RETURNING current_expires_at`, + [ + renewedTo, now, identity.userId, identity.relayHostId, identity.relayDeviceId, + number(device, 'current_version'), string(device, 'current_hash'), + now + RELAY_PROTOCOL_LIMITS.resumeTtlMs - RESUME_RENEWAL_INTERVAL_MS + ] + ) + return rows[0] ? number(rows[0], 'current_expires_at') : number(device, 'current_expires_at') + } + async deactivateBasis(basisConnId: string): Promise { await this.database.query(`UPDATE relay_connection_bases SET active = ? WHERE basis_conn_id = ?`, [ 0, diff --git a/cloud/apps/relay/src/host-session-registry.ts b/cloud/apps/relay/src/host-session-registry.ts index 46fbec84041..d33c6e36578 100644 --- a/cloud/apps/relay/src/host-session-registry.ts +++ b/cloud/apps/relay/src/host-session-registry.ts @@ -39,6 +39,7 @@ import { printableCloseReason } from './control-close-reason-text.js' import { ControlRenewalBatch } from './control-renewal-batch.js' import { RelayCredentialStore, type CredentialReservation } from './credential-store.js' import { HostCloseReasonMemory } from './host-close-reason-memory.js' +import { MemoryResumeConfirmations } from './memory-resume-confirmations.js' import { PhoneAcceptLog } from './phone-accept-log.js' import { postProofCloseCause, @@ -304,6 +305,7 @@ export class HostSessionRegistry { private readonly seatLog: CellSeatLog private readonly postProofCloses: PostProofCloseLog private readonly phoneAccepts: PhoneAcceptLog + private readonly memoryConfirmations: MemoryResumeConfirmations // 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() @@ -449,6 +451,9 @@ export class HostSessionRegistry { this.seatLog = new CellSeatLog(undefined, now) this.postProofCloses = new PostProofCloseLog(observer) this.phoneAccepts = new PhoneAcceptLog(config.cellId) + this.memoryConfirmations = new MemoryResumeConfirmations(store, now, (error) => + this.phoneAccepts.renewalFailed(isRelayDatabaseTransientError(error) ? 'database_transient' : 'error') + ) } // Reserve mode is only ever a cell's, and only while its applied switch says so. @@ -1151,12 +1156,16 @@ export class HostSessionRegistry { cellId: this.config.cellId }) } - await this.store.recordConnectionBasis({ - ...pending.reservation, - basisConnId: connId, - owningControlGeneration: session.generation, - deadline: basisDeadline - }) + if (pending.phonePath === 'memory') { + this.openMemoryBasis(pending, connId, session.generation, basisDeadline) + } else { + await this.store.recordConnectionBasis({ + ...pending.reservation, + basisConnId: connId, + owningControlGeneration: session.generation, + deadline: basisDeadline + }) + } } catch { session.activeConnIds.delete(connId) await this.store.failReservation(pending.reservation) @@ -1179,7 +1188,7 @@ export class HostSessionRegistry { ) { session.activeConnIds.delete(connId) pending.capacityReservation?.release() - this.deactivateBasisBestEffort(connId) + this.closeBasis(pending, connId) this.failReservationBestEffort(pending.reservation) if (spliceActivityId) this.releaseActivityBestEffort(identity, spliceActivityId) if (pending.credentialActivityId) { @@ -1196,7 +1205,7 @@ export class HostSessionRegistry { onClose: () => { session.activeConnIds.delete(connId) session.activeSplices.delete(connId) - this.deactivateBasisBestEffort(connId) + this.closeBasis(pending, connId) if (spliceActivityId) this.releaseActivityBestEffort(identity, spliceActivityId) if (pending.credentialActivityId) { this.releaseActivityBestEffort(identity, pending.credentialActivityId) @@ -2478,11 +2487,12 @@ export class HostSessionRegistry { } if (type === 'device-resume-confirm') { const request = DeviceResumeConfirmSchema.parse(payload(raw, type)) - const result = await this.store.confirmResume({ - ...identity, - ...request, - owningControlGeneration: session.generation - }) + const confirm = { ...identity, ...request, owningControlGeneration: session.generation } + if (this.memoryConfirmations.owns(confirm)) { + send(session.socket, 'device-resume-confirmed', await this.memoryConfirmations.confirm(confirm)) + return + } + const result = await this.store.confirmResume(confirm) await this.assignments.releaseActivity(identity, `confirmation:${request.basisConnId}`) send(session.socket, 'device-resume-confirmed', result) return @@ -2549,6 +2559,34 @@ export class HostSessionRegistry { }) } + // A memory resume's basis is the cell's own record; every other basis is a Postgres row. + private openMemoryBasis( + pending: PendingConnection, + connId: string, + owningControlGeneration: number, + deadline: number + ): void { + const { reservation } = pending + if (reservation.acceptedCredentialVersion === undefined || reservation.acceptedAs === undefined) { + throw new Error('memory basis needs an accepted resume credential') + } + this.memoryConfirmations.open({ + basisConnId: connId, + userId: reservation.userId, + relayHostId: reservation.relayHostId, + relayDeviceId: reservation.relayDeviceId, + acceptedCredentialVersion: reservation.acceptedCredentialVersion, + acceptedAs: reservation.acceptedAs, + owningControlGeneration, + deadline + }) + } + + private closeBasis(pending: PendingConnection, connId: string): void { + if (pending.phonePath === 'memory') this.memoryConfirmations.close(connId) + else this.deactivateBasisBestEffort(connId) + } + private deactivateBasisBestEffort(connId: string): void { void this.store.deactivateBasis(connId).catch(() => { console.warn('[orca-relay] basis deactivation deferred to credential cleanup') diff --git a/cloud/apps/relay/src/memory-resume-confirmations.ts b/cloud/apps/relay/src/memory-resume-confirmations.ts new file mode 100644 index 00000000000..c2de7a7be7a --- /dev/null +++ b/cloud/apps/relay/src/memory-resume-confirmations.ts @@ -0,0 +1,98 @@ +import type { DeviceResumeConfirmed } from '@orca-cloud/relay-contract' +import { RelayStoreError, type RelayCredentialStore, type RelayIdentity } from './credential-store.js' + +// Step 6: what a memory cell keeps instead of relay_connection_bases and relay_confirm_results. +// Both live and die with this process; a restart takes the splices with it. + +export type MemoryConnectionBasis = RelayIdentity & { + basisConnId: string + relayDeviceId: string + acceptedCredentialVersion: number + acceptedAs: 'current' | 'grace' + owningControlGeneration: number + deadline: number +} + +export type ResumeConfirmRequest = RelayIdentity & { + reqId: string + basisConnId: string + owningControlGeneration: number +} + +// A retried confirm of the same request answers from here, as relay_confirm_results does. +export const MEMORY_CONFIRM_RESULT_RETENTION_MS = 10 * 60 * 1000 + +type StoredResult = { basisConnId: string; result: DeviceResumeConfirmed; at: number } + +function resultKey(request: RelayIdentity & { reqId: string }): string { + return `${request.userId}\u0000${request.relayHostId}\u0000${request.reqId}` +} + +export class MemoryResumeConfirmations { + private readonly bases = new Map() + // Insertion order is age order, so expiry walks from the front. + private readonly results = new Map() + + constructor( + private readonly store: RelayCredentialStore, + private readonly now: () => number = Date.now, + private readonly onRenewalFailure?: (error: unknown) => void + ) {} + + open(basis: MemoryConnectionBasis): void { + this.bases.set(basis.basisConnId, basis) + } + + close(basisConnId: string): void { + this.bases.delete(basisConnId) + } + + // Routed by where the basis lives, not by the switch, so a flip between accept and confirm + // neither fails the confirm nor sends it to a database that holds no basis for it. + owns(request: ResumeConfirmRequest): boolean { + this.expireResults() + return this.bases.has(request.basisConnId) || this.results.has(resultKey(request)) + } + + async confirm(request: ResumeConfirmRequest): Promise { + this.expireResults() + const key = resultKey(request) + const prior = this.results.get(key) + if (prior) { + if (prior.basisConnId !== request.basisConnId) throw new RelayStoreError('confirmation_tuple_mismatch') + return prior.result + } + const basis = this.bases.get(request.basisConnId) + if ( + !basis || + basis.userId !== request.userId || + basis.relayHostId !== request.relayHostId || + basis.owningControlGeneration !== request.owningControlGeneration || + basis.deadline < this.now() + ) { + throw new RelayStoreError('confirmation_not_active') + } + const result = await this.store.confirmMemoryResume( + { + userId: basis.userId, + relayHostId: basis.relayHostId, + reqId: request.reqId, + basisConnId: basis.basisConnId, + relayDeviceId: basis.relayDeviceId, + acceptedCredentialVersion: basis.acceptedCredentialVersion, + acceptedAs: basis.acceptedAs + }, + { onRenewalFailure: this.onRenewalFailure } + ) + this.results.set(key, { basisConnId: basis.basisConnId, result, at: this.now() }) + return result + } + + private expireResults(): void { + const cutoff = this.now() - MEMORY_CONFIRM_RESULT_RETENTION_MS + for (const [key, stored] of this.results) { + if (stored.at > cutoff) break + this.results.delete(key) + } + } +} diff --git a/cloud/apps/relay/src/phone-accept-log.ts b/cloud/apps/relay/src/phone-accept-log.ts index 785148ff1c5..f565ee7b435 100644 --- a/cloud/apps/relay/src/phone-accept-log.ts +++ b/cloud/apps/relay/src/phone-accept-log.ts @@ -11,6 +11,7 @@ type PhoneAcceptAnomaly = | { event: 'orca_relay_client_accept_db_error'; cause: PhoneAcceptDatabaseCause } | { event: 'orca_relay_client_miss_row_names_here'; relayHostIdDigest: string } | { event: 'orca_relay_client_miss_read_error'; cause: PhoneAcceptDatabaseCause } + | { event: 'orca_relay_resume_renewal_failed'; cause: PhoneAcceptDatabaseCause } export class PhoneAcceptLog { private readonly window: AssignmentRejectionLogWindow @@ -45,6 +46,11 @@ export class PhoneAcceptLog { this.record({ event: 'orca_relay_client_miss_read_error', cause }) } + // A due renewal write failed; the confirm still answered the stored expiry. + renewalFailed(cause: PhoneAcceptDatabaseCause): void { + this.record({ event: 'orca_relay_resume_renewal_failed', cause }) + } + private record(anomaly: PhoneAcceptAnomaly): void { if (this.window.admit(anomaly.event, anomaly)) this.write(anomaly, 0) } diff --git a/cloud/apps/relay/src/test-fixtures/statement-recording-database.ts b/cloud/apps/relay/src/test-fixtures/statement-recording-database.ts index 9d2db6c727a..d195ed6c552 100644 --- a/cloud/apps/relay/src/test-fixtures/statement-recording-database.ts +++ b/cloud/apps/relay/src/test-fixtures/statement-recording-database.ts @@ -9,6 +9,8 @@ import type { // described. BEGIN and COMMIT are recorded too: each is a trip to a remote database. export class StatementRecordingDatabase implements RelayDatabase { readonly dialect: RelayDatabase['dialect'] + // A statement this matches fails as a dropped connection would, before reaching the database. + failing: ((sql: string) => boolean) | null = null constructor( private readonly inner: RelayDatabase, @@ -19,6 +21,7 @@ export class StatementRecordingDatabase implements RelayDatabase { async query(sql: string, params: unknown[] = []): Promise { this.statements.push(normalized(sql)) + if (this.failing?.(normalized(sql))) throw new Error('Connection terminated unexpectedly') return await this.inner.query(sql, params) } @@ -34,7 +37,11 @@ export class StatementRecordingDatabase implements RelayDatabase { this.statements.push('BEGIN') try { return await this.inner.transaction( - async (transaction) => await operation(new StatementRecordingDatabase(transaction, this.statements)), + async (transaction) => { + const recording = new StatementRecordingDatabase(transaction, this.statements) + recording.failing = this.failing + return await operation(recording) + }, options ) } finally {