Terminate owned subprocess trees

internal/adapters/subprocess/process_tree.go

@@ -0,0 +1,65 @@
+package subprocess
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os/exec"
+	"time"
+)
+
+const (
+	gracefulTerminationWait = 2 * time.Second
+	forcefulTerminationWait = 2 * time.Second
+)
+
+// ownedProcessTree owns every process started by a command invocation.
+// Implementations must tolerate a leader that has already exited.
+type ownedProcessTree interface {
+	Start(*exec.Cmd) error
+	TerminateGracefully() error
+	TerminateForcefully() error
+	Close() error
+}
+
+func waitForOwnedCommand(ctx context.Context, tree ownedProcessTree, waitCh <-chan error) (waitErr, ctxErr, cleanupErr error) {
+	select {
+	case waitErr = <-waitCh:
+		return waitErr, nil, nil
+	case <-ctx.Done():
+		ctxErr = ctx.Err()
+	}
+
+	cleanupErr = tree.TerminateGracefully()
+	gracefulTimer := time.NewTimer(gracefulTerminationWait)
+	defer gracefulTimer.Stop()
+
+	select {
+	case waitErr = <-waitCh:
+		// The leader may exit before descendants finish graceful shutdown.
+		cleanupErr = joinErrors(cleanupErr, tree.TerminateForcefully())
+		return waitErr, ctxErr, cleanupErr
+	case <-gracefulTimer.C:
+	}
+
+	cleanupErr = joinErrors(cleanupErr, tree.TerminateForcefully())
+	forcefulTimer := time.NewTimer(forcefulTerminationWait)
+	defer forcefulTimer.Stop()
+
+	select {
+	case waitErr = <-waitCh:
+		return waitErr, ctxErr, cleanupErr
+	case <-forcefulTimer.C:
+		return nil, ctxErr, joinErrors(cleanupErr, fmt.Errorf("owned subprocess did not reap within %s after forceful termination", forcefulTerminationWait))
+	}
+}
+
+func joinErrors(errs ...error) error {
+	filtered := make([]error, 0, len(errs))
+	for _, err := range errs {
+		if err != nil {
+			filtered = append(filtered, err)
+		}
+	}
+	return errors.Join(filtered...)
+}

internal/adapters/subprocess/process_tree_supported_test.go

@@ -0,0 +1,119 @@
+//go:build linux || darwin || windows
+
+package subprocess
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestRunCancellationTerminatesProcessTree(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	resultCh := make(chan runOutcome, 1)
+	req, sentinelPath := processTreeRequest(t)
+	go func() {
+		result, err := Run(ctx, req)
+		resultCh <- runOutcome{result: result, err: err}
+	}()
+
+	awaitHelperReady(t, req.EnvOverrides["SUBPROCESS_HELPER_READY_PATH"])
+	cancel()
+
+	outcome := awaitRunOutcome(t, resultCh)
+	if !outcome.result.Canceled {
+		t.Fatalf("Canceled = %v, want true", outcome.result.Canceled)
+	}
+	if !errors.Is(outcome.err, context.Canceled) {
+		t.Fatalf("error = %v, want context cancellation", outcome.err)
+	}
+	assertDescendantDidNotSurvive(t, sentinelPath)
+}
+
+func TestRunTimeoutTerminatesProcessTree(t *testing.T) {
+	req, sentinelPath := processTreeRequest(t)
+	req.Timeout = 100 * time.Millisecond
+
+	result, err := Run(context.Background(), req)
+	if !result.TimedOut {
+		t.Fatalf("TimedOut = %v, want true", result.TimedOut)
+	}
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("error = %v, want context deadline exceeded", err)
+	}
+	assertDescendantDidNotSurvive(t, sentinelPath)
+}
+
+type runOutcome struct {
+	result RunResult
+	err    error
+}
+
+func processTreeRequest(t *testing.T) (RunRequest, string) {
+	t.Helper()
+
+	executable, err := os.Executable()
+	if err != nil {
+		t.Fatalf("os.Executable() error = %v", err)
+	}
+	dir := t.TempDir()
+	readyPath := filepath.Join(dir, "ready")
+	sentinelPath := filepath.Join(dir, "descendant-survived")
+	return RunRequest{
+		Executable: executable,
+		Args:       []string{"-test.run=^TestSubprocessHelper$", "--", "tree"},
+		EnvOverrides: map[string]string{
+			"GO_WANT_SUBPROCESS_HELPER":       "1",
+			"SUBPROCESS_HELPER_READY_PATH":    readyPath,
+			"SUBPROCESS_HELPER_SENTINEL_PATH": sentinelPath,
+		},
+		StdoutLogPath: filepath.Join(dir, "stdout.log"),
+		StderrLogPath: filepath.Join(dir, "stderr.log"),
+	}, sentinelPath
+}
+
+func awaitHelperReady(t *testing.T, readyPath string) {
+	t.Helper()
+
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, err := os.Stat(readyPath); err == nil {
+			return
+		} else if !errors.Is(err, os.ErrNotExist) {
+			t.Fatalf("stat helper readiness: %v", err)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("helper did not start its descendant")
+}
+
+func awaitRunOutcome(t *testing.T, outcomes <-chan runOutcome) runOutcome {
+	t.Helper()
+
+	select {
+	case outcome := <-outcomes:
+		if outcome.err == nil {
+			t.Fatal("Run() error = nil, want cancellation error")
+		}
+		return outcome
+	case <-time.After(3 * time.Second):
+		t.Fatal("Run() did not return after cancellation")
+		return runOutcome{}
+	}
+}
+
+func assertDescendantDidNotSurvive(t *testing.T, sentinelPath string) {
+	t.Helper()
+
+	time.Sleep(700 * time.Millisecond)
+	if _, err := os.Stat(sentinelPath); err == nil {
+		t.Fatal("descendant survived cancellation and wrote its sentinel")
+	} else if !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("stat descendant sentinel: %v", err)
+	}
+}

internal/adapters/subprocess/process_tree_unix.go

@@ -0,0 +1,50 @@
+//go:build linux || darwin
+
+package subprocess
+
+import (
+	"errors"
+	"os"
+	"os/exec"
+	"syscall"
+)
+
+type unixProcessTree struct {
+	processGroupID int
+}
+
+func newOwnedProcessTree() (ownedProcessTree, error) {
+	return &unixProcessTree{}, nil
+}
+
+func (tree *unixProcessTree) Start(cmd *exec.Cmd) error {
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	if err := cmd.Start(); err != nil {
+		return err
+	}
+	tree.processGroupID = cmd.Process.Pid
+	return nil
+}
+
+func (tree *unixProcessTree) TerminateGracefully() error {
+	return tree.signal(syscall.SIGTERM)
+}
+
+func (tree *unixProcessTree) TerminateForcefully() error {
+	return tree.signal(syscall.SIGKILL)
+}
+
+func (tree *unixProcessTree) Close() error {
+	return nil
+}
+
+func (tree *unixProcessTree) signal(signal syscall.Signal) error {
+	if tree.processGroupID <= 0 {
+		return nil
+	}
+	err := syscall.Kill(-tree.processGroupID, signal)
+	if errors.Is(err, syscall.ESRCH) || errors.Is(err, os.ErrProcessDone) {
+		return nil
+	}
+	return err
+}

internal/adapters/subprocess/process_tree_unsupported.go

@@ -0,0 +1,12 @@
+//go:build !linux && !darwin && !windows
+
+package subprocess
+
+import (
+	"fmt"
+	"runtime"
+)
+
+func newOwnedProcessTree() (ownedProcessTree, error) {
+	return nil, fmt.Errorf("owned subprocess trees are unsupported on %s", runtime.GOOS)
+}

internal/adapters/subprocess/process_tree_windows.go

@@ -0,0 +1,122 @@
+//go:build windows
+
+package subprocess
+
+import (
+	"errors"
+	"fmt"
+	"os/exec"
+	"syscall"
+	"unsafe"
+
+	"golang.org/x/sys/windows"
+)
+
+type windowsProcessTree struct {
+	job windows.Handle
+}
+
+func newOwnedProcessTree() (ownedProcessTree, error) {
+	return &windowsProcessTree{}, nil
+}
+
+func (tree *windowsProcessTree) Start(cmd *exec.Cmd) error {
+	job, err := windows.CreateJobObject(nil, nil)
+	if err != nil {
+		return fmt.Errorf("create job object: %w", err)
+	}
+
+	limits := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{}
+	limits.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
+	if _, err := windows.SetInformationJobObject(job, windows.JobObjectExtendedLimitInformation, uintptr(unsafe.Pointer(&limits)), uint32(unsafe.Sizeof(limits))); err != nil {
+		_ = windows.CloseHandle(job)
+		return fmt.Errorf("configure job object: %w", err)
+	}
+
+	cmd.SysProcAttr = &syscall.SysProcAttr{CreationFlags: windows.CREATE_SUSPENDED}
+	if err := cmd.Start(); err != nil {
+		_ = windows.CloseHandle(job)
+		return err
+	}
+
+	process, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, false, uint32(cmd.Process.Pid))
+	if err == nil {
+		err = windows.AssignProcessToJobObject(job, process)
+		_ = windows.CloseHandle(process)
+	}
+	if err == nil {
+		err = resumeInitialThread(uint32(cmd.Process.Pid))
+	}
+	if err != nil {
+		killErr := cmd.Process.Kill()
+		waitErr := cmd.Wait()
+		_ = windows.CloseHandle(job)
+		return joinErrors(fmt.Errorf("assign process to job object: %w", err), killErr, waitErr)
+	}
+
+	tree.job = job
+	return nil
+}
+
+func resumeInitialThread(processID uint32) error {
+	snapshot, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0)
+	if err != nil {
+		return fmt.Errorf("snapshot initial thread: %w", err)
+	}
+	defer func() { _ = windows.CloseHandle(snapshot) }()
+
+	entry := windows.ThreadEntry32{Size: uint32(unsafe.Sizeof(windows.ThreadEntry32{}))}
+	if err := windows.Thread32First(snapshot, &entry); err != nil {
+		return fmt.Errorf("find initial thread: %w", err)
+	}
+	for {
+		if entry.OwnerProcessID != processID {
+			// Keep enumerating until the suspended process's only initial thread
+			// is found.
+		} else {
+			thread, openErr := windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, entry.ThreadID)
+			if openErr != nil {
+				return fmt.Errorf("open initial thread: %w", openErr)
+			}
+			defer func() { _ = windows.CloseHandle(thread) }()
+			if _, resumeErr := windows.ResumeThread(thread); resumeErr != nil {
+				return fmt.Errorf("resume initial thread: %w", resumeErr)
+			}
+			return nil
+		}
+
+		if err := windows.Thread32Next(snapshot, &entry); err != nil {
+			if errors.Is(err, windows.ERROR_NO_MORE_FILES) {
+				break
+			}
+			return fmt.Errorf("find initial thread: %w", err)
+		}
+	}
+	return fmt.Errorf("find initial thread: no thread found for process %d", processID)
+}
+
+func (tree *windowsProcessTree) TerminateGracefully() error {
+	// Windows jobs have no portable graceful signal. Terminating the owned job
+	// is the safe fallback and prevents a descendant from escaping cleanup.
+	return tree.terminate()
+}
+
+func (tree *windowsProcessTree) TerminateForcefully() error {
+	return tree.terminate()
+}
+
+func (tree *windowsProcessTree) Close() error {
+	if tree.job == 0 {
+		return nil
+	}
+	err := windows.CloseHandle(tree.job)
+	tree.job = 0
+	return err
+}
+
+func (tree *windowsProcessTree) terminate() error {
+	if tree.job == 0 {
+		return nil
+	}
+	return windows.TerminateJobObject(tree.job, 1)
+}

internal/adapters/subprocess/run.go

@@ -2,7 +2,6 @@ package subprocess
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -61,7 +60,12 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) {
 	}
 	defer logs.Close()
 
-	cmd := exec.CommandContext(runCtx, req.Executable, req.Args...)
+	tree, err := newOwnedProcessTree()
+	if err != nil {
+		return RunResult{}, fmt.Errorf("prepare owned subprocess tree: %w", err)
+	}
+
+	cmd := exec.Command(req.Executable, req.Args...)
 	cmd.Dir = req.WorkingDir
 	cmd.Env = mergeEnv(os.Environ(), req.EnvOverrides)
 	cmd.Stdout = logs.Stdout
@@ -75,28 +79,37 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) {
 		StderrLogPath: req.StderrLogPath,
 	}
 
-	if err := cmd.Start(); err != nil {
+	if err := runCtx.Err(); err != nil {
+		result.CompletedAt = time.Now().UTC()
+		result.Duration = result.CompletedAt.Sub(result.StartedAt)
+		return result, fmt.Errorf("command was not started: %w", err)
+	}
+
+	if err := tree.Start(cmd); err != nil {
 		result.CompletedAt = time.Now().UTC()
 		result.Duration = result.CompletedAt.Sub(result.StartedAt)
 		return result, fmt.Errorf("start command %q with args %v: %w", req.Executable, req.Args, err)
 	}
 
-	waitErr := cmd.Wait()
+	waitCh := make(chan error, 1)
+	go func() { waitCh <- cmd.Wait() }()
+
+	waitErr, ctxErr, cleanupErr := waitForOwnedCommand(runCtx, tree, waitCh)
+	cleanupErr = joinErrors(cleanupErr, tree.Close())
 	result.CompletedAt = time.Now().UTC()
 	result.Duration = result.CompletedAt.Sub(result.StartedAt)
 	if cmd.ProcessState != nil {
 		result.ExitCode = cmd.ProcessState.ExitCode()
 	}
 
-	ctxErr := runCtx.Err()
-	if errors.Is(ctxErr, context.DeadlineExceeded) {
+	if ctxErr == context.DeadlineExceeded {
 		result.TimedOut = true
 	}
-	if errors.Is(ctxErr, context.Canceled) && !result.TimedOut {
+	if ctxErr == context.Canceled && !result.TimedOut {
 		result.Canceled = true
 	}
 
-	if waitErr == nil {
+	if waitErr == nil && ctxErr == nil && cleanupErr == nil {
 		return result, nil
 	}
 
@@ -104,14 +117,17 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) {
 	diagnostics := buildDiagnostics(req, result, stderrTail)
 
 	if result.TimedOut {
-		return result, fmt.Errorf("command timed out after %s (%s)", req.Timeout, diagnostics)
+		return result, fmt.Errorf("command timed out after %s (%s): %w", req.Timeout, diagnostics, joinErrors(ctxErr, waitErr, cleanupErr))
 	}
 	if result.Canceled {
-		return result, fmt.Errorf("command canceled (%s)", diagnostics)
+		return result, fmt.Errorf("command canceled (%s): %w", diagnostics, joinErrors(ctxErr, waitErr, cleanupErr))
 	}
 	if exitErr, ok := waitErr.(*exec.ExitError); ok {
 		return result, fmt.Errorf("command failed with exit code %d (%s): %w", exitErr.ExitCode(), diagnostics, waitErr)
 	}
+	if cleanupErr != nil {
+		return result, fmt.Errorf("command cleanup failed (%s): %w", diagnostics, joinErrors(waitErr, cleanupErr))
+	}
 
 	return result, fmt.Errorf("command failed to run (%s): %w", diagnostics, waitErr)
 }
@@ -203,7 +219,13 @@ func cleanLogPath(path string) string {
 
 func logWriter(path string) (*os.File, io.Writer, error) {
 	if strings.TrimSpace(path) == "" {
-		return nil, io.Discard, nil
+		// Use a descriptor instead of io.Discard so os/exec does not create a
+		// pipe and wait for a descendant that inherited it after its leader exits.
+		f, err := os.OpenFile(os.DevNull, os.O_WRONLY, 0)
+		if err != nil {
+			return nil, nil, fmt.Errorf("open null output: %w", err)
+		}
+		return f, f, nil
 	}
 	f, err := openLogFile(path)
 	if err != nil {

internal/adapters/subprocess/run_test.go

@@ -3,6 +3,7 @@ package subprocess
 import (
 	"context"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -368,6 +369,24 @@ func TestSubprocessHelper(t *testing.T) {
 		key := os.Getenv("SUBPROCESS_HELPER_ENV_KEY")
 		_, _ = os.Stdout.WriteString(os.Getenv(key) + "\n")
 		os.Exit(0)
+	case "tree":
+		descendant := exec.Command(os.Args[0], "-test.run=^TestSubprocessHelper$", "--", "descendant")
+		descendant.Env = append(os.Environ(), "GO_WANT_SUBPROCESS_HELPER=1")
+		descendant.Stdout = os.Stdout
+		descendant.Stderr = os.Stderr
+		if err := descendant.Start(); err != nil {
+			os.Exit(3)
+		}
+		if err := os.WriteFile(os.Getenv("SUBPROCESS_HELPER_READY_PATH"), []byte("ready"), 0o600); err != nil {
+			os.Exit(4)
+		}
+		time.Sleep(10 * time.Second)
+		os.Exit(0)
+	case "descendant":
+		time.Sleep(500 * time.Millisecond)
+		_ = os.WriteFile(os.Getenv("SUBPROCESS_HELPER_SENTINEL_PATH"), []byte("survived"), 0o600)
+		time.Sleep(10 * time.Second)
+		os.Exit(0)
 	default:
 		os.Exit(2)
 	}