1// SPDX-License-Identifier: MIT
2
3package execstream
4
5import (
6 "bytes"
7 "context"
8 "encoding/binary"
9 "errors"
10 "io"
11 "net"
12 "strings"
13 "sync/atomic"
14 "testing"
15 "testing/iotest"
16 "time"
17
18 "bnema.dev/ermite-runner/internal/execstream/execpb"
19)
20
21func TestRoundTrip(t *testing.T) {
22 var buf bytes.Buffer
23 c := NewConn(&buf)
24 in := []*execpb.Frame{
25 {Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "v", Sandbox: "vm"}}},
26 {Body: &execpb.Frame_Output{Output: &execpb.Output{Stream: execpb.StreamKind_STREAM_KIND_STDERR, Data: []byte("x\x00y")}}},
27 {Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 3}}},
28 }
29 for _, f := range in {
30 if err := c.Send(f); err != nil {
31 t.Fatal(err)
32 }
33 }
34 // One byte at a time: framing must not depend on read sizes.
35 r := NewConn(struct {
36 io.Reader
37 io.Writer
38 }{iotest.OneByteReader(&buf), io.Discard})
39 for i := range in {
40 out, err := r.Recv()
41 if err != nil {
42 t.Fatalf("frame %d: %v", i, err)
43 }
44 if out.String() != in[i].String() {
45 t.Fatalf("frame %d: got %v want %v", i, out, in[i])
46 }
47 }
48 if _, err := r.Recv(); err != io.EOF {
49 t.Fatalf("want io.EOF at end, got %v", err)
50 }
51}
52
53func TestRecvErrors(t *testing.T) {
54 huge := make([]byte, 4)
55 binary.BigEndian.PutUint32(huge, MaxFrame+1)
56 if _, err := NewConn(bytes.NewBuffer(huge)).Recv(); !errors.Is(err, ErrFrameTooLarge) {
57 t.Fatalf("oversize: %v", err)
58 }
59 trunc := []byte{0, 0, 0, 9, 1, 2}
60 if _, err := NewConn(bytes.NewBuffer(trunc)).Recv(); !errors.Is(err, io.ErrUnexpectedEOF) {
61 t.Fatalf("truncated body: %v", err)
62 }
63 if _, err := NewConn(bytes.NewBuffer([]byte{0, 0})).Recv(); !errors.Is(err, io.ErrUnexpectedEOF) {
64 t.Fatalf("truncated header: %v", err)
65 }
66 garbage := []byte{0, 0, 0, 3, 0xff, 0xff, 0xff}
67 if _, err := NewConn(bytes.NewBuffer(garbage)).Recv(); err == nil {
68 t.Fatal("garbage payload accepted")
69 }
70}
71
72func TestSendTooLarge(t *testing.T) {
73 c := NewConn(struct {
74 io.Reader
75 io.Writer
76 }{bytes.NewReader(nil), io.Discard})
77 err := c.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{Data: make([]byte, MaxFrame)}}})
78 if !errors.Is(err, ErrFrameTooLarge) {
79 t.Fatalf("got %v", err)
80 }
81}
82
83func TestDrive(t *testing.T) {
84 host, guest := net.Pipe()
85 defer func() { _ = host.Close() }()
86 go func() {
87 defer func() { _ = guest.Close() }()
88 g := NewConn(guest)
89 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Sandbox: "container"}}})
90 f, err := g.Recv()
91 if err != nil || f.GetRun() == nil || f.GetRun().Script != "echo hi" {
92 return
93 }
94 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{Data: []byte("hi\n")}}})
95 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 7}}})
96 }()
97 var got []byte
98 hello, exit, err := Drive(host, &execpb.Run{Script: "echo hi"}, nil, func(o *execpb.Output) { got = append(got, o.Data...) })
99 if err != nil {
100 t.Fatal(err)
101 }
102 if hello.Sandbox != "container" || exit.Code != 7 || string(got) != "hi\n" {
103 t.Fatalf("hello=%v exit=%v out=%q", hello, exit, got)
104 }
105}
106
107func TestDriveCancelSendsCancel(t *testing.T) {
108 host, guest := net.Pipe()
109 defer func() { _ = host.Close() }()
110 gotCancel := make(chan bool, 1)
111 go func() {
112 defer func() { _ = guest.Close() }()
113 g := NewConn(guest)
114 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{}}})
115 _, _ = g.Recv() // run
116 f, err := g.Recv()
117 gotCancel <- err == nil && f.GetCancel() != nil && f.GetCancel().Reason == "cancelled by the user"
118 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 130}}})
119 }()
120 cancel := make(chan *execpb.Cancel, 1)
121 time.AfterFunc(50*time.Millisecond, func() { cancel <- &execpb.Cancel{Reason: "cancelled by the user"} })
122 _, exit, err := Drive(host, &execpb.Run{}, cancel, nil)
123 if err != nil || exit.Code != 130 {
124 t.Fatalf("exit=%v err=%v", exit, err)
125 }
126 if !<-gotCancel {
127 t.Fatal("guest did not receive a Cancel frame")
128 }
129}
130
131func TestDriveHelloMissing(t *testing.T) {
132 host, guest := net.Pipe()
133 defer func() { _ = host.Close() }()
134 go func() {
135 defer func() { _ = guest.Close() }()
136 _ = NewConn(guest).Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
137 }()
138 if _, _, err := Drive(host, &execpb.Run{}, nil, nil); err == nil {
139 t.Fatal("want error when the first frame is not hello")
140 }
141}
142
143func TestDriveJobDeliversJobFrames(t *testing.T) {
144 host, guest := net.Pipe()
145 go func() {
146 g := NewConn(guest)
147 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "v"}}})
148 r, _ := g.Recv()
149 if r.GetRun().GetJob().GetJobKey() != "build" {
150 t.Errorf("run: %v", r)
151 }
152 for _, b := range []*execpb.Frame{
153 {Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: 1, Name: "Set up job"}}},
154 {Body: &execpb.Frame_Log{Log: &execpb.Log{Step: 1, Text: "hello", UnixNano: 5}}},
155 {Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: 1, Conclusion: execpb.StepConclusion_STEP_CONCLUSION_SUCCESS}}},
156 {Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{Conclusion: execpb.StepConclusion_STEP_CONCLUSION_SUCCESS, Outputs: map[string]string{"a": "b"}}}},
157 {Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}},
158 } {
159 _ = g.Send(b)
160 }
161 }()
162 var got []string
163 _, exit, err := DriveJob(host, &execpb.Run{Job: &execpb.JobSpec{JobKey: "build"}}, nil, JobHandler{
164 StepStarted: func(s *execpb.StepStarted) { got = append(got, "start:"+s.Name) },
165 Log: func(l *execpb.Log) { got = append(got, "log:"+l.Text) },
166 StepCompleted: func(s *execpb.StepCompleted) { got = append(got, "done:"+s.Conclusion.String()) },
167 JobResult: func(r *execpb.JobResult) { got = append(got, "result:"+r.Outputs["a"]) },
168 })
169 if err != nil || exit == nil {
170 t.Fatal(err)
171 }
172 want := "start:Set up job log:hello done:STEP_CONCLUSION_SUCCESS result:b"
173 if strings.Join(got, " ") != want {
174 t.Fatalf("%v", got)
175 }
176}
177
178// GO-003: a flood of ResolveRequest frames is answered by one worker, the ones
179// beyond the queue with an error and no call, and DriveJob does not return
180// before the worker is done (a call in flight is cancelled).
181func TestDriveJobResolveBurst(t *testing.T) {
182 ln, err := net.Listen("tcp", "127.0.0.1:0")
183 if err != nil {
184 t.Skip(err)
185 }
186 defer func() { _ = ln.Close() }()
187 g, err := net.Dial("tcp", ln.Addr().String())
188 if err != nil {
189 t.Fatal(err)
190 }
191 h, err := ln.Accept()
192 if err != nil {
193 t.Fatal(err)
194 }
195 defer func() { _ = g.Close() }()
196 gc := NewConn(g)
197 if err := gc.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{}}}); err != nil {
198 t.Fatal(err)
199 }
200 var running, maxRunning, started atomic.Int32
201 release := make(chan struct{})
202 var exited atomic.Bool
203 done := make(chan struct{})
204 go func() {
205 defer close(done)
206 _, _, _ = DriveJob(h, &execpb.Run{}, nil, JobHandler{Resolve: func(ctx context.Context, r *execpb.ResolveRequest) *execpb.ResolveResponse {
207 n := running.Add(1)
208 defer running.Add(-1)
209 for {
210 m := maxRunning.Load()
211 if n <= m || maxRunning.CompareAndSwap(m, n) {
212 break
213 }
214 }
215 started.Add(1)
216 select {
217 case <-release:
218 case <-ctx.Done():
219 if exited.Load() {
220 t.Error("the worker outlived DriveJob")
221 }
222 }
223 return &execpb.ResolveResponse{}
224 }})
225 exited.Store(true)
226 }()
227 // 50 requests at once while the first call is held.
228 for i := int64(1); i <= 50; i++ {
229 if err := gc.Send(&execpb.Frame{Body: &execpb.Frame_ResolveRequest{ResolveRequest: &execpb.ResolveRequest{Id: i, Uses: []string{"a/b@v1"}}}}); err != nil {
230 t.Fatal(err)
231 }
232 }
233 // Collect the refusals, which are immediate: until the stream is quiet. The
234 // worker may or may not have taken the first request before the queue filled,
235 // so the queue and the running call hold 2 or 3 of the 50.
236 refused := 0
237 for {
238 _ = g.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
239 f, err := gc.Recv()
240 if err != nil {
241 break
242 }
243 if r := f.GetResolveResponse(); r != nil && r.Error != "" {
244 refused++
245 }
246 }
247 _ = g.SetReadDeadline(time.Time{})
248 if refused < 50-1-resolveQueue || refused > 50-resolveQueue {
249 t.Fatalf("%d refused of 50, want %d or %d", refused, 50-1-resolveQueue, 50-resolveQueue)
250 }
251 if maxRunning.Load() != 1 || started.Load() != 1 {
252 t.Errorf("running %d, started %d: resolutions are serial", maxRunning.Load(), started.Load())
253 }
254 // End the session with a call still in flight: it is cancelled and joined.
255 _ = g.Close()
256 select {
257 case <-done:
258 case <-time.After(5 * time.Second):
259 t.Fatal("DriveJob did not return")
260 }
261 close(release)
262}
263
264// GO-009: the resolution worker is in a Send that cannot finish because the
265// guest stopped reading, and the guest ends the session with Exit. DriveJob
266// must return, not wait for the write for ever.
267func TestDriveJobReturnsWhenTheResolverWriteIsStuck(t *testing.T) {
268 old := resolveJoinGrace
269 resolveJoinGrace = 50 * time.Millisecond
270 defer func() { resolveJoinGrace = old }()
271 hostSide, guestSide := net.Pipe() // synchronous: a write needs a reader
272 defer func() { _ = guestSide.Close() }()
273 gc := NewConn(guestSide)
274 go func() {
275 _ = gc.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{}}})
276 if _, err := gc.Recv(); err != nil { // the Run frame
277 return
278 }
279 _ = gc.Send(&execpb.Frame{Body: &execpb.Frame_ResolveRequest{ResolveRequest: &execpb.ResolveRequest{Id: 1, Uses: []string{"a/b@v1"}}}})
280 time.Sleep(100 * time.Millisecond) // the answer is being written: nobody reads it
281 _ = gc.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
282 }()
283 done := make(chan error, 1)
284 go func() {
285 _, exit, err := DriveJob(hostSide, &execpb.Run{}, nil, JobHandler{Resolve: func(context.Context, *execpb.ResolveRequest) *execpb.ResolveResponse {
286 return &execpb.ResolveResponse{}
287 }})
288 if err == nil && exit == nil {
289 err = errors.New("no exit")
290 }
291 done <- err
292 }()
293 select {
294 case err := <-done:
295 if err != nil {
296 t.Fatal(err)
297 }
298 case <-time.After(10 * time.Second):
299 t.Fatal("DriveJob is blocked by a write nobody reads")
300 }
301}
302
303// And a worker that has an answer when the session ends does not write it.
304func TestDriveJobDoesNotSendAfterTheSession(t *testing.T) {
305 hostSide, guestSide := net.Pipe()
306 defer func() { _ = guestSide.Close() }()
307 gc := NewConn(guestSide)
308 release := make(chan struct{})
309 go func() {
310 _ = gc.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{}}})
311 _, _ = gc.Recv()
312 _ = gc.Send(&execpb.Frame{Body: &execpb.Frame_ResolveRequest{ResolveRequest: &execpb.ResolveRequest{Id: 1, Uses: []string{"a/b@v1"}}}})
313 _ = gc.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
314 close(release)
315 }()
316 done := make(chan struct{})
317 go func() {
318 defer close(done)
319 _, _, _ = DriveJob(hostSide, &execpb.Run{}, nil, JobHandler{Resolve: func(ctx context.Context, _ *execpb.ResolveRequest) *execpb.ResolveResponse {
320 <-ctx.Done() // returns only when the session ends
321 return &execpb.ResolveResponse{}
322 }})
323 }()
324 select {
325 case <-done:
326 case <-time.After(10 * time.Second):
327 t.Fatal("DriveJob did not return")
328 }
329 <-release
330}