feat(host): tell the user when annotations were dropped for lack of room

The host counts the annotations it drops over the shared budget (a coalesced
counter, no content) and, after the outbox is closed and before CompleteJob,
appends one warning line to the job log through the final-report path, with a
slice of the report budget of its own. The text is ermite's own: the official
runner has no reporting backlog (GO-009).

commit: 41a0ba6bf883fb87802e50d54393398f997f5b7a
author: brice · 8d ago · 2026-10-14 03:09 UTC
parent: 349efbd
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +119 -1

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


@@ -77,6 +77,7 @@ 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	+	annDropped int // annotations dropped for lack of room, see reportDroppedAnnotations
80	81		// hard stops the sandbox at once; set before the session starts.
81	82		hard func()
82	83		result *execpb.JobResult

@@ -582,6 +583,7 @@ func (j *jobRun) report(ctx context.Context, out outcome, shutdown <-chan struct
582	583		if err := j.abandoned(); err != nil {
583	584		return
584	585		}
	586	+	j.reportDroppedAnnotations(rctx, budget/4)
585	587		j.complete(rctx, out.conclusion, out.outputs, out.shutdown)
586	588		}
587	589

@@ -835,13 +837,18 @@ func (j jobRun) sendComplete(ctx context.Context, req ermite.CompleteJobReques
835	837
836	838		// errorLog appends a line to the log after the outbox is closed.
837	839		func (j *jobRun) errorLog(ctx context.Context, msg string) {
	840	+	j.appendLogLine(ctx, "##[error]"+msg)
	841	+	}
	842	+
	843	+	// appendLogLine appends one line, in the last step, after the outbox is closed.
	844	+	func (j *jobRun) appendLogLine(ctx context.Context, text string) {
838	845		j.mu.Lock()
839	846		var last int32 = 1
840	847		for n := range j.steps {
841	848		last = max(last, n)
842	849		}
843	850		j.mu.Unlock()
844		-	j.outbox.AppendNow(ctx, logLine{step: last, unixNano: time.Now().UnixNano(), text: "##[error]" + msg})
	851	+	j.outbox.AppendNow(ctx, logLine{step: last, unixNano: time.Now().UnixNano(), text: text})
845	852		}
846	853
847	854		// rebuiltSteps returns every known step as a valid, completed snapshot: number

 	// annSeen and sumSeen are the steps that sent their annotations and their
 	// summary: each is sent once.
 	annSeen, sumSeen map[int32]bool
+	annDropped       int // annotations dropped for lack of room, see reportDroppedAnnotations
 	// hard stops the sandbox at once; set before the session starts.
 	hard    func()
 	result  *execpb.JobResult
 	if err := j.abandoned(); err != nil {
 		return
 	}
+	j.reportDroppedAnnotations(rctx, budget/4)
 	j.complete(rctx, out.conclusion, out.outputs, out.shutdown)
 }
 // errorLog appends a line to the log after the outbox is closed.
 func (j *jobRun) errorLog(ctx context.Context, msg string) {
+	j.appendLogLine(ctx, "##[error]"+msg)
+}
+// appendLogLine appends one line, in the last step, after the outbox is closed.
+func (j *jobRun) appendLogLine(ctx context.Context, text string) {
 	j.mu.Lock()
 	var last int32 = 1
 	for n := range j.steps {
 		last = max(last, n)
 	}
 	j.mu.Unlock()
-	j.outbox.AppendNow(ctx, logLine{step: last, unixNano: time.Now().UnixNano(), text: "##[error]" + msg})
+	j.outbox.AppendNow(ctx, logLine{step: last, unixNano: time.Now().UnixNano(), text: text})
 }
 // rebuiltSteps returns every known step as a valid, completed snapshot: number

M internal/host/results.go +34 -0


@@ -6,6 +6,7 @@ import (
6	6		"context"
7	7		"fmt"
8	8		"strings"
	9	+	"time"
9	10		"unicode/utf8"
10	11
11	12		"connectrpc.com/connect"

@@ -160,6 +161,8 @@ func (j jobRun) stepAnnotations(a execpb.StepAnnotations) {
160	161		n := annotationBytes(list)
161	162		if !j.outbox.ReserveAnnotations(n) {
162	163		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")
	164	+	// Only a count is kept (j.mu is held): the final report tells the user.
	165	+	j.annDropped += len(list)
163	166		return
164	167		}
165	168		j.annLists[a.Step] = list

@@ -254,3 +257,34 @@ func (j jobRun) allPendingAnnotations() []ermite.Annotation {
254	257		j.mu.Unlock()
255	258		return flattenAnnotations(m)
256	259		}
	260	+
	261	+	// droppedAnnotationsLine is the warning the user gets when annotations were
	262	+	// dropped because the backlog of reports to the server was full. The text is
	263	+	// ermite's own (the official runner has no such backlog); it carries no
	264	+	// annotation content, only the count, which the host coalesces over the whole
	265	+	// job. Empty when nothing was dropped.
	266	+	func droppedAnnotationsLine(n int) string {
	267	+	switch {
	268	+	case n <= 0:
	269	+	return ""
	270	+	case n == 1:
	271	+	return "##[warning]1 annotation was not reported because the reporting backlog was full."
	272	+	}
	273	+	return fmt.Sprintf("##[warning]%d annotations were not reported because the reporting backlog was full.", n)
	274	+	}
	275	+
	276	+	// reportDroppedAnnotations appends the one warning line to the job log, after
	277	+	// the outbox is closed (the queue is full, or gone: it is not used) and before
	278	+	// CompleteJob, with a slice of the report budget of its own so that a server
	279	+	// that keeps failing cannot take the time CompleteJob needs.
	280	+	func (j *jobRun) reportDroppedAnnotations(ctx context.Context, budget time.Duration) {
	281	+	j.mu.Lock()
	282	+	line := droppedAnnotationsLine(j.annDropped)
	283	+	j.mu.Unlock()
	284	+	if line == "" {
	285	+	return
	286	+	}
	287	+	actx, cancel := context.WithTimeout(ctx, budget)
	288	+	defer cancel()
	289	+	j.appendLogLine(actx, line)
	290	+	}


@@ -6,6 +6,7 @@ import (
6	"context"	6		"context"
7	"fmt"	7		"fmt"
8	"strings"	8		"strings"
		9	+	"time"
9	"unicode/utf8"	10		"unicode/utf8"
10		11
11	"connectrpc.com/connect"	12		"connectrpc.com/connect"

@@ -160,6 +161,8 @@ func (j jobRun) stepAnnotations(a execpb.StepAnnotations) {
160	n := annotationBytes(list)	161		n := annotationBytes(list)
161	if !j.outbox.ReserveAnnotations(n) {	162		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")	163		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")
		164	+	// Only a count is kept (j.mu is held): the final report tells the user.
		165	+	j.annDropped += len(list)
163	return	166		return
164	}	167		}
165	j.annLists[a.Step] = list	168		j.annLists[a.Step] = list

@@ -254,3 +257,34 @@ func (j jobRun) allPendingAnnotations() []ermite.Annotation {
254	j.mu.Unlock()	257		j.mu.Unlock()
255	return flattenAnnotations(m)	258		return flattenAnnotations(m)
256	}	259		}
		260	+
		261	+	// droppedAnnotationsLine is the warning the user gets when annotations were
		262	+	// dropped because the backlog of reports to the server was full. The text is
		263	+	// ermite's own (the official runner has no such backlog); it carries no
		264	+	// annotation content, only the count, which the host coalesces over the whole
		265	+	// job. Empty when nothing was dropped.
		266	+	func droppedAnnotationsLine(n int) string {
		267	+	switch {
		268	+	case n <= 0:
		269	+	return ""
		270	+	case n == 1:
		271	+	return "##[warning]1 annotation was not reported because the reporting backlog was full."
		272	+	}
		273	+	return fmt.Sprintf("##[warning]%d annotations were not reported because the reporting backlog was full.", n)
		274	+	}
		275	+
		276	+	// reportDroppedAnnotations appends the one warning line to the job log, after
		277	+	// the outbox is closed (the queue is full, or gone: it is not used) and before
		278	+	// CompleteJob, with a slice of the report budget of its own so that a server
		279	+	// that keeps failing cannot take the time CompleteJob needs.
		280	+	func (j *jobRun) reportDroppedAnnotations(ctx context.Context, budget time.Duration) {
		281	+	j.mu.Lock()
		282	+	line := droppedAnnotationsLine(j.annDropped)
		283	+	j.mu.Unlock()
		284	+	if line == "" {
		285	+	return
		286	+	}
		287	+	actx, cancel := context.WithTimeout(ctx, budget)
		288	+	defer cancel()
		289	+	j.appendLogLine(actx, line)
		290	+	}

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


@@ -5,6 +5,7 @@ package host
5	5		import (
6	6		"context"
7	7		"errors"
	8	+	"fmt"
8	9		"io"
9	10		"strings"
10	11		"sync/atomic"

@@ -557,3 +558,79 @@ func TestAnnotationBudgetIsGivenBack(t *testing.T) {
557	558		return o.queued == 0
558	559		})
559	560		}
	561	+
	562	+	// GO-009: annotations dropped for lack of room are told to the user with one
	563	+	// warning line in the job log, appended after the outbox is closed, without
	564	+	// annotation content.
	565	+	func TestDroppedAnnotationsAreReportedInTheLog(t *testing.T) {
	566	+	var list []*execpb.Annotation
	567	+	for _, l := range []execpb.AnnotationLevel{lWarn, lFail, execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE} {
	568	+	for range 10 {
	569	+	list = append(list, &execpb.Annotation{Level: l, Message: "SECRET-CONTENT" + strings.Repeat("m", 4080), Title: strings.Repeat("t", 255)})
	570	+	}
	571	+	}
	572	+	const steps = 200 // about 135 KiB each: 27 MB offered, 16 MiB of room
	573	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	574	+	for n := 1; n <= steps; n++ {
	575	+	stepStart(c, n, "step")
	576	+	annFrame(c, n, list...)
	577	+	stepDone(c, n, "step", cOK)
	578	+	}
	579	+	jobDone(c, cOK, nil, false)
	580	+	}
	581	+	h := newHarness(t, sc)
	582	+	h.runner.Timing.CompleteDeadline = 3 * time.Second
	583	+	// the server is blocked: every UpdateJob fails transiently, for ever
	584	+	h.srv.updateHook = func(int, *ermite.UpdateJobRequest) error { return connect.NewError(connect.CodeUnavailable, nil) }
	585	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	586	+	stop := h.run(t)
	587	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	588	+	if err := stop(); err != nil {
	589	+	t.Fatal(err)
	590	+	}
	591	+	h.srv.mu.Lock()
	592	+	defer h.srv.mu.Unlock()
	593	+	var found []string
	594	+	for _, l := range h.srv.lines {
	595	+	if strings.Contains(l, "were not reported because the reporting backlog was full.") {
	596	+	found = append(found, l)
	597	+	}
	598	+	if strings.Contains(l, "SECRET-CONTENT") {
	599	+	t.Error("annotation content reached the job log")
	600	+	}
	601	+	}
	602	+	if len(found) != 1 {
	603	+	t.Fatalf("%d lines about dropped annotations: %q", len(found), found)
	604	+	}
	605	+	var n int
	606	+	if _, err := fmt.Sscanf(found[0], "##[warning]%d annotations were not reported because the reporting backlog was full.", &n); err != nil \|\| n < 30 \|\| n > steps*30 \|\| n%30 != 0 {
	607	+	t.Errorf("%q (n=%d, %v)", found[0], n, err)
	608	+	}
	609	+	if h.srv.lines[len(h.srv.lines)-1] != found[0] {
	610	+	t.Error("the line is not the last of the log")
	611	+	}
	612	+	}
	613	+
	614	+	func TestDroppedAnnotationsLine(t *testing.T) {
	615	+	for n, want := range map[int]string{
	616	+	0: "", -1: "",
	617	+	1: "##[warning]1 annotation was not reported because the reporting backlog was full.",
	618	+	30: "##[warning]30 annotations were not reported because the reporting backlog was full.",
	619	+	} {
	620	+	if got := droppedAnnotationsLine(n); got != want {
	621	+	t.Errorf("%d: %q", n, got)
	622	+	}
	623	+	}
	624	+	}
	625	+
	626	+	// A job that drops nothing writes no such line.
	627	+	func TestNoDroppedAnnotationsNoLine(t *testing.T) {
	628	+	h := runResultsJob(t, resultsScript(nil), nil)
	629	+	h.srv.mu.Lock()
	630	+	defer h.srv.mu.Unlock()
	631	+	for _, l := range h.srv.lines {
	632	+	if strings.Contains(l, "reporting backlog") {
	633	+	t.Errorf("%q", l)
	634	+	}
	635	+	}
	636	+	}


@@ -5,6 +5,7 @@ package host
5	import (	5		import (
6	"context"	6		"context"
7	"errors"	7		"errors"
		8	+	"fmt"
8	"io"	9		"io"
9	"strings"	10		"strings"
10	"sync/atomic"	11		"sync/atomic"

@@ -557,3 +558,79 @@ func TestAnnotationBudgetIsGivenBack(t *testing.T) {
557	return o.queued == 0	558		return o.queued == 0
558	})	559		})
559	}	560		}
		561	+
		562	+	// GO-009: annotations dropped for lack of room are told to the user with one
		563	+	// warning line in the job log, appended after the outbox is closed, without
		564	+	// annotation content.
		565	+	func TestDroppedAnnotationsAreReportedInTheLog(t *testing.T) {
		566	+	var list []*execpb.Annotation
		567	+	for _, l := range []execpb.AnnotationLevel{lWarn, lFail, execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE} {
		568	+	for range 10 {
		569	+	list = append(list, &execpb.Annotation{Level: l, Message: "SECRET-CONTENT" + strings.Repeat("m", 4080), Title: strings.Repeat("t", 255)})
		570	+	}
		571	+	}
		572	+	const steps = 200 // about 135 KiB each: 27 MB offered, 16 MiB of room
		573	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
		574	+	for n := 1; n <= steps; n++ {
		575	+	stepStart(c, n, "step")
		576	+	annFrame(c, n, list...)
		577	+	stepDone(c, n, "step", cOK)
		578	+	}
		579	+	jobDone(c, cOK, nil, false)
		580	+	}
		581	+	h := newHarness(t, sc)
		582	+	h.runner.Timing.CompleteDeadline = 3 * time.Second
		583	+	// the server is blocked: every UpdateJob fails transiently, for ever
		584	+	h.srv.updateHook = func(int, *ermite.UpdateJobRequest) error { return connect.NewError(connect.CodeUnavailable, nil) }
		585	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
		586	+	stop := h.run(t)
		587	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
		588	+	if err := stop(); err != nil {
		589	+	t.Fatal(err)
		590	+	}
		591	+	h.srv.mu.Lock()
		592	+	defer h.srv.mu.Unlock()
		593	+	var found []string
		594	+	for _, l := range h.srv.lines {
		595	+	if strings.Contains(l, "were not reported because the reporting backlog was full.") {
		596	+	found = append(found, l)
		597	+	}
		598	+	if strings.Contains(l, "SECRET-CONTENT") {
		599	+	t.Error("annotation content reached the job log")
		600	+	}
		601	+	}
		602	+	if len(found) != 1 {
		603	+	t.Fatalf("%d lines about dropped annotations: %q", len(found), found)
		604	+	}
		605	+	var n int
		606	+	if _, err := fmt.Sscanf(found[0], "##[warning]%d annotations were not reported because the reporting backlog was full.", &n); err != nil \|\| n < 30 \|\| n > steps*30 \|\| n%30 != 0 {
		607	+	t.Errorf("%q (n=%d, %v)", found[0], n, err)
		608	+	}
		609	+	if h.srv.lines[len(h.srv.lines)-1] != found[0] {
		610	+	t.Error("the line is not the last of the log")
		611	+	}
		612	+	}
		613	+
		614	+	func TestDroppedAnnotationsLine(t *testing.T) {
		615	+	for n, want := range map[int]string{
		616	+	0: "", -1: "",
		617	+	1: "##[warning]1 annotation was not reported because the reporting backlog was full.",
		618	+	30: "##[warning]30 annotations were not reported because the reporting backlog was full.",
		619	+	} {
		620	+	if got := droppedAnnotationsLine(n); got != want {
		621	+	t.Errorf("%d: %q", n, got)
		622	+	}
		623	+	}
		624	+	}
		625	+
		626	+	// A job that drops nothing writes no such line.
		627	+	func TestNoDroppedAnnotationsNoLine(t *testing.T) {
		628	+	h := runResultsJob(t, resultsScript(nil), nil)
		629	+	h.srv.mu.Lock()
		630	+	defer h.srv.mu.Unlock()
		631	+	for _, l := range h.srv.lines {
		632	+	if strings.Contains(l, "reporting backlog") {
		633	+	t.Errorf("%q", l)
		634	+	}
		635	+	}
		636	+	}