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