fix(mr): retry failed mergeability checks with backoff

commit: f8b1d20a21727275851151108ddd20cdf9b738d0
author: brice · 2d ago · 2026-10-19 14:52 UTC
committer: brice · 2d ago · 2026-10-19 18:30 UTC
parent: f8bb86d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +291 -30

M internal/usecase/mergeability.go +152 -30


@@ -8,6 +8,7 @@ import (
8	8		"slices"
9	9		"strings"
10	10		"sync"
	11	+	"time"
11	12
12	13		"bnema.dev/ermite/internal/domain"
13	14		"bnema.dev/ermite/internal/ports"

@@ -20,8 +21,40 @@ const (
20	21		// race with a moving branch or a concurrent write; the check is then queued
21	22		// again.
22	23		mergeabilityAttempts = 3
	24	+	// Failed checks (Git or store errors) are retried with an exponential
	25	+	// backoff, up to mergeabilityRetries times, then dropped until the next
	26	+	// trigger.
	27	+	mergeabilityRetries = 5
	28	+	mergeabilityRetryInitial = time.Second
	29	+	mergeabilityRetryMax = time.Minute
23	30		)
24	31
	32	+	// retryKind tells what a scheduled retry redoes.
	33	+	type retryKind int
	34	+
	35	+	const (
	36	+	retryCheck retryKind = iota // one merge request
	37	+	retryRescan // one repository
	38	+	retryRescanAll
	39	+	)
	40	+
	41	+	type retryKey struct {
	42	+	kind retryKind
	43	+	mk mergeKey
	44	+	}
	45	+
	46	+	// mergeBackoff is the delay before retry number attempt (1-based).
	47	+	func mergeBackoff(attempt int) time.Duration {
	48	+	d := mergeabilityRetryInitial
	49	+	for i := 1; i < attempt; i++ {
	50	+	d *= 2
	51	+	if d >= mergeabilityRetryMax {
	52	+	return mergeabilityRetryMax
	53	+	}
	54	+	}
	55	+	return d
	56	+	}
	57	+
25	58		// MergeabilityConfig holds what the mergeability service needs. Workers and
26	59		// Queue default to 2 and 1024.
27	60		type MergeabilityConfig struct {

@@ -33,6 +66,9 @@ type MergeabilityConfig struct {
33	66		// Workers is the number of checks that run at once, Queue the number of
34	67		// checks that may wait; past it a repository is rescanned.
35	68		Workers, Queue int
	69	+	// After returns a channel that fires after d; it defaults to time.After.
	70	+	// Tests inject it to drive the retry backoff.
	71	+	After func(d time.Duration) <-chan time.Time
36	72		}
37	73
38	74		// Mergeability keeps up to date whether each open merge request merges with

@@ -45,6 +81,11 @@ type Mergeability struct {
45	81		ctx context.Context
46	82		cancel context.CancelFunc
47	83		wg sync.WaitGroup
	84	+
	85	+	retryMu sync.Mutex
	86	+	retries map[retryKey]int // attempts so far of the scheduled or failing work
	87	+	retrying map[retryKey]bool
	88	+	closed bool
48	89		}
49	90
50	91		// MergeabilityHook is what the merge request use cases need from the

@@ -67,8 +108,15 @@ func NewMergeability(ctx context.Context, cfg MergeabilityConfig) *Mergeability
67	108		if cfg.Queue <= 0 {
68	109		cfg.Queue = defaultMergeabilityQueue
69	110		}
70		-	wctx, cancel := context.WithCancel(ctx)
71		-	return &Mergeability{cfg: cfg, q: newMergeQueue(cfg.Queue), ctx: wctx, cancel: cancel}
	111	+	if cfg.After == nil {
	112	+	cfg.After = time.After
	113	+	}
	114	+	// The workers outlive the cancellation of ctx: Close controls their life.
	115	+	wctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
	116	+	return &Mergeability{
	117	+	cfg: cfg, q: newMergeQueue(cfg.Queue), ctx: wctx, cancel: cancel,
	118	+	retries: map[retryKey]int{}, retrying: map[retryKey]bool{},
	119	+	}
72	120		}
73	121
74	122		// Start rescans the open merge requests of every repository (the records may

@@ -83,16 +131,60 @@ func (s *Mergeability) Start() {
83	131
84	132		// Close stops the workers and waits for them.
85	133		func (s *Mergeability) Close(_ context.Context) {
	134	+	s.retryMu.Lock()
	135	+	s.closed = true
	136	+	s.retryMu.Unlock()
86	137		s.cancel()
87	138		s.wg.Wait()
88	139		}
89	140
	141	+	// retry schedules redo after a bounded exponential backoff. It is
	142	+	// deduplicated per key, gives up after mergeabilityRetries attempts and stops
	143	+	// on Close.
	144	+	func (s *Mergeability) retry(key retryKey, redo func()) {
	145	+	s.retryMu.Lock()
	146	+	defer s.retryMu.Unlock()
	147	+	if s.closed \|\| s.retrying[key] {
	148	+	return
	149	+	}
	150	+	s.retries[key]++
	151	+	n := s.retries[key]
	152	+	if n > mergeabilityRetries {
	153	+	delete(s.retries, key)
	154	+	return
	155	+	}
	156	+	s.retrying[key] = true
	157	+	after := s.cfg.After(mergeBackoff(n))
	158	+	s.wg.Add(1)
	159	+	go func() {
	160	+	defer s.wg.Done()
	161	+	select {
	162	+	case <-s.ctx.Done():
	163	+	return
	164	+	case <-after:
	165	+	}
	166	+	s.retryMu.Lock()
	167	+	delete(s.retrying, key)
	168	+	s.retryMu.Unlock()
	169	+	redo()
	170	+	}()
	171	+	}
	172	+
	173	+	// retryDone forgets the failures of key after a success.
	174	+	func (s *Mergeability) retryDone(key retryKey) {
	175	+	s.retryMu.Lock()
	176	+	delete(s.retries, key)
	177	+	s.retryMu.Unlock()
	178	+	}
	179	+
90	180		func (s *Mergeability) rescanAll() {
91	181		repos, err := s.cfg.Meta.List(s.ctx)
92	182		if err != nil {
93	183		ctxLog(s.ctx).Error().Err(err).Msg("mergeability: list the repositories to rescan")
	184	+	s.retry(retryKey{kind: retryRescanAll}, s.rescanAll)
94	185		return
95	186		}
	187	+	s.retryDone(retryKey{kind: retryRescanAll})
96	188		for _, r := range repos {
97	189		n, err := s.cfg.Store.CountOpen(s.ctx, r.Name)
98	190		if err != nil {

@@ -118,6 +210,7 @@ func (s *Mergeability) OnClosed(ctx context.Context, mr domain.MergeRequest) {
118	210		if err := s.cfg.Store.DeleteMergeability(context.WithoutCancel(ctx), mr.Repo, mr.Number); err != nil {
119	211		ctxLog(ctx).Error().Err(err).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: delete the record of a closed merge request")
120	212		}
	213	+	s.q.takeSeen(mergeKey{mr.Repo, mr.Number})
121	214		}
122	215
123	216		// OnBranchesMoved implements ports.MergeabilityTrigger: the open merge

@@ -192,7 +285,12 @@ func (s *Mergeability) loop() {
192	285		// meanwhile.
193	286		func (s *Mergeability) runKey(k mergeKey) {
194	287		for {
195		-	s.check(k)
	288	+	rk := retryKey{kind: retryCheck, mk: k}
	289	+	if s.check(k) {
	290	+	s.retryDone(rk)
	291	+	} else {
	292	+	s.retry(rk, func() { s.q.add(k) })
	293	+	}
196	294		if !s.q.finish(k) \|\| s.ctx.Err() != nil {
197	295		return
198	296		}

@@ -205,8 +303,10 @@ func (s *Mergeability) rescan(repo domain.RepoName) {
205	303		mrs, err := s.cfg.Store.ListOpen(s.ctx, repo)
206	304		if err != nil {
207	305		ctxLog(s.ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: list the open merge requests to rescan")
	306	+	s.retry(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}}, func() { s.q.markDirty(repo) })
208	307		return
209	308		}
	309	+	s.retryDone(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})
210	310		for _, mr := range mrs {
211	311		if s.ctx.Err() != nil {
212	312		return

@@ -227,47 +327,58 @@ func (s *Mergeability) Check(repo domain.RepoName, number int64) {
227	327		}
228	328		}
229	329
230		-	// check brings the record of k up to date.
231		-	func (s *Mergeability) check(k mergeKey) {
	330	+	// check brings the record of k up to date. It reports false when a transient
	331	+	// error (store, Git) dropped the work, which is then retried.
	332	+	func (s *Mergeability) check(k mergeKey) (done bool) {
232	333		ctx := s.ctx
233	334		log := ctxLog(ctx).With().Str("repo", k.repo.String()).Int64("number", k.number).Logger()
234	335		for range mergeabilityAttempts {
235	336		mr, err := s.cfg.Store.Get(ctx, k.repo, k.number)
236	337		if err != nil {
237		-	if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
238		-	log.Error().Err(err).Msg("mergeability: read the merge request")
	338	+	if errors.Is(err, domain.ErrNotFound) {
	339	+	s.q.takeSeen(k)
	340	+	return true
239	341		}
240		-	return
	342	+	if ctx.Err() != nil {
	343	+	return true
	344	+	}
	345	+	log.Error().Err(err).Msg("mergeability: read the merge request")
	346	+	return false
241	347		}
242	348		if !mr.IsOpen() {
243		-	return // the store dropped the record when it left the open state
	349	+	s.q.takeSeen(k)
	350	+	return true // the store dropped the record when it left the open state
244	351		}
245		-	head, tip, ok := s.tips(ctx, mr)
	352	+	head, tip, ok, transient := s.tips(ctx, mr)
246	353		if !ok {
247		-	return
	354	+	return !transient
248	355		}
249	356		prev, err := s.cfg.Store.GetMergeability(ctx, k.repo, k.number)
250	357		if err != nil && !errors.Is(err, domain.ErrNotFound) {
	358	+	if ctx.Err() != nil {
	359	+	return true
	360	+	}
251	361		log.Error().Err(err).Msg("mergeability: read the record")
252		-	return
	362	+	return false
253	363		}
254	364		if prev.IsFor(tip, head) {
255	365		if s.q.takeSeen(k) {
256	366		s.publish(ctx, prev) // a reader saw "checking" before this record
257	367		}
258		-	return
	368	+	return true
259	369		}
260	370		res, err := s.compute(ctx, k, tip, head)
261	371		if err != nil {
262		-	if ctx.Err() == nil {
263		-	log.Error().Err(err).Msg("mergeability: compute")
	372	+	if ctx.Err() != nil {
	373	+	return true
264	374		}
265		-	return
	375	+	log.Error().Err(err).Msg("mergeability: compute")
	376	+	return false
266	377		}
267	378		// The branches may have moved during the computation: the result is
268	379		// then for tips nobody asks about, and is dropped.
269		-	if h2, t2, ok := s.tips(ctx, mr); !ok {
270		-	return
	380	+	if h2, t2, ok, transient := s.tips(ctx, mr); !ok {
	381	+	return !transient
271	382		} else if h2 != head \|\| t2 != tip {
272	383		continue
273	384		}

@@ -276,38 +387,49 @@ func (s *Mergeability) check(k mergeKey) {
276	387		case errors.Is(err, domain.ErrConflict):
277	388		continue
278	389		case errors.Is(err, domain.ErrNotFound):
279		-	return // closed meanwhile
	390	+	s.q.takeSeen(k)
	391	+	return true // closed meanwhile
280	392		case err != nil:
	393	+	if ctx.Err() != nil {
	394	+	return true
	395	+	}
281	396		log.Error().Err(err).Msg("mergeability: store the record")
282		-	return
	397	+	return false
283	398		}
284	399		seen := s.q.takeSeen(k)
285	400		if seen \|\| prev.State == "" \|\| changedMergeability(prev, res) {
286	401		s.publish(ctx, res)
287	402		}
288		-	return
	403	+	return true
289	404		}
290	405		s.q.add(k) // kept losing races: try again later
	406	+	return true
291	407		}
292	408
293	409		// tips reads the heads of the source and the target of mr. ok is false when
294		-	// the check is over: a branch is gone (the record is deleted) or Git failed.
295		-	func (s *Mergeability) tips(ctx context.Context, mr domain.MergeRequest) (head, tip domain.ObjectID, ok bool) {
	410	+	// the check is over: a branch is gone (the record is deleted) or Git failed,
	411	+	// which is transient.
	412	+	func (s *Mergeability) tips(ctx context.Context, mr domain.MergeRequest) (head, tip domain.ObjectID, ok, transient bool) {
296	413		head, herr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
297	414		tip, terr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
298	415		switch {
299	416		case errors.Is(herr, domain.ErrNotFound) \|\| errors.Is(terr, domain.ErrNotFound):
300		-	if err := s.cfg.Store.DeleteMergeability(ctx, mr.Repo, mr.Number); err != nil && ctx.Err() == nil {
301		-	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")
	417	+	if err := s.cfg.Store.DeleteMergeability(ctx, mr.Repo, mr.Number); err != nil {
	418	+	if ctx.Err() == nil {
	419	+	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")
	420	+	}
	421	+	transient = ctx.Err() == nil
302	422		}
303		-	return domain.ObjectID{}, domain.ObjectID{}, false
	423	+	s.q.takeSeen(mergeKey{mr.Repo, mr.Number})
	424	+	return domain.ObjectID{}, domain.ObjectID{}, false, transient
304	425		case herr != nil \|\| terr != nil:
305		-	if ctx.Err() == nil {
306		-	ctxLog(ctx).Error().Err(errors.Join(herr, terr)).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: read the branches")
	426	+	if ctx.Err() != nil {
	427	+	return domain.ObjectID{}, domain.ObjectID{}, false, false
307	428		}
308		-	return domain.ObjectID{}, domain.ObjectID{}, false
	429	+	ctxLog(ctx).Error().Err(errors.Join(herr, terr)).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: read the branches")
	430	+	return domain.ObjectID{}, domain.ObjectID{}, false, true
309	431		}
310		-	return head, tip, true
	432	+	return head, tip, true, false
311	433		}
312	434
313	435		// compute runs the Git reads that decide the state of head against tip.

 	"slices"
 	"strings"
 	"sync"
+	"time"
 	"bnema.dev/ermite/internal/domain"
 	"bnema.dev/ermite/internal/ports"
 	// race with a moving branch or a concurrent write; the check is then queued
 	// again.
 	mergeabilityAttempts = 3
+	// Failed checks (Git or store errors) are retried with an exponential
+	// backoff, up to mergeabilityRetries times, then dropped until the next
+	// trigger.
+	mergeabilityRetries      = 5
+	mergeabilityRetryInitial = time.Second
+	mergeabilityRetryMax     = time.Minute
 )
+// retryKind tells what a scheduled retry redoes.
+type retryKind int
+const (
+	retryCheck  retryKind = iota // one merge request
+	retryRescan                  // one repository
+	retryRescanAll
+)
+type retryKey struct {
+	kind retryKind
+	mk   mergeKey
+}
+// mergeBackoff is the delay before retry number attempt (1-based).
+func mergeBackoff(attempt int) time.Duration {
+	d := mergeabilityRetryInitial
+	for i := 1; i < attempt; i++ {
+		d *= 2
+		if d >= mergeabilityRetryMax {
+			return mergeabilityRetryMax
+		}
+	}
+	return d
+}
 // MergeabilityConfig holds what the mergeability service needs. Workers and
 // Queue default to 2 and 1024.
 type MergeabilityConfig struct {
 	// Workers is the number of checks that run at once, Queue the number of
 	// checks that may wait; past it a repository is rescanned.
 	Workers, Queue int
+	// After returns a channel that fires after d; it defaults to time.After.
+	// Tests inject it to drive the retry backoff.
+	After func(d time.Duration) <-chan time.Time
 }
 // Mergeability keeps up to date whether each open merge request merges with
 	ctx    context.Context
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
+	retryMu  sync.Mutex
+	retries  map[retryKey]int // attempts so far of the scheduled or failing work
+	retrying map[retryKey]bool
+	closed   bool
 }
 // MergeabilityHook is what the merge request use cases need from the
 	if cfg.Queue <= 0 {
 		cfg.Queue = defaultMergeabilityQueue
 	}
-	wctx, cancel := context.WithCancel(ctx)
+	if cfg.After == nil {
-	return &Mergeability{cfg: cfg, q: newMergeQueue(cfg.Queue), ctx: wctx, cancel: cancel}
+		cfg.After = time.After
+	}
+	// The workers outlive the cancellation of ctx: Close controls their life.
+	wctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
+	return &Mergeability{
+		cfg: cfg, q: newMergeQueue(cfg.Queue), ctx: wctx, cancel: cancel,
+		retries: map[retryKey]int{}, retrying: map[retryKey]bool{},
+	}
 }
 // Start rescans the open merge requests of every repository (the records may
 // Close stops the workers and waits for them.
 func (s *Mergeability) Close(_ context.Context) {
+	s.retryMu.Lock()
+	s.closed = true
+	s.retryMu.Unlock()
 	s.cancel()
 	s.wg.Wait()
 }
+// retry schedules redo after a bounded exponential backoff. It is
+// deduplicated per key, gives up after mergeabilityRetries attempts and stops
+// on Close.
+func (s *Mergeability) retry(key retryKey, redo func()) {
+	s.retryMu.Lock()
+	defer s.retryMu.Unlock()
+	if s.closed || s.retrying[key] {
+		return
+	}
+	s.retries[key]++
+	n := s.retries[key]
+	if n > mergeabilityRetries {
+		delete(s.retries, key)
+		return
+	}
+	s.retrying[key] = true
+	after := s.cfg.After(mergeBackoff(n))
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		select {
+		case <-s.ctx.Done():
+			return
+		case <-after:
+		}
+		s.retryMu.Lock()
+		delete(s.retrying, key)
+		s.retryMu.Unlock()
+		redo()
+	}()
+}
+// retryDone forgets the failures of key after a success.
+func (s *Mergeability) retryDone(key retryKey) {
+	s.retryMu.Lock()
+	delete(s.retries, key)
+	s.retryMu.Unlock()
+}
 func (s *Mergeability) rescanAll() {
 	repos, err := s.cfg.Meta.List(s.ctx)
 	if err != nil {
 		ctxLog(s.ctx).Error().Err(err).Msg("mergeability: list the repositories to rescan")
+		s.retry(retryKey{kind: retryRescanAll}, s.rescanAll)
 		return
 	}
+	s.retryDone(retryKey{kind: retryRescanAll})
 	for _, r := range repos {
 		n, err := s.cfg.Store.CountOpen(s.ctx, r.Name)
 		if err != nil {
 	if err := s.cfg.Store.DeleteMergeability(context.WithoutCancel(ctx), mr.Repo, mr.Number); err != nil {
 		ctxLog(ctx).Error().Err(err).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: delete the record of a closed merge request")
 	}
+	s.q.takeSeen(mergeKey{mr.Repo, mr.Number})
 }
 // OnBranchesMoved implements ports.MergeabilityTrigger: the open merge
 // meanwhile.
 func (s *Mergeability) runKey(k mergeKey) {
 	for {
-		s.check(k)
+		rk := retryKey{kind: retryCheck, mk: k}
+		if s.check(k) {
+			s.retryDone(rk)
+		} else {
+			s.retry(rk, func() { s.q.add(k) })
+		}
 		if !s.q.finish(k) || s.ctx.Err() != nil {
 			return
 		}
 	mrs, err := s.cfg.Store.ListOpen(s.ctx, repo)
 	if err != nil {
 		ctxLog(s.ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: list the open merge requests to rescan")
+		s.retry(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}}, func() { s.q.markDirty(repo) })
 		return
 	}
+	s.retryDone(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})
 	for _, mr := range mrs {
 		if s.ctx.Err() != nil {
 			return
 	}
 }
-// check brings the record of k up to date.
+// check brings the record of k up to date. It reports false when a transient
-func (s *Mergeability) check(k mergeKey) {
+// error (store, Git) dropped the work, which is then retried.
+func (s *Mergeability) check(k mergeKey) (done bool) {
 	ctx := s.ctx
 	log := ctxLog(ctx).With().Str("repo", k.repo.String()).Int64("number", k.number).Logger()
 	for range mergeabilityAttempts {
 		mr, err := s.cfg.Store.Get(ctx, k.repo, k.number)
 		if err != nil {
-			if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
+			if errors.Is(err, domain.ErrNotFound) {
-				log.Error().Err(err).Msg("mergeability: read the merge request")
+				s.q.takeSeen(k)
+				return true
 			}
-			return
+			if ctx.Err() != nil {
+				return true
+			}
+			log.Error().Err(err).Msg("mergeability: read the merge request")
+			return false
 		}
 		if !mr.IsOpen() {
-			return // the store dropped the record when it left the open state
+			s.q.takeSeen(k)
+			return true // the store dropped the record when it left the open state
 		}
-		head, tip, ok := s.tips(ctx, mr)
+		head, tip, ok, transient := s.tips(ctx, mr)
 		if !ok {
-			return
+			return !transient
 		}
 		prev, err := s.cfg.Store.GetMergeability(ctx, k.repo, k.number)
 		if err != nil && !errors.Is(err, domain.ErrNotFound) {
+			if ctx.Err() != nil {
+				return true
+			}
 			log.Error().Err(err).Msg("mergeability: read the record")
-			return
+			return false
 		}
 		if prev.IsFor(tip, head) {
 			if s.q.takeSeen(k) {
 				s.publish(ctx, prev) // a reader saw "checking" before this record
 			}
-			return
+			return true
 		}
 		res, err := s.compute(ctx, k, tip, head)
 		if err != nil {
-			if ctx.Err() == nil {
+			if ctx.Err() != nil {
-				log.Error().Err(err).Msg("mergeability: compute")
+				return true
 			}
-			return
+			log.Error().Err(err).Msg("mergeability: compute")
+			return false
 		}
 		// The branches may have moved during the computation: the result is
 		// then for tips nobody asks about, and is dropped.
-		if h2, t2, ok := s.tips(ctx, mr); !ok {
+		if h2, t2, ok, transient := s.tips(ctx, mr); !ok {
-			return
+			return !transient
 		} else if h2 != head || t2 != tip {
 			continue
 		}
 		case errors.Is(err, domain.ErrConflict):
 			continue
 		case errors.Is(err, domain.ErrNotFound):
-			return // closed meanwhile
+			s.q.takeSeen(k)
+			return true // closed meanwhile
 		case err != nil:
+			if ctx.Err() != nil {
+				return true
+			}
 			log.Error().Err(err).Msg("mergeability: store the record")
-			return
+			return false
 		}
 		seen := s.q.takeSeen(k)
 		if seen || prev.State == "" || changedMergeability(prev, res) {
 			s.publish(ctx, res)
 		}
-		return
+		return true
 	}
 	s.q.add(k) // kept losing races: try again later
+	return true
 }
 // tips reads the heads of the source and the target of mr. ok is false when
-// the check is over: a branch is gone (the record is deleted) or Git failed.
+// the check is over: a branch is gone (the record is deleted) or Git failed,
-func (s *Mergeability) tips(ctx context.Context, mr domain.MergeRequest) (head, tip domain.ObjectID, ok bool) {
+// which is transient.
+func (s *Mergeability) tips(ctx context.Context, mr domain.MergeRequest) (head, tip domain.ObjectID, ok, transient bool) {
 	head, herr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
 	tip, terr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
 	switch {
 	case errors.Is(herr, domain.ErrNotFound) || errors.Is(terr, domain.ErrNotFound):
-		if err := s.cfg.Store.DeleteMergeability(ctx, mr.Repo, mr.Number); err != nil && ctx.Err() == nil {
+		if err := s.cfg.Store.DeleteMergeability(ctx, mr.Repo, mr.Number); err != nil {
-			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")
+			if ctx.Err() == nil {
+				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")
+			}
+			transient = ctx.Err() == nil
 		}
-		return domain.ObjectID{}, domain.ObjectID{}, false
+		s.q.takeSeen(mergeKey{mr.Repo, mr.Number})
+		return domain.ObjectID{}, domain.ObjectID{}, false, transient
 	case herr != nil || terr != nil:
-		if ctx.Err() == nil {
+		if ctx.Err() != nil {
-			ctxLog(ctx).Error().Err(errors.Join(herr, terr)).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: read the branches")
+			return domain.ObjectID{}, domain.ObjectID{}, false, false
 		}
-		return domain.ObjectID{}, domain.ObjectID{}, false
+		ctxLog(ctx).Error().Err(errors.Join(herr, terr)).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: read the branches")
+		return domain.ObjectID{}, domain.ObjectID{}, false, true
 	}
-	return head, tip, true
+	return head, tip, true, false
 }
 // compute runs the Git reads that decide the state of head against tip.

A internal/usecase/mergeability_seen_test.go +47 -0

@@ -0,0 +1,47 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/stretchr/testify/assert"
	11	+	"github.com/stretchr/testify/mock"
	12	+	"github.com/stretchr/testify/require"
	13	+
	14	+	"bnema.dev/ermite/internal/domain"
	15	+	"bnema.dev/ermite/internal/ports/mocks"
	16	+	)
	17	+
	18	+	type zeroClock struct{}
	19	+
	20	+	func (zeroClock) Now() time.Time { return time.Time{} }
	21	+
	22	+	func TestMergeabilityBackoffIsBounded(t *testing.T) {
	23	+	assert.Equal(t, time.Second, mergeBackoff(1))
	24	+	assert.Equal(t, 2*time.Second, mergeBackoff(2))
	25	+	assert.Equal(t, 4*time.Second, mergeBackoff(3))
	26	+	assert.Equal(t, time.Minute, mergeBackoff(20))
	27	+	}
	28	+
	29	+	func TestMergeabilityClosingForgetsTheSeenKey(t *testing.T) {
	30	+	repo, err := domain.ParseRepoName("proj")
	31	+	require.NoError(t, err)
	32	+	store := mocks.NewMergeRequestStore(t)
	33	+	store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
	34	+	store.EXPECT().DeleteMergeability(mock.Anything, repo, int64(7)).Return(nil)
	35	+	svc := NewMergeability(context.Background(), MergeabilityConfig{Store: store, Clock: zeroClock{}})
	36	+	t.Cleanup(func() { svc.Close(context.Background()) })
	37	+
	38	+	mr := domain.MergeRequest{Repo: repo, Number: 7, State: domain.MRStateOpen}
	39	+	oid, err := domain.ParseObjectID("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
	40	+	require.NoError(t, err)
	41	+	mr.HeadID = oid
	42	+	svc.Read(context.Background(), mr, oid)
	43	+	require.Len(t, svc.q.seen, 1)
	44	+
	45	+	svc.OnClosed(context.Background(), mr)
	46	+	assert.Empty(t, svc.q.seen)
	47	+	}

@@ -0,0 +1,47 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/stretchr/testify/assert"
	11	+	"github.com/stretchr/testify/mock"
	12	+	"github.com/stretchr/testify/require"
	13	+
	14	+	"bnema.dev/ermite/internal/domain"
	15	+	"bnema.dev/ermite/internal/ports/mocks"
	16	+	)
	17	+
	18	+	type zeroClock struct{}
	19	+
	20	+	func (zeroClock) Now() time.Time { return time.Time{} }
	21	+
	22	+	func TestMergeabilityBackoffIsBounded(t *testing.T) {
	23	+	assert.Equal(t, time.Second, mergeBackoff(1))
	24	+	assert.Equal(t, 2*time.Second, mergeBackoff(2))
	25	+	assert.Equal(t, 4*time.Second, mergeBackoff(3))
	26	+	assert.Equal(t, time.Minute, mergeBackoff(20))
	27	+	}
	28	+
	29	+	func TestMergeabilityClosingForgetsTheSeenKey(t *testing.T) {
	30	+	repo, err := domain.ParseRepoName("proj")
	31	+	require.NoError(t, err)
	32	+	store := mocks.NewMergeRequestStore(t)
	33	+	store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
	34	+	store.EXPECT().DeleteMergeability(mock.Anything, repo, int64(7)).Return(nil)
	35	+	svc := NewMergeability(context.Background(), MergeabilityConfig{Store: store, Clock: zeroClock{}})
	36	+	t.Cleanup(func() { svc.Close(context.Background()) })
	37	+
	38	+	mr := domain.MergeRequest{Repo: repo, Number: 7, State: domain.MRStateOpen}
	39	+	oid, err := domain.ParseObjectID("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
	40	+	require.NoError(t, err)
	41	+	mr.HeadID = oid
	42	+	svc.Read(context.Background(), mr, oid)
	43	+	require.Len(t, svc.q.seen, 1)
	44	+
	45	+	svc.OnClosed(context.Background(), mr)
	46	+	assert.Empty(t, svc.q.seen)
	47	+	}

M internal/usecase/mergeability_test.go +92 -0


@@ -4,6 +4,7 @@ package usecase_test
4	4
5	5		import (
6	6		"context"
	7	+	"errors"
7	8		"strings"
8	9		"sync/atomic"
9	10		"testing"

@@ -460,3 +461,94 @@ func TestMergeabilityBoundsConflictPathsAndBehind(t *testing.T) {
460	461		assert.True(t, put.BehindCapped)
461	462		})
462	463		}
	464	+
	465	+	// A transient Git failure is retried once after the backoff.
	466	+	func TestMergeabilityRetriesATransientFailureWithBackoff(t *testing.T) {
	467	+	repo := name(t, "proj")
	468	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
	469	+	mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
	470	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
	471	+	store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)
	472	+	store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
	473	+	store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
	474	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
	475	+	meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
	476	+	var fails atomic.Int32
	477	+	git.EXPECT().BranchHead(mock.Anything, repo, "feat").RunAndReturn(
	478	+	func(context.Context, domain.RepoName, string) (domain.ObjectID, error) {
	479	+	if fails.Add(1) == 1 {
	480	+	return domain.ObjectID{}, errors.New("transient")
	481	+	}
	482	+	return head, nil
	483	+	})
	484	+	git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
	485	+	git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
	486	+	published := make(chan struct{})
	487	+	bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, domain.Event) { close(published) }).Return().Once()
	488	+
	489	+	asked, fire := make(chan time.Duration, 4), make(chan time.Time)
	490	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
	491	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{},
	492	+	After: func(d time.Duration) <-chan time.Time { asked <- d; return fire },
	493	+	})
	494	+	t.Cleanup(func() { svc.Close(context.Background()) })
	495	+	svc.Start()
	496	+	svc.OnOpened(context.Background(), mr)
	497	+
	498	+	assert.Equal(t, time.Second, <-asked)
	499	+	select {
	500	+	case <-published:
	501	+	t.Fatal("published before the backoff elapsed")
	502	+	default:
	503	+	}
	504	+	fire <- time.Time{}
	505	+	<-published
	506	+	assert.Empty(t, asked, "one retry was enough and only one was scheduled")
	507	+	}
	508	+
	509	+	// Close stops a pending retry timer's goroutine.
	510	+	func TestMergeabilityCloseStopsPendingRetries(t *testing.T) {
	511	+	repo := name(t, "proj")
	512	+	mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
	513	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
	514	+	store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(domain.MergeRequest{}, errors.New("db down"))
	515	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
	516	+	asked := make(chan time.Duration, 4)
	517	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
	518	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{},
	519	+	After: func(d time.Duration) <-chan time.Time { asked <- d; return make(chan time.Time) },
	520	+	})
	521	+	svc.Start()
	522	+	svc.OnOpened(context.Background(), mr)
	523	+	<-asked
	524	+	done := make(chan struct{})
	525	+	go func() { svc.Close(context.Background()); close(done) }()
	526	+	<-done // Close waits for the timer goroutine: it returns only if it stopped
	527	+	}
	528	+
	529	+	// Workers are not stopped by the cancellation of the context given to the
	530	+	// constructor; Close stops them.
	531	+	func TestMergeabilityWorkersOutliveTheParentContext(t *testing.T) {
	532	+	repo := name(t, "proj")
	533	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
	534	+	mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
	535	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
	536	+	store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)
	537	+	store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
	538	+	store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
	539	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
	540	+	meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
	541	+	git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
	542	+	git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
	543	+	git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
	544	+	published := make(chan struct{})
	545	+	bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, domain.Event) { close(published) }).Return().Once()
	546	+
	547	+	parent, cancel := context.WithCancel(context.Background())
	548	+	svc := usecase.NewMergeability(parent, usecase.MergeabilityConfig{Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}})
	549	+	svc.Start()
	550	+	cancel()
	551	+	svc.OnOpened(context.Background(), mr)
	552	+	<-published // computed although the parent is cancelled
	553	+	svc.Close(context.Background())
	554	+	}


@@ -4,6 +4,7 @@ package usecase_test
4		4
5	import (	5		import (
6	"context"	6		"context"
		7	+	"errors"
7	"strings"	8		"strings"
8	"sync/atomic"	9		"sync/atomic"
9	"testing"	10		"testing"

@@ -460,3 +461,94 @@ func TestMergeabilityBoundsConflictPathsAndBehind(t *testing.T) {
460	assert.True(t, put.BehindCapped)	461		assert.True(t, put.BehindCapped)
461	})	462		})
462	}	463		}
		464	+
		465	+	// A transient Git failure is retried once after the backoff.
		466	+	func TestMergeabilityRetriesATransientFailureWithBackoff(t *testing.T) {
		467	+	repo := name(t, "proj")
		468	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
		469	+	mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
		470	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
		471	+	store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)
		472	+	store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
		473	+	store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
		474	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
		475	+	meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
		476	+	var fails atomic.Int32
		477	+	git.EXPECT().BranchHead(mock.Anything, repo, "feat").RunAndReturn(
		478	+	func(context.Context, domain.RepoName, string) (domain.ObjectID, error) {
		479	+	if fails.Add(1) == 1 {
		480	+	return domain.ObjectID{}, errors.New("transient")
		481	+	}
		482	+	return head, nil
		483	+	})
		484	+	git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
		485	+	git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
		486	+	published := make(chan struct{})
		487	+	bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, domain.Event) { close(published) }).Return().Once()
		488	+
		489	+	asked, fire := make(chan time.Duration, 4), make(chan time.Time)
		490	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
		491	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{},
		492	+	After: func(d time.Duration) <-chan time.Time { asked <- d; return fire },
		493	+	})
		494	+	t.Cleanup(func() { svc.Close(context.Background()) })
		495	+	svc.Start()
		496	+	svc.OnOpened(context.Background(), mr)
		497	+
		498	+	assert.Equal(t, time.Second, <-asked)
		499	+	select {
		500	+	case <-published:
		501	+	t.Fatal("published before the backoff elapsed")
		502	+	default:
		503	+	}
		504	+	fire <- time.Time{}
		505	+	<-published
		506	+	assert.Empty(t, asked, "one retry was enough and only one was scheduled")
		507	+	}
		508	+
		509	+	// Close stops a pending retry timer's goroutine.
		510	+	func TestMergeabilityCloseStopsPendingRetries(t *testing.T) {
		511	+	repo := name(t, "proj")
		512	+	mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
		513	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
		514	+	store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(domain.MergeRequest{}, errors.New("db down"))
		515	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
		516	+	asked := make(chan time.Duration, 4)
		517	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
		518	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{},
		519	+	After: func(d time.Duration) <-chan time.Time { asked <- d; return make(chan time.Time) },
		520	+	})
		521	+	svc.Start()
		522	+	svc.OnOpened(context.Background(), mr)
		523	+	<-asked
		524	+	done := make(chan struct{})
		525	+	go func() { svc.Close(context.Background()); close(done) }()
		526	+	<-done // Close waits for the timer goroutine: it returns only if it stopped
		527	+	}
		528	+
		529	+	// Workers are not stopped by the cancellation of the context given to the
		530	+	// constructor; Close stops them.
		531	+	func TestMergeabilityWorkersOutliveTheParentContext(t *testing.T) {
		532	+	repo := name(t, "proj")
		533	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
		534	+	mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
		535	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
		536	+	store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)
		537	+	store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
		538	+	store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
		539	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
		540	+	meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
		541	+	git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
		542	+	git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
		543	+	git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
		544	+	published := make(chan struct{})
		545	+	bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, domain.Event) { close(published) }).Return().Once()
		546	+
		547	+	parent, cancel := context.WithCancel(context.Background())
		548	+	svc := usecase.NewMergeability(parent, usecase.MergeabilityConfig{Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}})
		549	+	svc.Start()
		550	+	cancel()
		551	+	svc.OnOpened(context.Background(), mr)
		552	+	<-published // computed although the parent is cancelled
		553	+	svc.Close(context.Background())
		554	+	}