1// SPDX-License-Identifier: MIT
2
3package cachesvc
4
5import (
6 "bufio"
7 "fmt"
8 "io"
9 "net"
10 "net/http"
11 "strings"
12 "testing"
13 "time"
14
15 "bnema.dev/ermite-runner/internal/registryproxy"
16)
17
18const registryProgress = 300 * time.Millisecond
19
20// registryRig serves a fake registry behind the service's real HTTP server, so
21// that the connection deadlines apply.
22func registryRig(t *testing.T, h http.HandlerFunc) *rig {
23 t.Helper()
24 return newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, ProgressTimeout: registryProgress, Registry: h})
25}
26
27// A registry download that keeps moving may last longer than the progress
28// period set when the request started.
29func TestRegistryProgressingTransferSucceeds(t *testing.T) {
30 const chunks = 14
31 r := registryRig(t, func(w http.ResponseWriter, _ *http.Request) {
32 w.Header().Set("Content-Length", fmt.Sprint(chunks*1024))
33 for range chunks {
34 if _, err := w.Write([]byte(strings.Repeat("x", 1024))); err != nil {
35 return
36 }
37 _ = http.NewResponseController(w).Flush()
38 time.Sleep(registryProgress / 6)
39 }
40 })
41 start := time.Now()
42 resp, err := http.Get(r.srv.URL + RegistryPath + "library/alpine/blobs/x")
43 if err != nil {
44 t.Fatal(err)
45 }
46 defer func() { _ = resp.Body.Close() }()
47 b, err := io.ReadAll(resp.Body)
48 if err != nil || len(b) != chunks*1024 {
49 t.Fatalf("read %d bytes: %v", len(b), err)
50 }
51 if d := time.Since(start); d < 2*registryProgress {
52 t.Fatalf("transfer took %v, not longer than the progress period", d)
53 }
54}
55
56// The proxy keeps a client alive while it waits for a shared upstream
57// download, before the first byte; the read deadline moves too, so the request
58// context stays alive.
59func TestRegistryKeepAliveWhileWaiting(t *testing.T) {
60 r := registryRig(t, func(w http.ResponseWriter, req *http.Request) {
61 k, ok := w.(registryproxy.KeepAliver)
62 if !ok {
63 t.Error("the registry writer is no KeepAliver")
64 return
65 }
66 for range 12 { // two progress periods
67 time.Sleep(registryProgress / 6)
68 k.KeepAlive()
69 if req.Context().Err() != nil {
70 t.Error("request context ended while kept alive")
71 return
72 }
73 }
74 _, _ = io.WriteString(w, "verified")
75 })
76 resp, err := http.Get(r.srv.URL + RegistryPath + "library/alpine/blobs/x")
77 if err != nil {
78 t.Fatal(err)
79 }
80 defer func() { _ = resp.Body.Close() }()
81 if b, err := io.ReadAll(resp.Body); err != nil || string(b) != "verified" {
82 t.Fatalf("%q %v", b, err)
83 }
84}
85
86// A client that stops reading is cut after one progress period.
87func TestRegistryStalledReaderIsCut(t *testing.T) {
88 result := make(chan error, 1)
89 r := registryRig(t, func(w http.ResponseWriter, _ *http.Request) {
90 chunk := []byte(strings.Repeat("x", 64<<10))
91 deadline := time.Now().Add(5 * time.Second)
92 for time.Now().Before(deadline) {
93 if _, err := w.Write(chunk); err != nil {
94 result <- err
95 return
96 }
97 }
98 result <- nil
99 })
100 conn, err := net.Dial("tcp", strings.TrimPrefix(r.srv.URL, "http://"))
101 if err != nil {
102 t.Fatal(err)
103 }
104 defer func() { _ = conn.Close() }()
105 if tc, ok := conn.(*net.TCPConn); ok {
106 _ = tc.SetReadBuffer(4096)
107 }
108 if _, err := fmt.Fprint(conn, "GET "+RegistryPath+"library/alpine/blobs/x HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
109 t.Fatal(err)
110 }
111 resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
112 if err != nil {
113 t.Fatal(err)
114 }
115 if resp.StatusCode != http.StatusOK {
116 t.Fatal(resp.StatusCode)
117 }
118 // Read nothing more.
119 select {
120 case err := <-result:
121 if err == nil {
122 t.Fatal("a stalled reader was not cut")
123 }
124 case <-time.After(10 * time.Second):
125 t.Fatal("handler still blocked")
126 }
127}