fix(subprocess): preserve forced Linux outcomes
This commit is contained in:
1 parent
df8e3d7c37
commit
b6dc88be51
5 files changed
+60
-24
No files matched your search
@@ -4,9 +4,9 @@ import { randomBytes } from 'node:crypto'
|
||||
import { execFile, spawn, spawnSync } from 'node:child_process'
|
||||
import type { ChildProcess } from 'node:child_process'
|
||||
import { setTimeout as sleepMs } from 'node:timers/promises'
|
||||
import type { SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
|
||||
import type { SubprocessOutcome, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
|
||||
import type { BoundProcessOwner, ManagedProcessLaunch } from './managed-owner.ts'
|
||||
import { observeChildClose, waitWithAbort } from './managed-owner.ts'
|
||||
import { DirectResultUnavailableError, observeChildClose, waitWithAbort } from './managed-owner.ts'
|
||||
import { childEnv } from './spawn.ts'
|
||||
import {
|
||||
cleanupAfterRunner,
|
||||
@@ -114,6 +114,7 @@ class SystemdScopeOwner implements BoundProcessOwner {
|
||||
private readonly runSync: typeof spawnSync,
|
||||
private readonly query: (command: string, args: readonly string[]) => Promise<SystemctlResult>,
|
||||
private readonly runner: ChildProcess,
|
||||
private readonly onForceKillAttempt: () => void,
|
||||
) {}
|
||||
|
||||
signal(signal: NodeJS.Signals): void {
|
||||
@@ -125,6 +126,7 @@ class SystemdScopeOwner implements BoundProcessOwner {
|
||||
`--signal=${signal}`,
|
||||
this.unit,
|
||||
], { encoding: 'utf8', env: systemctlEnv(), timeout: SYSTEMCTL_TIMEOUT_MS })
|
||||
if (signal === 'SIGKILL' && result.error === undefined) this.onForceKillAttempt()
|
||||
if (result.error === undefined && result.status === 0) {
|
||||
return
|
||||
}
|
||||
@@ -211,15 +213,38 @@ export function launchLinuxScope(
|
||||
stdio: runnerStdio(spec),
|
||||
})
|
||||
const closed = observeChildClose(child)
|
||||
const owner = new SystemdScopeOwner(`${unitBase}.scope`, systemctl, runSync, query, child)
|
||||
const result = runnerDirectResult(child, files, closed)
|
||||
cleanupAfterRunner(files, result.direct, closed)
|
||||
let forceKillAttempted = false
|
||||
let scopeSettled = false
|
||||
const owner = new SystemdScopeOwner(
|
||||
`${unitBase}.scope`,
|
||||
systemctl,
|
||||
runSync,
|
||||
query,
|
||||
child,
|
||||
() => { forceKillAttempted = true },
|
||||
)
|
||||
const resultFinalized = closed.then(async () => {
|
||||
try {
|
||||
await owner.waitForExit()
|
||||
scopeSettled = true
|
||||
} catch (_rangeObservationFailed) {
|
||||
// waitForExit retains the authoritative owner failure for its caller.
|
||||
}
|
||||
})
|
||||
const result = runnerDirectResult(child, files, resultFinalized)
|
||||
const direct = result.direct.catch((error: unknown): SubprocessOutcome => {
|
||||
if (forceKillAttempted && scopeSettled && error instanceof DirectResultUnavailableError) {
|
||||
return { exitCode: null, signal: 'SIGKILL' }
|
||||
}
|
||||
throw error
|
||||
})
|
||||
cleanupAfterRunner(files, direct, closed)
|
||||
return {
|
||||
stdin: child.stdin,
|
||||
stdout: child.stdout,
|
||||
stderr: child.stderr,
|
||||
pid: result.pid,
|
||||
direct: result.direct,
|
||||
direct,
|
||||
closed,
|
||||
owner,
|
||||
}
|
||||
|
||||
@@ -4,6 +4,11 @@ import type { ChildProcess } from 'node:child_process'
|
||||
import type { Readable, Writable } from 'node:stream'
|
||||
import type { SubprocessOutcome } from '@deepseek-ai/dsh-subprocess'
|
||||
|
||||
/** Direct target started, but its runner closed before publishing an exit event. */
|
||||
export class DirectResultUnavailableError extends Error {
|
||||
override name = 'DirectResultUnavailableError'
|
||||
}
|
||||
|
||||
/** Platform owner used by termination and whole-range settlement. */
|
||||
export interface BoundProcessOwner {
|
||||
/** Signal the established managed range; a confirmed-stopped owner stays inert. */
|
||||
|
||||
@@ -14,6 +14,7 @@ import {
|
||||
readRunnerEventsAsync,
|
||||
} from './runner-protocol.ts'
|
||||
import type { RunnerEvent, RunnerFiles, RunnerRequest } from './runner-protocol.ts'
|
||||
import { DirectResultUnavailableError } from './managed-owner.ts'
|
||||
import { childEnv } from './spawn.ts'
|
||||
|
||||
let handshakeWait: Int32Array | undefined
|
||||
@@ -114,24 +115,23 @@ function waitForRunnerHandshake(child: ChildProcess, files: RunnerFiles): Runner
|
||||
async function waitForDirectResult(
|
||||
files: RunnerFiles,
|
||||
initial: RunnerEvent[],
|
||||
closed: Promise<void>,
|
||||
resultFinalized: Promise<void>,
|
||||
): Promise<SubprocessOutcome> {
|
||||
let seen = 0
|
||||
const wrapperState = { closed: false }
|
||||
void closed.then(() => { wrapperState.closed = true })
|
||||
const resultState = { finalized: false }
|
||||
void resultFinalized.then(() => { resultState.finalized = true })
|
||||
for (;;) {
|
||||
// A read started before close may return a stale snapshot after close has
|
||||
// become visible. Only a read started after close can prove no terminal
|
||||
// event was written before the runner exited.
|
||||
const closedBeforeRead = wrapperState.closed
|
||||
// A read started before finalization may return a stale snapshot. Only a
|
||||
// read started afterward can prove no terminal event remains forthcoming.
|
||||
const finalizedBeforeRead = resultState.finalized
|
||||
const events = await readRunnerEventsAsync(files.eventsPath)
|
||||
for (const event of events.slice(seen)) {
|
||||
if (event.type === 'exit') return { exitCode: event.exitCode, signal: event.signal }
|
||||
if (event.type === 'spawn-error' || event.type === 'runner-error') throw deserializeSpawnError(event.error)
|
||||
}
|
||||
seen = Math.max(seen, events.length, initial.length)
|
||||
if (closedBeforeRead) {
|
||||
throw new Error('native subprocess runner exited without a direct-command result')
|
||||
if (finalizedBeforeRead) {
|
||||
throw new DirectResultUnavailableError('native subprocess runner exited without a direct-command result')
|
||||
}
|
||||
await sleepMs(RUNNER_EVENT_POLL_MS)
|
||||
}
|
||||
@@ -141,13 +141,13 @@ async function waitForDirectResult(
|
||||
* Bind runner events into one direct result while preserving the target pid.
|
||||
* @param child - native wrapper process.
|
||||
* @param files - private request and result paths.
|
||||
* @param closed - wrapper close observation attached before the start handshake.
|
||||
* @param resultFinalized - platform proof that no later runner event can arrive.
|
||||
* @returns target pid, direct result, and whether the runner already reported a pre-start terminal failure.
|
||||
*/
|
||||
export function runnerDirectResult(
|
||||
child: ChildProcess,
|
||||
files: RunnerFiles,
|
||||
closed: Promise<void>,
|
||||
resultFinalized: Promise<void>,
|
||||
): {
|
||||
pid: number
|
||||
direct: Promise<SubprocessOutcome>
|
||||
@@ -162,7 +162,7 @@ export function runnerDirectResult(
|
||||
}
|
||||
return {
|
||||
pid: handshake.pid,
|
||||
direct: waitForDirectResult(files, handshake.events, closed),
|
||||
direct: waitForDirectResult(files, handshake.events, resultFinalized),
|
||||
failureReported: handshake.failureReported,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -113,7 +113,7 @@ describe.skipIf(process.platform === 'win32')('Linux systemd scope adapter', ()
|
||||
expect(systemdArgs).not.toContain('literal $VALUE')
|
||||
})
|
||||
|
||||
it('still escalates after a missing-unit TERM response without inventing a direct result', async () => {
|
||||
it('uses a successful scope KILL when the runner cannot publish the direct result', async () => {
|
||||
let wrapper: ReturnType<typeof spawn> | undefined
|
||||
const run = vi.fn((_command: string, args: readonly string[], options: Parameters<typeof spawn>[2]) => {
|
||||
const separator = args.indexOf('--')
|
||||
@@ -127,6 +127,12 @@ describe.skipIf(process.platform === 'win32')('Linux systemd scope adapter', ()
|
||||
return { status: 1, stdout: '', stderr: 'Unit could not be found', error: undefined }
|
||||
}
|
||||
if (wrapper?.pid !== undefined) process.kill(-wrapper.pid, 'SIGKILL')
|
||||
return {
|
||||
status: 1,
|
||||
stdout: '',
|
||||
stderr: 'Failed to send signal SIGKILL to auxiliary processes: Invalid argument',
|
||||
error: undefined,
|
||||
}
|
||||
}
|
||||
if (command === 'systemctl' && args[1] === 'show') {
|
||||
const active = wrapper?.exitCode === null && wrapper.signalCode === null
|
||||
@@ -142,7 +148,7 @@ describe.skipIf(process.platform === 'win32')('Linux systemd scope adapter', ()
|
||||
})
|
||||
launch.owner.signal('SIGTERM')
|
||||
launch.owner.signal('SIGKILL')
|
||||
await expect(launch.direct).rejects.toThrow('exited without a direct-command result')
|
||||
await expect(launch.direct).resolves.toEqual({ exitCode: null, signal: 'SIGKILL' })
|
||||
await expect(launch.owner.waitForExit()).resolves.toBe(true)
|
||||
})
|
||||
|
||||
|
||||
@@ -333,7 +333,7 @@ describe('spawn runner transport', () => {
|
||||
|
||||
})
|
||||
|
||||
it('requires an event snapshot started after wrapper close before reporting a missing result', async () => {
|
||||
it('requires an event snapshot started after result finalization before reporting a missing result', async () => {
|
||||
const staleRead = Promise.withResolvers<Awaited<ReturnType<typeof readRunnerEventsAsync>>>()
|
||||
let readCount = 0
|
||||
vi.resetModules()
|
||||
@@ -351,12 +351,12 @@ describe('spawn runner transport', () => {
|
||||
const files = createRunnerFiles({ argv: ['node'], cwd: '.', env: {} })
|
||||
try {
|
||||
appendRunnerEvent(files.eventsPath, { type: 'started', pid: 456 })
|
||||
const closed = Promise.withResolvers<undefined>()
|
||||
const finalized = Promise.withResolvers<undefined>()
|
||||
const isolated = await import('../src/runner-launch.ts')
|
||||
const result = isolated.runnerDirectResult(fakeChild(123), files, closed.promise)
|
||||
const result = isolated.runnerDirectResult(fakeChild(123), files, finalized.promise)
|
||||
expect(result.failureReported).toBe(false)
|
||||
expect(readCount).toBe(1)
|
||||
closed.resolve(undefined)
|
||||
finalized.resolve(undefined)
|
||||
await Promise.resolve()
|
||||
appendRunnerEvent(files.eventsPath, { type: 'exit', exitCode: 0, signal: null })
|
||||
staleRead.resolve([{ type: 'started', pid: 456 }])
|
||||
|
||||
Reference in new issue
Block a user