1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "bytes"
7 "context"
8 "errors"
9 "fmt"
10 "maps"
11 "net/url"
12 "slices"
13 "strings"
14
15 kv "bnema.dev/ermite-kv"
16 "google.golang.org/protobuf/proto"
17
18 storagepb "bnema.dev/ermite/gen/ermite/storage"
19 "bnema.dev/ermite/internal/domain"
20 "bnema.dev/ermite/internal/ports"
21)
22
23// Key layout of the concurrency groups:
24//
25// concur/<repo>/<key(group)> CIConcurrencyRecord (running and pending entry)
26// concurmark/<repo>/<escaped key(group)>/<run>/<attempt>/<job id>
27// CIConcurrencyMark, written in the same batch
28//
29// The group key disappears only when both the record and its marks are empty.
30// Both prefixes are removed with their repository (ciKeys).
31const (
32 concurPrefix = "concur/"
33 concurMarkPrefixAll = "concurmark/"
34)
35
36// updateConcurrencyTries bounds the retries of UpdateConcurrency under
37// contention.
38const updateConcurrencyTries = 32
39
40func concurKey(repo domain.RepoName, group string) []byte {
41 return []byte(concurPrefix + repo.String() + "/" + domain.ConcurrencyKey(group))
42}
43
44var concurrencyStates = []struct {
45 d domain.CIConcurrencyState
46 p storagepb.CIConcurrencyState
47}{
48 {domain.CIConcurrencyPending, storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_PENDING},
49 {domain.CIConcurrencyHolding, storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_HOLDING},
50}
51
52// markStates are the states a mark may hold on top of concurrencyStates.
53var markStates = append(slices.Clone(concurrencyStates), []struct {
54 d domain.CIConcurrencyState
55 p storagepb.CIConcurrencyState
56}{
57 {domain.CIConcurrencyReplaced, storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_REPLACED},
58}...)
59
60func encodeConcurrency(c domain.CIConcurrency) *storagepb.CIConcurrency {
61 if c == (domain.CIConcurrency{}) {
62 return nil
63 }
64 out := &storagepb.CIConcurrency{Group: c.Group, Cancel: c.Cancel, Evaluated: c.Evaluated, FromRun: c.FromRun, Preempted: c.Preempted, QueueMax: c.Max}
65 for _, e := range concurrencyStates {
66 if e.d == c.State {
67 out.State = e.p
68 }
69 }
70 return out
71}
72
73func decodeConcurrency(pb *storagepb.CIConcurrency) domain.CIConcurrency {
74 if pb == nil {
75 return domain.CIConcurrency{}
76 }
77 c := domain.CIConcurrency{Group: pb.GetGroup(), Cancel: pb.GetCancel(), Evaluated: pb.GetEvaluated(), FromRun: pb.GetFromRun(), Preempted: pb.GetPreempted(), Max: pb.GetQueueMax()}
78 for _, e := range concurrencyStates {
79 if e.p == pb.GetState() {
80 c.State = e.d
81 }
82 }
83 return c
84}
85
86func encodeConcurrencyEntry(e *domain.CIConcurrencyEntry) *storagepb.CIConcurrencyEntry {
87 if e == nil {
88 return nil
89 }
90 return &storagepb.CIConcurrencyEntry{RunId: e.RunID, Attempt: e.Attempt, JobId: e.JobID, Cancel: e.Cancel, Canceling: e.Canceling, QueueMax: e.Max}
91}
92
93// encodeConcurrencyRecord writes the waiting entries to queue, and the head of
94// the line to pending too, for a binary that predates queue (a downgrade).
95func encodeConcurrencyRecord(r domain.CIConcurrencyRecord) []byte {
96 pb := &storagepb.CIConcurrencyRecord{Running: encodeConcurrencyEntry(r.Running)}
97 for i := range r.Pending {
98 pb.Queue = append(pb.Queue, encodeConcurrencyEntry(&r.Pending[i]))
99 }
100 if len(r.Pending) > 0 {
101 pb.Pending = encodeConcurrencyEntry(&r.Pending[0])
102 }
103 return mustMarshal(pb)
104}
105
106//hexcheck:ignore suspicious-business-logic-in-adapter maps the stored record to the domain record field by field: no rule is decided here
107func decodeConcurrencyRecord(val []byte) (domain.CIConcurrencyRecord, error) {
108 var pb storagepb.CIConcurrencyRecord
109 if err := proto.Unmarshal(val, &pb); err != nil {
110 return domain.CIConcurrencyRecord{}, fmt.Errorf("decode concurrency group: %w", err)
111 }
112 rec := domain.CIConcurrencyRecord{Running: decodeConcurrencyEntry(pb.GetRunning())}
113 for _, q := range pb.GetQueue() {
114 rec.Pending = append(rec.Pending, *decodeConcurrencyEntry(q))
115 }
116 // A record stored before queue: max holds its only waiting entry in pending.
117 if p := decodeConcurrencyEntry(pb.GetPending()); p != nil && len(rec.Pending) == 0 {
118 rec.Pending = []domain.CIConcurrencyEntry{*p}
119 }
120 return rec, nil
121}
122
123func decodeConcurrencyEntry(pb *storagepb.CIConcurrencyEntry) *domain.CIConcurrencyEntry {
124 if pb == nil {
125 return nil
126 }
127 return &domain.CIConcurrencyEntry{RunID: pb.GetRunId(), Attempt: pb.GetAttempt(), JobID: pb.GetJobId(), Cancel: pb.GetCancel(), Canceling: pb.GetCanceling(), Max: pb.GetQueueMax()}
128}
129
130// concurMarkPrefix is where the marks of a group are: the group key is escaped,
131// so that the marks of "a" never fall under the prefix of "a/b".
132func concurMarkPrefix(repo domain.RepoName, group string) string {
133 return concurMarkPrefixAll + repo.String() + "/" + url.PathEscape(domain.ConcurrencyKey(group)) + "/"
134}
135
136func encodeConcurrencyMark(m domain.CIConcurrencyMark) []byte {
137 pb := &storagepb.CIConcurrencyMark{Entry: encodeConcurrencyEntry(&m.Entry), Group: m.Group}
138 for _, e := range markStates {
139 if e.d == m.State {
140 pb.State = e.p
141 }
142 }
143 if m.State == domain.CIConcurrencyRejected {
144 pb.State, pb.QueueFull = storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_REPLACED, true
145 }
146 return mustMarshal(pb)
147}
148
149//hexcheck:ignore suspicious-business-logic-in-adapter maps the stored mark to the domain mark field by field: no rule is decided here
150func decodeConcurrencyMark(val []byte) (domain.CIConcurrencyMark, error) {
151 var pb storagepb.CIConcurrencyMark
152 if err := proto.Unmarshal(val, &pb); err != nil {
153 return domain.CIConcurrencyMark{}, fmt.Errorf("decode concurrency mark: %w", err)
154 }
155 m := domain.CIConcurrencyMark{Group: pb.GetGroup()}
156 if e := decodeConcurrencyEntry(pb.GetEntry()); e != nil {
157 m.Entry = *e
158 }
159 for _, e := range markStates {
160 if e.p == pb.GetState() {
161 m.State = e.d
162 }
163 }
164 if m.State == domain.CIConcurrencyReplaced && pb.GetQueueFull() {
165 m.State = domain.CIConcurrencyRejected
166 }
167 return m, nil
168}
169
170// encodeConcurrencyState is a canonical image of a group: the record, then the
171// marks in key order. Two states are the same when their images are.
172func encodeConcurrencyState(s domain.CIConcurrencyGroupState) []byte {
173 out := encodeConcurrencyRecord(s.Record)
174 for _, k := range slices.Sorted(maps.Keys(s.Marks)) {
175 out = append(out, 0)
176 out = append(out, k...)
177 out = append(out, 0)
178 out = append(out, encodeConcurrencyMark(s.Marks[k])...)
179 }
180 return out
181}
182
183// UpdateConcurrency reads a group with its marks, calls fn on it and stores the
184// result in ONE batch: the record (key "concur/..."), every mark that changed
185// ("concurmark/...", written or deleted) and, when something changed, the guard
186// of the repository. The batch requires the version the record had when it was
187// read (or its absence) whatever fn does, and every change rewrites the record,
188// so of two transactions that read the same state only one commits; the other
189// calls fn again on the new state.
190//
191//hexcheck:ignore suspicious-business-logic-in-adapter compare-and-swap loop over a record and its marks in one batch: the rule is fn, decided by the caller
192func (c *CI) UpdateConcurrency(ctx context.Context, repo domain.RepoName, group string, fn func(*domain.CIConcurrencyGroupState) error) error {
193 key, mprefix := concurKey(repo, group), concurMarkPrefix(repo, group)
194 for range updateConcurrencyTries {
195 var state domain.CIConcurrencyGroupState
196 var version uint64
197 if item, ok := c.db.Get(ctx, key); ok {
198 var err error
199 if state.Record, err = decodeConcurrencyRecord(item.Value); err != nil {
200 return err
201 }
202 version = item.Version
203 }
204 // The marks are read after the record: a transaction that commits in
205 // between has changed the record, and the batch below conflicts.
206 state.Marks = map[string]domain.CIConcurrencyMark{}
207 for _, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(mprefix)) {
208 m, err := decodeConcurrencyMark(item.Value)
209 if err != nil {
210 return err
211 }
212 state.Marks[m.Entry.Key()] = m
213 }
214 // The state as read, in bytes, before fn sees it: fn may change the
215 // entries it is given in place, and a copy of the struct shares them.
216 before, stored := encodeConcurrencyState(state), maps.Clone(state.Marks)
217 if err := fn(&state); err != nil {
218 return err
219 }
220 var b kv.Batch
221 if version == 0 {
222 b.RequireAbsent(key)
223 } else {
224 b.RequireVersion(key, version)
225 }
226 // A state left as it was is not written, but the batch still holds the
227 // precondition: the decision was made on the state that was read.
228 if !bytes.Equal(before, encodeConcurrencyState(state)) {
229 for k, m := range state.Marks {
230 if old, ok := stored[k]; !ok || old != m {
231 b.Put([]byte(mprefix+k), encodeConcurrencyMark(m))
232 }
233 }
234 for k := range stored {
235 if _, ok := state.Marks[k]; !ok {
236 b.Delete([]byte(mprefix + k))
237 }
238 }
239 if state.IsEmpty() {
240 b.Delete(key)
241 } else {
242 b.Put(key, encodeConcurrencyRecord(state.Record)) // rewritten even if equal: it is the version gate
243 if err := c.guardRepo(ctx, &b, repo); err != nil {
244 return err
245 }
246 }
247 }
248 if _, err := c.db.Apply(ctx, &b); err != nil {
249 if errors.Is(err, kv.ErrConflict) {
250 continue
251 }
252 return fmt.Errorf("write concurrency group: %w", err)
253 }
254 return nil
255 }
256 return fmt.Errorf("%w: concurrency group is busy", domain.ErrConflict)
257}
258
259// ListConcurrency returns every stored group with its marks.
260//
261//hexcheck:ignore suspicious-business-logic-in-adapter lists and decodes records and marks: no rule is decided here
262func (c *CI) ListConcurrency(ctx context.Context) ([]ports.CIConcurrencyGroup, error) {
263 type id struct{ repo, group string }
264 snap := c.db.Snapshot(ctx)
265 byID := map[id]*ports.CIConcurrencyGroup{}
266 get := func(repoName, group string) *ports.CIConcurrencyGroup {
267 repo, err := domain.ParseRepoName(repoName)
268 if err != nil {
269 return nil
270 }
271 g, ok := byID[id{repoName, group}]
272 if !ok {
273 g = &ports.CIConcurrencyGroup{Repo: repo, Group: group, State: domain.CIConcurrencyGroupState{Marks: map[string]domain.CIConcurrencyMark{}}}
274 byID[id{repoName, group}] = g
275 }
276 return g
277 }
278 for key, item := range snap.Prefix(ctx, []byte(concurPrefix)) {
279 repoName, group, ok := strings.Cut(strings.TrimPrefix(string(key), concurPrefix), "/")
280 if !ok {
281 continue
282 }
283 g := get(repoName, group)
284 if g == nil {
285 continue
286 }
287 rec, err := decodeConcurrencyRecord(item.Value)
288 if err != nil {
289 return nil, err
290 }
291 g.State.Record = rec
292 }
293 for key, item := range snap.Prefix(ctx, []byte(concurMarkPrefixAll)) {
294 repoName, rest, ok := strings.Cut(strings.TrimPrefix(string(key), concurMarkPrefixAll), "/")
295 if !ok {
296 continue
297 }
298 escaped, _, ok := strings.Cut(rest, "/")
299 if !ok {
300 continue
301 }
302 group, err := url.PathUnescape(escaped)
303 if err != nil {
304 continue
305 }
306 g := get(repoName, group)
307 if g == nil {
308 continue
309 }
310 m, err := decodeConcurrencyMark(item.Value)
311 if err != nil {
312 return nil, err
313 }
314 g.State.Marks[m.Entry.Key()] = m
315 }
316 out := make([]ports.CIConcurrencyGroup, 0, len(byID))
317 for _, g := range byID {
318 out = append(out, *g)
319 }
320 return out, nil
321}