fix(mobile): fence reconnect lifecycle races

2026-10-20 00:03:45 +00:00 · 2026-08-01 04:21:36 -07:00
parent 39dc249792
commit e4813c8225
9 changed files with 268 additions and 50 deletions
@@ -3,6 +3,7 @@ import { loadHosts } from './host-store'
 import { serializeHostProfilePublication } from './host-profile-publication'
 import * as hostListLoads from './host-list-load-sharing'
 import { saveMobileRelayHostOverlay } from './mobile-relay-host-overlay-store'
+import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle'

 export class MobileRelayUpgradeHostRemovedError extends Error {}
 export class MobileRelayUpgradeHostSupersededError extends Error {}
@@ -13,13 +14,7 @@ export async function saveExistingHostRelayRouting(
 ): Promise<void> {
  const validated = HostProfileSchema.parse(host)
  await serializeHostProfilePublication(validated.id, async () => {
-    const existing = (await loadHosts()).find(({ id }) => id === validated.id)
-    if (!existing || existing.publicKeyB64 !== validated.publicKeyB64) {
-      throw new MobileRelayUpgradeHostRemovedError('mobile relay upgrade host was removed')
-    }
-    if (existing.deviceToken !== validated.deviceToken) {
-      throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired')
-    }
+    const existing = await requireCurrentHostIdentity(validated)
    const { endpoints, relayHostId, relay } = validated
    if (!endpoints || !relayHostId || !relay) {
      throw new Error('mobile relay upgrade routing metadata missing')
@@ -29,10 +24,40 @@ export async function saveExistingHostRelayRouting(
    await saveMobileRelayHostOverlay({
      v: 2,
      hostId: validated.id,
-      endpoints,
+      endpoints: endpoints.map((endpoint) =>
+        endpoint.id === 'direct-primary' && endpoint.kind !== 'relay'
+          ? { ...endpoint, url: existing.endpoint }
+          : endpoint
+      ),
      relayHostId,
      relay
    })
    hostListLoads.dropSharedHostListLoad()
  })
 }
+
+export async function writeExistingHostRelayCredentialBundle(
+  host: HostProfile,
+  bundle: MobileRelayCredentialBundle,
+  writeBundle: (bundle: MobileRelayCredentialBundle) => Promise<void>
+): Promise<void> {
+  const validated = HostProfileSchema.parse(host)
+  await serializeHostProfilePublication(validated.id, async () => {
+    await requireCurrentHostIdentity(validated)
+    if (bundle.hostId !== validated.id || bundle.deviceToken !== validated.deviceToken) {
+      throw new MobileRelayUpgradeHostSupersededError('mobile relay credential identity mismatch')
+    }
+    await writeBundle(bundle)
+  })
+}
+
+async function requireCurrentHostIdentity(host: HostProfile): Promise<HostProfile> {
+  const existing = (await loadHosts()).find(({ id }) => id === host.id)
+  if (!existing || existing.publicKeyB64 !== host.publicKeyB64) {
+    throw new MobileRelayUpgradeHostRemovedError('mobile relay upgrade host was removed')
+  }
+  if (existing.deviceToken !== host.deviceToken) {
+    throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired')
+  }
+  return existing
+}
@@ -27,4 +27,26 @@ describe('HostForceReconnectCoordinator', () => {
    await expect(outcome).resolves.toBe('Force Reconnect timed out')
    expect(cancelPendingOpen).toHaveBeenCalledTimes(2)
  })
+
+  it('settles a cancelled reconnect without waiting for its host lookup deadline', async () => {
+    vi.useFakeTimers()
+    const cancelPendingOpen = vi.fn()
+    const coordinator = new HostForceReconnectCoordinator()
+    const reconnect = coordinator.run({
+      hostId: 'host-1',
+      profileVersion: 0,
+      getEntry: () => undefined,
+      getListenerCount: () => 1,
+      removeEntry: vi.fn(),
+      cancelPendingOpen,
+      openReplacement: () => new Promise<never>(() => {})
+    })
+
+    coordinator.cancel('host-1')
+
+    await expect(reconnect).resolves.toBeUndefined()
+    expect(cancelPendingOpen).toHaveBeenCalledTimes(2)
+    await vi.advanceTimersByTimeAsync(15_000)
+    await expect(reconnect).resolves.toBeUndefined()
+  })
 })
@@ -24,6 +24,7 @@ type PendingReconnect = {
  profileVersion: number
  generation: number
  promise: Promise<void>
+  cancel: () => void
 }

 export class HostForceReconnectCoordinator {
@@ -32,6 +33,7 @@ export class HostForceReconnectCoordinator {

  cancel(hostId: string): void {
    this.generations.set(hostId, this.generation(hostId) + 1)
+    this.pendingByHost.get(hostId)?.cancel()
    this.pendingByHost.delete(hostId)
  }

@@ -50,18 +52,28 @@ export class HostForceReconnectCoordinator {
    if (pending) {
      generation += 1
      this.generations.set(operation.hostId, generation)
+      pending.cancel()
    }
    // Why: a first reconnect must not join an ordinary open whose Keychain read may never settle.
    operation.cancelPendingOpen()
+    let cancelReconnect: () => void = () => {}
+    const cancelled = new Promise<void>((resolve) => {
+      cancelReconnect = () => {
+        operation.cancelPendingOpen()
+        resolve()
+      }
+    })
    const reconnect = this.replaceAndVerify(
      operation,
      generation,
-      Date.now() + FORCE_RECONNECT_TIMEOUT_MS
+      Date.now() + FORCE_RECONNECT_TIMEOUT_MS,
+      cancelled
    )
    this.pendingByHost.set(operation.hostId, {
      profileVersion: operation.profileVersion,
      generation,
-      promise: reconnect
+      promise: reconnect,
+      cancel: cancelReconnect
    })
    const clearPending = () => {
      if (this.pendingByHost.get(operation.hostId)?.promise === reconnect) {
@@ -75,7 +87,8 @@ export class HostForceReconnectCoordinator {
  private async replaceAndVerify(
    operation: HostReconnectOperation,
    generation: number,
-    deadline: number
+    deadline: number,
+    cancelled: Promise<void>
  ): Promise<void> {
    if (this.wasCancelled(operation.hostId, generation)) {
      return
@@ -87,7 +100,7 @@ export class HostForceReconnectCoordinator {
      entry.client.close()
      operation.removeEntry()
    }
-    const fresh = await this.openBeforeDeadline(operation, deadline)
+    const fresh = await this.openBeforeDeadline(operation, deadline, cancelled)
    if (this.wasCancelled(operation.hostId, generation)) {
      return
    }
@@ -106,7 +119,8 @@ export class HostForceReconnectCoordinator {

  private async openBeforeDeadline(
    operation: HostReconnectOperation,
-    deadline: number
+    deadline: number,
+    cancelled: Promise<void>
  ): Promise<HostReconnectEntry | null> {
    const timeoutMs = deadline - Date.now()
    if (timeoutMs <= 0) {
@@ -116,6 +130,7 @@ export class HostForceReconnectCoordinator {
    try {
      return await Promise.race([
        operation.openReplacement(),
+        cancelled.then(() => null),
        new Promise<never>((_resolve, reject) => {
          timer = setTimeout(() => {
            operation.cancelPendingOpen()
@@ -46,7 +46,8 @@ import {
 import {
  MobileRelayUpgradeHostRemovedError,
  MobileRelayUpgradeHostSupersededError,
-  saveExistingHostRelayRouting
+  saveExistingHostRelayRouting,
+  writeExistingHostRelayCredentialBundle
 } from './existing-host-relay-routing'
 import { resetMobileRelayHostOverlayStoreForTests } from './mobile-relay-host-overlay-store'
 import { publishHostProfileTransaction } from './host-profile-publication'
@@ -284,10 +285,51 @@ describe('host-store list mutations', () => {
      expect.objectContaining({ id: HOST_ONE.id, endpoint: 'ws://127.0.0.1:9' })
    )
    expect(JSON.parse(storedOverlayRaw ?? '[]')).toContainEqual(
-      expect.objectContaining({ hostId: HOST_ONE.id, relayHostId })
+      expect.objectContaining({
+        hostId: HOST_ONE.id,
+        relayHostId,
+        endpoints: expect.arrayContaining([
+          expect.objectContaining({
+            id: 'direct-primary',
+            url: 'ws://127.0.0.1:9'
+          })
+        ])
+      })
    )
  })

+  it('updates relay overlay routing when the primary host address changes later', async () => {
+    const overlay: MobileRelayHostOverlay = {
+      v: 2,
+      hostId: HOST_ONE.id,
+      endpoints: [
+        { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint },
+        { id: 'tailscale-backup', kind: 'tailscale', url: 'ws://desk.example.ts.net:6768' },
+        { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' }
+      ],
+      relayHostId: 'AbCdEf0123_-xyZ9',
+      relay: {
+        v: 1,
+        directorUrl: 'https://relay.onorca.dev',
+        cellUrl: 'https://relay.onorca.dev',
+        assignmentEpoch: 1,
+        relayHostId: 'AbCdEf0123_-xyZ9',
+        e2eeFraming: 2
+      }
+    }
+    storedOverlayRaw = JSON.stringify([overlay])
+
+    await updateHostNameAndEndpoint(HOST_ONE.id, { endpoint: 'ws://127.0.0.1:9' })
+
+    const [updated] = JSON.parse(storedOverlayRaw ?? '[]') as MobileRelayHostOverlay[]
+    expect(updated?.endpoints).toContainEqual({
+      id: 'direct-primary',
+      kind: 'lan',
+      url: 'ws://127.0.0.1:9'
+    })
+    expect(updated?.endpoints).toContainEqual(overlay.endpoints[1])
+  })
+
  it('rejects relay routing from a lifecycle superseded by same-host re-pairing', async () => {
    await saveHost({ ...HOST_ONE, deviceToken: 'replacement-token' })
    const relayHostId = 'AbCdEf0123_-xyZ9'
@@ -356,6 +398,52 @@ describe('host-store list mutations', () => {
    expect(JSON.parse(storedOverlayRaw ?? '[]')).toEqual([])
  })

+  it('orders an old lifecycle credential write before same-host re-pair credentials', async () => {
+    let releaseOldWrite: (() => void) | null = null
+    let markOldWriteStarted: (() => void) | null = null
+    const oldWritePending = new Promise<void>((resolve) => {
+      releaseOldWrite = resolve
+    })
+    const oldWriteStarted = new Promise<void>((resolve) => {
+      markOldWriteStarted = resolve
+    })
+    const writes: string[] = []
+    const oldWrite = writeExistingHostRelayCredentialBundle(
+      { ...HOST_ONE, deviceToken: 'token-1' },
+      {
+        v: 1,
+        hostId: HOST_ONE.id,
+        deviceToken: 'token-1',
+        current: {
+          token: 'A'.repeat(43),
+          hash: 'B'.repeat(43),
+          version: 1,
+          expiresAt: 1
+        }
+      },
+      async () => {
+        markOldWriteStarted?.()
+        await oldWritePending
+        writes.push('old')
+      }
+    )
+    await oldWriteStarted
+
+    const replacement = publishHostProfileTransaction(
+      { ...HOST_ONE, deviceToken: 'replacement-token' },
+      async () => {
+        writes.push('replacement')
+      },
+      saveHost
+    )
+    await Promise.resolve()
+    expect(writes).toEqual([])
+
+    releaseOldWrite?.()
+    await Promise.all([oldWrite, replacement])
+    expect(writes).toEqual(['old', 'replacement'])
+  })
+
  it('awaits cleanup scheduling after metadata commit', async () => {
    let resolveSchedule: (() => void) | null = null
    scheduleCleanupMock.mockReturnValue(
@@ -22,47 +22,41 @@ import {
  loadMobileRelayHostOverlayState,
  removeMobileRelayHostOverlay,
  removeMobileRelayHostOverlays,
-  saveMobileRelayHostOverlay
+  saveMobileRelayHostOverlay,
+  updateMobileRelayHostOverlayDirectEndpoint
 } from './mobile-relay-host-overlay-store'
 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'

 const STORAGE_KEY = 'orca:hosts'
 // Why: SecureStore keys must match [A-Za-z0-9._-] (colons rejected), so use dots as the separator.
 const TOKEN_KEY_PREFIX = 'orca.host-token.'
 const WEB_TOKEN_KEY_PREFIX = 'orca:web-host-token:'

-function tokenKey(hostId: string): string {
-  return `${TOKEN_KEY_PREFIX}${hostId}`
-}
-
-function webTokenKey(hostId: string): string {
-  return `${WEB_TOKEN_KEY_PREFIX}${hostId}`
-}
-
 async function readDeviceToken(hostId: string): Promise<string | null> {
  // Why: Expo SecureStore has no working web backend; fall back to AsyncStorage only on web so native still uses the keychain.
  if (Platform.OS === 'web') {
-    return AsyncStorage.getItem(webTokenKey(hostId))
+    return AsyncStorage.getItem(`${WEB_TOKEN_KEY_PREFIX}${hostId}`)
  }
-  return readPairingKeychainItem(tokenKey(hostId))
+  return readPairingKeychainItem(`${TOKEN_KEY_PREFIX}${hostId}`)
 }

 async function writeDeviceToken(hostId: string, token: string): Promise<void> {
  if (Platform.OS === 'web') {
-    await AsyncStorage.setItem(webTokenKey(hostId), token)
+    await AsyncStorage.setItem(`${WEB_TOKEN_KEY_PREFIX}${hostId}`, token)
    return
  }
-  await writePairingKeychainItem(tokenKey(hostId), token)
+  await writePairingKeychainItem(`${TOKEN_KEY_PREFIX}${hostId}`, token)
 }

 async function deleteDeviceToken(hostId: string): Promise<void> {
  if (Platform.OS === 'web') {
-    await AsyncStorage.removeItem(webTokenKey(hostId))
+    await AsyncStorage.removeItem(`${WEB_TOKEN_KEY_PREFIX}${hostId}`)
    return
  }
-  await deletePairingKeychainItem(tokenKey(hostId))
+  await deletePairingKeychainItem(`${TOKEN_KEY_PREFIX}${hostId}`)
 }

 async function deleteHostCredentials(hostId: string): Promise<void> {
@@ -299,18 +293,25 @@ export async function updateHostNameAndEndpoint(
  hostId: string,
  updates: { name?: string; endpoint?: string }
 ): Promise<void> {
-  await mutateStoredHosts((hosts) => {
-    const index = hosts.findIndex((host) => host.id === hostId)
-    if (index < 0) {
-      throw new Error('Host not found')
+  await serializeHostProfilePublication(hostId, async () => {
+    await mutateStoredHosts((hosts) => {
+      const index = hosts.findIndex((host) => host.id === hostId)
+      if (index < 0) {
+        throw new Error('Host not found')
+      }
+      const next = hosts.slice()
+      if (updates.name !== undefined) {
+        next[index] = { ...next[index]!, name: updates.name }
+      }
+      if (updates.endpoint !== undefined) {
+        next[index] = { ...next[index]!, endpoint: updates.endpoint }
+      }
+      return next
+    })
+    if (updates.endpoint !== undefined) {
+      await updateMobileRelayHostOverlayDirectEndpoint(hostId, updates.endpoint)
+      hostListLoads.dropSharedHostListLoad()
    }
-    const next = hosts.slice()
-    next[index] = {
-      ...next[index]!,
-      ...(updates.name !== undefined ? { name: updates.name } : {}),
-      ...(updates.endpoint !== undefined ? { endpoint: updates.endpoint } : {})
-    }
-    return next
  })
 }

@@ -8,7 +8,10 @@ import {
  readMobileRelayCredentialBundle,
  writeMobileRelayCredentialBundle
 } from './mobile-relay-credential-bundle'
-import { saveExistingHostRelayRouting } from './existing-host-relay-routing'
+import {
+  saveExistingHostRelayRouting,
+  writeExistingHostRelayCredentialBundle
+} from './existing-host-relay-routing'
 import { upgradeDirectMobileRelay } from './mobile-relay-direct-upgrade'
 import { MobileRelayDirectUpgradeController } from './mobile-relay-direct-upgrade-controller'
 import type { StableLogicalRpcClient } from './stable-logical-rpc-client'
@@ -98,7 +101,8 @@ function createSupervisor(
      }),
    resolveRelay: resolveMobileRelayEndpoint,
    readBundle: readMobileRelayCredentialBundle,
-    writeBundle: writeMobileRelayCredentialBundle,
+    writeBundle: (bundle) =>
+      writeExistingHostRelayCredentialBundle(host, bundle, writeMobileRelayCredentialBundle),
    saveHost: saveExistingHostRelayRouting,
    onHostUpdated,
    now: Date.now,
@@ -108,6 +108,37 @@ export function removeMobileRelayHostOverlays(hostIds: readonly string[]): Promi
  })
 }

+export function updateMobileRelayHostOverlayDirectEndpoint(
+  hostId: string,
+  endpointUrl: string
+): Promise<void> {
+  return mutateOverlays((overlays) => {
+    const index = overlays.findIndex((overlay) => overlay.hostId === hostId)
+    if (index < 0) {
+      return overlays
+    }
+    const overlay = overlays[index]!
+    let changed = false
+    const endpoints = overlay.endpoints.map((endpoint) => {
+      if (
+        endpoint.id !== 'direct-primary' ||
+        endpoint.kind === 'relay' ||
+        endpoint.url === endpointUrl
+      ) {
+        return endpoint
+      }
+      changed = true
+      return { ...endpoint, url: endpointUrl }
+    })
+    if (!changed) {
+      return overlays
+    }
+    const next = overlays.slice()
+    next[index] = { ...overlay, endpoints }
+    return next
+  })
+}
+
 /** Test-only: drain the module mutation chain between cases. */
 export function resetMobileRelayHostOverlayStoreForTests(): void {
  overlayMutation = Promise.resolve()
@@ -381,6 +381,33 @@ describe('mobile relay RPC session', () => {
    expect(session.getState()).toBe('auth-failed')
  })

+  it('keeps concurrent written RPCs delivery-ambiguous when one request is unauthorized', async () => {
+    const { session } = await authenticateSession()
+    const rejected = session.sendRequest('status.get').catch((error: unknown) => error)
+    const concurrent = session
+      .sendRequest('terminal.send', { terminal: 'term', text: 'hi' })
+      .catch((error: unknown) => error)
+    await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledTimes(2))
+    const requests = fakes.sendText.mock.calls.map(
+      ([payload]) => JSON.parse(payload as string) as { id: string; method: string }
+    )
+    const rejectedRequest = requests.find(({ method }) => method === 'status.get')!
+
+    fakes.linkOptions!.onText(
+      JSON.stringify({
+        id: rejectedRequest.id,
+        ok: false,
+        error: { code: 'unauthorized', message: 'Invalid device token' },
+        _meta: {}
+      })
+    )
+
+    await expect(rejected).resolves.toBeInstanceOf(MobileE2EEAuthenticationError)
+    await expect(rejected.then(isRpcDeliveryUnknown)).resolves.toBe(false)
+    await expect(concurrent.then(isRpcDeliveryUnknown)).resolves.toBe(true)
+    expect(session.getState()).toBe('auth-failed')
+  })
+
  it('keeps a Relay session when a fresh timeout probe proves it live', async () => {
    const { session } = await authenticateSession()
    vi.useFakeTimers()
@@ -193,11 +193,18 @@ export function connectMobileRelayRpcSession(args: {
    if (!isRpcResponse(value)) {
      return
    }
+    const request = pending.get(value.id)
    if (!value.ok && value.error.code === 'unauthorized') {
-      fail(new MobileE2EEAuthenticationError())
+      const error = new MobileE2EEAuthenticationError()
+      if (request) {
+        clearTimeout(request.timer)
+        pending.delete(value.id)
+        request.reject(error)
+      }
+      // Why: only the rejected request is definite; concurrent written RPCs may have executed.
+      fail(error, new Error(error.message))
      return
    }
-    const request = pending.get(value.id)
    if (request || timedOutControlRequestIds.delete(value.id)) {
      controlResponseSequence += 1
    }
@@ -275,7 +282,7 @@ export function connectMobileRelayRpcSession(args: {
    }
  }

-  function fail(error: Error): void {
+  function fail(error: Error, pendingError = error): void {
    if (closed) {
      return
    }
@@ -283,7 +290,7 @@ export function connectMobileRelayRpcSession(args: {
    stopControlProbeTimer()
    failure = error
    link.close()
-    rejectPending(error)
+    rejectPending(pendingError)
    publishState(error instanceof MobileE2EEAuthenticationError ? 'auth-failed' : 'disconnected')
  }

@@ -294,9 +301,7 @@ export function connectMobileRelayRpcSession(args: {
    // 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.
-    if (!(error instanceof MobileE2EEAuthenticationError)) {
-      markRpcDeliveryUnknown(error)
-    }
+    markRpcDeliveryUnknown(error)
    for (const request of pending.values()) {
      clearTimeout(request.timer)
      request.reject(error)