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