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