mirror of
https://github.com/stablyai/orca.git
synced 2026-11-15 18:37:24 +00:00
fix(cloud): review fixes for fast wake (stale background re-pair guarded by resume generation, no reconnect over a socket already dialing, pane recovery scoped to the host again, ready poll backs off)
This commit is contained in:
@@ -228,6 +228,43 @@ describe('resuming a recipe VM server', () => {
|
||||
expect(invalidateTransport).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('does not let a stale background re-pair overwrite a newer resume', async () => {
|
||||
const environmentId = provision()
|
||||
rmSync(serverDataPath, { recursive: true, force: true })
|
||||
serverDataPath = mkdtempSync(join(tmpdir(), 'orca-resume-server-'))
|
||||
let releaseFirst!: () => void
|
||||
const firstGate = new Promise<void>((resolve) => (releaseFirst = resolve))
|
||||
const auth = serverAuth(serverDataPath)
|
||||
const repairedA = vi.fn()
|
||||
// Resume A: its probe is held, so its re-pair would land late.
|
||||
await applyResumedRuntimePairing({
|
||||
userDataPath: desktopDataPath,
|
||||
environmentId,
|
||||
pairingCode: bootVm(serverDataPath, 'wss://vm-a.example/runtime', unpaddedHostKey),
|
||||
invalidateTransport: vi.fn(),
|
||||
probeSavedPairing: async (connection) => {
|
||||
await firstGate
|
||||
return auth(connection)
|
||||
},
|
||||
confirmInBackground: true,
|
||||
onRepairedInBackground: repairedA
|
||||
})
|
||||
// Resume B lands first and re-pairs synchronously (probe answers at once).
|
||||
const outcomeB = await applyResumedRuntimePairing({
|
||||
userDataPath: desktopDataPath,
|
||||
environmentId,
|
||||
pairingCode: bootVm(serverDataPath, 'wss://vm-b.example/runtime', unpaddedHostKey),
|
||||
invalidateTransport: vi.fn(),
|
||||
probeSavedPairing: auth
|
||||
})
|
||||
const afterB = resolveEnvironmentPairingOffer(desktopDataPath, environmentId)
|
||||
releaseFirst()
|
||||
await new Promise((resolve) => setTimeout(resolve, 20))
|
||||
expect(repairedA).not.toHaveBeenCalled()
|
||||
expect(resolveEnvironmentPairingOffer(desktopDataPath, environmentId)).toEqual(afterB)
|
||||
expect(outcomeB.environment.id).toBe(environmentId)
|
||||
})
|
||||
|
||||
it('keeps the saved pairing when the server cannot be reached to check it', async () => {
|
||||
const environmentId = provision()
|
||||
const original = resolveEnvironmentPairingOffer(desktopDataPath, environmentId)
|
||||
|
||||
@@ -10,6 +10,11 @@ import { ELECTRON_REMOTE_RUNTIME_CLIENT_CAPABILITIES } from '../../shared/electr
|
||||
|
||||
const SAVED_PAIRING_PROBE_TIMEOUT_MS = 10_000
|
||||
|
||||
// The newest resume per environment. A background confirmation may only re-pair for the resume that
|
||||
// started it: a later resume (Restart right after a wake) owns the environment's pairing now.
|
||||
const latestResume = new Map<string, number>()
|
||||
let resumeSequence = 0
|
||||
|
||||
export type ResumedRuntimePairingOutcome = {
|
||||
environment: KnownRuntimeEnvironment
|
||||
reusedPairing: boolean
|
||||
@@ -36,6 +41,8 @@ export async function applyResumedRuntimePairing(args: {
|
||||
onRepairedInBackground?: (environment: KnownRuntimeEnvironment) => void
|
||||
now?: number
|
||||
}): Promise<ResumedRuntimePairingOutcome> {
|
||||
const resume = ++resumeSequence
|
||||
latestResume.set(args.environmentId, resume)
|
||||
const { repairFromFreshOffer, ...refreshed } = refreshEnvironmentFromResumePairingCode(
|
||||
args.userDataPath,
|
||||
args.environmentId,
|
||||
@@ -50,16 +57,20 @@ export async function applyResumedRuntimePairing(args: {
|
||||
// while the probe confirms it instead of after (one connection setup off the wake path). In the
|
||||
// rare case the server lost the device, the re-pair below drops that socket and dials again.
|
||||
void args.invalidateTransport(args.environmentId)
|
||||
const confirmed = confirmSavedPairing(
|
||||
refreshed,
|
||||
repairFromFreshOffer,
|
||||
args.probeSavedPairing
|
||||
).then((outcome) => {
|
||||
if (!outcome.reusedPairing) {
|
||||
const isCurrent = (): boolean => latestResume.get(args.environmentId) === resume
|
||||
// A stale confirmation must not overwrite the pairing a newer resume wrote.
|
||||
const guardedRepair = (): KnownRuntimeEnvironment =>
|
||||
isCurrent() ? repairFromFreshOffer() : refreshed.environment
|
||||
const confirmed = confirmSavedPairing(refreshed, guardedRepair, args.probeSavedPairing).then(
|
||||
(outcome) => {
|
||||
if (outcome.environment === refreshed.environment) {
|
||||
// Kept, or superseded by a newer resume before the re-pair: nothing changed here.
|
||||
return { ...outcome, reusedPairing: true }
|
||||
}
|
||||
void args.invalidateTransport(args.environmentId)
|
||||
return outcome
|
||||
}
|
||||
return outcome
|
||||
})
|
||||
)
|
||||
if (!args.confirmInBackground) {
|
||||
return confirmed
|
||||
}
|
||||
|
||||
@@ -29,6 +29,7 @@ import {
|
||||
import type { VmWakeRequest } from '../runtime/workspaces/sleep/workspace-sleep-platform'
|
||||
import { getRuntimeRecipeContext } from '../ipc/ephemeral-vm-recipe-context'
|
||||
import { invalidateRuntimeEnvironmentTransport } from '../ipc/runtime-environments/runtime-environments'
|
||||
import { reconnectSharedControlIfIdle } from '../ipc/runtime-environments/runtime-environment-shared-control-reconnect'
|
||||
import {
|
||||
getRemoteRuntimeSharedControlDiagnostics,
|
||||
ensureRemoteRuntimeSharedControlConnection,
|
||||
@@ -228,7 +229,10 @@ function reestablishWokenServer(
|
||||
getPreferredPairedRuntimeConnection(resolveEnvironment(userDataPath, environmentId))
|
||||
)
|
||||
}
|
||||
reconnectRemoteRuntimeSharedControlConnection(environmentId)
|
||||
reconnectSharedControlIfIdle({
|
||||
getDiagnostics: () => getRemoteRuntimeSharedControlDiagnostics(environmentId),
|
||||
reconnectNow: () => reconnectRemoteRuntimeSharedControlConnection(environmentId)
|
||||
})
|
||||
void getRuntimeEnvironmentStatusOwner(userDataPath, environmentId)
|
||||
.refresh({ reconnect: true })
|
||||
.catch(() => {})
|
||||
|
||||
@@ -214,6 +214,7 @@ export async function announceWakeUsable(
|
||||
const runtime = findCloudRuntime(userDataPath, workspaceId)?.runtime
|
||||
return runtime?.status === 'running' ? runtime : null
|
||||
}
|
||||
let beat = 0
|
||||
while (Date.now() < deadline && hooks.isCurrent()) {
|
||||
const runtime = running()
|
||||
if (runtime) {
|
||||
@@ -240,8 +241,10 @@ export async function announceWakeUsable(
|
||||
}
|
||||
return
|
||||
}
|
||||
// Short beat: this sits on the wake's critical path until the runtime record flips to running.
|
||||
await new Promise((resolve) => setTimeout(resolve, 50))
|
||||
// Short first beats: this sits on the wake's critical path until the runtime record flips to
|
||||
// running, but each check reads the runtime store, so later beats back off.
|
||||
beat += 1
|
||||
await new Promise((resolve) => setTimeout(resolve, beat <= 4 ? 50 : beat <= 8 ? 100 : 250))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -31,6 +31,7 @@ import {
|
||||
getRuntimeEnvironmentTransportGeneration
|
||||
} from './runtime-environment-transport-generation'
|
||||
import { noteRuntimeEnvironmentSharedControlContact } from './runtime-environment-contact-loss'
|
||||
import { reconnectSharedControlIfIdle } from './runtime-environment-shared-control-reconnect'
|
||||
import { declareWslHostClient } from '../../wsl/host-lifecycle/wsl-host-client-declaration'
|
||||
|
||||
type CachedRuntimeConnection = {
|
||||
@@ -85,7 +86,7 @@ export function getRuntimeEnvironmentStatusOwner(
|
||||
),
|
||||
establish: () => {
|
||||
ensureRemoteRuntimeSharedControlConnection(environment.id, pairedConnection)
|
||||
reconnectRemoteRuntimeSharedControlConnection(environment.id)
|
||||
reconnectSharedControlIfIdle(sharedControlConnections.get(environment.id)?.connection)
|
||||
},
|
||||
pause: () => pauseRemoteRuntimeSharedControlRetry(environment.id)
|
||||
})
|
||||
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { reconnectSharedControlIfIdle } from './runtime-environment-shared-control-reconnect'
|
||||
|
||||
function connection(state: string | null) {
|
||||
return {
|
||||
getDiagnostics: () => (state === null ? null : ({ state } as never)),
|
||||
reconnectNow: vi.fn()
|
||||
}
|
||||
}
|
||||
|
||||
describe('reconnectSharedControlIfIdle', () => {
|
||||
it.each(['awaiting_ready', 'awaiting_authenticated', 'ready'])(
|
||||
'leaves a socket that is %s alone',
|
||||
(state) => {
|
||||
const socket = connection(state)
|
||||
reconnectSharedControlIfIdle(socket)
|
||||
expect(socket.reconnectNow).not.toHaveBeenCalled()
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['closed', 'reconnecting', null])('reconnects when the socket is %s', (state) => {
|
||||
const socket = connection(state)
|
||||
reconnectSharedControlIfIdle(socket)
|
||||
expect(socket.reconnectNow).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('does nothing without a connection', () => {
|
||||
expect(() => reconnectSharedControlIfIdle(undefined)).not.toThrow()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,21 @@
|
||||
import type { RemoteRuntimeSharedConnectionDiagnostics } from '../../../shared/remote-runtime/control/remote-runtime-shared-control-types'
|
||||
|
||||
type ReconnectableSharedControl = {
|
||||
getDiagnostics(): Pick<RemoteRuntimeSharedConnectionDiagnostics, 'state'> | null
|
||||
reconnectNow(): void
|
||||
}
|
||||
|
||||
/**
|
||||
* Reconnect unless a socket is already dialing or up: reconnecting closes and reopens a socket that
|
||||
* is not ready yet, so it would throw away a connection a moment from ready (e.g. one pre-opened
|
||||
* for a wake) and start over.
|
||||
*/
|
||||
export function reconnectSharedControlIfIdle(
|
||||
connection: ReconnectableSharedControl | undefined
|
||||
): void {
|
||||
const state = connection?.getDiagnostics()?.state
|
||||
if (state === 'awaiting_ready' || state === 'awaiting_authenticated' || state === 'ready') {
|
||||
return
|
||||
}
|
||||
connection?.reconnectNow()
|
||||
}
|
||||
@@ -24,11 +24,6 @@ export function useRemoteRuntimeRecoveryTriggers(): void {
|
||||
readyByEnvironment.set(environmentId, ready)
|
||||
if (ready && !wasReady) {
|
||||
reviveRemoteRuntimePtyRecoveriesForEnvironment(environmentId)
|
||||
// A host that just came back (e.g. a woken VM) would otherwise wait out a pane's
|
||||
// backoff step (up to seconds) before re-attaching. Pending pane backoffs are not
|
||||
// keyed by host, so advance them all, as the online/resume triggers do; panes of
|
||||
// hosts that are still down simply fail once more and back off again.
|
||||
retryAllRemoteRuntimePtyRecoveriesNow()
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user