# Pipelines
[‹ docs index](README.md)
`a | b | c` **without a shell**. Each stage's stdout feeds the next stage's
stdin through an in-process relay (a `tokio::io::copy` task per boundary) — there
is no shell string anywhere, so no quoting rules, no word splitting, no injection
surface. Each stage spawns into its **own** kill-on-drop
[process group](process-groups.md) sub-group, so a per-stage
[`Command::timeout`](#timeouts) tears down that stage's whole subtree
(grandchildren of a forking `sh -c …` included); a chain-wide teardown fans the
kill across every stage's sub-group, so the chain still lives and dies as a
unit. (The relay is an implementation detail, not a kernel splice: a producer
whose consumer exits early stops on a *broken pipe* when the relay's next write
fails, rather than instantly via `SIGPIPE`.)
- [Building and running](#building-and-running)
- [Semantics: pipefail and the ends](#semantics-pipefail-and-the-ends)
- [Unchecked stages](#unchecked-stages)
- [Timeouts](#timeouts)
- [Streaming a live chain](#streaming-a-live-chain)
- [Re-running a pipeline](#re-running-a-pipeline)
## Building and running
`Command::pipe(next)` starts a `Pipeline`; chain more stages with
`Pipeline::pipe`; drive it with `output_string()` or `run()`:
```rust,no_run
use processkit::Command;
#[tokio::main]
async fn main() -> processkit::Result<()> {
// git log --format=%an | sort | uniq -c
let authors = Command::new("git").args(["log", "--format=%an"])
.pipe(Command::new("sort"))
.pipe(Command::new("uniq").arg("-c"))
.run() // require every stage to succeed
.await?;
println!("{authors}");
Ok(())
}
```
The verbs mirror `Command`'s, each operating on the pipefail outcome:
| `output_string()` | `ProcessResult<String>` | …reported in the result (code/stderr/program of the first unclean stage) |
| `output_bytes()` | `ProcessResult<Vec<u8>>` | …same, with the last stage's stdout captured raw (binary pipes) |
| `run()` | trimmed final stdout | …raised as that stage's `ErrorReason::Exit`; fails loud on a truncated capture |
| `checked()` | full `ProcessResult<String>` | …raised as `ErrorReason::Exit` (untrimmed stdout) |
| `run_unit()` | `()` | …raised as `ErrorReason::Exit` (output discarded) |
| `exit_code()` | `i32` | …its attributed code (no code → `ErrorReason::Timeout`/`Signalled`) |
| `probe()` | `bool` | `0` → `true`, `1` → `false`, else `Err` |
| `parse(\|s\| …)` / `try_parse(\|s\| …)` | `T` | …raised as `ErrorReason::Exit`; fails loud on a truncated capture |
`Err` from `output_string` itself means a stage couldn't be *started or
driven* at all (spawn failure, broken plumbing) — never a mere non-zero exit.
The `first_line` probe is deliberately not a *buffering* pipeline verb: those
verbs consume the last stage in full to fold the pipefail outcome. To *capture*
the first matching line of a finished chain, add a `| head -n1` (Unix) / `grep
-m1` / `findstr` stage and capture. To instead **read a chain that keeps
running** — wait for a banner line, then stream the rest — use
[`Pipeline::start()`](#streaming-a-live-chain), which gives a chain the same live
streaming surface a single `Command::start()` gives a process.
The `|` operator is sugar for the same thing — `a | b | c` ≡
`a.pipe(b).pipe(c)`. Parenthesize the chain before a terminal verb, since
method calls bind tighter than `|`:
```rust,no_run
use processkit::Command;
#[tokio::main]
async fn main() -> processkit::Result<()> {
let authors = (Command::new("git").args(["log", "--format=%an"])
| Command::new("sort")
| Command::new("uniq").arg("-c"))
.run()
.await?;
Ok(())
}
```
## Semantics: pipefail and the ends
The outcome is **pipefail**, like `set -o pipefail` in a shell:
- `stdout` is always the **last** stage's output — that's what the chain
produced.
- `code`, `stderr`, and the reported program come from the **culprit** stage:
the leftmost stage that didn't exit cleanly (non-zero, signal-killed, or timed
out), but **preferring a real failure over a downstream `SIGPIPE` victim** — a
stage killed only because a later stage closed the pipe early. If every failure
is such a broken-pipe victim, the leftmost one wins; when every stage succeeded,
the last stage speaks.
```rust,no_run
use processkit::Command;
#[tokio::main]
async fn main() -> processkit::Result<()> {
let result = Command::new("cat").arg("data.txt")
.pipe(Command::new("grep").arg("ERROR")) // suppose grep exits 2 (bad pattern)
.pipe(Command::new("wc").arg("-l"))
.output_string()
.await?;
// Diagnostics point at grep — the first unclean stage — while stdout is
// whatever wc managed to print:
assert_eq!(result.code(), Some(2));
println!("blamed: {}", result.ensure_success().unwrap_err()); // names `grep`
Ok(())
}
```
**Failure tears the chain down proactively.** The moment a stage ends with a
checked failure (a non-zero exit outside its `ok_codes`, a signal kill, or its
own per-stage timeout), every stage's sub-group is torn down at once — the failure does
not wait to trickle out through closing pipes. This matters for a *quiet*
sibling that would otherwise hang: an upstream producer that never writes never
dies of a broken pipe, so under a purely passive teardown a downstream failure
could be held open indefinitely by that silent producer. Now the failure
surfaces immediately, and the killed siblings are treated as victims (like a
downstream `SIGPIPE` death) — the stage that actually failed keeps the blame.
The one death that does **not** trigger this is an
[`unchecked_in_pipe()`](#unchecked-stages) stage's: its unclean exit is forgiven,
so it leaves the rest of the chain running. (A stuck stage that never *fails* —
a healthy producer that simply never finishes — is still bounded only by
[`Pipeline::timeout`](#timeouts) or [cancellation](#timeouts).)
The ends of the chain behave like a single `Command`:
- The **first** stage's configured [`stdin`](commands.md#standard-input)
source is honored — feed the whole pipeline from a string, file, or stream.
- **Inner** stages read from the pipe, full stop: any `stdin` source or
`keep_stdin_open` configured on them is overridden.
- Inner stages' **stderr** is captured per-stage for pipefail diagnostics;
only the last stage's stdout reaches you.
```rust,no_run
use processkit::{Command, Stdin};
#[tokio::main]
async fn main() -> processkit::Result<()> {
let unique_count = Command::new("sort")
.stdin(Stdin::from_iter_lines(["b", "a", "b", "c"]))
.pipe(Command::new("uniq"))
.pipe(Command::new("wc").arg("-l"))
.run()
.await?;
assert_eq!(unique_count.trim(), "3");
Ok(())
}
```
## Unchecked stages
Strict pipefail has one classic false positive: a consumer that legitimately
stops reading early. In `producer | head -1` the consumer exits `0` after one
line and closes the pipe; the producer then stops on a **broken pipe** — its
next write fails once the relay's downstream is gone (a broken-pipe write error,
or `SIGPIPE` where the OS delivers it) — a perfectly normal death that strict
pipefail would blame the chain for. Mark that stage [`unchecked_in_pipe()`](https://docs.rs/processkit/latest/processkit/struct.Command.html#method.unchecked_in_pipe):
```rust,no_run
use processkit::Command;
#[tokio::main]
async fn main() -> processkit::Result<()> {
// seq 1 1000000 | head -1 — the producer's broken-pipe death is expected.
let first = (Command::new("seq").args(["1", "1000000"]).unchecked_in_pipe()
| Command::new("head").args(["-n", "1"]))
.run()
.await?;
assert_eq!(first.trim(), "1");
Ok(())
}
```
The rules (a design borrowed from `duct`'s `unchecked()` — the idea, not the
code):
- An unchecked stage's unclean exit — a non-zero code, a broken-pipe write
failure (or `SIGPIPE` where the OS delivers it) from a consumer that closed
early, or its own per-stage timeout kill — is **skipped** when the chain
decides what to report.
- A **checked** failure always trumps an unchecked one, regardless of
position: `unchecked` never shields another stage's real failure.
- A chain whose only failures are unchecked reports **success** with the last
stage's stdout and its **real** exit code preserved (not a fabricated `0` — the
accepted-code set is widened to include it). The carve-out is for an exit
*status* only: a last stage killed by a **signal** or its own **timeout** is not
a status to forgive and still surfaces as the failure.
- `unchecked` forgives exit *status* only — never a whole-chain
[`Pipeline::timeout`](#timeouts), and it has no effect on a `Command` run
outside a pipeline (a single run's status is already plain data in its
`ProcessResult`).
## Timeouts
Two scopes, deliberately distinct:
```rust,no_run
use processkit::Command;
use std::time::Duration;
#[tokio::main]
async fn main() -> processkit::Result<()> {
let out = Command::new("producer")
.timeout(Duration::from_secs(10)) // per-STAGE: kills just `producer`
.pipe(Command::new("consumer"))
.timeout(Duration::from_secs(30)) // whole-CHAIN: Pipeline::timeout
.output_string()
.await?;
Ok(())
}
```
- **`Pipeline::timeout`** bounds the whole chain: at the deadline every
stage's sub-group is torn down and the result reports `timed_out` (no
partial stdout — unlike a single command's captured timeout).
- A **per-stage `Command::timeout`** kills that stage's *whole subtree* — its
own sub-group, grandchildren of a forking `sh -c …` included, not just its
direct child. Every stage is evaluated by the same pipefail rule (D14): a
stage that hit its own deadline — inner *or* last — surfaces on `run()` as
that stage's `ErrorReason::Timeout`, reporting **that stage's own deadline** (not
the chain's, and never `0ns`).
Cancellation has two forms. **`Pipeline::cancel_on(token)`** is the chain-level
control: the token **gap-fills** into every stage that doesn't already carry its
own `Command::cancel_on` (an explicit per-stage token is left intact), so firing
it tears the whole chain down and the run resolves to `ErrorReason::Cancelled`. (A
`cancel_on` token on an individual stage `Command` also cancels that stage and
errors the pipeline, but
the pipeline-level builder is the clearer authority.) See
[Timeouts & cancellation](timeouts-and-cancellation.md).
## Streaming a live chain
The verbs above buffer the whole run. For a long-lived chain you read *from*
rather than wait *out* — `journalctl -f | grep ERROR`, `tail -F access.log | jq`
— `Pipeline::start()` returns a `PipelineSession`: the multi-stage analogue of
the [`RunningProcess`](streaming.md) a single `Command::start()` gives you. It
streams the **last** stage's stdout as it arrives while every inner stage drains
in the background, then folds the same pipefail outcome at `finish()`.
```rust,no_run
use processkit::prelude::StreamExt;
use processkit::{Command, Finished, Outcome};
#[tokio::main]
async fn main() -> processkit::Result<()> {
// journalctl -f | grep --line-buffered ERROR
let mut session = Command::new("journalctl").arg("-f")
.pipe(Command::new("grep").args(["--line-buffered", "ERROR"]))
.start()
.await?;
let mut lines = session.stdout_lines()?; // the *last* stage's stdout, live
while let Some(line) = lines.next().await {
println!("error: {line}");
// …break out when you've seen enough, then tear the chain down…
break;
}
drop(lines);
// Fold the pipefail outcome. The last stage's stdout was already streamed,
// so — like RunningProcess::finish — none is re-bundled; `outcome` and
// `stderr` come from the pipefail-attributed (culprit) stage.
session.start_kill()?; // stop the whole chain now
let Finished { outcome, stderr, .. } = session.finish().await?;
if outcome != Outcome::Exited(0) {
eprintln!("chain ended {outcome:?}: {stderr}");
}
Ok(())
}
```
The session mirrors `RunningProcess`:
- **`stdout_lines()` / `events()`** — the last stage's stdout (lines) or its full
lifecycle (`Started` → interleaved stdout+stderr → `Exited`), each
**consume-once** (a second take is a loud `Err`, never a silently-empty stream).
- **`wait_for_line(pred, within)`** — wait for a readiness banner on that stream
without tearing the chain down (an `ErrorReason::NotReady` on timeout, like the
single-process probe).
- **`finish()`** — the streaming analogue of `output_string()`: the
pipefail-attributed stage's `outcome` and *its own* `stderr` in a `Finished`
(no stdout — you already streamed it). A chain-wide `Pipeline::timeout` that
elapsed reports `Outcome::TimedOut`; a `Pipeline::cancel_on` that fired surfaces
as `ErrorReason::Cancelled` — exactly as in the buffering verbs.
- **`start_kill()`** and **kill-on-drop** — stop the whole chain now, or drop the
session and every stage's tree dies. The no-orphan invariant holds for a live
chain (including a partially-started one) just as it does for a single process.
Whole-chain teardown still applies while streaming: an inner stage's checked
failure proactively tears the chain down (so a quiet upstream can't hold a failed
live chain open), and the chain-wide [timeout / cancellation](#timeouts) bounds
the session regardless of which stage is slow.
## Re-running a pipeline
A `Pipeline` is `Clone` and re-runnable — stages are re-cloned per run. The
one caveat is inherited from `Command`: a **one-shot** stdin source on the
first stage (`Stdin::from_reader` / `from_lines`) is consumed by the first run;
re-running then **fails loud** (an `ErrorReason::Io` at launch, D10) rather than
silently feeding empty stdin. Use the reusable sources
(`from_string` / `from_bytes` / `from_iter_lines` / `from_file`) when a chain
runs more than once.
---
Next: [Timeouts, retries & cancellation](timeouts-and-cancellation.md) ·
[Running commands](commands.md) ·
[Process groups](process-groups.md)