diff --git a/mobile/app/h/[hostId]/edit.tsx b/mobile/app/h/[hostId]/edit.tsx index b6aef56672e..13eb40dc032 100644 --- a/mobile/app/h/[hostId]/edit.tsx +++ b/mobile/app/h/[hostId]/edit.tsx @@ -17,11 +17,12 @@ 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 { resolveHostEndpointEdit } from '../../../src/transport/host-endpoint-edit' import { - hostProfileAfterEdit, - resolveHostEndpointEdit -} from '../../../src/transport/host-endpoint-edit' -import { useForceReconnect, usePrimeHosts } from '../../../src/transport/client-context' + useForceReconnect, + useForceReconnectAfterEdit, + usePrimeHosts +} from '../../../src/transport/client-context' import { showForceReconnectError } from '../../../src/transport/force-reconnect-feedback' import type { HostProfile } from '../../../src/transport/types' @@ -31,6 +32,7 @@ export default function EditHostScreen() { const { hostId } = useLocalSearchParams<{ hostId: string }>() const primeHosts = usePrimeHosts() const forceReconnectHost = useForceReconnect() + const forceReconnectHostAfterEdit = useForceReconnectAfterEdit() const [host, setHost] = useState(null) const [loadError, setLoadError] = useState(null) @@ -109,7 +111,7 @@ export default function EditHostScreen() { ...(willRename ? { name: nextName } : {}), ...(nextEndpoint !== undefined ? { endpoint: nextEndpoint } : {}) } - let reconnectProfile = hostProfileAfterEdit(host, updates) + let reconnectProfile: HostProfile | null = null savingRef.current = true setSaving(true) setSaveError(null) @@ -157,7 +159,10 @@ export default function EditHostScreen() { // 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) + const reconnect = reconnectProfile + ? forceReconnectHost(host.id, reconnectProfile) + : forceReconnectHostAfterEdit(host.id, host, updates) + void reconnect.catch(showForceReconnectError) } } diff --git a/mobile/src/host-edit-route-accessibility.test.ts b/mobile/src/host-edit-route-accessibility.test.ts index bb5ddf4faa6..be23c603c5a 100644 --- a/mobile/src/host-edit-route-accessibility.test.ts +++ b/mobile/src/host-edit-route-accessibility.test.ts @@ -5,6 +5,7 @@ import EditHostScreen from '../app/h/[hostId]/edit' const dependencies = vi.hoisted(() => ({ back: vi.fn(), + forceReconnectHostAfterEdit: vi.fn(), forceReconnectHost: vi.fn(), loadHosts: vi.fn(), primeHosts: vi.fn(), @@ -43,6 +44,7 @@ vi.mock('./transport/host-store', () => ({ vi.mock('./transport/client-context', () => ({ useForceReconnect: () => dependencies.forceReconnectHost, + useForceReconnectAfterEdit: () => dependencies.forceReconnectHostAfterEdit, usePrimeHosts: () => dependencies.primeHosts })) diff --git a/mobile/src/host-edit-save-flow.test.ts b/mobile/src/host-edit-save-flow.test.ts index b895b7b6f89..dba79ee1ced 100644 --- a/mobile/src/host-edit-save-flow.test.ts +++ b/mobile/src/host-edit-save-flow.test.ts @@ -7,6 +7,7 @@ const dependencies = vi.hoisted(() => ({ alert: vi.fn(), back: vi.fn(), forceReconnectHost: vi.fn(), + forceReconnectHostAfterEdit: vi.fn(), getHostListLoadRevision: vi.fn(() => 0), loadHosts: vi.fn(), primeHosts: vi.fn(), @@ -51,6 +52,7 @@ vi.mock('./transport/host-list-load-sharing', () => ({ vi.mock('./transport/client-context', () => ({ useForceReconnect: () => dependencies.forceReconnectHost, + useForceReconnectAfterEdit: () => dependencies.forceReconnectHostAfterEdit, usePrimeHosts: () => dependencies.primeHosts })) @@ -147,6 +149,7 @@ describe('edit host handleSave', () => { dependencies.alert.mockReset() dependencies.back.mockReset() dependencies.forceReconnectHost.mockReset().mockResolvedValue(undefined) + dependencies.forceReconnectHostAfterEdit.mockReset().mockResolvedValue(undefined) dependencies.getHostListLoadRevision.mockReset().mockReturnValue(0) dependencies.loadHosts.mockReset().mockResolvedValue([HOST_FIXTURE]) dependencies.primeHosts.mockReset() @@ -264,8 +267,23 @@ describe('edit host handleSave', () => { await pressSave(renderer) expect(dependencies.primeHosts).not.toHaveBeenCalled() - expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', { - ...HOST_FIXTURE, + expect(dependencies.forceReconnectHost).not.toHaveBeenCalled() + expect(dependencies.forceReconnectHostAfterEdit).toHaveBeenCalledWith('host-1', HOST_FIXTURE, { + endpoint: 'ws://192.168.1.20:6768' + }) + expect(dependencies.back).toHaveBeenCalledTimes(1) + + act(() => renderer.unmount()) + }) + + it('merges the endpoint edit into the current cache when the host reload omits it', async () => { + dependencies.loadHosts.mockResolvedValueOnce([HOST_FIXTURE]).mockResolvedValueOnce([]) + const renderer = await renderEditHostRoute() + setFieldValue(renderer, 'Address', '192.168.1.20:6768') + await pressSave(renderer) + + expect(dependencies.forceReconnectHost).not.toHaveBeenCalled() + expect(dependencies.forceReconnectHostAfterEdit).toHaveBeenCalledWith('host-1', HOST_FIXTURE, { endpoint: 'ws://192.168.1.20:6768' }) expect(dependencies.back).toHaveBeenCalledTimes(1) @@ -297,8 +315,16 @@ describe('edit host handleSave', () => { ...HOST_FIXTURE, endpoint: 'ws://192.168.1.20:6768', endpoints: [ - { id: 'direct-primary', kind: 'lan' as const, url: 'ws://192.168.1.20:6768' }, - { id: 'relay-primary', kind: 'relay' as const, url: 'wss://relay.invalid/host-1' } + { + id: 'direct-primary', + kind: 'lan' as const, + url: 'ws://192.168.1.20:6768' + }, + { + id: 'relay-primary', + kind: 'relay' as const, + url: 'wss://relay.invalid/host-1' + } ] } dependencies.loadHosts.mockResolvedValueOnce([HOST_FIXTURE]).mockResolvedValueOnce([relayHost]) @@ -370,6 +396,7 @@ describe('edit host load() error states', () => { dependencies.alert.mockReset() dependencies.back.mockReset() dependencies.forceReconnectHost.mockReset().mockResolvedValue(undefined) + dependencies.forceReconnectHostAfterEdit.mockReset().mockResolvedValue(undefined) dependencies.getHostListLoadRevision.mockReset().mockReturnValue(0) dependencies.loadHosts.mockReset().mockResolvedValue([HOST_FIXTURE]) dependencies.primeHosts.mockReset() diff --git a/mobile/src/transport/client-context-actions.ts b/mobile/src/transport/client-context-actions.ts index 86900550d85..f677d3d8dde 100644 --- a/mobile/src/transport/client-context-actions.ts +++ b/mobile/src/transport/client-context-actions.ts @@ -2,4 +2,5 @@ import { useRpcClientContext } from './rpc-client-react-context' export const useCloseHost = () => useRpcClientContext().closeHost export const useForceReconnect = () => useRpcClientContext().forceReconnect +export const useForceReconnectAfterEdit = () => useRpcClientContext().forceReconnectAfterEdit export const usePrimeHosts = () => useRpcClientContext().primeHosts diff --git a/mobile/src/transport/client-context.tsx b/mobile/src/transport/client-context.tsx index 133a72dea13..0172fc83bbe 100644 --- a/mobile/src/transport/client-context.tsx +++ b/mobile/src/transport/client-context.tsx @@ -193,13 +193,8 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { entry.refCount = Math.max(0, entry.refCount - 1) }, []) - const forceReconnect = useCallback( - (hostId: string, requestedHost?: HostProfile): Promise => { - const profile = primedHostsRef.current.reconnectProfile( - hostId, - getHostListLoadRevision(), - requestedHost - ) + const runForceReconnect = useCallback( + (hostId: string, profile: HostOpenProfile): Promise => { return forceReconnectCoordinatorRef.current.run({ hostId, profileVersion: profile.version, @@ -220,6 +215,35 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { [openEntry] ) + const forceReconnect = useCallback( + (hostId: string, requestedHost?: HostProfile): Promise => { + const profile = primedHostsRef.current.reconnectProfile( + hostId, + getHostListLoadRevision(), + requestedHost + ) + return runForceReconnect(hostId, profile) + }, + [runForceReconnect] + ) + + const forceReconnectAfterEdit = useCallback( + ( + hostId: string, + fallbackHost: HostProfile, + updates: { name?: string; endpoint?: string } + ): Promise => { + const profile = primedHostsRef.current.reconnectEditedProfile( + hostId, + getHostListLoadRevision(), + fallbackHost, + updates + ) + return runForceReconnect(hostId, profile) + }, + [runForceReconnect] + ) + const getState = useCallback((hostId: string): ConnectionState => { return storeRef.current.get(hostId)?.state ?? 'disconnected' }, []) @@ -262,7 +286,10 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { [] ) - const getAllClients = useCallback((): Array<{ hostId: string; client: RpcClient }> => { + const getAllClients = useCallback((): Array<{ + hostId: string + client: RpcClient + }> => { const out: Array<{ hostId: string; client: RpcClient }> = [] for (const [hostId, entry] of storeRef.current) { out.push({ hostId, client: entry.client }) @@ -302,6 +329,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { acquire, release, forceReconnect, + forceReconnectAfterEdit, closeHost: closeEntry, getState, getReconnectAttempt, @@ -317,6 +345,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { acquire, release, forceReconnect, + forceReconnectAfterEdit, closeEntry, getState, getReconnectAttempt, @@ -394,5 +423,10 @@ export function useHostClient(hostId: string | undefined): { } // Why: host-store's removeHost() must close the live client but has no React-side handle; this hook bridges to it. -export { useCloseHost, useForceReconnect, usePrimeHosts } from './client-context-actions' +export { + useCloseHost, + useForceReconnect, + useForceReconnectAfterEdit, + usePrimeHosts +} from './client-context-actions' export { useAllHostClients } from './client-context-all-host-clients' diff --git a/mobile/src/transport/host-endpoint-edit.ts b/mobile/src/transport/host-endpoint-edit.ts index 670a6a2150b..39abf278edd 100644 --- a/mobile/src/transport/host-endpoint-edit.ts +++ b/mobile/src/transport/host-endpoint-edit.ts @@ -6,6 +6,8 @@ import { } from './host-endpoint' import type { HostProfile } from './types' +export type HostProfileEdit = { name?: string; endpoint?: string } + export type HostEndpointEditResolution = | { kind: 'unchanged'; endpoint: string } | { kind: 'changed'; endpoint: string } @@ -44,10 +46,7 @@ export function resolveHostEndpointEdit( } } -export function hostProfileAfterEdit( - host: HostProfile, - updates: { name?: string; endpoint?: string } -): HostProfile { +export function hostProfileAfterEdit(host: HostProfile, updates: HostProfileEdit): HostProfile { const updatedEndpoint = updates.endpoint const endpoints = updatedEndpoint === undefined || !host.endpoints diff --git a/mobile/src/transport/host-profile-publication.ts b/mobile/src/transport/host-profile-publication.ts index 9668ffefcc9..783945e4a07 100644 --- a/mobile/src/transport/host-profile-publication.ts +++ b/mobile/src/transport/host-profile-publication.ts @@ -14,6 +14,10 @@ export function getHostProfilePublicationRevision(hostId: string): number { return revisionByHost.get(hostId) ?? 0 } +export function recordHostProfileMutation(hostId: string): void { + revisionByHost.set(hostId, getHostProfilePublicationRevision(hostId) + 1) +} + export function getPublishedHostIdentity( hostId: string ): Pick | undefined { @@ -43,7 +47,10 @@ export function beginHostEndpointPublicationLifecycle( ): HostEndpointPublicationLifecycle { const generation = (endpointGenerationByHost.get(hostId) ?? 0) + 1 endpointGenerationByHost.set(hostId, generation) - return { generation, profileRevision: getHostProfilePublicationRevision(hostId) } + return { + generation, + profileRevision: getHostProfilePublicationRevision(hostId) + } } export function getHostEndpointPublicationLifecycle( @@ -77,9 +84,9 @@ export function serializeHostProfilePublication( export function publishHostProfileTransaction( host: HostProfile, beforeHostSave: (() => Promise) | null, - saveHost: (host: HostProfile) => Promise + saveHost: (host: HostProfile) => Promise, + publicationRevision = getHostProfilePublicationRevision(host.id) ): Promise { - const publicationRevision = getHostProfilePublicationRevision(host.id) return serializeHostProfilePublication(host.id, async () => { requirePublicationRevision(host.id, publicationRevision) await beforeHostSave?.() diff --git a/mobile/src/transport/host-reconnect-profile-cache.test.ts b/mobile/src/transport/host-reconnect-profile-cache.test.ts index 39906a159f3..2a0acebd9a4 100644 --- a/mobile/src/transport/host-reconnect-profile-cache.test.ts +++ b/mobile/src/transport/host-reconnect-profile-cache.test.ts @@ -18,7 +18,11 @@ describe('HostReconnectProfileCache', () => { ...HOST, endpoints: [ { id: 'direct-primary', kind: 'lan', url: HOST.endpoint }, - { id: 'relay-primary', kind: 'relay', url: 'wss://relay.invalid/v1/connect/host' } + { + id: 'relay-primary', + kind: 'relay', + url: 'wss://relay.invalid/v1/connect/host' + } ], relayHostId: 'AbCdEf0123_-xyZ9', relay: { @@ -95,4 +99,43 @@ describe('HostReconnectProfileCache', () => { expect(cache.get(HOST.id, 4)).toEqual(editedHost) }) + + it('merges a fallback endpoint edit into the latest cached relay routing', () => { + const cache = new HostReconnectProfileCache() + const relayHost: HostProfile = { + ...HOST, + endpoints: [ + { id: 'direct-primary', kind: 'lan', url: HOST.endpoint }, + { + id: 'relay-primary', + kind: 'relay', + url: 'wss://relay.invalid/v1/connect/host' + } + ], + relayHostId: 'AbCdEf0123_-xyZ9', + relay: { + v: 1, + directorUrl: 'https://relay.invalid', + cellUrl: 'https://relay.invalid', + assignmentEpoch: 1, + relayHostId: 'AbCdEf0123_-xyZ9', + e2eeFraming: 2 + } + } + cache.prime(HOST, 4) + cache.prime(relayHost, 5) + + const edited = cache.reconnectEditedProfile(HOST.id, 6, HOST, { + endpoint: 'ws://127.0.0.1:2' + }) + + expect(edited.host).toEqual({ + ...relayHost, + endpoint: 'ws://127.0.0.1:2', + endpoints: [ + { id: 'direct-primary', kind: 'lan', url: 'ws://127.0.0.1:2' }, + relayHost.endpoints![1] + ] + }) + }) }) diff --git a/mobile/src/transport/host-reconnect-profile-cache.ts b/mobile/src/transport/host-reconnect-profile-cache.ts index 12e72928a7a..ee8d4e4d534 100644 --- a/mobile/src/transport/host-reconnect-profile-cache.ts +++ b/mobile/src/transport/host-reconnect-profile-cache.ts @@ -1,4 +1,5 @@ import type { HostProfile } from './types' +import { hostProfileAfterEdit, type HostProfileEdit } from './host-endpoint-edit' type CachedHostProfile = { host: HostProfile @@ -31,7 +32,12 @@ export class HostReconnectProfileCache { ? previous.publicationVersion : (this.publicationVersions.get(host.id) ?? 0) + 1 this.publicationVersions.set(host.id, publicationVersion) - this.profiles.set(host.id, { host, version, publicationVersion, sourceRevision }) + this.profiles.set(host.id, { + host, + version, + publicationVersion, + sourceRevision + }) return version } @@ -53,6 +59,22 @@ export class HostReconnectProfileCache { } } + reconnectEditedProfile( + hostId: string, + currentRevision: number, + fallbackHost: HostProfile, + updates: HostProfileEdit + ): HostOpenProfile { + const cached = this.profiles.get(hostId)?.host + const base = cached && reconnectIdentityMatches(cached, fallbackHost) ? cached : fallbackHost + const host = hostProfileAfterEdit(base, updates) + return { + host, + sourceRevision: currentRevision, + version: this.prime(host, currentRevision) + } + } + openProfile( hostId: string, currentRevision: number, @@ -92,7 +114,12 @@ export class HostReconnectProfileCache { const publicationVersion = sourcePublicationVersion + 1 this.latestVersions.set(host.id, version) this.publicationVersions.set(host.id, publicationVersion) - this.profiles.set(host.id, { host, version, publicationVersion, sourceRevision }) + this.profiles.set(host.id, { + host, + version, + publicationVersion, + sourceRevision + }) return version } @@ -154,3 +181,11 @@ function reconnectProfileMatches(left: HostProfile, right: HostProfile): boolean JSON.stringify(left.relay ?? null) === JSON.stringify(right.relay ?? null) ) } + +function reconnectIdentityMatches(left: HostProfile, right: HostProfile): boolean { + return ( + left.id === right.id && + left.deviceToken === right.deviceToken && + left.publicKeyB64 === right.publicKeyB64 + ) +} diff --git a/mobile/src/transport/host-store.test.ts b/mobile/src/transport/host-store.test.ts index 560df5c0758..0df45253698 100644 --- a/mobile/src/transport/host-store.test.ts +++ b/mobile/src/transport/host-store.test.ts @@ -44,6 +44,7 @@ import { updateLastConnected } from './host-store' import { + MobileRelayUpgradeLifecycleRetiredError, MobileRelayUpgradeHostRemovedError, MobileRelayUpgradeHostSupersededError, saveExistingHostRelayRouting, @@ -463,13 +464,8 @@ describe('host-store list mutations', () => { }) releaseOldToken?.('token-1') - await stalePublication - const [overlay] = JSON.parse(storedOverlayRaw ?? '[]') as MobileRelayHostOverlay[] - expect(overlay?.endpoints).toContainEqual({ - id: 'direct-primary', - kind: 'lan', - url: 'ws://127.0.0.1:9' - }) + await expect(stalePublication).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError) + expect(JSON.parse(storedOverlayRaw ?? '[]')).toEqual([]) }) it('lets same-host re-pair supersede a publication stalled on identity lookup', async () => { diff --git a/mobile/src/transport/host-store.ts b/mobile/src/transport/host-store.ts index ca1d0bb025c..b0ff35e3620 100644 --- a/mobile/src/transport/host-store.ts +++ b/mobile/src/transport/host-store.ts @@ -22,6 +22,7 @@ import { import { scheduleOrphanedMobileRelayCleanup } from './mobile-relay-orphan-cleanup' import { recordDurableHostIdentity, + recordHostProfileMutation, resetHostProfilePublicationForTests, retireHostProfilePublication, serializeHostProfilePublication @@ -303,6 +304,7 @@ export async function updateHostNameAndEndpoint( ? undefined : () => updateMobileRelayHostOverlayDirectEndpoint(hostId, updates.endpoint!) ) + recordHostProfileMutation(hostId) }) } diff --git a/mobile/src/transport/mobile-relay-pairing-recovery.test.ts b/mobile/src/transport/mobile-relay-pairing-recovery.test.ts index 93611f3912b..e980393fcf0 100644 --- a/mobile/src/transport/mobile-relay-pairing-recovery.test.ts +++ b/mobile/src/transport/mobile-relay-pairing-recovery.test.ts @@ -7,10 +7,17 @@ import { } from './mobile-relay-pairing-recovery' import type { PairingCandidateClient } from './mobile-relay-physical-client' import type { PairingOffer, RpcResponse } from './types' +import { + recordHostProfileMutation, + resetHostProfilePublicationForTests, + retireHostProfilePublication +} from './host-profile-publication' vi.mock('react-native', () => ({ Platform: { OS: 'ios' } })) vi.mock('expo-crypto', () => ({ getRandomBytes: vi.fn() })) -vi.mock('expo-secure-store', () => ({ WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'WHEN_UNLOCKED' })) +vi.mock('expo-secure-store', () => ({ + WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'WHEN_UNLOCKED' +})) vi.mock('@react-native-async-storage/async-storage', () => ({ default: {} })) const now = Date.UTC(2026, 6, 13) @@ -62,13 +69,19 @@ function endpoints( relayHostId: offer.relay.relayHostId, e2eeFraming: 2 as const }, - installStatus: { v: 1 as const, reqId: journal.metadata.installReqId, ...state } + installStatus: { + v: 1 as const, + reqId: journal.metadata.installReqId, + ...state + } } } function journal(mode: 'authenticated-direct' | 'relay-basis' = 'authenticated-direct') { const value = createMobileRelayPairingJournal({ - offer: offer as PairingOffer & { relay: NonNullable }, + offer: offer as PairingOffer & { + relay: NonNullable + }, hostId: 'host-1', hostName: 'Blue Whale', now, @@ -85,7 +98,10 @@ function journal(mode: 'authenticated-direct' | 'relay-basis' = 'authenticated-d } function client(handler: (method: string, params: unknown) => Promise) { - return { sendRequest: vi.fn(handler), close: vi.fn() } satisfies PairingCandidateClient + return { + sendRequest: vi.fn(handler), + close: vi.fn() + } satisfies PairingCandidateClient } function dependencies(args: { @@ -116,6 +132,7 @@ function dependencies(args: { describe('mobile relay pairing recovery', () => { beforeEach(() => { resetMobileRelayPairingRecoveryForTests() + resetHostProfilePublicationForTests() }) it('recovers a lost direct-install response with the pending credential first', async () => { @@ -224,4 +241,95 @@ describe('mobile relay pairing recovery', () => { expect(saved.metadata.authorizationMode).toBe('authenticated-direct') expect(deps.updateJournal).toHaveBeenCalledTimes(2) }) + + it('does not resurrect a host removed while startup recovery waits on Relay', async () => { + const saved = journal() + const committed = installed(saved, 'authenticated-direct') + let releaseStatus: () => void = () => {} + const statusPending = new Promise((resolve) => { + releaseStatus = resolve + }) + const statusStarted = vi.fn() + const pending = client(async () => { + statusStarted() + await statusPending + return response(endpoints(saved, { state: 'committed', result: committed })) + }) + const deps = dependencies({ + journal: saved, + connectRelay: vi.fn(() => pending) + }) + const recovery = recoverMobileRelayPairing(deps) + await vi.waitFor(() => expect(statusStarted).toHaveBeenCalledOnce()) + + retireHostProfilePublication(saved.metadata.host.id) + releaseStatus() + + await expect(recovery).resolves.toBe('deferred') + expect(deps.writeCredentialBundle).not.toHaveBeenCalled() + expect(deps.saveHost).not.toHaveBeenCalled() + expect(deps.onHostPublished).not.toHaveBeenCalled() + expect(deps.clearJournal).not.toHaveBeenCalled() + }) + + it('does not overwrite a host edited while startup recovery waits on Relay', async () => { + const saved = journal() + const committed = installed(saved, 'authenticated-direct') + let releaseStatus: () => void = () => {} + const statusPending = new Promise((resolve) => { + releaseStatus = resolve + }) + const statusStarted = vi.fn() + const pending = client(async () => { + statusStarted() + await statusPending + return response(endpoints(saved, { state: 'committed', result: committed })) + }) + const deps = dependencies({ + journal: saved, + connectRelay: vi.fn(() => pending) + }) + const recovery = recoverMobileRelayPairing(deps) + await vi.waitFor(() => expect(statusStarted).toHaveBeenCalledOnce()) + + recordHostProfileMutation(saved.metadata.host.id) + releaseStatus() + + await expect(recovery).resolves.toBe('deferred') + expect(deps.writeCredentialBundle).not.toHaveBeenCalled() + expect(deps.saveHost).not.toHaveBeenCalled() + expect(deps.onHostPublished).not.toHaveBeenCalled() + }) + + it('preserves current durable host fields when retrying a retained journal', async () => { + const saved = journal() + const committed = installed(saved, 'authenticated-direct') + const editedHost = { + ...saved.metadata.host, + name: 'Edited name', + endpoint: 'ws://192.168.1.99:6768', + deviceToken: saved.secrets.deviceToken, + lastConnected: 42 + } + const pending = client(async () => + response(endpoints(saved, { state: 'committed', result: committed })) + ) + const deps = dependencies({ + journal: saved, + connectRelay: vi.fn(() => pending) + }) + deps.loadHosts.mockResolvedValue([editedHost]) + + await expect(recoverMobileRelayPairing(deps)).resolves.toBe('recovered') + expect(deps.saveHost).toHaveBeenCalledWith( + expect.objectContaining({ + name: editedHost.name, + endpoint: editedHost.endpoint, + lastConnected: editedHost.lastConnected, + endpoints: expect.arrayContaining([ + { id: 'direct-primary', kind: 'lan', url: editedHost.endpoint } + ]) + }) + ) + }) }) diff --git a/mobile/src/transport/mobile-relay-pairing-recovery.ts b/mobile/src/transport/mobile-relay-pairing-recovery.ts index 9a98691a497..88ce5661606 100644 --- a/mobile/src/transport/mobile-relay-pairing-recovery.ts +++ b/mobile/src/transport/mobile-relay-pairing-recovery.ts @@ -2,8 +2,7 @@ import { Platform } from 'react-native' import { DeviceCredentialInstalledSchema, PairingGetEndpointsResultSchema, - type DeviceCredentialInstalled, - type MobileRelayEndpoint + type DeviceCredentialInstalled } from '../../../src/shared/mobile-relay-credential-contract' import type { PairingRelay } from '../../../src/shared/mobile-relay-pairing-offer' import { loadHosts, saveHost } from './host-store' @@ -26,7 +25,11 @@ import { } from './mobile-relay-physical-client' import { createRecoveringPairingRelayCandidate } from './pairing-relay-candidate' import type { HostProfile, RpcResponse } from './types' -import { publishHostProfileTransaction } from './host-profile-publication' +import { createMobileRelayRecoveredHost } from './mobile-relay-recovered-host' +import { + getHostProfilePublicationRevision, + publishHostProfileTransaction +} from './host-profile-publication' export type MobileRelayPairingRecoveryResult = 'none' | 'recovered' | 'deferred' @@ -90,9 +93,11 @@ async function runRecovery( if (!journal) { return 'none' } - const bundle = await dependencies.readCredentialBundle(journal.metadata.host.id).catch(() => null) + const hostId = journal.metadata.host.id + const publicationRevision = getHostProfilePublicationRevision(hostId) + const bundle = await dependencies.readCredentialBundle(hostId).catch(() => null) const hosts = await dependencies.loadHosts().catch(() => []) - const existing = hosts.find(({ id }) => id === journal!.metadata.host.id) + const existing = hosts.find(({ id }) => id === hostId) if (existing?.relayHostId === journal.metadata.relay.relayHostId && bundle) { await dependencies.clearJournal(journal.metadata.journalId) return 'recovered' @@ -116,7 +121,7 @@ async function runRecovery( }) const endpoints = await getRecoveryStatus(client, journal, credential.kind) if (endpoints.installStatus?.state === 'committed') { - await publishCommitted(journal, endpoints, dependencies) + await publishCommitted(journal, endpoints, dependencies, publicationRevision, existing) return 'recovered' } if (credential.kind === 'invite' && endpoints.installStatus?.state === 'not-found') { @@ -131,7 +136,7 @@ async function runRecovery( ) const reconciled = await getRecoveryStatus(client, journal, 'invite') assertCommitted(reconciled, installed) - await publishCommitted(journal, reconciled, dependencies) + await publishCommitted(journal, reconciled, dependencies, publicationRevision, existing) return 'recovered' } } catch { @@ -229,7 +234,9 @@ async function transitionToInviteAuthorization( async function publishCommitted( journal: MobileRelayPairingJournal, endpoints: ReturnType, - dependencies: RecoveryDependencies + dependencies: RecoveryDependencies, + publicationRevision: number, + existing: HostProfile | undefined ): Promise { if (endpoints.installStatus?.state !== 'committed' || !endpoints.relay) { throw new Error('relay pairing recovery was not committed') @@ -246,38 +253,28 @@ async function publishCommitted( if (journal.metadata.authorizationMode !== installed.authorizationMode) { await dependencies.updateJournal(journal.metadata.journalId, () => reconciledJournal.metadata) } - const host = relayHost(reconciledJournal, endpoints.relay) + const host = createMobileRelayRecoveredHost(reconciledJournal, endpoints.relay, existing) await publishHostProfileTransaction( host, () => dependencies.writeCredentialBundle( - promotePairingJournalCredential({ journal: reconciledJournal, installed }) + promotePairingJournalCredential({ + journal: reconciledJournal, + installed + }) ), - dependencies.saveHost + dependencies.saveHost, + publicationRevision ) dependencies.onHostPublished(host) await dependencies.clearJournal(journal.metadata.journalId) } -function relayHost(journal: MobileRelayPairingJournal, relay: MobileRelayEndpoint): HostProfile { - const host = journal.metadata.host - const url = new URL(relay.cellUrl) - url.protocol = 'wss:' - url.pathname = `/v1/connect/${encodeURIComponent(relay.relayHostId)}` - return { - ...host, - deviceToken: journal.secrets.deviceToken, - endpoints: [ - { id: 'direct-primary', kind: 'lan', url: host.endpoint }, - { id: 'relay-primary', kind: 'relay', url: url.toString() } - ], - relayHostId: relay.relayHostId, - relay - } -} - function pairingRelay(journal: MobileRelayPairingJournal): PairingRelay { - return { ...journal.metadata.relay, inviteToken: journal.secrets.inviteToken } + return { + ...journal.metadata.relay, + inviteToken: journal.secrets.inviteToken + } } function requireSuccess(response: RpcResponse): unknown { diff --git a/mobile/src/transport/mobile-relay-recovered-host.ts b/mobile/src/transport/mobile-relay-recovered-host.ts new file mode 100644 index 00000000000..656b2b19a8e --- /dev/null +++ b/mobile/src/transport/mobile-relay-recovered-host.ts @@ -0,0 +1,27 @@ +import type { MobileRelayEndpoint } from '../../../src/shared/mobile-relay-credential-contract' +import type { MobileRelayPairingJournal } from './mobile-relay-pairing-journal' +import { relayWebSocketUrl } from './mobile-endpoint-supervisor-support' +import type { HostProfile } from './types' + +export function createMobileRelayRecoveredHost( + journal: MobileRelayPairingJournal, + relay: MobileRelayEndpoint, + existing: HostProfile | undefined +): HostProfile { + const journalHost = journal.metadata.host + const host = + existing?.publicKeyB64 === journalHost.publicKeyB64 && + existing.deviceToken === journal.secrets.deviceToken + ? existing + : journalHost + return { + ...host, + deviceToken: journal.secrets.deviceToken, + endpoints: [ + { id: 'direct-primary', kind: 'lan', url: host.endpoint }, + { id: 'relay-primary', kind: 'relay', url: relayWebSocketUrl(relay) } + ], + relayHostId: relay.relayHostId, + relay + } +} diff --git a/mobile/src/transport/mobile-relay-rpc-session-application-responsiveness.test.ts b/mobile/src/transport/mobile-relay-rpc-session-application-responsiveness.test.ts new file mode 100644 index 00000000000..6c92adee193 --- /dev/null +++ b/mobile/src/transport/mobile-relay-rpc-session-application-responsiveness.test.ts @@ -0,0 +1,130 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const fakes = vi.hoisted(() => ({ + linkOptions: null as null | { + onHello(value: unknown): void + onAuthenticated(): void + onText(value: string): void + }, + sendText: vi.fn(() => true), + close: vi.fn() +})) + +vi.mock('./mobile-relay-e2ee-link', () => ({ + MobileRelayE2eeLink: class { + constructor(options: NonNullable) { + fakes.linkOptions = options + } + sendText = fakes.sendText + close = fakes.close + } +})) + +import { connectMobileRelayRpcSession } from './mobile-relay-rpc-session' + +const relay = { + v: 1 as const, + directorUrl: 'https://relay.onorca.dev', + cellUrl: 'https://relay-c1.onorca.dev', + assignmentEpoch: 7, + relayHostId: 'AbCdEf0123_-xyZ9', + e2eeFraming: 2 as const +} + +async function authenticateSession() { + const session = connectMobileRelayRpcSession({ + relay, + resumeToken: 'resume-secret', + resumeCredentialVersion: 3, + resumeConfirmReqId: 'confirm-1', + deviceToken: 'device-token', + desktopPublicKeyB64: 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=', + requestTimeoutMs: 1000 + }) + fakes.linkOptions!.onHello({ + type: 'relay-hello', + ok: true, + credentialKind: 'resume', + leaseExpiresAt: Date.now() + 60_000, + acceptedCredentialVersion: 3, + acceptedAs: 'current', + resumeExpiresAt: Date.now() + 300_000 + }) + 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 } + fakes.linkOptions!.onText( + JSON.stringify({ + id: request.id, + ok: true, + result: { + v: 1, + relay, + resumeConfirmation: { + v: 1, + reqId: 'confirm-1', + currentVersion: 3, + acceptedAs: 'current', + renewed: true, + resumeExpiresAt: Date.now() + 300_000 + } + }, + _meta: { runtimeId: 'runtime-1' } + }) + ) + await vi.waitFor(() => expect(session.getState()).toBe('connected')) + fakes.sendText.mockClear() + return session +} + +describe('mobile relay subscription application responsiveness', () => { + beforeEach(() => { + vi.clearAllMocks() + fakes.linkOptions = null + fakes.sendText.mockReturnValue(true) + }) + + it('clears an application stall when a subscription answers', async () => { + const session = await authenticateSession() + vi.useFakeTimers() + try { + const first = session + .sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + .catch((error: unknown) => error) + await vi.advanceTimersByTimeAsync(100) + await expect(first).resolves.toMatchObject({ + message: 'relay RPC timed out: browser.screenshot' + }) + expect(session.getRpcUnresponsiveSince?.()).not.toBeNull() + + session.subscribe('terminal.subscribe', { terminal: 'term-1' }, 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 === 'terminal.subscribe')! + fakes.linkOptions!.onText( + JSON.stringify({ + id: subscribe.id, + ok: true, + streaming: true, + result: { type: 'subscribed', streamId: 42 }, + _meta: {} + }) + ) + expect(session.getRpcUnresponsiveSince?.()).toBeNull() + + const second = session + .sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + .catch((error: unknown) => error) + await vi.advanceTimersByTimeAsync(100) + await expect(second).resolves.toMatchObject({ + message: 'relay RPC timed out: browser.screenshot' + }) + 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 baeabc54a36..56b6cb003d1 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -52,7 +52,8 @@ export function connectMobileRelayRpcSession( const streams = new MobileRelayRpcStreams({ nextId, sendFrame, - waitForConnected: () => waitForConnected() + waitForConnected: () => waitForConnected(), + onApplicationResponse: (method) => applicationResponseTracker.recordResponse(method) }) const link = new MobileRelayE2eeLink({ diff --git a/mobile/src/transport/mobile-relay-rpc-streams.ts b/mobile/src/transport/mobile-relay-rpc-streams.ts index d15167e5665..25f3a3587d6 100644 --- a/mobile/src/transport/mobile-relay-rpc-streams.ts +++ b/mobile/src/transport/mobile-relay-rpc-streams.ts @@ -30,6 +30,7 @@ type StreamManagerOptions = { nextId: () => string sendFrame: (request: { id: string; method: string; params?: unknown }) => boolean waitForConnected: () => Promise + onApplicationResponse?: (method: string) => void } export class MobileRelayRpcStreams { @@ -62,7 +63,11 @@ export class MobileRelayRpcStreams { .waitForConnected() .then(() => { if (!stream.cancelled) { - stream.sent = this.options.sendFrame({ id, method, params: stream.params }) + stream.sent = this.options.sendFrame({ + id, + method, + params: stream.params + }) if (!stream.sent) { this.fail(id, stream, 'Connection interrupted') } @@ -90,7 +95,11 @@ export class MobileRelayRpcStreams { } const result = (response as RpcSuccess).result if (result && typeof result === 'object') { - const metadata = result as { subscriptionId?: unknown; streamId?: unknown; type?: unknown } + const metadata = result as { + subscriptionId?: unknown + streamId?: unknown + type?: unknown + } if (typeof metadata.subscriptionId === 'string') { stream.subscriptionId = metadata.subscriptionId } @@ -119,6 +128,9 @@ export class MobileRelayRpcStreams { return false } stream.controlResponseReceived = true + if (response.ok) { + this.options.onApplicationResponse?.(stream.method) + } return true } diff --git a/mobile/src/transport/rpc-client-application-responsiveness.test.ts b/mobile/src/transport/rpc-client-application-responsiveness.test.ts index 0769ec821a6..1ca090b2d78 100644 --- a/mobile/src/transport/rpc-client-application-responsiveness.test.ts +++ b/mobile/src/transport/rpc-client-application-responsiveness.test.ts @@ -93,7 +93,9 @@ describe('direct RPC application responsiveness', () => { .catch((error: unknown) => error) await vi.advanceTimersByTimeAsync(100) - await expect(first).resolves.toMatchObject({ message: 'Request timed out: browser.screenshot' }) + await expect(first).resolves.toMatchObject({ + message: 'Request timed out: browser.screenshot' + }) const probe = sentRequest(socket, 'status.get') socket.receive(`encrypted:${JSON.stringify({ id: probe.id, ok: true, result: {} })}`) await vi.advanceTimersByTimeAsync(0) @@ -121,4 +123,42 @@ describe('direct RPC application responsiveness', () => { expect(client.getState()).toBe('reconnecting') client.close() }) + + it('clears an application stall when a subscription answers', async () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = sockets[0]! + socket.openAndAuthenticate() + const first = client + .sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + .catch((error: unknown) => error) + + await vi.advanceTimersByTimeAsync(100) + await expect(first).resolves.toMatchObject({ + message: 'Request timed out: browser.screenshot' + }) + expect(client.getRpcUnresponsiveSince?.()).not.toBeNull() + + client.subscribe('terminal.subscribe', { terminal: 'term-1' }, () => {}) + const subscribe = sentRequest(socket, 'terminal.subscribe') + socket.receive( + `encrypted:${JSON.stringify({ + id: subscribe.id, + ok: true, + streaming: true, + result: { type: 'subscribed', streamId: 42 } + })}` + ) + expect(client.getRpcUnresponsiveSince?.()).toBeNull() + + const second = client + .sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + .catch((error: unknown) => error) + await vi.advanceTimersByTimeAsync(100) + await expect(second).resolves.toMatchObject({ + message: 'Request timed out: browser.screenshot' + }) + expect(socket.close).not.toHaveBeenCalled() + expect(client.getState()).toBe('connected') + client.close() + }) }) diff --git a/mobile/src/transport/rpc-client-context-contract.ts b/mobile/src/transport/rpc-client-context-contract.ts index e578fcf8b73..ef8a70a06cb 100644 --- a/mobile/src/transport/rpc-client-context-contract.ts +++ b/mobile/src/transport/rpc-client-context-contract.ts @@ -1,11 +1,17 @@ import type { RpcClient } from './rpc-client' import type { MobileConnectionPath } from './stable-logical-rpc-client' import type { ConnectionState, HostProfile } from './types' +import type { HostProfileEdit } from './host-endpoint-edit' export type RpcClientContextValue = { acquire: (hostId: string, host?: HostProfile) => RpcClient | null release: (hostId: string) => void forceReconnect: (hostId: string, host?: HostProfile) => Promise + forceReconnectAfterEdit: ( + hostId: string, + fallbackHost: HostProfile, + updates: HostProfileEdit + ) => Promise closeHost: (hostId: string) => void getState: (hostId: string) => ConnectionState getReconnectAttempt: (hostId: string) => number diff --git a/mobile/src/transport/rpc-client-stream-frame-routing.ts b/mobile/src/transport/rpc-client-stream-frame-routing.ts new file mode 100644 index 00000000000..b774aa4f7b1 --- /dev/null +++ b/mobile/src/transport/rpc-client-stream-frame-routing.ts @@ -0,0 +1,47 @@ +import { + decodeBrowserScreencastFrame, + type BrowserScreencastFrame +} from './browser-screencast-protocol' +import { + handleTerminalBinaryFrame, + type TerminalSnapshotState +} from './rpc-client-terminal-binary-frame' + +export type RpcStreamingListener = (result: unknown) => void + +export type RpcClientStreamRequest = { + method: string + params: unknown + listener: RpcStreamingListener + onBinaryFrame?: (frame: BrowserScreencastFrame) => void + subscriptionId?: string + cancelled?: boolean + sent?: 'awaiting' | 'received' +} + +type StreamFrameRoutingOptions = { + activeBrowserRequestId: string | null + streams: Map + terminalSnapshots: Map + terminalListeners: Map +} + +export function routeRpcClientStreamFrame( + bytes: Uint8Array, + options: StreamFrameRoutingOptions +): void { + const browserFrame = decodeBrowserScreencastFrame(bytes) + if (browserFrame) { + const stream = options.activeBrowserRequestId + ? options.streams.get(options.activeBrowserRequestId) + : undefined + if (stream && !stream.cancelled && stream.method === 'browser.screencast') { + stream.onBinaryFrame?.(browserFrame) + } + return + } + handleTerminalBinaryFrame(bytes, { + terminalSnapshots: options.terminalSnapshots, + getListener: (streamId) => options.terminalListeners.get(streamId) + }) +} diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index 375f1198326..0e1fa13c4e6 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -8,14 +8,12 @@ import { decrypt, decryptBytes } from './e2ee' +import type { TerminalSnapshotState } from './rpc-client-terminal-binary-frame' import { - handleTerminalBinaryFrame, - type TerminalSnapshotState -} from './rpc-client-terminal-binary-frame' -import { - decodeBrowserScreencastFrame, - type BrowserScreencastFrame -} from './browser-screencast-protocol' + routeRpcClientStreamFrame, + type RpcClientStreamRequest as StreamRequest, + type RpcStreamingListener as StreamingListener +} from './rpc-client-stream-frame-routing' import type { ConnectOptions, RpcClient, @@ -59,18 +57,6 @@ type ConnectWaiter = { timeout: ReturnType | null } -type StreamingListener = (result: unknown) => void - -type StreamRequest = { - method: string - params: unknown - listener: StreamingListener - onBinaryFrame?: (frame: BrowserScreencastFrame) => void - subscriptionId?: string - cancelled?: boolean - sent?: 'awaiting' | 'received' -} - // Why: tiered backoff — fast early entries recover blips; the slow tail avoids burning a SYN every 4s on an unreachable desktop. const RECONNECT_DELAYS = [500, 1000, 2000, 4000, 8000, 15_000, 30_000, 60_000] // Why: ≈6 min of failure before the re-pair banner; MUST stay aligned with connection-health.ts UNREACHABLE_ATTEMPTS. @@ -412,7 +398,12 @@ export function connect( } resetTerminalStreamRoutingForRequest(id) if ( - sendEncrypted({ id, deviceToken, method: stream.method, params: stream.params }) + sendEncrypted({ + id, + deviceToken, + method: stream.method, + params: stream.params + }) ) { stream.sent = 'awaiting' } else { @@ -421,7 +412,10 @@ export function connect( } } } else if (msg.type === 'e2ee_error' || (!msg.ok && msg.error?.code === 'unauthorized')) { - console.log('[net] e2ee auth FAILED', { msgType: msg.type, error: msg.error }) + console.log('[net] e2ee auth FAILED', { + msgType: msg.type, + error: msg.error + }) if (handshakeTimer) { clearTimeout(handshakeTimer) handshakeTimer = null @@ -451,7 +445,12 @@ export function connect( if (!plaintextBytes) { return } - handleBinaryFrame(plaintextBytes) + routeRpcClientStreamFrame(plaintextBytes, { + activeBrowserRequestId: activeBrowserScreencastRequestId, + streams: streamListeners, + terminalSnapshots, + terminalListeners: terminalStreamListeners + }) return } @@ -471,11 +470,13 @@ export function connect( } const request = pending.get(response.id) const lateApplicationResponse = applicationResponseTracker.consumeLateResponse(response.id) + const responseStream = streamListeners.get(response.id) + const firstStreamResponse = consumeFirstStreamControlResponse(responseStream) if ( request || timedOutControlRequestIds.consume(response.id) || lateApplicationResponse || - consumeFirstStreamControlResponse(streamListeners.get(response.id)) + firstStreamResponse ) { controlResponseSequence++ } @@ -485,20 +486,22 @@ export function connect( handleAuthRejection('Unauthorized — pairing may be revoked') return } + if (response.ok && firstStreamResponse && responseStream) { + applicationResponseTracker.recordResponse(responseStream.method) + } const isStreaming = response.ok && (response as RpcSuccess).streaming === true if (isStreaming) { - const stream = streamListeners.get(response.id) - if (stream && response.ok) { + if (responseStream && response.ok) { const result = (response as RpcSuccess).result if (isStreamingSubscriptionReadyResult(result)) { - stream.subscriptionId = result.subscriptionId - if (stream.cancelled) { - sendServerSubscriptionUnsubscribe(stream) + responseStream.subscriptionId = result.subscriptionId + if (responseStream.cancelled) { + sendServerSubscriptionUnsubscribe(responseStream) removeStreamListener(response.id) return } - if (stream.method === 'browser.screencast') { + if (responseStream.method === 'browser.screencast') { if ( pendingBrowserScreencastRequestId !== response.id && activeBrowserScreencastRequestId !== response.id @@ -518,10 +521,10 @@ export function connect( terminalStreamIdsByRequest.set(response.id, ids) } ids.add(result.streamId) - terminalStreamListeners.set(result.streamId, stream.listener) + terminalStreamListeners.set(result.streamId, responseStream.listener) } - if (!stream.cancelled) { - stream.listener(result) + if (!responseStream.cancelled) { + responseStream.listener(result) } } return @@ -762,7 +765,9 @@ export function connect( if (controlResponded) { return } - console.log('[net] activity-probe TIMEOUT — forcing reconnect', { state }) + console.log('[net] activity-probe TIMEOUT — forcing reconnect', { + state + }) forceSocketReconnect(probeWs) }, 8_000) pending.set(id, { @@ -910,29 +915,6 @@ export function connect( } } - function handleBinaryFrame(bytes: Uint8Array): void { - const browserFrame = decodeBrowserScreencastFrame(bytes) - if (browserFrame) { - handleBrowserBinaryFrame(browserFrame) - return - } - handleTerminalBinaryFrame(bytes, { - terminalSnapshots, - getListener: (streamId) => terminalStreamListeners.get(streamId) - }) - } - - function handleBrowserBinaryFrame(frame: BrowserScreencastFrame) { - if (!activeBrowserScreencastRequestId) { - return - } - const stream = streamListeners.get(activeBrowserScreencastRequestId) - if (!stream || stream.cancelled || stream.method !== 'browser.screencast') { - return - } - stream.onBinaryFrame?.(frame) - } - function sendEncrypted(request: unknown): boolean { if (ws && ws.readyState === WebSocket.OPEN && sharedKey) { ws.send(encrypt(JSON.stringify(request), sharedKey)) @@ -1014,7 +996,11 @@ export function connect( const id = nextId() const timeout = setTimeout(() => { pending.delete(id) - console.log('[net] sendRequest TIMEOUT', { method, timeoutMs, state }) + console.log('[net] sendRequest TIMEOUT', { + method, + timeoutMs, + state + }) // Why: the frame was written 30s ago — the host may have processed it. reject(markRpcDeliveryUnknown(new Error(`Request timed out: ${method}`))) if (requestWs === ws) { @@ -1084,7 +1070,10 @@ export function connect( } } else { // Registered now; the outbound subscribe is (re-)sent once the channel reaches 'connected'. - console.log('[net] subscribe queued — waiting for connected', { method, state }) + console.log('[net] subscribe queued — waiting for connected', { + method, + state + }) } return () => { @@ -1111,7 +1100,12 @@ export function connect( } else { const unsub = buildStreamUnsubscribe(stream?.method, stream?.params) if (unsub) { - sendEncrypted({ id: nextId(), deviceToken, method: unsub.method, params: unsub.params }) + sendEncrypted({ + id: nextId(), + deviceToken, + method: unsub.method, + params: unsub.params + }) } } removeStreamListener(id)