1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore_test
4
5import (
6 "context"
7 "sync"
8 "testing"
9
10 "github.com/stretchr/testify/assert"
11 "github.com/stretchr/testify/require"
12 "google.golang.org/protobuf/proto"
13
14 storagepb "bnema.dev/ermite/gen/ermite/storage"
15 "bnema.dev/ermite/internal/adapters/kvstore"
16 "bnema.dev/ermite/internal/domain"
17)
18
19func entry(run int64) *domain.CIConcurrencyEntry {
20 return &domain.CIConcurrencyEntry{RunID: run, Attempt: 1}
21}
22
23// update runs one transaction on a group.
24func update(t *testing.T, ci interface {
25 UpdateConcurrency(context.Context, domain.RepoName, string, func(*domain.CIConcurrencyGroupState) error) error
26}, n domain.RepoName, group string, fn func(*domain.CIConcurrencyGroupState)) {
27 t.Helper()
28 require.NoError(t, ci.UpdateConcurrency(context.Background(), n, group, func(st *domain.CIConcurrencyGroupState) error {
29 fn(st)
30 return nil
31 }))
32}
33
34type txer interface {
35 UpdateConcurrency(context.Context, domain.RepoName, string, func(*domain.CIConcurrencyGroupState) error) error
36}
37
38func read(t *testing.T, ci txer, n domain.RepoName, group string) (st domain.CIConcurrencyGroupState) {
39 t.Helper()
40 update(t, ci, n, group, func(s *domain.CIConcurrencyGroupState) { st = *s })
41 return st
42}
43
44func markOf(e domain.CIConcurrencyEntry, state domain.CIConcurrencyState) domain.CIConcurrencyMark {
45 return domain.CIConcurrencyMark{Entry: e, Group: "Deploy", State: state}
46}
47
48// A group is one record per repository, compared with domain.ConcurrencyKey, and
49// an empty group (no record, no mark) leaves no key.
50func TestCIConcurrencyGroupRecord(t *testing.T) {
51 _, ci, n := ciRepo(t)
52 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
53 s.Record.Running, s.Record.Pending = entry(1), []domain.CIConcurrencyEntry{{RunID: 2, Attempt: 1, JobID: "2-0-0", Cancel: true}}
54 })
55 got := read(t, ci, n, "DEPLOY")
56 require.NotNil(t, got.Record.Running)
57 require.Len(t, got.Record.Pending, 1)
58 assert.EqualValues(t, 1, got.Record.Running.RunID)
59 assert.Equal(t, "2-0-0", got.Record.Pending[0].JobID)
60 assert.True(t, got.Record.Pending[0].Cancel)
61 assert.True(t, read(t, ci, n, "other").IsEmpty(), "another group is another record")
62
63 update(t, ci, n, "deploy", func(s *domain.CIConcurrencyGroupState) { *s = domain.CIConcurrencyGroupState{} })
64 assert.True(t, read(t, ci, n, "Deploy").IsEmpty())
65 groups, err := ci.ListConcurrency(context.Background())
66 require.NoError(t, err)
67 assert.Empty(t, groups, "no key is left")
68}
69
70// The record and the marks of its entries are written by one batch, and a mark
71// that a transaction removes goes in the same batch.
72func TestCIConcurrencyMarksAreWrittenWithTheRecord(t *testing.T) {
73 _, ci, n := ciRepo(t)
74 run, wait := *entry(1), *entry(2)
75 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
76 s.Record.Running, s.Record.Pending = &run, []domain.CIConcurrencyEntry{wait}
77 s.Marks = map[string]domain.CIConcurrencyMark{
78 run.Key(): markOf(run, domain.CIConcurrencyHolding),
79 wait.Key(): markOf(wait, domain.CIConcurrencyPending),
80 }
81 })
82 got := read(t, ci, n, "deploy")
83 assert.Equal(t, domain.CIConcurrencyHolding, got.Marks[run.Key()].State)
84 assert.Equal(t, domain.CIConcurrencyPending, got.Marks[wait.Key()].State)
85 assert.Equal(t, "Deploy", got.Marks[wait.Key()].Group)
86
87 // The waiting entry is replaced: unnamed and marked in one transaction.
88 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
89 s.Record.Pending = []domain.CIConcurrencyEntry{*entry(3)}
90 s.Marks[wait.Key()] = markOf(wait, domain.CIConcurrencyReplaced)
91 s.Marks[entry(3).Key()] = markOf(*entry(3), domain.CIConcurrencyPending)
92 })
93 got = read(t, ci, n, "Deploy")
94 require.Len(t, got.Record.Pending, 1)
95 assert.EqualValues(t, 3, got.Record.Pending[0].RunID)
96 assert.Equal(t, domain.CIConcurrencyReplaced, got.Marks[wait.Key()].State)
97 assert.Len(t, got.Marks, 3)
98
99 // A group with marks only (nothing named) is kept, and listed with them.
100 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
101 s.Record = domain.CIConcurrencyRecord{}
102 delete(s.Marks, run.Key())
103 delete(s.Marks, entry(3).Key())
104 })
105 groups, err := ci.ListConcurrency(context.Background())
106 require.NoError(t, err)
107 require.Len(t, groups, 1)
108 assert.Equal(t, domain.ConcurrencyKey("Deploy"), groups[0].Group)
109 assert.Equal(t, n, groups[0].Repo)
110 assert.True(t, groups[0].State.Record.IsEmpty())
111 assert.Equal(t, domain.CIConcurrencyReplaced, groups[0].State.Marks[wait.Key()].State)
112
113 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) { delete(s.Marks, wait.Key()) })
114 groups, err = ci.ListConcurrency(context.Background())
115 require.NoError(t, err)
116 assert.Empty(t, groups, "no record and no mark: no key")
117}
118
119// The marks of "a" are not the marks of "a/b": a group may hold a slash.
120func TestCIConcurrencyMarksBelongToTheirOwnGroup(t *testing.T) {
121 _, ci, n := ciRepo(t)
122 for _, g := range []string{"a", "a/b"} {
123 e := *entry(int64(len(g)))
124 update(t, ci, n, g, func(s *domain.CIConcurrencyGroupState) {
125 s.Record.Running = &e
126 s.Marks = map[string]domain.CIConcurrencyMark{e.Key(): {Entry: e, Group: g, State: domain.CIConcurrencyHolding}}
127 })
128 }
129 a, ab := read(t, ci, n, "a"), read(t, ci, n, "a/b")
130 assert.Len(t, a.Marks, 1)
131 assert.Len(t, ab.Marks, 1)
132 assert.Equal(t, "a", a.Marks["1/1/"].Group)
133 assert.Equal(t, "a/b", ab.Marks["3/1/"].Group)
134}
135
136// Two writers that both read an empty group never both claim it: the batch of
137// the second carries the precondition of the state it read, so it conflicts and
138// its callback runs again on the winner's record.
139func TestCIConcurrencyGroupCompareAndSwap(t *testing.T) {
140 ctx := context.Background()
141 _, ci, n := ciRepo(t)
142
143 // Both first callbacks see the empty group before either write happens.
144 var arrived sync.WaitGroup
145 arrived.Add(2)
146 type writer struct {
147 claimed bool // reset by every call: it is the outcome of the last one
148 calls int
149 }
150 ws := []*writer{{}, {}}
151 var wg sync.WaitGroup
152 for i, w := range ws {
153 wg.Add(1)
154 go func() {
155 defer wg.Done()
156 err := ci.UpdateConcurrency(ctx, n, "g", func(s *domain.CIConcurrencyGroupState) error {
157 w.calls++
158 w.claimed = false
159 if w.calls == 1 {
160 assert.Nil(t, s.Record.Running, "the first call sees the empty group")
161 arrived.Done()
162 arrived.Wait()
163 }
164 if s.Record.Running == nil {
165 e := *entry(int64(i + 1))
166 s.Record.Running, w.claimed = &e, true
167 s.Marks = map[string]domain.CIConcurrencyMark{e.Key(): markOf(e, domain.CIConcurrencyHolding)}
168 }
169 return nil
170 })
171 assert.NoError(t, err)
172 }()
173 }
174 wg.Wait()
175
176 claimed, winner := 0, int64(0)
177 for i, w := range ws {
178 if w.claimed {
179 claimed, winner = claimed+1, int64(i+1)
180 }
181 }
182 assert.Equal(t, 1, claimed, "exactly one writer holds the group")
183 assert.Equal(t, 3, ws[0].calls+ws[1].calls, "the loser's callback ran again on the winner's record")
184 final := read(t, ci, n, "g")
185 require.NotNil(t, final.Record.Running)
186 assert.Equal(t, winner, final.Record.Running.RunID, "the stored owner is the one that claimed")
187 assert.Len(t, final.Marks, 1, "and it alone is marked")
188}
189
190// A transaction that only changes a mark still conflicts with one that changed
191// the record meanwhile: the mark rewrites the record, the version gate.
192func TestCIConcurrencyAMarkChangeIsGatedByTheRecord(t *testing.T) {
193 ctx := context.Background()
194 _, ci, n := ciRepo(t)
195 update(t, ci, n, "g", func(s *domain.CIConcurrencyGroupState) {
196 s.Record.Running = entry(1)
197 s.Marks = map[string]domain.CIConcurrencyMark{entry(1).Key(): markOf(*entry(1), domain.CIConcurrencyHolding)}
198 })
199 calls := 0
200 require.NoError(t, ci.UpdateConcurrency(ctx, n, "g", func(s *domain.CIConcurrencyGroupState) error {
201 calls++
202 if calls == 1 {
203 update(t, ci, n, "g", func(o *domain.CIConcurrencyGroupState) { o.Record.Pending = []domain.CIConcurrencyEntry{*entry(2)} })
204 }
205 m := s.Marks[entry(1).Key()]
206 m.State = domain.CIConcurrencyReplaced
207 s.Marks[entry(1).Key()] = m
208 return nil
209 }))
210 assert.Equal(t, 2, calls, "the first attempt was made on a stale record")
211 got := read(t, ci, n, "g")
212 require.Len(t, got.Record.Pending, 1, "the concurrent change is kept")
213 assert.Equal(t, domain.CIConcurrencyReplaced, got.Marks[entry(1).Key()].State)
214}
215
216// A callback that leaves the group as it was writes nothing, but its batch
217// still carries the version precondition: a writer that got in meanwhile makes
218// it run again.
219func TestCIConcurrencyUnchangedRecordKeepsItsPrecondition(t *testing.T) {
220 ctx := context.Background()
221 _, ci, n := ciRepo(t)
222 calls := 0
223 var second domain.CIConcurrencyGroupState
224 err := ci.UpdateConcurrency(ctx, n, "g", func(s *domain.CIConcurrencyGroupState) error {
225 calls++
226 if calls == 1 {
227 // Another writer creates the group while this decision is made.
228 update(t, ci, n, "g", func(o *domain.CIConcurrencyGroupState) { o.Record.Running = entry(7) })
229 } else {
230 second = *s
231 }
232 return nil // unchanged
233 })
234 require.NoError(t, err)
235 assert.Equal(t, 2, calls, "the unchanged decision was made on stale state: it ran again")
236 require.NotNil(t, second.Record.Running)
237 assert.EqualValues(t, 7, second.Record.Running.RunID)
238}
239
240// fn may change the entries it is given in place: the update is still written,
241// because the state is compared with its bytes as read, not with a shallow
242// copy that shares the entries.
243func TestCIConcurrencyInPlaceMutationIsWritten(t *testing.T) {
244 _, ci, n := ciRepo(t)
245 update(t, ci, n, "g", func(s *domain.CIConcurrencyGroupState) {
246 s.Record.Running, s.Record.Pending = entry(1), []domain.CIConcurrencyEntry{*entry(2)}
247 })
248
249 update(t, ci, n, "g", func(s *domain.CIConcurrencyGroupState) {
250 s.Record.Running.Canceling = true // through the pointer, the struct is returned as given
251 })
252 got := read(t, ci, n, "g")
253 require.NotNil(t, got.Record.Running)
254 assert.True(t, got.Record.Running.Canceling, "the in-place change of the running entry is stored")
255
256 update(t, ci, n, "g", func(s *domain.CIConcurrencyGroupState) {
257 s.Record.Pending[0].JobID, s.Record.Pending[0].Cancel = "2-0-0", true
258 })
259 got = read(t, ci, n, "g")
260 require.Len(t, got.Record.Pending, 1)
261 assert.Equal(t, "2-0-0", got.Record.Pending[0].JobID)
262 assert.True(t, got.Record.Pending[0].Cancel)
263}
264
265// ListConcurrency returns every group with its repository and canonical key.
266func TestCIConcurrencyListGroups(t *testing.T) {
267 _, ci, n := ciRepo(t)
268 for i, g := range []string{"Deploy", "Build/Linux"} {
269 update(t, ci, n, g, func(s *domain.CIConcurrencyGroupState) { s.Record.Running = entry(int64(i + 1)) })
270 }
271 got, err := ci.ListConcurrency(context.Background())
272 require.NoError(t, err)
273 require.Len(t, got, 2)
274 keys := []string{got[0].Group, got[1].Group}
275 assert.ElementsMatch(t, []string{"DEPLOY", "BUILD/LINUX"}, keys, "a group may hold a slash")
276 assert.Equal(t, n, got[0].Repo)
277 assert.NotNil(t, got[0].State.Record.Running)
278}
279
280// Groups and their marks go with their repository, and nothing is created under
281// one that is gone.
282func TestCIConcurrencyGroupsAreDeletedWithTheRepository(t *testing.T) {
283 ctx := context.Background()
284 s, ci, n := ciRepo(t)
285 update(t, ci, n, "g", func(st *domain.CIConcurrencyGroupState) {
286 st.Record.Running = entry(1)
287 st.Marks = map[string]domain.CIConcurrencyMark{entry(1).Key(): markOf(*entry(1), domain.CIConcurrencyHolding)}
288 })
289 require.NoError(t, s.Delete(ctx, n))
290 err := ci.UpdateConcurrency(ctx, n, "g", func(st *domain.CIConcurrencyGroupState) error {
291 st.Record.Running = entry(2)
292 return nil
293 })
294 assert.ErrorIs(t, err, domain.ErrNotFound, "no key is created under a repository that is gone")
295 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
296 assert.True(t, read(t, ci, n, "g").IsEmpty(), "the repository created again starts without groups or marks")
297 groups, err := ci.ListConcurrency(ctx)
298 require.NoError(t, err)
299 assert.Empty(t, groups)
300}
301
302// queue: max: the waiting line is stored in order with the queue mode of each
303// entry, and a rejected mark is kept as such.
304func TestCIConcurrencyQueueRoundTrip(t *testing.T) {
305 _, ci, n := ciRepo(t)
306 line := []domain.CIConcurrencyEntry{
307 {RunID: 4, Attempt: 1, Max: true},
308 {RunID: 2, Attempt: 1, JobID: "2-0-0", Max: true},
309 {RunID: 3, Attempt: 2},
310 }
311 rejected := domain.CIConcurrencyEntry{RunID: 5, Attempt: 1, Max: true}
312 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
313 s.Record.Running, s.Record.Pending = entry(1), line
314 s.Marks = map[string]domain.CIConcurrencyMark{rejected.Key(): markOf(rejected, domain.CIConcurrencyRejected)}
315 })
316 got := read(t, ci, n, "deploy")
317 assert.Equal(t, line, got.Record.Pending, "the order and the queue modes are kept")
318 assert.Equal(t, domain.CIConcurrencyRejected, got.Marks[rejected.Key()].State)
319 assert.True(t, got.Marks[rejected.Key()].Entry.Max)
320
321 // Stored as replaced with queue_full: a binary that predates it still
322 // cancels the entry.
323 var pb storagepb.CIConcurrencyMark
324 require.NoError(t, proto.Unmarshal(kvstore.EncodeConcurrencyMark(markOf(rejected, domain.CIConcurrencyRejected)), &pb))
325 assert.Equal(t, storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_REPLACED, pb.GetState())
326 assert.True(t, pb.GetQueueFull())
327}
328
329// A record stored before queue: max holds its only waiting entry in the
330// pending field: it reads as a line of one, and is written back in the new
331// layout.
332func TestCIConcurrencyRecordOldLayout(t *testing.T) {
333 old, err := proto.Marshal(&storagepb.CIConcurrencyRecord{
334 Running: &storagepb.CIConcurrencyEntry{RunId: 1, Attempt: 1},
335 Pending: &storagepb.CIConcurrencyEntry{RunId: 2, Attempt: 1, JobId: "2-0-0", Cancel: true},
336 })
337 require.NoError(t, err)
338 rec, err := kvstore.DecodeConcurrencyRecord(old)
339 require.NoError(t, err)
340 require.NotNil(t, rec.Running)
341 assert.EqualValues(t, 1, rec.Running.RunID)
342 assert.Equal(t, []domain.CIConcurrencyEntry{{RunID: 2, Attempt: 1, JobID: "2-0-0", Cancel: true}}, rec.Pending)
343
344 var pb storagepb.CIConcurrencyRecord
345 require.NoError(t, proto.Unmarshal(kvstore.EncodeConcurrencyRecord(rec), &pb))
346 assert.EqualValues(t, 2, pb.GetPending().GetRunId(), "the head of the line is written to the old field too, for a downgrade")
347 assert.Len(t, pb.GetQueue(), 1)
348
349 // A line of two: an older binary sees its head only.
350 rec.Pending = append(rec.Pending, domain.CIConcurrencyEntry{RunID: 3, Attempt: 1, Max: true})
351 pb.Reset()
352 require.NoError(t, proto.Unmarshal(kvstore.EncodeConcurrencyRecord(rec), &pb))
353 assert.EqualValues(t, 2, pb.GetPending().GetRunId())
354 assert.Len(t, pb.GetQueue(), 2)
355 back, err := kvstore.DecodeConcurrencyRecord(kvstore.EncodeConcurrencyRecord(rec))
356 require.NoError(t, err)
357 assert.Equal(t, rec, back, "the queue wins over the old field")
358}