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