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