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
13 "bnema.dev/ermite/internal/domain"
14)
15
16func entry(run int64) *domain.CIConcurrencyEntry {
17 return &domain.CIConcurrencyEntry{RunID: run, Attempt: 1}
18}
19
20// update runs one transaction on a group.
21func update(t *testing.T, ci interface {
22 UpdateConcurrency(context.Context, domain.RepoName, string, func(*domain.CIConcurrencyGroupState) error) error
23}, n domain.RepoName, group string, fn func(*domain.CIConcurrencyGroupState)) {
24 t.Helper()
25 require.NoError(t, ci.UpdateConcurrency(context.Background(), n, group, func(st *domain.CIConcurrencyGroupState) error {
26 fn(st)
27 return nil
28 }))
29}
30
31type txer interface {
32 UpdateConcurrency(context.Context, domain.RepoName, string, func(*domain.CIConcurrencyGroupState) error) error
33}
34
35func read(t *testing.T, ci txer, n domain.RepoName, group string) (st domain.CIConcurrencyGroupState) {
36 t.Helper()
37 update(t, ci, n, group, func(s *domain.CIConcurrencyGroupState) { st = *s })
38 return st
39}
40
41func markOf(e domain.CIConcurrencyEntry, state domain.CIConcurrencyState) domain.CIConcurrencyMark {
42 return domain.CIConcurrencyMark{Entry: e, Group: "Deploy", State: state}
43}
44
45// A group is one record per repository, compared with domain.ConcurrencyKey, and
46// an empty group (no record, no mark) leaves no key.
47func TestCIConcurrencyGroupRecord(t *testing.T) {
48 _, ci, n := ciRepo(t)
49 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
50 s.Record.Running, s.Record.Pending = entry(1), &domain.CIConcurrencyEntry{RunID: 2, Attempt: 1, JobID: "2-0-0", Cancel: true}
51 })
52 got := read(t, ci, n, "DEPLOY")
53 require.NotNil(t, got.Record.Running)
54 require.NotNil(t, got.Record.Pending)
55 assert.EqualValues(t, 1, got.Record.Running.RunID)
56 assert.Equal(t, "2-0-0", got.Record.Pending.JobID)
57 assert.True(t, got.Record.Pending.Cancel)
58 assert.True(t, read(t, ci, n, "other").IsEmpty(), "another group is another record")
59
60 update(t, ci, n, "deploy", func(s *domain.CIConcurrencyGroupState) { *s = domain.CIConcurrencyGroupState{} })
61 assert.True(t, read(t, ci, n, "Deploy").IsEmpty())
62 groups, err := ci.ListConcurrency(context.Background())
63 require.NoError(t, err)
64 assert.Empty(t, groups, "no key is left")
65}
66
67// The record and the marks of its entries are written by one batch, and a mark
68// that a transaction removes goes in the same batch.
69func TestCIConcurrencyMarksAreWrittenWithTheRecord(t *testing.T) {
70 _, ci, n := ciRepo(t)
71 run, wait := *entry(1), *entry(2)
72 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
73 s.Record.Running, s.Record.Pending = &run, &wait
74 s.Marks = map[string]domain.CIConcurrencyMark{
75 run.Key(): markOf(run, domain.CIConcurrencyHolding),
76 wait.Key(): markOf(wait, domain.CIConcurrencyPending),
77 }
78 })
79 got := read(t, ci, n, "deploy")
80 assert.Equal(t, domain.CIConcurrencyHolding, got.Marks[run.Key()].State)
81 assert.Equal(t, domain.CIConcurrencyPending, got.Marks[wait.Key()].State)
82 assert.Equal(t, "Deploy", got.Marks[wait.Key()].Group)
83
84 // The waiting entry is replaced: unnamed and marked in one transaction.
85 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
86 s.Record.Pending = entry(3)
87 s.Marks[wait.Key()] = markOf(wait, domain.CIConcurrencyReplaced)
88 s.Marks[entry(3).Key()] = markOf(*entry(3), domain.CIConcurrencyPending)
89 })
90 got = read(t, ci, n, "Deploy")
91 assert.EqualValues(t, 3, got.Record.Pending.RunID)
92 assert.Equal(t, domain.CIConcurrencyReplaced, got.Marks[wait.Key()].State)
93 assert.Len(t, got.Marks, 3)
94
95 // A group with marks only (nothing named) is kept, and listed with them.
96 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) {
97 s.Record = domain.CIConcurrencyRecord{}
98 delete(s.Marks, run.Key())
99 delete(s.Marks, entry(3).Key())
100 })
101 groups, err := ci.ListConcurrency(context.Background())
102 require.NoError(t, err)
103 require.Len(t, groups, 1)
104 assert.Equal(t, domain.ConcurrencyKey("Deploy"), groups[0].Group)
105 assert.Equal(t, n, groups[0].Repo)
106 assert.True(t, groups[0].State.Record.IsEmpty())
107 assert.Equal(t, domain.CIConcurrencyReplaced, groups[0].State.Marks[wait.Key()].State)
108
109 update(t, ci, n, "Deploy", func(s *domain.CIConcurrencyGroupState) { delete(s.Marks, wait.Key()) })
110 groups, err = ci.ListConcurrency(context.Background())
111 require.NoError(t, err)
112 assert.Empty(t, groups, "no record and no mark: no key")
113}
114
115// The marks of "a" are not the marks of "a/b": a group may hold a slash.
116func TestCIConcurrencyMarksBelongToTheirOwnGroup(t *testing.T) {
117 _, ci, n := ciRepo(t)
118 for _, g := range []string{"a", "a/b"} {
119 e := *entry(int64(len(g)))
120 update(t, ci, n, g, func(s *domain.CIConcurrencyGroupState) {
121 s.Record.Running = &e
122 s.Marks = map[string]domain.CIConcurrencyMark{e.Key(): {Entry: e, Group: g, State: domain.CIConcurrencyHolding}}
123 })
124 }
125 a, ab := read(t, ci, n, "a"), read(t, ci, n, "a/b")
126 assert.Len(t, a.Marks, 1)
127 assert.Len(t, ab.Marks, 1)
128 assert.Equal(t, "a", a.Marks["1/1/"].Group)
129 assert.Equal(t, "a/b", ab.Marks["3/1/"].Group)
130}
131
132// Two writers that both read an empty group never both claim it: the batch of
133// the second carries the precondition of the state it read, so it conflicts and
134// its callback runs again on the winner's record.
135func TestCIConcurrencyGroupCompareAndSwap(t *testing.T) {
136 ctx := context.Background()
137 _, ci, n := ciRepo(t)
138
139 // Both first callbacks see the empty group before either write happens.
140 var arrived sync.WaitGroup
141 arrived.Add(2)
142 type writer struct {
143 claimed bool // reset by every call: it is the outcome of the last one
144 calls int
145 }
146 ws := []*writer{{}, {}}
147 var wg sync.WaitGroup
148 for i, w := range ws {
149 wg.Add(1)
150 go func() {
151 defer wg.Done()
152 err := ci.UpdateConcurrency(ctx, n, "g", func(s *domain.CIConcurrencyGroupState) error {
153 w.calls++
154 w.claimed = false
155 if w.calls == 1 {
156 assert.Nil(t, s.Record.Running, "the first call sees the empty group")
157 arrived.Done()
158 arrived.Wait()
159 }
160 if s.Record.Running == nil {
161 e := *entry(int64(i + 1))
162 s.Record.Running, w.claimed = &e, true
163 s.Marks = map[string]domain.CIConcurrencyMark{e.Key(): markOf(e, domain.CIConcurrencyHolding)}
164 }
165 return nil
166 })
167 assert.NoError(t, err)
168 }()
169 }
170 wg.Wait()
171
172 claimed, winner := 0, int64(0)
173 for i, w := range ws {
174 if w.claimed {
175 claimed, winner = claimed+1, int64(i+1)
176 }
177 }
178 assert.Equal(t, 1, claimed, "exactly one writer holds the group")
179 assert.Equal(t, 3, ws[0].calls+ws[1].calls, "the loser's callback ran again on the winner's record")
180 final := read(t, ci, n, "g")
181 require.NotNil(t, final.Record.Running)
182 assert.Equal(t, winner, final.Record.Running.RunID, "the stored owner is the one that claimed")
183 assert.Len(t, final.Marks, 1, "and it alone is marked")
184}
185
186// A transaction that only changes a mark still conflicts with one that changed
187// the record meanwhile: the mark rewrites the record, the version gate.
188func TestCIConcurrencyAMarkChangeIsGatedByTheRecord(t *testing.T) {
189 ctx := context.Background()
190 _, ci, n := ciRepo(t)
191 update(t, ci, n, "g", func(s *domain.CIConcurrencyGroupState) {
192 s.Record.Running = entry(1)
193 s.Marks = map[string]domain.CIConcurrencyMark{entry(1).Key(): markOf(*entry(1), domain.CIConcurrencyHolding)}
194 })
195 calls := 0
196 require.NoError(t, ci.UpdateConcurrency(ctx, n, "g", func(s *domain.CIConcurrencyGroupState) error {
197 calls++
198 if calls == 1 {
199 update(t, ci, n, "g", func(o *domain.CIConcurrencyGroupState) { o.Record.Pending = entry(2) })
200 }
201 m := s.Marks[entry(1).Key()]
202 m.State = domain.CIConcurrencyReplaced
203 s.Marks[entry(1).Key()] = m
204 return nil
205 }))
206 assert.Equal(t, 2, calls, "the first attempt was made on a stale record")
207 got := read(t, ci, n, "g")
208 require.NotNil(t, got.Record.Pending, "the concurrent change is kept")
209 assert.Equal(t, domain.CIConcurrencyReplaced, got.Marks[entry(1).Key()].State)
210}
211
212// A callback that leaves the group as it was writes nothing, but its batch
213// still carries the version precondition: a writer that got in meanwhile makes
214// it run again.
215func TestCIConcurrencyUnchangedRecordKeepsItsPrecondition(t *testing.T) {
216 ctx := context.Background()
217 _, ci, n := ciRepo(t)
218 calls := 0
219 var second domain.CIConcurrencyGroupState
220 err := ci.UpdateConcurrency(ctx, n, "g", func(s *domain.CIConcurrencyGroupState) error {
221 calls++
222 if calls == 1 {
223 // Another writer creates the group while this decision is made.
224 update(t, ci, n, "g", func(o *domain.CIConcurrencyGroupState) { o.Record.Running = entry(7) })
225 } else {
226 second = *s
227 }
228 return nil // unchanged
229 })
230 require.NoError(t, err)
231 assert.Equal(t, 2, calls, "the unchanged decision was made on stale state: it ran again")
232 require.NotNil(t, second.Record.Running)
233 assert.EqualValues(t, 7, second.Record.Running.RunID)
234}
235
236// fn may change the entries it is given in place: the update is still written,
237// because the state is compared with its bytes as read, not with a shallow
238// copy that shares the entries.
239func TestCIConcurrencyInPlaceMutationIsWritten(t *testing.T) {
240 _, ci, n := ciRepo(t)
241 update(t, ci, n, "g", func(s *domain.CIConcurrencyGroupState) { s.Record.Running, s.Record.Pending = entry(1), entry(2) })
242
243 update(t, ci, n, "g", func(s *domain.CIConcurrencyGroupState) {
244 s.Record.Running.Canceling = true // through the pointer, the struct is returned as given
245 })
246 got := read(t, ci, n, "g")
247 require.NotNil(t, got.Record.Running)
248 assert.True(t, got.Record.Running.Canceling, "the in-place change of the running entry is stored")
249
250 update(t, ci, n, "g", func(s *domain.CIConcurrencyGroupState) {
251 s.Record.Pending.JobID, s.Record.Pending.Cancel = "2-0-0", true
252 })
253 got = read(t, ci, n, "g")
254 require.NotNil(t, got.Record.Pending)
255 assert.Equal(t, "2-0-0", got.Record.Pending.JobID)
256 assert.True(t, got.Record.Pending.Cancel)
257}
258
259// ListConcurrency returns every group with its repository and canonical key.
260func TestCIConcurrencyListGroups(t *testing.T) {
261 _, ci, n := ciRepo(t)
262 for i, g := range []string{"Deploy", "Build/Linux"} {
263 update(t, ci, n, g, func(s *domain.CIConcurrencyGroupState) { s.Record.Running = entry(int64(i + 1)) })
264 }
265 got, err := ci.ListConcurrency(context.Background())
266 require.NoError(t, err)
267 require.Len(t, got, 2)
268 keys := []string{got[0].Group, got[1].Group}
269 assert.ElementsMatch(t, []string{"DEPLOY", "BUILD/LINUX"}, keys, "a group may hold a slash")
270 assert.Equal(t, n, got[0].Repo)
271 assert.NotNil(t, got[0].State.Record.Running)
272}
273
274// Groups and their marks go with their repository, and nothing is created under
275// one that is gone.
276func TestCIConcurrencyGroupsAreDeletedWithTheRepository(t *testing.T) {
277 ctx := context.Background()
278 s, ci, n := ciRepo(t)
279 update(t, ci, n, "g", func(st *domain.CIConcurrencyGroupState) {
280 st.Record.Running = entry(1)
281 st.Marks = map[string]domain.CIConcurrencyMark{entry(1).Key(): markOf(*entry(1), domain.CIConcurrencyHolding)}
282 })
283 require.NoError(t, s.Delete(ctx, n))
284 err := ci.UpdateConcurrency(ctx, n, "g", func(st *domain.CIConcurrencyGroupState) error {
285 st.Record.Running = entry(2)
286 return nil
287 })
288 assert.ErrorIs(t, err, domain.ErrNotFound, "no key is created under a repository that is gone")
289 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
290 assert.True(t, read(t, ci, n, "g").IsEmpty(), "the repository created again starts without groups or marks")
291 groups, err := ci.ListConcurrency(ctx)
292 require.NoError(t, err)
293 assert.Empty(t, groups)
294}