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
52func encodeConcurrency(c domain.CIConcurrency) *storagepb.CIConcurrency {
53 if c == (domain.CIConcurrency{}) {
54 return nil
55 }
56 out := &storagepb.CIConcurrency{Group: c.Group, Cancel: c.Cancel, Evaluated: c.Evaluated, FromRun: c.FromRun, Preempted: c.Preempted}
57 for _, e := range concurrencyStates {
58 if e.d == c.State {
59 out.State = e.p
60 }
61 }
62 return out
63}
64
65func decodeConcurrency(pb *storagepb.CIConcurrency) domain.CIConcurrency {
66 if pb == nil {
67 return domain.CIConcurrency{}
68 }
69 c := domain.CIConcurrency{Group: pb.GetGroup(), Cancel: pb.GetCancel(), Evaluated: pb.GetEvaluated(), FromRun: pb.GetFromRun(), Preempted: pb.GetPreempted()}
70 for _, e := range concurrencyStates {
71 if e.p == pb.GetState() {
72 c.State = e.d
73 }
74 }
75 return c
76}
77
78func encodeConcurrencyEntry(e *domain.CIConcurrencyEntry) *storagepb.CIConcurrencyEntry {
79 if e == nil {
80 return nil
81 }
82 return &storagepb.CIConcurrencyEntry{RunId: e.RunID, Attempt: e.Attempt, JobId: e.JobID, Cancel: e.Cancel, Canceling: e.Canceling}
83}
84
85func encodeConcurrencyRecord(r domain.CIConcurrencyRecord) []byte {
86 return mustMarshal(&storagepb.CIConcurrencyRecord{Running: encodeConcurrencyEntry(r.Running), Pending: encodeConcurrencyEntry(r.Pending)})
87}
88
89//hexcheck:ignore suspicious-business-logic-in-adapter maps the stored record to the domain record field by field: no rule is decided here
90func decodeConcurrencyRecord(val []byte) (domain.CIConcurrencyRecord, error) {
91 var pb storagepb.CIConcurrencyRecord
92 if err := proto.Unmarshal(val, &pb); err != nil {
93 return domain.CIConcurrencyRecord{}, fmt.Errorf("decode concurrency group: %w", err)
94 }
95 return domain.CIConcurrencyRecord{Running: decodeConcurrencyEntry(pb.GetRunning()), Pending: decodeConcurrencyEntry(pb.GetPending())}, nil
96}
97
98func decodeConcurrencyEntry(pb *storagepb.CIConcurrencyEntry) *domain.CIConcurrencyEntry {
99 if pb == nil {
100 return nil
101 }
102 return &domain.CIConcurrencyEntry{RunID: pb.GetRunId(), Attempt: pb.GetAttempt(), JobID: pb.GetJobId(), Cancel: pb.GetCancel(), Canceling: pb.GetCanceling()}
103}
104
105// concurMarkPrefix is where the marks of a group are: the group key is escaped,
106// so that the marks of "a" never fall under the prefix of "a/b".
107func concurMarkPrefix(repo domain.RepoName, group string) string {
108 return concurMarkPrefixAll + repo.String() + "/" + url.PathEscape(domain.ConcurrencyKey(group)) + "/"
109}
110
111func encodeConcurrencyMark(m domain.CIConcurrencyMark) []byte {
112 pb := &storagepb.CIConcurrencyMark{Entry: encodeConcurrencyEntry(&m.Entry), Group: m.Group}
113 for _, e := range concurrencyStates {
114 if e.d == m.State {
115 pb.State = e.p
116 }
117 }
118 if m.State == domain.CIConcurrencyReplaced {
119 pb.State = storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_REPLACED
120 }
121 return mustMarshal(pb)
122}
123
124//hexcheck:ignore suspicious-business-logic-in-adapter maps the stored mark to the domain mark field by field: no rule is decided here
125func decodeConcurrencyMark(val []byte) (domain.CIConcurrencyMark, error) {
126 var pb storagepb.CIConcurrencyMark
127 if err := proto.Unmarshal(val, &pb); err != nil {
128 return domain.CIConcurrencyMark{}, fmt.Errorf("decode concurrency mark: %w", err)
129 }
130 m := domain.CIConcurrencyMark{Group: pb.GetGroup()}
131 if e := decodeConcurrencyEntry(pb.GetEntry()); e != nil {
132 m.Entry = *e
133 }
134 switch pb.GetState() {
135 case storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_PENDING:
136 m.State = domain.CIConcurrencyPending
137 case storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_HOLDING:
138 m.State = domain.CIConcurrencyHolding
139 case storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_REPLACED:
140 m.State = domain.CIConcurrencyReplaced
141 }
142 return m, nil
143}
144
145// encodeConcurrencyState is a canonical image of a group: the record, then the
146// marks in key order. Two states are the same when their images are.
147func encodeConcurrencyState(s domain.CIConcurrencyGroupState) []byte {
148 out := encodeConcurrencyRecord(s.Record)
149 for _, k := range slices.Sorted(maps.Keys(s.Marks)) {
150 out = append(out, 0)
151 out = append(out, k...)
152 out = append(out, 0)
153 out = append(out, encodeConcurrencyMark(s.Marks[k])...)
154 }
155 return out
156}
157
158// UpdateConcurrency reads a group with its marks, calls fn on it and stores the
159// result in ONE batch: the record (key "concur/..."), every mark that changed
160// ("concurmark/...", written or deleted) and, when something changed, the guard
161// of the repository. The batch requires the version the record had when it was
162// read (or its absence) whatever fn does, and every change rewrites the record,
163// so of two transactions that read the same state only one commits; the other
164// calls fn again on the new state.
165//
166//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
167func (c *CI) UpdateConcurrency(ctx context.Context, repo domain.RepoName, group string, fn func(*domain.CIConcurrencyGroupState) error) error {
168 key, mprefix := concurKey(repo, group), concurMarkPrefix(repo, group)
169 for range updateConcurrencyTries {
170 var state domain.CIConcurrencyGroupState
171 var version uint64
172 if item, ok := c.db.Get(ctx, key); ok {
173 var err error
174 if state.Record, err = decodeConcurrencyRecord(item.Value); err != nil {
175 return err
176 }
177 version = item.Version
178 }
179 // The marks are read after the record: a transaction that commits in
180 // between has changed the record, and the batch below conflicts.
181 state.Marks = map[string]domain.CIConcurrencyMark{}
182 for _, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(mprefix)) {
183 m, err := decodeConcurrencyMark(item.Value)
184 if err != nil {
185 return err
186 }
187 state.Marks[m.Entry.Key()] = m
188 }
189 // The state as read, in bytes, before fn sees it: fn may change the
190 // entries it is given in place, and a copy of the struct shares them.
191 before, stored := encodeConcurrencyState(state), maps.Clone(state.Marks)
192 if err := fn(&state); err != nil {
193 return err
194 }
195 var b kv.Batch
196 if version == 0 {
197 b.RequireAbsent(key)
198 } else {
199 b.RequireVersion(key, version)
200 }
201 // A state left as it was is not written, but the batch still holds the
202 // precondition: the decision was made on the state that was read.
203 if !bytes.Equal(before, encodeConcurrencyState(state)) {
204 for k, m := range state.Marks {
205 if old, ok := stored[k]; !ok || old != m {
206 b.Put([]byte(mprefix+k), encodeConcurrencyMark(m))
207 }
208 }
209 for k := range stored {
210 if _, ok := state.Marks[k]; !ok {
211 b.Delete([]byte(mprefix + k))
212 }
213 }
214 if state.IsEmpty() {
215 b.Delete(key)
216 } else {
217 b.Put(key, encodeConcurrencyRecord(state.Record)) // rewritten even if equal: it is the version gate
218 if err := c.guardRepo(ctx, &b, repo); err != nil {
219 return err
220 }
221 }
222 }
223 if _, err := c.db.Apply(ctx, &b); err != nil {
224 if errors.Is(err, kv.ErrConflict) {
225 continue
226 }
227 return fmt.Errorf("write concurrency group: %w", err)
228 }
229 return nil
230 }
231 return fmt.Errorf("%w: concurrency group is busy", domain.ErrConflict)
232}
233
234// ListConcurrency returns every stored group with its marks.
235//
236//hexcheck:ignore suspicious-business-logic-in-adapter lists and decodes records and marks: no rule is decided here
237func (c *CI) ListConcurrency(ctx context.Context) ([]ports.CIConcurrencyGroup, error) {
238 type id struct{ repo, group string }
239 snap := c.db.Snapshot(ctx)
240 byID := map[id]*ports.CIConcurrencyGroup{}
241 get := func(repoName, group string) *ports.CIConcurrencyGroup {
242 repo, err := domain.ParseRepoName(repoName)
243 if err != nil {
244 return nil
245 }
246 g, ok := byID[id{repoName, group}]
247 if !ok {
248 g = &ports.CIConcurrencyGroup{Repo: repo, Group: group, State: domain.CIConcurrencyGroupState{Marks: map[string]domain.CIConcurrencyMark{}}}
249 byID[id{repoName, group}] = g
250 }
251 return g
252 }
253 for key, item := range snap.Prefix(ctx, []byte(concurPrefix)) {
254 repoName, group, ok := strings.Cut(strings.TrimPrefix(string(key), concurPrefix), "/")
255 if !ok {
256 continue
257 }
258 g := get(repoName, group)
259 if g == nil {
260 continue
261 }
262 rec, err := decodeConcurrencyRecord(item.Value)
263 if err != nil {
264 return nil, err
265 }
266 g.State.Record = rec
267 }
268 for key, item := range snap.Prefix(ctx, []byte(concurMarkPrefixAll)) {
269 repoName, rest, ok := strings.Cut(strings.TrimPrefix(string(key), concurMarkPrefixAll), "/")
270 if !ok {
271 continue
272 }
273 escaped, _, ok := strings.Cut(rest, "/")
274 if !ok {
275 continue
276 }
277 group, err := url.PathUnescape(escaped)
278 if err != nil {
279 continue
280 }
281 g := get(repoName, group)
282 if g == nil {
283 continue
284 }
285 m, err := decodeConcurrencyMark(item.Value)
286 if err != nil {
287 return nil, err
288 }
289 g.State.Marks[m.Entry.Key()] = m
290 }
291 out := make([]ports.CIConcurrencyGroup, 0, len(byID))
292 for _, g := range byID {
293 out = append(out, *g)
294 }
295 return out, nil
296}