fix(mirror): mark mirrors dirty on merges and default branch changes, validate config before saving and keep manual throttling on coalesced syncs

commit: 335d134480a771ea99c98e0eb9e5d9b386ae952c
author: brice · 2d ago · 2026-10-20 06:15 UTC
committer: brice · 2d ago · 2026-10-20 10:02 UTC
parent: c59b5fa
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

7 files · +254 -26

M internal/app/forge.go +2 -1


@@ -298,6 +298,7 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
298	298		}
299	299		f.Mirrors = usecase.NewMirrors(mirrorCfg)
300	300		f.Pushes.WithMirror(f.Mirrors)
	301	+	f.Repos.WithMirror(f.Mirrors)
301	302		f.hooks.WithMirrorWarner(f.Mirrors)
302	303
303	304		f.Settings, err = usecase.NewCISettings(ctx, store, store.Settings(), sealer, clock)

@@ -368,7 +369,7 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
368	369		CI: f.CI, Checks: f.CI, PullRequests: f.CI, RefTips: store.CI(), Mergeability: f.Mergeability,
369	370		})
370	371		f.Mergeability.Start()
371		-	f.MergeRequests.WithVerifier(f.Signatures)
	372	+	f.MergeRequests.WithVerifier(f.Signatures).WithMirror(f.Mirrors)
372	373		f.Issues = usecase.NewIssues(usecase.IssuesConfig{
373	374		Meta: store, Store: store.Issues(), MergeRequests: store.MergeRequests(), Bus: f.Bus, Clock: clock,
374	375		})

 	}
 	f.Mirrors = usecase.NewMirrors(mirrorCfg)
 	f.Pushes.WithMirror(f.Mirrors)
+	f.Repos.WithMirror(f.Mirrors)
 	f.hooks.WithMirrorWarner(f.Mirrors)
 	f.Settings, err = usecase.NewCISettings(ctx, store, store.Settings(), sealer, clock)
 		CI: f.CI, Checks: f.CI, PullRequests: f.CI, RefTips: store.CI(), Mergeability: f.Mergeability,
 	})
 	f.Mergeability.Start()
-	f.MergeRequests.WithVerifier(f.Signatures)
+	f.MergeRequests.WithVerifier(f.Signatures).WithMirror(f.Mirrors)
 	f.Issues = usecase.NewIssues(usecase.IssuesConfig{
 		Meta: store, Store: store.Issues(), MergeRequests: store.MergeRequests(), Bus: f.Bus, Clock: clock,
 	})

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


@@ -58,6 +58,7 @@ type MergeRequests struct {
58	58		tips ports.CIRefTipStore // nil: who merges is not recorded
59	59		verifier ports.CommitVerifier // nil: commits are not checked (ADR 0030)
60	60		mergeab MergeabilityHook // nil: no mergeability
	61	+	mirror ports.MirrorTrigger // nil: no outbound mirror
61	62
62	63		mu sync.Mutex
63	64		locks map[domain.RepoName]chan struct{}

@@ -116,6 +117,12 @@ func NewMergeRequests(cfg MergeRequestsConfig) *MergeRequests {
116	117		}
117	118		}
118	119
	120	+	// WithMirror makes a merge notify outbound mirrors. Call before the first merge.
	121	+	func (m MergeRequests) WithMirror(t ports.MirrorTrigger) MergeRequests {
	122	+	m.mirror = t
	123	+	return m
	124	+	}
	125	+
119	126		// WithVerifier makes GetMergeRequest fill the signature state of the commits
120	127		// it lists (ADR 0030).
121	128		func (m MergeRequests) WithVerifier(v ports.CommitVerifier) MergeRequests {

@@ -1172,6 +1179,9 @@ func (m *MergeRequests) afterMerge(ctx context.Context, p domain.Principal, repo
1172	1179		m.bus.Publish(ctx, domain.PushEvent{
1173	1180		Repo: name, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, Updates: updates,
1174	1181		})
	1182	+	if m.mirror != nil {
	1183	+	m.mirror.OnPush(ctx, name, updates) // marks dirty only; never blocks
	1184	+	}
1175	1185		if m.pulls != nil {
1176	1186		m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: mr.Number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID, Merged: true}, p)
1177	1187		}


@@ -58,6 +58,7 @@ type MergeRequests struct {
58	tips ports.CIRefTipStore // nil: who merges is not recorded	58		tips ports.CIRefTipStore // nil: who merges is not recorded
59	verifier ports.CommitVerifier // nil: commits are not checked (ADR 0030)	59		verifier ports.CommitVerifier // nil: commits are not checked (ADR 0030)
60	mergeab MergeabilityHook // nil: no mergeability	60		mergeab MergeabilityHook // nil: no mergeability
		61	+	mirror ports.MirrorTrigger // nil: no outbound mirror
61		62
62	mu sync.Mutex	63		mu sync.Mutex
63	locks map[domain.RepoName]chan struct{}	64		locks map[domain.RepoName]chan struct{}

@@ -116,6 +117,12 @@ func NewMergeRequests(cfg MergeRequestsConfig) *MergeRequests {
116	}	117		}
117	}	118		}
118		119
		120	+	// WithMirror makes a merge notify outbound mirrors. Call before the first merge.
		121	+	func (m MergeRequests) WithMirror(t ports.MirrorTrigger) MergeRequests {
		122	+	m.mirror = t
		123	+	return m
		124	+	}
		125	+
119	// WithVerifier makes GetMergeRequest fill the signature state of the commits	126		// WithVerifier makes GetMergeRequest fill the signature state of the commits
120	// it lists (ADR 0030).	127		// it lists (ADR 0030).
121	func (m MergeRequests) WithVerifier(v ports.CommitVerifier) MergeRequests {	128		func (m MergeRequests) WithVerifier(v ports.CommitVerifier) MergeRequests {

@@ -1172,6 +1179,9 @@ func (m *MergeRequests) afterMerge(ctx context.Context, p domain.Principal, repo
1172	m.bus.Publish(ctx, domain.PushEvent{	1179		m.bus.Publish(ctx, domain.PushEvent{
1173	Repo: name, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, Updates: updates,	1180		Repo: name, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, Updates: updates,
1174	})	1181		})
		1182	+	if m.mirror != nil {
		1183	+	m.mirror.OnPush(ctx, name, updates) // marks dirty only; never blocks
		1184	+	}
1175	if m.pulls != nil {	1185		if m.pulls != nil {
1176	m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: mr.Number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID, Merged: true}, p)	1186		m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: mr.Number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID, Merged: true}, p)
1177	}	1187		}

M internal/usecase/mirrors_config.go +54 -14


@@ -74,10 +74,11 @@ func (m *Mirrors) SetGlobalConfig(ctx context.Context, p domain.Principal, patch
74	74		}
75	75		m.configMu.Lock()
76	76		defer m.configMu.Unlock()
77		-	g, err := m.cfg.Store.GetGlobal(ctx)
	77	+	old, err := m.cfg.Store.GetGlobal(ctx)
78	78		if err != nil {
79		-	return g, err
	79	+	return old, err
80	80		}
	81	+	g := old
81	82		if patch.Enabled != nil {
82	83		g.Enabled = *patch.Enabled
83	84		}

@@ -91,30 +92,55 @@ func (m *Mirrors) SetGlobalConfig(ctx context.Context, p domain.Principal, patch
91	92		g.AllBranches = *patch.AllBranches
92	93		}
93	94		if err := m.validateGlobal(g); err != nil {
94		-	return g, err
95		-	}
96		-	g.UpdatedAt = m.cfg.Clock.Now().UTC()
97		-	if err := m.cfg.Store.PutGlobal(ctx, g); err != nil {
98		-	return g, err
	95	+	return old, err
99	96		}
100	97		repos, err := m.cfg.Meta.List(ctx)
101	98		if err != nil {
102		-	return g, err
	99	+	return old, err
103	100		}
	101	+	// Resolve every effective target before and after, and persist nothing
	102	+	// unless all of them stay valid.
	103	+	var changed []domain.RepoName
104	104		for _, r := range repos {
105	105		c, e := m.cfg.Store.GetRepoConfig(ctx, r.Name)
106		-	if e != nil && !errors.Is(e, domain.ErrNotFound) {
107		-	return g, e
	106	+	var cp *domain.MirrorRepoConfig
	107	+	if e == nil {
	108	+	cp = &c
	109	+	} else if !errors.Is(e, domain.ErrNotFound) {
	110	+	return old, e
	111	+	}
	112	+	before, _ := m.resolve(old, cp, r.Name)
	113	+	after, e := m.resolve(g, cp, r.Name)
	114	+	if e != nil {
	115	+	return old, invalidCron(e)
108	116		}
109		-	if c.URL != "" {
110		-	continue
	117	+	if !sameMirrorTarget(before, after) {
	118	+	changed = append(changed, r.Name)
111	119		}
112		-	if err := m.reset(ctx, r.Name); err != nil {
	120	+	}
	121	+	g.UpdatedAt = m.cfg.Clock.Now().UTC()
	122	+	if err := m.cfg.Store.PutGlobal(ctx, g); err != nil {
	123	+	return old, err
	124	+	}
	125	+	for _, name := range changed {
	126	+	if err := m.reset(ctx, name); err != nil {
113	127		return g, err
114	128		}
115	129		}
116	130		return g, nil
117	131		}
	132	+
	133	+	func invalidCron(e error) error {
	134	+	var inv *domain.InvalidFieldsError
	135	+	if errors.As(e, &inv) {
	136	+	return e
	137	+	}
	138	+	return &domain.InvalidFieldsError{Fields: []domain.FieldViolation{{Field: "cron", Reason: e.Error()}}}
	139	+	}
	140	+
	141	+	func sameMirrorTarget(a, b domain.MirrorTarget) bool {
	142	+	return a.Enabled == b.Enabled && a.URL == b.URL && a.CronSpec == b.CronSpec && a.AllBranches == b.AllBranches && a.Source == b.Source
	143	+	}
118	144		func (m *Mirrors) reset(ctx context.Context, repo domain.RepoName) error {
119	145		t, err := m.target(ctx, repo)
120	146		if err != nil {

@@ -146,6 +172,18 @@ func (m *Mirrors) SetRepoMirror(ctx context.Context, p domain.Principal, c domai
146	172		}
147	173		m.configMu.Lock()
148	174		defer m.configMu.Unlock()
	175	+	return m.setRepoLocked(ctx, c)
	176	+	}
	177	+
	178	+	// setRepoLocked validates the prospective target, then persists. configMu must be held.
	179	+	func (m *Mirrors) setRepoLocked(ctx context.Context, c domain.MirrorRepoConfig) (domain.MirrorTarget, error) {
	180	+	g, err := m.cfg.Store.GetGlobal(ctx)
	181	+	if err != nil {
	182	+	return domain.MirrorTarget{}, err
	183	+	}
	184	+	if _, err := m.resolve(g, &c, c.Repo); err != nil {
	185	+	return domain.MirrorTarget{}, invalidCron(err)
	186	+	}
149	187		c.UpdatedAt = m.cfg.Clock.Now().UTC()
150	188		if err := m.cfg.Store.PutRepoConfig(ctx, c); err != nil {
151	189		return domain.MirrorTarget{}, err

@@ -174,12 +212,14 @@ func (m *Mirrors) SetRepoExcluded(ctx context.Context, p domain.Principal, repo
174	212		if err := m.admin(ctx, p, repo); err != nil {
175	213		return domain.MirrorTarget{}, err
176	214		}
	215	+	m.configMu.Lock()
	216	+	defer m.configMu.Unlock()
177	217		c, err := m.cfg.Store.GetRepoConfig(ctx, repo)
178	218		if err != nil && !errors.Is(err, domain.ErrNotFound) {
179	219		return domain.MirrorTarget{}, err
180	220		}
181	221		c.Repo, c.Excluded = repo, excluded
182		-	return m.SetRepoMirror(ctx, p, c)
	222	+	return m.setRepoLocked(ctx, c)
183	223		}
184	224
185	225		// GetStatus returns a read-authorized status, redacting administrative diagnostics.

 	}
 	m.configMu.Lock()
 	defer m.configMu.Unlock()
-	g, err := m.cfg.Store.GetGlobal(ctx)
+	old, err := m.cfg.Store.GetGlobal(ctx)
 	if err != nil {
-		return g, err
+		return old, err
 	}
+	g := old
 	if patch.Enabled != nil {
 		g.Enabled = *patch.Enabled
 	}
 		g.AllBranches = *patch.AllBranches
 	}
 	if err := m.validateGlobal(g); err != nil {
-		return g, err
+		return old, err
-	}
-	g.UpdatedAt = m.cfg.Clock.Now().UTC()
-	if err := m.cfg.Store.PutGlobal(ctx, g); err != nil {
-		return g, err
 	}
 	repos, err := m.cfg.Meta.List(ctx)
 	if err != nil {
-		return g, err
+		return old, err
 	}
+	// Resolve every effective target before and after, and persist nothing
+	// unless all of them stay valid.
+	var changed []domain.RepoName
 	for _, r := range repos {
 		c, e := m.cfg.Store.GetRepoConfig(ctx, r.Name)
-		if e != nil && !errors.Is(e, domain.ErrNotFound) {
+		var cp *domain.MirrorRepoConfig
-			return g, e
+		if e == nil {
+			cp = &c
+		} else if !errors.Is(e, domain.ErrNotFound) {
+			return old, e
+		}
+		before, _ := m.resolve(old, cp, r.Name)
+		after, e := m.resolve(g, cp, r.Name)
+		if e != nil {
+			return old, invalidCron(e)
 		}
-		if c.URL != "" {
+		if !sameMirrorTarget(before, after) {
-			continue
+			changed = append(changed, r.Name)
 		}
-		if err := m.reset(ctx, r.Name); err != nil {
+	}
+	g.UpdatedAt = m.cfg.Clock.Now().UTC()
+	if err := m.cfg.Store.PutGlobal(ctx, g); err != nil {
+		return old, err
+	}
+	for _, name := range changed {
+		if err := m.reset(ctx, name); err != nil {
 			return g, err
 		}
 	}
 	return g, nil
 }
+func invalidCron(e error) error {
+	var inv *domain.InvalidFieldsError
+	if errors.As(e, &inv) {
+		return e
+	}
+	return &domain.InvalidFieldsError{Fields: []domain.FieldViolation{{Field: "cron", Reason: e.Error()}}}
+}
+func sameMirrorTarget(a, b domain.MirrorTarget) bool {
+	return a.Enabled == b.Enabled && a.URL == b.URL && a.CronSpec == b.CronSpec && a.AllBranches == b.AllBranches && a.Source == b.Source
+}
 func (m *Mirrors) reset(ctx context.Context, repo domain.RepoName) error {
 	t, err := m.target(ctx, repo)
 	if err != nil {
 	}
 	m.configMu.Lock()
 	defer m.configMu.Unlock()
+	return m.setRepoLocked(ctx, c)
+}
+// setRepoLocked validates the prospective target, then persists. configMu must be held.
+func (m *Mirrors) setRepoLocked(ctx context.Context, c domain.MirrorRepoConfig) (domain.MirrorTarget, error) {
+	g, err := m.cfg.Store.GetGlobal(ctx)
+	if err != nil {
+		return domain.MirrorTarget{}, err
+	}
+	if _, err := m.resolve(g, &c, c.Repo); err != nil {
+		return domain.MirrorTarget{}, invalidCron(err)
+	}
 	c.UpdatedAt = m.cfg.Clock.Now().UTC()
 	if err := m.cfg.Store.PutRepoConfig(ctx, c); err != nil {
 		return domain.MirrorTarget{}, err
 	if err := m.admin(ctx, p, repo); err != nil {
 		return domain.MirrorTarget{}, err
 	}
+	m.configMu.Lock()
+	defer m.configMu.Unlock()
 	c, err := m.cfg.Store.GetRepoConfig(ctx, repo)
 	if err != nil && !errors.Is(err, domain.ErrNotFound) {
 		return domain.MirrorTarget{}, err
 	}
 	c.Repo, c.Excluded = repo, excluded
-	return m.SetRepoMirror(ctx, p, c)
+	return m.setRepoLocked(ctx, c)
 }
 // GetStatus returns a read-authorized status, redacting administrative diagnostics.

M internal/usecase/mirrors_sync.go +10 -6


@@ -14,6 +14,7 @@ import (
14	14		type mirrorFlight struct {
15	15		pending, force, manual bool
16	16		pendingForce bool
	17	+	pendingManual bool
17	18		done chan struct{}
18	19		completed bool
19	20		result ports.MirrorSyncResult

@@ -77,6 +78,7 @@ func (m Mirrors) enqueue(repo domain.RepoName, force, manual bool) (mirrorFlig
77	78		if f := m.flights[repo]; f != nil {
78	79		f.pending = true
79	80		f.pendingForce = f.pendingForce \|\| force
	81	+	f.pendingManual = f.pendingManual \|\| manual
80	82		return f, true, nil
81	83		}
82	84		f := &mirrorFlight{force: force, manual: manual, done: make(chan struct{})}

@@ -106,15 +108,12 @@ func (m *Mirrors) run(repo domain.RepoName) {
106	108		m.mu.Unlock()
107	109		return
108	110		}
109		-	f.force, f.manual = f.pendingForce, false
110		-	f.pending, f.pendingForce = false, false
	111	+	f.force, f.manual = f.pendingForce, f.pendingManual
	112	+	f.pending, f.pendingForce, f.pendingManual = false, false, false
111	113		m.mu.Unlock()
112	114		}
113	115		}
114	116
115		-	func (m *Mirrors) syncOnce(ctx context.Context, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
116		-	return m.attempt(ctx, repo, force, true)
117		-	}
118	117		func (m *Mirrors) attempt(ctx context.Context, repo domain.RepoName, force, manual bool) (ports.MirrorSyncResult, error) {
119	118		m.configMu.RLock()
120	119		defer m.configMu.RUnlock()

@@ -130,7 +129,8 @@ func (m *Mirrors) attempt(ctx context.Context, repo domain.RepoName, force, manu
130	129		if err != nil {
131	130		return ports.MirrorSyncResult{}, err
132	131		}
133		-	now := m.cfg.Clock.Now().UTC()
	132	+	start := m.cfg.Clock.Now().UTC()
	133	+	now := start
134	134		if !t.Enabled {
135	135		s := domain.MirrorDisable(prev, now)
136	136		if prev.State != domain.MirrorDisabled {

@@ -193,7 +193,11 @@ func (m *Mirrors) attempt(ctx context.Context, repo domain.RepoName, force, manu
193	193		runErr = fmt.Errorf("%w: %s", domain.ErrPreconditionFailed, detail)
194	194		}
195	195		}
	196	+	// The push may be slow: completion, retry and success times come from the
	197	+	// clock after it returns; only LastAttemptAt marks when the attempt began.
	198	+	now = m.cfg.Clock.Now().UTC()
196	199		transition := domain.ApplyMirrorOutcome(prev, domain.MirrorOutcome{Err: failure, NoOp: noOp}, now)
	200	+	transition.Status.LastAttemptAt = start
197	201		if failure != nil && m.cfg.Sealer == nil {
198	202		transition.Status.Reason = failure.Detail
199	203		}

 type mirrorFlight struct {
 	pending, force, manual bool
 	pendingForce           bool
+	pendingManual          bool
 	done                   chan struct{}
 	completed              bool
 	result                 ports.MirrorSyncResult
 	if f := m.flights[repo]; f != nil {
 		f.pending = true
 		f.pendingForce = f.pendingForce || force
+		f.pendingManual = f.pendingManual || manual
 		return f, true, nil
 	}
 	f := &mirrorFlight{force: force, manual: manual, done: make(chan struct{})}
 			m.mu.Unlock()
 			return
 		}
-		f.force, f.manual = f.pendingForce, false
+		f.force, f.manual = f.pendingForce, f.pendingManual
-		f.pending, f.pendingForce = false, false
+		f.pending, f.pendingForce, f.pendingManual = false, false, false
 		m.mu.Unlock()
 	}
 }
-func (m *Mirrors) syncOnce(ctx context.Context, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
-	return m.attempt(ctx, repo, force, true)
-}
 func (m *Mirrors) attempt(ctx context.Context, repo domain.RepoName, force, manual bool) (ports.MirrorSyncResult, error) {
 	m.configMu.RLock()
 	defer m.configMu.RUnlock()
 	if err != nil {
 		return ports.MirrorSyncResult{}, err
 	}
-	now := m.cfg.Clock.Now().UTC()
+	start := m.cfg.Clock.Now().UTC()
+	now := start
 	if !t.Enabled {
 		s := domain.MirrorDisable(prev, now)
 		if prev.State != domain.MirrorDisabled {
 			runErr = fmt.Errorf("%w: %s", domain.ErrPreconditionFailed, detail)
 		}
 	}
+	// The push may be slow: completion, retry and success times come from the
+	// clock after it returns; only LastAttemptAt marks when the attempt began.
+	now = m.cfg.Clock.Now().UTC()
 	transition := domain.ApplyMirrorOutcome(prev, domain.MirrorOutcome{Err: failure, NoOp: noOp}, now)
+	transition.Status.LastAttemptAt = start
 	if failure != nil && m.cfg.Sealer == nil {
 		transition.Status.Reason = failure.Detail
 	}

M internal/usecase/mirrors_test.go +162 -0


@@ -432,3 +432,165 @@ func TestMirrorsFileURLTestOption(t *testing.T) {
432	432		})
433	433		}
434	434		}
	435	+
	436	+	func TestMirrorsMergeNotifiesMirror(t *testing.T) {
	437	+	r := newMRRig(t)
	438	+	r.seed()
	439	+	trigger := mocks.NewMirrorTrigger(t)
	440	+	notified := make(chan []domain.RefUpdate, 1)
	441	+	trigger.On("OnPush", mock.Anything, r.repo, mock.Anything).Run(func(a mock.Arguments) { notified <- a.Get(2).([]domain.RefUpdate) }).Return().Once()
	442	+	r.mrs.WithMirror(trigger)
	443	+	head := r.feature("f0")
	444	+	mr := r.open("f0")
	445	+	_, err := r.mrs.Merge(r.ctx, root, r.repo, mr.Number, ports.MergeParams{HeadID: head})
	446	+	require.NoError(t, err)
	447	+	r.mrs.Wait(r.ctx)
	448	+	select {
	449	+	case u := <-notified:
	450	+	require.Equal(t, "refs/heads/main", u[0].Ref)
	451	+	case <-time.After(time.Second):
	452	+	t.Fatal("merge did not notify mirrors")
	453	+	}
	454	+	}
	455	+
	456	+	func TestMirrorsDefaultBranchChangeNotifiesMirror(t *testing.T) {
	457	+	f := newFixture(t)
	458	+	ctx := context.Background()
	459	+	n := name(t, "tools")
	460	+	repo := domain.Repo{Name: n, Visibility: domain.VisibilityPrivate}
	461	+	trigger := mocks.NewMirrorTrigger(t)
	462	+	trigger.On("OnPush", mock.Anything, n, mock.Anything).Return().Once()
	463	+	f.repos.WithMirror(trigger)
	464	+	f.meta.EXPECT().Get(ctx, n).Return(repo, nil)
	465	+	f.git.EXPECT().HasBranch(ctx, n, "dev").Return(true, nil)
	466	+	f.git.EXPECT().SetDefaultBranch(ctx, n, "dev").Return(nil)
	467	+	f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
	468	+	_, err := f.repos.Update(ctx, tokenPrincipal(domain.ScopeRepoAdmin, "tools"), n, ports.UpdateRepoParams{DefaultBranch: ptr("dev")})
	469	+	require.NoError(t, err)
	470	+	}
	471	+
	472	+	func TestMirrorsExternalMutationMakesNextDueTickSync(t *testing.T) {
	473	+	f := newMirrorFixture(t)
	474	+	f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, nil).Once()
	475	+	_, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
	476	+	require.NoError(t, err)
	477	+	f.mu.Lock()
	478	+	f.at = f.at.Add(24 * time.Hour).Truncate(time.Hour).Add(time.Minute)
	479	+	f.mu.Unlock()
	480	+	f.m.ScheduleNow(context.Background())
	481	+	f.advance(4 * time.Minute)
	482	+	f.m.ScheduleNow(context.Background()) // due and clean: skipped
	483	+	f.reader.AssertNumberOfCalls(t, "ListRefs", 1)
	484	+	// A merge and a default-branch change both call OnPush (idempotent dirty mark).
	485	+	f.m.OnPush(context.Background(), f.repo, []domain.RefUpdate{{Ref: "refs/heads/main"}})
	486	+	f.m.OnPush(context.Background(), f.repo, []domain.RefUpdate{{Ref: "HEAD"}})
	487	+	read := make(chan struct{})
	488	+	f.reader.ExpectedCalls = nil
	489	+	f.reader.On("ListRefs", mock.Anything, f.repo).Run(func(mock.Arguments) { close(read) }).Return([]domain.Ref{{Name: "refs/heads/main", ID: mirrorOID(t)}}, nil).Once()
	490	+	f.advance(5 * time.Minute)
	491	+	f.m.ScheduleNow(context.Background())
	492	+	select {
	493	+	case <-read:
	494	+	case <-time.After(time.Second):
	495	+	t.Fatal("dirty repository not synced on the next due tick")
	496	+	}
	497	+	}
	498	+
	499	+	func setupRepoOverrideFixture(t testing.T, global domain.MirrorGlobalConfig) mirrorFixture {
	500	+	f := newMirrorFixture(t)
	501	+	f.store.ExpectedCalls = nil
	502	+	f.store.On("GetGlobal", mock.Anything).Return(global, nil)
	503	+	f.store.On("GetRepoConfig", mock.Anything, f.repo).Return(domain.MirrorRepoConfig{}, domain.ErrNotFound).Maybe()
	504	+	f.store.On("GetStatus", mock.Anything, f.repo).Return(domain.MirrorStatus{Repo: f.repo}, nil).Maybe()
	505	+	return f
	506	+	}
	507	+
	508	+	func TestMirrorsOverrideWithoutCronRejectedWhenGlobalCronEmpty(t *testing.T) {
	509	+	f := setupRepoOverrideFixture(t, domain.MirrorGlobalConfig{})
	510	+	_, err := f.m.SetRepoMirror(context.Background(), domain.Root(), domain.MirrorRepoConfig{Repo: f.repo, URL: "git@example.com:acme/tools"})
	511	+	var fe *domain.InvalidFieldsError
	512	+	require.ErrorAs(t, err, &fe)
	513	+	require.Equal(t, "cron", fe.Fields[0].Field)
	514	+	// PutRepoConfig/PutStatus unexpected: strict mock fails if called.
	515	+	}
	516	+
	517	+	func TestMirrorsGlobalChangeInvalidatingOverrideRejected(t *testing.T) {
	518	+	f := setupRepoOverrideFixture(t, domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.com:acme/{repo}", Cron: "/5 * * *"})
	519	+	f.store.ExpectedCalls = nil
	520	+	f.store.On("GetGlobal", mock.Anything).Return(domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.com:acme/{repo}", Cron: "/5 * * *"}, nil)
	521	+	f.store.On("GetRepoConfig", mock.Anything, f.repo).Return(domain.MirrorRepoConfig{Repo: f.repo, URL: "git@other.example:x/y"}, nil)
	522	+	// Disabling the global with an empty cron leaves the inheriting override with no cron.
	523	+	_, err := f.m.SetGlobalConfig(context.Background(), domain.Root(), ports.MirrorGlobalPatch{Enabled: ptr(false), Cron: ptr("")})
	524	+	var fe *domain.InvalidFieldsError
	525	+	require.ErrorAs(t, err, &fe)
	526	+	}
	527	+
	528	+	func TestMirrorsGlobalChangeResetsInheritingURLOverride(t *testing.T) {
	529	+	f := newMirrorFixture(t)
	530	+	f.store.ExpectedCalls = nil
	531	+	f.store.On("GetGlobal", mock.Anything).Return(domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.com:acme/{repo}", Cron: "/5 * * *"}, nil)
	532	+	f.store.On("GetRepoConfig", mock.Anything, f.repo).Return(domain.MirrorRepoConfig{Repo: f.repo, URL: "git@other.example:x/y"}, nil)
	533	+	f.store.On("PutGlobal", mock.Anything, mock.Anything).Return(nil).Once()
	534	+	f.store.On("GetStatus", mock.Anything, f.repo).Return(domain.MirrorStatus{Repo: f.repo, State: domain.MirrorPaused, Class: domain.MirrorClassDiverged}, nil)
	535	+	reset := make(chan domain.MirrorStatus, 1)
	536	+	f.store.On("PutStatus", mock.Anything, mock.Anything).Run(func(a mock.Arguments) { reset <- a.Get(1).(domain.MirrorStatus) }).Return(nil)
	537	+	_, err := f.m.SetGlobalConfig(context.Background(), domain.Root(), ports.MirrorGlobalPatch{Cron: ptr("/10 * * *")})
	538	+	require.NoError(t, err)
	539	+	require.Equal(t, domain.MirrorOK, (<-reset).State)
	540	+	}
	541	+
	542	+	func TestMirrorsSetRepoExcludedConcurrent(t *testing.T) {
	543	+	f := newMirrorFixture(t)
	544	+	var mu sync.Mutex
	545	+	active, maxActive := 0, 0
	546	+	f.store.On("PutRepoConfig", mock.Anything, mock.Anything).Run(func(mock.Arguments) {
	547	+	mu.Lock()
	548	+	active++
	549	+	if active > maxActive {
	550	+	maxActive = active
	551	+	}
	552	+	mu.Unlock()
	553	+	time.Sleep(2 * time.Millisecond)
	554	+	mu.Lock()
	555	+	active--
	556	+	mu.Unlock()
	557	+	}).Return(nil)
	558	+	var wg sync.WaitGroup
	559	+	for i := 0; i < 8; i++ {
	560	+	wg.Add(1)
	561	+	go func(ex bool) {
	562	+	defer wg.Done()
	563	+	_, err := f.m.SetRepoExcluded(context.Background(), domain.Root(), f.repo, ex)
	564	+	require.NoError(t, err)
	565	+	}(i%2 == 0)
	566	+	}
	567	+	wg.Wait()
	568	+	require.Equal(t, 1, maxActive, "read-modify-write must be serialized")
	569	+	}
	570	+
	571	+	func TestMirrorsCoalescedManualAppliesThrottle(t *testing.T) {
	572	+	f := newMirrorFixture(t)
	573	+	entered, release := make(chan struct{}), make(chan struct{})
	574	+	f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { close(entered); <-release }).Return(ports.MirrorPushResult{}, nil).Once()
	575	+	done := make(chan error, 1)
	576	+	go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
	577	+	<-entered
	578	+	v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
	579	+	require.NoError(t, err)
	580	+	require.True(t, v.Queued)
	581	+	close(release)
	582	+	require.NoError(t, <-done)
	583	+	// The rerun is manual and not forced: the throttle (just attempted) stops it, so no second push.
	584	+	time.Sleep(200 * time.Millisecond)
	585	+	f.push.AssertNumberOfCalls(t, "Push", 1)
	586	+	}
	587	+
	588	+	func TestMirrorsSlowPushTimestamps(t *testing.T) {
	589	+	f := newMirrorFixture(t)
	590	+	start := f.at
	591	+	f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { f.advance(10 * time.Minute) }).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Once()
	592	+	v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
	593	+	require.Error(t, err)
	594	+	require.True(t, v.Status.LastAttemptAt.Equal(start))
	595	+	require.True(t, v.Status.NextRetryAt.Equal(start.Add(10*time.Minute+domain.MirrorBackoff(1))))
	596	+	}


@@ -432,3 +432,165 @@ func TestMirrorsFileURLTestOption(t *testing.T) {
432	})	432		})
433	}	433		}
434	}	434		}
		435	+
		436	+	func TestMirrorsMergeNotifiesMirror(t *testing.T) {
		437	+	r := newMRRig(t)
		438	+	r.seed()
		439	+	trigger := mocks.NewMirrorTrigger(t)
		440	+	notified := make(chan []domain.RefUpdate, 1)
		441	+	trigger.On("OnPush", mock.Anything, r.repo, mock.Anything).Run(func(a mock.Arguments) { notified <- a.Get(2).([]domain.RefUpdate) }).Return().Once()
		442	+	r.mrs.WithMirror(trigger)
		443	+	head := r.feature("f0")
		444	+	mr := r.open("f0")
		445	+	_, err := r.mrs.Merge(r.ctx, root, r.repo, mr.Number, ports.MergeParams{HeadID: head})
		446	+	require.NoError(t, err)
		447	+	r.mrs.Wait(r.ctx)
		448	+	select {
		449	+	case u := <-notified:
		450	+	require.Equal(t, "refs/heads/main", u[0].Ref)
		451	+	case <-time.After(time.Second):
		452	+	t.Fatal("merge did not notify mirrors")
		453	+	}
		454	+	}
		455	+
		456	+	func TestMirrorsDefaultBranchChangeNotifiesMirror(t *testing.T) {
		457	+	f := newFixture(t)
		458	+	ctx := context.Background()
		459	+	n := name(t, "tools")
		460	+	repo := domain.Repo{Name: n, Visibility: domain.VisibilityPrivate}
		461	+	trigger := mocks.NewMirrorTrigger(t)
		462	+	trigger.On("OnPush", mock.Anything, n, mock.Anything).Return().Once()
		463	+	f.repos.WithMirror(trigger)
		464	+	f.meta.EXPECT().Get(ctx, n).Return(repo, nil)
		465	+	f.git.EXPECT().HasBranch(ctx, n, "dev").Return(true, nil)
		466	+	f.git.EXPECT().SetDefaultBranch(ctx, n, "dev").Return(nil)
		467	+	f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
		468	+	_, err := f.repos.Update(ctx, tokenPrincipal(domain.ScopeRepoAdmin, "tools"), n, ports.UpdateRepoParams{DefaultBranch: ptr("dev")})
		469	+	require.NoError(t, err)
		470	+	}
		471	+
		472	+	func TestMirrorsExternalMutationMakesNextDueTickSync(t *testing.T) {
		473	+	f := newMirrorFixture(t)
		474	+	f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, nil).Once()
		475	+	_, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
		476	+	require.NoError(t, err)
		477	+	f.mu.Lock()
		478	+	f.at = f.at.Add(24 * time.Hour).Truncate(time.Hour).Add(time.Minute)
		479	+	f.mu.Unlock()
		480	+	f.m.ScheduleNow(context.Background())
		481	+	f.advance(4 * time.Minute)
		482	+	f.m.ScheduleNow(context.Background()) // due and clean: skipped
		483	+	f.reader.AssertNumberOfCalls(t, "ListRefs", 1)
		484	+	// A merge and a default-branch change both call OnPush (idempotent dirty mark).
		485	+	f.m.OnPush(context.Background(), f.repo, []domain.RefUpdate{{Ref: "refs/heads/main"}})
		486	+	f.m.OnPush(context.Background(), f.repo, []domain.RefUpdate{{Ref: "HEAD"}})
		487	+	read := make(chan struct{})
		488	+	f.reader.ExpectedCalls = nil
		489	+	f.reader.On("ListRefs", mock.Anything, f.repo).Run(func(mock.Arguments) { close(read) }).Return([]domain.Ref{{Name: "refs/heads/main", ID: mirrorOID(t)}}, nil).Once()
		490	+	f.advance(5 * time.Minute)
		491	+	f.m.ScheduleNow(context.Background())
		492	+	select {
		493	+	case <-read:
		494	+	case <-time.After(time.Second):
		495	+	t.Fatal("dirty repository not synced on the next due tick")
		496	+	}
		497	+	}
		498	+
		499	+	func setupRepoOverrideFixture(t testing.T, global domain.MirrorGlobalConfig) mirrorFixture {
		500	+	f := newMirrorFixture(t)
		501	+	f.store.ExpectedCalls = nil
		502	+	f.store.On("GetGlobal", mock.Anything).Return(global, nil)
		503	+	f.store.On("GetRepoConfig", mock.Anything, f.repo).Return(domain.MirrorRepoConfig{}, domain.ErrNotFound).Maybe()
		504	+	f.store.On("GetStatus", mock.Anything, f.repo).Return(domain.MirrorStatus{Repo: f.repo}, nil).Maybe()
		505	+	return f
		506	+	}
		507	+
		508	+	func TestMirrorsOverrideWithoutCronRejectedWhenGlobalCronEmpty(t *testing.T) {
		509	+	f := setupRepoOverrideFixture(t, domain.MirrorGlobalConfig{})
		510	+	_, err := f.m.SetRepoMirror(context.Background(), domain.Root(), domain.MirrorRepoConfig{Repo: f.repo, URL: "git@example.com:acme/tools"})
		511	+	var fe *domain.InvalidFieldsError
		512	+	require.ErrorAs(t, err, &fe)
		513	+	require.Equal(t, "cron", fe.Fields[0].Field)
		514	+	// PutRepoConfig/PutStatus unexpected: strict mock fails if called.
		515	+	}
		516	+
		517	+	func TestMirrorsGlobalChangeInvalidatingOverrideRejected(t *testing.T) {
		518	+	f := setupRepoOverrideFixture(t, domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.com:acme/{repo}", Cron: "/5 * * *"})
		519	+	f.store.ExpectedCalls = nil
		520	+	f.store.On("GetGlobal", mock.Anything).Return(domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.com:acme/{repo}", Cron: "/5 * * *"}, nil)
		521	+	f.store.On("GetRepoConfig", mock.Anything, f.repo).Return(domain.MirrorRepoConfig{Repo: f.repo, URL: "git@other.example:x/y"}, nil)
		522	+	// Disabling the global with an empty cron leaves the inheriting override with no cron.
		523	+	_, err := f.m.SetGlobalConfig(context.Background(), domain.Root(), ports.MirrorGlobalPatch{Enabled: ptr(false), Cron: ptr("")})
		524	+	var fe *domain.InvalidFieldsError
		525	+	require.ErrorAs(t, err, &fe)
		526	+	}
		527	+
		528	+	func TestMirrorsGlobalChangeResetsInheritingURLOverride(t *testing.T) {
		529	+	f := newMirrorFixture(t)
		530	+	f.store.ExpectedCalls = nil
		531	+	f.store.On("GetGlobal", mock.Anything).Return(domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.com:acme/{repo}", Cron: "/5 * * *"}, nil)
		532	+	f.store.On("GetRepoConfig", mock.Anything, f.repo).Return(domain.MirrorRepoConfig{Repo: f.repo, URL: "git@other.example:x/y"}, nil)
		533	+	f.store.On("PutGlobal", mock.Anything, mock.Anything).Return(nil).Once()
		534	+	f.store.On("GetStatus", mock.Anything, f.repo).Return(domain.MirrorStatus{Repo: f.repo, State: domain.MirrorPaused, Class: domain.MirrorClassDiverged}, nil)
		535	+	reset := make(chan domain.MirrorStatus, 1)
		536	+	f.store.On("PutStatus", mock.Anything, mock.Anything).Run(func(a mock.Arguments) { reset <- a.Get(1).(domain.MirrorStatus) }).Return(nil)
		537	+	_, err := f.m.SetGlobalConfig(context.Background(), domain.Root(), ports.MirrorGlobalPatch{Cron: ptr("/10 * * *")})
		538	+	require.NoError(t, err)
		539	+	require.Equal(t, domain.MirrorOK, (<-reset).State)
		540	+	}
		541	+
		542	+	func TestMirrorsSetRepoExcludedConcurrent(t *testing.T) {
		543	+	f := newMirrorFixture(t)
		544	+	var mu sync.Mutex
		545	+	active, maxActive := 0, 0
		546	+	f.store.On("PutRepoConfig", mock.Anything, mock.Anything).Run(func(mock.Arguments) {
		547	+	mu.Lock()
		548	+	active++
		549	+	if active > maxActive {
		550	+	maxActive = active
		551	+	}
		552	+	mu.Unlock()
		553	+	time.Sleep(2 * time.Millisecond)
		554	+	mu.Lock()
		555	+	active--
		556	+	mu.Unlock()
		557	+	}).Return(nil)
		558	+	var wg sync.WaitGroup
		559	+	for i := 0; i < 8; i++ {
		560	+	wg.Add(1)
		561	+	go func(ex bool) {
		562	+	defer wg.Done()
		563	+	_, err := f.m.SetRepoExcluded(context.Background(), domain.Root(), f.repo, ex)
		564	+	require.NoError(t, err)
		565	+	}(i%2 == 0)
		566	+	}
		567	+	wg.Wait()
		568	+	require.Equal(t, 1, maxActive, "read-modify-write must be serialized")
		569	+	}
		570	+
		571	+	func TestMirrorsCoalescedManualAppliesThrottle(t *testing.T) {
		572	+	f := newMirrorFixture(t)
		573	+	entered, release := make(chan struct{}), make(chan struct{})
		574	+	f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { close(entered); <-release }).Return(ports.MirrorPushResult{}, nil).Once()
		575	+	done := make(chan error, 1)
		576	+	go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
		577	+	<-entered
		578	+	v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
		579	+	require.NoError(t, err)
		580	+	require.True(t, v.Queued)
		581	+	close(release)
		582	+	require.NoError(t, <-done)
		583	+	// The rerun is manual and not forced: the throttle (just attempted) stops it, so no second push.
		584	+	time.Sleep(200 * time.Millisecond)
		585	+	f.push.AssertNumberOfCalls(t, "Push", 1)
		586	+	}
		587	+
		588	+	func TestMirrorsSlowPushTimestamps(t *testing.T) {
		589	+	f := newMirrorFixture(t)
		590	+	start := f.at
		591	+	f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { f.advance(10 * time.Minute) }).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Once()
		592	+	v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
		593	+	require.Error(t, err)
		594	+	require.True(t, v.Status.LastAttemptAt.Equal(start))
		595	+	require.True(t, v.Status.NextRetryAt.Equal(start.Add(10*time.Minute+domain.MirrorBackoff(1))))
		596	+	}

M internal/usecase/repos.go +13 -5


@@ -17,11 +17,12 @@ import (
17	17
18	18		// Repos are the repository use cases.
19	19		type Repos struct {
20		-	meta ports.RepoMetaStore
21		-	git ports.GitRepos
22		-	bus ports.EventBus
23		-	clock ports.Clock
24		-	ci ports.CIRunCleaner // nil: no CI logs to remove
	20	+	meta ports.RepoMetaStore
	21	+	git ports.GitRepos
	22	+	bus ports.EventBus
	23	+	clock ports.Clock
	24	+	ci ports.CIRunCleaner // nil: no CI logs to remove
	25	+	mirror ports.MirrorTrigger // nil: no outbound mirror
25	26		}
26	27
27	28		var _ ports.RepoService = (*Repos)(nil)

@@ -32,6 +33,13 @@ func NewRepos(meta ports.RepoMetaStore, git ports.GitRepos, bus ports.EventBus,
32	33		return &Repos{meta: meta, git: git, bus: bus, clock: clock}
33	34		}
34	35
	36	+	// WithMirror makes a default-branch change notify outbound mirrors. Call it
	37	+	// before the first request; it returns r.
	38	+	func (r Repos) WithMirror(t ports.MirrorTrigger) Repos {
	39	+	r.mirror = t
	40	+	return r
	41	+	}
	42	+
35	43		// WithCIRunCleaner makes Delete remove the CI logs of the repository. Call it
36	44		// before the first request; it returns r.
37	45		func (r Repos) WithCIRunCleaner(c ports.CIRunCleaner) Repos {

 // Repos are the repository use cases.
 type Repos struct {
-	meta  ports.RepoMetaStore
+	meta   ports.RepoMetaStore
-	git   ports.GitRepos
+	git    ports.GitRepos
-	bus   ports.EventBus
+	bus    ports.EventBus
-	clock ports.Clock
+	clock  ports.Clock
-	ci    ports.CIRunCleaner // nil: no CI logs to remove
+	ci     ports.CIRunCleaner  // nil: no CI logs to remove
+	mirror ports.MirrorTrigger // nil: no outbound mirror
 }
 var _ ports.RepoService = (*Repos)(nil)
 	return &Repos{meta: meta, git: git, bus: bus, clock: clock}
 }
+// WithMirror makes a default-branch change notify outbound mirrors. Call it
+// before the first request; it returns r.
+func (r *Repos) WithMirror(t ports.MirrorTrigger) *Repos {
+	r.mirror = t
+	return r
+}
 // WithCIRunCleaner makes Delete remove the CI logs of the repository. Call it
 // before the first request; it returns r.
 func (r *Repos) WithCIRunCleaner(c ports.CIRunCleaner) *Repos {

M internal/usecase/repos_write.go +3 -0


@@ -129,6 +129,9 @@ func (r *Repos) Update(ctx context.Context, p domain.Principal, name domain.Repo
129	129		if !metaDone {
130	130		return domain.Repo{}, headErr
131	131		}
	132	+	} else if r.mirror != nil {
	133	+	// The default branch decides which refs are mirrored.
	134	+	r.mirror.OnPush(ctx, name, []domain.RefUpdate{{Ref: "HEAD"}})
132	135		}
133	136		}
134	137


@@ -129,6 +129,9 @@ func (r *Repos) Update(ctx context.Context, p domain.Principal, name domain.Repo
129	if !metaDone {	129		if !metaDone {
130	return domain.Repo{}, headErr	130		return domain.Repo{}, headErr
131	}	131		}
		132	+	} else if r.mirror != nil {
		133	+	// The default branch decides which refs are mirrored.
		134	+	r.mirror.OnPush(ctx, name, []domain.RefUpdate{{Ref: "HEAD"}})
132	}	135		}
133	}	136		}
134		137