1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "io"
10 "path"
11 "runtime/debug"
12 "slices"
13 "strconv"
14 "strings"
15 "sync/atomic"
16 "time"
17
18 "github.com/bnema/zerowrap"
19
20 "bnema.dev/ermite-actions/expr"
21 "bnema.dev/ermite-actions/template"
22 "bnema.dev/ermite-actions/workflow"
23 "bnema.dev/ermite/internal/domain"
24 "bnema.dev/ermite/internal/ports"
25)
26
27const (
28 workflowDir = ".github/workflows"
29 // maxRunError bounds Run.error (8 KiB, run.proto).
30 maxRunError = 8 << 10
31 // maxPushCommits is the number of commits of the push event payload.
32 maxPushCommits = 20
33)
34
35// ciTrigger is a push waiting to be planned.
36type ciTrigger struct {
37 done func()
38 repo domain.RepoName
39 updates []domain.RefUpdate
40 actor string
41 pusher domain.CIPusher
42 // trusted: the principal that pushed or merged is the owner
43 // (domain.TrustedActor), so the run gets secrets.
44 trusted bool
45 at time.Time
46 log zerowrap.Logger
47}
48
49// OnPush queues the refs of a push or a merge for planning and returns; a
50// worker plans them. A full queue drops the trigger, logged at error: a push is
51// never slowed down or failed by CI. The push workflows are queued first; the
52// merge requests the push touches are then handed to their own worker, with a
53// short lock and no I/O, so nothing here waits for Git or the store.
54//
55// A push made with a job token starts nothing, as on GitHub (events created
56// with GITHUB_TOKEN start no run): that is what stops a workflow that pushes
57// from running itself again.
58func (c *CI) OnPush(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, pusher domain.Principal) {
59 done := c.pushLifecycle(ctx, repo, 2)
60 defer done()
61
62 // Whoever pushed, the workflow files of the default branch may have changed.
63 // (Who pushed was recorded before the refs moved: Pushes.Check.)
64 c.markSchedulesDirty(repo, updates)
65 // The merge requests follow the push whoever made it (their refs); only a
66 // push that is not a job's starts pull_request runs.
67 defer c.onBranchesMoved(ctx, repo, updates, domain.AuthorName(pusher), domain.PusherOf(pusher), domain.TrustedActor(pusher), !pusher.ViaJobToken())
68 if pusher.ViaJobToken() {
69 log := zerowrap.FromCtx(ctx)
70 log.Debug().Str("repo", repo.String()).Int("refs", len(updates)).Msg("push by a job token starts no run")
71 done()
72 return
73 }
74 c.enqueuePush(ctx, repo, updates, domain.AuthorName(pusher), domain.PusherOf(pusher), domain.TrustedActor(pusher), done)
75}
76
77// OnMerge queues the move of a target branch by a merge. The actor of the run is
78// the owner, as GitHub shows; the pusher recorded with it, whose rights the
79// actions of other repositories are resolved with, is the principal that merged:
80// a token limited to some repositories does not gain the owner's reach by
81// merging. For the same reason the run is trusted (gets secrets) only when the
82// merger is: the displayed actor is not the one that decides.
83func (c *CI) OnMerge(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, merger domain.Principal) {
84 done := c.pushLifecycle(ctx, repo, 2)
85 defer done()
86
87 c.markSchedulesDirty(repo, updates)
88 // The merge moved a target branch: the test merges of the other requests
89 // into it are computed again (a merge starts no synchronize run), after the
90 // push workflows are queued.
91 defer c.onBranchesMoved(ctx, repo, updates, domain.AuthorName(domain.Owner()), domain.PusherOf(merger), domain.TrustedActor(merger), false)
92 if merger.ViaJobToken() {
93 log := zerowrap.FromCtx(ctx)
94 log.Debug().Str("repo", repo.String()).Int("refs", len(updates)).Msg("merge by a job token starts no run")
95 done()
96 return
97 }
98 c.enqueuePush(ctx, repo, updates, domain.AuthorName(domain.Owner()), domain.PusherOf(merger), domain.TrustedActor(merger), done)
99}
100
101func (c *CI) enqueue(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, actor string, pusher domain.CIPusher, trusted bool) {
102 c.enqueuePush(ctx, repo, updates, actor, pusher, trusted, c.pushLifecycle(ctx, repo, 1))
103}
104
105// pushLifecycle holds one repository activity count until all parts finish.
106func (c *CI) pushLifecycle(ctx context.Context, repo domain.RepoName, parts int32) func() {
107 c.mu.Lock()
108 c.pushes[repo]++
109 c.mu.Unlock()
110 var remaining atomic.Int32
111 remaining.Store(parts)
112 return func() {
113 if remaining.Add(-1) == 0 {
114 c.finishPush(repo)
115 c.kickAutoMerge(ctx, repo)
116 }
117 }
118}
119
120func (c *CI) enqueuePush(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, actor string, pusher domain.CIPusher, trusted bool, done func()) {
121 var kept []domain.RefUpdate
122 for _, u := range updates {
123 if strings.HasPrefix(u.Ref, "refs/heads/") || strings.HasPrefix(u.Ref, "refs/tags/") {
124 kept = append(kept, u)
125 }
126 }
127 if len(kept) == 0 {
128 done()
129 return
130 }
131 t := ciTrigger{done: done, repo: repo, updates: kept, actor: actor, pusher: pusher, trusted: trusted, at: c.now(), log: zerowrap.FromCtx(ctx)}
132 c.mu.Lock()
133 if c.closed {
134 c.mu.Unlock()
135 done()
136 return
137 }
138 select {
139 case c.queue <- t:
140 c.mu.Unlock()
141 default:
142 c.mu.Unlock()
143 done()
144 t.log.Error().Str("repo", repo.String()).Int("refs", len(kept)).Msg("ci trigger queue is full: the workflows of this push are not run")
145 }
146
147}
148
149func (c *CI) triggerLoop() {
150 defer c.wg.Done()
151 for {
152 select {
153 case <-c.ctx.Done():
154 return
155 case t := <-c.queue:
156 c.planPush(t)
157 }
158 }
159}
160
161// planPush plans one push; a panic or an error is logged and never escapes.
162func (c *CI) planPush(t ciTrigger) {
163 ctx := zerowrap.WithCtx(c.ctx, t.log)
164 defer t.done()
165 defer func() {
166 if r := recover(); r != nil {
167 t.log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Str("repo", t.repo.String()).Msg("ci trigger panicked")
168 }
169 }()
170 if err := c.handlePush(ctx, t); err != nil && ctx.Err() == nil {
171 t.log.Error().Err(err).Str("repo", t.repo.String()).Msg("ci trigger failed")
172 }
173}
174
175// handlePush starts the workflows that match each ref of the push.
176func (c *CI) handlePush(ctx context.Context, t ciTrigger) error {
177 // CI is on once a runner is registered (plan, "Config"): without one a push
178 // makes no run and no check, so a forge that does not use CI shows nothing.
179 runners, err := c.cfg.Store.ListRunners(ctx)
180 if err != nil {
181 return fmt.Errorf("list the runners: %w", err)
182 }
183 if len(runners) == 0 {
184 return nil
185 }
186 var tags, created, deleted int
187 for _, u := range t.updates {
188 if strings.HasPrefix(u.Ref, "refs/tags/") {
189 switch {
190 case u.IsDelete():
191 deleted++
192 case u.IsCreate():
193 tags++
194 created++
195 default:
196 tags++
197 }
198 }
199 }
200 var errs []error
201 for _, u := range t.updates {
202 var err error
203 if u.IsDelete() {
204 err = c.handleDelete(ctx, t, u, deleted)
205 } else {
206 err = c.handleRef(ctx, t, u, tags, created)
207 }
208 if err != nil {
209 errs = append(errs, fmt.Errorf("%s: %w", u.Ref, err))
210 }
211 }
212 return errors.Join(errs...)
213}
214
215// pushFacts is what is known of one pushed ref.
216type pushFacts struct {
217 update domain.RefUpdate
218 commit domain.ObjectID // the commit the ref points to, tags followed
219 head domain.Commit
220 commits []domain.Commit // oldest first, at most maxPushCommits
221 forced bool
222 created bool
223 files []string
224 // filesTruncated: the list of changed files is not available.
225 filesTruncated bool
226 tagCount int
227 isTag bool
228 // defaultBranch is the default branch of the repository, "" when unknown.
229 defaultBranch string
230 // payload is the commit part of the push event, shared by the workflows
231 // of the ref; handleRef builds it once, before the first one that matches.
232 payload *pushPayload
233 // pull is set for a pull_request run: update.Ref is then
234 // refs/pull/<n>/merge (or /head), commit the test merge commit (or the
235 // head) and head the head commit. For a pull_request_target run
236 // (pull.target) update.Ref is the default branch and commit its tip.
237 pull *pullFacts
238 // custom is set for the events that are not a push or a merge request:
239 // workflow_dispatch, schedule, workflow_run, create and delete (ci_events.go).
240 custom *customEvent
241}
242
243// eventName is the event of the run, as the run records and github.event_name
244// show it.
245func (f pushFacts) eventName() string {
246 switch {
247 case f.pull != nil && f.pull.target:
248 return eventPullTarget
249 case f.pull != nil:
250 return "pull_request"
251 case f.custom != nil:
252 return f.custom.name
253 }
254 return "push"
255}
256
257// runTitle is the title of a run without run-name: the commit message of a
258// push, the title of the merge request for a pull_request.
259func (f pushFacts) runTitle() string {
260 switch {
261 case f.pull != nil:
262 return firstLine(f.pull.mr.Title.String())
263 case f.custom != nil && f.custom.title != "":
264 return firstLine(f.custom.title)
265 }
266 return firstLine(f.head.Message)
267}
268
269// pullNumber is CIRun.PullNumber: the merge request, 0 for a push.
270func (f pushFacts) pullNumber() int64 {
271 if f.pull != nil {
272 return f.pull.mr.Number
273 }
274 return 0
275}
276
277// headSha is CIRun.HeadSha: the head of the merge request (pull_request and
278// pull_request_target), "" for a push.
279func (f pushFacts) headSha() string {
280 if f.pull != nil {
281 return f.pull.head.String()
282 }
283 return ""
284}
285
286// handleRef starts the push workflows of a pushed ref and, for a created one,
287// its create workflows. tagCount is the number of tags the push sets,
288// tagsCreated the number it creates.
289func (c *CI) handleRef(ctx context.Context, t ciTrigger, u domain.RefUpdate, tagCount, tagsCreated int) error {
290 isTag := strings.HasPrefix(u.Ref, "refs/tags/")
291 // GitHub creates no push event for the tags of a push that sets more than
292 // three, and no create event when it creates more than three.
293 push := !isTag || tagCount <= workflow.MaxTagsPerPush
294 create := u.IsCreate() && (!isTag || tagsCreated <= workflow.MaxTagsPerPush)
295 if !push && !create {
296 return nil
297 }
298 repo, err := c.cfg.Meta.Get(ctx, t.repo)
299 if err != nil {
300 return nil // deleted meanwhile
301 }
302 facts, err := c.collect(ctx, t.repo, u, isTag, tagCount)
303 if err != nil {
304 return err
305 }
306 files, err := c.workflowFiles(ctx, t.repo, facts.commit)
307 if err != nil || len(files) == 0 {
308 return err
309 }
310 skip := workflow.SkipCI(facts.head.Message)
311 var errs []error
312 for _, f := range files {
313 wf, perrs := parseWorkflow(f.path, f.content)
314 var problems []string
315 for _, e := range perrs {
316 problems = append(problems, e.Error())
317 }
318 switch {
319 case !push:
320 case f.problem != "":
321 errs = append(errs, c.startInvalid(ctx, repo, t, facts, f.path, f.path, f.problem))
322 case len(problems) > 0:
323 errs = append(errs, c.startInvalid(ctx, repo, t, facts, f.path, f.path, strings.Join(problems, "\n")))
324 case skip || !wf.MatchPush(workflow.PushInfo{
325 Ref: u.Ref, Created: facts.created, ChangedFiles: facts.files, FilesTruncated: facts.filesTruncated, TagCount: tagCount,
326 }):
327 default:
328 if facts.payload == nil {
329 facts.payload = c.pushPayload(ctx, repo, facts)
330 }
331 if problem := c.matrixProblem(ctx, repo, t, facts, f.path, wf); problem.text != "" {
332 name := f.path
333 if problem.named {
334 name = workflowName(wf, f.path)
335 }
336 errs = append(errs, c.startInvalid(ctx, repo, t, facts, f.path, name, problem.text))
337 continue
338 }
339 errs = append(errs, c.startRun(ctx, repo, t, facts, f.path, string(f.content), wf))
340 }
341 }
342 if create {
343 errs = append(errs, c.startCreateRuns(ctx, repo, t, facts, files))
344 }
345 return errors.Join(errs...)
346}
347
348// collect reads what the payload and the filters need from Git.
349func (c *CI) collect(ctx context.Context, repo domain.RepoName, u domain.RefUpdate, isTag bool, tagCount int) (pushFacts, error) {
350 newID, err := domain.ParseObjectID(u.New)
351 if err != nil {
352 return pushFacts{}, err
353 }
354 f := pushFacts{update: u, created: u.IsCreate(), isTag: isTag, tagCount: tagCount}
355 if def, _, err := c.cfg.Repos.DefaultBranch(ctx, repo); err == nil {
356 f.defaultBranch = def
357 }
358 res, err := c.cfg.Reader.ResolveRevision(ctx, repo, domain.RevisionOf(newID))
359 if err != nil {
360 return f, fmt.Errorf("resolve %s: %w", u.New, err)
361 }
362 f.commit = res.CommitID
363 if isTag {
364 // Recorded (push-tag, skip-directive-tag, push-tags-limit): the event of
365 // a tag carries the commits that no branch has, newest as head_commit
366 // ("compare" is then that commit's page, or "<first>^...<head>" for
367 // several). Not the range from the previous push, and not the commits
368 // the default branch lacks: in push-tag the branch commit under the
369 // tagged one is not listed, in skip-directive-tag, which has no branch,
370 // both commits are. A tag on a commit that a branch has lists that
371 // commit alone (not recorded). No file list: paths filters are not
372 // evaluated for tags.
373 cs, _, err := c.cfg.Diff.CommitsOffBranches(ctx, repo, f.commit, ports.MaxCommitLimit)
374 if err != nil {
375 return f, fmt.Errorf("read the commits of the tag: %w", err)
376 }
377 if len(cs) == 0 {
378 cs, _, err = c.cfg.Git.ListRange(ctx, repo, domain.ObjectID{}, f.commit, 1)
379 if err != nil {
380 return f, fmt.Errorf("read the commit of the tag: %w", err)
381 }
382 if len(cs) == 0 {
383 return f, fmt.Errorf("read the commit of the tag: %w", domain.ErrNotFound)
384 }
385 }
386 f.head = cs[0]
387 f.commits = slices.Clone(cs[:min(len(cs), maxPushCommits)])
388 slices.Reverse(f.commits)
389 return f, nil
390 }
391
392 var exclude domain.ObjectID
393 if !f.created {
394 if old, err := domain.ParseObjectID(u.Old); err == nil {
395 exclude = old
396 if ok, err := c.cfg.Git.IsAncestor(ctx, repo, old, f.commit); err == nil && !ok {
397 f.forced = true
398 }
399 }
400 }
401 base := exclude
402 if f.created {
403 base = c.defaultBase(ctx, repo, strings.TrimPrefix(u.Ref, "refs/heads/"), f.commit)
404 exclude = base
405 }
406 cs, more, err := c.cfg.Git.ListRange(ctx, repo, exclude, f.commit, ports.MaxCommitLimit)
407 if err != nil {
408 return f, fmt.Errorf("read the commits of the push: %w", err)
409 }
410 if len(cs) == 0 {
411 // A new branch at a commit the default branch has, or a branch rewound
412 // to an ancestor: no new commit, no changed file. GitHub still runs
413 // the push workflows that have no paths filter, for the head commit.
414 head, _, err := c.cfg.Git.ListRange(ctx, repo, domain.ObjectID{}, f.commit, 1)
415 if err != nil {
416 return f, fmt.Errorf("read the head commit of the push: %w", err)
417 }
418 if len(head) == 0 {
419 return f, fmt.Errorf("read the head commit of the push: %w", domain.ErrNotFound)
420 }
421 f.head = head[0]
422 return f, nil
423 }
424 f.head = cs[0]
425 n := min(len(cs), maxPushCommits)
426 f.commits = slices.Clone(cs[:n])
427 slices.Reverse(f.commits)
428 if more || base == f.commit {
429 // More than 1,000 commits: the diff is not looked at (it runs).
430 f.filesTruncated = more
431 }
432 if !more && base != f.commit {
433 paths, _, err := c.cfg.Diff.ChangedPaths(ctx, repo, base, f.commit, workflow.MaxFilesEvaluated)
434 if err != nil {
435 f.filesTruncated = true // diff unavailable: runs
436 } else {
437 f.files = paths
438 }
439 }
440 return f, nil
441}
442
443// defaultBase is the merge base of a new branch with the default branch, the
444// zero id when there is none (every file of the branch counts).
445func (c *CI) defaultBase(ctx context.Context, repo domain.RepoName, branch string, head domain.ObjectID) domain.ObjectID {
446 def, exists, err := c.cfg.Repos.DefaultBranch(ctx, repo)
447 if err != nil || !exists || def == branch {
448 return domain.ObjectID{}
449 }
450 tip, err := c.cfg.Git.BranchHead(ctx, repo, def)
451 if err != nil {
452 return domain.ObjectID{}
453 }
454 base, err := c.cfg.Git.MergeBase(ctx, repo, tip, head)
455 if err != nil {
456 return domain.ObjectID{}
457 }
458 return base
459}
460
461type workflowFile struct {
462 path string
463 content []byte
464 // problem is set when the file cannot even be parsed (too large).
465 problem string
466}
467
468// workflowFiles reads .github/workflows/*.yml|*.yaml at the commit, without the
469// workflows the repository disabled: every event that starts a run reads its
470// workflows here, so a disabled one starts none (ci_workflows.go).
471func (c *CI) workflowFiles(ctx context.Context, repo domain.RepoName, commit domain.ObjectID) ([]workflowFile, error) {
472 files, err := c.listWorkflowFiles(ctx, repo, commit)
473 if err != nil || len(files) == 0 {
474 return files, err
475 }
476 off, err := c.disabledPaths(ctx, repo)
477 if err != nil || len(off) == 0 {
478 return files, err
479 }
480 return slices.DeleteFunc(files, func(f workflowFile) bool { return slices.Contains(off, f.path) }), nil
481}
482
483// listWorkflowFiles is workflowFiles with the disabled ones. The disabled state
484// is also enforced where a run is stored (CIStore.CreateRun), which is what
485// holds against a change made while a push is being planned.
486func (c *CI) listWorkflowFiles(ctx context.Context, repo domain.RepoName, commit domain.ObjectID) ([]workflowFile, error) {
487 dir, err := domain.ParseTreePath(workflowDir)
488 if err != nil {
489 return nil, err
490 }
491 tree, err := c.cfg.Reader.GetTree(ctx, repo, domain.RevisionOf(commit), dir, false)
492 if errors.Is(err, domain.ErrNotFound) || errors.Is(err, domain.ErrNotDirectory) {
493 return nil, nil
494 }
495 if err != nil {
496 return nil, fmt.Errorf("list %s: %w", workflowDir, err)
497 }
498 var out []workflowFile
499 for _, e := range tree.Entries {
500 ext := strings.ToLower(path.Ext(e.Name))
501 if (e.Kind != domain.EntryFile && e.Kind != domain.EntryExecutable) || (ext != ".yml" && ext != ".yaml") {
502 continue
503 }
504 if len(out) == maxCIWorkflowFiles {
505 break
506 }
507 wf := workflowFile{path: e.Path}
508 if wf.path == "" {
509 wf.path = workflowDir + "/" + e.Name
510 }
511 if e.Size > maxCIWorkflowBytes {
512 wf.problem = fmt.Sprintf("%s: the workflow file is larger than 1 MiB", wf.path)
513 out = append(out, wf)
514 continue
515 }
516 tp, err := domain.ParseTreePath(wf.path)
517 if err != nil {
518 continue
519 }
520 _, rc, err := c.cfg.Reader.OpenBlob(ctx, repo, domain.RevisionOf(commit), tp)
521 if err != nil {
522 return nil, fmt.Errorf("read %s: %w", wf.path, err)
523 }
524 wf.content, err = io.ReadAll(io.LimitReader(rc, maxCIWorkflowBytes+1))
525 _ = rc.Close()
526 if err != nil {
527 return nil, fmt.Errorf("read %s: %w", wf.path, err)
528 }
529 if len(wf.content) > maxCIWorkflowBytes {
530 wf.content, wf.problem = nil, fmt.Sprintf("%s: the workflow file is larger than 1 MiB", wf.path)
531 }
532 out = append(out, wf)
533 }
534 return out, nil
535}
536
537// --- the run ---
538
539func firstLine(s string) string {
540 line, _, _ := strings.Cut(s, "\n")
541 return truncateRunes(strings.TrimSpace(line), 256)
542}
543
544// startInvalid records the failed run of an invalid workflow file: no job, the
545// errors as Run.error, like GitHub's "Invalid workflow file".
546//
547// name is what the run is called: the path of the file when the file is not a
548// workflow GitHub accepts, or the workflow's own name for a workflow that
549// parses but whose strategy GitHub refuses (recorded: matrix-over-256).
550func (c *CI) startInvalid(ctx context.Context, repo domain.Repo, t ciTrigger, f pushFacts, wfPath, name, problem string) error {
551 now := c.now()
552 draft, err := c.createRun(ctx, f, repo.Name, wfPath, func(id, number int64) ports.CIRunDraft {
553 run := domain.CIRun{
554 Repo: repo.Name, ID: id, Number: number, Attempt: 1, WorkflowPath: wfPath, WorkflowName: name, Name: name,
555 Title: f.runTitle(), Event: f.eventName(), Ref: f.update.Ref, Sha: f.commit.String(), HeadSha: f.headSha(), PullNumber: f.pullNumber(), Actor: t.actor,
556 Status: domain.CICompleted, Conclusion: domain.CIFailure, Error: cutText(problem, maxRunError), Trusted: t.trusted,
557 CreatedAt: now, StartedAt: now, CompletedAt: now,
558 }
559 return ports.CIRunDraft{
560 Run: run, Source: domain.CIRunSource{Repo: repo.Name, WorkflowYAML: "", GithubJSON: "{}", Pusher: t.pusher},
561 Checks: []domain.CICheck{{
562 ID: "run-" + strconv.FormatInt(id, 10), Repo: repo.Name, Sha: run.CheckSha(), Name: wfPath, Status: domain.CICompleted,
563 Conclusion: domain.CIFailure, RunID: id, Attempt: 1, WorkflowName: name, CreatedAt: now, StartedAt: now, CompletedAt: now,
564 }},
565 }
566 })
567 if errors.Is(err, domain.ErrNotFound) || errors.Is(err, errPullClosed) || errors.Is(err, domain.ErrWorkflowDisabled) {
568 return nil
569 }
570 if err != nil {
571 return err
572 }
573 c.publishRun(ctx, repo, draft.Run)
574 c.onRunEvent(ctx, draft.Run, runRequested)
575 c.onRunEvent(ctx, draft.Run, runCompleted)
576 log := logRun(ctx, draft.Run)
577 log.Info().Str("workflow", wfPath).Msg("ci run failed: invalid workflow file")
578 return nil
579}
580
581func (c *CI) publishRun(ctx context.Context, repo domain.Repo, run domain.CIRun) {
582 c.cfg.Bus.Publish(ctx, domain.RunUpdated{Run: run, At: c.now(), Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
583}
584
585// startRun creates the run of a matching workflow with one waiting record per
586// job of the file, then plans the jobs that can start.
587func (c *CI) startRun(ctx context.Context, repo domain.Repo, t ciTrigger, f pushFacts, wfPath, yaml string, wf *workflow.Workflow) error {
588 _, err := c.launchRun(ctx, repo, t, f, wfPath, yaml, wf)
589 if errors.Is(err, domain.ErrWorkflowDisabled) {
590 return nil // disabled since it was read: no run (the store decides, atomically)
591 }
592 return err
593}
594
595// launchRun is startRun that also returns the run it created, as it was when
596// created (the zero run when the repository or the merge request is gone). A
597// workflow that is disabled when the run is stored gets none, whatever was
598// checked before: the error is domain.ErrWorkflowDisabled.
599func (c *CI) launchRun(ctx context.Context, repo domain.Repo, t ciTrigger, f pushFacts, wfPath, yaml string, wf *workflow.Workflow) (domain.CIRun, error) {
600 now := c.now()
601 name := workflowName(wf, wfPath)
602 // The title of a run without run-name is the commit message for a branch.
603 // For a tag GitHub shows the name of the workflow registered for the path,
604 // not the commit message (recorded: filter-tags, filter-paths-tag,
605 // filter-branches-ignore, and push-tag, push-tags-limit, push-delete-tag,
606 // skip-directive-tag, whose ci.yml is called "push-branch-create" on the
607 // default branch of the recording repository although each run's own
608 // workflow has another name: 14 of 14 tag runs). Registered is read as the
609 // "name:" of the file on the default branch; the first version GitHub saw
610 // is the same file in the recording and cannot be told apart.
611 title := f.runTitle()
612 if f.isTag {
613 title = c.registeredName(ctx, repo.Name, wfPath, name)
614 }
615 vars, err := c.varsContext(ctx, repo.Name)
616 if err != nil {
617 return domain.CIRun{}, err
618 }
619 draft, err := c.createRun(ctx, f, repo.Name, wfPath, func(id, number int64) ports.CIRunDraft {
620 run := domain.CIRun{
621 Repo: repo.Name, ID: id, Number: number, Attempt: 1, WorkflowPath: wfPath, WorkflowName: name, Name: name,
622 Title: title, Event: f.eventName(), Ref: f.update.Ref, Sha: f.commit.String(), HeadSha: f.headSha(), PullNumber: f.pullNumber(), Actor: t.actor,
623 Status: domain.CIQueued, Trusted: t.trusted, CreatedAt: now,
624 }
625 githubJSON := c.githubJSON(ctx, repo, run, f)
626 if runName, ok := c.evalRunName(ctx, wf, githubJSON, vars); ok {
627 run.Name, run.Title = runName, runName
628 }
629 d := ports.CIRunDraft{Run: run, Source: domain.CIRunSource{
630 Repo: repo.Name, WorkflowYAML: yaml, GithubJSON: githubJSON, Pusher: t.pusher,
631 }}
632 for i, j := range wf.Jobs {
633 job := domain.CIJob{
634 ID: domain.NewJobID(id, i, 0), RunID: id, Attempt: 1, RunAttempt: 1, Repo: repo.Name, Key: j.ID, Name: j.ID,
635 Phase: domain.CIPhaseWaiting, Needs: slices.Clone(j.Needs), Index: i, CreatedAt: now,
636 }
637 d.Jobs = append(d.Jobs, job)
638 d.Checks = append(d.Checks, domain.CICheck{
639 ID: job.ID, Repo: repo.Name, Sha: run.CheckSha(), Name: job.Name, Status: domain.CIQueued, RunID: id, JobID: job.ID,
640 Attempt: 1, WorkflowName: name, CreatedAt: now,
641 })
642 }
643 return d
644 })
645 if errors.Is(err, domain.ErrNotFound) || errors.Is(err, errPullClosed) {
646 return domain.CIRun{}, nil
647 }
648 if err != nil {
649 return domain.CIRun{}, err
650 }
651 c.publishRun(ctx, repo, draft.Run)
652 for _, j := range draft.Jobs {
653 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
654 }
655 c.onRunEvent(ctx, draft.Run, runRequested)
656 log := logRun(ctx, draft.Run)
657 log.Info().Str("workflow", wfPath).Str("ref", f.update.Ref).Str("event", draft.Run.Event).Int("jobs", len(draft.Jobs)).Msg("ci run created")
658 return draft.Run, c.advanceRun(ctx, draft.Run.ID)
659}
660
661// workflowName is the name of a workflow, the path of its file when it has no
662// "name:".
663func workflowName(wf *workflow.Workflow, path string) string {
664 if wf.Name == "" {
665 return path
666 }
667 return wf.Name
668}
669
670// registeredName is the name GitHub shows for a tag run without run-name: the
671// name of the workflow registered for the path, here the "name:" of the file on
672// the default branch (the path when that file has none). It falls back to own,
673// the name of the workflow of the run, when the default branch has no such
674// file, the file does not parse, or anything cannot be read; a failure other
675// than a missing file is logged at debug.
676func (c *CI) registeredName(ctx context.Context, repo domain.RepoName, wfPath, own string) string {
677 name, err := c.registeredNameOf(ctx, repo, wfPath)
678 if err != nil {
679 if !errors.Is(err, domain.ErrNotFound) {
680 log := zerowrap.FromCtx(ctx)
681 log.Debug().Err(err).Str("workflow", wfPath).Msg("ci tag run title: the registered workflow name is not readable, using the run's own")
682 }
683 return own
684 }
685 return name
686}
687
688// registeredNameOf is registeredName without the fallback; a missing default
689// branch or file is domain.ErrNotFound.
690func (c *CI) registeredNameOf(ctx context.Context, repo domain.RepoName, wfPath string) (string, error) {
691 def, exists, err := c.cfg.Repos.DefaultBranch(ctx, repo)
692 if err != nil {
693 return "", err
694 }
695 if !exists {
696 return "", domain.ErrNotFound
697 }
698 tip, err := c.cfg.Git.BranchHead(ctx, repo, def)
699 if err != nil {
700 return "", err
701 }
702 tp, err := domain.ParseTreePath(wfPath)
703 if err != nil {
704 return "", err
705 }
706 _, rc, err := c.cfg.Reader.OpenBlob(ctx, repo, domain.RevisionOf(tip), tp)
707 if err != nil {
708 return "", err
709 }
710 content, err := io.ReadAll(io.LimitReader(rc, maxCIWorkflowBytes+1))
711 _ = rc.Close()
712 if err != nil {
713 return "", err
714 }
715 if len(content) > maxCIWorkflowBytes {
716 return "", fmt.Errorf("workflow file over %d bytes", maxCIWorkflowBytes)
717 }
718 wf, errs := workflow.Parse(wfPath, content)
719 if len(errs) > 0 {
720 return "", fmt.Errorf("workflow does not parse: %v", errs[0])
721 }
722 return workflowName(wf, wfPath), nil
723}
724
725// evalRunName evaluates the run-name of a workflow with the github, inputs and
726// vars contexts (the contexts the schema allows there). ok is false when the
727// workflow has none, when it does not evaluate (logged), or when it evaluates
728// to nothing: the run then has no name of its own, GitHub's display title falls
729// back to the commit message (a branch) or the registered name (a tag), and
730// the name of the run to the name of the workflow.
731func (c *CI) evalRunName(ctx context.Context, wf *workflow.Workflow, githubJSON string, vars *expr.Dictionary) (string, bool) {
732 if wf.RunName == nil {
733 return "", false
734 }
735 gh, err := expr.FromJSON([]byte(githubJSON))
736 if err != nil {
737 return "", false
738 }
739 env := template.Env{Contexts: map[string]expr.Value{
740 "github": gh, "inputs": eventInputs(wf, githubJSON), "vars": vars,
741 }}
742 name, err := template.EvaluateString(ctx, wf.RunName, env)
743 if err != nil {
744 log := zerowrap.FromCtx(ctx)
745 log.Warn().Err(err).Msg("ci run-name does not evaluate: the run keeps the workflow name and the default title")
746 return "", false
747 }
748 if name = firstLine(name); name == "" {
749 return "", false
750 }
751 return name, true
752}
753
754// matrixProblem expands the strategy of every job of the file, whatever its
755// "if:" says, with an empty needs context (a matrix built from a needs output does not
756// evaluate here, and that error is ignored),
757// and returns the error text of the first strategy that produces more
758// configurations than GitHub allows: GitHub then fails the run with no job,
759// like an invalid workflow file (recorded: matrix-over-256; the run keeps the
760// name of the workflow). An "exclude" entry with a key that is no vector of the
761// matrix is refused too, and there GitHub names the run after the file path
762// (recorded: matrix-edge-values, a-exclude-unknown-key.yml). Any other error
763// of a strategy is ignored here and stays a job failure, found when the job is
764// planned. Whether GitHub counts the matrix of a job that "if:" skips is not
765// recorded yet (needs an "if: false" scenario over 256). The github context
766// is that of a run with no id yet, on the commits shared by the ref.
767func (c *CI) matrixProblem(ctx context.Context, repo domain.Repo, t ciTrigger, f pushFacts, wfPath string, wf *workflow.Workflow) matrixIssue {
768 if !slices.ContainsFunc(wf.Jobs, func(j *workflow.Job) bool { return j.Strategy != nil }) {
769 return matrixIssue{} // nothing to expand: no context to build
770 }
771 probe := domain.CIRun{Repo: repo.Name, Ref: f.update.Ref, Sha: f.commit.String(), Actor: t.actor, WorkflowPath: wfPath, WorkflowName: wf.Name, CreatedAt: c.now()}
772 ghJSON := c.githubJSON(ctx, repo, probe, f)
773 gh, err := expr.FromJSON([]byte(ghJSON))
774 if err != nil {
775 return matrixIssue{}
776 }
777 vars, err := c.varsContext(ctx, repo.Name)
778 if err != nil {
779 // Only the sizes of the matrices are looked at here: the job fails on
780 // the same error when it is planned.
781 vars = expr.NewDictionary()
782 }
783 env := template.Env{Contexts: map[string]expr.Value{
784 "github": gh, "needs": expr.NewDictionary(), "inputs": eventInputs(wf, ghJSON), "vars": vars,
785 }}
786 for _, j := range wf.Jobs {
787 _, err := j.EvaluateStrategy(ctx, env)
788 switch {
789 case errors.Is(err, workflow.ErrTooManyConfigurations):
790 return matrixIssue{text: wfPath + ": " + err.Error(), named: true}
791 case errors.Is(err, workflow.ErrExcludeUnknownKey):
792 // One line per offending entry, each with the file.
793 return matrixIssue{text: wfPath + ": " + strings.ReplaceAll(err.Error(), "\n", "\n"+wfPath+": ")}
794 }
795 }
796 return matrixIssue{}
797}
798
799// matrixIssue is a strategy GitHub refuses: the error text, and whether the run
800// keeps the name of the workflow (else it is named after the file).
801type matrixIssue struct {
802 text string
803 named bool
804}
805
806// --- the github context (see ci_github.go) ---
807
808// shortID is the 12-character form of an object id that GitHub puts in the
809// compare URL of a push.
810func shortID(id string) string {
811 if len(id) > 12 {
812 return id[:12]
813 }
814 return id
815}
816
817// retentionDays is the retention in days, as github.retention_days shows it
818// (at least 1).
819func (c *CI) retentionDays() int {
820 return max(int(c.cfg.Retention/(24*time.Hour)), 1)
821}