1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "slices"
10 "strings"
11 "time"
12 "unicode/utf8"
13
14 "github.com/bnema/zerowrap"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// maxVersionLen bounds RegisterRunner.version.
21const maxVersionLen = 64
22
23// --- registration ---
24
25// Register adds a runner and returns its token, shown once. The principal
26// needs runner:register on a token that is not restricted to repositories, or
27// to be root. A name in use is domain.ErrAlreadyExists unless Replace is set,
28// which removes the old runner (its token stops working at once).
29func (c *CI) Register(ctx context.Context, p domain.Principal, params ports.RegisterRunnerParams) (ports.RegisteredRunner, error) {
30 if err := canManageRunners(p); err != nil {
31 return ports.RegisteredRunner{}, err
32 }
33 if err := domain.ParseRunnerName(params.Name); err != nil {
34 return ports.RegisteredRunner{}, err
35 }
36 labels, err := domain.ParseRunnerLabels(params.Labels)
37 if err != nil {
38 return ports.RegisteredRunner{}, err
39 }
40 var v domain.Violations
41 if len(params.Version) > maxVersionLen || !utf8.ValidString(params.Version) {
42 v.Add("version", "is at most 64 characters")
43 }
44 switch params.Isolation {
45 case domain.CIIsolationVM, domain.CIIsolationContainerSysbox, domain.CIIsolationContainerPrivileged:
46 default:
47 v.Add("isolation", "is required")
48 }
49 if params.Arch != "x64" && params.Arch != "arm64" {
50 v.Add("arch", `must be "x64" or "arm64"`)
51 }
52 if params.OS != "linux" {
53 v.Add("os", `must be "linux"`)
54 }
55 if err := v.Err(); err != nil {
56 return ports.RegisteredRunner{}, err
57 }
58
59 token := newToken(domain.RunnerTokenPrefix)
60 r := domain.CIRunner{
61 ID: newRunnerID(), Name: params.Name, Labels: domain.RunnerLabels(labels, params.OS, params.Arch),
62 Isolation: params.Isolation, Version: params.Version, Arch: params.Arch, OS: params.OS,
63 RegisteredAt: c.now(), TokenHash: domain.HashTokenSecret(token),
64 }
65 removed, err := c.cfg.Store.CreateRunner(ctx, r, params.Replace)
66 if err != nil {
67 return ports.RegisteredRunner{}, err
68 }
69 log := zerowrap.FromCtx(ctx)
70 log.Info().Str("runner_id", r.ID).Str("runner", r.Name).Str("isolation", string(r.Isolation)).
71 Int("replaced", len(removed)).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("runner registered")
72 c.forgetRunners(removed)
73 return ports.RegisteredRunner{Runner: r, Token: token}, nil
74}
75
76// forgetRunners drops what is kept in memory for removed runners and ends
77// their polls in progress.
78func (c *CI) forgetRunners(rs []domain.CIRunner) {
79 c.mu.Lock()
80 for _, r := range rs {
81 delete(c.polls, r.ID)
82 delete(c.seen, r.ID)
83 }
84 c.mu.Unlock()
85 c.wakeQueue()
86}
87
88// runnerFor authenticates a runner token.
89func (c *CI) runnerFor(ctx context.Context, secret string) (domain.CIRunner, error) {
90 if !domain.IsRunnerToken(secret) {
91 return domain.CIRunner{}, domain.ErrUnauthenticated
92 }
93 r, err := c.cfg.Store.FindRunnerByToken(ctx, domain.HashTokenSecret(secret))
94 if errors.Is(err, domain.ErrNotFound) {
95 return domain.CIRunner{}, domain.ErrUnauthenticated
96 }
97 return r, err
98}
99
100// RunnerOnlineWindow is how long after its last sign of life (a poll or a
101// heartbeat) a runner is online, by the server's clock.
102const RunnerOnlineWindow = 2 * time.Minute
103
104func (c *CI) runnerOnline(lastSeen, now time.Time) bool {
105 return !lastSeen.IsZero() && now.Sub(lastSeen) < RunnerOnlineWindow
106}
107
108// noteVersion records the version a poll reports, once: the runner read for
109// this poll already holds the stored one, so an unchanged version costs no
110// write. A runner that sends none keeps its stored version. A failure to
111// record is logged and does not fail the poll.
112func (c *CI) noteVersion(ctx context.Context, r domain.CIRunner, version string) {
113 if version == "" || version == r.Version {
114 return
115 }
116 if err := c.cfg.Store.SetRunnerVersion(ctx, r.ID, version); err != nil && !errors.Is(err, domain.ErrNotFound) {
117 log := zerowrap.FromCtx(ctx)
118 log.Warn().Err(err).Str("runner_id", r.ID).Msg("runner version not recorded")
119 }
120}
121
122func (c *CI) markSeen(runnerID string) {
123 c.mu.Lock()
124 c.seen[runnerID] = c.now()
125 c.mu.Unlock()
126}
127
128// --- FetchJob ---
129
130var errNoJob = errors.New("no job")
131
132// FetchJob holds the call until a queued job matches the runner, or the poll
133// time passes. A job is given to exactly one runner: assignments are made one
134// at a time under a lock and recorded before the answer is sent.
135func (c *CI) FetchJob(ctx context.Context, secret string, params ports.FetchJobParams) (ports.FetchJobResult, error) {
136 runner, err := c.runnerFor(ctx, secret)
137 if err != nil {
138 return ports.FetchJobResult{}, err
139 }
140 if len(params.Version) > maxVersionLen || !utf8.ValidString(params.Version) {
141 return ports.FetchJobResult{}, fmt.Errorf("%w: version is at most 64 characters", domain.ErrInvalid)
142 }
143 offered, err := offeredLabels(runner, params.Labels)
144 if err != nil {
145 return ports.FetchJobResult{}, err
146 }
147 if params.PollID != "" {
148 if err := checkPollID(params.PollID); err != nil {
149 return ports.FetchJobResult{}, err
150 }
151 j, ok, err := c.repeatPoll(ctx, runner, params.PollID)
152 if err != nil {
153 return ports.FetchJobResult{}, err // the runner asks again with the same poll id
154 }
155 if ok {
156 return ports.FetchJobResult{Job: j}, nil
157 }
158 }
159 c.markSeen(runner.ID)
160 c.noteVersion(ctx, runner, params.Version)
161 deadline := time.NewTimer(c.cfg.PollTimeout)
162 defer deadline.Stop()
163 for {
164 wake := c.pending.wait(queueKey) // before looking: a job made meanwhile wakes us
165 job, err := c.assignTo(ctx, runner, offered, params.PollID)
166 switch {
167 case err == nil:
168 c.markSeen(runner.ID)
169 return ports.FetchJobResult{Job: job}, nil
170 case !errors.Is(err, errNoJob):
171 return ports.FetchJobResult{}, err
172 }
173 select {
174 case <-wake:
175 case <-deadline.C:
176 c.markSeen(runner.ID)
177 return ports.FetchJobResult{}, nil
178 case <-ctx.Done():
179 return ports.FetchJobResult{}, ctx.Err()
180 case <-c.ctx.Done():
181 return ports.FetchJobResult{RetryAfter: time.Second}, nil
182 }
183 // A removed runner's poll ends with unauthenticated.
184 if _, err := c.cfg.Store.GetRunner(ctx, runner.ID); errors.Is(err, domain.ErrNotFound) {
185 return ports.FetchJobResult{}, domain.ErrUnauthenticated
186 }
187 }
188}
189
190func checkPollID(id string) error {
191 if len(id) > 64 {
192 return fmt.Errorf("%w: poll_id is at most 64 characters", domain.ErrInvalid)
193 }
194 for _, r := range id {
195 if (r < 'a' || r > 'z') && (r < 'A' || r > 'Z') && (r < '0' || r > '9') && r != '_' && r != '-' {
196 return fmt.Errorf("%w: poll_id has letters, digits, '_' and '-' only", domain.ErrInvalid)
197 }
198 }
199 return nil
200}
201
202// offeredLabels are the labels of one poll: all those of the runner, or the
203// subset the call names.
204func offeredLabels(r domain.CIRunner, asked []string) ([]string, error) {
205 if len(asked) == 0 {
206 return r.Labels, nil
207 }
208 var out []string
209 for _, l := range asked {
210 l = strings.ToLower(l)
211 if !slices.Contains(r.Labels, l) {
212 return nil, fmt.Errorf("%w: label %q is not a label of the runner", domain.ErrInvalid, l)
213 }
214 out = append(out, l)
215 }
216 return out, nil
217}
218
219// repeatPoll answers a retried poll with the assignment the first call made,
220// as long as it has not been acknowledged or cancelled.
221func (c *CI) repeatPoll(ctx context.Context, runner domain.CIRunner, pollID string) (*ports.RunnerJob, bool, error) {
222 c.mu.Lock()
223 rec, ok := c.polls[runner.ID]
224 c.mu.Unlock()
225 if !ok || rec.pollID != pollID {
226 return nil, false, nil
227 }
228 var rid int64
229 var ra int32
230 var id string
231 if _, err := fmt.Sscanf(rec.ref, "%d/%d/%s", &rid, &ra, &id); err != nil {
232 return nil, false, nil
233 }
234 var out *ports.RunnerJob
235 err := c.withRun(ctx, rid, ra, func(rs *runState) error {
236 j := rs.find(id)
237 if j == nil || j.Phase != domain.CIPhaseAssigned || j.TokenHash != rec.hash || j.RunnerID != runner.ID {
238 return errNoJob
239 }
240 // The secrets are not kept with the poll: they are read again.
241 settings, err := c.settingsOf(ctx, rs, *j)
242 if err != nil {
243 return err
244 }
245 out = c.runnerJob(rs, *j, jobSecrets{token: rec.token, ghToken: rec.ghToken, idToken: rec.idToken}, settings)
246 return nil
247 })
248 switch {
249 case err == nil && out != nil:
250 return out, true, nil
251 case isSettingsError(err):
252 return nil, false, err // checked first: it may wrap ErrConflict or ErrNotFound
253 case err == nil, errors.Is(err, errNoJob), errors.Is(err, domain.ErrNotFound), errors.Is(err, domain.ErrConflict):
254 return nil, false, nil // nothing to repeat: the poll goes on as a new one
255 }
256 return nil, false, err // the settings could not be read: not a lost assignment
257}
258
259// runnerJob is the job as the runner gets it. settings are read when the job
260// is assigned and live in this value only: j is a copy, so the contexts it
261// carries (with the variables as they are now) are not stored, and the secrets
262// go nowhere but to the runner (never to the run record, the github context or
263// a log; RunnerJob prints without them).
264//
265// TODO(ci3-c2): the runner writes the "Set up job" log; it states there that
266// secrets were withheld when Trusted is false (the server adds no setup line).
267func (c *CI) runnerJob(rs *runState, j domain.CIJob, sec jobSecrets, settings ports.JobSettings) *ports.RunnerJob {
268 if settings.Vars != nil {
269 j.ContextsJSON = withVars(j.ContextsJSON, settings.Vars)
270 }
271 path, yaml := rs.run.WorkflowPath, rs.src.WorkflowYAML
272 if call := rs.scopeOf(j); call != nil && call.Call != nil {
273 // A job of a called workflow runs from the file of the call, under its
274 // id in that file.
275 path, yaml, j.Key = call.Call.Path, call.Call.YAML, j.CallID
276 }
277 out := &ports.RunnerJob{
278 Job: j, Token: sec.token, GithubToken: sec.ghToken, Repo: rs.run.Repo, Sha: rs.run.Sha, Ref: rs.run.Ref,
279 WorkflowPath: path, WorkflowYAML: yaml,
280 Environment: j.Environment, Vars: settings.Vars, Secrets: settings.Secrets, Trusted: rs.run.Trusted,
281 Permissions: domain.TokenPermissions(j.Permissions, rs.run.Trusted),
282 Cache: c.cacheScope(rs.run, rs.src.GithubJSON),
283 }
284 if sec.idToken != "" && c.cfg.OIDC != nil {
285 out.IDTokenURL, out.IDTokenRequestToken = c.cfg.OIDC.TokenURL(), sec.idToken
286 }
287 return out
288}
289
290// jobSecrets are the secrets issued with an assignment: the job token, the
291// repository token and the ID token request token ("" when the job gets
292// none).
293type jobSecrets struct {
294 token, ghToken, idToken string
295}
296
297// assignTo gives the oldest queued job that matches the labels to the runner,
298// or errNoJob. The assignment lock makes the choice and the write one step.
299func (c *CI) assignTo(ctx context.Context, runner domain.CIRunner, offered []string, pollID string) (*ports.RunnerJob, error) {
300 c.assign.Lock()
301 defer c.assign.Unlock()
302 ready, err := c.cfg.Store.ListReadyJobs(ctx, 0)
303 if err != nil {
304 return nil, err
305 }
306 for _, cand := range ready {
307 if !domain.LabelsMatch(cand.Labels, offered) {
308 continue
309 }
310 var out *ports.RunnerJob
311 err := c.withRun(ctx, cand.RunID, cand.RunAttempt, func(rs *runState) error {
312 j := rs.find(cand.ID)
313 if j == nil || j.Phase != domain.CIPhaseReady || rs.run.CancelRequested {
314 return errNoJob
315 }
316 // Before anything changes: a failure here leaves the job ready.
317 settings, err := c.settingsOf(ctx, rs, *j)
318 if err != nil {
319 return err
320 }
321 token := newToken(domain.JobTokenPrefix)
322 ghToken := rs.issueRepoToken(j)
323 idToken := rs.issueIDTokenRequest(j)
324 j.Phase, j.AssignedAt = domain.CIPhaseAssigned, c.now()
325 j.TokenHash, j.TokenValidUntil = domain.HashTokenSecret(token), time.Time{}
326 j.RunnerID, j.RunnerName, j.Isolation = runner.ID, runner.Name, runner.Isolation
327 rs.touch(j)
328 if err := rs.flush(ctx); err != nil {
329 return err
330 }
331 out = c.runnerJob(rs, *j, jobSecrets{token: token, ghToken: ghToken, idToken: idToken}, settings)
332 if pollID != "" {
333 c.mu.Lock()
334 c.polls[runner.ID] = pollRecord{pollID: pollID, ref: jobRef(*j), token: token, ghToken: ghToken, idToken: idToken, hash: j.TokenHash}
335 c.mu.Unlock()
336 }
337 return nil
338 })
339 switch {
340 case err == nil:
341 log := zerowrap.FromCtx(ctx)
342 log.Info().Str("runner", runner.Name).Str("job_id", out.Job.ID).Int64("run_id", out.Job.RunID).Msg("job assigned")
343 return out, nil
344 case isSettingsError(err):
345 return nil, err // the job stays ready; the runner polls again
346 case errors.Is(err, errNoJob), errors.Is(err, domain.ErrNotFound), errors.Is(err, domain.ErrConflict):
347 continue // taken or gone meanwhile
348 default:
349 return nil, err
350 }
351 }
352 return nil, errNoJob
353}
354
355// --- the calls of a job token ---
356
357// jobFor authenticates a job token and checks that it is the one of jobID.
358func (c *CI) jobFor(ctx context.Context, secret, jobID string) (domain.CIJob, error) {
359 if !domain.IsJobToken(secret) {
360 return domain.CIJob{}, domain.ErrUnauthenticated
361 }
362 j, err := c.cfg.Store.FindJobByToken(ctx, domain.HashTokenSecret(secret), c.now())
363 if errors.Is(err, domain.ErrNotFound) {
364 return domain.CIJob{}, domain.ErrUnauthenticated
365 }
366 if err != nil {
367 return domain.CIJob{}, err
368 }
369 if j.ID != jobID {
370 return domain.CIJob{}, fmt.Errorf("%w: this token is the token of another job", domain.ErrForbidden)
371 }
372 return j, nil
373}
374
375func endedError(j domain.CIJob) error {
376 return &domain.JobEndedError{Conclusion: j.Conclusion, Message: j.Error}
377}
378
379// lockedJob runs fn with the lock of the run and the job read again.
380func (c *CI) lockedJob(ctx context.Context, j domain.CIJob, fn func(cur *domain.CIJob) error) error {
381 lock := c.runLock(j.RunID)
382 lock.Lock()
383 defer lock.Unlock()
384 cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
385 if err != nil {
386 return err
387 }
388 if err := sameAssignment(j, cur); err != nil {
389 return err
390 }
391 return fn(&cur)
392}
393
394// sameAssignment checks, under the lock of the run, that the job a token was
395// found for is still given under that token: an assignment cancelled and given
396// again since has a new one, and the old runner is unauthenticated. A job that
397// has ended answers with its end instead (the callers do).
398func sameAssignment(was, cur domain.CIJob) error {
399 if !cur.Ended() && cur.TokenHash != was.TokenHash {
400 return domain.ErrUnauthenticated
401 }
402 return nil
403}
404
405// saveJob writes one job (its steps changed) with the annotations that change
406// with it, and publishes JobUpdated. The status of the run does not depend on
407// steps.
408func (c *CI) saveJob(ctx context.Context, j *domain.CIJob, anns ...ports.CIStepAnnotations) error {
409 v, err := c.cfg.Store.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{*j}, Annotations: anns})
410 if err != nil {
411 return fmt.Errorf("write job %s: %w", j.ID, err)
412 }
413 j.Version = v
414 if repo, err := c.cfg.Meta.Get(ctx, j.Repo); err == nil {
415 lines, _ := c.cfg.Logs.Count(ctx, logRef(*j))
416 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: *j, Lines: lines, At: c.now(), Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
417 }
418 c.jobs.notify(jobRef(*j))
419 return nil
420}
421
422func logRef(j domain.CIJob) ports.CILogRef {
423 return ports.CILogRef{RunID: j.RunID, JobID: j.ID, Attempt: j.Attempt}
424}
425
426// UpdateJob applies the steps and the annotations of the request whole or not
427// at all. The annotations of each step they name replace those stored, whether
428// the step is completed or not.
429func (c *CI) UpdateJob(ctx context.Context, secret, jobID string, steps []ports.StepUpdate, annotations []domain.CIAnnotation) error {
430 j, err := c.jobFor(ctx, secret, jobID)
431 if err != nil {
432 return err
433 }
434 if j.Ended() {
435 return endedError(j)
436 }
437 if err := checkStepUpdates(steps); err != nil {
438 return err
439 }
440 if err := domain.CheckAnnotations(annotations); err != nil {
441 return err
442 }
443 return c.lockedJob(ctx, j, func(cur *domain.CIJob) error {
444 if cur.Ended() {
445 return endedError(*cur)
446 }
447 next, changed, err := applySteps(cur.Steps, steps)
448 if err != nil {
449 return err
450 }
451 writes, err := c.annotationWrites(ctx, *cur, annotations)
452 if err != nil {
453 return err
454 }
455 if !changed && len(writes) == 0 {
456 return nil
457 }
458 cur.Steps = next
459 return c.saveJob(ctx, cur, writes...)
460 })
461}
462
463// checkStepUpdates validates the shape of the steps of a request.
464func checkStepUpdates(steps []ports.StepUpdate) error {
465 if len(steps) > domain.CIMaxSteps {
466 return fmt.Errorf("%w: at most %d steps", domain.ErrInvalid, domain.CIMaxSteps)
467 }
468 for _, s := range steps {
469 if s.Number < 1 || s.Number > domain.CIMaxSteps {
470 return fmt.Errorf("%w: step number %d is not between 1 and %d", domain.ErrInvalid, s.Number, domain.CIMaxSteps)
471 }
472 switch s.State {
473 case domain.CIStepPending:
474 if s.Conclusion != domain.CINoConclusion || s.Outcome != domain.CINoConclusion {
475 return fmt.Errorf("%w: step %d is not completed: it has no conclusion or outcome", domain.ErrInvalid, s.Number)
476 }
477 case domain.CIStepInProgress:
478 if s.Conclusion != domain.CINoConclusion || s.Outcome != domain.CINoConclusion || s.StartedAt.IsZero() {
479 return fmt.Errorf("%w: step %d is in progress: started_at is required, conclusion and outcome are not allowed", domain.ErrInvalid, s.Number)
480 }
481 case domain.CIStepCompleted:
482 switch s.Conclusion {
483 case domain.CISuccess, domain.CIFailure, domain.CICancelled, domain.CISkipped:
484 default:
485 return fmt.Errorf("%w: step %d is completed: its conclusion must be success, failure, cancelled or skipped", domain.ErrInvalid, s.Number)
486 }
487 switch s.Outcome {
488 case domain.CINoConclusion, domain.CISuccess, domain.CIFailure, domain.CICancelled, domain.CISkipped:
489 default:
490 return fmt.Errorf("%w: step %d is completed: its outcome must be success, failure, cancelled or skipped", domain.ErrInvalid, s.Number)
491 }
492 if s.StartedAt.IsZero() && s.Conclusion != domain.CISkipped {
493 return fmt.Errorf("%w: step %d: started_at is required", domain.ErrInvalid, s.Number)
494 }
495 if s.CompletedAt.IsZero() {
496 return fmt.Errorf("%w: step %d is completed: completed_at is required", domain.ErrInvalid, s.Number)
497 }
498 default:
499 return fmt.Errorf("%w: step %d has no state", domain.ErrInvalid, s.Number)
500 }
501 }
502 return nil
503}
504
505// applySteps returns the steps after the updates, and whether anything
506// changed. The input is not modified.
507func applySteps(have []domain.CIStep, ups []ports.StepUpdate) ([]domain.CIStep, bool, error) {
508 out := slices.Clone(have)
509 changed := false
510 for _, u := range ups {
511 name := truncateRunes(u.Name, domain.CIMaxStepName)
512 i := slices.IndexFunc(out, func(s domain.CIStep) bool { return s.Number == u.Number })
513 if i < 0 {
514 out = append(out, domain.CIStep{
515 Number: u.Number, Name: name, State: u.State, Conclusion: u.Conclusion, Outcome: u.Outcome,
516 StartedAt: u.StartedAt, CompletedAt: u.CompletedAt,
517 })
518 changed = true
519 continue
520 }
521 cur := out[i]
522 next := cur
523 next.Name, next.State, next.Conclusion, next.Outcome = name, u.State, u.Conclusion, u.Outcome
524 next.StartedAt, next.CompletedAt = u.StartedAt, u.CompletedAt
525 if cur.Name != "" && name == "" {
526 next.Name = cur.Name // an empty name never erases one
527 }
528 switch {
529 case cur.State == domain.CIStepCompleted:
530 same := next.State == cur.State && next.Conclusion == cur.Conclusion && effectiveOutcome(next) == effectiveOutcome(cur) && next.Name == cur.Name &&
531 next.StartedAt.Equal(cur.StartedAt) && next.CompletedAt.Equal(cur.CompletedAt)
532 if !same {
533 return nil, false, fmt.Errorf("%w: step %d is completed: it cannot change", domain.ErrPreconditionFailed, u.Number)
534 }
535 continue
536 case cur.State == domain.CIStepInProgress && next.State == domain.CIStepPending:
537 return nil, false, fmt.Errorf("%w: step %d cannot go from in progress back to pending", domain.ErrPreconditionFailed, u.Number)
538 }
539 if !stepsEqual(cur, next) {
540 out[i], changed = next, true
541 }
542 }
543 slices.SortFunc(out, func(a, b domain.CIStep) int { return int(a.Number) - int(b.Number) })
544 return out, changed, nil
545}
546
547// effectiveOutcome is the outcome of a step, read as its conclusion when the
548// runner did not report one.
549func effectiveOutcome(s domain.CIStep) domain.CIConclusion {
550 if s.Outcome == domain.CINoConclusion {
551 return s.Conclusion
552 }
553 return s.Outcome
554}
555
556func stepsEqual(a, b domain.CIStep) bool {
557 return a.Number == b.Number && a.Name == b.Name && a.State == b.State && a.Conclusion == b.Conclusion && a.Outcome == b.Outcome &&
558 a.StartedAt.Equal(b.StartedAt) && a.CompletedAt.Equal(b.CompletedAt) && a.LogStart == b.LogStart && a.HasLines == b.HasLines
559}
560
561// lastLineStep returns the highest step number that has lines, 0 for none.
562func lastLineStep(steps []domain.CIStep) int32 {
563 var last int32
564 for _, s := range steps {
565 if s.HasLines {
566 last = max(last, s.Number)
567 }
568 }
569 return last
570}
571
572// checkLines validates the shape of the lines of an AppendLog request.
573func checkLines(step int32, offset int64, lines []domain.CILogLine) error {
574 if step < 1 || step > domain.CIMaxSteps {
575 return fmt.Errorf("%w: step %d is not between 1 and %d", domain.ErrInvalid, step, domain.CIMaxSteps)
576 }
577 if offset < 0 {
578 return fmt.Errorf("%w: negative offset", domain.ErrInvalid)
579 }
580 if len(lines) > domain.CIMaxLogLines {
581 return fmt.Errorf("%w: at most %d lines per request", domain.ErrInvalid, domain.CIMaxLogLines)
582 }
583 total := 0
584 for i, l := range lines {
585 if strings.ContainsRune(l.Text, '\n') || !utf8.ValidString(l.Text) {
586 return fmt.Errorf("%w: line %d has a line feed or is not valid UTF-8", domain.ErrInvalid, i)
587 }
588 if l.Time.IsZero() {
589 return fmt.Errorf("%w: line %d has no time", domain.ErrInvalid, i)
590 }
591 if l.Continued && offset == 0 && i == 0 {
592 return fmt.Errorf("%w: the first line of a job cannot be continued", domain.ErrInvalid)
593 }
594 total += len(l.Text)
595 }
596 if total > domain.CIMaxLogChunkBytes {
597 return fmt.Errorf("%w: at most %d bytes of text per request", domain.ErrInvalid, domain.CIMaxLogChunkBytes)
598 }
599 return nil
600}
601
602// AppendLog appends lines of a step to the log of the job; see runner.proto
603// for the rules of offsets. The metadata store is written only when a step
604// gets its first line.
605func (c *CI) AppendLog(ctx context.Context, secret, jobID string, step int32, offset int64, lines []domain.CILogLine) (int64, error) {
606 j, err := c.jobFor(ctx, secret, jobID)
607 if err != nil {
608 return 0, err
609 }
610 if j.Ended() {
611 return 0, endedError(j)
612 }
613 if err := checkLines(step, offset, lines); err != nil {
614 return 0, err
615 }
616 var next int64
617 limitHit := false
618 err = c.lockedJob(ctx, j, func(cur *domain.CIJob) error {
619 if cur.Ended() {
620 return endedError(*cur)
621 }
622 ref := logRef(*cur)
623 count, err := c.cfg.Logs.Count(ctx, ref)
624 if err != nil {
625 return err
626 }
627 if offset > count {
628 return &ports.ErrCILogGap{Count: count}
629 }
630 skip := count - offset
631 if skip >= int64(len(lines)) {
632 next = count // everything is stored already
633 return nil
634 }
635 last := lastLineStep(cur.Steps)
636 if step < last {
637 return fmt.Errorf("%w: step %d is below step %d, which already has lines", domain.ErrPreconditionFailed, step, last)
638 }
639 if first := lines[skip]; first.Continued && count > 0 && step != last {
640 return fmt.Errorf("%w: a continued line must be in the step of the line before it", domain.ErrPreconditionFailed)
641 }
642 n, added, err := c.cfg.Logs.Append(ctx, ref, offset, lines, domain.CIMaxLogBytes)
643 if errors.Is(err, ports.ErrCILogLimit) {
644 limitHit = true
645 return nil
646 }
647 if err != nil {
648 return err
649 }
650 next = n
651 if added == 0 {
652 return nil
653 }
654 return c.startStepLines(ctx, cur, step, count, lines[skip].Time)
655 })
656 if err != nil {
657 return 0, err
658 }
659 if limitHit {
660 return 0, c.stopForLogLimit(ctx, j)
661 }
662 return next, nil
663}
664
665// startStepLines records where the lines of a step begin, creating the step
666// when UpdateJob did not announce it.
667func (c *CI) startStepLines(ctx context.Context, cur *domain.CIJob, step int32, firstLine int64, at time.Time) error {
668 i := slices.IndexFunc(cur.Steps, func(s domain.CIStep) bool { return s.Number == step })
669 switch {
670 case i < 0:
671 cur.Steps = append(cur.Steps, domain.CIStep{
672 Number: step, State: domain.CIStepInProgress, StartedAt: at, HasLines: true, LogStart: firstLine,
673 })
674 slices.SortFunc(cur.Steps, func(a, b domain.CIStep) int { return int(a.Number) - int(b.Number) })
675 case cur.Steps[i].HasLines:
676 return nil
677 default:
678 cur.Steps[i].HasLines, cur.Steps[i].LogStart = true, firstLine
679 }
680 return c.saveJob(ctx, cur)
681}
682
683// stopForLogLimit ends a job whose log reached its limit: a last line says
684// so, the job fails, and the call answers like any call after the end.
685func (c *CI) stopForLogLimit(ctx context.Context, j domain.CIJob) error {
686 var ended *domain.JobEndedError
687 err := c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
688 cur := rs.find(j.ID)
689 if cur == nil {
690 return domain.ErrNotFound
691 }
692 if cur.Ended() {
693 ended = &domain.JobEndedError{Conclusion: cur.Conclusion, Message: cur.Error}
694 return nil
695 }
696 if err := sameAssignment(j, *cur); err != nil {
697 return err
698 }
699 _, _ = c.cfg.Logs.AppendForced(ctx, logRef(*cur), []domain.CILogLine{{Time: c.now(), Text: "##[error]" + domain.CILogLimitMessage}})
700 rs.end(cur, domain.CIFailure, domain.CILogLimitMessage)
701 ended = &domain.JobEndedError{Conclusion: domain.CIFailure, Message: domain.CILogLimitMessage}
702 if err := rs.advance(ctx); err != nil {
703 return err
704 }
705 return rs.flush(ctx)
706 })
707 if err != nil {
708 return err
709 }
710 if ended == nil {
711 return domain.ErrNotFound // a typed nil would be taken for an error
712 }
713 return ended
714}
715
716// Heartbeat tells the server that the runner is alive. The first one
717// acknowledges the assignment; the others only move the deadline.
718func (c *CI) Heartbeat(ctx context.Context, secret, jobID string) error {
719 j, err := c.jobFor(ctx, secret, jobID)
720 if err != nil {
721 return err
722 }
723 if j.Ended() {
724 return endedError(j)
725 }
726 // Everything is done under the lock of the run: the acknowledgement, and
727 // the note of the heartbeat, which must not be written once the job ended
728 // (end() forgets it under the same lock).
729 return c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
730 cur := rs.find(j.ID)
731 switch {
732 case cur == nil:
733 return domain.ErrUnauthenticated
734 case cur.Ended():
735 return endedError(*cur)
736 case cur.TokenHash != j.TokenHash:
737 return domain.ErrUnauthenticated // the assignment was cancelled and given again
738 }
739 now := c.now()
740 if cur.Phase == domain.CIPhaseAssigned {
741 cur.Phase, cur.AckedAt, cur.StartedAt, cur.AssignCancels = domain.CIPhaseRunning, now, now, 0
742 rs.touch(cur)
743 if err := rs.flush(ctx); err != nil {
744 return err
745 }
746 c.forgetPoll(jobRef(*cur)) // the token it holds is the runner's now
747 }
748 c.mu.Lock()
749 c.heartbeats[jobRef(*cur)] = now
750 c.seen[cur.RunnerID] = now
751 c.mu.Unlock()
752 return nil
753 })
754}
755
756// forgetPoll drops the remembered assignment (and its plaintext token) of a
757// job: after the acknowledgement or the cancel of the assignment, a retried
758// poll has nothing to repeat.
759func (c *CI) forgetPoll(ref string) {
760 c.mu.Lock()
761 defer c.mu.Unlock()
762 for id, p := range c.polls {
763 if p.ref == ref {
764 delete(c.polls, id)
765 }
766 }
767}
768
769// WatchJob streams what the runner must react to: the cancel request (the
770// first message when one is already pending), keepalives, and the end.
771func (c *CI) WatchJob(ctx context.Context, secret, jobID string, ready func(), send func(ports.JobSignal) error) error {
772 j, err := c.jobFor(ctx, secret, jobID)
773 if err != nil {
774 return err
775 }
776 if j.Ended() {
777 return endedError(j)
778 }
779 ref := jobRef(j)
780 keepalive := time.NewTicker(c.cfg.Keepalive)
781 defer keepalive.Stop()
782 cancelSent := false
783 if ready != nil {
784 ready()
785 }
786 for {
787 wake := c.jobs.wait(ref)
788 cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
789 if err != nil {
790 return err
791 }
792 if cur.TokenHash != j.TokenHash && !cur.Ended() {
793 return domain.ErrUnauthenticated // given again
794 }
795 if cur.Ended() {
796 return send(ports.JobSignal{Ended: &domain.JobEndedError{Conclusion: cur.Conclusion, Message: cur.Error}})
797 }
798 if cur.CancelReason != domain.CICancelNone && !cancelSent {
799 cancelSent = true
800 if err := send(ports.JobSignal{Cancel: &ports.JobCancel{
801 Reason: cur.CancelReason, Message: cur.CancelMessage, RequestedAt: cur.CancelRequestedAt,
802 }}); err != nil {
803 return err
804 }
805 }
806 select {
807 case <-wake:
808 case <-keepalive.C:
809 if err := send(ports.JobSignal{Keepalive: true}); err != nil {
810 return err
811 }
812 case <-ctx.Done():
813 return nil
814 case <-c.ctx.Done():
815 return nil
816 }
817 }
818}
819
820// CompleteJob ends a job, once. A later call returns the stored result.
821func (c *CI) CompleteJob(ctx context.Context, secret string, params ports.CompleteJobParams) (ports.CompleteJobResult, error) {
822 j, err := c.jobFor(ctx, secret, params.JobID)
823 if err != nil {
824 return ports.CompleteJobResult{}, err
825 }
826 if j.Ended() {
827 return ports.CompleteJobResult{Conclusion: j.Conclusion, Error: j.Error}, nil
828 }
829 switch params.Conclusion {
830 case domain.CISuccess, domain.CIFailure, domain.CICancelled, domain.CITimedOut:
831 default:
832 return ports.CompleteJobResult{}, fmt.Errorf("%w: a runner reports success, failure, cancelled or timed_out", domain.ErrInvalid)
833 }
834 if params.RunnerShutdown && params.Conclusion != domain.CICancelled {
835 return ports.CompleteJobResult{}, fmt.Errorf("%w: runner_shutdown goes with cancelled", domain.ErrInvalid)
836 }
837 if err := domain.CheckOutputs(params.Outputs); err != nil {
838 return ports.CompleteJobResult{}, err
839 }
840 if err := checkStepUpdates(params.Steps); err != nil {
841 return ports.CompleteJobResult{}, err
842 }
843 if err := domain.CheckAnnotations(params.Annotations); err != nil {
844 return ports.CompleteJobResult{}, err
845 }
846 var res ports.CompleteJobResult
847 err = c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
848 cur := rs.find(j.ID)
849 if cur == nil {
850 return domain.ErrNotFound
851 }
852 if cur.Ended() { // ended by the server since
853 res = ports.CompleteJobResult{Conclusion: cur.Conclusion, Error: cur.Error}
854 return nil
855 }
856 if err := sameAssignment(j, *cur); err != nil {
857 return err
858 }
859 steps, _, err := applySteps(cur.Steps, params.Steps)
860 if err != nil {
861 return err
862 }
863 writes, err := c.annotationWrites(ctx, *cur, params.Annotations)
864 if err != nil {
865 return err
866 }
867 rs.annotations = append(rs.annotations, writes...)
868 cur.Steps = steps
869 cur.Outputs = params.Outputs
870 msg := ""
871 if params.RunnerShutdown {
872 msg = domain.CIShutdownMessage
873 }
874 rs.end(cur, params.Conclusion, msg)
875 if err := rs.advance(ctx); err != nil {
876 return err
877 }
878 res = ports.CompleteJobResult{Conclusion: cur.Conclusion, Error: cur.Error, Applied: true}
879 return rs.flush(ctx)
880 })
881 if err != nil {
882 return ports.CompleteJobResult{}, err
883 }
884 if res.Applied {
885 log := zerowrap.FromCtx(ctx)
886 log.Info().Str("job_id", j.ID).Int64("run_id", j.RunID).Str("conclusion", string(res.Conclusion)).Msg("job completed")
887 }
888 return res, nil
889}