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