mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 00:57:45 +00:00
feat(relay): keep a memory resume's basis in the cell and confirm it from one read (#30651)
* feat(relay): add the per-cell phonePath switch, effective only in reserve mode Step 6 of the relay redesign keys every phone-path change on this switch. The cell reports the effective value in runtime status, and the flag tool refuses phonePath=memory unless the cell is reserve, and a reserve flip that would turn phones over unnamed. * feat(relay): resume paired phones from one device read on memory cells With phonePath=memory in reserve mode, a phone resume is admitted from one device read and the cell's own seat, and takes no activity leases. A refused device closes 4401, a failed read 4503, and only a missing device row falls through to the invite path. A phone whose host has no seat gets one assignment read: 4409 when the row names another cell or the read fails, otherwise 4404. * feat(relay): keep a memory resume's basis in the cell and confirm it from one read A phone resumed on the memory path keeps its connection basis in the cell, not in relay_connection_bases. Its confirm rereads the device row once, renews it only when less than 29 days remain, and always answers a current-version confirm renewed:true with the expiry the database holds. Confirms route by where their basis lives, so a flip between accept and confirm completes either way. Retries answer from a 10-min in-memory result map. * fix(relay-ops): store phonePath=db as no key, so old images keep taking writes An explicit phonePath=db left in the switch file would make the flag tool refuse every later write after an image rollback, including a watcher's admitMode=db flip back. * fix(relay): log a failed no-seat read, recheck the switch after the device read A failed assignment read for a phone with no seat now logs a rate-limited line before its 4409. A switch that turned off during the device read sends the phone down today's path. resolveResume reuses the one-read lookup, so the expiry and grace rules have one copy. * fix(relay): log a failed memory renewal; test flips, replays and basis cleanup A renewal write that fails still answers the stored expiry, but now logs a rate-limited line so a persistently failing update is visible before phones age toward re-pairing.
This commit is contained in:
@@ -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> | 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<ReturnType<typeof cell>>) =>
|
||||
(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<string, [string, unknown[]]> = {
|
||||
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)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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<DeviceResumeConfirmed> {
|
||||
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<number> {
|
||||
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<void> {
|
||||
await this.database.query(`UPDATE relay_connection_bases SET active = ? WHERE basis_conn_id = ?`, [
|
||||
0,
|
||||
|
||||
@@ -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<string, WebSocket>()
|
||||
@@ -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')
|
||||
|
||||
@@ -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<string, MemoryConnectionBasis>()
|
||||
// Insertion order is age order, so expiry walks from the front.
|
||||
private readonly results = new Map<string, StoredResult>()
|
||||
|
||||
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<DeviceResumeConfirmed> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<PhoneAcceptAnomaly>
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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<SqlRow[]> {
|
||||
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 {
|
||||
|
||||
Reference in New Issue
Block a user