From 6af0ae2b9f33afd54d12f18610eee01efe788ac4 Mon Sep 17 00:00:00 2001 From: Altay Date: Sat, 5 Sep 2026 15:03:43 +0300 Subject: [PATCH] fix(slopmachine): bound verification process cleanup --- cli/slopmachine/cmd/slopmachine/main.go | 75 ++++--- cli/slopmachine/cmd/slopmachine/main_test.go | 19 +- .../cmd/slopmachine/process_darwin.go | 104 ++++++++++ .../cmd/slopmachine/process_darwin_test.go | 33 +++ .../cmd/slopmachine/process_linux.go | 38 ++++ .../cmd/slopmachine/process_linux_test.go | 21 ++ .../cmd/slopmachine/process_test.go | 188 ++++++++++++++++++ cli/slopmachine/docs/AGENT_INTERFACE.md | 6 + cli/slopmachine/go.mod | 2 +- 9 files changed, 433 insertions(+), 53 deletions(-) create mode 100644 cli/slopmachine/cmd/slopmachine/process_darwin.go create mode 100644 cli/slopmachine/cmd/slopmachine/process_darwin_test.go create mode 100644 cli/slopmachine/cmd/slopmachine/process_linux.go create mode 100644 cli/slopmachine/cmd/slopmachine/process_linux_test.go create mode 100644 cli/slopmachine/cmd/slopmachine/process_test.go diff --git a/cli/slopmachine/cmd/slopmachine/main.go b/cli/slopmachine/cmd/slopmachine/main.go index b0c5391..ba11b96 100644 --- a/cli/slopmachine/cmd/slopmachine/main.go +++ b/cli/slopmachine/cmd/slopmachine/main.go @@ -809,7 +809,10 @@ func cmdVerify(st *store.Store, args []string, opts runOptions) int { err = errVerificationCommandCancelled } if err != nil { - cancelCode := 130 + cancelCode := 1 + if errors.Is(err, errVerificationCommandCancelled) { + cancelCode = 130 + } if unixSignal, ok := received.(syscall.Signal); ok { cancelCode = 128 + int(unixSignal) } @@ -1788,6 +1791,7 @@ var errVerificationCommandCancelled = errors.New("verification command cancelled func runShell(ctx context.Context, command string, jsonOut bool) (int, string, error) { cmd := exec.Command("sh", "-c", command) cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.WaitDelay = 2 * time.Second stdoutDigest := newOutputDigester() stderrDigest := newOutputDigester() if jsonOut { @@ -1800,41 +1804,37 @@ func runShell(ctx context.Context, command string, jsonOut bool) (int, string, e if err := cmd.Start(); err != nil { return 1, digestOutputs(stdoutDigest, stderrDigest), nil } - waited := make(chan error, 1) - go func() { waited <- cmd.Wait() }() - - select { - case err := <-waited: - return shellExitCode(err), digestOutputs(stdoutDigest, stderrDigest), nil - case <-ctx.Done(): - select { - case err := <-waited: - return shellExitCode(err), digestOutputs(stdoutDigest, stderrDigest), nil - default: - } - } - + // Keep the exited leader unreaped until cleanup is complete: its PID + // reserves the process-group identity while descendants are signalled. pid := cmd.Process.Pid - _ = signalShellGroup(pid, syscall.SIGTERM) - timer := time.NewTimer(verificationTerminationGrace) - ticker := time.NewTicker(10 * time.Millisecond) - defer timer.Stop() - defer ticker.Stop() - waitComplete := false - for { - if waitComplete && !shellGroupAlive(pid) { - return 130, digestOutputs(stdoutDigest, stderrDigest), errVerificationCommandCancelled - } - select { - case <-waited: - waitComplete = true - case <-timer.C: - if shellGroupAlive(pid) { - _ = signalShellGroup(pid, syscall.SIGKILL) - } - case <-ticker.C: - } - } + watchErr := waitForExit(ctx, pid) + cancelled := errors.Is(watchErr, context.Canceled) || errors.Is(watchErr, context.DeadlineExceeded) + var cleanupErr error + if cancelled { + cleanupErr = signalShellGroup(pid, syscall.SIGTERM) + // The leader remains reserved for the whole grace period, including + // when it exits before a descendant finishes its TERM handler. + time.Sleep(verificationTerminationGrace) + } + killErr := signalShellGroup(pid, syscall.SIGKILL) + if ignoreCleanupErrorAfterExit(pid, killErr) { + killErr = nil + } + cleanupErr = errors.Join(cleanupErr, killErr) + if watchErr != nil { + if err := cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { + cleanupErr = errors.Join(cleanupErr, err) + } + } + waitErr := cmd.Wait() + digest := digestOutputs(stdoutDigest, stderrDigest) + if cancelled || ctx.Err() != nil { + return 130, digest, errors.Join(errVerificationCommandCancelled, cleanupErr) + } + if watchErr != nil || cleanupErr != nil { + return 1, digest, fmt.Errorf("verification process cleanup: %w", errors.Join(watchErr, cleanupErr)) + } + return shellExitCode(waitErr), digest, nil } func shellExitCode(err error) int { @@ -1856,11 +1856,6 @@ func signalShellGroup(pid int, sig syscall.Signal) error { return err } -func shellGroupAlive(pid int) bool { - err := syscall.Kill(-pid, 0) - return err == nil || errors.Is(err, syscall.EPERM) -} - func randomID() (string, error) { var b [4]byte if _, err := rand.Read(b[:]); err != nil { diff --git a/cli/slopmachine/cmd/slopmachine/main_test.go b/cli/slopmachine/cmd/slopmachine/main_test.go index 7f52a03..250e8b9 100644 --- a/cli/slopmachine/cmd/slopmachine/main_test.go +++ b/cli/slopmachine/cmd/slopmachine/main_test.go @@ -13,7 +13,6 @@ import ( "path/filepath" "strconv" "strings" - "syscall" "testing" "time" @@ -26,29 +25,30 @@ import ( func TestRunShellCancellationReapsProcessGroup(t *testing.T) { pidFile := filepath.Join(t.TempDir(), "pids") ctx, cancel := context.WithCancel(context.Background()) + defer cancel() type result struct { code int err error } done := make(chan result, 1) - command := fmt.Sprintf(`trap '' TERM; sh -c 'trap "" TERM; printf "%%s %%s\n" "$PPID" "$$" > "$1"; while :; do sleep 1; done' child %q & wait`, pidFile) + command := fmt.Sprintf(`trap '' TERM; sh -c 'trap "" TERM; sleep 30 & printf "%%s %%s %%s\n" "$PPID" "$$" "$!" > "$1"; wait' child %q & wait`, pidFile) go func() { code, _, err := runShell(ctx, command, true) done <- result{code: code, err: err} }() - var groupPID, childPID int + var groupPID, childPID, grandchildPID int deadline := time.Now().Add(3 * time.Second) for time.Now().Before(deadline) { contents, err := os.ReadFile(pidFile) if err == nil { - if n, _ := fmt.Sscanf(string(contents), "%d %d", &groupPID, &childPID); n == 2 { + if n, _ := fmt.Sscanf(string(contents), "%d %d %d", &groupPID, &childPID, &grandchildPID); n == 3 { break } } time.Sleep(10 * time.Millisecond) } - if groupPID == 0 || childPID == 0 { + if groupPID == 0 || childPID == 0 || grandchildPID == 0 { t.Fatal("verification descendants did not start") } cancel() @@ -61,13 +61,8 @@ func TestRunShellCancellationReapsProcessGroup(t *testing.T) { case <-time.After(3 * time.Second): t.Fatal("runShell did not finish cancellation") } - if shellGroupAlive(groupPID) { - t.Fatalf("process group %d remains alive", groupPID) - } - for _, pid := range []int{groupPID, childPID} { - if err := syscall.Kill(pid, 0); !errors.Is(err, syscall.ESRCH) { - t.Fatalf("process %d remains: %v", pid, err) - } + for _, pid := range []int{groupPID, childPID, grandchildPID} { + waitVerificationProcessTerminated(t, pid) } } diff --git a/cli/slopmachine/cmd/slopmachine/process_darwin.go b/cli/slopmachine/cmd/slopmachine/process_darwin.go new file mode 100644 index 0000000..a146312 --- /dev/null +++ b/cli/slopmachine/cmd/slopmachine/process_darwin.go @@ -0,0 +1,104 @@ +//go:build darwin + +package main + +import ( + "context" + "errors" + "time" + + "golang.org/x/sys/unix" +) + +func waitForExit(ctx context.Context, pid int) error { + queue, err := unix.Kqueue() + if err != nil { + return err + } + defer unix.Close(queue) + change := unix.Kevent_t{ + Ident: uint64(pid), + Filter: unix.EVFILT_PROC, + Flags: unix.EV_ADD | unix.EV_ENABLE | unix.EV_ONESHOT, + Fflags: unix.NOTE_EXIT, + } + if _, err := unix.Kevent(queue, []unix.Kevent_t{change}, nil, nil); err != nil { + if errors.Is(err, unix.ESRCH) { + return nil + } + return err + } + events := make([]unix.Kevent_t, 1) + for { + timeout := unix.NsecToTimespec((10 * time.Millisecond).Nanoseconds()) + count, err := unix.Kevent(queue, nil, events, &timeout) + if errors.Is(err, unix.EINTR) { + continue + } + if err != nil { + return err + } + if count != 0 { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + } +} + +func ignoreCleanupErrorAfterExit(leaderPID int, err error) bool { + if !errors.Is(err, unix.EPERM) { + return false + } + for range 10 { + safe, pending := cleanupStateAfterExit(leaderPID) + if safe { + return true + } + if !pending { + return false + } + time.Sleep(time.Millisecond) + } + return false +} + +func cleanupStateAfterExit(leaderPID int) (safe, pending bool) { + processes, queryErr := unix.SysctlKinfoProcSlice("kern.proc.pgrp", leaderPID) + if queryErr != nil { + return false, false + } + if containsOnlyZombieLeader(processes, leaderPID) { + return true, false + } + if len(processes) == 1 && int(processes[0].Proc.P_pid) == leaderPID { + return false, true + } + if len(processes) != 0 { + return false, false + } + leader, queryErr := unix.SysctlKinfoProc("kern.proc.pid", leaderPID) + if queryErr != nil { + return false, false + } + if isZombieLeader(leader, leaderPID) { + return true, false + } + return false, int(leader.Proc.P_pid) == leaderPID +} + +func containsOnlyZombieLeader(processes []unix.KinfoProc, leaderPID int) bool { + return len(processes) == 1 && + isZombieLeader(&processes[0], leaderPID) +} + +func isZombieLeader(process *unix.KinfoProc, leaderPID int) bool { + // Darwin's SZOMB value is 5 in sys/proc.h but is not exported by x/sys. + const zombieState = 5 + return process != nil && + int(process.Proc.P_pid) == leaderPID && + process.Proc.P_stat == zombieState +} diff --git a/cli/slopmachine/cmd/slopmachine/process_darwin_test.go b/cli/slopmachine/cmd/slopmachine/process_darwin_test.go new file mode 100644 index 0000000..65be5f8 --- /dev/null +++ b/cli/slopmachine/cmd/slopmachine/process_darwin_test.go @@ -0,0 +1,33 @@ +package main + +import ( + "os/exec" + "syscall" + "testing" + + "golang.org/x/sys/unix" +) + +func TestCleanupDoesNotIgnoreLiveLeader(t *testing.T) { + command := exec.Command("sleep", "30") + command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := command.Start(); err != nil { + t.Fatal(err) + } + defer func() { + _ = command.Process.Kill() + _ = command.Wait() + }() + pid := command.Process.Pid + if safe, pending := cleanupStateAfterExit(pid); safe || !pending { + t.Fatalf("live leader cleanup state = (%t, %t)", safe, pending) + } + if ignoreCleanupErrorAfterExit(pid, unix.EPERM) { + t.Fatal("permission error ignored for a live process group") + } +} + +func verificationProcessZombie(pid int) bool { + process, err := unix.SysctlKinfoProc("kern.proc.pid", pid) + return err == nil && isZombieLeader(process, pid) +} diff --git a/cli/slopmachine/cmd/slopmachine/process_linux.go b/cli/slopmachine/cmd/slopmachine/process_linux.go new file mode 100644 index 0000000..4a34b6e --- /dev/null +++ b/cli/slopmachine/cmd/slopmachine/process_linux.go @@ -0,0 +1,38 @@ +//go:build linux + +package main + +import ( + "context" + "errors" + "time" + + "golang.org/x/sys/unix" +) + +func waitForExit(ctx context.Context, pid int) error { + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for { + var info unix.Siginfo + err := unix.Waitid(unix.P_PID, pid, &info, unix.WEXITED|unix.WNOHANG|unix.WNOWAIT, nil) + if errors.Is(err, unix.EINTR) { + continue + } + if err != nil { + return err + } + if info.Signo != 0 { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + +func ignoreCleanupErrorAfterExit(_ int, _ error) bool { + return false +} diff --git a/cli/slopmachine/cmd/slopmachine/process_linux_test.go b/cli/slopmachine/cmd/slopmachine/process_linux_test.go new file mode 100644 index 0000000..5068859 --- /dev/null +++ b/cli/slopmachine/cmd/slopmachine/process_linux_test.go @@ -0,0 +1,21 @@ +package main + +import ( + "fmt" + "os" + "strings" +) + +func verificationProcessZombie(pid int) bool { + data, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)) + if err != nil { + return false + } + // The parenthesized command name may itself contain spaces and ')'. + end := strings.LastIndexByte(string(data), ')') + if end < 0 { + return false + } + fields := strings.Fields(string(data)[end+1:]) + return len(fields) > 0 && fields[0] == "Z" +} diff --git a/cli/slopmachine/cmd/slopmachine/process_test.go b/cli/slopmachine/cmd/slopmachine/process_test.go new file mode 100644 index 0000000..8a94438 --- /dev/null +++ b/cli/slopmachine/cmd/slopmachine/process_test.go @@ -0,0 +1,188 @@ +package main + +import ( + "context" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + "syscall" + "testing" + "time" +) + +func TestVerificationPipeHelper(t *testing.T) { + path := os.Getenv("SLOPMACHINE_PIPE_HELPER") + if path == "" { + return + } + if _, err := syscall.Setsid(); err != nil { + os.Exit(2) + } + fmt.Fprint(os.Stdout, "helper stdout") + fmt.Fprint(os.Stderr, "helper stderr") + if err := os.WriteFile(path, []byte(strconv.Itoa(os.Getpid())), 0600); err != nil { + os.Exit(2) + } + time.Sleep(30 * time.Second) + os.Exit(0) +} + +func TestRunShellDetachedPipes(t *testing.T) { + for _, cancelRun := range []bool{false, true} { + t.Run(fmt.Sprintf("cancel=%t", cancelRun), func(t *testing.T) { + path := filepath.Join(t.TempDir(), "pid") + executable, err := os.Executable() + if err != nil { + t.Fatal(err) + } + command := fmt.Sprintf(`SLOPMACHINE_PIPE_HELPER=%q %q -test.run='^TestVerificationPipeHelper$' & while [ ! -s %q ]; do sleep 0.01; done`, path, executable, path) + if cancelRun { + command += "; wait" + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + type result struct { + code int + digest string + err error + } + done := make(chan result, 1) + go func() { code, digest, err := runShell(ctx, command, true); done <- result{code, digest, err} }() + pid := waitVerificationPID(t, path) + defer syscall.Kill(pid, syscall.SIGKILL) + if cancelRun { + cancel() + } + select { + case got := <-done: + if cancelRun { + if got.code != 130 || !errors.Is(got.err, errVerificationCommandCancelled) { + t.Fatalf("cancellation = %+v", got) + } + } else if got.code != 1 || got.err != nil { + t.Fatalf("incomplete output = %+v", got) + } + stdout, stderr := newOutputDigester(), newOutputDigester() + stdout.Write([]byte("helper stdout")) + stderr.Write([]byte("helper stderr")) + if want := digestOutputs(stdout, stderr); got.digest != want { + t.Fatalf("digest = %s, want %s", got.digest, want) + } + case <-time.After(5 * time.Second): + t.Fatal("inherited pipes blocked verification") + } + }) + } +} + +func waitVerificationPID(t *testing.T, path string) int { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if data, err := os.ReadFile(path); err == nil { + if pid, err := strconv.Atoi(string(data)); err == nil && pid > 0 { + return pid + } + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("helper did not start") + return 0 +} + +func TestRunShellNormalExitCleansGroup(t *testing.T) { + for _, redirect := range []string{"", ">/dev/null 2>&1"} { + t.Run(redirect, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "pid") + command := fmt.Sprintf(`sleep 30 %s & printf '%%s' "$!" > %q; exit 7`, redirect, path) + code, _, err := runShell(context.Background(), command, true) + pid := waitVerificationPID(t, path) + defer syscall.Kill(pid, syscall.SIGKILL) + if code != 7 || err != nil { + t.Fatalf("exit = (%d,%v)", code, err) + } + waitVerificationProcessTerminated(t, pid) + }) + } +} + +func TestRunShellGracefulCancellation(t *testing.T) { + path := filepath.Join(t.TempDir(), "pid") + handled := filepath.Join(t.TempDir(), "handled") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + command := fmt.Sprintf(`trap 'printf done > %q; exit 0' TERM; printf '%%s' "$$" > %q; while :; do sleep 0.05; done`, handled, path) + go func() { _, _, err := runShell(ctx, command, true); done <- err }() + pid := waitVerificationPID(t, path) + defer syscall.Kill(-pid, syscall.SIGKILL) + cancel() + select { + case err := <-done: + if !errors.Is(err, errVerificationCommandCancelled) { + t.Fatal(err) + } + case <-time.After(4 * time.Second): + t.Fatal("cancellation blocked") + } + if data, err := os.ReadFile(handled); err != nil || string(data) != "done" { + t.Fatalf("TERM handler = %q,%v", data, err) + } +} + +// A killed orphan can remain a zombie until init reaps it. Check kernel +// termination state instead of making fixture success depend on init timing. +func verificationProcessTerminated(pid int) bool { + if errors.Is(syscall.Kill(pid, 0), syscall.ESRCH) { + return true + } + return verificationProcessZombie(pid) +} + +func waitVerificationProcessTerminated(t *testing.T, pid int) { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + if verificationProcessTerminated(pid) { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("owned process %d remains live", pid) +} + +func TestVerificationProcessTerminationStates(t *testing.T) { + command := exec.Command("sleep", "30") + if err := command.Start(); err != nil { + t.Fatal(err) + } + defer func() { _ = command.Process.Kill(); _ = command.Wait() }() + pid := command.Process.Pid + if verificationProcessTerminated(pid) { + t.Fatal("live child accepted as terminated") + } + if err := command.Process.Kill(); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + if err := waitForExit(ctx, pid); err != nil { + t.Fatal(err) + } + // WNOWAIT/kqueue preserves this exact child's zombie for the assertion. + // Darwin may publish NOTE_EXIT before its zombie state becomes visible. + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) && !verificationProcessZombie(pid) { + time.Sleep(time.Millisecond) + } + if !verificationProcessZombie(pid) || !verificationProcessTerminated(pid) { + t.Fatal("unreaped zombie rejected as terminated") + } + _ = command.Wait() + if !verificationProcessTerminated(pid) { + t.Fatal("reaped child rejected as terminated") + } +} diff --git a/cli/slopmachine/docs/AGENT_INTERFACE.md b/cli/slopmachine/docs/AGENT_INTERFACE.md index 34b2325..14ffd5a 100644 --- a/cli/slopmachine/docs/AGENT_INTERFACE.md +++ b/cli/slopmachine/docs/AGENT_INTERFACE.md @@ -107,6 +107,12 @@ hashed stdout and stderr streams. The final SHA-256 hashes the labeled text `stdout=sha256:\nstderr=sha256:`. JSON mode streams both child channels to stderr so stdout remains machine-readable; plain mode preserves the child's stdout and stderr. +Verification owns its shell process group and terminates remaining group members +before reaping the shell. Cancellation gives them 500 ms after SIGTERM before +SIGKILL. Output pipes drain for at most two seconds after shell exit; an escaped +helper holding a pipe causes failed verification when that limit expires. Helpers +that create a separate session are outside the owned group. + A non-dry `verify --cmd` intentionally invokes the local shell; never construct it from untrusted text. diff --git a/cli/slopmachine/go.mod b/cli/slopmachine/go.mod index 8854eb7..3edf48a 100644 --- a/cli/slopmachine/go.mod +++ b/cli/slopmachine/go.mod @@ -7,6 +7,7 @@ toolchain go1.26.6 require ( github.com/mattn/go-isatty v0.0.24 github.com/uinaf/ffss/cli/lib v0.1.0 + golang.org/x/sys v0.47.0 modernc.org/sqlite v1.56.0 ) @@ -15,7 +16,6 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect - golang.org/x/sys v0.47.0 // indirect modernc.org/libc v1.74.4 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect