diff --git a/src/renderer/src/runtime/runtime-environment-ssh-state-read-concurrency.test.ts b/src/renderer/src/runtime/runtime-environment-ssh-state-read-concurrency.test.ts new file mode 100644 index 00000000000..88d8f4d313a --- /dev/null +++ b/src/renderer/src/runtime/runtime-environment-ssh-state-read-concurrency.test.ts @@ -0,0 +1,138 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { RuntimeRpcCallQueuePool } from '../../../shared/runtime-rpc-call-queue' +import { useAppStore } from '@/store' +import { + hydrateRuntimeEnvironmentSshState, + SSH_STATE_READ_CONCURRENCY +} from './runtime-environment-ssh-state' +import { callRuntimeRpc } from './runtime-rpc-client' + +vi.mock('./runtime-rpc-client', async (importOriginal) => ({ + ...(await importOriginal>()), + callRuntimeRpc: vi.fn() +})) + +const callRuntimeRpcMock = vi.mocked(callRuntimeRpc) +const TARGET_COUNT = 8 + +// The hydrator admits the wire state itself, so fixtures stay untyped wire payloads. +function connState(targetId: string): Record { + return { + targetId, + status: 'connected', + error: null, + reconnectAttempt: 0, + providerEpoch: `${targetId}-provider-epoch`, + connectionGeneration: 1 + } +} + +function rpcResult(value: unknown): Promise { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: callRuntimeRpc is generic in its result; each fixture returns the shape its method's caller reads. + return Promise.resolve(value as T) +} + +function targetIdOf(params: unknown): string { + return typeof params === 'object' && params !== null && 'targetId' in params + ? String(params.targetId) + : '' +} + +type HeldRead = { targetId: string; release: () => void } + +// Routes calls through the production queue so lane occupancy is the real one. +function installHeldStateReads(pool: RuntimeRpcCallQueuePool, envId: string): HeldRead[] { + const held: HeldRead[] = [] + const targets = Array.from({ length: TARGET_COUNT }, (_, index) => ({ + id: `ssh-${index}`, + label: `box ${index}` + })) + callRuntimeRpcMock.mockImplementation((_target, method, params) => + pool.enqueue(envId, method, () => { + if (method === 'ssh.listTargetSummaries') { + return rpcResult({ targets }) + } + if (method === 'ssh.listRemovedTargetLabels') { + return rpcResult({ labels: {} }) + } + const targetId = targetIdOf(params) + return new Promise((resolve) => { + held.push({ targetId, release: resolve }) + }).then(() => rpcResult({ state: connState(targetId) })) + }) + ) + return held +} + +async function flush(): Promise { + for (let i = 0; i < 20; i += 1) { + await Promise.resolve() + } +} + +beforeEach(() => { + useAppStore.setState(useAppStore.getInitialState(), true) + callRuntimeRpcMock.mockReset() +}) + +describe('remote SSH state reads', () => { + it('reads several targets at once instead of waiting on each in turn', async () => { + const envId = 'env-concurrent' + const held = installHeldStateReads(new RuntimeRpcCallQueuePool(), envId) + + const hydration = hydrateRuntimeEnvironmentSshState(envId) + await flush() + expect(held).toHaveLength(SSH_STATE_READ_CONCURRENCY) + + let maxInFlight = held.length + for (let read = held.shift(); read; read = held.shift()) { + read.release() + await flush() + maxInFlight = Math.max(maxInFlight, held.length) + } + await hydration + + expect(maxInFlight).toBe(SSH_STATE_READ_CONCURRENCY) + const states = useAppStore.getState().sshStateByEnvironment.get(envId)?.connectionStates + expect(states?.size).toBe(TARGET_COUNT) + }) + + it('leaves foreground capacity for an interactive call while every read is held', async () => { + const envId = 'env-lane' + const pool = new RuntimeRpcCallQueuePool() + const held = installHeldStateReads(pool, envId) + + const hydration = hydrateRuntimeEnvironmentSshState(envId) + await flush() + const send = vi.fn(() => Promise.resolve('sent')) + const sendResult = pool.enqueue(envId, 'terminal.send', send) + await flush() + + expect(send).toHaveBeenCalledTimes(1) + await expect(sendResult).resolves.toBe('sent') + for (let read = held.shift(); read; read = held.shift()) { + read.release() + await flush() + } + await hydration + }) + + it('stops starting reads once the environment is marked stale', async () => { + const envId = 'env-stale' + const held = installHeldStateReads(new RuntimeRpcCallQueuePool(), envId) + + const hydration = hydrateRuntimeEnvironmentSshState(envId) + await flush() + useAppStore.getState().markEnvironmentSshStateStale(envId) + for (let read = held.shift(); read; read = held.shift()) { + read.release() + await flush() + } + await hydration + + const stateReads = callRuntimeRpcMock.mock.calls.filter( + ([, method]) => method === 'ssh.getState' + ) + expect(stateReads).toHaveLength(SSH_STATE_READ_CONCURRENCY) + }) +}) diff --git a/src/renderer/src/runtime/runtime-environment-ssh-state.ts b/src/renderer/src/runtime/runtime-environment-ssh-state.ts index 59f1a3256ae..2fb48a121dc 100644 --- a/src/renderer/src/runtime/runtime-environment-ssh-state.ts +++ b/src/renderer/src/runtime/runtime-environment-ssh-state.ts @@ -9,6 +9,7 @@ import { sshTargetLabelsEqual } from '@/store/slices/ssh-target-cleanup' import { sanitizeSshTargetGeneration } from '../../../shared/ssh-target-generation' +import { forEachWithConcurrency } from '../../../shared/map-with-concurrency' /** * Mirrors a remote Orca server's own SSH targets into that environment's @@ -19,6 +20,9 @@ import { sanitizeSshTargetGeneration } from '../../../shared/ssh-target-generati */ const SSH_RPC_TIMEOUT_MS = 15_000 +// Why: these decorative reads share the host's 8-slot foreground RPC lane with sends and listings, +// so a small pool cuts N x 15 s stalls without letting them take the lane interactions need. +export const SSH_STATE_READ_CONCURRENCY = 3 function environmentTarget(environmentId: string): { kind: 'environment'; environmentId: string } { return { kind: 'environment', environmentId } @@ -86,7 +90,7 @@ async function fetchEnvironmentSshConnectionStates( targets: readonly SshTargetSummary[], generation: number ): Promise { - for (const target of targets) { + await forEachWithConcurrency(targets, SSH_STATE_READ_CONCURRENCY, async (target) => { if (generation !== getEnvironmentSshStateGeneration(environmentId)) { return } @@ -106,7 +110,7 @@ async function fetchEnvironmentSshConnectionStates( } catch { // Why: a timeout or unsupported RPC is not authoritative evidence that the HUB's SSH link disconnected. } - } + }) } type SshRefreshKind = 'metadata' | 'full'