mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
fix(ssh): no probe stall on the first remote action after a pause (#29240)
* fix(ssh): skip the managed-server probe while the control socket is answering A remote call after more than 5 s idle first ran a status.get probe on a new one-shot socket (SSH channel, WebSocket upgrade, hello, auth, request). A ready shared control socket that heard from the server in the last 20 s already proves it serves; a kill, reboot, rebind or stop closes that socket, so the probe still runs then. * fix(ssh): read control-socket answers through a leaf module orcad-managed-tunnel imported the request-connection cache, which (through the status owner) imports the serving gate back. Tests that mock that cache with an async factory deadlocked importing it. A leaf module now carries the answer source between them.
This commit is contained in:
@@ -21,6 +21,7 @@ import {
|
||||
sendRemoteRuntimeSharedControlRequest,
|
||||
subscribeRemoteRuntimeSharedControlRequest
|
||||
} from './runtime-environment-request-connections'
|
||||
import { getRemoteRuntimeSharedControlAnsweredAt } from './runtime-environment-shared-control-answers'
|
||||
|
||||
const ENVIRONMENT_ID = 'standing-intent-test'
|
||||
|
||||
@@ -65,6 +66,25 @@ describe('runtime environment shared-control connection cache', () => {
|
||||
await waitFor(() => server.connectionCount() === 2)
|
||||
})
|
||||
|
||||
it('reports when the ready control socket last heard from its server, never once it drops', async () => {
|
||||
const server = await createSharedControlTestServer()
|
||||
expect(getRemoteRuntimeSharedControlAnsweredAt(ENVIRONMENT_ID)).toBeNull()
|
||||
|
||||
const before = Date.now()
|
||||
await sendRemoteRuntimeSharedControlRequest(
|
||||
ENVIRONMENT_ID,
|
||||
server.pairing,
|
||||
'status.get',
|
||||
undefined,
|
||||
1000
|
||||
)
|
||||
expect(getRemoteRuntimeSharedControlAnsweredAt(ENVIRONMENT_ID)).toBeGreaterThanOrEqual(before)
|
||||
|
||||
server.closeClients()
|
||||
await waitFor(() => getRemoteRuntimeSharedControlDiagnostics(ENVIRONMENT_ID)?.state !== 'ready')
|
||||
expect(getRemoteRuntimeSharedControlAnsweredAt(ENVIRONMENT_ID)).toBeNull()
|
||||
})
|
||||
|
||||
it('blocks standing creation while manual intent is disconnected', async () => {
|
||||
const server = await createSharedControlTestServer()
|
||||
markRuntimeEnvironmentManuallyDisconnected(ENVIRONMENT_ID)
|
||||
|
||||
@@ -18,6 +18,10 @@ import type {
|
||||
import { isRuntimeEnvironmentCapabilityPaused } from './runtime-environment-capability-evidence'
|
||||
import { isRuntimeEnvironmentManuallyDisconnected } from './runtime-environment-manual-disconnect'
|
||||
import { publishRuntimeEnvironmentDiagnostics } from './runtime-environment-diagnostics-broadcast'
|
||||
import {
|
||||
clearSharedControlAnswerSource,
|
||||
setSharedControlAnswerSource
|
||||
} from './runtime-environment-shared-control-answers'
|
||||
import {
|
||||
advanceRuntimeEnvironmentTransportGeneration,
|
||||
getRuntimeEnvironmentTransportGeneration
|
||||
@@ -31,6 +35,7 @@ type CachedRuntimeConnection = {
|
||||
type CachedSharedControlConnection = {
|
||||
pairingKey: string
|
||||
connection: RemoteRuntimeSharedControlConnection
|
||||
answeredAt: () => number | null
|
||||
}
|
||||
|
||||
const requestConnections = new Map<string, CachedRuntimeConnection>()
|
||||
@@ -164,6 +169,7 @@ export function subscribeRemoteRuntimeSharedControlRequest<TResult>(
|
||||
export function closeRemoteRuntimeSharedControlConnection(environmentId: string): void {
|
||||
const cached = sharedControlConnections.get(environmentId)
|
||||
sharedControlConnections.delete(environmentId)
|
||||
clearSharedControlAnswerSource(environmentId)
|
||||
cached?.connection.close()
|
||||
}
|
||||
|
||||
@@ -210,13 +216,19 @@ function getSharedControlConnection(
|
||||
advanceRuntimeEnvironmentTransportGeneration(environmentId)
|
||||
cached?.connection.close()
|
||||
const transportGeneration = getRuntimeEnvironmentTransportGeneration(environmentId)
|
||||
let answeredAt: number | null = null
|
||||
cached = {
|
||||
pairingKey,
|
||||
answeredAt: () => answeredAt,
|
||||
connection: new RemoteRuntimeSharedControlConnection(pairedConnection, {
|
||||
environmentId,
|
||||
clientCapabilities: ELECTRON_REMOTE_RUNTIME_CLIENT_CAPABILITIES,
|
||||
isManuallyDisconnected: () => isRuntimeEnvironmentManuallyDisconnected(environmentId),
|
||||
isCapabilityPaused: () => isRuntimeEnvironmentCapabilityPaused(environmentId),
|
||||
// Why wall clock: it advances across a suspend, so a pre-sleep answer reads as stale.
|
||||
onAnswered: () => {
|
||||
answeredAt = Date.now()
|
||||
},
|
||||
onDiagnosticsChanged: (diagnostics) => {
|
||||
if (getRuntimeEnvironmentTransportGeneration(environmentId) !== transportGeneration) {
|
||||
return
|
||||
@@ -240,6 +252,11 @@ function getSharedControlConnection(
|
||||
})
|
||||
}
|
||||
sharedControlConnections.set(environmentId, cached)
|
||||
const entry = cached
|
||||
// Why ready is enough: a socket reaches ready only on its own server frame, which stamps it.
|
||||
setSharedControlAnswerSource(environmentId, () =>
|
||||
entry.connection.getDiagnostics().state === 'ready' ? entry.answeredAt() : null
|
||||
)
|
||||
}
|
||||
return cached.connection
|
||||
}
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
/**
|
||||
* When each environment's ready control socket last heard from its server. A leaf module: the
|
||||
* serving gate reads it without importing the connection cache, which imports the gate back.
|
||||
*/
|
||||
const sources = new Map<string, () => number | null>()
|
||||
|
||||
export function setSharedControlAnswerSource(
|
||||
environmentId: string,
|
||||
source: () => number | null
|
||||
): void {
|
||||
sources.set(environmentId, source)
|
||||
}
|
||||
|
||||
export function clearSharedControlAnswerSource(environmentId: string): void {
|
||||
sources.delete(environmentId)
|
||||
}
|
||||
|
||||
/** Null while there is no ready control socket for the environment. */
|
||||
export function getRemoteRuntimeSharedControlAnsweredAt(environmentId: string): number | null {
|
||||
return sources.get(environmentId)?.() ?? null
|
||||
}
|
||||
@@ -13,6 +13,7 @@ vi.mock('./orcad-managed-wake', () => ({ wakeStoppedManagedOrcad: vi.fn() }))
|
||||
import { wakeStoppedManagedOrcad } from './orcad-managed-wake'
|
||||
import {
|
||||
ensureManagedOrcadServing,
|
||||
MANAGED_ORCAD_NOT_ACTIVATED_DETAIL,
|
||||
resetManagedOrcadServingForTests,
|
||||
setManagedOrcadStartListener,
|
||||
type OrcadManagedServingInput
|
||||
@@ -174,4 +175,58 @@ describe('ensureManagedOrcadServing', () => {
|
||||
await ensureManagedOrcadServing({ ...input(probe), remotePort: 40_001 }, () => 2)
|
||||
expect(probe).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('skips the probe after an idle pause while the control socket keeps answering', async () => {
|
||||
let now = 0
|
||||
const probe = vi.fn(async () => true)
|
||||
const answeredAt = vi.fn((): number | null => null)
|
||||
const live = { ...input(probe), answeredAt }
|
||||
await ensureManagedOrcadServing(live, () => now)
|
||||
expect(probe).toHaveBeenCalledOnce()
|
||||
|
||||
now = 6_000
|
||||
answeredAt.mockReturnValue(4_000)
|
||||
expect(await ensureManagedOrcadServing(live, () => now)).toEqual({ state: 'serving' })
|
||||
expect(answeredAt).toHaveBeenCalledWith('env-1')
|
||||
expect(probe).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('shares one probe between concurrent calls once the last answer is stale', async () => {
|
||||
let answer!: (value: boolean) => void
|
||||
const probe = vi.fn(() => new Promise<boolean>((resolve) => (answer = resolve)))
|
||||
// Last heard before a suspend: too old to prove anything.
|
||||
const stale = { ...input(probe), answeredAt: () => 0 }
|
||||
const calls = [1, 2, 3].map(() => ensureManagedOrcadServing(stale, () => 60_000))
|
||||
answer(true)
|
||||
expect(await Promise.all(calls)).toEqual([
|
||||
{ state: 'serving' },
|
||||
{ state: 'serving' },
|
||||
{ state: 'serving' }
|
||||
])
|
||||
expect(probe).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('still detects a stopped server once its control socket closed', async () => {
|
||||
vi.mocked(wakeStoppedManagedOrcad).mockResolvedValue({ outcome: 'not-activated' })
|
||||
const answeredAt = vi.fn((): number | null => 0)
|
||||
const probe = vi.fn(async () => true)
|
||||
const watched = { ...input(probe), answeredAt }
|
||||
expect(await ensureManagedOrcadServing(watched, () => 0)).toEqual({ state: 'serving' })
|
||||
expect(probe).not.toHaveBeenCalled()
|
||||
|
||||
// Another desktop stopped the server: its socket closed, so there is no live answer.
|
||||
answeredAt.mockReturnValue(null)
|
||||
probe.mockResolvedValue(false)
|
||||
expect(await ensureManagedOrcadServing(watched, () => 1_000)).toEqual({
|
||||
state: 'unverifiable',
|
||||
detail: MANAGED_ORCAD_NOT_ACTIVATED_DETAIL
|
||||
})
|
||||
expect(probe).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('does not take an answer from a wall clock that moved backwards as proof', async () => {
|
||||
const probe = vi.fn(async () => true)
|
||||
await ensureManagedOrcadServing({ ...input(probe), answeredAt: () => 50_000 }, () => 10_000)
|
||||
expect(probe).toHaveBeenCalledOnce()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -27,6 +27,8 @@ export type OrcadManagedServingInput = {
|
||||
connection: OrcadHostTransport
|
||||
remotePort: number
|
||||
probe: (environment: KnownRuntimeEnvironment, timeoutMs: number) => Promise<boolean>
|
||||
/** When this environment's ready control socket last heard from its server, if it has. */
|
||||
answeredAt?: (environmentId: string) => number | null
|
||||
}
|
||||
|
||||
/** A wake refused because another update, rollback or recovery holds the host; it clears itself. */
|
||||
@@ -41,6 +43,9 @@ const PROBE_TIMEOUT_MS = 5_000
|
||||
// Why: a connect checks right after its fresh tunnel did; one verdict serves both, on that
|
||||
// transport and port only, so a kill, reboot or rebind is never answered from cache.
|
||||
const VERDICT_REUSE_MS = 5_000
|
||||
// Why: the control socket hears a pong every 10 s while healthy, and a kill, reboot, rebind or
|
||||
// stop closes it, so a recent answer on it proves serving without a fresh probe socket.
|
||||
const LIVE_ANSWER_PROOF_MS = 20_000
|
||||
|
||||
type ServingTransport = { connection: OrcadHostTransport; generation: number; remotePort: number }
|
||||
// Why per transport: a check on a dropped connection fails, and a caller on the reconnected one
|
||||
@@ -85,6 +90,12 @@ export function ensureManagedOrcadServing(
|
||||
) {
|
||||
return entry.check
|
||||
}
|
||||
const answeredAt = input.answeredAt?.(id) ?? null
|
||||
const answerAge = answeredAt === null ? null : now() - answeredAt
|
||||
// Why the lower bound: a wall clock set backwards must not stretch the proof.
|
||||
if (answerAge !== null && answerAge >= 0 && answerAge < LIVE_ANSWER_PROOF_MS) {
|
||||
return Promise.resolve({ state: 'serving' })
|
||||
}
|
||||
const next: ServingCheck = { ...transport, check: checkAndStart(input) }
|
||||
checks.set(id, next)
|
||||
// A rejected check is never reused.
|
||||
|
||||
@@ -5,6 +5,7 @@ import type { SshConnection } from './ssh-connection'
|
||||
import { probeManagedOrcadTunnel } from './orcad-managed-tunnel-resume'
|
||||
import { ensureManagedOrcadServing } from './orcad-managed-serving'
|
||||
import { OrcadManagedTunnelManager } from './orcad-managed-tunnel-manager'
|
||||
import { getRemoteRuntimeSharedControlAnsweredAt } from '../ipc/runtime-environment-shared-control-answers'
|
||||
import { getSshConnectionManager, getSshTargetRegistryStore } from './ssh-target-registry'
|
||||
import {
|
||||
MANAGED_ORCAD_TUNNEL_TARGETING,
|
||||
@@ -17,7 +18,12 @@ const managedTunnels = new OrcadManagedTunnelManager({
|
||||
getConnectionManager: getSshConnectionManager,
|
||||
getTargetStore: getSshTargetRegistryStore,
|
||||
targeting: MANAGED_ORCAD_TUNNEL_TARGETING,
|
||||
ensureServing: (input) => ensureManagedOrcadServing({ ...input, probe: probeManagedOrcadTunnel })
|
||||
ensureServing: (input) =>
|
||||
ensureManagedOrcadServing({
|
||||
...input,
|
||||
probe: probeManagedOrcadTunnel,
|
||||
answeredAt: getRemoteRuntimeSharedControlAnsweredAt
|
||||
})
|
||||
})
|
||||
|
||||
export async function ensureOrcadManagedTunnel(
|
||||
|
||||
@@ -186,6 +186,7 @@ export class RemoteRuntimeSharedControlConnection {
|
||||
},
|
||||
onError: (error) => this.handleSocketClosed(error, socketGeneration),
|
||||
onTextFrame: (frame) => this.handleTextFrame(frame, socketGeneration),
|
||||
onActivity: this.options.onAnswered,
|
||||
liveness: {
|
||||
options: this.options.liveness,
|
||||
onDead: (error) => this.handleSocketClosed(error, socketGeneration)
|
||||
|
||||
@@ -19,6 +19,8 @@ export function openSharedControlSocket(
|
||||
onClose: (close: { code: number; reason: string }, error: RemoteRuntimeClientError) => void
|
||||
onError: (error: RemoteRuntimeClientError) => void
|
||||
onTextFrame: (frame: string) => void
|
||||
/** Any inbound frame, ping or pong on the current socket. */
|
||||
onActivity?: () => void
|
||||
// Why: reconnect is edge-triggered on `close`, but a half-open tunnel
|
||||
// never delivers one. Liveness pings and declares the socket dead when
|
||||
// the server goes silent, so the close/reconnect path can run (#7718).
|
||||
@@ -28,7 +30,11 @@ export function openSharedControlSocket(
|
||||
}
|
||||
}
|
||||
): { ok: true; socket: RemoteRuntimeWebSocket } | { ok: false; error: RemoteRuntimeClientError } {
|
||||
let noteActivity: () => void = () => {}
|
||||
let noteLiveness: () => void = () => {}
|
||||
const noteActivity = (): void => {
|
||||
noteLiveness()
|
||||
callbacks.onActivity?.()
|
||||
}
|
||||
const opened = openRemoteRuntimeWebSocket(connection, {
|
||||
onClose: (ws, code, reason) => {
|
||||
if (callbacks.getCurrentSocket() === ws) {
|
||||
@@ -90,7 +96,7 @@ export function openSharedControlSocket(
|
||||
},
|
||||
options: liveness.options
|
||||
})
|
||||
noteActivity = monitor.noteActivity
|
||||
noteLiveness = monitor.noteActivity
|
||||
|
||||
return {
|
||||
ok: true,
|
||||
|
||||
@@ -79,6 +79,8 @@ export type RemoteRuntimeSharedControlConnectionOptions = {
|
||||
isCapabilityPaused?: () => boolean
|
||||
/** Publishes local transport diagnostics after a meaningful state transition. */
|
||||
onDiagnosticsChanged?: (diagnostics: RemoteRuntimeSharedConnectionDiagnostics) => void
|
||||
/** Any inbound frame, ping or pong on the current socket. */
|
||||
onAnswered?: () => void
|
||||
reconnectStableResetMs?: number
|
||||
liveness?: RemoteRuntimeSocketLivenessOptions
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user