detached ee276af

refactor(host): split run and execute into acknowledge, startStopTimers, report, startSandbox, drive and outcomeOf (G2)

commit
ee276afd055073ac3ace42b550c540e0d5770f0f
author
brice · · 2026-10-12 19:43 UTC
parent
c5b1e6d
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
1 file · +131 -89
Unified
UnifiedSplit
M internal/host/job.go +131 -89
@@ -422,27 +422,7 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
422422 defer j.cancel(nil)
423423 ctx := j.ctx
424424
425- // 1. The acknowledgement comes first, before anything else. A shutdown ends
426- // the wait: the job is not started and not acknowledged, and the server
427- // gives it to another runner when no heartbeat comes within 30 s (runner.proto,
428- // FetchJob: this is not a failure of the job).
429- ackCtx, stopAck := context.WithCancel(ctx)
430- go func() {
431- select {
432- case <-shutdown:
433- stopAck()
434- case <-ackCtx.Done():
435- }
436- }()
437- acked := j.ack(ackCtx)
438- stopAck()
439- if !acked {
440- select {
441- case <-shutdown:
442- j.log.Warn().Msg("the runner is shutting down before the job was acknowledged: the server will give it to another runner")
443- default:
444- j.log.Warn().Msg("the assignment was lost before the job started")
445- }
425+ if !j.acknowledge(ctx, shutdown) {
446426 return
447427 }
448428 var bg sync.WaitGroup
@@ -455,15 +435,58 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
455435 j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.log)
456436 j.outbox.start(ctx)
457437
458- // Two reasons to stop the job from the host, each a cancellation followed
459- // by a hard stop of the sandbox if the executor does not answer: a shutdown
460- // of the runner (ShutdownGrace), and the job timeout, which the executor
461- // enforces itself but a stuck executor would not (TimeoutGrace after the
462- // deadline: the escalation SIGINT, SIGTERM, SIGKILL takes 10 s).
438+ stopHard := j.startStopTimers(ctx, shutdown)
439+ out := j.execute(ctx, rt, stopHard)
440+ j.report(ctx, out)
441+}
442+
443+// acknowledge sends the first heartbeat, before anything else. A shutdown ends
444+// the wait: the job is not started and not acknowledged, and the server gives it
445+// to another runner when no heartbeat comes within 30 s (runner.proto, FetchJob:
446+// this is not a failure of the job). It reports whether the job goes on.
447+func (j *jobRun) acknowledge(ctx context.Context, shutdown <-chan struct{}) bool {
448+ ackCtx, stopAck := context.WithCancel(ctx)
449+ defer stopAck()
450+ go func() {
451+ select {
452+ case <-shutdown:
453+ stopAck()
454+ case <-ackCtx.Done():
455+ }
456+ }()
457+ if j.ack(ackCtx) {
458+ return true
459+ }
460+ select {
461+ case <-shutdown:
462+ j.log.Warn().Msg("the runner is shutting down before the job was acknowledged: the server will give it to another runner")
463+ default:
464+ j.log.Warn().Msg("the assignment was lost before the job started")
465+ }
466+ return false
467+}
468+
469+// startStopTimers arms the two reasons to stop the job from the host, each a
470+// cancellation followed by a hard stop of the sandbox (the returned channel is
471+// closed) if the executor does not answer: a shutdown of the runner
472+// (ShutdownGrace), and the job timeout, which the executor enforces itself but
473+// a stuck executor would not (TimeoutGrace after the deadline: the escalation
474+// SIGINT, SIGTERM, SIGKILL takes 10 s). The goroutines end with ctx.
475+func (j *jobRun) startStopTimers(ctx context.Context, shutdown <-chan struct{}) <-chan struct{} {
463476 stopHard := make(chan struct{})
464477 var hardOnce sync.Once
465478 hard := func() { hardOnce.Do(func() { close(stopHard) }) }
466479 j.hard = hard
480+ // after waits grace, then stops the sandbox; it returns early with ctx.
481+ after := func(grace time.Duration) {
482+ timer := time.NewTimer(grace)
483+ defer timer.Stop()
484+ select {
485+ case <-timer.C:
486+ hard()
487+ case <-ctx.Done():
488+ }
489+ }
467490 go func() {
468491 select {
469492 case <-shutdown:
@@ -474,13 +497,7 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
474497 j.stopping = true
475498 j.cancelMu.Unlock()
476499 j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)
477- timer := time.NewTimer(j.t.ShutdownGrace)
478- defer timer.Stop()
479- select {
480- case <-timer.C:
481- hard()
482- case <-ctx.Done():
483- }
500+ after(j.t.ShutdownGrace)
484501 }()
485502 go func() {
486503 wait := time.NewTimer(time.Until(j.deadline()))
@@ -492,87 +509,70 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
492509 }
493510 j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed")
494511 j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)
495- timer := time.NewTimer(j.t.TimeoutGrace)
496- defer timer.Stop()
497- select {
498- case <-timer.C:
499- hard()
500- case <-ctx.Done():
501- }
512+ after(j.t.TimeoutGrace)
502513 }()
514+ return stopHard
515+}
503516
504- conclusion, outputs, shutdownFlag := j.execute(ctx, rt, stopHard)
505-
506- // 2. Nothing more is sent to a job that is over.
517+// report ends the job towards the server: the log is flushed, then CompleteJob.
518+// Nothing is sent to a job that the server has ended.
519+func (j *jobRun) report(ctx context.Context, out outcome) {
507520 if err := j.abandoned(); err != nil {
508521 j.log.Warn().Err(err).Msg("the server ended the job: no result is reported")
509522 return
510523 }
511524 // The report is made even when the runner is shutting down: it uses its own
512525 // deadline.
513- reportBudget := j.t.CompleteDeadline
526+ budget := j.t.CompleteDeadline
514527 if j.isStopping() {
515528 // The service manager waits only so long (see Timing): the report of a job
516529 // cut short by a shutdown gets what is left of the budget.
517- reportBudget = min(reportBudget, j.t.ShutdownReport)
530+ budget = min(budget, j.t.ShutdownReport)
518531 }
519- rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), reportBudget)
532+ rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget)
520533 defer cancel()
521534 // Output limits are checked before the log is flushed: a job whose outputs
522535 // the server would refuse fails without outputs, with a line in the log
523536 // (runner.proto: CompleteJobRequest.outputs), and lines sent after
524537 // CompleteJob are rejected.
525- if msg := checkOutputs(outputs); msg != "" {
538+ if msg := checkOutputs(out.outputs); msg != "" {
526539 j.sandboxError(msg)
527- outputs, conclusion, shutdownFlag = nil, ermite.CiConclusion_CI_CONCLUSION_FAILURE, false
540+ out = failure()
528541 }
529542 j.outbox.Close(rctx)
530543 if err := j.abandoned(); err != nil {
531544 return
532545 }
533- j.complete(rctx, conclusion, outputs, shutdownFlag)
546+ j.complete(rctx, out.conclusion, out.outputs, out.shutdown)
534547 }
535548
536-// execute starts the sandbox, runs the job in it and returns the conclusion.
537-func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct{}) (ermite.CiConclusion, map[string]string, bool) {
538- fail := func(format string, a ...any) (ermite.CiConclusion, map[string]string, bool) {
539- j.failEarly(fmt.Sprintf(format, a...))
540- return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
541- }
549+// outcome is how a job ended, as far as the host can tell.
550+type outcome struct {
551+ conclusion ermite.CiConclusion
552+ outputs map[string]string
553+ // shutdown is true when the runner's shutdown cut the job short.
554+ shutdown bool
555+}
556+
557+func failure() outcome { return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE} }
558+
559+// execute starts the sandbox, runs the job in it and returns how it ended.
560+func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct{}) outcome {
542561 sb, err := rt.Driver.New()
543562 if err != nil {
544- return fail("The runner has no free sandbox: %v", err)
563+ j.failEarly(fmt.Sprintf("The runner has no free sandbox: %v", err))
564+ return failure()
545565 }
546566 // Stop runs to completion even when the job context is done.
547- defer func() {
548- sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)
549- defer cancel()
550- if err := sb.Stop(sctx); err != nil {
551- j.log.Error().Err(err).Msg("stopping the sandbox")
552- if errors.Is(err, sandbox.ErrLeaked) {
553- rt.markLeaked()
554- }
555- }
556- }()
567+ defer j.stopSandbox(ctx, rt, sb)
557568
558- // Starting the sandbox is bounded by the job context and a hard stop.
559- startCtx, cancelStart := context.WithCancel(ctx)
560- stopWatch := make(chan struct{})
561- go func() {
562- select {
563- case <-stopHard:
564- cancelStart()
565- case <-stopWatch:
566- }
567- }()
568- conn, err := sb.Start(startCtx, rt.spec(j.job))
569- close(stopWatch)
570- cancelStart()
569+ conn, err := j.startSandbox(ctx, rt, sb, stopHard)
571570 if err != nil {
572571 if ctx.Err() != nil || j.cancelRequested() {
573- return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, j.isStopping()
572+ return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: j.isStopping()}
574573 }
575- return fail("The sandbox did not start: %v", err)
574+ j.failEarly(fmt.Sprintf("The sandbox did not start: %v", err))
575+ return failure()
576576 }
577577 defer func() { _ = conn.Close() }()
578578
@@ -590,6 +590,42 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct
590590 }
591591 }()
592592
593+ exit, err := j.drive(conn, rt)
594+ return j.outcomeOf(ctx, exit, err)
595+}
596+
597+// stopSandbox destroys the sandbox; a cleanup that left something behind stops
598+// the runner from taking jobs until it is reaped.
599+func (j *jobRun) stopSandbox(ctx context.Context, rt *Runner, sb sandbox.Sandbox) {
600+ sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)
601+ defer cancel()
602+ if err := sb.Stop(sctx); err != nil {
603+ j.log.Error().Err(err).Msg("stopping the sandbox")
604+ if errors.Is(err, sandbox.ErrLeaked) {
605+ rt.markLeaked()
606+ }
607+ }
608+}
609+
610+// startSandbox starts the sandbox; the start is bounded by the job context and
611+// by the hard stop.
612+func (j *jobRun) startSandbox(ctx context.Context, rt *Runner, sb sandbox.Sandbox, stopHard <-chan struct{}) (sandbox.Conn, error) {
613+ startCtx, cancelStart := context.WithCancel(ctx)
614+ defer cancelStart()
615+ stopWatch := make(chan struct{})
616+ defer close(stopWatch)
617+ go func() {
618+ select {
619+ case <-stopHard:
620+ cancelStart()
621+ case <-stopWatch:
622+ }
623+ }()
624+ return sb.Start(startCtx, rt.spec(j.job))
625+}
626+
627+// drive sends the job to the executor and handles its frames until it exits.
628+func (j *jobRun) drive(conn sandbox.Conn, rt *Runner) (*execpb.Exit, error) {
593629 run := &execpb.Run{Job: &execpb.JobSpec{
594630 WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,
595631 Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),
@@ -601,6 +637,12 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct
601637 Log: j.logLine,
602638 JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
603639 })
640+ return exit, err
641+}
642+
643+// outcomeOf decides how the job ended from what the executor reported and what
644+// the host asked of it.
645+func (j *jobRun) outcomeOf(ctx context.Context, exit *execpb.Exit, driveErr error) outcome {
604646 j.mu.Lock()
605647 res := j.result
606648 j.mu.Unlock()
@@ -609,7 +651,7 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct
609651 case j.violated():
610652 // The step numbers or lines of the executor were wrong: whatever it
611653 // reported is not trusted, the job fails (the reason is in the log).
612- return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
654+ return failure()
613655 case res != nil:
614656 c := conclusionOf(res.Conclusion)
615657 if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {
@@ -618,27 +660,27 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct
618660 if res.Error != "" {
619661 j.log.Info().Str("error", res.Error).Msg("the job reported an error")
620662 }
621- return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED
663+ return outcome{conclusion: c, outputs: res.Outputs, shutdown: res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED}
622664 case j.cancelRequested() || ctx.Err() != nil:
623665 // The executor did not report: it was stopped. A timeout wins over a
624666 // shutdown or a user cancel that came later, as on the server (the first
625667 // reason decides, and the timeout is the first when it fired first).
626668 switch {
627669 case j.firstCancelKind() == execpb.CancelKind_CANCEL_KIND_TIMEOUT:
628- return ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT, nil, false
670+ return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT}
629671 case j.isStopping():
630- return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, true
672+ return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: true}
631673 }
632- return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, false
674+ return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED}
633675 }
634676 why := "the executor ended without a result"
635- if err != nil {
636- why = err.Error()
677+ if driveErr != nil {
678+ why = driveErr.Error()
637679 } else if exit != nil && exit.Error != "" {
638680 why = exit.Error
639681 }
640682 j.sandboxError(fmt.Sprintf("The job's sandbox ended unexpectedly: %s", why))
641- return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
683+ return failure()
642684 }
643685
644686 // timeoutMinutes is the job timeout: at least 1, 360 when the job has none
@@ -422,27 +422,7 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
422 defer j.cancel(nil)422 defer j.cancel(nil)
423 ctx := j.ctx423 ctx := j.ctx
424 424
425- // 1. The acknowledgement comes first, before anything else. A shutdown ends425+ if !j.acknowledge(ctx, shutdown) {
426- // the wait: the job is not started and not acknowledged, and the server
427- // gives it to another runner when no heartbeat comes within 30 s (runner.proto,
428- // FetchJob: this is not a failure of the job).
429- ackCtx, stopAck := context.WithCancel(ctx)
430- go func() {
431- select {
432- case <-shutdown:
433- stopAck()
434- case <-ackCtx.Done():
435- }
436- }()
437- acked := j.ack(ackCtx)
438- stopAck()
439- if !acked {
440- select {
441- case <-shutdown:
442- j.log.Warn().Msg("the runner is shutting down before the job was acknowledged: the server will give it to another runner")
443- default:
444- j.log.Warn().Msg("the assignment was lost before the job started")
445- }
446 return426 return
447 }427 }
448 var bg sync.WaitGroup428 var bg sync.WaitGroup
@@ -455,15 +435,58 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
455 j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.log)435 j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.log)
456 j.outbox.start(ctx)436 j.outbox.start(ctx)
457 437
458- // Two reasons to stop the job from the host, each a cancellation followed438+ stopHard := j.startStopTimers(ctx, shutdown)
459- // by a hard stop of the sandbox if the executor does not answer: a shutdown439+ out := j.execute(ctx, rt, stopHard)
460- // of the runner (ShutdownGrace), and the job timeout, which the executor440+ j.report(ctx, out)
461- // enforces itself but a stuck executor would not (TimeoutGrace after the441+}
462- // deadline: the escalation SIGINT, SIGTERM, SIGKILL takes 10 s).442+
443+// acknowledge sends the first heartbeat, before anything else. A shutdown ends
444+// the wait: the job is not started and not acknowledged, and the server gives it
445+// to another runner when no heartbeat comes within 30 s (runner.proto, FetchJob:
446+// this is not a failure of the job). It reports whether the job goes on.
447+func (j *jobRun) acknowledge(ctx context.Context, shutdown <-chan struct{}) bool {
448+ ackCtx, stopAck := context.WithCancel(ctx)
449+ defer stopAck()
450+ go func() {
451+ select {
452+ case <-shutdown:
453+ stopAck()
454+ case <-ackCtx.Done():
455+ }
456+ }()
457+ if j.ack(ackCtx) {
458+ return true
459+ }
460+ select {
461+ case <-shutdown:
462+ j.log.Warn().Msg("the runner is shutting down before the job was acknowledged: the server will give it to another runner")
463+ default:
464+ j.log.Warn().Msg("the assignment was lost before the job started")
465+ }
466+ return false
467+}
468+
469+// startStopTimers arms the two reasons to stop the job from the host, each a
470+// cancellation followed by a hard stop of the sandbox (the returned channel is
471+// closed) if the executor does not answer: a shutdown of the runner
472+// (ShutdownGrace), and the job timeout, which the executor enforces itself but
473+// a stuck executor would not (TimeoutGrace after the deadline: the escalation
474+// SIGINT, SIGTERM, SIGKILL takes 10 s). The goroutines end with ctx.
475+func (j *jobRun) startStopTimers(ctx context.Context, shutdown <-chan struct{}) <-chan struct{} {
463 stopHard := make(chan struct{})476 stopHard := make(chan struct{})
464 var hardOnce sync.Once477 var hardOnce sync.Once
465 hard := func() { hardOnce.Do(func() { close(stopHard) }) }478 hard := func() { hardOnce.Do(func() { close(stopHard) }) }
466 j.hard = hard479 j.hard = hard
480+ // after waits grace, then stops the sandbox; it returns early with ctx.
481+ after := func(grace time.Duration) {
482+ timer := time.NewTimer(grace)
483+ defer timer.Stop()
484+ select {
485+ case <-timer.C:
486+ hard()
487+ case <-ctx.Done():
488+ }
489+ }
467 go func() {490 go func() {
468 select {491 select {
469 case <-shutdown:492 case <-shutdown:
@@ -474,13 +497,7 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
474 j.stopping = true497 j.stopping = true
475 j.cancelMu.Unlock()498 j.cancelMu.Unlock()
476 j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)499 j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)
477- timer := time.NewTimer(j.t.ShutdownGrace)500+ after(j.t.ShutdownGrace)
478- defer timer.Stop()
479- select {
480- case <-timer.C:
481- hard()
482- case <-ctx.Done():
483- }
484 }()501 }()
485 go func() {502 go func() {
486 wait := time.NewTimer(time.Until(j.deadline()))503 wait := time.NewTimer(time.Until(j.deadline()))
@@ -492,87 +509,70 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
492 }509 }
493 j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed")510 j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed")
494 j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)511 j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)
495- timer := time.NewTimer(j.t.TimeoutGrace)512+ after(j.t.TimeoutGrace)
496- defer timer.Stop()
497- select {
498- case <-timer.C:
499- hard()
500- case <-ctx.Done():
501- }
502 }()513 }()
514+ return stopHard
515+}
503 516
504- conclusion, outputs, shutdownFlag := j.execute(ctx, rt, stopHard)517+// report ends the job towards the server: the log is flushed, then CompleteJob.
505-518+// Nothing is sent to a job that the server has ended.
506- // 2. Nothing more is sent to a job that is over.519+func (j *jobRun) report(ctx context.Context, out outcome) {
507 if err := j.abandoned(); err != nil {520 if err := j.abandoned(); err != nil {
508 j.log.Warn().Err(err).Msg("the server ended the job: no result is reported")521 j.log.Warn().Err(err).Msg("the server ended the job: no result is reported")
509 return522 return
510 }523 }
511 // The report is made even when the runner is shutting down: it uses its own524 // The report is made even when the runner is shutting down: it uses its own
512 // deadline.525 // deadline.
513- reportBudget := j.t.CompleteDeadline526+ budget := j.t.CompleteDeadline
514 if j.isStopping() {527 if j.isStopping() {
515 // The service manager waits only so long (see Timing): the report of a job528 // The service manager waits only so long (see Timing): the report of a job
516 // cut short by a shutdown gets what is left of the budget.529 // cut short by a shutdown gets what is left of the budget.
517- reportBudget = min(reportBudget, j.t.ShutdownReport)530+ budget = min(budget, j.t.ShutdownReport)
518 }531 }
519- rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), reportBudget)532+ rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget)
520 defer cancel()533 defer cancel()
521 // Output limits are checked before the log is flushed: a job whose outputs534 // Output limits are checked before the log is flushed: a job whose outputs
522 // the server would refuse fails without outputs, with a line in the log535 // the server would refuse fails without outputs, with a line in the log
523 // (runner.proto: CompleteJobRequest.outputs), and lines sent after536 // (runner.proto: CompleteJobRequest.outputs), and lines sent after
524 // CompleteJob are rejected.537 // CompleteJob are rejected.
525- if msg := checkOutputs(outputs); msg != "" {538+ if msg := checkOutputs(out.outputs); msg != "" {
526 j.sandboxError(msg)539 j.sandboxError(msg)
527- outputs, conclusion, shutdownFlag = nil, ermite.CiConclusion_CI_CONCLUSION_FAILURE, false540+ out = failure()
528 }541 }
529 j.outbox.Close(rctx)542 j.outbox.Close(rctx)
530 if err := j.abandoned(); err != nil {543 if err := j.abandoned(); err != nil {
531 return544 return
532 }545 }
533- j.complete(rctx, conclusion, outputs, shutdownFlag)546+ j.complete(rctx, out.conclusion, out.outputs, out.shutdown)
534 }547 }
535 548
536-// execute starts the sandbox, runs the job in it and returns the conclusion.549+// outcome is how a job ended, as far as the host can tell.
537-func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct{}) (ermite.CiConclusion, map[string]string, bool) {550+type outcome struct {
538- fail := func(format string, a ...any) (ermite.CiConclusion, map[string]string, bool) {551+ conclusion ermite.CiConclusion
539- j.failEarly(fmt.Sprintf(format, a...))552+ outputs map[string]string
540- return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false553+ // shutdown is true when the runner's shutdown cut the job short.
541- }554+ shutdown bool
555+}
556+
557+func failure() outcome { return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE} }
558+
559+// execute starts the sandbox, runs the job in it and returns how it ended.
560+func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct{}) outcome {
542 sb, err := rt.Driver.New()561 sb, err := rt.Driver.New()
543 if err != nil {562 if err != nil {
544- return fail("The runner has no free sandbox: %v", err)563+ j.failEarly(fmt.Sprintf("The runner has no free sandbox: %v", err))
564+ return failure()
545 }565 }
546 // Stop runs to completion even when the job context is done.566 // Stop runs to completion even when the job context is done.
547- defer func() {567+ defer j.stopSandbox(ctx, rt, sb)
548- sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)
549- defer cancel()
550- if err := sb.Stop(sctx); err != nil {
551- j.log.Error().Err(err).Msg("stopping the sandbox")
552- if errors.Is(err, sandbox.ErrLeaked) {
553- rt.markLeaked()
554- }
555- }
556- }()
557 568
558- // Starting the sandbox is bounded by the job context and a hard stop.569+ conn, err := j.startSandbox(ctx, rt, sb, stopHard)
559- startCtx, cancelStart := context.WithCancel(ctx)
560- stopWatch := make(chan struct{})
561- go func() {
562- select {
563- case <-stopHard:
564- cancelStart()
565- case <-stopWatch:
566- }
567- }()
568- conn, err := sb.Start(startCtx, rt.spec(j.job))
569- close(stopWatch)
570- cancelStart()
571 if err != nil {570 if err != nil {
572 if ctx.Err() != nil || j.cancelRequested() {571 if ctx.Err() != nil || j.cancelRequested() {
573- return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, j.isStopping()572+ return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: j.isStopping()}
574 }573 }
575- return fail("The sandbox did not start: %v", err)574+ j.failEarly(fmt.Sprintf("The sandbox did not start: %v", err))
575+ return failure()
576 }576 }
577 defer func() { _ = conn.Close() }()577 defer func() { _ = conn.Close() }()
578 578
@@ -590,6 +590,42 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct
590 }590 }
591 }()591 }()
592 592
593+ exit, err := j.drive(conn, rt)
594+ return j.outcomeOf(ctx, exit, err)
595+}
596+
597+// stopSandbox destroys the sandbox; a cleanup that left something behind stops
598+// the runner from taking jobs until it is reaped.
599+func (j *jobRun) stopSandbox(ctx context.Context, rt *Runner, sb sandbox.Sandbox) {
600+ sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)
601+ defer cancel()
602+ if err := sb.Stop(sctx); err != nil {
603+ j.log.Error().Err(err).Msg("stopping the sandbox")
604+ if errors.Is(err, sandbox.ErrLeaked) {
605+ rt.markLeaked()
606+ }
607+ }
608+}
609+
610+// startSandbox starts the sandbox; the start is bounded by the job context and
611+// by the hard stop.
612+func (j *jobRun) startSandbox(ctx context.Context, rt *Runner, sb sandbox.Sandbox, stopHard <-chan struct{}) (sandbox.Conn, error) {
613+ startCtx, cancelStart := context.WithCancel(ctx)
614+ defer cancelStart()
615+ stopWatch := make(chan struct{})
616+ defer close(stopWatch)
617+ go func() {
618+ select {
619+ case <-stopHard:
620+ cancelStart()
621+ case <-stopWatch:
622+ }
623+ }()
624+ return sb.Start(startCtx, rt.spec(j.job))
625+}
626+
627+// drive sends the job to the executor and handles its frames until it exits.
628+func (j *jobRun) drive(conn sandbox.Conn, rt *Runner) (*execpb.Exit, error) {
593 run := &execpb.Run{Job: &execpb.JobSpec{629 run := &execpb.Run{Job: &execpb.JobSpec{
594 WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,630 WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,
595 Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),631 Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),
@@ -601,6 +637,12 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct
601 Log: j.logLine,637 Log: j.logLine,
602 JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },638 JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
603 })639 })
640+ return exit, err
641+}
642+
643+// outcomeOf decides how the job ended from what the executor reported and what
644+// the host asked of it.
645+func (j *jobRun) outcomeOf(ctx context.Context, exit *execpb.Exit, driveErr error) outcome {
604 j.mu.Lock()646 j.mu.Lock()
605 res := j.result647 res := j.result
606 j.mu.Unlock()648 j.mu.Unlock()
@@ -609,7 +651,7 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct
609 case j.violated():651 case j.violated():
610 // The step numbers or lines of the executor were wrong: whatever it652 // The step numbers or lines of the executor were wrong: whatever it
611 // reported is not trusted, the job fails (the reason is in the log).653 // reported is not trusted, the job fails (the reason is in the log).
612- return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false654+ return failure()
613 case res != nil:655 case res != nil:
614 c := conclusionOf(res.Conclusion)656 c := conclusionOf(res.Conclusion)
615 if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {657 if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {
@@ -618,27 +660,27 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct
618 if res.Error != "" {660 if res.Error != "" {
619 j.log.Info().Str("error", res.Error).Msg("the job reported an error")661 j.log.Info().Str("error", res.Error).Msg("the job reported an error")
620 }662 }
621- return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED663+ return outcome{conclusion: c, outputs: res.Outputs, shutdown: res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED}
622 case j.cancelRequested() || ctx.Err() != nil:664 case j.cancelRequested() || ctx.Err() != nil:
623 // The executor did not report: it was stopped. A timeout wins over a665 // The executor did not report: it was stopped. A timeout wins over a
624 // shutdown or a user cancel that came later, as on the server (the first666 // shutdown or a user cancel that came later, as on the server (the first
625 // reason decides, and the timeout is the first when it fired first).667 // reason decides, and the timeout is the first when it fired first).
626 switch {668 switch {
627 case j.firstCancelKind() == execpb.CancelKind_CANCEL_KIND_TIMEOUT:669 case j.firstCancelKind() == execpb.CancelKind_CANCEL_KIND_TIMEOUT:
628- return ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT, nil, false670+ return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT}
629 case j.isStopping():671 case j.isStopping():
630- return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, true672+ return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: true}
631 }673 }
632- return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, false674+ return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED}
633 }675 }
634 why := "the executor ended without a result"676 why := "the executor ended without a result"
635- if err != nil {677+ if driveErr != nil {
636- why = err.Error()678+ why = driveErr.Error()
637 } else if exit != nil && exit.Error != "" {679 } else if exit != nil && exit.Error != "" {
638 why = exit.Error680 why = exit.Error
639 }681 }
640 j.sandboxError(fmt.Sprintf("The job's sandbox ended unexpectedly: %s", why))682 j.sandboxError(fmt.Sprintf("The job's sandbox ended unexpectedly: %s", why))
641- return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false683+ return failure()
642 }684 }
643 685
644 // timeoutMinutes is the job timeout: at least 1, 360 when the job has none686 // timeoutMinutes is the job timeout: at least 1, 360 when the job has none

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 4

Tags 11

↑↓ moveenter openesc close