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