1// SPDX-License-Identifier: MIT
2
3package guest
4
5import (
6 "context"
7 "errors"
8 "io"
9 "net"
10 "path/filepath"
11 "sync/atomic"
12 "testing"
13 "time"
14)
15
16// hostSide is a Unix socket server standing for the host end of the channel.
17type hostSide struct {
18 ln net.Listener
19 accepts atomic.Int32
20}
21
22func newHostSide(t *testing.T, serve func(net.Conn)) *hostSide {
23 t.Helper()
24 ln, err := net.Listen("unix", filepath.Join(t.TempDir(), "h"))
25 if err != nil {
26 t.Fatal(err)
27 }
28 h := &hostSide{ln: ln}
29 t.Cleanup(func() { _ = ln.Close() })
30 go func() {
31 for {
32 c, err := ln.Accept()
33 if err != nil {
34 return
35 }
36 h.accepts.Add(1)
37 go serve(c)
38 }
39 }()
40 return h
41}
42
43func (h *hostSide) dial(ctx context.Context) (io.ReadWriteCloser, error) {
44 var d net.Dialer
45 return d.DialContext(ctx, "unix", h.ln.Addr().String())
46}
47
48func startRelay(t *testing.T, dial DialFunc, max int) *Relay {
49 t.Helper()
50 r, err := StartRelay(context.Background(), "127.0.0.1:0", dial, max)
51 if err != nil {
52 t.Skipf("loopback sockets are not allowed here: %v", err)
53 }
54 t.Cleanup(func() { _ = r.Close() })
55 return r
56}
57
58func dialRelay(t *testing.T, r *Relay) *net.TCPConn {
59 t.Helper()
60 c, err := net.Dial("tcp", r.Addr().String())
61 if err != nil {
62 t.Fatal(err)
63 }
64 t.Cleanup(func() { _ = c.Close() })
65 return c.(*net.TCPConn)
66}
67
68func readAllTimeout(t *testing.T, c net.Conn) []byte {
69 t.Helper()
70 _ = c.SetReadDeadline(time.Now().Add(5 * time.Second))
71 b, err := io.ReadAll(c)
72 if err != nil {
73 t.Fatalf("read: %v", err)
74 }
75 return b
76}
77
78// Bytes go both ways, and the end of the client's stream reaches the host as an
79// EOF while the host can still answer (half-close).
80func TestRelayCopiesBothWaysAndHalfCloses(t *testing.T) {
81 h := newHostSide(t, func(c net.Conn) {
82 defer func() { _ = c.Close() }()
83 req, _ := io.ReadAll(c) // returns only at the client's half-close
84 _, _ = c.Write(append([]byte("echo:"), req...))
85 })
86 r := startRelay(t, h.dial, 4)
87 c := dialRelay(t, r)
88 if _, err := c.Write([]byte("hello")); err != nil {
89 t.Fatal(err)
90 }
91 if err := c.CloseWrite(); err != nil {
92 t.Fatal(err)
93 }
94 if got := string(readAllTimeout(t, c)); got != "echo:hello" {
95 t.Fatalf("got %q", got)
96 }
97}
98
99// The host closing its side reaches the client as an EOF, and bulk data in both
100// directions is not lost.
101func TestRelayBulkAndHostClose(t *testing.T) {
102 const n = 3 << 20
103 h := newHostSide(t, func(c net.Conn) {
104 defer func() { _ = c.Close() }()
105 got, _ := io.CopyN(io.Discard, c, n)
106 _, _ = io.CopyN(c, zeroReader{}, got)
107 })
108 r := startRelay(t, h.dial, 4)
109 c := dialRelay(t, r)
110 go func() { _, _ = io.CopyN(c, zeroReader{}, n) }()
111 if got := len(readAllTimeout(t, c)); got != n {
112 t.Fatalf("read %d bytes, want %d", got, n)
113 }
114}
115
116type zeroReader struct{}
117
118func (zeroReader) Read(p []byte) (int, error) { clear(p); return len(p), nil }
119
120// Every accepted connection is a new stream to the host.
121func TestRelayOneStreamPerConnection(t *testing.T) {
122 h := newHostSide(t, func(c net.Conn) {
123 defer func() { _ = c.Close() }()
124 _, _ = io.Copy(c, c)
125 })
126 r := startRelay(t, h.dial, 8)
127 for i := 0; i < 3; i++ {
128 c := dialRelay(t, r)
129 _, _ = c.Write([]byte("x"))
130 _ = c.CloseWrite()
131 if got := string(readAllTimeout(t, c)); got != "x" {
132 t.Fatalf("got %q", got)
133 }
134 }
135 if n := h.accepts.Load(); n != 3 {
136 t.Fatalf("%d host streams for 3 connections", n)
137 }
138}
139
140// The cap is on open connections: one over it is closed at once, never queued,
141// and a slot freed by a closing connection is usable again.
142func TestRelayConcurrencyCap(t *testing.T) {
143 h := newHostSide(t, func(c net.Conn) {
144 defer func() { _ = c.Close() }()
145 _, _ = io.Copy(c, c)
146 })
147 r := startRelay(t, h.dial, 2)
148 a, b := dialRelay(t, r), dialRelay(t, r)
149 roundTrip := func(c net.Conn) {
150 t.Helper()
151 _ = c.SetDeadline(time.Now().Add(5 * time.Second))
152 if _, err := c.Write([]byte("p")); err != nil {
153 t.Fatal(err)
154 }
155 buf := make([]byte, 1)
156 if _, err := io.ReadFull(c, buf); err != nil {
157 t.Fatal(err)
158 }
159 }
160 roundTrip(a)
161 roundTrip(b)
162 over := dialRelay(t, r)
163 _ = over.SetReadDeadline(time.Now().Add(5 * time.Second))
164 if _, err := over.Read(make([]byte, 1)); !errors.Is(err, io.EOF) && !errors.Is(err, net.ErrClosed) {
165 // A reset is as good as an EOF: the connection was refused service.
166 var ne net.Error
167 if !errors.As(err, &ne) || ne.Timeout() {
168 t.Fatalf("a connection over the cap must be closed, got %v", err)
169 }
170 }
171 if n := h.accepts.Load(); n != 2 {
172 t.Fatalf("%d host streams, want 2", n)
173 }
174 _ = a.Close()
175 deadline := time.Now().Add(5 * time.Second)
176 for {
177 c := dialRelay(t, r)
178 _ = c.SetDeadline(time.Now().Add(time.Second))
179 _, _ = c.Write([]byte("p"))
180 if _, err := io.ReadFull(c, make([]byte, 1)); err == nil {
181 break
182 }
183 if time.Now().After(deadline) {
184 t.Fatal("the slot of a closed connection is never freed")
185 }
186 time.Sleep(20 * time.Millisecond)
187 }
188}
189
190// A channel that does not carry the half-close (Firecracker's vsock proxy may
191// not) must not hold a slot for ever once the client has gone: the idle half is
192// cut after the grace, while a transfer that makes progress is not.
193func TestRelayCutsAnIdleHalfAfterTheGrace(t *testing.T) {
194 release := make(chan struct{})
195 t.Cleanup(func() { close(release) })
196 h := newHostSide(t, func(c net.Conn) {
197 defer func() { _ = c.Close() }()
198 <-release // never answers and never closes its side
199 })
200 const grace = 150 * time.Millisecond
201 r, err := newRelay(context.Background(), "127.0.0.1:0", h.dial, 1, grace)
202 if err != nil {
203 t.Skipf("loopback sockets are not allowed here: %v", err)
204 }
205 t.Cleanup(func() { _ = r.Close() })
206 c := dialRelay(t, r)
207 _, _ = c.Write([]byte("x"))
208 _ = c.CloseWrite() // the client is done: the host half stays idle
209 start := time.Now()
210 // The host never ends its half, so an EOF here is the relay cutting it.
211 if b := readAllTimeout(t, c); len(b) != 0 {
212 t.Fatalf("read %q", b)
213 }
214 if took := time.Since(start); took < grace {
215 t.Fatalf("cut after %v, before the grace of %v", took, grace)
216 }
217 // The slot is back (the cap is 1): a new connection reaches the host.
218 waitFor(t, func() bool {
219 d := dialRelay(t, r)
220 _, _ = d.Write([]byte("y"))
221 return h.accepts.Load() >= 2
222 })
223}
224
225// A slow but live transfer after the first direction ended is not cut: the
226// grace counts idle time, not total time.
227func TestRelayGraceIsIdleTimeNotTotalTime(t *testing.T) {
228 h := newHostSide(t, func(c net.Conn) {
229 defer func() { _ = c.Close() }()
230 _, _ = io.Copy(io.Discard, c) // the request, until the client's half-close
231 for i := 0; i < 8; i++ { // a response slower than the grace in total
232 _, _ = c.Write([]byte("0123456789"))
233 time.Sleep(60 * time.Millisecond)
234 }
235 })
236 r, err := newRelay(context.Background(), "127.0.0.1:0", h.dial, 2, 200*time.Millisecond)
237 if err != nil {
238 t.Skipf("loopback sockets are not allowed here: %v", err)
239 }
240 t.Cleanup(func() { _ = r.Close() })
241 c := dialRelay(t, r)
242 _, _ = c.Write([]byte("req"))
243 _ = c.CloseWrite()
244 if got := len(readAllTimeout(t, c)); got != 80 {
245 t.Fatalf("read %d bytes, want 80", got)
246 }
247}
248
249// A host that cannot be reached closes the client's connection.
250func TestRelayDialFailureClosesClient(t *testing.T) {
251 var calls atomic.Int32
252 r := startRelay(t, func(context.Context) (io.ReadWriteCloser, error) {
253 calls.Add(1)
254 return nil, errors.New("no host")
255 }, 2)
256 c := dialRelay(t, r)
257 _ = c.SetReadDeadline(time.Now().Add(5 * time.Second))
258 if _, err := c.Read(make([]byte, 1)); err == nil || calls.Load() != 1 {
259 t.Fatalf("read err %v after %d dials", err, calls.Load())
260 }
261 // The failed connection holds no slot.
262 for i := 0; i < 4; i++ {
263 d := dialRelay(t, r)
264 _ = d.SetReadDeadline(time.Now().Add(5 * time.Second))
265 if _, err := d.Read(make([]byte, 1)); err == nil {
266 t.Fatal("expected a close")
267 }
268 }
269 if calls.Load() != 5 {
270 t.Fatalf("%d dials, a leaked slot would have refused some", calls.Load())
271 }
272}
273
274// Close stops the listener, cuts the open connections on both sides and returns
275// once every goroutine has ended; it can be called twice.
276func TestRelayCloseCutsEverything(t *testing.T) {
277 hostGone := make(chan struct{}, 1)
278 h := newHostSide(t, func(c net.Conn) {
279 defer func() { _ = c.Close() }()
280 _, _ = io.Copy(io.Discard, c) // ends when the relay closes its stream
281 hostGone <- struct{}{}
282 })
283 r, err := StartRelay(context.Background(), "127.0.0.1:0", h.dial, 4)
284 if err != nil {
285 t.Skipf("loopback sockets are not allowed here: %v", err)
286 }
287 c := dialRelay(t, r)
288 _, _ = c.Write([]byte("x"))
289 waitFor(t, func() bool { return h.accepts.Load() == 1 })
290
291 done := make(chan error, 1)
292 go func() { done <- r.Close() }()
293 select {
294 case err := <-done:
295 if err != nil {
296 t.Fatal(err)
297 }
298 case <-time.After(5 * time.Second):
299 t.Fatal("Close does not return")
300 }
301 select {
302 case <-hostGone:
303 case <-time.After(5 * time.Second):
304 t.Fatal("the host stream was not closed")
305 }
306 _ = c.SetReadDeadline(time.Now().Add(5 * time.Second))
307 if _, err := c.Read(make([]byte, 1)); err == nil {
308 t.Fatal("the client connection is still open")
309 }
310 if c2, err := net.DialTimeout("tcp", r.Addr().String(), time.Second); err == nil {
311 _ = c2.Close()
312 t.Fatal("the relay still listens")
313 }
314 if err := r.Close(); err != nil {
315 t.Fatalf("second Close: %v", err)
316 }
317}
318
319// The relay ends with the context of the job.
320func TestRelayEndsWithItsContext(t *testing.T) {
321 h := newHostSide(t, func(c net.Conn) { _ = c.Close() })
322 ctx, cancel := context.WithCancel(context.Background())
323 r, err := StartRelay(ctx, "127.0.0.1:0", h.dial, 2)
324 if err != nil {
325 t.Skipf("loopback sockets are not allowed here: %v", err)
326 }
327 addr := r.Addr().String()
328 cancel()
329 waitFor(t, func() bool {
330 c, err := net.DialTimeout("tcp", addr, 200*time.Millisecond)
331 if err == nil {
332 _ = c.Close()
333 }
334 return err != nil
335 })
336}
337
338// A port that is taken is an error, not a silent no-cache.
339func TestRelayListenError(t *testing.T) {
340 busy, err := net.Listen("tcp", "127.0.0.1:0")
341 if err != nil {
342 t.Skipf("loopback sockets are not allowed here: %v", err)
343 }
344 defer func() { _ = busy.Close() }()
345 if _, err := StartRelay(context.Background(), busy.Addr().String(), nil, 1); err == nil {
346 t.Fatal("expected an error")
347 }
348}
349
350// The cache channel is a sibling of the executor's: its own vsock port in a VM,
351// a second socket in the same private directory in a container.
352func TestCacheTarget(t *testing.T) {
353 for in, want := range map[string]string{
354 "vsock:1024": "vsock:9089",
355 "unix:/run/ermite/exec.sock": "unix:/run/ermite/cache.sock",
356 } {
357 tg, err := ParseTarget(in)
358 if err != nil {
359 t.Fatal(err)
360 }
361 if got := tg.CacheTarget().String(); got != want {
362 t.Errorf("%s: cache target %s, want %s", in, got, want)
363 }
364 }
365}
366
367// DialStream opens a new connection at every call and does not retry.
368func TestDialStreamUnix(t *testing.T) {
369 h := newHostSide(t, func(c net.Conn) { _ = c.Close() })
370 tg := Target{Network: "unix", Addr: h.ln.Addr().String()}
371 for i := 0; i < 2; i++ {
372 c, err := tg.DialStream(context.Background())
373 if err != nil {
374 t.Fatal(err)
375 }
376 _ = c.Close()
377 }
378 waitFor(t, func() bool { return h.accepts.Load() == 2 })
379 start := time.Now()
380 if _, err := (Target{Network: "unix", Addr: filepath.Join(t.TempDir(), "none")}).DialStream(context.Background()); err == nil || time.Since(start) > time.Second {
381 t.Fatalf("want an immediate error, got %v after %s", err, time.Since(start))
382 }
383}
384
385func waitFor(t *testing.T, ok func() bool) {
386 t.Helper()
387 for end := time.Now().Add(5 * time.Second); time.Now().Before(end); time.Sleep(10 * time.Millisecond) {
388 if ok() {
389 return
390 }
391 }
392 t.Fatal("condition not reached in 5s")
393}