mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
feat(updater): auto-park registered serve supervisors during a desktop install (#27722)
* feat(updater): auto-park registered serve supervisors during a desktop install * test(serve): drop a type assertion in the bundle park test * fix(updater): stop any server started during an install hold and re-list blockers right before commit --------- Co-authored-by: m4air <m4air@Mac.localdomain>
This commit is contained in:
@@ -36,6 +36,8 @@ import {
|
||||
} from './serve-electron-preflight'
|
||||
import { MacServeStartupAbortError } from './serve-signal-exit-diagnostic'
|
||||
import { superviseServeService } from './serve-supervised-service'
|
||||
import { isSupervisedSpawnHeld } from './serve-supervised-stop'
|
||||
import { openServeBundlePark } from './serve-bundle-park'
|
||||
|
||||
const USER_NAMESPACE_PROBE_TIMEOUT_MS = 2_000
|
||||
const STARTUP_ABORT_MS = 30_000
|
||||
@@ -126,6 +128,7 @@ export function serveOrcaApp(args: ServeOrcaAppArgs = {}): Promise<number> {
|
||||
executable,
|
||||
appRoot: resolveAppRoot(),
|
||||
force: args.force === true,
|
||||
park: openServeBundlePark({ executable }),
|
||||
// Why first only: --force overrides a remembered abort once; the brake owns later attempts.
|
||||
runOnce: () =>
|
||||
serveOnce(
|
||||
@@ -238,6 +241,9 @@ function spawnElectronServe(
|
||||
handoff: interruptedHandoff
|
||||
})
|
||||
}
|
||||
if (isSupervisedSpawnHeld()) {
|
||||
return Promise.resolve(0)
|
||||
}
|
||||
const child = spawnProcess(executable, childArgs, spawnOptions)
|
||||
const spawnedAt = Date.now()
|
||||
|
||||
|
||||
@@ -0,0 +1,175 @@
|
||||
import { mkdir, mkdtemp, readFile, readdir, rm, stat, writeFile } from 'node:fs/promises'
|
||||
import { realpathSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
acksDir,
|
||||
bundleCoordinationDir,
|
||||
formatBundleIdentity,
|
||||
parkRecordPath,
|
||||
supervisorsDir,
|
||||
type ServeParkRecord
|
||||
} from '../../shared/serve-bundle-coordination'
|
||||
import { openServeBundlePark } from './serve-bundle-park'
|
||||
import { superviseServeService } from './serve-supervised-service'
|
||||
import { createSupervisedParkWatch } from './serve-supervised-park-watch'
|
||||
|
||||
const LABEL = 'com.stablyai.orca.serve.default'
|
||||
|
||||
describe('openServeBundlePark', () => {
|
||||
let root: string
|
||||
let executable: string
|
||||
let dir: string
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-bundle-park-'))
|
||||
const bundle = join(root, 'Orca.app')
|
||||
executable = join(bundle, 'Contents', 'MacOS', 'Orca')
|
||||
await mkdir(join(bundle, 'Contents', 'MacOS'), { recursive: true })
|
||||
await writeFile(executable, '')
|
||||
await writeFile(join(bundle, 'Contents', 'Info.plist'), 'v1')
|
||||
dir = bundleCoordinationDir(realpathSync(bundle), join(root, 'data'))
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
const open = (
|
||||
env: NodeJS.ProcessEnv = { XPC_SERVICE_NAME: LABEL },
|
||||
platform: NodeJS.Platform = 'darwin'
|
||||
) =>
|
||||
openServeBundlePark({
|
||||
executable,
|
||||
env,
|
||||
platform,
|
||||
userDataRoot: join(root, 'data')
|
||||
})
|
||||
|
||||
async function writeRecord(overrides: Partial<ServeParkRecord> = {}): Promise<void> {
|
||||
const record: ServeParkRecord = {
|
||||
schemaVersion: 1,
|
||||
attemptId: 'attempt-1',
|
||||
installerPid: process.pid,
|
||||
phase: 'park',
|
||||
createdAt: Date.now(),
|
||||
parkDeadlineAt: Date.now() + 60_000,
|
||||
commitDeadlineAt: null,
|
||||
bundleBefore: 'unused',
|
||||
supervisorPids: [process.pid],
|
||||
...overrides
|
||||
}
|
||||
await mkdir(dir, { recursive: true })
|
||||
await writeFile(parkRecordPath(dir), JSON.stringify(record))
|
||||
}
|
||||
|
||||
it('takes part only under an installed launchd label on macOS', () => {
|
||||
expect(open({})).toBeNull()
|
||||
expect(open({ XPC_SERVICE_NAME: 'application.com.other' })).toBeNull()
|
||||
expect(open({ XPC_SERVICE_NAME: LABEL }, 'linux')).toBeNull()
|
||||
expect(open()).not.toBeNull()
|
||||
})
|
||||
|
||||
it('registers with its child pid and removes the registration on close', async () => {
|
||||
const park = open()!
|
||||
park.heartbeat(77)
|
||||
const [name] = await readdir(supervisorsDir(dir))
|
||||
expect(JSON.parse(await readFile(join(supervisorsDir(dir), name), 'utf8'))).toMatchObject({
|
||||
pid: process.pid,
|
||||
childPid: 77,
|
||||
label: LABEL
|
||||
})
|
||||
park.unregister()
|
||||
await expect(readdir(supervisorsDir(dir))).resolves.toEqual([])
|
||||
})
|
||||
|
||||
it('stops the running server for a park, acknowledges after it exits, and never counts it', async () => {
|
||||
const park = open()!
|
||||
const stop = vi.fn()
|
||||
let tick: (() => void) | null = null
|
||||
let runs = 0
|
||||
let signalStop: (() => void) | null = null
|
||||
const result = superviseServeService({
|
||||
userDataPath: join(root, 'profile'),
|
||||
executable,
|
||||
force: false,
|
||||
park,
|
||||
readInstallEvidence: async () => ({ kind: 'clear' }),
|
||||
currentChild: () => ({ pid: 99, stop }),
|
||||
setRepeating: (callback) => {
|
||||
tick = callback
|
||||
return () => {
|
||||
tick = null
|
||||
}
|
||||
},
|
||||
setTimer: (callback) => {
|
||||
const timer = setTimeout(callback, 1)
|
||||
return () => clearTimeout(timer)
|
||||
},
|
||||
onStopSignal: (listener) => {
|
||||
signalStop = listener
|
||||
return () => undefined
|
||||
},
|
||||
log: () => undefined,
|
||||
runOnce: async () => {
|
||||
runs++
|
||||
if (runs > 1) {
|
||||
signalStop?.()
|
||||
return 0
|
||||
}
|
||||
await writeRecord()
|
||||
tick?.()
|
||||
expect(stop).toHaveBeenCalledOnce()
|
||||
await expect(readdir(acksDir(dir))).rejects.toThrow()
|
||||
return 0
|
||||
}
|
||||
})
|
||||
// While parked the loop holds; release it so the next attempt runs.
|
||||
await vi.waitFor(async () => expect(await readdir(acksDir(dir))).toHaveLength(1))
|
||||
await writeRecord({ phase: 'unpark' })
|
||||
await expect(result).resolves.toBe(0)
|
||||
expect(runs).toBe(2)
|
||||
await expect(readdir(join(root, 'profile'))).rejects.toThrow()
|
||||
})
|
||||
|
||||
it('stops a server it is not listed for while a park holds, without counting it', async () => {
|
||||
const park = open()!
|
||||
const watch = createSupervisedParkWatch(park, {
|
||||
currentChild: () => ({ pid: 5, stop }),
|
||||
setRepeating: (callback) => {
|
||||
tick = callback
|
||||
return () => undefined
|
||||
},
|
||||
now: Date.now
|
||||
})
|
||||
const stop = vi.fn()
|
||||
let tick: (() => void) | null = null
|
||||
const attempt = await watch.during(async () => {
|
||||
await writeRecord({ supervisorPids: [process.pid + 1] })
|
||||
tick?.()
|
||||
return 0
|
||||
})
|
||||
expect(stop).toHaveBeenCalledOnce()
|
||||
expect(attempt.parked).toBe(true)
|
||||
await expect(readdir(acksDir(dir))).rejects.toThrow()
|
||||
})
|
||||
|
||||
it('holds while committed until the bundle changes, then releases', async () => {
|
||||
const park = open()!
|
||||
await writeRecord({ phase: 'commit', commitDeadlineAt: Date.now() + 60_000 })
|
||||
const bundleBefore = JSON.parse(await readFile(parkRecordPath(dir), 'utf8')).bundleBefore
|
||||
expect(bundleBefore).toBe('unused')
|
||||
// 'unused' differs from the real identity, which models a bundle ShipIt already replaced.
|
||||
expect(park.holds()).toBe(false)
|
||||
const identity = formatBundleIdentity(
|
||||
await stat(join(root, 'Orca.app', 'Contents', 'Info.plist'))
|
||||
)
|
||||
await writeRecord({
|
||||
phase: 'commit',
|
||||
commitDeadlineAt: Date.now() + 60_000,
|
||||
bundleBefore: identity
|
||||
})
|
||||
expect(park.holds()).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,152 @@
|
||||
import {
|
||||
mkdirSync,
|
||||
readFileSync,
|
||||
realpathSync,
|
||||
renameSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
writeFileSync
|
||||
} from 'node:fs'
|
||||
import { dirname, join } from 'node:path'
|
||||
import {
|
||||
ackFileName,
|
||||
acksDir,
|
||||
bundleCoordinationDir,
|
||||
bundleInfoPlistPath,
|
||||
decideServeParkHold,
|
||||
formatBundleIdentity,
|
||||
isProcessAlive,
|
||||
isServeSupervisorLabel,
|
||||
macCoordinationRoot,
|
||||
parkRecordPath,
|
||||
parseServeParkRecord,
|
||||
supervisorsDir,
|
||||
SERVE_COORDINATION_PROTOCOL,
|
||||
type ServeParkRecord,
|
||||
type ServeSupervisorRegistration
|
||||
} from '../../shared/serve-bundle-coordination'
|
||||
import { homedir } from 'node:os'
|
||||
|
||||
/** The supervisor's half of the desktop-install park protocol. */
|
||||
export type ServeBundlePark = {
|
||||
heartbeat: (childPid: number | null) => void
|
||||
unregister: () => void
|
||||
/** A park the installer is asking for now, or null. */
|
||||
pendingPark: () => { attemptId: string } | null
|
||||
ack: (attemptId: string) => void
|
||||
/** True while the shared record says this bundle must stay stopped. */
|
||||
holds: () => boolean
|
||||
}
|
||||
|
||||
function writeJsonAtomic(path: string, value: unknown): void {
|
||||
mkdirSync(dirname(path), { recursive: true, mode: 0o700 })
|
||||
const temporary = `${path}.${process.pid}.tmp`
|
||||
writeFileSync(temporary, JSON.stringify(value), { mode: 0o600 })
|
||||
renameSync(temporary, path)
|
||||
}
|
||||
|
||||
function readParkRecord(dir: string): ServeParkRecord | null {
|
||||
try {
|
||||
return parseServeParkRecord(JSON.parse(readFileSync(parkRecordPath(dir), 'utf8')))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function readBundleIdentity(bundlePath: string): string {
|
||||
try {
|
||||
return formatBundleIdentity(statSync(bundleInfoPlistPath(bundlePath)))
|
||||
} catch {
|
||||
return formatBundleIdentity(null)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Only a launchd service whose label `orca serve service install` wrote takes part: the installer
|
||||
* stops a server only when a service manager it can name keeps the supervisor running. Everyone
|
||||
* else keeps today's refusal.
|
||||
*/
|
||||
export function openServeBundlePark(input: {
|
||||
executable: string
|
||||
env?: NodeJS.ProcessEnv
|
||||
platform?: NodeJS.Platform
|
||||
userDataRoot?: string
|
||||
now?: () => number
|
||||
}): ServeBundlePark | null {
|
||||
const env = input.env ?? process.env
|
||||
const label = env.XPC_SERVICE_NAME
|
||||
if ((input.platform ?? process.platform) !== 'darwin' || !isServeSupervisorLabel(label)) {
|
||||
return null
|
||||
}
|
||||
const bundle = dirname(dirname(dirname(input.executable)))
|
||||
if (!bundle.endsWith('.app')) {
|
||||
return null
|
||||
}
|
||||
let realBundle: string
|
||||
try {
|
||||
realBundle = realpathSync(bundle)
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
const now = input.now ?? Date.now
|
||||
const dir = bundleCoordinationDir(
|
||||
realBundle,
|
||||
input.userDataRoot ?? macCoordinationRoot(homedir())
|
||||
)
|
||||
const registrationPath = join(supervisorsDir(dir), `${process.pid}.json`)
|
||||
return {
|
||||
heartbeat: (childPid) => {
|
||||
const registration: ServeSupervisorRegistration = {
|
||||
protocol: SERVE_COORDINATION_PROTOCOL,
|
||||
pid: process.pid,
|
||||
childPid,
|
||||
heartbeatAt: now(),
|
||||
label
|
||||
}
|
||||
try {
|
||||
writeJsonAtomic(registrationPath, registration)
|
||||
} catch {
|
||||
// Unregistered is safe: the installer then refuses, exactly as before this protocol.
|
||||
}
|
||||
},
|
||||
unregister: () => {
|
||||
try {
|
||||
rmSync(registrationPath, { force: true })
|
||||
} catch {
|
||||
// A leftover registration goes stale on its own heartbeat.
|
||||
}
|
||||
},
|
||||
pendingPark: () => {
|
||||
const record = readParkRecord(dir)
|
||||
return record?.phase === 'park' &&
|
||||
record.supervisorPids.includes(process.pid) &&
|
||||
decideServeParkHold(record, {
|
||||
now: now(),
|
||||
installerAlive: isProcessAlive(record.installerPid),
|
||||
bundleNow: record.bundleBefore
|
||||
}) === 'hold'
|
||||
? { attemptId: record.attemptId }
|
||||
: null
|
||||
},
|
||||
ack: (attemptId) => {
|
||||
try {
|
||||
writeJsonAtomic(join(acksDir(dir), ackFileName(attemptId, process.pid)), {
|
||||
pid: process.pid,
|
||||
at: now()
|
||||
})
|
||||
} catch {
|
||||
// No ack means the installer gives up and the park ends; never a stuck service.
|
||||
}
|
||||
},
|
||||
holds: () => {
|
||||
const record = readParkRecord(dir)
|
||||
return (
|
||||
decideServeParkHold(record, {
|
||||
now: now(),
|
||||
installerAlive: record ? isProcessAlive(record.installerPid) : false,
|
||||
bundleNow: readBundleIdentity(realBundle)
|
||||
}) === 'hold'
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -4,7 +4,10 @@ import { runProcess } from '@orca/process-host'
|
||||
|
||||
export type ServeInstallEvidence =
|
||||
| { kind: 'clear' }
|
||||
| { kind: 'hold'; reason: 'bundle-missing' | 'installer-running' | 'installer-unknown' }
|
||||
| {
|
||||
kind: 'hold'
|
||||
reason: 'bundle-missing' | 'installer-running' | 'installer-unknown' | 'install-park'
|
||||
}
|
||||
|
||||
export const SERVE_INSTALL_HOLD_REASON_TEXT: Record<
|
||||
Extract<ServeInstallEvidence, { kind: 'hold' }>['reason'],
|
||||
@@ -12,7 +15,9 @@ export const SERVE_INSTALL_HOLD_REASON_TEXT: Record<
|
||||
> = {
|
||||
'bundle-missing': 'the Orca executable is missing, likely mid-install',
|
||||
'installer-running': "macOS's updater (ShipIt) is replacing this Orca app",
|
||||
'installer-unknown': "could not check whether macOS's updater (ShipIt) is replacing this Orca app"
|
||||
'installer-unknown':
|
||||
"could not check whether macOS's updater (ShipIt) is replacing this Orca app",
|
||||
'install-park': 'the Orca desktop app paused this server to install an update'
|
||||
}
|
||||
|
||||
type ListProcesses = () => Promise<string | null>
|
||||
|
||||
@@ -12,6 +12,7 @@ import {
|
||||
} from '../../shared/orcad-local-serve-selection'
|
||||
import { waitForRecipeJson } from './serve-recipe-json'
|
||||
import { superviseForegroundServe } from './serve-update-supervisor'
|
||||
import { isSupervisedSpawnHeld } from './serve-supervised-stop'
|
||||
|
||||
type SupervisorArgs = Parameters<typeof superviseForegroundServe>[0]
|
||||
|
||||
@@ -109,6 +110,9 @@ export function serveWithOrcad(
|
||||
ORCA_VERSION: selection.version
|
||||
}
|
||||
}
|
||||
if (isSupervisedSpawnHeld()) {
|
||||
return Promise.resolve(0)
|
||||
}
|
||||
const child = spawnChild(selection.runtime, childArgs, spawnOptions)
|
||||
if (args.recipeJson) {
|
||||
return waitForRecipeJson(child)
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
import { SERVE_SUPERVISOR_HEARTBEAT_MS } from '../../shared/serve-bundle-coordination'
|
||||
import type { ServeBundlePark } from './serve-bundle-park'
|
||||
|
||||
export const SERVE_PARK_POLL_MS = 1_000
|
||||
|
||||
type ServeChild = { pid: number | null; stop: () => void } | null
|
||||
|
||||
export type SupervisedParkWatch = {
|
||||
/** Between attempts: no server runs, so a pending park is acknowledged at once. True = hold. */
|
||||
idle: () => boolean
|
||||
/** Runs one attempt; stops its server when the desktop asks to park, and acknowledges after. */
|
||||
during: (
|
||||
run: () => Promise<number>
|
||||
) => Promise<{ parked: boolean; settled: PromiseSettledResult<number> }>
|
||||
close: () => void
|
||||
}
|
||||
|
||||
export function createSupervisedParkWatch(
|
||||
park: ServeBundlePark,
|
||||
deps: {
|
||||
currentChild: () => ServeChild
|
||||
setRepeating: (callback: () => void, ms: number) => () => void
|
||||
now: () => number
|
||||
}
|
||||
): SupervisedParkWatch {
|
||||
const acked = new Set<string>()
|
||||
let lastBeat: { pid: number | null | undefined; at: number } = { pid: undefined, at: 0 }
|
||||
const beat = (pid: number | null): void => {
|
||||
if (pid !== lastBeat.pid || deps.now() - lastBeat.at >= SERVE_SUPERVISOR_HEARTBEAT_MS) {
|
||||
park.heartbeat(pid)
|
||||
lastBeat = { pid, at: deps.now() }
|
||||
}
|
||||
}
|
||||
const ack = (attemptId: string): void => {
|
||||
if (!acked.has(attemptId)) {
|
||||
acked.add(attemptId)
|
||||
park.ack(attemptId)
|
||||
}
|
||||
}
|
||||
return {
|
||||
idle: () => {
|
||||
beat(null)
|
||||
const pending = park.pendingPark()
|
||||
if (pending) {
|
||||
ack(pending.attemptId)
|
||||
}
|
||||
return park.holds()
|
||||
},
|
||||
during: async (run) => {
|
||||
let parking: string | null = null
|
||||
let holding = false
|
||||
let stoppedPid: number | null = null
|
||||
const tick = (): void => {
|
||||
const child = deps.currentChild()
|
||||
beat(child?.pid ?? null)
|
||||
if (parking === null) {
|
||||
const pending = park.pendingPark()
|
||||
parking = pending && !acked.has(pending.attemptId) ? pending.attemptId : null
|
||||
}
|
||||
// Why holds() too: a server another profile's supervisor started mid-install is not listed.
|
||||
holding = holding || park.holds()
|
||||
// Why per pid: a park can land between preflight and spawn, before any child exists.
|
||||
if ((parking !== null || holding) && child && child.pid !== stoppedPid) {
|
||||
stoppedPid = child.pid
|
||||
child.stop()
|
||||
}
|
||||
}
|
||||
const cancel = deps.setRepeating(tick, SERVE_PARK_POLL_MS)
|
||||
let settled: PromiseSettledResult<number>
|
||||
try {
|
||||
settled = { status: 'fulfilled', value: await run() }
|
||||
} catch (reason) {
|
||||
settled = { status: 'rejected', reason }
|
||||
} finally {
|
||||
cancel()
|
||||
}
|
||||
if (parking !== null) {
|
||||
ack(parking)
|
||||
return { parked: true, settled }
|
||||
}
|
||||
// A first spawn refused by the hold, or a server stopped for it, is not a failure either.
|
||||
return { parked: holding || park.holds(), settled }
|
||||
},
|
||||
close: () => park.unregister()
|
||||
}
|
||||
}
|
||||
@@ -14,7 +14,13 @@ import {
|
||||
SERVE_INSTALL_HOLD_REASON_TEXT,
|
||||
type ServeInstallEvidence
|
||||
} from './serve-install-hold'
|
||||
import { requestSupervisedServeStop, resetSupervisedServeStop } from './serve-supervised-stop'
|
||||
import {
|
||||
currentServeChild,
|
||||
requestSupervisedServeStop,
|
||||
resetSupervisedServeStop
|
||||
} from './serve-supervised-stop'
|
||||
import type { ServeBundlePark } from './serve-bundle-park'
|
||||
import { createSupervisedParkWatch } from './serve-supervised-park-watch'
|
||||
|
||||
export const SERVE_INSTALL_HOLD_POLL_MS = 2_000
|
||||
/** A hold this long is reported as stuck; it still never times out into a spawn. */
|
||||
@@ -42,6 +48,10 @@ export type SupervisedServeOptions = {
|
||||
setTimer?: (callback: () => void, ms: number) => () => void
|
||||
onStopSignal?: (listener: StopListener) => () => void
|
||||
log?: (line: string) => void
|
||||
/** The desktop-install park protocol; null when this supervisor does not take part. */
|
||||
park?: ServeBundlePark | null
|
||||
currentChild?: typeof currentServeChild
|
||||
setRepeating?: (callback: () => void, ms: number) => () => void
|
||||
}
|
||||
|
||||
export function getServeSuperviseValidationError(input: {
|
||||
@@ -79,6 +89,19 @@ function defaultSetTimer(callback: () => void, ms: number): () => void {
|
||||
return () => clearTimeout(timer)
|
||||
}
|
||||
|
||||
function defaultSetRepeating(callback: () => void, ms: number): () => void {
|
||||
const timer = setInterval(callback, ms)
|
||||
return () => clearInterval(timer)
|
||||
}
|
||||
|
||||
async function settle(run: () => Promise<number>): Promise<PromiseSettledResult<number>> {
|
||||
try {
|
||||
return { status: 'fulfilled', value: await run() }
|
||||
} catch (reason) {
|
||||
return { status: 'rejected', reason }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* `orca serve --supervise`: the restart owner for a service manager. It never exits because its
|
||||
* child died; it exits on a stop signal, or with 0 after giving up so the manager stops too.
|
||||
@@ -112,6 +135,13 @@ export async function superviseServeService(options: SupervisedServeOptions): Pr
|
||||
return refuseGaveUp(record.gaveUp)
|
||||
}
|
||||
|
||||
const parkWatch = options.park
|
||||
? createSupervisedParkWatch(options.park, {
|
||||
currentChild: options.currentChild ?? currentServeChild,
|
||||
setRepeating: options.setRepeating ?? defaultSetRepeating,
|
||||
now
|
||||
})
|
||||
: null
|
||||
let stopped = false
|
||||
let wake: (() => void) | null = null
|
||||
const removeStopListener = (options.onStopSignal ?? defaultOnStopSignal)(() => {
|
||||
@@ -139,7 +169,9 @@ export async function superviseServeService(options: SupervisedServeOptions): Pr
|
||||
const awaitClearToSpawn = async (): Promise<boolean> => {
|
||||
const since = now()
|
||||
while (!stopped) {
|
||||
const evidence = await readEvidence(options.executable)
|
||||
const evidence: ServeInstallEvidence = parkWatch?.idle()
|
||||
? { kind: 'hold', reason: 'install-park' }
|
||||
: await readEvidence(options.executable)
|
||||
if (evidence.kind === 'clear') {
|
||||
return !stopped
|
||||
}
|
||||
@@ -154,7 +186,7 @@ export async function superviseServeService(options: SupervisedServeOptions): Pr
|
||||
}
|
||||
return false
|
||||
}
|
||||
resetSupervisedServeStop(awaitClearToSpawn)
|
||||
resetSupervisedServeStop(awaitClearToSpawn, () => options.park?.holds() ?? false)
|
||||
|
||||
try {
|
||||
let standbyAttempt = 0
|
||||
@@ -162,7 +194,9 @@ export async function superviseServeService(options: SupervisedServeOptions): Pr
|
||||
current: null
|
||||
}
|
||||
while (!stopped) {
|
||||
const evidence = await readEvidence(options.executable)
|
||||
const evidence: ServeInstallEvidence = parkWatch?.idle()
|
||||
? { kind: 'hold', reason: 'install-park' }
|
||||
: await readEvidence(options.executable)
|
||||
if (stopped) {
|
||||
break
|
||||
}
|
||||
@@ -197,12 +231,19 @@ export async function superviseServeService(options: SupervisedServeOptions): Pr
|
||||
return refuseGaveUp(record.gaveUp)
|
||||
}
|
||||
|
||||
const attempt = parkWatch
|
||||
? await parkWatch.during(options.runOnce)
|
||||
: { parked: false, settled: await settle(options.runOnce) }
|
||||
if (stopped) {
|
||||
break
|
||||
}
|
||||
if (attempt.parked) {
|
||||
// Why not a failure: the desktop stopped this server to install an update.
|
||||
continue
|
||||
}
|
||||
let failure: string
|
||||
try {
|
||||
const code = await options.runOnce()
|
||||
if (stopped) {
|
||||
break
|
||||
}
|
||||
if (attempt.settled.status === 'fulfilled') {
|
||||
const code = attempt.settled.value
|
||||
if (code === SINGLE_INSTANCE_ALREADY_RUNNING_EXIT_CODE) {
|
||||
// Why not a failure: another Orca serves this profile, so wait for it instead of racing it.
|
||||
const delayMs = standbyDelayMs(standbyAttempt++, random)
|
||||
@@ -213,10 +254,8 @@ export async function superviseServeService(options: SupervisedServeOptions): Pr
|
||||
continue
|
||||
}
|
||||
failure = `exited with code ${code}`
|
||||
} catch (error) {
|
||||
if (stopped) {
|
||||
break
|
||||
}
|
||||
} else {
|
||||
const error: unknown = attempt.settled.reason
|
||||
failure = error instanceof Error ? error.message : String(error)
|
||||
}
|
||||
standbyAttempt = 0
|
||||
@@ -236,5 +275,6 @@ export async function superviseServeService(options: SupervisedServeOptions): Pr
|
||||
} finally {
|
||||
removeStopListener()
|
||||
resetSupervisedServeStop()
|
||||
parkWatch?.close()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
let supervisedStopRequested = false
|
||||
let supervisedStop = new AbortController()
|
||||
let supervisedSpawnGate: (() => Promise<boolean>) | null = null
|
||||
let supervisedSpawnHeld: (() => boolean) | null = null
|
||||
|
||||
/** A stop that reached `--supervise` before its child existed must still reach that child. */
|
||||
export function requestSupervisedServeStop(): void {
|
||||
@@ -10,10 +11,19 @@ export function requestSupervisedServeStop(): void {
|
||||
}
|
||||
|
||||
/** Starts a supervised run: `gate` resolves true when a replacement may spawn, false on stop. */
|
||||
export function resetSupervisedServeStop(gate: (() => Promise<boolean>) | null = null): void {
|
||||
export function resetSupervisedServeStop(
|
||||
gate: (() => Promise<boolean>) | null = null,
|
||||
held: (() => boolean) | null = null
|
||||
): void {
|
||||
supervisedStopRequested = false
|
||||
supervisedStop = new AbortController()
|
||||
supervisedSpawnGate = gate
|
||||
supervisedSpawnHeld = held
|
||||
}
|
||||
|
||||
/** Checked synchronously right before a first spawn: a desktop install may have begun since. */
|
||||
export function isSupervisedSpawnHeld(): boolean {
|
||||
return supervisedStopRequested || (supervisedSpawnHeld?.() ?? false)
|
||||
}
|
||||
|
||||
/** Why: update replacements spawn inside this loop, so the install hold must gate them too. */
|
||||
@@ -24,10 +34,28 @@ export async function mayRespawn(): Promise<boolean> {
|
||||
return supervisedSpawnGate ? supervisedSpawnGate() : true
|
||||
}
|
||||
|
||||
export function isSupervisedStopRequested(): boolean {
|
||||
return supervisedStopRequested
|
||||
}
|
||||
|
||||
export function supervisedStopSignal(): AbortSignal {
|
||||
return supervisedStop.signal
|
||||
}
|
||||
|
||||
type ServeChild = { pid: number | null; stop: () => void }
|
||||
let currentChild: ServeChild | null = null
|
||||
|
||||
/** The child a supervised serve is waiting on right now, so a desktop install can park it. */
|
||||
export function currentServeChild(): ServeChild | null {
|
||||
return currentChild
|
||||
}
|
||||
|
||||
/** A stop that reached `--supervise` before this child existed is passed on to it now. */
|
||||
export function trackServeChild(child: ServeChild): void {
|
||||
currentChild = child
|
||||
if (supervisedStopRequested) {
|
||||
child.stop()
|
||||
}
|
||||
}
|
||||
|
||||
export function untrackServeChild(child: ServeChild): void {
|
||||
if (currentChild === child) {
|
||||
currentChild = null
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,9 +13,10 @@ import {
|
||||
import { serveSignalExitError } from './serve-signal-exit-diagnostic'
|
||||
import { waitForMacBundleVersion } from './mac-app-update-bundle'
|
||||
import {
|
||||
isSupervisedStopRequested,
|
||||
mayRespawn,
|
||||
supervisedStopSignal
|
||||
supervisedStopSignal,
|
||||
trackServeChild,
|
||||
untrackServeChild
|
||||
} from './serve-supervised-stop'
|
||||
import {
|
||||
decideAfterServeChildExit,
|
||||
@@ -184,6 +185,7 @@ function waitForForegroundChild(
|
||||
}
|
||||
forceKillTimer ??= setTimeout(() => child.kill('SIGKILL'), SERVE_CHILD_FORCE_KILL_GRACE_MS)
|
||||
}
|
||||
const tracked = { pid: child.pid ?? null, stop: () => forwardSignal('SIGTERM') }
|
||||
const handleMessage = (value: unknown): void => {
|
||||
const message = parseServeSupervisorMessage(value)
|
||||
if (!message || !expected || readiness !== 'pending') {
|
||||
@@ -211,6 +213,7 @@ function waitForForegroundChild(
|
||||
})
|
||||
}
|
||||
const cleanup = (): void => {
|
||||
untrackServeChild(tracked)
|
||||
process.off('SIGINT', forwardSignal)
|
||||
process.off('SIGTERM', forwardSignal)
|
||||
if (forwardsHangup) {
|
||||
@@ -234,9 +237,7 @@ function waitForForegroundChild(
|
||||
if (typeof child.on === 'function') {
|
||||
child.on('message', handleMessage)
|
||||
}
|
||||
if (isSupervisedStopRequested()) {
|
||||
forwardSignal('SIGTERM')
|
||||
}
|
||||
trackServeChild(tracked)
|
||||
if (expected) {
|
||||
readyTimer = setTimeout(() => {
|
||||
rejectReplacement(
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type * as ServeSupervisorInstallPark from './updater/serve-supervisor-install-park'
|
||||
import { loadUpdaterModule, warmUpdaterModule } from './updater-test-module-loader'
|
||||
|
||||
const {
|
||||
@@ -127,6 +128,18 @@ vi.mock('./macos-update-running-instances', () => ({
|
||||
getMacUpdateRunningInstances: getMacUpdateRunningInstancesMock
|
||||
}))
|
||||
|
||||
// Why: no registered serve supervisors here, so every blocker is refused as before parking existed.
|
||||
vi.mock('./updater/serve-supervisor-install-park', async (importOriginal) => ({
|
||||
...(await importOriginal<typeof ServeSupervisorInstallPark>()),
|
||||
parkServeSupervisorsForInstall: (_executable: string, blockers: number[]) => ({
|
||||
kind: 'refused',
|
||||
reason: 'unregistered',
|
||||
pids: blockers
|
||||
}),
|
||||
commitServeSupervisorPark: async () => undefined,
|
||||
releaseServeSupervisorPark: async () => undefined
|
||||
}))
|
||||
|
||||
warmUpdaterModule()
|
||||
|
||||
async function prepareStagedMacUpdate(downloadUpdate: () => void): Promise<void> {
|
||||
|
||||
@@ -0,0 +1,230 @@
|
||||
import { mkdir, mkdtemp, readFile, readdir, rm, stat, utimes, writeFile } from 'node:fs/promises'
|
||||
import { realpathSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
import {
|
||||
ackFileName,
|
||||
acksDir,
|
||||
bundleCoordinationDir,
|
||||
decideServeParkHold,
|
||||
formatBundleIdentity,
|
||||
parkRecordPath,
|
||||
parseServeParkRecord,
|
||||
supervisorsDir,
|
||||
type ServeParkRecord
|
||||
} from '../../shared/serve-bundle-coordination'
|
||||
import {
|
||||
commitServeSupervisorPark,
|
||||
parkServeSupervisorsForInstall,
|
||||
releaseServeSupervisorPark
|
||||
} from './serve-supervisor-install-park'
|
||||
|
||||
const CHILD = 424242
|
||||
|
||||
describe('parkServeSupervisorsForInstall', () => {
|
||||
let root: string
|
||||
let executable: string
|
||||
let dir: string
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-install-park-'))
|
||||
const bundle = join(root, 'Orca.app')
|
||||
executable = join(bundle, 'Contents', 'MacOS', 'Orca')
|
||||
await mkdir(join(bundle, 'Contents', 'MacOS'), { recursive: true })
|
||||
await writeFile(join(bundle, 'Contents', 'Info.plist'), 'v1')
|
||||
dir = bundleCoordinationDir(realpathSync(bundle), join(root, 'data'))
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await releaseServeSupervisorPark()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function register(childPid: number, heartbeatAt = Date.now()): Promise<void> {
|
||||
await mkdir(supervisorsDir(dir), { recursive: true })
|
||||
await writeFile(
|
||||
join(supervisorsDir(dir), `${process.pid}.json`),
|
||||
JSON.stringify({
|
||||
protocol: 1,
|
||||
pid: process.pid,
|
||||
childPid,
|
||||
heartbeatAt,
|
||||
label: 'com.stablyai.orca.serve.default'
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
async function readRecord(): Promise<ServeParkRecord | null> {
|
||||
return parseServeParkRecord(JSON.parse(await readFile(parkRecordPath(dir), 'utf8')))
|
||||
}
|
||||
|
||||
const deps = (overrides: Partial<Parameters<typeof parkServeSupervisorsForInstall>[2]> = {}) => ({
|
||||
macUserDataRoot: join(root, 'data'),
|
||||
listBlockers: async () => [],
|
||||
sleep: async () => undefined,
|
||||
...overrides
|
||||
})
|
||||
|
||||
it('refuses as today when any blocker is not a registered supervisor child', async () => {
|
||||
await register(CHILD)
|
||||
await expect(parkServeSupervisorsForInstall(executable, [CHILD, 7], deps())).resolves.toEqual({
|
||||
kind: 'refused',
|
||||
reason: 'unregistered',
|
||||
pids: [7]
|
||||
})
|
||||
// A stale registration counts as unregistered.
|
||||
await register(CHILD, Date.now() - 60_000)
|
||||
await expect(parkServeSupervisorsForInstall(executable, [CHILD], deps())).resolves.toEqual({
|
||||
kind: 'refused',
|
||||
reason: 'unregistered',
|
||||
pids: [CHILD]
|
||||
})
|
||||
await expect(readdir(dir)).resolves.not.toContain('park.json')
|
||||
})
|
||||
|
||||
it('parks, commits, then lets supervisors wait for the bundle to change', async () => {
|
||||
await register(CHILD)
|
||||
const result = await parkServeSupervisorsForInstall(
|
||||
executable,
|
||||
[CHILD],
|
||||
deps({
|
||||
sleep: async () => {
|
||||
const record = await readRecord()
|
||||
await mkdir(acksDir(dir), { recursive: true })
|
||||
await writeFile(join(acksDir(dir), ackFileName(record!.attemptId, process.pid)), '{}')
|
||||
}
|
||||
})
|
||||
)
|
||||
expect(result).toEqual({ kind: 'parked' })
|
||||
expect((await readRecord())?.phase).toBe('park')
|
||||
await commitServeSupervisorPark(async () => [])
|
||||
const committed = await readRecord()
|
||||
expect(committed?.phase).toBe('commit')
|
||||
|
||||
const facts = { now: Date.now(), installerAlive: false, bundleNow: committed!.bundleBefore }
|
||||
expect(decideServeParkHold(committed, facts)).toBe('hold')
|
||||
expect(decideServeParkHold(committed, { ...facts, bundleNow: 'new' })).toBe('release')
|
||||
expect(
|
||||
decideServeParkHold(committed, { ...facts, now: committed!.commitDeadlineAt! + 1 })
|
||||
).toBe('release')
|
||||
})
|
||||
|
||||
it('releases and names the supervisors that never acknowledged', async () => {
|
||||
await register(CHILD)
|
||||
let clock = 0
|
||||
const result = await parkServeSupervisorsForInstall(
|
||||
executable,
|
||||
[CHILD],
|
||||
deps({
|
||||
now: () => clock,
|
||||
sleep: async () => {
|
||||
clock += 10_000
|
||||
}
|
||||
})
|
||||
)
|
||||
expect(result).toEqual({ kind: 'refused', reason: 'unresponsive', pids: [process.pid] })
|
||||
expect((await readRecord())?.phase).toBe('unpark')
|
||||
})
|
||||
|
||||
it('refuses when a still-blocking server remains after every acknowledgement', async () => {
|
||||
await register(CHILD)
|
||||
const result = await parkServeSupervisorsForInstall(
|
||||
executable,
|
||||
[CHILD],
|
||||
deps({
|
||||
listBlockers: async () => [CHILD],
|
||||
ackWaitMs: 50,
|
||||
sleep: async () => {
|
||||
const record = await readRecord()
|
||||
await mkdir(acksDir(dir), { recursive: true })
|
||||
await writeFile(join(acksDir(dir), ackFileName(record!.attemptId, process.pid)), '{}')
|
||||
await new Promise((r) => setTimeout(r, 20))
|
||||
}
|
||||
})
|
||||
)
|
||||
expect(result).toMatchObject({ kind: 'refused', reason: 'unresponsive' })
|
||||
expect((await readRecord())?.phase).toBe('unpark')
|
||||
})
|
||||
|
||||
it('fails the commit when the park already lapsed, so supervisors are never raced', async () => {
|
||||
await register(CHILD)
|
||||
await parkServeSupervisorsForInstall(
|
||||
executable,
|
||||
[CHILD],
|
||||
deps({
|
||||
sleep: async () => {
|
||||
const record = await readRecord()
|
||||
await mkdir(acksDir(dir), { recursive: true })
|
||||
await writeFile(join(acksDir(dir), ackFileName(record!.attemptId, process.pid)), '{}')
|
||||
}
|
||||
})
|
||||
)
|
||||
await expect(
|
||||
commitServeSupervisorPark(
|
||||
async () => [],
|
||||
() => Date.now() + 10 * 60_000
|
||||
)
|
||||
).rejects.toThrow()
|
||||
expect((await readRecord())?.phase).toBe('unpark')
|
||||
})
|
||||
|
||||
it('fails the commit when a server is running again', async () => {
|
||||
await register(CHILD)
|
||||
await parkServeSupervisorsForInstall(
|
||||
executable,
|
||||
[CHILD],
|
||||
deps({
|
||||
sleep: async () => {
|
||||
const record = await readRecord()
|
||||
await mkdir(acksDir(dir), { recursive: true })
|
||||
await writeFile(join(acksDir(dir), ackFileName(record!.attemptId, process.pid)), '{}')
|
||||
}
|
||||
})
|
||||
)
|
||||
await expect(commitServeSupervisorPark(async () => [CHILD + 1])).rejects.toThrow()
|
||||
expect((await readRecord())?.phase).toBe('unpark')
|
||||
})
|
||||
|
||||
it('a same-version reinstall still changes the bundle identity', async () => {
|
||||
const plist = join(root, 'Orca.app', 'Contents', 'Info.plist')
|
||||
await register(CHILD)
|
||||
await parkServeSupervisorsForInstall(
|
||||
executable,
|
||||
[CHILD],
|
||||
deps({
|
||||
sleep: async () => {
|
||||
const record = await readRecord()
|
||||
await mkdir(acksDir(dir), { recursive: true })
|
||||
await writeFile(join(acksDir(dir), ackFileName(record!.attemptId, process.pid)), '{}')
|
||||
}
|
||||
})
|
||||
)
|
||||
const before = (await readRecord())!.bundleBefore
|
||||
await utimes(plist, new Date(), new Date(Date.now() + 5_000))
|
||||
expect(formatBundleIdentity(await stat(plist))).not.toBe(before)
|
||||
})
|
||||
})
|
||||
|
||||
describe('decideServeParkHold', () => {
|
||||
const record: ServeParkRecord = {
|
||||
schemaVersion: 1,
|
||||
attemptId: 'a',
|
||||
installerPid: 1,
|
||||
phase: 'park',
|
||||
createdAt: 0,
|
||||
parkDeadlineAt: 100,
|
||||
commitDeadlineAt: null,
|
||||
bundleBefore: 'b',
|
||||
supervisorPids: [2]
|
||||
}
|
||||
|
||||
it('ends a park on unpark, deadline, or installer death', () => {
|
||||
const facts = { now: 50, installerAlive: true, bundleNow: 'b' }
|
||||
expect(decideServeParkHold(record, facts)).toBe('hold')
|
||||
expect(decideServeParkHold({ ...record, phase: 'unpark' }, facts)).toBe('release')
|
||||
expect(decideServeParkHold(record, { ...facts, now: 100 })).toBe('release')
|
||||
expect(decideServeParkHold(record, { ...facts, installerAlive: false })).toBe('release')
|
||||
expect(decideServeParkHold(null, facts)).toBe('release')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,230 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { mkdir, open, readdir, readFile, realpath, rename, stat } from 'node:fs/promises'
|
||||
import { dirname, join } from 'node:path'
|
||||
import {
|
||||
ackFileName,
|
||||
acksDir,
|
||||
bundleCoordinationDir,
|
||||
bundleInfoPlistPath,
|
||||
formatBundleIdentity,
|
||||
isLiveRegistration,
|
||||
isProcessAlive,
|
||||
parkRecordPath,
|
||||
parseServeParkRecord,
|
||||
parseServeSupervisorRegistration,
|
||||
supervisorsDir,
|
||||
SERVE_COMMIT_DEADLINE_MS,
|
||||
SERVE_PARK_ACK_WAIT_MS,
|
||||
SERVE_PARK_DEADLINE_MS,
|
||||
type ServeParkRecord,
|
||||
type ServeSupervisorRegistration
|
||||
} from '../../shared/serve-bundle-coordination'
|
||||
|
||||
export type ServeInstallParkResult =
|
||||
| { kind: 'parked' }
|
||||
/** `unregistered` keeps today's refusal; the others name supervisors that did not cooperate. */
|
||||
| { kind: 'refused'; reason: 'unregistered' | 'unresponsive' | 'busy'; pids: number[] }
|
||||
|
||||
type ParkDeps = {
|
||||
macUserDataRoot: string
|
||||
listBlockers: () => Promise<number[]>
|
||||
now?: () => number
|
||||
sleep?: (ms: number) => Promise<void>
|
||||
ackWaitMs?: number
|
||||
}
|
||||
|
||||
const ACK_POLL_MS = 500
|
||||
|
||||
/** The attempt this process parked, so commit and release find it without a second lookup. */
|
||||
let activePark: { dir: string; record: ServeParkRecord } | null = null
|
||||
|
||||
async function writeRecordDurably(dir: string, record: ServeParkRecord): Promise<void> {
|
||||
const target = parkRecordPath(dir)
|
||||
await mkdir(dirname(target), { recursive: true, mode: 0o700 })
|
||||
const temporary = `${target}.${process.pid}.tmp`
|
||||
const handle = await open(temporary, 'w', 0o600)
|
||||
try {
|
||||
await handle.writeFile(JSON.stringify(record))
|
||||
// Why fsync: supervisors trust this record after the installer is gone, even across a panic.
|
||||
await handle.sync()
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
await rename(temporary, target)
|
||||
}
|
||||
|
||||
async function readJson(path: string): Promise<unknown> {
|
||||
try {
|
||||
return JSON.parse(await readFile(path, 'utf8'))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
async function liveRegistrations(dir: string, now: number): Promise<ServeSupervisorRegistration[]> {
|
||||
let names: string[]
|
||||
try {
|
||||
names = await readdir(supervisorsDir(dir))
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
const registrations = await Promise.all(
|
||||
names
|
||||
.filter((name) => name.endsWith('.json'))
|
||||
.map(async (name) =>
|
||||
parseServeSupervisorRegistration(await readJson(join(supervisorsDir(dir), name)))
|
||||
)
|
||||
)
|
||||
return registrations.filter(
|
||||
(entry): entry is ServeSupervisorRegistration =>
|
||||
entry !== null && isLiveRegistration(entry, now) && isProcessAlive(entry.pid)
|
||||
)
|
||||
}
|
||||
|
||||
async function readBundleIdentity(bundle: string): Promise<string> {
|
||||
try {
|
||||
return formatBundleIdentity(await stat(bundleInfoPlistPath(bundle)))
|
||||
} catch {
|
||||
return formatBundleIdentity(null)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Before a desktop install: ask registered `orca serve --supervise` services to stop their server
|
||||
* and hold. Any blocker that is not such a server is refused exactly as before.
|
||||
*/
|
||||
export async function parkServeSupervisorsForInstall(
|
||||
executable: string,
|
||||
blockers: number[],
|
||||
deps: ParkDeps
|
||||
): Promise<ServeInstallParkResult> {
|
||||
const now = deps.now ?? Date.now
|
||||
const sleep = deps.sleep ?? ((ms: number) => new Promise((r) => setTimeout(r, ms)))
|
||||
let bundle: string
|
||||
try {
|
||||
bundle = await realpath(dirname(dirname(dirname(executable))))
|
||||
} catch {
|
||||
return { kind: 'refused', reason: 'unregistered', pids: blockers }
|
||||
}
|
||||
const dir = bundleCoordinationDir(bundle, deps.macUserDataRoot)
|
||||
const byChild = new Map<number, ServeSupervisorRegistration>()
|
||||
for (const registration of await liveRegistrations(dir, now())) {
|
||||
if (registration.childPid !== null) {
|
||||
byChild.set(registration.childPid, registration)
|
||||
}
|
||||
}
|
||||
const unregistered = blockers.filter((pid) => !byChild.has(pid))
|
||||
if (unregistered.length > 0) {
|
||||
return { kind: 'refused', reason: 'unregistered', pids: unregistered }
|
||||
}
|
||||
const existing = parseServeParkRecord(await readJson(parkRecordPath(dir)))
|
||||
if (
|
||||
existing &&
|
||||
existing.phase === 'park' &&
|
||||
existing.installerPid !== process.pid &&
|
||||
isProcessAlive(existing.installerPid) &&
|
||||
now() < existing.parkDeadlineAt
|
||||
) {
|
||||
return { kind: 'refused', reason: 'busy', pids: [existing.installerPid] }
|
||||
}
|
||||
const supervisorPids = [...new Set(blockers.flatMap((pid) => byChild.get(pid)?.pid ?? []))]
|
||||
const createdAt = now()
|
||||
const record: ServeParkRecord = {
|
||||
schemaVersion: 1,
|
||||
attemptId: randomUUID(),
|
||||
installerPid: process.pid,
|
||||
phase: 'park',
|
||||
createdAt,
|
||||
parkDeadlineAt: createdAt + SERVE_PARK_DEADLINE_MS,
|
||||
commitDeadlineAt: null,
|
||||
bundleBefore: await readBundleIdentity(bundle),
|
||||
supervisorPids
|
||||
}
|
||||
await writeRecordDurably(dir, record)
|
||||
activePark = { dir, record }
|
||||
|
||||
const deadline = createdAt + (deps.ackWaitMs ?? SERVE_PARK_ACK_WAIT_MS)
|
||||
let missing = supervisorPids
|
||||
while (now() < deadline) {
|
||||
let acked: string[]
|
||||
try {
|
||||
acked = await readdir(acksDir(dir))
|
||||
} catch {
|
||||
acked = []
|
||||
}
|
||||
missing = supervisorPids.filter(
|
||||
(pid) =>
|
||||
!acked.includes(ackFileName(record.attemptId, pid)) &&
|
||||
// A supervisor that died while parking no longer blocks anything.
|
||||
isProcessAlive(pid)
|
||||
)
|
||||
if (missing.length === 0) {
|
||||
const remaining = await deps.listBlockers().catch(() => blockers)
|
||||
if (remaining.length === 0) {
|
||||
return { kind: 'parked' }
|
||||
}
|
||||
missing = remaining
|
||||
}
|
||||
await sleep(ACK_POLL_MS)
|
||||
}
|
||||
await releaseServeSupervisorPark()
|
||||
return { kind: 'refused', reason: 'unresponsive', pids: missing }
|
||||
}
|
||||
|
||||
/** Margin so a supervisor polling at the park deadline cannot resume while commit is written. */
|
||||
const COMMIT_DEADLINE_MARGIN_MS = 5_000
|
||||
|
||||
/** Right before the native install: from here supervisors wait for the bundle to change. */
|
||||
export async function commitServeSupervisorPark(
|
||||
listBlockers: () => Promise<number[]>,
|
||||
now: () => number = Date.now
|
||||
): Promise<void> {
|
||||
if (!activePark) {
|
||||
return
|
||||
}
|
||||
// Why re-list: a server another supervisor started after the acks would block ShipIt.
|
||||
const blockers = await listBlockers().catch(() => null)
|
||||
if (blockers === null || blockers.length > 0) {
|
||||
await releaseServeSupervisorPark()
|
||||
throw new Error('An Orca server started again before the install began.')
|
||||
}
|
||||
if (now() + COMMIT_DEADLINE_MARGIN_MS >= activePark.record.parkDeadlineAt) {
|
||||
// Why throw: supervisors may already have restarted servers that would block ShipIt.
|
||||
await releaseServeSupervisorPark()
|
||||
throw new Error('Paused Orca serve services resumed before the install began.')
|
||||
}
|
||||
const record: ServeParkRecord = {
|
||||
...activePark.record,
|
||||
phase: 'commit',
|
||||
commitDeadlineAt: now() + SERVE_COMMIT_DEADLINE_MS
|
||||
}
|
||||
await writeRecordDurably(activePark.dir, record)
|
||||
activePark = { ...activePark, record }
|
||||
}
|
||||
|
||||
/** Any aborted install: parked supervisors resume the old bundle right away. */
|
||||
export async function releaseServeSupervisorPark(): Promise<void> {
|
||||
const park = activePark
|
||||
activePark = null
|
||||
if (!park) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
await writeRecordDurably(park.dir, { ...park.record, phase: 'unpark' })
|
||||
} catch {
|
||||
// The park and commit deadlines still end the hold on their own.
|
||||
}
|
||||
}
|
||||
|
||||
export function serveInstallRefusalMessage(
|
||||
result: Extract<ServeInstallParkResult, { kind: 'refused' }>
|
||||
): string {
|
||||
const pids = result.pids.slice(0, 10).join(', ')
|
||||
if (result.reason === 'unresponsive') {
|
||||
return `Background orca serve services (process IDs: ${pids}) did not pause for the update in time. Orca remains open; retry the update, or stop those services first.`
|
||||
}
|
||||
if (result.reason === 'busy') {
|
||||
return 'Another Orca window is already installing this update. Orca remains open; retry in a minute.'
|
||||
}
|
||||
return `Close the other Orca instances (process IDs: ${pids}) before installing this update. Background orca serve instances also need to stop. Orca remains open; retry the update after closing them.`
|
||||
}
|
||||
@@ -7,7 +7,14 @@ import {
|
||||
isMacInstallerReady,
|
||||
setMacInstallPreflightInProgress
|
||||
} from '../updater-mac-install'
|
||||
import { homedir } from 'node:os'
|
||||
import { getMacUpdateRunningInstances } from '../macos-update-running-instances'
|
||||
import { macCoordinationRoot } from '../../shared/serve-bundle-coordination'
|
||||
import {
|
||||
commitServeSupervisorPark,
|
||||
parkServeSupervisorsForInstall,
|
||||
serveInstallRefusalMessage
|
||||
} from './serve-supervisor-install-park'
|
||||
import { armUpdateInstallExitWatchdog } from '../update-install-exit-watchdog'
|
||||
import { getLinuxPackageType } from '../linux-update-package-type'
|
||||
import { LINUX_PACKAGE_MARKER_UNUSABLE_MESSAGE } from '../linux-package-downloaded-status'
|
||||
@@ -83,18 +90,27 @@ export abstract class UpdaterInstallExecution extends UpdaterPackageRecovery {
|
||||
recordUpdaterLifecycle('macos_running_instances_check_failed')
|
||||
return
|
||||
}
|
||||
if (blockers.length > 0) {
|
||||
// Why: registered `orca serve --supervise` services can pause their server for us; anything else is refused as before.
|
||||
const park =
|
||||
blockers.length > 0
|
||||
? await parkServeSupervisorsForInstall(process.execPath, blockers, {
|
||||
macUserDataRoot: macCoordinationRoot(homedir()),
|
||||
listBlockers: () => getMacUpdateRunningInstances()
|
||||
})
|
||||
: null
|
||||
if (park && park.kind === 'refused') {
|
||||
this.resetQuitForUpdateState()
|
||||
this.mainWindowRef?.webContents.send('updater:quitAndInstallAborted')
|
||||
this.sendInstallFailureStatus({
|
||||
state: 'error',
|
||||
version: pendingVersion,
|
||||
retryAction: 'install',
|
||||
message: `Close the other Orca instances (process IDs: ${blockers.slice(0, 10).join(', ')}) before installing this update. Background orca serve instances also need to stop. Orca remains open; retry the update after closing them.`
|
||||
message: serveInstallRefusalMessage(park)
|
||||
})
|
||||
recordUpdaterLifecycle('macos_install_blocked_by_running_instances', {
|
||||
pids: blockers.slice(0, 10).join(', '),
|
||||
count: blockers.length
|
||||
pids: park.pids.slice(0, 10).join(', '),
|
||||
count: park.pids.length,
|
||||
reason: park.reason
|
||||
})
|
||||
return
|
||||
}
|
||||
@@ -137,6 +153,7 @@ export abstract class UpdaterInstallExecution extends UpdaterPackageRecovery {
|
||||
return
|
||||
}
|
||||
|
||||
await commitServeSupervisorPark(() => getMacUpdateRunningInstances())
|
||||
recordUpdaterLifecycle('quit_and_install_invoking_native', {
|
||||
version: pendingVersion || null
|
||||
})
|
||||
|
||||
@@ -12,6 +12,7 @@ import type { LinuxPackageInstallRecovery, UpdateStatus } from '../../shared/upd
|
||||
import { compareVersions } from '../updater-fallback'
|
||||
import { PRE_QUIT_CLEANUP_TIMEOUT_MS, REQUIRED_PRE_QUIT_CLEANUP_TIMEOUT_MS } from './updater-state'
|
||||
import { UpdaterCheckState } from './updater-check-state'
|
||||
import { releaseServeSupervisorPark } from './serve-supervisor-install-park'
|
||||
|
||||
export abstract class UpdaterInstallSupport extends UpdaterCheckState {
|
||||
protected getKnownReleaseUrl(): string | undefined {
|
||||
@@ -79,6 +80,7 @@ export abstract class UpdaterInstallSupport extends UpdaterCheckState {
|
||||
this.clearMacDeferredStagingTimer()
|
||||
disarmUpdateInstallExitWatchdog()
|
||||
resetMacInstallState()
|
||||
void releaseServeSupervisorPark()
|
||||
}
|
||||
|
||||
protected clearMacDeferredStagingTimer(): void {
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
/**
|
||||
* The record a desktop install and `orca serve --supervise` share to park a supervised server
|
||||
* while macOS replaces the app bundle both run from. Lives outside the bundle, keyed by its path.
|
||||
*/
|
||||
import { createHash } from 'node:crypto'
|
||||
import { join } from 'node:path'
|
||||
|
||||
/** launchd label prefix `orca serve service install` writes; only those supervisors register. */
|
||||
export const SERVE_LAUNCHD_LABEL_PREFIX = 'com.stablyai.orca.serve'
|
||||
export const SERVE_COORDINATION_PROTOCOL = 1
|
||||
/** A registration this stale belongs to a supervisor that is gone or wedged. */
|
||||
export const SERVE_SUPERVISOR_HEARTBEAT_STALE_MS = 15_000
|
||||
export const SERVE_SUPERVISOR_HEARTBEAT_MS = 3_000
|
||||
/** How long the installer waits for every parked supervisor to stop its server. */
|
||||
export const SERVE_PARK_ACK_WAIT_MS = 60_000
|
||||
/** A park the installer never committed or released ends on its own after this. */
|
||||
export const SERVE_PARK_DEADLINE_MS = 2 * 60_000
|
||||
/** A committed install that never changed the bundle is treated as cancelled after this. */
|
||||
export const SERVE_COMMIT_DEADLINE_MS = 10 * 60_000
|
||||
|
||||
export type ServeSupervisorRegistration = {
|
||||
protocol: 1
|
||||
pid: number
|
||||
childPid: number | null
|
||||
heartbeatAt: number
|
||||
label: string
|
||||
}
|
||||
|
||||
export type ServeParkRecord = {
|
||||
schemaVersion: 1
|
||||
attemptId: string
|
||||
installerPid: number
|
||||
phase: 'park' | 'commit' | 'unpark'
|
||||
createdAt: number
|
||||
parkDeadlineAt: number
|
||||
commitDeadlineAt: number | null
|
||||
/** Info.plist identity when the park began: a change proves the install happened. */
|
||||
bundleBefore: string
|
||||
supervisorPids: number[]
|
||||
}
|
||||
|
||||
/** Electron's default macOS userData base, shared by every profile of one user. */
|
||||
export function macCoordinationRoot(homeDir: string): string {
|
||||
return join(homeDir, 'Library', 'Application Support', 'orca')
|
||||
}
|
||||
|
||||
export function bundleCoordinationDir(realBundlePath: string, macUserDataRoot: string): string {
|
||||
const key = createHash('sha256').update(realBundlePath).digest('hex').slice(0, 32)
|
||||
return join(macUserDataRoot, 'bundle-coordination', key)
|
||||
}
|
||||
|
||||
export const supervisorsDir = (dir: string): string => join(dir, 'supervisors')
|
||||
export const parkRecordPath = (dir: string): string => join(dir, 'park.json')
|
||||
export const acksDir = (dir: string): string => join(dir, 'acks')
|
||||
export const ackFileName = (attemptId: string, pid: number): string => `${attemptId}.${pid}.json`
|
||||
|
||||
export function isServeSupervisorLabel(label: string | undefined): label is string {
|
||||
return (
|
||||
typeof label === 'string' &&
|
||||
(label === SERVE_LAUNCHD_LABEL_PREFIX || label.startsWith(`${SERVE_LAUNCHD_LABEL_PREFIX}.`))
|
||||
)
|
||||
}
|
||||
|
||||
const isPid = (value: unknown): value is number =>
|
||||
typeof value === 'number' && Number.isSafeInteger(value) && value > 0
|
||||
|
||||
export function parseServeSupervisorRegistration(
|
||||
value: unknown
|
||||
): ServeSupervisorRegistration | null {
|
||||
if (typeof value !== 'object' || value === null) {
|
||||
return null
|
||||
}
|
||||
if (
|
||||
!('protocol' in value) ||
|
||||
value.protocol !== SERVE_COORDINATION_PROTOCOL ||
|
||||
!('pid' in value) ||
|
||||
!isPid(value.pid) ||
|
||||
!('childPid' in value) ||
|
||||
(value.childPid !== null && !isPid(value.childPid)) ||
|
||||
!('heartbeatAt' in value) ||
|
||||
typeof value.heartbeatAt !== 'number' ||
|
||||
!('label' in value) ||
|
||||
!isServeSupervisorLabel(typeof value.label === 'string' ? value.label : undefined)
|
||||
) {
|
||||
return null
|
||||
}
|
||||
return {
|
||||
protocol: 1,
|
||||
pid: value.pid,
|
||||
childPid: value.childPid,
|
||||
heartbeatAt: value.heartbeatAt,
|
||||
label: String(value.label)
|
||||
}
|
||||
}
|
||||
|
||||
export function parseServeParkRecord(value: unknown): ServeParkRecord | null {
|
||||
if (typeof value !== 'object' || value === null) {
|
||||
return null
|
||||
}
|
||||
if (
|
||||
!('schemaVersion' in value) ||
|
||||
value.schemaVersion !== 1 ||
|
||||
!('attemptId' in value) ||
|
||||
typeof value.attemptId !== 'string' ||
|
||||
value.attemptId.length === 0 ||
|
||||
!('installerPid' in value) ||
|
||||
!isPid(value.installerPid) ||
|
||||
!('phase' in value) ||
|
||||
(value.phase !== 'park' && value.phase !== 'commit' && value.phase !== 'unpark') ||
|
||||
!('createdAt' in value) ||
|
||||
typeof value.createdAt !== 'number' ||
|
||||
!('parkDeadlineAt' in value) ||
|
||||
typeof value.parkDeadlineAt !== 'number' ||
|
||||
!('commitDeadlineAt' in value) ||
|
||||
(value.commitDeadlineAt !== null && typeof value.commitDeadlineAt !== 'number') ||
|
||||
!('bundleBefore' in value) ||
|
||||
typeof value.bundleBefore !== 'string' ||
|
||||
!('supervisorPids' in value) ||
|
||||
!Array.isArray(value.supervisorPids)
|
||||
) {
|
||||
return null
|
||||
}
|
||||
return {
|
||||
schemaVersion: 1,
|
||||
attemptId: value.attemptId,
|
||||
installerPid: value.installerPid,
|
||||
phase: value.phase,
|
||||
createdAt: value.createdAt,
|
||||
parkDeadlineAt: value.parkDeadlineAt,
|
||||
commitDeadlineAt: value.commitDeadlineAt,
|
||||
bundleBefore: value.bundleBefore,
|
||||
supervisorPids: value.supervisorPids.filter(isPid)
|
||||
}
|
||||
}
|
||||
|
||||
export function isLiveRegistration(
|
||||
registration: ServeSupervisorRegistration,
|
||||
now: number
|
||||
): boolean {
|
||||
return now - registration.heartbeatAt < SERVE_SUPERVISOR_HEARTBEAT_STALE_MS
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a supervisor must keep its server stopped. Every way out is evidence or a deadline, so
|
||||
* a crashed installer or supervisor never leaves the service parked for good.
|
||||
*/
|
||||
export function decideServeParkHold(
|
||||
record: ServeParkRecord | null,
|
||||
facts: { now: number; installerAlive: boolean; bundleNow: string }
|
||||
): 'hold' | 'release' {
|
||||
if (!record || record.phase === 'unpark') {
|
||||
return 'release'
|
||||
}
|
||||
if (record.phase === 'park') {
|
||||
return facts.installerAlive && facts.now < record.parkDeadlineAt ? 'hold' : 'release'
|
||||
}
|
||||
// Committed: the installer quits into ShipIt by design, so its death proves nothing.
|
||||
if (facts.bundleNow !== record.bundleBefore) {
|
||||
return 'release'
|
||||
}
|
||||
return record.commitDeadlineAt !== null && facts.now < record.commitDeadlineAt
|
||||
? 'hold'
|
||||
: 'release'
|
||||
}
|
||||
|
||||
/** Same-user liveness; EPERM still means the process exists. */
|
||||
export function isProcessAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
return true
|
||||
} catch (error) {
|
||||
return error instanceof Error && 'code' in error && error.code === 'EPERM'
|
||||
}
|
||||
}
|
||||
|
||||
export const bundleInfoPlistPath = (bundlePath: string): string =>
|
||||
join(bundlePath, 'Contents', 'Info.plist')
|
||||
|
||||
/** ShipIt swaps the whole .app, so Info.plist size or mtime changes even for a same-version reinstall. */
|
||||
export function formatBundleIdentity(stat: { size: number; mtimeMs: number } | null): string {
|
||||
return stat ? `${stat.size}:${Math.trunc(stat.mtimeMs)}` : 'missing'
|
||||
}
|
||||
Reference in New Issue
Block a user