test(e2e): speed up and cover exec and Kubernetes modes (#1363)

Follow-up to https://gitea.com/gitea/runner/pulls/1362, whose crash e2e missed because it only drove the daemon on Docker.

- `dispatch`, `failure_paths` and `reusable_workflow` run through the daemon and `exec`, each on Docker and one shared k3s cluster.
- Fixed sleeps become gates the test releases, the job timeout uses a 15s runner deadline, k3s boots alongside Gitea, Gitea's data is on tmpfs.
- Outside a container, `make test-e2e` runs `go test` in a Linux container on its own network like CI, so local runs take the same paths.

Local wall time:

| | Before | After |
|---|---|---|
| Suite | 84s | 27–35s |
| `job_timeout` | 64s | 17s |
| `reconnect` | 22s | 4.5s |
| `capacity/serial` | 20s | 2s |
| `cli_shutdown_drains` | 14s | 2.3s |
| `kubernetes` | 31s | 21–23s |

_Written by Claude (Opus 5.5)._

---------

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

e2e/cancel_test.go

@@ -28,7 +28,7 @@ func testRunCancellation(t *testing.T) {
 	pushWorkflow(t, api, repo, "cancel.yml")
 
 	wfRun := waitForRun(t, api, repo)
-	waitForRunningJobLog(t, api, repo, wfRun.ID, "e2e-live-log-marker")
+	waitForRunningJobLog(t, api, repo, wfRun.ID, "e2e-live-log-marker", 1)
 	if !cancelSupported {
 		requireSuccess(t, api, repo, wfRun.ID)
 		return
@@ -49,31 +49,32 @@ func testRunCancellation(t *testing.T) {
 	}
 }
 
-func waitForRunningJobLog(t *testing.T, api *GiteaAPI, repo string, runID int64, substr string) {
+func waitForRunningJobLog(t *testing.T, api *GiteaAPI, repo string, runID int64, prefix string, count int) {
 	t.Helper()
 	// The budget covers scheduling and the container start, which a loaded runner makes slow.
 	ctx, cancel := context.WithTimeout(t.Context(), runTimeout)
 	defer cancel()
 
-	status := "none"
 	for {
 		jobs, err := api.Jobs(ctx, repo, runID)
 		if err != nil {
 			t.Fatalf("list jobs: %v", err)
 		}
-		if len(jobs) > 0 {
-			status = jobs[0].Status
-			if status == "in_progress" {
-				logs, err := api.JobLogs(ctx, repo, jobs[0].ID)
-				if err == nil && commandRow(logs, substr) == substr {
-					return
+		logged := 0
+		for _, job := range jobs {
+			if job.Status == "in_progress" {
+				if logs, err := api.JobLogs(ctx, repo, job.ID); err == nil && commandRow(logs, prefix) != "" {
+					logged++
 				}
 			}
 		}
+		if logged >= count {
+			return
+		}
 		select {
 		case <-ctx.Done():
 			dumpRunLogs(t, api, repo, runID)
-			t.Fatalf("job of run %d is %q and never logged %q", runID, status, substr)
+			t.Fatalf("%d of %d running jobs of run %d logged %q", logged, count, runID, prefix)
 		case <-time.After(pollInterval):
 		}
 	}

e2e/cli_test.go

@@ -67,10 +67,9 @@ func newCLIRunner(t *testing.T, label, extraRunnerConfig string) *cliRunner {
   log_report_interval: 500ms
 %scache:
   dir: %s
-`, filepath.Join(dir, ".runner"), extraRunnerConfig, filepath.Join(dir, "cache"))
-	if fixture.network != "" {
-		cfg += "container:\n  network: " + fixture.network + "\n"
-	}
+container:
+  network: %s
+`, filepath.Join(dir, ".runner"), extraRunnerConfig, filepath.Join(dir, "cache"), fixture.network)
 	c := &cliRunner{api: api, repo: repo, bin: bin, cfgFile: filepath.Join(dir, "config.yaml"), dir: dir}
 	if err := os.WriteFile(c.cfgFile, []byte(cfg), 0o600); err != nil {
 		t.Fatalf("write config: %v", err)
@@ -221,13 +220,26 @@ func testCLILabelsOverride(t *testing.T) {
 func testCLIShutdownDrains(t *testing.T) {
 	t.Parallel()
 	c := newCLIRunner(t, "e2e-shutdown", "  shutdown_timeout: 1m\n")
+	gate, release := newGate(t)
+	if err := c.api.CreateVariable(t.Context(), c.repo, "E2E_GATE", gate); err != nil {
+		t.Fatalf("create gate variable: %v", err)
+	}
 	d := c.startDaemon(t)
 
 	pushWorkflow(t, c.api, c.repo, "cli_shutdown.yml")
 	wfRun := waitForRun(t, c.api, c.repo)
-	waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started")
+	waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started", 1)
 
 	d.terminate(t)
+	deadline := time.After(time.Minute)
+	for !strings.Contains(d.log.String(), "shutdown initiated") {
+		select {
+		case <-deadline:
+			t.Fatalf("daemon never started shutting down\n%s", d.log)
+		case <-time.After(pollInterval):
+		}
+	}
+	release()
 	if err := d.wait(t, runTimeout); err != nil {
 		t.Errorf("daemon did not exit cleanly: %v\n%s", err, d.log)
 	}
@@ -248,7 +260,7 @@ func testCLIShutdownCancels(t *testing.T) {
 
 	pushWorkflow(t, c.api, c.repo, "cli_shutdown.yml")
 	wfRun := waitForRun(t, c.api, c.repo)
-	waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started")
+	waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started", 1)
 
 	d.terminate(t)
 	if err := d.wait(t, runTimeout); err != nil {
@@ -266,7 +278,7 @@ func testCLIShutdownCancels(t *testing.T) {
 	}
 }
 
-func testExec(t *testing.T) {
+func testExecArtifacts(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
 
@@ -274,25 +286,18 @@ func testExec(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	workflows, workdir := t.TempDir(), t.TempDir()
-	for dst, src := range map[string]string{
-		filepath.Join(workflows, "a.yml"):                                    "exec_artifact.yml",
-		filepath.Join(workflows, "b.yml"):                                    "exec_artifact.yml",
-		filepath.Join(workflows, "reusable_caller.yml"):                      "reusable_caller.yml",
-		filepath.Join(workdir, ".gitea", "workflows", "reusable_callee.yml"): "reusable_callee.yml",
-	} {
-		content, err := os.ReadFile(filepath.Join("testdata", "workflows", src))
-		if err != nil {
-			t.Fatal(err)
-		}
-		if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil {
-			t.Fatal(err)
-		}
-		if err := os.WriteFile(dst, content, 0o644); err != nil {
+	workflow, err := os.ReadFile("testdata/workflows/exec_artifact.yml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	workflows := t.TempDir()
+	for _, name := range []string{"a.yml", "b.yml"} {
+		if err := os.WriteFile(filepath.Join(workflows, name), workflow, 0o644); err != nil {
 			t.Fatal(err)
 		}
 	}
 
+	workdir := t.TempDir()
 	for _, args := range [][]string{
 		{"init", "-q"},
 		{"remote", "add", "origin", "https://example.com/e2e/exec.git"},
@@ -304,19 +309,12 @@ func testExec(t *testing.T) {
 	}
 
 	artifacts := t.TempDir()
-	args := []string{"exec", "-C", workdir, "-W", workflows, "-i", os.Getenv("E2E_JOB_IMAGE"), "--artifact-server-path", artifacts, "--no-skip-checkout"}
-	if fixture.network != "" {
-		args = append(args, "--network", fixture.network)
-	}
+	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_") })
-	out, err := cmd.CombinedOutput()
-	if err != nil {
+	if out, err := cmd.CombinedOutput(); err != nil {
 		t.Fatalf("exec: %v\n%s", err, out)
 	}
-	if !strings.Contains(string(out), "e2e-reusable-ran") {
-		t.Errorf("the called workflow did not run:\n%s", out)
-	}
 	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

@@ -28,19 +28,20 @@ func TestCompatibility(t *testing.T) {
 	t.Run("payloads", testPayloads)
 	t.Run("cache_and_artifacts", testCacheAndArtifactRoundTrips)
 	t.Run("cancellation_and_log_streaming", testRunCancellation)
-	t.Run("dispatch", testWorkflowDispatch)
+	t.Run("dispatch", inAllModes(testWorkflowDispatch))
 	t.Run("builtin_checkout", testBuiltinCheckout)
-	t.Run("exec", testExec)
+	t.Run("exec_artifacts", testExecArtifacts)
 	t.Run("moving_action_ref", testMovingActionRef)
 	t.Run("ephemeral", testEphemeralRunner)
-	t.Run("failure_paths", testFailurePaths)
+	t.Run("failure_paths", inAllModes(testFailurePaths))
+	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)
-	t.Run("reusable_workflow", testReusableWorkflow)
+	t.Run("reusable_workflow", inAllModes(testReusableWorkflow))
 	t.Run("label_routing", testLabelRouting)
 	t.Run("capacity", testCapacity)
 	t.Run("reconnect", testReconnect)

e2e/dispatch_test.go

@@ -11,27 +11,11 @@ import (
 	"time"
 )
 
-func testWorkflowDispatch(t *testing.T) {
-	t.Parallel()
-	ctx := t.Context()
-
-	api, repo := newScenario(t)
-	pushWorkflow(t, api, repo, "dispatch.yml")
-
-	branch, err := api.DefaultBranch(ctx, repo)
-	if err != nil {
-		t.Fatalf("get default branch: %v", err)
+func testWorkflowDispatch(t *testing.T, m mode) {
+	result := runWorkflow(t, m, workflowSpec{files: []string{"dispatch.yml"}, inputs: map[string]string{"subject": "dispatch-input-value"}})
+	if result.conclusion != "success" {
+		t.Errorf("run concluded %q, want success", result.conclusion)
 	}
-	inputs := map[string]string{"subject": "dispatch-input-value"}
-	if err := waitForDispatch(ctx, api, repo, "dispatch.yml", branch, inputs); err != nil {
-		t.Fatalf("dispatch workflow: %v", err)
-	}
-
-	dispatched := waitForRun(t, api, repo)
-	if dispatched.Event != "workflow_dispatch" {
-		t.Fatalf("run event is %q, want workflow_dispatch", dispatched.Event)
-	}
-	requireSuccess(t, api, repo, dispatched.ID)
 }
 
 // Gitea indexes workflows asynchronously after the Contents API commit.

e2e/failure_test.go

@@ -5,55 +5,41 @@
 
 package e2e
 
-import "testing"
+import (
+	"testing"
+	"time"
+)
 
-func testFailurePaths(t *testing.T) {
+func testFailurePaths(t *testing.T, m mode) {
+	result := runWorkflow(t, m, workflowSpec{files: []string{"failure.yml"}})
+	if result.conclusion != "failure" {
+		t.Errorf("run concluded %q, want failure", result.conclusion)
+	}
+	for job, want := range map[string]string{"fail": "failure", "tolerated": "success", "dependent": "skipped"} {
+		if got := result.job(job); got != want {
+			t.Errorf("job %q concluded %q, want %q", job, got, want)
+		}
+	}
+	for _, marker := range []string{"e2e-before-fail", "e2e-failure-cleanup", "e2e-tolerated-next"} {
+		if !result.hasLine(marker) {
+			t.Errorf("logs are missing %q", marker)
+		}
+	}
+	if result.hasLine("e2e-skipped-step") {
+		t.Error("a success() step ran after a failed step")
+	}
+}
+
+func testJobTimeout(t *testing.T) {
 	t.Parallel()
-	ctx := t.Context()
-
-	api, repo := newScenario(t)
-	pushWorkflow(t, api, repo, "failure.yml")
-
+	api, repo, _ := startIsolatedScenario(t, "timeout.yml", "e2e-timeout", runnerOptions{timeout: 15 * time.Second})
 	wfRun := waitForRun(t, api, repo)
-	completed, err := api.WaitForRunConclusion(ctx, repo, wfRun.ID, runTimeout)
+	completed, err := api.WaitForRunConclusion(t.Context(), repo, wfRun.ID, runTimeout)
 	if err != nil {
 		dumpRunLogs(t, api, repo, wfRun.ID)
 		t.Fatalf("wait for run: %v", err)
 	}
-	if completed.Conclusion != "failure" {
-		t.Errorf("run concluded %q, want failure", completed.Conclusion)
-	}
-
-	jobs, err := api.Jobs(ctx, repo, wfRun.ID)
-	if err != nil {
-		t.Fatalf("list jobs: %v", err)
-	}
-	want := map[string]string{
-		"fail":      "failure",
-		"tolerated": "success",
-		"dependent": "skipped",
-		"timeout":   "failure",
-	}
-	for _, job := range jobs {
-		if w, ok := want[job.Name]; ok && job.Conclusion != w {
-			t.Errorf("job %q concluded %q, want %q", job.Name, job.Conclusion, w)
-		}
-		delete(want, job.Name)
-	}
-	for name := range want {
-		t.Errorf("job %q not found", name)
-	}
-
-	logs := runLogs(t, api, repo, wfRun.ID)
-	for _, marker := range []string{"e2e-before-fail", "e2e-failure-cleanup", "e2e-tolerated-next", "e2e-timeout-started"} {
-		if commandRow(logs, marker) != marker {
-			t.Errorf("logs are missing %q", marker)
-		}
-	}
-	if commandRow(logs, "e2e-skipped-step") != "" {
-		t.Error("a success() step ran after a failed step")
-	}
-	if t.Failed() {
-		t.Logf("full stored logs:\n%s", logs)
+	if logs := runLogs(t, api, repo, wfRun.ID); completed.Conclusion != "failure" || commandRow(logs, "e2e-timeout-started") == "" {
+		t.Errorf("run concluded %q, want failure after the job started, logs:\n%s", completed.Conclusion, logs)
 	}
 }

e2e/gitea_api.go

@@ -53,12 +53,16 @@ func (a *GiteaAPI) CreateRepo(ctx context.Context, name string) error {
 }
 
 func (a *GiteaAPI) CreateFile(ctx context.Context, repo, path, content, message string) error {
-	body := map[string]any{
-		"content": base64.StdEncoding.EncodeToString([]byte(content)),
-		"message": message,
+	return a.CreateFiles(ctx, repo, map[string]string{path: content}, message)
+}
+
+func (a *GiteaAPI) CreateFiles(ctx context.Context, repo string, files map[string]string, message string) error {
+	var operations []map[string]string
+	for path, content := range files {
+		operations = append(operations, map[string]string{"operation": "create", "path": path, "content": base64.StdEncoding.EncodeToString([]byte(content))})
 	}
-	url := fmt.Sprintf("/api/v1/repos/%s/%s/contents/%s", giteaAdminUser, repo, path)
-	return a.doJSON(ctx, http.MethodPost, url, body, nil)
+	body := map[string]any{"files": operations, "message": message}
+	return a.doJSON(ctx, http.MethodPost, fmt.Sprintf("/api/v1/repos/%s/%s/contents", giteaAdminUser, repo), body, nil)
 }
 
 func (a *GiteaAPI) CreateTag(ctx context.Context, repo, name string) error {

e2e/gitea_fixture.go

@@ -15,9 +15,7 @@ import (
 	"io"
 	"net"
 	"net/http"
-	"net/netip"
 	"os"
-	"strconv"
 	"strings"
 	"time"
 
@@ -40,7 +38,8 @@ type GiteaFixture struct {
 	image      string
 	version    string
 	baseURL    string
-	network    string // set in container mode; jobs must join it
+	network    string // the user-defined network every container of the suite joins
+	selfIP     string // the test process's address on network, which job containers can reach
 	adminToken string
 }
 
@@ -55,44 +54,10 @@ func dockerClient(ctx context.Context) (mobyclient.APIClient, error) {
 	return cli, nil
 }
 
-// Prefer docker bridge gateway, then LAN IP; never 127.0.0.1 (ACTIONS_RUNTIME_URL must reach the host from job containers).
-func hostAddress(ctx context.Context, cli mobyclient.APIClient) netip.Addr {
-	if gateway, ok := bridgeGateway(ctx, cli); ok && bindable(gateway) {
-		return gateway
-	}
-
-	loopback := netip.AddrFrom4([4]byte{127, 0, 0, 1})
-
-	conn, err := net.Dial("udp", "192.0.2.1:80") // TEST-NET-1: routing table only
-	if err != nil {
-		return loopback
-	}
-	defer conn.Close()
-
-	addr, ok := conn.LocalAddr().(*net.UDPAddr)
-	if !ok {
-		return loopback
-	}
-	return addr.AddrPort().Addr().Unmap()
-}
-
-func bridgeGateway(ctx context.Context, cli mobyclient.APIClient) (netip.Addr, bool) {
-	inspected, err := cli.NetworkInspect(ctx, "bridge", mobyclient.NetworkInspectOptions{})
-	if err != nil {
-		return netip.Addr{}, false
-	}
-	for _, cfg := range inspected.Network.IPAM.Config {
-		if cfg.Gateway.Is4() {
-			return cfg.Gateway, true
-		}
-	}
-	return netip.Addr{}, false
-}
-
 // When tests run inside a container (sibling docker daemon), join that network and address by name.
-func selfNetwork(ctx context.Context, cli mobyclient.APIClient) (string, bool) {
+func selfNetwork(ctx context.Context, cli mobyclient.APIClient) (string, string, bool) {
 	if _, err := os.Stat("/.dockerenv"); err != nil {
-		return "", false
+		return "", "", false
 	}
 
 	id, err := os.ReadFile("/proc/sys/kernel/hostname")
@@ -100,42 +65,19 @@ func selfNetwork(ctx context.Context, cli mobyclient.APIClient) (string, bool) {
 		id, err = os.ReadFile("/etc/hostname")
 	}
 	if err != nil {
-		return "", false
+		return "", "", false
 	}
 
 	inspected, err := cli.ContainerInspect(ctx, strings.TrimSpace(string(id)), mobyclient.ContainerInspectOptions{})
 	if err != nil || inspected.Container.NetworkSettings == nil {
-		return "", false
+		return "", "", false
 	}
-	for name := range inspected.Container.NetworkSettings.Networks {
+	for name, endpoint := range inspected.Container.NetworkSettings.Networks {
 		if name != "host" && name != "none" && name != "bridge" { // only user-defined nets have DNS
-			return name, true
+			return name, endpoint.IPAddress.String(), true
 		}
 	}
-	return "", false
-}
-
-func bindable(addr netip.Addr) bool {
-	l, err := net.Listen("tcp", net.JoinHostPort(addr.String(), "0"))
-	if err != nil {
-		return false
-	}
-	_ = l.Close()
-	return true
-}
-
-func freeHostPort(host string) (int, error) {
-	l, err := net.Listen("tcp", net.JoinHostPort(host, "0"))
-	if err != nil {
-		return 0, err
-	}
-	defer l.Close()
-
-	addr, ok := l.Addr().(*net.TCPAddr)
-	if !ok {
-		return 0, fmt.Errorf("unexpected listener address type %T", l.Addr())
-	}
-	return addr.Port, nil
+	return "", "", false
 }
 
 func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, error) {
@@ -151,31 +93,12 @@ func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, e
 		return nil, errors.New("E2E_GITEA_IMAGE is not set")
 	}
 
-	name := fmt.Sprintf("gitea-runner-e2e-%d", time.Now().UnixNano())
-	containerPort := network.MustParsePort("3000/tcp")
-
-	var (
-		baseURL      string
-		hostConfig   = &apicontainer.HostConfig{}
-		netConfig    *network.NetworkingConfig
-		sharedNet, _ = selfNetwork(ctx, cli)
-	)
-	if sharedNet != "" {
-		baseURL = "http://" + net.JoinHostPort(name, "3000")
-		netConfig = &network.NetworkingConfig{
-			EndpointsConfig: map[string]*network.EndpointSettings{sharedNet: {}},
-		}
-	} else {
-		host := hostAddress(ctx, cli)
-		port, err := freeHostPort(host.String())
-		if err != nil {
-			return nil, fmt.Errorf("find a free host port: %w", err)
-		}
-		baseURL = "http://" + net.JoinHostPort(host.String(), strconv.Itoa(port))
-		hostConfig.PortBindings = network.PortMap{
-			containerPort: []network.PortBinding{{HostIP: host, HostPort: strconv.Itoa(port)}},
-		}
+	sharedNet, selfIP, ok := selfNetwork(ctx, cli)
+	if !ok {
+		return nil, errors.New("e2e tests must run in a container on a user-defined network, `make test-e2e` provides one")
 	}
+	name := fmt.Sprintf("gitea-runner-e2e-%d", time.Now().UnixNano())
+	baseURL := "http://" + net.JoinHostPort(name, "3000")
 
 	if _, err := cli.ImageInspect(ctx, image); err != nil {
 		return nil, fmt.Errorf("inspect gitea image %s: %w", image, err)
@@ -191,17 +114,16 @@ func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, e
 				"GITEA__server__LFS_START_SERVER=true",
 				"GITEA__server__ROOT_URL=" + baseURL + "/",
 			},
-			ExposedPorts: network.PortSet{containerPort: struct{}{}},
 		},
-		HostConfig:       hostConfig,
-		NetworkingConfig: netConfig,
+		HostConfig:       &apicontainer.HostConfig{Tmpfs: map[string]string{"/data": "exec"}},
+		NetworkingConfig: &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{sharedNet: {}}},
 		Name:             name,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("create gitea container: %w", err)
 	}
 
-	f := &GiteaFixture{cli: cli, id: resp.ID, image: image, baseURL: baseURL, network: sharedNet}
+	f := &GiteaFixture{cli: cli, id: resp.ID, image: image, baseURL: baseURL, network: sharedNet, selfIP: selfIP}
 	closeClient = false
 
 	if _, err := cli.ContainerStart(ctx, f.id, mobyclient.ContainerStartOptions{}); err != nil {

e2e/kubernetes_test.go

@@ -9,15 +9,16 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
-	"net"
-	"net/netip"
 	"net/url"
 	"os"
 	"path/filepath"
-	"strconv"
+	"slices"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -32,125 +33,168 @@ import (
 func testKubernetes(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
-	k3sID, kubernetes, images := startK3s(t)
+	cluster := requireCluster(t)
 
 	api, repo := newScenario(t)
-	if err := api.CreateVariable(ctx, repo, "E2E_SERVICE_IMAGE", images[1]); err != nil {
+	if err := api.CreateVariable(ctx, repo, "E2E_SERVICE_IMAGE", cluster.serviceImage); err != nil {
 		t.Fatalf("create service image variable: %v", err)
 	}
-	startRunner(t, repo, "e2e-kubernetes", runnerOptions{image: images[0], kubernetes: kubernetes})
+	startRunner(t, repo, "e2e-kubernetes", runnerOptions{image: cluster.jobImage, kubernetes: cluster.kubernetes})
 	pushWorkflow(t, api, repo, "kubernetes.yml")
 
 	wfRun := waitForRun(t, api, repo)
 	requireSuccess(t, api, repo, wfRun.ID)
+}
 
-	if out, err := containerExec(ctx, fixture.cli, k3sID, "", nil,
+// checkClusterLeaks runs after every test, so a pod or secret any of them left behind fails the suite.
+func checkClusterLeaks() error {
+	if !clusterStarted.Load() {
+		return nil
+	}
+	cluster, err := k3s()
+	if err != nil {
+		return nil // the tests that needed the cluster already failed
+	}
+	if out, err := containerExec(context.Background(), dockerCLI, cluster.id, "", nil,
 		"kubectl", "wait", "--for=delete", "pods,secrets", "--selector=app.kubernetes.io/managed-by=gitea-runner", "--timeout=1m",
 	); err != nil {
-		t.Errorf("job pods or secrets left behind: %v: %s", err, out)
+		return fmt.Errorf("job pods or secrets left behind: %w: %s", err, out)
+	}
+	return nil
+}
+
+type kubeCluster struct {
+	id                     string
+	kubernetes             *config.Kubernetes
+	jobImage, serviceImage string
+}
+
+var (
+	k3s            = sync.OnceValues(startK3s) // independent of Gitea, so it can boot alongside it
+	clusterStarted atomic.Bool
+	clusterCleanup []func()
+	sharedCluster  = sync.OnceValues(func() (*kubeCluster, error) {
+		cluster, err := k3s()
+		if err != nil {
+			return nil, err
+		}
+		gitea, err := fixture.cli.ContainerInspect(context.Background(), fixture.id, mobyclient.ContainerInspectOptions{})
+		if err != nil {
+			return nil, fmt.Errorf("inspect gitea container: %w", err)
+		}
+		giteaURL, err := url.Parse(fixture.baseURL)
+		if err != nil {
+			return nil, fmt.Errorf("parse gitea url: %w", err)
+		}
+		cluster.kubernetes.PodTemplate = map[string]any{"spec": map[string]any{"hostAliases": []any{map[string]any{
+			"ip":        gitea.Container.NetworkSettings.Networks[fixture.network].IPAddress.String(),
+			"hostnames": []any{giteaURL.Hostname()},
+		}}}}
+		return cluster, nil
+	})
+)
+
+func requireCluster(t *testing.T) *kubeCluster {
+	t.Helper()
+	cluster, err := sharedCluster()
+	if err != nil {
+		t.Fatal(err)
+	}
+	return cluster
+}
+
+func stopCluster() {
+	if !clusterStarted.Load() {
+		return
+	}
+	_, _ = k3s()
+	for _, cleanup := range slices.Backward(clusterCleanup) {
+		cleanup()
 	}
 }
 
-func startK3s(t *testing.T) (string, *config.Kubernetes, []string) {
-	t.Helper()
-	ctx := t.Context()
-	cli := fixture.cli
+func startK3s() (*kubeCluster, error) {
+	clusterStarted.Store(true)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+	cli := dockerCLI
+	sharedNet, _, ok := selfNetwork(ctx, cli)
+	if !ok {
+		return nil, errors.New("e2e tests must run in a container on a user-defined network")
+	}
 
 	name := fmt.Sprintf("gitea-runner-e2e-k3s-%d", time.Now().UnixNano())
-	apiPort := network.MustParsePort("6443/tcp")
 	server := "https://" + name + ":6443"
-	hostConfig := &apicontainer.HostConfig{Privileged: true}
-	var netConfig *network.NetworkingConfig
-	if fixture.network != "" {
-		netConfig = &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{fixture.network: {}}}
-	} else {
-		loopback := netip.MustParseAddr("127.0.0.1")
-		port, err := freeHostPort(loopback.String())
-		if err != nil {
-			t.Fatalf("find a free host port: %v", err)
-		}
-		server = "https://" + net.JoinHostPort(loopback.String(), strconv.Itoa(port))
-		hostConfig.PortBindings = network.PortMap{apiPort: []network.PortBinding{{HostIP: loopback, HostPort: strconv.Itoa(port)}}}
-	}
 
 	created, err := cli.ContainerCreate(ctx, mobyclient.ContainerCreateOptions{
 		Config: &apicontainer.Config{
-			Image:        os.Getenv("E2E_K3S_IMAGE"),
-			Cmd:          []string{"server", "--disable=traefik,metrics-server,servicelb,local-storage,coredns", "--tls-san=" + name, "--kubelet-arg=fail-cgroupv1=false"},
-			ExposedPorts: network.PortSet{apiPort: struct{}{}},
+			Image: os.Getenv("E2E_K3S_IMAGE"),
+			Cmd:   []string{"server", "--disable=traefik,metrics-server,servicelb,local-storage,coredns", "--tls-san=" + name, "--kubelet-arg=fail-cgroupv1=false"},
 		},
-		HostConfig:       hostConfig,
-		NetworkingConfig: netConfig,
+		HostConfig:       &apicontainer.HostConfig{Privileged: true},
+		NetworkingConfig: &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{sharedNet: {}}},
 		Name:             name,
 	})
 	if err != nil {
-		t.Fatalf("create k3s container: %v", err)
+		return nil, fmt.Errorf("create k3s container: %w", err)
 	}
-	t.Cleanup(func() {
+	clusterCleanup = append(clusterCleanup, func() {
 		_, _ = cli.ContainerRemove(context.Background(), created.ID, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true})
 	})
 	if _, err := cli.ContainerStart(ctx, created.ID, mobyclient.ContainerStartOptions{}); err != nil {
-		t.Fatalf("start k3s container: %v", err)
+		return nil, fmt.Errorf("start k3s container: %w", err)
 	}
 
-	containerdCtx, stopWaiting := context.WithTimeout(ctx, 2*time.Minute)
-	defer stopWaiting()
-	if err := waitForContainerLog(containerdCtx, created.ID, "containerd is now running"); err != nil {
-		t.Fatalf("wait for k3s containerd: %v", err)
+	if err := waitForContainerLog(ctx, created.ID, "containerd is now running"); err != nil {
+		return nil, fmt.Errorf("wait for k3s containerd: %w", err)
+	}
+	sandboxPulled := make(chan struct{})
+	go func() { // first pods would otherwise pull the sandbox image themselves
+		defer close(sandboxPulled)
+		pullCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+		defer cancel()
+		_, _ = containerExec(pullCtx, cli, created.ID, "", nil, "sh", "-c",
+			`ctr --namespace=k8s.io images pull "docker.io/$(sed -n 's/^ *sandbox = "\(.*\)"$/\1/p' /var/lib/rancher/k3s/agent/etc/containerd/config.toml)"`)
+	}()
+	images, err := preloadImages(ctx, created.ID, os.Getenv("E2E_JOB_IMAGE"), os.Getenv("SERVICE_IMAGE"))
+	<-sandboxPulled
+	if err != nil {
+		return nil, err
 	}
-	images := preloadImages(t, created.ID, os.Getenv("E2E_JOB_IMAGE"), os.Getenv("SERVICE_IMAGE"))
 
-	waitCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
-	defer cancel()
 	for {
-		_, err := containerExec(waitCtx, cli, created.ID, "", nil, "kubectl", "get", "serviceaccount", "default")
+		_, err := containerExec(ctx, cli, created.ID, "", nil, "kubectl", "get", "serviceaccount", "default")
 		if err == nil {
 			break
 		}
 		inspected, inspectErr := cli.ContainerInspect(ctx, created.ID, mobyclient.ContainerInspectOptions{})
-		if inspectErr != nil || !inspected.Container.State.Running || waitCtx.Err() != nil {
+		if inspectErr != nil || !inspected.Container.State.Running || ctx.Err() != nil {
 			var logs bytes.Buffer
-			if reader, logsErr := cli.ContainerLogs(ctx, created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil {
+			if reader, logsErr := cli.ContainerLogs(context.Background(), created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil {
 				_, _ = stdcopy.StdCopy(&logs, &logs, reader)
 				_ = reader.Close()
 			}
-			t.Fatalf("k3s did not start: %v, state %+v %v, last log lines:\n%s", err, inspected.Container.State, inspectErr, logs.String())
+			return nil, fmt.Errorf("k3s did not start: %w, state %+v %v, last log lines:\n%s", err, inspected.Container.State, inspectErr, logs.String())
 		}
 		select {
-		case <-waitCtx.Done():
+		case <-ctx.Done():
 		case <-time.After(pollInterval):
 		}
 	}
 
 	kubeconfig, err := containerExec(ctx, cli, created.ID, "", nil, "cat", "/etc/rancher/k3s/k3s.yaml")
 	if err != nil {
-		t.Fatalf("read kubeconfig: %v", err)
+		return nil, fmt.Errorf("read kubeconfig: %w", err)
 	}
-	path := filepath.Join(t.TempDir(), "kubeconfig")
+	path := filepath.Join(binDir, "kubeconfig")
 	if err := os.WriteFile(path, []byte(strings.ReplaceAll(kubeconfig, "https://127.0.0.1:6443", server)), 0o600); err != nil {
-		t.Fatalf("write kubeconfig: %v", err)
+		return nil, fmt.Errorf("write kubeconfig: %w", err)
 	}
-	kubernetes := &config.Kubernetes{Kubeconfig: path, Namespace: "default"}
-
-	if fixture.network != "" {
-		gitea, err := cli.ContainerInspect(ctx, fixture.id, mobyclient.ContainerInspectOptions{})
-		if err != nil {
-			t.Fatalf("inspect gitea container: %v", err)
-		}
-		giteaURL, err := url.Parse(fixture.baseURL)
-		if err != nil {
-			t.Fatalf("parse gitea url: %v", err)
-		}
-		kubernetes.PodTemplate = map[string]any{"spec": map[string]any{"hostAliases": []any{map[string]any{
-			"ip":        gitea.Container.NetworkSettings.Networks[fixture.network].IPAddress.String(),
-			"hostnames": []any{giteaURL.Hostname()},
-		}}}}
-	}
-	return created.ID, kubernetes, images
+	return &kubeCluster{id: created.ID, kubernetes: &config.Kubernetes{Kubeconfig: path, Namespace: "default"}, jobImage: images[0], serviceImage: images[1]}, nil
 }
 
 func waitForContainerLog(ctx context.Context, id, message string) error {
-	logs, err := fixture.cli.ContainerLogs(ctx, id, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true})
+	logs, err := dockerCLI.ContainerLogs(ctx, id, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true})
 	if err != nil {
 		return err
 	}
@@ -175,25 +219,23 @@ func waitForContainerLog(ctx context.Context, id, message string) error {
 	return fmt.Errorf("container exited before logging %q", message)
 }
 
-func preloadImages(t *testing.T, k3sID string, images ...string) []string {
-	t.Helper()
-	ctx := t.Context()
+func preloadImages(ctx context.Context, k3sID string, images ...string) ([]string, error) {
 	var aliases []string
 	for index, image := range images {
 		alias := fmt.Sprintf("gitea-runner-e2e/preload-%d:%d", index, time.Now().UnixNano())
-		if _, err := fixture.cli.ImageTag(ctx, mobyclient.ImageTagOptions{Source: image, Target: alias}); err != nil {
-			t.Fatalf("tag %s: %v", image, err)
+		if _, err := dockerCLI.ImageTag(ctx, mobyclient.ImageTagOptions{Source: image, Target: alias}); err != nil {
+			return nil, fmt.Errorf("tag %s: %w", image, err)
 		}
-		t.Cleanup(func() { _, _ = fixture.cli.ImageRemove(context.Background(), alias, mobyclient.ImageRemoveOptions{}) })
+		clusterCleanup = append(clusterCleanup, func() { _, _ = dockerCLI.ImageRemove(context.Background(), alias, mobyclient.ImageRemoveOptions{}) })
 		aliases = append(aliases, alias)
 	}
-	archive, err := fixture.cli.ImageSave(ctx, aliases)
+	archive, err := dockerCLI.ImageSave(ctx, aliases)
 	if err != nil {
-		t.Fatalf("save %v: %v", images, err)
+		return nil, fmt.Errorf("save %v: %w", images, err)
 	}
 	defer archive.Close()
-	if out, err := containerExec(ctx, fixture.cli, k3sID, "", archive, "ctr", "--namespace=k8s.io", "images", "import", "-"); err != nil {
-		t.Fatalf("import %v into k3s: %v: %s", images, err, out)
+	if out, err := containerExec(ctx, dockerCLI, k3sID, "", archive, "ctr", "--namespace=k8s.io", "images", "import", "-"); err != nil {
+		return nil, fmt.Errorf("import %v into k3s: %w: %s", images, err, out)
 	}
-	return aliases
+	return aliases, nil
 }

e2e/main_test.go

@@ -9,18 +9,24 @@ import (
 	"cmp"
 	"context"
 	"errors"
+	"flag"
 	"fmt"
+	"net"
 	"net/http"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"regexp"
+	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"gitea.com/gitea/runner/internal/app/poll"
+
+	mobyclient "github.com/moby/moby/client"
 )
 
 const runTimeout = 3 * time.Minute
@@ -33,6 +39,8 @@ var skipReason string
 
 var binDir string
 
+var dockerCLI mobyclient.APIClient
+
 var runnerBinary = sync.OnceValues(func() (string, error) {
 	bin := filepath.Join(binDir, "gitea-runner")
 	if out, err := exec.Command(cmp.Or(os.Getenv("GO"), "go"), "build", "-o", bin, "..").CombinedOutput(); err != nil {
@@ -54,6 +62,17 @@ func runSuite(m *testing.M) int {
 		skipReason = fmt.Sprintf("docker unavailable: %v", err)
 		return m.Run()
 	}
+	dockerCLI = cli
+
+	if binDir, err = os.MkdirTemp("", "gitea-runner-e2e-"); err != nil {
+		fmt.Fprintf(os.Stderr, "create binary dir: %v\n", err)
+		return 1
+	}
+	defer os.RemoveAll(binDir)
+	defer stopCluster()
+	if flag.Parse(); flag.Lookup("test.run").Value.String() == "" { // a filtered run starts the cluster only if it needs it
+		go func() { _, _ = k3s() }()
+	}
 
 	f, err := StartGitea(ctx, cli)
 	if err != nil {
@@ -63,15 +82,14 @@ func runSuite(m *testing.M) int {
 	fixture = f
 	defer func() { _ = fixture.Close(context.Background()) }()
 
-	if binDir, err = os.MkdirTemp("", "gitea-runner-e2e-"); err != nil {
-		fmt.Fprintf(os.Stderr, "create binary dir: %v\n", err)
-		return 1
-	}
-	defer os.RemoveAll(binDir)
-
 	fmt.Fprintf(os.Stderr, "gitea fixture: image=%s version=%s\n", fixture.image, fixture.version)
 
-	return m.Run()
+	code := m.Run()
+	if err := checkClusterLeaks(); err != nil && code == 0 {
+		fmt.Fprintln(os.Stderr, err)
+		code = 1
+	}
+	return code
 }
 
 func newScenario(t *testing.T) (*GiteaAPI, string) {
@@ -162,6 +180,24 @@ func dumpRunLogs(t *testing.T, api *GiteaAPI, repo string, runID int64) {
 	}
 }
 
+// newGate serves a URL that answers 503 until release, for jobs to poll instead of sleeping.
+func newGate(t *testing.T) (string, func()) {
+	t.Helper()
+	var released atomic.Bool
+	listener, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatalf("listen for the gate: %v", err)
+	}
+	server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		if !released.Load() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+		}
+	}), ReadHeaderTimeout: time.Second}
+	go func() { _ = server.Serve(listener) }()
+	t.Cleanup(func() { _ = server.Close() })
+	return fmt.Sprintf("http://%s/", net.JoinHostPort(fixture.selfIP, strconv.Itoa(listener.Addr().(*net.TCPAddr).Port))), func() { released.Store(true) }
+}
+
 func waitForRun(t *testing.T, api *GiteaAPI, repo string) *ActionRun {
 	t.Helper()
 	return waitForRunAfter(t, api, repo, 0)

e2e/modes_test.go

@@ -0,0 +1,227 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+//go:build e2e
+
+package e2e
+
+import (
+	"bufio"
+	"bytes"
+	"cmp"
+	"encoding/json"
+	"errors"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"slices"
+	"strings"
+	"testing"
+)
+
+type mode struct {
+	exec, kubernetes bool
+}
+
+var allModes = []mode{{}, {kubernetes: true}, {exec: true}, {exec: true, kubernetes: true}}
+
+func (m mode) String() string {
+	driver, backend := "daemon", "docker"
+	if m.exec {
+		driver = "exec"
+	}
+	if m.kubernetes {
+		backend = "kubernetes"
+	}
+	return driver + "_" + backend
+}
+
+func inAllModes(scenario func(*testing.T, mode)) func(*testing.T) {
+	return func(t *testing.T) {
+		t.Parallel()
+		for _, m := range allModes {
+			t.Run(m.String(), func(t *testing.T) {
+				t.Parallel()
+				scenario(t, m)
+			})
+		}
+	}
+}
+
+type workflowSpec struct {
+	files    []string
+	inputs   map[string]string // dispatches files[0] instead of relying on push
+	execArgs []string
+}
+
+type workflowResult struct {
+	conclusion string
+	jobs       map[string]string
+	lines      []string
+}
+
+func (r *workflowResult) job(id string) string {
+	return cmp.Or(r.jobs[id], "skipped") // exec reports nothing for a job it skipped
+}
+
+func (r *workflowResult) hasLine(line string) bool {
+	return slices.Contains(r.lines, line)
+}
+
+func runWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
+	t.Helper()
+	run := daemonWorkflow
+	if m.exec {
+		run = execWorkflow
+	}
+	result := run(t, m, spec)
+	t.Cleanup(func() {
+		if t.Failed() {
+			t.Logf("%s concluded %q, jobs %v, logs:\n%s", m, result.conclusion, result.jobs, strings.Join(result.lines, "\n"))
+		}
+	})
+	return result
+}
+
+func daemonWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
+	t.Helper()
+	ctx := t.Context()
+
+	api, repo := newScenario(t)
+	if m.kubernetes {
+		cluster := requireCluster(t)
+		startRunner(t, repo, "e2e-kubernetes", runnerOptions{image: cluster.jobImage, kubernetes: cluster.kubernetes, capacity: 16})
+	}
+	files := map[string]string{}
+	for _, file := range spec.files {
+		content, err := os.ReadFile(filepath.Join("testdata", "workflows", file))
+		if err != nil {
+			t.Fatalf("read workflow fixture %s: %v", file, err)
+		}
+		if m.kubernetes {
+			content = bytes.ReplaceAll(content, []byte("runs-on: ubuntu-latest"), []byte("runs-on: e2e-kubernetes"))
+		}
+		files[".gitea/workflows/"+file] = string(content)
+	}
+	if err := api.CreateFiles(ctx, repo, files, "add workflows"); err != nil {
+		t.Fatalf("push workflows: %v", err)
+	}
+	if spec.inputs != nil {
+		branch, err := api.DefaultBranch(ctx, repo)
+		if err != nil {
+			t.Fatalf("get default branch: %v", err)
+		}
+		if err := waitForDispatch(ctx, api, repo, spec.files[0], branch, spec.inputs); err != nil {
+			t.Fatalf("dispatch workflow: %v", err)
+		}
+	}
+
+	wfRun := waitForRun(t, api, repo)
+	completed, err := api.WaitForRunConclusion(ctx, repo, wfRun.ID, runTimeout)
+	if err != nil {
+		dumpRunLogs(t, api, repo, wfRun.ID)
+		t.Fatalf("wait for run: %v", err)
+	}
+	jobs, err := api.Jobs(ctx, repo, wfRun.ID)
+	if err != nil {
+		t.Fatalf("list jobs: %v", err)
+	}
+	result := &workflowResult{conclusion: completed.Conclusion, jobs: map[string]string{}}
+	for _, job := range jobs {
+		result.jobs[job.Name] = job.Conclusion
+	}
+	for line := range strings.SplitSeq(runLogs(t, api, repo, wfRun.ID), "\n") {
+		if _, payload, found := strings.Cut(line, "Z "); found {
+			result.lines = append(result.lines, strings.TrimRight(payload, "\r"))
+		}
+	}
+	return result
+}
+
+func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult {
+	t.Helper()
+	ctx := t.Context()
+
+	bin, err := runnerBinary()
+	if err != nil {
+		t.Fatal(err)
+	}
+	workdir := t.TempDir()
+	workflows := filepath.Join(workdir, ".gitea", "workflows")
+	if err := os.MkdirAll(workflows, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	for _, file := range spec.files {
+		content, err := os.ReadFile(filepath.Join("testdata", "workflows", file))
+		if err != nil {
+			t.Fatalf("read workflow fixture %s: %v", file, err)
+		}
+		if err := os.WriteFile(filepath.Join(workflows, file), content, 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for _, args := range [][]string{
+		{"init", "-q"},
+		{"remote", "add", "origin", "https://example.com/e2e/exec.git"},
+		{"-c", "user.name=e2e", "-c", "[email protected]", "commit", "-q", "--allow-empty", "-m", "init"},
+	} {
+		if out, err := exec.CommandContext(ctx, "git", append([]string{"-C", workdir}, args...)...).CombinedOutput(); err != nil {
+			t.Fatalf("git %v: %v\n%s", args, err, out)
+		}
+	}
+
+	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_") })
+	if m.kubernetes {
+		cluster := requireCluster(t)
+		args = append(args, "-i", "kubernetes://"+cluster.jobImage)
+		env = append(env, "KUBECONFIG="+cluster.kubernetes.Kubeconfig)
+	} else {
+		args = append(args, "-i", os.Getenv("E2E_JOB_IMAGE"), "--network", fixture.network)
+	}
+	if spec.inputs != nil {
+		args = append(args, "-E", "workflow_dispatch")
+		for name, value := range spec.inputs {
+			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()
+	var exitErr *exec.ExitError
+	if runErr != nil && !errors.As(runErr, &exitErr) {
+		t.Fatalf("exec: %v\n%s", runErr, stderr.String())
+	}
+
+	result := &workflowResult{conclusion: "success", jobs: map[string]string{}}
+	if runErr != nil {
+		result.conclusion = "failure"
+	}
+	scanner := bufio.NewScanner(&stdout)
+	scanner.Buffer(nil, 1<<20)
+	for scanner.Scan() {
+		var entry struct {
+			JobID     string `json:"jobID"`
+			JobResult string `json:"jobResult"`
+			Msg       string `json:"msg"`
+			RawOutput bool   `json:"raw_output"`
+		}
+		if json.Unmarshal(scanner.Bytes(), &entry) != nil {
+			continue
+		}
+		if entry.JobResult != "" {
+			result.jobs[entry.JobID] = entry.JobResult
+		}
+		if entry.RawOutput {
+			for line := range strings.SplitSeq(strings.TrimRight(entry.Msg, "\n"), "\n") {
+				result.lines = append(result.lines, strings.TrimRight(line, "\r"))
+			}
+		}
+	}
+	if result.conclusion != "success" && len(result.jobs) == 0 {
+		t.Fatalf("exec ran no job: %v\n%s", runErr, stderr.String())
+	}
+	return result
+}

e2e/reconnect_test.go

@@ -10,6 +10,8 @@ import (
 	"io"
 	"net"
 	"net/url"
+	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -101,6 +103,10 @@ func (p *flakyProxy) restore() { p.down.Store(false) }
 func testReconnect(t *testing.T) {
 	t.Parallel()
 	api, repo := newScenario(t)
+	gate, release := newGate(t)
+	if err := api.CreateVariable(t.Context(), repo, "E2E_GATE", gate); err != nil {
+		t.Fatalf("create gate variable: %v", err)
+	}
 	proxy := newFlakyProxy(t, fixture.baseURL)
 	startRunner(t, repo, "e2e-reconnect", runnerOptions{host: true, address: proxy.URL()})
 
@@ -112,14 +118,19 @@ func testReconnect(t *testing.T) {
 
 	pushWorkflow(t, api, repo, "reconnect.yml")
 	slow := waitForRunAfter(t, api, repo, idle.ID)
-	waitForRunningJobLog(t, api, repo, slow.ID, "e2e-before-outage")
-	// the job logs a line per second, so a long outage spans several that can only arrive after it
+	waitForRunningJobLog(t, api, repo, slow.ID, "e2e-before-outage", 1)
+	// the job logs a line every 200ms until released, so the outage spans several that can only arrive after it
 	proxy.cut(t, 8)
 	proxy.restore()
+	release()
 	requireSuccess(t, api, repo, slow.ID)
 	logs := runLogs(t, api, repo, slow.ID)
-	for i := 1; i <= 15; i++ {
-		if tick := fmt.Sprintf("tick-%02d", i); commandRow(logs, tick) != tick {
+	ticks, err := strconv.Atoi(strings.TrimPrefix(commandRow(logs, "ticks="), "ticks="))
+	if err != nil {
+		t.Fatalf("the job did not report its tick count: %v\n%s", err, logs)
+	}
+	for i := 1; i <= ticks; i++ {
+		if tick := fmt.Sprintf("tick-%d", i); commandRow(logs, tick) != tick {
 			t.Errorf("log line %s written during or after the outage was lost:\n%s", tick, logs)
 			break
 		}

e2e/reusable_test.go

@@ -8,15 +8,12 @@ package e2e
 import "testing"
 
 // testReusableWorkflow runs a local workflow_call workflow and consumes its output.
-func testReusableWorkflow(t *testing.T) {
-	t.Parallel()
-	api, repo := newScenario(t)
-	pushWorkflow(t, api, repo, "reusable_callee.yml")
-	pushWorkflow(t, api, repo, "reusable_caller.yml")
-
-	wfRun := waitForRun(t, api, repo)
-	requireSuccess(t, api, repo, wfRun.ID)
-	if logs := runLogs(t, api, repo, wfRun.ID); commandRow(logs, "e2e-reusable-ran") == "" {
-		t.Errorf("the called workflow did not run:\n%s", logs)
+func testReusableWorkflow(t *testing.T, m mode) {
+	result := runWorkflow(t, m, workflowSpec{
+		files:    []string{"reusable_callee.yml", "reusable_caller.yml"},
+		execArgs: []string{"--no-skip-checkout"},
+	})
+	if result.conclusion != "success" || !result.hasLine("e2e-reusable-ran") {
+		t.Errorf("run concluded %q, want success with the called workflow's marker", result.conclusion)
 	}
 }

e2e/runner_harness.go

@@ -33,6 +33,7 @@ type runnerOptions struct {
 	host       bool   // register a host-mode label instead of a docker one
 	name       string // suffix that keeps runner names unique when a test starts several
 	address    string // Gitea address the runner talks to, default the fixture
+	timeout    time.Duration
 }
 
 func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *poll.Poller {
@@ -60,9 +61,8 @@ func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *p
 	cfg.Runner.StateReportInterval = 500 * time.Millisecond // so cancel reaches the job quickly
 	cfg.Runner.LogReportInterval = 500 * time.Millisecond
 	cfg.Runner.Capacity = max(options.capacity, 1)
-	if fixture.network != "" {
-		cfg.Container.Network = fixture.network
-	}
+	cfg.Runner.Timeout = cmp.Or(options.timeout, cfg.Runner.Timeout)
+	cfg.Container.Network = fixture.network
 
 	scheme := labels.SchemeDocker
 	if options.kubernetes != nil {

e2e/scheduling_test.go

@@ -75,10 +75,16 @@ func testCapacity(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 			api, repo := newScenario(t)
+			gate, release := newGate(t)
+			if err := api.CreateVariable(t.Context(), repo, "E2E_GATE", gate); err != nil {
+				t.Fatalf("create gate variable: %v", err)
+			}
 			startRunner(t, repo, "e2e-capacity", runnerOptions{capacity: tc.capacity})
 			pushWorkflow(t, api, repo, "capacity.yml")
 
 			wfRun := waitForRun(t, api, repo)
+			waitForRunningJobLog(t, api, repo, wfRun.ID, "span-start=", tc.capacity)
+			release()
 			requireSuccess(t, api, repo, wfRun.ID)
 
 			jobs, err := api.Jobs(t.Context(), repo, wfRun.ID)

e2e/testdata/workflows/capacity.yml

@@ -6,12 +6,12 @@ jobs:
     steps:
       - run: |
           echo "span-start=$(date +%s%N)"
-          sleep 8
+          until curl -fs "${{ vars.E2E_GATE }}" > /dev/null; do sleep 0.1; done
           echo "span-end=$(date +%s%N)"
   two:
     runs-on: e2e-capacity
     steps:
       - run: |
           echo "span-start=$(date +%s%N)"
-          sleep 8
+          until curl -fs "${{ vars.E2E_GATE }}" > /dev/null; do sleep 0.1; done
           echo "span-end=$(date +%s%N)"

e2e/testdata/workflows/cli_shutdown.yml

@@ -6,5 +6,5 @@ jobs:
     steps:
       - run: |
           echo e2e-shutdown-started
-          sleep 8
+          until curl -fs "${{ vars.E2E_GATE }}" > /dev/null; do sleep 0.1; done
           echo e2e-shutdown-finished

e2e/testdata/workflows/failure.yml

@@ -21,10 +21,3 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - run: echo dependent
-  timeout:
-    runs-on: ubuntu-latest
-    timeout-minutes: 1
-    steps:
-      - run: |
-          echo e2e-timeout-started
-          sleep 300

e2e/testdata/workflows/reconnect.yml

@@ -9,7 +9,10 @@ jobs:
     steps:
       - run: |
           echo e2e-before-outage
-          for i in $(seq -w 1 15); do
+          i=0
+          until curl -fs "${{ vars.E2E_GATE }}" > /dev/null; do
+            i=$((i + 1))
             echo "tick-$i"
-            sleep 1
+            sleep 0.2
           done
+          echo "ticks=$i"

e2e/testdata/workflows/timeout.yml

@@ -0,0 +1,9 @@
+name: timeout
+on: push
+jobs:
+  timeout:
+    runs-on: e2e-timeout
+    steps:
+      - run: |
+          echo e2e-timeout-started
+          sleep 300