1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "runtime/debug"
10 "slices"
11 "strconv"
12 "strings"
13 "sync"
14 "time"
15
16 "github.com/bnema/zerowrap"
17
18 "bnema.dev/ermite-actions/workflow"
19 "bnema.dev/ermite/internal/domain"
20 "bnema.dev/ermite/internal/ports"
21)
22
23// The pull_request event (CI-3, D5). A merge request is what GitHub calls a pull
24// request: ermite keeps refs/pull/<n>/head at the head of its source branch and,
25// while the head merges cleanly into the target, refs/pull/<n>/merge at a test
26// merge commit (parents: target tip, then head; message "Merge <head> into
27// <base>"). A pull_request run is for that merge commit: github.ref is the merge
28// ref, github.sha the merge commit, the workflow is read from it, and its checks
29// attach to the head (domain.CIRun.CheckSha).
30//
31// Two workers, apart from the one that plans pushes (so a push never waits for
32// them):
33//
34// - the event worker handles the activities one at a time, in order: a merge
35// request opened (OnPullRequest), its source branch pushed (a synchronize),
36// closed or merged. An event carries the commits it is for, fixed when it
37// happened, and the actor and trust of whoever caused it: it runs for exactly
38// those commits, with that trust, even when the branches have moved on by the
39// time it is handled (GitHub runs each synchronize). A run is trusted only
40// if the head and the base of its event were pushed by trusted principals
41// too (pullTipsTrusted, pullBaseTrusted). The merge commit of a run is computed from the head and
42// base of its event, never from the branches as they are now.
43// - the refresh worker keeps refs/pull/<n>/{head,merge} at the latest branches
44// when a source or a target branch moves; the refreshes of one merge request
45// coalesce, and the work is bounded.
46//
47// Whatever publishes a ref or creates a run does it through the PullGate, under
48// the merge lock of the repository and only while the request is open, so a close
49// or a merge that completes meanwhile discards it: a closed request gets no
50// refs and no run but its "closed" one, which uses the last merge commit that was
51// published, or the head, and is never computed again.
52//
53// A head that does not merge cleanly makes no run for that update, as on GitHub;
54// the merge ref then keeps the last commit that merged, as GitHub's does.
55//
56// The pull_request_target event (CI-4, D4) has the same activities, payload
57// and filters, but runs the workflows of the default branch's tip, as taken
58// when the activity happened, conflict or not: github.ref is the default
59// branch, github.sha that tip. Its checks attach to the head too. It is
60// trusted when the cause of the activity is trusted and the default tip was
61// pushed by a trusted principal: the head and the base do not matter, as none
62// of their files run unless a workflow checks them out (GitHub's "pwn
63// request"). It is stricter than GitHub, which gives the secrets to any pull
64// request of the repository: an agent's activity runs it without secrets.
65
66const (
67 // DefaultCIPullRefreshes bounds the merge requests (and the pushes) waiting
68 // for the refresh worker. Past it the repository is marked dirty instead and
69 // the worker rescans its open merge requests when it has capacity, so that
70 // nothing stays stale for good.
71 DefaultCIPullRefreshes = 4096
72 // DefaultCIPullSnapshot bounds the merge requests one push takes the snapshot
73 // of (the open requests whose source branch it pushed). Past it the rest is
74 // skipped, an error is logged, and the repository is rescanned for the refs.
75 DefaultCIPullSnapshot = 256
76 // Stages of the work of the merge requests, for the test hook
77 // CIConfig.PullStage.
78 pullStageDiscover = "discover"
79 pullStageRescan = "rescan"
80 pullStageEvent = "event"
81 pullStageComputed = "computed"
82 pullStageRefresh = "refresh"
83)
84
85// errPullClosed means a merge request was closed or merged before the work for
86// it could be published.
87var errPullClosed = errors.New("merge request is not open")
88
89// pullEvent is one activity of a merge request waiting to be handled.
90type pullEvent struct {
91 repo domain.RepoName
92 number int64
93 action string
94 // head and base are the commits the activity is for; before is the head
95 // before a synchronize, zero otherwise.
96 head, base, before domain.ObjectID
97 actor string
98 pusher domain.CIPusher
99 trusted bool
100 // actorTrusted is the trust of the cause alone (the pusher of a
101 // synchronize, whoever opened or closed): trusted also requires the head
102 // and the base to be, pull_request_target the default tip.
103 actorTrusted bool
104 // target is the default branch's tip when the activity happened: what its
105 // pull_request_target runs execute.
106 target defaultTip
107 // merge is the last merge commit the trust of a closing was checked
108 // against, zero when there was none.
109 merge domain.ObjectID
110 // run is false when the cause is a job token: no run starts.
111 run bool
112 log zerowrap.Logger
113}
114
115func (e pullEvent) trigger(at time.Time) ciTrigger {
116 return ciTrigger{repo: e.repo, actor: e.actor, pusher: e.pusher, trusted: e.trusted, at: at, log: e.log}
117}
118
119// defaultTip is the tip of the default branch at one moment (CI-4, D4).
120type defaultTip struct {
121 branch string
122 commit domain.ObjectID
123 // trusted: a trusted principal set the branch to exactly that commit.
124 trusted bool
125}
126
127// defaultTipOf takes the tip of the default branch of a repository now, and
128// whether a trusted principal pushed it. The zero value when the repository
129// has no default branch (no pull_request_target run), and not trusted on any
130// error or without a record (refTrusted).
131func (c *CI) defaultTipOf(ctx context.Context, repo domain.RepoName) defaultTip {
132 def, exists, err := c.cfg.Repos.DefaultBranch(ctx, repo)
133 if err != nil || !exists {
134 return defaultTip{}
135 }
136 tip, err := c.cfg.Git.BranchHead(ctx, repo, def)
137 if err != nil {
138 return defaultTip{}
139 }
140 return defaultTip{branch: def, commit: tip, trusted: c.refTrusted(ctx, repo, pinnedRef{ref: "refs/heads/" + def, commit: tip, object: tip})}
141}
142
143type pullKey struct {
144 repo domain.RepoName
145 number int64
146}
147
148// pullQueue is what the two workers read.
149//
150// Overload policy. The activities (events) are never coalesced and never
151// dropped by the cap of the refreshes: they go on their own queue, which behaves
152// as the queue of the push triggers (CI.queue) does when it is full: the item is
153// dropped and an error is logged, so a push is never slowed or failed by CI. Only
154// the refresh of refs/pull/<n>/{head,merge}, which is idempotent and recomputed
155// from the branches as they are, is coalesced (one waiting refresh per merge
156// request) and bounded (max); past the bound the repository is marked dirty and
157// its open merge requests are rescanned when the worker has capacity.
158type pullQueue struct {
159 events chan pullEvent
160
161 mu sync.Mutex
162 max int // bound of the waiting refreshes
163 pending map[pullKey]bool
164 order []pullKey
165 // dirty are the repositories whose merge requests may have stale refs: the
166 // bound is the number of repositories.
167 dirty map[domain.RepoName]bool
168 dirtyOrder []domain.RepoName
169 // attempts remember the last state tried for each merge request, so that a
170 // rescan does not try again what it already tried (a conflict).
171 attempts *pullAttempts
172 wake chan struct{}
173 gate ports.PullGate // set by SetPullGate
174}
175
176func newPullQueue(size, max, attempts int) *pullQueue {
177 return &pullQueue{
178 events: make(chan pullEvent, size), max: max, pending: map[pullKey]bool{},
179 dirty: map[domain.RepoName]bool{}, attempts: newPullAttempts(attempts), wake: make(chan struct{}, 1),
180 }
181}
182
183// SetPullGate gives the CI the gate of the merge requests (see PullGate). Call
184// it before merge requests can be opened.
185func (c *CI) SetPullGate(g ports.PullGate) {
186 c.pull.mu.Lock()
187 c.pull.gate = g
188 c.pull.mu.Unlock()
189}
190
191func (c *CI) pullEnabled() bool { return c.cfg.MRs != nil && c.cfg.Pull != nil }
192
193func (c *CI) stage(stage string) {
194 if c.cfg.PullStage != nil {
195 c.cfg.PullStage(stage)
196 }
197}
198
199// OnPullRequest queues an activity of a merge request.
200func (c *CI) OnPullRequest(ctx context.Context, repo domain.RepoName, ev ports.PullRequestEvent, p domain.Principal) {
201 if !c.pullEnabled() {
202 return
203 }
204 e := pullEvent{
205 repo: repo, number: ev.Number, action: ev.Action, head: ev.Head, base: ev.Base,
206 actor: domain.AuthorName(p), pusher: domain.PusherOf(p), run: !p.ViaJobToken(), log: zerowrap.FromCtx(ctx),
207 actorTrusted: domain.TrustedActor(p), target: c.defaultTipOf(ctx, repo),
208 }
209 if e.actorTrusted {
210 e.trusted, e.merge = c.pullTipsTrusted(ctx, repo, ev)
211 }
212 c.queuePullEvent(e)
213}
214
215// pullTipsTrusted reports whether the code the runs of an activity that is not
216// a push execute was pushed by trusted principals (CI-4, D1), and, for a
217// closing, the last merge commit it checked (zero when there is none). The
218// workflow of a pull_request run comes from its merge commit, so from the head
219// and the base: whoever opens or closes the request did not necessarily write
220// either, and without this the owner opening the request of an agent's branch
221// (or into an agent's branch) would run the agent's workflow with the owner's
222// secrets. Every check is positive: an error, an unknown commit or a missing
223// record is not trusted.
224//
225// The records of the source and target branches (domain.CIRefTip) must name
226// exactly the head and the base of the event; they are read here, when the
227// activity happens, as the commits of the event are fixed: a push of either
228// branch before the event is handled changes nothing (a push of the source has
229// its own event and trust). A tip with no record (pushed before the records
230// existed: the next trusted push restores it), or a record for another commit
231// (the branch was force-pushed, moved or deleted) is not trusted. The base of
232// an opening is ev.Base, the target tip of that moment. A closing runs the last
233// merge commit published (which no longer changes once the request is closed:
234// the event worker checks it is still the one checked here), whose parents are
235// its head and base; without one it runs its head alone.
236//
237// A synchronize carries the trust of the push that set the source to its head
238// (onBranchesMoved), and its base is checked there. The closing a merge makes
239// is checked by mergedTrusted.
240func (c *CI) pullTipsTrusted(ctx context.Context, repo domain.RepoName, ev ports.PullRequestEvent) (bool, domain.ObjectID) {
241 mr, err := c.cfg.MRs.Get(ctx, repo, ev.Number)
242 if err != nil {
243 pullUntrusted(ctx, repo, ev, err, "the merge request cannot be read")
244 return false, domain.ObjectID{}
245 }
246 base := ev.Base
247 var merge domain.ObjectID
248 if ev.Action == domain.PullClosed {
249 var parents []domain.ObjectID
250 if merge, parents, err = c.lastMerge(ctx, repo, mr.Number); err != nil {
251 pullUntrusted(ctx, repo, ev, err, "the last merge commit cannot be read")
252 return false, merge
253 }
254 base = domain.ObjectID{} // no merge commit: the head runs alone
255 if !merge.IsZero() {
256 if parents[1] != ev.Head {
257 pullUntrusted(ctx, repo, ev, nil, "the last merge commit is of another head")
258 return false, merge
259 }
260 base = parents[0]
261 }
262 }
263 if ev.Merged {
264 return c.mergedTrusted(ctx, repo, ev, mr, base), merge
265 }
266 if !c.refTrusted(ctx, repo, pinnedRef{ref: "refs/heads/" + mr.Source, commit: ev.Head, object: ev.Head}) {
267 pullUntrusted(ctx, repo, ev, nil, "the head was not pushed by a trusted principal")
268 return false, merge
269 }
270 if ev.Action == domain.PullClosed && merge.IsZero() {
271 return true, merge
272 }
273 return c.pullBaseTrusted(ctx, repo, ev.Number, ev.Action, mr.Target, base), merge
274}
275
276// pullUntrusted logs why the runs of an activity are not trusted.
277func pullUntrusted(ctx context.Context, repo domain.RepoName, ev ports.PullRequestEvent, err error, why string) {
278 ctxLog(ctx).Info().Err(err).Str("repo", repo.String()).Int64("number", ev.Number).Str("action", ev.Action).
279 Str("head", ev.Head.String()).Msg("ci pull request runs are not trusted: " + why)
280}
281
282// lastMerge returns the last merge commit published for a request and its two
283// parents (base, head); zero and no error when there is none.
284func (c *CI) lastMerge(ctx context.Context, repo domain.RepoName, number int64) (domain.ObjectID, []domain.ObjectID, error) {
285 merge, err := c.cfg.Pull.PullRef(ctx, repo, domain.PullMergeRef(number))
286 if errors.Is(err, domain.ErrNotFound) {
287 return domain.ObjectID{}, nil, nil
288 }
289 if err != nil {
290 return merge, nil, err
291 }
292 cs, _, err := c.cfg.Git.ListRange(ctx, repo, domain.ObjectID{}, merge, 1)
293 if err != nil {
294 return merge, nil, err
295 }
296 if len(cs) != 1 || len(cs[0].Parents) != 2 {
297 return merge, nil, fmt.Errorf("%s is not a merge commit", merge)
298 }
299 return merge, cs[0].Parents, nil
300}
301
302// mergedTrusted reports whether the closing a merge made runs code the merger
303// took into the target: the request is merged, the target was set to its merge
304// result by a trusted principal (the merge records the merger), the head the
305// run executes is the one merged (the parents of the last merge commit were
306// checked against it), and the base it executes, if any, is in the merge
307// result's history. The records of the source branch are gone by then.
308func (c *CI) mergedTrusted(ctx context.Context, repo domain.RepoName, ev ports.PullRequestEvent, mr domain.MergeRequest, base domain.ObjectID) bool {
309 switch {
310 case ev.Action != domain.PullClosed || mr.State != domain.MRStateMerged || mr.MergedCommitID.IsZero():
311 pullUntrusted(ctx, repo, ev, nil, "the request is not merged")
312 return false
313 case ev.Head.IsZero() || ev.Head != mr.HeadID:
314 pullUntrusted(ctx, repo, ev, nil, "the head is not the merged one")
315 return false
316 case !c.refTrusted(ctx, repo, pinnedRef{ref: "refs/heads/" + mr.Target, commit: mr.MergedCommitID, object: mr.MergedCommitID}):
317 pullUntrusted(ctx, repo, ev, nil, "the merge result was not set by a trusted principal")
318 return false
319 }
320 if base.IsZero() {
321 return true
322 }
323 if mb, err := c.cfg.Git.MergeBase(ctx, repo, base, mr.MergedCommitID); err != nil || mb != base {
324 pullUntrusted(ctx, repo, ev, err, "the base is not in the merge result")
325 return false
326 }
327 return true
328}
329
330// pullBaseTrusted reports whether the target branch was set to exactly base by
331// a trusted principal: the merge commit of a run carries the target's files
332// too, its workflow included.
333func (c *CI) pullBaseTrusted(ctx context.Context, repo domain.RepoName, number int64, action, target string, base domain.ObjectID) bool {
334 if c.refTrusted(ctx, repo, pinnedRef{ref: "refs/heads/" + target, commit: base, object: base}) {
335 return true
336 }
337 ctxLog(ctx).Info().Str("repo", repo.String()).Int64("number", number).Str("action", action).Str("base", base.String()).
338 Msg("ci pull request runs are not trusted: the base was not pushed by a trusted principal")
339 return false
340}
341
342// onBranchesMoved is called for every push and merge, after its push
343// workflows are queued (so a push run never waits for this). It takes the
344// snapshot of the push for the merge requests it touches, here and now: the open
345// merge requests are listed (a scan), and for those whose source branch was
346// pushed a synchronize event is queued carrying the pushed head, the target tip
347// as it is at this moment (read once per target for the push), and the actor and
348// trust of the push. The event runs for exactly that snapshot, however the
349// branches move before it is handled: a later push to the target, possibly by
350// someone less trusted, changes nothing about what a trusted run executes.
351// The refresh of the refs of every merge request touched (those that have a
352// pushed branch as source or as target, from the two indexes) is queued for the
353// refresh worker, which is the only fully asynchronous part. A push that touches
354// no merge request queues nothing and marks nothing; the whole-repository rescan
355// is for overflow, truncation and startup.
356func (c *CI) onBranchesMoved(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, actor string, pusher domain.CIPusher, trusted, run bool) {
357 if !c.pullEnabled() {
358 return
359 }
360 moved := map[string]domain.RefUpdate{}
361 for _, u := range updates {
362 if b, ok := strings.CutPrefix(u.Ref, "refs/heads/"); ok && !u.IsDelete() {
363 moved[b] = u
364 }
365 }
366 if len(moved) == 0 {
367 return
368 }
369 c.stage(pullStageDiscover)
370 log := zerowrap.FromCtx(ctx)
371 branches := make([]string, 0, len(moved))
372 for b := range moved {
373 branches = append(branches, b)
374 }
375
376 // The requests whose branch is the target of a push (a target move): their
377 // refs are queued for a refresh. Only these: an unrelated push queues nothing
378 // and marks nothing. The lookup is bounded like the snapshot; past the bound
379 // or on an error the repository is marked dirty, and its rescan finds the
380 // stale ones.
381 targets, tgtTruncated, err := c.cfg.MRs.OpenByTarget(ctx, repo, branches, c.cfg.PullSnapshot)
382 switch {
383 case err != nil:
384 log.Error().Err(err).Str("repo", repo.String()).Msg("ci: look up the merge requests targeting a pushed branch: the repository is rescanned")
385 c.markPullDirty(repo)
386 case tgtTruncated:
387 log.Error().Str("repo", repo.String()).Int("bound", c.cfg.PullSnapshot).
388 Msg("ci pull request refresh of a push is cut short: more than the bound of merge requests target the pushed branches; the repository is rescanned")
389 c.markPullDirty(repo)
390 }
391 for _, n := range targets {
392 c.queuePullRefresh(pullKey{repo, n})
393 }
394
395 // The requests whose source branch was pushed come from the index of the open
396 // merge requests by source: a scan of the keys of those branches, bounded and
397 // cancellable, never of every merge request the repository ever had. Past the
398 // bound or on cancellation the rest is skipped and logged (the overload policy
399 // of the event queue: those events are lost), and the repository is marked
400 // dirty, so that the refs of the skipped requests are repaired by the rescan.
401 numbers, truncated, err := c.cfg.MRs.OpenBySource(ctx, repo, branches, c.cfg.PullSnapshot)
402 if err != nil {
403 log.Error().Err(err).Str("repo", repo.String()).Msg("ci: look up the merge requests of a push: its pull_request runs are not started; the repository is rescanned")
404 c.markPullDirty(repo)
405 if len(numbers) == 0 {
406 return
407 }
408 } else if truncated {
409 log.Error().Str("repo", repo.String()).Int("bound", c.cfg.PullSnapshot).
410 Msg("ci pull request snapshot of a push is cut short: more than the bound of merge requests have a pushed branch as source; the pull_request runs of the others are not started and the repository is rescanned")
411 c.markPullDirty(repo)
412 }
413 tips := map[string]domain.ObjectID{}
414 baseTrusted := map[string]bool{}
415 var target *defaultTip // the default tip of the snapshot, read once
416 for _, n := range numbers {
417 mr, err := c.cfg.MRs.Get(ctx, repo, n)
418 if err != nil || !mr.IsOpen() {
419 continue // closed or gone since the index was read
420 }
421 u, srcMoved := moved[mr.Source]
422 if !srcMoved {
423 continue
424 }
425 if run {
426 if head, herr := domain.ParseObjectID(u.New); herr == nil {
427 base, seen := tips[mr.Target]
428 if !seen {
429 base, _ = c.cfg.Git.BranchHead(ctx, repo, mr.Target)
430 tips[mr.Target] = base
431 }
432 before, _ := domain.ParseObjectID(u.Old) // zero for a new branch
433 // The pusher's trust covers the head; the base of the snapshot
434 // must have been pushed by a trusted principal too.
435 bt, checked := baseTrusted[mr.Target]
436 if trusted && !checked {
437 bt = c.pullBaseTrusted(ctx, repo, mr.Number, domain.PullSynchronize, mr.Target, base)
438 baseTrusted[mr.Target] = bt
439 }
440 evTrusted := trusted && bt
441 if target == nil {
442 t := c.defaultTipOf(ctx, repo)
443 target = &t
444 }
445 c.queuePullEvent(pullEvent{
446 repo: repo, number: mr.Number, action: domain.PullSynchronize, head: head, base: base, before: before,
447 actor: actor, pusher: pusher, trusted: evTrusted, actorTrusted: trusted, target: *target, run: true, log: log,
448 })
449 }
450 }
451 c.queuePullRefresh(pullKey{repo, mr.Number})
452 }
453}
454
455// markPullDirty marks a repository for a rescan of its merge requests.
456func (c *CI) markPullDirty(repo domain.RepoName) {
457 c.pull.mu.Lock()
458 c.pull.markDirty(repo)
459 c.pull.mu.Unlock()
460 c.wakePull()
461}
462
463// markDirty records that the merge requests of a repository may be stale. The
464// caller holds mu.
465func (q *pullQueue) markDirty(repo domain.RepoName) {
466 if !q.dirty[repo] {
467 q.dirty[repo] = true
468 q.dirtyOrder = append(q.dirtyOrder, repo)
469 }
470}
471
472func (c *CI) wakePull() {
473 select {
474 case c.pull.wake <- struct{}{}:
475 default:
476 }
477}
478
479// queuePullEvent queues an activity. A full queue drops it and logs an error,
480// exactly as the queue of the push triggers does (CI.enqueue): CI never slows
481// or fails a push. The refresh cap does not apply to events.
482func (c *CI) queuePullEvent(e pullEvent) {
483 c.mu.Lock()
484 closed := c.closed
485 if !closed {
486 c.pulls[pullKey{repo: e.repo, number: e.number}]++
487 }
488 c.mu.Unlock()
489 if closed {
490 return
491 }
492 select {
493 case c.pull.events <- e:
494 default:
495 c.finishPull(pullKey{repo: e.repo, number: e.number})
496 e.log.Error().Str("repo", e.repo.String()).Int64("number", e.number).Str("action", e.action).Msg("ci pull request queue is full: this activity of a merge request is not run")
497 }
498}
499
500// queuePullRefresh asks for the refs of a merge request to be brought up to
501// date; asking again while it waits changes nothing. When the queue is full the
502// repository is marked dirty instead: the worker rescans its open merge requests
503// once it has capacity.
504func (c *CI) queuePullRefresh(k pullKey) {
505 q := c.pull
506 q.mu.Lock()
507 switch {
508 case q.pending[k]:
509 case len(q.pending) < q.max:
510 q.pending[k] = true
511 q.order = append(q.order, k)
512 default:
513 q.markDirty(k.repo)
514 }
515 q.mu.Unlock()
516 c.wakePull()
517}
518
519// nextPullWork takes what the refresh worker does next: a refresh, or a dirty
520// repository to rescan, in that order.
521func (c *CI) nextPullWork() (key *pullKey, dirty *domain.RepoName) {
522 q := c.pull
523 q.mu.Lock()
524 defer q.mu.Unlock()
525 switch {
526 case len(q.order) > 0:
527 k := q.order[0]
528 q.order = q.order[1:]
529 delete(q.pending, k)
530 return &k, nil
531 case len(q.dirtyOrder) > 0:
532 r := q.dirtyOrder[0]
533 q.dirtyOrder = q.dirtyOrder[1:]
534 delete(q.dirty, r)
535 return nil, &r
536 }
537 return nil, nil
538}
539
540// RescanPulls rescans the merge requests of a repository now, on the calling
541// goroutine, as the refresh worker does for a repository it marked dirty. It is
542// for tests and operators' tooling; the server never needs to call it.
543func (c *CI) RescanPulls(ctx context.Context, repo domain.RepoName) error {
544 return c.rescanPulls(ctx, repo)
545}
546
547// rescanPulls checks the open merge requests of a repository against their refs
548// and refreshes, one after the other and here (not through the bounded queue, so
549// a rescan never marks its repository dirty again), each whose
550// refs/pull/<n>/head is not the tip of its source branch or whose merge ref is
551// not a merge of the two current tips. A merge request whose current state was
552// already tried (a head that conflicts has no merge ref, so it never looks
553// current) is skipped until one of its branches moves. It repairs what a
554// saturated queue, a failed discovery or a restart left behind.
555func (c *CI) rescanPulls(ctx context.Context, repo domain.RepoName) error {
556 c.stage(pullStageRescan)
557 mrs, err := c.cfg.MRs.ListOpen(ctx, repo)
558 if err != nil {
559 return fmt.Errorf("list the open merge requests: %w", err)
560 }
561 for _, mr := range mrs {
562 if ctx.Err() != nil {
563 return nil
564 }
565 head, herr := c.cfg.Git.BranchHead(ctx, repo, mr.Source)
566 base, berr := c.cfg.Git.BranchHead(ctx, repo, mr.Target)
567 if herr != nil || berr != nil {
568 continue // a branch is gone: nothing to publish
569 }
570 if c.pull.attempts.tried(pullKey{repo, mr.Number}, base, head) || c.pullRefsCurrent(ctx, mr, head, base) {
571 continue
572 }
573 if err := c.refreshPull(ctx, pullKey{repo, mr.Number}); err != nil && !errors.Is(err, errPullClosed) {
574 ctxLog(ctx).Error().Err(err).Str("repo", repo.String()).Int64("number", mr.Number).Msg("ci: refresh the refs of a merge request in a rescan")
575 }
576 }
577 return nil
578}
579
580// pullRefsCurrent reports whether the head ref is head and the merge ref, if
581// there is one, merges base and head. A head that conflicts has no merge ref and
582// counts as current once the head ref is right: the next push or move tries again.
583func (c *CI) pullRefsCurrent(ctx context.Context, mr domain.MergeRequest, head, base domain.ObjectID) bool {
584 if cur, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil || cur != head {
585 return false
586 }
587 merge, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number))
588 if err != nil {
589 return false // none: a refresh decides whether the merge is clean
590 }
591 cs, _, err := c.cfg.Git.ListRange(ctx, mr.Repo, domain.ObjectID{}, merge, 1)
592 return err == nil && len(cs) == 1 && len(cs[0].Parents) == 2 && cs[0].Parents[0] == base && cs[0].Parents[1] == head
593}
594
595// rescanAllPulls marks dirty every repository that has open merge requests, at
596// startup: the refs may be stale after a restart or a crash.
597func (c *CI) rescanAllPulls(ctx context.Context) {
598 repos, err := c.cfg.Meta.List(ctx)
599 if err != nil {
600 ctxLog(ctx).Error().Err(err).Msg("ci: list the repositories to rescan their merge requests")
601 return
602 }
603 q := c.pull
604 q.mu.Lock()
605 for _, r := range repos {
606 q.markDirty(r.Name)
607 }
608 q.mu.Unlock()
609 c.wakePull()
610}
611
612// pullLoop is the event worker.
613func (c *CI) pullLoop() {
614 defer c.wg.Done()
615 for {
616 select {
617 case <-c.ctx.Done():
618 return
619 case e := <-c.pull.events:
620 c.guardPull(e.log, e.repo, func(ctx context.Context) error { return c.handlePullEvent(ctx, e) })
621 c.finishPull(pullKey{repo: e.repo, number: e.number})
622 c.kickAutoMerge(c.ctx, e.repo)
623 }
624 }
625}
626
627// refreshLoop is the refresh worker: it refreshes the refs of the merge
628// requests a push touched, and rescans the repositories that fell behind.
629func (c *CI) refreshLoop() {
630 defer c.wg.Done()
631 c.rescanAllPulls(c.ctx)
632 for {
633 key, dirty := c.nextPullWork()
634 if c.ctx.Err() != nil {
635 return
636 }
637 switch {
638 case key != nil:
639 c.guardPull(zerowrap.FromCtx(c.ctx), key.repo, func(ctx context.Context) error { return c.refreshPull(ctx, *key) })
640 case dirty != nil:
641 c.guardPull(zerowrap.FromCtx(c.ctx), *dirty, func(ctx context.Context) error { return c.rescanPulls(ctx, *dirty) })
642 default:
643 select {
644 case <-c.ctx.Done():
645 return
646 case <-c.pull.wake:
647 }
648 }
649 }
650}
651
652// guardPull runs fn; a panic or an error is logged and never escapes.
653func (c *CI) guardPull(log zerowrap.Logger, repo domain.RepoName, fn func(context.Context) error) {
654 ctx := zerowrap.WithCtx(c.ctx, log)
655 defer func() {
656 if r := recover(); r != nil {
657 log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Str("repo", repo.String()).Msg("ci pull request work panicked")
658 }
659 }()
660 if err := fn(ctx); err != nil && ctx.Err() == nil && !errors.Is(err, errPullClosed) {
661 log.Error().Err(err).Str("repo", repo.String()).Msg("ci pull request work failed")
662 }
663}
664
665// whileOpen runs fn under the merge lock while the request is open; errPullClosed
666// when it is not. Without a gate (tests of the CI alone) it checks and runs.
667func (c *CI) whileOpen(ctx context.Context, repo domain.RepoName, number int64, fn func() error) error {
668 c.pull.mu.Lock()
669 gate := c.pull.gate
670 c.pull.mu.Unlock()
671 var err error
672 if gate != nil {
673 err = gate.WhileOpen(ctx, repo, number, fn)
674 } else if mr, gerr := c.cfg.MRs.Get(ctx, repo, number); gerr != nil {
675 err = gerr
676 } else if !mr.IsOpen() {
677 err = domain.ErrPreconditionFailed
678 } else {
679 err = fn()
680 }
681 if errors.Is(err, domain.ErrPreconditionFailed) || errors.Is(err, domain.ErrNotFound) {
682 return errPullClosed
683 }
684 return err
685}
686
687// createRun stores a run. The run of an open merge request is stored under the
688// gate, so that one that is closed meanwhile gets none (errPullClosed).
689func (c *CI) createRun(ctx context.Context, f pushFacts, repo domain.RepoName, wfPath string, build func(id, number int64) ports.CIRunDraft) (ports.CIRunDraft, error) {
690 off, err := c.workflowDisabled(ctx, repo, wfPath)
691 if err != nil {
692 return ports.CIRunDraft{}, err
693 }
694 if off {
695 return ports.CIRunDraft{}, domain.ErrWorkflowDisabled
696 }
697
698 if f.pull == nil || f.pull.action == domain.PullClosed {
699 return c.cfg.Store.CreateRun(ctx, repo, wfPath, build)
700 }
701 var d ports.CIRunDraft
702 err = c.whileOpen(ctx, repo, f.pull.mr.Number, func() error {
703 var err error
704 d, err = c.cfg.Store.CreateRun(ctx, repo, wfPath, build)
705 return err
706 })
707 return d, err
708}
709
710// pullFacts is what a pull_request run is for.
711type pullFacts struct {
712 mr domain.MergeRequest
713 action string
714 // head and base are the commits of the event; the merge commit of the run
715 // has them as parents.
716 head, base domain.ObjectID
717 // before is the head before a synchronize.
718 before domain.ObjectID
719 merged bool
720 // merge is the test merge commit of head and base, zero when there is
721 // none (a conflict, or a closed request that never merged cleanly).
722 merge domain.ObjectID
723 // target is set for a pull_request_target run.
724 target bool
725}
726
727// pullState is the merge of a head into a base.
728type pullState struct {
729 head, base domain.ObjectID
730 // merge is the test merge commit of head and base when clean; for a closed
731 // request, the last one published, zero if none.
732 merge domain.ObjectID
733 clean bool
734 // otherHead is set for a closed request whose last merge commit is of
735 // another head than its event's: the provenance of the event's head says
736 // nothing of the code the run executes.
737 otherHead bool
738}
739
740func (c *CI) handlePullEvent(ctx context.Context, e pullEvent) error {
741 c.stage(pullStageEvent)
742 mr, err := c.cfg.MRs.Get(ctx, e.repo, e.number)
743 if errors.Is(err, domain.ErrNotFound) {
744 return nil
745 }
746 if err != nil {
747 return err
748 }
749 switch e.action {
750 case domain.PullOpened, domain.PullSynchronize:
751 if !mr.IsOpen() {
752 return nil // closed meanwhile: the closed event follows
753 }
754 st, err := c.mergeOf(ctx, mr, e.head, e.base)
755 if err != nil {
756 return err
757 }
758 c.stage(pullStageComputed)
759 if err := c.publishPull(ctx, mr, st); err != nil || !e.run {
760 return err
761 }
762 var err1 error
763 if st.clean { // a conflict makes no pull_request run, as on GitHub
764 err1 = c.startPullRuns(ctx, e, mr, st, false)
765 }
766 return errors.Join(err1, c.startPullRuns(ctx, e, mr, st, true))
767 case domain.PullClosed:
768 if mr.IsOpen() || !e.run {
769 return nil
770 }
771 st, ok, err := c.closedPull(ctx, e, mr)
772 if err != nil || !ok {
773 return err
774 }
775 pe := e
776 if st.otherHead || st.merge != e.merge {
777 pe.trusted = false // not the code the trust was checked for
778 }
779 return errors.Join(c.startPullRuns(ctx, pe, mr, st, false), c.startPullRuns(ctx, e, mr, st, true))
780 }
781 return nil
782}
783
784// refreshPull brings the refs of a merge request to its branches as they are.
785func (c *CI) refreshPull(ctx context.Context, k pullKey) error {
786 c.stage(pullStageRefresh)
787 mr, err := c.cfg.MRs.Get(ctx, k.repo, k.number)
788 if errors.Is(err, domain.ErrNotFound) || (err == nil && !mr.IsOpen()) {
789 return nil
790 }
791 if err != nil {
792 return err
793 }
794 head, err := c.cfg.Git.BranchHead(ctx, k.repo, mr.Source)
795 if errors.Is(err, domain.ErrNotFound) {
796 return nil
797 }
798 if err != nil {
799 return err
800 }
801 base, err := c.cfg.Git.BranchHead(ctx, k.repo, mr.Target)
802 if errors.Is(err, domain.ErrNotFound) {
803 return nil
804 }
805 if err != nil {
806 return err
807 }
808 st, err := c.mergeOf(ctx, mr, head, base)
809 if err != nil {
810 return err
811 }
812 return c.publishPull(ctx, mr, st)
813}
814
815// mergeOf is the test merge of head into base. It computes the commit only; a
816// merge ref that is already that merge is reused.
817func (c *CI) mergeOf(ctx context.Context, mr domain.MergeRequest, head, base domain.ObjectID) (pullState, error) {
818 st := pullState{head: head, base: base}
819 if head.IsZero() || base.IsZero() {
820 return st, nil
821 }
822 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err == nil {
823 cs, _, err := c.cfg.Git.ListRange(ctx, mr.Repo, domain.ObjectID{}, old, 1)
824 if err == nil && len(cs) == 1 && len(cs[0].Parents) == 2 && cs[0].Parents[0] == base && cs[0].Parents[1] == head {
825 st.merge, st.clean = old, true
826 return st, nil
827 }
828 } else if !errors.Is(err, domain.ErrNotFound) {
829 return st, err
830 }
831 tree, clean, err := c.cfg.Pull.TestMerge(ctx, mr.Repo, base, head)
832 k := pullKey{mr.Repo, mr.Number}
833 if err != nil {
834 return st, err
835 }
836 if !clean {
837 c.pull.attempts.conflicted(k, base, head) // a rescan does not try these tips again
838 return st, nil
839 }
840 c.pull.attempts.merged(k)
841 committer := c.cfg.Committer
842 if committer.Name == "" {
843 committer = domain.Identity{Name: "ermite", Email: "ermite@localhost"}
844 }
845 id, err := c.cfg.Git.WriteCommit(ctx, mr.Repo, domain.TestMergeCommit(tree, base, head, committer, c.now()))
846 if err != nil {
847 return st, err
848 }
849 st.merge, st.clean = id, true
850 return st, nil
851}
852
853// publishPull points refs/pull/<n>/head and /merge at st, under the gate. A ref
854// moves only to the latest: the head ref when st.head is still the head of the
855// source branch, the merge ref when the target tip is st.base too, so that an
856// older event never moves a ref back. errPullClosed when the request is closed.
857func (c *CI) publishPull(ctx context.Context, mr domain.MergeRequest, st pullState) error {
858 if st.head.IsZero() {
859 return nil
860 }
861 changed := false
862 err := c.whileOpen(ctx, mr.Repo, mr.Number, func() error {
863 curHead, err := c.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
864 if err != nil || curHead != st.head {
865 return ignoreNotFound(err)
866 }
867 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil || old != st.head {
868 if err != nil && !errors.Is(err, domain.ErrNotFound) {
869 return err
870 }
871 if err := c.cfg.Pull.SetPullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number), st.head); err != nil {
872 return err
873 }
874 changed = true
875 }
876 if !st.clean {
877 return nil
878 }
879 curBase, err := c.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
880 if err != nil || curBase != st.base {
881 return ignoreNotFound(err)
882 }
883 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err != nil || old != st.merge {
884 if err != nil && !errors.Is(err, domain.ErrNotFound) {
885 return err
886 }
887 if err := c.cfg.Pull.SetPullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number), st.merge); err != nil {
888 return err
889 }
890 changed = true
891 }
892 return nil
893 })
894 if changed && c.cfg.Fresh != nil {
895 c.cfg.Fresh.Refresh(ctx, mr.Repo)
896 }
897 return err
898}
899
900func ignoreNotFound(err error) error {
901 if errors.Is(err, domain.ErrNotFound) {
902 return nil
903 }
904 return err
905}
906
907// closedPull is the state a closed or merged merge request runs with: the last
908// test merge commit published, or the head when there is none. Nothing is
909// computed again. ok is false when the request has neither (its branch never
910// existed).
911func (c *CI) closedPull(ctx context.Context, e pullEvent, mr domain.MergeRequest) (st pullState, ok bool, err error) {
912 if st.merge, err = c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err != nil && !errors.Is(err, domain.ErrNotFound) {
913 return st, false, err
914 }
915 st.head = e.head
916 if st.head.IsZero() {
917 st.head = mr.HeadID
918 }
919 if st.head.IsZero() {
920 if st.head, err = c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil && !errors.Is(err, domain.ErrNotFound) {
921 return st, false, err
922 }
923 }
924 st.base = e.base
925 if !st.merge.IsZero() {
926 if cs, _, err := c.cfg.Git.ListRange(ctx, mr.Repo, domain.ObjectID{}, st.merge, 1); err == nil && len(cs) == 1 && len(cs[0].Parents) == 2 {
927 st.base = cs[0].Parents[0]
928 if st.head.IsZero() {
929 st.head = cs[0].Parents[1]
930 }
931 st.otherHead = cs[0].Parents[1] != e.head
932 }
933 }
934 if st.base.IsZero() {
935 st.base = mr.BaseID
936 }
937 st.clean = true
938 return st, !st.head.IsZero(), nil
939}
940
941// startPullRuns starts the pull_request workflows of the merge request that
942// match the activity and the files it changes, or with target its
943// pull_request_target workflows: those of the default tip of the event.
944func (c *CI) startPullRuns(ctx context.Context, e pullEvent, mr domain.MergeRequest, st pullState, target bool) error {
945 if target && (e.target.commit.IsZero() || st.head.IsZero()) {
946 return nil // no default branch, or no head to run for
947 }
948 runners, err := c.cfg.Store.ListRunners(ctx)
949 if err != nil {
950 return fmt.Errorf("list the runners: %w", err)
951 }
952 if len(runners) == 0 {
953 return nil
954 }
955 repo, err := c.cfg.Meta.Get(ctx, e.repo)
956 if err != nil {
957 return nil // deleted meanwhile
958 }
959 t := e.trigger(c.now())
960 ref, sha := domain.PullMergeRef(mr.Number), st.merge
961 switch {
962 case target:
963 // The code is the default tip's: the cause and that tip decide.
964 ref, sha = "refs/heads/"+e.target.branch, e.target.commit
965 t.trusted = e.actorTrusted && e.target.trusted
966 if e.actorTrusted && !e.target.trusted {
967 ctxLog(ctx).Info().Str("repo", e.repo.String()).Int64("number", mr.Number).Str("action", e.action).Str("tip", sha.String()).
968 Msg("ci pull_request_target runs are not trusted: the default branch tip was not pushed by a trusted principal")
969 }
970 case sha.IsZero():
971 ref, sha = domain.PullHeadRef(mr.Number), st.head
972 }
973 pf := &pullFacts{
974 mr: mr, action: e.action, head: st.head, base: st.base, before: e.before, merged: mr.State == domain.MRStateMerged,
975 merge: st.merge, target: target,
976 }
977 f := pushFacts{
978 update: domain.RefUpdate{Ref: ref, Old: domain.ZeroOID, New: sha.String()}, commit: sha, pull: pf,
979 }
980 if def, _, err := c.cfg.Repos.DefaultBranch(ctx, e.repo); err == nil {
981 f.defaultBranch = def
982 }
983 head, _, err := c.cfg.Git.ListRange(ctx, e.repo, domain.ObjectID{}, st.head, 1)
984 if err != nil || len(head) == 0 {
985 return fmt.Errorf("read the head commit of !%d: %w", mr.Number, err)
986 }
987 f.head = head[0]
988 // The files of the merge request are those between the merge base of the
989 // target and the head, and the head: what the review shows.
990 f.filesTruncated = true // unavailable unless read below: a paths filter then passes
991 if !st.base.IsZero() {
992 if mb, err := c.cfg.Git.MergeBase(ctx, e.repo, st.base, st.head); err == nil {
993 if paths, truncated, err := c.cfg.Diff.ChangedPaths(ctx, e.repo, mb, st.head, workflow.MaxFilesEvaluated); err == nil {
994 f.files, f.filesTruncated = paths, truncated
995 }
996 }
997 }
998 files, err := c.workflowFiles(ctx, e.repo, sha)
999 if err != nil || len(files) == 0 {
1000 return err
1001 }
1002 if !target && workflow.SkipCI(f.head.Message) {
1003 return nil // GitHub's skip directives are for push and pull_request only
1004 }
1005 var errs []error
1006 for _, file := range files {
1007 if file.problem != "" {
1008 continue // reported by the push of the branch
1009 }
1010 wf, perrs := parseWorkflow(file.path, file.content)
1011 if len(perrs) > 0 || !matchPullRequest(wf, f.eventName(), e.action, mr.Target, f.files, f.filesTruncated) {
1012 continue
1013 }
1014 if problem := c.matrixProblem(ctx, repo, t, f, file.path, wf); problem.text != "" {
1015 name := file.path
1016 if problem.named {
1017 name = workflowName(wf, file.path)
1018 }
1019 errs = append(errs, c.startInvalid(ctx, repo, t, f, file.path, name, problem.text))
1020 continue
1021 }
1022 errs = append(errs, c.startRun(ctx, repo, t, f, file.path, string(file.content), wf))
1023 }
1024 return errors.Join(errs...)
1025}
1026
1027// matchPullRequest reports whether the workflow runs for the activity of a merge
1028// request into base: the event (pull_request or pull_request_target) is
1029// listed, the activity is one of its types (opened, synchronize and reopened
1030// when it lists none), and the branches and paths filters, which apply to the
1031// base branch and the changed files, let it through. The filters are those of
1032// the push event (ermite-actions), run on the event under another name.
1033func matchPullRequest(wf *workflow.Workflow, event, action, base string, files []string, truncated bool) bool {
1034 if wf == nil || wf.On == nil {
1035 return false
1036 }
1037 ev := wf.On.Event(event)
1038 if ev == nil {
1039 return false
1040 }
1041 types := ev.Types
1042 if types == nil {
1043 types = domain.DefaultPullRequestTypes
1044 }
1045 if !slices.Contains(types, action) {
1046 return false
1047 }
1048 as := *ev
1049 as.Name = "push"
1050 probe := &workflow.Workflow{On: &workflow.On{Events: []*workflow.Event{&as}}}
1051 return probe.MatchPush(workflow.PushInfo{Ref: "refs/heads/" + base, ChangedFiles: files, FilesTruncated: truncated})
1052}
1053
1054// --- the github context ---
1055
1056// pullGithubJSON builds the github context of a pull_request run: github.ref is
1057// the merge ref, github.sha the merge commit (the head for a closed request
1058// that never merged), and the event is shaped like GitHub's pull_request
1059// webhook payload with the fields ermite can fill. A pull_request_target run
1060// has the same payload under its own event name, on the default branch.
1061func (c *CI) pullGithubJSON(repo domain.Repo, run domain.CIRun, f pushFacts) string {
1062 p := f.pull
1063 mr := p.mr
1064 u := c.urls(repo)
1065 id := c.identity(run.Actor)
1066 owner := c.identity("owner")
1067 repoObj := c.repoObj(repo, u, f.defaultBranch, run.CreatedAt)
1068 num := strconv.FormatInt(mr.Number, 10)
1069 prAPI := u.api + "/repos/" + u.full + "/pulls/" + num
1070 html := u.repo + "/merge-requests/" + num
1071 side := func(branch string, sha domain.ObjectID) map[string]any {
1072 return map[string]any{
1073 "label": u.owner + ":" + branch, "ref": branch, "sha": sha.String(),
1074 "user": u.userObj(owner, false), "repo": repoObj,
1075 }
1076 }
1077 tm := func(t time.Time) any {
1078 if t.IsZero() {
1079 return nil
1080 }
1081 return t.UTC().Format(time.RFC3339)
1082 }
1083 var body any
1084 if s := mr.Description.String(); s != "" {
1085 body = s
1086 }
1087 state, closedAt, mergedAt := "open", any(nil), any(nil)
1088 mergeSha := any(nil)
1089 if !mr.IsOpen() {
1090 state, closedAt = "closed", tm(mr.UpdatedAt)
1091 }
1092 switch {
1093 case p.merged:
1094 mergedAt = tm(mr.UpdatedAt)
1095 mergeSha = mr.MergedCommitID.String()
1096 case !p.merge.IsZero() && p.merge != p.head:
1097 mergeSha = p.merge.String() // the test merge commit
1098 }
1099 prID := stableID("pull", u.full+"#"+num)
1100 pr := map[string]any{
1101 "url": prAPI, "id": prID, "node_id": nodeID("011", "PullRequest", prID), "html_url": html,
1102 "diff_url": html + ".diff", "patch_url": html + ".patch", "issue_url": u.api + "/repos/" + u.full + "/issues/" + num,
1103 "number": mr.Number, "state": state, "locked": false, "title": mr.Title.String(), "user": u.userObj(c.identity(mr.Author), false),
1104 "body": body, "created_at": tm(mr.CreatedAt), "updated_at": tm(mr.UpdatedAt), "closed_at": closedAt, "merged_at": mergedAt,
1105 "merge_commit_sha": mergeSha, "assignee": nil, "assignees": []any{}, "requested_reviewers": []any{},
1106 "requested_teams": []any{}, "labels": []any{}, "milestone": nil, "draft": false,
1107 "head": side(mr.Source, p.head), "base": side(mr.Target, p.base), "author_association": "OWNER",
1108 "auto_merge": nil, "active_lock_reason": nil, "merged": p.merged,
1109 }
1110 event := map[string]any{
1111 "action": p.action, "number": mr.Number, "pull_request": pr, "repository": repoObj, "sender": u.userObj(id, false),
1112 }
1113 if p.action == domain.PullSynchronize {
1114 before := p.before.String()
1115 if p.before.IsZero() {
1116 before = domain.ZeroOID
1117 }
1118 event["before"], event["after"] = before, p.head.String()
1119 }
1120 refName := strings.TrimPrefix(run.Ref, domain.PullRefPrefix)
1121 if p.target {
1122 refName = strings.TrimPrefix(run.Ref, "refs/heads/")
1123 }
1124 return c.githubContext(repo, run, ghEvent{
1125 name: f.eventName(), refName: refName, refType: "branch",
1126 headRef: mr.Source, baseRef: mr.Target, payload: event,
1127 })
1128}