1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore_test
4
5import (
6 "context"
7 "testing"
8
9 "github.com/stretchr/testify/assert"
10 "github.com/stretchr/testify/require"
11
12 storagepb "bnema.dev/ermite/gen/ermite/storage"
13 "bnema.dev/ermite/internal/adapters/kvstore"
14 "bnema.dev/ermite/internal/domain"
15)
16
17func bySource(t *testing.T, mrs *kvstore.MergeRequests, repo domain.RepoName, branches ...string) []int64 {
18 t.Helper()
19 got, truncated, err := mrs.OpenBySource(context.Background(), repo, branches, 0)
20 require.NoError(t, err)
21 require.False(t, truncated, "no limit: never truncated")
22 return got
23}
24
25func byTarget(t *testing.T, mrs *kvstore.MergeRequests, repo domain.RepoName, branches ...string) []int64 {
26 t.Helper()
27 got, truncated, err := mrs.OpenByTarget(context.Background(), repo, branches, 0)
28 require.NoError(t, err)
29 require.False(t, truncated)
30 return got
31}
32
33func mrFromTo(t *testing.T, repo domain.RepoName, source, target string) domain.MergeRequest {
34 mr := mrOn(t, repo, source)
35 mr.Target = target
36 return mr
37}
38
39func openNumbers(t *testing.T, mrs *kvstore.MergeRequests, repo domain.RepoName) []int64 {
40 t.Helper()
41 list, err := mrs.ListOpen(context.Background(), repo)
42 require.NoError(t, err)
43 var out []int64
44 for _, mr := range list {
45 out = append(out, mr.Number)
46 }
47 return out
48}
49
50func mrOn(t *testing.T, repo domain.RepoName, source string) domain.MergeRequest {
51 mr := newMR(t, repo, "t")
52 mr.Source = source
53 return mr
54}
55
56// The index follows every write of a merge request: create, comment, source
57// change, close and merge.
58func TestOpenMergeRequestIndexFollowsTheWrites(t *testing.T) {
59 ctx := context.Background()
60 s, mrs, n := repoWithMRs(t)
61
62 a, err := mrs.Create(ctx, mrOn(t, n, "feat/a"))
63 require.NoError(t, err)
64 b, err := mrs.Create(ctx, mrOn(t, n, "feat/b"))
65 require.NoError(t, err)
66 c, err := mrs.Create(ctx, mrOn(t, n, "feat/a")) // a second open request on the same source
67 require.NoError(t, err)
68 assert.Equal(t, []int64{a.Number, c.Number}, bySource(t, mrs, n, "feat/a"))
69 assert.Equal(t, []int64{b.Number}, bySource(t, mrs, n, "feat/b"))
70 assert.Empty(t, bySource(t, mrs, n, "feat"), "a prefix of a branch is not the branch")
71 assert.Empty(t, bySource(t, mrs, n, "feat/a/x"))
72 assert.Equal(t, []int64{c.Number, b.Number, a.Number}, openNumbers(t, mrs, n), "highest number first")
73
74 // A comment rewrites the request: the index stays.
75 text, err := domain.ParsePlainText("hi")
76 require.NoError(t, err)
77 cm := domain.Comment{ID: 1, Author: "x", Text: text, CreatedAt: created}
78 a.CommentSeq = 1
79 require.NoError(t, mrs.SaveComment(ctx, a, cm))
80 assert.Equal(t, []int64{a.Number, c.Number}, bySource(t, mrs, n, "feat/a"))
81
82 // The source changes (a rewrite of the request): the key moves.
83 b, err = mrs.Get(ctx, n, b.Number)
84 require.NoError(t, err)
85 b.Source = "feat/moved"
86 require.NoError(t, mrs.Update(ctx, b))
87 assert.Empty(t, bySource(t, mrs, n, "feat/b"))
88 assert.Equal(t, []int64{b.Number}, bySource(t, mrs, n, "feat/moved"))
89
90 // Close and merge remove the key.
91 c, err = mrs.Get(ctx, n, c.Number)
92 require.NoError(t, err)
93 c.State = domain.MRStateClosed
94 require.NoError(t, mrs.Update(ctx, c))
95 assert.Equal(t, []int64{a.Number}, bySource(t, mrs, n, "feat/a"))
96 a, err = mrs.Get(ctx, n, a.Number)
97 require.NoError(t, err)
98 a.State, a.MergeMode = domain.MRStateMerged, domain.MergeModeFastForward
99 require.NoError(t, mrs.Update(ctx, a))
100 assert.Empty(t, bySource(t, mrs, n, "feat/a"))
101 assert.Equal(t, []int64{b.Number}, openNumbers(t, mrs, n))
102
103 // Several branches in one lookup, each scanned on its own.
104 assert.Equal(t, []int64{b.Number}, bySource(t, mrs, n, "feat/a", "feat/moved", "nothing"))
105
106 // Another repository has its own.
107 other := name(t, "other")
108 require.NoError(t, s.Create(ctx, domain.Repo{Name: other, CreatedAt: created, Visibility: domain.VisibilityPublic}))
109 _, err = mrs.Create(ctx, mrOn(t, other, "feat/moved"))
110 require.NoError(t, err)
111 assert.Equal(t, []int64{b.Number}, bySource(t, mrs, n, "feat/moved"))
112}
113
114// The target index follows the same writes as the source one: create, comment,
115// target change, close and merge.
116func TestOpenMergeRequestTargetIndexFollowsTheWrites(t *testing.T) {
117 ctx := context.Background()
118 s, mrs, n := repoWithMRs(t)
119
120 a, err := mrs.Create(ctx, mrFromTo(t, n, "feat/a", "main"))
121 require.NoError(t, err)
122 b, err := mrs.Create(ctx, mrFromTo(t, n, "feat/b", "release/1"))
123 require.NoError(t, err)
124 c, err := mrs.Create(ctx, mrFromTo(t, n, "feat/c", "main"))
125 require.NoError(t, err)
126 assert.Equal(t, []int64{a.Number, c.Number}, byTarget(t, mrs, n, "main"))
127 assert.Equal(t, []int64{b.Number}, byTarget(t, mrs, n, "release/1"))
128 assert.Empty(t, byTarget(t, mrs, n, "release"), "a prefix of a branch is not the branch")
129 assert.Empty(t, byTarget(t, mrs, n, "feat/a"), "a source is not a target")
130 assert.Equal(t, []int64{a.Number}, bySource(t, mrs, n, "feat/a"), "and the other way round")
131
132 text, err := domain.ParsePlainText("hi")
133 require.NoError(t, err)
134 a.CommentSeq = 1
135 require.NoError(t, mrs.SaveComment(ctx, a, domain.Comment{ID: 1, Author: "x", Text: text, CreatedAt: created}))
136 assert.Equal(t, []int64{a.Number, c.Number}, byTarget(t, mrs, n, "main"), "a comment keeps the index")
137
138 b, err = mrs.Get(ctx, n, b.Number)
139 require.NoError(t, err)
140 b.Target = "main"
141 require.NoError(t, mrs.Update(ctx, b))
142 assert.Empty(t, byTarget(t, mrs, n, "release/1"), "a changed target moves the key")
143 assert.Equal(t, []int64{a.Number, b.Number, c.Number}, byTarget(t, mrs, n, "main"))
144
145 c, err = mrs.Get(ctx, n, c.Number)
146 require.NoError(t, err)
147 c.State = domain.MRStateClosed
148 require.NoError(t, mrs.Update(ctx, c))
149 a, err = mrs.Get(ctx, n, a.Number)
150 require.NoError(t, err)
151 a.State, a.MergeMode = domain.MRStateMerged, domain.MergeModeFastForward
152 require.NoError(t, mrs.Update(ctx, a))
153 assert.Equal(t, []int64{b.Number}, byTarget(t, mrs, n, "main"), "close and merge remove the key")
154
155 // The keys of a deleted repository go with it, both indexes.
156 require.NoError(t, s.Delete(ctx, n))
157 assert.Empty(t, byTarget(t, mrs, n, "main"))
158 assert.Empty(t, bySource(t, mrs, n, "feat/b"))
159}
160
161// The lookup stops at limit+1 keys, whatever the number of open requests on the
162// branch, and says it was cut without counting the rest.
163func TestOpenLookupStopsAtTheLimit(t *testing.T) {
164 ctx := context.Background()
165 _, mrs, n := repoWithMRs(t)
166 for range 40 {
167 _, err := mrs.Create(ctx, mrFromTo(t, n, "busy", "main"))
168 require.NoError(t, err)
169 }
170 _, err := mrs.Create(ctx, mrFromTo(t, n, "calm", "main"))
171 require.NoError(t, err)
172
173 got, truncated, err := mrs.OpenBySource(ctx, n, []string{"busy"}, 5)
174 require.NoError(t, err)
175 assert.True(t, truncated)
176 assert.Equal(t, []int64{1, 2, 3, 4, 5}, got, "the first limit numbers")
177
178 got, truncated, err = mrs.OpenByTarget(ctx, n, []string{"main"}, 7)
179 require.NoError(t, err)
180 assert.True(t, truncated)
181 assert.Len(t, got, 7)
182
183 // Exactly limit keys is not truncated: only a limit+1th key is.
184 got, truncated, err = mrs.OpenBySource(ctx, n, []string{"calm"}, 1)
185 require.NoError(t, err)
186 assert.False(t, truncated)
187 assert.Len(t, got, 1)
188 got, truncated, err = mrs.OpenBySource(ctx, n, []string{"busy"}, 40)
189 require.NoError(t, err)
190 assert.False(t, truncated)
191 assert.Len(t, got, 40)
192
193 // The limit spans the branches of one lookup.
194 got, truncated, err = mrs.OpenBySource(ctx, n, []string{"calm", "busy"}, 3)
195 require.NoError(t, err)
196 assert.True(t, truncated)
197 assert.Len(t, got, 3)
198}
199
200// A cancelled context stops the lookup: nothing is scanned, and the caller
201// sees the error of the context and that the result is partial.
202func TestOpenLookupStopsOnCancel(t *testing.T) {
203 _, mrs, n := repoWithMRs(t)
204 for range 10 {
205 _, err := mrs.Create(context.Background(), mrFromTo(t, n, "busy", "main"))
206 require.NoError(t, err)
207 }
208 ctx, cancel := context.WithCancel(context.Background())
209 cancel()
210 got, truncated, err := mrs.OpenBySource(ctx, n, []string{"busy"}, 0)
211 assert.ErrorIs(t, err, context.Canceled)
212 assert.True(t, truncated)
213 assert.Empty(t, got)
214 _, truncated, err = mrs.OpenByTarget(ctx, n, []string{"main"}, 5)
215 assert.ErrorIs(t, err, context.Canceled)
216 assert.True(t, truncated)
217}
218
219// A branch name with the characters of a key separator stays in its own prefix.
220func TestOpenMergeRequestIndexEscapesBranchNames(t *testing.T) {
221 ctx := context.Background()
222 _, mrs, n := repoWithMRs(t)
223 x, err := mrs.Create(ctx, mrOn(t, n, "a/b"))
224 require.NoError(t, err)
225 y, err := mrs.Create(ctx, mrOn(t, n, "a%2Fb"))
226 require.NoError(t, err)
227 assert.Equal(t, []int64{x.Number}, bySource(t, mrs, n, "a/b"))
228 assert.Equal(t, []int64{y.Number}, bySource(t, mrs, n, "a%2Fb"))
229}
230
231// Many historical merge requests are not read by a lookup of a push: the index
232// holds the keys of the open ones, and only those are fetched.
233func TestOpenBySourceDoesNotReadHistoricalMergeRequests(t *testing.T) {
234 ctx := context.Background()
235 _, mrs, n := repoWithMRs(t)
236 for range 200 {
237 mr, err := mrs.Create(ctx, mrOn(t, n, "old/closed"))
238 require.NoError(t, err)
239 mr.State = domain.MRStateClosed
240 require.NoError(t, mrs.Update(ctx, mr))
241 }
242 open, err := mrs.Create(ctx, mrOn(t, n, "feat/live"))
243 require.NoError(t, err)
244 assert.Empty(t, bySource(t, mrs, n, "old/closed"), "none of the 200 closed ones is indexed")
245 assert.Equal(t, []int64{open.Number}, bySource(t, mrs, n, "feat/live"))
246 assert.Equal(t, []int64{open.Number}, openNumbers(t, mrs, n))
247}
248
249// Data that predates the index (a schema 2 database, or one an older build wrote
250// to) is indexed once at open, and the index is not built twice.
251func TestOpenIndexIsBackfilledOnce(t *testing.T) {
252 ctx := context.Background()
253 dir := t.TempDir()
254 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: 2}))
255 putRaw(t, dir, "repo/old", mustProto(t, &storagepb.Repo{Name: "old", Visibility: storagepb.Visibility_VISIBILITY_PUBLIC}))
256 mrRaw := func(number int64, source string, state storagepb.MergeRequestState) []byte {
257 return mustProto(t, &storagepb.MergeRequest{
258 Number: number, Title: "t", SourceBranch: source, TargetBranch: "main", State: state, Author: "a",
259 })
260 }
261 putRaw(t, dir, "mr/old/0000000001", mrRaw(1, "feat/one", storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN))
262 putRaw(t, dir, "mr/old/0000000002", mrRaw(2, "feat/one", storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED))
263 putRaw(t, dir, "mr/old/0000000003", mrRaw(3, "feat/two", storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED))
264 putRaw(t, dir, "mr/old/0000000004", mrRaw(4, "feat/two", storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN))
265 putRaw(t, dir, "mrseq/old", mustProto(t, &storagepb.Sequence{Last: 4}))
266
267 s := openStore(t, dir)
268 mrs := s.MergeRequests()
269 old := name(t, "old")
270 assert.Equal(t, []int64{1}, bySource(t, mrs, old, "feat/one"))
271 assert.Equal(t, []int64{4}, bySource(t, mrs, old, "feat/two"))
272 assert.Equal(t, []int64{4, 1}, openNumbers(t, mrs, old))
273 assert.Equal(t, []int64{1, 4}, byTarget(t, mrs, old, "main"), "the target index is built by the same pass")
274 require.NoError(t, s.Close(ctx))
275
276 // Opened again: the marker is there, nothing is built, and the index kept
277 // following the writes in between.
278 s2, err := kvstore.Open(ctx, dir)
279 require.NoError(t, err)
280 t.Cleanup(func() { _ = s2.Close(ctx) })
281 m2 := s2.MergeRequests()
282 five, err := m2.Create(ctx, mrOn(t, old, "feat/one"))
283 require.NoError(t, err)
284 assert.Equal(t, int64(5), five.Number)
285 assert.Equal(t, []int64{1, 5}, bySource(t, m2, old, "feat/one"))
286}