refactor(mr): retry failed mergeability work per repository only

commit: 0019f14a4f57270e4c7a93968272b797839936cd
author: brice · 2d ago · 2026-10-19 15:46 UTC
committer: brice · 2d ago · 2026-10-19 18:30 UTC
parent: 99b00c3
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +134 -101

M internal/usecase/export_test.go +2 -1


@@ -183,7 +183,8 @@ func (s *Mergeability) ScheduledRetries() int {
183	183		return len(s.heap)
184	184		}
185	185
186		-	// RetryAttempts gives the number of keys whose failed attempts are counted.
	186	+	// RetryAttempts gives the number of repositories (and the global key) whose
	187	+	// failed attempts are counted.
187	188		func (s *Mergeability) RetryAttempts() int {
188	189		s.retryMu.Lock()
189	190		defer s.retryMu.Unlock()


@@ -183,7 +183,8 @@ func (s *Mergeability) ScheduledRetries() int {
183		return len(s.heap)	183		return len(s.heap)
184		}	184		}
185			185
186	-	// RetryAttempts gives the number of keys whose failed attempts are counted.	186	+	// RetryAttempts gives the number of repositories (and the global key) whose
			187	+	// failed attempts are counted.
187		func (s *Mergeability) RetryAttempts() int {	188		func (s *Mergeability) RetryAttempts() int {
188		s.retryMu.Lock()	189		s.retryMu.Lock()
189		defer s.retryMu.Unlock()	190		defer s.retryMu.Unlock()

M internal/usecase/mergeability.go +47 -66


@@ -29,20 +29,18 @@ const (
29	29		mergeabilityRetryMax = time.Minute
30	30		)
31	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		-
	32	+	// retryKey names the work a scheduled retry redoes: the rescan of one
	33	+	// repository, or (all) the rescan of every repository. Retry state is thus
	34	+	// bounded by the number of repositories plus one.
41	35		type retryKey struct {
42		-	kind retryKind
43		-	mk mergeKey
	36	+	repo domain.RepoName
	37	+	all bool
44	38		}
45	39
	40	+	func repoKey(repo domain.RepoName) retryKey { return retryKey{repo: repo} }
	41	+
	42	+	var globalKey = retryKey{all: true}
	43	+
46	44		// mergeBackoff is the delay before retry number attempt (1-based).
47	45		func mergeBackoff(attempt int) time.Duration {
48	46		d := mergeabilityRetryInitial

@@ -82,13 +80,12 @@ type Mergeability struct {
82	80		cancel context.CancelFunc
83	81		wg sync.WaitGroup
84	82
85		-	retryMu sync.Mutex
86		-	retries map[retryKey]int // attempts so far of the scheduled or failing work
87		-	scheduled map[retryKey]bool
88		-	heap []retryItem // sorted by due time
89		-	checkRetries int // scheduled retries of kind retryCheck
90		-	retryKick chan struct{}
91		-	closed bool
	83	+	retryMu sync.Mutex
	84	+	retries map[retryKey]int // attempts so far of the scheduled or failing work
	85	+	scheduled map[retryKey]bool
	86	+	heap []retryItem // sorted by due time
	87	+	retryKick chan struct{}
	88	+	closed bool
92	89		}
93	90
94	91		// MergeabilityHook is what the merge request use cases need from the

@@ -151,37 +148,21 @@ type retryItem struct {
151	148
152	149		// retry schedules a retry of key after a bounded exponential backoff. It is
153	150		// deduplicated per key, gives up after mergeabilityRetries attempts and stops
154		-	// on Close. All retries share one scheduler goroutine and at most cfg.Queue
155		-	// per-request retries wait at once; past it the retry of a request becomes
156		-	// the retry of its repository's rescan (one per repository), so memory stays
157		-	// bounded.
	151	+	// on Close. All retries share one scheduler goroutine.
158	152		func (s *Mergeability) retry(key retryKey) {
159	153		s.retryMu.Lock()
160	154		defer s.retryMu.Unlock()
161	155		if s.closed \|\| s.scheduled[key] {
162	156		return
163	157		}
164		-	// The key that waits is chosen before counting attempts, so that only
165		-	// the waiting keys hold attempt counts: at most cfg.Queue requests plus
166		-	// one per repository.
167		-	sk := key
168		-	if key.kind == retryCheck && s.retries[key] == 0 && s.checkRetries >= s.cfg.Queue {
169		-	sk = retryKey{kind: retryRescan, mk: mergeKey{repo: key.mk.repo}}
170		-	if s.scheduled[sk] {
171		-	return
172		-	}
173		-	}
174		-	s.retries[sk]++
175		-	n := s.retries[sk]
	158	+	s.retries[key]++
	159	+	n := s.retries[key]
176	160		if n > mergeabilityRetries {
177		-	delete(s.retries, sk)
	161	+	delete(s.retries, key)
178	162		return
179	163		}
180		-	if sk.kind == retryCheck {
181		-	s.checkRetries++
182		-	}
183		-	s.scheduled[sk] = true
184		-	it := retryItem{due: s.cfg.Clock.Now().Add(mergeBackoff(n)), key: sk}
	164	+	s.scheduled[key] = true
	165	+	it := retryItem{due: s.cfg.Clock.Now().Add(mergeBackoff(n)), key: key}
185	166		i, _ := slices.BinarySearchFunc(s.heap, it, func(a, b retryItem) int {
186	167		if a.due.After(b.due) {
187	168		return 1

@@ -256,9 +237,6 @@ func (s *Mergeability) popDue(waited retryKey) []retryKey {
256	237		}
257	238		out = append(out, it.key)
258	239		delete(s.scheduled, it.key)
259		-	if it.key.kind == retryCheck {
260		-	s.checkRetries--
261		-	}
262	240		return true
263	241		})
264	242		return out

@@ -267,14 +245,11 @@ func (s *Mergeability) popDue(waited retryKey) []retryKey {
267	245		// fire puts the work of a due retry back in the normal queue, so that the
268	246		// worker limits apply.
269	247		func (s *Mergeability) fire(k retryKey) {
270		-	switch k.kind {
271		-	case retryCheck:
272		-	s.q.add(k.mk)
273		-	case retryRescan:
274		-	s.q.markDirty(k.mk.repo)
275		-	case retryRescanAll:
	248	+	if k.all {
276	249		s.rescanAll()
	250	+	return
277	251		}
	252	+	s.q.markDirty(k.repo)
278	253		}
279	254
280	255		// retryDone forgets the failures of key after a success.

@@ -288,10 +263,10 @@ func (s *Mergeability) rescanAll() {
288	263		repos, err := s.cfg.Meta.List(s.ctx)
289	264		if err != nil {
290	265		ctxLog(s.ctx).Error().Err(err).Msg("mergeability: list the repositories to rescan")
291		-	s.retry(retryKey{kind: retryRescanAll})
	266	+	s.retry(globalKey)
292	267		return
293	268		}
294		-	s.retryDone(retryKey{kind: retryRescanAll})
	269	+	s.retryDone(globalKey)
295	270		for _, r := range repos {
296	271		n, err := s.cfg.Store.CountOpen(s.ctx, r.Name)
297	272		if err != nil {

@@ -372,7 +347,9 @@ func (s *Mergeability) loop() {
372	347		key, dirty := s.q.next()
373	348		switch {
374	349		case key != nil:
375		-	s.runKey(*key)
	350	+	if !s.runKey(*key) {
	351	+	s.retry(repoKey(key.repo))
	352	+	}
376	353		case dirty != nil:
377	354		s.rescan(*dirty)
378	355		default:

@@ -389,48 +366,52 @@ func (s *Mergeability) loop() {
389	366		}
390	367
391	368		// runKey checks k, which is marked running, as often as it was queued again
392		-	// meanwhile.
393		-	func (s *Mergeability) runKey(k mergeKey) {
	369	+	// meanwhile. It reports false when a check failed transiently; the caller
	370	+	// then retries the repository.
	371	+	func (s *Mergeability) runKey(k mergeKey) bool {
	372	+	ok := true
394	373		for {
395		-	rk := retryKey{kind: retryCheck, mk: k}
396		-	if s.check(k) {
397		-	s.retryDone(rk)
398		-	} else {
399		-	s.retry(rk)
400		-	}
	374	+	ok = s.check(k) && ok
401	375		if !s.q.finish(k) \|\| s.ctx.Err() != nil {
402		-	return
	376	+	return ok
403	377		}
404	378		}
405	379		}
406	380
407	381		// rescan checks every open merge request of a repository, here and not through
408	382		// the bounded queue, so that a rescan never marks its repository dirty again.
	383	+	// It succeeds, and resets the retry budget of the repository, only if the
	384	+	// listing and every check succeeded; otherwise the repository is retried once.
409	385		func (s *Mergeability) rescan(repo domain.RepoName) {
410	386		mrs, err := s.cfg.Store.ListOpen(s.ctx, repo)
411	387		if err != nil {
412	388		ctxLog(s.ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: list the open merge requests to rescan")
413		-	s.retry(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})
	389	+	s.retry(repoKey(repo))
414	390		return
415	391		}
416		-	s.retryDone(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})
	392	+	ok := true
417	393		for _, mr := range mrs {
418	394		if s.ctx.Err() != nil {
419	395		return
420	396		}
421	397		k := mergeKey{repo, mr.Number}
422	398		if s.q.tryRun(k) {
423		-	s.runKey(k)
	399	+	ok = s.runKey(k) && ok
424	400		}
425	401		}
	402	+	if ok {
	403	+	s.retryDone(repoKey(repo))
	404	+	} else {
	405	+	s.retry(repoKey(repo))
	406	+	}
426	407		}
427	408
428	409		// Check computes the mergeability of a merge request now, on the calling
429	410		// goroutine, as a worker does. It is for tests; the server never calls it.
430	411		func (s *Mergeability) Check(repo domain.RepoName, number int64) {
431	412		k := mergeKey{repo, number}
432		-	if s.q.tryRun(k) {
433		-	s.runKey(k)
	413	+	if s.q.tryRun(k) && !s.runKey(k) {
	414	+	s.retry(repoKey(k.repo))
434	415		}
435	416		}
436	417


@@ -29,20 +29,18 @@ const (
29		mergeabilityRetryMax = time.Minute	29		mergeabilityRetryMax = time.Minute
30		)	30		)
31			31
32	-	// retryKind tells what a scheduled retry redoes.	32	+	// retryKey names the work a scheduled retry redoes: the rescan of one
33	-	type retryKind int	33	+	// repository, or (all) the rescan of every repository. Retry state is thus
34	-		34	+	// bounded by the number of repositories plus one.
35	-	const (
36	-	retryCheck retryKind = iota // one merge request
37	-	retryRescan // one repository
38	-	retryRescanAll
39	-	)
40	-
41		type retryKey struct {	35		type retryKey struct {
42	-	kind retryKind	36	+	repo domain.RepoName
43	-	mk mergeKey	37	+	all bool
44		}	38		}
45			39
			40	+	func repoKey(repo domain.RepoName) retryKey { return retryKey{repo: repo} }
			41	+
			42	+	var globalKey = retryKey{all: true}
			43	+
46		// mergeBackoff is the delay before retry number attempt (1-based).	44		// mergeBackoff is the delay before retry number attempt (1-based).
47		func mergeBackoff(attempt int) time.Duration {	45		func mergeBackoff(attempt int) time.Duration {
48		d := mergeabilityRetryInitial	46		d := mergeabilityRetryInitial

@@ -82,13 +80,12 @@ type Mergeability struct {
82		cancel context.CancelFunc	80		cancel context.CancelFunc
83		wg sync.WaitGroup	81		wg sync.WaitGroup
84			82
85	-	retryMu sync.Mutex	83	+	retryMu sync.Mutex
86	-	retries map[retryKey]int // attempts so far of the scheduled or failing work	84	+	retries map[retryKey]int // attempts so far of the scheduled or failing work
87	-	scheduled map[retryKey]bool	85	+	scheduled map[retryKey]bool
88	-	heap []retryItem // sorted by due time	86	+	heap []retryItem // sorted by due time
89	-	checkRetries int // scheduled retries of kind retryCheck	87	+	retryKick chan struct{}
90	-	retryKick chan struct{}	88	+	closed bool
91	-	closed bool
92		}	89		}
93			90
94		// MergeabilityHook is what the merge request use cases need from the	91		// MergeabilityHook is what the merge request use cases need from the

@@ -151,37 +148,21 @@ type retryItem struct {
151			148
152		// retry schedules a retry of key after a bounded exponential backoff. It is	149		// retry schedules a retry of key after a bounded exponential backoff. It is
153		// deduplicated per key, gives up after mergeabilityRetries attempts and stops	150		// deduplicated per key, gives up after mergeabilityRetries attempts and stops
154	-	// on Close. All retries share one scheduler goroutine and at most cfg.Queue	151	+	// on Close. All retries share one scheduler goroutine.
155	-	// per-request retries wait at once; past it the retry of a request becomes
156	-	// the retry of its repository's rescan (one per repository), so memory stays
157	-	// bounded.
158		func (s *Mergeability) retry(key retryKey) {	152		func (s *Mergeability) retry(key retryKey) {
159		s.retryMu.Lock()	153		s.retryMu.Lock()
160		defer s.retryMu.Unlock()	154		defer s.retryMu.Unlock()
161		if s.closed \|\| s.scheduled[key] {	155		if s.closed \|\| s.scheduled[key] {
162		return	156		return
163		}	157		}
164	-	// The key that waits is chosen before counting attempts, so that only	158	+	s.retries[key]++
165	-	// the waiting keys hold attempt counts: at most cfg.Queue requests plus	159	+	n := s.retries[key]
166	-	// one per repository.
167	-	sk := key
168	-	if key.kind == retryCheck && s.retries[key] == 0 && s.checkRetries >= s.cfg.Queue {
169	-	sk = retryKey{kind: retryRescan, mk: mergeKey{repo: key.mk.repo}}
170	-	if s.scheduled[sk] {
171	-	return
172	-	}
173	-	}
174	-	s.retries[sk]++
175	-	n := s.retries[sk]
176		if n > mergeabilityRetries {	160		if n > mergeabilityRetries {
177	-	delete(s.retries, sk)	161	+	delete(s.retries, key)
178		return	162		return
179		}	163		}
180	-	if sk.kind == retryCheck {	164	+	s.scheduled[key] = true
181	-	s.checkRetries++	165	+	it := retryItem{due: s.cfg.Clock.Now().Add(mergeBackoff(n)), key: key}
182	-	}
183	-	s.scheduled[sk] = true
184	-	it := retryItem{due: s.cfg.Clock.Now().Add(mergeBackoff(n)), key: sk}
185		i, _ := slices.BinarySearchFunc(s.heap, it, func(a, b retryItem) int {	166		i, _ := slices.BinarySearchFunc(s.heap, it, func(a, b retryItem) int {
186		if a.due.After(b.due) {	167		if a.due.After(b.due) {
187		return 1	168		return 1

@@ -256,9 +237,6 @@ func (s *Mergeability) popDue(waited retryKey) []retryKey {
256		}	237		}
257		out = append(out, it.key)	238		out = append(out, it.key)
258		delete(s.scheduled, it.key)	239		delete(s.scheduled, it.key)
259	-	if it.key.kind == retryCheck {
260	-	s.checkRetries--
261	-	}
262		return true	240		return true
263		})	241		})
264		return out	242		return out

@@ -267,14 +245,11 @@ func (s *Mergeability) popDue(waited retryKey) []retryKey {
267		// fire puts the work of a due retry back in the normal queue, so that the	245		// fire puts the work of a due retry back in the normal queue, so that the
268		// worker limits apply.	246		// worker limits apply.
269		func (s *Mergeability) fire(k retryKey) {	247		func (s *Mergeability) fire(k retryKey) {
270	-	switch k.kind {	248	+	if k.all {
271	-	case retryCheck:
272	-	s.q.add(k.mk)
273	-	case retryRescan:
274	-	s.q.markDirty(k.mk.repo)
275	-	case retryRescanAll:
276		s.rescanAll()	249		s.rescanAll()
			250	+	return
277		}	251		}
			252	+	s.q.markDirty(k.repo)
278		}	253		}
279			254
280		// retryDone forgets the failures of key after a success.	255		// retryDone forgets the failures of key after a success.

@@ -288,10 +263,10 @@ func (s *Mergeability) rescanAll() {
288		repos, err := s.cfg.Meta.List(s.ctx)	263		repos, err := s.cfg.Meta.List(s.ctx)
289		if err != nil {	264		if err != nil {
290		ctxLog(s.ctx).Error().Err(err).Msg("mergeability: list the repositories to rescan")	265		ctxLog(s.ctx).Error().Err(err).Msg("mergeability: list the repositories to rescan")
291	-	s.retry(retryKey{kind: retryRescanAll})	266	+	s.retry(globalKey)
292		return	267		return
293		}	268		}
294	-	s.retryDone(retryKey{kind: retryRescanAll})	269	+	s.retryDone(globalKey)
295		for _, r := range repos {	270		for _, r := range repos {
296		n, err := s.cfg.Store.CountOpen(s.ctx, r.Name)	271		n, err := s.cfg.Store.CountOpen(s.ctx, r.Name)
297		if err != nil {	272		if err != nil {

@@ -372,7 +347,9 @@ func (s *Mergeability) loop() {
372		key, dirty := s.q.next()	347		key, dirty := s.q.next()
373		switch {	348		switch {
374		case key != nil:	349		case key != nil:
375	-	s.runKey(*key)	350	+	if !s.runKey(*key) {
			351	+	s.retry(repoKey(key.repo))
			352	+	}
376		case dirty != nil:	353		case dirty != nil:
377		s.rescan(*dirty)	354		s.rescan(*dirty)
378		default:	355		default:

@@ -389,48 +366,52 @@ func (s *Mergeability) loop() {
389		}	366		}
390			367
391		// runKey checks k, which is marked running, as often as it was queued again	368		// runKey checks k, which is marked running, as often as it was queued again
392	-	// meanwhile.	369	+	// meanwhile. It reports false when a check failed transiently; the caller
393	-	func (s *Mergeability) runKey(k mergeKey) {	370	+	// then retries the repository.
			371	+	func (s *Mergeability) runKey(k mergeKey) bool {
			372	+	ok := true
394		for {	373		for {
395	-	rk := retryKey{kind: retryCheck, mk: k}	374	+	ok = s.check(k) && ok
396	-	if s.check(k) {
397	-	s.retryDone(rk)
398	-	} else {
399	-	s.retry(rk)
400	-	}
401		if !s.q.finish(k) \|\| s.ctx.Err() != nil {	375		if !s.q.finish(k) \|\| s.ctx.Err() != nil {
402	-	return	376	+	return ok
403		}	377		}
404		}	378		}
405		}	379		}
406			380
407		// rescan checks every open merge request of a repository, here and not through	381		// rescan checks every open merge request of a repository, here and not through
408		// the bounded queue, so that a rescan never marks its repository dirty again.	382		// the bounded queue, so that a rescan never marks its repository dirty again.
			383	+	// It succeeds, and resets the retry budget of the repository, only if the
			384	+	// listing and every check succeeded; otherwise the repository is retried once.
409		func (s *Mergeability) rescan(repo domain.RepoName) {	385		func (s *Mergeability) rescan(repo domain.RepoName) {
410		mrs, err := s.cfg.Store.ListOpen(s.ctx, repo)	386		mrs, err := s.cfg.Store.ListOpen(s.ctx, repo)
411		if err != nil {	387		if err != nil {
412		ctxLog(s.ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: list the open merge requests to rescan")	388		ctxLog(s.ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: list the open merge requests to rescan")
413	-	s.retry(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})	389	+	s.retry(repoKey(repo))
414		return	390		return
415		}	391		}
416	-	s.retryDone(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})	392	+	ok := true
417		for _, mr := range mrs {	393		for _, mr := range mrs {
418		if s.ctx.Err() != nil {	394		if s.ctx.Err() != nil {
419		return	395		return
420		}	396		}
421		k := mergeKey{repo, mr.Number}	397		k := mergeKey{repo, mr.Number}
422		if s.q.tryRun(k) {	398		if s.q.tryRun(k) {
423	-	s.runKey(k)	399	+	ok = s.runKey(k) && ok
424		}	400		}
425		}	401		}
			402	+	if ok {
			403	+	s.retryDone(repoKey(repo))
			404	+	} else {
			405	+	s.retry(repoKey(repo))
			406	+	}
426		}	407		}
427			408
428		// Check computes the mergeability of a merge request now, on the calling	409		// Check computes the mergeability of a merge request now, on the calling
429		// goroutine, as a worker does. It is for tests; the server never calls it.	410		// goroutine, as a worker does. It is for tests; the server never calls it.
430		func (s *Mergeability) Check(repo domain.RepoName, number int64) {	411		func (s *Mergeability) Check(repo domain.RepoName, number int64) {
431		k := mergeKey{repo, number}	412		k := mergeKey{repo, number}
432	-	if s.q.tryRun(k) {	413	+	if s.q.tryRun(k) && !s.runKey(k) {
433	-	s.runKey(k)	414	+	s.retry(repoKey(k.repo))
434		}	415		}
435		}	416		}
436			417

M internal/usecase/mergeability_test.go +85 -34


@@ -464,7 +464,8 @@ func TestMergeabilityBoundsConflictPathsAndBehind(t *testing.T) {
464	464		})
465	465		}
466	466
467		-	// A transient Git failure is retried once after the backoff.
	467	+	// A transient Git failure is retried once after the backoff, by rescanning
	468	+	// the repository through the queue.
468	469		func TestMergeabilityRetriesATransientFailureWithBackoff(t *testing.T) {
469	470		repo := name(t, "proj")
470	471		head, tip := fakeOID(t, "a"), fakeOID(t, "b")

@@ -473,6 +474,7 @@ func TestMergeabilityRetriesATransientFailureWithBackoff(t *testing.T) {
473	474		store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)
474	475		store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
475	476		store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
	477	+	store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{mr}, nil).Once()
476	478		meta.EXPECT().List(mock.Anything).Return(nil, nil)
477	479		meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
478	480		var fails atomic.Int32

@@ -555,15 +557,15 @@ func TestMergeabilityWorkersOutliveTheParentContext(t *testing.T) {
555	557		svc.Close(context.Background())
556	558		}
557	559
558		-	// A global failure schedules a bounded number of retries on a single
559		-	// scheduler goroutine, however many requests fail.
	560	+	// Failures of many requests of 3 repositories schedule at most one retry per
	561	+	// repository on a single scheduler goroutine.
560	562		func TestMergeabilityRetriesAreBounded(t *testing.T) {
561		-	const queue, repos, perRepo = 4, 3, 20
	563	+	const repos, perRepo = 3, 20
562	564		store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
563	565		store.EXPECT().Get(mock.Anything, mock.Anything, mock.Anything).Return(domain.MergeRequest{}, errors.New("db down"))
564	566		meta.EXPECT().List(mock.Anything).Return(nil, nil)
565	567		svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
566		-	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: queue,
	568	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: 4,
567	569		After: func(time.Duration) <-chan time.Time { return make(chan time.Time) },
568	570		})
569	571		svc.Start()

@@ -574,46 +576,94 @@ func TestMergeabilityRetriesAreBounded(t *testing.T) {
574	576		svc.Check(repo, int64(n+1))
575	577		}
576	578		}
577		-	assert.LessOrEqual(t, svc.ScheduledRetries(), queue+repos)
578	579		assert.Positive(t, svc.ScheduledRetries())
579		-	assert.LessOrEqual(t, svc.RetryAttempts(), queue+repos, "only the waiting keys count their attempts")
	580	+	assert.LessOrEqual(t, svc.ScheduledRetries(), repos+1)
	581	+	assert.LessOrEqual(t, svc.RetryAttempts(), repos+1)
580	582		assert.LessOrEqual(t, runtime.NumGoroutine(), before, "retries must not start goroutines")
581	583		svc.Close(context.Background()) // returns although retries are pending
582	584		}
583	585
584		-	// A request whose retry was coalesced into a repository rescan is checked
585		-	// again through the queue when the rescan is due.
586		-	func TestMergeabilityDueRetriesReenterTheQueue(t *testing.T) {
587		-	repo := name(t, "proj")
	586	+	// retryRig is a service over mocks whose only repository holds request !7,
	587	+	// whose read fails for its first failFor calls. Each repository rescan signals passes.
	588	+	type retryRig struct {
	589	+	svc *usecase.Mergeability
	590	+	repo domain.RepoName
	591	+	reads atomic.Int32
	592	+	failFor atomic.Int32
	593	+	passes chan struct{}
	594	+	afters chan chan time.Time
	595	+	}
	596	+
	597	+	func newRetryRig(t testing.T) retryRig {
	598	+	t.Helper()
	599	+	r := &retryRig{repo: name(t, "proj"), passes: make(chan struct{}, 32), afters: make(chan chan time.Time, 32)}
588	600		store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
589		-	var calls [9]atomic.Int32
590		-	done := make(chan struct{})
591		-	store.EXPECT().Get(mock.Anything, repo, mock.Anything).RunAndReturn(
592		-	func(_ context.Context, _ domain.RepoName, n int64) (domain.MergeRequest, error) {
593		-	if calls[n].Add(1) == 1 {
	601	+	r.failFor.Store(1 << 30)
	602	+	store.EXPECT().Get(mock.Anything, r.repo, int64(7)).RunAndReturn(
	603	+	func(context.Context, domain.RepoName, int64) (domain.MergeRequest, error) {
	604	+	if r.reads.Add(1) <= r.failFor.Load() {
594	605		return domain.MergeRequest{}, errors.New("db down")
595	606		}
596		-	if n == 8 {
597		-	close(done)
598		-	}
599	607		return domain.MergeRequest{}, domain.ErrNotFound
600	608		})
601		-	store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{{Repo: repo, Number: 8}}, nil)
602		-	meta.EXPECT().List(mock.Anything).Return(nil, nil)
603		-	afters := make(chan chan time.Time, 4)
604		-	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
605		-	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: 1,
606		-	After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); afters <- ch; return ch },
	609	+	store.EXPECT().ListOpen(mock.Anything, r.repo).RunAndReturn(
	610	+	func(context.Context, domain.RepoName) ([]domain.MergeRequest, error) {
	611	+	r.passes <- struct{}{}
	612	+	return []domain.MergeRequest{{Repo: r.repo, Number: 7}}, nil
	613	+	})
	614	+	store.EXPECT().CountOpen(mock.Anything, r.repo).Return(1, nil)
	615	+	meta.EXPECT().List(mock.Anything).Return([]domain.Repo{{Name: r.repo}}, nil)
	616	+	r.svc = usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
	617	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1,
	618	+	After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); r.afters <- ch; return ch },
607	619		})
608		-	t.Cleanup(func() { svc.Close(context.Background()) })
609		-	svc.Start()
610		-	svc.Check(repo, 7) // scheduled itself
611		-	svc.Check(repo, 8) // queue full: coalesced into a rescan
612		-	assert.Equal(t, 2, svc.ScheduledRetries())
613		-	stop := fireAll(afters) // !7 is due and queued again, then the rescan checks !8
614		-	defer stop()
615		-	<-done
616		-	assert.EqualValues(t, 2, calls[7].Load())
	620	+	t.Cleanup(func() { r.svc.Close(context.Background()) })
	621	+	return r
	622	+	}
	623	+
	624	+	// next fires the pending retry and waits for the rescan it queues.
	625	+	func (r *retryRig) next() {
	626	+	ch := <-r.afters
	627	+	ch <- time.Time{}
	628	+	<-r.passes
	629	+	}
	630	+
	631	+	func (r retryRig) settled(t testing.T) {
	632	+	t.Helper()
	633	+	require.Eventually(t, func() bool { return r.svc.RetryAttempts() == 0 && r.svc.ScheduledRetries() == 0 },
	634	+	10*time.Second, time.Millisecond)
	635	+	}
	636	+
	637	+	// A persistent failure is retried 5 times, then given up with no state kept.
	638	+	func TestMergeabilityGivesUpAfterFiveRetries(t *testing.T) {
	639	+	r := newRetryRig(t)
	640	+	r.svc.Start()
	641	+	<-r.passes // the startup rescan fails
	642	+	for range 5 {
	643	+	r.next()
	644	+	}
	645	+	r.settled(t)
	646	+	assert.Empty(t, r.afters, "no sixth retry")
	647	+	assert.Empty(t, r.passes)
	648	+	}
	649	+
	650	+	// A successful full rescan resets the budget: a later failure gets 5 again.
	651	+	func TestMergeabilityFullRescanResetsTheRetryBudget(t *testing.T) {
	652	+	r := newRetryRig(t)
	653	+	r.failFor.Store(2) // the first two reads fail, the third succeeds
	654	+	r.svc.Start()
	655	+	<-r.passes
	656	+	r.next() // second failing pass
	657	+	r.next() // success
	658	+	r.settled(t)
	659	+	r.failFor.Store(1 << 30)
	660	+	r.svc.Check(r.repo, 7) // fails alone: retries the repository
	661	+	for range 5 {
	662	+	r.next()
	663	+	}
	664	+	r.settled(t)
	665	+	assert.Empty(t, r.afters)
	666	+	assert.Empty(t, r.passes)
617	667		}
618	668
619	669		// A transient failure to publish after the store is retried and the event is

@@ -634,6 +684,7 @@ func TestMergeabilityRetriesAFailedPublish(t *testing.T) {
634	684		return rec, nil
635	685		})
636	686		store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil).Once()
	687	+	store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{mr}, nil)
637	688		meta.EXPECT().List(mock.Anything).Return(nil, nil)
638	689		var metaGets atomic.Int32
639	690		meta.EXPECT().Get(mock.Anything, repo).RunAndReturn(


@@ -464,7 +464,8 @@ func TestMergeabilityBoundsConflictPathsAndBehind(t *testing.T) {
464		})	464		})
465		}	465		}
466			466
467	-	// A transient Git failure is retried once after the backoff.	467	+	// A transient Git failure is retried once after the backoff, by rescanning
			468	+	// the repository through the queue.
468		func TestMergeabilityRetriesATransientFailureWithBackoff(t *testing.T) {	469		func TestMergeabilityRetriesATransientFailureWithBackoff(t *testing.T) {
469		repo := name(t, "proj")	470		repo := name(t, "proj")
470		head, tip := fakeOID(t, "a"), fakeOID(t, "b")	471		head, tip := fakeOID(t, "a"), fakeOID(t, "b")

@@ -473,6 +474,7 @@ func TestMergeabilityRetriesATransientFailureWithBackoff(t *testing.T) {
473		store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)	474		store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)
474		store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)	475		store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
475		store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)	476		store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
			477	+	store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{mr}, nil).Once()
476		meta.EXPECT().List(mock.Anything).Return(nil, nil)	478		meta.EXPECT().List(mock.Anything).Return(nil, nil)
477		meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)	479		meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
478		var fails atomic.Int32	480		var fails atomic.Int32

@@ -555,15 +557,15 @@ func TestMergeabilityWorkersOutliveTheParentContext(t *testing.T) {
555		svc.Close(context.Background())	557		svc.Close(context.Background())
556		}	558		}
557			559
558	-	// A global failure schedules a bounded number of retries on a single	560	+	// Failures of many requests of 3 repositories schedule at most one retry per
559	-	// scheduler goroutine, however many requests fail.	561	+	// repository on a single scheduler goroutine.
560		func TestMergeabilityRetriesAreBounded(t *testing.T) {	562		func TestMergeabilityRetriesAreBounded(t *testing.T) {
561	-	const queue, repos, perRepo = 4, 3, 20	563	+	const repos, perRepo = 3, 20
562		store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)	564		store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
563		store.EXPECT().Get(mock.Anything, mock.Anything, mock.Anything).Return(domain.MergeRequest{}, errors.New("db down"))	565		store.EXPECT().Get(mock.Anything, mock.Anything, mock.Anything).Return(domain.MergeRequest{}, errors.New("db down"))
564		meta.EXPECT().List(mock.Anything).Return(nil, nil)	566		meta.EXPECT().List(mock.Anything).Return(nil, nil)
565		svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{	567		svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
566	-	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: queue,	568	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: 4,
567		After: func(time.Duration) <-chan time.Time { return make(chan time.Time) },	569		After: func(time.Duration) <-chan time.Time { return make(chan time.Time) },
568		})	570		})
569		svc.Start()	571		svc.Start()

@@ -574,46 +576,94 @@ func TestMergeabilityRetriesAreBounded(t *testing.T) {
574		svc.Check(repo, int64(n+1))	576		svc.Check(repo, int64(n+1))
575		}	577		}
576		}	578		}
577	-	assert.LessOrEqual(t, svc.ScheduledRetries(), queue+repos)
578		assert.Positive(t, svc.ScheduledRetries())	579		assert.Positive(t, svc.ScheduledRetries())
579	-	assert.LessOrEqual(t, svc.RetryAttempts(), queue+repos, "only the waiting keys count their attempts")	580	+	assert.LessOrEqual(t, svc.ScheduledRetries(), repos+1)
			581	+	assert.LessOrEqual(t, svc.RetryAttempts(), repos+1)
580		assert.LessOrEqual(t, runtime.NumGoroutine(), before, "retries must not start goroutines")	582		assert.LessOrEqual(t, runtime.NumGoroutine(), before, "retries must not start goroutines")
581		svc.Close(context.Background()) // returns although retries are pending	583		svc.Close(context.Background()) // returns although retries are pending
582		}	584		}
583			585
584	-	// A request whose retry was coalesced into a repository rescan is checked	586	+	// retryRig is a service over mocks whose only repository holds request !7,
585	-	// again through the queue when the rescan is due.	587	+	// whose read fails for its first failFor calls. Each repository rescan signals passes.
586	-	func TestMergeabilityDueRetriesReenterTheQueue(t *testing.T) {	588	+	type retryRig struct {
587	-	repo := name(t, "proj")	589	+	svc *usecase.Mergeability
			590	+	repo domain.RepoName
			591	+	reads atomic.Int32
			592	+	failFor atomic.Int32
			593	+	passes chan struct{}
			594	+	afters chan chan time.Time
			595	+	}
			596	+
			597	+	func newRetryRig(t testing.T) retryRig {
			598	+	t.Helper()
			599	+	r := &retryRig{repo: name(t, "proj"), passes: make(chan struct{}, 32), afters: make(chan chan time.Time, 32)}
588		store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)	600		store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
589	-	var calls [9]atomic.Int32	601	+	r.failFor.Store(1 << 30)
590	-	done := make(chan struct{})	602	+	store.EXPECT().Get(mock.Anything, r.repo, int64(7)).RunAndReturn(
591	-	store.EXPECT().Get(mock.Anything, repo, mock.Anything).RunAndReturn(	603	+	func(context.Context, domain.RepoName, int64) (domain.MergeRequest, error) {
592	-	func(_ context.Context, _ domain.RepoName, n int64) (domain.MergeRequest, error) {	604	+	if r.reads.Add(1) <= r.failFor.Load() {
593	-	if calls[n].Add(1) == 1 {
594		return domain.MergeRequest{}, errors.New("db down")	605		return domain.MergeRequest{}, errors.New("db down")
595		}	606		}
596	-	if n == 8 {
597	-	close(done)
598	-	}
599		return domain.MergeRequest{}, domain.ErrNotFound	607		return domain.MergeRequest{}, domain.ErrNotFound
600		})	608		})
601	-	store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{{Repo: repo, Number: 8}}, nil)	609	+	store.EXPECT().ListOpen(mock.Anything, r.repo).RunAndReturn(
602	-	meta.EXPECT().List(mock.Anything).Return(nil, nil)	610	+	func(context.Context, domain.RepoName) ([]domain.MergeRequest, error) {
603	-	afters := make(chan chan time.Time, 4)	611	+	r.passes <- struct{}{}
604	-	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{	612	+	return []domain.MergeRequest{{Repo: r.repo, Number: 7}}, nil
605	-	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: 1,	613	+	})
606	-	After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); afters <- ch; return ch },	614	+	store.EXPECT().CountOpen(mock.Anything, r.repo).Return(1, nil)
			615	+	meta.EXPECT().List(mock.Anything).Return([]domain.Repo{{Name: r.repo}}, nil)
			616	+	r.svc = usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
			617	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1,
			618	+	After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); r.afters <- ch; return ch },
607		})	619		})
608	-	t.Cleanup(func() { svc.Close(context.Background()) })	620	+	t.Cleanup(func() { r.svc.Close(context.Background()) })
609	-	svc.Start()	621	+	return r
610	-	svc.Check(repo, 7) // scheduled itself	622	+	}
611	-	svc.Check(repo, 8) // queue full: coalesced into a rescan	623	+
612	-	assert.Equal(t, 2, svc.ScheduledRetries())	624	+	// next fires the pending retry and waits for the rescan it queues.
613	-	stop := fireAll(afters) // !7 is due and queued again, then the rescan checks !8	625	+	func (r *retryRig) next() {
614	-	defer stop()	626	+	ch := <-r.afters
615	-	<-done	627	+	ch <- time.Time{}
616	-	assert.EqualValues(t, 2, calls[7].Load())	628	+	<-r.passes
			629	+	}
			630	+
			631	+	func (r retryRig) settled(t testing.T) {
			632	+	t.Helper()
			633	+	require.Eventually(t, func() bool { return r.svc.RetryAttempts() == 0 && r.svc.ScheduledRetries() == 0 },
			634	+	10*time.Second, time.Millisecond)
			635	+	}
			636	+
			637	+	// A persistent failure is retried 5 times, then given up with no state kept.
			638	+	func TestMergeabilityGivesUpAfterFiveRetries(t *testing.T) {
			639	+	r := newRetryRig(t)
			640	+	r.svc.Start()
			641	+	<-r.passes // the startup rescan fails
			642	+	for range 5 {
			643	+	r.next()
			644	+	}
			645	+	r.settled(t)
			646	+	assert.Empty(t, r.afters, "no sixth retry")
			647	+	assert.Empty(t, r.passes)
			648	+	}
			649	+
			650	+	// A successful full rescan resets the budget: a later failure gets 5 again.
			651	+	func TestMergeabilityFullRescanResetsTheRetryBudget(t *testing.T) {
			652	+	r := newRetryRig(t)
			653	+	r.failFor.Store(2) // the first two reads fail, the third succeeds
			654	+	r.svc.Start()
			655	+	<-r.passes
			656	+	r.next() // second failing pass
			657	+	r.next() // success
			658	+	r.settled(t)
			659	+	r.failFor.Store(1 << 30)
			660	+	r.svc.Check(r.repo, 7) // fails alone: retries the repository
			661	+	for range 5 {
			662	+	r.next()
			663	+	}
			664	+	r.settled(t)
			665	+	assert.Empty(t, r.afters)
			666	+	assert.Empty(t, r.passes)
617		}	667		}
618			668
619		// A transient failure to publish after the store is retried and the event is	669		// A transient failure to publish after the store is retried and the event is

@@ -634,6 +684,7 @@ func TestMergeabilityRetriesAFailedPublish(t *testing.T) {
634		return rec, nil	684		return rec, nil
635		})	685		})
636		store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil).Once()	686		store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil).Once()
			687	+	store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{mr}, nil)
637		meta.EXPECT().List(mock.Anything).Return(nil, nil)	688		meta.EXPECT().List(mock.Anything).Return(nil, nil)
638		var metaGets atomic.Int32	689		var metaGets atomic.Int32
639		meta.EXPECT().Get(mock.Anything, repo).RunAndReturn(	690		meta.EXPECT().Get(mock.Anything, repo).RunAndReturn(