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
This commit is contained in:
Neil
2026-11-09 05:02:55 -08:00
parent 5c1e87e093
commit 0533b5cd9f
11 changed files with 461 additions and 135 deletions
+25 -11
View File
@@ -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<unknown> {
async function readJson(request: Request, maxBytes = MAX_JSON_BYTES): Promise<unknown> {
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)
+17 -5
View File
@@ -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<Ready> {
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<Ready> {
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<Ready>
} 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)
}
@@ -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<typeof VmMetricsSchema>
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('\\${', '${')
+3 -2
View File
@@ -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,
@@ -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<MachineSize, number> = { 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<void>((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<unknown> | 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<SandboxRow> {
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<ArchiveMetadata>,
config: WorkspaceConfig = {}
): Promise<SandboxRow> {
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<WorkspaceStatus> {
export async function workspaceStatus(deps: LifecycleDeps, input: SandboxRow, config: WorkspaceConfig = {}): Promise<WorkspaceStatus> {
// 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()
@@ -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')
})
})
@@ -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),
@@ -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'
})
})
@@ -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([])
})
})
@@ -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<HostedArchiveCheckout[] | null>) | null =
null
export function setArchiveCheckoutsProvider(
provider: ((workspaceId: string) => Promise<HostedArchiveCheckout[] | null>) | 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<T>(error: unknown): CloudWorkspaceResult<T> {
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<HostedArchiveCheckout[] | null> {
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<string, { bootAt: number; kills: number }>()
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<string | undefined>): 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
}
@@ -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<CloudWakeProgressEvent, 'at'>): 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<HostedArchiveCheckout[] | null>) | null =
null
export function setArchiveCheckoutsProvider(
provider: ((workspaceId: string) => Promise<HostedArchiveCheckout[] | null>) | 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<T>(error: unknown): CloudWorkspaceResult<T> {
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<string, CloudWakeSource>()
// Last OOM-kill count seen per sandbox session, so a banner fires once per new kill.
const oomSeen = new Map<string, number>()
// Wakes in flight per workspace, so every phase carries the source the first one named.
const wakes = new Map<string, { source: CloudWakeSource; startedAt: number }>()
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<void> {
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<void> {
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
}