fix(host): count annotations in the outbox budget for their whole lifetime, release producers on the hard stop

Annotations count in the one byte budget of the outbox from the moment the host
holds a list until the call that carries it is answered (they left the budget
when their step completed, while the events and pending lists kept growing).
Over the budget a list is dropped, with a log line.

The producers that wait for room (log lines, summaries) are released by the
hard stop of the job and drop what does not fit, so a server that keeps
failing the uploads cannot keep the reader of the sandbox stream stuck past the
deadline. kill() ends the sender's context, and the final flush gets half of
the report budget so that CompleteJob still has the other half (GO-006, GO-007).

commit: 349efbd4d904acca0999de3c43657700f05678fd
author: brice · 8d ago · 2026-10-14 03:01 UTC
parent: c19a481
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +269 -40

M internal/host/fake_test.go +18 -12


@@ -44,18 +44,19 @@ type fakeServer struct {
44	44		beatHook func(n int) error
45	45		registers []*ermite.RegisterRunnerRequest
46	46
47		-	store *hosttest.LogStore
48		-	violations []string
49		-	lines []string // log text, one per line (mirror of store)
50		-	cont []bool
51		-	steps map[int32]*ermite.StepUpdate
52		-	annots map[int32][]*ermite.Annotation // by step, replaced by each call that names the step
53		-	summaries map[int32]string
54		-	updates []*ermite.UpdateJobRequest
55		-	updateHook func(n int, req *ermite.UpdateJobRequest) error
56		-	appendReq []*ermite.AppendLogRequest
57		-	appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)
58		-	appendN int
	47	+	store *hosttest.LogStore
	48	+	violations []string
	49	+	lines []string // log text, one per line (mirror of store)
	50	+	cont []bool
	51	+	steps map[int32]*ermite.StepUpdate
	52	+	annots map[int32][]*ermite.Annotation // by step, replaced by each call that names the step
	53	+	summaries map[int32]string
	54	+	updates []*ermite.UpdateJobRequest
	55	+	updateHook func(n int, req *ermite.UpdateJobRequest) error
	56	+	summaryHook func(req *ermite.UploadSummaryRequest) error
	57	+	appendReq []*ermite.AppendLogRequest
	58	+	appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)
	59	+	appendN int
59	60
60	61		completes []*ermite.CompleteJobRequest
61	62		completeHook func(n int) (applyThenFail bool, err error)

@@ -208,6 +209,11 @@ func (s fakeServer) UploadSummary(_ context.Context, r connect.Request[ermite.
208	209		if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
209	210		return nil, err
210	211		}
	212	+	if s.summaryHook != nil {
	213	+	if err := s.summaryHook(r.Msg); err != nil {
	214	+	return nil, err
	215	+	}
	216	+	}
211	217		if s.summaries == nil {
212	218		s.summaries = map[int32]string{}
213	219		}


@@ -44,18 +44,19 @@ type fakeServer struct {
44		beatHook func(n int) error	44		beatHook func(n int) error
45		registers []*ermite.RegisterRunnerRequest	45		registers []*ermite.RegisterRunnerRequest
46			46
47	-	store *hosttest.LogStore	47	+	store *hosttest.LogStore
48	-	violations []string	48	+	violations []string
49	-	lines []string // log text, one per line (mirror of store)	49	+	lines []string // log text, one per line (mirror of store)
50	-	cont []bool	50	+	cont []bool
51	-	steps map[int32]*ermite.StepUpdate	51	+	steps map[int32]*ermite.StepUpdate
52	-	annots map[int32][]*ermite.Annotation // by step, replaced by each call that names the step	52	+	annots map[int32][]*ermite.Annotation // by step, replaced by each call that names the step
53	-	summaries map[int32]string	53	+	summaries map[int32]string
54	-	updates []*ermite.UpdateJobRequest	54	+	updates []*ermite.UpdateJobRequest
55	-	updateHook func(n int, req *ermite.UpdateJobRequest) error	55	+	updateHook func(n int, req *ermite.UpdateJobRequest) error
56	-	appendReq []*ermite.AppendLogRequest	56	+	summaryHook func(req *ermite.UploadSummaryRequest) error
57	-	appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)	57	+	appendReq []*ermite.AppendLogRequest
58	-	appendN int	58	+	appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)
			59	+	appendN int
59			60
60		completes []*ermite.CompleteJobRequest	61		completes []*ermite.CompleteJobRequest
61		completeHook func(n int) (applyThenFail bool, err error)	62		completeHook func(n int) (applyThenFail bool, err error)

@@ -208,6 +209,11 @@ func (s fakeServer) UploadSummary(_ context.Context, r connect.Request[ermite.
208		if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {	209		if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
209		return nil, err	210		return nil, err
210		}	211		}
			212	+	if s.summaryHook != nil {
			213	+	if err := s.summaryHook(r.Msg); err != nil {
			214	+	return nil, err
			215	+	}
			216	+	}
211		if s.summaries == nil {	217		if s.summaries == nil {
212		s.summaries = map[int32]string{}	218		s.summaries = map[int32]string{}
213		}	219		}

M internal/host/job.go +15 -2


@@ -77,7 +77,6 @@ type jobRun struct {
77	77		// annSeen and sumSeen are the steps that sent their annotations and their
78	78		// summary: each is sent once.
79	79		annSeen, sumSeen map[int32]bool
80		-	annBytes int // bytes of annLists
81	80		// hard stops the sandbox at once; set before the session starts.
82	81		hard func()
83	82		result *execpb.JobResult

@@ -458,6 +457,15 @@ func (j jobRun) run(rt Runner, shutdown <-chan struct{}) {
458	457		j.outbox.start(ctx)
459	458
460	459		stopHard := j.startStopTimers(ctx, shutdown)
	460	+	// The hard stop frees the reader of the stream if it waits for room in the
	461	+	// outbox (the server keeps failing): nothing else would wake it.
	462	+	go func() {
	463	+	select {
	464	+	case <-stopHard:
	465	+	j.outbox.releaseProducers()
	466	+	case <-ctx.Done():
	467	+	}
	468	+	}()
461	469		out := j.execute(ctx, rt, stopHard)
462	470		j.report(ctx, out, shutdown)
463	471		}

@@ -565,7 +573,12 @@ func (j *jobRun) report(ctx context.Context, out outcome, shutdown <-chan struct
565	573		j.sandboxError(msg)
566	574		out = failure()
567	575		}
568		-	j.outbox.Close(rctx)
	576	+	// The flush gets half of the budget: when the server keeps failing the
	577	+	// uploads, the other half is still there for CompleteJob, which is what
	578	+	// ends the job for the server.
	579	+	fctx, fcancel := context.WithTimeout(rctx, budget/2)
	580	+	j.outbox.Close(fctx)
	581	+	fcancel()
569	582		if err := j.abandoned(); err != nil {
570	583		return
571	584		}

 	// 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
 	j.outbox.start(ctx)
 	stopHard := j.startStopTimers(ctx, shutdown)
+	// The hard stop frees the reader of the stream if it waits for room in the
+	// outbox (the server keeps failing): nothing else would wake it.
+	go func() {
+		select {
+		case <-stopHard:
+			j.outbox.releaseProducers()
+		case <-ctx.Done():
+		}
+	}()
 	out := j.execute(ctx, rt, stopHard)
 	j.report(ctx, out, shutdown)
 }
 		j.sandboxError(msg)
 		out = failure()
 	}
-	j.outbox.Close(rctx)
+	// The flush gets half of the budget: when the server keeps failing the
+	// uploads, the other half is still there for CompleteJob, which is what
+	// ends the job for the server.
+	fctx, fcancel := context.WithTimeout(rctx, budget/2)
+	j.outbox.Close(fctx)
+	fcancel()
 	if err := j.abandoned(); err != nil {
 		return
 	}

M internal/host/outbox.go +89 -20


@@ -65,13 +65,25 @@ type outbox struct {
65	65		onEnded func(error)
66	66		log zerowrap.Logger
67	67
68		-	mu sync.Mutex
69		-	cond *sync.Cond
70		-	events []event
71		-	queued int // bytes of log text in events
	68	+	mu sync.Mutex
	69	+	cond *sync.Cond
	70	+	events []event
	71	+	// queued is the memory the outbox holds for the server, in bytes: log text
	72	+	// and summaries in events, and annotations from the moment the host
	73	+	// received them until the call that carries them is answered. One budget,
	74	+	// maxQueuedBytes, for all of it.
	75	+	queued int
72	76		closing bool
73	77		dead bool
74		-	done chan struct{}
	78	+	// released is set by the hard stop of the job (releaseProducers): the
	79	+	// producers that wait for room stop waiting and drop what they hold, and
	80	+	// what is offered over the budget is dropped instead of waited for.
	81	+	released bool
	82	+	// stopSender ends the context the sender works with (start): kill calls
	83	+	// it, so that a call or a retry delay in progress ends with the outbox and
	84	+	// Close can wait for the sender after giving up.
	85	+	stopSender context.CancelFunc
	86	+	done chan struct{}
75	87
76	88		// pendingAnns are the annotations the server may not hold yet, by step: the
77	89		// last list queued for the step, until an UpdateJob that carried it

@@ -100,9 +112,14 @@ func newOutbox(c callClient, jobID string, t Timing, onEnded func(error), log ze
100	112
101	113		// start runs the sender until Close or until ctx ends.
102	114		func (o *outbox) start(ctx context.Context) {
	115	+	sctx, stop := context.WithCancel(ctx)
	116	+	o.mu.Lock()
	117	+	o.stopSender = stop
	118	+	o.mu.Unlock()
103	119		go func() {
104	120		defer close(o.done)
105		-	o.loop(ctx)
	121	+	defer stop()
	122	+	o.loop(sctx)
106	123		}()
107	124		// A cancelled context must wake the loop and the producers.
108	125		go func() {

@@ -119,17 +136,44 @@ func (o *outbox) kill() {
119	136		o.dead = true
120	137		o.events = nil
121	138		o.queued = 0
	139	+	stop := o.stopSender
122	140		o.mu.Unlock()
	141	+	if stop != nil {
	142	+	stop()
	143	+	}
123	144		o.cond.Broadcast()
124	145		}
125	146
126		-	// Log queues a line of the job log. It blocks while too much is queued.
127		-	func (o *outbox) Log(l logLine) {
128		-	o.mu.Lock()
129		-	for o.queued > maxQueuedBytes && !o.dead && !o.closing {
	147	+	// waitRoomLocked blocks while too much is queued. It returns false when the
	148	+	// item must be dropped: the outbox is dead, or the hard stop released the
	149	+	// producers and there is still no room. Closing lets the last items in.
	150	+	func (o *outbox) waitRoomLocked() bool {
	151	+	for o.queued > maxQueuedBytes && !o.dead && !o.closing && !o.released {
130	152		o.cond.Wait()
131	153		}
132	154		if o.dead {
	155	+	return false
	156	+	}
	157	+	return o.queued <= maxQueuedBytes \|\| o.closing
	158	+	}
	159	+
	160	+	// releaseProducers wakes the producers that wait for room and makes them (and
	161	+	// the ones that come) drop their items when there is none: called by the hard
	162	+	// stop of the job, because the sender may never make room (the server keeps
	163	+	// failing) and the producer is the reader of the stream of the sandbox. What
	164	+	// is queued is still sent by Close, within its budget.
	165	+	func (o *outbox) releaseProducers() {
	166	+	o.mu.Lock()
	167	+	o.released = true
	168	+	o.mu.Unlock()
	169	+	o.cond.Broadcast()
	170	+	}
	171	+
	172	+	// Log queues a line of the job log. It blocks while too much is queued, until
	173	+	// the hard stop of the job releases it.
	174	+	func (o *outbox) Log(l logLine) {
	175	+	o.mu.Lock()
	176	+	if !o.waitRoomLocked() {
133	177		o.mu.Unlock()
134	178		return
135	179		}

@@ -139,6 +183,32 @@ func (o *outbox) Log(l logLine) {
139	183		o.cond.Broadcast()
140	184		}
141	185
	186	+	// ReserveAnnotations counts the bytes of a list of annotations in the budget,
	187	+	// from the moment the host holds it. It returns false, and counts nothing,
	188	+	// when there is no room: the caller drops the list (annotations are not worth
	189	+	// stopping the reader of the stream for: it also carries the frames that
	190	+	// complete the steps and free the room).
	191	+	func (o *outbox) ReserveAnnotations(n int) bool {
	192	+	o.mu.Lock()
	193	+	defer o.mu.Unlock()
	194	+	if o.dead \|\| o.queued+n > maxQueuedBytes {
	195	+	return false
	196	+	}
	197	+	o.queued += n
	198	+	return true
	199	+	}
	200	+
	201	+	// releaseAnnotations gives the bytes back once the call that carried the
	202	+	// lists was answered (or refused).
	203	+	func (o *outbox) releaseAnnotations(n int) {
	204	+	o.mu.Lock()
	205	+	if !o.dead {
	206	+	o.queued -= n
	207	+	}
	208	+	o.mu.Unlock()
	209	+	o.cond.Broadcast()
	210	+	}
	211	+
142	212		// Step queues a step snapshot, with the annotations that changed (may be nil).
143	213		func (o outbox) Step(s ermite.StepUpdate, anns map[int32][]*ermite.Annotation) {
144	214		o.mu.Lock()

@@ -158,13 +228,11 @@ func (o outbox) Step(s ermite.StepUpdate, anns map[int32][]*ermite.Annotation)
158	228		}
159	229
160	230		// 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.
	231	+	// the bounded memory of the outbox: it blocks while too much is queued, until
	232	+	// the hard stop of the job releases it.
162	233		func (o *outbox) Summary(step int32, markdown string) {
163	234		o.mu.Lock()
164		-	for o.queued > maxQueuedBytes && !o.dead && !o.closing {
165		-	o.cond.Wait()
166		-	}
167		-	if o.dead {
	235	+	if !o.waitRoomLocked() {
168	236		o.mu.Unlock()
169	237		return
170	238		}

@@ -327,14 +395,14 @@ func (o outbox) sendStep(ctx context.Context, s ermite.StepUpdate, anns map[in
327	395		return err
328	396		})
329	397		if err == nil {
330		-	// The server holds these lists now, unless a later one replaced them.
	398	+	// The server holds these lists now: they leave the pending ones and
	399	+	// the budget.
331	400		o.mu.Lock()
332		-	for n, l := range anns {
333		-	if cur := o.pendingAnns[n]; len(cur) == len(l) && (len(l) == 0 \|\| &cur[0] == &l[0]) {
334		-	delete(o.pendingAnns, n)
335		-	}
	401	+	for n := range anns {
	402	+	delete(o.pendingAnns, n)
336	403		}
337	404		o.mu.Unlock()
	405	+	o.releaseAnnotations(annotationsBytes(anns))
338	406		return
339	407		}
340	408		if ctx.Err() == nil && classify(err) == kindPermanent {

@@ -347,6 +415,7 @@ func (o outbox) sendStep(ctx context.Context, s ermite.StepUpdate, anns map[in
347	415		delete(o.pendingAnns, n)
348	416		}
349	417		o.mu.Unlock()
	418	+	o.releaseAnnotations(annotationsBytes(anns))
350	419		if err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
351	420		_, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))
352	421		return err

 	onEnded func(error)
 	log     zerowrap.Logger
-	mu      sync.Mutex
+	mu     sync.Mutex
-	cond    *sync.Cond
+	cond   *sync.Cond
-	events  []event
+	events []event
-	queued  int // bytes of log text in events
+	// queued is the memory the outbox holds for the server, in bytes: log text
+	// and summaries in events, and annotations from the moment the host
+	// received them until the call that carries them is answered. One budget,
+	// maxQueuedBytes, for all of it.
+	queued  int
 	closing bool
 	dead    bool
-	done    chan struct{}
+	// released is set by the hard stop of the job (releaseProducers): the
+	// producers that wait for room stop waiting and drop what they hold, and
+	// what is offered over the budget is dropped instead of waited for.
+	released bool
+	// stopSender ends the context the sender works with (start): kill calls
+	// it, so that a call or a retry delay in progress ends with the outbox and
+	// Close can wait for the sender after giving up.
+	stopSender context.CancelFunc
+	done       chan struct{}
 	// pendingAnns are the annotations the server may not hold yet, by step: the
 	// last list queued for the step, until an UpdateJob that carried it
 // start runs the sender until Close or until ctx ends.
 func (o *outbox) start(ctx context.Context) {
+	sctx, stop := context.WithCancel(ctx)
+	o.mu.Lock()
+	o.stopSender = stop
+	o.mu.Unlock()
 	go func() {
 		defer close(o.done)
-		o.loop(ctx)
+		defer stop()
+		o.loop(sctx)
 	}()
 	// A cancelled context must wake the loop and the producers.
 	go func() {
 	o.dead = true
 	o.events = nil
 	o.queued = 0
+	stop := o.stopSender
 	o.mu.Unlock()
+	if stop != nil {
+		stop()
+	}
 	o.cond.Broadcast()
 }
-// Log queues a line of the job log. It blocks while too much is queued.
+// waitRoomLocked blocks while too much is queued. It returns false when the
-func (o *outbox) Log(l logLine) {
+// item must be dropped: the outbox is dead, or the hard stop released the
-	o.mu.Lock()
+// producers and there is still no room. Closing lets the last items in.
-	for o.queued > maxQueuedBytes && !o.dead && !o.closing {
+func (o *outbox) waitRoomLocked() bool {
+	for o.queued > maxQueuedBytes && !o.dead && !o.closing && !o.released {
 		o.cond.Wait()
 	}
 	if o.dead {
+		return false
+	}
+	return o.queued <= maxQueuedBytes || o.closing
+}
+// releaseProducers wakes the producers that wait for room and makes them (and
+// the ones that come) drop their items when there is none: called by the hard
+// stop of the job, because the sender may never make room (the server keeps
+// failing) and the producer is the reader of the stream of the sandbox. What
+// is queued is still sent by Close, within its budget.
+func (o *outbox) releaseProducers() {
+	o.mu.Lock()
+	o.released = true
+	o.mu.Unlock()
+	o.cond.Broadcast()
+}
+// Log queues a line of the job log. It blocks while too much is queued, until
+// the hard stop of the job releases it.
+func (o *outbox) Log(l logLine) {
+	o.mu.Lock()
+	if !o.waitRoomLocked() {
 		o.mu.Unlock()
 		return
 	}
 	o.cond.Broadcast()
 }
+// ReserveAnnotations counts the bytes of a list of annotations in the budget,
+// from the moment the host holds it. It returns false, and counts nothing,
+// when there is no room: the caller drops the list (annotations are not worth
+// stopping the reader of the stream for: it also carries the frames that
+// complete the steps and free the room).
+func (o *outbox) ReserveAnnotations(n int) bool {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if o.dead || o.queued+n > maxQueuedBytes {
+		return false
+	}
+	o.queued += n
+	return true
+}
+// releaseAnnotations gives the bytes back once the call that carried the
+// lists was answered (or refused).
+func (o *outbox) releaseAnnotations(n int) {
+	o.mu.Lock()
+	if !o.dead {
+		o.queued -= n
+	}
+	o.mu.Unlock()
+	o.cond.Broadcast()
+}
 // Step queues a step snapshot, with the annotations that changed (may be nil).
 func (o *outbox) Step(s *ermite.StepUpdate, anns map[int32][]*ermite.Annotation) {
 	o.mu.Lock()
 }
 // 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.
+// the bounded memory of the outbox: it blocks while too much is queued, until
+// the hard stop of the job releases it.
 func (o *outbox) Summary(step int32, markdown string) {
 	o.mu.Lock()
-	for o.queued > maxQueuedBytes && !o.dead && !o.closing {
+	if !o.waitRoomLocked() {
-		o.cond.Wait()
-	}
-	if o.dead {
 		o.mu.Unlock()
 		return
 	}
 		return err
 	})
 	if err == nil {
-		// The server holds these lists now, unless a later one replaced them.
+		// The server holds these lists now: they leave the pending ones and
+		// the budget.
 		o.mu.Lock()
-		for n, l := range anns {
+		for n := range anns {
-			if cur := o.pendingAnns[n]; len(cur) == len(l) && (len(l) == 0 || &cur[0] == &l[0]) {
+			delete(o.pendingAnns, n)
-				delete(o.pendingAnns, n)
-			}
 		}
 		o.mu.Unlock()
+		o.releaseAnnotations(annotationsBytes(anns))
 		return
 	}
 	if ctx.Err() == nil && classify(err) == kindPermanent {
 				delete(o.pendingAnns, n)
 			}
 			o.mu.Unlock()
+			o.releaseAnnotations(annotationsBytes(anns))
 			if err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
 				_, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))
 				return err

M internal/host/results.go +16 -6


@@ -152,17 +152,28 @@ func (j jobRun) stepAnnotations(a execpb.StepAnnotations) {
152	152		}
153	153		j.annSeen[a.Step] = true
154	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.
	155	+	// The list stays in memory until the call that carries it is answered,
	156	+	// first here while its step runs, then in the outbox: all of it counts in
	157	+	// the one budget of the outbox, from now (ReserveAnnotations) to the answer
	158	+	// (sendStep). When there is no room (the server has been failing for long)
	159	+	// the list is dropped, and the log says so.
157	160		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")
	161	+	if !j.outbox.ReserveAnnotations(n) {
	162	+	j.log.Warn().Int32("step", a.Step).Int("bytes", n).Msg("too much is waiting to be sent to the server: the annotations of this step are dropped")
160	163		return
161	164		}
162		-	j.annBytes += n
163	165		j.annLists[a.Step] = list
164	166		}
165	167
	168	+	// annotationsBytes is the size of the lists of a call.
	169	+	func annotationsBytes(m map[int32][]*ermite.Annotation) int {
	170	+	n := 0
	171	+	for _, l := range m {
	172	+	n += annotationBytes(l)
	173	+	}
	174	+	return n
	175	+	}
	176	+
166	177		func annotationBytes(l []*ermite.Annotation) int {
167	178		n := 0
168	179		for _, a := range l {

@@ -196,7 +207,6 @@ func (j jobRun) takeAnnotations(n int32) map[int32][]ermite.Annotation {
196	207		return nil
197	208		}
198	209		delete(j.annLists, n)
199		-	j.annBytes -= annotationBytes(l)
200	210		return map[int32][]*ermite.Annotation{n: l}
201	211		}
202	212

 	}
 	j.annSeen[a.Step] = true
 	list := annotationsFrom(a.Step, a.Annotations)
-	// The lists wait in memory until their step completes and the call that
+	// The list stays in memory until the call that carries it is answered,
-	// carries them is answered: bounded like the queue of the log.
+	// first here while its step runs, then in the outbox: all of it counts in
+	// the one budget of the outbox, from now (ReserveAnnotations) to the answer
+	// (sendStep). When there is no room (the server has been failing for long)
+	// the list is dropped, and the log says so.
 	n := annotationBytes(list)
-	if j.annBytes+n > maxQueuedBytes {
+	if !j.outbox.ReserveAnnotations(n) {
-		j.log.Warn().Int32("step", a.Step).Int("bytes", n).Msg("too many annotations are waiting to be sent: this step's are dropped")
+		j.log.Warn().Int32("step", a.Step).Int("bytes", n).Msg("too much is waiting to be sent to the server: the annotations of this step are dropped")
 		return
 	}
-	j.annBytes += n
 	j.annLists[a.Step] = list
 }
+// annotationsBytes is the size of the lists of a call.
+func annotationsBytes(m map[int32][]*ermite.Annotation) int {
+	n := 0
+	for _, l := range m {
+		n += annotationBytes(l)
+	}
+	return n
+}
 func annotationBytes(l []*ermite.Annotation) int {
 	n := 0
 	for _, a := range l {
 		return nil
 	}
 	delete(j.annLists, n)
-	j.annBytes -= annotationBytes(l)
 	return map[int32][]*ermite.Annotation{n: l}
 }

M internal/host/results_test.go +131 -0


@@ -426,3 +426,134 @@ func (discard) Read([]byte) (int, error) { select {} }
426	426		func zerowrapNop() zerowrap.Logger {
427	427		return zerowrap.New(zerowrap.Config{Level: "error", Format: "json", Output: io.Discard})
428	428		}
	429	+
	430	+	// GO-006: the reader of the stream of the sandbox waits in the outbox for room
	431	+	// when the summaries fill the budget and the server keeps failing; the hard
	432	+	// stop of the job must wake it, or the job outlives its deadline and the
	433	+	// shutdown waits for ever. Heartbeats work, every upload fails transiently,
	434	+	// summaries of distinct steps fill the queue, the runner is stopped.
	435	+	func TestHardStopReleasesAProducerWaitingForRoom(t *testing.T) {
	436	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, cancels <-chan execpb.Cancel) {
	437	+	big := strings.Repeat("s", 1<<20)
	438	+	for i := 1; i <= 30; i++ { // 30 MiB of summaries, 16 MiB of room
	439	+	stepStart(c, i, "step")
	440	+	sumFrame(c, i, big)
	441	+	}
	442	+	// the executor ignores the cancellation, as a stuck one would
	443	+	<-cancels
	444	+	time.Sleep(time.Hour)
	445	+	}
	446	+	h := newHarness(t, sc)
	447	+	h.runner.Timing.ShutdownGrace = 500 * time.Millisecond
	448	+	h.runner.Timing.ShutdownReport = time.Second
	449	+	h.runner.Timing.StopDeadline = 5 * time.Second
	450	+	h.srv.summaryHook = func(*ermite.UploadSummaryRequest) error { return connect.NewError(connect.CodeUnavailable, nil) }
	451	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	452	+	stop := h.run(t)
	453	+	eventually(t, "the queue to fill", func() bool {
	454	+	h.srv.mu.Lock()
	455	+	defer h.srv.mu.Unlock()
	456	+	return len(h.srv.beats) > 2
	457	+	})
	458	+	time.Sleep(300 * time.Millisecond) // the reader is stuck in Summary by now
	459	+	start := time.Now()
	460	+	done := make(chan error, 1)
	461	+	go func() { done <- stop() }()
	462	+	select {
	463	+	case <-done:
	464	+	case <-time.After(15 * time.Second):
	465	+	t.Fatal("the job did not end after the hard stop: the stream reader is stuck waiting for room")
	466	+	}
	467	+	if took := time.Since(start); took > 10*time.Second {
	468	+	t.Errorf("shutdown took %v", took)
	469	+	}
	470	+	if h.srv.completeCount() != 1 {
	471	+	t.Errorf("the job was not reported: %d CompleteJob", h.srv.completeCount())
	472	+	}
	473	+	}
	474	+
	475	+	// GO-007: annotations count in the budget of the outbox from the moment the
	476	+	// host holds them until the call that carries them is answered. With the
	477	+	// server not answering and 400 steps that each ended with the largest list the
	478	+	// contract allows, the memory held stays within the budget and the lists over
	479	+	// it are dropped, with a log line.
	480	+	func TestAnnotationsWaitingForTheServerAreBounded(t *testing.T) {
	481	+	cl := &blockingUpdateClient{release: make(chan struct{})}
	482	+	defer close(cl.release)
	483	+	logs := &logBuf{}
	484	+	tm := fastTiming()
	485	+	ctx, cancel := context.WithCancel(context.Background())
	486	+	defer cancel()
	487	+	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},
	488	+	annSeen: map[int32]bool{}, sumSeen: map[int32]bool{}, features: jobFeatures{annotations: true},
	489	+	cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: logs})}
	490	+	j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)
	491	+	j.outbox.start(ctx)
	492	+	var list []*execpb.Annotation
	493	+	for _, l := range []execpb.AnnotationLevel{lWarn, lFail, execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE} {
	494	+	for range 10 {
	495	+	list = append(list, &execpb.Annotation{Level: l, Message: strings.Repeat("m", 4096), Title: strings.Repeat("t", 255)})
	496	+	}
	497	+	}
	498	+	for n := int32(1); n <= 400; n++ { // 400 x about 135 KiB offered, 16 MiB of room
	499	+	j.stepStarted(&execpb.StepStarted{Number: n, Name: "s"})
	500	+	j.stepAnnotations(&execpb.StepAnnotations{Step: n, Annotations: list})
	501	+	j.stepCompleted(&execpb.StepCompleted{Number: n, Name: "s", Conclusion: cOK})
	502	+	}
	503	+	j.outbox.mu.Lock()
	504	+	held := annotationsBytes(j.outbox.pendingAnns)
	505	+	queued := j.outbox.queued
	506	+	j.outbox.mu.Unlock()
	507	+	if held > maxQueuedBytes {
	508	+	t.Errorf("%d bytes of annotations wait for the server, the budget is %d", held, maxQueuedBytes)
	509	+	}
	510	+	if queued > maxQueuedBytes {
	511	+	t.Errorf("%d bytes counted in the outbox, the budget is %d", queued, maxQueuedBytes)
	512	+	}
	513	+	if held < maxQueuedBytes/2 {
	514	+	t.Errorf("only %d bytes held: the annotations are dropped too early", held)
	515	+	}
	516	+	if !strings.Contains(logs.String(), "the annotations of this step are dropped") {
	517	+	t.Error("dropping annotations is not logged")
	518	+	}
	519	+	}
	520	+
	521	+	type blockingUpdateClient struct {
	522	+	stubClient
	523	+	release chan struct{}
	524	+	}
	525	+
	526	+	func (c blockingUpdateClient) UpdateJob(ctx context.Context, _ connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
	527	+	select {
	528	+	case <-c.release:
	529	+	return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
	530	+	case <-ctx.Done():
	531	+	return nil, ctx.Err()
	532	+	}
	533	+	}
	534	+
	535	+	// Once the call that carried a list is answered, its bytes leave the budget.
	536	+	func TestAnnotationBudgetIsGivenBack(t *testing.T) {
	537	+	var ob *outbox
	538	+	h := runResultsJob(t, resultsScript(nil), nil)
	539	+	_ = ob
	540	+	if got := h.srv.annots[2]; len(got) != 2 {
	541	+	t.Fatalf("annotations %v", got)
	542	+	}
	543	+	cl := &stubClient{}
	544	+	o := newOutbox(cl, "j1", fastTiming(), func(error) {}, zerowrapNop())
	545	+	ctx, cancel := context.WithCancel(context.Background())
	546	+	defer cancel()
	547	+	o.start(ctx)
	548	+	list := []*ermite.Annotation{{Level: ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING, Message: "m", StepNumber: 2}}
	549	+	n := annotationBytes(list)
	550	+	if !o.ReserveAnnotations(n) {
	551	+	t.Fatal("no room")
	552	+	}
	553	+	o.Step(&ermite.StepUpdate{Number: 2}, map[int32][]*ermite.Annotation{2: list})
	554	+	eventually(t, "the call to be answered", func() bool {
	555	+	o.mu.Lock()
	556	+	defer o.mu.Unlock()
	557	+	return o.queued == 0
	558	+	})
	559	+	}


@@ -426,3 +426,134 @@ func (discard) Read([]byte) (int, error) { select {} }
426	func zerowrapNop() zerowrap.Logger {	426		func zerowrapNop() zerowrap.Logger {
427	return zerowrap.New(zerowrap.Config{Level: "error", Format: "json", Output: io.Discard})	427		return zerowrap.New(zerowrap.Config{Level: "error", Format: "json", Output: io.Discard})
428	}	428		}
		429	+
		430	+	// GO-006: the reader of the stream of the sandbox waits in the outbox for room
		431	+	// when the summaries fill the budget and the server keeps failing; the hard
		432	+	// stop of the job must wake it, or the job outlives its deadline and the
		433	+	// shutdown waits for ever. Heartbeats work, every upload fails transiently,
		434	+	// summaries of distinct steps fill the queue, the runner is stopped.
		435	+	func TestHardStopReleasesAProducerWaitingForRoom(t *testing.T) {
		436	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, cancels <-chan execpb.Cancel) {
		437	+	big := strings.Repeat("s", 1<<20)
		438	+	for i := 1; i <= 30; i++ { // 30 MiB of summaries, 16 MiB of room
		439	+	stepStart(c, i, "step")
		440	+	sumFrame(c, i, big)
		441	+	}
		442	+	// the executor ignores the cancellation, as a stuck one would
		443	+	<-cancels
		444	+	time.Sleep(time.Hour)
		445	+	}
		446	+	h := newHarness(t, sc)
		447	+	h.runner.Timing.ShutdownGrace = 500 * time.Millisecond
		448	+	h.runner.Timing.ShutdownReport = time.Second
		449	+	h.runner.Timing.StopDeadline = 5 * time.Second
		450	+	h.srv.summaryHook = func(*ermite.UploadSummaryRequest) error { return connect.NewError(connect.CodeUnavailable, nil) }
		451	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
		452	+	stop := h.run(t)
		453	+	eventually(t, "the queue to fill", func() bool {
		454	+	h.srv.mu.Lock()
		455	+	defer h.srv.mu.Unlock()
		456	+	return len(h.srv.beats) > 2
		457	+	})
		458	+	time.Sleep(300 * time.Millisecond) // the reader is stuck in Summary by now
		459	+	start := time.Now()
		460	+	done := make(chan error, 1)
		461	+	go func() { done <- stop() }()
		462	+	select {
		463	+	case <-done:
		464	+	case <-time.After(15 * time.Second):
		465	+	t.Fatal("the job did not end after the hard stop: the stream reader is stuck waiting for room")
		466	+	}
		467	+	if took := time.Since(start); took > 10*time.Second {
		468	+	t.Errorf("shutdown took %v", took)
		469	+	}
		470	+	if h.srv.completeCount() != 1 {
		471	+	t.Errorf("the job was not reported: %d CompleteJob", h.srv.completeCount())
		472	+	}
		473	+	}
		474	+
		475	+	// GO-007: annotations count in the budget of the outbox from the moment the
		476	+	// host holds them until the call that carries them is answered. With the
		477	+	// server not answering and 400 steps that each ended with the largest list the
		478	+	// contract allows, the memory held stays within the budget and the lists over
		479	+	// it are dropped, with a log line.
		480	+	func TestAnnotationsWaitingForTheServerAreBounded(t *testing.T) {
		481	+	cl := &blockingUpdateClient{release: make(chan struct{})}
		482	+	defer close(cl.release)
		483	+	logs := &logBuf{}
		484	+	tm := fastTiming()
		485	+	ctx, cancel := context.WithCancel(context.Background())
		486	+	defer cancel()
		487	+	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},
		488	+	annSeen: map[int32]bool{}, sumSeen: map[int32]bool{}, features: jobFeatures{annotations: true},
		489	+	cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: logs})}
		490	+	j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)
		491	+	j.outbox.start(ctx)
		492	+	var list []*execpb.Annotation
		493	+	for _, l := range []execpb.AnnotationLevel{lWarn, lFail, execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE} {
		494	+	for range 10 {
		495	+	list = append(list, &execpb.Annotation{Level: l, Message: strings.Repeat("m", 4096), Title: strings.Repeat("t", 255)})
		496	+	}
		497	+	}
		498	+	for n := int32(1); n <= 400; n++ { // 400 x about 135 KiB offered, 16 MiB of room
		499	+	j.stepStarted(&execpb.StepStarted{Number: n, Name: "s"})
		500	+	j.stepAnnotations(&execpb.StepAnnotations{Step: n, Annotations: list})
		501	+	j.stepCompleted(&execpb.StepCompleted{Number: n, Name: "s", Conclusion: cOK})
		502	+	}
		503	+	j.outbox.mu.Lock()
		504	+	held := annotationsBytes(j.outbox.pendingAnns)
		505	+	queued := j.outbox.queued
		506	+	j.outbox.mu.Unlock()
		507	+	if held > maxQueuedBytes {
		508	+	t.Errorf("%d bytes of annotations wait for the server, the budget is %d", held, maxQueuedBytes)
		509	+	}
		510	+	if queued > maxQueuedBytes {
		511	+	t.Errorf("%d bytes counted in the outbox, the budget is %d", queued, maxQueuedBytes)
		512	+	}
		513	+	if held < maxQueuedBytes/2 {
		514	+	t.Errorf("only %d bytes held: the annotations are dropped too early", held)
		515	+	}
		516	+	if !strings.Contains(logs.String(), "the annotations of this step are dropped") {
		517	+	t.Error("dropping annotations is not logged")
		518	+	}
		519	+	}
		520	+
		521	+	type blockingUpdateClient struct {
		522	+	stubClient
		523	+	release chan struct{}
		524	+	}
		525	+
		526	+	func (c blockingUpdateClient) UpdateJob(ctx context.Context, _ connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
		527	+	select {
		528	+	case <-c.release:
		529	+	return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
		530	+	case <-ctx.Done():
		531	+	return nil, ctx.Err()
		532	+	}
		533	+	}
		534	+
		535	+	// Once the call that carried a list is answered, its bytes leave the budget.
		536	+	func TestAnnotationBudgetIsGivenBack(t *testing.T) {
		537	+	var ob *outbox
		538	+	h := runResultsJob(t, resultsScript(nil), nil)
		539	+	_ = ob
		540	+	if got := h.srv.annots[2]; len(got) != 2 {
		541	+	t.Fatalf("annotations %v", got)
		542	+	}
		543	+	cl := &stubClient{}
		544	+	o := newOutbox(cl, "j1", fastTiming(), func(error) {}, zerowrapNop())
		545	+	ctx, cancel := context.WithCancel(context.Background())
		546	+	defer cancel()
		547	+	o.start(ctx)
		548	+	list := []*ermite.Annotation{{Level: ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING, Message: "m", StepNumber: 2}}
		549	+	n := annotationBytes(list)
		550	+	if !o.ReserveAnnotations(n) {
		551	+	t.Fatal("no room")
		552	+	}
		553	+	o.Step(&ermite.StepUpdate{Number: 2}, map[int32][]*ermite.Annotation{2: list})
		554	+	eventually(t, "the call to be answered", func() bool {
		555	+	o.mu.Lock()
		556	+	defer o.mu.Unlock()
		557	+	return o.queued == 0
		558	+	})
		559	+	}