a21f495 fix(host): enforce the job timeout from the receipt of the job and send the remaining time to the executor (M6, m1) · ermite-runner

M internal/execstream/exec.proto +6 -0


@@ -60,6 +60,12 @@ message JobSpec {
60	60		// runner_environment is the runner.environment value ("self-hosted").
61	61		string runner_environment = 10;
62	62		bool debug = 11; // ACTIONS_STEP_DEBUG
	63	+	// deadline_ms is what is left of the job timeout when the host sends this
	64	+	// Run, in milliseconds: the timeout counts from the moment the host received
	65	+	// the job (runner.proto: Job.timeout_minutes), not from the start of the
	66	+	// executor, and booting the sandbox takes time. 0 means "use
	67	+	// timeout_minutes from now".
	68	+	int64 deadline_ms = 12;
63	69		}
64	70
65	71		enum StreamKind {


@@ -60,6 +60,12 @@ message JobSpec {
60	// runner_environment is the runner.environment value ("self-hosted").	60		// runner_environment is the runner.environment value ("self-hosted").
61	string runner_environment = 10;	61		string runner_environment = 10;
62	bool debug = 11; // ACTIONS_STEP_DEBUG	62		bool debug = 11; // ACTIONS_STEP_DEBUG
		63	+	// deadline_ms is what is left of the job timeout when the host sends this
		64	+	// Run, in milliseconds: the timeout counts from the moment the host received
		65	+	// the job (runner.proto: Job.timeout_minutes), not from the start of the
		66	+	// executor, and booting the sandbox takes time. 0 means "use
		67	+	// timeout_minutes from now".
		68	+	int64 deadline_ms = 12;
63	}	69		}
64		70
65	enum StreamKind {	71		enum StreamKind {

M internal/execstream/execpb/exec.pb.go +19 -4 generated file: collapsed

load diff

M internal/executor/job.go +10 -5


@@ -50,8 +50,13 @@ type Config struct {
50	50		MatrixJSON string
51	51		// Repo is the repository name used when the github context lacks one.
52	52		Repo string
53		-	// TimeoutMinutes is the job timeout; 0 means DefaultTimeoutMinutes.
	53	+	// TimeoutMinutes is the job timeout; 0 means DefaultTimeoutMinutes. It is
	54	+	// used for the message and, without Deadline, for the timer.
54	55		TimeoutMinutes int
	56	+	// Deadline is the time left of the job timeout when Run is called, when the
	57	+	// timeout started counting earlier (at the receipt of the job by the host);
	58	+	// 0 means TimeoutMinutes from now.
	59	+	Deadline time.Duration
55	60
56	61		Runner RunnerInfo
57	62		RunnerVersion string

@@ -165,11 +170,11 @@ func Run(cancel *Canceller, cfg Config) Result {
165	170		}
166	171
167	172		func (j *job) run() Result {
168		-	minutes := j.cfg.TimeoutMinutes
169		-	if minutes <= 0 {
170		-	minutes = DefaultTimeoutMinutes
	173	+	limit := j.cfg.Deadline
	174	+	if limit <= 0 {
	175	+	limit = time.Duration(j.timeoutMinutes()) * time.Minute
171	176		}
172		-	timer := time.AfterFunc(time.Duration(minutes)*time.Minute, func() { j.cancel.Cancel(CancelTimeout) })
	177	+	timer := time.AfterFunc(limit, func() { j.cancel.Cancel(CancelTimeout) })
173	178		defer timer.Stop()
174	179
175	180		// ---- Set up job (JobExtension.InitializeJob) ----

 	MatrixJSON   string
 	// Repo is the repository name used when the github context lacks one.
 	Repo string
-	// TimeoutMinutes is the job timeout; 0 means DefaultTimeoutMinutes.
+	// TimeoutMinutes is the job timeout; 0 means DefaultTimeoutMinutes. It is
+	// used for the message and, without Deadline, for the timer.
 	TimeoutMinutes int
+	// Deadline is the time left of the job timeout when Run is called, when the
+	// timeout started counting earlier (at the receipt of the job by the host);
+	// 0 means TimeoutMinutes from now.
+	Deadline time.Duration
 	Runner        RunnerInfo
 	RunnerVersion string
 }
 func (j *job) run() Result {
-	minutes := j.cfg.TimeoutMinutes
+	limit := j.cfg.Deadline
-	if minutes <= 0 {
+	if limit <= 0 {
-		minutes = DefaultTimeoutMinutes
+		limit = time.Duration(j.timeoutMinutes()) * time.Minute
 	}
-	timer := time.AfterFunc(time.Duration(minutes)*time.Minute, func() { j.cancel.Cancel(CancelTimeout) })
+	timer := time.AfterFunc(limit, func() { j.cancel.Cancel(CancelTimeout) })
 	defer timer.Stop()
 	// ---- Set up job (JobExtension.InitializeJob) ----

M internal/executor/job_test.go +17 -0


@@ -601,3 +601,20 @@ func TestCancelSignalsTheShellNotTheGroup(t *testing.T) {
601	601		t.Fatalf("the foreground child was signalled with the shell: %v", err)
602	602		}
603	603		}
	604	+
	605	+	// Deadline is what is left of the timeout when the host sent the job: the
	606	+	// executor must not grant the whole timeout again.
	607	+	func TestDeadlineOverridesTheTimeout(t *testing.T) {
	608	+	needTools(t, "bash")
	609	+	start := time.Now()
	610	+	rec, res := runLocal(t, wfHead+" steps:\n - run: sleep 30\n", "j", func(c *Config) {
	611	+	c.TimeoutMinutes = 60
	612	+	c.Deadline = 500 * time.Millisecond
	613	+	})
	614	+	if res.Conclusion != ConclusionTimedOut \|\| !strings.Contains(res.Error, "60 minutes") {
	615	+	t.Fatalf("%+v", res)
	616	+	}
	617	+	if time.Since(start) > 10*time.Second \|\| rec.steps[1].Conclusion != ConclusionCancelled {
	618	+	t.Fatalf("took %v: %+v", time.Since(start), rec.steps[1])
	619	+	}
	620	+	}


@@ -601,3 +601,20 @@ func TestCancelSignalsTheShellNotTheGroup(t *testing.T) {
601	t.Fatalf("the foreground child was signalled with the shell: %v", err)	601		t.Fatalf("the foreground child was signalled with the shell: %v", err)
602	}	602		}
603	}	603		}
		604	+
		605	+	// Deadline is what is left of the timeout when the host sent the job: the
		606	+	// executor must not grant the whole timeout again.
		607	+	func TestDeadlineOverridesTheTimeout(t *testing.T) {
		608	+	needTools(t, "bash")
		609	+	start := time.Now()
		610	+	rec, res := runLocal(t, wfHead+" steps:\n - run: sleep 30\n", "j", func(c *Config) {
		611	+	c.TimeoutMinutes = 60
		612	+	c.Deadline = 500 * time.Millisecond
		613	+	})
		614	+	if res.Conclusion != ConclusionTimedOut \|\| !strings.Contains(res.Error, "60 minutes") {
		615	+	t.Fatalf("%+v", res)
		616	+	}
		617	+	if time.Since(start) > 10*time.Second \|\| rec.steps[1].Conclusion != ConclusionCancelled {
		618	+	t.Fatalf("took %v: %+v", time.Since(start), rec.steps[1])
		619	+	}
		620	+	}

M internal/guest/jobexec.go +4 -4


@@ -179,10 +179,10 @@ func runJob(ctx context.Context, conn execstream.Conn, cfg ExecConfig, spec ex
179	179		res := executor.Run(canceller, executor.Config{
180	180		WorkflowPath: spec.GetWorkflowPath(), WorkflowYAML: []byte(spec.GetWorkflowYaml()), JobKey: spec.GetJobKey(),
181	181		ContextsJSON: spec.GetContexts(), MatrixJSON: spec.GetMatrix(), Repo: spec.GetRepo(),
182		-	TimeoutMinutes: int(spec.GetTimeoutMinutes()),
183		-	Runner: executor.RunnerInfo{Name: spec.GetRunnerName(), Environment: env},
184		-	RunnerVersion: spec.GetRunnerVersion(),
185		-	Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,
	182	+	TimeoutMinutes: int(spec.GetTimeoutMinutes()), Deadline: time.Duration(spec.GetDeadlineMs()) * time.Millisecond,
	183	+	Runner: executor.RunnerInfo{Name: spec.GetRunnerName(), Environment: env},
	184	+	RunnerVersion: spec.GetRunnerVersion(),
	185	+	Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,
186	186		Debug: spec.GetDebug(), Reporter: rep,
187	187		TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay,
188	188		})


@@ -179,10 +179,10 @@ func runJob(ctx context.Context, conn execstream.Conn, cfg ExecConfig, spec ex
179		res := executor.Run(canceller, executor.Config{	179		res := executor.Run(canceller, executor.Config{
180		WorkflowPath: spec.GetWorkflowPath(), WorkflowYAML: []byte(spec.GetWorkflowYaml()), JobKey: spec.GetJobKey(),	180		WorkflowPath: spec.GetWorkflowPath(), WorkflowYAML: []byte(spec.GetWorkflowYaml()), JobKey: spec.GetJobKey(),
181		ContextsJSON: spec.GetContexts(), MatrixJSON: spec.GetMatrix(), Repo: spec.GetRepo(),	181		ContextsJSON: spec.GetContexts(), MatrixJSON: spec.GetMatrix(), Repo: spec.GetRepo(),
182	-	TimeoutMinutes: int(spec.GetTimeoutMinutes()),	182	+	TimeoutMinutes: int(spec.GetTimeoutMinutes()), Deadline: time.Duration(spec.GetDeadlineMs()) * time.Millisecond,
183	-	Runner: executor.RunnerInfo{Name: spec.GetRunnerName(), Environment: env},	183	+	Runner: executor.RunnerInfo{Name: spec.GetRunnerName(), Environment: env},
184	-	RunnerVersion: spec.GetRunnerVersion(),	184	+	RunnerVersion: spec.GetRunnerVersion(),
185	-	Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,	185	+	Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,
186		Debug: spec.GetDebug(), Reporter: rep,	186		Debug: spec.GetDebug(), Reporter: rep,
187		TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay,	187		TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay,
188		})	188		})

M internal/host/job.go +62 -9


@@ -47,10 +47,15 @@ type jobRun struct {
47	47		ctx context.Context
48	48		cancel context.CancelCauseFunc
49	49
50		-	cancelCh chan *execpb.Cancel // the first value is sent to the executor
51		-	cancelMu sync.Mutex
52		-	cancelAt string // reason of the first cancel request, "" if none
53		-	stopping bool // the runner is shutting down
	50	+	// receivedAt is when the host got the job: the job timeout counts from it
	51	+	// (runner.proto: Job.timeout_minutes).
	52	+	receivedAt time.Time
	53	+
	54	+	cancelCh chan *execpb.Cancel // the first value is sent to the executor
	55	+	cancelMu sync.Mutex
	56	+	cancelAt string // reason of the first cancel request, "" if none
	57	+	cancelKind execpb.CancelKind // kind of the first cancel request
	58	+	stopping bool // the runner is shutting down
54	59
55	60		mu sync.Mutex
56	61		steps map[int32]*ermite.StepUpdate

@@ -84,9 +89,17 @@ func (j *jobRun) requestCancel(reason string, kind execpb.CancelKind) {
84	89		return
85	90		}
86	91		j.cancelAt = reason
	92	+	j.cancelKind = kind
87	93		j.cancelCh <- &execpb.Cancel{Reason: reason, Kind: kind}
88	94		}
89	95
	96	+	// firstCancelKind is the kind of the first cancel request.
	97	+	func (j *jobRun) firstCancelKind() execpb.CancelKind {
	98	+	j.cancelMu.Lock()
	99	+	defer j.cancelMu.Unlock()
	100	+	return j.cancelKind
	101	+	}
	102	+
90	103		func (j *jobRun) cancelRequested() bool {
91	104		j.cancelMu.Lock()
92	105		defer j.cancelMu.Unlock()

@@ -304,8 +317,14 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
304	317		j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)
305	318		j.outbox.start(ctx)
306	319
307		-	// A shutdown is a cancellation of kind shutdown, then a hard stop.
	320	+	// Two reasons to stop the job from the host, each a cancellation followed
	321	+	// by a hard stop of the sandbox if the executor does not answer: a shutdown
	322	+	// of the runner (ShutdownGrace), and the job timeout, which the executor
	323	+	// enforces itself but a stuck executor would not (TimeoutGrace after the
	324	+	// deadline: the escalation SIGINT, SIGTERM, SIGKILL takes 10 s).
308	325		stopHard := make(chan struct{})
	326	+	var hardOnce sync.Once
	327	+	hard := func() { hardOnce.Do(func() { close(stopHard) }) }
309	328		go func() {
310	329		select {
311	330		case <-shutdown:

@@ -318,7 +337,21 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
318	337		j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)
319	338		select {
320	339		case <-time.After(j.t.ShutdownGrace):
321		-	close(stopHard)
	340	+	hard()
	341	+	case <-ctx.Done():
	342	+	}
	343	+	}()
	344	+	go func() {
	345	+	select {
	346	+	case <-time.After(time.Until(j.deadline())):
	347	+	case <-ctx.Done():
	348	+	return
	349	+	}
	350	+	j.logf("the job timeout of %d minutes has passed", j.timeoutMinutes())
	351	+	j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)
	352	+	select {
	353	+	case <-time.After(j.t.TimeoutGrace):
	354	+	hard()
322	355		case <-ctx.Done():
323	356		}
324	357		}()

@@ -403,7 +436,8 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
403	436
404	437		run := &execpb.Run{Job: &execpb.JobSpec{
405	438		WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,
406		-	Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: j.job.TimeoutMinutes,
	439	+	Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),
	440	+	DeadlineMs: max(1, time.Until(j.deadline()).Milliseconds()),
407	441		RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: "self-hosted",
408	442		}}
409	443		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{

@@ -428,8 +462,13 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
428	462		}
429	463		return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED
430	464		case j.cancelRequested() \|\| ctx.Err() != nil:
431		-	// The executor did not report: it was stopped.
432		-	if j.isStopping() {
	465	+	// The executor did not report: it was stopped. A timeout wins over a
	466	+	// shutdown or a user cancel that came later, as on the server (the first
	467	+	// reason decides, and the timeout is the first when it fired first).
	468	+	switch {
	469	+	case j.firstCancelKind() == execpb.CancelKind_CANCEL_KIND_TIMEOUT:
	470	+	return ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT, nil, false
	471	+	case j.isStopping():
433	472		return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, true
434	473		}
435	474		return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, false

@@ -444,6 +483,20 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
444	483		return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
445	484		}
446	485
	486	+	// timeoutMinutes is the job timeout: at least 1, 360 when the job has none
	487	+	// (runner.proto: Job.timeout_minutes).
	488	+	func (j *jobRun) timeoutMinutes() int {
	489	+	if m := int(j.job.TimeoutMinutes); m > 0 {
	490	+	return m
	491	+	}
	492	+	return 360
	493	+	}
	494	+
	495	+	// deadline is when the job times out, counted from its receipt.
	496	+	func (j *jobRun) deadline() time.Time {
	497	+	return j.receivedAt.Add(time.Duration(j.timeoutMinutes()) * time.Minute)
	498	+	}
	499	+
447	500		func (j *jobRun) isStopping() bool {
448	501		j.cancelMu.Lock()
449	502		defer j.cancelMu.Unlock()


@@ -47,10 +47,15 @@ type jobRun struct {
47		ctx context.Context	47		ctx context.Context
48		cancel context.CancelCauseFunc	48		cancel context.CancelCauseFunc
49			49
50	-	cancelCh chan *execpb.Cancel // the first value is sent to the executor	50	+	// receivedAt is when the host got the job: the job timeout counts from it
51	-	cancelMu sync.Mutex	51	+	// (runner.proto: Job.timeout_minutes).
52	-	cancelAt string // reason of the first cancel request, "" if none	52	+	receivedAt time.Time
53	-	stopping bool // the runner is shutting down	53	+
			54	+	cancelCh chan *execpb.Cancel // the first value is sent to the executor
			55	+	cancelMu sync.Mutex
			56	+	cancelAt string // reason of the first cancel request, "" if none
			57	+	cancelKind execpb.CancelKind // kind of the first cancel request
			58	+	stopping bool // the runner is shutting down
54			59
55		mu sync.Mutex	60		mu sync.Mutex
56		steps map[int32]*ermite.StepUpdate	61		steps map[int32]*ermite.StepUpdate

@@ -84,9 +89,17 @@ func (j *jobRun) requestCancel(reason string, kind execpb.CancelKind) {
84		return	89		return
85		}	90		}
86		j.cancelAt = reason	91		j.cancelAt = reason
			92	+	j.cancelKind = kind
87		j.cancelCh <- &execpb.Cancel{Reason: reason, Kind: kind}	93		j.cancelCh <- &execpb.Cancel{Reason: reason, Kind: kind}
88		}	94		}
89			95
			96	+	// firstCancelKind is the kind of the first cancel request.
			97	+	func (j *jobRun) firstCancelKind() execpb.CancelKind {
			98	+	j.cancelMu.Lock()
			99	+	defer j.cancelMu.Unlock()
			100	+	return j.cancelKind
			101	+	}
			102	+
90		func (j *jobRun) cancelRequested() bool {	103		func (j *jobRun) cancelRequested() bool {
91		j.cancelMu.Lock()	104		j.cancelMu.Lock()
92		defer j.cancelMu.Unlock()	105		defer j.cancelMu.Unlock()

@@ -304,8 +317,14 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
304		j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)	317		j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)
305		j.outbox.start(ctx)	318		j.outbox.start(ctx)
306			319
307	-	// A shutdown is a cancellation of kind shutdown, then a hard stop.	320	+	// Two reasons to stop the job from the host, each a cancellation followed
			321	+	// by a hard stop of the sandbox if the executor does not answer: a shutdown
			322	+	// of the runner (ShutdownGrace), and the job timeout, which the executor
			323	+	// enforces itself but a stuck executor would not (TimeoutGrace after the
			324	+	// deadline: the escalation SIGINT, SIGTERM, SIGKILL takes 10 s).
308		stopHard := make(chan struct{})	325		stopHard := make(chan struct{})
			326	+	var hardOnce sync.Once
			327	+	hard := func() { hardOnce.Do(func() { close(stopHard) }) }
309		go func() {	328		go func() {
310		select {	329		select {
311		case <-shutdown:	330		case <-shutdown:

@@ -318,7 +337,21 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
318		j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)	337		j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)
319		select {	338		select {
320		case <-time.After(j.t.ShutdownGrace):	339		case <-time.After(j.t.ShutdownGrace):
321	-	close(stopHard)	340	+	hard()
			341	+	case <-ctx.Done():
			342	+	}
			343	+	}()
			344	+	go func() {
			345	+	select {
			346	+	case <-time.After(time.Until(j.deadline())):
			347	+	case <-ctx.Done():
			348	+	return
			349	+	}
			350	+	j.logf("the job timeout of %d minutes has passed", j.timeoutMinutes())
			351	+	j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)
			352	+	select {
			353	+	case <-time.After(j.t.TimeoutGrace):
			354	+	hard()
322		case <-ctx.Done():	355		case <-ctx.Done():
323		}	356		}
324		}()	357		}()

@@ -403,7 +436,8 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
403			436
404		run := &execpb.Run{Job: &execpb.JobSpec{	437		run := &execpb.Run{Job: &execpb.JobSpec{
405		WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,	438		WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,
406	-	Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: j.job.TimeoutMinutes,	439	+	Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),
			440	+	DeadlineMs: max(1, time.Until(j.deadline()).Milliseconds()),
407		RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: "self-hosted",	441		RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: "self-hosted",
408		}}	442		}}
409		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{	443		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{

@@ -428,8 +462,13 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
428		}	462		}
429		return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED	463		return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED
430		case j.cancelRequested() \|\| ctx.Err() != nil:	464		case j.cancelRequested() \|\| ctx.Err() != nil:
431	-	// The executor did not report: it was stopped.	465	+	// The executor did not report: it was stopped. A timeout wins over a
432	-	if j.isStopping() {	466	+	// shutdown or a user cancel that came later, as on the server (the first
			467	+	// reason decides, and the timeout is the first when it fired first).
			468	+	switch {
			469	+	case j.firstCancelKind() == execpb.CancelKind_CANCEL_KIND_TIMEOUT:
			470	+	return ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT, nil, false
			471	+	case j.isStopping():
433		return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, true	472		return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, true
434		}	473		}
435		return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, false	474		return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, false

@@ -444,6 +483,20 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
444		return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false	483		return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
445		}	484		}
446			485
			486	+	// timeoutMinutes is the job timeout: at least 1, 360 when the job has none
			487	+	// (runner.proto: Job.timeout_minutes).
			488	+	func (j *jobRun) timeoutMinutes() int {
			489	+	if m := int(j.job.TimeoutMinutes); m > 0 {
			490	+	return m
			491	+	}
			492	+	return 360
			493	+	}
			494	+
			495	+	// deadline is when the job times out, counted from its receipt.
			496	+	func (j *jobRun) deadline() time.Time {
			497	+	return j.receivedAt.Add(time.Duration(j.timeoutMinutes()) * time.Minute)
			498	+	}
			499	+
447		func (j *jobRun) isStopping() bool {	500		func (j *jobRun) isStopping() bool {
448		j.cancelMu.Lock()	501		j.cancelMu.Lock()
449		defer j.cancelMu.Unlock()	502		defer j.cancelMu.Unlock()

M internal/host/rpc.go +5 -0


@@ -111,6 +111,10 @@ type Timing struct {
111	111		LogFlush time.Duration
112	112		// WatchRetry is the delay before WatchJob is reopened.
113	113		WatchRetry time.Duration
	114	+	// TimeoutGrace is how long the executor gets after the job timeout to stop
	115	+	// its steps (SIGINT, then SIGTERM after 7.5 s, then SIGKILL 2.5 s later) and
	116	+	// report, before the sandbox is destroyed.
	117	+	TimeoutGrace time.Duration
114	118		// ShutdownGrace is how long jobs get to stop after a shutdown signal before
115	119		// their sandbox is destroyed.
116	120		ShutdownGrace time.Duration

@@ -138,6 +142,7 @@ func (t Timing) withDefaults() Timing {
138	142		def(&t.BackoffMax, 30*time.Second)
139	143		def(&t.LogFlush, 250*time.Millisecond)
140	144		def(&t.WatchRetry, time.Second)
	145	+	def(&t.TimeoutGrace, 30*time.Second)
141	146		def(&t.ShutdownGrace, 30*time.Second)
142	147		def(&t.CompleteDeadline, 5*time.Minute)
143	148		def(&t.NoJobPause, 0)


@@ -111,6 +111,10 @@ type Timing struct {
111	LogFlush time.Duration	111		LogFlush time.Duration
112	// WatchRetry is the delay before WatchJob is reopened.	112		// WatchRetry is the delay before WatchJob is reopened.
113	WatchRetry time.Duration	113		WatchRetry time.Duration
		114	+	// TimeoutGrace is how long the executor gets after the job timeout to stop
		115	+	// its steps (SIGINT, then SIGTERM after 7.5 s, then SIGKILL 2.5 s later) and
		116	+	// report, before the sandbox is destroyed.
		117	+	TimeoutGrace time.Duration
114	// ShutdownGrace is how long jobs get to stop after a shutdown signal before	118		// ShutdownGrace is how long jobs get to stop after a shutdown signal before
115	// their sandbox is destroyed.	119		// their sandbox is destroyed.
116	ShutdownGrace time.Duration	120		ShutdownGrace time.Duration

@@ -138,6 +142,7 @@ func (t Timing) withDefaults() Timing {
138	def(&t.BackoffMax, 30*time.Second)	142		def(&t.BackoffMax, 30*time.Second)
139	def(&t.LogFlush, 250*time.Millisecond)	143		def(&t.LogFlush, 250*time.Millisecond)
140	def(&t.WatchRetry, time.Second)	144		def(&t.WatchRetry, time.Second)
		145	+	def(&t.TimeoutGrace, 30*time.Second)
141	def(&t.ShutdownGrace, 30*time.Second)	146		def(&t.ShutdownGrace, 30*time.Second)
142	def(&t.CompleteDeadline, 5*time.Minute)	147		def(&t.CompleteDeadline, 5*time.Minute)
143	def(&t.NoJobPause, 0)	148		def(&t.NoJobPause, 0)

M internal/host/runner.go +12 -2


@@ -35,10 +35,20 @@ type Runner struct {
35	35		// JobSpec sizes the sandbox of a job; the zero value takes the defaults.
36	36		JobSpec func(*ermite.Job) sandbox.JobSpec
37	37
	38	+	// now returns the current time; tests replace it.
	39	+	now func() time.Time
	40	+
38	41		leaked atomic.Bool
39	42		active sync.WaitGroup
40	43		}
41	44
	45	+	func (r *Runner) clock() time.Time {
	46	+	if r.now != nil {
	47	+	return r.now()
	48	+	}
	49	+	return time.Now()
	50	+	}
	51	+
42	52		func (r *Runner) logf(format string, a ...any) {
43	53		if r.Logf != nil {
44	54		r.Logf(format, a...)

@@ -175,8 +185,8 @@ func (r Runner) startJob(ctx context.Context, job ermite.Job, t Timing, shutdo
175	185		jctx, cancel := context.WithCancelCause(context.WithoutCancel(ctx))
176	186		jr := &jobRun{
177	187		r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
178		-	logf: func(f string, a ...any) { r.logf("job "+job.JobId+": "+f, a...) },
179		-	cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},
	188	+	logf: func(f string, a ...any) { r.logf("job "+job.JobId+": "+f, a...) },
	189	+	receivedAt: r.clock(), cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},
180	190		}
181	191		r.logf("job %s (run %d, attempt %d, %s) assigned", job.JobId, job.RunId, job.Attempt, job.JobKey)
182	192		r.active.Add(1)

 	// JobSpec sizes the sandbox of a job; the zero value takes the defaults.
 	JobSpec func(*ermite.Job) sandbox.JobSpec
+	// now returns the current time; tests replace it.
+	now func() time.Time
 	leaked atomic.Bool
 	active sync.WaitGroup
 }
+func (r *Runner) clock() time.Time {
+	if r.now != nil {
+		return r.now()
+	}
+	return time.Now()
+}
 func (r *Runner) logf(format string, a ...any) {
 	if r.Logf != nil {
 		r.Logf(format, a...)
 	jctx, cancel := context.WithCancelCause(context.WithoutCancel(ctx))
 	jr := &jobRun{
 		r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
-		logf:     func(f string, a ...any) { r.logf("job "+job.JobId+": "+f, a...) },
+		logf:       func(f string, a ...any) { r.logf("job "+job.JobId+": "+f, a...) },
-		cancelCh: make(chan *execpb.Cancel, 1), steps: map[int32]*ermite.StepUpdate{},
+		receivedAt: r.clock(), cancelCh: make(chan *execpb.Cancel, 1), steps: map[int32]*ermite.StepUpdate{},
 	}
 	r.logf("job %s (run %d, attempt %d, %s) assigned", job.JobId, job.RunId, job.Attempt, job.JobKey)
 	r.active.Add(1)

M internal/host/runner_test.go +57 -0


@@ -790,3 +790,60 @@ func TestLoadConfig(t *testing.T) {
790	790		}
791	791
792	792		var _ = ermiteconnect.RunnerServiceClient(nil)
	793	+
	794	+	// The job timeout counts from the receipt of the job, and the host enforces it
	795	+	// even when the executor does not: the sandbox is stopped and the job ends
	796	+	// timed_out.
	797	+	func TestHostEnforcesTheJobTimeout(t *testing.T) {
	798	+	h := newHarness(t, func(_ testing.T, c execstream.Conn, _ execpb.Run, cancels <-chan execpb.Cancel) {
	799	+	stepStart(c, 1, "Set up job")
	800	+	logLineFrame(c, 1, "stuck", false)
	801	+	cn := <-cancels
	802	+	if cn.Kind != execpb.CancelKind_CANCEL_KIND_TIMEOUT {
	803	+	t.Errorf("kind %v", cn.Kind)
	804	+	}
	805	+	select {} // the executor ignores the cancel
	806	+	})
	807	+	h.runner.Timing.TimeoutGrace = 200 * time.Millisecond
	808	+	job := testJob("j1")
	809	+	job.TimeoutMinutes = 1
	810	+	h.runner.Job = func(tok string) ermiteconnect.RunnerServiceClient {
	811	+	return NewClient(h.url, tok, http.DefaultTransport)
	812	+	}
	813	+	h.runner.now = func() time.Time { return time.Now().Add(-59time.Second - 700time.Millisecond) } // received 59.7 s ago
	814	+	h.srv.jobs = []*ermite.Job{job}
	815	+	start := time.Now()
	816	+	h.run(t)
	817	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	818	+	if d := time.Since(start); d > 5*time.Second {
	819	+	t.Fatalf("took %v", d)
	820	+	}
	821	+	h.srv.mu.Lock()
	822	+	defer h.srv.mu.Unlock()
	823	+	c := h.srv.completes[0]
	824	+	if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT \|\| c.RunnerShutdown {
	825	+	t.Fatalf("%v", c.Conclusion)
	826	+	}
	827	+	for _, u := range c.Steps {
	828	+	if u.State != ermite.StepState_STEP_STATE_COMPLETED {
	829	+	t.Errorf("step %d not completed", u.Number)
	830	+	}
	831	+	}
	832	+	}
	833	+
	834	+	// The remaining time, not the whole timeout, goes to the executor.
	835	+	func TestRunCarriesTheRemainingDeadline(t *testing.T) {
	836	+	var spec *execpb.JobSpec
	837	+	inner := simpleScript("x")
	838	+	h := newHarness(t, func(t testing.T, c execstream.Conn, run execpb.Run, cn <-chan execpb.Cancel) {
	839	+	spec = run.Job
	840	+	inner(t, c, run, cn)
	841	+	})
	842	+	h.runner.now = func() time.Time { return time.Now().Add(-3 * time.Minute) }
	843	+	h.srv.jobs = []*ermite.Job{testJob("j1")} // 5 minutes
	844	+	h.run(t)
	845	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	846	+	if spec == nil \|\| spec.TimeoutMinutes != 5 \|\| spec.DeadlineMs < 110_000 \|\| spec.DeadlineMs > 120_000 {
	847	+	t.Fatalf("%v", spec)
	848	+	}
	849	+	}


@@ -790,3 +790,60 @@ func TestLoadConfig(t *testing.T) {
790	}	790		}
791		791
792	var _ = ermiteconnect.RunnerServiceClient(nil)	792		var _ = ermiteconnect.RunnerServiceClient(nil)
		793	+
		794	+	// The job timeout counts from the receipt of the job, and the host enforces it
		795	+	// even when the executor does not: the sandbox is stopped and the job ends
		796	+	// timed_out.
		797	+	func TestHostEnforcesTheJobTimeout(t *testing.T) {
		798	+	h := newHarness(t, func(_ testing.T, c execstream.Conn, _ execpb.Run, cancels <-chan execpb.Cancel) {
		799	+	stepStart(c, 1, "Set up job")
		800	+	logLineFrame(c, 1, "stuck", false)
		801	+	cn := <-cancels
		802	+	if cn.Kind != execpb.CancelKind_CANCEL_KIND_TIMEOUT {
		803	+	t.Errorf("kind %v", cn.Kind)
		804	+	}
		805	+	select {} // the executor ignores the cancel
		806	+	})
		807	+	h.runner.Timing.TimeoutGrace = 200 * time.Millisecond
		808	+	job := testJob("j1")
		809	+	job.TimeoutMinutes = 1
		810	+	h.runner.Job = func(tok string) ermiteconnect.RunnerServiceClient {
		811	+	return NewClient(h.url, tok, http.DefaultTransport)
		812	+	}
		813	+	h.runner.now = func() time.Time { return time.Now().Add(-59time.Second - 700time.Millisecond) } // received 59.7 s ago
		814	+	h.srv.jobs = []*ermite.Job{job}
		815	+	start := time.Now()
		816	+	h.run(t)
		817	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
		818	+	if d := time.Since(start); d > 5*time.Second {
		819	+	t.Fatalf("took %v", d)
		820	+	}
		821	+	h.srv.mu.Lock()
		822	+	defer h.srv.mu.Unlock()
		823	+	c := h.srv.completes[0]
		824	+	if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT \|\| c.RunnerShutdown {
		825	+	t.Fatalf("%v", c.Conclusion)
		826	+	}
		827	+	for _, u := range c.Steps {
		828	+	if u.State != ermite.StepState_STEP_STATE_COMPLETED {
		829	+	t.Errorf("step %d not completed", u.Number)
		830	+	}
		831	+	}
		832	+	}
		833	+
		834	+	// The remaining time, not the whole timeout, goes to the executor.
		835	+	func TestRunCarriesTheRemainingDeadline(t *testing.T) {
		836	+	var spec *execpb.JobSpec
		837	+	inner := simpleScript("x")
		838	+	h := newHarness(t, func(t testing.T, c execstream.Conn, run execpb.Run, cn <-chan execpb.Cancel) {
		839	+	spec = run.Job
		840	+	inner(t, c, run, cn)
		841	+	})
		842	+	h.runner.now = func() time.Time { return time.Now().Add(-3 * time.Minute) }
		843	+	h.srv.jobs = []*ermite.Job{testJob("j1")} // 5 minutes
		844	+	h.run(t)
		845	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
		846	+	if spec == nil \|\| spec.TimeoutMinutes != 5 \|\| spec.DeadlineMs < 110_000 \|\| spec.DeadlineMs > 120_000 {
		847	+	t.Fatalf("%v", spec)
		848	+	}
		849	+	}