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