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, 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 return err == nil && !j.Ended() && j.Phase == domain.CIPhaseWaiting && j.Concurrency.State != want
329}
330
331// ReconcileConcurrency repairs the groups. It runs at startup and every minute.
332// For each stored group it prunes the entries that are gone or over, asks again
333// the running entry a cancel-in-progress entry replaced, cancels every entry
334// marked replaced (their cancellation is owed), and plans the entries the record
335// names whose own record does not know it yet.
336//
337// Cost, per pass: a scan of the stored groups and their marks, and for each
338// group a version-checked transaction (a write only if something is repaired;
339// at most CIConfig.ConcurrencyPruneBatch stale marks go in one, the group is
340// taken again until none is left) plus a read of the run or job of each entry
341// the record or the marks name. It grows with the groups in use, not with the
342// open runs.
343func (c *CI) ReconcileConcurrency(ctx context.Context) {
344 log := ctxLog(ctx)
345 groups, err := c.cfg.Store.ListConcurrency(ctx)
346 if err != nil {
347 log.Error().Err(err).Msg("ci: list the concurrency groups")
348 return
349 }
350 for _, g := range groups {
351 var st domain.CIConcurrencyGroupState
352 var err error
353 // One transaction prunes a bounded number of marks: go on until none is
354 // left to prune (each pass removes some, so this ends).
355 for more := true; more && err == nil && ctx.Err() == nil; {
356 _, more, err = c.updateGroupMore(ctx, g.Repo, g.Group, func(cur *domain.CIConcurrencyGroupState) error {
357 st = domain.CIConcurrencyGroupState{Marks: maps.Clone(cur.Marks)}
358 if cur.Record.Running != nil {
359 r := *cur.Record.Running
360 st.Record.Running = &r
361 }
362 if cur.Record.Pending != nil {
363 p := *cur.Record.Pending
364 st.Record.Pending = &p
365 }
366 return nil
367 })
368 }
369 if err != nil {
370 if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
371 log.Error().Err(err).Str("group", g.Group).Msg("ci: repair a concurrency group")
372 }
373 continue
374 }
375 if r := st.Record.Running; r != nil {
376 if r.Canceling {
377 c.cancelLater(st.Marks[r.Key()].Group, *r)
378 }
379 if c.entryNeeds(ctx, g.Repo, *r, domain.CIConcurrencyHolding) {
380 c.wake(g.Repo, g.Group, *r)
381 }
382 }
383 if p := st.Record.Pending; p != nil && c.entryNeeds(ctx, g.Repo, *p, domain.CIConcurrencyPending) {
384 c.wake(g.Repo, g.Group, *p)
385 }
386 for _, m := range st.Marks {
387 if m.State == domain.CIConcurrencyReplaced {
388 c.cancelLater(m.Group, m.Entry)
389 }
390 }
391 }
392}
393
394// acquire asks for the turn of an entry in a group and returns where it stands:
395// holding, pending behind the running entry, or replaced. The decision and the
396// marks of every entry it touches are one batch (Store.UpdateConcurrency), so an
397// entry is marked pending or holding if and only if the record names it.
398//
399// An entry marked replaced never acquires, whatever the state of the group
400// (empty, or its waiting place free): it is owed a cancellation. Otherwise a
401// pending entry stays so; a newer one replaces the waiting entry, which is
402// marked replaced and cancelled, and with cancel-in-progress the running entry
403// is cancelled too (it stays named, with Canceling set, until it has ended).
404// The call is repeatable: an entry that holds or waits is told so again.
405func (rs *runState) acquire(ctx context.Context, group string, e domain.CIConcurrencyEntry) (domain.CIConcurrencyState, error) {
406 var state domain.CIConcurrencyState
407 var victims []domain.CIConcurrencyEntry
408 promoted, err := rs.c.updateGroup(ctx, rs.run.Repo, group, func(st *domain.CIConcurrencyGroupState) error {
409 victims = nil
410 rec := &st.Record
411 if m, ok := st.Marks[e.Key()]; ok && m.State == domain.CIConcurrencyReplaced {
412 state = domain.CIConcurrencyReplaced
413 victims = append(victims, e) // its cancellation may have been lost
414 return nil
415 }
416 switch {
417 case rec.Running == nil:
418 me := e
419 rec.Running = &me
420 if rec.Pending != nil && rec.Pending.Same(e) {
421 rec.Pending = nil
422 }
423 mark(st, e, group, domain.CIConcurrencyHolding)
424 state = domain.CIConcurrencyHolding
425 case rec.Running.Same(e):
426 state = domain.CIConcurrencyHolding
427 case rec.Pending != nil && rec.Pending.Same(e):
428 state = domain.CIConcurrencyPending
429 default:
430 state = domain.CIConcurrencyPending
431 if rec.Pending != nil {
432 old := *rec.Pending
433 mark(st, old, group, domain.CIConcurrencyReplaced)
434 victims = append(victims, old)
435 }
436 me := e
437 rec.Pending = &me
438 mark(st, e, group, domain.CIConcurrencyPending)
439 if e.Cancel && !rec.Running.Canceling {
440 run := *rec.Running
441 run.Canceling = true
442 rec.Running = &run
443 victims = append(victims, run)
444 }
445 }
446 return nil
447 })
448 if err != nil {
449 return "", err
450 }
451 if promoted != nil && !promoted.Same(e) {
452 rs.c.wake(rs.run.Repo, group, *promoted)
453 }
454 for _, v := range victims {
455 rs.c.cancelLater(group, v)
456 }
457 if state == domain.CIConcurrencyReplaced {
458 state = domain.CIConcurrencyPending // it shows as waiting until its cancellation lands
459 }
460 return state, nil
461}
462
463// releaseEntry takes an entry out of its group, because its run or job has
464// ended, and gives the turn to the waiting entry, which is planned again. The
465// mark of the entry goes in the same batch.
466func (c *CI) releaseEntry(ctx context.Context, repo domain.RepoName, group string, e domain.CIConcurrencyEntry) error {
467 promoted, err := c.updateGroup(ctx, repo, group, func(st *domain.CIConcurrencyGroupState) error {
468 switch {
469 case st.Record.Running != nil && st.Record.Running.Same(e):
470 st.Record.Running = nil
471 promote(st, group)
472 case st.Record.Pending != nil && st.Record.Pending.Same(e):
473 st.Record.Pending = nil
474 }
475 delete(st.Marks, e.Key())
476 return nil
477 })
478 if err != nil || promoted == nil {
479 return err
480 }
481 c.wake(repo, group, *promoted)
482 return nil
483}
484
485// cancelEntry cancels a run or a job that a newer entry of the group
486// replaced: it ends cancelled with GitHub's annotation, and a job a runner has
487// gets the cancel request the owner's cancel sends.
488func (c *CI) cancelEntry(ctx context.Context, e domain.CIConcurrencyEntry, group string) error {
489 return c.withRun(ctx, e.RunID, e.Attempt, func(rs *runState) error {
490 if group == "" { // no name in the mark: use the entry's own group
491 group = rs.run.Concurrency.Group
492 if j := rs.find(e.JobID); e.JobID != "" && j != nil {
493 group = j.Concurrency.Group
494 }
495 }
496 msg := domain.ConcurrencyCancelMessage(group)
497 if e.JobID == "" {
498 if !rs.latest() || rs.run.Status == domain.CICompleted || rs.run.CancelRequested {
499 return nil
500 }
501 rs.run.CancelRequested, rs.run.CancelRequestedAt, rs.runDirty = true, c.now(), true
502 rs.run.Concurrency.Preempted = true
503 for i := range rs.jobs {
504 if !rs.jobs[i].Ended() {
505 rs.cancelForGroup(&rs.jobs[i], msg)
506 }
507 }
508 return rs.flush(ctx)
509 }
510 j := rs.find(e.JobID)
511 if j == nil || !j.Planned {
512 // The group names a leg whose record the run has not written yet (the
513 // flush that followed the planning was lost): the record found under
514 // the id of leg 0 is the placeholder of the whole job, not the leg.
515 // Plan the job first, so that every leg exists, and only then act on
516 // the one that was targeted.
517 if err := rs.advance(ctx); err != nil {
518 return err
519 }
520 if j = rs.find(e.JobID); j == nil || !j.Planned {
521 return rs.flush(ctx) // not plannable (yet): nothing to cancel, keep what advance did
522 }
523 }
524 if j.Ended() || j.Concurrency.Preempted {
525 return rs.flush(ctx) // over, or cancelled by an earlier delivery of the same item
526 }
527 rs.cancelForGroup(j, msg)
528 if err := rs.advance(ctx); err != nil {
529 return err
530 }
531 return rs.flush(ctx)
532 })
533}
534
535// cancelForGroup ends a job a newer entry of its group replaced, or asks its
536// runner to stop. The annotation is GitHub's.
537func (rs *runState) cancelForGroup(j *domain.CIJob, msg string) {
538 j.Concurrency.Preempted = true
539 rs.annotations = append(rs.annotations, ports.CIStepAnnotations{
540 Repo: j.Repo, Exec: logRef(*j), Step: 1,
541 Items: []domain.CIAnnotation{{Level: domain.AnnotationWarning, Message: msg, Step: 1}},
542 })
543 if j.Active() {
544 rs.requestCancel(j, domain.CICancelUser, msg)
545 rs.touch(j)
546 return
547 }
548 rs.end(j, domain.CICancelled, msg)
549}
550
551// acquireRun evaluates the workflow-level concurrency of the attempt once and
552// asks for its turn. It returns whether the run may go on: a pending run
553// plans nothing, and its jobs show the pending status.
554func (rs *runState) acquireRun(ctx context.Context, wf *workflow.Workflow) (bool, error) {
555 run := &rs.run
556 if !slices.ContainsFunc(rs.jobs, func(j domain.CIJob) bool { return !j.Ended() }) {
557 return true, nil // nothing left to run: the group is freed at the flush
558 }
559 if !run.Concurrency.Evaluated {
560 conc, err := rs.evalRunConcurrency(ctx, wf)
561 if err != nil {
562 for i := range rs.jobs {
563 if !rs.jobs[i].Ended() && !rs.jobs[i].Active() {
564 rs.end(&rs.jobs[i], domain.CIFailure, err.Error())
565 }
566 }
567 run.Concurrency, rs.runDirty = domain.CIConcurrency{Evaluated: true}, true
568 return false, nil
569 }
570 conc.Evaluated = true
571 run.Concurrency, rs.runDirty = conc, true
572 }
573 if run.Concurrency.Group == "" {
574 return true, nil
575 }
576 st, err := rs.acquire(ctx, run.Concurrency.Group, domain.CIConcurrencyEntry{RunID: run.ID, Attempt: run.Attempt, Cancel: run.Concurrency.Cancel})
577 if err != nil {
578 return false, err
579 }
580 if st != run.Concurrency.State {
581 run.Concurrency.State, rs.runDirty = st, true
582 }
583 want := run.Concurrency
584 want.FromRun, want.Preempted = true, false
585 for i := range rs.jobs {
586 j := &rs.jobs[i]
587 if j.Ended() || (j.Concurrency.Group != "" && !j.Concurrency.FromRun) {
588 continue // a job with a group of its own shows that one
589 }
590 if j.Concurrency != want {
591 j.Concurrency = want
592 rs.touch(j)
593 }
594 }
595 return st == domain.CIConcurrencyHolding, nil
596}
597
598func (rs *runState) evalRunConcurrency(ctx context.Context, wf *workflow.Workflow) (domain.CIConcurrency, error) {
599 if wf.Concurrency == nil {
600 return domain.CIConcurrency{}, nil
601 }
602 gh, err := rs.githubValue("", rs.run.Attempt)
603 if err != nil {
604 return domain.CIConcurrency{}, err
605 }
606 vars, err := rs.c.varsContext(ctx, rs.run.Repo)
607 if err != nil {
608 return domain.CIConcurrency{}, err
609 }
610 env := template.Env{Contexts: map[string]expr.Value{"github": gh, "inputs": expr.NewDictionary(), "vars": vars}}
611 return evalConcurrency(ctx, wf.Concurrency, env)
612}
613
614// releaseEnded frees the groups of what ended in the write that was just
615// applied: the jobs that hold one of their own, and the run itself when it
616// completed.
617func (rs *runState) releaseEnded(ctx context.Context, changed []int, runDone bool) {
618 c := rs.c
619 for _, i := range changed {
620 j := rs.jobs[i]
621 if !j.Ended() || j.Concurrency.Group == "" || j.Concurrency.FromRun {
622 continue
623 }
624 e := domain.CIConcurrencyEntry{RunID: j.RunID, Attempt: j.RunAttempt, JobID: j.ID}
625 if err := c.releaseEntry(ctx, rs.run.Repo, j.Concurrency.Group, e); err != nil && !errors.Is(err, domain.ErrNotFound) {
626 ctxLog(ctx).Error().Err(err).Str("job_id", j.ID).Msg("ci: free a concurrency group")
627 }
628 }
629 if runDone && rs.run.Concurrency.Group != "" {
630 e := domain.CIConcurrencyEntry{RunID: rs.run.ID, Attempt: rs.run.Attempt}
631 if err := c.releaseEntry(ctx, rs.run.Repo, rs.run.Concurrency.Group, e); err != nil && !errors.Is(err, domain.ErrNotFound) {
632 ctxLog(ctx).Error().Err(err).Int64("run_id", rs.run.ID).Msg("ci: free a concurrency group")
633 }
634 }
635}