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..a5f79b1 100644 --- a/src/core/io.ts +++ b/src/core/io.ts @@ -3,28 +3,64 @@ * (`-`), 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 } 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; +} + +/** 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?)`, + ); } /** @@ -32,9 +68,16 @@ 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. 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}. * @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,23 +85,39 @@ 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 { + let handle; try { - return await readFile(path); + handle = await open(path, "r"); } 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); + } + try { + const info = await handle.stat(); + if (info.isFile() && info.size > limit) throw inputTooLargeError(limit); + return await handle.readFile(); + } catch (e) { + // The over-limit refusal is final; anything else is an unreadable input, reported value-free. + if (e instanceof CliError) throw e; + throw noInputError(path); + } finally { + await handle.close(); } } /** - * 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 +127,76 @@ 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 { + let handle; + try { + handle = await open(path, "r"); + } catch { + throw noInputError(path); + } + try { + const stream = handle.createReadStream(); + try { + for await (const chunk of stream as AsyncIterable) yield chunk; + } catch { + throw noInputError(path); + } + } 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"); + }); +});