22ff0d6 fix(ci): keep a pushed head unsettled until its pull events are discovered · ermite

M internal/usecase/automerge_test.go +17 -3


@@ -122,7 +122,7 @@ func TestAutoMergeWaitsWhileCINotSettledOrChecksPending(t *testing.T) {
122	122		enableAuto(t, r, mr)
123	123		activity := mocks.NewCIActivity(t)
124	124		activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(false).Once()
125		-	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(true).Once()
	125	+	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(true).Twice()
126	126		checks.EXPECT().CommitChecks(mock.Anything, r.repo, r.head().String()).Return([]domain.CICheck{{Status: domain.CICompleted, Conclusion: domain.CISuccess}, {Status: domain.CIQueued, Name: "pending"}}, nil)
127	127		a := autoEvaluator(r, checks, activity, nil)
128	128		defer a.Close(r.ctx)

@@ -223,7 +223,7 @@ func TestAutoMergeCancelsWhenEnablerTokenRevokedOrExpired(t *testing.T) {
223	223		}
224	224		}
225	225
226		-	func TestAutoMergeCancelsOnPermanentMergeFailure(t *testing.T) {
	226	+	func TestAutoMergeWaitsOnConflictingOrBehindHead(t *testing.T) {
227	227		for _, mode := range []domain.MergeMode{domain.MergeModeFastForward, domain.MergeModeSquash} {
228	228		t.Run(mode.String(), func(t *testing.T) {
229	229		r, mr, checks := autoRig(t)

@@ -237,7 +237,7 @@ func TestAutoMergeCancelsOnPermanentMergeFailure(t *testing.T) {
237	237		a := autoEvaluator(r, checks, nil, nil)
238	238		defer a.Close(r.ctx)
239	239		require.True(t, a.Evaluate(r.repo))
240		-	assert.Nil(t, storedAuto(t, r, mr).AutoMerge)
	240	+	assert.NotNil(t, storedAuto(t, r, mr).AutoMerge)
241	241		assert.True(t, storedAuto(t, r, mr).IsOpen())
242	242		})
243	243		}

@@ -424,3 +424,17 @@ func TestAutoMergeResolvesEnablerAfterMergeGate(t *testing.T) {
424	424		require.Nil(t, got.AutoMerge)
425	425		require.True(t, got.IsOpen())
426	426		}
	427	+
	428	+	func TestAutoMergeRechecksSettledUnderMergeGate(t *testing.T) {
	429	+	r, mr, checks := autoRig(t)
	430	+	enableAuto(t, r, mr)
	431	+	activity := mocks.NewCIActivity(t)
	432	+	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(true).Once()
	433	+	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(false).Once()
	434	+	checks.EXPECT().CommitChecks(mock.Anything, r.repo, r.head().String()).Return([]domain.CICheck{{Status: domain.CICompleted, Conclusion: domain.CISuccess}}, nil).Once()
	435	+	a := autoEvaluator(r, checks, activity, nil)
	436	+	defer a.Close(r.ctx)
	437	+	require.True(t, a.Evaluate(r.repo))
	438	+	require.True(t, storedAuto(t, r, mr).IsOpen())
	439	+	require.NotNil(t, storedAuto(t, r, mr).AutoMerge)
	440	+	}

 	enableAuto(t, r, mr)
 	activity := mocks.NewCIActivity(t)
 	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(false).Once()
-	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(true).Once()
+	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(true).Twice()
 	checks.EXPECT().CommitChecks(mock.Anything, r.repo, r.head().String()).Return([]domain.CICheck{{Status: domain.CICompleted, Conclusion: domain.CISuccess}, {Status: domain.CIQueued, Name: "pending"}}, nil)
 	a := autoEvaluator(r, checks, activity, nil)
 	defer a.Close(r.ctx)
 	}
 }
-func TestAutoMergeCancelsOnPermanentMergeFailure(t *testing.T) {
+func TestAutoMergeWaitsOnConflictingOrBehindHead(t *testing.T) {
 	for _, mode := range []domain.MergeMode{domain.MergeModeFastForward, domain.MergeModeSquash} {
 		t.Run(mode.String(), func(t *testing.T) {
 			r, mr, checks := autoRig(t)
 			a := autoEvaluator(r, checks, nil, nil)
 			defer a.Close(r.ctx)
 			require.True(t, a.Evaluate(r.repo))
-			assert.Nil(t, storedAuto(t, r, mr).AutoMerge)
+			assert.NotNil(t, storedAuto(t, r, mr).AutoMerge)
 			assert.True(t, storedAuto(t, r, mr).IsOpen())
 		})
 	}
 	require.Nil(t, got.AutoMerge)
 	require.True(t, got.IsOpen())
 }
+func TestAutoMergeRechecksSettledUnderMergeGate(t *testing.T) {
+	r, mr, checks := autoRig(t)
+	enableAuto(t, r, mr)
+	activity := mocks.NewCIActivity(t)
+	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(true).Once()
+	activity.EXPECT().Settled(mock.Anything, r.repo, mr.Number, mr.Source, r.head()).Return(false).Once()
+	checks.EXPECT().CommitChecks(mock.Anything, r.repo, r.head().String()).Return([]domain.CICheck{{Status: domain.CICompleted, Conclusion: domain.CISuccess}}, nil).Once()
+	a := autoEvaluator(r, checks, activity, nil)
+	defer a.Close(r.ctx)
+	require.True(t, a.Evaluate(r.repo))
+	require.True(t, storedAuto(t, r, mr).IsOpen())
+	require.NotNil(t, storedAuto(t, r, mr).AutoMerge)
+}

M internal/usecase/ci.go +9 -0


@@ -348,6 +348,15 @@ func (c *CI) Close(_ context.Context) {
348	348		c.mu.Unlock()
349	349		c.stop()
350	350		c.wg.Wait()
	351	+	// Workers stop on cancellation; release the planning parts they did not take.
	352	+	for {
	353	+	select {
	354	+	case t := <-c.queue:
	355	+	t.done()
	356	+	default:
	357	+	return
	358	+	}
	359	+	}
351	360		}
352	361
353	362		func (c *CI) now() time.Time { return c.cfg.Clock.Now().UTC() }


@@ -348,6 +348,15 @@ func (c *CI) Close(_ context.Context) {
348	c.mu.Unlock()	348		c.mu.Unlock()
349	c.stop()	349		c.stop()
350	c.wg.Wait()	350		c.wg.Wait()
		351	+	// Workers stop on cancellation; release the planning parts they did not take.
		352	+	for {
		353	+	select {
		354	+	case t := <-c.queue:
		355	+	t.done()
		356	+	default:
		357	+	return
		358	+	}
		359	+	}
351	}	360		}
352		361
353	func (c *CI) now() time.Time { return c.cfg.Clock.Now().UTC() }	362		func (c *CI) now() time.Time { return c.cfg.Clock.Now().UTC() }

M internal/usecase/ci_activity_test.go +36 -1


@@ -18,7 +18,7 @@ func TestSettledCountsQueuedPushesAndPullEvents(t *testing.T) {
18	18		trigger := mocks.NewAutoMergeTrigger(t)
19	19		u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.TriggerQueue = 1; cfg.AutoMerge = trigger })
20	20		repo := name(t, "proj")
21		-	trigger.EXPECT().OnRepoChanged(mock.Anything, repo).Return().Twice()
	21	+	trigger.EXPECT().OnRepoChanged(mock.Anything, repo).Return().Times(3)
22	22		u.store.EXPECT().ListRunners(mock.Anything).Return(nil, nil).Once()
23	23		ctx := context.Background()
24	24		settled := func(n int64) bool { return u.ci.Settled(ctx, repo, n, "feat", domain.ObjectID{}) }

@@ -35,3 +35,38 @@ func TestSettledCountsQueuedPushesAndPullEvents(t *testing.T) {
35	35		u.ci.FinishQueuedPull()
36	36		assert.True(t, settled(1))
37	37		}
	38	+
	39	+	func TestSettledWaitsForPushDiscoveryAndReleasesOnError(t *testing.T) {
	40	+	mrs := mocks.NewMergeRequestStore(t)
	41	+	entered, release := make(chan struct{}), make(chan struct{})
	42	+	u := newCIUnit(t, func(cfg *usecase.CIConfig) {
	43	+	cfg.MRs = mrs
	44	+	cfg.Pull = mocks.NewPullGit(t)
	45	+	cfg.PullStage = func(string) { close(entered); <-release }
	46	+	})
	47	+	repo := name(t, "proj")
	48	+	mrs.EXPECT().OpenByTarget(mock.Anything, repo, mock.Anything, mock.Anything).Return(nil, false, nil)
	49	+	mrs.EXPECT().OpenBySource(mock.Anything, repo, mock.Anything, mock.Anything).Return(nil, false, domain.ErrUnavailable)
	50	+	u.store.EXPECT().ListRunners(mock.Anything).Return(nil, nil).Once()
	51	+	done := make(chan struct{})
	52	+	go func() {
	53	+	u.ci.OnPush(context.Background(), repo, []domain.RefUpdate{{Ref: "refs/heads/feat", New: "1111111111111111111111111111111111111111"}}, domain.Root())
	54	+	close(done)
	55	+	}()
	56	+	<-entered
	57	+	u.ci.PlanQueuedPush()
	58	+	assert.False(t, u.ci.Settled(context.Background(), repo, 1, "feat", domain.ObjectID{}))
	59	+	close(release)
	60	+	<-done
	61	+	assert.True(t, u.ci.Settled(context.Background(), repo, 1, "feat", domain.ObjectID{}))
	62	+	}
	63	+
	64	+	func TestSettledReleasesQueuedPushOnShutdown(t *testing.T) {
	65	+	u := newCIUnit(t, nil)
	66	+	repo := name(t, "proj")
	67	+	u.ci.QueuePush(context.Background(), repo)
	68	+	u.ci.Close(context.Background())
	69	+	assert.True(t, u.ci.Settled(context.Background(), repo, 1, "feat", domain.ObjectID{}))
	70	+	u.ci.QueuePush(context.Background(), repo)
	71	+	assert.True(t, u.ci.Settled(context.Background(), repo, 1, "feat", domain.ObjectID{}))
	72	+	}


@@ -18,7 +18,7 @@ func TestSettledCountsQueuedPushesAndPullEvents(t *testing.T) {
18		trigger := mocks.NewAutoMergeTrigger(t)	18		trigger := mocks.NewAutoMergeTrigger(t)
19		u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.TriggerQueue = 1; cfg.AutoMerge = trigger })	19		u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.TriggerQueue = 1; cfg.AutoMerge = trigger })
20		repo := name(t, "proj")	20		repo := name(t, "proj")
21	-	trigger.EXPECT().OnRepoChanged(mock.Anything, repo).Return().Twice()	21	+	trigger.EXPECT().OnRepoChanged(mock.Anything, repo).Return().Times(3)
22		u.store.EXPECT().ListRunners(mock.Anything).Return(nil, nil).Once()	22		u.store.EXPECT().ListRunners(mock.Anything).Return(nil, nil).Once()
23		ctx := context.Background()	23		ctx := context.Background()
24		settled := func(n int64) bool { return u.ci.Settled(ctx, repo, n, "feat", domain.ObjectID{}) }	24		settled := func(n int64) bool { return u.ci.Settled(ctx, repo, n, "feat", domain.ObjectID{}) }

@@ -35,3 +35,38 @@ func TestSettledCountsQueuedPushesAndPullEvents(t *testing.T) {
35		u.ci.FinishQueuedPull()	35		u.ci.FinishQueuedPull()
36		assert.True(t, settled(1))	36		assert.True(t, settled(1))
37		}	37		}
			38	+
			39	+	func TestSettledWaitsForPushDiscoveryAndReleasesOnError(t *testing.T) {
			40	+	mrs := mocks.NewMergeRequestStore(t)
			41	+	entered, release := make(chan struct{}), make(chan struct{})
			42	+	u := newCIUnit(t, func(cfg *usecase.CIConfig) {
			43	+	cfg.MRs = mrs
			44	+	cfg.Pull = mocks.NewPullGit(t)
			45	+	cfg.PullStage = func(string) { close(entered); <-release }
			46	+	})
			47	+	repo := name(t, "proj")
			48	+	mrs.EXPECT().OpenByTarget(mock.Anything, repo, mock.Anything, mock.Anything).Return(nil, false, nil)
			49	+	mrs.EXPECT().OpenBySource(mock.Anything, repo, mock.Anything, mock.Anything).Return(nil, false, domain.ErrUnavailable)
			50	+	u.store.EXPECT().ListRunners(mock.Anything).Return(nil, nil).Once()
			51	+	done := make(chan struct{})
			52	+	go func() {
			53	+	u.ci.OnPush(context.Background(), repo, []domain.RefUpdate{{Ref: "refs/heads/feat", New: "1111111111111111111111111111111111111111"}}, domain.Root())
			54	+	close(done)
			55	+	}()
			56	+	<-entered
			57	+	u.ci.PlanQueuedPush()
			58	+	assert.False(t, u.ci.Settled(context.Background(), repo, 1, "feat", domain.ObjectID{}))
			59	+	close(release)
			60	+	<-done
			61	+	assert.True(t, u.ci.Settled(context.Background(), repo, 1, "feat", domain.ObjectID{}))
			62	+	}
			63	+
			64	+	func TestSettledReleasesQueuedPushOnShutdown(t *testing.T) {
			65	+	u := newCIUnit(t, nil)
			66	+	repo := name(t, "proj")
			67	+	u.ci.QueuePush(context.Background(), repo)
			68	+	u.ci.Close(context.Background())
			69	+	assert.True(t, u.ci.Settled(context.Background(), repo, 1, "feat", domain.ObjectID{}))
			70	+	u.ci.QueuePush(context.Background(), repo)
			71	+	assert.True(t, u.ci.Settled(context.Background(), repo, 1, "feat", domain.ObjectID{}))
			72	+	}

M internal/usecase/ci_trigger.go +41 -11


@@ -12,6 +12,7 @@ import (
12	12		"slices"
13	13		"strconv"
14	14		"strings"
	15	+	"sync/atomic"
15	16		"time"
16	17
17	18		"github.com/bnema/zerowrap"

@@ -33,6 +34,7 @@ const (
33	34
34	35		// ciTrigger is a push waiting to be planned.
35	36		type ciTrigger struct {
	37	+	done func()
36	38		repo domain.RepoName
37	39		updates []domain.RefUpdate
38	40		actor string

@@ -54,6 +56,9 @@ type ciTrigger struct {
54	56		// with GITHUB_TOKEN start no run): that is what stops a workflow that pushes
55	57		// from running itself again.
56	58		func (c *CI) OnPush(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, pusher domain.Principal) {
	59	+	done := c.pushLifecycle(ctx, repo, 2)
	60	+	defer done()
	61	+
57	62		// Whoever pushed, the workflow files of the default branch may have changed.
58	63		// (Who pushed was recorded before the refs moved: Pushes.Check.)
59	64		c.markSchedulesDirty(repo, updates)

@@ -63,9 +68,10 @@ func (c *CI) OnPush(ctx context.Context, repo domain.RepoName, updates []domain.
63	68		if pusher.ViaJobToken() {
64	69		log := zerowrap.FromCtx(ctx)
65	70		log.Debug().Str("repo", repo.String()).Int("refs", len(updates)).Msg("push by a job token starts no run")
	71	+	done()
66	72		return
67	73		}
68		-	c.enqueue(ctx, repo, updates, domain.AuthorName(pusher), domain.PusherOf(pusher), domain.TrustedActor(pusher))
	74	+	c.enqueuePush(ctx, repo, updates, domain.AuthorName(pusher), domain.PusherOf(pusher), domain.TrustedActor(pusher), done)
69	75		}
70	76
71	77		// OnMerge queues the move of a target branch by a merge. The actor of the run is

@@ -75,6 +81,9 @@ func (c *CI) OnPush(ctx context.Context, repo domain.RepoName, updates []domain.
75	81		// merging. For the same reason the run is trusted (gets secrets) only when the
76	82		// merger is: the displayed actor is not the one that decides.
77	83		func (c *CI) OnMerge(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, merger domain.Principal) {
	84	+	done := c.pushLifecycle(ctx, repo, 2)
	85	+	defer done()
	86	+
78	87		c.markSchedulesDirty(repo, updates)
79	88		// The merge moved a target branch: the test merges of the other requests
80	89		// into it are computed again (a merge starts no synchronize run), after the

@@ -83,12 +92,32 @@ func (c *CI) OnMerge(ctx context.Context, repo domain.RepoName, updates []domain
83	92		if merger.ViaJobToken() {
84	93		log := zerowrap.FromCtx(ctx)
85	94		log.Debug().Str("repo", repo.String()).Int("refs", len(updates)).Msg("merge by a job token starts no run")
	95	+	done()
86	96		return
87	97		}
88		-	c.enqueue(ctx, repo, updates, domain.AuthorName(domain.Owner()), domain.PusherOf(merger), domain.TrustedActor(merger))
	98	+	c.enqueuePush(ctx, repo, updates, domain.AuthorName(domain.Owner()), domain.PusherOf(merger), domain.TrustedActor(merger), done)
89	99		}
90	100
91	101		func (c *CI) enqueue(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, actor string, pusher domain.CIPusher, trusted bool) {
	102	+	c.enqueuePush(ctx, repo, updates, actor, pusher, trusted, c.pushLifecycle(ctx, repo, 1))
	103	+	}
	104	+
	105	+	// pushLifecycle holds one repository activity count until all parts finish.
	106	+	func (c *CI) pushLifecycle(ctx context.Context, repo domain.RepoName, parts int32) func() {
	107	+	c.mu.Lock()
	108	+	c.pushes[repo]++
	109	+	c.mu.Unlock()
	110	+	var remaining atomic.Int32
	111	+	remaining.Store(parts)
	112	+	return func() {
	113	+	if remaining.Add(-1) == 0 {
	114	+	c.finishPush(repo)
	115	+	c.kickAutoMerge(ctx, repo)
	116	+	}
	117	+	}
	118	+	}
	119	+
	120	+	func (c *CI) enqueuePush(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, actor string, pusher domain.CIPusher, trusted bool, done func()) {
92	121		var kept []domain.RefUpdate
93	122		for _, u := range updates {
94	123		if strings.HasPrefix(u.Ref, "refs/heads/") \|\| strings.HasPrefix(u.Ref, "refs/tags/") {

@@ -96,24 +125,25 @@ func (c *CI) enqueue(ctx context.Context, repo domain.RepoName, updates []domain
96	125		}
97	126		}
98	127		if len(kept) == 0 {
	128	+	done()
99	129		return
100	130		}
101		-	t := ciTrigger{repo: repo, updates: kept, actor: actor, pusher: pusher, trusted: trusted, at: c.now(), log: zerowrap.FromCtx(ctx)}
	131	+	t := ciTrigger{done: done, repo: repo, updates: kept, actor: actor, pusher: pusher, trusted: trusted, at: c.now(), log: zerowrap.FromCtx(ctx)}
102	132		c.mu.Lock()
103		-	closed := c.closed
104		-	if !closed {
105		-	c.pushes[repo]++
106		-	}
107		-	c.mu.Unlock()
108		-	if closed {
	133	+	if c.closed {
	134	+	c.mu.Unlock()
	135	+	done()
109	136		return
110	137		}
111	138		select {
112	139		case c.queue <- t:
	140	+	c.mu.Unlock()
113	141		default:
114		-	c.finishPush(repo)
	142	+	c.mu.Unlock()
	143	+	done()
115	144		t.log.Error().Str("repo", repo.String()).Int("refs", len(kept)).Msg("ci trigger queue is full: the workflows of this push are not run")
116	145		}
	146	+
117	147		}
118	148
119	149		func (c *CI) triggerLoop() {

@@ -131,7 +161,7 @@ func (c *CI) triggerLoop() {
131	161		// planPush plans one push; a panic or an error is logged and never escapes.
132	162		func (c *CI) planPush(t ciTrigger) {
133	163		ctx := zerowrap.WithCtx(c.ctx, t.log)
134		-	defer func() { c.finishPush(t.repo); c.kickAutoMerge(ctx, t.repo) }()
	164	+	defer t.done()
135	165		defer func() {
136	166		if r := recover(); r != nil {
137	167		t.log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Str("repo", t.repo.String()).Msg("ci trigger panicked")

 	"slices"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 	"github.com/bnema/zerowrap"
 // ciTrigger is a push waiting to be planned.
 type ciTrigger struct {
+	done    func()
 	repo    domain.RepoName
 	updates []domain.RefUpdate
 	actor   string
 // with GITHUB_TOKEN start no run): that is what stops a workflow that pushes
 // from running itself again.
 func (c *CI) OnPush(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, pusher domain.Principal) {
+	done := c.pushLifecycle(ctx, repo, 2)
+	defer done()
 	// Whoever pushed, the workflow files of the default branch may have changed.
 	// (Who pushed was recorded before the refs moved: Pushes.Check.)
 	c.markSchedulesDirty(repo, updates)
 	if pusher.ViaJobToken() {
 		log := zerowrap.FromCtx(ctx)
 		log.Debug().Str("repo", repo.String()).Int("refs", len(updates)).Msg("push by a job token starts no run")
+		done()
 		return
 	}
-	c.enqueue(ctx, repo, updates, domain.AuthorName(pusher), domain.PusherOf(pusher), domain.TrustedActor(pusher))
+	c.enqueuePush(ctx, repo, updates, domain.AuthorName(pusher), domain.PusherOf(pusher), domain.TrustedActor(pusher), done)
 }
 // OnMerge queues the move of a target branch by a merge. The actor of the run is
 // merging. For the same reason the run is trusted (gets secrets) only when the
 // merger is: the displayed actor is not the one that decides.
 func (c *CI) OnMerge(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, merger domain.Principal) {
+	done := c.pushLifecycle(ctx, repo, 2)
+	defer done()
 	c.markSchedulesDirty(repo, updates)
 	// The merge moved a target branch: the test merges of the other requests
 	// into it are computed again (a merge starts no synchronize run), after the
 	if merger.ViaJobToken() {
 		log := zerowrap.FromCtx(ctx)
 		log.Debug().Str("repo", repo.String()).Int("refs", len(updates)).Msg("merge by a job token starts no run")
+		done()
 		return
 	}
-	c.enqueue(ctx, repo, updates, domain.AuthorName(domain.Owner()), domain.PusherOf(merger), domain.TrustedActor(merger))
+	c.enqueuePush(ctx, repo, updates, domain.AuthorName(domain.Owner()), domain.PusherOf(merger), domain.TrustedActor(merger), done)
 }
 func (c *CI) enqueue(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, actor string, pusher domain.CIPusher, trusted bool) {
+	c.enqueuePush(ctx, repo, updates, actor, pusher, trusted, c.pushLifecycle(ctx, repo, 1))
+}
+// pushLifecycle holds one repository activity count until all parts finish.
+func (c *CI) pushLifecycle(ctx context.Context, repo domain.RepoName, parts int32) func() {
+	c.mu.Lock()
+	c.pushes[repo]++
+	c.mu.Unlock()
+	var remaining atomic.Int32
+	remaining.Store(parts)
+	return func() {
+		if remaining.Add(-1) == 0 {
+			c.finishPush(repo)
+			c.kickAutoMerge(ctx, repo)
+		}
+	}
+}
+func (c *CI) enqueuePush(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, actor string, pusher domain.CIPusher, trusted bool, done func()) {
 	var kept []domain.RefUpdate
 	for _, u := range updates {
 		if strings.HasPrefix(u.Ref, "refs/heads/") || strings.HasPrefix(u.Ref, "refs/tags/") {
 		}
 	}
 	if len(kept) == 0 {
+		done()
 		return
 	}
-	t := ciTrigger{repo: repo, updates: kept, actor: actor, pusher: pusher, trusted: trusted, at: c.now(), log: zerowrap.FromCtx(ctx)}
+	t := ciTrigger{done: done, repo: repo, updates: kept, actor: actor, pusher: pusher, trusted: trusted, at: c.now(), log: zerowrap.FromCtx(ctx)}
 	c.mu.Lock()
-	closed := c.closed
+	if c.closed {
-	if !closed {
+		c.mu.Unlock()
-		c.pushes[repo]++
+		done()
-	}
-	c.mu.Unlock()
-	if closed {
 		return
 	}
 	select {
 	case c.queue <- t:
+		c.mu.Unlock()
 	default:
-		c.finishPush(repo)
+		c.mu.Unlock()
+		done()
 		t.log.Error().Str("repo", repo.String()).Int("refs", len(kept)).Msg("ci trigger queue is full: the workflows of this push are not run")
 	}
 }
 func (c *CI) triggerLoop() {
 // planPush plans one push; a panic or an error is logged and never escapes.
 func (c *CI) planPush(t ciTrigger) {
 	ctx := zerowrap.WithCtx(c.ctx, t.log)
-	defer func() { c.finishPush(t.repo); c.kickAutoMerge(ctx, t.repo) }()
+	defer t.done()
 	defer func() {
 		if r := recover(); r != nil {
 			t.log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Str("repo", t.repo.String()).Msg("ci trigger panicked")

M internal/usecase/mergerequests.go +9 -0


@@ -1147,8 +1147,14 @@ func (m *MergeRequests) mergeAs(ctx context.Context, p domain.Principal, name do
1147	1147		} else if behind {
1148	1148		return domain.MergeRequest{}, fmt.Errorf("%w: nothing to merge, %q already contains %s", domain.ErrPreconditionFailed, mr.Target, head)
1149	1149		}
	1150	+	if opts.auto && opts.activity != nil && !opts.activity.Settled(ctx, name, number, mr.Source, head) {
	1151	+	return domain.MergeRequest{}, errAutoMergeWait
	1152	+	}
1150	1153		tree, err := m.checkMode(ctx, mr, mode, tip, head)
1151	1154		if err != nil {
	1155	+	if opts.auto && errors.Is(err, domain.ErrPreconditionFailed) {
	1156	+	return domain.MergeRequest{}, errAutoMergeWait
	1157	+	}
1152	1158		return domain.MergeRequest{}, err
1153	1159		}
1154	1160		if opts.auto && !settings.AllowAutoMerge {

@@ -1156,6 +1162,9 @@ func (m *MergeRequests) mergeAs(ctx context.Context, p domain.Principal, name do
1156	1162		}
1157	1163		if settings.FailingChecksBlock \|\| opts.auto {
1158	1164		if err := m.checksAllow(ctx, name, head); err != nil {
	1165	+	if opts.auto && errors.Is(err, domain.ErrPreconditionFailed) {
	1166	+	return domain.MergeRequest{}, errAutoMergeWait
	1167	+	}
1159	1168		return domain.MergeRequest{}, err
1160	1169		}
1161	1170		}


@@ -1147,8 +1147,14 @@ func (m *MergeRequests) mergeAs(ctx context.Context, p domain.Principal, name do
1147	} else if behind {	1147		} else if behind {
1148	return domain.MergeRequest{}, fmt.Errorf("%w: nothing to merge, %q already contains %s", domain.ErrPreconditionFailed, mr.Target, head)	1148		return domain.MergeRequest{}, fmt.Errorf("%w: nothing to merge, %q already contains %s", domain.ErrPreconditionFailed, mr.Target, head)
1149	}	1149		}
		1150	+	if opts.auto && opts.activity != nil && !opts.activity.Settled(ctx, name, number, mr.Source, head) {
		1151	+	return domain.MergeRequest{}, errAutoMergeWait
		1152	+	}
1150	tree, err := m.checkMode(ctx, mr, mode, tip, head)	1153		tree, err := m.checkMode(ctx, mr, mode, tip, head)
1151	if err != nil {	1154		if err != nil {
		1155	+	if opts.auto && errors.Is(err, domain.ErrPreconditionFailed) {
		1156	+	return domain.MergeRequest{}, errAutoMergeWait
		1157	+	}
1152	return domain.MergeRequest{}, err	1158		return domain.MergeRequest{}, err
1153	}	1159		}
1154	if opts.auto && !settings.AllowAutoMerge {	1160		if opts.auto && !settings.AllowAutoMerge {

@@ -1156,6 +1162,9 @@ func (m *MergeRequests) mergeAs(ctx context.Context, p domain.Principal, name do
1156	}	1162		}
1157	if settings.FailingChecksBlock \|\| opts.auto {	1163		if settings.FailingChecksBlock \|\| opts.auto {
1158	if err := m.checksAllow(ctx, name, head); err != nil {	1164		if err := m.checksAllow(ctx, name, head); err != nil {
		1165	+	if opts.auto && errors.Is(err, domain.ErrPreconditionFailed) {
		1166	+	return domain.MergeRequest{}, errAutoMergeWait
		1167	+	}
1159	return domain.MergeRequest{}, err	1168		return domain.MergeRequest{}, err
1160	}	1169		}
1161	}	1170		}