From 4d24fb340b6fd6596fac5ed37d25fe875f4d77bd Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Thu, 3 Sep 2026 17:16:35 -0400 Subject: [PATCH] fix(mobile): stage-aware relay dial bound so a slow cell is not hung up on (#18518) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A phone returning to foreground on 2026-09-03 logged "replacement session authentication timed out" five dials in a row while the desktop's relay control was live. The cell (production-gce-c27) had taken relay-auth but its assignment/reservation transactions were lock-contended (55P03 retries, 14–16s per accept); the phone's flat 12s migrateTo bound closed the socket 2–4s before the cell finished (cell logged host_data_reservation_already_bound), and because the timeout counted as a director-class failure the phone re-resolved the same cell and waited 12s again before logging — every retry landed in the same contended window. - MobileRelayE2eeLink reports onOpen once relay-auth is on the wire; MobileRelayRpcSession exposes a dial stage (opening → awaiting-hello → handshaking → confirming). - waitForAuthenticated keeps the caller's bound until the socket opens, then re-arms a per-stage budget (30s awaiting-hello, 12s handshaking, 35s confirming) so a reachable, slow cell is not treated as a black hole. - The timeout error carries the stalled stage and shows up in the "relay dial failed" log line; a stall past the open socket no longer triggers the director re-resolve round. Phone-local only: no wire change. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb --- ...e-endpoint-supervisor-stalled-cell.test.ts | 80 +++++++++++ .../mobile-endpoint-supervisor-support.ts | 6 + .../mobile-endpoint-supervisor-test-fakes.ts | 5 + .../transport/mobile-relay-e2ee-link.test.ts | 55 ++++++++ .../src/transport/mobile-relay-e2ee-link.ts | 4 + .../mobile-relay-rpc-session.test.ts | 30 +++++ .../src/transport/mobile-relay-rpc-session.ts | 24 ++-- .../mobile-relay-runtime-failover.test.ts | 5 + mobile/src/transport/relay-dial-stage.ts | 64 +++++++++ ...replacement-session-authentication.test.ts | 127 ++++++++++++++++++ .../replacement-session-authentication.ts | 58 +++++++- .../stable-logical-rpc-client.test.ts | 29 ++++ 12 files changed, 472 insertions(+), 15 deletions(-) create mode 100644 mobile/src/transport/mobile-endpoint-supervisor-stalled-cell.test.ts create mode 100644 mobile/src/transport/relay-dial-stage.ts create mode 100644 mobile/src/transport/replacement-session-authentication.test.ts diff --git a/mobile/src/transport/mobile-endpoint-supervisor-stalled-cell.test.ts b/mobile/src/transport/mobile-endpoint-supervisor-stalled-cell.test.ts new file mode 100644 index 00000000000..be4050cee5b --- /dev/null +++ b/mobile/src/transport/mobile-endpoint-supervisor-stalled-cell.test.ts @@ -0,0 +1,80 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { MobileEndpointSupervisor } from './mobile-endpoint-supervisor' +import { + dependencies, + FakeLogicalClient, + FakeRelaySession, + host, + relay +} from './mobile-endpoint-supervisor-test-fakes' +import { ReplacementAuthenticationTimeoutError } from './replacement-session-authentication' +import type { RpcClient } from './rpc-client' + +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) })) + +// The 2026-09-03 incident: five consecutive "authentication timed out" dials against a +// live desktop while the cell's assignment tables were lock-contended. Each logged +// failure was two dials — the timeout counted as a director-class failure, so the phone +// re-resolved the same cell and waited the full bound again. +describe('relay dial against a cell that took the dial and stalled', () => { + beforeEach(() => { + vi.useFakeTimers() + vi.spyOn(console, 'log').mockImplementation(() => {}) + }) + afterEach(() => { + vi.useRealTimers() + vi.restoreAllMocks() + }) + + function timingOut(logical: FakeLogicalClient, error: Error): void { + logical.migrateTo.mockImplementation(async (session: RpcClient) => { + session.close() + throw error + }) + } + + it('does not re-resolve the director and names the stalled stage', async () => { + const logical = new FakeLogicalClient('disconnected', 'lan') + timingOut(logical, new ReplacementAuthenticationTimeoutError('awaiting-hello', 30_000)) + const openRelay = vi.fn(() => new FakeRelaySession('connecting')) + const resolveRelay = vi.fn(async () => relay) + const onLog = vi.fn() + const supervisor = new MobileEndpointSupervisor( + logical, + host, + dependencies({ openRelay, resolveRelay, onLog }) + ) + + await supervisor.start() + + expect(openRelay).toHaveBeenCalledOnce() + expect(resolveRelay).not.toHaveBeenCalled() + expect(onLog).toHaveBeenCalledWith( + expect.objectContaining({ + code: 'relay-dial-failed', + detail: expect.stringContaining('timed out (awaiting-hello, 30s)') + }) + ) + supervisor.stop() + }) + + it('still re-resolves the director when the cell socket never opened', async () => { + const logical = new FakeLogicalClient('disconnected', 'lan') + timingOut(logical, new ReplacementAuthenticationTimeoutError('opening', 12_000)) + const openRelay = vi.fn(() => new FakeRelaySession('connecting')) + const resolveRelay = vi.fn(async () => relay) + const supervisor = new MobileEndpointSupervisor( + logical, + host, + dependencies({ openRelay, resolveRelay }) + ) + + await supervisor.start() + + expect(resolveRelay).toHaveBeenCalledOnce() + expect(openRelay).toHaveBeenCalledTimes(2) + supervisor.stop() + }) +}) diff --git a/mobile/src/transport/mobile-endpoint-supervisor-support.ts b/mobile/src/transport/mobile-endpoint-supervisor-support.ts index 6ee0a6b42cb..1a2c00f12de 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-support.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-support.ts @@ -1,5 +1,6 @@ import { RelayOuterError } from './mobile-relay-e2ee-link' import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel' +import { ReplacementAuthenticationTimeoutError } from './replacement-session-authentication' import type { RelayReconnectController } from './mobile-relay-reconnect-controller' import type { StableLogicalRpcClient } from './stable-logical-rpc-client' import type { HostProfile } from './types' @@ -68,6 +69,11 @@ export async function dialRelayThroughDirectorFallback(args: { } export function isDirectorResolutionFailure(error: Error): boolean { + // Why: a cell that took relay-auth and went quiet is the right cell working slowly; + // re-resolving it just doubles the wait against the same contended window. + if (error instanceof ReplacementAuthenticationTimeoutError) { + return error.stage === null || error.stage === 'opening' + } return ( !(error instanceof MobileE2EEAuthenticationError) && (!(error instanceof RelayOuterError) || [4409, 4503, 1006].includes(error.code)) diff --git a/mobile/src/transport/mobile-endpoint-supervisor-test-fakes.ts b/mobile/src/transport/mobile-endpoint-supervisor-test-fakes.ts index 4023a1a8e39..1dc1473d9db 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-test-fakes.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-test-fakes.ts @@ -1,6 +1,7 @@ import { vi } from 'vitest' import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle' import type { MobileRelayRpcSession } from './mobile-relay-rpc-session' +import { RelayDialStageTracker, type RelayDialStage } from './relay-dial-stage' import type { MobileEndpointSupervisorDependencies } from './mobile-endpoint-supervisor' import type { RpcClient } from './rpc-client' import type { MobileConnectionPath, StableLogicalRpcClient } from './stable-logical-rpc-client' @@ -51,6 +52,10 @@ export class FakeRelaySession extends FakeSession implements MobileRelayRpcSessi // Why: production-realistic defaults — fictional fake values hid three // live defects in this subsystem (latch, churn, int32 timer overflow). getAttachDeadlineAt = () => Date.now() + 10_000 + readonly dialStage = new RelayDialStageTracker() + getDialStage = () => this.dialStage.getDialStage() + onDialStageChange = (listener: (stage: RelayDialStage) => void) => + this.dialStage.onDialStageChange(listener) getResumeExpiresAt = () => this.resumeExpiry getResumeConfirmation = () => ({ v: 1 as const, diff --git a/mobile/src/transport/mobile-relay-e2ee-link.test.ts b/mobile/src/transport/mobile-relay-e2ee-link.test.ts index aa2d42b13f0..965135511eb 100644 --- a/mobile/src/transport/mobile-relay-e2ee-link.test.ts +++ b/mobile/src/transport/mobile-relay-e2ee-link.test.ts @@ -60,6 +60,61 @@ describe('MobileRelayE2eeLink', () => { expect(socket.close).toHaveBeenCalledOnce() }) + it('reports open only once relay-auth is on the wire', () => { + const socket = new ThrowingSocket() + const onOpen = vi.fn() + const sent: string[] = [] + socket.send.mockImplementation((frame: string) => { + sent.push(frame) + }) + new MobileRelayE2eeLink({ + endpoint: { + cellUrl: 'https://relay-c1.onorca.dev', + relayHostId: 'AbCdEf0123_-xyZ9' + }, + credential: 'credential', + expectedCredentialKind: 'resume', + deviceToken: 'device-token', + desktopPublicKeyB64: 'desktop-key', + onAuthenticated: vi.fn(), + onText: vi.fn(), + onBinary: vi.fn(), + onOpen, + onError: vi.fn(), + createSocket: () => socket as unknown as WebSocket + }) + + expect(onOpen).not.toHaveBeenCalled() + socket.onopen?.() + expect(sent).toHaveLength(1) + expect(JSON.parse(sent[0]!)).toMatchObject({ type: 'relay-auth', mode: 'connect' }) + expect(onOpen).toHaveBeenCalledOnce() + }) + + it('does not report open when the relay-auth write fails', () => { + const socket = new ThrowingSocket() + const onOpen = vi.fn() + new MobileRelayE2eeLink({ + endpoint: { + cellUrl: 'https://relay-c1.onorca.dev', + relayHostId: 'AbCdEf0123_-xyZ9' + }, + credential: 'credential', + expectedCredentialKind: 'resume', + deviceToken: 'device-token', + desktopPublicKeyB64: 'desktop-key', + onAuthenticated: vi.fn(), + onText: vi.fn(), + onBinary: vi.fn(), + onOpen, + onError: vi.fn(), + createSocket: () => socket as unknown as WebSocket + }) + + socket.onopen?.() + expect(onOpen).not.toHaveBeenCalled() + }) + it('keeps a typed close code when transport error precedes close', () => { const socket = new ThrowingSocket() const onError = vi.fn() diff --git a/mobile/src/transport/mobile-relay-e2ee-link.ts b/mobile/src/transport/mobile-relay-e2ee-link.ts index f19417a60f1..7743deb23dd 100644 --- a/mobile/src/transport/mobile-relay-e2ee-link.ts +++ b/mobile/src/transport/mobile-relay-e2ee-link.ts @@ -26,6 +26,8 @@ type MobileRelayE2eeLinkOptions = { onText: (plaintext: string) => void onBinary: (plaintext: Uint8Array) => void onHello?: (hello: Extract) => void + // Fired once relay-auth is on the wire: from here the cell owns the wait. + onOpen?: () => void onError: (error: Error) => void createSocket?: (url: string) => WebSocket } @@ -96,7 +98,9 @@ export class MobileRelayE2eeLink { ) } catch (error) { this.fail(asError(error)) + return } + this.options.onOpen?.() } this.socket.onmessage = (event) => { this.inboundChain = this.inboundChain diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index d5b547885cf..7f98436c63b 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -11,6 +11,7 @@ const fakes = vi.hoisted(() => ({ endpoint: { cellUrl: string; relayHostId: string } credential: string expectedCredentialKind: string + onOpen(): void onHello(value: unknown): void onAuthenticated(): void onText(value: string): void @@ -123,6 +124,35 @@ describe('mobile relay RPC session', () => { expect(session.getAttachDeadlineAt()).toEqual(expect.any(Number)) }) + // 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". + it('reports the dial stage as the link opens, receives hello, and authenticates', async () => { + const session = openSession() + const stages: string[] = [] + session.onDialStageChange((stage) => stages.push(stage)) + expect(session.getDialStage()).toBe('opening') + + fakes.linkOptions!.onOpen() + expect(session.getDialStage()).toBe('awaiting-hello') + expect(session.getState()).toBe('connecting') + fakes.linkOptions!.onHello({ + type: 'relay-hello', + ok: true, + credentialKind: 'resume', + leaseExpiresAt: Date.now() + 10_000, + acceptedCredentialVersion: 3, + acceptedAs: 'current', + resumeExpiresAt: Date.now() + 300_000 + }) + expect(session.getDialStage()).toBe('handshaking') + fakes.linkOptions!.onAuthenticated() + expect(session.getDialStage()).toBe('confirming') + await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) + expect(stages).toEqual(['awaiting-hello', 'handshaking', 'confirming']) + session.close() + }) + it('rejects a mismatched outer credential version and closes the physical link', () => { const session = openSession() fakes.linkOptions!.onHello({ diff --git a/mobile/src/transport/mobile-relay-rpc-session.ts b/mobile/src/transport/mobile-relay-rpc-session.ts index f242fce07ba..40b93927139 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -9,6 +9,7 @@ import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity' import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget' import { isRpcResponse } from './rpc-response-shape' +import { RelayDialStageTracker, type RelayDialStageSource } from './relay-dial-stage' import { RpcSessionLivenessWatchdog } from './rpc-session-liveness-watchdog' import type { RpcClient } from './rpc-client' import type { ConnectionLogSink, ConnectionState, RpcResponse } from './types' @@ -24,14 +25,15 @@ type PendingRequest = { timer: ReturnType } -export type MobileRelayRpcSession = RpcClient & { - // The cell's attach-reservation deadline (~10s). Diagnostics only — never - // schedule anything from it; rotation keys off getResumeExpiresAt(). - getAttachDeadlineAt(): number | null - getResumeExpiresAt(): number | null - getResumeConfirmation(): DeviceResumeConfirmed | null - getFailure(): Error | null -} +export type MobileRelayRpcSession = RpcClient & + RelayDialStageSource & { + // The cell's attach-reservation deadline (~10s). Diagnostics only — never + // schedule anything from it; rotation keys off getResumeExpiresAt(). + getAttachDeadlineAt(): number | null + getResumeExpiresAt(): number | null + getResumeConfirmation(): DeviceResumeConfirmed | null + getFailure(): Error | null + } export function connectMobileRelayRpcSession(args: { relay: MobileRelayEndpoint @@ -58,6 +60,7 @@ export function connectMobileRelayRpcSession(args: { let logSequence = 0 const logSessionId = `${Date.now().toString(36)}-${(++relayRpcSessionSequence).toString(36)}` const livenessIdentity = {} + const dialStage = new RelayDialStageTracker() const streams = new MobileRelayRpcStreams({ nextId, sendFrame, @@ -71,6 +74,7 @@ export function connectMobileRelayRpcSession(args: { deviceToken: args.deviceToken, desktopPublicKeyB64: args.desktopPublicKeyB64, createSocket: args.createSocket, + onOpen: () => dialStage.advance('awaiting-hello'), onHello: (hello) => { if ( hello.credentialKind !== 'resume' || @@ -81,6 +85,7 @@ export function connectMobileRelayRpcSession(args: { } attachDeadlineAt = hello.leaseExpiresAt resumeExpiresAt = hello.resumeExpiresAt + dialStage.advance('handshaking') publishState('handshaking') }, onAuthenticated: () => void confirmResume(), @@ -136,6 +141,8 @@ export function connectMobileRelayRpcSession(args: { streams.clear() publishState('disconnected') }, + getDialStage: () => dialStage.getDialStage(), + onDialStageChange: (listener) => dialStage.onDialStageChange(listener), getAttachDeadlineAt: () => attachDeadlineAt, getResumeExpiresAt: () => resumeExpiresAt, getResumeConfirmation: () => resumeConfirmation, @@ -165,6 +172,7 @@ export function connectMobileRelayRpcSession(args: { return client async function confirmResume(): Promise { + dialStage.advance('confirming') try { const response = await sendRpc( 'pairing.getEndpoints', diff --git a/mobile/src/transport/mobile-relay-runtime-failover.test.ts b/mobile/src/transport/mobile-relay-runtime-failover.test.ts index 795f2618dfb..01f4d45feb0 100644 --- a/mobile/src/transport/mobile-relay-runtime-failover.test.ts +++ b/mobile/src/transport/mobile-relay-runtime-failover.test.ts @@ -9,6 +9,7 @@ import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel import { RelayOuterError } from './mobile-relay-e2ee-link' import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle' import type { MobileRelayRpcSession } from './mobile-relay-rpc-session' +import { RelayDialStageTracker, type RelayDialStage } from './relay-dial-stage' import { MobileEndpointSupervisor, type MobileEndpointSupervisorDependencies @@ -81,6 +82,10 @@ class FakeRelaySession extends FakeSession implements MobileRelayRpcSession { // Why: production-realistic constants — fictional fake values hid three // live defects in this subsystem (latch, churn, int32 timer overflow). getAttachDeadlineAt = () => Date.now() + 10_000 + readonly dialStage = new RelayDialStageTracker() + getDialStage = () => this.dialStage.getDialStage() + onDialStageChange = (listener: (stage: RelayDialStage) => void) => + this.dialStage.onDialStageChange(listener) getResumeExpiresAt = () => Date.now() + 30 * 24 * 3_600_000 getResumeConfirmation = () => null getFailure = () => this.failure diff --git a/mobile/src/transport/relay-dial-stage.ts b/mobile/src/transport/relay-dial-stage.ts new file mode 100644 index 00000000000..c4a743f84f4 --- /dev/null +++ b/mobile/src/transport/relay-dial-stage.ts @@ -0,0 +1,64 @@ +// Where a relay dial is waiting, so a bound can tell "the cell never answered the +// upgrade" from "the cell took the dial and is slow" — the two look identical from +// ConnectionState, which stays 'connecting' until relay-hello arrives. +export type RelayDialStage = + // WebSocket upgrade not yet open. + | 'opening' + // Socket open and relay-auth sent; the cell is resolving/reserving and asking the + // desktop to attach before it can answer with relay-hello. + | 'awaiting-hello' + // relay-hello accepted; E2EE handshake with the desktop in flight. + | 'handshaking' + // E2EE authenticated; waiting on the desktop's resume confirmation. + | 'confirming' + +export type RelayDialStageSource = { + getDialStage(): RelayDialStage + onDialStageChange(listener: (stage: RelayDialStage) => void): () => void +} + +export function relayDialStageSource(session: object): RelayDialStageSource | null { + const candidate = session as Partial + return typeof candidate.getDialStage === 'function' && + typeof candidate.onDialStageChange === 'function' + ? (candidate as RelayDialStageSource) + : null +} + +export class RelayDialStageTracker implements RelayDialStageSource { + private stage: RelayDialStage = 'opening' + private readonly listeners = new Set<(stage: RelayDialStage) => void>() + + getDialStage(): RelayDialStage { + return this.stage + } + + onDialStageChange(listener: (stage: RelayDialStage) => void): () => void { + this.listeners.add(listener) + return () => this.listeners.delete(listener) + } + + advance(stage: RelayDialStage): void { + if (this.stage === stage) { + return + } + this.stage = stage + for (const listener of this.listeners) { + listener(stage) + } + } +} + +// Budget per stage once the cell holds the dial. awaiting-hello covers the cell's +// assignment/reservation transactions (observed 14–16s under lock contention) plus its +// 10s host-attach deadline; handshaking is two E2EE round trips; confirming is bounded +// by the session's own 30s resume-confirmation request, with slack so that error wins. +const RELAY_DIAL_STAGE_BUDGET_MS: Record, number> = { + 'awaiting-hello': 30_000, + handshaking: 12_000, + confirming: 35_000 +} + +export function relayDialStageBudgetMs(stage: Exclude): number { + return RELAY_DIAL_STAGE_BUDGET_MS[stage] +} diff --git a/mobile/src/transport/replacement-session-authentication.test.ts b/mobile/src/transport/replacement-session-authentication.test.ts new file mode 100644 index 00000000000..096721fa02d --- /dev/null +++ b/mobile/src/transport/replacement-session-authentication.test.ts @@ -0,0 +1,127 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { RelayDialStageTracker } from './relay-dial-stage' +import { + ReplacementAuthenticationTimeoutError, + waitForAuthenticated +} from './replacement-session-authentication' +import type { RpcClient } from './rpc-client' +import type { ConnectionState } from './types' + +class FakeSession implements RpcClient { + readonly sendRequest = vi.fn() + readonly subscribe = vi.fn(() => () => {}) + readonly updateTerminalSubscriptionViewport = vi.fn() + readonly notifyForeground = vi.fn() + readonly close = vi.fn() + private readonly listeners = new Set<(state: ConnectionState) => void>() + constructor(private state: ConnectionState = 'connecting') {} + getState = () => this.state + getReconnectAttempt = () => 0 + getLastConnectedAt = () => null + onStateChange = (listener: (state: ConnectionState) => void) => { + this.listeners.add(listener) + return () => this.listeners.delete(listener) + } + setState(state: ConnectionState): void { + this.state = state + for (const listener of this.listeners) { + listener(state) + } + } +} + +class FakeRelaySession extends FakeSession { + readonly dialStage = new RelayDialStageTracker() + getDialStage = () => this.dialStage.getDialStage() + onDialStageChange = this.dialStage.onDialStageChange.bind(this.dialStage) +} + +// Why: fake timers are active, so "still pending" is decided on the microtask queue. +async function settle( + promise: Promise +): Promise<{ status: 'pending' | 'settled'; error?: Error }> { + let outcome: { status: 'pending' | 'settled'; error?: Error } = { status: 'pending' } + void promise.then( + () => (outcome = { status: 'settled' }), + (error: Error) => (outcome = { status: 'settled', error }) + ) + await Promise.resolve() + await Promise.resolve() + return outcome +} + +describe('waitForAuthenticated', () => { + beforeEach(() => vi.useFakeTimers()) + afterEach(() => vi.useRealTimers()) + + it('keeps the flat bound for a session that reports no dial stages', async () => { + const session = new FakeSession() + const waiting = waitForAuthenticated(session, 12_000) + waiting.catch(() => {}) + await vi.advanceTimersByTimeAsync(11_999) + expect((await settle(waiting)).status).toBe('pending') + await vi.advanceTimersByTimeAsync(1) + const outcome = await settle(waiting) + expect(outcome.error).toBeInstanceOf(ReplacementAuthenticationTimeoutError) + expect(outcome.error?.message).toBe('replacement session authentication timed out') + }) + + // The 2026-09-03 incident: the cell accepted relay-auth and spent 14–16s in its + // lock-contended assignment transactions. The flat 12s bound hung up 2–4s before + // the cell finished, five dials in a row, while the desktop was live the whole time. + it('re-arms the bound per stage once the cell holds the dial', async () => { + const session = new FakeRelaySession() + const waiting = waitForAuthenticated(session, 12_000) + waiting.catch(() => {}) + await vi.advanceTimersByTimeAsync(11_000) + session.dialStage.advance('awaiting-hello') + await vi.advanceTimersByTimeAsync(5_000) + expect((await settle(waiting)).status).toBe('pending') + session.dialStage.advance('handshaking') + session.setState('handshaking') + await vi.advanceTimersByTimeAsync(11_000) + expect((await settle(waiting)).status).toBe('pending') + session.dialStage.advance('confirming') + await vi.advanceTimersByTimeAsync(20_000) + session.setState('connected') + await expect(waiting).resolves.toBeUndefined() + }) + + it('bounds a cell that took the dial and never answers, naming the stage', async () => { + const session = new FakeRelaySession() + const waiting = waitForAuthenticated(session, 12_000) + waiting.catch(() => {}) + await vi.advanceTimersByTimeAsync(2_000) + session.dialStage.advance('awaiting-hello') + await vi.advanceTimersByTimeAsync(29_999) + expect((await settle(waiting)).status).toBe('pending') + await vi.advanceTimersByTimeAsync(1) + const outcome = await settle(waiting) + expect(outcome.error).toBeInstanceOf(ReplacementAuthenticationTimeoutError) + expect((outcome.error as ReplacementAuthenticationTimeoutError).stage).toBe('awaiting-hello') + expect(outcome.error?.message).toBe( + 'replacement session authentication timed out (awaiting-hello, 30s)' + ) + }) + + it('keeps the caller bound while the socket never opens', async () => { + const session = new FakeRelaySession() + const waiting = waitForAuthenticated(session, 12_000) + waiting.catch(() => {}) + await vi.advanceTimersByTimeAsync(12_000) + const outcome = await settle(waiting) + expect((outcome.error as ReplacementAuthenticationTimeoutError).stage).toBe('opening') + expect(outcome.error?.message).toBe( + 'replacement session authentication timed out (opening, 12s)' + ) + }) + + it('ignores stage advances after the wait has settled', async () => { + const session = new FakeRelaySession() + const waiting = waitForAuthenticated(session, 12_000) + session.setState('disconnected') + await expect(waiting).rejects.toThrow('replacement session disconnected') + session.dialStage.advance('awaiting-hello') + expect(vi.getTimerCount()).toBe(0) + }) +}) diff --git a/mobile/src/transport/replacement-session-authentication.ts b/mobile/src/transport/replacement-session-authentication.ts index 0ba54a9d611..5ef9a5f1f46 100644 --- a/mobile/src/transport/replacement-session-authentication.ts +++ b/mobile/src/transport/replacement-session-authentication.ts @@ -1,21 +1,43 @@ import type { RpcClient } from './rpc-client' +import { + relayDialStageBudgetMs, + relayDialStageSource, + type RelayDialStage +} from './relay-dial-stage' + +export class ReplacementAuthenticationTimeoutError extends Error { + constructor( + readonly stage: RelayDialStage | null, + budgetMs: number + ) { + super( + stage + ? `replacement session authentication timed out (${stage}, ${Math.round(budgetMs / 1000)}s)` + : 'replacement session authentication timed out' + ) + this.name = 'ReplacementAuthenticationTimeoutError' + } +} // Why: a migration must not cut over to a session that has only opened a socket — the -// replacement has to reach 'connected' (E2EE authenticated) first, and a relay dial can -// sit in handshaking for seconds, so the wait is bounded by the caller's timeout. +// replacement has to reach 'connected' (E2EE authenticated) first. The caller's bound +// covers reaching an open socket; a relay session that reports dial stages re-arms a +// per-stage budget on every advance, so a cell that accepted the dial and is working +// slowly (lock-contended assignment tables) is not hung up on like a black hole — the +// retry would land in the same window and burn a director round on the way. export function waitForAuthenticated(session: RpcClient, timeoutMs: number): Promise { if (session.getState() === 'connected') { return Promise.resolve() } + const stages = relayDialStageSource(session) return new Promise((resolve, reject) => { let settled = false let unsubscribe: (() => void) | null = null + let unsubscribeStage: (() => void) | null = null + let timer: ReturnType | null = null // Why: armed before subscribing — a synchronous notification during registration // must find a timer to clear, or a settled wait leaves it running for 12s. - const timer = setTimeout(() => { - finish() - reject(new Error('replacement session authentication timed out')) - }, timeoutMs) + arm(stages?.getDialStage() ?? null) unsubscribe = session.onStateChange((state) => { if (state === 'connected') { finish() @@ -29,6 +51,23 @@ export function waitForAuthenticated(session: RpcClient, timeoutMs: number): Pro // Why: the notification fired inside onStateChange, before we held the handle. unsubscribe() unsubscribe = null + } else if (stages) { + unsubscribeStage = stages.onDialStageChange((stage) => arm(stage)) + } + + function arm(stage: RelayDialStage | null): void { + if (settled) { + return + } + if (timer) { + clearTimeout(timer) + } + const budgetMs = + stage === null || stage === 'opening' ? timeoutMs : relayDialStageBudgetMs(stage) + timer = setTimeout(() => { + finish() + reject(new ReplacementAuthenticationTimeoutError(stage, budgetMs)) + }, budgetMs) } function finish(): void { @@ -36,9 +75,14 @@ export function waitForAuthenticated(session: RpcClient, timeoutMs: number): Pro return } settled = true - clearTimeout(timer) + if (timer) { + clearTimeout(timer) + timer = null + } unsubscribe?.() unsubscribe = null + unsubscribeStage?.() + unsubscribeStage = null } }) } diff --git a/mobile/src/transport/stable-logical-rpc-client.test.ts b/mobile/src/transport/stable-logical-rpc-client.test.ts index 0ba7a1f2ea7..faa236a88ca 100644 --- a/mobile/src/transport/stable-logical-rpc-client.test.ts +++ b/mobile/src/transport/stable-logical-rpc-client.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it, vi } from 'vitest' +import { RelayDialStageTracker, type RelayDialStage } from './relay-dial-stage' import type { ConnectionState, RpcResponse } from './types' import type { RpcClient } from './rpc-client' import { isRpcDeliveryUnknown, markRpcDeliveryUnknown } from './rpc-delivery-ambiguity' @@ -399,6 +400,34 @@ describe('stable logical RPC client', () => { expect(client.getPendingPath()).toBeNull() }) + // Pins the shipping wiring: migrateTo's bound honors the replacement's dial stages. + it('outlives the flat bound when the relay cell holds the dial', async () => { + vi.useFakeTimers() + try { + const oldSession = new FakeSession('connected') + const replacement = Object.assign(new FakeSession('connecting'), { + dialStage: new RelayDialStageTracker(), + getDialStage(): RelayDialStage { + return this.dialStage.getDialStage() + }, + onDialStageChange(listener: (stage: RelayDialStage) => void) { + return this.dialStage.onDialStageChange(listener) + } + }) + const client = createStableLogicalRpcClient(oldSession, 'lan') + const migrating = client.migrateTo(replacement, 'relay', 12_000) + await vi.advanceTimersByTimeAsync(1_000) + replacement.dialStage.advance('awaiting-hello') + await vi.advanceTimersByTimeAsync(20_000) + expect(replacement.close).not.toHaveBeenCalled() + replacement.setState('connected') + await migrating + expect(client.getActivePath()).toBe('relay') + } finally { + vi.useRealTimers() + } + }) + it('closes a replacement that fails authentication and preserves the active session', async () => { const oldSession = new FakeSession('connected') const replacement = new FakeSession('connecting')