Files
deepseek-harness/packages/subprocess/subprocess-local/src/linux-scope.ts
T
turtle1999 aaa02a3970 fix(subprocess): settle a Linux scope left active with no processes
Range observation waited for an active transient scope to leave `active`
before it would prove the managed range empty. systemd ends a scope only when
it observes the populated-to-empty transition, so a payload killed before the
manager placed it in the scope's cgroup leaves the unit active with an empty
cgroup forever. `waitForExit()` then polled a unit that could not change, and
`LocalSubprocessRuntime.disposeManagedProcesses()` awaited it without bound —
teardown hung for any command cancelled immediately after spawn.

Each state query now also reads the unit's process count. Once termination was
requested and the launcher has exited, a still-active unit reporting no
processes is the empty range: the launcher forked every process it will ever
fork, so nothing later joins that cgroup. The owner stops the leftover unit so
transient units cannot accumulate, and an unreported or `[not set]` count
stays unknown and keeps waiting.

The race needs a live user-systemd manager, so it stayed dormant wherever the
native scope path was unavailable and surfaced once CI routed lanes onto the
Blacksmith hosted image, whose runners accept that path. Ten repetitions of
`local.spec.ts > disposal kills still-running processes and awaits their exit`
in a Linux container with a user manager failed seven times before the change
and passed ten times after it.
2026-09-10 17:27:52 +08:00

550 lines
19 KiB
TypeScript

/** Linux user-systemd scope launch and managed-range ownership. */
import { execFile, spawn, spawnSync } from 'node:child_process'
import { randomBytes } from 'node:crypto'
import { existsSync } from 'node:fs'
import { setTimeout as sleepMs } from 'node:timers/promises'
import type {
SubprocessOutcome,
SubprocessSpawnSpec,
SubprocessTerminalSpawnSpec,
} from '@deepseek-ai/dsh-subprocess'
import { loadLinuxExecve } from './linux-execve.ts'
import type { BoundProcessOwner, ManagedProcessLaunch } from './managed-owner.ts'
import {
cleanupLinuxLaunchFiles,
createLinuxLaunchFiles,
deserializeRunnerError,
readLinuxStartupError,
} from './runner-protocol.ts'
import type { LinuxLaunchFiles } from './runner-protocol.ts'
import {
runnerEnvironment,
runnerInvocationAvailable,
runnerStdio,
spawnRunnerInvocation,
} from './runner-launch.ts'
import type { RunnerInvocation } from './runner-launch.ts'
import { childEnv } from './spawn.ts'
/** Test seams for systemd command execution. */
export interface LinuxScopeInternals {
spawn?: typeof spawn
spawnSync?: typeof spawnSync
systemctlQuery?: (command: string, args: readonly string[]) => Promise<SystemctlResult>
systemdRun?: string
systemctl?: string
runnerInvocation?: RunnerInvocation
resolveRunnerInvocation?: () => RunnerInvocation
runnerAvailable?: (invocation: RunnerInvocation) => boolean
loadLinuxExecve?: typeof loadLinuxExecve
sleep?: (delayMs: number, signal?: AbortSignal) => Promise<void>
}
interface SystemctlResult {
status: number | null
stdout: string
stderr: string
error?: Error
}
const SYSTEMCTL_TIMEOUT_MS = 5_000
const SCOPE_INITIAL_POLL_INTERVAL_MS = 50
const MISSING_UNIT = /\bunit\b[^\r\n]*(?:could not be found|not found|not loaded)/iu
function managerEnvironment(): NodeJS.ProcessEnv {
const environment = childEnv({ LC_ALL: 'C' })
delete environment.SYSTEMD_LOG_TARGET
return environment
}
function quietSystemdEnvironment(): NodeJS.ProcessEnv {
return childEnv({ LC_ALL: 'C', SYSTEMD_LOG_TARGET: 'null' })
}
function querySystemctl(command: string, args: readonly string[]): Promise<SystemctlResult> {
return new Promise((resolveResult) => {
execFile(command, [...args], {
encoding: 'utf8',
env: managerEnvironment(),
timeout: SYSTEMCTL_TIMEOUT_MS,
}, (error, stdout, stderr) => {
const code = error === null ? 0 : (error as Error & { code?: string | number }).code
resolveResult({
status: typeof code === 'number' ? code : null,
stdout,
stderr,
...error === null ? {} : { error },
})
})
})
}
function unitStem(prefix: string): string {
return `${prefix}-${String(process.pid)}-${randomBytes(6).toString('hex')}`
}
function sleepWithAbort(delayMs: number, signal?: AbortSignal): Promise<void> {
return sleepMs(delayMs, undefined, { signal })
}
/**
* Confirm this exact runner entry and libc execve binding without a probe mode.
* @param internals - optional runner and libc-binding seams used by tests.
* @returns whether the bootstrap can enter the final target.
*/
export function probeLinuxBootstrap(internals: LinuxScopeInternals = {}): boolean {
try {
;(internals.loadLinuxExecve ?? loadLinuxExecve)()
const invocation = internals.runnerInvocation
?? (internals.resolveRunnerInvocation ?? spawnRunnerInvocation)()
return (internals.runnerAvailable ?? runnerInvocationAvailable)(invocation)
} catch {
return false
}
}
/**
* Confirm current literal-argv transient-scope support before selecting native launch.
* @param internals - optional systemd command seams used by tests.
* @returns whether the current user manager supports the required scope invocation.
*/
export function probeLinuxScope(internals: LinuxScopeInternals = {}): boolean {
const unitBase = unitStem('dsh-subprocess-probe')
const result = (internals.spawnSync ?? spawnSync)(internals.systemdRun ?? 'systemd-run', [
'--user',
'--scope',
'--quiet',
'--collect',
'--expand-environment=no',
`--unit=${unitBase}`,
'--',
internals.systemctl ?? 'systemctl',
'--user',
'show',
`${unitBase}.scope`,
'--property=ActiveState',
'--value',
], { env: quietSystemdEnvironment(), stdio: 'ignore', timeout: SYSTEMCTL_TIMEOUT_MS })
return result.error === undefined && result.status === 0
}
/**
* Confirm that the current user manager remains reachable after a positive deep probe.
* @param internals - optional systemctl seam used by tests.
* @returns whether one lightweight manager query succeeds.
*/
export function probeLinuxManager(internals: LinuxScopeInternals = {}): boolean {
const result = (internals.spawnSync ?? spawnSync)(internals.systemctl ?? 'systemctl', [
'--user',
'show',
'--property=Version',
'--value',
], { env: managerEnvironment(), stdio: 'ignore', timeout: SYSTEMCTL_TIMEOUT_MS })
return result.error === undefined && result.status === 0
}
/**
* Re-check every Linux native prerequisite for one eligible spawn.
* @param internals - optional native capability seams used by tests.
* @returns whether the Linux native containment path is currently available.
*/
export function probeLinuxNative(internals: LinuxScopeInternals = {}): boolean {
return probeLinuxBootstrap(internals)
&& probeLinuxScope(internals)
}
interface DirectRange {
running(): boolean
signal(signal: 'SIGTERM' | 'SIGKILL'): void
}
class LinuxScopeStartup {
readonly terminationSignals = new Set<NodeJS.Signals>()
constructor(readonly files: LinuxLaunchFiles, private readonly kind: 'subprocess' | 'terminal') {}
resolveOutcome(outcome: SubprocessOutcome): SubprocessOutcome {
const startup = readLinuxStartupError(this.files.startupErrorPath)
if (startup !== undefined) throw deserializeRunnerError(startup.error)
if (existsSync(this.files.requestPath)
&& !(outcome.signal !== null && this.terminationSignals.has(outcome.signal))) {
throw new Error(`${this.kind} scope exited before its bootstrap consumed the launch request`)
}
return outcome
}
}
class SystemdScopeOwner implements BoundProcessOwner {
private establishment: 'pending' | 'established' = 'pending'
private stopped = false
private terminationRequested = false
private observation: Promise<void> | undefined
private killFailure: Error | undefined
private wakeGeneration = 0
private wakeWaiter: { generation: number; resolve: () => void } | undefined
constructor(
private readonly unit: string,
private readonly startup: LinuxScopeStartup,
private readonly direct: DirectRange,
private readonly systemctl: string,
private readonly runSync: typeof spawnSync,
private readonly query: (command: string, args: readonly string[]) => Promise<SystemctlResult>,
private readonly sleep: (delayMs: number, signal?: AbortSignal) => Promise<void>,
) {}
signal(signal: 'SIGTERM' | 'SIGKILL'): void {
if (this.stopped) return
this.terminationRequested = true
if (this.direct.running()) this.startup.terminationSignals.add(signal)
this.observeRequestConsumption()
const directFallbackRequired = this.establishment === 'pending'
if (directFallbackRequired && this.direct.running()) this.direct.signal(signal)
const result = this.runSync(this.systemctl, [
'--user',
'kill',
'--kill-whom=all',
`--signal=${signal}`,
this.unit,
], { encoding: 'utf8', env: managerEnvironment(), timeout: SYSTEMCTL_TIMEOUT_MS })
this.wakeObservation()
if (result.error === undefined && result.status === 0) {
if (signal === 'SIGKILL') this.killFailure = undefined
return
}
if (!directFallbackRequired && this.direct.running()) this.direct.signal(signal)
if (signal === 'SIGKILL') {
const output = `${result.stdout}\n${result.stderr}`
if (!MISSING_UNIT.test(output)) {
this.killFailure = result.error ?? new Error(
`systemctl could not signal ${this.unit}: ${output.trim() || `exit ${String(result.status)}`}`,
)
}
}
}
terminateForHostExit(): void {
if (this.stopped) return
try {
if (this.direct.running()) this.direct.signal('SIGKILL')
} catch { /* Continue with the native owner. */ }
try {
this.runSync(this.systemctl, [
'--user',
'kill',
'--kill-whom=all',
'--signal=SIGKILL',
this.unit,
], { env: managerEnvironment(), stdio: 'ignore', timeout: SYSTEMCTL_TIMEOUT_MS })
} catch {
// Host exit cannot report one range; the runtime continues with the rest.
}
}
private observeRequestConsumption(): void {
if (this.establishment === 'pending' && !existsSync(this.startup.files.requestPath)) {
this.establishment = 'established'
}
}
private absentUnit(): boolean {
this.observeRequestConsumption()
if (this.establishment === 'established') return false
if (!this.direct.running() && existsSync(this.startup.files.requestPath)) {
return false
}
if (this.killFailure !== undefined) throw this.killFailure
return true
}
/**
* Prove an active unit with no processes is the empty managed range rather
* than a launch still placing its payload. systemd ends a scope only on the
* populated-to-empty transition, so a payload killed before it entered the
* cgroup leaves the unit active forever. Requested termination plus a
* departed client makes that leftover conclusive: the client forked every
* process it will ever fork.
*/
private emptyRange(tasksCurrent: number | undefined): boolean {
return this.terminationRequested && tasksCurrent === 0 && !this.direct.running()
}
/** Release a leftover empty scope so the transient unit is collected and cannot accumulate. */
private releaseEmptyRange(): void {
try {
this.runSync(this.systemctl, ['--user', 'stop', this.unit], {
env: managerEnvironment(),
stdio: 'ignore',
timeout: SYSTEMCTL_TIMEOUT_MS,
})
} catch {
// The range is already empty; a failed cleanup leaves only the transient unit.
}
}
private parseUnitState(stdout: string): { loadState: string; activeState: string; tasksCurrent: number | undefined } {
const values = new Map<string, string>()
for (const line of stdout.split(/\r?\n/u)) {
if (line === '') continue
const separator = line.indexOf('=')
if (separator <= 0) {
throw new Error(`systemctl returned malformed state for ${this.unit}: ${JSON.stringify(stdout.trim())}`)
}
const name = line.slice(0, separator)
if (values.has(name)) {
throw new Error(`systemctl returned duplicate ${name} for ${this.unit}`)
}
values.set(name, line.slice(separator + 1))
}
const loadState = values.get('LoadState')
const activeState = values.get('ActiveState')
// The manager prints this sentinel for a property the unit does not carry.
const reportedTasks = values.get('TasksCurrent')
const tasksCurrent = reportedTasks === '[not set]' ? undefined : reportedTasks
if (values.size !== (reportedTasks === undefined ? 2 : 3)
|| loadState === undefined || activeState === undefined) {
throw new Error(`systemctl returned incomplete state for ${this.unit}: ${JSON.stringify(stdout.trim())}`)
}
if (tasksCurrent !== undefined && !/^\d+$/u.test(tasksCurrent)) {
throw new Error(`systemctl returned a non-numeric TasksCurrent for ${this.unit}: ${JSON.stringify(tasksCurrent)}`)
}
return {
loadState,
activeState,
tasksCurrent: tasksCurrent === undefined ? undefined : Number(tasksCurrent),
}
}
private async rangeActive(): Promise<boolean> {
this.observeRequestConsumption()
const result = await this.query(this.systemctl, [
'--user',
'show',
this.unit,
'--property=LoadState',
'--property=ActiveState',
'--property=TasksCurrent',
])
const output = `${result.stdout}\n${result.stderr}`
if (result.status === 0) {
const { loadState, activeState, tasksCurrent } = this.parseUnitState(result.stdout)
if (loadState === 'not-found' && activeState === 'inactive') return this.absentUnit()
if (loadState !== 'loaded') {
throw new Error(
`systemctl returned unknown state for ${this.unit}: ${JSON.stringify({ loadState, activeState })}`,
)
}
this.establishment = 'established'
if (activeState === 'inactive' || activeState === 'failed') return false
if (!['active', 'activating', 'reloading', 'deactivating'].includes(activeState)) {
throw new Error(`systemctl returned unknown ActiveState for ${this.unit}: ${JSON.stringify(activeState)}`)
}
if (this.killFailure !== undefined) throw this.killFailure
if (this.emptyRange(tasksCurrent)) {
this.releaseEmptyRange()
return false
}
return true
}
if (!MISSING_UNIT.test(output)) {
if (result.error !== undefined) throw result.error
throw new Error(`systemctl could not read ${this.unit}: ${output.trim() || `exit ${String(result.status)}`}`)
}
return this.absentUnit()
}
private wakeObservation(): void {
this.wakeGeneration += 1
this.wakeWaiter?.resolve()
this.wakeWaiter = undefined
}
private async waitForPoll(delayMs: number, generation: number): Promise<void> {
if (generation !== this.wakeGeneration) return
const wake = Promise.withResolvers<void>()
const waiter = { generation, resolve: wake.resolve }
const sleepController = new AbortController()
this.wakeWaiter = waiter
try {
await Promise.race([this.sleep(delayMs, sleepController.signal), wake.promise])
} finally {
sleepController.abort()
if (this.wakeWaiter === waiter) this.wakeWaiter = undefined
}
}
async waitForExit(): Promise<void> {
if (this.stopped) return
this.observation ??= (async () => {
let pollIntervalMs = SCOPE_INITIAL_POLL_INTERVAL_MS
let generation = this.wakeGeneration
while (await this.rangeActive()) {
await this.waitForPoll(pollIntervalMs, generation)
generation = this.wakeGeneration
// Keep establishment responsive, then reduce systemctl process churn
// while systemd remains the authoritative owner of an active range.
if (this.establishment === 'established') {
pollIntervalMs = Math.min(pollIntervalMs * 2, SYSTEMCTL_TIMEOUT_MS)
}
}
this.stopped = true
})().catch((error: unknown) => {
this.observation = undefined
throw error
})
await this.observation
}
cleanup(): void {
cleanupLinuxLaunchFiles(this.startup.files)
}
}
function scopeArgs(unitBase: string, invocation: RunnerInvocation, argv: readonly string[]): string[] {
return [
'--user',
'--scope',
'--quiet',
'--collect',
'--expand-environment=no',
`--unit=${unitBase}`,
'--',
...invocation,
'--',
...argv,
]
}
function directOutcome(
child: ReturnType<typeof spawn>,
startup: LinuxScopeStartup,
): Promise<SubprocessOutcome> {
return new Promise((resolveOutcome, rejectOutcome) => {
let settled = false
child.once('error', (error) => {
if (settled) return
settled = true
rejectOutcome(error)
})
child.once('exit', (exitCode, signal) => {
if (settled) return
settled = true
try {
resolveOutcome(startup.resolveOutcome({ exitCode, signal }))
} catch (error) {
/* v8 ignore next -- Node filesystem operations throw Error instances. */
const failure = error instanceof Error ? error : new Error(String(error))
rejectOutcome(failure)
}
})
})
}
function signalChildGroup(child: ReturnType<typeof spawn>, signal: 'SIGTERM' | 'SIGKILL'): void {
try {
process.kill(-(child.pid as number), signal)
} catch {
try { child.kill(signal) } catch { /* The direct process already exited. */ }
}
}
/** Linux PTY invocation and owner for the exact one-shot scope/bootstrap. */
export interface LinuxTerminalScopeLaunch {
command: string
args: string[]
cwd: string
env: NodeJS.ProcessEnv
bindOwner: (direct: DirectRange) => BoundProcessOwner
resolveOutcome: (outcome: SubprocessOutcome) => SubprocessOutcome
cleanup: () => void
}
/**
* Prepare one Linux PTY scope using the same launch request and bootstrap core.
* @param spec - terminal target request.
* @param targetEnv - validated complete target environment.
* @param internals - optional runner and systemd seams used by tests.
* @returns invocation and ownership callbacks; requested termination preserves the observed signal even before bootstrap consumption.
*/
export function prepareLinuxTerminalScope(
spec: SubprocessTerminalSpawnSpec,
targetEnv: Record<string, string>,
internals: LinuxScopeInternals = {},
): LinuxTerminalScopeLaunch {
const invocation = internals.runnerInvocation ?? spawnRunnerInvocation()
const files = createLinuxLaunchFiles({ cwd: spec.cwd, env: targetEnv })
const startup = new LinuxScopeStartup(files, 'terminal')
const unitBase = unitStem('dsh-terminal')
return {
command: internals.systemdRun ?? 'systemd-run',
args: scopeArgs(unitBase, invocation, spec.argv),
cwd: process.cwd(),
env: runnerEnvironment(files.requestPath, invocation),
bindOwner: direct => new SystemdScopeOwner(
`${unitBase}.scope`,
startup,
direct,
internals.systemctl ?? 'systemctl',
internals.spawnSync ?? spawnSync,
internals.systemctlQuery ?? querySystemctl,
internals.sleep ?? sleepWithAbort,
),
resolveOutcome: outcome => startup.resolveOutcome(outcome),
cleanup: () => { cleanupLinuxLaunchFiles(files) },
}
}
/**
* Launch one ordinary target inside a transient user scope.
* @param spec - ordinary target request.
* @param targetEnv - validated complete target environment.
* @param internals - optional runner and systemd seams used by tests.
* @returns streams, result, and scope owner; requested termination preserves the observed signal even before bootstrap consumption.
*/
export function launchLinuxScope(
spec: SubprocessSpawnSpec,
targetEnv: Record<string, string>,
internals: LinuxScopeInternals = {},
): ManagedProcessLaunch {
const invocation = internals.runnerInvocation ?? spawnRunnerInvocation()
const files = createLinuxLaunchFiles({ cwd: spec.cwd, env: targetEnv })
const startup = new LinuxScopeStartup(files, 'subprocess')
const unitBase = unitStem('dsh-subprocess')
let child: ReturnType<typeof spawn>
try {
child = (internals.spawn ?? spawn)(internals.systemdRun ?? 'systemd-run', scopeArgs(
unitBase,
invocation,
spec.argv,
), {
cwd: process.cwd(),
env: runnerEnvironment(files.requestPath, invocation),
stdio: runnerStdio(spec, false),
detached: true,
})
} catch (error) {
cleanupLinuxLaunchFiles(files)
throw error
}
const owner = new SystemdScopeOwner(
`${unitBase}.scope`,
startup,
{
running: () => child.pid !== undefined && child.exitCode === null && child.signalCode === null,
signal: (signal) => { signalChildGroup(child, signal) },
},
internals.systemctl ?? 'systemctl',
internals.spawnSync ?? spawnSync,
internals.systemctlQuery ?? querySystemctl,
internals.sleep ?? sleepWithAbort,
)
return {
stdin: child.stdin,
stdout: child.stdout,
stderr: child.stderr,
direct: directOutcome(child, startup),
owner,
}
}