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