diff --git a/mobile/src/transport/mobile-direct-return-probe.test.ts b/mobile/src/transport/mobile-direct-return-probe.test.ts new file mode 100644 index 00000000000..8be5c00c805 --- /dev/null +++ b/mobile/src/transport/mobile-direct-return-probe.test.ts @@ -0,0 +1,93 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { DirectReturnProbe } from './mobile-direct-return-probe' +import { MobileEndpointHysteresis } from './mobile-endpoint-hysteresis' +import { FakeSession, host } from './mobile-endpoint-supervisor-test-fakes' + +vi.mock('react-native', () => ({ Platform: { OS: 'ios' } })) + +// A LAN that never answers: every dial sits open until the probe's own 12s budget. +function fixture() { + const opened: FakeSession[] = [] + const probe = new DirectReturnProbe( + { + now: Date.now, + setTimer: setTimeout, + clearTimer: clearTimeout, + openDirect: () => { + const candidate = new FakeSession('connecting') + opened.push(candidate) + return candidate + } + }, + { + hysteresis: new MobileEndpointHysteresis(Date.now(), { + directSuccessesRequired: 1, + directObservationMs: 60_000, + failureCooldownMs: 0, + minimumDwellMs: 0 + }), + host: () => host, + canSchedule: () => true, + canAttempt: () => true, + beginOperation: () => {}, + migrate: async () => {}, + onDirectMigrated: async () => {}, + afterProbe: () => {} + } + ) + return { opened, probe } +} + +beforeEach(() => vi.useFakeTimers()) +afterEach(() => vi.useRealTimers()) + +it('never opens a second dial while one is still in flight', async () => { + const { opened, probe } = fixture() + probe.schedule(0) + await vi.advanceTimersByTimeAsync(0) + expect(opened).toHaveLength(1) + + // A relay drop and a foreground return both ask for an immediate probe while the + // first dial is still awaiting authentication. + probe.schedule(0) + probe.schedule(0) + await vi.advanceTimersByTimeAsync(0) + expect(opened).toHaveLength(1) + + // Why this is the assertion that matters: a second probe would have overwritten + // activeProbe, so stop() would abort only the newest dial and leave this socket + // open for the rest of its 12s budget. + probe.stop() + await vi.advanceTimersByTimeAsync(0) + expect(opened[0]!.close).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) +}) + +it('honors an urgent reprobe asked for mid-dial instead of dropping it on the 15s floor', async () => { + const { opened, probe } = fixture() + probe.schedule(0) + await vi.advanceTimersByTimeAsync(0) + probe.schedule(0) + await vi.advanceTimersByTimeAsync(0) + expect(opened).toHaveLength(1) + + // The deferred ask survives the dial and runs at once when it settles, so holding + // the slot does not cost the caller the 15s it was trying to skip. + await vi.advanceTimersByTimeAsync(12_000) + await vi.advanceTimersByTimeAsync(1) + expect(opened).toHaveLength(2) + probe.stop() +}) + +it('falls back to the ordinary interval when nothing asked for a sooner probe', async () => { + const { opened, probe } = fixture() + probe.schedule(0) + await vi.advanceTimersByTimeAsync(12_000) + expect(opened).toHaveLength(1) + + await vi.advanceTimersByTimeAsync(14_999) + expect(opened).toHaveLength(1) + await vi.advanceTimersByTimeAsync(1) + expect(opened).toHaveLength(2) + probe.stop() +}) diff --git a/mobile/src/transport/mobile-direct-return-probe.ts b/mobile/src/transport/mobile-direct-return-probe.ts index 3ae31edd07f..7bb4d97ce7a 100644 --- a/mobile/src/transport/mobile-direct-return-probe.ts +++ b/mobile/src/transport/mobile-direct-return-probe.ts @@ -13,6 +13,8 @@ export class DirectReturnProbe { private stopped = false private activeProbe: AbortController | null = null + // Soonest delay a caller asked for while a dial was in flight. + private deferredDelayMs: number | null = null constructor( private readonly deps: { @@ -26,6 +28,7 @@ export class DirectReturnProbe { host: () => HostProfile canSchedule: () => boolean canAttempt: () => boolean + // Takes the supervisor's operation mutex, now held for the cutover only. beginOperation: () => void migrate: ( client: RpcClient, @@ -38,7 +41,18 @@ export class DirectReturnProbe { ) {} schedule(delayMs = DIRECT_PROBE_INTERVAL_MS): void { - if (this.stopped || !this.hooks.canSchedule() || this.timer) { + if (this.stopped || !this.hooks.canSchedule()) { + return + } + // Why: the dial no longer holds the supervisor's mutex, so nothing else stops a + // second probe from overwriting activeProbe — stop() would then reach only the + // newest socket and leave the earlier one dialing for its full 12s budget. The + // in-flight probe owns the next slot and re-arms it on the soonest ask. + if (this.activeProbe) { + this.deferredDelayMs = Math.min(this.deferredDelayMs ?? delayMs, delayMs) + return + } + if (this.timer) { return } this.timer = this.deps.setTimer(() => { @@ -48,6 +62,7 @@ export class DirectReturnProbe { } clear(): void { + this.deferredDelayMs = null if (this.timer) { this.deps.clearTimer(this.timer) this.timer = null @@ -70,9 +85,12 @@ export class DirectReturnProbe { } const controller = new AbortController() this.activeProbe = controller - this.hooks.beginOperation() + let owned = false let successful: Awaited> = null try { + // Why: the dial is a pure observation on its own socket — holding the + // supervisor's mutex across its 12s budget stalled every relay recovery + // that landed during a foreground return. Only the cutover needs the mutex. successful = await openAuthenticatedDirectEndpoint( this.hooks.host(), this.deps.openDirect, @@ -86,10 +104,18 @@ export class DirectReturnProbe { this.hooks.hysteresis.recordDirectFailure(this.deps.now()) return } + // Both early returns leave the candidate to the finally, which owns it until + // migration takes over — closing here too would double-close it. if (!this.hooks.hysteresis.recordDirectSuccess(this.deps.now())) { - successful.client.close() return } + if (!this.hooks.canAttempt()) { + // A relay dial owns the mutex; the streak survives, so the next probe + // promotes direct instead of this one. + return + } + this.hooks.beginOperation() + owned = true const candidate = successful // Migration owns the candidate, including closing it if cutover is canceled. successful = null @@ -109,10 +135,14 @@ export class DirectReturnProbe { } finally { this.activeProbe = null successful?.client.close() - // Why: a relay drop or backoff timer can arrive while the probe owns the + // Why: a relay drop or backoff timer can arrive while the cutover owns the // operation mutex; afterProbe releases it and replays deferred recovery. - this.hooks.afterProbe() - this.schedule() + if (owned) { + this.hooks.afterProbe() + } + const deferred = this.deferredDelayMs + this.deferredDelayMs = null + this.schedule(deferred ?? undefined) } } } diff --git a/mobile/src/transport/mobile-endpoint-lifecycle.ts b/mobile/src/transport/mobile-endpoint-lifecycle.ts index 7ec5f28b945..1542de9da7d 100644 --- a/mobile/src/transport/mobile-endpoint-lifecycle.ts +++ b/mobile/src/transport/mobile-endpoint-lifecycle.ts @@ -86,7 +86,7 @@ function createSupervisor( ): MobileEndpointSupervisor { return new MobileEndpointSupervisor(logical, host, { openDirect: (endpoint) => connect(endpoint, host.deviceToken, host.publicKeyB64, { onLog }), - openRelay: (relay, credential, confirmReqId, onHostCloseReason) => + openRelay: (relay, credential, confirmReqId, onHostCloseReason, isForeground) => connectMobileRelayRpcSession({ relay, resumeToken: credential.token, @@ -94,6 +94,7 @@ function createSupervisor( resumeConfirmReqId: confirmReqId, deviceToken: host.deviceToken, desktopPublicKeyB64: host.publicKeyB64, + isForeground, onHostCloseReason, onLog }), diff --git a/mobile/src/transport/mobile-endpoint-supervisor-contract.ts b/mobile/src/transport/mobile-endpoint-supervisor-contract.ts index 2a784fd8895..29ec807e649 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-contract.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-contract.ts @@ -12,7 +12,9 @@ export type MobileEndpointSupervisorDependencies = { relay: MobileRelayEndpoint, credential: { token: string; version: number }, confirmReqId: string, - onHostCloseReason?: (reason: RelayHostCloseReason) => void + onHostCloseReason?: (reason: RelayHostCloseReason) => void, + // Gates the session's idle liveness sweep; a backgrounded app spends no probes. + isForeground?: () => boolean ) => MobileRelayRpcSession resolveRelay: typeof resolveMobileRelayEndpoint readBundle: (hostId: string) => Promise diff --git a/mobile/src/transport/mobile-endpoint-supervisor-direct-probe.test.ts b/mobile/src/transport/mobile-endpoint-supervisor-direct-probe.test.ts index 3ee52fc7ddf..0e8f32ee5e3 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-direct-probe.test.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-direct-probe.test.ts @@ -1,5 +1,6 @@ import { beforeEach, afterEach, describe, expect, it, vi } from 'vitest' import { MobileEndpointSupervisor } from './mobile-endpoint-supervisor' +import { MobileEndpointHysteresis } from './mobile-endpoint-hysteresis' import { dependencies, FakeLogicalClient, @@ -8,6 +9,17 @@ import { host } from './mobile-endpoint-supervisor-test-fakes' +// A cell that authenticates and then answers the confirm for a different relay host +// — what a rehomed desktop produces. The session fails after the logical cutover. +function confirmRejectingRelaySession(logical: FakeLogicalClient): FakeRelaySession { + const session = new FakeRelaySession('connected', new Error('relay resume confirmation missing')) + session.whenResumeConfirmed = async () => { + session.publishState('disconnected') + logical.publishState('disconnected') + } + return session +} + vi.mock('react-native', () => ({ Platform: { OS: 'ios' } })) vi.mock('expo-secure-store', () => ({ WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'when-unlocked' })) vi.mock('expo-crypto', () => ({ getRandomBytes: (length: number) => new Uint8Array(length) })) @@ -48,4 +60,98 @@ describe('mobile endpoint supervisor direct probe', () => { expect(logical.getActivePath()).toBe('relay') supervisor.stop() }) + + it('recovers the relay at once while the probe is still dialing direct', async () => { + const logical = new FakeLogicalClient('connected', 'relay') + // A black-holed LAN endpoint: the dial sits unanswered for its whole 12s budget. + const direct = new FakeSession('connecting') + const openRelay = vi.fn(() => new FakeRelaySession('connected')) + const deps = dependencies({ openDirect: vi.fn(() => direct), openRelay }) + const supervisor = new MobileEndpointSupervisor(logical, host, deps) + await supervisor.start() + + await vi.advanceTimersByTimeAsync(15_000) + expect(deps.openDirect).toHaveBeenCalledOnce() + logical.publishState('disconnected') + await vi.advanceTimersByTimeAsync(0) + + // Why: the dial is a pure observation, so it no longer owns the operation + // mutex — recovery does not wait out the probe's budget. + expect(openRelay).toHaveBeenCalledOnce() + expect(logical.getState()).toBe('connected') + expect(logical.getActivePath()).toBe('relay') + supervisor.stop() + }) + + it('backs off a dial whose resume confirm fails after the cutover', async () => { + const recordMigration = vi.spyOn(MobileEndpointHysteresis.prototype, 'recordMigration') + const logical = new FakeLogicalClient('disconnected', 'lan') + const openRelay = vi.fn(() => confirmRejectingRelaySession(logical)) + const deps = dependencies({ openRelay, randomBytes: () => new Uint8Array([128, 0]) }) + const supervisor = new MobileEndpointSupervisor(logical, host, deps) + + await supervisor.start() + // Two sockets per pass: a confirm mismatch reads as a stale cell assignment, so + // the existing director fallback re-resolves and dials the authoritative target. + expect(openRelay).toHaveBeenCalledTimes(2) + expect(logical.migrateTo).toHaveBeenCalledTimes(2) + + // Why: `connected` is published at authentication, so the cutover happens before + // the confirm answers. A confirm that then fails must still book the shared + // cooldown — reporting it as an established dial redials in a tight loop. + await vi.advanceTimersByTimeAsync(0) + expect(openRelay).toHaveBeenCalledTimes(2) + + // 250ms, then 500ms, then 1000ms: the streak grows instead of resetting, which + // it could not do if setActiveSession had run for this dying session. + await vi.advanceTimersByTimeAsync(249) + expect(openRelay).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(1) + expect(openRelay).toHaveBeenCalledTimes(4) + await vi.advanceTimersByTimeAsync(250) + expect(openRelay).toHaveBeenCalledTimes(4) + await vi.advanceTimersByTimeAsync(250) + expect(openRelay).toHaveBeenCalledTimes(6) + await vi.advanceTimersByTimeAsync(999) + expect(openRelay).toHaveBeenCalledTimes(6) + await vi.advanceTimersByTimeAsync(1) + expect(openRelay).toHaveBeenCalledTimes(8) + + // No session whose confirm failed is ever booked as a migration. + expect(recordMigration).not.toHaveBeenCalled() + supervisor.stop() + }) + + it('replays a relay recovery that landed while the direct cutover owned the mutex', async () => { + const logical = new FakeLogicalClient('connected', 'relay') + const openRelay = vi.fn(() => new FakeRelaySession('connected')) + const deps = dependencies({ openDirect: vi.fn(() => new FakeSession('connected')), openRelay }) + const supervisor = new MobileEndpointSupervisor(logical, host, deps) + await supervisor.start() + + let release!: () => void + const cutover = new Promise((resolve) => { + release = resolve + }) + // The candidate loses the cutover, so the logical client stays on the relay path. + logical.migrateTo.mockImplementationOnce(async (candidate) => { + await cutover + candidate.close() + }) + // Three authenticated probes plus the observation and dwell windows. + await vi.advanceTimersByTimeAsync(60_000) + expect(logical.migrateTo).toHaveBeenCalledOnce() + + logical.publishState('disconnected') + await vi.advanceTimersByTimeAsync(0) + expect(openRelay).not.toHaveBeenCalled() + + release() + await vi.advanceTimersByTimeAsync(0) + + // The queued request is replayed by afterProbe, never dropped. + expect(openRelay).toHaveBeenCalledOnce() + expect(logical.getState()).toBe('connected') + supervisor.stop() + }) }) diff --git a/mobile/src/transport/mobile-endpoint-supervisor-test-fakes.ts b/mobile/src/transport/mobile-endpoint-supervisor-test-fakes.ts index 80f4438c160..cc4d91ea9da 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-test-fakes.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-test-fakes.ts @@ -65,6 +65,7 @@ export class FakeRelaySession extends FakeSession implements MobileRelayRpcSessi renewed: this.renewed, resumeExpiresAt: this.resumeExpiry }) + whenResumeConfirmed = () => Promise.resolve() getFailure = () => this.failure } diff --git a/mobile/src/transport/mobile-endpoint-supervisor.test.ts b/mobile/src/transport/mobile-endpoint-supervisor.test.ts index 10ef892a479..028387d8232 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor.test.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor.test.ts @@ -189,6 +189,7 @@ describe('mobile endpoint supervisor', () => { resolved, expect.any(Object), expect.any(String), + expect.any(Function), expect.any(Function) ) expect(deps.saveHost).toHaveBeenCalledWith( @@ -562,6 +563,7 @@ describe('mobile endpoint supervisor', () => { relay, expect.objectContaining({ version: 3 }), expect.any(String), + expect.any(Function), expect.any(Function) ) supervisor.stop() @@ -610,6 +612,7 @@ describe('mobile endpoint supervisor', () => { relay, expect.objectContaining({ version: 3 }), expect.any(String), + expect.any(Function), expect.any(Function) ) supervisor.stop() diff --git a/mobile/src/transport/mobile-endpoint-supervisor.ts b/mobile/src/transport/mobile-endpoint-supervisor.ts index 9ba12f35112..372fd7372a2 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor.ts @@ -16,6 +16,7 @@ import { } from './mobile-relay-credential-rotation' import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle' import { MobileEndpointNudgeRouter } from './mobile-endpoint-nudge-router' +import { RelayRecoveryIntentQueue } from './relay-recovery-intent-queue' import { MobileRelayDirectGraceTimer } from './mobile-relay-direct-grace-timer' import { MobileRelaySessionEstablisher } from './mobile-relay-session-establisher' import * as recoveryPresentation from './mobile-relay-recovery-presentation' @@ -38,7 +39,7 @@ export class MobileEndpointSupervisor { private bundle: MobileRelayCredentialBundle | null = null private stopped = false private operationInFlight = false - private pendingReplace = false + private readonly pending = new RelayRecoveryIntentQueue() private readonly nudgeRouter: MobileEndpointNudgeRouter private credentialRotationInFlight = false private relayRotationPending = false @@ -128,11 +129,8 @@ export class MobileEndpointSupervisor { }, afterProbe: () => { this.operationInFlight = false - if ( - this.pendingReplace || - this.relayRotationPending || - this.logical.getState() !== 'connected' - ) { + const queued = this.pending.takeRecovery() || this.pending.hasReplacement() + if (queued || this.relayRotationPending || this.logical.getState() !== 'connected') { void this.recoverRelay(this.relayRotationPending) } } @@ -195,6 +193,7 @@ export class MobileEndpointSupervisor { stop(): void { this.stopped = true + this.pending.clear() this.directProbe.stop() this.unsubscribeState?.() this.unsubscribeState = null @@ -215,13 +214,14 @@ export class MobileEndpointSupervisor { return } if (this.operationInFlight) { - // Why: a 12s direct probe can own the mutex when a network handoff lands; - // afterProbe replays the queued replacement so the signal is never lost. - this.pendingReplace ||= forceReplacement && ownsRecovery + // Why: a direct cutover or a slow post-migration write can own the mutex when + // a handoff lands. Every request is queued — an owning replacement keeps its + // force/owns intent, anything else replays as a plain recovery — so the + // holder's release replays it instead of dropping it. + this.pending.queue(forceReplacement, ownsRecovery) return } - if (this.pendingReplace) { - this.pendingReplace = false + if (this.pending.takeReplacement()) { forceReplacement = true ownsRecovery = true } @@ -236,7 +236,7 @@ export class MobileEndpointSupervisor { if (ownsRecovery) { // Why: never tear down a session no dial has disproven — the intent stays // queued so the armed retry runs forced once the cooldown lapses. - this.pendingReplace = true + this.pending.holdReplacement() } this.logRelay('recovery deferred by cooldown or gate') return @@ -260,7 +260,7 @@ export class MobileEndpointSupervisor { if (ownsRecovery) { // Why: no dial happened — keep the session and the intent; the reprobe // runs forced and replaces make-before-break once a credential exists. - this.pendingReplace = true + this.pending.holdReplacement() } return } @@ -273,7 +273,7 @@ export class MobileEndpointSupervisor { const dialed = await this.sessionEstablisher.dialEligible(selection.credentials) if (dialed.outcome === 'established') { // Why: a fresh socket satisfies any replacement intent queued mid-dial. - this.pendingReplace = false + this.pending.clearReplacement() retryAfterOperation = this.logical.getState() !== 'connected' return } @@ -293,11 +293,12 @@ export class MobileEndpointSupervisor { } } finally { this.operationInFlight = false + const queued = this.pending.takeRecovery() if (forceReplacement && this.relayRotationPending && this.isActive()) { this.leaseRotation.armRetry(this.relayReconnect.retryDelayMs(5000)) } // Why: the active relay can drop while migration follow-up still owns the mutex. - if (retryAfterOperation && this.isActive()) { + if ((retryAfterOperation || queued) && this.isActive()) { void this.recoverRelay() } } diff --git a/mobile/src/transport/mobile-relay-credential-rotation.ts b/mobile/src/transport/mobile-relay-credential-rotation.ts index 9b8a038e8e4..ef2630c8a67 100644 --- a/mobile/src/transport/mobile-relay-credential-rotation.ts +++ b/mobile/src/transport/mobile-relay-credential-rotation.ts @@ -142,11 +142,15 @@ export async function persistResumeConfirmation(args: { session: { getResumeConfirmation(): DeviceResumeConfirmed | null getResumeExpiresAt(): number | null + whenResumeConfirmed(): Promise } bundle: MobileRelayCredentialBundle usedCredentialVersion: number writeBundle: (bundle: MobileRelayCredentialBundle) => Promise }): Promise<{ bundle: MobileRelayCredentialBundle; leaseExpiry: number | null }> { + // Why: 'connected' is published at E2EE authentication now, so the confirm round + // trip can still be in flight here — its answer is what makes the bundle durable. + await args.session.whenResumeConfirmed() const confirmation = args.session.getResumeConfirmation() let bundle = args.bundle if (confirmation) { diff --git a/mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts b/mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts index b811721e562..fb8d2b5ffea 100644 --- a/mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts @@ -32,7 +32,10 @@ const relay = { e2eeFraming: 2 as const } -async function authenticateSession(onLog?: ConnectionLogSink) { +async function authenticateSession( + onLog?: ConnectionLogSink, + isForeground: () => boolean = () => true +) { const session = connectMobileRelayRpcSession({ relay, resumeToken: 'resume-secret', @@ -41,6 +44,7 @@ async function authenticateSession(onLog?: ConnectionLogSink) { deviceToken: 'device-token', desktopPublicKeyB64: 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=', requestTimeoutMs: 30_000, + isForeground, onLog }) fakes.linkOptions!.onHello({ @@ -52,12 +56,12 @@ async function authenticateSession(onLog?: ConnectionLogSink) { acceptedAs: 'current', resumeExpiresAt: Date.now() + 300_000 }) + // Authentication publishes 'connected' and puts both advisories on the wire. fakes.linkOptions!.onAuthenticated() - await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) - const confirmation = sentRequests()[0]! + const [confirmation, capabilities] = sentRequests() fakes.linkOptions!.onText( JSON.stringify({ - id: confirmation.id, + id: confirmation!.id, ok: true, result: { v: 1, @@ -74,17 +78,16 @@ async function authenticateSession(onLog?: ConnectionLogSink) { _meta: { runtimeId: 'runtime-1' } }) ) - await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledTimes(2)) - const capabilities = sentRequests()[1]! fakes.linkOptions!.onText( JSON.stringify({ - id: capabilities.id, + id: capabilities!.id, ok: true, result: {}, _meta: { runtimeId: 'runtime-1' } }) ) - await vi.waitFor(() => expect(session.getState()).toBe('connected')) + await session.whenResumeConfirmed() + expect(session.getState()).toBe('connected') fakes.sendText.mockClear() return session } @@ -95,6 +98,13 @@ function sentRequests(): Array<{ id: string; method: string }> { ) } +function answerProbe(): void { + const probe = sentRequests().at(-1)! + fakes.linkOptions!.onText( + JSON.stringify({ id: probe.id, ok: true, result: {}, _meta: { runtimeId: 'r1' } }) + ) +} + describe('mobile relay RPC session liveness', () => { beforeEach(() => { vi.useFakeTimers() @@ -104,16 +114,66 @@ describe('mobile relay RPC session liveness', () => { }) afterEach(() => vi.useRealTimers()) - it('sends no periodic traffic while an authenticated relay is idle', async () => { + it('sweeps an idle foregrounded relay once per idle interval', async () => { const session = await authenticateSession() - await vi.advanceTimersByTimeAsync(60_000) + await vi.advanceTimersByTimeAsync(24_999) + expect(fakes.sendText).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(1) + expect(sentRequests().map(({ method }) => method)).toEqual(['status.get']) + answerProbe() + + // Inbound traffic re-arms the sweep rather than stacking probes on it. + await vi.advanceTimersByTimeAsync(24_999) + expect(fakes.sendText).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(1) + expect(fakes.sendText).toHaveBeenCalledTimes(2) + expect(session.getState()).toBe('connected') + session.close() + }) + + it('spends no idle probe while the app is backgrounded', async () => { + let foreground = true + const session = await authenticateSession(undefined, () => foreground) + foreground = false + + await vi.advanceTimersByTimeAsync(120_000) expect(fakes.sendText).not.toHaveBeenCalled() expect(session.getState()).toBe('connected') + + // The resume that follows probes at once instead of waiting out the sweep. + foreground = true + session.notifyForeground('app-resume') + expect(sentRequests().map(({ method }) => method)).toEqual(['status.get']) session.close() }) + it('terminates a relay whose socket died in the background on two 2s resume misses', async () => { + const onLog = vi.fn() + const session = await authenticateSession(onLog) + + session.notifyForeground('app-resume') + expect(fakes.sendText).toHaveBeenCalledOnce() + // Why: the first frame after a resume rides a cold radio, so one slow answer is + // tolerated — but the verdict still lands at 4s instead of the old 8s. + await vi.advanceTimersByTimeAsync(2_000) + expect(session.getState()).toBe('connected') + expect(fakes.sendText).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(1_999) + expect(session.getState()).toBe('connected') + await vi.advanceTimersByTimeAsync(1) + + expect(session.getState()).toBe('disconnected') + expect(fakes.close).toHaveBeenCalledOnce() + expect(onLog).toHaveBeenCalledWith( + expect.objectContaining({ + code: 'liveness-timeout', + detail: expect.stringMatching(/^probe-timeout; 2\/2 probes missed;/) + }) + ) + }) + it('disconnects after two fair foreground misses', async () => { const onLog = vi.fn() const session = await authenticateSession(onLog) @@ -161,22 +221,25 @@ describe('mobile relay RPC session liveness', () => { expect(secondId).not.toBe(firstId) }) - it('rate-limits foreground sequences without suppressing a retry', async () => { + it('rate-limits focus nudges but never an app resume', async () => { const session = await authenticateSession() session.notifyForeground('focus') - const firstProbe = sentRequests()[0]! - fakes.linkOptions!.onText( - JSON.stringify({ id: firstProbe.id, ok: true, result: {}, _meta: { runtimeId: 'r1' } }) - ) + answerProbe() session.notifyForeground('focus') await vi.advanceTimersByTimeAsync(9_999) - session.notifyForeground('app-resume') expect(fakes.sendText).toHaveBeenCalledOnce() - await vi.advanceTimersByTimeAsync(1) + + // The resume owns the only evidence that the suspended socket is still alive. + session.notifyForeground('app-resume') + expect(fakes.sendText).toHaveBeenCalledTimes(2) + answerProbe() + session.notifyForeground('focus') + expect(fakes.sendText).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(10_000) session.notifyForeground('focus') - expect(fakes.sendText).toHaveBeenCalledTimes(2) + expect(fakes.sendText).toHaveBeenCalledTimes(3) session.close() }) @@ -189,9 +252,9 @@ describe('mobile relay RPC session liveness', () => { session.close() }) - it('does not probe when work follows prolonged inbound silence', async () => { + it('does not probe when work follows inbound silence', async () => { const session = await authenticateSession() - await vi.advanceTimersByTimeAsync(60_000) + await vi.advanceTimersByTimeAsync(20_000) const pending = session.sendRequest('terminal.send', { terminal: 'term', text: 'hi' }) const outcome = pending.catch(() => undefined) diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index 4bf617faf50..05ffce0f1e8 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -22,6 +22,10 @@ const fakes = vi.hoisted(() => ({ close: vi.fn() })) +vi.mock('react-native', () => ({ Platform: { OS: 'ios' } })) +vi.mock('expo-secure-store', () => ({ WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'when-unlocked' })) +vi.mock('expo-crypto', () => ({ getRandomBytes: (length: number) => new Uint8Array(length) })) + vi.mock('./mobile-relay-e2ee-link', () => ({ MobileRelayE2eeLink: class { constructor(options: NonNullable) { @@ -33,6 +37,8 @@ vi.mock('./mobile-relay-e2ee-link', () => ({ })) import { connectMobileRelayRpcSession } from './mobile-relay-rpc-session' +import { persistResumeConfirmation } from './mobile-relay-credential-rotation' +import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle' const relay = { v: 1 as const, @@ -43,6 +49,13 @@ const relay = { e2eeFraming: 2 as const } +type SentRequest = { + id: string + method: string + deviceToken: string + params: Record | undefined +} + function openSession() { return connectMobileRelayRpcSession({ relay, @@ -55,8 +68,11 @@ function openSession() { }) } -async function confirmResume() { - const session = openSession() +function sentRequests(): SentRequest[] { + return fakes.sendText.mock.calls.map(([value]) => JSON.parse(value as string) as SentRequest) +} + +function receiveHello(): void { fakes.linkOptions!.onHello({ type: 'relay-hello', ok: true, @@ -66,21 +82,31 @@ async function confirmResume() { acceptedAs: 'current', resumeExpiresAt: Date.now() + 300_000 }) +} + +// E2EE authentication alone publishes 'connected'; the confirm and the capability +// advisory are already on the wire by the time it returns. +function authenticateSession() { + const session = openSession() + receiveHello() expect(session.getState()).toBe('handshaking') fakes.linkOptions!.onAuthenticated() - await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) - const request = JSON.parse(fakes.sendText.mock.calls[0]![0] as string) as { - id: string - method: string - params: unknown + const [confirmationRequest, capabilityRequest] = sentRequests() + return { + session, + confirmationRequest: confirmationRequest!, + capabilityRequest: capabilityRequest! } +} + +function answerConfirm(request: SentRequest, relayHostId = relay.relayHostId): void { fakes.linkOptions!.onText( JSON.stringify({ id: request.id, ok: true, result: { v: 1, - relay, + relay: { ...relay, relayHostId }, resumeConfirmation: { v: 1, reqId: 'confirm-1', @@ -93,39 +119,32 @@ async function confirmResume() { _meta: { runtimeId: 'runtime-1' } }) ) - await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledTimes(2)) - const capabilityRequest = JSON.parse(fakes.sendText.mock.calls[1]![0] as string) as { - id: string - method: string - deviceToken: string - params: { clientCapabilities?: string[] } - } - return { session, confirmationRequest: request, capabilityRequest } } -async function authenticateSession(capabilitySupported = true) { - const { session, confirmationRequest, capabilityRequest } = await confirmResume() - expect(session.getState()).toBe('handshaking') +function answerCapability(request: SentRequest, supported = true): void { fakes.linkOptions!.onText( JSON.stringify( - capabilitySupported - ? { - id: capabilityRequest.id, - ok: true, - result: capabilityRequest.params, - _meta: { runtimeId: 'runtime-1' } - } + supported + ? { id: request.id, ok: true, result: request.params, _meta: { runtimeId: 'runtime-1' } } : { - id: capabilityRequest.id, + id: request.id, ok: false, error: { code: 'method_not_found', message: 'Unknown method' }, _meta: { runtimeId: 'runtime-1' } } ) ) - await vi.waitFor(() => expect(session.getState()).toBe('connected')) +} + +// Both advisories answered and the send log cleared, so a test can read its own frames. +async function settledSession(capabilitySupported = true) { + const authenticated = authenticateSession() + answerConfirm(authenticated.confirmationRequest) + answerCapability(authenticated.capabilityRequest, capabilitySupported) + await authenticated.session.whenResumeConfirmed() + expect(authenticated.session.getState()).toBe('connected') fakes.sendText.mockClear() - return { session, confirmationRequest, capabilityRequest } + return authenticated } describe('mobile relay RPC session', () => { @@ -137,7 +156,7 @@ describe('mobile relay RPC session', () => { afterEach(() => vi.useRealTimers()) it('releases stream listeners on failure even when close follows it', async () => { - const { session } = await authenticateSession() + const { session } = await settledSession() const listener = vi.fn() session.subscribe('runtime.clientEvents.subscribe', {}, listener) await Promise.resolve() @@ -166,8 +185,8 @@ describe('mobile relay RPC session', () => { expect(listener).toHaveBeenCalledTimes(1) }) - it('requires exact resume observations and confirms by request ID before becoming connected', async () => { - const { session, confirmationRequest, capabilityRequest } = await authenticateSession() + it('sends the resume confirm by request ID and the capability advisory concurrently', async () => { + const { session, confirmationRequest, capabilityRequest } = await settledSession() expect(fakes.linkOptions).toMatchObject({ endpoint: relay, @@ -192,21 +211,103 @@ describe('mobile relay RPC session', () => { }) it('connects when an older runtime rejects capability negotiation', async () => { - const { session } = await authenticateSession(false) + const { session } = await settledSession(false) expect(session.getState()).toBe('connected') expect(session.getFailure()).toBeNull() }) it('connects when the relay never answers capability negotiation', async () => { - const { session } = await confirmResume() + const { session, confirmationRequest } = authenticateSession() + answerConfirm(confirmationRequest) - // Why: the advisory's own deadline used to fail confirmResume, so a link too slow to + // Why: the advisory's own deadline used to fail the confirm, so a link too slow to // answer within the request timeout never published 'connected' — it just redialled. - await vi.waitFor(() => expect(session.getState()).toBe('connected'), { timeout: 5_000 }) + await session.whenResumeConfirmed() + expect(session.getState()).toBe('connected') expect(session.getFailure()).toBeNull() }) + it('publishes connected at authentication, ahead of the confirm answer', async () => { + const states: string[] = [] + const session = openSession() + session.onStateChange((state) => states.push(state)) + receiveHello() + fakes.linkOptions!.onAuthenticated() + + // Why: the transport carries traffic from here; two serialized advisory round + // trips used to add ~200ms to every phone reconnect before anything rendered. + expect(session.getState()).toBe('connected') + expect(states).toEqual(['handshaking', 'connected']) + expect(session.getResumeConfirmation()).toBeNull() + expect(sentRequests().map(({ method }) => method)).toEqual([ + 'pairing.getEndpoints', + 'runtime.clientCapabilities.update' + ]) + + const [confirmationRequest] = sentRequests() + answerConfirm(confirmationRequest!) + await session.whenResumeConfirmed() + expect(session.getResumeConfirmation()).toMatchObject({ reqId: 'confirm-1' }) + session.close() + }) + + it('fails a session whose confirm answers for another relay host after connected', async () => { + const { session, confirmationRequest } = authenticateSession() + expect(session.getState()).toBe('connected') + + answerConfirm(confirmationRequest, 'ZZZZZZZZZZZZZZZZ') + await session.whenResumeConfirmed() + + // A late failure is fine; a lost one is not. + expect(session.getState()).toBe('disconnected') + expect(session.getFailure()?.message).toBe('relay resume confirmation missing') + expect(fakes.close).toHaveBeenCalledOnce() + }) + + it('fails a session whose confirm never answers', async () => { + vi.useFakeTimers() + try { + const { session } = authenticateSession() + expect(session.getState()).toBe('connected') + + await vi.advanceTimersByTimeAsync(1_000) + + expect(session.getState()).toBe('disconnected') + expect(session.getFailure()?.message).toBe('relay RPC timed out: pairing.getEndpoints') + } finally { + vi.useRealTimers() + } + }) + + it('hands the landed confirmation to resume persistence', async () => { + const { session, confirmationRequest } = authenticateSession() + const bundle: MobileRelayCredentialBundle = { + v: 1, + hostId: 'host-1', + deviceToken: 'device-token', + current: { token: 'A'.repeat(43), hash: 'B'.repeat(43), version: 3, expiresAt: 1 } + } + const writeBundle = vi.fn(async () => {}) + // Why: persistence runs right after the migration, while the confirm is still + // in flight — it must wait for the answer instead of reading a null. + const persisting = persistResumeConfirmation({ + session, + bundle, + usedCredentialVersion: 3, + writeBundle + }) + expect(writeBundle).not.toHaveBeenCalled() + + answerConfirm(confirmationRequest) + const applied = await persisting + + expect(writeBundle).toHaveBeenCalledOnce() + expect(applied.bundle.current.expiresAt).toBe(session.getResumeExpiresAt()) + expect(applied.leaseExpiry).toBe(session.getResumeExpiresAt()) + session.close() + }) + // Why: ConnectionState stays 'connecting' until relay-hello, so the migration bound // needs a separate signal to tell "cell never answered the upgrade" from "cell took // relay-auth and is still resolving the assignment". @@ -231,7 +332,7 @@ describe('mobile relay RPC session', () => { expect(session.getDialStage()).toBe('handshaking') fakes.linkOptions!.onAuthenticated() expect(session.getDialStage()).toBe('confirming') - await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) + expect(fakes.sendText).toHaveBeenCalledTimes(2) expect(stages).toEqual(['awaiting-hello', 'handshaking', 'confirming']) session.close() }) @@ -254,7 +355,7 @@ describe('mobile relay RPC session', () => { }) it('routes terminal and browser binary streams after confirmation', async () => { - const { session } = await authenticateSession() + const { session } = await settledSession() const terminalListener = vi.fn() session.subscribe('terminal.subscribe', { terminal: 'term-1' }, terminalListener) await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) @@ -311,7 +412,7 @@ describe('mobile relay RPC session', () => { }) it('rejects pending RPC work when the physical link fails', async () => { - const { session } = await authenticateSession() + const { session } = await settledSession() const pending = session.sendRequest('status.get') await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) fakes.linkOptions!.onError(new Error('relay transport error')) @@ -323,7 +424,7 @@ describe('mobile relay RPC session', () => { }) it('marks in-flight requests delivery-unknown when the session closes', async () => { - const { session } = await authenticateSession() + const { session } = await settledSession() const pending = session.sendRequest('terminal.send', { terminal: 'term', text: 'hi' }) await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) session.close() @@ -333,7 +434,7 @@ describe('mobile relay RPC session', () => { }) it('marks a relay RPC timeout delivery-unknown', async () => { - const { session } = await authenticateSession() + const { session } = await settledSession() vi.useFakeTimers() try { const pending = session.sendRequest('terminal.send', { terminal: 'term', text: 'hi' }) @@ -352,4 +453,57 @@ describe('mobile relay RPC session', () => { vi.useRealTimers() } }) + it('keeps whenResumeConfirmed() pending until the session has an answer', async () => { + // The contract callers rely on is "settles when the confirm has answered or the + // session is over". A promise already resolved during the dial would let a caller + // read getResumeConfirmation() as null and persist that as the answer. + const session = openSession() + const settled = vi.fn() + void session.whenResumeConfirmed().then(settled) + receiveHello() + await Promise.resolve() + expect(settled).not.toHaveBeenCalled() + + fakes.linkOptions!.onAuthenticated() + await Promise.resolve() + expect(settled).not.toHaveBeenCalled() + + answerConfirm(sentRequests()[0]!) + await session.whenResumeConfirmed() + expect(settled).toHaveBeenCalled() + expect(session.getResumeConfirmation()).toMatchObject({ reqId: 'confirm-1' }) + }) + + it('settles whenResumeConfirmed() when the session dies before authenticating', async () => { + const session = openSession() + const settled = vi.fn() + void session.whenResumeConfirmed().then(settled) + + // A credential-version mismatch fails the session inside onHello, so no confirm + // is ever sent. Awaiting the answer must not hang a caller forever. + fakes.linkOptions!.onHello({ + type: 'relay-hello', + ok: true, + credentialKind: 'resume', + leaseExpiresAt: Date.now() + 60_000, + acceptedCredentialVersion: 2, + acceptedAs: 'current', + resumeExpiresAt: Date.now() + 300_000 + }) + + await session.whenResumeConfirmed() + expect(settled).toHaveBeenCalled() + expect(session.getState()).toBe('disconnected') + }) + + it('settles whenResumeConfirmed() when a caller closes an unconfirmed session', async () => { + const session = openSession() + const settled = vi.fn() + void session.whenResumeConfirmed().then(settled) + + session.close() + + await session.whenResumeConfirmed() + expect(settled).toHaveBeenCalled() + }) }) diff --git a/mobile/src/transport/mobile-relay-rpc-session.ts b/mobile/src/transport/mobile-relay-rpc-session.ts index 67b50ea591e..8108f21d021 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -17,9 +17,17 @@ import type { RelayHostCloseReason } from '../../../src/shared/relay-host-close- import type { RpcClient } from './rpc-client' import type { ConnectionLogSink, ConnectionState, RpcResponse } from './types' -const RELAY_PROBE_TIMEOUT_MS = 4_000 -const RELAY_MISSED_PROBE_LIMIT = 2 -const RELAY_FOREGROUND_PROBE_MIN_INTERVAL_MS = 10_000 +// Ordinary foreground checks: two 4s misses, at most one voluntary probe per 10s. +const RELAY_PROBE = { timeoutMs: 4_000, missedProbeLimit: 2, minIntervalMs: 10_000 } +// A socket that died while the process was suspended must be admitted before the +// user reads the screen as broken. Two 2s misses, not one: the first frame after a +// resume rides a cold radio, and a single slow answer is not proof of a dead link. +const RELAY_RESUME_PROBE = { timeoutMs: 2_000, missedProbeLimit: 2 } +// Bounds the confirm exactly as migrateTo's own wait used to, so the supervisor's +// mutex is never held for the full request timeout waiting on a silent cell. +const RELAY_CONFIRM_TIMEOUT_MS = 12_000 +// Foreground-only sweep so a silently-dead relay surfaces without a user action. +const RELAY_IDLE_PROBE_MS = 25_000 let relayRpcSessionSequence = 0 export type MobileRelayRpcSession = RpcClient & @@ -29,6 +37,10 @@ export type MobileRelayRpcSession = RpcClient & getAttachDeadlineAt(): number | null getResumeExpiresAt(): number | null getResumeConfirmation(): DeviceResumeConfirmed | null + // Settles once the resume confirm has answered or failed the session. Never + // rejects. Anyone reading getResumeConfirmation()/getResumeExpiresAt() must + // await it: 'connected' is published at authentication, ahead of the confirm. + whenResumeConfirmed(): Promise getFailure(): Error | null } @@ -40,6 +52,8 @@ export function connectMobileRelayRpcSession(args: { deviceToken: string desktopPublicKeyB64: string requestTimeoutMs?: number + // Gates the idle liveness sweep; a backgrounded app must not spend probes. + isForeground?: () => boolean createSocket?: (url: string) => WebSocket onHostCloseReason?: (reason: RelayHostCloseReason) => void onLog?: ConnectionLogSink @@ -57,6 +71,14 @@ export function connectMobileRelayRpcSession(args: { let logSequence = 0 const logSessionId = `${Date.now().toString(36)}-${(++relayRpcSessionSequence).toString(36)}` const livenessIdentity = {} + // Why created here and not at authentication: handing a pre-auth caller an + // already-resolved promise would let it read getResumeConfirmation() as null and + // treat that as the answer. Every terminal path settles it — the confirm, fail(), + // and close() — so awaiting it can never outlive the session. + let settleResumeConfirmed!: () => void + const resumeConfirmed = new Promise((resolve) => { + settleResumeConfirmed = resolve + }) const dialStage = new RelayDialStageTracker() const streams = new MobileRelayRpcStreams({ nextId: () => pending.nextId(), @@ -86,7 +108,7 @@ export function connectMobileRelayRpcSession(args: { dialStage.advance('handshaking') publishState('handshaking') }, - onAuthenticated: () => void confirmResume(), + onAuthenticated: () => publishAuthenticated(), onText: (plaintext) => { livenessWatchdog.noteAuthenticatedInbound(livenessIdentity) handleText(plaintext) @@ -125,33 +147,27 @@ export function connectMobileRelayRpcSession(args: { }, notifyForeground: (reason) => { if (state === 'connected' && reason !== 'network-change') { - livenessWatchdog.probeNow(livenessIdentity) + livenessWatchdog.probeNow(livenessIdentity, reason === 'app-resume' ? 'resume' : 'nudge') } }, - close() { - if (closed) { - return - } - closed = true - livenessWatchdog.stop(livenessIdentity) - link.close() - pending.rejectAll(new Error('Client closed')) - streams.clear() - publishState('disconnected') - }, + close: () => terminate(new Error('Client closed')), getDialStage: () => dialStage.getDialStage(), onDialStageChange: (listener) => dialStage.onDialStageChange(listener), getAttachDeadlineAt: () => attachDeadlineAt, getResumeExpiresAt: () => resumeExpiresAt, getResumeConfirmation: () => resumeConfirmation, + whenResumeConfirmed: () => resumeConfirmed, getFailure: () => failure } const livenessWatchdog = new RpcSessionLivenessWatchdog({ transport: 'relay', - idleProbeMs: null, - probeTimeoutMs: RELAY_PROBE_TIMEOUT_MS, - missedProbeLimit: RELAY_MISSED_PROBE_LIMIT, - voluntaryProbeMinIntervalMs: RELAY_FOREGROUND_PROBE_MIN_INTERVAL_MS, + idleProbeMs: RELAY_IDLE_PROBE_MS, + probeTimeoutMs: RELAY_PROBE.timeoutMs, + missedProbeLimit: RELAY_PROBE.missedProbeLimit, + voluntaryProbeMinIntervalMs: RELAY_PROBE.minIntervalMs, + urgentProbeTimeoutMs: RELAY_RESUME_PROBE.timeoutMs, + urgentMissedProbeLimit: RELAY_RESUME_PROBE.missedProbeLimit, + shouldIdleProbe: () => args.isForeground?.() ?? true, sendProbe: () => state === 'connected' && sendFrame({ id: pending.nextId(), method: 'status.get', params: undefined }), @@ -170,13 +186,33 @@ export function connectMobileRelayRpcSession(args: { }) return client - async function confirmResume(): Promise { + // Why: the transport carries traffic the moment E2EE authenticates. The resume + // confirm and the capability advisory ride it concurrently instead of putting + // two serialized round trips in front of 'connected'. + function publishAuthenticated(): void { + if (closed) { + return + } dialStage.advance('confirming') + void confirmResume().then(settleResumeConfirmed, settleResumeConfirmed) + // Why: an unanswered advisory says nothing, but a frame that never reached the + // wire proves the socket cannot carry traffic — that alone still fails. + void settleMobileRuntimeCapabilities((method, params) => + sendRpc(method, params, requestTimeoutMs, true) + ).catch((error: unknown) => fail(asError(error))) + lastConnectedAt = Date.now() + livenessWatchdog.start(livenessIdentity) + publishState('connected') + } + + // Off the critical path but never optional: a failed confirm or a relayHostId + // that is not ours still fails the session, only later than it used to. + async function confirmResume(): Promise { try { const response = await sendRpc( 'pairing.getEndpoints', { resumeConfirmReqId: args.resumeConfirmReqId }, - requestTimeoutMs, + Math.min(requestTimeoutMs, RELAY_CONFIRM_TIMEOUT_MS), true ) if (!response.ok) { @@ -188,13 +224,6 @@ export function connectMobileRelayRpcSession(args: { } resumeConfirmation = result.resumeConfirmation resumeExpiresAt = result.resumeConfirmation.resumeExpiresAt - lastConnectedAt = Date.now() - // Why: an unanswered advisory must not keep a slow relay from ever reaching connected. - await settleMobileRuntimeCapabilities((method, params) => - sendRpc(method, params, requestTimeoutMs, true) - ) - livenessWatchdog.start(livenessIdentity) - publishState('connected') } catch (error) { fail(asError(error)) } @@ -287,18 +316,28 @@ export function connectMobileRelayRpcSession(args: { } } - function fail(error: Error): void { + // One teardown for both endings; only whether the session is to blame differs, and + // recording a failure for a caller's close would make the establisher report a + // deliberate teardown as a dial error. + function terminate(error: Error): void { if (closed) { return } closed = true - failure = error + settleResumeConfirmed() livenessWatchdog.stop(livenessIdentity) streams.clear() link.close() pending.rejectAll(error) publishState(error instanceof MobileE2EEAuthenticationError ? 'auth-failed' : 'disconnected') } + + function fail(error: Error): void { + if (!closed) { + failure = error + } + terminate(error) + } } function asError(error: unknown): Error { diff --git a/mobile/src/transport/mobile-relay-runtime-failover.test.ts b/mobile/src/transport/mobile-relay-runtime-failover.test.ts index ce7cca3fd9f..7098746587a 100644 --- a/mobile/src/transport/mobile-relay-runtime-failover.test.ts +++ b/mobile/src/transport/mobile-relay-runtime-failover.test.ts @@ -88,6 +88,7 @@ class FakeRelaySession extends FakeSession implements MobileRelayRpcSession { this.dialStage.onDialStageChange(listener) getResumeExpiresAt = () => Date.now() + 30 * 24 * 3_600_000 getResumeConfirmation = () => null + whenResumeConfirmed = () => Promise.resolve() getFailure = () => this.failure } @@ -277,6 +278,7 @@ describe('relay runtime recovery without direct connectivity', () => { relay, expect.objectContaining({ version: 3 }), expect.any(String), + expect.any(Function), expect.any(Function) ) expect(logical.getActivePath()).toBe('relay') @@ -367,6 +369,7 @@ describe('relay runtime recovery without direct connectivity', () => { relay, expect.objectContaining({ version: 2 }), expect.any(String), + expect.any(Function), expect.any(Function) ) expect(logical.getActivePath()).toBe('relay') @@ -397,6 +400,7 @@ describe('relay runtime recovery without direct connectivity', () => { relay, expect.objectContaining({ version: 1 }), expect.any(String), + expect.any(Function), expect.any(Function) ) expect(logical.getActivePath()).toBe('relay') diff --git a/mobile/src/transport/mobile-relay-session-establisher.ts b/mobile/src/transport/mobile-relay-session-establisher.ts index 9a04ae44137..9ec8ebb3a37 100644 --- a/mobile/src/transport/mobile-relay-session-establisher.ts +++ b/mobile/src/transport/mobile-relay-session-establisher.ts @@ -110,7 +110,8 @@ export class MobileRelaySessionEstablisher { if (reason === RELAY_HOST_CLOSE_REASON.SIGNED_OUT) { args.logical.setHostSignedOut(true) } - } + }, + args.isForeground ) try { // Why: backgrounding or a direct winner withdraws this dial before cutover. @@ -126,6 +127,17 @@ export class MobileRelaySessionEstablisher { } return { ok: false, error: session.getFailure() ?? toError(error) } } + // Why: migrateTo now resolves at E2EE authentication, so the resume confirm can + // still fail this session after the cutover. Booking a dying session as an + // established dial skips backoff and redials in a tight loop — the supervisor's + // bookkeeping waits for the verdict even though the UI is already connected. + await session.whenResumeConfirmed() + if (session.getState() !== 'connected') { + if (!args.isActive() || directWon(args.logical)) { + return { ok: false, error: new RelayDialAbortedError() } + } + return { ok: false, error: session.getFailure() ?? new Error('relay lost at confirm') } + } args.controller.setActiveSession(session) if (!args.isForeground()) { args.controller.suspendActiveRelay(args.logical) diff --git a/mobile/src/transport/relay-recovery-intent-queue.ts b/mobile/src/transport/relay-recovery-intent-queue.ts new file mode 100644 index 00000000000..c34e40b8990 --- /dev/null +++ b/mobile/src/transport/relay-recovery-intent-queue.ts @@ -0,0 +1,45 @@ +// Recovery requests that arrive while the supervisor's operation mutex is held. +// Two latches, because the intents are not interchangeable: an owning forced +// replacement books the shared cooldown and may bring a stale session down, while +// every other request must replay as a plain recovery. Nothing is ever dropped. +export class RelayRecoveryIntentQueue { + private replacement = false + private recovery = false + + queue(forceReplacement: boolean, ownsRecovery: boolean): void { + if (forceReplacement && ownsRecovery) { + this.replacement = true + return + } + this.recovery = true + } + + holdReplacement(): void { + this.replacement = true + } + + hasReplacement(): boolean { + return this.replacement + } + + clearReplacement(): void { + this.replacement = false + } + + takeReplacement(): boolean { + const queued = this.replacement + this.replacement = false + return queued + } + + takeRecovery(): boolean { + const queued = this.recovery + this.recovery = false + return queued + } + + clear(): void { + this.replacement = false + this.recovery = false + } +} diff --git a/mobile/src/transport/rpc-session-liveness-watchdog.test.ts b/mobile/src/transport/rpc-session-liveness-watchdog.test.ts index aa1398e44b6..4b25aaa1fd5 100644 --- a/mobile/src/transport/rpc-session-liveness-watchdog.test.ts +++ b/mobile/src/transport/rpc-session-liveness-watchdog.test.ts @@ -173,4 +173,97 @@ describe('RpcSessionLivenessWatchdog', () => { watchdog.probeNow(identity) expect(terminate).toHaveBeenCalledWith(identity) }) + function backgroundableFixture() { + const sendProbe = vi.fn(() => true) + const terminate = vi.fn() + const identity = {} + const state = { foreground: true } + const watchdog = new RpcSessionLivenessWatchdog({ + transport: 'relay', + sendProbe, + terminate, + shouldIdleProbe: () => state.foreground, + now: Date.now + }) + watchdog.start(identity) + return { identity, sendProbe, state, terminate, watchdog } + } + + it('stops retrying an idle probe once the app backgrounds under it', async () => { + const { sendProbe, state, terminate } = backgroundableFixture() + await vi.advanceTimersByTimeAsync(LIVENESS_IDLE_MS) + expect(sendProbe).toHaveBeenCalledOnce() + + // iOS suspends the socket in the background, so every further miss is evidence + // about the app and not about the peer. Retrying would spend the whole budget on + // the suspension and terminate a relay that is fine. + state.foreground = false + await vi.advanceTimersByTimeAsync(LIVENESS_PROBE_TIMEOUT_MS * 4) + expect(sendProbe).toHaveBeenCalledOnce() + expect(terminate).not.toHaveBeenCalled() + }) + + it('re-arms the idle sweep with a clean slate after a backgrounded probe', async () => { + const { sendProbe, state, terminate } = backgroundableFixture() + await vi.advanceTimersByTimeAsync(LIVENESS_IDLE_MS) + state.foreground = false + await vi.advanceTimersByTimeAsync(LIVENESS_PROBE_TIMEOUT_MS) + state.foreground = true + + // The abandoned probe must not be carried forward as a miss: the sweep needs its + // full three fair misses again before it may call the session dead. + await vi.advanceTimersByTimeAsync(LIVENESS_IDLE_MS) + expect(sendProbe).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(LIVENESS_PROBE_TIMEOUT_MS * 2) + expect(terminate).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(LIVENESS_PROBE_TIMEOUT_MS) + expect(terminate).toHaveBeenCalledOnce() + }) + + it('gives a resume probe its own miss budget, not the one the ordinary probe spent', async () => { + // Why: the urgent profile exists to tolerate one slow answer from a cold radio. Inheriting + // an ordinary miss spends that tolerance before the resume probe is even sent, so the first + // slow answer on a healthy socket kills the session -- the case the profile was added for. + const terminate = vi.fn() + const sendProbe = vi.fn(() => true) + const identity = {} + const watchdog = new RpcSessionLivenessWatchdog({ + transport: 'relay', + idleProbeMs: 20_000, + probeTimeoutMs: 4_000, + missedProbeLimit: 2, + urgentProbeTimeoutMs: 2_000, + urgentMissedProbeLimit: 2, + shouldIdleProbe: () => true, + sendProbe, + terminate, + now: Date.now + }) + watchdog.start(identity) + + // One ordinary miss on the idle sweep, tolerated, and a second ordinary probe in flight. + await vi.advanceTimersByTimeAsync(20_000) + await vi.advanceTimersByTimeAsync(4_000) + expect(terminate).not.toHaveBeenCalled() + + // Foreground: the resume probe supersedes the ordinary one still in flight. + watchdog.probeNow(identity, 'resume') + await vi.advanceTimersByTimeAsync(2_000) + expect(terminate).not.toHaveBeenCalled() + + // The second urgent miss is the one that may terminate. + await vi.advanceTimersByTimeAsync(2_000) + expect(terminate).toHaveBeenCalledOnce() + }) + + it('still reaches a verdict on a caller probe when the app backgrounds', async () => { + // The gate covers the idle sweep only. A nudge or resume probe was asked for on + // purpose, and abandoning it would leave a genuinely dead socket unreported. + const { identity, state, terminate, watchdog } = backgroundableFixture() + watchdog.probeNow(identity) + state.foreground = false + + await vi.advanceTimersByTimeAsync(LIVENESS_PROBE_TIMEOUT_MS * 3) + expect(terminate).toHaveBeenCalledOnce() + }) }) diff --git a/mobile/src/transport/rpc-session-liveness-watchdog.ts b/mobile/src/transport/rpc-session-liveness-watchdog.ts index 36525f60fb0..e34c47bd67a 100644 --- a/mobile/src/transport/rpc-session-liveness-watchdog.ts +++ b/mobile/src/transport/rpc-session-liveness-watchdog.ts @@ -13,11 +13,19 @@ type WatchdogOptions = { probeTimeoutMs?: number missedProbeLimit?: number voluntaryProbeMinIntervalMs?: number + // Bounds for probeImmediately(); default to the ordinary probe bounds. + urgentProbeTimeoutMs?: number + urgentMissedProbeLimit?: number + // Gates the idle sweep only. False re-arms without probing — a backgrounded app + // must not spend a probe, and its resume probes immediately anyway. + shouldIdleProbe?: () => boolean now?: () => number setTimer?: typeof setTimeout clearTimer?: typeof clearTimeout } +type ProbeProfile = { timeoutMs: number; missedProbeLimit: number } + export type LivenessTimeoutEvidence = { transport: 'direct' | 'relay' reason: 'probe-send-failed' | 'probe-timeout' @@ -30,12 +38,15 @@ export class RpcSessionLivenessWatchdog { private identity: RpcSessionIdentity | null = null private timer: ReturnType | null = null private probing = false + // Whether the probe in flight came from the idle sweep rather than a caller. + private idleSweepProbe = false private missedProbes = 0 private lastInboundAt = 0 private lastVoluntaryProbeAt: number | null = null + private profile: ProbeProfile private readonly idleProbeMs: number | null - private readonly probeTimeoutMs: number - private readonly missedProbeLimit: number + private readonly ordinaryProfile: ProbeProfile + private readonly urgentProfile: ProbeProfile private readonly voluntaryProbeMinIntervalMs: number private readonly now: () => number private readonly setTimer: typeof setTimeout @@ -43,8 +54,15 @@ export class RpcSessionLivenessWatchdog { constructor(private readonly options: WatchdogOptions) { this.idleProbeMs = options.idleProbeMs === undefined ? LIVENESS_IDLE_MS : options.idleProbeMs - this.probeTimeoutMs = options.probeTimeoutMs ?? LIVENESS_PROBE_TIMEOUT_MS - this.missedProbeLimit = options.missedProbeLimit ?? MISSED_PROBE_LIMIT + this.ordinaryProfile = { + timeoutMs: options.probeTimeoutMs ?? LIVENESS_PROBE_TIMEOUT_MS, + missedProbeLimit: options.missedProbeLimit ?? MISSED_PROBE_LIMIT + } + this.urgentProfile = { + timeoutMs: options.urgentProbeTimeoutMs ?? this.ordinaryProfile.timeoutMs, + missedProbeLimit: options.urgentMissedProbeLimit ?? this.ordinaryProfile.missedProbeLimit + } + this.profile = this.ordinaryProfile this.voluntaryProbeMinIntervalMs = options.voluntaryProbeMinIntervalMs ?? 0 this.now = options.now ?? Date.now this.setTimer = options.setTimer ?? setTimeout @@ -55,9 +73,11 @@ export class RpcSessionLivenessWatchdog { this.clearActiveTimer() this.identity = identity this.probing = false + this.idleSweepProbe = false this.missedProbes = 0 this.lastInboundAt = this.now() this.lastVoluntaryProbeAt = null + this.profile = this.ordinaryProfile this.armIdle(identity) } @@ -84,22 +104,28 @@ export class RpcSessionLivenessWatchdog { } this.missedProbes = 0 this.probing = false + this.idleSweepProbe = false this.armIdle(identity) } - probeNow(identity: RpcSessionIdentity): void { - if (this.identity !== identity || this.probing) { + // 'resume' is evidence the socket may have died while the process was suspended: + // it ignores the voluntary minimum, runs on the urgent bounds, and replaces any + // probe already in flight so the verdict lands on the short clock. + probeNow(identity: RpcSessionIdentity, urgency: 'nudge' | 'resume' = 'nudge'): void { + const urgent = urgency === 'resume' + if (this.identity !== identity || (this.probing && !urgent)) { return } const now = this.now() if ( + !urgent && this.lastVoluntaryProbeAt !== null && now - this.lastVoluntaryProbeAt < this.voluntaryProbeMinIntervalMs ) { return } this.lastVoluntaryProbeAt = now - this.startProbe(identity) + this.startProbe(identity, urgent ? this.urgentProfile : this.ordinaryProfile) } stop(identity: RpcSessionIdentity): void { @@ -109,9 +135,11 @@ export class RpcSessionLivenessWatchdog { this.clearActiveTimer() this.identity = null this.probing = false + this.idleSweepProbe = false this.missedProbes = 0 this.lastInboundAt = 0 this.lastVoluntaryProbeAt = null + this.profile = this.ordinaryProfile } private armIdle(identity: RpcSessionIdentity, delayMs = this.idleProbeMs): void { @@ -124,21 +152,37 @@ export class RpcSessionLivenessWatchdog { if (this.identity !== identity) { return } + if (this.options.shouldIdleProbe && !this.options.shouldIdleProbe()) { + this.armIdle(identity) + return + } const idleMs = this.now() - this.lastInboundAt if (this.idleProbeMs !== null && idleMs < this.idleProbeMs) { this.armIdle(identity, Math.max(1, this.idleProbeMs - Math.max(0, idleMs))) } else { - this.startProbe(identity) + this.startProbe(identity, this.ordinaryProfile, true) } }, delayMs) } - private startProbe(identity: RpcSessionIdentity): void { + private startProbe( + identity: RpcSessionIdentity, + profile = this.ordinaryProfile, + fromIdleSweep = false + ): void { if (this.identity !== identity) { return } this.clearActiveTimer() + // Why: switching profile starts a new observation window on a different clock. Carrying the + // ordinary probe's misses into the urgent one spends the tolerated slow answer that profile + // exists to give a cold radio, so the first 2s miss would kill a healthy socket. + if (profile !== this.profile) { + this.missedProbes = 0 + } + this.profile = profile this.probing = true + this.idleSweepProbe = fromIdleSweep const sentAt = this.now() let sent = false try { @@ -150,7 +194,7 @@ export class RpcSessionLivenessWatchdog { this.terminateCurrent(identity, 'probe-send-failed') return } - this.timer = this.setTimer(() => this.handleProbeTimeout(identity, sentAt), this.probeTimeoutMs) + this.timer = this.setTimer(() => this.handleProbeTimeout(identity, sentAt), profile.timeoutMs) } private handleProbeTimeout(identity: RpcSessionIdentity, sentAt: number): void { @@ -158,27 +202,38 @@ export class RpcSessionLivenessWatchdog { if (this.identity !== identity) { return } + // Why: the idle sweep is foreground-only because iOS suspends sockets in the + // background, where a miss is not evidence of a dead peer. Retrying here would + // spend the whole miss budget on that suspension and kill a healthy session. + if (this.idleSweepProbe && this.options.shouldIdleProbe && !this.options.shouldIdleProbe()) { + this.probing = false + this.idleSweepProbe = false + this.missedProbes = 0 + this.armIdle(identity) + return + } + const profile = this.profile const elapsedMs = this.now() - sentAt - if (elapsedMs < 0 || elapsedMs > this.probeTimeoutMs * 1.5) { + if (elapsedMs < 0 || elapsedMs > profile.timeoutMs * 1.5) { console.log('[net] activity-probe unfair window skipped', { transport: this.options.transport, elapsedMs, - timeoutMs: this.probeTimeoutMs + timeoutMs: profile.timeoutMs }) - this.startProbe(identity) + this.startProbe(identity, profile, this.idleSweepProbe) return } this.missedProbes += 1 - if (this.missedProbes >= this.missedProbeLimit) { + if (this.missedProbes >= profile.missedProbeLimit) { this.terminateCurrent(identity, 'probe-timeout') return } console.log('[net] activity-probe timeout tolerated', { transport: this.options.transport, missedProbes: this.missedProbes, - missedProbeLimit: this.missedProbeLimit + missedProbeLimit: profile.missedProbeLimit }) - this.startProbe(identity) + this.startProbe(identity, profile, this.idleSweepProbe) } private terminateCurrent( @@ -191,16 +246,17 @@ export class RpcSessionLivenessWatchdog { this.clearActiveTimer() this.identity = null this.probing = false + this.idleSweepProbe = false console.log('[net] activity-probe TIMEOUT — forcing reconnect', { transport: this.options.transport, missedProbes: this.missedProbes, - missedProbeLimit: this.missedProbeLimit + missedProbeLimit: this.profile.missedProbeLimit }) this.options.onTimeout?.({ transport: this.options.transport, reason, missedProbes: this.missedProbes, - missedProbeLimit: this.missedProbeLimit, + missedProbeLimit: this.profile.missedProbeLimit, lastInboundAgeMs: Math.max(0, this.now() - this.lastInboundAt) }) this.options.terminate(identity)