1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "io"
10 "strings"
11 "sync/atomic"
12 "testing"
13 "time"
14 "unsafe"
15
16 "connectrpc.com/connect"
17 "github.com/bnema/zerowrap"
18
19 ermite "bnema.dev/ermite-api/gen/ermite"
20
21 "bnema.dev/ermite-runner/internal/execstream"
22 "bnema.dev/ermite-runner/internal/execstream/execpb"
23)
24
25func annFrame(c *execstream.Conn, step int, as ...*execpb.Annotation) {
26 send(c, &execpb.Frame{Body: &execpb.Frame_StepAnnotations{StepAnnotations: &execpb.StepAnnotations{Step: int32(step), Annotations: as}}})
27}
28
29func sumFrame(c *execstream.Conn, step int, md string) {
30 send(c, &execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: int32(step), Markdown: md}}})
31}
32
33func stepDoneOutcome(c *execstream.Conn, n int, name string, concl, outcome execpb.StepConclusion) {
34 send(c, &execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: concl, Outcome: outcome, UnixNano: time.Now().UnixNano()}}})
35}
36
37const (
38 lWarn = execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING
39 lFail = execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
40)
41
42// resultsScript runs a job whose step 2 has annotations, a summary and an
43// outcome that differs from its conclusion (continue-on-error).
44func resultsScript(extra func(c *execstream.Conn)) script {
45 return func(_ *testing.T, c *execstream.Conn, run *execpb.Run, _ <-chan *execpb.Cancel) {
46 seenRun.Store(run)
47 stepStart(c, 1, "Set up job")
48 stepDone(c, 1, "Set up job", cOK)
49 stepStart(c, 2, "Run it")
50 logLineFrame(c, 2, "hi", false)
51 annFrame(c, 2,
52 &execpb.Annotation{Level: lWarn, Title: "T", Message: "careful", File: "a.go", StartLine: 3, EndLine: 4},
53 &execpb.Annotation{Level: lFail, Message: "broken"})
54 sumFrame(c, 2, "# Summary\n")
55 stepDoneOutcome(c, 2, "Run it", cOK, cFail)
56 if extra != nil {
57 extra(c)
58 }
59 stepStart(c, 3, "Complete job")
60 stepDone(c, 3, "Complete job", cOK)
61 jobDone(c, cOK, nil, false)
62 }
63}
64
65// seenRun is the last Run frame the scripted executor received.
66var seenRun atomic.Pointer[execpb.Run]
67
68func runResultsJob(t *testing.T, sc script, tweak func(*harness)) *harness {
69 t.Helper()
70 h := newHarness(t, sc)
71 if tweak != nil {
72 tweak(h)
73 }
74 h.srv.jobs = []*ermite.Job{testJob("j1")}
75 stop := h.run(t)
76 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
77 if err := stop(); err != nil {
78 t.Fatal(err)
79 }
80 return h
81}
82
83func TestAnnotationsSummariesAndOutcomeAreForwarded(t *testing.T) {
84 h := runResultsJob(t, resultsScript(nil), nil)
85 if r := seenRun.Load(); r == nil || !r.Job.SendAnnotations || !r.Job.SendSummaries {
86 t.Errorf("the executor was not told to send: %+v", r)
87 }
88 s := h.srv
89 s.mu.Lock()
90 defer s.mu.Unlock()
91 // the annotations travel with the UpdateJob that completes the step
92 var withAnn *ermite.UpdateJobRequest
93 for _, u := range s.updates {
94 if len(u.Annotations) > 0 {
95 withAnn = u
96 }
97 }
98 if withAnn == nil || len(withAnn.Steps) != 1 || withAnn.Steps[0].Number != 2 || withAnn.Steps[0].State != ermite.StepState_STEP_STATE_COMPLETED {
99 t.Fatalf("annotations did not ride the completion of step 2: %+v", s.updates)
100 }
101 got := s.annots[2]
102 if len(got) != 2 || got[0].Level != ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING || got[0].Title != "T" || got[0].File != "a.go" ||
103 got[0].StartLine != 3 || got[0].EndLine != 4 || got[0].StepNumber != 2 || got[1].Level != ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE {
104 t.Errorf("annotations %+v", got)
105 }
106 if s.summaries[2] != "# Summary\n" {
107 t.Errorf("summaries %v", s.summaries)
108 }
109 u := s.steps[2]
110 if u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS || u.Outcome != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
111 t.Errorf("conclusion %v outcome %v", u.Conclusion, u.Outcome)
112 }
113 if o := s.steps[3].Outcome; o != ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
114 t.Errorf("a step that sends no outcome carries none: %v", o)
115 }
116 // nothing is left for CompleteJob when the UpdateJob went through
117 if n := len(s.completes[0].Annotations); n != 0 {
118 t.Errorf("CompleteJob repeats %d annotations", n)
119 }
120}
121
122// Without the server features, nothing is sent.
123func TestResultsAreNotSentWithoutTheFeatures(t *testing.T) {
124 h := runResultsJob(t, resultsScript(nil), func(h *harness) {
125 h.runner.Features = func(context.Context) ([]string, error) { return []string{"ci.runners"}, nil }
126 })
127 if r := seenRun.Load(); r == nil || r.Job.SendAnnotations || r.Job.SendSummaries {
128 t.Errorf("the executor was told to send: %+v", r)
129 }
130 s := h.srv
131 s.mu.Lock()
132 defer s.mu.Unlock()
133 if len(s.annots) != 0 || len(s.summaries) != 0 {
134 t.Errorf("annotations %v summaries %v", s.annots, s.summaries)
135 }
136 if len(s.completes[0].Annotations) != 0 {
137 t.Error("CompleteJob carries annotations")
138 }
139 // the outcome is part of the StepUpdate, not of a feature
140 if s.steps[2].Outcome != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
141 t.Error("outcome")
142 }
143}
144
145func TestFeaturesUnreachableMeansNothingIsSent(t *testing.T) {
146 h := runResultsJob(t, resultsScript(nil), func(h *harness) {
147 h.runner.Features = func(context.Context) ([]string, error) { return nil, errors.New("down") }
148 })
149 h.srv.mu.Lock()
150 defer h.srv.mu.Unlock()
151 if len(h.srv.annots) != 0 || len(h.srv.summaries) != 0 {
152 t.Error("sent without knowing the features")
153 }
154}
155
156// The host enforces the limits of the contract on what the guest sends: ten
157// per level, 4096 characters, titles of 255, valid numbers.
158func TestGuestAnnotationsAreBoundedByTheHost(t *testing.T) {
159 var many []*execpb.Annotation
160 for range 12 {
161 many = append(many, &execpb.Annotation{Level: lWarn, Message: "w"})
162 }
163 many = append(many,
164 &execpb.Annotation{Level: lFail, Message: strings.Repeat("é", 4097)}, // too long: dropped
165 &execpb.Annotation{Level: lFail, Message: strings.Repeat("é", 4096), Title: strings.Repeat("t", 300)}, // kept, title cut
166 &execpb.Annotation{Level: lFail, Message: ""}, // empty: dropped
167 &execpb.Annotation{Level: execpb.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED, Message: "no level"}, // dropped
168 &execpb.Annotation{Level: lFail, Message: "bad numbers", StartLine: -4, EndLine: -1, StartColumn: 5}, // numbers made valid
169 &execpb.Annotation{Level: lFail, Message: "end before", StartLine: 9, EndLine: 3},
170 )
171 sc := func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
172 stepStart(c, 1, "Set up job")
173 annFrame(c, 1, many...)
174 stepDone(c, 1, "Set up job", cOK)
175 jobDone(c, cOK, nil, false)
176 }
177 h := runResultsJob(t, sc, nil)
178 s := h.srv
179 s.mu.Lock()
180 defer s.mu.Unlock()
181 count := map[ermite.AnnotationLevel]int{}
182 for _, a := range s.annots[1] {
183 count[a.Level]++
184 }
185 if count[ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING] != 10 || count[ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE] != 3 {
186 t.Fatalf("counts %v", count)
187 }
188 var cut, neg, end *ermite.Annotation
189 for _, a := range s.annots[1] {
190 switch a.Message {
191 case strings.Repeat("é", 4096):
192 cut = a
193 case "bad numbers":
194 neg = a
195 case "end before":
196 end = a
197 }
198 }
199 if cut == nil || len([]rune(cut.Title)) != 255 {
200 t.Errorf("title: %v", cut)
201 }
202 if neg == nil || neg.StartLine != 0 || neg.EndLine != 0 || neg.StartColumn != 5 {
203 t.Errorf("%+v", neg)
204 }
205 if end == nil || end.StartLine != 9 || end.EndLine != 0 {
206 t.Errorf("%+v", end)
207 }
208 if v := s.violations; len(v) != 0 {
209 t.Error(v)
210 }
211}
212
213func TestGuestResultsForAnUnstartedStepFailTheJob(t *testing.T) {
214 h := hostile(t, "annotations for step 9, which has not started", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
215 stepStart(c, 1, "Set up job")
216 annFrame(c, 9, &execpb.Annotation{Level: lWarn, Message: "x"})
217 time.Sleep(50 * time.Millisecond)
218 })
219 _ = h
220}
221
222// The annotations of a step that never completes (the executor is stopped) are
223// sent with CompleteJob.
224func TestAnnotationsOfAnUnfinishedStepGoWithCompleteJob(t *testing.T) {
225 sc := func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
226 stepStart(c, 1, "Set up job")
227 stepDone(c, 1, "Set up job", cOK)
228 stepStart(c, 2, "Run it")
229 annFrame(c, 2, &execpb.Annotation{Level: lWarn, Message: "last words"})
230 // the executor dies here, without completing step 2 or the job
231 }
232 h := runResultsJob(t, sc, nil)
233 h.srv.mu.Lock()
234 defer h.srv.mu.Unlock()
235 if len(h.srv.completes[0].Annotations) != 1 || h.srv.completes[0].Annotations[0].Message != "last words" {
236 t.Errorf("%+v", h.srv.completes[0].Annotations)
237 }
238}
239
240// UpdateJob that fails and is retried: the annotations are not duplicated,
241// since the list replaces the stored one.
242func TestAnnotationsSurviveARetriedUpdateJob(t *testing.T) {
243 var fails atomic.Int32
244 h := runResultsJob(t, resultsScript(nil), func(h *harness) {
245 h.srv.updateHook = func(_ int, r *ermite.UpdateJobRequest) error {
246 if len(r.Annotations) > 0 && fails.Add(1) <= 2 {
247 return connect.NewError(connect.CodeUnavailable, nil)
248 }
249 return nil
250 }
251 })
252 h.srv.mu.Lock()
253 defer h.srv.mu.Unlock()
254 if len(h.srv.annots[2]) != 2 {
255 t.Errorf("%d annotations after retries", len(h.srv.annots[2]))
256 }
257}
258
259// A refused UpdateJob (invalid_argument) does not lose the step update: the
260// step is resent without its annotations.
261func TestRefusedAnnotationsDoNotLoseTheStep(t *testing.T) {
262 h := runResultsJob(t, resultsScript(nil), func(h *harness) {
263 h.srv.updateHook = func(_ int, r *ermite.UpdateJobRequest) error {
264 if len(r.Annotations) > 0 {
265 return connect.NewError(connect.CodeInvalidArgument, nil)
266 }
267 return nil
268 }
269 })
270 h.srv.mu.Lock()
271 defer h.srv.mu.Unlock()
272 if u := h.srv.steps[2]; u == nil || u.State != ermite.StepState_STEP_STATE_COMPLETED {
273 t.Errorf("step 2: %v", u)
274 }
275 if len(h.srv.annots[2]) != 0 {
276 t.Error("refused annotations stored")
277 }
278 if len(h.srv.completes[0].Annotations) != 0 {
279 t.Error("a refused list is sent again with CompleteJob")
280 }
281}
282
283func TestSummaryOverOneMiBIsDropped(t *testing.T) {
284 sc := func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
285 stepStart(c, 1, "Set up job")
286 sumFrame(c, 1, strings.Repeat("x", 1<<20+1))
287 stepDone(c, 1, "Set up job", cOK)
288 stepStart(c, 2, "Run it")
289 sumFrame(c, 2, strings.Repeat("y", 1<<20))
290 stepDone(c, 2, "Run it", cOK)
291 jobDone(c, cOK, nil, false)
292 }
293 h := runResultsJob(t, sc, nil)
294 h.srv.mu.Lock()
295 defer h.srv.mu.Unlock()
296 if _, has := h.srv.summaries[1]; has {
297 t.Error("a summary over 1 MiB was sent")
298 }
299 if len(h.srv.summaries[2]) != 1<<20 || h.srv.summaries[2][0] != 'y' {
300 t.Errorf("summary of %d bytes", len(h.srv.summaries[2]))
301 }
302}
303
304// One summary and one list of annotations per step, sent while the step runs
305// (GO-002): a second one, or one for a completed step, is the guest
306// misbehaving, and it would let it queue unbounded data.
307func TestGuestSecondSummaryFailsTheJob(t *testing.T) {
308 hostile(t, "a summary for step 1, which already sent them", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
309 stepStart(c, 1, "Set up job")
310 sumFrame(c, 1, "one")
311 sumFrame(c, 1, "two")
312 time.Sleep(50 * time.Millisecond)
313 })
314}
315
316func TestGuestSummaryForACompletedStepFailsTheJob(t *testing.T) {
317 hostile(t, "a summary for step 1, which is completed", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
318 stepStart(c, 1, "Set up job")
319 stepDone(c, 1, "Set up job", cOK)
320 sumFrame(c, 1, "late")
321 time.Sleep(50 * time.Millisecond)
322 })
323}
324
325func TestGuestSecondAnnotationListFailsTheJob(t *testing.T) {
326 hostile(t, "annotations for step 1, which already sent them", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
327 stepStart(c, 1, "Set up job")
328 annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "a"})
329 annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "b"})
330 time.Sleep(50 * time.Millisecond)
331 })
332}
333
334func TestGuestAnnotationsForACompletedStepFailTheJob(t *testing.T) {
335 hostile(t, "annotations for step 1, which is completed", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
336 stepStart(c, 1, "Set up job")
337 stepDone(c, 1, "Set up job", cOK)
338 annFrame(c, 1, &execpb.Annotation{Level: lWarn, Message: "late"})
339 time.Sleep(50 * time.Millisecond)
340 })
341}
342
343// The summaries count in the memory the outbox bounds, and backpressure holds
344// the producer while the server does not answer (GO-002).
345func TestSummaryBytesAreBoundedByTheOutbox(t *testing.T) {
346 release := make(chan struct{})
347 cl := &blockingSummaryClient{release: release}
348 o := newOutbox(cl, "j1", fastTiming(), func(error) {}, zerowrapNop())
349 ctx, cancel := context.WithCancel(context.Background())
350 defer cancel()
351 o.start(ctx)
352 defer close(release)
353 big := strings.Repeat("x", 1<<20)
354 go func() {
355 for i := 1; i <= 40; i++ { // 40 MiB offered, 16 MiB allowed in memory
356 o.Summary(int32(i), big)
357 }
358 }()
359 time.Sleep(500 * time.Millisecond)
360 o.mu.Lock()
361 queued := o.queued
362 o.mu.Unlock()
363 if queued > maxQueuedBytes+len(big)+64 {
364 t.Errorf("%d bytes queued, the bound is %d", queued, maxQueuedBytes)
365 }
366 if queued < maxQueuedBytes/2 {
367 t.Errorf("only %d bytes queued: the summaries are not counted?", queued)
368 }
369}
370
371type blockingSummaryClient struct {
372 stubClient
373 release chan struct{}
374}
375
376func (c *blockingSummaryClient) UploadSummary(ctx context.Context, _ *connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
377 select {
378 case <-c.release:
379 case <-ctx.Done():
380 return nil, ctx.Err()
381 }
382 return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
383}
384
385// A string cut by the host must not keep the allocation of the frame the guest
386// sent it in (GO-003): the cut copy has its own memory.
387func TestCutStringsDoNotRetainTheGuestAllocation(t *testing.T) {
388 big := strings.Repeat("a", 1<<20)
389 t0 := cutRunes(big, 255)
390 if len(t0) != 255 {
391 t.Fatal(len(t0))
392 }
393 if unsafe.StringData(t0) == unsafe.StringData(big) {
394 t.Error("cutRunes returns a substring of its argument")
395 }
396 name := cutName(big)
397 if len(name) != maxNameLen || unsafe.StringData(name) == unsafe.StringData(big) {
398 t.Error("cutName returns a substring of its argument")
399 }
400 // an annotation that is kept whole (not cut) is not copied: its message is
401 // within 4096 characters, a frame cannot make it retain more than that
402 // through a substring because the host does not slice it.
403 a := annotationsFrom(1, []*execpb.Annotation{{Level: lWarn, Message: "m", Title: big, File: strings.Repeat("f", 2000)}})
404 if len(a) != 1 || len(a[0].Title) != 255 || unsafe.StringData(a[0].Title) == unsafe.StringData(big) {
405 t.Errorf("title %d", len(a[0].Title))
406 }
407 if a[0].File != "" {
408 t.Error("a file over 1,024 characters is dropped")
409 }
410}
411
412// A message of the frame is bounded by execstream.MaxFrame: a summary of 1 MiB
413// fits.
414func TestSummaryFitsAFrame(t *testing.T) {
415 f := &execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: 1, Markdown: strings.Repeat("x", 1<<20)}}}
416 c := execstream.NewConn(&discard{})
417 if err := c.Send(f); err != nil {
418 t.Fatal(err)
419 }
420}
421
422type discard struct{}
423
424func (discard) Write(p []byte) (int, error) { return len(p), nil }
425func (discard) Read([]byte) (int, error) { select {} }
426
427func zerowrapNop() zerowrap.Logger {
428 return zerowrap.New(zerowrap.Config{Level: "error", Format: "json", Output: io.Discard})
429}
430
431// GO-006: the reader of the stream of the sandbox waits in the outbox for room
432// when the summaries fill the budget and the server keeps failing; the hard
433// stop of the job must wake it, or the job outlives its deadline and the
434// shutdown waits for ever. Heartbeats work, every upload fails transiently,
435// summaries of distinct steps fill the queue, the runner is stopped.
436func TestHardStopReleasesAProducerWaitingForRoom(t *testing.T) {
437 sc := func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
438 big := strings.Repeat("s", 1<<20)
439 for i := 1; i <= 30; i++ { // 30 MiB of summaries, 16 MiB of room
440 stepStart(c, i, "step")
441 sumFrame(c, i, big)
442 }
443 // the executor ignores the cancellation, as a stuck one would
444 <-cancels
445 time.Sleep(time.Hour)
446 }
447 h := newHarness(t, sc)
448 h.runner.Timing.ShutdownGrace = 500 * time.Millisecond
449 h.runner.Timing.ShutdownReport = time.Second
450 h.runner.Timing.StopDeadline = 5 * time.Second
451 h.srv.summaryHook = func(*ermite.UploadSummaryRequest) error { return connect.NewError(connect.CodeUnavailable, nil) }
452 h.srv.jobs = []*ermite.Job{testJob("j1")}
453 stop := h.run(t)
454 eventually(t, "the queue to fill", func() bool {
455 h.srv.mu.Lock()
456 defer h.srv.mu.Unlock()
457 return len(h.srv.beats) > 2
458 })
459 time.Sleep(300 * time.Millisecond) // the reader is stuck in Summary by now
460 start := time.Now()
461 done := make(chan error, 1)
462 go func() { done <- stop() }()
463 select {
464 case <-done:
465 case <-time.After(15 * time.Second):
466 t.Fatal("the job did not end after the hard stop: the stream reader is stuck waiting for room")
467 }
468 if took := time.Since(start); took > 10*time.Second {
469 t.Errorf("shutdown took %v", took)
470 }
471 if h.srv.completeCount() != 1 {
472 t.Errorf("the job was not reported: %d CompleteJob", h.srv.completeCount())
473 }
474}
475
476// GO-007: annotations count in the budget of the outbox from the moment the
477// host holds them until the call that carries them is answered. With the
478// server not answering and 400 steps that each ended with the largest list the
479// contract allows, the memory held stays within the budget and the lists over
480// it are dropped, with a log line.
481func TestAnnotationsWaitingForTheServerAreBounded(t *testing.T) {
482 cl := &blockingUpdateClient{release: make(chan struct{})}
483 defer close(cl.release)
484 logs := &logBuf{}
485 tm := fastTiming()
486 ctx, cancel := context.WithCancel(context.Background())
487 defer cancel()
488 j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},
489 annSeen: map[int32]bool{}, sumSeen: map[int32]bool{}, features: jobFeatures{annotations: true},
490 cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: logs})}
491 j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)
492 j.outbox.start(ctx)
493 var list []*execpb.Annotation
494 for _, l := range []execpb.AnnotationLevel{lWarn, lFail, execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE} {
495 for range 10 {
496 list = append(list, &execpb.Annotation{Level: l, Message: strings.Repeat("m", 4096), Title: strings.Repeat("t", 255)})
497 }
498 }
499 for n := int32(1); n <= 400; n++ { // 400 x about 135 KiB offered, 16 MiB of room
500 j.stepStarted(&execpb.StepStarted{Number: n, Name: "s"})
501 j.stepAnnotations(&execpb.StepAnnotations{Step: n, Annotations: list})
502 j.stepCompleted(&execpb.StepCompleted{Number: n, Name: "s", Conclusion: cOK})
503 }
504 j.outbox.mu.Lock()
505 held := annotationsBytes(j.outbox.pendingAnns)
506 queued := j.outbox.queued
507 j.outbox.mu.Unlock()
508 if held > maxQueuedBytes {
509 t.Errorf("%d bytes of annotations wait for the server, the budget is %d", held, maxQueuedBytes)
510 }
511 if queued > maxQueuedBytes {
512 t.Errorf("%d bytes counted in the outbox, the budget is %d", queued, maxQueuedBytes)
513 }
514 if held < maxQueuedBytes/2 {
515 t.Errorf("only %d bytes held: the annotations are dropped too early", held)
516 }
517 if !strings.Contains(logs.String(), "the annotations of this step are dropped") {
518 t.Error("dropping annotations is not logged")
519 }
520}
521
522type blockingUpdateClient struct {
523 stubClient
524 release chan struct{}
525}
526
527func (c *blockingUpdateClient) UpdateJob(ctx context.Context, _ *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
528 select {
529 case <-c.release:
530 return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
531 case <-ctx.Done():
532 return nil, ctx.Err()
533 }
534}
535
536// Once the call that carried a list is answered, its bytes leave the budget.
537func TestAnnotationBudgetIsGivenBack(t *testing.T) {
538 var ob *outbox
539 h := runResultsJob(t, resultsScript(nil), nil)
540 _ = ob
541 if got := h.srv.annots[2]; len(got) != 2 {
542 t.Fatalf("annotations %v", got)
543 }
544 cl := &stubClient{}
545 o := newOutbox(cl, "j1", fastTiming(), func(error) {}, zerowrapNop())
546 ctx, cancel := context.WithCancel(context.Background())
547 defer cancel()
548 o.start(ctx)
549 list := []*ermite.Annotation{{Level: ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING, Message: "m", StepNumber: 2}}
550 n := annotationBytes(list)
551 if !o.ReserveAnnotations(n) {
552 t.Fatal("no room")
553 }
554 o.Step(&ermite.StepUpdate{Number: 2}, map[int32][]*ermite.Annotation{2: list})
555 eventually(t, "the call to be answered", func() bool {
556 o.mu.Lock()
557 defer o.mu.Unlock()
558 return o.queued == 0
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.
565func 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
614func 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.
627func 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}