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 // WorkflowState keeps the workflows a repository disabled (ci_workflows.go).
171 // Nil: none is. Leave it unset rather than assigning a typed nil pointer.
172 WorkflowState ports.CIWorkflowStore
173 // DispatchResolved, when set, is called by DispatchWorkflow after it resolved
174 // its ref and before it judges the trust of the run. Tests use it to move the
175 // ref in between.
176 DispatchResolved func()
177 // RunEventHandled, when set, is called after the workflows waiting for a
178 // "requested" or "completed" event of a run were started (or none matched).
179 // Tests use it as a barrier.
180 RunEventHandled func(runID int64, action string)
181 // ScheduleLoaded, when set, is called after the scheduler read the workflows
182 // of a repository and before it caches and fires them. Tests use it to
183 // change the disabled state in that window.
184 ScheduleLoaded func(repo domain.RepoName)
185 // ScheduleVisit, when set, is called with each repository the scheduler
186 // looks at. Tests use it.
187 ScheduleVisit func(repo domain.RepoName)
188}
189
190func (c CIConfig) withDefaults() CIConfig {
191 def := func(d *time.Duration, v time.Duration) {
192 if *d <= 0 {
193 *d = v
194 }
195 }
196 def(&c.PollTimeout, DefaultCIPollTimeout)
197 def(&c.Tick, DefaultCITick)
198 def(&c.Keepalive, DefaultCIKeepalive)
199 def(&c.AckTimeout, domain.CIAckTimeout)
200 def(&c.HeartbeatDeadline, domain.CIHeartbeatDeadline)
201 def(&c.CancelGrace, domain.CICancelGrace)
202 def(&c.TimeoutSlack, domain.CITimeoutSlack)
203 def(&c.QueueTimeout, domain.CIQueueTimeout)
204 def(&c.Retention, domain.CIRetentionDays*24*time.Hour)
205 def(&c.ArtifactLimits.Retention, domain.CIRetentionDays*24*time.Hour)
206 if c.ArtifactLimits.Quota <= 0 {
207 c.ArtifactLimits.Quota = domain.DefaultCIArtifactQuota
208 }
209 if c.ArtifactLimits.MaxSize <= 0 {
210 c.ArtifactLimits.MaxSize = domain.DefaultCIArtifactMaxSize
211 }
212 if c.RemoteConcurrency <= 0 {
213 c.RemoteConcurrency = DefaultCIRemoteConcurrency
214 }
215 def(&c.ResolveTimeout, DefaultCIResolveTimeout)
216 if c.PullSnapshot <= 0 {
217 c.PullSnapshot = DefaultCIPullSnapshot
218 }
219 if c.PullRefreshes <= 0 {
220 c.PullRefreshes = DefaultCIPullRefreshes
221 }
222 if c.TriggerQueue <= 0 {
223 c.TriggerQueue = DefaultCITriggerQueue
224 }
225 if c.ConcurrencyQueueMax <= 0 {
226 c.ConcurrencyQueueMax = domain.CIMaxConcurrencyQueue
227 }
228 if c.ConcurrencyPruneBatch <= 0 {
229 c.ConcurrencyPruneBatch = DefaultCIConcurrencyPruneBatch
230 }
231 def(&c.ScheduleTick, DefaultCIScheduleTick)
232 if c.ScheduleBudget <= 0 {
233 c.ScheduleBudget = DefaultCIScheduleBudget
234 }
235 return c
236}
237
238// CI is the CI use cases: the workflow trigger, the planner, the assignment of
239// jobs to runners, the runner protocol and the owner side (ADR 0025).
240//
241// Every change of a run goes through its lock (withRun): the records of a run
242// are read, changed and written again by one goroutine at a time, so that two
243// calls (a runner's update and an owner's cancel) never lose one another's
244// write. Assignment takes the assignment lock before the lock of the run.
245type CI struct {
246 cfg CIConfig
247 boot time.Time
248 queue chan ciTrigger
249 // runEvents are the requested and completed events of runs, for the
250 // workflow_run workflows (ci_workflowrun.go).
251 runEvents chan runEvent
252 sched *scheduler // the schedule trigger (ci_schedule.go)
253 pull *pullQueue // the work of the merge requests (ci_pull.go)
254
255 pullFileLists pullFilesCache // the files of the merge requests (ci_pullfiles.go)
256
257 stop context.CancelFunc
258 ctx context.Context
259 wg sync.WaitGroup
260
261 remoteSlots chan struct{} // slots of the git ls-remote processes
262 remoteCache *refsCache // the refs of github.com repositories, for every job
263 locks sync.Map // run id -> *sync.Mutex
264 assign sync.Mutex
265 jobs notifier // woken when a job changes: key is the job reference
266 pending notifier // woken when a job may be ready or a runner was removed
267 flows *wfCache
268
269 mu sync.Mutex
270 heartbeats map[string]time.Time // job reference -> last heartbeat
271 seen map[string]time.Time // runner id -> last seen
272 polls map[string]pollRecord
273 // archiveTokens are the secrets of the tokens given for the archives of
274 // ermite actions, per job reference then "<repository>@<sha>", so that a job
275 // that resolves the same commit twice gets one token (memory only: a
276 // restart issues new ones).
277 archiveTokens map[string]map[string]string
278 streams int
279 perClient map[string]int
280 closed bool
281 // artifactLocks serialize the commits of each artifact (ci_artifacts.go).
282 artifactLocks map[int64]*artifactLock
283 // reservations are the artifact bytes being written (ci_artifacts.go).
284 reservations artifactReservations
285}
286
287var (
288 _ ports.RunnerService = (*CI)(nil)
289 _ ports.RunService = (*CI)(nil)
290 _ ports.WorkflowTrigger = (*CI)(nil)
291 _ ports.CIChecksReader = (*CI)(nil)
292)
293
294// pollRecord remembers the assignment a poll made, for the retry of a lost
295// response (FetchJobRequest.poll_id). The token is kept in memory only.
296type pollRecord struct {
297 pollID string
298 ref string
299 token string
300 ghToken string
301 idToken string
302 hash domain.TokenHash
303}
304
305// NewCI builds the CI use cases. ctx carries the logger; its cancellation is
306// not followed: call Start to run the workers and Close to stop them.
307func NewCI(ctx context.Context, cfg CIConfig) *CI {
308 if cfg.Owner == "" {
309 panic("usecase: CIConfig.Owner is required")
310 }
311 cfg = cfg.withDefaults()
312 run, stop := context.WithCancel(context.WithoutCancel(ctx))
313 return &CI{
314 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),
315 ctx: run, stop: stop, flows: newWfCache(64), remoteSlots: make(chan struct{}, cfg.RemoteConcurrency), remoteCache: newRefsCache(),
316 heartbeats: map[string]time.Time{}, seen: map[string]time.Time{}, polls: map[string]pollRecord{},
317 archiveTokens: map[string]map[string]string{},
318 perClient: map[string]int{},
319 }
320}
321
322// Start runs the trigger worker and the watchdog, and plans again the runs
323// that were open when the server stopped.
324func (c *CI) Start() {
325 c.wg.Add(6)
326 go c.scheduleLoop()
327 go c.runEventLoop()
328 go c.pullLoop()
329 go c.refreshLoop()
330 go c.triggerLoop()
331 go c.watchdog()
332 c.wg.Add(1)
333 go func() {
334 defer c.wg.Done()
335 c.resume()
336 }()
337}
338
339// Close stops the workers and waits for them.
340func (c *CI) Close(_ context.Context) {
341 c.mu.Lock()
342 c.closed = true
343 c.mu.Unlock()
344 c.stop()
345 c.wg.Wait()
346}
347
348func (c *CI) now() time.Time { return c.cfg.Clock.Now().UTC() }
349
350// --- identifiers and secrets ---
351
352func newToken(prefix string) string {
353 var raw [32]byte
354 _, _ = rand.Read(raw[:])
355 return prefix + base64.RawURLEncoding.EncodeToString(raw[:])
356}
357
358func newRunnerID() string {
359 var raw [8]byte
360 _, _ = rand.Read(raw[:])
361 return hex.EncodeToString(raw[:])
362}
363
364func jobRef(j domain.CIJob) string { return fmt.Sprintf("%d/%d/%s", j.RunID, j.RunAttempt, j.ID) }
365
366// --- locks and notifications ---
367
368// notifier wakes the goroutines waiting on a key. A channel exists only while
369// somebody waits on it.
370type notifier struct {
371 mu sync.Mutex
372 ch map[string]chan struct{}
373}
374
375// wait returns a channel closed by the next notify of key.
376func (n *notifier) wait(key string) <-chan struct{} {
377 n.mu.Lock()
378 defer n.mu.Unlock()
379 if n.ch == nil {
380 n.ch = map[string]chan struct{}{}
381 }
382 c, ok := n.ch[key]
383 if !ok {
384 c = make(chan struct{})
385 n.ch[key] = c
386 }
387 return c
388}
389
390func (n *notifier) notify(key string) {
391 n.mu.Lock()
392 defer n.mu.Unlock()
393 if c, ok := n.ch[key]; ok {
394 close(c)
395 delete(n.ch, key)
396 }
397}
398
399const queueKey = "q"
400
401func (c *CI) wakeQueue() { c.pending.notify(queueKey) }
402
403func (c *CI) runLock(runID int64) *sync.Mutex {
404 m, _ := c.locks.LoadOrStore(runID, &sync.Mutex{})
405 return m.(*sync.Mutex)
406}
407
408// --- run state ---
409
410// runState is a run being changed under its lock: the run, the jobs of one of
411// its attempts, and what must be written. flush writes everything in one batch
412// and publishes the events.
413type runState struct {
414 c *CI
415 run domain.CIRun
416 attempt int32
417 jobs []domain.CIJob
418 src *domain.CIRunSource
419
420 runDirty bool
421 dirty map[string]bool
422 // chk marks jobs whose check must be written even when the job itself did
423 // not change (a rerun moves every check to the new attempt).
424 checks map[string]bool
425 // tokens are the job repository tokens to create, revoke the jobs whose
426 // tokens are deleted, with the next flush.
427 tokens []domain.CIRepoToken
428 revoke []ports.CIJobRef
429 // resolutions are the action resolutions to keep with the next flush.
430 resolutions []domain.CIActionResolution
431 // annotations replace the annotations of steps, clear drops all those of
432 // executions of jobs, with the next flush.
433 annotations []ports.CIStepAnnotations
434 clear []ports.CILogRef
435}
436
437// withRun locks the run, loads it with the jobs of attempt (0 for the latest)
438// and calls fn. The run is found by id alone through its source.
439func (c *CI) withRun(ctx context.Context, runID int64, attempt int32, fn func(rs *runState) error) error {
440 lock := c.runLock(runID)
441 lock.Lock()
442 defer lock.Unlock()
443 rs, err := c.loadRun(ctx, runID, attempt)
444 if err != nil {
445 return err
446 }
447 return fn(rs)
448}
449
450func (c *CI) loadRun(ctx context.Context, runID int64, attempt int32) (*runState, error) {
451 src, err := c.cfg.Store.GetRunSource(ctx, runID)
452 if err != nil {
453 return nil, err
454 }
455 run, err := c.cfg.Store.GetRun(ctx, src.Repo, runID)
456 if err != nil {
457 return nil, err
458 }
459 if attempt <= 0 {
460 attempt = run.Attempt
461 }
462 jobs, err := c.cfg.Store.ListJobs(ctx, runID, attempt)
463 if err != nil {
464 return nil, err
465 }
466 return &runState{c: c, run: run, attempt: attempt, jobs: jobs, src: &src, dirty: map[string]bool{}, checks: map[string]bool{}}, nil
467}
468
469// latest reports whether the state holds the latest attempt of the run.
470func (rs *runState) latest() bool { return rs.attempt == rs.run.Attempt }
471
472func (rs *runState) find(id string) *domain.CIJob {
473 for i := range rs.jobs {
474 if rs.jobs[i].ID == id {
475 return &rs.jobs[i]
476 }
477 }
478 return nil
479}
480
481func (rs *runState) touch(j *domain.CIJob) { rs.dirty[j.ID] = true }
482
483// checkOf builds the check of a job.
484func (rs *runState) checkOf(j domain.CIJob) domain.CICheck {
485 return domain.CICheck{
486 ID: j.ID, Repo: rs.run.Repo, Sha: rs.run.CheckSha(), Name: j.Name, Status: j.Status(), Conclusion: j.Conclusion,
487 RunID: rs.run.ID, JobID: j.ID, Attempt: rs.run.Attempt, WorkflowName: rs.run.WorkflowName,
488 CreatedAt: j.CreatedAt, StartedAt: j.StartedAt, CompletedAt: j.CompletedAt,
489 }
490}
491
492// flush recomputes the status of the run, writes the changed jobs, their
493// checks and the run in one batch, and publishes RunUpdated and JobUpdated.
494func (rs *runState) flush(ctx context.Context) error {
495 c := rs.c
496 now := c.now()
497 var w ports.CIWrite
498 var changedJobs []int
499 completing := false
500 for i := range rs.jobs {
501 if rs.dirty[rs.jobs[i].ID] {
502 w.Jobs = append(w.Jobs, rs.jobs[i])
503 changedJobs = append(changedJobs, i)
504 }
505 }
506 if rs.latest() {
507 for i := range rs.jobs {
508 if rs.dirty[rs.jobs[i].ID] || rs.checks[rs.jobs[i].ID] {
509 w.Checks = append(w.Checks, rs.checkOf(rs.jobs[i]))
510 }
511 }
512 if next := rs.recompute(now); next != rs.run {
513 completing = rs.run.Status != domain.CICompleted && next.Status == domain.CICompleted
514 rs.run, rs.runDirty = next, true
515 }
516 }
517 if rs.runDirty {
518 w.Runs = []domain.CIRun{rs.run}
519 }
520 w.RepoTokens, w.RevokeJobs, w.Resolutions = rs.tokens, rs.revoke, rs.resolutions
521 w.Annotations, w.ClearAnnotations = rs.annotations, rs.clear
522 if len(w.Jobs) == 0 && len(w.Runs) == 0 && len(w.Checks) == 0 && len(w.RepoTokens) == 0 && len(w.RevokeJobs) == 0 &&
523 len(w.Resolutions) == 0 && len(w.Annotations) == 0 && len(w.ClearAnnotations) == 0 {
524 return nil
525 }
526 v, err := c.cfg.Store.Apply(ctx, w)
527 if err != nil {
528 return fmt.Errorf("write run %d: %w", rs.run.ID, err)
529 }
530 for _, i := range changedJobs {
531 rs.jobs[i].Version = v
532 }
533 if rs.runDirty {
534 rs.run.Version = v
535 }
536 changedRun := rs.runDirty
537 rs.releaseEnded(ctx, changedJobs, changedRun && rs.run.Status == domain.CICompleted)
538 rs.dirty, rs.checks, rs.runDirty = map[string]bool{}, map[string]bool{}, false
539 rs.tokens, rs.revoke, rs.resolutions = nil, nil, nil
540 rs.annotations, rs.clear = nil, nil
541
542 repo, err := c.cfg.Meta.Get(ctx, rs.run.Repo)
543 if err != nil {
544 for _, i := range changedJobs {
545 c.jobs.notify(jobRef(rs.jobs[i]))
546 }
547 return nil // the repository is gone: nobody to tell
548 }
549 for _, i := range changedJobs {
550 j := rs.jobs[i]
551 lines, _ := c.cfg.Logs.Count(ctx, ports.CILogRef{RunID: j.RunID, JobID: j.ID, Attempt: j.Attempt})
552 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, Lines: lines, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
553 c.jobs.notify(jobRef(j))
554 if j.Phase == domain.CIPhaseReady {
555 c.wakeQueue()
556 }
557 }
558 if changedRun {
559 c.cfg.Bus.Publish(ctx, domain.RunUpdated{Run: rs.run, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
560 }
561 if completing {
562 c.onRunEvent(ctx, rs.run, runCompleted)
563 }
564 return nil
565}
566
567// recompute returns the run with the status, the conclusion and the
568// timestamps its jobs imply.
569func (rs *runState) recompute(now time.Time) domain.CIRun {
570 run := rs.run
571 if run.Status == domain.CICompleted && len(rs.jobs) == 0 {
572 return run // an invalid workflow file: failed from the start
573 }
574 allEnded := len(rs.jobs) > 0
575 started := time.Time{}
576 var failed, cancelled bool
577 for _, j := range rs.jobs {
578 if !j.Ended() {
579 allEnded = false
580 }
581 // A job kept from an earlier attempt did not start in this one.
582 if j.Attempt == rs.attempt && !j.StartedAt.IsZero() && (started.IsZero() || j.StartedAt.Before(started)) {
583 started = j.StartedAt
584 }
585 switch {
586 case j.Conclusion.IsFailure() && j.ContinueOnError:
587 // continue-on-error: the job failed, the run does not.
588 case j.Conclusion.IsFailure():
589 failed = true
590 case j.Conclusion == domain.CICancelled:
591 cancelled = true
592 }
593 }
594 if run.StartedAt.IsZero() && !started.IsZero() {
595 run.StartedAt = started
596 }
597 switch {
598 case allEnded:
599 run.Status, run.CompletedAt = domain.CICompleted, now
600 switch {
601 case run.CancelRequested:
602 run.Conclusion = domain.CICancelled
603 case failed:
604 run.Conclusion = domain.CIFailure
605 case cancelled:
606 run.Conclusion = domain.CICancelled
607 default:
608 run.Conclusion = domain.CISuccess
609 }
610 if rs.run.Status == domain.CICompleted {
611 run.CompletedAt = rs.run.CompletedAt
612 }
613 case !run.StartedAt.IsZero():
614 run.Status, run.Conclusion, run.CompletedAt = domain.CIInProgress, domain.CINoConclusion, time.Time{}
615 default:
616 run.Status, run.Conclusion, run.CompletedAt = domain.CIQueued, domain.CINoConclusion, time.Time{}
617 }
618 return run
619}
620
621// --- access ---
622
623// readableRepo looks the repository up and applies the read rule.
624func (c *CI) readableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
625 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
626 if err != nil {
627 return domain.Repo{}, err
628 }
629 if !canRead(p, repo) {
630 return domain.Repo{}, domain.ErrNotFound
631 }
632 return repo, nil
633}
634
635// writableRepo is the access rule of cancelling and re-running: a principal
636// whose token holds ci:write on the repository sees the repository for this
637// action even without repo:read, as for merge requests.
638func (c *CI) writableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
639 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
640 if err != nil {
641 return domain.Repo{}, err
642 }
643 if !canRead(p, repo) && !p.Allows(domain.ScopeCIWrite, name) {
644 return domain.Repo{}, domain.ErrNotFound
645 }
646 if err := scopeAccess(p, repo, domain.ScopeCIWrite); err != nil {
647 return domain.Repo{}, err
648 }
649 return repo, nil
650}
651
652// runWriteAccess is the branch rule of the write actions on a run, cancel and
653// re-run, which come after the ci:write check. A run is tied to the ref that
654// made it, and re-running it executes the workflow of that ref again while
655// cancelling it stops the checks of that ref. So, like a push, a merge or a
656// branch delete, these actions fall under the branch prefixes of a limited
657// token (ADR 0027): it may act only on the runs of branches under its prefixes,
658// and never on a run of a tag or of any other ref. Without prefixes, every run.
659//
660// Reading runs, jobs, logs and checks is not a write: it needs only the access
661// to read the repository, as for the code, whatever the prefixes. The caller
662// has seen the run by then, so the refusal is forbidden, not not found.
663func runWriteAccess(p domain.Principal, run domain.CIRun) error {
664 if !p.AllowsRef(run.Ref) {
665 return fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, run.Ref)
666 }
667 return nil
668}
669
670// canManageRunners is the access rule of runners: runner:register on a token
671// that is not restricted to repositories (a runner serves every repository),
672// or root. A token limited to branch prefixes never manages runners, whatever
673// scopes it holds: a runner runs the workflows of every branch of every
674// repository, with their code and secrets, so registering or removing one
675// reaches far beyond the prefixes (a replaced runner could also be a way to
676// take over the jobs of main). Like repository administration and owner
677// rights, it is out of reach of a limited token.
678func canManageRunners(p domain.Principal) error {
679 switch {
680 case p.IsAnonymous():
681 return domain.ErrUnauthenticated
682 case !p.HasScope(domain.ScopeRunnerRegister):
683 return fmt.Errorf("%w: this needs the %s scope", domain.ErrForbidden, domain.ScopeRunnerRegister)
684 case len(p.Repos()) > 0:
685 return fmt.Errorf("%w: a runner serves every repository: the token must not be restricted to some", domain.ErrForbidden)
686 case p.IsBranchLimited():
687 return fmt.Errorf("%w: a token limited to branch prefixes may not manage runners", domain.ErrForbidden)
688 }
689 return nil
690}
691
692// --- small helpers ---
693
694func truncateRunes(s string, n int) string {
695 if len(s) <= n {
696 return s
697 }
698 r := []rune(s)
699 if len(r) <= n {
700 return s
701 }
702 return string(r[:n])
703}
704
705// cutText cuts s to n bytes at a character boundary.
706func cutText(s string, n int) string {
707 if len(s) <= n {
708 return s
709 }
710 for n > 0 && n < len(s) && s[n]&0xC0 == 0x80 {
711 n--
712 }
713 return s[:n]
714}
715
716func sortedKeys[V any](m map[string]V) []string {
717 keys := make([]string, 0, len(m))
718 for k := range m {
719 keys = append(keys, k)
720 }
721 slices.Sort(keys)
722 return keys
723}
724
725// ctxLog returns the logger of the context, addressable so that the level
726// methods can be chained on it.
727func ctxLog(ctx context.Context) *zerowrap.Logger {
728 l := zerowrap.FromCtx(ctx)
729 return &l
730}
731
732// logRun returns the logger of the context with the fields of the run.
733func logRun(ctx context.Context, run domain.CIRun) zerowrap.Logger {
734 return zerowrap.FromCtx(ctx).WithFields(map[string]any{"repo": run.Repo.String(), "run_id": run.ID})
735}
736
737func isHex40(s string) bool {
738 if len(s) != 40 {
739 return false
740 }
741 return strings.Trim(s, "0123456789abcdef") == ""
742}