From 006e1a3ded2d472b4d8ad30c9366ca2bbb4207b6 Mon Sep 17 00:00:00 2001 From: Dmitry Meyer Date: Tue, 8 Sep 2026 15:31:48 +0000 Subject: [PATCH] Stop reading job output once the command has exited Previously, `execJob` copied the command's output to completion and only then waited for the command. A read on the pty master returns EIO only once every process holding the slave has closed it, so a job that leaves one behind blocked `io.Copy` forever. `cmd.Wait()` was then never reached: the shell stayed an unreaped zombie, the terminal job state was never reported, and the run hung until the container was destroyed. Now the command is waited for first, and the copy is given `logsDrainDelay` to drain what the command already wrote before the master is closed, which unblocks the read. Closing the master only works if it is pollable. `pty.Open()` wraps the descriptor with `os.NewFile` in blocking mode, so it never reaches the runtime poller, and closing such a file does not interrupt a `Read` already in flight -- the close is deferred until that read returns, which may be never. `/dev/ptmx` is now opened with `O_NONBLOCK`, which was the last use of `creack/pty`. The log quota watchdog moves into a goroutine, so output keeps being copied until the command exits and a full pty buffer cannot keep it from exiting. What gets killed is unchanged: the processes the job leaves behind still survive, and are cleaned up when the container is destroyed. Co-authored-by: Claude Opus 5 (1M context) --- runner/go.mod | 1 - runner/go.sum | 2 - runner/internal/runner/executor/executor.go | 114 +++++++++++++----- .../internal/runner/executor/executor_test.go | 58 +++++++++ 4 files changed, 143 insertions(+), 32 deletions(-) diff --git a/runner/go.mod b/runner/go.mod index f338e78edd..95f3a2f424 100644 --- a/runner/go.mod +++ b/runner/go.mod @@ -7,7 +7,6 @@ require ( github.com/alexellis/go-execute/v2 v2.2.1 github.com/bluekeyes/go-gitdiff v0.7.2 github.com/codeclysm/extract/v4 v4.0.0 - github.com/creack/pty v1.1.24 github.com/docker/docker v26.0.0+incompatible github.com/docker/go-connections v0.5.0 github.com/docker/go-units v0.5.0 diff --git a/runner/go.sum b/runner/go.sum index 655ea59dc0..9eefc2a075 100644 --- a/runner/go.sum +++ b/runner/go.sum @@ -34,8 +34,6 @@ github.com/codeclysm/extract/v4 v4.0.0 h1:H87LFsUNaJTu2e/8p/oiuiUsOK/TaPQ5wxsjPn github.com/codeclysm/extract/v4 v4.0.0/go.mod h1:SFju1lj6as7FvUgalpSct7torJE0zttbJUWtryPRG6s= github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= -github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= -github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= github.com/cyphar/filepath-securejoin v0.2.4 h1:Ugdm7cg7i6ZK6x3xDF1oEu1nfkyfH53EtKeQYTC3kyg= github.com/cyphar/filepath-securejoin v0.2.4/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/runner/internal/runner/executor/executor.go b/runner/internal/runner/executor/executor.go index 45cb429636..d78bf629fb 100644 --- a/runner/internal/runner/executor/executor.go +++ b/runner/internal/runner/executor/executor.go @@ -17,7 +17,6 @@ import ( "syscall" "time" - "github.com/creack/pty" "github.com/dstackai/ansistrip" "github.com/prometheus/procfs" "github.com/sirupsen/logrus" @@ -85,7 +84,10 @@ type RunExecutor struct { runnerLogs *appendWriter timestamp *MonotonicTimestamp - killDelay time.Duration + killDelay time.Duration + // How long output may go on being copied after the command has exited, before the pty + // master is closed. Only reached when the job leaves processes holding the terminal open. + logsDrainDelay time.Duration connectionTracker ConnectionTracker } @@ -121,6 +123,7 @@ func NewRunExecutor(tempDir string, dstackDir string, currentUser linuxuser.User timestamp: timestamp, killDelay: 10 * time.Second, + logsDrainDelay: 2 * time.Second, connectionTracker: connectionTracker, }, nil } @@ -595,38 +598,21 @@ func (ex *RunExecutor) execJob(ctx context.Context, jobLogFile io.Writer) error return fmt.Errorf("start command: %w", err) } defer func() { _ = ptm.Close() }() - defer func() { _ = cmd.Wait() }() // release resources if copy fails stripper := ansistrip.NewWriter(ex.jobLogs, AnsiStripFlushInterval, AnsiStripMaxDelay, MaxBufferSize) logger := io.MultiWriter(jobLogFile, ex.jobWsLogs, stripper) - if err := ex.copyOutputWithQuota(cmd, ptm, stripper, logger); err != nil { - return err - } - if err = cmd.Wait(); err != nil { - return fmt.Errorf("wait for command: %w", err) - } - return nil -} - -// copyOutputWithQuota streams process output through the log pipeline and -// monitors for log quota exceeded. The quota signal is out-of-band (via channel) -// because the ansistrip writer is async and swallows downstream write errors. -func (ex *RunExecutor) copyOutputWithQuota(cmd *exec.Cmd, ptm io.Reader, stripper io.Closer, logger io.Writer) error { copyDone := make(chan error, 1) go func() { - _, err := io.Copy(logger, ptm) - copyDone <- err + _, copyErr := io.Copy(logger, ptm) + copyDone <- copyErr }() - // Wait for either io.Copy to finish or quota to be exceeded. - var copyErr error - select { - case copyErr = <-copyDone: - case <-ex.jobLogs.QuotaExceeded(): - _ = cmd.Process.Kill() - <-copyDone - } + stopQuotaWatch := watchLogQuota(cmd, ex.jobLogs.QuotaExceeded()) + defer stopQuotaWatch() + + waitErr := cmd.Wait() + copyErr := ex.finishOutputCopy(ctx, ptm, copyDone) // Flush the ansistrip buffer — may also trigger quota exceeded. _ = stripper.Close() @@ -636,13 +622,52 @@ func (ex *RunExecutor) copyOutputWithQuota(cmd *exec.Cmd, ptm io.Reader, strippe return ErrLogQuotaExceeded default: } - if copyErr != nil && !isPtyError(copyErr) { return fmt.Errorf("copy command output: %w", copyErr) } + if waitErr != nil { + return fmt.Errorf("wait for command: %w", waitErr) + } return nil } +// finishOutputCopy waits for the output copy to finish, bounding how long it may run after +// the command has exited. +// +// A read on the pty master returns EIO only once every process holding the slave has closed +// it. A job that leaves a process behind -- a `cmd &` job, a daemon -- would otherwise keep +// the copy running forever, and with it the executor: the job state would never be reported +// and the run would hang until the container is destroyed. Give the output the command has +// already written a moment to drain, then close the master, which unblocks the read. +func (ex *RunExecutor) finishOutputCopy(ctx context.Context, ptm *os.File, copyDone <-chan error) error { + select { + case copyErr := <-copyDone: + return copyErr + case <-time.After(ex.logsDrainDelay): + } + log.Warning(ctx, "The job left processes holding the terminal open, stopped reading output") + _ = ptm.Close() + <-copyDone // fails with os.ErrClosed, which is what closing the master is for + return nil +} + +// watchLogQuota kills the command if the job exceeds its log quota. Output keeps being copied +// until the command exits, so a full pty buffer cannot keep it from exiting. +// +// The quota signal is out-of-band (via channel) because the ansistrip writer is async and +// swallows downstream write errors. +func watchLogQuota(cmd *exec.Cmd, quotaExceeded <-chan struct{}) (stop func()) { + done := make(chan struct{}) + go func() { + select { + case <-quotaExceeded: + _ = cmd.Process.Kill() + case <-done: + } + }() + return func() { close(done) } +} + // setupGitCredentials must be called from Run after setJobUser func (ex *RunExecutor) setupGitCredentials(ctx context.Context) (func(), error) { if ex.repoCredentials == nil { @@ -704,6 +729,37 @@ func (ex *RunExecutor) setupGitCredentials(ctx context.Context) (func(), error) return nil, fmt.Errorf("unknown protocol %s", ex.repoCredentials.GetProtocol()) } +// openPty opens a new pty pair. +// +// The master is opened non-blocking so that Go registers it with the runtime poller. A +// blocking os.File never reaches the poller, and closing one does not interrupt a Read already +// in flight -- the close is deferred until that read returns, which may be never. execJob +// relies on closing the master to stop reading output. +func openPty() (*os.File, *os.File, error) { + ptmFd, err := unix.Open("/dev/ptmx", unix.O_RDWR|unix.O_NOCTTY|unix.O_NONBLOCK, 0) + if err != nil { + return nil, nil, fmt.Errorf("open pty master: %w", err) + } + ptm := os.NewFile(uintptr(ptmFd), "/dev/ptmx") + + if err := unix.IoctlSetPointerInt(ptmFd, unix.TIOCSPTLCK, 0); err != nil { + _ = ptm.Close() + return nil, nil, fmt.Errorf("unlock pty slave: %w", err) + } + ptsNum, err := unix.IoctlGetInt(ptmFd, unix.TIOCGPTN) + if err != nil { + _ = ptm.Close() + return nil, nil, fmt.Errorf("get pty slave number: %w", err) + } + ptsName := fmt.Sprintf("/dev/pts/%d", ptsNum) + pts, err := os.OpenFile(ptsName, os.O_RDWR|unix.O_NOCTTY, 0) + if err != nil { + _ = ptm.Close() + return nil, nil, fmt.Errorf("open pty slave: %w", err) + } + return ptm, pts, nil +} + func isPtyError(err error) bool { /* read /dev/ptmx: input/output error */ var e *os.PathError @@ -715,9 +771,9 @@ func isPtyError(err error) bool { // * controlling terminal is properly set (cmd.Extrafiles, Cmd.SysProcAttr.Ctty) // * owner of slave pty is changed to the child process uid func startCommand(cmd *exec.Cmd) (*os.File, error) { - ptm, pts, err := pty.Open() + ptm, pts, err := openPty() if err != nil { - return nil, fmt.Errorf("open pty: %w", err) + return nil, err } defer func() { _ = pts.Close() }() diff --git a/runner/internal/runner/executor/executor_test.go b/runner/internal/runner/executor/executor_test.go index 90a942328a..acbf49381a 100644 --- a/runner/internal/runner/executor/executor_test.go +++ b/runner/internal/runner/executor/executor_test.go @@ -10,7 +10,9 @@ import ( "os/exec" "path" "path/filepath" + "strconv" "strings" + "syscall" "testing" "time" @@ -162,6 +164,49 @@ func TestExecutor_LogQuota(t *testing.T) { assert.Equal(t, schemas.JobStateFailed, lastState.State) } +// A job that leaves a process behind keeps the pty slave open, so reading the master never +// returns EIO. The executor must stop reading anyway instead of hanging forever. +func TestExecutor_SurvivingProcessDoesNotHangRun(t *testing.T) { + if testing.Short() { + t.Skip() + } + + ex := makeTestExecutor(t) + ex.logsDrainDelay = 500 * time.Millisecond + // `-i` as the server sends it: job control puts the backgrounded process in its own + // process group, so it does not get the SIGHUP the kernel sends to the foreground group + // when the shell exits, and goes on holding the pty slave open. It must outlive the + // assertion below, or the executor would be let off the hook by the process exiting. + pidPath := filepath.Join(t.TempDir(), "survivor.pid") + ex.jobSpec.Commands = []string{ + "/bin/bash", "-i", "-c", + fmt.Sprintf("sleep 300 & echo $! > %s; echo done", pidPath), + } + t.Cleanup(func() { killRecordedPid(t, pidPath) }) + makeCodeTar(t, ex) + + runDone := make(chan error, 1) + go func() { runDone <- ex.Run(t.Context()) }() + + select { + case err := <-runDone: + assert.NoError(t, err) + case <-time.After(20 * time.Second): + t.Fatal("Run did not return while a process left by the job held the terminal open") + } + + history := ex.GetHistory(0) + lastState := history.JobStates[len(history.JobStates)-1] + assert.Equal(t, schemas.JobStateDone, lastState.State) + + // Output written before the command exited must still be drained. + var logs strings.Builder + for _, event := range history.JobLogs { + logs.Write(event.Message) + } + assert.Contains(t, logs.String(), "done") +} + func TestExecutor_RemoteRepo(t *testing.T) { if testing.Short() { t.Skip() @@ -482,3 +527,16 @@ func combineLogMessages(logHistory []schemas.LogEvent) string { } return logOutput.String() } + +func killRecordedPid(t *testing.T, pidPath string) { + t.Helper() + data, err := os.ReadFile(pidPath) + if err != nil { + return + } + pid, err := strconv.Atoi(strings.TrimSpace(string(data))) + if err != nil { + return + } + _ = syscall.Kill(pid, syscall.SIGKILL) +}