diff --git a/mobile/src/host-screen/host-show-pinned-in-groups.test.ts b/mobile/src/host-screen/host-show-pinned-in-groups.test.ts index 376b8dbebc2..a225a1a48bd 100644 --- a/mobile/src/host-screen/host-show-pinned-in-groups.test.ts +++ b/mobile/src/host-screen/host-show-pinned-in-groups.test.ts @@ -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, diff --git a/mobile/src/host-screen/host-ssh-health-badge-wiring.test.ts b/mobile/src/host-screen/host-ssh-health-badge-wiring.test.ts index ffd7e73f63a..a47f15d359d 100644 --- a/mobile/src/host-screen/host-ssh-health-badge-wiring.test.ts +++ b/mobile/src/host-screen/host-ssh-health-badge-wiring.test.ts @@ -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: () => {} diff --git a/mobile/src/host-screen/host-worktree-catalog-action-error.test.ts b/mobile/src/host-screen/host-worktree-catalog-action-error.test.ts index 79edb5209e1..1cd6df0f5fc 100644 --- a/mobile/src/host-screen/host-worktree-catalog-action-error.test.ts +++ b/mobile/src/host-screen/host-worktree-catalog-action-error.test.ts @@ -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 () => {} })) diff --git a/mobile/src/navigation/workspace-route.test.tsx b/mobile/src/navigation/workspace-route.test.tsx index 13156e5685c..6159ef33f9d 100644 --- a/mobile/src/navigation/workspace-route.test.tsx +++ b/mobile/src/navigation/workspace-route.test.tsx @@ -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() diff --git a/mobile/src/transport/workspace-server.ts b/mobile/src/transport/workspace-server.ts index 32b52274331..4df40813200 100644 --- a/mobile/src/transport/workspace-server.ts +++ b/mobile/src/transport/workspace-server.ts @@ -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 => { - 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' diff --git a/mobile/src/worktree/relay-hosts-feed.test.ts b/mobile/src/worktree/relay-hosts-feed.test.ts new file mode 100644 index 00000000000..ee5f00d6a96 --- /dev/null +++ b/mobile/src/worktree/relay-hosts-feed.test.ts @@ -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() + }) +}) diff --git a/mobile/src/worktree/relay-hosts-feed.ts b/mobile/src/worktree/relay-hosts-feed.ts new file mode 100644 index 00000000000..adca7ae3f1a --- /dev/null +++ b/mobile/src/worktree/relay-hosts-feed.ts @@ -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 +} + +const feeds = new WeakMap() + +function createFeed(client: RpcClient): RelayHostsFeed { + let snapshot: readonly MobileRelayHost[] | null = null + let inFlight: Promise | 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 | 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 => { + 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 => { + 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 +} diff --git a/mobile/src/worktree/server-workspaces.ts b/mobile/src/worktree/server-workspaces.ts index b1eea4d0816..7ad1ffda130 100644 --- a/mobile/src/worktree/server-workspaces.ts +++ b/mobile/src/worktree/server-workspaces.ts @@ -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 { - 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))) } }