fix(ptc): retain per-call cleanup until process exit is observed
This commit is contained in:
1 parent
49f7919762
commit
209684aaae
2 files changed
+33
-2
No files matched your search
@@ -137,6 +137,17 @@ export class NodePtcRuntime extends PtcRuntime {
|
||||
let parsing = true
|
||||
const wallTimer = spec.timeoutMs === null ? undefined
|
||||
: setTimeout(() => { timedOut = true; controller.abort('execution deadline reached') }, spec.timeoutMs)
|
||||
let stopping: Promise<boolean> | undefined
|
||||
let unregister: (() => void) | undefined
|
||||
const stop = (): Promise<boolean> => {
|
||||
if (stopping !== undefined) return stopping
|
||||
const cleanup = handle === undefined ? Promise.resolve(true) : terminateProcess(handle, this.config.graceMs)
|
||||
stopping = cleanup.then((drained) => { unregister?.(); return drained }, (error: unknown) => {
|
||||
stopping = undefined
|
||||
throw error
|
||||
})
|
||||
return stopping
|
||||
}
|
||||
const finish = (failure?: PtcRunFailure, value?: PtcJsonValue): void => {
|
||||
if (settled) return
|
||||
settled = true
|
||||
@@ -144,9 +155,9 @@ export class NodePtcRuntime extends PtcRuntime {
|
||||
signal.removeEventListener('abort', onAbort)
|
||||
channel?.close()
|
||||
void (async () => {
|
||||
if (handle !== undefined) {
|
||||
if (handle !== undefined || unregister !== undefined) {
|
||||
try {
|
||||
const drained = await terminateProcess(handle, this.config.graceMs)
|
||||
const drained = await stop()
|
||||
if (!drained && failure === undefined) {
|
||||
failure = { kind: 'worker-exit', message: 'Node process output did not close cleanly' }
|
||||
}
|
||||
@@ -200,6 +211,11 @@ export class NodePtcRuntime extends PtcRuntime {
|
||||
env.DSH_PTC_RUNTIME_NODE = '1'
|
||||
env.NODE_OPTIONS = heapFlag
|
||||
}
|
||||
if (policy.sessionId !== undefined) {
|
||||
unregister = this.ctx.sandboxPolicy.registerResource(policy.sessionId, {
|
||||
release: async () => { controller.abort(new Error('sandbox mode is changing')); await stop() },
|
||||
})
|
||||
}
|
||||
handle = this.ctx.subprocess.spawn({ argv: confined?.argv ?? argv, cwd: spec.cwd, env, stdio: { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe', control: 'pipe' }, graceMs: this.config.graceMs, signal })
|
||||
const launched = handle
|
||||
if (launched.control === undefined || launched.stdout === undefined || launched.stderr === undefined) {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { Duplex, PassThrough } from 'node:stream'
|
||||
import { setImmediate } from 'node:timers/promises'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import { describe, expect, it, onTestFinished, vi } from 'vitest'
|
||||
import type { PtcBindingFunction, PtcRunRequest } from '@deepseek-ai/dsh-ptc-runtime'
|
||||
@@ -515,3 +516,17 @@ describe('Node runtime host failures', () => {
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('keeps failed per-call cleanup registered until process observation succeeds', async () => {
|
||||
const h = await setup()
|
||||
const session = h.ctx.sessions.create(SessionId('per-call-cleanup-retry'))
|
||||
h.waitForExit.mockRejectedValue(new Error('range cannot be observed'))
|
||||
h.onBoot(() => { h.emit({ type: 'done', value: encodePtcJsonWire(42) }) })
|
||||
const result = await h.start({ ...request, sandboxPolicy: h.ctx.sandboxPolicy.resolve({ session }) })
|
||||
expect(result.error?.kind).toBe('worker-exit')
|
||||
await expect(h.ctx.sandboxPolicy.setMode(session, 'read-only')).rejects.toThrow('range cannot be observed')
|
||||
expect(h.ctx.sandboxPolicy.resolve({ session }).mode).toBe('danger-full-access')
|
||||
h.waitForExit.mockResolvedValue(true)
|
||||
expect(await h.ctx.sandboxPolicy.setMode(session, 'read-only')).toEqual([])
|
||||
expect(h.waitForExit).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
Reference in new issue
Block a user