mirror of
https://github.com/stablyai/orca.git
synced 2026-11-02 10:08:13 +00:00
fix(mobile): fetch a server's full workspace list within a bound, and forget removed servers
The desktop now asks for the phone's full-catalog limit, answers within 5s with the last rows when a server hangs, drops a server's rows when it is removed or re-paired, and checks the server's identity after the fetch so a reply from before a re-pair is never served. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb
This commit is contained in:
@@ -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))
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<MobileDesktopRelayHostListing, 'pairingRevision' | 'runtimeId'>
|
||||
}
|
||||
|
||||
/** The desktop's configured servers, as the relay sees them. */
|
||||
@@ -32,9 +39,10 @@ export type MobileDesktopRelayHosts = {
|
||||
resolve: (environmentId: string) => Promise<MobileDesktopRelayHost | null>
|
||||
/** A call as the desktop itself, over its own connection to the server. */
|
||||
call: (
|
||||
host: MobileDesktopRelayHost,
|
||||
host: Pick<MobileDesktopRelayHost, 'environmentId'>,
|
||||
method: string,
|
||||
params: unknown
|
||||
params: unknown,
|
||||
options?: MobileDesktopRelayCallOptions
|
||||
) => Promise<RuntimeRpcResponse<unknown>>
|
||||
/** Fires when a server is removed, re-paired or disconnected; returns an unsubscribe. */
|
||||
onEnvironmentRetired: (listener: (environmentId: string) => void) => () => void
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -51,11 +51,16 @@ function psReply(rows: Record<string, unknown>[]): RuntimeRpcResponse<unknown> {
|
||||
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<MobileDesktopRelayHosts['call']>()
|
||||
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<unknown>) => 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') }
|
||||
|
||||
@@ -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<MobileDesktopRelayHostListing, 'pairingRevision' | 'runtimeId'>
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<string, CachedServerWorktrees>()
|
||||
private readonly refreshing = new Map<string, Promise<boolean>>()
|
||||
private readonly stopRetirementWatch: () => void
|
||||
|
||||
constructor(
|
||||
private readonly options: {
|
||||
@@ -49,7 +61,16 @@ export class MobileRelayHostCatalog {
|
||||
hostLabelOverrides: () => ReadonlyMap<ExecutionHostId, string>
|
||||
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<boolean> {
|
||||
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<boolean>): Promise<boolean> {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
const bound = new Promise<false>((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
|
||||
|
||||
@@ -95,6 +95,7 @@ export class RuntimeRpcPairing extends RuntimeRpcNetworkExposure {
|
||||
allocateStreamId: allocateTerminalSubscriptionStreamId
|
||||
})
|
||||
: null
|
||||
this.mobileRelayHostCatalog?.dispose()
|
||||
this.mobileRelayHostCatalog = hosts
|
||||
? new MobileRelayHostCatalog({
|
||||
hosts,
|
||||
|
||||
Reference in New Issue
Block a user