From 81c70e3950ab196959cc68790effca6ff6bfc164 Mon Sep 17 00:00:00 2001 From: Noah Schatz Date: Thu, 20 Aug 2026 14:26:19 +0000 Subject: [PATCH 1/2] S0054-cli-bulk-1: a documented input-size limit and incremental record output for parse A legitimately large input used to walk into one of Node's own allocation ceilings, throw past every handler, and be reported as CLI_INTERNAL / exit 70: "an unexpected exception, i.e. a bug", about an input that is merely big. And multi-record parse output was accumulated in full and written once, so nothing reached stdout until the last byte of input had been read. The CLI now declares a limit of its own, 67108864 bytes (64 MiB), far below both platform ceilings, and checks it against the running byte count as the input arrives. An over-limit invocation is refused with a value-free CLI_INPUT_TOO_LARGE naming the limit and the data-error exit code (65), before anything allocates memory proportional to the oversized input. The number is rendered from one constant into `cosyte --help` and the command reference, and a test reds if the two ever disagree. Multi-record input (--ndjson and MLLP frames) is now read as chunks and emitted record by record as each record is parsed. Per-record isolation and the exit-code contract are unchanged. A fatal condition part way through keeps the lines already written and still resolves to that failure's own non-zero exit code, so a partial record stream is never presented as a complete one; a truncated MLLP stream is the visible case, where the frames that completed are now emitted before the truncation is detected at end of stream. A downstream consumer that closes the pipe is a value-free CLI_OUTPUT_WRITE_FAILED rather than an unhandled write error: one write per record is a failure surface a single write did not have. Spec: work/specs/S0054-cli-bulk-1/spec.md --- .changeset/brave-pumas-stream.md | 20 ++ CHANGELOG.md | 29 +++ README.md | 28 ++- docs-content/reference-commands.md | 17 +- src/bin/cosyte.ts | 22 +- src/commands/parse.ts | 164 ++++++++----- src/core/diagnostics.ts | 8 + src/core/format.ts | 17 +- src/core/input.ts | 204 ++++++++++++++-- src/core/io.ts | 139 ++++++++++- src/core/limits.ts | 91 +++++++ src/core/parsers.ts | 28 ++- src/core/records.ts | 221 +++++++++++++++++ src/core/run.ts | 34 ++- src/index.ts | 32 ++- test/formats.test.ts | 8 +- test/limit-docs.test.ts | 89 +++++++ test/limits.test.ts | 171 +++++++++++++ test/parse-bulk.test.ts | 369 +++++++++++++++++++++++++++++ 19 files changed, 1576 insertions(+), 115 deletions(-) create mode 100644 .changeset/brave-pumas-stream.md create mode 100644 src/core/limits.ts create mode 100644 src/core/records.ts create mode 100644 test/limit-docs.test.ts create mode 100644 test/limits.test.ts create mode 100644 test/parse-bulk.test.ts diff --git a/.changeset/brave-pumas-stream.md b/.changeset/brave-pumas-stream.md new file mode 100644 index 0000000..249f800 --- /dev/null +++ b/.changeset/brave-pumas-stream.md @@ -0,0 +1,20 @@ +--- +"@cosyte/cli": patch +--- + +Give `cosyte parse` a documented input-size limit and incremental multi-record output. + +An input past 67108864 bytes (64 MiB) is now refused with a value-free `CLI_INPUT_TOO_LARGE` +diagnostic naming the limit and the data-error exit code (`65`), never the internal-error code a +platform allocation failure used to produce. The check runs against the running byte count as the +input arrives, so the refusal lands before anything allocates memory proportional to the oversized +input. The limit is rendered from one constant into `cosyte --help` and the command reference, and a +test reds if those two ever disagree. + +Multi-record output (`--ndjson` and MLLP frames) is emitted record by record as each record is +parsed, rather than accumulated and written once at the end. Per-record isolation and the exit-code +contract are unchanged. A fatal condition part way through keeps the lines already written and still +resolves to that failure's own non-zero exit code, so a partial record stream is never presented as a +complete one; a truncated MLLP stream is the visible case, where the frames that completed are now +emitted before the truncation is detected. A downstream consumer that closes the pipe is a value-free +`CLI_OUTPUT_WRITE_FAILED` rather than an unhandled write error. diff --git a/CHANGELOG.md b/CHANGELOG.md index 70a7fc2..b7ebee7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -80,8 +80,37 @@ still do. Each entry was assigned to the release whose tag first contains it, re gate's own OK line.** Both key on an actual NUL byte; the wider set is git's own binary classification, which is why neither gate may be reduced to `grep -I`. +- **`parse` has a documented input-size limit: 67108864 bytes (64 MiB) per invocation, refused as a + data error.** Node has two hard allocation ceilings (`buffer.constants.MAX_LENGTH` and + `buffer.constants.MAX_STRING_LENGTH`), and a legitimately large input that crossed one of them threw + past every handler and was reported as `CLI_INTERNAL` / exit `70`, which says "this is a bug in the + tool" about an input that is merely big. The CLI now declares its own limit far below both, checks it + against the **running byte count as the input arrives**, and refuses with a value-free + `CLI_INPUT_TOO_LARGE` naming the limit and exit `65`. The number is rendered from one constant into + `cosyte --help` and the command reference, and a test reds if the two ever disagree. + - The refusal fires before anything allocates memory proportional to the oversized input, which is + what makes it a refusal rather than a slower way to reach the same crash. The regression suite + proves it with sources that would run past `MAX_STRING_LENGTH` if anything drained them. + - The MLLP frame reader's own smaller default ceiling is raised to the CLI's limit on this path, so + the documented number is the binding one rather than an undocumented number underneath it. + ### Changed +- **Multi-record `parse` output (`--ndjson` and MLLP) is now emitted record by record, as each record + is parsed**, instead of being accumulated and written once at the end. The first line reaches stdout + before the rest of the input has been read, so a bulk batch pipes into the next process instead of + waiting on the whole file. Per-record isolation and the exit-code contract are unchanged: a record + that fails to parse is still a value-free `{ record, error }` line, the stream still continues, and + any failed record still resolves the invocation to exit `65`. + - **A fatal condition part way through keeps the lines already written and still exits non-zero.** + A truncated MLLP stream is the visible case: the frames that completed have already been emitted + when the unterminated one is detected at end of stream, so `stdout` is no longer empty for that + input. It was never a success and still is not: the exit code carries the failure, and a partial + record stream is never reported as a complete one. + - **A downstream consumer that closes the pipe part way through is now a value-free + `CLI_OUTPUT_WRITE_FAILED`** rather than an unhandled write error. One write per record is a + failure surface a single write did not have. + - **`CLAUDE.md` narrative was relocated into `documentation/agent-notes.md` to make room for the gate's rules.** The branch-protection, PHI-scanner-residual and em-dash blocks were compressed to their imperatives; every trap keeps a one-line rule and a pointer, and the reasoning each one compresses diff --git a/README.md b/README.md index a316916..aef6ef2 100644 --- a/README.md +++ b/README.md @@ -146,7 +146,15 @@ astm | ncpdp | ccda | dicom | mllp`. **MLLP** stream, one record per frame, or any input under **`--ndjson`**, one record per non-empty line) streams as **NDJSON**, one `{ record, format, model, warnings }` line each, with per-record isolation: a record that fails to parse becomes a value-free `{ record, error }` line and the stream -continues; the overall exit is a data error (`65`) if any record failed. +continues; the overall exit is a data error (`65`) if any record failed. Each line is written **as +its record is parsed**, before the rest of the input has been read, so a batch pipes straight into +the next process. + +**Input size.** `parse` reads at most **67108864 bytes (64 MiB)** per invocation, the number +`cosyte --help` prints. A larger input is a value-free `CLI_INPUT_TOO_LARGE` refusal naming that +limit, with the data-error exit code (`65`) and never the internal-error code: split the input and +re-run. Since records are emitted as they are parsed, a refusal can land after some records have +already reached stdout; the exit code, not the output, says the run did not complete. Support is honest **per (format, operation)**: `x12`/`astm`/`ncpdp` support all of parse/inspect/fmt/ validate; `ccda` supports inspect/fmt/validate (parse deferred); `dicom` supports inspect/validate @@ -232,15 +240,15 @@ valid JSON or not a loadable ConceptMap is a `CLI_MAP_INVALID` data error (`65`) Every command is safe to branch on in CI. The exit code carries the outcome (`sysexits.h`): -| Code | Meaning | -| ---- | ---------------------------------------------------------- | -| `0` | success / **valid** (`validate`) | -| `1` | **invalid**: `validate` found a parseable-but-bad message | -| `2` | usage error (unknown flag, missing argument) | -| `65` | data error (unparseable input, or format undetected) | -| `66` | no input (missing/unreadable file) | -| `69` | unavailable (a capability is not yet built, e.g. `redact`) | -| `70` | internal error (a bug) | +| Code | Meaning | +| ---- | ------------------------------------------------------------------------------- | +| `0` | success / **valid** (`validate`) | +| `1` | **invalid**: `validate` found a parseable-but-bad message | +| `2` | usage error (unknown flag, missing argument) | +| `65` | data error (unparseable input, format undetected, or input past the size limit) | +| `66` | no input (missing/unreadable file) | +| `69` | unavailable (a capability is not yet built, e.g. `redact`) | +| `70` | internal error (a bug) | The load-bearing rule: the CLI **never prints a reassuring line and exits `0`** on input it could not handle, or on an invalid message. diff --git a/docs-content/reference-commands.md b/docs-content/reference-commands.md index 1d5e457..4e0940e 100644 --- a/docs-content/reference-commands.md +++ b/docs-content/reference-commands.md @@ -44,12 +44,27 @@ Parse a healthcare message to typed JSON on stdout (the data channel). Autodetec record per frame with per-record isolation (a failed record becomes a value-free `{ record, error }` line and the stream continues; any failure → exit `65`). +Multi-record output is **written record by record, as each record is parsed**, so the first line +reaches stdout long before the last byte of input has been read and a large batch can be piped +straight into another process. + ```bash cosyte parse adt.hl7 cat adt.hl7 | cosyte parse - cosyte parse --ndjson bulk.ndjson | jq . ``` +#### Input size limit + +`parse` reads at most **67108864 bytes (64 MiB)** of input per invocation. A larger input is refused +with a value-free `CLI_INPUT_TOO_LARGE` diagnostic naming that limit and the **data-error** exit code +(`65`), never the internal-error code (`70`): a large input is not a bug in the tool. Split the input +and re-run. The same number is printed by `cosyte --help`. + +Because records are emitted as they are parsed, a refusal can arrive **after** some records have +already been written to stdout. The exit code, not the output, is the signal that the run did not +complete: a partial record stream always carries a non-zero exit, never `0`. + ### `cosyte validate [--profile ]` Validate a message; **the exit code carries the verdict**: `0` valid, `1` invalid (parseable but @@ -104,7 +119,7 @@ tools over the same core. See [MCP server](./mcp). | `0` | success / `validate` found the input **valid** | | `1` | operation-level failure: `validate` found the input **invalid** (a real CI signal) | | `2` | usage error: unknown command, bad flag, missing argument | -| `65` | data error: input could not be parsed / format not detected / (format, op) unsupported | +| `65` | data error: input could not be parsed / format not detected / (format, op) unsupported / input larger than the size limit | | `66` | no input: the file does not exist or is unreadable | | `69` | unavailable: a capability is not yet built (e.g. `redact`, `--profile`) | | `70` | internal error: an unexpected exception (a bug) | diff --git a/src/bin/cosyte.ts b/src/bin/cosyte.ts index cfa3059..a97558c 100644 --- a/src/bin/cosyte.ts +++ b/src/bin/cosyte.ts @@ -10,12 +10,32 @@ */ /* v8 ignore start -- process wiring: argv/stdin/stdout/exit glue, exercised by the packaged bin smoke, not unit-covered */ -import { readFileBytes, readStreamBytes, type RunDeps } from "../core/io.js"; +import { + fileChunks, + readFileBytes, + readStreamBytes, + streamChunks, + type RunDeps, +} from "../core/io.js"; import { run } from "../core/run.js"; +// A consumer that closes the pipe (`cosyte parse big.ndjson | head -3`) makes the next write fail. +// Node reports that as an 'error' event on the stream, which is fatal if nothing is listening, so +// the sink below is marked closed instead and the command resolves it as a write failure. +let stdoutOpen = true; +process.stdout.on("error", () => { + stdoutOpen = false; +}); + const deps: RunDeps = { readFile: (path) => readFileBytes(path), readStdin: () => readStreamBytes(process.stdin), + openFile: (path) => fileChunks(path), + openStdin: () => streamChunks(process.stdin), + writeStdout: (chunk) => { + if (!stdoutOpen) throw new Error("stdout closed"); + process.stdout.write(chunk); + }, }; // The `cosyte mcp` subcommand starts the stdio MCP server (also reachable as the `cosyte-mcp` bin). diff --git a/src/commands/parse.ts b/src/commands/parse.ts index d704f88..b1c08c5 100644 --- a/src/commands/parse.ts +++ b/src/commands/parse.ts @@ -12,7 +12,14 @@ * `{ record, error }` line and the stream continues, and the overall exit is a data error (`65`) if any * record failed. Two inputs are multi-record: an **MLLP** stream (each VT-framed frame is an enclosed * HL7 message) and any input under **`--ndjson`** (each non-empty line is a record: the FHIR bulk-data - * convention). + * convention). Multi-record output is emitted **record by record, as each is parsed**, so the first + * line reaches the data channel long before the last byte of input has been read. + * + * **The size limit.** One invocation reads at most the documented number of input bytes (see + * `core/limits.ts`), counted against the running total as the input arrives. A larger input is a + * value-free `CLI_INPUT_TOO_LARGE` **data error**, never the internal-error code, and never a partial + * record stream presented as a complete one: whatever reached stdout before the refusal stands, and + * the invocation still resolves to the failure's own non-zero exit code. * * The CLI adds **no** parsing of its own: it routes, reads, and shapes output; `cosyte parse` equals the * wrapped library's programmatic parse. @@ -22,13 +29,14 @@ import { parseArgs } from "node:util"; -import { CLI_CODES, CliError, errorResult } from "../core/diagnostics.js"; +import { CLI_CODES, CliError, errorResult, formatDiagnostic } from "../core/diagnostics.js"; import { EXIT } from "../core/exit-codes.js"; import type { CosyteFormat } from "../core/format.js"; -import { resolveInput } from "../core/input.js"; +import { resolveInputStream } from "../core/input.js"; import type { RunDeps } from "../core/io.js"; -import { deframeMllp, parseFormat, type ParseWarning } from "../core/parsers.js"; +import { parseFormat, type ParseWarning } from "../core/parsers.js"; import { VALUE_FREE, type PhiPosture } from "../core/phi.js"; +import { collectChunks, mllpFrames, ndjsonRecords, type ByteChunks } from "../core/records.js"; import type { RunResult } from "../core/result.js"; import { extractStableCode, parseFailureResult } from "../core/wrap.js"; @@ -65,13 +73,16 @@ const PARSE_OPTIONS = { * Run the `parse` command. * * @param args - The arguments after the `parse` subcommand token. - * @param deps - Injected input readers ({@link RunDeps}). + * @param deps - Injected I/O ({@link RunDeps}). When it carries the chunk readers and the output + * sink, a multi-record input is read and emitted incrementally; when it does not, the same path + * runs over one chunk and the output comes back on `stdout` instead. * @param posture - The resolved {@link PhiPosture}. Defaults to {@link VALUE_FREE}; under * `--unsafe-show-values` a bounded excerpt of the offending input is appended to a `CLI_PARSE_FAILED` * diagnostic (the single, opt-in value-echoing surface): single-record mode only. * @returns A {@link RunResult}: the typed-JSON model (or NDJSON records) on `stdout`, a value-free note - * (or nothing) on `stderr`, and the resolved exit code. Never throws a {@link CliError}: it resolves - * it to a result; unexpected exceptions are caught by the dispatcher and mapped to `CLI_INTERNAL`. + * (or nothing) on `stderr`, and the resolved exit code. An input past the documented size limit is a + * value-free data error naming the limit. Never throws a {@link CliError}: it resolves it to a + * result; unexpected exceptions are caught by the dispatcher and mapped to `CLI_INTERNAL`. * @throws Never {@link CliError}; may propagate a truly unexpected error for the dispatcher to map. * @example * ```ts @@ -112,13 +123,28 @@ export async function parseCommand( ); } - const resolved = await resolveInput(positionals[0], values.format, deps, "parse"); + const resolved = await resolveInputStream(positionals[0], values.format, deps, "parse"); if (!resolved.ok) return resolved.result; - const { format, bytes } = resolved.input; + const { format, chunks } = resolved.input; // MLLP is a transport container and `--ndjson` is explicit batch mode: both are multi-record. if (format === "mllp" || values.ndjson === true) { - return await parseMulti(format, bytes, values.ndjson === true, values.quiet === true); + return await parseMulti( + format, + chunks, + values.ndjson === true, + values.quiet === true, + deps.writeStdout, + ); + } + + // A single message is parsed whole, so the rest of the input is drained here (still size-limited). + let bytes: Uint8Array; + try { + bytes = await collectChunks(chunks); + } catch (e) { + if (e instanceof CliError) return errorResult(e); + throw e; } return await parseSingle(format, bytes, values.json === true, values.quiet === true, posture); } @@ -150,31 +176,83 @@ async function parseSingle( return { stdout, stderr, exit: EXIT.OK }; } -/** Parse a multi-record input (MLLP frames, or `--ndjson` lines) → NDJSON, with per-record isolation. */ +/** + * Parse a multi-record input (MLLP frames, or `--ndjson` lines) → NDJSON, with per-record isolation + * and **per-record emission**: each line reaches the data channel as soon as its record is parsed, + * while the rest of the input is still arriving. + * + * Failure isolation is unchanged by that move: a record the parser rejects becomes a value-free + * `{ record, error }` line, the stream continues, and any failed record resolves the invocation to + * the data-error exit. A **fatal** condition (the over-limit refusal, a truncated MLLP stream, a + * parser that is not installed, a downstream consumer closing the pipe) ends the stream and resolves + * to that failure's own non-zero code, **keeping** whatever already reached stdout: a partial record + * stream is never dressed up as a complete one, and never reported as a success. + */ async function parseMulti( format: CosyteFormat, - bytes: Uint8Array, + chunks: ByteChunks, ndjson: boolean, quiet: boolean, + writeStdout: ((chunk: string) => void) | undefined, ): Promise { - // Resolve the records + the format each record is parsed as. MLLP de-frames to enclosed HL7 payloads. - let records: Uint8Array[]; - let recordFormat: CosyteFormat; + // MLLP de-frames to enclosed HL7 payloads; every other multi-record input is one record per line. + const recordFormat: CosyteFormat = format === "mllp" ? "hl7" : format; + const records = format === "mllp" ? mllpFrames(chunks) : ndjsonRecords(chunks); + + const held: string[] = []; + const emit = (line: string): void => { + if (writeStdout === undefined) { + held.push(line); + return; + } + try { + writeStdout(line); + } catch { + // The consumer went away part way through the stream (a closed pipe). Value-free, and never + // an unhandled platform error reaching the terminal. + throw new CliError( + CLI_CODES.CLI_OUTPUT_WRITE_FAILED, + EXIT.SOFTWARE, + "could not write to the output stream; it closed before the record stream finished", + ); + } + }; + const written = (): string => held.join(""); + + let total = 0; + let failed = 0; + let warnings = 0; + try { - if (format === "mllp") { - const { payloads } = await deframeMllp(bytes); - records = payloads; - recordFormat = "hl7"; - } else { - records = splitLines(bytes); - recordFormat = format; + for await (const record of records) { + let line: RecordLine; + try { + const { model, warnings: ws } = await parseFormat(recordFormat, record); + warnings += ws.length; + line = { record: total, format: recordFormat, model, warnings: ws }; + } catch (e) { + if (e instanceof CliError) throw e; // a parser-unavailable is fatal for the whole stream + failed += 1; + // Value-free per-record error: a stable code (if the throw carried one), never the bytes. + line = { + record: total, + format: recordFormat, + error: extractStableCode(e) ?? "CLI_PARSE_FAILED", + }; + } + total += 1; + emit(JSON.stringify(line) + "\n"); } } catch (e) { - if (e instanceof CliError) return errorResult(e); - return parseFailureResult(format, bytes, VALUE_FREE, e); + // Fatal, part way through: the exit code is the failure's own, and the lines already emitted stay. + if (e instanceof CliError) { + return { stdout: written(), stderr: `${formatDiagnostic(e)}\n`, exit: e.exit }; + } + const rejected = parseFailureResult(format, new Uint8Array(), VALUE_FREE, e); + return { stdout: written(), stderr: rejected.stderr, exit: rejected.exit }; } - if (records.length === 0) { + if (total === 0) { // A framed/ndjson input that yielded no record is a data error, never a silent success. return errorResult( new CliError( @@ -185,44 +263,12 @@ async function parseMulti( ); } - const lines: RecordLine[] = []; - let failed = 0; - let warnings = 0; - for (let i = 0; i < records.length; i += 1) { - const rec = records[i] as Uint8Array; - try { - const { model, warnings: ws } = await parseFormat(recordFormat, rec); - warnings += ws.length; - lines.push({ record: i, format: recordFormat, model, warnings: ws }); - } catch (e) { - if (e instanceof CliError) return errorResult(e); // a parser-unavailable is fatal for the stream - failed += 1; - // Value-free per-record error: a stable code (if the throw carried one), never the bytes. - lines.push({ - record: i, - format: recordFormat, - error: extractStableCode(e) ?? "CLI_PARSE_FAILED", - }); - } - } - - const stdout = lines.map((l) => JSON.stringify(l)).join("\n") + "\n"; const exit = failed > 0 ? EXIT.DATAERR : EXIT.OK; const stderr = quiet ? "" - : `cosyte: parsed ${String(records.length)} ${recordFormat} record(s)` + + : `cosyte: parsed ${String(total)} ${recordFormat} record(s)` + ` (${String(warnings)} warning(s), ${String(failed)} failed)` + (ndjson ? " [ndjson]" : format === "mllp" ? " [mllp]" : "") + "\n"; - return { stdout, stderr, exit }; -} - -/** Split input bytes into non-empty, whitespace-trimmed newline-delimited records (NDJSON input). */ -function splitLines(bytes: Uint8Array): Uint8Array[] { - const text = new TextDecoder("utf-8", { fatal: false }).decode(bytes); - const enc = new TextEncoder(); - return text - .split(/\r?\n/) - .filter((line) => line.trim().length > 0) - .map((line) => enc.encode(line)); + return { stdout: written(), stderr, exit }; } diff --git a/src/core/diagnostics.ts b/src/core/diagnostics.ts index b36cc5e..3f6c123 100644 --- a/src/core/diagnostics.ts +++ b/src/core/diagnostics.ts @@ -46,6 +46,14 @@ export const CLI_CODES = { CLI_PARSER_UNAVAILABLE: "CLI_PARSER_UNAVAILABLE", /** The wrapped parser rejected the input. Positional context only, never the offending bytes. Exit `65`. */ CLI_PARSE_FAILED: "CLI_PARSE_FAILED", + /** The input is larger than the documented size limit the CLI reads in one invocation. Raised + * against the running byte count while the input is read, so the refusal always arrives before an + * allocation ceiling could. Names the limit, never the input. Exit `65`, never `70`. */ + CLI_INPUT_TOO_LARGE: "CLI_INPUT_TOO_LARGE", + /** The output stream could not be written: a record stream is emitted line by line, so a consumer + * that closes the pipe part way through ends the invocation here rather than as an unhandled + * platform error. Names no input. Exit `70`. */ + CLI_OUTPUT_WRITE_FAILED: "CLI_OUTPUT_WRITE_FAILED", /** The BYO ConceptMap supplied to `map-codes` is not valid JSON or not a loadable FHIR ConceptMap. * Names the stable terminology-loader code, never the map's bytes. Exit `65`. */ CLI_MAP_INVALID: "CLI_MAP_INVALID", diff --git a/src/core/format.ts b/src/core/format.ts index e274ae2..e3d8def 100644 --- a/src/core/format.ts +++ b/src/core/format.ts @@ -56,8 +56,23 @@ interface Signature { readonly match: (prefix: string, bytes: Uint8Array) => boolean; } +/** + * How many leading bytes detection reads. Small: every signature lives in the first line (the + * deepest is DICOM's magic at byte 128). Exported because a **streaming** caller has to know how much + * of the input to hold back before it can detect the format, and reading a different amount than + * detection uses would make the two disagree. + * + * @example + * ```ts + * import { DETECT_PREFIX_BYTES } from "@cosyte/cli"; + * + * DETECT_PREFIX_BYTES; // => 512 + * ``` + */ +export const DETECT_PREFIX_BYTES = 512; + /** How many leading bytes to decode for text sniffing. Small: signatures live in the first line. */ -const SNIFF_BYTES = 512; +const SNIFF_BYTES = DETECT_PREFIX_BYTES; /** Strip a leading UTF-8 BOM and any leading ASCII whitespace for a tolerant sniff. Deliberately does * **not** strip the MLLP `0x0B` VT frame byte (`\v`): that byte is the mllp signature, so consuming it diff --git a/src/core/input.ts b/src/core/input.ts index 9464987..9084cbc 100644 --- a/src/core/input.ts +++ b/src/core/input.ts @@ -8,16 +8,30 @@ * * Every failure is a value-free {@link CliError} rendered to a {@link RunResult}: a missing argument * is a usage error (`2`), an unreadable file a no-input error (`66`), empty/undetected/unwired input a - * data error (`65`). None ever echoes an input byte. + * data error (`65`), an input past the documented size limit a data error too (`65`, never the + * internal-error code). None ever echoes an input byte. + * + * Two front doors, one contract: {@link resolveInput} hands back the whole input as one buffer, and + * {@link resolveInputStream} hands back a chunk stream for a command that emits output before the + * input has finished arriving. They share the format-resolution and exit-code decisions, so a bad + * `--format`, an undetectable input and an unwired (format, op) resolve identically either way. * * @packageDocumentation */ -import { asCosyteFormat, detectFormat, detectionError, type CosyteFormat } from "./format.js"; +import { + asCosyteFormat, + detectFormat, + detectionError, + DETECT_PREFIX_BYTES, + type CosyteFormat, +} from "./format.js"; import { CLI_CODES, CliError, errorResult } from "./diagnostics.js"; import { EXIT } from "./exit-codes.js"; import type { RunDeps } from "./io.js"; +import { MAX_INPUT_BYTES } from "./limits.js"; import { formatsSupporting, supportsOp, type Op } from "./parsers.js"; +import { oneChunk, withinLimit, type ByteChunks } from "./records.js"; import type { RunResult } from "./result.js"; /** A successfully-resolved input: the format (guaranteed to support the requested op) and the bytes. */ @@ -89,42 +103,188 @@ export async function resolveInput( return fail(new CliError(CLI_CODES.CLI_EMPTY_INPUT, EXIT.DATAERR, "input is empty")); } - // Resolve the format: an explicit --format override (validated), else content autodetection. + const format = resolveFormat(bytes, formatOverride, op); + if (format instanceof CliError) return fail(format); + + return { ok: true, input: { format, bytes } }; +} + +/** + * A successfully-resolved **streaming** input: the format, plus the input as a stream of chunks with + * the leading bytes detection consumed put back in front, so the consumer still sees the whole input + * in order. + */ +export interface ResolvedInputStream { + /** The resolved format: guaranteed to satisfy `supportsOp(format, op)` for the requested op. */ + readonly format: CosyteFormat; + /** The whole input as chunks, size-limited: reading past the limit raises `CLI_INPUT_TOO_LARGE`. */ + readonly chunks: ByteChunks; +} + +/** + * The outcome of {@link resolveInputStream}: the resolved streaming input, or a ready-to-return + * value-free {@link RunResult}. The same discriminated-union shape as {@link InputResolution}. + */ +export type InputStreamResolution = + | { readonly ok: true; readonly input: ResolvedInputStream } + | { readonly ok: false; readonly result: RunResult }; + +/** + * Resolve the input as a **stream of chunks** rather than one buffer, for a command that can act on a + * record before the whole input has arrived. + * + * It reads only as far as the detection window ({@link DETECT_PREFIX_BYTES}) before deciding the + * format, then hands back those bytes followed by the rest of the stream, unread. The size limit is + * applied to the running total from the first chunk, so an over-limit input is refused while it is + * still arriving and never assembled. + * + * When the caller's {@link RunDeps} carry no chunk readers, the whole-input readers are wrapped as a + * single chunk: the same code path, the same output, only the granularity differs. + * + * @param source - The positional `` argument (or `undefined` when it was omitted). + * @param formatOverride - The raw `--format` value, or `undefined` to autodetect by content. + * @param deps - Injected readers ({@link RunDeps}); the chunk readers are used when present. + * @param op - The wrapping operation the caller will run; the resolved format is confirmed to support it. + * @param limit - The maximum number of input bytes to accept. Defaults to {@link MAX_INPUT_BYTES}. + * @returns `{ ok: true, input }` with the format and the chunk stream, else `{ ok: false, result }` + * carrying the value-free usage / no-input / data-error {@link RunResult}. + * @throws Propagates a **non-`CliError`** read failure unchanged, so the dispatcher maps it to + * `CLI_INTERNAL`. + * @example + * ```ts + * import { resolveInputStream } from "@cosyte/cli"; + * + * const deps = { + * readFile: async () => new TextEncoder().encode('{"resourceType":"Patient"}'), + * readStdin: async () => new Uint8Array(), + * }; + * const r = await resolveInputStream("patient.json", undefined, deps, "parse"); + * if (r.ok) r.input.format; // => "fhir" + * ``` + */ +export async function resolveInputStream( + source: string | undefined, + formatOverride: string | undefined, + deps: RunDeps, + op: Op, + limit: number = MAX_INPUT_BYTES, +): Promise { + if (source === undefined) { + return fail( + new CliError( + CLI_CODES.CLI_USAGE, + EXIT.USAGE, + "missing argument; pass a path or `-` to read stdin", + ), + ); + } + + const iterator = withinLimit(openChunks(source, deps), limit)[Symbol.asyncIterator](); + + // Read just far enough to decide the format, and no further: the detection window when the format + // is autodetected, a single byte (the empty-input check) when `--format` already decided it. Those + // chunks are put back in front below, so nothing is consumed twice. + const need = formatOverride === undefined ? DETECT_PREFIX_BYTES : 1; + const head: Uint8Array[] = []; + let headBytes = 0; + try { + while (headBytes < need) { + const step = await iterator.next(); + if (step.done === true) break; + head.push(step.value); + headBytes += step.value.length; + } + } catch (e) { + if (e instanceof CliError) return fail(e); + throw e; + } + + if (headBytes === 0) { + return fail(new CliError(CLI_CODES.CLI_EMPTY_INPUT, EXIT.DATAERR, "input is empty")); + } + + const prefix = Buffer.concat( + head.map((c) => Buffer.from(c.buffer, c.byteOffset, c.length)), + Math.min(headBytes, DETECT_PREFIX_BYTES), + ); + const format = resolveFormat(prefix, formatOverride, op); + if (format instanceof CliError) { + await iterator.return(undefined); + return fail(format); + } + + return { ok: true, input: { format, chunks: rejoin(head, iterator) } }; +} + +/** The chunk stream for ``: the injected chunk reader, or the whole-input reader as one chunk. */ +function openChunks(source: string, deps: RunDeps): ByteChunks { + if (source === "-") { + return deps.openStdin === undefined ? readerAsChunks(() => deps.readStdin()) : deps.openStdin(); + } + return deps.openFile === undefined + ? readerAsChunks(() => deps.readFile(source)) + : deps.openFile(source); +} + +/** Adapt a whole-input reader to the chunk-stream shape (one chunk, or none when it is empty). */ +async function* readerAsChunks(read: () => Promise): AsyncGenerator { + yield* oneChunk(await read()); +} + +/** Re-attach the chunks detection consumed to the front of the unread remainder. */ +async function* rejoin( + head: readonly Uint8Array[], + rest: AsyncIterator, +): AsyncGenerator { + for (const chunk of head) yield chunk; + for (;;) { + const step = await rest.next(); + if (step.done === true) return; + yield step.value; + } +} + +/** + * Resolve the format for an input from its **leading bytes**: an explicit `--format` override + * (validated), else conservative content autodetection, then a check that the format supports the + * requested operation. Shared by the buffered and streaming resolvers so the exit-code contract for a + * bad `--format`, an undetectable input and an unwired (format, op) is applied in exactly one place. + */ +function resolveFormat( + prefix: Uint8Array, + formatOverride: string | undefined, + op: Op, +): CosyteFormat | CliError { let format: CosyteFormat; if (formatOverride !== undefined) { const narrowed = asCosyteFormat(formatOverride); if (narrowed === null) { - return fail( - new CliError( - CLI_CODES.CLI_USAGE, - EXIT.USAGE, - "unknown --format value; expected one of hl7, fhir, dicom, x12, ccda, ncpdp, astm, mllp", - ), + return new CliError( + CLI_CODES.CLI_USAGE, + EXIT.USAGE, + "unknown --format value; expected one of hl7, fhir, dicom, x12, ccda, ncpdp, astm, mllp", ); } format = narrowed; } else { - const detected = detectFormat(bytes); + const detected = detectFormat(prefix); // `format` is non-null iff detection is `certain`; `none`/`ambiguous` become a value-free data error. - if (detected.format === null) return fail(detectionError(detected)); + if (detected.format === null) return detectionError(detected); format = detected.format; } if (!supportsOp(format, op)) { - return fail( - new CliError( - CLI_CODES.CLI_FORMAT_UNSUPPORTED, - EXIT.DATAERR, - `format '${format}' does not support \`${op}\` in this CLI build ` + - `(${op} supports: ${formatsSupporting(op).join(", ")})`, - ), + return new CliError( + CLI_CODES.CLI_FORMAT_UNSUPPORTED, + EXIT.DATAERR, + `format '${format}' does not support \`${op}\` in this CLI build ` + + `(${op} supports: ${formatsSupporting(op).join(", ")})`, ); } - - return { ok: true, input: { format, bytes } }; + return format; } -/** Wrap a {@link CliError} as a failed {@link InputResolution}. */ -function fail(e: CliError): InputResolution { +/** Wrap a {@link CliError} as a failed resolution (either shape). */ +function fail(e: CliError): { readonly ok: false; readonly result: RunResult } { return { ok: false, result: errorResult(e) }; } diff --git a/src/core/io.ts b/src/core/io.ts index cb4d3b8..543dfe8 100644 --- a/src/core/io.ts +++ b/src/core/io.ts @@ -3,28 +3,55 @@ * (`-`), with a value-free failure mode. The CLI never writes a temp file and never logs to a file; * this module only *reads*. * + * Reading comes in two shapes. The **whole-input** readers ({@link readFileBytes}, + * {@link readStreamBytes}) hand a command one buffer, which is what a single-message command wants. + * The **chunk** readers ({@link fileChunks}, {@link streamChunks}) hand it an async stream of chunks, + * which is what a multi-record command wants: it can emit a record's output before the rest of the + * input has arrived, and the CLI's own size limit can be enforced against the **running total** + * rather than after an oversized input has already been assembled. + * * The reader functions are injected into the command layer as {@link RunDeps} so the whole dispatch * path is testable without touching `process`: the real `process.stdin` wiring lives in the thin - * `bin` entry. + * `bin` entry, and so does the one function in {@link RunDeps} that writes rather than reads. * * @packageDocumentation */ -import { readFile } from "node:fs/promises"; +import { open, readFile, stat } from "node:fs/promises"; import type { Readable } from "node:stream"; import { CLI_CODES, CliError } from "./diagnostics.js"; import { EXIT } from "./exit-codes.js"; +import { inputTooLargeError, MAX_INPUT_BYTES } from "./limits.js"; /** - * The injectable input side-effects the dispatcher needs. Kept tiny and pure-ish so tests drive the - * CLI end to end with in-memory fakes and no real filesystem or stdin. + * The injectable I/O side-effects the dispatcher needs. Kept tiny and pure-ish so tests drive the + * CLI end to end with in-memory fakes and no real filesystem, stdin or stdout. + * + * The two whole-input readers are required; the chunk readers and the output sink are **optional**, + * and a caller that omits them still gets identical output. They are the difference between reading + * a stream as it arrives and being handed one buffer: with them, a multi-record `parse` emits each + * record as it is produced and refuses an over-limit input while it is still arriving; without them, + * the same code path runs over a single chunk and returns its output on the result's `stdout` as + * before. */ export interface RunDeps { /** Read a file's bytes, or raise a `CLI_NO_INPUT` {@link CliError} if it cannot be read. */ readonly readFile: (path: string) => Promise; /** Read all of stdin's bytes. */ readonly readStdin: () => Promise; + /** Optional: open a file as a stream of byte chunks (see {@link fileChunks}). */ + readonly openFile?: (path: string) => AsyncIterable; + /** Optional: open stdin as a stream of byte chunks (see {@link streamChunks}). */ + readonly openStdin?: () => AsyncIterable; + /** + * Optional: write one piece of the data channel **as it is produced**. When present, a multi-record + * `parse` writes each record's NDJSON line through here instead of accumulating them, and the + * returned result's `stdout` is empty because the output has already been delivered. May throw (a + * consumer that closed the pipe); the caller turns that into a value-free + * `CLI_OUTPUT_WRITE_FAILED`. + */ + readonly writeStdout?: (chunk: string) => void; } /** @@ -32,9 +59,14 @@ export interface RunDeps { * {@link EXIT.NOINPUT} error. The path is structural context (the user supplied it), so it may appear * in the message; the file *contents* never do. * + * A file whose size already exceeds `limit` is refused **before it is read**, with the value-free + * over-limit data error, so an oversized file is never allocated in the first place. + * * @param path - The file path to read. + * @param limit - The maximum number of bytes to accept. Defaults to {@link MAX_INPUT_BYTES}. * @returns The file bytes. - * @throws {CliError} `CLI_NO_INPUT` (exit `66`) when the file is missing, a directory, or unreadable. + * @throws {CliError} `CLI_NO_INPUT` (exit `66`) when the file is missing, a directory, or unreadable; + * `CLI_INPUT_TOO_LARGE` (exit `65`) when it is larger than `limit`. * @example * ```ts throws * import { readFileBytes } from "@cosyte/cli"; @@ -42,7 +74,18 @@ export interface RunDeps { * await readFileBytes("/no/such/file"); // throws CliError CLI_NO_INPUT * ``` */ -export async function readFileBytes(path: string): Promise { +export async function readFileBytes( + path: string, + limit: number = MAX_INPUT_BYTES, +): Promise { + try { + const info = await stat(path); + if (info.isFile() && info.size > limit) throw inputTooLargeError(limit); + } catch (e) { + // A CliError here is the over-limit refusal and is final; anything else means the path could not + // be stat-ed, which the read below reports as the value-free CLI_NO_INPUT. + if (e instanceof CliError) throw e; + } try { return await readFile(path); } catch { @@ -55,10 +98,15 @@ export async function readFileBytes(path: string): Promise { } /** - * Drain a readable stream (e.g. `process.stdin`) into a single byte buffer. + * Drain a readable stream (e.g. `process.stdin`) into a single byte buffer, refusing an over-limit + * stream against the **running total**: the read is abandoned the moment the accumulated size passes + * `limit`, so the bytes past it are never allocated and no platform allocation ceiling is ever + * reached. * * @param stream - The readable stream to drain. + * @param limit - The maximum number of bytes to accept. Defaults to {@link MAX_INPUT_BYTES}. * @returns The concatenated bytes. + * @throws {CliError} `CLI_INPUT_TOO_LARGE` (exit `65`) as soon as the running total passes `limit`. * @example * ```ts * import { Readable } from "node:stream"; @@ -68,10 +116,83 @@ export async function readFileBytes(path: string): Promise { * bytes.length; // => 4 * ``` */ -export async function readStreamBytes(stream: Readable): Promise { +export async function readStreamBytes( + stream: Readable, + limit: number = MAX_INPUT_BYTES, +): Promise { const chunks: Buffer[] = []; + let total = 0; for await (const chunk of stream as AsyncIterable) { - chunks.push(typeof chunk === "string" ? Buffer.from(chunk, "utf-8") : chunk); + const buf = typeof chunk === "string" ? Buffer.from(chunk, "utf-8") : chunk; + total += buf.length; + if (total > limit) throw inputTooLargeError(limit); + chunks.push(buf); } return Buffer.concat(chunks); } + +/** + * Open a readable stream (e.g. `process.stdin`) as a stream of byte **chunks**, in arrival order. + * Unlike {@link readStreamBytes} this never concatenates: the consumer decides what to keep, which is + * what lets a multi-record command emit a record before the rest of the input has been read. + * + * @param stream - The readable stream to iterate. + * @returns An async iterable of the stream's chunks, string chunks encoded as UTF-8. + * @example + * ```ts + * import { Readable } from "node:stream"; + * import { streamChunks } from "@cosyte/cli"; + * + * const seen: number[] = []; + * for await (const c of streamChunks(Readable.from([Buffer.from("MS"), Buffer.from("H|")]))) { + * seen.push(c.length); + * } + * seen.length; // => 2 + * ``` + */ +export async function* streamChunks(stream: Readable): AsyncGenerator { + for await (const chunk of stream as AsyncIterable) { + yield typeof chunk === "string" ? Buffer.from(chunk, "utf-8") : chunk; + } +} + +/** + * Open a file as a stream of byte **chunks**, with the same value-free failure mode as + * {@link readFileBytes}: a missing, unreadable or non-file path is a `CLI_NO_INPUT` error, never a + * raw filesystem exception. + * + * @param path - The file path to read. + * @returns An async iterable of the file's chunks. + * @throws {CliError} `CLI_NO_INPUT` (exit `66`) when the file cannot be opened or read. + * @example + * ```ts throws + * import { fileChunks } from "@cosyte/cli"; + * + * for await (const chunk of fileChunks("/no/such/file")) chunk.length; // throws CLI_NO_INPUT + * ``` + */ +export async function* fileChunks(path: string): AsyncGenerator { + const noInput = (): CliError => + new CliError( + CLI_CODES.CLI_NO_INPUT, + EXIT.NOINPUT, + `cannot read input file: ${path} (does it exist and is it readable?)`, + ); + + let handle; + try { + handle = await open(path, "r"); + } catch { + throw noInput(); + } + try { + const stream = handle.createReadStream(); + try { + for await (const chunk of stream as AsyncIterable) yield chunk; + } catch { + throw noInput(); + } + } finally { + await handle.close(); + } +} diff --git a/src/core/limits.ts b/src/core/limits.ts new file mode 100644 index 0000000..cfc75c3 --- /dev/null +++ b/src/core/limits.ts @@ -0,0 +1,91 @@ +/** + * The CLI's **own input-size ceiling**, and the value-free refusal it raises. + * + * Node has two hard allocation ceilings a byte pipeline can walk into: a single `Buffer` cannot + * exceed `buffer.constants.MAX_LENGTH` and a single string cannot exceed + * `buffer.constants.MAX_STRING_LENGTH`. Crossing either raises a *platform* exception, which is + * indistinguishable, at the CLI edge, from a bug: it would be reported as an internal error even + * though the input was merely large. That is a lie about whose fault it is, and it is the reason this + * module exists. + * + * So the CLI declares a limit of its **own**, far below both ceilings, checks it against the + * **running total as input is read** (never after a whole oversized input has been assembled), and + * refuses an over-limit invocation with a stable, value-free diagnostic and the **data-error** exit + * code. The number is a documented part of the command surface: it is stated in `cosyte --help` and + * in the command reference, and both are rendered from {@link MAX_INPUT_BYTES} so they cannot drift. + * + * @packageDocumentation + */ + +import { CLI_CODES, CliError } from "./diagnostics.js"; +import { EXIT } from "./exit-codes.js"; + +/** Bytes in one mebibyte: the unit the human half of the limit text is expressed in. */ +const BYTES_PER_MIB = 1024 * 1024; + +/** + * The maximum number of input bytes `cosyte parse` reads in one invocation: **67108864 bytes + * (64 MiB)**. + * + * Chosen to sit far below the smaller of Node's two allocation ceilings + * (`buffer.constants.MAX_STRING_LENGTH`, 536870888 characters), because the bytes read are not the + * peak: a parsed model rendered as JSON is routinely several times the size of the input that + * produced it, so the limit has to leave room for the output as well as the input. A power of two, so + * the byte count and its mebibyte rendering are both exact. + * + * @example + * ```ts + * import { MAX_INPUT_BYTES } from "@cosyte/cli"; + * + * MAX_INPUT_BYTES; // => 67108864 + * ``` + */ +export const MAX_INPUT_BYTES = 64 * BYTES_PER_MIB; + +/** + * Render a byte limit as a **concrete number with an explicit byte-based unit**: `"67108864 bytes + * (64 MiB)"`. Every surface that states the limit (the refusal diagnostic, `cosyte --help`, the + * command reference) renders it through here, so the number a user reads is always the number the + * code enforces, and never a vague description of one. + * + * @param limit - The limit in bytes. Defaults to {@link MAX_INPUT_BYTES}. + * @returns The limit as bytes, plus a mebibyte rendering when it is a whole number of mebibytes. + * @example + * ```ts + * import { describeByteLimit } from "@cosyte/cli"; + * + * describeByteLimit(67108864); // => "67108864 bytes (64 MiB)" + * describeByteLimit(1000); // => "1000 bytes" + * ``` + */ +export function describeByteLimit(limit: number = MAX_INPUT_BYTES): string { + const mib = limit / BYTES_PER_MIB; + return Number.isInteger(mib) + ? `${String(limit)} bytes (${String(mib)} MiB)` + : `${String(limit)} bytes`; +} + +/** + * Build the value-free over-limit refusal: a stable `CLI_INPUT_TOO_LARGE` code, the **data-error** + * exit (`65`, never the internal-error `70`), and a message that names the limit and nothing else. + * The offending input's own size is deliberately not reported either: the limit is the actionable + * fact, and a size read off the input is a property of the input. + * + * @param limit - The limit that was exceeded, in bytes. Defaults to {@link MAX_INPUT_BYTES}. + * @returns The value-free {@link CliError} to resolve the invocation with. + * @example + * ```ts + * import { inputTooLargeError } from "@cosyte/cli"; + * + * inputTooLargeError().exit; // => 65 + * inputTooLargeError().code; // => "CLI_INPUT_TOO_LARGE" + * ``` + */ +export function inputTooLargeError(limit: number = MAX_INPUT_BYTES): CliError { + return new CliError( + CLI_CODES.CLI_INPUT_TOO_LARGE, + EXIT.DATAERR, + `input is larger than the ${describeByteLimit(limit)} this command reads in one ` + + `invocation; split the input and re-run`, + ); +} diff --git a/src/core/parsers.ts b/src/core/parsers.ts index 037fc52..1560f1c 100644 --- a/src/core/parsers.ts +++ b/src/core/parsers.ts @@ -751,11 +751,7 @@ export async function deframeMllp( // Reject an unterminated trailing frame BEFORE de-framing: a VT after the last FS is an open frame // the streaming reader would buffer and silently drop. (MLLP payloads never contain 0x0B/0x1C.) if (bytes.lastIndexOf(0x0b) > bytes.lastIndexOf(0x1c)) { - throw new CliError( - CLI_CODES.CLI_PARSE_FAILED, - EXIT.DATAERR, - "truncated MLLP stream: an unterminated frame at the end of input (no closing FS/CR)", - ); + throw truncatedMllpError(); } const { FrameReader } = await loadOptional("mllp", () => import("@cosyte/mllp")); const payloads: Buffer[] = []; @@ -768,6 +764,28 @@ export async function deframeMllp( return { payloads, warningCount }; } +/** + * The value-free data error for a **truncated MLLP stream**: a frame opened and never closed, which + * the streaming reader buffers and delivers to no callback. Built here so the whole-buffer de-framer + * above and the chunk-by-chunk one in `core/records.ts` refuse the same input with the same code and + * the same words. + * + * @returns A `CLI_PARSE_FAILED` (exit `65`) {@link CliError}. + * @example + * ```ts + * import { truncatedMllpError } from "@cosyte/cli"; + * + * truncatedMllpError().exit; // => 65 + * ``` + */ +export function truncatedMllpError(): CliError { + return new CliError( + CLI_CODES.CLI_PARSE_FAILED, + EXIT.DATAERR, + "truncated MLLP stream: an unterminated frame at the end of input (no closing FS/CR)", + ); +} + /* ── shared internals ────────────────────────────────────────────────────────────────────────── */ /** Decode input bytes as tolerant UTF-8 (the text parsers accept a string). */ diff --git a/src/core/records.ts b/src/core/records.ts new file mode 100644 index 0000000..aaac721 --- /dev/null +++ b/src/core/records.ts @@ -0,0 +1,221 @@ +/** + * **Record streaming**: turning a stream of input byte chunks into a stream of whole records, so a + * multi-record command can emit a record's output *before* the rest of its input has been read. + * + * Two input shapes are multi-record, and both arrive here as chunks rather than as one buffer: an + * **MLLP** stream (each VT-framed frame encloses one HL7 v2 message) and any input under + * **`--ndjson`** (each non-empty line is a record). Neither splitter re-implements a wire format: the + * MLLP one drives `@cosyte/mllp`'s own streaming `FrameReader`, and the NDJSON one splits on the line + * terminator, which is the bulk-data convention's own framing, not a parse of the records inside it. + * + * The size limit is enforced here too, as {@link withinLimit}: a running byte count over the chunk + * stream, so an over-limit input is refused **while it is arriving** rather than after it has been + * assembled. + * + * @packageDocumentation + */ + +import { inputTooLargeError, MAX_INPUT_BYTES } from "./limits.js"; +import { loadOptional, truncatedMllpError } from "./parsers.js"; + +/** A stream of input byte chunks, in arrival order: what every function here consumes. */ +export type ByteChunks = AsyncIterable; + +/** The NDJSON record terminator (`\n`); a preceding `\r` is part of the terminator, not the record. */ +const LINE_FEED = 0x0a; +/** The carriage return that may precede a line feed on a CRLF stream. */ +const CARRIAGE_RETURN = 0x0d; +/** The MLLP start-of-block byte: a frame opens with it. */ +const VERTICAL_TAB = 0x0b; +/** The MLLP end-of-block byte: a frame closes with it (followed by a carriage return). */ +const FILE_SEPARATOR = 0x1c; + +/** View a chunk as a `Buffer` without copying it. */ +function asBuffer(chunk: Uint8Array): Buffer { + return Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk.buffer, chunk.byteOffset, chunk.length); +} + +/** + * Wrap `bytes` as a one-chunk stream, so the streaming record path can also serve a caller that + * hands the CLI a whole buffer (an injected reader, or the programmatic API). The path is the same; + * only the granularity the input arrives at differs. + * + * @param bytes - The whole input. + * @returns A stream yielding exactly one chunk (or none, when `bytes` is empty). + * @example + * ```ts + * import { oneChunk } from "@cosyte/cli"; + * + * let n = 0; + * for await (const c of oneChunk(new Uint8Array([1, 2, 3]))) n += c.length; + * n; // => 3 + * ``` + */ +export async function* oneChunk(bytes: Uint8Array): AsyncGenerator { + // A one-element stream still has to be a real async iterable, so the tick is deliberate: every + // consumer here is written against a source that arrives over time. + await Promise.resolve(); + if (bytes.length > 0) yield bytes; +} + +/** + * Pass chunks through while counting them against the CLI's own size limit, and **refuse the moment + * the running total passes it**: the chunk that crosses the limit is never yielded, the source is + * never pulled again, and nothing downstream ever allocates memory proportional to an oversized + * input. This is why the refusal a caller sees is the CLI's own data error rather than a platform + * allocation failure reported as an internal error. + * + * @param chunks - The input chunk stream. + * @param limit - The maximum number of bytes to accept. Defaults to {@link MAX_INPUT_BYTES}. + * @returns The same chunks, up to the limit. + * @throws {CliError} `CLI_INPUT_TOO_LARGE` (exit `65`) as soon as the running total passes `limit`. + * @example + * ```ts + * import { oneChunk, withinLimit } from "@cosyte/cli"; + * + * const src = withinLimit(oneChunk(new Uint8Array(10)), 4); + * await src.next().catch((e: unknown) => (e as { code: string }).code); // => "CLI_INPUT_TOO_LARGE" + * ``` + */ +export async function* withinLimit( + chunks: ByteChunks, + limit: number = MAX_INPUT_BYTES, +): AsyncGenerator { + let total = 0; + for await (const chunk of chunks) { + total += chunk.length; + if (total > limit) throw inputTooLargeError(limit); + yield chunk; + } +} + +/** + * Drain a chunk stream into one buffer: the single-message path, where the parser needs the whole + * message anyway. The size limit is already applied by {@link withinLimit} upstream, so this cannot + * accumulate more than the limit allows. + * + * @param chunks - The input chunk stream. + * @returns The concatenated bytes. + * @example + * ```ts + * import { collectChunks, oneChunk } from "@cosyte/cli"; + * + * (await collectChunks(oneChunk(new TextEncoder().encode("MSH|")))).length; // => 4 + * ``` + */ +export async function collectChunks(chunks: ByteChunks): Promise { + const parts: Buffer[] = []; + for await (const chunk of chunks) parts.push(asBuffer(chunk)); + return Buffer.concat(parts); +} + +/** True iff every byte of `line` is whitespace (so the line carries no record). */ +function isBlank(line: Uint8Array): boolean { + return new TextDecoder("utf-8", { fatal: false }).decode(line).trim().length === 0; +} + +/** Drop the `\r` of a CRLF terminator, which belongs to the terminator rather than to the record. */ +function stripCarriageReturn(line: Buffer): Buffer { + return line.length > 0 && line[line.length - 1] === CARRIAGE_RETURN + ? line.subarray(0, line.length - 1) + : line; +} + +/** + * Split a chunk stream into **NDJSON records**: each non-empty, newline-terminated line, yielded as + * soon as its terminator arrives. A record split across two chunks is carried over; a final record + * with no trailing newline is yielded at end of stream; a blank line carries no record and is + * skipped. + * + * @param chunks - The input chunk stream. + * @returns A stream of record bytes, in input order. + * @example + * ```ts + * import { ndjsonRecords, oneChunk } from "@cosyte/cli"; + * + * const src = oneChunk(new TextEncoder().encode('{"a":1}\n\n{"b":2}\n')); + * const out: string[] = []; + * for await (const r of ndjsonRecords(src)) out.push(new TextDecoder().decode(r)); + * out.length; // => 2 + * ``` + */ +export async function* ndjsonRecords(chunks: ByteChunks): AsyncGenerator { + let carry: Buffer = Buffer.alloc(0); + for await (const chunk of chunks) { + const buf = carry.length > 0 ? Buffer.concat([carry, asBuffer(chunk)]) : asBuffer(chunk); + let start = 0; + for (;;) { + const end = buf.indexOf(LINE_FEED, start); + if (end < 0) break; + const line = stripCarriageReturn(buf.subarray(start, end)); + if (!isBlank(line)) yield line; + start = end + 1; + } + // Copy the tail so the chunk it came from can be released while we wait for the next one. + carry = start === 0 ? asBuffer(buf) : Buffer.from(buf.subarray(start)); + } + const last = stripCarriageReturn(carry); + if (!isBlank(last)) yield last; +} + +/** + * De-frame a chunk stream into **MLLP frame payloads** (each an enclosed HL7 v2 message), yielded as + * each frame completes. The framing is `@cosyte/mllp`'s own streaming `FrameReader`, fed chunk by + * chunk; the CLI only tracks where the frame bytes fell so it can tell a **truncated** stream from a + * complete one. + * + * **Truncation is a data error, never a silent drop.** An unterminated trailing frame (a start-of-block + * byte after the last end-of-block byte) is delivered by no callback, so it would otherwise vanish + * with a green exit. It is detected at end of stream and raised, after the frames that did complete + * have already been yielded: their output stands, and the invocation still resolves to the data error. + * + * @param chunks - The input chunk stream. + * @returns A stream of enclosed HL7 payloads, in frame order. + * @throws {CliError} `CLI_PARSER_UNAVAILABLE` (exit `69`) if `@cosyte/mllp` is absent; + * `CLI_PARSE_FAILED` (exit `65`) at end of stream on a truncated final frame. + * @example + * ```ts + * import { mllpFrames, oneChunk } from "@cosyte/cli"; + * + * const framed = new Uint8Array([0x0b, 0x4d, 0x53, 0x48, 0x1c, 0x0d]); + * const seen: number[] = []; + * for await (const p of mllpFrames(oneChunk(framed))) seen.push(p.length); + * seen.length; // => 1 + * ``` + */ +export async function* mllpFrames(chunks: ByteChunks): AsyncGenerator { + const { FrameReader } = await loadOptional("mllp", () => import("@cosyte/mllp")); + const ready: Buffer[] = []; + const reader = new FrameReader({ + onFrame: (payload) => ready.push(Buffer.from(payload)), + onWarning: () => undefined, + // The CLI's own documented input limit is the binding one, so the reader's smaller default frame + // ceiling is raised to it: an over-limit MLLP stream is refused by the limit the help output and + // the command reference state, not by an undocumented one underneath it. + maxFrameSizeBytes: MAX_INPUT_BYTES, + }); + + // Offsets of the last start-of-block and end-of-block bytes seen anywhere in the stream: an + // enclosed HL7 v2 payload never carries either byte, so their order at end of stream is the same + // truncation test the whole-buffer de-framer applies. + let lastStart = -1; + let lastEnd = -1; + let offset = 0; + + for await (const chunk of chunks) { + const buf = asBuffer(chunk); + const start = buf.lastIndexOf(VERTICAL_TAB); + if (start >= 0) lastStart = offset + start; + const end = buf.lastIndexOf(FILE_SEPARATOR); + if (end >= 0) lastEnd = offset + end; + offset += buf.length; + + reader.push(buf); + while (ready.length > 0) { + const frame = ready.shift(); + if (frame !== undefined) yield frame; + } + } + + if (lastStart > lastEnd) throw truncatedMllpError(); +} diff --git a/src/core/run.ts b/src/core/run.ts index 1117baa..3f629a5 100644 --- a/src/core/run.ts +++ b/src/core/run.ts @@ -19,12 +19,29 @@ import { validateCommand } from "../commands/validate.js"; import { CLI_CODES, CliError, toCliError } from "./diagnostics.js"; import { EXIT } from "./exit-codes.js"; import type { RunDeps } from "./io.js"; +import { describeByteLimit, MAX_INPUT_BYTES } from "./limits.js"; import { extractPhiPosture } from "./phi.js"; import type { RunResult } from "./result.js"; import { VERSION } from "./version.js"; -/** The value-free `--help` text. Names commands, flags, and exit codes, never any input. */ -const HELP = `cosyte: a PHI-safe developer CLI over the @cosyte/* healthcare parsers +/** + * Build the value-free `--help` text. Names commands, flags, the input-size limit, and exit codes, + * never any input. + * + * The limit is **rendered from the constant the code enforces**, not typed in: the help output and + * the published command reference state one number because there is only one number to state. + * + * @param limitBytes - The input-size limit to state. Defaults to {@link MAX_INPUT_BYTES}. + * @returns The full help text. + * @example + * ```ts + * import { helpText } from "@cosyte/cli"; + * + * helpText(1024).includes("1024 bytes"); // => true + * ``` + */ +export function helpText(limitBytes: number = MAX_INPUT_BYTES): string { + return `cosyte: a PHI-safe developer CLI over the @cosyte/* healthcare parsers Usage: cosyte [options] @@ -63,8 +80,15 @@ map-codes options (the positional is a BYO FHIR ConceptMap; a code is not PHI): --version The source code system version (optional) --display The source display (optional) +Input size: + parse reads at most ${describeByteLimit(limitBytes)} of input per invocation. A larger input is + refused with CLI_INPUT_TOO_LARGE and the data-error exit code (65), never the internal-error + code; split the input and re-run. Multi-record output (--ndjson, MLLP) is written record by + record as it is parsed, so a refusal can arrive after some records have already been emitted: + the exit code, not the output, is the signal that the run did not complete. + Exit codes: - 0 success / valid 65 data error (unparseable / undetected format) + 0 success / valid 65 data error (unparseable / undetected format / input too large) 1 invalid (validate) 66 no input (missing/unreadable file) 2 usage error 69 unavailable (a capability is not yet built, e.g. redact) 70 internal error @@ -73,6 +97,10 @@ PHI posture: the parsed model goes to stdout (the data channel you chose); every stderr is value-free (codes and positions only, never a field value) unless you pass the loud, opt-in --unsafe-show-values (which permits a bounded input excerpt in a failure diagnostic). `; +} + +/** The rendered help text for this build: the one place the default limit is stated. */ +const HELP = helpText(); /** True if `argv` requests help (`-h`/`--help` anywhere). */ function wantsHelp(argv: readonly string[]): boolean { diff --git a/src/index.ts b/src/index.ts index 6160111..d587822 100644 --- a/src/index.ts +++ b/src/index.ts @@ -24,11 +24,23 @@ export { detectionError, asCosyteFormat, DETECTABLE_FORMATS, + DETECT_PREFIX_BYTES, KNOWN_FORMATS, type CosyteFormat, type DetectResult, } from "./core/format.js"; +export { MAX_INPUT_BYTES, describeByteLimit, inputTooLargeError } from "./core/limits.js"; + +export { + collectChunks, + mllpFrames, + ndjsonRecords, + oneChunk, + withinLimit, + type ByteChunks, +} from "./core/records.js"; + export { OP_SUPPORT, supportsOp, @@ -38,6 +50,7 @@ export { fmtFormat, validateFormat, deframeMllp, + truncatedMllpError, loadOptional, loadOptionalPackage, valueFreeLocator, @@ -58,13 +71,26 @@ export { type CliCode, } from "./core/diagnostics.js"; -export { resolveInput, type ResolvedInput, type InputResolution } from "./core/input.js"; +export { + resolveInput, + resolveInputStream, + type ResolvedInput, + type InputResolution, + type ResolvedInputStream, + type InputStreamResolution, +} from "./core/input.js"; export { extractStableCode, parseFailureResult } from "./core/wrap.js"; export { formatHl7Position, type Finding } from "./core/findings.js"; -export { readFileBytes, readStreamBytes, type RunDeps } from "./core/io.js"; +export { + readFileBytes, + readStreamBytes, + fileChunks, + streamChunks, + type RunDeps, +} from "./core/io.js"; export { VALUE_FREE, @@ -80,7 +106,7 @@ export { deidStatus, DEID_UNAVAILABLE_REASON, type DeidAvailability } from "./co export type { RunResult } from "./core/result.js"; -export { run } from "./core/run.js"; +export { run, helpText } from "./core/run.js"; export { parseCommand } from "./commands/parse.js"; diff --git a/test/formats.test.ts b/test/formats.test.ts index 9d74b20..1332144 100644 --- a/test/formats.test.ts +++ b/test/formats.test.ts @@ -288,11 +288,17 @@ describe("streaming, per-record isolation + quiet on multi-record", () => { it("a truncated trailing MLLP frame is a data error (65), never a silent-dropped message", async () => { // One complete frame, then a VT that opens a second message with no closing FS/CR: the streaming // de-framer would buffer and silently drop it. It must be a value-free data error, not exit 0. + // + // Records are emitted as they are parsed, so the frame that DID complete has already reached the + // data channel by the time the truncation is detected at end of stream. That partial stream is + // never presented as complete: the exit code carries the failure, and only the exit code can. const truncated = new Uint8Array([...frame(HL7_MSG), 0x0b, ...enc.encode("MSH|^~\\&|C|D\r")]); const r = await run(["parse", "s.mllp"], bytesDeps(truncated)); expect(r.exit).toBe(EXIT.DATAERR); + expect(r.exit).not.toBe(EXIT.OK); expect(r.stderr).toContain("CLI_PARSE_FAILED"); - expect(r.stdout).toBe(""); + expect(r.stdout.trim().split("\n")).toHaveLength(1); // the one complete frame, and no more + expect(r.stderr).not.toContain("MSH"); }); it("inspect rejects a truncated MLLP stream too (never a fake frame count + exit 0)", async () => { diff --git a/test/limit-docs.test.ts b/test/limit-docs.test.ts new file mode 100644 index 0000000..eae8fbb --- /dev/null +++ b/test/limit-docs.test.ts @@ -0,0 +1,89 @@ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import type { CliError } from "../src/core/diagnostics.js"; +import type { RunDeps } from "../src/core/io.js"; +import { describeByteLimit, inputTooLargeError, MAX_INPUT_BYTES } from "../src/core/limits.js"; +import { helpText, run } from "../src/core/run.js"; + +/** + * The documented size limit is **one number, stated in three places**: the refusal a caller reads on + * stderr, `cosyte --help`, and the published command reference. + * + * The help output is rendered from the constant the code enforces, so changing the constant changes + * the help text with it (proved below by rendering a different limit). The command reference is a + * static document, so this suite is what keeps it honest: move the constant without editing the + * reference and these assertions red, in the same build, before the two can disagree in public. + */ + +const REFERENCE = readFileSync( + join(import.meta.dirname, "..", "docs-content", "reference-commands.md"), + "utf-8", +); +const README = readFileSync(join(import.meta.dirname, "..", "README.md"), "utf-8"); + +const noDeps: RunDeps = { + readFile: () => Promise.resolve(new Uint8Array()), + readStdin: () => Promise.resolve(new Uint8Array()), +}; + +/** Every " bytes" claim in a text, as numbers. */ +function byteClaims(text: string): number[] { + return [...text.matchAll(/(\d[\d,]*)\s*bytes/g)].map((m) => Number(m[1]?.replace(/,/g, ""))); +} + +describe("the limit is stated as a concrete number with an explicit byte-based unit", () => { + it("on the refusal diagnostic", () => { + const message = inputTooLargeError().message; + expect(message).toMatch(/\b\d+ bytes\b/); + expect(byteClaims(message)).toContain(MAX_INPUT_BYTES); + }); + + it("in the help output", async () => { + const help = (await run(["--help"], noDeps)).stdout; + expect(help).toMatch(/\b\d+ bytes\b/); + expect(byteClaims(help)).toContain(MAX_INPUT_BYTES); + }); + + it("in the published command reference", () => { + expect(REFERENCE).toMatch(/\b\d+ bytes\b/); + expect(byteClaims(REFERENCE)).toContain(MAX_INPUT_BYTES); + }); +}); + +describe("the surfaces cannot disagree in one build", () => { + it("the help output and the command reference state the same limit, and no other", async () => { + const help = (await run(["--help"], noDeps)).stdout; + expect(new Set(byteClaims(help))).toStrictEqual(new Set([MAX_INPUT_BYTES])); + expect(new Set(byteClaims(REFERENCE))).toStrictEqual(new Set([MAX_INPUT_BYTES])); + expect(new Set(byteClaims(README))).toStrictEqual(new Set([MAX_INPUT_BYTES])); + }); + + it("all three carry the same rendered limit text", async () => { + const rendered = describeByteLimit(); + expect((await run(["--help"], noDeps)).stdout).toContain(rendered); + expect(REFERENCE).toContain(rendered); + expect(README).toContain(rendered); + expect(inputTooLargeError().message).toContain(rendered); + }); + + it("changing the limit changes the help text: it is rendered, never typed in", () => { + expect(helpText(1024)).toContain("1024 bytes"); + expect(helpText(1024)).not.toContain(describeByteLimit()); + expect(helpText()).toContain(describeByteLimit()); + }); + + it("the refusal names whatever limit was applied", () => { + const e: CliError = inputTooLargeError(4096); + expect(e.message).toContain("4096 bytes"); + expect(e.code).toBe("CLI_INPUT_TOO_LARGE"); + }); + + it("the help output names the refusal's code and its data-error exit code", async () => { + const help = (await run(["--help"], noDeps)).stdout; + expect(help).toContain("CLI_INPUT_TOO_LARGE"); + expect(help).toContain("65"); + }); +}); diff --git a/test/limits.test.ts b/test/limits.test.ts new file mode 100644 index 0000000..86b7e70 --- /dev/null +++ b/test/limits.test.ts @@ -0,0 +1,171 @@ +import { constants as bufferConstants } from "node:buffer"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Readable } from "node:stream"; + +import { afterAll, describe, expect, it } from "vitest"; + +import { CliError } from "../src/core/diagnostics.js"; +import { EXIT } from "../src/core/exit-codes.js"; +import { fileChunks, readFileBytes, readStreamBytes, streamChunks } from "../src/core/io.js"; +import { describeByteLimit, inputTooLargeError, MAX_INPUT_BYTES } from "../src/core/limits.js"; +import { collectChunks, oneChunk, withinLimit } from "../src/core/records.js"; + +/** + * The CLI's own input-size ceiling: where it sits relative to the platform's, how it is stated, and + * that it is enforced against the **running total** as input arrives rather than after an oversized + * input has been assembled. The last property is the whole point: a check that runs only after the + * bytes are in memory cannot stop the platform allocation failure it exists to pre-empt. + */ + +const dir = mkdtempSync(join(tmpdir(), "cosyte-cli-limits-")); +afterAll(() => rmSync(dir, { recursive: true, force: true })); + +/** A chunk source that would produce `total` bytes, and records how many it was actually asked for. */ +function countedSource( + total: number, + chunkSize: number, +): { chunks: () => AsyncGenerator; produced: () => number } { + let produced = 0; + const filler = Buffer.alloc(chunkSize, 0x61); + async function* chunks(): AsyncGenerator { + await Promise.resolve(); + while (produced < total) { + const size = Math.min(chunkSize, total - produced); + produced += size; + yield size === chunkSize ? filler : filler.subarray(0, size); + } + } + return { chunks, produced: () => produced }; +} + +describe("the documented limit sits below the platform's own ceilings", () => { + it("is strictly less than the smaller ceiling (MAX_STRING_LENGTH)", () => { + expect(MAX_INPUT_BYTES).toBeLessThan(bufferConstants.MAX_STRING_LENGTH); + expect(MAX_INPUT_BYTES).toBeLessThan(bufferConstants.MAX_LENGTH); + expect(bufferConstants.MAX_STRING_LENGTH).toBeLessThanOrEqual(bufferConstants.MAX_LENGTH); + }); + + it("is a concrete number of bytes, stated with an explicit byte-based unit", () => { + expect(Number.isInteger(MAX_INPUT_BYTES)).toBe(true); + expect(describeByteLimit()).toBe("67108864 bytes (64 MiB)"); + expect(describeByteLimit()).toMatch(/\b\d+ bytes\b/); + expect(describeByteLimit(1000)).toBe("1000 bytes"); + }); + + it("the refusal is a value-free data error naming the limit, never the internal-error code", () => { + const e = inputTooLargeError(); + expect(e.code).toBe("CLI_INPUT_TOO_LARGE"); + expect(e.exit).toBe(EXIT.DATAERR); + expect(e.exit).not.toBe(EXIT.SOFTWARE); + expect(e.message).toContain(describeByteLimit()); + }); +}); + +describe("withinLimit: the count runs as the input arrives", () => { + it("passes a stream that stays under the limit through unchanged", async () => { + const bytes = new TextEncoder().encode("MSH|^~\\&|"); + const out = await collectChunks(withinLimit(oneChunk(bytes), 1024)); + expect(new TextDecoder().decode(out)).toBe("MSH|^~\\&|"); + }); + + it("refuses at limit + 1 byte with the value-free data error", async () => { + const bytes = new Uint8Array(65); + await expect(collectChunks(withinLimit(oneChunk(bytes), 64))).rejects.toMatchObject({ + code: "CLI_INPUT_TOO_LARGE", + exit: EXIT.DATAERR, + }); + }); + + it("accepts exactly the limit", async () => { + const bytes = new Uint8Array(64); + expect((await collectChunks(withinLimit(oneChunk(bytes), 64))).length).toBe(64); + }); + + it("stops pulling the source the moment the total crosses: never drains a huge input", async () => { + // The source would produce more bytes than a single string can hold, which is the failure mode + // this limit exists to pre-empt: assembling it first and checking afterwards is exactly what a + // post-hoc check does, and it cannot get here. 8 MiB chunks, one refusal, nothing near the ceiling. + const chunkSize = 8 * 1024 * 1024; + const src = countedSource(bufferConstants.MAX_STRING_LENGTH + chunkSize, chunkSize); + await expect(collectChunks(withinLimit(src.chunks(), MAX_INPUT_BYTES))).rejects.toMatchObject({ + code: "CLI_INPUT_TOO_LARGE", + }); + expect(src.produced()).toBeLessThanOrEqual(MAX_INPUT_BYTES + chunkSize); + expect(src.produced()).toBeLessThan(bufferConstants.MAX_STRING_LENGTH); + }, 30_000); +}); + +describe("readStreamBytes: the same running count on the whole-input reader", () => { + it("refuses a stream past the limit without concatenating it", async () => { + const stream = Readable.from([Buffer.alloc(40), Buffer.alloc(40)]); + await expect(readStreamBytes(stream, 64)).rejects.toMatchObject({ + code: "CLI_INPUT_TOO_LARGE", + exit: EXIT.DATAERR, + }); + }); + + it("still returns a stream that fits", async () => { + const bytes = await readStreamBytes(Readable.from([Buffer.from("MS"), Buffer.from("H|")]), 64); + expect(new TextDecoder().decode(bytes)).toBe("MSH|"); + }); +}); + +describe("readFileBytes: an oversized file is refused before it is read", () => { + it("refuses a file whose size already exceeds the limit", async () => { + const p = join(dir, "big.txt"); + writeFileSync(p, "x".repeat(200)); + await expect(readFileBytes(p, 100)).rejects.toMatchObject({ + code: "CLI_INPUT_TOO_LARGE", + exit: EXIT.DATAERR, + }); + }); + + it("reads a file that fits", async () => { + const p = join(dir, "small.txt"); + writeFileSync(p, "MSH|"); + expect(new TextDecoder().decode(await readFileBytes(p, 100))).toBe("MSH|"); + }); + + it("keeps the value-free no-input error for a missing file", async () => { + await expect(readFileBytes(join(dir, "nope.txt"), 100)).rejects.toMatchObject({ + code: "CLI_NO_INPUT", + exit: EXIT.NOINPUT, + }); + }); +}); + +describe("chunk readers", () => { + it("fileChunks streams a file's bytes", async () => { + const p = join(dir, "chunked.txt"); + writeFileSync(p, "MSH|^~\\&|A|B\r"); + const out = await collectChunks(fileChunks(p)); + expect(new TextDecoder().decode(out)).toBe("MSH|^~\\&|A|B\r"); + }); + + it("fileChunks raises the value-free CLI_NO_INPUT for a missing file", async () => { + await expect(collectChunks(fileChunks(join(dir, "gone.txt")))).rejects.toBeInstanceOf(CliError); + await expect(collectChunks(fileChunks(join(dir, "gone.txt")))).rejects.toMatchObject({ + code: "CLI_NO_INPUT", + }); + }); + + it("fileChunks raises CLI_NO_INPUT for a directory (openable, unreadable as a file)", async () => { + await expect(collectChunks(fileChunks(dir))).rejects.toMatchObject({ code: "CLI_NO_INPUT" }); + }); + + it("streamChunks yields each chunk as it arrives, encoding string chunks as utf-8", async () => { + const sizes: number[] = []; + for await (const chunk of streamChunks(Readable.from([Buffer.from("MS"), "H|"]))) { + sizes.push(chunk.length); + } + expect(sizes).toStrictEqual([2, 2]); + }); + + it("oneChunk yields nothing for empty input", async () => { + const seen: number[] = []; + for await (const chunk of oneChunk(new Uint8Array())) seen.push(chunk.length); + expect(seen).toStrictEqual([]); + }); +}); diff --git a/test/parse-bulk.test.ts b/test/parse-bulk.test.ts new file mode 100644 index 0000000..01bb27b --- /dev/null +++ b/test/parse-bulk.test.ts @@ -0,0 +1,369 @@ +import { constants as bufferConstants } from "node:buffer"; + +import { describe, expect, it } from "vitest"; + +import { EXIT } from "../src/core/exit-codes.js"; +import type { RunDeps } from "../src/core/io.js"; +import { describeByteLimit, MAX_INPUT_BYTES } from "../src/core/limits.js"; +import { oneChunk } from "../src/core/records.js"; +import { run } from "../src/core/run.js"; + +/** + * Bulk input for `cosyte parse`: an input past the documented size limit is a value-free refusal + * naming the limit and a **data error**, never the internal-error code; and a multi-record input + * (`--ndjson`, or MLLP frames) emits each record's line **as it is parsed**, before the rest of the + * input has been read. + * + * The refusal is checked on all three input shapes independently, at the real documented limit, with + * sources that would run past the platform's own string ceiling if anything here drained them. + */ + +const enc = new TextEncoder(); +const CHUNK = 8 * 1024 * 1024; + +/** A VT/FS-framed MLLP frame around an HL7 payload. */ +const frame = (hl7: string): number[] => [0x0b, ...enc.encode(hl7), 0x1c, 0x0d]; +const HL7_MSG = "MSH|^~\\&|A|B|C|D|20240101||ADT^A01|1|P|2.5\rPID|1||123^^^HOSP\r"; +const HL7_HEAD = "MSH|^~\\&|A|B|C|D|20240101||ADT^A01|1|P|2.5\r"; +const FHIR_RECORD = '{"resourceType":"Patient","id":"ZZSENTINELBULK"}'; + +/** + * A chunk source that opens with `lead` and then pads with filler until it would have produced + * `total` bytes, counting what it was actually asked for. `total` is past + * `buffer.constants.MAX_STRING_LENGTH`, so an implementation that read the whole input before + * checking its size would fail on the platform's ceiling instead of refusing on the CLI's limit. + */ +function overLimitSource( + lead: Uint8Array, + total: number = bufferConstants.MAX_STRING_LENGTH + CHUNK, +): { open: () => AsyncGenerator; produced: () => number } { + let produced = 0; + const filler = Buffer.alloc(CHUNK, 0x61); + async function* open(): AsyncGenerator { + await Promise.resolve(); + const first = Buffer.concat([Buffer.from(lead), filler.subarray(0, CHUNK - lead.length)]); + produced += first.length; + yield first; + while (produced < total) { + produced += CHUNK; + yield filler; + } + } + return { open, produced: () => produced }; +} + +/** Deps whose stdin is a chunk stream; the whole-input readers reject, proving they are unused. */ +function chunkDeps(open: () => AsyncIterable, sink?: (chunk: string) => void): RunDeps { + const base = { + readFile: () => Promise.reject(new Error("the whole-input reader must not be used here")), + readStdin: () => Promise.reject(new Error("the whole-input reader must not be used here")), + openStdin: open, + }; + return sink === undefined ? base : { ...base, writeStdout: sink }; +} + +/** A promise plus its resolver, for holding an input open until the output has been observed. */ +function deferred(): { promise: Promise; resolve: (value: T) => void } { + let resolve: (value: T) => void = () => undefined; + const promise = new Promise((r) => (resolve = r)); + return { promise, resolve }; +} + +/** Resolve to `marker` if `promise` has not settled within `ms`, without leaving a timer behind. */ +async function within(promise: Promise, ms: number, marker: T): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(marker), ms); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + +describe("an input past the documented limit is refused, value-free, as a data error", () => { + it("a single message: exit 65 naming the limit, never exit 70, never a byte of the input", async () => { + const src = overLimitSource(enc.encode(HL7_HEAD)); + const r = await run(["parse", "-"], chunkDeps(src.open)); + + expect(r.exit).toBe(EXIT.DATAERR); + expect(r.exit).not.toBe(EXIT.SOFTWARE); + expect(r.stderr).toContain("CLI_INPUT_TOO_LARGE"); + expect(r.stderr).toContain(describeByteLimit()); + expect(r.stderr).not.toContain("MSH"); + expect(r.stderr).not.toContain("aaa"); + expect(r.stdout).toBe(""); + // The refusal arrived while the input was still arriving: the source was never drained. + expect(src.produced()).toBeLessThanOrEqual(MAX_INPUT_BYTES + CHUNK); + expect(src.produced()).toBeLessThan(bufferConstants.MAX_STRING_LENGTH); + }, 30_000); + + it("an --ndjson input: exit 65 naming the limit, after the records that did parse", async () => { + const src = overLimitSource(enc.encode(`${FHIR_RECORD}\n`)); + const r = await run(["parse", "-", "--ndjson"], chunkDeps(src.open)); + + expect(r.exit).toBe(EXIT.DATAERR); + expect(r.exit).not.toBe(EXIT.SOFTWARE); + expect(r.stderr).toContain("CLI_INPUT_TOO_LARGE"); + expect(r.stderr).toContain(describeByteLimit()); + expect(r.stderr).not.toContain("ZZSENTINELBULK"); + // The one complete record reached the data channel before the refusal; the exit code, not the + // output, says the run did not complete. + expect(r.stdout.trim().split("\n")).toHaveLength(1); + expect(src.produced()).toBeLessThanOrEqual(MAX_INPUT_BYTES + CHUNK); + expect(src.produced()).toBeLessThan(bufferConstants.MAX_STRING_LENGTH); + }, 30_000); + + it("an MLLP input: exit 65 naming the limit, after the frames that did complete", async () => { + // One complete frame, then a frame that opens and never closes: the stream runs past the limit. + const src = overLimitSource(new Uint8Array([...frame(HL7_MSG), 0x0b])); + const r = await run(["parse", "-"], chunkDeps(src.open)); + + expect(r.exit).toBe(EXIT.DATAERR); + expect(r.exit).not.toBe(EXIT.SOFTWARE); + expect(r.stderr).toContain("CLI_INPUT_TOO_LARGE"); + expect(r.stderr).toContain(describeByteLimit()); + expect(r.stderr).not.toContain("MSH"); + expect(r.stdout.trim().split("\n")).toHaveLength(1); + expect(src.produced()).toBeLessThanOrEqual(MAX_INPUT_BYTES + CHUNK); + expect(src.produced()).toBeLessThan(bufferConstants.MAX_STRING_LENGTH); + }, 30_000); + + it("a whole-input reader handing over exactly the limit plus one byte is refused too", async () => { + const bytes = Buffer.alloc(MAX_INPUT_BYTES + 1, 0x61); + bytes.write(HL7_HEAD, 0, "utf-8"); + const r = await run(["parse", "big.hl7"], { + readFile: () => Promise.resolve(bytes), + readStdin: () => Promise.resolve(new Uint8Array()), + }); + + expect(r.exit).toBe(EXIT.DATAERR); + expect(r.exit).not.toBe(EXIT.SOFTWARE); + expect(r.stderr).toContain("CLI_INPUT_TOO_LARGE"); + expect(r.stderr).toContain(describeByteLimit()); + }, 30_000); + + it("an input under the limit is never refused for its size (the negative control)", async () => { + // The same input shape, well under the ceiling: nothing about it triggers the refusal, so the + // assertions above are about the size and not about the shape of the fixture. + const bytes = Buffer.alloc(1024, 0x61); + bytes.write(HL7_HEAD, 0, "utf-8"); + const r = await run(["parse", "small.hl7"], { + readFile: () => Promise.resolve(bytes), + readStdin: () => Promise.resolve(new Uint8Array()), + }); + expect(r.stderr).not.toContain("CLI_INPUT_TOO_LARGE"); + expect(r.exit).toBe(EXIT.OK); + }); +}); + +describe("multi-record output is emitted before the whole input has been read", () => { + it("--ndjson: the first record's line lands while the rest of the input is still withheld", async () => { + const gate = deferred(); + const firstLine = deferred(); + const lines: string[] = []; + let drained = false; + + async function* source(): AsyncGenerator { + yield enc.encode(`{"resourceType":"Patient","id":"one"}\n`); + await gate.promise; // the remainder of the input never arrives until the output is observed + yield enc.encode(`{"resourceType":"Patient","id":"two"}\n`); + drained = true; + } + + const running = run( + ["parse", "-", "--ndjson", "--format", "fhir"], + chunkDeps(source, (chunk) => { + lines.push(chunk); + if (lines.length === 1) firstLine.resolve(chunk); + }), + ); + + const first = await within(firstLine.promise, 2_000, "NEVER ARRIVED"); + expect(first).not.toBe("NEVER ARRIVED"); + expect(JSON.parse(first) as { record: number }).toMatchObject({ record: 0 }); + expect(drained).toBe(false); // a read-it-all-then-emit implementation cannot reach this line + + gate.resolve(); + const r = await running; + expect(r.exit).toBe(EXIT.OK); + expect(lines).toHaveLength(2); + expect(r.stdout).toBe(""); // the output went to the sink as it was produced, not into the result + expect(drained).toBe(true); + }); + + it("MLLP: the first frame's line lands while the rest of the stream is still withheld", async () => { + const gate = deferred(); + const firstLine = deferred(); + const lines: string[] = []; + let drained = false; + + async function* source(): AsyncGenerator { + yield new Uint8Array(frame(HL7_MSG)); + await gate.promise; + yield new Uint8Array(frame(HL7_MSG)); + drained = true; + } + + const running = run( + ["parse", "-", "--format", "mllp"], + chunkDeps(source, (chunk) => { + lines.push(chunk); + if (lines.length === 1) firstLine.resolve(chunk); + }), + ); + + const first = await within(firstLine.promise, 2_000, "NEVER ARRIVED"); + expect(first).not.toBe("NEVER ARRIVED"); + expect(JSON.parse(first) as { record: number; format: string }).toMatchObject({ + record: 0, + format: "hl7", + }); + expect(drained).toBe(false); + + gate.resolve(); + const r = await running; + expect(r.exit).toBe(EXIT.OK); + expect(lines).toHaveLength(2); + }); + + it("a record split across two chunks is still one record", async () => { + const lines: string[] = []; + async function* source(): AsyncGenerator { + await Promise.resolve(); + yield enc.encode('{"resourceType":"Pat'); + yield enc.encode('ient","id":"split"}\n{"resourceType":"Patient","id":"b"}'); + } + const r = await run( + ["parse", "-", "--ndjson", "--format", "fhir"], + chunkDeps(source, (chunk) => lines.push(chunk)), + ); + expect(r.exit).toBe(EXIT.OK); + expect(lines).toHaveLength(2); + expect(lines.join("")).toContain('"split"'); + }); +}); + +describe("a fatal condition part way through never resolves to OK", () => { + it("keeps the emitted lines and exits with the failing record stream's own code", async () => { + // Three good records, then a frame that opens and never closes: a truncated stream, detected at + // the end, after three lines have already been written. + const lines: string[] = []; + async function* source(): AsyncGenerator { + await Promise.resolve(); + yield new Uint8Array([...frame(HL7_MSG), ...frame(HL7_MSG), ...frame(HL7_MSG), 0x0b]); + yield enc.encode("MSH|^~\\&|C|D\r"); + } + const r = await run( + ["parse", "-"], + chunkDeps(source, (chunk) => lines.push(chunk)), + ); + + expect(lines).toHaveLength(3); + expect(r.exit).not.toBe(EXIT.OK); + expect(r.exit).toBe(EXIT.DATAERR); + expect(r.stderr).toContain("CLI_PARSE_FAILED"); + expect(r.stderr).not.toContain("MSH"); + }); + + it("a downstream consumer closing the pipe ends the stream with a value-free write failure", async () => { + const lines: string[] = []; + async function* source(): AsyncGenerator { + await Promise.resolve(); + for (let i = 0; i < 5; i += 1) yield enc.encode(`{"resourceType":"Patient","id":"p"}\n`); + } + const r = await run( + ["parse", "-", "--ndjson", "--format", "fhir"], + chunkDeps(source, (chunk) => { + lines.push(chunk); + if (lines.length >= 2) throw new Error("EPIPE"); + }), + ); + + expect(lines).toHaveLength(2); + expect(r.exit).not.toBe(EXIT.OK); + expect(r.stderr).toContain("CLI_OUTPUT_WRITE_FAILED"); + expect(r.stderr).not.toContain("EPIPE"); + expect(r.stderr).not.toContain("resourceType"); + }); +}); + +describe("per-record isolation survives incremental emission", () => { + it("a bad record among good ones gets its own value-free line, and the run exits 65", async () => { + const lines: string[] = []; + async function* source(): AsyncGenerator { + await Promise.resolve(); + yield enc.encode( + `{"resourceType":"Patient","id":"a"}\n` + + `{ not json ZZSENTINELBAD\n` + + `{"resourceType":"Patient","id":"c"}\n`, + ); + } + const r = await run( + ["parse", "-", "--ndjson", "--format", "fhir"], + chunkDeps(source, (chunk) => lines.push(chunk)), + ); + + expect(lines).toHaveLength(3); + expect(r.exit).toBe(EXIT.DATAERR); + const parsed = lines.map((l) => JSON.parse(l) as { record: number; error?: string }); + expect(parsed.map((p) => p.record)).toStrictEqual([0, 1, 2]); + expect(parsed[1]?.error).toBeDefined(); + expect(parsed[0]?.error).toBeUndefined(); + expect(lines.join("")).not.toContain("ZZSENTINELBAD"); + expect(r.stderr).not.toContain("ZZSENTINELBAD"); + expect(r.stderr).toContain("1 failed"); + }); + + it("blank lines carry no record and are skipped, not failed", async () => { + const lines: string[] = []; + async function* source(): AsyncGenerator { + await Promise.resolve(); + yield enc.encode(`{"resourceType":"Patient","id":"a"}\n\n \n{"resourceType":"Patient"}\n`); + } + const r = await run( + ["parse", "-", "--ndjson", "--format", "fhir"], + chunkDeps(source, (chunk) => lines.push(chunk)), + ); + expect(r.exit).toBe(EXIT.OK); + expect(lines).toHaveLength(2); + }); +}); + +describe("an input that frames no record at all is a data error, never a silent success", () => { + it("zero bytes is the empty-input data error", async () => { + const source = (): AsyncGenerator => oneChunk(new Uint8Array()); + const r = await run(["parse", "-", "--ndjson", "--format", "fhir"], chunkDeps(source)); + expect(r.exit).toBe(EXIT.DATAERR); + expect(r.exit).not.toBe(EXIT.OK); + expect(r.stderr).toContain("CLI_EMPTY_INPUT"); + }); + + it("whitespace that frames no --ndjson record is a data error", async () => { + const lines: string[] = []; + async function* source(): AsyncGenerator { + await Promise.resolve(); + yield enc.encode("\n \n\n"); + } + const r = await run( + ["parse", "-", "--ndjson", "--format", "fhir"], + chunkDeps(source, (chunk) => lines.push(chunk)), + ); + expect(r.exit).toBe(EXIT.DATAERR); + expect(r.exit).not.toBe(EXIT.OK); + expect(r.stderr).toContain("CLI_PARSE_FAILED"); + expect(lines).toHaveLength(0); + }); + + it("an MLLP stream with no frame in it is a data error", async () => { + async function* source(): AsyncGenerator { + await Promise.resolve(); + yield new Uint8Array([0x1c, 0x0d]); + } + const r = await run(["parse", "-", "--format", "mllp"], chunkDeps(source)); + expect(r.exit).toBe(EXIT.DATAERR); + expect(r.exit).not.toBe(EXIT.OK); + expect(r.stderr).toContain("CLI_PARSE_FAILED"); + }); +}); From 71eb26e7bc7c8bfa19013856f9162510b1612cb7 Mon Sep 17 00:00:00 2001 From: Noah Schatz Date: Thu, 20 Aug 2026 14:31:43 +0000 Subject: [PATCH 2/2] S0054-cli-bulk-1: measure and read the same file descriptor, not the same path The size pre-check stat-ed the path and then read the path, which is a time-of-check to time-of-use race: the file measured and the file read need not be the same one. Code scanning flagged it as `js/file-system-race`, high severity, on the first push of this branch. The reader now opens the file once and does both operations on that handle, so the size that gates the read is the size of the bytes that get read. The value-free failure modes are unchanged: a missing path, a directory and an unreadable file are all still CLI_NO_INPUT / exit 66, and an oversized file is still refused before its bytes are read. --- src/core/io.ts | 50 +++++++++++++++++++++++++++----------------------- 1 file changed, 27 insertions(+), 23 deletions(-) diff --git a/src/core/io.ts b/src/core/io.ts index 543dfe8..a5f79b1 100644 --- a/src/core/io.ts +++ b/src/core/io.ts @@ -17,7 +17,7 @@ * @packageDocumentation */ -import { open, readFile, stat } from "node:fs/promises"; +import { open } from "node:fs/promises"; import type { Readable } from "node:stream"; import { CLI_CODES, CliError } from "./diagnostics.js"; @@ -54,13 +54,24 @@ export interface RunDeps { readonly writeStdout?: (chunk: string) => void; } +/** The value-free failure for a file that is missing, a directory, or otherwise unreadable. */ +function noInputError(path: string): CliError { + return new CliError( + CLI_CODES.CLI_NO_INPUT, + EXIT.NOINPUT, + `cannot read input file: ${path} (does it exist and is it readable?)`, + ); +} + /** * Read a file into bytes, mapping any read failure to a **value-free** `CLI_NO_INPUT` / * {@link EXIT.NOINPUT} error. The path is structural context (the user supplied it), so it may appear * in the message; the file *contents* never do. * * A file whose size already exceeds `limit` is refused **before it is read**, with the value-free - * over-limit data error, so an oversized file is never allocated in the first place. + * over-limit data error, so an oversized file is never allocated in the first place. The size is read + * from the **open descriptor** rather than from the path, so what is measured and what is read are the + * same file even if the path is replaced in between. * * @param path - The file path to read. * @param limit - The maximum number of bytes to accept. Defaults to {@link MAX_INPUT_BYTES}. @@ -78,22 +89,22 @@ export async function readFileBytes( path: string, limit: number = MAX_INPUT_BYTES, ): Promise { + let handle; + try { + handle = await open(path, "r"); + } catch { + throw noInputError(path); + } try { - const info = await stat(path); + const info = await handle.stat(); if (info.isFile() && info.size > limit) throw inputTooLargeError(limit); + return await handle.readFile(); } catch (e) { - // A CliError here is the over-limit refusal and is final; anything else means the path could not - // be stat-ed, which the read below reports as the value-free CLI_NO_INPUT. + // The over-limit refusal is final; anything else is an unreadable input, reported value-free. if (e instanceof CliError) throw e; - } - try { - return await readFile(path); - } catch { - throw new CliError( - CLI_CODES.CLI_NO_INPUT, - EXIT.NOINPUT, - `cannot read input file: ${path} (does it exist and is it readable?)`, - ); + throw noInputError(path); + } finally { + await handle.close(); } } @@ -172,25 +183,18 @@ export async function* streamChunks(stream: Readable): AsyncGenerator { - const noInput = (): CliError => - new CliError( - CLI_CODES.CLI_NO_INPUT, - EXIT.NOINPUT, - `cannot read input file: ${path} (does it exist and is it readable?)`, - ); - let handle; try { handle = await open(path, "r"); } catch { - throw noInput(); + throw noInputError(path); } try { const stream = handle.createReadStream(); try { for await (const chunk of stream as AsyncIterable) yield chunk; } catch { - throw noInput(); + throw noInputError(path); } } finally { await handle.close();