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