1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "slices"
10 "strconv"
11 "strings"
12 "time"
13
14 kv "bnema.dev/ermite-kv"
15 "google.golang.org/protobuf/proto"
16
17 storagepb "bnema.dev/ermite/gen/ermite/storage"
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20)
21
22// Key layout of CI. Repository names hold no '/', so a prefix ending in "/"
23// never matches another repository; ids are zero-padded so that keys sort
24// numerically. Job records are keyed by run, attempt of the run and job id.
25//
26// run/<repo>/<id:020d> CIRun
27// runsrc/<id:020d> CIRunSource
28// job/<run:020d>/<attempt:010d>/<id> CIJob
29// check/<repo>/<sha>/<check id> CICheck
30// runseq Sequence: last run id
31// runnum/<repo>/<workflow path> Sequence: last run number
32// runner/<id>, runnername/<name>, runnertoken/<hash>
33//
34// Indexes, derived from the job and run records by every write:
35//
36// ciopen/<id:020d> run not completed (value: repo)
37// ciready/<ready_at:020d>/<job key> job waiting for a runner
38// ciactive/<job key> job assigned or running
39// jobtoken/<hash> job token (value: job key)
40const (
41 runPrefix = "run/"
42 runSrcPrefix = "runsrc/"
43 jobPrefix = "job/"
44 checkPrefix = "check/"
45 runSeqKey = "runseq"
46 runNumPrefix = "runnum/"
47 runnerPrefix = "runner/"
48 runnerName = "runnername/"
49 runnerTokPref = "runnertoken/"
50 openPrefix = "ciopen/"
51 readyPrefix = "ciready/"
52 activePrefix = "ciactive/"
53 jobTokPrefix = "jobtoken/"
54 idFormat = "%020d"
55)
56
57// CI is the ermite-kv backed ports.CIStore. It shares the database of the
58// Store it comes from.
59type CI struct {
60 db *kv.DB
61}
62
63var _ ports.CIStore = (*CI)(nil)
64
65// CI returns the CI store over the same database. It is valid until the Store
66// is closed.
67func (s *Store) CI() *CI { return &CI{db: s.db} }
68
69func runKey(repo domain.RepoName, id int64) []byte {
70 return []byte(runPrefix + repo.String() + "/" + fmt.Sprintf(idFormat, id))
71}
72func runsPrefix(repo domain.RepoName) string { return runPrefix + repo.String() + "/" }
73func runSrcKey(id int64) []byte { return []byte(runSrcPrefix + fmt.Sprintf(idFormat, id)) }
74func openKey(id int64) []byte { return []byte(openPrefix + fmt.Sprintf(idFormat, id)) }
75func jobsPrefix(runID int64) string { return jobPrefix + fmt.Sprintf(idFormat, runID) + "/" }
76func jobAttemptPrefix(runID int64, attempt int32) string {
77 return jobsPrefix(runID) + fmt.Sprintf("%010d", attempt) + "/"
78}
79func jobRef(runID int64, attempt int32, id string) string {
80 return jobAttemptPrefix(runID, attempt) + id
81}
82func jobKey(runID int64, attempt int32, id string) []byte { return []byte(jobRef(runID, attempt, id)) }
83func checksPrefix(repo domain.RepoName, sha string) string {
84 return checkPrefix + repo.String() + "/" + sha + "/"
85}
86func checkKey(repo domain.RepoName, sha, id string) []byte {
87 return []byte(checksPrefix(repo, sha) + id)
88}
89func runNumKey(repo domain.RepoName, workflow string) []byte {
90 return []byte(runNumPrefix + repo.String() + "/" + workflow)
91}
92func readyKey(at time.Time, ref string) []byte {
93 return []byte(readyPrefix + fmt.Sprintf(idFormat, at.UnixNano()) + "/" + ref)
94}
95func activeKey(ref string) []byte { return []byte(activePrefix + ref) }
96func jobTokenKey(h domain.TokenHash) []byte {
97 return []byte(jobTokPrefix + h.String())
98}
99
100// --- encoding ---
101
102func encodeStatus(s domain.CIStatus) storagepb.CIStatus {
103 switch s {
104 case domain.CIQueued:
105 return storagepb.CIStatus_CI_STATUS_QUEUED
106 case domain.CIInProgress:
107 return storagepb.CIStatus_CI_STATUS_IN_PROGRESS
108 case domain.CICompleted:
109 return storagepb.CIStatus_CI_STATUS_COMPLETED
110 }
111 return storagepb.CIStatus_CI_STATUS_UNSPECIFIED
112}
113
114func decodeStatus(s storagepb.CIStatus) domain.CIStatus {
115 switch s {
116 case storagepb.CIStatus_CI_STATUS_IN_PROGRESS:
117 return domain.CIInProgress
118 case storagepb.CIStatus_CI_STATUS_COMPLETED:
119 return domain.CICompleted
120 }
121 return domain.CIQueued
122}
123
124var conclusions = []struct {
125 d domain.CIConclusion
126 p storagepb.CIConclusion
127}{
128 {domain.CISuccess, storagepb.CIConclusion_CI_CONCLUSION_SUCCESS},
129 {domain.CIFailure, storagepb.CIConclusion_CI_CONCLUSION_FAILURE},
130 {domain.CICancelled, storagepb.CIConclusion_CI_CONCLUSION_CANCELLED},
131 {domain.CISkipped, storagepb.CIConclusion_CI_CONCLUSION_SKIPPED},
132 {domain.CITimedOut, storagepb.CIConclusion_CI_CONCLUSION_TIMED_OUT},
133 {domain.CINeutral, storagepb.CIConclusion_CI_CONCLUSION_NEUTRAL},
134 {domain.CIActionRequired, storagepb.CIConclusion_CI_CONCLUSION_ACTION_REQUIRED},
135}
136
137func encodeConclusion(c domain.CIConclusion) storagepb.CIConclusion {
138 for _, e := range conclusions {
139 if e.d == c {
140 return e.p
141 }
142 }
143 return storagepb.CIConclusion_CI_CONCLUSION_UNSPECIFIED
144}
145
146func decodeConclusion(c storagepb.CIConclusion) domain.CIConclusion {
147 for _, e := range conclusions {
148 if e.p == c {
149 return e.d
150 }
151 }
152 return domain.CINoConclusion
153}
154
155var phases = []struct {
156 d domain.CIJobPhase
157 p storagepb.CIJobPhase
158}{
159 {domain.CIPhaseWaiting, storagepb.CIJobPhase_CI_JOB_PHASE_WAITING},
160 {domain.CIPhaseReady, storagepb.CIJobPhase_CI_JOB_PHASE_READY},
161 {domain.CIPhaseAssigned, storagepb.CIJobPhase_CI_JOB_PHASE_ASSIGNED},
162 {domain.CIPhaseRunning, storagepb.CIJobPhase_CI_JOB_PHASE_RUNNING},
163 {domain.CIPhaseEnded, storagepb.CIJobPhase_CI_JOB_PHASE_ENDED},
164}
165
166func encodePhase(p domain.CIJobPhase) (storagepb.CIJobPhase, error) {
167 for _, e := range phases {
168 if e.d == p {
169 return e.p, nil
170 }
171 }
172 return 0, fmt.Errorf("invalid job phase %q", p)
173}
174
175func decodePhase(p storagepb.CIJobPhase) domain.CIJobPhase {
176 for _, e := range phases {
177 if e.p == p {
178 return e.d
179 }
180 }
181 return domain.CIPhaseWaiting
182}
183
184var isolations = []struct {
185 d domain.CIIsolation
186 p storagepb.CIIsolation
187}{
188 {domain.CIIsolationVM, storagepb.CIIsolation_CI_ISOLATION_VM},
189 {domain.CIIsolationContainerSysbox, storagepb.CIIsolation_CI_ISOLATION_CONTAINER_SYSBOX},
190 {domain.CIIsolationContainerPrivileged, storagepb.CIIsolation_CI_ISOLATION_CONTAINER_PRIVILEGED},
191}
192
193func encodeIsolation(i domain.CIIsolation) storagepb.CIIsolation {
194 for _, e := range isolations {
195 if e.d == i {
196 return e.p
197 }
198 }
199 return storagepb.CIIsolation_CI_ISOLATION_UNSPECIFIED
200}
201
202func decodeIsolation(i storagepb.CIIsolation) domain.CIIsolation {
203 for _, e := range isolations {
204 if e.p == i {
205 return e.d
206 }
207 }
208 return domain.CIIsolationNone
209}
210
211var stepStates = []struct {
212 d domain.CIStepState
213 p storagepb.CIStepState
214}{
215 {domain.CIStepPending, storagepb.CIStepState_CI_STEP_STATE_PENDING},
216 {domain.CIStepInProgress, storagepb.CIStepState_CI_STEP_STATE_IN_PROGRESS},
217 {domain.CIStepCompleted, storagepb.CIStepState_CI_STEP_STATE_COMPLETED},
218}
219
220func encodeStepState(s domain.CIStepState) storagepb.CIStepState {
221 for _, e := range stepStates {
222 if e.d == s {
223 return e.p
224 }
225 }
226 return storagepb.CIStepState_CI_STEP_STATE_UNSPECIFIED
227}
228
229func decodeStepState(s storagepb.CIStepState) domain.CIStepState {
230 for _, e := range stepStates {
231 if e.p == s {
232 return e.d
233 }
234 }
235 return domain.CIStepPending
236}
237
238var cancelReasons = []struct {
239 d domain.CICancelReason
240 p storagepb.CICancelReason
241}{
242 {domain.CICancelUser, storagepb.CICancelReason_CI_CANCEL_REASON_USER},
243 {domain.CICancelFailFast, storagepb.CICancelReason_CI_CANCEL_REASON_FAIL_FAST},
244 {domain.CICancelTimeout, storagepb.CICancelReason_CI_CANCEL_REASON_TIMEOUT},
245}
246
247func encodeCancelReason(r domain.CICancelReason) storagepb.CICancelReason {
248 for _, e := range cancelReasons {
249 if e.d == r {
250 return e.p
251 }
252 }
253 return storagepb.CICancelReason_CI_CANCEL_REASON_UNSPECIFIED
254}
255
256func decodeCancelReason(r storagepb.CICancelReason) domain.CICancelReason {
257 for _, e := range cancelReasons {
258 if e.p == r {
259 return e.d
260 }
261 }
262 return domain.CICancelNone
263}
264
265func encodeRun(r domain.CIRun) ([]byte, error) {
266 if r.Repo.IsZero() {
267 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
268 }
269 return mustMarshal(&storagepb.CIRun{
270 Repo: r.Repo.String(), Id: r.ID, Number: r.Number, Attempt: r.Attempt,
271 WorkflowPath: r.WorkflowPath, WorkflowName: r.WorkflowName, Name: r.Name, Title: r.Title,
272 Event: r.Event, Ref: r.Ref, Sha: r.Sha, Actor: r.Actor,
273 Status: encodeStatus(r.Status), Conclusion: encodeConclusion(r.Conclusion), Error: r.Error,
274 CreatedAt: encodeTime(r.CreatedAt), StartedAt: encodeTime(r.StartedAt), CompletedAt: encodeTime(r.CompletedAt),
275 CancelRequested: r.CancelRequested, CancelRequestedAt: encodeTime(r.CancelRequestedAt),
276 }), nil
277}
278
279func decodeRun(item kv.Item) (domain.CIRun, error) {
280 var pb storagepb.CIRun
281 if err := proto.Unmarshal(item.Value, &pb); err != nil {
282 return domain.CIRun{}, fmt.Errorf("decode run: %w", err)
283 }
284 repo, err := domain.ParseRepoName(pb.GetRepo())
285 if err != nil {
286 return domain.CIRun{}, fmt.Errorf("decode run: %w", err)
287 }
288 name := pb.GetName()
289 if name == "" {
290 name = pb.GetWorkflowName() // stored before the field existed
291 }
292 return domain.CIRun{
293 Repo: repo, ID: pb.GetId(), Number: pb.GetNumber(), Attempt: pb.GetAttempt(),
294 WorkflowPath: pb.GetWorkflowPath(), WorkflowName: pb.GetWorkflowName(), Name: name, Title: pb.GetTitle(),
295 Event: pb.GetEvent(), Ref: pb.GetRef(), Sha: pb.GetSha(), Actor: pb.GetActor(),
296 Status: decodeStatus(pb.GetStatus()), Conclusion: decodeConclusion(pb.GetConclusion()), Error: pb.GetError(),
297 CreatedAt: decodeTime(pb.GetCreatedAt()), StartedAt: decodeTime(pb.GetStartedAt()), CompletedAt: decodeTime(pb.GetCompletedAt()),
298 CancelRequested: pb.GetCancelRequested(), CancelRequestedAt: decodeTime(pb.GetCancelRequestedAt()),
299 Version: item.Version,
300 }, nil
301}
302
303func encodeJob(j domain.CIJob) ([]byte, error) {
304 if j.Repo.IsZero() {
305 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
306 }
307 phase, err := encodePhase(j.Phase)
308 if err != nil {
309 return nil, err
310 }
311 steps := make([]*storagepb.CIStep, 0, len(j.Steps))
312 for _, s := range j.Steps {
313 steps = append(steps, &storagepb.CIStep{
314 Number: s.Number, Name: s.Name, State: encodeStepState(s.State), Conclusion: encodeConclusion(s.Conclusion),
315 StartedAt: encodeTime(s.StartedAt), CompletedAt: encodeTime(s.CompletedAt),
316 LogStart: s.LogStart, HasLines: s.HasLines,
317 })
318 }
319 pb := &storagepb.CIJob{
320 Id: j.ID, RunId: j.RunID, Attempt: j.Attempt, RunAttempt: j.RunAttempt, Repo: j.Repo.String(),
321 Key: j.Key, Name: j.Name, Phase: phase, Conclusion: encodeConclusion(j.Conclusion), Error: j.Error,
322 Labels: j.Labels, Needs: j.Needs, RunnerId: j.RunnerID, RunnerName: j.RunnerName,
323 Isolation: encodeIsolation(j.Isolation), Steps: steps,
324 CreatedAt: encodeTime(j.CreatedAt), StartedAt: encodeTime(j.StartedAt), CompletedAt: encodeTime(j.CompletedAt),
325 Index: int32(j.Index), Combo: int32(j.Combo), MatrixJson: j.MatrixJSON, ContextsJson: j.ContextsJSON, ComboTotal: int32(j.ComboTotal), //nolint:gosec // bounded by the 256 job limit
326 TimeoutMinutes: j.TimeoutMinutes, FailFast: j.FailFast, ContinueOnError: j.ContinueOnError, MaxParallel: int32(j.MaxParallel), //nolint:gosec // validated at planning
327 Outputs: j.Outputs, ReadyAt: encodeTime(j.ReadyAt), AssignedAt: encodeTime(j.AssignedAt),
328 AckedAt: encodeTime(j.AckedAt), AssignCancels: int32(j.AssignCancels), //nolint:gosec // at most 3
329 TokenValidUntil: encodeTime(j.TokenValidUntil), CancelReason: encodeCancelReason(j.CancelReason), CancelMessage: j.CancelMessage,
330 CancelRequestedAt: encodeTime(j.CancelRequestedAt), Planned: j.Planned,
331 }
332 if j.HasToken() {
333 pb.TokenHash = j.TokenHash[:]
334 }
335 return mustMarshal(pb), nil
336}
337
338//hexcheck:ignore suspicious-business-logic-in-adapter maps the stored record to the domain job field by field: no rule is decided here
339func decodeJob(item kv.Item) (domain.CIJob, error) {
340 var pb storagepb.CIJob
341 if err := proto.Unmarshal(item.Value, &pb); err != nil {
342 return domain.CIJob{}, fmt.Errorf("decode job: %w", err)
343 }
344 repo, err := domain.ParseRepoName(pb.GetRepo())
345 if err != nil {
346 return domain.CIJob{}, fmt.Errorf("decode job: %w", err)
347 }
348 j := domain.CIJob{
349 ID: pb.GetId(), RunID: pb.GetRunId(), Attempt: pb.GetAttempt(), RunAttempt: pb.GetRunAttempt(), Repo: repo,
350 Key: pb.GetKey(), Name: pb.GetName(), Phase: decodePhase(pb.GetPhase()), Conclusion: decodeConclusion(pb.GetConclusion()),
351 Error: pb.GetError(), Labels: pb.GetLabels(), Needs: pb.GetNeeds(), RunnerID: pb.GetRunnerId(), RunnerName: pb.GetRunnerName(),
352 Isolation: decodeIsolation(pb.GetIsolation()),
353 CreatedAt: decodeTime(pb.GetCreatedAt()), StartedAt: decodeTime(pb.GetStartedAt()), CompletedAt: decodeTime(pb.GetCompletedAt()),
354 Index: int(pb.GetIndex()), Combo: int(pb.GetCombo()), Planned: pb.GetPlanned(), MatrixJSON: pb.GetMatrixJson(), ContextsJSON: pb.GetContextsJson(),
355 ComboTotal: int(pb.GetComboTotal()), TimeoutMinutes: pb.GetTimeoutMinutes(), FailFast: pb.GetFailFast(), ContinueOnError: pb.GetContinueOnError(),
356 MaxParallel: int(pb.GetMaxParallel()), Outputs: pb.GetOutputs(),
357 TokenValidUntil: decodeTime(pb.GetTokenValidUntil()), ReadyAt: decodeTime(pb.GetReadyAt()),
358 AssignedAt: decodeTime(pb.GetAssignedAt()), AckedAt: decodeTime(pb.GetAckedAt()), AssignCancels: int(pb.GetAssignCancels()),
359 CancelMessage: pb.GetCancelMessage(), CancelRequestedAt: decodeTime(pb.GetCancelRequestedAt()),
360 Version: item.Version,
361 }
362 j.CancelReason = decodeCancelReason(pb.GetCancelReason())
363 copy(j.TokenHash[:], pb.GetTokenHash())
364 for _, s := range pb.GetSteps() {
365 j.Steps = append(j.Steps, domain.CIStep{
366 Number: s.GetNumber(), Name: s.GetName(), State: decodeStepState(s.GetState()), Conclusion: decodeConclusion(s.GetConclusion()),
367 StartedAt: decodeTime(s.GetStartedAt()), CompletedAt: decodeTime(s.GetCompletedAt()),
368 LogStart: s.GetLogStart(), HasLines: s.GetHasLines(),
369 })
370 }
371 return j, nil
372}
373
374func encodeCheck(c domain.CICheck) []byte {
375 return mustMarshal(&storagepb.CICheck{
376 Id: c.ID, Repo: c.Repo.String(), Sha: c.Sha, Name: c.Name, Status: encodeStatus(c.Status),
377 Conclusion: encodeConclusion(c.Conclusion), RunId: c.RunID, JobId: c.JobID, Attempt: c.Attempt,
378 WorkflowName: c.WorkflowName, CreatedAt: encodeTime(c.CreatedAt), StartedAt: encodeTime(c.StartedAt),
379 CompletedAt: encodeTime(c.CompletedAt),
380 })
381}
382
383func decodeCheck(item kv.Item) (domain.CICheck, error) {
384 var pb storagepb.CICheck
385 if err := proto.Unmarshal(item.Value, &pb); err != nil {
386 return domain.CICheck{}, fmt.Errorf("decode check: %w", err)
387 }
388 repo, err := domain.ParseRepoName(pb.GetRepo())
389 if err != nil {
390 return domain.CICheck{}, fmt.Errorf("decode check: %w", err)
391 }
392 return domain.CICheck{
393 ID: pb.GetId(), Repo: repo, Sha: pb.GetSha(), Name: pb.GetName(), Status: decodeStatus(pb.GetStatus()),
394 Conclusion: decodeConclusion(pb.GetConclusion()), RunID: pb.GetRunId(), JobID: pb.GetJobId(), Attempt: pb.GetAttempt(),
395 WorkflowName: pb.GetWorkflowName(), CreatedAt: decodeTime(pb.GetCreatedAt()), StartedAt: decodeTime(pb.GetStartedAt()),
396 CompletedAt: decodeTime(pb.GetCompletedAt()), Version: item.Version,
397 }, nil
398}
399
400func encodeRunner(r domain.CIRunner) []byte {
401 return mustMarshal(&storagepb.CIRunner{
402 Id: r.ID, Name: r.Name, Labels: r.Labels, Isolation: encodeIsolation(r.Isolation), Version: r.Version,
403 Arch: r.Arch, Os: r.OS, RegisteredAt: encodeTime(r.RegisteredAt), TokenHash: r.TokenHash[:],
404 })
405}
406
407func decodeRunner(val []byte) (domain.CIRunner, error) {
408 var pb storagepb.CIRunner
409 if err := proto.Unmarshal(val, &pb); err != nil {
410 return domain.CIRunner{}, fmt.Errorf("decode runner: %w", err)
411 }
412 r := domain.CIRunner{
413 ID: pb.GetId(), Name: pb.GetName(), Labels: pb.GetLabels(), Isolation: decodeIsolation(pb.GetIsolation()),
414 Version: pb.GetVersion(), Arch: pb.GetArch(), OS: pb.GetOs(), RegisteredAt: decodeTime(pb.GetRegisteredAt()),
415 }
416 copy(r.TokenHash[:], pb.GetTokenHash())
417 return r, nil
418}
419
420// --- writes ---
421
422// stage adds the preconditions and writes of one record to the batch.
423func stage(b *kv.Batch, key, val []byte, version uint64) {
424 if version == 0 {
425 b.RequireAbsent(key)
426 } else {
427 b.RequireVersion(key, version)
428 }
429 b.Put(key, val)
430}
431
432// runIndexes stages the open-run index of the run.
433func runIndexes(b *kv.Batch, r domain.CIRun) {
434 if r.Status == domain.CICompleted {
435 b.Delete(openKey(r.ID))
436 return
437 }
438 b.Put(openKey(r.ID), []byte(r.Repo.String()))
439}
440
441// jobIndexes stages the ready, active and token indexes of the job. old is
442// the job as stored before, if any: its token index goes when the token
443// changed.
444func (c *CI) jobIndexes(b *kv.Batch, j domain.CIJob, old *domain.CIJob) {
445 ref := jobRef(j.RunID, j.RunAttempt, j.ID)
446 if j.Phase == domain.CIPhaseReady {
447 b.Put(readyKey(j.ReadyAt, ref), nil)
448 } else {
449 b.Delete(readyKey(j.ReadyAt, ref))
450 }
451 if old != nil && old.Phase == domain.CIPhaseReady && (j.Phase != domain.CIPhaseReady || !old.ReadyAt.Equal(j.ReadyAt)) {
452 b.Delete(readyKey(old.ReadyAt, ref))
453 }
454 if j.Active() {
455 b.Put(activeKey(ref), nil)
456 } else {
457 b.Delete(activeKey(ref))
458 }
459 if old != nil && old.HasToken() && old.TokenHash != j.TokenHash {
460 b.Delete(jobTokenKey(old.TokenHash))
461 }
462 if j.HasToken() {
463 b.Put(jobTokenKey(j.TokenHash), []byte(ref))
464 }
465}
466
467// guardRepo is guardRepoKey on the database of the CI store, for the writes that
468// create keys under a repository.
469func (c *CI) guardRepo(ctx context.Context, b *kv.Batch, repo domain.RepoName) error {
470 return guardRepoKey(ctx, c.db, b, repo)
471}
472
473// stageRecords stages the writes of w. guarded collects the repositories that
474// need guarding.
475func (c *CI) stageRecords(ctx context.Context, b *kv.Batch, w ports.CIWrite) (guarded map[domain.RepoName]bool, err error) {
476 guarded = map[domain.RepoName]bool{}
477 for _, r := range w.Runs {
478 val, err := encodeRun(r)
479 if err != nil {
480 return nil, err
481 }
482 stage(b, runKey(r.Repo, r.ID), val, r.Version)
483 runIndexes(b, r)
484 if r.Version == 0 {
485 guarded[r.Repo] = true
486 }
487 }
488 for _, j := range w.Jobs {
489 val, err := encodeJob(j)
490 if err != nil {
491 return nil, err
492 }
493 key := jobKey(j.RunID, j.RunAttempt, j.ID)
494 var old *domain.CIJob
495 if j.Version != 0 {
496 item, ok := c.db.Get(ctx, key)
497 if !ok {
498 return nil, domain.ErrConflict
499 }
500 if o, err := decodeJob(item); err == nil {
501 old = &o
502 }
503 }
504 stage(b, key, val, j.Version)
505 c.jobIndexes(b, j, old)
506 if j.Version == 0 {
507 guarded[j.Repo] = true
508 }
509 }
510 // Checks are derived from the jobs, written in the same batch and under
511 // the lock of their run: they are upserted without a precondition.
512 for _, ck := range w.Checks {
513 b.Put(checkKey(ck.Repo, ck.Sha, ck.ID), encodeCheck(ck))
514 }
515 return guarded, nil
516}
517
518// Apply writes the records in one batch.
519//
520//hexcheck:ignore suspicious-business-logic-in-adapter stages records and their derived indexes under version preconditions: a storage concern, no policy
521func (c *CI) Apply(ctx context.Context, w ports.CIWrite) (uint64, error) {
522 var b kv.Batch
523 guarded, err := c.stageRecords(ctx, &b, w)
524 if err != nil {
525 return 0, err
526 }
527 for repo := range guarded {
528 if err := c.guardRepo(ctx, &b, repo); err != nil {
529 return 0, err
530 }
531 }
532 v, err := c.db.Apply(ctx, &b)
533 if errors.Is(err, kv.ErrConflict) {
534 return 0, domain.ErrConflict
535 }
536 if err != nil {
537 return 0, fmt.Errorf("write CI records: %w", err)
538 }
539 return v, nil
540}
541
542func (c *CI) readSeq(ctx context.Context, key []byte) (last int64, item kv.Item, ok bool, err error) {
543 item, ok = c.db.Get(ctx, key)
544 if !ok {
545 return 0, kv.Item{}, false, nil
546 }
547 var seq storagepb.Sequence
548 if err := proto.Unmarshal(item.Value, &seq); err != nil {
549 return 0, kv.Item{}, false, fmt.Errorf("decode sequence %s: %w", key, err)
550 }
551 return seq.GetLast(), item, true, nil
552}
553
554func stageSeq(b *kv.Batch, key []byte, item kv.Item, ok bool, last int64) {
555 if ok {
556 b.RequireVersion(key, item.Version)
557 } else {
558 b.RequireAbsent(key)
559 }
560 b.Put(key, mustMarshal(&storagepb.Sequence{Last: last}))
561}
562
563const createRunTries = 20
564
565// CreateRun takes the next run id and number and stores the draft.
566//
567//hexcheck:ignore suspicious-business-logic-in-adapter takes the next numbers under version preconditions: a storage concern, no policy
568func (c *CI) CreateRun(ctx context.Context, repo domain.RepoName, workflowPath string, build func(id, number int64) ports.CIRunDraft) (ports.CIRunDraft, error) {
569 for range createRunTries {
570 lastID, idItem, idOK, err := c.readSeq(ctx, []byte(runSeqKey))
571 if err != nil {
572 return ports.CIRunDraft{}, err
573 }
574 numKey := runNumKey(repo, workflowPath)
575 lastNum, numItem, numOK, err := c.readSeq(ctx, numKey)
576 if err != nil {
577 return ports.CIRunDraft{}, err
578 }
579 draft := build(lastID+1, lastNum+1)
580 var b kv.Batch
581 stageSeq(&b, []byte(runSeqKey), idItem, idOK, lastID+1)
582 stageSeq(&b, numKey, numItem, numOK, lastNum+1)
583 w := ports.CIWrite{Runs: []domain.CIRun{draft.Run}, Jobs: draft.Jobs, Checks: draft.Checks}
584 guarded, err := c.stageRecords(ctx, &b, w)
585 if err != nil {
586 return ports.CIRunDraft{}, err
587 }
588 src := mustMarshal(&storagepb.CIRunSource{
589 RunId: draft.Run.ID, Repo: repo.String(), WorkflowYaml: draft.Source.WorkflowYAML, GithubJson: draft.Source.GithubJSON,
590 })
591 b.RequireAbsent(runSrcKey(draft.Run.ID))
592 b.Put(runSrcKey(draft.Run.ID), src)
593 for r := range guarded {
594 if err := c.guardRepo(ctx, &b, r); err != nil {
595 return ports.CIRunDraft{}, err
596 }
597 }
598 v, err := c.db.Apply(ctx, &b)
599 if errors.Is(err, kv.ErrConflict) {
600 continue
601 }
602 if err != nil {
603 return ports.CIRunDraft{}, fmt.Errorf("create run of %q: %w", repo, err)
604 }
605 draft.Run.Version = v
606 for i := range draft.Jobs {
607 draft.Jobs[i].Version = v
608 }
609 for i := range draft.Checks {
610 draft.Checks[i].Version = v
611 }
612 return draft, nil
613 }
614 return ports.CIRunDraft{}, fmt.Errorf("create run of %q: %w", repo, domain.ErrConflict)
615}
616
617// --- reads ---
618
619// GetRun reads run/<repo>/<id>.
620func (c *CI) GetRun(ctx context.Context, repo domain.RepoName, id int64) (domain.CIRun, error) {
621 item, ok := c.db.Get(ctx, runKey(repo, id))
622 if !ok || id < 1 {
623 return domain.CIRun{}, domain.ErrNotFound
624 }
625 return decodeRun(item)
626}
627
628// GetRunSource reads runsrc/<id>.
629func (c *CI) GetRunSource(ctx context.Context, runID int64) (domain.CIRunSource, error) {
630 item, ok := c.db.Get(ctx, runSrcKey(runID))
631 if !ok || runID < 1 {
632 return domain.CIRunSource{}, domain.ErrNotFound
633 }
634 var pb storagepb.CIRunSource
635 if err := proto.Unmarshal(item.Value, &pb); err != nil {
636 return domain.CIRunSource{}, fmt.Errorf("decode run source: %w", err)
637 }
638 repo, err := domain.ParseRepoName(pb.GetRepo())
639 if err != nil {
640 return domain.CIRunSource{}, fmt.Errorf("decode run source: %w", err)
641 }
642 return domain.CIRunSource{RunID: pb.GetRunId(), Repo: repo, WorkflowYAML: pb.GetWorkflowYaml(), GithubJSON: pb.GetGithubJson()}, nil
643}
644
645// ListRuns returns the runs of the repository, newest first.
646//
647//hexcheck:ignore suspicious-business-logic-in-adapter port method name only: reads a key range and reverses it
648func (c *CI) ListRuns(ctx context.Context, repo domain.RepoName, before int64, limit int) ([]domain.CIRun, error) {
649 start := []byte(runsPrefix(repo))
650 end := []byte(runPrefix + repo.String() + "0") // '/' + 1
651 if before > 0 {
652 end = runKey(repo, before)
653 }
654 // Collect then reverse: the snapshot iterates in key order only.
655 var all []domain.CIRun
656 for key, item := range c.db.Snapshot(ctx).Range(ctx, start, end) {
657 r, err := decodeRun(item)
658 if err != nil {
659 return nil, fmt.Errorf("decode %s: %w", key, err)
660 }
661 all = append(all, r)
662 }
663 slices.Reverse(all)
664 if limit > 0 && len(all) > limit {
665 all = all[:limit]
666 }
667 return all, nil
668}
669
670// RunIDs returns the ids of the runs of the repository.
671func (c *CI) RunIDs(ctx context.Context, repo domain.RepoName) ([]int64, error) {
672 var ids []int64
673 prefix := runsPrefix(repo)
674 for key := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
675 id, err := strconv.ParseInt(strings.TrimPrefix(string(key), prefix), 10, 64)
676 if err != nil {
677 return nil, fmt.Errorf("decode %s: %w", key, err)
678 }
679 ids = append(ids, id)
680 }
681 return ids, nil
682}
683
684// ListOpenRuns returns the runs that have not completed.
685func (c *CI) ListOpenRuns(ctx context.Context) ([]domain.CIRun, error) {
686 snap := c.db.Snapshot(ctx)
687 var out []domain.CIRun
688 for key, item := range snap.Prefix(ctx, []byte(openPrefix)) {
689 id, err := strconv.ParseInt(strings.TrimPrefix(string(key), openPrefix), 10, 64)
690 if err != nil {
691 return nil, fmt.Errorf("decode %s: %w", key, err)
692 }
693 repo, err := domain.ParseRepoName(string(item.Value))
694 if err != nil {
695 return nil, err
696 }
697 it, ok := snap.Get(ctx, runKey(repo, id))
698 if !ok {
699 continue
700 }
701 r, err := decodeRun(it)
702 if err != nil {
703 return nil, err
704 }
705 out = append(out, r)
706 }
707 return out, nil
708}
709
710// GetJob reads one job record.
711func (c *CI) GetJob(ctx context.Context, runID int64, runAttempt int32, jobID string) (domain.CIJob, error) {
712 item, ok := c.db.Get(ctx, jobKey(runID, runAttempt, jobID))
713 if !ok {
714 return domain.CIJob{}, domain.ErrNotFound
715 }
716 return decodeJob(item)
717}
718
719// ListJobs returns the jobs of a run at an attempt, ordered by index then
720// combination.
721func (c *CI) ListJobs(ctx context.Context, runID int64, runAttempt int32) ([]domain.CIJob, error) {
722 var out []domain.CIJob
723 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(jobAttemptPrefix(runID, runAttempt))) {
724 j, err := decodeJob(item)
725 if err != nil {
726 return nil, fmt.Errorf("decode %s: %w", key, err)
727 }
728 out = append(out, j)
729 }
730 slices.SortFunc(out, func(a, b domain.CIJob) int {
731 if a.Index != b.Index {
732 return a.Index - b.Index
733 }
734 return a.Combo - b.Combo
735 })
736 return out, nil
737}
738
739func (c *CI) jobByRef(snap *kv.Snapshot, ctx context.Context, ref string) (domain.CIJob, error) {
740 item, ok := snap.Get(ctx, []byte(ref))
741 if !ok {
742 return domain.CIJob{}, domain.ErrNotFound
743 }
744 return decodeJob(item)
745}
746
747// FindJobByToken resolves the token index, then reads the job.
748func (c *CI) FindJobByToken(ctx context.Context, hash domain.TokenHash, now time.Time) (domain.CIJob, error) {
749 snap := c.db.Snapshot(ctx)
750 idx, ok := snap.Get(ctx, jobTokenKey(hash))
751 if !ok {
752 return domain.CIJob{}, domain.ErrNotFound
753 }
754 j, err := c.jobByRef(snap, ctx, string(idx.Value))
755 if err != nil {
756 return domain.CIJob{}, err
757 }
758 if j.TokenHash != hash || (!j.TokenValidUntil.IsZero() && !now.Before(j.TokenValidUntil)) {
759 return domain.CIJob{}, domain.ErrNotFound
760 }
761 return j, nil
762}
763
764// ListReadyJobs returns the jobs waiting for a runner, oldest first.
765func (c *CI) ListReadyJobs(ctx context.Context, limit int) ([]domain.CIJob, error) {
766 snap := c.db.Snapshot(ctx)
767 var out []domain.CIJob
768 for key := range snap.Prefix(ctx, []byte(readyPrefix)) {
769 ref := string(key)[len(readyPrefix)+len(fmt.Sprintf(idFormat, 0))+1:]
770 j, err := c.jobByRef(snap, ctx, ref)
771 if errors.Is(err, domain.ErrNotFound) {
772 continue
773 }
774 if err != nil {
775 return nil, err
776 }
777 out = append(out, j)
778 if limit > 0 && len(out) >= limit {
779 break
780 }
781 }
782 return out, nil
783}
784
785// ListActiveJobs returns the jobs given to a runner and not over.
786func (c *CI) ListActiveJobs(ctx context.Context) ([]domain.CIJob, error) {
787 snap := c.db.Snapshot(ctx)
788 var out []domain.CIJob
789 for key := range snap.Prefix(ctx, []byte(activePrefix)) {
790 j, err := c.jobByRef(snap, ctx, strings.TrimPrefix(string(key), activePrefix))
791 if errors.Is(err, domain.ErrNotFound) {
792 continue
793 }
794 if err != nil {
795 return nil, err
796 }
797 out = append(out, j)
798 }
799 return out, nil
800}
801
802// ListChecks returns the checks of a commit ordered by creation time then
803// name.
804func (c *CI) ListChecks(ctx context.Context, repo domain.RepoName, sha string, limit int) ([]domain.CICheck, bool, error) {
805 var out []domain.CICheck
806 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(checksPrefix(repo, sha))) {
807 ck, err := decodeCheck(item)
808 if err != nil {
809 return nil, false, fmt.Errorf("decode %s: %w", key, err)
810 }
811 out = append(out, ck)
812 }
813 slices.SortStableFunc(out, func(a, b domain.CICheck) int {
814 if d := a.CreatedAt.Compare(b.CreatedAt); d != 0 {
815 return d
816 }
817 return strings.Compare(a.Name, b.Name)
818 })
819 more := limit > 0 && len(out) > limit
820 if more {
821 out = out[:limit]
822 }
823 return out, more, nil
824}
825
826// --- retention and deletion ---
827
828// runKeys lists every key of a run except its record: source, jobs, indexes.
829func (c *CI) runKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName, runID int64) [][]byte {
830 keys := [][]byte{runKey(repo, runID), runSrcKey(runID), openKey(runID)}
831 for key, item := range snap.Prefix(ctx, []byte(jobsPrefix(runID))) {
832 keys = append(keys, slices.Clone(key))
833 if j, err := decodeJob(item); err == nil {
834 ref := jobRef(j.RunID, j.RunAttempt, j.ID)
835 keys = append(keys, readyKey(j.ReadyAt, ref), activeKey(ref))
836 if j.HasToken() {
837 keys = append(keys, jobTokenKey(j.TokenHash))
838 }
839 }
840 }
841 return keys
842}
843
844// ciKeys lists every key of the snapshot that belongs to the repository's CI
845// runs, checks and numbers.
846func ciKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
847 c := &CI{}
848 var keys [][]byte
849 for key, item := range snap.Prefix(ctx, []byte(runsPrefix(repo))) {
850 if r, err := decodeRun(item); err == nil {
851 keys = append(keys, c.runKeys(ctx, snap, repo, r.ID)...)
852 }
853 keys = append(keys, slices.Clone(key))
854 }
855 for _, p := range []string{checkPrefix + repo.String() + "/", runNumPrefix + repo.String() + "/"} {
856 for k := range snap.Prefix(ctx, []byte(p)) {
857 keys = append(keys, slices.Clone(k))
858 }
859 }
860 return keys
861}
862
863const sweepChunk = 500
864
865// Sweep removes the runs completed before the time.
866//
867//hexcheck:ignore suspicious-business-logic-in-adapter chunked delete of expired records: a storage concern, the retention is decided by the caller
868func (c *CI) Sweep(ctx context.Context, before time.Time) ([]int64, error) {
869 snap := c.db.Snapshot(ctx)
870 var ids []int64
871 var keys [][]byte
872 for _, rp := range []string{runPrefix} {
873 for _, item := range snap.Prefix(ctx, []byte(rp)) {
874 r, err := decodeRun(item)
875 if err != nil {
876 return nil, err
877 }
878 if r.Status != domain.CICompleted || r.CompletedAt.IsZero() || !r.CompletedAt.Before(before) {
879 continue
880 }
881 ids = append(ids, r.ID)
882 keys = append(keys, c.runKeys(ctx, snap, r.Repo, r.ID)...)
883 for ck := range snap.Prefix(ctx, []byte(checkPrefix+r.Repo.String()+"/"+r.Sha+"/")) {
884 // A check belongs to the run only while it is the latest one
885 // for its job: leave it when another run replaced it.
886 if it, ok := snap.Get(ctx, ck); ok {
887 if chk, err := decodeCheck(it); err == nil && chk.RunID == r.ID {
888 keys = append(keys, slices.Clone(ck))
889 }
890 }
891 }
892 }
893 }
894 for len(keys) > 0 {
895 n := min(sweepChunk, len(keys))
896 var b kv.Batch
897 for _, k := range keys[:n] {
898 b.Delete(k)
899 }
900 if _, err := c.db.Apply(ctx, &b); err != nil {
901 return nil, fmt.Errorf("sweep CI runs: %w", err)
902 }
903 keys = keys[n:]
904 }
905 return ids, nil
906}
907
908// --- runners ---
909
910func runnerKey(id string) []byte { return []byte(runnerPrefix + id) }
911func runnerNameKey(n string) []byte { return []byte(runnerName + strings.ToLower(n)) }
912func runnerTokenKey(h domain.TokenHash) []byte { return []byte(runnerTokPref + h.String()) }
913
914// CreateRunner stores the runner, its name index and its token index.
915//
916//hexcheck:ignore suspicious-business-logic-in-adapter replaces a runner of the same name in one batch: a storage concern, the rule is decided by the caller
917func (c *CI) CreateRunner(ctx context.Context, r domain.CIRunner, replace bool) ([]domain.CIRunner, error) {
918 for range createRunTries {
919 var b kv.Batch
920 var removed []domain.CIRunner
921 if idx, ok := c.db.Get(ctx, runnerNameKey(r.Name)); ok {
922 if !replace {
923 return nil, domain.ErrAlreadyExists
924 }
925 old, err := c.GetRunner(ctx, string(idx.Value))
926 if err != nil && !errors.Is(err, domain.ErrNotFound) {
927 return nil, err
928 }
929 b.RequireVersion(runnerNameKey(r.Name), idx.Version)
930 if err == nil {
931 removed = append(removed, old)
932 b.Delete(runnerKey(old.ID))
933 b.Delete(runnerTokenKey(old.TokenHash))
934 }
935 } else {
936 b.RequireAbsent(runnerNameKey(r.Name))
937 }
938 b.RequireAbsent(runnerKey(r.ID))
939 b.RequireAbsent(runnerTokenKey(r.TokenHash))
940 b.Put(runnerKey(r.ID), encodeRunner(r))
941 b.Put(runnerNameKey(r.Name), []byte(r.ID))
942 b.Put(runnerTokenKey(r.TokenHash), []byte(r.ID))
943 _, err := c.db.Apply(ctx, &b)
944 if errors.Is(err, kv.ErrConflict) {
945 if !replace {
946 if _, ok := c.db.Get(ctx, runnerNameKey(r.Name)); ok {
947 return nil, domain.ErrAlreadyExists
948 }
949 }
950 continue
951 }
952 if err != nil {
953 return nil, fmt.Errorf("create runner %q: %w", r.Name, err)
954 }
955 return removed, nil
956 }
957 return nil, fmt.Errorf("create runner %q: %w", r.Name, domain.ErrConflict)
958}
959
960// GetRunner reads runner/<id>.
961func (c *CI) GetRunner(ctx context.Context, id string) (domain.CIRunner, error) {
962 item, ok := c.db.Get(ctx, runnerKey(id))
963 if !ok || id == "" || strings.Contains(id, "/") {
964 return domain.CIRunner{}, domain.ErrNotFound
965 }
966 return decodeRunner(item.Value)
967}
968
969// FindRunnerByToken resolves the token index, then reads the runner.
970func (c *CI) FindRunnerByToken(ctx context.Context, hash domain.TokenHash) (domain.CIRunner, error) {
971 snap := c.db.Snapshot(ctx)
972 idx, ok := snap.Get(ctx, runnerTokenKey(hash))
973 if !ok {
974 return domain.CIRunner{}, domain.ErrNotFound
975 }
976 item, ok := snap.Get(ctx, runnerKey(string(idx.Value)))
977 if !ok {
978 return domain.CIRunner{}, domain.ErrNotFound
979 }
980 r, err := decodeRunner(item.Value)
981 if err != nil {
982 return domain.CIRunner{}, err
983 }
984 if r.TokenHash != hash {
985 return domain.CIRunner{}, fmt.Errorf("runner %q: token index does not match the stored hash", r.ID)
986 }
987 return r, nil
988}
989
990// ListRunners returns every runner ordered by name.
991func (c *CI) ListRunners(ctx context.Context) ([]domain.CIRunner, error) {
992 var out []domain.CIRunner
993 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(runnerPrefix)) {
994 r, err := decodeRunner(item.Value)
995 if err != nil {
996 return nil, fmt.Errorf("decode %s: %w", key, err)
997 }
998 out = append(out, r)
999 }
1000 slices.SortFunc(out, func(a, b domain.CIRunner) int {
1001 return strings.Compare(strings.ToLower(a.Name), strings.ToLower(b.Name))
1002 })
1003 return out, nil
1004}
1005
1006// RemoveRunner removes the runner, its name and its token.
1007func (c *CI) RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error) {
1008 for range createRunTries {
1009 item, ok := c.db.Get(ctx, runnerKey(id))
1010 if !ok || id == "" || strings.Contains(id, "/") {
1011 return domain.CIRunner{}, domain.ErrNotFound
1012 }
1013 r, err := decodeRunner(item.Value)
1014 if err != nil {
1015 return domain.CIRunner{}, err
1016 }
1017 var b kv.Batch
1018 b.RequireVersion(runnerKey(id), item.Version)
1019 b.Delete(runnerKey(id))
1020 b.Delete(runnerNameKey(r.Name))
1021 b.Delete(runnerTokenKey(r.TokenHash))
1022 _, err = c.db.Apply(ctx, &b)
1023 if errors.Is(err, kv.ErrConflict) {
1024 continue
1025 }
1026 if err != nil {
1027 return domain.CIRunner{}, fmt.Errorf("remove runner %q: %w", id, err)
1028 }
1029 return r, nil
1030 }
1031 return domain.CIRunner{}, fmt.Errorf("remove runner %q: %w", id, domain.ErrConflict)
1032}