1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "bytes"
7 "context"
8 "encoding/json"
9 "fmt"
10 "math"
11 "slices"
12 "strconv"
13 "strings"
14 "sync"
15
16 "bnema.dev/ermite-actions/expr"
17 "bnema.dev/ermite-actions/template"
18 "bnema.dev/ermite-actions/workflow"
19 "bnema.dev/ermite/internal/domain"
20)
21
22// maxJobError bounds JobError text (4 KiB, run.proto).
23const maxJobError = 4 << 10
24
25// wfCache keeps the parsed workflows of the runs in flight, so that planning a
26// job does not parse the file again.
27type wfCache struct {
28 mu sync.Mutex
29 max int
30 m map[int64]*workflow.Workflow
31 // uses keeps which secrets each job of a run refers to (the names only,
32 // never a value), per run then job key: the workflow file of a run does not
33 // change, so the walk of its YAML is done once per job.
34 uses map[int64]map[string]secretUse
35}
36
37func newWfCache(max int) *wfCache {
38 return &wfCache{max: max, m: map[int64]*workflow.Workflow{}, uses: map[int64]map[string]secretUse{}}
39}
40
41// secretUse returns the secrets the job refers to, walking the YAML once per
42// run and job.
43func (w *wfCache) secretUse(runID int64, yaml, jobKey string) secretUse {
44 w.mu.Lock()
45 if u, ok := w.uses[runID][jobKey]; ok {
46 w.mu.Unlock()
47 return u
48 }
49 w.mu.Unlock()
50 u := secretsUsedBy(yaml, jobKey)
51 w.mu.Lock()
52 defer w.mu.Unlock()
53 if w.uses[runID] == nil {
54 if len(w.uses) >= w.max {
55 clear(w.uses)
56 }
57 w.uses[runID] = map[string]secretUse{}
58 }
59 w.uses[runID][jobKey] = u
60 return u
61}
62
63func (w *wfCache) get(runID int64, path, yaml string) (*workflow.Workflow, error) {
64 w.mu.Lock()
65 if wf, ok := w.m[runID]; ok {
66 w.mu.Unlock()
67 return wf, nil
68 }
69 w.mu.Unlock()
70 wf, errs := workflow.Parse(path, []byte(yaml))
71 if len(errs) > 0 {
72 return nil, fmt.Errorf("the workflow file no longer parses: %s", errs[0].Error())
73 }
74 w.mu.Lock()
75 defer w.mu.Unlock()
76 if len(w.m) >= w.max {
77 clear(w.m)
78 }
79 w.m[runID] = wf
80 return wf, nil
81}
82
83// unsupported names the feature of a job that ermite does not run yet, or "".
84// Such a job fails with "not supported yet: <feature>" instead of running
85// wrong. CI-2 runs steps with uses:, job container:, services: and step
86// timeout-minutes (ADR 0025), CI-3 runs environment:; what stays is the rest of
87// CI-3.
88func unsupported(j *workflow.Job) string {
89 switch {
90 case j.IsReusable():
91 return "reusable workflows (uses at the job level)"
92 }
93 return ""
94}
95
96// instances returns the indexes of the records of the job key, in order.
97func (rs *runState) instances(key string) []int {
98 var out []int
99 for i, j := range rs.jobs {
100 if j.Key == key {
101 out = append(out, i)
102 }
103 }
104 return out
105}
106
107func (rs *runState) sortJobs() {
108 slices.SortStableFunc(rs.jobs, func(a, b domain.CIJob) int {
109 if a.Index != b.Index {
110 return a.Index - b.Index
111 }
112 return a.Combo - b.Combo
113 })
114}
115
116// result is the aggregate conclusion of the instances of a job as the needs
117// context shows it: success, failure, cancelled or skipped.
118func result(insts []domain.CIJob) string {
119 var failed, cancelled, ok bool
120 for _, j := range insts {
121 if j.ContinueOnError && j.Conclusion.IsFailure() {
122 ok = true // recorded: needs.<job>.result is success
123 continue
124 }
125 switch j.Conclusion {
126 case domain.CIFailure, domain.CITimedOut, domain.CIActionRequired:
127 failed = true
128 case domain.CICancelled:
129 cancelled = true
130 case domain.CISkipped:
131 default:
132 ok = true
133 }
134 }
135 switch {
136 case failed:
137 return "failure"
138 case cancelled:
139 return "cancelled"
140 case ok:
141 return "success"
142 }
143 return "skipped"
144}
145
146// needsOf computes the needs context and the status functions of a job from
147// the jobs it needs, and whether they all have ended.
148//
149// The needs context holds the direct needs only, but the status functions look
150// at the whole chain (wf.Ancestors): a job whose need ran through always()
151// after a failure further up is skipped by its default success() (recorded:
152// needs-failed-dependency, chain-after-always), and failure() is true when an
153// ancestor failed (documented, not recorded). An ancestor beyond the direct
154// needs only changes success() and failure(): cancelled() is not recorded for
155// them, and a skipped ancestor behind a need that did run is not either.
156func (rs *runState) needsOf(wf *workflow.Workflow, j *workflow.Job) (needs *expr.Dictionary, st workflow.StatusFuncs, ready bool) {
157 needs = expr.NewDictionary()
158 st = workflow.StatusFuncs{Success: true, Cancelled: rs.run.CancelRequested}
159 for _, n := range wf.Ancestors(j.ID) {
160 var insts []domain.CIJob
161 for _, i := range rs.instances(n) {
162 if rs.jobs[i].Ended() {
163 insts = append(insts, rs.jobs[i])
164 }
165 }
166 switch result(insts) {
167 case "failure":
168 st.Failure, st.Success = true, false
169 case "cancelled":
170 st.Success = false
171 }
172 }
173 for _, n := range j.Needs {
174 var insts []domain.CIJob
175 for _, i := range rs.instances(n) {
176 if !rs.jobs[i].Ended() {
177 return nil, st, false
178 }
179 insts = append(insts, rs.jobs[i])
180 }
181 if len(insts) == 0 {
182 return nil, st, false
183 }
184 res := result(insts)
185 switch res {
186 case "failure":
187 st.Failure, st.Success = true, false
188 case "cancelled":
189 st.Cancelled, st.Success = true, false
190 case "skipped":
191 st.Success = false
192 }
193 outs := expr.NewDictionary()
194 for _, in := range insts {
195 for _, k := range sortedKeys(in.Outputs) {
196 // Instances of a matrix share names: the last non-empty value wins.
197 if v := in.Outputs[k]; v != "" || !hasKey(outs, k) {
198 setKey(outs, k, expr.String(v))
199 }
200 }
201 }
202 d := expr.NewDictionary()
203 d.Add("result", expr.String(res))
204 d.Add("outputs", outs)
205 needs.Add(n, d)
206 }
207 return needs, st, true
208}
209
210func hasKey(d *expr.Dictionary, k string) bool { _, ok := d.Get(k); return ok }
211
212// setKey sets a key of a dictionary, replacing an existing value.
213func setKey(d *expr.Dictionary, k string, v expr.Value) {
214 if !d.Add(k, v) {
215 pairs := d.Pairs()
216 *d = *expr.NewDictionary()
217 for _, p := range pairs {
218 if strings.EqualFold(p.Key, k) {
219 d.Add(p.Key, v)
220 } else {
221 d.Add(p.Key, p.Value)
222 }
223 }
224 }
225}
226
227// githubValue is the github context of the run with the keys of the job.
228func (rs *runState) githubValue(key string, attempt int32) (*expr.Dictionary, error) {
229 v, err := expr.FromJSON([]byte(rs.src.GithubJSON))
230 if err != nil {
231 return nil, fmt.Errorf("the stored github context is not valid: %w", err)
232 }
233 d, ok := v.(*expr.Dictionary)
234 if !ok {
235 return nil, fmt.Errorf("the stored github context is not an object")
236 }
237 setKey(d, "job", expr.String(key))
238 setKey(d, "run_attempt", expr.String(strconv.Itoa(int(attempt))))
239 return d, nil
240}
241
242// trim cuts a message for a job error.
243func trim(s string) string { return cutText(s, maxJobError) }
244
245// end ends a job with a conclusion and an error text, completes its steps like
246// the server does, and applies fail-fast to its siblings.
247func (rs *runState) end(j *domain.CIJob, conclusion domain.CIConclusion, msg string) {
248 now := rs.c.now()
249 cancelish := conclusion == domain.CICancelled || conclusion == domain.CITimedOut
250 for i := range j.Steps {
251 s := &j.Steps[i]
252 if s.State == domain.CIStepCompleted {
253 continue
254 }
255 s.Conclusion = domain.CICancelled
256 if s.State == domain.CIStepPending && !cancelish {
257 s.Conclusion = domain.CISkipped
258 }
259 s.Outcome = s.Conclusion
260 s.State, s.CompletedAt = domain.CIStepCompleted, now
261 }
262 j.Phase, j.Conclusion, j.CompletedAt = domain.CIPhaseEnded, conclusion, now
263 if msg != "" {
264 j.Error = trim(msg)
265 }
266 if j.HasToken() {
267 j.TokenValidUntil = now.Add(domain.CITokenGrace)
268 }
269 rs.revokeTokens(j)
270 rs.touch(j)
271 rs.c.mu.Lock()
272 delete(rs.c.heartbeats, jobRef(*j))
273 rs.c.mu.Unlock()
274
275 if j.FailFast && !j.ContinueOnError && conclusion.IsFailure() {
276 for i := range rs.jobs {
277 s := &rs.jobs[i]
278 if s.Key != j.Key || s.ID == j.ID || s.Ended() {
279 continue
280 }
281 if s.Active() {
282 rs.requestCancel(s, domain.CICancelFailFast, "Canceled because another job of the matrix failed (fail-fast)")
283 } else {
284 rs.end(s, domain.CICancelled, "")
285 }
286 }
287 }
288}
289
290// requestCancel asks the runner of a job to stop, once: the first reason wins.
291func (rs *runState) requestCancel(j *domain.CIJob, reason domain.CICancelReason, msg string) {
292 if j.CancelReason != domain.CICancelNone || j.Ended() {
293 return
294 }
295 j.CancelReason, j.CancelMessage, j.CancelRequestedAt = reason, msg, rs.c.now()
296 rs.touch(j)
297}
298
299// cancelAll cancels what is left of the run: jobs that no runner has are over
300// at once, the ones a runner has get a cancel request.
301func (rs *runState) cancelAll(msg string) {
302 for i := range rs.jobs {
303 j := &rs.jobs[i]
304 switch {
305 case j.Ended():
306 case j.Active():
307 rs.requestCancel(j, domain.CICancelUser, msg)
308 default:
309 rs.end(j, domain.CICancelled, "")
310 }
311 }
312}
313
314// advance plans the jobs that became ready and promotes queued matrix
315// instances, until nothing changes.
316func (rs *runState) advance(ctx context.Context) error {
317 if !rs.latest() {
318 return nil
319 }
320 if rs.run.CancelRequested {
321 rs.cancelAll("Canceled by the owner")
322 return nil
323 }
324 wf, err := rs.c.flows.get(rs.run.ID, rs.run.WorkflowPath, rs.src.WorkflowYAML)
325 if err != nil {
326 for i := range rs.jobs {
327 if !rs.jobs[i].Ended() && !rs.jobs[i].Active() {
328 rs.end(&rs.jobs[i], domain.CIFailure, err.Error())
329 }
330 }
331 return nil
332 }
333 if ok, err := rs.acquireRun(ctx, wf); err != nil || !ok {
334 return err // waits for its group, or was failed by it
335 }
336 for changed := true; changed; {
337 changed = false
338 for _, wj := range wf.Jobs {
339 insts := rs.instances(wj.ID)
340 if len(insts) != 1 {
341 continue
342 }
343 rec := rs.jobs[insts[0]]
344 if rec.Planned || rec.Ended() || rec.Phase != domain.CIPhaseWaiting {
345 continue
346 }
347 needs, st, ready := rs.needsOf(wf, wj)
348 if !ready {
349 continue
350 }
351 rs.planJob(ctx, wf, wj, insts[0], needs, st)
352 changed = true
353 }
354 }
355 return rs.promote(ctx)
356}
357
358// promote makes planned instances ready while max-parallel allows it and the
359// concurrency group of the job, if it has one of its own, gives it the turn.
360// An instance that waits for its group is pending.
361func (rs *runState) promote(ctx context.Context) error {
362 now := rs.c.now()
363 counts := map[string]int{}
364 for _, j := range rs.jobs {
365 if j.Phase == domain.CIPhaseReady || j.Active() {
366 counts[j.Key]++
367 }
368 }
369 for i := range rs.jobs {
370 j := &rs.jobs[i]
371 if !j.Planned || j.Phase != domain.CIPhaseWaiting {
372 continue
373 }
374 if j.MaxParallel > 0 && counts[j.Key] >= j.MaxParallel {
375 continue
376 }
377 if g := j.Concurrency; g.Group != "" && !g.FromRun {
378 st, err := rs.acquire(ctx, g.Group, domain.CIConcurrencyEntry{RunID: j.RunID, Attempt: j.RunAttempt, JobID: j.ID, Cancel: g.Cancel})
379 if err != nil {
380 return err
381 }
382 if st != g.State {
383 j.Concurrency.State = st
384 rs.touch(j)
385 }
386 if st != domain.CIConcurrencyHolding {
387 continue
388 }
389 }
390 j.Phase, j.ReadyAt = domain.CIPhaseReady, now
391 counts[j.Key]++
392 rs.touch(j)
393 }
394 return nil
395}
396
397// plan evaluates a job whose needs have ended: its condition, its strategy and
398// the parts of every instance the server decides (name, labels, timeout,
399// contexts). The record at idx becomes the first instance.
400func (rs *runState) planJob(ctx context.Context, wf *workflow.Workflow, wj *workflow.Job, idx int, needs *expr.Dictionary, st workflow.StatusFuncs) {
401 rec := rs.jobs[idx]
402 fail := func(msg string) { rs.end(&rs.jobs[idx], domain.CIFailure, msg) }
403
404 // github.job is empty while the job is evaluated (if, strategy, name,
405 // runs-on...): the runner sets it. Recorded: ctx-github-names, where
406 // "name: Display name of ${{ github.job }}" shows as "Display name of".
407 gh, err := rs.githubValue("", rs.run.Attempt)
408 if err != nil {
409 fail(err.Error())
410 return
411 }
412 vars, err := rs.c.varsContext(ctx, rs.run.Repo)
413 if err != nil {
414 fail(err.Error())
415 return
416 }
417 env := template.Env{Contexts: map[string]expr.Value{
418 "github": gh, "needs": needs, "inputs": expr.NewDictionary(), "vars": vars,
419 }}
420 run, err := wj.EvaluateIf(ctx, env, st)
421 switch {
422 case err != nil:
423 fail(err.Error())
424 return
425 case !run:
426 rs.jobs[idx].Name = rs.skippedName(ctx, wj, env)
427 rs.end(&rs.jobs[idx], domain.CISkipped, "")
428 return
429 }
430 if feature := unsupported(wj); feature != "" {
431 fail("not supported yet: " + feature)
432 return
433 }
434 plan, err := wj.EvaluateStrategy(ctx, env)
435 if err != nil {
436 fail(err.Error())
437 return
438 }
439
440 insts := make([]domain.CIJob, 0, len(plan.Combos))
441 for i := range plan.Combos {
442 menv, err := plan.MatrixEnv(env, i)
443 if err != nil {
444 fail(err.Error())
445 return
446 }
447 inst, err := rs.instance(ctx, wf, wj, plan, rec, i, menv, needs)
448 if err != nil {
449 fail(err.Error())
450 return
451 }
452 insts = append(insts, inst)
453 }
454 rs.jobs[idx] = insts[0]
455 rs.touch(&rs.jobs[idx])
456 for _, inst := range insts[1:] {
457 rs.jobs = append(rs.jobs, inst)
458 rs.dirty[inst.ID] = true
459 }
460 rs.sortJobs()
461}
462
463// skippedName is the best name for a job that did not run: its evaluated name
464// when that does not read the matrix, else its key.
465func (rs *runState) skippedName(ctx context.Context, wj *workflow.Job, env template.Env) string {
466 if wj.HasName {
467 if name, err := template.EvaluateString(ctx, wj.Name, env); err == nil && name != "" {
468 return name
469 }
470 }
471 return wj.ID
472}
473
474// instance builds the planned record of combination i of a job.
475func (rs *runState) instance(ctx context.Context, wf *workflow.Workflow, wj *workflow.Job, plan *workflow.Strategy, base domain.CIJob, i int, env template.Env, needs *expr.Dictionary) (domain.CIJob, error) {
476 j := base
477 j.Combo = i
478 j.ID = domain.NewJobID(rs.run.ID, base.Index, i)
479 if i > 0 {
480 j.Version = 0
481 }
482 name, err := wj.DisplayName(ctx, env, plan, i)
483 if err != nil {
484 return j, err
485 }
486 name = strings.TrimSpace(name) // GitHub shows "Display name of", not "Display name of "
487 labels, _, err := wj.EvaluateRunsOn(ctx, env)
488 if err != nil {
489 return j, err
490 }
491 if len(labels) == 0 {
492 return j, fmt.Errorf("job '%s' has no runs-on labels", wj.ID)
493 }
494 timeout := int32(domain.CIDefaultTimeoutMinutes)
495 if wj.TimeoutMinutes != nil {
496 n, err := template.EvaluateNumber(ctx, wj.TimeoutMinutes, env)
497 if err != nil {
498 return j, err
499 }
500 timeout = int32(min(max(math.Round(n), 1), 525600)) //nolint:gosec // clamped
501 }
502 gh, err := rs.githubValue(wj.ID, rs.run.Attempt)
503 if err != nil {
504 return j, err
505 }
506 matrix := plan.Combos[i].Values
507 var matrixCtx expr.Value = matrix
508 if !plan.HasMatrix {
509 matrixCtx = expr.Null{} // GitHub: "matrix": null for a job without a matrix
510 }
511 environment, err := environmentOf(ctx, wj, env)
512 if err != nil {
513 return j, err
514 }
515 // The vars of planning (see planJob), not read again for each leg.
516 vars, _ := env.Contexts["vars"].(*expr.Dictionary)
517 if vars == nil {
518 vars = expr.NewDictionary()
519 }
520 contexts, err := contextsJSON(gh, needs, plan.Context(i), matrixCtx, vars, rs.c.jobKeys(ctx, rs.run, base.Index, i))
521 if err != nil {
522 return j, err
523 }
524 j.Environment = environment
525 j.Permissions = domain.ResolvePermissions(permissionSpec(wf.Permissions), permissionSpec(wj.Permissions))
526 j.Name, j.Labels, j.TimeoutMinutes = name, labels, timeout
527 j.MatrixJSON = compactJSON(expr.ToJSON(matrix))
528 j.ContextsJSON, j.ComboTotal = contexts, len(plan.Combos)
529 j.FailFast, j.MaxParallel = plan.FailFast, plan.MaxParallel
530 j.ContinueOnError = false
531 if wj.ContinueOnError != nil {
532 // Evaluated with github, needs, strategy, matrix, vars and inputs.
533 coe, err := template.EvaluateBool(ctx, wj.ContinueOnError, env)
534 if err != nil {
535 return j, err
536 }
537 j.ContinueOnError = coe
538 }
539 if wj.Concurrency != nil {
540 // Evaluated with github, needs, strategy, matrix, inputs and vars;
541 // this leg's own group, which it takes before it may run.
542 conc, err := evalConcurrency(ctx, wj.Concurrency, env)
543 if err != nil {
544 return j, err
545 }
546 conc.Evaluated = true
547 j.Concurrency = conc
548 }
549 j.Planned, j.Phase = true, domain.CIPhaseWaiting
550 return j, nil
551}
552
553func compactJSON(s string) string {
554 var b bytes.Buffer
555 if err := json.Compact(&b, []byte(s)); err != nil {
556 return "{}"
557 }
558 return b.String()
559}
560
561// contextsJSON is the JSON object of the contexts the server computes for a
562// job (Job.contexts in runner.proto).
563//
564// "job" holds the keys of the job context the server knows (check_run_id,
565// workflow_ref, workflow_sha, workflow_repository, workflow_file_path); the
566// runner adds status. The runner copies the keys of this "job" into its job
567// context (executor.State.initRest), whatever they are.
568func contextsJSON(github, needs, strategy *expr.Dictionary, matrix expr.Value, vars *expr.Dictionary, job ordered) (string, error) {
569 out := map[string]json.RawMessage{}
570 for k, v := range map[string]expr.Value{
571 "github": github, "needs": needs, "strategy": strategy, "matrix": matrix,
572 "inputs": expr.NewDictionary(), "vars": vars,
573 } {
574 out[k] = json.RawMessage(compactJSON(expr.ToJSON(v)))
575 }
576 jb, err := json.Marshal(job)
577 if err != nil {
578 return "", err
579 }
580 out["job"] = jb
581 b, err := json.Marshal(out)
582 if err != nil {
583 return "", err
584 }
585 return string(b), nil
586}
587
588// permissionSpec reads the "permissions:" of a workflow or a job, nil for none.
589func permissionSpec(p *workflow.Permissions) *domain.PermissionSpec {
590 if p == nil {
591 return nil
592 }
593 spec := &domain.PermissionSpec{All: p.All, Levels: make(map[string]string, len(p.Scopes))}
594 for _, sc := range p.Scopes {
595 spec.Levels[sc.Scope] = sc.Level
596 }
597 return spec
598}