mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 00:03:52 +00:00
perf(runtime): avoid forced SSH hydration on project refresh
This commit is contained in:
Neil
@@ -101,7 +101,8 @@ import { dispatchTerminalSideEffectBatch } from '@/components/terminal-pane/term
|
||||
import { subscribeToUnpairedDeviceAuthNotification } from './unpaired-device-auth-notification'
|
||||
import {
|
||||
applyRuntimeEnvironmentSshStateChanged,
|
||||
hydrateRuntimeEnvironmentSshState
|
||||
hydrateRuntimeEnvironmentSshState,
|
||||
refreshRuntimeEnvironmentSshTargetMetadata
|
||||
} from '@/runtime/runtime-environment-ssh-state'
|
||||
import {
|
||||
createRuntimeProjectRefreshScheduler,
|
||||
@@ -907,8 +908,8 @@ export function useIpcEvents(): void {
|
||||
|
||||
const runtimeProjectRefreshScheduler = createRuntimeProjectRefreshScheduler({
|
||||
refresh: async (environmentId) => {
|
||||
// Why: refresh the env's SSH bucket on (re)connect so a pre-drop snapshot can't keep a reconnect overlay stale.
|
||||
void hydrateRuntimeEnvironmentSshState(environmentId, { force: true }).catch(() => {})
|
||||
// Why: project events can reveal target CRUD, but known target states already arrive by push.
|
||||
void refreshRuntimeEnvironmentSshTargetMetadata(environmentId).catch(() => {})
|
||||
const repos = await useAppStore.getState().fetchRuntimeEnvironmentRepos(environmentId)
|
||||
await refreshRuntimeProjectWorktreesAndLineage(
|
||||
environmentId,
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
applyRuntimeEnvironmentSshStateChanged,
|
||||
connectRuntimeEnvironmentSshTarget,
|
||||
hydrateRuntimeEnvironmentSshState,
|
||||
refreshRuntimeEnvironmentSshTargetMetadata,
|
||||
resyncRuntimeEnvironmentSshTargets
|
||||
} from './runtime-environment-ssh-state'
|
||||
import { callRuntimeRpc } from './runtime-rpc-client'
|
||||
@@ -119,6 +120,195 @@ describe('hydrateRuntimeEnvironmentSshState', () => {
|
||||
expect(callRuntimeRpcMock.mock.calls.length).toBeGreaterThan(callsAfterFirst)
|
||||
})
|
||||
|
||||
it('does not rerun when an ordinary refresh joins a forced hydration', async () => {
|
||||
const envId = nextEnvId()
|
||||
useAppStore
|
||||
.getState()
|
||||
.setEnvironmentSshTargetsMetadata(envId, [{ id: 'ssh-1', label: 'devbox' }])
|
||||
let resolveTargets!: (value: { targets: { id: string; label: string }[] }) => void
|
||||
const targetsPromise = new Promise<{ targets: { id: string; label: string }[] }>((resolve) => {
|
||||
resolveTargets = resolve
|
||||
})
|
||||
callRuntimeRpcMock.mockImplementation((_target, method) => {
|
||||
if (method === 'ssh.listTargetSummaries') {
|
||||
return targetsPromise as never
|
||||
}
|
||||
if (method === 'ssh.listRemovedTargetLabels') {
|
||||
return Promise.resolve({ labels: {} } as never)
|
||||
}
|
||||
return Promise.resolve({ state: connState('ssh-1') } as never)
|
||||
})
|
||||
|
||||
const forced = hydrateRuntimeEnvironmentSshState(envId, { force: true })
|
||||
const ordinary = hydrateRuntimeEnvironmentSshState(envId)
|
||||
resolveTargets({ targets: [{ id: 'ssh-1', label: 'devbox' }] })
|
||||
await Promise.all([forced, ordinary])
|
||||
|
||||
expect(
|
||||
callRuntimeRpcMock.mock.calls.filter(([, method]) => method === 'ssh.listTargetSummaries')
|
||||
).toHaveLength(1)
|
||||
expect(
|
||||
callRuntimeRpcMock.mock.calls.filter(([, method]) => method === 'ssh.listRemovedTargetLabels')
|
||||
).toHaveLength(1)
|
||||
expect(
|
||||
callRuntimeRpcMock.mock.calls.filter(([, method]) => method === 'ssh.getState')
|
||||
).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('hydrates without force after the environment bucket is marked stale', async () => {
|
||||
const envId = nextEnvId()
|
||||
useAppStore
|
||||
.getState()
|
||||
.setEnvironmentSshTargetsMetadata(envId, [{ id: 'ssh-old', label: 'old box' }])
|
||||
useAppStore.getState().markEnvironmentSshStateStale(envId)
|
||||
installRpcResponses({
|
||||
targets: [{ id: 'ssh-new', label: 'new box' }],
|
||||
states: { 'ssh-new': connState('ssh-new') }
|
||||
})
|
||||
|
||||
await hydrateRuntimeEnvironmentSshState(envId)
|
||||
|
||||
const bucket = useAppStore.getState().sshStateByEnvironment.get(envId)
|
||||
expect(bucket?.targetsHydrated).toBe(true)
|
||||
expect(bucket?.targetLabels.get('ssh-new')).toBe('new box')
|
||||
expect(bucket?.connectionStates.get('ssh-new')?.status).toBe('connected')
|
||||
})
|
||||
|
||||
it('refreshes only target summaries when hydrated metadata is unchanged', async () => {
|
||||
const envId = nextEnvId()
|
||||
installRpcResponses({
|
||||
targets: [{ id: 'ssh-1', label: 'devbox' }],
|
||||
states: { 'ssh-1': connState('ssh-1') }
|
||||
})
|
||||
await hydrateRuntimeEnvironmentSshState(envId)
|
||||
callRuntimeRpcMock.mockClear()
|
||||
|
||||
await refreshRuntimeEnvironmentSshTargetMetadata(envId)
|
||||
|
||||
expect(callRuntimeRpcMock.mock.calls.map(([, method]) => method)).toEqual([
|
||||
'ssh.listTargetSummaries'
|
||||
])
|
||||
})
|
||||
|
||||
it('prunes removed targets and fetches state only for newly discovered targets', async () => {
|
||||
const envId = nextEnvId()
|
||||
useAppStore.getState().setEnvironmentSshTargetsMetadata(envId, [
|
||||
{ id: 'ssh-keep', label: 'keep' },
|
||||
{ id: 'ssh-old', label: 'old' }
|
||||
])
|
||||
useAppStore
|
||||
.getState()
|
||||
.setEnvironmentSshConnectionState(envId, 'ssh-keep', connState('ssh-keep'))
|
||||
useAppStore.getState().setEnvironmentSshConnectionState(envId, 'ssh-old', connState('ssh-old'))
|
||||
installRpcResponses({
|
||||
targets: [
|
||||
{ id: 'ssh-keep', label: 'keep' },
|
||||
{ id: 'ssh-new', label: 'new' }
|
||||
],
|
||||
labels: { 'ssh-old': 'old' },
|
||||
states: { 'ssh-new': connState('ssh-new', 'connecting') }
|
||||
})
|
||||
|
||||
await refreshRuntimeEnvironmentSshTargetMetadata(envId)
|
||||
|
||||
const bucket = useAppStore.getState().sshStateByEnvironment.get(envId)
|
||||
expect(bucket?.targetLabels.has('ssh-old')).toBe(false)
|
||||
expect(bucket?.connectionStates.has('ssh-old')).toBe(false)
|
||||
expect(bucket?.removedTargetLabels.get('ssh-old')).toBe('old')
|
||||
expect(bucket?.connectionStates.get('ssh-keep')?.status).toBe('connected')
|
||||
expect(bucket?.connectionStates.get('ssh-new')?.status).toBe('connecting')
|
||||
expect(callRuntimeRpcMock.mock.calls.filter(([, method]) => method === 'ssh.getState')).toEqual(
|
||||
[
|
||||
[
|
||||
{ kind: 'environment', environmentId: envId },
|
||||
'ssh.getState',
|
||||
{ targetId: 'ssh-new' },
|
||||
expect.anything()
|
||||
]
|
||||
]
|
||||
)
|
||||
})
|
||||
|
||||
it('reruns hydration when a new generation joins an older in-flight request', async () => {
|
||||
const envId = nextEnvId()
|
||||
let resolveOlderTargets!: (value: { targets: { id: string; label: string }[] }) => void
|
||||
const olderTargets = new Promise<{ targets: { id: string; label: string }[] }>((resolve) => {
|
||||
resolveOlderTargets = resolve
|
||||
})
|
||||
let targetListCallCount = 0
|
||||
callRuntimeRpcMock.mockImplementation((_target, method, params) => {
|
||||
if (method === 'ssh.listTargetSummaries') {
|
||||
targetListCallCount += 1
|
||||
return (
|
||||
targetListCallCount === 1
|
||||
? olderTargets
|
||||
: Promise.resolve({ targets: [{ id: 'ssh-new', label: 'new' }] })
|
||||
) as never
|
||||
}
|
||||
if (method === 'ssh.listRemovedTargetLabels') {
|
||||
return Promise.resolve({ labels: {} } as never)
|
||||
}
|
||||
const targetId = (params as { targetId: string }).targetId
|
||||
return Promise.resolve({ state: connState(targetId) } as never)
|
||||
})
|
||||
|
||||
const olderHydration = hydrateRuntimeEnvironmentSshState(envId)
|
||||
useAppStore.getState().markEnvironmentSshStateStale(envId)
|
||||
const newerHydration = hydrateRuntimeEnvironmentSshState(envId)
|
||||
resolveOlderTargets({ targets: [{ id: 'ssh-old', label: 'old' }] })
|
||||
await Promise.all([olderHydration, newerHydration])
|
||||
|
||||
const bucket = useAppStore.getState().sshStateByEnvironment.get(envId)
|
||||
expect(targetListCallCount).toBe(2)
|
||||
expect(bucket?.targetsHydrated).toBe(true)
|
||||
expect(bucket?.targetLabels.has('ssh-old')).toBe(false)
|
||||
expect(bucket?.targetLabels.get('ssh-new')).toBe('new')
|
||||
expect(bucket?.connectionStates.get('ssh-new')?.status).toBe('connected')
|
||||
})
|
||||
|
||||
it('keeps a failed full refresh full when a metadata refresh queued behind it', async () => {
|
||||
const envId = nextEnvId()
|
||||
useAppStore
|
||||
.getState()
|
||||
.setEnvironmentSshTargetsMetadata(envId, [{ id: 'ssh-1', label: 'devbox' }])
|
||||
useAppStore.getState().markEnvironmentSshStateStale(envId)
|
||||
let rejectFirstTargets!: (error: Error) => void
|
||||
const firstTargets = new Promise<never>((_resolve, reject) => {
|
||||
rejectFirstTargets = reject
|
||||
})
|
||||
let targetListCallCount = 0
|
||||
callRuntimeRpcMock.mockImplementation((_target, method) => {
|
||||
if (method === 'ssh.listTargetSummaries') {
|
||||
targetListCallCount += 1
|
||||
return (
|
||||
targetListCallCount === 1
|
||||
? firstTargets
|
||||
: Promise.resolve({ targets: [{ id: 'ssh-1', label: 'devbox' }] })
|
||||
) as never
|
||||
}
|
||||
if (method === 'ssh.listRemovedTargetLabels') {
|
||||
return Promise.resolve({ labels: {} } as never)
|
||||
}
|
||||
return Promise.resolve({ state: connState('ssh-1') } as never)
|
||||
})
|
||||
|
||||
const forced = hydrateRuntimeEnvironmentSshState(envId, { force: true })
|
||||
const metadata = refreshRuntimeEnvironmentSshTargetMetadata(envId)
|
||||
rejectFirstTargets(new Error('transient target-list failure'))
|
||||
await Promise.all([forced, metadata])
|
||||
|
||||
const bucket = useAppStore.getState().sshStateByEnvironment.get(envId)
|
||||
expect(targetListCallCount).toBe(2)
|
||||
expect(bucket?.targetsHydrated).toBe(true)
|
||||
expect(bucket?.connectionStates.get('ssh-1')?.status).toBe('connected')
|
||||
expect(callRuntimeRpcMock.mock.calls.map(([, method]) => method)).toEqual([
|
||||
'ssh.listTargetSummaries',
|
||||
'ssh.listTargetSummaries',
|
||||
'ssh.listRemovedTargetLabels',
|
||||
'ssh.getState'
|
||||
])
|
||||
})
|
||||
|
||||
it('leaves the bucket un-hydrated when the host lacks the ssh RPC methods', async () => {
|
||||
const envId = nextEnvId()
|
||||
installRpcResponses({ failListTargets: true })
|
||||
|
||||
@@ -3,6 +3,7 @@ import type { SshConnectionState, SshTargetSummary } from '../../../shared/ssh-t
|
||||
import { callRuntimeRpc } from './runtime-rpc-client'
|
||||
import { getEnvironmentSshStateGeneration } from '@/store/slices/runtime-environment-ssh'
|
||||
import { admitSshConnectionState } from '../../../shared/ssh-retained-payload-admission'
|
||||
import { sshTargetLabelsEqual } from '@/store/slices/ssh-target-cleanup'
|
||||
|
||||
/**
|
||||
* Mirrors a remote Orca server's own SSH targets into that environment's
|
||||
@@ -44,7 +45,18 @@ async function syncEnvironmentSshTargetMetadata(
|
||||
generation: number
|
||||
): Promise<SshTargetSummary[]> {
|
||||
const targets = await fetchEnvironmentSshTargets(environmentId)
|
||||
if (generation !== getEnvironmentSshStateGeneration(environmentId)) {
|
||||
return []
|
||||
}
|
||||
useAppStore.getState().setEnvironmentSshTargetsMetadata(environmentId, targets, generation)
|
||||
await syncEnvironmentRemovedSshTargetLabels(environmentId, generation)
|
||||
return generation === getEnvironmentSshStateGeneration(environmentId) ? targets : []
|
||||
}
|
||||
|
||||
async function syncEnvironmentRemovedSshTargetLabels(
|
||||
environmentId: string,
|
||||
generation: number
|
||||
): Promise<void> {
|
||||
try {
|
||||
const { labels } = await callRuntimeRpc<{ labels: Record<string, string> }>(
|
||||
environmentTarget(environmentId),
|
||||
@@ -56,7 +68,6 @@ async function syncEnvironmentSshTargetMetadata(
|
||||
} catch {
|
||||
// Best-effort — a missing map just falls back to the raw target id.
|
||||
}
|
||||
return targets
|
||||
}
|
||||
|
||||
async function fetchEnvironmentSshConnectionStates(
|
||||
@@ -65,6 +76,9 @@ async function fetchEnvironmentSshConnectionStates(
|
||||
generation: number
|
||||
): Promise<void> {
|
||||
for (const target of targets) {
|
||||
if (generation !== getEnvironmentSshStateGeneration(environmentId)) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
const { state } = await callRuntimeRpc<{ state: SshConnectionState | null }>(
|
||||
environmentTarget(environmentId),
|
||||
@@ -84,8 +98,21 @@ async function fetchEnvironmentSshConnectionStates(
|
||||
}
|
||||
}
|
||||
|
||||
type HydrationEntry = { promise: Promise<void>; rerunRequested: boolean }
|
||||
const hydrationsInFlight = new Map<string, HydrationEntry>()
|
||||
type SshRefreshKind = 'metadata' | 'full'
|
||||
type SshRefreshEntry = {
|
||||
promise: Promise<void>
|
||||
generation: number
|
||||
kind: SshRefreshKind
|
||||
rerunKind: SshRefreshKind | null
|
||||
}
|
||||
const sshRefreshesInFlight = new Map<string, SshRefreshEntry>()
|
||||
|
||||
function mergeSshRefreshKind(
|
||||
current: SshRefreshKind | null,
|
||||
requested: SshRefreshKind
|
||||
): SshRefreshKind {
|
||||
return current === 'full' || requested === 'full' ? 'full' : 'metadata'
|
||||
}
|
||||
|
||||
async function runEnvironmentSshHydration(environmentId: string): Promise<void> {
|
||||
const generation = getEnvironmentSshStateGeneration(environmentId)
|
||||
@@ -93,6 +120,76 @@ async function runEnvironmentSshHydration(environmentId: string): Promise<void>
|
||||
await fetchEnvironmentSshConnectionStates(environmentId, targets, generation)
|
||||
}
|
||||
|
||||
async function runEnvironmentSshTargetMetadataRefresh(environmentId: string): Promise<void> {
|
||||
const generation = getEnvironmentSshStateGeneration(environmentId)
|
||||
const bucket = useAppStore.getState().sshStateByEnvironment.get(environmentId)
|
||||
const targets = await fetchEnvironmentSshTargets(environmentId)
|
||||
if (generation !== getEnvironmentSshStateGeneration(environmentId)) {
|
||||
return
|
||||
}
|
||||
const metadataChanged =
|
||||
!bucket?.targetsHydrated || !sshTargetLabelsEqual(bucket.targetLabels, targets)
|
||||
useAppStore.getState().setEnvironmentSshTargetsMetadata(environmentId, targets, generation)
|
||||
if (!metadataChanged) {
|
||||
return
|
||||
}
|
||||
await syncEnvironmentRemovedSshTargetLabels(environmentId, generation)
|
||||
const priorTargetIds = new Set(bucket?.targetLabels.keys() ?? [])
|
||||
await fetchEnvironmentSshConnectionStates(
|
||||
environmentId,
|
||||
targets.filter((target) => !priorTargetIds.has(target.id)),
|
||||
generation
|
||||
)
|
||||
}
|
||||
|
||||
function requestSshRefreshRerun(entry: SshRefreshEntry, kind: SshRefreshKind): void {
|
||||
entry.rerunKind = mergeSshRefreshKind(entry.rerunKind, kind)
|
||||
}
|
||||
|
||||
function startEnvironmentSshRefresh(
|
||||
environmentId: string,
|
||||
initialKind: SshRefreshKind
|
||||
): Promise<void> {
|
||||
const entry: SshRefreshEntry = {
|
||||
promise: Promise.resolve(),
|
||||
generation: getEnvironmentSshStateGeneration(environmentId),
|
||||
kind: initialKind,
|
||||
rerunKind: null
|
||||
}
|
||||
entry.promise = (async () => {
|
||||
let nextKind: SshRefreshKind | null = initialKind
|
||||
let lastError: unknown = null
|
||||
try {
|
||||
while (nextKind) {
|
||||
entry.kind = nextKind
|
||||
entry.generation = getEnvironmentSshStateGeneration(environmentId)
|
||||
entry.rerunKind = null
|
||||
try {
|
||||
await (entry.kind === 'full'
|
||||
? runEnvironmentSshHydration(environmentId)
|
||||
: runEnvironmentSshTargetMetadataRefresh(environmentId))
|
||||
lastError = null
|
||||
} catch (error) {
|
||||
lastError = error
|
||||
if (entry.rerunKind) {
|
||||
entry.rerunKind = mergeSshRefreshKind(entry.rerunKind, entry.kind)
|
||||
}
|
||||
}
|
||||
nextKind = entry.rerunKind
|
||||
}
|
||||
if (lastError) {
|
||||
throw lastError
|
||||
}
|
||||
} finally {
|
||||
if (sshRefreshesInFlight.get(environmentId) === entry) {
|
||||
sshRefreshesInFlight.delete(environmentId)
|
||||
}
|
||||
}
|
||||
})()
|
||||
sshRefreshesInFlight.set(environmentId, entry)
|
||||
return entry.promise
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches the environment's SSH targets, removal tombstones, and per-target
|
||||
* connection states into its bucket. Single-flight per environment; a `force`
|
||||
@@ -107,10 +204,11 @@ export async function hydrateRuntimeEnvironmentSshState(
|
||||
environmentId: string,
|
||||
options: { force?: boolean } = {}
|
||||
): Promise<void> {
|
||||
const inFlight = hydrationsInFlight.get(environmentId)
|
||||
const generation = getEnvironmentSshStateGeneration(environmentId)
|
||||
const inFlight = sshRefreshesInFlight.get(environmentId)
|
||||
if (inFlight) {
|
||||
if (options.force) {
|
||||
inFlight.rerunRequested = true
|
||||
if (options.force || inFlight.generation !== generation) {
|
||||
requestSshRefreshRerun(inFlight, 'full')
|
||||
}
|
||||
return inFlight.promise
|
||||
}
|
||||
@@ -118,28 +216,21 @@ export async function hydrateRuntimeEnvironmentSshState(
|
||||
if (!options.force && bucket?.targetsHydrated) {
|
||||
return
|
||||
}
|
||||
const entry: HydrationEntry = { promise: Promise.resolve(), rerunRequested: false }
|
||||
entry.promise = (async () => {
|
||||
let lastError: unknown = null
|
||||
try {
|
||||
do {
|
||||
entry.rerunRequested = false
|
||||
try {
|
||||
await runEnvironmentSshHydration(environmentId)
|
||||
lastError = null
|
||||
} catch (error) {
|
||||
lastError = error
|
||||
}
|
||||
} while (entry.rerunRequested)
|
||||
if (lastError) {
|
||||
throw lastError
|
||||
}
|
||||
} finally {
|
||||
hydrationsInFlight.delete(environmentId)
|
||||
}
|
||||
})()
|
||||
hydrationsInFlight.set(environmentId, entry)
|
||||
return entry.promise
|
||||
return startEnvironmentSshRefresh(environmentId, 'full')
|
||||
}
|
||||
|
||||
/** Refreshes target membership without re-reading every known target state. */
|
||||
export async function refreshRuntimeEnvironmentSshTargetMetadata(
|
||||
environmentId: string
|
||||
): Promise<void> {
|
||||
const generation = getEnvironmentSshStateGeneration(environmentId)
|
||||
const inFlight = sshRefreshesInFlight.get(environmentId)
|
||||
if (inFlight) {
|
||||
requestSshRefreshRerun(inFlight, inFlight.generation === generation ? 'metadata' : 'full')
|
||||
return inFlight.promise
|
||||
}
|
||||
const bucket = useAppStore.getState().sshStateByEnvironment.get(environmentId)
|
||||
return startEnvironmentSshRefresh(environmentId, bucket?.targetsHydrated ? 'metadata' : 'full')
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user