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: authorOf(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 c.mu.Unlock()
486 if closed {
487 return
488 }
489 select {
490 case c.pull.events <- e:
491 default:
492 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")
493 }
494}
495
496// queuePullRefresh asks for the refs of a merge request to be brought up to
497// date; asking again while it waits changes nothing. When the queue is full the
498// repository is marked dirty instead: the worker rescans its open merge requests
499// once it has capacity.
500func (c *CI) queuePullRefresh(k pullKey) {
501 q := c.pull
502 q.mu.Lock()
503 switch {
504 case q.pending[k]:
505 case len(q.pending) < q.max:
506 q.pending[k] = true
507 q.order = append(q.order, k)
508 default:
509 q.markDirty(k.repo)
510 }
511 q.mu.Unlock()
512 c.wakePull()
513}
514
515// nextPullWork takes what the refresh worker does next: a refresh, or a dirty
516// repository to rescan, in that order.
517func (c *CI) nextPullWork() (key *pullKey, dirty *domain.RepoName) {
518 q := c.pull
519 q.mu.Lock()
520 defer q.mu.Unlock()
521 switch {
522 case len(q.order) > 0:
523 k := q.order[0]
524 q.order = q.order[1:]
525 delete(q.pending, k)
526 return &k, nil
527 case len(q.dirtyOrder) > 0:
528 r := q.dirtyOrder[0]
529 q.dirtyOrder = q.dirtyOrder[1:]
530 delete(q.dirty, r)
531 return nil, &r
532 }
533 return nil, nil
534}
535
536// RescanPulls rescans the merge requests of a repository now, on the calling
537// goroutine, as the refresh worker does for a repository it marked dirty. It is
538// for tests and operators' tooling; the server never needs to call it.
539func (c *CI) RescanPulls(ctx context.Context, repo domain.RepoName) error {
540 return c.rescanPulls(ctx, repo)
541}
542
543// rescanPulls checks the open merge requests of a repository against their refs
544// and refreshes, one after the other and here (not through the bounded queue, so
545// a rescan never marks its repository dirty again), each whose
546// refs/pull/<n>/head is not the tip of its source branch or whose merge ref is
547// not a merge of the two current tips. A merge request whose current state was
548// already tried (a head that conflicts has no merge ref, so it never looks
549// current) is skipped until one of its branches moves. It repairs what a
550// saturated queue, a failed discovery or a restart left behind.
551func (c *CI) rescanPulls(ctx context.Context, repo domain.RepoName) error {
552 c.stage(pullStageRescan)
553 mrs, err := c.cfg.MRs.ListOpen(ctx, repo)
554 if err != nil {
555 return fmt.Errorf("list the open merge requests: %w", err)
556 }
557 for _, mr := range mrs {
558 if ctx.Err() != nil {
559 return nil
560 }
561 head, herr := c.cfg.Git.BranchHead(ctx, repo, mr.Source)
562 base, berr := c.cfg.Git.BranchHead(ctx, repo, mr.Target)
563 if herr != nil || berr != nil {
564 continue // a branch is gone: nothing to publish
565 }
566 if c.pull.attempts.tried(pullKey{repo, mr.Number}, base, head) || c.pullRefsCurrent(ctx, mr, head, base) {
567 continue
568 }
569 if err := c.refreshPull(ctx, pullKey{repo, mr.Number}); err != nil && !errors.Is(err, errPullClosed) {
570 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")
571 }
572 }
573 return nil
574}
575
576// pullRefsCurrent reports whether the head ref is head and the merge ref, if
577// there is one, merges base and head. A head that conflicts has no merge ref and
578// counts as current once the head ref is right: the next push or move tries again.
579func (c *CI) pullRefsCurrent(ctx context.Context, mr domain.MergeRequest, head, base domain.ObjectID) bool {
580 if cur, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil || cur != head {
581 return false
582 }
583 merge, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number))
584 if err != nil {
585 return false // none: a refresh decides whether the merge is clean
586 }
587 cs, _, err := c.cfg.Git.ListRange(ctx, mr.Repo, domain.ObjectID{}, merge, 1)
588 return err == nil && len(cs) == 1 && len(cs[0].Parents) == 2 && cs[0].Parents[0] == base && cs[0].Parents[1] == head
589}
590
591// rescanAllPulls marks dirty every repository that has open merge requests, at
592// startup: the refs may be stale after a restart or a crash.
593func (c *CI) rescanAllPulls(ctx context.Context) {
594 repos, err := c.cfg.Meta.List(ctx)
595 if err != nil {
596 ctxLog(ctx).Error().Err(err).Msg("ci: list the repositories to rescan their merge requests")
597 return
598 }
599 q := c.pull
600 q.mu.Lock()
601 for _, r := range repos {
602 q.markDirty(r.Name)
603 }
604 q.mu.Unlock()
605 c.wakePull()
606}
607
608// pullLoop is the event worker.
609func (c *CI) pullLoop() {
610 defer c.wg.Done()
611 for {
612 select {
613 case <-c.ctx.Done():
614 return
615 case e := <-c.pull.events:
616 c.guardPull(e.log, e.repo, func(ctx context.Context) error { return c.handlePullEvent(ctx, e) })
617 }
618 }
619}
620
621// refreshLoop is the refresh worker: it refreshes the refs of the merge
622// requests a push touched, and rescans the repositories that fell behind.
623func (c *CI) refreshLoop() {
624 defer c.wg.Done()
625 c.rescanAllPulls(c.ctx)
626 for {
627 key, dirty := c.nextPullWork()
628 if c.ctx.Err() != nil {
629 return
630 }
631 switch {
632 case key != nil:
633 c.guardPull(zerowrap.FromCtx(c.ctx), key.repo, func(ctx context.Context) error { return c.refreshPull(ctx, *key) })
634 case dirty != nil:
635 c.guardPull(zerowrap.FromCtx(c.ctx), *dirty, func(ctx context.Context) error { return c.rescanPulls(ctx, *dirty) })
636 default:
637 select {
638 case <-c.ctx.Done():
639 return
640 case <-c.pull.wake:
641 }
642 }
643 }
644}
645
646// guardPull runs fn; a panic or an error is logged and never escapes.
647func (c *CI) guardPull(log zerowrap.Logger, repo domain.RepoName, fn func(context.Context) error) {
648 ctx := zerowrap.WithCtx(c.ctx, log)
649 defer func() {
650 if r := recover(); r != nil {
651 log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Str("repo", repo.String()).Msg("ci pull request work panicked")
652 }
653 }()
654 if err := fn(ctx); err != nil && ctx.Err() == nil && !errors.Is(err, errPullClosed) {
655 log.Error().Err(err).Str("repo", repo.String()).Msg("ci pull request work failed")
656 }
657}
658
659// whileOpen runs fn under the merge lock while the request is open; errPullClosed
660// when it is not. Without a gate (tests of the CI alone) it checks and runs.
661func (c *CI) whileOpen(ctx context.Context, repo domain.RepoName, number int64, fn func() error) error {
662 c.pull.mu.Lock()
663 gate := c.pull.gate
664 c.pull.mu.Unlock()
665 var err error
666 if gate != nil {
667 err = gate.WhileOpen(ctx, repo, number, fn)
668 } else if mr, gerr := c.cfg.MRs.Get(ctx, repo, number); gerr != nil {
669 err = gerr
670 } else if !mr.IsOpen() {
671 err = domain.ErrPreconditionFailed
672 } else {
673 err = fn()
674 }
675 if errors.Is(err, domain.ErrPreconditionFailed) || errors.Is(err, domain.ErrNotFound) {
676 return errPullClosed
677 }
678 return err
679}
680
681// createRun stores a run. The run of an open merge request is stored under the
682// gate, so that one that is closed meanwhile gets none (errPullClosed).
683func (c *CI) createRun(ctx context.Context, f pushFacts, repo domain.RepoName, wfPath string, build func(id, number int64) ports.CIRunDraft) (ports.CIRunDraft, error) {
684 if f.pull == nil || f.pull.action == domain.PullClosed {
685 return c.cfg.Store.CreateRun(ctx, repo, wfPath, build)
686 }
687 var d ports.CIRunDraft
688 err := c.whileOpen(ctx, repo, f.pull.mr.Number, func() error {
689 var err error
690 d, err = c.cfg.Store.CreateRun(ctx, repo, wfPath, build)
691 return err
692 })
693 return d, err
694}
695
696// pullFacts is what a pull_request run is for.
697type pullFacts struct {
698 mr domain.MergeRequest
699 action string
700 // head and base are the commits of the event; the merge commit of the run
701 // has them as parents.
702 head, base domain.ObjectID
703 // before is the head before a synchronize.
704 before domain.ObjectID
705 merged bool
706 // merge is the test merge commit of head and base, zero when there is
707 // none (a conflict, or a closed request that never merged cleanly).
708 merge domain.ObjectID
709 // target is set for a pull_request_target run.
710 target bool
711}
712
713// pullState is the merge of a head into a base.
714type pullState struct {
715 head, base domain.ObjectID
716 // merge is the test merge commit of head and base when clean; for a closed
717 // request, the last one published, zero if none.
718 merge domain.ObjectID
719 clean bool
720 // otherHead is set for a closed request whose last merge commit is of
721 // another head than its event's: the provenance of the event's head says
722 // nothing of the code the run executes.
723 otherHead bool
724}
725
726func (c *CI) handlePullEvent(ctx context.Context, e pullEvent) error {
727 c.stage(pullStageEvent)
728 mr, err := c.cfg.MRs.Get(ctx, e.repo, e.number)
729 if errors.Is(err, domain.ErrNotFound) {
730 return nil
731 }
732 if err != nil {
733 return err
734 }
735 switch e.action {
736 case domain.PullOpened, domain.PullSynchronize:
737 if !mr.IsOpen() {
738 return nil // closed meanwhile: the closed event follows
739 }
740 st, err := c.mergeOf(ctx, mr, e.head, e.base)
741 if err != nil {
742 return err
743 }
744 c.stage(pullStageComputed)
745 if err := c.publishPull(ctx, mr, st); err != nil || !e.run {
746 return err
747 }
748 var err1 error
749 if st.clean { // a conflict makes no pull_request run, as on GitHub
750 err1 = c.startPullRuns(ctx, e, mr, st, false)
751 }
752 return errors.Join(err1, c.startPullRuns(ctx, e, mr, st, true))
753 case domain.PullClosed:
754 if mr.IsOpen() || !e.run {
755 return nil
756 }
757 st, ok, err := c.closedPull(ctx, e, mr)
758 if err != nil || !ok {
759 return err
760 }
761 pe := e
762 if st.otherHead || st.merge != e.merge {
763 pe.trusted = false // not the code the trust was checked for
764 }
765 return errors.Join(c.startPullRuns(ctx, pe, mr, st, false), c.startPullRuns(ctx, e, mr, st, true))
766 }
767 return nil
768}
769
770// refreshPull brings the refs of a merge request to its branches as they are.
771func (c *CI) refreshPull(ctx context.Context, k pullKey) error {
772 c.stage(pullStageRefresh)
773 mr, err := c.cfg.MRs.Get(ctx, k.repo, k.number)
774 if errors.Is(err, domain.ErrNotFound) || (err == nil && !mr.IsOpen()) {
775 return nil
776 }
777 if err != nil {
778 return err
779 }
780 head, err := c.cfg.Git.BranchHead(ctx, k.repo, mr.Source)
781 if errors.Is(err, domain.ErrNotFound) {
782 return nil
783 }
784 if err != nil {
785 return err
786 }
787 base, err := c.cfg.Git.BranchHead(ctx, k.repo, mr.Target)
788 if errors.Is(err, domain.ErrNotFound) {
789 return nil
790 }
791 if err != nil {
792 return err
793 }
794 st, err := c.mergeOf(ctx, mr, head, base)
795 if err != nil {
796 return err
797 }
798 return c.publishPull(ctx, mr, st)
799}
800
801// mergeOf is the test merge of head into base. It computes the commit only; a
802// merge ref that is already that merge is reused.
803func (c *CI) mergeOf(ctx context.Context, mr domain.MergeRequest, head, base domain.ObjectID) (pullState, error) {
804 st := pullState{head: head, base: base}
805 if head.IsZero() || base.IsZero() {
806 return st, nil
807 }
808 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err == nil {
809 cs, _, err := c.cfg.Git.ListRange(ctx, mr.Repo, domain.ObjectID{}, old, 1)
810 if err == nil && len(cs) == 1 && len(cs[0].Parents) == 2 && cs[0].Parents[0] == base && cs[0].Parents[1] == head {
811 st.merge, st.clean = old, true
812 return st, nil
813 }
814 } else if !errors.Is(err, domain.ErrNotFound) {
815 return st, err
816 }
817 tree, clean, err := c.cfg.Pull.TestMerge(ctx, mr.Repo, base, head)
818 k := pullKey{mr.Repo, mr.Number}
819 if err != nil {
820 return st, err
821 }
822 if !clean {
823 c.pull.attempts.conflicted(k, base, head) // a rescan does not try these tips again
824 return st, nil
825 }
826 c.pull.attempts.merged(k)
827 committer := c.cfg.Committer
828 if committer.Name == "" {
829 committer = domain.Identity{Name: "ermite", Email: "ermite@localhost"}
830 }
831 id, err := c.cfg.Git.WriteCommit(ctx, mr.Repo, domain.TestMergeCommit(tree, base, head, committer, c.now()))
832 if err != nil {
833 return st, err
834 }
835 st.merge, st.clean = id, true
836 return st, nil
837}
838
839// publishPull points refs/pull/<n>/head and /merge at st, under the gate. A ref
840// moves only to the latest: the head ref when st.head is still the head of the
841// source branch, the merge ref when the target tip is st.base too, so that an
842// older event never moves a ref back. errPullClosed when the request is closed.
843func (c *CI) publishPull(ctx context.Context, mr domain.MergeRequest, st pullState) error {
844 if st.head.IsZero() {
845 return nil
846 }
847 changed := false
848 err := c.whileOpen(ctx, mr.Repo, mr.Number, func() error {
849 curHead, err := c.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
850 if err != nil || curHead != st.head {
851 return ignoreNotFound(err)
852 }
853 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil || old != st.head {
854 if err != nil && !errors.Is(err, domain.ErrNotFound) {
855 return err
856 }
857 if err := c.cfg.Pull.SetPullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number), st.head); err != nil {
858 return err
859 }
860 changed = true
861 }
862 if !st.clean {
863 return nil
864 }
865 curBase, err := c.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
866 if err != nil || curBase != st.base {
867 return ignoreNotFound(err)
868 }
869 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err != nil || old != st.merge {
870 if err != nil && !errors.Is(err, domain.ErrNotFound) {
871 return err
872 }
873 if err := c.cfg.Pull.SetPullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number), st.merge); err != nil {
874 return err
875 }
876 changed = true
877 }
878 return nil
879 })
880 if changed && c.cfg.Fresh != nil {
881 c.cfg.Fresh.Refresh(ctx, mr.Repo)
882 }
883 return err
884}
885
886func ignoreNotFound(err error) error {
887 if errors.Is(err, domain.ErrNotFound) {
888 return nil
889 }
890 return err
891}
892
893// closedPull is the state a closed or merged merge request runs with: the last
894// test merge commit published, or the head when there is none. Nothing is
895// computed again. ok is false when the request has neither (its branch never
896// existed).
897func (c *CI) closedPull(ctx context.Context, e pullEvent, mr domain.MergeRequest) (st pullState, ok bool, err error) {
898 if st.merge, err = c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err != nil && !errors.Is(err, domain.ErrNotFound) {
899 return st, false, err
900 }
901 st.head = e.head
902 if st.head.IsZero() {
903 st.head = mr.HeadID
904 }
905 if st.head.IsZero() {
906 if st.head, err = c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil && !errors.Is(err, domain.ErrNotFound) {
907 return st, false, err
908 }
909 }
910 st.base = e.base
911 if !st.merge.IsZero() {
912 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 {
913 st.base = cs[0].Parents[0]
914 if st.head.IsZero() {
915 st.head = cs[0].Parents[1]
916 }
917 st.otherHead = cs[0].Parents[1] != e.head
918 }
919 }
920 if st.base.IsZero() {
921 st.base = mr.BaseID
922 }
923 st.clean = true
924 return st, !st.head.IsZero(), nil
925}
926
927// startPullRuns starts the pull_request workflows of the merge request that
928// match the activity and the files it changes, or with target its
929// pull_request_target workflows: those of the default tip of the event.
930func (c *CI) startPullRuns(ctx context.Context, e pullEvent, mr domain.MergeRequest, st pullState, target bool) error {
931 if target && (e.target.commit.IsZero() || st.head.IsZero()) {
932 return nil // no default branch, or no head to run for
933 }
934 runners, err := c.cfg.Store.ListRunners(ctx)
935 if err != nil {
936 return fmt.Errorf("list the runners: %w", err)
937 }
938 if len(runners) == 0 {
939 return nil
940 }
941 repo, err := c.cfg.Meta.Get(ctx, e.repo)
942 if err != nil {
943 return nil // deleted meanwhile
944 }
945 t := e.trigger(c.now())
946 ref, sha := domain.PullMergeRef(mr.Number), st.merge
947 switch {
948 case target:
949 // The code is the default tip's: the cause and that tip decide.
950 ref, sha = "refs/heads/"+e.target.branch, e.target.commit
951 t.trusted = e.actorTrusted && e.target.trusted
952 if e.actorTrusted && !e.target.trusted {
953 ctxLog(ctx).Info().Str("repo", e.repo.String()).Int64("number", mr.Number).Str("action", e.action).Str("tip", sha.String()).
954 Msg("ci pull_request_target runs are not trusted: the default branch tip was not pushed by a trusted principal")
955 }
956 case sha.IsZero():
957 ref, sha = domain.PullHeadRef(mr.Number), st.head
958 }
959 pf := &pullFacts{
960 mr: mr, action: e.action, head: st.head, base: st.base, before: e.before, merged: mr.State == domain.MRStateMerged,
961 merge: st.merge, target: target,
962 }
963 f := pushFacts{
964 update: domain.RefUpdate{Ref: ref, Old: domain.ZeroOID, New: sha.String()}, commit: sha, pull: pf,
965 }
966 if def, _, err := c.cfg.Repos.DefaultBranch(ctx, e.repo); err == nil {
967 f.defaultBranch = def
968 }
969 head, _, err := c.cfg.Git.ListRange(ctx, e.repo, domain.ObjectID{}, st.head, 1)
970 if err != nil || len(head) == 0 {
971 return fmt.Errorf("read the head commit of !%d: %w", mr.Number, err)
972 }
973 f.head = head[0]
974 // The files of the merge request are those between the merge base of the
975 // target and the head, and the head: what the review shows.
976 f.filesTruncated = true // unavailable unless read below: a paths filter then passes
977 if !st.base.IsZero() {
978 if mb, err := c.cfg.Git.MergeBase(ctx, e.repo, st.base, st.head); err == nil {
979 if paths, truncated, err := c.cfg.Diff.ChangedPaths(ctx, e.repo, mb, st.head, workflow.MaxFilesEvaluated); err == nil {
980 f.files, f.filesTruncated = paths, truncated
981 }
982 }
983 }
984 files, err := c.workflowFiles(ctx, e.repo, sha)
985 if err != nil || len(files) == 0 {
986 return err
987 }
988 if !target && workflow.SkipCI(f.head.Message) {
989 return nil // GitHub's skip directives are for push and pull_request only
990 }
991 var errs []error
992 for _, file := range files {
993 if file.problem != "" {
994 continue // reported by the push of the branch
995 }
996 wf, perrs := parseWorkflow(file.path, file.content)
997 if len(perrs) > 0 || !matchPullRequest(wf, f.eventName(), e.action, mr.Target, f.files, f.filesTruncated) {
998 continue
999 }
1000 if problem := c.matrixProblem(ctx, repo, t, f, file.path, wf); problem.text != "" {
1001 name := file.path
1002 if problem.named {
1003 name = workflowName(wf, file.path)
1004 }
1005 errs = append(errs, c.startInvalid(ctx, repo, t, f, file.path, name, problem.text))
1006 continue
1007 }
1008 errs = append(errs, c.startRun(ctx, repo, t, f, file.path, string(file.content), wf))
1009 }
1010 return errors.Join(errs...)
1011}
1012
1013// matchPullRequest reports whether the workflow runs for the activity of a merge
1014// request into base: the event (pull_request or pull_request_target) is
1015// listed, the activity is one of its types (opened, synchronize and reopened
1016// when it lists none), and the branches and paths filters, which apply to the
1017// base branch and the changed files, let it through. The filters are those of
1018// the push event (ermite-actions), run on the event under another name.
1019func matchPullRequest(wf *workflow.Workflow, event, action, base string, files []string, truncated bool) bool {
1020 if wf == nil || wf.On == nil {
1021 return false
1022 }
1023 ev := wf.On.Event(event)
1024 if ev == nil {
1025 return false
1026 }
1027 types := ev.Types
1028 if types == nil {
1029 types = domain.DefaultPullRequestTypes
1030 }
1031 if !slices.Contains(types, action) {
1032 return false
1033 }
1034 as := *ev
1035 as.Name = "push"
1036 probe := &workflow.Workflow{On: &workflow.On{Events: []*workflow.Event{&as}}}
1037 return probe.MatchPush(workflow.PushInfo{Ref: "refs/heads/" + base, ChangedFiles: files, FilesTruncated: truncated})
1038}
1039
1040// --- the github context ---
1041
1042// pullGithubJSON builds the github context of a pull_request run: github.ref is
1043// the merge ref, github.sha the merge commit (the head for a closed request
1044// that never merged), and the event is shaped like GitHub's pull_request
1045// webhook payload with the fields ermite can fill. A pull_request_target run
1046// has the same payload under its own event name, on the default branch.
1047func (c *CI) pullGithubJSON(repo domain.Repo, run domain.CIRun, f pushFacts) string {
1048 p := f.pull
1049 mr := p.mr
1050 u := c.urls(repo)
1051 id := c.identity(run.Actor)
1052 owner := c.identity("owner")
1053 repoObj := c.repoObj(repo, u, f.defaultBranch, run.CreatedAt)
1054 num := strconv.FormatInt(mr.Number, 10)
1055 prAPI := u.api + "/repos/" + u.full + "/pulls/" + num
1056 html := u.repo + "/merge-requests/" + num
1057 side := func(branch string, sha domain.ObjectID) map[string]any {
1058 return map[string]any{
1059 "label": u.owner + ":" + branch, "ref": branch, "sha": sha.String(),
1060 "user": u.userObj(owner, false), "repo": repoObj,
1061 }
1062 }
1063 tm := func(t time.Time) any {
1064 if t.IsZero() {
1065 return nil
1066 }
1067 return t.UTC().Format(time.RFC3339)
1068 }
1069 var body any
1070 if s := mr.Description.String(); s != "" {
1071 body = s
1072 }
1073 state, closedAt, mergedAt := "open", any(nil), any(nil)
1074 mergeSha := any(nil)
1075 if !mr.IsOpen() {
1076 state, closedAt = "closed", tm(mr.UpdatedAt)
1077 }
1078 switch {
1079 case p.merged:
1080 mergedAt = tm(mr.UpdatedAt)
1081 mergeSha = mr.MergedCommitID.String()
1082 case !p.merge.IsZero() && p.merge != p.head:
1083 mergeSha = p.merge.String() // the test merge commit
1084 }
1085 prID := stableID("pull", u.full+"#"+num)
1086 pr := map[string]any{
1087 "url": prAPI, "id": prID, "node_id": nodeID("011", "PullRequest", prID), "html_url": html,
1088 "diff_url": html + ".diff", "patch_url": html + ".patch", "issue_url": u.api + "/repos/" + u.full + "/issues/" + num,
1089 "number": mr.Number, "state": state, "locked": false, "title": mr.Title.String(), "user": u.userObj(c.identity(mr.Author), false),
1090 "body": body, "created_at": tm(mr.CreatedAt), "updated_at": tm(mr.UpdatedAt), "closed_at": closedAt, "merged_at": mergedAt,
1091 "merge_commit_sha": mergeSha, "assignee": nil, "assignees": []any{}, "requested_reviewers": []any{},
1092 "requested_teams": []any{}, "labels": []any{}, "milestone": nil, "draft": false,
1093 "head": side(mr.Source, p.head), "base": side(mr.Target, p.base), "author_association": "OWNER",
1094 "auto_merge": nil, "active_lock_reason": nil, "merged": p.merged,
1095 }
1096 event := map[string]any{
1097 "action": p.action, "number": mr.Number, "pull_request": pr, "repository": repoObj, "sender": u.userObj(id, false),
1098 }
1099 if p.action == domain.PullSynchronize {
1100 before := p.before.String()
1101 if p.before.IsZero() {
1102 before = domain.ZeroOID
1103 }
1104 event["before"], event["after"] = before, p.head.String()
1105 }
1106 refName := strings.TrimPrefix(run.Ref, domain.PullRefPrefix)
1107 if p.target {
1108 refName = strings.TrimPrefix(run.Ref, "refs/heads/")
1109 }
1110 return c.githubContext(repo, run, ghEvent{
1111 name: f.eventName(), refName: refName, refType: "branch",
1112 headRef: mr.Source, baseRef: mr.Target, payload: event,
1113 })
1114}