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 bootMs map[string]int64 // the guest's Hello.boot_ms; nil: dockerd=2
355}
356
357func (d *fakeDriver) Mode() sandbox.Mode { return sandbox.Container }
358func (d *fakeDriver) Isolation() sandbox.Isolation { return d.iso }
359func (d *fakeDriver) Close() error { return nil }
360func (d *fakeDriver) Reap(context.Context) error {
361 d.mu.Lock()
362 defer d.mu.Unlock()
363 d.reaps++
364 return d.reapErr
365}
366func (d *fakeDriver) New() (sandbox.Sandbox, error) {
367 d.mu.Lock()
368 defer d.mu.Unlock()
369 if d.capacity > 0 && d.live >= d.capacity {
370 return nil, sandbox.ErrCapacity
371 }
372 d.news++
373 d.live++
374 return &fakeSandbox{d: d}, nil
375}
376
377type fakeSandbox struct {
378 d *fakeDriver
379 mu sync.Mutex
380 conn net.Conn
381 stoped bool
382}
383
384func (s *fakeSandbox) ID() string { return "er00000000" }
385func (s *fakeSandbox) StartTimings() map[string]int64 { return map[string]int64{"socket_ms": 1} }
386func (s *fakeSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
387 d := s.d
388 d.mu.Lock()
389 d.specs = append(d.specs, spec)
390 d.mu.Unlock()
391 if d.blockStart != nil {
392 select {
393 case <-d.blockStart:
394 case <-ctx.Done():
395 return nil, ctx.Err()
396 }
397 }
398 d.mu.Lock()
399 if d.startErr != nil {
400 err := d.startErr
401 d.mu.Unlock()
402 return nil, err
403 }
404 d.started++
405 sc := d.script
406 bootMs := d.bootMs
407 if bootMs == nil {
408 bootMs = map[string]int64{"dockerd": 2}
409 }
410 d.mu.Unlock()
411 host, guest := net.Pipe()
412 s.mu.Lock()
413 s.conn = guest
414 s.mu.Unlock()
415 go func() {
416 defer func() { _ = guest.Close() }()
417 c := execstream.NewConn(guest)
418 _ = c.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "test", Sandbox: "container", BootMs: bootMs}}})
419 // NamedValue frames (secrets, variables) come before Run.
420 vals := &seenValues{secrets: map[string]string{}, vars: map[string]string{}}
421 var f *execpb.Frame
422 for {
423 var err error
424 if f, err = c.Recv(); err != nil {
425 return
426 }
427 if f.GetSecret() != nil {
428 vals.secrets[f.GetSecret().Name] = f.GetSecret().Value
429 } else if f.GetVar() != nil {
430 vals.vars[f.GetVar().Name] = f.GetVar().Value
431 } else {
432 break
433 }
434 }
435 if f.GetRun() == nil {
436 return
437 }
438 lastValues.Store(vals)
439 cancels := make(chan *execpb.Cancel, 4)
440 go func() {
441 for {
442 f, err := c.Recv()
443 if err != nil {
444 close(cancels)
445 return
446 }
447 if cn := f.GetCancel(); cn != nil {
448 cancels <- cn
449 }
450 }
451 }()
452 sc(d.t, c, f.GetRun(), cancels)
453 }()
454 return host, nil
455}
456
457func (s *fakeSandbox) Stop(context.Context) error {
458 s.mu.Lock()
459 if s.conn != nil {
460 _ = s.conn.Close()
461 }
462 first := !s.stoped
463 s.stoped = true
464 s.mu.Unlock()
465 if first {
466 s.d.mu.Lock()
467 s.d.stops++
468 s.d.live--
469 err := s.d.stopErr
470 s.d.mu.Unlock()
471 return err
472 }
473 return nil
474}
475
476// ---- script helpers
477
478func send(c *execstream.Conn, b *execpb.Frame) { _ = c.Send(b) }
479
480func stepStart(c *execstream.Conn, n int, name string) {
481 send(c, &execpb.Frame{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: int32(n), Name: name, UnixNano: time.Now().UnixNano()}}})
482}
483func stepDone(c *execstream.Conn, n int, name string, concl execpb.StepConclusion) {
484 send(c, &execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: concl, UnixNano: time.Now().UnixNano()}}})
485}
486func logLineFrame(c *execstream.Conn, n int, text string, continued bool) {
487 send(c, &execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: time.Now().UnixNano(), Text: text, Continued: continued}}})
488}
489func jobDone(c *execstream.Conn, concl execpb.StepConclusion, outputs map[string]string, shutdown bool) {
490 send(c, &execpb.Frame{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{Conclusion: concl, Outputs: outputs, RunnerShutdown: shutdown}}})
491 send(c, &execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
492}
493
494const (
495 cOK = execpb.StepConclusion_STEP_CONCLUSION_SUCCESS
496 cFail = execpb.StepConclusion_STEP_CONCLUSION_FAILURE
497 cCancel = execpb.StepConclusion_STEP_CONCLUSION_CANCELLED
498)
499
500// simpleScript runs a one-step job that logs the given lines.
501func simpleScript(lines ...string) script {
502 return func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
503 stepStart(c, 1, "Set up job")
504 stepDone(c, 1, "Set up job", cOK)
505 stepStart(c, 2, "Run it")
506 for _, l := range lines {
507 logLineFrame(c, 2, l, false)
508 }
509 stepDone(c, 2, "Run it", cOK)
510 stepStart(c, 3, "Complete job")
511 stepDone(c, 3, "Complete job", cOK)
512 jobDone(c, cOK, map[string]string{"out": "v"}, false)
513 }
514}
515
516// ---- harness
517
518func testJob(id string) *ermite.Job {
519 return &ermite.Job{
520 JobId: id, RunId: 7, Attempt: 1, JobToken: "ermitejob_" + id, WorkflowPath: ".github/workflows/ci.yml",
521 WorkflowYaml: "on: push\njobs:\n j:\n runs-on: x\n steps: [{run: echo}]\n", JobKey: "j", Matrix: "{}",
522 Contexts: `{"github":{"repository":"o/r"}}`, Repo: "r", TimeoutMinutes: 5,
523 }
524}
525
526func fastTiming() Timing {
527 return Timing{
528 PollTimeout: 5 * time.Second, HeartbeatEvery: 100 * time.Millisecond, HeartbeatRetry: 30 * time.Millisecond,
529 CallTimeout: 5 * time.Second, BackoffMin: 10 * time.Millisecond, BackoffMax: 40 * time.Millisecond,
530 LogFlush: 5 * time.Millisecond, WatchRetry: 20 * time.Millisecond, ShutdownGrace: 2 * time.Second,
531 CompleteDeadline: 10 * time.Second, NoJobPause: 10 * time.Millisecond,
532 }
533}
534
535type harness struct {
536 srv *fakeServer
537 url string
538 driver *fakeDriver
539 runner *Runner
540 logs *logBuf
541}
542
543// logBuf collects what the runner logs (zerowrap, console format without
544// colors), so that tests can look at it.
545type logBuf struct {
546 mu sync.Mutex
547 b strings.Builder
548}
549
550func (l *logBuf) Write(p []byte) (int, error) {
551 l.mu.Lock()
552 defer l.mu.Unlock()
553 return l.b.Write(p)
554}
555
556func (l *logBuf) String() string { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() }
557
558// ctx returns a context with a logger that writes to the buffer (JSON, one
559// event per line, debug level).
560func (l *logBuf) ctx() context.Context {
561 return zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: l}))
562}
563
564func newHarness(t *testing.T, sc script) *harness {
565 t.Helper()
566 srv := newFakeServer()
567 url := srv.start(t)
568 drv := &fakeDriver{script: sc, t: t, iso: sandbox.IsolationVM}
569 lb := &logBuf{}
570 h := &harness{srv: srv, url: url, driver: drv, logs: lb}
571 t.Cleanup(func() {
572 srv.mu.Lock()
573 defer srv.mu.Unlock()
574 if len(srv.violations) > 0 {
575 t.Errorf("the host broke the contract of AppendLog: %q", srv.violations)
576 }
577 })
578 h.runner = &Runner{
579 Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
580 Driver: drv,
581 Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),
582 Features: func(context.Context) ([]string, error) { return []string{FeatureAnnotations, FeatureSummaries}, nil },
583 Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
584 Timing: fastTiming(),
585 }
586 return h
587}
588
589// run starts the loop; stop cancels it and waits for Run to return.
590func (h *harness) run(t *testing.T) (stop func() error) {
591 t.Helper()
592 ctx, cancel := context.WithCancel(h.logs.ctx())
593 done := make(chan error, 1)
594 go func() { done <- h.runner.Run(ctx) }()
595 var once sync.Once
596 var res error
597 stop = func() error {
598 once.Do(func() {
599 cancel()
600 select {
601 case res = <-done:
602 case <-time.After(20 * time.Second):
603 t.Error("Run did not return after shutdown")
604 }
605 })
606 return res
607 }
608 t.Cleanup(func() { _ = stop() })
609 return stop
610}
611
612func eventually(t *testing.T, what string, cond func() bool) {
613 t.Helper()
614 deadline := time.Now().Add(15 * time.Second)
615 for time.Now().Before(deadline) {
616 if cond() {
617 return
618 }
619 time.Sleep(10 * time.Millisecond)
620 }
621 t.Fatalf("timed out waiting for %s", what)
622}
623
624func (s *fakeServer) completeCount() int { s.mu.Lock(); defer s.mu.Unlock(); return len(s.completes) }
625
626// Result returns what a job reported, like hosttest.Server.Result.
627func (s *fakeServer) Result(jobID string) hosttest.Result {
628 s.mu.Lock()
629 defer s.mu.Unlock()
630 return hosttest.Result{Steps: s.steps}
631}
632
633// seenValues are the secrets and variables the scripted executor received
634// before the Run frame.
635type seenValues struct{ secrets, vars map[string]string }
636
637var lastValues atomic.Pointer[seenValues]