fix(host): bound the report by ShutdownReport once the runner's shutdown channel is closed, not CompleteDeadline (GO-004)

commit: ca9abad1b65a59b6d50f75bfff5e5d609220ae5a
author: brice · 9d ago · 2026-10-12 20:13 UTC
parent: abd0a01
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +108 -5

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


@@ -437,7 +437,7 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
437	437
438	438		stopHard := j.startStopTimers(ctx, shutdown)
439	439		out := j.execute(ctx, rt, stopHard)
440		-	j.report(ctx, out)
	440	+	j.report(ctx, out, shutdown)
441	441		}
442	442
443	443		// acknowledge sends the first heartbeat, before anything else. A shutdown ends

@@ -516,7 +516,7 @@ func (j *jobRun) startStopTimers(ctx context.Context, shutdown <-chan struct{})
516	516
517	517		// report ends the job towards the server: the log is flushed, then CompleteJob.
518	518		// Nothing is sent to a job that the server has ended.
519		-	func (j *jobRun) report(ctx context.Context, out outcome) {
	519	+	func (j *jobRun) report(ctx context.Context, out outcome, shutdown <-chan struct{}) {
520	520		if err := j.abandoned(); err != nil {
521	521		j.log.Warn().Err(err).Msg("the server ended the job: no result is reported")
522	522		return

@@ -524,10 +524,14 @@ func (j *jobRun) report(ctx context.Context, out outcome) {
524	524		// The report is made even when the runner is shutting down: it uses its own
525	525		// deadline.
526	526		budget := j.t.CompleteDeadline
527		-	if j.isStopping() {
528		-	// The service manager waits only so long (see Timing): the report of a job
529		-	// cut short by a shutdown gets what is left of the budget.
	527	+	select {
	528	+	case <-shutdown:
	529	+	// The service manager waits only so long (see Timing): once the runner's
	530	+	// shutdown channel is closed, whatever the job's outcome, the report gets
	531	+	// what is left of the budget. The channel is read here, not the flag set
	532	+	// by the stop timers, which may not have run yet.
530	533		budget = min(budget, j.t.ShutdownReport)
	534	+	default:
531	535		}
532	536		rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget)
533	537		defer cancel()

 	stopHard := j.startStopTimers(ctx, shutdown)
 	out := j.execute(ctx, rt, stopHard)
-	j.report(ctx, out)
+	j.report(ctx, out, shutdown)
 }
 // acknowledge sends the first heartbeat, before anything else. A shutdown ends
 // report ends the job towards the server: the log is flushed, then CompleteJob.
 // Nothing is sent to a job that the server has ended.
-func (j *jobRun) report(ctx context.Context, out outcome) {
+func (j *jobRun) report(ctx context.Context, out outcome, shutdown <-chan struct{}) {
 	if err := j.abandoned(); err != nil {
 		j.log.Warn().Err(err).Msg("the server ended the job: no result is reported")
 		return
 	// The report is made even when the runner is shutting down: it uses its own
 	// deadline.
 	budget := j.t.CompleteDeadline
-	if j.isStopping() {
+	select {
-		// The service manager waits only so long (see Timing): the report of a job
+	case <-shutdown:
-		// cut short by a shutdown gets what is left of the budget.
+		// The service manager waits only so long (see Timing): once the runner's
+		// shutdown channel is closed, whatever the job's outcome, the report gets
+		// what is left of the budget. The channel is read here, not the flag set
+		// by the stop timers, which may not have run yet.
 		budget = min(budget, j.t.ShutdownReport)
+	default:
 	}
 	rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget)
 	defer cancel()

A internal/host/shutdown_test.go +99 -0

@@ -0,0 +1,99 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"context"
	7	+	"sync"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"connectrpc.com/connect"
	12	+	"github.com/bnema/zerowrap"
	13	+
	14	+	ermite "bnema.dev/ermite-api/gen/ermite"
	15	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	16	+	)
	17	+
	18	+	// stubClient answers the calls of the final report. CompleteJob records the
	19	+	// deadline of its context, then waits for the context to end.
	20	+	type stubClient struct {
	21	+	mu sync.Mutex
	22	+	deadline time.Time
	23	+	hang bool // AppendLog waits for its context to end
	24	+	}
	25	+
	26	+	func (c stubClient) UpdateJob(context.Context, connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
	27	+	return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
	28	+	}
	29	+
	30	+	func (c stubClient) AppendLog(ctx context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
	31	+	c.mu.Lock()
	32	+	hang := c.hang
	33	+	c.mu.Unlock()
	34	+	if hang {
	35	+	<-ctx.Done()
	36	+	return nil, ctx.Err()
	37	+	}
	38	+	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: r.Msg.Offset + int64(len(r.Msg.Lines))}), nil
	39	+	}
	40	+
	41	+	func (c stubClient) Heartbeat(context.Context, connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) {
	42	+	return connect.NewResponse(&ermite.HeartbeatResponse{}), nil
	43	+	}
	44	+
	45	+	func (c stubClient) WatchJob(context.Context, connect.Request[ermite.WatchJobRequest]) (*connect.ServerStreamForClient[ermite.WatchJobResponse], error) {
	46	+	return nil, connect.NewError(connect.CodeUnimplemented, nil)
	47	+	}
	48	+
	49	+	func (c stubClient) CompleteJob(ctx context.Context, _ connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
	50	+	c.mu.Lock()
	51	+	c.deadline, _ = ctx.Deadline()
	52	+	c.mu.Unlock()
	53	+	return connect.NewResponse(&ermite.CompleteJobResponse{}), nil
	54	+	}
	55	+
	56	+	// The report of a job gets ShutdownReport once the runner's shutdown channel is
	57	+	// closed, even when the stop timers did not mark the job as stopping (GO-004),
	58	+	// and CompleteDeadline otherwise.
	59	+	func TestReportBudgetFollowsTheShutdownChannel(t *testing.T) {
	60	+	for _, tc := range []struct {
	61	+	name string
	62	+	shutdown bool
	63	+	want time.Duration
	64	+	}{
	65	+	{"running", false, time.Hour},
	66	+	{"shutdown", true, 30 * time.Second},
	67	+	} {
	68	+	t.Run(tc.name, func(t *testing.T) {
	69	+	tm := Timing{
	70	+	CallTimeout: 24 * time.Hour, CompleteDeadline: time.Hour, ShutdownReport: 30 * time.Second,
	71	+	BackoffMin: time.Millisecond, BackoffMax: time.Millisecond,
	72	+	}
	73	+	cl := &stubClient{}
	74	+	ctx, cancel := context.WithCancel(context.Background())
	75	+	defer cancel()
	76	+	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},
	77	+	cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logBuf{}})}
	78	+	j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)
	79	+	j.outbox.start(ctx)
	80	+	shutdown := make(chan struct{})
	81	+	if tc.shutdown {
	82	+	close(shutdown)
	83	+	}
	84	+	if j.isStopping() {
	85	+	t.Fatal("the job must not be marked as stopping: only the channel is closed")
	86	+	}
	87	+	start := time.Now()
	88	+	j.report(ctx, outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}, shutdown)
	89	+	cl.mu.Lock()
	90	+	defer cl.mu.Unlock()
	91	+	if cl.deadline.IsZero() {
	92	+	t.Fatal("CompleteJob was not called")
	93	+	}
	94	+	if got := cl.deadline.Sub(start); got > tc.want+time.Second \|\| got < tc.want-time.Second {
	95	+	t.Fatalf("budget %v, want about %v", got, tc.want)
	96	+	}
	97	+	})
	98	+	}
	99	+	}

@@ -0,0 +1,99 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"context"
	7	+	"sync"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"connectrpc.com/connect"
	12	+	"github.com/bnema/zerowrap"
	13	+
	14	+	ermite "bnema.dev/ermite-api/gen/ermite"
	15	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	16	+	)
	17	+
	18	+	// stubClient answers the calls of the final report. CompleteJob records the
	19	+	// deadline of its context, then waits for the context to end.
	20	+	type stubClient struct {
	21	+	mu sync.Mutex
	22	+	deadline time.Time
	23	+	hang bool // AppendLog waits for its context to end
	24	+	}
	25	+
	26	+	func (c stubClient) UpdateJob(context.Context, connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
	27	+	return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
	28	+	}
	29	+
	30	+	func (c stubClient) AppendLog(ctx context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
	31	+	c.mu.Lock()
	32	+	hang := c.hang
	33	+	c.mu.Unlock()
	34	+	if hang {
	35	+	<-ctx.Done()
	36	+	return nil, ctx.Err()
	37	+	}
	38	+	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: r.Msg.Offset + int64(len(r.Msg.Lines))}), nil
	39	+	}
	40	+
	41	+	func (c stubClient) Heartbeat(context.Context, connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) {
	42	+	return connect.NewResponse(&ermite.HeartbeatResponse{}), nil
	43	+	}
	44	+
	45	+	func (c stubClient) WatchJob(context.Context, connect.Request[ermite.WatchJobRequest]) (*connect.ServerStreamForClient[ermite.WatchJobResponse], error) {
	46	+	return nil, connect.NewError(connect.CodeUnimplemented, nil)
	47	+	}
	48	+
	49	+	func (c stubClient) CompleteJob(ctx context.Context, _ connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
	50	+	c.mu.Lock()
	51	+	c.deadline, _ = ctx.Deadline()
	52	+	c.mu.Unlock()
	53	+	return connect.NewResponse(&ermite.CompleteJobResponse{}), nil
	54	+	}
	55	+
	56	+	// The report of a job gets ShutdownReport once the runner's shutdown channel is
	57	+	// closed, even when the stop timers did not mark the job as stopping (GO-004),
	58	+	// and CompleteDeadline otherwise.
	59	+	func TestReportBudgetFollowsTheShutdownChannel(t *testing.T) {
	60	+	for _, tc := range []struct {
	61	+	name string
	62	+	shutdown bool
	63	+	want time.Duration
	64	+	}{
	65	+	{"running", false, time.Hour},
	66	+	{"shutdown", true, 30 * time.Second},
	67	+	} {
	68	+	t.Run(tc.name, func(t *testing.T) {
	69	+	tm := Timing{
	70	+	CallTimeout: 24 * time.Hour, CompleteDeadline: time.Hour, ShutdownReport: 30 * time.Second,
	71	+	BackoffMin: time.Millisecond, BackoffMax: time.Millisecond,
	72	+	}
	73	+	cl := &stubClient{}
	74	+	ctx, cancel := context.WithCancel(context.Background())
	75	+	defer cancel()
	76	+	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},
	77	+	cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logBuf{}})}
	78	+	j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)
	79	+	j.outbox.start(ctx)
	80	+	shutdown := make(chan struct{})
	81	+	if tc.shutdown {
	82	+	close(shutdown)
	83	+	}
	84	+	if j.isStopping() {
	85	+	t.Fatal("the job must not be marked as stopping: only the channel is closed")
	86	+	}
	87	+	start := time.Now()
	88	+	j.report(ctx, outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}, shutdown)
	89	+	cl.mu.Lock()
	90	+	defer cl.mu.Unlock()
	91	+	if cl.deadline.IsZero() {
	92	+	t.Fatal("CompleteJob was not called")
	93	+	}
	94	+	if got := cl.deadline.Sub(start); got > tc.want+time.Second \|\| got < tc.want-time.Second {
	95	+	t.Fatalf("budget %v, want about %v", got, tc.want)
	96	+	}
	97	+	})
	98	+	}
	99	+	}