vole
Clone
Read only, no credentials:
https://rodent.software/vole.git
Write over SSH with a registered key:
git clone ssh://git@rodent.software:2222/vole.git
A free and open-source forge that lives in your Git repo.
| 1 | package ssh |
| 2 | |
| 3 | // A resident runner should cost the queue nothing while it is idle. These tests pin |
| 4 | // the two halves of that: a lease with a wait parks until work is queued, and a lease |
| 5 | // without one answers immediately. The wait is a hint — the lease still decides — so |
| 6 | // what is measured here is latency, not permission. |
| 7 | |
| 8 | import ( |
| 9 | "fmt" |
| 10 | "strings" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "rodent.software/vole/ci/runs" |
| 15 | "rodent.software/vole/core/id" |
| 16 | ) |
| 17 | |
| 18 | func (h *harness) enqueueJob(t *testing.T, jobName string) { |
| 19 | t.Helper() |
| 20 | runID, err := id.New(time.Now().UTC()) |
| 21 | if err != nil { |
| 22 | t.Fatal(err) |
| 23 | } |
| 24 | if _, err := h.runs.Enqueue(runs.Run{ |
| 25 | ID: runID, |
| 26 | Repo: h.repoName, |
| 27 | Ref: "refs/heads/main", |
| 28 | Branch: "main", |
| 29 | SHA: strings.Repeat("ab", 20), |
| 30 | Workflow: ".vole/workflows/ci.yaml", |
| 31 | WorkflowSHA: strings.Repeat("cd", 20), |
| 32 | Job: jobName, |
| 33 | Steps: []runs.Step{{Name: "test", Run: "true"}}, |
| 34 | Created: time.Now().UTC(), |
| 35 | }, nil); err != nil { |
| 36 | t.Fatal(err) |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | // A connected runner is a row the page can show, and the row says what it is doing. |
| 41 | // Nothing else records that a runner exists, so this is the only proof it is visible. |
| 42 | func TestARunnerSessionIsVisibleWhileItIsConnected(t *testing.T) { |
| 43 | t.Parallel() |
| 44 | h := newHarness(t) |
| 45 | c := h.runner(t, h.keyPath) |
| 46 | if rep := c.call(runnerRequest{Op: "ping"}); rep.Error != "" { |
| 47 | t.Fatal(rep.Error) |
| 48 | } |
| 49 | |
| 50 | sessions, err := h.runs.Sessions(time.Time{}) |
| 51 | if err != nil { |
| 52 | t.Fatal(err) |
| 53 | } |
| 54 | if len(sessions) != 1 { |
| 55 | t.Fatalf("%d session(s) recorded for one connected runner", len(sessions)) |
| 56 | } |
| 57 | if sessions[0].Name != h.writerID { |
| 58 | t.Fatalf("the session is recorded for %q, want the account the key resolves to", sessions[0].Name) |
| 59 | } |
| 60 | if sessions[0].RunID != "" { |
| 61 | t.Fatalf("an idle runner is recorded as building %q", sessions[0].RunID) |
| 62 | } |
| 63 | |
| 64 | h.enqueueJob(t, "visible") |
| 65 | lease := c.call(runnerRequest{Op: "lease"}) |
| 66 | if lease.Run == nil { |
| 67 | t.Fatalf("no job was leased: %+v", lease) |
| 68 | } |
| 69 | sessions, err = h.runs.Sessions(time.Time{}) |
| 70 | if err != nil { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | if sessions[0].RunID != lease.Run.ID || sessions[0].Repo != h.repoName { |
| 74 | t.Fatalf("the session shows run %q in %q, want %q in %q", |
| 75 | sessions[0].RunID, sessions[0].Repo, lease.Run.ID, h.repoName) |
| 76 | } |
| 77 | } |
| 78 | |
| 79 | // The reply to a parked lease arrives because a job was queued, not because the wait |
| 80 | // ran out: the wait is 30 seconds and the assertion is five, so a poll would fail it. |
| 81 | func TestALeaseWaitsForWorkInsteadOfPolling(t *testing.T) { |
| 82 | t.Parallel() |
| 83 | h := newHarness(t) |
| 84 | c := h.runner(t, h.keyPath) |
| 85 | if rep := c.call(runnerRequest{Op: "ping"}); rep.Error != "" { |
| 86 | t.Fatal(rep.Error) |
| 87 | } |
| 88 | |
| 89 | type result struct { |
| 90 | rep runnerReply |
| 91 | err error |
| 92 | } |
| 93 | got := make(chan result, 1) |
| 94 | go func() { |
| 95 | if err := c.enc.Encode(runnerRequest{Op: "lease", Wait: 30}); err != nil { |
| 96 | got <- result{err: err} |
| 97 | return |
| 98 | } |
| 99 | var rep runnerReply |
| 100 | err := c.dec.Decode(&rep) |
| 101 | got <- result{rep: rep, err: err} |
| 102 | }() |
| 103 | |
| 104 | // Let the session park before queueing, so the wake-up is what answers it rather |
| 105 | // than the initial lease attempt. |
| 106 | time.Sleep(300 * time.Millisecond) |
| 107 | h.enqueueJob(t, "waited") |
| 108 | |
| 109 | select { |
| 110 | case r := <-got: |
| 111 | if r.err != nil { |
| 112 | t.Fatalf("reading the parked lease: %v", r.err) |
| 113 | } |
| 114 | if r.rep.Run == nil { |
| 115 | t.Fatalf("the parked lease answered with no job: %+v", r.rep) |
| 116 | } |
| 117 | if r.rep.Run.Job != "waited" { |
| 118 | t.Fatalf("leased %q, want the job that was queued", r.rep.Run.Job) |
| 119 | } |
| 120 | case <-time.After(5 * time.Second): |
| 121 | t.Fatal("a lease parked for 30s did not return when a job was queued") |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | // A one-shot runner is here for the job it was started for, so it must never park: an |
| 126 | // empty queue answers at once. |
| 127 | func TestALeaseWithoutAWaitAnswersAtOnce(t *testing.T) { |
| 128 | t.Parallel() |
| 129 | h := newHarness(t) |
| 130 | c := h.runner(t, h.keyPath) |
| 131 | |
| 132 | start := time.Now() |
| 133 | rep := c.call(runnerRequest{Op: "lease"}) |
| 134 | if rep.Error != "" { |
| 135 | t.Fatal(rep.Error) |
| 136 | } |
| 137 | if rep.Run != nil { |
| 138 | t.Fatalf("leased %q from an empty queue", rep.Run.Job) |
| 139 | } |
| 140 | if waited := time.Since(start); waited > time.Second { |
| 141 | t.Fatalf("a lease with no wait blocked for %s", waited) |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | // A runner with --jobs N is N sessions, and the queue has to hand each one a |
| 146 | // different job: two sessions that lease the same run would run it twice and race |
| 147 | // each other to report it. |
| 148 | func TestSeveralSessionsLeaseDifferentJobs(t *testing.T) { |
| 149 | t.Parallel() |
| 150 | h := newHarness(t) |
| 151 | const slots = 4 |
| 152 | // A distinct job name per run: enqueue is idempotent per (repo, ref, sha, |
| 153 | // workflow, job), so four identical enqueues are one run. |
| 154 | for i := 0; i < slots; i++ { |
| 155 | h.enqueueJob(t, fmt.Sprintf("slot-%d", i)) |
| 156 | } |
| 157 | |
| 158 | seen := make(map[string]bool, slots) |
| 159 | for i := 0; i < slots; i++ { |
| 160 | c := h.runner(t, h.keyPath) |
| 161 | rep := c.call(runnerRequest{Op: "lease"}) |
| 162 | if rep.Run == nil { |
| 163 | t.Fatalf("slot %d leased nothing: %+v", i, rep) |
| 164 | } |
| 165 | if seen[rep.Run.ID] { |
| 166 | t.Fatalf("two sessions leased the same run %s", rep.Run.ID) |
| 167 | } |
| 168 | seen[rep.Run.ID] = true |
| 169 | } |
| 170 | if len(seen) != slots { |
| 171 | t.Fatalf("%d distinct runs leased across %d sessions", len(seen), slots) |
| 172 | } |
| 173 | |
| 174 | // One job, one attempt: each session's lease is the attempt it reports under, and |
| 175 | // they cannot share one. |
| 176 | for id := range seen { |
| 177 | run, err := h.runs.Get(id) |
| 178 | if err != nil || run == nil { |
| 179 | t.Fatalf("leased run %s is not in the queue: %v", id, err) |
| 180 | } |
| 181 | if run.Attempts != 1 { |
| 182 | t.Fatalf("run %s is on attempt %d after one lease", id, run.Attempts) |
| 183 | } |
| 184 | } |
| 185 | } |
| 186 | |
| 187 | // max_jobs_per_repo is what stops one repository's burst from taking every slot on |
| 188 | // the instance. A repository at its cap is skipped rather than blocking the lease. |
| 189 | func TestARepositoryAtItsCapLeavesTheSlotsForOthers(t *testing.T) { |
| 190 | t.Parallel() |
| 191 | h := newHarnessWith(t, func(c *Config) { c.MaxJobsPerRepo = 2 }) |
| 192 | for i := 0; i < 3; i++ { |
| 193 | h.enqueueJob(t, fmt.Sprintf("burst-%d", i)) |
| 194 | } |
| 195 | |
| 196 | leased := 0 |
| 197 | for i := 0; i < 3; i++ { |
| 198 | c := h.runner(t, h.keyPath) |
| 199 | if c.call(runnerRequest{Op: "lease"}).Run != nil { |
| 200 | leased++ |
| 201 | } |
| 202 | } |
| 203 | if leased != 2 { |
| 204 | t.Fatalf("%d job(s) leased from a repository capped at 2", leased) |
| 205 | } |
| 206 | } |
| 207 |