diff --git a/packages/ssh/ssh/src/helper-processes.ts b/packages/ssh/ssh/src/helper-processes.ts index 3b33e14c75..bdf864f2f7 100644 --- a/packages/ssh/ssh/src/helper-processes.ts +++ b/packages/ssh/ssh/src/helper-processes.ts @@ -99,6 +99,14 @@ export class RemoteProcesses { private readonly limit: number, private readonly preparationMs: number, ) {} + /** Log one spill failure; the remote caller then receives the tail without a spill path. */ + private readonly reportSpillFailure = (error: unknown, label: string): void => { + this.ctx.logger.error( + `ssh helper could not write the complete ${label} stream to its spill file; the remote result keeps only the in-memory tail and reports no full-output path.`, + error, + ) + } + /** * Allocate private stream listeners; no target executes until start(). * @param raw - untrusted process request received over SSH. @@ -217,7 +225,10 @@ export class RemoteProcesses { const mode = stdio[name] const socket = await (record.endpoints[name] as Endpoint).connected if (typeof mode === 'object') { - const collector = new OutputCollector(mode.maxBytes, mode.spill?.maxBytes, name, prepareManagedProcessBinding().spillDir) + const binding = prepareManagedProcessBinding({ onSpillFailure: this.reportSpillFailure }) + const collector = new OutputCollector(mode.maxBytes, name, mode.spill === undefined ? undefined : { + maxBytes: mode.spill.maxBytes, dir: binding.spillDir, onFailure: binding.onSpillFailure, + }) collectors[name] = collector const forwarder = new CollectedOutputForwarder(socket, collector, mode.maxBytes) forwarders.push(forwarder) diff --git a/packages/ssh/subprocess-ssh/src/index.ts b/packages/ssh/subprocess-ssh/src/index.ts index b95908c8cd..9e8476594b 100644 --- a/packages/ssh/subprocess-ssh/src/index.ts +++ b/packages/ssh/subprocess-ssh/src/index.ts @@ -52,7 +52,7 @@ class RemoteProcess implements SubprocessHandle { const collect = (name: 'stdout' | 'stderr', stream: PassThrough, mode: SubprocessOutputMode) => { if (mode === 'pipe') return undefined if (mode === 'inherit') { stream.pipe(name === 'stdout' ? process.stdout : process.stderr, { end: false }); return undefined } - let collector = new OutputCollector(mode.maxBytes, undefined, name, '') + let collector = new OutputCollector(mode.maxBytes, name, undefined) let base = 0 let total = 0 let finalized = false @@ -62,7 +62,7 @@ class RemoteProcess implements SubprocessHandle { if (finalized) return if (snapshot.totalBytes < total) throw new Error('SSH helper rewound collected output') finalized = final - collector = new OutputCollector(mode.maxBytes, undefined, name, '') + collector = new OutputCollector(mode.maxBytes, name, undefined) collector.push(bytes) base = snapshot.totalBytes - bytes.length total = snapshot.totalBytes diff --git a/packages/subprocess/subprocess-local/README.i18n.yaml b/packages/subprocess/subprocess-local/README.i18n.yaml index d4858f0dad..52a8c84a8e 100644 --- a/packages/subprocess/subprocess-local/README.i18n.yaml +++ b/packages/subprocess/subprocess-local/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/subprocess/subprocess-local/README.md -README.md: 24480565de2f324a451dfdb40101eb50703b30b6 -README.zh.md: c86a7eb930834d960fa25dd7b5d175af63d0b31a +README.md: 0e5423aecd46a18cde79437ca5707be5ee58d59b +README.zh.md: 32a1b937e5768ad87d72914fb6c8a90a33c5899f diff --git a/packages/subprocess/subprocess-local/README.md b/packages/subprocess/subprocess-local/README.md index 24480565de..0e5423aecd 100644 --- a/packages/subprocess/subprocess-local/README.md +++ b/packages/subprocess/subprocess-local/README.md @@ -44,7 +44,7 @@ Windows ordinary subprocesses start the private Job runner with `windowsHide` an ### Collecting output -Collect mode keeps the last `maxBytes` of a stream in memory — errors and final results cluster at the end — and, when a `spill` cap is configured, appends the complete stream to a private file under a per-process directory in the OS temp dir (a `0700` directory, `0600` random-named files). A stream larger than the spill cap discards its incomplete spill and returns only the marked truncated tail. Reads are offset-based and non-consuming, so background and batch readers coexist before and after exit. +Collect mode keeps the last `maxBytes` of a stream in memory — errors and final results cluster at the end — and, when a `spill` cap is configured, appends the complete stream to a private file under a per-process directory in the OS temp dir (a `0700` directory, `0600` random-named files). A stream larger than the spill cap discards its incomplete spill and returns only the marked truncated tail. Spilling is best-effort: when the spill file cannot be opened or appended (the per-process directory removed by a temporary-file cleaner, `EACCES`, `EMFILE`, `ENOSPC`), the collector discards the spill, logs one `error` through the plugin logger, and keeps collecting the in-memory tail, so the result is truncated with no spill path. Reads are offset-based and non-consuming, so background and batch readers coexist before and after exit. The `./output` export shares this collector and retained-spill storage with process adapters. `snapshot()` returns the retained raw bytes and total byte count, allowing remote adapters to preserve offsets without forwarding the complete stream. @@ -109,7 +109,7 @@ A spawn synchronously validates the final argv, cwd, and environment, selects co ### Safety invariants -Spill files are opened `0600` with `O_EXCL` and random names under a `0700` per-process directory, defeating symlink planting in shared temp dirs; a failed final close withholds the spill path. Fallback process identities carry start times, so cleanup never follows PID reuse. A selected native failure is reported instead of replaying argv through fallback, and a range is removed from the live set only after cleanup completes or the failure remains observable. Host-exit finalization creates no promises or timers, preserves the host exit code and diagnostic, contains each target's failure, and does not claim quiescence. +Spill files are opened `0600` with `O_EXCL` and random names under a `0700` per-process directory, defeating symlink planting in shared temp dirs; a failed open, append, or final close withholds the spill path and never interrupts collection, because collection runs inside the stream's `'data'` listener where a thrown error would kill the host process. Fallback process identities carry start times, so cleanup never follows PID reuse. A selected native failure is reported instead of replaying argv through fallback, and a range is removed from the live set only after cleanup completes or the failure remains observable. Host-exit finalization creates no promises or timers, preserves the host exit code and diagnostic, contains each target's failure, and does not claim quiescence. @@ -155,6 +155,7 @@ These limits define when the provider is a poor fit or needs special operational - **In-process cleanup requires a JavaScript-observable exit** — direct `process.exit()`, default uncaught exceptions, and default unhandled rejections emit Node's synchronous `exit` event. The default OS disposition for an unhandled `SIGTERM`, `SIGINT`, or `SIGHUP` bypasses that event; an application covers those signals only by installing a handler that performs normal disposal or calls `process.exit()`. `SIGKILL`, fatal OOM, `process.abort()`, native crashes, power loss, and any failure that cannot run JavaScript require an external supervisor, container init, or equivalent OS owner. - **The credential scrub is a name heuristic** — `*KEY*`/`*PASSWORD*`/`*SECRET*`/`*TOKEN*` only; differently named secrets (for example `*PASSPHRASE*`) pass through, and a whitelist for over-scrubbed variables is noted future work. - **Completed spill files are not deleted** — bounded full-output recovery files accumulate under the OS tmpdir until something external cleans them; the private per-process spill directory is removed at a JavaScript-observable exit only when it holds no completed spill file. +- **A removed spill directory is not recreated** — the private per-process directory is created once; after an external cleaner removes it, every later spill in that process degrades to the in-memory tail with an `error` log until the host restarts. Recreating a fresh random directory on `ENOENT` is deferred work. ### Dev Note diff --git a/packages/subprocess/subprocess-local/README.zh.md b/packages/subprocess/subprocess-local/README.zh.md index c86a7eb930..32a1b937e5 100644 --- a/packages/subprocess/subprocess-local/README.zh.md +++ b/packages/subprocess/subprocess-local/README.zh.md @@ -44,7 +44,7 @@ Windows 普通子进程通过 `windowsHide` 启动私有 Job runner,并为原 ### 收集输出 -收集模式在内存中保留一条流的最后 `maxBytes`——错误与最终结果通常聚集在末尾——并在配置了 `spill` 上限时把完整流追加到 OS 临时目录下每进程目录中的私有文件(`0700` 目录、`0600` 随机命名文件)。某条流大于 spill 上限时,会丢弃不完整的 spill,只返回带截断标记的尾部。读取基于偏移量且从不消费,因此后台读取与批量读取在退出前后都可以共存。 +收集模式在内存中保留一条流的最后 `maxBytes`——错误与最终结果通常聚集在末尾——并在配置了 `spill` 上限时把完整流追加到 OS 临时目录下每进程目录中的私有文件(`0700` 目录、`0600` 随机命名文件)。某条流大于 spill 上限时,会丢弃不完整的 spill,只返回带截断标记的尾部。spill 是尽力而为:当 spill 文件无法打开或追加(每进程目录被临时文件清理工具删除、`EACCES`、`EMFILE`、`ENOSPC`)时,收集器丢弃该 spill,通过插件 logger 记录一条 `error`,并继续收集内存尾部,因此结果带截断标记且没有 spill 路径。读取基于偏移量且从不消费,因此后台读取与批量读取在退出前后都可以共存。 `./output` 导出向进程适配器共享该收集器与保留 spill 的存储。`snapshot()` 返回保留的原始字节及总字节数,使远程适配器能够保留偏移量,而无需转发完整的流。 @@ -109,7 +109,7 @@ Linux 普通进程和终端进程即使在 bootstrap 消费启动请求前被取 ### 安全不变式 -spill 文件以 `0600` 权限、`O_EXCL` 与随机名称在 `0700` 每进程目录下创建,可抵御共享临时目录中的符号链接植入;最终关闭失败时不公布 spill 路径。fallback 进程身份携带启动时间,因此清理绝不会跟随 PID 复用。选定的 native 路径失败时会报告错误,而不会通过 fallback 重放 argv;受管范围只有在清理完成后才从存活集合移除,否则失败仍保持可观察。宿主退出最终清理不创建 Promise 或定时器,保留宿主退出码与诊断,分别包含每个目标的失败,也不会声称已经完全停稳。 +spill 文件以 `0600` 权限、`O_EXCL` 与随机名称在 `0700` 每进程目录下创建,可抵御共享临时目录中的符号链接植入;打开、追加或最终关闭失败时不公布 spill 路径,且绝不中断收集,因为收集运行在流的 `'data'` 监听器内,抛出的错误会杀死宿主进程。fallback 进程身份携带启动时间,因此清理绝不会跟随 PID 复用。选定的 native 路径失败时会报告错误,而不会通过 fallback 重放 argv;受管范围只有在清理完成后才从存活集合移除,否则失败仍保持可观察。宿主退出最终清理不创建 Promise 或定时器,保留宿主退出码与诊断,分别包含每个目标的失败,也不会声称已经完全停稳。 @@ -155,6 +155,7 @@ spill 文件以 `0600` 权限、`O_EXCL` 与随机名称在 `0700` 每进程目 - **进程内清理要求退出阶段仍能执行 JavaScript**——直接 `process.exit()`、默认未捕获异常和默认未处理 rejection 会发出 Node 同步 `exit` 事件。未安装 handler 时,`SIGTERM`、`SIGINT` 或 `SIGHUP` 的默认 OS 处置不会发出该事件;应用只有安装执行正常 dispose 或调用 `process.exit()` 的 handler 才能覆盖这些信号。`SIGKILL`、fatal OOM、`process.abort()`、native crash、断电,以及任何无法运行 JavaScript 的故障,都需要外部 supervisor、容器 init 或等价的 OS owner 负责。 - **凭据清除依赖名称启发式规则**——只匹配 `*KEY*`/`*PASSWORD*`/`*SECRET*`/`*TOKEN*`;名称不同的 secret(例如 `*PASSPHRASE*`)会继续传递,对误删变量引入白名单属于已记录的后续工作。 - **不会删除已完成的 spill 文件**——有界的完整输出恢复文件会在 OS tmpdir 下累积,直到外部机制进行清理;每进程私有 spill 目录仅在未持有任何已完成 spill 文件时于 JavaScript 可观察的退出阶段删除。 +- **被删除的 spill 目录不会重建**——每进程私有目录只创建一次;被外部清理工具删除后,该进程内之后的每次 spill 都降级为内存尾部并记录一条 `error`,直到宿主重启。在 `ENOENT` 时重新创建一个新的随机目录是待办工作。 ### 开发备注 diff --git a/packages/subprocess/subprocess-local/src/index.ts b/packages/subprocess/subprocess-local/src/index.ts index ff03daebdf..c106b071b8 100644 --- a/packages/subprocess/subprocess-local/src/index.ts +++ b/packages/subprocess/subprocess-local/src/index.ts @@ -84,6 +84,19 @@ export class LocalSubprocessRuntime extends SubprocessRuntime { }, 'local subprocess teardown') } + /** + * Log one spill failure through the plugin logger. The collector keeps only + * its in-memory tail afterwards, so the model sees a truncated result with + * no spill path; the log line is the only trace of why. + */ + private readonly reportSpillFailure = (error: unknown, label: string): void => { + this.ctx.logger.error( + `subprocess-local could not write the complete ${label} stream to its spill file; the tool result keeps only the in-memory tail and reports no full-output path. ` + + 'A removed private spill directory under the OS temp dir (ENOENT) points at a temporary-file cleaner.', + error, + ) + } + private terminateForHostExit(): void { for (const handle of this.live) { try { @@ -181,10 +194,11 @@ export class LocalSubprocessRuntime extends SubprocessRuntime { const env = targetEnvironment(spec) const containmentMode = this.selectContainmentMode('ordinary') let handle: LocalSubprocessHandle + const internals: SpawnInternals = { ...this.internals, onSpillFailure: this.reportSpillFailure } if (containmentMode === 'fallback') { - handle = spawnSubprocess(spec, this.internals) + handle = spawnSubprocess(spec, internals) } else { - const binding = prepareManagedProcessBinding(this.internals) + const binding = prepareManagedProcessBinding(internals) const launch = containmentMode === 'linux-scope' ? launchLinuxScope(spec, env) : launchWindowsJob(spec, env) diff --git a/packages/subprocess/subprocess-local/src/output.ts b/packages/subprocess/subprocess-local/src/output.ts index 2cd3a1d082..f94c11480e 100644 --- a/packages/subprocess/subprocess-local/src/output.ts +++ b/packages/subprocess/subprocess-local/src/output.ts @@ -5,16 +5,38 @@ import { join } from 'node:path' import { tmpdir } from 'node:os' import type { CollectedOutput } from '@deepseek-ai/dsh-subprocess' +/** + * Receives one spill failure so the owner can log it through its own logger. + * Called at most once per collector, after the spill has been discarded and + * the in-memory tail has kept collecting. + * @param error - the `node:fs` failure from opening or appending the spill file. + * @param label - the stream label of the collector that failed. + */ +export type SpillFailureReporter = (error: unknown, label: string) => void + +/** Spill storage for one collected stream; `undefined` means tail-only collection. */ +export interface SpillOptions { + /** Whole-stream byte cap beyond which an incomplete spill is discarded. */ + maxBytes: number + /** Private directory receiving the spill file. */ + dir: string + /** Owner-side report of a spill open or write failure. */ + onFailure: SpillFailureReporter +} + let spillCounter = 0 let defaultSpillDir: string | undefined /** * The default spill location: a private (0700) per-process directory under * the OS tmpdir, created lazily. Predictable world-readable paths would let - * other local users read command output or pre-create symlinks. At a - * JavaScript-observable process exit the directory is removed only when it - * holds no completed spill file (spill files are retained as full-output - * recovery artifacts until an external cleanup). + * other local users read command output or pre-create symlinks. The directory + * is created once per process and never recreated: when an external + * temporary-file cleaner removes it while empty, the next spill open fails and + * the collector degrades to its in-memory tail. At a JavaScript-observable + * process exit the directory is removed only when it holds no completed spill + * file (spill files are retained as full-output recovery artifacts until an + * external cleanup). */ function privateSpillDir(): string { defaultSpillDir ??= mkdtempSync(join(tmpdir(), 'dsh-subprocess-')) @@ -32,25 +54,55 @@ process.once('exit', () => { try { rmdirSync(defaultSpillDir) } catch { /* best-effort: ENOENT/ENOTEMPTY/EBUSY/EPERM must not change the exit code. */ } }) +/** + * The stderr reporter used when no owner supplies one: a bare + * {@link prepareManagedProcessBinding} caller has no plugin logger, and the + * failure must still reach the process diagnostics. + * @param error - the spill failure. + * @param label - the failed stream label. + */ +function reportSpillFailureToStderr(error: unknown, label: string): void { + process.stderr.write(`dsh-subprocess-local: ${label} spill failed; only the in-memory tail is retained: ${String(error)}\n`) +} + +/** Inputs a managed native process needs before its output streams are bound. */ +export interface ManagedProcessBinding { + /** Directory receiving spill files. */ + spillDir: string + /** Receives a spill open or write failure. */ + onSpillFailure: SpillFailureReporter +} + /** * Prepare fallible output storage before starting a managed native process. - * @param internals - optional caller-owned spill directory. + * This is the explicit resolve step for spill inputs: the spill directory + * defaults to the private per-process directory, and the failure reporter + * defaults to a stderr line when the caller has no logger. + * @param internals - optional caller-owned spill directory and failure reporter. * @returns binding inputs whose spill directory is ready for use. */ export function prepareManagedProcessBinding( - internals: { spillDir?: string } = {}, -): { spillDir: string } { - return { spillDir: internals.spillDir ?? privateSpillDir() } + internals: { spillDir?: string; onSpillFailure?: SpillFailureReporter } = {}, +): ManagedProcessBinding { + return { + spillDir: internals.spillDir ?? privateSpillDir(), + onSpillFailure: internals.onSpillFailure ?? reportSpillFailureToStderr, + } } /** - * Collects one stream with a bounded in-memory tail. With a spill cap, on + * Collects one stream with a bounded in-memory tail. With spill options, on * first overflow a spill file is created and every chunk (including those * already collected) is appended there while the full stream remains within - * the cap; without one, only the in-memory tail is ever retained (the + * the cap; without them, only the in-memory tail is ever retained (the * diagnostic-tail shape — a language server's stderr). * + * Spilling is best-effort: a spill open or write failure discards the spill, + * reports once through {@link SpillOptions.onFailure}, and never interrupts + * in-memory collection, because `push()` runs inside the stream's `'data'` + * listener where a thrown error would become an uncaught exception. + * * Tail-keep rationale (pi/OpenCode): errors and final results cluster at the * end of command output; the spill file covers the head. */ @@ -64,13 +116,17 @@ export class OutputCollector { /** Total bytes ever pushed (not just retained). */ private total = 0 + /** + * @param maxBytes - in-memory tail cap in bytes. + * @param label - stream label used in spill file names and failure reports. + * @param spill - spill storage; omit for tail-only collection. + */ constructor( private readonly maxBytes: number, - private readonly maxSpillBytes: number | undefined, private readonly label: string, - private readonly spillDir: string, + private readonly spill: SpillOptions | undefined, ) { - this.spillDisabled = maxSpillBytes === undefined + this.spillDisabled = spill === undefined } /** @@ -84,7 +140,8 @@ export class OutputCollector { push(chunk: Buffer): void { this.total += chunk.length const overflows = this.bytes + chunk.length > this.maxBytes - if (!this.spillDisabled && (overflows || this.spillFd !== undefined)) this.spillAll(chunk) + const spill = this.spill + if (spill !== undefined && !this.spillDisabled && (overflows || this.spillFd !== undefined)) this.spillAll(spill, chunk) this.chunks.push(chunk) this.bytes += chunk.length while (this.bytes > this.maxBytes) { @@ -105,24 +162,37 @@ export class OutputCollector { } } - /** Open the spill file lazily and append `chunk` (and any prior chunks once). */ - private spillAll(chunk: Buffer): void { - if (this.maxSpillBytes !== undefined && this.total > this.maxSpillBytes) { + /** + * Open the spill file lazily and append `chunk` (and any prior chunks once). + * Runs inside the stream's `'data'` listener, so every filesystem failure is + * contained here: the spill is discarded, reported once, and collection + * continues with the in-memory tail alone. + */ + private spillAll(spill: SpillOptions, chunk: Buffer): void { + if (this.total > spill.maxBytes) { this.discardSpill() return } - if (this.spillFd === undefined) { - // Random suffix + O_EXCL + no-follow-equivalent ('wx' fails on any - // existing path, symlink or not) + owner-only mode: defeats spill-path - // prediction and symlink planting in shared tmp dirs. - this.spillFile = join( - this.spillDir, - `dsh-subprocess-${process.pid}-${++spillCounter}-${randomBytes(6).toString('hex')}-${this.label}.log`, - ) - this.spillFd = openSync(this.spillFile, 'wx', 0o600) - for (const prior of this.chunks) writeSync(this.spillFd, prior) + try { + if (this.spillFd === undefined) { + // Random suffix + O_EXCL + no-follow-equivalent ('wx' fails on any + // existing path, symlink or not) + owner-only mode: defeats spill-path + // prediction and symlink planting in shared tmp dirs. + this.spillFile = join( + spill.dir, + `dsh-subprocess-${process.pid}-${++spillCounter}-${randomBytes(6).toString('hex')}-${this.label}.log`, + ) + this.spillFd = openSync(this.spillFile, 'wx', 0o600) + for (const prior of this.chunks) writeSync(this.spillFd, prior) + } + writeSync(this.spillFd, chunk) + } catch (error) { + // ENOENT (spill directory removed by a temp cleaner), EACCES/EPERM, + // EMFILE, or ENOSPC: the spill file is a recovery aid, not a + // precondition of collection. + this.discardSpill() + spill.onFailure(error, this.label) } - writeSync(this.spillFd, chunk) } /** Stop spilling and remove the file once it can no longer hold the complete stream. */ diff --git a/packages/subprocess/subprocess-local/src/spawn.ts b/packages/subprocess/subprocess-local/src/spawn.ts index 9862a0d0dd..ed9baa2020 100644 --- a/packages/subprocess/subprocess-local/src/spawn.ts +++ b/packages/subprocess/subprocess-local/src/spawn.ts @@ -25,7 +25,7 @@ import { waitWithAbort } from './managed-owner.ts' import { controlEnvironment, controlPipe } from './control-spawn.ts' import { SUBPROCESS_CONTROL_FD } from '@deepseek-ai/dsh-subprocess/control' import { linuxProcessGroupHasLiveMembers } from './process-inspector.ts' -import { OutputCollector, prepareManagedProcessBinding } from './output.ts' +import { OutputCollector, prepareManagedProcessBinding, type SpillFailureReporter } from './output.ts' type SpawnProcess = ( program: string, @@ -59,6 +59,8 @@ export interface SpawnInternals { spawn?: SpawnProcess /** Directory for spill files (defaults to the OS temp dir). */ spillDir?: string + /** Receives spill open/write failures; the runtime supplies its plugin logger, bare callers get a stderr line. */ + onSpillFailure?: SpillFailureReporter /** Windows tree-termination runner (defaults to `taskkill /PID /T /F`). */ taskkill?: (pid: number) => void /** Host platform override for signalling decisions. */ @@ -259,15 +261,15 @@ function fallbackOwner( * Bind platform launch facts to the existing stdio, outcome, abort, and termination lifecycle. * @param spec - fully resolved argv, cwd, stdio, grace, cancellation, environment. * @param launch - platform streams, direct outcome, and managed-range owner. - * @param internals - test-only spill-directory override. + * @param internals - spill-directory override and spill failure reporter. * @returns live subprocess handle. */ export function bindManagedProcess( spec: SubprocessSpawnSpec, launch: ManagedProcessLaunch, - internals: Pick = {}, + internals: Pick = {}, ): LocalSubprocessHandle { - const { spillDir } = prepareManagedProcessBinding(internals) + const { spillDir, onSpillFailure } = prepareManagedProcessBinding(internals) const { stdin, stdout, stderr } = launch const isCollect = (mode: SubprocessOutputMode): mode is SubprocessCollect => @@ -278,7 +280,11 @@ export function bindManagedProcess( const collectStream = (mode: SubprocessOutputMode, stream: Readable | null, label: string): OutputCollector | undefined => { if (!isCollect(mode) || stream === null) return undefined - const collector = new OutputCollector(mode.maxBytes, mode.spill?.maxBytes, label, spillDir) + const collector = new OutputCollector( + mode.maxBytes, + label, + mode.spill === undefined ? undefined : { maxBytes: mode.spill.maxBytes, dir: spillDir, onFailure: onSpillFailure }, + ) stream.on('data', (chunk: Buffer) => { collector.push(chunk) }) return collector } diff --git a/packages/subprocess/subprocess-local/tests/spawn.spec.ts b/packages/subprocess/subprocess-local/tests/spawn.spec.ts index d57e66a6cd..22c77f5d52 100644 --- a/packages/subprocess/subprocess-local/tests/spawn.spec.ts +++ b/packages/subprocess/subprocess-local/tests/spawn.spec.ts @@ -1,6 +1,6 @@ -import { OutputCollector } from '../src/output.ts' +import { OutputCollector, type SpillOptions } from '../src/output.ts' import { spawn as nodeSpawn, spawnSync as nodeSpawnSync } from 'node:child_process' -import { mkdtempSync, readFileSync, rmSync, statSync, unlinkSync } from 'node:fs' +import { mkdtempSync, readFileSync, rmSync, statSync, unlinkSync, writeFileSync } from 'node:fs' import { tmpdir } from 'node:os' import { dirname, join } from 'node:path' import { afterAll, describe, expect, it, vi } from 'vitest' @@ -64,9 +64,10 @@ function shellArgv(command: string): string[] { } } -const { failNextClose, failNextUnlink } = vi.hoisted(() => ({ +const { failNextClose, failNextUnlink, failNextWrite } = vi.hoisted(() => ({ failNextClose: { value: false }, failNextUnlink: { value: false }, + failNextWrite: { value: false }, })) vi.mock('node:fs', async (importOriginal) => { const actual = await importOriginal() @@ -86,11 +87,24 @@ vi.mock('node:fs', async (importOriginal) => { } actual.unlinkSync(path) }, + writeSync(...args: Parameters): number { + if (failNextWrite.value) { + failNextWrite.value = false + throw Object.assign(new Error('simulated ENOSPC on write'), { code: 'ENOSPC' }) + } + return actual.writeSync(...args) + }, } }) const spillDir = mkdtempSync(join(tmpdir(), 'dsh-subprocess-spec-')) +/** Spill options for one collector test; failures are collected for assertions. */ +function spillOptions(maxBytes: number, dir = spillDir): { options: SpillOptions; failures: { error: unknown; label: string }[] } { + const failures: { error: unknown; label: string }[] = [] + return { options: { maxBytes, dir, onFailure: (error, label) => { failures.push({ error, label }) } }, failures } +} + /** The per-process default spill dir captured by the default-spill test. */ let defaultSpillDir: string | undefined @@ -516,7 +530,7 @@ describe('output truncation and spill', () => { describe('OutputCollector', () => { it('keeps the tail of a single oversized chunk', () => { - const collector = new OutputCollector(10, 100, 'test', spillDir) + const collector = new OutputCollector(10, 'test', spillOptions(100).options) collector.push(Buffer.from('0123456789abcdef')) const out = collector.finalize() expect(out.text).toBe('6789abcdef') @@ -527,7 +541,7 @@ describe('OutputCollector', () => { it('retains a byte-exact tail across uneven chunk boundaries', () => { // A diagnostic tail must be exactly the LAST maxBytes regardless of // chunking; dropping only whole chunks would under-retain. - const collector = new OutputCollector(10, undefined, 'exact-tail', spillDir) + const collector = new OutputCollector(10, 'exact-tail', undefined) collector.push(Buffer.from('aaaa')) collector.push(Buffer.from('bbbbbb')) collector.push(Buffer.from('cc')) @@ -538,7 +552,7 @@ describe('OutputCollector', () => { }) it('readFrom returns increments and flags lossy reads', () => { - const collector = new OutputCollector(10, 100, 'test', spillDir) + const collector = new OutputCollector(10, 'test', spillOptions(100).options) collector.push(Buffer.from('aaaaa')) const first = collector.readFrom(0) expect(first.text).toBe('aaaaa') @@ -559,7 +573,7 @@ describe('OutputCollector', () => { }) it('contains close failures and drops the spill path', () => { - const collector = new OutputCollector(4, 100, 'closefail', spillDir) + const collector = new OutputCollector(4, 'closefail', spillOptions(100).options) collector.push(Buffer.from('aaaa')) collector.push(Buffer.from('bbbb')) expect(collector.readFrom(0).spillPath).toBeDefined() @@ -575,7 +589,7 @@ describe('OutputCollector', () => { }) it('discards a spill that exceeds its configured cap', () => { - const collector = new OutputCollector(4, 8, 'bounded', spillDir) + const collector = new OutputCollector(4, 'bounded', spillOptions(8).options) collector.push(Buffer.from('aaaa')) collector.push(Buffer.from('bbbb')) const spillPath = collector.readFrom(0).spillPath! @@ -591,7 +605,7 @@ describe('OutputCollector', () => { }) it('does not create a spill when the first overflowing chunk exceeds the cap', () => { - const collector = new OutputCollector(4, 4, 'no-spill', spillDir) + const collector = new OutputCollector(4, 'no-spill', spillOptions(4).options) collector.push(Buffer.from('abcdefgh')) const out = collector.finalize() expect(out.text).toBe('efgh') @@ -600,7 +614,7 @@ describe('OutputCollector', () => { }) it('contains cleanup failures while disabling an oversize spill', () => { - const collector = new OutputCollector(4, 8, 'cleanup-fail', spillDir) + const collector = new OutputCollector(4, 'cleanup-fail', spillOptions(8).options) collector.push(Buffer.from('aaaa')) collector.push(Buffer.from('bbbb')) const spillPath = collector.readFrom(0).spillPath! @@ -613,6 +627,58 @@ describe('OutputCollector', () => { expect(collector.finalize().spillPath).toBeUndefined() unlinkSync(spillPath) }) + + it('keeps collecting when the spill directory has been removed (ENOENT on open)', () => { + const removedDir = mkdtempSync(join(tmpdir(), 'dsh-subprocess-removed-')) + rmSync(removedDir, { recursive: true, force: true }) + const { options, failures } = spillOptions(100, removedDir) + const collector = new OutputCollector(4, 'enoent', options) + collector.push(Buffer.from('aaaa')) + expect(() => { collector.push(Buffer.from('bbbb')) }).not.toThrow() + collector.push(Buffer.from('cc')) + expect(failures).toHaveLength(1) + expect((failures[0]!.error as NodeJS.ErrnoException).code).toBe('ENOENT') + expect(failures[0]!.label).toBe('enoent') + const out = collector.finalize() + expect(out.text).toBe('bbcc') + expect(out.truncated).toBe(true) + expect(out.spillPath).toBeUndefined() + }) + + it('keeps collecting when the spill directory is a file (ENOTDIR on open)', () => { + const fileAsDir = join(spillDir, `not-a-dir-${Date.now()}`) + writeFileSync(fileAsDir, '') + const { options, failures } = spillOptions(100, fileAsDir) + const collector = new OutputCollector(4, 'enotdir', options) + collector.push(Buffer.from('aaaa')) + expect(() => { collector.push(Buffer.from('bbbb')) }).not.toThrow() + expect(failures).toHaveLength(1) + expect(['ENOTDIR', 'ENOENT']).toContain((failures[0]!.error as NodeJS.ErrnoException).code) + expect(collector.finalize()).toEqual({ text: 'bbbb', truncated: true }) + }) + + it('withdraws a spill whose append fails after the file exists (ENOSPC on write)', () => { + const { options, failures } = spillOptions(100) + const collector = new OutputCollector(4, 'enospc', options) + collector.push(Buffer.from('aaaa')) + collector.push(Buffer.from('bbbb')) + const spillPath = collector.readFrom(0).spillPath! + expect(readFileSync(spillPath, 'utf8')).toBe('aaaabbbb') + + failNextWrite.value = true + expect(() => { collector.push(Buffer.from('cccc')) }).not.toThrow() + expect(failNextWrite.value).toBe(false) + expect(failures).toHaveLength(1) + expect((failures[0]!.error as NodeJS.ErrnoException).code).toBe('ENOSPC') + expect(() => readFileSync(spillPath)).toThrow() + + collector.push(Buffer.from('dd')) + expect(failures).toHaveLength(1) + const out = collector.finalize() + expect(out.text).toBe('ccdd') + expect(out.truncated).toBe(true) + expect(out.spillPath).toBeUndefined() + }) }) describe('killGroup', () => {