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