[runner] Interrupt the job through the terminal, not the shell

runner/internal/runner/executor/executor.go

@@ -43,6 +43,11 @@ const (
 
 	// Maximum buffer size for ansistrip
 	MaxBufferSize = 32 * 1024 // 32KB
+
+	// intrChar is the terminal's INTR character (Ctrl-C) in the default configuration.
+	intrChar = 0x03
+	// intrWriteTimeout bounds how long writing INTR to the pty master may block.
+	intrWriteTimeout = 5 * time.Second
 )
 
 type ConnectionTracker interface {
@@ -505,13 +510,6 @@ func (ex *RunExecutor) execJob(ctx context.Context, jobLogFile io.Writer) error
 	}
 
 	cmd := exec.CommandContext(ctx, ex.jobSpec.Commands[0], ex.jobSpec.Commands[1:]...)
-	cmd.Cancel = func() error {
-		// returns error on Windows
-		if signalErr := cmd.Process.Signal(os.Interrupt); signalErr != nil {
-			return fmt.Errorf("send interrupt signal: %w", signalErr)
-		}
-		return nil
-	}
 	cmd.WaitDelay = ex.killDelay // kills the process if it doesn't exit in time
 
 	if err := utils.MkdirAll(ctx, ex.jobWorkingDir, ex.jobUser.Uid, ex.jobUser.Gid, 0o755); err != nil {
@@ -802,13 +800,50 @@ func startCommand(cmd *exec.Cmd) (*os.File, error) {
 		}
 	}
 
+	// Cancel must be set before Start, which installs the goroutine that calls it.
+	cmd.Cancel = func() error { return interruptJob(ptm) }
+
 	if err := cmd.Start(); err != nil {
 		_ = ptm.Close()
 		return nil, fmt.Errorf("start command: %w", err)
 	}
 	return ptm, nil
 }
 
+// interruptJob asks the job to stop the way Ctrl-C does. Writing the terminal's INTR character
+// to the pty master makes the line discipline raise SIGINT in the terminal's foreground process
+// group -- the command the shell is currently running, together with everything sharing its
+// process group.
+//
+// Signalling cmd.Process reaches the wrong process instead. The server runs commands under
+// `sh -i -c`, and an interactive shell turns on job control, which puts the job in a process
+// group of its own, while the shell ignores SIGINT for as long as it is waiting for that job.
+// The signal reached neither, so nothing stopped the job until WaitDelay expired and SIGKILL
+// went to the shell alone.
+//
+// The job may still ignore this: a program that puts the terminal in raw mode clears ISIG, and
+// the INTR character then delivers no signal at all. WaitDelay stays the backstop.
+func interruptJob(ptm *os.File) error {
+	// The master is pollable (see openPty), so a deadline is honoured here. Without one, a job
+	// that never reads its stdin could fill the terminal's input buffer and block this write
+	// indefinitely -- and Cmd only starts the WaitDelay timer once Cancel has returned.
+	if err := ptm.SetWriteDeadline(time.Now().Add(intrWriteTimeout)); err != nil {
+		return fmt.Errorf("set INTR write deadline: %w", err)
+	}
+	defer func() { _ = ptm.SetWriteDeadline(time.Time{}) }()
+
+	if _, err := ptm.Write([]byte{intrChar}); err != nil {
+		if isPtyError(err) || errors.Is(err, os.ErrClosed) {
+			// The terminal is gone, so the job is gone with it. Reporting the process as
+			// already done keeps Wait returning the command's own exit status rather than
+			// replacing it with the context error.
+			return fmt.Errorf("write INTR: %w", errors.Join(err, os.ErrProcessDone))
+		}
+		return fmt.Errorf("write INTR: %w", err)
+	}
+	return nil
+}
+
 func prepareUserSshDir(user *linuxuser.User) (string, error) {
 	sshDir := filepath.Join(user.HomeDir, ".ssh")
 	info, err := os.Stat(sshDir)

runner/internal/runner/executor/executor_test.go

@@ -16,6 +16,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/dstackai/dstack/runner/internal/common/types"
 	linuxuser "github.com/dstackai/dstack/runner/internal/runner/linux/user"
 	"github.com/dstackai/dstack/runner/internal/runner/schemas"
 	"github.com/stretchr/testify/assert"
@@ -140,7 +141,14 @@ func TestExecutor_MaxDuration(t *testing.T) {
 	makeCodeTar(t, ex)
 
 	err := ex.Run(t.Context())
-	assert.ErrorContains(t, err, "killed")
+	// The job is interrupted rather than killed: INTR reaches the workload through the
+	// terminal, so it exits on SIGINT long before the SIGKILL backstop would fire.
+	assert.ErrorContains(t, err, "interrupt")
+
+	history := ex.GetHistory(0)
+	lastState := history.JobStates[len(history.JobStates)-1]
+	assert.Equal(t, schemas.JobStateTerminated, lastState.State)
+	assert.Equal(t, types.TerminationReasonMaxDurationExceeded, lastState.TerminationReason)
 }
 
 func TestExecutor_LogQuota(t *testing.T) {
@@ -207,6 +215,64 @@ func TestExecutor_SurvivingProcessDoesNotHangRun(t *testing.T) {
 	assert.Contains(t, logs.String(), "done")
 }
 
+// Stopping a job must interrupt the workload, not just the wrapper shell: the workload gets a
+// chance to shut down cleanly instead of being killed once the grace period expires.
+func TestExecutor_StopInterruptsWorkload(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+
+	workload := filepath.Join(t.TempDir(), "workload.sh")
+	require.NoError(t, os.WriteFile(workload, []byte(
+		"trap 'echo graceful shutdown; exit 0' INT\n"+
+			"echo ready\n"+
+			"sleep 300\n"), 0o600))
+
+	ex := makeTestExecutor(t)
+	// The SIGKILL backstop must not be what stops the job.
+	ex.killDelay = 60 * time.Second
+	// The trailing `&& :` is load-bearing. Given a single simple command, bash and BusyBox ash
+	// exec it in place, leaving no shell at all -- and the interrupt used to reach a workload
+	// that was the direct child just fine. A command list keeps the wrapper shell, which is the
+	// arrangement that used to swallow the interrupt. Verified against bash, dash and BusyBox
+	// ash; dash does not do the optimization either way.
+	ex.jobSpec.Commands = []string{"/bin/sh", "-i", "-c", "/bin/sh " + workload + " && :"}
+	makeCodeTar(t, ex)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	runDone := make(chan error, 1)
+	go func() { runDone <- ex.Run(ctx) }()
+
+	require.Eventually(t, func() bool {
+		return strings.Contains(jobLogsSoFar(ex), "ready")
+	}, 30*time.Second, 100*time.Millisecond, "the workload never started")
+
+	cancel() // what /api/stop does
+	stoppedAt := time.Now()
+
+	select {
+	case <-runDone:
+	case <-time.After(30 * time.Second):
+		t.Fatal("Run did not return after the job was stopped")
+	}
+	assert.Less(t, time.Since(stoppedAt), ex.killDelay,
+		"the job was stopped by the SIGKILL backstop rather than by INTR")
+
+	history := ex.GetHistory(0)
+	assert.Contains(t, combineLogMessages(history.JobLogs), "graceful shutdown",
+		"the workload did not receive SIGINT")
+	lastState := history.JobStates[len(history.JobStates)-1]
+	assert.Equal(t, schemas.JobStateTerminated, lastState.State)
+}
+
+// jobLogsSoFar reads the log history while the job is still running, under the lock the
+// executor's writers share.
+func jobLogsSoFar(ex *RunExecutor) string {
+	ex.mu.RLock()
+	defer ex.mu.RUnlock()
+	return combineLogMessages(ex.jobLogs.history)
+}
+
 func TestExecutor_RemoteRepo(t *testing.T) {
 	if testing.Short() {
 		t.Skip()