17b5f65 fix(kvstore): bound the numbering locks · ermite

M internal/adapters/kvstore/issues.go +4 -3


@@ -31,14 +31,15 @@ const (
31	31		// Issues is the ermite-kv backed ports.IssueStore. It shares the database of
32	32		// the Store it comes from.
33	33		type Issues struct {
34		-	db *kv.DB
	34	+	db *kv.DB
	35	+	locks *numberLocks
35	36		}
36	37
37	38		var _ ports.IssueStore = (*Issues)(nil)
38	39
39	40		// Issues returns the issue store over the same database. It is valid until the
40	41		// Store is closed.
41		-	func (s Store) Issues() Issues { return &Issues{db: s.db} }
	42	+	func (s Store) Issues() Issues { return &Issues{db: s.db, locks: s.locks} }
42	43
43	44		func issueRepoPrefix(repo domain.RepoName) string { return issuePrefix + repo.String() + "/" }
44	45		func issueKey(repo domain.RepoName, n int64) []byte {

@@ -86,7 +87,7 @@ func issueKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][
86	87		//
87	88		//hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
88	89		func (i *Issues) Create(ctx context.Context, issue domain.Issue) (domain.Issue, error) {
89		-	err := createNumbered(ctx, i.db, issue.Repo, func(b *kv.Batch, n int64) error {
	90	+	err := createNumbered(ctx, i.db, i.locks, issue.Repo, func(b *kv.Batch, n int64) error {
90	91		issue.Number = n
91	92		val, err := encodeIssue(issue)
92	93		if err != nil {

 // Issues is the ermite-kv backed ports.IssueStore. It shares the database of
 // the Store it comes from.
 type Issues struct {
-	db *kv.DB
+	db    *kv.DB
+	locks *numberLocks
 }
 var _ ports.IssueStore = (*Issues)(nil)
 // Issues returns the issue store over the same database. It is valid until the
 // Store is closed.
-func (s *Store) Issues() *Issues { return &Issues{db: s.db} }
+func (s *Store) Issues() *Issues { return &Issues{db: s.db, locks: s.locks} }
 func issueRepoPrefix(repo domain.RepoName) string { return issuePrefix + repo.String() + "/" }
 func issueKey(repo domain.RepoName, n int64) []byte {
 //
 //hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
 func (i *Issues) Create(ctx context.Context, issue domain.Issue) (domain.Issue, error) {
-	err := createNumbered(ctx, i.db, issue.Repo, func(b *kv.Batch, n int64) error {
+	err := createNumbered(ctx, i.db, i.locks, issue.Repo, func(b *kv.Batch, n int64) error {
 		issue.Number = n
 		val, err := encodeIssue(issue)
 		if err != nil {

M internal/adapters/kvstore/issues_pending.go +5 -4


@@ -37,14 +37,15 @@ const (
37	37		// PendingIssues is the ermite-kv backed ports.PendingIssueStore. It shares the
38	38		// database of the Store it comes from.
39	39		type PendingIssues struct {
40		-	db *kv.DB
	40	+	db *kv.DB
	41	+	locks *numberLocks
41	42		}
42	43
43	44		var _ ports.PendingIssueStore = (*PendingIssues)(nil)
44	45
45	46		// PendingIssues returns the pending issue store over the same database. It is
46	47		// valid until the Store is closed.
47		-	func (s Store) PendingIssues() PendingIssues { return &PendingIssues{db: s.db} }
	48	+	func (s Store) PendingIssues() PendingIssues { return &PendingIssues{db: s.db, locks: s.locks} }
48	49
49	50		func pendingRepoPrefix(repo domain.RepoName) string { return pendingPrefix + repo.String() + "/" }
50	51		func pendingKey(repo domain.RepoName, id domain.PendingIssueID) []byte {

@@ -222,7 +223,7 @@ func (s *PendingIssues) Count(ctx context.Context) (int, error) {
222	223		//hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
223	224		func (s *PendingIssues) Approve(ctx context.Context, repo domain.RepoName, id domain.PendingIssueID) (domain.Issue, error) {
224	225		var number int64
225		-	err := createNumbered(ctx, s.db, repo, func(b *kv.Batch, n int64) error {
	226	+	err := createNumbered(ctx, s.db, s.locks, repo, func(b *kv.Batch, n int64) error {
226	227		key := pendingKey(repo, id)
227	228		item, ok := s.db.Get(ctx, key)
228	229		if !ok {

@@ -258,7 +259,7 @@ func (s *PendingIssues) Approve(ctx context.Context, repo domain.RepoName, id do
258	259		}
259	260		return domain.Issue{}, fmt.Errorf("approve pending issue of %q: %w", repo, err)
260	261		}
261		-	return (&Issues{db: s.db}).Get(ctx, repo, number)
	262	+	return (&Issues{db: s.db, locks: s.locks}).Get(ctx, repo, number)
262	263		}
263	264
264	265		// Delete removes the pending record, or returns domain.ErrNotFound.

 // PendingIssues is the ermite-kv backed ports.PendingIssueStore. It shares the
 // database of the Store it comes from.
 type PendingIssues struct {
-	db *kv.DB
+	db    *kv.DB
+	locks *numberLocks
 }
 var _ ports.PendingIssueStore = (*PendingIssues)(nil)
 // PendingIssues returns the pending issue store over the same database. It is
 // valid until the Store is closed.
-func (s *Store) PendingIssues() *PendingIssues { return &PendingIssues{db: s.db} }
+func (s *Store) PendingIssues() *PendingIssues { return &PendingIssues{db: s.db, locks: s.locks} }
 func pendingRepoPrefix(repo domain.RepoName) string { return pendingPrefix + repo.String() + "/" }
 func pendingKey(repo domain.RepoName, id domain.PendingIssueID) []byte {
 //hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
 func (s *PendingIssues) Approve(ctx context.Context, repo domain.RepoName, id domain.PendingIssueID) (domain.Issue, error) {
 	var number int64
-	err := createNumbered(ctx, s.db, repo, func(b *kv.Batch, n int64) error {
+	err := createNumbered(ctx, s.db, s.locks, repo, func(b *kv.Batch, n int64) error {
 		key := pendingKey(repo, id)
 		item, ok := s.db.Get(ctx, key)
 		if !ok {
 		}
 		return domain.Issue{}, fmt.Errorf("approve pending issue of %q: %w", repo, err)
 	}
-	return (&Issues{db: s.db}).Get(ctx, repo, number)
+	return (&Issues{db: s.db, locks: s.locks}).Get(ctx, repo, number)
 }
 // Delete removes the pending record, or returns domain.ErrNotFound.

M internal/adapters/kvstore/mergerequests.go +22 -12


@@ -6,6 +6,7 @@ import (
6	6		"context"
7	7		"errors"
8	8		"fmt"
	9	+	"hash/fnv"
9	10		"net/url"
10	11		"slices"
11	12		"strconv"

@@ -45,14 +46,15 @@ const (
45	46		// MergeRequests is the ermite-kv backed ports.MergeRequestStore. It shares the
46	47		// database of the Store it comes from.
47	48		type MergeRequests struct {
48		-	db *kv.DB
	49	+	db *kv.DB
	50	+	locks *numberLocks
49	51		}
50	52
51	53		var _ ports.MergeRequestStore = (*MergeRequests)(nil)
52	54
53	55		// MergeRequests returns the merge request store over the same database. It is
54	56		// valid until the Store is closed.
55		-	func (s Store) MergeRequests() MergeRequests { return &MergeRequests{db: s.db} }
	57	+	func (s Store) MergeRequests() MergeRequests { return &MergeRequests{db: s.db, locks: s.locks} }
56	58
57	59		func mrRepoPrefix(repo domain.RepoName) string { return mrPrefix + repo.String() + "/" }
58	60		func mrKey(repo domain.RepoName, n int64) []byte {

@@ -199,7 +201,7 @@ func (m MergeRequests) guardRepo(ctx context.Context, b kv.Batch, repo domain.
199	201		//
200	202		//hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
201	203		func (m *MergeRequests) Create(ctx context.Context, mr domain.MergeRequest) (domain.MergeRequest, error) {
202		-	err := createNumbered(ctx, m.db, mr.Repo, func(b *kv.Batch, n int64) error {
	204	+	err := createNumbered(ctx, m.db, m.locks, mr.Repo, func(b *kv.Batch, n int64) error {
203	205		mr.Number = n
204	206		val, err := encodeMergeRequest(mr)
205	207		if err != nil {

@@ -224,11 +226,11 @@ func (m *MergeRequests) Create(ctx context.Context, mr domain.MergeRequest) (dom
224	226		// number. It retries when the batch conflicts; a missing repository gives
225	227		// domain.ErrNotFound.
226	228		//
227		-	// Creations in one repository are serialized in the process (numberLock), so
	229	+	// Creations in one repository are serialized in the process (numberLocks), so
228	230		// a burst of merge requests and issues never exhausts the retries; the version
229	231		// checks still guard against any other writer.
230		-	func createNumbered(ctx context.Context, db kv.DB, repo domain.RepoName, add func(b kv.Batch, number int64) error) error {
231		-	unlock := numberLock(repo)
	232	+	func createNumbered(ctx context.Context, db kv.DB, locks numberLocks, repo domain.RepoName, add func(b *kv.Batch, number int64) error) error {
	233	+	unlock := locks.lock(repo)
232	234		defer unlock()
233	235		for range createTries {
234	236		repoItem, repoOK := db.Get(ctx, repoKey(repo))

@@ -266,13 +268,21 @@ func createNumbered(ctx context.Context, db *kv.DB, repo domain.RepoName, add fu
266	268		return domain.ErrConflict
267	269		}
268	270
269		-	// numberLocks holds one mutex per repository for createNumbered.
270		-	var numberLocks sync.Map // domain.RepoName -> *sync.Mutex
	271	+	// numberLockStripes is the size of the table of numbering locks.
	272	+	const numberLockStripes = 64
271	273
272		-	// numberLock locks the numbering of repo and returns the unlock function.
273		-	func numberLock(repo domain.RepoName) func() {
274		-	v, _ := numberLocks.LoadOrStore(repo, new(sync.Mutex))
275		-	mu := v.(*sync.Mutex)
	274	+	// numberLocks is a fixed table of mutexes, one stripe per hash of the
	275	+	// repository name, for createNumbered: the memory is bounded whatever the
	276	+	// number of repositories, and two repositories sharing a stripe only wait for
	277	+	// each other. One table is shared by every store of a Store, so the merge
	278	+	// requests and the issues of a repository are numbered one at a time.
	279	+	type numberLocks [numberLockStripes]sync.Mutex
	280	+
	281	+	// lock locks the numbering of repo and returns the unlock function.
	282	+	func (l *numberLocks) lock(repo domain.RepoName) func() {
	283	+	h := fnv.New32a()
	284	+	_, _ = h.Write([]byte(repo.String()))
	285	+	mu := &l[h.Sum32()%numberLockStripes]
276	286		mu.Lock()
277	287		return mu.Unlock
278	288		}

 	"context"
 	"errors"
 	"fmt"
+	"hash/fnv"
 	"net/url"
 	"slices"
 	"strconv"
 // MergeRequests is the ermite-kv backed ports.MergeRequestStore. It shares the
 // database of the Store it comes from.
 type MergeRequests struct {
-	db *kv.DB
+	db    *kv.DB
+	locks *numberLocks
 }
 var _ ports.MergeRequestStore = (*MergeRequests)(nil)
 // MergeRequests returns the merge request store over the same database. It is
 // valid until the Store is closed.
-func (s *Store) MergeRequests() *MergeRequests { return &MergeRequests{db: s.db} }
+func (s *Store) MergeRequests() *MergeRequests { return &MergeRequests{db: s.db, locks: s.locks} }
 func mrRepoPrefix(repo domain.RepoName) string { return mrPrefix + repo.String() + "/" }
 func mrKey(repo domain.RepoName, n int64) []byte {
 //
 //hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
 func (m *MergeRequests) Create(ctx context.Context, mr domain.MergeRequest) (domain.MergeRequest, error) {
-	err := createNumbered(ctx, m.db, mr.Repo, func(b *kv.Batch, n int64) error {
+	err := createNumbered(ctx, m.db, m.locks, mr.Repo, func(b *kv.Batch, n int64) error {
 		mr.Number = n
 		val, err := encodeMergeRequest(mr)
 		if err != nil {
 // number. It retries when the batch conflicts; a missing repository gives
 // domain.ErrNotFound.
 //
-// Creations in one repository are serialized in the process (numberLock), so
+// Creations in one repository are serialized in the process (numberLocks), so
 // a burst of merge requests and issues never exhausts the retries; the version
 // checks still guard against any other writer.
-func createNumbered(ctx context.Context, db *kv.DB, repo domain.RepoName, add func(b *kv.Batch, number int64) error) error {
+func createNumbered(ctx context.Context, db *kv.DB, locks *numberLocks, repo domain.RepoName, add func(b *kv.Batch, number int64) error) error {
-	unlock := numberLock(repo)
+	unlock := locks.lock(repo)
 	defer unlock()
 	for range createTries {
 		repoItem, repoOK := db.Get(ctx, repoKey(repo))
 	return domain.ErrConflict
 }
-// numberLocks holds one mutex per repository for createNumbered.
+// numberLockStripes is the size of the table of numbering locks.
-var numberLocks sync.Map // domain.RepoName -> *sync.Mutex
+const numberLockStripes = 64
-// numberLock locks the numbering of repo and returns the unlock function.
+// numberLocks is a fixed table of mutexes, one stripe per hash of the
-func numberLock(repo domain.RepoName) func() {
+// repository name, for createNumbered: the memory is bounded whatever the
-	v, _ := numberLocks.LoadOrStore(repo, new(sync.Mutex))
+// number of repositories, and two repositories sharing a stripe only wait for
-	mu := v.(*sync.Mutex)
+// each other. One table is shared by every store of a Store, so the merge
+// requests and the issues of a repository are numbered one at a time.
+type numberLocks [numberLockStripes]sync.Mutex
+// lock locks the numbering of repo and returns the unlock function.
+func (l *numberLocks) lock(repo domain.RepoName) func() {
+	h := fnv.New32a()
+	_, _ = h.Write([]byte(repo.String()))
+	mu := &l[h.Sum32()%numberLockStripes]
 	mu.Lock()
 	return mu.Unlock
 }

M internal/adapters/kvstore/store.go +3 -2


@@ -48,7 +48,8 @@ var ErrSchemaTooNew = errors.New("storage schema is newer than this ermite suppo
48	48
49	49		// Store is the ermite-kv backed metadata store.
50	50		type Store struct {
51		-	db *kv.DB
	51	+	db *kv.DB
	52	+	locks *numberLocks // shared by the merge request and issue stores
52	53		}
53	54
54	55		var _ ports.RepoMetaStore = (*Store)(nil)

@@ -61,7 +62,7 @@ func Open(ctx context.Context, dir string) (*Store, error) {
61	62		if err != nil {
62	63		return nil, fmt.Errorf("open ermite-kv in %s: %w", dir, err)
63	64		}
64		-	s := &Store{db: db}
	65	+	s := &Store{db: db, locks: new(numberLocks)}
65	66		if err := s.checkSchema(ctx); err != nil {
66	67		_ = db.Close(context.WithoutCancel(ctx))
67	68		return nil, err

 // Store is the ermite-kv backed metadata store.
 type Store struct {
-	db *kv.DB
+	db    *kv.DB
+	locks *numberLocks // shared by the merge request and issue stores
 }
 var _ ports.RepoMetaStore = (*Store)(nil)
 	if err != nil {
 		return nil, fmt.Errorf("open ermite-kv in %s: %w", dir, err)
 	}
-	s := &Store{db: db}
+	s := &Store{db: db, locks: new(numberLocks)}
 	if err := s.checkSchema(ctx); err != nil {
 		_ = db.Close(context.WithoutCancel(ctx))
 		return nil, err