From 4748c627aee2f3b34dd3dd63f6a2bbe073407cf9 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sun, 4 Oct 2026 21:20:52 +0200 Subject: [PATCH] feat(durable): bounded binary and directory readers, argv exec with stream info - FileSystem.openBinaryReader(): positional reads of one opened regular file, noFollow - FileSystem.openDirReader(): paged listings, vanished entries skipped - Shell.exec() accepts an argv array (no shell); onOutput reports the stream - Env conformance suite in /testing, run for NodeExecutionEnv - Work-branch workflow running env tests on Linux, macOS and Windows --- .github/workflows/durable-env.yml | 2 +- packages/durable/CHANGELOG.md | 12 + packages/durable/README.md | 13 + packages/durable/docs/spec.md | 19 ++ packages/durable/src/env/index.ts | 50 ++- packages/durable/src/env/node.ts | 290 +++++++++++++++--- .../durable/src/testing/env-conformance.ts | 243 +++++++++++++++ packages/durable/src/testing/index.ts | 7 +- packages/durable/src/testing/runner.ts | 16 +- packages/durable/src/testing/types.ts | 20 ++ .../durable/test/env-node-conformance.test.ts | 64 ++++ packages/durable/test/tools.test.ts | 4 +- 12 files changed, 690 insertions(+), 50 deletions(-) create mode 100644 packages/durable/src/testing/env-conformance.ts create mode 100644 packages/durable/test/env-node-conformance.test.ts diff --git a/.github/workflows/durable-env.yml b/.github/workflows/durable-env.yml index a33f0f3d5..68cc05b86 100644 --- a/.github/workflows/durable-env.yml +++ b/.github/workflows/durable-env.yml @@ -63,4 +63,4 @@ jobs: - name: Env tests working-directory: packages/durable - run: npx vitest --run test/env-node.test.ts test/env-node-spill.test.ts test/tools.test.ts + run: npx vitest --run test/env-node.test.ts test/env-node-spill.test.ts test/env-node-conformance.test.ts test/tools.test.ts diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index a1c9f44d9..a52502175 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -2,6 +2,18 @@ ## [Unreleased] +### Breaking Changes + +- `FileSystem` requires `openBinaryReader()` and `openDirReader()`; custom environments must implement them. +- `Shell.exec()` accepts an argv array besides a shell string, and `ShellExecOptions.onOutput` receives a third `info` argument naming the stream (`stdout` or `stderr`); custom environments must accept both forms and pass the stream. + +### Added + +- `openBinaryReader()` for bounded positional reads of one opened regular file, with `noFollow` to refuse a final-component symlink. +- `openDirReader()` for paged directory listings that read metadata only for returned entries. +- Argv form of `exec()`, which runs a program without a shell. +- `createEnvConformance()` and `registerEnvConformance()` in `@earendil-works/pi-durable/testing` for checking custom `ExecutionEnv` implementations. + ### Fixed - `NodeExecutionEnv.flushFile()` on a directory fails with `is_directory` on Windows, as on POSIX. diff --git a/packages/durable/README.md b/packages/durable/README.md index 521ffcaee..f13921d83 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -246,6 +246,19 @@ const harness = await Harness.open(storage, { A throw from `env` becomes the call's error result. Without an environment, the built-in tools fail with an error result. A fresh environment object per call is fine: `edit` and `write` serialize changes to one file by the environment's `id` and path. A custom `ExecutionEnv` sets `id` so that equal ids see the same files at the same paths, for example one id per container. +Hosts can use the environment directly too, for example to show a project's files. `openBinaryReader()` reads byte ranges of one opened file, `openDirReader()` pages a directory, and `exec()` with an argv array runs a program without a shell, reporting which stream each output chunk came from: + +```typescript +const status = { stdout: "", stderr: "" }; +await env.exec(["git", "status", "--porcelain=v2", "-z"], { + onOutput: (text, _context, { stream }) => { + status[stream] += text; + }, +}, context); +``` + +Abort the context to stop one call; `cleanup()` is for shutting the environment down. A custom environment can check itself with `registerEnvConformance()` from `@earendil-works/pi-durable/testing`, like storage below. + ## Reload Installing an extension with an installed name replaces it in place, in one step: diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 862129117..2b9c53f0c 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -2819,6 +2819,25 @@ sanitized, and throttled. The `bash` tool pipes those chunks into `output()`, reports the spill path as a diagnostic, and throws on a nonzero exit or timeout; the error result still carries the retained output and diagnostics. +`exec` takes a string or an argv array. A string runs through the +environment's shell. An array runs its first element directly with the rest as +arguments, without a shell, so a host that builds a command from data, such as +a file name, never quotes it for a particular shell. Each `onOutput` chunk names +the stream it came from; the `bash` tool ignores it, while a host that needs +stdout and stderr apart collects them separately, bounds them itself, and +aborts the call when it has enough. `openBinaryReader` opens one regular file +for positional reads, so a host reads a bounded range instead of the whole +file, and every read sees the file it opened even if the path is renamed; +`noFollow` refuses a symbolic link as the final path component. `openDirReader` +pages a directory in file-system order, reading metadata only for the entries +it returns and skipping entries removed meanwhile. All operations stop when +their context is aborted, and only that call's work stops: a timeout or abort +kills only that command's processes. `cleanup()` kills every command the +environment still runs and belongs to its owner's shutdown, never to a single +request. An environment is trusted, not a confinement boundary: checking a +canonical path before opening it does not prevent a concurrent rename or +symlink swap, so callers that restrict paths do so for hygiene, not security. + The `details()` promise resolves after the corresponding or coalesced document commit. During normal settlement the tool task stops its throttle and awaits the commit in flight; the terminal commit, which appends the result entry, is the diff --git a/packages/durable/src/env/index.ts b/packages/durable/src/env/index.ts index bb0177854..a9047c038 100644 --- a/packages/durable/src/env/index.ts +++ b/packages/durable/src/env/index.ts @@ -92,6 +92,26 @@ export interface TextLineReader { close(context: Context): Promise; } +/** Positional reads from one opened regular file; all calls see the same file even if its path is renamed. */ +export interface BinaryReader { + /** Metadata of the opened file, not of whatever its path names now. */ + info(context: Context): Promise>; + /** Up to `length` bytes at `offset`; fewer only at end of file. */ + read(offset: number, length: number, context: Context): Promise>; + close(context: Context): Promise; +} + +/** Pages of one directory's entries. */ +export interface DirReader { + /** + * Up to `maxEntries` entries in the order the file system returns them, continuing where the previous call stopped. + * `done` marks the end; it may come with the last entries or with an empty page. An entry that disappears before its + * metadata is read is skipped, as are entries of unsupported kinds. After a failed or aborted call, close the reader. + */ + next(maxEntries: number, context: Context): Promise>; + close(context: Context): Promise; +} + /** Portable filesystem capability. Operations return failures rather than throwing. */ export interface FileSystem { /** @@ -110,6 +130,16 @@ export interface FileSystem { context: Context, ): Promise>; readBinaryFile(path: string, context: Context): Promise>; + /** + * Open a regular file for bounded positional reads. A directory fails with `is_directory`, other non-regular files + * with `invalid`. With `noFollow`, a symbolic link as the final path component fails with `invalid` instead of being + * followed; earlier components are still resolved. + */ + openBinaryReader( + path: string, + options: { noFollow?: boolean } | undefined, + context: Context, + ): Promise>; writeFile(path: string, content: string | Uint8Array, context: Context): Promise>; appendFile(path: string, content: string | Uint8Array, context: Context): Promise>; /** Truncate or extend a file to exactly `size` bytes. */ @@ -119,6 +149,7 @@ export interface FileSystem { renameFile(sourcePath: string, destinationPath: string, context: Context): Promise>; fileInfo(path: string, context: Context): Promise>; listDir(path: string, context: Context): Promise>; + openDirReader(path: string, context: Context): Promise>; canonicalPath(path: string, context: Context): Promise>; exists(path: string, context: Context): Promise>; createDir( @@ -157,17 +188,30 @@ export interface ShellExecOptions { env?: Record; inheritEnv?: boolean; timeout?: number; - /** Every decoded chunk of combined stdout and stderr as it arrives: raw, unbounded, and unthrottled. */ - onOutput?: (text: string, context: Context) => void; + /** + * Every decoded chunk of stdout and stderr as it arrives, in arrival order, with the stream it came from: raw, + * unbounded, and unthrottled. Each stream is decoded separately, so a character split across chunks survives. + */ + onOutput?: (text: string, context: Context, info: ShellOutputInfo) => void; spill?: ShellSpillOptions; } +export interface ShellOutputInfo { + stream: "stdout" | "stderr"; +} + export interface Shell { + /** + * Run a command. A string runs through the environment's shell. An array runs `command[0]` directly with the rest + * as its arguments, without a shell, so they reach the program unparsed. Aborting the context or a timeout kills + * only this command's processes. + */ exec( - command: string, + command: string | readonly string[], options: ShellExecOptions | undefined, context: Context, ): Promise>; + /** Kill every command this environment still runs; for its owner's shutdown, never for one request. */ cleanup(context: Context): Promise; } diff --git a/packages/durable/src/env/node.ts b/packages/durable/src/env/node.ts index bbdcdf0c7..5ad7c7207 100644 --- a/packages/durable/src/env/node.ts +++ b/packages/durable/src/env/node.ts @@ -1,12 +1,14 @@ import { type ChildProcess, spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { constants, createWriteStream, type WriteStream } from "node:fs"; +import { constants, createWriteStream, type Dir, type Stats, type WriteStream } from "node:fs"; import { access, appendFile, + type FileHandle, lstat, mkdir, mkdtemp, + opendir, open as openFile, readdir, readFile, @@ -20,6 +22,8 @@ import { basename, isAbsolute, join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import type { Context } from "@earendil-works/chord"; import { + type BinaryReader, + type DirReader, type ExecutionEnv, ExecutionError, err, @@ -30,6 +34,7 @@ import { type Result, type ShellExecOptions, type ShellExecResult, + type ShellOutputInfo, type TextLine, type TextLineReader, toError, @@ -39,6 +44,8 @@ const MAX_TIMEOUT_MS = 2_147_483_647; const MAX_TIMEOUT_SECONDS = MAX_TIMEOUT_MS / 1000; const EXIT_STDIO_GRACE_MS = 100; const SPILL_HIGH_WATER_MARK = 1024 * 1024; +/** Largest single read of a `BinaryReader`, so a huge `length` allocates only as much as the file yields. */ +const BINARY_READ_CHUNK = 1024 * 1024; type SpillChunk = string | Uint8Array; @@ -434,6 +441,130 @@ class NodeTextLineReader implements TextLineReader { } } +function closedResult(what: string, path: string): Result { + return err(new FileError("invalid", `${what} is closed`, path)); +} + +class NodeBinaryReader implements BinaryReader { + private readonly file: FileHandle; + private readonly path: string; + private closed = false; + + constructor(file: FileHandle, path: string) { + this.file = file; + this.path = path; + } + + async info(context: Context): Promise> { + const aborted = abortResult(context.abortSignal, this.path); + if (aborted) return aborted; + if (this.closed) return closedResult("Binary reader", this.path); + try { + return fileInfoFromStats(this.path, await this.file.stat()); + } catch (error) { + return err(toFileError(error, this.path)); + } + } + + async read(offset: number, length: number, context: Context): Promise> { + const aborted = abortResult(context.abortSignal, this.path); + if (aborted) return aborted; + if (this.closed) return closedResult("Binary reader", this.path); + if (!Number.isSafeInteger(offset) || offset < 0 || !Number.isSafeInteger(length) || length < 0) { + return err(new FileError("invalid", "Offset and length must be non-negative safe integers", this.path)); + } + const chunks: Uint8Array[] = []; + let total = 0; + try { + while (total < length) { + const chunk = new Uint8Array(Math.min(length - total, BINARY_READ_CHUNK)); + const { bytesRead } = await this.file.read(chunk, 0, chunk.length, offset + total); + const afterReadAbort = abortResult(context.abortSignal, this.path); + if (afterReadAbort) return afterReadAbort; + if (bytesRead === 0) break; + chunks.push(bytesRead === chunk.length ? chunk : chunk.subarray(0, bytesRead)); + total += bytesRead; + } + } catch (error) { + return err(toFileError(error, this.path)); + } + if (chunks.length === 1) return ok(chunks[0]!); + const bytes = new Uint8Array(total); + let position = 0; + for (const chunk of chunks) { + bytes.set(chunk, position); + position += chunk.length; + } + return ok(bytes); + } + + async close(_context: Context): Promise { + if (this.closed) return; + this.closed = true; + await this.file.close().catch(() => undefined); + } +} + +class NodeDirReader implements DirReader { + private readonly dir: Dir; + private readonly path: string; + private done = false; + private closed = false; + + constructor(dir: Dir, path: string) { + this.dir = dir; + this.path = path; + } + + async next( + maxEntries: number, + context: Context, + ): Promise> { + const aborted = abortResult<{ entries: FileInfo[]; done: boolean }>(context.abortSignal, this.path); + if (aborted) return aborted; + if (this.closed) return closedResult("Directory reader", this.path); + if (!Number.isSafeInteger(maxEntries) || maxEntries <= 0) { + return err(new FileError("invalid", "maxEntries must be a positive safe integer", this.path)); + } + const entries: FileInfo[] = []; + try { + while (!this.done && entries.length < maxEntries) { + const entry = await this.dir.read(); + if (entry === null) { + this.done = true; + break; + } + const loopAbort = abortResult<{ entries: FileInfo[]; done: boolean }>(context.abortSignal, this.path); + if (loopAbort) return loopAbort; + const entryPath = resolve(this.path, entry.name); + let stats: Stats; + try { + stats = await lstat(entryPath); + } catch (error) { + // Removed between enumeration and lstat: not part of the listing any more. + if (isNodeError(error) && error.code === "ENOENT") continue; + return err(toFileError(error, entryPath)); + } + const info = fileInfoFromStats(entryPath, stats); + if (info.ok) entries.push(info.value); + } + return ok({ entries, done: this.done }); + } catch (error) { + return err(toFileError(error, this.path)); + } + } + + async close(_context: Context): Promise { + if (this.closed) return; + this.closed = true; + await this.dir.close().catch(() => undefined); + } +} + +function symlinkRefused(path: string, cause?: Error): FileError { + return new FileError("invalid", "Refusing to follow a symbolic link", path, cause); +} + export class NodeExecutionEnv implements ExecutionEnv { /** Every local environment sees the same files. */ readonly id: string = "node:local"; @@ -457,7 +588,7 @@ export class NodeExecutionEnv implements ExecutionEnv { } async exec( - command: string, + command: string | readonly string[], options: ShellExecOptions | undefined, context: Context, ): Promise> { @@ -468,8 +599,23 @@ export class NodeExecutionEnv implements ExecutionEnv { const timeoutMs = timeoutMsResult.value; const cwd = options?.cwd ? resolvePath(this.cwd, options.cwd) : this.cwd; - const shellConfig = await getShellConfig(this.shellPath); - if (!shellConfig.ok) return shellConfig; + // A string runs through the shell; an argv array runs its program directly, without shell parsing. + let program: string; + let args: readonly string[]; + let stdinCommand: string | undefined; + if (typeof command === "string") { + const shellConfig = await getShellConfig(this.shellPath); + if (!shellConfig.ok) return shellConfig; + const commandFromStdin = shellConfig.value.commandTransport === "stdin"; + program = shellConfig.value.shell; + args = commandFromStdin ? shellConfig.value.args : [...shellConfig.value.args, command]; + stdinCommand = commandFromStdin ? command : undefined; + } else { + const [first, ...rest] = command; + if (first === undefined) return err(new ExecutionError("spawn_error", "Empty argv: no program to run")); + program = first; + args = rest; + } try { await access(cwd, constants.F_OK); } catch (error) { @@ -510,10 +656,10 @@ export class NodeExecutionEnv implements ExecutionEnv { const stdoutDecoder = new TextDecoder(); const stderrDecoder = new TextDecoder(); // No output reaches the caller after exec() settled, for example from a descendant holding stdio open. - const emit = (text: string): void => { + const emit = (text: string, stream: ShellOutputInfo["stream"]): void => { if (settled || text === "" || options?.onOutput === undefined || callbackError !== undefined) return; try { - options.onOutput(text, context); + options.onOutput(text, context, { stream }); } catch (error) { failCallback(error); } @@ -592,22 +738,17 @@ export class NodeExecutionEnv implements ExecutionEnv { }; try { - const commandFromStdin = shellConfig.value.commandTransport === "stdin"; - child = spawn( - shellConfig.value.shell, - commandFromStdin ? shellConfig.value.args : [...shellConfig.value.args, command], - { - cwd, - detached: process.platform !== "win32", - env: getShellEnv(this.shellEnv, options?.env, options?.inheritEnv), - stdio: [commandFromStdin ? "pipe" : "ignore", "pipe", "pipe"], - windowsHide: true, - }, - ); + child = spawn(program, args, { + cwd, + detached: process.platform !== "win32", + env: getShellEnv(this.shellEnv, options?.env, options?.inheritEnv), + stdio: [stdinCommand === undefined ? "ignore" : "pipe", "pipe", "pipe"], + windowsHide: true, + }); if (child.pid) this.activeChildPids.add(child.pid); - if (commandFromStdin) { + if (stdinCommand !== undefined) { child.stdin?.on("error", () => {}); - child.stdin?.end(command); + child.stdin?.end(stdinCommand); } } catch (error) { const cause = toError(error); @@ -628,27 +769,29 @@ export class NodeExecutionEnv implements ExecutionEnv { else signal.addEventListener("abort", onAbort, { once: true }); } - const feed = (decoder: InstanceType) => (chunk: Uint8Array) => { - emit(decoder.decode(chunk, { stream: true })); - const spill = options?.spill; - if (spill === undefined || chunk.length === 0) return; - if (spillStart !== undefined) { + const feed = + (source: ShellOutputInfo["stream"], decoder: InstanceType) => (chunk: Uint8Array) => { + emit(decoder.decode(chunk, { stream: true }), source); + const spill = options?.spill; + if (spill === undefined || chunk.length === 0) return; + if (spillStart !== undefined) { + startSpill(chunk); + return; + } + seenBytes += chunk.length; + for (let index = chunk.indexOf(0x0a); index !== -1; index = chunk.indexOf(0x0a, index + 1)) + seenNewlines++; + const lines = seenNewlines + (chunk[chunk.length - 1] === 0x0a ? 0 : 1); + if (seenBytes <= spill.afterBytes && lines <= spill.afterLines) { + spillPrefix.push(chunk); + return; + } + for (const prefix of spillPrefix) startSpill(prefix); + spillPrefix.length = 0; startSpill(chunk); - return; - } - seenBytes += chunk.length; - for (let index = chunk.indexOf(0x0a); index !== -1; index = chunk.indexOf(0x0a, index + 1)) seenNewlines++; - const lines = seenNewlines + (chunk[chunk.length - 1] === 0x0a ? 0 : 1); - if (seenBytes <= spill.afterBytes && lines <= spill.afterLines) { - spillPrefix.push(chunk); - return; - } - for (const prefix of spillPrefix) startSpill(prefix); - spillPrefix.length = 0; - startSpill(chunk); - }; - child.stdout?.on("data", feed(stdoutDecoder)); - child.stderr?.on("data", feed(stderrDecoder)); + }; + child.stdout?.on("data", feed("stdout", stdoutDecoder)); + child.stderr?.on("data", feed("stderr", stderrDecoder)); void waitForChildProcess( child, @@ -659,8 +802,8 @@ export class NodeExecutionEnv implements ExecutionEnv { ).then( async ({ code, signal: exitSignal }) => { await finishSpill(); - emit(stdoutDecoder.decode()); - emit(stderrDecoder.decode()); + emit(stdoutDecoder.decode(), "stdout"); + emit(stderrDecoder.decode(), "stderr"); if (callbackError) { settle(err(callbackError)); return; @@ -753,6 +896,52 @@ export class NodeExecutionEnv implements ExecutionEnv { } } + async openBinaryReader( + path: string, + options: { noFollow?: boolean } | undefined, + context: Context, + ): Promise> { + const resolved = resolvePath(this.cwd, path); + const aborted = abortResult(context.abortSignal, resolved); + if (aborted) return aborted; + const noFollow = options?.noFollow === true; + let file: FileHandle | undefined; + try { + let flags = constants.O_RDONLY; + if (process.platform === "win32") { + // Windows has no O_NOFOLLOW; check the final component before opening (not race-free). + if (noFollow && (await lstat(resolved)).isSymbolicLink()) return err(symlinkRefused(resolved)); + } else { + // Nonblocking, so opening a FIFO does not wait for a writer. + flags |= constants.O_NONBLOCK; + if (noFollow) flags |= constants.O_NOFOLLOW; + } + file = await openFile(resolved, flags); + const stats = await file.stat(); + if (!stats.isFile()) { + await file.close().catch(() => undefined); + return err( + stats.isDirectory() + ? new FileError("is_directory", "Is a directory", resolved) + : new FileError("invalid", "Not a regular file", resolved), + ); + } + const afterOpenAbort = abortResult(context.abortSignal, resolved); + if (afterOpenAbort) { + await file.close().catch(() => undefined); + return afterOpenAbort; + } + return ok(new NodeBinaryReader(file, resolved)); + } catch (error) { + await file?.close().catch(() => undefined); + // O_NOFOLLOW reports a final-component symlink as ELOOP (EMLINK on some BSDs). + if (noFollow && isNodeError(error) && (error.code === "ELOOP" || error.code === "EMLINK")) { + return err(symlinkRefused(resolved, toError(error))); + } + return err(toFileError(error, resolved)); + } + } + async writeFile(path: string, content: string | Uint8Array, context: Context): Promise> { const resolved = resolvePath(this.cwd, path); const signal = context.abortSignal; @@ -876,6 +1065,23 @@ export class NodeExecutionEnv implements ExecutionEnv { } } + async openDirReader(path: string, context: Context): Promise> { + const resolved = resolvePath(this.cwd, path); + const aborted = abortResult(context.abortSignal, resolved); + if (aborted) return aborted; + try { + const dir = await opendir(resolved); + const afterOpenAbort = abortResult(context.abortSignal, resolved); + if (afterOpenAbort) { + await dir.close().catch(() => undefined); + return afterOpenAbort; + } + return ok(new NodeDirReader(dir, resolved)); + } catch (error) { + return err(toFileError(error, resolved)); + } + } + async canonicalPath(path: string, context: Context): Promise> { const resolved = resolvePath(this.cwd, path); const aborted = abortResult(context.abortSignal, resolved); diff --git a/packages/durable/src/testing/env-conformance.ts b/packages/durable/src/testing/env-conformance.ts new file mode 100644 index 000000000..474a9648d --- /dev/null +++ b/packages/durable/src/testing/env-conformance.ts @@ -0,0 +1,243 @@ +import type { Context } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT, withAbortSignal } from "@earendil-works/chord/context"; +import { type ExecutionEnv, type FileInfo, getOrThrow, type Result, type ShellOutputInfo } from "../env/index.ts"; +import type { EnvConformanceCase, EnvConformanceOptions } from "./types.ts"; + +const context = BACKGROUND_CONTEXT; +const decoder = new TextDecoder(); + +type EnvTest = (env: ExecutionEnv) => Promise; + +function abortedContext(): Context { + const controller = new AbortController(); + controller.abort(); + return withAbortSignal(controller.signal, BACKGROUND_CONTEXT); +} + +function errorCode(result: Result): string | undefined { + return result.ok ? undefined : result.error.code; +} + +async function readAll( + env: ExecutionEnv, + path: string, + maxEntries: number, +): Promise<{ pages: FileInfo[][]; done: boolean }> { + const reader = getOrThrow(await env.openDirReader(path, context)); + const pages: FileInfo[][] = []; + try { + for (let page = 0; page < 1000; page++) { + const next = getOrThrow(await reader.next(maxEntries, context)); + pages.push(next.entries); + if (next.done) return { pages, done: true }; + } + return { pages, done: false }; + } finally { + await reader.close(context); + } +} + +/** + * Creates runner-independent cases for an `ExecutionEnv`. `withEnv` must call and await its callback exactly once per + * case with an environment whose `cwd` is a fresh, empty, writable directory. + */ +export function createEnvConformance(options: EnvConformanceOptions): readonly EnvConformanceCase[] { + const assert = options.assertions; + const shell = options.shell ?? ["sh", "-c"]; + const symlinks = options.symlinks ?? true; + const createCase = (name: string, test: EnvTest): EnvConformanceCase => ({ + name, + run: () => options.withEnv(test), + }); + const execCollect = async (env: ExecutionEnv, command: string | readonly string[], cwd?: string) => { + const output: Record = { stdout: "", stderr: "" }; + const result = await env.exec( + command, + { + ...(cwd === undefined ? {} : { cwd }), + onOutput: (text, _context, info) => { + output[info.stream] += text; + }, + }, + context, + ); + return { result, ...output }; + }; + + const cases: EnvConformanceCase[] = [ + createCase("binary reader reads byte ranges of the opened file", async (env) => { + getOrThrow(await env.writeFile("data.txt", "hello world", context)); + const reader = getOrThrow(await env.openBinaryReader("data.txt", undefined, context)); + const info = getOrThrow(await reader.info(context)); + assert.partialDeepEqual(info, { name: "data.txt", kind: "file", size: 11 }); + assert.strictEqual(decoder.decode(getOrThrow(await reader.read(0, 5, context))), "hello"); + assert.strictEqual(decoder.decode(getOrThrow(await reader.read(6, 100, context))), "world"); + assert.strictEqual(getOrThrow(await reader.read(11, 4, context)).length, 0); + assert.strictEqual(getOrThrow(await reader.read(50, 1, context)).length, 0); + assert.strictEqual(getOrThrow(await reader.read(3, 0, context)).length, 0); + assert.strictEqual(errorCode(await reader.read(-1, 1, context)), "invalid"); + assert.strictEqual(errorCode(await reader.read(0, 1.5, context)), "invalid"); + assert.strictEqual(errorCode(await reader.read(0, 1, abortedContext())), "aborted"); + await reader.close(context); + await reader.close(context); + assert.strictEqual(errorCode(await reader.read(0, 1, context)), "invalid"); + assert.strictEqual(errorCode(await reader.info(context)), "invalid"); + }), + + createCase("binary reader keeps reading the file it opened after a rename", async (env) => { + getOrThrow(await env.writeFile("a.txt", "one", context)); + const reader = getOrThrow(await env.openBinaryReader("a.txt", undefined, context)); + try { + getOrThrow(await env.renameFile("a.txt", "b.txt", context)); + getOrThrow(await env.writeFile("a.txt", "two", context)); + assert.strictEqual(decoder.decode(getOrThrow(await reader.read(0, 10, context))), "one"); + } finally { + await reader.close(context); + } + }), + + createCase("binary reader refuses directories, missing files and aborted opens", async (env) => { + getOrThrow(await env.createDir("dir", undefined, context)); + getOrThrow(await env.writeFile("file.txt", "x", context)); + assert.strictEqual(errorCode(await env.openBinaryReader("dir", undefined, context)), "is_directory"); + assert.strictEqual(errorCode(await env.openBinaryReader("missing.txt", undefined, context)), "not_found"); + assert.strictEqual(errorCode(await env.openBinaryReader("file.txt", undefined, abortedContext())), "aborted"); + }), + + createCase("directory reader pages every entry exactly once", async (env) => { + const names = ["a.txt", "b.txt", "c.txt", "d.txt", "e.txt"]; + for (const name of names) getOrThrow(await env.writeFile(name, name, context)); + getOrThrow(await env.createDir("sub", undefined, context)); + const { pages, done } = await readAll(env, ".", 2); + assert.ok(done, "directory reader reached the end"); + for (const page of pages) assert.ok(page.length <= 2, "page within maxEntries"); + const entries = pages.flat(); + assert.deepEqual(entries.map((entry) => entry.name).sort(), [...names, "sub"].sort()); + assert.strictEqual(entries.find((entry) => entry.name === "sub")?.kind, "directory"); + assert.partialDeepEqual( + entries.find((entry) => entry.name === "a.txt"), + { kind: "file", size: 5 }, + ); + }), + + createCase("directory reader reports the end and refuses use after close", async (env) => { + getOrThrow(await env.createDir("empty", undefined, context)); + const reader = getOrThrow(await env.openDirReader("empty", context)); + assert.deepEqual(getOrThrow(await reader.next(10, context)), { entries: [], done: true }); + assert.deepEqual(getOrThrow(await reader.next(10, context)), { entries: [], done: true }); + assert.strictEqual(errorCode(await reader.next(0, context)), "invalid"); + assert.strictEqual(errorCode(await reader.next(1, abortedContext())), "aborted"); + await reader.close(context); + await reader.close(context); + assert.strictEqual(errorCode(await reader.next(1, context)), "invalid"); + }), + + createCase("directory reader refuses missing paths and files", async (env) => { + getOrThrow(await env.writeFile("file.txt", "x", context)); + assert.strictEqual(errorCode(await env.openDirReader("missing", context)), "not_found"); + const file = await env.openDirReader("file.txt", context); + assert.strictEqual(errorCode(file), "not_directory"); + assert.strictEqual(errorCode(await env.openDirReader(".", abortedContext())), "aborted"); + }), + + createCase("directory reader skips entries removed during enumeration", async (env) => { + getOrThrow(await env.createDir("dir", undefined, context)); + for (const name of ["x", "y", "z"]) getOrThrow(await env.writeFile(`dir/${name}`, name, context)); + const reader = getOrThrow(await env.openDirReader("dir", context)); + try { + for (const name of ["x", "y", "z"]) getOrThrow(await env.remove(`dir/${name}`, undefined, context)); + const entries: FileInfo[] = []; + for (let page = 0; page < 10; page++) { + const next = getOrThrow(await reader.next(10, context)); + entries.push(...next.entries); + if (next.done) break; + } + assert.deepEqual(entries, []); + } finally { + await reader.close(context); + } + }), + + createCase("argv exec passes arguments to the program without shell parsing", async (env) => { + const hostile = "it's $(touch pwned) `touch pwned` *; touch pwned"; + const { result, stdout } = await execCollect(env, [ + ...shell, + 'printf "%s|%s" "$1" "$2"', + "argv0", + hostile, + "a b", + ]); + assert.strictEqual(getOrThrow(result).exitCode, 0); + assert.strictEqual(stdout, `${hostile}|a b`); + assert.strictEqual(getOrThrow(await env.exists("pwned", context)), false); + }), + + createCase("exec reports the stream of every chunk in both forms", async (env) => { + const script = "printf out; printf err >&2; printf more"; + const argv = await execCollect(env, [...shell, script]); + assert.strictEqual(getOrThrow(argv.result).exitCode, 0); + assert.strictEqual(argv.stdout, "outmore"); + assert.strictEqual(argv.stderr, "err"); + const string = await execCollect(env, script); + assert.strictEqual(getOrThrow(string.result).exitCode, 0); + assert.strictEqual(string.stdout, "outmore"); + assert.strictEqual(string.stderr, "err"); + }), + + createCase("argv exec honors cwd and exit codes", async (env) => { + getOrThrow(await env.createDir("sub", undefined, context)); + const made = await execCollect(env, [...shell, "printf x > made.txt; exit 3"], "sub"); + assert.strictEqual(getOrThrow(made.result).exitCode, 3); + assert.strictEqual(getOrThrow(await env.readTextFile("sub/made.txt", context)), "x"); + }), + + createCase("argv exec reports missing programs and empty argv as spawn errors", async (env) => { + assert.strictEqual( + errorCode(await env.exec(["pi-durable-conformance-missing-program"], undefined, context)), + "spawn_error", + ); + assert.strictEqual(errorCode(await env.exec([], undefined, context)), "spawn_error"); + }), + + createCase("argv exec distinguishes timeout from abort", async (env) => { + const timedOut = await env.exec([...shell, "sleep 5"], { timeout: 0.1 }, context); + assert.strictEqual(errorCode(timedOut), "timeout"); + const controller = new AbortController(); + const running = env.exec([...shell, "sleep 5"], undefined, withAbortSignal(controller.signal, context)); + setTimeout(() => controller.abort(), 100); + assert.strictEqual(errorCode(await running), "aborted"); + }), + ]; + + if (symlinks) { + cases.push( + createCase("binary reader follows symlinks unless noFollow refuses the final one", async (env) => { + getOrThrow(await env.writeFile("target.txt", "target", context)); + getOrThrow(await env.createDir("sub", undefined, context)); + getOrThrow(await env.writeFile("sub/inner.txt", "inner", context)); + const linked = await env.exec( + [...shell, "ln -s target.txt link.txt && ln -s sub dirlink"], + undefined, + context, + ); + assert.strictEqual(getOrThrow(linked).exitCode, 0); + + const followed = getOrThrow(await env.openBinaryReader("link.txt", undefined, context)); + assert.strictEqual(decoder.decode(getOrThrow(await followed.read(0, 10, context))), "target"); + await followed.close(context); + + assert.strictEqual( + errorCode(await env.openBinaryReader("link.txt", { noFollow: true }, context)), + "invalid", + ); + + // Only the final component is refused; earlier symlinked directories still resolve. + const inner = getOrThrow(await env.openBinaryReader("dirlink/inner.txt", { noFollow: true }, context)); + assert.strictEqual(decoder.decode(getOrThrow(await inner.read(0, 10, context))), "inner"); + await inner.close(context); + }), + ); + } + + return cases; +} diff --git a/packages/durable/src/testing/index.ts b/packages/durable/src/testing/index.ts index d8f84cce8..fdaa421f5 100644 --- a/packages/durable/src/testing/index.ts +++ b/packages/durable/src/testing/index.ts @@ -1,5 +1,6 @@ export { createExpectAssertions, type ExpectLike } from "./assertions.ts"; -export { registerStorageConformance, type StorageConformanceRunner } from "./runner.ts"; +export { createEnvConformance } from "./env-conformance.ts"; +export { registerEnvConformance, registerStorageConformance, type StorageConformanceRunner } from "./runner.ts"; export { STORAGE_MEMORY_SCALES, STORAGE_READ_BENCHMARKS, @@ -15,6 +16,10 @@ export { } from "./storage-benchmark.ts"; export { createStorageConformance } from "./storage-conformance.ts"; export type { + EnvConformanceAssertions, + EnvConformanceCase, + EnvConformanceOptions, + EnvConformanceProvider, StorageConformanceAssertions, StorageConformanceCase, StorageConformanceOptions, diff --git a/packages/durable/src/testing/runner.ts b/packages/durable/src/testing/runner.ts index 8f63519a5..04bcd2a0c 100644 --- a/packages/durable/src/testing/runner.ts +++ b/packages/durable/src/testing/runner.ts @@ -1,6 +1,7 @@ import { createExpectAssertions, type ExpectLike } from "./assertions.ts"; +import { createEnvConformance } from "./env-conformance.ts"; import { createStorageConformance } from "./storage-conformance.ts"; -import type { StorageConformanceProvider } from "./types.ts"; +import type { EnvConformanceOptions, EnvConformanceProvider, StorageConformanceProvider } from "./types.ts"; export interface StorageConformanceRunner { readonly describe: (name: string, suite: () => void) => unknown; @@ -22,3 +23,16 @@ export function registerStorageConformance( for (const testCase of cases) runner.it(testCase.name, testCase.run); }); } + +/** Registers the runner-independent `ExecutionEnv` cases with a Vitest/Jest-compatible test runner. */ +export function registerEnvConformance( + runner: StorageConformanceRunner, + name: string, + withEnv: EnvConformanceProvider, + options: Pick = {}, +): void { + const cases = createEnvConformance({ ...options, assertions: createExpectAssertions(runner.expect), withEnv }); + runner.describe(name, () => { + for (const testCase of cases) runner.it(testCase.name, testCase.run); + }); +} diff --git a/packages/durable/src/testing/types.ts b/packages/durable/src/testing/types.ts index a23ffa0db..adfcc33c9 100644 --- a/packages/durable/src/testing/types.ts +++ b/packages/durable/src/testing/types.ts @@ -1,3 +1,4 @@ +import type { ExecutionEnv } from "../env/index.ts"; import type { Storage } from "../types.ts"; export interface StorageConformanceAssertions { @@ -20,3 +21,22 @@ export interface StorageConformanceCase { readonly name: string; run(): Promise; } + +export type EnvConformanceAssertions = StorageConformanceAssertions; + +/** Calls `use` once with an environment whose `cwd` is a fresh, empty, writable directory, then cleans up. */ +export type EnvConformanceProvider = (use: (env: ExecutionEnv) => Promise) => Promise; + +export interface EnvConformanceOptions { + readonly assertions: EnvConformanceAssertions; + readonly withEnv: EnvConformanceProvider; + /** Program and flag that run a POSIX shell script from the next argument; default `["sh", "-c"]`. */ + readonly shell?: readonly string[]; + /** Whether the shell's `ln -s` creates symbolic links; default true. */ + readonly symlinks?: boolean; +} + +export interface EnvConformanceCase { + readonly name: string; + run(): Promise; +} diff --git a/packages/durable/test/env-node-conformance.test.ts b/packages/durable/test/env-node-conformance.test.ts new file mode 100644 index 000000000..8736e54be --- /dev/null +++ b/packages/durable/test/env-node-conformance.test.ts @@ -0,0 +1,64 @@ +import { execFileSync } from "node:child_process"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { afterEach, describe, expect, it } from "vitest"; +import { getOrThrow } from "../src/env/index.ts"; +import { NodeExecutionEnv } from "../src/env/node.ts"; +import { registerEnvConformance } from "../src/testing/index.ts"; + +const windows = process.platform === "win32"; +const gitBash = join(process.env.ProgramFiles ?? "C:\\Program Files", "Git", "bin", "bash.exe"); +const context = BACKGROUND_CONTEXT; + +registerEnvConformance( + { describe, expect, it }, + "NodeExecutionEnv conformance", + async (use) => { + const cwd = mkdtempSync(join(tmpdir(), "pi-durable-env-conformance-")); + try { + await use(new NodeExecutionEnv({ cwd })); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }, + // Git Bash's `ln -s` copies instead of linking unless native symlinks are enabled. + windows ? { shell: [gitBash, "-c"], symlinks: false } : {}, +); + +describe("NodeExecutionEnv readers", () => { + const dirs: string[] = []; + const tempDir = (): string => { + const dir = mkdtempSync(join(tmpdir(), "pi-durable-env-readers-")); + dirs.push(dir); + return dir; + }; + afterEach(() => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); + }); + + it("reads ranges spanning several internal chunks exactly", async () => { + const cwd = tempDir(); + const bytes = new Uint8Array(2.5 * 1024 * 1024); + for (let index = 0; index < bytes.length; index++) bytes[index] = (index * 31) % 251; + writeFileSync(join(cwd, "big.bin"), bytes); + const reader = getOrThrow(await new NodeExecutionEnv({ cwd }).openBinaryReader("big.bin", undefined, context)); + try { + const all = getOrThrow(await reader.read(0, bytes.length + 10, context)); + expect(all.length).toBe(bytes.length); + expect(Buffer.from(all).equals(Buffer.from(bytes))).toBe(true); + const middle = getOrThrow(await reader.read(1024 * 1024 - 3, 7, context)); + expect([...middle]).toEqual([...bytes.subarray(1024 * 1024 - 3, 1024 * 1024 + 4)]); + } finally { + await reader.close(context); + } + }); + + it.skipIf(windows)("refuses a FIFO without waiting for a writer", async () => { + const cwd = tempDir(); + execFileSync("mkfifo", [join(cwd, "pipe")]); + const result = await new NodeExecutionEnv({ cwd }).openBinaryReader("pipe", undefined, context); + expect(result).toMatchObject({ ok: false, error: { code: "invalid" } }); + }); +}); diff --git a/packages/durable/test/tools.test.ts b/packages/durable/test/tools.test.ts index 37d65a2dc..48268d1f9 100644 --- a/packages/durable/test/tools.test.ts +++ b/packages/durable/test/tools.test.ts @@ -169,14 +169,14 @@ const TRUNCATED_OUTPUT_LINES = DEFAULT_MAX_LINES + 1; class TimeoutOutputExecutionEnv extends NodeExecutionEnv { override async exec( - _command: string, + _command: string | readonly string[], options: ShellExecOptions | undefined, context: Context, ): Promise> { const output = `${Array.from({ length: TRUNCATED_OUTPUT_LINES }, (_, index) => `line-${index + 1}`).join("\n")}\n`; const spillPath = getOrThrow(await this.createTempFile({ prefix: "timeout-", suffix: ".log" }, context)); getOrThrow(await this.writeFile(spillPath, output, context)); - options?.onOutput?.(output, context); + options?.onOutput?.(output, context, { stream: "stdout" }); const error = new ExecutionError("timeout", `timeout:${options?.timeout}`); error.spillPath = spillPath; return err(error);