From 65f1afdf6075ac8d61e1da668f71bafce4caf04b Mon Sep 17 00:00:00 2001 From: alex-the-third <139810351+alex-the-third@users.noreply.github.com> Date: Wed, 12 Aug 2026 22:00:05 +0200 Subject: [PATCH 1/3] Clean up Windows action process trees Create action processes suspended and assign them to non-breakaway job objects before allowing them to execute. Terminate and wait for surviving descendants when the root process exits so build directory cleanup does not race lingering processes. Serialize startup cancellation with process resumption and cover descendant cleanup and the cancellation race with Windows tests. --- pkg/runner/BUILD.bazel | 5 +- pkg/runner/local_runner.go | 28 +- pkg/runner/local_runner_process.go | 29 ++ pkg/runner/local_runner_process_windows.go | 308 ++++++++++++++++++ .../local_runner_process_windows_test.go | 92 ++++++ pkg/runner/local_runner_test.go | 202 ++++++++++++ 6 files changed, 659 insertions(+), 5 deletions(-) create mode 100644 pkg/runner/local_runner_process.go create mode 100644 pkg/runner/local_runner_process_windows.go create mode 100644 pkg/runner/local_runner_process_windows_test.go diff --git a/pkg/runner/BUILD.bazel b/pkg/runner/BUILD.bazel index fb596676..706dfd24 100644 --- a/pkg/runner/BUILD.bazel +++ b/pkg/runner/BUILD.bazel @@ -7,6 +7,8 @@ go_library( "clean_runner.go", "local_runner.go", "local_runner_darwin.go", + "local_runner_process.go", + "local_runner_process_windows.go", "local_runner_rss_bytes.go", "local_runner_rss_kibibytes.go", "local_runner_unix.go", @@ -69,12 +71,13 @@ go_test( srcs = [ "apple_xcode_resolving_runner_test.go", "clean_runner_test.go", + "local_runner_process_windows_test.go", "local_runner_test.go", "path_existence_checking_runner_test.go", "temporary_directory_symlinking_runner_test.go", ], + embed = [":runner"], deps = [ - ":runner", "//internal/mock", "//pkg/cleaner", "//pkg/proto/resourceusage", diff --git a/pkg/runner/local_runner.go b/pkg/runner/local_runner.go index 31d7e63b..ba6085ea 100644 --- a/pkg/runner/local_runner.go +++ b/pkg/runner/local_runner.go @@ -125,7 +125,9 @@ func NewPlainCommandCreator(sysProcAttr *syscall.SysProcAttr) CommandCreator { } // NewLocalRunner returns a Runner capable of running commands on the -// local system directly. +// local system directly. On Windows, commands are placed in a +// non-breakaway job object, and any surviving descendants are terminated +// and waited for when the root process exits. func NewLocalRunner(buildDirectory filesystem.Directory, buildDirectoryPath *path.Builder, commandCreator CommandCreator, setTmpdirEnvironmentVariable bool) runner.RunnerServer { return &localRunner{ buildDirectory: buildDirectory, @@ -185,10 +187,17 @@ func (r *localRunner) Run(ctx context.Context, request *runner.RunRequest) (*run // Start the subprocess. We can already close the output files // while the process is running. + commandProcess, err := prepareCommandForStart(cmd) + if err != nil { + stdout.Close() + stderr.Close() + return nil, util.StatusWrap(err, "Failed to prepare process") + } err = cmd.Start() stdout.Close() stderr.Close() if err != nil { + commandProcess.Close() code := codes.Internal for _, invalidArgumentErr := range invalidArgumentErrs { if errors.Is(err, invalidArgumentErr) { @@ -198,13 +207,24 @@ func (r *localRunner) Run(ctx context.Context, request *runner.RunRequest) (*run } return nil, util.StatusWrapWithCode(err, code, "Failed to start process") } + if err := commandProcess.AfterStart(cmd); err != nil { + return nil, util.StatusWrap(err, "Failed to finish process startup") + } // Wait for execution to complete. Permit non-zero exit codes. - if err := cmd.Wait(); err != nil { - if _, ok := err.(*exec.ExitError); !ok { - return nil, err + waitErr := cmd.Wait() + afterWaitErr := commandProcess.AfterWait(cmd) + if waitErr != nil { + if afterWaitErr != nil { + return nil, afterWaitErr + } + if _, ok := waitErr.(*exec.ExitError); !ok { + return nil, waitErr } } + if afterWaitErr != nil { + return nil, afterWaitErr + } // Attach rusage information to the response. posixResourceUsage, err := anypb.New(getPOSIXResourceUsage(cmd)) diff --git a/pkg/runner/local_runner_process.go b/pkg/runner/local_runner_process.go new file mode 100644 index 00000000..6548bc9e --- /dev/null +++ b/pkg/runner/local_runner_process.go @@ -0,0 +1,29 @@ +//go:build !windows +// +build !windows + +package runner + +import "os/exec" + +type commandProcess struct{} + +// prepareCommandForStart is called after the command is fully configured and +// before Start. Non-Windows platforms do not need extra process-tree state. +func prepareCommandForStart(cmd *exec.Cmd) (*commandProcess, error) { + return &commandProcess{}, nil +} + +// AfterStart is called after Start succeeds and before the caller waits. +func (commandProcess) AfterStart(cmd *exec.Cmd) error { + return nil +} + +// AfterWait is called after Wait returns, before build directory cleanup can +// proceed. +func (commandProcess) AfterWait(cmd *exec.Cmd) error { + return nil +} + +// Close releases any resources allocated by prepareCommandForStart. It must be +// safe to call if Start fails. +func (commandProcess) Close() {} diff --git a/pkg/runner/local_runner_process_windows.go b/pkg/runner/local_runner_process_windows.go new file mode 100644 index 00000000..831fc88d --- /dev/null +++ b/pkg/runner/local_runner_process_windows.go @@ -0,0 +1,308 @@ +//go:build windows +// +build windows + +// This file intentionally mirrors the native Bazel Windows launcher: +// https://github.com/bazelbuild/bazel/blob/master/src/main/native/windows/process.cc +// +// The root process is created suspended, assigned to a non-breakaway job, and +// only then resumed. Assigning while suspended is the important race fix: an +// unsuspended process could spawn descendants before the runner has contained +// it, leaving those descendants alive to keep the input root undeletable. + +package runner + +import ( + "errors" + "fmt" + "os" + "os/exec" + "sync" + "syscall" + "unsafe" + + "golang.org/x/sys/windows" +) + +const ( + // jobObjectMsgActiveProcessZero is JOB_OBJECT_MSG_ACTIVE_PROCESS_ZERO. + // The Windows SDK's winnt.h defines this message value as 4, but + // x/sys/windows does not expose it. The official job completion-port docs + // describe the message: + // https://learn.microsoft.com/en-us/windows/win32/api/winnt/ns-winnt-jobobject_associate_completion_port + jobObjectMsgActiveProcessZero = 4 + + windowsTerminatedExitCode = 130 +) + +type jobObjectAssociateCompletionPort struct { + CompletionKey uintptr + CompletionPort windows.Handle +} + +type commandProcess struct { + lock sync.Mutex + job windows.Handle + ioport windows.Handle + assigned bool + closed bool + canceled bool + // afterCancellationCheck is a test-only hook for pausing AfterStart while + // it holds lock between checking canceled and resuming the process. + afterCancellationCheck func() +} + +// prepareCommandForStart creates the job object and completion port before the +// process exists. It also amends cmd so Start creates the root process suspended +// and in a new process group. The returned object owns those handles until +// Close or AfterWait. +func prepareCommandForStart(cmd *exec.Cmd) (*commandProcess, error) { + job, err := windows.CreateJobObject(nil, nil) + if err != nil { + return nil, err + } + p := &commandProcess{ + job: job, + } + success := false + defer func() { + if !success { + p.Close() + } + }() + + jobInfo := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{} + jobInfo.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + if _, err := windows.SetInformationJobObject( + job, + windows.JobObjectExtendedLimitInformation, + uintptr(unsafe.Pointer(&jobInfo)), + uint32(unsafe.Sizeof(jobInfo)), + ); err != nil { + return nil, err + } + + ioport, err := windows.CreateIoCompletionPort(windows.InvalidHandle, 0, 0, 1) + if err != nil { + return nil, err + } + p.ioport = ioport + + port := jobObjectAssociateCompletionPort{ + CompletionKey: uintptr(job), + CompletionPort: ioport, + } + if _, err := windows.SetInformationJobObject( + job, + windows.JobObjectAssociateCompletionPortInformation, + uintptr(unsafe.Pointer(&port)), + uint32(unsafe.Sizeof(port)), + ); err != nil { + return nil, err + } + + var sysProcAttr syscall.SysProcAttr + if cmd.SysProcAttr != nil { + sysProcAttr = *cmd.SysProcAttr + } + sysProcAttr.CreationFlags |= windows.CREATE_NEW_PROCESS_GROUP | windows.CREATE_SUSPENDED + cmd.SysProcAttr = &sysProcAttr + if cmd.Cancel != nil { + cmd.Cancel = p.Cancel + } + + success = true + return p, nil +} + +// Cancel may run as soon as cmd.Context is canceled, including before AfterStart +// has assigned the suspended root process to the job. In that pre-assignment +// race, record the cancellation and let AfterStart terminate the job after the +// assignment has made termination cover the whole process tree. +func (p *commandProcess) Cancel() error { + p.lock.Lock() + p.canceled = true + assigned := p.assigned + closed := p.closed + p.lock.Unlock() + if closed { + return os.ErrProcessDone + } + if !assigned { + return nil + } + return p.terminateJob() +} + +// AfterStart assigns the still-suspended root process to the job, then either +// applies a cancellation that arrived early or resumes the process. On any +// failure after Start, it kills/reaps the partially started process tree before +// returning to the caller. +func (p *commandProcess) AfterStart(cmd *exec.Cmd) error { + processHandle, err := windows.OpenProcess( + windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, + false, + uint32(cmd.Process.Pid), + ) + if err != nil { + return p.cleanupAfterStartFailure(cmd, err, false) + } + err = windows.AssignProcessToJobObject(p.job, processHandle) + windows.CloseHandle(processHandle) + if err != nil { + return p.cleanupAfterStartFailure(cmd, err, false) + } + + p.lock.Lock() + p.assigned = true + if p.canceled { + p.lock.Unlock() + if err := p.terminateJob(); err != nil { + return p.cleanupAfterStartFailure(cmd, err, true) + } + return nil + } + if p.afterCancellationCheck != nil { + p.afterCancellationCheck() + } + err = resumeProcessThreads(uint32(cmd.Process.Pid)) + p.lock.Unlock() + if err != nil { + return p.cleanupAfterStartFailure(cmd, err, true) + } + return nil +} + +// AfterWait runs after the root process has exited. It terminates anything that +// remains in the job, waits for the active-process-zero notification, and only +// then releases the job handles so build-directory cleanup cannot race lingering +// descendants. +func (p *commandProcess) AfterWait(cmd *exec.Cmd) error { + defer p.Close() + if err := p.terminateJob(); err != nil && !errors.Is(err, os.ErrProcessDone) { + return err + } + return p.waitForActiveProcessZero() +} + +// Close releases job resources. Once a process has been assigned to the job, +// callers should prefer AfterWait so descendants are terminated and waited for +// before these handles are closed. +func (p *commandProcess) Close() { + p.lock.Lock() + if p.closed { + p.lock.Unlock() + return + } + p.closed = true + job := p.job + ioport := p.ioport + p.job = 0 + p.ioport = 0 + p.lock.Unlock() + + if job != 0 { + windows.CloseHandle(job) + } + if ioport != 0 { + windows.CloseHandle(ioport) + } +} + +func (p *commandProcess) cleanupAfterStartFailure(cmd *exec.Cmd, cause error, assigned bool) error { + if assigned { + _ = p.terminateJob() + } else if cmd.Process != nil { + _ = cmd.Process.Kill() + } + _ = cmd.Wait() + if assigned { + _ = p.waitForActiveProcessZero() + } + p.Close() + return cause +} + +func (p *commandProcess) terminateJob() error { + p.lock.Lock() + closed := p.closed + job := p.job + p.lock.Unlock() + if closed || job == 0 { + return os.ErrProcessDone + } + return windows.TerminateJobObject(job, windowsTerminatedExitCode) +} + +func (p *commandProcess) waitForActiveProcessZero() error { + p.lock.Lock() + closed := p.closed + job := p.job + ioport := p.ioport + p.lock.Unlock() + if closed || job == 0 || ioport == 0 { + return nil + } + for { + var completionCode uint32 + var completionKey uintptr + var overlapped *windows.Overlapped + if err := windows.GetQueuedCompletionStatus( + ioport, + &completionCode, + &completionKey, + &overlapped, + windows.INFINITE, + ); err != nil { + return err + } + if windows.Handle(completionKey) == job && completionCode == jobObjectMsgActiveProcessZero { + return nil + } + } +} + +// resumeProcessThreads resumes all threads belonging to the suspended root +// process. Go's os/exec path closes the primary thread handle returned by +// CreateProcess, so this code has to rediscover the thread handle by PID. +func resumeProcessThreads(pid uint32) error { + snapshot, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0) + if err != nil { + return err + } + defer windows.CloseHandle(snapshot) + + var threadHandles []windows.Handle + defer func() { + for _, threadHandle := range threadHandles { + windows.CloseHandle(threadHandle) + } + }() + + entry := windows.ThreadEntry32{ + Size: uint32(unsafe.Sizeof(windows.ThreadEntry32{})), + } + for err := windows.Thread32First(snapshot, &entry); ; err = windows.Thread32Next(snapshot, &entry) { + if err != nil { + if errors.Is(err, windows.ERROR_NO_MORE_FILES) { + break + } + return err + } + if entry.OwnerProcessID == pid { + threadHandle, err := windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, entry.ThreadID) + if err != nil { + return err + } + threadHandles = append(threadHandles, threadHandle) + } + } + if len(threadHandles) == 0 { + return fmt.Errorf("process %d has no threads to resume", pid) + } + for _, threadHandle := range threadHandles { + if _, err := windows.ResumeThread(threadHandle); err != nil { + return err + } + } + return nil +} diff --git a/pkg/runner/local_runner_process_windows_test.go b/pkg/runner/local_runner_process_windows_test.go new file mode 100644 index 00000000..b866e3bb --- /dev/null +++ b/pkg/runner/local_runner_process_windows_test.go @@ -0,0 +1,92 @@ +package runner + +import ( + "errors" + "os" + "os/exec" + "path/filepath" + "testing" + "time" +) + +func TestCommandProcessCancellationBetweenCheckAndResume(t *testing.T) { + // Pause AfterStart after it has checked for cancellation, at which point it + // must still hold the mutex. Cancel must block until AfterStart resumes the + // process and releases the mutex. The old implementation released the mutex + // before resume, allowing Cancel to terminate the suspended process first. + cmd := exec.Command( + filepath.Join(os.Getenv("SYSTEMROOT"), "System32", "cmd.exe"), + "/d", "/c", "ping -n 60 127.0.0.1 > nul", + ) + commandProcess, err := prepareCommandForStart(cmd) + if err != nil { + t.Fatal(err) + } + + afterCancellationCheck := make(chan struct{}) + allowResume := make(chan struct{}) + commandProcess.afterCancellationCheck = func() { + close(afterCancellationCheck) + <-allowResume + } + + if err := cmd.Start(); err != nil { + commandProcess.Close() + t.Fatal(err) + } + afterStartResult := make(chan error, 1) + go func() { + afterStartResult <- commandProcess.AfterStart(cmd) + }() + + select { + case <-afterCancellationCheck: + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for the cancellation check") + } + + var cancelErr error + cancellationReturnedBeforeResume := commandProcess.lock.TryLock() + var cancelResult chan error + if cancellationReturnedBeforeResume { + commandProcess.lock.Unlock() + cancelErr = commandProcess.Cancel() + } else { + cancelResult = make(chan error, 1) + go func() { + cancelResult <- commandProcess.Cancel() + }() + } + close(allowResume) + + var afterStartErr error + select { + case afterStartErr = <-afterStartResult: + case <-time.After(10 * time.Second): + t.Fatal("AfterStart() hung") + } + if !cancellationReturnedBeforeResume { + select { + case cancelErr = <-cancelResult: + case <-time.After(10 * time.Second): + t.Fatal("Cancel() hung") + } + } + + if cancelErr != nil && !errors.Is(cancelErr, os.ErrProcessDone) { + t.Errorf("Cancel() failed: %v", cancelErr) + } + if afterStartErr != nil { + t.Errorf("AfterStart() failed: %v", afterStartErr) + } + if cancellationReturnedBeforeResume { + t.Error("cancellation terminated the suspended job before resume") + } + + if afterStartErr == nil { + _ = cmd.Wait() + if err := commandProcess.AfterWait(cmd); err != nil { + t.Errorf("AfterWait() failed: %v", err) + } + } +} diff --git a/pkg/runner/local_runner_test.go b/pkg/runner/local_runner_test.go index b284eccc..c1e2ae78 100644 --- a/pkg/runner/local_runner_test.go +++ b/pkg/runner/local_runner_test.go @@ -3,11 +3,13 @@ package runner_test import ( "context" "os" + "os/exec" "path/filepath" "runtime" "strings" "syscall" "testing" + "time" "github.com/buildbarn/bb-remote-execution/internal/mock" "github.com/buildbarn/bb-remote-execution/pkg/proto/resourceusage" @@ -549,3 +551,203 @@ func TestLocalRunnerRun(t *testing.T) { // TODO: Improve testing coverage of LocalRunner. } + +func TestLocalRunnerRunWindowsSubprocessCleanup(t *testing.T) { + if runtime.GOOS != "windows" { + return + } + + // The child helper keeps its current directory inside the input root and + // opens a file there. Go's Windows syscall.Open shares read/write but not + // delete access, so RemoveAll fails while that descendant is alive. If this + // test can immediately remove the root, Run() waited for descendant cleanup. + buildDirectoryPath := t.TempDir() + buildDirectory, err := filesystem.NewLocalDirectory(path.LocalFormat.NewParser(buildDirectoryPath)) + require.NoError(t, err) + defer buildDirectory.Close() + + buildDirectoryPathBuilder, scopeWalker := path.EmptyBuilder.Join(path.VoidScopeWalker) + require.NoError(t, path.Resolve(path.LocalFormat.NewParser(buildDirectoryPath), scopeWalker)) + + testBinaryPath, err := os.Executable() + require.NoError(t, err) + + testName := "WindowsSubprocessCleanup" + testPath := filepath.Join(buildDirectoryPath, testName) + rootPath := filepath.Join(testPath, "root") + require.NoError(t, os.Mkdir(testPath, 0o777)) + require.NoError(t, os.Mkdir(rootPath, 0o777)) + require.NoError(t, os.Mkdir(filepath.Join(testPath, "tmp"), 0o777)) + + environmentVariables := map[string]string{ + "BB_RE_TEST_HELPER": "parent", + "BB_RE_TEST_BINARY": testBinaryPath, + "BB_RE_TEST_LOCKED_FILE": filepath.Join(rootPath, "locked"), + "BB_RE_TEST_READY_FILE": filepath.Join(rootPath, "ready"), + } + for _, name := range []string{"COMSPEC", "PATH", "SYSTEMROOT", "TEMP", "TMP", "WINDIR"} { + if value, ok := os.LookupEnv(name); ok { + environmentVariables[name] = value + } + } + + runner := runner.NewLocalRunner(buildDirectory, buildDirectoryPathBuilder, runner.NewPlainCommandCreator(&syscall.SysProcAttr{}), false) + response, err := runner.Run(context.Background(), &runner_pb.RunRequest{ + Arguments: []string{ + testBinaryPath, + "-test.run=TestLocalRunnerRunWindowsSubprocessCleanupHelper", + }, + EnvironmentVariables: environmentVariables, + StdoutPath: testName + "/stdout", + StderrPath: testName + "/stderr", + InputRootDirectory: testName + "/root", + TemporaryDirectory: testName + "/tmp", + }) + require.NoError(t, err) + require.Equal(t, int64(0), response.ExitCode) + require.FileExists(t, filepath.Join(rootPath, "ready")) + + require.NoError(t, os.RemoveAll(rootPath)) +} + +func TestLocalRunnerRunWindowsCancellationCleanup(t *testing.T) { + if runtime.GOOS != "windows" { + return + } + + // The root helper starts a descendant that holds an input-root file open, + // then remains alive. Cancellation must terminate both processes, and Run + // must wait for the entire job to be reaped so the root is immediately + // removable. + buildDirectoryPath := t.TempDir() + buildDirectory, err := filesystem.NewLocalDirectory(path.LocalFormat.NewParser(buildDirectoryPath)) + require.NoError(t, err) + defer buildDirectory.Close() + + buildDirectoryPathBuilder, scopeWalker := path.EmptyBuilder.Join(path.VoidScopeWalker) + require.NoError(t, path.Resolve(path.LocalFormat.NewParser(buildDirectoryPath), scopeWalker)) + + testName := "CancellationCleanup" + testPath := filepath.Join(buildDirectoryPath, testName) + rootPath := filepath.Join(testPath, "root") + require.NoError(t, os.Mkdir(testPath, 0o777)) + require.NoError(t, os.Mkdir(rootPath, 0o777)) + require.NoError(t, os.Mkdir(filepath.Join(testPath, "tmp"), 0o777)) + + testBinaryPath, err := os.Executable() + require.NoError(t, err) + readyFilePath := filepath.Join(rootPath, "ready") + environmentVariables := map[string]string{ + "BB_RE_TEST_HELPER": "parent_wait", + "BB_RE_TEST_BINARY": testBinaryPath, + "BB_RE_TEST_LOCKED_FILE": filepath.Join(rootPath, "locked"), + "BB_RE_TEST_READY_FILE": readyFilePath, + } + for _, name := range []string{"COMSPEC", "PATH", "SYSTEMROOT", "TEMP", "TMP", "WINDIR"} { + if value, ok := os.LookupEnv(name); ok { + environmentVariables[name] = value + } + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + type runResult struct { + response *runner_pb.RunResponse + err error + } + runResultChannel := make(chan runResult, 1) + localRunner := runner.NewLocalRunner(buildDirectory, buildDirectoryPathBuilder, runner.NewPlainCommandCreator(&syscall.SysProcAttr{}), false) + go func() { + response, err := localRunner.Run(ctx, &runner_pb.RunRequest{ + Arguments: []string{ + testBinaryPath, + "-test.run=TestLocalRunnerRunWindowsSubprocessCleanupHelper", + }, + EnvironmentVariables: environmentVariables, + StdoutPath: testName + "/stdout", + StderrPath: testName + "/stderr", + InputRootDirectory: testName + "/root", + TemporaryDirectory: testName + "/tmp", + }) + runResultChannel <- runResult{response: response, err: err} + }() + + deadline := time.Now().Add(10 * time.Second) + for { + if _, err := os.Stat(readyFilePath); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for the descendant helper") + } + time.Sleep(10 * time.Millisecond) + } + cancel() + + var result runResult + select { + case result = <-runResultChannel: + case <-time.After(10 * time.Second): + t.Fatal("Run() hung after cancellation") + } + if result.err != nil { + require.NotContains(t, result.err.Error(), "Failed to finish process startup") + } else { + require.NotNil(t, result.response) + } + + require.NoError(t, os.RemoveAll(rootPath)) + require.NoDirExists(t, rootPath) +} + +func TestLocalRunnerRunWindowsSubprocessCleanupHelper(t *testing.T) { + // parent spawns child and exits, parent_wait spawns child and remains alive, + // and child locks an input-root file before signaling that it is ready. + switch os.Getenv("BB_RE_TEST_HELPER") { + case "": + return + case "parent": + testBinaryPath := os.Getenv("BB_RE_TEST_BINARY") + if testBinaryPath == "" { + t.Fatal("BB_RE_TEST_BINARY is not set") + } + cmd := exec.Command(testBinaryPath, "-test.run=TestLocalRunnerRunWindowsSubprocessCleanupHelper") + cmd.Env = append(os.Environ(), "BB_RE_TEST_HELPER=child") + cmd.Dir = "." + require.NoError(t, cmd.Start()) + + readyFilePath := os.Getenv("BB_RE_TEST_READY_FILE") + deadline := time.Now().Add(10 * time.Second) + for { + if _, err := os.Stat(readyFilePath); err == nil { + return + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for child helper") + } + time.Sleep(10 * time.Millisecond) + } + case "parent_wait": + testBinaryPath := os.Getenv("BB_RE_TEST_BINARY") + if testBinaryPath == "" { + t.Fatal("BB_RE_TEST_BINARY is not set") + } + cmd := exec.Command(testBinaryPath, "-test.run=TestLocalRunnerRunWindowsSubprocessCleanupHelper") + cmd.Env = append(os.Environ(), "BB_RE_TEST_HELPER=child") + cmd.Dir = "." + require.NoError(t, cmd.Start()) + time.Sleep(time.Minute) + case "child": + lockedFilePath := os.Getenv("BB_RE_TEST_LOCKED_FILE") + readyFilePath := os.Getenv("BB_RE_TEST_READY_FILE") + lockedFile, err := os.OpenFile(lockedFilePath, os.O_CREATE|os.O_RDWR, 0o666) + require.NoError(t, err) + defer lockedFile.Close() + _, err = lockedFile.WriteString("locked") + require.NoError(t, err) + require.NoError(t, os.WriteFile(readyFilePath, []byte("ready"), 0o666)) + time.Sleep(time.Minute) + default: + t.Fatalf("unknown helper mode %#v", os.Getenv("BB_RE_TEST_HELPER")) + } +} From d0c1213e21fc023277155459bc09e68eca798aee Mon Sep 17 00:00:00 2001 From: alex-the-third <139810351+alex-the-third@users.noreply.github.com> Date: Sat, 22 Aug 2026 19:45:46 +0200 Subject: [PATCH 2/3] Dispatch Windows process tree helpers from TestMain Handle helper modes before running the generated test entry point. This follows rules_go's respawn-test pattern and avoids depending on -test.run behavior when re-executing the test binary. --- pkg/runner/local_runner_test.go | 73 ++++++++++++++++++--------------- 1 file changed, 39 insertions(+), 34 deletions(-) diff --git a/pkg/runner/local_runner_test.go b/pkg/runner/local_runner_test.go index c1e2ae78..7f2e29d0 100644 --- a/pkg/runner/local_runner_test.go +++ b/pkg/runner/local_runner_test.go @@ -2,6 +2,7 @@ package runner_test import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -26,6 +27,17 @@ import ( "go.uber.org/mock/gomock" ) +func TestMain(m *testing.M) { + if mode := os.Getenv("BB_RE_TEST_HELPER"); mode != "" { + if err := runWindowsProcessTreeHelper(mode); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + os.Exit(0) + } + os.Exit(m.Run()) +} + func TestLocalRunnerCheckReadiness(t *testing.T) { ctrl, ctx := gomock.WithContext(context.Background(), t) @@ -593,10 +605,7 @@ func TestLocalRunnerRunWindowsSubprocessCleanup(t *testing.T) { runner := runner.NewLocalRunner(buildDirectory, buildDirectoryPathBuilder, runner.NewPlainCommandCreator(&syscall.SysProcAttr{}), false) response, err := runner.Run(context.Background(), &runner_pb.RunRequest{ - Arguments: []string{ - testBinaryPath, - "-test.run=TestLocalRunnerRunWindowsSubprocessCleanupHelper", - }, + Arguments: []string{testBinaryPath}, EnvironmentVariables: environmentVariables, StdoutPath: testName + "/stdout", StderrPath: testName + "/stderr", @@ -659,10 +668,7 @@ func TestLocalRunnerRunWindowsCancellationCleanup(t *testing.T) { localRunner := runner.NewLocalRunner(buildDirectory, buildDirectoryPathBuilder, runner.NewPlainCommandCreator(&syscall.SysProcAttr{}), false) go func() { response, err := localRunner.Run(ctx, &runner_pb.RunRequest{ - Arguments: []string{ - testBinaryPath, - "-test.run=TestLocalRunnerRunWindowsSubprocessCleanupHelper", - }, + Arguments: []string{testBinaryPath}, EnvironmentVariables: environmentVariables, StdoutPath: testName + "/stdout", StderrPath: testName + "/stderr", @@ -700,54 +706,53 @@ func TestLocalRunnerRunWindowsCancellationCleanup(t *testing.T) { require.NoDirExists(t, rootPath) } -func TestLocalRunnerRunWindowsSubprocessCleanupHelper(t *testing.T) { +func runWindowsProcessTreeHelper(mode string) error { // parent spawns child and exits, parent_wait spawns child and remains alive, // and child locks an input-root file before signaling that it is ready. - switch os.Getenv("BB_RE_TEST_HELPER") { - case "": - return - case "parent": + switch mode { + case "parent", "parent_wait": testBinaryPath := os.Getenv("BB_RE_TEST_BINARY") if testBinaryPath == "" { - t.Fatal("BB_RE_TEST_BINARY is not set") + return fmt.Errorf("BB_RE_TEST_BINARY is not set") } - cmd := exec.Command(testBinaryPath, "-test.run=TestLocalRunnerRunWindowsSubprocessCleanupHelper") + cmd := exec.Command(testBinaryPath) cmd.Env = append(os.Environ(), "BB_RE_TEST_HELPER=child") cmd.Dir = "." - require.NoError(t, cmd.Start()) - + if err := cmd.Start(); err != nil { + return fmt.Errorf("failed to start child helper: %w", err) + } + if mode == "parent_wait" { + time.Sleep(time.Minute) + return nil + } readyFilePath := os.Getenv("BB_RE_TEST_READY_FILE") deadline := time.Now().Add(10 * time.Second) for { if _, err := os.Stat(readyFilePath); err == nil { - return + return nil } if time.Now().After(deadline) { - t.Fatal("timed out waiting for child helper") + return fmt.Errorf("timed out waiting for child helper") } time.Sleep(10 * time.Millisecond) } - case "parent_wait": - testBinaryPath := os.Getenv("BB_RE_TEST_BINARY") - if testBinaryPath == "" { - t.Fatal("BB_RE_TEST_BINARY is not set") - } - cmd := exec.Command(testBinaryPath, "-test.run=TestLocalRunnerRunWindowsSubprocessCleanupHelper") - cmd.Env = append(os.Environ(), "BB_RE_TEST_HELPER=child") - cmd.Dir = "." - require.NoError(t, cmd.Start()) - time.Sleep(time.Minute) case "child": lockedFilePath := os.Getenv("BB_RE_TEST_LOCKED_FILE") readyFilePath := os.Getenv("BB_RE_TEST_READY_FILE") lockedFile, err := os.OpenFile(lockedFilePath, os.O_CREATE|os.O_RDWR, 0o666) - require.NoError(t, err) + if err != nil { + return fmt.Errorf("failed to open locked file: %w", err) + } defer lockedFile.Close() - _, err = lockedFile.WriteString("locked") - require.NoError(t, err) - require.NoError(t, os.WriteFile(readyFilePath, []byte("ready"), 0o666)) + if _, err := lockedFile.WriteString("locked"); err != nil { + return fmt.Errorf("failed to write locked file: %w", err) + } + if err := os.WriteFile(readyFilePath, []byte("ready"), 0o666); err != nil { + return fmt.Errorf("failed to write ready file: %w", err) + } time.Sleep(time.Minute) + return nil default: - t.Fatalf("unknown helper mode %#v", os.Getenv("BB_RE_TEST_HELPER")) + return fmt.Errorf("unknown helper mode %#v", mode) } } From 5e8bd9756666110d9c6a0954de059838abd09954 Mon Sep 17 00:00:00 2001 From: alex-the-third <139810351+alex-the-third@users.noreply.github.com> Date: Sat, 22 Aug 2026 19:59:55 +0200 Subject: [PATCH 3/3] Use SysProcAttr.Jobs for Windows process trees Adopt the proposed SysProcAttr.Jobs support from Go CL 801640 in the Windows runner. Assign the action process to its job during process creation and remove the suspended-process, manual assignment, thread enumeration, and resume workaround while retaining the Buildbarn job lifecycle and cleanup policy. This change requires Go 1.28 once the proposal lands. --- pkg/runner/local_runner.go | 4 - pkg/runner/local_runner_process.go | 5 - pkg/runner/local_runner_process_windows.go | 148 ++---------------- .../local_runner_process_windows_test.go | 99 +++--------- pkg/runner/local_runner_test.go | 7 +- 5 files changed, 44 insertions(+), 219 deletions(-) diff --git a/pkg/runner/local_runner.go b/pkg/runner/local_runner.go index ba6085ea..9148f8ea 100644 --- a/pkg/runner/local_runner.go +++ b/pkg/runner/local_runner.go @@ -207,10 +207,6 @@ func (r *localRunner) Run(ctx context.Context, request *runner.RunRequest) (*run } return nil, util.StatusWrapWithCode(err, code, "Failed to start process") } - if err := commandProcess.AfterStart(cmd); err != nil { - return nil, util.StatusWrap(err, "Failed to finish process startup") - } - // Wait for execution to complete. Permit non-zero exit codes. waitErr := cmd.Wait() afterWaitErr := commandProcess.AfterWait(cmd) diff --git a/pkg/runner/local_runner_process.go b/pkg/runner/local_runner_process.go index 6548bc9e..b2acf270 100644 --- a/pkg/runner/local_runner_process.go +++ b/pkg/runner/local_runner_process.go @@ -13,11 +13,6 @@ func prepareCommandForStart(cmd *exec.Cmd) (*commandProcess, error) { return &commandProcess{}, nil } -// AfterStart is called after Start succeeds and before the caller waits. -func (commandProcess) AfterStart(cmd *exec.Cmd) error { - return nil -} - // AfterWait is called after Wait returns, before build directory cleanup can // proceed. func (commandProcess) AfterWait(cmd *exec.Cmd) error { diff --git a/pkg/runner/local_runner_process_windows.go b/pkg/runner/local_runner_process_windows.go index 831fc88d..37b104d9 100644 --- a/pkg/runner/local_runner_process_windows.go +++ b/pkg/runner/local_runner_process_windows.go @@ -4,16 +4,15 @@ // This file intentionally mirrors the native Bazel Windows launcher: // https://github.com/bazelbuild/bazel/blob/master/src/main/native/windows/process.cc // -// The root process is created suspended, assigned to a non-breakaway job, and -// only then resumed. Assigning while suspended is the important race fix: an -// unsuspended process could spawn descendants before the runner has contained -// it, leaving those descendants alive to keep the input root undeletable. +// The root process is assigned to a non-breakaway job as part of process +// creation. This prevents it from spawning descendants before the runner has +// contained it, which could leave those descendants alive to keep the input +// root undeletable. package runner import ( "errors" - "fmt" "os" "os/exec" "sync" @@ -40,20 +39,15 @@ type jobObjectAssociateCompletionPort struct { } type commandProcess struct { - lock sync.Mutex - job windows.Handle - ioport windows.Handle - assigned bool - closed bool - canceled bool - // afterCancellationCheck is a test-only hook for pausing AfterStart while - // it holds lock between checking canceled and resuming the process. - afterCancellationCheck func() + lock sync.Mutex + job windows.Handle + ioport windows.Handle + closed bool } // prepareCommandForStart creates the job object and completion port before the -// process exists. It also amends cmd so Start creates the root process suspended -// and in a new process group. The returned object owns those handles until +// process exists. It also amends cmd so Start creates the root process in the +// job and in a new process group. The returned object owns those handles until // Close or AfterWait. func prepareCommandForStart(cmd *exec.Cmd) (*commandProcess, error) { job, err := windows.CreateJobObject(nil, nil) @@ -104,7 +98,11 @@ func prepareCommandForStart(cmd *exec.Cmd) (*commandProcess, error) { if cmd.SysProcAttr != nil { sysProcAttr = *cmd.SysProcAttr } - sysProcAttr.CreationFlags |= windows.CREATE_NEW_PROCESS_GROUP | windows.CREATE_SUSPENDED + // Clone Jobs before appending, so preparing the command does not mutate the + // backing array of a caller-provided SysProcAttr. + sysProcAttr.Jobs = append([]syscall.Handle(nil), sysProcAttr.Jobs...) + sysProcAttr.Jobs = append(sysProcAttr.Jobs, syscall.Handle(job)) + sysProcAttr.CreationFlags |= windows.CREATE_NEW_PROCESS_GROUP cmd.SysProcAttr = &sysProcAttr if cmd.Cancel != nil { cmd.Cancel = p.Cancel @@ -114,64 +112,12 @@ func prepareCommandForStart(cmd *exec.Cmd) (*commandProcess, error) { return p, nil } -// Cancel may run as soon as cmd.Context is canceled, including before AfterStart -// has assigned the suspended root process to the job. In that pre-assignment -// race, record the cancellation and let AfterStart terminate the job after the -// assignment has made termination cover the whole process tree. +// Cancel terminates the root process and all of its descendants through the job +// to which Start assigned the process atomically. func (p *commandProcess) Cancel() error { - p.lock.Lock() - p.canceled = true - assigned := p.assigned - closed := p.closed - p.lock.Unlock() - if closed { - return os.ErrProcessDone - } - if !assigned { - return nil - } return p.terminateJob() } -// AfterStart assigns the still-suspended root process to the job, then either -// applies a cancellation that arrived early or resumes the process. On any -// failure after Start, it kills/reaps the partially started process tree before -// returning to the caller. -func (p *commandProcess) AfterStart(cmd *exec.Cmd) error { - processHandle, err := windows.OpenProcess( - windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, - false, - uint32(cmd.Process.Pid), - ) - if err != nil { - return p.cleanupAfterStartFailure(cmd, err, false) - } - err = windows.AssignProcessToJobObject(p.job, processHandle) - windows.CloseHandle(processHandle) - if err != nil { - return p.cleanupAfterStartFailure(cmd, err, false) - } - - p.lock.Lock() - p.assigned = true - if p.canceled { - p.lock.Unlock() - if err := p.terminateJob(); err != nil { - return p.cleanupAfterStartFailure(cmd, err, true) - } - return nil - } - if p.afterCancellationCheck != nil { - p.afterCancellationCheck() - } - err = resumeProcessThreads(uint32(cmd.Process.Pid)) - p.lock.Unlock() - if err != nil { - return p.cleanupAfterStartFailure(cmd, err, true) - } - return nil -} - // AfterWait runs after the root process has exited. It terminates anything that // remains in the job, waits for the active-process-zero notification, and only // then releases the job handles so build-directory cleanup cannot race lingering @@ -208,20 +154,6 @@ func (p *commandProcess) Close() { } } -func (p *commandProcess) cleanupAfterStartFailure(cmd *exec.Cmd, cause error, assigned bool) error { - if assigned { - _ = p.terminateJob() - } else if cmd.Process != nil { - _ = cmd.Process.Kill() - } - _ = cmd.Wait() - if assigned { - _ = p.waitForActiveProcessZero() - } - p.Close() - return cause -} - func (p *commandProcess) terminateJob() error { p.lock.Lock() closed := p.closed @@ -260,49 +192,3 @@ func (p *commandProcess) waitForActiveProcessZero() error { } } } - -// resumeProcessThreads resumes all threads belonging to the suspended root -// process. Go's os/exec path closes the primary thread handle returned by -// CreateProcess, so this code has to rediscover the thread handle by PID. -func resumeProcessThreads(pid uint32) error { - snapshot, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0) - if err != nil { - return err - } - defer windows.CloseHandle(snapshot) - - var threadHandles []windows.Handle - defer func() { - for _, threadHandle := range threadHandles { - windows.CloseHandle(threadHandle) - } - }() - - entry := windows.ThreadEntry32{ - Size: uint32(unsafe.Sizeof(windows.ThreadEntry32{})), - } - for err := windows.Thread32First(snapshot, &entry); ; err = windows.Thread32Next(snapshot, &entry) { - if err != nil { - if errors.Is(err, windows.ERROR_NO_MORE_FILES) { - break - } - return err - } - if entry.OwnerProcessID == pid { - threadHandle, err := windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, entry.ThreadID) - if err != nil { - return err - } - threadHandles = append(threadHandles, threadHandle) - } - } - if len(threadHandles) == 0 { - return fmt.Errorf("process %d has no threads to resume", pid) - } - for _, threadHandle := range threadHandles { - if _, err := windows.ResumeThread(threadHandle); err != nil { - return err - } - } - return nil -} diff --git a/pkg/runner/local_runner_process_windows_test.go b/pkg/runner/local_runner_process_windows_test.go index b866e3bb..9acef8ba 100644 --- a/pkg/runner/local_runner_process_windows_test.go +++ b/pkg/runner/local_runner_process_windows_test.go @@ -1,92 +1,43 @@ package runner import ( - "errors" - "os" "os/exec" - "path/filepath" + "syscall" "testing" - "time" + + "golang.org/x/sys/windows" ) -func TestCommandProcessCancellationBetweenCheckAndResume(t *testing.T) { - // Pause AfterStart after it has checked for cancellation, at which point it - // must still hold the mutex. Cancel must block until AfterStart resumes the - // process and releases the mutex. The old implementation released the mutex - // before resume, allowing Cancel to terminate the suspended process first. - cmd := exec.Command( - filepath.Join(os.Getenv("SYSTEMROOT"), "System32", "cmd.exe"), - "/d", "/c", "ping -n 60 127.0.0.1 > nul", - ) +func TestPrepareCommandForStartAppendsJob(t *testing.T) { + const existingJob syscall.Handle = 123 + jobs := make([]syscall.Handle, 1, 2) + jobs[0] = existingJob + cmd := exec.Command("does-not-need-to-exist") + originalSysProcAttr := &syscall.SysProcAttr{ + CreationFlags: windows.CREATE_NO_WINDOW, + Jobs: jobs, + } + cmd.SysProcAttr = originalSysProcAttr + commandProcess, err := prepareCommandForStart(cmd) if err != nil { t.Fatal(err) } + defer commandProcess.Close() - afterCancellationCheck := make(chan struct{}) - allowResume := make(chan struct{}) - commandProcess.afterCancellationCheck = func() { - close(afterCancellationCheck) - <-allowResume + if cmd.SysProcAttr == originalSysProcAttr { + t.Fatal("prepareCommandForStart() reused the caller's SysProcAttr") } - - if err := cmd.Start(); err != nil { - commandProcess.Close() - t.Fatal(err) + if got, want := cmd.SysProcAttr.CreationFlags, uint32(windows.CREATE_NO_WINDOW|windows.CREATE_NEW_PROCESS_GROUP); got != want { + t.Errorf("CreationFlags = %#x, want %#x", got, want) } - afterStartResult := make(chan error, 1) - go func() { - afterStartResult <- commandProcess.AfterStart(cmd) - }() - - select { - case <-afterCancellationCheck: - case <-time.After(10 * time.Second): - t.Fatal("timed out waiting for the cancellation check") + if got, want := cmd.SysProcAttr.Jobs, []syscall.Handle{existingJob, syscall.Handle(commandProcess.job)}; len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { + t.Errorf("Jobs = %v, want %v", got, want) } - - var cancelErr error - cancellationReturnedBeforeResume := commandProcess.lock.TryLock() - var cancelResult chan error - if cancellationReturnedBeforeResume { - commandProcess.lock.Unlock() - cancelErr = commandProcess.Cancel() - } else { - cancelResult = make(chan error, 1) - go func() { - cancelResult <- commandProcess.Cancel() - }() + if got, want := originalSysProcAttr.Jobs, []syscall.Handle{existingJob}; len(got) != len(want) || got[0] != want[0] { + t.Errorf("original Jobs = %v, want %v", got, want) } - close(allowResume) - - var afterStartErr error - select { - case afterStartErr = <-afterStartResult: - case <-time.After(10 * time.Second): - t.Fatal("AfterStart() hung") - } - if !cancellationReturnedBeforeResume { - select { - case cancelErr = <-cancelResult: - case <-time.After(10 * time.Second): - t.Fatal("Cancel() hung") - } - } - - if cancelErr != nil && !errors.Is(cancelErr, os.ErrProcessDone) { - t.Errorf("Cancel() failed: %v", cancelErr) - } - if afterStartErr != nil { - t.Errorf("AfterStart() failed: %v", afterStartErr) - } - if cancellationReturnedBeforeResume { - t.Error("cancellation terminated the suspended job before resume") - } - - if afterStartErr == nil { - _ = cmd.Wait() - if err := commandProcess.AfterWait(cmd); err != nil { - t.Errorf("AfterWait() failed: %v", err) - } + if got := jobs[:cap(jobs)][1]; got != 0 { + t.Errorf("caller-owned Jobs backing array was modified: jobs[1] = %v", got) } } diff --git a/pkg/runner/local_runner_test.go b/pkg/runner/local_runner_test.go index 7f2e29d0..a92e1788 100644 --- a/pkg/runner/local_runner_test.go +++ b/pkg/runner/local_runner_test.go @@ -696,11 +696,8 @@ func TestLocalRunnerRunWindowsCancellationCleanup(t *testing.T) { case <-time.After(10 * time.Second): t.Fatal("Run() hung after cancellation") } - if result.err != nil { - require.NotContains(t, result.err.Error(), "Failed to finish process startup") - } else { - require.NotNil(t, result.response) - } + require.NoError(t, result.err) + require.NotNil(t, result.response) require.NoError(t, os.RemoveAll(rootPath)) require.NoDirExists(t, rootPath)