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