mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
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
This commit is contained in:
Mario Zechner
parent
2e63fcdfbf
commit
4748c627ae
12 changed files with 690 additions and 50 deletions
2
.github/workflows/durable-env.yml
vendored
2
.github/workflows/durable-env.yml
vendored
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
50
packages/durable/src/env/index.ts
vendored
50
packages/durable/src/env/index.ts
vendored
|
|
@ -92,6 +92,26 @@ export interface TextLineReader {
|
|||
close(context: Context): Promise<void>;
|
||||
}
|
||||
|
||||
/** 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<Result<FileInfo, FileError>>;
|
||||
/** Up to `length` bytes at `offset`; fewer only at end of file. */
|
||||
read(offset: number, length: number, context: Context): Promise<Result<Uint8Array, FileError>>;
|
||||
close(context: Context): Promise<void>;
|
||||
}
|
||||
|
||||
/** 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<Result<{ entries: FileInfo[]; done: boolean }, FileError>>;
|
||||
close(context: Context): Promise<void>;
|
||||
}
|
||||
|
||||
/** Portable filesystem capability. Operations return failures rather than throwing. */
|
||||
export interface FileSystem {
|
||||
/**
|
||||
|
|
@ -110,6 +130,16 @@ export interface FileSystem {
|
|||
context: Context,
|
||||
): Promise<Result<string[], FileError>>;
|
||||
readBinaryFile(path: string, context: Context): Promise<Result<Uint8Array, FileError>>;
|
||||
/**
|
||||
* 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<Result<BinaryReader, FileError>>;
|
||||
writeFile(path: string, content: string | Uint8Array, context: Context): Promise<Result<void, FileError>>;
|
||||
appendFile(path: string, content: string | Uint8Array, context: Context): Promise<Result<void, FileError>>;
|
||||
/** Truncate or extend a file to exactly `size` bytes. */
|
||||
|
|
@ -119,6 +149,7 @@ export interface FileSystem {
|
|||
renameFile(sourcePath: string, destinationPath: string, context: Context): Promise<Result<void, FileError>>;
|
||||
fileInfo(path: string, context: Context): Promise<Result<FileInfo, FileError>>;
|
||||
listDir(path: string, context: Context): Promise<Result<FileInfo[], FileError>>;
|
||||
openDirReader(path: string, context: Context): Promise<Result<DirReader, FileError>>;
|
||||
canonicalPath(path: string, context: Context): Promise<Result<string, FileError>>;
|
||||
exists(path: string, context: Context): Promise<Result<boolean, FileError>>;
|
||||
createDir(
|
||||
|
|
@ -157,17 +188,30 @@ export interface ShellExecOptions {
|
|||
env?: Record<string, string>;
|
||||
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<Result<ShellExecResult, ExecutionError>>;
|
||||
/** Kill every command this environment still runs; for its owner's shutdown, never for one request. */
|
||||
cleanup(context: Context): Promise<void>;
|
||||
}
|
||||
|
||||
|
|
|
|||
290
packages/durable/src/env/node.ts
vendored
290
packages/durable/src/env/node.ts
vendored
|
|
@ -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<TValue>(what: string, path: string): Result<TValue, FileError> {
|
||||
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<Result<FileInfo, FileError>> {
|
||||
const aborted = abortResult<FileInfo>(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<Result<Uint8Array, FileError>> {
|
||||
const aborted = abortResult<Uint8Array>(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<Uint8Array>(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<void> {
|
||||
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<Result<{ entries: FileInfo[]; done: boolean }, FileError>> {
|
||||
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<void> {
|
||||
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<Result<ShellExecResult, ExecutionError>> {
|
||||
|
|
@ -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<typeof TextDecoder>) => (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<typeof TextDecoder>) => (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<Result<BinaryReader, FileError>> {
|
||||
const resolved = resolvePath(this.cwd, path);
|
||||
const aborted = abortResult<BinaryReader>(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<BinaryReader>(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<Result<void, FileError>> {
|
||||
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<Result<DirReader, FileError>> {
|
||||
const resolved = resolvePath(this.cwd, path);
|
||||
const aborted = abortResult<DirReader>(context.abortSignal, resolved);
|
||||
if (aborted) return aborted;
|
||||
try {
|
||||
const dir = await opendir(resolved);
|
||||
const afterOpenAbort = abortResult<DirReader>(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<Result<string, FileError>> {
|
||||
const resolved = resolvePath(this.cwd, path);
|
||||
const aborted = abortResult<string>(context.abortSignal, resolved);
|
||||
|
|
|
|||
243
packages/durable/src/testing/env-conformance.ts
Normal file
243
packages/durable/src/testing/env-conformance.ts
Normal file
|
|
@ -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<void>;
|
||||
|
||||
function abortedContext(): Context {
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
return withAbortSignal(controller.signal, BACKGROUND_CONTEXT);
|
||||
}
|
||||
|
||||
function errorCode(result: Result<unknown, { code: string }>): 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<ShellOutputInfo["stream"], string> = { 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;
|
||||
}
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<EnvConformanceOptions, "shell" | "symlinks"> = {},
|
||||
): void {
|
||||
const cases = createEnvConformance({ ...options, assertions: createExpectAssertions(runner.expect), withEnv });
|
||||
runner.describe(name, () => {
|
||||
for (const testCase of cases) runner.it(testCase.name, testCase.run);
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
}
|
||||
|
||||
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<void>) => Promise<void>;
|
||||
|
||||
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<void>;
|
||||
}
|
||||
|
|
|
|||
64
packages/durable/test/env-node-conformance.test.ts
Normal file
64
packages/durable/test/env-node-conformance.test.ts
Normal file
|
|
@ -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" } });
|
||||
});
|
||||
});
|
||||
|
|
@ -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<Result<ShellExecResult, ExecutionError>> {
|
||||
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);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue