[runner] Stop reading job output once the command has exited

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

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=

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

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