1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "sort"
10 "strings"
11 "sync"
12 "time"
13 "unicode/utf8"
14
15 "connectrpc.com/connect"
16 "github.com/bnema/zerowrap"
17 "google.golang.org/protobuf/proto"
18 "google.golang.org/protobuf/types/known/timestamppb"
19
20 ermite "bnema.dev/ermite-api/gen/ermite"
21 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
22
23 "bnema.dev/ermite-runner/internal/execstream"
24 "bnema.dev/ermite-runner/internal/execstream/execpb"
25 "bnema.dev/ermite-runner/internal/sandbox"
26)
27
28// jobClient is what a job needs from the RunnerService client (job token).
29type jobClient interface {
30 callClient
31 Heartbeat(context.Context, *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error)
32 WatchJob(context.Context, *connect.Request[ermite.WatchJobRequest]) (*connect.ServerStreamForClient[ermite.WatchJobResponse], error)
33 CompleteJob(context.Context, *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error)
34 ResolveActions(context.Context, *connect.Request[ermite.ResolveActionsRequest]) (*connect.Response[ermite.ResolveActionsResponse], error)
35}
36
37var _ jobClient = ermiteconnect.RunnerServiceClient(nil)
38
39// jobRun is the host's side of one job: the assignment, the sandbox, and the
40// calls to ermite with the job token.
41type jobRun struct {
42 r *Runner
43 job *ermite.Job
44 client jobClient
45 t Timing
46 log zerowrap.Logger // with job_id, run_id and attempt
47
48 // ctx lives until the job is finished or abandoned. Shutdown does not end
49 // it: a shutdown is a cancellation that the job reports.
50 ctx context.Context
51 cancel context.CancelCauseFunc
52
53 // receivedAt is when the host got the job: the job timeout counts from it
54 // (runner.proto: Job.timeout_minutes).
55 receivedAt time.Time
56
57 cancelCh chan *execpb.Cancel // the first value is sent to the executor
58 cancelMu sync.Mutex
59 cancelAt string // reason of the first cancel request, "" if none
60 cancelKind execpb.CancelKind // kind of the first cancel request
61 stopping bool // the runner is shutting down
62
63 mu sync.Mutex
64 steps map[int32]*ermite.StepUpdate
65 // timeline is the job's phase timing, guarded by mu.
66 timeline timeline
67 // What the guest has told us, for the validation of the next frame: the
68 // guest is untrusted code, its step numbers and lines are checked before
69 // they reach the server (see violate).
70 maxStarted int32
71 lastLogStep int32
72 violation string
73 // contexts is Job.Contexts as handed to the executor: with the server URLs
74 // pointed at the runner's URL when guest forge access is on.
75 contexts string
76 // Origins that are ermite for the archive tokens (see resolveResponse), and
77 // the budget of resolution requests of the job (see resolveActions).
78 publicOrigin, serverOrigin, guestForge string
79 resolveRounds, resolveRefs int
80 resolved map[string]*execpb.ResolvedAction // by "uses", for the job
81 // features is what the server accepts; annLists are the annotations the
82 // executor sent for steps that have not completed yet (see results.go).
83 features jobFeatures
84 // results says the job gets the results service (see resultsOn).
85 results bool
86 annLists map[int32][]*ermite.Annotation
87 // annSeen and sumSeen are the steps that sent their annotations and their
88 // summary: each is sent once.
89 annSeen, sumSeen map[int32]bool
90 annDropped int // annotations dropped for lack of room, see reportDroppedAnnotations
91 // hard stops the sandbox at once; set before the session starts.
92 hard func()
93 result *execpb.JobResult
94 outbox *outbox
95 endedBy error // the server ended the job, or the assignment is gone
96}
97
98var errAbandoned = errors.New("the server ended the job or revoked the assignment")
99
100func (j *jobRun) abandon(err error) {
101 j.mu.Lock()
102 if j.endedBy == nil {
103 j.endedBy = err
104 }
105 j.mu.Unlock()
106 j.cancel(errAbandoned)
107}
108
109func (j *jobRun) abandoned() error {
110 j.mu.Lock()
111 defer j.mu.Unlock()
112 return j.endedBy
113}
114
115// requestCancel asks the executor to stop; only the first request counts.
116func (j *jobRun) requestCancel(reason string, kind execpb.CancelKind) {
117 j.cancelMu.Lock()
118 defer j.cancelMu.Unlock()
119 if j.cancelAt != "" {
120 return
121 }
122 j.cancelAt = reason
123 j.cancelKind = kind
124 j.cancelCh <- &execpb.Cancel{Reason: reason, Kind: kind}
125}
126
127// firstCancelKind is the kind of the first cancel request.
128func (j *jobRun) firstCancelKind() execpb.CancelKind {
129 j.cancelMu.Lock()
130 defer j.cancelMu.Unlock()
131 return j.cancelKind
132}
133
134func (j *jobRun) cancelRequested() bool {
135 j.cancelMu.Lock()
136 defer j.cancelMu.Unlock()
137 return j.cancelAt != ""
138}
139
140// heartbeat sends the acknowledgement. It retries every HeartbeatRetry until
141// it succeeds, so that the 30 s of the server are used well; it returns false
142// when the assignment is gone.
143func (j *jobRun) ack(ctx context.Context) bool {
144 for {
145 err := j.beat(ctx)
146 if err == nil {
147 return true
148 }
149 if ctx.Err() != nil {
150 return false
151 }
152 switch classify(err) {
153 case kindEnded, kindAbandoned:
154 j.abandon(err)
155 return false
156 }
157 j.log.Warn().Err(err).Msg("first heartbeat failed, retrying")
158 if !sleep(ctx, j.t.HeartbeatRetry) {
159 return false
160 }
161 }
162}
163
164func (j *jobRun) beat(ctx context.Context) error {
165 cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
166 defer cancel()
167 _, err := j.client.Heartbeat(cctx, connect.NewRequest(&ermite.HeartbeatRequest{JobId: j.job.JobId}))
168 return err
169}
170
171// heartbeats keeps the job alive: every HeartbeatEvery, and after a failure
172// every HeartbeatRetry until one succeeds (runner.proto: Heartbeat).
173func (j *jobRun) heartbeats(ctx context.Context) {
174 wait := j.t.HeartbeatEvery
175 for {
176 if !sleep(ctx, wait) {
177 return
178 }
179 if err := j.beat(ctx); err != nil {
180 if ctx.Err() != nil {
181 return
182 }
183 switch classify(err) {
184 case kindEnded, kindAbandoned:
185 j.abandon(err)
186 return
187 }
188 j.log.Warn().Err(err).Dur("retry_in", j.t.HeartbeatRetry).Msg("heartbeat failed")
189 wait = j.t.HeartbeatRetry
190 continue
191 }
192 wait = j.t.HeartbeatEvery
193 }
194}
195
196// watch follows WatchJob: a cancel request stops the job, an Ended message or
197// the end-of-job error abandons it. The stream is reopened after any other
198// error and after a clean close without Ended.
199func (j *jobRun) watch(ctx context.Context) {
200 for ctx.Err() == nil {
201 stream, err := j.client.WatchJob(ctx, connect.NewRequest(&ermite.WatchJobRequest{JobId: j.job.JobId}))
202 if err == nil {
203 for stream.Receive() {
204 switch m := stream.Msg().Msg.(type) {
205 case *ermite.WatchJobResponse_Cancel:
206 j.requestCancel(cancelText(m.Cancel), kindOf(m.Cancel.GetReason()))
207 case *ermite.WatchJobResponse_Ended:
208 _ = stream.Close()
209 j.abandon(fmt.Errorf("job ended: %s", m.Ended.GetConclusion()))
210 return
211 }
212 }
213 err = stream.Err()
214 _ = stream.Close()
215 }
216 if ctx.Err() != nil {
217 return
218 }
219 if err != nil {
220 switch classify(err) {
221 case kindEnded, kindAbandoned:
222 j.abandon(err)
223 return
224 }
225 j.log.Warn().Err(err).Msg("WatchJob failed, reopening")
226 }
227 if !sleep(ctx, j.t.WatchRetry) {
228 return
229 }
230 }
231}
232
233func cancelText(c *ermite.CancelRequest) string {
234 if c.GetMessage() != "" {
235 return c.GetMessage()
236 }
237 return c.GetReason().String()
238}
239
240func kindOf(r ermite.CancelReason) execpb.CancelKind {
241 if r == ermite.CancelReason_CANCEL_REASON_TIMEOUT {
242 return execpb.CancelKind_CANCEL_KIND_TIMEOUT
243 }
244 // User, fail-fast, and any reason this runner does not know: cancelled.
245 return execpb.CancelKind_CANCEL_KIND_USER
246}
247
248// ---- steps
249
250// Limits on what the guest may send, which keep the memory of the host bounded
251// whatever the guest does: runner.proto allows 1,000 steps per job, names of
252// 1,024 characters, and lines of 64 KiB.
253const (
254 maxSteps = 1000
255 maxNameLen = 1024
256 maxLineText = 64 * 1024
257)
258
259// violate fails the job because the guest broke the protocol: it writes the
260// reason in the log, asks the executor to stop and destroys the sandbox. Only
261// the first violation counts; the frames that follow are dropped.
262func (j *jobRun) violate(format string, a ...any) {
263 j.mu.Lock()
264 if j.violation != "" {
265 j.mu.Unlock()
266 return
267 }
268 j.violation = fmt.Sprintf(format, a...)
269 msg := j.violation
270 j.mu.Unlock()
271 j.log.Error().Str("violation", msg).Msg("the executor broke the protocol")
272 j.sandboxError("The runner stopped the job: the executor broke the protocol: " + msg)
273 if j.hard != nil {
274 j.hard()
275 }
276}
277
278func (j *jobRun) violated() bool {
279 j.mu.Lock()
280 defer j.mu.Unlock()
281 return j.violation != ""
282}
283
284func (j *jobRun) stepStarted(s *execpb.StepStarted) {
285 j.mu.Lock()
286 switch {
287 case j.violation != "":
288 j.mu.Unlock()
289 return
290 case s.Number < 1 || s.Number > maxSteps:
291 j.mu.Unlock()
292 j.violate("step number %d is outside 1..%d", s.Number, maxSteps)
293 return
294 case s.Number <= j.maxStarted:
295 j.mu.Unlock()
296 j.violate("step %d started after step %d", s.Number, j.maxStarted)
297 return
298 }
299 j.maxStarted = s.Number
300 u := &ermite.StepUpdate{Number: s.Number, Name: cutName(s.Name), State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
301 j.steps[s.Number] = u
302 j.mu.Unlock()
303 j.outbox.Step(u, nil)
304}
305
306func (j *jobRun) stepCompleted(s *execpb.StepCompleted) {
307 j.mu.Lock()
308 if j.violation != "" {
309 j.mu.Unlock()
310 return
311 }
312 old := j.steps[s.Number]
313 switch {
314 case old == nil:
315 j.mu.Unlock()
316 j.violate("step %d completed without having started", s.Number)
317 return
318 case old.State == ermite.StepState_STEP_STATE_COMPLETED:
319 j.mu.Unlock()
320 j.violate("step %d completed twice", s.Number)
321 return
322 }
323 u := cloneStep(old)
324 if s.Name != "" {
325 u.Name = cutName(s.Name)
326 }
327 u.State = ermite.StepState_STEP_STATE_COMPLETED
328 u.Conclusion = conclusionOf(s.Conclusion)
329 if u.Conclusion == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || u.Conclusion == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
330 u.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
331 }
332 u.CompletedAt = tsOf(s.UnixNano)
333 // The outcome is how the step ended before continue-on-error; a runner that
334 // does not say leaves it unspecified, which means the conclusion.
335 if o := conclusionOf(s.Outcome); o != ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
336 if o == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
337 o = ermite.CiConclusion_CI_CONCLUSION_FAILURE
338 }
339 u.Outcome = o
340 }
341 j.steps[s.Number] = u
342 anns := j.takeAnnotations(s.Number)
343 j.mu.Unlock()
344 j.outbox.Step(u, anns)
345}
346
347// logLine checks and queues a line of the log. The step must be one that was
348// started and not completed, and not lower than the step of the previous line
349// (the server refuses a log that goes back); the text is made valid for the
350// server: UTF-8, no line feed (a line feed in a line makes more lines).
351func (j *jobRun) logLine(l *execpb.Log) {
352 j.mu.Lock()
353 if j.violation != "" {
354 j.mu.Unlock()
355 return
356 }
357 st := j.steps[l.Step]
358 switch {
359 case l.Step < 1 || l.Step > maxSteps || st == nil:
360 j.mu.Unlock()
361 j.violate("a log line for step %d, which has not started", l.Step)
362 return
363 case st.State == ermite.StepState_STEP_STATE_COMPLETED:
364 j.mu.Unlock()
365 j.violate("a log line for step %d, which is completed", l.Step)
366 return
367 case l.Step < j.lastLogStep:
368 j.mu.Unlock()
369 j.violate("a log line for step %d after one for step %d", l.Step, j.lastLogStep)
370 return
371 case len(l.Text) > maxLineText:
372 j.mu.Unlock()
373 j.violate("a log line of %d bytes (the limit is %d)", len(l.Text), maxLineText)
374 return
375 }
376 j.lastLogStep = l.Step
377 j.mu.Unlock()
378 text := l.Text
379 if !utf8.ValidString(text) {
380 text = strings.ToValidUTF8(text, "\ufffd")
381 }
382 for i, part := range strings.Split(text, "\n") {
383 j.outbox.Log(logLine{step: l.Step, unixNano: l.UnixNano, text: part, continued: l.Continued && i == 0})
384 }
385}
386
387// cutName bounds a step name to what the server keeps, on a character boundary.
388// The result is a copy: a substring would keep the whole name the guest sent
389// alive (see cutRunes).
390func cutName(s string) string {
391 if len(s) <= maxNameLen {
392 return strings.ToValidUTF8(s, "\ufffd")
393 }
394 s = s[:maxNameLen]
395 for !utf8.ValidString(s) && len(s) > 0 {
396 s = s[:len(s)-1]
397 }
398 return strings.Clone(s)
399}
400
401func cloneStep(u *ermite.StepUpdate) *ermite.StepUpdate {
402 return proto.Clone(u).(*ermite.StepUpdate)
403}
404
405func tsOf(unixNano int64) *timestamppb.Timestamp {
406 if unixNano == 0 {
407 return timestamppb.Now()
408 }
409 return timestamppb.New(time.Unix(0, unixNano))
410}
411
412func conclusionOf(c execpb.StepConclusion) ermite.CiConclusion {
413 switch c {
414 case execpb.StepConclusion_STEP_CONCLUSION_SUCCESS:
415 return ermite.CiConclusion_CI_CONCLUSION_SUCCESS
416 case execpb.StepConclusion_STEP_CONCLUSION_FAILURE:
417 return ermite.CiConclusion_CI_CONCLUSION_FAILURE
418 case execpb.StepConclusion_STEP_CONCLUSION_CANCELLED:
419 return ermite.CiConclusion_CI_CONCLUSION_CANCELLED
420 case execpb.StepConclusion_STEP_CONCLUSION_SKIPPED:
421 return ermite.CiConclusion_CI_CONCLUSION_SKIPPED
422 case execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT:
423 return ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT
424 }
425 return ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED
426}
427
428// finalSteps closes the steps that never completed (the executor died or was
429// stopped) with the conclusion of the job, and returns all snapshots in order.
430func (j *jobRun) finalSteps(jobConclusion ermite.CiConclusion) []*ermite.StepUpdate {
431 j.mu.Lock()
432 defer j.mu.Unlock()
433 end := ermite.CiConclusion_CI_CONCLUSION_FAILURE
434 if jobConclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED || jobConclusion == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
435 end = ermite.CiConclusion_CI_CONCLUSION_CANCELLED
436 }
437 var out []*ermite.StepUpdate
438 for _, u := range j.steps {
439 if u.State != ermite.StepState_STEP_STATE_COMPLETED {
440 u = cloneStep(u)
441 u.State, u.Conclusion, u.CompletedAt = ermite.StepState_STEP_STATE_COMPLETED, end, timestamppb.Now()
442 }
443 out = append(out, u)
444 }
445 sort.Slice(out, func(a, b int) bool { return out[a].Number < out[b].Number })
446 return out
447}
448
449// ---- the job
450
451// run runs the whole job: acknowledgement, sandbox, executor session, final
452// report. It returns when everything is over and the sandbox is gone.
453func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
454 defer j.cancel(nil)
455 ctx := j.ctx
456
457 if !j.acknowledge(ctx, shutdown) {
458 return
459 }
460 var bg sync.WaitGroup
461 bg.Add(2)
462 go func() { defer bg.Done(); j.heartbeats(ctx) }()
463 go func() { defer bg.Done(); j.watch(ctx) }()
464 // The context is cancelled first, so that the heartbeat and watch loops end.
465 defer func() { j.cancel(nil); bg.Wait() }()
466
467 j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.log)
468 j.outbox.start(ctx)
469
470 stopHard := j.startStopTimers(ctx, shutdown)
471 // The hard stop frees the reader of the stream if it waits for room in the
472 // outbox (the server keeps failing): nothing else would wake it.
473 go func() {
474 select {
475 case <-stopHard:
476 j.outbox.releaseProducers()
477 case <-ctx.Done():
478 }
479 }()
480 out := j.execute(ctx, rt, stopHard)
481 j.report(ctx, out, shutdown)
482}
483
484// acknowledge sends the first heartbeat, before anything else. A shutdown ends
485// the wait: the job is not started and not acknowledged, and the server gives it
486// to another runner when no heartbeat comes within 30 s (runner.proto, FetchJob:
487// this is not a failure of the job). It reports whether the job goes on.
488func (j *jobRun) acknowledge(ctx context.Context, shutdown <-chan struct{}) bool {
489 ackCtx, stopAck := context.WithCancel(ctx)
490 defer stopAck()
491 go func() {
492 select {
493 case <-shutdown:
494 stopAck()
495 case <-ackCtx.Done():
496 }
497 }()
498 if j.ack(ackCtx) {
499 j.timeSet("ack_ms", time.Since(j.receivedAt))
500 return true
501 }
502 select {
503 case <-shutdown:
504 j.log.Warn().Msg("the runner is shutting down before the job was acknowledged: the server will give it to another runner")
505 default:
506 j.log.Warn().Msg("the assignment was lost before the job started")
507 }
508 return false
509}
510
511// startStopTimers arms the two reasons to stop the job from the host, each a
512// cancellation followed by a hard stop of the sandbox (the returned channel is
513// closed) if the executor does not answer: a shutdown of the runner
514// (ShutdownGrace), and the job timeout, which the executor enforces itself but
515// a stuck executor would not (TimeoutGrace after the deadline: the escalation
516// SIGINT, SIGTERM, SIGKILL takes 10 s). The goroutines end with ctx.
517func (j *jobRun) startStopTimers(ctx context.Context, shutdown <-chan struct{}) <-chan struct{} {
518 stopHard := make(chan struct{})
519 var hardOnce sync.Once
520 hard := func() { hardOnce.Do(func() { close(stopHard) }) }
521 j.hard = hard
522 // after waits grace, then stops the sandbox; it returns early with ctx.
523 after := func(grace time.Duration) {
524 timer := time.NewTimer(grace)
525 defer timer.Stop()
526 select {
527 case <-timer.C:
528 hard()
529 case <-ctx.Done():
530 }
531 }
532 go func() {
533 select {
534 case <-shutdown:
535 case <-ctx.Done():
536 return
537 }
538 j.cancelMu.Lock()
539 j.stopping = true
540 j.cancelMu.Unlock()
541 j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)
542 after(j.t.ShutdownGrace)
543 }()
544 go func() {
545 wait := time.NewTimer(time.Until(j.deadline()))
546 defer wait.Stop()
547 select {
548 case <-wait.C:
549 case <-ctx.Done():
550 return
551 }
552 j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed")
553 j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)
554 after(j.t.TimeoutGrace)
555 }()
556 return stopHard
557}
558
559// report ends the job towards the server: the log is flushed, then CompleteJob.
560// Nothing is sent to a job that the server has ended.
561func (j *jobRun) report(ctx context.Context, out outcome, shutdown <-chan struct{}) {
562 if err := j.abandoned(); err != nil {
563 j.log.Warn().Err(err).Msg("the server ended the job: no result is reported")
564 return
565 }
566 // The report is made even when the runner is shutting down: it uses its own
567 // deadline.
568 budget := j.t.CompleteDeadline
569 select {
570 case <-shutdown:
571 // The service manager waits only so long (see Timing): once the runner's
572 // shutdown channel is closed, whatever the job's outcome, the report gets
573 // what is left of the budget. The channel is read here, not the flag set
574 // by the stop timers, which may not have run yet.
575 budget = min(budget, j.t.ShutdownReport)
576 default:
577 }
578 rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget)
579 defer cancel()
580 // Output limits are checked before the log is flushed: a job whose outputs
581 // the server would refuse fails without outputs, with a line in the log
582 // (runner.proto: CompleteJobRequest.outputs), and lines sent after
583 // CompleteJob are rejected.
584 if msg := checkOutputs(out.outputs); msg != "" {
585 j.sandboxError(msg)
586 out = failure()
587 }
588 // The flush gets half of the budget: when the server keeps failing the
589 // uploads, the other half is still there for CompleteJob, which is what
590 // ends the job for the server.
591 fctx, fcancel := context.WithTimeout(rctx, budget/2)
592 j.outbox.Close(fctx)
593 fcancel()
594 if err := j.abandoned(); err != nil {
595 return
596 }
597 j.reportDroppedAnnotations(rctx, budget/4)
598 j.complete(rctx, out.conclusion, out.outputs, out.shutdown)
599}
600
601// outcome is how a job ended, as far as the host can tell.
602type outcome struct {
603 conclusion ermite.CiConclusion
604 outputs map[string]string
605 // shutdown is true when the runner's shutdown cut the job short.
606 shutdown bool
607}
608
609func failure() outcome { return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE} }
610
611// execute starts the sandbox, runs the job in it and returns how it ended.
612func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct{}) outcome {
613 var err error
614 j.setOrigins(j.job.Contexts, rt.Cfg.URL, rt.Cfg.GuestForgeOrigin)
615 if j.contexts, err = RewriteServerURLs(j.job.Contexts, rt.Cfg.GuestForgeOrigin); err != nil {
616 j.failEarly(fmt.Sprintf("The job's contexts cannot be prepared: %v", err))
617 return failure()
618 }
619 if j.contexts, err = SetGitHubToken(j.contexts, j.job.GithubToken); err != nil {
620 j.failEarly(fmt.Sprintf("The job's contexts cannot be prepared: %v", err))
621 return failure()
622 }
623 sb, err := rt.Driver.New()
624 if err != nil {
625 j.failEarly(fmt.Sprintf("The runner has no free sandbox: %v", err))
626 return failure()
627 }
628 // Stop runs to completion even when the job context is done.
629 defer j.stopSandbox(ctx, rt, sb)
630 j.features = j.loadFeatures(ctx, rt)
631 j.results = j.resultsOn(rt)
632
633 start := time.Now()
634 conn, err := j.startSandbox(ctx, rt, sb, stopHard)
635 j.timeSet("sandbox_ms", time.Since(start))
636 if timed, ok := sb.(sandbox.Timed); ok && err == nil {
637 j.mu.Lock()
638 for k, v := range timed.StartTimings() {
639 j.timeline["sb_"+k] = max(0, v)
640 }
641 j.mu.Unlock()
642 }
643 if err != nil {
644 if ctx.Err() != nil || j.cancelRequested() {
645 return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: j.isStopping()}
646 }
647 j.failEarly(fmt.Sprintf("The sandbox did not start: %v", err))
648 return failure()
649 }
650 defer func() { _ = conn.Close() }()
651
652 // Hard stops (grace elapsed, or the job was abandoned) close the stream. The
653 // goroutine ends when this function returns, whatever happens to ctx.
654 sessionDone := make(chan struct{})
655 defer close(sessionDone)
656 go func() {
657 select {
658 case <-stopHard:
659 _ = conn.Close()
660 case <-ctx.Done():
661 _ = conn.Close()
662 case <-sessionDone:
663 }
664 }()
665
666 start = time.Now()
667 exit, err := j.drive(conn, rt)
668 j.timeSet("run_ms", time.Since(start))
669 return j.outcomeOf(ctx, exit, err)
670}
671
672// stopSandbox destroys the sandbox; a cleanup that left something behind stops
673// the runner from taking jobs until it is reaped.
674func (j *jobRun) stopSandbox(ctx context.Context, rt *Runner, sb sandbox.Sandbox) {
675 sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)
676 defer cancel()
677 start := time.Now()
678 defer func() { j.timeSet("stop_ms", time.Since(start)) }()
679 if err := sb.Stop(sctx); err != nil {
680 j.log.Error().Err(err).Msg("stopping the sandbox")
681 if errors.Is(err, sandbox.ErrLeaked) {
682 rt.markLeaked()
683 }
684 }
685}
686
687// startSandbox starts the sandbox; the start is bounded by the job context and
688// by the hard stop.
689func (j *jobRun) startSandbox(ctx context.Context, rt *Runner, sb sandbox.Sandbox, stopHard <-chan struct{}) (sandbox.Conn, error) {
690 startCtx, cancelStart := context.WithCancel(ctx)
691 defer cancelStart()
692 stopWatch := make(chan struct{})
693 defer close(stopWatch)
694 go func() {
695 select {
696 case <-stopHard:
697 cancelStart()
698 case <-stopWatch:
699 }
700 }()
701 return sb.Start(startCtx, rt.spec(j.job, j.resultsHandler(rt)))
702}
703
704// drive sends the job to the executor and handles its frames until it exits.
705func (j *jobRun) drive(conn sandbox.Conn, rt *Runner) (*execpb.Exit, error) {
706 run := &execpb.Run{Job: &execpb.JobSpec{
707 WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,
708 Matrix: j.job.Matrix, Contexts: j.contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),
709 DeadlineMs: max(1, time.Until(j.deadline()).Milliseconds()),
710 RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: rt.Cfg.environment(),
711 SendAnnotations: j.features.annotations, SendSummaries: j.features.summaries,
712 // Job.secrets and Job.vars are not in this frame (it is limited to
713 // 2 MiB): DriveJob sends them in NamedValue frames before it. The host
714 // never logs them nor stores them. An old server leaves trusted false
715 // because it does not send it, so the executor says "not available"
716 // only when the server has ci.secrets.
717 SecretsWithheld: j.features.secrets && !j.job.Trusted,
718 }}
719 run.Job.IdTokenUrl, run.Job.IdTokenRequestToken = j.idTokenRequest()
720 if p := j.job.Permissions; p != nil && j.features.secrets {
721 run.Job.TokenPermissions = &execpb.TokenPermissions{Levels: p.GetLevels()}
722 }
723 run.Job.CachePort, run.Job.CacheToken, run.Job.CacheMode, run.Job.Results = j.runtimeFor(rt)
724 _, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
725 Hello: func(h *execpb.Hello) {
726 j.mu.Lock()
727 defer j.mu.Unlock()
728 j.timeline.addGuestBoot(h.GetBootMs())
729 },
730 StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
731 Log: j.logLine,
732 JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
733 Resolve: j.resolveActions,
734 StepAnnotations: j.stepAnnotations, StepSummary: j.stepSummary,
735 Secrets: j.job.Secrets, Vars: j.job.Vars,
736 })
737 return exit, err
738}
739
740// idTokenRequest returns Job.id_token_url, its origin rewritten to the guest
741// forge as for github.server_url (RewriteServerURLs), and
742// Job.id_token_request_token: the ACTIONS_ID_TOKEN_REQUEST_URL and
743// ACTIONS_ID_TOKEN_REQUEST_TOKEN of the steps. Both are empty unless the
744// server sent both. The token is never logged.
745func (j *jobRun) idTokenRequest() (url, token string) {
746 url, token = j.job.GetIdTokenUrl(), j.job.GetIdTokenRequestToken()
747 if url == "" || token == "" {
748 return "", ""
749 }
750 if j.guestForge != "" {
751 url, _ = rewriteOrigin(url, j.publicOrigin, j.guestForge)
752 }
753 return url, token
754}
755
756// outcomeOf decides how the job ended from what the executor reported and what
757// the host asked of it.
758func (j *jobRun) outcomeOf(ctx context.Context, exit *execpb.Exit, driveErr error) outcome {
759 j.mu.Lock()
760 res := j.result
761 j.mu.Unlock()
762
763 switch {
764 case j.violated():
765 // The step numbers or lines of the executor were wrong: whatever it
766 // reported is not trusted, the job fails (the reason is in the log).
767 return failure()
768 case res != nil:
769 c := conclusionOf(res.Conclusion)
770 if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {
771 c = ermite.CiConclusion_CI_CONCLUSION_FAILURE
772 }
773 if res.Error != "" {
774 j.log.Info().Str("error", res.Error).Msg("the job reported an error")
775 }
776 return outcome{conclusion: c, outputs: res.Outputs, shutdown: res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED}
777 case j.cancelRequested() || ctx.Err() != nil:
778 // The executor did not report: it was stopped. A timeout wins over a
779 // shutdown or a user cancel that came later, as on the server (the first
780 // reason decides, and the timeout is the first when it fired first).
781 switch {
782 case j.firstCancelKind() == execpb.CancelKind_CANCEL_KIND_TIMEOUT:
783 return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT}
784 case j.isStopping():
785 return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: true}
786 }
787 return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED}
788 }
789 why := "the executor ended without a result"
790 if driveErr != nil {
791 why = driveErr.Error()
792 } else if exit != nil && exit.Error != "" {
793 why = exit.Error
794 }
795 j.sandboxError(fmt.Sprintf("The job's sandbox ended unexpectedly: %s", why))
796 return failure()
797}
798
799// timeoutMinutes is the job timeout: at least 1, 360 when the job has none
800// (runner.proto: Job.timeout_minutes).
801func (j *jobRun) timeoutMinutes() int {
802 if m := int(j.job.TimeoutMinutes); m > 0 {
803 return m
804 }
805 return 360
806}
807
808// deadline is when the job times out, counted from its receipt.
809func (j *jobRun) deadline() time.Time {
810 return j.receivedAt.Add(time.Duration(j.timeoutMinutes()) * time.Minute)
811}
812
813func (j *jobRun) isStopping() bool {
814 j.cancelMu.Lock()
815 defer j.cancelMu.Unlock()
816 return j.stopping
817}
818
819// failEarly reports a job that could not start: step 1 fails with the reason
820// in its log.
821func (j *jobRun) failEarly(msg string) {
822 now := time.Now()
823 j.stepStarted(&execpb.StepStarted{Number: 1, Name: "Set up job", UnixNano: now.UnixNano()})
824 j.outbox.Log(logLine{step: 1, unixNano: now.UnixNano(), text: "##[error]" + msg})
825 j.stepCompleted(&execpb.StepCompleted{Number: 1, Name: "Set up job", Conclusion: execpb.StepConclusion_STEP_CONCLUSION_FAILURE, UnixNano: time.Now().UnixNano()})
826}
827
828// sandboxError writes the reason into the log of the step that was running.
829func (j *jobRun) sandboxError(msg string) {
830 j.mu.Lock()
831 var last int32 = 1
832 for n := range j.steps {
833 last = max(last, n)
834 }
835 j.mu.Unlock()
836 j.outbox.Log(logLine{step: last, unixNano: time.Now().UnixNano(), text: "##[error]" + msg})
837}
838
839// complete reports the end of the job. CompleteJob is idempotent, so a lost
840// response is retried until it is answered (applied or not).
841func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, outputs map[string]string, shutdown bool) {
842 start := time.Now()
843 defer func() { j.timeSet("complete_ms", time.Since(start)) }()
844 req := &ermite.CompleteJobRequest{
845 JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,
846 RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,
847 Steps: j.finalSteps(conclusion),
848 // The annotations no UpdateJob has carried: all of them with the
849 // step they belong to, whole lists (replace semantics).
850 Annotations: j.allPendingAnnotations(),
851 }
852 err := j.sendComplete(ctx, req, conclusion)
853 if err == nil || ctx.Err() != nil || classify(err) != kindPermanent {
854 return
855 }
856 // The server refused the result itself (invalid_argument, or
857 // failed_precondition that is not the end of the job). CompleteJob is
858 // "applied whole or not at all", so the job is still open: say why in the
859 // log, then report a plain failure without outputs and with steps that are
860 // valid by construction. Without this the job would stay running until the
861 // heartbeat deadline of the server and then be reported as a lost runner.
862 j.errorLog(ctx, fmt.Sprintf("The runner could not report the result of the job: %s. The job is reported as failed.", connect.CodeOf(err)))
863 j.log.Error().Err(err).Msg("CompleteJob refused: retrying as a failure without outputs")
864 req = &ermite.CompleteJobRequest{
865 JobId: j.job.JobId, Conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE,
866 Steps: j.rebuiltSteps(),
867 }
868 if err := j.sendComplete(ctx, req, req.Conclusion); err != nil && ctx.Err() == nil {
869 j.log.Error().Err(err).Msg("CompleteJob refused again, giving up")
870 }
871}
872
873// sendComplete sends the request until the server answers: success, the end of
874// the job, a lost assignment, or a refusal (returned).
875func (j *jobRun) sendComplete(ctx context.Context, req *ermite.CompleteJobRequest, conclusion ermite.CiConclusion) error {
876 b := backoff{min: j.t.BackoffMin, max: j.t.BackoffMax}
877 for {
878 cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
879 resp, err := j.client.CompleteJob(cctx, connect.NewRequest(req))
880 cancel()
881 if err == nil {
882 if resp.Msg.Conclusion != conclusion {
883 j.log.Warn().Str("reported", conclusion.String()).Str("stored", resp.Msg.Conclusion.String()).Msg("the server holds another conclusion")
884 }
885 return nil
886 }
887 if ctx.Err() != nil {
888 j.log.Error().Err(ctx.Err()).Msg("could not report the result")
889 return err
890 }
891 switch classify(err) {
892 case kindEnded, kindAbandoned:
893 return nil
894 case kindPermanent:
895 return err
896 }
897 j.log.Warn().Err(err).Msg("CompleteJob failed, retrying")
898 if !sleep(ctx, b.next()) {
899 return err
900 }
901 }
902}
903
904// errorLog appends a line to the log after the outbox is closed.
905func (j *jobRun) errorLog(ctx context.Context, msg string) {
906 j.appendLogLine(ctx, "##[error]"+msg)
907}
908
909// appendLogLine appends one line, in the last step, after the outbox is closed.
910func (j *jobRun) appendLogLine(ctx context.Context, text string) {
911 j.mu.Lock()
912 var last int32 = 1
913 for n := range j.steps {
914 last = max(last, n)
915 }
916 j.mu.Unlock()
917 j.outbox.AppendNow(ctx, logLine{step: last, unixNano: time.Now().UnixNano(), text: text})
918}
919
920// rebuiltSteps returns every known step as a valid, completed snapshot: number
921// in 1..maxSteps, a name of at most 1,024 bytes, a conclusion the server
922// accepts (success, failure, cancelled, skipped), both times set, and no step
923// that finished before it began. A step that was running is failed.
924func (j *jobRun) rebuiltSteps() []*ermite.StepUpdate {
925 j.mu.Lock()
926 defer j.mu.Unlock()
927 now := timestamppb.Now()
928 var out []*ermite.StepUpdate
929 for n, u := range j.steps {
930 if n < 1 || n > maxSteps {
931 continue
932 }
933 c := proto.Clone(u).(*ermite.StepUpdate)
934 c.Name = cutName(c.Name)
935 if c.StartedAt == nil {
936 c.StartedAt = now
937 }
938 if c.State != ermite.StepState_STEP_STATE_COMPLETED {
939 c.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
940 c.CompletedAt = now
941 }
942 switch c.Conclusion {
943 case ermite.CiConclusion_CI_CONCLUSION_SUCCESS, ermite.CiConclusion_CI_CONCLUSION_FAILURE,
944 ermite.CiConclusion_CI_CONCLUSION_CANCELLED, ermite.CiConclusion_CI_CONCLUSION_SKIPPED:
945 default:
946 c.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
947 }
948 c.State = ermite.StepState_STEP_STATE_COMPLETED
949 if c.CompletedAt == nil || c.CompletedAt.AsTime().Before(c.StartedAt.AsTime()) {
950 c.CompletedAt = c.StartedAt
951 }
952 out = append(out, c)
953 }
954 sort.Slice(out, func(a, b int) bool { return out[a].Number < out[b].Number })
955 return out
956}
957
958// Limits of CompleteJobRequest.outputs.
959const (
960 maxOutputBytes = 1 << 20
961 maxOutputName = 255
962)
963
964// checkOutputs returns a message when the server would refuse the outputs: a
965// name of 1 to 255 letters, digits, "_" and "-" starting with a letter or "_",
966// no two names equal ignoring case, each value and the sum of names and values
967// at most 1 MiB.
968func checkOutputs(outputs map[string]string) string {
969 total := 0
970 seen := map[string]string{}
971 names := make([]string, 0, len(outputs))
972 for n := range outputs {
973 names = append(names, n)
974 }
975 sort.Strings(names)
976 for _, n := range names {
977 v := outputs[n]
978 if !validOutputName(n) {
979 return fmt.Sprintf("Job output name %q is not valid: use 1 to 255 letters, digits, '_' and '-', starting with a letter or '_'.", n)
980 }
981 if prev, dup := seen[strings.ToLower(n)]; dup {
982 return fmt.Sprintf("Job outputs %q and %q differ only by case.", prev, n)
983 }
984 seen[strings.ToLower(n)] = n
985 if len(v) > maxOutputBytes {
986 return fmt.Sprintf("Job output %q is larger than 1 MiB.", n)
987 }
988 total += len(n) + len(v)
989 }
990 if total > maxOutputBytes {
991 return "The job outputs are larger than 1 MiB in total."
992 }
993 return ""
994}
995
996func validOutputName(n string) bool {
997 if n == "" || len(n) > maxOutputName {
998 return false
999 }
1000 for i, r := range n {
1001 letter := r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r == '_'
1002 if ok := letter || (i > 0 && (r >= '0' && r <= '9' || r == '-')); !ok {
1003 return false
1004 }
1005 }
1006 return true
1007}