fix(host): retry CompleteJob once as a failure when the server refuses the result (M4); structured logs with the job's fields instead of Logf (G1)

commit: 508a002ef0c55fb4598aff397803975d08b4cf45
author: brice · 9d ago · 2026-10-12 19:32 UTC
committer: brice · 9d ago · 2026-10-12 19:33 UTC
parent: af4d447
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

8 files · +288 -67

M host_cmd.go +0 -3


@@ -9,7 +9,6 @@ import (
9	9		"fmt"
10	10		"net/http"
11	11		"os"
12		-	"strings"
13	12
14	13		"github.com/bnema/zerowrap"
15	14

@@ -120,8 +119,6 @@ func runCmd(ctx context.Context) int {
120	119		Job: func(tok string) ermiteconnect.RunnerServiceClient {
121	120		return host.NewClient(url, tok, http.DefaultTransport)
122	121		},
123		-	Logf: func(format string, args ...any) { log.Info().Msg(strings.TrimSpace(fmt.Sprintf(format, args...))) },
124		-	JobSpec: nil,
125	122		}
126	123		r.Cfg.URL = url
127	124		if err := r.Run(ctx); err != nil {


@@ -9,7 +9,6 @@ import (
9		"fmt"	9	"fmt"
10		"net/http"	10	"net/http"
11		"os"	11	"os"
12	-	"strings"
13			12
14		"github.com/bnema/zerowrap"	13	"github.com/bnema/zerowrap"
15			14

@@ -120,8 +119,6 @@ func runCmd(ctx context.Context) int {
120		Job: func(tok string) ermiteconnect.RunnerServiceClient {	119	Job: func(tok string) ermiteconnect.RunnerServiceClient {
121		return host.NewClient(url, tok, http.DefaultTransport)	120	return host.NewClient(url, tok, http.DefaultTransport)
122		},	121	},
123	-	Logf: func(format string, args ...any) { log.Info().Msg(strings.TrimSpace(fmt.Sprintf(format, args...))) },
124	-	JobSpec: nil,
125		}	122	}
126		r.Cfg.URL = url	123	r.Cfg.URL = url
127		if err := r.Run(ctx); err != nil {	124	if err := r.Run(ctx); err != nil {

M internal/e2e/chain_test.go +10 -2


@@ -15,6 +15,8 @@ import (
15	15		"testing"
16	16		"time"
17	17
	18	+	"github.com/bnema/zerowrap"
	19	+
18	20		ermite "bnema.dev/ermite-api/gen/ermite"
19	21		"bnema.dev/ermite-api/gen/ermite/ermiteconnect"
20	22

@@ -164,9 +166,8 @@ func TestChainCorpus(t *testing.T) {
164	166		return host.NewClient(srv.URL, tok, http.DefaultTransport)
165	167		},
166	168		Timing: host.Timing{LogFlush: 20 * time.Millisecond, NoJobPause: 100 * time.Millisecond},
167		-	Logf: func(f string, a ...any) { t.Logf(f, a...) },
168	169		}
169		-	ctx, cancel := context.WithCancel(context.Background())
	170	+	ctx, cancel := context.WithCancel(zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "info", Format: "console", Output: testWriter{t}})))
170	171		done := make(chan error, 1)
171	172		go func() { done <- r.Run(ctx) }()
172	173		defer func() {

@@ -318,3 +319,10 @@ func checkProducerOutputs(t *testing.T, got hosttest.Result) {
318	319		}
319	320		}
320	321		}
	322	+
	323	+	type testWriter struct{ t *testing.T }
	324	+
	325	+	func (w testWriter) Write(p []byte) (int, error) {
	326	+	w.t.Log(strings.TrimRight(string(p), "\n"))
	327	+	return len(p), nil
	328	+	}

 	"testing"
 	"time"
+	"github.com/bnema/zerowrap"
 	ermite "bnema.dev/ermite-api/gen/ermite"
 	"bnema.dev/ermite-api/gen/ermite/ermiteconnect"
 			return host.NewClient(srv.URL, tok, http.DefaultTransport)
 		},
 		Timing: host.Timing{LogFlush: 20 * time.Millisecond, NoJobPause: 100 * time.Millisecond},
-		Logf:   func(f string, a ...any) { t.Logf(f, a...) },
 	}
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "info", Format: "console", Output: testWriter{t}})))
 	done := make(chan error, 1)
 	go func() { done <- r.Run(ctx) }()
 	defer func() {
 		}
 	}
 }
+type testWriter struct{ t *testing.T }
+func (w testWriter) Write(p []byte) (int, error) {
+	w.t.Log(strings.TrimRight(string(p), "\n"))
+	return len(p), nil
+}

M internal/host/fake_test.go +20 -6


@@ -14,6 +14,7 @@ import (
14	14		"time"
15	15
16	16		"connectrpc.com/connect"
	17	+	"github.com/bnema/zerowrap"
17	18
18	19		ermite "bnema.dev/ermite-api/gen/ermite"
19	20		"bnema.dev/ermite-api/gen/ermite/ermiteconnect"

@@ -301,7 +302,8 @@ type fakeDriver struct {
301	302		stopErr error
302	303		capacity int
303	304		live int
304		-	blockStart chan struct{} // when set, Start waits on it
	305	+	blockStart chan struct{} // when set, Start waits on it
	306	+	specs []sandbox.JobSpec // what each Start was asked for
305	307		iso sandbox.Isolation
306	308		}
307	309

@@ -333,8 +335,11 @@ type fakeSandbox struct {
333	335		}
334	336
335	337		func (s *fakeSandbox) ID() string { return "er00000000" }
336		-	func (s *fakeSandbox) Start(ctx context.Context, _ sandbox.JobSpec) (sandbox.Conn, error) {
	338	+	func (s *fakeSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
337	339		d := s.d
	340	+	d.mu.Lock()
	341	+	d.specs = append(d.specs, spec)
	342	+	d.mu.Unlock()
338	343		if d.blockStart != nil {
339	344		select {
340	345		case <-d.blockStart:

@@ -467,18 +472,27 @@ type harness struct {
467	472		logs *logBuf
468	473		}
469	474
	475	+	// logBuf collects what the runner logs (zerowrap, console format without
	476	+	// colors), so that tests can look at it.
470	477		type logBuf struct {
471	478		mu sync.Mutex
472	479		b strings.Builder
473	480		}
474	481
475		-	func (l *logBuf) logf(f string, a ...any) {
	482	+	func (l *logBuf) Write(p []byte) (int, error) {
476	483		l.mu.Lock()
477	484		defer l.mu.Unlock()
478		-	l.b.WriteString(strings.TrimSpace(sprintf(f, a...)) + "\n")
	485	+	return l.b.Write(p)
479	486		}
	487	+
480	488		func (l *logBuf) String() string { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() }
481	489
	490	+	// ctx returns a context with a logger that writes to the buffer (JSON, one
	491	+	// event per line, debug level).
	492	+	func (l *logBuf) ctx() context.Context {
	493	+	return zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: l}))
	494	+	}
	495	+
482	496		func newHarness(t testing.T, sc script) harness {
483	497		t.Helper()
484	498		srv := newFakeServer()

@@ -498,7 +512,7 @@ func newHarness(t testing.T, sc script) harness {
498	512		Driver: drv,
499	513		Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),
500	514		Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
501		-	Timing: fastTiming(), Logf: lb.logf,
	515	+	Timing: fastTiming(),
502	516		}
503	517		return h
504	518		}

@@ -506,7 +520,7 @@ func newHarness(t testing.T, sc script) harness {
506	520		// run starts the loop; stop cancels it and waits for Run to return.
507	521		func (h harness) run(t testing.T) (stop func() error) {
508	522		t.Helper()
509		-	ctx, cancel := context.WithCancel(context.Background())
	523	+	ctx, cancel := context.WithCancel(h.logs.ctx())
510	524		done := make(chan error, 1)
511	525		go func() { done <- h.runner.Run(ctx) }()
512	526		var once sync.Once

 	"time"
 	"connectrpc.com/connect"
+	"github.com/bnema/zerowrap"
 	ermite "bnema.dev/ermite-api/gen/ermite"
 	"bnema.dev/ermite-api/gen/ermite/ermiteconnect"
 	stopErr    error
 	capacity   int
 	live       int
-	blockStart chan struct{} // when set, Start waits on it
+	blockStart chan struct{}     // when set, Start waits on it
+	specs      []sandbox.JobSpec // what each Start was asked for
 	iso        sandbox.Isolation
 }
 }
 func (s *fakeSandbox) ID() string { return "er00000000" }
-func (s *fakeSandbox) Start(ctx context.Context, _ sandbox.JobSpec) (sandbox.Conn, error) {
+func (s *fakeSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
 	d := s.d
+	d.mu.Lock()
+	d.specs = append(d.specs, spec)
+	d.mu.Unlock()
 	if d.blockStart != nil {
 		select {
 		case <-d.blockStart:
 	logs   *logBuf
 }
+// logBuf collects what the runner logs (zerowrap, console format without
+// colors), so that tests can look at it.
 type logBuf struct {
 	mu sync.Mutex
 	b  strings.Builder
 }
-func (l *logBuf) logf(f string, a ...any) {
+func (l *logBuf) Write(p []byte) (int, error) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
-	l.b.WriteString(strings.TrimSpace(sprintf(f, a...)) + "\n")
+	return l.b.Write(p)
 }
 func (l *logBuf) String() string { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() }
+// ctx returns a context with a logger that writes to the buffer (JSON, one
+// event per line, debug level).
+func (l *logBuf) ctx() context.Context {
+	return zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: l}))
+}
 func newHarness(t *testing.T, sc script) *harness {
 	t.Helper()
 	srv := newFakeServer()
 		Driver: drv,
 		Poll:   NewClient(url, "ermiterun_tok", http.DefaultTransport),
 		Job:    func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
-		Timing: fastTiming(), Logf: lb.logf,
+		Timing: fastTiming(),
 	}
 	return h
 }
 // run starts the loop; stop cancels it and waits for Run to return.
 func (h *harness) run(t *testing.T) (stop func() error) {
 	t.Helper()
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(h.logs.ctx())
 	done := make(chan error, 1)
 	go func() { done <- h.runner.Run(ctx) }()
 	var once sync.Once

M internal/host/job.go +93 -20


@@ -13,6 +13,7 @@ import (
13	13		"unicode/utf8"
14	14
15	15		"connectrpc.com/connect"
	16	+	"github.com/bnema/zerowrap"
16	17		"google.golang.org/protobuf/proto"
17	18		"google.golang.org/protobuf/types/known/timestamppb"
18	19

@@ -41,7 +42,7 @@ type jobRun struct {
41	42		job *ermite.Job
42	43		client jobClient
43	44		t Timing
44		-	logf func(string, ...any)
	45	+	log zerowrap.Logger // with job_id, run_id and attempt
45	46
46	47		// ctx lives until the job is finished or abandoned. Shutdown does not end
47	48		// it: a shutdown is a cancellation that the job reports.

@@ -132,7 +133,7 @@ func (j *jobRun) ack(ctx context.Context) bool {
132	133		j.abandon(err)
133	134		return false
134	135		}
135		-	j.logf("first heartbeat failed, retrying: %v", err)
	136	+	j.log.Warn().Err(err).Msg("first heartbeat failed, retrying")
136	137		if !sleep(ctx, j.t.HeartbeatRetry) {
137	138		return false
138	139		}

@@ -163,7 +164,7 @@ func (j *jobRun) heartbeats(ctx context.Context) {
163	164		j.abandon(err)
164	165		return
165	166		}
166		-	j.logf("heartbeat failed, retrying in %s: %v", j.t.HeartbeatRetry, err)
	167	+	j.log.Warn().Err(err).Dur("retry_in", j.t.HeartbeatRetry).Msg("heartbeat failed")
167	168		wait = j.t.HeartbeatRetry
168	169		continue
169	170		}

@@ -200,7 +201,7 @@ func (j *jobRun) watch(ctx context.Context) {
200	201		j.abandon(err)
201	202		return
202	203		}
203		-	j.logf("WatchJob: %v", err)
	204	+	j.log.Warn().Err(err).Msg("WatchJob failed, reopening")
204	205		}
205	206		if !sleep(ctx, j.t.WatchRetry) {
206	207		return

@@ -246,7 +247,7 @@ func (j *jobRun) violate(format string, a ...any) {
246	247		j.violation = fmt.Sprintf(format, a...)
247	248		msg := j.violation
248	249		j.mu.Unlock()
249		-	j.logf("the executor broke the protocol: %s", msg)
	250	+	j.log.Error().Str("violation", msg).Msg("the executor broke the protocol")
250	251		j.sandboxError("The runner stopped the job: the executor broke the protocol: " + msg)
251	252		if j.hard != nil {
252	253		j.hard()

@@ -423,7 +424,7 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
423	424
424	425		// 1. The acknowledgement comes first, before anything else.
425	426		if !j.ack(ctx) {
426		-	j.logf("job %s: assignment lost before it started", j.job.JobId)
	427	+	j.log.Warn().Msg("the assignment was lost before the job started")
427	428		return
428	429		}
429	430		var bg sync.WaitGroup

@@ -433,7 +434,7 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
433	434		// The context is cancelled first, so that the heartbeat and watch loops end.
434	435		defer func() { j.cancel(nil); bg.Wait() }()
435	436
436		-	j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)
	437	+	j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.log)
437	438		j.outbox.start(ctx)
438	439
439	440		// Two reasons to stop the job from the host, each a cancellation followed

@@ -467,7 +468,7 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
467	468		case <-ctx.Done():
468	469		return
469	470		}
470		-	j.logf("the job timeout of %d minutes has passed", j.timeoutMinutes())
	471	+	j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed")
471	472		j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)
472	473		select {
473	474		case <-time.After(j.t.TimeoutGrace):

@@ -480,7 +481,7 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
480	481
481	482		// 2. Nothing more is sent to a job that is over.
482	483		if err := j.abandoned(); err != nil {
483		-	j.logf("job %s: not reporting a result: %v", j.job.JobId, err)
	484	+	j.log.Warn().Err(err).Msg("the server ended the job: no result is reported")
484	485		return
485	486		}
486	487		// The report is made even when the runner is shutting down: it uses its own

@@ -517,7 +518,7 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
517	518		sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute)
518	519		defer cancel()
519	520		if err := sb.Stop(sctx); err != nil {
520		-	j.logf("job %s: stopping the sandbox: %v", j.job.JobId, err)
	521	+	j.log.Error().Err(err).Msg("stopping the sandbox")
521	522		if errors.Is(err, sandbox.ErrLeaked) {
522	523		rt.markLeaked()
523	524		}

@@ -580,7 +581,7 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
580	581		c = ermite.CiConclusion_CI_CONCLUSION_FAILURE
581	582		}
582	583		if res.Error != "" {
583		-	j.logf("job %s: %s", j.job.JobId, res.Error)
	584	+	j.log.Info().Str("error", res.Error).Msg("the job reported an error")
584	585		}
585	586		return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED
586	587		case j.cancelRequested() \|\| ctx.Err() != nil:

@@ -653,6 +654,30 @@ func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, o
653	654		RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,
654	655		Steps: j.finalSteps(conclusion),
655	656		}
	657	+	err := j.sendComplete(ctx, req, conclusion)
	658	+	if err == nil \|\| ctx.Err() != nil \|\| classify(err) != kindPermanent {
	659	+	return
	660	+	}
	661	+	// The server refused the result itself (invalid_argument, or
	662	+	// failed_precondition that is not the end of the job). CompleteJob is
	663	+	// "applied whole or not at all", so the job is still open: say why in the
	664	+	// log, then report a plain failure without outputs and with steps that are
	665	+	// valid by construction. Without this the job would stay running until the
	666	+	// heartbeat deadline of the server and then be reported as a lost runner.
	667	+	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)))
	668	+	j.log.Error().Err(err).Msg("CompleteJob refused: retrying as a failure without outputs")
	669	+	req = &ermite.CompleteJobRequest{
	670	+	JobId: j.job.JobId, Conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE,
	671	+	Steps: j.rebuiltSteps(),
	672	+	}
	673	+	if err := j.sendComplete(ctx, req, req.Conclusion); err != nil && ctx.Err() == nil {
	674	+	j.log.Error().Err(err).Msg("CompleteJob refused again, giving up")
	675	+	}
	676	+	}
	677	+
	678	+	// sendComplete sends the request until the server answers: success, the end of
	679	+	// the job, a lost assignment, or a refusal (returned).
	680	+	func (j jobRun) sendComplete(ctx context.Context, req ermite.CompleteJobRequest, conclusion ermite.CiConclusion) error {
656	681		b := backoff{min: j.t.BackoffMin, max: j.t.BackoffMax}
657	682		for {
658	683		cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)

@@ -660,28 +685,76 @@ func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, o
660	685		cancel()
661	686		if err == nil {
662	687		if resp.Msg.Conclusion != conclusion {
663		-	j.logf("job %s: ended as %s, the server holds %s", j.job.JobId, conclusion, resp.Msg.Conclusion)
	688	+	j.log.Warn().Str("reported", conclusion.String()).Str("stored", resp.Msg.Conclusion.String()).Msg("the server holds another conclusion")
664	689		}
665		-	return
	690	+	return nil
666	691		}
667	692		if ctx.Err() != nil {
668		-	j.logf("job %s: could not report the result: %v", j.job.JobId, ctx.Err())
669		-	return
	693	+	j.log.Error().Err(ctx.Err()).Msg("could not report the result")
	694	+	return err
670	695		}
671	696		switch classify(err) {
672	697		case kindEnded, kindAbandoned:
673		-	return
	698	+	return nil
674	699		case kindPermanent:
675		-	j.logf("job %s: CompleteJob refused: %v", j.job.JobId, err)
676		-	return
	700	+	return err
677	701		}
678		-	j.logf("CompleteJob failed, retrying: %v", err)
	702	+	j.log.Warn().Err(err).Msg("CompleteJob failed, retrying")
679	703		if !sleep(ctx, b.next()) {
680		-	return
	704	+	return err
681	705		}
682	706		}
683	707		}
684	708
	709	+	// errorLog appends a line to the log after the outbox is closed.
	710	+	func (j *jobRun) errorLog(ctx context.Context, msg string) {
	711	+	j.mu.Lock()
	712	+	var last int32 = 1
	713	+	for n := range j.steps {
	714	+	last = max(last, n)
	715	+	}
	716	+	j.mu.Unlock()
	717	+	j.outbox.AppendNow(ctx, logLine{step: last, unixNano: time.Now().UnixNano(), text: "##[error]" + msg})
	718	+	}
	719	+
	720	+	// rebuiltSteps returns every known step as a valid, completed snapshot: number
	721	+	// in 1..maxSteps, a name of at most 1,024 bytes, a conclusion the server
	722	+	// accepts (success, failure, cancelled, skipped), both times set, and no step
	723	+	// that finished before it began. A step that was running is failed.
	724	+	func (j jobRun) rebuiltSteps() []ermite.StepUpdate {
	725	+	j.mu.Lock()
	726	+	defer j.mu.Unlock()
	727	+	now := timestamppb.Now()
	728	+	var out []*ermite.StepUpdate
	729	+	for n, u := range j.steps {
	730	+	if n < 1 \|\| n > maxSteps {
	731	+	continue
	732	+	}
	733	+	c := proto.Clone(u).(*ermite.StepUpdate)
	734	+	c.Name = cutName(c.Name)
	735	+	if c.StartedAt == nil {
	736	+	c.StartedAt = now
	737	+	}
	738	+	if c.State != ermite.StepState_STEP_STATE_COMPLETED {
	739	+	c.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
	740	+	c.CompletedAt = now
	741	+	}
	742	+	switch c.Conclusion {
	743	+	case ermite.CiConclusion_CI_CONCLUSION_SUCCESS, ermite.CiConclusion_CI_CONCLUSION_FAILURE,
	744	+	ermite.CiConclusion_CI_CONCLUSION_CANCELLED, ermite.CiConclusion_CI_CONCLUSION_SKIPPED:
	745	+	default:
	746	+	c.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
	747	+	}
	748	+	c.State = ermite.StepState_STEP_STATE_COMPLETED
	749	+	if c.CompletedAt == nil \|\| c.CompletedAt.AsTime().Before(c.StartedAt.AsTime()) {
	750	+	c.CompletedAt = c.StartedAt
	751	+	}
	752	+	out = append(out, c)
	753	+	}
	754	+	sort.Slice(out, func(a, b int) bool { return out[a].Number < out[b].Number })
	755	+	return out
	756	+	}
	757	+
685	758		// Limits of CompleteJobRequest.outputs.
686	759		const (
687	760		maxOutputBytes = 1 << 20

 	"unicode/utf8"
 	"connectrpc.com/connect"
+	"github.com/bnema/zerowrap"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	job    *ermite.Job
 	client jobClient
 	t      Timing
-	logf   func(string, ...any)
+	log    zerowrap.Logger // with job_id, run_id and attempt
 	// ctx lives until the job is finished or abandoned. Shutdown does not end
 	// it: a shutdown is a cancellation that the job reports.
 			j.abandon(err)
 			return false
 		}
-		j.logf("first heartbeat failed, retrying: %v", err)
+		j.log.Warn().Err(err).Msg("first heartbeat failed, retrying")
 		if !sleep(ctx, j.t.HeartbeatRetry) {
 			return false
 		}
 				j.abandon(err)
 				return
 			}
-			j.logf("heartbeat failed, retrying in %s: %v", j.t.HeartbeatRetry, err)
+			j.log.Warn().Err(err).Dur("retry_in", j.t.HeartbeatRetry).Msg("heartbeat failed")
 			wait = j.t.HeartbeatRetry
 			continue
 		}
 				j.abandon(err)
 				return
 			}
-			j.logf("WatchJob: %v", err)
+			j.log.Warn().Err(err).Msg("WatchJob failed, reopening")
 		}
 		if !sleep(ctx, j.t.WatchRetry) {
 			return
 	j.violation = fmt.Sprintf(format, a...)
 	msg := j.violation
 	j.mu.Unlock()
-	j.logf("the executor broke the protocol: %s", msg)
+	j.log.Error().Str("violation", msg).Msg("the executor broke the protocol")
 	j.sandboxError("The runner stopped the job: the executor broke the protocol: " + msg)
 	if j.hard != nil {
 		j.hard()
 	// 1. The acknowledgement comes first, before anything else.
 	if !j.ack(ctx) {
-		j.logf("job %s: assignment lost before it started", j.job.JobId)
+		j.log.Warn().Msg("the assignment was lost before the job started")
 		return
 	}
 	var bg sync.WaitGroup
 	// The context is cancelled first, so that the heartbeat and watch loops end.
 	defer func() { j.cancel(nil); bg.Wait() }()
-	j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)
+	j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.log)
 	j.outbox.start(ctx)
 	// Two reasons to stop the job from the host, each a cancellation followed
 		case <-ctx.Done():
 			return
 		}
-		j.logf("the job timeout of %d minutes has passed", j.timeoutMinutes())
+		j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed")
 		j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)
 		select {
 		case <-time.After(j.t.TimeoutGrace):
 	// 2. Nothing more is sent to a job that is over.
 	if err := j.abandoned(); err != nil {
-		j.logf("job %s: not reporting a result: %v", j.job.JobId, err)
+		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
 		sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute)
 		defer cancel()
 		if err := sb.Stop(sctx); err != nil {
-			j.logf("job %s: stopping the sandbox: %v", j.job.JobId, err)
+			j.log.Error().Err(err).Msg("stopping the sandbox")
 			if errors.Is(err, sandbox.ErrLeaked) {
 				rt.markLeaked()
 			}
 			c = ermite.CiConclusion_CI_CONCLUSION_FAILURE
 		}
 		if res.Error != "" {
-			j.logf("job %s: %s", j.job.JobId, res.Error)
+			j.log.Info().Str("error", res.Error).Msg("the job reported an error")
 		}
 		return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED
 	case j.cancelRequested() || ctx.Err() != nil:
 		RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,
 		Steps:          j.finalSteps(conclusion),
 	}
+	err := j.sendComplete(ctx, req, conclusion)
+	if err == nil || ctx.Err() != nil || classify(err) != kindPermanent {
+		return
+	}
+	// The server refused the result itself (invalid_argument, or
+	// failed_precondition that is not the end of the job). CompleteJob is
+	// "applied whole or not at all", so the job is still open: say why in the
+	// log, then report a plain failure without outputs and with steps that are
+	// valid by construction. Without this the job would stay running until the
+	// heartbeat deadline of the server and then be reported as a lost runner.
+	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)))
+	j.log.Error().Err(err).Msg("CompleteJob refused: retrying as a failure without outputs")
+	req = &ermite.CompleteJobRequest{
+		JobId: j.job.JobId, Conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE,
+		Steps: j.rebuiltSteps(),
+	}
+	if err := j.sendComplete(ctx, req, req.Conclusion); err != nil && ctx.Err() == nil {
+		j.log.Error().Err(err).Msg("CompleteJob refused again, giving up")
+	}
+}
+// sendComplete sends the request until the server answers: success, the end of
+// the job, a lost assignment, or a refusal (returned).
+func (j *jobRun) sendComplete(ctx context.Context, req *ermite.CompleteJobRequest, conclusion ermite.CiConclusion) error {
 	b := backoff{min: j.t.BackoffMin, max: j.t.BackoffMax}
 	for {
 		cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
 		cancel()
 		if err == nil {
 			if resp.Msg.Conclusion != conclusion {
-				j.logf("job %s: ended as %s, the server holds %s", j.job.JobId, conclusion, resp.Msg.Conclusion)
+				j.log.Warn().Str("reported", conclusion.String()).Str("stored", resp.Msg.Conclusion.String()).Msg("the server holds another conclusion")
 			}
-			return
+			return nil
 		}
 		if ctx.Err() != nil {
-			j.logf("job %s: could not report the result: %v", j.job.JobId, ctx.Err())
+			j.log.Error().Err(ctx.Err()).Msg("could not report the result")
-			return
+			return err
 		}
 		switch classify(err) {
 		case kindEnded, kindAbandoned:
-			return
+			return nil
 		case kindPermanent:
-			j.logf("job %s: CompleteJob refused: %v", j.job.JobId, err)
+			return err
-			return
 		}
-		j.logf("CompleteJob failed, retrying: %v", err)
+		j.log.Warn().Err(err).Msg("CompleteJob failed, retrying")
 		if !sleep(ctx, b.next()) {
-			return
+			return err
 		}
 	}
 }
+// errorLog appends a line to the log after the outbox is closed.
+func (j *jobRun) errorLog(ctx context.Context, msg string) {
+	j.mu.Lock()
+	var last int32 = 1
+	for n := range j.steps {
+		last = max(last, n)
+	}
+	j.mu.Unlock()
+	j.outbox.AppendNow(ctx, logLine{step: last, unixNano: time.Now().UnixNano(), text: "##[error]" + msg})
+}
+// rebuiltSteps returns every known step as a valid, completed snapshot: number
+// in 1..maxSteps, a name of at most 1,024 bytes, a conclusion the server
+// accepts (success, failure, cancelled, skipped), both times set, and no step
+// that finished before it began. A step that was running is failed.
+func (j *jobRun) rebuiltSteps() []*ermite.StepUpdate {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	now := timestamppb.Now()
+	var out []*ermite.StepUpdate
+	for n, u := range j.steps {
+		if n < 1 || n > maxSteps {
+			continue
+		}
+		c := proto.Clone(u).(*ermite.StepUpdate)
+		c.Name = cutName(c.Name)
+		if c.StartedAt == nil {
+			c.StartedAt = now
+		}
+		if c.State != ermite.StepState_STEP_STATE_COMPLETED {
+			c.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
+			c.CompletedAt = now
+		}
+		switch c.Conclusion {
+		case ermite.CiConclusion_CI_CONCLUSION_SUCCESS, ermite.CiConclusion_CI_CONCLUSION_FAILURE,
+			ermite.CiConclusion_CI_CONCLUSION_CANCELLED, ermite.CiConclusion_CI_CONCLUSION_SKIPPED:
+		default:
+			c.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
+		}
+		c.State = ermite.StepState_STEP_STATE_COMPLETED
+		if c.CompletedAt == nil || c.CompletedAt.AsTime().Before(c.StartedAt.AsTime()) {
+			c.CompletedAt = c.StartedAt
+		}
+		out = append(out, c)
+	}
+	sort.Slice(out, func(a, b int) bool { return out[a].Number < out[b].Number })
+	return out
+}
 // Limits of CompleteJobRequest.outputs.
 const (
 	maxOutputBytes = 1 << 20

M internal/host/outbox.go +15 -7


@@ -8,6 +8,7 @@ import (
8	8		"time"
9	9
10	10		"connectrpc.com/connect"
	11	+	"github.com/bnema/zerowrap"
11	12		"google.golang.org/protobuf/types/known/timestamppb"
12	13
13	14		ermite "bnema.dev/ermite-api/gen/ermite"

@@ -50,7 +51,7 @@ type outbox struct {
50	51		// onEnded is called once when the server says the job is over (JobEnded)
51	52		// or the assignment is gone; the outbox then drops everything.
52	53		onEnded func(error)
53		-	logf func(format string, args ...any)
	54	+	log zerowrap.Logger
54	55
55	56		mu sync.Mutex
56	57		cond *sync.Cond

@@ -69,8 +70,8 @@ type callClient interface {
69	70		AppendLog(context.Context, connect.Request[ermite.AppendLogRequest]) (connect.Response[ermite.AppendLogResponse], error)
70	71		}
71	72
72		-	func newOutbox(c callClient, jobID string, t Timing, onEnded func(error), logf func(string, ...any)) *outbox {
73		-	o := &outbox{client: c, jobID: jobID, t: t, onEnded: onEnded, logf: logf, done: make(chan struct{})}
	73	+	func newOutbox(c callClient, jobID string, t Timing, onEnded func(error), log zerowrap.Logger) *outbox {
	74	+	o := &outbox{client: c, jobID: jobID, t: t, onEnded: onEnded, log: log, done: make(chan struct{})}
74	75		o.cond = sync.NewCond(&o.mu)
75	76		return o
76	77		}

@@ -239,7 +240,7 @@ func (o *outbox) retry(ctx context.Context, what string, call func(context.Conte
239	240		case kindPermanent:
240	241		return err
241	242		}
242		-	o.logf("%s failed, retrying: %v", what, err)
	243	+	o.log.Warn().Err(err).Str("call", what).Msg("call failed, retrying")
243	244		if !sleep(ctx, b.next()) {
244	245		return ctx.Err()
245	246		}

@@ -252,7 +253,7 @@ func (o outbox) sendStep(ctx context.Context, s ermite.StepUpdate) {
252	253		return err
253	254		})
254	255		if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
255		-	o.logf("UpdateJob refused step %d: %v", s.Number, err)
	256	+	o.log.Error().Err(err).Int32("step", s.Number).Msg("UpdateJob refused")
256	257		}
257	258		}
258	259

@@ -282,13 +283,20 @@ func (o *outbox) sendLines(ctx context.Context, batch []event) {
282	283		// so that the next offset is never a gap (out_of_range) and never a
283	284		// repeat of lines that would be skipped.
284	285		if want := req.Offset + int64(len(req.Lines)); next != want {
285		-	o.logf("AppendLog: the server holds %d lines, expected %d: continuing from %d", next, want, next)
	286	+	o.log.Warn().Int64("server_lines", next).Int64("expected", want).Msg("AppendLog: the server holds another number of lines, continuing from its count")
286	287		}
287	288		o.offset = next
288	289		o.mu.Unlock()
289	290		case ctx.Err() == nil && classify(err) == kindPermanent:
290	291		// The server refused these lines; they are dropped and the offset stays
291	292		// where the server is.
292		-	o.logf("AppendLog refused %d lines of step %d: %v", len(req.Lines), req.Step, err)
	293	+	o.log.Error().Err(err).Int("lines", len(req.Lines)).Int32("step", req.Step).Msg("AppendLog refused, the lines are dropped")
293	294		}
294	295		}
	296	+
	297	+	// AppendNow sends one line synchronously, after Close: the sender has stopped,
	298	+	// so nothing else uses the offset. It is for the last words of a job (an error
	299	+	// that explains why the result could not be reported).
	300	+	func (o *outbox) AppendNow(ctx context.Context, l logLine) {
	301	+	o.sendLines(ctx, []event{{line: &l}})
	302	+	}

 	"time"
 	"connectrpc.com/connect"
+	"github.com/bnema/zerowrap"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	ermite "bnema.dev/ermite-api/gen/ermite"
 	// onEnded is called once when the server says the job is over (JobEnded)
 	// or the assignment is gone; the outbox then drops everything.
 	onEnded func(error)
-	logf    func(format string, args ...any)
+	log     zerowrap.Logger
 	mu      sync.Mutex
 	cond    *sync.Cond
 	AppendLog(context.Context, *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error)
 }
-func newOutbox(c callClient, jobID string, t Timing, onEnded func(error), logf func(string, ...any)) *outbox {
+func newOutbox(c callClient, jobID string, t Timing, onEnded func(error), log zerowrap.Logger) *outbox {
-	o := &outbox{client: c, jobID: jobID, t: t, onEnded: onEnded, logf: logf, done: make(chan struct{})}
+	o := &outbox{client: c, jobID: jobID, t: t, onEnded: onEnded, log: log, done: make(chan struct{})}
 	o.cond = sync.NewCond(&o.mu)
 	return o
 }
 		case kindPermanent:
 			return err
 		}
-		o.logf("%s failed, retrying: %v", what, err)
+		o.log.Warn().Err(err).Str("call", what).Msg("call failed, retrying")
 		if !sleep(ctx, b.next()) {
 			return ctx.Err()
 		}
 		return err
 	})
 	if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
-		o.logf("UpdateJob refused step %d: %v", s.Number, err)
+		o.log.Error().Err(err).Int32("step", s.Number).Msg("UpdateJob refused")
 	}
 }
 		// so that the next offset is never a gap (out_of_range) and never a
 		// repeat of lines that would be skipped.
 		if want := req.Offset + int64(len(req.Lines)); next != want {
-			o.logf("AppendLog: the server holds %d lines, expected %d: continuing from %d", next, want, next)
+			o.log.Warn().Int64("server_lines", next).Int64("expected", want).Msg("AppendLog: the server holds another number of lines, continuing from its count")
 		}
 		o.offset = next
 		o.mu.Unlock()
 	case ctx.Err() == nil && classify(err) == kindPermanent:
 		// The server refused these lines; they are dropped and the offset stays
 		// where the server is.
-		o.logf("AppendLog refused %d lines of step %d: %v", len(req.Lines), req.Step, err)
+		o.log.Error().Err(err).Int("lines", len(req.Lines)).Int32("step", req.Step).Msg("AppendLog refused, the lines are dropped")
 	}
 }
+// AppendNow sends one line synchronously, after Close: the sender has stopped,
+// so nothing else uses the offset. It is for the last words of a job (an error
+// that explains why the result could not be reported).
+func (o *outbox) AppendNow(ctx context.Context, l logLine) {
+	o.sendLines(ctx, []event{{line: &l}})
+}

M internal/host/runner.go +19 -24


@@ -14,6 +14,7 @@ import (
14	14		"time"
15	15
16	16		"connectrpc.com/connect"
	17	+	"github.com/bnema/zerowrap"
17	18
18	19		ermite "bnema.dev/ermite-api/gen/ermite"
19	20		"bnema.dev/ermite-api/gen/ermite/ermiteconnect"

@@ -31,9 +32,6 @@ type Runner struct {
31	32		// Job builds the client of a job from its job token.
32	33		Job func(jobToken string) ermiteconnect.RunnerServiceClient
33	34		Timing Timing
34		-	Logf func(format string, args ...any)
35		-	// JobSpec sizes the sandbox of a job; the zero value takes the defaults.
36		-	JobSpec func(*ermite.Job) sandbox.JobSpec
37	35
38	36		// now returns the current time; tests replace it.
39	37		now func() time.Time

@@ -49,16 +47,12 @@ func (r *Runner) clock() time.Time {
49	47		return time.Now()
50	48		}
51	49
52		-	func (r *Runner) logf(format string, a ...any) {
53		-	if r.Logf != nil {
54		-	r.Logf(format, a...)
55		-	}
56		-	}
57		-
	50	+	// spec is the sandbox request of a job. It names the job and leaves CPUs,
	51	+	// memory and disk at zero: the driver fills them from its configuration
	52	+	// (sandbox.Config.Resolve in Start), that is from ERMITE_RUNNER_JOB_CPUS,
	53	+	// _JOB_MEMORY and _JOB_DISK, so that the limits are one setting of the runner
	54	+	// and not something a job or this package can change.
58	55		func (r Runner) spec(j ermite.Job) sandbox.JobSpec {
59		-	if r.JobSpec != nil {
60		-	return r.JobSpec(j)
61		-	}
62	56		return sandbox.JobSpec{JobID: j.JobId}
63	57		}
64	58

@@ -84,6 +78,7 @@ func newPollID() string {
84	78		// Capacity: a poll is made only when a slot is free. A lost response is retried
85	79		// with the same poll_id, so that the server hands back the same assignment.
86	80		func (r *Runner) Run(ctx context.Context) error {
	81	+	log := zerowrap.FromCtx(ctx)
87	82		t := r.Timing.withDefaults()
88	83		shutdown := make(chan struct{})
89	84		slots := make(chan struct{}, max(1, r.Cfg.Capacity))

@@ -95,7 +90,7 @@ func (r *Runner) Run(ctx context.Context) error {
95	90		}()
96	91
97	92		if err := r.Driver.Reap(ctx); err != nil && !errors.Is(err, sandbox.ErrBusy) {
98		-	r.logf("reaping what a previous run left: %v", err)
	93	+	log.Error().Err(err).Msg("reaping what a previous run left")
99	94		}
100	95
101	96		var pollID string

@@ -109,7 +104,7 @@ func (r *Runner) Run(ctx context.Context) error {
109	104		}
110	105		if r.leaked.Load() {
111	106		<-slots
112		-	if !r.recover(ctx, slots, t.ReapRetry) {
	107	+	if !r.recover(ctx, log, slots, t.ReapRetry) {
113	108		return nil
114	109		}
115	110		continue

@@ -132,7 +127,7 @@ func (r *Runner) Run(ctx context.Context) error {
132	127		pollID = "" // our id was refused: a new one
133	128		}
134	129		d := bo.next()
135		-	r.logf("FetchJob failed, retrying in %s: %v", d, err)
	130	+	log.Warn().Err(err).Dur("retry_in", d).Msg("FetchJob failed")
136	131		if !sleep(ctx, d) {
137	132		return nil
138	133		}

@@ -151,7 +146,7 @@ func (r *Runner) Run(ctx context.Context) error {
151	146		}
152	147		continue
153	148		}
154		-	r.startJob(ctx, resp.Msg.Job, t, shutdown, slots)
	149	+	r.startJob(ctx, log, resp.Msg.Job, t, shutdown, slots)
155	150		if ra := resp.Msg.RetryAfter; ra != nil && ra.AsDuration() > 0 {
156	151		if !sleep(ctx, ra.AsDuration()) {
157	152		return nil

@@ -162,16 +157,16 @@ func (r *Runner) Run(ctx context.Context) error {
162	157		}
163	158
164	159		// recover waits until no job runs, then reaps. It returns false when ctx ends.
165		-	func (r *Runner) recover(ctx context.Context, slots chan struct{}, retry time.Duration) bool {
166		-	r.logf("a sandbox could not be cleaned up: waiting for the running jobs, then reaping")
	160	+	func (r *Runner) recover(ctx context.Context, log zerowrap.Logger, slots chan struct{}, retry time.Duration) bool {
	161	+	log.Error().Msg("a sandbox could not be cleaned up: no new job until the running ones end and the leftovers are reaped")
167	162		for ctx.Err() == nil {
168	163		if len(slots) == 0 {
169	164		if err := r.Driver.Reap(ctx); err == nil {
170	165		r.leaked.Store(false)
171		-	r.logf("reaped: taking jobs again")
	166	+	log.Info().Msg("reaped: taking jobs again")
172	167		return true
173	168		} else if !errors.Is(err, sandbox.ErrBusy) {
174		-	r.logf("reap failed, retrying: %v", err)
	169	+	log.Error().Err(err).Msg("reaping failed, retrying")
175	170		}
176	171		}
177	172		if !sleep(ctx, retry) {

@@ -181,20 +176,20 @@ func (r *Runner) recover(ctx context.Context, slots chan struct{}, retry time.Du
181	176		return false
182	177		}
183	178
184		-	func (r Runner) startJob(ctx context.Context, job ermite.Job, t Timing, shutdown <-chan struct{}, slots chan struct{}) {
	179	+	func (r Runner) startJob(ctx context.Context, log zerowrap.Logger, job ermite.Job, t Timing, shutdown <-chan struct{}, slots chan struct{}) {
185	180		jctx, cancel := context.WithCancelCause(context.WithoutCancel(ctx))
186	181		jr := &jobRun{
187	182		r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
188		-	logf: func(f string, a ...any) { r.logf("job "+job.JobId+": "+f, a...) },
	183	+	log: zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},
189	184		receivedAt: r.clock(), cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},
190	185		}
191		-	r.logf("job %s (run %d, attempt %d, %s) assigned", job.JobId, job.RunId, job.Attempt, job.JobKey)
	186	+	jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")
192	187		r.active.Add(1)
193	188		go func() {
194	189		defer r.active.Done()
195	190		defer func() { <-slots }()
196	191		jr.run(r, shutdown)
197		-	r.logf("job %s finished", job.JobId)
	192	+	jr.log.Info().Msg("job finished")
198	193		}()
199	194		}
200	195

 	"time"
 	"connectrpc.com/connect"
+	"github.com/bnema/zerowrap"
 	ermite "bnema.dev/ermite-api/gen/ermite"
 	"bnema.dev/ermite-api/gen/ermite/ermiteconnect"
 	// Job builds the client of a job from its job token.
 	Job    func(jobToken string) ermiteconnect.RunnerServiceClient
 	Timing Timing
-	Logf   func(format string, args ...any)
-	// 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
 	return time.Now()
 }
-func (r *Runner) logf(format string, a ...any) {
+// spec is the sandbox request of a job. It names the job and leaves CPUs,
-	if r.Logf != nil {
+// memory and disk at zero: the driver fills them from its configuration
-		r.Logf(format, a...)
+// (sandbox.Config.Resolve in Start), that is from ERMITE_RUNNER_JOB_CPUS,
-	}
+// _JOB_MEMORY and _JOB_DISK, so that the limits are one setting of the runner
-}
+// and not something a job or this package can change.
 func (r *Runner) spec(j *ermite.Job) sandbox.JobSpec {
-	if r.JobSpec != nil {
-		return r.JobSpec(j)
-	}
 	return sandbox.JobSpec{JobID: j.JobId}
 }
 // Capacity: a poll is made only when a slot is free. A lost response is retried
 // with the same poll_id, so that the server hands back the same assignment.
 func (r *Runner) Run(ctx context.Context) error {
+	log := zerowrap.FromCtx(ctx)
 	t := r.Timing.withDefaults()
 	shutdown := make(chan struct{})
 	slots := make(chan struct{}, max(1, r.Cfg.Capacity))
 	}()
 	if err := r.Driver.Reap(ctx); err != nil && !errors.Is(err, sandbox.ErrBusy) {
-		r.logf("reaping what a previous run left: %v", err)
+		log.Error().Err(err).Msg("reaping what a previous run left")
 	}
 	var pollID string
 		}
 		if r.leaked.Load() {
 			<-slots
-			if !r.recover(ctx, slots, t.ReapRetry) {
+			if !r.recover(ctx, log, slots, t.ReapRetry) {
 				return nil
 			}
 			continue
 				pollID = "" // our id was refused: a new one
 			}
 			d := bo.next()
-			r.logf("FetchJob failed, retrying in %s: %v", d, err)
+			log.Warn().Err(err).Dur("retry_in", d).Msg("FetchJob failed")
 			if !sleep(ctx, d) {
 				return nil
 			}
 			}
 			continue
 		}
-		r.startJob(ctx, resp.Msg.Job, t, shutdown, slots)
+		r.startJob(ctx, log, resp.Msg.Job, t, shutdown, slots)
 		if ra := resp.Msg.RetryAfter; ra != nil && ra.AsDuration() > 0 {
 			if !sleep(ctx, ra.AsDuration()) {
 				return nil
 }
 // recover waits until no job runs, then reaps. It returns false when ctx ends.
-func (r *Runner) recover(ctx context.Context, slots chan struct{}, retry time.Duration) bool {
+func (r *Runner) recover(ctx context.Context, log zerowrap.Logger, slots chan struct{}, retry time.Duration) bool {
-	r.logf("a sandbox could not be cleaned up: waiting for the running jobs, then reaping")
+	log.Error().Msg("a sandbox could not be cleaned up: no new job until the running ones end and the leftovers are reaped")
 	for ctx.Err() == nil {
 		if len(slots) == 0 {
 			if err := r.Driver.Reap(ctx); err == nil {
 				r.leaked.Store(false)
-				r.logf("reaped: taking jobs again")
+				log.Info().Msg("reaped: taking jobs again")
 				return true
 			} else if !errors.Is(err, sandbox.ErrBusy) {
-				r.logf("reap failed, retrying: %v", err)
+				log.Error().Err(err).Msg("reaping failed, retrying")
 			}
 		}
 		if !sleep(ctx, retry) {
 	return false
 }
-func (r *Runner) startJob(ctx context.Context, job *ermite.Job, t Timing, shutdown <-chan struct{}, slots chan struct{}) {
+func (r *Runner) startJob(ctx context.Context, log zerowrap.Logger, job *ermite.Job, t Timing, shutdown <-chan struct{}, slots chan struct{}) {
 	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...) },
+		log:        zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},
 		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)
+	jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")
 	r.active.Add(1)
 	go func() {
 		defer r.active.Done()
 		defer func() { <-slots }()
 		jr.run(r, shutdown)
-		r.logf("job %s finished", job.JobId)
+		jr.log.Info().Msg("job finished")
 	}()
 }

M internal/host/runner_test.go +131 -2


@@ -4,6 +4,7 @@ package host
4	4
5	5		import (
6	6		"context"
	7	+	"encoding/json"
7	8		"errors"
8	9		"io"
9	10		"net/http"

@@ -130,7 +131,7 @@ func TestPollRespectsRetryAfter(t *testing.T) {
130	131		func TestPollRemovedRunnerStops(t *testing.T) {
131	132		h := newHarness(t, simpleScript("x"))
132	133		h.runner.Poll = NewClient(h.url, "ermiterun_wrong", http.DefaultTransport)
133		-	err := h.runner.Run(context.Background())
	134	+	err := h.runner.Run(h.logs.ctx())
134	135		if !errors.Is(err, ErrRemoved) {
135	136		t.Fatalf("%v", err)
136	137		}

@@ -1040,7 +1041,135 @@ func TestAppendLogResyncsToTheServersCount(t *testing.T) {
1040	1041		if len(h.srv.lines) < 3 \|\| h.srv.lines[0] != "planted" \|\| h.srv.lines[len(h.srv.lines)-1] != "l3" {
1041	1042		t.Fatalf("%q", h.srv.lines)
1042	1043		}
1043		-	if !strings.Contains(h.logs.String(), "continuing from") {
	1044	+	if !strings.Contains(h.logs.String(), "continuing from its count") {
1044	1045		t.Fatalf("no warning about the resync:\n%s", h.logs.String())
1045	1046		}
1046	1047		}
	1048	+
	1049	+	// ---- M4: the server refuses the result itself
	1050	+
	1051	+	func TestCompleteJobRefusedIsRetriedAsAFailureWithoutOutputs(t *testing.T) {
	1052	+	h := newHarness(t, simpleScript("hello"))
	1053	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	1054	+	h.srv.completeHook = func(n int) (bool, error) {
	1055	+	if n == 1 { // e.g. an output the server's rules refuse
	1056	+	return false, connect.NewError(connect.CodeInvalidArgument, io.EOF)
	1057	+	}
	1058	+	return false, nil
	1059	+	}
	1060	+	h.run(t)
	1061	+	eventually(t, "two CompleteJob calls", func() bool { return h.srv.completeCount() == 2 })
	1062	+	time.Sleep(100 * time.Millisecond)
	1063	+	h.srv.mu.Lock()
	1064	+	defer h.srv.mu.Unlock()
	1065	+	if len(h.srv.completes) != 2 {
	1066	+	t.Fatalf("%d calls", len(h.srv.completes))
	1067	+	}
	1068	+	first, second := h.srv.completes[0], h.srv.completes[1]
	1069	+	if first.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS \|\| first.Outputs["out"] != "v" {
	1070	+	t.Fatalf("first: %v %v", first.Conclusion, first.Outputs)
	1071	+	}
	1072	+	if second.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE \|\| len(second.Outputs) != 0 \|\| second.RunnerShutdown {
	1073	+	t.Fatalf("second: %v %v", second.Conclusion, second.Outputs)
	1074	+	}
	1075	+	if len(second.Steps) != 3 {
	1076	+	t.Fatalf("steps %v", second.Steps)
	1077	+	}
	1078	+	for _, u := range second.Steps {
	1079	+	if u.State != ermite.StepState_STEP_STATE_COMPLETED \|\| u.StartedAt == nil \|\| u.CompletedAt == nil \|\| u.CompletedAt.AsTime().Before(u.StartedAt.AsTime()) {
	1080	+	t.Errorf("invalid step %v", u)
	1081	+	}
	1082	+	}
	1083	+	last := h.srv.lines[len(h.srv.lines)-1]
	1084	+	if !strings.HasPrefix(last, "##[error]The runner could not report the result of the job: invalid_argument") {
	1085	+	t.Fatalf("last line %q", last)
	1086	+	}
	1087	+	}
	1088	+
	1089	+	func TestCompleteJobRefusedTwiceGivesUp(t *testing.T) {
	1090	+	h := newHarness(t, simpleScript("x"))
	1091	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	1092	+	h.srv.completeHook = func(int) (bool, error) {
	1093	+	return false, connect.NewError(connect.CodeFailedPrecondition, io.EOF) // not a JobEnded detail
	1094	+	}
	1095	+	h.run(t)
	1096	+	eventually(t, "two calls", func() bool { return h.srv.completeCount() == 2 })
	1097	+	time.Sleep(200 * time.Millisecond)
	1098	+	if n := h.srv.completeCount(); n != 2 {
	1099	+	t.Fatalf("%d calls: one retry only", n)
	1100	+	}
	1101	+	if !strings.Contains(h.logs.String(), `"level":"error"`) \|\| !strings.Contains(h.logs.String(), "refused again") {
	1102	+	t.Fatalf("no error log:\n%s", h.logs.String())
	1103	+	}
	1104	+	}
	1105	+
	1106	+	// ---- G1: structured logs with the job's fields, warn for retries, error for refusals
	1107	+
	1108	+	func TestLogsCarryTheJobFieldsAndLevels(t *testing.T) {
	1109	+	h := newHarness(t, simpleScript("x"))
	1110	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	1111	+	h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
	1112	+	if n == 1 {
	1113	+	return false, connect.NewError(connect.CodeUnavailable, io.EOF)
	1114	+	}
	1115	+	return false, nil
	1116	+	}
	1117	+	h.srv.beatHook = func(n int) error { return nil }
	1118	+	h.run(t)
	1119	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	1120	+	var retry, assigned map[string]any
	1121	+	for _, l := range strings.Split(h.logs.String(), "\n") {
	1122	+	if l == "" {
	1123	+	continue
	1124	+	}
	1125	+	var m map[string]any
	1126	+	if err := json.Unmarshal([]byte(l), &m); err != nil {
	1127	+	t.Fatalf("not JSON: %q", l)
	1128	+	}
	1129	+	switch m["message"] {
	1130	+	case "call failed, retrying":
	1131	+	retry = m
	1132	+	case "job assigned":
	1133	+	assigned = m
	1134	+	}
	1135	+	}
	1136	+	for name, m := range map[string]map[string]any{"retry": retry, "assigned": assigned} {
	1137	+	if m == nil {
	1138	+	t.Fatalf("no %s event in\n%s", name, h.logs.String())
	1139	+	}
	1140	+	if m["job_id"] != "j1" \|\| m["run_id"] != float64(7) \|\| m["attempt"] != float64(1) {
	1141	+	t.Errorf("%s event lacks the job fields: %v", name, m)
	1142	+	}
	1143	+	}
	1144	+	if retry["level"] != "warn" \|\| retry["call"] != "AppendLog" {
	1145	+	t.Errorf("a retry is a warning: %v", retry)
	1146	+	}
	1147	+	}
	1148	+
	1149	+	// The sizes of a sandbox come from ERMITE_RUNNER_JOB_CPUS, _JOB_MEMORY and
	1150	+	// _JOB_DISK (sandbox.Config.Resolve, applied by the driver at Start): the host
	1151	+	// asks for none, so that nothing in a job can change them.
	1152	+	func TestSandboxSizesAreLeftToTheDriversConfiguration(t *testing.T) {
	1153	+	h := newHarness(t, simpleScript("x"))
	1154	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	1155	+	h.run(t)
	1156	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	1157	+	h.driver.mu.Lock()
	1158	+	defer h.driver.mu.Unlock()
	1159	+	if len(h.driver.specs) != 1 {
	1160	+	t.Fatalf("%v", h.driver.specs)
	1161	+	}
	1162	+	if s := h.driver.specs[0]; s.JobID != "j1" \|\| s.CPUs != 0 \|\| s.MemoryMiB != 0 \|\| s.DiskMiB != 0 {
	1163	+	t.Fatalf("%+v", s)
	1164	+	}
	1165	+	// And the configuration fills them (the real rule, in package sandbox).
	1166	+	cfg, err := sandbox.LoadConfig(func(k string) string {
	1167	+	return map[string]string{sandbox.EnvJobCPUs: "3", sandbox.EnvJobMemory: "2G", sandbox.EnvJobDisk: "10G"}[k]
	1168	+	})
	1169	+	if err != nil {
	1170	+	t.Fatal(err)
	1171	+	}
	1172	+	if got := cfg.Resolve(h.driver.specs[0]); got.CPUs != 3 \|\| got.MemoryMiB != 2048 \|\| got.DiskMiB != 10240 {
	1173	+	t.Fatalf("%+v", got)
	1174	+	}
	1175	+	}


@@ -4,6 +4,7 @@ package host
4			4
5		import (	5		import (
6		"context"	6		"context"
			7	+	"encoding/json"
7		"errors"	8		"errors"
8		"io"	9		"io"
9		"net/http"	10		"net/http"

@@ -130,7 +131,7 @@ func TestPollRespectsRetryAfter(t *testing.T) {
130		func TestPollRemovedRunnerStops(t *testing.T) {	131		func TestPollRemovedRunnerStops(t *testing.T) {
131		h := newHarness(t, simpleScript("x"))	132		h := newHarness(t, simpleScript("x"))
132		h.runner.Poll = NewClient(h.url, "ermiterun_wrong", http.DefaultTransport)	133		h.runner.Poll = NewClient(h.url, "ermiterun_wrong", http.DefaultTransport)
133	-	err := h.runner.Run(context.Background())	134	+	err := h.runner.Run(h.logs.ctx())
134		if !errors.Is(err, ErrRemoved) {	135		if !errors.Is(err, ErrRemoved) {
135		t.Fatalf("%v", err)	136		t.Fatalf("%v", err)
136		}	137		}

@@ -1040,7 +1041,135 @@ func TestAppendLogResyncsToTheServersCount(t *testing.T) {
1040		if len(h.srv.lines) < 3 \|\| h.srv.lines[0] != "planted" \|\| h.srv.lines[len(h.srv.lines)-1] != "l3" {	1041		if len(h.srv.lines) < 3 \|\| h.srv.lines[0] != "planted" \|\| h.srv.lines[len(h.srv.lines)-1] != "l3" {
1041		t.Fatalf("%q", h.srv.lines)	1042		t.Fatalf("%q", h.srv.lines)
1042		}	1043		}
1043	-	if !strings.Contains(h.logs.String(), "continuing from") {	1044	+	if !strings.Contains(h.logs.String(), "continuing from its count") {
1044		t.Fatalf("no warning about the resync:\n%s", h.logs.String())	1045		t.Fatalf("no warning about the resync:\n%s", h.logs.String())
1045		}	1046		}
1046		}	1047		}
			1048	+
			1049	+	// ---- M4: the server refuses the result itself
			1050	+
			1051	+	func TestCompleteJobRefusedIsRetriedAsAFailureWithoutOutputs(t *testing.T) {
			1052	+	h := newHarness(t, simpleScript("hello"))
			1053	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
			1054	+	h.srv.completeHook = func(n int) (bool, error) {
			1055	+	if n == 1 { // e.g. an output the server's rules refuse
			1056	+	return false, connect.NewError(connect.CodeInvalidArgument, io.EOF)
			1057	+	}
			1058	+	return false, nil
			1059	+	}
			1060	+	h.run(t)
			1061	+	eventually(t, "two CompleteJob calls", func() bool { return h.srv.completeCount() == 2 })
			1062	+	time.Sleep(100 * time.Millisecond)
			1063	+	h.srv.mu.Lock()
			1064	+	defer h.srv.mu.Unlock()
			1065	+	if len(h.srv.completes) != 2 {
			1066	+	t.Fatalf("%d calls", len(h.srv.completes))
			1067	+	}
			1068	+	first, second := h.srv.completes[0], h.srv.completes[1]
			1069	+	if first.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS \|\| first.Outputs["out"] != "v" {
			1070	+	t.Fatalf("first: %v %v", first.Conclusion, first.Outputs)
			1071	+	}
			1072	+	if second.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE \|\| len(second.Outputs) != 0 \|\| second.RunnerShutdown {
			1073	+	t.Fatalf("second: %v %v", second.Conclusion, second.Outputs)
			1074	+	}
			1075	+	if len(second.Steps) != 3 {
			1076	+	t.Fatalf("steps %v", second.Steps)
			1077	+	}
			1078	+	for _, u := range second.Steps {
			1079	+	if u.State != ermite.StepState_STEP_STATE_COMPLETED \|\| u.StartedAt == nil \|\| u.CompletedAt == nil \|\| u.CompletedAt.AsTime().Before(u.StartedAt.AsTime()) {
			1080	+	t.Errorf("invalid step %v", u)
			1081	+	}
			1082	+	}
			1083	+	last := h.srv.lines[len(h.srv.lines)-1]
			1084	+	if !strings.HasPrefix(last, "##[error]The runner could not report the result of the job: invalid_argument") {
			1085	+	t.Fatalf("last line %q", last)
			1086	+	}
			1087	+	}
			1088	+
			1089	+	func TestCompleteJobRefusedTwiceGivesUp(t *testing.T) {
			1090	+	h := newHarness(t, simpleScript("x"))
			1091	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
			1092	+	h.srv.completeHook = func(int) (bool, error) {
			1093	+	return false, connect.NewError(connect.CodeFailedPrecondition, io.EOF) // not a JobEnded detail
			1094	+	}
			1095	+	h.run(t)
			1096	+	eventually(t, "two calls", func() bool { return h.srv.completeCount() == 2 })
			1097	+	time.Sleep(200 * time.Millisecond)
			1098	+	if n := h.srv.completeCount(); n != 2 {
			1099	+	t.Fatalf("%d calls: one retry only", n)
			1100	+	}
			1101	+	if !strings.Contains(h.logs.String(), `"level":"error"`) \|\| !strings.Contains(h.logs.String(), "refused again") {
			1102	+	t.Fatalf("no error log:\n%s", h.logs.String())
			1103	+	}
			1104	+	}
			1105	+
			1106	+	// ---- G1: structured logs with the job's fields, warn for retries, error for refusals
			1107	+
			1108	+	func TestLogsCarryTheJobFieldsAndLevels(t *testing.T) {
			1109	+	h := newHarness(t, simpleScript("x"))
			1110	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
			1111	+	h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
			1112	+	if n == 1 {
			1113	+	return false, connect.NewError(connect.CodeUnavailable, io.EOF)
			1114	+	}
			1115	+	return false, nil
			1116	+	}
			1117	+	h.srv.beatHook = func(n int) error { return nil }
			1118	+	h.run(t)
			1119	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
			1120	+	var retry, assigned map[string]any
			1121	+	for _, l := range strings.Split(h.logs.String(), "\n") {
			1122	+	if l == "" {
			1123	+	continue
			1124	+	}
			1125	+	var m map[string]any
			1126	+	if err := json.Unmarshal([]byte(l), &m); err != nil {
			1127	+	t.Fatalf("not JSON: %q", l)
			1128	+	}
			1129	+	switch m["message"] {
			1130	+	case "call failed, retrying":
			1131	+	retry = m
			1132	+	case "job assigned":
			1133	+	assigned = m
			1134	+	}
			1135	+	}
			1136	+	for name, m := range map[string]map[string]any{"retry": retry, "assigned": assigned} {
			1137	+	if m == nil {
			1138	+	t.Fatalf("no %s event in\n%s", name, h.logs.String())
			1139	+	}
			1140	+	if m["job_id"] != "j1" \|\| m["run_id"] != float64(7) \|\| m["attempt"] != float64(1) {
			1141	+	t.Errorf("%s event lacks the job fields: %v", name, m)
			1142	+	}
			1143	+	}
			1144	+	if retry["level"] != "warn" \|\| retry["call"] != "AppendLog" {
			1145	+	t.Errorf("a retry is a warning: %v", retry)
			1146	+	}
			1147	+	}
			1148	+
			1149	+	// The sizes of a sandbox come from ERMITE_RUNNER_JOB_CPUS, _JOB_MEMORY and
			1150	+	// _JOB_DISK (sandbox.Config.Resolve, applied by the driver at Start): the host
			1151	+	// asks for none, so that nothing in a job can change them.
			1152	+	func TestSandboxSizesAreLeftToTheDriversConfiguration(t *testing.T) {
			1153	+	h := newHarness(t, simpleScript("x"))
			1154	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
			1155	+	h.run(t)
			1156	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
			1157	+	h.driver.mu.Lock()
			1158	+	defer h.driver.mu.Unlock()
			1159	+	if len(h.driver.specs) != 1 {
			1160	+	t.Fatalf("%v", h.driver.specs)
			1161	+	}
			1162	+	if s := h.driver.specs[0]; s.JobID != "j1" \|\| s.CPUs != 0 \|\| s.MemoryMiB != 0 \|\| s.DiskMiB != 0 {
			1163	+	t.Fatalf("%+v", s)
			1164	+	}
			1165	+	// And the configuration fills them (the real rule, in package sandbox).
			1166	+	cfg, err := sandbox.LoadConfig(func(k string) string {
			1167	+	return map[string]string{sandbox.EnvJobCPUs: "3", sandbox.EnvJobMemory: "2G", sandbox.EnvJobDisk: "10G"}[k]
			1168	+	})
			1169	+	if err != nil {
			1170	+	t.Fatal(err)
			1171	+	}
			1172	+	if got := cfg.Resolve(h.driver.specs[0]); got.CPUs != 3 \|\| got.MemoryMiB != 2048 \|\| got.DiskMiB != 10240 {
			1173	+	t.Fatalf("%+v", got)
			1174	+	}
			1175	+	}

M internal/host/util_test.go +0 -3


@@ -3,12 +3,9 @@
3	3		package host
4	4
5	5		import (
6		-	"fmt"
7	6		"time"
8	7
9	8		"google.golang.org/protobuf/types/known/durationpb"
10	9		)
11	10
12		-	func sprintf(f string, a ...any) string { return fmt.Sprintf(f, a...) }
13		-
14	11		func durationpbOf(d time.Duration) *durationpb.Duration { return durationpb.New(d) }


@@ -3,12 +3,9 @@
3		package host	3	package host
4			4
5		import (	5	import (
6	-	"fmt"
7		"time"	6	"time"
8			7
9		"google.golang.org/protobuf/types/known/durationpb"	8	"google.golang.org/protobuf/types/known/durationpb"
10		)	9	)
11			10
12	-	func sprintf(f string, a ...any) string { return fmt.Sprintf(f, a...) }
13	-
14		func durationpbOf(d time.Duration) *durationpb.Duration { return durationpb.New(d) }	11	func durationpbOf(d time.Duration) *durationpb.Duration { return durationpb.New(d) }