From 7bd575187e30f30d2b6d881f52e9089a0b5b4baa Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 18:44:10 +0000 Subject: [PATCH] Terminate owned subprocess trees --- docs/internal/adapters.md | 6 + docs/operations.md | 10 ++ docs/roadmap/implementation.md | 2 +- internal/adapters/subprocess/process_tree.go | 65 ++++++++++ .../subprocess/process_tree_supported_test.go | 119 +++++++++++++++++ .../adapters/subprocess/process_tree_unix.go | 50 +++++++ .../subprocess/process_tree_unsupported.go | 12 ++ .../subprocess/process_tree_windows.go | 122 ++++++++++++++++++ internal/adapters/subprocess/run.go | 44 +++++-- internal/adapters/subprocess/run_test.go | 19 +++ 10 files changed, 437 insertions(+), 12 deletions(-) create mode 100644 internal/adapters/subprocess/process_tree.go create mode 100644 internal/adapters/subprocess/process_tree_supported_test.go create mode 100644 internal/adapters/subprocess/process_tree_unix.go create mode 100644 internal/adapters/subprocess/process_tree_unsupported.go create mode 100644 internal/adapters/subprocess/process_tree_windows.go diff --git a/docs/internal/adapters.md b/docs/internal/adapters.md index ce2a939..5ec2b63 100644 --- a/docs/internal/adapters.md +++ b/docs/internal/adapters.md @@ -56,6 +56,12 @@ configured filesystem secrets before adapter initialization. - Constructor errors fail stage execution setup early. - Runtime adapter errors propagate to stage code and then manifest failure handling. - Subprocess adapters persist stage logs/generated configs through stage-managed paths. +- Shared subprocess execution starts an owned process group on Linux/macOS or a + kill-on-close job object on Windows. Cancellation and deadlines request + termination, use a bounded forceful fallback, and wait for the leader before + returning. Unlogged stdout/stderr use direct null-device descriptors so a + descendant cannot retain an adapter pipe after its leader exits. Unsupported + platforms reject owned command execution. ## Implementation And Tests diff --git a/docs/operations.md b/docs/operations.md index 55d0232..f9372f2 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -139,6 +139,16 @@ temporary promotion tree because Narratio has no verified atomic no-replace directory primitive there. This is an extraction limitation, not a broader platform-support guarantee for every Narratio workflow. +## External Command Lifecycle + +When an external command is cancelled or times out, Narratio terminates its +owned descendants as well as the command itself. Cancellation first requests +termination where the platform supports it, then force terminates after a +bounded wait. A command is not considered finished until its leader has been +reaped, and descendants that keep standard output or error open cannot keep +the invocation blocked. Other operating systems fail closed rather than launch +a command without tree ownership. + Run-local diagnostics are: - `runs/{run_id}/extract/notarius.receipt.json` diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 08ef2c9..ab0de11 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -23,7 +23,7 @@ All stages are pending when this plan is created. | 5 | Confine recursive cleanup and replace sentinel locks | RSK-003 | Completed | | 6 | Harden API-key file acquisition | RSK-010 | Completed | | 7 | Bound and verify external result acquisition | RSK-013, TST-007 | Completed | -| 8 | Terminate owned subprocess trees | RSK-011 | Pending | +| 8 | Terminate owned subprocess trees | RSK-011 | Completed | | 9 | Redact and cap subprocess diagnostics | RSK-012 | Pending | | 10 | Confine publish archive reads | COR-005 | Pending | | 11 | Make manifest and run identity singular | COR-001, TST-006 | Pending | diff --git a/internal/adapters/subprocess/process_tree.go b/internal/adapters/subprocess/process_tree.go new file mode 100644 index 0000000..3fa140c --- /dev/null +++ b/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...) +} diff --git a/internal/adapters/subprocess/process_tree_supported_test.go b/internal/adapters/subprocess/process_tree_supported_test.go new file mode 100644 index 0000000..147239a --- /dev/null +++ b/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) + } +} diff --git a/internal/adapters/subprocess/process_tree_unix.go b/internal/adapters/subprocess/process_tree_unix.go new file mode 100644 index 0000000..b00ac35 --- /dev/null +++ b/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 +} diff --git a/internal/adapters/subprocess/process_tree_unsupported.go b/internal/adapters/subprocess/process_tree_unsupported.go new file mode 100644 index 0000000..f644320 --- /dev/null +++ b/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) +} diff --git a/internal/adapters/subprocess/process_tree_windows.go b/internal/adapters/subprocess/process_tree_windows.go new file mode 100644 index 0000000..b4e0f4b --- /dev/null +++ b/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) +} diff --git a/internal/adapters/subprocess/run.go b/internal/adapters/subprocess/run.go index c99a5c3..15c8eb6 100644 --- a/internal/adapters/subprocess/run.go +++ b/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 { diff --git a/internal/adapters/subprocess/run_test.go b/internal/adapters/subprocess/run_test.go index 42cc538..7fc2d87 100644 --- a/internal/adapters/subprocess/run_test.go +++ b/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) }