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