diff --git a/cloud/apps/relay-edge/src/env.ts b/cloud/apps/relay-edge/src/env.ts index 31de529480a..88cb6a0ca06 100644 --- a/cloud/apps/relay-edge/src/env.ts +++ b/cloud/apps/relay-edge/src/env.ts @@ -21,6 +21,9 @@ export type Env = { PING_INTERVAL_MS: string SILENCE_TIMEOUT_MS: string ORPHAN_GRACE_MS: string + // '0' stops offering the host-sent keepalive (host-relay-control.ts) and pings every control from + // the alarm again, including controls that took the offer earlier. Default: offered. + CLIENT_KEEPALIVE?: string // Object backend: store cleanup cadence (default 24 h) and how long an object keeps the data of // a host that moved away before wiping it (default 30 days). Tests shorten both. SWEEP_INTERVAL_MS?: string @@ -69,6 +72,7 @@ export type EdgeConfig = { orphanGraceMs: number sweepIntervalMs: number leftRetentionMs: number + clientKeepalive: boolean } const configs = new WeakMap() @@ -109,8 +113,11 @@ export function edgeConfig(env: Env): EdgeConfig { silenceTimeoutMs: positiveInt('silence_timeout', env.SILENCE_TIMEOUT_MS), orphanGraceMs: positiveInt('orphan_grace', env.ORPHAN_GRACE_MS), sweepIntervalMs: positiveInt('sweep_interval', env.SWEEP_INTERVAL_MS ?? String(24 * 60 * 60 * 1000)), - leftRetentionMs: positiveInt('left_retention', env.LEFT_RETENTION_MS ?? String(30 * 24 * 60 * 60 * 1000)) + leftRetentionMs: positiveInt('left_retention', env.LEFT_RETENTION_MS ?? String(30 * 24 * 60 * 60 * 1000)), + clientKeepalive: env.CLIENT_KEEPALIVE !== '0' } + // The offer carries the ping interval, which the contract bounds to 1..60 s. + config.clientKeepalive &&= config.pingIntervalMs >= 1_000 && config.pingIntervalMs <= 60_000 // Shipped desktops close a control that is silent for 75 s; a ping must arrive well inside it. if (config.pingIntervalMs >= config.silenceTimeoutMs) { throw new Error('relay_edge_config_ping_interval') } configs.set(env, config) diff --git a/cloud/apps/relay-edge/src/host-relay-base.ts b/cloud/apps/relay-edge/src/host-relay-base.ts index 672887fbc88..f3f34839b10 100644 --- a/cloud/apps/relay-edge/src/host-relay-base.ts +++ b/cloud/apps/relay-edge/src/host-relay-base.ts @@ -7,6 +7,7 @@ import { RELAY_CLOSE_CODE } from '../../../packages/relay-contract/src/close-cod import { CONTROL_CONTINUITY_LIMITS } from '../../../packages/relay-contract/src/control-continuity' import { ASSIGNMENT_LIMITS } from '../../../packages/relay-contract/src/assignment-invariants' import { relayHostCloseReasonFrom } from '../../../packages/relay-contract/src/host-close-reason' +import { RELAY_CLIENT_KEEPALIVE_ACK_FRAME, RELAY_CLIENT_KEEPALIVE_FRAME } from '../../../packages/relay-contract/src/control-messages' import { parseConnId } from './splice-slot' import { metric } from './metrics' import { ObjectStore } from './object-store' @@ -39,6 +40,11 @@ export type ControlAttachment = { lastRefreshAt?: number // Client ingress '/', for metrics. place?: string + // The host took the keepalive offer: it sends the keepalive frame, which the runtime answers + // without waking this object, so the alarm sends it no pings (keepalive() below). + clientKeepalive?: boolean + // A keepalive control found silent: one relay ping sent at this time (see probeSilentControls). + probeAt?: number } export type Attachment = ControlAttachment export type Conn = { @@ -65,6 +71,8 @@ export type AuthorizeResult = export type ConfirmResult = { ok: true; hello: RelayHello } | { ok: false; code: number } export const NO_DEADLINE = Number.MAX_SAFE_INTEGER +// How long a silent keepalive control has to answer the relay ping before it is closed. +export const KEEPALIVE_PROBE_MS = 10_000 export const SWEEP_AT_KEY = 'sweepAt' // 6 h ± 30 min, as the cells: desktops re-ask the director at least this often. export const LEASE_BASE_MS = 6 * 60 * 60 * 1000 @@ -80,6 +88,12 @@ export abstract class HostRelayBase extends DurableObject { protected recentPicks: { slot: number; at: number }[] = [] protected backendCache: EdgeBackend | null = null + constructor(ctx: DurableObjectState, env: Env) { + super(ctx, env) + // Answered by the runtime while the object sleeps: no wake, no alarm, no duration. + ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair(RELAY_CLIENT_KEEPALIVE_FRAME, RELAY_CLIENT_KEEPALIVE_ACK_FRAME)) + } + protected get config(): EdgeConfig { return edgeConfig(this.env) } @@ -101,6 +115,17 @@ export abstract class HostRelayBase extends DurableObject { return a.phase === 'active' && now <= a.claims.exp } + // Host-sent keepalive in use for this control (and not switched off since it was offered). + protected keepalive(a: ControlAttachment): boolean { + return a.clientKeepalive === true && this.config.clientKeepalive + } + + // Last sign of life: any frame that woke the object, or a keepalive the runtime answered. + protected lastSeen(ws: WebSocket, a: ControlAttachment): number { + const answered = this.ctx.getWebSocketAutoResponseTimestamp(ws)?.getTime() ?? 0 + return Math.max(a.lastPongAt ?? 0, answered) + } + protected otherActiveControls(ws: WebSocket): WebSocket[] { return this.ctx.getWebSockets('control').filter((s) => s !== ws && this.attachment(s)?.phase === 'active') } @@ -144,7 +169,9 @@ export abstract class HostRelayBase extends DurableObject { // ---------------- lifecycle ---------------- - protected async onSocketGone(ws: WebSocket, code: number, reason: string): Promise { + // `goneAt`: when the control was last known alive plus the silence timeout, for a control found + // dead later than the alarm would have found it; the orphan grace counts from then. + protected async onSocketGone(ws: WebSocket, code: number, reason: string, goneAt = Date.now()): Promise { const a = this.attachment(ws) if (!a) { return } // During webSocketClose the closing socket is still listed; clear it first so it can't count as live. @@ -156,7 +183,7 @@ export abstract class HostRelayBase extends DurableObject { const hostReason = relayHostCloseReasonFrom(reason) await this.ctx.storage.put('session', { ...session, - orphanedAt: Date.now(), + orphanedAt: goneAt, ...(hostReason ? { closeReason: { reason: hostReason, expiresAt: Date.now() + ASSIGNMENT_LIMITS.dormantTtlMs } } : {}) }) } @@ -168,19 +195,32 @@ export abstract class HostRelayBase extends DurableObject { protected async reschedule(): Promise { let next = NO_DEADLINE + let conns: Record | null = null for (const ws of this.ctx.getWebSockets()) { const a = this.attachment(ws) if (!a) { continue } + if (a.k !== 'control' || a.phase !== 'active') { + next = Math.min(next, a.deadline) + continue + } next = Math.min( next, - ...(a.k === 'control' && a.phase === 'active' - ? [ - (a.lastPingAt ?? 0) + this.config.pingIntervalMs, - a.claims.exp + CONTROL_CONTINUITY_LIMITS.expiredAuthExistingSpliceGraceMs + 1, - (a.leaseExpiresAt ?? NO_DEADLINE - 1) + 1 - ] - : [a.deadline]) + a.claims.exp + CONTROL_CONTINUITY_LIMITS.expiredAuthExistingSpliceGraceMs + 1, + (a.leaseExpiresAt ?? NO_DEADLINE - 1) + 1 ) + if (!this.keepalive(a)) { + next = Math.min(next, (a.lastPingAt ?? 0) + this.config.pingIntervalMs) + continue + } + const seen = this.lastSeen(ws, a) + if (a.probeAt !== undefined && a.probeAt >= seen) { + next = Math.min(next, a.probeAt + KEEPALIVE_PROBE_MS) + continue + } + // Phones attached or waiting: their splices must close on time if the host dies, so check + // its silence when it would run out. Otherwise no alarm for liveness at all. + conns ??= await this.conns() + if (Object.keys(conns).length > 0) { next = Math.min(next, seen + this.config.silenceTimeoutMs + 1) } } const session = await this.ctx.storage.get('session') if (session?.orphanedAt != null) { next = Math.min(next, session.orphanedAt + this.config.orphanGraceMs) } diff --git a/cloud/apps/relay-edge/src/host-relay-control.ts b/cloud/apps/relay-edge/src/host-relay-control.ts index dddb466028d..16697097b1c 100644 --- a/cloud/apps/relay-edge/src/host-relay-control.ts +++ b/cloud/apps/relay-edge/src/host-relay-control.ts @@ -1,6 +1,6 @@ -// HostRelay control frames once a host is past the handshake: liveness, pongs, auth refresh and +// HostRelay control frames once a host is past the handshake: liveness, pongs, keepalives, auth refresh and // the host's credential commands. -import { AuthRefreshSchema, DeviceRevokeSchema, InviteCreateSchema } from '../../../packages/relay-contract/src/control-messages' +import { AuthRefreshSchema, DeviceRevokeSchema, InviteCreateSchema, RELAY_CLIENT_KEEPALIVE_ACK_FRAME } from '../../../packages/relay-contract/src/control-messages' import { DeviceCredentialInstallSchema, DeviceCredentialInstallStatusSchema, DeviceResumeConfirmSchema } from '../../../packages/relay-contract/src/credential-messages' import { RELAY_CLOSE_CODE } from '../../../packages/relay-contract/src/close-codes' import { RELAY_PROTOCOL_LIMITS } from '../../../packages/relay-contract/src/protocol-limits' @@ -12,6 +12,45 @@ import { HostRelayHandshake } from './host-relay-handshake' import type { ControlAttachment } from './host-relay-base' export abstract class HostRelayControl extends HostRelayHandshake { + // ---------------- silent keepalive controls ---------------- + + // Active controls silent past the timeout. No liveness field (attachments older than it): the + // alarm judges those, as before. + protected silentControls(now: number): { ws: WebSocket; a: ControlAttachment; seen: number }[] { + return this.ctx.getWebSockets('control').flatMap((ws) => { + const a = this.attachment(ws) + if (a?.k !== 'control' || a.phase !== 'active' || a.lastPongAt === undefined) { return [] } + const seen = this.lastSeen(ws, a) + return now - seen > this.config.silenceTimeoutMs ? [{ ws, a, seen }] : [] + }) + } + + // Where the host is going away anyway (export, erase, drain, a new control of the same host): a + // silent control is closed now. Its orphan time is the latest the ping alarm would have found it, + // so rebind and the orphan grace are never shorter than before. + protected async closeSilentControls(now = Date.now()): Promise { + for (const { ws, seen } of this.silentControls(now)) { + this.close(ws, RELAY_CLOSE_CODE.PEER_DROPPED, 'control silence timeout') + await this.onSocketGone(ws, RELAY_CLOSE_CODE.PEER_DROPPED, '', Math.min(now, seen + this.config.silenceTimeoutMs + this.config.pingIntervalMs)) + } + } + + // Where a phone needs the host: a silent keepalive control may be alive with its own timer held + // back (a sleeping app), so it is sent one relay ping, answered from socket I/O, and closed only if + // nothing arrives within KEEPALIVE_PROBE_MS (alarm). Meanwhile it counts as live, as a control + // inside the silence timeout always did. + protected async probeSilentControls(now = Date.now()): Promise { + let probed = false + for (const { ws, a, seen } of this.silentControls(now)) { + if (!this.keepalive(a) || (a.probeAt !== undefined && a.probeAt >= seen)) { continue } + this.send(ws, { type: 'ping', t: now }) + a.probeAt = now + this.attach(ws, a) + probed = true + } + if (probed) { await this.reschedule() } + } + // ---------------- control ---------------- protected async onControl(ws: WebSocket, a: ControlAttachment, message: string | ArrayBuffer): Promise { @@ -47,6 +86,12 @@ export abstract class HostRelayControl extends HostRelayHandshake { if (typeof body.t === 'number') { metric(this.env, 'pong_rtt_ms', now - body.t, '', [], '', a.place) } return } + // Normally answered by the runtime (host-relay-base.ts); a frame that differs only in spacing + // still gets its answer here. + if (type === 'keepalive') { + ws.send(RELAY_CLIENT_KEEPALIVE_ACK_FRAME) + return + } if (type === 'auth-refresh') { return this.onAuthRefresh(ws, body) } return this.onCommand(ws, a, type, body) } diff --git a/cloud/apps/relay-edge/src/host-relay-handshake.ts b/cloud/apps/relay-edge/src/host-relay-handshake.ts index ca9d83498d7..bc153be8a0c 100644 --- a/cloud/apps/relay-edge/src/host-relay-handshake.ts +++ b/cloud/apps/relay-edge/src/host-relay-handshake.ts @@ -2,7 +2,7 @@ // eligibility, and the proof that activates a control (generation, resume secret, pending conns). import nacl from 'tweetnacl' import { buildHostChallengePlaintext, buildHostProofMacInput, buildHostProofTranscript } from '../../../packages/relay-contract/src/host-proof-transcript' -import { HostChallengeAckSchema, HostHelloSchema, RELAY_HOST_CAPABILITY_PENDING_CONN_DETAILS } from '../../../packages/relay-contract/src/control-messages' +import { HostChallengeAckSchema, HostHelloSchema, RELAY_HOST_CAPABILITY_CLIENT_KEEPALIVE, RELAY_HOST_CAPABILITY_PENDING_CONN_DETAILS } from '../../../packages/relay-contract/src/control-messages' import { RELAY_CLOSE_CODE } from '../../../packages/relay-contract/src/close-codes' import { metric } from './metrics' import type { EdgeClaims } from './env' @@ -162,6 +162,7 @@ export abstract class HostRelayHandshake extends HostRelayBase { a.lastPingAt = now a.leaseExpiresAt = leaseExpiresAt a.deadline = NO_DEADLINE + a.clientKeepalive = this.config.clientKeepalive && a.caps.includes(RELAY_HOST_CAPABILITY_CLIENT_KEEPALIVE) delete a.expectedProof delete a.hello delete a.rebind @@ -188,6 +189,8 @@ export abstract class HostRelayHandshake extends HostRelayBase { activeConnIds: activeConnIds.slice(0, 8), pendingConns: pendingConns.slice(0, 8) }) + // After the ack, which strict shipped parsers must see unchanged; only hosts that asked get it. + if (a.clientKeepalive) { this.send(ws, { type: 'keepalive-offer', v: 1, intervalMs: this.config.pingIntervalMs }) } metric(this.env, 'control_active', 1, rebind ? 'rebind' : 'new', [], '', a.place) await this.reschedule() } diff --git a/cloud/apps/relay-edge/src/host-relay-phones.ts b/cloud/apps/relay-edge/src/host-relay-phones.ts index 22e5da6a35d..f9278f9589e 100644 --- a/cloud/apps/relay-edge/src/host-relay-phones.ts +++ b/cloud/apps/relay-edge/src/host-relay-phones.ts @@ -61,6 +61,8 @@ export abstract class HostRelayPhones extends HostRelayControl { return { ok: false, code: RELAY_CLOSE_CODE.DRAINING } } if (!reservation) { return { ok: false, code: RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL } } + // A keepalive host may have died unnoticed (no alarm watches an idle one): probe it first. + await this.probeSilentControls() const control = this.activeControl() const seated = control !== null && @@ -95,6 +97,8 @@ export abstract class HostRelayPhones extends HostRelayControl { relayDeviceId: reservation.relayDeviceId, attachDeadlineMs: RELAY_PROTOCOL_LIMITS.hostAttachDeadlineMs }) + // The first phone of a keepalive host arms its silence check (reschedule). + if (this.keepalive(control.a)) { await this.reschedule() } return { ok: true, connTicket, generation } } diff --git a/cloud/apps/relay-edge/src/host-relay.ts b/cloud/apps/relay-edge/src/host-relay.ts index 397b9dd3844..958232f8ff4 100644 --- a/cloud/apps/relay-edge/src/host-relay.ts +++ b/cloud/apps/relay-edge/src/host-relay.ts @@ -16,7 +16,7 @@ import type { AssignBody, DeviceRow } from './admin' import { ADMIN_JTI_KEY, admitJti, type AdminReplayGuard, type ReplayRefusal, type SeenJtis } from './admin-jti' import type { AssignResult, StoredAssignment } from './object-store' import type { EdgeClaims } from './env' -import { SWEEP_AT_KEY, type Session } from './host-relay-base' +import { KEEPALIVE_PROBE_MS, SWEEP_AT_KEY, type Session } from './host-relay-base' import { HostRelayPhones } from './host-relay-phones' export type { AuthorizeResult, ConfirmResult } from './host-relay-base' @@ -48,6 +48,7 @@ export class HostRelay extends HostRelayPhones { > { const refused = await this.claimAdminJti(replay) if (refused) { return refused } + await this.closeSilentControls() if (this.liveControls().length > 0) { return { outcome: 'busy' } } const devices = this.store.freezeForExport() await this.closeSplices(() => true, RELAY_CLOSE_CODE.WRONG_CELL, 'host moving') @@ -64,7 +65,8 @@ export class HostRelay extends HostRelayPhones { const assignment = this.store.assignment() // Another account now holds this machine's object; its move on wiped the earlier account's data. if (assignment && assignment.userId !== userId) { return { outcome: 'mismatch' } } - if (this.ctx.getWebSockets('control').length > 0 || this.store.activeBases() > 0) { return { outcome: 'busy' } } + await this.closeSilentControls() + if (this.liveControls().length > 0 || this.store.activeBases() > 0) { return { outcome: 'busy' } } await this.ctx.storage.deleteAlarm() await this.ctx.storage.deleteAll() this.storeCache = null @@ -76,6 +78,7 @@ export class HostRelay extends HostRelayPhones { // desktop re-asks the director), and goes through the same path as that expiry. The assignment, // devices and splices are left as they are. async adminDrain(replay?: AdminReplayGuard | null): Promise { + await this.closeSilentControls() const controls = this.liveControls() // An idle drain of a host nobody pushed changes nothing and must not give the object storage, // so its jti is not recorded; a copy of that token can at most drain a control that arrived @@ -128,6 +131,8 @@ export class HostRelay extends HostRelayPhones { }) } if (kind !== 'control') { return new Response(null, { status: 404 }) } + // A new control's hello decides rebind against the controls still listed. + await this.closeSilentControls() const origin = request.headers.get('x-edge-origin') ?? '' const claims = JSON.parse(request.headers.get('x-edge-claims') ?? 'null') as EdgeClaims const pair = new WebSocketPair() @@ -166,7 +171,15 @@ export class HostRelay extends HostRelayPhones { const a = this.attachment(ws) if (!a) { continue } if (a.k === 'control' && a.phase === 'active') { - if (now - (a.lastPongAt ?? now) > this.config.silenceTimeoutMs) { + const seen = a.lastPongAt === undefined ? now : this.lastSeen(ws, a) + // A keepalive control gets one relay ping first (probeSilentControls); closed if it stays silent. + if (this.keepalive(a) && now - seen > this.config.silenceTimeoutMs && (a.probeAt === undefined || a.probeAt < seen)) { + this.send(ws, { type: 'ping', t: now }) + a.probeAt = now + this.attach(ws, a) + continue + } + if (now - seen > this.config.silenceTimeoutMs && (!this.keepalive(a) || now - a.probeAt! >= KEEPALIVE_PROBE_MS - 50)) { this.close(ws, RELAY_CLOSE_CODE.PEER_DROPPED, 'control silence timeout') await this.onSocketGone(ws, RELAY_CLOSE_CODE.PEER_DROPPED, '') continue @@ -185,7 +198,7 @@ export class HostRelay extends HostRelayPhones { await this.onSocketGone(ws, RELAY_CLOSE_CODE.DRAINING, '') continue } - if (now - (a.lastPingAt ?? 0) >= this.config.pingIntervalMs - 50) { + if (!this.keepalive(a) && now - (a.lastPingAt ?? 0) >= this.config.pingIntervalMs - 50) { if (a.lastPingAt) { metric(this.env, 'alarm_lag_ms', Math.max(0, now - (a.lastPingAt + this.config.pingIntervalMs)), '', [], '', a.place) } this.send(ws, { type: 'ping', t: now }) a.lastPingAt = now diff --git a/cloud/apps/relay-edge/test/harness-mvp-object.ts b/cloud/apps/relay-edge/test/harness-mvp-object.ts index aedc78680d6..89ec43b1e7c 100644 --- a/cloud/apps/relay-edge/test/harness-mvp-object.ts +++ b/cloud/apps/relay-edge/test/harness-mvp-object.ts @@ -59,8 +59,10 @@ export async function mint(key: KeyLike, relayHostId: string, issuer = ISSUER): } // `drop` returns true for an outbound frame the test wants lost (a host that stops answering pings). -export function control(host: Host, jwt: string, epoch = EPOCH, opts: { drop?: (frame: string) => boolean } = {}) { - const events = { closes: [] as number[], opens: [] as { connId: string }[], drains: [] as Record[], dropped: 0 } +// `caps` replaces the shipped desktop's capability header (default: as shipped today). +export const SHIPPED_CAPS = 'pending-conn-details,idle-regional-rehome-v1' +export function control(host: Host, jwt: string, epoch = EPOCH, opts: { drop?: (frame: string) => boolean; caps?: string } = {}) { + const events = { closes: [] as number[], opens: [] as { connId: string }[], drains: [] as Record[], dropped: 0, pings: 0, offers: 0, acks: 0 } let transport: CloudRelayTransport | null = null const client = new RelayControlClient({ cellUrl: ORIGIN, @@ -78,10 +80,16 @@ export function control(host: Host, jwt: string, epoch = EPOCH, opts: { drop?: ( onClose: (code) => events.closes.push(code), createSocket: (url, token) => { const ws = new WebSocket(url, { - headers: { authorization: `Bearer ${token}`, 'x-orca-host-capabilities': 'pending-conn-details,idle-regional-rehome-v1' }, + headers: { authorization: `Bearer ${token}`, 'x-orca-host-capabilities': opts.caps ?? process.env.MVP_HOST_CAPS ?? SHIPPED_CAPS }, perMessageDeflate: false, maxPayload: 64 * 1024 }) + ws.on('message', (raw) => { + const type = (JSON.parse(raw.toString()) as { type?: unknown }).type + if (type === 'ping') { events.pings++ } + if (type === 'keepalive-offer') { events.offers++ } + if (type === 'keepalive-ack') { events.acks++ } + }) const send = ws.send.bind(ws) ws.send = ((data: string, ...rest: never[]) => { if (opts.drop?.(String(data))) { diff --git a/cloud/apps/relay-edge/test/mvp-object-hardening.ts b/cloud/apps/relay-edge/test/mvp-object-hardening.ts index a304dabe9e4..5bd6553acab 100644 --- a/cloud/apps/relay-edge/test/mvp-object-hardening.ts +++ b/cloud/apps/relay-edge/test/mvp-object-hardening.ts @@ -5,7 +5,7 @@ import { readdirSync } from 'node:fs' import { join } from 'node:path' import { DatabaseSync } from 'node:sqlite' import type { KeyLike } from 'jose' -import { EPOCH, ORIGIN, adminCall, assign, control, mint, newHost, record, sleep, waitFor } from './harness-mvp-object' +import { EPOCH, ORIGIN, SHIPPED_CAPS, adminCall, assign, control, mint, newHost, record, sleep, waitFor } from './harness-mvp-object' // Local timing (mvp-object.ts): ping every 3 s, silence timeout 9 s. const SILENCE_MS = 9_000 @@ -47,6 +47,87 @@ export async function heartbeatCheck(key: KeyLike): Promise { } } +// K1: host-sent keepalive. A host advertising client-keepalive-v1 gets the offer, then sends the +// keepalive, which the runtime answers: the edge sends it no pings and it stays up. One that goes +// silent is not watched by any alarm while idle, but is closed 4408 the moment the director needs +// the host (export), and a live one is not. A host without the capability is pinged as before. +export async function keepaliveCheck(key: KeyLike): Promise { + const caps = `${SHIPPED_CAPS},client-keepalive-v1` + const liveHost = newHost() + const silentHost = newHost() + const legacyHost = newHost() + await Promise.all([liveHost, silentHost, legacyHost].map((h) => assign(h.relayHostId, 'edge', EPOCH, []))) + let muted = false + const live = control(liveHost, await mint(key, liveHost.relayHostId), EPOCH, { caps }) + const silent = control(silentHost, await mint(key, silentHost.relayHostId), EPOCH, { caps, drop: () => muted }) + const legacy = control(legacyHost, await mint(key, legacyHost.relayHostId)) + // Its keepalive timer stalls (a throttled app) but it still answers relay pings from socket I/O. + const nappingHost = newHost() + await assign(nappingHost.relayHostId, 'edge', EPOCH, []) + const napping = control(nappingHost, await mint(key, nappingHost.relayHostId), EPOCH, { caps, drop: (f) => muted && f === '{"type":"keepalive"}' }) + await Promise.all([live.client.connect(), silent.client.connect(), legacy.client.connect(), napping.client.connect()]) + try { + await sleep(1_000) + muted = true + await sleep(SILENCE_MS + 4_000) + record( + 'K1 keepalive host: offered, answered, no edge pings, up past the silence timeout', + live.events.offers === 1 && live.events.acks >= 3 && live.events.pings === 0 && live.client.isLive() && live.events.closes.length === 0, + `offers=${live.events.offers} acks=${live.events.acks} pings=${live.events.pings} closes=${live.events.closes}` + ) + record( + 'K1 host without the capability: no offer, pinged as before', + legacy.events.offers === 0 && legacy.events.pings >= 3 && legacy.client.isLive(), + `offers=${legacy.events.offers} pings=${legacy.events.pings}` + ) + // Local sweep alarm every 3 s: it finds the napping host silent and probes it with a relay ping. + record( + 'K2 keepalive host with a stalled timer: probed with a relay ping, answers, stays up', + napping.events.pings >= 1 && napping.client.isLive() && napping.events.closes.length === 0, + `pings=${napping.events.pings} dropped=${napping.events.dropped} closes=${napping.events.closes}` + ) + const nappingExport = await adminCall('export', { relayHostId: nappingHost.relayHostId }) + record('K2 export of the probed, answering host: busy', nappingExport.json?.outcome === 'busy', JSON.stringify(nappingExport.json?.outcome)) + const liveExport = await adminCall('export', { relayHostId: liveHost.relayHostId }) + const silentExport = await adminCall('export', { relayHostId: silentHost.relayHostId }) + await waitFor(() => silent.events.closes.length > 0, 3_000) + record( + 'K1 export: live keepalive host busy; silent one closed 4408 and exported', + liveExport.json?.outcome === 'busy' && silentExport.json?.outcome === 'exported' && silent.events.closes.includes(4408) && live.events.closes.length === 0, + `live=${JSON.stringify(liveExport.json?.outcome)} silent=${JSON.stringify(silentExport.json?.outcome)} closes=${silent.events.closes}` + ) + } finally { + live.client.closeNow() + silent.client.closeNow() + legacy.client.closeNow() + napping.client.closeNow() + } +} + +// K3: a keepalive host that goes fully silent is probed when an alarm runs (here the local 3 s +// sweep; in production the silence check armed while phones are attached), and closed 4408 when +// the probe goes unanswered: silence timeout plus the 10 s probe. +export async function keepaliveProbeCheck(key: KeyLike): Promise { + const host = newHost() + await assign(host.relayHostId, 'edge', EPOCH, []) + let muted = false + const c = control(host, await mint(key, host.relayHostId), EPOCH, { caps: `${SHIPPED_CAPS},client-keepalive-v1`, drop: () => muted }) + await c.client.connect() + muted = true + const started = Date.now() + try { + const closed = await waitFor(() => c.events.closes.length > 0, SILENCE_MS + 10_000 + 8_000) + const after = Date.now() - started + record( + 'K3 silent keepalive host: probed, then closed 4408 after the silence timeout plus the probe', + closed && c.events.closes.includes(4408) && c.events.pings >= 1 && after >= SILENCE_MS + 10_000 - 500, + `closedAfter=${after}ms pings=${c.events.pings} closes=${c.events.closes}` + ) + } finally { + c.client.closeNow() + } +} + // F5: POST /__admin/drain sends the host's controls back to the director (drain frame, 4503). export async function drainCheck(key: KeyLike): Promise { const host = newHost() diff --git a/cloud/apps/relay-edge/test/mvp-object.ts b/cloud/apps/relay-edge/test/mvp-object.ts index 8183bd102ba..5558bb2d179 100644 --- a/cloud/apps/relay-edge/test/mvp-object.ts +++ b/cloud/apps/relay-edge/test/mvp-object.ts @@ -41,7 +41,7 @@ import { waitFor, type Host } from './harness-mvp-object' -import { drainCheck, heartbeatCheck, objectHealthCheck } from './mvp-object-hardening' +import { drainCheck, heartbeatCheck, keepaliveCheck, keepaliveProbeCheck, objectHealthCheck } from './mvp-object-hardening' import { replayCheck, rotationCheck } from './mvp-object-replay' async function mainRemote(): Promise { @@ -107,7 +107,7 @@ async function main(): Promise { await sleep(500) } await run(host, privateKey, () => jwksHits) - await Promise.all([heartbeatCheck(privateKey), drainCheck(privateKey), replayCheck(), rotationCheck()]) + await Promise.all([heartbeatCheck(privateKey), keepaliveCheck(privateKey), keepaliveProbeCheck(privateKey), drainCheck(privateKey), replayCheck(), rotationCheck()]) await sweepCheck() await preAuthCheck(stateDir) await objectHealthCheck(stateDir) diff --git a/cloud/packages/relay-contract/src/control-messages.ts b/cloud/packages/relay-contract/src/control-messages.ts index c7b74de118a..bc7c8ffda89 100644 --- a/cloud/packages/relay-contract/src/control-messages.ts +++ b/cloud/packages/relay-contract/src/control-messages.ts @@ -51,6 +51,20 @@ export const RELAY_ASSIGNMENT_LEASE_HEADER = 'x-orca-relay-assignment-lease' // Advertise before sending pendingConns details: legacy strict parsers would reject the entire ack. export const RELAY_HOST_CAPABILITY_PENDING_CONN_DETAILS = 'pending-conn-details' export const RELAY_HOST_CAPABILITY_IDLE_REGIONAL_REHOME = 'idle-regional-rehome-v1' +// The host can send the keepalive below. A relay that supports it says so with a keepalive-offer +// after the hello-ack; only then does the host send it. Relays that never offer keep pinging. +export const RELAY_HOST_CAPABILITY_CLIENT_KEEPALIVE = 'client-keepalive-v1' +// Exact frames: the edge answers the request without waking the host's object, so they never vary. +export const RELAY_CLIENT_KEEPALIVE_FRAME = '{"type":"keepalive"}' +export const RELAY_CLIENT_KEEPALIVE_ACK_FRAME = '{"type":"keepalive-ack"}' + +export const KeepaliveOfferSchema = z + .object({ + type: z.literal('keepalive-offer'), + v: z.literal(1), + intervalMs: z.number().int().min(1_000).max(60_000) + }) + .strict() export function parseRelayHostCapabilities( header: string | string[] | undefined diff --git a/config/recurring-timer-baseline.txt b/config/recurring-timer-baseline.txt index 4f313845137..343b138c771 100644 --- a/config/recurring-timer-baseline.txt +++ b/config/recurring-timer-baseline.txt @@ -48,7 +48,7 @@ 1 src/main/rate-limits/service/service-polling.ts 1 src/main/runtime/agent-state-rules/agent-state-rules-live-update.ts 1 src/main/runtime/mobile-cloud-relay/desktop-relay-service.ts -2 src/main/runtime/mobile-cloud-relay/relay-control-liveness.ts +3 src/main/runtime/mobile-cloud-relay/relay-control-liveness.ts 1 src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts 1 src/main/runtime/rpc/mobile-socket-wiring.ts 1 src/main/runtime/rpc/remote-runtime-server-heartbeat.ts diff --git a/src/main/runtime/mobile-cloud-relay/relay-control-client.test.ts b/src/main/runtime/mobile-cloud-relay/relay-control-client.test.ts index b134747fe0b..a43d8abd3d7 100644 --- a/src/main/runtime/mobile-cloud-relay/relay-control-client.test.ts +++ b/src/main/runtime/mobile-cloud-relay/relay-control-client.test.ts @@ -204,7 +204,7 @@ describe('RelayControlClient', () => { const { socket, authorization, capabilities, path } = await accepted expect(authorization).toBe('Bearer scoped-token') // Strict older cells reject new hello keys; advertise capabilities in upgrade headers. - expect(capabilities).toBe('pending-conn-details,idle-regional-rehome-v1') + expect(capabilities).toBe('pending-conn-details,idle-regional-rehome-v1,client-keepalive-v1') expect(path).toBe('/v1/host/control') const hello = await nextJson(socket) expect(hello).toMatchObject({ @@ -274,6 +274,13 @@ describe('RelayControlClient', () => { // ast-grep-ignore: test-no-wall-clock -- Real loopback protocol timestamps must match the client clock. socket.send(JSON.stringify({ type: 'ping', t: Date.now() })) await expect(nextJson(socket)).resolves.toMatchObject({ type: 'pong' }) + // Offered: the host sends the exact keepalive frame (the edge answers it without waking). + socket.send(JSON.stringify({ type: 'keepalive-offer', v: 1, intervalMs: 1_000 })) + const keepalive = await new Promise((resolve) => { + socket.once('message', (raw) => resolve(raw.toString())) + }) + expect(keepalive).toBe('{"type":"keepalive"}') + socket.send('{"type":"keepalive-ack"}') socket.send( JSON.stringify({ type: 'conn-open', diff --git a/src/main/runtime/mobile-cloud-relay/relay-control-client.ts b/src/main/runtime/mobile-cloud-relay/relay-control-client.ts index 0744919c9d2..e5227a79034 100644 --- a/src/main/runtime/mobile-cloud-relay/relay-control-client.ts +++ b/src/main/runtime/mobile-cloud-relay/relay-control-client.ts @@ -9,7 +9,6 @@ import { RelayDrainMessageSchema, RelayHostChallengeMessageSchema, RelayHostHelloAckMessageSchema, - RelayPingMessageSchema, RELAY_HOST_CAPABILITY_HEADERS, encodeRelayHostHello, parseRelayControlMessage, @@ -48,6 +47,7 @@ export class RelayControlClient { this.liveness = new RelayControlLiveness({ cellUrl: this.relayOrigin, ping: () => this.socket?.ping(), + send: (frame) => this.socket?.send(frame), isLive: () => this.isLive(), terminate: () => this.socket?.terminate(), random: options.livenessRandom @@ -207,8 +207,7 @@ export class RelayControlClient { this.failProtocol('control message before activation') return } - if (RelayPingMessageSchema.safeParse(message).success) { - this.socket?.send(JSON.stringify({ type: 'pong', t: message.t })) + if (this.liveness.handleLivenessMessage(message)) { return } const connection = RelayConnectionOpenMessageSchema.safeParse(message) diff --git a/src/main/runtime/mobile-cloud-relay/relay-control-liveness.test.ts b/src/main/runtime/mobile-cloud-relay/relay-control-liveness.test.ts new file mode 100644 index 00000000000..202d376fc46 --- /dev/null +++ b/src/main/runtime/mobile-cloud-relay/relay-control-liveness.test.ts @@ -0,0 +1,85 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { RelayControlLiveness } from './relay-control-liveness' + +describe('RelayControlLiveness keepalive', () => { + afterEach(() => { + vi.useRealTimers() + }) + + function liveness(live = { value: true }) { + const terminate = vi.fn() + // The relay's ack is an inbound frame. + const sendKeepalive = vi.fn((frame: string) => { + expect(frame).toBe('{"type":"keepalive"}') + instance.noteInbound() + }) + const instance: RelayControlLiveness = new RelayControlLiveness({ + cellUrl: 'https://relay.test', + ping: vi.fn(), + isLive: () => live.value, + terminate, + send: sendKeepalive + }) + return { instance, sendKeepalive, terminate, live } + } + + it('sends nothing until the relay offers, then one keepalive per interval while live', () => { + vi.useFakeTimers() + const { instance, sendKeepalive, terminate, live } = liveness() + instance.start() + vi.advanceTimersByTime(30_000) + expect(sendKeepalive).not.toHaveBeenCalled() + instance.startKeepalive(25_000) + vi.advanceTimersByTime(50_000) + expect(sendKeepalive).toHaveBeenCalledTimes(2) + // A second offer does not add a second timer. + instance.startKeepalive(1_000) + vi.advanceTimersByTime(25_000) + expect(sendKeepalive).toHaveBeenCalledTimes(3) + // Answered keepalives alone keep the control inside the silence bound. + vi.advanceTimersByTime(200_000) + expect(terminate).not.toHaveBeenCalled() + expect(sendKeepalive).toHaveBeenCalledTimes(11) + live.value = false + vi.advanceTimersByTime(25_000) + expect(sendKeepalive).toHaveBeenCalledTimes(11) + instance.stop() + }) + + it('sends a keepalive instead of tearing down when its own timers were held back', () => { + vi.useFakeTimers() + const terminate = vi.fn() + const send = vi.fn() + const instance = new RelayControlLiveness({ + cellUrl: 'https://relay.test', + ping: vi.fn(), + isLive: () => true, + terminate, + send + }) + instance.start() + instance.startKeepalive(25_000) + // A throttled app: no timer ran for 100 s, then the silence check runs before the keepalive. + vi.setSystemTime(Date.now() + 100_000) + vi.advanceTimersToNextTimer() + expect(terminate).not.toHaveBeenCalled() + expect(send).toHaveBeenCalledWith('{"type":"keepalive"}') + // Still unanswered at a later check: the link is judged dead as before. + vi.advanceTimersByTime(80_000) + expect(terminate).toHaveBeenCalledOnce() + }) + + it('ignores an offer before the control is active and stops with the control', () => { + vi.useFakeTimers() + const { instance, sendKeepalive } = liveness() + instance.startKeepalive(25_000) + vi.advanceTimersByTime(50_000) + expect(sendKeepalive).not.toHaveBeenCalled() + instance.start() + instance.startKeepalive(25_000) + instance.stop() + vi.advanceTimersByTime(100_000) + expect(sendKeepalive).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/runtime/mobile-cloud-relay/relay-control-liveness.ts b/src/main/runtime/mobile-cloud-relay/relay-control-liveness.ts index 9368f4f4660..1d85b1431a0 100644 --- a/src/main/runtime/mobile-cloud-relay/relay-control-liveness.ts +++ b/src/main/runtime/mobile-cloud-relay/relay-control-liveness.ts @@ -1,5 +1,11 @@ import type { RelayControlRequestTimeout } from './relay-control-requests' import { RELAY_RENEWAL_JITTER_RATIO } from './relay-renewal-jitter' +import { + RELAY_CLIENT_KEEPALIVE_FRAME, + RelayKeepaliveAckMessageSchema, + RelayKeepaliveOfferMessageSchema, + RelayPingMessageSchema +} from './relay-control-protocol' // STA-7672: request timeout may be relay load or half-open TCP; probe before inferring death. @@ -19,6 +25,7 @@ type TeardownReason = 'probe-unanswered' | 'silence-limit' export type RelayControlLivenessOptions = { cellUrl: string ping: () => void + send?: (frame: string) => void isLive: () => boolean terminate: () => void random?: () => number @@ -28,6 +35,8 @@ export class RelayControlLiveness { private readonly random: () => number private probe: { timer: ReturnType; misses: number } | null = null private silenceTimer: ReturnType | null = null + private keepaliveTimer: ReturnType | null = null + private keepaliveSentAt = 0 private openedAt = 0 private lastInboundAt = 0 @@ -40,6 +49,12 @@ export class RelayControlLiveness { this.lastInboundAt = this.openedAt this.silenceTimer = setInterval(() => { if (Date.now() - this.lastInboundAt > SILENCE_LIMIT_MS) { + // Our own keepalive timer was held back too (a throttled app), so the silence says nothing + // about the link: send one now and judge on the next check. + if (this.keepaliveTimer && Date.now() - this.keepaliveSentAt > SILENCE_LIMIT_MS) { + this.sendKeepalive() + return + } this.tearDown('silence-limit') } }, SILENCE_CHECK_INTERVAL_MS) @@ -56,11 +71,48 @@ export class RelayControlLiveness { this.clearProbe() } + /** Relay pings, the keepalive offer and its acks; the caller already counted them as inbound. */ + handleLivenessMessage(message: Record): boolean { + const ping = RelayPingMessageSchema.safeParse(message) + if (ping.success) { + this.options.send?.(JSON.stringify({ type: 'pong', t: ping.data.t })) + return true + } + const offer = RelayKeepaliveOfferMessageSchema.safeParse(message) + if (offer.success) { + this.startKeepalive(offer.data.intervalMs) + return true + } + return RelayKeepaliveAckMessageSchema.safeParse(message).success + } + + // The relay offered the host-sent keepalive: its answer is an inbound frame, so it keeps the + // silence bound met without relay pings. Relays that never offer keep pinging as before. + startKeepalive(intervalMs: number): void { + if (this.keepaliveTimer || !this.silenceTimer || !this.options.send) { + return + } + this.keepaliveSentAt = Date.now() + this.keepaliveTimer = setInterval(() => this.sendKeepalive(), intervalMs) + this.keepaliveTimer.unref?.() + } + + private sendKeepalive(): void { + if (this.options.isLive()) { + this.keepaliveSentAt = Date.now() + this.options.send?.(RELAY_CLIENT_KEEPALIVE_FRAME) + } + } + stop(): void { if (this.silenceTimer) { clearInterval(this.silenceTimer) this.silenceTimer = null } + if (this.keepaliveTimer) { + clearInterval(this.keepaliveTimer) + this.keepaliveTimer = null + } this.clearProbe() } diff --git a/src/main/runtime/mobile-cloud-relay/relay-control-protocol.ts b/src/main/runtime/mobile-cloud-relay/relay-control-protocol.ts index d1609c628b9..a0fca00db85 100644 --- a/src/main/runtime/mobile-cloud-relay/relay-control-protocol.ts +++ b/src/main/runtime/mobile-cloud-relay/relay-control-protocol.ts @@ -30,9 +30,13 @@ const ConnectionKindSchema = z.enum(['invite', 'resume']) // Strict deployed cells reject new hello keys; mirror contract capabilities in upgrade headers. export const RELAY_HOST_CAPABILITY_HEADERS = { - 'x-orca-host-capabilities': 'pending-conn-details,idle-regional-rehome-v1' + 'x-orca-host-capabilities': 'pending-conn-details,idle-regional-rehome-v1,client-keepalive-v1' } as const +// Mirror relay-contract RELAY_CLIENT_KEEPALIVE_*: sent only after a relay's keepalive-offer. The +// relay answers it without waking, so the bytes must match exactly. +export const RELAY_CLIENT_KEEPALIVE_FRAME = '{"type":"keepalive"}' + // Mirror RELAY_PROTOCOL_LIMITS.hostAttachDeadlineMs, the phone's data-socket attach window. export const RELAY_HOST_ATTACH_DEADLINE_MS = 10_000 @@ -85,6 +89,18 @@ export const RelayPingMessageSchema = z .object({ type: z.literal('ping'), t: EpochMsSchema }) .strict() +export const RelayKeepaliveOfferMessageSchema = z + .object({ + type: z.literal('keepalive-offer'), + v: z.literal(1), + intervalMs: z.number().int().min(1_000).max(60_000) + }) + .strict() + +export const RelayKeepaliveAckMessageSchema = z + .object({ type: z.literal('keepalive-ack') }) + .strict() + export const RelayInviteCreatedMessageSchema = z .object({ type: z.literal('invite-created'),