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