detached 5d5c671

test(host): fake RunnerService tests for polling, acknowledgement, cancel, log retries, completion and shutdown; job outputs checked by the executor

commit
5d5c671191cce7bab55dca7994a6998bfa968f89
author
brice · · 2026-10-12 18:43 UTC
parent
7d17a05
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
10 files · +1410 -8
M go.mod +3 -3
@@ -4,17 +4,19 @@ go 1.27.1
44
55 require (
66 bnema.dev/ermite-actions v0.0.0-20261012165742-b6c4c3a2f7ff
7+ bnema.dev/ermite-api v0.0.0-20261012155218-f8f64413e63c
8+ connectrpc.com/connect v1.21.0
79 github.com/bnema/zerowrap v1.4.1
810 github.com/containerd/errdefs v1.0.0
911 github.com/google/uuid v1.6.0
1012 github.com/moby/moby/api v1.56.1
1113 github.com/moby/moby/client v0.6.2
14+ golang.org/x/net v0.58.0
1215 golang.org/x/sys v0.49.0
1316 google.golang.org/protobuf v1.36.12
1417 )
1518
1619 require (
17- bnema.dev/ermite-api v0.0.0-20261012155218-f8f64413e63c // indirect
1820 buf.build/gen/go/bufbuild/bufplugin/protocolbuffers/go v1.36.12-20260722160903-4d94f3df3a7b.2 // indirect
1921 buf.build/gen/go/bufbuild/protodescriptor/protocolbuffers/go v1.36.12-20250109164928-1da0de137947.2 // indirect
2022 buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2 // indirect
@@ -31,7 +33,6 @@ require (
3133 buf.build/go/standard v0.1.1-0.20260325175353-2b287e071df5 // indirect
3234 cel.dev/cel-go v0.32.0 // indirect
3335 cel.dev/expr v0.25.3 // indirect
34- connectrpc.com/connect v1.21.0 // indirect
3536 connectrpc.com/otelconnect v0.9.0 // indirect
3637 github.com/Microsoft/go-winio v0.6.2 // indirect
3738 github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
@@ -90,7 +91,6 @@ require (
9091 golang.org/x/crypto v0.56.0 // indirect
9192 golang.org/x/exp v0.0.0-20260824195058-e88cd73687aa // indirect
9293 golang.org/x/mod v0.40.0 // indirect
93- golang.org/x/net v0.58.0 // indirect
9494 golang.org/x/sync v0.23.0 // indirect
9595 golang.org/x/term v0.45.0 // indirect
9696 golang.org/x/text v0.41.0 // indirect
@@ -4,17 +4,19 @@ go 1.27.1
4 4
5 require (5 require (
6 bnema.dev/ermite-actions v0.0.0-20261012165742-b6c4c3a2f7ff6 bnema.dev/ermite-actions v0.0.0-20261012165742-b6c4c3a2f7ff
7+ bnema.dev/ermite-api v0.0.0-20261012155218-f8f64413e63c
8+ connectrpc.com/connect v1.21.0
7 github.com/bnema/zerowrap v1.4.19 github.com/bnema/zerowrap v1.4.1
8 github.com/containerd/errdefs v1.0.010 github.com/containerd/errdefs v1.0.0
9 github.com/google/uuid v1.6.011 github.com/google/uuid v1.6.0
10 github.com/moby/moby/api v1.56.112 github.com/moby/moby/api v1.56.1
11 github.com/moby/moby/client v0.6.213 github.com/moby/moby/client v0.6.2
14+ golang.org/x/net v0.58.0
12 golang.org/x/sys v0.49.015 golang.org/x/sys v0.49.0
13 google.golang.org/protobuf v1.36.1216 google.golang.org/protobuf v1.36.12
14 )17 )
15 18
16 require (19 require (
17- bnema.dev/ermite-api v0.0.0-20261012155218-f8f64413e63c // indirect
18 buf.build/gen/go/bufbuild/bufplugin/protocolbuffers/go v1.36.12-20260722160903-4d94f3df3a7b.2 // indirect20 buf.build/gen/go/bufbuild/bufplugin/protocolbuffers/go v1.36.12-20260722160903-4d94f3df3a7b.2 // indirect
19 buf.build/gen/go/bufbuild/protodescriptor/protocolbuffers/go v1.36.12-20250109164928-1da0de137947.2 // indirect21 buf.build/gen/go/bufbuild/protodescriptor/protocolbuffers/go v1.36.12-20250109164928-1da0de137947.2 // indirect
20 buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2 // indirect22 buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2 // indirect
@@ -31,7 +33,6 @@ require (
31 buf.build/go/standard v0.1.1-0.20260325175353-2b287e071df5 // indirect33 buf.build/go/standard v0.1.1-0.20260325175353-2b287e071df5 // indirect
32 cel.dev/cel-go v0.32.0 // indirect34 cel.dev/cel-go v0.32.0 // indirect
33 cel.dev/expr v0.25.3 // indirect35 cel.dev/expr v0.25.3 // indirect
34- connectrpc.com/connect v1.21.0 // indirect
35 connectrpc.com/otelconnect v0.9.0 // indirect36 connectrpc.com/otelconnect v0.9.0 // indirect
36 github.com/Microsoft/go-winio v0.6.2 // indirect37 github.com/Microsoft/go-winio v0.6.2 // indirect
37 github.com/antlr4-go/antlr/v4 v4.13.1 // indirect38 github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
@@ -90,7 +91,6 @@ require (
90 golang.org/x/crypto v0.56.0 // indirect91 golang.org/x/crypto v0.56.0 // indirect
91 golang.org/x/exp v0.0.0-20260824195058-e88cd73687aa // indirect92 golang.org/x/exp v0.0.0-20260824195058-e88cd73687aa // indirect
92 golang.org/x/mod v0.40.0 // indirect93 golang.org/x/mod v0.40.0 // indirect
93- golang.org/x/net v0.58.0 // indirect
94 golang.org/x/sync v0.23.0 // indirect94 golang.org/x/sync v0.23.0 // indirect
95 golang.org/x/term v0.45.0 // indirect95 golang.org/x/term v0.45.0 // indirect
96 golang.org/x/text v0.41.0 // indirect96 golang.org/x/text v0.41.0 // indirect
M internal/execstream/frame.go +3 -1
@@ -25,7 +25,9 @@ import (
2525 )
2626
2727 // MaxFrame is the largest accepted serialized frame.
28-const MaxFrame = 1 << 20
28+// It leaves room for the largest job result: outputs of 1 MiB in total (the
29+// limit of ermite) plus the overhead of the encoding.
30+const MaxFrame = 2 << 20
2931
3032 // ChunkSize is the largest output payload placed in one Output frame.
3133 const ChunkSize = 32 * 1024
@@ -25,7 +25,9 @@ import (
25 )25 )
26 26
27 // MaxFrame is the largest accepted serialized frame.27 // MaxFrame is the largest accepted serialized frame.
28-const MaxFrame = 1 << 2028+// It leaves room for the largest job result: outputs of 1 MiB in total (the
29+// limit of ermite) plus the overhead of the encoding.
30+const MaxFrame = 2 << 20
29 31
30 // ChunkSize is the largest output payload placed in one Output frame.32 // ChunkSize is the largest output payload placed in one Output frame.
31 const ChunkSize = 32 * 102433 const ChunkSize = 32 * 1024
M internal/executor/steps.go +19 -0
@@ -29,6 +29,10 @@ const (
2929 // DefaultTimeoutMinutes is the timeout of a job without timeout-minutes.
3030 const DefaultTimeoutMinutes = 360
3131
32+// MaxOutputBytes is the limit of ermite on the job outputs, names and values
33+// together (1 MiB).
34+const MaxOutputBytes = 1 << 20
35+
3236 // maxIssueMessageLength is ExecutionContext._maxIssueMessageLength.
3337 const maxIssueMessageLength = 4096
3438
@@ -933,11 +937,26 @@ func (j *job) completeJob(num int) map[string]string {
933937 j.result = &m
934938 j.mu.Unlock()
935939 }
940+ total := 0
936941 for _, p := range pairs {
937942 if p.Value == "" {
938943 j.debug(num, fmt.Sprintf("Skip output '%s' since it's empty", p.Key))
939944 continue
940945 }
946+ // ermite refuses outputs above 1 MiB in total (runner.proto,
947+ // CompleteJobRequest.outputs), and they would not fit a frame of the
948+ // stream either: the job fails without outputs.
949+ if total += len(p.Key) + len(p.Value); total > MaxOutputBytes {
950+ result = resFailed
951+ j.errorf(num, "Fail to evaluate job outputs")
952+ j.errorf(num, "The job outputs are larger than 1 MiB in total.")
953+ j.mu.Lock()
954+ m := merge(j.result, resFailed)
955+ j.result = &m
956+ j.mu.Unlock()
957+ outputs = map[string]string{}
958+ break
959+ }
941960 outputs[p.Key] = p.Value
942961 }
943962 }
@@ -29,6 +29,10 @@ const (
29 // DefaultTimeoutMinutes is the timeout of a job without timeout-minutes.29 // DefaultTimeoutMinutes is the timeout of a job without timeout-minutes.
30 const DefaultTimeoutMinutes = 36030 const DefaultTimeoutMinutes = 360
31 31
32+// MaxOutputBytes is the limit of ermite on the job outputs, names and values
33+// together (1 MiB).
34+const MaxOutputBytes = 1 << 20
35+
32 // maxIssueMessageLength is ExecutionContext._maxIssueMessageLength.36 // maxIssueMessageLength is ExecutionContext._maxIssueMessageLength.
33 const maxIssueMessageLength = 409637 const maxIssueMessageLength = 4096
34 38
@@ -933,11 +937,26 @@ func (j *job) completeJob(num int) map[string]string {
933 j.result = &m937 j.result = &m
934 j.mu.Unlock()938 j.mu.Unlock()
935 }939 }
940+ total := 0
936 for _, p := range pairs {941 for _, p := range pairs {
937 if p.Value == "" {942 if p.Value == "" {
938 j.debug(num, fmt.Sprintf("Skip output '%s' since it's empty", p.Key))943 j.debug(num, fmt.Sprintf("Skip output '%s' since it's empty", p.Key))
939 continue944 continue
940 }945 }
946+ // ermite refuses outputs above 1 MiB in total (runner.proto,
947+ // CompleteJobRequest.outputs), and they would not fit a frame of the
948+ // stream either: the job fails without outputs.
949+ if total += len(p.Key) + len(p.Value); total > MaxOutputBytes {
950+ result = resFailed
951+ j.errorf(num, "Fail to evaluate job outputs")
952+ j.errorf(num, "The job outputs are larger than 1 MiB in total.")
953+ j.mu.Lock()
954+ m := merge(j.result, resFailed)
955+ j.result = &m
956+ j.mu.Unlock()
957+ outputs = map[string]string{}
958+ break
959+ }
941 outputs[p.Key] = p.Value960 outputs[p.Key] = p.Value
942 }961 }
943 }962 }
M internal/executor/steps_test.go +22 -0
@@ -475,3 +475,25 @@ func TestStopCommands(t *testing.T) {
475475 t.Errorf("%s", got)
476476 }
477477 }
478+
479+func TestJobOutputsOverOneMiBFailTheJob(t *testing.T) {
480+ needTools(t, "bash")
481+ rec, res := runLocal(t, `on: push
482+jobs:
483+ j:
484+ runs-on: x
485+ outputs:
486+ big: ${{ steps.s.outputs.big }}
487+ other: ${{ steps.s.outputs.big }}
488+ steps:
489+ - id: s
490+ run: echo "big=$(head -c 600000 /dev/zero | tr '\0' x)" >> "$GITHUB_OUTPUT"
491+`, "j", nil)
492+ if res.Conclusion != ConclusionFailure || len(res.Outputs) != 0 {
493+ t.Fatalf("%+v", res.Conclusion)
494+ }
495+ last := rec.steps[len(rec.steps)-1]
496+ if last.Name != CompleteJobName || last.Conclusion != ConclusionFailure || !strings.Contains(logOf(last), "##[error]The job outputs are larger than 1 MiB in total.") {
497+ t.Fatalf("%s %s", last.Conclusion, logOf(last))
498+ }
499+}
@@ -475,3 +475,25 @@ func TestStopCommands(t *testing.T) {
475 t.Errorf("%s", got)475 t.Errorf("%s", got)
476 }476 }
477 }477 }
478+
479+func TestJobOutputsOverOneMiBFailTheJob(t *testing.T) {
480+ needTools(t, "bash")
481+ rec, res := runLocal(t, `on: push
482+jobs:
483+ j:
484+ runs-on: x
485+ outputs:
486+ big: ${{ steps.s.outputs.big }}
487+ other: ${{ steps.s.outputs.big }}
488+ steps:
489+ - id: s
490+ run: echo "big=$(head -c 600000 /dev/zero | tr '\0' x)" >> "$GITHUB_OUTPUT"
491+`, "j", nil)
492+ if res.Conclusion != ConclusionFailure || len(res.Outputs) != 0 {
493+ t.Fatalf("%+v", res.Conclusion)
494+ }
495+ last := rec.steps[len(rec.steps)-1]
496+ if last.Name != CompleteJobName || last.Conclusion != ConclusionFailure || !strings.Contains(logOf(last), "##[error]The job outputs are larger than 1 MiB in total.") {
497+ t.Fatalf("%s %s", last.Conclusion, logOf(last))
498+ }
499+}
A internal/host/fake_test.go +549 -0
@@ -0,0 +1,549 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "io"
8+ "net"
9+ "net/http"
10+ "net/http/httptest"
11+ "strings"
12+ "sync"
13+ "testing"
14+ "time"
15+
16+ "connectrpc.com/connect"
17+ "golang.org/x/net/http2"
18+ "golang.org/x/net/http2/h2c"
19+
20+ ermite "bnema.dev/ermite-api/gen/ermite"
21+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
22+
23+ "bnema.dev/ermite-runner/internal/execstream"
24+ "bnema.dev/ermite-runner/internal/execstream/execpb"
25+ "bnema.dev/ermite-runner/internal/sandbox"
26+)
27+
28+// fakeServer is an in-memory RunnerService with the contract's rules for
29+// AppendLog (offsets), tokens and the end of a job. Hooks let a test inject
30+// failures.
31+type fakeServer struct {
32+ ermiteconnect.UnimplementedRunnerServiceHandler
33+
34+ mu sync.Mutex
35+ runnerTok string
36+ jobs []*ermite.Job // queued, handed out in order
37+ assigned map[string]*ermite.Job
38+ polls []string // poll ids seen
39+ pollHook func(n int, req *ermite.FetchJobRequest) (*ermite.FetchJobResponse, error)
40+ pollCalls int
41+ retryAfter time.Duration
42+
43+ beats []time.Time
44+ beatHook func(n int) error
45+ registers []*ermite.RegisterRunnerRequest
46+ registerTok string
47+
48+ lines []string // log text, joined pieces are not merged: one per line
49+ cont []bool
50+ steps map[int32]*ermite.StepUpdate
51+ appendReq []*ermite.AppendLogRequest
52+ appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)
53+ appendN int
54+
55+ completes []*ermite.CompleteJobRequest
56+ completeHook func(n int) (applyThenFail bool, err error)
57+ completeN int
58+ ended *ermite.JobEnded // when set, job calls answer JobEnded
59+ completed map[string]bool
60+
61+ watchMsgs chan *ermite.WatchJobResponse
62+ watches int
63+}
64+
65+func newFakeServer() *fakeServer {
66+ return &fakeServer{
67+ runnerTok: "ermiterun_tok", assigned: map[string]*ermite.Job{}, completed: map[string]bool{}, steps: map[int32]*ermite.StepUpdate{},
68+ watchMsgs: make(chan *ermite.WatchJobResponse, 16),
69+ }
70+}
71+
72+func endedErr(je *ermite.JobEnded) error {
73+ e := connect.NewError(connect.CodeFailedPrecondition, io.EOF)
74+ if d, err := connect.NewErrorDetail(je); err == nil {
75+ e.AddDetail(d)
76+ }
77+ return e
78+}
79+
80+func (s *fakeServer) checkRunner(h http.Header) error {
81+ if h.Get("Authorization") != "Bearer "+s.runnerTok {
82+ return connect.NewError(connect.CodeUnauthenticated, io.EOF)
83+ }
84+ return nil
85+}
86+
87+func (s *fakeServer) checkJob(h http.Header, id string) (*ermite.Job, error) {
88+ tok := strings.TrimPrefix(h.Get("Authorization"), "Bearer ")
89+ for _, j := range s.assigned {
90+ if j.JobToken == tok {
91+ if j.JobId != id {
92+ return nil, connect.NewError(connect.CodePermissionDenied, io.EOF)
93+ }
94+ return j, nil
95+ }
96+ }
97+ return nil, connect.NewError(connect.CodeUnauthenticated, io.EOF)
98+}
99+
100+func (s *fakeServer) RegisterRunner(_ context.Context, r *connect.Request[ermite.RegisterRunnerRequest]) (*connect.Response[ermite.RegisterRunnerResponse], error) {
101+ s.mu.Lock()
102+ defer s.mu.Unlock()
103+ s.registers = append(s.registers, r.Msg)
104+ if r.Header().Get("Authorization") != "Bearer reg-token" {
105+ return nil, connect.NewError(connect.CodePermissionDenied, io.EOF)
106+ }
107+ return connect.NewResponse(&ermite.RegisterRunnerResponse{RunnerId: "rid1", RunnerToken: "ermiterun_new"}), nil
108+}
109+
110+func (s *fakeServer) FetchJob(_ context.Context, r *connect.Request[ermite.FetchJobRequest]) (*connect.Response[ermite.FetchJobResponse], error) {
111+ s.mu.Lock()
112+ defer s.mu.Unlock()
113+ if err := s.checkRunner(r.Header()); err != nil {
114+ return nil, err
115+ }
116+ s.pollCalls++
117+ s.polls = append(s.polls, r.Msg.PollId)
118+ if s.pollHook != nil {
119+ if resp, err := s.pollHook(s.pollCalls, r.Msg); resp != nil || err != nil {
120+ if resp != nil && resp.Job != nil {
121+ s.assigned[resp.Job.JobId] = resp.Job
122+ }
123+ return connect.NewResponse(resp), err
124+ }
125+ }
126+ if len(s.jobs) == 0 {
127+ resp := &ermite.FetchJobResponse{}
128+ if s.retryAfter > 0 {
129+ resp.RetryAfter = durationpbOf(s.retryAfter)
130+ }
131+ return connect.NewResponse(resp), nil
132+ }
133+ j := s.jobs[0]
134+ s.jobs = s.jobs[1:]
135+ s.assigned[j.JobId] = j
136+ return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
137+}
138+
139+func (s *fakeServer) Heartbeat(_ context.Context, r *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) {
140+ s.mu.Lock()
141+ defer s.mu.Unlock()
142+ if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
143+ return nil, err
144+ }
145+ if s.ended != nil {
146+ return nil, endedErr(s.ended)
147+ }
148+ s.beats = append(s.beats, time.Now())
149+ if s.beatHook != nil {
150+ if err := s.beatHook(len(s.beats)); err != nil {
151+ return nil, err
152+ }
153+ }
154+ return connect.NewResponse(&ermite.HeartbeatResponse{}), nil
155+}
156+
157+func (s *fakeServer) UpdateJob(_ context.Context, r *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
158+ s.mu.Lock()
159+ defer s.mu.Unlock()
160+ if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
161+ return nil, err
162+ }
163+ if s.ended != nil || s.completed[r.Msg.JobId] {
164+ je := s.ended
165+ if je == nil {
166+ je = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}
167+ }
168+ return nil, endedErr(je)
169+ }
170+ for _, u := range r.Msg.Steps {
171+ s.steps[u.Number] = u
172+ }
173+ return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
174+}
175+
176+func (s *fakeServer) AppendLog(_ context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
177+ s.mu.Lock()
178+ defer s.mu.Unlock()
179+ if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
180+ return nil, err
181+ }
182+ if s.ended != nil || s.completed[r.Msg.JobId] {
183+ je := s.ended
184+ if je == nil {
185+ je = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}
186+ }
187+ return nil, endedErr(je)
188+ }
189+ s.appendN++
190+ s.appendReq = append(s.appendReq, r.Msg)
191+ if len(r.Msg.Lines) == 0 || len(r.Msg.Lines) > 1000 {
192+ return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
193+ }
194+ bytes := 0
195+ for _, l := range r.Msg.Lines {
196+ bytes += len(l.Text)
197+ if strings.Contains(l.Text, "\n") || l.Time == nil {
198+ return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
199+ }
200+ }
201+ if bytes > 64*1024 {
202+ return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
203+ }
204+ have := int64(len(s.lines))
205+ if r.Msg.Offset > have {
206+ return nil, connect.NewError(connect.CodeOutOfRange, io.EOF)
207+ }
208+ skip := have - r.Msg.Offset
209+ apply := func() {
210+ for i, l := range r.Msg.Lines {
211+ if int64(i) < skip {
212+ continue
213+ }
214+ s.lines = append(s.lines, l.Text)
215+ s.cont = append(s.cont, l.Continued)
216+ }
217+ }
218+ if s.appendHook != nil {
219+ thenFail, err := s.appendHook(s.appendN, r.Msg)
220+ if err != nil {
221+ if thenFail {
222+ apply()
223+ }
224+ return nil, err
225+ }
226+ }
227+ apply()
228+ return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: int64(len(s.lines))}), nil
229+}
230+
231+func (s *fakeServer) CompleteJob(_ context.Context, r *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
232+ s.mu.Lock()
233+ defer s.mu.Unlock()
234+ if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
235+ return nil, err
236+ }
237+ s.completeN++
238+ s.completes = append(s.completes, r.Msg)
239+ if s.completeHook != nil {
240+ thenFail, err := s.completeHook(s.completeN)
241+ if err != nil {
242+ if thenFail {
243+ s.completed[r.Msg.JobId] = true
244+ }
245+ return nil, err
246+ }
247+ }
248+ if s.completed[r.Msg.JobId] || s.ended != nil {
249+ c := ermite.CiConclusion_CI_CONCLUSION_FAILURE
250+ if s.ended != nil {
251+ c = s.ended.Conclusion
252+ }
253+ return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: c, Applied: false}), nil
254+ }
255+ s.completed[r.Msg.JobId] = true
256+ return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil
257+}
258+
259+func (s *fakeServer) WatchJob(ctx context.Context, r *connect.Request[ermite.WatchJobRequest], st *connect.ServerStream[ermite.WatchJobResponse]) error {
260+ s.mu.Lock()
261+ _, err := s.checkJob(r.Header(), r.Msg.JobId)
262+ s.watches++
263+ ended := s.ended
264+ s.mu.Unlock()
265+ if err != nil {
266+ return err
267+ }
268+ if ended != nil {
269+ return endedErr(ended)
270+ }
271+ for {
272+ select {
273+ case m := <-s.watchMsgs:
274+ if err := st.Send(m); err != nil {
275+ return err
276+ }
277+ if m.GetEnded() != nil {
278+ return nil
279+ }
280+ case <-ctx.Done():
281+ return ctx.Err()
282+ }
283+ }
284+}
285+
286+// start serves the handler over h2c on a local port and returns its URL.
287+func (s *fakeServer) start(t *testing.T) string {
288+ t.Helper()
289+ _, h := ermiteconnect.NewRunnerServiceHandler(s)
290+ srv := httptest.NewUnstartedServer(h2c.NewHandler(h, &http2.Server{}))
291+ srv.Start()
292+ t.Cleanup(srv.Close)
293+ return srv.URL
294+}
295+
296+func (s *fakeServer) snapshotLines() []string {
297+ s.mu.Lock()
298+ defer s.mu.Unlock()
299+ return append([]string(nil), s.lines...)
300+}
301+
302+// ---- fake sandbox: the "executor" is a function on the other end of a pipe
303+
304+type script func(t *testing.T, c *execstream.Conn, run *execpb.Run, cancels <-chan *execpb.Cancel)
305+
306+type fakeDriver struct {
307+ mu sync.Mutex
308+ script script
309+ t *testing.T
310+ news int
311+ stops int
312+ started int
313+ reaps int
314+ reapErr error
315+ startErr error
316+ stopErr error
317+ capacity int
318+ live int
319+ blockStart chan struct{} // when set, Start waits on it
320+ iso sandbox.Isolation
321+}
322+
323+func (d *fakeDriver) Mode() sandbox.Mode { return sandbox.Container }
324+func (d *fakeDriver) Isolation() sandbox.Isolation { return d.iso }
325+func (d *fakeDriver) Close() error { return nil }
326+func (d *fakeDriver) Reap(context.Context) error {
327+ d.mu.Lock()
328+ defer d.mu.Unlock()
329+ d.reaps++
330+ return d.reapErr
331+}
332+func (d *fakeDriver) New() (sandbox.Sandbox, error) {
333+ d.mu.Lock()
334+ defer d.mu.Unlock()
335+ if d.capacity > 0 && d.live >= d.capacity {
336+ return nil, sandbox.ErrCapacity
337+ }
338+ d.news++
339+ d.live++
340+ return &fakeSandbox{d: d}, nil
341+}
342+
343+type fakeSandbox struct {
344+ d *fakeDriver
345+ mu sync.Mutex
346+ conn net.Conn
347+ stoped bool
348+}
349+
350+func (s *fakeSandbox) ID() string { return "er00000000" }
351+func (s *fakeSandbox) Start(ctx context.Context, _ sandbox.JobSpec) (sandbox.Conn, error) {
352+ d := s.d
353+ if d.blockStart != nil {
354+ select {
355+ case <-d.blockStart:
356+ case <-ctx.Done():
357+ return nil, ctx.Err()
358+ }
359+ }
360+ d.mu.Lock()
361+ if d.startErr != nil {
362+ err := d.startErr
363+ d.mu.Unlock()
364+ return nil, err
365+ }
366+ d.started++
367+ sc := d.script
368+ d.mu.Unlock()
369+ host, guest := net.Pipe()
370+ s.mu.Lock()
371+ s.conn = guest
372+ s.mu.Unlock()
373+ go func() {
374+ defer func() { _ = guest.Close() }()
375+ c := execstream.NewConn(guest)
376+ _ = c.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "test", Sandbox: "container"}}})
377+ f, err := c.Recv()
378+ if err != nil || f.GetRun() == nil {
379+ return
380+ }
381+ cancels := make(chan *execpb.Cancel, 4)
382+ go func() {
383+ for {
384+ f, err := c.Recv()
385+ if err != nil {
386+ close(cancels)
387+ return
388+ }
389+ if cn := f.GetCancel(); cn != nil {
390+ cancels <- cn
391+ }
392+ }
393+ }()
394+ sc(d.t, c, f.GetRun(), cancels)
395+ }()
396+ return host, nil
397+}
398+
399+func (s *fakeSandbox) Stop(context.Context) error {
400+ s.mu.Lock()
401+ if s.conn != nil {
402+ _ = s.conn.Close()
403+ }
404+ first := !s.stoped
405+ s.stoped = true
406+ s.mu.Unlock()
407+ if first {
408+ s.d.mu.Lock()
409+ s.d.stops++
410+ s.d.live--
411+ err := s.d.stopErr
412+ s.d.mu.Unlock()
413+ return err
414+ }
415+ return nil
416+}
417+
418+// ---- script helpers
419+
420+func send(c *execstream.Conn, b *execpb.Frame) { _ = c.Send(b) }
421+
422+func stepStart(c *execstream.Conn, n int, name string) {
423+ send(c, &execpb.Frame{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: int32(n), Name: name, UnixNano: time.Now().UnixNano()}}})
424+}
425+func stepDone(c *execstream.Conn, n int, name string, concl execpb.StepConclusion) {
426+ send(c, &execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: concl, UnixNano: time.Now().UnixNano()}}})
427+}
428+func logLineFrame(c *execstream.Conn, n int, text string, continued bool) {
429+ send(c, &execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: time.Now().UnixNano(), Text: text, Continued: continued}}})
430+}
431+func jobDone(c *execstream.Conn, concl execpb.StepConclusion, outputs map[string]string, shutdown bool) {
432+ send(c, &execpb.Frame{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{Conclusion: concl, Outputs: outputs, RunnerShutdown: shutdown}}})
433+ send(c, &execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
434+}
435+
436+const (
437+ cOK = execpb.StepConclusion_STEP_CONCLUSION_SUCCESS
438+ cFail = execpb.StepConclusion_STEP_CONCLUSION_FAILURE
439+ cCancel = execpb.StepConclusion_STEP_CONCLUSION_CANCELLED
440+)
441+
442+// simpleScript runs a one-step job that logs the given lines.
443+func simpleScript(lines ...string) script {
444+ return func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
445+ stepStart(c, 1, "Set up job")
446+ stepDone(c, 1, "Set up job", cOK)
447+ stepStart(c, 2, "Run it")
448+ for _, l := range lines {
449+ logLineFrame(c, 2, l, false)
450+ }
451+ stepDone(c, 2, "Run it", cOK)
452+ stepStart(c, 3, "Complete job")
453+ stepDone(c, 3, "Complete job", cOK)
454+ jobDone(c, cOK, map[string]string{"out": "v"}, false)
455+ }
456+}
457+
458+// ---- harness
459+
460+func testJob(id string) *ermite.Job {
461+ return &ermite.Job{
462+ JobId: id, RunId: 7, Attempt: 1, JobToken: "ermitejob_" + id, WorkflowPath: ".github/workflows/ci.yml",
463+ WorkflowYaml: "on: push\njobs:\n j:\n runs-on: x\n steps: [{run: echo}]\n", JobKey: "j", Matrix: "{}",
464+ Contexts: `{"github":{"repository":"o/r"}}`, Repo: "r", TimeoutMinutes: 5,
465+ }
466+}
467+
468+func fastTiming() Timing {
469+ return Timing{
470+ PollTimeout: 5 * time.Second, HeartbeatEvery: 100 * time.Millisecond, HeartbeatRetry: 30 * time.Millisecond,
471+ CallTimeout: 5 * time.Second, BackoffMin: 10 * time.Millisecond, BackoffMax: 40 * time.Millisecond,
472+ LogFlush: 5 * time.Millisecond, WatchRetry: 20 * time.Millisecond, ShutdownGrace: 2 * time.Second,
473+ CompleteDeadline: 10 * time.Second, NoJobPause: 10 * time.Millisecond,
474+ }
475+}
476+
477+type harness struct {
478+ srv *fakeServer
479+ url string
480+ driver *fakeDriver
481+ runner *Runner
482+ logs *logBuf
483+}
484+
485+type logBuf struct {
486+ mu sync.Mutex
487+ b strings.Builder
488+}
489+
490+func (l *logBuf) logf(f string, a ...any) {
491+ l.mu.Lock()
492+ defer l.mu.Unlock()
493+ l.b.WriteString(strings.TrimSpace(sprintf(f, a...)) + "\n")
494+}
495+func (l *logBuf) String() string { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() }
496+
497+func newHarness(t *testing.T, sc script) *harness {
498+ t.Helper()
499+ srv := newFakeServer()
500+ url := srv.start(t)
501+ drv := &fakeDriver{script: sc, t: t, iso: sandbox.IsolationVM}
502+ lb := &logBuf{}
503+ h := &harness{srv: srv, url: url, driver: drv, logs: lb}
504+ h.runner = &Runner{
505+ Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
506+ Driver: drv,
507+ Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),
508+ Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
509+ Timing: fastTiming(), Logf: lb.logf,
510+ }
511+ return h
512+}
513+
514+// run starts the loop; stop cancels it and waits for Run to return.
515+func (h *harness) run(t *testing.T) (stop func() error) {
516+ t.Helper()
517+ ctx, cancel := context.WithCancel(context.Background())
518+ done := make(chan error, 1)
519+ go func() { done <- h.runner.Run(ctx) }()
520+ var once sync.Once
521+ var res error
522+ stop = func() error {
523+ once.Do(func() {
524+ cancel()
525+ select {
526+ case res = <-done:
527+ case <-time.After(20 * time.Second):
528+ t.Error("Run did not return after shutdown")
529+ }
530+ })
531+ return res
532+ }
533+ t.Cleanup(func() { _ = stop() })
534+ return stop
535+}
536+
537+func eventually(t *testing.T, what string, cond func() bool) {
538+ t.Helper()
539+ deadline := time.Now().Add(15 * time.Second)
540+ for time.Now().Before(deadline) {
541+ if cond() {
542+ return
543+ }
544+ time.Sleep(10 * time.Millisecond)
545+ }
546+ t.Fatalf("timed out waiting for %s", what)
547+}
548+
549+func (s *fakeServer) completeCount() int { s.mu.Lock(); defer s.mu.Unlock(); return len(s.completes) }
@@ -0,0 +1,549 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "io"
8+ "net"
9+ "net/http"
10+ "net/http/httptest"
11+ "strings"
12+ "sync"
13+ "testing"
14+ "time"
15+
16+ "connectrpc.com/connect"
17+ "golang.org/x/net/http2"
18+ "golang.org/x/net/http2/h2c"
19+
20+ ermite "bnema.dev/ermite-api/gen/ermite"
21+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
22+
23+ "bnema.dev/ermite-runner/internal/execstream"
24+ "bnema.dev/ermite-runner/internal/execstream/execpb"
25+ "bnema.dev/ermite-runner/internal/sandbox"
26+)
27+
28+// fakeServer is an in-memory RunnerService with the contract's rules for
29+// AppendLog (offsets), tokens and the end of a job. Hooks let a test inject
30+// failures.
31+type fakeServer struct {
32+ ermiteconnect.UnimplementedRunnerServiceHandler
33+
34+ mu sync.Mutex
35+ runnerTok string
36+ jobs []*ermite.Job // queued, handed out in order
37+ assigned map[string]*ermite.Job
38+ polls []string // poll ids seen
39+ pollHook func(n int, req *ermite.FetchJobRequest) (*ermite.FetchJobResponse, error)
40+ pollCalls int
41+ retryAfter time.Duration
42+
43+ beats []time.Time
44+ beatHook func(n int) error
45+ registers []*ermite.RegisterRunnerRequest
46+ registerTok string
47+
48+ lines []string // log text, joined pieces are not merged: one per line
49+ cont []bool
50+ steps map[int32]*ermite.StepUpdate
51+ appendReq []*ermite.AppendLogRequest
52+ appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)
53+ appendN int
54+
55+ completes []*ermite.CompleteJobRequest
56+ completeHook func(n int) (applyThenFail bool, err error)
57+ completeN int
58+ ended *ermite.JobEnded // when set, job calls answer JobEnded
59+ completed map[string]bool
60+
61+ watchMsgs chan *ermite.WatchJobResponse
62+ watches int
63+}
64+
65+func newFakeServer() *fakeServer {
66+ return &fakeServer{
67+ runnerTok: "ermiterun_tok", assigned: map[string]*ermite.Job{}, completed: map[string]bool{}, steps: map[int32]*ermite.StepUpdate{},
68+ watchMsgs: make(chan *ermite.WatchJobResponse, 16),
69+ }
70+}
71+
72+func endedErr(je *ermite.JobEnded) error {
73+ e := connect.NewError(connect.CodeFailedPrecondition, io.EOF)
74+ if d, err := connect.NewErrorDetail(je); err == nil {
75+ e.AddDetail(d)
76+ }
77+ return e
78+}
79+
80+func (s *fakeServer) checkRunner(h http.Header) error {
81+ if h.Get("Authorization") != "Bearer "+s.runnerTok {
82+ return connect.NewError(connect.CodeUnauthenticated, io.EOF)
83+ }
84+ return nil
85+}
86+
87+func (s *fakeServer) checkJob(h http.Header, id string) (*ermite.Job, error) {
88+ tok := strings.TrimPrefix(h.Get("Authorization"), "Bearer ")
89+ for _, j := range s.assigned {
90+ if j.JobToken == tok {
91+ if j.JobId != id {
92+ return nil, connect.NewError(connect.CodePermissionDenied, io.EOF)
93+ }
94+ return j, nil
95+ }
96+ }
97+ return nil, connect.NewError(connect.CodeUnauthenticated, io.EOF)
98+}
99+
100+func (s *fakeServer) RegisterRunner(_ context.Context, r *connect.Request[ermite.RegisterRunnerRequest]) (*connect.Response[ermite.RegisterRunnerResponse], error) {
101+ s.mu.Lock()
102+ defer s.mu.Unlock()
103+ s.registers = append(s.registers, r.Msg)
104+ if r.Header().Get("Authorization") != "Bearer reg-token" {
105+ return nil, connect.NewError(connect.CodePermissionDenied, io.EOF)
106+ }
107+ return connect.NewResponse(&ermite.RegisterRunnerResponse{RunnerId: "rid1", RunnerToken: "ermiterun_new"}), nil
108+}
109+
110+func (s *fakeServer) FetchJob(_ context.Context, r *connect.Request[ermite.FetchJobRequest]) (*connect.Response[ermite.FetchJobResponse], error) {
111+ s.mu.Lock()
112+ defer s.mu.Unlock()
113+ if err := s.checkRunner(r.Header()); err != nil {
114+ return nil, err
115+ }
116+ s.pollCalls++
117+ s.polls = append(s.polls, r.Msg.PollId)
118+ if s.pollHook != nil {
119+ if resp, err := s.pollHook(s.pollCalls, r.Msg); resp != nil || err != nil {
120+ if resp != nil && resp.Job != nil {
121+ s.assigned[resp.Job.JobId] = resp.Job
122+ }
123+ return connect.NewResponse(resp), err
124+ }
125+ }
126+ if len(s.jobs) == 0 {
127+ resp := &ermite.FetchJobResponse{}
128+ if s.retryAfter > 0 {
129+ resp.RetryAfter = durationpbOf(s.retryAfter)
130+ }
131+ return connect.NewResponse(resp), nil
132+ }
133+ j := s.jobs[0]
134+ s.jobs = s.jobs[1:]
135+ s.assigned[j.JobId] = j
136+ return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
137+}
138+
139+func (s *fakeServer) Heartbeat(_ context.Context, r *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) {
140+ s.mu.Lock()
141+ defer s.mu.Unlock()
142+ if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
143+ return nil, err
144+ }
145+ if s.ended != nil {
146+ return nil, endedErr(s.ended)
147+ }
148+ s.beats = append(s.beats, time.Now())
149+ if s.beatHook != nil {
150+ if err := s.beatHook(len(s.beats)); err != nil {
151+ return nil, err
152+ }
153+ }
154+ return connect.NewResponse(&ermite.HeartbeatResponse{}), nil
155+}
156+
157+func (s *fakeServer) UpdateJob(_ context.Context, r *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
158+ s.mu.Lock()
159+ defer s.mu.Unlock()
160+ if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
161+ return nil, err
162+ }
163+ if s.ended != nil || s.completed[r.Msg.JobId] {
164+ je := s.ended
165+ if je == nil {
166+ je = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}
167+ }
168+ return nil, endedErr(je)
169+ }
170+ for _, u := range r.Msg.Steps {
171+ s.steps[u.Number] = u
172+ }
173+ return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
174+}
175+
176+func (s *fakeServer) AppendLog(_ context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
177+ s.mu.Lock()
178+ defer s.mu.Unlock()
179+ if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
180+ return nil, err
181+ }
182+ if s.ended != nil || s.completed[r.Msg.JobId] {
183+ je := s.ended
184+ if je == nil {
185+ je = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}
186+ }
187+ return nil, endedErr(je)
188+ }
189+ s.appendN++
190+ s.appendReq = append(s.appendReq, r.Msg)
191+ if len(r.Msg.Lines) == 0 || len(r.Msg.Lines) > 1000 {
192+ return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
193+ }
194+ bytes := 0
195+ for _, l := range r.Msg.Lines {
196+ bytes += len(l.Text)
197+ if strings.Contains(l.Text, "\n") || l.Time == nil {
198+ return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
199+ }
200+ }
201+ if bytes > 64*1024 {
202+ return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF)
203+ }
204+ have := int64(len(s.lines))
205+ if r.Msg.Offset > have {
206+ return nil, connect.NewError(connect.CodeOutOfRange, io.EOF)
207+ }
208+ skip := have - r.Msg.Offset
209+ apply := func() {
210+ for i, l := range r.Msg.Lines {
211+ if int64(i) < skip {
212+ continue
213+ }
214+ s.lines = append(s.lines, l.Text)
215+ s.cont = append(s.cont, l.Continued)
216+ }
217+ }
218+ if s.appendHook != nil {
219+ thenFail, err := s.appendHook(s.appendN, r.Msg)
220+ if err != nil {
221+ if thenFail {
222+ apply()
223+ }
224+ return nil, err
225+ }
226+ }
227+ apply()
228+ return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: int64(len(s.lines))}), nil
229+}
230+
231+func (s *fakeServer) CompleteJob(_ context.Context, r *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
232+ s.mu.Lock()
233+ defer s.mu.Unlock()
234+ if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
235+ return nil, err
236+ }
237+ s.completeN++
238+ s.completes = append(s.completes, r.Msg)
239+ if s.completeHook != nil {
240+ thenFail, err := s.completeHook(s.completeN)
241+ if err != nil {
242+ if thenFail {
243+ s.completed[r.Msg.JobId] = true
244+ }
245+ return nil, err
246+ }
247+ }
248+ if s.completed[r.Msg.JobId] || s.ended != nil {
249+ c := ermite.CiConclusion_CI_CONCLUSION_FAILURE
250+ if s.ended != nil {
251+ c = s.ended.Conclusion
252+ }
253+ return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: c, Applied: false}), nil
254+ }
255+ s.completed[r.Msg.JobId] = true
256+ return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil
257+}
258+
259+func (s *fakeServer) WatchJob(ctx context.Context, r *connect.Request[ermite.WatchJobRequest], st *connect.ServerStream[ermite.WatchJobResponse]) error {
260+ s.mu.Lock()
261+ _, err := s.checkJob(r.Header(), r.Msg.JobId)
262+ s.watches++
263+ ended := s.ended
264+ s.mu.Unlock()
265+ if err != nil {
266+ return err
267+ }
268+ if ended != nil {
269+ return endedErr(ended)
270+ }
271+ for {
272+ select {
273+ case m := <-s.watchMsgs:
274+ if err := st.Send(m); err != nil {
275+ return err
276+ }
277+ if m.GetEnded() != nil {
278+ return nil
279+ }
280+ case <-ctx.Done():
281+ return ctx.Err()
282+ }
283+ }
284+}
285+
286+// start serves the handler over h2c on a local port and returns its URL.
287+func (s *fakeServer) start(t *testing.T) string {
288+ t.Helper()
289+ _, h := ermiteconnect.NewRunnerServiceHandler(s)
290+ srv := httptest.NewUnstartedServer(h2c.NewHandler(h, &http2.Server{}))
291+ srv.Start()
292+ t.Cleanup(srv.Close)
293+ return srv.URL
294+}
295+
296+func (s *fakeServer) snapshotLines() []string {
297+ s.mu.Lock()
298+ defer s.mu.Unlock()
299+ return append([]string(nil), s.lines...)
300+}
301+
302+// ---- fake sandbox: the "executor" is a function on the other end of a pipe
303+
304+type script func(t *testing.T, c *execstream.Conn, run *execpb.Run, cancels <-chan *execpb.Cancel)
305+
306+type fakeDriver struct {
307+ mu sync.Mutex
308+ script script
309+ t *testing.T
310+ news int
311+ stops int
312+ started int
313+ reaps int
314+ reapErr error
315+ startErr error
316+ stopErr error
317+ capacity int
318+ live int
319+ blockStart chan struct{} // when set, Start waits on it
320+ iso sandbox.Isolation
321+}
322+
323+func (d *fakeDriver) Mode() sandbox.Mode { return sandbox.Container }
324+func (d *fakeDriver) Isolation() sandbox.Isolation { return d.iso }
325+func (d *fakeDriver) Close() error { return nil }
326+func (d *fakeDriver) Reap(context.Context) error {
327+ d.mu.Lock()
328+ defer d.mu.Unlock()
329+ d.reaps++
330+ return d.reapErr
331+}
332+func (d *fakeDriver) New() (sandbox.Sandbox, error) {
333+ d.mu.Lock()
334+ defer d.mu.Unlock()
335+ if d.capacity > 0 && d.live >= d.capacity {
336+ return nil, sandbox.ErrCapacity
337+ }
338+ d.news++
339+ d.live++
340+ return &fakeSandbox{d: d}, nil
341+}
342+
343+type fakeSandbox struct {
344+ d *fakeDriver
345+ mu sync.Mutex
346+ conn net.Conn
347+ stoped bool
348+}
349+
350+func (s *fakeSandbox) ID() string { return "er00000000" }
351+func (s *fakeSandbox) Start(ctx context.Context, _ sandbox.JobSpec) (sandbox.Conn, error) {
352+ d := s.d
353+ if d.blockStart != nil {
354+ select {
355+ case <-d.blockStart:
356+ case <-ctx.Done():
357+ return nil, ctx.Err()
358+ }
359+ }
360+ d.mu.Lock()
361+ if d.startErr != nil {
362+ err := d.startErr
363+ d.mu.Unlock()
364+ return nil, err
365+ }
366+ d.started++
367+ sc := d.script
368+ d.mu.Unlock()
369+ host, guest := net.Pipe()
370+ s.mu.Lock()
371+ s.conn = guest
372+ s.mu.Unlock()
373+ go func() {
374+ defer func() { _ = guest.Close() }()
375+ c := execstream.NewConn(guest)
376+ _ = c.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "test", Sandbox: "container"}}})
377+ f, err := c.Recv()
378+ if err != nil || f.GetRun() == nil {
379+ return
380+ }
381+ cancels := make(chan *execpb.Cancel, 4)
382+ go func() {
383+ for {
384+ f, err := c.Recv()
385+ if err != nil {
386+ close(cancels)
387+ return
388+ }
389+ if cn := f.GetCancel(); cn != nil {
390+ cancels <- cn
391+ }
392+ }
393+ }()
394+ sc(d.t, c, f.GetRun(), cancels)
395+ }()
396+ return host, nil
397+}
398+
399+func (s *fakeSandbox) Stop(context.Context) error {
400+ s.mu.Lock()
401+ if s.conn != nil {
402+ _ = s.conn.Close()
403+ }
404+ first := !s.stoped
405+ s.stoped = true
406+ s.mu.Unlock()
407+ if first {
408+ s.d.mu.Lock()
409+ s.d.stops++
410+ s.d.live--
411+ err := s.d.stopErr
412+ s.d.mu.Unlock()
413+ return err
414+ }
415+ return nil
416+}
417+
418+// ---- script helpers
419+
420+func send(c *execstream.Conn, b *execpb.Frame) { _ = c.Send(b) }
421+
422+func stepStart(c *execstream.Conn, n int, name string) {
423+ send(c, &execpb.Frame{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: int32(n), Name: name, UnixNano: time.Now().UnixNano()}}})
424+}
425+func stepDone(c *execstream.Conn, n int, name string, concl execpb.StepConclusion) {
426+ send(c, &execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: concl, UnixNano: time.Now().UnixNano()}}})
427+}
428+func logLineFrame(c *execstream.Conn, n int, text string, continued bool) {
429+ send(c, &execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: time.Now().UnixNano(), Text: text, Continued: continued}}})
430+}
431+func jobDone(c *execstream.Conn, concl execpb.StepConclusion, outputs map[string]string, shutdown bool) {
432+ send(c, &execpb.Frame{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{Conclusion: concl, Outputs: outputs, RunnerShutdown: shutdown}}})
433+ send(c, &execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
434+}
435+
436+const (
437+ cOK = execpb.StepConclusion_STEP_CONCLUSION_SUCCESS
438+ cFail = execpb.StepConclusion_STEP_CONCLUSION_FAILURE
439+ cCancel = execpb.StepConclusion_STEP_CONCLUSION_CANCELLED
440+)
441+
442+// simpleScript runs a one-step job that logs the given lines.
443+func simpleScript(lines ...string) script {
444+ return func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
445+ stepStart(c, 1, "Set up job")
446+ stepDone(c, 1, "Set up job", cOK)
447+ stepStart(c, 2, "Run it")
448+ for _, l := range lines {
449+ logLineFrame(c, 2, l, false)
450+ }
451+ stepDone(c, 2, "Run it", cOK)
452+ stepStart(c, 3, "Complete job")
453+ stepDone(c, 3, "Complete job", cOK)
454+ jobDone(c, cOK, map[string]string{"out": "v"}, false)
455+ }
456+}
457+
458+// ---- harness
459+
460+func testJob(id string) *ermite.Job {
461+ return &ermite.Job{
462+ JobId: id, RunId: 7, Attempt: 1, JobToken: "ermitejob_" + id, WorkflowPath: ".github/workflows/ci.yml",
463+ WorkflowYaml: "on: push\njobs:\n j:\n runs-on: x\n steps: [{run: echo}]\n", JobKey: "j", Matrix: "{}",
464+ Contexts: `{"github":{"repository":"o/r"}}`, Repo: "r", TimeoutMinutes: 5,
465+ }
466+}
467+
468+func fastTiming() Timing {
469+ return Timing{
470+ PollTimeout: 5 * time.Second, HeartbeatEvery: 100 * time.Millisecond, HeartbeatRetry: 30 * time.Millisecond,
471+ CallTimeout: 5 * time.Second, BackoffMin: 10 * time.Millisecond, BackoffMax: 40 * time.Millisecond,
472+ LogFlush: 5 * time.Millisecond, WatchRetry: 20 * time.Millisecond, ShutdownGrace: 2 * time.Second,
473+ CompleteDeadline: 10 * time.Second, NoJobPause: 10 * time.Millisecond,
474+ }
475+}
476+
477+type harness struct {
478+ srv *fakeServer
479+ url string
480+ driver *fakeDriver
481+ runner *Runner
482+ logs *logBuf
483+}
484+
485+type logBuf struct {
486+ mu sync.Mutex
487+ b strings.Builder
488+}
489+
490+func (l *logBuf) logf(f string, a ...any) {
491+ l.mu.Lock()
492+ defer l.mu.Unlock()
493+ l.b.WriteString(strings.TrimSpace(sprintf(f, a...)) + "\n")
494+}
495+func (l *logBuf) String() string { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() }
496+
497+func newHarness(t *testing.T, sc script) *harness {
498+ t.Helper()
499+ srv := newFakeServer()
500+ url := srv.start(t)
501+ drv := &fakeDriver{script: sc, t: t, iso: sandbox.IsolationVM}
502+ lb := &logBuf{}
503+ h := &harness{srv: srv, url: url, driver: drv, logs: lb}
504+ h.runner = &Runner{
505+ Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
506+ Driver: drv,
507+ Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),
508+ Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
509+ Timing: fastTiming(), Logf: lb.logf,
510+ }
511+ return h
512+}
513+
514+// run starts the loop; stop cancels it and waits for Run to return.
515+func (h *harness) run(t *testing.T) (stop func() error) {
516+ t.Helper()
517+ ctx, cancel := context.WithCancel(context.Background())
518+ done := make(chan error, 1)
519+ go func() { done <- h.runner.Run(ctx) }()
520+ var once sync.Once
521+ var res error
522+ stop = func() error {
523+ once.Do(func() {
524+ cancel()
525+ select {
526+ case res = <-done:
527+ case <-time.After(20 * time.Second):
528+ t.Error("Run did not return after shutdown")
529+ }
530+ })
531+ return res
532+ }
533+ t.Cleanup(func() { _ = stop() })
534+ return stop
535+}
536+
537+func eventually(t *testing.T, what string, cond func() bool) {
538+ t.Helper()
539+ deadline := time.Now().Add(15 * time.Second)
540+ for time.Now().Before(deadline) {
541+ if cond() {
542+ return
543+ }
544+ time.Sleep(10 * time.Millisecond)
545+ }
546+ t.Fatalf("timed out waiting for %s", what)
547+}
548+
549+func (s *fakeServer) completeCount() int { s.mu.Lock(); defer s.mu.Unlock(); return len(s.completes) }
M internal/host/job.go +2 -1
@@ -298,7 +298,8 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
298298 bg.Add(2)
299299 go func() { defer bg.Done(); j.heartbeats(ctx) }()
300300 go func() { defer bg.Done(); j.watch(ctx) }()
301- defer bg.Wait()
301+ // The context is cancelled first, so that the heartbeat and watch loops end.
302+ defer func() { j.cancel(nil); bg.Wait() }()
302303
303304 j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)
304305 j.outbox.start(ctx)
@@ -298,7 +298,8 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
298 bg.Add(2)298 bg.Add(2)
299 go func() { defer bg.Done(); j.heartbeats(ctx) }()299 go func() { defer bg.Done(); j.heartbeats(ctx) }()
300 go func() { defer bg.Done(); j.watch(ctx) }()300 go func() { defer bg.Done(); j.watch(ctx) }()
301- defer bg.Wait()301+ // The context is cancelled first, so that the heartbeat and watch loops end.
302+ defer func() { j.cancel(nil); bg.Wait() }()
302 303
303 j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)304 j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)
304 j.outbox.start(ctx)305 j.outbox.start(ctx)
M internal/host/rpc.go +3 -0
@@ -120,6 +120,8 @@ type Timing struct {
120120 // NoJobPause is the pause after a poll without a job and without
121121 // retry_after.
122122 NoJobPause time.Duration
123+ // ReapRetry is the delay between attempts to reap after a leaked sandbox.
124+ ReapRetry time.Duration
123125 }
124126
125127 func (t Timing) withDefaults() Timing {
@@ -139,6 +141,7 @@ func (t Timing) withDefaults() Timing {
139141 def(&t.ShutdownGrace, 30*time.Second)
140142 def(&t.CompleteDeadline, 5*time.Minute)
141143 def(&t.NoJobPause, 0)
144+ def(&t.ReapRetry, 5*time.Second)
142145 return t
143146 }
144147
@@ -120,6 +120,8 @@ type Timing struct {
120 // NoJobPause is the pause after a poll without a job and without120 // NoJobPause is the pause after a poll without a job and without
121 // retry_after.121 // retry_after.
122 NoJobPause time.Duration122 NoJobPause time.Duration
123+ // ReapRetry is the delay between attempts to reap after a leaked sandbox.
124+ ReapRetry time.Duration
123 }125 }
124 126
125 func (t Timing) withDefaults() Timing {127 func (t Timing) withDefaults() Timing {
@@ -139,6 +141,7 @@ func (t Timing) withDefaults() Timing {
139 def(&t.ShutdownGrace, 30*time.Second)141 def(&t.ShutdownGrace, 30*time.Second)
140 def(&t.CompleteDeadline, 5*time.Minute)142 def(&t.CompleteDeadline, 5*time.Minute)
141 def(&t.NoJobPause, 0)143 def(&t.NoJobPause, 0)
144+ def(&t.ReapRetry, 5*time.Second)
142 return t145 return t
143 }146 }
144 147
M internal/host/runner.go +3 -3
@@ -99,7 +99,7 @@ func (r *Runner) Run(ctx context.Context) error {
9999 }
100100 if r.leaked.Load() {
101101 <-slots
102- if !r.recover(ctx, slots) {
102+ if !r.recover(ctx, slots, t.ReapRetry) {
103103 return nil
104104 }
105105 continue
@@ -152,7 +152,7 @@ func (r *Runner) Run(ctx context.Context) error {
152152 }
153153
154154 // recover waits until no job runs, then reaps. It returns false when ctx ends.
155-func (r *Runner) recover(ctx context.Context, slots chan struct{}) bool {
155+func (r *Runner) recover(ctx context.Context, slots chan struct{}, retry time.Duration) bool {
156156 r.logf("a sandbox could not be cleaned up: waiting for the running jobs, then reaping")
157157 for ctx.Err() == nil {
158158 if len(slots) == 0 {
@@ -164,7 +164,7 @@ func (r *Runner) recover(ctx context.Context, slots chan struct{}) bool {
164164 r.logf("reap failed, retrying: %v", err)
165165 }
166166 }
167- if !sleep(ctx, 5*time.Second) {
167+ if !sleep(ctx, retry) {
168168 return false
169169 }
170170 }
@@ -99,7 +99,7 @@ func (r *Runner) Run(ctx context.Context) error {
99 }99 }
100 if r.leaked.Load() {100 if r.leaked.Load() {
101 <-slots101 <-slots
102- if !r.recover(ctx, slots) {102+ if !r.recover(ctx, slots, t.ReapRetry) {
103 return nil103 return nil
104 }104 }
105 continue105 continue
@@ -152,7 +152,7 @@ func (r *Runner) Run(ctx context.Context) error {
152 }152 }
153 153
154 // recover waits until no job runs, then reaps. It returns false when ctx ends.154 // recover waits until no job runs, then reaps. It returns false when ctx ends.
155-func (r *Runner) recover(ctx context.Context, slots chan struct{}) bool {155+func (r *Runner) recover(ctx context.Context, slots chan struct{}, retry time.Duration) bool {
156 r.logf("a sandbox could not be cleaned up: waiting for the running jobs, then reaping")156 r.logf("a sandbox could not be cleaned up: waiting for the running jobs, then reaping")
157 for ctx.Err() == nil {157 for ctx.Err() == nil {
158 if len(slots) == 0 {158 if len(slots) == 0 {
@@ -164,7 +164,7 @@ func (r *Runner) recover(ctx context.Context, slots chan struct{}) bool {
164 r.logf("reap failed, retrying: %v", err)164 r.logf("reap failed, retrying: %v", err)
165 }165 }
166 }166 }
167- if !sleep(ctx, 5*time.Second) {167+ if !sleep(ctx, retry) {
168 return false168 return false
169 }169 }
170 }170 }
A internal/host/runner_test.go +792 -0
@@ -0,0 +1,792 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "errors"
8+ "io"
9+ "net/http"
10+ "os"
11+ "strings"
12+ "testing"
13+ "time"
14+
15+ "connectrpc.com/connect"
16+
17+ ermite "bnema.dev/ermite-api/gen/ermite"
18+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
19+
20+ "bnema.dev/ermite-runner/internal/execstream"
21+ "bnema.dev/ermite-runner/internal/execstream/execpb"
22+ "bnema.dev/ermite-runner/internal/sandbox"
23+)
24+
25+func TestJobRunsAndReports(t *testing.T) {
26+ h := newHarness(t, simpleScript("hello", "world"))
27+ h.srv.jobs = []*ermite.Job{testJob("j1")}
28+ stop := h.run(t)
29+
30+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
31+ if err := stop(); err != nil {
32+ t.Fatal(err)
33+ }
34+ s := h.srv
35+ s.mu.Lock()
36+ defer s.mu.Unlock()
37+ c := s.completes[0]
38+ if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS || c.Outputs["out"] != "v" || c.RunnerShutdown {
39+ t.Fatalf("%+v", c)
40+ }
41+ if len(s.beats) == 0 {
42+ t.Error("no heartbeat: the first one is the acknowledgement")
43+ }
44+ if strings.Join(s.lines, "|") != "hello|world" {
45+ t.Errorf("lines %q", s.lines)
46+ }
47+ for n, name := range map[int32]string{1: "Set up job", 2: "Run it", 3: "Complete job"} {
48+ u := s.steps[n]
49+ if u == nil || u.Name != name || u.State != ermite.StepState_STEP_STATE_COMPLETED || u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS || u.StartedAt == nil || u.CompletedAt == nil {
50+ t.Errorf("step %d: %v", n, u)
51+ }
52+ }
53+ if len(c.Steps) != 3 {
54+ t.Errorf("CompleteJob carries %d final steps", len(c.Steps))
55+ }
56+ if h.driver.stops != 1 || h.driver.reaps < 1 {
57+ t.Errorf("stops=%d reaps=%d: the sandbox must be stopped and the start must reap", h.driver.stops, h.driver.reaps)
58+ }
59+ if strings.Contains(h.logs.String(), "ermitejob_") || strings.Contains(h.logs.String(), "ermiterun_") {
60+ t.Errorf("a token reached the log: %s", h.logs.String())
61+ }
62+}
63+
64+func TestTheJobTokenAndTheRunnerTokenStayApart(t *testing.T) {
65+ h := newHarness(t, simpleScript("x"))
66+ h.srv.jobs = []*ermite.Job{testJob("j1")}
67+ var sawJobSpec *execpb.JobSpec
68+ inner := h.driver.script
69+ h.driver.script = func(t *testing.T, c *execstream.Conn, run *execpb.Run, cn <-chan *execpb.Cancel) {
70+ sawJobSpec = run.Job
71+ inner(t, c, run, cn)
72+ }
73+ h.run(t)
74+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
75+ // The sandbox gets the workflow and contexts, never a token.
76+ if sawJobSpec == nil || sawJobSpec.JobKey != "j" || sawJobSpec.RunnerName != "r1" || sawJobSpec.RunnerEnvironment != "self-hosted" || sawJobSpec.TimeoutMinutes != 5 {
77+ t.Fatalf("%v", sawJobSpec)
78+ }
79+ if strings.Contains(sawJobSpec.String(), "ermitejob_") || strings.Contains(sawJobSpec.String(), "ermiterun_") {
80+ t.Fatal("a token reached the sandbox")
81+ }
82+}
83+
84+func TestPollRetryUsesTheSamePollIDAfterAnError(t *testing.T) {
85+ h := newHarness(t, simpleScript("x"))
86+ h.srv.pollHook = func(n int, req *ermite.FetchJobRequest) (*ermite.FetchJobResponse, error) {
87+ if n <= 2 { // the response is lost twice
88+ return nil, connect.NewError(connect.CodeUnavailable, io.EOF)
89+ }
90+ if n == 3 {
91+ return &ermite.FetchJobResponse{Job: testJob("j1")}, nil
92+ }
93+ return nil, nil
94+ }
95+ h.run(t)
96+ eventually(t, "CompleteJob and the next poll", func() bool {
97+ h.srv.mu.Lock()
98+ defer h.srv.mu.Unlock()
99+ return len(h.srv.completes) == 1 && len(h.srv.polls) >= 4
100+ })
101+ h.srv.mu.Lock()
102+ defer h.srv.mu.Unlock()
103+ if len(h.srv.polls) < 4 {
104+ t.Fatalf("polls %v", h.srv.polls)
105+ }
106+ if h.srv.polls[0] == "" || h.srv.polls[0] != h.srv.polls[1] || h.srv.polls[1] != h.srv.polls[2] {
107+ t.Errorf("a retried poll must keep its poll_id: %v", h.srv.polls)
108+ }
109+ if h.srv.polls[3] == h.srv.polls[2] {
110+ t.Errorf("an answered poll is followed by a new id: %v", h.srv.polls)
111+ }
112+}
113+
114+func TestPollRespectsRetryAfter(t *testing.T) {
115+ h := newHarness(t, simpleScript("x"))
116+ h.srv.retryAfter = 300 * time.Millisecond
117+ h.run(t)
118+ time.Sleep(1000 * time.Millisecond)
119+ h.srv.mu.Lock()
120+ n := h.srv.pollCalls
121+ h.srv.mu.Unlock()
122+ if n < 2 || n > 5 {
123+ t.Fatalf("%d polls in 1 s with retry_after=300ms", n)
124+ }
125+}
126+
127+func TestPollRemovedRunnerStops(t *testing.T) {
128+ h := newHarness(t, simpleScript("x"))
129+ h.runner.Poll = NewClient(h.url, "ermiterun_wrong", http.DefaultTransport)
130+ err := h.runner.Run(context.Background())
131+ if !errors.Is(err, ErrRemoved) {
132+ t.Fatalf("%v", err)
133+ }
134+}
135+
136+func TestNoPollWithoutFreeCapacity(t *testing.T) {
137+ release := make(chan struct{})
138+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
139+ stepStart(c, 1, "Set up job")
140+ <-release
141+ stepDone(c, 1, "Set up job", cOK)
142+ jobDone(c, cOK, nil, false)
143+ })
144+ h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2")}
145+ h.run(t)
146+ eventually(t, "first job started", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.started == 1 })
147+ time.Sleep(300 * time.Millisecond)
148+ h.srv.mu.Lock()
149+ calls := h.srv.pollCalls
150+ h.srv.mu.Unlock()
151+ if calls != 1 {
152+ t.Fatalf("polled %d times with the only slot busy", calls)
153+ }
154+ h.driver.mu.Lock()
155+ if h.driver.started != 1 {
156+ t.Errorf("started=%d", h.driver.started)
157+ }
158+ h.driver.mu.Unlock()
159+ close(release)
160+ eventually(t, "both jobs done", func() bool { return h.srv.completeCount() == 2 })
161+}
162+
163+func TestCapacityTwoRunsTwoJobsAtOnce(t *testing.T) {
164+ release := make(chan struct{})
165+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
166+ stepStart(c, 1, "Set up job")
167+ <-release
168+ stepDone(c, 1, "Set up job", cOK)
169+ jobDone(c, cOK, nil, false)
170+ })
171+ h.runner.Cfg.Capacity = 2
172+ h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2"), testJob("j3")}
173+ h.run(t)
174+ eventually(t, "two sandboxes", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.started == 2 })
175+ time.Sleep(200 * time.Millisecond)
176+ h.driver.mu.Lock()
177+ if h.driver.started != 2 {
178+ t.Errorf("started=%d with capacity 2", h.driver.started)
179+ }
180+ h.driver.mu.Unlock()
181+ close(release)
182+ eventually(t, "three jobs done", func() bool { return h.srv.completeCount() == 3 })
183+}
184+
185+func TestFirstHeartbeatRetriesEvery5sStyle(t *testing.T) {
186+ h := newHarness(t, simpleScript("x"))
187+ h.srv.jobs = []*ermite.Job{testJob("j1")}
188+ h.srv.beatHook = func(n int) error {
189+ if n <= 2 {
190+ return connect.NewError(connect.CodeUnavailable, io.EOF)
191+ }
192+ return nil
193+ }
194+ h.run(t)
195+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
196+ h.srv.mu.Lock()
197+ defer h.srv.mu.Unlock()
198+ if len(h.srv.beats) < 3 {
199+ t.Fatalf("beats %d", len(h.srv.beats))
200+ }
201+ // The job starts only after the acknowledgement: no log before the third beat.
202+ if gap := h.srv.beats[1].Sub(h.srv.beats[0]); gap > 90*time.Millisecond {
203+ t.Errorf("retry after %v, want about HeartbeatRetry (30ms), not the 100ms period", gap)
204+ }
205+}
206+
207+func TestHeartbeatTicksAndRetriesQuickly(t *testing.T) {
208+ release := make(chan struct{})
209+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
210+ stepStart(c, 1, "Set up job")
211+ <-release
212+ stepDone(c, 1, "Set up job", cOK)
213+ jobDone(c, cOK, nil, false)
214+ })
215+ h.srv.jobs = []*ermite.Job{testJob("j1")}
216+ h.srv.beatHook = func(n int) error {
217+ if n == 3 { // one failure in the middle of the job
218+ return connect.NewError(connect.CodeUnavailable, io.EOF)
219+ }
220+ return nil
221+ }
222+ h.run(t)
223+ eventually(t, "five heartbeats", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return len(h.srv.beats) >= 5 })
224+ h.srv.mu.Lock()
225+ b := append([]time.Time(nil), h.srv.beats...)
226+ h.srv.mu.Unlock()
227+ if d := b[2].Sub(b[1]); d < 80*time.Millisecond {
228+ t.Errorf("period %v, want about 100ms", d)
229+ }
230+ if d := b[3].Sub(b[2]); d > 80*time.Millisecond {
231+ t.Errorf("the retry after a failure came after %v, want about 30ms", d)
232+ }
233+ close(release)
234+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
235+}
236+
237+func TestCancelFromWatchJobReachesTheExecutor(t *testing.T) {
238+ got := make(chan *execpb.Cancel, 1)
239+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
240+ stepStart(c, 1, "Set up job")
241+ stepStart(c, 2, "Run sleep")
242+ logLineFrame(c, 2, "working", false)
243+ cn := <-cancels
244+ got <- cn
245+ stepDone(c, 2, "Run sleep", cCancel)
246+ jobDone(c, cCancel, nil, false)
247+ })
248+ h.srv.jobs = []*ermite.Job{testJob("j1")}
249+ h.run(t)
250+ eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
251+ h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Keepalive{Keepalive: &ermite.Keepalive{}}}
252+ h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Cancel{Cancel: &ermite.CancelRequest{Reason: ermite.CancelReason_CANCEL_REASON_USER, Message: "Canceled by the owner"}}}
253+ select {
254+ case cn := <-got:
255+ if cn.Reason != "Canceled by the owner" || cn.Kind != execpb.CancelKind_CANCEL_KIND_USER {
256+ t.Fatalf("%v", cn)
257+ }
258+ case <-time.After(5 * time.Second):
259+ t.Fatal("the cancel did not reach the executor")
260+ }
261+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
262+ c := h.srv.completes[0]
263+ if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || c.RunnerShutdown {
264+ t.Fatalf("%v", c)
265+ }
266+}
267+
268+func TestTimeoutCancelKind(t *testing.T) {
269+ got := make(chan *execpb.Cancel, 1)
270+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
271+ stepStart(c, 1, "Set up job")
272+ logLineFrame(c, 1, "x", false)
273+ got <- <-cancels
274+ jobDone(c, execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT, nil, false)
275+ })
276+ h.srv.jobs = []*ermite.Job{testJob("j1")}
277+ h.run(t)
278+ eventually(t, "log", func() bool { return len(h.srv.snapshotLines()) == 1 })
279+ h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Cancel{Cancel: &ermite.CancelRequest{Reason: ermite.CancelReason_CANCEL_REASON_TIMEOUT}}}
280+ if cn := <-got; cn.Kind != execpb.CancelKind_CANCEL_KIND_TIMEOUT {
281+ t.Fatalf("%v", cn)
282+ }
283+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
284+ if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
285+ t.Fatalf("%v", c.Conclusion)
286+ }
287+}
288+
289+func TestWatchJobIsReopenedAfterAnError(t *testing.T) {
290+ release := make(chan struct{})
291+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
292+ stepStart(c, 1, "Set up job")
293+ <-release
294+ jobDone(c, cOK, nil, false)
295+ })
296+ h.srv.jobs = []*ermite.Job{testJob("j1")}
297+ h.run(t)
298+ eventually(t, "the first watch", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return h.srv.watches >= 1 })
299+ // A message that ends the stream cleanly without Ended: the runner reopens.
300+ // The fake closes the stream only on Ended or context end, so cut the
301+ // transport instead: close idle connections of the test server.
302+ http.DefaultTransport.(*http.Transport).CloseIdleConnections()
303+ close(release)
304+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
305+}
306+
307+func TestLogBatchesRespectTheLimitsAndKeepOrder(t *testing.T) {
308+ var lines []string
309+ big := strings.Repeat("x", 1000)
310+ for i := 0; i < 2500; i++ {
311+ lines = append(lines, big[:1+i%1000])
312+ }
313+ h := newHarness(t, simpleScript(lines...))
314+ h.runner.Timing.LogFlush = 50 * time.Millisecond
315+ h.srv.jobs = []*ermite.Job{testJob("j1")}
316+ h.run(t)
317+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
318+ h.srv.mu.Lock()
319+ defer h.srv.mu.Unlock()
320+ if len(h.srv.lines) != 2500 {
321+ t.Fatalf("the server holds %d lines", len(h.srv.lines))
322+ }
323+ for i, l := range h.srv.lines {
324+ if len(l) != 1+(i)%1000 {
325+ t.Fatalf("line %d out of order or wrong", i)
326+ }
327+ }
328+ var next int64
329+ for _, r := range h.srv.appendReq {
330+ if len(r.Lines) > 1000 {
331+ t.Fatalf("batch of %d lines", len(r.Lines))
332+ }
333+ n := 0
334+ for _, l := range r.Lines {
335+ n += len(l.Text)
336+ }
337+ if n > 64*1024 {
338+ t.Fatalf("batch of %d bytes", n)
339+ }
340+ if r.Offset != next {
341+ t.Fatalf("offset %d, want %d", r.Offset, next)
342+ }
343+ next += int64(len(r.Lines))
344+ }
345+ if next != 2500 {
346+ t.Fatalf("offsets end at %d", next)
347+ }
348+ // Batches never mix steps.
349+ for _, r := range h.srv.appendReq {
350+ if r.Step < 1 {
351+ t.Fatalf("step %d", r.Step)
352+ }
353+ }
354+}
355+
356+func TestLogStepsAreSeparateRequests(t *testing.T) {
357+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
358+ for n := 1; n <= 3; n++ {
359+ stepStart(c, n, "s")
360+ logLineFrame(c, n, "a", false)
361+ logLineFrame(c, n, "b", false)
362+ stepDone(c, n, "s", cOK)
363+ }
364+ jobDone(c, cOK, nil, false)
365+ })
366+ h.runner.Timing.LogFlush = 50 * time.Millisecond
367+ h.srv.jobs = []*ermite.Job{testJob("j1")}
368+ h.run(t)
369+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
370+ h.srv.mu.Lock()
371+ defer h.srv.mu.Unlock()
372+ if len(h.srv.appendReq) != 3 {
373+ t.Fatalf("%d requests, want one per step", len(h.srv.appendReq))
374+ }
375+ for i, r := range h.srv.appendReq {
376+ if r.Step != int32(i+1) || r.Offset != int64(2*i) || len(r.Lines) != 2 {
377+ t.Errorf("request %d: step %d offset %d lines %d", i, r.Step, r.Offset, len(r.Lines))
378+ }
379+ }
380+}
381+
382+func TestContinuedLinesAreForwarded(t *testing.T) {
383+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
384+ stepStart(c, 1, "s")
385+ logLineFrame(c, 1, "first", false)
386+ logLineFrame(c, 1, "piece1", false)
387+ logLineFrame(c, 1, "piece2", true)
388+ stepDone(c, 1, "s", cOK)
389+ jobDone(c, cOK, nil, false)
390+ })
391+ h.srv.jobs = []*ermite.Job{testJob("j1")}
392+ h.run(t)
393+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
394+ h.srv.mu.Lock()
395+ defer h.srv.mu.Unlock()
396+ if len(h.srv.cont) != 3 || h.srv.cont[0] || h.srv.cont[1] || !h.srv.cont[2] {
397+ t.Fatalf("%v", h.srv.cont)
398+ }
399+}
400+
401+func TestAppendLogRetriesAreIdempotent(t *testing.T) {
402+ h := newHarness(t, simpleScript("l1", "l2", "l3"))
403+ h.srv.jobs = []*ermite.Job{testJob("j1")}
404+ // The first AppendLog is applied but its response is lost; the second fails
405+ // before being applied. The runner resends the same lines at the same offset.
406+ h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
407+ switch n {
408+ case 1:
409+ return true, connect.NewError(connect.CodeUnavailable, io.EOF)
410+ case 2:
411+ return false, connect.NewError(connect.CodeInternal, io.EOF)
412+ }
413+ return false, nil
414+ }
415+ h.run(t)
416+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
417+ h.srv.mu.Lock()
418+ defer h.srv.mu.Unlock()
419+ if strings.Join(h.srv.lines, "|") != "l1|l2|l3" {
420+ t.Fatalf("lines %q: a retry must neither lose nor duplicate", h.srv.lines)
421+ }
422+ if h.srv.appendN < 3 {
423+ t.Fatalf("only %d calls", h.srv.appendN)
424+ }
425+ first := h.srv.appendReq[0]
426+ for _, r := range h.srv.appendReq[1:3] {
427+ if r.Offset != first.Offset || len(r.Lines) != len(first.Lines) {
428+ t.Errorf("a retry changed the request: offset %d/%d lines %d/%d", r.Offset, first.Offset, len(r.Lines), len(first.Lines))
429+ }
430+ }
431+}
432+
433+func TestStepUpdatesAreRetried(t *testing.T) {
434+ h := newHarness(t, simpleScript("x"))
435+ h.srv.jobs = []*ermite.Job{testJob("j1")}
436+ h.run(t)
437+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
438+ // UpdateJob snapshots carry the state at each step boundary.
439+ h.srv.mu.Lock()
440+ defer h.srv.mu.Unlock()
441+ if u := h.srv.steps[2]; u == nil || u.State != ermite.StepState_STEP_STATE_COMPLETED {
442+ t.Fatalf("%v", u)
443+ }
444+}
445+
446+func TestCompleteJobIsRetriedAndIdempotent(t *testing.T) {
447+ h := newHarness(t, simpleScript("x"))
448+ h.srv.jobs = []*ermite.Job{testJob("j1")}
449+ h.srv.completeHook = func(n int) (bool, error) {
450+ switch n {
451+ case 1: // applied, response lost
452+ return true, connect.NewError(connect.CodeUnavailable, io.EOF)
453+ case 2: // not even reached
454+ return false, connect.NewError(connect.CodeInternal, io.EOF)
455+ }
456+ return false, nil
457+ }
458+ h.run(t)
459+ eventually(t, "three CompleteJob calls", func() bool { return h.srv.completeCount() == 3 })
460+ time.Sleep(100 * time.Millisecond)
461+ if n := h.srv.completeCount(); n != 3 {
462+ t.Fatalf("%d calls", n)
463+ }
464+ h.srv.mu.Lock()
465+ defer h.srv.mu.Unlock()
466+ a, b := h.srv.completes[0], h.srv.completes[2]
467+ if a.Conclusion != b.Conclusion || a.Outputs["out"] != b.Outputs["out"] || len(a.Steps) != len(b.Steps) {
468+ t.Error("a retry must repeat the same result")
469+ }
470+}
471+
472+func TestJobEndedStopsEverything(t *testing.T) {
473+ release := make(chan struct{})
474+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
475+ stepStart(c, 1, "Set up job")
476+ logLineFrame(c, 1, "before", false)
477+ <-release
478+ logLineFrame(c, 1, "after", false)
479+ stepDone(c, 1, "Set up job", cOK)
480+ jobDone(c, cOK, nil, false)
481+ })
482+ h.srv.jobs = []*ermite.Job{testJob("j1")}
483+ h.run(t)
484+ eventually(t, "first line", func() bool { return len(h.srv.snapshotLines()) == 1 })
485+ // The server ends the job (lost runner, log limit, ...): every call now says so.
486+ h.srv.mu.Lock()
487+ h.srv.ended = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, Error: "ended by the server"}
488+ h.srv.mu.Unlock()
489+ h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Ended{Ended: &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED}}}
490+ eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
491+ close(release)
492+ time.Sleep(300 * time.Millisecond)
493+ if n := h.srv.completeCount(); n != 0 {
494+ t.Fatalf("CompleteJob was called %d times for a job the server ended", n)
495+ }
496+ if got := h.srv.snapshotLines(); len(got) != 1 {
497+ t.Fatalf("lines after the end: %q", got)
498+ }
499+}
500+
501+func TestJobEndedFromAppendLogAbandonsTheJob(t *testing.T) {
502+ release := make(chan struct{})
503+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
504+ stepStart(c, 1, "Set up job")
505+ logLineFrame(c, 1, "a", false)
506+ <-release
507+ jobDone(c, cOK, nil, false)
508+ })
509+ h.srv.jobs = []*ermite.Job{testJob("j1")}
510+ h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
511+ return false, endedErr(&ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE, Error: "log limit reached"})
512+ }
513+ h.run(t)
514+ eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
515+ close(release)
516+ time.Sleep(200 * time.Millisecond)
517+ if n := h.srv.completeCount(); n != 0 {
518+ t.Fatalf("CompleteJob was called %d times", n)
519+ }
520+}
521+
522+func TestUnauthenticatedJobCallAbandonsAtOnce(t *testing.T) {
523+ release := make(chan struct{})
524+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
525+ stepStart(c, 1, "Set up job")
526+ <-release
527+ jobDone(c, cOK, nil, false)
528+ })
529+ h.srv.jobs = []*ermite.Job{testJob("j1")}
530+ h.srv.beatHook = func(n int) error {
531+ if n >= 2 { // the assignment was replaced
532+ return connect.NewError(connect.CodeUnauthenticated, io.EOF)
533+ }
534+ return nil
535+ }
536+ h.run(t)
537+ eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
538+ close(release)
539+ time.Sleep(200 * time.Millisecond)
540+ if n := h.srv.completeCount(); n != 0 {
541+ t.Fatalf("CompleteJob was called %d times", n)
542+ }
543+}
544+
545+func TestGracefulShutdownReportsRunnerShutdown(t *testing.T) {
546+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
547+ stepStart(c, 1, "Set up job")
548+ stepStart(c, 2, "Run sleep")
549+ logLineFrame(c, 2, "working", false)
550+ cn := <-cancels
551+ if cn.Kind != execpb.CancelKind_CANCEL_KIND_SHUTDOWN {
552+ t.Errorf("kind %v", cn.Kind)
553+ }
554+ stepDone(c, 2, "Run sleep", cCancel)
555+ jobDone(c, cCancel, nil, true)
556+ })
557+ h.srv.jobs = []*ermite.Job{testJob("j1")}
558+ stop := h.run(t)
559+ eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
560+ if err := stop(); err != nil {
561+ t.Fatal(err)
562+ }
563+ h.srv.mu.Lock()
564+ defer h.srv.mu.Unlock()
565+ if len(h.srv.completes) != 1 {
566+ t.Fatalf("%d CompleteJob calls", len(h.srv.completes))
567+ }
568+ if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || !c.RunnerShutdown {
569+ t.Fatalf("%+v", c)
570+ }
571+ if h.driver.stops != 1 {
572+ t.Errorf("stops=%d", h.driver.stops)
573+ }
574+}
575+
576+func TestShutdownHardStopsAnExecutorThatIgnoresIt(t *testing.T) {
577+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
578+ stepStart(c, 1, "Set up job")
579+ stepStart(c, 2, "Run hang")
580+ logLineFrame(c, 2, "hung", false)
581+ <-cancels // received, ignored: keep running until the stream is cut
582+ select {}
583+ })
584+ h.runner.Timing.ShutdownGrace = 300 * time.Millisecond
585+ h.srv.jobs = []*ermite.Job{testJob("j1")}
586+ stop := h.run(t)
587+ eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
588+ start := time.Now()
589+ _ = stop()
590+ if time.Since(start) > 5*time.Second {
591+ t.Fatal("shutdown waited for an executor that does not answer")
592+ }
593+ h.srv.mu.Lock()
594+ defer h.srv.mu.Unlock()
595+ if len(h.srv.completes) != 1 || h.srv.completes[0].Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || !h.srv.completes[0].RunnerShutdown {
596+ t.Fatalf("%+v", h.srv.completes)
597+ }
598+ // Steps that never completed are closed by the runner.
599+ for _, u := range h.srv.completes[0].Steps {
600+ if u.State != ermite.StepState_STEP_STATE_COMPLETED || u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED {
601+ t.Errorf("step %d: %v/%v", u.Number, u.State, u.Conclusion)
602+ }
603+ }
604+}
605+
606+func TestExecutorDyingFailsTheJob(t *testing.T) {
607+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
608+ stepStart(c, 1, "Set up job")
609+ logLineFrame(c, 1, "partial", false)
610+ // the stream closes without a result
611+ })
612+ h.srv.jobs = []*ermite.Job{testJob("j1")}
613+ h.run(t)
614+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
615+ h.srv.mu.Lock()
616+ defer h.srv.mu.Unlock()
617+ c := h.srv.completes[0]
618+ if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE || c.RunnerShutdown {
619+ t.Fatalf("%+v", c)
620+ }
621+ if len(h.srv.lines) != 2 || !strings.HasPrefix(h.srv.lines[1], "##[error]The job's sandbox ended unexpectedly") {
622+ t.Fatalf("lines %q", h.srv.lines)
623+ }
624+ if len(c.Steps) != 1 || c.Steps[0].State != ermite.StepState_STEP_STATE_COMPLETED || c.Steps[0].Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
625+ t.Fatalf("%v", c.Steps)
626+ }
627+}
628+
629+func TestSandboxStartFailureFailsTheJobInStepOne(t *testing.T) {
630+ h := newHarness(t, simpleScript("x"))
631+ h.driver.startErr = errors.New("no kvm today")
632+ h.srv.jobs = []*ermite.Job{testJob("j1")}
633+ h.run(t)
634+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
635+ h.srv.mu.Lock()
636+ defer h.srv.mu.Unlock()
637+ if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
638+ t.Fatalf("%v", c.Conclusion)
639+ }
640+ if len(h.srv.lines) != 1 || !strings.Contains(h.srv.lines[0], "The sandbox did not start: no kvm today") {
641+ t.Fatalf("%q", h.srv.lines)
642+ }
643+ if h.driver.stops != 1 {
644+ t.Errorf("a sandbox that failed to start is still stopped: stops=%d", h.driver.stops)
645+ }
646+}
647+
648+func TestOutputsOverTheLimitFailTheJob(t *testing.T) {
649+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
650+ stepStart(c, 1, "Set up job")
651+ stepDone(c, 1, "Set up job", cOK)
652+ // Names that ermite would refuse: the job fails without outputs.
653+ jobDone(c, cOK, map[string]string{"1bad": "x", "fine": "y"}, false)
654+ })
655+ h.srv.jobs = []*ermite.Job{testJob("j1")}
656+ h.run(t)
657+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
658+ h.srv.mu.Lock()
659+ defer h.srv.mu.Unlock()
660+ c := h.srv.completes[0]
661+ if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE || len(c.Outputs) != 0 {
662+ t.Fatalf("%v %v", c.Conclusion, len(c.Outputs))
663+ }
664+ if len(h.srv.lines) != 1 || !strings.Contains(h.srv.lines[0], "is not valid") {
665+ t.Fatalf("%q", h.srv.lines)
666+ }
667+}
668+
669+func TestCheckOutputs(t *testing.T) {
670+ for _, tc := range []struct {
671+ in map[string]string
672+ want string
673+ }{
674+ {nil, ""},
675+ {map[string]string{"a-b_c1": "v", "_x": ""}, ""},
676+ {map[string]string{"1a": "v"}, "not valid"},
677+ {map[string]string{"a b": "v"}, "not valid"},
678+ {map[string]string{"": "v"}, "not valid"},
679+ {map[string]string{strings.Repeat("a", 256): "v"}, "not valid"},
680+ {map[string]string{"Foo": "1", "foo": "2"}, "differ only by case"},
681+ {map[string]string{"a": strings.Repeat("x", 1<<20+1)}, "larger than 1 MiB"},
682+ {map[string]string{"a": strings.Repeat("x", 600000), "b": strings.Repeat("x", 600000)}, "in total"},
683+ } {
684+ got := checkOutputs(tc.in)
685+ if (tc.want == "") != (got == "") || !strings.Contains(got, tc.want) {
686+ t.Errorf("%v: %q, want %q", tc.in, got, tc.want)
687+ }
688+ }
689+}
690+
691+func TestLeakedSandboxStopsTakingJobsUntilReaped(t *testing.T) {
692+ h := newHarness(t, simpleScript("x"))
693+ h.driver.stopErr = sandbox.ErrLeaked
694+ h.driver.reapErr = errors.New("still leaked")
695+ h.runner.Timing.ReapRetry = 50 * time.Millisecond
696+ h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2")}
697+ h.run(t)
698+ eventually(t, "first job done", func() bool { return h.srv.completeCount() == 1 })
699+ time.Sleep(400 * time.Millisecond)
700+ if h.srv.completeCount() != 1 {
701+ t.Fatal("a second job started while a sandbox is leaked and cannot be reaped")
702+ }
703+ // Once the reap works, jobs are taken again.
704+ h.driver.mu.Lock()
705+ h.driver.reapErr = nil
706+ h.driver.stopErr = nil
707+ h.driver.mu.Unlock()
708+ eventually(t, "the second job", func() bool { return h.srv.completeCount() == 2 })
709+}
710+
711+func TestReapAtStartAndIsolation(t *testing.T) {
712+ h := newHarness(t, simpleScript("x"))
713+ h.run(t)
714+ eventually(t, "reap", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.reaps >= 1 })
715+ for in, want := range map[sandbox.Isolation]ermite.RunnerIsolation{
716+ sandbox.IsolationVM: ermite.RunnerIsolation_RUNNER_ISOLATION_VM,
717+ sandbox.IsolationContainerSysbox: ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX,
718+ sandbox.IsolationContainerPrivileged: ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED,
719+ "other": ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED,
720+ } {
721+ if got := Isolation(in); got != want {
722+ t.Errorf("%s: %v", in, got)
723+ }
724+ }
725+}
726+
727+func TestRegisterStoresTheToken(t *testing.T) {
728+ srv := newFakeServer()
729+ url := srv.start(t)
730+ dir := t.TempDir() + "/state"
731+ cfg := Config{URL: url, Name: "r1", Labels: []string{"ubuntu-latest"}, StateDir: dir, Version: "v1"}
732+ st, err := Register(context.Background(), cfg, NewClient(url, "reg-token", http.DefaultTransport), sandbox.IsolationContainerSysbox, true)
733+ if err != nil {
734+ t.Fatal(err)
735+ }
736+ if st.RunnerToken != "ermiterun_new" || st.RunnerID != "rid1" {
737+ t.Fatalf("%+v", st)
738+ }
739+ req := srv.registers[0]
740+ if req.Name != "r1" || req.Isolation != ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX || req.Os != "linux" || (req.Arch != "x64" && req.Arch != "arm64") || !req.Replace || req.Version != "v1" || len(req.Labels) != 1 {
741+ t.Fatalf("%v", req)
742+ }
743+ fi, err := os.Stat(StateFile(dir))
744+ if err != nil || fi.Mode().Perm() != 0o600 {
745+ t.Fatalf("%v %v", fi, err)
746+ }
747+ got, err := LoadState(dir)
748+ if err != nil || got != st {
749+ t.Fatalf("%+v %v", got, err)
750+ }
751+ // Anyone can read a state file with a loose mode: refuse it.
752+ if err := os.Chmod(StateFile(dir), 0o644); err != nil {
753+ t.Fatal(err)
754+ }
755+ if _, err := LoadState(dir); err == nil || !strings.Contains(err.Error(), "accessible by other users") {
756+ t.Fatalf("%v", err)
757+ }
758+ if _, err := LoadState(t.TempDir()); !errors.Is(err, ErrNotRegistered) {
759+ t.Fatalf("%v", err)
760+ }
761+ // A refused registration stores nothing.
762+ dir2 := t.TempDir()
763+ cfg.StateDir = dir2
764+ if _, err := Register(context.Background(), cfg, NewClient(url, "bad", http.DefaultTransport), sandbox.IsolationVM, false); err == nil {
765+ t.Fatal("registration with a bad token succeeded")
766+ }
767+ if _, err := os.Stat(StateFile(dir2)); err == nil {
768+ t.Fatal("a state file was written")
769+ }
770+}
771+
772+func TestLoadConfig(t *testing.T) {
773+ env := map[string]string{EnvURL: " https://ermite.example ", EnvLabels: "ubuntu-latest, Ubuntu-24.04 ,ubuntu-latest,"}
774+ c, err := LoadConfig(func(k string) string { return env[k] }, "my host.local", "/state", 2, "v1")
775+ if err != nil {
776+ t.Fatal(err)
777+ }
778+ if c.URL != "https://ermite.example" || c.Name != "my-host.local" || strings.Join(c.Labels, ",") != "ubuntu-latest,Ubuntu-24.04" || c.Capacity != 2 {
779+ t.Fatalf("%+v", c)
780+ }
781+ env[EnvName] = "bad name"
782+ if _, err := LoadConfig(func(k string) string { return env[k] }, "h", "/s", 1, ""); err == nil {
783+ t.Error("an invalid name was accepted")
784+ }
785+ env[EnvName] = "ok"
786+ env[EnvLabels] = "good,bad label"
787+ if _, err := LoadConfig(func(k string) string { return env[k] }, "h", "/s", 1, ""); err == nil {
788+ t.Error("an invalid label was accepted")
789+ }
790+}
791+
792+var _ = ermiteconnect.RunnerServiceClient(nil)
@@ -0,0 +1,792 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "errors"
8+ "io"
9+ "net/http"
10+ "os"
11+ "strings"
12+ "testing"
13+ "time"
14+
15+ "connectrpc.com/connect"
16+
17+ ermite "bnema.dev/ermite-api/gen/ermite"
18+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
19+
20+ "bnema.dev/ermite-runner/internal/execstream"
21+ "bnema.dev/ermite-runner/internal/execstream/execpb"
22+ "bnema.dev/ermite-runner/internal/sandbox"
23+)
24+
25+func TestJobRunsAndReports(t *testing.T) {
26+ h := newHarness(t, simpleScript("hello", "world"))
27+ h.srv.jobs = []*ermite.Job{testJob("j1")}
28+ stop := h.run(t)
29+
30+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
31+ if err := stop(); err != nil {
32+ t.Fatal(err)
33+ }
34+ s := h.srv
35+ s.mu.Lock()
36+ defer s.mu.Unlock()
37+ c := s.completes[0]
38+ if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS || c.Outputs["out"] != "v" || c.RunnerShutdown {
39+ t.Fatalf("%+v", c)
40+ }
41+ if len(s.beats) == 0 {
42+ t.Error("no heartbeat: the first one is the acknowledgement")
43+ }
44+ if strings.Join(s.lines, "|") != "hello|world" {
45+ t.Errorf("lines %q", s.lines)
46+ }
47+ for n, name := range map[int32]string{1: "Set up job", 2: "Run it", 3: "Complete job"} {
48+ u := s.steps[n]
49+ if u == nil || u.Name != name || u.State != ermite.StepState_STEP_STATE_COMPLETED || u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS || u.StartedAt == nil || u.CompletedAt == nil {
50+ t.Errorf("step %d: %v", n, u)
51+ }
52+ }
53+ if len(c.Steps) != 3 {
54+ t.Errorf("CompleteJob carries %d final steps", len(c.Steps))
55+ }
56+ if h.driver.stops != 1 || h.driver.reaps < 1 {
57+ t.Errorf("stops=%d reaps=%d: the sandbox must be stopped and the start must reap", h.driver.stops, h.driver.reaps)
58+ }
59+ if strings.Contains(h.logs.String(), "ermitejob_") || strings.Contains(h.logs.String(), "ermiterun_") {
60+ t.Errorf("a token reached the log: %s", h.logs.String())
61+ }
62+}
63+
64+func TestTheJobTokenAndTheRunnerTokenStayApart(t *testing.T) {
65+ h := newHarness(t, simpleScript("x"))
66+ h.srv.jobs = []*ermite.Job{testJob("j1")}
67+ var sawJobSpec *execpb.JobSpec
68+ inner := h.driver.script
69+ h.driver.script = func(t *testing.T, c *execstream.Conn, run *execpb.Run, cn <-chan *execpb.Cancel) {
70+ sawJobSpec = run.Job
71+ inner(t, c, run, cn)
72+ }
73+ h.run(t)
74+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
75+ // The sandbox gets the workflow and contexts, never a token.
76+ if sawJobSpec == nil || sawJobSpec.JobKey != "j" || sawJobSpec.RunnerName != "r1" || sawJobSpec.RunnerEnvironment != "self-hosted" || sawJobSpec.TimeoutMinutes != 5 {
77+ t.Fatalf("%v", sawJobSpec)
78+ }
79+ if strings.Contains(sawJobSpec.String(), "ermitejob_") || strings.Contains(sawJobSpec.String(), "ermiterun_") {
80+ t.Fatal("a token reached the sandbox")
81+ }
82+}
83+
84+func TestPollRetryUsesTheSamePollIDAfterAnError(t *testing.T) {
85+ h := newHarness(t, simpleScript("x"))
86+ h.srv.pollHook = func(n int, req *ermite.FetchJobRequest) (*ermite.FetchJobResponse, error) {
87+ if n <= 2 { // the response is lost twice
88+ return nil, connect.NewError(connect.CodeUnavailable, io.EOF)
89+ }
90+ if n == 3 {
91+ return &ermite.FetchJobResponse{Job: testJob("j1")}, nil
92+ }
93+ return nil, nil
94+ }
95+ h.run(t)
96+ eventually(t, "CompleteJob and the next poll", func() bool {
97+ h.srv.mu.Lock()
98+ defer h.srv.mu.Unlock()
99+ return len(h.srv.completes) == 1 && len(h.srv.polls) >= 4
100+ })
101+ h.srv.mu.Lock()
102+ defer h.srv.mu.Unlock()
103+ if len(h.srv.polls) < 4 {
104+ t.Fatalf("polls %v", h.srv.polls)
105+ }
106+ if h.srv.polls[0] == "" || h.srv.polls[0] != h.srv.polls[1] || h.srv.polls[1] != h.srv.polls[2] {
107+ t.Errorf("a retried poll must keep its poll_id: %v", h.srv.polls)
108+ }
109+ if h.srv.polls[3] == h.srv.polls[2] {
110+ t.Errorf("an answered poll is followed by a new id: %v", h.srv.polls)
111+ }
112+}
113+
114+func TestPollRespectsRetryAfter(t *testing.T) {
115+ h := newHarness(t, simpleScript("x"))
116+ h.srv.retryAfter = 300 * time.Millisecond
117+ h.run(t)
118+ time.Sleep(1000 * time.Millisecond)
119+ h.srv.mu.Lock()
120+ n := h.srv.pollCalls
121+ h.srv.mu.Unlock()
122+ if n < 2 || n > 5 {
123+ t.Fatalf("%d polls in 1 s with retry_after=300ms", n)
124+ }
125+}
126+
127+func TestPollRemovedRunnerStops(t *testing.T) {
128+ h := newHarness(t, simpleScript("x"))
129+ h.runner.Poll = NewClient(h.url, "ermiterun_wrong", http.DefaultTransport)
130+ err := h.runner.Run(context.Background())
131+ if !errors.Is(err, ErrRemoved) {
132+ t.Fatalf("%v", err)
133+ }
134+}
135+
136+func TestNoPollWithoutFreeCapacity(t *testing.T) {
137+ release := make(chan struct{})
138+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
139+ stepStart(c, 1, "Set up job")
140+ <-release
141+ stepDone(c, 1, "Set up job", cOK)
142+ jobDone(c, cOK, nil, false)
143+ })
144+ h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2")}
145+ h.run(t)
146+ eventually(t, "first job started", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.started == 1 })
147+ time.Sleep(300 * time.Millisecond)
148+ h.srv.mu.Lock()
149+ calls := h.srv.pollCalls
150+ h.srv.mu.Unlock()
151+ if calls != 1 {
152+ t.Fatalf("polled %d times with the only slot busy", calls)
153+ }
154+ h.driver.mu.Lock()
155+ if h.driver.started != 1 {
156+ t.Errorf("started=%d", h.driver.started)
157+ }
158+ h.driver.mu.Unlock()
159+ close(release)
160+ eventually(t, "both jobs done", func() bool { return h.srv.completeCount() == 2 })
161+}
162+
163+func TestCapacityTwoRunsTwoJobsAtOnce(t *testing.T) {
164+ release := make(chan struct{})
165+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
166+ stepStart(c, 1, "Set up job")
167+ <-release
168+ stepDone(c, 1, "Set up job", cOK)
169+ jobDone(c, cOK, nil, false)
170+ })
171+ h.runner.Cfg.Capacity = 2
172+ h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2"), testJob("j3")}
173+ h.run(t)
174+ eventually(t, "two sandboxes", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.started == 2 })
175+ time.Sleep(200 * time.Millisecond)
176+ h.driver.mu.Lock()
177+ if h.driver.started != 2 {
178+ t.Errorf("started=%d with capacity 2", h.driver.started)
179+ }
180+ h.driver.mu.Unlock()
181+ close(release)
182+ eventually(t, "three jobs done", func() bool { return h.srv.completeCount() == 3 })
183+}
184+
185+func TestFirstHeartbeatRetriesEvery5sStyle(t *testing.T) {
186+ h := newHarness(t, simpleScript("x"))
187+ h.srv.jobs = []*ermite.Job{testJob("j1")}
188+ h.srv.beatHook = func(n int) error {
189+ if n <= 2 {
190+ return connect.NewError(connect.CodeUnavailable, io.EOF)
191+ }
192+ return nil
193+ }
194+ h.run(t)
195+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
196+ h.srv.mu.Lock()
197+ defer h.srv.mu.Unlock()
198+ if len(h.srv.beats) < 3 {
199+ t.Fatalf("beats %d", len(h.srv.beats))
200+ }
201+ // The job starts only after the acknowledgement: no log before the third beat.
202+ if gap := h.srv.beats[1].Sub(h.srv.beats[0]); gap > 90*time.Millisecond {
203+ t.Errorf("retry after %v, want about HeartbeatRetry (30ms), not the 100ms period", gap)
204+ }
205+}
206+
207+func TestHeartbeatTicksAndRetriesQuickly(t *testing.T) {
208+ release := make(chan struct{})
209+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
210+ stepStart(c, 1, "Set up job")
211+ <-release
212+ stepDone(c, 1, "Set up job", cOK)
213+ jobDone(c, cOK, nil, false)
214+ })
215+ h.srv.jobs = []*ermite.Job{testJob("j1")}
216+ h.srv.beatHook = func(n int) error {
217+ if n == 3 { // one failure in the middle of the job
218+ return connect.NewError(connect.CodeUnavailable, io.EOF)
219+ }
220+ return nil
221+ }
222+ h.run(t)
223+ eventually(t, "five heartbeats", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return len(h.srv.beats) >= 5 })
224+ h.srv.mu.Lock()
225+ b := append([]time.Time(nil), h.srv.beats...)
226+ h.srv.mu.Unlock()
227+ if d := b[2].Sub(b[1]); d < 80*time.Millisecond {
228+ t.Errorf("period %v, want about 100ms", d)
229+ }
230+ if d := b[3].Sub(b[2]); d > 80*time.Millisecond {
231+ t.Errorf("the retry after a failure came after %v, want about 30ms", d)
232+ }
233+ close(release)
234+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
235+}
236+
237+func TestCancelFromWatchJobReachesTheExecutor(t *testing.T) {
238+ got := make(chan *execpb.Cancel, 1)
239+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
240+ stepStart(c, 1, "Set up job")
241+ stepStart(c, 2, "Run sleep")
242+ logLineFrame(c, 2, "working", false)
243+ cn := <-cancels
244+ got <- cn
245+ stepDone(c, 2, "Run sleep", cCancel)
246+ jobDone(c, cCancel, nil, false)
247+ })
248+ h.srv.jobs = []*ermite.Job{testJob("j1")}
249+ h.run(t)
250+ eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
251+ h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Keepalive{Keepalive: &ermite.Keepalive{}}}
252+ h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Cancel{Cancel: &ermite.CancelRequest{Reason: ermite.CancelReason_CANCEL_REASON_USER, Message: "Canceled by the owner"}}}
253+ select {
254+ case cn := <-got:
255+ if cn.Reason != "Canceled by the owner" || cn.Kind != execpb.CancelKind_CANCEL_KIND_USER {
256+ t.Fatalf("%v", cn)
257+ }
258+ case <-time.After(5 * time.Second):
259+ t.Fatal("the cancel did not reach the executor")
260+ }
261+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
262+ c := h.srv.completes[0]
263+ if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || c.RunnerShutdown {
264+ t.Fatalf("%v", c)
265+ }
266+}
267+
268+func TestTimeoutCancelKind(t *testing.T) {
269+ got := make(chan *execpb.Cancel, 1)
270+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
271+ stepStart(c, 1, "Set up job")
272+ logLineFrame(c, 1, "x", false)
273+ got <- <-cancels
274+ jobDone(c, execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT, nil, false)
275+ })
276+ h.srv.jobs = []*ermite.Job{testJob("j1")}
277+ h.run(t)
278+ eventually(t, "log", func() bool { return len(h.srv.snapshotLines()) == 1 })
279+ h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Cancel{Cancel: &ermite.CancelRequest{Reason: ermite.CancelReason_CANCEL_REASON_TIMEOUT}}}
280+ if cn := <-got; cn.Kind != execpb.CancelKind_CANCEL_KIND_TIMEOUT {
281+ t.Fatalf("%v", cn)
282+ }
283+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
284+ if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
285+ t.Fatalf("%v", c.Conclusion)
286+ }
287+}
288+
289+func TestWatchJobIsReopenedAfterAnError(t *testing.T) {
290+ release := make(chan struct{})
291+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
292+ stepStart(c, 1, "Set up job")
293+ <-release
294+ jobDone(c, cOK, nil, false)
295+ })
296+ h.srv.jobs = []*ermite.Job{testJob("j1")}
297+ h.run(t)
298+ eventually(t, "the first watch", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return h.srv.watches >= 1 })
299+ // A message that ends the stream cleanly without Ended: the runner reopens.
300+ // The fake closes the stream only on Ended or context end, so cut the
301+ // transport instead: close idle connections of the test server.
302+ http.DefaultTransport.(*http.Transport).CloseIdleConnections()
303+ close(release)
304+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
305+}
306+
307+func TestLogBatchesRespectTheLimitsAndKeepOrder(t *testing.T) {
308+ var lines []string
309+ big := strings.Repeat("x", 1000)
310+ for i := 0; i < 2500; i++ {
311+ lines = append(lines, big[:1+i%1000])
312+ }
313+ h := newHarness(t, simpleScript(lines...))
314+ h.runner.Timing.LogFlush = 50 * time.Millisecond
315+ h.srv.jobs = []*ermite.Job{testJob("j1")}
316+ h.run(t)
317+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
318+ h.srv.mu.Lock()
319+ defer h.srv.mu.Unlock()
320+ if len(h.srv.lines) != 2500 {
321+ t.Fatalf("the server holds %d lines", len(h.srv.lines))
322+ }
323+ for i, l := range h.srv.lines {
324+ if len(l) != 1+(i)%1000 {
325+ t.Fatalf("line %d out of order or wrong", i)
326+ }
327+ }
328+ var next int64
329+ for _, r := range h.srv.appendReq {
330+ if len(r.Lines) > 1000 {
331+ t.Fatalf("batch of %d lines", len(r.Lines))
332+ }
333+ n := 0
334+ for _, l := range r.Lines {
335+ n += len(l.Text)
336+ }
337+ if n > 64*1024 {
338+ t.Fatalf("batch of %d bytes", n)
339+ }
340+ if r.Offset != next {
341+ t.Fatalf("offset %d, want %d", r.Offset, next)
342+ }
343+ next += int64(len(r.Lines))
344+ }
345+ if next != 2500 {
346+ t.Fatalf("offsets end at %d", next)
347+ }
348+ // Batches never mix steps.
349+ for _, r := range h.srv.appendReq {
350+ if r.Step < 1 {
351+ t.Fatalf("step %d", r.Step)
352+ }
353+ }
354+}
355+
356+func TestLogStepsAreSeparateRequests(t *testing.T) {
357+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
358+ for n := 1; n <= 3; n++ {
359+ stepStart(c, n, "s")
360+ logLineFrame(c, n, "a", false)
361+ logLineFrame(c, n, "b", false)
362+ stepDone(c, n, "s", cOK)
363+ }
364+ jobDone(c, cOK, nil, false)
365+ })
366+ h.runner.Timing.LogFlush = 50 * time.Millisecond
367+ h.srv.jobs = []*ermite.Job{testJob("j1")}
368+ h.run(t)
369+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
370+ h.srv.mu.Lock()
371+ defer h.srv.mu.Unlock()
372+ if len(h.srv.appendReq) != 3 {
373+ t.Fatalf("%d requests, want one per step", len(h.srv.appendReq))
374+ }
375+ for i, r := range h.srv.appendReq {
376+ if r.Step != int32(i+1) || r.Offset != int64(2*i) || len(r.Lines) != 2 {
377+ t.Errorf("request %d: step %d offset %d lines %d", i, r.Step, r.Offset, len(r.Lines))
378+ }
379+ }
380+}
381+
382+func TestContinuedLinesAreForwarded(t *testing.T) {
383+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
384+ stepStart(c, 1, "s")
385+ logLineFrame(c, 1, "first", false)
386+ logLineFrame(c, 1, "piece1", false)
387+ logLineFrame(c, 1, "piece2", true)
388+ stepDone(c, 1, "s", cOK)
389+ jobDone(c, cOK, nil, false)
390+ })
391+ h.srv.jobs = []*ermite.Job{testJob("j1")}
392+ h.run(t)
393+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
394+ h.srv.mu.Lock()
395+ defer h.srv.mu.Unlock()
396+ if len(h.srv.cont) != 3 || h.srv.cont[0] || h.srv.cont[1] || !h.srv.cont[2] {
397+ t.Fatalf("%v", h.srv.cont)
398+ }
399+}
400+
401+func TestAppendLogRetriesAreIdempotent(t *testing.T) {
402+ h := newHarness(t, simpleScript("l1", "l2", "l3"))
403+ h.srv.jobs = []*ermite.Job{testJob("j1")}
404+ // The first AppendLog is applied but its response is lost; the second fails
405+ // before being applied. The runner resends the same lines at the same offset.
406+ h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
407+ switch n {
408+ case 1:
409+ return true, connect.NewError(connect.CodeUnavailable, io.EOF)
410+ case 2:
411+ return false, connect.NewError(connect.CodeInternal, io.EOF)
412+ }
413+ return false, nil
414+ }
415+ h.run(t)
416+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
417+ h.srv.mu.Lock()
418+ defer h.srv.mu.Unlock()
419+ if strings.Join(h.srv.lines, "|") != "l1|l2|l3" {
420+ t.Fatalf("lines %q: a retry must neither lose nor duplicate", h.srv.lines)
421+ }
422+ if h.srv.appendN < 3 {
423+ t.Fatalf("only %d calls", h.srv.appendN)
424+ }
425+ first := h.srv.appendReq[0]
426+ for _, r := range h.srv.appendReq[1:3] {
427+ if r.Offset != first.Offset || len(r.Lines) != len(first.Lines) {
428+ t.Errorf("a retry changed the request: offset %d/%d lines %d/%d", r.Offset, first.Offset, len(r.Lines), len(first.Lines))
429+ }
430+ }
431+}
432+
433+func TestStepUpdatesAreRetried(t *testing.T) {
434+ h := newHarness(t, simpleScript("x"))
435+ h.srv.jobs = []*ermite.Job{testJob("j1")}
436+ h.run(t)
437+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
438+ // UpdateJob snapshots carry the state at each step boundary.
439+ h.srv.mu.Lock()
440+ defer h.srv.mu.Unlock()
441+ if u := h.srv.steps[2]; u == nil || u.State != ermite.StepState_STEP_STATE_COMPLETED {
442+ t.Fatalf("%v", u)
443+ }
444+}
445+
446+func TestCompleteJobIsRetriedAndIdempotent(t *testing.T) {
447+ h := newHarness(t, simpleScript("x"))
448+ h.srv.jobs = []*ermite.Job{testJob("j1")}
449+ h.srv.completeHook = func(n int) (bool, error) {
450+ switch n {
451+ case 1: // applied, response lost
452+ return true, connect.NewError(connect.CodeUnavailable, io.EOF)
453+ case 2: // not even reached
454+ return false, connect.NewError(connect.CodeInternal, io.EOF)
455+ }
456+ return false, nil
457+ }
458+ h.run(t)
459+ eventually(t, "three CompleteJob calls", func() bool { return h.srv.completeCount() == 3 })
460+ time.Sleep(100 * time.Millisecond)
461+ if n := h.srv.completeCount(); n != 3 {
462+ t.Fatalf("%d calls", n)
463+ }
464+ h.srv.mu.Lock()
465+ defer h.srv.mu.Unlock()
466+ a, b := h.srv.completes[0], h.srv.completes[2]
467+ if a.Conclusion != b.Conclusion || a.Outputs["out"] != b.Outputs["out"] || len(a.Steps) != len(b.Steps) {
468+ t.Error("a retry must repeat the same result")
469+ }
470+}
471+
472+func TestJobEndedStopsEverything(t *testing.T) {
473+ release := make(chan struct{})
474+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
475+ stepStart(c, 1, "Set up job")
476+ logLineFrame(c, 1, "before", false)
477+ <-release
478+ logLineFrame(c, 1, "after", false)
479+ stepDone(c, 1, "Set up job", cOK)
480+ jobDone(c, cOK, nil, false)
481+ })
482+ h.srv.jobs = []*ermite.Job{testJob("j1")}
483+ h.run(t)
484+ eventually(t, "first line", func() bool { return len(h.srv.snapshotLines()) == 1 })
485+ // The server ends the job (lost runner, log limit, ...): every call now says so.
486+ h.srv.mu.Lock()
487+ h.srv.ended = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, Error: "ended by the server"}
488+ h.srv.mu.Unlock()
489+ h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Ended{Ended: &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED}}}
490+ eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
491+ close(release)
492+ time.Sleep(300 * time.Millisecond)
493+ if n := h.srv.completeCount(); n != 0 {
494+ t.Fatalf("CompleteJob was called %d times for a job the server ended", n)
495+ }
496+ if got := h.srv.snapshotLines(); len(got) != 1 {
497+ t.Fatalf("lines after the end: %q", got)
498+ }
499+}
500+
501+func TestJobEndedFromAppendLogAbandonsTheJob(t *testing.T) {
502+ release := make(chan struct{})
503+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
504+ stepStart(c, 1, "Set up job")
505+ logLineFrame(c, 1, "a", false)
506+ <-release
507+ jobDone(c, cOK, nil, false)
508+ })
509+ h.srv.jobs = []*ermite.Job{testJob("j1")}
510+ h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
511+ return false, endedErr(&ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE, Error: "log limit reached"})
512+ }
513+ h.run(t)
514+ eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
515+ close(release)
516+ time.Sleep(200 * time.Millisecond)
517+ if n := h.srv.completeCount(); n != 0 {
518+ t.Fatalf("CompleteJob was called %d times", n)
519+ }
520+}
521+
522+func TestUnauthenticatedJobCallAbandonsAtOnce(t *testing.T) {
523+ release := make(chan struct{})
524+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
525+ stepStart(c, 1, "Set up job")
526+ <-release
527+ jobDone(c, cOK, nil, false)
528+ })
529+ h.srv.jobs = []*ermite.Job{testJob("j1")}
530+ h.srv.beatHook = func(n int) error {
531+ if n >= 2 { // the assignment was replaced
532+ return connect.NewError(connect.CodeUnauthenticated, io.EOF)
533+ }
534+ return nil
535+ }
536+ h.run(t)
537+ eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
538+ close(release)
539+ time.Sleep(200 * time.Millisecond)
540+ if n := h.srv.completeCount(); n != 0 {
541+ t.Fatalf("CompleteJob was called %d times", n)
542+ }
543+}
544+
545+func TestGracefulShutdownReportsRunnerShutdown(t *testing.T) {
546+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
547+ stepStart(c, 1, "Set up job")
548+ stepStart(c, 2, "Run sleep")
549+ logLineFrame(c, 2, "working", false)
550+ cn := <-cancels
551+ if cn.Kind != execpb.CancelKind_CANCEL_KIND_SHUTDOWN {
552+ t.Errorf("kind %v", cn.Kind)
553+ }
554+ stepDone(c, 2, "Run sleep", cCancel)
555+ jobDone(c, cCancel, nil, true)
556+ })
557+ h.srv.jobs = []*ermite.Job{testJob("j1")}
558+ stop := h.run(t)
559+ eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
560+ if err := stop(); err != nil {
561+ t.Fatal(err)
562+ }
563+ h.srv.mu.Lock()
564+ defer h.srv.mu.Unlock()
565+ if len(h.srv.completes) != 1 {
566+ t.Fatalf("%d CompleteJob calls", len(h.srv.completes))
567+ }
568+ if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || !c.RunnerShutdown {
569+ t.Fatalf("%+v", c)
570+ }
571+ if h.driver.stops != 1 {
572+ t.Errorf("stops=%d", h.driver.stops)
573+ }
574+}
575+
576+func TestShutdownHardStopsAnExecutorThatIgnoresIt(t *testing.T) {
577+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
578+ stepStart(c, 1, "Set up job")
579+ stepStart(c, 2, "Run hang")
580+ logLineFrame(c, 2, "hung", false)
581+ <-cancels // received, ignored: keep running until the stream is cut
582+ select {}
583+ })
584+ h.runner.Timing.ShutdownGrace = 300 * time.Millisecond
585+ h.srv.jobs = []*ermite.Job{testJob("j1")}
586+ stop := h.run(t)
587+ eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
588+ start := time.Now()
589+ _ = stop()
590+ if time.Since(start) > 5*time.Second {
591+ t.Fatal("shutdown waited for an executor that does not answer")
592+ }
593+ h.srv.mu.Lock()
594+ defer h.srv.mu.Unlock()
595+ if len(h.srv.completes) != 1 || h.srv.completes[0].Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || !h.srv.completes[0].RunnerShutdown {
596+ t.Fatalf("%+v", h.srv.completes)
597+ }
598+ // Steps that never completed are closed by the runner.
599+ for _, u := range h.srv.completes[0].Steps {
600+ if u.State != ermite.StepState_STEP_STATE_COMPLETED || u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED {
601+ t.Errorf("step %d: %v/%v", u.Number, u.State, u.Conclusion)
602+ }
603+ }
604+}
605+
606+func TestExecutorDyingFailsTheJob(t *testing.T) {
607+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
608+ stepStart(c, 1, "Set up job")
609+ logLineFrame(c, 1, "partial", false)
610+ // the stream closes without a result
611+ })
612+ h.srv.jobs = []*ermite.Job{testJob("j1")}
613+ h.run(t)
614+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
615+ h.srv.mu.Lock()
616+ defer h.srv.mu.Unlock()
617+ c := h.srv.completes[0]
618+ if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE || c.RunnerShutdown {
619+ t.Fatalf("%+v", c)
620+ }
621+ if len(h.srv.lines) != 2 || !strings.HasPrefix(h.srv.lines[1], "##[error]The job's sandbox ended unexpectedly") {
622+ t.Fatalf("lines %q", h.srv.lines)
623+ }
624+ if len(c.Steps) != 1 || c.Steps[0].State != ermite.StepState_STEP_STATE_COMPLETED || c.Steps[0].Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
625+ t.Fatalf("%v", c.Steps)
626+ }
627+}
628+
629+func TestSandboxStartFailureFailsTheJobInStepOne(t *testing.T) {
630+ h := newHarness(t, simpleScript("x"))
631+ h.driver.startErr = errors.New("no kvm today")
632+ h.srv.jobs = []*ermite.Job{testJob("j1")}
633+ h.run(t)
634+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
635+ h.srv.mu.Lock()
636+ defer h.srv.mu.Unlock()
637+ if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
638+ t.Fatalf("%v", c.Conclusion)
639+ }
640+ if len(h.srv.lines) != 1 || !strings.Contains(h.srv.lines[0], "The sandbox did not start: no kvm today") {
641+ t.Fatalf("%q", h.srv.lines)
642+ }
643+ if h.driver.stops != 1 {
644+ t.Errorf("a sandbox that failed to start is still stopped: stops=%d", h.driver.stops)
645+ }
646+}
647+
648+func TestOutputsOverTheLimitFailTheJob(t *testing.T) {
649+ h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
650+ stepStart(c, 1, "Set up job")
651+ stepDone(c, 1, "Set up job", cOK)
652+ // Names that ermite would refuse: the job fails without outputs.
653+ jobDone(c, cOK, map[string]string{"1bad": "x", "fine": "y"}, false)
654+ })
655+ h.srv.jobs = []*ermite.Job{testJob("j1")}
656+ h.run(t)
657+ eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
658+ h.srv.mu.Lock()
659+ defer h.srv.mu.Unlock()
660+ c := h.srv.completes[0]
661+ if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE || len(c.Outputs) != 0 {
662+ t.Fatalf("%v %v", c.Conclusion, len(c.Outputs))
663+ }
664+ if len(h.srv.lines) != 1 || !strings.Contains(h.srv.lines[0], "is not valid") {
665+ t.Fatalf("%q", h.srv.lines)
666+ }
667+}
668+
669+func TestCheckOutputs(t *testing.T) {
670+ for _, tc := range []struct {
671+ in map[string]string
672+ want string
673+ }{
674+ {nil, ""},
675+ {map[string]string{"a-b_c1": "v", "_x": ""}, ""},
676+ {map[string]string{"1a": "v"}, "not valid"},
677+ {map[string]string{"a b": "v"}, "not valid"},
678+ {map[string]string{"": "v"}, "not valid"},
679+ {map[string]string{strings.Repeat("a", 256): "v"}, "not valid"},
680+ {map[string]string{"Foo": "1", "foo": "2"}, "differ only by case"},
681+ {map[string]string{"a": strings.Repeat("x", 1<<20+1)}, "larger than 1 MiB"},
682+ {map[string]string{"a": strings.Repeat("x", 600000), "b": strings.Repeat("x", 600000)}, "in total"},
683+ } {
684+ got := checkOutputs(tc.in)
685+ if (tc.want == "") != (got == "") || !strings.Contains(got, tc.want) {
686+ t.Errorf("%v: %q, want %q", tc.in, got, tc.want)
687+ }
688+ }
689+}
690+
691+func TestLeakedSandboxStopsTakingJobsUntilReaped(t *testing.T) {
692+ h := newHarness(t, simpleScript("x"))
693+ h.driver.stopErr = sandbox.ErrLeaked
694+ h.driver.reapErr = errors.New("still leaked")
695+ h.runner.Timing.ReapRetry = 50 * time.Millisecond
696+ h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2")}
697+ h.run(t)
698+ eventually(t, "first job done", func() bool { return h.srv.completeCount() == 1 })
699+ time.Sleep(400 * time.Millisecond)
700+ if h.srv.completeCount() != 1 {
701+ t.Fatal("a second job started while a sandbox is leaked and cannot be reaped")
702+ }
703+ // Once the reap works, jobs are taken again.
704+ h.driver.mu.Lock()
705+ h.driver.reapErr = nil
706+ h.driver.stopErr = nil
707+ h.driver.mu.Unlock()
708+ eventually(t, "the second job", func() bool { return h.srv.completeCount() == 2 })
709+}
710+
711+func TestReapAtStartAndIsolation(t *testing.T) {
712+ h := newHarness(t, simpleScript("x"))
713+ h.run(t)
714+ eventually(t, "reap", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.reaps >= 1 })
715+ for in, want := range map[sandbox.Isolation]ermite.RunnerIsolation{
716+ sandbox.IsolationVM: ermite.RunnerIsolation_RUNNER_ISOLATION_VM,
717+ sandbox.IsolationContainerSysbox: ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX,
718+ sandbox.IsolationContainerPrivileged: ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED,
719+ "other": ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED,
720+ } {
721+ if got := Isolation(in); got != want {
722+ t.Errorf("%s: %v", in, got)
723+ }
724+ }
725+}
726+
727+func TestRegisterStoresTheToken(t *testing.T) {
728+ srv := newFakeServer()
729+ url := srv.start(t)
730+ dir := t.TempDir() + "/state"
731+ cfg := Config{URL: url, Name: "r1", Labels: []string{"ubuntu-latest"}, StateDir: dir, Version: "v1"}
732+ st, err := Register(context.Background(), cfg, NewClient(url, "reg-token", http.DefaultTransport), sandbox.IsolationContainerSysbox, true)
733+ if err != nil {
734+ t.Fatal(err)
735+ }
736+ if st.RunnerToken != "ermiterun_new" || st.RunnerID != "rid1" {
737+ t.Fatalf("%+v", st)
738+ }
739+ req := srv.registers[0]
740+ if req.Name != "r1" || req.Isolation != ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX || req.Os != "linux" || (req.Arch != "x64" && req.Arch != "arm64") || !req.Replace || req.Version != "v1" || len(req.Labels) != 1 {
741+ t.Fatalf("%v", req)
742+ }
743+ fi, err := os.Stat(StateFile(dir))
744+ if err != nil || fi.Mode().Perm() != 0o600 {
745+ t.Fatalf("%v %v", fi, err)
746+ }
747+ got, err := LoadState(dir)
748+ if err != nil || got != st {
749+ t.Fatalf("%+v %v", got, err)
750+ }
751+ // Anyone can read a state file with a loose mode: refuse it.
752+ if err := os.Chmod(StateFile(dir), 0o644); err != nil {
753+ t.Fatal(err)
754+ }
755+ if _, err := LoadState(dir); err == nil || !strings.Contains(err.Error(), "accessible by other users") {
756+ t.Fatalf("%v", err)
757+ }
758+ if _, err := LoadState(t.TempDir()); !errors.Is(err, ErrNotRegistered) {
759+ t.Fatalf("%v", err)
760+ }
761+ // A refused registration stores nothing.
762+ dir2 := t.TempDir()
763+ cfg.StateDir = dir2
764+ if _, err := Register(context.Background(), cfg, NewClient(url, "bad", http.DefaultTransport), sandbox.IsolationVM, false); err == nil {
765+ t.Fatal("registration with a bad token succeeded")
766+ }
767+ if _, err := os.Stat(StateFile(dir2)); err == nil {
768+ t.Fatal("a state file was written")
769+ }
770+}
771+
772+func TestLoadConfig(t *testing.T) {
773+ env := map[string]string{EnvURL: " https://ermite.example ", EnvLabels: "ubuntu-latest, Ubuntu-24.04 ,ubuntu-latest,"}
774+ c, err := LoadConfig(func(k string) string { return env[k] }, "my host.local", "/state", 2, "v1")
775+ if err != nil {
776+ t.Fatal(err)
777+ }
778+ if c.URL != "https://ermite.example" || c.Name != "my-host.local" || strings.Join(c.Labels, ",") != "ubuntu-latest,Ubuntu-24.04" || c.Capacity != 2 {
779+ t.Fatalf("%+v", c)
780+ }
781+ env[EnvName] = "bad name"
782+ if _, err := LoadConfig(func(k string) string { return env[k] }, "h", "/s", 1, ""); err == nil {
783+ t.Error("an invalid name was accepted")
784+ }
785+ env[EnvName] = "ok"
786+ env[EnvLabels] = "good,bad label"
787+ if _, err := LoadConfig(func(k string) string { return env[k] }, "h", "/s", 1, ""); err == nil {
788+ t.Error("an invalid label was accepted")
789+ }
790+}
791+
792+var _ = ermiteconnect.RunnerServiceClient(nil)
A internal/host/util_test.go +14 -0
@@ -0,0 +1,14 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "fmt"
7+ "time"
8+
9+ "google.golang.org/protobuf/types/known/durationpb"
10+)
11+
12+func sprintf(f string, a ...any) string { return fmt.Sprintf(f, a...) }
13+
14+func durationpbOf(d time.Duration) *durationpb.Duration { return durationpb.New(d) }
@@ -0,0 +1,14 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "fmt"
7+ "time"
8+
9+ "google.golang.org/protobuf/types/known/durationpb"
10+)
11+
12+func sprintf(f string, a ...any) string { return fmt.Sprintf(f, a...) }
13+
14+func durationpbOf(d time.Duration) *durationpb.Duration { return durationpb.New(d) }

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 4

Tags 11

↑↓ moveenter openesc close