fix: stop a stalled docker daemon from hanging jobs and the e2e suite (#1404)

When the docker daemon stopped answering on https://gitea.com/gitea/runner/actions/runs/1204971, unbounded exec runs hung the e2e suite until the 20m `go test` panic, which skipped teardown. Exec runs and teardown are now bounded, and attached exec streams, including the runner's user ID probe, close with their context.

CLI daemons now stop with SIGTERM, so job volumes no longer leak.

_Written by Claude (Opus 5.5)._

---------

Co-authored-by: bircni <[email protected]>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1404
Reviewed-by: bircni <[email protected]>
Co-authored-by: silverwind <[email protected]>

act/container/docker_run.go

@@ -895,6 +895,7 @@ func (cr *containerReference) tryReadID(opt string, cbk func(id int)) common.Exe
 			return nil
 		}
 		defer resp.Close()
+		defer context.AfterFunc(ctx, resp.Close)()
 
 		sid, err := resp.Reader.ReadString('\n')
 		if err != nil {

act/container/docker_run_test.go

@@ -378,6 +378,21 @@ func TestDockerAttachStreamClosesWithItsContext(t *testing.T) {
 	client.AssertExpectations(t)
 }
 
+func TestDockerReadIDStreamClosesWithItsContext(t *testing.T) {
+	conn, peer := net.Pipe()
+	t.Cleanup(func() { _ = peer.Close() })
+	client := &mockDockerClient{}
+	client.On("ExecCreate", mock.Anything, "123", mock.AnythingOfType("client.ExecCreateOptions")).
+		Return(mobyclient.ExecCreateResult{ID: "id"}, nil)
+	client.On("ExecAttach", mock.Anything, "id", mobyclient.ExecAttachOptions{}).
+		Return(mobyclient.ExecAttachResult{HijackedResponse: mobyclient.NewHijackedResponse(conn, "")}, nil)
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	require.NoError(t, (&containerReference{id: "123", cli: client}).tryReadID("-u", func(int) { t.Error("read an ID after cancellation") })(ctx))
+	client.AssertExpectations(t)
+}
+
 func TestDockerWaitFailure(t *testing.T) {
 	ctx := t.Context()
 	statusCh := make(chan container.WaitResponse, 1)

e2e/cli_test.go

@@ -90,20 +90,29 @@ container:
 type cliDaemon struct {
 	cmd    *exec.Cmd
 	log    *syncBuffer
-	exited chan error
+	exited chan struct{}
+	err    error
 }
 
 func (c *cliRunner) startDaemon(t *testing.T, args ...string) *cliDaemon {
 	t.Helper()
-	d := &cliDaemon{log: &syncBuffer{}, exited: make(chan error, 1)}
+	d := &cliDaemon{log: &syncBuffer{}, exited: make(chan struct{})}
 	d.cmd = exec.Command(c.bin, append([]string{"daemon", "-c", c.cfgFile}, args...)...)
 	d.cmd.Stdout, d.cmd.Stderr = d.log, d.log
 	if err := d.cmd.Start(); err != nil {
 		t.Fatalf("start daemon: %v", err)
 	}
-	go func() { d.exited <- d.cmd.Wait() }()
+	go func() {
+		d.err = d.cmd.Wait()
+		close(d.exited)
+	}()
 	t.Cleanup(func() {
-		_ = d.cmd.Process.Kill() // no-op once the daemon has exited
+		_ = d.cmd.Process.Signal(syscall.SIGTERM)
+		select {
+		case <-d.exited:
+		case <-time.After(30 * time.Second):
+			_ = d.cmd.Process.Kill()
+		}
 	})
 	return d
 }
@@ -112,8 +121,8 @@ func (c *cliRunner) startDaemon(t *testing.T, args ...string) *cliDaemon {
 func (d *cliDaemon) wait(t *testing.T, timeout time.Duration) error {
 	t.Helper()
 	select {
-	case err := <-d.exited:
-		return err
+	case <-d.exited:
+		return d.err
 	case <-time.After(timeout):
 		t.Fatalf("daemon still running after %s\n%s", timeout, d.log)
 		return nil
@@ -140,8 +149,8 @@ func (d *cliDaemon) requireSuccess(t *testing.T, c *cliRunner, runID int64) {
 // testCLIRegisterDaemon drives the built binary the way a user does: register, then daemon.
 func testCLIRegisterDaemon(t *testing.T) {
 	t.Parallel()
-	c := newCLIRunner(t, "e2e-cli", "")
-	d := c.startDaemon(t)
+	c := newCLIRunner(t, "e2e-registered", "")
+	d := c.startDaemon(t, "--labels", "e2e-cli:docker://"+os.Getenv("E2E_JOB_IMAGE"))
 
 	pushWorkflow(t, c.api, c.repo, "cli.yml")
 	wfRun := waitForRun(t, c.api, c.repo)
@@ -150,6 +159,22 @@ func testCLIRegisterDaemon(t *testing.T) {
 		t.Errorf("job logs are missing the marker:\n%s", logs)
 	}
 
+	runners, err := c.api.Runners(t.Context(), c.repo)
+	if err != nil {
+		t.Fatalf("list runners: %v", err)
+	}
+	var labels []string
+	for _, runner := range runners {
+		if runner.Name == c.repo {
+			for _, label := range runner.Labels {
+				labels = append(labels, label.Name)
+			}
+		}
+	}
+	if !slices.Equal(labels, []string{"e2e-cli"}) {
+		t.Errorf("runner %q has labels %v on the server, want only the --labels override", c.repo, labels)
+	}
+
 	d.terminate(t)
 	if err := d.wait(t, 30*time.Second); err != nil {
 		t.Errorf("daemon did not exit cleanly on SIGTERM: %v\n%s", err, d.log)
@@ -186,36 +211,6 @@ func testCLIOnce(t *testing.T) {
 	}
 }
 
-// testCLILabelsOverride checks that daemon --labels replaces the labels given at registration.
-func testCLILabelsOverride(t *testing.T) {
-	t.Parallel()
-	c := newCLIRunner(t, "e2e-registered", "")
-	d := c.startDaemon(t, "--labels", "e2e-override:docker://"+os.Getenv("E2E_JOB_IMAGE"))
-
-	pushWorkflow(t, c.api, c.repo, "cli_labels.yml")
-	wfRun := waitForRun(t, c.api, c.repo)
-	d.requireSuccess(t, c, wfRun.ID)
-
-	runners, err := c.api.Runners(t.Context(), c.repo)
-	if err != nil {
-		t.Fatalf("list runners: %v", err)
-	}
-	for _, runner := range runners {
-		if runner.Name != c.repo {
-			continue
-		}
-		var labels []string
-		for _, label := range runner.Labels {
-			labels = append(labels, label.Name)
-		}
-		if !slices.Equal(labels, []string{"e2e-override"}) {
-			t.Errorf("runner has labels %v on the server, want only the --labels override", labels)
-		}
-		return
-	}
-	t.Errorf("runner %q not found among %+v", c.repo, runners)
-}
-
 // testCLIShutdownDrains checks that SIGTERM lets a running job finish within shutdown_timeout.
 func testCLIShutdownDrains(t *testing.T) {
 	t.Parallel()
@@ -282,10 +277,6 @@ func testExecArtifacts(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
 
-	bin, err := runnerBinary()
-	if err != nil {
-		t.Fatal(err)
-	}
 	workflow, err := os.ReadFile("testdata/workflows/exec_artifact.yml")
 	if err != nil {
 		t.Fatal(err)
@@ -309,11 +300,9 @@ func testExecArtifacts(t *testing.T) {
 	}
 
 	artifacts := t.TempDir()
-	args := []string{"exec", "-C", workdir, "-W", workflows, "-i", os.Getenv("E2E_JOB_IMAGE"), "--artifact-server-path", artifacts, "--network", fixture.network}
-	cmd := exec.CommandContext(ctx, bin, args...)
-	cmd.Env = slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") })
-	if out, err := cmd.CombinedOutput(); err != nil {
-		t.Fatalf("exec: %v\n%s", err, out)
+	stdout, stderr, err := runExec(t, nil, "-C", workdir, "-W", workflows, "-i", os.Getenv("E2E_JOB_IMAGE"), "--artifact-server-path", artifacts, "--network", fixture.network)
+	if err != nil {
+		t.Fatalf("exec: %v\n%s%s", err, stdout, stderr)
 	}
 	if blobs, _ := filepath.Glob(filepath.Join(artifacts, "v4", "*", "*", "blob")); len(blobs) != 2 {
 		t.Fatalf("want one v4 artifact blob per workflow run, got %v", blobs)

e2e/compatibility_test.go

@@ -37,7 +37,6 @@ func TestCompatibility(t *testing.T) {
 	t.Run("job_timeout", testJobTimeout)
 	t.Run("cli_register_daemon", testCLIRegisterDaemon)
 	t.Run("cli_once", testCLIOnce)
-	t.Run("cli_labels_override", testCLILabelsOverride)
 	t.Run("cli_shutdown_drains", testCLIShutdownDrains)
 	t.Run("cli_shutdown_cancels", testCLIShutdownCancels)
 	t.Run("host_mode", testHostMode)

e2e/gitea_fixture.go

@@ -80,11 +80,11 @@ func selfNetwork(ctx context.Context, cli mobyclient.APIClient) (string, string,
 	return "", "", false
 }
 
-func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, error) {
-	closeClient := true
+func StartGitea(ctx context.Context, cli mobyclient.APIClient) (_ *GiteaFixture, err error) {
+	f := &GiteaFixture{cli: cli}
 	defer func() {
-		if closeClient {
-			_ = cli.Close()
+		if err != nil {
+			_ = f.Close()
 		}
 	}()
 
@@ -123,26 +123,21 @@ func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, e
 		return nil, fmt.Errorf("create gitea container: %w", err)
 	}
 
-	f := &GiteaFixture{cli: cli, id: resp.ID, image: image, baseURL: baseURL, network: sharedNet, selfIP: selfIP}
-	closeClient = false
+	f.id, f.image, f.baseURL, f.network, f.selfIP = resp.ID, image, baseURL, sharedNet, selfIP
 
 	if _, err := cli.ContainerStart(ctx, f.id, mobyclient.ContainerStartOptions{}); err != nil {
-		_ = f.Close(ctx)
 		return nil, fmt.Errorf("start gitea container: %w", err)
 	}
 
 	if err := f.waitHealthy(ctx); err != nil {
-		_ = f.Close(ctx)
 		return nil, fmt.Errorf("gitea did not become healthy: %w", err)
 	}
 
 	if err := f.bootstrapAdmin(ctx); err != nil {
-		_ = f.Close(ctx)
 		return nil, fmt.Errorf("bootstrap gitea admin: %w", err)
 	}
 
 	if err := f.readVersion(ctx); err != nil {
-		_ = f.Close(ctx)
 		return nil, fmt.Errorf("read gitea version: %w", err)
 	}
 
@@ -249,6 +244,7 @@ func containerExec(ctx context.Context, cli mobyclient.APIClient, id, user strin
 	if err != nil {
 		return "", err
 	}
+	defer context.AfterFunc(ctx, attached.Close)()
 	defer attached.Close()
 	if stdin != nil {
 		go func() {
@@ -258,7 +254,11 @@ func containerExec(ctx context.Context, cli mobyclient.APIClient, id, user strin
 	}
 
 	var out bytes.Buffer
-	if _, err := stdcopy.StdCopy(&out, &out, attached.Reader); err != nil {
+	_, err = stdcopy.StdCopy(&out, &out, attached.Reader)
+	if ctx.Err() != nil {
+		return "", ctx.Err()
+	}
+	if err != nil {
 		return "", err
 	}
 
@@ -291,10 +291,12 @@ func (f *GiteaFixture) doJSON(ctx context.Context, method, path string, reqBody,
 	return api.doJSON(ctx, method, path, reqBody, respBody)
 }
 
-func (f *GiteaFixture) Close(ctx context.Context) error {
+func (f *GiteaFixture) Close() error {
 	if f.id == "" {
 		return f.cli.Close()
 	}
+	ctx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
+	defer cancel()
 	_, removeErr := f.cli.ContainerRemove(ctx, f.id, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true})
 	return errors.Join(removeErr, f.cli.Close())
 }

e2e/kubernetes_test.go

@@ -55,7 +55,9 @@ func checkClusterLeaks() error {
 	if err != nil {
 		return nil // the tests that needed the cluster already failed
 	}
-	if out, err := containerExec(context.Background(), dockerCLI, cluster.id, "", nil,
+	ctx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
+	defer cancel()
+	if out, err := containerExec(ctx, dockerCLI, cluster.id, "", nil,
 		"kubectl", "wait", "--for=delete", "pods,secrets", "--selector=app.kubernetes.io/managed-by=gitea-runner", "--timeout=1m",
 	); err != nil {
 		return fmt.Errorf("job pods or secrets left behind: %w: %s", err, out)
@@ -72,7 +74,7 @@ type kubeCluster struct {
 var (
 	k3s            = sync.OnceValues(startK3s) // independent of Gitea, so it can boot alongside it
 	clusterStarted atomic.Bool
-	clusterCleanup []func()
+	clusterCleanup []func(context.Context)
 	sharedCluster  = sync.OnceValues(func() (*kubeCluster, error) {
 		cluster, err := k3s()
 		if err != nil {
@@ -109,7 +111,9 @@ func stopCluster() {
 	}
 	_, _ = k3s()
 	for _, cleanup := range slices.Backward(clusterCleanup) {
-		cleanup()
+		ctx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
+		cleanup(ctx)
+		cancel()
 	}
 }
 
@@ -138,8 +142,8 @@ func startK3s() (*kubeCluster, error) {
 	if err != nil {
 		return nil, fmt.Errorf("create k3s container: %w", err)
 	}
-	clusterCleanup = append(clusterCleanup, func() {
-		_, _ = cli.ContainerRemove(context.Background(), created.ID, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true})
+	clusterCleanup = append(clusterCleanup, func(ctx context.Context) {
+		_, _ = cli.ContainerRemove(ctx, created.ID, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true})
 	})
 	if _, err := cli.ContainerStart(ctx, created.ID, mobyclient.ContainerStartOptions{}); err != nil {
 		return nil, fmt.Errorf("start k3s container: %w", err)
@@ -170,7 +174,9 @@ func startK3s() (*kubeCluster, error) {
 		inspected, inspectErr := cli.ContainerInspect(ctx, created.ID, mobyclient.ContainerInspectOptions{})
 		if inspectErr != nil || !inspected.Container.State.Running || ctx.Err() != nil {
 			var logs bytes.Buffer
-			if reader, logsErr := cli.ContainerLogs(context.Background(), created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil {
+			logsCtx, logsCancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer logsCancel()
+			if reader, logsErr := cli.ContainerLogs(logsCtx, created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil {
 				_, _ = stdcopy.StdCopy(&logs, &logs, reader)
 				_ = reader.Close()
 			}
@@ -226,7 +232,7 @@ func preloadImages(ctx context.Context, k3sID string, images ...string) ([]strin
 		if _, err := dockerCLI.ImageTag(ctx, mobyclient.ImageTagOptions{Source: image, Target: alias}); err != nil {
 			return nil, fmt.Errorf("tag %s: %w", image, err)
 		}
-		clusterCleanup = append(clusterCleanup, func() { _, _ = dockerCLI.ImageRemove(context.Background(), alias, mobyclient.ImageRemoveOptions{}) })
+		clusterCleanup = append(clusterCleanup, func(ctx context.Context) { _, _ = dockerCLI.ImageRemove(ctx, alias, mobyclient.ImageRemoveOptions{}) })
 		aliases = append(aliases, alias)
 	}
 	archive, err := dockerCLI.ImageSave(ctx, aliases)

e2e/main_test.go

@@ -31,6 +31,8 @@ import (
 
 const runTimeout = 3 * time.Minute
 
+const cleanupTimeout = 2 * time.Minute
+
 var nonRepoChars = regexp.MustCompile(`[^a-zA-Z0-9._-]+`)
 
 var fixture *GiteaFixture
@@ -42,8 +44,10 @@ var binDir string
 var dockerCLI mobyclient.APIClient
 
 var runnerBinary = sync.OnceValues(func() (string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), runTimeout)
+	defer cancel()
 	bin := filepath.Join(binDir, "gitea-runner")
-	if out, err := exec.Command(cmp.Or(os.Getenv("GO"), "go"), "build", "-o", bin, "..").CombinedOutput(); err != nil {
+	if out, err := exec.CommandContext(ctx, cmp.Or(os.Getenv("GO"), "go"), "build", "-o", bin, "..").CombinedOutput(); err != nil {
 		return "", fmt.Errorf("build runner: %w\n%s", err, out)
 	}
 	return bin, nil
@@ -80,7 +84,7 @@ func runSuite(m *testing.M) int {
 		return 1
 	}
 	fixture = f
-	defer func() { _ = fixture.Close(context.Background()) }()
+	defer func() { _ = fixture.Close() }()
 
 	fmt.Fprintf(os.Stderr, "gitea fixture: image=%s version=%s\n", fixture.image, fixture.version)
 

e2e/modes_test.go

@@ -9,6 +9,7 @@ import (
 	"bufio"
 	"bytes"
 	"cmp"
+	"context"
 	"encoding/json"
 	"errors"
 	"os"
@@ -138,14 +139,31 @@ func daemonWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
 	return result
 }
 
-func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
+func runExec(t *testing.T, env []string, args ...string) (stdout, stderr []byte, err error) {
 	t.Helper()
-	ctx := t.Context()
-
 	bin, err := runnerBinary()
 	if err != nil {
 		t.Fatal(err)
 	}
+	ctx, cancel := context.WithTimeout(t.Context(), runTimeout)
+	defer cancel()
+	cmd := exec.CommandContext(ctx, bin, append([]string{"exec"}, args...)...)
+	cmd.Env = append(slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") }), env...)
+	cmd.Cancel = func() error { return cmd.Process.Signal(os.Interrupt) }
+	cmd.WaitDelay = cleanupTimeout
+	var stdoutBuf, stderrBuf bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &stdoutBuf, &stderrBuf
+	err = cmd.Run()
+	if ctx.Err() != nil {
+		t.Fatalf("exec did not finish within %s\n%s%s", runTimeout, stdoutBuf.Bytes(), stderrBuf.Bytes())
+	}
+	return stdoutBuf.Bytes(), stderrBuf.Bytes(), err
+}
+
+func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
+	t.Helper()
+	ctx := t.Context()
+
 	workdir := t.TempDir()
 	workflows := filepath.Join(workdir, ".gitea", "workflows")
 	if err := os.MkdirAll(workflows, 0o755); err != nil {
@@ -170,12 +188,12 @@ func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
 		}
 	}
 
-	args := []string{"exec", "--json", "-C", workdir, "-W", workflows, "--artifact-server-path", t.TempDir(), "--artifact-server-port", "0"}
-	env := slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") })
+	args := []string{"--json", "-C", workdir, "-W", workflows, "--artifact-server-path", t.TempDir(), "--artifact-server-port", "0"}
+	var env []string
 	if m.kubernetes {
 		cluster := requireCluster(t)
 		args = append(args, "-i", "kubernetes://"+cluster.jobImage)
-		env = append(env, "KUBECONFIG="+cluster.kubernetes.Kubeconfig)
+		env = []string{"KUBECONFIG=" + cluster.kubernetes.Kubeconfig}
 	} else {
 		args = append(args, "-i", os.Getenv("E2E_JOB_IMAGE"), "--network", fixture.network)
 	}
@@ -185,21 +203,17 @@ func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
 			args = append(args, "--input", name+"="+value)
 		}
 	}
-	cmd := exec.CommandContext(ctx, bin, append(args, spec.execArgs...)...)
-	cmd.Env = env
-	var stdout, stderr bytes.Buffer
-	cmd.Stdout, cmd.Stderr = &stdout, &stderr
-	runErr := cmd.Run()
+	stdout, stderr, runErr := runExec(t, env, append(args, spec.execArgs...)...)
 	var exitErr *exec.ExitError
 	if runErr != nil && !errors.As(runErr, &exitErr) {
-		t.Fatalf("exec: %v\n%s", runErr, stderr.String())
+		t.Fatalf("exec: %v\n%s", runErr, stderr)
 	}
 
 	result := &workflowResult{conclusion: "success", jobs: map[string]string{}}
 	if runErr != nil {
 		result.conclusion = "failure"
 	}
-	scanner := bufio.NewScanner(&stdout)
+	scanner := bufio.NewScanner(bytes.NewReader(stdout))
 	scanner.Buffer(nil, 1<<20)
 	for scanner.Scan() {
 		var entry struct {
@@ -220,8 +234,8 @@ func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
 			}
 		}
 	}
-	if result.conclusion != "success" && len(result.jobs) == 0 {
-		t.Fatalf("exec ran no job: %v\n%s", runErr, stderr.String())
+	if len(result.jobs) == 0 {
+		t.Fatalf("exec ran no job: %v\n%s", runErr, stderr)
 	}
 	return result
 }

e2e/testdata/workflows/cli_labels.yml

@@ -1,7 +0,0 @@
-name: cli-labels
-on: push
-jobs:
-  hello:
-    runs-on: e2e-override
-    steps:
-      - run: echo e2e-override-ran