fix(mr): bound mergeability retries with one scheduler and retry failed publishes

commit: 0efd0e8529b85091d1737bd8df940070992982ea
author: brice · 3d ago · 2026-10-19 15:10 UTC
parent: bc99f06
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +275 -29

M internal/usecase/export_test.go +7 -0


@@ -165,3 +165,10 @@ const MaxSpent = maxSpent
165	165		// SetAfterMergeGate makes the post-merge goroutine of m run f before anything
166	166		// else, to place a test between a merge and its side effects.
167	167		func (m *MergeRequests) SetAfterMergeGate(f func()) { m.afterGate = f }
	168	+
	169	+	// ScheduledRetries gives the number of retries waiting in the scheduler.
	170	+	func (s *Mergeability) ScheduledRetries() int {
	171	+	s.retryMu.Lock()
	172	+	defer s.retryMu.Unlock()
	173	+	return len(s.heap)
	174	+	}


@@ -165,3 +165,10 @@ const MaxSpent = maxSpent
165	// SetAfterMergeGate makes the post-merge goroutine of m run f before anything	165		// SetAfterMergeGate makes the post-merge goroutine of m run f before anything
166	// else, to place a test between a merge and its side effects.	166		// else, to place a test between a merge and its side effects.
167	func (m *MergeRequests) SetAfterMergeGate(f func()) { m.afterGate = f }	167		func (m *MergeRequests) SetAfterMergeGate(f func()) { m.afterGate = f }
		168	+
		169	+	// ScheduledRetries gives the number of retries waiting in the scheduler.
		170	+	func (s *Mergeability) ScheduledRetries() int {
		171	+	s.retryMu.Lock()
		172	+	defer s.retryMu.Unlock()
		173	+	return len(s.heap)
		174	+	}

M internal/usecase/mergeability.go +131 -29


@@ -82,10 +82,13 @@ type Mergeability struct {
82	82		cancel context.CancelFunc
83	83		wg sync.WaitGroup
84	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
	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
89	92		}
90	93
91	94		// MergeabilityHook is what the merge request use cases need from the

@@ -115,13 +118,15 @@ func NewMergeability(ctx context.Context, cfg MergeabilityConfig) *Mergeability
115	118		wctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
116	119		return &Mergeability{
117	120		cfg: cfg, q: newMergeQueue(cfg.Queue), ctx: wctx, cancel: cancel,
118		-	retries: map[retryKey]int{}, retrying: map[retryKey]bool{},
	121	+	retries: map[retryKey]int{}, scheduled: map[retryKey]bool{}, retryKick: make(chan struct{}, 1),
119	122		}
120	123		}
121	124
122	125		// Start rescans the open merge requests of every repository (the records may
123	126		// be missing or stale after a restart) and starts the workers.
124	127		func (s *Mergeability) Start() {
	128	+	s.wg.Add(1)
	129	+	go s.scheduler()
125	130		s.rescanAll()
126	131		for range s.cfg.Workers {
127	132		s.wg.Add(1)

@@ -138,13 +143,22 @@ func (s *Mergeability) Close(_ context.Context) {
138	143		s.wg.Wait()
139	144		}
140	145
141		-	// retry schedules redo after a bounded exponential backoff. It is
	146	+	// retryItem is one scheduled retry.
	147	+	type retryItem struct {
	148	+	due time.Time
	149	+	key retryKey
	150	+	}
	151	+
	152	+	// retry schedules a retry of key after a bounded exponential backoff. It is
142	153		// deduplicated per key, gives up after mergeabilityRetries attempts and stops
143		-	// on Close.
144		-	func (s *Mergeability) retry(key retryKey, redo func()) {
	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.
	158	+	func (s *Mergeability) retry(key retryKey) {
145	159		s.retryMu.Lock()
146	160		defer s.retryMu.Unlock()
147		-	if s.closed \|\| s.retrying[key] {
	161	+	if s.closed \|\| s.scheduled[key] {
148	162		return
149	163		}
150	164		s.retries[key]++

@@ -153,21 +167,101 @@ func (s *Mergeability) retry(key retryKey, redo func()) {
153	167		delete(s.retries, key)
154	168		return
155	169		}
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()
	170	+	sk := key
	171	+	if key.kind == retryCheck && s.checkRetries >= s.cfg.Queue {
	172	+	sk = retryKey{kind: retryRescan, mk: mergeKey{repo: key.mk.repo}}
	173	+	if s.scheduled[sk] {
	174	+	return
	175	+	}
	176	+	}
	177	+	if sk.kind == retryCheck {
	178	+	s.checkRetries++
	179	+	}
	180	+	s.scheduled[sk] = true
	181	+	it := retryItem{due: s.cfg.Clock.Now().Add(mergeBackoff(n)), key: sk}
	182	+	i, _ := slices.BinarySearchFunc(s.heap, it, func(a, b retryItem) int {
	183	+	if a.due.After(b.due) {
	184	+	return 1
	185	+	}
	186	+	return -1 // equal dues keep their insertion order
	187	+	})
	188	+	s.heap = slices.Insert(s.heap, i, it)
	189	+	if i == 0 {
	190	+	select {
	191	+	case s.retryKick <- struct{}{}:
	192	+	default:
	193	+	}
	194	+	}
	195	+	}
	196	+
	197	+	// scheduler is the only goroutine that waits for retries: it sleeps until the
	198	+	// earliest is due and puts it back in the normal queue.
	199	+	func (s *Mergeability) scheduler() {
	200	+	defer s.wg.Done()
	201	+	for {
	202	+	s.retryMu.Lock()
	203	+	var head retryItem
	204	+	has := len(s.heap) > 0
	205	+	if has {
	206	+	head = s.heap[0]
	207	+	}
	208	+	s.retryMu.Unlock()
	209	+	if !has {
	210	+	select {
	211	+	case <-s.ctx.Done():
	212	+	return
	213	+	case <-s.retryKick:
	214	+	continue
	215	+	}
	216	+	}
	217	+	after := s.cfg.After(max(0, head.due.Sub(s.cfg.Clock.Now())))
161	218		select {
162	219		case <-s.ctx.Done():
163	220		return
	221	+	case <-s.retryKick:
	222	+	continue // an earlier retry arrived
164	223		case <-after:
165	224		}
166		-	s.retryMu.Lock()
167		-	delete(s.retrying, key)
168		-	s.retryMu.Unlock()
169		-	redo()
170		-	}()
	225	+	for _, k := range s.popDue(head.key) {
	226	+	if s.ctx.Err() != nil {
	227	+	return
	228	+	}
	229	+	s.fire(k)
	230	+	}
	231	+	}
	232	+	}
	233	+
	234	+	// popDue removes the waited item and the ones due as well.
	235	+	func (s *Mergeability) popDue(waited retryKey) []retryKey {
	236	+	s.retryMu.Lock()
	237	+	defer s.retryMu.Unlock()
	238	+	now := s.cfg.Clock.Now()
	239	+	var out []retryKey
	240	+	s.heap = slices.DeleteFunc(s.heap, func(it retryItem) bool {
	241	+	if it.key != waited && it.due.After(now) {
	242	+	return false
	243	+	}
	244	+	out = append(out, it.key)
	245	+	delete(s.scheduled, it.key)
	246	+	if it.key.kind == retryCheck {
	247	+	s.checkRetries--
	248	+	}
	249	+	return true
	250	+	})
	251	+	return out
	252	+	}
	253	+
	254	+	// fire puts the work of a due retry back in the normal queue, so that the
	255	+	// worker limits apply.
	256	+	func (s *Mergeability) fire(k retryKey) {
	257	+	switch k.kind {
	258	+	case retryCheck:
	259	+	s.q.add(k.mk)
	260	+	case retryRescan:
	261	+	s.q.markDirty(k.mk.repo)
	262	+	case retryRescanAll:
	263	+	s.rescanAll()
	264	+	}
171	265		}
172	266
173	267		// retryDone forgets the failures of key after a success.

@@ -181,7 +275,7 @@ func (s *Mergeability) rescanAll() {
181	275		repos, err := s.cfg.Meta.List(s.ctx)
182	276		if err != nil {
183	277		ctxLog(s.ctx).Error().Err(err).Msg("mergeability: list the repositories to rescan")
184		-	s.retry(retryKey{kind: retryRescanAll}, s.rescanAll)
	278	+	s.retry(retryKey{kind: retryRescanAll})
185	279		return
186	280		}
187	281		s.retryDone(retryKey{kind: retryRescanAll})

@@ -289,7 +383,7 @@ func (s *Mergeability) runKey(k mergeKey) {
289	383		if s.check(k) {
290	384		s.retryDone(rk)
291	385		} else {
292		-	s.retry(rk, func() { s.q.add(k) })
	386	+	s.retry(rk)
293	387		}
294	388		if !s.q.finish(k) \|\| s.ctx.Err() != nil {
295	389		return

@@ -303,7 +397,7 @@ func (s *Mergeability) rescan(repo domain.RepoName) {
303	397		mrs, err := s.cfg.Store.ListOpen(s.ctx, repo)
304	398		if err != nil {
305	399		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) })
	400	+	s.retry(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})
307	401		return
308	402		}
309	403		s.retryDone(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})

@@ -362,8 +456,9 @@ func (s *Mergeability) check(k mergeKey) (done bool) {
362	456		return false
363	457		}
364	458		if prev.IsFor(tip, head) {
365		-	if s.q.takeSeen(k) {
366		-	s.publish(ctx, prev) // a reader saw "checking" before this record
	459	+	if s.q.takeSeen(k) && s.publish(ctx, prev) != nil { // a reader saw "checking" before this record
	460	+	s.q.markSeen(k)
	461	+	return false
367	462		}
368	463		return true
369	464		}

@@ -398,7 +493,10 @@ func (s *Mergeability) check(k mergeKey) (done bool) {
398	493		}
399	494		seen := s.q.takeSeen(k)
400	495		if seen \|\| prev.State == "" \|\| changedMergeability(prev, res) {
401		-	s.publish(ctx, res)
	496	+	if s.publish(ctx, res) != nil {
	497	+	s.q.markSeen(k) // the next check publishes it
	498	+	return false
	499	+	}
402	500		}
403	501		return true
404	502		}

@@ -471,16 +569,20 @@ func changedMergeability(a, b domain.Mergeability) bool {
471	569		a.ConflictPathsTruncated != b.ConflictPathsTruncated \|\| !slices.Equal(a.ConflictPaths, b.ConflictPaths)
472	570		}
473	571
474		-	func (s *Mergeability) publish(ctx context.Context, m domain.Mergeability) {
	572	+	// publish sends the result on the bus. It fails on a transient error; a
	573	+	// repository that is gone is terminal and not an error.
	574	+	func (s *Mergeability) publish(ctx context.Context, m domain.Mergeability) error {
475	575		repo, err := s.cfg.Meta.Get(ctx, m.Repo)
476	576		if err != nil {
477		-	if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
478		-	ctxLog(ctx).Error().Err(err).Str("repo", m.Repo.String()).Msg("mergeability: read the repository to publish a result")
	577	+	if errors.Is(err, domain.ErrNotFound) \|\| ctx.Err() != nil {
	578	+	return nil
479	579		}
480		-	return
	580	+	ctxLog(ctx).Error().Err(err).Str("repo", m.Repo.String()).Msg("mergeability: read the repository to publish a result")
	581	+	return err
481	582		}
482	583		s.cfg.Bus.Publish(ctx, domain.MergeRequestMergeability{
483	584		Repo: m.Repo, Number: m.Number, Mergeability: m, At: m.CheckedAt,
484	585		Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
485	586		})
	587	+	return nil
486	588		}

 	cancel context.CancelFunc
 	wg     sync.WaitGroup
-	retryMu  sync.Mutex
+	retryMu      sync.Mutex
-	retries  map[retryKey]int // attempts so far of the scheduled or failing work
+	retries      map[retryKey]int // attempts so far of the scheduled or failing work
-	retrying map[retryKey]bool
+	scheduled    map[retryKey]bool
-	closed   bool
+	heap         []retryItem // sorted by due time
+	checkRetries int         // scheduled retries of kind retryCheck
+	retryKick    chan struct{}
+	closed       bool
 }
 // MergeabilityHook is what the merge request use cases need from the
 	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{},
+		retries: map[retryKey]int{}, scheduled: map[retryKey]bool{}, retryKick: make(chan struct{}, 1),
 	}
 }
 // Start rescans the open merge requests of every repository (the records may
 // be missing or stale after a restart) and starts the workers.
 func (s *Mergeability) Start() {
+	s.wg.Add(1)
+	go s.scheduler()
 	s.rescanAll()
 	for range s.cfg.Workers {
 		s.wg.Add(1)
 	s.wg.Wait()
 }
-// retry schedules redo after a bounded exponential backoff. It is
+// retryItem is one scheduled retry.
+type retryItem struct {
+	due time.Time
+	key retryKey
+}
+// retry schedules a retry of key after a bounded exponential backoff. It is
 // deduplicated per key, gives up after mergeabilityRetries attempts and stops
-// on Close.
+// on Close. All retries share one scheduler goroutine and at most cfg.Queue
-func (s *Mergeability) retry(key retryKey, redo func()) {
+// per-request retries wait at once; past it the retry of a request becomes
+// the retry of its repository's rescan (one per repository), so memory stays
+// bounded.
+func (s *Mergeability) retry(key retryKey) {
 	s.retryMu.Lock()
 	defer s.retryMu.Unlock()
-	if s.closed || s.retrying[key] {
+	if s.closed || s.scheduled[key] {
 		return
 	}
 	s.retries[key]++
 		delete(s.retries, key)
 		return
 	}
-	s.retrying[key] = true
+	sk := key
-	after := s.cfg.After(mergeBackoff(n))
+	if key.kind == retryCheck && s.checkRetries >= s.cfg.Queue {
-	s.wg.Add(1)
+		sk = retryKey{kind: retryRescan, mk: mergeKey{repo: key.mk.repo}}
-	go func() {
+		if s.scheduled[sk] {
-		defer s.wg.Done()
+			return
+		}
+	}
+	if sk.kind == retryCheck {
+		s.checkRetries++
+	}
+	s.scheduled[sk] = true
+	it := retryItem{due: s.cfg.Clock.Now().Add(mergeBackoff(n)), key: sk}
+	i, _ := slices.BinarySearchFunc(s.heap, it, func(a, b retryItem) int {
+		if a.due.After(b.due) {
+			return 1
+		}
+		return -1 // equal dues keep their insertion order
+	})
+	s.heap = slices.Insert(s.heap, i, it)
+	if i == 0 {
+		select {
+		case s.retryKick <- struct{}{}:
+		default:
+		}
+	}
+}
+// scheduler is the only goroutine that waits for retries: it sleeps until the
+// earliest is due and puts it back in the normal queue.
+func (s *Mergeability) scheduler() {
+	defer s.wg.Done()
+	for {
+		s.retryMu.Lock()
+		var head retryItem
+		has := len(s.heap) > 0
+		if has {
+			head = s.heap[0]
+		}
+		s.retryMu.Unlock()
+		if !has {
+			select {
+			case <-s.ctx.Done():
+				return
+			case <-s.retryKick:
+				continue
+			}
+		}
+		after := s.cfg.After(max(0, head.due.Sub(s.cfg.Clock.Now())))
 		select {
 		case <-s.ctx.Done():
 			return
+		case <-s.retryKick:
+			continue // an earlier retry arrived
 		case <-after:
 		}
-		s.retryMu.Lock()
+		for _, k := range s.popDue(head.key) {
-		delete(s.retrying, key)
+			if s.ctx.Err() != nil {
-		s.retryMu.Unlock()
+				return
-		redo()
+			}
-	}()
+			s.fire(k)
+		}
+	}
+}
+// popDue removes the waited item and the ones due as well.
+func (s *Mergeability) popDue(waited retryKey) []retryKey {
+	s.retryMu.Lock()
+	defer s.retryMu.Unlock()
+	now := s.cfg.Clock.Now()
+	var out []retryKey
+	s.heap = slices.DeleteFunc(s.heap, func(it retryItem) bool {
+		if it.key != waited && it.due.After(now) {
+			return false
+		}
+		out = append(out, it.key)
+		delete(s.scheduled, it.key)
+		if it.key.kind == retryCheck {
+			s.checkRetries--
+		}
+		return true
+	})
+	return out
+}
+// fire puts the work of a due retry back in the normal queue, so that the
+// worker limits apply.
+func (s *Mergeability) fire(k retryKey) {
+	switch k.kind {
+	case retryCheck:
+		s.q.add(k.mk)
+	case retryRescan:
+		s.q.markDirty(k.mk.repo)
+	case retryRescanAll:
+		s.rescanAll()
+	}
 }
 // retryDone forgets the failures of key after a success.
 	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)
+		s.retry(retryKey{kind: retryRescanAll})
 		return
 	}
 	s.retryDone(retryKey{kind: retryRescanAll})
 		if s.check(k) {
 			s.retryDone(rk)
 		} else {
-			s.retry(rk, func() { s.q.add(k) })
+			s.retry(rk)
 		}
 		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) })
+		s.retry(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})
 		return
 	}
 	s.retryDone(retryKey{kind: retryRescan, mk: mergeKey{repo: repo}})
 			return false
 		}
 		if prev.IsFor(tip, head) {
-			if s.q.takeSeen(k) {
+			if s.q.takeSeen(k) && s.publish(ctx, prev) != nil { // a reader saw "checking" before this record
-				s.publish(ctx, prev) // a reader saw "checking" before this record
+				s.q.markSeen(k)
+				return false
 			}
 			return true
 		}
 		}
 		seen := s.q.takeSeen(k)
 		if seen || prev.State == "" || changedMergeability(prev, res) {
-			s.publish(ctx, res)
+			if s.publish(ctx, res) != nil {
+				s.q.markSeen(k) // the next check publishes it
+				return false
+			}
 		}
 		return true
 	}
 		a.ConflictPathsTruncated != b.ConflictPathsTruncated || !slices.Equal(a.ConflictPaths, b.ConflictPaths)
 }
-func (s *Mergeability) publish(ctx context.Context, m domain.Mergeability) {
+// publish sends the result on the bus. It fails on a transient error; a
+// repository that is gone is terminal and not an error.
+func (s *Mergeability) publish(ctx context.Context, m domain.Mergeability) error {
 	repo, err := s.cfg.Meta.Get(ctx, m.Repo)
 	if err != nil {
-		if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
+		if errors.Is(err, domain.ErrNotFound) || ctx.Err() != nil {
-			ctxLog(ctx).Error().Err(err).Str("repo", m.Repo.String()).Msg("mergeability: read the repository to publish a result")
+			return nil
 		}
-		return
+		ctxLog(ctx).Error().Err(err).Str("repo", m.Repo.String()).Msg("mergeability: read the repository to publish a result")
+		return err
 	}
 	s.cfg.Bus.Publish(ctx, domain.MergeRequestMergeability{
 		Repo: m.Repo, Number: m.Number, Mergeability: m, At: m.CheckedAt,
 		Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
 	})
+	return nil
 }

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


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

@@ -552,3 +554,138 @@ func TestMergeabilityWorkersOutliveTheParentContext(t *testing.T) {
552	554		<-published // computed although the parent is cancelled
553	555		svc.Close(context.Background())
554	556		}
	557	+
	558	+	// A global failure schedules a bounded number of retries on a single
	559	+	// scheduler goroutine, however many requests fail.
	560	+	func TestMergeabilityRetriesAreBounded(t *testing.T) {
	561	+	const queue, repos, perRepo = 4, 3, 20
	562	+	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"))
	564	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
	565	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
	566	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: queue,
	567	+	After: func(time.Duration) <-chan time.Time { return make(chan time.Time) },
	568	+	})
	569	+	svc.Start()
	570	+	before := runtime.NumGoroutine()
	571	+	for r := range repos {
	572	+	repo := name(t, "proj"+strconv.Itoa(r))
	573	+	for n := range perRepo {
	574	+	svc.Check(repo, int64(n+1))
	575	+	}
	576	+	}
	577	+	assert.LessOrEqual(t, svc.ScheduledRetries(), queue+repos)
	578	+	assert.Positive(t, svc.ScheduledRetries())
	579	+	assert.LessOrEqual(t, runtime.NumGoroutine(), before, "retries must not start goroutines")
	580	+	svc.Close(context.Background()) // returns although retries are pending
	581	+	}
	582	+
	583	+	// A request whose retry was coalesced into a repository rescan is checked
	584	+	// again through the queue when the rescan is due.
	585	+	func TestMergeabilityDueRetriesReenterTheQueue(t *testing.T) {
	586	+	repo := name(t, "proj")
	587	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
	588	+	var calls [9]atomic.Int32
	589	+	done := make(chan struct{})
	590	+	store.EXPECT().Get(mock.Anything, repo, mock.Anything).RunAndReturn(
	591	+	func(_ context.Context, _ domain.RepoName, n int64) (domain.MergeRequest, error) {
	592	+	if calls[n].Add(1) == 1 {
	593	+	return domain.MergeRequest{}, errors.New("db down")
	594	+	}
	595	+	if n == 8 {
	596	+	close(done)
	597	+	}
	598	+	return domain.MergeRequest{}, domain.ErrNotFound
	599	+	})
	600	+	store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{{Repo: repo, Number: 8}}, nil)
	601	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
	602	+	afters := make(chan chan time.Time, 4)
	603	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
	604	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: 1,
	605	+	After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); afters <- ch; return ch },
	606	+	})
	607	+	t.Cleanup(func() { svc.Close(context.Background()) })
	608	+	svc.Start()
	609	+	svc.Check(repo, 7) // scheduled itself
	610	+	svc.Check(repo, 8) // queue full: coalesced into a rescan
	611	+	assert.Equal(t, 2, svc.ScheduledRetries())
	612	+	stop := fireAll(afters) // !7 is due and queued again, then the rescan checks !8
	613	+	defer stop()
	614	+	<-done
	615	+	assert.EqualValues(t, 2, calls[7].Load())
	616	+	}
	617	+
	618	+	// A transient failure to publish after the store is retried and the event is
	619	+	// published without any other trigger.
	620	+	func TestMergeabilityRetriesAFailedPublish(t *testing.T) {
	621	+	repo := name(t, "proj")
	622	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
	623	+	mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
	624	+	rec := domain.Mergeability{Repo: repo, Number: 7, TargetID: tip, HeadID: head, State: domain.MergeabilityUpToDate}
	625	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
	626	+	store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)
	627	+	var gets atomic.Int32
	628	+	store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).RunAndReturn(
	629	+	func(context.Context, domain.RepoName, int64) (domain.Mergeability, error) {
	630	+	if gets.Add(1) == 1 {
	631	+	return domain.Mergeability{}, domain.ErrNotFound
	632	+	}
	633	+	return rec, nil
	634	+	})
	635	+	store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil).Once()
	636	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
	637	+	var metaGets atomic.Int32
	638	+	meta.EXPECT().Get(mock.Anything, repo).RunAndReturn(
	639	+	func(context.Context, domain.RepoName) (domain.Repo, error) {
	640	+	if metaGets.Add(1) == 1 {
	641	+	return domain.Repo{}, errors.New("transient")
	642	+	}
	643	+	return domain.Repo{Name: repo, CreatedAt: now}, nil
	644	+	})
	645	+	git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
	646	+	git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
	647	+	git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
	648	+	published := make(chan struct{})
	649	+	bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, domain.Event) { close(published) }).Return().Once()
	650	+
	651	+	afters := make(chan chan time.Time, 4)
	652	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
	653	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{},
	654	+	After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); afters <- ch; return ch },
	655	+	})
	656	+	t.Cleanup(func() { svc.Close(context.Background()) })
	657	+	svc.Start()
	658	+	svc.OnOpened(context.Background(), mr)
	659	+	<-svcWaiting(afters)
	660	+	select {
	661	+	case <-published:
	662	+	t.Fatal("published although the repository could not be read")
	663	+	default:
	664	+	}
	665	+	stop := fireAll(afters)
	666	+	defer stop()
	667	+	<-published
	668	+	}
	669	+
	670	+	// svcWaiting blocks until the scheduler asked for a timer, and puts it back.
	671	+	func svcWaiting(afters chan chan time.Time) <-chan struct{} {
	672	+	ready := make(chan struct{})
	673	+	go func() { ch := <-afters; afters <- ch; close(ready) }()
	674	+	return ready
	675	+	}
	676	+
	677	+	// fireAll fires every timer the scheduler asks for (stale ones are harmless).
	678	+	func fireAll(afters chan chan time.Time) (stop func()) {
	679	+	quit := make(chan struct{})
	680	+	go func() {
	681	+	for {
	682	+	select {
	683	+	case ch := <-afters:
	684	+	ch <- time.Time{}
	685	+	case <-quit:
	686	+	return
	687	+	}
	688	+	}
	689	+	}()
	690	+	return func() { close(quit) }
	691	+	}


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

@@ -552,3 +554,138 @@ func TestMergeabilityWorkersOutliveTheParentContext(t *testing.T) {
552	<-published // computed although the parent is cancelled	554		<-published // computed although the parent is cancelled
553	svc.Close(context.Background())	555		svc.Close(context.Background())
554	}	556		}
		557	+
		558	+	// A global failure schedules a bounded number of retries on a single
		559	+	// scheduler goroutine, however many requests fail.
		560	+	func TestMergeabilityRetriesAreBounded(t *testing.T) {
		561	+	const queue, repos, perRepo = 4, 3, 20
		562	+	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"))
		564	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
		565	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
		566	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: queue,
		567	+	After: func(time.Duration) <-chan time.Time { return make(chan time.Time) },
		568	+	})
		569	+	svc.Start()
		570	+	before := runtime.NumGoroutine()
		571	+	for r := range repos {
		572	+	repo := name(t, "proj"+strconv.Itoa(r))
		573	+	for n := range perRepo {
		574	+	svc.Check(repo, int64(n+1))
		575	+	}
		576	+	}
		577	+	assert.LessOrEqual(t, svc.ScheduledRetries(), queue+repos)
		578	+	assert.Positive(t, svc.ScheduledRetries())
		579	+	assert.LessOrEqual(t, runtime.NumGoroutine(), before, "retries must not start goroutines")
		580	+	svc.Close(context.Background()) // returns although retries are pending
		581	+	}
		582	+
		583	+	// A request whose retry was coalesced into a repository rescan is checked
		584	+	// again through the queue when the rescan is due.
		585	+	func TestMergeabilityDueRetriesReenterTheQueue(t *testing.T) {
		586	+	repo := name(t, "proj")
		587	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
		588	+	var calls [9]atomic.Int32
		589	+	done := make(chan struct{})
		590	+	store.EXPECT().Get(mock.Anything, repo, mock.Anything).RunAndReturn(
		591	+	func(_ context.Context, _ domain.RepoName, n int64) (domain.MergeRequest, error) {
		592	+	if calls[n].Add(1) == 1 {
		593	+	return domain.MergeRequest{}, errors.New("db down")
		594	+	}
		595	+	if n == 8 {
		596	+	close(done)
		597	+	}
		598	+	return domain.MergeRequest{}, domain.ErrNotFound
		599	+	})
		600	+	store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{{Repo: repo, Number: 8}}, nil)
		601	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
		602	+	afters := make(chan chan time.Time, 4)
		603	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
		604	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1, Queue: 1,
		605	+	After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); afters <- ch; return ch },
		606	+	})
		607	+	t.Cleanup(func() { svc.Close(context.Background()) })
		608	+	svc.Start()
		609	+	svc.Check(repo, 7) // scheduled itself
		610	+	svc.Check(repo, 8) // queue full: coalesced into a rescan
		611	+	assert.Equal(t, 2, svc.ScheduledRetries())
		612	+	stop := fireAll(afters) // !7 is due and queued again, then the rescan checks !8
		613	+	defer stop()
		614	+	<-done
		615	+	assert.EqualValues(t, 2, calls[7].Load())
		616	+	}
		617	+
		618	+	// A transient failure to publish after the store is retried and the event is
		619	+	// published without any other trigger.
		620	+	func TestMergeabilityRetriesAFailedPublish(t *testing.T) {
		621	+	repo := name(t, "proj")
		622	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
		623	+	mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
		624	+	rec := domain.Mergeability{Repo: repo, Number: 7, TargetID: tip, HeadID: head, State: domain.MergeabilityUpToDate}
		625	+	store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
		626	+	store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(mr, nil)
		627	+	var gets atomic.Int32
		628	+	store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).RunAndReturn(
		629	+	func(context.Context, domain.RepoName, int64) (domain.Mergeability, error) {
		630	+	if gets.Add(1) == 1 {
		631	+	return domain.Mergeability{}, domain.ErrNotFound
		632	+	}
		633	+	return rec, nil
		634	+	})
		635	+	store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil).Once()
		636	+	meta.EXPECT().List(mock.Anything).Return(nil, nil)
		637	+	var metaGets atomic.Int32
		638	+	meta.EXPECT().Get(mock.Anything, repo).RunAndReturn(
		639	+	func(context.Context, domain.RepoName) (domain.Repo, error) {
		640	+	if metaGets.Add(1) == 1 {
		641	+	return domain.Repo{}, errors.New("transient")
		642	+	}
		643	+	return domain.Repo{Name: repo, CreatedAt: now}, nil
		644	+	})
		645	+	git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
		646	+	git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
		647	+	git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
		648	+	published := make(chan struct{})
		649	+	bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, domain.Event) { close(published) }).Return().Once()
		650	+
		651	+	afters := make(chan chan time.Time, 4)
		652	+	svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
		653	+	Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{},
		654	+	After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); afters <- ch; return ch },
		655	+	})
		656	+	t.Cleanup(func() { svc.Close(context.Background()) })
		657	+	svc.Start()
		658	+	svc.OnOpened(context.Background(), mr)
		659	+	<-svcWaiting(afters)
		660	+	select {
		661	+	case <-published:
		662	+	t.Fatal("published although the repository could not be read")
		663	+	default:
		664	+	}
		665	+	stop := fireAll(afters)
		666	+	defer stop()
		667	+	<-published
		668	+	}
		669	+
		670	+	// svcWaiting blocks until the scheduler asked for a timer, and puts it back.
		671	+	func svcWaiting(afters chan chan time.Time) <-chan struct{} {
		672	+	ready := make(chan struct{})
		673	+	go func() { ch := <-afters; afters <- ch; close(ready) }()
		674	+	return ready
		675	+	}
		676	+
		677	+	// fireAll fires every timer the scheduler asks for (stale ones are harmless).
		678	+	func fireAll(afters chan chan time.Time) (stop func()) {
		679	+	quit := make(chan struct{})
		680	+	go func() {
		681	+	for {
		682	+	select {
		683	+	case ch := <-afters:
		684	+	ch <- time.Time{}
		685	+	case <-quit:
		686	+	return
		687	+	}
		688	+	}
		689	+	}()
		690	+	return func() { close(quit) }
		691	+	}