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