9ca7c68 fix(repos): mutate metadata atomically and record approved issue activity · ermite

M internal/adapters/kvstore/issues_test.go +4 -2


@@ -258,13 +258,15 @@ func TestRepoSettingsKeepTheIssuesMode(t *testing.T) {
258	258		assert.Equal(t, domain.IssuesOff, got.Settings.Issues)
259	259		for _, m := range []domain.IssuesMode{domain.IssuesOwner, domain.IssuesPublic, domain.IssuesOff} {
260	260		got.Settings.Issues = m
261		-	require.NoError(t, s.Update(ctx, got))
	261	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Settings = got.Settings; return true, nil })
	262	+	require.NoError(t, err)
262	263		again, err := s.Get(ctx, n)
263	264		require.NoError(t, err)
264	265		assert.Equal(t, m, again.Settings.Issues)
265	266		}
266	267		got.Settings.Issues = "everyone"
267		-	assert.ErrorIs(t, s.Update(ctx, got), domain.ErrInvalid)
	268	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Settings = got.Settings; return true, nil })
	269	+	assert.ErrorIs(t, err, domain.ErrInvalid)
268	270		}
269	271
270	272		// TestSchema4UpgradesTheTokensWithAllScopes: the tokens that held every

 	assert.Equal(t, domain.IssuesOff, got.Settings.Issues)
 	for _, m := range []domain.IssuesMode{domain.IssuesOwner, domain.IssuesPublic, domain.IssuesOff} {
 		got.Settings.Issues = m
-		require.NoError(t, s.Update(ctx, got))
+		_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Settings = got.Settings; return true, nil })
+		require.NoError(t, err)
 		again, err := s.Get(ctx, n)
 		require.NoError(t, err)
 		assert.Equal(t, m, again.Settings.Issues)
 	}
 	got.Settings.Issues = "everyone"
-	assert.ErrorIs(t, s.Update(ctx, got), domain.ErrInvalid)
+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Settings = got.Settings; return true, nil })
+	assert.ErrorIs(t, err, domain.ErrInvalid)
 }
 // TestSchema4UpgradesTheTokensWithAllScopes: the tokens that held every

M internal/adapters/kvstore/mergerequests_test.go +4 -2


@@ -442,13 +442,15 @@ func TestRepoSettings(t *testing.T) {
442	442		assert.Equal(t, domain.DefaultRepoSettings(), got.Settings)
443	443
444	444		got.Settings = domain.RepoSettings{AllowedMergeModes: []domain.MergeMode{domain.MergeModeSquash}, FailingChecksBlock: false, Issues: domain.IssuesOwner}
445		-	require.NoError(t, s.Update(ctx, got))
	445	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Settings = got.Settings; return true, nil })
	446	+	require.NoError(t, err)
446	447		again, err := s.Get(ctx, n)
447	448		require.NoError(t, err)
448	449		assert.Equal(t, got.Settings, again.Settings)
449	450
450	451		got.Settings = domain.RepoSettings{AllowedMergeModes: []domain.MergeMode{"rebase"}}
451		-	assert.ErrorIs(t, s.Update(ctx, got), domain.ErrInvalid)
	452	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Settings = got.Settings; return true, nil })
	453	+	assert.ErrorIs(t, err, domain.ErrInvalid)
452	454		}
453	455
454	456		// TestRepoWithoutStoredSettingsReadsTheDefaults: a repository stored before

 	assert.Equal(t, domain.DefaultRepoSettings(), got.Settings)
 	got.Settings = domain.RepoSettings{AllowedMergeModes: []domain.MergeMode{domain.MergeModeSquash}, FailingChecksBlock: false, Issues: domain.IssuesOwner}
-	require.NoError(t, s.Update(ctx, got))
+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Settings = got.Settings; return true, nil })
+	require.NoError(t, err)
 	again, err := s.Get(ctx, n)
 	require.NoError(t, err)
 	assert.Equal(t, got.Settings, again.Settings)
 	got.Settings = domain.RepoSettings{AllowedMergeModes: []domain.MergeMode{"rebase"}}
-	assert.ErrorIs(t, s.Update(ctx, got), domain.ErrInvalid)
+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Settings = got.Settings; return true, nil })
+	assert.ErrorIs(t, err, domain.ErrInvalid)
 }
 // TestRepoWithoutStoredSettingsReadsTheDefaults: a repository stored before

M internal/adapters/kvstore/repos.go +42 -22


@@ -64,31 +64,51 @@ func (s *Store) List(ctx context.Context) ([]domain.Repo, error) {
64	64		return repos, nil
65	65		}
66	66
67		-	// Update replaces the stored repository. It returns domain.ErrNotFound if
68		-	// there is none and domain.ErrConflict if another write changed it since it
69		-	// was read.
	67	+	const repoMutationTries = 8
	68	+
	69	+	// Mutate applies a retry-safe mutation under the version it read.
70	70		//
71		-	//hexcheck:ignore suspicious-business-logic-in-adapter port method name only
72		-	func (s *Store) Update(ctx context.Context, repo domain.Repo) error {
73		-	val, err := encodeRepo(repo)
74		-	if err != nil {
75		-	return err
76		-	}
77		-	key := repoKey(repo.Name)
78		-	item, ok := s.db.Get(ctx, key)
79		-	if !ok {
80		-	return domain.ErrNotFound
81		-	}
82		-	var b kv.Batch
83		-	b.RequireVersion(key, item.Version)
84		-	b.Put(key, val)
85		-	if _, err := s.db.Apply(ctx, &b); err != nil {
86		-	if errors.Is(err, kv.ErrConflict) {
87		-	return domain.ErrConflict
	71	+	//hexcheck:ignore suspicious-business-logic-in-adapter optimistic concurrency is a storage concern
	72	+	func (s Store) Mutate(ctx context.Context, name domain.RepoName, fn func(domain.Repo) (bool, error)) (domain.Repo, error) {
	73	+	key := repoKey(name)
	74	+	for range repoMutationTries {
	75	+	if err := ctx.Err(); err != nil {
	76	+	return domain.Repo{}, err
	77	+	}
	78	+	item, ok := s.db.Get(ctx, key)
	79	+	if !ok {
	80	+	return domain.Repo{}, domain.ErrNotFound
88	81		}
89		-	return fmt.Errorf("update repository %q: %w", repo.Name, err)
	82	+	repo, err := decodeRepo(item.Value)
	83	+	if err != nil {
	84	+	return domain.Repo{}, err
	85	+	}
	86	+	changed, err := fn(&repo)
	87	+	if err != nil {
	88	+	return domain.Repo{}, err
	89	+	}
	90	+	if repo.Name != name {
	91	+	return domain.Repo{}, fmt.Errorf("mutate repository name: %w", domain.ErrInvalidName)
	92	+	}
	93	+	if !changed {
	94	+	return decodeRepo(item.Value)
	95	+	}
	96	+	val, err := encodeRepo(repo)
	97	+	if err != nil {
	98	+	return domain.Repo{}, err
	99	+	}
	100	+	var b kv.Batch
	101	+	b.RequireVersion(key, item.Version)
	102	+	b.Put(key, val)
	103	+	if _, err := s.db.Apply(ctx, &b); err != nil {
	104	+	if errors.Is(err, kv.ErrConflict) {
	105	+	continue
	106	+	}
	107	+	return domain.Repo{}, fmt.Errorf("mutate repository %q: %w", name, err)
	108	+	}
	109	+	return decodeRepo(val)
90	110		}
91		-	return nil
	111	+	return domain.Repo{}, domain.ErrConflict
92	112		}
93	113
94	114		// guardRepoKey makes the batch depend on the repository: it requires the


@@ -64,31 +64,51 @@ func (s *Store) List(ctx context.Context) ([]domain.Repo, error) {
64		return repos, nil	64		return repos, nil
65		}	65		}
66			66
67	-	// Update replaces the stored repository. It returns domain.ErrNotFound if	67	+	const repoMutationTries = 8
68	-	// there is none and domain.ErrConflict if another write changed it since it	68	+
69	-	// was read.	69	+	// Mutate applies a retry-safe mutation under the version it read.
70		//	70		//
71	-	//hexcheck:ignore suspicious-business-logic-in-adapter port method name only	71	+	//hexcheck:ignore suspicious-business-logic-in-adapter optimistic concurrency is a storage concern
72	-	func (s *Store) Update(ctx context.Context, repo domain.Repo) error {	72	+	func (s Store) Mutate(ctx context.Context, name domain.RepoName, fn func(domain.Repo) (bool, error)) (domain.Repo, error) {
73	-	val, err := encodeRepo(repo)	73	+	key := repoKey(name)
74	-	if err != nil {	74	+	for range repoMutationTries {
75	-	return err	75	+	if err := ctx.Err(); err != nil {
76	-	}	76	+	return domain.Repo{}, err
77	-	key := repoKey(repo.Name)	77	+	}
78	-	item, ok := s.db.Get(ctx, key)	78	+	item, ok := s.db.Get(ctx, key)
79	-	if !ok {	79	+	if !ok {
80	-	return domain.ErrNotFound	80	+	return domain.Repo{}, domain.ErrNotFound
81	-	}
82	-	var b kv.Batch
83	-	b.RequireVersion(key, item.Version)
84	-	b.Put(key, val)
85	-	if _, err := s.db.Apply(ctx, &b); err != nil {
86	-	if errors.Is(err, kv.ErrConflict) {
87	-	return domain.ErrConflict
88		}	81		}
89	-	return fmt.Errorf("update repository %q: %w", repo.Name, err)	82	+	repo, err := decodeRepo(item.Value)
			83	+	if err != nil {
			84	+	return domain.Repo{}, err
			85	+	}
			86	+	changed, err := fn(&repo)
			87	+	if err != nil {
			88	+	return domain.Repo{}, err
			89	+	}
			90	+	if repo.Name != name {
			91	+	return domain.Repo{}, fmt.Errorf("mutate repository name: %w", domain.ErrInvalidName)
			92	+	}
			93	+	if !changed {
			94	+	return decodeRepo(item.Value)
			95	+	}
			96	+	val, err := encodeRepo(repo)
			97	+	if err != nil {
			98	+	return domain.Repo{}, err
			99	+	}
			100	+	var b kv.Batch
			101	+	b.RequireVersion(key, item.Version)
			102	+	b.Put(key, val)
			103	+	if _, err := s.db.Apply(ctx, &b); err != nil {
			104	+	if errors.Is(err, kv.ErrConflict) {
			105	+	continue
			106	+	}
			107	+	return domain.Repo{}, fmt.Errorf("mutate repository %q: %w", name, err)
			108	+	}
			109	+	return decodeRepo(val)
90		}	110		}
91	-	return nil	111	+	return domain.Repo{}, domain.ErrConflict
92		}	112		}
93			113
94		// guardRepoKey makes the batch depend on the repository: it requires the	114		// guardRepoKey makes the batch depend on the repository: it requires the

A internal/adapters/kvstore/repos_mutate_test.go +87 -0

@@ -0,0 +1,87 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore_test
	4	+
	5	+	import (
	6	+	"errors"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/stretchr/testify/require"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	)
	14	+
	15	+	func TestRepoMutateInterleavedWrites(t *testing.T) {
	16	+	ctx := t.Context()
	17	+	s := openStore(t, t.TempDir())
	18	+	n := name(t, "proj")
	19	+	original := domain.Repo{Name: n, Visibility: domain.VisibilityPublic, CreatedAt: created}
	20	+	require.NoError(t, s.Create(ctx, original))
	21	+	later := created.Add(2 * time.Hour)
	22	+	calls := 0
	23	+	got, err := s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) {
	24	+	calls++
	25	+	// Test-only interleaving: inject a committed write after the outer read.
	26	+	if calls == 1 {
	27	+	_, err := s.Mutate(ctx, n, func(current *domain.Repo) (bool, error) {
	28	+	current.Visibility = domain.VisibilityPrivate
	29	+	current.ActiveAt = later
	30	+	current.PushedAt = later
	31	+	return true, nil
	32	+	})
	33	+	require.NoError(t, err)
	34	+	}
	35	+	r.Description = "updated"
	36	+	at := created.Add(time.Hour)
	37	+	if at.After(r.ActiveAt) {
	38	+	r.ActiveAt = at
	39	+	}
	40	+	if at.After(r.PushedAt) {
	41	+	r.PushedAt = at
	42	+	}
	43	+	return true, nil
	44	+	})
	45	+	require.NoError(t, err)
	46	+	require.Equal(t, 2, calls)
	47	+	require.Equal(t, domain.VisibilityPrivate, got.Visibility)
	48	+	require.Equal(t, "updated", got.Description)
	49	+	require.Equal(t, later, got.ActiveAt)
	50	+	require.Equal(t, later, got.PushedAt)
	51	+	stored, err := s.Get(ctx, n)
	52	+	require.NoError(t, err)
	53	+	require.Equal(t, got, stored)
	54	+	}
	55	+
	56	+	func TestRepoMutateContracts(t *testing.T) {
	57	+	ctx := t.Context()
	58	+	s := openStore(t, t.TempDir())
	59	+	n := name(t, "proj")
	60	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, CreatedAt: created}))
	61	+	original, err := s.Get(ctx, n)
	62	+	require.NoError(t, err)
	63	+	boom := errors.New("callback failed")
	64	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Description = "discard"; return true, boom })
	65	+	require.ErrorIs(t, err, boom)
	66	+	got, err := s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Description = "discard"; return false, nil })
	67	+	require.NoError(t, err)
	68	+	require.Equal(t, original, got)
	69	+	for _, changed := range []bool{false, true} {
	70	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Name = name(t, "other"); return changed, nil })
	71	+	require.ErrorIs(t, err, domain.ErrInvalidName)
	72	+	}
	73	+	got, err = s.Get(ctx, n)
	74	+	require.NoError(t, err)
	75	+	require.Equal(t, original, got)
	76	+	calls := 0
	77	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) {
	78	+	calls++
	79	+	_, err := s.Mutate(ctx, n, func(current *domain.Repo) (bool, error) { current.Description = "concurrent"; return true, nil })
	80	+	require.NoError(t, err)
	81	+	return true, nil
	82	+	})
	83	+	require.ErrorIs(t, err, domain.ErrConflict)
	84	+	require.Equal(t, 8, calls)
	85	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { require.NoError(t, s.Delete(ctx, n)); return true, nil })
	86	+	require.ErrorIs(t, err, domain.ErrNotFound)
	87	+	}

@@ -0,0 +1,87 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore_test
	4	+
	5	+	import (
	6	+	"errors"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/stretchr/testify/require"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	)
	14	+
	15	+	func TestRepoMutateInterleavedWrites(t *testing.T) {
	16	+	ctx := t.Context()
	17	+	s := openStore(t, t.TempDir())
	18	+	n := name(t, "proj")
	19	+	original := domain.Repo{Name: n, Visibility: domain.VisibilityPublic, CreatedAt: created}
	20	+	require.NoError(t, s.Create(ctx, original))
	21	+	later := created.Add(2 * time.Hour)
	22	+	calls := 0
	23	+	got, err := s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) {
	24	+	calls++
	25	+	// Test-only interleaving: inject a committed write after the outer read.
	26	+	if calls == 1 {
	27	+	_, err := s.Mutate(ctx, n, func(current *domain.Repo) (bool, error) {
	28	+	current.Visibility = domain.VisibilityPrivate
	29	+	current.ActiveAt = later
	30	+	current.PushedAt = later
	31	+	return true, nil
	32	+	})
	33	+	require.NoError(t, err)
	34	+	}
	35	+	r.Description = "updated"
	36	+	at := created.Add(time.Hour)
	37	+	if at.After(r.ActiveAt) {
	38	+	r.ActiveAt = at
	39	+	}
	40	+	if at.After(r.PushedAt) {
	41	+	r.PushedAt = at
	42	+	}
	43	+	return true, nil
	44	+	})
	45	+	require.NoError(t, err)
	46	+	require.Equal(t, 2, calls)
	47	+	require.Equal(t, domain.VisibilityPrivate, got.Visibility)
	48	+	require.Equal(t, "updated", got.Description)
	49	+	require.Equal(t, later, got.ActiveAt)
	50	+	require.Equal(t, later, got.PushedAt)
	51	+	stored, err := s.Get(ctx, n)
	52	+	require.NoError(t, err)
	53	+	require.Equal(t, got, stored)
	54	+	}
	55	+
	56	+	func TestRepoMutateContracts(t *testing.T) {
	57	+	ctx := t.Context()
	58	+	s := openStore(t, t.TempDir())
	59	+	n := name(t, "proj")
	60	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, CreatedAt: created}))
	61	+	original, err := s.Get(ctx, n)
	62	+	require.NoError(t, err)
	63	+	boom := errors.New("callback failed")
	64	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Description = "discard"; return true, boom })
	65	+	require.ErrorIs(t, err, boom)
	66	+	got, err := s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Description = "discard"; return false, nil })
	67	+	require.NoError(t, err)
	68	+	require.Equal(t, original, got)
	69	+	for _, changed := range []bool{false, true} {
	70	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { r.Name = name(t, "other"); return changed, nil })
	71	+	require.ErrorIs(t, err, domain.ErrInvalidName)
	72	+	}
	73	+	got, err = s.Get(ctx, n)
	74	+	require.NoError(t, err)
	75	+	require.Equal(t, original, got)
	76	+	calls := 0
	77	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) {
	78	+	calls++
	79	+	_, err := s.Mutate(ctx, n, func(current *domain.Repo) (bool, error) { current.Description = "concurrent"; return true, nil })
	80	+	require.NoError(t, err)
	81	+	return true, nil
	82	+	})
	83	+	require.ErrorIs(t, err, domain.ErrConflict)
	84	+	require.Equal(t, 8, calls)
	85	+	_, err = s.Mutate(ctx, n, func(r *domain.Repo) (bool, error) { require.NoError(t, s.Delete(ctx, n)); return true, nil })
	86	+	require.ErrorIs(t, err, domain.ErrNotFound)
	87	+	}

M internal/adapters/kvstore/store_test.go +6 -3


@@ -66,13 +66,15 @@ func TestCreateGetListUpdateDelete(t *testing.T) {
66	66		b.Description = "changed"
67	67		b.Topics = []string{"forge"}
68	68		b.Visibility = domain.VisibilityPrivate
69		-	require.NoError(t, s.Update(ctx, b))
	69	+	_, err = s.Mutate(ctx, b.Name, func(r domain.Repo) (bool, error) { r = b; return true, nil })
	70	+	require.NoError(t, err)
70	71		got, err = s.Get(ctx, b.Name)
71	72		require.NoError(t, err)
72	73		assert.Equal(t, b, got)
73	74
74	75		b.Topics = nil
75		-	require.NoError(t, s.Update(ctx, b))
	76	+	_, err = s.Mutate(ctx, b.Name, func(r domain.Repo) (bool, error) { r = b; return true, nil })
	77	+	require.NoError(t, err)
76	78		got, err = s.Get(ctx, b.Name)
77	79		require.NoError(t, err)
78	80		assert.Nil(t, got.Topics, "cleared topics read back as nil")

@@ -81,7 +83,8 @@ func TestCreateGetListUpdateDelete(t *testing.T) {
81	83		_, err = s.Get(ctx, b.Name)
82	84		assert.ErrorIs(t, err, domain.ErrNotFound)
83	85		require.NoError(t, s.Delete(ctx, b.Name), "deleting a missing repository is not an error")
84		-	assert.ErrorIs(t, s.Update(ctx, b), domain.ErrNotFound)
	86	+	_, err = s.Mutate(ctx, b.Name, func(r *domain.Repo) (bool, error) { return true, nil })
	87	+	assert.ErrorIs(t, err, domain.ErrNotFound)
85	88
86	89		_, err = s.Get(ctx, name(t, "never"))
87	90		assert.ErrorIs(t, err, domain.ErrNotFound)

 	b.Description = "changed"
 	b.Topics = []string{"forge"}
 	b.Visibility = domain.VisibilityPrivate
-	require.NoError(t, s.Update(ctx, b))
+	_, err = s.Mutate(ctx, b.Name, func(r *domain.Repo) (bool, error) { *r = b; return true, nil })
+	require.NoError(t, err)
 	got, err = s.Get(ctx, b.Name)
 	require.NoError(t, err)
 	assert.Equal(t, b, got)
 	b.Topics = nil
-	require.NoError(t, s.Update(ctx, b))
+	_, err = s.Mutate(ctx, b.Name, func(r *domain.Repo) (bool, error) { *r = b; return true, nil })
+	require.NoError(t, err)
 	got, err = s.Get(ctx, b.Name)
 	require.NoError(t, err)
 	assert.Nil(t, got.Topics, "cleared topics read back as nil")
 	_, err = s.Get(ctx, b.Name)
 	assert.ErrorIs(t, err, domain.ErrNotFound)
 	require.NoError(t, s.Delete(ctx, b.Name), "deleting a missing repository is not an error")
-	assert.ErrorIs(t, s.Update(ctx, b), domain.ErrNotFound)
+	_, err = s.Mutate(ctx, b.Name, func(r *domain.Repo) (bool, error) { return true, nil })
+	assert.ErrorIs(t, err, domain.ErrNotFound)
 	_, err = s.Get(ctx, name(t, "never"))
 	assert.ErrorIs(t, err, domain.ErrNotFound)

M internal/ports/mocks/RepoMetaStore.go +36 -21


@@ -289,59 +289,74 @@ func (_c *RepoMetaStore_List_Call) RunAndReturn(run func(ctx context.Context) ([
289	289		return _c
290	290		}
291	291
292		-	// Update provides a mock function for the type RepoMetaStore
293		-	func (_mock *RepoMetaStore) Update(ctx context.Context, repo domain.Repo) error {
294		-	ret := _mock.Called(ctx, repo)
	292	+	// Mutate provides a mock function for the type RepoMetaStore
	293	+	func (_mock RepoMetaStore) Mutate(ctx context.Context, name domain.RepoName, fn func(domain.Repo) (changed bool, err error)) (domain.Repo, error) {
	294	+	ret := _mock.Called(ctx, name, fn)
295	295
296	296		if len(ret) == 0 {
297		-	panic("no return value specified for Update")
	297	+	panic("no return value specified for Mutate")
298	298		}
299	299
300		-	var r0 error
301		-	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.Repo) error); ok {
302		-	r0 = returnFunc(ctx, repo)
	300	+	var r0 domain.Repo
	301	+	var r1 error
	302	+	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.RepoName, func(*domain.Repo) (changed bool, err error)) (domain.Repo, error)); ok {
	303	+	return returnFunc(ctx, name, fn)
	304	+	}
	305	+	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.RepoName, func(*domain.Repo) (changed bool, err error)) domain.Repo); ok {
	306	+	r0 = returnFunc(ctx, name, fn)
303	307		} else {
304		-	r0 = ret.Error(0)
	308	+	r0 = ret.Get(0).(domain.Repo)
305	309		}
306		-	return r0
	310	+	if returnFunc, ok := ret.Get(1).(func(context.Context, domain.RepoName, func(*domain.Repo) (changed bool, err error)) error); ok {
	311	+	r1 = returnFunc(ctx, name, fn)
	312	+	} else {
	313	+	r1 = ret.Error(1)
	314	+	}
	315	+	return r0, r1
307	316		}
308	317
309		-	// RepoMetaStore_Update_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Update'
310		-	type RepoMetaStore_Update_Call struct {
	318	+	// RepoMetaStore_Mutate_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Mutate'
	319	+	type RepoMetaStore_Mutate_Call struct {
311	320		*mock.Call
312	321		}
313	322
314		-	// Update is a helper method to define mock.On call
	323	+	// Mutate is a helper method to define mock.On call
315	324		// - ctx context.Context
316		-	// - repo domain.Repo
317		-	func (_e RepoMetaStore_Expecter) Update(ctx any, repo any) RepoMetaStore_Update_Call {
318		-	return &RepoMetaStore_Update_Call{Call: _e.mock.On("Update", ctx, repo)}
	325	+	// - name domain.RepoName
	326	+	// - fn func(*domain.Repo) (changed bool, err error)
	327	+	func (_e RepoMetaStore_Expecter) Mutate(ctx any, name any, fn any) RepoMetaStore_Mutate_Call {
	328	+	return &RepoMetaStore_Mutate_Call{Call: _e.mock.On("Mutate", ctx, name, fn)}
319	329		}
320	330
321		-	func (_c RepoMetaStore_Update_Call) Run(run func(ctx context.Context, repo domain.Repo)) RepoMetaStore_Update_Call {
	331	+	func (_c RepoMetaStore_Mutate_Call) Run(run func(ctx context.Context, name domain.RepoName, fn func(domain.Repo) (changed bool, err error))) *RepoMetaStore_Mutate_Call {
322	332		_c.Call.Run(func(args mock.Arguments) {
323	333		var arg0 context.Context
324	334		if args[0] != nil {
325	335		arg0 = args[0].(context.Context)
326	336		}
327		-	var arg1 domain.Repo
	337	+	var arg1 domain.RepoName
328	338		if args[1] != nil {
329		-	arg1 = args[1].(domain.Repo)
	339	+	arg1 = args[1].(domain.RepoName)
	340	+	}
	341	+	var arg2 func(*domain.Repo) (changed bool, err error)
	342	+	if args[2] != nil {
	343	+	arg2 = args[2].(func(*domain.Repo) (changed bool, err error))
330	344		}
331	345		run(
332	346		arg0,
333	347		arg1,
	348	+	arg2,
334	349		)
335	350		})
336	351		return _c
337	352		}
338	353
339		-	func (_c RepoMetaStore_Update_Call) Return(err error) RepoMetaStore_Update_Call {
340		-	_c.Call.Return(err)
	354	+	func (_c RepoMetaStore_Mutate_Call) Return(repo domain.Repo, err error) RepoMetaStore_Mutate_Call {
	355	+	_c.Call.Return(repo, err)
341	356		return _c
342	357		}
343	358
344		-	func (_c RepoMetaStore_Update_Call) RunAndReturn(run func(ctx context.Context, repo domain.Repo) error) RepoMetaStore_Update_Call {
	359	+	func (_c RepoMetaStore_Mutate_Call) RunAndReturn(run func(ctx context.Context, name domain.RepoName, fn func(domain.Repo) (changed bool, err error)) (domain.Repo, error)) *RepoMetaStore_Mutate_Call {
345	360		_c.Call.Return(run)
346	361		return _c
347	362		}

 	return _c
 }
-// Update provides a mock function for the type RepoMetaStore
+// Mutate provides a mock function for the type RepoMetaStore
-func (_mock *RepoMetaStore) Update(ctx context.Context, repo domain.Repo) error {
+func (_mock *RepoMetaStore) Mutate(ctx context.Context, name domain.RepoName, fn func(*domain.Repo) (changed bool, err error)) (domain.Repo, error) {
-	ret := _mock.Called(ctx, repo)
+	ret := _mock.Called(ctx, name, fn)
 	if len(ret) == 0 {
-		panic("no return value specified for Update")
+		panic("no return value specified for Mutate")
 	}
-	var r0 error
+	var r0 domain.Repo
-	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.Repo) error); ok {
+	var r1 error
-		r0 = returnFunc(ctx, repo)
+	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.RepoName, func(*domain.Repo) (changed bool, err error)) (domain.Repo, error)); ok {
+		return returnFunc(ctx, name, fn)
+	}
+	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.RepoName, func(*domain.Repo) (changed bool, err error)) domain.Repo); ok {
+		r0 = returnFunc(ctx, name, fn)
 	} else {
-		r0 = ret.Error(0)
+		r0 = ret.Get(0).(domain.Repo)
 	}
-	return r0
+	if returnFunc, ok := ret.Get(1).(func(context.Context, domain.RepoName, func(*domain.Repo) (changed bool, err error)) error); ok {
+		r1 = returnFunc(ctx, name, fn)
+	} else {
+		r1 = ret.Error(1)
+	}
+	return r0, r1
 }
-// RepoMetaStore_Update_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Update'
+// RepoMetaStore_Mutate_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Mutate'
-type RepoMetaStore_Update_Call struct {
+type RepoMetaStore_Mutate_Call struct {
 	*mock.Call
 }
-// Update is a helper method to define mock.On call
+// Mutate is a helper method to define mock.On call
 //   - ctx context.Context
-//   - repo domain.Repo
+//   - name domain.RepoName
-func (_e *RepoMetaStore_Expecter) Update(ctx any, repo any) *RepoMetaStore_Update_Call {
+//   - fn func(*domain.Repo) (changed bool, err error)
-	return &RepoMetaStore_Update_Call{Call: _e.mock.On("Update", ctx, repo)}
+func (_e *RepoMetaStore_Expecter) Mutate(ctx any, name any, fn any) *RepoMetaStore_Mutate_Call {
+	return &RepoMetaStore_Mutate_Call{Call: _e.mock.On("Mutate", ctx, name, fn)}
 }
-func (_c *RepoMetaStore_Update_Call) Run(run func(ctx context.Context, repo domain.Repo)) *RepoMetaStore_Update_Call {
+func (_c *RepoMetaStore_Mutate_Call) Run(run func(ctx context.Context, name domain.RepoName, fn func(*domain.Repo) (changed bool, err error))) *RepoMetaStore_Mutate_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		var arg0 context.Context
 		if args[0] != nil {
 			arg0 = args[0].(context.Context)
 		}
-		var arg1 domain.Repo
+		var arg1 domain.RepoName
 		if args[1] != nil {
-			arg1 = args[1].(domain.Repo)
+			arg1 = args[1].(domain.RepoName)
+		}
+		var arg2 func(*domain.Repo) (changed bool, err error)
+		if args[2] != nil {
+			arg2 = args[2].(func(*domain.Repo) (changed bool, err error))
 		}
 		run(
 			arg0,
 			arg1,
+			arg2,
 		)
 	})
 	return _c
 }
-func (_c *RepoMetaStore_Update_Call) Return(err error) *RepoMetaStore_Update_Call {
+func (_c *RepoMetaStore_Mutate_Call) Return(repo domain.Repo, err error) *RepoMetaStore_Mutate_Call {
-	_c.Call.Return(err)
+	_c.Call.Return(repo, err)
 	return _c
 }
-func (_c *RepoMetaStore_Update_Call) RunAndReturn(run func(ctx context.Context, repo domain.Repo) error) *RepoMetaStore_Update_Call {
+func (_c *RepoMetaStore_Mutate_Call) RunAndReturn(run func(ctx context.Context, name domain.RepoName, fn func(*domain.Repo) (changed bool, err error)) (domain.Repo, error)) *RepoMetaStore_Mutate_Call {
 	_c.Call.Return(run)
 	return _c
 }

M internal/ports/ports.go +7 -4


@@ -32,10 +32,13 @@ type RepoMetaStore interface {
32	32		Get(ctx context.Context, name domain.RepoName) (domain.Repo, error)
33	33		// List returns every repository ordered by name.
34	34		List(ctx context.Context) ([]domain.Repo, error)
35		-	// Update replaces the stored repository. It returns domain.ErrNotFound if
36		-	// there is none and domain.ErrConflict if another write changed the
37		-	// stored repository since it was read.
38		-	Update(ctx context.Context, repo domain.Repo) error
	35	+	// Mutate applies fn to the stored repository and writes under the version read.
	36	+	// Concurrent writes retry fn up to a bounded number of attempts, then return
	37	+	// domain.ErrConflict. fn must be pure and safe to call several times, and
	38	+	// must not change Name. changed=false writes nothing. It returns the stored
	39	+	// repository, or domain.ErrNotFound when absent. Errors from fn are returned
	40	+	// unchanged with nothing written.
	41	+	Mutate(ctx context.Context, name domain.RepoName, fn func(*domain.Repo) (changed bool, err error)) (domain.Repo, error)
39	42		// Delete removes the repository and everything keyed to it: its merge
40	43		// requests, their comments and read positions, and its number sequence.
41	44		// A repository created again under the same name starts empty, with its

 	Get(ctx context.Context, name domain.RepoName) (domain.Repo, error)
 	// List returns every repository ordered by name.
 	List(ctx context.Context) ([]domain.Repo, error)
-	// Update replaces the stored repository. It returns domain.ErrNotFound if
+	// Mutate applies fn to the stored repository and writes under the version read.
-	// there is none and domain.ErrConflict if another write changed the
+	// Concurrent writes retry fn up to a bounded number of attempts, then return
-	// stored repository since it was read.
+	// domain.ErrConflict. fn must be pure and safe to call several times, and
-	Update(ctx context.Context, repo domain.Repo) error
+	// must not change Name. changed=false writes nothing. It returns the stored
+	// repository, or domain.ErrNotFound when absent. Errors from fn are returned
+	// unchanged with nothing written.
+	Mutate(ctx context.Context, name domain.RepoName, fn func(*domain.Repo) (changed bool, err error)) (domain.Repo, error)
 	// Delete removes the repository and everything keyed to it: its merge
 	// requests, their comments and read positions, and its number sequence.
 	// A repository created again under the same name starts empty, with its

M internal/usecase/activity.go +13 -23


@@ -25,29 +25,19 @@ func recordActivity(ctx context.Context, meta ports.RepoMetaStore, name domain.R
25	25		// touchActiveAt preserves later activity and retries concurrent metadata writes.
26	26		// A repository deleted meanwhile is not an error.
27	27		func touchActiveAt(ctx context.Context, meta ports.RepoMetaStore, name domain.RepoName, now time.Time) (domain.Repo, error) {
28		-	for range pushRetries {
29		-	repo, err := meta.Get(ctx, name)
30		-	if errors.Is(err, domain.ErrNotFound) {
31		-	return domain.Repo{}, nil
	28	+	repo, err := meta.Mutate(ctx, name, func(repo *domain.Repo) (bool, error) {
	29	+	changed := false
	30	+	if now.After(repo.ActiveAt) {
	31	+	repo.ActiveAt = now
	32	+	changed = true
32	33		}
33		-	if err != nil {
34		-	return domain.Repo{}, fmt.Errorf("get repository %q: %w", name, err)
35		-	}
36		-	if !now.After(repo.ActiveAt) {
37		-	return repo, nil // a later activity already recorded
38		-	}
39		-	repo.ActiveAt = now
40		-	err = meta.Update(ctx, repo)
41		-	if errors.Is(err, domain.ErrConflict) {
42		-	continue
43		-	}
44		-	if errors.Is(err, domain.ErrNotFound) {
45		-	return domain.Repo{}, nil
46		-	}
47		-	if err != nil {
48		-	return repo, fmt.Errorf("update repository %q: %w", name, err)
49		-	}
50		-	return repo, nil
	34	+	return changed, nil
	35	+	})
	36	+	if errors.Is(err, domain.ErrNotFound) {
	37	+	return domain.Repo{}, nil
	38	+	}
	39	+	if err != nil {
	40	+	return repo, fmt.Errorf("update repository %q: %w", name, err)
51	41		}
52		-	return domain.Repo{}, fmt.Errorf("update active_at of %q: %w", name, domain.ErrConflict)
	42	+	return repo, nil
53	43		}

 // touchActiveAt preserves later activity and retries concurrent metadata writes.
 // A repository deleted meanwhile is not an error.
 func touchActiveAt(ctx context.Context, meta ports.RepoMetaStore, name domain.RepoName, now time.Time) (domain.Repo, error) {
-	for range pushRetries {
+	repo, err := meta.Mutate(ctx, name, func(repo *domain.Repo) (bool, error) {
-		repo, err := meta.Get(ctx, name)
+		changed := false
-		if errors.Is(err, domain.ErrNotFound) {
+		if now.After(repo.ActiveAt) {
-			return domain.Repo{}, nil
+			repo.ActiveAt = now
+			changed = true
 		}
-		if err != nil {
+		return changed, nil
-			return domain.Repo{}, fmt.Errorf("get repository %q: %w", name, err)
+	})
-		}
+	if errors.Is(err, domain.ErrNotFound) {
-		if !now.After(repo.ActiveAt) {
+		return domain.Repo{}, nil
-			return repo, nil // a later activity already recorded
+	}
-		}
+	if err != nil {
-		repo.ActiveAt = now
+		return repo, fmt.Errorf("update repository %q: %w", name, err)
-		err = meta.Update(ctx, repo)
-		if errors.Is(err, domain.ErrConflict) {
-			continue
-		}
-		if errors.Is(err, domain.ErrNotFound) {
-			return domain.Repo{}, nil
-		}
-		if err != nil {
-			return repo, fmt.Errorf("update repository %q: %w", name, err)
-		}
-		return repo, nil
 	}
-	return domain.Repo{}, fmt.Errorf("update active_at of %q: %w", name, domain.ErrConflict)
+	return repo, nil
 }

A internal/usecase/activity_store_test.go +45 -0

@@ -0,0 +1,45 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase
	4	+
	5	+	import (
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+
	11	+	"bnema.dev/ermite/internal/adapters/kvstore"
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	)
	14	+
	15	+	func TestActivityTimestampsOverStore(t *testing.T) {
	16	+	ctx := t.Context()
	17	+	s, err := kvstore.Open(ctx, t.TempDir())
	18	+	require.NoError(t, err)
	19	+	t.Cleanup(func() { require.NoError(t, s.Close(ctx)) })
	20	+	n, err := domain.ParseRepoName("proj")
	21	+	require.NoError(t, err)
	22	+	at := time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC)
	23	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: at}))
	24	+	got, err := touchPushedAt(ctx, s, n, at.Add(time.Hour))
	25	+	require.NoError(t, err)
	26	+	require.Equal(t, got.PushedAt, got.ActiveAt)
	27	+	_, err = touchActiveAt(ctx, s, n, at.Add(3*time.Hour))
	28	+	require.NoError(t, err)
	29	+	_, err = touchPushedAt(ctx, s, n, at.Add(2*time.Hour))
	30	+	require.NoError(t, err)
	31	+	_, err = touchPushedAt(ctx, s, n, at)
	32	+	require.NoError(t, err)
	33	+	_, err = touchActiveAt(ctx, s, n, at)
	34	+	require.NoError(t, err)
	35	+	got, err = s.Get(ctx, n)
	36	+	require.NoError(t, err)
	37	+	require.Equal(t, at.Add(2*time.Hour), got.PushedAt)
	38	+	require.Equal(t, at.Add(3*time.Hour), got.ActiveAt)
	39	+	require.Equal(t, domain.VisibilityPrivate, got.Visibility)
	40	+	require.NoError(t, s.Delete(ctx, n))
	41	+	_, err = touchPushedAt(ctx, s, n, at)
	42	+	require.NoError(t, err)
	43	+	_, err = touchActiveAt(ctx, s, n, at)
	44	+	require.NoError(t, err)
	45	+	}

@@ -0,0 +1,45 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase
	4	+
	5	+	import (
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+
	11	+	"bnema.dev/ermite/internal/adapters/kvstore"
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	)
	14	+
	15	+	func TestActivityTimestampsOverStore(t *testing.T) {
	16	+	ctx := t.Context()
	17	+	s, err := kvstore.Open(ctx, t.TempDir())
	18	+	require.NoError(t, err)
	19	+	t.Cleanup(func() { require.NoError(t, s.Close(ctx)) })
	20	+	n, err := domain.ParseRepoName("proj")
	21	+	require.NoError(t, err)
	22	+	at := time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC)
	23	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: at}))
	24	+	got, err := touchPushedAt(ctx, s, n, at.Add(time.Hour))
	25	+	require.NoError(t, err)
	26	+	require.Equal(t, got.PushedAt, got.ActiveAt)
	27	+	_, err = touchActiveAt(ctx, s, n, at.Add(3*time.Hour))
	28	+	require.NoError(t, err)
	29	+	_, err = touchPushedAt(ctx, s, n, at.Add(2*time.Hour))
	30	+	require.NoError(t, err)
	31	+	_, err = touchPushedAt(ctx, s, n, at)
	32	+	require.NoError(t, err)
	33	+	_, err = touchActiveAt(ctx, s, n, at)
	34	+	require.NoError(t, err)
	35	+	got, err = s.Get(ctx, n)
	36	+	require.NoError(t, err)
	37	+	require.Equal(t, at.Add(2*time.Hour), got.PushedAt)
	38	+	require.Equal(t, at.Add(3*time.Hour), got.ActiveAt)
	39	+	require.Equal(t, domain.VisibilityPrivate, got.Visibility)
	40	+	require.NoError(t, s.Delete(ctx, n))
	41	+	_, err = touchPushedAt(ctx, s, n, at)
	42	+	require.NoError(t, err)
	43	+	_, err = touchActiveAt(ctx, s, n, at)
	44	+	require.NoError(t, err)
	45	+	}

M internal/usecase/activity_test.go +11 -6


@@ -3,6 +3,7 @@
3	3		package usecase
4	4
5	5		import (
	6	+	"context"
6	7		"errors"
7	8		"testing"
8	9		"time"

@@ -18,25 +19,29 @@ func TestTouchActiveAt(t *testing.T) {
18	19		name, err := domain.ParseRepoName("proj")
19	20		require.NoError(t, err)
20	21		now := time.Now().UTC()
21		-	t.Run("retries and preserves later activity", func(t *testing.T) {
	22	+	t.Run("preserves later activity", func(t *testing.T) {
22	23		meta := mocks.NewRepoMetaStore(t)
23		-	meta.EXPECT().Get(mock.Anything, name).Return(domain.Repo{Name: name}, nil).Once()
24		-	meta.EXPECT().Update(mock.Anything, mock.MatchedBy(func(r domain.Repo) bool { return r.ActiveAt.Equal(now) && r.PushedAt.IsZero() })).Return(domain.ErrConflict).Once()
25	24		later := domain.Repo{Name: name, ActiveAt: now.Add(time.Hour)}
26		-	meta.EXPECT().Get(mock.Anything, name).Return(later, nil).Once()
	25	+	meta.EXPECT().Mutate(mock.Anything, name, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.RepoName, fn func(*domain.Repo) (bool, error)) (domain.Repo, error) {
	26	+	repo := later
	27	+	changed, err := fn(&repo)
	28	+	require.NoError(t, err)
	29	+	require.False(t, changed)
	30	+	return repo, nil
	31	+	})
27	32		got, err := touchActiveAt(t.Context(), meta, name, now)
28	33		require.NoError(t, err)
29	34		require.Equal(t, later, got)
30	35		})
31	36		t.Run("deleted repository", func(t *testing.T) {
32	37		meta := mocks.NewRepoMetaStore(t)
33		-	meta.EXPECT().Get(mock.Anything, name).Return(domain.Repo{}, domain.ErrNotFound)
	38	+	meta.EXPECT().Mutate(mock.Anything, name, mock.Anything).Return(domain.Repo{}, domain.ErrNotFound)
34	39		_, err := touchActiveAt(t.Context(), meta, name, now)
35	40		require.NoError(t, err)
36	41		})
37	42		t.Run("best effort failure", func(t *testing.T) {
38	43		meta := mocks.NewRepoMetaStore(t)
39		-	meta.EXPECT().Get(mock.Anything, name).Return(domain.Repo{}, errors.New("offline"))
	44	+	meta.EXPECT().Mutate(mock.Anything, name, mock.Anything).Return(domain.Repo{}, errors.New("offline"))
40	45		require.NotPanics(t, func() { recordActivity(t.Context(), meta, name, now) })
41	46		})
42	47		}

 package usecase
 import (
+	"context"
 	"errors"
 	"testing"
 	"time"
 	name, err := domain.ParseRepoName("proj")
 	require.NoError(t, err)
 	now := time.Now().UTC()
-	t.Run("retries and preserves later activity", func(t *testing.T) {
+	t.Run("preserves later activity", func(t *testing.T) {
 		meta := mocks.NewRepoMetaStore(t)
-		meta.EXPECT().Get(mock.Anything, name).Return(domain.Repo{Name: name}, nil).Once()
-		meta.EXPECT().Update(mock.Anything, mock.MatchedBy(func(r domain.Repo) bool { return r.ActiveAt.Equal(now) && r.PushedAt.IsZero() })).Return(domain.ErrConflict).Once()
 		later := domain.Repo{Name: name, ActiveAt: now.Add(time.Hour)}
-		meta.EXPECT().Get(mock.Anything, name).Return(later, nil).Once()
+		meta.EXPECT().Mutate(mock.Anything, name, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.RepoName, fn func(*domain.Repo) (bool, error)) (domain.Repo, error) {
+			repo := later
+			changed, err := fn(&repo)
+			require.NoError(t, err)
+			require.False(t, changed)
+			return repo, nil
+		})
 		got, err := touchActiveAt(t.Context(), meta, name, now)
 		require.NoError(t, err)
 		require.Equal(t, later, got)
 	})
 	t.Run("deleted repository", func(t *testing.T) {
 		meta := mocks.NewRepoMetaStore(t)
-		meta.EXPECT().Get(mock.Anything, name).Return(domain.Repo{}, domain.ErrNotFound)
+		meta.EXPECT().Mutate(mock.Anything, name, mock.Anything).Return(domain.Repo{}, domain.ErrNotFound)
 		_, err := touchActiveAt(t.Context(), meta, name, now)
 		require.NoError(t, err)
 	})
 	t.Run("best effort failure", func(t *testing.T) {
 		meta := mocks.NewRepoMetaStore(t)
-		meta.EXPECT().Get(mock.Anything, name).Return(domain.Repo{}, errors.New("offline"))
+		meta.EXPECT().Mutate(mock.Anything, name, mock.Anything).Return(domain.Repo{}, errors.New("offline"))
 		require.NotPanics(t, func() { recordActivity(t.Context(), meta, name, now) })
 	})
 }

M internal/usecase/issue_moderation.go +1 -0


@@ -85,6 +85,7 @@ func (m *IssueModeration) Approve(ctx context.Context, p domain.Principal, repo
85	85		log := zerowrap.FromCtx(ctx)
86	86		log.Info().Str("repo", repo.String()).Int64("number", issue.Number).
87	87		Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("pending issue approved")
	88	+	recordActivity(ctx, m.meta, repo, now)
88	89		m.bus.Publish(ctx, domain.IssueOpened{Issue: issue, At: now, Visibility: r.Visibility, CreatedAt: r.CreatedAt})
89	90		return issue, nil
90	91		}


@@ -85,6 +85,7 @@ func (m *IssueModeration) Approve(ctx context.Context, p domain.Principal, repo
85	log := zerowrap.FromCtx(ctx)	85		log := zerowrap.FromCtx(ctx)
86	log.Info().Str("repo", repo.String()).Int64("number", issue.Number).	86		log.Info().Str("repo", repo.String()).Int64("number", issue.Number).
87	Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("pending issue approved")	87		Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("pending issue approved")
		88	+	recordActivity(ctx, m.meta, repo, now)
88	m.bus.Publish(ctx, domain.IssueOpened{Issue: issue, At: now, Visibility: r.Visibility, CreatedAt: r.CreatedAt})	89		m.bus.Publish(ctx, domain.IssueOpened{Issue: issue, At: now, Visibility: r.Visibility, CreatedAt: r.CreatedAt})
89	return issue, nil	90		return issue, nil
90	}	91		}

M internal/usecase/issue_moderation_test.go +1 -0


@@ -89,6 +89,7 @@ func TestIssueModerationApprove(t *testing.T) {
89	89		u := newModerationUnit(t, domain.IssuesPublic)
90	90		issue := domain.Issue{Repo: u.repo.Name, Number: 7, State: domain.IssueStateOpen}
91	91		u.pending.EXPECT().Approve(mock.Anything, u.repo.Name, u.id, now).Return(issue, nil).Once()
	92	+	expectRepoMutation(t, u.meta, u.repo, mock.MatchedBy(func(r domain.Repo) bool { return r.ActiveAt.Equal(now) })).Once()
92	93		u.bus.EXPECT().Publish(mock.Anything, domain.IssueOpened{Issue: issue, At: now, Visibility: u.repo.Visibility, CreatedAt: u.repo.CreatedAt}).Once()
93	94		got, err := u.mod.Approve(ctx, domain.Owner(), u.repo.Name, u.id)
94	95		require.NoError(t, err)


@@ -89,6 +89,7 @@ func TestIssueModerationApprove(t *testing.T) {
89	u := newModerationUnit(t, domain.IssuesPublic)	89		u := newModerationUnit(t, domain.IssuesPublic)
90	issue := domain.Issue{Repo: u.repo.Name, Number: 7, State: domain.IssueStateOpen}	90		issue := domain.Issue{Repo: u.repo.Name, Number: 7, State: domain.IssueStateOpen}
91	u.pending.EXPECT().Approve(mock.Anything, u.repo.Name, u.id, now).Return(issue, nil).Once()	91		u.pending.EXPECT().Approve(mock.Anything, u.repo.Name, u.id, now).Return(issue, nil).Once()
		92	+	expectRepoMutation(t, u.meta, u.repo, mock.MatchedBy(func(r domain.Repo) bool { return r.ActiveAt.Equal(now) })).Once()
92	u.bus.EXPECT().Publish(mock.Anything, domain.IssueOpened{Issue: issue, At: now, Visibility: u.repo.Visibility, CreatedAt: u.repo.CreatedAt}).Once()	93		u.bus.EXPECT().Publish(mock.Anything, domain.IssueOpened{Issue: issue, At: now, Visibility: u.repo.Visibility, CreatedAt: u.repo.CreatedAt}).Once()
93	got, err := u.mod.Approve(ctx, domain.Owner(), u.repo.Name, u.id)	94		got, err := u.mod.Approve(ctx, domain.Owner(), u.repo.Name, u.id)
94	require.NoError(t, err)	95		require.NoError(t, err)

M internal/usecase/issues_integration_test.go +63 -0


@@ -6,14 +6,17 @@ import (
6	6		"context"
7	7		"path/filepath"
8	8		"testing"
	9	+	"time"
9	10
10	11		"github.com/stretchr/testify/assert"
	12	+	"github.com/stretchr/testify/mock"
11	13		"github.com/stretchr/testify/require"
12	14
13	15		"bnema.dev/ermite/internal/adapters/eventbus"
14	16		"bnema.dev/ermite/internal/adapters/kvstore"
15	17		"bnema.dev/ermite/internal/domain"
16	18		"bnema.dev/ermite/internal/ports"
	19	+	"bnema.dev/ermite/internal/ports/mocks"
17	20		"bnema.dev/ermite/internal/usecase"
18	21		)
19	22

@@ -89,3 +92,63 @@ func TestIssuesOverTheStore(t *testing.T) {
89	92		}
90	93		assert.Equal(t, []string{"issue.opened", "issue.commented", "issue.closed", "issue.commented", "issue.reopened"}, kinds)
91	94		}
	95	+
	96	+	// Moderated submissions are private work until approval makes them public activity.
	97	+	func TestModeratedIssueActivityOverStore(t *testing.T) {
	98	+	ctx := t.Context()
	99	+	store, err := kvstore.Open(ctx, t.TempDir())
	100	+	require.NoError(t, err)
	101	+	t.Cleanup(func() { require.NoError(t, store.Close(ctx)) })
	102	+	n := name(t, "proj")
	103	+	initial := now.Add(-time.Hour)
	104	+	require.NoError(t, store.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, CreatedAt: initial, ActiveAt: initial, Settings: withIssues(domain.IssuesPublic)}))
	105	+	clock := mocks.NewClock(t)
	106	+	at := now
	107	+	clock.EXPECT().Now().RunAndReturn(func() time.Time { return at }).Maybe()
	108	+	keys, err := usecase.NewIssueKeys(ctx, store.IssueKeys())
	109	+	require.NoError(t, err)
	110	+	limiter := mocks.NewSubmissionLimiter(t)
	111	+	limiter.EXPECT().Recent(mock.Anything, mock.Anything).Return(0, 0).Maybe()
	112	+	limiter.EXPECT().Reserve(mock.Anything, mock.Anything).Return(true, time.Duration(0), func() {}).Maybe()
	113	+	bus := mocks.NewEventBus(t)
	114	+	var kinds []string
	115	+	bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(_ context.Context, ev domain.Event) {
	116	+	repo, err := store.Get(ctx, n)
	117	+	require.NoError(t, err)
	118	+	if ev.EventKind() == "issue.opened" {
	119	+	require.Equal(t, at, repo.ActiveAt)
	120	+	} else {
	121	+	require.Equal(t, initial, repo.ActiveAt)
	122	+	}
	123	+	kinds = append(kinds, ev.EventKind())
	124	+	}).Maybe()
	125	+	submissions := usecase.NewIssueSubmissions(usecase.IssueSubmissionsConfig{Meta: store, Pending: store.PendingIssues(), Bans: store.IssueBans(), Limiter: limiter, Keys: keys, Bus: bus, Clock: clock, BaseDifficulty: testBits, MaxDifficulty: testBits})
	126	+	moderation := usecase.NewIssueModeration(usecase.IssueModerationConfig{Meta: store, Pending: store.PendingIssues(), Bans: store.IssueBans(), Bus: bus, Clock: clock})
	127	+	submit := func() domain.PendingIssueID {
	128	+	ch, err := submissions.Challenge(ctx, n, "client")
	129	+	require.NoError(t, err)
	130	+	sub := ports.IssueSubmission{Token: ch.Token, FormToken: ch.FormToken, Solution: solve(t, ch), Title: "Bug", Body: "Details"}
	131	+	at = at.Add(usecase.MinFillTime)
	132	+	require.NoError(t, submissions.Submit(ctx, n, "client", sub))
	133	+	repo, err := store.Get(ctx, n)
	134	+	require.NoError(t, err)
	135	+	require.Equal(t, initial, repo.ActiveAt)
	136	+	page, err := store.PendingIssues().List(ctx, n, ports.ListPendingParams{PageSize: 10})
	137	+	require.NoError(t, err)
	138	+	require.Len(t, page.Issues, 1)
	139	+	return page.Issues[0].ID
	140	+	}
	141	+	id := submit()
	142	+	require.NoError(t, moderation.Reject(ctx, domain.Owner(), n, id, false))
	143	+	repo, err := store.Get(ctx, n)
	144	+	require.NoError(t, err)
	145	+	require.Equal(t, initial, repo.ActiveAt)
	146	+	id = submit()
	147	+	_, err = moderation.Approve(ctx, domain.Owner(), n, id)
	148	+	require.NoError(t, err)
	149	+	repo, err = store.Get(ctx, n)
	150	+	require.NoError(t, err)
	151	+	require.Equal(t, at, repo.ActiveAt)
	152	+	require.True(t, repo.PushedAt.IsZero())
	153	+	require.Equal(t, []string{"issue.submitted", "issue.pending_removed", "issue.submitted", "issue.opened"}, kinds)
	154	+	}


@@ -6,14 +6,17 @@ import (
6	"context"	6		"context"
7	"path/filepath"	7		"path/filepath"
8	"testing"	8		"testing"
		9	+	"time"
9		10
10	"github.com/stretchr/testify/assert"	11		"github.com/stretchr/testify/assert"
		12	+	"github.com/stretchr/testify/mock"
11	"github.com/stretchr/testify/require"	13		"github.com/stretchr/testify/require"
12		14
13	"bnema.dev/ermite/internal/adapters/eventbus"	15		"bnema.dev/ermite/internal/adapters/eventbus"
14	"bnema.dev/ermite/internal/adapters/kvstore"	16		"bnema.dev/ermite/internal/adapters/kvstore"
15	"bnema.dev/ermite/internal/domain"	17		"bnema.dev/ermite/internal/domain"
16	"bnema.dev/ermite/internal/ports"	18		"bnema.dev/ermite/internal/ports"
		19	+	"bnema.dev/ermite/internal/ports/mocks"
17	"bnema.dev/ermite/internal/usecase"	20		"bnema.dev/ermite/internal/usecase"
18	)	21		)
19		22

@@ -89,3 +92,63 @@ func TestIssuesOverTheStore(t *testing.T) {
89	}	92		}
90	assert.Equal(t, []string{"issue.opened", "issue.commented", "issue.closed", "issue.commented", "issue.reopened"}, kinds)	93		assert.Equal(t, []string{"issue.opened", "issue.commented", "issue.closed", "issue.commented", "issue.reopened"}, kinds)
91	}	94		}
		95	+
		96	+	// Moderated submissions are private work until approval makes them public activity.
		97	+	func TestModeratedIssueActivityOverStore(t *testing.T) {
		98	+	ctx := t.Context()
		99	+	store, err := kvstore.Open(ctx, t.TempDir())
		100	+	require.NoError(t, err)
		101	+	t.Cleanup(func() { require.NoError(t, store.Close(ctx)) })
		102	+	n := name(t, "proj")
		103	+	initial := now.Add(-time.Hour)
		104	+	require.NoError(t, store.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, CreatedAt: initial, ActiveAt: initial, Settings: withIssues(domain.IssuesPublic)}))
		105	+	clock := mocks.NewClock(t)
		106	+	at := now
		107	+	clock.EXPECT().Now().RunAndReturn(func() time.Time { return at }).Maybe()
		108	+	keys, err := usecase.NewIssueKeys(ctx, store.IssueKeys())
		109	+	require.NoError(t, err)
		110	+	limiter := mocks.NewSubmissionLimiter(t)
		111	+	limiter.EXPECT().Recent(mock.Anything, mock.Anything).Return(0, 0).Maybe()
		112	+	limiter.EXPECT().Reserve(mock.Anything, mock.Anything).Return(true, time.Duration(0), func() {}).Maybe()
		113	+	bus := mocks.NewEventBus(t)
		114	+	var kinds []string
		115	+	bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(_ context.Context, ev domain.Event) {
		116	+	repo, err := store.Get(ctx, n)
		117	+	require.NoError(t, err)
		118	+	if ev.EventKind() == "issue.opened" {
		119	+	require.Equal(t, at, repo.ActiveAt)
		120	+	} else {
		121	+	require.Equal(t, initial, repo.ActiveAt)
		122	+	}
		123	+	kinds = append(kinds, ev.EventKind())
		124	+	}).Maybe()
		125	+	submissions := usecase.NewIssueSubmissions(usecase.IssueSubmissionsConfig{Meta: store, Pending: store.PendingIssues(), Bans: store.IssueBans(), Limiter: limiter, Keys: keys, Bus: bus, Clock: clock, BaseDifficulty: testBits, MaxDifficulty: testBits})
		126	+	moderation := usecase.NewIssueModeration(usecase.IssueModerationConfig{Meta: store, Pending: store.PendingIssues(), Bans: store.IssueBans(), Bus: bus, Clock: clock})
		127	+	submit := func() domain.PendingIssueID {
		128	+	ch, err := submissions.Challenge(ctx, n, "client")
		129	+	require.NoError(t, err)
		130	+	sub := ports.IssueSubmission{Token: ch.Token, FormToken: ch.FormToken, Solution: solve(t, ch), Title: "Bug", Body: "Details"}
		131	+	at = at.Add(usecase.MinFillTime)
		132	+	require.NoError(t, submissions.Submit(ctx, n, "client", sub))
		133	+	repo, err := store.Get(ctx, n)
		134	+	require.NoError(t, err)
		135	+	require.Equal(t, initial, repo.ActiveAt)
		136	+	page, err := store.PendingIssues().List(ctx, n, ports.ListPendingParams{PageSize: 10})
		137	+	require.NoError(t, err)
		138	+	require.Len(t, page.Issues, 1)
		139	+	return page.Issues[0].ID
		140	+	}
		141	+	id := submit()
		142	+	require.NoError(t, moderation.Reject(ctx, domain.Owner(), n, id, false))
		143	+	repo, err := store.Get(ctx, n)
		144	+	require.NoError(t, err)
		145	+	require.Equal(t, initial, repo.ActiveAt)
		146	+	id = submit()
		147	+	_, err = moderation.Approve(ctx, domain.Owner(), n, id)
		148	+	require.NoError(t, err)
		149	+	repo, err = store.Get(ctx, n)
		150	+	require.NoError(t, err)
		151	+	require.Equal(t, at, repo.ActiveAt)
		152	+	require.True(t, repo.PushedAt.IsZero())
		153	+	require.Equal(t, []string{"issue.submitted", "issue.pending_removed", "issue.submitted", "issue.opened"}, kinds)
		154	+	}

M internal/usecase/issues_test.go +2 -2


@@ -39,7 +39,7 @@ func newIssueUnit(t testing.T, vis domain.Visibility, mode domain.IssuesMode)
39	39		clock := mocks.NewClock(t)
40	40		clock.EXPECT().Now().Return(now).Maybe()
41	41		u.meta.EXPECT().Get(mock.Anything, u.repo.Name).Return(u.repo, nil).Maybe()
42		-	u.meta.EXPECT().Update(mock.Anything, mock.MatchedBy(func(r domain.Repo) bool { return r.Name == u.repo.Name && r.ActiveAt.Equal(now) && r.PushedAt.IsZero() })).Return(nil).Maybe()
	42	+	expectRepoMutation(t, u.meta, u.repo, mock.MatchedBy(func(r domain.Repo) bool { return r.ActiveAt.Equal(now) && r.PushedAt.IsZero() })).Maybe()
43	43		u.issues = usecase.NewIssues(usecase.IssuesConfig{Meta: u.meta, Store: u.store, MergeRequests: u.mrs, Bus: u.bus, Clock: clock})
44	44		return u
45	45		}

@@ -117,7 +117,7 @@ func TestIssueCreateAndTransitions(t *testing.T) {
117	117		got, err := u.issues.Create(ctx, p, u.repo.Name, ports.CreateIssueParams{Title: " Crash ", Body: "why"})
118	118		require.NoError(t, err)
119	119		assert.EqualValues(t, 7, got.Number)
120		-	u.meta.AssertCalled(t, "Update", mock.Anything, mock.MatchedBy(func(r domain.Repo) bool { return r.ActiveAt.Equal(now) && r.PushedAt.IsZero() }))
	120	+	u.meta.AssertCalled(t, "Mutate", mock.Anything, u.repo.Name, mock.Anything)
121	121		})
122	122
123	123		t.Run("invalid text stores nothing", func(t *testing.T) {


@@ -39,7 +39,7 @@ func newIssueUnit(t testing.T, vis domain.Visibility, mode domain.IssuesMode)
39		clock := mocks.NewClock(t)	39		clock := mocks.NewClock(t)
40		clock.EXPECT().Now().Return(now).Maybe()	40		clock.EXPECT().Now().Return(now).Maybe()
41		u.meta.EXPECT().Get(mock.Anything, u.repo.Name).Return(u.repo, nil).Maybe()	41		u.meta.EXPECT().Get(mock.Anything, u.repo.Name).Return(u.repo, nil).Maybe()
42	-	u.meta.EXPECT().Update(mock.Anything, mock.MatchedBy(func(r domain.Repo) bool { return r.Name == u.repo.Name && r.ActiveAt.Equal(now) && r.PushedAt.IsZero() })).Return(nil).Maybe()	42	+	expectRepoMutation(t, u.meta, u.repo, mock.MatchedBy(func(r domain.Repo) bool { return r.ActiveAt.Equal(now) && r.PushedAt.IsZero() })).Maybe()
43		u.issues = usecase.NewIssues(usecase.IssuesConfig{Meta: u.meta, Store: u.store, MergeRequests: u.mrs, Bus: u.bus, Clock: clock})	43		u.issues = usecase.NewIssues(usecase.IssuesConfig{Meta: u.meta, Store: u.store, MergeRequests: u.mrs, Bus: u.bus, Clock: clock})
44		return u	44		return u
45		}	45		}

@@ -117,7 +117,7 @@ func TestIssueCreateAndTransitions(t *testing.T) {
117		got, err := u.issues.Create(ctx, p, u.repo.Name, ports.CreateIssueParams{Title: " Crash ", Body: "why"})	117		got, err := u.issues.Create(ctx, p, u.repo.Name, ports.CreateIssueParams{Title: " Crash ", Body: "why"})
118		require.NoError(t, err)	118		require.NoError(t, err)
119		assert.EqualValues(t, 7, got.Number)	119		assert.EqualValues(t, 7, got.Number)
120	-	u.meta.AssertCalled(t, "Update", mock.Anything, mock.MatchedBy(func(r domain.Repo) bool { return r.ActiveAt.Equal(now) && r.PushedAt.IsZero() }))	120	+	u.meta.AssertCalled(t, "Mutate", mock.Anything, u.repo.Name, mock.Anything)
121		})	121		})
122			122
123		t.Run("invalid text stores nothing", func(t *testing.T) {	123		t.Run("invalid text stores nothing", func(t *testing.T) {

M internal/usecase/pushes.go +14 -28


@@ -15,9 +15,6 @@ import (
15	15		"bnema.dev/ermite/internal/ports"
16	16		)
17	17
18		-	// pushRetries bounds the retries of a metadata update that loses a race.
19		-	const pushRetries = 5
20		-
21	18		// Pushes are the use cases around a Git push.
22	19		type Pushes struct {
23	20		meta ports.RepoMetaStore

@@ -248,39 +245,28 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep
248	245		// touchPushedAt stores pushed_at and active_at, preserving later timestamps
249	246		// and retrying when another write got in between.
250	247		// A repository deleted meanwhile is not an error. It returns the repository
251		-	// as read, or the zero value when it is unknown. A push (Pushes.Record) and a
252		-	// merge, which moves a branch too, both record through it.
	248	+	// as stored after mutation, or the zero value when it is unknown. A push
	249	+	// (Pushes.Record) and a merge, which moves a branch too, both record through it.
253	250		func touchPushedAt(ctx context.Context, meta ports.RepoMetaStore, name domain.RepoName, now time.Time) (domain.Repo, error) {
254		-	for range pushRetries {
255		-	repo, err := meta.Get(ctx, name)
256		-	if errors.Is(err, domain.ErrNotFound) {
257		-	return domain.Repo{}, nil
258		-	}
259		-	if err != nil {
260		-	return domain.Repo{}, fmt.Errorf("get repository %q: %w", name, err)
261		-	}
262		-	if !now.After(repo.PushedAt) && !now.After(repo.ActiveAt) {
263		-	return repo, nil // a later push already recorded
264		-	}
	251	+	repo, err := meta.Mutate(ctx, name, func(repo *domain.Repo) (bool, error) {
	252	+	changed := false
265	253		if now.After(repo.PushedAt) {
266	254		repo.PushedAt = now
	255	+	changed = true
267	256		}
268	257		if now.After(repo.ActiveAt) {
269	258		repo.ActiveAt = now
	259	+	changed = true
270	260		}
271		-	err = meta.Update(ctx, repo)
272		-	if errors.Is(err, domain.ErrConflict) {
273		-	continue
274		-	}
275		-	if errors.Is(err, domain.ErrNotFound) {
276		-	return domain.Repo{}, nil
277		-	}
278		-	if err != nil {
279		-	return repo, fmt.Errorf("update repository %q: %w", name, err)
280		-	}
281		-	return repo, nil
	261	+	return changed, nil
	262	+	})
	263	+	if errors.Is(err, domain.ErrNotFound) {
	264	+	return domain.Repo{}, nil
	265	+	}
	266	+	if err != nil {
	267	+	return repo, fmt.Errorf("update repository %q: %w", name, err)
282	268		}
283		-	return domain.Repo{}, fmt.Errorf("update pushed_at of %q: %w", name, domain.ErrConflict)
	269	+	return repo, nil
284	270		}
285	271
286	272		// fixDefaultBranch points HEAD to a branch that exists when it points to a

 	"bnema.dev/ermite/internal/ports"
 )
-// pushRetries bounds the retries of a metadata update that loses a race.
-const pushRetries = 5
 // Pushes are the use cases around a Git push.
 type Pushes struct {
 	meta  ports.RepoMetaStore
 // touchPushedAt stores pushed_at and active_at, preserving later timestamps
 // and 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. A push (Pushes.Record) and a
+// as stored after mutation, or the zero value when it is unknown. A push
-// merge, which moves a branch too, both record through it.
+// (Pushes.Record) and a 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 := meta.Mutate(ctx, name, func(repo *domain.Repo) (bool, error) {
-		repo, err := meta.Get(ctx, name)
+		changed := false
-		if errors.Is(err, domain.ErrNotFound) {
-			return domain.Repo{}, nil
-		}
-		if err != nil {
-			return domain.Repo{}, fmt.Errorf("get repository %q: %w", name, err)
-		}
-		if !now.After(repo.PushedAt) && !now.After(repo.ActiveAt) {
-			return repo, nil // a later push already recorded
-		}
 		if now.After(repo.PushedAt) {
 			repo.PushedAt = now
+			changed = true
 		}
 		if now.After(repo.ActiveAt) {
 			repo.ActiveAt = now
+			changed = true
 		}
-		err = meta.Update(ctx, repo)
+		return changed, nil
-		if errors.Is(err, domain.ErrConflict) {
+	})
-			continue
+	if errors.Is(err, domain.ErrNotFound) {
-		}
+		return domain.Repo{}, nil
-		if errors.Is(err, domain.ErrNotFound) {
+	}
-			return domain.Repo{}, nil
+	if err != nil {
-		}
+		return repo, fmt.Errorf("update repository %q: %w", name, err)
-		if err != nil {
-			return repo, fmt.Errorf("update repository %q: %w", name, err)
-		}
-		return repo, nil
 	}
-	return domain.Repo{}, fmt.Errorf("update pushed_at of %q: %w", name, domain.ErrConflict)
+	return repo, nil
 }
 // fixDefaultBranch points HEAD to a branch that exists when it points to a

M internal/usecase/pushes_test.go +10 -15


@@ -305,8 +305,7 @@ func TestPushesRecord(t *testing.T) {
305	305		f := newPushFixture(t)
306	306		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
307	307		repo := domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: now.Add(-time.Hour)}
308		-	f.meta.EXPECT().Get(ctx, n).Return(repo, nil)
309		-	f.meta.EXPECT().Update(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: repo.CreatedAt, PushedAt: now, ActiveAt: now}).Return(nil)
	308	+	expectRepoMutation(t, f.meta, repo, domain.Repo{Name: n, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, PushedAt: now, ActiveAt: now}).Once()
310	309		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
311	310		f.bus.EXPECT().Publish(ctx, domain.PushEvent{Repo: n, Updates: ups, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}).Return()
312	311

@@ -320,22 +319,18 @@ func TestPushesRecord(t *testing.T) {
320	319		fresh := mocks.NewRepoFreshener(t)
321	320		fresh.EXPECT().Refresh(ctx, n).Return().Once()
322	321		f.pushes.WithFreshener(fresh)
323		-	f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{Name: n}, nil)
324		-	f.meta.EXPECT().Update(ctx, mock.Anything).Return(nil)
	322	+	expectPushMutation(t, f.meta, domain.Repo{Name: n}, nil)
325	323		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
326	324		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
327	325		require.NoError(t, f.pushes.Record(ctx, p, n, ups))
328	326		})
329	327
330		-	t.Run("pushed_at retries after a conflict and never goes backwards", func(t *testing.T) {
	328	+	t.Run("pushed_at preserves a later push", func(t *testing.T) {
331	329		t.Parallel()
332	330		f := newPushFixture(t)
333	331		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
334		-	repo := domain.Repo{Name: n}
335		-	f.meta.EXPECT().Get(ctx, n).Return(repo, nil).Once()
336		-	f.meta.EXPECT().Update(ctx, mock.Anything).Return(domain.ErrConflict).Once()
337	332		later := domain.Repo{Name: n, PushedAt: now.Add(time.Second), ActiveAt: now.Add(time.Second)}
338		-	f.meta.EXPECT().Get(ctx, n).Return(later, nil).Once() // a later push recorded meanwhile
	333	+	expectPushMutation(t, f.meta, later, nil) // a later push recorded meanwhile
339	334		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
340	335		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
341	336		require.NoError(t, f.pushes.Record(ctx, p, n, ups))

@@ -345,7 +340,7 @@ func TestPushesRecord(t *testing.T) {
345	340		t.Parallel()
346	341		f := newPushFixture(t)
347	342		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
348		-	f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{}, domain.ErrNotFound)
	343	+	expectPushMutation(t, f.meta, domain.Repo{}, domain.ErrNotFound)
349	344		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
350	345		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
351	346		require.NoError(t, f.pushes.Record(ctx, p, n, ups))

@@ -355,7 +350,7 @@ func TestPushesRecord(t *testing.T) {
355	350		t.Parallel()
356	351		f := newPushFixture(t)
357	352		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
358		-	f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{}, domain.ErrNotFound)
	353	+	expectPushMutation(t, f.meta, domain.Repo{}, domain.ErrNotFound)
359	354		f.git.EXPECT().DefaultBranch(ctx, n).Return("", false, domain.ErrNotFound)
360	355		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
361	356		require.NoError(t, f.pushes.Record(ctx, p, n, ups))

@@ -365,7 +360,7 @@ func TestPushesRecord(t *testing.T) {
365	360		t.Parallel()
366	361		f := newPushFixture(t)
367	362		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
368		-	f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{}, domain.ErrNotFound)
	363	+	expectPushMutation(t, f.meta, domain.Repo{}, domain.ErrNotFound)
369	364		f.git.EXPECT().DefaultBranch(ctx, n).Return("trunk", false, nil)
370	365		f.git.EXPECT().SetDefaultBranch(ctx, n, "main").Return(domain.ErrNotFound)
371	366		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()

@@ -377,7 +372,7 @@ func TestPushesRecord(t *testing.T) {
377	372		f := newPushFixture(t)
378	373		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
379	374		boom := errors.New("boom")
380		-	f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{}, boom)
	375	+	expectPushMutation(t, f.meta, domain.Repo{}, boom)
381	376		f.git.EXPECT().DefaultBranch(ctx, n).Return("", false, boom)
382	377		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
383	378		err := f.pushes.Record(ctx, p, n, ups)

@@ -414,7 +409,7 @@ func TestPushesDefaultBranchRule(t *testing.T) {
414	409		t.Parallel()
415	410		f := newPushFixture(t)
416	411		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
417		-	f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{Name: n, PushedAt: now, ActiveAt: now}, nil)
	412	+	expectPushMutation(t, f.meta, domain.Repo{Name: n, PushedAt: now, ActiveAt: now}, nil)
418	413		f.git.EXPECT().DefaultBranch(ctx, n).Return(tt.head, tt.exists, nil)
419	414		if tt.want != "" {
420	415		f.git.EXPECT().SetDefaultBranch(ctx, n, tt.want).Return(nil)

@@ -429,7 +424,7 @@ func TestPushesDefaultBranchRule(t *testing.T) {
429	424		f := newPushFixture(t)
430	425		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
431	426		boom := errors.New("boom")
432		-	f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{Name: n, PushedAt: now, ActiveAt: now}, nil)
	427	+	expectPushMutation(t, f.meta, domain.Repo{Name: n, PushedAt: now, ActiveAt: now}, nil)
433	428		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", false, nil)
434	429		f.git.EXPECT().SetDefaultBranch(ctx, n, "dev").Return(boom)
435	430		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()

 		f := newPushFixture(t)
 		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
 		repo := domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: now.Add(-time.Hour)}
-		f.meta.EXPECT().Get(ctx, n).Return(repo, nil)
+		expectRepoMutation(t, f.meta, repo, domain.Repo{Name: n, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, PushedAt: now, ActiveAt: now}).Once()
-		f.meta.EXPECT().Update(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: repo.CreatedAt, PushedAt: now, ActiveAt: now}).Return(nil)
 		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
 		f.bus.EXPECT().Publish(ctx, domain.PushEvent{Repo: n, Updates: ups, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}).Return()
 		fresh := mocks.NewRepoFreshener(t)
 		fresh.EXPECT().Refresh(ctx, n).Return().Once()
 		f.pushes.WithFreshener(fresh)
-		f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{Name: n}, nil)
+		expectPushMutation(t, f.meta, domain.Repo{Name: n}, nil)
-		f.meta.EXPECT().Update(ctx, mock.Anything).Return(nil)
 		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
 		require.NoError(t, f.pushes.Record(ctx, p, n, ups))
 	})
-	t.Run("pushed_at retries after a conflict and never goes backwards", func(t *testing.T) {
+	t.Run("pushed_at preserves a later push", func(t *testing.T) {
 		t.Parallel()
 		f := newPushFixture(t)
 		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
-		repo := domain.Repo{Name: n}
-		f.meta.EXPECT().Get(ctx, n).Return(repo, nil).Once()
-		f.meta.EXPECT().Update(ctx, mock.Anything).Return(domain.ErrConflict).Once()
 		later := domain.Repo{Name: n, PushedAt: now.Add(time.Second), ActiveAt: now.Add(time.Second)}
-		f.meta.EXPECT().Get(ctx, n).Return(later, nil).Once() // a later push recorded meanwhile
+		expectPushMutation(t, f.meta, later, nil) // a later push recorded meanwhile
 		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
 		require.NoError(t, f.pushes.Record(ctx, p, n, ups))
 		t.Parallel()
 		f := newPushFixture(t)
 		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
-		f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{}, domain.ErrNotFound)
+		expectPushMutation(t, f.meta, domain.Repo{}, domain.ErrNotFound)
 		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
 		require.NoError(t, f.pushes.Record(ctx, p, n, ups))
 		t.Parallel()
 		f := newPushFixture(t)
 		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
-		f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{}, domain.ErrNotFound)
+		expectPushMutation(t, f.meta, domain.Repo{}, domain.ErrNotFound)
 		f.git.EXPECT().DefaultBranch(ctx, n).Return("", false, domain.ErrNotFound)
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
 		require.NoError(t, f.pushes.Record(ctx, p, n, ups))
 		t.Parallel()
 		f := newPushFixture(t)
 		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
-		f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{}, domain.ErrNotFound)
+		expectPushMutation(t, f.meta, domain.Repo{}, domain.ErrNotFound)
 		f.git.EXPECT().DefaultBranch(ctx, n).Return("trunk", false, nil)
 		f.git.EXPECT().SetDefaultBranch(ctx, n, "main").Return(domain.ErrNotFound)
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
 		f := newPushFixture(t)
 		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
 		boom := errors.New("boom")
-		f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{}, boom)
+		expectPushMutation(t, f.meta, domain.Repo{}, boom)
 		f.git.EXPECT().DefaultBranch(ctx, n).Return("", false, boom)
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()
 		err := f.pushes.Record(ctx, p, n, ups)
 			t.Parallel()
 			f := newPushFixture(t)
 			f.sched.EXPECT().Schedule(ctx, n).Return().Once()
-			f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{Name: n, PushedAt: now, ActiveAt: now}, nil)
+			expectPushMutation(t, f.meta, domain.Repo{Name: n, PushedAt: now, ActiveAt: now}, nil)
 			f.git.EXPECT().DefaultBranch(ctx, n).Return(tt.head, tt.exists, nil)
 			if tt.want != "" {
 				f.git.EXPECT().SetDefaultBranch(ctx, n, tt.want).Return(nil)
 		f := newPushFixture(t)
 		f.sched.EXPECT().Schedule(ctx, n).Return().Once()
 		boom := errors.New("boom")
-		f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{Name: n, PushedAt: now, ActiveAt: now}, nil)
+		expectPushMutation(t, f.meta, domain.Repo{Name: n, PushedAt: now, ActiveAt: now}, nil)
 		f.git.EXPECT().DefaultBranch(ctx, n).Return("main", false, nil)
 		f.git.EXPECT().SetDefaultBranch(ctx, n, "dev").Return(boom)
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return()

A internal/usecase/repo_mutation_test.go +34 -0

@@ -0,0 +1,34 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+
	9	+	"github.com/stretchr/testify/mock"
	10	+	"github.com/stretchr/testify/require"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	"bnema.dev/ermite/internal/ports/mocks"
	14	+	)
	15	+
	16	+	func expectRepoMutation(t testing.T, meta mocks.RepoMetaStore, repo domain.Repo, want any) *mocks.RepoMetaStore_Mutate_Call {
	17	+	t.Helper()
	18	+	return meta.EXPECT().Mutate(mock.Anything, repo.Name, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.RepoName, fn func(*domain.Repo) (bool, error)) (domain.Repo, error) {
	19	+	result := repo
	20	+	_, err := fn(&result)
	21	+	require.NoError(t, err)
	22	+	_, differences := (mock.Arguments{want}).Diff([]any{result})
	23	+	require.Zero(t, differences, "unexpected mutated repository: %+v", result)
	24	+	return result, err
	25	+	})
	26	+	}
	27	+
	28	+	func expectPushMutation(t testing.T, meta mocks.RepoMetaStore, repo domain.Repo, err error) *mocks.RepoMetaStore_Mutate_Call {
	29	+	t.Helper()
	30	+	if err != nil {
	31	+	return meta.EXPECT().Mutate(mock.Anything, mock.Anything, mock.Anything).Return(repo, err)
	32	+	}
	33	+	return expectRepoMutation(t, meta, repo, mock.Anything)
	34	+	}

@@ -0,0 +1,34 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+
	9	+	"github.com/stretchr/testify/mock"
	10	+	"github.com/stretchr/testify/require"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	"bnema.dev/ermite/internal/ports/mocks"
	14	+	)
	15	+
	16	+	func expectRepoMutation(t testing.T, meta mocks.RepoMetaStore, repo domain.Repo, want any) *mocks.RepoMetaStore_Mutate_Call {
	17	+	t.Helper()
	18	+	return meta.EXPECT().Mutate(mock.Anything, repo.Name, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.RepoName, fn func(*domain.Repo) (bool, error)) (domain.Repo, error) {
	19	+	result := repo
	20	+	_, err := fn(&result)
	21	+	require.NoError(t, err)
	22	+	_, differences := (mock.Arguments{want}).Diff([]any{result})
	23	+	require.Zero(t, differences, "unexpected mutated repository: %+v", result)
	24	+	return result, err
	25	+	})
	26	+	}
	27	+
	28	+	func expectPushMutation(t testing.T, meta mocks.RepoMetaStore, repo domain.Repo, err error) *mocks.RepoMetaStore_Mutate_Call {
	29	+	t.Helper()
	30	+	if err != nil {
	31	+	return meta.EXPECT().Mutate(mock.Anything, mock.Anything, mock.Anything).Return(repo, err)
	32	+	}
	33	+	return expectRepoMutation(t, meta, repo, mock.Anything)
	34	+	}

M internal/usecase/repos_write.go +6 -14


@@ -155,11 +155,7 @@ func (r *Repos) Update(ctx context.Context, p domain.Principal, name domain.Repo
155	155		// updateMeta applies the metadata fields, retrying when a write (a push
156	156		// recording pushed_at) gets in between.
157	157		func (r *Repos) updateMeta(ctx context.Context, name domain.RepoName, params ports.UpdateRepoParams) (domain.Repo, error) {
158		-	for range pushRetries {
159		-	repo, err := r.meta.Get(ctx, name)
160		-	if err != nil {
161		-	return domain.Repo{}, err
162		-	}
	158	+	repo, err := r.meta.Mutate(ctx, name, func(repo *domain.Repo) (bool, error) {
163	159		if params.Description != nil {
164	160		repo.Description = *params.Description
165	161		}

@@ -181,16 +177,12 @@ func (r *Repos) updateMeta(ctx context.Context, name domain.RepoName, params por
181	177		if params.Issues != nil {
182	178		repo.Settings.Issues = *params.Issues
183	179		}
184		-	err = r.meta.Update(ctx, repo)
185		-	if errors.Is(err, domain.ErrConflict) {
186		-	continue
187		-	}
188		-	if err != nil {
189		-	return domain.Repo{}, fmt.Errorf("update repository %q: %w", name, err)
190		-	}
191		-	return repo, nil
	180	+	return true, nil
	181	+	})
	182	+	if err != nil {
	183	+	return domain.Repo{}, fmt.Errorf("update repository %q: %w", name, err)
192	184		}
193		-	return domain.Repo{}, fmt.Errorf("update repository %q: %w", name, domain.ErrConflict)
	185	+	return repo, nil
194	186		}
195	187
196	188		// Delete moves the repository directory to the trash, removes its metadata

 // updateMeta applies the metadata fields, retrying when a write (a push
 // recording pushed_at) gets in between.
 func (r *Repos) updateMeta(ctx context.Context, name domain.RepoName, params ports.UpdateRepoParams) (domain.Repo, error) {
-	for range pushRetries {
+	repo, err := r.meta.Mutate(ctx, name, func(repo *domain.Repo) (bool, error) {
-		repo, err := r.meta.Get(ctx, name)
-		if err != nil {
-			return domain.Repo{}, err
-		}
 		if params.Description != nil {
 			repo.Description = *params.Description
 		}
 		if params.Issues != nil {
 			repo.Settings.Issues = *params.Issues
 		}
-		err = r.meta.Update(ctx, repo)
+		return true, nil
-		if errors.Is(err, domain.ErrConflict) {
+	})
-			continue
+	if err != nil {
-		}
+		return domain.Repo{}, fmt.Errorf("update repository %q: %w", name, err)
-		if err != nil {
-			return domain.Repo{}, fmt.Errorf("update repository %q: %w", name, err)
-		}
-		return repo, nil
 	}
-	return domain.Repo{}, fmt.Errorf("update repository %q: %w", name, domain.ErrConflict)
+	return repo, nil
 }
 // Delete moves the repository directory to the trash, removes its metadata

M internal/usecase/repos_write_test.go +20 -21


@@ -30,9 +30,9 @@ func TestUpdate(t *testing.T) {
30	30		t.Run("description and visibility are stored and RepoUpdated is published", func(t *testing.T) {
31	31		t.Parallel()
32	32		f := newFixture(t)
33		-	f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
	33	+	f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Once()
34	34		want := domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "new", Settings: domain.DefaultRepoSettings()}
35		-	f.meta.EXPECT().Update(ctx, want).Return(nil)
	35	+	expectRepoMutation(t, f.meta, priv, want)
36	36		f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{
37	37		Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate,
38	38		}).Return().Once()

@@ -47,11 +47,11 @@ func TestUpdate(t *testing.T) {
47	47		t.Run("topics are normalized and stored, and RepoUpdated is published", func(t *testing.T) {
48	48		t.Parallel()
49	49		f := newFixture(t)
50		-	f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
	50	+	f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Once()
51	51		want := priv
52	52		want.Topics = []string{"cli", "go"}
53	53		want.Settings = domain.DefaultRepoSettings()
54		-	f.meta.EXPECT().Update(ctx, want).Return(nil)
	54	+	expectRepoMutation(t, f.meta, priv, want)
55	55		f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{
56	56		Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate,
57	57		}).Return().Once()

@@ -66,10 +66,10 @@ func TestUpdate(t *testing.T) {
66	66		f := newFixture(t)
67	67		withTopics := priv
68	68		withTopics.Topics = []string{"go"}
69		-	f.meta.EXPECT().Get(ctx, n).Return(withTopics, nil).Twice()
	69	+	f.meta.EXPECT().Get(ctx, n).Return(withTopics, nil).Once()
70	70		want := priv
71	71		want.Settings = domain.DefaultRepoSettings()
72		-	f.meta.EXPECT().Update(ctx, want).Return(nil)
	72	+	expectRepoMutation(t, f.meta, withTopics, want)
73	73		f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
74	74
75	75		got, err := f.repos.Update(ctx, admin, n, ports.UpdateRepoParams{Topics: &[]string{}})

@@ -77,15 +77,12 @@ func TestUpdate(t *testing.T) {
77	77		assert.Nil(t, got.Topics)
78	78		})
79	79
80		-	t.Run("a lost race on the metadata is retried", func(t *testing.T) {
81		-	t.Parallel()
	80	+	t.Run("exhausted mutation conflicts are returned", func(t *testing.T) {
82	81		f := newFixture(t)
83	82		f.meta.EXPECT().Get(ctx, n).Return(priv, nil)
84		-	f.meta.EXPECT().Update(ctx, mock.Anything).Return(domain.ErrConflict).Once()
85		-	f.meta.EXPECT().Update(ctx, mock.Anything).Return(nil).Once()
86		-	f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
	83	+	f.meta.EXPECT().Mutate(ctx, n, mock.Anything).Return(domain.Repo{}, domain.ErrConflict).Once()
87	84		_, err := f.repos.Update(ctx, admin, n, ports.UpdateRepoParams{Description: ptr("x")})
88		-	require.NoError(t, err)
	85	+	require.ErrorIs(t, err, domain.ErrConflict)
89	86		})
90	87
91	88		t.Run("the default branch is checked, then HEAD moves", func(t *testing.T) {

@@ -106,9 +103,11 @@ func TestUpdate(t *testing.T) {
106	103		var order []string
107	104		f.meta.EXPECT().Get(ctx, n).Return(priv, nil)
108	105		f.git.EXPECT().HasBranch(ctx, n, "dev").Return(true, nil)
109		-	f.meta.EXPECT().Update(ctx, mock.Anything).RunAndReturn(func(context.Context, domain.Repo) error {
	106	+	f.meta.EXPECT().Mutate(ctx, n, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.RepoName, fn func(*domain.Repo) (bool, error)) (domain.Repo, error) {
110	107		order = append(order, "meta")
111		-	return nil
	108	+	repo := priv
	109	+	_, err := fn(&repo)
	110	+	return repo, err
112	111		})
113	112		f.git.EXPECT().SetDefaultBranch(ctx, n, "dev").RunAndReturn(func(context.Context, domain.RepoName, string) error {
114	113		order = append(order, "head")

@@ -127,7 +126,7 @@ func TestUpdate(t *testing.T) {
127	126		pub := domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "old"}
128	127		f.meta.EXPECT().Get(ctx, n).Return(pub, nil)
129	128		f.git.EXPECT().HasBranch(ctx, n, "dev").Return(true, nil)
130		-	f.meta.EXPECT().Update(ctx, mock.Anything).Return(nil)
	129	+	expectRepoMutation(t, f.meta, pub, mock.Anything)
131	130		f.git.EXPECT().SetDefaultBranch(ctx, n, "dev").Return(boom)
132	131		f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{
133	132		Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPublic,

@@ -250,10 +249,10 @@ func TestUpdateIssuesMode(t *testing.T) {
250	249		t.Run("public is stored and published", func(t *testing.T) {
251	250		t.Parallel()
252	251		f := newFixture(t)
253		-	f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
254		-	f.meta.EXPECT().Update(ctx, mock.MatchedBy(func(r domain.Repo) bool {
	252	+	f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Once()
	253	+	expectRepoMutation(t, f.meta, priv, mock.MatchedBy(func(r domain.Repo) bool {
255	254		return r.Settings.Issues == domain.IssuesPublic
256		-	})).Return(nil)
	255	+	}))
257	256		f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
258	257		repo, err := f.repos.Update(ctx, domain.Root(), n, ports.UpdateRepoParams{Issues: ptr(domain.IssuesPublic)})
259	258		require.NoError(t, err)

@@ -271,10 +270,10 @@ func TestUpdateIssuesMode(t *testing.T) {
271	270		t.Run("owner is stored and published", func(t *testing.T) {
272	271		t.Parallel()
273	272		f := newFixture(t)
274		-	f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
275		-	f.meta.EXPECT().Update(ctx, mock.MatchedBy(func(r domain.Repo) bool {
	273	+	f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Once()
	274	+	expectRepoMutation(t, f.meta, priv, mock.MatchedBy(func(r domain.Repo) bool {
276	275		return r.Settings.Issues == domain.IssuesOwner && r.Settings.FailingChecksBlock
277		-	})).Return(nil)
	276	+	}))
278	277		f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
279	278		repo, err := f.repos.Update(ctx, domain.Root(), n, ports.UpdateRepoParams{Issues: ptr(domain.IssuesOwner)})
280	279		require.NoError(t, err)

 	t.Run("description and visibility are stored and RepoUpdated is published", func(t *testing.T) {
 		t.Parallel()
 		f := newFixture(t)
-		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
+		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Once()
 		want := domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "new", Settings: domain.DefaultRepoSettings()}
-		f.meta.EXPECT().Update(ctx, want).Return(nil)
+		expectRepoMutation(t, f.meta, priv, want)
 		f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{
 			Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate,
 		}).Return().Once()
 	t.Run("topics are normalized and stored, and RepoUpdated is published", func(t *testing.T) {
 		t.Parallel()
 		f := newFixture(t)
-		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
+		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Once()
 		want := priv
 		want.Topics = []string{"cli", "go"}
 		want.Settings = domain.DefaultRepoSettings()
-		f.meta.EXPECT().Update(ctx, want).Return(nil)
+		expectRepoMutation(t, f.meta, priv, want)
 		f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{
 			Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate,
 		}).Return().Once()
 		f := newFixture(t)
 		withTopics := priv
 		withTopics.Topics = []string{"go"}
-		f.meta.EXPECT().Get(ctx, n).Return(withTopics, nil).Twice()
+		f.meta.EXPECT().Get(ctx, n).Return(withTopics, nil).Once()
 		want := priv
 		want.Settings = domain.DefaultRepoSettings()
-		f.meta.EXPECT().Update(ctx, want).Return(nil)
+		expectRepoMutation(t, f.meta, withTopics, want)
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
 		got, err := f.repos.Update(ctx, admin, n, ports.UpdateRepoParams{Topics: &[]string{}})
 		assert.Nil(t, got.Topics)
 	})
-	t.Run("a lost race on the metadata is retried", func(t *testing.T) {
+	t.Run("exhausted mutation conflicts are returned", func(t *testing.T) {
-		t.Parallel()
 		f := newFixture(t)
 		f.meta.EXPECT().Get(ctx, n).Return(priv, nil)
-		f.meta.EXPECT().Update(ctx, mock.Anything).Return(domain.ErrConflict).Once()
+		f.meta.EXPECT().Mutate(ctx, n, mock.Anything).Return(domain.Repo{}, domain.ErrConflict).Once()
-		f.meta.EXPECT().Update(ctx, mock.Anything).Return(nil).Once()
-		f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
 		_, err := f.repos.Update(ctx, admin, n, ports.UpdateRepoParams{Description: ptr("x")})
-		require.NoError(t, err)
+		require.ErrorIs(t, err, domain.ErrConflict)
 	})
 	t.Run("the default branch is checked, then HEAD moves", func(t *testing.T) {
 		var order []string
 		f.meta.EXPECT().Get(ctx, n).Return(priv, nil)
 		f.git.EXPECT().HasBranch(ctx, n, "dev").Return(true, nil)
-		f.meta.EXPECT().Update(ctx, mock.Anything).RunAndReturn(func(context.Context, domain.Repo) error {
+		f.meta.EXPECT().Mutate(ctx, n, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.RepoName, fn func(*domain.Repo) (bool, error)) (domain.Repo, error) {
 			order = append(order, "meta")
-			return nil
+			repo := priv
+			_, err := fn(&repo)
+			return repo, err
 		})
 		f.git.EXPECT().SetDefaultBranch(ctx, n, "dev").RunAndReturn(func(context.Context, domain.RepoName, string) error {
 			order = append(order, "head")
 		pub := domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "old"}
 		f.meta.EXPECT().Get(ctx, n).Return(pub, nil)
 		f.git.EXPECT().HasBranch(ctx, n, "dev").Return(true, nil)
-		f.meta.EXPECT().Update(ctx, mock.Anything).Return(nil)
+		expectRepoMutation(t, f.meta, pub, mock.Anything)
 		f.git.EXPECT().SetDefaultBranch(ctx, n, "dev").Return(boom)
 		f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{
 			Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPublic,
 	t.Run("public is stored and published", func(t *testing.T) {
 		t.Parallel()
 		f := newFixture(t)
-		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
+		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Once()
-		f.meta.EXPECT().Update(ctx, mock.MatchedBy(func(r domain.Repo) bool {
+		expectRepoMutation(t, f.meta, priv, mock.MatchedBy(func(r domain.Repo) bool {
 			return r.Settings.Issues == domain.IssuesPublic
-		})).Return(nil)
+		}))
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
 		repo, err := f.repos.Update(ctx, domain.Root(), n, ports.UpdateRepoParams{Issues: ptr(domain.IssuesPublic)})
 		require.NoError(t, err)
 	t.Run("owner is stored and published", func(t *testing.T) {
 		t.Parallel()
 		f := newFixture(t)
-		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
+		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Once()
-		f.meta.EXPECT().Update(ctx, mock.MatchedBy(func(r domain.Repo) bool {
+		expectRepoMutation(t, f.meta, priv, mock.MatchedBy(func(r domain.Repo) bool {
 			return r.Settings.Issues == domain.IssuesOwner && r.Settings.FailingChecksBlock
-		})).Return(nil)
+		}))
 		f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
 		repo, err := f.repos.Update(ctx, domain.Root(), n, ports.UpdateRepoParams{Issues: ptr(domain.IssuesOwner)})
 		require.NoError(t, err)