From d34e4cf11a872f34cd176853a1a317026468ddb5 Mon Sep 17 00:00:00 2001 From: silverwind Date: Sat, 7 Nov 2026 08:07:41 +0000 Subject: [PATCH] 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 Reviewed-on: https://gitea.com/gitea/runner/pulls/1404 Reviewed-by: bircni Co-authored-by: silverwind --- act/container/docker_run.go | 1 + act/container/docker_run_test.go | 15 +++++ e2e/cli_test.go | 83 ++++++++++++--------------- e2e/compatibility_test.go | 1 - e2e/gitea_fixture.go | 26 +++++---- e2e/kubernetes_test.go | 20 ++++--- e2e/main_test.go | 8 ++- e2e/modes_test.go | 44 +++++++++----- e2e/testdata/workflows/cli_labels.yml | 7 --- 9 files changed, 114 insertions(+), 91 deletions(-) delete mode 100644 e2e/testdata/workflows/cli_labels.yml diff --git a/act/container/docker_run.go b/act/container/docker_run.go index b2266ee8..714c646b 100644 --- a/act/container/docker_run.go +++ b/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 { diff --git a/act/container/docker_run_test.go b/act/container/docker_run_test.go index c1fbaa61..3af5103e 100644 --- a/act/container/docker_run_test.go +++ b/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) diff --git a/e2e/cli_test.go b/e2e/cli_test.go index d295701b..97d443ee 100644 --- a/e2e/cli_test.go +++ b/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) diff --git a/e2e/compatibility_test.go b/e2e/compatibility_test.go index 7fa9331b..6b2f2c0d 100644 --- a/e2e/compatibility_test.go +++ b/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) diff --git a/e2e/gitea_fixture.go b/e2e/gitea_fixture.go index 3d19da8d..135d14ec 100644 --- a/e2e/gitea_fixture.go +++ b/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()) } diff --git a/e2e/kubernetes_test.go b/e2e/kubernetes_test.go index 33a35b05..05a55491 100644 --- a/e2e/kubernetes_test.go +++ b/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) diff --git a/e2e/main_test.go b/e2e/main_test.go index 5500987c..c746a951 100644 --- a/e2e/main_test.go +++ b/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) diff --git a/e2e/modes_test.go b/e2e/modes_test.go index 72841b1f..e46a3b6a 100644 --- a/e2e/modes_test.go +++ b/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 } diff --git a/e2e/testdata/workflows/cli_labels.yml b/e2e/testdata/workflows/cli_labels.yml deleted file mode 100644 index dfc4eb7c..00000000 --- a/e2e/testdata/workflows/cli_labels.yml +++ /dev/null @@ -1,7 +0,0 @@ -name: cli-labels -on: push -jobs: - hello: - runs-on: e2e-override - steps: - - run: echo e2e-override-ran