fix(host): bound result frames of the guest and copy what is cut

A step sends at most one summary and one list of annotations, while it runs:
a second one, or one for a completed step, fails the job as a protocol
violation. Summary bytes count in the bounded memory of the outbox with
backpressure, and the annotations waiting for their step are bounded too.
Strings the host cuts (annotation titles, step names) are copied so that they
do not keep the allocation of the frame the guest sent.

commit: 8a89149f7cb7017444c0bd356bc38fc3bf21f897
author: brice · 8d ago · 2026-10-14 02:44 UTC
parent: 3d4e205
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +193 -13

M internal/host/job.go +7 -1


@@ -74,6 +74,10 @@ type jobRun struct {
74	74		// executor sent for steps that have not completed yet (see results.go).
75	75		features jobFeatures
76	76		annLists map[int32][]*ermite.Annotation
	77	+	// annSeen and sumSeen are the steps that sent their annotations and their
	78	+	// summary: each is sent once.
	79	+	annSeen, sumSeen map[int32]bool
	80	+	annBytes int // bytes of annLists
77	81		// hard stops the sandbox at once; set before the session starts.
78	82		hard func()
79	83		result *execpb.JobResult

@@ -371,6 +375,8 @@ func (j jobRun) logLine(l execpb.Log) {
371	375		}
372	376
373	377		// cutName bounds a step name to what the server keeps, on a character boundary.
	378	+	// The result is a copy: a substring would keep the whole name the guest sent
	379	+	// alive (see cutRunes).
374	380		func cutName(s string) string {
375	381		if len(s) <= maxNameLen {
376	382		return strings.ToValidUTF8(s, "\ufffd")

@@ -379,7 +385,7 @@ func cutName(s string) string {
379	385		for !utf8.ValidString(s) && len(s) > 0 {
380	386		s = s[:len(s)-1]
381	387		}
382		-	return s
	388	+	return strings.Clone(s)
383	389		}
384	390
385	391		func cloneStep(u ermite.StepUpdate) ermite.StepUpdate {

 	// executor sent for steps that have not completed yet (see results.go).
 	features jobFeatures
 	annLists map[int32][]*ermite.Annotation
+	// annSeen and sumSeen are the steps that sent their annotations and their
+	// summary: each is sent once.
+	annSeen, sumSeen map[int32]bool
+	annBytes         int // bytes of annLists
 	// hard stops the sandbox at once; set before the session starts.
 	hard    func()
 	result  *execpb.JobResult
 }
 // cutName bounds a step name to what the server keeps, on a character boundary.
+// The result is a copy: a substring would keep the whole name the guest sent
+// alive (see cutRunes).
 func cutName(s string) string {
 	if len(s) <= maxNameLen {
 		return strings.ToValidUTF8(s, "\ufffd")
 	for !utf8.ValidString(s) && len(s) > 0 {
 		s = s[:len(s)-1]
 	}
-	return s
+	return strings.Clone(s)
 }
 func cloneStep(u *ermite.StepUpdate) *ermite.StepUpdate {

M internal/host/outbox.go +10 -1


@@ -157,14 +157,19 @@ func (o outbox) Step(s ermite.StepUpdate, anns map[int32][]*ermite.Annotation)
157	157		o.cond.Broadcast()
158	158		}
159	159
160		-	// Summary queues the summary of a step.
	160	+	// Summary queues the summary of a step. Its bytes count with the log text in
	161	+	// the bounded memory of the outbox: it blocks while too much is queued.
161	162		func (o *outbox) Summary(step int32, markdown string) {
162	163		o.mu.Lock()
	164	+	for o.queued > maxQueuedBytes && !o.dead && !o.closing {
	165	+	o.cond.Wait()
	166	+	}
163	167		if o.dead {
164	168		o.mu.Unlock()
165	169		return
166	170		}
167	171		o.events = append(o.events, event{summary: &summary{step, markdown}})
	172	+	o.queued += len(markdown) + 32
168	173		o.mu.Unlock()
169	174		o.cond.Broadcast()
170	175		}

@@ -238,6 +243,10 @@ func (o *outbox) next(ctx context.Context) []event {
238	243		first := o.events[0]
239	244		if first.step != nil \|\| first.summary != nil {
240	245		o.events = o.events[1:]
	246	+	if first.summary != nil {
	247	+	o.queued -= len(first.summary.markdown) + 32
	248	+	o.cond.Broadcast() // producers waiting for room
	249	+	}
241	250		return []event{first}
242	251		}
243	252		// A run of lines of the same step, within the limits.

 	o.cond.Broadcast()
 }
-// Summary queues the summary of a step.
+// Summary queues the summary of a step. Its bytes count with the log text in
+// the bounded memory of the outbox: it blocks while too much is queued.
 func (o *outbox) Summary(step int32, markdown string) {
 	o.mu.Lock()
+	for o.queued > maxQueuedBytes && !o.dead && !o.closing {
+		o.cond.Wait()
+	}
 	if o.dead {
 		o.mu.Unlock()
 		return
 	}
 	o.events = append(o.events, event{summary: &summary{step, markdown}})
+	o.queued += len(markdown) + 32
 	o.mu.Unlock()
 	o.cond.Broadcast()
 }
 	first := o.events[0]
 	if first.step != nil || first.summary != nil {
 		o.events = o.events[1:]
+		if first.summary != nil {
+			o.queued -= len(first.summary.markdown) + 32
+			o.cond.Broadcast() // producers waiting for room
+		}
 		return []event{first}
 	}
 	// A run of lines of the same step, within the limits.

M internal/host/results.go +52 -8


@@ -4,6 +4,7 @@ package host
4	4
5	5		import (
6	6		"context"
	7	+	"fmt"
7	8		"strings"
8	9		"unicode/utf8"
9	10

@@ -80,6 +81,9 @@ func levelOf(l execpb.AnnotationLevel) ermite.AnnotationLevel {
80	81		}
81	82
82	83		// cutRunes bounds s to n characters (code points), after making it valid UTF-8.
	84	+	// A string that is cut is copied: a substring would keep the whole allocation
	85	+	// of what the guest sent (up to a frame of 2 MiB) alive for as long as the
	86	+	// short string is, and a guest can open 1,000 steps.
83	87		func cutRunes(s string, n int) string {
84	88		s = strings.ToValidUTF8(s, "\ufffd")
85	89		if utf8.RuneCountInString(s) <= n {

@@ -88,7 +92,7 @@ func cutRunes(s string, n int) string {
88	92		i := 0
89	93		for k := range s {
90	94		if i == n {
91		-	return s[:k]
	95	+	return strings.Clone(s[:k])
92	96		}
93	97		i++
94	98		}

@@ -137,16 +141,52 @@ func (j jobRun) stepAnnotations(a execpb.StepAnnotations) {
137	141		if j.violation != "" \|\| !j.features.annotations {
138	142		return
139	143		}
140		-	if _, ok := j.steps[a.Step]; !ok {
	144	+	if why := j.resultFrameProblemLocked("annotations", a.Step, j.annSeen); why != "" {
141	145		j.mu.Unlock()
142		-	j.violate("annotations for step %d, which has not started", a.Step)
	146	+	j.violate("%s", why)
143	147		j.mu.Lock()
144	148		return
145	149		}
146	150		if j.annLists == nil {
147	151		j.annLists = map[int32][]*ermite.Annotation{}
148	152		}
149		-	j.annLists[a.Step] = annotationsFrom(a.Step, a.Annotations)
	153	+	j.annSeen[a.Step] = true
	154	+	list := annotationsFrom(a.Step, a.Annotations)
	155	+	// The lists wait in memory until their step completes and the call that
	156	+	// carries them is answered: bounded like the queue of the log.
	157	+	n := annotationBytes(list)
	158	+	if j.annBytes+n > maxQueuedBytes {
	159	+	j.log.Warn().Int32("step", a.Step).Int("bytes", n).Msg("too many annotations are waiting to be sent: this step's are dropped")
	160	+	return
	161	+	}
	162	+	j.annBytes += n
	163	+	j.annLists[a.Step] = list
	164	+	}
	165	+
	166	+	func annotationBytes(l []*ermite.Annotation) int {
	167	+	n := 0
	168	+	for _, a := range l {
	169	+	n += len(a.Title) + len(a.Message) + len(a.File) + 64
	170	+	}
	171	+	return n
	172	+	}
	173	+
	174	+	// resultFrameProblemLocked checks a frame that carries a result of a step
	175	+	// (annotations, summary). The executor sends each at most once per step, while
	176	+	// the step runs, just before it completes (the runner reads GITHUB_STEP_SUMMARY
	177	+	// once, after the step, and keeps the issues of a step in one list): anything
	178	+	// else is the guest misbehaving, and would let it queue unbounded data.
	179	+	func (j *jobRun) resultFrameProblemLocked(what string, step int32, seen map[int32]bool) string {
	180	+	u, ok := j.steps[step]
	181	+	switch {
	182	+	case !ok:
	183	+	return fmt.Sprintf("%s for step %d, which has not started", what, step)
	184	+	case u.State == ermite.StepState_STEP_STATE_COMPLETED:
	185	+	return fmt.Sprintf("%s for step %d, which is completed", what, step)
	186	+	case seen[step]:
	187	+	return fmt.Sprintf("%s for step %d, which already sent them", what, step)
	188	+	}
	189	+	return ""
150	190		}
151	191
152	192		// takeAnnotations returns the list recorded for step n, once.

@@ -156,22 +196,26 @@ func (j jobRun) takeAnnotations(n int32) map[int32][]ermite.Annotation {
156	196		return nil
157	197		}
158	198		delete(j.annLists, n)
	199	+	j.annBytes -= annotationBytes(l)
159	200		return map[int32][]*ermite.Annotation{n: l}
160	201		}
161	202
162	203		// stepSummary queues UploadSummary for a step that wrote GITHUB_STEP_SUMMARY.
	204	+	// The bytes count in the bounded memory of the outbox, like log lines: the call
	205	+	// blocks while too much is queued.
163	206		func (j jobRun) stepSummary(s execpb.StepSummary) {
164	207		j.mu.Lock()
165	208		if j.violation != "" \|\| !j.features.summaries {
166	209		j.mu.Unlock()
167	210		return
168	211		}
169		-	_, started := j.steps[s.Step]
170		-	j.mu.Unlock()
171		-	if !started {
172		-	j.violate("a summary for step %d, which has not started", s.Step)
	212	+	if why := j.resultFrameProblemLocked("a summary", s.Step, j.sumSeen); why != "" {
	213	+	j.mu.Unlock()
	214	+	j.violate("%s", why)
173	215		return
174	216		}
	217	+	j.sumSeen[s.Step] = true
	218	+	j.mu.Unlock()
175	219		md := strings.ToValidUTF8(s.Markdown, "\ufffd")
176	220		if len(md) > maxSummaryBytes {
177	221		j.log.Warn().Int32("step", s.Step).Int("bytes", len(md)).Msg("a step summary over 1 MiB is dropped")

 import (
 	"context"
+	"fmt"
 	"strings"
 	"unicode/utf8"
 }
 // cutRunes bounds s to n characters (code points), after making it valid UTF-8.
+// A string that is cut is copied: a substring would keep the whole allocation
+// of what the guest sent (up to a frame of 2 MiB) alive for as long as the
+// short string is, and a guest can open 1,000 steps.
 func cutRunes(s string, n int) string {
 	s = strings.ToValidUTF8(s, "\ufffd")
 	if utf8.RuneCountInString(s) <= n {
 	i := 0
 	for k := range s {
 		if i == n {
-			return s[:k]
+			return strings.Clone(s[:k])
 		}
 		i++
 	}
 	if j.violation != "" || !j.features.annotations {
 		return
 	}
-	if _, ok := j.steps[a.Step]; !ok {
+	if why := j.resultFrameProblemLocked("annotations", a.Step, j.annSeen); why != "" {
 		j.mu.Unlock()
-		j.violate("annotations for step %d, which has not started", a.Step)
+		j.violate("%s", why)
 		j.mu.Lock()
 		return
 	}
 	if j.annLists == nil {
 		j.annLists = map[int32][]*ermite.Annotation{}
 	}
-	j.annLists[a.Step] = annotationsFrom(a.Step, a.Annotations)
+	j.annSeen[a.Step] = true
+	list := annotationsFrom(a.Step, a.Annotations)
+	// The lists wait in memory until their step completes and the call that
+	// carries them is answered: bounded like the queue of the log.
+	n := annotationBytes(list)
+	if j.annBytes+n > maxQueuedBytes {
+		j.log.Warn().Int32("step", a.Step).Int("bytes", n).Msg("too many annotations are waiting to be sent: this step's are dropped")
+		return
+	}
+	j.annBytes += n
+	j.annLists[a.Step] = list
+}
+func annotationBytes(l []*ermite.Annotation) int {
+	n := 0
+	for _, a := range l {
+		n += len(a.Title) + len(a.Message) + len(a.File) + 64
+	}
+	return n
+}
+// resultFrameProblemLocked checks a frame that carries a result of a step
+// (annotations, summary). The executor sends each at most once per step, while
+// the step runs, just before it completes (the runner reads GITHUB_STEP_SUMMARY
+// once, after the step, and keeps the issues of a step in one list): anything
+// else is the guest misbehaving, and would let it queue unbounded data.
+func (j *jobRun) resultFrameProblemLocked(what string, step int32, seen map[int32]bool) string {
+	u, ok := j.steps[step]
+	switch {
+	case !ok:
+		return fmt.Sprintf("%s for step %d, which has not started", what, step)
+	case u.State == ermite.StepState_STEP_STATE_COMPLETED:
+		return fmt.Sprintf("%s for step %d, which is completed", what, step)
+	case seen[step]:
+		return fmt.Sprintf("%s for step %d, which already sent them", what, step)
+	}
+	return ""
 }
 // takeAnnotations returns the list recorded for step n, once.
 		return nil
 	}
 	delete(j.annLists, n)
+	j.annBytes -= annotationBytes(l)
 	return map[int32][]*ermite.Annotation{n: l}
 }
 // stepSummary queues UploadSummary for a step that wrote GITHUB_STEP_SUMMARY.
+// The bytes count in the bounded memory of the outbox, like log lines: the call
+// blocks while too much is queued.
 func (j *jobRun) stepSummary(s *execpb.StepSummary) {
 	j.mu.Lock()
 	if j.violation != "" || !j.features.summaries {
 		j.mu.Unlock()
 		return
 	}
-	_, started := j.steps[s.Step]
+	if why := j.resultFrameProblemLocked("a summary", s.Step, j.sumSeen); why != "" {
-	j.mu.Unlock()
+		j.mu.Unlock()
-	if !started {
+		j.violate("%s", why)
-		j.violate("a summary for step %d, which has not started", s.Step)
 		return
 	}
+	j.sumSeen[s.Step] = true
+	j.mu.Unlock()
 	md := strings.ToValidUTF8(s.Markdown, "\ufffd")
 	if len(md) > maxSummaryBytes {
 		j.log.Warn().Int32("step", s.Step).Int("bytes", len(md)).Msg("a step summary over 1 MiB is dropped")

M internal/host/results_test.go +123 -3


@@ -5,12 +5,15 @@ package host
5	5		import (
6	6		"context"
7	7		"errors"
	8	+	"io"
8	9		"strings"
9	10		"sync/atomic"
10	11		"testing"
11	12		"time"
	13	+	"unsafe"
12	14
13	15		"connectrpc.com/connect"
	16	+	"github.com/bnema/zerowrap"
14	17
15	18		ermite "bnema.dev/ermite-api/gen/ermite"
16	19

@@ -280,15 +283,128 @@ func TestSummaryOverOneMiBIsDropped(t *testing.T) {
280	283		sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
281	284		stepStart(c, 1, "Set up job")
282	285		sumFrame(c, 1, strings.Repeat("x", 1<<20+1))
283		-	sumFrame(c, 1, strings.Repeat("y", 1<<20))
284	286		stepDone(c, 1, "Set up job", cOK)
	287	+	stepStart(c, 2, "Run it")
	288	+	sumFrame(c, 2, strings.Repeat("y", 1<<20))
	289	+	stepDone(c, 2, "Run it", cOK)
285	290		jobDone(c, cOK, nil, false)
286	291		}
287	292		h := runResultsJob(t, sc, nil)
288	293		h.srv.mu.Lock()
289	294		defer h.srv.mu.Unlock()
290		-	if len(h.srv.summaries[1]) != 1<<20 \|\| h.srv.summaries[1][0] != 'y' {
291		-	t.Errorf("summary of %d bytes", len(h.srv.summaries[1]))
	295	+	if _, has := h.srv.summaries[1]; has {
	296	+	t.Error("a summary over 1 MiB was sent")
	297	+	}
	298	+	if len(h.srv.summaries[2]) != 1<<20 \|\| h.srv.summaries[2][0] != 'y' {
	299	+	t.Errorf("summary of %d bytes", len(h.srv.summaries[2]))
	300	+	}
	301	+	}
	302	+
	303	+	// One summary and one list of annotations per step, sent while the step runs
	304	+	// (GO-002): a second one, or one for a completed step, is the guest
	305	+	// misbehaving, and it would let it queue unbounded data.
	306	+	func TestGuestSecondSummaryFailsTheJob(t *testing.T) {
	307	+	hostile(t, "a summary for step 1, which already sent them", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	308	+	stepStart(c, 1, "Set up job")
	309	+	sumFrame(c, 1, "one")
	310	+	sumFrame(c, 1, "two")
	311	+	time.Sleep(50 * time.Millisecond)
	312	+	})
	313	+	}
	314	+
	315	+	func TestGuestSummaryForACompletedStepFailsTheJob(t *testing.T) {
	316	+	hostile(t, "a summary for step 1, which is completed", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	317	+	stepStart(c, 1, "Set up job")
	318	+	stepDone(c, 1, "Set up job", cOK)
	319	+	sumFrame(c, 1, "late")
	320	+	time.Sleep(50 * time.Millisecond)
	321	+	})
	322	+	}
	323	+
	324	+	func TestGuestSecondAnnotationListFailsTheJob(t *testing.T) {
	325	+	hostile(t, "annotations for step 1, which already sent them", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	326	+	stepStart(c, 1, "Set up job")
	327	+	annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "a"})
	328	+	annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "b"})
	329	+	time.Sleep(50 * time.Millisecond)
	330	+	})
	331	+	}
	332	+
	333	+	func TestGuestAnnotationsForACompletedStepFailTheJob(t *testing.T) {
	334	+	hostile(t, "annotations for step 1, which is completed", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	335	+	stepStart(c, 1, "Set up job")
	336	+	stepDone(c, 1, "Set up job", cOK)
	337	+	annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "late"})
	338	+	time.Sleep(50 * time.Millisecond)
	339	+	})
	340	+	}
	341	+
	342	+	// The summaries count in the memory the outbox bounds, and backpressure holds
	343	+	// the producer while the server does not answer (GO-002).
	344	+	func TestSummaryBytesAreBoundedByTheOutbox(t *testing.T) {
	345	+	release := make(chan struct{})
	346	+	cl := &blockingSummaryClient{release: release}
	347	+	o := newOutbox(cl, "j1", fastTiming(), func(error) {}, zerowrapNop())
	348	+	ctx, cancel := context.WithCancel(context.Background())
	349	+	defer cancel()
	350	+	o.start(ctx)
	351	+	defer close(release)
	352	+	big := strings.Repeat("x", 1<<20)
	353	+	go func() {
	354	+	for i := 1; i <= 40; i++ { // 40 MiB offered, 16 MiB allowed in memory
	355	+	o.Summary(int32(i), big)
	356	+	}
	357	+	}()
	358	+	time.Sleep(500 * time.Millisecond)
	359	+	o.mu.Lock()
	360	+	queued := o.queued
	361	+	o.mu.Unlock()
	362	+	if queued > maxQueuedBytes+len(big)+64 {
	363	+	t.Errorf("%d bytes queued, the bound is %d", queued, maxQueuedBytes)
	364	+	}
	365	+	if queued < maxQueuedBytes/2 {
	366	+	t.Errorf("only %d bytes queued: the summaries are not counted?", queued)
	367	+	}
	368	+	}
	369	+
	370	+	type blockingSummaryClient struct {
	371	+	stubClient
	372	+	release chan struct{}
	373	+	}
	374	+
	375	+	func (c blockingSummaryClient) UploadSummary(ctx context.Context, _ connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
	376	+	select {
	377	+	case <-c.release:
	378	+	case <-ctx.Done():
	379	+	return nil, ctx.Err()
	380	+	}
	381	+	return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
	382	+	}
	383	+
	384	+	// A string cut by the host must not keep the allocation of the frame the guest
	385	+	// sent it in (GO-003): the cut copy has its own memory.
	386	+	func TestCutStringsDoNotRetainTheGuestAllocation(t *testing.T) {
	387	+	big := strings.Repeat("a", 1<<20)
	388	+	t0 := cutRunes(big, 255)
	389	+	if len(t0) != 255 {
	390	+	t.Fatal(len(t0))
	391	+	}
	392	+	if unsafe.StringData(t0) == unsafe.StringData(big) {
	393	+	t.Error("cutRunes returns a substring of its argument")
	394	+	}
	395	+	name := cutName(big)
	396	+	if len(name) != maxNameLen \|\| unsafe.StringData(name) == unsafe.StringData(big) {
	397	+	t.Error("cutName returns a substring of its argument")
	398	+	}
	399	+	// an annotation that is kept whole (not cut) is not copied: its message is
	400	+	// within 4096 characters, a frame cannot make it retain more than that
	401	+	// through a substring because the host does not slice it.
	402	+	a := annotationsFrom(1, []*execpb.Annotation{{Level: lWarn, Message: "m", Title: big, File: strings.Repeat("f", 2000)}})
	403	+	if len(a) != 1 \|\| len(a[0].Title) != 255 \|\| unsafe.StringData(a[0].Title) == unsafe.StringData(big) {
	404	+	t.Errorf("title %d", len(a[0].Title))
	405	+	}
	406	+	if a[0].File != "" {
	407	+	t.Error("a file over 1,024 characters is dropped")
292	408		}
293	409		}
294	410

@@ -306,3 +422,7 @@ type discard struct{}
306	422
307	423		func (discard) Write(p []byte) (int, error) { return len(p), nil }
308	424		func (discard) Read([]byte) (int, error) { select {} }
	425	+
	426	+	func zerowrapNop() zerowrap.Logger {
	427	+	return zerowrap.New(zerowrap.Config{Level: "error", Format: "json", Output: io.Discard})
	428	+	}


@@ -5,12 +5,15 @@ package host
5		import (	5		import (
6		"context"	6		"context"
7		"errors"	7		"errors"
			8	+	"io"
8		"strings"	9		"strings"
9		"sync/atomic"	10		"sync/atomic"
10		"testing"	11		"testing"
11		"time"	12		"time"
			13	+	"unsafe"
12			14
13		"connectrpc.com/connect"	15		"connectrpc.com/connect"
			16	+	"github.com/bnema/zerowrap"
14			17
15		ermite "bnema.dev/ermite-api/gen/ermite"	18		ermite "bnema.dev/ermite-api/gen/ermite"
16			19

@@ -280,15 +283,128 @@ func TestSummaryOverOneMiBIsDropped(t *testing.T) {
280		sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {	283		sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
281		stepStart(c, 1, "Set up job")	284		stepStart(c, 1, "Set up job")
282		sumFrame(c, 1, strings.Repeat("x", 1<<20+1))	285		sumFrame(c, 1, strings.Repeat("x", 1<<20+1))
283	-	sumFrame(c, 1, strings.Repeat("y", 1<<20))
284		stepDone(c, 1, "Set up job", cOK)	286		stepDone(c, 1, "Set up job", cOK)
			287	+	stepStart(c, 2, "Run it")
			288	+	sumFrame(c, 2, strings.Repeat("y", 1<<20))
			289	+	stepDone(c, 2, "Run it", cOK)
285		jobDone(c, cOK, nil, false)	290		jobDone(c, cOK, nil, false)
286		}	291		}
287		h := runResultsJob(t, sc, nil)	292		h := runResultsJob(t, sc, nil)
288		h.srv.mu.Lock()	293		h.srv.mu.Lock()
289		defer h.srv.mu.Unlock()	294		defer h.srv.mu.Unlock()
290	-	if len(h.srv.summaries[1]) != 1<<20 \|\| h.srv.summaries[1][0] != 'y' {	295	+	if _, has := h.srv.summaries[1]; has {
291	-	t.Errorf("summary of %d bytes", len(h.srv.summaries[1]))	296	+	t.Error("a summary over 1 MiB was sent")
			297	+	}
			298	+	if len(h.srv.summaries[2]) != 1<<20 \|\| h.srv.summaries[2][0] != 'y' {
			299	+	t.Errorf("summary of %d bytes", len(h.srv.summaries[2]))
			300	+	}
			301	+	}
			302	+
			303	+	// One summary and one list of annotations per step, sent while the step runs
			304	+	// (GO-002): a second one, or one for a completed step, is the guest
			305	+	// misbehaving, and it would let it queue unbounded data.
			306	+	func TestGuestSecondSummaryFailsTheJob(t *testing.T) {
			307	+	hostile(t, "a summary for step 1, which already sent them", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
			308	+	stepStart(c, 1, "Set up job")
			309	+	sumFrame(c, 1, "one")
			310	+	sumFrame(c, 1, "two")
			311	+	time.Sleep(50 * time.Millisecond)
			312	+	})
			313	+	}
			314	+
			315	+	func TestGuestSummaryForACompletedStepFailsTheJob(t *testing.T) {
			316	+	hostile(t, "a summary for step 1, which is completed", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
			317	+	stepStart(c, 1, "Set up job")
			318	+	stepDone(c, 1, "Set up job", cOK)
			319	+	sumFrame(c, 1, "late")
			320	+	time.Sleep(50 * time.Millisecond)
			321	+	})
			322	+	}
			323	+
			324	+	func TestGuestSecondAnnotationListFailsTheJob(t *testing.T) {
			325	+	hostile(t, "annotations for step 1, which already sent them", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
			326	+	stepStart(c, 1, "Set up job")
			327	+	annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "a"})
			328	+	annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "b"})
			329	+	time.Sleep(50 * time.Millisecond)
			330	+	})
			331	+	}
			332	+
			333	+	func TestGuestAnnotationsForACompletedStepFailTheJob(t *testing.T) {
			334	+	hostile(t, "annotations for step 1, which is completed", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
			335	+	stepStart(c, 1, "Set up job")
			336	+	stepDone(c, 1, "Set up job", cOK)
			337	+	annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "late"})
			338	+	time.Sleep(50 * time.Millisecond)
			339	+	})
			340	+	}
			341	+
			342	+	// The summaries count in the memory the outbox bounds, and backpressure holds
			343	+	// the producer while the server does not answer (GO-002).
			344	+	func TestSummaryBytesAreBoundedByTheOutbox(t *testing.T) {
			345	+	release := make(chan struct{})
			346	+	cl := &blockingSummaryClient{release: release}
			347	+	o := newOutbox(cl, "j1", fastTiming(), func(error) {}, zerowrapNop())
			348	+	ctx, cancel := context.WithCancel(context.Background())
			349	+	defer cancel()
			350	+	o.start(ctx)
			351	+	defer close(release)
			352	+	big := strings.Repeat("x", 1<<20)
			353	+	go func() {
			354	+	for i := 1; i <= 40; i++ { // 40 MiB offered, 16 MiB allowed in memory
			355	+	o.Summary(int32(i), big)
			356	+	}
			357	+	}()
			358	+	time.Sleep(500 * time.Millisecond)
			359	+	o.mu.Lock()
			360	+	queued := o.queued
			361	+	o.mu.Unlock()
			362	+	if queued > maxQueuedBytes+len(big)+64 {
			363	+	t.Errorf("%d bytes queued, the bound is %d", queued, maxQueuedBytes)
			364	+	}
			365	+	if queued < maxQueuedBytes/2 {
			366	+	t.Errorf("only %d bytes queued: the summaries are not counted?", queued)
			367	+	}
			368	+	}
			369	+
			370	+	type blockingSummaryClient struct {
			371	+	stubClient
			372	+	release chan struct{}
			373	+	}
			374	+
			375	+	func (c blockingSummaryClient) UploadSummary(ctx context.Context, _ connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
			376	+	select {
			377	+	case <-c.release:
			378	+	case <-ctx.Done():
			379	+	return nil, ctx.Err()
			380	+	}
			381	+	return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
			382	+	}
			383	+
			384	+	// A string cut by the host must not keep the allocation of the frame the guest
			385	+	// sent it in (GO-003): the cut copy has its own memory.
			386	+	func TestCutStringsDoNotRetainTheGuestAllocation(t *testing.T) {
			387	+	big := strings.Repeat("a", 1<<20)
			388	+	t0 := cutRunes(big, 255)
			389	+	if len(t0) != 255 {
			390	+	t.Fatal(len(t0))
			391	+	}
			392	+	if unsafe.StringData(t0) == unsafe.StringData(big) {
			393	+	t.Error("cutRunes returns a substring of its argument")
			394	+	}
			395	+	name := cutName(big)
			396	+	if len(name) != maxNameLen \|\| unsafe.StringData(name) == unsafe.StringData(big) {
			397	+	t.Error("cutName returns a substring of its argument")
			398	+	}
			399	+	// an annotation that is kept whole (not cut) is not copied: its message is
			400	+	// within 4096 characters, a frame cannot make it retain more than that
			401	+	// through a substring because the host does not slice it.
			402	+	a := annotationsFrom(1, []*execpb.Annotation{{Level: lWarn, Message: "m", Title: big, File: strings.Repeat("f", 2000)}})
			403	+	if len(a) != 1 \|\| len(a[0].Title) != 255 \|\| unsafe.StringData(a[0].Title) == unsafe.StringData(big) {
			404	+	t.Errorf("title %d", len(a[0].Title))
			405	+	}
			406	+	if a[0].File != "" {
			407	+	t.Error("a file over 1,024 characters is dropped")
292		}	408		}
293		}	409		}
294			410

@@ -306,3 +422,7 @@ type discard struct{}
306			422
307		func (discard) Write(p []byte) (int, error) { return len(p), nil }	423		func (discard) Write(p []byte) (int, error) { return len(p), nil }
308		func (discard) Read([]byte) (int, error) { select {} }	424		func (discard) Read([]byte) (int, error) { select {} }
			425	+
			426	+	func zerowrapNop() zerowrap.Logger {
			427	+	return zerowrap.New(zerowrap.Config{Level: "error", Format: "json", Output: io.Discard})
			428	+	}

M internal/host/runner.go +1 -0


@@ -186,6 +186,7 @@ func (r Runner) startJob(ctx context.Context, log zerowrap.Logger, job ermite.
186	186		r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
187	187		log: zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},
188	188		receivedAt: r.clock(), cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},
	189	+	annSeen: map[int32]bool{}, sumSeen: map[int32]bool{},
189	190		}
190	191		jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")
191	192		r.active.Add(1)


@@ -186,6 +186,7 @@ func (r Runner) startJob(ctx context.Context, log zerowrap.Logger, job ermite.
186	r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,	186		r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
187	log: zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},	187		log: zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},
188	receivedAt: r.clock(), cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},	188		receivedAt: r.clock(), cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},
		189	+	annSeen: map[int32]bool{}, sumSeen: map[int32]bool{},
189	}	190		}
190	jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")	191		jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")
191	r.active.Add(1)	192		r.active.Add(1)