1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
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 "github.com/bnema/zerowrap"
18
19 ermite "bnema.dev/ermite-api/gen/ermite"
20 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
21
22 "bnema.dev/ermite-runner/internal/execstream"
23 "bnema.dev/ermite-runner/internal/execstream/execpb"
24 "bnema.dev/ermite-runner/internal/host/hosttest"
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.
31type 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
47 store *hosttest.LogStore
48 violations []string
49 lines []string // log text, one per line (mirror of store)
50 cont []bool
51 steps map[int32]*ermite.StepUpdate
52 appendReq []*ermite.AppendLogRequest
53 appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)
54 appendN int
55
56 completes []*ermite.CompleteJobRequest
57 completeHook func(n int) (applyThenFail bool, err error)
58 completeN int
59 ended *ermite.JobEnded // when set, job calls answer JobEnded
60 completed map[string]bool
61
62 watchMsgs chan *ermite.WatchJobResponse
63 watches int
64}
65
66func newFakeServer() *fakeServer {
67 s := &fakeServer{
68 runnerTok: "ermiterun_tok", assigned: map[string]*ermite.Job{}, completed: map[string]bool{}, steps: map[int32]*ermite.StepUpdate{},
69 watchMsgs: make(chan *ermite.WatchJobResponse, 16),
70 }
71 s.store = &hosttest.LogStore{OnViolation: func(m string) { s.violations = append(s.violations, m) }}
72 return s
73}
74
75func endedErr(je *ermite.JobEnded) error {
76 e := connect.NewError(connect.CodeFailedPrecondition, io.EOF)
77 if d, err := connect.NewErrorDetail(je); err == nil {
78 e.AddDetail(d)
79 }
80 return e
81}
82
83func (s *fakeServer) checkRunner(h http.Header) error {
84 if h.Get("Authorization") != "Bearer "+s.runnerTok {
85 return connect.NewError(connect.CodeUnauthenticated, io.EOF)
86 }
87 return nil
88}
89
90func (s *fakeServer) checkJob(h http.Header, id string) (*ermite.Job, error) {
91 tok := strings.TrimPrefix(h.Get("Authorization"), "Bearer ")
92 for _, j := range s.assigned {
93 if j.JobToken == tok {
94 if j.JobId != id {
95 return nil, connect.NewError(connect.CodePermissionDenied, io.EOF)
96 }
97 return j, nil
98 }
99 }
100 return nil, connect.NewError(connect.CodeUnauthenticated, io.EOF)
101}
102
103func (s *fakeServer) RegisterRunner(_ context.Context, r *connect.Request[ermite.RegisterRunnerRequest]) (*connect.Response[ermite.RegisterRunnerResponse], error) {
104 s.mu.Lock()
105 defer s.mu.Unlock()
106 s.registers = append(s.registers, r.Msg)
107 if r.Header().Get("Authorization") != "Bearer reg-token" {
108 return nil, connect.NewError(connect.CodePermissionDenied, io.EOF)
109 }
110 return connect.NewResponse(&ermite.RegisterRunnerResponse{RunnerId: "rid1", RunnerToken: "ermiterun_new"}), nil
111}
112
113func (s *fakeServer) FetchJob(_ context.Context, r *connect.Request[ermite.FetchJobRequest]) (*connect.Response[ermite.FetchJobResponse], error) {
114 s.mu.Lock()
115 defer s.mu.Unlock()
116 if err := s.checkRunner(r.Header()); err != nil {
117 return nil, err
118 }
119 s.pollCalls++
120 s.polls = append(s.polls, r.Msg.PollId)
121 if s.pollHook != nil {
122 if resp, err := s.pollHook(s.pollCalls, r.Msg); resp != nil || err != nil {
123 if resp != nil && resp.Job != nil {
124 s.assigned[resp.Job.JobId] = resp.Job
125 }
126 return connect.NewResponse(resp), err
127 }
128 }
129 if len(s.jobs) == 0 {
130 resp := &ermite.FetchJobResponse{}
131 if s.retryAfter > 0 {
132 resp.RetryAfter = durationpbOf(s.retryAfter)
133 }
134 return connect.NewResponse(resp), nil
135 }
136 j := s.jobs[0]
137 s.jobs = s.jobs[1:]
138 s.assigned[j.JobId] = j
139 return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
140}
141
142func (s *fakeServer) Heartbeat(_ context.Context, r *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) {
143 s.mu.Lock()
144 defer s.mu.Unlock()
145 if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
146 return nil, err
147 }
148 if s.ended != nil {
149 return nil, endedErr(s.ended)
150 }
151 s.beats = append(s.beats, time.Now())
152 if s.beatHook != nil {
153 if err := s.beatHook(len(s.beats)); err != nil {
154 return nil, err
155 }
156 }
157 return connect.NewResponse(&ermite.HeartbeatResponse{}), nil
158}
159
160func (s *fakeServer) UpdateJob(_ context.Context, r *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
161 s.mu.Lock()
162 defer s.mu.Unlock()
163 if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
164 return nil, err
165 }
166 if s.ended != nil || s.completed[r.Msg.JobId] {
167 je := s.ended
168 if je == nil {
169 je = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}
170 }
171 return nil, endedErr(je)
172 }
173 for _, u := range r.Msg.Steps {
174 s.steps[u.Number] = u
175 }
176 return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
177}
178
179func (s *fakeServer) AppendLog(_ context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
180 s.mu.Lock()
181 defer s.mu.Unlock()
182 if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
183 return nil, err
184 }
185 if s.ended != nil || s.completed[r.Msg.JobId] {
186 je := s.ended
187 if je == nil {
188 je = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_SUCCESS}
189 }
190 return nil, endedErr(je)
191 }
192 s.appendN++
193 s.appendReq = append(s.appendReq, r.Msg)
194 // The rules of runner.proto are written once, in hosttest.LogStore.
195 var failAfter error
196 if s.appendHook != nil {
197 thenFail, err := s.appendHook(s.appendN, r.Msg)
198 if err != nil && !thenFail {
199 return nil, err
200 }
201 failAfter = err
202 }
203 next, err := s.store.Apply(r.Msg)
204 if err != nil {
205 return nil, err
206 }
207 s.lines, s.cont = s.lines[:0], s.cont[:0]
208 for _, l := range s.store.Lines() {
209 s.lines = append(s.lines, l.Text)
210 s.cont = append(s.cont, l.Continued)
211 }
212 if failAfter != nil {
213 return nil, failAfter
214 }
215 return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: next}), nil
216}
217
218func (s *fakeServer) CompleteJob(_ context.Context, r *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
219 s.mu.Lock()
220 defer s.mu.Unlock()
221 if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
222 return nil, err
223 }
224 s.completeN++
225 s.completes = append(s.completes, r.Msg)
226 if s.completeHook != nil {
227 thenFail, err := s.completeHook(s.completeN)
228 if err != nil {
229 if thenFail {
230 s.completed[r.Msg.JobId] = true
231 }
232 return nil, err
233 }
234 }
235 if s.completed[r.Msg.JobId] || s.ended != nil {
236 c := ermite.CiConclusion_CI_CONCLUSION_FAILURE
237 if s.ended != nil {
238 c = s.ended.Conclusion
239 }
240 return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: c, Applied: false}), nil
241 }
242 s.completed[r.Msg.JobId] = true
243 return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil
244}
245
246func (s *fakeServer) WatchJob(ctx context.Context, r *connect.Request[ermite.WatchJobRequest], st *connect.ServerStream[ermite.WatchJobResponse]) error {
247 s.mu.Lock()
248 _, err := s.checkJob(r.Header(), r.Msg.JobId)
249 s.watches++
250 ended := s.ended
251 s.mu.Unlock()
252 if err != nil {
253 return err
254 }
255 if ended != nil {
256 return endedErr(ended)
257 }
258 for {
259 select {
260 case m := <-s.watchMsgs:
261 if err := st.Send(m); err != nil {
262 return err
263 }
264 if m.GetEnded() != nil {
265 return nil
266 }
267 case <-ctx.Done():
268 return ctx.Err()
269 }
270 }
271}
272
273// start serves the handler over HTTP/1.1 on a local port (Connect streams server messages over it) and returns its URL.
274func (s *fakeServer) start(t *testing.T) string {
275 t.Helper()
276 srv := httptest.NewServer(hosttest.Mount(ermiteconnect.NewRunnerServiceHandler(s)))
277 t.Cleanup(srv.Close)
278 return srv.URL
279}
280
281func (s *fakeServer) snapshotLines() []string {
282 s.mu.Lock()
283 defer s.mu.Unlock()
284 return append([]string(nil), s.lines...)
285}
286
287// ---- fake sandbox: the "executor" is a function on the other end of a pipe
288
289type script func(t *testing.T, c *execstream.Conn, run *execpb.Run, cancels <-chan *execpb.Cancel)
290
291type fakeDriver struct {
292 mu sync.Mutex
293 script script
294 t *testing.T
295 news int
296 stops int
297 started int
298 reaps int
299 reapErr error
300 startErr error
301 stopErr error
302 capacity int
303 live int
304 blockStart chan struct{} // when set, Start waits on it
305 specs []sandbox.JobSpec // what each Start was asked for
306 iso sandbox.Isolation
307}
308
309func (d *fakeDriver) Mode() sandbox.Mode { return sandbox.Container }
310func (d *fakeDriver) Isolation() sandbox.Isolation { return d.iso }
311func (d *fakeDriver) Close() error { return nil }
312func (d *fakeDriver) Reap(context.Context) error {
313 d.mu.Lock()
314 defer d.mu.Unlock()
315 d.reaps++
316 return d.reapErr
317}
318func (d *fakeDriver) New() (sandbox.Sandbox, error) {
319 d.mu.Lock()
320 defer d.mu.Unlock()
321 if d.capacity > 0 && d.live >= d.capacity {
322 return nil, sandbox.ErrCapacity
323 }
324 d.news++
325 d.live++
326 return &fakeSandbox{d: d}, nil
327}
328
329type fakeSandbox struct {
330 d *fakeDriver
331 mu sync.Mutex
332 conn net.Conn
333 stoped bool
334}
335
336func (s *fakeSandbox) ID() string { return "er00000000" }
337func (s *fakeSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
338 d := s.d
339 d.mu.Lock()
340 d.specs = append(d.specs, spec)
341 d.mu.Unlock()
342 if d.blockStart != nil {
343 select {
344 case <-d.blockStart:
345 case <-ctx.Done():
346 return nil, ctx.Err()
347 }
348 }
349 d.mu.Lock()
350 if d.startErr != nil {
351 err := d.startErr
352 d.mu.Unlock()
353 return nil, err
354 }
355 d.started++
356 sc := d.script
357 d.mu.Unlock()
358 host, guest := net.Pipe()
359 s.mu.Lock()
360 s.conn = guest
361 s.mu.Unlock()
362 go func() {
363 defer func() { _ = guest.Close() }()
364 c := execstream.NewConn(guest)
365 _ = c.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "test", Sandbox: "container"}}})
366 f, err := c.Recv()
367 if err != nil || f.GetRun() == nil {
368 return
369 }
370 cancels := make(chan *execpb.Cancel, 4)
371 go func() {
372 for {
373 f, err := c.Recv()
374 if err != nil {
375 close(cancels)
376 return
377 }
378 if cn := f.GetCancel(); cn != nil {
379 cancels <- cn
380 }
381 }
382 }()
383 sc(d.t, c, f.GetRun(), cancels)
384 }()
385 return host, nil
386}
387
388func (s *fakeSandbox) Stop(context.Context) error {
389 s.mu.Lock()
390 if s.conn != nil {
391 _ = s.conn.Close()
392 }
393 first := !s.stoped
394 s.stoped = true
395 s.mu.Unlock()
396 if first {
397 s.d.mu.Lock()
398 s.d.stops++
399 s.d.live--
400 err := s.d.stopErr
401 s.d.mu.Unlock()
402 return err
403 }
404 return nil
405}
406
407// ---- script helpers
408
409func send(c *execstream.Conn, b *execpb.Frame) { _ = c.Send(b) }
410
411func stepStart(c *execstream.Conn, n int, name string) {
412 send(c, &execpb.Frame{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: int32(n), Name: name, UnixNano: time.Now().UnixNano()}}})
413}
414func stepDone(c *execstream.Conn, n int, name string, concl execpb.StepConclusion) {
415 send(c, &execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: concl, UnixNano: time.Now().UnixNano()}}})
416}
417func logLineFrame(c *execstream.Conn, n int, text string, continued bool) {
418 send(c, &execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: time.Now().UnixNano(), Text: text, Continued: continued}}})
419}
420func jobDone(c *execstream.Conn, concl execpb.StepConclusion, outputs map[string]string, shutdown bool) {
421 send(c, &execpb.Frame{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{Conclusion: concl, Outputs: outputs, RunnerShutdown: shutdown}}})
422 send(c, &execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
423}
424
425const (
426 cOK = execpb.StepConclusion_STEP_CONCLUSION_SUCCESS
427 cFail = execpb.StepConclusion_STEP_CONCLUSION_FAILURE
428 cCancel = execpb.StepConclusion_STEP_CONCLUSION_CANCELLED
429)
430
431// simpleScript runs a one-step job that logs the given lines.
432func simpleScript(lines ...string) script {
433 return func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
434 stepStart(c, 1, "Set up job")
435 stepDone(c, 1, "Set up job", cOK)
436 stepStart(c, 2, "Run it")
437 for _, l := range lines {
438 logLineFrame(c, 2, l, false)
439 }
440 stepDone(c, 2, "Run it", cOK)
441 stepStart(c, 3, "Complete job")
442 stepDone(c, 3, "Complete job", cOK)
443 jobDone(c, cOK, map[string]string{"out": "v"}, false)
444 }
445}
446
447// ---- harness
448
449func testJob(id string) *ermite.Job {
450 return &ermite.Job{
451 JobId: id, RunId: 7, Attempt: 1, JobToken: "ermitejob_" + id, WorkflowPath: ".github/workflows/ci.yml",
452 WorkflowYaml: "on: push\njobs:\n j:\n runs-on: x\n steps: [{run: echo}]\n", JobKey: "j", Matrix: "{}",
453 Contexts: `{"github":{"repository":"o/r"}}`, Repo: "r", TimeoutMinutes: 5,
454 }
455}
456
457func fastTiming() Timing {
458 return Timing{
459 PollTimeout: 5 * time.Second, HeartbeatEvery: 100 * time.Millisecond, HeartbeatRetry: 30 * time.Millisecond,
460 CallTimeout: 5 * time.Second, BackoffMin: 10 * time.Millisecond, BackoffMax: 40 * time.Millisecond,
461 LogFlush: 5 * time.Millisecond, WatchRetry: 20 * time.Millisecond, ShutdownGrace: 2 * time.Second,
462 CompleteDeadline: 10 * time.Second, NoJobPause: 10 * time.Millisecond,
463 }
464}
465
466type harness struct {
467 srv *fakeServer
468 url string
469 driver *fakeDriver
470 runner *Runner
471 logs *logBuf
472}
473
474// logBuf collects what the runner logs (zerowrap, console format without
475// colors), so that tests can look at it.
476type logBuf struct {
477 mu sync.Mutex
478 b strings.Builder
479}
480
481func (l *logBuf) Write(p []byte) (int, error) {
482 l.mu.Lock()
483 defer l.mu.Unlock()
484 return l.b.Write(p)
485}
486
487func (l *logBuf) String() string { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() }
488
489// ctx returns a context with a logger that writes to the buffer (JSON, one
490// event per line, debug level).
491func (l *logBuf) ctx() context.Context {
492 return zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: l}))
493}
494
495func newHarness(t *testing.T, sc script) *harness {
496 t.Helper()
497 srv := newFakeServer()
498 url := srv.start(t)
499 drv := &fakeDriver{script: sc, t: t, iso: sandbox.IsolationVM}
500 lb := &logBuf{}
501 h := &harness{srv: srv, url: url, driver: drv, logs: lb}
502 t.Cleanup(func() {
503 srv.mu.Lock()
504 defer srv.mu.Unlock()
505 if len(srv.violations) > 0 {
506 t.Errorf("the host broke the contract of AppendLog: %q", srv.violations)
507 }
508 })
509 h.runner = &Runner{
510 Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
511 Driver: drv,
512 Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),
513 Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
514 Timing: fastTiming(),
515 }
516 return h
517}
518
519// run starts the loop; stop cancels it and waits for Run to return.
520func (h *harness) run(t *testing.T) (stop func() error) {
521 t.Helper()
522 ctx, cancel := context.WithCancel(h.logs.ctx())
523 done := make(chan error, 1)
524 go func() { done <- h.runner.Run(ctx) }()
525 var once sync.Once
526 var res error
527 stop = func() error {
528 once.Do(func() {
529 cancel()
530 select {
531 case res = <-done:
532 case <-time.After(20 * time.Second):
533 t.Error("Run did not return after shutdown")
534 }
535 })
536 return res
537 }
538 t.Cleanup(func() { _ = stop() })
539 return stop
540}
541
542func eventually(t *testing.T, what string, cond func() bool) {
543 t.Helper()
544 deadline := time.Now().Add(15 * time.Second)
545 for time.Now().Before(deadline) {
546 if cond() {
547 return
548 }
549 time.Sleep(10 * time.Millisecond)
550 }
551 t.Fatalf("timed out waiting for %s", what)
552}
553
554func (s *fakeServer) completeCount() int { s.mu.Lock(); defer s.mu.Unlock(); return len(s.completes) }
555
556// Result returns what a job reported, like hosttest.Server.Result.
557func (s *fakeServer) Result(jobID string) hosttest.Result {
558 s.mu.Lock()
559 defer s.mu.Unlock()
560 return hosttest.Result{Steps: s.steps}
561}