mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(process): three P1s from review — a crash vector and two wedge bugs
STDIN EPIPE COULD TAKE DOWN THE MAIN PROCESS. A child that exits without
reading makes the queued write fail with EPIPE, and an unhandled error on
a stream is an uncaught exception. The child's own error listener does
not cover its stdin stream, so runProcess({ input }) against a
short-lived child was a crash, not a failed call.
THE COOLDOWN LEAKED A BATCH PER CYCLE INSTEAD OF BOUNDING IT. At expiry
every concurrent caller passed the check before any of them re-armed it,
so each enqueued a callback into the still-latched native queue and each
cycle leaked another batch. The cooldown is now re-armed BEFORE probing,
so exactly one caller gets through.
A SYNCHRONOUS THROW LEFT ITS DEADLINE RUNNING. The timer was declared
inside the try, so catch could not clear it; it fired later and wedged a
reader that had already recovered. Hoisted and cleared, and wedge state
now carries a generation so a request that lost its deadline cannot
mutate it on behalf of the one that replaced it.
Found by review once the prompts were short enough for the reviewer to
finish -- the previous two rounds died on prompt length.
This commit is contained in:
@@ -144,4 +144,55 @@ describe('wedge cooldown', () => {
|
||||
await expect(readWindowsProcessTableFresh()).rejects.toThrow(/cooling down/)
|
||||
expect(getAllProcesses).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('lets exactly one caller probe when the cooldown expires', async () => {
|
||||
// Without claiming the recovery slot before probing, every concurrent
|
||||
// caller passes the cooldown check at expiry and each enqueues a callback
|
||||
// into the still-latched native queue -- so each cycle leaks another batch
|
||||
// rather than bounding it to one probe.
|
||||
vi.useFakeTimers()
|
||||
const getAllProcesses = vi.fn(() => {})
|
||||
__setWindowsProcessTreeLoaderForTests(() => ({
|
||||
ProcessDataFlag: { None: 0, Memory: 1, CommandLine: 2 },
|
||||
getAllProcesses
|
||||
}))
|
||||
|
||||
const wedge = readWindowsProcessTableFresh()
|
||||
const wedgeAssertion = expect(wedge).rejects.toThrow(/timed out/)
|
||||
await vi.advanceTimersByTimeAsync(3_000)
|
||||
await wedgeAssertion
|
||||
expect(getAllProcesses).toHaveBeenCalledTimes(1)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(30_000)
|
||||
const attempts = [
|
||||
readWindowsProcessTableFresh().catch(() => 'rejected'),
|
||||
readWindowsProcessTableFresh().catch(() => 'rejected'),
|
||||
readWindowsProcessTableFresh().catch(() => 'rejected')
|
||||
]
|
||||
await vi.advanceTimersByTimeAsync(3_000)
|
||||
await Promise.all(attempts)
|
||||
|
||||
// One recovery probe, not three.
|
||||
expect(getAllProcesses).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('clears the deadline when the reader throws synchronously', async () => {
|
||||
// An orphaned timer would fire later and wedge a reader that had recovered.
|
||||
vi.useFakeTimers()
|
||||
const getAllProcesses = vi.fn(() => {
|
||||
throw new Error('addon exploded')
|
||||
})
|
||||
__setWindowsProcessTreeLoaderForTests(() => ({
|
||||
ProcessDataFlag: { None: 0, Memory: 1, CommandLine: 2 },
|
||||
getAllProcesses
|
||||
}))
|
||||
|
||||
await expect(readWindowsProcessTableFresh()).rejects.toThrow(/exploded/)
|
||||
await vi.advanceTimersByTimeAsync(10_000)
|
||||
|
||||
// The recovered reader must answer, not report a wedge left by a dead timer.
|
||||
getAllProcesses.mockImplementation((cb: (rows: unknown) => void) => cb(NATIVE))
|
||||
resetWindowsProcessTableForTests()
|
||||
await expect(readWindowsProcessTableFresh()).resolves.toHaveLength(NATIVE.length)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -103,6 +103,9 @@ const WINDOWS_PROCESS_QUERY_TIMEOUT_MS = 3_000
|
||||
const WINDOWS_PROCESS_QUERY_COOLDOWN_MS = 30_000
|
||||
|
||||
let wedgedUntilMs = 0
|
||||
// Why a generation: a request that already lost its deadline must not later
|
||||
// clear or re-arm the wedge on behalf of the request that replaced it.
|
||||
let readGeneration = 0
|
||||
|
||||
function readNativeRows(): Promise<WindowsProcessRow[]> {
|
||||
const native = moduleLoader()
|
||||
@@ -112,9 +115,19 @@ function readNativeRows(): Promise<WindowsProcessRow[]> {
|
||||
// tree dead. "Unavailable" has to stay distinguishable from "empty".
|
||||
return Promise.reject(new Error('windows process table unavailable'))
|
||||
}
|
||||
if (Date.now() < wedgedUntilMs) {
|
||||
const startedAt = Date.now()
|
||||
if (startedAt < wedgedUntilMs) {
|
||||
return Promise.reject(new Error('windows process table is cooling down after a timeout'))
|
||||
}
|
||||
if (wedgedUntilMs > 0) {
|
||||
// Coming out of a wedge: re-arm the cooldown BEFORE probing, so exactly one
|
||||
// caller gets through. Without this every concurrent caller passes the
|
||||
// check above at expiry, each enqueues a callback into the still-latched
|
||||
// native queue, and each cooldown cycle leaks another batch rather than
|
||||
// bounding it to one probe.
|
||||
wedgedUntilMs = startedAt + WINDOWS_PROCESS_QUERY_COOLDOWN_MS
|
||||
}
|
||||
const generation = ++readGeneration
|
||||
// Why always both flags: each adds an OpenProcess per process (Memory a
|
||||
// GetProcessMemoryInfo, CommandLine a PEB read), so asking for less would be
|
||||
// cheaper -- 15.9ms p50 versus 30.6ms at 1050 processes. But every read shares
|
||||
@@ -123,16 +136,24 @@ function readNativeRows(): Promise<WindowsProcessRow[]> {
|
||||
// would restore exactly the fan-out it exists to prevent.
|
||||
const flags = native.ProcessDataFlag.Memory | native.ProcessDataFlag.CommandLine
|
||||
return new Promise((resolve, reject) => {
|
||||
// Hoisted so a synchronous throw from getAllProcesses can clear it. An
|
||||
// orphaned timer would otherwise fire later and wedge a reader that had
|
||||
// already recovered.
|
||||
let deadline: ReturnType<typeof setTimeout> | undefined
|
||||
try {
|
||||
const deadline = setTimeout(() => {
|
||||
wedgedUntilMs = Date.now() + WINDOWS_PROCESS_QUERY_COOLDOWN_MS
|
||||
deadline = setTimeout(() => {
|
||||
if (generation === readGeneration) {
|
||||
wedgedUntilMs = Date.now() + WINDOWS_PROCESS_QUERY_COOLDOWN_MS
|
||||
}
|
||||
reject(new Error('windows process table timed out'))
|
||||
}, WINDOWS_PROCESS_QUERY_TIMEOUT_MS)
|
||||
deadline.unref?.()
|
||||
native.getAllProcesses((processes) => {
|
||||
clearTimeout(deadline)
|
||||
// A late callback proves the reader recovered, so stop refusing.
|
||||
wedgedUntilMs = 0
|
||||
// A callback proves the reader is answering, so stop refusing.
|
||||
if (generation === readGeneration) {
|
||||
wedgedUntilMs = 0
|
||||
}
|
||||
if (!processes) {
|
||||
reject(new Error('windows process table returned no snapshot'))
|
||||
return
|
||||
@@ -159,6 +180,7 @@ function readNativeRows(): Promise<WindowsProcessRow[]> {
|
||||
)
|
||||
}, flags)
|
||||
} catch (error) {
|
||||
clearTimeout(deadline)
|
||||
reject(error instanceof Error ? error : new Error(String(error)))
|
||||
}
|
||||
})
|
||||
|
||||
@@ -248,6 +248,13 @@ export function runProcess(spec: ProcessSpec): Promise<ProcessResult> {
|
||||
// Why close rather than leave open: a child that reads stdin (a hook
|
||||
// draining its payload, a CLI probing for a TTY) otherwise blocks until the
|
||||
// timeout instead of seeing EOF immediately.
|
||||
// Why an error listener and not just end(): a child that exits without
|
||||
// reading makes the queued write fail with EPIPE, and an unhandled error on
|
||||
// a stream is an uncaught exception -- which takes the whole main process
|
||||
// down. The child's own error listener does not cover this stream. The
|
||||
// exit code already carries the outcome, so failing to deliver stdin is
|
||||
// not separately interesting.
|
||||
child.stdin?.on('error', () => {})
|
||||
child.stdin?.end(spec.input)
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user