perf(kvstore): page issues without decoding the whole history
- commit
- 028460980c51e14a0de1f404d66d221ca074d5e6
- author
- brice · · 2026-10-19 08:52 UTC
- parent
- 17b5f65
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +84 -10
M internal/adapters/kvstore/issues.go +35 -10
| @@ -129,21 +129,46 @@ func (i *Issues) List(ctx context.Context, repo domain.RepoName, params ports.Li | |||
| 129 | 129 | } | |
| 130 | 130 | before = n | |
| 131 | 131 | } | |
| 132 | - | var all []domain.Issue | |
| 133 | - | for key, item := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueRepoPrefix(repo))) { | |
| 134 | - | issue, err := decodeIssue(repo, item) | |
| 135 | - | if err != nil { | |
| 136 | - | return ports.IssuePage{}, fmt.Errorf("decode %s: %w", key, err) | |
| 132 | + | // The keys sort by number: walk [first, cursor) ascending, keep the last | |
| 133 | + | // PageSize+1 matching records undecoded (the extra one tells whether | |
| 134 | + | // another page follows), then decode only those. | |
| 135 | + | start := []byte(issueRepoPrefix(repo)) | |
| 136 | + | end := append(slices.Clone(start[:len(start)-1]), '/'+1) // the prefix ends in '/' | |
| 137 | + | if before > 0 { | |
| 138 | + | end = issueKey(repo, before) | |
| 139 | + | } | |
| 140 | + | keep := params.PageSize + 1 | |
| 141 | + | var window []kv.Item | |
| 142 | + | for _, item := range i.db.Snapshot(ctx).Range(ctx, start, end) { | |
| 143 | + | if params.State != "" { | |
| 144 | + | var pb storagepb.Issue | |
| 145 | + | if err := proto.Unmarshal(item.Value, &pb); err != nil { | |
| 146 | + | return ports.IssuePage{}, fmt.Errorf("decode issue: %w", err) | |
| 147 | + | } | |
| 148 | + | st, err := decodeIssueState(pb.GetState()) | |
| 149 | + | if err != nil { | |
| 150 | + | return ports.IssuePage{}, fmt.Errorf("decode issue of %q: %w", repo, err) | |
| 151 | + | } | |
| 152 | + | if st != params.State { | |
| 153 | + | continue | |
| 154 | + | } | |
| 137 | 155 | } | |
| 138 | - | if params.State != "" && issue.State != params.State { | |
| 139 | - | continue | |
| 156 | + | window = append(window, item) | |
| 157 | + | if params.PageSize > 0 && len(window) >= 2*keep { | |
| 158 | + | window = append(window[:0], window[len(window)-keep:]...) | |
| 140 | 159 | } | |
| 141 | - | if before > 0 && issue.Number >= before { | |
| 142 | - | continue | |
| 160 | + | } | |
| 161 | + | if params.PageSize > 0 && len(window) > keep { | |
| 162 | + | window = window[len(window)-keep:] | |
| 163 | + | } | |
| 164 | + | all := make([]domain.Issue, 0, len(window)) | |
| 165 | + | for _, item := range slices.Backward(window) { | |
| 166 | + | issue, err := decodeIssue(repo, item) | |
| 167 | + | if err != nil { | |
| 168 | + | return ports.IssuePage{}, err | |
| 143 | 169 | } | |
| 144 | 170 | all = append(all, issue) | |
| 145 | 171 | } | |
| 146 | - | slices.Reverse(all) | |
| 147 | 172 | var page ports.IssuePage | |
| 148 | 173 | if params.PageSize > 0 && len(all) > params.PageSize { | |
| 149 | 174 | all = all[:params.PageSize] | |
| @@ -129,21 +129,46 @@ func (i *Issues) List(ctx context.Context, repo domain.RepoName, params ports.Li | |||||
| 129 | } | 129 | } | ||
| 130 | before = n | 130 | before = n | ||
| 131 | } | 131 | } | ||
| 132 | - | var all []domain.Issue | 132 | + | // The keys sort by number: walk [first, cursor) ascending, keep the last |
| 133 | - | for key, item := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueRepoPrefix(repo))) { | 133 | + | // PageSize+1 matching records undecoded (the extra one tells whether |
| 134 | - | issue, err := decodeIssue(repo, item) | 134 | + | // another page follows), then decode only those. |
| 135 | - | if err != nil { | 135 | + | start := []byte(issueRepoPrefix(repo)) |
| 136 | - | return ports.IssuePage{}, fmt.Errorf("decode %s: %w", key, err) | 136 | + | end := append(slices.Clone(start[:len(start)-1]), '/'+1) // the prefix ends in '/' |
| 137 | + | if before > 0 { | |||
| 138 | + | end = issueKey(repo, before) | |||
| 139 | + | } | |||
| 140 | + | keep := params.PageSize + 1 | |||
| 141 | + | var window []kv.Item | |||
| 142 | + | for _, item := range i.db.Snapshot(ctx).Range(ctx, start, end) { | |||
| 143 | + | if params.State != "" { | |||
| 144 | + | var pb storagepb.Issue | |||
| 145 | + | if err := proto.Unmarshal(item.Value, &pb); err != nil { | |||
| 146 | + | return ports.IssuePage{}, fmt.Errorf("decode issue: %w", err) | |||
| 147 | + | } | |||
| 148 | + | st, err := decodeIssueState(pb.GetState()) | |||
| 149 | + | if err != nil { | |||
| 150 | + | return ports.IssuePage{}, fmt.Errorf("decode issue of %q: %w", repo, err) | |||
| 151 | + | } | |||
| 152 | + | if st != params.State { | |||
| 153 | + | continue | |||
| 154 | + | } | |||
| 137 | } | 155 | } | ||
| 138 | - | if params.State != "" && issue.State != params.State { | 156 | + | window = append(window, item) |
| 139 | - | continue | 157 | + | if params.PageSize > 0 && len(window) >= 2*keep { |
| 158 | + | window = append(window[:0], window[len(window)-keep:]...) | |||
| 140 | } | 159 | } | ||
| 141 | - | if before > 0 && issue.Number >= before { | 160 | + | } |
| 142 | - | continue | 161 | + | if params.PageSize > 0 && len(window) > keep { |
| 162 | + | window = window[len(window)-keep:] | |||
| 163 | + | } | |||
| 164 | + | all := make([]domain.Issue, 0, len(window)) | |||
| 165 | + | for _, item := range slices.Backward(window) { | |||
| 166 | + | issue, err := decodeIssue(repo, item) | |||
| 167 | + | if err != nil { | |||
| 168 | + | return ports.IssuePage{}, err | |||
| 143 | } | 169 | } | ||
| 144 | all = append(all, issue) | 170 | all = append(all, issue) | ||
| 145 | } | 171 | } | ||
| 146 | - | slices.Reverse(all) | |||
| 147 | var page ports.IssuePage | 172 | var page ports.IssuePage | ||
| 148 | if params.PageSize > 0 && len(all) > params.PageSize { | 173 | if params.PageSize > 0 && len(all) > params.PageSize { | ||
| 149 | all = all[:params.PageSize] | 174 | all = all[:params.PageSize] | ||
M internal/adapters/kvstore/issues_test.go +49 -0
| @@ -453,3 +453,52 @@ func TestListOpenAllSkipsAStaleIndexEntry(t *testing.T) { | |||
| 453 | 453 | require.Len(t, open, 1) | |
| 454 | 454 | assert.Equal(t, keepM.Number, open[0].Number) | |
| 455 | 455 | } | |
| 456 | + | ||
| 457 | + | func TestIssueListPagesThroughManyIssues(t *testing.T) { | |
| 458 | + | ctx := context.Background() | |
| 459 | + | s, _, n := repoWithMRs(t) | |
| 460 | + | is := s.Issues() | |
| 461 | + | const total = 53 | |
| 462 | + | for i := 1; i <= total; i++ { | |
| 463 | + | _, err := is.Create(ctx, newIssue(t, n, "issue "+strconv.Itoa(i))) | |
| 464 | + | require.NoError(t, err) | |
| 465 | + | } | |
| 466 | + | for i := 2; i <= total; i += 3 { // 2, 5, 8, ... | |
| 467 | + | is, err := is.Get(ctx, n, int64(i)) | |
| 468 | + | require.NoError(t, err) | |
| 469 | + | is.State = domain.IssueStateClosed | |
| 470 | + | require.NoError(t, s.Issues().Update(ctx, is)) | |
| 471 | + | } | |
| 472 | + | ||
| 473 | + | walk := func(state domain.IssueState, size int) []int64 { | |
| 474 | + | var got []int64 | |
| 475 | + | token := "" | |
| 476 | + | for { | |
| 477 | + | page, err := s.Issues().List(ctx, n, ports.ListIssuesParams{State: state, PageSize: size, PageToken: token}) | |
| 478 | + | require.NoError(t, err) | |
| 479 | + | require.LessOrEqual(t, len(page.Issues), size) | |
| 480 | + | for _, is := range page.Issues { | |
| 481 | + | got = append(got, is.Number) | |
| 482 | + | } | |
| 483 | + | if page.NextPageToken == "" { | |
| 484 | + | return got | |
| 485 | + | } | |
| 486 | + | token = page.NextPageToken | |
| 487 | + | } | |
| 488 | + | } | |
| 489 | + | for _, size := range []int{1, 7, 10, 53, 100} { | |
| 490 | + | var all, closed []int64 | |
| 491 | + | for i := int64(total); i >= 1; i-- { | |
| 492 | + | all = append(all, i) | |
| 493 | + | if i%3 == 2 { | |
| 494 | + | closed = append(closed, i) | |
| 495 | + | } | |
| 496 | + | } | |
| 497 | + | assert.Equal(t, all, walk("", size), "size %d", size) | |
| 498 | + | assert.Equal(t, closed, walk(domain.IssueStateClosed, size), "closed, size %d", size) | |
| 499 | + | } | |
| 500 | + | page, err := s.Issues().List(ctx, n, ports.ListIssuesParams{PageSize: 10, PageToken: "1"}) | |
| 501 | + | require.NoError(t, err) | |
| 502 | + | assert.Empty(t, page.Issues) | |
| 503 | + | assert.Empty(t, page.NextPageToken) | |
| 504 | + | } | |
| @@ -453,3 +453,52 @@ func TestListOpenAllSkipsAStaleIndexEntry(t *testing.T) { | |||||
| 453 | require.Len(t, open, 1) | 453 | require.Len(t, open, 1) | ||
| 454 | assert.Equal(t, keepM.Number, open[0].Number) | 454 | assert.Equal(t, keepM.Number, open[0].Number) | ||
| 455 | } | 455 | } | ||
| 456 | + | ||||
| 457 | + | func TestIssueListPagesThroughManyIssues(t *testing.T) { | |||
| 458 | + | ctx := context.Background() | |||
| 459 | + | s, _, n := repoWithMRs(t) | |||
| 460 | + | is := s.Issues() | |||
| 461 | + | const total = 53 | |||
| 462 | + | for i := 1; i <= total; i++ { | |||
| 463 | + | _, err := is.Create(ctx, newIssue(t, n, "issue "+strconv.Itoa(i))) | |||
| 464 | + | require.NoError(t, err) | |||
| 465 | + | } | |||
| 466 | + | for i := 2; i <= total; i += 3 { // 2, 5, 8, ... | |||
| 467 | + | is, err := is.Get(ctx, n, int64(i)) | |||
| 468 | + | require.NoError(t, err) | |||
| 469 | + | is.State = domain.IssueStateClosed | |||
| 470 | + | require.NoError(t, s.Issues().Update(ctx, is)) | |||
| 471 | + | } | |||
| 472 | + | ||||
| 473 | + | walk := func(state domain.IssueState, size int) []int64 { | |||
| 474 | + | var got []int64 | |||
| 475 | + | token := "" | |||
| 476 | + | for { | |||
| 477 | + | page, err := s.Issues().List(ctx, n, ports.ListIssuesParams{State: state, PageSize: size, PageToken: token}) | |||
| 478 | + | require.NoError(t, err) | |||
| 479 | + | require.LessOrEqual(t, len(page.Issues), size) | |||
| 480 | + | for _, is := range page.Issues { | |||
| 481 | + | got = append(got, is.Number) | |||
| 482 | + | } | |||
| 483 | + | if page.NextPageToken == "" { | |||
| 484 | + | return got | |||
| 485 | + | } | |||
| 486 | + | token = page.NextPageToken | |||
| 487 | + | } | |||
| 488 | + | } | |||
| 489 | + | for _, size := range []int{1, 7, 10, 53, 100} { | |||
| 490 | + | var all, closed []int64 | |||
| 491 | + | for i := int64(total); i >= 1; i-- { | |||
| 492 | + | all = append(all, i) | |||
| 493 | + | if i%3 == 2 { | |||
| 494 | + | closed = append(closed, i) | |||
| 495 | + | } | |||
| 496 | + | } | |||
| 497 | + | assert.Equal(t, all, walk("", size), "size %d", size) | |||
| 498 | + | assert.Equal(t, closed, walk(domain.IssueStateClosed, size), "closed, size %d", size) | |||
| 499 | + | } | |||
| 500 | + | page, err := s.Issues().List(ctx, n, ports.ListIssuesParams{PageSize: 10, PageToken: "1"}) | |||
| 501 | + | require.NoError(t, err) | |||
| 502 | + | assert.Empty(t, page.Issues) | |||
| 503 | + | assert.Empty(t, page.NextPageToken) | |||
| 504 | + | } | |||