perf(kvstore): bound merge request history reads with a max-heap

commit: 35917d077b0598046928d23939dd04cbcff296cd
author: brice · 9h ago · 2026-10-22 18:36 UTC
committer: brice · 9h ago · 2026-10-22 18:41 UTC
parent: 9564f12
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +53 -5

M internal/adapters/kvstore/issue_links.go +24 -5


@@ -3,6 +3,7 @@
3	3		package kvstore
4	4
5	5		import (
	6	+	"container/heap"
6	7		"context"
7	8		"errors"
8	9		"fmt"

@@ -158,7 +159,7 @@ func toSet(numbers []int64) map[int64]struct{} {
158	159		// EventsOfMergeRequest returns history, including references no longer linked,
159	160		// ordered by time, then issue and sequence, capped at MaxIssueEvents.
160	161		func (l *IssueLinks) EventsOfMergeRequest(ctx context.Context, repo domain.RepoName, mr int64) ([]domain.IssueEvent, error) {
161		-	var out []domain.IssueEvent
	162	+	out := make(issueEventMaxHeap, 0, MaxIssueEvents)
162	163		prefix := mrEventsPrefix(repo, mr)
163	164		for key, item := range l.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
164	165		if err := ctx.Err(); err != nil {

@@ -176,11 +177,29 @@ func (l *IssueLinks) EventsOfMergeRequest(ctx context.Context, repo domain.RepoN
176	177		if err != nil {
177	178		return nil, err
178	179		}
179		-	out = append(out, ev)
	180	+	if len(out) < MaxIssueEvents {
	181	+	heap.Push(&out, ev)
	182	+	} else if domain.CompareIssueEvents(ev, out[0]) < 0 {
	183	+	out[0] = ev
	184	+	heap.Fix(&out, 0)
	185	+	}
180	186		}
181	187		slices.SortFunc(out, domain.CompareIssueEvents)
182		-	if len(out) > MaxIssueEvents {
183		-	out = out[:MaxIssueEvents]
184		-	}
185	188		return out, ctx.Err()
186	189		}
	190	+
	191	+	// issueEventMaxHeap keeps the latest retained event at the root.
	192	+	type issueEventMaxHeap []domain.IssueEvent
	193	+
	194	+	func (h issueEventMaxHeap) Len() int { return len(h) }
	195	+	func (h issueEventMaxHeap) Less(i, j int) bool { return domain.CompareIssueEvents(h[i], h[j]) > 0 }
	196	+	func (h issueEventMaxHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
	197	+	func (h issueEventMaxHeap) Push(x any) { h = append(*h, x.(domain.IssueEvent)) }
	198	+	func (h *issueEventMaxHeap) Pop() any {
	199	+	old := *h
	200	+	n := len(old) - 1
	201	+	x := old[n]
	202	+	old[n] = domain.IssueEvent{}
	203	+	*h = old[:n]
	204	+	return x
	205	+	}

 package kvstore
 import (
+	"container/heap"
 	"context"
 	"errors"
 	"fmt"
 // EventsOfMergeRequest returns history, including references no longer linked,
 // ordered by time, then issue and sequence, capped at MaxIssueEvents.
 func (l *IssueLinks) EventsOfMergeRequest(ctx context.Context, repo domain.RepoName, mr int64) ([]domain.IssueEvent, error) {
-	var out []domain.IssueEvent
+	out := make(issueEventMaxHeap, 0, MaxIssueEvents)
 	prefix := mrEventsPrefix(repo, mr)
 	for key, item := range l.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
 		if err := ctx.Err(); err != nil {
 		if err != nil {
 			return nil, err
 		}
-		out = append(out, ev)
+		if len(out) < MaxIssueEvents {
+			heap.Push(&out, ev)
+		} else if domain.CompareIssueEvents(ev, out[0]) < 0 {
+			out[0] = ev
+			heap.Fix(&out, 0)
+		}
 	}
 	slices.SortFunc(out, domain.CompareIssueEvents)
-	if len(out) > MaxIssueEvents {
-		out = out[:MaxIssueEvents]
-	}
 	return out, ctx.Err()
 }
+// issueEventMaxHeap keeps the latest retained event at the root.
+type issueEventMaxHeap []domain.IssueEvent
+func (h issueEventMaxHeap) Len() int           { return len(h) }
+func (h issueEventMaxHeap) Less(i, j int) bool { return domain.CompareIssueEvents(h[i], h[j]) > 0 }
+func (h issueEventMaxHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
+func (h *issueEventMaxHeap) Push(x any)        { *h = append(*h, x.(domain.IssueEvent)) }
+func (h *issueEventMaxHeap) Pop() any {
+	old := *h
+	n := len(old) - 1
+	x := old[n]
+	old[n] = domain.IssueEvent{}
+	*h = old[:n]
+	return x
+}

M internal/adapters/kvstore/issue_links_test.go +29 -0


@@ -2,12 +2,15 @@
2	2		package kvstore_test
3	3
4	4		import (
	5	+	"bnema.dev/ermite/internal/adapters/kvstore"
5	6		"bnema.dev/ermite/internal/domain"
6	7		"bnema.dev/ermite/internal/ports"
7	8		"context"
8	9		"github.com/stretchr/testify/assert"
9	10		"github.com/stretchr/testify/require"
	11	+	"slices"
10	12		"testing"
	13	+	"time"
11	14		)
12	15
13	16		func TestIssueLinksReplace(t *testing.T) {

@@ -65,3 +68,29 @@ func TestIssueClosedByRoundTrip(t *testing.T) {
65	68		require.NoError(t, err)
66	69		assert.Zero(t, stored.ClosedByMergeRequest)
67	70		}
	71	+
	72	+	func TestMRHistoryKeepsEarliestEvents(t *testing.T) {
	73	+	ctx := context.Background()
	74	+	s, _, repo := repoWithMRs(t)
	75	+	var want []domain.IssueEvent
	76	+	// Issue keys and insertion order both differ from timestamp order. Equal
	77	+	// timestamps across issues also exercise the issue/sequence tie breakers.
	78	+	for i := 0; i < 3; i++ {
	79	+	issue, err := s.Issues().Create(ctx, newIssue(t, repo, "history"))
	80	+	require.NoError(t, err)
	81	+	var events []domain.IssueEvent
	82	+	for j := 0; j < 240; j++ {
	83	+	events = append(events, domain.IssueEvent{Kind: domain.IssueEventReferenced, MergeRequest: 9, CreatedAt: time.Unix(int64(240-j), 0)})
	84	+	}
	85	+	require.NoError(t, s.Issues().Update(ctx, issue, events...))
	86	+	stored, err := s.Issues().ListEvents(ctx, repo, issue.Number)
	87	+	require.NoError(t, err)
	88	+	want = append(want, stored...)
	89	+	}
	90	+	slices.SortFunc(want, domain.CompareIssueEvents)
	91	+	require.Greater(t, len(want), kvstore.MaxIssueEvents)
	92	+	got, err := s.IssueLinks().EventsOfMergeRequest(ctx, repo, 9)
	93	+	require.NoError(t, err)
	94	+	assert.Equal(t, want[:kvstore.MaxIssueEvents], got)
	95	+	assert.Equal(t, kvstore.MaxIssueEvents, cap(got))
	96	+	}


@@ -2,12 +2,15 @@
2	package kvstore_test	2		package kvstore_test
3		3
4	import (	4		import (
		5	+	"bnema.dev/ermite/internal/adapters/kvstore"
5	"bnema.dev/ermite/internal/domain"	6		"bnema.dev/ermite/internal/domain"
6	"bnema.dev/ermite/internal/ports"	7		"bnema.dev/ermite/internal/ports"
7	"context"	8		"context"
8	"github.com/stretchr/testify/assert"	9		"github.com/stretchr/testify/assert"
9	"github.com/stretchr/testify/require"	10		"github.com/stretchr/testify/require"
		11	+	"slices"
10	"testing"	12		"testing"
		13	+	"time"
11	)	14		)
12		15
13	func TestIssueLinksReplace(t *testing.T) {	16		func TestIssueLinksReplace(t *testing.T) {

@@ -65,3 +68,29 @@ func TestIssueClosedByRoundTrip(t *testing.T) {
65	require.NoError(t, err)	68		require.NoError(t, err)
66	assert.Zero(t, stored.ClosedByMergeRequest)	69		assert.Zero(t, stored.ClosedByMergeRequest)
67	}	70		}
		71	+
		72	+	func TestMRHistoryKeepsEarliestEvents(t *testing.T) {
		73	+	ctx := context.Background()
		74	+	s, _, repo := repoWithMRs(t)
		75	+	var want []domain.IssueEvent
		76	+	// Issue keys and insertion order both differ from timestamp order. Equal
		77	+	// timestamps across issues also exercise the issue/sequence tie breakers.
		78	+	for i := 0; i < 3; i++ {
		79	+	issue, err := s.Issues().Create(ctx, newIssue(t, repo, "history"))
		80	+	require.NoError(t, err)
		81	+	var events []domain.IssueEvent
		82	+	for j := 0; j < 240; j++ {
		83	+	events = append(events, domain.IssueEvent{Kind: domain.IssueEventReferenced, MergeRequest: 9, CreatedAt: time.Unix(int64(240-j), 0)})
		84	+	}
		85	+	require.NoError(t, s.Issues().Update(ctx, issue, events...))
		86	+	stored, err := s.Issues().ListEvents(ctx, repo, issue.Number)
		87	+	require.NoError(t, err)
		88	+	want = append(want, stored...)
		89	+	}
		90	+	slices.SortFunc(want, domain.CompareIssueEvents)
		91	+	require.Greater(t, len(want), kvstore.MaxIssueEvents)
		92	+	got, err := s.IssueLinks().EventsOfMergeRequest(ctx, repo, 9)
		93	+	require.NoError(t, err)
		94	+	assert.Equal(t, want[:kvstore.MaxIssueEvents], got)
		95	+	assert.Equal(t, kvstore.MaxIssueEvents, cap(got))
		96	+	}