fix(kvstore): guard the repository key when updating an issue so a repo delete leaves no open-issue index keys
- commit
- 2794dd48b17ea72ce2cede01a87760233d3464b5
- author
- brice · · 2026-10-19 06:10 UTC
- parent
- 02692c4
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +36 -1
M internal/adapters/kvstore/issues.go +5 -1
| @@ -186,7 +186,8 @@ func (i *Issues) ListOpen(ctx context.Context, repo domain.RepoName) ([]domain.I | |||
| 186 | 186 | } | |
| 187 | 187 | ||
| 188 | 188 | // Update replaces the issue with a version precondition and moves its index key | |
| 189 | - | // in the same batch. | |
| 189 | + | // in the same batch. The repository must exist (see guardRepoKey), so that a | |
| 190 | + | // concurrent delete of the repository leaves no index key behind. | |
| 190 | 191 | // | |
| 191 | 192 | //hexcheck:ignore suspicious-business-logic-in-adapter port method name only | |
| 192 | 193 | func (i *Issues) Update(ctx context.Context, issue domain.Issue) error { | |
| @@ -202,6 +203,9 @@ func (i *Issues) Update(ctx context.Context, issue domain.Issue) error { | |||
| 202 | 203 | return fmt.Errorf("update issue %d of %q: read it first: %w", issue.Number, issue.Repo, domain.ErrConflict) | |
| 203 | 204 | } | |
| 204 | 205 | var b kv.Batch | |
| 206 | + | if err := guardRepoKey(ctx, i.db, &b, issue.Repo); err != nil { | |
| 207 | + | return err | |
| 208 | + | } | |
| 205 | 209 | b.RequireVersion(key, issue.Version) | |
| 206 | 210 | b.Put(key, val) | |
| 207 | 211 | indexOpenIssue(&b, issue) | |
| @@ -186,7 +186,8 @@ func (i *Issues) ListOpen(ctx context.Context, repo domain.RepoName) ([]domain.I | |||||
| 186 | } | 186 | } | ||
| 187 | 187 | ||||
| 188 | // Update replaces the issue with a version precondition and moves its index key | 188 | // Update replaces the issue with a version precondition and moves its index key | ||
| 189 | - | // in the same batch. | 189 | + | // in the same batch. The repository must exist (see guardRepoKey), so that a |
| 190 | + | // concurrent delete of the repository leaves no index key behind. | |||
| 190 | // | 191 | // | ||
| 191 | //hexcheck:ignore suspicious-business-logic-in-adapter port method name only | 192 | //hexcheck:ignore suspicious-business-logic-in-adapter port method name only | ||
| 192 | func (i *Issues) Update(ctx context.Context, issue domain.Issue) error { | 193 | func (i *Issues) Update(ctx context.Context, issue domain.Issue) error { | ||
| @@ -202,6 +203,9 @@ func (i *Issues) Update(ctx context.Context, issue domain.Issue) error { | |||||
| 202 | return fmt.Errorf("update issue %d of %q: read it first: %w", issue.Number, issue.Repo, domain.ErrConflict) | 203 | return fmt.Errorf("update issue %d of %q: read it first: %w", issue.Number, issue.Repo, domain.ErrConflict) | ||
| 203 | } | 204 | } | ||
| 204 | var b kv.Batch | 205 | var b kv.Batch | ||
| 206 | + | if err := guardRepoKey(ctx, i.db, &b, issue.Repo); err != nil { | |||
| 207 | + | return err | |||
| 208 | + | } | |||
| 205 | b.RequireVersion(key, issue.Version) | 209 | b.RequireVersion(key, issue.Version) | ||
| 206 | b.Put(key, val) | 210 | b.Put(key, val) | ||
| 207 | indexOpenIssue(&b, issue) | 211 | indexOpenIssue(&b, issue) | ||
M internal/adapters/kvstore/issues_test.go +31 -0
| @@ -337,3 +337,34 @@ func TestSchemaNewerThanThisBuildStopsIssuesData(t *testing.T) { | |||
| 337 | 337 | require.ErrorIs(t, err, kvstore.ErrSchemaTooNew) | |
| 338 | 338 | assert.Equal(t, uint32(4), uint32(kvstore.SchemaVersion)) | |
| 339 | 339 | } | |
| 340 | + | ||
| 341 | + | // A reopen racing the delete of the repository either lands before the delete | |
| 342 | + | // removes it or fails; it never leaves an isopen/ key of a repository that is | |
| 343 | + | // gone. | |
| 344 | + | func TestDeleteVersusReopenLeavesNoIssueKeys(t *testing.T) { | |
| 345 | + | ctx := context.Background() | |
| 346 | + | for range 30 { | |
| 347 | + | s, _, n := repoWithMRs(t) | |
| 348 | + | is := s.Issues() | |
| 349 | + | issue, err := is.Create(ctx, newIssue(t, n, "t")) | |
| 350 | + | require.NoError(t, err) | |
| 351 | + | issue.State = domain.IssueStateClosed | |
| 352 | + | require.NoError(t, is.Update(ctx, issue)) | |
| 353 | + | closed, err := is.Get(ctx, n, issue.Number) | |
| 354 | + | require.NoError(t, err) | |
| 355 | + | closed.State = domain.IssueStateOpen | |
| 356 | + | ||
| 357 | + | var wg sync.WaitGroup | |
| 358 | + | start := make(chan struct{}) | |
| 359 | + | wg.Add(2) | |
| 360 | + | go func() { defer wg.Done(); <-start; _ = is.Update(ctx, closed) }() | |
| 361 | + | go func() { defer wg.Done(); <-start; _ = s.Delete(ctx, n) }() | |
| 362 | + | close(start) | |
| 363 | + | wg.Wait() | |
| 364 | + | ||
| 365 | + | for _, prefix := range []string{"is/proj/", "isc/proj/", "isopen/proj/"} { | |
| 366 | + | assert.Zero(t, countKeys(t, s, prefix), prefix) | |
| 367 | + | } | |
| 368 | + | assert.ErrorIs(t, is.Update(ctx, closed), domain.ErrNotFound) | |
| 369 | + | } | |
| 370 | + | } | |
| @@ -337,3 +337,34 @@ func TestSchemaNewerThanThisBuildStopsIssuesData(t *testing.T) { | |||||
| 337 | require.ErrorIs(t, err, kvstore.ErrSchemaTooNew) | 337 | require.ErrorIs(t, err, kvstore.ErrSchemaTooNew) | ||
| 338 | assert.Equal(t, uint32(4), uint32(kvstore.SchemaVersion)) | 338 | assert.Equal(t, uint32(4), uint32(kvstore.SchemaVersion)) | ||
| 339 | } | 339 | } | ||
| 340 | + | ||||
| 341 | + | // A reopen racing the delete of the repository either lands before the delete | |||
| 342 | + | // removes it or fails; it never leaves an isopen/ key of a repository that is | |||
| 343 | + | // gone. | |||
| 344 | + | func TestDeleteVersusReopenLeavesNoIssueKeys(t *testing.T) { | |||
| 345 | + | ctx := context.Background() | |||
| 346 | + | for range 30 { | |||
| 347 | + | s, _, n := repoWithMRs(t) | |||
| 348 | + | is := s.Issues() | |||
| 349 | + | issue, err := is.Create(ctx, newIssue(t, n, "t")) | |||
| 350 | + | require.NoError(t, err) | |||
| 351 | + | issue.State = domain.IssueStateClosed | |||
| 352 | + | require.NoError(t, is.Update(ctx, issue)) | |||
| 353 | + | closed, err := is.Get(ctx, n, issue.Number) | |||
| 354 | + | require.NoError(t, err) | |||
| 355 | + | closed.State = domain.IssueStateOpen | |||
| 356 | + | ||||
| 357 | + | var wg sync.WaitGroup | |||
| 358 | + | start := make(chan struct{}) | |||
| 359 | + | wg.Add(2) | |||
| 360 | + | go func() { defer wg.Done(); <-start; _ = is.Update(ctx, closed) }() | |||
| 361 | + | go func() { defer wg.Done(); <-start; _ = s.Delete(ctx, n) }() | |||
| 362 | + | close(start) | |||
| 363 | + | wg.Wait() | |||
| 364 | + | ||||
| 365 | + | for _, prefix := range []string{"is/proj/", "isc/proj/", "isopen/proj/"} { | |||
| 366 | + | assert.Zero(t, countKeys(t, s, prefix), prefix) | |||
| 367 | + | } | |||
| 368 | + | assert.ErrorIs(t, is.Update(ctx, closed), domain.ErrNotFound) | |||
| 369 | + | } | |||
| 370 | + | } | |||