1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "crypto/rand"
8 "encoding/base64"
9 "encoding/hex"
10 "fmt"
11 "slices"
12 "strings"
13 "sync"
14 "time"
15
16 "github.com/bnema/zerowrap"
17
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20)
21
22// Defaults of the CI engine.
23const (
24 // DefaultCIPollTimeout is how long FetchJob holds a call: the contract's 50 s.
25 DefaultCIPollTimeout = 50 * time.Second
26 // DefaultCITick is how often the watchdog looks at the jobs in flight.
27 DefaultCITick = time.Second
28 // DefaultCIKeepalive is how often WatchJob sends a keepalive.
29 DefaultCIKeepalive = 30 * time.Second
30 // DefaultCITriggerQueue is the size of the queue of pushes to plan.
31 DefaultCITriggerQueue = 256
32 // DefaultCIRemoteConcurrency is the git ls-remote processes ResolveActions
33 // runs at once, whatever the number of jobs that resolve.
34 DefaultCIRemoteConcurrency = 4
35 // DefaultCIResolveTimeout bounds one ResolveActions call.
36 DefaultCIResolveTimeout = 60 * time.Second
37 // DefaultCIConcurrencyPruneBatch is the stale marks of a concurrency group a
38 // transaction removes at most.
39 DefaultCIConcurrencyPruneBatch = 256
40 // MaxCIChecks is the most checks ListChecks returns.
41 MaxCIChecks = 1000
42 // MaxCILogStreams and MaxCILogStreamsPerClient bound the concurrent log
43 // streams (GetJobLog).
44 MaxCILogStreams = 64
45 MaxCILogStreamsPerClient = 8
46 // maxCIWorkflowBytes is the largest workflow file (the parser's limit).
47 maxCIWorkflowBytes = 1 << 20
48 // maxCIWorkflowFiles bounds the workflow files read per push.
49 maxCIWorkflowFiles = 256
50)
51
52// CIConfig holds what the CI use cases need. Store, Logs, Summaries, Meta, Repos, Reader,
53// Git, Diff, Bus, Clock, Owner and PublicURL are required (the config supplies
54// the defaults of Owner and Retention).
55type CIConfig struct {
56 Store ports.CIStore
57 Logs ports.CILogStore
58 // Summaries keeps the job summaries, deleted with the logs of their run.
59 Summaries ports.CISummaryStore
60 Meta ports.RepoMetaStore
61 Repos ports.GitRepos
62 Reader ports.RepoReader
63 Git ports.MergeGit
64 Diff ports.CIGit
65 // MRs, Pull and Committer give the workflows their pull_request event and
66 // the refs/pull/<n>/{head,merge} refs of the merge requests (CI-3, D5);
67 // without MRs or Pull the merge requests start no run and have no such
68 // refs. Committer is the identity of the test merge commits (default
69 // "ermite"). Fresh, when set, is told when the refs of a repository changed
70 // so that the reader sees them at once.
71 MRs ports.MergeRequestStore
72 Pull ports.PullGit
73 Committer domain.Identity
74 Fresh ports.RepoFreshener
75 // PullStage, when set, is called by the workers of the merge requests at
76 // the stages "event" (an activity is taken), "computed" (its merge is
77 // computed, before it is published) and "refresh" (a refresh is taken).
78 // Tests use it to interleave other changes.
79 PullStage func(stage string)
80 // PullRefreshes bounds the merge requests waiting for a refresh of their refs
81 // (default DefaultCIPullRefreshes); past it repositories are rescanned. It
82 // does not bound the activities (synchronize runs), which have their own queue
83 // with the overload policy of the push triggers.
84 PullRefreshes int
85 // PullAttempts bounds the merge requests whose last conflict is remembered
86 // (default DefaultCIPullAttempts), so that a rescan does not try a conflicting
87 // merge again until a branch moves.
88 PullAttempts int
89 // PullSnapshot bounds the merge requests one push takes the snapshot of, in the
90 // push handler (default DefaultCIPullSnapshot): past it the others are skipped
91 // with an error in the log and the repository is rescanned for its refs.
92 PullSnapshot int
93 Bus ports.EventBus
94 Clock ports.Clock
95 // Auth re-validates the token of a log stream; nil skips it.
96 Auth ports.Revalidator
97 // Remote lists the refs of github.com repositories for ResolveActions; nil
98 // answers every github.com action with unavailable.
99 Remote ports.GitRemote
100 // Workflows downloads reusable workflow files from github.com at a resolved
101 // commit; nil fails such a call as unavailable.
102 Workflows ports.WorkflowFetcher
103 // Archive serves the tarballs of the REST subset.
104 Archive ports.GitArchive
105 // Tokens finds the stored token of the pusher of a run, to ask what it may
106 // read when an action of another repository is resolved for the run. Nil
107 // reads nothing for a token (fail closed).
108 Tokens ports.TokenStore
109 // OIDC signs the ID tokens of the jobs (CI-4); nil issues none: jobs get no
110 // token endpoint, and the provider is not served.
111 OIDC *OIDC
112 // Settings resolves the variables and secrets of a job (CI-3). Nil gives
113 // jobs none, and an empty "vars" context. Leave it unset rather than
114 // assigning a typed nil pointer.
115 Settings ports.CIJobSettings
116 // RemoteConcurrency bounds the git ls-remote processes at once (default
117 // DefaultCIRemoteConcurrency); ResolveTimeout bounds one ResolveActions call
118 // (default DefaultCIResolveTimeout).
119 RemoteConcurrency int
120 ResolveTimeout time.Duration
121 // Owner is the owner login of the github context.
122 Owner string
123 // PublicURL is the base URL of the instance (github.server_url).
124 PublicURL string
125 // Retention is how long runs, jobs, checks and logs are kept.
126 Retention time.Duration
127 // Artifacts and ArtifactFiles keep the artifacts of the runs (CI-4 D3):
128 // the records and the contents. Without both the artifact services refuse
129 // every call. ArtifactLimits are their settings; a zero field takes its
130 // default (the run retention of domain.CIRetentionDays,
131 // domain.DefaultCIArtifactQuota, domain.DefaultCIArtifactMaxSize). Leave
132 // them unset rather than assigning typed nil pointers.
133 Artifacts ports.CIArtifactStore
134 ArtifactFiles ports.CIArtifactFiles
135 ArtifactLimits ports.ArtifactLimits
136
137 // The durations below default to the contract's; tests shorten them.
138 PollTimeout time.Duration
139 Tick time.Duration
140 Keepalive time.Duration
141 AckTimeout time.Duration
142 HeartbeatDeadline time.Duration
143 CancelGrace time.Duration
144 TimeoutSlack time.Duration
145 QueueTimeout time.Duration
146 TriggerQueue int
147 // ConcurrencyRetries bounds the retries of a delivery of the concurrency
148 // groups that failed (a cancel, the planning of a promoted entry): 0 is the
149 // default (8), a negative number none, which the tests use to leave the
150 // recovery to the reconciliation.
151 ConcurrencyRetries int
152 // ConcurrencyPruneBatch bounds the stale marks of a concurrency group one
153 // transaction removes (default 256, far below the size limit of a kv batch);
154 // the others go with the next decisions and the reconciliation.
155 ConcurrencyPruneBatch int
156 // ConcurrencyQueueMax is how many entries may wait in a queue: max group:
157 // 0 is GitHub's limit (domain.CIMaxConcurrencyQueue); the tests lower it so
158 // as not to start a hundred runs.
159 ConcurrencyQueueMax int
160 // ScheduleTick is how often the scheduler looks at the clock (default one
161 // second) and ScheduleBudget the repositories it handles per tick (default
162 // 64); see ci_schedule.go.
163 ScheduleTick time.Duration
164 ScheduleBudget int
165 // RefTips holds who moves each branch and tag to its object, written before
166 // the ref moves (Pushes, MergeRequests), for the actor and trust of scheduled
167 // runs and the trust of dispatched ones (ci_tips.go). Nil: they are never
168 // trusted. Leave it unset rather than assigning a typed nil pointer.
169 RefTips ports.CIRefTipStore
170 // DispatchResolved, when set, is called by DispatchWorkflow after it resolved
171 // its ref and before it judges the trust of the run. Tests use it to move the
172 // ref in between.
173 DispatchResolved func()
174 // RunEventHandled, when set, is called after the workflows waiting for a
175 // "requested" or "completed" event of a run were started (or none matched).
176 // Tests use it as a barrier.
177 RunEventHandled func(runID int64, action string)
178 // ScheduleVisit, when set, is called with each repository the scheduler
179 // looks at. Tests use it.
180 ScheduleVisit func(repo domain.RepoName)
181}
182
183func (c CIConfig) withDefaults() CIConfig {
184 def := func(d *time.Duration, v time.Duration) {
185 if *d <= 0 {
186 *d = v
187 }
188 }
189 def(&c.PollTimeout, DefaultCIPollTimeout)
190 def(&c.Tick, DefaultCITick)
191 def(&c.Keepalive, DefaultCIKeepalive)
192 def(&c.AckTimeout, domain.CIAckTimeout)
193 def(&c.HeartbeatDeadline, domain.CIHeartbeatDeadline)
194 def(&c.CancelGrace, domain.CICancelGrace)
195 def(&c.TimeoutSlack, domain.CITimeoutSlack)
196 def(&c.QueueTimeout, domain.CIQueueTimeout)
197 def(&c.Retention, domain.CIRetentionDays*24*time.Hour)
198 def(&c.ArtifactLimits.Retention, domain.CIRetentionDays*24*time.Hour)
199 if c.ArtifactLimits.Quota <= 0 {
200 c.ArtifactLimits.Quota = domain.DefaultCIArtifactQuota
201 }
202 if c.ArtifactLimits.MaxSize <= 0 {
203 c.ArtifactLimits.MaxSize = domain.DefaultCIArtifactMaxSize
204 }
205 if c.RemoteConcurrency <= 0 {
206 c.RemoteConcurrency = DefaultCIRemoteConcurrency
207 }
208 def(&c.ResolveTimeout, DefaultCIResolveTimeout)
209 if c.PullSnapshot <= 0 {
210 c.PullSnapshot = DefaultCIPullSnapshot
211 }
212 if c.PullRefreshes <= 0 {
213 c.PullRefreshes = DefaultCIPullRefreshes
214 }
215 if c.TriggerQueue <= 0 {
216 c.TriggerQueue = DefaultCITriggerQueue
217 }
218 if c.ConcurrencyQueueMax <= 0 {
219 c.ConcurrencyQueueMax = domain.CIMaxConcurrencyQueue
220 }
221 if c.ConcurrencyPruneBatch <= 0 {
222 c.ConcurrencyPruneBatch = DefaultCIConcurrencyPruneBatch
223 }
224 def(&c.ScheduleTick, DefaultCIScheduleTick)
225 if c.ScheduleBudget <= 0 {
226 c.ScheduleBudget = DefaultCIScheduleBudget
227 }
228 return c
229}
230
231// CI is the CI use cases: the workflow trigger, the planner, the assignment of
232// jobs to runners, the runner protocol and the owner side (ADR 0025).
233//
234// Every change of a run goes through its lock (withRun): the records of a run
235// are read, changed and written again by one goroutine at a time, so that two
236// calls (a runner's update and an owner's cancel) never lose one another's
237// write. Assignment takes the assignment lock before the lock of the run.
238type CI struct {
239 cfg CIConfig
240 boot time.Time
241 queue chan ciTrigger
242 // runEvents are the requested and completed events of runs, for the
243 // workflow_run workflows (ci_workflowrun.go).
244 runEvents chan runEvent
245 sched *scheduler // the schedule trigger (ci_schedule.go)
246 pull *pullQueue // the work of the merge requests (ci_pull.go)
247
248 pullFileLists pullFilesCache // the files of the merge requests (ci_pullfiles.go)
249
250 stop context.CancelFunc
251 ctx context.Context
252 wg sync.WaitGroup
253
254 remoteSlots chan struct{} // slots of the git ls-remote processes
255 remoteCache *refsCache // the refs of github.com repositories, for every job
256 locks sync.Map // run id -> *sync.Mutex
257 assign sync.Mutex
258 jobs notifier // woken when a job changes: key is the job reference
259 pending notifier // woken when a job may be ready or a runner was removed
260 flows *wfCache
261
262 mu sync.Mutex
263 heartbeats map[string]time.Time // job reference -> last heartbeat
264 seen map[string]time.Time // runner id -> last seen
265 polls map[string]pollRecord
266 // archiveTokens are the secrets of the tokens given for the archives of
267 // ermite actions, per job reference then "<repository>@<sha>", so that a job
268 // that resolves the same commit twice gets one token (memory only: a
269 // restart issues new ones).
270 archiveTokens map[string]map[string]string
271 streams int
272 perClient map[string]int
273 closed bool
274 // artifactLocks serialize the commits of each artifact (ci_artifacts.go).
275 artifactLocks map[int64]*artifactLock
276 // reservations are the artifact bytes being written (ci_artifacts.go).
277 reservations artifactReservations
278}
279
280var (
281 _ ports.RunnerService = (*CI)(nil)
282 _ ports.RunService = (*CI)(nil)
283 _ ports.WorkflowTrigger = (*CI)(nil)
284 _ ports.CIChecksReader = (*CI)(nil)
285)
286
287// pollRecord remembers the assignment a poll made, for the retry of a lost
288// response (FetchJobRequest.poll_id). The token is kept in memory only.
289type pollRecord struct {
290 pollID string
291 ref string
292 token string
293 ghToken string
294 idToken string
295 hash domain.TokenHash
296}
297
298// NewCI builds the CI use cases. ctx carries the logger; its cancellation is
299// not followed: call Start to run the workers and Close to stop them.
300func NewCI(ctx context.Context, cfg CIConfig) *CI {
301 if cfg.Owner == "" {
302 panic("usecase: CIConfig.Owner is required")
303 }
304 cfg = cfg.withDefaults()
305 run, stop := context.WithCancel(context.WithoutCancel(ctx))
306 return &CI{
307 cfg: cfg, boot: cfg.Clock.Now(), queue: make(chan ciTrigger, cfg.TriggerQueue), runEvents: make(chan runEvent, cfg.TriggerQueue), sched: newScheduler(cfg.Clock.Now()), pull: newPullQueue(cfg.TriggerQueue, cfg.PullRefreshes, cfg.PullAttempts),
308 ctx: run, stop: stop, flows: newWfCache(64), remoteSlots: make(chan struct{}, cfg.RemoteConcurrency), remoteCache: newRefsCache(),
309 heartbeats: map[string]time.Time{}, seen: map[string]time.Time{}, polls: map[string]pollRecord{},
310 archiveTokens: map[string]map[string]string{},
311 perClient: map[string]int{},
312 }
313}
314
315// Start runs the trigger worker and the watchdog, and plans again the runs
316// that were open when the server stopped.
317func (c *CI) Start() {
318 c.wg.Add(6)
319 go c.scheduleLoop()
320 go c.runEventLoop()
321 go c.pullLoop()
322 go c.refreshLoop()
323 go c.triggerLoop()
324 go c.watchdog()
325 c.wg.Add(1)
326 go func() {
327 defer c.wg.Done()
328 c.resume()
329 }()
330}
331
332// Close stops the workers and waits for them.
333func (c *CI) Close(_ context.Context) {
334 c.mu.Lock()
335 c.closed = true
336 c.mu.Unlock()
337 c.stop()
338 c.wg.Wait()
339}
340
341func (c *CI) now() time.Time { return c.cfg.Clock.Now().UTC() }
342
343// --- identifiers and secrets ---
344
345func newToken(prefix string) string {
346 var raw [32]byte
347 _, _ = rand.Read(raw[:])
348 return prefix + base64.RawURLEncoding.EncodeToString(raw[:])
349}
350
351func newRunnerID() string {
352 var raw [8]byte
353 _, _ = rand.Read(raw[:])
354 return hex.EncodeToString(raw[:])
355}
356
357func jobRef(j domain.CIJob) string { return fmt.Sprintf("%d/%d/%s", j.RunID, j.RunAttempt, j.ID) }
358
359// --- locks and notifications ---
360
361// notifier wakes the goroutines waiting on a key. A channel exists only while
362// somebody waits on it.
363type notifier struct {
364 mu sync.Mutex
365 ch map[string]chan struct{}
366}
367
368// wait returns a channel closed by the next notify of key.
369func (n *notifier) wait(key string) <-chan struct{} {
370 n.mu.Lock()
371 defer n.mu.Unlock()
372 if n.ch == nil {
373 n.ch = map[string]chan struct{}{}
374 }
375 c, ok := n.ch[key]
376 if !ok {
377 c = make(chan struct{})
378 n.ch[key] = c
379 }
380 return c
381}
382
383func (n *notifier) notify(key string) {
384 n.mu.Lock()
385 defer n.mu.Unlock()
386 if c, ok := n.ch[key]; ok {
387 close(c)
388 delete(n.ch, key)
389 }
390}
391
392const queueKey = "q"
393
394func (c *CI) wakeQueue() { c.pending.notify(queueKey) }
395
396func (c *CI) runLock(runID int64) *sync.Mutex {
397 m, _ := c.locks.LoadOrStore(runID, &sync.Mutex{})
398 return m.(*sync.Mutex)
399}
400
401// --- run state ---
402
403// runState is a run being changed under its lock: the run, the jobs of one of
404// its attempts, and what must be written. flush writes everything in one batch
405// and publishes the events.
406type runState struct {
407 c *CI
408 run domain.CIRun
409 attempt int32
410 jobs []domain.CIJob
411 src *domain.CIRunSource
412
413 runDirty bool
414 dirty map[string]bool
415 // chk marks jobs whose check must be written even when the job itself did
416 // not change (a rerun moves every check to the new attempt).
417 checks map[string]bool
418 // tokens are the job repository tokens to create, revoke the jobs whose
419 // tokens are deleted, with the next flush.
420 tokens []domain.CIRepoToken
421 revoke []ports.CIJobRef
422 // resolutions are the action resolutions to keep with the next flush.
423 resolutions []domain.CIActionResolution
424 // annotations replace the annotations of steps, clear drops all those of
425 // executions of jobs, with the next flush.
426 annotations []ports.CIStepAnnotations
427 clear []ports.CILogRef
428}
429
430// withRun locks the run, loads it with the jobs of attempt (0 for the latest)
431// and calls fn. The run is found by id alone through its source.
432func (c *CI) withRun(ctx context.Context, runID int64, attempt int32, fn func(rs *runState) error) error {
433 lock := c.runLock(runID)
434 lock.Lock()
435 defer lock.Unlock()
436 rs, err := c.loadRun(ctx, runID, attempt)
437 if err != nil {
438 return err
439 }
440 return fn(rs)
441}
442
443func (c *CI) loadRun(ctx context.Context, runID int64, attempt int32) (*runState, error) {
444 src, err := c.cfg.Store.GetRunSource(ctx, runID)
445 if err != nil {
446 return nil, err
447 }
448 run, err := c.cfg.Store.GetRun(ctx, src.Repo, runID)
449 if err != nil {
450 return nil, err
451 }
452 if attempt <= 0 {
453 attempt = run.Attempt
454 }
455 jobs, err := c.cfg.Store.ListJobs(ctx, runID, attempt)
456 if err != nil {
457 return nil, err
458 }
459 return &runState{c: c, run: run, attempt: attempt, jobs: jobs, src: &src, dirty: map[string]bool{}, checks: map[string]bool{}}, nil
460}
461
462// latest reports whether the state holds the latest attempt of the run.
463func (rs *runState) latest() bool { return rs.attempt == rs.run.Attempt }
464
465func (rs *runState) find(id string) *domain.CIJob {
466 for i := range rs.jobs {
467 if rs.jobs[i].ID == id {
468 return &rs.jobs[i]
469 }
470 }
471 return nil
472}
473
474func (rs *runState) touch(j *domain.CIJob) { rs.dirty[j.ID] = true }
475
476// checkOf builds the check of a job.
477func (rs *runState) checkOf(j domain.CIJob) domain.CICheck {
478 return domain.CICheck{
479 ID: j.ID, Repo: rs.run.Repo, Sha: rs.run.CheckSha(), Name: j.Name, Status: j.Status(), Conclusion: j.Conclusion,
480 RunID: rs.run.ID, JobID: j.ID, Attempt: rs.run.Attempt, WorkflowName: rs.run.WorkflowName,
481 CreatedAt: j.CreatedAt, StartedAt: j.StartedAt, CompletedAt: j.CompletedAt,
482 }
483}
484
485// flush recomputes the status of the run, writes the changed jobs, their
486// checks and the run in one batch, and publishes RunUpdated and JobUpdated.
487func (rs *runState) flush(ctx context.Context) error {
488 c := rs.c
489 now := c.now()
490 var w ports.CIWrite
491 var changedJobs []int
492 completing := false
493 for i := range rs.jobs {
494 if rs.dirty[rs.jobs[i].ID] {
495 w.Jobs = append(w.Jobs, rs.jobs[i])
496 changedJobs = append(changedJobs, i)
497 }
498 }
499 if rs.latest() {
500 for i := range rs.jobs {
501 if rs.dirty[rs.jobs[i].ID] || rs.checks[rs.jobs[i].ID] {
502 w.Checks = append(w.Checks, rs.checkOf(rs.jobs[i]))
503 }
504 }
505 if next := rs.recompute(now); next != rs.run {
506 completing = rs.run.Status != domain.CICompleted && next.Status == domain.CICompleted
507 rs.run, rs.runDirty = next, true
508 }
509 }
510 if rs.runDirty {
511 w.Runs = []domain.CIRun{rs.run}
512 }
513 w.RepoTokens, w.RevokeJobs, w.Resolutions = rs.tokens, rs.revoke, rs.resolutions
514 w.Annotations, w.ClearAnnotations = rs.annotations, rs.clear
515 if len(w.Jobs) == 0 && len(w.Runs) == 0 && len(w.Checks) == 0 && len(w.RepoTokens) == 0 && len(w.RevokeJobs) == 0 &&
516 len(w.Resolutions) == 0 && len(w.Annotations) == 0 && len(w.ClearAnnotations) == 0 {
517 return nil
518 }
519 v, err := c.cfg.Store.Apply(ctx, w)
520 if err != nil {
521 return fmt.Errorf("write run %d: %w", rs.run.ID, err)
522 }
523 for _, i := range changedJobs {
524 rs.jobs[i].Version = v
525 }
526 if rs.runDirty {
527 rs.run.Version = v
528 }
529 changedRun := rs.runDirty
530 rs.releaseEnded(ctx, changedJobs, changedRun && rs.run.Status == domain.CICompleted)
531 rs.dirty, rs.checks, rs.runDirty = map[string]bool{}, map[string]bool{}, false
532 rs.tokens, rs.revoke, rs.resolutions = nil, nil, nil
533 rs.annotations, rs.clear = nil, nil
534
535 repo, err := c.cfg.Meta.Get(ctx, rs.run.Repo)
536 if err != nil {
537 for _, i := range changedJobs {
538 c.jobs.notify(jobRef(rs.jobs[i]))
539 }
540 return nil // the repository is gone: nobody to tell
541 }
542 for _, i := range changedJobs {
543 j := rs.jobs[i]
544 lines, _ := c.cfg.Logs.Count(ctx, ports.CILogRef{RunID: j.RunID, JobID: j.ID, Attempt: j.Attempt})
545 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, Lines: lines, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
546 c.jobs.notify(jobRef(j))
547 if j.Phase == domain.CIPhaseReady {
548 c.wakeQueue()
549 }
550 }
551 if changedRun {
552 c.cfg.Bus.Publish(ctx, domain.RunUpdated{Run: rs.run, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
553 }
554 if completing {
555 c.onRunEvent(ctx, rs.run, runCompleted)
556 }
557 return nil
558}
559
560// recompute returns the run with the status, the conclusion and the
561// timestamps its jobs imply.
562func (rs *runState) recompute(now time.Time) domain.CIRun {
563 run := rs.run
564 if run.Status == domain.CICompleted && len(rs.jobs) == 0 {
565 return run // an invalid workflow file: failed from the start
566 }
567 allEnded := len(rs.jobs) > 0
568 started := time.Time{}
569 var failed, cancelled bool
570 for _, j := range rs.jobs {
571 if !j.Ended() {
572 allEnded = false
573 }
574 // A job kept from an earlier attempt did not start in this one.
575 if j.Attempt == rs.attempt && !j.StartedAt.IsZero() && (started.IsZero() || j.StartedAt.Before(started)) {
576 started = j.StartedAt
577 }
578 switch {
579 case j.Conclusion.IsFailure() && j.ContinueOnError:
580 // continue-on-error: the job failed, the run does not.
581 case j.Conclusion.IsFailure():
582 failed = true
583 case j.Conclusion == domain.CICancelled:
584 cancelled = true
585 }
586 }
587 if run.StartedAt.IsZero() && !started.IsZero() {
588 run.StartedAt = started
589 }
590 switch {
591 case allEnded:
592 run.Status, run.CompletedAt = domain.CICompleted, now
593 switch {
594 case run.CancelRequested:
595 run.Conclusion = domain.CICancelled
596 case failed:
597 run.Conclusion = domain.CIFailure
598 case cancelled:
599 run.Conclusion = domain.CICancelled
600 default:
601 run.Conclusion = domain.CISuccess
602 }
603 if rs.run.Status == domain.CICompleted {
604 run.CompletedAt = rs.run.CompletedAt
605 }
606 case !run.StartedAt.IsZero():
607 run.Status, run.Conclusion, run.CompletedAt = domain.CIInProgress, domain.CINoConclusion, time.Time{}
608 default:
609 run.Status, run.Conclusion, run.CompletedAt = domain.CIQueued, domain.CINoConclusion, time.Time{}
610 }
611 return run
612}
613
614// --- access ---
615
616// readableRepo looks the repository up and applies the read rule.
617func (c *CI) readableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
618 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
619 if err != nil {
620 return domain.Repo{}, err
621 }
622 if !canRead(p, repo) {
623 return domain.Repo{}, domain.ErrNotFound
624 }
625 return repo, nil
626}
627
628// writableRepo is the access rule of cancelling and re-running: a principal
629// whose token holds ci:write on the repository sees the repository for this
630// action even without repo:read, as for merge requests.
631func (c *CI) writableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
632 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
633 if err != nil {
634 return domain.Repo{}, err
635 }
636 if !canRead(p, repo) && !p.Allows(domain.ScopeCIWrite, name) {
637 return domain.Repo{}, domain.ErrNotFound
638 }
639 if err := scopeAccess(p, repo, domain.ScopeCIWrite); err != nil {
640 return domain.Repo{}, err
641 }
642 return repo, nil
643}
644
645// runWriteAccess is the branch rule of the write actions on a run, cancel and
646// re-run, which come after the ci:write check. A run is tied to the ref that
647// made it, and re-running it executes the workflow of that ref again while
648// cancelling it stops the checks of that ref. So, like a push, a merge or a
649// branch delete, these actions fall under the branch prefixes of a limited
650// token (ADR 0027): it may act only on the runs of branches under its prefixes,
651// and never on a run of a tag or of any other ref. Without prefixes, every run.
652//
653// Reading runs, jobs, logs and checks is not a write: it needs only the access
654// to read the repository, as for the code, whatever the prefixes. The caller
655// has seen the run by then, so the refusal is forbidden, not not found.
656func runWriteAccess(p domain.Principal, run domain.CIRun) error {
657 if !p.AllowsRef(run.Ref) {
658 return fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, run.Ref)
659 }
660 return nil
661}
662
663// canManageRunners is the access rule of runners: runner:register on a token
664// that is not restricted to repositories (a runner serves every repository),
665// or root. A token limited to branch prefixes never manages runners, whatever
666// scopes it holds: a runner runs the workflows of every branch of every
667// repository, with their code and secrets, so registering or removing one
668// reaches far beyond the prefixes (a replaced runner could also be a way to
669// take over the jobs of main). Like repository administration and owner
670// rights, it is out of reach of a limited token.
671func canManageRunners(p domain.Principal) error {
672 switch {
673 case p.IsAnonymous():
674 return domain.ErrUnauthenticated
675 case !p.HasScope(domain.ScopeRunnerRegister):
676 return fmt.Errorf("%w: this needs the %s scope", domain.ErrForbidden, domain.ScopeRunnerRegister)
677 case len(p.Repos()) > 0:
678 return fmt.Errorf("%w: a runner serves every repository: the token must not be restricted to some", domain.ErrForbidden)
679 case p.IsBranchLimited():
680 return fmt.Errorf("%w: a token limited to branch prefixes may not manage runners", domain.ErrForbidden)
681 }
682 return nil
683}
684
685// --- small helpers ---
686
687func truncateRunes(s string, n int) string {
688 if len(s) <= n {
689 return s
690 }
691 r := []rune(s)
692 if len(r) <= n {
693 return s
694 }
695 return string(r[:n])
696}
697
698// cutText cuts s to n bytes at a character boundary.
699func cutText(s string, n int) string {
700 if len(s) <= n {
701 return s
702 }
703 for n > 0 && n < len(s) && s[n]&0xC0 == 0x80 {
704 n--
705 }
706 return s[:n]
707}
708
709func sortedKeys[V any](m map[string]V) []string {
710 keys := make([]string, 0, len(m))
711 for k := range m {
712 keys = append(keys, k)
713 }
714 slices.Sort(keys)
715 return keys
716}
717
718// ctxLog returns the logger of the context, addressable so that the level
719// methods can be chained on it.
720func ctxLog(ctx context.Context) *zerowrap.Logger {
721 l := zerowrap.FromCtx(ctx)
722 return &l
723}
724
725// logRun returns the logger of the context with the fields of the run.
726func logRun(ctx context.Context, run domain.CIRun) zerowrap.Logger {
727 return zerowrap.FromCtx(ctx).WithFields(map[string]any{"repo": run.Repo.String(), "run_id": run.ID})
728}
729
730func isHex40(s string) bool {
731 if len(s) != 40 {
732 return false
733 }
734 return strings.Trim(s, "0123456789abcdef") == ""
735}