From 2545faef6caeed4c53ae3e1f76d840ad923ccc12 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 11 Aug 2026 03:22:21 +0000 Subject: [PATCH] Dispose subprocess descendants after leader exit --- docs/internal/adapters.md | 19 +++-- docs/roadmap/implementation.md | 4 +- internal/adapters/subprocess/process_tree.go | 2 +- .../subprocess/process_tree_supported_test.go | 83 +++++++++++++++++++ .../adapters/subprocess/process_tree_unix.go | 58 ++++++++++++- .../subprocess/process_tree_windows.go | 2 +- internal/adapters/subprocess/run.go | 9 +- internal/adapters/subprocess/run_test.go | 50 +++++++++++ 8 files changed, 207 insertions(+), 20 deletions(-) diff --git a/docs/internal/adapters.md b/docs/internal/adapters.md index e2f3daa..a5bc0d2 100644 --- a/docs/internal/adapters.md +++ b/docs/internal/adapters.md @@ -57,14 +57,17 @@ configured filesystem secrets before adapter initialization. - 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. Child environments contain only the execution baseline and - adapter-specified values; configured credentials are explicit sensitive - values. Stdout and stderr are redacted while streaming into separate 8 MiB - diagnostic captures; a bounded wait closes a stream retained by a departed - leader's descendant. Reaching either limit terminates the owned tree. - Unsupported platforms reject owned command execution. + kill-on-close job object on Windows. Every terminal path disposes of that + owned tree before returning. After a natural leader exit, Unix checks for + remaining group members and uses bounded graceful then forceful termination; + Windows closes the job so kill-on-close applies. Cancellation, deadlines, and + diagnostic limits use the same terminal disposal path without losing their + original result classification. Child environments contain only the execution + baseline and adapter-specified values; configured credentials are explicit + sensitive values. Stdout and stderr are redacted while streaming into separate + 8 MiB diagnostic captures; a bounded wait closes a stream retained by a + departed leader's descendant. Unsupported platforms reject owned command + execution. ## Implementation And Tests diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 379d85d..b381377 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -55,7 +55,7 @@ the full audit or any audit line range. | 31 | Enforced repository-wide CI/release validation and streamlined redundant test matrices. | TST-012, TST-015 | Completed | | 32 | Reconciled lifecycle/analyze documentation and closed the original audit traceability inventory. | COM-002, COM-005 | Completed | | 33 | Confine diagnostic log destinations and eliminate pathname-based tail reads. | Follow-up review | Completed | -| 34 | Dispose of owned subprocess descendants after natural leader exit. | Follow-up review | Pending | +| 34 | Dispose of owned subprocess descendants after natural leader exit. | Follow-up review | Completed | | 35 | Bound remote current-state and lock control-plane reads. | Follow-up review | Pending | Completed stages must not be reimplemented wholesale. A pending stage may adjust @@ -243,7 +243,7 @@ suite repeatedly, shuffled, and under `-race`. Cross-build the affected packages for Linux, macOS, and Windows; run native platform tests only where runners are actually available. -**Status:** Pending. +**Status:** Completed. ## Stage 35 — Bound remote current-state and lock control-plane reads diff --git a/internal/adapters/subprocess/process_tree.go b/internal/adapters/subprocess/process_tree.go index 453ed96..4f7ad37 100644 --- a/internal/adapters/subprocess/process_tree.go +++ b/internal/adapters/subprocess/process_tree.go @@ -19,7 +19,7 @@ type ownedProcessTree interface { Start(*exec.Cmd) error TerminateGracefully() error TerminateForcefully() error - Close() error + Dispose() error } func waitForOwnedCommand(ctx context.Context, tree ownedProcessTree, waitCh <-chan error, captureLimits <-chan *captureLimitError) (waitErr, ctxErr error, captureLimit *captureLimitError, cleanupErr error) { diff --git a/internal/adapters/subprocess/process_tree_supported_test.go b/internal/adapters/subprocess/process_tree_supported_test.go index 0f0e70c..dd09826 100644 --- a/internal/adapters/subprocess/process_tree_supported_test.go +++ b/internal/adapters/subprocess/process_tree_supported_test.go @@ -6,6 +6,7 @@ import ( "context" "errors" "os" + "os/exec" "path/filepath" "strings" "testing" @@ -74,6 +75,63 @@ func TestRunCaptureLimitTerminatesProcessTree(t *testing.T) { assertDescendantDidNotSurvive(t, sentinelPath) } +func TestRunDisposesDescendantsAfterLeaderExit(t *testing.T) { + tests := []struct { + name string + mode string + wantExitCode int + wantWaitDelay bool + ignoreTerm bool + }{ + {name: "success retaining streams", mode: "leader-exit-retained", wantExitCode: 0, wantWaitDelay: true}, + {name: "success redirecting streams", mode: "leader-exit-redirected", wantExitCode: 0}, + {name: "failed leader", mode: "leader-fail-redirected", wantExitCode: 9, ignoreTerm: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req, sentinelPath, releasePath := leaderExitRequest(t, tt.mode) + if tt.ignoreTerm { + req.EnvOverrides["SUBPROCESS_HELPER_IGNORE_TERM"] = "1" + } + outcomes := make(chan runOutcome, 1) + go func() { + result, err := Run(context.Background(), req) + outcomes <- runOutcome{result: result, err: err} + }() + + var outcome runOutcome + select { + case outcome = <-outcomes: + case <-time.After(6 * time.Second): + t.Fatal("Run() did not complete bounded owned-tree disposal") + } + if outcome.result.ExitCode != tt.wantExitCode { + t.Fatalf("ExitCode = %d, want %d", outcome.result.ExitCode, tt.wantExitCode) + } + if tt.wantWaitDelay { + if !errors.Is(outcome.err, exec.ErrWaitDelay) { + t.Fatalf("error = %v, want exec.ErrWaitDelay", outcome.err) + } + } else if tt.wantExitCode == 0 && outcome.err != nil { + t.Fatalf("Run() error = %v, want nil", outcome.err) + } else if tt.wantExitCode != 0 { + var exitErr *exec.ExitError + if !errors.As(outcome.err, &exitErr) || exitErr.ExitCode() != tt.wantExitCode { + t.Fatalf("error = %v, want exit code %d", outcome.err, tt.wantExitCode) + } + } + if _, err := os.Stat(req.EnvOverrides["SUBPROCESS_HELPER_READY_PATH"]); err != nil { + t.Fatalf("descendant readiness file: %v", err) + } + if err := os.WriteFile(releasePath, []byte("release"), 0o600); err != nil { + t.Fatalf("WriteFile(release) error = %v", err) + } + assertDescendantDidNotSurvive(t, sentinelPath) + }) + } +} + type runOutcome struct { result RunResult err error @@ -102,6 +160,31 @@ func processTreeRequest(t *testing.T) (RunRequest, string) { }, sentinelPath } +func leaderExitRequest(t *testing.T, mode string) (RunRequest, string, 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") + releasePath := filepath.Join(dir, "release") + sentinelPath := filepath.Join(dir, "descendant-survived") + return RunRequest{ + Executable: executable, + Args: []string{"-test.run=^TestSubprocessHelper$", "--", mode}, + EnvOverrides: map[string]string{ + "GO_WANT_SUBPROCESS_HELPER": "1", + "SUBPROCESS_HELPER_READY_PATH": readyPath, + "SUBPROCESS_HELPER_RELEASE_PATH": releasePath, + "SUBPROCESS_HELPER_SENTINEL_PATH": sentinelPath, + }, + StdoutLogPath: filepath.Join(dir, "stdout.log"), + StderrLogPath: filepath.Join(dir, "stderr.log"), + }, sentinelPath, releasePath +} + func awaitHelperReady(t *testing.T, readyPath string) { t.Helper() diff --git a/internal/adapters/subprocess/process_tree_unix.go b/internal/adapters/subprocess/process_tree_unix.go index b00ac35..cd7dd45 100644 --- a/internal/adapters/subprocess/process_tree_unix.go +++ b/internal/adapters/subprocess/process_tree_unix.go @@ -4,11 +4,15 @@ package subprocess import ( "errors" + "fmt" "os" "os/exec" "syscall" + "time" ) +const processGroupPollInterval = 10 * time.Millisecond + type unixProcessTree struct { processGroupID int } @@ -34,8 +38,26 @@ func (tree *unixProcessTree) TerminateForcefully() error { return tree.signal(syscall.SIGKILL) } -func (tree *unixProcessTree) Close() error { - return nil +func (tree *unixProcessTree) Dispose() error { + hasMembers, err := tree.hasMembers() + if err != nil || !hasMembers { + return err + } + + cleanupErr := tree.TerminateGracefully() + empty, waitErr := tree.waitUntilEmpty(gracefulTerminationWait) + cleanupErr = joinErrors(cleanupErr, waitErr) + if empty { + return cleanupErr + } + + cleanupErr = joinErrors(cleanupErr, tree.TerminateForcefully()) + empty, waitErr = tree.waitUntilEmpty(forcefulTerminationWait) + cleanupErr = joinErrors(cleanupErr, waitErr) + if !empty { + cleanupErr = joinErrors(cleanupErr, fmt.Errorf("owned subprocess group did not exit within %s after forceful termination", forcefulTerminationWait)) + } + return cleanupErr } func (tree *unixProcessTree) signal(signal syscall.Signal) error { @@ -48,3 +70,35 @@ func (tree *unixProcessTree) signal(signal syscall.Signal) error { } return err } + +func (tree *unixProcessTree) hasMembers() (bool, error) { + if tree.processGroupID <= 0 { + return false, nil + } + err := syscall.Kill(-tree.processGroupID, 0) + if err == nil || errors.Is(err, syscall.EPERM) { + return true, nil + } + if errors.Is(err, syscall.ESRCH) || errors.Is(err, os.ErrProcessDone) { + return false, nil + } + return false, fmt.Errorf("inspect owned subprocess group: %w", err) +} + +func (tree *unixProcessTree) waitUntilEmpty(timeout time.Duration) (bool, error) { + deadline := time.Now().Add(timeout) + for { + hasMembers, err := tree.hasMembers() + if err != nil || !hasMembers { + return !hasMembers, err + } + remaining := time.Until(deadline) + if remaining <= 0 { + return false, nil + } + if remaining > processGroupPollInterval { + remaining = processGroupPollInterval + } + time.Sleep(remaining) + } +} diff --git a/internal/adapters/subprocess/process_tree_windows.go b/internal/adapters/subprocess/process_tree_windows.go index b4e0f4b..e3fb6e4 100644 --- a/internal/adapters/subprocess/process_tree_windows.go +++ b/internal/adapters/subprocess/process_tree_windows.go @@ -105,7 +105,7 @@ func (tree *windowsProcessTree) TerminateForcefully() error { return tree.terminate() } -func (tree *windowsProcessTree) Close() error { +func (tree *windowsProcessTree) Dispose() error { if tree.job == 0 { return nil } diff --git a/internal/adapters/subprocess/run.go b/internal/adapters/subprocess/run.go index 038f3a8..4709684 100644 --- a/internal/adapters/subprocess/run.go +++ b/internal/adapters/subprocess/run.go @@ -98,14 +98,11 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) { go func() { waitCh <- cmd.Wait() }() waitErr, ctxErr, captureLimit, cleanupErr := waitForOwnedCommand(runCtx, tree, waitCh, logs.Limits()) + cleanupErr = joinErrors(cleanupErr, tree.Dispose()) cleanupErr = joinErrors(cleanupErr, logs.Flush()) if captureLimit == nil { captureLimit = logs.Limit() - if captureLimit != nil { - cleanupErr = joinErrors(cleanupErr, tree.TerminateForcefully()) - } } - cleanupErr = joinErrors(cleanupErr, tree.Close()) result.CompletedAt = time.Now().UTC() result.Duration = result.CompletedAt.Sub(result.StartedAt) if cmd.ProcessState != nil { @@ -139,13 +136,13 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) { 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) + return result, fmt.Errorf("command failed with exit code %d (%s): %w", exitErr.ExitCode(), diagnostics, joinErrors(waitErr, cleanupErr)) } 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) + return result, fmt.Errorf("command failed to run (%s): %w", diagnostics, joinErrors(waitErr, cleanupErr)) } // WriteYAMLAtomic marshals value as YAML and atomically writes it to path. diff --git a/internal/adapters/subprocess/run_test.go b/internal/adapters/subprocess/run_test.go index 87675ee..5512c59 100644 --- a/internal/adapters/subprocess/run_test.go +++ b/internal/adapters/subprocess/run_test.go @@ -6,10 +6,12 @@ import ( "errors" "os" "os/exec" + "os/signal" "path/filepath" "runtime" "strconv" "strings" + "syscall" "testing" "time" @@ -747,6 +749,41 @@ func TestSubprocessHelper(t *testing.T) { } time.Sleep(10 * time.Second) os.Exit(0) + case "leader-exit-retained", "leader-exit-redirected", "leader-fail-redirected": + descendant := exec.Command(os.Args[0], "-test.run=^TestSubprocessHelper$", "--", "descendant-after-release") + descendant.Env = append(os.Environ(), "GO_WANT_SUBPROCESS_HELPER=1") + if mode == "leader-exit-retained" { + descendant.Stdout = os.Stdout + descendant.Stderr = os.Stderr + } + if err := descendant.Start(); err != nil { + os.Exit(3) + } + if !helperFileAppeared(os.Getenv("SUBPROCESS_HELPER_READY_PATH"), 2*time.Second) { + os.Exit(4) + } + if mode == "leader-fail-redirected" { + os.Exit(9) + } + os.Exit(0) + case "descendant-after-release": + if os.Getenv("SUBPROCESS_HELPER_IGNORE_TERM") == "1" { + signal.Ignore(syscall.SIGTERM) + } + if err := os.WriteFile(os.Getenv("SUBPROCESS_HELPER_READY_PATH"), []byte("ready"), 0o600); err != nil { + os.Exit(4) + } + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + if _, err := os.Stat(os.Getenv("SUBPROCESS_HELPER_RELEASE_PATH")); err == nil { + _ = os.WriteFile(os.Getenv("SUBPROCESS_HELPER_SENTINEL_PATH"), []byte("survived"), 0o600) + os.Exit(0) + } else if !errors.Is(err, os.ErrNotExist) { + os.Exit(5) + } + time.Sleep(10 * time.Millisecond) + } + os.Exit(0) case "descendant": time.Sleep(500 * time.Millisecond) _ = os.WriteFile(os.Getenv("SUBPROCESS_HELPER_SENTINEL_PATH"), []byte("survived"), 0o600) @@ -770,3 +807,16 @@ func waitForHelperFile(t *testing.T, path string) { } t.Fatalf("helper file %q was not created", path) } + +func helperFileAppeared(path string, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if _, err := os.Stat(path); err == nil { + return true + } else if !errors.Is(err, os.ErrNotExist) { + return false + } + time.Sleep(10 * time.Millisecond) + } + return false +}