fix(mobile): fence reconnect publication lifecycle

This commit is contained in:
Brennan Benson
2026-08-01 11:09:38 -07:00
parent 7126150774
commit 7ee1184a0e
17 changed files with 283 additions and 77 deletions
+1 -4
View File
@@ -106,8 +106,6 @@ export default function EditHostScreen() {
...(willRename ? { name: nextName } : {}),
...(nextEndpoint !== undefined ? { endpoint: nextEndpoint } : {})
}
let reconnectWithCurrentProfile = false
savingRef.current = true
setSaving(true)
setSaveError(null)
@@ -138,7 +136,6 @@ export default function EditHostScreen() {
}
primeHosts(hosts, hostLoadRevision)
if (hostLoadRevision === getHostListLoadRevision()) {
reconnectWithCurrentProfile = true
break
}
}
@@ -150,7 +147,7 @@ export default function EditHostScreen() {
setSaving(false)
router.back()
if (nextEndpoint !== undefined && reconnectWithCurrentProfile) {
if (nextEndpoint !== undefined) {
// 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.
+15
View File
@@ -255,6 +255,21 @@ describe('edit host handleSave', () => {
act(() => renderer.unmount())
})
it('retires the old endpoint client when post-save profile loading fails', async () => {
dependencies.loadHosts
.mockResolvedValueOnce([HOST_FIXTURE])
.mockRejectedValueOnce(new Error('keychain locked'))
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')
expect(dependencies.back).toHaveBeenCalledTimes(1)
act(() => renderer.unmount())
})
it('still navigates back and alerts when the post-save reconnect rejects', async () => {
const savedHost = { ...HOST_FIXTURE, endpoint: 'ws://192.168.1.20:6768' }
dependencies.loadHosts.mockResolvedValueOnce([HOST_FIXTURE]).mockResolvedValueOnce([savedHost])
@@ -0,0 +1,5 @@
import { useRpcClientContext } from './client-context'
export const useCloseHost = () => useRpcClientContext().closeHost
export const useForceReconnect = () => useRpcClientContext().forceReconnect
export const usePrimeHosts = () => useRpcClientContext().primeHosts
@@ -541,6 +541,7 @@ describe('useHostClient', () => {
loadHostsMock.mockResolvedValue([HOST])
const harness = await renderHarness(HOST.id)
harness.primeHosts([HOST])
const first = harness.forceReconnect(HOST.id)
await act(async () => {
await Promise.resolve()
+25 -41
View File
@@ -1,5 +1,3 @@
// Single shared RpcClient per host, collapsing the old per-screen WebSocket connections.
// Design: docs/mobile-shared-client-per-host.md.
import {
createContext,
useCallback,
@@ -30,16 +28,13 @@ type StoreEntry = HostReconnectEntry & { state: ConnectionState }
const Ctx = createContext<RpcClientContextValue | null>(null)
export function RpcClientProvider({ children }: { children: ReactNode }) {
// Why: entries in a ref so state changes don't re-render the whole tree; propagation goes through per-host listener Sets.
const storeRef = useRef<Map<string, StoreEntry>>(new Map())
const stateListenersRef = useRef<Map<string, Set<(state: ConnectionState) => void>>>(new Map())
const allHostsListenersRef = useRef<Set<() => void>>(new Set())
// Pending opens keyed by hostId so two acquire() callers in the same render don't race the host lookup.
const pendingOpensRef = useRef(new HostClientOpenRegistry())
const forceReconnectCoordinatorRef = useRef(new HostForceReconnectCoordinator())
// Why: cache of already-loaded HostProfiles so openEntry can skip a second loadHosts()/Keychain pass on cold start.
const primedHostsRef = useRef(new HostReconnectProfileCache())
function notifyHostState(hostId: string, state: ConnectionState) {
@@ -87,26 +82,34 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
const pendingOpen = pendingOpensRef.current.register(hostId, profileVersion, promise)
try {
const cachedHost =
requestedProfile?.host ?? primedHostsRef.current.get(hostId, currentRevision)
const profile = cachedHost
? { host: cachedHost, version: requestedProfile?.version ?? profileVersion }
: await loadHostClientOpenProfile({
hostId,
cache: primedHostsRef.current,
ticket: pendingOpen,
loadHosts,
onUnavailable: () => {
notifyHostState(hostId, 'disconnected')
notifyAllHosts()
}
})
if (!profile || pendingOpen.cancelled) {
const onUnavailable = () => {
notifyHostState(hostId, 'disconnected')
notifyAllHosts()
}
const loadProfile = () =>
loadHostClientOpenProfile({
hostId,
cache: primedHostsRef.current,
ticket: pendingOpen,
loadHosts,
onUnavailable
})
let profile: HostOpenProfile | null = primedHostsRef.current.openProfile(
hostId,
currentRevision,
requestedProfile
)
if (!profile.host) {
profile = await loadProfile()
}
while (profile && profile.sourceRevision !== getHostListLoadRevision()) {
profile = await loadProfile()
}
if (!profile?.host || pendingOpen.cancelled) {
return null
}
pendingOpen.profileVersion = profile.version
// Re-check after any await — another acquire() may have completed.
const after = storeRef.current.get(hostId)
if (after) {
return after
@@ -152,7 +155,6 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
[]
)
// Synchronous get-or-open: returns an existing client immediately, else kicks off an async open and returns null this tick.
const acquire = useCallback(
(hostId: string, host?: HostProfile): RpcClient | null => {
if (host) {
@@ -163,7 +165,6 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
existing.refCount += 1
return existing.client
}
// Trigger async open; returns null this tick — consumers re-call acquire() from an effect that re-runs on state changes.
void openEntry(hostId).then((entry) => {
if (!entry) {
return
@@ -268,7 +269,6 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
}
}, [])
// Close all clients on provider unmount. Empty deps: [closeEntry] would let Fast Refresh tear down all live sockets.
// eslint-disable-next-line react-hooks/exhaustive-deps
useEffect(() => {
const store = storeRef.current
@@ -281,7 +281,6 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
}
}, [])
// Why: nudge live clients when the OS signals the link may be back so sessions recover without a restart (issue #5049).
useEffect(() => {
return subscribeConnectionRevivalTriggers(() => {
for (const entry of storeRef.current.values()) {
@@ -332,7 +331,6 @@ export function useRpcClientContext(): RpcClientContextValue {
return ctx
}
// Primary hook for screens: acquires the shared client on mount, releases on unmount, re-renders on state change.
export function useHostClient(hostId: string | undefined): {
client: RpcClient | null
state: ConnectionState
@@ -393,10 +391,8 @@ export function useHostClient(hostId: string | undefined): {
return { client: bound ? clientRef.current : null, state: boundState }
}
// Why: refcounting prevents a double-open when a host-detail screen shares one of these hosts.
export function useAllHostClients(hostIds: string[]) {
const ctx = useRpcClientContext()
// Stable key so we don't tear down on every render of the array.
const key = useMemo(() => [...hostIds].sort().join(','), [hostIds])
const [tick, setTick] = useState(0)
@@ -447,16 +443,4 @@ export function useAllHostClients(hostIds: string[]) {
}
// Why: host-store's removeHost() must close the live client but has no React-side handle; this hook bridges to it.
export function useCloseHost(): (hostId: string) => void {
return useRpcClientContext().closeHost
}
// Why: future-proof "Connection issues — try again" affordance.
export function useForceReconnect(): (hostId: string, host?: HostProfile) => Promise<void> {
return useRpcClientContext().forceReconnect
}
// Why: primes already-loaded HostProfiles so the provider can skip a second loadHosts()/Keychain pass on cold start.
export function usePrimeHosts(): (hosts: HostProfile[], sourceRevision: number) => void {
return useRpcClientContext().primeHosts
}
export { useCloseHost, useForceReconnect, usePrimeHosts } from './client-context-actions'
@@ -20,7 +20,10 @@ import {
import {
beginHostEndpointPublicationLifecycle,
getHostProfilePublicationRevision,
publishHostProfileTransaction
publishHostProfileTransaction,
recordDurableHostIdentity,
resetHostProfilePublicationForTests,
serializeHostProfilePublication
} from './host-profile-publication'
const HOST: HostProfile = {
@@ -35,6 +38,8 @@ const HOST: HostProfile = {
describe('existing host relay endpoint lifecycle publication', () => {
beforeEach(() => {
vi.clearAllMocks()
resetHostProfilePublicationForTests()
recordDurableHostIdentity(HOST)
hostStoreMock.loadStoredHostIdentity.mockResolvedValue(HOST)
tokenStoreMock.readHostDeviceToken.mockResolvedValue(HOST.deviceToken)
overlayStoreMock.saveMobileRelayHostOverlay.mockResolvedValue(undefined)
@@ -157,6 +162,29 @@ describe('existing host relay endpoint lifecycle publication', () => {
expect(overlayStoreMock.saveMobileRelayHostOverlay).not.toHaveBeenCalled()
})
it('orders host removal after an in-flight publication', async () => {
let releasePublication: (() => void) | null = null
const events: string[] = []
const publication = serializeHostProfilePublication(HOST.id, async () => {
events.push('publication-started')
await new Promise<void>((resolve) => {
releasePublication = resolve
})
events.push('publication-finished')
})
await vi.waitFor(() => expect(events).toEqual(['publication-started']))
const removal = serializeHostProfilePublication(HOST.id, async () => {
events.push('removed')
})
await Promise.resolve()
expect(events).toEqual(['publication-started'])
releasePublication?.()
await Promise.all([publication, removal])
expect(events).toEqual(['publication-started', 'publication-finished', 'removed'])
})
it('rejects an old credential write that starts during same-host re-pair publication', async () => {
let releaseReplacement: (() => void) | null = null
let markReplacementStarted: (() => void) | null = null
@@ -194,6 +222,34 @@ describe('existing host relay endpoint lifecycle publication', () => {
expect(writeBundle).not.toHaveBeenCalled()
})
it('rejects an old credential after a replacement token commits but publication fails', async () => {
let releaseOldMetadata: ((host: HostProfile | null) => void) | null = null
hostStoreMock.loadStoredHostIdentity.mockImplementationOnce(
() =>
new Promise<HostProfile | null>((resolve) => {
releaseOldMetadata = resolve
})
)
const writeBundle = vi.fn(async () => {})
const staleWrite = writeExistingHostRelayCredentialBundle(
HOST,
credentialBundle(1),
writeBundle
)
await vi.waitFor(() => expect(hostStoreMock.loadStoredHostIdentity).toHaveBeenCalledOnce())
const replacementHost = { ...HOST, deviceToken: 'token-2' }
const replacement = publishHostProfileTransaction(replacementHost, null, async () => {
recordDurableHostIdentity(replacementHost)
throw new Error('overlay write failed')
})
releaseOldMetadata?.(HOST)
await expect(replacement).rejects.toThrow('overlay write failed')
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
@@ -105,18 +105,18 @@ function requireCurrentPublication(
host: HostProfile,
endpointLifecycle: HostEndpointPublicationLifecycle
): void {
const publishedIdentity = getPublishedHostIdentity(host.id)
if (
publishedIdentity &&
(publishedIdentity.deviceToken !== host.deviceToken ||
publishedIdentity.publicKeyB64 !== host.publicKeyB64)
) {
throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired')
}
if (
getHostProfilePublicationRevision(host.id) !== endpointLifecycle.profileRevision ||
getHostEndpointPublicationLifecycle(host.id).generation !== endpointLifecycle.generation
) {
const publishedIdentity = getPublishedHostIdentity(host.id)
if (
publishedIdentity &&
(publishedIdentity.deviceToken !== host.deviceToken ||
publishedIdentity.publicKeyB64 !== host.publicKeyB64)
) {
throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired')
}
throw new MobileRelayUpgradeLifecycleRetiredError('mobile relay endpoint lifecycle was retired')
}
}
@@ -0,0 +1,38 @@
import { describe, expect, it, vi } from 'vitest'
import { dropSharedHostListLoad, getHostListLoadRevision } from './host-list-load-sharing'
import { loadHostClientOpenProfile } from './host-client-open-profile'
import { HostReconnectProfileCache } from './host-reconnect-profile-cache'
import type { HostClientOpenTicket } from './host-client-open-registry'
import type { HostProfile } from './types'
const HOST: HostProfile = {
id: 'host-1',
name: 'Host 1',
endpoint: 'ws://127.0.0.1:1',
deviceToken: 'token',
publicKeyB64: 'key',
lastConnected: 0
}
describe('host client open profile', () => {
it('returns the exact durable revision that produced the loaded profile', async () => {
dropSharedHostListLoad()
const ticket: HostClientOpenTicket = {
cancelled: false,
profileVersion: 0,
promise: Promise.resolve()
}
const profile = await loadHostClientOpenProfile({
hostId: HOST.id,
cache: new HostReconnectProfileCache(),
ticket,
loadHosts: vi.fn(async () => [HOST]),
onUnavailable: vi.fn()
})
expect(profile?.sourceRevision).toBe(getHostListLoadRevision())
dropSharedHostListLoad()
expect(profile?.sourceRevision).not.toBe(getHostListLoadRevision())
})
})
@@ -1,6 +1,6 @@
import type { HostClientOpenTicket, HostClientOpenRegistry } from './host-client-open-registry'
import type { HostReconnectProfileCache } from './host-reconnect-profile-cache'
import { dropSharedHostListLoad, getHostListLoadRevision } from './host-list-load-sharing'
import { abandonSharedHostListLoad, getHostListLoadRevision } from './host-list-load-sharing'
import type { HostProfile } from './types'
type HostClientOpenProfileOptions = {
@@ -16,12 +16,12 @@ export function cancelHostClientOpenProfile(
hostId: string
): void {
registry.cancel(hostId)
dropSharedHostListLoad()
abandonSharedHostListLoad()
}
export async function loadHostClientOpenProfile(
options: HostClientOpenProfileOptions
): Promise<{ host: HostProfile; version: number } | null> {
): Promise<{ host: HostProfile; sourceRevision: number; version: number } | null> {
const { hostId, cache, ticket } = options
while (!ticket.cancelled) {
const sourceRevision = getHostListLoadRevision()
@@ -47,6 +47,7 @@ export async function loadHostClientOpenProfile(
}
return {
host: cache.get(hostId, currentRevision) ?? loadedHost,
sourceRevision: currentRevision,
version: cache.version(hostId, currentRevision)
}
}
@@ -29,3 +29,8 @@ export function dropSharedHostListLoad(): void {
revision += 1
inflight = null
}
/** Stop offering a possibly stalled pass without invalidating durable snapshot revisions. */
export function abandonSharedHostListLoad(): void {
inflight = null
}
@@ -20,6 +20,21 @@ export function getPublishedHostIdentity(
return publishedIdentityByHost.get(hostId)
}
export function recordDurableHostIdentity(
host: Pick<HostProfile, 'id' | 'deviceToken' | 'publicKeyB64'>
): void {
publishedIdentityByHost.set(host.id, {
deviceToken: host.deviceToken,
publicKeyB64: host.publicKeyB64
})
}
export function retireHostProfilePublication(hostId: string): void {
revisionByHost.set(hostId, getHostProfilePublicationRevision(hostId) + 1)
endpointGenerationByHost.set(hostId, (endpointGenerationByHost.get(hostId) ?? 0) + 1)
publishedIdentityByHost.delete(hostId)
}
export function beginHostEndpointPublicationLifecycle(
hostId: string
): HostEndpointPublicationLifecycle {
@@ -65,9 +80,14 @@ export function publishHostProfileTransaction(
await beforeHostSave?.()
await saveHost(host)
revisionByHost.set(host.id, getHostProfilePublicationRevision(host.id) + 1)
publishedIdentityByHost.set(host.id, {
deviceToken: host.deviceToken,
publicKeyB64: host.publicKeyB64
})
recordDurableHostIdentity(host)
})
}
/** Test-only: reset publication state between module-level storage cases. */
export function resetHostProfilePublicationForTests(): void {
pendingByHost.clear()
revisionByHost.clear()
endpointGenerationByHost.clear()
publishedIdentityByHost.clear()
}
@@ -44,6 +44,7 @@ describe('HostReconnectProfileCache', () => {
expect(cache.reconnectProfile(HOST.id, 5)).toEqual({
host: undefined,
sourceRevision: 5,
version: cachedVersion + 1
})
})
@@ -8,6 +8,7 @@ type CachedHostProfile = {
export type HostOpenProfile = {
host: HostProfile | undefined
sourceRevision: number
version: number
}
@@ -32,9 +33,28 @@ export class HostReconnectProfileCache {
requestedHost?: HostProfile
): HostOpenProfile {
return requestedHost
? { host: requestedHost, version: this.prime(requestedHost, currentRevision) }
? {
host: requestedHost,
sourceRevision: currentRevision,
version: this.prime(requestedHost, currentRevision)
}
: {
host: this.get(hostId, currentRevision),
sourceRevision: currentRevision,
version: this.version(hostId, currentRevision)
}
}
openProfile(
hostId: string,
currentRevision: number,
requested?: HostOpenProfile
): HostOpenProfile {
return requested?.host
? requested
: {
host: this.get(hostId, currentRevision),
sourceRevision: currentRevision,
version: this.version(hostId, currentRevision)
}
}
+26 -15
View File
@@ -22,7 +22,12 @@ import {
import { deleteMobileRelayCredentialBundle } from './mobile-relay-credential-bundle'
import { deleteMobileRelayDirectUpgradeJournal } from './mobile-relay-direct-upgrade-journal'
import { scheduleOrphanedMobileRelayCleanup } from './mobile-relay-orphan-cleanup'
import { serializeHostProfilePublication } from './host-profile-publication'
import {
recordDurableHostIdentity,
resetHostProfilePublicationForTests,
retireHostProfilePublication,
serializeHostProfilePublication
} from './host-profile-publication'
import {
deleteHostDeviceToken,
readHostDeviceToken,
@@ -218,6 +223,8 @@ async function persistHost(host: HostProfile): Promise<void> {
// Why: write metadata before the keychain token so a crash leaves recoverable orphaned metadata, not an orphaned token that persists forever.
await writeHostDeviceToken(stored.id, validated.deviceToken)
tokenCache.set(stored.id, validated.deviceToken)
// Why: a later overlay failure must not hide the already-committed credential identity.
recordDurableHostIdentity(validated)
hostListLoads.dropSharedHostListLoad()
if (validated.endpoints) {
await saveMobileRelayHostOverlay({
@@ -249,20 +256,23 @@ async function persistHost(host: HostProfile): Promise<void> {
}
export async function removeHost(hostId: string): Promise<void> {
await mutateStoredHosts((hosts) => hosts.filter((h) => h.id !== hostId))
tokenCache.delete(hostId)
try {
await removeMobileRelayHostOverlay(hostId)
hostListLoads.dropSharedHostListLoad()
} catch {
// Base removal is authoritative; a retained overlay can't resurrect the host and is cleaned on a later retry.
}
// Why: keychain delete can stall/reject; await only the durable cleanup intent so removeHost can't freeze the UI.
try {
await scheduleHostCredentialCleanup(hostId, deleteHostCredentials)
} catch {
// Metadata is already committed; orphan-token recovery is best-effort.
}
await serializeHostProfilePublication(hostId, async () => {
await mutateStoredHosts((hosts) => hosts.filter((h) => h.id !== hostId))
tokenCache.delete(hostId)
retireHostProfilePublication(hostId)
try {
await removeMobileRelayHostOverlay(hostId)
hostListLoads.dropSharedHostListLoad()
} catch {
// Base removal is authoritative; a retained overlay can't resurrect the host and is cleaned on a later retry.
}
// Why: keychain delete can stall/reject; await only the durable cleanup intent so removeHost can't freeze the UI.
try {
await scheduleHostCredentialCleanup(hostId, deleteHostCredentials)
} catch {
// Metadata is already committed; orphan-token recovery is best-effort.
}
})
}
export async function retryPendingHostCredentialCleanup(): Promise<{
@@ -322,5 +332,6 @@ export function resetHostStoreForTests(): void {
hostListMutation = Promise.resolve()
tokenCache.clear()
hostListLoads.dropSharedHostListLoad()
resetHostProfilePublicationForTests()
resetPairingKeychainForTests()
}
@@ -17,6 +17,7 @@ import {
type HostEndpointPublicationLifecycle
} from './host-profile-publication'
import { upgradeDirectMobileRelay } from './mobile-relay-direct-upgrade'
import { retireMobileRelayDirectUpgradeJournalForRelayHost } from './mobile-relay-direct-upgrade-journal'
import { MobileRelayDirectUpgradeController } from './mobile-relay-direct-upgrade-controller'
import type { StableLogicalRpcClient } from './stable-logical-rpc-client'
@@ -57,6 +58,8 @@ export function startMobileEndpointLifecycle(
}
if (initialHost.relay) {
// Why: relay publication supersedes any committed journal retained by the retired direct owner.
void retireMobileRelayDirectUpgradeJournalForRelayHost(initialHost.id).catch(() => {})
owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate, endpointLifecycle)
void owner.start()
} else {
@@ -0,0 +1,40 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
const keychain = vi.hoisted(() => ({
deletePairingKeychainItemIfMatches: vi.fn(),
readPairingKeychainItem: vi.fn()
}))
vi.mock('react-native', () => ({ Platform: { OS: 'ios' } }))
vi.mock('./pairing-keychain', () => ({
deletePairingKeychainItem: vi.fn(),
deletePairingKeychainItemIfMatches: keychain.deletePairingKeychainItemIfMatches,
readPairingKeychainItem: keychain.readPairingKeychainItem,
writePairingKeychainItem: vi.fn()
}))
import {
createMobileRelayDirectUpgradeJournal,
retireMobileRelayDirectUpgradeJournalForRelayHost
} from './mobile-relay-direct-upgrade-journal'
describe('mobile relay direct upgrade journal retirement', () => {
beforeEach(() => {
vi.clearAllMocks()
keychain.deletePairingKeychainItemIfMatches.mockResolvedValue(true)
})
it('conditionally clears the retained journal when a relay profile takes ownership', async () => {
const journal = createMobileRelayDirectUpgradeJournal('host-1', (length) =>
new Uint8Array(length).fill(7)
)
keychain.readPairingKeychainItem.mockResolvedValue(JSON.stringify(journal))
await retireMobileRelayDirectUpgradeJournalForRelayHost(journal.hostId)
expect(keychain.deletePairingKeychainItemIfMatches).toHaveBeenCalledWith(
`orca.mobile-relay.direct-upgrade.${journal.hostId}`,
JSON.stringify(journal)
)
})
})
@@ -81,6 +81,15 @@ export function deleteMobileRelayDirectUpgradeJournalIfCurrent(
return deletePairingKeychainItemIfMatches(journalKey(parsed.hostId), JSON.stringify(parsed))
}
export async function retireMobileRelayDirectUpgradeJournalForRelayHost(
hostId: string
): Promise<void> {
const journal = await readMobileRelayDirectUpgradeJournal(hostId)
if (journal) {
await deleteMobileRelayDirectUpgradeJournalIfCurrent(journal)
}
}
function encodeBase64Url(value: Uint8Array): string {
let binary = ''
for (const byte of value) {