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