diff --git a/cmd/environments/sync.go b/cmd/environments/sync.go index 2e6c5dc2567..35256c8d149 100644 --- a/cmd/environments/sync.go +++ b/cmd/environments/sync.go @@ -1,8 +1,11 @@ package environments import ( + "context" "os" + "os/signal" "path/filepath" + "syscall" "github.com/databricks/cli/cmd/root" "github.com/databricks/cli/libs/cmdctx" @@ -51,9 +54,55 @@ func addComputeFlags(cmd *cobra.Command) { // consumer relies on, instead of a bare pre-RunE Cobra error. } +// watchInterruptSignals cancels ctx on the first SIGINT (Ctrl-C) or SIGTERM (how +// a supervisor, CI timeout, or VS Code stops the child), which propagates to the +// uv subprocesses the pipeline spawns so they are reaped instead of orphaned +// mid-provision. The CLI root installs no signal handler of its own. +// +// The returned stop function uninstalls the handler and joins the goroutine; the +// caller must defer it. +// +// The handler must give the *second* signal back to the OS. signal.Notify (like +// signal.NotifyContext, which wraps it) disables the default disposition for +// SIGINT/SIGTERM for as long as the channel stays registered, so without the +// signal.Stop below a second Ctrl-C is merely buffered and dropped: the user +// would have no way to abort during the process group's SIGKILL grace window. +// Stopping the relay as soon as the first signal lands restores SIG_DFL, so a +// second signal terminates the CLI immediately. That matters more here than in +// most commands because WithProcessGroup moves uv out of the foreground process +// group, so the tty no longer delivers Ctrl-C to it directly — this handler is +// the only delivery path. +func watchInterruptSignals(ctx context.Context, cancel context.CancelFunc) func() { + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM) + + done := make(chan struct{}) + go func() { + defer close(done) + // Selecting on ctx.Done() too lets the goroutine exit on the normal (no + // signal) path rather than blocking on sigCh for the rest of the process: + // signal.Stop unregisters the channel but never closes it. + select { + case <-sigCh: + signal.Stop(sigCh) + cancel() + case <-ctx.Done(): + } + }() + + return func() { + signal.Stop(sigCh) + // Wake the goroutine in case neither sigCh nor ctx.Done has fired. + cancel() + <-done + } +} + // runPipeline builds and runs the setup-local Pipeline. func runPipeline(cmd *cobra.Command) error { - ctx := cmd.Context() + ctx, cancel := context.WithCancel(cmd.Context()) + defer cancel() + defer watchInterruptSignals(ctx, cancel)() cluster, _ := cmd.Flags().GetString("cluster-id") clusterName, _ := cmd.Flags().GetString("cluster-name") diff --git a/cmd/environments/sync_signal_test.go b/cmd/environments/sync_signal_test.go new file mode 100644 index 00000000000..59477bccd67 --- /dev/null +++ b/cmd/environments/sync_signal_test.go @@ -0,0 +1,131 @@ +//go:build unix + +// The second-signal escape hatch is a Unix signal-delivery behavior: the test +// re-execs itself and sends SIGINT/SIGTERM, which os/signal does not support on +// Windows (the CLI's own signal handling there is likewise a no-op path). +package environments + +import ( + "bufio" + "context" + "os" + "os/exec" + "strings" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The second-signal behavior can only be observed in a real process: signal +// dispositions are process-wide, and the test binary's own handlers would mask +// them. So the test re-executes itself as a child (Go resets dispositions to +// SIG_DFL across exec, unlike a shell background job, which inherits SIGINT as +// SIG_IGN and would silently invalidate the result). +const signalChildEnv = "TEST_ENVIRONMENTS_SIGNAL_CHILD" + +// TestMain runs the signal-handler-under-test when re-executed as the child. +func TestMain(m *testing.M) { + if os.Getenv(signalChildEnv) == "" { + os.Exit(m.Run()) + } + + // TestMain has no *testing.T, so t.Context() is unavailable here. + //nolint:gocritic + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + stop := watchInterruptSignals(ctx, cancel) + defer stop() + + // Stand in for the pipeline still draining the uv process group during the + // SIGKILL grace window — the interval in which the user needs the hatch. + os.Stdout.WriteString("READY\n") + <-ctx.Done() + os.Stdout.WriteString("CANCELLED\n") + time.Sleep(30 * time.Second) + os.Stdout.WriteString("SURVIVED\n") + os.Exit(0) +} + +// TestWatchInterruptSignalsSecondSignalStillKills is the regression test for the +// escape hatch: signal.Notify disables the default SIGINT/SIGTERM disposition +// process-wide, so a handler that only relays the first signal leaves the user +// unable to abort. The first signal must cancel the context, and the second must +// terminate the process outright. +func TestWatchInterruptSignalsSecondSignalStillKills(t *testing.T) { + for _, sig := range []syscall.Signal{syscall.SIGINT, syscall.SIGTERM} { + t.Run(sig.String(), func(t *testing.T) { + cmd := exec.Command(os.Args[0], "-test.run=TestWatchInterruptSignalsSecondSignalStillKills") + cmd.Env = append(os.Environ(), signalChildEnv+"=1") + stdout, err := cmd.StdoutPipe() + require.NoError(t, err) + require.NoError(t, cmd.Start()) + t.Cleanup(func() { _ = cmd.Process.Kill() }) + + lines := make(chan string, 8) + go func() { + sc := bufio.NewScanner(stdout) + for sc.Scan() { + lines <- sc.Text() + } + close(lines) + }() + await := func(want string) bool { + deadline := time.After(30 * time.Second) + for { + select { + case l, ok := <-lines: + if !ok { + return false + } + if strings.Contains(l, want) { + return true + } + case <-deadline: + return false + } + } + } + + require.True(t, await("READY"), "child never started") + require.NoError(t, cmd.Process.Signal(sig)) + require.True(t, await("CANCELLED"), "first %s did not cancel the context", sig) + + // The second signal must reach the default disposition and kill the + // child rather than being buffered and dropped by the handler. + require.NoError(t, cmd.Process.Signal(sig)) + waitErr := make(chan error, 1) + go func() { waitErr <- cmd.Wait() }() + select { + case err := <-waitErr: + // Killed by the signal, so a non-nil (non-exit-zero) error. + assert.Error(t, err, "child should die by signal, not exit cleanly") + case <-time.After(15 * time.Second): + t.Fatalf("second %s was swallowed: the user has no escape hatch "+ + "during the process group's SIGKILL grace window", sig) + } + }) + } +} + +// TestWatchInterruptSignalsStopsWithoutSignal covers the no-signal path: stop() +// must return (joining its goroutine) rather than leaving it parked on the +// channel for the life of the process. +func TestWatchInterruptSignalsStopsWithoutSignal(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + stop := watchInterruptSignals(ctx, cancel) + + returned := make(chan struct{}) + go func() { + stop() + close(returned) + }() + select { + case <-returned: + case <-time.After(10 * time.Second): + t.Fatal("stop() blocked: the signal goroutine was never joined") + } +} diff --git a/libs/localenv/pipeline.go b/libs/localenv/pipeline.go index 194bc5c1f6a..30f803a21e4 100644 --- a/libs/localenv/pipeline.go +++ b/libs/localenv/pipeline.go @@ -82,6 +82,38 @@ func (p *Pipeline) Run(ctx context.Context) (*Result, error) { p.res.Phases = initialPhases() if err := p.run(ctx); err != nil { + // A cancelled context means the user or parent interrupted us (SIGINT/ + // SIGTERM). The phase that was running reports its own failure (e.g. uv + // sync exiting on the signal surfaces as E_PROVISION with "signal: + // terminated"), which misleads a --json consumer into thinking something + // broke. Reclassify to E_CANCELED here — the single funnel where ctx is in + // scope — keeping the recorded FailurePhase and diskMutated so the consumer + // still knows where we stopped and whether disk was touched. + // + // The phase's own error is kept as the wrapped cause rather than replaced: + // a real failure can race with the signal (uv sync failing on a dependency + // conflict while the user gives up and hits Ctrl-C), and that cause is the + // only diagnostic there is. + if ctx.Err() != nil && p.res.Error != nil { + // Snapshot the phase's error *before* overwriting Code/Msg below. uvFailure + // folds uv's stderr — the actual diagnostic (e.g. a dependency-conflict + // "no solution found") — into Msg, so wrapping only the inner .Err would + // drop it, leaving less than main in exactly the racing-failure case this + // is meant to preserve. Wrapping the whole original PipelineError keeps + // Msg (stderr and all) in the chain. + orig := &PipelineError{Code: p.res.Error.Code, Msg: p.res.Error.Msg, Err: p.res.Error.Err} + p.res.Error.Code = ErrCanceled + p.res.Error.Msg = "interrupted" + // Two %w verbs keep both the context error and the phase's original error + // matchable by errors.Is, on one line — errors.Join would embed a newline + // and break the single-line phase row text mode prints. + p.res.Error.Err = fmt.Errorf("%w; %w", ctx.Err(), orig) + // fail() already snapshotted the pre-reclassification text into the + // errored phase's Detail, which is what text mode prints. Re-sync it so + // text and --json agree on cancellation (see PipelineError.MarshalJSON). + p.syncFailureDetail() + return p.res, p.res.Error + } return p.res, err } p.res.OK = true @@ -474,6 +506,22 @@ func (p *Pipeline) fail(phase PhaseName, diskMutated bool, pe *PipelineError) er return pe } +// syncFailureDetail re-copies the recorded error's text into its phase's Detail. +// fail() sets Detail when the failure happens; a caller that rewrites the error +// afterwards (Run's E_CANCELED reclassification) must call this so text output — +// which prints Detail — keeps agreeing with the --json error object. +func (p *Pipeline) syncFailureDetail() { + if p.res.Error == nil { + return + } + for i := range p.res.Phases { + if p.res.Phases[i].Phase == p.res.Error.FailurePhase { + p.res.Phases[i].Detail = p.res.Error.Error() + return + } + } +} + // asPipelineError returns err as a *PipelineError if it already is one, otherwise // wraps it with the fallback code and message. func asPipelineError(err error, fallback ErrorCode, format string, args ...any) *PipelineError { diff --git a/libs/localenv/pipeline_test.go b/libs/localenv/pipeline_test.go index eb25935573e..865e4e30263 100644 --- a/libs/localenv/pipeline_test.go +++ b/libs/localenv/pipeline_test.go @@ -8,12 +8,19 @@ import ( "os" "path/filepath" "runtime" + "strings" "testing" + "github.com/databricks/cli/libs/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// cancelPMStderr is the stderr the interrupted uv sync emits. It stands in for a +// real resolver diagnostic — the thing the cancellation reclassification must not +// drop when it races with a Ctrl-C. +const cancelPMStderr = "error: no solution found: databricks-connect==17.2 conflicts with pyspark==3.5" + type fakePM struct{ py, dbc string } func (fakePM) Name() string { return "fake" } @@ -60,6 +67,30 @@ func (uvMissingPM) EnsureAvailable(context.Context) (string, error) { return "", errors.New("uv not found and install failed") } +// cancelPM simulates uv being interrupted: Provision closes entered (so the test +// knows the pipeline reached this phase), blocks until the context is cancelled, +// then returns a *process.ProcessError carrying uv's stderr (NOT context.Canceled), +// exactly as a real `uv sync` does when it exits on SIGTERM mid-resolution. The +// stderr is the real diagnostic; the pipeline's uvFailure folds it into the +// PipelineError's Msg, which the cancellation reclassification must preserve. +type cancelPM struct { + fakePM + entered chan struct{} +} + +func (c cancelPM) Provision(ctx context.Context, _, _ string) error { + close(c.entered) + <-ctx.Done() + // Mirror uvManager.Provision's real return: a *PipelineError from uvFailure, + // which folds uv's stderr into Msg. Returning a bare ProcessError would not + // reproduce the stderr-in-Msg shape the reclassification must preserve. + return uvFailure(ErrProvision, &process.ProcessError{ + Command: "uv sync", + Err: errors.New("signal: terminated"), + Stderr: cancelPMStderr, + }, "uv sync") +} + func writeProject(t *testing.T) string { dir := t.TempDir() require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), []byte(`[project] @@ -125,6 +156,71 @@ func TestPipelineCheckMutatesNothing(t *testing.T) { assert.Empty(t, entries) } +func TestPipelineReportsCancellationNotProvisionFailure(t *testing.T) { + // When the context is cancelled mid-provision (a Ctrl-C / SIGTERM), the run + // must surface E_CANCELED, not E_PROVISION — the provision phase's own error + // ("signal: terminated") would otherwise imply something broke. + dir := writeProject(t) + srv := newTestServer(t) + defer srv.Close() + + ctx, cancel := context.WithCancel(t.Context()) + pm := cancelPM{fakePM: fakePM{py: "3.12", dbc: "17.2.0"}, entered: make(chan struct{})} + p := &Pipeline{ + Mode: ModeDefault, Check: false, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: ComputeFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: pm, + } + + // Cancel once Provision is running, so the run unblocks and returns through the + // interrupt path (mirrors a Ctrl-C landing mid-`uv sync`). + go func() { + <-pm.entered + cancel() + }() + + res, err := p.Run(ctx) + var pe *PipelineError + require.ErrorAs(t, err, &pe) + assert.Equal(t, ErrCanceled, pe.Code) + assert.Equal(t, PhaseProvision, pe.FailurePhase, "should still record where it stopped") + require.NotNil(t, res.Error) + assert.Equal(t, ErrCanceled, res.Error.Code) + assert.False(t, res.OK) + // The wrapped cause is the context error, so errors.Is works upstream. + assert.ErrorIs(t, pe, context.Canceled) + + // The phase's own error is kept as a second cause, not discarded: a genuine + // failure can race with the signal, and its stderr is the only diagnostic + // there is. uvFailure folds that stderr into Msg, so preserving only the inner + // .Err would drop it — assert the actual resolver output survives. + assert.Contains(t, pe.Error(), "signal: terminated", + "the phase's cause must survive the reclassification") + assert.Contains(t, pe.Error(), cancelPMStderr, + "uv's stderr (the real diagnostic) must survive the cancellation reclassification") + + // Text mode prints the errored phase's Detail while --json prints the error + // object; they must agree (see PipelineError.MarshalJSON). Detail is set when + // the phase fails, i.e. before the reclassification, so this catches a stale one. + var detail string + for _, ph := range res.Phases { + if ph.Phase == PhaseProvision { + detail = ph.Detail + } + } + assert.Equal(t, pe.Error(), detail, "text-mode phase detail must match the JSON error") + // A Ctrl-C must be *classified* as cancellation, not a provision failure: the + // message leads with "interrupted" (Code is E_CANCELED). The retained cause may + // still contain the phase's own "... failed" text — that is the preserved + // diagnostic, not the classification — so assert the prefix, not absence. + assert.True(t, strings.HasPrefix(detail, "interrupted"), + "a Ctrl-C must read as interrupted, not a provision failure: %q", detail) + + // Both causes render on one line: a phase row is a single line of output. + assert.NotContains(t, pe.Error(), "\n", "the error must stay single-line") +} + func TestPipelineCheckReRunPlanMatchesRealRun(t *testing.T) { // On a re-run where the .bak already exists and the live file already equals // the merged output, --dry-run must report a plan a real run would perform: no diff --git a/libs/localenv/result.go b/libs/localenv/result.go index 8146690e9cb..244769727af 100644 --- a/libs/localenv/result.go +++ b/libs/localenv/result.go @@ -87,6 +87,13 @@ const ( ErrPythonInstall ErrorCode = "E_PYTHON_INSTALL" // provision: uv python install failed ErrProvision ErrorCode = "E_PROVISION" // provision: uv sync failed ErrValidate ErrorCode = "E_VALIDATE" // validate: post-provision version mismatch + + // ErrCanceled is not in the spec's error-code table: it reports a user/parent + // interrupt (SIGINT/SIGTERM cancels the context), not a failure of the phase + // it happened to be in. Without it an interrupt mid-`uv sync` surfaces as + // E_PROVISION with a "provision failed" message, implying something broke when + // the user simply pressed Ctrl-C. FailurePhase still records where it stopped. + ErrCanceled ErrorCode = "E_CANCELED" // any phase: interrupted by SIGINT/SIGTERM ) // PipelineError is a failure carrying a stable code, the phase at which it diff --git a/libs/localenv/uv.go b/libs/localenv/uv.go index 81fbc8802c8..dc8c5bb35e0 100644 --- a/libs/localenv/uv.go +++ b/libs/localenv/uv.go @@ -63,7 +63,7 @@ func (m *uvManager) EnsureAvailable(ctx context.Context) (string, error) { m.bin = bin // Use --version (not "version") to avoid project-scoped sub-command that requires pyproject.toml. - version, err := process.Background(ctx, []string{m.bin, "--version"}) + version, err := process.Background(ctx, []string{m.bin, "--version"}, process.WithProcessGroup()) if err != nil { return "", uvFailure(ErrUvMissing, err, "uv version check") } @@ -75,12 +75,15 @@ func (m *uvManager) EnsureAvailable(ctx context.Context) (string, error) { // (process.WithDir("") is a no-op). The index-url is injected only when // resolveIndexURL returns non-empty; it returns "" when UV_INDEX_URL is already // set, so an explicit value in the environment is never clobbered. +// WithProcessGroup is applied because uv fans out to its own subprocesses +// (Python, build backends); on SIGINT/SIGTERM they must be reaped as a group +// rather than left as orphans holding locks over a half-written .venv. func (m *uvManager) runUv(ctx context.Context, args []string, dir string) error { if indexURL := m.resolveIndexURL(ctx); indexURL != "" { - _, err := process.Background(ctx, args, process.WithDir(dir), process.WithEnv("UV_INDEX_URL", indexURL)) + _, err := process.Background(ctx, args, process.WithDir(dir), process.WithEnv("UV_INDEX_URL", indexURL), process.WithProcessGroup()) return err } - _, err := process.Background(ctx, args, process.WithDir(dir)) + _, err := process.Background(ctx, args, process.WithDir(dir), process.WithProcessGroup()) return err } @@ -155,6 +158,7 @@ except importlib.metadata.PackageNotFoundError: out, err := process.Background(ctx, []string{venvPython(projectDir), "-c", pyCode}, process.WithDir(projectDir), + process.WithProcessGroup(), ) if err != nil { return "", "", uvFailure(ErrValidate, err, "venv python validation") @@ -361,7 +365,10 @@ func installUv(ctx context.Context) error { // (~/.local/bin), so record exactly what ran before it fires — visible under // --debug for anyone auditing where uv came from. log.Debugf(ctx, "uv: not found; running installer: %s", strings.Join(cmd, " ")) - _, err := process.Background(ctx, cmd) + // The installer is a shell/PowerShell pipeline that spawns curl and the + // downloaded script; reap the whole group on cancellation so an interrupted + // install leaves no orphaned downloader behind. + _, err := process.Background(ctx, cmd, process.WithProcessGroup()) return err } diff --git a/libs/process/background.go b/libs/process/background.go index 2649d0ef2c6..e1b20414aec 100644 --- a/libs/process/background.go +++ b/libs/process/background.go @@ -47,7 +47,11 @@ func Background(ctx context.Context, args []string, opts ...execOption) (string, return "", err } } - if err := runCmd(ctx, cmd); err != nil { + err := runCmd(ctx, cmd) + // Sweep the process group (WithProcessGroup + cancelled context only) so a + // grandchild that outlived a SIGKILLed leader is not re-orphaned. + reapProcessGroup(ctx, cmd) + if err != nil { return stdout.String(), &ProcessError{ Err: err, Command: commandStr, diff --git a/libs/process/forwarded.go b/libs/process/forwarded.go index 1d7fdb71e4d..91070f6c96b 100644 --- a/libs/process/forwarded.go +++ b/libs/process/forwarded.go @@ -34,5 +34,9 @@ func Forwarded(ctx context.Context, args []string, src io.Reader, outWriter, err } } - return runCmd(ctx, cmd) + err := runCmd(ctx, cmd) + // Sweep the process group (WithProcessGroup + cancelled context only) so a + // grandchild that outlived a SIGKILLed leader is not re-orphaned. + reapProcessGroup(ctx, cmd) + return err } diff --git a/libs/process/group.go b/libs/process/group.go new file mode 100644 index 00000000000..24b3e289dd0 --- /dev/null +++ b/libs/process/group.go @@ -0,0 +1,9 @@ +package process + +import "time" + +// processGroupGracePeriod bounds how long WithProcessGroup waits after the +// context is cancelled before escalating to SIGKILL. It mirrors the 10s grace +// period used elsewhere for subprocess termination (see experimental/ssh). It is +// a var, not a const, only so the escalation test can shorten it. +var processGroupGracePeriod = 10 * time.Second diff --git a/libs/process/group_other.go b/libs/process/group_other.go new file mode 100644 index 00000000000..29c1a1b8c33 --- /dev/null +++ b/libs/process/group_other.go @@ -0,0 +1,29 @@ +//go:build !unix + +package process + +import ( + "context" + "os/exec" +) + +// WithProcessGroup sets a WaitDelay so a cancelled command does not block +// indefinitely on a stuck child. +// +// Unlike the Unix build, this does not reap the child's descendants: killing a +// whole process tree on Windows requires a Job Object (CreateJobObject + +// AssignProcessToJobObject with JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE), which is +// out of scope here. The direct child is still terminated by the default +// cancellation, and the caller's signal handler still fires; only grandchildren +// spawned by the child may outlive it. +func WithProcessGroup() execOption { + return func(_ context.Context, c *exec.Cmd) error { + c.WaitDelay = processGroupGracePeriod + return nil + } +} + +// reapProcessGroup is a no-op on non-unix builds: there is no process group to +// sweep (WithProcessGroup does not set Setpgid here). See the unix build for the +// group-SIGKILL escalation this closes. +func reapProcessGroup(_ context.Context, _ *exec.Cmd) {} diff --git a/libs/process/group_unix.go b/libs/process/group_unix.go new file mode 100644 index 00000000000..8f0adb225aa --- /dev/null +++ b/libs/process/group_unix.go @@ -0,0 +1,71 @@ +//go:build unix + +package process + +import ( + "context" + "errors" + "os" + "os/exec" + "syscall" +) + +// WithProcessGroup makes the child the leader of a new process group and, when +// the context is cancelled, signals the entire group rather than just the child. +// +// exec.CommandContext's default cancellation only SIGKILLs the direct child, so +// a tool that fans out to its own subprocesses (e.g. `uv sync` spawning Python +// and build backends) leaves those grandchildren running as orphans when the CLI +// receives SIGINT/SIGTERM. Putting the child in its own group and signalling the +// group (negative PID) delivers SIGTERM to every descendant at once, giving them +// a chance to exit cleanly. +// +// Two backstops handle a member that ignores SIGTERM: WaitDelay bounds how long +// Wait blocks on a hung leader (Go then SIGKILLs the leader and closes the pipes +// so Wait returns), and reapProcessGroup sends a final group-wide SIGKILL after +// Wait returns. The second is necessary because Go's WaitDelay escalation targets +// the leader PID only, not the group — without it a grandchild that outlives a +// SIGKILLed leader would be re-orphaned. +func WithProcessGroup() execOption { + return func(_ context.Context, c *exec.Cmd) error { + if c.SysProcAttr == nil { + c.SysProcAttr = &syscall.SysProcAttr{} + } + c.SysProcAttr.Setpgid = true + + c.WaitDelay = processGroupGracePeriod + c.Cancel = func() error { + // With Setpgid and Pgid unset, the child's group ID equals its PID; + // a negative PID targets the whole group. Map "no such process" to + // os.ErrProcessDone so a benign exit/cancel race is not surfaced as a + // Wait error. + err := syscall.Kill(-c.Process.Pid, syscall.SIGTERM) + if errors.Is(err, syscall.ESRCH) { + return os.ErrProcessDone + } + return err + } + return nil + } +} + +// reapProcessGroup sends a final SIGKILL to the child's process group after the +// command has been waited on, closing the gap left by Go's WaitDelay escalation +// (which SIGKILLs only the leader PID). It runs only for a WithProcessGroup child +// whose context was cancelled — the escalation path — so a normally-exited +// command is never signalled. +// +// It is safe against PID reuse: this runs synchronously after Wait has returned +// (the leader is reaped), not on a delayed timer. The group ID stays reserved by +// the kernel while any member is alive, so kill(-pgid) hits surviving descendants +// or returns ESRCH on an already-empty group; there is no 10s window in which the +// PGID could be reused by an unrelated group before the signal is sent. +func reapProcessGroup(ctx context.Context, c *exec.Cmd) { + if c.SysProcAttr == nil || !c.SysProcAttr.Setpgid { + return + } + if c.Process == nil || ctx.Err() == nil { + return + } + _ = syscall.Kill(-c.Process.Pid, syscall.SIGKILL) +} diff --git a/libs/process/group_unix_test.go b/libs/process/group_unix_test.go new file mode 100644 index 00000000000..3db07a9bac9 --- /dev/null +++ b/libs/process/group_unix_test.go @@ -0,0 +1,109 @@ +//go:build unix + +package process + +import ( + "context" + "errors" + "os" + "path/filepath" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestWithProcessGroupReapsGrandchild verifies that cancelling the context kills +// the whole process group, not just the direct child. The shell (the group +// leader) backgrounds a long sleep — a grandchild of the test process — and +// records its PID; after cancellation that grandchild must be gone. +func TestWithProcessGroupReapsGrandchild(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + + pidFile := filepath.Join(t.TempDir(), "grandchild.pid") + // $1 is pidFile: background a sleep, record its PID, then wait on it so the + // shell stays alive as the group leader until the group is signalled. + script := []string{"sh", "-c", `sleep 300 & echo $! > "$1"; wait`, "sh", pidFile} + + done := make(chan struct{}) + go func() { + defer close(done) + _, _ = Background(ctx, script, WithProcessGroup()) + }() + + grandchildPid := waitForPid(t, pidFile) + + cancel() + + select { + case <-done: + case <-time.After(processGroupGracePeriod + 5*time.Second): + t.Fatal("Background did not return after context cancellation") + } + + // The grandchild inherited the leader's group, so the group SIGTERM reaches + // it directly. Poll briefly to let the kernel deliver the signal and reap it. + assert.Eventually(t, func() bool { + return errors.Is(syscall.Kill(grandchildPid, 0), syscall.ESRCH) + }, 5*time.Second, 20*time.Millisecond, "grandchild %d was orphaned, not reaped", grandchildPid) +} + +// TestWithProcessGroupReapsGrandchildAfterEscalation covers the SIGKILL +// escalation path: a leader that ignores SIGTERM (trap ” TERM). The group +// SIGTERM from Cancel does nothing, so WaitDelay expires and Go SIGKILLs the +// leader PID only — leaving the grandchild that this option exists to reap. The +// post-Wait group sweep (reapProcessGroup) must SIGKILL the whole group so the +// grandchild does not survive. +func TestWithProcessGroupReapsGrandchildAfterEscalation(t *testing.T) { + // Shorten the grace period so the WaitDelay escalation fires quickly. + orig := processGroupGracePeriod + processGroupGracePeriod = 500 * time.Millisecond + t.Cleanup(func() { processGroupGracePeriod = orig }) + + ctx, cancel := context.WithCancel(t.Context()) + + pidFile := filepath.Join(t.TempDir(), "grandchild.pid") + // The leader ignores SIGTERM, so only the escalation can stop it; the sleep + // grandchild inherits the group but not the trap. + script := []string{"sh", "-c", `trap '' TERM; sleep 300 & echo $! > "$1"; wait`, "sh", pidFile} + + done := make(chan struct{}) + go func() { + defer close(done) + _, _ = Background(ctx, script, WithProcessGroup()) + }() + + grandchildPid := waitForPid(t, pidFile) + + cancel() + + select { + case <-done: + case <-time.After(processGroupGracePeriod + 10*time.Second): + t.Fatal("Background did not return after escalation") + } + + assert.Eventually(t, func() bool { + return errors.Is(syscall.Kill(grandchildPid, 0), syscall.ESRCH) + }, 5*time.Second, 20*time.Millisecond, + "grandchild %d survived the SIGKILL escalation (re-orphaned)", grandchildPid) +} + +// waitForPid waits for the shell to write the grandchild PID and returns it. +func waitForPid(t *testing.T, pidFile string) int { + t.Helper() + var pid int + require.Eventually(t, func() bool { + b, err := os.ReadFile(pidFile) + if err != nil { + return false + } + pid, err = strconv.Atoi(strings.TrimSpace(string(b))) + return err == nil && pid > 0 + }, 5*time.Second, 20*time.Millisecond, "grandchild PID was never recorded") + return pid +}