From 0533b5cd9feb494aaec3ff1b5bcb649c69e33d0a Mon Sep 17 00:00:00 2001 From: Neil Date: Mon, 9 Nov 2026 05:02:55 -0800 Subject: [PATCH] fix(cloud): review fixes for wake/restart/archive races, output budget, archived compat, OOM dedupe, scheduler handoff # Conflicts: # cloud/apps/sandbox-control/src/index.ts # cloud/apps/sandbox-control/src/lifecycle.ts # Conflicts: # cloud/apps/sandbox-control/src/index.ts --- cloud/apps/sandbox-control/src/index.ts | 36 ++-- cloud/apps/sandbox-control/src/lifecycle.ts | 22 ++- .../sandbox-control/src/sandbox-vm-inspect.ts | 13 +- cloud/apps/sandbox-control/src/wire.ts | 5 +- .../src/workspace-lifecycle.ts | 83 +++++++- .../test/workspace-lifecycle.test.ts | 55 +++++- .../hosted-sandbox-control-plane-client.ts | 2 + .../hosted-sandbox/hosted-sandbox-install.ts | 4 +- .../hosted-workspace-runtime.test.ts | 62 ++++++ .../hosted-workspace-runtime.ts | 136 +++++++++++++ .../hosted-workspace-service.ts | 178 +++++++----------- 11 files changed, 461 insertions(+), 135 deletions(-) create mode 100644 src/main/hosted-sandbox/hosted-workspace-runtime.test.ts create mode 100644 src/main/hosted-sandbox/hosted-workspace-runtime.ts diff --git a/cloud/apps/sandbox-control/src/index.ts b/cloud/apps/sandbox-control/src/index.ts index 0f4479d778f..fa819570da5 100644 --- a/cloud/apps/sandbox-control/src/index.ts +++ b/cloud/apps/sandbox-control/src/index.ts @@ -19,6 +19,7 @@ // Browser (GitHub redirects): GET /v1/github/setup, GET /v1/github/callback // Operator (OPERATOR_TOKEN): /v1/internal/* (see operator.ts) // Vercel network-policy forwardURL target (Vercel sandbox OIDC): /v1/egress/proxy +import { z } from 'zod' import { maintainPool, poolConfigFromEnv } from './pool' import { authenticateDesktop } from './auth' import { handleApi, isApiPath, type ApiExtension } from './api/router' @@ -82,6 +83,14 @@ import { } from './workspace-lifecycle' const MAX_JSON_BYTES = 16 * 1024 +/** Archive bodies carry the desktop's cached repo state (file trees) when the VM is already asleep. */ +const MAX_ARCHIVE_JSON_BYTES = 256 * 1024 +const CentsPerHourSchema = z + .object({ standard: z.number().int().nonnegative(), large: z.number().int().nonnegative() }) + .partial() + .strict() +/** Desktops that understand `archived` say so; older strict clients see it as `suspended`. */ +const FEATURES_HEADER = 'x-orca-sandbox-features' export type Deps = { provider?: SandboxProviderApi @@ -108,7 +117,8 @@ function workspaceConfig(env: Env, deps: Deps): WorkspaceConfig { const days = Number(env.ARCHIVE_TTL_DAYS) let centsPerHour: WorkspaceConfig['centsPerHour'] try { - centsPerHour = env.COST_CENTS_PER_HOUR ? (JSON.parse(env.COST_CENTS_PER_HOUR) as WorkspaceConfig['centsPerHour']) : undefined + const parsed = CentsPerHourSchema.safeParse(env.COST_CENTS_PER_HOUR ? JSON.parse(env.COST_CENTS_PER_HOUR) : undefined) + centsPerHour = parsed.success ? parsed.data : undefined } catch { centsPerHour = undefined } @@ -319,9 +329,9 @@ function fail(error: string, status: number, reason?: string): Response { return json(reason ? { error, reason } : { error }, status) } -async function readJson(request: Request): Promise { +async function readJson(request: Request, maxBytes = MAX_JSON_BYTES): Promise { const text = await request.text() - if (text.length > MAX_JSON_BYTES) { + if (text.length > maxBytes) { throw new SyntaxError('too_large') } return text ? JSON.parse(text) : {} @@ -563,11 +573,13 @@ export async function handleRequest(request: Request, env: Env, deps: Deps = {}) return fail('not_configured', 503) } const [, id, action] = match + const archivedAware = (request.headers.get(FEATURES_HEADER) ?? '').split(',').map((f) => f.trim()).includes('archive') + const view = (row: SandboxRow) => sandboxView(row, archivedAware) if (!id) { if (request.method === 'GET') { // Routine runs and other machine-made workspaces stay out of the desktop's list. const rows = (await listAccountSandboxes(env.SANDBOX_DB, caller.accountId)).filter((row) => (row as { hidden?: number }).hidden !== 1) - return json({ sandboxes: rows.map(sandboxView) }) + return json({ sandboxes: rows.map(view) }) } if (request.method !== 'POST') { return fail('not_found', 404) @@ -588,11 +600,11 @@ export async function handleRequest(request: Request, env: Env, deps: Deps = {}) const route = `${request.method} ${action ?? ''}` switch (route) { case 'GET ': - return json({ sandbox: sandboxView(row) }) + return json({ sandbox: view(row) }) case 'DELETE ': - return json({ sandbox: sandboxView(await destroySandbox(life, row)) }) + return json({ sandbox: view(await destroySandbox(life, row)) }) case 'POST suspend': - return json({ sandbox: sandboxView(await suspendSandbox(life, row)) }) + return json({ sandbox: view(await suspendSandbox(life, row)) }) case 'POST resume': { const ready = await resumeSandbox(life, row) return readyResponse(ready) @@ -604,17 +616,19 @@ export async function handleRequest(request: Request, env: Env, deps: Deps = {}) case 'POST wake': return json(await wakeSandbox(life, row, { waitMs: parseWakeWait(url, await readJson(request)), ...workspaceConfig(env, deps) })) case 'POST restart': { + // User-triggered and unbounded otherwise: shares the per-account start rate limit. + await consumeCreateToken(life.db, life.config.guardrails!, caller.accountId, life.now()) const ready = await restartSandbox(life, row) return json(readyView(ready.row, ready.connection)) } case 'GET status': - return json({ sandbox: sandboxView(row), status: await workspaceStatus(life, row, workspaceConfig(env, deps)) }) + return json({ sandbox: view(row), status: await workspaceStatus(life, row, workspaceConfig(env, deps)) }) case 'POST archive': { - const body = ArchiveRequestSchema.safeParse(await readJson(request)) + const body = ArchiveRequestSchema.safeParse(await readJson(request, MAX_ARCHIVE_JSON_BYTES)) if (!body.success) { return fail('invalid_request', 400) } - return json({ sandbox: sandboxView(await archiveSandbox(life, row, body.data, workspaceConfig(env, deps))) }) + return json({ sandbox: view(await archiveSandbox(life, row, body.data, workspaceConfig(env, deps))) }) } case 'GET archive': return json({ archive: await readArchive(env.SANDBOX_DB, row.id) }) @@ -627,7 +641,7 @@ export async function handleRequest(request: Request, env: Env, deps: Deps = {}) if (!body.success) { return fail('invalid_request', 400) } - return json({ sandbox: sandboxView(await renewLease(life, row, body.data.extendMs)) }) + return json({ sandbox: view(await renewLease(life, row, body.data.extendMs)) }) } default: return fail('not_found', 404) diff --git a/cloud/apps/sandbox-control/src/lifecycle.ts b/cloud/apps/sandbox-control/src/lifecycle.ts index 07661bd4f78..0e0efc4f593 100644 --- a/cloud/apps/sandbox-control/src/lifecycle.ts +++ b/cloud/apps/sandbox-control/src/lifecycle.ts @@ -549,10 +549,7 @@ export async function resumeSandbox(deps: LifecycleDeps, row: SandboxRow): Promi const state = await deps.provider.getSandbox(row.provider_name, { resume: false }) if (state?.session.status === 'running') { // Idempotent: a running, serving sandbox answers at once (desktop wake-on-open after a restart). - const connection = - (await runningOrcad(deps.provider, state.session.id)) ?? - (await startOrcad(deps.provider, state, { sandboxId: row.id })) - return { row, connection } + return adoptRunning(deps, row, state) } // The provider ended the session (lease lapsed): fall through to a real resume. row = await markProviderStopped(deps, row) @@ -565,6 +562,20 @@ export async function resumeSandbox(deps: LifecycleDeps, row: SandboxRow): Promi return wake(deps, resuming) } +/** + * A running session whose orcad may not be up (a retried call, or a row the cron repaired after a + * wake died mid-way): start orcad (idempotent per boot) and record its endpoint if missing. + */ +export async function adoptRunning(deps: LifecycleDeps, row: SandboxRow, state: ProviderSandboxState): Promise { + const connection = + (await runningOrcad(deps.provider, state.session.id)) ?? (await startOrcad(deps.provider, state, { sandboxId: row.id })) + if (row.orcad_endpoint && row.ready_at !== null) { + return { row, connection } + } + const updated = await transition(deps.db, row, { orcad_endpoint: orcadEndpoint(state), ready_at: deps.now() }, deps.now()) + return { row: updated ?? row, connection } +} + export async function wake(deps: LifecycleDeps, row: SandboxRow): Promise { const timer = phaseTimer(deps.now) // The provider boot is the long pole; secrets and the repo token resolve alongside it. @@ -583,7 +594,8 @@ export async function wake(deps: LifecycleDeps, row: SandboxRow): Promise } catch (error) { // Leave it resumable: the disk is intact, only this boot failed. const current = await getSandbox(deps.db, row.id) - if (current?.status === 'resuming') { + // Only undo our own claim: a Restart that took the row over owns it now. + if (current?.status === 'resuming' && current.generation === row.generation) { await transition(deps.db, current, { status: 'suspended', ...NO_SESSION, last_error: 'resume_failed' }, deps.now()) await deps.hooks?.afterStop?.(current).catch(() => undefined) } diff --git a/cloud/apps/sandbox-control/src/sandbox-vm-inspect.ts b/cloud/apps/sandbox-control/src/sandbox-vm-inspect.ts index 0714f92fa42..0f5aadb85a3 100644 --- a/cloud/apps/sandbox-control/src/sandbox-vm-inspect.ts +++ b/cloud/apps/sandbox-control/src/sandbox-vm-inspect.ts @@ -24,7 +24,7 @@ read disk_total disk_used < <(df -B1 --output=size,used ${PROJECT} / 2>/dev/null uptime_s=$(awk '{printf "%d", $1}' /proc/uptime) oom=$(awk '/^oom_kill /{print $2}' /proc/vmstat 2>/dev/null); oom=\${oom:-0} victim=$(sudo -n dmesg 2>/dev/null | grep -oE 'Killed process [0-9]+ \([^)]{1,64}\)' | tail -n 1 | sed -E 's/.*\((.*)\)/\1/' | tr -cd 'A-Za-z0-9._+-' | head -c 64) -orcad=false; pid=$(cat "$HOME/.orca-sandbox/orcad.pid" 2>/dev/null || echo 0); kill -0 "$pid" 2>/dev/null && orcad=true +orcad=false; pid=$(cat "$HOME/.orca-sandbox/orcad.pid" 2>/dev/null || echo 0); [ "$pid" -gt 1 ] 2>/dev/null && kill -0 "$pid" 2>/dev/null && orcad=true load=$(awk '{print $1}' /proc/loadavg) printf '{"cpuPercent":%d,"cpus":%d,"load1":%s,"memTotalBytes":%d,"memUsedBytes":%d,"diskTotalBytes":%d,"diskUsedBytes":%d,"uptimeSeconds":%d,"oomKills":%d,"lastOomVictim":%s,"orcadAlive":%s}\n' \ "$cpu_pct" "$(nproc)" "$load" "$mem_total" "$((mem_total-mem_avail))" "\${disk_total:-0}" "\${disk_used:-0}" "$uptime_s" "$oom" \ @@ -52,6 +52,8 @@ export type VmMetrics = z.infer const MAX_TREE = 2000 const MAX_CHANGED = 300 const MAX_COMMITS = 20 +/** Under the provider's 256 KiB command-output cap, with room for the stream framing. */ +export const MAX_OUTPUT_BYTES = 200 * 1024 /** * Archive metadata for every git checkout in the project (the repo plus its worktrees): branch, @@ -89,7 +91,14 @@ const checkouts = fs.readFileSync("/tmp/orca-archive-wt", "utf8").split("\n").fi files: tree.slice(0, ${MAX_TREE}).map((f) => f.slice(0, 1024)) } }) -process.stdout.write(JSON.stringify({ checkouts: checkouts.filter((c) => c.head || c.branch) }) + "\n") +// The provider caps command output (256 KiB): keep the answer well under it. The main checkout keeps +// its file tree first; trees are trimmed, then dropped, until the JSON fits. +const kept = checkouts.filter((c) => c.head || c.branch) +const fits = () => Buffer.byteLength(JSON.stringify({ checkouts: kept })) <= ${MAX_OUTPUT_BYTES} +for (let i = kept.length - 1; i >= 0 && !fits(); i--) { kept[i].files = i === 0 ? kept[i].files.slice(0, 500) : [] } +for (let i = kept.length - 1; i >= 0 && !fits(); i--) { kept[i].files = []; kept[i].changes = kept[i].changes.slice(0, 50) } +while (kept.length > 1 && !fits()) kept.pop() +process.stdout.write(JSON.stringify({ checkouts: kept }) + "\n") ' `.replaceAll('\\${', '${') diff --git a/cloud/apps/sandbox-control/src/wire.ts b/cloud/apps/sandbox-control/src/wire.ts index d270017a8a6..473e2603450 100644 --- a/cloud/apps/sandbox-control/src/wire.ts +++ b/cloud/apps/sandbox-control/src/wire.ts @@ -38,7 +38,8 @@ export const LeaseRequestSchema = z }) .strict() -export function sandboxView(row: SandboxRow) { +/** `archivedAware`: the caller's schema knows `archived`; older strict desktops see `suspended`. */ +export function sandboxView(row: SandboxRow, archivedAware = false) { const session = row.session_id !== null && row.session_started_at !== null && @@ -54,7 +55,7 @@ export function sandboxView(row: SandboxRow) { return { id: row.id, workspaceId: row.workspace_id, - status: row.status, + status: row.status === 'archived' && !archivedAware ? ('suspended' as const) : row.status, machineSize: row.machine_size, region: row.region, session, diff --git a/cloud/apps/sandbox-control/src/workspace-lifecycle.ts b/cloud/apps/sandbox-control/src/workspace-lifecycle.ts index f534be2b65d..52e913ec2a7 100644 --- a/cloud/apps/sandbox-control/src/workspace-lifecycle.ts +++ b/cloud/apps/sandbox-control/src/workspace-lifecycle.ts @@ -6,13 +6,14 @@ import { GuardrailError, killSwitchOn } from './guardrails' import { DEFAULT_SNAPSHOT_EXPIRATION_MS, LifecycleError, + SESSION_DEADLINE_MS, + adoptRunning, claim, guardrails, markProviderStopped, requireStart, resumeSandbox, stopForSuspend, - suspendSandbox, wake, type LifecycleDeps, type Ready @@ -36,7 +37,13 @@ const POLL_MS = 400 /** Restart gives a wedged orcad this long to write durable offers before the stop. */ const RESTART_DRAIN_SECONDS = 15 /** A resuming/suspending row older than this is treated as stuck by Restart (the cron's bar is 15 min). */ -const RESTART_STUCK_MS = 90_000 +const RESTART_STUCK_MS = 5 * 60_000 +/** A wake whose Worker died (waitUntil cut it off) is picked up by the next wake call after this. */ +const WAKE_STALE_MS = 60_000 +/** A wake that just failed answers `failed` for this long instead of booting again on every poll. */ +const WAKE_FAILURE_HOLD_MS = 60_000 +/** Archive gives orcad this long to write durable offers (the desktop already ran its Sleep). */ +const ARCHIVE_DRAIN_SECONDS = 60 export const DEFAULT_CENTS_PER_HOUR: Record = { standard: 75, large: 150 } export type WorkspaceConfig = { @@ -81,6 +88,10 @@ function wakeAnswer(row: SandboxRow | null): WakeAnswer { } } +function recentlyFailed(deps: LifecycleDeps, row: SandboxRow): boolean { + return row.status === 'suspended' && row.last_error === 'resume_failed' && deps.now() - row.updated_at < WAKE_FAILURE_HOLD_MS +} + const defaultSleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) /** The row says running, but a lapsed lease means the provider may have stopped it on its own. */ @@ -106,6 +117,17 @@ export async function wakeSandbox( const sleep = options.sleep ?? defaultSleep let current = row.status === 'running' ? await confirmRunning(deps, row) : row let ours: Promise | null = null + if (current.status === 'running' && !current.orcad_endpoint) { + // Repaired by the cron after a wake died: the VM runs, orcad may not. + const state = await deps.provider.getSandbox(current.provider_name, { resume: false }) + current = state?.session.status === 'running' ? (await adoptRunning(deps, current, state)).row : await markProviderStopped(deps, current) + } + if (current.status === 'resuming' && deps.now() - current.updated_at >= WAKE_STALE_MS) { + current = await adoptStaleWake(deps, current) + } + if (recentlyFailed(deps, current)) { + return { state: 'failed', endpoint: null, readyAt: null, sandboxId: current.id } + } if (current.status === 'suspended') { await requireStart(deps, current.account_id, current.user_id, current.machine_size) const resuming = await transition(deps.db, current, { status: 'resuming' }, deps.now()) @@ -127,9 +149,43 @@ export async function wakeSandbox( await sleep(Math.min(POLL_MS, Math.max(1, deadline - deps.now()))) current = (await getSandbox(deps.db, row.id)) ?? current } + if (recentlyFailed(deps, current)) { + return { state: 'failed', endpoint: null, readyAt: null, sandboxId: current.id } + } return wakeAnswer(current) } +/** + * The Worker running a wake was cut off (waitUntil limit) and left the row `resuming`: adopt what + * the provider says — a running session gets orcad started and becomes `running`; otherwise the + * row goes back to `suspended` so this wake can start a fresh one. + */ +async function adoptStaleWake(deps: LifecycleDeps, row: SandboxRow): Promise { + const state = await deps.provider.getSandbox(row.provider_name, { resume: false }) + if (state?.session.status !== 'running') { + return markProviderStopped(deps, row) + } + const now = deps.now() + const running = await transition( + deps.db, + row, + { + status: 'running', + session_id: state.session.id, + session_started_at: state.session.startedAt ?? now, + session_deadline_at: (state.session.startedAt ?? now) + (deps.config.sessionDeadlineMs ?? SESSION_DEADLINE_MS), + lease_expires_at: (state.session.startedAt ?? now) + state.session.timeoutMs, + usage_settled_at: state.session.startedAt ?? now, + orcad_endpoint: null + }, + now + ) + if (!running) { + return (await getSandbox(deps.db, row.id)) ?? row + } + return (await adoptRunning(deps, running, state)).row +} + /** * "Restart" for an unresponsive workspace: one rehydrate path (short drain, stop with snapshot, * boot the same disk). Works from running, sleeping, or a wake/sleep that got stuck mid-way. @@ -146,6 +202,10 @@ export async function restartSandbox(deps: LifecycleDeps, row: SandboxRow): Prom } current = await markProviderStopped(deps, current) } + if (current.status === 'running') { + // A lapsed lease means the provider already stopped it; resume instead of stopping a dead session. + current = await confirmRunning(deps, current) + } if (current.status === 'suspended') { return resumeSandbox(deps, current) } @@ -190,10 +250,11 @@ export async function deleteArchive(db: D1Database, sandboxId: string): Promise< */ export async function archiveSandbox( deps: LifecycleDeps, - row: SandboxRow, + input: SandboxRow, request: Partial, config: WorkspaceConfig = {} ): Promise { + let row = input if (row.status === 'archived') { return row } @@ -201,6 +262,11 @@ export async function archiveSandbox( throw new LifecycleError('conflict', `sandbox_${row.status}`) } const captured = row.status === 'running' && row.session_id ? await captureArchiveMetadata(deps.provider, row.session_id) : null + // The capture takes seconds; a lease renew may have moved the row meanwhile. + row = (await getSandbox(deps.db, row.id)) ?? row + if (row.status !== 'running' && row.status !== 'suspended') { + throw new LifecycleError('conflict', `sandbox_${row.status}`) + } const previous = await readArchive(deps.db, row.id) const checkouts = captured?.checkouts ?? request.checkouts ?? previous?.metadata.checkouts ?? [] const source = captured ? 'vm' : request.checkouts ? 'client' : previous ? previous.source : 'none' @@ -211,7 +277,12 @@ export async function archiveSandbox( { checkouts, pullRequests: request.pullRequests ?? previous?.metadata.pullRequests ?? [] }, deps.now() ) - const stopped = row.status === 'running' ? await suspendSandbox(deps, row) : row + let stopped = row + if (row.status === 'running') { + // Short drain: the desktop's Sleep normally ran first; a long drain would outlive the caller. + stopped = await stopForSuspend(deps, await claim(deps, row, { status: 'suspending' }), { drainSeconds: ARCHIVE_DRAIN_SECONDS }) + await flushUsage(deps.db, deps.entitlements, deps.now(), row.id) + } const ttl = config.archiveTtlMs ?? DEFAULT_ARCHIVE_TTL_MS await deps.provider .updateSnapshotExpiration(stopped.provider_name, ttl + ARCHIVE_PROVIDER_GRACE_MS) @@ -281,7 +352,9 @@ function stateOf(row: SandboxRow, metrics: VmMetrics | null, probed: boolean): W } /** The status panel's one call. Metrics are read live from the VM only when it is running. */ -export async function workspaceStatus(deps: LifecycleDeps, row: SandboxRow, config: WorkspaceConfig = {}): Promise { +export async function workspaceStatus(deps: LifecycleDeps, input: SandboxRow, config: WorkspaceConfig = {}): Promise { + // A lapsed lease: the provider may have stopped it, which is "sleeping", not "unresponsive". + const row = input.status === 'running' ? await confirmRunning(deps, input) : input const probed = row.status === 'running' && row.session_id !== null const metrics = probed ? await readVmMetrics(deps.provider, row.session_id!) : null const now = deps.now() diff --git a/cloud/apps/sandbox-control/test/workspace-lifecycle.test.ts b/cloud/apps/sandbox-control/test/workspace-lifecycle.test.ts index f7cf9ee7b0e..6b2d022d9d7 100644 --- a/cloud/apps/sandbox-control/test/workspace-lifecycle.test.ts +++ b/cloud/apps/sandbox-control/test/workspace-lifecycle.test.ts @@ -121,6 +121,9 @@ describe('restart', () => { const { h, id } = await running() h.raw.prepare(`UPDATE sandboxes SET status = 'resuming', updated_at = ? WHERE id = ?`).run(h.now(), id) h.advance(2 * 60_000) + // A slow-but-alive wake is not taken over. + expect((await h.call('POST', `/v1/sandboxes/${id}/restart`)).status).toBe(409) + h.advance(4 * 60_000) const res = await h.call('POST', `/v1/sandboxes/${id}/restart`) expect(res.status).toBe(200) expect(res.body.sandbox.status).toBe('running') @@ -168,7 +171,14 @@ describe('archive', () => { script === ARCHIVE_SCRIPT ? { exitCode: 0, stdout: `${JSON.stringify({ checkouts: [checkout] })}\n`, stderr: '' } : undefined const pr = { number: 7, title: 'Add x', state: 'open', url: 'https://github.com/o/r/pull/7', branch: 'feature/x' } const archived = await h.call('POST', `/v1/sandboxes/${id}/archive`, { body: { pullRequests: [pr] } }) - expect(archived.body.sandbox.status).toBe('archived') + // Callers that don't advertise `archive` (older strict desktops) see it as suspended. + expect(archived.body.sandbox.status).toBe('suspended') + const aware = await h.send( + new Request(`https://sandbox.test/v1/sandboxes/${id}`, { + headers: { authorization: `Bearer ${await h.token('alice')}`, 'x-orca-sandbox-features': 'archive' } + }) + ) + expect(((await aware.json()) as any).sandbox.status).toBe('archived') const name = `orca-test-${id}` expect(h.vercel.boxes.get(name)!.status).toBe('stopped') expect(h.vercel.snapshotExpirations.get(name)).toBe(37 * DAY) @@ -237,3 +247,46 @@ describe('desktop client against the real handler', () => { expect((await client.unarchive(id)).sandbox.status).toBe('running') }) }) + +describe('wake recovery', () => { + it('adopts a wake whose Worker was cut off: running VM gets orcad and an endpoint', async () => { + const { h, id } = await running() + await h.call('POST', `/v1/sandboxes/${id}/suspend`) + // Simulate a wake that booted the VM, then lost its Worker before orcad/claim. + await h.vercel.getSandbox(`orca-test-${id}`, { resume: true }) + h.raw.prepare(`UPDATE sandboxes SET status = 'resuming', updated_at = ? WHERE id = ?`).run(h.now(), id) + expect((await h.call('POST', '/v1/workspaces/ws-1/wake')).body.state).toBe('waking') + h.advance(61_000) + const answer = HostedWakeResponseSchema.parse((await h.call('POST', '/v1/workspaces/ws-1/wake')).body) + expect(answer.state).toBe('ready') + expect(answer.endpoint).toMatch(/^wss:/) + }) + + it('a stale wake whose VM never came up goes back to sleep and wakes again', async () => { + const { h, id } = await running() + await h.call('POST', `/v1/sandboxes/${id}/suspend`) + h.raw.prepare(`UPDATE sandboxes SET status = 'resuming', updated_at = ? WHERE id = ?`).run(h.now(), id) + h.advance(61_000) + expect((await h.call('POST', '/v1/workspaces/ws-1/wake?wait=30s')).body.state).toBe('ready') + }) + + it('answers failed after a failed wake instead of booting on every poll', async () => { + const { h, id } = await running() + await h.call('POST', `/v1/sandboxes/${id}/suspend`) + h.vercel.failStart = true + expect((await h.call('POST', '/v1/workspaces/ws-1/wake?wait=30s')).body.state).toBe('failed') + const starts = h.vercel.commands.filter((c) => c.command === START_COMMAND).length + expect((await h.call('POST', '/v1/workspaces/ws-1/wake?wait=30s')).body.state).toBe('failed') + expect(h.vercel.commands.filter((c) => c.command === START_COMMAND)).toHaveLength(starts) + h.vercel.failStart = false + h.advance(61_000) + expect((await h.call('POST', '/v1/workspaces/ws-1/wake?wait=30s')).body.state).toBe('ready') + }) + + it('reports a provider-stopped VM as sleeping, not unresponsive', async () => { + const { h, id } = await running() + h.vercel.expire(`orca-test-${id}`) + h.advance(31 * 60_000) + expect((await h.call('GET', `/v1/sandboxes/${id}/status`)).body.status.state).toBe('sleeping') + }) +}) diff --git a/src/main/hosted-sandbox/hosted-sandbox-control-plane-client.ts b/src/main/hosted-sandbox/hosted-sandbox-control-plane-client.ts index 8331021a354..8f32e39f21b 100644 --- a/src/main/hosted-sandbox/hosted-sandbox-control-plane-client.ts +++ b/src/main/hosted-sandbox/hosted-sandbox-control-plane-client.ts @@ -87,6 +87,8 @@ export function createHostedSandboxControlPlaneClient( headers: { authorization: `Bearer ${token}`, accept: 'application/json', + // This client's schemas know the `archived` status. + 'x-orca-sandbox-features': 'archive', ...(init.body === undefined ? {} : { 'content-type': 'application/json' }) }, body: init.body === undefined ? undefined : JSON.stringify(init.body), diff --git a/src/main/hosted-sandbox/hosted-sandbox-install.ts b/src/main/hosted-sandbox/hosted-sandbox-install.ts index a01e43a0d83..1c132e320af 100644 --- a/src/main/hosted-sandbox/hosted-sandbox-install.ts +++ b/src/main/hosted-sandbox/hosted-sandbox-install.ts @@ -57,7 +57,8 @@ export function installHostedSandbox(): void { userDataPath, client, windows: () => BrowserWindow.getAllWindows(), - syncSessions: () => void registry.sync() + syncSessions: () => void registry.sync(), + stopSession: (environmentId) => registry.stopEnvironment(environmentId) }) setHostedWorkspaceService(workspaces) // Records land after the step answers and wakes can come from any path, so reconcile on a slow beat too. @@ -80,6 +81,7 @@ export function installHostedSandbox(): void { phase, startedAt, message, + // Later phases inherit the source recorded at `waking`. source: phase === 'waking' ? workspaces.takeWakeSource(workspaceId) : 'open' }) }) diff --git a/src/main/hosted-sandbox/hosted-workspace-runtime.test.ts b/src/main/hosted-sandbox/hosted-workspace-runtime.test.ts new file mode 100644 index 00000000000..99c7ab33592 --- /dev/null +++ b/src/main/hosted-sandbox/hosted-workspace-runtime.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, it } from 'vitest' +import type { CloudOomEvent } from '../../shared/hosted-sandbox/cloud-workspace-api' +import type { HostedWorkspaceStatus } from '../../shared/hosted-sandbox/hosted-sandbox-wire' +import { createOomTracker } from './hosted-workspace-runtime' + +function status(checkedAt: number, uptimeSeconds: number, oomKills: number): HostedWorkspaceStatus { + return { + state: 'ready', + metrics: { + cpuPercent: 1, + cpus: 4, + load1: 0, + memTotalBytes: 8, + memUsedBytes: 1, + diskTotalBytes: 8, + diskUsedBytes: 1, + uptimeSeconds, + oomKills, + lastOomVictim: oomKills ? 'node' : null, + orcadAlive: true + }, + uptimeMs: 0, + usage: { totalMs: 0, centsPerHour: 75, costCents: 0 }, + archive: null, + checkedAt + } +} + +describe('OOM tracker', () => { + it('announces each new kill once per boot, despite clock jitter', () => { + const events: CloudOomEvent[] = [] + const tracker = createOomTracker((event) => events.push(event)) + const t0 = 1_000_000_000 + tracker.observe('ws', 'sbx', status(t0, 100, 1)) + // Same boot, readings straddle a minute boundary: no repeat. + tracker.observe('ws', 'sbx', status(t0 + 60_400, 160, 1)) + tracker.observe('ws', 'sbx', status(t0 + 120_000, 220, 2)) + expect(events.map((e) => e.oomKills)).toEqual([1, 2]) + expect(events[0]?.victim).toBe('node') + }) + + it('treats a new boot as a fresh counter and forgets stopped sandboxes', () => { + const events: CloudOomEvent[] = [] + const tracker = createOomTracker((event) => events.push(event)) + tracker.observe('ws', 'sbx', status(1_000_000_000, 100, 1)) + // Restarted: uptime is small again, counter restarts at 1. + tracker.observe('ws', 'sbx', status(1_000_600_000, 20, 1)) + expect(events).toHaveLength(2) + tracker.prune(new Set()) + tracker.observe('ws', 'sbx', status(1_000_660_000, 80, 1)) + expect(events).toHaveLength(3) + }) + + it('ignores a status without metrics', () => { + const events: CloudOomEvent[] = [] + createOomTracker((event) => events.push(event)).observe('ws', 'sbx', { + ...status(1, 1, 0), + metrics: null + }) + expect(events).toEqual([]) + }) +}) diff --git a/src/main/hosted-sandbox/hosted-workspace-runtime.ts b/src/main/hosted-sandbox/hosted-workspace-runtime.ts new file mode 100644 index 00000000000..8c470d3573c --- /dev/null +++ b/src/main/hosted-sandbox/hosted-workspace-runtime.ts @@ -0,0 +1,136 @@ +// Orca Cloud workspace lookups shared by the workspace service and its IPC: workspace id → runtime +// record → sandbox id, error mapping for the window, and OOM-notice dedupe. +import { listEphemeralVmRuntimes } from '../../shared/ephemeral-vm-runtime-store' +import type { EphemeralVmRuntimeRecord } from '../../shared/ephemeral-vm-runtimes' +import type { + CloudOomEvent, + CloudWorkspaceErrorCode, + CloudWorkspaceResult +} from '../../shared/hosted-sandbox/cloud-workspace-api' +import type { + HostedArchiveCheckout, + HostedWorkspaceStatus +} from '../../shared/hosted-sandbox/hosted-sandbox-wire' +import { + isEnvironmentProviderError, + type EnvironmentProviderErrorCode +} from '../environment-providers/environment-provider-driver' +import { environmentHandleFromRecipeResult } from '../environment-providers/environment-provider-recipe-adapter' +import { HOSTED_SANDBOX_DRIVER_ID } from './hosted-sandbox-driver' +import type { HostedWorkspaceService } from './hosted-workspace-service' + +export type CloudRuntime = { runtime: EphemeralVmRuntimeRecord; sandboxId: string } + +let cachedCheckouts: ((workspaceId: string) => Promise) | null = + null +export function setArchiveCheckoutsProvider( + provider: ((workspaceId: string) => Promise) | null +): void { + cachedCheckouts = provider +} + +export function findCloudRuntime(userDataPath: string, workspaceId: string): CloudRuntime | null { + for (const runtime of listEphemeralVmRuntimes(userDataPath)) { + if (runtime.recipe?.provider !== 'orca-cloud' || runtime.workspaceId !== workspaceId) { + continue + } + const handle = runtime.recipeResult + ? environmentHandleFromRecipeResult(runtime.recipeResult, HOSTED_SANDBOX_DRIVER_ID) + : null + const sandboxId = handle?.driverState.sandboxId + if (typeof sandboxId === 'string' && sandboxId) { + return { runtime, sandboxId } + } + } + return null +} + +export function failure(error: unknown): CloudWorkspaceResult { + const codes: (CloudWorkspaceErrorCode & EnvironmentProviderErrorCode)[] = [ + 'unauthorized', + 'entitlement_denied', + 'conflict', + 'not_found' + ] + for (const code of codes) { + if (isEnvironmentProviderError(error, code)) { + return { ok: false, code, message: error.message, reason: error.reason ?? null } + } + } + if ( + isEnvironmentProviderError(error, 'unavailable') || + isEnvironmentProviderError(error, 'timeout') + ) { + return { ok: false, code: 'unavailable', message: error.message } + } + return { + ok: false, + code: 'failed', + message: error instanceof Error ? error.message : String(error) + } +} + +export const NOT_CLOUD = { + ok: false, + code: 'not_cloud', + message: 'Not an Orca Cloud workspace.' +} as const + +export function readCachedCheckouts(workspaceId: string): Promise { + return cachedCheckouts ? cachedCheckouts(workspaceId).catch(() => null) : Promise.resolve(null) +} + +/** + * Announces each new kernel OOM kill once. The counter resets per VM boot; two readings belong to + * one boot when their implied boot times agree within two minutes (uptime and checkedAt come from + * different clocks). + */ +export function createOomTracker(announce: (event: CloudOomEvent) => void) { + const seen = new Map() + return { + observe( + workspaceId: string, + sandboxId: string, + status: HostedWorkspaceStatus, + now = Date.now() + ): void { + const metrics = status.metrics + if (!metrics) { + return + } + const bootAt = status.checkedAt - metrics.uptimeSeconds * 1000 + const previous = seen.get(sandboxId) + const sameBoot = previous !== undefined && Math.abs(previous.bootAt - bootAt) < 120_000 + const announced = sameBoot ? previous.kills : 0 + seen.set(sandboxId, { + bootAt: sameBoot ? previous.bootAt : bootAt, + kills: Math.max(announced, metrics.oomKills) + }) + if (metrics.oomKills > announced) { + announce({ + workspaceId, + oomKills: metrics.oomKills, + victim: metrics.lastOomVictim, + at: now + }) + } + }, + /** Forget sandboxes that are no longer running. */ + prune(live: ReadonlySet): void { + for (const sandboxId of seen.keys()) { + if (!live.has(sandboxId)) { + seen.delete(sandboxId) + } + } + } + } +} + +let current: HostedWorkspaceService | null = null +export function setHostedWorkspaceService(next: HostedWorkspaceService | null): void { + current = next +} +/** Null when Orca Cloud is not configured in this build. */ +export function getHostedWorkspaceService(): HostedWorkspaceService | null { + return current +} diff --git a/src/main/hosted-sandbox/hosted-workspace-service.ts b/src/main/hosted-sandbox/hosted-workspace-service.ts index ed5f9d9b7a3..418a4e37974 100644 --- a/src/main/hosted-sandbox/hosted-workspace-service.ts +++ b/src/main/hosted-sandbox/hosted-workspace-service.ts @@ -5,38 +5,38 @@ import { listEphemeralVmRuntimes, updateEphemeralVmRuntimeStatus } from '../../shared/ephemeral-vm-runtime-store' -import type { EphemeralVmRuntimeRecord } from '../../shared/ephemeral-vm-runtimes' import { ORCA_CLOUD_WORKSPACE_CHANNELS as CH, - type CloudOomEvent, type CloudWakeProgressEvent, - type CloudWakeSource, - type CloudWorkspaceErrorCode, - type CloudWorkspaceResult + type CloudWakeSource } from '../../shared/hosted-sandbox/cloud-workspace-api' import type { - HostedArchiveCheckout, HostedArchivePullRequest, - HostedSandboxReadyResponse, - HostedWorkspaceStatus + HostedSandboxReadyResponse } from '../../shared/hosted-sandbox/hosted-sandbox-wire' -import { - isEnvironmentProviderError, - type EnvironmentProviderErrorCode -} from '../environment-providers/environment-provider-driver' -import { - environmentHandleFromRecipeResult, - environmentReadyToRecipeResult -} from '../environment-providers/environment-provider-recipe-adapter' +import { environmentReadyToRecipeResult } from '../environment-providers/environment-provider-recipe-adapter' import { applyResumedRuntimePairing } from '../ephemeral-vm/runtime-resume-pairing' import { noteRecipeVmWakePairing } from '../ephemeral-vm/recipe-vm-wake-pairings' import { invalidateRuntimeEnvironmentTransport } from '../ipc/runtime-environments/runtime-environments' import { autoResumeAgentsAfterRollover } from './hosted-agent-auto-resume' import type { HostedSandboxControlPlaneClient } from './hosted-sandbox-control-plane-client' -import { HOSTED_SANDBOX_DRIVER_ID, toReady } from './hosted-sandbox-driver' +import { toReady } from './hosted-sandbox-driver' import { createHostedRuntimeEnvironmentCall } from './hosted-runtime-environment-call' +import { + NOT_CLOUD, + createOomTracker, + failure, + findCloudRuntime, + readCachedCheckouts, + type CloudRuntime +} from './hosted-workspace-runtime' -type CloudRuntime = { runtime: EphemeralVmRuntimeRecord; sandboxId: string } +export { + findCloudRuntime, + getHostedWorkspaceService, + setArchiveCheckoutsProvider, + setHostedWorkspaceService +} from './hosted-workspace-runtime' export type HostedWorkspaceService = { emitWakeProgress(event: Omit): void @@ -49,73 +49,20 @@ export type HostedWorkspaceService = { } /** Git state from the client-side offline cache, used when archiving a VM that is already asleep. */ -let cachedCheckouts: ((workspaceId: string) => Promise) | null = - null -export function setArchiveCheckoutsProvider( - provider: ((workspaceId: string) => Promise) | null -): void { - cachedCheckouts = provider -} - -export function findCloudRuntime(userDataPath: string, workspaceId: string): CloudRuntime | null { - for (const runtime of listEphemeralVmRuntimes(userDataPath)) { - if (runtime.recipe?.provider !== 'orca-cloud' || runtime.workspaceId !== workspaceId) { - continue - } - const handle = runtime.recipeResult - ? environmentHandleFromRecipeResult(runtime.recipeResult, HOSTED_SANDBOX_DRIVER_ID) - : null - const sandboxId = handle?.driverState.sandboxId - if (typeof sandboxId === 'string' && sandboxId) { - return { runtime, sandboxId } - } - } - return null -} - -function failure(error: unknown): CloudWorkspaceResult { - const codes: (CloudWorkspaceErrorCode & EnvironmentProviderErrorCode)[] = [ - 'unauthorized', - 'entitlement_denied', - 'conflict', - 'not_found' - ] - for (const code of codes) { - if (isEnvironmentProviderError(error, code)) { - return { ok: false, code, message: error.message, reason: error.reason ?? null } - } - } - if ( - isEnvironmentProviderError(error, 'unavailable') || - isEnvironmentProviderError(error, 'timeout') - ) { - return { ok: false, code: 'unavailable', message: error.message } - } - return { - ok: false, - code: 'failed', - message: error instanceof Error ? error.message : String(error) - } -} - -const NOT_CLOUD = { - ok: false, - code: 'not_cloud', - message: 'Not an Orca Cloud workspace.' -} as const - export function installHostedWorkspaceService(deps: { userDataPath: string client: HostedSandboxControlPlaneClient windows: () => readonly BrowserWindow[] /** Restart lease schedulers after a lifecycle change. */ syncSessions: () => void + /** Stop the lease scheduler of one environment (its session is about to be replaced). */ + stopSession: (environmentId: string) => void log?: (line: string) => void }): HostedWorkspaceService { const log = deps.log ?? ((line: string) => console.error(line)) const wakeSources = new Map() - // Last OOM-kill count seen per sandbox session, so a banner fires once per new kill. - const oomSeen = new Map() + // Wakes in flight per workspace, so every phase carries the source the first one named. + const wakes = new Map() const broadcast = (channel: string, payload: unknown): void => { for (const window of deps.windows()) { if (!window.isDestroyed()) { @@ -123,26 +70,7 @@ export function installHostedWorkspaceService(deps: { } } } - - function noticeOom(workspaceId: string, sandboxId: string, status: HostedWorkspaceStatus): void { - const metrics = status.metrics - if (!metrics) { - return - } - // Per boot: uptime resets with each session, so key by sandbox and boot time (minute precision). - const bootKey = `${sandboxId}:${Math.round((status.checkedAt - metrics.uptimeSeconds * 1000) / 60_000)}` - const seen = oomSeen.get(bootKey) ?? 0 - if (metrics.oomKills > seen) { - oomSeen.set(bootKey, metrics.oomKills) - const event: CloudOomEvent = { - workspaceId, - oomKills: metrics.oomKills, - victim: metrics.lastOomVictim, - at: Date.now() - } - broadcast(CH.oom, event) - } - } + const oom = createOomTracker((event) => broadcast(CH.oom, event)) async function reconnect(found: CloudRuntime, ready: HostedSandboxReadyResponse): Promise { const runtimeEnvironmentId = found.runtime.runtimeEnvironmentId @@ -166,9 +94,37 @@ export function installHostedWorkspaceService(deps: { ) } + /** The step answers before the window re-pairs; `ready` fires once the runtime record says running. */ + async function announceReadyWhenRunning(workspaceId: string, startedAt: number): Promise { + const deadline = Date.now() + 60_000 + while (Date.now() < deadline && wakes.has(workspaceId)) { + const runtime = findCloudRuntime(deps.userDataPath, workspaceId)?.runtime + if (runtime?.status === 'running') { + service.emitWakeProgress({ workspaceId, phase: 'ready', source: 'open', startedAt }) + return + } + await new Promise((resolve) => setTimeout(resolve, 250)) + } + } + const service: HostedWorkspaceService = { emitWakeProgress(event) { - broadcast(CH.wakeProgress, { ...event, at: Date.now() } satisfies CloudWakeProgressEvent) + if (event.phase === 'waking') { + wakes.set(event.workspaceId, { source: event.source, startedAt: event.startedAt }) + } + const inFlight = wakes.get(event.workspaceId) + const source = inFlight?.source ?? event.source + if (event.phase === 'ready' || event.phase === 'failed') { + wakes.delete(event.workspaceId) + } + broadcast(CH.wakeProgress, { + ...event, + source, + at: Date.now() + } satisfies CloudWakeProgressEvent) + if (event.phase === 'connecting' && event.source !== 'restart') { + void announceReadyWhenRunning(event.workspaceId, event.startedAt) + } }, noteWakeSource(workspaceId, source) { wakeSources.set(workspaceId, source) @@ -179,6 +135,17 @@ export function installHostedWorkspaceService(deps: { return source }, async pollOom() { + const live = new Set( + listEphemeralVmRuntimes(deps.userDataPath) + .filter( + (runtime) => + runtime.recipe?.provider === 'orca-cloud' && + runtime.status === 'running' && + runtime.workspaceId + ) + .map((runtime) => findCloudRuntime(deps.userDataPath, runtime.workspaceId!)?.sandboxId) + ) + oom.prune(live) for (const runtime of listEphemeralVmRuntimes(deps.userDataPath)) { if ( runtime.recipe?.provider !== 'orca-cloud' || @@ -193,7 +160,7 @@ export function installHostedWorkspaceService(deps: { } try { const { status } = await deps.client.workspaceStatus(found.sandboxId) - noticeOom(runtime.workspaceId, found.sandboxId, status) + oom.observe(runtime.workspaceId, found.sandboxId, status) } catch { // Status is advisory; the next beat retries. } @@ -227,7 +194,7 @@ export function installHostedWorkspaceService(deps: { } try { const { status } = await deps.client.workspaceStatus(found.sandboxId) - noticeOom(args.workspaceId, found.sandboxId, status) + oom.observe(args.workspaceId, found.sandboxId, status) return { ok: true, value: status } } catch (error) { return failure(error) @@ -246,6 +213,8 @@ export function installHostedWorkspaceService(deps: { source: 'restart', startedAt }) + // The scheduler holds the old session's deadline and would warn/roll over the new one; a fresh one starts on sync. + deps.stopSession(found.sandboxId) try { const ready = await deps.client.restart(found.sandboxId) service.emitWakeProgress({ @@ -281,6 +250,7 @@ export function installHostedWorkspaceService(deps: { startedAt, message: error instanceof Error ? error.message : String(error) }) + deps.syncSessions() return failure(error) } }) @@ -293,7 +263,7 @@ export function installHostedWorkspaceService(deps: { return NOT_CLOUD } try { - const checkouts = (await cachedCheckouts?.(args.workspaceId).catch(() => null)) ?? undefined + const checkouts = (await readCachedCheckouts(args.workspaceId)) ?? undefined await deps.client.archive(found.sandboxId, { ...(args.pullRequests?.length ? { pullRequests: args.pullRequests.slice(0, 50) } : {}), ...(checkouts?.length ? { checkouts: checkouts.slice(0, 20) } : {}) @@ -322,6 +292,7 @@ export function installHostedWorkspaceService(deps: { try { // The control plane resumes the disk; the window then runs its ordinary wake, which re-pairs. await deps.client.unarchive(found.sandboxId) + deps.syncSessions() return { ok: true, value: null } } catch (error) { service.emitWakeProgress({ @@ -349,12 +320,3 @@ export function installHostedWorkspaceService(deps: { return service } - -let current: HostedWorkspaceService | null = null -export function setHostedWorkspaceService(next: HostedWorkspaceService | null): void { - current = next -} -/** Null when Orca Cloud is not configured in this build. */ -export function getHostedWorkspaceService(): HostedWorkspaceService | null { - return current -}