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). The merge commit of a
40// run is computed from the head and base of its event, never from the branches
41// as they are now.
42// - the refresh worker keeps refs/pull/<n>/{head,merge} at the latest branches
43// when a source or a target branch moves; the refreshes of one merge request
44// coalesce, and the work is bounded.
45//
46// Whatever publishes a ref or creates a run does it through the PullGate, under
47// the merge lock of the repository and only while the request is open, so a close
48// or a merge that completes meanwhile discards it: a closed request gets no
49// refs and no run but its "closed" one, which uses the last merge commit that was
50// published, or the head, and is never computed again.
51//
52// A head that does not merge cleanly makes no run for that update, as on GitHub;
53// the merge ref then keeps the last commit that merged, as GitHub's does.
54
55const (
56 // DefaultCIPullRefreshes bounds the merge requests (and the pushes) waiting
57 // for the refresh worker. Past it the repository is marked dirty instead and
58 // the worker rescans its open merge requests when it has capacity, so that
59 // nothing stays stale for good.
60 DefaultCIPullRefreshes = 4096
61 // DefaultCIPullSnapshot bounds the merge requests one push takes the snapshot
62 // of (the open requests whose source branch it pushed). Past it the rest is
63 // skipped, an error is logged, and the repository is rescanned for the refs.
64 DefaultCIPullSnapshot = 256
65 // Stages of the work of the merge requests, for the test hook
66 // CIConfig.PullStage.
67 pullStageDiscover = "discover"
68 pullStageRescan = "rescan"
69 pullStageEvent = "event"
70 pullStageComputed = "computed"
71 pullStageRefresh = "refresh"
72)
73
74// errPullClosed means a merge request was closed or merged before the work for
75// it could be published.
76var errPullClosed = errors.New("merge request is not open")
77
78// pullEvent is one activity of a merge request waiting to be handled.
79type pullEvent struct {
80 repo domain.RepoName
81 number int64
82 action string
83 // head and base are the commits the activity is for; before is the head
84 // before a synchronize, zero otherwise.
85 head, base, before domain.ObjectID
86 actor string
87 pusher domain.CIPusher
88 trusted bool
89 // run is false when the cause is a job token: no run starts.
90 run bool
91 log zerowrap.Logger
92}
93
94func (e pullEvent) trigger(at time.Time) ciTrigger {
95 return ciTrigger{repo: e.repo, actor: e.actor, pusher: e.pusher, trusted: e.trusted, at: at, log: e.log}
96}
97
98type pullKey struct {
99 repo domain.RepoName
100 number int64
101}
102
103// pullQueue is what the two workers read.
104//
105// Overload policy. The activities (events) are never coalesced and never
106// dropped by the cap of the refreshes: they go on their own queue, which behaves
107// as the queue of the push triggers (CI.queue) does when it is full: the item is
108// dropped and an error is logged, so a push is never slowed or failed by CI. Only
109// the refresh of refs/pull/<n>/{head,merge}, which is idempotent and recomputed
110// from the branches as they are, is coalesced (one waiting refresh per merge
111// request) and bounded (max); past the bound the repository is marked dirty and
112// its open merge requests are rescanned when the worker has capacity.
113type pullQueue struct {
114 events chan pullEvent
115
116 mu sync.Mutex
117 max int // bound of the waiting refreshes
118 pending map[pullKey]bool
119 order []pullKey
120 // dirty are the repositories whose merge requests may have stale refs: the
121 // bound is the number of repositories.
122 dirty map[domain.RepoName]bool
123 dirtyOrder []domain.RepoName
124 // attempts remember the last state tried for each merge request, so that a
125 // rescan does not try again what it already tried (a conflict).
126 attempts *pullAttempts
127 wake chan struct{}
128 gate ports.PullGate // set by SetPullGate
129}
130
131func newPullQueue(size, max, attempts int) *pullQueue {
132 return &pullQueue{
133 events: make(chan pullEvent, size), max: max, pending: map[pullKey]bool{},
134 dirty: map[domain.RepoName]bool{}, attempts: newPullAttempts(attempts), wake: make(chan struct{}, 1),
135 }
136}
137
138// SetPullGate gives the CI the gate of the merge requests (see PullGate). Call
139// it before merge requests can be opened.
140func (c *CI) SetPullGate(g ports.PullGate) {
141 c.pull.mu.Lock()
142 c.pull.gate = g
143 c.pull.mu.Unlock()
144}
145
146func (c *CI) pullEnabled() bool { return c.cfg.MRs != nil && c.cfg.Pull != nil }
147
148func (c *CI) stage(stage string) {
149 if c.cfg.PullStage != nil {
150 c.cfg.PullStage(stage)
151 }
152}
153
154// OnPullRequest queues an activity of a merge request.
155func (c *CI) OnPullRequest(ctx context.Context, repo domain.RepoName, ev ports.PullRequestEvent, p domain.Principal) {
156 if !c.pullEnabled() {
157 return
158 }
159 c.queuePullEvent(pullEvent{
160 repo: repo, number: ev.Number, action: ev.Action, head: ev.Head, base: ev.Base,
161 actor: authorOf(p), pusher: domain.PusherOf(p), trusted: domain.TrustedActor(p), run: !p.ViaJobToken(),
162 log: zerowrap.FromCtx(ctx),
163 })
164}
165
166// onBranchesMoved is called for every push and merge, after its push
167// workflows are queued (so a push run never waits for this). It takes the
168// snapshot of the push for the merge requests it touches, here and now: the open
169// merge requests are listed (a scan), and for those whose source branch was
170// pushed a synchronize event is queued carrying the pushed head, the target tip
171// as it is at this moment (read once per target for the push), and the actor and
172// trust of the push. The event runs for exactly that snapshot, however the
173// branches move before it is handled: a later push to the target, possibly by
174// someone less trusted, changes nothing about what a trusted run executes.
175// The refresh of the refs of every merge request touched (those that have a
176// pushed branch as source or as target, from the two indexes) is queued for the
177// refresh worker, which is the only fully asynchronous part. A push that touches
178// no merge request queues nothing and marks nothing; the whole-repository rescan
179// is for overflow, truncation and startup.
180func (c *CI) onBranchesMoved(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, actor string, pusher domain.CIPusher, trusted, run bool) {
181 if !c.pullEnabled() {
182 return
183 }
184 moved := map[string]domain.RefUpdate{}
185 for _, u := range updates {
186 if b, ok := strings.CutPrefix(u.Ref, "refs/heads/"); ok && !u.IsDelete() {
187 moved[b] = u
188 }
189 }
190 if len(moved) == 0 {
191 return
192 }
193 c.stage(pullStageDiscover)
194 log := zerowrap.FromCtx(ctx)
195 branches := make([]string, 0, len(moved))
196 for b := range moved {
197 branches = append(branches, b)
198 }
199
200 // The requests whose branch is the target of a push (a target move): their
201 // refs are queued for a refresh. Only these: an unrelated push queues nothing
202 // and marks nothing. The lookup is bounded like the snapshot; past the bound
203 // or on an error the repository is marked dirty, and its rescan finds the
204 // stale ones.
205 targets, tgtTruncated, err := c.cfg.MRs.OpenByTarget(ctx, repo, branches, c.cfg.PullSnapshot)
206 switch {
207 case err != nil:
208 log.Error().Err(err).Str("repo", repo.String()).Msg("ci: look up the merge requests targeting a pushed branch: the repository is rescanned")
209 c.markPullDirty(repo)
210 case tgtTruncated:
211 log.Error().Str("repo", repo.String()).Int("bound", c.cfg.PullSnapshot).
212 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")
213 c.markPullDirty(repo)
214 }
215 for _, n := range targets {
216 c.queuePullRefresh(pullKey{repo, n})
217 }
218
219 // The requests whose source branch was pushed come from the index of the open
220 // merge requests by source: a scan of the keys of those branches, bounded and
221 // cancellable, never of every merge request the repository ever had. Past the
222 // bound or on cancellation the rest is skipped and logged (the overload policy
223 // of the event queue: those events are lost), and the repository is marked
224 // dirty, so that the refs of the skipped requests are repaired by the rescan.
225 numbers, truncated, err := c.cfg.MRs.OpenBySource(ctx, repo, branches, c.cfg.PullSnapshot)
226 if err != nil {
227 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")
228 c.markPullDirty(repo)
229 if len(numbers) == 0 {
230 return
231 }
232 } else if truncated {
233 log.Error().Str("repo", repo.String()).Int("bound", c.cfg.PullSnapshot).
234 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")
235 c.markPullDirty(repo)
236 }
237 tips := map[string]domain.ObjectID{}
238 for _, n := range numbers {
239 mr, err := c.cfg.MRs.Get(ctx, repo, n)
240 if err != nil || !mr.IsOpen() {
241 continue // closed or gone since the index was read
242 }
243 u, srcMoved := moved[mr.Source]
244 if !srcMoved {
245 continue
246 }
247 if run {
248 if head, herr := domain.ParseObjectID(u.New); herr == nil {
249 base, seen := tips[mr.Target]
250 if !seen {
251 base, _ = c.cfg.Git.BranchHead(ctx, repo, mr.Target)
252 tips[mr.Target] = base
253 }
254 before, _ := domain.ParseObjectID(u.Old) // zero for a new branch
255 c.queuePullEvent(pullEvent{
256 repo: repo, number: mr.Number, action: domain.PullSynchronize, head: head, base: base, before: before,
257 actor: actor, pusher: pusher, trusted: trusted, run: true, log: log,
258 })
259 }
260 }
261 c.queuePullRefresh(pullKey{repo, mr.Number})
262 }
263}
264
265// markPullDirty marks a repository for a rescan of its merge requests.
266func (c *CI) markPullDirty(repo domain.RepoName) {
267 c.pull.mu.Lock()
268 c.pull.markDirty(repo)
269 c.pull.mu.Unlock()
270 c.wakePull()
271}
272
273// markDirty records that the merge requests of a repository may be stale. The
274// caller holds mu.
275func (q *pullQueue) markDirty(repo domain.RepoName) {
276 if !q.dirty[repo] {
277 q.dirty[repo] = true
278 q.dirtyOrder = append(q.dirtyOrder, repo)
279 }
280}
281
282func (c *CI) wakePull() {
283 select {
284 case c.pull.wake <- struct{}{}:
285 default:
286 }
287}
288
289// queuePullEvent queues an activity. A full queue drops it and logs an error,
290// exactly as the queue of the push triggers does (CI.enqueue): CI never slows
291// or fails a push. The refresh cap does not apply to events.
292func (c *CI) queuePullEvent(e pullEvent) {
293 c.mu.Lock()
294 closed := c.closed
295 c.mu.Unlock()
296 if closed {
297 return
298 }
299 select {
300 case c.pull.events <- e:
301 default:
302 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")
303 }
304}
305
306// queuePullRefresh asks for the refs of a merge request to be brought up to
307// date; asking again while it waits changes nothing. When the queue is full the
308// repository is marked dirty instead: the worker rescans its open merge requests
309// once it has capacity.
310func (c *CI) queuePullRefresh(k pullKey) {
311 q := c.pull
312 q.mu.Lock()
313 switch {
314 case q.pending[k]:
315 case len(q.pending) < q.max:
316 q.pending[k] = true
317 q.order = append(q.order, k)
318 default:
319 q.markDirty(k.repo)
320 }
321 q.mu.Unlock()
322 c.wakePull()
323}
324
325// nextPullWork takes what the refresh worker does next: a refresh, or a dirty
326// repository to rescan, in that order.
327func (c *CI) nextPullWork() (key *pullKey, dirty *domain.RepoName) {
328 q := c.pull
329 q.mu.Lock()
330 defer q.mu.Unlock()
331 switch {
332 case len(q.order) > 0:
333 k := q.order[0]
334 q.order = q.order[1:]
335 delete(q.pending, k)
336 return &k, nil
337 case len(q.dirtyOrder) > 0:
338 r := q.dirtyOrder[0]
339 q.dirtyOrder = q.dirtyOrder[1:]
340 delete(q.dirty, r)
341 return nil, &r
342 }
343 return nil, nil
344}
345
346// RescanPulls rescans the merge requests of a repository now, on the calling
347// goroutine, as the refresh worker does for a repository it marked dirty. It is
348// for tests and operators' tooling; the server never needs to call it.
349func (c *CI) RescanPulls(ctx context.Context, repo domain.RepoName) error {
350 return c.rescanPulls(ctx, repo)
351}
352
353// rescanPulls checks the open merge requests of a repository against their refs
354// and refreshes, one after the other and here (not through the bounded queue, so
355// a rescan never marks its repository dirty again), each whose
356// refs/pull/<n>/head is not the tip of its source branch or whose merge ref is
357// not a merge of the two current tips. A merge request whose current state was
358// already tried (a head that conflicts has no merge ref, so it never looks
359// current) is skipped until one of its branches moves. It repairs what a
360// saturated queue, a failed discovery or a restart left behind.
361func (c *CI) rescanPulls(ctx context.Context, repo domain.RepoName) error {
362 c.stage(pullStageRescan)
363 mrs, err := c.cfg.MRs.ListOpen(ctx, repo)
364 if err != nil {
365 return fmt.Errorf("list the open merge requests: %w", err)
366 }
367 for _, mr := range mrs {
368 if ctx.Err() != nil {
369 return nil
370 }
371 head, herr := c.cfg.Git.BranchHead(ctx, repo, mr.Source)
372 base, berr := c.cfg.Git.BranchHead(ctx, repo, mr.Target)
373 if herr != nil || berr != nil {
374 continue // a branch is gone: nothing to publish
375 }
376 if c.pull.attempts.tried(pullKey{repo, mr.Number}, base, head) || c.pullRefsCurrent(ctx, mr, head, base) {
377 continue
378 }
379 if err := c.refreshPull(ctx, pullKey{repo, mr.Number}); err != nil && !errors.Is(err, errPullClosed) {
380 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")
381 }
382 }
383 return nil
384}
385
386// pullRefsCurrent reports whether the head ref is head and the merge ref, if
387// there is one, merges base and head. A head that conflicts has no merge ref and
388// counts as current once the head ref is right: the next push or move tries again.
389func (c *CI) pullRefsCurrent(ctx context.Context, mr domain.MergeRequest, head, base domain.ObjectID) bool {
390 if cur, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil || cur != head {
391 return false
392 }
393 merge, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number))
394 if err != nil {
395 return false // none: a refresh decides whether the merge is clean
396 }
397 cs, _, err := c.cfg.Git.ListRange(ctx, mr.Repo, domain.ObjectID{}, merge, 1)
398 return err == nil && len(cs) == 1 && len(cs[0].Parents) == 2 && cs[0].Parents[0] == base && cs[0].Parents[1] == head
399}
400
401// rescanAllPulls marks dirty every repository that has open merge requests, at
402// startup: the refs may be stale after a restart or a crash.
403func (c *CI) rescanAllPulls(ctx context.Context) {
404 repos, err := c.cfg.Meta.List(ctx)
405 if err != nil {
406 ctxLog(ctx).Error().Err(err).Msg("ci: list the repositories to rescan their merge requests")
407 return
408 }
409 q := c.pull
410 q.mu.Lock()
411 for _, r := range repos {
412 q.markDirty(r.Name)
413 }
414 q.mu.Unlock()
415 c.wakePull()
416}
417
418// pullLoop is the event worker.
419func (c *CI) pullLoop() {
420 defer c.wg.Done()
421 for {
422 select {
423 case <-c.ctx.Done():
424 return
425 case e := <-c.pull.events:
426 c.guardPull(e.log, e.repo, func(ctx context.Context) error { return c.handlePullEvent(ctx, e) })
427 }
428 }
429}
430
431// refreshLoop is the refresh worker: it refreshes the refs of the merge
432// requests a push touched, and rescans the repositories that fell behind.
433func (c *CI) refreshLoop() {
434 defer c.wg.Done()
435 c.rescanAllPulls(c.ctx)
436 for {
437 key, dirty := c.nextPullWork()
438 if c.ctx.Err() != nil {
439 return
440 }
441 switch {
442 case key != nil:
443 c.guardPull(zerowrap.FromCtx(c.ctx), key.repo, func(ctx context.Context) error { return c.refreshPull(ctx, *key) })
444 case dirty != nil:
445 c.guardPull(zerowrap.FromCtx(c.ctx), *dirty, func(ctx context.Context) error { return c.rescanPulls(ctx, *dirty) })
446 default:
447 select {
448 case <-c.ctx.Done():
449 return
450 case <-c.pull.wake:
451 }
452 }
453 }
454}
455
456// guardPull runs fn; a panic or an error is logged and never escapes.
457func (c *CI) guardPull(log zerowrap.Logger, repo domain.RepoName, fn func(context.Context) error) {
458 ctx := zerowrap.WithCtx(c.ctx, log)
459 defer func() {
460 if r := recover(); r != nil {
461 log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Str("repo", repo.String()).Msg("ci pull request work panicked")
462 }
463 }()
464 if err := fn(ctx); err != nil && ctx.Err() == nil && !errors.Is(err, errPullClosed) {
465 log.Error().Err(err).Str("repo", repo.String()).Msg("ci pull request work failed")
466 }
467}
468
469// whileOpen runs fn under the merge lock while the request is open; errPullClosed
470// when it is not. Without a gate (tests of the CI alone) it checks and runs.
471func (c *CI) whileOpen(ctx context.Context, repo domain.RepoName, number int64, fn func() error) error {
472 c.pull.mu.Lock()
473 gate := c.pull.gate
474 c.pull.mu.Unlock()
475 var err error
476 if gate != nil {
477 err = gate.WhileOpen(ctx, repo, number, fn)
478 } else if mr, gerr := c.cfg.MRs.Get(ctx, repo, number); gerr != nil {
479 err = gerr
480 } else if !mr.IsOpen() {
481 err = domain.ErrPreconditionFailed
482 } else {
483 err = fn()
484 }
485 if errors.Is(err, domain.ErrPreconditionFailed) || errors.Is(err, domain.ErrNotFound) {
486 return errPullClosed
487 }
488 return err
489}
490
491// createRun stores a run. The run of an open merge request is stored under the
492// gate, so that one that is closed meanwhile gets none (errPullClosed).
493func (c *CI) createRun(ctx context.Context, f pushFacts, repo domain.RepoName, wfPath string, build func(id, number int64) ports.CIRunDraft) (ports.CIRunDraft, error) {
494 if f.pull == nil || f.pull.action == domain.PullClosed {
495 return c.cfg.Store.CreateRun(ctx, repo, wfPath, build)
496 }
497 var d ports.CIRunDraft
498 err := c.whileOpen(ctx, repo, f.pull.mr.Number, func() error {
499 var err error
500 d, err = c.cfg.Store.CreateRun(ctx, repo, wfPath, build)
501 return err
502 })
503 return d, err
504}
505
506// pullFacts is what a pull_request run is for.
507type pullFacts struct {
508 mr domain.MergeRequest
509 action string
510 // head and base are the commits of the event; the merge commit of the run
511 // has them as parents.
512 head, base domain.ObjectID
513 // before is the head before a synchronize.
514 before domain.ObjectID
515 merged bool
516}
517
518// pullState is the merge of a head into a base.
519type pullState struct {
520 head, base domain.ObjectID
521 // merge is the test merge commit of head and base when clean; for a closed
522 // request, the last one published, zero if none.
523 merge domain.ObjectID
524 clean bool
525}
526
527func (c *CI) handlePullEvent(ctx context.Context, e pullEvent) error {
528 c.stage(pullStageEvent)
529 mr, err := c.cfg.MRs.Get(ctx, e.repo, e.number)
530 if errors.Is(err, domain.ErrNotFound) {
531 return nil
532 }
533 if err != nil {
534 return err
535 }
536 switch e.action {
537 case domain.PullOpened, domain.PullSynchronize:
538 if !mr.IsOpen() {
539 return nil // closed meanwhile: the closed event follows
540 }
541 st, err := c.mergeOf(ctx, mr, e.head, e.base)
542 if err != nil {
543 return err
544 }
545 c.stage(pullStageComputed)
546 if err := c.publishPull(ctx, mr, st); err != nil || !st.clean || !e.run {
547 return err
548 }
549 return c.startPullRuns(ctx, e, mr, st)
550 case domain.PullClosed:
551 if mr.IsOpen() || !e.run {
552 return nil
553 }
554 st, ok, err := c.closedPull(ctx, e, mr)
555 if err != nil || !ok {
556 return err
557 }
558 return c.startPullRuns(ctx, e, mr, st)
559 }
560 return nil
561}
562
563// refreshPull brings the refs of a merge request to its branches as they are.
564func (c *CI) refreshPull(ctx context.Context, k pullKey) error {
565 c.stage(pullStageRefresh)
566 mr, err := c.cfg.MRs.Get(ctx, k.repo, k.number)
567 if errors.Is(err, domain.ErrNotFound) || (err == nil && !mr.IsOpen()) {
568 return nil
569 }
570 if err != nil {
571 return err
572 }
573 head, err := c.cfg.Git.BranchHead(ctx, k.repo, mr.Source)
574 if errors.Is(err, domain.ErrNotFound) {
575 return nil
576 }
577 if err != nil {
578 return err
579 }
580 base, err := c.cfg.Git.BranchHead(ctx, k.repo, mr.Target)
581 if errors.Is(err, domain.ErrNotFound) {
582 return nil
583 }
584 if err != nil {
585 return err
586 }
587 st, err := c.mergeOf(ctx, mr, head, base)
588 if err != nil {
589 return err
590 }
591 return c.publishPull(ctx, mr, st)
592}
593
594// mergeOf is the test merge of head into base. It computes the commit only; a
595// merge ref that is already that merge is reused.
596func (c *CI) mergeOf(ctx context.Context, mr domain.MergeRequest, head, base domain.ObjectID) (pullState, error) {
597 st := pullState{head: head, base: base}
598 if head.IsZero() || base.IsZero() {
599 return st, nil
600 }
601 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err == nil {
602 cs, _, err := c.cfg.Git.ListRange(ctx, mr.Repo, domain.ObjectID{}, old, 1)
603 if err == nil && len(cs) == 1 && len(cs[0].Parents) == 2 && cs[0].Parents[0] == base && cs[0].Parents[1] == head {
604 st.merge, st.clean = old, true
605 return st, nil
606 }
607 } else if !errors.Is(err, domain.ErrNotFound) {
608 return st, err
609 }
610 tree, clean, err := c.cfg.Pull.TestMerge(ctx, mr.Repo, base, head)
611 k := pullKey{mr.Repo, mr.Number}
612 if err != nil {
613 return st, err
614 }
615 if !clean {
616 c.pull.attempts.conflicted(k, base, head) // a rescan does not try these tips again
617 return st, nil
618 }
619 c.pull.attempts.merged(k)
620 committer := c.cfg.Committer
621 if committer.Name == "" {
622 committer = domain.Identity{Name: "ermite", Email: "ermite@localhost"}
623 }
624 id, err := c.cfg.Git.WriteCommit(ctx, mr.Repo, domain.TestMergeCommit(tree, base, head, committer, c.now()))
625 if err != nil {
626 return st, err
627 }
628 st.merge, st.clean = id, true
629 return st, nil
630}
631
632// publishPull points refs/pull/<n>/head and /merge at st, under the gate. A ref
633// moves only to the latest: the head ref when st.head is still the head of the
634// source branch, the merge ref when the target tip is st.base too, so that an
635// older event never moves a ref back. errPullClosed when the request is closed.
636func (c *CI) publishPull(ctx context.Context, mr domain.MergeRequest, st pullState) error {
637 if st.head.IsZero() {
638 return nil
639 }
640 changed := false
641 err := c.whileOpen(ctx, mr.Repo, mr.Number, func() error {
642 curHead, err := c.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
643 if err != nil || curHead != st.head {
644 return ignoreNotFound(err)
645 }
646 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil || old != st.head {
647 if err != nil && !errors.Is(err, domain.ErrNotFound) {
648 return err
649 }
650 if err := c.cfg.Pull.SetPullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number), st.head); err != nil {
651 return err
652 }
653 changed = true
654 }
655 if !st.clean {
656 return nil
657 }
658 curBase, err := c.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
659 if err != nil || curBase != st.base {
660 return ignoreNotFound(err)
661 }
662 if old, err := c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err != nil || old != st.merge {
663 if err != nil && !errors.Is(err, domain.ErrNotFound) {
664 return err
665 }
666 if err := c.cfg.Pull.SetPullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number), st.merge); err != nil {
667 return err
668 }
669 changed = true
670 }
671 return nil
672 })
673 if changed && c.cfg.Fresh != nil {
674 c.cfg.Fresh.Refresh(ctx, mr.Repo)
675 }
676 return err
677}
678
679func ignoreNotFound(err error) error {
680 if errors.Is(err, domain.ErrNotFound) {
681 return nil
682 }
683 return err
684}
685
686// closedPull is the state a closed or merged merge request runs with: the last
687// test merge commit published, or the head when there is none. Nothing is
688// computed again. ok is false when the request has neither (its branch never
689// existed).
690func (c *CI) closedPull(ctx context.Context, e pullEvent, mr domain.MergeRequest) (st pullState, ok bool, err error) {
691 if st.merge, err = c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullMergeRef(mr.Number)); err != nil && !errors.Is(err, domain.ErrNotFound) {
692 return st, false, err
693 }
694 st.head = e.head
695 if st.head.IsZero() {
696 st.head = mr.HeadID
697 }
698 if st.head.IsZero() {
699 if st.head, err = c.cfg.Pull.PullRef(ctx, mr.Repo, domain.PullHeadRef(mr.Number)); err != nil && !errors.Is(err, domain.ErrNotFound) {
700 return st, false, err
701 }
702 }
703 st.base = e.base
704 if !st.merge.IsZero() {
705 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 {
706 st.base = cs[0].Parents[0]
707 if st.head.IsZero() {
708 st.head = cs[0].Parents[1]
709 }
710 }
711 }
712 if st.base.IsZero() {
713 st.base = mr.BaseID
714 }
715 st.clean = true
716 return st, !st.head.IsZero(), nil
717}
718
719// startPullRuns starts the pull_request workflows of the merge request that
720// match the activity and the files it changes.
721func (c *CI) startPullRuns(ctx context.Context, e pullEvent, mr domain.MergeRequest, st pullState) error {
722 runners, err := c.cfg.Store.ListRunners(ctx)
723 if err != nil {
724 return fmt.Errorf("list the runners: %w", err)
725 }
726 if len(runners) == 0 {
727 return nil
728 }
729 repo, err := c.cfg.Meta.Get(ctx, e.repo)
730 if err != nil {
731 return nil // deleted meanwhile
732 }
733 t := e.trigger(c.now())
734 ref, sha := domain.PullMergeRef(mr.Number), st.merge
735 if sha.IsZero() {
736 ref, sha = domain.PullHeadRef(mr.Number), st.head
737 }
738 pf := &pullFacts{mr: mr, action: e.action, head: st.head, base: st.base, before: e.before, merged: mr.State == domain.MRStateMerged}
739 f := pushFacts{
740 update: domain.RefUpdate{Ref: ref, Old: domain.ZeroOID, New: sha.String()}, commit: sha, pull: pf,
741 }
742 if def, _, err := c.cfg.Repos.DefaultBranch(ctx, e.repo); err == nil {
743 f.defaultBranch = def
744 }
745 head, _, err := c.cfg.Git.ListRange(ctx, e.repo, domain.ObjectID{}, st.head, 1)
746 if err != nil || len(head) == 0 {
747 return fmt.Errorf("read the head commit of !%d: %w", mr.Number, err)
748 }
749 f.head = head[0]
750 // The files of the merge request are those between the merge base of the
751 // target and the head, and the head: what the review shows.
752 f.filesTruncated = true // unavailable unless read below: a paths filter then passes
753 if !st.base.IsZero() {
754 if mb, err := c.cfg.Git.MergeBase(ctx, e.repo, st.base, st.head); err == nil {
755 if paths, truncated, err := c.cfg.Diff.ChangedPaths(ctx, e.repo, mb, st.head, workflow.MaxFilesEvaluated); err == nil {
756 f.files, f.filesTruncated = paths, truncated
757 }
758 }
759 }
760 files, err := c.workflowFiles(ctx, e.repo, sha)
761 if err != nil || len(files) == 0 {
762 return err
763 }
764 if workflow.SkipCI(f.head.Message) {
765 return nil
766 }
767 var errs []error
768 for _, file := range files {
769 if file.problem != "" {
770 continue // reported by the push of the branch
771 }
772 wf, perrs := workflow.Parse(file.path, file.content)
773 if len(perrs) > 0 || !matchPullRequest(wf, e.action, mr.Target, f.files, f.filesTruncated) {
774 continue
775 }
776 if problem := c.matrixProblem(ctx, repo, t, f, file.path, wf); problem.text != "" {
777 name := file.path
778 if problem.named {
779 name = workflowName(wf, file.path)
780 }
781 errs = append(errs, c.startInvalid(ctx, repo, t, f, file.path, name, problem.text))
782 continue
783 }
784 errs = append(errs, c.startRun(ctx, repo, t, f, file.path, string(file.content), wf))
785 }
786 return errors.Join(errs...)
787}
788
789// matchPullRequest reports whether the workflow runs for the activity of a merge
790// request into base: the event is listed, the activity is one of its types
791// (opened, synchronize and reopened when it lists none), and the branches and
792// paths filters, which apply to the base branch and the changed files, let it
793// through. The filters are those of the push event (ermite-actions), run on the
794// pull_request event under another name.
795func matchPullRequest(wf *workflow.Workflow, action, base string, files []string, truncated bool) bool {
796 if wf == nil || wf.On == nil {
797 return false
798 }
799 ev := wf.On.Event("pull_request")
800 if ev == nil {
801 return false
802 }
803 types := ev.Types
804 if types == nil {
805 types = domain.DefaultPullRequestTypes
806 }
807 if !slices.Contains(types, action) {
808 return false
809 }
810 as := *ev
811 as.Name = "push"
812 probe := &workflow.Workflow{On: &workflow.On{Events: []*workflow.Event{&as}}}
813 return probe.MatchPush(workflow.PushInfo{Ref: "refs/heads/" + base, ChangedFiles: files, FilesTruncated: truncated})
814}
815
816// --- the github context ---
817
818// pullGithubJSON builds the github context of a pull_request run: github.ref is
819// the merge ref, github.sha the merge commit (the head for a closed request
820// that never merged), and the event is shaped like GitHub's pull_request
821// webhook payload with the fields ermite can fill.
822func (c *CI) pullGithubJSON(repo domain.Repo, run domain.CIRun, f pushFacts) string {
823 p := f.pull
824 mr := p.mr
825 u := c.urls(repo)
826 id := c.identity(run.Actor)
827 owner := c.identity("owner")
828 repoObj := c.repoObj(repo, u, f.defaultBranch, run.CreatedAt)
829 num := strconv.FormatInt(mr.Number, 10)
830 prAPI := u.api + "/repos/" + u.full + "/pulls/" + num
831 html := u.repo + "/merge-requests/" + num
832 side := func(branch string, sha domain.ObjectID) map[string]any {
833 return map[string]any{
834 "label": u.owner + ":" + branch, "ref": branch, "sha": sha.String(),
835 "user": u.userObj(owner, false), "repo": repoObj,
836 }
837 }
838 tm := func(t time.Time) any {
839 if t.IsZero() {
840 return nil
841 }
842 return t.UTC().Format(time.RFC3339)
843 }
844 var body any
845 if s := mr.Description.String(); s != "" {
846 body = s
847 }
848 state, closedAt, mergedAt := "open", any(nil), any(nil)
849 mergeSha := any(nil)
850 if !mr.IsOpen() {
851 state, closedAt = "closed", tm(mr.UpdatedAt)
852 }
853 switch {
854 case p.merged:
855 mergedAt = tm(mr.UpdatedAt)
856 mergeSha = mr.MergedCommitID.String()
857 case f.commit.String() != p.head.String():
858 mergeSha = f.commit.String() // the test merge commit
859 }
860 prID := stableID("pull", u.full+"#"+num)
861 pr := map[string]any{
862 "url": prAPI, "id": prID, "node_id": nodeID("011", "PullRequest", prID), "html_url": html,
863 "diff_url": html + ".diff", "patch_url": html + ".patch", "issue_url": u.api + "/repos/" + u.full + "/issues/" + num,
864 "number": mr.Number, "state": state, "locked": false, "title": mr.Title.String(), "user": u.userObj(c.identity(mr.Author), false),
865 "body": body, "created_at": tm(mr.CreatedAt), "updated_at": tm(mr.UpdatedAt), "closed_at": closedAt, "merged_at": mergedAt,
866 "merge_commit_sha": mergeSha, "assignee": nil, "assignees": []any{}, "requested_reviewers": []any{},
867 "requested_teams": []any{}, "labels": []any{}, "milestone": nil, "draft": false,
868 "head": side(mr.Source, p.head), "base": side(mr.Target, p.base), "author_association": "OWNER",
869 "auto_merge": nil, "active_lock_reason": nil, "merged": p.merged,
870 }
871 event := map[string]any{
872 "action": p.action, "number": mr.Number, "pull_request": pr, "repository": repoObj, "sender": u.userObj(id, false),
873 }
874 if p.action == domain.PullSynchronize {
875 before := p.before.String()
876 if p.before.IsZero() {
877 before = domain.ZeroOID
878 }
879 event["before"], event["after"] = before, p.head.String()
880 }
881 return c.githubContext(repo, run, ghEvent{
882 name: "pull_request", refName: strings.TrimPrefix(run.Ref, domain.PullRefPrefix), refType: "branch",
883 headRef: mr.Source, baseRef: mr.Target, payload: event,
884 })
885}