1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "slices"
10 "strings"
11 "sync"
12 "time"
13
14 "bnema.dev/ermite/internal/domain"
15 "bnema.dev/ermite/internal/ports"
16)
17
18// AutoMergeConfig holds dependencies of the asynchronous merge evaluator.
19type AutoMergeConfig struct {
20 MRs *MergeRequests
21 Store ports.MergeRequestStore
22 Meta ports.RepoMetaStore
23 Git ports.MergeGit
24 Checks ports.CIChecksReader
25 Activity ports.CIActivity
26 Tokens ports.TokenStore
27 Clock ports.Clock
28 After func(time.Duration) <-chan time.Time
29}
30
31// AutoMerge serializes evaluation within each repository. Triggers only enqueue
32// work, including when invoked while CI or merge-request locks are held.
33type AutoMerge struct {
34 cfg AutoMergeConfig
35 ctx context.Context
36 cancel context.CancelFunc
37 wg sync.WaitGroup
38 mu sync.Mutex
39 pending map[domain.RepoName]bool
40 dirty map[domain.RepoName]bool
41 order []domain.RepoName
42 running, again map[domain.RepoName]bool
43 wake chan struct{}
44 retryMu sync.Mutex
45 retries map[domain.RepoName]int
46 scheduled map[domain.RepoName]bool
47 heap []retryItem
48 retryKick chan struct{}
49 closed bool
50}
51
52var _ ports.AutoMergeTrigger = (*AutoMerge)(nil)
53
54// NewAutoMerge builds an evaluator; call Start once, and Close before MRs.Wait.
55func NewAutoMerge(ctx context.Context, cfg AutoMergeConfig) *AutoMerge {
56 if cfg.After == nil {
57 cfg.After = time.After
58 }
59 ctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
60 return &AutoMerge{cfg: cfg, ctx: ctx, cancel: cancel,
61 pending: map[domain.RepoName]bool{}, dirty: map[domain.RepoName]bool{}, running: map[domain.RepoName]bool{}, again: map[domain.RepoName]bool{}, wake: make(chan struct{}, 1),
62 retries: map[domain.RepoName]int{}, scheduled: map[domain.RepoName]bool{}, retryKick: make(chan struct{}, 1)}
63}
64
65// Start starts workers and discovers repositories, retrying failed discovery.
66func (s *AutoMerge) Start() {
67 s.wg.Add(2 + defaultMergeabilityWorkers)
68 go s.scheduler()
69 go func() {
70 defer s.wg.Done()
71 for attempt := 0; attempt <= mergeabilityRetries; attempt++ {
72 repos, err := s.cfg.Store.ListOpenAll(s.ctx)
73 if err != nil && s.ctx.Err() == nil {
74 ctxLog(s.ctx).Error().Err(err).Int("attempt", attempt+1).Msg("auto-merge: list the repositories at startup")
75 }
76 if err == nil {
77 for _, repo := range repos {
78 if repo.AutoMerge != nil {
79 s.OnAutoMergeEnabled(s.ctx, repo.Repo)
80 }
81 }
82 return
83 }
84 if attempt == mergeabilityRetries {
85 return
86 }
87 select {
88 case <-s.ctx.Done():
89 return
90 case <-s.cfg.After(mergeBackoff(attempt + 1)):
91 }
92 }
93 }()
94 for range defaultMergeabilityWorkers {
95 go s.loop()
96 }
97}
98
99// Close cancels workers and pending retry waits, then waits for them.
100func (s *AutoMerge) Close(_ context.Context) {
101 s.retryMu.Lock()
102 s.closed = true
103 s.retryMu.Unlock()
104 s.cancel()
105 s.wg.Wait()
106}
107
108func (s *AutoMerge) signal() {
109 select {
110 case s.wake <- struct{}{}:
111 default:
112 }
113}
114
115func (s *AutoMerge) markDirty(repo domain.RepoName) {
116 s.mu.Lock()
117 if !s.pending[repo] {
118 s.mu.Unlock()
119 return
120 }
121 if s.running[repo] {
122 s.again[repo] = true
123 } else if !s.dirty[repo] {
124 s.dirty[repo] = true
125 s.order = append(s.order, repo)
126 }
127 s.mu.Unlock()
128 s.signal()
129}
130
131func (s *AutoMerge) next() (domain.RepoName, bool) {
132 s.mu.Lock()
133 defer s.mu.Unlock()
134 if len(s.order) == 0 {
135 return domain.RepoName{}, false
136 }
137 repo := s.order[0]
138 s.order = s.order[1:]
139 delete(s.dirty, repo)
140 s.running[repo] = true
141 if len(s.order) > 0 {
142 s.signal()
143 }
144 return repo, true
145}
146
147func (s *AutoMerge) loop() {
148 defer s.wg.Done()
149 for s.ctx.Err() == nil {
150 repo, found := s.next()
151 if !found {
152 select {
153 case <-s.ctx.Done():
154 return
155 case <-s.wake:
156 }
157 continue
158 }
159 if s.evaluate(repo) {
160 s.retryDone(repo)
161 } else {
162 s.retry(repo)
163 }
164 s.mu.Lock()
165 delete(s.running, repo)
166 if s.again[repo] {
167 delete(s.again, repo)
168 s.dirty[repo] = true
169 s.order = append(s.order, repo)
170 s.signal()
171 }
172 s.mu.Unlock()
173 }
174}
175
176// OnRepoChanged marks and wakes only; it never merges on the caller's stack.
177func (s *AutoMerge) OnRepoChanged(_ context.Context, repo domain.RepoName) { s.markDirty(repo) }
178
179// OnAutoMergeEnabled remembers an intent before queuing evaluation.
180func (s *AutoMerge) OnAutoMergeEnabled(_ context.Context, repo domain.RepoName) {
181 s.mu.Lock()
182 s.pending[repo] = true
183 if s.running[repo] {
184 s.again[repo] = true
185 }
186 s.mu.Unlock()
187 s.markDirty(repo)
188}
189
190func (s *AutoMerge) evaluate(repo domain.RepoName) bool {
191 err := s.evaluateRepo(repo)
192 if err != nil && s.ctx.Err() == nil {
193 ctxLog(s.ctx).Error().Err(err).Str("repo", repo.String()).Msg("auto-merge: evaluate repository")
194 }
195 return err == nil || s.ctx.Err() != nil
196}
197
198func (s *AutoMerge) evaluateRepo(repo domain.RepoName) error {
199 ctx := s.ctx
200 meta, err := s.cfg.Meta.Get(ctx, repo)
201 if errors.Is(err, domain.ErrNotFound) {
202 return nil
203 }
204 if err != nil {
205 return err
206 }
207 mrs, err := s.cfg.Store.ListOpen(ctx, repo)
208 if err != nil {
209 return err
210 }
211 hasIntent := slices.ContainsFunc(mrs, func(mr domain.MergeRequest) bool { return mr.AutoMerge != nil })
212 if !hasIntent {
213 s.mu.Lock()
214 // An enable arriving during this scan requests another pass. Do not
215 // erase its mark based on the older store snapshot.
216 if !s.again[repo] {
217 delete(s.pending, repo)
218 }
219 s.mu.Unlock()
220 }
221 var failures []error
222 for _, mr := range mrs {
223 if ctx.Err() != nil {
224 return nil
225 }
226 if mr.AutoMerge == nil {
227 continue
228 }
229 if err := s.evaluateMR(ctx, meta, mr); err != nil {
230 failures = append(failures, err)
231 }
232 }
233 return errors.Join(failures...)
234}
235
236func (s *AutoMerge) evaluateMR(ctx context.Context, repo domain.Repo, mr domain.MergeRequest) error {
237 intent := mr.AutoMerge
238 settings := repo.EffectiveSettings()
239 cancel := func(reason string) error { return s.cfg.MRs.cancelAuto(ctx, mr, reason) }
240 if !settings.AllowAutoMerge {
241 return cancel("auto-merge was turned off in the repository settings")
242 }
243 if !settings.Allows(intent.Mode) {
244 return cancel(fmt.Sprintf("the %s merge mode is no longer allowed", intent.Mode))
245 }
246 head, err := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
247 if errors.Is(err, domain.ErrNotFound) || (err == nil && head.IsZero()) {
248 return cancel("the source branch was deleted")
249 }
250 if err != nil {
251 return err
252 }
253 if head != intent.HeadID {
254 return cancel("the branch moved to " + head.String()[:7])
255 }
256 // An enabler that can no longer merge cancels at once, even while CI is
257 // pending: the intent must not stay shown as enabled. This check only
258 // cancels; the merge itself resolves the enabler again under the merge lock.
259 if _, err := enablerCanMerge(ctx, s.cfg.Tokens, intent, mr, s.cfg.Clock.Now()); err != nil {
260 if errors.Is(err, domain.ErrForbidden) {
261 return cancel(autoMergeReason(err))
262 }
263 return err
264 }
265 if s.cfg.Activity != nil && !s.cfg.Activity.Settled(ctx, mr.Repo, mr.Number) {
266 return nil
267 }
268 checks, err := s.cfg.Checks.CommitChecks(ctx, mr.Repo, head.String())
269 if err != nil {
270 return err
271 }
272 if !hasSuccessfulCheck(checks) {
273 return nil
274 }
275 _, err = s.cfg.MRs.mergeAs(ctx, domain.Principal{}, mr.Repo, mr.Number, ports.MergeParams{Mode: intent.Mode, HeadID: intent.HeadID}, mergeOpts{auto: &autoMergeOpts{intent: intent, tokens: s.cfg.Tokens, activity: s.cfg.Activity}})
276 switch {
277 case err == nil, errors.Is(err, errAutoMergeWait):
278 return nil
279 case errors.Is(err, domain.ErrPreconditionFailed), errors.Is(err, domain.ErrForbidden), errors.Is(err, domain.ErrUnauthenticated):
280 return cancel(autoMergeReason(err))
281 default:
282 return err
283 }
284}
285
286// Removing a known sentinel's presentation prefix is display formatting, not
287// error classification (which always uses errors.Is).
288func autoMergeReason(err error) string {
289 text := err.Error()
290 for _, sentinel := range []error{domain.ErrPreconditionFailed, domain.ErrForbidden, domain.ErrUnauthenticated} {
291 text = strings.TrimPrefix(text, sentinel.Error()+": ")
292 }
293 return text
294}
295
296func (s *AutoMerge) retry(key domain.RepoName) {
297 s.retryMu.Lock()
298 defer s.retryMu.Unlock()
299 if s.closed || s.scheduled[key] {
300 return
301 }
302 s.retries[key]++
303 n := s.retries[key]
304 if n > mergeabilityRetries {
305 delete(s.retries, key)
306 return
307 }
308 s.scheduled[key] = true
309 it := retryItem{due: s.cfg.Clock.Now().Add(mergeBackoff(n)), key: key}
310 i, _ := slices.BinarySearchFunc(s.heap, it, func(a, b retryItem) int {
311 if a.due.After(b.due) {
312 return 1
313 }
314 return -1 // equal dues keep their insertion order
315 })
316 s.heap = slices.Insert(s.heap, i, it)
317 if i == 0 {
318 select {
319 case s.retryKick <- struct{}{}:
320 default:
321 }
322 }
323}
324
325// scheduler is the only goroutine that waits for retries: it sleeps until the
326// earliest is due and puts it back in the normal queue.
327func (s *AutoMerge) scheduler() {
328 defer s.wg.Done()
329 // The timer armed for waited; a kick re-arms it only when the earliest
330 // retry changed, so a stale kick never starts a second timer.
331 var (
332 after <-chan time.Time
333 waited retryItem
334 )
335 for {
336 s.retryMu.Lock()
337 var head retryItem
338 has := len(s.heap) > 0
339 if has {
340 head = s.heap[0]
341 }
342 s.retryMu.Unlock()
343 if !has {
344 after = nil
345 select {
346 case <-s.ctx.Done():
347 return
348 case <-s.retryKick:
349 continue
350 }
351 }
352 if after == nil || head.key != waited.key || !head.due.Equal(waited.due) {
353 after, waited = s.cfg.After(max(0, head.due.Sub(s.cfg.Clock.Now()))), head
354 }
355 select {
356 case <-s.ctx.Done():
357 return
358 case <-s.retryKick:
359 continue // an earlier retry may have arrived
360 case <-after:
361 after = nil
362 }
363 for _, k := range s.popDue(waited) {
364 if s.ctx.Err() != nil {
365 return
366 }
367 s.markDirty(k)
368 }
369 }
370}
371
372// popDue removes the waited item and the ones due as well. The waited item is
373// removed regardless of the clock only when both its key and its deadline
374// match: a replacement retry of the same repository is kept until it is due.
375func (s *AutoMerge) popDue(waited retryItem) []domain.RepoName {
376 s.retryMu.Lock()
377 defer s.retryMu.Unlock()
378 now := s.cfg.Clock.Now()
379 var out []domain.RepoName
380 s.heap = slices.DeleteFunc(s.heap, func(it retryItem) bool {
381 if (it.key != waited.key || !it.due.Equal(waited.due)) && it.due.After(now) {
382 return false
383 }
384 out = append(out, it.key)
385 delete(s.scheduled, it.key)
386 return true
387 })
388 return out
389}
390
391// retryDone forgets the failures of key after a success.
392func (s *AutoMerge) retryDone(key domain.RepoName) {
393 s.retryMu.Lock()
394 delete(s.retries, key)
395 delete(s.scheduled, key)
396 s.heap = slices.DeleteFunc(s.heap, func(it retryItem) bool { return it.key == key })
397 select {
398 case s.retryKick <- struct{}{}:
399 default:
400 }
401 s.retryMu.Unlock()
402}
403
404func enablerCanMerge(ctx context.Context, tokens ports.TokenStore, intent *domain.AutoMerge, mr domain.MergeRequest, now time.Time) (domain.Principal, error) {
405 fresh, valid, err := principalOfPusher(ctx, tokens, intent.EnabledBy, now)
406 if err != nil {
407 return domain.Principal{}, err
408 }
409 if !valid || !fresh.Allows(domain.ScopeMRMerge, mr.Repo) || !fresh.AllowsRef("refs/heads/"+mr.Target) {
410 return domain.Principal{}, fmt.Errorf("%w: the principal that enabled auto-merge can no longer merge %s", domain.ErrForbidden, mr.Target)
411 }
412 return fresh, nil
413}