fix(mr): alternate pending and dirty work in the mergeability queue

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

2 files · +45 -2

M internal/usecase/mergeability_queue.go +19 -2


@@ -32,6 +32,9 @@ type mergeQueue struct {
32	32		// their next check is published whatever it is.
33	33		seen map[mergeKey]bool
34	34		wake chan struct{}
	35	+	// preferDirty alternates the service between pending keys and dirty
	36	+	// repositories, so that neither starves the other.
	37	+	preferDirty bool
35	38		}
36	39
37	40		func newMergeQueue(max int) *mergeQueue {

@@ -80,10 +83,19 @@ func (q *mergeQueue) markDirtyLocked(repo domain.RepoName) {
80	83		}
81	84
82	85		// next takes the next check, which it marks running, or the next dirty
83		-	// repository to rescan; pending checks first.
	86	+	// repository to rescan, alternating between the two when both wait.
84	87		func (q mergeQueue) next() (key mergeKey, dirty *domain.RepoName) {
85	88		q.mu.Lock()
86	89		defer q.mu.Unlock()
	90	+	if q.preferDirty && len(q.dirtyOrder) > 0 {
	91	+	q.preferDirty = false
	92	+	r := q.dirtyOrder[0]
	93	+	q.dirtyOrder = q.dirtyOrder[1:]
	94	+	delete(q.dirty, r)
	95	+	dirty = &r
	96	+	q.signalIfWaitingLocked()
	97	+	return nil, dirty
	98	+	}
87	99		for key == nil && len(q.order) > 0 {
88	100		k := q.order[0]
89	101		q.order = q.order[1:]

@@ -98,16 +110,21 @@ func (q mergeQueue) next() (key mergeKey, dirty *domain.RepoName) {
98	110		q.running[k] = true
99	111		key = &k
100	112		}
	113	+	q.preferDirty = key != nil
101	114		if key == nil && len(q.dirtyOrder) > 0 {
102	115		r := q.dirtyOrder[0]
103	116		q.dirtyOrder = q.dirtyOrder[1:]
104	117		delete(q.dirty, r)
105	118		dirty = &r
106	119		}
	120	+	q.signalIfWaitingLocked()
	121	+	return key, dirty
	122	+	}
	123	+
	124	+	func (q *mergeQueue) signalIfWaitingLocked() {
107	125		if len(q.order) > 0 \|\| len(q.dirtyOrder) > 0 {
108	126		q.signal() // let another worker take the rest
109	127		}
110		-	return key, dirty
111	128		}
112	129
113	130		// tryRun marks k running for a rescan; when it already runs it asks for

 	// their next check is published whatever it is.
 	seen map[mergeKey]bool
 	wake chan struct{}
+	// preferDirty alternates the service between pending keys and dirty
+	// repositories, so that neither starves the other.
+	preferDirty bool
 }
 func newMergeQueue(max int) *mergeQueue {
 }
 // next takes the next check, which it marks running, or the next dirty
-// repository to rescan; pending checks first.
+// repository to rescan, alternating between the two when both wait.
 func (q *mergeQueue) next() (key *mergeKey, dirty *domain.RepoName) {
 	q.mu.Lock()
 	defer q.mu.Unlock()
+	if q.preferDirty && len(q.dirtyOrder) > 0 {
+		q.preferDirty = false
+		r := q.dirtyOrder[0]
+		q.dirtyOrder = q.dirtyOrder[1:]
+		delete(q.dirty, r)
+		dirty = &r
+		q.signalIfWaitingLocked()
+		return nil, dirty
+	}
 	for key == nil && len(q.order) > 0 {
 		k := q.order[0]
 		q.order = q.order[1:]
 		q.running[k] = true
 		key = &k
 	}
+	q.preferDirty = key != nil
 	if key == nil && len(q.dirtyOrder) > 0 {
 		r := q.dirtyOrder[0]
 		q.dirtyOrder = q.dirtyOrder[1:]
 		delete(q.dirty, r)
 		dirty = &r
 	}
+	q.signalIfWaitingLocked()
+	return key, dirty
+}
+func (q *mergeQueue) signalIfWaitingLocked() {
 	if len(q.order) > 0 || len(q.dirtyOrder) > 0 {
 		q.signal() // let another worker take the rest
 	}
-	return key, dirty
 }
 // tryRun marks k running for a rescan; when it already runs it asks for

M internal/usecase/mergeability_queue_test.go +26 -0


@@ -46,3 +46,29 @@ func TestMergeQueueNextMarksAgainForARunningKey(t *testing.T) {
46	46		assert.True(t, q.again[k])
47	47		assert.False(t, q.pending[k])
48	48		}
	49	+
	50	+	func TestMergeQueueNextServicesADirtyRepoUnderSustainedRefill(t *testing.T) {
	51	+	repo, err := domain.ParseRepoName("proj")
	52	+	require.NoError(t, err)
	53	+	q := newMergeQueue(2)
	54	+	q.markDirty(repo)
	55	+	n := int64(0)
	56	+	refill := func() {
	57	+	for range 2 {
	58	+	n++
	59	+	q.add(mergeKey{repo, n})
	60	+	}
	61	+	}
	62	+	refill()
	63	+	for range 3 {
	64	+	key, dirty := q.next()
	65	+	if dirty != nil {
	66	+	assert.Equal(t, repo, *dirty)
	67	+	return
	68	+	}
	69	+	require.NotNil(t, key)
	70	+	q.finish(*key)
	71	+	refill()
	72	+	}
	73	+	t.Fatal("the dirty repository starved")
	74	+	}


@@ -46,3 +46,29 @@ func TestMergeQueueNextMarksAgainForARunningKey(t *testing.T) {
46	assert.True(t, q.again[k])	46		assert.True(t, q.again[k])
47	assert.False(t, q.pending[k])	47		assert.False(t, q.pending[k])
48	}	48		}
		49	+
		50	+	func TestMergeQueueNextServicesADirtyRepoUnderSustainedRefill(t *testing.T) {
		51	+	repo, err := domain.ParseRepoName("proj")
		52	+	require.NoError(t, err)
		53	+	q := newMergeQueue(2)
		54	+	q.markDirty(repo)
		55	+	n := int64(0)
		56	+	refill := func() {
		57	+	for range 2 {
		58	+	n++
		59	+	q.add(mergeKey{repo, n})
		60	+	}
		61	+	}
		62	+	refill()
		63	+	for range 3 {
		64	+	key, dirty := q.next()
		65	+	if dirty != nil {
		66	+	assert.Equal(t, repo, *dirty)
		67	+	return
		68	+	}
		69	+	require.NotNil(t, key)
		70	+	q.finish(*key)
		71	+	refill()
		72	+	}
		73	+	t.Fatal("the dirty repository starved")
		74	+	}