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

Dispose subprocess descendants after leader exit

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-11 03:22:21 +00:00
parent 80be8be4d6
commit 2545faef6c
8 changed files with 207 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
docs/internal/adapters.md
View File

@@ -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

4
docs/roadmap/implementation.md
View File

@@ -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

2
internal/adapters/subprocess/process_tree.go
View File

@@ -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) {

83
internal/adapters/subprocess/process_tree_supported_test.go
View File

@@ -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()

58
internal/adapters/subprocess/process_tree_unix.go
View File

@@ -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)
}
}

2
internal/adapters/subprocess/process_tree_windows.go
View File

@@ -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
}

9
internal/adapters/subprocess/run.go
View File

@@ -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.

50
internal/adapters/subprocess/run_test.go
View File

@@ -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
}