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
271// queue: max is refused.
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 r3 := c.pushN(3, workflowAt("", " concurrency:\n group: g\n queue: max\n cancel-in-progress: true\n"))
280 c.waitJobs("failed", r3, allStatus(pb.CiStatus_CI_STATUS_COMPLETED))
281 assert.Equal(t, "'queue: max' cannot be combined with 'cancel-in-progress: true'.", c.jobsOf(r3)[0].GetError())
282
283 r2 := c.pushN(2, workflowAt("", " concurrency:\n group: g\n queue: max\n"))
284 c.waitJobs("failed", r2, allStatus(pb.CiStatus_CI_STATUS_COMPLETED))
285 assert.Equal(t, "not supported yet: queue: max", strings.TrimSpace(c.jobsOf(r2)[0].GetError()))
286}
287
288// holdGroup writes a group behind the use case's back, as a crash would leave
289// it: the entries are named and marked whatever the state of their runs.
290func (c *ciHarness) holdGroup(group string, running, pending *domain.CIConcurrencyEntry) {
291 c.t.Helper()
292 require.NoError(c.t, c.forge.store.CI().UpdateConcurrency(c.ctx, c.h("proj"), group, func(st *domain.CIConcurrencyGroupState) error {
293 st.Record.Running, st.Record.Pending = running, pending
294 st.Marks = map[string]domain.CIConcurrencyMark{}
295 if running != nil {
296 st.Marks[running.Key()] = domain.CIConcurrencyMark{Entry: *running, Group: group, State: domain.CIConcurrencyHolding}
297 }
298 if pending != nil {
299 st.Marks[pending.Key()] = domain.CIConcurrencyMark{Entry: *pending, Group: group, State: domain.CIConcurrencyPending}
300 }
301 return nil
302 }))
303}
304
305func (c *ciHarness) groupState(group string) (st domain.CIConcurrencyGroupState) {
306 c.t.Helper()
307 require.NoError(c.t, c.forge.store.CI().UpdateConcurrency(c.ctx, c.h("proj"), group, func(s *domain.CIConcurrencyGroupState) error {
308 st = *s
309 return nil
310 }))
311 return st
312}
313
314func (c *ciHarness) groupRecord(group string) domain.CIConcurrencyRecord {
315 return c.groupState(group).Record
316}
317
318// markState is the mark of an entry in a group, "" when it has none.
319func (c *ciHarness) markState(group string, e domain.CIConcurrencyEntry) domain.CIConcurrencyState {
320 return c.groupState(group).Marks[e.Key()].State
321}
322
323// A crash between the write of an ended job and the release of its group leaves
324// the record holding it: the next run to ask finds the entry dead and takes the
325// group.
326func TestCIConcurrencyHealsAfterACrash(t *testing.T) {
327 c, runner := concurrencyHarness(t)
328 r1 := c.pushN(1, workflowAt(wfGroup, ""))
329 runner.runJob(runner.await(), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
330 c.waitFor("the first run completes", func() bool { return c.getRun(r1.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
331 c.waitFor("the group is released", func() bool { return c.groupRecord("Deploy").IsEmpty() })
332 c.holdGroup("Deploy", &domain.CIConcurrencyEntry{RunID: r1.GetId(), Attempt: 1}, nil) // the release that did not happen
333
334 r2 := c.pushN(2, workflowAt(wfGroup, ""))
335 job2 := runner.await()
336 assert.Equal(t, r2.GetId(), job2.GetRunId(), "the dead entry does not hold the group")
337 rec := c.groupRecord("Deploy")
338 require.NotNil(t, rec.Running)
339 assert.Equal(t, r2.GetId(), rec.Running.RunID)
340}
341
342// A run that is deleted (the retention sweep, not its repository) frees the
343// group it held and the one it waited in.
344func TestCIConcurrencyHealsAfterRunDeletion(t *testing.T) {
345 c, runner := concurrencyHarness(t)
346 r1 := c.pushN(1, workflowAt(wfGroup, ""))
347 runner.runJob(runner.await(), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
348 c.waitFor("completes", func() bool { return c.getRun(r1.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
349 c.waitFor("released", func() bool { return c.groupRecord("Deploy").IsEmpty() })
350 ghost := &domain.CIConcurrencyEntry{RunID: r1.GetId(), Attempt: 1}
351 c.holdGroup("Deploy", ghost, nil)
352 ids, err := c.forge.store.CI().Sweep(c.ctx, time.Now().Add(time.Hour))
353 require.NoError(t, err)
354 require.Contains(t, ids, r1.GetId(), "the run is deleted")
355
356 r2 := c.pushN(2, workflowAt(wfGroup, ""))
357 job2 := runner.await()
358 assert.Equal(t, r2.GetId(), job2.GetRunId(), "a deleted run does not hold the group")
359}
360
361// flakyStore fails the writes that match, as many times as fails says. The
362// failure is returned before anything is written.
363type flakyStore struct {
364 ports.CIStore
365 match func(ports.CIWrite) bool
366 fails atomic.Int32
367 hits atomic.Int32
368}
369
370func (f *flakyStore) Apply(ctx context.Context, w ports.CIWrite) (uint64, error) {
371 if f.match != nil && f.match(w) && f.fails.Add(-1) >= 0 {
372 f.hits.Add(1)
373 return 0, domain.ErrConflict
374 }
375 return f.CIStore.Apply(ctx, w)
376}
377
378// flakyConcurrencyHarness is a harness whose store fails the writes fs.match
379// selects. retries is CIConfig.ConcurrencyRetries: 0 keeps the default, a
380// negative number leaves the recovery of a lost delivery to the reconciliation.
381func flakyConcurrencyHarness(t *testing.T, retries int) (*ciHarness, *fakeRunner, *flakyStore) {
382 t.Helper()
383 fs := &flakyStore{}
384 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
385 cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
386 cfg.ConcurrencyRetries = retries
387 fs.CIStore, cfg.Store = cfg.Store, fs
388 })
389 return c, c.registerRunner("alpha", "ubuntu-latest"), fs
390}
391
392// endsCancelled matches the write that ends a job of the run as cancelled.
393func endsCancelled(runID int64) func(ports.CIWrite) bool {
394 return func(w ports.CIWrite) bool {
395 for _, j := range w.Jobs {
396 if j.RunID == runID && j.Ended() && j.Conclusion == domain.CICancelled {
397 return true
398 }
399 }
400 return false
401 }
402}
403
404// A replaced pending run whose cancel write fails once is cancelled by the
405// retry.
406func TestCIConcurrencyCancelSurvivesAFailedDelivery(t *testing.T) {
407 c, runner, fs := flakyConcurrencyHarness(t, 0)
408 c.pushN(1, workflowAt(wfGroup, ""))
409 job1 := runner.await()
410 r2 := c.pushN(2, workflowAt(wfGroup, ""))
411 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
412
413 fs.match = endsCancelled(r2.GetId())
414 fs.fails.Store(1)
415 r3 := c.pushN(3, workflowAt(wfGroup, ""))
416 c.waitFor("the first delivery failed", func() bool { return fs.hits.Load() == 1 })
417 rec := c.groupRecord("Deploy")
418 require.NotNil(t, rec.Pending)
419 assert.Equal(t, r3.GetId(), rec.Pending.RunID)
420 c.waitFor("the replaced run is cancelled by the retry", func() bool { return c.conclusionOf(r2) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
421 assert.True(t, c.cancelAnnotation(r2, "Deploy"))
422 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
423 assert.Equal(t, r3.GetId(), runner.await().GetRunId())
424}
425
426// Nothing is bounded: more replacements than any list could hold, every cancel
427// write failing, and after the store works again the reconciliation discovers
428// each victim through its "replaced" mark in the group, and cancels it.
429func TestCIConcurrencyReconcileCancelsEveryReplacedVictim(t *testing.T) {
430 const replaced = 35
431 c, runner, fs := flakyConcurrencyHarness(t, -1)
432 c.pushN(1, workflowAt(wfGroup, ""))
433 runner.await()
434
435 // Each new run replaces the one that waited before it; no cancel lands.
436 fs.match = func(w ports.CIWrite) bool {
437 for _, j := range w.Jobs {
438 if j.Ended() && j.Conclusion == domain.CICancelled {
439 return true
440 }
441 }
442 return false
443 }
444 fs.fails.Store(1 << 20)
445 var victims []*pb.Run
446 prev := c.pushN(2, workflowAt(wfGroup, ""))
447 c.waitJobs("pending", prev, allStatus(pb.CiStatus_CI_STATUS_PENDING))
448 for i := range replaced {
449 next := c.pushN(3+i, workflowAt(wfGroup, ""))
450 c.waitFor("the newer run takes the waiting place", func() bool {
451 rec := c.groupRecord("Deploy")
452 return rec.Pending != nil && rec.Pending.RunID == next.GetId()
453 })
454 c.waitJobs("the newer run is pending", next, allStatus(pb.CiStatus_CI_STATUS_PENDING))
455 victims = append(victims, prev)
456 prev = next
457 }
458 for _, v := range victims {
459 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(v), "no cancel landed yet")
460 }
461 // The test has no in-process retry (ConcurrencyRetries is negative): only
462 // the reconciliation can recover the victims.
463 fs.fails.Store(0)
464 c.forge.CI.ReconcileConcurrency(c.ctx)
465 for _, v := range victims {
466 c.waitFor("a replaced run is cancelled", func() bool { return c.conclusionOf(v) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
467 assert.True(t, c.cancelAnnotation(v, "Deploy"))
468 }
469 c.waitJobs("the last one still waits", prev, allStatus(pb.CiStatus_CI_STATUS_PENDING))
470 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(prev))
471}
472
473// A running entry that cancel-in-progress replaced stays named with Canceling
474// set; when its cancel is lost, the reconciliation asks again.
475func TestCIConcurrencyReconcileAsksARunningVictimAgain(t *testing.T) {
476 c, runner, fs := flakyConcurrencyHarness(t, -1)
477 top := "concurrency:\n group: Deploy\n cancel-in-progress: true\n"
478 r1 := c.pushN(1, workflowAt(top, ""))
479 job1 := runner.await()
480 watch, err := runner.jobAPI(job1).runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: job1.GetJobId()}))
481 require.NoError(t, err)
482
483 // The cancel of the running run cannot be written.
484 fs.match = func(w ports.CIWrite) bool {
485 for _, r := range w.Runs {
486 if r.ID == r1.GetId() && r.CancelRequested {
487 return true
488 }
489 }
490 return false
491 }
492 fs.fails.Store(1 << 20)
493 c.pushN(2, workflowAt(top, ""))
494 c.waitFor("the cancel fails", func() bool { return fs.hits.Load() >= 1 })
495 rec := c.groupRecord("Deploy")
496 require.NotNil(t, rec.Running)
497 assert.True(t, rec.Running.Canceling, "the victim stays named, marked")
498 assert.Equal(t, r1.GetId(), rec.Running.RunID)
499
500 fs.fails.Store(0)
501 c.forge.CI.ReconcileConcurrency(c.ctx)
502 var cancel *pb.CancelRequest
503 for watch.Receive() {
504 if m := watch.Msg().GetCancel(); m != nil {
505 cancel = m
506 break
507 }
508 }
509 require.NotNil(t, cancel, "the runner is told after the reconciliation")
510 // The runner stops: its watch ends with the job.
511 _, err = runner.jobAPI(job1).runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
512 JobId: job1.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED,
513 }))
514 require.NoError(t, err)
515}
516
517// Crash between the write of the group and the write of the entry: the group
518// names a pending run whose own record does not say it waits yet. It is not
519// stale and not cancelled; the reconciliation plans it, which marks it
520// pending, and the group goes on to it when the holder ends.
521func TestCIConcurrencyCrashBetweenGroupAndEntry(t *testing.T) {
522 c, runner := concurrencyHarness(t)
523 c.pushN(1, workflowAt(wfGroup, ""))
524 job1 := runner.await()
525
526 // Stop the planning of the second run from marking it, as a crash would:
527 // name it in the group by hand, behind the first.
528 store := c.forge.store.CI()
529 r2 := c.pushN(2, workflowAt(wfGroup, ""))
530 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
531 run, err := store.GetRun(c.ctx, c.h("proj"), r2.GetId())
532 require.NoError(t, err)
533 run.Concurrency.State = domain.CIConcurrencyIdle
534 jobs, err := store.ListJobs(c.ctx, r2.GetId(), 1)
535 require.NoError(t, err)
536 for i := range jobs {
537 jobs[i].Concurrency.State = domain.CIConcurrencyIdle
538 }
539 _, err = store.Apply(c.ctx, ports.CIWrite{Runs: []domain.CIRun{run}, Jobs: jobs})
540 require.NoError(t, err)
541 rec := c.groupRecord("Deploy")
542 require.NotNil(t, rec.Pending)
543 require.Equal(t, r2.GetId(), rec.Pending.RunID, "the group names the run")
544 assert.Equal(t, domain.CIConcurrencyPending, c.markState("Deploy", *rec.Pending), "and marks it, in the same batch")
545
546 c.forge.CI.ReconcileConcurrency(c.ctx)
547 c.waitFor("the entry is marked pending again", func() bool {
548 r, err := store.GetRun(c.ctx, c.h("proj"), r2.GetId())
549 return err == nil && r.Concurrency.State == domain.CIConcurrencyPending
550 })
551 time.Sleep(200 * time.Millisecond)
552 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(r2), "a named entry is not a replaced one")
553 rec = c.groupRecord("Deploy")
554 require.NotNil(t, rec.Pending)
555 assert.Equal(t, r2.GetId(), rec.Pending.RunID)
556
557 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
558 assert.Equal(t, r2.GetId(), runner.await().GetRunId())
559}
560
561// holdsGroup matches the write that tells a run it holds its group.
562func holdsGroup(runID int64) func(ports.CIWrite) bool {
563 return func(w ports.CIWrite) bool {
564 for _, r := range w.Runs {
565 if r.ID == runID && r.Concurrency.State == domain.CIConcurrencyHolding {
566 return true
567 }
568 }
569 return false
570 }
571}
572
573// HIGH-2: a promotion whose planning fails once is tried again.
574func TestCIConcurrencyPromotionIsRetried(t *testing.T) {
575 c, runner, fs := flakyConcurrencyHarness(t, 0)
576 c.pushN(1, workflowAt(wfGroup, ""))
577 job1 := runner.await()
578 r2 := c.pushN(2, workflowAt(wfGroup, ""))
579 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
580
581 fs.match = holdsGroup(r2.GetId())
582 fs.fails.Store(1)
583 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
584 c.waitFor("the first planning failed", func() bool { return fs.hits.Load() == 1 })
585 job2 := runner.await()
586 assert.Equal(t, r2.GetId(), job2.GetRunId(), "the retry planned the promoted run")
587}
588
589// HIGH-2: a holder that was never told it holds the group is planned by the
590// reconciliation, whatever happened to the retries.
591func TestCIConcurrencyReconcilePlansAStaleHolder(t *testing.T) {
592 c, runner, fs := flakyConcurrencyHarness(t, -1)
593 c.pushN(1, workflowAt(wfGroup, ""))
594 job1 := runner.await()
595 r2 := c.pushN(2, workflowAt(wfGroup, ""))
596 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
597
598 fs.match = holdsGroup(r2.GetId())
599 fs.fails.Store(1000)
600 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
601 c.waitFor("the planning fails", func() bool { return fs.hits.Load() >= 1 })
602 rec := c.groupRecord("Deploy")
603 require.NotNil(t, rec.Running)
604 assert.Equal(t, r2.GetId(), rec.Running.RunID, "the group names the promoted run")
605 c.waitJobs("which still shows pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
606
607 fs.fails.Store(0)
608 c.forge.CI.ReconcileConcurrency(c.ctx)
609 job2 := runner.await()
610 assert.Equal(t, r2.GetId(), job2.GetRunId())
611}
612
613const siblingJobs = ` steps:
614 - run: echo grouped
615 sibling:
616 runs-on: ubuntu-latest
617 steps:
618 - run: echo sibling
619`
620
621// jobWorkflow is a workflow of a grouped job with the given concurrency lines
622// and an unrelated sibling.
623func jobWorkflow(concurrency string) string {
624 return "name: ci\non:\n push:\n branches: [main]\njobs:\n grouped:\n runs-on: ubuntu-latest\n" +
625 concurrency + siblingJobs
626}
627
628// awaitKey fetches jobs until the one of the run with the key comes.
629func (r *fakeRunner) awaitKey(runID int64, key string) *pb.Job {
630 r.c.t.Helper()
631 for range 20 {
632 if j := r.await(); j.GetRunId() == runID && j.GetJobKey() == key {
633 return j
634 }
635 }
636 r.c.t.Fatalf("no job %s of run %d", key, runID)
637 return nil
638}
639
640// T-2: a job-level cancel-in-progress (true, or an expression) cancels the
641// running job of the group only: its sibling goes on, and the replacement
642// becomes assignable.
643func TestCIJobConcurrencyCancelInProgress(t *testing.T) {
644 for _, tt := range []struct{ name, cancel string }{
645 {"true", "true"},
646 {"expression", "${{ github.ref == 'refs/heads/main' }}"},
647 } {
648 t.Run(tt.name, func(t *testing.T) {
649 c, runner := concurrencyHarness(t)
650 wf := jobWorkflow(" concurrency:\n group: jobgroup\n cancel-in-progress: " + tt.cancel + "\n")
651 r1 := c.pushN(1, wf)
652 grouped1, sibling1 := runner.awaitKey(r1.GetId(), "grouped"), runner.awaitKey(r1.GetId(), "sibling")
653 watch, err := runner.jobAPI(grouped1).runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: grouped1.GetJobId()}))
654 require.NoError(t, err)
655
656 r2 := c.pushN(2, wf)
657 var cancel *pb.CancelRequest
658 for watch.Receive() {
659 if m := watch.Msg().GetCancel(); m != nil {
660 cancel = m
661 break
662 }
663 }
664 require.NotNil(t, cancel, "the grouped job is asked to stop")
665 assert.Contains(t, cancel.GetMessage(), "higher priority waiting request for 'jobgroup'")
666
667 // The sibling of the first run was not touched.
668 runner.heartbeat(sibling1)
669 for _, j := range c.jobsOf(r1) {
670 if j.GetKey() == "sibling" {
671 assert.NotEqual(t, pb.CiStatus_CI_STATUS_COMPLETED, j.GetStatus())
672 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_UNSPECIFIED, j.GetConclusion())
673 }
674 }
675 _, err = runner.jobAPI(grouped1).runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
676 JobId: grouped1.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED,
677 }))
678 require.NoError(t, err)
679 runner.runJob(sibling1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
680
681 grouped2 := runner.awaitKey(r2.GetId(), "grouped")
682 assert.Equal(t, r2.GetId(), grouped2.GetRunId(), "the replacement is assignable")
683 c.waitFor("the first run is over", func() bool { return c.getRun(r1.GetId()).GetRun().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
684 for _, j := range c.jobsOf(r1) {
685 want := pb.CiConclusion_CI_CONCLUSION_CANCELLED
686 if j.GetKey() == "sibling" {
687 want = pb.CiConclusion_CI_CONCLUSION_SUCCESS
688 }
689 assert.Equal(t, want, j.GetConclusion(), j.GetKey())
690 }
691 })
692 }
693}
694
695// T-3: the server restarts: the groups are repaired at startup. A group held by
696// an ended run is freed and the run behind it is planned; a group whose only
697// run is over is removed.
698func TestCIConcurrencyReconcileAtRestart(t *testing.T) {
699 c, runner := concurrencyHarness(t)
700 r1 := c.pushN(1, workflowAt(wfGroup, ""))
701 runner.await()
702 r2 := c.pushN(2, workflowAt(wfGroup, ""))
703 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
704
705 // The first run ends in the store without the release (the crash), and a
706 // second group is held by a run that is over.
707 store := c.forge.store.CI()
708 jobs, err := store.ListJobs(c.ctx, r1.GetId(), 1)
709 require.NoError(t, err)
710 run, err := store.GetRun(c.ctx, c.h("proj"), r1.GetId())
711 require.NoError(t, err)
712 jobs[0].Phase, jobs[0].Conclusion = domain.CIPhaseEnded, domain.CISuccess
713 run.Status, run.Conclusion, run.CompletedAt = domain.CICompleted, domain.CISuccess, time.Now()
714 _, err = store.Apply(c.ctx, ports.CIWrite{Jobs: jobs, Runs: []domain.CIRun{run}})
715 require.NoError(t, err)
716 c.holdGroup("Stale", &domain.CIConcurrencyEntry{RunID: r1.GetId(), Attempt: 1}, nil)
717 rec := c.groupRecord("Deploy")
718 require.NotNil(t, rec.Running)
719 require.Equal(t, r1.GetId(), rec.Running.RunID, "the record still holds the ended run")
720
721 // Restart on the same data.
722 cfg, hookExe := c.forge.cfg, binary(t)
723 c.srv.Close()
724 c.forge.Close(c.ctx)
725 f2, err := newForge(c.ctx, cfg, options{hookExe: hookExe, ci: func(cc *usecase.CIConfig) {
726 cc.CancelGrace, cc.Tick, cc.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
727 }})
728 require.NoError(t, err)
729 t.Cleanup(func() { f2.Close(context.WithoutCancel(c.ctx)) })
730 srv := httptest.NewServer(f2.server.Handler())
731 t.Cleanup(srv.Close)
732
733 var got *pb.Job
734 c.waitFor("the waiting run is planned by the startup repair", func() bool {
735 res, err := newClients(srv.URL+"/-/api", runner.secret).runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{}))
736 if err == nil && res.Msg.GetJob() != nil {
737 got = res.Msg.GetJob()
738 }
739 return got != nil
740 })
741 assert.Equal(t, r2.GetId(), got.GetRunId())
742 groups, err := f2.store.CI().ListConcurrency(c.ctx)
743 require.NoError(t, err)
744 require.Len(t, groups, 1, "the stale group is gone")
745 assert.Equal(t, "DEPLOY", groups[0].Group)
746 require.NotNil(t, groups[0].State.Record.Running)
747 assert.Equal(t, r2.GetId(), groups[0].State.Record.Running.RunID)
748 assert.Nil(t, groups[0].State.Record.Pending)
749}
750
751// Both writes of a run's own record fail when it queues (its "pending" state),
752// and the cancel of the run fails for good: the group alone knows B. A newer run
753// replaces B; B's mark is "replaced", written in the batch that unnamed it. After
754// the store works again the reconciliation finds B from that mark and cancels it.
755func TestCIConcurrencyReplacedRunWithLostWritesIsCancelled(t *testing.T) {
756 c, runner, fs := flakyConcurrencyHarness(t, -1)
757 r1 := c.pushN(1, workflowAt(wfGroup, ""))
758 runner.await()
759
760 // B queues, but no run after the first gets the pending state in its own
761 // record, and no cancel lands.
762 fs.match = func(w ports.CIWrite) bool {
763 for _, r := range w.Runs {
764 if r.ID > r1.GetId() && r.Concurrency.State == domain.CIConcurrencyPending {
765 return true
766 }
767 }
768 for _, j := range w.Jobs {
769 if j.RunID > r1.GetId() && j.Ended() && j.Conclusion == domain.CICancelled {
770 return true
771 }
772 }
773 return false
774 }
775 fs.fails.Store(1 << 20)
776 b := c.pushN(2, workflowAt(wfGroup, ""))
777 c.waitFor("B is named by the group", func() bool {
778 rec := c.groupRecord("Deploy")
779 return rec.Pending != nil && rec.Pending.RunID == b.GetId()
780 })
781 bEntry := domain.CIConcurrencyEntry{RunID: b.GetId(), Attempt: 1}
782 assert.Equal(t, domain.CIConcurrencyPending, c.markState("Deploy", bEntry), "the group batch marked it")
783
784 // C replaces B: B is unnamed and marked replaced in one batch.
785 c.pushN(3, workflowAt(wfGroup, ""))
786 c.waitFor("B is marked replaced", func() bool { return c.markState("Deploy", bEntry) == domain.CIConcurrencyReplaced })
787 rec := c.groupRecord("Deploy")
788 require.NotNil(t, rec.Pending)
789 assert.NotEqual(t, b.GetId(), rec.Pending.RunID, "B is no longer named")
790 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(b))
791
792 fs.fails.Store(0)
793 c.forge.CI.ReconcileConcurrency(c.ctx)
794 c.waitFor("B ends cancelled", func() bool { return c.conclusionOf(b) == pb.CiConclusion_CI_CONCLUSION_CANCELLED })
795 assert.True(t, c.cancelAnnotation(b, "Deploy"))
796 c.waitFor("its mark goes with it", func() bool { return c.markState("Deploy", bEntry) == "" })
797}
798
799// A grouped job B is replaced by C, the cancel never lands, C is promoted when
800// the running job ends, and an unrelated sibling of B completes and advances B:
801// B's grouped job must not take the free place, it ends cancelled and never runs.
802func TestCIJobConcurrencyReplacedJobNeverAcquiresAgain(t *testing.T) {
803 c, runner, fs := flakyConcurrencyHarness(t, -1)
804 wf := jobWorkflow(" concurrency: jobgroup\n")
805 r1 := c.pushN(1, wf)
806 grouped1, sibling1 := runner.awaitKey(r1.GetId(), "grouped"), runner.awaitKey(r1.GetId(), "sibling")
807
808 rB := c.pushN(2, wf)
809 siblingB := runner.awaitKey(rB.GetId(), "sibling") // B's unrelated job is running
810 c.waitJobs("B's grouped job waits", rB, func(js []*pb.RunJob) bool {
811 for _, j := range js {
812 if j.GetKey() == "grouped" {
813 return j.GetStatus() == pb.CiStatus_CI_STATUS_PENDING
814 }
815 }
816 return false
817 })
818
819 fs.match = endsCancelled(rB.GetId())
820 fs.fails.Store(1 << 20)
821 rC := c.pushN(3, wf)
822 bEntry := domain.CIConcurrencyEntry{RunID: rB.GetId(), Attempt: 1, JobID: domain.NewJobID(rB.GetId(), 0, 0)}
823 c.waitFor("B's grouped job is replaced", func() bool { return c.markState("jobgroup", bEntry) == domain.CIConcurrencyReplaced })
824
825 // The running job ends: C's grouped job is promoted.
826 runner.runJob(grouped1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
827 runner.runJob(sibling1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
828 groupedC := runner.awaitKey(rC.GetId(), "grouped")
829 rec := c.groupRecord("jobgroup")
830 require.NotNil(t, rec.Running)
831 assert.Equal(t, rC.GetId(), rec.Running.RunID)
832
833 // B's sibling completes: B is advanced, its grouped job asks for the group.
834 runner.runJob(siblingB, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
835 c.waitFor("B was advanced", func() bool {
836 for _, j := range c.jobsOf(rB) {
837 if j.GetKey() == "sibling" {
838 return j.GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
839 }
840 }
841 return false
842 })
843 time.Sleep(300 * time.Millisecond)
844 rec = c.groupRecord("jobgroup")
845 require.NotNil(t, rec.Running)
846 assert.Equal(t, rC.GetId(), rec.Running.RunID, "C still holds the group")
847 if rec.Pending != nil {
848 assert.NotEqual(t, rB.GetId(), rec.Pending.RunID, "B did not take the waiting place back")
849 }
850 // Nothing of B's grouped job is given to a runner.
851 for range 5 {
852 j, err := runner.fetch("")
853 require.NoError(t, err)
854 if j != nil {
855 assert.False(t, j.GetRunId() == rB.GetId() && j.GetJobKey() == "grouped", "B's grouped job must not run")
856 }
857 }
858
859 // Once the store works, the reconciliation cancels it.
860 fs.fails.Store(0)
861 c.forge.CI.ReconcileConcurrency(c.ctx)
862 c.waitFor("B's grouped job ends cancelled", func() bool {
863 for _, j := range c.jobsOf(rB) {
864 if j.GetKey() == "grouped" {
865 return j.GetConclusion() == pb.CiConclusion_CI_CONCLUSION_CANCELLED
866 }
867 }
868 return false
869 })
870 runner.runJob(groupedC, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
871}
872
873// The same, with an empty group: B's grouped job is replaced and cancelled
874// before its replacement ends too; a replaced job never runs even when the
875// group is free.
876func TestCIJobConcurrencyReplacedJobDoesNotTakeAnEmptyGroup(t *testing.T) {
877 c, runner, fs := flakyConcurrencyHarness(t, -1)
878 wf := jobWorkflow(" concurrency: jobgroup\n")
879 r1 := c.pushN(1, wf)
880 grouped1, sibling1 := runner.awaitKey(r1.GetId(), "grouped"), runner.awaitKey(r1.GetId(), "sibling")
881 rB := c.pushN(2, wf)
882 siblingB := runner.awaitKey(rB.GetId(), "sibling")
883 c.waitJobs("pending", rB, func(js []*pb.RunJob) bool {
884 for _, j := range js {
885 if j.GetKey() == "grouped" {
886 return j.GetStatus() == pb.CiStatus_CI_STATUS_PENDING
887 }
888 }
889 return false
890 })
891 fs.match = endsCancelled(rB.GetId())
892 fs.fails.Store(1 << 20)
893 rC := c.pushN(3, wf)
894 bEntry := domain.CIConcurrencyEntry{RunID: rB.GetId(), Attempt: 1, JobID: domain.NewJobID(rB.GetId(), 0, 0)}
895 c.waitFor("replaced", func() bool { return c.markState("jobgroup", bEntry) == domain.CIConcurrencyReplaced })
896
897 // Both the running job and the promoted one finish: the group is empty.
898 runner.runJob(grouped1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
899 runner.runJob(sibling1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
900 runner.runJob(runner.awaitKey(rC.GetId(), "grouped"), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
901 c.waitFor("the group is free", func() bool { return c.groupRecord("jobgroup").Running == nil })
902
903 runner.runJob(siblingB, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil) // advances B
904 time.Sleep(300 * time.Millisecond)
905 assert.Nil(t, c.groupRecord("jobgroup").Running, "B's replaced job did not take the empty group")
906 for range 5 {
907 j, err := runner.fetch("")
908 require.NoError(t, err)
909 if j != nil {
910 assert.False(t, j.GetRunId() == rB.GetId() && j.GetJobKey() == "grouped", "B's grouped job must not run")
911 }
912 }
913 fs.fails.Store(0)
914 c.forge.CI.ReconcileConcurrency(c.ctx)
915 c.waitFor("it 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}
924
925// Crash between the group batch and the write of a JOB's own state: the group
926// names (and marks) the grouped job as waiting, the job's record does not say so
927// yet. It is not replaced and not cancelled; reconciliation tells it, and it runs
928// when its turn comes.
929func TestCIJobConcurrencyCrashBetweenGroupAndJob(t *testing.T) {
930 c, runner := concurrencyHarness(t)
931 group := " concurrency: jobgroup\n"
932 c.pushN(1, workflowAt("", group))
933 job1 := runner.await()
934 r2 := c.pushN(2, workflowAt("", group))
935 c.waitJobs("pending", r2, allStatus(pb.CiStatus_CI_STATUS_PENDING))
936
937 store := c.forge.store.CI()
938 jobs, err := store.ListJobs(c.ctx, r2.GetId(), 1)
939 require.NoError(t, err)
940 require.Len(t, jobs, 1)
941 entry := domain.CIConcurrencyEntry{RunID: r2.GetId(), Attempt: 1, JobID: jobs[0].ID}
942 jobs[0].Concurrency.State = domain.CIConcurrencyIdle // the write that was lost
943 _, err = store.Apply(c.ctx, ports.CIWrite{Jobs: jobs})
944 require.NoError(t, err)
945 rec := c.groupRecord("jobgroup")
946 require.NotNil(t, rec.Pending)
947 assert.Equal(t, entry.JobID, rec.Pending.JobID, "the group names the job")
948 assert.Equal(t, domain.CIConcurrencyPending, c.markState("jobgroup", entry), "and its mark says so")
949
950 c.forge.CI.ReconcileConcurrency(c.ctx)
951 c.waitFor("the job is told it waits", func() bool {
952 js, err := store.ListJobs(c.ctx, r2.GetId(), 1)
953 return err == nil && js[0].Concurrency.State == domain.CIConcurrencyPending
954 })
955 time.Sleep(200 * time.Millisecond)
956 assert.NotEqual(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, c.conclusionOf(r2), "a named job is not a replaced one")
957 assert.Equal(t, domain.CIConcurrencyPending, c.markState("jobgroup", entry))
958
959 runner.runJob(job1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
960 assert.Equal(t, r2.GetId(), runner.await().GetRunId(), "it runs when the group is free")
961}
962
963// plannedWrite matches a write of a run after the first that stores planned jobs.
964func plannedWrite(after int64) func(ports.CIWrite) bool {
965 return func(w ports.CIWrite) bool {
966 for _, j := range w.Jobs {
967 if j.RunID > after && j.Planned {
968 return true
969 }
970 }
971 return false
972 }
973}
974
975const matrixGroupWorkflow = `name: b
976on:
977 push:
978 branches: [other]
979jobs:
980 build:
981 runs-on: ubuntu-latest
982 strategy:
983 matrix:
984 n: [1, 2, 3]
985 concurrency: shared
986 steps:
987 - run: echo ${{ matrix.n }}
988`
989
990const holderWorkflow = `name: a
991on:
992 push:
993 branches: [main]
994jobs:
995 hold:
996 runs-on: ubuntu-latest
997 concurrency: shared
998 steps:
999 - run: echo hold
1000`
1001
1002// Run B plans three legs of a matrix that share a group held by A. The legs
1003// replace one another (leg 0 by leg 1 by leg 2: only one waits), the group
1004// batches commit, and the write of B's planned jobs fails, leaving the
1005// placeholder of the job in the store. After the store recovers, the
1006// reconciliation must plan B before it cancels anything: only legs 0 and 1
1007// end cancelled, and leg 2, the legitimate waiting one, runs when A ends.
1008func TestCIJobConcurrencyRecoveryPlansTheMatrixBeforeCancellingALeg(t *testing.T) {
1009 c, runner, fs := flakyConcurrencyHarness(t, -1)
1010 a := c.push("main", map[string]string{".github/workflows/a.yml": holderWorkflow}, "")
1011 _ = a
1012 holder := runner.await()
1013 require.Equal(t, "hold", holder.GetJobKey())
1014 aRun := holder.GetRunId()
1015
1016 fs.match = plannedWrite(aRun)
1017 fs.fails.Store(1 << 20)
1018 c.push("other", map[string]string{".github/workflows/b.yml": matrixGroupWorkflow}, "")
1019 c.waitFor("B's run exists", func() bool {
1020 for _, r := range c.runs() {
1021 if r.GetId() > aRun {
1022 return true
1023 }
1024 }
1025 return false
1026 })
1027 var bRun *pb.Run
1028 for _, r := range c.runs() {
1029 if r.GetId() > aRun {
1030 bRun = r
1031 }
1032 }
1033 require.NotNil(t, bRun)
1034 g := domain.ConcurrencyKey("shared")
1035 c.waitFor("the group batches committed: two legs are marked replaced", func() bool {
1036 replaced := 0
1037 for _, m := range c.groupState(g).Marks {
1038 if m.Entry.RunID == bRun.GetId() && m.State == domain.CIConcurrencyReplaced {
1039 replaced++
1040 }
1041 }
1042 return replaced == 2
1043 })
1044 assert.Equal(t, 1, len(c.jobsOf(bRun)), "B's job is still the placeholder: its planned legs were never stored")
1045
1046 fs.fails.Store(0)
1047 c.forge.CI.ReconcileConcurrency(c.ctx)
1048 c.waitFor("B has its three legs", func() bool { return len(c.jobsOf(bRun)) == 3 })
1049 c.waitFor("legs 0 and 1 end cancelled", func() bool {
1050 n := 0
1051 for _, j := range c.jobsOf(bRun) {
1052 if j.GetConclusion() == pb.CiConclusion_CI_CONCLUSION_CANCELLED {
1053 n++
1054 }
1055 }
1056 return n == 2
1057 })
1058 byCombo := map[string]*pb.RunJob{}
1059 for _, j := range c.jobsOf(bRun) {
1060 byCombo[j.GetName()] = j
1061 }
1062 var waiting []*pb.RunJob
1063 for _, j := range c.jobsOf(bRun) {
1064 if j.GetConclusion() != pb.CiConclusion_CI_CONCLUSION_CANCELLED {
1065 waiting = append(waiting, j)
1066 }
1067 }
1068 require.Len(t, waiting, 1, "exactly one leg is not cancelled")
1069 assert.Equal(t, pb.CiStatus_CI_STATUS_PENDING, waiting[0].GetStatus(), "the final leg waits for A")
1070 assert.Contains(t, waiting[0].GetName(), "3", "and it is the last leg")
1071
1072 runner.runJob(holder, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1073 last := runner.await()
1074 assert.Equal(t, bRun.GetId(), last.GetRunId())
1075 assert.Equal(t, waiting[0].GetId(), last.GetJobId(), "the final leg runs when A ends")
1076 runner.runJob(last, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1077}
1078
1079// Stale marks (of runs that are gone) are pruned a bounded number per
1080// transaction: a decision on a group with a long backlog succeeds at once, and
1081// the reconciliation takes the others, a bounded batch at a time.
1082func TestCIConcurrencyPrunesStaleMarksInBoundedBatches(t *testing.T) {
1083 const stale, bound = 10, 3
1084 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
1085 cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
1086 cfg.ConcurrencyPruneBatch = bound
1087 })
1088 runner := c.registerRunner("alpha", "ubuntu-latest")
1089 require.NoError(t, c.forge.store.CI().UpdateConcurrency(c.ctx, c.h("proj"), "Deploy", func(st *domain.CIConcurrencyGroupState) error {
1090 st.Marks = map[string]domain.CIConcurrencyMark{}
1091 for i := range stale {
1092 e := domain.CIConcurrencyEntry{RunID: int64(9000 + i), Attempt: 1, JobID: "9000-0-0"}
1093 st.Marks[e.Key()] = domain.CIConcurrencyMark{Entry: e, Group: "Deploy", State: domain.CIConcurrencyPending}
1094 }
1095 return nil
1096 }))
1097 require.Len(t, c.groupState("Deploy").Marks, stale)
1098
1099 r := c.pushN(1, workflowAt(wfGroup, ""))
1100 job := runner.await()
1101 assert.Equal(t, r.GetId(), job.GetRunId(), "the decision succeeded with the backlog in place")
1102 st := c.groupState("Deploy")
1103 require.NotNil(t, st.Record.Running)
1104 assert.Equal(t, r.GetId(), st.Record.Running.RunID)
1105 assert.Len(t, st.Marks, stale-bound+1, "one transaction pruned a bounded number of the stale marks, and marked the run")
1106
1107 c.forge.CI.ReconcileConcurrency(c.ctx)
1108 st = c.groupState("Deploy")
1109 assert.Len(t, st.Marks, 1, "the reconciliation prunes the rest, a batch at a time")
1110 assert.Contains(t, st.Marks, domain.CIConcurrencyEntry{RunID: r.GetId(), Attempt: 1}.Key())
1111 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
1112}