Stage interface redesign: pipe type negotiation, pooled buffer copies

[znull]

Rebases and modernizes @mhagger's Stage interface redesign (#21) onto current main, bumps the module version to v2, and takes advantage of the bump to make a couple other API changes.

What this is

go-pipe is the library used to wire up process pipelines (cat foo | sed | filter-fn | writer): it spawns the subprocesses, connects stdin/stdout between stages, and propagates errors. This PR reworks the core Stage interface so that the pipeline, not the stages, owns connecting adjacent stages. That allows for optimizations (like the copy-elimination below) and lets us delete the old synthetic ioCopier entirely.

The interface now hands each stage both its stdin and stdout, plus a new StartOptions, and asks each stage to declare its I/O preferences so the pipeline can pick the cheapest pipe type between neighbors:

type Stage interface {
    Name() string
    Preferences() StagePreferences
    Start(ctx context.Context, env Env, stdin io.ReadCloser, stdout io.WriteCloser, opts StartOptions) error
    Wait() error
}

Because this breaks the interface, the module path moves to /v2.

Perf optimizations

We want to minimize data copies, whether by letting the kernel move bytes instead of Go, or if we do copy in Go, doing it more efficiently. With the pipeline owning the connections it can:

• hand a command's stdout fd straight to the next stage (or to the final destination), so there's no Go-side goroutine copying between stages and dirtying the heap;
• pass an *os.File destination's fd directly into the child, letting the kernel (sendfile(2)/splice(2) and friends) do the copy (so-called "zero-copy" i/o);
• and where we do have to copy in-process, run it through a pooled 32KB buffer instead of letting exec.Cmd allocate a fresh one per pipeline.

#49 and #50 were aiming for similar optimizations, but now they fall out of the structure instead of being bolted onto ioCopier.

Panic Handling

go-pipe runs user code (function stages, memory-limit event handlers) in goroutines it spawns itself, so a panic there used to be able to take down the whole process. There was already panic handling, but it was part of the Stage interface, which caused a lot of problems with error-prone boilerplate in stages that don't do any panic handling (which is most of them). Panic handling is now a callback passed into Start that the pipeline threads to every stage, so a single handler covers all stages.

Supersedes

• Change the Stage interface to make stdin/stdout handling more flexible #21 (stale, merge conflicts)
• Implement an optional Stage2 interface that allows such stages to be started more flexibly #20 (the opt-in Stage2 variant)
• pool io.Copy buffers in ioCopier to reduce GC pressure #49 (sync.Pool for ioCopier copy buffers) — pool preserved in copy_pool.go
• Enable sendfile for network destinations; fix pool bypass on Go 1.26+ #50 (sendfile + pool-bypass fix in ioCopier) — sendfile preserved structurally; the WriterTo pool-bypass workaround is gone, since we control the copy site now

The git-systems/pooled-copies branch (which carried #49 + #50) can be deleted after this merges.

/cc @mhagger @migue @carlosmn

Copilot Summary

Interface change

Stage.Start gains stdout and a StartOptions struct (a struct so future run-scoped options don't break the interface again):

Start(ctx context.Context, env Env, stdin io.ReadCloser, stdout io.WriteCloser, opts StartOptions) error

Pipeline.Start() uses each stage's Preferences() to negotiate the pipe type between adjacent stages:

• os.Pipe() when either neighbor is a command (needs a real *os.File fd)
• io.Pipe() when both neighbors are Go functions (all userspace, cheaper)
• the pipeline's own stdout passed directly to the last stage — no synthetic ioCopier

Module path bumped to github.com/github/go-pipe/v2 for the breaking change.

Fast paths preserved from #49/#50

ioCopier is deleted; the optimizations it carried now live in the stage structure:

• fd dup-to-child / sendfile — a final commandStage writing to an *os.File destination dup's the fd into the child; for a non-*os.File writer that implements io.ReaderFrom, the copy goes through ReadFrom (sendfile where the kernel supports it). Pinned in pipe/command_stdout_fastpath_test.go.
• pooled 32KB copy buffers — for a non-*os.File, non-ReaderFrom stdout, setupPooledStdout builds an os.Pipe() and copies through a sync.Pool buffer rather than letting exec.Cmd allocate per-pipeline. See pipe/copy_pool.go.

Type unwrapping (replaces transparent NopCloser forwarding)

The old NopCloser re-exposed io.WriterTo by forwarding, so a naive io.Copy kept the fast path transparently through the wrapper. That's replaced by explicit, exported helpers:

• UnwrapReader(io.Reader) io.Reader / UnwrapWriter(io.Writer) io.Writer recover the concrete type go-pipe wrapped around stdin/stdout (nil-safe; non-wrappers pass through unchanged).
• goStage.Start and commandStage.Start call them internally, so every pipe.Function consumer transparently regains WriterTo / ReaderFrom / *os.File identity — a superset of the single method the old forwarding preserved, and it picked up a case goStage was previously missing.
• The exported helpers are the public extension point for external stages that consume raw stdin/stdout and want a fast path or fd identity. No such stage exists in go-pipe or gitrpcd today.

Panic handling

• StartOptions.PanicHandler (StagePanicHandler = func(p any) error) is set on the pipeline via WithStagePanicHandler and threaded to every stage's Start. The previous StagePanicHandlerAware opt-in interface is removed; wrapper stages just forward opts.
• Covered: Function stage StageFuncs and memory-limit event handlers (both run in library-spawned goroutines). If PanicHandler is nil the panic propagates and crashes — intentional, since gitrpcd treats an unhandled pipeline panic as catastrophic.
• A panic escaping the memory-watch goroutine is now recovered and surfaced via PanicHandler rather than crashing. The over-limit kill is guaranteed even if the event handler panics; a panic in a purely-informational handler (RSS-read error / peak usage) is recovered and the stage keeps running unmonitored ("fail open") — losing monitoring is not, on its own, a reason to stop a healthy stage. This is documented on MemoryLimit / StartOptions.PanicHandler.

Commit structure

• Cherry-picked from @mhagger's version-2 (authorship preserved): linter shush, pipeline_test.go cleanup, pipeline benchmarks, NopCloser simplification, the Stage interface change, pipe-matching tests.
• Reconciliation with main: port MemoryLimitWithObserver, restore the Function-stage panic handler, fix memoryWatchStage.Wait() to always stopWatching(), lint.
• Adversarial-review fixups: restore empty-pipeline identity copy, pin the command-stdout fd-pass fast path, restore pooled-buffer copy for non-*os.File stdout, avoid leaking the pooled-stdout goroutine when cmd.Start() fails.
• v2 API work: bump the module path to /v2, remove IOPreferenceNil, handle nil stdin/stdout cleanly, export Unwrap{Reader,Writer} and fully unwrap goStage stdin, thread the panic handler through Start (dropping StagePanicHandlerAware), and recover panics escaping the memory-watch goroutine.

[mhagger]

This PR is still in draft mode. Do you want review/feedback already?

[znull]

This PR is still in draft mode. Do you want review/feedback already?

@mhagger I ran out of time to review the LLM output before vacation. I wanted to read through the changes more fully myself before inflicting them on anyone else so I left it in draft mode.

[Copilot]

Pull request overview

This PR modernizes the pipeline stage contract for v2 by letting stages declare I/O preferences and receive both stdin and stdout from the pipeline, enabling better pipe selection and removing the synthetic ioCopier stage.

Changes:

• Redesigns Stage with Preferences() and Start(..., stdin, stdout) so Pipeline.Start() can negotiate os.Pipe vs io.Pipe.
• Reworks command/function/memory-limit stages for the new interface, including pooled stdout copies for non-file command destinations.
• Updates module path to /v2 and adds regression/benchmark coverage for pipe matching, empty pipelines, fast-path stdout, and start-failure cleanup.

Show a summary per file

File	Description
README.md	Updates documentation links for the v2 module path.
go.mod	Changes the module path to github.com/github/go-pipe/v2.
internal/ptree/ptree_test.go	Updates internal import path for v2.
pipe/stage.go	Redefines the public Stage interface and adds I/O preference types.
pipe/pipeline.go	Reworks pipeline startup to negotiate pipe types and pass stdout directly.
pipe/command.go	Adapts command stages to the new interface and adds pooled stdout copy handling.
pipe/function.go	Adapts function stages to receive caller-provided stdout and panic handling.
pipe/filter-error.go	Forwards panic handlers through error-filtering wrappers.
pipe/memorylimit.go	Ports memory-watching wrappers to the new stage interface.
pipe/nop_closer.go	Splits reader/writer nop closers and adds test unwrapping support.
pipe/copy_pool.go	Adds pooled-buffer copy helper with ReaderFrom fast-path support.
pipe/iocopier.go	Removes the old synthetic copier stage.
pipe/scanner.go	Simplifies scanner error return.
pipe/command_linux.go	Updates internal import path for v2.
pipe/command_test.go	Applies formatting cleanup.
pipe/command_nil_panic_test.go	Updates direct Start call for the new signature.
pipe/pipeline_test.go	Updates tests/benchmarks for v2 behavior, empty pipelines, and panic forwarding.
pipe/memorylimit_test.go	Reworks memory-limit tests for the new pipeline flow.
pipe/pipe_matching_test.go	Adds coverage for negotiated stdin/stdout pipe types.
pipe/export_test.go	Exposes nop-closer unwrapping for external package tests.
pipe/command_stdout_fastpath_test.go	Adds tests pinning direct *os.File stdout handoff.
pipe/command_starterror_test.go	Adds regression coverage for start-failure copy-goroutine cleanup.

Copilot's findings

Tip

Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

• Files reviewed: 22/22 changed files
• Comments generated: 2

[znull]

@mhagger I ran out of time to review the LLM output before vacation. I wanted to read through the changes more fully myself before inflicting them on anyone else so I left it in draft mode.

I think this is actually worth taking a look at now.

[Copilot]

Copilot's findings

• Files reviewed: 25/25 changed files
• Comments generated: 1

[mhagger]

Thanks for your patience with my comments ✨

I think that I like the new style of dealing with the panic handler. I have just one question: why do we need a new StartOptions type, as opposed to adding the panic handler to Env? They seem redundant with each other.

[znull]

Thanks for your patience with my comments ✨

Not at all, thanks for making them!

I think that I like the new style of dealing with the panic handler. I have just one question: why do we need a new StartOptions type, as opposed to adding the panic handler to Env? They seem redundant with each other.

I considered that briefly, but Env seemed more related to process specifics, and I was hesitant to make it a junk drawer of miscellaneous fields (which, admittedly, StartOptions pretty much is). In answering I'd consider not just the panic handler, but also #53's use of LeaveStdinOpen and LeaveStdoutOpen. On the whole, though, I can't say that any of those fields seems especially out of place in Env.

@mhagger EDIT: I noticed two things reviewing the code of #53:

• Env get passed to the user's goStage, whereas the uses of StartOptions is more library-internal
• In v2 Stage API: explicit close-responsibility + StageOptions consolidation #53, the handling of LeaveStdinOpen/LeaveStdoutOpen computes `StartOptions per-stage, to special case the first and last stages.

[mhagger]

Some thoughts about the tests related to handling panics.