6d7a853 fix(app): record a push when a merge moves the target branch · ermite

M cmd/ermite/forge.go +1 -1


@@ -184,7 +184,7 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
184	184		f.SigningKey = app.NewSigningKey(signer, committer)
185	185		f.MergeRequests = app.NewMergeRequests(app.MergeRequestsConfig{
186	186		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,
187		-	Committer: committer, Bus: f.Bus, Clock: clock,
	187	+	Committer: committer, Bus: f.Bus, Clock: clock, Maint: f.maint,
188	188		})
189	189		log.Info().Str("fingerprint", signer.PublicKey().Fingerprint).Msg("commit signing key ready")
190	190


@@ -184,7 +184,7 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
184		f.SigningKey = app.NewSigningKey(signer, committer)	184		f.SigningKey = app.NewSigningKey(signer, committer)
185		f.MergeRequests = app.NewMergeRequests(app.MergeRequestsConfig{	185		f.MergeRequests = app.NewMergeRequests(app.MergeRequestsConfig{
186		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,	186		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,
187	-	Committer: committer, Bus: f.Bus, Clock: clock,	187	+	Committer: committer, Bus: f.Bus, Clock: clock, Maint: f.maint,
188		})	188		})
189		log.Info().Str("fingerprint", signer.PublicKey().Fingerprint).Msg("commit signing key ready")	189		log.Info().Str("fingerprint", signer.PublicKey().Fingerprint).Msg("commit signing key ready")
190			190

M internal/app/mergerequests.go +18 -1


@@ -42,6 +42,7 @@ type MergeRequests struct {
42	42		committer domain.Identity
43	43		bus ports.EventBus
44	44		clock ports.Clock
	45	+	maint MaintenanceScheduler
45	46
46	47		mu sync.Mutex
47	48		locks map[domain.RepoName]chan struct{}

@@ -61,13 +62,15 @@ type MergeRequestsConfig struct {
61	62		Committer domain.Identity
62	63		Bus ports.EventBus
63	64		Clock ports.Clock
	65	+	// Maint queues the maintenance that follows a merge, as it does a push.
	66	+	Maint MaintenanceScheduler
64	67		}
65	68
66	69		// NewMergeRequests builds the merge request use cases.
67	70		func NewMergeRequests(cfg MergeRequestsConfig) *MergeRequests {
68	71		return &MergeRequests{
69	72		meta: cfg.Meta, store: cfg.Store, repos: cfg.Repos, git: cfg.Git, signer: cfg.Signer,
70		-	committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock,
	73	+	committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock, maint: cfg.Maint,
71	74		locks: make(map[domain.RepoName]chan struct{}),
72	75		}
73	76		}

@@ -788,6 +791,7 @@ func (m *MergeRequests) Merge(ctx context.Context, p domain.Principal, name doma
788	791		if err != nil {
789	792		return domain.MergeRequest{}, err
790	793		}
	794	+	m.recordPush(ctx, name) // the target moved, whatever happens next
791	795		mr, err = m.markMerged(ctx, name, number, mode, tip, head, merged)
792	796		if err != nil {
793	797		return domain.MergeRequest{}, err

@@ -805,6 +809,19 @@ func (m *MergeRequests) Merge(ctx context.Context, p domain.Principal, name doma
805	809		return mr, nil
806	810		}
807	811
	812	+	// recordPush does for a merge what Pushes.Record does for a push, as the merge
	813	+	// moved a branch: it sets pushed_at and queues the maintenance of the
	814	+	// repository. The target already moved, so a failure is logged, not returned,
	815	+	// and the caller giving up does not stop it.
	816	+	func (m *MergeRequests) recordPush(ctx context.Context, name domain.RepoName) {
	817	+	ctx = context.WithoutCancel(ctx)
	818	+	if _, err := touchPushedAt(ctx, m.meta, name, m.clock.Now().UTC()); err != nil {
	819	+	log := zerowrap.FromCtx(ctx)
	820	+	log.Error().Err(err).Str("repo", name.String()).Msg("record pushed_at after a merge")
	821	+	}
	822	+	m.maint.Schedule(ctx, name)
	823	+	}
	824	+
808	825		// moveTarget moves the target branch from tip to the head (fast-forward) or to
809	826		// a new signed squash commit, and returns the new tip.
810	827		func (m *MergeRequests) moveTarget(ctx context.Context, mr domain.MergeRequest, mode domain.MergeMode, tip, head domain.ObjectID) (domain.ObjectID, error) {

 	committer domain.Identity
 	bus       ports.EventBus
 	clock     ports.Clock
+	maint     MaintenanceScheduler
 	mu    sync.Mutex
 	locks map[domain.RepoName]chan struct{}
 	Committer domain.Identity
 	Bus       ports.EventBus
 	Clock     ports.Clock
+	// Maint queues the maintenance that follows a merge, as it does a push.
+	Maint MaintenanceScheduler
 }
 // NewMergeRequests builds the merge request use cases.
 func NewMergeRequests(cfg MergeRequestsConfig) *MergeRequests {
 	return &MergeRequests{
 		meta: cfg.Meta, store: cfg.Store, repos: cfg.Repos, git: cfg.Git, signer: cfg.Signer,
-		committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock,
+		committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock, maint: cfg.Maint,
 		locks: make(map[domain.RepoName]chan struct{}),
 	}
 }
 	if err != nil {
 		return domain.MergeRequest{}, err
 	}
+	m.recordPush(ctx, name) // the target moved, whatever happens next
 	mr, err = m.markMerged(ctx, name, number, mode, tip, head, merged)
 	if err != nil {
 		return domain.MergeRequest{}, err
 	return mr, nil
 }
+// recordPush does for a merge what Pushes.Record does for a push, as the merge
+// moved a branch: it sets pushed_at and queues the maintenance of the
+// repository. The target already moved, so a failure is logged, not returned,
+// and the caller giving up does not stop it.
+func (m *MergeRequests) recordPush(ctx context.Context, name domain.RepoName) {
+	ctx = context.WithoutCancel(ctx)
+	if _, err := touchPushedAt(ctx, m.meta, name, m.clock.Now().UTC()); err != nil {
+		log := zerowrap.FromCtx(ctx)
+		log.Error().Err(err).Str("repo", name.String()).Msg("record pushed_at after a merge")
+	}
+	m.maint.Schedule(ctx, name)
+}
 // moveTarget moves the target branch from tip to the head (fast-forward) or to
 // a new signed squash commit, and returns the new tip.
 func (m *MergeRequests) moveTarget(ctx context.Context, mr domain.MergeRequest, mode domain.MergeMode, tip, head domain.ObjectID) (domain.ObjectID, error) {

M internal/app/mergerequests_integration_test.go +28 -2


@@ -39,6 +39,7 @@ type mrRig struct {
39	39		store *kvstore.Store
40	40		git *gitcli.Git
41	41		bus *eventbus.Bus
	42	+	sched *scheduler
42	43		events <-chan domain.Event
43	44		work string
44	45		repo domain.RepoName

@@ -74,7 +75,7 @@ func newMRRig(t testing.T) mrRig {
74	75		bus := eventbus.NewWithBuffer(1024)
75	76		committer := domain.Identity{Name: "ermite", Email: "ermite@git.example.org"}
76	77		r := &mrRig{
77		-	t: t, ctx: ctx, data: data, store: store, git: git, bus: bus, events: bus.Subscribe(ctx),
	78	+	t: t, ctx: ctx, data: data, store: store, git: git, bus: bus, sched: &scheduler{}, events: bus.Subscribe(ctx),
78	79		repos: app.NewRepos(store, git, bus, fixedClock{}),
79	80		key: app.NewSigningKey(signer, committer),
80	81		work: t.TempDir(),

@@ -82,7 +83,7 @@ func newMRRig(t testing.T) mrRig {
82	83		}
83	84		r.mrs = app.NewMergeRequests(app.MergeRequestsConfig{
84	85		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,
85		-	Committer: committer, Bus: bus, Clock: fixedClock{},
	86	+	Committer: committer, Bus: bus, Clock: fixedClock{}, Maint: r.sched,
86	87		})
87	88		_, err = r.repos.Create(ctx, root, app.CreateParams{Name: r.repo, Visibility: domain.VisibilityPrivate})
88	89		require.NoError(t, err)

@@ -1224,3 +1225,28 @@ func TestAgentPresetWorksOnItsPrivateRepository(t *testing.T) {
1224	1225		_, err = r.mrs.Create(r.ctx, elsewhere, r.repo, app.CreateMergeRequestParams{Title: "t", Source: "feat"})
1225	1226		assert.ErrorIs(t, err, domain.ErrNotFound)
1226	1227		}
	1228	+
	1229	+	// GO-006: a merge moves the target branch, so it records pushed_at and queues
	1230	+	// the maintenance of the repository, as a push does.
	1231	+	func TestMergeRecordsAPush(t *testing.T) {
	1232	+	for _, mode := range []domain.MergeMode{domain.MergeModeFastForward, domain.MergeModeSquash} {
	1233	+	t.Run(string(mode), func(t *testing.T) {
	1234	+	r := newMRRig(t)
	1235	+	r.seed()
	1236	+	head := r.feature("feat")
	1237	+	mr := r.open("feat")
	1238	+	before, err := r.repos.Get(r.ctx, root, r.repo)
	1239	+	require.NoError(t, err)
	1240	+	require.True(t, before.PushedAt.IsZero(), "no push recorded yet")
	1241	+	assert.Empty(t, r.sched.names)
	1242	+
	1243	+	_, err = r.mrs.Merge(r.ctx, mrM, r.repo, mr.Number, app.MergeParams{Mode: mode, HeadID: head})
	1244	+	require.NoError(t, err)
	1245	+
	1246	+	after, err := r.repos.Get(r.ctx, root, r.repo)
	1247	+	require.NoError(t, err)
	1248	+	assert.Equal(t, now, after.PushedAt)
	1249	+	assert.Equal(t, []domain.RepoName{r.repo}, r.sched.names, "maintenance queued")
	1250	+	})
	1251	+	}
	1252	+	}


@@ -39,6 +39,7 @@ type mrRig struct {
39		store *kvstore.Store	39		store *kvstore.Store
40		git *gitcli.Git	40		git *gitcli.Git
41		bus *eventbus.Bus	41		bus *eventbus.Bus
			42	+	sched *scheduler
42		events <-chan domain.Event	43		events <-chan domain.Event
43		work string	44		work string
44		repo domain.RepoName	45		repo domain.RepoName

@@ -74,7 +75,7 @@ func newMRRig(t testing.T) mrRig {
74		bus := eventbus.NewWithBuffer(1024)	75		bus := eventbus.NewWithBuffer(1024)
75		committer := domain.Identity{Name: "ermite", Email: "ermite@git.example.org"}	76		committer := domain.Identity{Name: "ermite", Email: "ermite@git.example.org"}
76		r := &mrRig{	77		r := &mrRig{
77	-	t: t, ctx: ctx, data: data, store: store, git: git, bus: bus, events: bus.Subscribe(ctx),	78	+	t: t, ctx: ctx, data: data, store: store, git: git, bus: bus, sched: &scheduler{}, events: bus.Subscribe(ctx),
78		repos: app.NewRepos(store, git, bus, fixedClock{}),	79		repos: app.NewRepos(store, git, bus, fixedClock{}),
79		key: app.NewSigningKey(signer, committer),	80		key: app.NewSigningKey(signer, committer),
80		work: t.TempDir(),	81		work: t.TempDir(),

@@ -82,7 +83,7 @@ func newMRRig(t testing.T) mrRig {
82		}	83		}
83		r.mrs = app.NewMergeRequests(app.MergeRequestsConfig{	84		r.mrs = app.NewMergeRequests(app.MergeRequestsConfig{
84		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,	85		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,
85	-	Committer: committer, Bus: bus, Clock: fixedClock{},	86	+	Committer: committer, Bus: bus, Clock: fixedClock{}, Maint: r.sched,
86		})	87		})
87		_, err = r.repos.Create(ctx, root, app.CreateParams{Name: r.repo, Visibility: domain.VisibilityPrivate})	88		_, err = r.repos.Create(ctx, root, app.CreateParams{Name: r.repo, Visibility: domain.VisibilityPrivate})
88		require.NoError(t, err)	89		require.NoError(t, err)

@@ -1224,3 +1225,28 @@ func TestAgentPresetWorksOnItsPrivateRepository(t *testing.T) {
1224		_, err = r.mrs.Create(r.ctx, elsewhere, r.repo, app.CreateMergeRequestParams{Title: "t", Source: "feat"})	1225		_, err = r.mrs.Create(r.ctx, elsewhere, r.repo, app.CreateMergeRequestParams{Title: "t", Source: "feat"})
1225		assert.ErrorIs(t, err, domain.ErrNotFound)	1226		assert.ErrorIs(t, err, domain.ErrNotFound)
1226		}	1227		}
			1228	+
			1229	+	// GO-006: a merge moves the target branch, so it records pushed_at and queues
			1230	+	// the maintenance of the repository, as a push does.
			1231	+	func TestMergeRecordsAPush(t *testing.T) {
			1232	+	for _, mode := range []domain.MergeMode{domain.MergeModeFastForward, domain.MergeModeSquash} {
			1233	+	t.Run(string(mode), func(t *testing.T) {
			1234	+	r := newMRRig(t)
			1235	+	r.seed()
			1236	+	head := r.feature("feat")
			1237	+	mr := r.open("feat")
			1238	+	before, err := r.repos.Get(r.ctx, root, r.repo)
			1239	+	require.NoError(t, err)
			1240	+	require.True(t, before.PushedAt.IsZero(), "no push recorded yet")
			1241	+	assert.Empty(t, r.sched.names)
			1242	+
			1243	+	_, err = r.mrs.Merge(r.ctx, mrM, r.repo, mr.Number, app.MergeParams{Mode: mode, HeadID: head})
			1244	+	require.NoError(t, err)
			1245	+
			1246	+	after, err := r.repos.Get(r.ctx, root, r.repo)
			1247	+	require.NoError(t, err)
			1248	+	assert.Equal(t, now, after.PushedAt)
			1249	+	assert.Equal(t, []domain.RepoName{r.repo}, r.sched.names, "maintenance queued")
			1250	+	})
			1251	+	}
			1252	+	}

M internal/app/mergerequests_test.go +1 -1


@@ -38,7 +38,7 @@ func newMRUnit(t testing.T) mrUnit {
38	38		clock.EXPECT().Now().Return(now).Maybe()
39	39		u.mrs = app.NewMergeRequests(app.MergeRequestsConfig{
40	40		Meta: u.meta, Store: u.store, Repos: u.repos, Git: u.git,
41		-	Signer: mocks.NewCommitSigner(t), Bus: mocks.NewEventBus(t), Clock: clock,
	41	+	Signer: mocks.NewCommitSigner(t), Bus: mocks.NewEventBus(t), Clock: clock, Maint: &scheduler{},
42	42		})
43	43		return u
44	44		}


@@ -38,7 +38,7 @@ func newMRUnit(t testing.T) mrUnit {
38		clock.EXPECT().Now().Return(now).Maybe()	38		clock.EXPECT().Now().Return(now).Maybe()
39		u.mrs = app.NewMergeRequests(app.MergeRequestsConfig{	39		u.mrs = app.NewMergeRequests(app.MergeRequestsConfig{
40		Meta: u.meta, Store: u.store, Repos: u.repos, Git: u.git,	40		Meta: u.meta, Store: u.store, Repos: u.repos, Git: u.git,
41	-	Signer: mocks.NewCommitSigner(t), Bus: mocks.NewEventBus(t), Clock: clock,	41	+	Signer: mocks.NewCommitSigner(t), Bus: mocks.NewEventBus(t), Clock: clock, Maint: &scheduler{},
42		})	42		})
43		return u	43		return u
44		}	44		}

M internal/app/pushes.go +6 -5


@@ -126,7 +126,7 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep
126	126		now := u.clock.Now().UTC()
127	127		var errs []error
128	128
129		-	repo, err := u.touchPushedAt(ctx, name, now)
	129	+	repo, err := touchPushedAt(ctx, u.meta, name, now)
130	130		if err != nil {
131	131		errs = append(errs, err)
132	132		}

@@ -147,10 +147,11 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep
147	147
148	148		// touchPushedAt stores pushed_at, retrying when another write got in between.
149	149		// A repository deleted meanwhile is not an error. It returns the repository
150		-	// as read, or the zero value when it is unknown.
151		-	func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now time.Time) (domain.Repo, error) {
	150	+	// as read, or the zero value when it is unknown. A push (Pushes.Record) and a
	151	+	// merge, which moves a branch too, both record through it.
	152	+	func touchPushedAt(ctx context.Context, meta ports.RepoMetaStore, name domain.RepoName, now time.Time) (domain.Repo, error) {
152	153		for range pushRetries {
153		-	repo, err := u.meta.Get(ctx, name)
	154	+	repo, err := meta.Get(ctx, name)
154	155		if errors.Is(err, domain.ErrNotFound) {
155	156		return domain.Repo{}, nil
156	157		}

@@ -161,7 +162,7 @@ func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now ti
161	162		return repo, nil // a later push already recorded
162	163		}
163	164		repo.PushedAt = now
164		-	err = u.meta.Update(ctx, repo)
	165	+	err = meta.Update(ctx, repo)
165	166		if errors.Is(err, domain.ErrConflict) {
166	167		continue
167	168		}

 	now := u.clock.Now().UTC()
 	var errs []error
-	repo, err := u.touchPushedAt(ctx, name, now)
+	repo, err := touchPushedAt(ctx, u.meta, name, now)
 	if err != nil {
 		errs = append(errs, err)
 	}
 // touchPushedAt stores pushed_at, retrying when another write got in between.
 // A repository deleted meanwhile is not an error. It returns the repository
-// as read, or the zero value when it is unknown.
+// as read, or the zero value when it is unknown. A push (Pushes.Record) and a
-func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now time.Time) (domain.Repo, error) {
+// merge, which moves a branch too, both record through it.
+func touchPushedAt(ctx context.Context, meta ports.RepoMetaStore, name domain.RepoName, now time.Time) (domain.Repo, error) {
 	for range pushRetries {
-		repo, err := u.meta.Get(ctx, name)
+		repo, err := meta.Get(ctx, name)
 		if errors.Is(err, domain.ErrNotFound) {
 			return domain.Repo{}, nil
 		}
 			return repo, nil // a later push already recorded
 		}
 		repo.PushedAt = now
-		err = u.meta.Update(ctx, repo)
+		err = meta.Update(ctx, repo)
 		if errors.Is(err, domain.ErrConflict) {
 			continue
 		}