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 // MaxCIChecks is the most checks ListChecks returns.
38 MaxCIChecks = 1000
39 // MaxCILogStreams and MaxCILogStreamsPerClient bound the concurrent log
40 // streams (GetJobLog).
41 MaxCILogStreams = 64
42 MaxCILogStreamsPerClient = 8
43 // maxCIWorkflowBytes is the largest workflow file (the parser's limit).
44 maxCIWorkflowBytes = 1 << 20
45 // maxCIWorkflowFiles bounds the workflow files read per push.
46 maxCIWorkflowFiles = 256
47)
48
49// CIConfig holds what the CI use cases need. Store, Logs, Summaries, Meta, Repos, Reader,
50// Git, Diff, Bus, Clock, Owner and PublicURL are required (the config supplies
51// the defaults of Owner and Retention).
52type CIConfig struct {
53 Store ports.CIStore
54 Logs ports.CILogStore
55 // Summaries keeps the job summaries, deleted with the logs of their run.
56 Summaries ports.CISummaryStore
57 Meta ports.RepoMetaStore
58 Repos ports.GitRepos
59 Reader ports.RepoReader
60 Git ports.MergeGit
61 Diff ports.CIGit
62 Bus ports.EventBus
63 Clock ports.Clock
64 // Auth re-validates the token of a log stream; nil skips it.
65 Auth ports.Revalidator
66 // Remote lists the refs of github.com repositories for ResolveActions; nil
67 // answers every github.com action with unavailable.
68 Remote ports.GitRemote
69 // Archive serves the tarballs of the REST subset.
70 Archive ports.GitArchive
71 // Tokens finds the stored token of the pusher of a run, to ask what it may
72 // read when an action of another repository is resolved for the run. Nil
73 // reads nothing for a token (fail closed).
74 Tokens ports.TokenStore
75 // Settings resolves the variables and secrets of a job (CI-3). Nil gives
76 // jobs none, and an empty "vars" context. Leave it unset rather than
77 // assigning a typed nil pointer.
78 Settings ports.CIJobSettings
79 // RemoteConcurrency bounds the git ls-remote processes at once (default
80 // DefaultCIRemoteConcurrency); ResolveTimeout bounds one ResolveActions call
81 // (default DefaultCIResolveTimeout).
82 RemoteConcurrency int
83 ResolveTimeout time.Duration
84 // Owner is the owner login of the github context.
85 Owner string
86 // PublicURL is the base URL of the instance (github.server_url).
87 PublicURL string
88 // Retention is how long runs, jobs, checks and logs are kept.
89 Retention time.Duration
90
91 // The durations below default to the contract's; tests shorten them.
92 PollTimeout time.Duration
93 Tick time.Duration
94 Keepalive time.Duration
95 AckTimeout time.Duration
96 HeartbeatDeadline time.Duration
97 CancelGrace time.Duration
98 TimeoutSlack time.Duration
99 QueueTimeout time.Duration
100 TriggerQueue int
101}
102
103func (c CIConfig) withDefaults() CIConfig {
104 def := func(d *time.Duration, v time.Duration) {
105 if *d <= 0 {
106 *d = v
107 }
108 }
109 def(&c.PollTimeout, DefaultCIPollTimeout)
110 def(&c.Tick, DefaultCITick)
111 def(&c.Keepalive, DefaultCIKeepalive)
112 def(&c.AckTimeout, domain.CIAckTimeout)
113 def(&c.HeartbeatDeadline, domain.CIHeartbeatDeadline)
114 def(&c.CancelGrace, domain.CICancelGrace)
115 def(&c.TimeoutSlack, domain.CITimeoutSlack)
116 def(&c.QueueTimeout, domain.CIQueueTimeout)
117 def(&c.Retention, domain.CIRetentionDays*24*time.Hour)
118 if c.RemoteConcurrency <= 0 {
119 c.RemoteConcurrency = DefaultCIRemoteConcurrency
120 }
121 def(&c.ResolveTimeout, DefaultCIResolveTimeout)
122 if c.TriggerQueue <= 0 {
123 c.TriggerQueue = DefaultCITriggerQueue
124 }
125 return c
126}
127
128// CI is the CI use cases: the workflow trigger, the planner, the assignment of
129// jobs to runners, the runner protocol and the owner side (ADR 0025).
130//
131// Every change of a run goes through its lock (withRun): the records of a run
132// are read, changed and written again by one goroutine at a time, so that two
133// calls (a runner's update and an owner's cancel) never lose one another's
134// write. Assignment takes the assignment lock before the lock of the run.
135type CI struct {
136 cfg CIConfig
137 boot time.Time
138 queue chan ciTrigger
139
140 stop context.CancelFunc
141 ctx context.Context
142 wg sync.WaitGroup
143
144 remoteSlots chan struct{} // slots of the git ls-remote processes
145 locks sync.Map // run id -> *sync.Mutex
146 assign sync.Mutex
147 jobs notifier // woken when a job changes: key is the job reference
148 pending notifier // woken when a job may be ready or a runner was removed
149 flows *wfCache
150
151 mu sync.Mutex
152 heartbeats map[string]time.Time // job reference -> last heartbeat
153 seen map[string]time.Time // runner id -> last seen
154 polls map[string]pollRecord
155 // archiveTokens are the secrets of the tokens given for the archives of
156 // ermite actions, per job reference then "<repository>@<sha>", so that a job
157 // that resolves the same commit twice gets one token (memory only: a
158 // restart issues new ones).
159 archiveTokens map[string]map[string]string
160 streams int
161 perClient map[string]int
162 closed bool
163}
164
165var (
166 _ ports.RunnerService = (*CI)(nil)
167 _ ports.RunService = (*CI)(nil)
168 _ ports.WorkflowTrigger = (*CI)(nil)
169 _ ports.CIChecksReader = (*CI)(nil)
170)
171
172// pollRecord remembers the assignment a poll made, for the retry of a lost
173// response (FetchJobRequest.poll_id). The token is kept in memory only.
174type pollRecord struct {
175 pollID string
176 ref string
177 token string
178 ghToken string
179 hash domain.TokenHash
180}
181
182// NewCI builds the CI use cases. ctx carries the logger; its cancellation is
183// not followed: call Start to run the workers and Close to stop them.
184func NewCI(ctx context.Context, cfg CIConfig) *CI {
185 if cfg.Owner == "" {
186 panic("usecase: CIConfig.Owner is required")
187 }
188 cfg = cfg.withDefaults()
189 run, stop := context.WithCancel(context.WithoutCancel(ctx))
190 return &CI{
191 cfg: cfg, boot: cfg.Clock.Now(), queue: make(chan ciTrigger, cfg.TriggerQueue),
192 ctx: run, stop: stop, flows: newWfCache(64), remoteSlots: make(chan struct{}, cfg.RemoteConcurrency),
193 heartbeats: map[string]time.Time{}, seen: map[string]time.Time{}, polls: map[string]pollRecord{},
194 archiveTokens: map[string]map[string]string{},
195 perClient: map[string]int{},
196 }
197}
198
199// Start runs the trigger worker and the watchdog, and plans again the runs
200// that were open when the server stopped.
201func (c *CI) Start() {
202 c.wg.Add(2)
203 go c.triggerLoop()
204 go c.watchdog()
205 c.wg.Add(1)
206 go func() {
207 defer c.wg.Done()
208 c.resume()
209 }()
210}
211
212// Close stops the workers and waits for them.
213func (c *CI) Close(_ context.Context) {
214 c.mu.Lock()
215 c.closed = true
216 c.mu.Unlock()
217 c.stop()
218 c.wg.Wait()
219}
220
221func (c *CI) now() time.Time { return c.cfg.Clock.Now().UTC() }
222
223// --- identifiers and secrets ---
224
225func newToken(prefix string) string {
226 var raw [32]byte
227 _, _ = rand.Read(raw[:])
228 return prefix + base64.RawURLEncoding.EncodeToString(raw[:])
229}
230
231func newRunnerID() string {
232 var raw [8]byte
233 _, _ = rand.Read(raw[:])
234 return hex.EncodeToString(raw[:])
235}
236
237func jobRef(j domain.CIJob) string { return fmt.Sprintf("%d/%d/%s", j.RunID, j.RunAttempt, j.ID) }
238
239// --- locks and notifications ---
240
241// notifier wakes the goroutines waiting on a key. A channel exists only while
242// somebody waits on it.
243type notifier struct {
244 mu sync.Mutex
245 ch map[string]chan struct{}
246}
247
248// wait returns a channel closed by the next notify of key.
249func (n *notifier) wait(key string) <-chan struct{} {
250 n.mu.Lock()
251 defer n.mu.Unlock()
252 if n.ch == nil {
253 n.ch = map[string]chan struct{}{}
254 }
255 c, ok := n.ch[key]
256 if !ok {
257 c = make(chan struct{})
258 n.ch[key] = c
259 }
260 return c
261}
262
263func (n *notifier) notify(key string) {
264 n.mu.Lock()
265 defer n.mu.Unlock()
266 if c, ok := n.ch[key]; ok {
267 close(c)
268 delete(n.ch, key)
269 }
270}
271
272const queueKey = "q"
273
274func (c *CI) wakeQueue() { c.pending.notify(queueKey) }
275
276func (c *CI) runLock(runID int64) *sync.Mutex {
277 m, _ := c.locks.LoadOrStore(runID, &sync.Mutex{})
278 return m.(*sync.Mutex)
279}
280
281// --- run state ---
282
283// runState is a run being changed under its lock: the run, the jobs of one of
284// its attempts, and what must be written. flush writes everything in one batch
285// and publishes the events.
286type runState struct {
287 c *CI
288 run domain.CIRun
289 attempt int32
290 jobs []domain.CIJob
291 src *domain.CIRunSource
292
293 runDirty bool
294 dirty map[string]bool
295 // chk marks jobs whose check must be written even when the job itself did
296 // not change (a rerun moves every check to the new attempt).
297 checks map[string]bool
298 // tokens are the job repository tokens to create, revoke the jobs whose
299 // tokens are deleted, with the next flush.
300 tokens []domain.CIRepoToken
301 revoke []ports.CIJobRef
302 // resolutions are the action resolutions to keep with the next flush.
303 resolutions []domain.CIActionResolution
304 // annotations replace the annotations of steps, clear drops all those of
305 // executions of jobs, with the next flush.
306 annotations []ports.CIStepAnnotations
307 clear []ports.CILogRef
308}
309
310// withRun locks the run, loads it with the jobs of attempt (0 for the latest)
311// and calls fn. The run is found by id alone through its source.
312func (c *CI) withRun(ctx context.Context, runID int64, attempt int32, fn func(rs *runState) error) error {
313 lock := c.runLock(runID)
314 lock.Lock()
315 defer lock.Unlock()
316 rs, err := c.loadRun(ctx, runID, attempt)
317 if err != nil {
318 return err
319 }
320 return fn(rs)
321}
322
323func (c *CI) loadRun(ctx context.Context, runID int64, attempt int32) (*runState, error) {
324 src, err := c.cfg.Store.GetRunSource(ctx, runID)
325 if err != nil {
326 return nil, err
327 }
328 run, err := c.cfg.Store.GetRun(ctx, src.Repo, runID)
329 if err != nil {
330 return nil, err
331 }
332 if attempt <= 0 {
333 attempt = run.Attempt
334 }
335 jobs, err := c.cfg.Store.ListJobs(ctx, runID, attempt)
336 if err != nil {
337 return nil, err
338 }
339 return &runState{c: c, run: run, attempt: attempt, jobs: jobs, src: &src, dirty: map[string]bool{}, checks: map[string]bool{}}, nil
340}
341
342// latest reports whether the state holds the latest attempt of the run.
343func (rs *runState) latest() bool { return rs.attempt == rs.run.Attempt }
344
345func (rs *runState) find(id string) *domain.CIJob {
346 for i := range rs.jobs {
347 if rs.jobs[i].ID == id {
348 return &rs.jobs[i]
349 }
350 }
351 return nil
352}
353
354func (rs *runState) touch(j *domain.CIJob) { rs.dirty[j.ID] = true }
355
356// checkOf builds the check of a job.
357func (rs *runState) checkOf(j domain.CIJob) domain.CICheck {
358 return domain.CICheck{
359 ID: j.ID, Repo: rs.run.Repo, Sha: rs.run.Sha, Name: j.Name, Status: j.Status(), Conclusion: j.Conclusion,
360 RunID: rs.run.ID, JobID: j.ID, Attempt: rs.run.Attempt, WorkflowName: rs.run.WorkflowName,
361 CreatedAt: j.CreatedAt, StartedAt: j.StartedAt, CompletedAt: j.CompletedAt,
362 }
363}
364
365// flush recomputes the status of the run, writes the changed jobs, their
366// checks and the run in one batch, and publishes RunUpdated and JobUpdated.
367func (rs *runState) flush(ctx context.Context) error {
368 c := rs.c
369 now := c.now()
370 var w ports.CIWrite
371 var changedJobs []int
372 for i := range rs.jobs {
373 if rs.dirty[rs.jobs[i].ID] {
374 w.Jobs = append(w.Jobs, rs.jobs[i])
375 changedJobs = append(changedJobs, i)
376 }
377 }
378 if rs.latest() {
379 for i := range rs.jobs {
380 if rs.dirty[rs.jobs[i].ID] || rs.checks[rs.jobs[i].ID] {
381 w.Checks = append(w.Checks, rs.checkOf(rs.jobs[i]))
382 }
383 }
384 if next := rs.recompute(now); next != rs.run {
385 rs.run, rs.runDirty = next, true
386 }
387 }
388 if rs.runDirty {
389 w.Runs = []domain.CIRun{rs.run}
390 }
391 w.RepoTokens, w.RevokeJobs, w.Resolutions = rs.tokens, rs.revoke, rs.resolutions
392 w.Annotations, w.ClearAnnotations = rs.annotations, rs.clear
393 if len(w.Jobs) == 0 && len(w.Runs) == 0 && len(w.Checks) == 0 && len(w.RepoTokens) == 0 && len(w.RevokeJobs) == 0 &&
394 len(w.Resolutions) == 0 && len(w.Annotations) == 0 && len(w.ClearAnnotations) == 0 {
395 return nil
396 }
397 v, err := c.cfg.Store.Apply(ctx, w)
398 if err != nil {
399 return fmt.Errorf("write run %d: %w", rs.run.ID, err)
400 }
401 for _, i := range changedJobs {
402 rs.jobs[i].Version = v
403 }
404 if rs.runDirty {
405 rs.run.Version = v
406 }
407 changedRun := rs.runDirty
408 rs.dirty, rs.checks, rs.runDirty = map[string]bool{}, map[string]bool{}, false
409 rs.tokens, rs.revoke, rs.resolutions = nil, nil, nil
410 rs.annotations, rs.clear = nil, nil
411
412 repo, err := c.cfg.Meta.Get(ctx, rs.run.Repo)
413 if err != nil {
414 for _, i := range changedJobs {
415 c.jobs.notify(jobRef(rs.jobs[i]))
416 }
417 return nil // the repository is gone: nobody to tell
418 }
419 for _, i := range changedJobs {
420 j := rs.jobs[i]
421 lines, _ := c.cfg.Logs.Count(ctx, ports.CILogRef{RunID: j.RunID, JobID: j.ID, Attempt: j.Attempt})
422 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, Lines: lines, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
423 c.jobs.notify(jobRef(j))
424 if j.Phase == domain.CIPhaseReady {
425 c.wakeQueue()
426 }
427 }
428 if changedRun {
429 c.cfg.Bus.Publish(ctx, domain.RunUpdated{Run: rs.run, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
430 }
431 return nil
432}
433
434// recompute returns the run with the status, the conclusion and the
435// timestamps its jobs imply.
436func (rs *runState) recompute(now time.Time) domain.CIRun {
437 run := rs.run
438 if run.Status == domain.CICompleted && len(rs.jobs) == 0 {
439 return run // an invalid workflow file: failed from the start
440 }
441 allEnded := len(rs.jobs) > 0
442 started := time.Time{}
443 var failed, cancelled bool
444 for _, j := range rs.jobs {
445 if !j.Ended() {
446 allEnded = false
447 }
448 // A job kept from an earlier attempt did not start in this one.
449 if j.Attempt == rs.attempt && !j.StartedAt.IsZero() && (started.IsZero() || j.StartedAt.Before(started)) {
450 started = j.StartedAt
451 }
452 switch {
453 case j.Conclusion.IsFailure() && j.ContinueOnError:
454 // continue-on-error: the job failed, the run does not.
455 case j.Conclusion.IsFailure():
456 failed = true
457 case j.Conclusion == domain.CICancelled:
458 cancelled = true
459 }
460 }
461 if run.StartedAt.IsZero() && !started.IsZero() {
462 run.StartedAt = started
463 }
464 switch {
465 case allEnded:
466 run.Status, run.CompletedAt = domain.CICompleted, now
467 switch {
468 case run.CancelRequested:
469 run.Conclusion = domain.CICancelled
470 case failed:
471 run.Conclusion = domain.CIFailure
472 case cancelled:
473 run.Conclusion = domain.CICancelled
474 default:
475 run.Conclusion = domain.CISuccess
476 }
477 if rs.run.Status == domain.CICompleted {
478 run.CompletedAt = rs.run.CompletedAt
479 }
480 case !run.StartedAt.IsZero():
481 run.Status, run.Conclusion, run.CompletedAt = domain.CIInProgress, domain.CINoConclusion, time.Time{}
482 default:
483 run.Status, run.Conclusion, run.CompletedAt = domain.CIQueued, domain.CINoConclusion, time.Time{}
484 }
485 return run
486}
487
488// --- access ---
489
490// readableRepo looks the repository up and applies the read rule.
491func (c *CI) readableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
492 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
493 if err != nil {
494 return domain.Repo{}, err
495 }
496 if !canRead(p, repo) {
497 return domain.Repo{}, domain.ErrNotFound
498 }
499 return repo, nil
500}
501
502// writableRepo is the access rule of cancelling and re-running: a principal
503// whose token holds ci:write on the repository sees the repository for this
504// action even without repo:read, as for merge requests.
505func (c *CI) writableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
506 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
507 if err != nil {
508 return domain.Repo{}, err
509 }
510 if !canRead(p, repo) && !p.Allows(domain.ScopeCIWrite, name) {
511 return domain.Repo{}, domain.ErrNotFound
512 }
513 if err := scopeAccess(p, repo, domain.ScopeCIWrite); err != nil {
514 return domain.Repo{}, err
515 }
516 return repo, nil
517}
518
519// runWriteAccess is the branch rule of the write actions on a run, cancel and
520// re-run, which come after the ci:write check. A run is tied to the ref that
521// made it, and re-running it executes the workflow of that ref again while
522// cancelling it stops the checks of that ref. So, like a push, a merge or a
523// branch delete, these actions fall under the branch prefixes of a limited
524// token (ADR 0027): it may act only on the runs of branches under its prefixes,
525// and never on a run of a tag or of any other ref. Without prefixes, every run.
526//
527// Reading runs, jobs, logs and checks is not a write: it needs only the access
528// to read the repository, as for the code, whatever the prefixes. The caller
529// has seen the run by then, so the refusal is forbidden, not not found.
530func runWriteAccess(p domain.Principal, run domain.CIRun) error {
531 if !p.AllowsRef(run.Ref) {
532 return fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, run.Ref)
533 }
534 return nil
535}
536
537// canManageRunners is the access rule of runners: runner:register on a token
538// that is not restricted to repositories (a runner serves every repository),
539// or root. A token limited to branch prefixes never manages runners, whatever
540// scopes it holds: a runner runs the workflows of every branch of every
541// repository, with their code and secrets, so registering or removing one
542// reaches far beyond the prefixes (a replaced runner could also be a way to
543// take over the jobs of main). Like repository administration and owner
544// rights, it is out of reach of a limited token.
545func canManageRunners(p domain.Principal) error {
546 switch {
547 case p.IsAnonymous():
548 return domain.ErrUnauthenticated
549 case !p.HasScope(domain.ScopeRunnerRegister):
550 return fmt.Errorf("%w: this needs the %s scope", domain.ErrForbidden, domain.ScopeRunnerRegister)
551 case len(p.Repos()) > 0:
552 return fmt.Errorf("%w: a runner serves every repository: the token must not be restricted to some", domain.ErrForbidden)
553 case p.IsBranchLimited():
554 return fmt.Errorf("%w: a token limited to branch prefixes may not manage runners", domain.ErrForbidden)
555 }
556 return nil
557}
558
559// --- small helpers ---
560
561func truncateRunes(s string, n int) string {
562 if len(s) <= n {
563 return s
564 }
565 r := []rune(s)
566 if len(r) <= n {
567 return s
568 }
569 return string(r[:n])
570}
571
572// cutText cuts s to n bytes at a character boundary.
573func cutText(s string, n int) string {
574 if len(s) <= n {
575 return s
576 }
577 for n > 0 && n < len(s) && s[n]&0xC0 == 0x80 {
578 n--
579 }
580 return s[:n]
581}
582
583func sortedKeys[V any](m map[string]V) []string {
584 keys := make([]string, 0, len(m))
585 for k := range m {
586 keys = append(keys, k)
587 }
588 slices.Sort(keys)
589 return keys
590}
591
592// ctxLog returns the logger of the context, addressable so that the level
593// methods can be chained on it.
594func ctxLog(ctx context.Context) *zerowrap.Logger {
595 l := zerowrap.FromCtx(ctx)
596 return &l
597}
598
599// logRun returns the logger of the context with the fields of the run.
600func logRun(ctx context.Context, run domain.CIRun) zerowrap.Logger {
601 return zerowrap.FromCtx(ctx).WithFields(map[string]any{"repo": run.Repo.String(), "run_id": run.ID})
602}
603
604func isHex40(s string) bool {
605 if len(s) != 40 {
606 return false
607 }
608 return strings.Trim(s, "0123456789abcdef") == ""
609}