relay edge: host-sent keepalive answered without waking the object (#40010)

Desktops advertise client-keepalive-v1; the edge offers it after the hello-ack and
auto-responds to a fixed keepalive frame, so idle hosts need no 25 s ping alarm
(each setAlarm is a billed row write). Silent keepalive hosts are probed with one
relay ping before a phone is refused, closed at once where the host is leaving.
Shipped desktops and cells see nothing new. Kill switch CLIENT_KEEPALIVE=0.
This commit is contained in:
Neil
2026-11-08 15:03:20 -08:00
committed by GitHub
parent 139e954009
commit cc7d19e932
16 changed files with 403 additions and 29 deletions
+8 -1
View File
@@ -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<object, EdgeConfig>()
@@ -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)
+49 -9
View File
@@ -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 '<continent>/<colo>', 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<Env> {
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<Env> {
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<Env> {
// ---------------- lifecycle ----------------
protected async onSocketGone(ws: WebSocket, code: number, reason: string): Promise<void> {
// `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<void> {
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<Env> {
const hostReason = relayHostCloseReasonFrom(reason)
await this.ctx.storage.put<Session>('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<Env> {
protected async reschedule(): Promise<void> {
let next = NO_DEADLINE
let conns: Record<string, Conn> | 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>('session')
if (session?.orphanedAt != null) { next = Math.min(next, session.orphanedAt + this.config.orphanGraceMs) }
@@ -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<void> {
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<void> {
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<void> {
@@ -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)
}
@@ -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()
}
@@ -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 }
}
+17 -4
View File
@@ -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<ReplayRefusal | { outcome: 'idle' } | { outcome: 'drained'; controls: number }> {
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
@@ -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<string, unknown>[], 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<string, unknown>[], 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))) {
@@ -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<void> {
}
}
// 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<void> {
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<void> {
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<void> {
const host = newHost()
+2 -2
View File
@@ -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<void> {
@@ -107,7 +107,7 @@ async function main(): Promise<void> {
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)
@@ -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
+1 -1
View File
@@ -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
@@ -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<string>((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',
@@ -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)
@@ -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()
})
})
@@ -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<typeof setInterval>; misses: number } | null = null
private silenceTimer: ReturnType<typeof setInterval> | null = null
private keepaliveTimer: ReturnType<typeof setInterval> | 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<string, unknown>): 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()
}
@@ -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'),