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