refactor(mobile): share one desktop server-list feed between the host list and workspaces

A server workspace polled mobileRelay.hosts.list beside the host list's own read, so
the phone could ask the desktop twice per tick and the row badge and the screen could
disagree. One feed per desktop client now owns that read; both readers join it.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb
This commit is contained in:
Jinwoo-H
2026-10-11 05:09:55 -04:00
parent d6fc1c34de
commit e5cd05ffe6
8 changed files with 238 additions and 54 deletions
@@ -2,6 +2,8 @@ import { createElement } from 'react'
import { act, create } from 'react-test-renderer'
import { describe, expect, it, vi } from 'vitest'
// The desktop's server-list feed loads react-native; its timer never runs without a subscriber.
vi.mock('react-native', () => ({}))
vi.mock('expo-router', () => ({ useFocusEffect: () => {} }))
vi.mock('../cache/worktree-cache', () => ({
getCachedWorktrees: () => null,
@@ -2,6 +2,8 @@ import { createElement } from 'react'
import { act, create } from 'react-test-renderer'
import { describe, expect, it, vi } from 'vitest'
// The desktop's server-list feed loads react-native; its timer never runs without a subscriber.
vi.mock('react-native', () => ({}))
vi.mock('../cache/worktree-cache', () => ({
getCachedWorktrees: () => null,
setCachedWorktrees: () => {}
@@ -2,6 +2,8 @@ import { createElement } from 'react'
import { act, create } from 'react-test-renderer'
import { describe, expect, it, vi } from 'vitest'
// The desktop's server-list feed loads react-native; its timer never runs without a subscriber.
vi.mock('react-native', () => ({}))
vi.mock('expo-router', () => ({ useFocusEffect: () => {} }))
vi.mock('../cache/worktree-cache', () => ({ setCachedWorktrees: () => {} }))
vi.mock('../storage/preferences', () => ({ savePinnedIds: async () => {} }))
@@ -230,7 +230,7 @@ describe('a server workspace', () => {
})
it('reads as disconnected while the desktop reports its server down, and recovers', async () => {
vi.useFakeTimers({ toFake: ['setInterval', 'clearInterval'] })
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] })
try {
route.params = { hostId: 'host-1', executionHost: 'runtime:vm' }
desktop()
+15 -48
View File
@@ -1,9 +1,7 @@
import { useEffect, useState } from 'react'
import { AppState } from 'react-native'
import { useCallback, useSyncExternalStore } from 'react'
import type { ExecutionHostId } from '../../../src/shared/execution-host'
import type { ExecutionHostHealth } from '../../../src/shared/execution-host-health'
import { relayHostsFeed } from '../worktree/relay-hosts-feed'
import { serverHostHealth } from '../worktree/server-workspaces'
import { relayHostsListRead } from '../worktree/server-workspace-operations'
import {
scopeRpcClientToExecutionHost,
workspaceRouteExecutionHost
@@ -34,8 +32,7 @@ export function useWorkspaceServer(args: {
return { routeHost, server: routeHost === undefined ? null : { client: serverClient, gates } }
}
// The host list's poll cadence, so a workspace screen and its row agree within one tick.
const SERVER_HEALTH_POLL_MS = 3000
const noSubscription = (): (() => void) => () => {}
/**
* The connection state a server workspace's screens see: the desktop's, unless the desktop reports
@@ -47,50 +44,20 @@ export function useWorkspaceServerConnState(args: {
routeHost: ExecutionHostId | undefined | null
}): ConnectionState {
const { client, connState, routeHost } = args
const [reported, setReported] = useState<{
client: RpcClient
routeHost: ExecutionHostId
health: ExecutionHostHealth
} | null>(null)
useEffect(() => {
if (!client || connState !== 'connected' || !routeHost) {
return
}
let stopped = false
const read = async (): Promise<void> => {
if (AppState.currentState !== 'active') {
return
}
try {
const listed = relayHostsListRead.interpret(await relayHostsListRead.request(client))
if (stopped || !listed.accepted) {
return
}
const host = listed.value.find((entry) => entry.hostId === routeHost)
// A server the desktop no longer lists cannot be reached through it.
setReported({ client, routeHost, health: host ? serverHostHealth(host) : 'disconnected' })
} catch {
// Keeps the last answer; the next poll retries.
}
}
void read()
const timer = setInterval(() => void read(), SERVER_HEALTH_POLL_MS)
return () => {
stopped = true
clearInterval(timer)
}
}, [client, connState, routeHost])
// Why: keyed by client and server so a reused route never shows another server's answer.
if (
connState !== 'connected' ||
reported?.client !== client ||
reported.routeHost !== routeHost
) {
// The same feed the host list reads, so a row's badge and its screens agree.
const feed = client && routeHost ? relayHostsFeed(client) : null
const subscribe = useCallback(
(listener: () => void) => (feed ? feed.subscribe(listener) : noSubscription()),
[feed]
)
const readHosts = () => feed?.getSnapshot() ?? null
const hosts = useSyncExternalStore(subscribe, readHosts, readHosts)
if (connState !== 'connected' || !routeHost || !hosts) {
return connState
}
switch (reported.health) {
const host = hosts.find((entry) => entry.hostId === routeHost)
// A server the desktop no longer lists cannot be reached through it.
switch (host ? serverHostHealth(host) : 'disconnected') {
case 'local':
case 'available':
return 'connected'
@@ -0,0 +1,81 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
vi.mock('react-native', () => ({ AppState: { currentState: 'active' } }))
import { MOBILE_RELAY_HOSTS_LIST_METHOD } from '../../../src/shared/mobile-relay-hosts-contract'
import { FakeSession } from '../transport/mobile-endpoint-supervisor-test-fakes'
import { relayHostsFeed } from './relay-hosts-feed'
function desktop(health: () => string): FakeSession {
const session = new FakeSession('connected')
session.sendRequest.mockImplementation(async () => ({
id: 'reply',
ok: true,
result: { hosts: [{ hostId: 'runtime:vm', label: 'vm', health: health(), relay: 'ready' }] },
_meta: { runtimeId: 'runtime-1' }
}))
return session
}
function listCalls(session: FakeSession): number {
return session.sendRequest.mock.calls.filter(
([method]) => method === MOBILE_RELAY_HOSTS_LIST_METHOD
).length
}
describe('the desktop’s server list', () => {
beforeEach(() => {
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] })
})
afterEach(() => {
vi.useRealTimers()
})
it('is read once per poll however many readers share it', async () => {
const session = desktop(() => 'available')
const feed = relayHostsFeed(session)
// The host list reads on its own poll; an open workspace subscribes a second later.
await feed.read()
await vi.advanceTimersByTimeAsync(1000)
const unsubscribe = feed.subscribe(() => {})
const unsubscribeAgain = feed.subscribe(() => {})
expect(listCalls(session)).toBe(1)
// Its timed read waits out the host list's, then both readers share each one.
await vi.advanceTimersByTimeAsync(1999)
expect(listCalls(session)).toBe(1)
await vi.advanceTimersByTimeAsync(1)
expect(listCalls(session)).toBe(2)
await vi.advanceTimersByTimeAsync(3000)
expect(listCalls(session)).toBe(3)
unsubscribe()
unsubscribeAgain()
await vi.advanceTimersByTimeAsync(9000)
expect(listCalls(session)).toBe(3)
})
it('joins a read already in flight', async () => {
const session = desktop(() => 'available')
const feed = relayHostsFeed(session)
const [first, second] = await Promise.all([feed.read(), feed.read()])
expect(second).toBe(first)
expect(listCalls(session)).toBe(1)
})
it('forgets its answer on a drop, and reads again on reconnect', async () => {
let health = 'available'
const session = desktop(() => health)
const feed = relayHostsFeed(session)
const unsubscribe = feed.subscribe(() => {})
await vi.advanceTimersByTimeAsync(0)
expect(feed.getSnapshot()?.[0]?.health).toBe('available')
session.publishState('reconnecting')
expect(feed.getSnapshot()).toBeNull()
await vi.advanceTimersByTimeAsync(9000)
expect(listCalls(session)).toBe(1)
health = 'error'
session.publishState('connected')
await vi.advanceTimersByTimeAsync(0)
expect(feed.getSnapshot()?.[0]?.health).toBe('error')
unsubscribe()
})
})
+129
View File
@@ -0,0 +1,129 @@
// The paired desktop's servers (`mobileRelay.hosts.list`), one feed per desktop client: the host
// list and every open server workspace share one poll and one answer, so a row's badge and its
// workspace screen never disagree and two readers never double the desktop's load.
import { AppState } from 'react-native'
import type { MobileRelayHost } from '../../../src/shared/mobile-relay-hosts-contract'
import type { RpcClient } from '../transport/rpc-client'
import { relayHostsListRead } from './server-workspace-operations'
// The host list's own poll cadence; a read by any reader pushes the next timed one back.
const RELAY_HOSTS_POLL_MS = 3000
export type RelayHostsFeed = {
subscribe: (listener: () => void) => () => void
/** The last answer on this connection; null before one lands. */
getSnapshot: () => readonly MobileRelayHost[] | null
/** Reads now, or joins the read in flight; null when the desktop did not answer. */
read: () => Promise<readonly MobileRelayHost[] | null>
}
const feeds = new WeakMap<RpcClient, RelayHostsFeed>()
function createFeed(client: RpcClient): RelayHostsFeed {
let snapshot: readonly MobileRelayHost[] | null = null
let inFlight: Promise<readonly MobileRelayHost[] | null> | null = null
let lastReadAt = Number.NEGATIVE_INFINITY
// Bumped on every drop, so a read from an earlier connection never lands.
let generation = 0
let timer: ReturnType<typeof setTimeout> | null = null
const listeners = new Set<() => void>()
const publish = (next: readonly MobileRelayHost[] | null): void => {
// Why: an unchanged poll keeps its identity, so readers do not re-render every tick.
if (JSON.stringify(next) === JSON.stringify(snapshot)) {
return
}
snapshot = next
for (const listener of listeners) {
listener()
}
}
const stopTimer = (): void => {
if (timer) {
clearTimeout(timer)
timer = null
}
}
// Timed reads run only while someone subscribes and the desktop is connected.
const schedule = (): void => {
stopTimer()
if (listeners.size === 0 || client.getState() !== 'connected') {
return
}
const delay = lastReadAt + RELAY_HOSTS_POLL_MS - Date.now()
if (delay <= 0) {
tick()
return
}
timer = setTimeout(tick, delay)
}
const tick = (): void => {
timer = null
if (AppState.currentState === 'active') {
void read()
return
}
timer = setTimeout(tick, RELAY_HOSTS_POLL_MS)
}
const fetchHosts = async (readGeneration: number): Promise<readonly MobileRelayHost[] | null> => {
try {
const listed = relayHostsListRead.interpret(await relayHostsListRead.request(client))
// A reply from before a drop is not this connection's answer.
if (!listed.accepted || readGeneration !== generation) {
return null
}
publish(listed.value)
return listed.value
} catch {
return null
}
}
const read = (): Promise<readonly MobileRelayHost[] | null> => {
if (inFlight) {
return inFlight
}
lastReadAt = Date.now()
const readGeneration = generation
inFlight = fetchHosts(readGeneration).finally(() => {
if (readGeneration === generation) {
inFlight = null
}
})
schedule()
return inFlight
}
client.onStateChange((state) => {
if (state !== 'connected') {
generation += 1
inFlight = null
lastReadAt = Number.NEGATIVE_INFINITY
publish(null)
}
schedule()
})
return {
subscribe: (listener) => {
listeners.add(listener)
schedule()
return () => {
listeners.delete(listener)
if (listeners.size === 0) {
stopTimer()
}
}
},
getSnapshot: () => snapshot,
read
}
}
export function relayHostsFeed(client: RpcClient): RelayHostsFeed {
let feed = feeds.get(client)
if (!feed) {
feed = createFeed(client)
feeds.set(client, feed)
}
return feed
}
+6 -5
View File
@@ -3,7 +3,9 @@ import type { ExecutionHostId } from '../../../src/shared/execution-host'
import type { ExecutionHostHealth } from '../../../src/shared/execution-host-health'
import type { MobileRelayHost } from '../../../src/shared/mobile-relay-hosts-contract'
import type { RpcOperationSender } from '../transport/rpc-operation-sender'
import { relayHostsListRead, relayHostWorktreesRead } from './server-workspace-operations'
import type { RpcClient } from '../transport/rpc-client'
import { relayHostsFeed } from './relay-hosts-feed'
import { relayHostWorktreesRead } from './server-workspace-operations'
import type { Worktree } from './workspace-list-types'
import { areWorktreeListsEqual } from './worktree-list-snapshot'
@@ -60,13 +62,12 @@ export type FetchedServerWorkspaces = {
/** The servers' rows through the desktop's one listing path; null when it did not answer. */
export async function fetchServerWorkspaces(
client: RpcOperationSender
client: RpcClient
): Promise<FetchedServerWorkspaces | null> {
const listed = relayHostsListRead.interpret(await relayHostsListRead.request(client))
if (!listed.accepted) {
const hosts = await relayHostsFeed(client).read()
if (!hosts) {
return null
}
const hosts = listed.value
return { hosts, rows: await Promise.all(hosts.map((host) => fetchHostRows(client, host.hostId))) }
}