Commit `3b4f05eb52`

3b4f05eb5218b33d26ca96e8b286c6a360c5660c

parent: 6f184c2936

Verified · cmc ci/build: success ci/test: success ci/vuln: success

cmc <hello@cleberg.net> · 2026-09-04 16:49 UTC

runner: -jobs N runs that many builds at once

One build per process, and the next one waited however long the current
took. ClaimBuild has always been a single transaction that selects and
updates, and each build already works in its own build-<id> directory, so
several workers were safe the whole time — the runner just never used
more than one.

N workers, each running the loop that was in main. -once still means one
build, whatever -jobs says. Idle polls are staggered across the interval,
so N workers do not all wake in the same instant to ask the same question.

TestRunnerConcurrentJobs asserts overlap from the runner's own log — a
build started before the first one finished — rather than from wall
clock, which also covers N concurrent clones and would make a slow
machine look serial. Pinning the runner back to one worker fails it.

The service weights in the runner's systemd drop-in are per service, not
per build, so -jobs divides them rather than multiplying the host's load.
Noted there.

Isolation, the other half of #115, is now #144: it needs a mechanism
chosen before any code, and `image:` per job means nothing until one is.

Closes #115

Layout: unified · split

cmd/gitbay-runner/main.go +39 −11

		@@ -58,6 +58,7 @@ func main() {
58	58	timeout = flag.Duration("timeout", 30*time.Minute, "per-build time limit")
59	59	repos = flag.String("repos", "", "only claim builds for these repositories, comma-separated owner/name (default: any)")
60	60	once = flag.Bool("once", false, "process at most one build, then exit")
	61	jobs = flag.Int("jobs", 1, "builds to run at once")
61	62	version = flag.Bool("version", false, "print the commit this binary was built from, then exit")
62	63	)
63	64	flag.Parse()
		@@ -88,18 +89,45 @@ func main() {
88	89	if err := os.MkdirAll(r.workdir, 0o755); err != nil {
89	90	log.Fatal(err)
90	91	}
91		for {
92		ran, err := r.step()
93		if err != nil {
94		log.Printf("runner: %v", err)
95		}
96		if *once {
97		return
98		}
99		if !ran {
100		time.Sleep(*poll)
101		}
	92	n := *jobs
	93	if n < 1 {
	94	log.Fatal("-jobs must be at least 1")
	95	}
	96	if *once {
	97	// "at most one build" is one build, whatever -jobs says.
	98	n = 1
	99	}
	100	// `runner next` claims inside one transaction, so several workers
	101	// claiming at once is already safe; the runner just never used that.
	102	// Each build works in its own build-<id> directory, so they do not
	103	// meet on disk either.
	104	var wg sync.WaitGroup
	105	for i := 0; i < n; i++ {
	106	wg.Add(1)
	107	go func(i int) {
	108	defer wg.Done()
	109	// Spread the idle polls across the interval rather than
	110	// having every worker wake together: n workers asking the
	111	// same question in the same instant is n times the load for
	112	// one answer.
	113	if n > 1 {
	114	time.Sleep(time.Duration(i) * *poll / time.Duration(n))
	115	}
	116	for {
	117	ran, err := r.step()
	118	if err != nil {
	119	log.Printf("runner: %v", err)
	120	}
	121	if *once {
	122	return
	123	}
	124	if !ran {
	125	time.Sleep(*poll)
	126	}
	127	}
	128	}(i)
102	129	}
	130	wg.Wait()
103	131	}
104	132
105	133	// step claims and executes at most one build. ran reports whether there was

deploy/gitbay-runner.override.conf +5

		@@ -6,6 +6,11 @@
6	6	# the admin sshd stalled at 1%. Lower CPU and IO weight keep sshd,
7	7	# gitbayd and the backup timers responsive while a build runs.
8	8	#
	9	# These weights are for the service, not per build, so `-jobs N` divides
	10	# them among N builds rather than taking N times as much. Raising -jobs
	11	# does not need them raised; it makes each build slower, not the host
	12	# busier.
	13	#
9	14	# A build runs whatever the repository's ci.yml says, as the runner's
10	15	# own user. Keep that user unprivileged: its key is added with
11	16	# `keys add --scope runner`, which confines it to the runner protocol

e2e/ci_test.go +106

		@@ -1,12 +1,14 @@
1	1	package e2e
2	2
3	3	import (
	4	"bytes"
4	5	"fmt"
5	6	"os"
6	7	"os/exec"
7	8	"path/filepath"
8	9	"strings"
9	10	"testing"
	11	"time"
10	12	)
11	13
12	14	func buildRunner(t *testing.T) string {
		@@ -268,3 +270,107 @@ func TestCI(t *testing.T) {
268	270	t.Fatalf("config failure status missing:\n%s", out)
269	271	}
270	272	}
	273
	274	// runnerJobs runs a runner with -jobs n until it has nothing left to do,
	275	// and returns its output. Unlike runnerOnce it is not bounded to one
	276	// build, so it is stopped when the queue drains.
	277	func (i instance) runnerJobs(t testing.T, key, repo string, jobs int) string {
	278	t.Helper()
	279	opts := fmt.Sprintf("-p %d -i %s -o IdentitiesOnly=yes -o StrictHostKeyChecking=no -o UserKnownHostsFile=%s -o BatchMode=yes",
	280	i.port, key, filepath.Join(i.sshDir, "known_hosts"))
	281	cmd := exec.Command(i.runner,
	282	"-jobs", fmt.Sprint(jobs),
	283	"-poll", "200ms",
	284	"-remote", "git@127.0.0.1",
	285	"-ssh-opts", opts,
	286	"-clone-base", fmt.Sprintf("ssh://git@127.0.0.1:%d", i.port),
	287	"-workdir", t.TempDir())
	288	cmd.Env = append(os.Environ(), "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL=/dev/null")
	289	var buf bytes.Buffer
	290	cmd.Stdout, cmd.Stderr = &buf, &buf
	291	if err := cmd.Start(); err != nil {
	292	t.Fatal(err)
	293	}
	294	defer func() { cmd.Process.Kill(); cmd.Wait() }()
	295
	296	// Wait for every queued build to leave the pending and running states.
	297	deadline := time.Now().Add(60 * time.Second)
	298	for time.Now().Before(deadline) {
	299	out, _, code := i.ssh(t, key, "", "build", "list", repo, "--json")
	300	if code == 0 && !strings.Contains(out, `"status":"pending"`) &&
	301	!strings.Contains(out, `"status":"running"`) {
	302	break
	303	}
	304	time.Sleep(200 * time.Millisecond)
	305	}
	306	return buf.String()
	307	}
	308
	309	// -jobs N runs N builds at once. ClaimBuild has always been a single
	310	// transaction that selects and updates, so several workers claiming
	311	// together is safe; the runner simply never used more than one (#115).
	312	func TestRunnerConcurrentJobs(t *testing.T) {
	313	inst := startInstance(t)
	314	inst.runner = buildRunner(t)
	315	aliceKey := inst.newKey(t, "alice")
	316	inst.admin(t, "admin", "user", "create", "alice", "--key", aliceKey+".pub", "--admin")
	317	if _, errOut, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 {
	318	t.Fatalf("repo create: %s", errOut)
	319	}
	320
	321	// Four jobs, each sleeping longer than the poll interval, so serial
	322	// execution and concurrent execution are distinguishable.
	323	ci := "jobs:\n"
	324	for _, name := range []string{"one", "two", "three", "four"} {
	325	ci += fmt.Sprintf(" %s:\n steps:\n - sleep 1\n - echo done-%s\n", name, name)
	326	}
	327	env := inst.gitEnv(aliceKey)
	328	work := t.TempDir()
	329	mustGit(t, work, env, "clone", inst.sshURL("alice/app"), "w")
	330	dir := filepath.Join(work, "w")
	331	os.MkdirAll(filepath.Join(dir, ".gitbay"), 0o755)
	332	os.WriteFile(filepath.Join(dir, ".gitbay", "ci.yml"), []byte(ci), 0o644)
	333	mustGit(t, dir, env, "checkout", "-q", "-b", "main")
	334	mustGit(t, dir, env, "add", ".")
	335	mustGit(t, dir, env, "commit", "-q", "-m", "ci")
	336	mustGit(t, dir, env, "push", "-q", "origin", "main")
	337
	338	out := inst.runnerJobs(t, aliceKey, "alice/app", 4)
	339
	340	listing, _, code := inst.ssh(t, aliceKey, "", "build", "list", "alice/app", "--json")
	341	if code != 0 {
	342	t.Fatalf("build list failed:\n%s", out)
	343	}
	344	for _, name := range []string{"one", "two", "three", "four"} {
	345	if !strings.Contains(listing, `"job":"`+name+`"`) {
	346	t.Fatalf("%s never ran:\n%s\n%s", name, listing, out)
	347	}
	348	}
	349	if strings.Contains(listing, `"status":"pending"`) \|\| strings.Contains(listing, `"status":"running"`) {
	350	t.Fatalf("builds did not finish:\n%s", listing)
	351	}
	352	// Overlap, asserted from the runner's own log rather than from wall
	353	// clock: elapsed time also covers four concurrent clones and the
	354	// polling this test does, and would make a slow machine look serial.
	355	// Every build announces itself when it starts and again when it
	356	// finishes, so concurrency is "a build started before the first one
	357	// finished".
	358	started, firstFinish := 0, -1
	359	for _, line := range strings.Split(out, "\n") {
	360	switch {
	361	case strings.Contains(line, "alice/app"):
	362	started++
	363	case strings.Contains(line, ": success"), strings.Contains(line, ": failure"):
	364	if firstFinish < 0 {
	365	firstFinish = started
	366	}
	367	}
	368	}
	369	if started != 4 {
	370	t.Fatalf("%d builds started, want 4:\n%s", started, out)
	371	}
	372	if firstFinish < 2 {
	373	t.Errorf("only %d build(s) had started when the first finished; -jobs 4 ran them serially\n%s",
	374	firstFinish, out)
	375	}
	376	}

Commit 3b4f05eb52

Commit `3b4f05eb52`