Pipelines

Previous: Overview

Build a → b → c without a shell. Each stage's stdout is wired straight into the next stage's stdin by an in-process relay — there is no shell string anywhere, so there are no quoting rules, no word splitting, and no injection surface. Every stage spawns into one shared kill-on-dispose process group, so the whole chain lives and dies as a unit: tear the chain down (a timeout, a cancellation, an early return) and every stage goes with it. Each stage participates in the shared-group ownership rules detailed in Lifecycle state machine.

The relay is a copy loop, not a kernel splice. When a consumer exits early it closes the upstream read end, so the producer stops on a broken pipe — its next write fails once the relay's downstream is gone. On POSIX the OS may deliver that as SIGPIPE; Windows has no SIGPIPE, so there it surfaces as a failed write instead. See Unchecked stages for why that distinction matters.

The relay keeps read-side and write-side failures separate. A genuine failure while reading an upstream stage's stdout is reported as ProcessError.Io instead of being mistaken for EOF, including when the downstream stage observes a truncated stream. A downstream broken pipe, or an IOException/ObjectDisposedException caused by whole-chain teardown, remains a normal termination race and stays quiet.

Such a genuine relay failure also ends the run immediately: the chain is hard-killed and reaped the moment the failure is seen, rather than after every stage has exited on its own. Closing the pipe ends is not enough by itself — an upstream stage that stops writing and keeps running never earns a broken pipe — so waiting for its natural exit would stall the pipeline on a failure already diagnosed. Buffered verbs and a StartAsync session share that behaviour, and the reported error is the first relay failure seen. One consequence is worth knowing: because the whole chain is torn down at once, the final stage's output is truncated wherever the kill lands. A whole-chain timeout or cancellation that is already tearing the chain down still takes precedence — a relay exception raised into that teardown stays a quiet termination race, so the run reports TimedOut/Cancelled rather than ProcessError.Io.

The samples below run inside a task { } block and use match!; from C# the same surface is await-able fluent methods.

• Building a pipeline
• The verbs
• Pipefail: the result and the ends
• Fail-loud output overflow
• Unchecked stages
• Timeouts and cancellation
• Re-running a pipeline

Building a pipeline

There are two equivalent ways to wire stages together; pick whichever reads better in context. There is no | operator — F# reserves | for patterns and active patterns — so the fluent .Pipe method and the Pipeline module are the only two ways to build a chain.

The fluent way: Command.Pipe(next) turns a Command into a Pipeline, and chaining .Pipe(...) again appends another stage. Finish with a verb.

F#

task {
    // git log --format=%an | sort | uniq -c
    let pipeline =
        (Command.create "git" |> Command.args [ "log"; "--format=%an" ])
            .Pipe(Command.create "sort")
            .Pipe(Command.create "uniq" |> Command.arg "-c")

    match! pipeline.RunAsync() with
    | Ok authors -> printfn $"{authors}"
    | Error err -> eprintfn $"{err.Message}"
}

C#

// git log --format=%an | sort | uniq -c
var pipeline = new Command("git").Args(["log", "--format=%an"])
    .Pipe(new Command("sort"))
    .Pipe(new Command("uniq").Arg("-c"));

Console.WriteLine(await pipeline.RunAsync() switch
{
    { IsOk: true, ResultValue: var authors } => authors,
    { IsOk: false, ErrorValue: var err }    => err.Message,
});

The pipe-style module mirror builds the same value: Pipeline.create first second seeds a two-stage chain, and Pipeline.pipe next pipeline appends a stage — so it threads naturally through |>:

F#

task {
    let pipeline =
        Pipeline.create
            (Command.create "git" |> Command.args [ "log"; "--format=%an" ])
            (Command.create "sort")
        |> Pipeline.pipe (Command.create "uniq" |> Command.arg "-c")

    match! pipeline.OutputStringAsync() with
    | Ok result -> printfn $"{result.Stdout}"
    | Error err -> eprintfn $"{err.Message}"
}

C#

var pipeline = new Command("git").Args(["log", "--format=%an"])
    .Pipe(new Command("sort"))
    .Pipe(new Command("uniq").Arg("-c"));

Console.WriteLine(await pipeline.OutputStringAsync() switch
{
    { IsOk: true, ResultValue: var result } => result.Stdout,
    { IsOk: false, ErrorValue: var err }   => err.Message,
});

Pipeline.timeout and Pipeline.cancelOn round out the module mirror; they correspond to the fluent .Timeout and .CancelOn covered under Timeouts and cancellation. Building a pipeline spawns nothing — a Pipeline is an immutable value, and each builder call returns a new one. Nothing runs until you call a verb.

The verbs

A Pipeline finishes with the same verb vocabulary as a Command; each one folds the whole chain's outcome (see pipefail below) and returns Task<Result<_, ProcessError>>:

Every verb also accepts an optional CancellationToken — pipeline.RunAsync(token), pipeline.OutputStringAsync(token), and so on — for a per-call token alongside the chain-level CancelOn.

An Error from a capture verb such as OutputStringAsync means a stage couldn't be started or driven at all — a spawn failure, a not-found program, broken plumbing, or a genuine upstream relay read failure — never a mere non-zero exit. A non-zero exit is data in the ProcessResult.

The buffering verbs above run the whole chain to completion before returning. For a long-running or interactive pipeline whose final output you want to read line by line as it appears — a live handle to stream from, wait for a readiness line on, and stop on demand — start a streaming session with StartAsync instead (see Streaming a pipeline below). It is the pipeline analogue of a single command's StartAsync → RunningProcess.

Pipefail: the result and the ends

The outcome follows shell pipefail (set -o pipefail):

• stdout is always the last stage's output — that is what the chain produced.
• code, stderr, and the reported program come from the rightmost checked stage that did not finish successfully — a code outside its accepted OkCodes (just 0 unless widened), a signal kill, or a timeout. When no checked stage failed, they come from the real last stage; an unchecked last stage's voluntary exit is accepted without replacing its code.

So when an inner stage fails, the result's stdout is whatever the tail still printed, while the diagnostics point at the culprit:

F#

task {
    let pipeline =
        (Command.create "cat" |> Command.arg "data.txt")
            .Pipe(Command.create "grep" |> Command.arg "ERROR") // suppose grep exits 2 (bad regex)
            .Pipe(Command.create "wc" |> Command.arg "-l")

    match! pipeline.OutputStringAsync() with
    | Ok result ->
        // Blame points at grep — the rightmost unclean stage — while Stdout is whatever wc managed.
        printfn $"code={result.Code} program={result.Program} success={result.IsSuccess}"
        // code=Some 2  program=grep  success=false
    | Error err -> eprintfn $"{err.Message}"
}

C#

var pipeline = new Command("cat").Arg("data.txt")
    .Pipe(new Command("grep").Arg("ERROR")) // suppose grep exits 2 (bad regex)
    .Pipe(new Command("wc").Arg("-l"));

Console.WriteLine(await pipeline.OutputStringAsync() switch
{
    // Blame points at grep — the rightmost unclean stage — while Stdout is whatever wc managed.
    { IsOk: true, ResultValue: var result } => $"code={result.Code} program={result.Program} success={result.IsSuccess}", // code=Some 2  program=grep  success=false
    { IsOk: false, ErrorValue: var err }   => err.Message,
});

The success-requiring verbs turn that same pipefail outcome into a typed error attributed to the blamed stage. ProcessResult.ensureSuccess does it explicitly, and RunAsync does it for you:

F#

match! pipeline.RunAsync() with
| Ok out -> printfn $"{out}"
| Error(ProcessError.Exit(program, code, _, stderr)) ->
    eprintfn $"{program} exited {code}: {stderr}" // program = "grep", code = 2
| Error err -> eprintfn $"{err.Message}"

C#

Console.WriteLine(await pipeline.RunAsync() switch
{
    { IsOk: true, ResultValue: var output } => output,
    { IsOk: false, ErrorValue: ProcessError.Exit { Program: var p, Code: var c, Stderr: var s } } => $"{p} exited {c}: {s}", // program = "grep", code = 2
    { IsOk: false, ErrorValue: var err } => err.Message,
});

The two ends of the chain behave like a single Command:

• The first stage's configured Stdin source is honored — feed the whole pipeline from a string, bytes, a file, or a stream. A Stdin source on any later stage is a configuration error — that stage's stdin is always rewired to the previous stage's stdout — so .Pipe rejects it with an ArgumentException naming the offending stage.
• A KeepStdinOpen on any stage is a configuration error: a pipeline exposes no per-stage RunningProcess.TakeStdin handle, so the caller cannot write to a kept-open stdin pipe. .Pipe rejects it with an ArgumentException naming the offending stage; run the command separately if interactive stdin is needed.
• Every stage's stdout is wired into a pipe — feeding the next stage's stdin, or captured at the end — so a Stdout mode of Null / Inherit set on a stage is overridden to keep the chain connected.
• Every stage's stderr is captured per-stage for pipefail diagnostics under that stage's own OutputBuffer byte cap (MaxBytes + Overflow); only the last stage's stdout reaches you. A fail-loud (OverflowMode.Error) overflow is honoured on any stage's stderr, not only the final stdout — see Fail-loud output overflow below.
• Command.MergeStderr (a shell 2>&1) is allowed only on the last stage — its stdout is the pipeline's captured output, so merging there captures the final stage's combined stdout+stderr. On an earlier stage it is rejected (ArgumentException) the moment another stage is appended after it: the chain wires each stage's stdout into the next stage's stdin, so an OS-level merge on an intermediate stage would inject its stderr into the downstream stage's input data.
• A per-stage Timeout, Retry, or CancelOn is rejected when the stage is piped (see Timeouts and cancellation); a per-stage Logger or StreamBuffer has no effect inside a chain — observe or bound an individual command by running it on its own.
• A per-stage CapturePolicy is rejected on any stage: every capture a pipeline makes — the final stdout and each stage's stderr — is a raw byte capture, so that seam has no decoded line to shape. .Pipe throws an ArgumentException naming the stage rather than running the chain with the redaction hook inactive; run a command whose captured output must be scrubbed on its own.
• The last stage's OutputBuffer byte cap (MaxBytes + Overflow) bounds the captured stdout — the same way a single command's OutputBytesAsync does (Error -> OutputTooLarge, DropOldest/DropNewest -> a tail/head with Truncated set). Its MaxLines, and every intermediate stage's stdout buffer policy, do not apply (an intermediate stdout is plumbing into the next stage, not a capture). Each stage's stderr cap, by contrast, applies on every stage — see Fail-loud output overflow.
• A truncated capture is refused by the same verbs as for a single command: RunAsync and the ParseAsync/TryParseAsync/OutputJsonAsync verbs built on it fail with OutputTooLarge (quoting the last stage's byte ceiling) rather than presenting a tail or head as the chain's whole output, while OutputStringAsync/OutputBytesAsync hand it back with Truncated set and RunUnitAsync stays successful — see Which verb you use decides what a drop means. The refusal names the last stage and quotes that same stage's byte ceiling (the only ceiling that applies to the chain's captured stdout). Those never come apart: an earlier stage is the pipefail representative only when it is a checked failure, and the verb has then already failed with that stage's own Exit/Signalled/Timeout error before truncation is considered — so a refusal means no checked stage failed, which is exactly when the result belongs to the real last stage, UncheckedInPipe or not. What it refuses is therefore the last stage's own capture: its stdout, or the stderr published with it. A truncated stderr on any other stage is diagnostics the result never publishes and refuses nothing.

F#

task {
    let uniqueCount =
        (Command.create "sort" |> Command.stdin (Stdin.FromString "b\na\nb\nc\n"))
            .Pipe(Command.create "uniq")
            .Pipe(Command.create "wc" |> Command.arg "-l")

    match! uniqueCount.RunAsync() with
    | Ok n -> printfn $"{n}" // "3"
    | Error err -> eprintfn $"{err.Message}"
}

C#

var uniqueCount = new Command("sort").Stdin(Stdin.FromString("b\na\nb\nc\n"))
    .Pipe(new Command("uniq"))
    .Pipe(new Command("wc").Arg("-l"));

Console.WriteLine(await uniqueCount.RunAsync() switch
{
    { IsOk: true, ResultValue: var n }    => n, // "3"
    { IsOk: false, ErrorValue: var err } => err.Message,
});

Fail-loud output overflow

Every captured stream in a chain obeys its own stage's OutputBuffer byte cap (MaxBytes + Overflow) — not only the final stdout. Two kinds of capture are subject to a cap:

• the last stage's stdout — the pipeline's captured output, and
• every stage's stderr — drained per-stage for diagnostics, bounded so a chatty stage can never exhaust memory regardless of its position in the chain.

Under OverflowMode.Error a cap is fail-loud: once the stream exceeds it, the verb returns ProcessError.OutputTooLarge — naming the offending stage's program and its configured caps — exactly like a single command's byte capture, and consistently whether the overflow is on the final stdout or on any stage's stderr (an intermediate stage's stderr fail-loud overflow is no longer silently dropped). Under DropOldest / DropNewest the same overflow stays lossy but non-erroring. The result's Truncated combines final-stage stdout truncation with truncation of the representative stage's published stderr; truncation of an unselected stage's stderr remains local to diagnostics the result does not publish.

When more than one stream overflows at once — several stages, and/or a stage's stderr together with the final stdout — one deterministic error is chosen by first-offending-stage-in-pipeline-order: the leftmost stage in the chain wins (the earliest point the chain overflowed), and within a single stage its captured stdout (only the last stage has one) is preferred over its stderr. So an overflow on an earlier stage's stderr outranks the final stdout's, while the pre-existing "only the final stdout overflowed" case is reported exactly as before (same program, limits, and totals).

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 dies on a broken pipe — its next write fails once the relay's downstream is gone (a failed write on Windows, possibly delivered as SIGPIPE on POSIX). That is a perfectly normal death, but strict pipefail would blame the chain for it. Mark the producer Command.uncheckedInPipe (fluent: .UncheckedInPipe()) and pipefail skips it:

F#

task {
    // seq 1 1000000 | head -n 1 — the producer's broken-pipe death is expected.
    let first =
        (Command.create "seq" |> Command.args [ "1"; "1000000" ] |> Command.uncheckedInPipe)
            .Pipe(Command.create "head" |> Command.args [ "-n"; "1" ])

    match! first.RunAsync() with
    | Ok line -> printfn $"{line}" // "1"
    | Error err -> eprintfn $"{err.Message}"
}

C#

// seq 1 1000000 | head -n 1 — the producer's broken-pipe death is expected.
var first = new Command("seq").Args(["1", "1000000"]).UncheckedInPipe()
    .Pipe(new Command("head").Args(["-n", "1"]));

Console.WriteLine(await first.RunAsync() switch
{
    { IsOk: true, ResultValue: var line } => line, // "1"
    { IsOk: false, ErrorValue: var err } => err.Message,
});

The rules:

• Every unchecked non-last stage is excluded from culprit selection, whatever its outcome; this is why an expected broken-pipe failure or POSIX SIGPIPE is skipped. Its outcome is not made acceptable — it simply is not selected as the representative result.
• A checked failure always trumps an unchecked one, regardless of position: uncheckedInPipe never shields another stage's real failure.
• When no checked stage failed, the result preserves the real last stage's program, outcome, stderr, and exit code. If that last stage is unchecked and exited voluntarily, its actual code is included in AcceptedCodes, so the result is successful without rewriting it to 0; a signal, timeout, or unobserved outcome on that last stage has no code to accept and remains a failure.
• uncheckedInPipe forgives exit status only — never a whole-chain Pipeline.Timeout — and it has no effect on a Command run outside a pipeline, where a single run's status is already plain data in its ProcessResult.

Timeouts and cancellation

A pipeline is bounded as a whole chain — there is no per-stage timeout:

F#

task {
    let pipeline =
        (Command.create "producer")
            .Pipe(Command.create "consumer")
            .Timeout(TimeSpan.FromSeconds 30.0) // whole-CHAIN deadline

    match! pipeline.OutputStringAsync() with
    | Ok result -> printfn $"timedOut={result.IsTimedOut}"
    | Error err -> eprintfn $"{err.Message}"
}

C#

var pipeline = new Command("producer")
    .Pipe(new Command("consumer"))
    .Timeout(TimeSpan.FromSeconds(30)); // whole-CHAIN deadline

Console.WriteLine(await pipeline.OutputStringAsync() switch
{
    { IsOk: true, ResultValue: var result } => $"timedOut={result.IsTimedOut}",
    { IsOk: false, ErrorValue: var err }   => err.Message,
});

• Pipeline.Timeout (module mirror: Pipeline.timeout) bounds the whole chain: at the deadline the shared group is torn down and the run reports the timeout — on OutputStringAsync as IsTimedOut, on RunAsync as an Error. Unlike a single command's captured timeout, there is no salvaged partial stdout to read back.
• A per-stage Command.Timeout cannot bound one stage of a chain — a pipeline spawns its stages directly, so a stage's own deadline never fires. Setting one is a configuration error, so .Pipe rejects it with an ArgumentException. Bound the whole chain with Pipeline.Timeout, or run the stage as a standalone Command when it needs its own deadline.
• A per-stage Command.Retry is rejected the same way — retry is a verb-layer mechanism and pipeline stages are spawned directly, bypassing it. Retry the pipeline as a whole instead.

Cancellation has two forms:

• Pipeline.CancelOn(token) (module mirror: Pipeline.cancelOn) is the chain-level control: the token is applied to every stage, so firing it tears the whole chain down and the run resolves to ProcessError.Cancelled.
• Each verb's optional CancellationToken (pipeline.RunAsync(token)) ties a single call to a token without baking it into the pipeline.

A per-stage Command.CancelOn cannot cancel one stage of a chain — a pipeline spawns its stages directly, so a stage's own token is a verb-layer mechanism the spawn bypasses and never fires. Setting one is a configuration error, so .Pipe rejects it with an ArgumentException. Cancel the whole chain with the chain-level Pipeline.CancelOn (or pass a token to the verb) instead. For the full model — captured vs. raised deadlines, and how cancellation differs from a timeout — see timeouts-and-cancellation.md.

A cancelled chain is hard-killed at once by default. To let its stages clean up first, set Command.CancelGrace (and optionally CancelSignal) on stage 0, which owns the pipeline-wide control configuration alongside StopSignal: the whole chain is then sent one soft signal, given the grace window, and only then hard-killed. Setting either on a later stage is rejected with an ArgumentException — a chain has one shared group and therefore one broadcast soft-stop phase. The chain-level Pipeline.Timeout is a different event and keeps its immediate hard kill.

Streaming a pipeline

The buffering verbs run the whole chain to completion before handing back its folded result. For a long-running or interactive pipeline — journalctl -f | grep ERROR, a server started behind a filter, any chain whose final output you want to read as it arrives rather than buffered until every stage exits — start a live streaming session instead. Pipeline.StartAsync() spawns the whole chain into one shared kill-on-dispose group and returns a Task<Result<PipelineSession, ProcessError>>.

A PipelineSession is the pipeline analogue of a single command's RunningProcess: it streams the final stage's stdout as it arrives, waits for the whole chain with the same pipefail classification the buffering verbs use, and stops/reaps the entire chain on demand.

F#

task {
    let pipeline =
        (Command.create "journalctl" |> Command.args [ "-f" ])
            .Pipe(Command.create "grep" |> Command.args [ "--line-buffered"; "ERROR" ])

    match! pipeline.StartAsync() with
    | Error err -> eprintfn $"could not start: {err.Message}"
    | Ok session ->
        use session = session
        let e = session.StdoutLinesAsync().GetAsyncEnumerator()

        try
            // Read matching lines from the LAST stage (grep) as they appear.
            let mutable seen = 0

            while seen < 10 do
                match! e.MoveNextAsync() with
                | true ->
                    printfn $"error: {e.Current}"
                    seen <- seen + 1
                | false -> seen <- 10
        finally
            e.DisposeAsync().AsTask().Wait()

        // Tears down BOTH stages (journalctl AND grep), not just the last.
        let! _ = session.StopAsync()
        ()
}

C#

var pipeline = new Command("journalctl").Args(["-f"])
    .Pipe(new Command("grep").Args(["--line-buffered", "ERROR"]));

await using var session = (await pipeline.StartAsync()).GetValueOrThrow();

var seen = 0;
await foreach (var line in session.StdoutLinesAsync())
{
    Console.WriteLine($"error: {line}");
    if (++seen >= 10) break;
}

await session.StopAsync();   // reaps BOTH stages

The session mirrors RunningProcess:

Single consumption, exactly like RunningProcess: the final stage's stdout is pumped once, so StdoutLinesAsync and OutputEventsAsync are mutually exclusive (a second, different streaming consumer throws), and FinishAsync rejoins the stdout-streaming session StdoutLinesAsync started — call it after streaming lines; after OutputEventsAsync, reap with StopAsync (or dispose) rather than FinishAsync.

Whole-chain semantics. The stream is the final stage's stdout, but FinishAsync / StopAsync reap and classify the ENTIRE chain: the returned Finished.Outcome is the pipefail representative (the rightmost checked stage that did not exit with an accepted code, or a TimedOut/Cancelled for the whole chain), and Finished.Stderr is that stage's stderr — identical to what RunAsync reports, never a final-stage-only view. Finished.Truncated combines drops from the final stdout stream with truncation of that representative stderr only; truncation on another stage does not mark diagnostics that were not published. A non-zero pipefail exit is data in Finished.Outcome; a genuine upstream relay read failure is returned as ProcessError.Io, while a downstream broken pipe and teardown race remain quiet. A fail-loud output overflow on any stage or a stage-0 stdin-source failure surfaces as Error. The chain-level Timeout/CancelOn still apply — either firing hard-kills the tree, and FinishAsync then reports TimedOut/Cancelled. Stopping or disposing tears down every stage — including a partially started chain, when a later stage never spawned — never just the last, so no stage is ever orphaned.

Re-running a pipeline

A Pipeline is an immutable value: building it spawns nothing, and each verb call drives the chain afresh, so you can hold one and run it more than once. The one caveat is inherited from Command — when a chain runs repeatedly, feed the first stage from a reusable stdin source (Stdin.FromString / Stdin.FromBytes / Stdin.FromFile) rather than a stream you can only read once. A one-shot source (Stdin.FromStream / FromLines / FromAsyncLines) feeds one chain and one chain only: stage 0's spawn takes it, so a second run of the chain — buffered or StartAsync — is refused with ProcessError.Unsupported at stage 0, which means no stage of that chain starts at all, and a run over a source some other child already read is refused the same way. If stage 0's spawn itself fails, the source is handed back untouched. See One-shot stdin sources feed one incarnation for the whole contract and commands.md for the full set of stdin sources and their semantics.

Next: Timeouts, retries & cancellation