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 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 off, err := c.workflowDisabled(ctx, repo, wfPath)
685 if err != nil {
686 return ports.CIRunDraft{}, err
687 }
688 if off {
689 return ports.CIRunDraft{}, domain.ErrWorkflowDisabled
690 }
691
692 if f.pull == nil || f.pull.action == domain.PullClosed {
693 return c.cfg.Store.CreateRun(ctx, repo, wfPath, build)
694 }
695 var d ports.CIRunDraft
696 err = c.whileOpen(ctx, repo, f.pull.mr.Number, func() error {
697 var err error
698 d, err = c.cfg.Store.CreateRun(ctx, repo, wfPath, build)
699 return err
700 })
701 return d, err
702}
703
704// pullFacts is what a pull_request run is for.
705type pullFacts struct {
706 mr domain.MergeRequest
707 action string
708 // head and base are the commits of the event; the merge commit of the run
709 // has them as parents.
710 head, base domain.ObjectID
711 // before is the head before a synchronize.
712 before domain.ObjectID
713 merged bool
714 // merge is the test merge commit of head and base, zero when there is
715 // none (a conflict, or a closed request that never merged cleanly).
716 merge domain.ObjectID
717 // target is set for a pull_request_target run.
718 target bool
719}
720
721// pullState is the merge of a head into a base.
722type pullState struct {
723 head, base domain.ObjectID
724 // merge is the test merge commit of head and base when clean; for a closed
725 // request, the last one published, zero if none.
726 merge domain.ObjectID
727 clean bool
728 // otherHead is set for a closed request whose last merge commit is of
729 // another head than its event's: the provenance of the event's head says
730 // nothing of the code the run executes.
731 otherHead bool
732}
733
734func (c *CI) handlePullEvent(ctx context.Context, e pullEvent) error {
735 c.stage(pullStageEvent)
736 mr, err := c.cfg.MRs.Get(ctx, e.repo, e.number)
737 if errors.Is(err, domain.ErrNotFound) {
738 return nil
739 }
740 if err != nil {
741 return err
742 }
743 switch e.action {
744 case domain.PullOpened, domain.PullSynchronize:
745 if !mr.IsOpen() {
746 return nil // closed meanwhile: the closed event follows
747 }
748 st, err := c.mergeOf(ctx, mr, e.head, e.base)
749 if err != nil {
750 return err
751 }
752 c.stage(pullStageComputed)
753 if err := c.publishPull(ctx, mr, st); err != nil || !e.run {
754 return err
755 }
756 var err1 error
757 if st.clean { // a conflict makes no pull_request run, as on GitHub
758 err1 = c.startPullRuns(ctx, e, mr, st, false)
759 }
760 return errors.Join(err1, c.startPullRuns(ctx, e, mr, st, true))
761 case domain.PullClosed:
762 if mr.IsOpen() || !e.run {
763 return nil
764 }
765 st, ok, err := c.closedPull(ctx, e, mr)
766 if err != nil || !ok {
767 return err
768 }
769 pe := e
770 if st.otherHead || st.merge != e.merge {
771 pe.trusted = false // not the code the trust was checked for
772 }
773 return errors.Join(c.startPullRuns(ctx, pe, mr, st, false), c.startPullRuns(ctx, e, mr, st, true))
774 }
775 return nil
776}
777
778// refreshPull brings the refs of a merge request to its branches as they are.
779func (c *CI) refreshPull(ctx context.Context, k pullKey) error {
780 c.stage(pullStageRefresh)
781 mr, err := c.cfg.MRs.Get(ctx, k.repo, k.number)
782 if errors.Is(err, domain.ErrNotFound) || (err == nil && !mr.IsOpen()) {
783 return nil
784 }
785 if err != nil {
786 return err
787 }
788 head, err := c.cfg.Git.BranchHead(ctx, k.repo, mr.Source)
789 if errors.Is(err, domain.ErrNotFound) {
790 return nil
791 }
792 if err != nil {
793 return err
794 }
795 base, err := c.cfg.Git.BranchHead(ctx, k.repo, mr.Target)
796 if errors.Is(err, domain.ErrNotFound) {
797 return nil
798 }
799 if err != nil {
800 return err
801 }
802 st, err := c.mergeOf(ctx, mr, head, base)
803 if err != nil {
804 return err
805 }
806 return c.publishPull(ctx, mr, st)
807}
808
809// mergeOf is the test merge of head into base. It computes the commit only; a
810// merge ref that is already that merge is reused.
811func (c *CI) mergeOf(ctx context.Context, mr domain.MergeRequest, head, base domain.ObjectID) (pullState, error) {
812 st := pullState{head: head, base: base}
813 if head.IsZero() || base.IsZero() {
814 return st, nil
815 }
816 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err == nil {
817 cs, _, err := c.cfg.Git.ListRange(ctx, mr.Repo, domain.ObjectID{}, old, 1)
818 if err == nil && len(cs) == 1 && len(cs[0].Parents) == 2 && cs[0].Parents[0] == base && cs[0].Parents[1] == head {
819 st.merge, st.clean = old, true
820 return st, nil
821 }
822 } else if !errors.Is(err, domain.ErrNotFound) {
823 return st, err
824 }
825 tree, clean, err := c.cfg.Pull.TestMerge(ctx, mr.Repo, base, head)
826 k := pullKey{mr.Repo, mr.Number}
827 if err != nil {
828 return st, err
829 }
830 if !clean {
831 c.pull.attempts.conflicted(k, base, head) // a rescan does not try these tips again
832 return st, nil
833 }
834 c.pull.attempts.merged(k)
835 committer := c.cfg.Committer
836 if committer.Name == "" {
837 committer = domain.Identity{Name: "ermite", Email: "ermite@localhost"}
838 }
839 id, err := c.cfg.Git.WriteCommit(ctx, mr.Repo, domain.TestMergeCommit(tree, base, head, committer, c.now()))
840 if err != nil {
841 return st, err
842 }
843 st.merge, st.clean = id, true
844 return st, nil
845}
846
847// publishPull points refs/pull/<n>/head and /merge at st, under the gate. A ref
848// moves only to the latest: the head ref when st.head is still the head of the
849// source branch, the merge ref when the target tip is st.base too, so that an
850// older event never moves a ref back. errPullClosed when the request is closed.
851func (c *CI) publishPull(ctx context.Context, mr domain.MergeRequest, st pullState) error {
852 if st.head.IsZero() {
853 return nil
854 }
855 changed := false
856 err := c.whileOpen(ctx, mr.Repo, mr.Number, func() error {
857 curHead, err := c.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
858 if err != nil || curHead != st.head {
859 return ignoreNotFound(err)
860 }
861 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil || old != st.head {
862 if err != nil && !errors.Is(err, domain.ErrNotFound) {
863 return err
864 }
865 if err := c.cfg.Pull.SetPullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number), st.head); err != nil {
866 return err
867 }
868 changed = true
869 }
870 if !st.clean {
871 return nil
872 }
873 curBase, err := c.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
874 if err != nil || curBase != st.base {
875 return ignoreNotFound(err)
876 }
877 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err != nil || old != st.merge {
878 if err != nil && !errors.Is(err, domain.ErrNotFound) {
879 return err
880 }
881 if err := c.cfg.Pull.SetPullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number), st.merge); err != nil {
882 return err
883 }
884 changed = true
885 }
886 return nil
887 })
888 if changed && c.cfg.Fresh != nil {
889 c.cfg.Fresh.Refresh(ctx, mr.Repo)
890 }
891 return err
892}
893
894func ignoreNotFound(err error) error {
895 if errors.Is(err, domain.ErrNotFound) {
896 return nil
897 }
898 return err
899}
900
901// closedPull is the state a closed or merged merge request runs with: the last
902// test merge commit published, or the head when there is none. Nothing is
903// computed again. ok is false when the request has neither (its branch never
904// existed).
905func (c *CI) closedPull(ctx context.Context, e pullEvent, mr domain.MergeRequest) (st pullState, ok bool, err error) {
906 if st.merge, err = c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err != nil && !errors.Is(err, domain.ErrNotFound) {
907 return st, false, err
908 }
909 st.head = e.head
910 if st.head.IsZero() {
911 st.head = mr.HeadID
912 }
913 if st.head.IsZero() {
914 if st.head, err = c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil && !errors.Is(err, domain.ErrNotFound) {
915 return st, false, err
916 }
917 }
918 st.base = e.base
919 if !st.merge.IsZero() {
920 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 {
921 st.base = cs[0].Parents[0]
922 if st.head.IsZero() {
923 st.head = cs[0].Parents[1]
924 }
925 st.otherHead = cs[0].Parents[1] != e.head
926 }
927 }
928 if st.base.IsZero() {
929 st.base = mr.BaseID
930 }
931 st.clean = true
932 return st, !st.head.IsZero(), nil
933}
934
935// startPullRuns starts the pull_request workflows of the merge request that
936// match the activity and the files it changes, or with target its
937// pull_request_target workflows: those of the default tip of the event.
938func (c *CI) startPullRuns(ctx context.Context, e pullEvent, mr domain.MergeRequest, st pullState, target bool) error {
939 if target && (e.target.commit.IsZero() || st.head.IsZero()) {
940 return nil // no default branch, or no head to run for
941 }
942 runners, err := c.cfg.Store.ListRunners(ctx)
943 if err != nil {
944 return fmt.Errorf("list the runners: %w", err)
945 }
946 if len(runners) == 0 {
947 return nil
948 }
949 repo, err := c.cfg.Meta.Get(ctx, e.repo)
950 if err != nil {
951 return nil // deleted meanwhile
952 }
953 t := e.trigger(c.now())
954 ref, sha := domain.PullMergeRef(mr.Number), st.merge
955 switch {
956 case target:
957 // The code is the default tip's: the cause and that tip decide.
958 ref, sha = "refs/heads/"+e.target.branch, e.target.commit
959 t.trusted = e.actorTrusted && e.target.trusted
960 if e.actorTrusted && !e.target.trusted {
961 ctxLog(ctx).Info().Str("repo", e.repo.String()).Int64("number", mr.Number).Str("action", e.action).Str("tip", sha.String()).
962 Msg("ci pull_request_target runs are not trusted: the default branch tip was not pushed by a trusted principal")
963 }
964 case sha.IsZero():
965 ref, sha = domain.PullHeadRef(mr.Number), st.head
966 }
967 pf := &pullFacts{
968 mr: mr, action: e.action, head: st.head, base: st.base, before: e.before, merged: mr.State == domain.MRStateMerged,
969 merge: st.merge, target: target,
970 }
971 f := pushFacts{
972 update: domain.RefUpdate{Ref: ref, Old: domain.ZeroOID, New: sha.String()}, commit: sha, pull: pf,
973 }
974 if def, _, err := c.cfg.Repos.DefaultBranch(ctx, e.repo); err == nil {
975 f.defaultBranch = def
976 }
977 head, _, err := c.cfg.Git.ListRange(ctx, e.repo, domain.ObjectID{}, st.head, 1)
978 if err != nil || len(head) == 0 {
979 return fmt.Errorf("read the head commit of !%d: %w", mr.Number, err)
980 }
981 f.head = head[0]
982 // The files of the merge request are those between the merge base of the
983 // target and the head, and the head: what the review shows.
984 f.filesTruncated = true // unavailable unless read below: a paths filter then passes
985 if !st.base.IsZero() {
986 if mb, err := c.cfg.Git.MergeBase(ctx, e.repo, st.base, st.head); err == nil {
987 if paths, truncated, err := c.cfg.Diff.ChangedPaths(ctx, e.repo, mb, st.head, workflow.MaxFilesEvaluated); err == nil {
988 f.files, f.filesTruncated = paths, truncated
989 }
990 }
991 }
992 files, err := c.workflowFiles(ctx, e.repo, sha)
993 if err != nil || len(files) == 0 {
994 return err
995 }
996 if !target && workflow.SkipCI(f.head.Message) {
997 return nil // GitHub's skip directives are for push and pull_request only
998 }
999 var errs []error
1000 for _, file := range files {
1001 if file.problem != "" {
1002 continue // reported by the push of the branch
1003 }
1004 wf, perrs := parseWorkflow(file.path, file.content)
1005 if len(perrs) > 0 || !matchPullRequest(wf, f.eventName(), e.action, mr.Target, f.files, f.filesTruncated) {
1006 continue
1007 }
1008 if problem := c.matrixProblem(ctx, repo, t, f, file.path, wf); problem.text != "" {
1009 name := file.path
1010 if problem.named {
1011 name = workflowName(wf, file.path)
1012 }
1013 errs = append(errs, c.startInvalid(ctx, repo, t, f, file.path, name, problem.text))
1014 continue
1015 }
1016 errs = append(errs, c.startRun(ctx, repo, t, f, file.path, string(file.content), wf))
1017 }
1018 return errors.Join(errs...)
1019}
1020
1021// matchPullRequest reports whether the workflow runs for the activity of a merge
1022// request into base: the event (pull_request or pull_request_target) is
1023// listed, the activity is one of its types (opened, synchronize and reopened
1024// when it lists none), and the branches and paths filters, which apply to the
1025// base branch and the changed files, let it through. The filters are those of
1026// the push event (ermite-actions), run on the event under another name.
1027func matchPullRequest(wf *workflow.Workflow, event, action, base string, files []string, truncated bool) bool {
1028 if wf == nil || wf.On == nil {
1029 return false
1030 }
1031 ev := wf.On.Event(event)
1032 if ev == nil {
1033 return false
1034 }
1035 types := ev.Types
1036 if types == nil {
1037 types = domain.DefaultPullRequestTypes
1038 }
1039 if !slices.Contains(types, action) {
1040 return false
1041 }
1042 as := *ev
1043 as.Name = "push"
1044 probe := &workflow.Workflow{On: &workflow.On{Events: []*workflow.Event{&as}}}
1045 return probe.MatchPush(workflow.PushInfo{Ref: "refs/heads/" + base, ChangedFiles: files, FilesTruncated: truncated})
1046}
1047
1048// --- the github context ---
1049
1050// pullGithubJSON builds the github context of a pull_request run: github.ref is
1051// the merge ref, github.sha the merge commit (the head for a closed request
1052// that never merged), and the event is shaped like GitHub's pull_request
1053// webhook payload with the fields ermite can fill. A pull_request_target run
1054// has the same payload under its own event name, on the default branch.
1055func (c *CI) pullGithubJSON(repo domain.Repo, run domain.CIRun, f pushFacts) string {
1056 p := f.pull
1057 mr := p.mr
1058 u := c.urls(repo)
1059 id := c.identity(run.Actor)
1060 owner := c.identity("owner")
1061 repoObj := c.repoObj(repo, u, f.defaultBranch, run.CreatedAt)
1062 num := strconv.FormatInt(mr.Number, 10)
1063 prAPI := u.api + "/repos/" + u.full + "/pulls/" + num
1064 html := u.repo + "/merge-requests/" + num
1065 side := func(branch string, sha domain.ObjectID) map[string]any {
1066 return map[string]any{
1067 "label": u.owner + ":" + branch, "ref": branch, "sha": sha.String(),
1068 "user": u.userObj(owner, false), "repo": repoObj,
1069 }
1070 }
1071 tm := func(t time.Time) any {
1072 if t.IsZero() {
1073 return nil
1074 }
1075 return t.UTC().Format(time.RFC3339)
1076 }
1077 var body any
1078 if s := mr.Description.String(); s != "" {
1079 body = s
1080 }
1081 state, closedAt, mergedAt := "open", any(nil), any(nil)
1082 mergeSha := any(nil)
1083 if !mr.IsOpen() {
1084 state, closedAt = "closed", tm(mr.UpdatedAt)
1085 }
1086 switch {
1087 case p.merged:
1088 mergedAt = tm(mr.UpdatedAt)
1089 mergeSha = mr.MergedCommitID.String()
1090 case !p.merge.IsZero() && p.merge != p.head:
1091 mergeSha = p.merge.String() // the test merge commit
1092 }
1093 prID := stableID("pull", u.full+"#"+num)
1094 pr := map[string]any{
1095 "url": prAPI, "id": prID, "node_id": nodeID("011", "PullRequest", prID), "html_url": html,
1096 "diff_url": html + ".diff", "patch_url": html + ".patch", "issue_url": u.api + "/repos/" + u.full + "/issues/" + num,
1097 "number": mr.Number, "state": state, "locked": false, "title": mr.Title.String(), "user": u.userObj(c.identity(mr.Author), false),
1098 "body": body, "created_at": tm(mr.CreatedAt), "updated_at": tm(mr.UpdatedAt), "closed_at": closedAt, "merged_at": mergedAt,
1099 "merge_commit_sha": mergeSha, "assignee": nil, "assignees": []any{}, "requested_reviewers": []any{},
1100 "requested_teams": []any{}, "labels": []any{}, "milestone": nil, "draft": false,
1101 "head": side(mr.Source, p.head), "base": side(mr.Target, p.base), "author_association": "OWNER",
1102 "auto_merge": nil, "active_lock_reason": nil, "merged": p.merged,
1103 }
1104 event := map[string]any{
1105 "action": p.action, "number": mr.Number, "pull_request": pr, "repository": repoObj, "sender": u.userObj(id, false),
1106 }
1107 if p.action == domain.PullSynchronize {
1108 before := p.before.String()
1109 if p.before.IsZero() {
1110 before = domain.ZeroOID
1111 }
1112 event["before"], event["after"] = before, p.head.String()
1113 }
1114 refName := strings.TrimPrefix(run.Ref, domain.PullRefPrefix)
1115 if p.target {
1116 refName = strings.TrimPrefix(run.Ref, "refs/heads/")
1117 }
1118 return c.githubContext(repo, run, ghEvent{
1119 name: f.eventName(), refName: refName, refType: "branch",
1120 headRef: mr.Source, baseRef: mr.Target, payload: event,
1121 })
1122}