diff --git a/src/main/ipc/runtime-environment-mobile-relay-hosts.test.ts b/src/main/ipc/runtime-environment-mobile-relay-hosts.test.ts index 3e6ace75f71..fb0c26b5218 100644 --- a/src/main/ipc/runtime-environment-mobile-relay-hosts.test.ts +++ b/src/main/ipc/runtime-environment-mobile-relay-hosts.test.ts @@ -83,14 +83,16 @@ describe('runtime environment mobile relay hosts', () => { name: 'VM', source: undefined, orcadDeployment: undefined, - fence: '7\0runtime-a' + pairingRevision: 7, + runtimeId: 'runtime-a' }, { id: 'env-2', name: 'Box', source: 'ephemeral-vm', orcadDeployment: undefined, - fence: '2\0null' + pairingRevision: 2, + runtimeId: null } ]) expect([...listing.statusByEnvironmentId.keys()]).toEqual(['env-1']) @@ -122,7 +124,36 @@ describe('runtime environment mobile relay hosts', () => { mocks.call.mockResolvedValue({ id: 'x', ok: true, result: {}, _meta: { runtimeId: 'r' } }) const host = (await hosts.resolve('env-1'))! await hosts.call(host, 'status.get', undefined) - expect(mocks.call).toHaveBeenCalledWith('/user-data', 'env-1', 'status.get', undefined) + expect(mocks.call).toHaveBeenLastCalledWith( + '/user-data', + 'env-1', + 'status.get', + undefined, + undefined, + undefined, + undefined, + undefined + ) + // A bounded call refused by the transport once the server was re-paired or replaced. + await hosts.call( + { environmentId: 'env-1' }, + 'worktree.ps', + {}, + { + timeoutMs: 5_000, + expected: { pairingRevision: 7, runtimeId: 'runtime-a' } + } + ) + expect(mocks.call).toHaveBeenLastCalledWith( + '/user-data', + 'env-1', + 'worktree.ps', + {}, + 5_000, + 7, + undefined, + { expectedEnvironmentRuntimeId: 'runtime-a' } + ) const retired: string[] = [] const stop = hosts.onEnvironmentRetired((id) => retired.push(id)) diff --git a/src/main/ipc/runtime-environment-mobile-relay-hosts.ts b/src/main/ipc/runtime-environment-mobile-relay-hosts.ts index a8c288ed4b9..2a6ac007709 100644 --- a/src/main/ipc/runtime-environment-mobile-relay-hosts.ts +++ b/src/main/ipc/runtime-environment-mobile-relay-hosts.ts @@ -43,7 +43,8 @@ export function createRuntimeEnvironmentMobileRelayHosts(): MobileDesktopRelayHo name: environment.name, source: environment.source, orcadDeployment: environment.orcadDeployment, - fence: relayHostFence(environment) + pairingRevision: pairingRevision(environment), + runtimeId: environment.runtimeId })), statusByEnvironmentId } @@ -59,8 +60,19 @@ export function createRuntimeEnvironmentMobileRelayHosts(): MobileDesktopRelayHo pairing: getPreferredPairingOffer(environment) } }, - call: (host, method, params) => - callRuntimeEnvironment(userDataPath, host.environmentId, method, params), + call: (host, method, params, options) => + callRuntimeEnvironment( + userDataPath, + host.environmentId, + method, + params, + options?.timeoutMs, + options?.expected?.pairingRevision, + undefined, + options?.expected?.runtimeId + ? { expectedEnvironmentRuntimeId: options.expected.runtimeId } + : undefined + ), onEnvironmentRetired: (listener) => { retirementListeners.add(listener) return () => retirementListeners.delete(listener) diff --git a/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay-hosts.ts b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay-hosts.ts index e0653cd196c..fd86049eed7 100644 --- a/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay-hosts.ts +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay-hosts.ts @@ -18,7 +18,14 @@ export type MobileDesktopRelayHostListing = { name: string source?: RuntimeEnvironmentSource orcadDeployment?: { sshTargetId: string } | null - fence: string + pairingRevision: number + runtimeId: string | null +} + +/** Bounds a call; `expected` refuses it once the server was re-paired or its runtime changed. */ +export type MobileDesktopRelayCallOptions = { + timeoutMs?: number + expected?: Pick } /** The desktop's configured servers, as the relay sees them. */ @@ -32,9 +39,10 @@ export type MobileDesktopRelayHosts = { resolve: (environmentId: string) => Promise /** A call as the desktop itself, over its own connection to the server. */ call: ( - host: MobileDesktopRelayHost, + host: Pick, method: string, - params: unknown + params: unknown, + options?: MobileDesktopRelayCallOptions ) => Promise> /** Fires when a server is removed, re-paired or disconnected; returns an unsubscribe. */ onEnvironmentRetired: (listener: (environmentId: string) => void) => () => void diff --git a/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts index c0b3147e1bd..ac401f19885 100644 --- a/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts +++ b/src/main/runtime/mobile-desktop-relay/mobile-desktop-relay.integration.test.ts @@ -144,7 +144,12 @@ describe('mobile desktop relay: phone -> desktop -> server', () => { const registry = host.server.getDeviceRegistry()! const children = () => registry.listDelegatedMobileDevices(desktopOnHost.pairedDeviceId!) const syncedNames: string[][] = [] - const state = { capable: true, retire: (_id: string) => {} } + const retirementListeners = new Set<(environmentId: string) => void>() + const state = { + capable: true, + retire: (environmentId: string) => + retirementListeners.forEach((listener) => listener(environmentId)) + } const hosts: MobileDesktopRelayHosts = { list: () => ({ environments: [], statusByEnvironmentId: new Map() }), resolve: async (environmentId) => @@ -166,8 +171,8 @@ describe('mobile desktop relay: phone -> desktop -> server', () => { : response }, onEnvironmentRetired: (listener) => { - state.retire = listener - return () => {} + retirementListeners.add(listener) + return () => retirementListeners.delete(listener) } } desktop.server.setMobileDesktopRelayHosts(hosts) diff --git a/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.integration.test.ts b/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.integration.test.ts index 79b8bf77eb2..df25ac02f72 100644 --- a/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.integration.test.ts +++ b/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.integration.test.ts @@ -115,7 +115,12 @@ describe('mobile relay hosts: phone -> desktop lists the servers it shows', () = ]) const hosts: MobileDesktopRelayHosts = { list: () => ({ - environments: [...servers].map(([id, { name }]) => ({ id, name, fence: 'pairing-1' })), + environments: [...servers].map(([id, { name }]) => ({ + id, + name, + pairingRevision: 1, + runtimeId: id + })), statusByEnvironmentId: new Map( [...servers].map(([id, { capable }]) => [ id, diff --git a/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.test.ts b/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.test.ts index 0016054a0ca..67ff379718b 100644 --- a/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.test.ts +++ b/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.test.ts @@ -51,11 +51,16 @@ function psReply(rows: Record[]): RuntimeRpcResponse { function fakeHosts( environments: { id: string; name: string; snapshot: RuntimeHostStatusSnapshot }[] ) { - const state = { environments, fence: 'fence-1' } + const state = { environments, pairingRevision: 1, retire: (_environmentId: string) => {} } const call = vi.fn() const hosts: MobileDesktopRelayHosts = { list: () => ({ - environments: state.environments.map(({ id, name }) => ({ id, name, fence: state.fence })), + environments: state.environments.map(({ id, name }) => ({ + id, + name, + pairingRevision: state.pairingRevision, + runtimeId: 'runtime-a' + })), statusByEnvironmentId: new Map( state.environments.map((entry) => [ entry.id, @@ -65,11 +70,14 @@ function fakeHosts( }), resolve: async (environmentId) => ({ environmentId, - fence: state.fence, + fence: 'unused', pairing: { v: 2, endpoint: 'ws://server', deviceToken: 'desktop', publicKeyB64: 'k' } }), call, - onEnvironmentRetired: () => () => {} + onEnvironmentRetired: (listener) => { + state.retire = listener + return () => {} + } } return { hosts, call, state } } @@ -147,9 +155,9 @@ describe('mobile relay host catalog', () => { expect(call).toHaveBeenCalledWith( expect.objectContaining({ environmentId: 'env' }), 'worktree.ps', - { - supportsWorktreeVisibilitySourceDefaults: true - } + // The phone's full-catalog limit: the server's default page would silently truncate. + { limit: 10_000, supportsWorktreeVisibilitySourceDefaults: true }, + { timeoutMs: 5_000, expected: { pairingRevision: 1, runtimeId: 'runtime-a' } } ) }) @@ -206,11 +214,66 @@ describe('mobile relay host catalog', () => { state.environments = [{ id: 'env', name: 'Box', snapshot: snapshot('env', 'live') }] call.mockResolvedValue(psReply([{ worktreeId: 'w', hostId: 'local' }])) await hostCatalog.worktrees('runtime:env') - state.fence = 'fence-2' + state.pairingRevision = 2 state.environments = [{ id: 'env', name: 'Box', snapshot: snapshot('env', 'unreachable') }] await expect(hostCatalog.worktrees('runtime:env')).resolves.toEqual({ worktrees: null }) }) + it('has no rows when the server is re-paired while its fetch is in flight', async () => { + const { hosts, call, state } = fakeHosts([ + { id: 'env', name: 'Box', snapshot: snapshot('env', 'live') } + ]) + let answerFetch: (response: RuntimeRpcResponse) => void = () => {} + call.mockReturnValue(new Promise((resolve) => (answerFetch = resolve))) + const { catalog: hostCatalog } = catalog(hosts) + const rows = hostCatalog.worktrees('runtime:env') + await vi.waitFor(() => expect(call).toHaveBeenCalledTimes(1)) + state.pairingRevision = 2 + state.retire('env') + answerFetch(psReply([{ worktreeId: 'old-server-row', hostId: 'local' }])) + await expect(rows).resolves.toEqual({ worktrees: null }) + }) + + it("drops a removed server's rows, so a server added back under the same id starts empty", async () => { + const { hosts, call, state } = fakeHosts([ + { id: 'env', name: 'Box', snapshot: snapshot('env', 'live') } + ]) + call.mockResolvedValue(psReply([{ worktreeId: 'w', hostId: 'local' }])) + const { catalog: hostCatalog } = catalog(hosts) + await hostCatalog.worktrees('runtime:env') + const removed = state.environments + state.environments = [] + state.retire('env') + await expect(hostCatalog.worktrees('runtime:env')).resolves.toEqual({ worktrees: null }) + state.environments = [{ ...removed[0]!, snapshot: snapshot('env', 'unreachable') }] + await expect(hostCatalog.worktrees('runtime:env')).resolves.toEqual({ worktrees: null }) + }) + + it('answers with the last rows, marked stale, when the server never answers the fetch', async () => { + vi.useFakeTimers() + try { + const { hosts, call } = fakeHosts([ + { id: 'env', name: 'Box', snapshot: snapshot('env', 'live') } + ]) + call.mockResolvedValueOnce(psReply([{ worktreeId: 'w', hostId: 'local' }])) + const { catalog: hostCatalog } = catalog(hosts) + await hostCatalog.worktrees('runtime:env') + call.mockReturnValue(new Promise(() => {})) + let settled = false + const rows = hostCatalog.worktrees('runtime:env').finally(() => (settled = true)) + await vi.advanceTimersByTimeAsync(4_999) + expect(settled).toBe(false) + await vi.advanceTimersByTimeAsync(1) + await expect(rows).resolves.toMatchObject({ + worktrees: [{ worktreeId: 'w', hostId: 'runtime:env' }], + fetchedAt: 1_000, + stale: true + }) + } finally { + vi.useRealTimers() + } + }) + it('shares one in-flight fetch per server between concurrent rows requests', async () => { const { hosts, call } = fakeHosts([ { id: 'env', name: 'Box', snapshot: snapshot('env', 'live') } diff --git a/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.ts b/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.ts index 6730ebde725..7d013ac96a9 100644 --- a/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.ts +++ b/src/main/runtime/mobile-desktop-relay/mobile-relay-host-catalog.ts @@ -15,7 +15,15 @@ import type { } from '../../../shared/mobile-relay-hosts-contract' import { lastVerifiedRuntimeStatus } from '../../../shared/runtime-host-status' import type { RuntimeStatus } from '../../../shared/runtime-types' -import type { MobileDesktopRelayHosts } from './mobile-desktop-relay-hosts' +import type { + MobileDesktopRelayHostListing, + MobileDesktopRelayHosts +} from './mobile-desktop-relay-hosts' + +// Why the phone's own full-catalog limit: the server's default page is 200 and nothing pages on. +const SERVER_WORKTREE_LIMIT = 10_000 +// Why well under the phone's 30s request timeout: a hung server must answer with its last rows. +const SERVER_FETCH_TIMEOUT_MS = 5_000 const WorktreePsReplySchema = z.object({ worktrees: z.array(z.looseObject({})), @@ -31,7 +39,10 @@ type CachedServerWorktrees = { truncated: boolean } -type DescribedHost = MobileRelayHost & { environmentId: string; fence: string } +type DescribedHost = MobileRelayHost & { + environmentId: string + identity: Pick +} /** * The configured servers the desktop shows, for the phone: health from the desktop's own status @@ -42,6 +53,7 @@ type DescribedHost = MobileRelayHost & { environmentId: string; fence: string } export class MobileRelayHostCatalog { private readonly cached = new Map() private readonly refreshing = new Map>() + private readonly stopRetirementWatch: () => void constructor( private readonly options: { @@ -49,7 +61,16 @@ export class MobileRelayHostCatalog { hostLabelOverrides: () => ReadonlyMap now?: () => number } - ) {} + ) { + this.stopRetirementWatch = options.hosts.onEnvironmentRetired((environmentId) => + this.cached.delete(environmentId) + ) + } + + dispose(): void { + this.stopRetirementWatch() + this.cached.clear() + } list(): MobileRelayHostsListResult { return { @@ -71,10 +92,13 @@ export class MobileRelayHostCatalog { if (!host) { return { worktrees: null } } - const refreshed = host.health === 'available' && (await this.refresh(host)) + const refreshed = host.health === 'available' && (await withinFetchBound(this.refresh(host))) const cached = this.cached.get(host.environmentId) - // Why: rows from before a re-pair may belong to a different server. - if (!cached || cached.fence !== host.fence) { + const current = this.options.hosts + .list() + .environments.find((environment) => environment.id === host.environmentId) + // Why the identity read after the fetch: one answered after a re-pair stored the old server's rows. + if (!cached || !current || cached.fence !== fenceOf(current)) { return { worktrees: null } } const { fence: _fence, ...rows } = cached @@ -83,7 +107,7 @@ export class MobileRelayHostCatalog { private describe(): DescribedHost[] { const { environments, statusByEnvironmentId } = this.options.hosts.list() - const fences = new Map(environments.map((environment) => [environment.id, environment.fence])) + const identities = new Map(environments.map((environment) => [environment.id, environment])) return buildExecutionHostRegistry({ repos: [], settings: null, @@ -93,8 +117,8 @@ export class MobileRelayHostCatalog { hostLabelOverrides: this.options.hostLabelOverrides() }).flatMap((entry) => { const parsed = parseExecutionHostId(entry.id) - const fence = parsed?.kind === 'runtime' ? fences.get(parsed.environmentId) : undefined - if (parsed?.kind !== 'runtime' || fence === undefined) { + const identity = parsed?.kind === 'runtime' ? identities.get(parsed.environmentId) : undefined + if (parsed?.kind !== 'runtime' || !identity) { return [] } const answer = lastVerifiedRuntimeStatus(statusByEnvironmentId.get(parsed.environmentId)) @@ -102,7 +126,7 @@ export class MobileRelayHostCatalog { { hostId: parsed.id, environmentId: parsed.environmentId, - fence, + identity: { pairingRevision: identity.pairingRevision, runtimeId: identity.runtimeId }, label: entry.label, health: entry.health, relay: relayVerdict(entry.health, answer) @@ -125,20 +149,19 @@ export class MobileRelayHostCatalog { private async fetchAsDesktop(host: DescribedHost): Promise { try { - const resolved = await this.options.hosts.resolve(host.environmentId) - if (!resolved) { - return false - } - const response = await this.options.hosts.call(resolved, 'worktree.ps', { - supportsWorktreeVisibilitySourceDefaults: true - }) + const response = await this.options.hosts.call( + host, + 'worktree.ps', + { limit: SERVER_WORKTREE_LIMIT, supportsWorktreeVisibilitySourceDefaults: true }, + { timeoutMs: SERVER_FETCH_TIMEOUT_MS, expected: host.identity } + ) const reply = response.ok ? WorktreePsReplySchema.safeParse(response.result) : null if (!reply?.success) { return false } this.cached.set(host.environmentId, { - // Why the listing's fence, taken before the fetch: a re-pair mid-fetch must not adopt these rows. - fence: host.fence, + // Why the identity taken before the fetch: the call's own check cannot see a re-pair after dispatch. + fence: fenceOf(host.identity), fetchedAt: (this.options.now ?? Date.now)(), worktrees: stampServerWorktreeRows(host.environmentId, reply.data.worktrees), totalCount: reply.data.totalCount, @@ -152,6 +175,23 @@ export class MobileRelayHostCatalog { } } +function fenceOf(identity: DescribedHost['identity']): string { + return `${identity.pairingRevision}\0${identity.runtimeId}` +} + +/** False once the bound passes; the call's own timeout starts only after the desktop's per-server queue. */ +async function withinFetchBound(refresh: Promise): Promise { + let timer: ReturnType | undefined + const bound = new Promise((resolve) => { + timer = setTimeout(() => resolve(false), SERVER_FETCH_TIMEOUT_MS) + }) + try { + return await Promise.race([refresh, bound]) + } finally { + clearTimeout(timer) + } +} + function relayVerdict( health: MobileRelayHost['health'], answer: RuntimeStatus | null diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts b/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts index 9db3697217d..097aa947bde 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-pairing.ts @@ -95,6 +95,7 @@ export class RuntimeRpcPairing extends RuntimeRpcNetworkExposure { allocateStreamId: allocateTerminalSubscriptionStreamId }) : null + this.mobileRelayHostCatalog?.dispose() this.mobileRelayHostCatalog = hosts ? new MobileRelayHostCatalog({ hosts,