proc

proc runs child processes from Deno the way a shell pipes them, with errors that reach one catch. It also gives any async iterable the Array methods you already know: map, filter, reduce, take, and concurrent versions of map.

import { run } from "@j50n/proc";

// cat app.log | grep ERROR | cut -c 21-
const errors = await run("cat", "app.log")
  .run("grep", "ERROR")
  .run("cut", "-c", "21-")
  .lines
  .collect();

console.log(errors);

[ "ERROR db: connection refused", "ERROR GET /orders/17 500" ]

Each .run() pipes the output of one command into the next. .lines turns bytes into lines of text, and collect() gathers them into an array. If any of the three commands fails, the await throws.

What it is for

• Scripts that drive other programs: git, compilers, ffmpeg, cloud CLIs, anything you would otherwise chain in bash, with TypeScript’s types and real error handling.
• Streaming data: logs, CSV exports, and compressed files, a line or a row at a time, in constant memory.
• Long-running children: programs that launch other programs and must give them time to shut down, including in containers.
• Async iterables in general: the same methods work on anything you can for await over, with or without processes.

What you get

• run() to start a command, and .run() to pipe into the next one.
• enumerate() to wrap any iterable in the same methods.
• read(), writeTo(), and toStdout() for files and standard streams.
• main() to let child processes finish cleaning up before your program exits.
• WritableIterable to turn callbacks and events into an iterable.
• Data transforms between CSV, TSV, JSON lines, and a record format.

When you don’t need it

If you run one command and want all of its output at once, await new Deno.Command("git", { args: ["status"] }).output() does that, and you check success yourself. proc earns its place when you pipe commands together, read output as it arrives, or want a failed command to throw.

Next

Install it and run a first script, then read Key ideas: eight short points that cover how every part of proc behaves.

Install and run a first script

Add proc to your project:

deno add jsr:@j50n/proc

This records the package in deno.json, and from then on you import it by name. The core library is @j50n/proc; the data transforms (CSV, TSV, JSON lines) are a separate entry point, @j50n/proc/transforms, so you load them only if you use them.

import { enumerate, read, run } from "@j50n/proc";
import { fromCsvToRows, toTsv } from "@j50n/proc/transforms";

For a one-off script without a deno.json, import from JSR directly: import { run } from "jsr:@j50n/proc@0.29.0";.

A first script

Save this as errors.ts, next to a log file called app.log:

import { run } from "@j50n/proc";

const errors = await run("grep", "ERROR", "app.log").lines.collect();

console.log(`${errors.length} errors`);
for (const line of errors) {
  console.log(line);
}

Run it, allowing it to start grep:

deno run --allow-run=grep errors.ts

2 errors
2026-10-05 09:01:13 ERROR db: connection refused
2026-10-05 09:02:45 ERROR GET /orders/17 500

If app.log has no errors, grep exits with code 1, and the script stops with an ExitCodeError. That is proc treating a failed command as an error, the way set -e does in bash. Errors shows how to handle it.

Permissions

Deno asks before a script runs a program or touches a file. Grant only what the script needs:

A child inherits your environment. Passing env to run() adds or overrides variables for the child, and doesn’t need --allow-env.

Key ideas

Eight points cover how every part of proc behaves. The rest of the book is detail.

1. A pipeline is a source, some steps, and a consumer

import { read } from "@j50n/proc";

const errors = await read("app.log") // a source: the file's bytes
  .lines // a step: bytes to lines of text
  .filter((line) => line.includes("ERROR")) // another step
  .count(); // the consumer: pulls everything through

console.log(errors);

2

A source produces items: run() (a command’s output), read() (a file), or enumerate() (any iterable or async iterable). Each step (lines, map, filter, take, transform, …) returns a new Enumerable and does nothing yet; .run() is the exception, since it starts its command at once (point 2). The consumer (collect, forEach, count, reduce, first, writeTo, toStdout, or a for await loop) pulls items through one at a time and returns a promise. Await it.

2. A command starts when you call run(), and you must read its output

run() starts the child process at once; only the steps after it wait. The child’s stdout is a pipe with a small buffer (typically 64 KB). A child that writes more than that waits until someone reads, so if your code never consumes the output, the child never finishes and neither does your program. Always end a command’s pipeline with a consumer. If you don’t want the output, consume it anyway: await run("make").lines.forEach(() => {}).

stderr is not piped by default: it goes straight to your terminal.

3. Errors come out of the consumer’s await

import { ExitCodeError, run } from "@j50n/proc";

try {
  await run("sh", "-c", "echo one; echo two; exit 3")
    .lines
    .forEach((line) => console.log(line));
} catch (error) {
  if (error instanceof ExitCodeError) {
    console.log(`failed with exit code ${error.code}`);
  } else {
    throw error;
  }
}

one
two
failed with exit code 3

A command that exits with a non-zero code throws ExitCodeError once you have read all of its output, so the lines it wrote before failing still reach you. The same catch also gets an error from any command earlier in a pipeline (as an UpstreamError whose cause is the original), and anything your own callbacks throw. One try around the consumer covers the whole pipeline. See Errors.

4. Stopping early is fine

import { run } from "@j50n/proc";

// `yes` prints "y" forever; taking three stops it.
const ys = await run("yes").lines.take(3).collect();
console.log(ys);

[ "y", "y", "y" ]

When a consumer stops before the end (take, first, find, a break out of for await), proc closes the pipeline behind it, and the await returns at once, without an error. A command that is still writing dies at its next write; one that runs on quietly, like a server that printed “ready”, keeps running until it exits, or until main() stops it on the way out. Deno doesn’t exit while a child is running, so without main() the script ends only when that child does.

5. An Enumerable is used once

import { enumerate } from "@j50n/proc";

const doubled = enumerate([1, 2, 3]).map((n) => n * 2);

console.log(await doubled.collect());
console.log(await doubled.collect()); // already used up: nothing left

const saved = await enumerate([1, 2, 3]).map((n) => n * 2).collect();
console.log(await enumerate(saved).count(), await enumerate(saved).count());

[ 2, 4, 6 ]
[]
3 3

Consuming an Enumerable uses it up, and a second pass finds nothing, without an error. To go over data twice, collect it into an array first, or split the stream with tee().

6. Some members are properties

.lines, .chunkedLines, .first, .status, and .pid take no parentheses. .first and .status are promises: await p.status, not p.status(). Everything else is a method.

7. Parsers yield batches

import { read } from "@j50n/proc";
import { fromCsvToRows } from "@j50n/proc/transforms";

const rows = await read("people.csv")
  .transform(fromCsvToRows()) // yields batches: arrays of rows
  .flatten() // one row at a time
  .collect();

console.log(rows);

[ [ "name", "city" ], [ "Ada", "London" ], [ "Grace", "Arlington" ] ]

For speed, the parsers in @j50n/proc/transforms yield arrays of rows rather than one row at a time. Add .flatten() before a step that works on one row. The row writers (toCsv(), toTsv(), toRecord()) take either; toJson() takes one value per item.

8. Wrap a long-running program in main()

import { main, run } from "@j50n/proc";

await main(async () => {
  await run("./long-job.sh").lines.toStdout();
});

When Deno is told to stop (Ctrl-C, or a container’s SIGTERM), it exits at once without waiting for its children; in a container they are killed with it before they can clean up. When an error goes uncaught, Deno kills its children as it exits, on any host. main() runs your program, and however it ends (it returns, it throws, or a signal arrives), it makes sure every child has been asked to stop, and waits for them, up to 30 seconds, before exiting. See Shutting down cleanly.

Running a command

import { run } from "@j50n/proc";

const lines = await run("head", "-n", "3", "app.log").lines.collect();

console.log(lines);

[
  "2026-10-05 09:00:01 INFO  server started on :8080",
  "2026-10-05 09:00:07 INFO  GET /health 200",
  "2026-10-05 09:01:13 ERROR db: connection refused"
]

run() takes the program and its arguments as separate strings. No shell is involved, so nothing needs quoting, and run("head -n 3 app.log") looks for a program with that whole name. The program is looked up on PATH, or given as a path or a file URL. A relative path such as ./build.sh is found from the child’s working directory, so with the cwd option below it is looked for in cwd, not where your program runs.

The child starts at the call. run() returns a ProcessEnumerable: the child’s stdout as an async iterable of bytes, with all the Enumerable methods. Steps such as .lines wait until a consumer pulls; the process doesn’t. If the command fails, the consumer’s await throws (see Errors).

Reading the output

import { concat, run } from "@j50n/proc";

// Every line, in an array.
const all = await run("cat", "fruit.txt").lines.collect();
console.log(all);

// One line at a time, as the command writes them.
for await (const line of run("cat", "fruit.txt").lines) {
  console.log(`- ${line}`);
}

// Only the first line.
const first = await run("cat", "fruit.txt").lines.first;
console.log(first);

// Raw bytes, in chunks as they arrive; `concat` joins them.
const chunks = await run("cat", "app.log.gz").collect();
console.log(`${concat(chunks).length} bytes`);

// Straight to this program's stdout, unchanged.
await run("cat", "fruit.txt").toStdout();

[ "cherry", "apple", "banana" ]
- cherry
- apple
- banana
cherry
176 bytes
cherry
apple
banana

• .lines decodes UTF-8 and splits on "\n" (a "\r" before it goes too). It is a property, not a method.
• .collect() gathers everything into an array. For large output, use for await or .forEach() instead, which handle one line at a time and keep nothing.
• .first is a promise of the first line. It stops reading there, which ends the command early (see Pipelines), and throws RangeError if there is no output at all.
• Without .lines the items are Uint8Array chunks of whatever size the pipe delivered, not lines. concat() joins them.
• .toStdout() copies the output to your program’s stdout. Use it when you only want the user to see it.

The output can be read once (Key ideas): a second pass finds nothing, and doesn’t throw a failed command’s error again.

Working directory and environment

Options go before the command:

import { run } from "@j50n/proc";

await Deno.mkdir("reports", { recursive: true });
await Deno.writeTextFile("reports/q3.txt", "");

const line = await run(
  { cwd: "reports", env: { REGION: "west" } },
  "sh",
  "-c",
  'echo "$REGION: $(ls)"',
).lines.first;

console.log(line);

west: q3.txt

cwd sets the child’s working directory. env adds variables to the environment the child inherits, or overrides them; it can’t remove one. A PATH in env also changes where the program is looked up, so run({ env }, "ls") runs whatever ls comes first on that PATH. Don’t build env from untrusted input. clearEnv: true starts the child with only env, to keep secrets in your environment from reaching it. proc still finds the program on your PATH, but the child gets none unless env has one, so put a PATH in env if it starts other programs by name.

timeoutMs stops a child that runs too long: proc sends it SIGTERM, and reading its output throws a TimeoutError (Errors). The other options, fnStderr and fnError, are about errors and are covered in Errors. All of them are listed under ProcessOptions.

Status and PID

import { run } from "@j50n/proc";

// No output to read, so waiting on the status alone is safe. It doesn't
// throw for a failed exit; it reports it.
const { success, code } = await run("test", "-e", "missing.txt").status;
console.log(success, code);

// With output: read it first, then look at the status.
const p = run("sh", "-c", "echo hello");
console.log(p.pid > 0);
console.log(await p.lines.collect());
console.log((await p.status).success);

false 1
true
[ "hello" ]
true

.status resolves when the child exits, with success, code, and signal. It never throws for a failed exit; it tells you about it. .pid is the child’s process ID, available as soon as run() returns.

Waiting for a command you don’t want output from

.status resolves only when the child exits, and a child that writes more than its stdout pipe holds (typically 64 KB) can’t exit until someone reads. This fragment hangs forever, because nothing reads seq’s output:

await run("seq", "1", "100000").status; // hangs: nobody reads the output

So read the output even when you don’t want it:

import { run } from "@j50n/proc";

// `seq` writes about 600 KB, far more than the pipe holds. Reading the output
// and throwing it away lets it finish, and a failure would still throw.
await run("seq", "1", "100000").forEach(() => {});

console.log("done");

done

Calling .forEach() on the raw bytes skips decoding them, and a failed exit still throws. Use .status alone only for a command that prints little or nothing, such as test or grep -q, and when a failed exit is an answer rather than an error.

stderr goes to your terminal

import { run } from "@j50n/proc";

const out = await run("sh", "-c", "echo to stdout; echo to stderr >&2")
  .lines
  .collect();

console.log(out);

[ "to stdout" ]

Only stdout is captured. The child’s stderr is connected straight to your program’s, so to stderr appeared on the terminal and isn’t in the result. To capture it, pass fnStderr; see Capturing stderr.

The child’s stdin is closed: a program that reads stdin sees end of input at once. To feed it data, pipe into it with .run(), as Pipelines and input shows.

A program that isn’t there

If the program doesn’t exist, run() itself throws Deno.errors.NotFound, before any consumer runs. Keep the run() call inside the same try as the await, and it is caught with everything else; see Errors.

Permissions

Running a command needs --allow-run, or --allow-run=head,sort to allow only those programs. Passing env doesn’t need --allow-env. See Install.

Pipelines and input

import { run } from "@j50n/proc";

// cut -d " " -f 3 app.log | sort | uniq -c
const counts = await run("cut", "-d", " ", "-f", "3", "app.log")
  .run("sort")
  .run("uniq", "-c")
  .lines
  .map((line) => line.trim())
  .collect();

console.log(counts);

[ "2 ERROR", "4 INFO", "1 WARN" ]

.run() on a command’s output starts another command with that output as its stdin, as | does in a shell. Every command in the chain runs at the same time, and the bytes pass from one to the next in chunks; nothing is collected in between unless you collect it. If any command fails, the consumer at the end throws; see Errors.

Feeding a command from an array

import { enumerate } from "@j50n/proc";

const sorted = await enumerate(["pear", "apple", "fig"])
  .run("sort")
  .lines
  .collect();
console.log(sorted);

// Only text and bytes can be written; turn anything else into strings first.
const numbers = await enumerate([10, 9, 100])
  .map((n) => `${n}`)
  .run("sort", "-n")
  .lines
  .collect();
console.log(numbers);

[ "apple", "fig", "pear" ]
[ "9", "10", "100" ]

.run() works on any Enumerable, not only on a command’s output. enumerate() wraps an array, a Set, a generator, or any other iterable or async iterable, and .run() writes its items to the command’s stdin.

Feeding a command from a file

import { read } from "@j50n/proc";

// grep -c ERROR < app.log
const count = await read("app.log").run("grep", "-c", "ERROR").lines.first;

console.log(count);

2

read() yields a file’s bytes, and .run() sends them to the command unchanged. It is the same as grep -c ERROR < app.log, without an extra cat process. The file is opened when reading starts, which is at the .run() call. If it is missing, the command sees empty input, and the consumer throws with Deno.errors.NotFound as the cause: an UpstreamError, or the command’s own ExitCodeError if empty input makes it fail, as it does grep -c (which prints 0 and exits 1).

How items become stdin

import { concat, enumerate } from "@j50n/proc";

const decoder = new TextDecoder();
const encoder = new TextEncoder();

// `cat` echoes its stdin, so this shows exactly what the command received.
async function received(
  items: Iterable<string | string[] | Uint8Array | Uint8Array[]>,
) {
  const bytes = concat(await enumerate(items).run("cat").collect());
  return JSON.stringify(decoder.decode(bytes));
}

console.log(await received(["one", "two"])); // each string is a line
console.log(await received([["one", "two"], ["three"]])); // so is each element
console.log(await received([encoder.encode("raw"), encoder.encode("bytes")]));

"one\ntwo\n"
"one\ntwo\nthree\n"
"rawbytes"

So lines from .lines go back in as lines, and bytes from a command or a file go through untouched. For any other item type (numbers, objects), the return type of .run() is never and the code doesn’t type-check: turn the items into strings with .map() first, as the array example does with numbers.

Lots of lines

Each item written to a command is a write of its own, and each step pays for every item it handles. With a few thousand lines that doesn’t matter; with millions it does. Work a chunk of lines at a time instead:

import { run } from "@j50n/proc";

// Lines arrive in arrays, one per chunk read. A step works on the whole
// array, and .run() writes each array to the next command in one go.
const count = await run("seq", "1", "1000000")
  .chunkedLines
  .map((lines) => lines.filter((line) => !line.endsWith("7")))
  .run("wc", "-l")
  .lines
  .first;

console.log(count);

900000

.chunkedLines yields the same lines as .lines, in arrays, and .run() writes each array in one go. On two million short lines, a filter between two commands took 16 seconds a line at a time and 0.3 seconds a chunk at a time.

run({ buffer: true }, ...) is a smaller fix for a source you can’t chunk: it collects small items into writes of at least 16 KiB, about twice as fast. The child sees nothing until 16 KiB has collected or the input ends, so leave it off for a child that must answer each line as it arrives.

Mixing commands and steps

import { run } from "@j50n/proc";

const messages = await run("cat", "app.log")
  .lines
  .filter((line) => line.includes(" ERROR "))
  .map((line) => line.slice(26)) // drop the timestamp and level
  .run("sort", "-f") // -f: ignore case
  .lines
  .collect();

console.log(messages);

[ "db: connection refused", "GET /orders/17 500" ]

Any step can sit between two commands: .lines turns the bytes into text, .filter() and .map() work on each line, and the next .run() writes the strings back out as lines. Use a command where a tool already does the job well (sort on data larger than memory, grep on a big file), and a TypeScript step where the logic is easier to write and test in code.

Stopping early

import { run } from "@j50n/proc";

// The consumer stops: take(2) closes the pipeline behind it.
const sevens = await run("seq", "1", "1000000")
  .run("grep", "7")
  .lines
  .take(2)
  .collect();
console.log(sevens);

// A command stops: head exits after two lines, and `yes` stops with it.
const ys = await run("yes").run("head", "-n", "2").lines.collect();
console.log(ys);

[ "7", "17" ]
[ "y", "y" ]

There are two ways a pipeline stops before its first command runs out, and neither is an error:

• The consumer stops (take, first, find, a break out of for await). proc closes the pipeline behind it, and the await returns at once. seq and grep die of SIGPIPE at their next write. Their exit codes are not checked.
• A command stops reading, like head. Writing to it ends quietly, and the command before it is closed the same way.

A command that goes quiet instead keeps running after the await has returned, until it writes again, exits, or is stopped. That is what lets you start a server and wait for its “ready” line. Deno doesn’t exit while a child is still running, though, so without main() the script ends only when the server does; under main(), the server is stopped when the program ends.

When a pipeline is cut short, the commands before the cut may or may not have finished, so don’t rely on their failures being reported. A command whose output proc read to the end has its exit code checked; one that was closed early doesn’t. With head in the middle, which of the two happens depends on timing: run("sh", "-c", "echo a; echo b; exit 3").run("head", "-n", "1") throws an UpstreamError for the exit 3 on some runs, and returns ["a"] without one on others.

Pipelines run for their side effects

import { read, run } from "@j50n/proc";

// Into a file: writeTo is the consumer.
await read("app.log").run("grep", "WARN").writeTo("warnings.log");
console.log((await Deno.readTextFile("warnings.log")).trimEnd());

// A command run only for what it does: consume its (empty) output, so a
// failure throws.
await run("cp", "warnings.log", "warnings.bak").forEach(() => {});
console.log((await Deno.stat("warnings.bak")).isFile);

2026-10-05 09:01:14 WARN  retrying db connection
true

A pipeline that writes a file ends in .writeTo(path), which is its consumer; it takes bytes, which is what a command produces. A command you run only for what it does (cp, mkdir, git commit) still needs a consumer, and .forEach(() => {}) is the one to use: it reads and discards any output, and throws if the command fails. .toStdout() is the other common ending, when the output is for the user to see.

.run() starts at once

Like run(), .run() starts its command when you call it, and it begins reading the items before it into the command’s stdin straight away, whether or not anything reads the output yet. Callbacks in the steps before it (a .map() that logs, say) run then too. The steps after it are lazy, as usual. Build a pipeline only when you are about to consume it, and always consume it: an unread pipeline can hang (see Waiting for a command).

To write a child’s stdin a piece at a time, as your program produces the data, use Process and its stdin, or feed .run() from a WritableIterable.

Errors

import { ExitCodeError, run } from "@j50n/proc";

try {
  await run("sh", "-c", "echo partial; exit 3").lines.forEach(console.log);
} catch (error) {
  if (error instanceof ExitCodeError) {
    console.log(error.message);
    console.log(error.command, error.code);
  } else {
    throw error;
  }
}

partial
sh exited with code 3
[ "sh", "-c", "echo partial; exit 3" ] 3

A command that exits with a non-zero code throws ExitCodeError from the consumer’s await, after every line it wrote has been delivered: partial printed first. Its message names the program and the code; command holds the whole command line (left out of the message, since arguments can hold secrets), and code the exit code. One try around the consumer catches the errors of every command and every callback in the pipeline.

What you see, and what to do

All four process errors extend ProcessError, so error instanceof ProcessError catches any of them. Narrow with instanceof before using code, signal, or command.

Killed by a signal

import { run, SignalError } from "@j50n/proc";

try {
  // The shell kills itself, as the OOM killer or `kill -9` might.
  await run("sh", "-c", "echo started; kill -KILL $$").lines.forEach(
    console.log,
  );
} catch (error) {
  if (error instanceof SignalError) {
    console.log(`${error.command[0]} was killed by ${error.signal}`);
  } else {
    throw error;
  }
}

started
sh was killed by SIGKILL

A command killed by a signal throws SignalError, with the signal’s name in signal. SIGKILL usually means the out-of-memory killer or a kill -9; SIGTERM means something asked it to stop. A command you stopped early yourself (with take, first, or break) dies of SIGPIPE, but that throws nothing.

Timed out

import { run, TimeoutError } from "@j50n/proc";

try {
  await run({ timeoutMs: 200 }, "sleep", "5").lines.collect();
} catch (error) {
  if (error instanceof TimeoutError) console.log(error.message);
  else throw error;
}

sleep timed out after 200 ms

With timeoutMs, proc sends the child SIGTERM when the time runs out, and its output throws TimeoutError once the child has exited, even if it caught the signal and exited cleanly: a run that was cut short didn’t succeed. The timer runs from the start until the child exits, so it also stops a child whose output you stopped reading early. A child that ignores SIGTERM keeps running; proc never sends SIGKILL. And reading waits for the output to end, which a child’s own children can hold open, as a wrapper script’s do (see Wrapper scripts must exec).

Which error a pipeline throws

import { ExitCodeError, run, UpstreamError } from "@j50n/proc";

/** One line per error in the cause chain. */
function chain(error: unknown): string[] {
  const lines: string[] = [];
  for (let e = error; e instanceof Error; e = e.cause) {
    if (e instanceof ExitCodeError) {
      lines.push(`${e.name}: ${e.command[0]} exited with ${e.code}`);
    } else if (e instanceof UpstreamError) {
      lines.push(`${e.name}: thrown by ${e.command[0]}`);
    } else {
      lines.push(`${e.name}: ${e.message}`);
    }
  }
  return lines;
}

// The first command fails; the last one succeeds.
try {
  await run("sh", "-c", "echo data; exit 2").run("cat").lines.collect();
} catch (error) {
  console.log(chain(error));
}

// Both fail: grep finds nothing in the empty input and exits 1.
try {
  await run("sh", "-c", "exit 2").run("grep", "data").lines.collect();
} catch (error) {
  console.log(chain(error));
}

[ "UpstreamError: thrown by cat", "ExitCodeError: sh exited with 2" ]
[
  "ExitCodeError: grep exited with 1",
  "ExitCodeError: sh exited with 2"
]

The consumer sees only the last command, so an earlier failure reaches it through the chain, as the error’s cause:

• An earlier command fails and the last one succeeds: the last command throws UpstreamError, and cause is the earlier command’s error. UpstreamError.command is the command that threw, not the one that failed; the message is copied from the cause.
• Both fail: the last command throws its own ExitCodeError (or SignalError), still with the earlier error as cause. This is common, since a command that gets cut-off or empty input often fails too: in the second case, grep exits 1 because it saw nothing to match.

In a longer pipeline the chain is longer, one link per command, in order from the last command back to the first failure. Walk cause to find where it started, as chain() does above.

Putting stderr into the error

import { ExitCodeError, run } from "@j50n/proc";

try {
  await run(
    {
      fnStderr: (stderr) => stderr.lines.collect(),
      fnError: (error, stderrLines) => {
        if (error instanceof ExitCodeError) {
          const detail = stderrLines?.join("; ") ?? "";
          throw new Error(`backup failed (exit ${error.code}): ${detail}`, {
            cause: error,
          });
        }
        if (error) throw error;
      },
    },
    "sh",
    "-c",
    "echo 'writing archive' >&2; echo 'disk full' >&2; exit 4",
  ).lines.collect();
} catch (error) {
  if (error instanceof Error) console.log(error.message);
}

backup failed (exit 4): writing archive; disk full

Two options, passed before the command, work together:

• fnStderr receives the child’s stderr as bytes (use .lines for text) and returns whatever you want to keep. With it, stderr no longer goes to the terminal. Read it to the end: a child that fills its stderr pipe (about 64 KB) blocks, and the program hangs.
• fnError decides what the process throws. It is called once, after the output has ended and the child has exited, with the error the process would throw (or undefined) and what fnStderr returned. Whatever it throws is what the consumer’s await throws; set cause when you wrap the original. If it returns normally, nothing is thrown. It may be async.

Don’t throw from fnStderr to fail the process: that replaces the process’s own error, and the exit code is lost. Return the data, and throw from fnError. ErrorHandler and StderrHandler have the exact rules.

Accepting expected exit codes

import { ExitCodeError, run } from "@j50n/proc";

/** grep, where exit code 1 ("no lines matched") is not an error. */
function grep(pattern: string, file: string): Promise<string[]> {
  return run(
    {
      fnError: (error) => {
        if (error instanceof ExitCodeError && error.code === 1) return;
        if (error) throw error;
      },
    },
    "grep",
    pattern,
    file,
  ).lines.collect();
}

console.log(await grep("WARN", "app.log"));
console.log(await grep("FATAL", "app.log"));

try {
  await grep("WARN", "missing.log"); // exit 2: a real failure
} catch (error) {
  if (error instanceof ExitCodeError) console.log(`exit ${error.code}`);
}

// When only the answer matters, `grep -q` prints nothing; ask the status.
const { success } = await run("grep", "-q", "ERROR", "app.log").status;
console.log(success);

[ "2026-10-05 09:01:14 WARN  retrying db connection" ]
[]
exit 2
true

grep exits 1 when no line matches and 2 when something went wrong, such as a missing file. An fnError that returns for code 1 and rethrows everything else turns “no match” into an empty result and keeps real failures as errors. (The missing file’s message went to the terminal, from grep’s stderr.) Other commands with a meaningful code 1: diff (files differ), cmp, test.

If you only need the answer, not the output, ask the status instead: grep -q prints nothing, so .status can be awaited without reading anything, and it reports a failed exit rather than throwing it.

Catching the error afterward works too, but the consumer has thrown by then, so any output it collected is lost. In a pipeline, put fnError on the command that has the expected code (.run({ fnError }, "grep", ...)); the commands after it then see no failure.

Errors from your own callbacks

import { enumerate, UpstreamError } from "@j50n/proc";

function parse(line: string): number {
  const n = Number(line);
  if (Number.isNaN(n)) throw new Error(`not a number: ${line}`);
  return n;
}

// The callback's error arrives as it was thrown.
try {
  await enumerate(["1", "two", "3"]).map(parse).collect();
} catch (error) {
  if (error instanceof Error) console.log(`${error.name}: ${error.message}`);
}

// With a .run() after the callback, it is the cause of an UpstreamError.
try {
  await enumerate(["1", "two", "3"])
    .map(parse)
    .map((n) => `${n}`)
    .run("cat")
    .lines
    .collect();
} catch (error) {
  if (error instanceof UpstreamError && error.cause instanceof Error) {
    console.log(
      `${error.name} from ${error.command[0]}: ${error.cause.message}`,
    );
  }
}

Error: not a number: two
UpstreamError from cat: not a number: two

An error thrown in a callback (map, filter, forEach, …) stops the pipeline and arrives at the consumer as it was thrown. If a .run() comes after the callback, the error stops that command’s input, and the command throws UpstreamError with your error as its cause, as for a failed command upstream. The same try catches both.

A missing program or file

import { read, run } from "@j50n/proc";

try {
  await run("no-such-program", "--version").lines.collect();
} catch (error) {
  if (error instanceof Deno.errors.NotFound) console.log(error.message);
}

try {
  await read("no-such-file.txt").lines.collect();
} catch (error) {
  if (error instanceof Deno.errors.NotFound) console.log(error.message);
}

Failed to spawn 'no-such-program': entity not found
No such file or directory (os error 2): open 'no-such-file.txt'

Both throw Deno.errors.NotFound, at different times. run() throws it itself, at the call, before any consumer runs; keep the call inside the try. A cwd that doesn’t exist throws it from run() too, with No such cwd in the message. read() throws it from the consumer, when reading starts. Fed into a command with .run(), a missing file becomes the cause of that command’s error (Feeding a command from a file).

Stopping early skips the check

import { run } from "@j50n/proc";

// The command fails, but the consumer stopped before the end, so the exit
// code is never checked and nothing throws.
const first = await run("sh", "-c", "echo one; echo two; exit 1").lines.first;

console.log(first);

one

A consumer that stops before the end (first, take, find, some, a break) closes the command’s output, and its exit code is never checked. If the failure matters, read to the end. The same goes for a command cut short by one after it, such as head; see Stopping early.

Shutting down cleanly

import { main, run } from "@j50n/proc";

await main(async () => {
  await run("./processes-service.sh").lines.forEach(console.log);
});

service: ready
service: finished

main() runs your program, and however it ends, it signals every child process proc started that is still running, waits for them to exit and for their fnStderr handlers to finish, and only then exits. Wrap any program that runs long-lived children in it, and always one that runs in a container.

The service here is a stand-in shell script: it holds a lock file while it runs and removes it when it gets SIGTERM. Left alone, it finishes by itself, and the program behaves as it would without main.

Why it’s needed

When Deno gets SIGTERM or SIGINT and nothing handles it, it exits at once, and its children are never told. On a normal host they carry on alone, orphaned. In a container, Deno is usually the main process, so its exit ends the container, and the runtime kills every child still running before its cleanup code can run. A child holding a lock, a temp directory, or a cloud resource never gets the chance to release it.

An uncaught error is worse, on any host: Deno kills its children itself as it exits, and their SIGTERM handlers never run.

This program starts the same service twice, once under main and once without, and sends SIGTERM to each as soon as the service is ready:

shutdown-main.ts
  service: ready
  exited with 143
  lock file left behind: false
shutdown-plain.ts
  service: ready
  killed by SIGTERM
  lock file left behind: true

Under main, the service got SIGTERM, removed its lock, and the program exited with 143, as a program killed by SIGTERM reports itself. Without main, Deno died at once; the service never heard anything, and was still running with its lock held when Deno was gone. (The program that ran the two then stopped it.)

What main does

main needs no permissions of its own: listening for signals and signalling the children it started are both allowed with the --allow-run that starting them needed.

• On return, any child still running (one you started and never awaited) gets SIGTERM, and main waits for it as it would on a signal.
• A child is done when it has exited and its fnStderr, if it has one, has finished. A handler that gathers stderr and writes it out after the exit gets to finish writing, within the same timeoutMs.
• On an error, main prints it to stderr before it signals the children. That includes an unhandled promise rejection or an error thrown in a timer.
• On SIGTERM, main passes it on and waits. A SIGTERM comes to Deno alone, from docker stop, Kubernetes, systemd, or kill, so the children hear of it only through main.
• After a signal, once the children have exited, main lets the program finish with what they printed on the way out: it waits for their output to be read to the end, then up to half a second for the program to return, so a pipeline writing it to a file completes. A child that dies of the signal makes its consumer throw SignalError; main doesn’t print it, since the signal decides the exit code.
• On SIGINT or SIGHUP, main doesn’t pass it on. These usually come from the terminal (Ctrl-C, or the terminal closing), which sends them to the whole foreground process group, children included. Forwarding would make it their second, and many programs take a second Ctrl-C to mean “quit now, skip the cleanup”. So main gives the children a second to act on the one they got, then sends SIGTERM to any still running and waits. That also covers a SIGINT sent to Deno alone, by Docker’s STOPSIGNAL SIGINT, systemd’s KillSignal=SIGINT, an IDE’s stop button, or kill -INT <pid>: the children got nothing, so after the second they get SIGTERM.
• A second signal exits at once, without waiting, so a person pressing Ctrl-C twice always gets out; children still running get SIGTERM on the way out.
• The first ending wins. If the program fails and the container’s SIGTERM arrives while the children are still cleaning up, main keeps waiting for them and still exits 1. One exception: a Ctrl-C usually reaches a child first, and a child dying of it fails the program before the signal reaches main. When a child ends that way, main gives the signal half a second to arrive, so the exit is still 130 and the error isn’t printed.
• Children started during the wait, by a program that carries on, get SIGTERM as Deno exits but aren’t waited for.

Call main once, around the whole program: however it ends, it ends the process.

Deno doesn’t honor nohup: under it, a hangup still reaches Deno, and main handles it as above.

On Windows

Windows has no signals a child can catch. Deno’s kill ends a child at once, with exit code 1 (TerminateProcess), whatever signal it is given. So on Windows main and terminateAll can’t ask a child to clean up; they can only end it. What still holds:

• Ctrl-C reaches every process on the console, the children included, and each can clean up after it. It is the only signal main handles there. It then waits for the children, up to timeoutMs, without ending them, so their cleanup isn’t cut short, and exits 130.
• On return or an error, children still running are ended at once, not asked; Deno would end them as it exits anyway.
• timeoutMs on a command ends the child at once.

A child that must clean up on Windows has to be told to stop some other way, such as by closing its stdin or with a message of its own, before the program returns.

How long it waits

main waits up to timeoutMs, 30 seconds by default. Children still running then are left running, not killed, and Deno exits; in a container they die with it. In a container, the real deadline is the runtime’s: it sends SIGTERM, waits a grace period, then sends SIGKILL to everything.

These come from the docker stop reference, the Kubernetes Pod API, and the ECS task definition parameters. In Kubernetes, a preStop hook’s time counts against the same grace period.

Set timeoutMs a little under the grace period, leaving a second or two for Deno itself to exit:

import { main, run } from "@j50n/proc";

// Under Kubernetes' default 30-second grace period: give the children 25.
await main(async () => {
  await run("./long-job.sh").lines.forEach(console.log);
}, { timeoutMs: 25_000 });

Too high, and the SIGKILL arrives first: the children are cut off anyway, and your program never gets to exit with its own code. Too low, and main gives up while the children could still have finished. Under Docker’s default 10 seconds, the default of 30 is too high; set about 8_000, or give the container a longer stop timeout.

Several children at once

A supervisor runs several long-lived children and waits on all of them:

import { main, run } from "@j50n/proc";

// Long-running workers, each output line tagged with the worker's name.
// (These stand-ins end after three ticks; real ones would run until stopped.)
const workers: Record<string, string> = {
  web: "for i in 1 2 3; do echo tick $i; sleep 0.05; done",
  queue: "for i in 1 2 3; do echo tick $i; sleep 0.05; done",
};

await main(async () => {
  await Promise.all(
    Object.entries(workers).map(([name, script]) =>
      run("sh", "-c", script).lines
        .forEach((line) => console.log(`[${name}] ${line}`))
    ),
  );
});

Promise.all rejects as soon as one worker fails, so main prints that error, sends SIGTERM to the others, waits for them, and exits 1. On SIGTERM or Ctrl-C, every worker is stopped as described above. Each worker’s stderr goes to yours untagged; to tag it as well, read it with fnStderr.

Stopping the children without exiting

import { run, terminateAll } from "@j50n/proc";

const service = run("./processes-service.sh");

for await (const line of service.lines) {
  console.log(line);
  if (line === "service: ready") {
    await terminateAll({ timeoutMs: 5_000 }); // SIGTERM, then wait for it
    console.log("terminateAll returned");
  }
}

const left = await Deno.stat("service.lock").then(() => true, () => false);
console.log(`lock file left behind: ${left}`);

service: ready
terminateAll returned
lock file left behind: false

terminateAll() is the part of main that signals and waits, on its own: it sends SIGTERM (or the signal you pass) to every running child proc started, all at once, and resolves when they have exited or timeoutMs (default 30 s) has passed. Your program carries on. It signals every child, not just one; to stop a single command, stop reading it (see Stopping early) or Deno.kill(p.pid, "SIGTERM").

A child that dies of the signal, rather than catching it and exiting cleanly, makes its consumer throw SignalError; the service above traps SIGTERM and exits 0, so nothing is thrown.

Wrapper scripts must exec

proc signals only the processes it started: children of run(), .run(), and new Process, not ones started with Deno.Command directly, and not their children in turn. If a child is a shell script that starts the real program, the signal reaches the shell, and the program under it never hears about the shutdown. Make the script replace itself with the program:

#!/bin/sh
export APP_ENV=production
exec java -jar app.jar "$@"

Without exec, SIGTERM ends the shell, and java carries on, orphaned and unaware. proc sees its child exit and stops waiting, so the cleanup never happens. Some launchers are such wrappers themselves, so check yours. If a script can’t exec, it has to trap the signal and pass it on (trap 'kill -TERM "$pid"' TERM, then wait).

The same applies to Deno itself in a container. Use the exec form, CMD ["deno", "run", "--allow-run", "main.ts"]. The shell form of CMD or ENTRYPOINT (CMD deno run main.ts) runs Deno under /bin/sh -c, which, as the Dockerfile reference warns, does not pass signals on, so main never sees the SIGTERM.

What main can’t do

It can’t wait for output your program is still reading. A pipeline writing a command’s stdout to a file, or a tee branch nobody awaits, is on its own once the program has returned or thrown: main waits for the children and their stderr handlers, not for your code. Await your own pipelines before the program returns.

Nothing runs if Deno is killed outright: by SIGKILL, the out-of-memory killer, or a failed node. A call to Deno.exit() elsewhere in your code also exits at once; proc sends SIGTERM to the children on the way out, but can’t wait for them. Anything that must be released needs a backstop of its own on the resource’s side, such as a lease or an idle timeout.

Enumerable and its methods

An Enumerable is an async sequence with the methods you know from arrays. run(), read(), and range() return one, and enumerate() wraps anything else you can iterate: an array, a Set, a generator, a ReadableStream.

import { read } from "@j50n/proc";

// How many lines at each level, leaving out INFO.
const levels = await read("app.log")
  .lines
  .map((line) => line.split(/\s+/)[2]) // "INFO", "ERROR", "WARN"
  .filterNot((level) => level === "INFO")
  .reduce(
    (counts, level) => counts.set(level, (counts.get(level) ?? 0) + 1),
    new Map<string, number>(),
  );

console.log(levels);

Map(2) { "ERROR" => 2, "WARN" => 1 }

Each step returns a new Enumerable, and the consumer at the end (reduce here) pulls the items through and returns a promise. Callbacks may be async; each call is awaited before the next one starts. To run several at once, use concurrentMap. Key ideas covers how sources, steps, and consumers fit together.

enumerate() is not .enum()

import { enumerate } from "@j50n/proc";

// enumerate() wraps an iterable. The items stay as they are.
console.log(await enumerate(["apple", "pear"]).collect());

// .enum() numbers the items. Each becomes [item, index].
console.log(await enumerate(["apple", "pear"]).enum().collect());

[ "apple", "pear" ]
[ [ "apple", 0 ], [ "pear", 1 ] ]

enumerate(iterable) is a function: it wraps an iterable so you can call the methods on it, and leaves the items alone. .enum() is a method on an Enumerable: it numbers the items, turning each into [item, index] from 0. Python’s enumerate is proc’s .enum().

The methods

Change items:

Filter:

Slice:

Combine:

Split:

Consume (each returns a promise):

A for await loop is a consumer too.

Text, processes, and output:

The output of run() also has pid and status; see Running a command. Files and stdout are in Files and standard streams.

Steps to pass to transform()

The data formats (fromCsvToRows(), toTsv(), …) are steps too, from @j50n/proc/transforms; see Data formats. To write your own, see Writing your own steps.

More examples

import { enumerate, range } from "@j50n/proc";

const words = ["one", "two", "three", "four", "five"];

console.log(await enumerate(words).drop(1).take(2).collect());
console.log(await range({ from: 1, to: 4 }).zip(enumerate(words)).collect());
console.log(
  await enumerate(["a b", "c"]).flatMap((s) => s.split(" ")).collect(),
);
console.log(
  await enumerate(["x", "y"]).concat(enumerate(["z"])).collect(),
);

[ "two", "three" ]
[ [ 1, "one" ], [ 2, "two" ], [ 3, "three" ] ]
[ "a", "b", "c" ]
[ "x", "y", "z" ]

concat and zip take an async iterable, so wrap an array in enumerate() first; the types refuse a plain array.

import { enumerate, range } from "@j50n/proc";

const numbers = () => range({ from: 1, until: 6 }); // 1 to 6, a new one each time

console.log(await numbers().count((n) => n % 2 === 0)); // 3
console.log(await numbers().find((n) => n > 4)); // 5
console.log(await numbers().find((n) => n > 10)); // undefined
console.log(await numbers().some((n) => n > 5)); // true
console.log(await numbers().every((n) => n > 5)); // false
console.log(await numbers().first); // 1, a getter

try {
  await enumerate<number>([]).first;
} catch (error) {
  if (error instanceof RangeError) console.log("empty: no first item");
}

3
5
undefined
true
false
1
empty: no first item

numbers() makes a new Enumerable for each line because each one can be read only once. find, some, every, and first stop reading as soon as they have their answer, and close the source.

import { read } from "@j50n/proc";

// Read the file once, and go over the lines two ways side by side.
const [a, b] = read("app.log").lines.tee();

const [total, errors] = await Promise.all([
  a.count(),
  b.count((line) => line.includes("ERROR")),
]);

console.log(`${errors} of ${total} lines are errors`);

2 of 7 lines are errors

Traps

• An Enumerable is read once. A second collect() gets nothing, without an error. Keep the array from the first, or tee() the sequence.
• tee() and unzip() keep an item until every branch has read it. Branches read side by side hold little; a branch that runs far ahead of another makes everything in between pile up in memory. If the source throws, every branch throws, after the items before the error.
• Read every branch of a tee(), at least to a break. The source is closed once every branch has stopped; a branch that is never read keeps it open, and keeps every item from the start.
• first throws on an empty sequence (RangeError), and reduce with no starting value throws TypeError. find returns undefined instead.
• There is no sort. Sorting needs every item at once, so it isn’t a step. Filter first to keep only what you need, collect(), and sort the array (or pipe lines through the sort command, which handles more than fits in memory).
• flatten() on strings gives characters, because a string is iterable.
• lines and chunkedLines are getters: .lines, not .lines(). Only an Enumerable of bytes has them; on anything else the type is never.

Doing work concurrently

concurrentMap and concurrentUnorderedMap are map with several calls running at once. The first gives results in input order; the second gives them as they finish.

import { enumerate, sleep } from "@j50n/proc";

const jobs = [
  { name: "slow", ms: 300 },
  { name: "fast", ms: 100 },
  { name: "medium", ms: 200 },
];

async function work(job: { name: string; ms: number }) {
  await sleep(job.ms);
  return job.name;
}

// In input order: "fast" and "medium" wait behind "slow".
console.log(
  await enumerate(jobs).concurrentMap(work, { concurrency: 3 }).collect(),
);

// In the order they finish.
console.log(
  await enumerate(jobs).concurrentUnorderedMap(work, { concurrency: 3 })
    .collect(),
);

[ "slow", "fast", "medium" ]
[ "fast", "medium", "slow" ]

Which one to use

concurrentMap keeps the order, and pays for it: a slow item holds back the results after it, and while it does, no new call starts. With one slow item in every few, fewer calls run than you asked for. Use it when the output must line up with the input.

concurrentUnorderedMap starts a new call as soon as any finishes, so every slot stays busy. Use it when order doesn’t matter, or carry the input along in the result (return { file, size }) so you can tell which result is which.

Plain map makes one call at a time, even when the callback is async.

How many at once

The concurrency option sets how many calls may be running. It defaults to navigator.hardwareConcurrency, the number of CPUs. A fraction rounds up, and a value below 1 throws an Error when reading starts, not at the call.

Calls overlap only while they wait: on a timer, a file, the network, or a child process. The JavaScript inside the callbacks still runs one piece at a time, so CPU-heavy work in a callback gets no faster. Put that work in a child process, where it does run in parallel. The CPU count suits commands that keep one CPU busy each; work that mostly waits on the network can go higher.

Running several commands at once

import { enumerate, run } from "@j50n/proc";

const files = ["a.txt", "b.txt", "c.txt"];
for (const file of files) {
  await Deno.writeTextFile(file, `${file}\n`.repeat(10_000));
}

// Up to two gzip processes at a time.
const results = await enumerate(files)
  .concurrentMap(async (file) => {
    await run("gzip", "-k", file).collect(); // wait for it, reading its output
    const before = (await Deno.stat(file)).size;
    const after = (await Deno.stat(`${file}.gz`)).size;
    return `${file}: ${before} -> ${after} bytes`;
  }, { concurrency: 2 })
  .collect();

console.log(results);

[
  "a.txt: 60000 -> 137 bytes",
  "b.txt: 60000 -> 137 bytes",
  "c.txt: 60000 -> 137 bytes"
]

Each call must consume its command’s output, here with collect(), even when there is none to speak of. That is how the call waits for the command to finish, and how a failed command throws: as an ExitCodeError from the callback, which then surfaces as below. A command that writes to stdout and is never read blocks, and its call never finishes (see Key ideas).

When a call fails

import { enumerate, sleep } from "@j50n/proc";

try {
  await enumerate(["a", "b", "c"])
    .concurrentMap(async (name) => {
      await sleep(name === "c" ? 100 : 10);
      if (name === "b") throw new Error(`${name} failed`);
      console.log(`${name} done`);
      return name;
    }, { concurrency: 3 })
    .forEach((name) => console.log(`got ${name}`));
} catch (error) {
  if (error instanceof Error) console.log(`caught: ${error.message}`);
}

// The error didn't stop "c", which was already running.
await sleep(200);

a done
got a
caught: b failed
c done

An error thrown by the callback comes out of the consumer, so one try around the await catches it. concurrentMap throws it when the failed item’s turn comes, after the results before it; concurrentUnorderedMap throws it in the order it happened.

Nothing cancels the calls already running: “c” finished after the error was caught. Their results are dropped, but whatever they do (write a file, start a command) still happens. A consumer that stops early, with take or break, leaves running calls going in the same way.

To let every job finish and see each outcome, catch inside the callback and return the outcome as the result:

import { enumerate, ExitCodeError, run } from "@j50n/proc";

// Catch inside the callback to let every job finish and report each one.
const checks = [["true"], ["sh", "-c", "exit 3"], ["echo", "fine"]];

const results = await enumerate(checks)
  .concurrentMap(async ([cmd, ...args]) => {
    try {
      await run(cmd, ...args).collect();
      return `${cmd}: ok`;
    } catch (error) {
      if (error instanceof ExitCodeError) return `${cmd}: exit ${error.code}`;
      throw error;
    }
  })
  .collect();

console.log(results);

[ "true: ok", "sh: exit 3", "echo: ok" ]

For a larger worked example, see Running jobs in parallel.

Files and standard streams

read(path) gives a file’s bytes as an Enumerable, and writeTo(path) writes bytes to a file. In between, the data streams a chunk at a time.

import { read } from "@j50n/proc";

// Copy the error lines into a file of their own.
await read("app.log")
  .lines
  .filter((line) => line.includes("ERROR"))
  .writeTo("errors.txt"); // each line written with a "\n"

console.log(await read("errors.txt").lines.collect());

[
  "2026-10-05 09:01:13 ERROR db: connection refused",
  "2026-10-05 09:02:45 ERROR GET /orders/17 500"
]

read() opens the file when reading starts, not at the call, and closes it when reading ends, including when the consumer stops early. A missing file throws Deno.errors.NotFound from the consumer’s await. readLines(path) is shorthand for read(path).lines.

Writing a file

writeTo(path) creates the file, or replaces what it held, and closes it when the sequence ends. It writes items as toStdout() does: bytes as they are, and each string as a line, with a "\n" added. If the source throws, including a command in the pipeline that fails, the file is closed holding what was written so far, and the error comes out of writeTo. The file is emptied before anything is read, so what it held before is gone either way, and a pipeline that reads the same file (read(path) … writeTo(path)) finds it already empty, as cmd < f > f does in a shell.

To replace a file only once everything has worked, or to rewrite one in place, pass { atomic: true }:

import { read } from "@j50n/proc";

// Rewrite fruit.txt in place: atomic writes a new file and renames it over
// the old one, so reading the old one while writing works.
await read("fruit.txt")
  .lines
  .map((line) => line.toUpperCase())
  .writeTo("fruit.txt", { atomic: true });

console.log((await Deno.readTextFile("fruit.txt")).trimEnd());

CHERRY
APPLE
BANANA

With atomic, proc writes a new file beside the old one, flushes it to disk, and renames it into place once everything is written. A failure leaves the old file as it was, with nothing beside it, and so does a program that exits partway under main. A symlink stays a symlink, and the file keeps its mode. Anything reached through /dev or /proc, such as /dev/stdout, is written in place. It needs read permission on the file and write permission on its directory, and the result is a new file, so a hard link to the old one still shows the old content.

To add to a file instead of replacing it, open it yourself and pass its writable:

import { enumerate, toBytes } from "@j50n/proc";

// writeTo(path) replaces the file. To add to it, open it yourself.
await enumerate(["first run"]).transform(toBytes).writeTo("notes.txt");

const file = await Deno.open("notes.txt", { append: true });
await enumerate(["second run"]).transform(toBytes).writeTo(file.writable);

console.log(JSON.stringify(await Deno.readTextFile("notes.txt")));

"first run\nsecond run\n"

writeTo(stream) takes any WritableStream and closes it at the end, which closes the file too.

stdout

import { enumerate, read, toBytes } from "@j50n/proc";

// Strings are written as lines; bytes as they are.
await enumerate(["one", "two"]).toStdout();
await read("fruit.txt").toStdout();

// writeTo() closes what it writes to, unless told not to.
await enumerate(["three"])
  .transform(toBytes)
  .writeTo(Deno.stdout.writable, { noclose: true });
console.log("stdout is still open");

one
two
cherry
apple
banana
three
stdout is still open

toStdout() writes each string with a "\n" added, and bytes as they are, and leaves stdout open. Use it for a command’s output too: run("ls").toStdout(). When stdout’s reader goes away before the program is done writing, as when its output is piped into head, toStdout() stops as a consumer that stops early does: it closes the source and resolves, without an error.

writeTo(Deno.stdout.writable) works, but closes stdout when it finishes unless you pass { noclose: true }. After that, every console.log in the program throws BadResource.

stdin

Deno.stdin.readable is a stream of bytes, so enumerate() wraps it like any other source:

import { enumerate } from "@j50n/proc";

// Print the lines of stdin that mention ERROR.
await enumerate(Deno.stdin.readable)
  .lines
  .filter((line) => line.includes("ERROR"))
  .toStdout();

deno run stdin-errors.ts < app.log

2026-10-05 09:01:13 ERROR db: connection refused
2026-10-05 09:02:45 ERROR GET /orders/17 500

Reading stdin needs no permission. A child process started with run() gets no stdin by default; to feed it, pipe into it with .run() (Pipelines and input).

Compressed files

CompressionStream and DecompressionStream take byte chunks, so the bytes from read(), a command’s output, or toBytes go straight in:

import { enumerate, gunzip, gzip, read } from "@j50n/proc";

// Bytes go straight into the web streams.
const errors = await read("app.log.gz")
  .transform(new DecompressionStream("gzip"))
  .lines
  .count((line) => line.includes("ERROR"));
console.log(`${errors} errors`);

await read("app.log")
  .transform(new CompressionStream("gzip"))
  .writeTo("copy.log.gz");

// gzip and gunzip also take lines of text.
await enumerate(["alpha", "beta"]).transform(gzip).writeTo("words.gz");
console.log(await read("words.gz").transform(gunzip).lines.collect());

2 errors
[ "alpha", "beta" ]

The gzip and gunzip steps do the same, and also accept lines of text, which they turn into bytes first. Data that isn’t valid gzip throws TypeError. .run("gzip", "-dc") works too, in a child process.

Lines or bytes

.lines decodes UTF-8 and splits on "\n", dropping a "\r" before it, so CRLF files work. Invalid UTF-8 throws TypeError. Keep the bytes for anything that isn’t text: compressed data, images, or a copy that must be exact. toByteLines splits bytes into lines without decoding them. For files with a great many short lines, .chunkedLines gives the same lines an array at a time, and is several times faster.

Large files

A chain that streams from read() to writeTo() holds only a few chunks at a time, whatever the size of the file: filtering a 1.2 GB log through gzip peaked at 150 MB of process memory, no more than a 300 MB log took. Two things break that: collect(), which keeps every item, and a single line so long that it doesn’t fit, since .lines holds a line until it ends. Count, reduce, or write as you go instead of collecting.

From callbacks to iterables

WritableIterable is a queue: push-style code (callbacks, event handlers, timers) writes items into it, and a for await loop or an Enumerable chain reads them out.

import { enumerate, WritableIterable } from "@j50n/proc";

// A producer that calls back on a timer, as many event sources do.
const ticks = new WritableIterable<number>();
let n = 0;
const timer = setInterval(() => {
  n += 1;
  ticks.write(n);
  if (n === 3) {
    clearInterval(timer);
    ticks.close(); // without this, the loop below waits forever
  }
}, 10);

await enumerate(ticks)
  .map((tick) => `tick ${tick}`)
  .forEach((line) => console.log(line));
console.log("closed");

tick 1
tick 2
tick 3
closed

It has three operations:

• write(item) adds an item to the queue and returns at once.
• close() ends the data. The reader gets every item written, then its loop ends.
• close(error) ends the data with an error. The reader gets every item written before it, then the error is thrown from its loop.

Only the first close counts; later ones are ignored. isClosed tells you whether it has happened.

Bridging events

import { WritableIterable } from "@j50n/proc";

const target = new EventTarget();
const messages = new WritableIterable<string>();

target.addEventListener("message", (event) => {
  // write() rejects once the queue is closed; catch it in a handler, or the
  // unhandled rejection ends the program.
  messages.write((event as CustomEvent<string>).detail).catch(() => {});
});
target.addEventListener("error", () => {
  messages.close(new Error("the source failed"));
});

// Nothing is reading yet, so these just queue.
target.dispatchEvent(new CustomEvent("message", { detail: "hello" }));
target.dispatchEvent(new CustomEvent("message", { detail: "world" }));
target.dispatchEvent(new Event("error"));
target.dispatchEvent(new CustomEvent("message", { detail: "too late" }));

try {
  for await (const message of messages) {
    console.log(message);
  }
} catch (error) {
  if (error instanceof Error) console.log(`error: ${error.message}`);
}

hello
world
error: the source failed

The handlers run before anything reads, so the items wait in the queue until the loop starts. The error event closes the queue with an error, which the reader sees after “hello” and “world”; the message after that is refused.

Feeding a command

A WritableIterable can be a command’s stdin. .run() starts the command at once and passes items to it as they are written:

import { enumerate, WritableIterable } from "@j50n/proc";

// The command starts now, and reads its stdin as items are written.
const input = new WritableIterable<string>();
const sorted = enumerate(input).run("sort").lines.collect();

for (const fruit of ["pear", "apple", "fig"]) {
  await input.write(fruit);
}
await input.close(); // sort sees the end of its input

console.log(await sorted);

[ "apple", "fig", "pear" ]

Enumerable.writeTo() also accepts a WritableIterable, to copy a sequence into one.

Traps

Nothing closes it for you. Until close() is called, the reader waits for the next item. If something else keeps the program alive (a timer, a server), it waits forever; if nothing does, Deno stops with error: Top-level await promise never resolved. Make sure every way the source can end, including failure, leads to a close.

There is no backpressure. write() doesn’t wait for the reader; its promise resolves right away. When the writer is faster than the reader, or nothing reads at all, the queue grows without limit and every item stays in memory. That suits events, which can’t be paused anyway. If the producer can wait, and the data is large, make it a generator instead and let the reader pull.

write() after close() rejects. In an event handler, nobody awaits that promise, so the rejection is unhandled and ends the program. Catch it, as the example does, or check isClosed first.

A reader that stops doesn’t stop the producer. After a break, take, or first, write() still succeeds, but the items go nowhere. If the producer should stop when the reader does, close the queue yourself when the loop ends, in a finally, and have the producer check isClosed.

One reader, once. Read it with a single loop or chain. A second reader, at the same time or after the first, throws TypeError: a WritableIterable can be read only once.

The constructor takes an onclose callback, called on the first close(), which waits for it: a place to remove event listeners or clear a timer. See the API reference.

Writing your own steps

When no method does what you need, write the step as an async generator function and pass it to transform(). It gets the whole sequence, so it can keep state between items and yield as many or as few as it likes.

import { enumerate, read, type TransformerFunction } from "@j50n/proc";

// Fewer items, with state: drop a line that repeats the one before it.
async function* dedupe(lines: AsyncIterable<string>) {
  let previous: string | undefined;
  for await (const line of lines) {
    if (line !== previous) yield line;
    previous = line;
  }
}

// Batching: a function that makes the step, so the size can vary.
function batches<T>(size: number): TransformerFunction<T, T[]> {
  return async function* (items) {
    let batch: T[] = [];
    for await (const item of items) {
      batch.push(item);
      if (batch.length === size) {
        yield batch;
        batch = [];
      }
    }
    if (batch.length > 0) yield batch; // the last, short batch
  };
}

console.log(
  await enumerate(["a", "a", "b", "a", "a"]).transform(dedupe).collect(),
);
console.log(await read("fruit.txt").lines.transform(batches(2)).collect());

[ "a", "b", "a" ]
[ [ "cherry", "apple" ], [ "banana" ] ]

A step is any function from an AsyncIterable<T> to an AsyncIterable<U>; the type TransformerFunction<T, U> names it. A generator function is the step itself: pass dedupe, not dedupe(). For a step that takes settings, write a function that returns the step, and call that: batches(2). The built-in steps come in both kinds (toBytes and gzip are passed as they are; buffer(size) and the data formats’ fromCsvToRows() are called); the list is in Enumerable and its methods.

More items than came in

yield* hands on every item of an iterable:

import { enumerate } from "@j50n/proc";

// More items: each line becomes its words.
async function* words(lines: AsyncIterable<string>) {
  for await (const line of lines) {
    yield* line.split(/\s+/).filter((word) => word.length > 0);
  }
}

console.log(
  await enumerate(["the quick", "  brown fox "]).transform(words).collect(),
);

[ "the", "quick", "brown", "fox" ]

For a step that only changes or drops single items, map, filter, and flatMap are shorter. Reach for a generator when the step needs memory of earlier items, needs to see the end (to flush a last batch, or emit a total), or needs to stop early.

Stopping early

import { run } from "@j50n/proc";

// A step that returns before its input ends closes the source behind it.
async function* untilEnd(lines: AsyncIterable<string>) {
  for await (const line of lines) {
    if (line === "END") return;
    yield line;
  }
}

const lines = await run("sh", "-c", "echo a; echo b; echo END; echo c; exit 3")
  .lines
  .transform(untilEnd)
  .collect();
console.log(lines); // no error: the exit code isn't checked after an early stop

[ "a", "b" ]

Returning from the generator before its input ends closes the source, as take does: the command’s output is closed, and its exit code isn’t checked. Code in a finally block of the generator runs when the step stops for any reason, including a consumer that stops early downstream.

Errors

An error thrown in a step comes out of the consumer’s await, unchanged. An error from upstream (a failed command, a callback that threw) is thrown inside the step, from its for await loop, so a step can catch it, to add context or to recover:

import { ExitCodeError, run } from "@j50n/proc";

// A step sees errors from upstream in its own loop, and can add context.
async function* numbered(lines: AsyncIterable<string>) {
  let n = 0;
  try {
    for await (const line of lines) {
      n += 1;
      yield `${n}: ${line}`;
    }
  } catch (error) {
    throw new Error(`input failed after line ${n}`, { cause: error });
  }
}

try {
  await run("sh", "-c", "echo a; echo b; exit 2")
    .lines
    .transform(numbered)
    .forEach((line) => console.log(line));
} catch (error) {
  if (error instanceof Error) {
    console.log(error.message);
    if (error.cause instanceof ExitCodeError) {
      console.log(`cause: exit code ${error.cause.code}`);
    }
  }
}

1: a
2: b
input failed after line 2
cause: exit code 2

Rethrow with the original as cause rather than swallowing it. A step that catches an error and simply returns hides the failure: the pipeline ends as if the input had run out.

With a TransformStream

transform() also takes a TransformStream, or any { writable, readable } pair, such as CompressionStream:

import { enumerate } from "@j50n/proc";

// A TransformStream works too.
function upper() {
  return new TransformStream<string, string>({
    transform(line, controller) {
      controller.enqueue(line.toUpperCase());
    },
  });
}

console.log(await enumerate(["one", "two"]).transform(upper()).collect());

// A stream is used up after one pass: the second use yields nothing.
const once = upper();
console.log(await enumerate(["three"]).transform(once).collect());
console.log(await enumerate(["four"]).transform(once).collect());

[ "ONE", "TWO" ]
[ "THREE" ]
[]

A stream works once. Used a second time, it yields nothing or throws, and the source goes unread, so create a new one for each use, as upper() does. An error from upstream of the stream reaches the consumer unchanged, and so does one thrown in its transform(). A generator is usually simpler to write; use a stream when you already have one.

Utilities

Small helpers that come with proc, for the jobs scripts keep needing.

range

import { range } from "@j50n/proc";

console.log(await range({ to: 3 }).collect()); // stops before `to`
console.log(await range({ from: 1, until: 3 }).collect()); // may end on `until`
console.log(await range({ from: 10, to: 0, step: -3 }).collect());
console.log(await range({ from: 1, to: Infinity }).take(3).collect());

[ 0, 1, 2 ]
[ 1, 2, 3 ]
[ 10, 7, 4, 1 ]
[ 1, 2, 3 ]

range counts from from (default 0) by step (default 1), as an Enumerable. With to it stops before to; with until it includes until when a step lands on it. A negative step counts down, a step that moves away from the limit gives nothing, and a step of 0 throws RangeError. Numbers are made as they are read, so to: Infinity is fine with take. Fractional steps gather floating-point error: { until: 0.3, step: 0.1 } stops at 0.2.

sleep and the time constants

import { MINUTES, SECONDS, sleep } from "@j50n/proc";

console.log(`${5 * MINUTES} ms in five minutes`);

// Retry a flaky step, waiting longer each time.
let attempts = 0;
async function flaky() {
  attempts += 1;
  await sleep(1); // stands in for the real work
  if (attempts < 3) throw new Error("not yet");
  return "ok";
}

for (let wait = 0.01 * SECONDS;; wait *= 2) {
  try {
    console.log(`${await flaky()} after ${attempts} attempts`);
    break;
  } catch {
    await sleep(wait);
  }
}

300000 ms in five minutes
ok after 3 attempts

sleep(ms) resolves after ms milliseconds without blocking anything else. SECONDS, MINUTES, HOURS, DAYS, and WEEKS are plain numbers of milliseconds, for sleep, cache, or anything else that takes milliseconds: sleep(2 * SECONDS).

shuffle

import { enumerate, shuffle } from "@j50n/proc";

// Run the tests in a random order. shuffle() changes the array in place.
const tests = ["parse", "render", "save", "load"];
shuffle(tests);

await enumerate(tests).forEach((test) => console.log(test)); // order varies

shuffle reorders an array in place and returns nothing. It uses Math.random, so it is not for anything that must be unpredictable, such as tokens.

concat and concatLines

import { concat, concatLines, read } from "@j50n/proc";

// The whole file as one Uint8Array.
const bytes = concat(await read("fruit.txt").collect());
console.log(bytes.length);

// Lines of bytes back into text, with "\n" after each.
const encoder = new TextEncoder();
const text = concatLines([encoder.encode("one"), encoder.encode("two")]);
console.log(JSON.stringify(new TextDecoder().decode(text)));

20
"one\ntwo\n"

concat joins byte arrays into one; concatLines does the same with a "\n" after each. Given a single array, concat returns that array itself, not a copy. These are not Enumerable.concat, which appends one sequence to another.

cache

cache(key, compute, { timeout }) returns the value stored under key if it is younger than timeout (default 24 hours); otherwise it calls compute, stores the result, and returns it. The values live in Deno KV’s default database, so they last across runs, which is the point: an expensive lookup is done once a day, not every time the script runs.

This is a fragment, not a checked example, because Deno KV is unstable: the program needs --unstable-kv (or "unstable": ["kv"] in deno.json), and the book’s examples run without it. Without the flag, cache throws a TypeError that says so.

import { cache, HOURS, run } from "@j50n/proc";

const branches = await cache(
  ["git", "remote-branches"],
  () => run("git", "ls-remote", "--heads", "origin").lines.collect(),
  { timeout: 4 * HOURS },
);

null and undefined are never stored, so compute runs every time for them. A value must fit in a KV entry (structured-cloneable, at most 64 KiB) to be cached; one that doesn’t is returned without being stored, so compute runs every time for it. Deno KV deletes each entry once it is older than the timeout it was stored with. Two calls that miss at the same time both compute.

To recompute whatever is stored, pass refresh: true: compute runs, and its result is stored as on a miss, so later calls read it. A script’s --refresh flag maps straight onto it. A timeout of 0 also skips the read, but stores nothing: use it for a call that shouldn’t touch the cache at all. A refresh whose result can’t be stored removes the old entry, so a later call computes again rather than getting the value the refresh replaced.

A value read back from the cache is a structured clone. A class instance comes back as a plain object, without its methods or getters, though its type still says it is the class; cache plain data.

Deno picks which database “default” means. With a deno.json, every script in that project shares one, so two scripts that both use the key "config" get each other’s values; without one, each main script has its own. The database is a SQLite file under Deno’s cache directory (DENO_DIR), unencrypted, and an entry stays in it until it expires, so don’t cache secrets such as tokens. See the API reference.

debug

import { debug, read } from "@j50n/proc";

// Print each item as it passes this point, then carry on.
const b = await read("fruit.txt")
  .lines
  .transform(debug<string>)
  .filter((fruit) => fruit.startsWith("b"))
  .collect();

console.log(b);

"cherry"
"apple"
"banana"
[ "banana" ]

debug is a step that prints each item as JSON, in blue, as it passes, and hands it on unchanged. It prints with console.log, so the lines land on stdout among the program’s own output. Remove it when you’re done; an item JSON can’t hold (a BigInt) throws.

isString

import { isString } from "@j50n/proc";

const values: unknown[] = ["text", 42, new String("boxed")];
console.log(values.map(isString));

[ true, false, false ]

isString is typeof value === "string" as a type guard, so TypeScript narrows the value after it. A String object is not a string by this test.

Reading and writing data formats

@j50n/proc/transforms turns bytes into rows and rows back into bytes, a batch at a time, so a CSV export or a process’s output streams through your code without being read whole into memory.

import { read } from "@j50n/proc";
import { fromCsvToRows, toTsv } from "@j50n/proc/transforms";

await read("data-orders.csv")
  .transform(fromCsvToRows()) // bytes in, batches of rows out
  .flatten() // one row at a time
  .filter((row) => row[3] !== "1") // row is a string[]
  .transform(toTsv()) // rows in, bytes out
  .toStdout();

id	customer	item	qty
1	Ada	Widget, large	2
3	Linus	Bolt	10

Each function returns a transformer for .transform(). A parser (fromCsvToRows(), …) takes bytes from read(), a command’s output, or any async iterable of Uint8Array. A writer (toTsv(), …) yields bytes, ready for .writeTo(), .toStdout(), or the stdin of the next command with .run(). The transforms are a separate entry point, so import them from "@j50n/proc/transforms".

Which format

• CSV to exchange data with spreadsheets, databases, and other people. Its parser is lenient; CSV says exactly what it accepts.
• TSV for simple tabular text whose fields never hold a tab or line break: logs, command output, files you grep.
• JSON lines when each item is an object (or any JSON value) rather than a row of strings.
• Record to hand rows to another program when fields may hold tabs, quotes, or newlines. Any language splits it with two calls.

To convert, chain any parser to any writer: .transform(fromCsvToRows()) then .transform(toRecord()). The row writers take what the row parsers yield, batches and LazyRows included. Between CSV and TSV, csvToTsv() and tsvToCsv() convert bytes to bytes without making rows, several times faster; see CSV. JSON is the exception, since it holds values rather than rows: map rows to objects before toJson(), and objects to arrays of strings after fromJsonToRows(). JSON lines shows both.

Batches

Parsers yield batches (arrays of rows), not single rows. Add .flatten() before a step that works on one row (filter, map, take), as in Key ideas. The row writers take a row or a batch per item, so there is no need to flatten just to write. toJson() takes one value per item, so flatten before it; see JSON lines.

Row or LazyRow

A Row is a string[], every field decoded. A LazyRow from fromCsvToLazyRows() or fromTsvToLazyRows() holds the row’s bytes and decodes a field when you ask for it with getField(), or compares one without decoding it with fieldEquals(). Use plain rows unless you filter on or read a few fields of each row; LazyRow has the details.

Header rows

No format has a header. A header line is the first row, like any other, so in the example above it went through the filter and into the TSV. Skip it with .drop(1), or keep it to name the fields:

import { read } from "@j50n/proc";
import { fromCsvToRows } from "@j50n/proc/transforms";

// The header is the first row, like any other.
let header: string[] = [];
const orders = await read("data-orders.csv")
  .transform(fromCsvToRows())
  .flatten()
  .enum()
  .filter(([row, index]) => {
    if (index === 0) header = row;
    return index > 0;
  })
  .map(([row]) => Object.fromEntries(header.map((name, i) => [name, row[i]])))
  .collect();

console.log(orders);

// Or skip it.
const quantities = await read("data-orders.csv")
  .transform(fromCsvToRows())
  .flatten()
  .drop(1)
  .map((row) => Number(row[3]))
  .collect();

console.log(quantities);

[
  { id: "1", customer: "Ada", item: "Widget, large", qty: "2" },
  { id: "2", customer: "Grace", item: 'Gear "XL"', qty: "1" },
  { id: "3", customer: "Linus", item: "Bolt", qty: "10" }
]
[ 2, 1, 10 ]

enum() pairs each row with its index, counted from 0. Every field is a string; convert numbers yourself, and turn them back into strings before writing.

To write a header in front of rows, put it first: enumerate([header]).concat(rows), where rows is any Enumerable of rows.

The CSV and TSV parsers, and csvToTsv() and tsvToCsv(), run in WebAssembly bundled with the package; the rest is TypeScript. None of them needs permissions of its own; only read() and writeTo() do.

What writers refuse

A writer throws rather than write a field its format can’t hold, since the reader would split it into extra fields or rows:

import { enumerate } from "@j50n/proc";
import { toCsv, toTsv } from "@j50n/proc/transforms";

const rows = [["id", "note"], ["1", "line one\nline two"]];

try {
  await enumerate(rows).transform(toTsv()).toStdout();
} catch (error) {
  if (error instanceof Error) console.log(`[${error.message}]`);
}

// CSV can hold it: the field is quoted.
await enumerate(rows).transform(toCsv()).toStdout();

id	note
[Invalid character (LF) in TSV data at row 2, field 2]
id,note
1,"line one
line two"

The Error names the row and field, both counted from 1 across the whole stream. Items before the bad one have already been written, so a file you were writing holds the rows up to that point. toCsv() refuses no field.

A writer also refuses a row that would read back as no row, or as a different one, with an error such as Invalid row (no fields) in CSV data at row 3:

csvToTsv() refuses a CSV row of one empty field, as toTsv() does. Every row writer refuses a field holding a lone surrogate, which UTF-8 can’t hold (TextEncoder would write U+FFFD in its place). A first field starting with U+FEFF would be dropped by the reader as a byte order mark, so toCsv() and tsvToCsv() quote it and the others refuse it.

What parsers refuse

The CSV and TSV parsers take lines ending in LF or CRLF. Any other CR outside a quoted field, as in a file with old Mac CR-only line ends, throws an Error such as Invalid character (CR) in CSV data at row 1, field 2, rather than read the file as one long row. The CSV parser throws on a quote still open at the end of the input, Unclosed quote in CSV data at row 7, field 3, rather than make the rest of the file one field. Invalid UTF-8 throws a TypeError such as Invalid UTF-8 in CSV data at row 3001, field 2 (a file saved as Latin-1 or Windows-1252 is the usual cause) from the parser, or, for a LazyRow, when the field is decoded. Batches before the one holding the error have already gone down the pipeline.

A parser’s row numbers count rows, from 1, with the header row included. Blank lines are skipped and not counted, and a quoted CSV field can span lines, so a row number is the line number only in a file with neither.

The flatdata CLI converts between the same formats in a separate process, with the same checks.

How fast

From benchmarks/transforms-throughput.ts: 100,000 rows of 20 fields (UTF-8, about one field in ten quoted, some with newlines), median MB/s on one machine. Your numbers will differ; the ratios are what matter.

The LazyRow parsers are fast because they make no strings until asked, and converting between CSV and TSV never makes any.

Lots of rows

The figures above are for a program that handles each batch with plain code. A step after .flatten() (filter, map, forEach) is an await per row, about a microsecond each, which on small rows costs more than the parsing. With millions of rows, work a batch at a time, with the array’s own methods inside one step:

import { enumerate } from "@j50n/proc";
import { fromCsvToLazyRows } from "@j50n/proc/transforms";

// 100,000 orders; every tenth is from New Zealand.
const csv = Array.from(
  { length: 100_000 },
  (_, i) => `${i},customer${i},${i % 10 === 0 ? "NZ" : "US"},${i % 97}\n`,
).join("");
const orders = () =>
  enumerate([new TextEncoder().encode(csv)]).transform(fromCsvToLazyRows());

// Row by row: every step after .flatten() is an await per row.
const rowByRow = await orders()
  .flatten()
  .filter((row) => row.fieldEquals(2, "NZ"))
  .count();

// A batch at a time: plain array methods inside one step per batch.
const byBatch = await orders()
  .map((rows) => rows.filter((row) => row.fieldEquals(2, "NZ")))
  .flatten()
  .count();

console.log(rowByRow, byBatch);

10000 10000

Both count the same rows. On 100,000 rows of 20 fields, the filter ran at about 290 MB/s a batch at a time and 135 MB/s row by row; on rows of 25 bytes, row by row fell to about 25 MB/s. It is the same advice as .chunkedLines for lines of text.

Against other JavaScript

From benchmarks/compare.ts, on the same data. Here every reader keeps all 2,000,000 fields in memory, as a whole-string parser must, so the figures are lower than the table above. The others get their best case, the whole file decoded to one string; proc streams it in 64 KB chunks.

Converting CSV to TSV, proc’s csvToTsv() runs at about 540 MB/s here; reading with the hand-written parser and joining the rows back runs at 30. Reading TSV, proc and a plain split on the whole string are close (72 and 59): TSV has no quoting to work out, so making the strings is nearly all the work.

CSV

fromCsvToRows() parses CSV into batches of rows, and toCsv() writes rows as CSV, quoting the fields that need it.

import { read } from "@j50n/proc";
import { fromCsvToRows } from "@j50n/proc/transforms";

const rows = await read("data-orders.csv")
  .transform(fromCsvToRows())
  .flatten()
  .drop(1) // the header
  .collect();

console.log(rows);

[
  [ "1", "Ada", "Widget, large", "2" ],
  [ "2", "Grace", 'Gear "XL"', "1" ],
  [ "3", "Linus", "Bolt", "10" ]
]

The quotes are gone: "Gear ""XL""" in the file is Gear "XL" in the row. The first row is the header, read like any other row; see Header rows. The parser runs in WebAssembly and yields a batch of rows for about every 128 KiB of input, so .flatten() comes before the per-row steps.

fromCsvToLazyRows() parses the same way but yields LazyRows, which decode a field only when you read it.

Writing

import { enumerate } from "@j50n/proc";
import { toCsv } from "@j50n/proc/transforms";

const rows = [
  ["item", "note"],
  ["Widget, large", 'says "hi"'],
  ["Bolt", "two\nlines"],
  ["Nut", " spaces kept "],
];

await enumerate(rows).transform(toCsv()).toStdout();
console.log("--");
await enumerate(rows).transform(toCsv({ separator: ";" })).toStdout();

item,note
"Widget, large","says ""hi"""
Bolt,"two
lines"
Nut, spaces kept 
--
item;note
Widget, large;"says ""hi"""
Bolt;"two
lines"
Nut; spaces kept 

toCsv() quotes a field that holds the separator, a quote, CR, or LF, and doubles the quotes inside it. Other fields are written as they are, spaces included, except that a first field starting with U+FEFF is quoted, since readers drop a byte order mark at the start. It refuses only what CSV can’t hold: a row with no fields ([] inside a batch), which would be a blank line and read back as no row, and a field holding a lone surrogate, which UTF-8 can’t hold. Each item can be a row or a batch, of Rows or LazyRows.

Options

The separator must be one ASCII character other than ", CR, or LF. ";", "|", and "\t" all work. Anything else throws a RangeError from the call that takes the options, such as fromCsvToRows() itself, before any data moves. There are no other options: no header handling, no quote character, no comment lines (a line starting with # is data), no trimming.

What the parser accepts

It reads RFC 4180 and is lenient about the rest. It refuses two things: a CR outside quotes that isn’t part of a CRLF, and a quote still open at the end of the input:

import { enumerate } from "@j50n/proc";
import { fromCsvToRows, toCsv } from "@j50n/proc/transforms";

async function parse(text: string): Promise<string> {
  try {
    const rows = await enumerate([new TextEncoder().encode(text)])
      .transform(fromCsvToRows())
      .flatten()
      .collect();
    return JSON.stringify(rows);
  } catch (error) {
    return error instanceof Error ? error.message : String(error);
  }
}

const inputs = [
  "a,b\r\nc,d", // CRLF; no line end on the last row
  "a\n\n\nb\n", // blank lines are skipped
  ",a,,\n", // empty fields
  '""\n', // one empty field: a row, not a blank line
  'a,"x\ny, ""z"""\n', // quoted line break, separator, quotes
  "a\nb,c,d\n", // rows may differ in length
  ' a , "b"\n', // spaces are kept, so this quote is text
  'a"b,c\n', // a quote inside an unquoted field is text
  '"ab" ,c\n', // text after a closing quote is kept
  'a,"bc\nd,e\n', // a quote still open at the end is an error
  '"x\ry",z\n', // a CR inside quotes is text
  "a,b\rc,d\r", // a lone CR outside quotes is an error
];
for (const text of inputs) {
  console.log(JSON.stringify(text).padEnd(26), await parse(text));
}

// toCsv writes a row holding one empty field as "", so it survives.
const text = await enumerate([["a"], [""], ["b"]])
  .transform(toCsv())
  .map((bytes) => new TextDecoder().decode(bytes))
  .reduce((all, chunk) => all + chunk, "");
console.log(JSON.stringify(text), await parse(text));

"a,b\r\nc,d"               [["a","b"],["c","d"]]
"a\n\n\nb\n"               [["a"],["b"]]
",a,,\n"                   [["","a","",""]]
"\"\"\n"                   [[""]]
"a,\"x\ny, \"\"z\"\"\"\n"  [["a","x\ny, \"z\""]]
"a\nb,c,d\n"               [["a"],["b","c","d"]]
" a , \"b\"\n"             [[" a "," \"b\""]]
"a\"b,c\n"                 [["a\"b","c"]]
"\"ab\" ,c\n"              [["ab ","c"]]
"a,\"bc\nd,e\n"            Unclosed quote in CSV data at row 1, field 2
"\"x\ry\",z\n"             [["x\ry","z"]]
"a,b\rc,d\r"               Invalid character (CR) in CSV data at row 1, field 2
"a\n\"\"\nb\n" [["a"],[""],["b"]]

• LF and CRLF end a row, and the last row needs no line end.
• Blank lines are skipped, but "" on a line is a row holding one empty field; toCsv() writes such a row that way, so it survives a round trip, as the last line shows.
• A quoted field can hold separators, line breaks (CR included), and doubled quotes.
• Rows may have different numbers of fields. Check row.length (a LazyRow’s columnCount) if yours must match: getField past the end throws a RangeError that can’t say which row, so count rows yourself to report one.
• Spaces around a field are kept. A quote opens a quoted field only at the very start of a field, so a quote after a space, or inside an unquoted field, is text. Text after a closing quote is kept: "ab" ,c reads as ["ab ", "c"].
• A UTF-8 byte order mark at the start, as spreadsheet programs write, is dropped.
• Any other CR outside quotes throws an Error naming the row and field, so a file with CR-only line ends fails at its first line rather than reading as one long row.
• A quote still open at the end of the input throws an Error naming the row and field where it opened, as in Unclosed quote in CSV data at row 1, field 2, rather than make the rest of the file one field.

Either error comes after the batches before the one holding it have been yielded.

Converting to and from TSV

csvToTsv() and tsvToCsv() convert bytes to bytes in WebAssembly, without making rows or strings, several times faster than a parser and a writer:

import { read } from "@j50n/proc";
import { csvToTsv, tsvToCsv } from "@j50n/proc/transforms";

// Bytes to bytes, with no rows in between.
await read("data-orders.csv").transform(csvToTsv()).writeTo("orders.tsv");
await read("orders.tsv").transform(tsvToCsv({ separator: ";" })).toStdout();

id;customer;item;qty
1;Ada;Widget, large;2
2;Grace;"Gear ""XL""";1
3;Linus;Bolt;10

They read and write as the parsers and writers do, errors included. TSV can’t hold a tab, CR, or LF in a field, so csvToTsv() throws on a CSV field holding one, as toTsv() does. Nor can it hold a row of one empty field ("" on a line), which would be a blank line, so csvToTsv() throws on that too: Invalid row (one empty field) in TSV data at row 4. The first error in the input is the one thrown: an unclosed quote whose field takes in a line break reports the LF. The output passed on before it can end partway through a row.

They copy field bytes without decoding them, so invalid UTF-8 doesn’t throw here: it passes through to the output unchanged, as text in any other ASCII-compatible encoding does. The parsers would throw a TypeError on the same input.

Other traps

• Invalid UTF-8 throws a TypeError naming the row and field, from fromCsvToRows() as its batch is converted, and from fromCsvToLazyRows() only when the bad field is decoded.
• Every field is a string. Number(row[3]) for numbers; an empty field is "", and Number("") is 0.
• A row is held whole until it ends, in the WebAssembly module’s memory, which can’t pass 4 GiB. A row takes about its own size plus 8 bytes for each of its fields, and up to twice that while its buffers grow by doubling: about 2 times its size for a row of long fields, and up to 18 times for one of empty fields (,,,,). So a row of about a gigabyte of long fields, or about 150 MB of empty ones, is too large, and throws Row too large for the WebAssembly module's memory in CSV data at row 7. tsvToCsv() holds the longest field whole instead, with about the same limit. The memory a row took isn’t given back until the stream ends. A quote opened near the start of untrusted input and never closed holds all the rest before the error, at about twice its size. Each open stream also holds a floor of about 1 MiB.

See fromCsvToRows, toCsv, and csvToTsv for the reference.

TSV

TSV is a tab between fields and a line feed after each row, with no quoting. fromTsvToRows() reads it and toTsv() writes it.

import { read } from "@j50n/proc";
import { fromTsvToRows } from "@j50n/proc/transforms";

const big = await read("data-orders.tsv")
  .transform(fromTsvToRows())
  .flatten()
  .drop(1) // the header
  .filter((row) => Number(row[3]) > 1)
  .map(([, customer, item]) => `${customer}: ${item}`)
  .collect();

console.log(big);

[ "Ada: Widget, large", "Linus: Bolt" ]

The parser is the CSV parser with a tab separator and quoting turned off, in WebAssembly, and yields a batch of rows for about every 128 KiB of input. There are no options.

Because nothing is quoted, TSV suits data you also want to read with cut, awk, or grep, and whose fields never hold a tab or a line break. When they might, use CSV or the record format.

Edge cases

import { enumerate } from "@j50n/proc";
import { fromTsvToRows } from "@j50n/proc/transforms";

async function parse(text: string): Promise<string> {
  try {
    const rows = await enumerate([new TextEncoder().encode(text)])
      .transform(fromTsvToRows())
      .flatten()
      .collect();
    return JSON.stringify(rows);
  } catch (error) {
    return error instanceof Error ? error.message : String(error);
  }
}

const inputs = [
  "a\tb\r\nc\td", // CRLF; no line end on the last row
  "a\n\nb\n", // blank lines are skipped
  "\t\n", // but a lone tab is two empty fields
  '"a\tb"\tc\n', // quotes are text, so they don't protect the tab
  "a\t\tb\t\n", // empty fields
  "a\tb\rc\n", // a CR that isn't part of CRLF is an error
];
for (const text of inputs) {
  console.log(JSON.stringify(text).padEnd(18), await parse(text));
}

"a\tb\r\nc\td"     [["a","b"],["c","d"]]
"a\n\nb\n"         [["a"],["b"]]
"\t\n"             [["",""]]
"\"a\tb\"\tc\n"    [["\"a","b\"","c"]]
"a\t\tb\t\n"       [["a","","b",""]]
"a\tb\rc\n"        Invalid character (CR) in TSV data at row 1, field 2

• Lines end in LF or CRLF, and the last line needs no LF.
• Blank lines are skipped. A line holding only spaces, or only a tab, is a row.
• Quotes are text. A CSV-style quoted field doesn’t protect a tab.
• A CR anywhere else throws an Error naming the row and field, so a file with CR-only line ends fails at its first line. TSV can’t hold a CR in a field.
• A UTF-8 byte order mark at the start is dropped.

Writing

toTsv() joins fields with tabs and ends each row with LF. A field holding a tab, CR, or LF throws an Error naming the row and field, as in Invalid character (tab) in TSV data at row 2, field 2; rows before it have already been written (What writers refuse shows one). A row of one empty field, [""], throws too, since it would be a blank line and the parser skips those: Invalid row (one empty field) in TSV data at row 3. So do a lone surrogate, which UTF-8 can’t hold, and a first field starting with U+FEFF, which the parser would drop as a byte order mark. If your fields might hold these and a lossy fix is acceptable, replace them first:

import { enumerate } from "@j50n/proc";
import { toTsv } from "@j50n/proc/transforms";

const rows = [["id", "note"], ["1", "tab\there"], ["2", "two\r\nlines"]];

// TSV can't hold a tab, CR, or LF in a field. If yours might, replace them.
await enumerate(rows)
  .map((row) => row.map((field) => field.replace(/[\t\r\n]+/g, " ")))
  .transform(toTsv())
  .toStdout();

id	note
1	tab here
2	two lines

LazyRows from TSV

fromTsvToLazyRows() yields LazyRows that decode a field only when you read it. To filter on a field or two, it is several times faster than fromTsvToRows().

To convert TSV to CSV, tsvToCsv() skips the rows altogether; see CSV.

See fromTsvToRows and toTsv for the reference.

JSON lines

JSON lines is one JSON value per line. fromJsonToRows() parses each line with JSON.parse, and toJson() writes each value with JSON.stringify.

import { read } from "@j50n/proc";
import { fromJsonToRows } from "@j50n/proc/transforms";

type Event = { id: string; level: string; ms: number };

const slow = await read("data-events.jsonl")
  .transform(fromJsonToRows<Event>()) // the type is asserted, not checked
  .flatten()
  .filter((event) => event.ms > 10)
  .map((event) => event.id)
  .collect();

console.log(slow);

[ "e1", "e2" ]

Despite its name, fromJsonToRows() yields values, not rows: objects, arrays, strings, numbers, anything JSON holds. They come in batches of about 128 KiB of text, so .flatten() before working value by value. Blank lines, and lines of only whitespace, are skipped. A CR at the end of a line is whitespace to JSON.parse, so CRLF files read fine.

The type parameter (fromJsonToRows<Event>()) is only an assertion. Nothing checks the values unless you pass a schema.

Checking values

import { enumerate } from "@j50n/proc";
import { fromJsonToRows } from "@j50n/proc/transforms";

type Event = { id: string; ms: number };

// Anything with a parse() that throws on a bad value works; a Zod schema does.
const EventSchema = {
  parse(value: unknown): Event {
    const e = value as Partial<Event>;
    if (typeof e?.id !== "string" || typeof e.ms !== "number") {
      throw new TypeError(`not an event: ${JSON.stringify(value)}`);
    }
    return e as Event;
  },
};

const text = '{"id":"e1","ms":12}\n{"id":"e2","ms":"slow"}\n';

try {
  await enumerate([new TextEncoder().encode(text)])
    .transform(fromJsonToRows({ schema: EventSchema }))
    .flatten()
    .forEach((event) => console.log(event.id));
} catch (error) {
  if (error instanceof TypeError) console.log(error.message);
}

not an event: {"id":"e2","ms":"slow"}

Whatever schema.parse throws stops the stream and reaches your catch as it is, so instanceof finds a Zod error. What it returns is the value you get, so a Zod schema’s defaults and transforms apply, and keys it doesn’t know are stripped, as with any parse call.

With sampleSize, only the first sampleSize values go through the schema. The rest come as JSON.parse made them, unchecked and untransformed, though they are typed as the schema’s output, just as values are with no schema at all. So use sampleSize only with a schema that checks values without changing them.

Note that e1 never printed. A batch is parsed whole before it is yielded, so an error stops the stream before any value in the same batch reaches you.

A line that isn’t JSON throws a SyntaxError naming the line, counted from 1 with blank lines included: Invalid JSON at line 4001: Unexpected token ....

Writing

import { enumerate, read } from "@j50n/proc";
import { fromCsvToRows, toJson } from "@j50n/proc/transforms";

// toJson takes one value per item, so flatten the parser's batches first.
await read("data-orders.csv")
  .transform(fromCsvToRows())
  .flatten()
  .drop(1) // the header
  .map(([id, customer, , qty]) => ({ id, customer, qty: Number(qty) }))
  .transform(toJson())
  .toStdout();

// A value with no JSON form is refused, naming the item.
try {
  await enumerate([{ total: 13 }, undefined]).transform(toJson()).toStdout();
} catch (error) {
  if (error instanceof TypeError) console.log(error.message);
}

{"id":"1","customer":"Ada","qty":2}
{"id":"2","customer":"Grace","qty":1}
{"id":"3","customer":"Linus","qty":10}
{"total":13}
Item 2 can't be written as JSON (undefined)

toJson() takes one value per item and writes it on a line of its own, the reverse of fromJsonToRows() followed by .flatten(). The parsers yield batches, so flatten them first; a batch passed as it is would be written as one JSON array on one line.

Inside a value, JSON.stringify’s rules apply: a property holding undefined or a function is left out, and in an array it becomes null. NaN and Infinity become null too, without an error, so check numbers parsed from text (Number("12,5") is NaN) before they get here. An item with no JSON form at all (undefined, a function, a symbol) throws a TypeError naming the item, counted from 1, as the second half of the example shows. So does an item JSON.stringify throws on, such as a BigInt. Items before it have already been written.

From JSON to rows

To write JSON values as CSV or TSV, turn each into an array of strings: .map((e) => [e.id, String(e.ms)]) after .flatten(), before toCsv().

See fromJsonToRows and toJson for the reference.

Record format

The record format puts \x1F (ASCII unit separator) between fields and \x1E (record separator) after each row, with no quoting. A field can hold anything else, tabs, quotes, and newlines included, and a program in any language reads it by splitting twice.

import { enumerate } from "@j50n/proc";
import { toRecord } from "@j50n/proc/transforms";

const notes = [
  ["1", "first line\nsecond line"],
  ["2", "a\ttab"],
];

// awk splits records on \036 (\x1E) and fields on \037 (\x1F).
const summary = await enumerate(notes)
  .transform(toRecord())
  .run(
    "awk",
    'BEGIN { RS = "\\036"; FS = "\\037" } { print $1 ": " length($2) }',
  )
  .lines
  .collect();

console.log(summary);

[ "1: 22", "2: 5" ]

toRecord() writes it and fromRecordToRows() reads it. Reach for it when you hand rows to another program, or take them from one, and the fields might hold characters that would break TSV. Unlike CSV, the reading side needs no parser: split the input on \x1E, dropping the empty piece after the last one, then each record on \x1F. In awk, set RS and FS as above. The separators are exported as RECORD_SEPARATOR and FIELD_SEPARATOR.

Reading

import { run } from "@j50n/proc";
import { fromRecordToRows } from "@j50n/proc/transforms";

// printf writes \037 between fields and \036 after each record.
const rows = await run(
  "printf",
  "Ada\\037line one\\nline two\\036Grace\\037\\036",
)
  .transform(fromRecordToRows())
  .flatten()
  .collect();

console.log(rows);

// A newline after the last record, as echo adds, is a record of its own.
const withNewline = await run("sh", "-c", "printf 'a\\037b\\036'; echo")
  .transform(fromRecordToRows())
  .flatten()
  .collect();

console.log(withNewline);

[ [ "Ada", "line one\nline two" ], [ "Grace", "" ] ]
[ [ "a", "b" ], [ "\n" ] ]

The input is split on \x1E, then each record on \x1F. Nothing else is special. Text after the last \x1E is a final record, so a record needs no \x1E at the end of the input. Every piece between separators counts:

• An empty record (\x1E\x1E) reads as [""].
• A newline after the last \x1E, as echo or print adds, reads as a row ["\n"]. Make the writer leave it off, or filter that row out.

A UTF-8 byte order mark at the start is dropped. Batches close at about 128 KiB of text. fromRecordToLazyRows() yields string-backed LazyRows and does the same work, so it is no faster; use it only when the code downstream expects LazyRows.

Writing

toRecord() refuses a field holding \x1E or \x1F, throwing an Error such as Invalid character (field separator) in record data at row 2, field 2; rows before it have already been written. It refuses a lone surrogate too, which UTF-8 can’t hold, and a first field starting with U+FEFF, which fromRecordToRows() would drop as a byte order mark. Any other text is written as it is.

Unlike TSV, the record format keeps a row holding one empty field: [""] is written as \x1E and reads back as [""]. A row with no fields, [] inside a batch, would be written the same way, so toRecord() refuses it: Invalid row (no fields) in record data at row 3.

The record format and flatdata

The flatdata CLI converts CSV and TSV to records in a separate process (flatdata csv2record), so a program in another language can read CSV without a CSV parser. It writes with toRecord(), so a CSV field holding \x1E or \x1F stops it with the same error and exit code 1, rather than coming out as extra records or fields.

See toRecord and fromRecordToRows for the reference.

LazyRow

A LazyRow is a row that decodes a field only when you read it. Use one when a pipeline filters on a field or two, or reads a few fields of wide rows.

import { read } from "@j50n/proc";
import { fromCsvToLazyRows, toCsv } from "@j50n/proc/transforms";

await read("data-orders.csv")
  .transform(fromCsvToLazyRows())
  .flatten()
  .drop(1) // the header
  .filter((row) => !row.fieldEquals(3, "1")) // compares bytes; decodes nothing
  .map((row) => {
    const fields = row.toStringArray(); // a LazyRow can't change; its array can
    fields[1] = fields[1].toUpperCase();
    return fields;
  })
  .transform(toCsv())
  .toStdout();

1,ADA,"Widget, large",2
3,LINUS,Bolt,10

fromCsvToLazyRows() and fromTsvToLazyRows() yield rows that are views of the bytes the WebAssembly parser produced. getField(i) decodes field i alone, and fieldEquals(i, value) compares field i with value byte by byte, without making a string, so the filter above decodes nothing. It is the fastest way to filter: about three times the speed of fromCsvToRows(), as How fast shows.

Methods

A LazyRow can’t be changed. To change a row, take toStringArray() and change the array, as the example does; the writers take plain rows as well. Every row writer (toCsv, toTsv, toRecord) takes LazyRows, singly or in batches, mixed with plain rows if you like.

When it doesn’t help

fromRecordToLazyRows() splits every record into strings first and wraps the array, so it does all the work fromRecordToRows() does. If you read every field anyway, toStringArray() decodes them all, and a plain Row is simpler.

Traps

• A row from CSV or TSV keeps its whole batch alive, the rows of about 128 KiB of input. To hold on to a few rows from a large stream, keep their toStringArray() instead.
• Invalid UTF-8 throws a TypeError when it is decoded, not from the parser, so a bad field nobody reads goes unnoticed. toStringArray() decodes the row’s whole batch at once, so it throws if any row in the batch is bad.

See LazyRow for the reference.

The flatdata CLI

flatdata converts between CSV, TSV, and the record format on the command line. Put it in front of a program, and the program reads simple records instead of CSV, with the parsing done in another process.

cat data.csv | flatdata csv2record | ./process | flatdata record2csv > out.csv

Install it with Deno:

deno install -g --allow-read --allow-write -n flatdata jsr:@j50n/proc@0.29.0/flatdata

The permissions are for -i and -o; without them it still works on stdin and stdout. To run it without installing, deno run jsr:@j50n/proc@0.29.0/flatdata csv2tsv < data.csv. flatdata --help lists the commands, and flatdata <command> --help a command’s options.

Commands

Commands are named <from>2<to>. Each reads stdin and writes stdout, or -i <file> and -o <file>.

--crlf ends rows with CRLF. The commands are the library’s transforms behind a command line: csv2tsv is csvToTsv(), csv2record is fromCsvToRows() into toRecord(), and so on. So they read and refuse what the library does, its parsers read flatdata’s output, and its writers make flatdata’s input.

In a pipeline

import { read } from "@j50n/proc";
import { fromRecordToRows } from "@j50n/proc/transforms";

// Installed, the command is just "flatdata". Here it runs from the package.
const flatdata = import.meta.resolve("@j50n/proc/flatdata");

await read("data-orders.csv")
  .run("deno", "run", flatdata, "csv2record") // parses in the child process
  .transform(fromRecordToRows())
  .flatten()
  .drop(1) // the header
  .map(([id, customer, item]) => `${id} ${customer}: ${item}`)
  .toStdout();

1 Ada: Widget, large
2 Grace: Gear "XL"
3 Linus: Bolt

With flatdata installed, write .run("flatdata", "csv2record"). The parsing then runs in its own process, alongside your program instead of in it. Records end in \x1E, not newlines, so read them with fromRecordToRows(), not .lines.

Errors

A field the output format can’t hold stops the conversion, as the library’s writers do: a tab, CR, or LF for TSV, \x1E or \x1F for the record format. So does a row the output would lose, a CR in CSV or TSV input that isn’t part of a CRLF, and a CSV quote still open at the end of the input. flatdata prints the message to stderr as one line naming the row and field, as in flatdata: Invalid character (tab) in TSV data at row 2, field 1, and exits with code 1, which proc turns into an ExitCodeError:

import { enumerate, ExitCodeError } from "@j50n/proc";

const flatdata = import.meta.resolve("@j50n/proc/flatdata");
const csv = (text: string) => enumerate([new TextEncoder().encode(text)]);

// A tab inside a quoted CSV field: TSV can't hold it, so csv2tsv fails. It
// writes one line to stderr, naming the row and field, and exits with 1.
try {
  await csv('a,b\n"c\td",e\n')
    .run(
      {
        fnStderr: (stderr) => stderr.lines.collect(),
        fnError: (error, stderrLines) => {
          if (error instanceof ExitCodeError) {
            console.log(`exit code ${error.code}: ${stderrLines?.join("; ")}`);
          }
          if (error) throw error;
        },
      },
      "deno",
      "run",
      flatdata,
      "csv2tsv",
    )
    .lines
    .forEach((line) => console.log(JSON.stringify(line)));
} catch (error) {
  if (!(error instanceof ExitCodeError)) throw error;
}

exit code 1: flatdata: Invalid character (tab) in TSV data at row 2, field 1

On stdout, output written before the error stays, and can end partway through a row; here there was none. -o is safer: flatdata writes a new file beside it and renames it into place only once the conversion has succeeded, so after an error the old file is as it was. That also lets -o name the file it reads, by -i or by <, to convert a file in place. -d must be one ASCII character other than ", CR, or LF.

When the reader of its output goes away, as with | head, flatdata stops and exits with 0, as other commands do.

Invalid UTF-8 stops the commands that make rows (csv2record, tsv2record, record2csv, record2tsv) with a TypeError. csv2tsv and tsv2csv copy bytes without decoding them, so they pass it through unchanged.

Replacing a shell script

The task: a nightly bash script that writes a manifest of the day’s CSV exports (rows, refunds, a checksum) and then compresses them. Here it is:

#!/bin/bash
# List the day's CSV exports in a manifest, then compress them.
set -euo pipefail

dir=recipes-exports
: > manifest.txt
for f in "$dir"/*.csv; do
  name=$(basename "$f")
  rows=$(tail -n +2 "$f" | wc -l)
  refunds=$(grep -c ',refunded$' "$f" || true)
  sum=$(sha256sum "$f" | cut -c 1-12)
  echo "$name rows=$rows refunds=$refunds sha256=$sum" >> manifest.txt
  gzip -f "$f"
done
cat manifest.txt

And the same script with proc:

import { type Enumerable, read, run } from "@j50n/proc";

/** Like `$(...)`: all the output as one string, without the last newline. */
async function text(output: Enumerable<Uint8Array>): Promise<string> {
  return (await output.lines.collect()).join("\n");
}

const dir = "recipes-exports";
const files: string[] = [];
for await (const entry of Deno.readDir(dir)) {
  if (entry.name.endsWith(".csv")) files.push(entry.name);
}
files.sort(); // a glob is sorted; readDir is not

const manifest: string[] = [];
for (const name of files) {
  const f = `${dir}/${name}`;
  const rows = await read(f).lines.drop(1).count();
  const refunds = await read(f).lines.count((l) => l.endsWith(",refunded"));
  const sum = await text(run("sha256sum", f).run("cut", "-c", "1-12"));
  manifest.push(`${name} rows=${rows} refunds=${refunds} sha256=${sum}`);
  await run("gzip", "-f", f).collect();
}

await Deno.writeTextFile("manifest.txt", manifest.join("\n") + "\n");
await read("manifest.txt").toStdout(); // cat manifest.txt

orders-east.csv rows=4 refunds=1 sha256=4d08cbb7e44c
orders-north.csv rows=3 refunds=2 sha256=7d787de6f357
orders-west.csv rows=2 refunds=0 sha256=a6ce666dd7b3

Run it with only what it touches:

deno run --allow-read=recipes-exports,manifest.txt --allow-write=manifest.txt \
  --allow-run=sha256sum,cut,gzip manifest.ts

What maps to what

Some steps are simpler in TypeScript than as a command. tail -n +2 | wc -l became read(f).lines.drop(1).count(), and the refund count became a count() with a test, which removed the bash script’s most fragile line: grep -c exits with code 1 when it counts zero, so under set -e it needs || true, and forgetting that kills the script on the first file with no refunds. When you do keep grep, Searching logs shows how to let “no match” through.

When a command fails

Nothing catches errors in the script, so a failure ends it the way set -e does, with exit code 1. The difference is what you see. If sha256sum can’t read a file:

sha256sum: missing.csv: No such file or directory
error: Uncaught (in promise) UpstreamError: sha256sum exited with code 1
    ...
Caused by: ExitCodeError: sha256sum exited with code 1

The first line is sha256sum’s own stderr, which goes to your terminal as it would from bash. cut succeeded, so the error at the end is an UpstreamError from cut, and its cause is the ExitCodeError of the command that failed. To handle failures rather than stop, put a try around the await; see Errors.

Traps when translating

• A command you don’t consume isn’t checked. run("gzip", f) on its own starts gzip, and your script waits for it to exit, but nothing reads its exit code, so a failure passes silently. Always await a consumer: await run("gzip", f).collect().
• .lines.first is not $(...). It stops reading after one line, and stopping early skips the exit-code check, so a command that prints a line and then fails gives you the line and no error. Collect the output, as text() does, when the exit code matters.
• There is no glob expansion. run("ls", "*.csv") passes the text *.csv to ls, which fails because no file has that name. List the files in TypeScript, or run a real shell for the parts that need one: run("sh", "-c", "ls *.csv"). Never build a sh -c string from data you didn’t write; that is where quoting bugs and injection come from.
• Deno.readDir is unsorted. A glob is sorted by name; sort the list yourself if order matters.

Searching logs

The task: find the database errors in an application’s logs, where the current file is plain text and the rotated ones are gzipped (app.log, app.log.1.gz, app.log.2.gz, as logrotate leaves them).

import { enumerate, read } from "@j50n/proc";

/** The lines of a log file, unzipping it first if it ends in `.gz`. */
function logLines(path: string) {
  const bytes = read(path);
  return path.endsWith(".gz")
    ? bytes.transform(new DecompressionStream("gzip")).lines
    : bytes.lines;
}

// app.log, app.log.1.gz, app.log.2.gz, ...
const dir = "recipes-logs";
const files: string[] = [];
for await (const entry of Deno.readDir(dir)) {
  if (entry.name.startsWith("app.log")) files.push(`${dir}/${entry.name}`);
}
// Oldest first: app.log.2.gz, app.log.1.gz, app.log (a plain sort would put
// app.log.10.gz before app.log.2.gz).
const age = (path: string) => Number(path.match(/\.(\d+)\.gz$/)?.[1] ?? 0);
files.sort((a, b) => age(b) - age(a));

await enumerate(files)
  .flatMap(logLines)
  .filter((line) => line.includes(" ERROR db:"))
  .toStdout();

2026-10-03 08:14:02 ERROR db: connection refused
2026-10-03 17:42:30 ERROR db: query timeout after 30s
2026-10-04 06:31:41 ERROR db: query timeout after 30s
2026-10-04 13:48:09 ERROR db: connection refused
2026-10-05 09:01:13 ERROR db: connection refused

logLines() is the one piece of logic: a .gz file goes through DecompressionStream before .lines, a plain one doesn’t. flatMap reads the files one after another as a single stream of lines, so memory stays flat however big the logs are, and toStdout() prints each match as it is found. Deno.readDir lists files in no particular order, hence the sort.

The script needs --allow-read=recipes-logs and nothing else: no command is run.

Summarizing

Counting is a forEach into a Map. Slicing fixed columns is the fastest way to pick a line apart when the format is fixed; use a regular expression when it isn’t.

import { enumerate, read } from "@j50n/proc";

const files = ["app.log.2.gz", "app.log.1.gz", "app.log"]
  .map((name) => `recipes-logs/${name}`);

const perDay = new Map<string, number>();
const perSource = new Map<string, number>();

await enumerate(files)
  .flatMap((path) =>
    path.endsWith(".gz")
      ? read(path).transform(new DecompressionStream("gzip")).lines
      : read(path).lines
  )
  .filter((line) => line.slice(20, 25) === "ERROR")
  .forEach((line) => {
    // 2026-10-05 09:01:13 ERROR db: connection refused
    const day = line.slice(0, 10);
    const source = line.slice(26).split(":")[0];
    perDay.set(day, (perDay.get(day) ?? 0) + 1);
    perSource.set(source, (perSource.get(source) ?? 0) + 1);
  });

console.log("errors per day:", Object.fromEntries(perDay));
const ranked = [...perSource].sort(([, a], [, b]) => b - a);
console.log("by source:", ranked.map(([s, n]) => `${s} ${n}`).join(", "));

errors per day: { "2026-10-03": 3, "2026-10-04": 4, "2026-10-05": 2 }
by source: db 5, http 3, disk 1

Timestamps in this format sort as strings, so a time window is a string comparison: .filter((line) => line >= "2026-10-04 06:00" && line < "2026-10-04 14:00").

Letting grep do the searching

For gigabytes of logs, grep finds matches faster than a JavaScript filter, and zgrep reads plain and gzipped files alike. The trap is that grep exits with code 1 when nothing matches, which proc, like set -e, treats as a failure. An fnError handler that lets exit code 1 through turns “no matches” back into “no lines”:

import { ExitCodeError, run } from "@j50n/proc";

/** grep and zgrep exit 1 when nothing matches; treat that as no lines. */
function noMatchIsFine(error?: Error) {
  if (error instanceof ExitCodeError && error.code === 1) return;
  if (error) throw error;
}

const logs = ["app.log.2.gz", "app.log.1.gz", "app.log"]
  .map((name) => `recipes-logs/${name}`);

for (const pattern of ["timeout", "segfault"]) {
  const hits = await run(
    { fnError: noMatchIsFine },
    "zgrep",
    "-h",
    pattern,
    ...logs,
  ).lines.collect();
  console.log(`${pattern}: ${hits.length}`, hits);
}

timeout: 2 [
  "2026-10-03 17:42:30 ERROR db: query timeout after 30s",
  "2026-10-04 06:31:41 ERROR db: query timeout after 30s"
]
segfault: 0 []

Exit code 2 (a missing file, a bad pattern) still throws ExitCodeError. Errors has more on fnError.

The same applies to grep in the middle of a pipeline: read(path).run({ fnError: noMatchIsFine }, "grep", "ERROR").

Variations

• Following a log as it grows: read() stops at the end of the file. For tail -f, run it: await run("tail", "-F", "app.log").lines.forEach(...) goes on until you stop it.
• Many large files at once: wrap the per-file work in concurrentMap, one file per call, and combine the counts at the end.
• The first match only: .find((line) => ...) stops reading at the match and closes the file.
• Structured logs (one JSON object per line): .transform(jsonParse) after .lines, or see JSON lines.
• Writing the matches to a file: end with .writeTo(path) instead of .toStdout(); it writes each line with a newline, as toStdout does.

Running jobs in parallel

The task: check a directory of nightly database backups with gzip -t, four at a time, and report every broken one, not just the first.

import { enumerate, run } from "@j50n/proc";

const dir = "recipes-backups";
const files: string[] = [];
for await (const entry of Deno.readDir(dir)) {
  if (entry.name.endsWith(".gz")) files.push(`${dir}/${entry.name}`);
}

/** Test one archive. A failure is returned, not thrown, so the rest go on. */
async function check(file: string) {
  try {
    await run(
      {
        fnStderr: (stderr) => stderr.lines.collect(),
        fnError: (error, stderr) => {
          if (error) throw new Error(stderr?.join(" ").trim() || error.message);
        },
      },
      "gzip",
      "-t",
      file,
    ).collect();
    return { file, problem: undefined };
  } catch (error) {
    return {
      file,
      problem: error instanceof Error ? error.message : `${error}`,
    };
  }
}

const results = await enumerate(files)
  .concurrentUnorderedMap(check, { concurrency: 4 })
  .collect();

const failed = results.filter((r) => r.problem !== undefined)
  .sort((a, b) => a.file.localeCompare(b.file));

console.log(`${results.length - failed.length} of ${results.length} are good`);
for (const { problem } of failed) console.log(problem);

4 of 6 are good
gzip: recipes-backups/db-2026-10-04.sql.gz: unexpected end of file
gzip: recipes-backups/db-2026-10-06.sql.gz: not in gzip format

Three pieces do the work:

• concurrentUnorderedMap(check, { concurrency: 4 }) keeps up to four calls of check running, and starts the next file as soon as any one finishes. Results come out in the order they finish, which is why the report sorts them. Without concurrency, it runs as many at once as the machine has CPUs.
• check returns its failure instead of throwing. That is what lets the other jobs carry on. The try sits inside the job, around the one command.
• fnStderr and fnError capture what gzip wrote to stderr and put it in the error, so the report says why each file failed. Without them, each child’s stderr goes straight to your terminal, interleaved with the others, and the error says only gzip exited with code 1.

The script needs --allow-read=recipes-backups --allow-run=gzip.

To make the script itself fail when any job did, as a CI step or a cron job should, end it with if (failed.length > 0) Deno.exitCode = 1;.

Why the try goes inside

This is a fragment, the same loop without the try:

await enumerate(files)
  .concurrentUnorderedMap((file) => run("gzip", "-t", file).collect())
  .collect();

The first job that fails throws its ExitCodeError out of the await, and the loop is over. The jobs already running carry on to the end in the background, with no one waiting for them; the files not yet started are never checked. That is the behavior you want when one failure makes the rest pointless (a build whose first step broke). For a batch where each job stands alone, catch inside.

Ordered results

concurrentMap takes the same arguments and yields results in input order. The cost is that a slow job holds back the results behind it, and while it does, fewer than concurrency jobs run. Use it when you print as you go and the order matters; otherwise concurrentUnorderedMap keeps every slot busy. Doing work concurrently covers both.

Variations

• A time limit per job: put timeout in front of the command, run("timeout", "60", "gzip", "-t", file). A job that runs over is stopped and exits with code 124, an ExitCodeError like any other failure.
• Keeping each job’s output: return await run(...).lines.collect() from the job along with the file name. It is all held in memory until the end, so for large output, write each job’s output to its own file with .writeTo().
• Jobs that are not commands: the same shape works for any async function, such as fetch calls. concurrency is the number of requests in flight.
• Showing progress: log from inside check when it finishes, or use .forEach() instead of .collect() to handle each result as it arrives.

Counting words

The task: the ten most common words in War and Peace, read from a gzipped copy (1.1 MB compressed, 3.2 MB of text). First with the classic shell pipeline, run through proc:

import { read } from "@j50n/proc";

// zcat book.gz | tr -cs A-Za-z '\n' | tr A-Z a-z | sort | uniq -c | sort -rn | head
const top = await read("recipes-warandpeace.txt.gz")
  .transform(new DecompressionStream("gzip"))
  .run("tr", "-cs", "A-Za-z", "\n") // anything but a letter ends a word
  .run("tr", "A-Z", "a-z")
  .run("sort")
  .run("uniq", "-c")
  .run("sort", "-rn")
  .lines
  .take(10)
  .collect();

console.log(top.join("\n"));

  34549 the
  22230 and
  16677 to
  14891 of
  10525 a
  10004 he
   8979 in
   8190 that
   7984 his
   7363 was

read() gives the file’s bytes, DecompressionStream unzips them, and each .run() is one |. The commands run at the same time, as they do in a shell, and proc hands each one’s output to the next as bytes, without decoding it. .take(10) does the job of head: once it has ten lines it closes the pipeline, the last sort dies of SIGPIPE on its next write, and that is not an error (stopping early). If tr or sort failed, the await would throw.

Arguments go to each program as they are, with no shell in between, so "\n" is a real newline character and nothing needs quoting.

The same in TypeScript

import { read } from "@j50n/proc";

const counts = new Map<string, number>();

await read("recipes-warandpeace.txt.gz")
  .transform(new DecompressionStream("gzip"))
  .lines
  .flatMap((line) => line.toLowerCase().match(/[a-z]+/g) ?? [])
  .forEach((word) => {
    counts.set(word, (counts.get(word) ?? 0) + 1);
  });

const top = [...counts].sort(([, a], [, b]) => b - a).slice(0, 10);
for (const [word, count] of top) {
  console.log(`${String(count).padStart(7)} ${word}`);
}

  34549 the
  22230 and
  16677 to
  14891 of
  10525 a
  10004 he
   8979 in
   8190 that
   7984 his
   7363 was

Same answer, no child processes, and no --allow-run. flatMap turns each line into its words, and forEach counts them into a Map. The callback has braces because forEach wants a function that returns nothing, and counts.set() returns the map, which fails the type check.

Making it faster

Every step in a pipeline costs an await per item, and this book has about 580,000 words. Handling lines in arrays, with plain loops inside, cuts that to one await per chunk of input:

import { read } from "@j50n/proc";

const counts = new Map<string, number>();

await read("recipes-warandpeace.txt.gz")
  .transform(new DecompressionStream("gzip"))
  .chunkedLines // arrays of lines: one await per chunk, not one per word
  .forEach((lines) => {
    for (const line of lines) {
      for (const word of line.toLowerCase().match(/[a-z]+/g) ?? []) {
        counts.set(word, (counts.get(word) ?? 0) + 1);
      }
    }
  });

const top = [...counts].sort(([, a], [, b]) => b - a).slice(0, 3);
console.log(top);

[ [ "the", 34549 ], [ "and", 22230 ], [ "to", 16677 ] ]

Measured on one laptop, this ran two to three times faster than the flatMap version, and faster than the shell pipeline. That one spends most of its time in sort, comparing by the rules of your locale; run({ env: { LC_ALL: "C" } }, "sort") compares bytes instead and was about four times faster. Reach for .chunkedLines when a pipeline does very little per item and there are a great many items; for most scripts, .lines is fast enough.

Getting the words right

The shell version has a flaw that is easy to miss. tr works on bytes, so every letter outside A-Z is a word break:

const line = "Anna Pávlovna said, “Don’t tease!”".toLowerCase();

console.log(line.match(/[a-z]+/g)?.join(" ")); // what tr -cs A-Za-z sees
console.log(line.match(/\p{L}+(?:’\p{L}+)*/gu)?.join(" ")); // any script

anna p vlovna said don t tease
anna pávlovna said don’t tease

The book is full of names like Rostóv and Bolkónski, and the shell pipeline counts their pieces as words. A regular expression with \p{L} (any letter, with the u flag) gets them right; swap it into either TypeScript version. This is the usual reason to move a text pipeline from the shell into code: not speed, but control over what counts as a word.

Variations

• A plain file: drop the .transform(...) line.
• Many files: run the count for each with concurrentMap and add up the maps.
• Total words only: .flatMap(...).count().
• A file too big for memory: the pipeline streams, but the Map holds every distinct word. For a word count that is small; for distinct values from a huge log it may not be, and the shell’s sort (which spills to disk) is the better tool.

API at a glance

Everything @j50n/proc and @j50n/proc/transforms export, grouped by what you would use it for, one line each. Each name links to its full entry on JSR. Key ideas explains how the pieces fit.

import { enumerate, read, run } from "@j50n/proc";
import { fromCsvToRows, toTsv } from "@j50n/proc/transforms";

Running commands

• run(...cmd): start a command now; returns its stdout to read. Options go first: run({ cwd, env }, "ls").
• .run(...cmd): pipe the items (lines or bytes) into a command’s stdin; returns its output.
• ProcessEnumerable: what run() returns; an Enumerable of stdout bytes, plus .pid and .status.
• .status: a promise of the exit status, without throwing on failure. A getter.
• .pid: the child’s process ID. A getter.
• ProcessOptions: cwd, env, fnStderr, fnError, buffer.
• StderrHandler: the type of fnStderr, which reads the child’s stderr instead of letting it reach the terminal.
• ErrorHandler: the type of fnError, which decides what a failure throws, or suppresses it.
• Cmd: a command and its arguments, [program, ...args].
• Process: the low-level child process under run(); use it to write stdin as you go, or to choose how each stream is connected.
• ProcessStreamOptions: options for new Process: ProcessOptions plus stdin, stdout, stderr.
• PipeKinds: "piped", "inherit", or "null".

Errors

• ProcessError: base class of the four below; catch it to handle any process failure.
• ExitCodeError: a command exited non-zero; .code, .command.
• SignalError: a command was killed by a signal; .signal, .command.
• TimeoutError: a command ran past its timeoutMs and was stopped; .timeoutMs, .command.
• UpstreamError: a command succeeded but its input failed; .cause is the original error.

Shutting down

• main(program, { timeoutMs }): run your program, and however it ends, stop the children and wait for them before exiting. For containers and services.
• terminateAll({ signal, timeoutMs }): signal every child proc started and wait for them, without exiting.

Sources

• enumerate(iterable): wrap an array, Set, generator, stream, or any (async) iterable as an Enumerable.
• read(path): a file’s bytes, opened when reading starts.
• readLines(path): a UTF-8 file’s lines; the same as read(path).lines.
• range({ from, to | until, step }): a sequence of numbers.
• RangeToOptions, RangeUntilOptions: the two shapes of range’s options (to excludes the end, until includes it).
• WritableIterable: a queue you write() to from callbacks and read with for await. No backpressure: writes never wait.
• Writable: something with write() and close(): a WritableIterable or a process’s stdin.

Enumerable

Enumerable is the async sequence every source returns. Steps return a new Enumerable; consumers return a promise. Read it once.

Steps:

• map(fn): transform each item; fn may be async.
• filter(fn), filterNot(fn): keep, or drop, the items fn accepts.
• flatMap(fn): map each item to an iterable and yield its items.
• flatten(): yield the items of each item; turns parser batches into rows.
• enum(): number the items as [item, index].
• concurrentMap(fn, { concurrency }): map with several calls at once; results in input order.
• concurrentUnorderedMap(fn, { concurrency }): map with several calls at once; results as they finish.
• ConcurrentOptions: concurrency, default navigator.hardwareConcurrency.
• transform(fn | stream): pass the whole sequence through a transformer function or a TransformStream, such as DecompressionStream or fromCsvToRows().
• take(n): the first n items, then close the source.
• drop(n): skip the first n items.
• concat(other): these items, then other’s.
• zip(other): pair items by position.
• unzip(): split pairs into two Enumerables (holds items in memory, as tee does).
• tee(n): split into n copies that each see every item; keeps items in memory until all have read them.
• .lines: decode bytes into lines of text. A getter.
• .chunkedLines: the same lines in arrays, one per chunk; faster for many short lines. A getter.
• run(...cmd): pipe into a command (see above). Starts at the call, unlike the other steps.

Consumers:

• collect(), toArray(): every item, in an array.
• forEach(fn): call fn on each item, waiting for each.
• reduce(fn, zero): fold the items into one value.
• count(fn?): how many items, or how many pass fn.
• find(fn): the first item fn accepts, or undefined.
• some(fn), every(fn): whether any, or all, items pass fn.
• .first: a promise of the first item; RangeError if there is none. A getter.
• writeTo(path | stream | writable): write to a file (bytes, or strings as lines; { atomic: true } replaces it only once everything is written), a WritableStream, or a Writable.
• writeBytesTo(writer): write bytes to a Writer & Closer such as a Deno.FsFile, then close it.
• toStdout(): write lines or bytes to stdout.
• for await (const item of e): iterate it yourself.

Transformers

Functions to pass to .transform().

• toLines, toChunkedLines: bytes to lines, or arrays of lines; what .lines and .chunkedLines use.
• toByteLines: split bytes into lines without decoding, for data that isn’t UTF-8.
• toBytes: lines of text (or bytes) to byte chunks, a newline after each string; use before writeTo or a CompressionStream.
• buffer(size): join small byte chunks into chunks of at least size bytes.
• gzip, gunzip: compress and decompress; for plain bytes, CompressionStream and DecompressionStream do the same.
• jsonParse: parse each line as JSON; a blank line throws.
• jsonStringify: each item to a line of JSON.
• debug: log each item as it passes (to stdout), unchanged.
• transformerFromTransformStream(stream): wrap a TransformStream as a transformer function.
• TransformerFunction: the type of a transformer, (AsyncIterable<T>) => AsyncIterable<U>; an async function* is the usual way to write one.
• TransformStream: the { writable, readable } pair .transform() also accepts.
• StandardData: what toBytes and a process’s stdin accept: string, string[], Uint8Array, Uint8Array[].

toBufferSource is deprecated; use toBytes.

Utilities

• sleep(ms): wait.
• SECONDS, MINUTES, HOURS, DAYS, WEEKS: milliseconds, for sleep(2 * SECONDS) or a cache timeout.
• cache(key, value, { timeout, refresh }): keep a computed value in Deno KV across runs; needs --unstable-kv.
• fetchRecord(key): read the raw cache entry, expired or not, for debugging.
• concat(arrays): join byte arrays into one.
• concatLines(arrays): join byte arrays with a newline after each.
• writeAll(data, writer): write all the bytes to a Writer, such as Deno.stdout.
• shuffle(array): shuffle in place.
• isString(value): type guard for a string primitive.

Helper types

These name the result types of some methods; you rarely write them.

• Lines, ChunkedLines, ByteSink, Run: what .lines, .chunkedLines, writeBytesTo, and .run() return; never when the items are the wrong type, so the mistake fails to type-check.
• ElementType: the item type of an iterable; what flatten() yields.
• Unzip: what unzip() returns.
• Tuple, TupleOf: what tee(n) returns.

Data formats: @j50n/proc/transforms

Parsers take bytes and yield batches (arrays of rows); add .flatten() for one row at a time. Writers take rows or batches and yield bytes. See Data formats.

• fromCsvToRows(options): parse CSV into batches of string[] rows.
• fromCsvToLazyRows(options): parse CSV into batches of LazyRows, which decode a field only when read.
• toCsv(options): write rows as CSV, quoting where needed.
• csvToTsv(options), tsvToCsv(options): convert CSV to TSV and back, bytes to bytes, without making rows.
• CsvParseOptions, CsvStringifyOptions: separator, and crlf for writing.
• fromTsvToRows(), fromTsvToLazyRows(): parse TSV.
• toTsv(): write TSV; throws on a tab, CR, or LF in a field.
• fromJsonToRows(options): parse JSON lines into batches of values, optionally checked by a schema.
• toJson(): write one value per item as JSON lines; throws a TypeError on an item with no JSON form.
• JsonOptions: schema and sampleSize.
• ZodSchema: anything with a parse(value) method, a Zod schema included.
• fromRecordToRows(), fromRecordToLazyRows(): parse the record format.
• toRecord(): write the record format, for handing rows to another program.
• FIELD_SEPARATOR, RECORD_SEPARATOR: "\x1F" and "\x1E", the record format’s separators.
• Row: one row, a string[].
• LazyRow: a row that decodes fields on demand: getField(i), fieldEquals(i, value), columnCount, toStringArray(), and LazyRow.fromStringArray(fields) to make one.
• BATCH_SIZE_BYTES: about 128 KiB, how much input makes a batch.

The package also has a command-line converter, jsr:@j50n/proc/flatdata; see The flatdata CLI.

Common mistakes

The traps that actually bite, by what you see: the symptom, why it happens, and the fix. Error messages are what Deno prints; deno check errors appear in your editor too.

The program hangs

Nothing reads a command’s output. A child’s stdout is a pipe that holds about 64 KB. A child that writes more waits until someone reads, so this fragment never finishes:

const p = run("seq", "1", "100000");
await p.status; // waits for an exit that never comes

Small output fits in the pipe, which is why the same code “works” in a test and hangs on real data. Fix: end every command’s pipeline with a consumer, await run(...).lines.collect(), or .forEach(() => {}) to throw the output away. .status is for after, or alongside, reading. The same goes for fnStderr: read stderr to the end, or a child that writes a lot to it blocks.

It finished its work, but doesn’t exit. A command it stopped reading early is still running, such as a server whose “ready” line .first returned, and Deno doesn’t exit while a child runs. Wrap the program in main(), which stops the children on the way out, or stop that one yourself with Deno.kill(p.pid).

A WritableIterable is never closed. The reader waits for more items after the last one. Call close() when the data ends, and close(error) when it fails.

It throws

ExitCodeError: grep exited with code 1. grep exits 1 when nothing matches, and proc treats every non-zero exit as a failure. So do diff (files differ), cmp, and test. Fix: an fnError handler that lets code 1 through, as in Searching logs, or catch ExitCodeError and check .code.

RangeError: .first: the sequence is empty. .first on a sequence with no items: a command that printed nothing and succeeded (one that failed throws its ExitCodeError instead), or a filter that matched nothing. .first never resolves to undefined. When the output may be empty, take an array of at most one:

import { run } from "@j50n/proc";

// find succeeds and prints nothing when no file matches.
const find = ["find", ".", "-name", "*.bak"] as const;

try {
  console.log(await run(...find).lines.first);
} catch (error) {
  console.log(String(error));
}

const [path] = await run(...find).lines.take(1).collect();
console.log(path ?? "no backups");

RangeError: .first: the sequence is empty
no backups

NotFound: Failed to spawn 'ls -la': entity not found. The whole command was passed as one string, so Deno looked for a program named ls -la. Pass each argument separately: run("ls", "-la"). The same error, with a real name, means the program isn’t installed or isn’t on PATH.

NotCapable: Requires run access to "grep", run again with the --allow-run flag. Deno’s permissions. Grant what the script uses: --allow-run=grep, --allow-read=./logs, --allow-write=out.txt.

TypeError: cache needs Deno KV from cache(). cache uses Deno KV, which is unstable. Run with --unstable-kv, or add "unstable": ["kv"] to deno.json.

BrokenPipe: Broken pipe (os error 32) when you pipe the script’s output into head or less and quit early, from a write of your own to Deno.stdout. toStdout() and writeTo(Deno.stdout.writable) stop quietly instead, and console.log ignores it. Use one of those, or catch it: if (!(error instanceof Deno.errors.BrokenPipe)) throw error;.

SyntaxError: Unexpected end of JSON input from jsonParse. A blank line is not JSON. Filter blank lines out before it, or use fromJsonToRows() from @j50n/proc/transforms, which skips them.

The error says exited with code 1 and nothing else. The command’s own explanation went to stderr, which is your terminal by default. To put it in the error, capture it with fnStderr and rethrow from fnError; see Errors.

It doesn’t type-check

This expression is not callable because it is a 'get' accessor. Did you mean to use it without '()'? (TS6234). .lines, .chunkedLines, .first, .status, and .pid are properties: await run("ls").lines.first, not .lines().first(). Without type checking, the same mistake fails at run time with TypeError: run(...).lines is not a function.

This comparison appears to be unintentional because the types 'string[]' and 'string' have no overlap. (TS2367), on a row of parsed data. The parsers yield batches of rows, so without .flatten() each item is an array of rows, not a row. Where the types don’t catch it, the counts are wrong:

import { read } from "@j50n/proc";
import { fromCsvToRows } from "@j50n/proc/transforms";

const batches = await read("people.csv").transform(fromCsvToRows()).count();
const rows = await read("people.csv").transform(fromCsvToRows()).flatten()
  .count();

console.log({ batches, rows });

{ batches: 1, rows: 3 }

Add .flatten() after the parser. toJson() takes one value per item, so flatten before it too: a batch left whole is written as one JSON array.

'error' is of type 'unknown'. (TS18046), on error.code in a catch. Narrow first: if (error instanceof ExitCodeError) console.log(error.code); else throw error;.

Module '".../mod.ts"' has no exported member 'fromCsvToRows'. The data transforms are a separate entry point: import them from "@j50n/proc/transforms".

Property 'collect' does not exist on type 'never'. .lines and .chunkedLines need bytes, and .run() needs strings or bytes; on other items the type is never. Strings are lines already, so drop the .lines; before a .run(), turn numbers or objects into strings with .map(String) or .transform(jsonStringify).

Wrong results, no error

The file came out empty. writeTo(path) empties the file before anything is read, so a pipeline that reads the same file finds nothing. Pass { atomic: true }, which writes a new file and renames it over the old one; see Files.

The second pass finds nothing. An Enumerable is used once; a second collect() on it, or on another chain built from it, yields nothing and throws nothing. Collect into an array first, or split with tee(). See Key ideas.

A failed command didn’t throw. Stopping early (.first, take, find, break) closes the pipeline without checking the exit code:

import { run } from "@j50n/proc";

const cmd = ["sh", "-c", "echo partial; exit 3"] as const;

console.log(await run(...cmd).lines.first); // no error: the exit code is unread

try {
  await run(...cmd).lines.collect(); // reads to the end, so it checks
} catch (error) {
  console.log(String(error));
}

partial
ExitCodeError: sh exited with code 3

Read the output to the end when the exit code matters. A command started with run() and never consumed isn’t checked either: the script waits for it to exit, then ignores how.

A result is a Promise { <pending> }. Every consumer (collect, forEach, count, first, …) returns a promise; await it. An un-awaited forEach may not have run when the next line does.

Memory keeps growing with a WritableIterable. write() returns at once, without waiting for the reader; there is no backpressure:

import { WritableIterable } from "@j50n/proc";

const queue = new WritableIterable<number>();

for (let n = 1; n <= 3; n++) {
  await queue.write(n); // resolves at once; nothing has read it
  console.log(`wrote ${n}`);
}
await queue.close();

for await (const n of queue) console.log(`read ${n}`);

wrote 1
wrote 2
wrote 3
read 1
read 2
read 3

A producer that outruns its reader fills memory. When the source can wait (a file, a socket, a paginated API), don’t push into a queue; pull from it with an async generator, or enumerate() a ReadableStream, so it is read only as fast as you consume it.

It’s slow

A pipeline of many short lines crawls, especially one that feeds a command with .run(): every step, and every write to the command, happens once per line. Use .chunkedLines instead of .lines, and work on each array: .chunkedLines.map((lines) => lines.filter(...)).run("sort"). On two million lines that took a filter between two commands from 16 seconds to 0.3. See Lots of lines.

Children die without cleaning up

In a container, stopping it sends SIGTERM to Deno, Deno exits at once, and the container’s end kills the children before their cleanup runs: temporary files are left behind, uploads cut off, locks held. On any host, an uncaught error does the same: Deno kills its children as it exits. Wrap the program in main(), which passes the signal on and waits for the children (30 seconds by default) before exiting. See Shutting down cleanly.

Contributing

These pages are for working on proc itself: where the code is, how it fits together, how it is tested, and how a release goes out. To use proc, start at Install and run a first script instead.

What’s where

deno publish sends JSR every file that .gitignore and the "publish" "exclude" list in deno.json don’t leave out: meant to be mod.ts, src/, scripts/flatdata/, the README, and the license. A stray file or directory in the working tree goes too, so check deno publish --dry-run before a release.

Getting set up

You need Deno. That is enough to change the TypeScript, run the tests, and type-check everything. Two more tools are needed only for some jobs:

• Swift 6.3.2 with its Embedded WebAssembly SDK (swift-6.3.2-RELEASE_wasm-embedded), to rebuild the WebAssembly module after a change under swift/.
• mdBook (installed with Rust’s cargo), to build the site.

The day-to-day loop, from the repository root:

deno fmt
deno lint
deno check mod.ts src/transforms/mod.ts
deno test --allow-read --allow-write=/tmp/ --allow-run ./tests

./build.sh does all of that and more (it also rebuilds the WASM and updates the toolchain and dependencies); Building and releasing says what each script does.

Pages in this section

• Architecture: the modules, and how a pipeline, its errors, and shutdown work inside.
• Coding standards: the conventions the code and the public API follow.
• Documentation guidelines: who the docs are for, and how doc comments and book pages are written.
• Testing: the test suites, including the one that runs every example in this book.
• Building and releasing: the build scripts, the WASM build, and the release.
• The CSV reader: what the WebAssembly module reads, how, and how it is tested.

Bugs and questions go to GitHub issues.

Architecture

proc is a few small layers over two things Deno already has: async iterables and Deno.Command. This page follows a pipeline down through them.

The modules

mod.ts re-exports enumerable.ts, run.ts, process.ts, transformers.ts, writable-iterable.ts, utility.ts and cache.ts, and main and terminateAll from shutdown.ts. The rest are internal, and transforms/ is its own export, @j50n/proc/transforms.

Enumerable

Enumerable<T> wraps one AsyncIterable<T>. Each step (map, filter, take, …) is an async generator over the one before, returned in a new Enumerable, so nothing runs until a consumer pulls. Errors need no special handling: an exception thrown anywhere in the chain unwinds through the generators to the consumer’s await. Stopping early works the same way in reverse: when a consumer stops, return() is called down the chain, and each generator’s finally closes what it holds.

Because the wrapped iterable is read once, so is the Enumerable. A second pass gets an exhausted generator, which yields nothing.

lines, run, writeBytesTo, and the rest of the methods that only make sense for some item types use conditional types (Lines<T>, Run<S, T>, …) that resolve to never for the wrong items, so the mistake fails to type-check instead of failing at run time.

A command

run(...cmd) parses its arguments (an options object may come first) and builds a Process, with stdin "null", stdout "piped", and stderr "inherit", or "piped" when there is an fnStderr. The Process constructor calls Deno.Command(...).spawn() at once, which is why a missing program throws from run() itself, and registers the child with shutdown.ts. run() returns a ProcessEnumerable, an Enumerable over Process.stdout.

Process.stdout is where failures become errors. It is an async generator that yields the child’s stdout chunks, which a ChildOutput reads; after the last one, it awaits the exit status and the fnStderr promise, and failureOf in failure.ts picks the error: TimeoutError if timeoutMs ran out, whatever the exit; else SignalError or ExitCodeError for a failed exit; else UpstreamError if the child succeeded but its input failed. If there is an fnError, settle hands the error to it instead, and whatever it throws is what the consumer sees. Since all of this happens after the last chunk, the consumer has every line before the error arrives. If the consumer stops early, the generator is closed before the check, so no exit code is examined.

ChildOutput also reads the output when the child exits and nobody has started reading it, into memory, so that a child run only for its status doesn’t keep its pipe open; a reader who comes later gets it from there. After a timeout, it ends the output at the first pause once the child has exited, since a program the child started may hold the pipe open for good.

A pipeline

.run(...cmd) on an Enumerable builds a Process with stdin "piped" and calls writeToStdin(this) straight away. That starts a loop that reads the upstream Enumerable and writes each item to the child’s stdin (through toBytes, and buffer when buffer: true), without waiting for anyone to read the output. So .run() is the one step that is not lazy.

When the upstream iterable throws, writeToStdin keeps the error as the process’s “pass error” and closes stdin. The child sees end of input, finishes, and exits; Process.stdout then attaches the pass error as cause to its own ExitCodeError, SignalError or TimeoutError, or throws an UpstreamError around it if the child exited 0. That is how an error from the first command in a chain reaches the catch after the last. A child that exits before reading all its input (head -1) makes the next write fail with BrokenPipe; the loop ignores that error and stops reading upstream, which closes it like any early stop.

Concurrency

concurrentMap and concurrentUnorderedMap in src/concurrent.ts keep up to concurrency promises from the mapping function in flight. The ordered one yields from the front of its buffer, so a slow item holds back the rest; the unordered one yields whichever settles first. Neither cancels calls already started when the consumer stops or an error is thrown.

Shutdown

shutdown.ts keeps a set of the children proc started, adding each in the Process constructor and removing it once it has exited and its fnStderr, if any, has finished. It also keeps the children’s outputs being read. main() installs listeners for SIGTERM, SIGINT, and SIGHUP (only SIGINT on Windows) and for uncaught errors and unhandled rejections, then runs the program. However it ends, the first ending wins. On SIGTERM, or when the program ends, it signals every child in the set and waits up to timeoutMs for them. After SIGINT or SIGHUP, which the terminal already sent the children, it gives them a second, then sends SIGTERM to any still running. After a signal, it then waits for the children’s output to be read and up to half a second more for the program, so a pipeline writing that output finishes. Then it calls Deno.exit with the code for that ending. A second signal exits at once. terminateAll() is the signal-and-wait half on its own. An unload listener sends SIGTERM to the children if the program exits some other way, though it can’t wait for them.

Small pieces

• WritableIterable is an unbounded queue of promises: write() resolves the promise at the tail and adds a new one, and the iterator awaits them in order. It has no backpressure by design; it is for push-style sources that can’t wait anyway. A process’s stdin is one.
• helpers.handled(promise) attaches an empty catch to a promise that will be awaited later, so that a rejection before then isn’t reported as unhandled. It is used wherever proc holds a promise while doing something else: the concurrent maps hold many, writeTo overlaps a write with reading the next item, zip pulls both sides at once, and the fnStderr reader runs beside stdout.
• helpers.abandon(source) closes a source nothing will read, for a step that fails before it reads (a file that can’t be opened, a bad concurrency). An async generator ignores return() until it has started, so it starts the source, takes at most one item, and stops.
• helpers.closeAll(sources, pending) closes sources a consumer stopped reading, in the background when a pull is still pending, since the close would wait behind it.
• cache() stores { timestamp, value } under the key in Deno KV’s default database, with KV’s expireIn set from the timeout, and checks the age on read.

Data formats and WebAssembly

src/transforms/ holds one file per format. The CSV and TSV parsers, and csvToTsv() and tsvToCsv(), run in a WebAssembly module; src/wasm/flatdata.ts wraps it, compiling it once and making an instance per stream. The reader hands back whole rows a batch at a time, with the end of every field, and common.ts and lazy-row.ts turn a batch into string rows or LazyRows. The record and JSON-lines parsers are TypeScript: split-text.ts splits the input on a separator, as it does lines for .lines, and they group rows into batches of about 128 KiB. The writers are TypeScript too: each builds one string per item and encodes it once, which is faster than crossing into WebAssembly when the strings already live in JavaScript.

The module is built from swift/ (Embedded Swift, with SIMD classifiers in C) into wasm/flatdata.wasm. tools/embed-wasm.ts writes it, base64-encoded, into src/wasm/flatdata-wasm.ts, because a package loaded from JSR has https: module URLs and can’t read a .wasm file beside it. The module has no imports. The flatdata CLI in scripts/flatdata/ runs the same transforms. See Building and releasing for the build, and The CSV reader for how the module reads.

Coding standards

The code should read like it was written by someone who cared, and the public API should be predictable enough that a reader, or an LLM, guesses right about a method it hasn’t seen. These are the conventions that make that so.

Tooling

• deno fmt, deno lint, and deno check pass on everything, with Deno’s default rules. That includes no any; use unknown and narrow.
• Dependencies in deno.json are pinned to exact versions, so nothing moves until someone runs deno update --latest on purpose (build.sh does).
• Every doc comment example type-checks: deno check --doc-only mod.ts src/. How to write doc comments is in Documentation guidelines.

Names and files

• Files in src/ are kebab-case (writable-iterable.ts); tests end in .test.ts.
• Classes are PascalCase (ProcessEnumerable), functions and variables camelCase (concurrentMap), constants SCREAMING_SNAKE_CASE (HOURS, FIELD_SEPARATOR).
• A name says what the thing is or does. enumerate() wraps an iterable; .enum() numbers its items. Two names that close need the doc comments to point at each other, as those do.

The shape of the public API

Options come first. run({ cwd }, "ls"), .run({ fnStderr }, "grep", "x"). The command and its arguments are the rest parameters, each argument its own string, with no shell parsing.

A member that names a value is a getter; one that does something is a method. .lines, .chunkedLines, .first, .status, and .pid take no parentheses; collect(), map(), and count() do. Add a getter only for something with no arguments that reads as a noun, and say “A getter” in its doc comment, because .first() is the most common mistake people make with the library.

Steps are lazy, consumers return promises. A method that returns an Enumerable does nothing until something pulls; a method that returns a promise is a consumer. .run() is the documented exception, because it starts a process. Keep it the only one.

Wrong calls fail to type-check. Members that only work on some item types use conditional types that resolve to never otherwise (Lines<T>, Run<S, T>, ByteSink<T>), so .lines on numbers is a compile error, not a run-time surprise.

Errors arrive at the consumer, after the data. A failure throws from the await that consumes the pipeline, after every item before it has been delivered. Process failures are subclasses of ProcessError. When one error leads to another, set cause; never swallow an error, and let an error from a user’s callback arrive unchanged where possible.

Strings are lines. Where text becomes bytes (toBytes, toStdout(), a process’s stdin), each string gets a "\n"; bytes pass as they are. So code using the library shouldn’t append "\n" to strings (the output gets blank lines) or call toBytes before toStdout() (it does that itself).

Transformers plug into .transform(). A transformer is a function from one async iterable to another. Ones with no configuration are passed as they are (.transform(toBytes), .transform(gunzip)); ones with configuration are factories that return one (buffer(size)). In @j50n/proc/transforms, every parser and writer is a factory, with or without options (fromTsvToRows(), toCsv({ crlf: true })), so they all read the same way.

Say what happens, not what we wish. If something holds memory, doesn’t wait, or skips a check, the doc comment says so (tee buffers, WritableIterable.write() doesn’t wait, stopping early skips the exit-code check). A design that is simpler but has a cost is fine; a hidden cost is not.

Inside the code

• Don’t leave a promise unobserved. When the code starts work now and awaits it later, wrap the promise in handled() from src/helpers.ts, or a rejection in between becomes an unhandled-rejection crash.
• Close what you open on every path. Iterators get return() when a consumer stops early; generator finally blocks are where files, writers, and stdin get closed. A test for early stopping goes with any new source.
• Comments say why. History and benchmarks belong in the book or a commit message. Internal helpers get a short comment only where the code doesn’t speak for itself.

Tests and examples

• A new feature comes with tests named for the claim they prove (“take(n) ends after the nth item, without waiting for another.”), and a bug fix with a test in tests/regressions/ that failed before the fix.
• Example data uses placeholders, never real names, addresses, or keys.

See Testing for where tests go and how the book’s examples are checked.

Documentation Guidelines

Who reads these docs

Two readers, and the docs have to work for both:

• A developer who just found proc. In a minute they should know what it is for, and in five they should have it working in their project.
• An LLM helping someone use proc. It reads the JSR page, the doc comments, and the book, often only part of them, and then writes code. It learns the library’s idioms only from what we wrote, so a claim that is false or vague turns straight into broken code in someone’s project.

Everything below follows from those two readers.

What makes a page or a comment good

It says what the thing is for before how it works. Lead with the purpose and when to reach for it. A reader who knows why a function exists can work out most of the details; a reader with only the details misuses it.

It is true, and checked. The code is the arbiter. Before writing a claim about behavior, read the code, and when the code doesn’t settle it, run a probe. Vague phrases (“handles errors properly”, “manages resources automatically”) hide whether a claim is true; say what actually happens (“throws ExitCodeError after the last line of output”).

It names the traps. Where people go wrong is the most useful thing to write down: what happens if they skip a step, and what they see when they do. “If you don’t read stdout, a child that writes more than the pipe holds blocks forever, and your program hangs” is worth more than “always consume output.”

It is short. Say a thing once, in the place a reader would look for it, and link to it from elsewhere. No preambles, no summaries of what was just said, no “Key points” lists that repeat the prose.

It sounds like a person who knows the library. Plain, direct, specific. No marketing (“superpowers”, “blazing”, “seamless”), no emoji, no comparisons with Node streams or other libraries unless a reader needs one to choose.

Doc comments

Doc comments are the API reference: they appear in deno doc, on the JSR pages, and in editors, and an LLM often sees nothing else.

• The first sentence says what it does, in plain words. It is the summary shown in lists, so it stands alone.
• Then, where it applies: why you would use it, what to watch for, and how it fails (what it throws, and when).
• Name parameters and return values with @param and @returns when the signature doesn’t already make them obvious. Don’t restate the type.
• One or two @example blocks, each minimal and complete: it imports what it uses from @j50n/proc (or @j50n/proc/transforms) and type-checks.
• History, design derivations, and benchmarks don’t go in comments. Put them in the book if they matter.
• Internal and private members get a short comment only where the code doesn’t explain itself.

src/shutdown.ts is a good model.

The book

The book (site/src) is for learning the library and for the topics that cross many functions: pipelines, errors, shutdown, data formats. It doesn’t repeat the API reference; it links to it.

• Each page answers one question a reader has (“How do I pipe one command into another?”), and its title says which.
• Examples come first and carry the explanation. Prose fills in what the code can’t show.
• Code that runs lives in site/examples/ as a file, and the page pulls it in with {{#include ../../examples/name.ts}}. tests/book_examples.test.ts checks and runs every file there, so an example can’t drift from the code.
• A fragment that can’t stand alone (a deliberate mistake, a sketch) is marked as such in the prose.

The README and mod.ts

mod.ts’s module comment is the JSR landing page. It and the README say the same short thing: what proc is for, a handful of examples, the few concepts everyone needs, and a link to the book. Keep them in step.

Checking your work

deno check --doc-only mod.ts src/   # every doc comment example type-checks
deno test -A tests/book_examples.test.ts   # every book example runs

Testing

Everything in tests/ runs with one command from the repository root:

deno test --allow-read --allow-write=/tmp/ --allow-run ./tests

It takes under a minute. build.sh runs the same tests with --allow-run narrowed to the commands they use. The tests run real commands (sh, cat, grep, sort, gzip, git, …), so they expect a Unix system with the usual tools.

What’s where

tests/mdbook_examples.test.ts and tests/tutorial.test.ts test examples from the previous edition of the book. tutorial.test.ts reads this repository’s git history, so it needs a clone, not a source download.

Name a test for the claim it proves, as a sentence: "take(n) ends after the nth item, without waiting for another." A failing test then says what broke. For a bug, write the test first and watch it fail.

The book’s examples

Every code block in this book that runs is a file under site/examples/, pulled into its page with {{#include}}. tests/book_examples.test.ts does two things with them:

1. Type-checks all of them at once, in strict mode, against the repository’s deno.json, so "@j50n/proc" resolves to the local code.
2. Runs each one in a fresh temporary directory holding a copy of site/examples/fixtures/, and compares what it prints on stdout with the .out file beside it, which the page also shows. An example without a .out file only has to exit 0.

So a page can’t show output the code doesn’t produce, and a change to the library that alters an example’s behavior fails the test.

To write or change an example, run it the same way the test does:

deno run -A tools/book-example.ts site/examples/recipes/logs-search.ts

Read what it prints. When it is what the page should show, save it as the .out file:

deno run -A tools/book-example.ts --save site/examples/recipes/logs-search.ts

The rules that keep these tests reliable:

• Deterministic output. No times, PIDs, absolute paths, random numbers, or results in completion order (sort them first).
• No network. Use local commands and files in fixtures/.
• Fixtures are shared. Every example sees every fixture. Name a new one after its section (recipes-exports/, data-sales.csv), and don’t change one another page relies on.
• Narrow before use in a catch (if (error instanceof ExitCodeError)), since the examples are type-checked strictly.

A fragment that can’t run, such as a deliberate mistake or code that would hang, stays inline in the Markdown, and the prose says it is a fragment.

Doc comment examples

The @example blocks in src/ are what JSR and editors show. They aren’t run (many touch files or the network), but they must type-check:

deno check --doc-only mod.ts src/

The WASM module

The module is tested from TypeScript, through the committed build, so the tests run without Swift installed. tests/transforms/reference.ts is an independent TypeScript CSV and TSV reader and writer, and readers.test.ts and convert.test.ts compare the module with it on edge cases at many chunk sizes, plus fields of several megabytes. The CSV reader says more.

Benchmarks

deno bench --allow-read --allow-write=/tmp/ --allow-run ./tests/comprehensive_benchmarks.test.ts

build.sh runs this one. The other benchmark files, and the studies in benchmarks/, are run by hand when performance is the question.

Building and releasing

Two scripts at the repository root do the building: build.sh checks and tests the library, and build-site.sh builds this book into docs/. The release itself is a short manual process, written down in MAINT.md.

build.sh

In order, it:

1. Updates the toolchain: rustup update, then cargo install mdbook.
2. Rebuilds the WebAssembly module with ./swift/build.sh if anything under swift/ is newer than wasm/flatdata.wasm, and embeds it with tools/embed-wasm.ts (below).
3. Updates every dependency to its latest version with deno update --latest. The versions in deno.json are pinned exactly, so this is the only thing that moves them; review the diff.
4. Formats the Markdown and TypeScript with deno fmt.
5. Runs deno lint and deno check on the TypeScript.
6. Runs the tests in ./tests with --allow-run limited to the commands they use, and then the benchmarks in tests/comprehensive_benchmarks.test.ts.

It stops at the first failure (set -e).

The WebAssembly module

The CSV and TSV readers and the CSV-TSV converters run in WebAssembly built from Embedded Swift in swift/. swift/build.sh compiles the package in release mode with the Embedded WebAssembly SDK (SWIFT_SDK names another) and copies the result to wasm/flatdata.wasm. swift/Package.swift explains the compiler flags: SIMD on for the Swift code too, and no stack protector, which keeps the module free of imports. tools/embed-wasm.ts then writes the file, base64-encoded, into src/wasm/flatdata-wasm.ts, which is what the library loads: a package installed from JSR has https: module URLs and can’t read a .wasm file next to it.

Both the .wasm and the generated .ts are committed, so Swift is needed only when something under swift/ changes, and build.sh skips the build otherwise: the compiler embeds a hash that differs on every build, so rebuilding unchanged sources would still change the bytes. To see what the compiler made of a loop, disassemble the module with llvm-objdump -d wasm/flatdata.wasm. If the two files ever disagree, tests/packaging.test.ts fails and says to run tools/embed-wasm.ts.

build-site.sh

1. Updates rustup and mdBook, as build.sh does.
2. Generates the HTML API docs with deno doc --html into site/src/api-docs/.
3. Formats the book’s Markdown and TypeScript.
4. Builds the book with mdbook build, using site/book.toml. Two preprocessors run: site/gitv.ts fills in the version placeholder (gitv in double braces) with the latest git tag, and tools/mdbook-deno-script-preprocessor.ts runs <script> blocks in pages (see its .md file).
5. Replaces docs/ with the built site. GitHub Pages serves docs/ from main.
6. Writes llms.txt (an index of the pages) and llms-full.txt (the whole book, examples and output included) into docs/ with tools/llms-txt.ts, for LLMs, which read those more reliably than HTML.

Because the version shown in the book comes from the latest tag, build the site after tagging a release, not before. Commit docs/ on its own: the generated files make a noisy diff that shouldn’t hide source changes.

Releasing

MAINT.md has the commands. In outline: set the version in deno.json (JSR takes it from there), commit, tag that commit with the same version and push the tag, run deno publish, build and commit the site, and finally import the new version in a clean directory to check the published package, since the tests only ever see local files.

The CSV reader

The CSV and TSV parsers, csvToTsv(), tsvToCsv(), and the flatdata CLI all run on one WebAssembly module, built from Embedded Swift in swift/. It is about 12 KB and has no imports. This page is the map; the doc comments in swift/Sources/ carry the details, and src/wasm/flatdata.ts is the TypeScript side.

What it reads

TSV is the same reader with a tab separator and quoting off, so a quote is content and the CR rule holds everywhere. A CR that ends the input is an error too, which is why a CR-only file fails at its first line end instead of reading as one long row.

The lexer

CSVLexer (swift/Sources/CSV/CSVLexer.swift) is the state machine behind the reader and csvToTsv(). It finds the bytes that can matter (quote, LF, CR, the separator, and one extra byte a sink asks for, such as a tab for csvToTsv) 64 at a time: the SIMD classifiers in swift/Sources/SIMDKernels/ compare four 16-byte vectors and return a 64-bit mask. The lexer walks the set bits; the bytes between them go to the sink as one run. A scalar step on each candidate byte decides what it is, with four states: field start, unquoted, quoted, and quote in quoted. The SIMD finds candidates and the state machine decides, so a separator inside quotes is just content. Runs are copied 16 bytes at a time.

A CR outside quotes looks at the next byte. When the CR ends a chunk, the next chunk’s first byte decides. On a CR it refuses, the scan stops and the sink records the row and field, counted from 1, through StreamOperation’s refusal; a feed then returns -1, and src/wasm/flatdata.ts turns that into the Invalid character (CR) in CSV data at row N, field M error. Refusing is kept out of the hot loop, because carrying its state through every iteration cost 10 to 20%. A quote still open when the stream ends is refused the same way, as a different Refusal kind, at the row and field where it opened: no field or row ends inside quotes, so the sink is still there.

The three operations differ in what they write:

• CSVReader writes each row in record format (a field ends in 0x1F, the last of a row in 0x1E) and records where every field ends: byteEnds in the output bytes and textEnds in UTF-16 code units of those bytes decoded. The lexer counts UTF-8 continuation bytes and 4-byte leads with SIMD too, so textEnds costs no extra pass. Only whole rows go out: a row cut off by the end of a chunk moves to the front of the buffer for the next feed, so a row never straddles two batches, and a row longer than a chunk just grows the buffer.
• CSVToTSV writes fields with tabs and rows with LF, and refuses a tab, CR, or LF inside a field, which TSV can’t hold, and a row of one empty field, which would be a blank line. Its output is not held back.
• TSVToCSV has a simpler scanner of its own, on the same SIMD classifiers, since TSV has no quoting. It quotes a field only when the field holds the separator or a quote, which it knows only at the field’s end, so a field cut off by a chunk is carried to the next one.

From bytes to rows

readRows() copies the input into the module’s buffer in chunks of BATCH_SIZE_BYTES (128 KiB), however the source chunked it, and feeds each one. A batch is the complete rows that chunk finished, as views of WASM memory. batchRows() in src/transforms/common.ts decodes the whole batch with one TextDecoder call and slices fields out by textEnds; that is several times faster than decoding field by field. lazyRows() in lazy-row.ts copies the batch out of WASM memory and wraps each row, so getField decodes one field by byteEnds, and fieldEquals compares bytes without decoding.

Memory

Each stream gets its own module instance (the module is compiled once), and drops it when the stream ends. That is the only way memory is released: the allocator, swift/Sources/Arena/arena.c, is a bump arena whose free does nothing. It suffices because wasm memory can’t shrink anyway, and buffers grow by doubling, so what a stream abandons is less than what it uses. One instance per stream also means two streams read at once never share state.

What bounds the memory is the longest row, not the input: the reader holds a row whole, and TSVToCSV holds a field whole, in its input and again in its output, where it reserves the field’s size plus its quotes. The arena grows memory by just what an allocation needs, so a stream’s memory runs to about 2 times its longest row in the reader, plus the row’s field ends: two UInt32s a field, so a row of empty fields (,,,,) takes about 18 times its size (measured: 16 MiB of commas took 287 MiB, 64 MiB took 1,151 MiB). In tsvToCsv() it is 5 to 10 times its longest field, depending on where the doubling buffers land and how many quotes need doubling (measured: a field of 128 MiB took 896 MiB, one of 512 MiB took 2,561 MiB, and 64 MiB of " took 640 MiB). Before, the arena doubled memory too and TSVToCSV reserved three times the field, and the same 128 MiB took 1,536 MiB. csvToTsv() holds nothing back and runs in constant memory.

A wasm32 memory can’t pass 4 GiB. When an allocation can’t be made, the arena traps: the Embedded Swift runtime doesn’t check for a null pointer, and would write through it into the module’s own memory at address 0. Every operation sets currentRow before it grows a buffer, so after the trap the TypeScript side reads current_row and throws Row too large for the WebAssembly module's memory in CSV data at row N (for tsvToCsv(), Field too large ...), with the RuntimeError as its cause. In practice that is a row, or a tsvToCsv() field, of around a gigabyte, or a row of about 150 MB of empty fields. The exports return pointers as i32, which JavaScript reads as negative past 2 GiB, so src/wasm/flatdata.ts reads every pointer through address() (pointer >>> 0).

Tests

tests/transforms/reference.ts is a plain TypeScript reader and writer for CSV and TSV, written the slow, obvious way and sharing no code with the module. readers.test.ts and convert.test.ts check the module against it on a list of edge cases at chunk sizes of 1, 7, 63, 64, 65, and 1,000 bytes, which puts a chunk boundary at every position that matters: inside a CRLF, a doubled quote, a UTF-8 character, a SIMD block. They also cover fields of several megabytes, CR and unclosed-quote refusals with their row and field, a byte order mark split across chunks, invalid UTF-8, and streams read at the same time. round-trip.test.ts checks that what the writers write reads back the same.

Building and releasing covers the build.