fix(mr): keep replacement retries from stale timers and log reconcile failures

commit: c22da12bdbdd89d5c639829cdcd8b26781f97d73
author: brice · 2d ago · 2026-10-19 16:11 UTC
committer: brice · 2d ago · 2026-10-19 18:30 UTC
parent: 1036507
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +52 -4

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


@@ -193,3 +193,27 @@ func (s *Mergeability) RetryAttempts() int {
193	193
194	194		// ReconcileForTest runs a repository pass synchronously without workers.
195	195		func (s *Mergeability) ReconcileForTest(repo domain.RepoName) bool { return s.reconcile(repo) }
	196	+
	197	+	// RetryForTest schedules a retry of repo as a failed pass does.
	198	+	func (s *Mergeability) RetryForTest(repo domain.RepoName) { s.retry(repo) }
	199	+
	200	+	// RetryDoneForTest forgets the retries of repo as a successful pass does.
	201	+	func (s *Mergeability) RetryDoneForTest(repo domain.RepoName) { s.retryDone(repo) }
	202	+
	203	+	// RetryDueForTest gives the deadline of the retry scheduled for repo.
	204	+	func (s *Mergeability) RetryDueForTest(repo domain.RepoName) (time.Time, bool) {
	205	+	s.retryMu.Lock()
	206	+	defer s.retryMu.Unlock()
	207	+	for _, it := range s.heap {
	208	+	if it.key == repo {
	209	+	return it.due, true
	210	+	}
	211	+	}
	212	+	return time.Time{}, false
	213	+	}
	214	+
	215	+	// PopDueForTest is the scheduler step taken when the timer armed for the
	216	+	// retry (repo, due) fires. It gives the repositories it released.
	217	+	func (s *Mergeability) PopDueForTest(repo domain.RepoName, due time.Time) []domain.RepoName {
	218	+	return s.popDue(retryItem{key: repo, due: due})
	219	+	}


@@ -193,3 +193,27 @@ func (s *Mergeability) RetryAttempts() int {
193		193
194	// ReconcileForTest runs a repository pass synchronously without workers.	194		// ReconcileForTest runs a repository pass synchronously without workers.
195	func (s *Mergeability) ReconcileForTest(repo domain.RepoName) bool { return s.reconcile(repo) }	195		func (s *Mergeability) ReconcileForTest(repo domain.RepoName) bool { return s.reconcile(repo) }
		196	+
		197	+	// RetryForTest schedules a retry of repo as a failed pass does.
		198	+	func (s *Mergeability) RetryForTest(repo domain.RepoName) { s.retry(repo) }
		199	+
		200	+	// RetryDoneForTest forgets the retries of repo as a successful pass does.
		201	+	func (s *Mergeability) RetryDoneForTest(repo domain.RepoName) { s.retryDone(repo) }
		202	+
		203	+	// RetryDueForTest gives the deadline of the retry scheduled for repo.
		204	+	func (s *Mergeability) RetryDueForTest(repo domain.RepoName) (time.Time, bool) {
		205	+	s.retryMu.Lock()
		206	+	defer s.retryMu.Unlock()
		207	+	for _, it := range s.heap {
		208	+	if it.key == repo {
		209	+	return it.due, true
		210	+	}
		211	+	}
		212	+	return time.Time{}, false
		213	+	}
		214	+
		215	+	// PopDueForTest is the scheduler step taken when the timer armed for the
		216	+	// retry (repo, due) fires. It gives the repositories it released.
		217	+	func (s *Mergeability) PopDueForTest(repo domain.RepoName, due time.Time) []domain.RepoName {
		218	+	return s.popDue(retryItem{key: repo, due: due})
		219	+	}

M internal/usecase/mergeability.go +28 -4


@@ -95,6 +95,9 @@ func (s *Mergeability) Start() {
95	95		defer s.wg.Done()
96	96		for attempt := 0; attempt <= mergeabilityRetries; attempt++ {
97	97		repos, err := s.cfg.Meta.List(s.ctx)
	98	+	if err != nil && s.ctx.Err() == nil {
	99	+	ctxLog(s.ctx).Error().Err(err).Int("attempt", attempt+1).Msg("mergeability: list the repositories at startup")
	100	+	}
98	101		if err == nil {
99	102		for _, repo := range repos {
100	103		s.markDirty(repo.Name)

@@ -223,10 +226,12 @@ func (s *Mergeability) reconcile(repo domain.RepoName) bool {
223	226		return true
224	227		}
225	228		if err != nil {
	229	+	s.logFailure(ctx, err, "get the repository", repo, 0)
226	230		return false
227	231		}
228	232		mrs, err := s.cfg.Store.ListOpen(ctx, repo)
229	233		if err != nil {
	234	+	s.logFailure(ctx, err, "list the open merge requests", repo, 0)
230	235		return false
231	236		}
232	237		ok := true

@@ -241,6 +246,7 @@ func (s *Mergeability) reconcile(repo domain.RepoName) bool {
241	246		}
242	247		prev, err := s.cfg.Store.GetMergeability(ctx, repo, mr.Number)
243	248		if err != nil && !errors.Is(err, domain.ErrNotFound) {
	249	+	s.logFailure(ctx, err, "get the mergeability", repo, mr.Number)
244	250		ok = false
245	251		continue
246	252		}

@@ -249,6 +255,7 @@ func (s *Mergeability) reconcile(repo domain.RepoName) bool {
249	255		}
250	256		res, err := s.compute(ctx, repo, mr.Number, tip, head)
251	257		if err != nil {
	258	+	s.logFailure(ctx, err, "compute the mergeability", repo, mr.Number)
252	259		ok = false
253	260		continue
254	261		}

@@ -258,6 +265,9 @@ func (s *Mergeability) reconcile(repo domain.RepoName) bool {
258	265		continue
259	266		}
260	267		if err != nil {
	268	+	if !errors.Is(err, domain.ErrConflict) { // retried by the next pass
	269	+	s.logFailure(ctx, err, "store the mergeability", repo, mr.Number)
	270	+	}
261	271		ok = false
262	272		continue
263	273		}

@@ -269,6 +279,18 @@ func (s *Mergeability) reconcile(repo domain.RepoName) bool {
269	279		return ok
270	280		}
271	281
	282	+	// logFailure logs an unexpected failure; cancellation is expected on Close.
	283	+	func (s *Mergeability) logFailure(ctx context.Context, err error, op string, repo domain.RepoName, number int64) {
	284	+	if ctx.Err() != nil \|\| errors.Is(err, context.Canceled) \|\| errors.Is(err, domain.ErrNotFound) {
	285	+	return
	286	+	}
	287	+	e := ctxLog(ctx).Error().Err(err).Str("repo", repo.String())
	288	+	if number != 0 {
	289	+	e = e.Int64("number", number)
	290	+	}
	291	+	e.Msg("mergeability: " + op)
	292	+	}
	293	+
272	294		// retryItem is one scheduled retry.
273	295		type retryItem struct {
274	296		due time.Time

@@ -345,7 +367,7 @@ func (s *Mergeability) scheduler() {
345	367		case <-after:
346	368		after = nil
347	369		}
348		-	for _, k := range s.popDue(head.key) {
	370	+	for _, k := range s.popDue(waited) {
349	371		if s.ctx.Err() != nil {
350	372		return
351	373		}

@@ -354,14 +376,16 @@ func (s *Mergeability) scheduler() {
354	376		}
355	377		}
356	378
357		-	// popDue removes the waited item and the ones due as well.
358		-	func (s *Mergeability) popDue(waited domain.RepoName) []domain.RepoName {
	379	+	// popDue removes the waited item and the ones due as well. The waited item is
	380	+	// removed regardless of the clock only when both its key and its deadline
	381	+	// match: a replacement retry of the same repository is kept until it is due.
	382	+	func (s *Mergeability) popDue(waited retryItem) []domain.RepoName {
359	383		s.retryMu.Lock()
360	384		defer s.retryMu.Unlock()
361	385		now := s.cfg.Clock.Now()
362	386		var out []domain.RepoName
363	387		s.heap = slices.DeleteFunc(s.heap, func(it retryItem) bool {
364		-	if it.key != waited && it.due.After(now) {
	388	+	if !(it.key == waited.key && it.due.Equal(waited.due)) && it.due.After(now) {
365	389		return false
366	390		}
367	391		out = append(out, it.key)

 		defer s.wg.Done()
 		for attempt := 0; attempt <= mergeabilityRetries; attempt++ {
 			repos, err := s.cfg.Meta.List(s.ctx)
+			if err != nil && s.ctx.Err() == nil {
+				ctxLog(s.ctx).Error().Err(err).Int("attempt", attempt+1).Msg("mergeability: list the repositories at startup")
+			}
 			if err == nil {
 				for _, repo := range repos {
 					s.markDirty(repo.Name)
 		return true
 	}
 	if err != nil {
+		s.logFailure(ctx, err, "get the repository", repo, 0)
 		return false
 	}
 	mrs, err := s.cfg.Store.ListOpen(ctx, repo)
 	if err != nil {
+		s.logFailure(ctx, err, "list the open merge requests", repo, 0)
 		return false
 	}
 	ok := true
 		}
 		prev, err := s.cfg.Store.GetMergeability(ctx, repo, mr.Number)
 		if err != nil && !errors.Is(err, domain.ErrNotFound) {
+			s.logFailure(ctx, err, "get the mergeability", repo, mr.Number)
 			ok = false
 			continue
 		}
 		}
 		res, err := s.compute(ctx, repo, mr.Number, tip, head)
 		if err != nil {
+			s.logFailure(ctx, err, "compute the mergeability", repo, mr.Number)
 			ok = false
 			continue
 		}
 			continue
 		}
 		if err != nil {
+			if !errors.Is(err, domain.ErrConflict) { // retried by the next pass
+				s.logFailure(ctx, err, "store the mergeability", repo, mr.Number)
+			}
 			ok = false
 			continue
 		}
 	return ok
 }
+// logFailure logs an unexpected failure; cancellation is expected on Close.
+func (s *Mergeability) logFailure(ctx context.Context, err error, op string, repo domain.RepoName, number int64) {
+	if ctx.Err() != nil || errors.Is(err, context.Canceled) || errors.Is(err, domain.ErrNotFound) {
+		return
+	}
+	e := ctxLog(ctx).Error().Err(err).Str("repo", repo.String())
+	if number != 0 {
+		e = e.Int64("number", number)
+	}
+	e.Msg("mergeability: " + op)
+}
 // retryItem is one scheduled retry.
 type retryItem struct {
 	due time.Time
 		case <-after:
 			after = nil
 		}
-		for _, k := range s.popDue(head.key) {
+		for _, k := range s.popDue(waited) {
 			if s.ctx.Err() != nil {
 				return
 			}
 	}
 }
-// popDue removes the waited item and the ones due as well.
+// popDue removes the waited item and the ones due as well. The waited item is
-func (s *Mergeability) popDue(waited domain.RepoName) []domain.RepoName {
+// removed regardless of the clock only when both its key and its deadline
+// match: a replacement retry of the same repository is kept until it is due.
+func (s *Mergeability) popDue(waited retryItem) []domain.RepoName {
 	s.retryMu.Lock()
 	defer s.retryMu.Unlock()
 	now := s.cfg.Clock.Now()
 	var out []domain.RepoName
 	s.heap = slices.DeleteFunc(s.heap, func(it retryItem) bool {
-		if it.key != waited && it.due.After(now) {
+		if !(it.key == waited.key && it.due.Equal(waited.due)) && it.due.After(now) {
 			return false
 		}
 		out = append(out, it.key)