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