1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "slices"
9 "strings"
10 "sync"
11 "time"
12
13 "bnema.dev/ermite/internal/domain"
14 "bnema.dev/ermite/internal/ports"
15)
16
17const (
18 defaultMergeabilityWorkers = 2
19 mergeabilityRetries = 5
20 mergeabilityRetryInitial = time.Second
21 mergeabilityRetryMax = time.Minute
22)
23
24// mergeBackoff is the delay before retry number attempt (1-based).
25func mergeBackoff(attempt int) time.Duration {
26 d := mergeabilityRetryInitial
27 for i := 1; i < attempt; i++ {
28 d *= 2
29 if d >= mergeabilityRetryMax {
30 return mergeabilityRetryMax
31 }
32 }
33 return d
34}
35
36// MergeabilityConfig holds the dependencies of the repository reconciler.
37type MergeabilityConfig struct {
38 AutoMerge ports.AutoMergeTrigger
39 Store ports.MergeRequestStore
40 Meta ports.RepoMetaStore
41 Git ports.MergeGit
42 Bus ports.EventBus
43 Clock ports.Clock
44 Workers int // defaults to 2
45 // After is injectable to drive backoff without wall-clock waits in tests.
46 After func(time.Duration) <-chan time.Time
47}
48
49// Mergeability reconciles derived records, with one writer per repository.
50// Its work and retry state are bounded by the number of repositories.
51type Mergeability struct {
52 cfg MergeabilityConfig
53 ctx context.Context
54 cancel context.CancelFunc
55 wg sync.WaitGroup
56 mu sync.Mutex
57 dirty map[domain.RepoName]bool
58 order []domain.RepoName
59 running, again map[domain.RepoName]bool
60 wake chan struct{}
61 retryMu sync.Mutex
62 retries map[domain.RepoName]int
63 scheduled map[domain.RepoName]bool
64 heap []retryItem
65 retryKick chan struct{}
66 closed bool
67}
68
69// MergeabilityHook is the unchanged trigger and reader contract.
70type MergeabilityHook interface {
71 ports.MergeabilityTrigger
72 ports.MergeabilityReader
73}
74
75var _ MergeabilityHook = (*Mergeability)(nil)
76
77// NewMergeability builds a reconciler whose lifetime is controlled by Close.
78func NewMergeability(ctx context.Context, cfg MergeabilityConfig) *Mergeability {
79 if cfg.Workers <= 0 {
80 cfg.Workers = defaultMergeabilityWorkers
81 }
82 if cfg.After == nil {
83 cfg.After = time.After
84 }
85 ctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
86 return &Mergeability{cfg: cfg, ctx: ctx, cancel: cancel,
87 dirty: map[domain.RepoName]bool{}, running: map[domain.RepoName]bool{}, again: map[domain.RepoName]bool{}, wake: make(chan struct{}, 1),
88 retries: map[domain.RepoName]int{}, scheduled: map[domain.RepoName]bool{}, retryKick: make(chan struct{}, 1)}
89}
90
91// Start starts workers and discovers repositories, retrying failed discovery.
92func (s *Mergeability) Start() {
93 s.wg.Add(2 + s.cfg.Workers)
94 go s.scheduler()
95 go func() {
96 defer s.wg.Done()
97 for attempt := 0; attempt <= mergeabilityRetries; attempt++ {
98 repos, err := s.cfg.Meta.List(s.ctx)
99 if err != nil && s.ctx.Err() == nil {
100 ctxLog(s.ctx).Error().Err(err).Int("attempt", attempt+1).Msg("mergeability: list the repositories at startup")
101 }
102 if err == nil {
103 for _, repo := range repos {
104 s.markDirty(repo.Name)
105 }
106 return
107 }
108 if attempt == mergeabilityRetries {
109 return
110 }
111 select {
112 case <-s.ctx.Done():
113 return
114 case <-s.cfg.After(mergeBackoff(attempt + 1)):
115 }
116 }
117 }()
118 for range s.cfg.Workers {
119 go s.loop()
120 }
121}
122
123// Close cancels workers and pending retry waits, then waits for them.
124func (s *Mergeability) Close(_ context.Context) {
125 s.retryMu.Lock()
126 s.closed = true
127 s.retryMu.Unlock()
128 s.cancel()
129 s.wg.Wait()
130}
131
132func (s *Mergeability) signal() {
133 select {
134 case s.wake <- struct{}{}:
135 default:
136 }
137}
138
139func (s *Mergeability) markDirty(repo domain.RepoName) {
140 s.mu.Lock()
141 if s.running[repo] {
142 s.again[repo] = true
143 } else if !s.dirty[repo] {
144 s.dirty[repo] = true
145 s.order = append(s.order, repo)
146 }
147 s.mu.Unlock()
148 s.signal()
149}
150
151func (s *Mergeability) next() (domain.RepoName, bool) {
152 s.mu.Lock()
153 defer s.mu.Unlock()
154 if len(s.order) == 0 {
155 return domain.RepoName{}, false
156 }
157 repo := s.order[0]
158 s.order = s.order[1:]
159 delete(s.dirty, repo)
160 s.running[repo] = true
161 if len(s.order) > 0 {
162 s.signal()
163 }
164 return repo, true
165}
166
167func (s *Mergeability) loop() {
168 defer s.wg.Done()
169 for s.ctx.Err() == nil {
170 repo, found := s.next()
171 if !found {
172 select {
173 case <-s.ctx.Done():
174 return
175 case <-s.wake:
176 }
177 continue
178 }
179 if s.reconcile(repo) {
180 s.retryDone(repo)
181 } else {
182 s.retry(repo)
183 }
184 s.mu.Lock()
185 delete(s.running, repo)
186 if s.again[repo] {
187 delete(s.again, repo)
188 s.dirty[repo] = true
189 s.order = append(s.order, repo)
190 s.signal()
191 }
192 s.mu.Unlock()
193 }
194}
195
196// OnOpened implements ports.MergeabilityTrigger.
197func (s *Mergeability) OnOpened(_ context.Context, mr domain.MergeRequest) { s.markDirty(mr.Repo) }
198
199// OnBranchesMoved dirties the repository for any branch update.
200func (s *Mergeability) OnBranchesMoved(_ context.Context, repo domain.RepoName, updates []domain.RefUpdate) {
201 for _, update := range updates {
202 if strings.HasPrefix(update.Ref, "refs/heads/") {
203 s.markDirty(repo)
204 return
205 }
206 }
207}
208
209// Read returns a current record, or checking and requests reconciliation.
210func (s *Mergeability) Read(ctx context.Context, mr domain.MergeRequest, targetTip domain.ObjectID) domain.Mergeability {
211 if !mr.HeadID.IsZero() && !targetTip.IsZero() {
212 rec, err := s.cfg.Store.GetMergeability(ctx, mr.Repo, mr.Number)
213 if err == nil && rec.IsFor(targetTip, mr.HeadID) {
214 return rec
215 }
216 }
217 s.markDirty(mr.Repo)
218 return domain.Mergeability{Repo: mr.Repo, Number: mr.Number, State: domain.MergeabilityChecking}
219}
220
221// reconcile settles each open request for the tips read in this pass. A push
222// during computation requests another pass; old-tip records are harmless.
223func (s *Mergeability) reconcile(repo domain.RepoName) bool {
224 ctx := s.ctx
225 meta, err := s.cfg.Meta.Get(ctx, repo)
226 if errors.Is(err, domain.ErrNotFound) {
227 return true
228 }
229 if err != nil {
230 s.logFailure(ctx, err, "get the repository", repo, 0)
231 return false
232 }
233 mrs, err := s.cfg.Store.ListOpen(ctx, repo)
234 if err != nil {
235 s.logFailure(ctx, err, "list the open merge requests", repo, 0)
236 return false
237 }
238 ok := true
239 changed := false
240 for _, mr := range mrs {
241 if ctx.Err() != nil {
242 return true
243 }
244 head, tip, found, transient := s.tips(ctx, mr)
245 if !found {
246 ok = ok && !transient
247 continue
248 }
249 prev, err := s.cfg.Store.GetMergeability(ctx, repo, mr.Number)
250 if err != nil && !errors.Is(err, domain.ErrNotFound) {
251 s.logFailure(ctx, err, "get the mergeability", repo, mr.Number)
252 ok = false
253 continue
254 }
255 if prev.IsFor(tip, head) {
256 continue
257 }
258 res, err := s.compute(ctx, repo, mr.Number, tip, head)
259 if err != nil {
260 s.logFailure(ctx, err, "compute the mergeability", repo, mr.Number)
261 ok = false
262 continue
263 }
264 res.Version = prev.Version
265 err = s.cfg.Store.PutMergeability(ctx, res)
266 if errors.Is(err, domain.ErrNotFound) {
267 continue
268 }
269 if err != nil {
270 if !errors.Is(err, domain.ErrConflict) { // retried by the next pass
271 s.logFailure(ctx, err, "store the mergeability", repo, mr.Number)
272 }
273 ok = false
274 continue
275 }
276 s.cfg.Bus.Publish(ctx, domain.MergeRequestMergeability{
277 Repo: repo, Number: mr.Number, Mergeability: res, At: res.CheckedAt,
278 Visibility: meta.Visibility, CreatedAt: meta.CreatedAt,
279 })
280 changed = true
281 }
282 if changed && s.cfg.AutoMerge != nil {
283 s.cfg.AutoMerge.OnRepoChanged(ctx, repo)
284 }
285 return ok
286}
287
288// logFailure logs an unexpected failure; cancellation is expected on Close.
289func (s *Mergeability) logFailure(ctx context.Context, err error, op string, repo domain.RepoName, number int64) {
290 if ctx.Err() != nil || errors.Is(err, context.Canceled) || errors.Is(err, domain.ErrNotFound) {
291 return
292 }
293 e := ctxLog(ctx).Error().Err(err).Str("repo", repo.String())
294 if number != 0 {
295 e = e.Int64("number", number)
296 }
297 e.Msg("mergeability: " + op)
298}
299
300// retryItem is one scheduled retry.
301type retryItem struct {
302 due time.Time
303 key domain.RepoName
304}
305
306// retry schedules a retry of key after a bounded exponential backoff. It is
307// deduplicated per key, gives up after mergeabilityRetries attempts and stops
308// on Close. All retries share one scheduler goroutine.
309func (s *Mergeability) retry(key domain.RepoName) {
310 s.retryMu.Lock()
311 defer s.retryMu.Unlock()
312 if s.closed || s.scheduled[key] {
313 return
314 }
315 s.retries[key]++
316 n := s.retries[key]
317 if n > mergeabilityRetries {
318 delete(s.retries, key)
319 return
320 }
321 s.scheduled[key] = true
322 it := retryItem{due: s.cfg.Clock.Now().Add(mergeBackoff(n)), key: key}
323 i, _ := slices.BinarySearchFunc(s.heap, it, func(a, b retryItem) int {
324 if a.due.After(b.due) {
325 return 1
326 }
327 return -1 // equal dues keep their insertion order
328 })
329 s.heap = slices.Insert(s.heap, i, it)
330 if i == 0 {
331 select {
332 case s.retryKick <- struct{}{}:
333 default:
334 }
335 }
336}
337
338// scheduler is the only goroutine that waits for retries: it sleeps until the
339// earliest is due and puts it back in the normal queue.
340func (s *Mergeability) scheduler() {
341 defer s.wg.Done()
342 // The timer armed for waited; a kick re-arms it only when the earliest
343 // retry changed, so a stale kick never starts a second timer.
344 var (
345 after <-chan time.Time
346 waited retryItem
347 )
348 for {
349 s.retryMu.Lock()
350 var head retryItem
351 has := len(s.heap) > 0
352 if has {
353 head = s.heap[0]
354 }
355 s.retryMu.Unlock()
356 if !has {
357 after = nil
358 select {
359 case <-s.ctx.Done():
360 return
361 case <-s.retryKick:
362 continue
363 }
364 }
365 if after == nil || head.key != waited.key || !head.due.Equal(waited.due) {
366 after, waited = s.cfg.After(max(0, head.due.Sub(s.cfg.Clock.Now()))), head
367 }
368 select {
369 case <-s.ctx.Done():
370 return
371 case <-s.retryKick:
372 continue // an earlier retry may have arrived
373 case <-after:
374 after = nil
375 }
376 for _, k := range s.popDue(waited) {
377 if s.ctx.Err() != nil {
378 return
379 }
380 s.markDirty(k)
381 }
382 }
383}
384
385// popDue removes the waited item and the ones due as well. The waited item is
386// removed regardless of the clock only when both its key and its deadline
387// match: a replacement retry of the same repository is kept until it is due.
388func (s *Mergeability) popDue(waited retryItem) []domain.RepoName {
389 s.retryMu.Lock()
390 defer s.retryMu.Unlock()
391 now := s.cfg.Clock.Now()
392 var out []domain.RepoName
393 s.heap = slices.DeleteFunc(s.heap, func(it retryItem) bool {
394 if (it.key != waited.key || !it.due.Equal(waited.due)) && it.due.After(now) {
395 return false
396 }
397 out = append(out, it.key)
398 delete(s.scheduled, it.key)
399 return true
400 })
401 return out
402}
403
404// retryDone forgets the failures of key after a success.
405func (s *Mergeability) retryDone(key domain.RepoName) {
406 s.retryMu.Lock()
407 delete(s.retries, key)
408 delete(s.scheduled, key)
409 s.heap = slices.DeleteFunc(s.heap, func(it retryItem) bool { return it.key == key })
410 select {
411 case s.retryKick <- struct{}{}:
412 default:
413 }
414 s.retryMu.Unlock()
415}
416
417// tips reads the heads of the source and the target of mr. ok is false when
418// the check is over: a branch is gone (the record is deleted) or Git failed,
419// which is transient.
420func (s *Mergeability) tips(ctx context.Context, mr domain.MergeRequest) (head, tip domain.ObjectID, ok, transient bool) {
421 head, herr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
422 tip, terr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
423 switch {
424 case errors.Is(herr, domain.ErrNotFound) || errors.Is(terr, domain.ErrNotFound):
425 if err := s.cfg.Store.DeleteMergeability(ctx, mr.Repo, mr.Number); err != nil {
426 if ctx.Err() == nil {
427 ctxLog(ctx).Error().Err(err).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: delete the record of a merge request without a branch")
428 }
429 transient = ctx.Err() == nil
430 }
431 return domain.ObjectID{}, domain.ObjectID{}, false, transient
432 case herr != nil || terr != nil:
433 if ctx.Err() != nil {
434 return domain.ObjectID{}, domain.ObjectID{}, false, false
435 }
436 ctxLog(ctx).Error().Err(errors.Join(herr, terr)).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: read the branches")
437 return domain.ObjectID{}, domain.ObjectID{}, false, true
438 }
439 return head, tip, true, false
440}
441
442// compute runs the Git reads that decide the state of head against tip.
443func (s *Mergeability) compute(ctx context.Context, repo domain.RepoName, number int64, tip, head domain.ObjectID) (domain.Mergeability, error) {
444 m := domain.Mergeability{Repo: repo, Number: number, TargetID: tip, HeadID: head, CheckedAt: s.cfg.Clock.Now().UTC()}
445 ancestor := tip == head
446 if !ancestor {
447 var err error
448 if ancestor, err = s.cfg.Git.IsAncestor(ctx, repo, tip, head); err != nil {
449 return m, err
450 }
451 }
452 if ancestor {
453 m.State = domain.MergeabilityUpToDate
454 return m, nil
455 }
456 tree, err := s.cfg.Git.MergeTree(ctx, repo, tip, head)
457 if err != nil {
458 return m, err
459 }
460 if !tree.Clean {
461 m.State = domain.MergeabilityConflict
462 tree = tree.Bounded()
463 m.ConflictPaths, m.ConflictPathsTruncated = tree.ConflictPaths, tree.Truncated
464 return m, nil
465 }
466 m.State = domain.MergeabilityBehind
467 n, err := s.cfg.Git.CountBehind(ctx, repo, head, tip, domain.BehindLimit)
468 if err != nil {
469 return m, err
470 }
471 m.Behind, m.BehindCapped = domain.BoundBehind(n)
472 return m, nil
473}
474
475// WithAutoMerge installs the evaluator before Start.
476func (s *Mergeability) WithAutoMerge(t ports.AutoMergeTrigger) *Mergeability {
477 s.cfg.AutoMerge = t
478 return s
479}