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
106func decodeConcurrencyRecord(val []byte) (domain.CIConcurrencyRecord, error) {
107 var pb storagepb.CIConcurrencyRecord
108 if err := proto.Unmarshal(val, &pb); err != nil {
109 return domain.CIConcurrencyRecord{}, fmt.Errorf("decode concurrency group: %w", err)
110 }
111 rec := domain.CIConcurrencyRecord{Running: decodeConcurrencyEntry(pb.GetRunning())}
112 for _, q := range pb.GetQueue() {
113 rec.Pending = append(rec.Pending, *decodeConcurrencyEntry(q))
114 }
115 // A record stored before queue: max holds its only waiting entry in pending.
116 if p := decodeConcurrencyEntry(pb.GetPending()); p != nil && len(rec.Pending) == 0 {
117 rec.Pending = []domain.CIConcurrencyEntry{*p}
118 }
119 return rec, nil
120}
121
122func decodeConcurrencyEntry(pb *storagepb.CIConcurrencyEntry) *domain.CIConcurrencyEntry {
123 if pb == nil {
124 return nil
125 }
126 return &domain.CIConcurrencyEntry{RunID: pb.GetRunId(), Attempt: pb.GetAttempt(), JobID: pb.GetJobId(), Cancel: pb.GetCancel(), Canceling: pb.GetCanceling(), Max: pb.GetQueueMax()}
127}
128
129// concurMarkPrefix is where the marks of a group are: the group key is escaped,
130// so that the marks of "a" never fall under the prefix of "a/b".
131func concurMarkPrefix(repo domain.RepoName, group string) string {
132 return concurMarkPrefixAll + repo.String() + "/" + url.PathEscape(domain.ConcurrencyKey(group)) + "/"
133}
134
135func encodeConcurrencyMark(m domain.CIConcurrencyMark) []byte {
136 pb := &storagepb.CIConcurrencyMark{Entry: encodeConcurrencyEntry(&m.Entry), Group: m.Group}
137 for _, e := range markStates {
138 if e.d == m.State {
139 pb.State = e.p
140 }
141 }
142 if m.State == domain.CIConcurrencyRejected {
143 pb.State, pb.QueueFull = storagepb.CIConcurrencyState_CI_CONCURRENCY_STATE_REPLACED, true
144 }
145 return mustMarshal(pb)
146}
147
148func decodeConcurrencyMark(val []byte) (domain.CIConcurrencyMark, error) {
149 var pb storagepb.CIConcurrencyMark
150 if err := proto.Unmarshal(val, &pb); err != nil {
151 return domain.CIConcurrencyMark{}, fmt.Errorf("decode concurrency mark: %w", err)
152 }
153 m := domain.CIConcurrencyMark{Group: pb.GetGroup()}
154 if e := decodeConcurrencyEntry(pb.GetEntry()); e != nil {
155 m.Entry = *e
156 }
157 for _, e := range markStates {
158 if e.p == pb.GetState() {
159 m.State = e.d
160 }
161 }
162 if m.State == domain.CIConcurrencyReplaced && pb.GetQueueFull() {
163 m.State = domain.CIConcurrencyRejected
164 }
165 return m, nil
166}
167
168// encodeConcurrencyState is a canonical image of a group: the record, then the
169// marks in key order. Two states are the same when their images are.
170func encodeConcurrencyState(s domain.CIConcurrencyGroupState) []byte {
171 out := encodeConcurrencyRecord(s.Record)
172 for _, k := range slices.Sorted(maps.Keys(s.Marks)) {
173 out = append(out, 0)
174 out = append(out, k...)
175 out = append(out, 0)
176 out = append(out, encodeConcurrencyMark(s.Marks[k])...)
177 }
178 return out
179}
180
181// UpdateConcurrency reads a group with its marks, calls fn on it and stores the
182// result in ONE batch: the record (key "concur/..."), every mark that changed
183// ("concurmark/...", written or deleted) and, when something changed, the guard
184// of the repository. The batch requires the version the record had when it was
185// read (or its absence) whatever fn does, and every change rewrites the record,
186// so of two transactions that read the same state only one commits; the other
187// calls fn again on the new state.
188//
189//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
190func (c *CI) UpdateConcurrency(ctx context.Context, repo domain.RepoName, group string, fn func(*domain.CIConcurrencyGroupState) error) error {
191 key, mprefix := concurKey(repo, group), concurMarkPrefix(repo, group)
192 for range updateConcurrencyTries {
193 var state domain.CIConcurrencyGroupState
194 var version uint64
195 if item, ok := c.db.Get(ctx, key); ok {
196 var err error
197 if state.Record, err = decodeConcurrencyRecord(item.Value); err != nil {
198 return err
199 }
200 version = item.Version
201 }
202 // The marks are read after the record: a transaction that commits in
203 // between has changed the record, and the batch below conflicts.
204 state.Marks = map[string]domain.CIConcurrencyMark{}
205 for _, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(mprefix)) {
206 m, err := decodeConcurrencyMark(item.Value)
207 if err != nil {
208 return err
209 }
210 state.Marks[m.Entry.Key()] = m
211 }
212 // The state as read, in bytes, before fn sees it: fn may change the
213 // entries it is given in place, and a copy of the struct shares them.
214 before, stored := encodeConcurrencyState(state), maps.Clone(state.Marks)
215 if err := fn(&state); err != nil {
216 return err
217 }
218 var b kv.Batch
219 if version == 0 {
220 b.RequireAbsent(key)
221 } else {
222 b.RequireVersion(key, version)
223 }
224 // A state left as it was is not written, but the batch still holds the
225 // precondition: the decision was made on the state that was read.
226 if !bytes.Equal(before, encodeConcurrencyState(state)) {
227 for k, m := range state.Marks {
228 if old, ok := stored[k]; !ok || old != m {
229 b.Put([]byte(mprefix+k), encodeConcurrencyMark(m))
230 }
231 }
232 for k := range stored {
233 if _, ok := state.Marks[k]; !ok {
234 b.Delete([]byte(mprefix + k))
235 }
236 }
237 if state.IsEmpty() {
238 b.Delete(key)
239 } else {
240 b.Put(key, encodeConcurrencyRecord(state.Record)) // rewritten even if equal: it is the version gate
241 if err := c.guardRepo(ctx, &b, repo); err != nil {
242 return err
243 }
244 }
245 }
246 if _, err := c.db.Apply(ctx, &b); err != nil {
247 if errors.Is(err, kv.ErrConflict) {
248 continue
249 }
250 return fmt.Errorf("write concurrency group: %w", err)
251 }
252 return nil
253 }
254 return fmt.Errorf("%w: concurrency group is busy", domain.ErrConflict)
255}
256
257// ListConcurrency returns every stored group with its marks.
258func (c *CI) ListConcurrency(ctx context.Context) ([]ports.CIConcurrencyGroup, error) {
259 type id struct{ repo, group string }
260 snap := c.db.Snapshot(ctx)
261 byID := map[id]*ports.CIConcurrencyGroup{}
262 get := func(repoName, group string) *ports.CIConcurrencyGroup {
263 repo, err := domain.ParseRepoName(repoName)
264 if err != nil {
265 return nil
266 }
267 g, ok := byID[id{repoName, group}]
268 if !ok {
269 g = &ports.CIConcurrencyGroup{Repo: repo, Group: group, State: domain.CIConcurrencyGroupState{Marks: map[string]domain.CIConcurrencyMark{}}}
270 byID[id{repoName, group}] = g
271 }
272 return g
273 }
274 for key, item := range snap.Prefix(ctx, []byte(concurPrefix)) {
275 repoName, group, ok := strings.Cut(strings.TrimPrefix(string(key), concurPrefix), "/")
276 if !ok {
277 continue
278 }
279 g := get(repoName, group)
280 if g == nil {
281 continue
282 }
283 rec, err := decodeConcurrencyRecord(item.Value)
284 if err != nil {
285 return nil, err
286 }
287 g.State.Record = rec
288 }
289 for key, item := range snap.Prefix(ctx, []byte(concurMarkPrefixAll)) {
290 repoName, rest, ok := strings.Cut(strings.TrimPrefix(string(key), concurMarkPrefixAll), "/")
291 if !ok {
292 continue
293 }
294 escaped, _, ok := strings.Cut(rest, "/")
295 if !ok {
296 continue
297 }
298 group, err := url.PathUnescape(escaped)
299 if err != nil {
300 continue
301 }
302 g := get(repoName, group)
303 if g == nil {
304 continue
305 }
306 m, err := decodeConcurrencyMark(item.Value)
307 if err != nil {
308 return nil, err
309 }
310 g.State.Marks[m.Entry.Key()] = m
311 }
312 out := make([]ports.CIConcurrencyGroup, 0, len(byID))
313 for _, g := range byID {
314 out = append(out, *g)
315 }
316 return out, nil
317}