1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "maps"
10 "slices"
11 "time"
12
13 "bnema.dev/ermite-actions/expr"
14 "bnema.dev/ermite-actions/template"
15 "bnema.dev/ermite-actions/workflow"
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// Concurrency groups (CI-3 phase E). A group is per repository and compared
21// without regard to case; a run (workflow level) or a job (job level) is an
22// entry. At most one entry runs; the others wait in line, first in first out.
23// With queue: single (the default) a newer waiting entry cancels every waiting
24// one, and cancel-in-progress also cancels the running one. With queue: max a
25// newer entry joins the end of the line, up to domain.CIMaxConcurrencyQueue
26// waiting entries; one more is cancelled (domain.ConcurrencyQueueFullMessage).
27// GitHub does not document a group whose entries mix both modes: ermite applies
28// each entry's own rule when it starts waiting (domain.CIConcurrencyRecord.
29// Enqueue), so a single entry clears the whole line, max ones included. An
30// entry holds its group from the moment it may run until the run or job has
31// ended, and the head of the line takes it then.
32//
33// The record of a group lives under "concur/<repo>/<key(group)>" and is
34// changed with a version-checked write (Store.UpdateConcurrency). Cancelling
35// another entry needs the lock of another run, which the caller of a change
36// does not hold in order (two runs would wait for each other): those effects
37// are queued with later and run after the lock is released.
38
39// errEmptyGroup and errLongGroup are GitHub's errors for a group that
40// evaluates to nothing or too much (actions/runner,
41// WorkflowTemplateConverter.cs ConvertToConcurrency, lines 679-687).
42var (
43 errEmptyGroup = errors.New("concurrency group name cannot be empty")
44 // errQueueMaxCancel is the message of languageservices
45 // (languageservice/src/validate.ts:743).
46 errQueueMaxCancel = errors.New("'queue: max' cannot be combined with 'cancel-in-progress: true'.") //nolint:staticcheck // ST1005: the text of languageservices, kept as is
47 errLongGroup = fmt.Errorf("concurrency group name must be less than %d characters", domain.CIMaxConcurrencyGroup)
48)
49
50// queueValue is the error of a queue that is neither single nor max. The schema
51// of workflow-parser allows only those ("concurrency-queue": {"allowed-values":
52// ["single", "max"]}), so a literal is refused when the workflow is parsed; an
53// expression that evaluates to another value gets the error of the template
54// reader of languageservices for a value not allowed
55// (workflow-parser/src/templates/template-reader.ts:444,
56// "this._context.error(literal, `Unexpected value '${literal.toString()}'`);").
57func queueValue(v string) error { return fmt.Errorf("Unexpected value '%s'", v) } //nolint:staticcheck // ST1005: the text of languageservices, kept as is
58
59// evalConcurrency evaluates the "concurrency:" token of a workflow or a job:
60// a string (or an expression) is the group; a mapping has group,
61// cancel-in-progress (an expression is allowed) and queue (single or max). An
62// empty group is an error, as in GitHub's runner.
63func evalConcurrency(ctx context.Context, tok template.Token, env template.Env) (domain.CIConcurrency, error) {
64 var out domain.CIConcurrency
65 if tok == nil {
66 return out, nil
67 }
68 groupTok := tok
69 if m, ok := tok.(*template.Mapping); ok {
70 if q := m.Find("queue"); q != nil {
71 queue, err := template.EvaluateString(ctx, q, env)
72 if err != nil {
73 return out, err
74 }
75 switch queue {
76 case "single":
77 case "max":
78 out.Max = true
79 default:
80 return out, queueValue(queue)
81 }
82 }
83 if groupTok = m.Find("group"); groupTok == nil {
84 return out, errEmptyGroup
85 }
86 if c := m.Find("cancel-in-progress"); c != nil {
87 cancel, err := template.EvaluateBool(ctx, c, env)
88 if err != nil {
89 return out, err
90 }
91 out.Cancel = cancel
92 }
93 // languageservices, languageservice/src/validate.ts:702-743, refuses the
94 // literal pair; ermite refuses it with the same message whatever wrote
95 // either value (an expression that evaluates to true included).
96 if out.Max && out.Cancel {
97 return domain.CIConcurrency{}, errQueueMaxCancel
98 }
99 }
100 group, err := template.EvaluateString(ctx, groupTok, env)
101 switch {
102 case err != nil:
103 return out, err
104 case group == "":
105 return out, errEmptyGroup
106 case domain.ConcurrencyGroupLength(group) > domain.CIMaxConcurrencyGroup:
107 return out, errLongGroup
108 }
109 out.Group = group
110 return out, nil
111}
112
113// later runs fn in the background, after the lock of the caller's run is
114// released. Close waits for it.
115func (c *CI) later(what string, fn func(ctx context.Context) error) {
116 c.mu.Lock()
117 if c.closed {
118 c.mu.Unlock()
119 return
120 }
121 c.wg.Add(1)
122 c.mu.Unlock()
123 go func() {
124 defer c.wg.Done()
125 c.guard(what, func() {
126 if err := fn(c.ctx); err != nil && c.ctx.Err() == nil && !errors.Is(err, domain.ErrNotFound) {
127 ctxLog(c.ctx).Error().Err(err).Str("worker", what).Msg("ci concurrency")
128 }
129 })
130 }()
131}
132
133// entryAlive reports whether the run or job of an entry can still hold or want
134// its group: it exists, and it has not ended (a run: it is on the attempt of the
135// entry and has not completed). Run ids are never reused, so a missing run or
136// job is gone for good. A read that fails says alive: nothing is dropped on a
137// doubt.
138func (c *CI) entryAlive(ctx context.Context, repo domain.RepoName, e domain.CIConcurrencyEntry) bool {
139 if e.JobID == "" {
140 run, err := c.cfg.Store.GetRun(ctx, repo, e.RunID)
141 if err != nil {
142 return !errors.Is(err, domain.ErrNotFound)
143 }
144 return run.Attempt == e.Attempt && run.Status != domain.CICompleted
145 }
146 j, err := c.cfg.Store.GetJob(ctx, e.RunID, e.Attempt, e.JobID)
147 switch {
148 case errors.Is(err, domain.ErrNotFound):
149 // A matrix leg that took its group is written with the flush that
150 // follows: without its record yet, it lives as long as its run's attempt.
151 return c.entryAlive(ctx, repo, domain.CIConcurrencyEntry{RunID: e.RunID, Attempt: e.Attempt})
152 case err != nil:
153 return true
154 }
155 return !j.Ended()
156}
157
158// mark sets the mark of an entry.
159func mark(st *domain.CIConcurrencyGroupState, e domain.CIConcurrencyEntry, group string, state domain.CIConcurrencyState) {
160 if st.Marks == nil {
161 st.Marks = map[string]domain.CIConcurrencyMark{}
162 }
163 st.Marks[e.Key()] = domain.CIConcurrencyMark{Entry: e, Group: group, State: state}
164}
165
166// promote gives the running place, when it is free, to the head of the line.
167func promote(st *domain.CIConcurrencyGroupState, group string) {
168 if head := st.Record.Promote(); head != nil {
169 m := st.Marks[head.Key()]
170 mark(st, *head, firstNonEmpty(m.Group, group), domain.CIConcurrencyHolding)
171 }
172}
173
174func firstNonEmpty(a, b string) string {
175 if a != "" {
176 return a
177 }
178 return b
179}
180
181// prune brings a group back to its invariant, in the transaction that is about to
182// decide on it: the entries whose run or job is gone or over leave the record
183// and lose their marks (a crash between the write of a job and the release of
184// its group, a run that was deleted), the others waiting in the same order; the
185// head of the line takes a free running place; an entry named by the record has
186// its mark; and an entry marked as pending or holding that the record does not
187// name any more is marked replaced (it should not happen, the batch writes both,
188// and it is cancelled if it does).
189func (c *CI) prune(ctx context.Context, repo domain.RepoName, group string, st *domain.CIConcurrencyGroupState) (more bool) {
190 rec := &st.Record
191 if rec.Running != nil && !c.entryAlive(ctx, repo, *rec.Running) {
192 delete(st.Marks, rec.Running.Key())
193 rec.Running = nil
194 }
195 alive := rec.Pending[:0:0]
196 for _, p := range rec.Pending {
197 if c.entryAlive(ctx, repo, p) {
198 alive = append(alive, p)
199 } else {
200 delete(st.Marks, p.Key())
201 }
202 }
203 if len(alive) == 0 {
204 alive = nil
205 }
206 rec.Pending = alive
207 promote(st, group)
208 if rec.Running != nil {
209 if m, ok := st.Marks[rec.Running.Key()]; !ok || m.State != domain.CIConcurrencyHolding {
210 mark(st, *rec.Running, firstNonEmpty(m.Group, group), domain.CIConcurrencyHolding)
211 }
212 }
213 for _, p := range rec.Pending {
214 if m, ok := st.Marks[p.Key()]; !ok || m.State != domain.CIConcurrencyPending {
215 mark(st, p, firstNonEmpty(m.Group, group), domain.CIConcurrencyPending)
216 }
217 }
218 // The other marks are pruned a bounded number per transaction, in key order:
219 // a batch has a size limit, and a group with a long backlog of stale marks
220 // must not wedge on one transaction that deletes them all. What is left is
221 // reported, and the next decision or the reconciliation takes the next ones.
222 left := c.cfg.ConcurrencyPruneBatch
223 for _, k := range slices.Sorted(maps.Keys(st.Marks)) {
224 m := st.Marks[k]
225 var prune, replace bool
226 switch {
227 case !c.entryAlive(ctx, repo, m.Entry):
228 prune = true
229 case !m.State.Owed() && !rec.Names(m.Entry):
230 replace = true
231 default:
232 continue
233 }
234 if left <= 0 {
235 return true
236 }
237 left--
238 if prune {
239 delete(st.Marks, k)
240 } else if replace {
241 m.State = domain.CIConcurrencyReplaced
242 st.Marks[k] = m
243 }
244 }
245 return false
246}
247
248// updateGroup is Store.UpdateConcurrency over the pruned group. It returns the
249// entry that holds the group after the change and did not before (the waiting
250// entry that was given the turn), or nil: its run is to be planned again.
251func (c *CI) updateGroup(ctx context.Context, repo domain.RepoName, group string, fn func(*domain.CIConcurrencyGroupState) error) (*domain.CIConcurrencyEntry, error) {
252 promoted, _, err := c.updateGroupMore(ctx, repo, group, fn)
253 return promoted, err
254}
255
256// updateGroupMore is updateGroup that also says whether the pruning stopped at
257// its bound, with stale marks left for another transaction.
258func (c *CI) updateGroupMore(ctx context.Context, repo domain.RepoName, group string, fn func(*domain.CIConcurrencyGroupState) error) (promoted *domain.CIConcurrencyEntry, more bool, err error) {
259 err = c.cfg.Store.UpdateConcurrency(ctx, repo, group, func(st *domain.CIConcurrencyGroupState) error {
260 promoted = nil
261 var was *domain.CIConcurrencyEntry
262 if r := st.Record.Running; r != nil {
263 e := *r
264 was = &e
265 }
266 more = c.prune(ctx, repo, group, st)
267 if err := fn(st); err != nil {
268 return err
269 }
270 if r := st.Record.Running; r != nil && (was == nil || !was.Same(*r)) {
271 e := *r
272 promoted = &e
273 }
274 return nil
275 })
276 return promoted, more, err
277}
278
279// concurrencyRetries bounds the retries of a delivery that failed; the delay
280// doubles from the watchdog tick up to maxConcurrencyBackoff.
281const (
282 defaultConcurrencyRetries = 8
283 maxConcurrencyBackoff = 30 * time.Second
284)
285
286// retry calls fn until it succeeds, at most concurrencyRetries more times, with
287// a doubling delay that ends with ctx. A domain.ErrNotFound is not retried.
288func (c *CI) retry(ctx context.Context, fn func() error) error {
289 delay := c.cfg.Tick
290 err := fn()
291 retries := c.cfg.ConcurrencyRetries
292 switch {
293 case retries == 0:
294 retries = defaultConcurrencyRetries
295 case retries < 0:
296 retries = 0
297 }
298 for range retries {
299 if err == nil || errors.Is(err, domain.ErrNotFound) {
300 return err
301 }
302 select {
303 case <-ctx.Done():
304 return errors.Join(err, ctx.Err())
305 case <-time.After(delay):
306 }
307 delay = min(delay*2, maxConcurrencyBackoff)
308 err = fn()
309 }
310 return err
311}
312
313// wake plans again the run of an entry that was given its group, outside any
314// run lock, and again if that failed. A run that is gone passes the group on.
315// What the retries do not reach, the reconciliation (startup, and every
316// minute) does: it advances every holder whose own record still says pending.
317func (c *CI) wake(repo domain.RepoName, group string, e domain.CIConcurrencyEntry) {
318 c.later("concurrency promote", func(ctx context.Context) error {
319 return c.retry(ctx, func() error {
320 err := c.advanceRun(ctx, e.RunID)
321 if errors.Is(err, domain.ErrNotFound) {
322 return c.releaseEntry(ctx, repo, group, e)
323 }
324 return err
325 })
326 })
327}
328
329// cancelLater cancels an entry that a newer one replaced (why is
330// domain.CIConcurrencyReplaced) or that found the line full (why is
331// domain.CIConcurrencyRejected), outside any run lock, and again if that
332// failed. The decision is in the group (the entry is marked replaced or
333// rejected in the batch that decided it, or stays named with Canceling set):
334// the reconciliation finds it from there if this cancellation is lost (crash,
335// retries used up).
336func (c *CI) cancelLater(group string, e domain.CIConcurrencyEntry, why domain.CIConcurrencyState) {
337 c.later("concurrency cancel", func(ctx context.Context) error {
338 return c.retry(ctx, func() error { return c.cancelEntry(ctx, e, group, why) })
339 })
340}
341
342// entryNeeds reports whether the entry named by its group still has to be told
343// where it stands: its own record does not say want yet (a promotion whose
344// planning failed or was lost, or a crash between the write of the group and the
345// write of the entry). An entry that is gone or over is the pruning's business.
346func (c *CI) entryNeeds(ctx context.Context, repo domain.RepoName, group string, e domain.CIConcurrencyEntry, want domain.CIConcurrencyState) bool {
347 if e.JobID == "" {
348 run, err := c.cfg.Store.GetRun(ctx, repo, e.RunID)
349 return err == nil && run.Status != domain.CICompleted && run.Attempt == e.Attempt && run.Concurrency.State != want
350 }
351 j, err := c.cfg.Store.GetJob(ctx, e.RunID, e.Attempt, e.JobID)
352 if errors.Is(err, domain.ErrNotFound) {
353 // A leg the group names whose record is not written: the run has to be
354 // planned, which expands its matrix, while its attempt is open.
355 run, err := c.cfg.Store.GetRun(ctx, repo, e.RunID)
356 return err == nil && run.Status != domain.CICompleted && run.Attempt == e.Attempt
357 }
358 if err != nil || j.Ended() || j.Phase != domain.CIPhaseWaiting {
359 return false
360 }
361 // A call holds two groups, its own and its callee's: the state to compare
362 // is the one of the group asked about.
363 if j.Call != nil && domain.ConcurrencyKey(j.Call.Concurrency.Group) == domain.ConcurrencyKey(group) &&
364 domain.ConcurrencyKey(j.Concurrency.Group) != domain.ConcurrencyKey(group) {
365 return j.Call.Concurrency.State != want
366 }
367 return j.Concurrency.State != want
368}
369
370// ReconcileConcurrency repairs the groups. It runs at startup and every minute.
371// For each stored group it prunes the entries that are gone or over, asks again
372// the running entry a cancel-in-progress entry replaced, cancels every entry
373// marked replaced or rejected (their cancellation is owed), and plans the
374// entries the record names whose own record does not know it yet.
375//
376// Cost, per pass: a scan of the stored groups and their marks, and for each
377// group a version-checked transaction (a write only if something is repaired;
378// at most CIConfig.ConcurrencyPruneBatch stale marks go in one, the group is
379// taken again until none is left) plus a read of the run or job of each entry
380// the record or the marks name (at most CIConfig.ConcurrencyQueueMax waiting
381// ones per group). It grows with the groups in use, not with the open runs.
382func (c *CI) ReconcileConcurrency(ctx context.Context) {
383 log := ctxLog(ctx)
384 groups, err := c.cfg.Store.ListConcurrency(ctx)
385 if err != nil {
386 log.Error().Err(err).Msg("ci: list the concurrency groups")
387 return
388 }
389 for _, g := range groups {
390 var st domain.CIConcurrencyGroupState
391 var err error
392 // One transaction prunes a bounded number of marks: go on until none is
393 // left to prune (each pass removes some, so this ends).
394 for more := true; more && err == nil && ctx.Err() == nil; {
395 _, more, err = c.updateGroupMore(ctx, g.Repo, g.Group, func(cur *domain.CIConcurrencyGroupState) error {
396 st = domain.CIConcurrencyGroupState{Record: cur.Record.Clone(), Marks: maps.Clone(cur.Marks)}
397 return nil
398 })
399 }
400 if err != nil {
401 if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
402 log.Error().Err(err).Str("group", g.Group).Msg("ci: repair a concurrency group")
403 }
404 continue
405 }
406 if r := st.Record.Running; r != nil {
407 if r.Canceling {
408 c.cancelLater(st.Marks[r.Key()].Group, *r, domain.CIConcurrencyReplaced)
409 }
410 if c.entryNeeds(ctx, g.Repo, g.Group, *r, domain.CIConcurrencyHolding) {
411 c.wake(g.Repo, g.Group, *r)
412 }
413 }
414 for _, p := range st.Record.Pending {
415 if c.entryNeeds(ctx, g.Repo, g.Group, p, domain.CIConcurrencyPending) {
416 c.wake(g.Repo, g.Group, p)
417 }
418 }
419 for _, m := range st.Marks {
420 if m.State.Owed() {
421 c.cancelLater(m.Group, m.Entry, m.State)
422 }
423 }
424 }
425}
426
427// acquire asks for the turn of an entry in a group and returns where it stands:
428// holding, pending behind the running entry, or replaced. The decision and the
429// marks of every entry it touches are one batch (Store.UpdateConcurrency), so an
430// entry is marked pending or holding if and only if the record names it.
431//
432// An entry marked replaced or rejected never acquires, whatever the state of
433// the group (empty, or its line free): it is owed a cancellation. Otherwise a
434// pending entry stays so, at its place in the line; a newer one is queued
435// (domain.CIConcurrencyRecord.Enqueue): with queue: single it replaces every
436// waiting entry, each marked replaced and cancelled, and with
437// cancel-in-progress the running entry is cancelled too (it stays named, with
438// Canceling set, until it has ended); with queue: max it joins the end of the
439// line, or, when the line is full, is marked rejected and cancelled itself.
440// The call is repeatable: an entry that holds or waits is told so again.
441func (rs *runState) acquire(ctx context.Context, group string, e domain.CIConcurrencyEntry) (domain.CIConcurrencyState, error) {
442 type victim struct {
443 e domain.CIConcurrencyEntry
444 why domain.CIConcurrencyState
445 }
446 var state domain.CIConcurrencyState
447 var victims []victim
448 promoted, err := rs.c.updateGroup(ctx, rs.run.Repo, group, func(st *domain.CIConcurrencyGroupState) error {
449 victims = nil
450 rec := &st.Record
451 if m, ok := st.Marks[e.Key()]; ok && m.State.Owed() {
452 state = domain.CIConcurrencyReplaced
453 victims = append(victims, victim{e, m.State}) // its cancellation may have been lost
454 return nil
455 }
456 switch {
457 case rec.Running == nil:
458 me := e
459 rec.Running = &me
460 rec.RemovePending(e)
461 mark(st, e, group, domain.CIConcurrencyHolding)
462 state = domain.CIConcurrencyHolding
463 case rec.Running.Same(e):
464 state = domain.CIConcurrencyHolding
465 case rec.PendingIndex(e) >= 0:
466 state = domain.CIConcurrencyPending
467 default:
468 replaced, rejected := rec.Enqueue(e, rs.c.cfg.ConcurrencyQueueMax)
469 if rejected {
470 state = domain.CIConcurrencyReplaced
471 mark(st, e, group, domain.CIConcurrencyRejected)
472 victims = append(victims, victim{e, domain.CIConcurrencyRejected})
473 return nil
474 }
475 state = domain.CIConcurrencyPending
476 for _, old := range replaced {
477 mark(st, old, group, domain.CIConcurrencyReplaced)
478 victims = append(victims, victim{old, domain.CIConcurrencyReplaced})
479 }
480 mark(st, e, group, domain.CIConcurrencyPending)
481 if e.Cancel && !rec.Running.Canceling {
482 run := *rec.Running
483 run.Canceling = true
484 rec.Running = &run
485 victims = append(victims, victim{run, domain.CIConcurrencyReplaced})
486 }
487 }
488 return nil
489 })
490 if err != nil {
491 return "", err
492 }
493 if promoted != nil && !promoted.Same(e) {
494 rs.c.wake(rs.run.Repo, group, *promoted)
495 }
496 for _, v := range victims {
497 rs.c.cancelLater(group, v.e, v.why)
498 }
499 if state == domain.CIConcurrencyReplaced {
500 state = domain.CIConcurrencyPending // it shows as waiting until its cancellation lands
501 }
502 return state, nil
503}
504
505// releaseEntry takes an entry out of its group, because its run or job has
506// ended, and gives the turn to the head of the line, which is planned again; a
507// waiting entry that leaves keeps the order of the others. The mark of the
508// entry goes in the same batch.
509func (c *CI) releaseEntry(ctx context.Context, repo domain.RepoName, group string, e domain.CIConcurrencyEntry) error {
510 promoted, err := c.updateGroup(ctx, repo, group, func(st *domain.CIConcurrencyGroupState) error {
511 switch {
512 case st.Record.Running != nil && st.Record.Running.Same(e):
513 st.Record.Running = nil
514 promote(st, group)
515 default:
516 st.Record.RemovePending(e)
517 }
518 delete(st.Marks, e.Key())
519 return nil
520 })
521 if err != nil || promoted == nil {
522 return err
523 }
524 c.wake(repo, group, *promoted)
525 return nil
526}
527
528// cancelEntry cancels a run or a job that a newer entry of the group replaced,
529// or that found the line full (why is domain.CIConcurrencyRejected): it ends
530// cancelled with GitHub's annotation (ermite's for a full line), and a job a
531// runner has gets the cancel request the owner's cancel sends.
532func (c *CI) cancelEntry(ctx context.Context, e domain.CIConcurrencyEntry, group string, why domain.CIConcurrencyState) error {
533 return c.withRun(ctx, e.RunID, e.Attempt, func(rs *runState) error {
534 if group == "" { // no name in the mark: use the entry's own group
535 group = rs.run.Concurrency.Group
536 if j := rs.find(e.JobID); e.JobID != "" && j != nil {
537 group = j.Concurrency.Group
538 }
539 }
540 msg := domain.ConcurrencyCancelMessage(group)
541 if why == domain.CIConcurrencyRejected {
542 msg = domain.ConcurrencyQueueFullMessage(group)
543 }
544 if e.JobID == "" {
545 if !rs.latest() || rs.run.Status == domain.CICompleted || rs.run.CancelRequested {
546 return nil
547 }
548 rs.run.CancelRequested, rs.run.CancelRequestedAt, rs.runDirty = true, c.now(), true
549 rs.run.Concurrency.Preempted = true
550 for i := range rs.jobs {
551 if !rs.jobs[i].Ended() {
552 rs.cancelForGroup(&rs.jobs[i], msg)
553 }
554 }
555 if err := rs.flush(ctx); err != nil {
556 return err
557 }
558 // The flush frees the group the run's own record names. A run whose
559 // writes of its group were lost names none: free the group that
560 // decided the cancel, and the mark with it.
561 if rs.run.Status == domain.CICompleted && domain.ConcurrencyKey(rs.run.Concurrency.Group) != domain.ConcurrencyKey(group) {
562 if err := c.releaseEntry(ctx, rs.run.Repo, group, e); err != nil && !errors.Is(err, domain.ErrNotFound) {
563 return err
564 }
565 }
566 return nil
567 }
568 j := rs.find(e.JobID)
569 if j == nil || !j.Planned {
570 // The group names a leg whose record the run has not written yet (the
571 // flush that followed the planning was lost): the record found under
572 // the id of leg 0 is the placeholder of the whole job, not the leg.
573 // Plan the job first, so that every leg exists, and only then act on
574 // the one that was targeted.
575 if err := rs.advance(ctx); err != nil {
576 return err
577 }
578 if j = rs.find(e.JobID); j == nil || !j.Planned {
579 return rs.flush(ctx) // not plannable (yet): nothing to cancel, keep what advance did
580 }
581 }
582 if j.Ended() || j.Concurrency.Preempted {
583 return rs.flush(ctx) // over, or cancelled by an earlier delivery of the same item
584 }
585 rs.cancelForGroup(j, msg)
586 if err := rs.advance(ctx); err != nil {
587 return err
588 }
589 return rs.flush(ctx)
590 })
591}
592
593// cancelForGroup ends a job a newer entry of its group replaced (or that found
594// the line full), or asks its runner to stop, with the annotation msg.
595func (rs *runState) cancelForGroup(j *domain.CIJob, msg string) {
596 j.Concurrency.Preempted = true
597 rs.annotations = append(rs.annotations, ports.CIStepAnnotations{
598 Repo: j.Repo, Exec: logRef(*j), Step: 1,
599 Items: []domain.CIAnnotation{{Level: domain.AnnotationWarning, Message: msg, Step: 1}},
600 })
601 if j.Active() {
602 rs.requestCancel(j, domain.CICancelUser, msg)
603 rs.touch(j)
604 return
605 }
606 rs.end(j, domain.CICancelled, msg)
607}
608
609// acquireRun evaluates the workflow-level concurrency of the attempt once and
610// asks for its turn. It returns whether the run may go on: a pending run
611// plans nothing, and its jobs show the pending status.
612func (rs *runState) acquireRun(ctx context.Context, wf *workflow.Workflow) (bool, error) {
613 run := &rs.run
614 if !slices.ContainsFunc(rs.jobs, func(j domain.CIJob) bool { return !j.Ended() }) {
615 return true, nil // nothing left to run: the group is freed at the flush
616 }
617 if !run.Concurrency.Evaluated {
618 conc, err := rs.evalRunConcurrency(ctx, wf)
619 if err != nil {
620 for i := range rs.jobs {
621 if !rs.jobs[i].Ended() && !rs.jobs[i].Active() {
622 rs.end(&rs.jobs[i], domain.CIFailure, err.Error())
623 }
624 }
625 run.Concurrency, rs.runDirty = domain.CIConcurrency{Evaluated: true}, true
626 return false, nil
627 }
628 conc.Evaluated = true
629 run.Concurrency, rs.runDirty = conc, true
630 }
631 if run.Concurrency.Group == "" {
632 return true, nil
633 }
634 st, err := rs.acquire(ctx, run.Concurrency.Group, domain.CIConcurrencyEntry{RunID: run.ID, Attempt: run.Attempt, Cancel: run.Concurrency.Cancel, Max: run.Concurrency.Max})
635 if err != nil {
636 return false, err
637 }
638 if st != run.Concurrency.State {
639 run.Concurrency.State, rs.runDirty = st, true
640 }
641 want := run.Concurrency
642 want.FromRun, want.Preempted = true, false
643 for i := range rs.jobs {
644 j := &rs.jobs[i]
645 if j.Ended() || (j.Concurrency.Group != "" && !j.Concurrency.FromRun) {
646 continue // a job with a group of its own shows that one
647 }
648 if j.Concurrency != want {
649 j.Concurrency = want
650 rs.touch(j)
651 }
652 }
653 return st == domain.CIConcurrencyHolding, nil
654}
655
656func (rs *runState) evalRunConcurrency(ctx context.Context, wf *workflow.Workflow) (domain.CIConcurrency, error) {
657 if wf.Concurrency == nil {
658 return domain.CIConcurrency{}, nil
659 }
660 gh, err := rs.githubValue("", rs.run.Attempt)
661 if err != nil {
662 return domain.CIConcurrency{}, err
663 }
664 vars, err := rs.c.varsContext(ctx, rs.run.Repo)
665 if err != nil {
666 return domain.CIConcurrency{}, err
667 }
668 env := template.Env{Contexts: map[string]expr.Value{"github": gh, "inputs": rs.inputsOf(domain.CIJob{}), "vars": vars}}
669 return evalConcurrency(ctx, wf.Concurrency, env)
670}
671
672// releaseEnded frees the groups of what ended in the write that was just
673// applied: the jobs that hold one of their own, and the run itself when it
674// completed.
675func (rs *runState) releaseEnded(ctx context.Context, changed []int, runDone bool) {
676 c := rs.c
677 for _, i := range changed {
678 j := rs.jobs[i]
679 if !j.Ended() {
680 continue
681 }
682 c.artifactsJobEnded(ctx, j)
683 e := domain.CIConcurrencyEntry{RunID: j.RunID, Attempt: j.RunAttempt, JobID: j.ID}
684 groups := []string{}
685 if j.Concurrency.Group != "" && !j.Concurrency.FromRun {
686 groups = append(groups, j.Concurrency.Group)
687 }
688 if j.Call != nil && j.Call.Concurrency.Group != "" {
689 groups = append(groups, j.Call.Concurrency.Group) // a call ends holding both
690 }
691 for _, g := range groups {
692 if err := c.releaseEntry(ctx, rs.run.Repo, g, e); err != nil && !errors.Is(err, domain.ErrNotFound) {
693 ctxLog(ctx).Error().Err(err).Str("job_id", j.ID).Msg("ci: free a concurrency group")
694 }
695 }
696 }
697 if runDone && rs.run.Concurrency.Group != "" {
698 e := domain.CIConcurrencyEntry{RunID: rs.run.ID, Attempt: rs.run.Attempt}
699 if err := c.releaseEntry(ctx, rs.run.Repo, rs.run.Concurrency.Group, e); err != nil && !errors.Is(err, domain.ErrNotFound) {
700 ctxLog(ctx).Error().Err(err).Int64("run_id", rs.run.ID).Msg("ci: free a concurrency group")
701 }
702 }
703}