eric/ReleaseKit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Add piped child process support

Browse Source
This commit is contained in:
Eric Rakestraw
2026-10-09 01:17:58 +00:00
parent 41e58ef894
commit 618a7c9dd3
4 changed files with 267 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
internal/process/process.go
View File

@@ -10,6 +10,7 @@ import (
"os/exec"
"sort"
"strings"
"sync"
)
// Invocation describes one child process. UnsetEnv removes inherited values
@@ -27,16 +28,11 @@ type Invocation struct {
// Run starts the command in the requested directory and waits for it to exit.
// Errors preserve the underlying exec error, including *exec.ExitError.
func Run(ctx context.Context, invocation Invocation) error {
if len(invocation.Args) == 0 || invocation.Args[0] == "" {
return errors.New("process invocation requires a command")
command, err := newCommand(ctx, invocation)
if err != nil {
return err
}
command := exec.CommandContext(ctx, invocation.Args[0], invocation.Args[1:]...)
command.Dir = invocation.Dir
command.Stdin = invocation.Stdin
if len(invocation.Env) != 0 || len(invocation.UnsetEnv) != 0 {
command.Env = environmentEntries(invocation.Env, invocation.UnsetEnv)
}
command.Stdout = invocation.Stdout
command.Stderr = invocation.Stderr
@@ -49,6 +45,102 @@ func Run(ctx context.Context, invocation Invocation) error {
return nil
}
// PipedProcess owns the command started by StartPiped. Callers close Stdin,
// read Stdout to EOF, and call Wait once after a normal exchange. On an abort,
// close active pipes, call Cancel, and then Wait to reap the child.
type PipedProcess struct {
command *exec.Cmd
stdin io.WriteCloser
stdout io.ReadCloser
parent context.Context
cancel context.CancelFunc
waitOnce sync.Once
waitErr error
}
// Stdin returns the child process input pipe. The caller owns its closure.
func (child *PipedProcess) Stdin() io.WriteCloser {
return child.stdin
}
// Stdout returns the child process output pipe. The caller must consume it
// before Wait on a normal exit and owns its closure on an abort.
func (child *PipedProcess) Stdout() io.ReadCloser {
return child.stdout
}
// Cancel interrupts the child without cancelling its parent context.
func (child *PipedProcess) Cancel() {
child.cancel()
}
// Wait waits for and reaps the child. It preserves the parent's cancellation
// cause and otherwise returns the command's wrapped exit error.
func (child *PipedProcess) Wait() error {
child.waitOnce.Do(func() {
err := child.command.Wait()
child.cancel()
if err == nil {
return
}
if contextErr := child.parent.Err(); contextErr != nil {
child.waitErr = fmt.Errorf("run %q: %w", child.command.Args[0], contextErr)
return
}
child.waitErr = fmt.Errorf("run %q: %w", child.command.Args[0], err)
})
return child.waitErr
}
// StartPiped starts a child with stdin and stdout connected to pipes. Command
// settings come from invocation; its Stdin and Stdout fields are ignored, while
// Stderr remains routed as requested.
func StartPiped(ctx context.Context, invocation Invocation) (*PipedProcess, error) {
childContext, cancel := context.WithCancel(ctx)
command, err := newCommand(childContext, invocation)
if err != nil {
cancel()
return nil, err
}
command.Stderr = invocation.Stderr
stdin, err := command.StdinPipe()
if err != nil {
cancel()
return nil, fmt.Errorf("pipe stdin for %q: %w", invocation.Args[0], err)
}
stdout, err := command.StdoutPipe()
if err != nil {
_ = stdin.Close()
cancel()
return nil, fmt.Errorf("pipe stdout for %q: %w", invocation.Args[0], err)
}
if err := command.Start(); err != nil {
_ = stdin.Close()
_ = stdout.Close()
cancel()
return nil, fmt.Errorf("start %q: %w", invocation.Args[0], err)
}
return &PipedProcess{
command: command,
stdin: stdin,
stdout: stdout,
parent: ctx,
cancel: cancel,
}, nil
}
func newCommand(ctx context.Context, invocation Invocation) (*exec.Cmd, error) {
if len(invocation.Args) == 0 || invocation.Args[0] == "" {
return nil, errors.New("process invocation requires a command")
}
command := exec.CommandContext(ctx, invocation.Args[0], invocation.Args[1:]...)
command.Dir = invocation.Dir
if len(invocation.Env) != 0 || len(invocation.UnsetEnv) != 0 {
command.Env = environmentEntries(invocation.Env, invocation.UnsetEnv)
}
return command, nil
}
func environmentEntries(overlay map[string]string, unset []string) []string {
environment := make(map[string]string, len(os.Environ())+len(overlay))
for _, entry := range os.Environ() {

157
internal/process/process_test.go
View File

@@ -1,10 +1,12 @@
package process
import (
"bufio"
"bytes"
"context"
"errors"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
@@ -117,6 +119,146 @@ func TestRunReturnsCancellation(t *testing.T) {
}
}
func TestStartPipedExchangesBeforeInputEOFAndPreservesSettings(t *testing.T) {
workdir := filepath.Join(t.TempDir(), "piped directory")
if err := os.Mkdir(workdir, 0o700); err != nil {
t.Fatal(err)
}
t.Setenv("RELEASEKIT_PROCESS_OVERLAY", "parent")
t.Setenv("RELEASEKIT_PROCESS_INHERITED", "inherited")
var stderr bytes.Buffer
child, err := StartPiped(context.Background(), Invocation{
Args: []string{helperExecutable(t), "-test.run=TestProcessHelper", "--", "argument with spaces", "exchange"},
Dir: workdir,
Env: map[string]string{
"RELEASEKIT_PROCESS_HELPER": "1",
"RELEASEKIT_PROCESS_OVERLAY": "child",
},
Stderr: &stderr,
})
if err != nil {
t.Fatal(err)
}
if _, err := io.WriteString(child.Stdin(), "request\n"); err != nil {
t.Fatal(err)
}
response, err := bufio.NewReader(child.Stdout()).ReadString('\n')
if err != nil {
t.Fatalf("read response before closing input: %v", err)
}
want := fmt.Sprintf("request|%s|argument with spaces|child|inherited\n", workdir)
if response != want {
t.Fatalf("piped response = %q, want %q", response, want)
}
if err := child.Stdin().Close(); err != nil {
t.Fatal(err)
}
if _, err := io.Copy(io.Discard, child.Stdout()); err != nil {
t.Fatal(err)
}
if err := child.Wait(); err != nil {
t.Fatal(err)
}
if got := stderr.String(); got != "child stderr\n" {
t.Fatalf("child stderr = %q, want %q", got, "child stderr\n")
}
if got := os.Getenv("RELEASEKIT_PROCESS_OVERLAY"); got != "parent" {
t.Fatalf("parent environment changed: got %q, want %q", got, "parent")
}
}
func TestStartPipedCancellationAfterReadinessReapsChild(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
child, err := StartPiped(ctx, Invocation{
Args: []string{helperExecutable(t), "-test.run=TestProcessHelper", "--", "unused", "wait"},
Env: map[string]string{"RELEASEKIT_PROCESS_HELPER": "1"},
})
if err != nil {
cancel()
t.Fatal(err)
}
line, err := bufio.NewReader(child.Stdout()).ReadString('\n')
if err != nil || line != "started\n" {
cancel()
_ = child.Stdin().Close()
_ = child.Stdout().Close()
_ = child.Wait()
t.Fatalf("readiness = %q, %v; want started signal", line, err)
}
cancel()
if err := child.Stdin().Close(); err != nil {
t.Fatal(err)
}
if err := child.Stdout().Close(); err != nil {
t.Fatal(err)
}
if err := child.Wait(); !errors.Is(err, context.Canceled) {
t.Fatalf("Wait() error = %v, want context cancellation", err)
}
}
func TestStartPipedCancelDoesNotCancelParentContext(t *testing.T) {
ctx := context.Background()
child, err := StartPiped(ctx, Invocation{
Args: []string{helperExecutable(t), "-test.run=TestProcessHelper", "--", "unused", "wait"},
Env: map[string]string{"RELEASEKIT_PROCESS_HELPER": "1"},
})
if err != nil {
t.Fatal(err)
}
line, err := bufio.NewReader(child.Stdout()).ReadString('\n')
if err != nil || line != "started\n" {
child.Cancel()
_ = child.Stdin().Close()
_ = child.Stdout().Close()
_ = child.Wait()
t.Fatalf("readiness = %q, %v; want started signal", line, err)
}
child.Cancel()
if err := child.Stdin().Close(); err != nil {
t.Fatal(err)
}
if err := child.Stdout().Close(); err != nil {
t.Fatal(err)
}
err = child.Wait()
var exitErr *exec.ExitError
if !errors.As(err, &exitErr) {
t.Fatalf("Wait() error = %v, want interrupted child exit", err)
}
if ctx.Err() != nil {
t.Fatalf("child cancellation changed parent context: %v", ctx.Err())
}
}
func TestStartPipedReportsStartFailure(t *testing.T) {
child, err := StartPiped(context.Background(), Invocation{Args: []string{"releasekit-command-that-does-not-exist"}})
if err == nil || child != nil || !strings.Contains(err.Error(), "releasekit-command-that-does-not-exist") {
t.Fatalf("StartPiped() = %#v, %v; want a missing executable error", child, err)
}
}
func TestStartPipedPreservesNonzeroExit(t *testing.T) {
child, err := StartPiped(context.Background(), Invocation{
Args: []string{helperExecutable(t), "-test.run=TestProcessHelper", "--", "unused", "exit"},
Env: map[string]string{"RELEASEKIT_PROCESS_HELPER": "1"},
})
if err != nil {
t.Fatal(err)
}
if err := child.Stdin().Close(); err != nil {
t.Fatal(err)
}
if _, err := io.Copy(io.Discard, child.Stdout()); err != nil {
t.Fatal(err)
}
err = child.Wait()
var exitErr *exec.ExitError
if !errors.As(err, &exitErr) || exitErr.ExitCode() != 7 {
t.Fatalf("Wait() error = %v, want exit code 7", err)
}
}
func TestProcessHelper(t *testing.T) {
if os.Getenv("RELEASEKIT_PROCESS_HELPER") != "1" {
return
@@ -142,6 +284,21 @@ func TestProcessHelper(t *testing.T) {
for {
time.Sleep(time.Second)
}
case "exchange":
reader := bufio.NewReader(os.Stdin)
request, err := reader.ReadString('\n')
if err != nil {
os.Exit(11)
}
workdir, err := os.Getwd()
if err != nil {
os.Exit(12)
}
fmt.Fprintf(os.Stdout, "%s|%s|%s|%s|%s\n", strings.TrimSuffix(request, "\n"), workdir, args[len(args)-2], os.Getenv("RELEASEKIT_PROCESS_OVERLAY"), os.Getenv("RELEASEKIT_PROCESS_INHERITED"))
fmt.Fprintln(os.Stderr, "child stderr")
if _, err := io.Copy(io.Discard, reader); err != nil {
os.Exit(13)
}
default:
os.Exit(10)
}