fix(kvstore): serialize numbering per repository so concurrent creates never exhaust retries
- commit
- 68460c13938ad84411cfbc24cb85e81abd579fd9
- author
- brice · · 2026-10-19 06:17 UTC
- parent
- b2212f2
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
1 file · +18 -0
M internal/adapters/kvstore/mergerequests.go +18 -0
| @@ -10,6 +10,7 @@ import ( | |||
| 10 | 10 | "slices" | |
| 11 | 11 | "strconv" | |
| 12 | 12 | "strings" | |
| 13 | + | "sync" | |
| 13 | 14 | "unicode/utf8" | |
| 14 | 15 | ||
| 15 | 16 | kv "bnema.dev/ermite-kv" | |
| @@ -222,7 +223,13 @@ func (m *MergeRequests) Create(ctx context.Context, mr domain.MergeRequest) (dom | |||
| 222 | 223 | // guardRepoKey): two concurrent creations, of either kind, never share a | |
| 223 | 224 | // number. It retries when the batch conflicts; a missing repository gives | |
| 224 | 225 | // domain.ErrNotFound. | |
| 226 | + | // | |
| 227 | + | // Creations in one repository are serialized in the process (numberLock), so | |
| 228 | + | // a burst of merge requests and issues never exhausts the retries; the version | |
| 229 | + | // checks still guard against any other writer. | |
| 225 | 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 | + | defer unlock() | |
| 226 | 233 | for range createTries { | |
| 227 | 234 | repoItem, repoOK := db.Get(ctx, repoKey(repo)) | |
| 228 | 235 | if !repoOK { | |
| @@ -259,6 +266,17 @@ func createNumbered(ctx context.Context, db *kv.DB, repo domain.RepoName, add fu | |||
| 259 | 266 | return domain.ErrConflict | |
| 260 | 267 | } | |
| 261 | 268 | ||
| 269 | + | // numberLocks holds one mutex per repository for createNumbered. | |
| 270 | + | var numberLocks sync.Map // domain.RepoName -> *sync.Mutex | |
| 271 | + | ||
| 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) | |
| 276 | + | mu.Lock() | |
| 277 | + | return mu.Unlock | |
| 278 | + | } | |
| 279 | + | ||
| 262 | 280 | // Get reads mr/<repo>/<number>. | |
| 263 | 281 | func (m *MergeRequests) Get(ctx context.Context, repo domain.RepoName, number int64) (domain.MergeRequest, error) { | |
| 264 | 282 | if number < 1 { | |
| @@ -10,6 +10,7 @@ import ( | |||||
| 10 | "slices" | 10 | "slices" | ||
| 11 | "strconv" | 11 | "strconv" | ||
| 12 | "strings" | 12 | "strings" | ||
| 13 | + | "sync" | |||
| 13 | "unicode/utf8" | 14 | "unicode/utf8" | ||
| 14 | 15 | ||||
| 15 | kv "bnema.dev/ermite-kv" | 16 | kv "bnema.dev/ermite-kv" | ||
| @@ -222,7 +223,13 @@ func (m *MergeRequests) Create(ctx context.Context, mr domain.MergeRequest) (dom | |||||
| 222 | // guardRepoKey): two concurrent creations, of either kind, never share a | 223 | // guardRepoKey): two concurrent creations, of either kind, never share a | ||
| 223 | // number. It retries when the batch conflicts; a missing repository gives | 224 | // number. It retries when the batch conflicts; a missing repository gives | ||
| 224 | // domain.ErrNotFound. | 225 | // domain.ErrNotFound. | ||
| 226 | + | // | |||
| 227 | + | // Creations in one repository are serialized in the process (numberLock), so | |||
| 228 | + | // a burst of merge requests and issues never exhausts the retries; the version | |||
| 229 | + | // checks still guard against any other writer. | |||
| 225 | func createNumbered(ctx context.Context, db *kv.DB, repo domain.RepoName, add func(b *kv.Batch, number int64) error) error { | 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 | + | defer unlock() | |||
| 226 | for range createTries { | 233 | for range createTries { | ||
| 227 | repoItem, repoOK := db.Get(ctx, repoKey(repo)) | 234 | repoItem, repoOK := db.Get(ctx, repoKey(repo)) | ||
| 228 | if !repoOK { | 235 | if !repoOK { | ||
| @@ -259,6 +266,17 @@ func createNumbered(ctx context.Context, db *kv.DB, repo domain.RepoName, add fu | |||||
| 259 | return domain.ErrConflict | 266 | return domain.ErrConflict | ||
| 260 | } | 267 | } | ||
| 261 | 268 | ||||
| 269 | + | // numberLocks holds one mutex per repository for createNumbered. | |||
| 270 | + | var numberLocks sync.Map // domain.RepoName -> *sync.Mutex | |||
| 271 | + | ||||
| 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) | |||
| 276 | + | mu.Lock() | |||
| 277 | + | return mu.Unlock | |||
| 278 | + | } | |||
| 279 | + | ||||
| 262 | // Get reads mr/<repo>/<number>. | 280 | // Get reads mr/<repo>/<number>. | ||
| 263 | func (m *MergeRequests) Get(ctx context.Context, repo domain.RepoName, number int64) (domain.MergeRequest, error) { | 281 | func (m *MergeRequests) Get(ctx context.Context, repo domain.RepoName, number int64) (domain.MergeRequest, error) { | ||
| 264 | if number < 1 { | 282 | if number < 1 { | ||