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