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():
402 failed = true
403 case j.Conclusion == domain.CICancelled:
404 cancelled = true
405 }
406 }
407 if run.StartedAt.IsZero() && !started.IsZero() {
408 run.StartedAt = started
409 }
410 switch {
411 case allEnded:
412 run.Status, run.CompletedAt = domain.CICompleted, now
413 switch {
414 case run.CancelRequested:
415 run.Conclusion = domain.CICancelled
416 case failed:
417 run.Conclusion = domain.CIFailure
418 case cancelled:
419 run.Conclusion = domain.CICancelled
420 default:
421 run.Conclusion = domain.CISuccess
422 }
423 if rs.run.Status == domain.CICompleted {
424 run.CompletedAt = rs.run.CompletedAt
425 }
426 case !run.StartedAt.IsZero():
427 run.Status, run.Conclusion, run.CompletedAt = domain.CIInProgress, domain.CINoConclusion, time.Time{}
428 default:
429 run.Status, run.Conclusion, run.CompletedAt = domain.CIQueued, domain.CINoConclusion, time.Time{}
430 }
431 return run
432}
433
434// --- access ---
435
436// readableRepo looks the repository up and applies the read rule.
437func (c *CI) readableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
438 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
439 if err != nil {
440 return domain.Repo{}, err
441 }
442 if !canRead(p, repo) {
443 return domain.Repo{}, domain.ErrNotFound
444 }
445 return repo, nil
446}
447
448// writableRepo is the access rule of cancelling and re-running: a principal
449// whose token holds ci:write on the repository sees the repository for this
450// action even without repo:read, as for merge requests.
451func (c *CI) writableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
452 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
453 if err != nil {
454 return domain.Repo{}, err
455 }
456 if !canRead(p, repo) && !p.Allows(domain.ScopeCIWrite, name) {
457 return domain.Repo{}, domain.ErrNotFound
458 }
459 if err := scopeAccess(p, repo, domain.ScopeCIWrite); err != nil {
460 return domain.Repo{}, err
461 }
462 return repo, nil
463}
464
465// runWriteAccess is the branch rule of the write actions on a run, cancel and
466// re-run, which come after the ci:write check. A run is tied to the ref that
467// made it, and re-running it executes the workflow of that ref again while
468// cancelling it stops the checks of that ref. So, like a push, a merge or a
469// branch delete, these actions fall under the branch prefixes of a limited
470// token (ADR 0027): it may act only on the runs of branches under its prefixes,
471// and never on a run of a tag or of any other ref. Without prefixes, every run.
472//
473// Reading runs, jobs, logs and checks is not a write: it needs only the access
474// to read the repository, as for the code, whatever the prefixes. The caller
475// has seen the run by then, so the refusal is forbidden, not not found.
476func runWriteAccess(p domain.Principal, run domain.CIRun) error {
477 if !p.AllowsRef(run.Ref) {
478 return fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, run.Ref)
479 }
480 return nil
481}
482
483// canManageRunners is the access rule of runners: runner:register on a token
484// that is not restricted to repositories (a runner serves every repository),
485// or root. A token limited to branch prefixes never manages runners, whatever
486// scopes it holds: a runner runs the workflows of every branch of every
487// repository, with their code and secrets, so registering or removing one
488// reaches far beyond the prefixes (a replaced runner could also be a way to
489// take over the jobs of main). Like repository administration and owner
490// rights, it is out of reach of a limited token.
491func canManageRunners(p domain.Principal) error {
492 switch {
493 case p.IsAnonymous():
494 return domain.ErrUnauthenticated
495 case !p.HasScope(domain.ScopeRunnerRegister):
496 return fmt.Errorf("%w: this needs the %s scope", domain.ErrForbidden, domain.ScopeRunnerRegister)
497 case len(p.Repos()) > 0:
498 return fmt.Errorf("%w: a runner serves every repository: the token must not be restricted to some", domain.ErrForbidden)
499 case p.IsBranchLimited():
500 return fmt.Errorf("%w: a token limited to branch prefixes may not manage runners", domain.ErrForbidden)
501 }
502 return nil
503}
504
505// --- small helpers ---
506
507func truncateRunes(s string, n int) string {
508 if len(s) <= n {
509 return s
510 }
511 r := []rune(s)
512 if len(r) <= n {
513 return s
514 }
515 return string(r[:n])
516}
517
518// cutText cuts s to n bytes at a character boundary.
519func cutText(s string, n int) string {
520 if len(s) <= n {
521 return s
522 }
523 for n > 0 && n < len(s) && s[n]&0xC0 == 0x80 {
524 n--
525 }
526 return s[:n]
527}
528
529func sortedKeys[V any](m map[string]V) []string {
530 keys := make([]string, 0, len(m))
531 for k := range m {
532 keys = append(keys, k)
533 }
534 slices.Sort(keys)
535 return keys
536}
537
538// ctxLog returns the logger of the context, addressable so that the level
539// methods can be chained on it.
540func ctxLog(ctx context.Context) *zerowrap.Logger {
541 l := zerowrap.FromCtx(ctx)
542 return &l
543}
544
545// logRun returns the logger of the context with the fields of the run.
546func logRun(ctx context.Context, run domain.CIRun) zerowrap.Logger {
547 return zerowrap.FromCtx(ctx).WithFields(map[string]any{"repo": run.Repo.String(), "run_id": run.ID})
548}
549
550func isHex40(s string) bool {
551 if len(s) != 40 {
552 return false
553 }
554 return strings.Trim(s, "0123456789abcdef") == ""
555}