1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore_test
4
5import (
6 "context"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/assert"
12 "github.com/stretchr/testify/require"
13
14 "bnema.dev/ermite/internal/adapters/kvstore"
15 "bnema.dev/ermite/internal/domain"
16 "bnema.dev/ermite/internal/ports"
17)
18
19// ciRepo opens a store holding repository "proj".
20func ciRepo(t *testing.T) (*kvstore.Store, *kvstore.CI, domain.RepoName) {
21 t.Helper()
22 s := openStore(t, t.TempDir())
23 n := name(t, "proj")
24 require.NoError(t, s.Create(context.Background(), domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
25 return s, s.CI(), n
26}
27
28func draftOf(repo domain.RepoName, workflow string, jobs int) func(id, number int64) ports.CIRunDraft {
29 return func(id, number int64) ports.CIRunDraft {
30 d := ports.CIRunDraft{
31 Run: domain.CIRun{
32 Repo: repo, ID: id, Number: number, Attempt: 1, WorkflowPath: workflow, WorkflowName: "ci", Event: "push",
33 Ref: "refs/heads/main", Sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Status: domain.CIQueued, CreatedAt: created,
34 },
35 Source: domain.CIRunSource{WorkflowYAML: "on: push", GithubJSON: "{}"},
36 }
37 for i := range jobs {
38 j := domain.CIJob{
39 ID: domain.NewJobID(id, i, 0), RunID: id, Attempt: 1, RunAttempt: 1, Repo: repo, Key: "build", Name: "build",
40 Phase: domain.CIPhaseReady, ReadyAt: created.Add(time.Duration(i) * time.Second), CreatedAt: created,
41 Labels: []string{"self-hosted"}, Index: i, TimeoutMinutes: 360, Planned: true,
42 }
43 d.Jobs = append(d.Jobs, j)
44 d.Checks = append(d.Checks, domain.CICheck{
45 ID: j.ID, Repo: repo, Sha: d.Run.Sha, Name: j.Name, Status: domain.CIQueued, RunID: id, JobID: j.ID, Attempt: 1,
46 WorkflowName: "ci", CreatedAt: created,
47 })
48 }
49 return d
50 }
51}
52
53func TestCICreateRunTakesIncreasingIdsAndPerWorkflowNumbers(t *testing.T) {
54 ctx := context.Background()
55 _, ci, n := ciRepo(t)
56
57 a, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 1))
58 require.NoError(t, err)
59 b, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 1))
60 require.NoError(t, err)
61 c, err := ci.CreateRun(ctx, n, ".github/workflows/lint.yml", draftOf(n, ".github/workflows/lint.yml", 0))
62 require.NoError(t, err)
63
64 assert.Equal(t, []int64{1, 2, 3}, []int64{a.Run.ID, b.Run.ID, c.Run.ID})
65 assert.Equal(t, []int64{1, 2, 1}, []int64{a.Run.Number, b.Run.Number, c.Run.Number})
66 assert.NotZero(t, a.Run.Version)
67}
68
69func TestCIConcurrentCreateRunNeverSharesAnId(t *testing.T) {
70 ctx := context.Background()
71 _, ci, n := ciRepo(t)
72 var mu sync.Mutex
73 ids := map[int64]bool{}
74 var wg sync.WaitGroup
75 for range 12 {
76 wg.Add(1)
77 go func() {
78 defer wg.Done()
79 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
80 if assert.NoError(t, err) {
81 mu.Lock()
82 defer mu.Unlock()
83 assert.False(t, ids[d.Run.ID], "id %d handed out twice", d.Run.ID)
84 ids[d.Run.ID] = true
85 }
86 }()
87 }
88 wg.Wait()
89 assert.Len(t, ids, 12)
90}
91
92func TestCICreateRunFailsOnAMissingRepository(t *testing.T) {
93 _, ci, _ := ciRepo(t)
94 ghost := name(t, "ghost")
95 _, err := ci.CreateRun(context.Background(), ghost, "w.yml", draftOf(ghost, "w.yml", 1))
96 assert.ErrorIs(t, err, domain.ErrNotFound)
97}
98
99func TestCIJobIndexesFollowThePhase(t *testing.T) {
100 ctx := context.Background()
101 _, ci, n := ciRepo(t)
102 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 2))
103 require.NoError(t, err)
104
105 ready, err := ci.ListReadyJobs(ctx, 10)
106 require.NoError(t, err)
107 require.Len(t, ready, 2)
108 assert.Equal(t, d.Jobs[0].ID, ready[0].ID, "oldest first")
109
110 open, err := ci.ListOpenRuns(ctx)
111 require.NoError(t, err)
112 assert.Len(t, open, 1)
113
114 // Assign the first job: it leaves the queue, enters the active set and its
115 // token becomes findable.
116 j := ready[0]
117 hash := domain.HashTokenSecret("ermitejob_x")
118 j.Phase, j.TokenHash, j.AssignedAt = domain.CIPhaseAssigned, hash, created.Add(time.Minute)
119 j.TokenValidUntil = created.Add(48 * time.Hour)
120 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
121 require.NoError(t, err)
122
123 ready, _ = ci.ListReadyJobs(ctx, 10)
124 assert.Len(t, ready, 1)
125 active, err := ci.ListActiveJobs(ctx)
126 require.NoError(t, err)
127 require.Len(t, active, 1)
128 assert.Equal(t, j.ID, active[0].ID)
129 found, err := ci.FindJobByToken(ctx, hash, created.Add(2*time.Minute))
130 require.NoError(t, err)
131 assert.Equal(t, j.ID, found.ID)
132 _, err = ci.FindJobByToken(ctx, hash, created.Add(72*time.Hour))
133 assert.ErrorIs(t, err, domain.ErrNotFound, "an expired token is not found")
134
135 // A new token replaces the old one.
136 j2, err := ci.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
137 require.NoError(t, err)
138 j2.TokenHash = domain.HashTokenSecret("ermitejob_y")
139 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j2}})
140 require.NoError(t, err)
141 _, err = ci.FindJobByToken(ctx, hash, created.Add(2*time.Minute))
142 assert.ErrorIs(t, err, domain.ErrNotFound, "the replaced token is gone")
143
144 // End it.
145 j3, _ := ci.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
146 j3.Phase, j3.Conclusion = domain.CIPhaseEnded, domain.CISuccess
147 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j3}})
148 require.NoError(t, err)
149 active, _ = ci.ListActiveJobs(ctx)
150 assert.Empty(t, active)
151}
152
153func TestCIApplyChecksVersions(t *testing.T) {
154 ctx := context.Background()
155 _, ci, n := ciRepo(t)
156 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
157 require.NoError(t, err)
158
159 a, err := ci.GetRun(ctx, n, d.Run.ID)
160 require.NoError(t, err)
161 b := a
162 a.Status = domain.CIInProgress
163 _, err = ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{a}})
164 require.NoError(t, err)
165 b.Status = domain.CICompleted
166 _, err = ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{b}})
167 assert.ErrorIs(t, err, domain.ErrConflict, "a write on a stale copy loses")
168}
169
170func TestCIChecksAreOrderedAndLimited(t *testing.T) {
171 ctx := context.Background()
172 _, ci, n := ciRepo(t)
173 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 3))
174 require.NoError(t, err)
175
176 checks, more, err := ci.ListChecks(ctx, n, d.Run.Sha, 10)
177 require.NoError(t, err)
178 assert.Len(t, checks, 3)
179 assert.False(t, more)
180 checks, more, err = ci.ListChecks(ctx, n, d.Run.Sha, 2)
181 require.NoError(t, err)
182 assert.Len(t, checks, 2)
183 assert.True(t, more)
184 none, _, err := ci.ListChecks(ctx, n, "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", 10)
185 require.NoError(t, err)
186 assert.Empty(t, none)
187}
188
189func TestCIListRunsNewestFirstAndBounded(t *testing.T) {
190 ctx := context.Background()
191 s, ci, n := ciRepo(t)
192 other := name(t, "other")
193 require.NoError(t, s.Create(ctx, domain.Repo{Name: other, Visibility: domain.VisibilityPublic, CreatedAt: created}))
194 _, err := ci.CreateRun(ctx, other, "w.yml", draftOf(other, "w.yml", 0))
195 require.NoError(t, err)
196 for range 5 {
197 _, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 0))
198 require.NoError(t, err)
199 }
200 runs, err := ci.ListRuns(ctx, n, 0, 3)
201 require.NoError(t, err)
202 require.Len(t, runs, 3)
203 assert.Equal(t, []int64{6, 5, 4}, []int64{runs[0].ID, runs[1].ID, runs[2].ID})
204 runs, err = ci.ListRuns(ctx, n, 4, 10)
205 require.NoError(t, err)
206 assert.Equal(t, []int64{3, 2}, []int64{runs[0].ID, runs[1].ID})
207 ids, err := ci.RunIDs(ctx, n)
208 require.NoError(t, err)
209 assert.Len(t, ids, 5)
210}
211
212func TestCIRunnersCreateReplaceFindRemove(t *testing.T) {
213 ctx := context.Background()
214 _, ci, _ := ciRepo(t)
215 h1, h2 := domain.HashTokenSecret("ermiterun_1"), domain.HashTokenSecret("ermiterun_2")
216 r1 := domain.CIRunner{ID: "r1", Name: "Alpha", Labels: []string{"linux"}, Isolation: domain.CIIsolationVM, TokenHash: h1, RegisteredAt: created}
217 _, err := ci.CreateRunner(ctx, r1, false)
218 require.NoError(t, err)
219
220 r2 := domain.CIRunner{ID: "r2", Name: "alpha", TokenHash: h2, RegisteredAt: created}
221 _, err = ci.CreateRunner(ctx, r2, false)
222 assert.ErrorIs(t, err, domain.ErrAlreadyExists, "names are case-insensitive")
223
224 removed, err := ci.CreateRunner(ctx, r2, true)
225 require.NoError(t, err)
226 require.Len(t, removed, 1)
227 assert.Equal(t, "r1", removed[0].ID)
228 _, err = ci.FindRunnerByToken(ctx, h1)
229 assert.ErrorIs(t, err, domain.ErrNotFound, "the replaced runner's token stops working")
230 got, err := ci.FindRunnerByToken(ctx, h2)
231 require.NoError(t, err)
232 assert.Equal(t, "r2", got.ID)
233
234 list, err := ci.ListRunners(ctx)
235 require.NoError(t, err)
236 assert.Len(t, list, 1)
237
238 _, err = ci.RemoveRunner(ctx, "r2")
239 require.NoError(t, err)
240 _, err = ci.RemoveRunner(ctx, "r2")
241 assert.ErrorIs(t, err, domain.ErrNotFound)
242 _, err = ci.FindRunnerByToken(ctx, h2)
243 assert.ErrorIs(t, err, domain.ErrNotFound)
244 // The name is free again.
245 _, err = ci.CreateRunner(ctx, domain.CIRunner{ID: "r3", Name: "alpha", TokenHash: domain.HashTokenSecret("ermiterun_3")}, false)
246 assert.NoError(t, err)
247}
248
249func TestCISweepRemovesOldCompletedRunsOnly(t *testing.T) {
250 ctx := context.Background()
251 _, ci, n := ciRepo(t)
252 oldRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
253 require.NoError(t, err)
254 newRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
255 require.NoError(t, err)
256 openRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
257 require.NoError(t, err)
258
259 for _, tc := range []struct {
260 d ports.CIRunDraft
261 at time.Time
262 }{{oldRun, created}, {newRun, created.Add(100 * 24 * time.Hour)}} {
263 run, _ := ci.GetRun(ctx, n, tc.d.Run.ID)
264 run.Status, run.Conclusion, run.CompletedAt = domain.CICompleted, domain.CISuccess, tc.at
265 job, _ := ci.GetJob(ctx, run.ID, 1, tc.d.Jobs[0].ID)
266 job.Phase, job.Conclusion = domain.CIPhaseEnded, domain.CISuccess
267 _, err := ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{run}, Jobs: []domain.CIJob{job}})
268 require.NoError(t, err)
269 }
270
271 ids, err := ci.Sweep(ctx, created.Add(90*24*time.Hour))
272 require.NoError(t, err)
273 assert.Equal(t, []int64{oldRun.Run.ID}, ids)
274
275 _, err = ci.GetRun(ctx, n, oldRun.Run.ID)
276 assert.ErrorIs(t, err, domain.ErrNotFound)
277 _, err = ci.GetRunSource(ctx, oldRun.Run.ID)
278 assert.ErrorIs(t, err, domain.ErrNotFound)
279 jobs, _ := ci.ListJobs(ctx, oldRun.Run.ID, 1)
280 assert.Empty(t, jobs)
281 _, err = ci.GetRun(ctx, n, newRun.Run.ID)
282 assert.NoError(t, err)
283 _, err = ci.GetRun(ctx, n, openRun.Run.ID)
284 assert.NoError(t, err, "an open run is never swept")
285
286 // Numbers are never reused.
287 next, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 0))
288 require.NoError(t, err)
289 assert.Equal(t, int64(4), next.Run.Number)
290}
291
292func TestCIRepositoryDeletionRemovesEverythingAndRestartsNumbers(t *testing.T) {
293 ctx := context.Background()
294 s, ci, n := ciRepo(t)
295 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 2))
296 require.NoError(t, err)
297 _, err = ci.CreateRunner(ctx, domain.CIRunner{ID: "r1", Name: "a", TokenHash: domain.HashTokenSecret("x")}, false)
298 require.NoError(t, err)
299
300 require.NoError(t, s.Delete(ctx, n))
301
302 _, err = ci.GetRun(ctx, n, d.Run.ID)
303 assert.ErrorIs(t, err, domain.ErrNotFound)
304 _, err = ci.GetRunSource(ctx, d.Run.ID)
305 assert.ErrorIs(t, err, domain.ErrNotFound)
306 jobs, _ := ci.ListJobs(ctx, d.Run.ID, 1)
307 assert.Empty(t, jobs)
308 checks, _, _ := ci.ListChecks(ctx, n, d.Run.Sha, 10)
309 assert.Empty(t, checks)
310 ready, _ := ci.ListReadyJobs(ctx, 10)
311 assert.Empty(t, ready)
312 open, _ := ci.ListOpenRuns(ctx)
313 assert.Empty(t, open)
314 runners, _ := ci.ListRunners(ctx)
315 assert.Len(t, runners, 1, "runners do not belong to a repository")
316
317 // A repository created again under the name starts its run numbers at 1,
318 // while ids are never reused.
319 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
320 again, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 0))
321 require.NoError(t, err)
322 assert.Equal(t, int64(1), again.Run.Number)
323 assert.Equal(t, int64(2), again.Run.ID)
324}
325
326func TestCIStateSurvivesARestart(t *testing.T) {
327 ctx := context.Background()
328 dir := t.TempDir()
329 s, err := kvstore.Open(ctx, dir)
330 require.NoError(t, err)
331 n := name(t, "proj")
332 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
333 ci := s.CI()
334 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
335 require.NoError(t, err)
336 j := d.Jobs[0]
337 hash := domain.HashTokenSecret("ermitejob_z")
338 j.Phase, j.TokenHash, j.RunnerID, j.RunnerName = domain.CIPhaseRunning, hash, "r1", "alpha"
339 j.Steps = []domain.CIStep{{Number: 1, Name: "Set up job", State: domain.CIStepCompleted, Conclusion: domain.CISuccess, HasLines: true, LogStart: 0}}
340 j.Outputs = map[string]string{"a": "b"}
341 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
342 require.NoError(t, err)
343 require.NoError(t, s.Close(ctx))
344
345 s2, err := kvstore.Open(ctx, dir)
346 require.NoError(t, err)
347 defer func() { _ = s2.Close(ctx) }()
348 ci2 := s2.CI()
349 got, err := ci2.FindJobByToken(ctx, hash, time.Time{})
350 require.NoError(t, err)
351 assert.Equal(t, domain.CIPhaseRunning, got.Phase)
352 assert.Equal(t, "alpha", got.RunnerName)
353 assert.Equal(t, map[string]string{"a": "b"}, got.Outputs)
354 require.Len(t, got.Steps, 1)
355 assert.True(t, got.Steps[0].HasLines)
356 active, _ := ci2.ListActiveJobs(ctx)
357 assert.Len(t, active, 1)
358}
359
360func TestCICancelReasonsRoundTrip(t *testing.T) {
361 ctx := context.Background()
362 _, ci, n := ciRepo(t)
363 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
364 require.NoError(t, err)
365 for _, reason := range []domain.CICancelReason{domain.CICancelUser, domain.CICancelFailFast, domain.CICancelTimeout, domain.CICancelNone} {
366 j, err := ci.GetJob(ctx, d.Run.ID, 1, d.Jobs[0].ID)
367 require.NoError(t, err)
368 j.CancelReason, j.CancelMessage, j.CancelRequestedAt = reason, "why", created
369 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
370 require.NoError(t, err)
371 got, err := ci.GetJob(ctx, d.Run.ID, 1, d.Jobs[0].ID)
372 require.NoError(t, err)
373 assert.Equal(t, reason, got.CancelReason)
374 }
375}
376
377// Run.Name round-trips, and a run stored without one (before the field existed)
378// reads as its workflow name.
379func TestCIRunNameRoundTripsAndDefaultsToTheWorkflowName(t *testing.T) {
380 ctx := context.Background()
381 _, ci, n := ciRepo(t)
382
383 old, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 0))
384 require.NoError(t, err)
385 got, err := ci.GetRun(ctx, n, old.Run.ID)
386 require.NoError(t, err)
387 assert.Equal(t, "ci", got.WorkflowName)
388 assert.Equal(t, "ci", got.Name, "no stored name: the workflow name")
389
390 named, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", func(id, number int64) ports.CIRunDraft {
391 d := draftOf(n, ".github/workflows/ci.yml", 0)(id, number)
392 d.Run.Name = "run push on branch by someone"
393 return d
394 })
395 require.NoError(t, err)
396 got, err = ci.GetRun(ctx, n, named.Run.ID)
397 require.NoError(t, err)
398 assert.Equal(t, "run push on branch by someone", got.Name)
399 assert.Equal(t, "ci", got.WorkflowName, "the workflow name is unchanged")
400}