1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "context"
7 "fmt"
8 "net/http/httptest"
9 "strings"
10 "sync/atomic"
11 "testing"
12 "time"
13
14 "bnema.dev/ermite/internal/domain"
15
16 "connectrpc.com/connect"
17 "github.com/stretchr/testify/assert"
18 "github.com/stretchr/testify/require"
19
20 pb "bnema.dev/ermite-api/gen/ermite"
21 "bnema.dev/ermite/internal/ports"
22 "bnema.dev/ermite/internal/usecase"
23)
24
25// concurrencyHarness is a CI harness with fast deadlines and one runner.
26func concurrencyHarness(t *testing.T) (*ciHarness, *fakeRunner) {
27 t.Helper()
28 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
29 cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
30 })
31 return c, c.registerRunner("alpha", "ubuntu-latest")
32}
33
34// workflowAt is a one-job workflow on main with extra top-level and job lines.
35func workflowAt(top, job string) string {
36 return "name: ci\non:\n push:\n branches: [main]\n" + top +
37 "jobs:\n build:\n runs-on: ubuntu-latest\n" + job +
38 " steps:\n - run: echo hello\n"
39}
40
41// pushN pushes a new commit with the workflow and returns the new run.
42func (c *ciHarness) pushN(n int, wf string) *pb.Run {
43 c.t.Helper()
44 before := len(c.runs())
45 // A comment makes each push a new commit even when the workflow repeats.
46 c.push("main", map[string]string{".github/workflows/ci.yml": fmt.Sprintf("# push %d\n%s", n, wf)}, "")
47 c.waitFor("the run exists", func() bool { return len(c.runs()) == before+1 })
48 return c.runs()[0] // newest first
49}
50
51func (c *ciHarness) jobsOf(run *pb.Run) []*pb.RunJob { return c.getRun(run.GetId()).GetJobs() }
52
53func (c *ciHarness) waitJobs(what string, run *pb.Run, cond func([]*pb.RunJob) bool) {
54 c.t.Helper()
55 c.waitFor(what, func() bool { return cond(c.jobsOf(run)) })
56}
57
58func allStatus(s pb.CiStatus) func([]*pb.RunJob) bool {
59 return func(js []*pb.RunJob) bool {
60 for _, j := range js {
61 if j.GetStatus() != s {
62 return false
63 }
64 }
65 return len(js) > 0
66 }
67}
68
69func (c *ciHarness) conclusionOf(run *pb.Run) pb.CiConclusion {
70 return c.getRun(run.GetId()).GetRun().GetConclusion()
71}
72
73func (c *ciHarness) cancelAnnotation(run *pb.Run, group string) bool {
74 for _, j := range c.jobsOf(run) {
75 for _, a := range c.annotations(run.GetId(), j.GetId()).GetAnnotations() {
76 if a.GetAnnotation().GetMessage() == "Canceling since a higher priority waiting request for '"+group+"' exists" {
77 return true
78 }
79 }
80 }
81 return false
82}
83
84const wfGroup = "concurrency:\n group: Deploy\n"
85
86// AC4: a workflow-level group lets one run go and one wait; a newer waiting run
87// replaces the older one, which ends cancelled with GitHub's annotation; the
88// group is handed over when the running one completes.
89func TestCIWorkflowConcurrencyPendingReplacement(t *testing.T) {
90 c, runner := concurrencyHarness(t)
91 r1 := c.pushN(1, workflowAt(wfGroup, ""))
92 job1 := runner.await()
93 require.Equal(t, r1.GetId(), job1.GetRunId())
94
95 r2 := c.pushN(2, workflowAt(wfGroup, ""))
96 c.waitJobs("the second run is pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
97 assert.Equal(t, "Deploy", c.jobsOf(r2)[0].GetConcurrencyGroup())
98 assert.Equal(t, pb.CiStatus_CI_STATUS_QUEUED, c.getRun(r2.GetId()).GetRun().GetStatus(), "a run is queued, only its jobs are pending")
99 // The pending job is not assignable.
100 j, err := runner.fetch("")
101 require.NoError(t, err)
102 assert.Nil(t, j)
103
104 r3 := c.pushN(3, workflowAt(wfGroup, ""))
105 c.waitFor("the older pending run is cancelled", func() bool {
106 return c.getRun(r2.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
107 })
108 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(r2))
109 assert.True(t, c.cancelAnnotation(r2, "Deploy"))
110 c.waitJobs("the third run is pending", r3, allStatus(pb.CiStatus_CI_STATUS_PENDING))
111 assert.NotEqual(t, pb.CiStatus_CI_STATUS_COMPLETED, c.getRun(r1.GetId()).GetRun().GetStatus(), "the running run is left alone")
112
113 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
114 job3 := runner.await()
115 assert.Equal(t, r3.GetId(), job3.GetRunId(), "the group goes to the waiting run")
116 runner.runJob(job3, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
117 c.waitFor("the third run completes", func() bool {
118 return c.getRun(r3.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
119 })
120
121 // The group is free: a new run takes it at once.
122 r4 := c.pushN(4, workflowAt(wfGroup, ""))
123 job4 := runner.await()
124 assert.Equal(t, r4.GetId(), job4.GetRunId())
125}
126
127// AC4: cancel-in-progress cancels the running run, whose runner is told through
128// the cancel mechanism of a user cancel; an expression decides it, and groups
129// are compared without regard to case.
130func TestCIWorkflowConcurrencyCancelInProgress(t *testing.T) {
131 for _, tt := range []struct{ name, cancel string }{
132 {"true", "true"},
133 {"expression", "${{ github.ref == 'refs/heads/main' }}"},
134 } {
135 t.Run(tt.name, func(t *testing.T) {
136 c, runner := concurrencyHarness(t)
137 r1 := c.pushN(1, workflowAt("concurrency:\n group: Release\n cancel-in-progress: "+tt.cancel+"\n", ""))
138 job1 := runner.await()
139 watch, err := runner.jobAPI(job1).runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: job1.GetJobId()}))
140 require.NoError(t, err)
141
142 r2 := c.pushN(2, workflowAt("concurrency:\n group: RELEASE\n cancel-in-progress: "+tt.cancel+"\n", ""))
143 var cancel *pb.CancelRequest
144 for watch.Receive() {
145 if m := watch.Msg().GetCancel(); m != nil {
146 cancel = m
147 break
148 }
149 }
150 require.NotNil(t, cancel, "the running job is asked to stop")
151 assert.Equal(t, pb.CancelReason_CANCEL_REASON_USER, cancel.GetReason())
152 assert.Contains(t, cancel.GetMessage(), "higher priority waiting request for 'RELEASE'")
153
154 _, err = runner.jobAPI(job1).runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
155 JobId: job1.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED,
156 }))
157 require.NoError(t, err)
158 c.waitFor("the first run is cancelled", func() bool { return c.conclusionOf(r1) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
159 job2 := runner.await()
160 assert.Equal(t, r2.GetId(), job2.GetRunId(), "the waiting run takes the group")
161 })
162 }
163 t.Run("false expression only queues", func(t *testing.T) {
164 c, runner := concurrencyHarness(t)
165 top := "concurrency:\n group: G\n cancel-in-progress: ${{ github.ref == 'refs/heads/other' }}\n"
166 c.pushN(1, workflowAt(top, ""))
167 job1 := runner.await()
168 r2 := c.pushN(2, workflowAt(top, ""))
169 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
170 runner.heartbeat(job1)
171 time.Sleep(200 * time.Millisecond)
172 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(c.runs()[1]))
173 })
174}
175
176const matrixJob = ` strategy:
177 matrix:
178 n: [1, 2]
179 concurrency:
180 group: shared-${{ 'X' }}
181`
182
183// AC4: the legs of a matrix share a job-level group: one runs, one waits, and
184// a leg is freed when the other ends.
185func TestCIJobConcurrencyMatrixLegsShareAGroup(t *testing.T) {
186 c, runner := concurrencyHarness(t)
187 r := c.pushN(1, workflowAt("", matrixJob))
188 job1 := runner.await()
189 c.waitJobs("one leg is pending", r, func(js []*pb.RunJob) bool {
190 pending, other := 0, 0
191 for _, j := range js {
192 switch j.GetStatus() {
193 case pb.CiStatus_CI_STATUS_PENDING:
194 pending++
195 default:
196 other++
197 }
198 }
199 return len(js) == 2 && pending == 1 && other == 1
200 })
201 j, err := runner.fetch("")
202 require.NoError(t, err)
203 assert.Nil(t, j, "the pending leg is not assignable")
204 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
205 job2 := runner.await()
206 assert.NotEqual(t, job1.GetJobId(), job2.GetJobId())
207 runner.runJob(job2, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
208 c.waitFor("the run completes", func() bool { return c.getRun(r.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
209 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, c.conclusionOf(r))
210}
211
212// AC4: a job-level group is shared between runs, case-insensitively; the third
213// waiting job replaces the second, which ends cancelled with the annotation.
214func TestCIJobConcurrencyAcrossRuns(t *testing.T) {
215 c, runner := concurrencyHarness(t)
216 group := func(g string) string { return " concurrency: " + g + "\n" }
217 r1 := c.pushN(1, workflowAt("", group("Build-Lock")))
218 job1 := runner.await()
219 r2 := c.pushN(2, workflowAt("", group("build-lock")))
220 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
221 assert.Equal(t, "build-lock", c.jobsOf(r2)[0].GetConcurrencyGroup())
222 r3 := c.pushN(3, workflowAt("", group("BUILD-LOCK")))
223 c.waitFor("the second job is cancelled", func() bool { return c.conclusionOf(r2) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
224 assert.True(t, c.cancelAnnotation(r2, "BUILD-LOCK"), "the group is named as the newer request wrote it")
225 c.waitJobs("pending", r3, allStatus(pb.CiStatus_CI_STATUS_PENDING))
226 _ = r1
227 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
228 job3 := runner.await()
229 assert.Equal(t, r3.GetId(), job3.GetRunId())
230}
231
232// AC4: the group is freed when its run is cancelled, and when a re-run joins
233// the group again it waits behind the run that took it meanwhile.
234func TestCIConcurrencyFreedByCancelAndRerunRejoins(t *testing.T) {
235 c, runner := concurrencyHarness(t)
236 r1 := c.pushN(1, workflowAt(wfGroup, ""))
237 job1 := runner.await()
238 _, err := c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: r1.GetId()}))
239 require.NoError(t, err)
240 runner.heartbeat(job1)
241 c.waitFor("the cancelled run completes", func() bool { return c.getRun(r1.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
242
243 r2 := c.pushN(2, workflowAt(wfGroup, ""))
244 job2 := runner.await()
245 assert.Equal(t, r2.GetId(), job2.GetRunId(), "a cancelled run freed the group")
246
247 _, err = c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: r1.GetId()}))
248 require.NoError(t, err)
249 c.waitJobs("the re-run waits for the group", r1, allStatus(pb.CiStatus_CI_STATUS_PENDING))
250 runner.runJob(job2, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
251 job1b := runner.await()
252 assert.Equal(t, r1.GetId(), job1b.GetRunId(), "the re-run takes the group once it is free")
253}
254
255// AC4: a job lost to a missing heartbeat frees its group.
256func TestCIConcurrencyFreedByLostJob(t *testing.T) {
257 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
258 cfg.HeartbeatDeadline, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
259 })
260 runner := c.registerRunner("alpha", "ubuntu-latest")
261 r1 := c.pushN(1, workflowAt(wfGroup, ""))
262 job1 := runner.await()
263 runner.heartbeat(job1)
264 r2 := c.pushN(2, workflowAt(wfGroup, ""))
265 c.waitFor("the lost job fails", func() bool { return c.conclusionOf(r1) == pb.CiConclusion_CI_CONCLUSION_FAILURE })
266 job2 := runner.await()
267 assert.Equal(t, r2.GetId(), job2.GetRunId())
268}
269
270// A group that evaluates to nothing fails the job as GitHub does, and so does
271// queue: max with cancel-in-progress, literal or evaluated to true.
272func TestCIConcurrencyErrors(t *testing.T) {
273 c, runner := concurrencyHarness(t)
274 _ = runner
275 r1 := c.pushN(1, workflowAt("", " concurrency:\n group: ${{ vars.NOPE }}\n"))
276 c.waitJobs("failed", r1, allStatus(pb.CiStatus_CI_STATUS_COMPLETED))
277 assert.Contains(t, c.jobsOf(r1)[0].GetError(), "concurrency group name cannot be empty")
278
279 // The literal pair is a workflow validation error (GitHub's docs): the run
280 // fails with no job, like an invalid workflow file, at job and workflow level.
281 r3 := c.pushN(3, workflowAt("", " concurrency:\n group: g\n queue: max\n cancel-in-progress: true\n"))
282 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, r3.GetConclusion())
283 assert.Empty(t, c.jobsOf(r3))
284 assert.Equal(t, ".github/workflows/ci.yml (Line: 12, Col: 27): 'queue: max' cannot be combined with 'cancel-in-progress: true'.", c.getRun(r3.GetId()).GetRun().GetError())
285 r4 := c.pushN(4, workflowAt("concurrency: {group: g, queue: max, cancel-in-progress: true}\n", ""))
286 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, r4.GetConclusion())
287 assert.Contains(t, c.getRun(r4.GetId()).GetRun().GetError(), "'queue: max' cannot be combined with 'cancel-in-progress: true'.")
288
289 r2 := c.pushN(2, workflowAt("", " concurrency:\n group: g\n queue: max\n cancel-in-progress: ${{ github.ref == 'refs/heads/main' }}\n"))
290 c.waitJobs("failed", r2, allStatus(pb.CiStatus_CI_STATUS_COMPLETED))
291 assert.Equal(t, "'queue: max' cannot be combined with 'cancel-in-progress: true'.", strings.TrimSpace(c.jobsOf(r2)[0].GetError()))
292}
293
294// holdGroup writes a group behind the use case's back, as a crash would leave
295// it: the entries are named and marked whatever the state of their runs.
296func (c *ciHarness) holdGroup(group string, running *domain.CIConcurrencyEntry, pending ...domain.CIConcurrencyEntry) {
297 c.t.Helper()
298 require.NoError(c.t, c.forge.store.CI().UpdateConcurrency(c.ctx, c.h("proj"), group, func(st *domain.CIConcurrencyGroupState) error {
299 st.Record.Running, st.Record.Pending = running, pending
300 st.Marks = map[string]domain.CIConcurrencyMark{}
301 if running != nil {
302 st.Marks[running.Key()] = domain.CIConcurrencyMark{Entry: *running, Group: group, State: domain.CIConcurrencyHolding}
303 }
304 for _, p := range pending {
305 st.Marks[p.Key()] = domain.CIConcurrencyMark{Entry: p, Group: group, State: domain.CIConcurrencyPending}
306 }
307 return nil
308 }))
309}
310
311func (c *ciHarness) groupState(group string) (st domain.CIConcurrencyGroupState) {
312 c.t.Helper()
313 require.NoError(c.t, c.forge.store.CI().UpdateConcurrency(c.ctx, c.h("proj"), group, func(s *domain.CIConcurrencyGroupState) error {
314 st = *s
315 return nil
316 }))
317 return st
318}
319
320func (c *ciHarness) groupRecord(group string) domain.CIConcurrencyRecord {
321 return c.groupState(group).Record
322}
323
324// markState is the mark of an entry in a group, "" when it has none.
325func (c *ciHarness) markState(group string, e domain.CIConcurrencyEntry) domain.CIConcurrencyState {
326 return c.groupState(group).Marks[e.Key()].State
327}
328
329// A crash between the write of an ended job and the release of its group leaves
330// the record holding it: the next run to ask finds the entry dead and takes the
331// group.
332func TestCIConcurrencyHealsAfterACrash(t *testing.T) {
333 c, runner := concurrencyHarness(t)
334 r1 := c.pushN(1, workflowAt(wfGroup, ""))
335 runner.runJob(runner.await(), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
336 c.waitFor("the first run completes", func() bool { return c.getRun(r1.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
337 c.waitFor("the group is released", func() bool { return c.groupRecord("Deploy").IsEmpty() })
338 c.holdGroup("Deploy", &domain.CIConcurrencyEntry{RunID: r1.GetId(), Attempt: 1}) // the release that did not happen
339
340 r2 := c.pushN(2, workflowAt(wfGroup, ""))
341 job2 := runner.await()
342 assert.Equal(t, r2.GetId(), job2.GetRunId(), "the dead entry does not hold the group")
343 rec := c.groupRecord("Deploy")
344 require.NotNil(t, rec.Running)
345 assert.Equal(t, r2.GetId(), rec.Running.RunID)
346}
347
348// A run that is deleted (the retention sweep, not its repository) frees the
349// group it held and the one it waited in.
350func TestCIConcurrencyHealsAfterRunDeletion(t *testing.T) {
351 c, runner := concurrencyHarness(t)
352 r1 := c.pushN(1, workflowAt(wfGroup, ""))
353 runner.runJob(runner.await(), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
354 c.waitFor("completes", func() bool { return c.getRun(r1.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
355 c.waitFor("released", func() bool { return c.groupRecord("Deploy").IsEmpty() })
356 ghost := &domain.CIConcurrencyEntry{RunID: r1.GetId(), Attempt: 1}
357 c.holdGroup("Deploy", ghost)
358 ids, err := c.forge.store.CI().Sweep(c.ctx, time.Now().Add(time.Hour))
359 require.NoError(t, err)
360 require.Contains(t, ids, r1.GetId(), "the run is deleted")
361
362 r2 := c.pushN(2, workflowAt(wfGroup, ""))
363 job2 := runner.await()
364 assert.Equal(t, r2.GetId(), job2.GetRunId(), "a deleted run does not hold the group")
365}
366
367// flakyStore fails the writes that match, as many times as fails says. The
368// failure is returned before anything is written.
369type flakyStore struct {
370 ports.CIStore
371 match func(ports.CIWrite) bool
372 fails atomic.Int32
373 hits atomic.Int32
374}
375
376func (f *flakyStore) Apply(ctx context.Context, w ports.CIWrite) (uint64, error) {
377 if f.match != nil && f.match(w) && f.fails.Add(-1) >= 0 {
378 f.hits.Add(1)
379 return 0, domain.ErrConflict
380 }
381 return f.CIStore.Apply(ctx, w)
382}
383
384// flakyConcurrencyHarness is a harness whose store fails the writes fs.match
385// selects. retries is CIConfig.ConcurrencyRetries: 0 keeps the default, a
386// negative number leaves the recovery of a lost delivery to the reconciliation.
387func flakyConcurrencyHarness(t *testing.T, retries int) (*ciHarness, *fakeRunner, *flakyStore) {
388 t.Helper()
389 fs := &flakyStore{}
390 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
391 cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
392 cfg.ConcurrencyRetries = retries
393 fs.CIStore, cfg.Store = cfg.Store, fs
394 })
395 return c, c.registerRunner("alpha", "ubuntu-latest"), fs
396}
397
398// endsCancelled matches the write that ends a job of the run as cancelled.
399func endsCancelled(runID int64) func(ports.CIWrite) bool {
400 return func(w ports.CIWrite) bool {
401 for _, j := range w.Jobs {
402 if j.RunID == runID && j.Ended() && j.Conclusion == domain.CICancelled {
403 return true
404 }
405 }
406 return false
407 }
408}
409
410// A replaced pending run whose cancel write fails once is cancelled by the
411// retry.
412func TestCIConcurrencyCancelSurvivesAFailedDelivery(t *testing.T) {
413 c, runner, fs := flakyConcurrencyHarness(t, 0)
414 c.pushN(1, workflowAt(wfGroup, ""))
415 job1 := runner.await()
416 r2 := c.pushN(2, workflowAt(wfGroup, ""))
417 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
418
419 fs.match = endsCancelled(r2.GetId())
420 fs.fails.Store(1)
421 r3 := c.pushN(3, workflowAt(wfGroup, ""))
422 c.waitFor("the first delivery failed", func() bool { return fs.hits.Load() == 1 })
423 rec := c.groupRecord("Deploy")
424 require.Len(t, rec.Pending, 1)
425 assert.Equal(t, r3.GetId(), rec.Pending[0].RunID)
426 c.waitFor("the replaced run is cancelled by the retry", func() bool { return c.conclusionOf(r2) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
427 assert.True(t, c.cancelAnnotation(r2, "Deploy"))
428 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
429 assert.Equal(t, r3.GetId(), runner.await().GetRunId())
430}
431
432// Nothing is bounded: more replacements than any list could hold, every cancel
433// write failing, and after the store works again the reconciliation discovers
434// each victim through its "replaced" mark in the group, and cancels it.
435func TestCIConcurrencyReconcileCancelsEveryReplacedVictim(t *testing.T) {
436 const replaced = 35
437 c, runner, fs := flakyConcurrencyHarness(t, -1)
438 c.pushN(1, workflowAt(wfGroup, ""))
439 runner.await()
440
441 // Each new run replaces the one that waited before it; no cancel lands.
442 fs.match = func(w ports.CIWrite) bool {
443 for _, j := range w.Jobs {
444 if j.Ended() && j.Conclusion == domain.CICancelled {
445 return true
446 }
447 }
448 return false
449 }
450 fs.fails.Store(1 << 20)
451 var victims []*pb.Run
452 prev := c.pushN(2, workflowAt(wfGroup, ""))
453 c.waitJobs("pending", prev, allStatus(pb.CiStatus_CI_STATUS_PENDING))
454 for i := range replaced {
455 next := c.pushN(3+i, workflowAt(wfGroup, ""))
456 c.waitFor("the newer run takes the waiting place", func() bool {
457 rec := c.groupRecord("Deploy")
458 return len(rec.Pending) == 1 && rec.Pending[0].RunID == next.GetId()
459 })
460 c.waitJobs("the newer run is pending", next, allStatus(pb.CiStatus_CI_STATUS_PENDING))
461 victims = append(victims, prev)
462 prev = next
463 }
464 for _, v := range victims {
465 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(v), "no cancel landed yet")
466 }
467 // The test has no in-process retry (ConcurrencyRetries is negative): only
468 // the reconciliation can recover the victims.
469 fs.fails.Store(0)
470 c.forge.CI.ReconcileConcurrency(c.ctx)
471 for _, v := range victims {
472 c.waitFor("a replaced run is cancelled", func() bool { return c.conclusionOf(v) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
473 assert.True(t, c.cancelAnnotation(v, "Deploy"))
474 }
475 c.waitJobs("the last one still waits", prev, allStatus(pb.CiStatus_CI_STATUS_PENDING))
476 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(prev))
477}
478
479// A running entry that cancel-in-progress replaced stays named with Canceling
480// set; when its cancel is lost, the reconciliation asks again.
481func TestCIConcurrencyReconcileAsksARunningVictimAgain(t *testing.T) {
482 c, runner, fs := flakyConcurrencyHarness(t, -1)
483 top := "concurrency:\n group: Deploy\n cancel-in-progress: true\n"
484 r1 := c.pushN(1, workflowAt(top, ""))
485 job1 := runner.await()
486 watch, err := runner.jobAPI(job1).runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: job1.GetJobId()}))
487 require.NoError(t, err)
488
489 // The cancel of the running run cannot be written.
490 fs.match = func(w ports.CIWrite) bool {
491 for _, r := range w.Runs {
492 if r.ID == r1.GetId() && r.CancelRequested {
493 return true
494 }
495 }
496 return false
497 }
498 fs.fails.Store(1 << 20)
499 c.pushN(2, workflowAt(top, ""))
500 c.waitFor("the cancel fails", func() bool { return fs.hits.Load() >= 1 })
501 rec := c.groupRecord("Deploy")
502 require.NotNil(t, rec.Running)
503 assert.True(t, rec.Running.Canceling, "the victim stays named, marked")
504 assert.Equal(t, r1.GetId(), rec.Running.RunID)
505
506 fs.fails.Store(0)
507 c.forge.CI.ReconcileConcurrency(c.ctx)
508 var cancel *pb.CancelRequest
509 for watch.Receive() {
510 if m := watch.Msg().GetCancel(); m != nil {
511 cancel = m
512 break
513 }
514 }
515 require.NotNil(t, cancel, "the runner is told after the reconciliation")
516 // The runner stops: its watch ends with the job.
517 _, err = runner.jobAPI(job1).runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
518 JobId: job1.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED,
519 }))
520 require.NoError(t, err)
521}
522
523// Crash between the write of the group and the write of the entry: the group
524// names a pending run whose own record does not say it waits yet. It is not
525// stale and not cancelled; the reconciliation plans it, which marks it
526// pending, and the group goes on to it when the holder ends.
527func TestCIConcurrencyCrashBetweenGroupAndEntry(t *testing.T) {
528 c, runner := concurrencyHarness(t)
529 c.pushN(1, workflowAt(wfGroup, ""))
530 job1 := runner.await()
531
532 // Stop the planning of the second run from marking it, as a crash would:
533 // name it in the group by hand, behind the first.
534 store := c.forge.store.CI()
535 r2 := c.pushN(2, workflowAt(wfGroup, ""))
536 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
537 run, err := store.GetRun(c.ctx, c.h("proj"), r2.GetId())
538 require.NoError(t, err)
539 run.Concurrency.State = domain.CIConcurrencyIdle
540 jobs, err := store.ListJobs(c.ctx, r2.GetId(), 1)
541 require.NoError(t, err)
542 for i := range jobs {
543 jobs[i].Concurrency.State = domain.CIConcurrencyIdle
544 }
545 _, err = store.Apply(c.ctx, ports.CIWrite{Runs: []domain.CIRun{run}, Jobs: jobs})
546 require.NoError(t, err)
547 rec := c.groupRecord("Deploy")
548 require.Len(t, rec.Pending, 1)
549 require.Equal(t, r2.GetId(), rec.Pending[0].RunID, "the group names the run")
550 assert.Equal(t, domain.CIConcurrencyPending, c.markState("Deploy", rec.Pending[0]), "and marks it, in the same batch")
551
552 c.forge.CI.ReconcileConcurrency(c.ctx)
553 c.waitFor("the entry is marked pending again", func() bool {
554 r, err := store.GetRun(c.ctx, c.h("proj"), r2.GetId())
555 return err == nil && r.Concurrency.State == domain.CIConcurrencyPending
556 })
557 time.Sleep(200 * time.Millisecond)
558 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(r2), "a named entry is not a replaced one")
559 rec = c.groupRecord("Deploy")
560 require.Len(t, rec.Pending, 1)
561 assert.Equal(t, r2.GetId(), rec.Pending[0].RunID)
562
563 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
564 assert.Equal(t, r2.GetId(), runner.await().GetRunId())
565}
566
567// holdsGroup matches the write that tells a run it holds its group.
568func holdsGroup(runID int64) func(ports.CIWrite) bool {
569 return func(w ports.CIWrite) bool {
570 for _, r := range w.Runs {
571 if r.ID == runID && r.Concurrency.State == domain.CIConcurrencyHolding {
572 return true
573 }
574 }
575 return false
576 }
577}
578
579// HIGH-2: a promotion whose planning fails once is tried again.
580func TestCIConcurrencyPromotionIsRetried(t *testing.T) {
581 c, runner, fs := flakyConcurrencyHarness(t, 0)
582 c.pushN(1, workflowAt(wfGroup, ""))
583 job1 := runner.await()
584 r2 := c.pushN(2, workflowAt(wfGroup, ""))
585 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
586
587 fs.match = holdsGroup(r2.GetId())
588 fs.fails.Store(1)
589 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
590 c.waitFor("the first planning failed", func() bool { return fs.hits.Load() == 1 })
591 job2 := runner.await()
592 assert.Equal(t, r2.GetId(), job2.GetRunId(), "the retry planned the promoted run")
593}
594
595// HIGH-2: a holder that was never told it holds the group is planned by the
596// reconciliation, whatever happened to the retries.
597func TestCIConcurrencyReconcilePlansAStaleHolder(t *testing.T) {
598 c, runner, fs := flakyConcurrencyHarness(t, -1)
599 c.pushN(1, workflowAt(wfGroup, ""))
600 job1 := runner.await()
601 r2 := c.pushN(2, workflowAt(wfGroup, ""))
602 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
603
604 fs.match = holdsGroup(r2.GetId())
605 fs.fails.Store(1000)
606 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
607 c.waitFor("the planning fails", func() bool { return fs.hits.Load() >= 1 })
608 rec := c.groupRecord("Deploy")
609 require.NotNil(t, rec.Running)
610 assert.Equal(t, r2.GetId(), rec.Running.RunID, "the group names the promoted run")
611 c.waitJobs("which still shows pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
612
613 fs.fails.Store(0)
614 c.forge.CI.ReconcileConcurrency(c.ctx)
615 job2 := runner.await()
616 assert.Equal(t, r2.GetId(), job2.GetRunId())
617}
618
619const siblingJobs = ` steps:
620 - run: echo grouped
621 sibling:
622 runs-on: ubuntu-latest
623 steps:
624 - run: echo sibling
625`
626
627// jobWorkflow is a workflow of a grouped job with the given concurrency lines
628// and an unrelated sibling.
629func jobWorkflow(concurrency string) string {
630 return "name: ci\non:\n push:\n branches: [main]\njobs:\n grouped:\n runs-on: ubuntu-latest\n" +
631 concurrency + siblingJobs
632}
633
634// awaitKey fetches jobs until the one of the run with the key comes.
635func (r *fakeRunner) awaitKey(runID int64, key string) *pb.Job {
636 r.c.t.Helper()
637 for range 20 {
638 if j := r.await(); j.GetRunId() == runID && j.GetJobKey() == key {
639 return j
640 }
641 }
642 r.c.t.Fatalf("no job %s of run %d", key, runID)
643 return nil
644}
645
646// T-2: a job-level cancel-in-progress (true, or an expression) cancels the
647// running job of the group only: its sibling goes on, and the replacement
648// becomes assignable.
649func TestCIJobConcurrencyCancelInProgress(t *testing.T) {
650 for _, tt := range []struct{ name, cancel string }{
651 {"true", "true"},
652 {"expression", "${{ github.ref == 'refs/heads/main' }}"},
653 } {
654 t.Run(tt.name, func(t *testing.T) {
655 c, runner := concurrencyHarness(t)
656 wf := jobWorkflow(" concurrency:\n group: jobgroup\n cancel-in-progress: " + tt.cancel + "\n")
657 r1 := c.pushN(1, wf)
658 grouped1, sibling1 := runner.awaitKey(r1.GetId(), "grouped"), runner.awaitKey(r1.GetId(), "sibling")
659 watch, err := runner.jobAPI(grouped1).runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: grouped1.GetJobId()}))
660 require.NoError(t, err)
661
662 r2 := c.pushN(2, wf)
663 var cancel *pb.CancelRequest
664 for watch.Receive() {
665 if m := watch.Msg().GetCancel(); m != nil {
666 cancel = m
667 break
668 }
669 }
670 require.NotNil(t, cancel, "the grouped job is asked to stop")
671 assert.Contains(t, cancel.GetMessage(), "higher priority waiting request for 'jobgroup'")
672
673 // The sibling of the first run was not touched.
674 runner.heartbeat(sibling1)
675 for _, j := range c.jobsOf(r1) {
676 if j.GetKey() == "sibling" {
677 assert.NotEqual(t, pb.CiStatus_CI_STATUS_COMPLETED, j.GetStatus())
678 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_UNSPECIFIED, j.GetConclusion())
679 }
680 }
681 _, err = runner.jobAPI(grouped1).runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
682 JobId: grouped1.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED,
683 }))
684 require.NoError(t, err)
685 runner.runJob(sibling1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
686
687 grouped2 := runner.awaitKey(r2.GetId(), "grouped")
688 assert.Equal(t, r2.GetId(), grouped2.GetRunId(), "the replacement is assignable")
689 c.waitFor("the first run is over", func() bool { return c.getRun(r1.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
690 for _, j := range c.jobsOf(r1) {
691 want := pb.CiConclusion_CI_CONCLUSION_CANCELLED
692 if j.GetKey() == "sibling" {
693 want = pb.CiConclusion_CI_CONCLUSION_SUCCESS
694 }
695 assert.Equal(t, want, j.GetConclusion(), j.GetKey())
696 }
697 })
698 }
699}
700
701// T-3: the server restarts: the groups are repaired at startup. A group held by
702// an ended run is freed and the run behind it is planned; a group whose only
703// run is over is removed.
704func TestCIConcurrencyReconcileAtRestart(t *testing.T) {
705 c, runner := concurrencyHarness(t)
706 r1 := c.pushN(1, workflowAt(wfGroup, ""))
707 runner.await()
708 r2 := c.pushN(2, workflowAt(wfGroup, ""))
709 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
710
711 // The first run ends in the store without the release (the crash), and a
712 // second group is held by a run that is over.
713 store := c.forge.store.CI()
714 jobs, err := store.ListJobs(c.ctx, r1.GetId(), 1)
715 require.NoError(t, err)
716 run, err := store.GetRun(c.ctx, c.h("proj"), r1.GetId())
717 require.NoError(t, err)
718 jobs[0].Phase, jobs[0].Conclusion = domain.CIPhaseEnded, domain.CISuccess
719 run.Status, run.Conclusion, run.CompletedAt = domain.CICompleted, domain.CISuccess, time.Now()
720 _, err = store.Apply(c.ctx, ports.CIWrite{Jobs: jobs, Runs: []domain.CIRun{run}})
721 require.NoError(t, err)
722 c.holdGroup("Stale", &domain.CIConcurrencyEntry{RunID: r1.GetId(), Attempt: 1})
723 rec := c.groupRecord("Deploy")
724 require.NotNil(t, rec.Running)
725 require.Equal(t, r1.GetId(), rec.Running.RunID, "the record still holds the ended run")
726
727 // Restart on the same data.
728 cfg, hookExe := c.forge.cfg, binary(t)
729 c.srv.Close()
730 c.forge.Close(c.ctx)
731 f2, err := newForge(c.ctx, cfg, options{hookExe: hookExe, ci: func(cc *usecase.CIConfig) {
732 cc.CancelGrace, cc.Tick, cc.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
733 }})
734 require.NoError(t, err)
735 t.Cleanup(func() { f2.Close(context.WithoutCancel(c.ctx)) })
736 srv := httptest.NewServer(f2.server.Handler())
737 t.Cleanup(srv.Close)
738
739 var got *pb.Job
740 c.waitFor("the waiting run is planned by the startup repair", func() bool {
741 res, err := newClients(srv.URL+"/-/api", runner.secret).runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{}))
742 if err == nil && res.Msg.GetJob() != nil {
743 got = res.Msg.GetJob()
744 }
745 return got != nil
746 })
747 assert.Equal(t, r2.GetId(), got.GetRunId())
748 groups, err := f2.store.CI().ListConcurrency(c.ctx)
749 require.NoError(t, err)
750 require.Len(t, groups, 1, "the stale group is gone")
751 assert.Equal(t, "DEPLOY", groups[0].Group)
752 require.NotNil(t, groups[0].State.Record.Running)
753 assert.Equal(t, r2.GetId(), groups[0].State.Record.Running.RunID)
754 assert.Empty(t, groups[0].State.Record.Pending)
755}
756
757// Both writes of a run's own record fail when it queues (its "pending" state),
758// and the cancel of the run fails for good: the group alone knows B. A newer run
759// replaces B; B's mark is "replaced", written in the batch that unnamed it. After
760// the store works again the reconciliation finds B from that mark and cancels it.
761func TestCIConcurrencyReplacedRunWithLostWritesIsCancelled(t *testing.T) {
762 c, runner, fs := flakyConcurrencyHarness(t, -1)
763 r1 := c.pushN(1, workflowAt(wfGroup, ""))
764 runner.await()
765
766 // B queues, but no run after the first gets the pending state in its own
767 // record, and no cancel lands.
768 fs.match = func(w ports.CIWrite) bool {
769 for _, r := range w.Runs {
770 if r.ID > r1.GetId() && r.Concurrency.State == domain.CIConcurrencyPending {
771 return true
772 }
773 }
774 for _, j := range w.Jobs {
775 if j.RunID > r1.GetId() && j.Ended() && j.Conclusion == domain.CICancelled {
776 return true
777 }
778 }
779 return false
780 }
781 fs.fails.Store(1 << 20)
782 b := c.pushN(2, workflowAt(wfGroup, ""))
783 c.waitFor("B is named by the group", func() bool {
784 rec := c.groupRecord("Deploy")
785 return len(rec.Pending) == 1 && rec.Pending[0].RunID == b.GetId()
786 })
787 bEntry := domain.CIConcurrencyEntry{RunID: b.GetId(), Attempt: 1}
788 assert.Equal(t, domain.CIConcurrencyPending, c.markState("Deploy", bEntry), "the group batch marked it")
789
790 // C replaces B: B is unnamed and marked replaced in one batch.
791 c.pushN(3, workflowAt(wfGroup, ""))
792 c.waitFor("B is marked replaced", func() bool { return c.markState("Deploy", bEntry) == domain.CIConcurrencyReplaced })
793 rec := c.groupRecord("Deploy")
794 require.Len(t, rec.Pending, 1)
795 assert.NotEqual(t, b.GetId(), rec.Pending[0].RunID, "B is no longer named")
796 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(b))
797
798 fs.fails.Store(0)
799 c.forge.CI.ReconcileConcurrency(c.ctx)
800 c.waitFor("B ends cancelled", func() bool { return c.conclusionOf(b) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
801 assert.True(t, c.cancelAnnotation(b, "Deploy"))
802 c.waitFor("its mark goes with it", func() bool { return c.markState("Deploy", bEntry) == "" })
803}
804
805// As above, but only B's own writes fail: C's record knows it is pending, so
806// the reconciliation does not plan C again, and no other decision on the group
807// prunes B's mark. The cancel that ends B frees the group that decided it,
808// although B's own record names no group.
809func TestCIConcurrencyCancelOfARunUnawareOfItsGroupDropsItsMark(t *testing.T) {
810 c, runner, fs := flakyConcurrencyHarness(t, -1)
811 r1 := c.pushN(1, workflowAt(wfGroup, ""))
812 runner.await()
813
814 bID := r1.GetId() + 1 // run ids are taken in sequence
815 fs.match = func(w ports.CIWrite) bool {
816 for _, r := range w.Runs {
817 if r.ID == bID && r.Concurrency.State == domain.CIConcurrencyPending {
818 return true
819 }
820 }
821 for _, j := range w.Jobs {
822 if j.RunID == bID && j.Ended() && j.Conclusion == domain.CICancelled {
823 return true
824 }
825 }
826 return false
827 }
828 fs.fails.Store(1 << 20)
829 b := c.pushN(2, workflowAt(wfGroup, ""))
830 require.Equal(t, bID, b.GetId())
831 bEntry := domain.CIConcurrencyEntry{RunID: bID, Attempt: 1}
832 c.waitFor("B is named by the group", func() bool { return c.markState("Deploy", bEntry) == domain.CIConcurrencyPending })
833
834 r3 := c.pushN(3, workflowAt(wfGroup, ""))
835 c.waitFor("B is marked replaced", func() bool { return c.markState("Deploy", bEntry) == domain.CIConcurrencyReplaced })
836 c.waitJobs("C knows it waits", r3, func(js []*pb.RunJob) bool {
837 return len(js) == 1 && js[0].GetStatus() == pb.CiStatus_CI_STATUS_PENDING
838 })
839 run, err := c.forge.store.CI().GetRun(c.ctx, c.h("proj"), bID)
840 require.NoError(t, err)
841 require.Empty(t, run.Concurrency.Group, "B's own record names no group")
842
843 fs.fails.Store(0)
844 c.forge.CI.ReconcileConcurrency(c.ctx)
845 c.waitFor("B ends cancelled", func() bool { return c.conclusionOf(b) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
846 c.waitFor("its mark goes with it", func() bool { return c.markState("Deploy", bEntry) == "" })
847 rec := c.groupRecord("Deploy")
848 require.Len(t, rec.Pending, 1)
849 assert.Equal(t, r3.GetId(), rec.Pending[0].RunID, "C still waits")
850}
851
852// A grouped job B is replaced by C, the cancel never lands, C is promoted when
853// the running job ends, and an unrelated sibling of B completes and advances B:
854// B's grouped job must not take the free place, it ends cancelled and never runs.
855func TestCIJobConcurrencyReplacedJobNeverAcquiresAgain(t *testing.T) {
856 c, runner, fs := flakyConcurrencyHarness(t, -1)
857 wf := jobWorkflow(" concurrency: jobgroup\n")
858 r1 := c.pushN(1, wf)
859 grouped1, sibling1 := runner.awaitKey(r1.GetId(), "grouped"), runner.awaitKey(r1.GetId(), "sibling")
860
861 rB := c.pushN(2, wf)
862 siblingB := runner.awaitKey(rB.GetId(), "sibling") // B's unrelated job is running
863 c.waitJobs("B's grouped job waits", rB, func(js []*pb.RunJob) bool {
864 for _, j := range js {
865 if j.GetKey() == "grouped" {
866 return j.GetStatus() == pb.CiStatus_CI_STATUS_PENDING
867 }
868 }
869 return false
870 })
871
872 fs.match = endsCancelled(rB.GetId())
873 fs.fails.Store(1 << 20)
874 rC := c.pushN(3, wf)
875 bEntry := domain.CIConcurrencyEntry{RunID: rB.GetId(), Attempt: 1, JobID: domain.NewJobID(rB.GetId(), 0, 0)}
876 c.waitFor("B's grouped job is replaced", func() bool { return c.markState("jobgroup", bEntry) == domain.CIConcurrencyReplaced })
877
878 // The running job ends: C's grouped job is promoted.
879 runner.runJob(grouped1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
880 runner.runJob(sibling1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
881 groupedC := runner.awaitKey(rC.GetId(), "grouped")
882 rec := c.groupRecord("jobgroup")
883 require.NotNil(t, rec.Running)
884 assert.Equal(t, rC.GetId(), rec.Running.RunID)
885
886 // B's sibling completes: B is advanced, its grouped job asks for the group.
887 runner.runJob(siblingB, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
888 c.waitFor("B was advanced", func() bool {
889 for _, j := range c.jobsOf(rB) {
890 if j.GetKey() == "sibling" {
891 return j.GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
892 }
893 }
894 return false
895 })
896 time.Sleep(300 * time.Millisecond)
897 rec = c.groupRecord("jobgroup")
898 require.NotNil(t, rec.Running)
899 assert.Equal(t, rC.GetId(), rec.Running.RunID, "C still holds the group")
900 if len(rec.Pending) > 0 {
901 assert.NotEqual(t, rB.GetId(), rec.Pending[0].RunID, "B did not take the waiting place back")
902 }
903 // Nothing of B's grouped job is given to a runner.
904 for range 5 {
905 j, err := runner.fetch("")
906 require.NoError(t, err)
907 if j != nil {
908 assert.False(t, j.GetRunId() == rB.GetId() && j.GetJobKey() == "grouped", "B's grouped job must not run")
909 }
910 }
911
912 // Once the store works, the reconciliation cancels it.
913 fs.fails.Store(0)
914 c.forge.CI.ReconcileConcurrency(c.ctx)
915 c.waitFor("B's grouped job ends cancelled", func() bool {
916 for _, j := range c.jobsOf(rB) {
917 if j.GetKey() == "grouped" {
918 return j.GetConclusion() == pb.CiConclusion_CI_CONCLUSION_CANCELLED
919 }
920 }
921 return false
922 })
923 runner.runJob(groupedC, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
924}
925
926// The same, with an empty group: B's grouped job is replaced and cancelled
927// before its replacement ends too; a replaced job never runs even when the
928// group is free.
929func TestCIJobConcurrencyReplacedJobDoesNotTakeAnEmptyGroup(t *testing.T) {
930 c, runner, fs := flakyConcurrencyHarness(t, -1)
931 wf := jobWorkflow(" concurrency: jobgroup\n")
932 r1 := c.pushN(1, wf)
933 grouped1, sibling1 := runner.awaitKey(r1.GetId(), "grouped"), runner.awaitKey(r1.GetId(), "sibling")
934 rB := c.pushN(2, wf)
935 siblingB := runner.awaitKey(rB.GetId(), "sibling")
936 c.waitJobs("pending", rB, func(js []*pb.RunJob) bool {
937 for _, j := range js {
938 if j.GetKey() == "grouped" {
939 return j.GetStatus() == pb.CiStatus_CI_STATUS_PENDING
940 }
941 }
942 return false
943 })
944 fs.match = endsCancelled(rB.GetId())
945 fs.fails.Store(1 << 20)
946 rC := c.pushN(3, wf)
947 bEntry := domain.CIConcurrencyEntry{RunID: rB.GetId(), Attempt: 1, JobID: domain.NewJobID(rB.GetId(), 0, 0)}
948 c.waitFor("replaced", func() bool { return c.markState("jobgroup", bEntry) == domain.CIConcurrencyReplaced })
949
950 // Both the running job and the promoted one finish: the group is empty.
951 runner.runJob(grouped1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
952 runner.runJob(sibling1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
953 runner.runJob(runner.awaitKey(rC.GetId(), "grouped"), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
954 c.waitFor("the group is free", func() bool { return c.groupRecord("jobgroup").Running == nil })
955
956 runner.runJob(siblingB, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil) // advances B
957 time.Sleep(300 * time.Millisecond)
958 assert.Nil(t, c.groupRecord("jobgroup").Running, "B's replaced job did not take the empty group")
959 for range 5 {
960 j, err := runner.fetch("")
961 require.NoError(t, err)
962 if j != nil {
963 assert.False(t, j.GetRunId() == rB.GetId() && j.GetJobKey() == "grouped", "B's grouped job must not run")
964 }
965 }
966 fs.fails.Store(0)
967 c.forge.CI.ReconcileConcurrency(c.ctx)
968 c.waitFor("it ends cancelled", func() bool {
969 for _, j := range c.jobsOf(rB) {
970 if j.GetKey() == "grouped" {
971 return j.GetConclusion() == pb.CiConclusion_CI_CONCLUSION_CANCELLED
972 }
973 }
974 return false
975 })
976}
977
978// Crash between the group batch and the write of a JOB's own state: the group
979// names (and marks) the grouped job as waiting, the job's record does not say so
980// yet. It is not replaced and not cancelled; reconciliation tells it, and it runs
981// when its turn comes.
982func TestCIJobConcurrencyCrashBetweenGroupAndJob(t *testing.T) {
983 c, runner := concurrencyHarness(t)
984 group := " concurrency: jobgroup\n"
985 c.pushN(1, workflowAt("", group))
986 job1 := runner.await()
987 r2 := c.pushN(2, workflowAt("", group))
988 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
989
990 store := c.forge.store.CI()
991 jobs, err := store.ListJobs(c.ctx, r2.GetId(), 1)
992 require.NoError(t, err)
993 require.Len(t, jobs, 1)
994 entry := domain.CIConcurrencyEntry{RunID: r2.GetId(), Attempt: 1, JobID: jobs[0].ID}
995 jobs[0].Concurrency.State = domain.CIConcurrencyIdle // the write that was lost
996 _, err = store.Apply(c.ctx, ports.CIWrite{Jobs: jobs})
997 require.NoError(t, err)
998 rec := c.groupRecord("jobgroup")
999 require.Len(t, rec.Pending, 1)
1000 assert.Equal(t, entry.JobID, rec.Pending[0].JobID, "the group names the job")
1001 assert.Equal(t, domain.CIConcurrencyPending, c.markState("jobgroup", entry), "and its mark says so")
1002
1003 c.forge.CI.ReconcileConcurrency(c.ctx)
1004 c.waitFor("the job is told it waits", func() bool {
1005 js, err := store.ListJobs(c.ctx, r2.GetId(), 1)
1006 return err == nil && js[0].Concurrency.State == domain.CIConcurrencyPending
1007 })
1008 time.Sleep(200 * time.Millisecond)
1009 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(r2), "a named job is not a replaced one")
1010 assert.Equal(t, domain.CIConcurrencyPending, c.markState("jobgroup", entry))
1011
1012 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1013 assert.Equal(t, r2.GetId(), runner.await().GetRunId(), "it runs when the group is free")
1014}
1015
1016// plannedWrite matches a write of a run after the first that stores planned jobs.
1017func plannedWrite(after int64) func(ports.CIWrite) bool {
1018 return func(w ports.CIWrite) bool {
1019 for _, j := range w.Jobs {
1020 if j.RunID > after && j.Planned {
1021 return true
1022 }
1023 }
1024 return false
1025 }
1026}
1027
1028const matrixGroupWorkflow = `name: b
1029on:
1030 push:
1031 branches: [other]
1032jobs:
1033 build:
1034 runs-on: ubuntu-latest
1035 strategy:
1036 matrix:
1037 n: [1, 2, 3]
1038 concurrency: shared
1039 steps:
1040 - run: echo ${{ matrix.n }}
1041`
1042
1043const holderWorkflow = `name: a
1044on:
1045 push:
1046 branches: [main]
1047jobs:
1048 hold:
1049 runs-on: ubuntu-latest
1050 concurrency: shared
1051 steps:
1052 - run: echo hold
1053`
1054
1055// Run B plans three legs of a matrix that share a group held by A. The legs
1056// replace one another (leg 0 by leg 1 by leg 2: only one waits), the group
1057// batches commit, and the write of B's planned jobs fails, leaving the
1058// placeholder of the job in the store. After the store recovers, the
1059// reconciliation must plan B before it cancels anything: only legs 0 and 1
1060// end cancelled, and leg 2, the legitimate waiting one, runs when A ends.
1061func TestCIJobConcurrencyRecoveryPlansTheMatrixBeforeCancellingALeg(t *testing.T) {
1062 c, runner, fs := flakyConcurrencyHarness(t, -1)
1063 a := c.push("main", map[string]string{".github/workflows/a.yml": holderWorkflow}, "")
1064 _ = a
1065 holder := runner.await()
1066 require.Equal(t, "hold", holder.GetJobKey())
1067 aRun := holder.GetRunId()
1068
1069 fs.match = plannedWrite(aRun)
1070 fs.fails.Store(1 << 20)
1071 c.push("other", map[string]string{".github/workflows/b.yml": matrixGroupWorkflow}, "")
1072 c.waitFor("B's run exists", func() bool {
1073 for _, r := range c.runs() {
1074 if r.GetId() > aRun {
1075 return true
1076 }
1077 }
1078 return false
1079 })
1080 var bRun *pb.Run
1081 for _, r := range c.runs() {
1082 if r.GetId() > aRun {
1083 bRun = r
1084 }
1085 }
1086 require.NotNil(t, bRun)
1087 g := domain.ConcurrencyKey("shared")
1088 c.waitFor("the group batches committed: two legs are marked replaced", func() bool {
1089 replaced := 0
1090 for _, m := range c.groupState(g).Marks {
1091 if m.Entry.RunID == bRun.GetId() && m.State == domain.CIConcurrencyReplaced {
1092 replaced++
1093 }
1094 }
1095 return replaced == 2
1096 })
1097 assert.Equal(t, 1, len(c.jobsOf(bRun)), "B's job is still the placeholder: its planned legs were never stored")
1098
1099 fs.fails.Store(0)
1100 c.forge.CI.ReconcileConcurrency(c.ctx)
1101 c.waitFor("B has its three legs", func() bool { return len(c.jobsOf(bRun)) == 3 })
1102 c.waitFor("legs 0 and 1 end cancelled", func() bool {
1103 n := 0
1104 for _, j := range c.jobsOf(bRun) {
1105 if j.GetConclusion() == pb.CiConclusion_CI_CONCLUSION_CANCELLED {
1106 n++
1107 }
1108 }
1109 return n == 2
1110 })
1111 byCombo := map[string]*pb.RunJob{}
1112 for _, j := range c.jobsOf(bRun) {
1113 byCombo[j.GetName()] = j
1114 }
1115 var waiting []*pb.RunJob
1116 for _, j := range c.jobsOf(bRun) {
1117 if j.GetConclusion() != pb.CiConclusion_CI_CONCLUSION_CANCELLED {
1118 waiting = append(waiting, j)
1119 }
1120 }
1121 require.Len(t, waiting, 1, "exactly one leg is not cancelled")
1122 assert.Equal(t, pb.CiStatus_CI_STATUS_PENDING, waiting[0].GetStatus(), "the final leg waits for A")
1123 assert.Contains(t, waiting[0].GetName(), "3", "and it is the last leg")
1124
1125 runner.runJob(holder, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1126 last := runner.await()
1127 assert.Equal(t, bRun.GetId(), last.GetRunId())
1128 assert.Equal(t, waiting[0].GetId(), last.GetJobId(), "the final leg runs when A ends")
1129 runner.runJob(last, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1130}
1131
1132// Stale marks (of runs that are gone) are pruned a bounded number per
1133// transaction: a decision on a group with a long backlog succeeds at once, and
1134// the reconciliation takes the others, a bounded batch at a time.
1135func TestCIConcurrencyPrunesStaleMarksInBoundedBatches(t *testing.T) {
1136 const stale, bound = 10, 3
1137 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
1138 cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
1139 cfg.ConcurrencyPruneBatch = bound
1140 })
1141 runner := c.registerRunner("alpha", "ubuntu-latest")
1142 require.NoError(t, c.forge.store.CI().UpdateConcurrency(c.ctx, c.h("proj"), "Deploy", func(st *domain.CIConcurrencyGroupState) error {
1143 st.Marks = map[string]domain.CIConcurrencyMark{}
1144 for i := range stale {
1145 e := domain.CIConcurrencyEntry{RunID: int64(9000 + i), Attempt: 1, JobID: "9000-0-0"}
1146 st.Marks[e.Key()] = domain.CIConcurrencyMark{Entry: e, Group: "Deploy", State: domain.CIConcurrencyPending}
1147 }
1148 return nil
1149 }))
1150 require.Len(t, c.groupState("Deploy").Marks, stale)
1151
1152 r := c.pushN(1, workflowAt(wfGroup, ""))
1153 job := runner.await()
1154 assert.Equal(t, r.GetId(), job.GetRunId(), "the decision succeeded with the backlog in place")
1155 st := c.groupState("Deploy")
1156 require.NotNil(t, st.Record.Running)
1157 assert.Equal(t, r.GetId(), st.Record.Running.RunID)
1158 assert.Len(t, st.Marks, stale-bound+1, "one transaction pruned a bounded number of the stale marks, and marked the run")
1159
1160 c.forge.CI.ReconcileConcurrency(c.ctx)
1161 st = c.groupState("Deploy")
1162 assert.Len(t, st.Marks, 1, "the reconciliation prunes the rest, a batch at a time")
1163 assert.Contains(t, st.Marks, domain.CIConcurrencyEntry{RunID: r.GetId(), Attempt: 1}.Key())
1164 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1165}
1166
1167func (c *ciHarness) hasAnnotation(run *pb.Run, msg string) bool {
1168 for _, j := range c.jobsOf(run) {
1169 for _, a := range c.annotations(run.GetId(), j.GetId()).GetAnnotations() {
1170 if a.GetAnnotation().GetMessage() == msg {
1171 return true
1172 }
1173 }
1174 }
1175 return false
1176}
1177
1178// pendingRuns is the waiting line of a group, as run ids.
1179func (c *ciHarness) pendingRuns(group string) []int64 {
1180 var ids []int64
1181 for _, p := range c.groupRecord(group).Pending {
1182 ids = append(ids, p.RunID)
1183 }
1184 return ids
1185}
1186
1187func (c *ciHarness) completed(run *pb.Run) bool {
1188 return c.getRun(run.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
1189}
1190
1191// queue: max at the workflow and at the job level: three runs of one group
1192// run one after the other in the order they started waiting, none cancelled.
1193func TestCIConcurrencyQueueMaxRunsInOrder(t *testing.T) {
1194 for _, tt := range []struct{ name, top, job string }{
1195 {"workflow", "concurrency:\n group: Deploy\n queue: max\n", ""},
1196 {"job", "", " concurrency:\n group: Deploy\n queue: max\n"},
1197 } {
1198 t.Run(tt.name, func(t *testing.T) {
1199 c, runner := concurrencyHarness(t)
1200 wf := workflowAt(tt.top, tt.job)
1201 r1 := c.pushN(1, wf)
1202 job1 := runner.await()
1203 require.Equal(t, r1.GetId(), job1.GetRunId())
1204 r2 := c.pushN(2, wf)
1205 c.waitJobs("the second run is pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
1206 r3 := c.pushN(3, wf)
1207 c.waitJobs("the third run is pending", r3, allStatus(pb.CiStatus_CI_STATUS_PENDING))
1208 assert.Equal(t, []int64{r2.GetId(), r3.GetId()}, c.pendingRuns("Deploy"), "both wait, in order")
1209 rec := c.groupRecord("Deploy")
1210 assert.True(t, rec.Running.Max && rec.Pending[0].Max && rec.Pending[1].Max, "each entry records its queue mode")
1211
1212 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1213 job2 := runner.await()
1214 assert.Equal(t, r2.GetId(), job2.GetRunId(), "the head of the line goes first")
1215 assert.Equal(t, []int64{r3.GetId()}, c.pendingRuns("Deploy"))
1216 runner.runJob(job2, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1217 job3 := runner.await()
1218 assert.Equal(t, r3.GetId(), job3.GetRunId())
1219 runner.runJob(job3, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1220 for _, r := range []*pb.Run{r1, r2, r3} {
1221 c.waitFor("the run completes", func() bool { return c.completed(r) })
1222 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, c.conclusionOf(r))
1223 }
1224 c.waitFor("the group is free", func() bool { return c.groupRecord("Deploy").IsEmpty() })
1225 })
1226 }
1227}
1228
1229// A queue: max entry that finds the line full is cancelled with ermite's
1230// annotation; the waiting ones stay. The cap is lowered to 2 here (it is
1231// domain.CIMaxConcurrencyQueue, 100, by default; see the domain tests).
1232func TestCIConcurrencyQueueMaxFullLineCancelsTheNewcomer(t *testing.T) {
1233 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
1234 cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
1235 cfg.ConcurrencyQueueMax = 2
1236 })
1237 runner := c.registerRunner("alpha", "ubuntu-latest")
1238 wf := workflowAt("concurrency:\n group: Deploy\n queue: max\n", "")
1239 r1 := c.pushN(1, wf)
1240 job1 := runner.await()
1241 r2 := c.pushN(2, wf)
1242 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
1243 r3 := c.pushN(3, wf)
1244 c.waitJobs("pending", r3, allStatus(pb.CiStatus_CI_STATUS_PENDING))
1245
1246 r4 := c.pushN(4, wf)
1247 c.waitFor("the newcomer is cancelled", func() bool { return c.completed(r4) })
1248 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(r4))
1249 assert.True(t, c.hasAnnotation(r4, domain.ConcurrencyQueueFullMessage("Deploy")))
1250 assert.False(t, c.cancelAnnotation(r4, "Deploy"))
1251 assert.Equal(t, []int64{r2.GetId(), r3.GetId()}, c.pendingRuns("Deploy"), "the line is kept")
1252 assert.False(t, c.groupRecord("Deploy").Names(domain.CIConcurrencyEntry{RunID: r4.GetId(), Attempt: 1}), "the newcomer was never named")
1253 assert.False(t, c.completed(r1))
1254 assert.False(t, c.completed(r2))
1255 assert.False(t, c.completed(r3))
1256
1257 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1258 job2 := runner.await()
1259 assert.Equal(t, r2.GetId(), job2.GetRunId())
1260 runner.runJob(job2, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1261 job3 := runner.await()
1262 assert.Equal(t, r3.GetId(), job3.GetRunId())
1263 runner.runJob(job3, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1264 c.waitFor("the last run completes", func() bool { return c.completed(r3) })
1265}
1266
1267// ermite's rule for a mixed group: a queue: single entry that starts waiting
1268// cancels every waiting entry, queue: max ones included, and is then the only
1269// one; the running entry is left alone (no cancel-in-progress).
1270func TestCIConcurrencyQueueSingleCancelsTheMaxLine(t *testing.T) {
1271 c, runner := concurrencyHarness(t)
1272 maxWF := workflowAt("concurrency:\n group: Deploy\n queue: max\n", "")
1273 r1 := c.pushN(1, maxWF)
1274 job1 := runner.await()
1275 r2 := c.pushN(2, maxWF)
1276 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
1277 r3 := c.pushN(3, maxWF)
1278 c.waitJobs("pending", r3, allStatus(pb.CiStatus_CI_STATUS_PENDING))
1279
1280 r4 := c.pushN(4, workflowAt(wfGroup, ""))
1281 c.waitJobs("the single run is pending", r4, allStatus(pb.CiStatus_CI_STATUS_PENDING))
1282 for _, r := range []*pb.Run{r2, r3} {
1283 c.waitFor("the max run is cancelled", func() bool { return c.completed(r) })
1284 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(r))
1285 assert.True(t, c.cancelAnnotation(r, "Deploy"))
1286 }
1287 assert.Equal(t, []int64{r4.GetId()}, c.pendingRuns("Deploy"))
1288 assert.False(t, c.completed(r1), "the running run is left alone")
1289
1290 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1291 job4 := runner.await()
1292 assert.Equal(t, r4.GetId(), job4.GetRunId())
1293 runner.runJob(job4, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1294 c.waitFor("the single run completes", func() bool { return c.completed(r4) })
1295}