Commit `67fca35eea`

67fca35eeafd1a4384c4e59595c143d50c3eaa70

parent: 9da9327c7a

Verified · cmc

cmc <hello@cleberg.net> · 2026-09-09 02:36 UTC

store: runner keys attach to repositories, heartbeat per key

Migration 0050 adds runner_repos and rekeys runner_seen by ssh key.

Ref #184

internal/store/migrations/0050_runner_repos.down.sql added +8

		@@ -0,0 +1,8 @@
	1	DROP TABLE runner_repos;
	2	DROP TABLE runner_seen;
	3	CREATE TABLE runner_seen (
	4	user_id INTEGER PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
	5	last_seen TEXT NOT NULL,
	6	scope TEXT NOT NULL DEFAULT '',
	7	build_id INTEGER REFERENCES builds(id) ON DELETE SET NULL
	8	);

internal/store/migrations/0050_runner_repos.up.sql added +19

		@@ -0,0 +1,19 @@
	1	-- A runner key is attached to the repositories it may claim builds for
	2	-- (#184). runner_seen is rekeyed by key so two runners on one account
	3	-- are two rows; what it held were heartbeats, so the rows are dropped.
	4	CREATE TABLE runner_repos (
	5	key_id INTEGER NOT NULL REFERENCES ssh_keys(id) ON DELETE CASCADE,
	6	repo_id INTEGER NOT NULL REFERENCES repos(id) ON DELETE CASCADE,
	7	added_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
	8	PRIMARY KEY (key_id, repo_id)
	9	);
	10	CREATE INDEX runner_repos_repo ON runner_repos(repo_id);
	11
	12	DROP TABLE runner_seen;
	13	CREATE TABLE runner_seen (
	14	key_id INTEGER PRIMARY KEY REFERENCES ssh_keys(id) ON DELETE CASCADE,
	15	user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
	16	last_seen TEXT NOT NULL,
	17	scope TEXT NOT NULL DEFAULT '',
	18	build_id INTEGER REFERENCES builds(id) ON DELETE SET NULL
	19	);

internal/store/runners.go +141 −18

		@@ -1,10 +1,17 @@
1	1	package store
2	2
3		// Runner is one runner account as the instance admin sees it.
	3	import "sort"
	4
	5	// Runner is one runner key as the instance admin sees it.
4	6	type Runner struct {
5		Username string `json:"username"`
6		LastSeen string `json:"last_seen"`
7		Scope string `json:"scope,omitempty"` // comma-joined owner/name, "" for any
	7	Username string `json:"username"`
	8	Fingerprint string `json:"fingerprint"`
	9	KeyID int64 `json:"-"`
	10	LastSeen string `json:"last_seen"`
	11	// Scope is what the runner asked for: comma-joined owner/name, ""
	12	// for any. admin runners replaces it with the attachments for a
	13	// runner key.
	14	Scope string `json:"scope,omitempty"`
8	15	// The build it holds, if any.
9	16	BuildRepo string `json:"build_repo,omitempty"`
10	17	BuildNumber int64 `json:"build_number,omitempty"`
		@@ -12,32 +19,147 @@ type Runner struct {
12	19	StartedAt string `json:"started_at,omitempty"`
13	20	}
14	21
15		// TouchRunner records a poll: the time, the scope the runner asked for,
16		// and the build it just claimed (0 for none).
17		func (s *Store) TouchRunner(userID int64, scope string, buildID int64) error {
18		_, err := s.DB.Exec(`INSERT INTO runner_seen (user_id, last_seen, scope, build_id)
19		VALUES (?1, strftime('%Y-%m-%dT%H:%M:%fZ','now'), ?2, NULLIF(?3, 0))
20		ON CONFLICT (user_id) DO UPDATE SET
	22	// RepoRunner is one key attached to a repository, as repo runner list
	23	// shows it.
	24	type RepoRunner struct {
	25	Fingerprint string `json:"fingerprint"`
	26	Algo string `json:"algo"`
	27	Username string `json:"username"`
	28	AddedAt string `json:"added_at"`
	29	LastSeen string `json:"last_seen,omitempty"`
	30	BuildRepo string `json:"build_repo,omitempty"`
	31	BuildNumber int64 `json:"build_number,omitempty"`
	32	BuildJob string `json:"build_job,omitempty"`
	33	StartedAt string `json:"started_at,omitempty"`
	34	}
	35
	36	// AttachRunner lets a key claim a repository's builds. Attaching twice is
	37	// one row.
	38	func (s *Store) AttachRunner(keyID, repoID int64) error {
	39	_, err := s.DB.Exec("INSERT OR IGNORE INTO runner_repos (key_id, repo_id) VALUES (?, ?)", keyID, repoID)
	40	return err
	41	}
	42
	43	// DetachRunner removes one attachment by fingerprint. The key itself stays.
	44	func (s *Store) DetachRunner(repoID int64, fingerprint string) error {
	45	res, err := s.DB.Exec(`DELETE FROM runner_repos WHERE repo_id = ?
	46	AND key_id = (SELECT id FROM ssh_keys WHERE fingerprint = ?)`, repoID, fingerprint)
	47	if err != nil {
	48	return err
	49	}
	50	if n, _ := res.RowsAffected(); n == 0 {
	51	return ErrNotFound
	52	}
	53	return nil
	54	}
	55
	56	// RunnerRepoIDs is every repository a key is attached to.
	57	func (s *Store) RunnerRepoIDs(keyID int64) ([]int64, error) {
	58	rows, err := s.DB.Query("SELECT repo_id FROM runner_repos WHERE key_id = ? ORDER BY repo_id", keyID)
	59	if err != nil {
	60	return nil, err
	61	}
	62	defer rows.Close()
	63	var ids []int64
	64	for rows.Next() {
	65	var id int64
	66	if err := rows.Scan(&id); err != nil {
	67	return nil, err
	68	}
	69	ids = append(ids, id)
	70	}
	71	return ids, rows.Err()
	72	}
	73
	74	// RunnerRepoPaths is RunnerRepoIDs as owner/name, sorted.
	75	func (s *Store) RunnerRepoPaths(keyID int64) ([]string, error) {
	76	rows, err := s.DB.Query(`SELECT COALESCE(u.username, o.name) \|\| '/' \|\| r.name
	77	FROM runner_repos rr JOIN repos r ON r.id = rr.repo_id
	78	LEFT JOIN users u ON r.owner_kind = 'user' AND u.id = r.owner_id
	79	LEFT JOIN orgs o ON r.owner_kind = 'org' AND o.id = r.owner_id
	80	WHERE rr.key_id = ?`, keyID)
	81	if err != nil {
	82	return nil, err
	83	}
	84	defer rows.Close()
	85	var paths []string
	86	for rows.Next() {
	87	var p string
	88	if err := rows.Scan(&p); err != nil {
	89	return nil, err
	90	}
	91	paths = append(paths, p)
	92	}
	93	sort.Strings(paths)
	94	return paths, rows.Err()
	95	}
	96
	97	// RunnerAttached reports whether a key may claim a repository's builds.
	98	func (s *Store) RunnerAttached(keyID, repoID int64) (bool, error) {
	99	var n int
	100	err := s.DB.QueryRow("SELECT count(*) FROM runner_repos WHERE key_id = ? AND repo_id = ?", keyID, repoID).Scan(&n)
	101	return n > 0, err
	102	}
	103
	104	// ListRepoRunners is every key attached to a repository with its last
	105	// poll and the build it holds, oldest attachment first.
	106	func (s *Store) ListRepoRunners(repoID int64) ([]RepoRunner, error) {
	107	rows, err := s.DB.Query(`SELECT k.fingerprint, k.algo, u.username, rr.added_at,
	108	COALESCE(rs.last_seen, ''),
	109	COALESCE(COALESCE(bu.username, bo.name) \|\| '/' \|\| br.name, ''),
	110	COALESCE(b.number, 0), COALESCE(b.job, ''), COALESCE(b.started_at, '')
	111	FROM runner_repos rr
	112	JOIN ssh_keys k ON k.id = rr.key_id
	113	JOIN users u ON u.id = k.user_id
	114	LEFT JOIN runner_seen rs ON rs.key_id = rr.key_id
	115	LEFT JOIN builds b ON b.id = rs.build_id AND b.status = 'running'
	116	LEFT JOIN repos br ON br.id = b.repo_id
	117	LEFT JOIN users bu ON br.owner_kind = 'user' AND bu.id = br.owner_id
	118	LEFT JOIN orgs bo ON br.owner_kind = 'org' AND bo.id = br.owner_id
	119	WHERE rr.repo_id = ? ORDER BY rr.added_at, k.id`, repoID)
	120	if err != nil {
	121	return nil, err
	122	}
	123	defer rows.Close()
	124	var out []RepoRunner
	125	for rows.Next() {
	126	var r RepoRunner
	127	if err := rows.Scan(&r.Fingerprint, &r.Algo, &r.Username, &r.AddedAt, &r.LastSeen,
	128	&r.BuildRepo, &r.BuildNumber, &r.BuildJob, &r.StartedAt); err != nil {
	129	return nil, err
	130	}
	131	out = append(out, r)
	132	}
	133	return out, rows.Err()
	134	}
	135
	136	// TouchRunner records a poll by one key: the time, the scope the runner
	137	// asked for, and the build it just claimed (0 for none).
	138	func (s *Store) TouchRunner(keyID, userID int64, scope string, buildID int64) error {
	139	_, err := s.DB.Exec(`INSERT INTO runner_seen (key_id, user_id, last_seen, scope, build_id)
	140	VALUES (?1, ?2, strftime('%Y-%m-%dT%H:%M:%fZ','now'), ?3, NULLIF(?4, 0))
	141	ON CONFLICT (key_id) DO UPDATE SET
21	142	last_seen = excluded.last_seen, scope = excluded.scope,
22	143	build_id = COALESCE(excluded.build_id, runner_seen.build_id)`,
23		userID, scope, buildID)
	144	keyID, userID, scope, buildID)
24	145	return err
25	146	}
26	147
27		// RunnerDone records that the runner reported and holds nothing now.
28		func (s *Store) RunnerDone(userID int64) error {
	148	// RunnerDone records that the key reported and holds nothing now.
	149	func (s *Store) RunnerDone(keyID int64) error {
29	150	_, err := s.DB.Exec(`UPDATE runner_seen SET last_seen = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
30		build_id = NULL WHERE user_id = ?`, userID)
	151	build_id = NULL WHERE key_id = ?`, keyID)
31	152	return err
32	153	}
33	154
34		// ListRunners lists every account that has ever polled as a runner,
35		// most recently seen first.
	155	// ListRunners lists every key that has ever polled as a runner, most
	156	// recently seen first.
36	157	func (s *Store) ListRunners() ([]Runner, error) {
37		rows, err := s.DB.Query(`SELECT u.username, r.last_seen, r.scope,
	158	rows, err := s.DB.Query(`SELECT u.username, k.fingerprint, k.id, r.last_seen, r.scope,
38	159	COALESCE(COALESCE(bu.username, bo.name) \|\| '/' \|\| br.name, ''),
39	160	COALESCE(b.number, 0), COALESCE(b.job, ''), COALESCE(b.started_at, '')
40	161	FROM runner_seen r JOIN users u ON u.id = r.user_id
	162	JOIN ssh_keys k ON k.id = r.key_id
41	163	LEFT JOIN builds b ON b.id = r.build_id AND b.status = 'running'
42	164	LEFT JOIN repos br ON br.id = b.repo_id
43	165	LEFT JOIN users bu ON br.owner_kind = 'user' AND bu.id = br.owner_id
		@@ -50,7 +172,8 @@ func (s *Store) ListRunners() ([]Runner, error) {
50	172	var out []Runner
51	173	for rows.Next() {
52	174	var r Runner
53		if err := rows.Scan(&r.Username, &r.LastSeen, &r.Scope, &r.BuildRepo, &r.BuildNumber, &r.BuildJob, &r.StartedAt); err != nil {
	175	if err := rows.Scan(&r.Username, &r.Fingerprint, &r.KeyID, &r.LastSeen, &r.Scope,
	176	&r.BuildRepo, &r.BuildNumber, &r.BuildJob, &r.StartedAt); err != nil {
54	177	return nil, err
55	178	}
56	179	out = append(out, r)

internal/store/runners_test.go added +148

		@@ -0,0 +1,148 @@
	1	package store
	2
	3	import (
	4	"errors"
	5	"testing"
	6	)
	7
	8	// runnerFixture is one user with a runner key and two repositories.
	9	func runnerFixture(t testing.T) (s Store, uid, keyID, repoA, repoB int64) {
	10	t.Helper()
	11	s = open(t)
	12	if err := s.MigrateUp(); err != nil {
	13	t.Fatal(err)
	14	}
	15	uid, err := s.CreateUser("alice", false)
	16	if err != nil {
	17	t.Fatal(err)
	18	}
	19	if err := s.AddSSHKey(uid, "SHA256:runnerkey", "ssh-ed25519", []byte("blob"), "runner"); err != nil {
	20	t.Fatal(err)
	21	}
	22	k, err := s.SSHKeyByFingerprint("SHA256:runnerkey")
	23	if err != nil {
	24	t.Fatal(err)
	25	}
	26	repoA, err = s.CreateRepo("user", uid, "a", "public")
	27	if err != nil {
	28	t.Fatal(err)
	29	}
	30	repoB, err = s.CreateRepo("user", uid, "b", "public")
	31	if err != nil {
	32	t.Fatal(err)
	33	}
	34	return s, uid, k.ID, repoA, repoB
	35	}
	36
	37	// Attaching twice is one row; detaching what is not attached is not found.
	38	func TestAttachRunnerIdempotentAndDetach(t *testing.T) {
	39	s, _, keyID, repoA, repoB := runnerFixture(t)
	40	for range 2 {
	41	if err := s.AttachRunner(keyID, repoA); err != nil {
	42	t.Fatal(err)
	43	}
	44	}
	45	ids, err := s.RunnerRepoIDs(keyID)
	46	if err != nil \|\| len(ids) != 1 \|\| ids[0] != repoA {
	47	t.Fatalf("attached repos %v err=%v, want [%d]", ids, err, repoA)
	48	}
	49	if ok, _ := s.RunnerAttached(keyID, repoB); ok {
	50	t.Fatal("attached to a repo it was never attached to")
	51	}
	52	if err := s.DetachRunner(repoB, "SHA256:runnerkey"); !errors.Is(err, ErrNotFound) {
	53	t.Fatalf("detach of an unattached repo: %v, want ErrNotFound", err)
	54	}
	55	if err := s.DetachRunner(repoA, "SHA256:runnerkey"); err != nil {
	56	t.Fatal(err)
	57	}
	58	if ok, _ := s.RunnerAttached(keyID, repoA); ok {
	59	t.Fatal("still attached after detach")
	60	}
	61	}
	62
	63	// Removing the key or the repository removes the attachment with it.
	64	func TestRunnerAttachmentCascades(t *testing.T) {
	65	s, uid, keyID, repoA, repoB := runnerFixture(t)
	66	if err := s.AttachRunner(keyID, repoA); err != nil {
	67	t.Fatal(err)
	68	}
	69	if err := s.AttachRunner(keyID, repoB); err != nil {
	70	t.Fatal(err)
	71	}
	72	if _, err := s.DB.Exec("DELETE FROM repos WHERE id = ?", repoB); err != nil {
	73	t.Fatal(err)
	74	}
	75	if ids, _ := s.RunnerRepoIDs(keyID); len(ids) != 1 {
	76	t.Fatalf("after repo delete: %v, want one attachment", ids)
	77	}
	78	if err := s.RemoveSSHKey(uid, "SHA256:runnerkey"); err != nil {
	79	t.Fatal(err)
	80	}
	81	var n int
	82	if err := s.DB.QueryRow("SELECT count(*) FROM runner_repos").Scan(&n); err != nil \|\| n != 0 {
	83	t.Fatalf("after key delete: %d rows err=%v, want 0", n, err)
	84	}
	85	}
	86
	87	// The heartbeat is per key: two keys on one account are two rows, and a
	88	// repository's runner list shows each key's last poll and the build it holds.
	89	func TestRunnerSeenPerKeyAndRepoList(t *testing.T) {
	90	s, uid, keyID, repoA, _ := runnerFixture(t)
	91	if err := s.AddSSHKey(uid, "SHA256:second", "ssh-ed25519", []byte("blob2"), "runner"); err != nil {
	92	t.Fatal(err)
	93	}
	94	k2, _ := s.SSHKeyByFingerprint("SHA256:second")
	95	for _, id := range []int64{keyID, k2.ID} {
	96	if err := s.AttachRunner(id, repoA); err != nil {
	97	t.Fatal(err)
	98	}
	99	}
	100	if _, err := s.CreateBuild(repoA, "unit", "abc123", "main", `["true"]`, "", "", true); err != nil {
	101	t.Fatal(err)
	102	}
	103	b, ok, err := s.ClaimBuild(nil, false)
	104	if err != nil \|\| !ok {
	105	t.Fatalf("claim: %v ok=%v", err, ok)
	106	}
	107	if err := s.TouchRunner(keyID, uid, "", b.ID); err != nil {
	108	t.Fatal(err)
	109	}
	110	if err := s.TouchRunner(k2.ID, uid, "", 0); err != nil {
	111	t.Fatal(err)
	112	}
	113	runners, err := s.ListRunners()
	114	if err != nil \|\| len(runners) != 2 {
	115	t.Fatalf("ListRunners: %v err=%v, want two rows", runners, err)
	116	}
	117	list, err := s.ListRepoRunners(repoA)
	118	if err != nil \|\| len(list) != 2 {
	119	t.Fatalf("ListRepoRunners: %v err=%v, want two rows", list, err)
	120	}
	121	var held, idle int
	122	for _, r := range list {
	123	if r.Username != "alice" \|\| r.LastSeen == "" \|\| r.AddedAt == "" {
	124	t.Fatalf("row %+v lacks username, last_seen or added_at", r)
	125	}
	126	if r.BuildNumber == b.Number && r.BuildJob == "unit" && r.BuildRepo == "alice/a" {
	127	held++
	128	} else if r.BuildNumber == 0 {
	129	idle++
	130	}
	131	}
	132	if held != 1 \|\| idle != 1 {
	133	t.Fatalf("held=%d idle=%d, want 1 and 1: %+v", held, idle, list)
	134	}
	135	if err := s.RunnerDone(keyID); err != nil {
	136	t.Fatal(err)
	137	}
	138	list, _ = s.ListRepoRunners(repoA)
	139	for _, r := range list {
	140	if r.BuildNumber != 0 {
	141	t.Fatalf("build still held after RunnerDone: %+v", r)
	142	}
	143	}
	144	paths, err := s.RunnerRepoPaths(keyID)
	145	if err != nil \|\| len(paths) != 1 \|\| paths[0] != "alice/a" {
	146	t.Fatalf("RunnerRepoPaths: %v err=%v", paths, err)
	147	}
	148	}

Commit 67fca35eea

Commit `67fca35eea`