From 50c43a6e11f197d05065d3eebfa514280b30841f Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sat, 1 Aug 2026 10:00:06 -0700 Subject: [PATCH] fix(mobile): close reconnect lifecycle review gaps --- mobile/app/h/[hostId]/edit.tsx | 28 +++--- mobile/src/host-edit-save-flow.test.ts | 23 +++-- ...sting-host-relay-routing-lifecycle.test.ts | 43 ++++++++- .../transport/existing-host-relay-routing.ts | 44 +++++---- .../force-reconnect-rpc-health.test.ts | 31 ++++++ .../transport/force-reconnect-rpc-health.ts | 58 ++++++++++-- .../src/transport/host-profile-publication.ts | 20 +++- mobile/src/transport/host-store.test.ts | 24 ++--- .../transport/mobile-endpoint-lifecycle.ts | 19 ++-- .../mobile-relay-direct-upgrade.test.ts | 25 +++++ .../mobile-relay-pending-requests.ts | 24 +++++ .../mobile-relay-rpc-session.test.ts | 94 +++++++++++++++++++ .../src/transport/mobile-relay-rpc-session.ts | 47 ++++------ .../src/transport/mobile-relay-rpc-streams.ts | 21 ++++- .../rpc-client-request-deadline.test.ts | 25 +++++ mobile/src/transport/rpc-client.test.ts | 30 +++++- mobile/src/transport/rpc-client.ts | 24 ++--- .../transport/rpc-stream-response-shape.ts | 25 ++--- .../timed-out-control-request-index.test.ts | 27 ++++++ .../timed-out-control-request-index.ts | 42 +++++++++ 20 files changed, 535 insertions(+), 139 deletions(-) create mode 100644 mobile/src/transport/mobile-relay-pending-requests.ts create mode 100644 mobile/src/transport/timed-out-control-request-index.test.ts create mode 100644 mobile/src/transport/timed-out-control-request-index.ts diff --git a/mobile/app/h/[hostId]/edit.tsx b/mobile/app/h/[hostId]/edit.tsx index f72d408d730..041e29a3ef0 100644 --- a/mobile/app/h/[hostId]/edit.tsx +++ b/mobile/app/h/[hostId]/edit.tsx @@ -17,10 +17,7 @@ import { colors, radii, spacing, typography } from '../../../src/theme/mobile-th import { loadHosts, updateHostNameAndEndpoint } from '../../../src/transport/host-store' import { getHostListLoadRevision } from '../../../src/transport/host-list-load-sharing' import { displayHostEndpoint } from '../../../src/transport/host-endpoint' -import { - hostProfileAfterEdit, - resolveHostEndpointEdit -} from '../../../src/transport/host-endpoint-edit' +import { resolveHostEndpointEdit } from '../../../src/transport/host-endpoint-edit' import { useForceReconnect, usePrimeHosts } from '../../../src/transport/client-context' import { showForceReconnectError } from '../../../src/transport/force-reconnect-feedback' import type { HostProfile } from '../../../src/transport/types' @@ -109,7 +106,7 @@ export default function EditHostScreen() { ...(willRename ? { name: nextName } : {}), ...(nextEndpoint !== undefined ? { endpoint: nextEndpoint } : {}) } - let reconnectProfile: HostProfile | null = hostProfileAfterEdit(host, updates) + let reconnectWithCurrentProfile = false savingRef.current = true setSaving(true) @@ -130,11 +127,20 @@ export default function EditHostScreen() { // Why: the write already committed above; a re-prime failure here // must not be reported as a save failure — the next loadHosts() call // elsewhere in the app picks up the fresh state regardless. - const hostLoadRevision = getHostListLoadRevision() - const hosts = await loadHosts() - if (hostLoadRevision === getHostListLoadRevision()) { + for (let attempt = 0; attempt < 3; attempt += 1) { + const hostLoadRevision = getHostListLoadRevision() + const hosts = await loadHosts() + if (hostLoadRevision !== getHostListLoadRevision()) { + continue + } + if (!hosts.some(({ id }) => id === host.id)) { + break + } primeHosts(hosts, hostLoadRevision) - reconnectProfile = hosts.find(({ id }) => id === host.id) ?? null + if (hostLoadRevision === getHostListLoadRevision()) { + reconnectWithCurrentProfile = true + break + } } } catch { // best-effort re-prime; persisted data is unaffected @@ -144,11 +150,11 @@ export default function EditHostScreen() { setSaving(false) router.back() - if (nextEndpoint !== undefined && reconnectProfile !== null) { + if (nextEndpoint !== undefined && reconnectWithCurrentProfile) { // Why: reconnect is a follow-on side effect of a save that already // committed — its failure or a hang must not be reported as a save // failure or block navigating back. - void forceReconnectHost(host.id, reconnectProfile).catch(showForceReconnectError) + void forceReconnectHost(host.id).catch(showForceReconnectError) } } diff --git a/mobile/src/host-edit-save-flow.test.ts b/mobile/src/host-edit-save-flow.test.ts index a63ca1cf79f..903256f8a47 100644 --- a/mobile/src/host-edit-save-flow.test.ts +++ b/mobile/src/host-edit-save-flow.test.ts @@ -186,7 +186,8 @@ describe('edit host handleSave', () => { expect(dependencies.updateHostNameAndEndpoint).toHaveBeenCalledWith('host-1', { endpoint: 'ws://192.168.1.20:6768' }) - expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', savedHost) + expect(dependencies.primeHosts).toHaveBeenCalledWith([savedHost], 0) + expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1') expect(dependencies.back).toHaveBeenCalledTimes(1) act(() => renderer.unmount()) @@ -209,7 +210,8 @@ describe('edit host handleSave', () => { name: 'Home Desk', endpoint: 'ws://192.168.1.20:6768' }) - expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', savedHost) + expect(dependencies.primeHosts).toHaveBeenCalledWith([savedHost], 0) + expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1') expect(dependencies.back).toHaveBeenCalledTimes(1) act(() => renderer.unmount()) @@ -265,7 +267,7 @@ describe('edit host handleSave', () => { await Promise.resolve() }) - expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', savedHost) + expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1') expect(dependencies.back).toHaveBeenCalledTimes(1) expect(dependencies.alert).toHaveBeenCalledWith('Unable to reconnect', 'connect failed') @@ -286,24 +288,29 @@ describe('edit host handleSave', () => { setFieldValue(renderer, 'Address', '192.168.1.20:6768') await pressSave(renderer) - expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', relayHost) + expect(dependencies.primeHosts).toHaveBeenCalledWith([relayHost], 0) + expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1') act(() => renderer.unmount()) }) - it('preserves the edited profile when an unrelated write invalidates the loaded snapshot', async () => { + it('reloads the current edited profile when an unrelated write invalidates a snapshot', async () => { const editedHost = { ...HOST_FIXTURE, endpoint: 'ws://192.168.1.20:6768' } const invalidatedHost = { ...HOST_FIXTURE, name: 'Stale snapshot' } dependencies.loadHosts .mockResolvedValueOnce([HOST_FIXTURE]) .mockResolvedValueOnce([invalidatedHost]) - dependencies.getHostListLoadRevision.mockReturnValueOnce(1).mockReturnValueOnce(2) + .mockResolvedValueOnce([editedHost]) + dependencies.getHostListLoadRevision + .mockReturnValueOnce(1) + .mockReturnValueOnce(2) + .mockReturnValue(2) const renderer = await renderEditHostRoute() setFieldValue(renderer, 'Address', '192.168.1.20:6768') await pressSave(renderer) - expect(dependencies.primeHosts).not.toHaveBeenCalled() - expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', editedHost) + expect(dependencies.primeHosts).toHaveBeenCalledWith([editedHost], 2) + expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1') act(() => renderer.unmount()) }) diff --git a/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts b/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts index 8e94f40184d..51969e05c5f 100644 --- a/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts +++ b/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts @@ -12,6 +12,7 @@ vi.mock('./mobile-relay-host-overlay-store', () => overlayStoreMock) vi.mock('./host-list-load-sharing', () => ({ dropSharedHostListLoad: vi.fn() })) import { + MobileRelayUpgradeLifecycleRetiredError, MobileRelayUpgradeHostSupersededError, saveExistingHostRelayRouting, writeExistingHostRelayCredentialBundle @@ -69,7 +70,7 @@ describe('existing host relay endpoint lifecycle publication', () => { ) releaseOldToken?.(HOST.deviceToken) - await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError) + await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError) expect(writes).toEqual(['replacement']) }) @@ -97,7 +98,7 @@ describe('existing host relay endpoint lifecycle publication', () => { ) releaseOldToken?.(HOST.deviceToken) - await expect(stalePublication).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError) + await expect(stalePublication).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError) expect(overlayStoreMock.saveMobileRelayHostOverlay).toHaveBeenCalledOnce() expect(overlayStoreMock.saveMobileRelayHostOverlay).toHaveBeenCalledWith( expect.objectContaining({ relayHostId: 'AbCdEf0123_-new9' }) @@ -125,7 +126,7 @@ describe('existing host relay endpoint lifecycle publication', () => { beginHostEndpointPublicationLifecycle(HOST.id) releaseMetadata?.(HOST) - await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError) + await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError) expect(writeBundle).not.toHaveBeenCalled() }) @@ -151,7 +152,7 @@ describe('existing host relay endpoint lifecycle publication', () => { beginHostEndpointPublicationLifecycle(HOST.id) releaseCredentialWrite?.() - await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError) + await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError) expect(overlayStoreMock.saveMobileRelayHostOverlay).not.toHaveBeenCalled() }) @@ -191,6 +192,40 @@ describe('existing host relay endpoint lifecycle publication', () => { await expect(staleWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError) expect(writeBundle).not.toHaveBeenCalled() }) + + it('binds an endpoint lifecycle to the profile revision before same-identity recovery', async () => { + let releaseRecovery: (() => void) | null = null + let markRecoveryStarted: (() => void) | null = null + const recoveryStarted = new Promise((resolve) => { + markRecoveryStarted = resolve + }) + const recoveryPending = new Promise((resolve) => { + releaseRecovery = resolve + }) + const oldLifecycle = beginHostEndpointPublicationLifecycle(HOST.id) + const recovery = publishHostProfileTransaction( + HOST, + async () => { + markRecoveryStarted?.() + await recoveryPending + }, + async () => {} + ) + await recoveryStarted + + const writeBundle = vi.fn(async () => {}) + const staleWrite = writeExistingHostRelayCredentialBundle( + HOST, + credentialBundle(1), + writeBundle, + oldLifecycle + ) + releaseRecovery?.() + + await recovery + await expect(staleWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError) + expect(writeBundle).not.toHaveBeenCalled() + }) }) function credentialBundle(version: number): MobileRelayCredentialBundle { diff --git a/mobile/src/transport/existing-host-relay-routing.ts b/mobile/src/transport/existing-host-relay-routing.ts index 71e78a07277..8aeb2b2417d 100644 --- a/mobile/src/transport/existing-host-relay-routing.ts +++ b/mobile/src/transport/existing-host-relay-routing.ts @@ -2,9 +2,10 @@ import { HostProfileSchema, type HostProfile } from './types' import { loadStoredHostIdentity } from './host-store' import { readHostDeviceToken } from './host-device-token-store' import { - getHostEndpointPublicationGeneration, + getHostEndpointPublicationLifecycle, getHostProfilePublicationRevision, - serializeHostProfilePublication + serializeHostProfilePublication, + type HostEndpointPublicationLifecycle } from './host-profile-publication' import * as hostListLoads from './host-list-load-sharing' import { saveMobileRelayHostOverlay } from './mobile-relay-host-overlay-store' @@ -12,28 +13,28 @@ import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bund export class MobileRelayUpgradeHostRemovedError extends Error {} export class MobileRelayUpgradeHostSupersededError extends Error {} +export class MobileRelayUpgradeLifecycleRetiredError extends Error {} export async function saveExistingHostRelayRouting( host: HostProfile, beforePublish?: () => Promise, - endpointGeneration = getHostEndpointPublicationGeneration(host.id) + endpointLifecycle = getHostEndpointPublicationLifecycle(host.id) ): Promise { const validated = HostProfileSchema.parse(host) - const revision = getHostProfilePublicationRevision(validated.id) await requireCurrentHostCredential(validated) await serializeHostProfilePublication(validated.id, async () => { - requireCurrentPublication(validated.id, revision, endpointGeneration) + requireCurrentEndpointGeneration(validated.id, endpointLifecycle) await requireCurrentHostDeviceToken(validated) - requireCurrentPublication(validated.id, revision, endpointGeneration) + requireCurrentEndpointGeneration(validated.id, endpointLifecycle) const existing = await requireCurrentHostMetadata(validated) - requireCurrentPublication(validated.id, revision, endpointGeneration) + requireCurrentPublication(validated.id, endpointLifecycle) const { endpoints, relayHostId, relay } = validated if (!endpoints || !relayHostId || !relay) { throw new Error('mobile relay upgrade routing metadata missing') } // Why: serialize the credential before its routing overlay so neither side can cross a re-pair. await beforePublish?.() - requireCurrentPublication(validated.id, revision, endpointGeneration) + requireCurrentPublication(validated.id, endpointLifecycle) await saveMobileRelayHostOverlay({ v: 2, hostId: validated.id, @@ -53,20 +54,19 @@ export async function writeExistingHostRelayCredentialBundle( host: HostProfile, bundle: MobileRelayCredentialBundle, writeBundle: (bundle: MobileRelayCredentialBundle) => Promise, - endpointGeneration = getHostEndpointPublicationGeneration(host.id) + endpointLifecycle = getHostEndpointPublicationLifecycle(host.id) ): Promise { const validated = HostProfileSchema.parse(host) - const revision = getHostProfilePublicationRevision(validated.id) await requireCurrentHostCredential(validated) await serializeHostProfilePublication(validated.id, async () => { - requireCurrentPublication(validated.id, revision, endpointGeneration) + requireCurrentEndpointGeneration(validated.id, endpointLifecycle) await requireCurrentHostDeviceToken(validated) - requireCurrentPublication(validated.id, revision, endpointGeneration) + requireCurrentEndpointGeneration(validated.id, endpointLifecycle) await requireCurrentHostMetadata(validated) if (bundle.hostId !== validated.id || bundle.deviceToken !== validated.deviceToken) { throw new MobileRelayUpgradeHostSupersededError('mobile relay credential identity mismatch') } - requireCurrentPublication(validated.id, revision, endpointGeneration) + requireCurrentPublication(validated.id, endpointLifecycle) await writeBundle(bundle) }) } @@ -106,13 +106,21 @@ function requireMatchingHostMetadata( function requireCurrentPublication( hostId: string, - revision: number, - endpointGeneration: number + endpointLifecycle: HostEndpointPublicationLifecycle ): void { if ( - getHostProfilePublicationRevision(hostId) !== revision || - getHostEndpointPublicationGeneration(hostId) !== endpointGeneration + getHostProfilePublicationRevision(hostId) !== endpointLifecycle.profileRevision || + getHostEndpointPublicationLifecycle(hostId).generation !== endpointLifecycle.generation ) { - throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired') + throw new MobileRelayUpgradeLifecycleRetiredError('mobile relay endpoint lifecycle was retired') + } +} + +function requireCurrentEndpointGeneration( + hostId: string, + endpointLifecycle: HostEndpointPublicationLifecycle +): void { + if (getHostEndpointPublicationLifecycle(hostId).generation !== endpointLifecycle.generation) { + throw new MobileRelayUpgradeLifecycleRetiredError('mobile relay endpoint lifecycle was retired') } } diff --git a/mobile/src/transport/force-reconnect-rpc-health.test.ts b/mobile/src/transport/force-reconnect-rpc-health.test.ts index 0da2a655fd0..3f0add2a9e9 100644 --- a/mobile/src/transport/force-reconnect-rpc-health.test.ts +++ b/mobile/src/transport/force-reconnect-rpc-health.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import type { RpcClient, SendRequestOptions } from './rpc-client' import type { ConnectionState, RpcResponse } from './types' import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health' +import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity' import { LogicalClientCutoverError } from './stable-logical-rpc-client' describe('Force Reconnect RPC health', () => { @@ -79,4 +80,34 @@ describe('Force Reconnect RPC health', () => { ) expect(sendRequest).toHaveBeenCalledOnce() }) + + it('waits for a Relay replacement after the active session drops', async () => { + let state: ConnectionState = 'connected' + const listeners = new Set<(next: ConnectionState) => void>() + const sendRequest = vi + .fn<() => Promise>() + .mockImplementationOnce(async () => { + state = 'disconnected' + throw markRpcDeliveryUnknown(new Error('relay RPC interrupted')) + }) + .mockResolvedValueOnce({ id: 'rpc-2', ok: true, result: {} }) + const client = { + sendRequest, + getState: () => state, + onStateChange: (listener: (next: ConnectionState) => void) => { + listeners.add(listener) + return () => listeners.delete(listener) + } + } as unknown as RpcClient + + const verification = verifyForceReconnectRpcHealth(client) + await vi.waitFor(() => expect(listeners.size).toBe(1)) + state = 'connected' + for (const listener of listeners) { + listener(state) + } + + await expect(verification).resolves.toBeUndefined() + expect(sendRequest).toHaveBeenCalledTimes(2) + }) }) diff --git a/mobile/src/transport/force-reconnect-rpc-health.ts b/mobile/src/transport/force-reconnect-rpc-health.ts index fc7b1779203..29262852845 100644 --- a/mobile/src/transport/force-reconnect-rpc-health.ts +++ b/mobile/src/transport/force-reconnect-rpc-health.ts @@ -22,15 +22,61 @@ export async function verifyForceReconnectRpcHealth( }) return } catch (error) { - if ( - client.getState() !== 'reconnecting' && - !isLogicalClientCutoverError(error) && - !isRpcDeliveryUnknown(error) && - !(error instanceof Error && error.message === 'Connection interrupted') - ) { + const state = client.getState() + if (state === 'auth-failed' || !isRecoverableHealthError(error, state)) { throw error } lastError = error + if (state === 'disconnected') { + await waitForReconnectState(client, deadline, error) + } } } } + +function isRecoverableHealthError( + error: unknown, + state: ReturnType +): boolean { + return ( + state === 'reconnecting' || + isLogicalClientCutoverError(error) || + isRpcDeliveryUnknown(error) || + (error instanceof Error && + ['Connection interrupted', 'relay session not connected'].includes(error.message)) + ) +} + +function waitForReconnectState( + client: RpcClient, + deadline: number, + lastError: unknown +): Promise { + if (client.getState() !== 'disconnected') { + return Promise.resolve() + } + return new Promise((resolve, reject) => { + const timeoutMs = deadline - Date.now() + if (timeoutMs <= 0) { + reject(lastError) + return + } + const timer = setTimeout(() => finish(() => reject(lastError)), timeoutMs) + const unsubscribe = client.onStateChange((state) => { + if (state === 'disconnected') { + return + } + finish(state === 'auth-failed' ? () => reject(lastError) : resolve) + }) + const currentState = client.getState() + if (currentState !== 'disconnected') { + finish(currentState === 'auth-failed' ? () => reject(lastError) : resolve) + } + + function finish(settle: () => void): void { + clearTimeout(timer) + unsubscribe() + settle() + } + }) +} diff --git a/mobile/src/transport/host-profile-publication.ts b/mobile/src/transport/host-profile-publication.ts index eb800ee5cbb..8cc9c07dfbc 100644 --- a/mobile/src/transport/host-profile-publication.ts +++ b/mobile/src/transport/host-profile-publication.ts @@ -4,18 +4,30 @@ const pendingByHost = new Map>() const revisionByHost = new Map() const endpointGenerationByHost = new Map() +export type HostEndpointPublicationLifecycle = Readonly<{ + generation: number + profileRevision: number +}> + export function getHostProfilePublicationRevision(hostId: string): number { return revisionByHost.get(hostId) ?? 0 } -export function beginHostEndpointPublicationLifecycle(hostId: string): number { +export function beginHostEndpointPublicationLifecycle( + hostId: string +): HostEndpointPublicationLifecycle { const generation = (endpointGenerationByHost.get(hostId) ?? 0) + 1 endpointGenerationByHost.set(hostId, generation) - return generation + return { generation, profileRevision: getHostProfilePublicationRevision(hostId) } } -export function getHostEndpointPublicationGeneration(hostId: string): number { - return endpointGenerationByHost.get(hostId) ?? 0 +export function getHostEndpointPublicationLifecycle( + hostId: string +): HostEndpointPublicationLifecycle { + return { + generation: endpointGenerationByHost.get(hostId) ?? 0, + profileRevision: getHostProfilePublicationRevision(hostId) + } } export function serializeHostProfilePublication( diff --git a/mobile/src/transport/host-store.test.ts b/mobile/src/transport/host-store.test.ts index d37d9c29ea1..1ed95272016 100644 --- a/mobile/src/transport/host-store.test.ts +++ b/mobile/src/transport/host-store.test.ts @@ -432,7 +432,6 @@ describe('host-store list mutations', () => { releaseOldToken = resolve }) ) - const relayHostId = 'AbCdEf0123_-xyZ9' const stalePublication = saveExistingHostRelayRouting({ ...HOST_ONE, deviceToken: 'token-1', @@ -440,13 +439,13 @@ describe('host-store list mutations', () => { { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint }, { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' } ], - relayHostId, + relayHostId: 'AbCdEf0123_-xyZ9', relay: { v: 1, directorUrl: 'https://relay.onorca.dev', cellUrl: 'https://relay.onorca.dev', assignmentEpoch: 1, - relayHostId, + relayHostId: 'AbCdEf0123_-xyZ9', e2eeFraming: 2 } }) @@ -476,13 +475,14 @@ describe('host-store list mutations', () => { it('lets same-host re-pair supersede a publication stalled on identity lookup', async () => { let releaseOldToken: ((token: string | null) => void) | null = null - secureStoreMock.getItemAsync.mockImplementationOnce( - () => - new Promise((resolve) => { - releaseOldToken = resolve - }) - ) - const relayHostId = 'AbCdEf0123_-xyZ9' + secureStoreMock.getItemAsync + .mockImplementationOnce( + () => + new Promise((resolve) => { + releaseOldToken = resolve + }) + ) + .mockResolvedValue('replacement-token') const stalePublication = saveExistingHostRelayRouting({ ...HOST_ONE, deviceToken: 'token-1', @@ -490,13 +490,13 @@ describe('host-store list mutations', () => { { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint }, { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' } ], - relayHostId, + relayHostId: 'AbCdEf0123_-xyZ9', relay: { v: 1, directorUrl: 'https://relay.onorca.dev', cellUrl: 'https://relay.onorca.dev', assignmentEpoch: 1, - relayHostId, + relayHostId: 'AbCdEf0123_-xyZ9', e2eeFraming: 2 } }) diff --git a/mobile/src/transport/mobile-endpoint-lifecycle.ts b/mobile/src/transport/mobile-endpoint-lifecycle.ts index b01ffb41dad..a7b03b6adc4 100644 --- a/mobile/src/transport/mobile-endpoint-lifecycle.ts +++ b/mobile/src/transport/mobile-endpoint-lifecycle.ts @@ -12,7 +12,10 @@ import { saveExistingHostRelayRouting, writeExistingHostRelayCredentialBundle } from './existing-host-relay-routing' -import { beginHostEndpointPublicationLifecycle } from './host-profile-publication' +import { + beginHostEndpointPublicationLifecycle, + type HostEndpointPublicationLifecycle +} from './host-profile-publication' import { upgradeDirectMobileRelay } from './mobile-relay-direct-upgrade' import { MobileRelayDirectUpgradeController } from './mobile-relay-direct-upgrade-controller' import type { StableLogicalRpcClient } from './stable-logical-rpc-client' @@ -35,7 +38,7 @@ export function startMobileEndpointLifecycle( let stopped = false let foreground = true let owner: EndpointOwner - const endpointGeneration = beginHostEndpointPublicationLifecycle(initialHost.id) + const endpointLifecycle = beginHostEndpointPublicationLifecycle(initialHost.id) const publishHostUpdate = (host: HostProfile): void => { if (!stopped) { onHostUpdated(host) @@ -46,7 +49,7 @@ export function startMobileEndpointLifecycle( if (stopped) { return } - const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate, endpointGeneration) + const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate, endpointLifecycle) owner.stop() owner = supervisor supervisor.setForeground(foreground) @@ -54,7 +57,7 @@ export function startMobileEndpointLifecycle( } if (initialHost.relay) { - owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate, endpointGeneration) + owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate, endpointLifecycle) void owner.start() } else { owner = new MobileRelayDirectUpgradeController(logical, initialHost, { @@ -65,7 +68,7 @@ export function startMobileEndpointLifecycle( dependencies: { randomBytes: ExpoCrypto.getRandomBytes, saveHost: (updatedHost, beforePublish) => - saveExistingHostRelayRouting(updatedHost, beforePublish, endpointGeneration) + saveExistingHostRelayRouting(updatedHost, beforePublish, endpointLifecycle) } }), onUpgraded: async ({ host }) => { @@ -93,7 +96,7 @@ function createSupervisor( host: HostProfile, onLog: ConnectionLogSink, onHostUpdated: (host: HostProfile) => void, - endpointGeneration: number + endpointLifecycle: HostEndpointPublicationLifecycle ): MobileEndpointSupervisor { return new MobileEndpointSupervisor(logical, host, { openDirect: (endpoint) => connect(endpoint, host.deviceToken, host.publicKeyB64, { onLog }), @@ -113,10 +116,10 @@ function createSupervisor( host, bundle, writeMobileRelayCredentialBundle, - endpointGeneration + endpointLifecycle ), saveHost: (updatedHost, beforePublish) => - saveExistingHostRelayRouting(updatedHost, beforePublish, endpointGeneration), + saveExistingHostRelayRouting(updatedHost, beforePublish, endpointLifecycle), onHostUpdated, now: Date.now, randomBytes: ExpoCrypto.getRandomBytes, diff --git a/mobile/src/transport/mobile-relay-direct-upgrade.test.ts b/mobile/src/transport/mobile-relay-direct-upgrade.test.ts index 46ff1fa20bb..aa1cd7b451a 100644 --- a/mobile/src/transport/mobile-relay-direct-upgrade.test.ts +++ b/mobile/src/transport/mobile-relay-direct-upgrade.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from 'vitest' import type { MobileRelayEndpoint } from '../../../src/shared/mobile-relay-credential-contract' import { + MobileRelayUpgradeLifecycleRetiredError, MobileRelayUpgradeHostRemovedError, MobileRelayUpgradeHostSupersededError } from './existing-host-relay-routing' @@ -181,6 +182,30 @@ describe('existing direct pairing relay upgrade', () => { await expect(deps.readJournal()).resolves.toEqual(replacement) }) + it('retains a committed journal when an endpoint lifecycle is retired', async () => { + const journal = createMobileRelayDirectUpgradeJournal(host.id, (length) => + new Uint8Array(length).fill(6) + ) + const committed = installed(journal) + const deps = dependencies(journal) + deps.saveHost.mockRejectedValue( + new MobileRelayUpgradeLifecycleRetiredError('mobile relay endpoint lifecycle was retired') + ) + const client = clientWith([ + success({ + v: 1, + relay, + installStatus: { v: 1, reqId: journal.reqId, state: 'committed', result: committed } + }) + ]) + + await expect( + upgradeDirectMobileRelay({ client, host, dependencies: deps }) + ).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError) + expect(deps.clearJournal).not.toHaveBeenCalled() + await expect(deps.readJournal()).resolves.toEqual(journal) + }) + it('retains recovery state when host identity credentials are temporarily unavailable', async () => { const journal = createMobileRelayDirectUpgradeJournal(host.id, (length) => new Uint8Array(length).fill(8) diff --git a/mobile/src/transport/mobile-relay-pending-requests.ts b/mobile/src/transport/mobile-relay-pending-requests.ts new file mode 100644 index 00000000000..ba98e1f35b1 --- /dev/null +++ b/mobile/src/transport/mobile-relay-pending-requests.ts @@ -0,0 +1,24 @@ +import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity' +import type { RpcResponse } from './types' + +export type MobileRelayPendingRequest = { + resolve: (response: RpcResponse) => void + reject: (error: Error) => void + timer: ReturnType +} + +export function rejectMobileRelayPendingRequests( + pending: Map, + error: Error +): void { + if (pending.size === 0) { + return + } + // Written frames may have executed before the Relay session failed. + markRpcDeliveryUnknown(error) + for (const request of pending.values()) { + clearTimeout(request.timer) + request.reject(error) + } + pending.clear() +} diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index 3b6380f967f..006e0ea02c0 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -334,6 +334,63 @@ describe('mobile relay RPC session', () => { } }) + it.each([ + ['lease-only terminal', { type: 'subscribed', streamId: null }], + ['native-chat snapshot', { type: 'snapshot' }], + ['session-tabs snapshot', { type: 'snapshot' }] + ])('counts a Relay %s reply as control-plane liveness', async (_name, result) => { + const { session } = await authenticateSession() + vi.useFakeTimers() + try { + session.subscribe('nativeChat.subscribe', {}, vi.fn()) + await vi.advanceTimersByTimeAsync(0) + const subscribe = fakes.sendText.mock.calls + .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string }) + .find(({ method }) => method === 'nativeChat.subscribe')! + + session.notifyForeground() + fakes.linkOptions!.onText( + JSON.stringify({ id: subscribe.id, ok: true, streaming: true, result, _meta: {} }) + ) + await vi.advanceTimersByTimeAsync(8_000) + + expect(session.getState()).toBe('connected') + expect(fakes.close).not.toHaveBeenCalled() + } finally { + session.close() + vi.useRealTimers() + } + }) + + it('does not count repeated Relay subscription frames as fresh control responses', async () => { + const { session } = await authenticateSession() + vi.useFakeTimers() + try { + session.subscribe('nativeChat.subscribe', {}, vi.fn()) + await vi.advanceTimersByTimeAsync(0) + const subscribe = fakes.sendText.mock.calls + .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string }) + .find(({ method }) => method === 'nativeChat.subscribe')! + const response = JSON.stringify({ + id: subscribe.id, + ok: true, + streaming: true, + result: { type: 'snapshot' }, + _meta: {} + }) + fakes.linkOptions!.onText(response) + + session.notifyForeground() + fakes.linkOptions!.onText(response) + await vi.advanceTimersByTimeAsync(8_000) + + expect(session.getState()).toBe('disconnected') + expect(fakes.close).toHaveBeenCalledOnce() + } finally { + vi.useRealTimers() + } + }) + it('does not let Relay terminal payload traffic mask a stalled control channel', async () => { const { session } = await authenticateSession() vi.useFakeTimers() @@ -621,4 +678,41 @@ describe('mobile relay RPC session', () => { vi.useRealTimers() } }) + + it('retains late Relay reply evidence after an earlier probe completes', async () => { + const { session } = await authenticateSession() + vi.useFakeTimers() + try { + const request = session.sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + const outcome = request.catch((error: unknown) => error) + await vi.advanceTimersByTimeAsync(0) + const timedOutRequest = fakes.sendText.mock.calls + .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string }) + .find(({ method }) => method === 'browser.screenshot')! + + await vi.advanceTimersByTimeAsync(100) + await expect(outcome).resolves.toMatchObject({ + message: 'relay RPC timed out: browser.screenshot' + }) + const firstProbe = fakes.sendText.mock.calls + .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string }) + .find(({ method }) => method === 'status.get')! + fakes.linkOptions!.onText( + JSON.stringify({ id: firstProbe.id, ok: true, result: {}, _meta: {} }) + ) + await vi.advanceTimersByTimeAsync(0) + + session.notifyForeground() + fakes.linkOptions!.onText( + JSON.stringify({ id: timedOutRequest.id, ok: true, result: {}, _meta: {} }) + ) + await vi.advanceTimersByTimeAsync(8_000) + + expect(session.getState()).toBe('connected') + expect(fakes.close).not.toHaveBeenCalled() + } finally { + session.close() + vi.useRealTimers() + } + }) }) diff --git a/mobile/src/transport/mobile-relay-rpc-session.ts b/mobile/src/transport/mobile-relay-rpc-session.ts index 3ae240ecd79..230fe1f4a29 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -8,21 +8,20 @@ import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams' import { waitForMobileRelayRpcConnected } from './mobile-relay-rpc-connect-wait' import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel' import { isRpcDeliveryUnknown, markRpcDeliveryUnknown } from './rpc-delivery-ambiguity' +import { + rejectMobileRelayPendingRequests, + type MobileRelayPendingRequest +} from './mobile-relay-pending-requests' import { RpcControlProbeFollowUp } from './rpc-control-probe-follow-up' import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget' import { isRpcResponse } from './rpc-response-shape' import type { RpcClient } from './rpc-client' import type { ConnectionState, RpcResponse } from './types' +import { TimedOutControlRequestIndex } from './timed-out-control-request-index' const CONTROL_PROBE_TIMEOUT_MS = 8_000 const CONTROL_PROBE_INTERVAL_MS = 20_000 -type PendingRequest = { - resolve: (response: RpcResponse) => void - reject: (error: Error) => void - timer: ReturnType -} - export type MobileRelayRpcSession = RpcClient & { getLeaseExpiresAt(): number | null getResumeConfirmation(): DeviceResumeConfirmed | null @@ -40,16 +39,15 @@ export function connectMobileRelayRpcSession(args: { createSocket?: (url: string) => WebSocket }): MobileRelayRpcSession { const requestTimeoutMs = args.requestTimeoutMs ?? 30_000 - const pending = new Map() - const timedOutControlRequestIds = new Set() + const pending = new Map() + const timedOutControlRequestIds = new TimedOutControlRequestIndex() const stateListeners = new Set<(state: ConnectionState) => void>() let state: ConnectionState = 'connecting' let requestCounter = 0 let controlResponseSequence = 0 const controlProbeFollowUp = new RpcControlProbeFollowUp( () => (!closed && state === 'connected' ? true : null), - probeControlPlane, - (hasQueuedFollowUp) => !hasQueuedFollowUp && timedOutControlRequestIds.clear() + probeControlPlane ) let controlProbeTimer: ReturnType | null = null let lastConnectedAt: number | null = null @@ -121,8 +119,9 @@ export function connectMobileRelayRpcSession(args: { } closed = true stopControlProbeTimer() + timedOutControlRequestIds.clear() link.close() - rejectPending(new Error('Client closed')) + rejectMobileRelayPendingRequests(pending, new Error('Client closed')) streams.clear() publishState('disconnected') }, @@ -170,7 +169,7 @@ export function connectMobileRelayRpcSession(args: { return new Promise((resolve, reject) => { const timer = setTimeout(() => { pending.delete(id) - timedOutControlRequestIds.add(id) + timedOutControlRequestIds.remember(id) // Why: the frame was written long ago — the desktop may have processed it. const error = markRpcDeliveryUnknown(new Error(`relay RPC timed out: ${method}`)) reject(error) @@ -213,7 +212,11 @@ export function connectMobileRelayRpcSession(args: { fail(error, new Error(error.message)) return } - if (request || timedOutControlRequestIds.delete(value.id) || streams.isControlResponse(value)) { + if ( + request || + timedOutControlRequestIds.consume(value.id) || + streams.isControlResponse(value) + ) { controlResponseSequence += 1 } if (request) { @@ -297,27 +300,13 @@ export function connectMobileRelayRpcSession(args: { } closed = true stopControlProbeTimer() + timedOutControlRequestIds.clear() failure = error link.close() - rejectPending(pendingError) + rejectMobileRelayPendingRequests(pending, pendingError) publishState(error instanceof MobileE2EEAuthenticationError ? 'auth-failed' : 'disconnected') } - function rejectPending(error: Error): void { - if (pending.size === 0) { - return - } - // Why: pending entries only exist after their frame reached the authenticated - // link (sendFrame failures delete them synchronously), so the desktop may - // have processed them — mark the ambiguity for callers. - markRpcDeliveryUnknown(error) - for (const request of pending.values()) { - clearTimeout(request.timer) - request.reject(error) - } - pending.clear() - } - function nextId(): string { return `relay-rpc-${++requestCounter}-${Date.now()}` } diff --git a/mobile/src/transport/mobile-relay-rpc-streams.ts b/mobile/src/transport/mobile-relay-rpc-streams.ts index 1be5dd6d3a6..d15167e5665 100644 --- a/mobile/src/transport/mobile-relay-rpc-streams.ts +++ b/mobile/src/transport/mobile-relay-rpc-streams.ts @@ -9,7 +9,6 @@ import { } from './rpc-client-terminal-subscription' import type { RpcClient } from './rpc-client' import type { RpcResponse, RpcSuccess } from './types' -import { isStreamControlResponse } from './rpc-stream-response-shape' type StreamRecord = { method: string @@ -23,6 +22,8 @@ type StreamRecord = { streamIds: Set subscriptionId?: string cancelled: boolean + sent: boolean + controlResponseReceived: boolean } type StreamManagerOptions = { @@ -52,14 +53,19 @@ export class MobileRelayRpcStreams { listener, onBinaryFrame: subscribeOptions?.onBinaryFrame, streamIds: new Set(), - cancelled: false + cancelled: false, + sent: false, + controlResponseReceived: false } this.streams.set(id, stream) void this.options .waitForConnected() .then(() => { - if (!stream.cancelled && !this.options.sendFrame({ id, method, params: stream.params })) { - this.fail(id, stream, 'Connection interrupted') + if (!stream.cancelled) { + stream.sent = this.options.sendFrame({ id, method, params: stream.params }) + if (!stream.sent) { + this.fail(id, stream, 'Connection interrupted') + } } }) .catch((error: unknown) => { @@ -108,7 +114,12 @@ export class MobileRelayRpcStreams { } isControlResponse(response: RpcResponse): boolean { - return this.streams.has(response.id) && isStreamControlResponse(response) + const stream = this.streams.get(response.id) + if (!stream?.sent || stream.controlResponseReceived) { + return false + } + stream.controlResponseReceived = true + return true } handleBinary(bytes: Uint8Array): void { diff --git a/mobile/src/transport/rpc-client-request-deadline.test.ts b/mobile/src/transport/rpc-client-request-deadline.test.ts index dee76b3ff04..56eb28f001f 100644 --- a/mobile/src/transport/rpc-client-request-deadline.test.ts +++ b/mobile/src/transport/rpc-client-request-deadline.test.ts @@ -231,6 +231,31 @@ describe('mobile rpc-client request deadline', () => { await request.catch(() => undefined) }) + it('retains late reply evidence after an earlier probe completes', async () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + socket.open() + const request = client.sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + const outcome = track(request) + await vi.advanceTimersByTimeAsync(0) + const timedOutRequest = sentRequest(socket, 'browser.screenshot') + + await vi.advanceTimersByTimeAsync(100) + expect(outcome.read()).toBe('Request timed out: browser.screenshot') + const firstProbe = sentRequest(socket, 'status.get') + socket.receive(`encrypted:${JSON.stringify({ id: firstProbe.id, ok: true, result: {} })}`) + await vi.advanceTimersByTimeAsync(0) + + client.notifyForeground() + socket.receive(`encrypted:${JSON.stringify({ id: timedOutRequest.id, ok: true, result: {} })}`) + await vi.advanceTimersByTimeAsync(8_000) + + expect(socket.close).not.toHaveBeenCalled() + expect(client.getState()).toBe('connected') + client.close() + await request.catch(() => undefined) + }) + it('demotes when the post-timeout control probe also stalls', async () => { const client = connect('ws://desktop.invalid', 'token', 'server-key') const socket = mockSockets[0]! diff --git a/mobile/src/transport/rpc-client.test.ts b/mobile/src/transport/rpc-client.test.ts index fe60dbb7e95..a8a4ee021da 100644 --- a/mobile/src/transport/rpc-client.test.ts +++ b/mobile/src/transport/rpc-client.test.ts @@ -798,10 +798,13 @@ describe('mobile rpc-client connection timeout', () => { }) it.each([ - ['subscribed', { ok: true, streaming: true, result: { type: 'subscribed', streamId: 42 } }], + [ + 'lease-only subscribed', + { ok: true, streaming: true, result: { type: 'subscribed', streamId: null } } + ], ['ready', { ok: true, streaming: true, result: { type: 'ready', subscriptionId: 'sub-1' } }], - ['error', { ok: false, error: { code: 'stream_failed', message: 'stream failed' } }], - ['end', { ok: true, result: { type: 'end' } }] + ['native-chat snapshot', { ok: true, streaming: true, result: { type: 'snapshot' } }], + ['session-tabs snapshot', { ok: true, streaming: true, result: { type: 'snapshot' } }] ])('counts a subscription %s reply as control liveness', async (_name, reply) => { const { client, socket } = connectAuthenticated() client.subscribe('terminal.subscribe', { terminal: 'term-1' }, () => {}) @@ -816,6 +819,27 @@ describe('mobile rpc-client connection timeout', () => { client.close() }) + it('does not count repeated subscription frames as fresh control responses', async () => { + const { client, socket } = connectAuthenticated() + client.subscribe('nativeChat.subscribe', {}, () => {}) + const subscribe = sentRequest(socket, 'nativeChat.subscribe') + const snapshot = { + id: subscribe.id, + ok: true, + streaming: true, + result: { type: 'ready', subscriptionId: 'sub-1' } + } + socket.receive(`encrypted:${JSON.stringify(snapshot)}`) + + client.notifyForeground() + socket.receive(`encrypted:${JSON.stringify(snapshot)}`) + await vi.advanceTimersByTimeAsync(8_000) + + expect(socket.close).toHaveBeenCalled() + expect(client.getState()).toBe('reconnecting') + client.close() + }) + it('does not let terminal stream traffic mask a stalled control channel', async () => { const { client, socket } = connectAuthenticated() const events: unknown[] = [] diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index b70d9a16335..067a6908697 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -38,11 +38,12 @@ import { RpcControlProbeFollowUp } from './rpc-control-probe-follow-up' import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget' import { isRpcResponse } from './rpc-response-shape' import { - isStreamControlResponse, + consumeFirstStreamControlResponse, isStreamingSubscriptionReadyResult, isTerminalSubscribedResult } from './rpc-stream-response-shape' import { websocketPayloadToUint8 } from './websocket-payload-bytes' +import { TimedOutControlRequestIndex } from './timed-out-control-request-index' type PendingRequest = { resolve: (response: RpcResponse) => void @@ -86,7 +87,7 @@ type StreamRequest = { onBinaryFrame?: (frame: BrowserScreencastFrame) => void subscriptionId?: string cancelled?: boolean - sent?: boolean + sent?: 'awaiting' | 'received' } export type RpcClient = { @@ -177,8 +178,7 @@ export function connect( let activityProbeTimer: ReturnType | null = null const activityProbeFollowUp = new RpcControlProbeFollowUp( () => (state === 'connected' ? ws : null), - runActivityProbe, - (hasQueuedFollowUp) => !hasQueuedFollowUp && timedOutControlRequestIds.clear() + runActivityProbe ) let intentionallyClosed = false // Consecutive auth rejections; tolerate up to AUTH_RETRY_BUDGET (issue #5200) before latching to avoid a needless re-pair. @@ -196,7 +196,7 @@ export function connect( const serverPublicKey = publicKeyFromBase64(serverPublicKeyB64) const pending = new Map() - const timedOutControlRequestIds = new Set() + const timedOutControlRequestIds = new TimedOutControlRequestIndex() const streamListeners = new Map() const terminalStreamListeners = new Map() const terminalStreamIdsByRequest = new Map>() @@ -458,7 +458,7 @@ export function connect( if ( sendEncrypted({ id, deviceToken, method: stream.method, params: stream.params }) ) { - stream.sent = true + stream.sent = 'awaiting' } else { emitStreamError(stream, 'Connection interrupted') removeStreamListener(id) @@ -515,8 +515,8 @@ export function connect( } if ( pending.has(response.id) || - timedOutControlRequestIds.delete(response.id) || - (streamListeners.has(response.id) && isStreamControlResponse(response)) + timedOutControlRequestIds.consume(response.id) || + consumeFirstStreamControlResponse(streamListeners.get(response.id)) ) { controlResponseSequence++ } @@ -798,7 +798,7 @@ export function connect( const timeout = setTimeout(() => { timedOut = true pending.delete(id) - timedOutControlRequestIds.add(id) + timedOutControlRequestIds.remember(id) const controlResponded = controlResponseSequence > probeControlResponseSequence activityProbeFollowUp.finish(probeWs) if (controlResponded) { @@ -892,7 +892,7 @@ export function connect( function markStreamsForReplay(): void { for (const [id, stream] of streamListeners) { - stream.sent = false + stream.sent = undefined resetTerminalStreamRoutingForRequest(id) } } @@ -1056,7 +1056,7 @@ export function connect( // Why: the frame was written 30s ago — the host may have processed it. reject(markRpcDeliveryUnknown(new Error(`Request timed out: ${method}`))) if (requestWs === ws) { - timedOutControlRequestIds.add(id) + timedOutControlRequestIds.remember(id) } runActivityProbe(requestWs, true) }, timeoutMs) @@ -1108,7 +1108,7 @@ export function connect( if (state === 'connected') { if (sendEncrypted({ id, deviceToken, method, params })) { - stream.sent = true + stream.sent = 'awaiting' } else { emitStreamError(stream, 'Connection interrupted') removeStreamListener(id) diff --git a/mobile/src/transport/rpc-stream-response-shape.ts b/mobile/src/transport/rpc-stream-response-shape.ts index 9378d876353..df4cb420300 100644 --- a/mobile/src/transport/rpc-stream-response-shape.ts +++ b/mobile/src/transport/rpc-stream-response-shape.ts @@ -1,5 +1,3 @@ -import type { RpcResponse, RpcSuccess } from './types' - export function isTerminalSubscribedResult( value: unknown ): value is { type: 'subscribed'; streamId: number } { @@ -22,23 +20,12 @@ export function isStreamingSubscriptionReadyResult( ) } -export function isStreamControlResponse(response: RpcResponse): boolean { - if (!response.ok) { - return true - } - const result = (response as RpcSuccess).result - if (!result || typeof result !== 'object') { +export function consumeFirstStreamControlResponse( + stream: { sent?: 'awaiting' | 'received' } | undefined +): boolean { + if (stream?.sent !== 'awaiting') { return false } - const metadata = result as { - type?: unknown - subscriptionId?: unknown - streamId?: unknown - } - return ( - metadata.type === 'end' || - metadata.type === 'error' || - typeof metadata.subscriptionId === 'string' || - typeof metadata.streamId === 'number' - ) + stream.sent = 'received' + return true } diff --git a/mobile/src/transport/timed-out-control-request-index.test.ts b/mobile/src/transport/timed-out-control-request-index.test.ts new file mode 100644 index 00000000000..bdf74c06dde --- /dev/null +++ b/mobile/src/transport/timed-out-control-request-index.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from 'vitest' +import { TimedOutControlRequestIndex } from './timed-out-control-request-index' + +describe('TimedOutControlRequestIndex', () => { + it('retains evidence until its bounded expiry', () => { + let now = 100 + const index = new TimedOutControlRequestIndex(() => now, 50) + index.remember('late') + + now = 149 + expect(index.consume('late')).toBe(true) + index.remember('expired') + now = 200 + expect(index.consume('expired')).toBe(false) + }) + + it('evicts the oldest evidence at the per-connection limit', () => { + const index = new TimedOutControlRequestIndex(() => 0, 100, 2) + index.remember('first') + index.remember('second') + index.remember('third') + + expect(index.consume('first')).toBe(false) + expect(index.consume('second')).toBe(true) + expect(index.consume('third')).toBe(true) + }) +}) diff --git a/mobile/src/transport/timed-out-control-request-index.ts b/mobile/src/transport/timed-out-control-request-index.ts new file mode 100644 index 00000000000..7dcbad4f2cd --- /dev/null +++ b/mobile/src/transport/timed-out-control-request-index.ts @@ -0,0 +1,42 @@ +const DEFAULT_EXPIRY_MS = 60_000 +const DEFAULT_MAX_ENTRIES = 256 + +export class TimedOutControlRequestIndex { + private readonly expiryById = new Map() + + constructor( + private readonly now: () => number = Date.now, + private readonly expiryMs = DEFAULT_EXPIRY_MS, + private readonly maxEntries = DEFAULT_MAX_ENTRIES + ) {} + + remember(id: string): void { + this.prune() + while (this.expiryById.size >= this.maxEntries) { + const oldest = this.expiryById.keys().next().value + if (oldest === undefined) { + break + } + this.expiryById.delete(oldest) + } + this.expiryById.set(id, this.now() + this.expiryMs) + } + + consume(id: string): boolean { + this.prune() + return this.expiryById.delete(id) + } + + prune(): void { + const now = this.now() + for (const [id, expiry] of this.expiryById) { + if (expiry <= now) { + this.expiryById.delete(id) + } + } + } + + clear(): void { + this.expiryById.clear() + } +}