1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "encoding/base64"
8 "errors"
9 "fmt"
10 "path"
11 "slices"
12 "strconv"
13 "strings"
14 "time"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20const runTokenPrefix = "ru1:"
21
22// --- runs and jobs ---
23
24// ListRuns returns a page of the runs of a repository, newest first. The
25// principal needs the access of reading the repository.
26func (c *CI) ListRuns(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.ListRunsParams) (ports.RunPage, error) {
27 if _, err := c.readableRepo(ctx, p, name); err != nil {
28 return ports.RunPage{}, err
29 }
30 n, err := pageSize(params.PageSize)
31 if err != nil {
32 return ports.RunPage{}, err
33 }
34 var before int64
35 if params.PageToken != "" {
36 raw, ok := decodeToken(params.PageToken, runTokenPrefix)
37 id, perr := strconv.ParseInt(raw, 10, 64)
38 if !ok || perr != nil || id < 1 {
39 return ports.RunPage{}, fmt.Errorf("%w: runs", domain.ErrInvalidPageToken)
40 }
41 before = id
42 }
43 match := runFilter(params)
44 // The store returns runs by id; the filters are applied while paging, so a
45 // page of a filtered listing is full unless the runs ran out.
46 var kept []domain.CIRun
47 cursor := before
48 for len(kept) <= n {
49 batch, err := c.cfg.Store.ListRuns(ctx, name, cursor, n+1)
50 if err != nil {
51 return ports.RunPage{}, err
52 }
53 for _, r := range batch {
54 if match(r) {
55 kept = append(kept, r)
56 }
57 }
58 if len(batch) < n+1 {
59 break
60 }
61 cursor = batch[len(batch)-1].ID
62 }
63 page := ports.RunPage{Runs: kept}
64 if len(kept) > n {
65 page.Runs = kept[:n]
66 page.NextPageToken = base64.RawURLEncoding.EncodeToString([]byte(runTokenPrefix + strconv.FormatInt(kept[n-1].ID, 10)))
67 }
68 return page, nil
69}
70
71// runFilter returns the test of the filters of a listing.
72func runFilter(p ports.ListRunsParams) func(domain.CIRun) bool {
73 return func(r domain.CIRun) bool {
74 switch {
75 case p.Status != "" && r.Status != p.Status:
76 return false
77 case p.Branch != "" && r.Ref != "refs/heads/"+p.Branch:
78 return false
79 case p.Workflow != "" && r.WorkflowPath != p.Workflow && path.Base(r.WorkflowPath) != p.Workflow:
80 return false
81 }
82 return true
83 }
84}
85
86// runOf finds a run of the repository; a run of another repository is not
87// found.
88func (c *CI) runOf(ctx context.Context, name domain.RepoName, runID int64) (domain.CIRun, error) {
89 run, err := c.cfg.Store.GetRun(ctx, name, runID)
90 if err != nil {
91 return domain.CIRun{}, err
92 }
93 return run, nil
94}
95
96// view adds the live line count to a job.
97func (c *CI) view(ctx context.Context, j domain.CIJob) ports.JobView {
98 v := ports.JobView{CIJob: j}
99 if j.Phase != domain.CIPhaseWaiting && j.Phase != domain.CIPhaseReady {
100 v.Lines, _ = c.cfg.Logs.Count(ctx, logRef(j))
101 }
102 if j.RunnerID != "" {
103 if _, err := c.cfg.Store.GetRunner(ctx, j.RunnerID); errors.Is(err, domain.ErrNotFound) {
104 v.RunnerGone = true
105 }
106 }
107 return v
108}
109
110// GetRun returns a run with the jobs of an attempt (0 for the latest).
111func (c *CI) GetRun(ctx context.Context, p domain.Principal, name domain.RepoName, runID int64, attempt int32) (ports.RunDetail, error) {
112 if _, err := c.readableRepo(ctx, p, name); err != nil {
113 return ports.RunDetail{}, err
114 }
115 run, err := c.runOf(ctx, name, runID)
116 if err != nil {
117 return ports.RunDetail{}, err
118 }
119 latest := run.Attempt
120 if attempt < 0 || attempt > latest {
121 return ports.RunDetail{}, domain.ErrNotFound
122 }
123 if attempt == 0 {
124 attempt = latest
125 }
126 jobs, err := c.cfg.Store.ListJobs(ctx, runID, attempt)
127 if err != nil {
128 return ports.RunDetail{}, err
129 }
130 if attempt != latest {
131 run = c.runAtAttempt(run, jobs, attempt)
132 }
133 d := ports.RunDetail{Run: run, LatestAttempt: latest}
134 for _, j := range jobs {
135 d.Jobs = append(d.Jobs, c.view(ctx, j))
136 }
137 return d, nil
138}
139
140// runAtAttempt describes an earlier attempt of a run from its jobs: the run
141// record holds the latest attempt only.
142func (c *CI) runAtAttempt(run domain.CIRun, jobs []domain.CIJob, attempt int32) domain.CIRun {
143 rs := &runState{c: c, run: run, jobs: jobs, attempt: attempt}
144 r := rs.recompute(c.now())
145 r.Attempt, r.Version = attempt, run.Version
146 r.CancelRequested = false
147 return r
148}
149
150// jobOf finds a job of the repository by its id and attempt (0 for the
151// latest). The attempt is the one of the run it was kept by, or executed in.
152func (c *CI) jobOf(ctx context.Context, name domain.RepoName, jobID string, attempt int32) (domain.CIRun, domain.CIJob, error) {
153 runID, _, _, ok := domain.ParseJobID(jobID)
154 if !ok {
155 return domain.CIRun{}, domain.CIJob{}, domain.ErrNotFound
156 }
157 run, err := c.runOf(ctx, name, runID)
158 if err != nil {
159 return domain.CIRun{}, domain.CIJob{}, err
160 }
161 if attempt < 0 || attempt > run.Attempt {
162 return domain.CIRun{}, domain.CIJob{}, domain.ErrNotFound
163 }
164 if attempt == 0 {
165 attempt = run.Attempt
166 }
167 j, err := c.cfg.Store.GetJob(ctx, runID, attempt, jobID)
168 if err != nil {
169 return domain.CIRun{}, domain.CIJob{}, err
170 }
171 return run, j, nil
172}
173
174// GetJob returns one job with its steps.
175func (c *CI) GetJob(ctx context.Context, p domain.Principal, name domain.RepoName, jobID string, attempt int32) (ports.JobView, error) {
176 if _, err := c.readableRepo(ctx, p, name); err != nil {
177 return ports.JobView{}, err
178 }
179 _, j, err := c.jobOf(ctx, name, jobID, attempt)
180 if err != nil {
181 return ports.JobView{}, err
182 }
183 return c.view(ctx, j), nil
184}
185
186// --- log streaming ---
187
188// acquireStream counts a log stream against the limits.
189func (c *CI) acquireStream(client string) (release func(), err error) {
190 c.mu.Lock()
191 defer c.mu.Unlock()
192 if c.streams >= MaxCILogStreams || (client != "" && c.perClient[client] >= MaxCILogStreamsPerClient) {
193 return nil, fmt.Errorf("%w: too many log streams", domain.ErrResourceExhausted)
194 }
195 c.streams++
196 if client != "" {
197 c.perClient[client]++
198 }
199 return func() {
200 c.mu.Lock()
201 defer c.mu.Unlock()
202 c.streams--
203 if client != "" {
204 if c.perClient[client]--; c.perClient[client] <= 0 {
205 delete(c.perClient, client)
206 }
207 }
208 }, nil
209}
210
211// GetJobLog streams the lines of a job log from an offset, in batches of at
212// most 1,000 lines and 64 KiB. With Follow it goes on while the job runs, and
213// ends once the job completed and every line was sent.
214func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.GetJobLogParams, send func(ports.LogBatch) error) error {
215 if _, err := c.readableRepo(ctx, p, name); err != nil {
216 return err
217 }
218 if params.Offset < 0 {
219 return fmt.Errorf("%w: negative offset", domain.ErrInvalid)
220 }
221 _, j, err := c.jobOf(ctx, name, params.JobID, params.Attempt)
222 if err != nil {
223 return err
224 }
225 release, err := c.acquireStream(params.Client)
226 if err != nil {
227 return err
228 }
229 defer release()
230 if params.Ready != nil {
231 params.Ready()
232 }
233 ref := logRef(j)
234 jref := jobRef(j)
235 offset := params.Offset
236 checked := time.Now()
237 // revalidate asks again whether the token of a stored-token caller still
238 // exists, at most every few seconds, whether the stream is busy or idle: a
239 // token revoked while a long log is sent stops the stream.
240 revalidate := func() error {
241 if p.TokenID() == "" || c.cfg.Auth == nil || time.Since(checked) <= DefaultRevalidateEvery {
242 return nil
243 }
244 if err := c.cfg.Auth.Revalidate(ctx, p); err != nil {
245 return err
246 }
247 checked = time.Now()
248 return nil
249 }
250 for {
251 wakeJob, wakeLog := c.jobs.wait(jref), c.cfg.Logs.Changed(ref)
252 cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
253 if err != nil {
254 return err
255 }
256 done := cur.Ended()
257 lines, err := c.cfg.Logs.Read(ctx, ref, offset, domain.CIMaxLogLines, domain.CIMaxLogChunkBytes)
258 switch {
259 case errors.Is(err, domain.ErrNotFound), errors.Is(err, domain.ErrOutOfRange):
260 // No line yet (a queued job has no log), or fewer lines than the
261 // offset. Without follow, and for a job that ended (follow changes
262 // nothing then), that is the answer (contract: not_found,
263 // out_of_range). A job that runs may still write the line: wait.
264 if !params.Follow || done {
265 if errors.Is(err, domain.ErrNotFound) {
266 return domain.ErrNotFound
267 }
268 return fmt.Errorf("%w: the log has fewer than %d lines", domain.ErrOutOfRange, offset)
269 }
270 lines = nil
271 case err != nil:
272 return err
273 }
274 if len(lines) > 0 {
275 if err := revalidate(); err != nil {
276 return err
277 }
278 batch := ports.LogBatch{Offset: offset, Lines: c.withSteps(ctx, cur, offset, lines)}
279 if err := send(batch); err != nil {
280 return err
281 }
282 offset += int64(len(lines))
283 continue // more may be waiting
284 }
285 if !params.Follow || done {
286 return nil
287 }
288 if err := revalidate(); err != nil {
289 return err
290 }
291 select {
292 case <-wakeJob:
293 case <-wakeLog:
294 case <-time.After(DefaultRevalidateEvery):
295 case <-ctx.Done():
296 return nil
297 case <-c.ctx.Done():
298 return nil
299 }
300 }
301}
302
303// withSteps tags the lines of a batch with their step from the step ranges.
304func (c *CI) withSteps(_ context.Context, j domain.CIJob, offset int64, lines []domain.CILogLine) []ports.JobLogLine {
305 out := make([]ports.JobLogLine, len(lines))
306 for i, l := range lines {
307 out[i] = ports.JobLogLine{Step: domain.StepOfLine(j.Steps, offset+int64(i)), CILogLine: l}
308 }
309 return out
310}
311
312// --- cancel and rerun ---
313
314// CancelRun cancels the latest attempt of a run. The principal needs ci:write
315// on the repository. Jobs not given to a runner end at once; the others get a
316// cancel request, and the run completes when they have ended.
317func (c *CI) CancelRun(ctx context.Context, p domain.Principal, name domain.RepoName, runID int64) (domain.CIRun, error) {
318 if _, err := c.writableRepo(ctx, p, name); err != nil {
319 return domain.CIRun{}, err
320 }
321 run, err := c.runOf(ctx, name, runID)
322 if err != nil {
323 return domain.CIRun{}, err
324 }
325 if err := runWriteAccess(p, run); err != nil {
326 return domain.CIRun{}, err
327 }
328 var out domain.CIRun
329 err = c.withRun(ctx, runID, 0, func(rs *runState) error {
330 if rs.run.Status == domain.CICompleted {
331 return fmt.Errorf("%w: run %d has completed", domain.ErrPreconditionFailed, runID)
332 }
333 if !rs.run.CancelRequested {
334 rs.run.CancelRequested, rs.run.CancelRequestedAt, rs.runDirty = true, c.now(), true
335 rs.cancelAll("Canceled by " + domain.AuthorName(p))
336 }
337 if err := rs.flush(ctx); err != nil {
338 return err
339 }
340 out = rs.run
341 return nil
342 })
343 if err != nil {
344 return domain.CIRun{}, err
345 }
346 ctxLog(ctx).Info().Str("repo", name.String()).Int64("run_id", runID).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("ci run cancelled")
347 return out, nil
348}
349
350// RerunRun starts a new attempt of a completed run, from the stored workflow
351// file: all the jobs, or with failedOnly the jobs that did not succeed and
352// those that need them; the others are kept with their result, steps and log.
353func (c *CI) RerunRun(ctx context.Context, p domain.Principal, name domain.RepoName, runID int64, failedOnly bool) (domain.CIRun, error) {
354 repo, err := c.writableRepo(ctx, p, name)
355 if err != nil {
356 return domain.CIRun{}, err
357 }
358 run, err := c.runOf(ctx, name, runID)
359 if err != nil {
360 return domain.CIRun{}, err
361 }
362 if err := runWriteAccess(p, run); err != nil {
363 return domain.CIRun{}, err
364 }
365 err = c.withRun(ctx, runID, 0, func(rs *runState) error {
366 if rs.run.Status != domain.CICompleted {
367 return fmt.Errorf("%w: run %d has not completed", domain.ErrPreconditionFailed, runID)
368 }
369 if len(rs.jobs) == 0 {
370 return fmt.Errorf("%w: run %d has no job to run again", domain.ErrPreconditionFailed, runID)
371 }
372 redo, err := rerunSet(rs, failedOnly)
373 if err != nil {
374 return err
375 }
376 next := rs.run.Attempt + 1
377 now := c.now()
378 w := ports.CIWrite{}
379 var fresh []domain.CIJob
380 for _, j := range rs.jobs {
381 if !redo[rootKey(j.Key)] {
382 kept := j // kept with its own attempt, steps and log
383 kept.RunAttempt, kept.Version = next, 0
384 w.Jobs = append(w.Jobs, kept)
385 fresh = append(fresh, kept)
386 continue
387 }
388 if j.Combo > 0 || j.Scope != "" {
389 continue // the matrix, and the jobs of a call, are expanded again when the job is planned
390 }
391 n := domain.CIJob{
392 ID: domain.NewJobID(runID, j.Index, 0), RunID: runID, Attempt: next, RunAttempt: next, Repo: rs.run.Repo,
393 Key: j.Key, Name: j.Key, Phase: domain.CIPhaseWaiting, Needs: slices.Clone(j.Needs), Index: j.Index, CreatedAt: now,
394 }
395 w.Jobs = append(w.Jobs, n)
396 fresh = append(fresh, n)
397 }
398 if failedOnly {
399 // The callee of a reusable workflow stays at the commit the run first
400 // resolved, as the jobs that are kept ran from it; a re-run of
401 // everything resolves afresh (as for the actions of CI-2, which are
402 // resolved again in every attempt).
403 res, err := c.callResolutions(ctx, rs, next)
404 if err != nil {
405 return err
406 }
407 w.Resolutions = append(w.Resolutions, res...)
408 }
409 run := rs.run
410 run.Attempt, run.Status, run.Conclusion = next, domain.CIQueued, domain.CINoConclusion
411 run.StartedAt, run.CompletedAt, run.CreatedAt = time.Time{}, time.Time{}, now
412 run.CancelRequested, run.CancelRequestedAt = false, time.Time{}
413 run.Concurrency = domain.CIConcurrency{} // evaluated again: the re-run joins its group again
414 // Kept or lowered, never raised: the owner who runs an agent's run
415 // again does not give it secrets (D1).
416 run.Trusted = domain.RerunTrusted(rs.run.Trusted, p)
417 w.Runs = []domain.CIRun{run}
418 rs2 := &runState{c: c, run: run, attempt: next, jobs: fresh, src: rs.src, dirty: map[string]bool{}, checks: map[string]bool{}}
419 for _, j := range fresh {
420 w.Checks = append(w.Checks, rs2.checkOf(j))
421 }
422 v, err := c.cfg.Store.Apply(ctx, w)
423 if err != nil {
424 return err
425 }
426 run.Version = v
427 c.flows.drop(runID)
428 c.publishRun(ctx, repo, run)
429 for _, j := range w.Jobs {
430 j.Version = v
431 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
432 }
433 return nil
434 })
435 if err != nil {
436 return domain.CIRun{}, err
437 }
438 // Plan the new attempt: the kept jobs count as the results of their needs.
439 if err := c.advanceRun(ctx, runID); err != nil {
440 return domain.CIRun{}, err
441 }
442 run, err = c.runOf(ctx, name, runID)
443 if err != nil {
444 return domain.CIRun{}, err
445 }
446 ctxLog(ctx).Info().Str("repo", name.String()).Int64("run_id", runID).Int32("attempt", run.Attempt).Bool("failed_only", failedOnly).
447 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("ci run started again")
448 return run, nil
449}
450
451// rerunSet returns the keys of the jobs to run again: all of them, or the ones
452// that did not succeed and those that need them, directly or not.
453func rerunSet(rs *runState, failedOnly bool) (map[string]bool, error) {
454 redo := map[string]bool{}
455 if !failedOnly {
456 for _, j := range rs.jobs {
457 redo[rootKey(j.Key)] = true
458 }
459 return redo, nil
460 }
461 for _, j := range rs.jobs {
462 switch j.Conclusion {
463 case domain.CIFailure, domain.CICancelled, domain.CITimedOut, domain.CIActionRequired:
464 redo[rootKey(j.Key)] = true
465 }
466 }
467 if len(redo) == 0 {
468 return nil, fmt.Errorf("%w: no job of run %d failed", domain.ErrPreconditionFailed, rs.run.ID)
469 }
470 for changed := true; changed; {
471 changed = false
472 for _, j := range rs.jobs {
473 if redo[j.Key] || j.Scope != "" {
474 continue
475 }
476 if slices.ContainsFunc(j.Needs, func(n string) bool { return redo[n] }) {
477 redo[j.Key], changed = true, true
478 }
479 }
480 }
481 return redo, nil
482}
483
484func (w *wfCache) drop(runID int64) {
485 w.mu.Lock()
486 delete(w.m, runID)
487 delete(w.uses, runID)
488 w.mu.Unlock()
489}
490
491// --- checks ---
492
493// CommitChecks returns the checks of a commit, without access control. A
494// commit with more than MaxCIChecks checks is an error (domain.ErrResourceExhausted):
495// the merge rule cannot answer from a part of them.
496func (c *CI) CommitChecks(ctx context.Context, repo domain.RepoName, sha string) ([]domain.CICheck, error) {
497 checks, more, err := c.cfg.Store.ListChecks(ctx, repo, sha, MaxCIChecks)
498 if err != nil {
499 return nil, err
500 }
501 if more {
502 return nil, fmt.Errorf("%w: the commit has more than %d checks", domain.ErrResourceExhausted, MaxCIChecks)
503 }
504 return checks, nil
505}
506
507// ListChecks returns the checks of a commit: the one of each job of the latest
508// attempt of each run on it. A commit without a run has none.
509func (c *CI) ListChecks(ctx context.Context, p domain.Principal, name domain.RepoName, sha string) (ports.ChecksResult, error) {
510 if _, err := c.readableRepo(ctx, p, name); err != nil {
511 return ports.ChecksResult{}, err
512 }
513 if !isHex40(sha) {
514 return ports.ChecksResult{}, fmt.Errorf("%w: a commit is 40 lowercase hexadecimal characters", domain.ErrInvalidObjectID)
515 }
516 checks, more, err := c.cfg.Store.ListChecks(ctx, name, sha, MaxCIChecks)
517 if err != nil {
518 return ports.ChecksResult{}, err
519 }
520 res := ports.ChecksResult{Truncated: more}
521 var runners []domain.CIRunner
522 events := make(map[int64]string)
523 for _, ck := range checks {
524 event, ok := events[ck.RunID]
525 if !ok {
526 run, err := c.cfg.Store.GetRun(ctx, name, ck.RunID)
527 if err != nil && !errors.Is(err, domain.ErrNotFound) {
528 return ports.ChecksResult{}, err
529 }
530 if err == nil {
531 event = run.Event
532 }
533 events[ck.RunID] = event
534 }
535 v := ports.CheckView{CICheck: ck, Event: event}
536 if ck.Status == domain.CIQueued {
537 if runners == nil {
538 runners, _ = c.cfg.Store.ListRunners(ctx)
539 }
540 v.WaitingForRunner = c.noRunnerFor(ctx, ck, runners)
541 }
542 res.Checks = append(res.Checks, v)
543 }
544 return res, nil
545}
546
547// noRunnerFor reports whether no registered runner can take a queued check's
548// job. A job that waits for its needs is not waiting for a runner.
549func (c *CI) noRunnerFor(ctx context.Context, ck domain.CICheck, runners []domain.CIRunner) bool {
550 if ck.JobID == "" {
551 return false
552 }
553 j, err := c.cfg.Store.GetJob(ctx, ck.RunID, ck.Attempt, ck.JobID)
554 if err != nil || j.Phase != domain.CIPhaseReady {
555 return false
556 }
557 return !slices.ContainsFunc(runners, func(r domain.CIRunner) bool { return domain.LabelsMatch(j.Labels, r.Labels) })
558}
559
560// --- runners ---
561
562// ListRunners returns a page of the runners ordered by name. The principal
563// needs runner:register on a token that is not restricted to repositories.
564func (c *CI) ListRunners(ctx context.Context, p domain.Principal, size int, token string) (ports.RunnerPage, error) {
565 if err := canManageRunners(p); err != nil {
566 return ports.RunnerPage{}, err
567 }
568 n, err := pageSize(size)
569 if err != nil {
570 return ports.RunnerPage{}, err
571 }
572 after := ""
573 if token != "" {
574 raw, ok := decodeToken(token, "rn1:")
575 if !ok || raw == "" {
576 return ports.RunnerPage{}, fmt.Errorf("%w: runners", domain.ErrInvalidPageToken)
577 }
578 after = raw
579 }
580 all, err := c.cfg.Store.ListRunners(ctx)
581 if err != nil {
582 return ports.RunnerPage{}, err
583 }
584 busy := map[string]bool{}
585 if active, err := c.cfg.Store.ListActiveJobs(ctx); err == nil {
586 for _, j := range active {
587 if j.Phase == domain.CIPhaseRunning {
588 busy[j.RunnerID] = true
589 }
590 }
591 }
592 var page ports.RunnerPage
593 var rows []domain.CIRunner
594 rows, page.NextPageToken = pageAfter(all, n,
595 func(r domain.CIRunner) bool { return after == "" || strings.ToLower(r.Name) > after },
596 func(r domain.CIRunner) string {
597 return base64.RawURLEncoding.EncodeToString([]byte("rn1:" + strings.ToLower(r.Name)))
598 })
599 now := c.now()
600 c.mu.Lock()
601 defer c.mu.Unlock()
602 for _, r := range rows {
603 page.Runners = append(page.Runners, ports.RunnerView{CIRunner: r, LastSeen: c.seen[r.ID], Busy: busy[r.ID], Online: c.runnerOnline(c.seen[r.ID], now)})
604 }
605 return page, nil
606}
607
608// RemoveRunner removes a runner: its token stops working at once and a poll of
609// it in progress ends with unauthenticated. A job it runs is not affected.
610func (c *CI) RemoveRunner(ctx context.Context, p domain.Principal, id string) error {
611 if err := canManageRunners(p); err != nil {
612 return err
613 }
614 r, err := c.cfg.Store.RemoveRunner(ctx, id)
615 if err != nil {
616 return err
617 }
618 c.forgetRunners([]domain.CIRunner{r})
619 ctxLog(ctx).Info().Str("runner_id", r.ID).Str("runner", r.Name).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("runner removed")
620 return nil
621}