fix(host): do not trust the guest's step numbers and lines (M3); resync to the server's line count (m6); one AppendLog validator for the host tests (G3)

commit: af4d4473f2c7655ee51466b0855a5d2c722cd48c
author: brice · 9d ago · 2026-10-12 19:31 UTC
parent: a21f495
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +478 -87

M internal/host/fake_test.go +38 -35


@@ -20,6 +20,7 @@ import (
20	20
21	21		"bnema.dev/ermite-runner/internal/execstream"
22	22		"bnema.dev/ermite-runner/internal/execstream/execpb"
	23	+	"bnema.dev/ermite-runner/internal/host/hosttest"
23	24		"bnema.dev/ermite-runner/internal/sandbox"
24	25		)
25	26

@@ -42,7 +43,9 @@ type fakeServer struct {
42	43		beatHook func(n int) error
43	44		registers []*ermite.RegisterRunnerRequest
44	45
45		-	lines []string // log text, joined pieces are not merged: one per line
	46	+	store *hosttest.LogStore
	47	+	violations []string
	48	+	lines []string // log text, one per line (mirror of store)
46	49		cont []bool
47	50		steps map[int32]*ermite.StepUpdate
48	51		appendReq []*ermite.AppendLogRequest

@@ -60,10 +63,12 @@ type fakeServer struct {
60	63		}
61	64
62	65		func newFakeServer() *fakeServer {
63		-	return &fakeServer{
	66	+	s := &fakeServer{
64	67		runnerTok: "ermiterun_tok", assigned: map[string]ermite.Job{}, completed: map[string]bool{}, steps: map[int32]ermite.StepUpdate{},
65	68		watchMsgs: make(chan *ermite.WatchJobResponse, 16),
66	69		}
	70	+	s.store = &hosttest.LogStore{OnViolation: func(m string) { s.violations = append(s.violations, m) }}
	71	+	return s
67	72		}
68	73
69	74		func endedErr(je *ermite.JobEnded) error {

@@ -185,44 +190,28 @@ func (s fakeServer) AppendLog(_ context.Context, r connect.Request[ermite.Appe
185	190		}
186	191		s.appendN++
187	192		s.appendReq = append(s.appendReq, r.Msg)
188		-	if len(r.Msg.Lines) == 0 \|\| len(r.Msg.Lines) > 1000 {
189		-	return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
190		-	}
191		-	bytes := 0
192		-	for _, l := range r.Msg.Lines {
193		-	bytes += len(l.Text)
194		-	if strings.Contains(l.Text, "\n") \|\| l.Time == nil {
195		-	return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
196		-	}
197		-	}
198		-	if bytes > 64*1024 {
199		-	return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
200		-	}
201		-	have := int64(len(s.lines))
202		-	if r.Msg.Offset > have {
203		-	return nil, connect.NewError(connect.CodeOutOfRange, io.EOF)
204		-	}
205		-	skip := have - r.Msg.Offset
206		-	apply := func() {
207		-	for i, l := range r.Msg.Lines {
208		-	if int64(i) < skip {
209		-	continue
210		-	}
211		-	s.lines = append(s.lines, l.Text)
212		-	s.cont = append(s.cont, l.Continued)
213		-	}
214		-	}
	193	+	// The rules of runner.proto are written once, in hosttest.LogStore.
	194	+	var failAfter error
215	195		if s.appendHook != nil {
216	196		thenFail, err := s.appendHook(s.appendN, r.Msg)
217		-	if err != nil {
218		-	if thenFail {
219		-	apply()
220		-	}
	197	+	if err != nil && !thenFail {
221	198		return nil, err
222	199		}
	200	+	failAfter = err
223	201		}
224		-	apply()
225		-	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: int64(len(s.lines))}), nil
	202	+	next, err := s.store.Apply(r.Msg)
	203	+	if err != nil {
	204	+	return nil, err
	205	+	}
	206	+	s.lines, s.cont = s.lines[:0], s.cont[:0]
	207	+	for _, l := range s.store.Lines() {
	208	+	s.lines = append(s.lines, l.Text)
	209	+	s.cont = append(s.cont, l.Continued)
	210	+	}
	211	+	if failAfter != nil {
	212	+	return nil, failAfter
	213	+	}
	214	+	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: next}), nil
226	215		}
227	216
228	217		func (s fakeServer) CompleteJob(_ context.Context, r connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {

@@ -497,6 +486,13 @@ func newHarness(t testing.T, sc script) harness {
497	486		drv := &fakeDriver{script: sc, t: t, iso: sandbox.IsolationVM}
498	487		lb := &logBuf{}
499	488		h := &harness{srv: srv, url: url, driver: drv, logs: lb}
	489	+	t.Cleanup(func() {
	490	+	srv.mu.Lock()
	491	+	defer srv.mu.Unlock()
	492	+	if len(srv.violations) > 0 {
	493	+	t.Errorf("the host broke the contract of AppendLog: %q", srv.violations)
	494	+	}
	495	+	})
500	496		h.runner = &Runner{
501	497		Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
502	498		Driver: drv,

@@ -543,3 +539,10 @@ func eventually(t *testing.T, what string, cond func() bool) {
543	539		}
544	540
545	541		func (s *fakeServer) completeCount() int { s.mu.Lock(); defer s.mu.Unlock(); return len(s.completes) }
	542	+
	543	+	// Result returns what a job reported, like hosttest.Server.Result.
	544	+	func (s *fakeServer) Result(jobID string) hosttest.Result {
	545	+	s.mu.Lock()
	546	+	defer s.mu.Unlock()
	547	+	return hosttest.Result{Steps: s.steps}
	548	+	}

 	"bnema.dev/ermite-runner/internal/execstream"
 	"bnema.dev/ermite-runner/internal/execstream/execpb"
+	"bnema.dev/ermite-runner/internal/host/hosttest"
 	"bnema.dev/ermite-runner/internal/sandbox"
 )
 	beatHook  func(n int) error
 	registers []*ermite.RegisterRunnerRequest
-	lines      []string // log text, joined pieces are not merged: one per line
+	store      *hosttest.LogStore
+	violations []string
+	lines      []string // log text, one per line (mirror of store)
 	cont       []bool
 	steps      map[int32]*ermite.StepUpdate
 	appendReq  []*ermite.AppendLogRequest
 }
 func newFakeServer() *fakeServer {
-	return &fakeServer{
+	s := &fakeServer{
 		runnerTok: "ermiterun_tok", assigned: map[string]*ermite.Job{}, completed: map[string]bool{}, steps: map[int32]*ermite.StepUpdate{},
 		watchMsgs: make(chan *ermite.WatchJobResponse, 16),
 	}
+	s.store = &hosttest.LogStore{OnViolation: func(m string) { s.violations = append(s.violations, m) }}
+	return s
 }
 func endedErr(je *ermite.JobEnded) error {
 	}
 	s.appendN++
 	s.appendReq = append(s.appendReq, r.Msg)
-	if len(r.Msg.Lines) == 0 || len(r.Msg.Lines) > 1000 {
+	// The rules of runner.proto are written once, in hosttest.LogStore.
-		return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
+	var failAfter error
-	}
-	bytes := 0
-	for _, l := range r.Msg.Lines {
-		bytes += len(l.Text)
-		if strings.Contains(l.Text, "\n") || l.Time == nil {
-			return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
-		}
-	}
-	if bytes > 64*1024 {
-		return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
-	}
-	have := int64(len(s.lines))
-	if r.Msg.Offset > have {
-		return nil, connect.NewError(connect.CodeOutOfRange, io.EOF)
-	}
-	skip := have - r.Msg.Offset
-	apply := func() {
-		for i, l := range r.Msg.Lines {
-			if int64(i) < skip {
-				continue
-			}
-			s.lines = append(s.lines, l.Text)
-			s.cont = append(s.cont, l.Continued)
-		}
-	}
 	if s.appendHook != nil {
 		thenFail, err := s.appendHook(s.appendN, r.Msg)
-		if err != nil {
+		if err != nil && !thenFail {
-			if thenFail {
-				apply()
-			}
 			return nil, err
 		}
+		failAfter = err
 	}
-	apply()
+	next, err := s.store.Apply(r.Msg)
-	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: int64(len(s.lines))}), nil
+	if err != nil {
+		return nil, err
+	}
+	s.lines, s.cont = s.lines[:0], s.cont[:0]
+	for _, l := range s.store.Lines() {
+		s.lines = append(s.lines, l.Text)
+		s.cont = append(s.cont, l.Continued)
+	}
+	if failAfter != nil {
+		return nil, failAfter
+	}
+	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: next}), nil
 }
 func (s *fakeServer) CompleteJob(_ context.Context, r *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
 	drv := &fakeDriver{script: sc, t: t, iso: sandbox.IsolationVM}
 	lb := &logBuf{}
 	h := &harness{srv: srv, url: url, driver: drv, logs: lb}
+	t.Cleanup(func() {
+		srv.mu.Lock()
+		defer srv.mu.Unlock()
+		if len(srv.violations) > 0 {
+			t.Errorf("the host broke the contract of AppendLog: %q", srv.violations)
+		}
+	})
 	h.runner = &Runner{
 		Cfg:    Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
 		Driver: drv,
 }
 func (s *fakeServer) completeCount() int { s.mu.Lock(); defer s.mu.Unlock(); return len(s.completes) }
+// Result returns what a job reported, like hosttest.Server.Result.
+func (s *fakeServer) Result(jobID string) hosttest.Result {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return hosttest.Result{Steps: s.steps}
+}

M internal/host/hosttest/hosttest.go +9 -36


@@ -55,7 +55,7 @@ type Server struct {
55	55		}
56	56
57	57		type jobState struct {
58		-	lines []Line
	58	+	log *LogStore
59	59		steps map[int32]*ermite.StepUpdate
60	60		complete *ermite.CompleteJobRequest
61	61		}

@@ -101,10 +101,10 @@ func (s *Server) Result(jobID string) Result {
101	101		for k, v := range st.steps {
102	102		steps[k] = v
103	103		}
104		-	return Result{Complete: st.complete, Lines: append([]Line(nil), st.lines...), Steps: steps}
	104	+	return Result{Complete: st.complete, Lines: st.log.Lines(), Steps: steps}
105	105		}
106	106
107		-	func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) }
	107	+	func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) } // called with s.mu held
108	108
109	109		func unauth() error { return connect.NewError(connect.CodeUnauthenticated, io.EOF) }
110	110

@@ -140,7 +140,7 @@ func (s Server) FetchJob(_ context.Context, r connect.Request[ermite.FetchJobR
140	140		j := s.queue[0]
141	141		s.queue = s.queue[1:]
142	142		s.assigned[j.JobToken] = j
143		-	s.jobs[j.JobId] = &jobState{steps: map[int32]*ermite.StepUpdate{}}
	143	+	s.jobs[j.JobId] = &jobState{log: &LogStore{OnViolation: s.violation}, steps: map[int32]*ermite.StepUpdate{}}
144	144		return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
145	145		}
146	146

@@ -178,7 +178,7 @@ func (s Server) UpdateJob(_ context.Context, r connect.Request[ermite.UpdateJo
178	178		return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
179	179		}
180	180
181		-	// AppendLog applies the offset rules of runner.proto.
	181	+	// AppendLog applies the rules of runner.proto through a LogStore.
182	182		func (s Server) AppendLog(_ context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
183	183		s.mu.Lock()
184	184		defer s.mu.Unlock()

@@ -190,38 +190,11 @@ func (s Server) AppendLog(_ context.Context, r connect.Request[ermite.AppendLo
190	190		s.violation("AppendLog after CompleteJob")
191	191		return nil, ended(st.complete.Conclusion)
192	192		}
193		-	n, bytes := len(r.Msg.Lines), 0
194		-	if n == 0 \|\| n > 1000 {
195		-	s.violation("AppendLog: bad number of lines")
196		-	return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
197		-	}
198		-	for i, l := range r.Msg.Lines {
199		-	bytes += len(l.Text)
200		-	if strings.Contains(l.Text, "\n") \|\| l.Time == nil \|\| (l.Continued && i == 0 && len(st.lines) == 0 && r.Msg.Offset == 0) {
201		-	s.violation("AppendLog: invalid line")
202		-	return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
203		-	}
204		-	}
205		-	if bytes > 64*1024 {
206		-	s.violation("AppendLog: more than 64 KiB")
207		-	return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
208		-	}
209		-	have := int64(len(st.lines))
210		-	if r.Msg.Offset > have {
211		-	s.violation("AppendLog: a gap in the offsets")
212		-	return nil, connect.NewError(connect.CodeOutOfRange, io.EOF)
213		-	}
214		-	if have > 0 && r.Msg.Offset == have && r.Msg.Step < st.lines[have-1].Step {
215		-	s.violation("AppendLog: a step went back")
216		-	return nil, connect.NewError(connect.CodeFailedPrecondition, io.EOF)
217		-	}
218		-	for i, l := range r.Msg.Lines {
219		-	if int64(i) < have-r.Msg.Offset {
220		-	continue
221		-	}
222		-	st.lines = append(st.lines, Line{Step: r.Msg.Step, Text: l.Text, Continued: l.Continued})
	193	+	next, err := st.log.Apply(r.Msg)
	194	+	if err != nil {
	195	+	return nil, err
223	196		}
224		-	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: int64(len(st.lines))}), nil
	197	+	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: next}), nil
225	198		}
226	199
227	200		// CompleteJob ends a job, idempotently.

 }
 type jobState struct {
-	lines    []Line
+	log      *LogStore
 	steps    map[int32]*ermite.StepUpdate
 	complete *ermite.CompleteJobRequest
 }
 	for k, v := range st.steps {
 		steps[k] = v
 	}
-	return Result{Complete: st.complete, Lines: append([]Line(nil), st.lines...), Steps: steps}
+	return Result{Complete: st.complete, Lines: st.log.Lines(), Steps: steps}
 }
-func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) }
+func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) } // called with s.mu held
 func unauth() error { return connect.NewError(connect.CodeUnauthenticated, io.EOF) }
 	j := s.queue[0]
 	s.queue = s.queue[1:]
 	s.assigned[j.JobToken] = j
-	s.jobs[j.JobId] = &jobState{steps: map[int32]*ermite.StepUpdate{}}
+	s.jobs[j.JobId] = &jobState{log: &LogStore{OnViolation: s.violation}, steps: map[int32]*ermite.StepUpdate{}}
 	return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
 }
 	return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
 }
-// AppendLog applies the offset rules of runner.proto.
+// AppendLog applies the rules of runner.proto through a LogStore.
 func (s *Server) AppendLog(_ context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 		s.violation("AppendLog after CompleteJob")
 		return nil, ended(st.complete.Conclusion)
 	}
-	n, bytes := len(r.Msg.Lines), 0
+	next, err := st.log.Apply(r.Msg)
-	if n == 0 || n > 1000 {
+	if err != nil {
-		s.violation("AppendLog: bad number of lines")
+		return nil, err
-		return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
-	}
-	for i, l := range r.Msg.Lines {
-		bytes += len(l.Text)
-		if strings.Contains(l.Text, "\n") || l.Time == nil || (l.Continued && i == 0 && len(st.lines) == 0 && r.Msg.Offset == 0) {
-			s.violation("AppendLog: invalid line")
-			return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
-		}
-	}
-	if bytes > 64*1024 {
-		s.violation("AppendLog: more than 64 KiB")
-		return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
-	}
-	have := int64(len(st.lines))
-	if r.Msg.Offset > have {
-		s.violation("AppendLog: a gap in the offsets")
-		return nil, connect.NewError(connect.CodeOutOfRange, io.EOF)
-	}
-	if have > 0 && r.Msg.Offset == have && r.Msg.Step < st.lines[have-1].Step {
-		s.violation("AppendLog: a step went back")
-		return nil, connect.NewError(connect.CodeFailedPrecondition, io.EOF)
-	}
-	for i, l := range r.Msg.Lines {
-		if int64(i) < have-r.Msg.Offset {
-			continue
-		}
-		st.lines = append(st.lines, Line{Step: r.Msg.Step, Text: l.Text, Continued: l.Continued})
 	}
-	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: int64(len(st.lines))}), nil
+	return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: next}), nil
 }
 // CompleteJob ends a job, idempotently.

A internal/host/hosttest/logstore.go +89 -0

@@ -0,0 +1,89 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package hosttest
	4	+
	5	+	import (
	6	+	"io"
	7	+	"strings"
	8	+	"sync"
	9	+
	10	+	"connectrpc.com/connect"
	11	+
	12	+	ermite "bnema.dev/ermite-api/gen/ermite"
	13	+	)
	14	+
	15	+	// LogStore is the log of one job as the server of runner.proto keeps it, with
	16	+	// the rules of AppendLog (the one place where they are written down for tests):
	17	+	//
	18	+	// - a request has 1 to 1,000 lines, at most 64 KiB of text in total, no line
	19	+	// feed in a line, a time on every line, and a continued first line only if
	20	+	// it follows a line;
	21	+	// - offset is the number of lines the server holds: equal appends, below skips
	22	+	// the first (count - offset) lines of the request (a retry), above is
	23	+	// out_of_range;
	24	+	// - the step of a request is at least the step of the last stored line
	25	+	// (failed_precondition otherwise; lines skipped as already stored are not
	26	+	// checked).
	27	+	//
	28	+	// A request that breaks a rule the client must never break (the first group)
	29	+	// is reported through OnViolation as well as answered with the error.
	30	+	type LogStore struct {
	31	+	// OnViolation, when set, receives the description of each client violation.
	32	+	OnViolation func(string)
	33	+
	34	+	mu sync.Mutex
	35	+	lines []Line
	36	+	}
	37	+
	38	+	// Lines returns a copy of the stored lines.
	39	+	func (s *LogStore) Lines() []Line {
	40	+	s.mu.Lock()
	41	+	defer s.mu.Unlock()
	42	+	return append([]Line(nil), s.lines...)
	43	+	}
	44	+
	45	+	// Apply applies a request and returns the number of lines held afterwards.
	46	+	func (s LogStore) Apply(r ermite.AppendLogRequest) (int64, error) {
	47	+	s.mu.Lock()
	48	+	defer s.mu.Unlock()
	49	+	bad := func(code connect.Code, why string) (int64, error) {
	50	+	if s.OnViolation != nil {
	51	+	s.OnViolation("AppendLog: " + why)
	52	+	}
	53	+	return 0, connect.NewError(code, io.EOF)
	54	+	}
	55	+	if r.Step < 1 {
	56	+	return bad(connect.CodeInvalidArgument, "step below 1")
	57	+	}
	58	+	n, bytes := len(r.Lines), 0
	59	+	if n == 0 \|\| n > 1000 {
	60	+	return bad(connect.CodeInvalidArgument, "bad number of lines")
	61	+	}
	62	+	have := int64(len(s.lines))
	63	+	for i, l := range r.Lines {
	64	+	bytes += len(l.Text)
	65	+	if strings.Contains(l.Text, "\n") \|\| l.Time == nil {
	66	+	return bad(connect.CodeInvalidArgument, "invalid line")
	67	+	}
	68	+	if l.Continued && i == 0 && have == 0 && r.Offset == 0 {
	69	+	return bad(connect.CodeInvalidArgument, "the first line of the job is continued")
	70	+	}
	71	+	}
	72	+	if bytes > 64*1024 {
	73	+	return bad(connect.CodeInvalidArgument, "more than 64 KiB")
	74	+	}
	75	+	if r.Offset > have {
	76	+	return bad(connect.CodeOutOfRange, "a gap in the offsets")
	77	+	}
	78	+	skip := have - r.Offset
	79	+	if skip < int64(n) && have > 0 && r.Step < s.lines[have-1].Step {
	80	+	return bad(connect.CodeFailedPrecondition, "a step went back")
	81	+	}
	82	+	for i, l := range r.Lines {
	83	+	if int64(i) < skip {
	84	+	continue
	85	+	}
	86	+	s.lines = append(s.lines, Line{Step: r.Step, Text: l.Text, Continued: l.Continued})
	87	+	}
	88	+	return int64(len(s.lines)), nil
	89	+	}

@@ -0,0 +1,89 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package hosttest
	4	+
	5	+	import (
	6	+	"io"
	7	+	"strings"
	8	+	"sync"
	9	+
	10	+	"connectrpc.com/connect"
	11	+
	12	+	ermite "bnema.dev/ermite-api/gen/ermite"
	13	+	)
	14	+
	15	+	// LogStore is the log of one job as the server of runner.proto keeps it, with
	16	+	// the rules of AppendLog (the one place where they are written down for tests):
	17	+	//
	18	+	// - a request has 1 to 1,000 lines, at most 64 KiB of text in total, no line
	19	+	// feed in a line, a time on every line, and a continued first line only if
	20	+	// it follows a line;
	21	+	// - offset is the number of lines the server holds: equal appends, below skips
	22	+	// the first (count - offset) lines of the request (a retry), above is
	23	+	// out_of_range;
	24	+	// - the step of a request is at least the step of the last stored line
	25	+	// (failed_precondition otherwise; lines skipped as already stored are not
	26	+	// checked).
	27	+	//
	28	+	// A request that breaks a rule the client must never break (the first group)
	29	+	// is reported through OnViolation as well as answered with the error.
	30	+	type LogStore struct {
	31	+	// OnViolation, when set, receives the description of each client violation.
	32	+	OnViolation func(string)
	33	+
	34	+	mu sync.Mutex
	35	+	lines []Line
	36	+	}
	37	+
	38	+	// Lines returns a copy of the stored lines.
	39	+	func (s *LogStore) Lines() []Line {
	40	+	s.mu.Lock()
	41	+	defer s.mu.Unlock()
	42	+	return append([]Line(nil), s.lines...)
	43	+	}
	44	+
	45	+	// Apply applies a request and returns the number of lines held afterwards.
	46	+	func (s LogStore) Apply(r ermite.AppendLogRequest) (int64, error) {
	47	+	s.mu.Lock()
	48	+	defer s.mu.Unlock()
	49	+	bad := func(code connect.Code, why string) (int64, error) {
	50	+	if s.OnViolation != nil {
	51	+	s.OnViolation("AppendLog: " + why)
	52	+	}
	53	+	return 0, connect.NewError(code, io.EOF)
	54	+	}
	55	+	if r.Step < 1 {
	56	+	return bad(connect.CodeInvalidArgument, "step below 1")
	57	+	}
	58	+	n, bytes := len(r.Lines), 0
	59	+	if n == 0 \|\| n > 1000 {
	60	+	return bad(connect.CodeInvalidArgument, "bad number of lines")
	61	+	}
	62	+	have := int64(len(s.lines))
	63	+	for i, l := range r.Lines {
	64	+	bytes += len(l.Text)
	65	+	if strings.Contains(l.Text, "\n") \|\| l.Time == nil {
	66	+	return bad(connect.CodeInvalidArgument, "invalid line")
	67	+	}
	68	+	if l.Continued && i == 0 && have == 0 && r.Offset == 0 {
	69	+	return bad(connect.CodeInvalidArgument, "the first line of the job is continued")
	70	+	}
	71	+	}
	72	+	if bytes > 64*1024 {
	73	+	return bad(connect.CodeInvalidArgument, "more than 64 KiB")
	74	+	}
	75	+	if r.Offset > have {
	76	+	return bad(connect.CodeOutOfRange, "a gap in the offsets")
	77	+	}
	78	+	skip := have - r.Offset
	79	+	if skip < int64(n) && have > 0 && r.Step < s.lines[have-1].Step {
	80	+	return bad(connect.CodeFailedPrecondition, "a step went back")
	81	+	}
	82	+	for i, l := range r.Lines {
	83	+	if int64(i) < skip {
	84	+	continue
	85	+	}
	86	+	s.lines = append(s.lines, Line{Step: r.Step, Text: l.Text, Continued: l.Continued})
	87	+	}
	88	+	return int64(len(s.lines)), nil
	89	+	}

M internal/host/job.go +134 -12


@@ -10,6 +10,7 @@ import (
10	10		"strings"
11	11		"sync"
12	12		"time"
	13	+	"unicode/utf8"
13	14
14	15		"connectrpc.com/connect"
15	16		"google.golang.org/protobuf/proto"

@@ -57,8 +58,16 @@ type jobRun struct {
57	58		cancelKind execpb.CancelKind // kind of the first cancel request
58	59		stopping bool // the runner is shutting down
59	60
60		-	mu sync.Mutex
61		-	steps map[int32]*ermite.StepUpdate
	61	+	mu sync.Mutex
	62	+	steps map[int32]*ermite.StepUpdate
	63	+	// What the guest has told us, for the validation of the next frame: the
	64	+	// guest is untrusted code, its step numbers and lines are checked before
	65	+	// they reach the server (see violate).
	66	+	maxStarted int32
	67	+	lastLogStep int32
	68	+	violation string
	69	+	// hard stops the sandbox at once; set before the session starts.
	70	+	hard func()
62	71		result *execpb.JobResult
63	72		outbox *outbox
64	73		endedBy error // the server ended the job, or the assignment is gone

@@ -216,9 +225,57 @@ func kindOf(r ermite.CancelReason) execpb.CancelKind {
216	225
217	226		// ---- steps
218	227
	228	+	// Limits on what the guest may send, which keep the memory of the host bounded
	229	+	// whatever the guest does: runner.proto allows 1,000 steps per job, names of
	230	+	// 1,024 characters, and lines of 64 KiB.
	231	+	const (
	232	+	maxSteps = 1000
	233	+	maxNameLen = 1024
	234	+	maxLineText = 64 * 1024
	235	+	)
	236	+
	237	+	// violate fails the job because the guest broke the protocol: it writes the
	238	+	// reason in the log, asks the executor to stop and destroys the sandbox. Only
	239	+	// the first violation counts; the frames that follow are dropped.
	240	+	func (j *jobRun) violate(format string, a ...any) {
	241	+	j.mu.Lock()
	242	+	if j.violation != "" {
	243	+	j.mu.Unlock()
	244	+	return
	245	+	}
	246	+	j.violation = fmt.Sprintf(format, a...)
	247	+	msg := j.violation
	248	+	j.mu.Unlock()
	249	+	j.logf("the executor broke the protocol: %s", msg)
	250	+	j.sandboxError("The runner stopped the job: the executor broke the protocol: " + msg)
	251	+	if j.hard != nil {
	252	+	j.hard()
	253	+	}
	254	+	}
	255	+
	256	+	func (j *jobRun) violated() bool {
	257	+	j.mu.Lock()
	258	+	defer j.mu.Unlock()
	259	+	return j.violation != ""
	260	+	}
	261	+
219	262		func (j jobRun) stepStarted(s execpb.StepStarted) {
220		-	u := &ermite.StepUpdate{Number: s.Number, Name: s.Name, State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
221	263		j.mu.Lock()
	264	+	switch {
	265	+	case j.violation != "":
	266	+	j.mu.Unlock()
	267	+	return
	268	+	case s.Number < 1 \|\| s.Number > maxSteps:
	269	+	j.mu.Unlock()
	270	+	j.violate("step number %d is outside 1..%d", s.Number, maxSteps)
	271	+	return
	272	+	case s.Number <= j.maxStarted:
	273	+	j.mu.Unlock()
	274	+	j.violate("step %d started after step %d", s.Number, j.maxStarted)
	275	+	return
	276	+	}
	277	+	j.maxStarted = s.Number
	278	+	u := &ermite.StepUpdate{Number: s.Number, Name: cutName(s.Name), State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
222	279		j.steps[s.Number] = u
223	280		j.mu.Unlock()
224	281		j.outbox.Step(u)

@@ -226,14 +283,24 @@ func (j jobRun) stepStarted(s execpb.StepStarted) {
226	283
227	284		func (j jobRun) stepCompleted(s execpb.StepCompleted) {
228	285		j.mu.Lock()
229		-	u := j.steps[s.Number]
230		-	if u == nil {
231		-	u = &ermite.StepUpdate{Number: s.Number, StartedAt: tsOf(s.UnixNano)}
232		-	} else {
233		-	u = cloneStep(u)
	286	+	if j.violation != "" {
	287	+	j.mu.Unlock()
	288	+	return
234	289		}
	290	+	old := j.steps[s.Number]
	291	+	switch {
	292	+	case old == nil:
	293	+	j.mu.Unlock()
	294	+	j.violate("step %d completed without having started", s.Number)
	295	+	return
	296	+	case old.State == ermite.StepState_STEP_STATE_COMPLETED:
	297	+	j.mu.Unlock()
	298	+	j.violate("step %d completed twice", s.Number)
	299	+	return
	300	+	}
	301	+	u := cloneStep(old)
235	302		if s.Name != "" {
236		-	u.Name = s.Name
	303	+	u.Name = cutName(s.Name)
237	304		}
238	305		u.State = ermite.StepState_STEP_STATE_COMPLETED
239	306		u.Conclusion = conclusionOf(s.Conclusion)

@@ -246,6 +313,58 @@ func (j jobRun) stepCompleted(s execpb.StepCompleted) {
246	313		j.outbox.Step(u)
247	314		}
248	315
	316	+	// logLine checks and queues a line of the log. The step must be one that was
	317	+	// started and not completed, and not lower than the step of the previous line
	318	+	// (the server refuses a log that goes back); the text is made valid for the
	319	+	// server: UTF-8, no line feed (a line feed in a line makes more lines).
	320	+	func (j jobRun) logLine(l execpb.Log) {
	321	+	j.mu.Lock()
	322	+	if j.violation != "" {
	323	+	j.mu.Unlock()
	324	+	return
	325	+	}
	326	+	st := j.steps[l.Step]
	327	+	switch {
	328	+	case l.Step < 1 \|\| l.Step > maxSteps \|\| st == nil:
	329	+	j.mu.Unlock()
	330	+	j.violate("a log line for step %d, which has not started", l.Step)
	331	+	return
	332	+	case st.State == ermite.StepState_STEP_STATE_COMPLETED:
	333	+	j.mu.Unlock()
	334	+	j.violate("a log line for step %d, which is completed", l.Step)
	335	+	return
	336	+	case l.Step < j.lastLogStep:
	337	+	j.mu.Unlock()
	338	+	j.violate("a log line for step %d after one for step %d", l.Step, j.lastLogStep)
	339	+	return
	340	+	case len(l.Text) > maxLineText:
	341	+	j.mu.Unlock()
	342	+	j.violate("a log line of %d bytes (the limit is %d)", len(l.Text), maxLineText)
	343	+	return
	344	+	}
	345	+	j.lastLogStep = l.Step
	346	+	j.mu.Unlock()
	347	+	text := l.Text
	348	+	if !utf8.ValidString(text) {
	349	+	text = strings.ToValidUTF8(text, "\ufffd")
	350	+	}
	351	+	for i, part := range strings.Split(text, "\n") {
	352	+	j.outbox.Log(logLine{step: l.Step, unixNano: l.UnixNano, text: part, continued: l.Continued && i == 0})
	353	+	}
	354	+	}
	355	+
	356	+	// cutName bounds a step name to what the server keeps, on a character boundary.
	357	+	func cutName(s string) string {
	358	+	if len(s) <= maxNameLen {
	359	+	return strings.ToValidUTF8(s, "\ufffd")
	360	+	}
	361	+	s = s[:maxNameLen]
	362	+	for !utf8.ValidString(s) && len(s) > 0 {
	363	+	s = s[:len(s)-1]
	364	+	}
	365	+	return s
	366	+	}
	367	+
249	368		func cloneStep(u ermite.StepUpdate) ermite.StepUpdate {
250	369		return proto.Clone(u).(*ermite.StepUpdate)
251	370		}

@@ -325,6 +444,7 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
325	444		stopHard := make(chan struct{})
326	445		var hardOnce sync.Once
327	446		hard := func() { hardOnce.Do(func() { close(stopHard) }) }
	447	+	j.hard = hard
328	448		go func() {
329	449		select {
330	450		case <-shutdown:

@@ -442,9 +562,7 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
442	562		}}
443	563		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
444	564		StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
445		-	Log: func(l *execpb.Log) {
446		-	j.outbox.Log(logLine{step: l.Step, unixNano: l.UnixNano, text: l.Text, continued: l.Continued})
447		-	},
	565	+	Log: j.logLine,
448	566		JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
449	567		})
450	568		j.mu.Lock()

@@ -452,6 +570,10 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
452	570		j.mu.Unlock()
453	571
454	572		switch {
	573	+	case j.violated():
	574	+	// The step numbers or lines of the executor were wrong: whatever it
	575	+	// reported is not trusted, the job fails (the reason is in the log).
	576	+	return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
455	577		case res != nil:
456	578		c := conclusionOf(res.Conclusion)
457	579		if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED \|\| c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {


@@ -10,6 +10,7 @@ import (
10		"strings"	10		"strings"
11		"sync"	11		"sync"
12		"time"	12		"time"
			13	+	"unicode/utf8"
13			14
14		"connectrpc.com/connect"	15		"connectrpc.com/connect"
15		"google.golang.org/protobuf/proto"	16		"google.golang.org/protobuf/proto"

@@ -57,8 +58,16 @@ type jobRun struct {
57		cancelKind execpb.CancelKind // kind of the first cancel request	58		cancelKind execpb.CancelKind // kind of the first cancel request
58		stopping bool // the runner is shutting down	59		stopping bool // the runner is shutting down
59			60
60	-	mu sync.Mutex	61	+	mu sync.Mutex
61	-	steps map[int32]*ermite.StepUpdate	62	+	steps map[int32]*ermite.StepUpdate
			63	+	// What the guest has told us, for the validation of the next frame: the
			64	+	// guest is untrusted code, its step numbers and lines are checked before
			65	+	// they reach the server (see violate).
			66	+	maxStarted int32
			67	+	lastLogStep int32
			68	+	violation string
			69	+	// hard stops the sandbox at once; set before the session starts.
			70	+	hard func()
62		result *execpb.JobResult	71		result *execpb.JobResult
63		outbox *outbox	72		outbox *outbox
64		endedBy error // the server ended the job, or the assignment is gone	73		endedBy error // the server ended the job, or the assignment is gone

@@ -216,9 +225,57 @@ func kindOf(r ermite.CancelReason) execpb.CancelKind {
216			225
217		// ---- steps	226		// ---- steps
218			227
			228	+	// Limits on what the guest may send, which keep the memory of the host bounded
			229	+	// whatever the guest does: runner.proto allows 1,000 steps per job, names of
			230	+	// 1,024 characters, and lines of 64 KiB.
			231	+	const (
			232	+	maxSteps = 1000
			233	+	maxNameLen = 1024
			234	+	maxLineText = 64 * 1024
			235	+	)
			236	+
			237	+	// violate fails the job because the guest broke the protocol: it writes the
			238	+	// reason in the log, asks the executor to stop and destroys the sandbox. Only
			239	+	// the first violation counts; the frames that follow are dropped.
			240	+	func (j *jobRun) violate(format string, a ...any) {
			241	+	j.mu.Lock()
			242	+	if j.violation != "" {
			243	+	j.mu.Unlock()
			244	+	return
			245	+	}
			246	+	j.violation = fmt.Sprintf(format, a...)
			247	+	msg := j.violation
			248	+	j.mu.Unlock()
			249	+	j.logf("the executor broke the protocol: %s", msg)
			250	+	j.sandboxError("The runner stopped the job: the executor broke the protocol: " + msg)
			251	+	if j.hard != nil {
			252	+	j.hard()
			253	+	}
			254	+	}
			255	+
			256	+	func (j *jobRun) violated() bool {
			257	+	j.mu.Lock()
			258	+	defer j.mu.Unlock()
			259	+	return j.violation != ""
			260	+	}
			261	+
219		func (j jobRun) stepStarted(s execpb.StepStarted) {	262		func (j jobRun) stepStarted(s execpb.StepStarted) {
220	-	u := &ermite.StepUpdate{Number: s.Number, Name: s.Name, State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
221		j.mu.Lock()	263		j.mu.Lock()
			264	+	switch {
			265	+	case j.violation != "":
			266	+	j.mu.Unlock()
			267	+	return
			268	+	case s.Number < 1 \|\| s.Number > maxSteps:
			269	+	j.mu.Unlock()
			270	+	j.violate("step number %d is outside 1..%d", s.Number, maxSteps)
			271	+	return
			272	+	case s.Number <= j.maxStarted:
			273	+	j.mu.Unlock()
			274	+	j.violate("step %d started after step %d", s.Number, j.maxStarted)
			275	+	return
			276	+	}
			277	+	j.maxStarted = s.Number
			278	+	u := &ermite.StepUpdate{Number: s.Number, Name: cutName(s.Name), State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
222		j.steps[s.Number] = u	279		j.steps[s.Number] = u
223		j.mu.Unlock()	280		j.mu.Unlock()
224		j.outbox.Step(u)	281		j.outbox.Step(u)

@@ -226,14 +283,24 @@ func (j jobRun) stepStarted(s execpb.StepStarted) {
226			283
227		func (j jobRun) stepCompleted(s execpb.StepCompleted) {	284		func (j jobRun) stepCompleted(s execpb.StepCompleted) {
228		j.mu.Lock()	285		j.mu.Lock()
229	-	u := j.steps[s.Number]	286	+	if j.violation != "" {
230	-	if u == nil {	287	+	j.mu.Unlock()
231	-	u = &ermite.StepUpdate{Number: s.Number, StartedAt: tsOf(s.UnixNano)}	288	+	return
232	-	} else {
233	-	u = cloneStep(u)
234		}	289		}
			290	+	old := j.steps[s.Number]
			291	+	switch {
			292	+	case old == nil:
			293	+	j.mu.Unlock()
			294	+	j.violate("step %d completed without having started", s.Number)
			295	+	return
			296	+	case old.State == ermite.StepState_STEP_STATE_COMPLETED:
			297	+	j.mu.Unlock()
			298	+	j.violate("step %d completed twice", s.Number)
			299	+	return
			300	+	}
			301	+	u := cloneStep(old)
235		if s.Name != "" {	302		if s.Name != "" {
236	-	u.Name = s.Name	303	+	u.Name = cutName(s.Name)
237		}	304		}
238		u.State = ermite.StepState_STEP_STATE_COMPLETED	305		u.State = ermite.StepState_STEP_STATE_COMPLETED
239		u.Conclusion = conclusionOf(s.Conclusion)	306		u.Conclusion = conclusionOf(s.Conclusion)

@@ -246,6 +313,58 @@ func (j jobRun) stepCompleted(s execpb.StepCompleted) {
246		j.outbox.Step(u)	313		j.outbox.Step(u)
247		}	314		}
248			315
			316	+	// logLine checks and queues a line of the log. The step must be one that was
			317	+	// started and not completed, and not lower than the step of the previous line
			318	+	// (the server refuses a log that goes back); the text is made valid for the
			319	+	// server: UTF-8, no line feed (a line feed in a line makes more lines).
			320	+	func (j jobRun) logLine(l execpb.Log) {
			321	+	j.mu.Lock()
			322	+	if j.violation != "" {
			323	+	j.mu.Unlock()
			324	+	return
			325	+	}
			326	+	st := j.steps[l.Step]
			327	+	switch {
			328	+	case l.Step < 1 \|\| l.Step > maxSteps \|\| st == nil:
			329	+	j.mu.Unlock()
			330	+	j.violate("a log line for step %d, which has not started", l.Step)
			331	+	return
			332	+	case st.State == ermite.StepState_STEP_STATE_COMPLETED:
			333	+	j.mu.Unlock()
			334	+	j.violate("a log line for step %d, which is completed", l.Step)
			335	+	return
			336	+	case l.Step < j.lastLogStep:
			337	+	j.mu.Unlock()
			338	+	j.violate("a log line for step %d after one for step %d", l.Step, j.lastLogStep)
			339	+	return
			340	+	case len(l.Text) > maxLineText:
			341	+	j.mu.Unlock()
			342	+	j.violate("a log line of %d bytes (the limit is %d)", len(l.Text), maxLineText)
			343	+	return
			344	+	}
			345	+	j.lastLogStep = l.Step
			346	+	j.mu.Unlock()
			347	+	text := l.Text
			348	+	if !utf8.ValidString(text) {
			349	+	text = strings.ToValidUTF8(text, "\ufffd")
			350	+	}
			351	+	for i, part := range strings.Split(text, "\n") {
			352	+	j.outbox.Log(logLine{step: l.Step, unixNano: l.UnixNano, text: part, continued: l.Continued && i == 0})
			353	+	}
			354	+	}
			355	+
			356	+	// cutName bounds a step name to what the server keeps, on a character boundary.
			357	+	func cutName(s string) string {
			358	+	if len(s) <= maxNameLen {
			359	+	return strings.ToValidUTF8(s, "\ufffd")
			360	+	}
			361	+	s = s[:maxNameLen]
			362	+	for !utf8.ValidString(s) && len(s) > 0 {
			363	+	s = s[:len(s)-1]
			364	+	}
			365	+	return s
			366	+	}
			367	+
249		func cloneStep(u ermite.StepUpdate) ermite.StepUpdate {	368		func cloneStep(u ermite.StepUpdate) ermite.StepUpdate {
250		return proto.Clone(u).(*ermite.StepUpdate)	369		return proto.Clone(u).(*ermite.StepUpdate)
251		}	370		}

@@ -325,6 +444,7 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
325		stopHard := make(chan struct{})	444		stopHard := make(chan struct{})
326		var hardOnce sync.Once	445		var hardOnce sync.Once
327		hard := func() { hardOnce.Do(func() { close(stopHard) }) }	446		hard := func() { hardOnce.Do(func() { close(stopHard) }) }
			447	+	j.hard = hard
328		go func() {	448		go func() {
329		select {	449		select {
330		case <-shutdown:	450		case <-shutdown:

@@ -442,9 +562,7 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
442		}}	562		}}
443		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{	563		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
444		StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,	564		StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
445	-	Log: func(l *execpb.Log) {	565	+	Log: j.logLine,
446	-	j.outbox.Log(logLine{step: l.Step, unixNano: l.UnixNano, text: l.Text, continued: l.Continued})
447	-	},
448		JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },	566		JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
449		})	567		})
450		j.mu.Lock()	568		j.mu.Lock()

@@ -452,6 +570,10 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
452		j.mu.Unlock()	570		j.mu.Unlock()
453			571
454		switch {	572		switch {
			573	+	case j.violated():
			574	+	// The step numbers or lines of the executor were wrong: whatever it
			575	+	// reported is not trusted, the job fails (the reason is in the log).
			576	+	return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
455		case res != nil:	577		case res != nil:
456		c := conclusionOf(res.Conclusion)	578		c := conclusionOf(res.Conclusion)
457		if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED \|\| c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {	579		if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED \|\| c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {

M internal/host/outbox.go +11 -4


@@ -264,20 +264,27 @@ func (o *outbox) sendLines(ctx context.Context, batch []event) {
264	264		// The offset is fixed before the first attempt: every retry sends the same
265	265		// lines at the same offset.
266	266		req.Offset = o.offset
	267	+	var next int64
267	268		err := o.retry(ctx, "AppendLog", func(c context.Context) error {
268	269		resp, err := o.client.AppendLog(c, connect.NewRequest(req))
269	270		if err != nil {
270	271		return err
271	272		}
272		-	if want := req.Offset + int64(len(req.Lines)); resp.Msg.NextOffset != want {
273		-	o.logf("AppendLog: the server holds %d lines, expected %d", resp.Msg.NextOffset, want)
274		-	}
	273	+	next = resp.Msg.NextOffset
275	274		return nil
276	275		})
277	276		switch {
278	277		case err == nil:
279	278		o.mu.Lock()
280		-	o.offset += int64(len(req.Lines))
	279	+	// The server says how many lines it holds: that is the truth. A
	280	+	// difference means lines were stored that we did not count (a retry
	281	+	// whose answer we lost) or dropped; continue from the server's count,
	282	+	// so that the next offset is never a gap (out_of_range) and never a
	283	+	// repeat of lines that would be skipped.
	284	+	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	+	}
	287	+	o.offset = next
281	288		o.mu.Unlock()
282	289		case ctx.Err() == nil && classify(err) == kindPermanent:
283	290		// The server refused these lines; they are dropped and the offset stays

 	// The offset is fixed before the first attempt: every retry sends the same
 	// lines at the same offset.
 	req.Offset = o.offset
+	var next int64
 	err := o.retry(ctx, "AppendLog", func(c context.Context) error {
 		resp, err := o.client.AppendLog(c, connect.NewRequest(req))
 		if err != nil {
 			return err
 		}
-		if want := req.Offset + int64(len(req.Lines)); resp.Msg.NextOffset != want {
+		next = resp.Msg.NextOffset
-			o.logf("AppendLog: the server holds %d lines, expected %d", resp.Msg.NextOffset, want)
-		}
 		return nil
 	})
 	switch {
 	case err == nil:
 		o.mu.Lock()
-		o.offset += int64(len(req.Lines))
+		// The server says how many lines it holds: that is the truth. A
+		// difference means lines were stored that we did not count (a retry
+		// whose answer we lost) or dropped; continue from the server's count,
+		// 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.offset = next
 		o.mu.Unlock()
 	case ctx.Err() == nil && classify(err) == kindPermanent:
 		// The server refused these lines; they are dropped and the offset stays

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


@@ -11,6 +11,9 @@ import (
11	11		"strings"
12	12		"testing"
13	13		"time"
	14	+	"unicode/utf8"
	15	+
	16	+	"google.golang.org/protobuf/types/known/timestamppb"
14	17
15	18		"connectrpc.com/connect"
16	19

@@ -847,3 +850,197 @@ func TestRunCarriesTheRemainingDeadline(t *testing.T) {
847	850		t.Fatalf("%v", spec)
848	851		}
849	852		}
	853	+
	854	+	// ---- the guest is untrusted: step numbers and lines are checked (M3)
	855	+
	856	+	// hostile runs a job whose executor sends the frames of body, and checks that
	857	+	// the job fails, the sandbox is destroyed, the server never saw a violation of
	858	+	// its contract, and a line in the log names the reason.
	859	+	func hostile(t testing.T, wantReason string, body script) harness {
	860	+	t.Helper()
	861	+	h := newHarness(t, func(t testing.T, c execstream.Conn, run execpb.Run, cn <-chan execpb.Cancel) {
	862	+	body(t, c, run, cn)
	863	+	// A real executor would go on; this one waits for the host to cut it.
	864	+	<-cn
	865	+	})
	866	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	867	+	h.run(t)
	868	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	869	+	h.srv.mu.Lock()
	870	+	defer h.srv.mu.Unlock()
	871	+	c := h.srv.completes[0]
	872	+	if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE \|\| len(c.Outputs) != 0 {
	873	+	t.Fatalf("conclusion %v outputs %v", c.Conclusion, c.Outputs)
	874	+	}
	875	+	found := false
	876	+	for _, l := range h.srv.lines {
	877	+	if strings.Contains(l, "broke the protocol") && strings.Contains(l, wantReason) {
	878	+	found = true
	879	+	}
	880	+	}
	881	+	if !found {
	882	+	t.Fatalf("no error line mentioning %q in %q", wantReason, h.srv.lines)
	883	+	}
	884	+	for _, u := range c.Steps {
	885	+	if u.State != ermite.StepState_STEP_STATE_COMPLETED \|\| u.Number < 1 \|\| u.Number > 1000 {
	886	+	t.Errorf("invalid final step %v", u)
	887	+	}
	888	+	}
	889	+	if h.driver.stops != 1 {
	890	+	t.Errorf("the sandbox was not destroyed: stops=%d", h.driver.stops)
	891	+	}
	892	+	return h
	893	+	}
	894	+
	895	+	func TestGuestStepNumberZeroFailsTheJob(t *testing.T) {
	896	+	hostile(t, "outside 1..1000", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	897	+	stepStart(c, 1, "Set up job")
	898	+	stepStart(c, 0, "zero")
	899	+	})
	900	+	}
	901	+
	902	+	func TestGuestStepNumberOverTheLimitFailsTheJob(t *testing.T) {
	903	+	hostile(t, "outside 1..1000", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	904	+	stepStart(c, 1, "Set up job")
	905	+	stepStart(c, 1001, "far")
	906	+	})
	907	+	}
	908	+
	909	+	func TestGuestStepGoingBackFailsTheJob(t *testing.T) {
	910	+	hostile(t, "started after step 3", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	911	+	stepStart(c, 1, "a")
	912	+	stepStart(c, 3, "c")
	913	+	stepStart(c, 2, "b")
	914	+	})
	915	+	}
	916	+
	917	+	func TestGuestStepRestartedFailsTheJob(t *testing.T) {
	918	+	hostile(t, "started after step 1", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	919	+	stepStart(c, 1, "a")
	920	+	stepDone(c, 1, "a", cOK)
	921	+	stepStart(c, 1, "again")
	922	+	})
	923	+	}
	924	+
	925	+	func TestGuestCompletedStepIsNeverTouchedAgain(t *testing.T) {
	926	+	h := hostile(t, "completed twice", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	927	+	stepStart(c, 1, "a")
	928	+	stepDone(c, 1, "a", cOK)
	929	+	stepDone(c, 1, "a", cFail)
	930	+	})
	931	+	if u := h.srv.Result("j1").Steps[1]; u == nil \|\| u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS {
	932	+	t.Errorf("a completed step was changed: %v", u.Conclusion)
	933	+	}
	934	+	}
	935	+
	936	+	func TestGuestCompletingAnUnknownStepFailsTheJob(t *testing.T) {
	937	+	hostile(t, "without having started", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	938	+	stepStart(c, 1, "a")
	939	+	stepDone(c, 7, "ghost", cOK)
	940	+	})
	941	+	}
	942	+
	943	+	func TestGuestLogForAnUnstartedOrCompletedStepFailsTheJob(t *testing.T) {
	944	+	hostile(t, "has not started", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	945	+	stepStart(c, 1, "a")
	946	+	logLineFrame(c, 5, "x", false)
	947	+	})
	948	+	hostile(t, "which is completed", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	949	+	stepStart(c, 1, "a")
	950	+	stepDone(c, 1, "a", cOK)
	951	+	logLineFrame(c, 1, "late", false)
	952	+	})
	953	+	}
	954	+
	955	+	func TestGuestLogGoingBackAStepFailsTheJob(t *testing.T) {
	956	+	hostile(t, "after one for step 2", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	957	+	stepStart(c, 1, "a")
	958	+	stepStart(c, 2, "b")
	959	+	logLineFrame(c, 2, "x", false)
	960	+	logLineFrame(c, 1, "y", false)
	961	+	})
	962	+	}
	963	+
	964	+	func TestGuestOversizedLineFailsTheJob(t *testing.T) {
	965	+	hostile(t, "bytes (the limit is 65536)", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	966	+	stepStart(c, 1, "a")
	967	+	logLineFrame(c, 1, strings.Repeat("x", 64*1024+1), false)
	968	+	})
	969	+	}
	970	+
	971	+	// Text that the server would refuse is repaired, not forwarded: a line feed
	972	+	// makes more lines and long names are cut. (Invalid UTF-8 cannot reach the host:
	973	+	// proto3 strings are validated when the frame is decoded, which fails the
	974	+	// session; the executor replaces invalid bytes before it sends.)
	975	+	func TestGuestTextIsMadeValidForTheServer(t *testing.T) {
	976	+	h := newHarness(t, func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	977	+	stepStart(c, 1, strings.Repeat("é", 800)) // 1,600 bytes
	978	+	logLineFrame(c, 1, "two\nlines", false)
	979	+	stepDone(c, 1, "", cOK)
	980	+	jobDone(c, cOK, nil, false)
	981	+	})
	982	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	983	+	h.run(t)
	984	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	985	+	h.srv.mu.Lock()
	986	+	defer h.srv.mu.Unlock()
	987	+	if strings.Join(h.srv.lines, "\|") != "two\|lines" {
	988	+	t.Fatalf("%q", h.srv.lines)
	989	+	}
	990	+	if n := h.srv.steps[1].Name; len(n) > 1024 \|\| !utf8.ValidString(n) {
	991	+	t.Fatalf("name of %d bytes", len(n))
	992	+	}
	993	+	}
	994	+
	995	+	// A guest that sends steps without end cannot grow the memory of the host.
	996	+	func TestGuestFloodIsBounded(t *testing.T) {
	997	+	h := hostile(t, "outside 1..1000", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	998	+	for n := 1; n <= 1100; n++ {
	999	+	stepStart(c, n, "s")
	1000	+	stepDone(c, n, "s", cOK)
	1001	+	}
	1002	+	})
	1003	+	_ = h
	1004	+	}
	1005	+
	1006	+	// The server's line count is the truth (m6): when the answer to an AppendLog
	1007	+	// says the server holds more lines than the request accounts for, the next
	1008	+	// offset starts from the server's count and no gap or repeat is created.
	1009	+	func TestAppendLogResyncsToTheServersCount(t *testing.T) {
	1010	+	h := newHarness(t, func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	1011	+	stepStart(c, 1, "a")
	1012	+	logLineFrame(c, 1, "l1", false)
	1013	+	time.Sleep(150 * time.Millisecond) // a first request of its own
	1014	+	logLineFrame(c, 1, "l2", false)
	1015	+	logLineFrame(c, 1, "l3", false)
	1016	+	stepDone(c, 1, "a", cOK)
	1017	+	jobDone(c, cOK, nil, false)
	1018	+	})
	1019	+	h.runner.Timing.LogFlush = 30 * time.Millisecond
	1020	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	1021	+	// A line the runner never sent appears in the server's log (a retry whose
	1022	+	// answer was lost, in a real server): the runner must continue after it.
	1023	+	h.srv.mu.Lock()
	1024	+	h.srv.appendHook = func(n int, r *ermite.AppendLogRequest) (bool, error) {
	1025	+	if n == 1 {
	1026	+	_, _ = h.srv.store.Apply(&ermite.AppendLogRequest{JobId: r.JobId, Step: 1, Offset: 0, Lines: []*ermite.LogLine{
	1027	+	{Time: timestamppb.Now(), Text: "planted"}, {Time: timestamppb.Now(), Text: "planted2"}}})
	1028	+	return false, nil
	1029	+	}
	1030	+	return false, nil
	1031	+	}
	1032	+	h.srv.mu.Unlock()
	1033	+	h.run(t)
	1034	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	1035	+	h.srv.mu.Lock()
	1036	+	defer h.srv.mu.Unlock()
	1037	+	// The first request (offset 0, "l1") was skipped by the server because the
	1038	+	// planted lines already took offsets 0 and 1; the runner learns it holds 2
	1039	+	// lines (it expected 1), and goes on at offset 2: no gap, no repeat.
	1040	+	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	+	}
	1043	+	if !strings.Contains(h.logs.String(), "continuing from") {
	1044	+	t.Fatalf("no warning about the resync:\n%s", h.logs.String())
	1045	+	}
	1046	+	}


@@ -11,6 +11,9 @@ import (
11	"strings"	11		"strings"
12	"testing"	12		"testing"
13	"time"	13		"time"
		14	+	"unicode/utf8"
		15	+
		16	+	"google.golang.org/protobuf/types/known/timestamppb"
14		17
15	"connectrpc.com/connect"	18		"connectrpc.com/connect"
16		19

@@ -847,3 +850,197 @@ func TestRunCarriesTheRemainingDeadline(t *testing.T) {
847	t.Fatalf("%v", spec)	850		t.Fatalf("%v", spec)
848	}	851		}
849	}	852		}
		853	+
		854	+	// ---- the guest is untrusted: step numbers and lines are checked (M3)
		855	+
		856	+	// hostile runs a job whose executor sends the frames of body, and checks that
		857	+	// the job fails, the sandbox is destroyed, the server never saw a violation of
		858	+	// its contract, and a line in the log names the reason.
		859	+	func hostile(t testing.T, wantReason string, body script) harness {
		860	+	t.Helper()
		861	+	h := newHarness(t, func(t testing.T, c execstream.Conn, run execpb.Run, cn <-chan execpb.Cancel) {
		862	+	body(t, c, run, cn)
		863	+	// A real executor would go on; this one waits for the host to cut it.
		864	+	<-cn
		865	+	})
		866	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
		867	+	h.run(t)
		868	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
		869	+	h.srv.mu.Lock()
		870	+	defer h.srv.mu.Unlock()
		871	+	c := h.srv.completes[0]
		872	+	if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE \|\| len(c.Outputs) != 0 {
		873	+	t.Fatalf("conclusion %v outputs %v", c.Conclusion, c.Outputs)
		874	+	}
		875	+	found := false
		876	+	for _, l := range h.srv.lines {
		877	+	if strings.Contains(l, "broke the protocol") && strings.Contains(l, wantReason) {
		878	+	found = true
		879	+	}
		880	+	}
		881	+	if !found {
		882	+	t.Fatalf("no error line mentioning %q in %q", wantReason, h.srv.lines)
		883	+	}
		884	+	for _, u := range c.Steps {
		885	+	if u.State != ermite.StepState_STEP_STATE_COMPLETED \|\| u.Number < 1 \|\| u.Number > 1000 {
		886	+	t.Errorf("invalid final step %v", u)
		887	+	}
		888	+	}
		889	+	if h.driver.stops != 1 {
		890	+	t.Errorf("the sandbox was not destroyed: stops=%d", h.driver.stops)
		891	+	}
		892	+	return h
		893	+	}
		894	+
		895	+	func TestGuestStepNumberZeroFailsTheJob(t *testing.T) {
		896	+	hostile(t, "outside 1..1000", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		897	+	stepStart(c, 1, "Set up job")
		898	+	stepStart(c, 0, "zero")
		899	+	})
		900	+	}
		901	+
		902	+	func TestGuestStepNumberOverTheLimitFailsTheJob(t *testing.T) {
		903	+	hostile(t, "outside 1..1000", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		904	+	stepStart(c, 1, "Set up job")
		905	+	stepStart(c, 1001, "far")
		906	+	})
		907	+	}
		908	+
		909	+	func TestGuestStepGoingBackFailsTheJob(t *testing.T) {
		910	+	hostile(t, "started after step 3", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		911	+	stepStart(c, 1, "a")
		912	+	stepStart(c, 3, "c")
		913	+	stepStart(c, 2, "b")
		914	+	})
		915	+	}
		916	+
		917	+	func TestGuestStepRestartedFailsTheJob(t *testing.T) {
		918	+	hostile(t, "started after step 1", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		919	+	stepStart(c, 1, "a")
		920	+	stepDone(c, 1, "a", cOK)
		921	+	stepStart(c, 1, "again")
		922	+	})
		923	+	}
		924	+
		925	+	func TestGuestCompletedStepIsNeverTouchedAgain(t *testing.T) {
		926	+	h := hostile(t, "completed twice", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		927	+	stepStart(c, 1, "a")
		928	+	stepDone(c, 1, "a", cOK)
		929	+	stepDone(c, 1, "a", cFail)
		930	+	})
		931	+	if u := h.srv.Result("j1").Steps[1]; u == nil \|\| u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS {
		932	+	t.Errorf("a completed step was changed: %v", u.Conclusion)
		933	+	}
		934	+	}
		935	+
		936	+	func TestGuestCompletingAnUnknownStepFailsTheJob(t *testing.T) {
		937	+	hostile(t, "without having started", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		938	+	stepStart(c, 1, "a")
		939	+	stepDone(c, 7, "ghost", cOK)
		940	+	})
		941	+	}
		942	+
		943	+	func TestGuestLogForAnUnstartedOrCompletedStepFailsTheJob(t *testing.T) {
		944	+	hostile(t, "has not started", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		945	+	stepStart(c, 1, "a")
		946	+	logLineFrame(c, 5, "x", false)
		947	+	})
		948	+	hostile(t, "which is completed", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		949	+	stepStart(c, 1, "a")
		950	+	stepDone(c, 1, "a", cOK)
		951	+	logLineFrame(c, 1, "late", false)
		952	+	})
		953	+	}
		954	+
		955	+	func TestGuestLogGoingBackAStepFailsTheJob(t *testing.T) {
		956	+	hostile(t, "after one for step 2", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		957	+	stepStart(c, 1, "a")
		958	+	stepStart(c, 2, "b")
		959	+	logLineFrame(c, 2, "x", false)
		960	+	logLineFrame(c, 1, "y", false)
		961	+	})
		962	+	}
		963	+
		964	+	func TestGuestOversizedLineFailsTheJob(t *testing.T) {
		965	+	hostile(t, "bytes (the limit is 65536)", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		966	+	stepStart(c, 1, "a")
		967	+	logLineFrame(c, 1, strings.Repeat("x", 64*1024+1), false)
		968	+	})
		969	+	}
		970	+
		971	+	// Text that the server would refuse is repaired, not forwarded: a line feed
		972	+	// makes more lines and long names are cut. (Invalid UTF-8 cannot reach the host:
		973	+	// proto3 strings are validated when the frame is decoded, which fails the
		974	+	// session; the executor replaces invalid bytes before it sends.)
		975	+	func TestGuestTextIsMadeValidForTheServer(t *testing.T) {
		976	+	h := newHarness(t, func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		977	+	stepStart(c, 1, strings.Repeat("é", 800)) // 1,600 bytes
		978	+	logLineFrame(c, 1, "two\nlines", false)
		979	+	stepDone(c, 1, "", cOK)
		980	+	jobDone(c, cOK, nil, false)
		981	+	})
		982	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
		983	+	h.run(t)
		984	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
		985	+	h.srv.mu.Lock()
		986	+	defer h.srv.mu.Unlock()
		987	+	if strings.Join(h.srv.lines, "\|") != "two\|lines" {
		988	+	t.Fatalf("%q", h.srv.lines)
		989	+	}
		990	+	if n := h.srv.steps[1].Name; len(n) > 1024 \|\| !utf8.ValidString(n) {
		991	+	t.Fatalf("name of %d bytes", len(n))
		992	+	}
		993	+	}
		994	+
		995	+	// A guest that sends steps without end cannot grow the memory of the host.
		996	+	func TestGuestFloodIsBounded(t *testing.T) {
		997	+	h := hostile(t, "outside 1..1000", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		998	+	for n := 1; n <= 1100; n++ {
		999	+	stepStart(c, n, "s")
		1000	+	stepDone(c, n, "s", cOK)
		1001	+	}
		1002	+	})
		1003	+	_ = h
		1004	+	}
		1005	+
		1006	+	// The server's line count is the truth (m6): when the answer to an AppendLog
		1007	+	// says the server holds more lines than the request accounts for, the next
		1008	+	// offset starts from the server's count and no gap or repeat is created.
		1009	+	func TestAppendLogResyncsToTheServersCount(t *testing.T) {
		1010	+	h := newHarness(t, func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		1011	+	stepStart(c, 1, "a")
		1012	+	logLineFrame(c, 1, "l1", false)
		1013	+	time.Sleep(150 * time.Millisecond) // a first request of its own
		1014	+	logLineFrame(c, 1, "l2", false)
		1015	+	logLineFrame(c, 1, "l3", false)
		1016	+	stepDone(c, 1, "a", cOK)
		1017	+	jobDone(c, cOK, nil, false)
		1018	+	})
		1019	+	h.runner.Timing.LogFlush = 30 * time.Millisecond
		1020	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
		1021	+	// A line the runner never sent appears in the server's log (a retry whose
		1022	+	// answer was lost, in a real server): the runner must continue after it.
		1023	+	h.srv.mu.Lock()
		1024	+	h.srv.appendHook = func(n int, r *ermite.AppendLogRequest) (bool, error) {
		1025	+	if n == 1 {
		1026	+	_, _ = h.srv.store.Apply(&ermite.AppendLogRequest{JobId: r.JobId, Step: 1, Offset: 0, Lines: []*ermite.LogLine{
		1027	+	{Time: timestamppb.Now(), Text: "planted"}, {Time: timestamppb.Now(), Text: "planted2"}}})
		1028	+	return false, nil
		1029	+	}
		1030	+	return false, nil
		1031	+	}
		1032	+	h.srv.mu.Unlock()
		1033	+	h.run(t)
		1034	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
		1035	+	h.srv.mu.Lock()
		1036	+	defer h.srv.mu.Unlock()
		1037	+	// The first request (offset 0, "l1") was skipped by the server because the
		1038	+	// planted lines already took offsets 0 and 1; the runner learns it holds 2
		1039	+	// lines (it expected 1), and goes on at offset 2: no gap, no repeat.
		1040	+	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	+	}
		1043	+	if !strings.Contains(h.logs.String(), "continuing from") {
		1044	+	t.Fatalf("no warning about the resync:\n%s", h.logs.String())
		1045	+	}
		1046	+	}