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