1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "sync"
8 "testing"
9 "time"
10
11 "connectrpc.com/connect"
12 "github.com/bnema/zerowrap"
13
14 ermite "bnema.dev/ermite-api/gen/ermite"
15 "bnema.dev/ermite-runner/internal/execstream/execpb"
16)
17
18// stubClient answers the calls of the final report. CompleteJob records the
19// deadline of its context, then waits for the context to end.
20type stubClient struct {
21 mu sync.Mutex
22 deadline time.Time
23 hang bool // AppendLog waits for its context to end
24}
25
26func (c *stubClient) UpdateJob(context.Context, *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
27 return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
28}
29
30func (c *stubClient) UploadSummary(context.Context, *connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
31 return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
32}
33
34func (c *stubClient) AppendLog(ctx context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
35 c.mu.Lock()
36 hang := c.hang
37 c.mu.Unlock()
38 if hang {
39 <-ctx.Done()
40 return nil, ctx.Err()
41 }
42 return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: r.Msg.Offset + int64(len(r.Msg.Lines))}), nil
43}
44
45func (c *stubClient) Heartbeat(context.Context, *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) {
46 return connect.NewResponse(&ermite.HeartbeatResponse{}), nil
47}
48
49func (c *stubClient) WatchJob(context.Context, *connect.Request[ermite.WatchJobRequest]) (*connect.ServerStreamForClient[ermite.WatchJobResponse], error) {
50 return nil, connect.NewError(connect.CodeUnimplemented, nil)
51}
52
53func (c *stubClient) ResolveActions(context.Context, *connect.Request[ermite.ResolveActionsRequest]) (*connect.Response[ermite.ResolveActionsResponse], error) {
54 return nil, connect.NewError(connect.CodeUnimplemented, nil)
55}
56
57func (c *stubClient) CompleteJob(ctx context.Context, _ *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
58 c.mu.Lock()
59 c.deadline, _ = ctx.Deadline()
60 c.mu.Unlock()
61 return connect.NewResponse(&ermite.CompleteJobResponse{}), nil
62}
63
64// The report of a job gets ShutdownReport once the runner's shutdown channel is
65// closed, even when the stop timers did not mark the job as stopping (GO-004),
66// and CompleteDeadline otherwise.
67func TestReportBudgetFollowsTheShutdownChannel(t *testing.T) {
68 for _, tc := range []struct {
69 name string
70 shutdown bool
71 want time.Duration
72 }{
73 {"running", false, time.Hour},
74 {"shutdown", true, 30 * time.Second},
75 } {
76 t.Run(tc.name, func(t *testing.T) {
77 tm := Timing{
78 CallTimeout: 24 * time.Hour, CompleteDeadline: time.Hour, ShutdownReport: 30 * time.Second,
79 BackoffMin: time.Millisecond, BackoffMax: time.Millisecond,
80 }
81 cl := &stubClient{}
82 ctx, cancel := context.WithCancel(context.Background())
83 defer cancel()
84 j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{}, timeline: timeline{},
85 cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logBuf{}})}
86 j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)
87 j.outbox.start(ctx)
88 shutdown := make(chan struct{})
89 if tc.shutdown {
90 close(shutdown)
91 }
92 if j.isStopping() {
93 t.Fatal("the job must not be marked as stopping: only the channel is closed")
94 }
95 start := time.Now()
96 j.report(ctx, outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}, shutdown)
97 cl.mu.Lock()
98 defer cl.mu.Unlock()
99 if cl.deadline.IsZero() {
100 t.Fatal("CompleteJob was not called")
101 }
102 if got := cl.deadline.Sub(start); got > tc.want+time.Second || got < tc.want-time.Second {
103 t.Fatalf("budget %v, want about %v", got, tc.want)
104 }
105 })
106 }
107}