detached 2b9f5c9

#4 perf(runner): job phase timing, preinstalled Go/Node tool cache and Docker Hub pull-through proxy

state
merged · aba298d · merge commit
branch
perf/job-timing-and-caches → main
opened
bnema ·
updated

Speeds up jobs and shows where their time goes.

  • Timing: the "job finished" log line carries the duration of each phase (ack, sandbox start with its sub-steps, guest boot, action resolution, archive misses, run, stop).
  • Tool cache: the ubuntu-26.04 job image ships Go and Node.js in the runner tool cache (pinned in toolcache.lock), so setup-go and setup-node find them without downloading.
  • Registry proxy: a Docker Hub pull-through cache on the runner host (internal/registryproxy), reached by the guest dockerd through --registry-mirror on the per-job cache channel. Content is checked by digest, bounded by ERMITE_RUNNER_REGISTRY_CACHE_MAX (default 10G, 0 turns it off), with per-job and global download limits.

Checks: go vet, go test -race ./..., staticcheck, the live Docker Hub test (ERMITE_RUNNER_IT_REGISTRY=1), and an end-to-end run on a local stack in vm mode (setup-go served from the tool cache; second pull of alpine:3 served entirely from the proxy). Not verified: the e2e suite after the last refactor commit, an arm64 image, container mode under rootless Docker.

CommitSubjectAuthorDateSignatureStatus
2b9f5c9refactor(registryproxy): simplify the proxy store, guest timing contract and size parsingbriceGPG Verified
57617dffeat(registryproxy): cache Docker Hub pulls of the sandboxes on the runner hostbriceGPG Verified
09df1abMerge branch 'perf/p1-timing' into perf/job-timing-and-cachesbriceGPG Verified
2b98cb5feat(timing): log the duration of each job phase on job finishedbriceGPG Verified
be87239feat(images): preinstall Go and Node.js in the tool cache of the job imagebriceGPG Verified

Checks 2

69 files · +4944 -687
Unified
UnifiedSplit
2/2
⋯

Merge settings

modes
fast-forward, squash, merge commit
checks
failing or unfinished checks block the merge
M docs/cache-and-artifacts.md +56 -1
show this file
M docs/configuration.md +2 -1
show this file
M docs/development.md +26 -0
show this file
M docs/images.md +17 -2
show this file
M docs/sandboxes.md +0 -1
show this file
M docs/service.md +31 -0
show this file
M host_cmd.go +42 -6
show this file
M images/ubuntu-26.04/Dockerfile +4 -0
show this file
A images/ubuntu-26.04/install-toolcache.sh +42 -0
show this file
A images/ubuntu-26.04/toolcache.lock +11 -0
show this file
M internal/actionscache/cache.go +17 -43
show this file
M internal/actionscache/cache_test.go +17 -30
show this file
M internal/cachesvc/archives_test.go +1 -1
show this file
M internal/cachesvc/handler.go +51 -3
show this file
A internal/cachesvc/registry_deadline_test.go +127 -0
show this file
A internal/cachesvc/registry_limits_test.go +195 -0
show this file
A internal/cachesvc/registry_test.go +37 -0
show this file
M internal/cachesvc/service.go +39 -13
show this file
M internal/cachesvc/service_test.go +10 -5
show this file
M internal/dnsfwd/dnsfwd.go +20 -140
show this file
M internal/dnsfwd/dnsfwd_test.go +31 -374
show this file
A internal/e2e/registry_test.go +114 -0
show this file
A internal/e2e/timing_test.go +83 -0
show this file
A internal/e2e/toolcache_test.go +82 -0
show this file
M internal/execstream/exec.proto +1 -0
show this file
M internal/execstream/execpb/exec.pb.go +39 -25
show this file
M internal/guest/cacheurl_test.go +3 -3
show this file
A internal/guest/dockerd.go +15 -0
show this file
M internal/guest/exec.go +2 -1
show this file
M internal/guest/init.go +31 -6
show this file
M internal/guest/jobexec.go +7 -1
show this file
A internal/guest/registry_test.go +35 -0
show this file
A internal/guest/timing.go +38 -0
show this file
A internal/guest/timing_test.go +33 -0
show this file
M internal/host/archives.go +39 -5
show this file
M internal/host/archives_test.go +83 -2
show this file
M internal/host/cache.go +8 -4
show this file
M internal/host/fake_test.go +8 -2
show this file
M internal/host/job.go +23 -0
show this file
A internal/host/registry_test.go +49 -0
show this file
M internal/host/resolve.go +8 -0
show this file
M internal/host/resolve_test.go +1 -1
show this file
M internal/host/results_test.go +1 -1
show this file
M internal/host/runner.go +2 -2
show this file
M internal/host/runner_test.go +12 -0
show this file
M internal/host/shutdown_test.go +1 -1
show this file
A internal/host/timeline.go +52 -0
show this file
A internal/host/timeline_test.go +74 -0
show this file
A internal/registryproxy/accept.go +41 -0
show this file
A internal/registryproxy/accept_test.go +112 -0
show this file
A internal/registryproxy/clients.go +105 -0
@@ -0,0 +1,105 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "errors"
8+ "io"
9+ "net/http"
10+ "sync"
11+ "time"
12+)
13+
14+// upstreamClients returns the metadata client (auth, manifests: a total
15+// timeout and no redirects) and the blob client (no total timeout, a response
16+// header timeout, and redirects that drop the bearer when leaving the host).
17+func upstreamClients(requestTimeout time.Duration) (metadata, blobs *http.Client) {
18+ metadata = &http.Client{
19+ Timeout: requestTimeout,
20+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
21+ }
22+ blobs = &http.Client{
23+ CheckRedirect: func(req *http.Request, via []*http.Request) error {
24+ if len(via) >= 10 {
25+ return errors.New("too many registry redirects")
26+ }
27+ if len(via) > 0 && req.URL.Host != via[0].URL.Host {
28+ req.Header.Del("Authorization")
29+ }
30+ return nil
31+ },
32+ }
33+ if tr, ok := http.DefaultTransport.(*http.Transport); ok {
34+ tr = tr.Clone()
35+ tr.ResponseHeaderTimeout = requestTimeout
36+ blobs.Transport = tr
37+ }
38+ return metadata, blobs
39+}
40+
41+// idleBody serializes timer reset/stop against expiry. Cancellation interrupts
42+// even a blocked transport read; only completed, nonempty reads renew the lease.
43+type idleBody struct {
44+ io.ReadCloser
45+ mu sync.Mutex
46+ timer *time.Timer
47+ cancel context.CancelFunc
48+ idle time.Duration
49+ expired bool
50+}
51+
52+func (b *idleBody) Read(p []byte) (int, error) {
53+ n, err := b.ReadCloser.Read(p)
54+ if n > 0 {
55+ b.mu.Lock()
56+ if !b.expired {
57+ b.timer.Reset(b.idle)
58+ }
59+ b.mu.Unlock()
60+ }
61+ return n, err
62+}
63+
64+func (b *idleBody) Close() error {
65+ b.mu.Lock()
66+ b.expired = true
67+ b.timer.Stop()
68+ b.mu.Unlock()
69+ b.cancel()
70+ return b.ReadCloser.Close()
71+}
72+
73+// blob GETs a blob. Each attempt has only a header deadline (RequestTimeout)
74+// and then a body progress deadline (BlobIdleTimeout), and no total transfer
75+// deadline: the response body is an idleBody that owns the attempt's context.
76+func (u *upstream) blob(ctx context.Context, name, digest string) (*http.Response, error) {
77+ return u.authorized(ctx, name, func(tok string) (*http.Response, error) {
78+ bodyCtx, cancel := context.WithCancel(ctx)
79+ lease := &idleBody{cancel: cancel, idle: u.blobIdle}
80+ lease.timer = time.AfterFunc(u.requestTimeout, func() { lease.mu.Lock(); lease.expired = true; cancel(); lease.mu.Unlock() })
81+ req, err := http.NewRequestWithContext(bodyCtx, http.MethodGet, u.base+"/v2/"+name+"/"+kindBlob.dir()+"/"+digest, nil)
82+ if err != nil {
83+ lease.timer.Stop()
84+ cancel()
85+ return nil, err
86+ }
87+ req.Header.Set("Authorization", "Bearer "+tok)
88+ resp, err := u.blobClient.Do(req)
89+ if err != nil {
90+ lease.timer.Stop()
91+ cancel()
92+ return nil, err
93+ }
94+ lease.ReadCloser = resp.Body
95+ resp.Body = lease
96+ if resp.StatusCode == http.StatusOK {
97+ lease.mu.Lock()
98+ if !lease.expired {
99+ lease.timer.Reset(u.blobIdle)
100+ }
101+ lease.mu.Unlock()
102+ }
103+ return resp, nil
104+ })
105+}
@@ -0,0 +1,105 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "errors"
8+ "io"
9+ "net/http"
10+ "sync"
11+ "time"
12+)
13+
14+// upstreamClients returns the metadata client (auth, manifests: a total
15+// timeout and no redirects) and the blob client (no total timeout, a response
16+// header timeout, and redirects that drop the bearer when leaving the host).
17+func upstreamClients(requestTimeout time.Duration) (metadata, blobs *http.Client) {
18+ metadata = &http.Client{
19+ Timeout: requestTimeout,
20+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
21+ }
22+ blobs = &http.Client{
23+ CheckRedirect: func(req *http.Request, via []*http.Request) error {
24+ if len(via) >= 10 {
25+ return errors.New("too many registry redirects")
26+ }
27+ if len(via) > 0 && req.URL.Host != via[0].URL.Host {
28+ req.Header.Del("Authorization")
29+ }
30+ return nil
31+ },
32+ }
33+ if tr, ok := http.DefaultTransport.(*http.Transport); ok {
34+ tr = tr.Clone()
35+ tr.ResponseHeaderTimeout = requestTimeout
36+ blobs.Transport = tr
37+ }
38+ return metadata, blobs
39+}
40+
41+// idleBody serializes timer reset/stop against expiry. Cancellation interrupts
42+// even a blocked transport read; only completed, nonempty reads renew the lease.
43+type idleBody struct {
44+ io.ReadCloser
45+ mu sync.Mutex
46+ timer *time.Timer
47+ cancel context.CancelFunc
48+ idle time.Duration
49+ expired bool
50+}
51+
52+func (b *idleBody) Read(p []byte) (int, error) {
53+ n, err := b.ReadCloser.Read(p)
54+ if n > 0 {
55+ b.mu.Lock()
56+ if !b.expired {
57+ b.timer.Reset(b.idle)
58+ }
59+ b.mu.Unlock()
60+ }
61+ return n, err
62+}
63+
64+func (b *idleBody) Close() error {
65+ b.mu.Lock()
66+ b.expired = true
67+ b.timer.Stop()
68+ b.mu.Unlock()
69+ b.cancel()
70+ return b.ReadCloser.Close()
71+}
72+
73+// blob GETs a blob. Each attempt has only a header deadline (RequestTimeout)
74+// and then a body progress deadline (BlobIdleTimeout), and no total transfer
75+// deadline: the response body is an idleBody that owns the attempt's context.
76+func (u *upstream) blob(ctx context.Context, name, digest string) (*http.Response, error) {
77+ return u.authorized(ctx, name, func(tok string) (*http.Response, error) {
78+ bodyCtx, cancel := context.WithCancel(ctx)
79+ lease := &idleBody{cancel: cancel, idle: u.blobIdle}
80+ lease.timer = time.AfterFunc(u.requestTimeout, func() { lease.mu.Lock(); lease.expired = true; cancel(); lease.mu.Unlock() })
81+ req, err := http.NewRequestWithContext(bodyCtx, http.MethodGet, u.base+"/v2/"+name+"/"+kindBlob.dir()+"/"+digest, nil)
82+ if err != nil {
83+ lease.timer.Stop()
84+ cancel()
85+ return nil, err
86+ }
87+ req.Header.Set("Authorization", "Bearer "+tok)
88+ resp, err := u.blobClient.Do(req)
89+ if err != nil {
90+ lease.timer.Stop()
91+ cancel()
92+ return nil, err
93+ }
94+ lease.ReadCloser = resp.Body
95+ resp.Body = lease
96+ if resp.StatusCode == http.StatusOK {
97+ lease.mu.Lock()
98+ if !lease.expired {
99+ lease.timer.Reset(u.blobIdle)
100+ }
101+ lease.mu.Unlock()
102+ }
103+ return resp, nil
104+ })
105+}
A internal/registryproxy/flights_test.go +248 -0
@@ -0,0 +1,248 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "fmt"
8+ "io"
9+ "net/http"
10+ "net/http/httptest"
11+ "os"
12+ "path/filepath"
13+ "strings"
14+ "sync"
15+ "sync/atomic"
16+ "testing"
17+ "time"
18+
19+ "github.com/bnema/zerowrap"
20+)
21+
22+// Concurrent requests for one digest share a single fetch.
23+func TestSingleFlight(t *testing.T) {
24+ s := mustOpen(t, storeCfg(t.TempDir(), 1<<20))
25+ var calls atomic.Int32
26+ var wg sync.WaitGroup
27+ start := make(chan struct{})
28+ for range 20 {
29+ wg.Go(func() {
30+ <-start
31+ f, _, _, err := s.ensure(context.Background(), kindBlob, sum("content"), func(_ context.Context, w io.Writer) (string, error) {
32+ calls.Add(1)
33+ _, err := io.WriteString(w, "content")
34+ return "", err
35+ }, keepalive{})
36+ if err != nil {
37+ t.Error(err)
38+ return
39+ }
40+ _ = f.Close()
41+ })
42+ }
43+ close(start)
44+ wg.Wait()
45+ if calls.Load() != 1 {
46+ t.Fatalf("fetches %d", calls.Load())
47+ }
48+}
49+
50+// A guest that requests many digests and disconnects leaves no download
51+// behind: once a flight has no waiter for the grace period, it is cancelled
52+// upstream and its temporary file removed.
53+func TestAbandonedDownloadsCancelled(t *testing.T) {
54+ u := newStallingUpstream(t)
55+ const n = 20
56+ p := proxyWith(t, u.srv, Config{AbandonGrace: 20 * time.Millisecond, MaxDownloads: n})
57+ for i := range n {
58+ serveCancelled(t, p, u, blobPath(i), int32(i+1))
59+ }
60+ waitFor(t, p.s, "every flight to end", func() bool { return len(p.s.flights) == 0 && p.s.downloads == 0 && p.s.pending == 0 })
61+ for end := time.Now().Add(5 * time.Second); u.cancelled.Load() != n; {
62+ if time.Now().After(end) {
63+ t.Fatalf("upstream saw %d cancellations of %d", u.cancelled.Load(), n)
64+ }
65+ time.Sleep(time.Millisecond)
66+ }
67+ waitFor(t, p.s, "tmp/ to be empty", func() bool {
68+ ds, _ := os.ReadDir(filepath.Join(p.s.dir, "tmp"))
69+ return len(ds) == 0
70+ })
71+}
72+
73+// A request that joins within the grace period keeps the download.
74+func TestRejoinWithinGraceKeepsDownload(t *testing.T) {
75+ body := strings.Repeat("layer", 4000)
76+ release := make(chan struct{})
77+ var gets, cancelled atomic.Int32
78+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
79+ gets.Add(1)
80+ w.Header().Set("Content-Length", fmt.Sprint(len(body)))
81+ _, _ = io.WriteString(w, body[:10])
82+ w.(http.Flusher).Flush()
83+ select {
84+ case <-release:
85+ _, _ = io.WriteString(w, body[10:])
86+ case <-r.Context().Done():
87+ cancelled.Add(1)
88+ }
89+ })
90+ var once sync.Once
91+ t.Cleanup(func() { once.Do(func() { close(release) }) })
92+ p := proxyWith(t, srv, Config{AbandonGrace: time.Hour})
93+ path := "/v2/library/alpine/blobs/" + sum(body)
94+ ctx, cancel := context.WithCancel(context.Background())
95+ first := make(chan struct{})
96+ go func() {
97+ defer close(first)
98+ p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
99+ }()
100+ key := kindBlob.key(sum(body)[7:])
101+ waitFor(t, p.s, "the first waiter", func() bool { return waiters(p.s, key) == 1 })
102+ cancel()
103+ <-first
104+ w := httptest.NewRecorder()
105+ second := make(chan struct{})
106+ go func() { defer close(second); p.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil)) }()
107+ waitFor(t, p.s, "the second waiter", func() bool { return waiters(p.s, key) == 1 })
108+ once.Do(func() { close(release) })
109+ <-second
110+ if w.Code != http.StatusOK || w.Body.String() != body || gets.Load() != 1 || cancelled.Load() != 0 {
111+ t.Fatalf("code %d, %d fetches, %d cancelled", w.Code, gets.Load(), cancelled.Load())
112+ }
113+}
114+
115+// The shared download does not belong to the client that started it. That
116+// client disconnects mid-download; a second client, joining within the grace
117+// period, still gets the blob and the blob is committed, from one upstream
118+// fetch.
119+func TestBlobFlightSurvivesInitiatorDisconnect(t *testing.T) {
120+ body := strings.Repeat("layer", 4000)
121+ half, release := make(chan struct{}), make(chan struct{})
122+ var releaseOnce sync.Once
123+ unblock := func() { releaseOnce.Do(func() { close(release) }) }
124+ var gets atomic.Int32
125+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
126+ gets.Add(1)
127+ w.Header().Set("Content-Length", fmt.Sprint(len(body)))
128+ _, _ = io.WriteString(w, body[:len(body)/2])
129+ w.(http.Flusher).Flush()
130+ close(half)
131+ <-release
132+ _, _ = io.WriteString(w, body[len(body)/2:])
133+ })
134+ t.Cleanup(unblock) // before srv.Close, which waits for the handler
135+ p := proxyWith(t, srv, Config{})
136+ path := "/v2/library/alpine/blobs/" + sum(body)
137+
138+ ctx, cancel := context.WithCancel(context.Background())
139+ first := httptest.NewRecorder()
140+ done := make(chan struct{})
141+ go func() {
142+ defer close(done)
143+ p.ServeHTTP(first, httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
144+ }()
145+ <-half
146+ cancel()
147+ <-done
148+ if strings.Contains(first.Body.String(), "layer") {
149+ t.Fatal("unverified bytes sent to the initiator")
150+ }
151+ key := kindBlob.key(sum(body)[7:])
152+ p.s.mu.Lock()
153+ left := waiters(p.s, key)
154+ p.s.mu.Unlock()
155+ if left != 0 {
156+ t.Fatalf("waiters after the disconnect: %d", left)
157+ }
158+
159+ second := httptest.NewRecorder()
160+ done = make(chan struct{})
161+ go func() {
162+ defer close(done)
163+ p.ServeHTTP(second, httptest.NewRequest(http.MethodGet, path, nil))
164+ }()
165+ waitFor(t, p.s, "the second client to join", func() bool { return waiters(p.s, key) == 1 })
166+ unblock()
167+ <-done
168+ if second.Code != http.StatusOK || second.Body.String() != body {
169+ t.Fatalf("second client: %d", second.Code)
170+ }
171+ if _, err := os.Stat(filepath.Join(p.s.dir, key)); err != nil {
172+ t.Fatalf("blob not committed: %v", err)
173+ }
174+ if gets.Load() != 1 {
175+ t.Fatalf("%d upstream fetches", gets.Load())
176+ }
177+}
178+
179+// A client waiting for a progressing download is kept alive, so the cache
180+// channel's progress deadline follows the download.
181+func TestWaiterKeepAlive(t *testing.T) {
182+ body := strings.Repeat("layer", 4000)
183+ srv := upstreamWith(t, slowBlob(body, 10, 30*time.Millisecond))
184+ p := proxyWith(t, srv, Config{KeepAliveInterval: 10 * time.Millisecond})
185+ w := &keepAliveRecorder{ResponseRecorder: httptest.NewRecorder()}
186+ p.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v2/library/alpine/blobs/"+sum(body), nil))
187+ if w.Code != http.StatusOK || w.calls.Load() == 0 {
188+ t.Fatalf("code %d, %d keepalives", w.Code, w.calls.Load())
189+ }
190+}
191+
192+// A panic in a download fails its flight and frees its slots; it does not
193+// leave the waiters hanging.
194+func TestPanickingFetchFailsFlight(t *testing.T) {
195+ c := storeCfg(t.TempDir(), 1<<20)
196+ c.maxDownloads = 1
197+ s := mustOpen(t, c)
198+ q := NewQuota(1)
199+ ctx, cancel := context.WithTimeout(WithQuota(context.Background(), q), 5*time.Second)
200+ defer cancel()
201+ _, _, _, err := s.ensure(ctx, kindBlob, sum("x"), func(context.Context, io.Writer) (string, error) {
202+ panic("boom")
203+ }, keepalive{})
204+ if err == nil || !strings.Contains(err.Error(), "panicked") {
205+ t.Fatalf("err %v", err)
206+ }
207+ s.mu.Lock()
208+ defer s.mu.Unlock()
209+ if s.downloads != 0 || len(s.flights) != 0 || q.used.Load() != 0 || s.pending != 0 {
210+ t.Fatalf("leaked: downloads=%d flights=%d quota=%d pending=%d", s.downloads, len(s.flights), q.used.Load(), s.pending)
211+ }
212+ if ds, _ := os.ReadDir(filepath.Join(s.dir, "tmp")); len(ds) != 0 {
213+ t.Fatal("temp file kept")
214+ }
215+}
216+
217+// A request that joins a manifest flight logs hit=false, with its job.
218+func TestJoinerLogsMiss(t *testing.T) {
219+ release := make(chan struct{})
220+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
221+ w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json")
222+ w.Header().Set("Docker-Content-Digest", sum(manifestBody))
223+ if r.Method == http.MethodGet {
224+ <-release
225+ _, _ = io.WriteString(w, manifestBody)
226+ }
227+ })
228+ var logs syncBuffer
229+ p := proxyWith(t, srv, Config{Log: zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs})})
230+ var wg sync.WaitGroup
231+ for i := range 2 {
232+ wg.Go(func() {
233+ ctx := WithJob(context.Background(), fmt.Sprint("job-", i))
234+ p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/v2/library/alpine/manifests/3", nil).WithContext(ctx))
235+ })
236+ }
237+ key := kindManifest.key(sum(manifestBody)[7:])
238+ waitFor(t, p.s, "two waiters", func() bool { return waiters(p.s, key) == 2 })
239+ close(release)
240+ wg.Wait()
241+ out := logs.String()
242+ if strings.Count(out, `"hit":false`) != 2 || strings.Contains(out, `"hit":true`) {
243+ t.Fatalf("logs: %s", out)
244+ }
245+ if !strings.Contains(out, `"job_id":"job-0"`) || !strings.Contains(out, `"job_id":"job-1"`) {
246+ t.Fatalf("no job_id on the registry lines: %s", out)
247+ }
248+}
@@ -0,0 +1,248 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "fmt"
8+ "io"
9+ "net/http"
10+ "net/http/httptest"
11+ "os"
12+ "path/filepath"
13+ "strings"
14+ "sync"
15+ "sync/atomic"
16+ "testing"
17+ "time"
18+
19+ "github.com/bnema/zerowrap"
20+)
21+
22+// Concurrent requests for one digest share a single fetch.
23+func TestSingleFlight(t *testing.T) {
24+ s := mustOpen(t, storeCfg(t.TempDir(), 1<<20))
25+ var calls atomic.Int32
26+ var wg sync.WaitGroup
27+ start := make(chan struct{})
28+ for range 20 {
29+ wg.Go(func() {
30+ <-start
31+ f, _, _, err := s.ensure(context.Background(), kindBlob, sum("content"), func(_ context.Context, w io.Writer) (string, error) {
32+ calls.Add(1)
33+ _, err := io.WriteString(w, "content")
34+ return "", err
35+ }, keepalive{})
36+ if err != nil {
37+ t.Error(err)
38+ return
39+ }
40+ _ = f.Close()
41+ })
42+ }
43+ close(start)
44+ wg.Wait()
45+ if calls.Load() != 1 {
46+ t.Fatalf("fetches %d", calls.Load())
47+ }
48+}
49+
50+// A guest that requests many digests and disconnects leaves no download
51+// behind: once a flight has no waiter for the grace period, it is cancelled
52+// upstream and its temporary file removed.
53+func TestAbandonedDownloadsCancelled(t *testing.T) {
54+ u := newStallingUpstream(t)
55+ const n = 20
56+ p := proxyWith(t, u.srv, Config{AbandonGrace: 20 * time.Millisecond, MaxDownloads: n})
57+ for i := range n {
58+ serveCancelled(t, p, u, blobPath(i), int32(i+1))
59+ }
60+ waitFor(t, p.s, "every flight to end", func() bool { return len(p.s.flights) == 0 && p.s.downloads == 0 && p.s.pending == 0 })
61+ for end := time.Now().Add(5 * time.Second); u.cancelled.Load() != n; {
62+ if time.Now().After(end) {
63+ t.Fatalf("upstream saw %d cancellations of %d", u.cancelled.Load(), n)
64+ }
65+ time.Sleep(time.Millisecond)
66+ }
67+ waitFor(t, p.s, "tmp/ to be empty", func() bool {
68+ ds, _ := os.ReadDir(filepath.Join(p.s.dir, "tmp"))
69+ return len(ds) == 0
70+ })
71+}
72+
73+// A request that joins within the grace period keeps the download.
74+func TestRejoinWithinGraceKeepsDownload(t *testing.T) {
75+ body := strings.Repeat("layer", 4000)
76+ release := make(chan struct{})
77+ var gets, cancelled atomic.Int32
78+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
79+ gets.Add(1)
80+ w.Header().Set("Content-Length", fmt.Sprint(len(body)))
81+ _, _ = io.WriteString(w, body[:10])
82+ w.(http.Flusher).Flush()
83+ select {
84+ case <-release:
85+ _, _ = io.WriteString(w, body[10:])
86+ case <-r.Context().Done():
87+ cancelled.Add(1)
88+ }
89+ })
90+ var once sync.Once
91+ t.Cleanup(func() { once.Do(func() { close(release) }) })
92+ p := proxyWith(t, srv, Config{AbandonGrace: time.Hour})
93+ path := "/v2/library/alpine/blobs/" + sum(body)
94+ ctx, cancel := context.WithCancel(context.Background())
95+ first := make(chan struct{})
96+ go func() {
97+ defer close(first)
98+ p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
99+ }()
100+ key := kindBlob.key(sum(body)[7:])
101+ waitFor(t, p.s, "the first waiter", func() bool { return waiters(p.s, key) == 1 })
102+ cancel()
103+ <-first
104+ w := httptest.NewRecorder()
105+ second := make(chan struct{})
106+ go func() { defer close(second); p.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil)) }()
107+ waitFor(t, p.s, "the second waiter", func() bool { return waiters(p.s, key) == 1 })
108+ once.Do(func() { close(release) })
109+ <-second
110+ if w.Code != http.StatusOK || w.Body.String() != body || gets.Load() != 1 || cancelled.Load() != 0 {
111+ t.Fatalf("code %d, %d fetches, %d cancelled", w.Code, gets.Load(), cancelled.Load())
112+ }
113+}
114+
115+// The shared download does not belong to the client that started it. That
116+// client disconnects mid-download; a second client, joining within the grace
117+// period, still gets the blob and the blob is committed, from one upstream
118+// fetch.
119+func TestBlobFlightSurvivesInitiatorDisconnect(t *testing.T) {
120+ body := strings.Repeat("layer", 4000)
121+ half, release := make(chan struct{}), make(chan struct{})
122+ var releaseOnce sync.Once
123+ unblock := func() { releaseOnce.Do(func() { close(release) }) }
124+ var gets atomic.Int32
125+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
126+ gets.Add(1)
127+ w.Header().Set("Content-Length", fmt.Sprint(len(body)))
128+ _, _ = io.WriteString(w, body[:len(body)/2])
129+ w.(http.Flusher).Flush()
130+ close(half)
131+ <-release
132+ _, _ = io.WriteString(w, body[len(body)/2:])
133+ })
134+ t.Cleanup(unblock) // before srv.Close, which waits for the handler
135+ p := proxyWith(t, srv, Config{})
136+ path := "/v2/library/alpine/blobs/" + sum(body)
137+
138+ ctx, cancel := context.WithCancel(context.Background())
139+ first := httptest.NewRecorder()
140+ done := make(chan struct{})
141+ go func() {
142+ defer close(done)
143+ p.ServeHTTP(first, httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
144+ }()
145+ <-half
146+ cancel()
147+ <-done
148+ if strings.Contains(first.Body.String(), "layer") {
149+ t.Fatal("unverified bytes sent to the initiator")
150+ }
151+ key := kindBlob.key(sum(body)[7:])
152+ p.s.mu.Lock()
153+ left := waiters(p.s, key)
154+ p.s.mu.Unlock()
155+ if left != 0 {
156+ t.Fatalf("waiters after the disconnect: %d", left)
157+ }
158+
159+ second := httptest.NewRecorder()
160+ done = make(chan struct{})
161+ go func() {
162+ defer close(done)
163+ p.ServeHTTP(second, httptest.NewRequest(http.MethodGet, path, nil))
164+ }()
165+ waitFor(t, p.s, "the second client to join", func() bool { return waiters(p.s, key) == 1 })
166+ unblock()
167+ <-done
168+ if second.Code != http.StatusOK || second.Body.String() != body {
169+ t.Fatalf("second client: %d", second.Code)
170+ }
171+ if _, err := os.Stat(filepath.Join(p.s.dir, key)); err != nil {
172+ t.Fatalf("blob not committed: %v", err)
173+ }
174+ if gets.Load() != 1 {
175+ t.Fatalf("%d upstream fetches", gets.Load())
176+ }
177+}
178+
179+// A client waiting for a progressing download is kept alive, so the cache
180+// channel's progress deadline follows the download.
181+func TestWaiterKeepAlive(t *testing.T) {
182+ body := strings.Repeat("layer", 4000)
183+ srv := upstreamWith(t, slowBlob(body, 10, 30*time.Millisecond))
184+ p := proxyWith(t, srv, Config{KeepAliveInterval: 10 * time.Millisecond})
185+ w := &keepAliveRecorder{ResponseRecorder: httptest.NewRecorder()}
186+ p.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v2/library/alpine/blobs/"+sum(body), nil))
187+ if w.Code != http.StatusOK || w.calls.Load() == 0 {
188+ t.Fatalf("code %d, %d keepalives", w.Code, w.calls.Load())
189+ }
190+}
191+
192+// A panic in a download fails its flight and frees its slots; it does not
193+// leave the waiters hanging.
194+func TestPanickingFetchFailsFlight(t *testing.T) {
195+ c := storeCfg(t.TempDir(), 1<<20)
196+ c.maxDownloads = 1
197+ s := mustOpen(t, c)
198+ q := NewQuota(1)
199+ ctx, cancel := context.WithTimeout(WithQuota(context.Background(), q), 5*time.Second)
200+ defer cancel()
201+ _, _, _, err := s.ensure(ctx, kindBlob, sum("x"), func(context.Context, io.Writer) (string, error) {
202+ panic("boom")
203+ }, keepalive{})
204+ if err == nil || !strings.Contains(err.Error(), "panicked") {
205+ t.Fatalf("err %v", err)
206+ }
207+ s.mu.Lock()
208+ defer s.mu.Unlock()
209+ if s.downloads != 0 || len(s.flights) != 0 || q.used.Load() != 0 || s.pending != 0 {
210+ t.Fatalf("leaked: downloads=%d flights=%d quota=%d pending=%d", s.downloads, len(s.flights), q.used.Load(), s.pending)
211+ }
212+ if ds, _ := os.ReadDir(filepath.Join(s.dir, "tmp")); len(ds) != 0 {
213+ t.Fatal("temp file kept")
214+ }
215+}
216+
217+// A request that joins a manifest flight logs hit=false, with its job.
218+func TestJoinerLogsMiss(t *testing.T) {
219+ release := make(chan struct{})
220+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
221+ w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json")
222+ w.Header().Set("Docker-Content-Digest", sum(manifestBody))
223+ if r.Method == http.MethodGet {
224+ <-release
225+ _, _ = io.WriteString(w, manifestBody)
226+ }
227+ })
228+ var logs syncBuffer
229+ p := proxyWith(t, srv, Config{Log: zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs})})
230+ var wg sync.WaitGroup
231+ for i := range 2 {
232+ wg.Go(func() {
233+ ctx := WithJob(context.Background(), fmt.Sprint("job-", i))
234+ p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/v2/library/alpine/manifests/3", nil).WithContext(ctx))
235+ })
236+ }
237+ key := kindManifest.key(sum(manifestBody)[7:])
238+ waitFor(t, p.s, "two waiters", func() bool { return waiters(p.s, key) == 2 })
239+ close(release)
240+ wg.Wait()
241+ out := logs.String()
242+ if strings.Count(out, `"hit":false`) != 2 || strings.Contains(out, `"hit":true`) {
243+ t.Fatalf("logs: %s", out)
244+ }
245+ if !strings.Contains(out, `"job_id":"job-0"`) || !strings.Contains(out, `"job_id":"job-1"`) {
246+ t.Fatalf("no job_id on the registry lines: %s", out)
247+ }
248+}
A internal/registryproxy/helpers_test.go +375 -0
@@ -0,0 +1,375 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "bytes"
7+ "context"
8+ "crypto/sha256"
9+ "fmt"
10+ "io"
11+ "net/http"
12+ "net/http/httptest"
13+ "os"
14+ "path/filepath"
15+ "strings"
16+ "sync"
17+ "sync/atomic"
18+ "testing"
19+ "time"
20+)
21+
22+// TestMain fails the run if any store hit an accounting bug (unpinLocked of
23+// an unpinned entry), whichever test caused it.
24+func TestMain(m *testing.M) {
25+ var bugs atomic.Int32
26+ defaultBugHook = func(msg string) {
27+ bugs.Add(1)
28+ fmt.Fprintln(os.Stderr, "registry cache bug:", msg)
29+ }
30+ code := m.Run()
31+ if n := bugs.Load(); n > 0 && code == 0 {
32+ fmt.Fprintf(os.Stderr, "FAIL: %d registry cache accounting bugs\n", n)
33+ code = 1
34+ }
35+ os.Exit(code)
36+}
37+
38+const manifestBody = `{"schemaVersion":2}`
39+const blobBody = "a layer of bytes"
40+
41+func sum(b string) string { return fmt.Sprintf("sha256:%x", sha256.Sum256([]byte(b))) }
42+
43+// storeCfg is a complete configuration for a test that opens a store directly.
44+func storeCfg(dir string, max int64) storeConfig {
45+ return storeConfig{dir: dir, max: max, now: time.Now, grace: time.Second, maxDownloads: 16, slotWait: time.Second}
46+}
47+
48+func mustOpen(t *testing.T, c storeConfig) *store {
49+ t.Helper()
50+ s, err := openStore(c)
51+ if err != nil {
52+ t.Fatal(err)
53+ }
54+ return s
55+}
56+
57+// put stores b as a blob.
58+func put(t *testing.T, s *store, b string) {
59+ t.Helper()
60+ f, _, _, err := s.ensure(context.Background(), kindBlob, sum(b), func(_ context.Context, w io.Writer) (string, error) { _, err := io.WriteString(w, b); return "", err }, keepalive{})
61+ if err != nil {
62+ t.Fatal(err)
63+ }
64+ _ = f.Close()
65+}
66+
67+// putManifest stores m as a manifest of media type media.
68+func putManifest(t *testing.T, s *store, m, media string) {
69+ t.Helper()
70+ f, _, _, err := s.ensure(context.Background(), kindManifest, sum(m), func(_ context.Context, w io.Writer) (string, error) {
71+ _, err := io.WriteString(w, m)
72+ return media, err
73+ }, keepalive{})
74+ if err != nil {
75+ t.Fatal(err)
76+ }
77+ _ = f.Close()
78+}
79+
80+// accounted checks that the store's total is exactly the sum of its entries and
81+// within Max, and returns the number of tag files on disk.
82+func accounted(t *testing.T, s *store) int {
83+ t.Helper()
84+ s.mu.Lock()
85+ defer s.mu.Unlock()
86+ var sum int64
87+ tags := 0
88+ for k, e := range s.entries {
89+ sum += e.size
90+ if e.size < minCharge {
91+ t.Errorf("%s charged %d, under the minimum", k, e.size)
92+ }
93+ if e.kind == kindTag {
94+ tags++
95+ if e.size != minCharge {
96+ t.Errorf("tag %s charged %d", k, e.size)
97+ }
98+ }
99+ }
100+ if sum != s.total || s.total+s.pending > s.max {
101+ t.Fatalf("accounting: sum=%d total=%d pending=%d max=%d", sum, s.total, s.pending, s.max)
102+ }
103+ files, err := os.ReadDir(filepath.Join(s.dir, kindTag.dir()))
104+ if err != nil {
105+ t.Fatal(err)
106+ }
107+ if len(files) != tags {
108+ t.Fatalf("%d tag files, %d tag entries", len(files), tags)
109+ }
110+ return tags
111+}
112+
113+// waitFor polls cond (under the store lock) until it holds.
114+func waitFor(t *testing.T, s *store, what string, cond func() bool) {
115+ t.Helper()
116+ for end := time.Now().Add(5 * time.Second); ; {
117+ s.mu.Lock()
118+ ok := cond()
119+ s.mu.Unlock()
120+ if ok {
121+ return
122+ }
123+ if time.Now().After(end) {
124+ t.Fatalf("timed out waiting for %s", what)
125+ }
126+ time.Sleep(time.Millisecond)
127+ }
128+}
129+
130+// waiters is the number of requests waiting for the flight of key (-1: none).
131+// The caller holds the store lock.
132+func waiters(s *store, key string) int {
133+ if fl, ok := s.flights[key]; ok {
134+ return fl.waiters
135+ }
136+ return -1
137+}
138+
139+// upstreamWith fakes Docker Hub: /token answers an anonymous token and h serves
140+// everything else.
141+func upstreamWith(t *testing.T, h http.HandlerFunc) *httptest.Server {
142+ t.Helper()
143+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
144+ if r.URL.Path == "/token" {
145+ _, _ = io.WriteString(w, `{"token":"public","expires_in":300}`)
146+ return
147+ }
148+ h(w, r)
149+ }))
150+ t.Cleanup(srv.Close)
151+ return srv
152+}
153+
154+func proxyWith(t *testing.T, srv *httptest.Server, c Config) *Proxy {
155+ t.Helper()
156+ if c.Dir == "" {
157+ c.Dir = t.TempDir()
158+ }
159+ if c.Max == 0 {
160+ c.Max = 8 << 20
161+ }
162+ c.Upstream, c.AuthURL = srv.URL, srv.URL+"/token"
163+ p, err := New(c)
164+ if err != nil {
165+ t.Fatal(err)
166+ }
167+ return p
168+}
169+
170+// fakeRegistry is a Docker Hub with one manifest and one blob, served from a
171+// CDN behind a redirect. It counts the requests it gets.
172+type fakeRegistry struct {
173+ server *httptest.Server
174+ heads, gets, blobs, tokens atomic.Int32
175+ status atomic.Int32
176+ corrupt atomic.Bool
177+}
178+
179+func fake(t *testing.T) *fakeRegistry {
180+ t.Helper()
181+ f := &fakeRegistry{}
182+ cdn := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
183+ if r.Header.Get("Authorization") != "" {
184+ t.Error("bearer leaked to CDN")
185+ }
186+ w.Header().Set("Content-Length", fmt.Sprint(len(blobBody)))
187+ if r.Method != http.MethodHead {
188+ b := blobBody
189+ if f.corrupt.Load() {
190+ b = strings.Repeat("x", len(b))
191+ }
192+ _, _ = io.WriteString(w, b)
193+ }
194+ }))
195+ t.Cleanup(cdn.Close)
196+ f.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
197+ if r.URL.Path == "/token" {
198+ f.tokens.Add(1)
199+ if r.Header.Get("Authorization") != "" {
200+ t.Error("guest credentials forwarded to auth")
201+ }
202+ if r.URL.Query().Get("scope") != "repository:library/alpine:pull" {
203+ t.Error("wrong scope")
204+ }
205+ _, _ = io.WriteString(w, `{"token":"public","expires_in":300}`)
206+ return
207+ }
208+ if r.Header.Get("Authorization") != "Bearer public" {
209+ t.Errorf("wrong bearer %q", r.Header.Get("Authorization"))
210+ }
211+ if s := f.status.Load(); s != 0 {
212+ w.WriteHeader(int(s))
213+ return
214+ }
215+ if strings.Contains(r.URL.Path, "/manifests/") {
216+ w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json")
217+ w.Header().Set("Docker-Content-Digest", sum(manifestBody))
218+ w.Header().Set("Content-Length", fmt.Sprint(len(manifestBody)))
219+ if r.Method == http.MethodHead {
220+ f.heads.Add(1)
221+ } else {
222+ f.gets.Add(1)
223+ _, _ = io.WriteString(w, manifestBody)
224+ }
225+ return
226+ }
227+ if strings.Contains(r.URL.Path, "/blobs/") {
228+ if r.Method == http.MethodGet {
229+ f.blobs.Add(1)
230+ }
231+ http.Redirect(w, r, cdn.URL+"/layer", http.StatusTemporaryRedirect)
232+ return
233+ }
234+ http.NotFound(w, r)
235+ }))
236+ t.Cleanup(f.server.Close)
237+ return f
238+}
239+
240+func proxyFor(t *testing.T, f *fakeRegistry) *Proxy {
241+ t.Helper()
242+ p, err := New(Config{Dir: t.TempDir(), Max: 8 << 20, Upstream: f.server.URL, AuthURL: f.server.URL + "/token"})
243+ if err != nil {
244+ t.Fatal(err)
245+ }
246+ return p
247+}
248+
249+// request serves one request with guest credentials that must never reach
250+// upstream; headers are name, value pairs.
251+func request(t *testing.T, p http.Handler, method, path string, headers ...string) *httptest.ResponseRecorder {
252+ t.Helper()
253+ r := httptest.NewRequest(method, path, nil)
254+ r.Header.Set("Authorization", "Bearer guest-secret")
255+ for i := 0; i < len(headers); i += 2 {
256+ r.Header.Set(headers[i], headers[i+1])
257+ }
258+ w := httptest.NewRecorder()
259+ p.ServeHTTP(w, r)
260+ return w
261+}
262+
263+// stallingUpstream serves a blob's first bytes and then holds the body until
264+// the request is cancelled; it counts started and cancelled downloads.
265+type stallingUpstream struct {
266+ srv *httptest.Server
267+ started, cancelled atomic.Int32
268+}
269+
270+func newStallingUpstream(t *testing.T) *stallingUpstream {
271+ t.Helper()
272+ u := &stallingUpstream{}
273+ u.srv = upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
274+ u.started.Add(1)
275+ w.Header().Set("Content-Length", "1000000")
276+ _, _ = io.WriteString(w, "partial")
277+ w.(http.Flusher).Flush()
278+ select {
279+ case <-r.Context().Done():
280+ u.cancelled.Add(1)
281+ case <-time.After(10 * time.Second):
282+ }
283+ })
284+ return u
285+}
286+
287+func blobPath(i int) string { return "/v2/library/alpine/blobs/" + sum(fmt.Sprint("blob-", i)) }
288+
289+// serveCancelled starts a request and cancels it once its download has begun.
290+func serveCancelled(t *testing.T, p *Proxy, u *stallingUpstream, path string, started int32) {
291+ t.Helper()
292+ ctx, cancel := context.WithCancel(context.Background())
293+ done := make(chan struct{})
294+ go func() {
295+ defer close(done)
296+ p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
297+ }()
298+ waitFor(t, p.s, "the download to start", func() bool { return u.started.Load() >= started })
299+ cancel()
300+ <-done
301+}
302+
303+// serveAsync serves path with ctx on its own goroutine, into w (a new
304+// recorder when nil); the status code comes on the returned channel.
305+func serveAsync(p *Proxy, ctx context.Context, path string, w *keepAliveRecorder) <-chan int {
306+ if w == nil {
307+ w = &keepAliveRecorder{ResponseRecorder: httptest.NewRecorder()}
308+ }
309+ code := make(chan int, 1)
310+ go func() {
311+ p.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
312+ code <- w.Code
313+ }()
314+ return code
315+}
316+
317+type keepAliveRecorder struct {
318+ *httptest.ResponseRecorder
319+ calls atomic.Int32
320+}
321+
322+func (k *keepAliveRecorder) KeepAlive() { k.calls.Add(1) }
323+
324+// waitKeptAlive waits until w has been kept alive at least once.
325+func waitKeptAlive(t *testing.T, w *keepAliveRecorder) {
326+ t.Helper()
327+ for end := time.Now().Add(5 * time.Second); w.calls.Load() == 0; time.Sleep(time.Millisecond) {
328+ if time.Now().After(end) {
329+ t.Fatal("request never kept alive")
330+ }
331+ }
332+}
333+
334+// blockingWriter holds a response in its first Write until release is closed.
335+type blockingWriter struct {
336+ *httptest.ResponseRecorder
337+ started, release chan struct{}
338+ once sync.Once
339+}
340+
341+func (b *blockingWriter) Write(p []byte) (int, error) {
342+ b.once.Do(func() { close(b.started) })
343+ <-b.release
344+ return b.ResponseRecorder.Write(p)
345+}
346+
347+// slowBlob streams body in chunks, one every gap.
348+func slowBlob(body string, chunks int, gap time.Duration) http.HandlerFunc {
349+ return func(w http.ResponseWriter, r *http.Request) {
350+ w.Header().Set("Content-Length", fmt.Sprint(len(body)))
351+ size := len(body) / chunks
352+ for i := 0; i < len(body); i += size {
353+ _, _ = io.WriteString(w, body[i:min(i+size, len(body))])
354+ w.(http.Flusher).Flush()
355+ select {
356+ case <-time.After(gap):
357+ case <-r.Context().Done():
358+ return
359+ }
360+ }
361+ }
362+}
363+
364+type syncBuffer struct {
365+ mu sync.Mutex
366+ b bytes.Buffer
367+}
368+
369+func (s *syncBuffer) Write(p []byte) (int, error) {
370+ s.mu.Lock()
371+ defer s.mu.Unlock()
372+ return s.b.Write(p)
373+}
374+
375+func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }
@@ -0,0 +1,375 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "bytes"
7+ "context"
8+ "crypto/sha256"
9+ "fmt"
10+ "io"
11+ "net/http"
12+ "net/http/httptest"
13+ "os"
14+ "path/filepath"
15+ "strings"
16+ "sync"
17+ "sync/atomic"
18+ "testing"
19+ "time"
20+)
21+
22+// TestMain fails the run if any store hit an accounting bug (unpinLocked of
23+// an unpinned entry), whichever test caused it.
24+func TestMain(m *testing.M) {
25+ var bugs atomic.Int32
26+ defaultBugHook = func(msg string) {
27+ bugs.Add(1)
28+ fmt.Fprintln(os.Stderr, "registry cache bug:", msg)
29+ }
30+ code := m.Run()
31+ if n := bugs.Load(); n > 0 && code == 0 {
32+ fmt.Fprintf(os.Stderr, "FAIL: %d registry cache accounting bugs\n", n)
33+ code = 1
34+ }
35+ os.Exit(code)
36+}
37+
38+const manifestBody = `{"schemaVersion":2}`
39+const blobBody = "a layer of bytes"
40+
41+func sum(b string) string { return fmt.Sprintf("sha256:%x", sha256.Sum256([]byte(b))) }
42+
43+// storeCfg is a complete configuration for a test that opens a store directly.
44+func storeCfg(dir string, max int64) storeConfig {
45+ return storeConfig{dir: dir, max: max, now: time.Now, grace: time.Second, maxDownloads: 16, slotWait: time.Second}
46+}
47+
48+func mustOpen(t *testing.T, c storeConfig) *store {
49+ t.Helper()
50+ s, err := openStore(c)
51+ if err != nil {
52+ t.Fatal(err)
53+ }
54+ return s
55+}
56+
57+// put stores b as a blob.
58+func put(t *testing.T, s *store, b string) {
59+ t.Helper()
60+ f, _, _, err := s.ensure(context.Background(), kindBlob, sum(b), func(_ context.Context, w io.Writer) (string, error) { _, err := io.WriteString(w, b); return "", err }, keepalive{})
61+ if err != nil {
62+ t.Fatal(err)
63+ }
64+ _ = f.Close()
65+}
66+
67+// putManifest stores m as a manifest of media type media.
68+func putManifest(t *testing.T, s *store, m, media string) {
69+ t.Helper()
70+ f, _, _, err := s.ensure(context.Background(), kindManifest, sum(m), func(_ context.Context, w io.Writer) (string, error) {
71+ _, err := io.WriteString(w, m)
72+ return media, err
73+ }, keepalive{})
74+ if err != nil {
75+ t.Fatal(err)
76+ }
77+ _ = f.Close()
78+}
79+
80+// accounted checks that the store's total is exactly the sum of its entries and
81+// within Max, and returns the number of tag files on disk.
82+func accounted(t *testing.T, s *store) int {
83+ t.Helper()
84+ s.mu.Lock()
85+ defer s.mu.Unlock()
86+ var sum int64
87+ tags := 0
88+ for k, e := range s.entries {
89+ sum += e.size
90+ if e.size < minCharge {
91+ t.Errorf("%s charged %d, under the minimum", k, e.size)
92+ }
93+ if e.kind == kindTag {
94+ tags++
95+ if e.size != minCharge {
96+ t.Errorf("tag %s charged %d", k, e.size)
97+ }
98+ }
99+ }
100+ if sum != s.total || s.total+s.pending > s.max {
101+ t.Fatalf("accounting: sum=%d total=%d pending=%d max=%d", sum, s.total, s.pending, s.max)
102+ }
103+ files, err := os.ReadDir(filepath.Join(s.dir, kindTag.dir()))
104+ if err != nil {
105+ t.Fatal(err)
106+ }
107+ if len(files) != tags {
108+ t.Fatalf("%d tag files, %d tag entries", len(files), tags)
109+ }
110+ return tags
111+}
112+
113+// waitFor polls cond (under the store lock) until it holds.
114+func waitFor(t *testing.T, s *store, what string, cond func() bool) {
115+ t.Helper()
116+ for end := time.Now().Add(5 * time.Second); ; {
117+ s.mu.Lock()
118+ ok := cond()
119+ s.mu.Unlock()
120+ if ok {
121+ return
122+ }
123+ if time.Now().After(end) {
124+ t.Fatalf("timed out waiting for %s", what)
125+ }
126+ time.Sleep(time.Millisecond)
127+ }
128+}
129+
130+// waiters is the number of requests waiting for the flight of key (-1: none).
131+// The caller holds the store lock.
132+func waiters(s *store, key string) int {
133+ if fl, ok := s.flights[key]; ok {
134+ return fl.waiters
135+ }
136+ return -1
137+}
138+
139+// upstreamWith fakes Docker Hub: /token answers an anonymous token and h serves
140+// everything else.
141+func upstreamWith(t *testing.T, h http.HandlerFunc) *httptest.Server {
142+ t.Helper()
143+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
144+ if r.URL.Path == "/token" {
145+ _, _ = io.WriteString(w, `{"token":"public","expires_in":300}`)
146+ return
147+ }
148+ h(w, r)
149+ }))
150+ t.Cleanup(srv.Close)
151+ return srv
152+}
153+
154+func proxyWith(t *testing.T, srv *httptest.Server, c Config) *Proxy {
155+ t.Helper()
156+ if c.Dir == "" {
157+ c.Dir = t.TempDir()
158+ }
159+ if c.Max == 0 {
160+ c.Max = 8 << 20
161+ }
162+ c.Upstream, c.AuthURL = srv.URL, srv.URL+"/token"
163+ p, err := New(c)
164+ if err != nil {
165+ t.Fatal(err)
166+ }
167+ return p
168+}
169+
170+// fakeRegistry is a Docker Hub with one manifest and one blob, served from a
171+// CDN behind a redirect. It counts the requests it gets.
172+type fakeRegistry struct {
173+ server *httptest.Server
174+ heads, gets, blobs, tokens atomic.Int32
175+ status atomic.Int32
176+ corrupt atomic.Bool
177+}
178+
179+func fake(t *testing.T) *fakeRegistry {
180+ t.Helper()
181+ f := &fakeRegistry{}
182+ cdn := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
183+ if r.Header.Get("Authorization") != "" {
184+ t.Error("bearer leaked to CDN")
185+ }
186+ w.Header().Set("Content-Length", fmt.Sprint(len(blobBody)))
187+ if r.Method != http.MethodHead {
188+ b := blobBody
189+ if f.corrupt.Load() {
190+ b = strings.Repeat("x", len(b))
191+ }
192+ _, _ = io.WriteString(w, b)
193+ }
194+ }))
195+ t.Cleanup(cdn.Close)
196+ f.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
197+ if r.URL.Path == "/token" {
198+ f.tokens.Add(1)
199+ if r.Header.Get("Authorization") != "" {
200+ t.Error("guest credentials forwarded to auth")
201+ }
202+ if r.URL.Query().Get("scope") != "repository:library/alpine:pull" {
203+ t.Error("wrong scope")
204+ }
205+ _, _ = io.WriteString(w, `{"token":"public","expires_in":300}`)
206+ return
207+ }
208+ if r.Header.Get("Authorization") != "Bearer public" {
209+ t.Errorf("wrong bearer %q", r.Header.Get("Authorization"))
210+ }
211+ if s := f.status.Load(); s != 0 {
212+ w.WriteHeader(int(s))
213+ return
214+ }
215+ if strings.Contains(r.URL.Path, "/manifests/") {
216+ w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json")
217+ w.Header().Set("Docker-Content-Digest", sum(manifestBody))
218+ w.Header().Set("Content-Length", fmt.Sprint(len(manifestBody)))
219+ if r.Method == http.MethodHead {
220+ f.heads.Add(1)
221+ } else {
222+ f.gets.Add(1)
223+ _, _ = io.WriteString(w, manifestBody)
224+ }
225+ return
226+ }
227+ if strings.Contains(r.URL.Path, "/blobs/") {
228+ if r.Method == http.MethodGet {
229+ f.blobs.Add(1)
230+ }
231+ http.Redirect(w, r, cdn.URL+"/layer", http.StatusTemporaryRedirect)
232+ return
233+ }
234+ http.NotFound(w, r)
235+ }))
236+ t.Cleanup(f.server.Close)
237+ return f
238+}
239+
240+func proxyFor(t *testing.T, f *fakeRegistry) *Proxy {
241+ t.Helper()
242+ p, err := New(Config{Dir: t.TempDir(), Max: 8 << 20, Upstream: f.server.URL, AuthURL: f.server.URL + "/token"})
243+ if err != nil {
244+ t.Fatal(err)
245+ }
246+ return p
247+}
248+
249+// request serves one request with guest credentials that must never reach
250+// upstream; headers are name, value pairs.
251+func request(t *testing.T, p http.Handler, method, path string, headers ...string) *httptest.ResponseRecorder {
252+ t.Helper()
253+ r := httptest.NewRequest(method, path, nil)
254+ r.Header.Set("Authorization", "Bearer guest-secret")
255+ for i := 0; i < len(headers); i += 2 {
256+ r.Header.Set(headers[i], headers[i+1])
257+ }
258+ w := httptest.NewRecorder()
259+ p.ServeHTTP(w, r)
260+ return w
261+}
262+
263+// stallingUpstream serves a blob's first bytes and then holds the body until
264+// the request is cancelled; it counts started and cancelled downloads.
265+type stallingUpstream struct {
266+ srv *httptest.Server
267+ started, cancelled atomic.Int32
268+}
269+
270+func newStallingUpstream(t *testing.T) *stallingUpstream {
271+ t.Helper()
272+ u := &stallingUpstream{}
273+ u.srv = upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
274+ u.started.Add(1)
275+ w.Header().Set("Content-Length", "1000000")
276+ _, _ = io.WriteString(w, "partial")
277+ w.(http.Flusher).Flush()
278+ select {
279+ case <-r.Context().Done():
280+ u.cancelled.Add(1)
281+ case <-time.After(10 * time.Second):
282+ }
283+ })
284+ return u
285+}
286+
287+func blobPath(i int) string { return "/v2/library/alpine/blobs/" + sum(fmt.Sprint("blob-", i)) }
288+
289+// serveCancelled starts a request and cancels it once its download has begun.
290+func serveCancelled(t *testing.T, p *Proxy, u *stallingUpstream, path string, started int32) {
291+ t.Helper()
292+ ctx, cancel := context.WithCancel(context.Background())
293+ done := make(chan struct{})
294+ go func() {
295+ defer close(done)
296+ p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
297+ }()
298+ waitFor(t, p.s, "the download to start", func() bool { return u.started.Load() >= started })
299+ cancel()
300+ <-done
301+}
302+
303+// serveAsync serves path with ctx on its own goroutine, into w (a new
304+// recorder when nil); the status code comes on the returned channel.
305+func serveAsync(p *Proxy, ctx context.Context, path string, w *keepAliveRecorder) <-chan int {
306+ if w == nil {
307+ w = &keepAliveRecorder{ResponseRecorder: httptest.NewRecorder()}
308+ }
309+ code := make(chan int, 1)
310+ go func() {
311+ p.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
312+ code <- w.Code
313+ }()
314+ return code
315+}
316+
317+type keepAliveRecorder struct {
318+ *httptest.ResponseRecorder
319+ calls atomic.Int32
320+}
321+
322+func (k *keepAliveRecorder) KeepAlive() { k.calls.Add(1) }
323+
324+// waitKeptAlive waits until w has been kept alive at least once.
325+func waitKeptAlive(t *testing.T, w *keepAliveRecorder) {
326+ t.Helper()
327+ for end := time.Now().Add(5 * time.Second); w.calls.Load() == 0; time.Sleep(time.Millisecond) {
328+ if time.Now().After(end) {
329+ t.Fatal("request never kept alive")
330+ }
331+ }
332+}
333+
334+// blockingWriter holds a response in its first Write until release is closed.
335+type blockingWriter struct {
336+ *httptest.ResponseRecorder
337+ started, release chan struct{}
338+ once sync.Once
339+}
340+
341+func (b *blockingWriter) Write(p []byte) (int, error) {
342+ b.once.Do(func() { close(b.started) })
343+ <-b.release
344+ return b.ResponseRecorder.Write(p)
345+}
346+
347+// slowBlob streams body in chunks, one every gap.
348+func slowBlob(body string, chunks int, gap time.Duration) http.HandlerFunc {
349+ return func(w http.ResponseWriter, r *http.Request) {
350+ w.Header().Set("Content-Length", fmt.Sprint(len(body)))
351+ size := len(body) / chunks
352+ for i := 0; i < len(body); i += size {
353+ _, _ = io.WriteString(w, body[i:min(i+size, len(body))])
354+ w.(http.Flusher).Flush()
355+ select {
356+ case <-time.After(gap):
357+ case <-r.Context().Done():
358+ return
359+ }
360+ }
361+ }
362+}
363+
364+type syncBuffer struct {
365+ mu sync.Mutex
366+ b bytes.Buffer
367+}
368+
369+func (s *syncBuffer) Write(p []byte) (int, error) {
370+ s.mu.Lock()
371+ defer s.mu.Unlock()
372+ return s.b.Write(p)
373+}
374+
375+func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }
A internal/registryproxy/hub_test.go +80 -0
@@ -0,0 +1,80 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "bytes"
7+ "encoding/json"
8+ "net/http/httptest"
9+ "os"
10+ "os/exec"
11+ "strings"
12+ "testing"
13+
14+ "github.com/bnema/zerowrap"
15+)
16+
17+// An opt-in manual check using curl against a real local HTTP listener, with
18+// Docker Hub as upstream. No daemon is required.
19+func TestDockerHubCurl(t *testing.T) {
20+ if os.Getenv("ERMITE_RUNNER_IT_REGISTRY") != "1" {
21+ t.Skip("set ERMITE_RUNNER_IT_REGISTRY=1 for live Docker Hub curl check")
22+ }
23+ var logs bytes.Buffer
24+ p, err := New(Config{Dir: t.TempDir(), Max: 64 << 20, Log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs})})
25+ if err != nil {
26+ t.Fatal(err)
27+ }
28+ srv := httptest.NewServer(p)
29+ defer srv.Close()
30+ curl := func(path string) []byte {
31+ t.Helper()
32+ out, err := exec.Command("curl", "--max-time", "20", "-fsS", "-H", "Accept: application/vnd.oci.image.index.v1+json, application/vnd.oci.image.manifest.v1+json", srv.URL+path).CombinedOutput()
33+ if err != nil {
34+ t.Fatalf("curl %s: %v: %s", path, err, out)
35+ }
36+ t.Logf("curl %s: HTTP 200, %d bytes", path, len(out))
37+ return out
38+ }
39+ var idx struct {
40+ Manifests []struct {
41+ Digest string
42+ Platform struct{ Architecture, OS string }
43+ }
44+ }
45+ b := curl("/v2/library/alpine/manifests/3")
46+ if err := json.Unmarshal(b, &idx); err != nil {
47+ t.Fatal(err)
48+ }
49+ if !bytes.Equal(b, curl("/v2/library/alpine/manifests/3")) {
50+ t.Fatal("tag changed during check")
51+ }
52+ var digest string
53+ for _, m := range idx.Manifests {
54+ if m.Platform.Architecture == "amd64" && m.Platform.OS == "linux" {
55+ digest = m.Digest
56+ break
57+ }
58+ }
59+ if !validDigest(digest) {
60+ t.Fatal("no linux/amd64 manifest")
61+ }
62+ var manifest struct{ Config struct{ Digest string } }
63+ if err := json.Unmarshal(curl("/v2/library/alpine/manifests/"+digest), &manifest); err != nil {
64+ t.Fatal(err)
65+ }
66+ if !validDigest(manifest.Config.Digest) {
67+ t.Fatal("no config blob")
68+ }
69+ path := "/v2/library/alpine/blobs/" + manifest.Config.Digest
70+ first := curl(path)
71+ second := curl(path)
72+ if !bytes.Equal(first, second) {
73+ t.Fatal("blob differs")
74+ }
75+ all := logs.String()
76+ t.Log(all)
77+ if !strings.Contains(all, `"hit":true`) || strings.Count(all, `"message":"registry blob"`) != 1 {
78+ t.Fatal("missing cache hits or repeated blob miss")
79+ }
80+}
@@ -0,0 +1,80 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "bytes"
7+ "encoding/json"
8+ "net/http/httptest"
9+ "os"
10+ "os/exec"
11+ "strings"
12+ "testing"
13+
14+ "github.com/bnema/zerowrap"
15+)
16+
17+// An opt-in manual check using curl against a real local HTTP listener, with
18+// Docker Hub as upstream. No daemon is required.
19+func TestDockerHubCurl(t *testing.T) {
20+ if os.Getenv("ERMITE_RUNNER_IT_REGISTRY") != "1" {
21+ t.Skip("set ERMITE_RUNNER_IT_REGISTRY=1 for live Docker Hub curl check")
22+ }
23+ var logs bytes.Buffer
24+ p, err := New(Config{Dir: t.TempDir(), Max: 64 << 20, Log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs})})
25+ if err != nil {
26+ t.Fatal(err)
27+ }
28+ srv := httptest.NewServer(p)
29+ defer srv.Close()
30+ curl := func(path string) []byte {
31+ t.Helper()
32+ out, err := exec.Command("curl", "--max-time", "20", "-fsS", "-H", "Accept: application/vnd.oci.image.index.v1+json, application/vnd.oci.image.manifest.v1+json", srv.URL+path).CombinedOutput()
33+ if err != nil {
34+ t.Fatalf("curl %s: %v: %s", path, err, out)
35+ }
36+ t.Logf("curl %s: HTTP 200, %d bytes", path, len(out))
37+ return out
38+ }
39+ var idx struct {
40+ Manifests []struct {
41+ Digest string
42+ Platform struct{ Architecture, OS string }
43+ }
44+ }
45+ b := curl("/v2/library/alpine/manifests/3")
46+ if err := json.Unmarshal(b, &idx); err != nil {
47+ t.Fatal(err)
48+ }
49+ if !bytes.Equal(b, curl("/v2/library/alpine/manifests/3")) {
50+ t.Fatal("tag changed during check")
51+ }
52+ var digest string
53+ for _, m := range idx.Manifests {
54+ if m.Platform.Architecture == "amd64" && m.Platform.OS == "linux" {
55+ digest = m.Digest
56+ break
57+ }
58+ }
59+ if !validDigest(digest) {
60+ t.Fatal("no linux/amd64 manifest")
61+ }
62+ var manifest struct{ Config struct{ Digest string } }
63+ if err := json.Unmarshal(curl("/v2/library/alpine/manifests/"+digest), &manifest); err != nil {
64+ t.Fatal(err)
65+ }
66+ if !validDigest(manifest.Config.Digest) {
67+ t.Fatal("no config blob")
68+ }
69+ path := "/v2/library/alpine/blobs/" + manifest.Config.Digest
70+ first := curl(path)
71+ second := curl(path)
72+ if !bytes.Equal(first, second) {
73+ t.Fatal("blob differs")
74+ }
75+ all := logs.String()
76+ t.Log(all)
77+ if !strings.Contains(all, `"hit":true`) || strings.Count(all, `"message":"registry blob"`) != 1 {
78+ t.Fatal("missing cache hits or repeated blob miss")
79+ }
80+}
A internal/registryproxy/proxy.go +338 -0
@@ -0,0 +1,338 @@
1+// SPDX-License-Identifier: MIT
2+
3+// Package registryproxy provides an anonymous, pull-only Docker Hub mirror.
4+// Its handler belongs on a sandbox's private runtime channel, not a public port.
5+package registryproxy
6+
7+import (
8+ "context"
9+ "encoding/hex"
10+ "errors"
11+ "io"
12+ "net/http"
13+ "net/url"
14+ "regexp"
15+ "strconv"
16+ "strings"
17+ "time"
18+
19+ "github.com/bnema/zerowrap"
20+)
21+
22+const manifestLimit = 4 << 20
23+
24+// Config configures a Proxy. Max must be positive. The directory is exclusively
25+// owned by one Proxy; opening it clears incomplete downloads from tmp/.
26+type Config struct {
27+ Dir string
28+ Max int64
29+ Upstream string // default https://registry-1.docker.io
30+ AuthURL string // default https://auth.docker.io/token
31+ Log zerowrap.Logger
32+ // RequestTimeout bounds auth, manifests and HEAD (default 10s).
33+ RequestTimeout time.Duration
34+ // BlobIdleTimeout bounds time without upstream body progress (default 30s).
35+ // A blob body has no total timeout.
36+ BlobIdleTimeout time.Duration
37+ // KeepAliveInterval is how often a client waiting for a cold blob is kept
38+ // alive while the download progresses (default 1s), when its
39+ // ResponseWriter is a KeepAliver.
40+ KeepAliveInterval time.Duration
41+ // AbandonGrace is how long a download nobody waits for any more keeps
42+ // running in case another request joins it; then it is cancelled and its
43+ // temporary file removed (default 5s).
44+ AbandonGrace time.Duration
45+ // MaxDownloads caps the concurrent cold blob downloads across all jobs
46+ // (default 16). A request may also carry a per-client Quota (WithQuota).
47+ MaxDownloads int
48+ // SlotWait is how long a request that would start a download waits for a
49+ // free slot, global or of its quota, before it gets 503 (default 30s).
50+ SlotWait time.Duration
51+}
52+
53+// KeepAliver is a ResponseWriter that can move its connection's deadlines
54+// without writing. While a request waits for a cold blob, the Proxy calls
55+// KeepAlive every KeepAliveInterval as long as the download progresses, so
56+// that a slow but moving download does not time the client out before the
57+// first byte of the reply.
58+type KeepAliver interface {
59+ KeepAlive()
60+}
61+
62+// Proxy implements http.Handler for GET and HEAD under /v2/.
63+type Proxy struct {
64+ s *store
65+ u *upstream
66+ log zerowrap.Logger
67+ keepEvery time.Duration
68+}
69+
70+// New opens the persistent cache. It starts no goroutines or listeners.
71+func New(c Config) (*Proxy, error) {
72+ if c.Upstream == "" {
73+ c.Upstream = "https://registry-1.docker.io"
74+ }
75+ if c.AuthURL == "" {
76+ c.AuthURL = "https://auth.docker.io/token"
77+ }
78+ for _, s := range []string{c.Upstream, c.AuthURL} {
79+ u, err := url.Parse(s)
80+ if err != nil || u.Host == "" || u.User != nil || (u.Scheme != "http" && u.Scheme != "https") {
81+ return nil, errors.New("invalid registry upstream URL")
82+ }
83+ }
84+ if c.RequestTimeout <= 0 {
85+ c.RequestTimeout = 10 * time.Second
86+ }
87+ if c.BlobIdleTimeout <= 0 {
88+ c.BlobIdleTimeout = 30 * time.Second
89+ }
90+ if c.KeepAliveInterval <= 0 {
91+ c.KeepAliveInterval = time.Second
92+ }
93+ if c.AbandonGrace <= 0 {
94+ c.AbandonGrace = 5 * time.Second
95+ }
96+ if c.MaxDownloads <= 0 {
97+ c.MaxDownloads = 16
98+ }
99+ if c.SlotWait <= 0 {
100+ c.SlotWait = 30 * time.Second
101+ }
102+ st, err := openStore(storeConfig{dir: c.Dir, max: c.Max, now: time.Now, grace: c.AbandonGrace,
103+ maxDownloads: c.MaxDownloads, slotWait: c.SlotWait, log: c.Log})
104+ if err != nil {
105+ return nil, err
106+ }
107+ client, blobs := upstreamClients(c.RequestTimeout)
108+ return &Proxy{s: st, u: &upstream{base: strings.TrimRight(c.Upstream, "/"), auth: c.AuthURL, client: client, blobClient: blobs,
109+ requestTimeout: c.RequestTimeout, blobIdle: c.BlobIdleTimeout, tokens: map[string]bearer{}}, log: c.Log, keepEvery: c.KeepAliveInterval}, nil
110+}
111+
112+var namePattern = regexp.MustCompile(`^[a-z0-9]+(?:[._-][a-z0-9]+)*(?:/[a-z0-9]+(?:[._-][a-z0-9]+)*)*$`)
113+var tagPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}$`)
114+
115+func validDigest(s string) bool {
116+ if len(s) != 71 || !strings.HasPrefix(s, "sha256:") {
117+ return false
118+ }
119+ _, err := hex.DecodeString(s[7:])
120+ return err == nil && s == strings.ToLower(s)
121+}
122+
123+func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
124+ if r.Method != http.MethodGet && r.Method != http.MethodHead {
125+ w.WriteHeader(http.StatusMethodNotAllowed)
126+ return
127+ }
128+ w.Header().Set("Docker-Distribution-Api-Version", "registry/2.0")
129+ if r.URL.Path == "/v2/" {
130+ w.Header().Set("Content-Type", "application/json")
131+ w.Header().Set("Content-Length", "2")
132+ if r.Method == http.MethodGet {
133+ _, _ = io.WriteString(w, "{}")
134+ }
135+ return
136+ }
137+ if !strings.HasPrefix(r.URL.Path, "/v2/") {
138+ http.NotFound(w, r)
139+ return
140+ }
141+ path := strings.TrimPrefix(r.URL.Path, "/v2/")
142+ i := strings.LastIndex(path, "/")
143+ if i < 0 {
144+ http.NotFound(w, r)
145+ return
146+ }
147+ ref := path[i+1:]
148+ path = path[:i]
149+ i = strings.LastIndex(path, "/")
150+ if i < 0 {
151+ http.NotFound(w, r)
152+ return
153+ }
154+ name, route := path[:i], path[i+1:]
155+ // dockerd's legacy client sends one Accept line per media type.
156+ accept := strings.Join(r.Header.Values("Accept"), ",")
157+ if len(name) > 255 || !namePattern.MatchString(name) || len(accept) > 8192 {
158+ http.NotFound(w, r)
159+ return
160+ }
161+ if route == kindManifest.dir() && (validDigest(ref) || tagPattern.MatchString(ref)) {
162+ p.manifest(w, r, name, ref, normalizeAccept(accept))
163+ return
164+ }
165+ if route == kindBlob.dir() && validDigest(ref) {
166+ p.blob(w, r, name, ref)
167+ return
168+ }
169+ http.NotFound(w, r)
170+}
171+
172+// logFor is the Proxy's logger, with the job_id of the request (WithJob).
173+func (p *Proxy) logFor(ctx context.Context) zerowrap.Logger {
174+ if job := jobFrom(ctx); job != "" {
175+ return zerowrap.Logger{Logger: p.log.With().Str("job_id", job).Logger()}
176+ }
177+ return p.log
178+}
179+
180+func serveFile(w http.ResponseWriter, r *http.Request, f *storedFile, media, digest string) {
181+ defer func() { _ = f.Close() }()
182+ fi, err := f.Stat()
183+ if err != nil {
184+ http.Error(w, "registry cache read failed", http.StatusBadGateway)
185+ return
186+ }
187+ w.Header().Set("Content-Type", media)
188+ w.Header().Set("Docker-Content-Digest", digest)
189+ http.ServeContent(w, r, "", fi.ModTime(), f)
190+}
191+
192+// serveStaleTag serves the manifest a tag pointed to when upstream could not
193+// answer, if that manifest is still stored. It reports whether it served one.
194+func (p *Proxy) serveStaleTag(w http.ResponseWriter, r *http.Request, name, tag, accept string) bool {
195+ digest := p.s.tag(name, tag, accept)
196+ if !validDigest(digest) {
197+ return false
198+ }
199+ f, media, err := p.s.open(kindManifest, digest)
200+ if err != nil {
201+ return false
202+ }
203+ serveFile(w, r, f, media, digest)
204+ return true
205+}
206+
207+// manifest serves a manifest by tag or digest. accept is normalized.
208+func (p *Proxy) manifest(w http.ResponseWriter, r *http.Request, name, ref, accept string) {
209+ start := time.Now()
210+ hit := false
211+ log := p.logFor(r.Context())
212+ defer func() {
213+ log.Info().Str("repo", name).Str("ref", ref).Bool("hit", hit).Int64("ms", time.Since(start).Milliseconds()).Msg("registry manifest")
214+ }()
215+ ctx, cancel := context.WithTimeout(r.Context(), p.u.requestTimeout)
216+ defer cancel()
217+ digest := ref
218+ tag := !validDigest(ref)
219+ if tag {
220+ resp, err := p.u.request(ctx, http.MethodHead, name, kindManifest, ref, accept)
221+ if err != nil {
222+ if unavailable(err) && p.serveStaleTag(w, r, name, ref, accept) {
223+ hit = true
224+ return
225+ }
226+ http.Error(w, "registry upstream unavailable", proxyStatus(err))
227+ return
228+ }
229+ digest = resp.Header.Get("Docker-Content-Digest")
230+ _ = resp.Body.Close()
231+ if !validDigest(digest) {
232+ http.Error(w, "invalid upstream digest", http.StatusBadGateway)
233+ return
234+ }
235+ }
236+ // Shared flight, as for blobs: it must not end with the initiator's request.
237+ f, media, cached, err := p.s.ensure(ctx, kindManifest, digest, func(flightCtx context.Context, dst io.Writer) (string, error) {
238+ fillCtx, fillCancel := context.WithTimeout(flightCtx, p.u.requestTimeout)
239+ defer fillCancel()
240+ resp, err := p.u.request(fillCtx, http.MethodGet, name, kindManifest, digest, accept)
241+ if err != nil {
242+ return "", err
243+ }
244+ defer func() { _ = resp.Body.Close() }()
245+ b, err := io.ReadAll(io.LimitReader(resp.Body, manifestLimit+1))
246+ if err != nil {
247+ return "", err
248+ }
249+ if len(b) > manifestLimit {
250+ return "", errTooLarge
251+ }
252+ _, err = dst.Write(b)
253+ return resp.Header.Get("Content-Type"), err
254+ }, keepalive{})
255+ hit = cached
256+ if err != nil {
257+ if tag && unavailable(err) && p.serveStaleTag(w, r, name, ref, accept) {
258+ hit = true
259+ return
260+ }
261+ http.Error(w, "registry fetch failed", proxyStatus(err))
262+ return
263+ }
264+ if tag {
265+ _ = p.s.setTag(name, ref, accept, digest)
266+ }
267+ serveFile(w, r, f, media, digest)
268+}
269+
270+func (p *Proxy) blob(w http.ResponseWriter, r *http.Request, name, digest string) {
271+ if f, media, err := p.s.open(kindBlob, digest); err == nil {
272+ serveFile(w, r, f, media, digest)
273+ return
274+ }
275+ // A cold HEAD is metadata-only and retains the short total timeout.
276+ if r.Method == http.MethodHead {
277+ ctx, cancel := context.WithTimeout(r.Context(), p.u.requestTimeout)
278+ defer cancel()
279+ resp, err := p.u.request(ctx, http.MethodHead, name, kindBlob, digest, "")
280+ if err != nil {
281+ http.Error(w, "registry fetch failed", proxyStatus(err))
282+ return
283+ }
284+ defer func() { _ = resp.Body.Close() }()
285+ if resp.ContentLength < 0 {
286+ http.Error(w, "unknown blob length", http.StatusBadGateway)
287+ return
288+ }
289+ w.Header().Set("Content-Type", "application/octet-stream")
290+ w.Header().Set("Content-Length", strconv.FormatInt(resp.ContentLength, 10))
291+ w.Header().Set("Docker-Content-Digest", digest)
292+ return
293+ }
294+ // The flight is shared by every client that asks for this blob, so no client
295+ // may own it: it downloads on a context of its own, into the temporary file
296+ // only, and is bounded by Max and the upstream idle timeout. Every client,
297+ // the initiator included, waits for the verified commit and then reads the
298+ // stored file (simplest correct option: no client streams unverified bytes,
299+ // and a client that disconnects fails only itself). The flight is cancelled
300+ // when no client has waited for it for AbandonGrace. At most MaxDownloads
301+ // run at once, and at most its Quota per client: a request over either
302+ // waits up to SlotWait for a slot, then gets 503, which dockerd retries
303+ // (about 50 s) before pulling from Docker Hub. While it waits, a
304+ // KeepAliver writer (the cache channel's progress deadline) is kept alive
305+ // as long as the download moves.
306+ keep := keepalive{every: p.keepEvery}
307+ if k, ok := w.(KeepAliver); ok {
308+ keep.fn = k.KeepAlive
309+ }
310+ log := p.logFor(r.Context())
311+ f, media, _, err := p.s.ensure(r.Context(), kindBlob, digest, func(downloadCtx context.Context, dst io.Writer) (string, error) {
312+ start := time.Now()
313+ var size int64
314+ defer func() {
315+ log.Debug().Str("digest", digest).Int64("bytes", size).Int64("ms", time.Since(start).Milliseconds()).Msg("registry blob")
316+ }()
317+ resp, err := p.u.blob(downloadCtx, name, digest)
318+ if err != nil {
319+ return "", err
320+ }
321+ defer func() { _ = resp.Body.Close() }()
322+ if resp.ContentLength > p.s.max {
323+ return "", errTooLarge
324+ }
325+ size, err = io.Copy(dst, resp.Body)
326+ return "application/octet-stream", err
327+ }, keep)
328+ if errors.Is(err, errBusy) {
329+ w.Header().Set("Retry-After", "1")
330+ http.Error(w, "registry downloads busy", http.StatusServiceUnavailable)
331+ return
332+ }
333+ if err != nil {
334+ http.Error(w, "registry fetch failed", proxyStatus(err))
335+ return
336+ }
337+ serveFile(w, r, f, media, digest)
338+}
@@ -0,0 +1,338 @@
1+// SPDX-License-Identifier: MIT
2+
3+// Package registryproxy provides an anonymous, pull-only Docker Hub mirror.
4+// Its handler belongs on a sandbox's private runtime channel, not a public port.
5+package registryproxy
6+
7+import (
8+ "context"
9+ "encoding/hex"
10+ "errors"
11+ "io"
12+ "net/http"
13+ "net/url"
14+ "regexp"
15+ "strconv"
16+ "strings"
17+ "time"
18+
19+ "github.com/bnema/zerowrap"
20+)
21+
22+const manifestLimit = 4 << 20
23+
24+// Config configures a Proxy. Max must be positive. The directory is exclusively
25+// owned by one Proxy; opening it clears incomplete downloads from tmp/.
26+type Config struct {
27+ Dir string
28+ Max int64
29+ Upstream string // default https://registry-1.docker.io
30+ AuthURL string // default https://auth.docker.io/token
31+ Log zerowrap.Logger
32+ // RequestTimeout bounds auth, manifests and HEAD (default 10s).
33+ RequestTimeout time.Duration
34+ // BlobIdleTimeout bounds time without upstream body progress (default 30s).
35+ // A blob body has no total timeout.
36+ BlobIdleTimeout time.Duration
37+ // KeepAliveInterval is how often a client waiting for a cold blob is kept
38+ // alive while the download progresses (default 1s), when its
39+ // ResponseWriter is a KeepAliver.
40+ KeepAliveInterval time.Duration
41+ // AbandonGrace is how long a download nobody waits for any more keeps
42+ // running in case another request joins it; then it is cancelled and its
43+ // temporary file removed (default 5s).
44+ AbandonGrace time.Duration
45+ // MaxDownloads caps the concurrent cold blob downloads across all jobs
46+ // (default 16). A request may also carry a per-client Quota (WithQuota).
47+ MaxDownloads int
48+ // SlotWait is how long a request that would start a download waits for a
49+ // free slot, global or of its quota, before it gets 503 (default 30s).
50+ SlotWait time.Duration
51+}
52+
53+// KeepAliver is a ResponseWriter that can move its connection's deadlines
54+// without writing. While a request waits for a cold blob, the Proxy calls
55+// KeepAlive every KeepAliveInterval as long as the download progresses, so
56+// that a slow but moving download does not time the client out before the
57+// first byte of the reply.
58+type KeepAliver interface {
59+ KeepAlive()
60+}
61+
62+// Proxy implements http.Handler for GET and HEAD under /v2/.
63+type Proxy struct {
64+ s *store
65+ u *upstream
66+ log zerowrap.Logger
67+ keepEvery time.Duration
68+}
69+
70+// New opens the persistent cache. It starts no goroutines or listeners.
71+func New(c Config) (*Proxy, error) {
72+ if c.Upstream == "" {
73+ c.Upstream = "https://registry-1.docker.io"
74+ }
75+ if c.AuthURL == "" {
76+ c.AuthURL = "https://auth.docker.io/token"
77+ }
78+ for _, s := range []string{c.Upstream, c.AuthURL} {
79+ u, err := url.Parse(s)
80+ if err != nil || u.Host == "" || u.User != nil || (u.Scheme != "http" && u.Scheme != "https") {
81+ return nil, errors.New("invalid registry upstream URL")
82+ }
83+ }
84+ if c.RequestTimeout <= 0 {
85+ c.RequestTimeout = 10 * time.Second
86+ }
87+ if c.BlobIdleTimeout <= 0 {
88+ c.BlobIdleTimeout = 30 * time.Second
89+ }
90+ if c.KeepAliveInterval <= 0 {
91+ c.KeepAliveInterval = time.Second
92+ }
93+ if c.AbandonGrace <= 0 {
94+ c.AbandonGrace = 5 * time.Second
95+ }
96+ if c.MaxDownloads <= 0 {
97+ c.MaxDownloads = 16
98+ }
99+ if c.SlotWait <= 0 {
100+ c.SlotWait = 30 * time.Second
101+ }
102+ st, err := openStore(storeConfig{dir: c.Dir, max: c.Max, now: time.Now, grace: c.AbandonGrace,
103+ maxDownloads: c.MaxDownloads, slotWait: c.SlotWait, log: c.Log})
104+ if err != nil {
105+ return nil, err
106+ }
107+ client, blobs := upstreamClients(c.RequestTimeout)
108+ return &Proxy{s: st, u: &upstream{base: strings.TrimRight(c.Upstream, "/"), auth: c.AuthURL, client: client, blobClient: blobs,
109+ requestTimeout: c.RequestTimeout, blobIdle: c.BlobIdleTimeout, tokens: map[string]bearer{}}, log: c.Log, keepEvery: c.KeepAliveInterval}, nil
110+}
111+
112+var namePattern = regexp.MustCompile(`^[a-z0-9]+(?:[._-][a-z0-9]+)*(?:/[a-z0-9]+(?:[._-][a-z0-9]+)*)*$`)
113+var tagPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}$`)
114+
115+func validDigest(s string) bool {
116+ if len(s) != 71 || !strings.HasPrefix(s, "sha256:") {
117+ return false
118+ }
119+ _, err := hex.DecodeString(s[7:])
120+ return err == nil && s == strings.ToLower(s)
121+}
122+
123+func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
124+ if r.Method != http.MethodGet && r.Method != http.MethodHead {
125+ w.WriteHeader(http.StatusMethodNotAllowed)
126+ return
127+ }
128+ w.Header().Set("Docker-Distribution-Api-Version", "registry/2.0")
129+ if r.URL.Path == "/v2/" {
130+ w.Header().Set("Content-Type", "application/json")
131+ w.Header().Set("Content-Length", "2")
132+ if r.Method == http.MethodGet {
133+ _, _ = io.WriteString(w, "{}")
134+ }
135+ return
136+ }
137+ if !strings.HasPrefix(r.URL.Path, "/v2/") {
138+ http.NotFound(w, r)
139+ return
140+ }
141+ path := strings.TrimPrefix(r.URL.Path, "/v2/")
142+ i := strings.LastIndex(path, "/")
143+ if i < 0 {
144+ http.NotFound(w, r)
145+ return
146+ }
147+ ref := path[i+1:]
148+ path = path[:i]
149+ i = strings.LastIndex(path, "/")
150+ if i < 0 {
151+ http.NotFound(w, r)
152+ return
153+ }
154+ name, route := path[:i], path[i+1:]
155+ // dockerd's legacy client sends one Accept line per media type.
156+ accept := strings.Join(r.Header.Values("Accept"), ",")
157+ if len(name) > 255 || !namePattern.MatchString(name) || len(accept) > 8192 {
158+ http.NotFound(w, r)
159+ return
160+ }
161+ if route == kindManifest.dir() && (validDigest(ref) || tagPattern.MatchString(ref)) {
162+ p.manifest(w, r, name, ref, normalizeAccept(accept))
163+ return
164+ }
165+ if route == kindBlob.dir() && validDigest(ref) {
166+ p.blob(w, r, name, ref)
167+ return
168+ }
169+ http.NotFound(w, r)
170+}
171+
172+// logFor is the Proxy's logger, with the job_id of the request (WithJob).
173+func (p *Proxy) logFor(ctx context.Context) zerowrap.Logger {
174+ if job := jobFrom(ctx); job != "" {
175+ return zerowrap.Logger{Logger: p.log.With().Str("job_id", job).Logger()}
176+ }
177+ return p.log
178+}
179+
180+func serveFile(w http.ResponseWriter, r *http.Request, f *storedFile, media, digest string) {
181+ defer func() { _ = f.Close() }()
182+ fi, err := f.Stat()
183+ if err != nil {
184+ http.Error(w, "registry cache read failed", http.StatusBadGateway)
185+ return
186+ }
187+ w.Header().Set("Content-Type", media)
188+ w.Header().Set("Docker-Content-Digest", digest)
189+ http.ServeContent(w, r, "", fi.ModTime(), f)
190+}
191+
192+// serveStaleTag serves the manifest a tag pointed to when upstream could not
193+// answer, if that manifest is still stored. It reports whether it served one.
194+func (p *Proxy) serveStaleTag(w http.ResponseWriter, r *http.Request, name, tag, accept string) bool {
195+ digest := p.s.tag(name, tag, accept)
196+ if !validDigest(digest) {
197+ return false
198+ }
199+ f, media, err := p.s.open(kindManifest, digest)
200+ if err != nil {
201+ return false
202+ }
203+ serveFile(w, r, f, media, digest)
204+ return true
205+}
206+
207+// manifest serves a manifest by tag or digest. accept is normalized.
208+func (p *Proxy) manifest(w http.ResponseWriter, r *http.Request, name, ref, accept string) {
209+ start := time.Now()
210+ hit := false
211+ log := p.logFor(r.Context())
212+ defer func() {
213+ log.Info().Str("repo", name).Str("ref", ref).Bool("hit", hit).Int64("ms", time.Since(start).Milliseconds()).Msg("registry manifest")
214+ }()
215+ ctx, cancel := context.WithTimeout(r.Context(), p.u.requestTimeout)
216+ defer cancel()
217+ digest := ref
218+ tag := !validDigest(ref)
219+ if tag {
220+ resp, err := p.u.request(ctx, http.MethodHead, name, kindManifest, ref, accept)
221+ if err != nil {
222+ if unavailable(err) && p.serveStaleTag(w, r, name, ref, accept) {
223+ hit = true
224+ return
225+ }
226+ http.Error(w, "registry upstream unavailable", proxyStatus(err))
227+ return
228+ }
229+ digest = resp.Header.Get("Docker-Content-Digest")
230+ _ = resp.Body.Close()
231+ if !validDigest(digest) {
232+ http.Error(w, "invalid upstream digest", http.StatusBadGateway)
233+ return
234+ }
235+ }
236+ // Shared flight, as for blobs: it must not end with the initiator's request.
237+ f, media, cached, err := p.s.ensure(ctx, kindManifest, digest, func(flightCtx context.Context, dst io.Writer) (string, error) {
238+ fillCtx, fillCancel := context.WithTimeout(flightCtx, p.u.requestTimeout)
239+ defer fillCancel()
240+ resp, err := p.u.request(fillCtx, http.MethodGet, name, kindManifest, digest, accept)
241+ if err != nil {
242+ return "", err
243+ }
244+ defer func() { _ = resp.Body.Close() }()
245+ b, err := io.ReadAll(io.LimitReader(resp.Body, manifestLimit+1))
246+ if err != nil {
247+ return "", err
248+ }
249+ if len(b) > manifestLimit {
250+ return "", errTooLarge
251+ }
252+ _, err = dst.Write(b)
253+ return resp.Header.Get("Content-Type"), err
254+ }, keepalive{})
255+ hit = cached
256+ if err != nil {
257+ if tag && unavailable(err) && p.serveStaleTag(w, r, name, ref, accept) {
258+ hit = true
259+ return
260+ }
261+ http.Error(w, "registry fetch failed", proxyStatus(err))
262+ return
263+ }
264+ if tag {
265+ _ = p.s.setTag(name, ref, accept, digest)
266+ }
267+ serveFile(w, r, f, media, digest)
268+}
269+
270+func (p *Proxy) blob(w http.ResponseWriter, r *http.Request, name, digest string) {
271+ if f, media, err := p.s.open(kindBlob, digest); err == nil {
272+ serveFile(w, r, f, media, digest)
273+ return
274+ }
275+ // A cold HEAD is metadata-only and retains the short total timeout.
276+ if r.Method == http.MethodHead {
277+ ctx, cancel := context.WithTimeout(r.Context(), p.u.requestTimeout)
278+ defer cancel()
279+ resp, err := p.u.request(ctx, http.MethodHead, name, kindBlob, digest, "")
280+ if err != nil {
281+ http.Error(w, "registry fetch failed", proxyStatus(err))
282+ return
283+ }
284+ defer func() { _ = resp.Body.Close() }()
285+ if resp.ContentLength < 0 {
286+ http.Error(w, "unknown blob length", http.StatusBadGateway)
287+ return
288+ }
289+ w.Header().Set("Content-Type", "application/octet-stream")
290+ w.Header().Set("Content-Length", strconv.FormatInt(resp.ContentLength, 10))
291+ w.Header().Set("Docker-Content-Digest", digest)
292+ return
293+ }
294+ // The flight is shared by every client that asks for this blob, so no client
295+ // may own it: it downloads on a context of its own, into the temporary file
296+ // only, and is bounded by Max and the upstream idle timeout. Every client,
297+ // the initiator included, waits for the verified commit and then reads the
298+ // stored file (simplest correct option: no client streams unverified bytes,
299+ // and a client that disconnects fails only itself). The flight is cancelled
300+ // when no client has waited for it for AbandonGrace. At most MaxDownloads
301+ // run at once, and at most its Quota per client: a request over either
302+ // waits up to SlotWait for a slot, then gets 503, which dockerd retries
303+ // (about 50 s) before pulling from Docker Hub. While it waits, a
304+ // KeepAliver writer (the cache channel's progress deadline) is kept alive
305+ // as long as the download moves.
306+ keep := keepalive{every: p.keepEvery}
307+ if k, ok := w.(KeepAliver); ok {
308+ keep.fn = k.KeepAlive
309+ }
310+ log := p.logFor(r.Context())
311+ f, media, _, err := p.s.ensure(r.Context(), kindBlob, digest, func(downloadCtx context.Context, dst io.Writer) (string, error) {
312+ start := time.Now()
313+ var size int64
314+ defer func() {
315+ log.Debug().Str("digest", digest).Int64("bytes", size).Int64("ms", time.Since(start).Milliseconds()).Msg("registry blob")
316+ }()
317+ resp, err := p.u.blob(downloadCtx, name, digest)
318+ if err != nil {
319+ return "", err
320+ }
321+ defer func() { _ = resp.Body.Close() }()
322+ if resp.ContentLength > p.s.max {
323+ return "", errTooLarge
324+ }
325+ size, err = io.Copy(dst, resp.Body)
326+ return "application/octet-stream", err
327+ }, keep)
328+ if errors.Is(err, errBusy) {
329+ w.Header().Set("Retry-After", "1")
330+ http.Error(w, "registry downloads busy", http.StatusServiceUnavailable)
331+ return
332+ }
333+ if err != nil {
334+ http.Error(w, "registry fetch failed", proxyStatus(err))
335+ return
336+ }
337+ serveFile(w, r, f, media, digest)
338+}
A internal/registryproxy/proxy_test.go +117 -0
@@ -0,0 +1,117 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "fmt"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "os"
11+ "path/filepath"
12+ "strings"
13+ "testing"
14+)
15+
16+// A pulled manifest and blob are served from the store the second time: a tag
17+// is revalidated by HEAD, a digest is not fetched again.
18+func TestPullCached(t *testing.T) {
19+ f := fake(t)
20+ p := proxyFor(t, f)
21+ for range 2 {
22+ w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/3")
23+ if w.Code != http.StatusOK || w.Body.String() != manifestBody {
24+ t.Fatalf("manifest: %d %s", w.Code, w.Body)
25+ }
26+ w = request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(blobBody))
27+ if w.Code != http.StatusOK || w.Body.String() != blobBody {
28+ t.Fatalf("blob: %d %s", w.Code, w.Body)
29+ }
30+ }
31+ if f.heads.Load() != 2 || f.gets.Load() != 1 || f.blobs.Load() != 1 || f.tokens.Load() != 1 {
32+ t.Fatalf("counts HEAD=%d GET=%d blobs=%d tokens=%d", f.heads.Load(), f.gets.Load(), f.blobs.Load(), f.tokens.Load())
33+ }
34+}
35+
36+// HEAD of a blob answers its length, cold or stored, with no body; a stored
37+// blob serves ranges.
38+func TestBlobHeadAndRange(t *testing.T) {
39+ f := fake(t)
40+ p := proxyFor(t, f)
41+ path := "/v2/library/alpine/blobs/" + sum(blobBody)
42+ for _, cached := range []bool{false, true} {
43+ if cached {
44+ if w := request(t, p, http.MethodGet, path); w.Code != http.StatusOK {
45+ t.Fatal(w.Code)
46+ }
47+ }
48+ w := request(t, p, http.MethodHead, path)
49+ if w.Code != http.StatusOK || w.Body.Len() != 0 || w.Header().Get("Content-Length") != fmt.Sprint(len(blobBody)) {
50+ t.Fatalf("HEAD %d %v %q", w.Code, w.Header(), w.Body.String())
51+ }
52+ }
53+ w := request(t, p, http.MethodGet, path, "Range", "bytes=2-5")
54+ if w.Code != http.StatusPartialContent || w.Body.String() != blobBody[2:6] {
55+ t.Fatalf("range: %d %s", w.Code, w.Body)
56+ }
57+}
58+
59+// A blob is verified before any byte of it is sent: a corrupt one is a 502
60+// with no body, and is not stored.
61+func TestCorruptBlobRejected(t *testing.T) {
62+ f := fake(t)
63+ f.corrupt.Store(true)
64+ p := proxyFor(t, f)
65+ w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(blobBody))
66+ if w.Code != http.StatusBadGateway || strings.Contains(w.Body.String(), "xxxx") {
67+ t.Fatalf("corrupt blob: %d %q", w.Code, w.Body)
68+ }
69+ if _, err := os.Stat(filepath.Join(p.s.dir, kindBlob.key(sum(blobBody)[7:]))); !os.IsNotExist(err) {
70+ t.Fatal("corrupt blob stored")
71+ }
72+ if ds, _ := os.ReadDir(filepath.Join(p.s.dir, "tmp")); len(ds) != 0 {
73+ t.Fatal("partial temp retained")
74+ }
75+}
76+
77+// Only GET and HEAD of valid names, tags and digests under /v2/ are served.
78+func TestRouteValidation(t *testing.T) {
79+ p := proxyFor(t, fake(t))
80+ for _, path := range []string{"/v2/../library/alpine/manifests/3", "/v2/Library/alpine/manifests/3", "/v2/library/alpine/manifests/.bad", "/v2/library/alpine/manifests/sha256:bad", "/v2/library/alpine/blobs/sha256:bad", "/v2/library/alpine/tags/list", "/v3/library/alpine/manifests/3"} {
81+ if w := request(t, p, http.MethodGet, path); w.Code != http.StatusNotFound {
82+ t.Errorf("%s: %d", path, w.Code)
83+ }
84+ }
85+ if w := request(t, p, http.MethodPut, "/v2/"); w.Code != http.StatusMethodNotAllowed {
86+ t.Fatal(w.Code)
87+ }
88+ if w := request(t, p, http.MethodGet, "/v2/"); w.Code != http.StatusOK || w.Body.String() != "{}" {
89+ t.Fatalf("/v2/: %d %s", w.Code, w.Body)
90+ }
91+}
92+
93+// A manifest whose bytes do not match its digest, or over the size limit, is
94+// a 502 and is not stored.
95+func TestManifestLimitsAndDigest(t *testing.T) {
96+ for _, body := range []string{"tampered", strings.Repeat("x", manifestLimit+1)} {
97+ up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
98+ if r.URL.Path == "/token" {
99+ _, _ = io.WriteString(w, `{"token":"public"}`)
100+ return
101+ }
102+ _, _ = io.WriteString(w, body)
103+ }))
104+ p, err := New(Config{Dir: t.TempDir(), Max: 8 << 20, Upstream: up.URL, AuthURL: up.URL + "/token"})
105+ if err != nil {
106+ t.Fatal(err)
107+ }
108+ w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/"+sum(manifestBody))
109+ up.Close()
110+ if w.Code != http.StatusBadGateway {
111+ t.Fatal(w.Code)
112+ }
113+ if _, _, err := p.s.open(kindManifest, sum(manifestBody)); err == nil {
114+ t.Fatal("invalid manifest stored")
115+ }
116+ }
117+}
@@ -0,0 +1,117 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "fmt"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "os"
11+ "path/filepath"
12+ "strings"
13+ "testing"
14+)
15+
16+// A pulled manifest and blob are served from the store the second time: a tag
17+// is revalidated by HEAD, a digest is not fetched again.
18+func TestPullCached(t *testing.T) {
19+ f := fake(t)
20+ p := proxyFor(t, f)
21+ for range 2 {
22+ w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/3")
23+ if w.Code != http.StatusOK || w.Body.String() != manifestBody {
24+ t.Fatalf("manifest: %d %s", w.Code, w.Body)
25+ }
26+ w = request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(blobBody))
27+ if w.Code != http.StatusOK || w.Body.String() != blobBody {
28+ t.Fatalf("blob: %d %s", w.Code, w.Body)
29+ }
30+ }
31+ if f.heads.Load() != 2 || f.gets.Load() != 1 || f.blobs.Load() != 1 || f.tokens.Load() != 1 {
32+ t.Fatalf("counts HEAD=%d GET=%d blobs=%d tokens=%d", f.heads.Load(), f.gets.Load(), f.blobs.Load(), f.tokens.Load())
33+ }
34+}
35+
36+// HEAD of a blob answers its length, cold or stored, with no body; a stored
37+// blob serves ranges.
38+func TestBlobHeadAndRange(t *testing.T) {
39+ f := fake(t)
40+ p := proxyFor(t, f)
41+ path := "/v2/library/alpine/blobs/" + sum(blobBody)
42+ for _, cached := range []bool{false, true} {
43+ if cached {
44+ if w := request(t, p, http.MethodGet, path); w.Code != http.StatusOK {
45+ t.Fatal(w.Code)
46+ }
47+ }
48+ w := request(t, p, http.MethodHead, path)
49+ if w.Code != http.StatusOK || w.Body.Len() != 0 || w.Header().Get("Content-Length") != fmt.Sprint(len(blobBody)) {
50+ t.Fatalf("HEAD %d %v %q", w.Code, w.Header(), w.Body.String())
51+ }
52+ }
53+ w := request(t, p, http.MethodGet, path, "Range", "bytes=2-5")
54+ if w.Code != http.StatusPartialContent || w.Body.String() != blobBody[2:6] {
55+ t.Fatalf("range: %d %s", w.Code, w.Body)
56+ }
57+}
58+
59+// A blob is verified before any byte of it is sent: a corrupt one is a 502
60+// with no body, and is not stored.
61+func TestCorruptBlobRejected(t *testing.T) {
62+ f := fake(t)
63+ f.corrupt.Store(true)
64+ p := proxyFor(t, f)
65+ w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(blobBody))
66+ if w.Code != http.StatusBadGateway || strings.Contains(w.Body.String(), "xxxx") {
67+ t.Fatalf("corrupt blob: %d %q", w.Code, w.Body)
68+ }
69+ if _, err := os.Stat(filepath.Join(p.s.dir, kindBlob.key(sum(blobBody)[7:]))); !os.IsNotExist(err) {
70+ t.Fatal("corrupt blob stored")
71+ }
72+ if ds, _ := os.ReadDir(filepath.Join(p.s.dir, "tmp")); len(ds) != 0 {
73+ t.Fatal("partial temp retained")
74+ }
75+}
76+
77+// Only GET and HEAD of valid names, tags and digests under /v2/ are served.
78+func TestRouteValidation(t *testing.T) {
79+ p := proxyFor(t, fake(t))
80+ for _, path := range []string{"/v2/../library/alpine/manifests/3", "/v2/Library/alpine/manifests/3", "/v2/library/alpine/manifests/.bad", "/v2/library/alpine/manifests/sha256:bad", "/v2/library/alpine/blobs/sha256:bad", "/v2/library/alpine/tags/list", "/v3/library/alpine/manifests/3"} {
81+ if w := request(t, p, http.MethodGet, path); w.Code != http.StatusNotFound {
82+ t.Errorf("%s: %d", path, w.Code)
83+ }
84+ }
85+ if w := request(t, p, http.MethodPut, "/v2/"); w.Code != http.StatusMethodNotAllowed {
86+ t.Fatal(w.Code)
87+ }
88+ if w := request(t, p, http.MethodGet, "/v2/"); w.Code != http.StatusOK || w.Body.String() != "{}" {
89+ t.Fatalf("/v2/: %d %s", w.Code, w.Body)
90+ }
91+}
92+
93+// A manifest whose bytes do not match its digest, or over the size limit, is
94+// a 502 and is not stored.
95+func TestManifestLimitsAndDigest(t *testing.T) {
96+ for _, body := range []string{"tampered", strings.Repeat("x", manifestLimit+1)} {
97+ up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
98+ if r.URL.Path == "/token" {
99+ _, _ = io.WriteString(w, `{"token":"public"}`)
100+ return
101+ }
102+ _, _ = io.WriteString(w, body)
103+ }))
104+ p, err := New(Config{Dir: t.TempDir(), Max: 8 << 20, Upstream: up.URL, AuthURL: up.URL + "/token"})
105+ if err != nil {
106+ t.Fatal(err)
107+ }
108+ w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/"+sum(manifestBody))
109+ up.Close()
110+ if w.Code != http.StatusBadGateway {
111+ t.Fatal(w.Code)
112+ }
113+ if _, _, err := p.s.open(kindManifest, sum(manifestBody)); err == nil {
114+ t.Fatal("invalid manifest stored")
115+ }
116+ }
117+}
A internal/registryproxy/quota.go +61 -0
@@ -0,0 +1,61 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "sync/atomic"
8+)
9+
10+// Quota bounds the cold blob downloads started on behalf of one client, such as
11+// the channel of one sandbox, so that no client can take every download slot of
12+// the Proxy. A download counts against the quota of the request that started it
13+// until it really ends, even when only other clients still wait for it;
14+// requests that join a download in progress take none.
15+type Quota struct {
16+ max int64
17+ used atomic.Int64
18+}
19+
20+// NewQuota returns a quota of n concurrent downloads; n must be positive.
21+func NewQuota(n int) *Quota { return &Quota{max: int64(max(n, 1))} }
22+
23+// WithQuota attaches q to the context of a request to the Proxy. A request
24+// without a quota is bound by the Proxy's global cap only.
25+func WithQuota(ctx context.Context, q *Quota) context.Context {
26+ return context.WithValue(ctx, quotaKey{}, q)
27+}
28+
29+type quotaKey struct{}
30+
31+func quotaFrom(ctx context.Context) *Quota {
32+ q, _ := ctx.Value(quotaKey{}).(*Quota)
33+ return q
34+}
35+
36+// WithJob attaches the id of the job a request serves to its context: the
37+// Proxy logs it as job_id on its registry lines.
38+func WithJob(ctx context.Context, job string) context.Context {
39+ return context.WithValue(ctx, jobKey{}, job)
40+}
41+
42+type jobKey struct{}
43+
44+func jobFrom(ctx context.Context) string {
45+ job, _ := ctx.Value(jobKey{}).(string)
46+ return job
47+}
48+
49+func (q *Quota) tryAcquire() bool {
50+ for {
51+ n := q.used.Load()
52+ if n >= q.max {
53+ return false
54+ }
55+ if q.used.CompareAndSwap(n, n+1) {
56+ return true
57+ }
58+ }
59+}
60+
61+func (q *Quota) release() { q.used.Add(-1) }
@@ -0,0 +1,61 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "sync/atomic"
8+)
9+
10+// Quota bounds the cold blob downloads started on behalf of one client, such as
11+// the channel of one sandbox, so that no client can take every download slot of
12+// the Proxy. A download counts against the quota of the request that started it
13+// until it really ends, even when only other clients still wait for it;
14+// requests that join a download in progress take none.
15+type Quota struct {
16+ max int64
17+ used atomic.Int64
18+}
19+
20+// NewQuota returns a quota of n concurrent downloads; n must be positive.
21+func NewQuota(n int) *Quota { return &Quota{max: int64(max(n, 1))} }
22+
23+// WithQuota attaches q to the context of a request to the Proxy. A request
24+// without a quota is bound by the Proxy's global cap only.
25+func WithQuota(ctx context.Context, q *Quota) context.Context {
26+ return context.WithValue(ctx, quotaKey{}, q)
27+}
28+
29+type quotaKey struct{}
30+
31+func quotaFrom(ctx context.Context) *Quota {
32+ q, _ := ctx.Value(quotaKey{}).(*Quota)
33+ return q
34+}
35+
36+// WithJob attaches the id of the job a request serves to its context: the
37+// Proxy logs it as job_id on its registry lines.
38+func WithJob(ctx context.Context, job string) context.Context {
39+ return context.WithValue(ctx, jobKey{}, job)
40+}
41+
42+type jobKey struct{}
43+
44+func jobFrom(ctx context.Context) string {
45+ job, _ := ctx.Value(jobKey{}).(string)
46+ return job
47+}
48+
49+func (q *Quota) tryAcquire() bool {
50+ for {
51+ n := q.used.Load()
52+ if n >= q.max {
53+ return false
54+ }
55+ if q.used.CompareAndSwap(n, n+1) {
56+ return true
57+ }
58+ }
59+}
60+
61+func (q *Quota) release() { q.used.Add(-1) }
A internal/registryproxy/slots_test.go +215 -0
@@ -0,0 +1,215 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "strings"
11+ "sync"
12+ "testing"
13+ "time"
14+)
15+
16+// At most MaxDownloads cold blobs download at once; one more waits for
17+// SlotWait and then gets 503 (dockerd retries or pulls from Docker Hub), a
18+// joiner of a running download takes no slot, and a slot frees once a
19+// download ends.
20+func TestColdDownloadCap(t *testing.T) {
21+ u := newStallingUpstream(t)
22+ const limit = 3
23+ const wait = 50 * time.Millisecond
24+ p := proxyWith(t, u.srv, Config{AbandonGrace: 10 * time.Millisecond, MaxDownloads: limit, SlotWait: wait})
25+ ctx, cancel := context.WithCancel(context.Background())
26+ var wg sync.WaitGroup
27+ for i := range limit {
28+ wg.Go(func() {
29+ p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, blobPath(i), nil).WithContext(ctx))
30+ })
31+ }
32+ waitFor(t, p.s, "the cap to fill", func() bool { return p.s.downloads == limit })
33+ start := time.Now()
34+ w := request(t, p, http.MethodGet, blobPath(limit))
35+ if w.Code != http.StatusServiceUnavailable || w.Header().Get("Retry-After") == "" {
36+ t.Fatalf("over the cap: %d", w.Code)
37+ }
38+ if d := time.Since(start); d < wait {
39+ t.Fatalf("503 after %v, before the slot wait", d)
40+ }
41+ // Joining a running download takes no slot: the joiner is a waiter of the
42+ // flight, and the number of downloads does not move.
43+ key := kindBlob.key(sum("blob-0")[7:])
44+ go p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, blobPath(0), nil).WithContext(ctx))
45+ waitFor(t, p.s, "the joiner", func() bool { return waiters(p.s, key) == 2 })
46+ p.s.mu.Lock()
47+ downloads := p.s.downloads
48+ p.s.mu.Unlock()
49+ if downloads != limit {
50+ t.Fatalf("%d downloads with a joiner, want %d", downloads, limit)
51+ }
52+ cancel()
53+ wg.Wait()
54+ waitFor(t, p.s, "the slots to free", func() bool { return p.s.downloads == 0 })
55+ if u.started.Load() != limit {
56+ t.Fatalf("%d downloads started", u.started.Load())
57+ }
58+}
59+
60+// A sandbox that saturates its quota does not hold back another sandbox: its
61+// download starts at once while the global cap still has room.
62+func TestQuotaIsolatesChannels(t *testing.T) {
63+ u := newStallingUpstream(t)
64+ p := proxyWith(t, u.srv, Config{SlotWait: time.Hour, AbandonGrace: 10 * time.Millisecond})
65+ a, b := NewQuota(3), NewQuota(3)
66+ ctxA, cancelA := context.WithCancel(WithQuota(context.Background(), a))
67+ defer cancelA()
68+ for i := range 3 {
69+ serveAsync(p, ctxA, blobPath(i), nil)
70+ }
71+ waitFor(t, p.s, "A's quota to fill", func() bool { return a.used.Load() == 3 })
72+ // A's fourth waits for its own quota and starts nothing.
73+ serveAsync(p, ctxA, blobPath(3), nil)
74+ ctxB, cancelB := context.WithCancel(WithQuota(context.Background(), b))
75+ defer cancelB()
76+ start := time.Now()
77+ serveAsync(p, ctxB, blobPath(10), nil)
78+ waitFor(t, p.s, "B's download to start", func() bool { return b.used.Load() == 1 })
79+ if d := time.Since(start); d > time.Second {
80+ t.Fatalf("B started after %v", d)
81+ }
82+ // The quotas are taken before the downloads reach upstream: wait for the
83+ // 4 of A and B there, then check under the store lock that A's fourth
84+ // took no slot.
85+ waitFor(t, p.s, "4 downloads upstream", func() bool { return u.started.Load() == 4 })
86+ p.s.mu.Lock()
87+ downloads := p.s.downloads
88+ p.s.mu.Unlock()
89+ if downloads != 4 {
90+ t.Fatalf("%d downloads, want A's 3 and B's 1", downloads)
91+ }
92+ cancelA()
93+ cancelB()
94+ waitFor(t, p.s, "every slot to free", func() bool { return p.s.downloads == 0 })
95+ if a.used.Load() != 0 || b.used.Load() != 0 {
96+ t.Fatalf("quota not released: %d %d", a.used.Load(), b.used.Load())
97+ }
98+}
99+
100+// A request over its quota waits, kept alive, and starts its download as soon
101+// as one of its channel's ends, with no 503.
102+func TestQuotaWaiterGetsFreedSlot(t *testing.T) {
103+ body := strings.Repeat("layer", 100)
104+ release := make(chan struct{})
105+ var once sync.Once
106+ unblock := func() { once.Do(func() { close(release) }) }
107+ t.Cleanup(unblock)
108+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
109+ if strings.HasSuffix(r.URL.Path, sum(body)) {
110+ _, _ = io.WriteString(w, body)
111+ return
112+ }
113+ w.Header().Set("Content-Length", "1000")
114+ _, _ = io.WriteString(w, "x")
115+ w.(http.Flusher).Flush()
116+ select {
117+ case <-release:
118+ case <-r.Context().Done():
119+ }
120+ })
121+ p := proxyWith(t, srv, Config{SlotWait: time.Hour, KeepAliveInterval: 5 * time.Millisecond})
122+ q := NewQuota(1)
123+ ctx := WithQuota(context.Background(), q)
124+ first := serveAsync(p, ctx, blobPath(0), nil)
125+ waitFor(t, p.s, "the quota to fill", func() bool { return q.used.Load() == 1 })
126+ w := &keepAliveRecorder{ResponseRecorder: httptest.NewRecorder()}
127+ second := serveAsync(p, ctx, "/v2/library/alpine/blobs/"+sum(body), w)
128+ waitKeptAlive(t, w) // waiting for a slot
129+ select {
130+ case code := <-second:
131+ t.Fatalf("did not wait for a slot: %d", code)
132+ default:
133+ }
134+ unblock() // the first download ends (digest mismatch), freeing the slot
135+ if code := <-first; code != http.StatusBadGateway {
136+ t.Fatalf("first: %d", code)
137+ }
138+ if code := <-second; code != http.StatusOK || w.Body.String() != body {
139+ t.Fatalf("second: %d", code)
140+ }
141+}
142+
143+// After SlotWait without a slot, the request gets 503.
144+func TestQuotaWaitTimesOut(t *testing.T) {
145+ u := newStallingUpstream(t)
146+ const wait = 50 * time.Millisecond
147+ p := proxyWith(t, u.srv, Config{SlotWait: wait, AbandonGrace: 10 * time.Millisecond})
148+ q := NewQuota(1)
149+ ctx, cancel := context.WithCancel(WithQuota(context.Background(), q))
150+ defer cancel()
151+ serveAsync(p, ctx, blobPath(0), nil)
152+ waitFor(t, p.s, "the quota to fill", func() bool { return q.used.Load() == 1 })
153+ start := time.Now()
154+ code := <-serveAsync(p, ctx, blobPath(1), nil)
155+ if code != http.StatusServiceUnavailable {
156+ t.Fatalf("over the quota: %d", code)
157+ }
158+ if d := time.Since(start); d < wait {
159+ t.Fatalf("503 after %v", d)
160+ }
161+ if u.started.Load() != 1 {
162+ t.Fatalf("%d downloads started", u.started.Load())
163+ }
164+}
165+
166+// A request that disconnects while it waits for a slot leaves at once, with no
167+// slot and no 503.
168+func TestQuotaWaiterDisconnects(t *testing.T) {
169+ u := newStallingUpstream(t)
170+ const wait = time.Hour
171+ p := proxyWith(t, u.srv, Config{SlotWait: wait, AbandonGrace: 10 * time.Millisecond, KeepAliveInterval: 5 * time.Millisecond})
172+ q := NewQuota(1)
173+ ctx, cancel := context.WithCancel(WithQuota(context.Background(), q))
174+ defer cancel()
175+ serveAsync(p, ctx, blobPath(0), nil)
176+ waitFor(t, p.s, "the quota to fill", func() bool { return q.used.Load() == 1 })
177+ wctx, wcancel := context.WithCancel(WithQuota(context.Background(), q))
178+ w := &keepAliveRecorder{ResponseRecorder: httptest.NewRecorder()}
179+ waiting := serveAsync(p, wctx, blobPath(1), w)
180+ waitKeptAlive(t, w) // really waiting for a slot
181+ start := time.Now()
182+ wcancel()
183+ code := <-waiting
184+ if d := time.Since(start); d > time.Second {
185+ t.Fatalf("left after %v", d)
186+ }
187+ if code == http.StatusServiceUnavailable {
188+ t.Fatal("a disconnected waiter got 503")
189+ }
190+ if u.started.Load() != 1 || q.used.Load() != 1 {
191+ t.Fatalf("%d downloads started, quota %d", u.started.Load(), q.used.Load())
192+ }
193+}
194+
195+// Joining a download in progress takes none of the joiner's quota.
196+func TestQuotaJoinersExempt(t *testing.T) {
197+ u := newStallingUpstream(t)
198+ p := proxyWith(t, u.srv, Config{SlotWait: 20 * time.Millisecond, AbandonGrace: 10 * time.Millisecond})
199+ a, b := NewQuota(1), NewQuota(1)
200+ ctxA, cancelA := context.WithCancel(WithQuota(context.Background(), a))
201+ defer cancelA()
202+ serveAsync(p, ctxA, blobPath(0), nil)
203+ waitFor(t, p.s, "A's download", func() bool { return a.used.Load() == 1 })
204+ ctxB, cancelB := context.WithCancel(WithQuota(context.Background(), b))
205+ defer cancelB()
206+ serveAsync(p, ctxB, blobPath(0), nil) // joins A's
207+ key := kindBlob.key(sum("blob-0")[7:])
208+ waitFor(t, p.s, "B to join", func() bool { return waiters(p.s, key) == 2 })
209+ if b.used.Load() != 0 {
210+ t.Fatal("a joiner took quota")
211+ }
212+ // B's quota is still free for a download of its own.
213+ serveAsync(p, ctxB, blobPath(1), nil)
214+ waitFor(t, p.s, "B's own download", func() bool { return b.used.Load() == 1 })
215+}
@@ -0,0 +1,215 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "strings"
11+ "sync"
12+ "testing"
13+ "time"
14+)
15+
16+// At most MaxDownloads cold blobs download at once; one more waits for
17+// SlotWait and then gets 503 (dockerd retries or pulls from Docker Hub), a
18+// joiner of a running download takes no slot, and a slot frees once a
19+// download ends.
20+func TestColdDownloadCap(t *testing.T) {
21+ u := newStallingUpstream(t)
22+ const limit = 3
23+ const wait = 50 * time.Millisecond
24+ p := proxyWith(t, u.srv, Config{AbandonGrace: 10 * time.Millisecond, MaxDownloads: limit, SlotWait: wait})
25+ ctx, cancel := context.WithCancel(context.Background())
26+ var wg sync.WaitGroup
27+ for i := range limit {
28+ wg.Go(func() {
29+ p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, blobPath(i), nil).WithContext(ctx))
30+ })
31+ }
32+ waitFor(t, p.s, "the cap to fill", func() bool { return p.s.downloads == limit })
33+ start := time.Now()
34+ w := request(t, p, http.MethodGet, blobPath(limit))
35+ if w.Code != http.StatusServiceUnavailable || w.Header().Get("Retry-After") == "" {
36+ t.Fatalf("over the cap: %d", w.Code)
37+ }
38+ if d := time.Since(start); d < wait {
39+ t.Fatalf("503 after %v, before the slot wait", d)
40+ }
41+ // Joining a running download takes no slot: the joiner is a waiter of the
42+ // flight, and the number of downloads does not move.
43+ key := kindBlob.key(sum("blob-0")[7:])
44+ go p.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, blobPath(0), nil).WithContext(ctx))
45+ waitFor(t, p.s, "the joiner", func() bool { return waiters(p.s, key) == 2 })
46+ p.s.mu.Lock()
47+ downloads := p.s.downloads
48+ p.s.mu.Unlock()
49+ if downloads != limit {
50+ t.Fatalf("%d downloads with a joiner, want %d", downloads, limit)
51+ }
52+ cancel()
53+ wg.Wait()
54+ waitFor(t, p.s, "the slots to free", func() bool { return p.s.downloads == 0 })
55+ if u.started.Load() != limit {
56+ t.Fatalf("%d downloads started", u.started.Load())
57+ }
58+}
59+
60+// A sandbox that saturates its quota does not hold back another sandbox: its
61+// download starts at once while the global cap still has room.
62+func TestQuotaIsolatesChannels(t *testing.T) {
63+ u := newStallingUpstream(t)
64+ p := proxyWith(t, u.srv, Config{SlotWait: time.Hour, AbandonGrace: 10 * time.Millisecond})
65+ a, b := NewQuota(3), NewQuota(3)
66+ ctxA, cancelA := context.WithCancel(WithQuota(context.Background(), a))
67+ defer cancelA()
68+ for i := range 3 {
69+ serveAsync(p, ctxA, blobPath(i), nil)
70+ }
71+ waitFor(t, p.s, "A's quota to fill", func() bool { return a.used.Load() == 3 })
72+ // A's fourth waits for its own quota and starts nothing.
73+ serveAsync(p, ctxA, blobPath(3), nil)
74+ ctxB, cancelB := context.WithCancel(WithQuota(context.Background(), b))
75+ defer cancelB()
76+ start := time.Now()
77+ serveAsync(p, ctxB, blobPath(10), nil)
78+ waitFor(t, p.s, "B's download to start", func() bool { return b.used.Load() == 1 })
79+ if d := time.Since(start); d > time.Second {
80+ t.Fatalf("B started after %v", d)
81+ }
82+ // The quotas are taken before the downloads reach upstream: wait for the
83+ // 4 of A and B there, then check under the store lock that A's fourth
84+ // took no slot.
85+ waitFor(t, p.s, "4 downloads upstream", func() bool { return u.started.Load() == 4 })
86+ p.s.mu.Lock()
87+ downloads := p.s.downloads
88+ p.s.mu.Unlock()
89+ if downloads != 4 {
90+ t.Fatalf("%d downloads, want A's 3 and B's 1", downloads)
91+ }
92+ cancelA()
93+ cancelB()
94+ waitFor(t, p.s, "every slot to free", func() bool { return p.s.downloads == 0 })
95+ if a.used.Load() != 0 || b.used.Load() != 0 {
96+ t.Fatalf("quota not released: %d %d", a.used.Load(), b.used.Load())
97+ }
98+}
99+
100+// A request over its quota waits, kept alive, and starts its download as soon
101+// as one of its channel's ends, with no 503.
102+func TestQuotaWaiterGetsFreedSlot(t *testing.T) {
103+ body := strings.Repeat("layer", 100)
104+ release := make(chan struct{})
105+ var once sync.Once
106+ unblock := func() { once.Do(func() { close(release) }) }
107+ t.Cleanup(unblock)
108+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
109+ if strings.HasSuffix(r.URL.Path, sum(body)) {
110+ _, _ = io.WriteString(w, body)
111+ return
112+ }
113+ w.Header().Set("Content-Length", "1000")
114+ _, _ = io.WriteString(w, "x")
115+ w.(http.Flusher).Flush()
116+ select {
117+ case <-release:
118+ case <-r.Context().Done():
119+ }
120+ })
121+ p := proxyWith(t, srv, Config{SlotWait: time.Hour, KeepAliveInterval: 5 * time.Millisecond})
122+ q := NewQuota(1)
123+ ctx := WithQuota(context.Background(), q)
124+ first := serveAsync(p, ctx, blobPath(0), nil)
125+ waitFor(t, p.s, "the quota to fill", func() bool { return q.used.Load() == 1 })
126+ w := &keepAliveRecorder{ResponseRecorder: httptest.NewRecorder()}
127+ second := serveAsync(p, ctx, "/v2/library/alpine/blobs/"+sum(body), w)
128+ waitKeptAlive(t, w) // waiting for a slot
129+ select {
130+ case code := <-second:
131+ t.Fatalf("did not wait for a slot: %d", code)
132+ default:
133+ }
134+ unblock() // the first download ends (digest mismatch), freeing the slot
135+ if code := <-first; code != http.StatusBadGateway {
136+ t.Fatalf("first: %d", code)
137+ }
138+ if code := <-second; code != http.StatusOK || w.Body.String() != body {
139+ t.Fatalf("second: %d", code)
140+ }
141+}
142+
143+// After SlotWait without a slot, the request gets 503.
144+func TestQuotaWaitTimesOut(t *testing.T) {
145+ u := newStallingUpstream(t)
146+ const wait = 50 * time.Millisecond
147+ p := proxyWith(t, u.srv, Config{SlotWait: wait, AbandonGrace: 10 * time.Millisecond})
148+ q := NewQuota(1)
149+ ctx, cancel := context.WithCancel(WithQuota(context.Background(), q))
150+ defer cancel()
151+ serveAsync(p, ctx, blobPath(0), nil)
152+ waitFor(t, p.s, "the quota to fill", func() bool { return q.used.Load() == 1 })
153+ start := time.Now()
154+ code := <-serveAsync(p, ctx, blobPath(1), nil)
155+ if code != http.StatusServiceUnavailable {
156+ t.Fatalf("over the quota: %d", code)
157+ }
158+ if d := time.Since(start); d < wait {
159+ t.Fatalf("503 after %v", d)
160+ }
161+ if u.started.Load() != 1 {
162+ t.Fatalf("%d downloads started", u.started.Load())
163+ }
164+}
165+
166+// A request that disconnects while it waits for a slot leaves at once, with no
167+// slot and no 503.
168+func TestQuotaWaiterDisconnects(t *testing.T) {
169+ u := newStallingUpstream(t)
170+ const wait = time.Hour
171+ p := proxyWith(t, u.srv, Config{SlotWait: wait, AbandonGrace: 10 * time.Millisecond, KeepAliveInterval: 5 * time.Millisecond})
172+ q := NewQuota(1)
173+ ctx, cancel := context.WithCancel(WithQuota(context.Background(), q))
174+ defer cancel()
175+ serveAsync(p, ctx, blobPath(0), nil)
176+ waitFor(t, p.s, "the quota to fill", func() bool { return q.used.Load() == 1 })
177+ wctx, wcancel := context.WithCancel(WithQuota(context.Background(), q))
178+ w := &keepAliveRecorder{ResponseRecorder: httptest.NewRecorder()}
179+ waiting := serveAsync(p, wctx, blobPath(1), w)
180+ waitKeptAlive(t, w) // really waiting for a slot
181+ start := time.Now()
182+ wcancel()
183+ code := <-waiting
184+ if d := time.Since(start); d > time.Second {
185+ t.Fatalf("left after %v", d)
186+ }
187+ if code == http.StatusServiceUnavailable {
188+ t.Fatal("a disconnected waiter got 503")
189+ }
190+ if u.started.Load() != 1 || q.used.Load() != 1 {
191+ t.Fatalf("%d downloads started, quota %d", u.started.Load(), q.used.Load())
192+ }
193+}
194+
195+// Joining a download in progress takes none of the joiner's quota.
196+func TestQuotaJoinersExempt(t *testing.T) {
197+ u := newStallingUpstream(t)
198+ p := proxyWith(t, u.srv, Config{SlotWait: 20 * time.Millisecond, AbandonGrace: 10 * time.Millisecond})
199+ a, b := NewQuota(1), NewQuota(1)
200+ ctxA, cancelA := context.WithCancel(WithQuota(context.Background(), a))
201+ defer cancelA()
202+ serveAsync(p, ctxA, blobPath(0), nil)
203+ waitFor(t, p.s, "A's download", func() bool { return a.used.Load() == 1 })
204+ ctxB, cancelB := context.WithCancel(WithQuota(context.Background(), b))
205+ defer cancelB()
206+ serveAsync(p, ctxB, blobPath(0), nil) // joins A's
207+ key := kindBlob.key(sum("blob-0")[7:])
208+ waitFor(t, p.s, "B to join", func() bool { return waiters(p.s, key) == 2 })
209+ if b.used.Load() != 0 {
210+ t.Fatal("a joiner took quota")
211+ }
212+ // B's quota is still free for a download of its own.
213+ serveAsync(p, ctxB, blobPath(1), nil)
214+ waitFor(t, p.s, "B's own download", func() bool { return b.used.Load() == 1 })
215+}
A internal/registryproxy/store.go +784 -0
@@ -0,0 +1,784 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "container/list"
7+ "context"
8+ "crypto/sha256"
9+ "encoding/hex"
10+ "errors"
11+ "fmt"
12+ "io"
13+ "os"
14+ "path/filepath"
15+ "sort"
16+ "strings"
17+ "sync"
18+ "sync/atomic"
19+ "time"
20+
21+ "github.com/bnema/zerowrap"
22+)
23+
24+var errTooLarge = errors.New("registry cache capacity exceeded")
25+var errDigest = errors.New("registry digest mismatch")
26+
27+// errBusy: no download slot came free within the slot wait. The proxy answers
28+// 503, and dockerd retries the layer (about 50 s) before pulling from Docker Hub.
29+var errBusy = errors.New("registry downloads busy")
30+
31+// minCharge is the least an entry is charged, whatever its size: about one
32+// disk block and inode. It bounds the number of entries, and so of files, to
33+// Max/minCharge. A tag file is charged exactly that.
34+const minCharge int64 = 4096
35+
36+// charge is what an entry of n bytes counts against Max.
37+func charge(n int64) int64 { return max(n, minCharge) }
38+
39+// kind is the directory of an entry: content (blobs, manifests) or a tag.
40+type kind uint8
41+
42+const (
43+ kindBlob kind = iota
44+ kindManifest
45+ kindTag
46+)
47+
48+var kinds = []kind{kindBlob, kindManifest, kindTag}
49+
50+// dir is the directory of the kind in the store, and its route in /v2/.
51+func (k kind) dir() string {
52+ switch k {
53+ case kindBlob:
54+ return "blobs"
55+ case kindManifest:
56+ return "manifests"
57+ default:
58+ return "tags"
59+ }
60+}
61+
62+// key is the key of an entry of kind k: its path in the store.
63+func (k kind) key(name string) string { return k.dir() + "/" + name }
64+
65+// lowWater bounds the extra room an eviction frees (with 5% of Max), so that a
66+// full store evicts once per that many bytes, not on every write.
67+const lowWater int64 = 64 << 20
68+
69+type entry struct {
70+ key string
71+ kind kind
72+ size int64
73+ used time.Time
74+ pins int
75+ digest string // tag target, empty for content
76+ elem *list.Element // in store.lru; nil once removed
77+}
78+
79+// flight is one download, shared by every request for its digest. Its fields
80+// below written are guarded by store.mu.
81+type flight struct {
82+ key string
83+ done chan struct{}
84+ err error
85+ written atomic.Int64 // bytes received so far, for the waiters' keepalive
86+
87+ waiters int
88+ gen int // bumped each time waiters drops to zero
89+ finished bool
90+ pinned bool // the committed entry holds a pin for the waiters
91+ kind kind // a blob flight takes a download slot
92+ quota *Quota // the starter's quota, held until the flight ends
93+ cancel context.CancelFunc
94+}
95+
96+// storeConfig is a store's complete configuration: openStore applies no
97+// defaults.
98+type storeConfig struct {
99+ dir string
100+ max int64
101+ now func() time.Time
102+ // grace is how long a blob flight with no waiter left keeps downloading in
103+ // case another request joins it; then it is cancelled.
104+ grace time.Duration
105+ // maxDownloads caps the concurrent blob downloads.
106+ maxDownloads int
107+ // slotWait is how long a request waits for a download slot before errBusy.
108+ slotWait time.Duration
109+ log zerowrap.Logger
110+}
111+
112+type store struct {
113+ storeConfig
114+
115+ mu sync.Mutex
116+ total, pending int64
117+ downloads int
118+ // slotFreed is closed and replaced each time a download slot frees.
119+ slotFreed chan struct{}
120+
121+ entries map[string]*entry
122+ lru *list.List // *entry, least recently used first
123+ tagsOf map[string]map[string]struct{} // manifest digest -> tag keys
124+ flights map[string]*flight
125+
126+ hookCommitted func(key string) // tests: after a commit, before waking waiters
127+ hookBug func(msg string) // tests: called on an accounting bug
128+}
129+
130+// defaultBugHook is each new store's hookBug. Tests set it in TestMain, before
131+// any store exists, so that an accounting bug fails the run.
132+var defaultBugHook func(msg string)
133+
134+// storedFile pins its entry until the descriptor is closed. In particular, an
135+// unlinked-but-open blob cannot silently escape the disk accounting.
136+type storedFile struct {
137+ *os.File
138+ s *store
139+ key string
140+ once sync.Once
141+ err error
142+}
143+
144+func (f *storedFile) Close() error {
145+ f.once.Do(func() {
146+ f.err = f.File.Close()
147+ f.s.mu.Lock()
148+ defer f.s.mu.Unlock()
149+ f.s.unpinLocked(f.key)
150+ })
151+ return f.err
152+}
153+
154+func openStore(c storeConfig) (*store, error) {
155+ if c.max < minCharge {
156+ return nil, fmt.Errorf("registry cache cap must be at least %d bytes", minCharge)
157+ }
158+ if c.now == nil || c.grace <= 0 || c.maxDownloads <= 0 || c.slotWait <= 0 {
159+ return nil, errors.New("registry store configuration incomplete")
160+ }
161+ dir := c.dir
162+ s := &store{storeConfig: c, entries: map[string]*entry{}, lru: list.New(), tagsOf: map[string]map[string]struct{}{}, flights: map[string]*flight{}, slotFreed: make(chan struct{}), hookBug: defaultBugHook}
163+ for _, sub := range []string{kindBlob.dir(), kindManifest.dir(), kindTag.dir(), "tmp"} {
164+ if err := os.MkdirAll(filepath.Join(dir, sub), 0700); err != nil {
165+ return nil, err
166+ }
167+ }
168+ temps, err := os.ReadDir(filepath.Join(dir, "tmp"))
169+ if err != nil {
170+ return nil, err
171+ }
172+ for _, de := range temps {
173+ if err := os.Remove(filepath.Join(dir, "tmp", de.Name())); err != nil {
174+ return nil, err
175+ }
176+ }
177+ var types []string
178+ for _, k := range kinds {
179+ ds, err := os.ReadDir(filepath.Join(dir, k.dir()))
180+ if err != nil {
181+ return nil, err
182+ }
183+ for _, de := range ds {
184+ if base, ok := strings.CutSuffix(de.Name(), ".type"); ok && k == kindManifest {
185+ types = append(types, base)
186+ continue
187+ }
188+ if !validDigest("sha256:" + de.Name()) {
189+ continue
190+ }
191+ fi, err := de.Info()
192+ if err != nil {
193+ return nil, err
194+ }
195+ if !fi.Mode().IsRegular() {
196+ continue
197+ }
198+ key := k.key(de.Name())
199+ size := fi.Size()
200+ e := &entry{key: key, kind: k, used: fi.ModTime()}
201+ switch k {
202+ case kindManifest:
203+ info, err := os.Stat(filepath.Join(dir, key) + ".type")
204+ if err != nil {
205+ _ = os.Remove(filepath.Join(dir, key))
206+ continue
207+ }
208+ size += info.Size()
209+ case kindTag:
210+ digest := s.readTagFile(key)
211+ if !validDigest(digest) {
212+ _ = os.Remove(filepath.Join(dir, key))
213+ continue
214+ }
215+ if _, ok := s.entries[kindManifest.key(digest[7:])]; !ok {
216+ _ = os.Remove(filepath.Join(dir, key))
217+ continue
218+ }
219+ size = 0
220+ e.digest = digest
221+ }
222+ e.size = charge(size)
223+ s.addLocked(e)
224+ }
225+ }
226+ // A .type whose manifest is gone (a crash between the two writes).
227+ for _, base := range types {
228+ if _, ok := s.entries[kindManifest.key(base)]; !ok {
229+ _ = os.Remove(filepath.Join(dir, kindManifest.key(base)+".type"))
230+ }
231+ }
232+ // Order the LRU by the access times kept in the mtimes.
233+ all := make([]*entry, 0, len(s.entries))
234+ for _, e := range s.entries {
235+ all = append(all, e)
236+ }
237+ sort.Slice(all, func(i, j int) bool { return all[i].used.Before(all[j].used) })
238+ s.lru.Init()
239+ for _, e := range all {
240+ e.elem = s.lru.PushBack(e)
241+ }
242+ s.evictLocked(0)
243+ return s, nil
244+}
245+
246+func (s *store) addLocked(e *entry) {
247+ s.entries[e.key] = e
248+ e.elem = s.lru.PushBack(e)
249+ s.total += e.size
250+ if e.digest != "" {
251+ s.indexTagLocked(e.key, e.digest)
252+ }
253+}
254+
255+func (s *store) indexTagLocked(key, digest string) {
256+ set := s.tagsOf[digest]
257+ if set == nil {
258+ set = map[string]struct{}{}
259+ s.tagsOf[digest] = set
260+ }
261+ set[key] = struct{}{}
262+}
263+
264+func (s *store) unindexTagLocked(key, digest string) {
265+ if set := s.tagsOf[digest]; set != nil {
266+ delete(set, key)
267+ if len(set) == 0 {
268+ delete(s.tagsOf, digest)
269+ }
270+ }
271+}
272+
273+func (s *store) touchLocked(e *entry) {
274+ e.used = s.now()
275+ s.lru.MoveToBack(e.elem)
276+ _ = os.Chtimes(filepath.Join(s.dir, e.key), e.used, e.used)
277+}
278+
279+func (s *store) unpinLocked(key string) {
280+ e, ok := s.entries[key]
281+ if !ok {
282+ return
283+ }
284+ if e.pins <= 0 {
285+ // A bug: the entry could have been evicted while open.
286+ s.log.Error().Str("key", key).Msg("registry cache unpin of an unpinned entry")
287+ if s.hookBug != nil {
288+ s.hookBug("unpin of an unpinned entry " + key)
289+ }
290+ return
291+ }
292+ e.pins--
293+}
294+
295+// removeLocked only releases a charge after the corresponding file is gone.
296+// Failures leave the entry charged so a full or unwritable disk fails closed.
297+func (s *store) removeLocked(key string) bool {
298+ e, ok := s.entries[key]
299+ if !ok {
300+ return true
301+ }
302+ if e.pins > 0 {
303+ return false
304+ }
305+ if e.kind == kindManifest {
306+ // Prune references first. If a tag cannot be removed, retain its target too.
307+ for k := range s.tagsOf["sha256:"+strings.TrimPrefix(key, kindManifest.key(""))] {
308+ if !s.removeLocked(k) {
309+ return false
310+ }
311+ }
312+ if err := os.Remove(filepath.Join(s.dir, key) + ".type"); err != nil && !os.IsNotExist(err) {
313+ return false
314+ }
315+ }
316+ if err := os.Remove(filepath.Join(s.dir, key)); err != nil && !os.IsNotExist(err) {
317+ return false
318+ }
319+ s.total -= e.size
320+ delete(s.entries, key)
321+ s.lru.Remove(e.elem)
322+ e.elem = nil
323+ if e.digest != "" {
324+ s.unindexTagLocked(key, e.digest)
325+ }
326+ return true
327+}
328+
329+// evictLocked makes room for need bytes. Once over the cap, it walks the LRU
330+// (no sort) and evicts unpinned entries until max(need, low water) bytes are
331+// free, so a full store evicts once per low-water bytes, not on every write.
332+func (s *store) evictLocked(need int64) {
333+ if s.total+s.pending+need <= s.max {
334+ return
335+ }
336+ target := s.max - max(need, min(lowWater, s.max/20))
337+ for el := s.lru.Front(); el != nil && s.total+s.pending > target; {
338+ e := el.Value.(*entry)
339+ next := el.Next()
340+ if e.pins == 0 {
341+ removed := s.removeLocked(e.key)
342+ if e.kind == kindManifest {
343+ // Removing a manifest removes its tags first, even when it then
344+ // fails: next may be gone. Start over; only pinned or
345+ // unremovable entries are walked again.
346+ next = s.lru.Front()
347+ if !removed {
348+ // Do not walk into it again: it stays where it is, and
349+ // the walk resumes after it.
350+ next = e.elem.Next()
351+ }
352+ }
353+ }
354+ el = next
355+ }
356+}
357+
358+func (s *store) openLocked(key string) (*storedFile, string, error) {
359+ e, ok := s.entries[key]
360+ if !ok {
361+ return nil, "", os.ErrNotExist
362+ }
363+ f, err := os.Open(filepath.Join(s.dir, key))
364+ if err != nil {
365+ return nil, "", err
366+ }
367+ media := "application/octet-stream"
368+ if e.kind == kindManifest {
369+ b, err := os.ReadFile(filepath.Join(s.dir, key) + ".type")
370+ if err != nil {
371+ _ = f.Close()
372+ return nil, "", err
373+ }
374+ media = string(b)
375+ }
376+ s.touchLocked(e)
377+ e.pins++
378+ return &storedFile{File: f, s: s, key: key}, media, nil
379+}
380+func (s *store) open(k kind, digest string) (*storedFile, string, error) {
381+ if !validDigest(digest) {
382+ return nil, "", errDigest
383+ }
384+ s.mu.Lock()
385+ defer s.mu.Unlock()
386+ return s.openLocked(k.key(digest[7:]))
387+}
388+
389+// fetch writes the upstream content to w. ctx belongs to the flight: it ends
390+// when the last waiter has been gone for the grace period.
391+type fetch func(ctx context.Context, w io.Writer) (string, error)
392+
393+type keepalive struct {
394+ every time.Duration
395+ fn func()
396+}
397+
398+// ensure returns the stored content, pinned for the caller; hit reports that it
399+// was already stored. Otherwise the request joins or starts the flight of that
400+// digest. The flight runs on its own goroutine and context, so no single
401+// request, the initiator included, can cancel it while another one waits; it is
402+// cancelled once no request has waited for it for the grace period. Every
403+// waiter waits for the verified commit and then reads the stored file. While
404+// waiting, keep.fn is called every keep.every when the download has made
405+// progress, so that the waiter's connection deadlines can follow it.
406+//
407+// Starting a blob flight takes a global download slot and one of the quota
408+// carried by ctx (WithQuota), both held until the flight really ends; joining
409+// one takes neither. When a slot is missing, the request waits for one, kept
410+// alive, for at most the slot wait, and then fails with errBusy.
411+func (s *store) ensure(ctx context.Context, k kind, digest string, fetch fetch, keep keepalive) (f *storedFile, media string, hit bool, err error) {
412+ if !validDigest(digest) {
413+ return nil, "", false, errDigest
414+ }
415+ key := k.key(digest[7:])
416+ blob := k == kindBlob
417+ quota := quotaFrom(ctx)
418+ var slotDeadline <-chan time.Time
419+ var fl *flight
420+ for {
421+ s.mu.Lock()
422+ if f, media, err := s.openLocked(key); err == nil {
423+ s.mu.Unlock()
424+ return f, media, true, nil
425+ }
426+ var ok bool
427+ if fl, ok = s.flights[key]; ok {
428+ break // join: no slot needed
429+ }
430+ if !blob || s.takeSlotLocked(quota) {
431+ dctx, cancel := context.WithCancel(context.Background())
432+ fl = &flight{key: key, kind: k, done: make(chan struct{}), cancel: cancel}
433+ if blob {
434+ fl.quota = quota
435+ }
436+ s.flights[key] = fl
437+ go s.run(dctx, fl, digest, fetch)
438+ break
439+ }
440+ freed := s.slotFreed
441+ s.mu.Unlock()
442+ if slotDeadline == nil {
443+ t := time.NewTimer(s.slotWait)
444+ defer t.Stop()
445+ slotDeadline = t.C
446+ }
447+ if err := waitSlot(ctx, freed, slotDeadline, keep); err != nil {
448+ return nil, "", false, err
449+ }
450+ }
451+ fl.waiters++
452+ s.mu.Unlock()
453+
454+ err = waitFlight(ctx, fl, keep)
455+ s.mu.Lock()
456+ defer s.mu.Unlock()
457+ if err == nil {
458+ f, media, err = s.openLocked(key)
459+ }
460+ s.leaveLocked(fl)
461+ return f, media, false, err
462+}
463+
464+// leaveLocked drops a waiter. The last one releases the commit pin, or, before
465+// the commit, starts the grace period after which the flight is cancelled.
466+func (s *store) leaveLocked(fl *flight) {
467+ fl.waiters--
468+ if fl.waiters > 0 {
469+ return
470+ }
471+ if fl.finished {
472+ if fl.pinned {
473+ fl.pinned = false
474+ s.unpinLocked(fl.key)
475+ }
476+ return
477+ }
478+ fl.gen++
479+ gen := fl.gen
480+ // A manifest is small and bounded by the request timeout: it is cancelled at
481+ // once, so its flights never outnumber the requests in progress. A blob
482+ // holds a download slot until it really ends, which bounds the abandoned
483+ // ones to MaxDownloads.
484+ if fl.kind != kindBlob {
485+ s.abandonLocked(fl)
486+ return
487+ }
488+ time.AfterFunc(s.grace, func() {
489+ s.mu.Lock()
490+ defer s.mu.Unlock()
491+ if fl.waiters == 0 && !fl.finished && fl.gen == gen {
492+ s.abandonLocked(fl)
493+ }
494+ })
495+}
496+
497+// abandonLocked cancels a flight nobody waits for. A later request for the
498+// digest starts a new one.
499+func (s *store) abandonLocked(fl *flight) {
500+ fl.cancel()
501+ if s.flights[fl.key] == fl {
502+ delete(s.flights, fl.key)
503+ }
504+}
505+
506+// takeSlotLocked takes a global download slot and one of quota (nil: none),
507+// both or neither.
508+func (s *store) takeSlotLocked(quota *Quota) bool {
509+ if s.downloads >= s.maxDownloads {
510+ return false
511+ }
512+ if quota != nil && !quota.tryAcquire() {
513+ return false
514+ }
515+ s.downloads++
516+ return true
517+}
518+
519+// waitSlot waits until a slot may have freed (freed is closed), keeping the
520+// client alive meanwhile. It fails with errBusy at the deadline.
521+func waitSlot(ctx context.Context, freed <-chan struct{}, deadline <-chan time.Time, keep keepalive) error {
522+ var tick <-chan time.Time
523+ if keep.every > 0 && keep.fn != nil {
524+ t := time.NewTicker(keep.every)
525+ defer t.Stop()
526+ tick = t.C
527+ }
528+ for {
529+ select {
530+ case <-freed:
531+ return nil
532+ case <-ctx.Done():
533+ return ctx.Err()
534+ case <-deadline:
535+ return errBusy
536+ case <-tick:
537+ keep.fn() // bounded by the slot wait
538+ }
539+ }
540+}
541+
542+// finishLocked ends a flight and releases its slots. A committed entry is
543+// pinned on behalf of the waiters, so that no eviction removes it before they
544+// open it.
545+func (s *store) finishLocked(fl *flight, err error, committed *entry) {
546+ fl.err = err
547+ fl.finished = true
548+ if err == nil && committed != nil && fl.waiters > 0 {
549+ committed.pins++
550+ fl.pinned = true
551+ }
552+ if s.flights[fl.key] == fl {
553+ delete(s.flights, fl.key)
554+ }
555+ if fl.kind == kindBlob {
556+ s.downloads--
557+ if fl.quota != nil {
558+ fl.quota.release()
559+ }
560+ close(s.slotFreed)
561+ s.slotFreed = make(chan struct{})
562+ }
563+}
564+
565+// run downloads and finishes the flight. A panic in fetch fails the flight
566+// (and frees its slots) rather than leave its waiters hanging.
567+func (s *store) run(ctx context.Context, fl *flight, digest string, fetch fetch) {
568+ var err error
569+ defer func() {
570+ if v := recover(); v != nil {
571+ err = fmt.Errorf("registry download panicked: %v", v)
572+ s.log.Error().Str("key", fl.key).Err(err).Msg("registry download")
573+ }
574+ fl.cancel()
575+ s.mu.Lock()
576+ if !fl.finished {
577+ if err == nil {
578+ err = errors.New("registry download ended without a commit")
579+ }
580+ s.finishLocked(fl, err, nil)
581+ }
582+ s.mu.Unlock()
583+ if fl.err == nil && s.hookCommitted != nil {
584+ s.hookCommitted(fl.key)
585+ }
586+ close(fl.done)
587+ }()
588+ err = s.fill(ctx, fl, digest, fetch)
589+}
590+
591+func waitFlight(ctx context.Context, fl *flight, keep keepalive) error {
592+ var tick <-chan time.Time
593+ if keep.every > 0 && keep.fn != nil {
594+ t := time.NewTicker(keep.every)
595+ defer t.Stop()
596+ tick = t.C
597+ }
598+ last := fl.written.Load()
599+ for {
600+ select {
601+ case <-ctx.Done():
602+ return ctx.Err()
603+ case <-fl.done:
604+ return fl.err
605+ case <-tick:
606+ if n := fl.written.Load(); n != last {
607+ last = n
608+ keep.fn()
609+ }
610+ }
611+ }
612+}
613+
614+type chargedWriter struct {
615+ s *store
616+ w io.Writer
617+ n int64
618+ written *atomic.Int64
619+}
620+
621+func (w *chargedWriter) Write(b []byte) (int, error) {
622+ s := w.s
623+ s.mu.Lock()
624+ need := int64(len(b))
625+ s.evictLocked(need)
626+ if need > s.max-s.total-s.pending {
627+ s.mu.Unlock()
628+ return 0, errTooLarge
629+ }
630+ s.pending += need
631+ w.n += need
632+ s.mu.Unlock()
633+ n, err := w.w.Write(b)
634+ if n < len(b) {
635+ s.mu.Lock()
636+ s.pending -= int64(len(b) - n)
637+ w.n -= int64(len(b) - n)
638+ s.mu.Unlock()
639+ }
640+ w.written.Add(int64(n))
641+ return n, err
642+}
643+
644+// fill downloads into tmp/, verifies the digest and commits. On success it has
645+// finished the flight (finishLocked) under the same lock as the commit.
646+func (s *store) fill(ctx context.Context, fl *flight, digest string, fetch fetch) error {
647+ key := fl.key
648+ f, err := os.CreateTemp(filepath.Join(s.dir, "tmp"), "download-")
649+ if err != nil {
650+ return err
651+ }
652+ defer func() { _ = f.Close(); _ = os.Remove(f.Name()) }()
653+ h := sha256.New()
654+ cw := &chargedWriter{s: s, w: io.MultiWriter(f, h), written: &fl.written}
655+ defer func() { s.mu.Lock(); s.pending -= cw.n; s.mu.Unlock() }()
656+ media, err := fetch(ctx, cw)
657+ if err != nil {
658+ return err
659+ }
660+ if "sha256:"+hex.EncodeToString(h.Sum(nil)) != digest {
661+ return errDigest
662+ }
663+ if err = f.Close(); err != nil {
664+ return err
665+ }
666+ s.mu.Lock()
667+ defer s.mu.Unlock()
668+ if e, ok := s.entries[key]; ok {
669+ // Same digest, same bytes: keep the stored file and its pins rather than
670+ // renaming over a file that may be open (its bytes would leave the
671+ // accounting while still allocated).
672+ s.finishLocked(fl, nil, e)
673+ return nil
674+ }
675+ path := filepath.Join(s.dir, key)
676+ metadata := int64(0)
677+ if fl.kind == kindManifest {
678+ if len(media) > 1024 {
679+ return errors.New("manifest media type too large")
680+ }
681+ metadata = int64(len(media))
682+ }
683+ // The download is charged its bytes in pending; the entry is charged at
684+ // least minCharge, and its media type.
685+ size := charge(cw.n + metadata)
686+ if extra := size - cw.n; extra > 0 {
687+ s.evictLocked(extra)
688+ if extra > s.max-s.total-s.pending {
689+ return errTooLarge
690+ }
691+ }
692+ if fl.kind == kindManifest {
693+ if err := os.WriteFile(path+".type", []byte(media), 0600); err != nil {
694+ return err
695+ }
696+ }
697+ if err := os.Rename(f.Name(), path); err != nil {
698+ if fl.kind == kindManifest {
699+ _ = os.Remove(path + ".type")
700+ }
701+ return err
702+ }
703+ e := &entry{key: key, kind: fl.kind, size: size, used: s.now()}
704+ s.addLocked(e)
705+ s.pending -= cw.n
706+ cw.n = 0
707+ s.finishLocked(fl, nil, e)
708+ return nil
709+}
710+
711+// tagKey is the key of a tag for a normalized Accept (normalizeAccept).
712+func tagKey(name, tag, accept string) string {
713+ return kindTag.key(fmt.Sprintf("%x", sha256.Sum256([]byte(name+"\x00"+tag+"\x00"+accept))))
714+}
715+func (s *store) readTagFile(key string) string {
716+ f, err := os.Open(filepath.Join(s.dir, key))
717+ if err != nil {
718+ return ""
719+ }
720+ defer func() { _ = f.Close() }()
721+ b, err := io.ReadAll(io.LimitReader(f, 72))
722+ if err != nil || !validDigest(string(b)) {
723+ return ""
724+ }
725+ return string(b)
726+}
727+func (s *store) tag(name, tag, accept string) string {
728+ s.mu.Lock()
729+ defer s.mu.Unlock()
730+ e, ok := s.entries[tagKey(name, tag, accept)]
731+ if !ok {
732+ return ""
733+ }
734+ s.touchLocked(e)
735+ return e.digest
736+}
737+func (s *store) setTag(name, tag, accept, digest string) error {
738+ if !validDigest(digest) {
739+ return errDigest
740+ }
741+ s.mu.Lock()
742+ defer s.mu.Unlock()
743+ manifest, ok := s.entries[kindManifest.key(digest[7:])]
744+ if !ok {
745+ return os.ErrNotExist
746+ }
747+ // Protect the target while making room for its metadata.
748+ manifest.pins++
749+ defer func() { manifest.pins-- }()
750+ key := tagKey(name, tag, accept)
751+ old, exists := s.entries[key]
752+ if exists {
753+ // Already charged: rewriting it needs no room.
754+ s.touchLocked(old)
755+ } else {
756+ s.evictLocked(charge(0))
757+ if charge(0) > s.max-s.total-s.pending {
758+ return errTooLarge
759+ }
760+ }
761+ f, err := os.CreateTemp(filepath.Join(s.dir, "tmp"), "tag-")
762+ if err != nil {
763+ return err
764+ }
765+ defer func() { _ = f.Close(); _ = os.Remove(f.Name()) }()
766+ if _, err = f.WriteString(digest); err != nil {
767+ return err
768+ }
769+ if err = f.Close(); err != nil {
770+ return err
771+ }
772+ if err = os.Rename(f.Name(), filepath.Join(s.dir, key)); err != nil {
773+ return err
774+ }
775+ if old, ok := s.entries[key]; ok {
776+ s.unindexTagLocked(key, old.digest)
777+ old.digest = digest
778+ old.used = s.now()
779+ s.indexTagLocked(key, digest)
780+ return nil
781+ }
782+ s.addLocked(&entry{key: key, kind: kindTag, size: charge(0), used: s.now(), digest: digest})
783+ return nil
784+}
@@ -0,0 +1,784 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "container/list"
7+ "context"
8+ "crypto/sha256"
9+ "encoding/hex"
10+ "errors"
11+ "fmt"
12+ "io"
13+ "os"
14+ "path/filepath"
15+ "sort"
16+ "strings"
17+ "sync"
18+ "sync/atomic"
19+ "time"
20+
21+ "github.com/bnema/zerowrap"
22+)
23+
24+var errTooLarge = errors.New("registry cache capacity exceeded")
25+var errDigest = errors.New("registry digest mismatch")
26+
27+// errBusy: no download slot came free within the slot wait. The proxy answers
28+// 503, and dockerd retries the layer (about 50 s) before pulling from Docker Hub.
29+var errBusy = errors.New("registry downloads busy")
30+
31+// minCharge is the least an entry is charged, whatever its size: about one
32+// disk block and inode. It bounds the number of entries, and so of files, to
33+// Max/minCharge. A tag file is charged exactly that.
34+const minCharge int64 = 4096
35+
36+// charge is what an entry of n bytes counts against Max.
37+func charge(n int64) int64 { return max(n, minCharge) }
38+
39+// kind is the directory of an entry: content (blobs, manifests) or a tag.
40+type kind uint8
41+
42+const (
43+ kindBlob kind = iota
44+ kindManifest
45+ kindTag
46+)
47+
48+var kinds = []kind{kindBlob, kindManifest, kindTag}
49+
50+// dir is the directory of the kind in the store, and its route in /v2/.
51+func (k kind) dir() string {
52+ switch k {
53+ case kindBlob:
54+ return "blobs"
55+ case kindManifest:
56+ return "manifests"
57+ default:
58+ return "tags"
59+ }
60+}
61+
62+// key is the key of an entry of kind k: its path in the store.
63+func (k kind) key(name string) string { return k.dir() + "/" + name }
64+
65+// lowWater bounds the extra room an eviction frees (with 5% of Max), so that a
66+// full store evicts once per that many bytes, not on every write.
67+const lowWater int64 = 64 << 20
68+
69+type entry struct {
70+ key string
71+ kind kind
72+ size int64
73+ used time.Time
74+ pins int
75+ digest string // tag target, empty for content
76+ elem *list.Element // in store.lru; nil once removed
77+}
78+
79+// flight is one download, shared by every request for its digest. Its fields
80+// below written are guarded by store.mu.
81+type flight struct {
82+ key string
83+ done chan struct{}
84+ err error
85+ written atomic.Int64 // bytes received so far, for the waiters' keepalive
86+
87+ waiters int
88+ gen int // bumped each time waiters drops to zero
89+ finished bool
90+ pinned bool // the committed entry holds a pin for the waiters
91+ kind kind // a blob flight takes a download slot
92+ quota *Quota // the starter's quota, held until the flight ends
93+ cancel context.CancelFunc
94+}
95+
96+// storeConfig is a store's complete configuration: openStore applies no
97+// defaults.
98+type storeConfig struct {
99+ dir string
100+ max int64
101+ now func() time.Time
102+ // grace is how long a blob flight with no waiter left keeps downloading in
103+ // case another request joins it; then it is cancelled.
104+ grace time.Duration
105+ // maxDownloads caps the concurrent blob downloads.
106+ maxDownloads int
107+ // slotWait is how long a request waits for a download slot before errBusy.
108+ slotWait time.Duration
109+ log zerowrap.Logger
110+}
111+
112+type store struct {
113+ storeConfig
114+
115+ mu sync.Mutex
116+ total, pending int64
117+ downloads int
118+ // slotFreed is closed and replaced each time a download slot frees.
119+ slotFreed chan struct{}
120+
121+ entries map[string]*entry
122+ lru *list.List // *entry, least recently used first
123+ tagsOf map[string]map[string]struct{} // manifest digest -> tag keys
124+ flights map[string]*flight
125+
126+ hookCommitted func(key string) // tests: after a commit, before waking waiters
127+ hookBug func(msg string) // tests: called on an accounting bug
128+}
129+
130+// defaultBugHook is each new store's hookBug. Tests set it in TestMain, before
131+// any store exists, so that an accounting bug fails the run.
132+var defaultBugHook func(msg string)
133+
134+// storedFile pins its entry until the descriptor is closed. In particular, an
135+// unlinked-but-open blob cannot silently escape the disk accounting.
136+type storedFile struct {
137+ *os.File
138+ s *store
139+ key string
140+ once sync.Once
141+ err error
142+}
143+
144+func (f *storedFile) Close() error {
145+ f.once.Do(func() {
146+ f.err = f.File.Close()
147+ f.s.mu.Lock()
148+ defer f.s.mu.Unlock()
149+ f.s.unpinLocked(f.key)
150+ })
151+ return f.err
152+}
153+
154+func openStore(c storeConfig) (*store, error) {
155+ if c.max < minCharge {
156+ return nil, fmt.Errorf("registry cache cap must be at least %d bytes", minCharge)
157+ }
158+ if c.now == nil || c.grace <= 0 || c.maxDownloads <= 0 || c.slotWait <= 0 {
159+ return nil, errors.New("registry store configuration incomplete")
160+ }
161+ dir := c.dir
162+ s := &store{storeConfig: c, entries: map[string]*entry{}, lru: list.New(), tagsOf: map[string]map[string]struct{}{}, flights: map[string]*flight{}, slotFreed: make(chan struct{}), hookBug: defaultBugHook}
163+ for _, sub := range []string{kindBlob.dir(), kindManifest.dir(), kindTag.dir(), "tmp"} {
164+ if err := os.MkdirAll(filepath.Join(dir, sub), 0700); err != nil {
165+ return nil, err
166+ }
167+ }
168+ temps, err := os.ReadDir(filepath.Join(dir, "tmp"))
169+ if err != nil {
170+ return nil, err
171+ }
172+ for _, de := range temps {
173+ if err := os.Remove(filepath.Join(dir, "tmp", de.Name())); err != nil {
174+ return nil, err
175+ }
176+ }
177+ var types []string
178+ for _, k := range kinds {
179+ ds, err := os.ReadDir(filepath.Join(dir, k.dir()))
180+ if err != nil {
181+ return nil, err
182+ }
183+ for _, de := range ds {
184+ if base, ok := strings.CutSuffix(de.Name(), ".type"); ok && k == kindManifest {
185+ types = append(types, base)
186+ continue
187+ }
188+ if !validDigest("sha256:" + de.Name()) {
189+ continue
190+ }
191+ fi, err := de.Info()
192+ if err != nil {
193+ return nil, err
194+ }
195+ if !fi.Mode().IsRegular() {
196+ continue
197+ }
198+ key := k.key(de.Name())
199+ size := fi.Size()
200+ e := &entry{key: key, kind: k, used: fi.ModTime()}
201+ switch k {
202+ case kindManifest:
203+ info, err := os.Stat(filepath.Join(dir, key) + ".type")
204+ if err != nil {
205+ _ = os.Remove(filepath.Join(dir, key))
206+ continue
207+ }
208+ size += info.Size()
209+ case kindTag:
210+ digest := s.readTagFile(key)
211+ if !validDigest(digest) {
212+ _ = os.Remove(filepath.Join(dir, key))
213+ continue
214+ }
215+ if _, ok := s.entries[kindManifest.key(digest[7:])]; !ok {
216+ _ = os.Remove(filepath.Join(dir, key))
217+ continue
218+ }
219+ size = 0
220+ e.digest = digest
221+ }
222+ e.size = charge(size)
223+ s.addLocked(e)
224+ }
225+ }
226+ // A .type whose manifest is gone (a crash between the two writes).
227+ for _, base := range types {
228+ if _, ok := s.entries[kindManifest.key(base)]; !ok {
229+ _ = os.Remove(filepath.Join(dir, kindManifest.key(base)+".type"))
230+ }
231+ }
232+ // Order the LRU by the access times kept in the mtimes.
233+ all := make([]*entry, 0, len(s.entries))
234+ for _, e := range s.entries {
235+ all = append(all, e)
236+ }
237+ sort.Slice(all, func(i, j int) bool { return all[i].used.Before(all[j].used) })
238+ s.lru.Init()
239+ for _, e := range all {
240+ e.elem = s.lru.PushBack(e)
241+ }
242+ s.evictLocked(0)
243+ return s, nil
244+}
245+
246+func (s *store) addLocked(e *entry) {
247+ s.entries[e.key] = e
248+ e.elem = s.lru.PushBack(e)
249+ s.total += e.size
250+ if e.digest != "" {
251+ s.indexTagLocked(e.key, e.digest)
252+ }
253+}
254+
255+func (s *store) indexTagLocked(key, digest string) {
256+ set := s.tagsOf[digest]
257+ if set == nil {
258+ set = map[string]struct{}{}
259+ s.tagsOf[digest] = set
260+ }
261+ set[key] = struct{}{}
262+}
263+
264+func (s *store) unindexTagLocked(key, digest string) {
265+ if set := s.tagsOf[digest]; set != nil {
266+ delete(set, key)
267+ if len(set) == 0 {
268+ delete(s.tagsOf, digest)
269+ }
270+ }
271+}
272+
273+func (s *store) touchLocked(e *entry) {
274+ e.used = s.now()
275+ s.lru.MoveToBack(e.elem)
276+ _ = os.Chtimes(filepath.Join(s.dir, e.key), e.used, e.used)
277+}
278+
279+func (s *store) unpinLocked(key string) {
280+ e, ok := s.entries[key]
281+ if !ok {
282+ return
283+ }
284+ if e.pins <= 0 {
285+ // A bug: the entry could have been evicted while open.
286+ s.log.Error().Str("key", key).Msg("registry cache unpin of an unpinned entry")
287+ if s.hookBug != nil {
288+ s.hookBug("unpin of an unpinned entry " + key)
289+ }
290+ return
291+ }
292+ e.pins--
293+}
294+
295+// removeLocked only releases a charge after the corresponding file is gone.
296+// Failures leave the entry charged so a full or unwritable disk fails closed.
297+func (s *store) removeLocked(key string) bool {
298+ e, ok := s.entries[key]
299+ if !ok {
300+ return true
301+ }
302+ if e.pins > 0 {
303+ return false
304+ }
305+ if e.kind == kindManifest {
306+ // Prune references first. If a tag cannot be removed, retain its target too.
307+ for k := range s.tagsOf["sha256:"+strings.TrimPrefix(key, kindManifest.key(""))] {
308+ if !s.removeLocked(k) {
309+ return false
310+ }
311+ }
312+ if err := os.Remove(filepath.Join(s.dir, key) + ".type"); err != nil && !os.IsNotExist(err) {
313+ return false
314+ }
315+ }
316+ if err := os.Remove(filepath.Join(s.dir, key)); err != nil && !os.IsNotExist(err) {
317+ return false
318+ }
319+ s.total -= e.size
320+ delete(s.entries, key)
321+ s.lru.Remove(e.elem)
322+ e.elem = nil
323+ if e.digest != "" {
324+ s.unindexTagLocked(key, e.digest)
325+ }
326+ return true
327+}
328+
329+// evictLocked makes room for need bytes. Once over the cap, it walks the LRU
330+// (no sort) and evicts unpinned entries until max(need, low water) bytes are
331+// free, so a full store evicts once per low-water bytes, not on every write.
332+func (s *store) evictLocked(need int64) {
333+ if s.total+s.pending+need <= s.max {
334+ return
335+ }
336+ target := s.max - max(need, min(lowWater, s.max/20))
337+ for el := s.lru.Front(); el != nil && s.total+s.pending > target; {
338+ e := el.Value.(*entry)
339+ next := el.Next()
340+ if e.pins == 0 {
341+ removed := s.removeLocked(e.key)
342+ if e.kind == kindManifest {
343+ // Removing a manifest removes its tags first, even when it then
344+ // fails: next may be gone. Start over; only pinned or
345+ // unremovable entries are walked again.
346+ next = s.lru.Front()
347+ if !removed {
348+ // Do not walk into it again: it stays where it is, and
349+ // the walk resumes after it.
350+ next = e.elem.Next()
351+ }
352+ }
353+ }
354+ el = next
355+ }
356+}
357+
358+func (s *store) openLocked(key string) (*storedFile, string, error) {
359+ e, ok := s.entries[key]
360+ if !ok {
361+ return nil, "", os.ErrNotExist
362+ }
363+ f, err := os.Open(filepath.Join(s.dir, key))
364+ if err != nil {
365+ return nil, "", err
366+ }
367+ media := "application/octet-stream"
368+ if e.kind == kindManifest {
369+ b, err := os.ReadFile(filepath.Join(s.dir, key) + ".type")
370+ if err != nil {
371+ _ = f.Close()
372+ return nil, "", err
373+ }
374+ media = string(b)
375+ }
376+ s.touchLocked(e)
377+ e.pins++
378+ return &storedFile{File: f, s: s, key: key}, media, nil
379+}
380+func (s *store) open(k kind, digest string) (*storedFile, string, error) {
381+ if !validDigest(digest) {
382+ return nil, "", errDigest
383+ }
384+ s.mu.Lock()
385+ defer s.mu.Unlock()
386+ return s.openLocked(k.key(digest[7:]))
387+}
388+
389+// fetch writes the upstream content to w. ctx belongs to the flight: it ends
390+// when the last waiter has been gone for the grace period.
391+type fetch func(ctx context.Context, w io.Writer) (string, error)
392+
393+type keepalive struct {
394+ every time.Duration
395+ fn func()
396+}
397+
398+// ensure returns the stored content, pinned for the caller; hit reports that it
399+// was already stored. Otherwise the request joins or starts the flight of that
400+// digest. The flight runs on its own goroutine and context, so no single
401+// request, the initiator included, can cancel it while another one waits; it is
402+// cancelled once no request has waited for it for the grace period. Every
403+// waiter waits for the verified commit and then reads the stored file. While
404+// waiting, keep.fn is called every keep.every when the download has made
405+// progress, so that the waiter's connection deadlines can follow it.
406+//
407+// Starting a blob flight takes a global download slot and one of the quota
408+// carried by ctx (WithQuota), both held until the flight really ends; joining
409+// one takes neither. When a slot is missing, the request waits for one, kept
410+// alive, for at most the slot wait, and then fails with errBusy.
411+func (s *store) ensure(ctx context.Context, k kind, digest string, fetch fetch, keep keepalive) (f *storedFile, media string, hit bool, err error) {
412+ if !validDigest(digest) {
413+ return nil, "", false, errDigest
414+ }
415+ key := k.key(digest[7:])
416+ blob := k == kindBlob
417+ quota := quotaFrom(ctx)
418+ var slotDeadline <-chan time.Time
419+ var fl *flight
420+ for {
421+ s.mu.Lock()
422+ if f, media, err := s.openLocked(key); err == nil {
423+ s.mu.Unlock()
424+ return f, media, true, nil
425+ }
426+ var ok bool
427+ if fl, ok = s.flights[key]; ok {
428+ break // join: no slot needed
429+ }
430+ if !blob || s.takeSlotLocked(quota) {
431+ dctx, cancel := context.WithCancel(context.Background())
432+ fl = &flight{key: key, kind: k, done: make(chan struct{}), cancel: cancel}
433+ if blob {
434+ fl.quota = quota
435+ }
436+ s.flights[key] = fl
437+ go s.run(dctx, fl, digest, fetch)
438+ break
439+ }
440+ freed := s.slotFreed
441+ s.mu.Unlock()
442+ if slotDeadline == nil {
443+ t := time.NewTimer(s.slotWait)
444+ defer t.Stop()
445+ slotDeadline = t.C
446+ }
447+ if err := waitSlot(ctx, freed, slotDeadline, keep); err != nil {
448+ return nil, "", false, err
449+ }
450+ }
451+ fl.waiters++
452+ s.mu.Unlock()
453+
454+ err = waitFlight(ctx, fl, keep)
455+ s.mu.Lock()
456+ defer s.mu.Unlock()
457+ if err == nil {
458+ f, media, err = s.openLocked(key)
459+ }
460+ s.leaveLocked(fl)
461+ return f, media, false, err
462+}
463+
464+// leaveLocked drops a waiter. The last one releases the commit pin, or, before
465+// the commit, starts the grace period after which the flight is cancelled.
466+func (s *store) leaveLocked(fl *flight) {
467+ fl.waiters--
468+ if fl.waiters > 0 {
469+ return
470+ }
471+ if fl.finished {
472+ if fl.pinned {
473+ fl.pinned = false
474+ s.unpinLocked(fl.key)
475+ }
476+ return
477+ }
478+ fl.gen++
479+ gen := fl.gen
480+ // A manifest is small and bounded by the request timeout: it is cancelled at
481+ // once, so its flights never outnumber the requests in progress. A blob
482+ // holds a download slot until it really ends, which bounds the abandoned
483+ // ones to MaxDownloads.
484+ if fl.kind != kindBlob {
485+ s.abandonLocked(fl)
486+ return
487+ }
488+ time.AfterFunc(s.grace, func() {
489+ s.mu.Lock()
490+ defer s.mu.Unlock()
491+ if fl.waiters == 0 && !fl.finished && fl.gen == gen {
492+ s.abandonLocked(fl)
493+ }
494+ })
495+}
496+
497+// abandonLocked cancels a flight nobody waits for. A later request for the
498+// digest starts a new one.
499+func (s *store) abandonLocked(fl *flight) {
500+ fl.cancel()
501+ if s.flights[fl.key] == fl {
502+ delete(s.flights, fl.key)
503+ }
504+}
505+
506+// takeSlotLocked takes a global download slot and one of quota (nil: none),
507+// both or neither.
508+func (s *store) takeSlotLocked(quota *Quota) bool {
509+ if s.downloads >= s.maxDownloads {
510+ return false
511+ }
512+ if quota != nil && !quota.tryAcquire() {
513+ return false
514+ }
515+ s.downloads++
516+ return true
517+}
518+
519+// waitSlot waits until a slot may have freed (freed is closed), keeping the
520+// client alive meanwhile. It fails with errBusy at the deadline.
521+func waitSlot(ctx context.Context, freed <-chan struct{}, deadline <-chan time.Time, keep keepalive) error {
522+ var tick <-chan time.Time
523+ if keep.every > 0 && keep.fn != nil {
524+ t := time.NewTicker(keep.every)
525+ defer t.Stop()
526+ tick = t.C
527+ }
528+ for {
529+ select {
530+ case <-freed:
531+ return nil
532+ case <-ctx.Done():
533+ return ctx.Err()
534+ case <-deadline:
535+ return errBusy
536+ case <-tick:
537+ keep.fn() // bounded by the slot wait
538+ }
539+ }
540+}
541+
542+// finishLocked ends a flight and releases its slots. A committed entry is
543+// pinned on behalf of the waiters, so that no eviction removes it before they
544+// open it.
545+func (s *store) finishLocked(fl *flight, err error, committed *entry) {
546+ fl.err = err
547+ fl.finished = true
548+ if err == nil && committed != nil && fl.waiters > 0 {
549+ committed.pins++
550+ fl.pinned = true
551+ }
552+ if s.flights[fl.key] == fl {
553+ delete(s.flights, fl.key)
554+ }
555+ if fl.kind == kindBlob {
556+ s.downloads--
557+ if fl.quota != nil {
558+ fl.quota.release()
559+ }
560+ close(s.slotFreed)
561+ s.slotFreed = make(chan struct{})
562+ }
563+}
564+
565+// run downloads and finishes the flight. A panic in fetch fails the flight
566+// (and frees its slots) rather than leave its waiters hanging.
567+func (s *store) run(ctx context.Context, fl *flight, digest string, fetch fetch) {
568+ var err error
569+ defer func() {
570+ if v := recover(); v != nil {
571+ err = fmt.Errorf("registry download panicked: %v", v)
572+ s.log.Error().Str("key", fl.key).Err(err).Msg("registry download")
573+ }
574+ fl.cancel()
575+ s.mu.Lock()
576+ if !fl.finished {
577+ if err == nil {
578+ err = errors.New("registry download ended without a commit")
579+ }
580+ s.finishLocked(fl, err, nil)
581+ }
582+ s.mu.Unlock()
583+ if fl.err == nil && s.hookCommitted != nil {
584+ s.hookCommitted(fl.key)
585+ }
586+ close(fl.done)
587+ }()
588+ err = s.fill(ctx, fl, digest, fetch)
589+}
590+
591+func waitFlight(ctx context.Context, fl *flight, keep keepalive) error {
592+ var tick <-chan time.Time
593+ if keep.every > 0 && keep.fn != nil {
594+ t := time.NewTicker(keep.every)
595+ defer t.Stop()
596+ tick = t.C
597+ }
598+ last := fl.written.Load()
599+ for {
600+ select {
601+ case <-ctx.Done():
602+ return ctx.Err()
603+ case <-fl.done:
604+ return fl.err
605+ case <-tick:
606+ if n := fl.written.Load(); n != last {
607+ last = n
608+ keep.fn()
609+ }
610+ }
611+ }
612+}
613+
614+type chargedWriter struct {
615+ s *store
616+ w io.Writer
617+ n int64
618+ written *atomic.Int64
619+}
620+
621+func (w *chargedWriter) Write(b []byte) (int, error) {
622+ s := w.s
623+ s.mu.Lock()
624+ need := int64(len(b))
625+ s.evictLocked(need)
626+ if need > s.max-s.total-s.pending {
627+ s.mu.Unlock()
628+ return 0, errTooLarge
629+ }
630+ s.pending += need
631+ w.n += need
632+ s.mu.Unlock()
633+ n, err := w.w.Write(b)
634+ if n < len(b) {
635+ s.mu.Lock()
636+ s.pending -= int64(len(b) - n)
637+ w.n -= int64(len(b) - n)
638+ s.mu.Unlock()
639+ }
640+ w.written.Add(int64(n))
641+ return n, err
642+}
643+
644+// fill downloads into tmp/, verifies the digest and commits. On success it has
645+// finished the flight (finishLocked) under the same lock as the commit.
646+func (s *store) fill(ctx context.Context, fl *flight, digest string, fetch fetch) error {
647+ key := fl.key
648+ f, err := os.CreateTemp(filepath.Join(s.dir, "tmp"), "download-")
649+ if err != nil {
650+ return err
651+ }
652+ defer func() { _ = f.Close(); _ = os.Remove(f.Name()) }()
653+ h := sha256.New()
654+ cw := &chargedWriter{s: s, w: io.MultiWriter(f, h), written: &fl.written}
655+ defer func() { s.mu.Lock(); s.pending -= cw.n; s.mu.Unlock() }()
656+ media, err := fetch(ctx, cw)
657+ if err != nil {
658+ return err
659+ }
660+ if "sha256:"+hex.EncodeToString(h.Sum(nil)) != digest {
661+ return errDigest
662+ }
663+ if err = f.Close(); err != nil {
664+ return err
665+ }
666+ s.mu.Lock()
667+ defer s.mu.Unlock()
668+ if e, ok := s.entries[key]; ok {
669+ // Same digest, same bytes: keep the stored file and its pins rather than
670+ // renaming over a file that may be open (its bytes would leave the
671+ // accounting while still allocated).
672+ s.finishLocked(fl, nil, e)
673+ return nil
674+ }
675+ path := filepath.Join(s.dir, key)
676+ metadata := int64(0)
677+ if fl.kind == kindManifest {
678+ if len(media) > 1024 {
679+ return errors.New("manifest media type too large")
680+ }
681+ metadata = int64(len(media))
682+ }
683+ // The download is charged its bytes in pending; the entry is charged at
684+ // least minCharge, and its media type.
685+ size := charge(cw.n + metadata)
686+ if extra := size - cw.n; extra > 0 {
687+ s.evictLocked(extra)
688+ if extra > s.max-s.total-s.pending {
689+ return errTooLarge
690+ }
691+ }
692+ if fl.kind == kindManifest {
693+ if err := os.WriteFile(path+".type", []byte(media), 0600); err != nil {
694+ return err
695+ }
696+ }
697+ if err := os.Rename(f.Name(), path); err != nil {
698+ if fl.kind == kindManifest {
699+ _ = os.Remove(path + ".type")
700+ }
701+ return err
702+ }
703+ e := &entry{key: key, kind: fl.kind, size: size, used: s.now()}
704+ s.addLocked(e)
705+ s.pending -= cw.n
706+ cw.n = 0
707+ s.finishLocked(fl, nil, e)
708+ return nil
709+}
710+
711+// tagKey is the key of a tag for a normalized Accept (normalizeAccept).
712+func tagKey(name, tag, accept string) string {
713+ return kindTag.key(fmt.Sprintf("%x", sha256.Sum256([]byte(name+"\x00"+tag+"\x00"+accept))))
714+}
715+func (s *store) readTagFile(key string) string {
716+ f, err := os.Open(filepath.Join(s.dir, key))
717+ if err != nil {
718+ return ""
719+ }
720+ defer func() { _ = f.Close() }()
721+ b, err := io.ReadAll(io.LimitReader(f, 72))
722+ if err != nil || !validDigest(string(b)) {
723+ return ""
724+ }
725+ return string(b)
726+}
727+func (s *store) tag(name, tag, accept string) string {
728+ s.mu.Lock()
729+ defer s.mu.Unlock()
730+ e, ok := s.entries[tagKey(name, tag, accept)]
731+ if !ok {
732+ return ""
733+ }
734+ s.touchLocked(e)
735+ return e.digest
736+}
737+func (s *store) setTag(name, tag, accept, digest string) error {
738+ if !validDigest(digest) {
739+ return errDigest
740+ }
741+ s.mu.Lock()
742+ defer s.mu.Unlock()
743+ manifest, ok := s.entries[kindManifest.key(digest[7:])]
744+ if !ok {
745+ return os.ErrNotExist
746+ }
747+ // Protect the target while making room for its metadata.
748+ manifest.pins++
749+ defer func() { manifest.pins-- }()
750+ key := tagKey(name, tag, accept)
751+ old, exists := s.entries[key]
752+ if exists {
753+ // Already charged: rewriting it needs no room.
754+ s.touchLocked(old)
755+ } else {
756+ s.evictLocked(charge(0))
757+ if charge(0) > s.max-s.total-s.pending {
758+ return errTooLarge
759+ }
760+ }
761+ f, err := os.CreateTemp(filepath.Join(s.dir, "tmp"), "tag-")
762+ if err != nil {
763+ return err
764+ }
765+ defer func() { _ = f.Close(); _ = os.Remove(f.Name()) }()
766+ if _, err = f.WriteString(digest); err != nil {
767+ return err
768+ }
769+ if err = f.Close(); err != nil {
770+ return err
771+ }
772+ if err = os.Rename(f.Name(), filepath.Join(s.dir, key)); err != nil {
773+ return err
774+ }
775+ if old, ok := s.entries[key]; ok {
776+ s.unindexTagLocked(key, old.digest)
777+ old.digest = digest
778+ old.used = s.now()
779+ s.indexTagLocked(key, digest)
780+ return nil
781+ }
782+ s.addLocked(&entry{key: key, kind: kindTag, size: charge(0), used: s.now(), digest: digest})
783+ return nil
784+}
A internal/registryproxy/store_test.go +327 -0
@@ -0,0 +1,327 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "fmt"
8+ "io"
9+ "net/http"
10+ "net/http/httptest"
11+ "os"
12+ "path/filepath"
13+ "strings"
14+ "testing"
15+ "time"
16+)
17+
18+// Reopening a store keeps its content, charges it again and clears the
19+// incomplete downloads in tmp/.
20+func TestStoreReopen(t *testing.T) {
21+ dir := t.TempDir()
22+ s := mustOpen(t, storeCfg(dir, 8*minCharge))
23+ put(t, s, "aaaa")
24+ putManifest(t, s, manifestBody, "application/vnd.oci.image.index.v1+json")
25+ if err := s.setTag("library/alpine", "3", normalizeAccept(""), sum(manifestBody)); err != nil {
26+ t.Fatal(err)
27+ }
28+ if err := os.WriteFile(filepath.Join(dir, "tmp", "leftover"), []byte("partial"), 0600); err != nil {
29+ t.Fatal(err)
30+ }
31+ total := s.total
32+ s = mustOpen(t, storeCfg(dir, 8*minCharge))
33+ f, _, err := s.open(kindBlob, sum("aaaa"))
34+ if err != nil {
35+ t.Fatal(err)
36+ }
37+ _ = f.Close()
38+ if got := s.tag("library/alpine", "3", normalizeAccept("")); got != sum(manifestBody) {
39+ t.Fatalf("tag after reopen: %q", got)
40+ }
41+ if ds, _ := os.ReadDir(filepath.Join(dir, "tmp")); len(ds) != 0 {
42+ t.Fatal("tmp not cleared")
43+ }
44+ if accounted(t, s); s.total != total {
45+ t.Fatalf("reopened total %d, was %d", s.total, total)
46+ }
47+}
48+
49+// Every entry is charged at least minCharge, on commit and on reopen, so tiny
50+// blobs cannot make more than Max/minCharge files.
51+func TestEntriesChargedAtLeastMinCharge(t *testing.T) {
52+ dir := t.TempDir()
53+ s := mustOpen(t, storeCfg(dir, 3*minCharge))
54+ for i := range 10 {
55+ put(t, s, fmt.Sprint("tiny-", i))
56+ }
57+ accounted(t, s)
58+ if n := len(s.entries); n > 3 {
59+ t.Fatalf("%d entries under a cap of 3 charges", n)
60+ }
61+ s = mustOpen(t, storeCfg(dir, 3*minCharge))
62+ accounted(t, s)
63+ if s.total != int64(len(s.entries))*minCharge {
64+ t.Fatalf("reopened total %d for %d entries", s.total, len(s.entries))
65+ }
66+}
67+
68+// A cap below one entry's minimum charge is refused; a blob larger than Max is
69+// refused and leaves nothing charged.
70+func TestStoreOversize(t *testing.T) {
71+ if _, err := openStore(storeCfg(t.TempDir(), minCharge-1)); err == nil {
72+ t.Fatal("cap below minCharge accepted")
73+ }
74+ s := mustOpen(t, storeCfg(t.TempDir(), minCharge))
75+ large := strings.Repeat("x", int(minCharge)+1)
76+ _, _, _, err := s.ensure(context.Background(), kindBlob, sum(large), func(_ context.Context, w io.Writer) (string, error) {
77+ _, err := io.WriteString(w, large)
78+ return "", err
79+ }, keepalive{})
80+ if err != errTooLarge {
81+ t.Fatal(err)
82+ }
83+ if s.pending != 0 || s.total != 0 {
84+ t.Fatalf("accounting: %d %d", s.pending, s.total)
85+ }
86+}
87+
88+// Eviction walks the LRU, evicts least recently used first, frees down to the
89+// low-water mark rather than the bare need, and skips pinned entries.
90+func TestEvictionLRUAndLowWater(t *testing.T) {
91+ now := time.Unix(1_700_000_000, 0)
92+ const n = 40
93+ c := storeCfg(t.TempDir(), n*minCharge) // low water: 5%, two entries
94+ c.now = func() time.Time { return now }
95+ s := mustOpen(t, c)
96+ // 40 blobs fill the store. Freeing only the need would evict one blob; the
97+ // low-water mark evicts two.
98+ blob := func(i int) string { return fmt.Sprintf("%03d", i) + strings.Repeat("x", 47) }
99+ for i := range n {
100+ now = now.Add(time.Second)
101+ put(t, s, blob(i))
102+ }
103+ // Touch 0 and 1, and keep 2 open: the least recently used are then 3, 4...
104+ for _, i := range []int{0, 1} {
105+ now = now.Add(time.Second)
106+ f, _, err := s.open(kindBlob, sum(blob(i)))
107+ if err != nil {
108+ t.Fatal(err)
109+ }
110+ _ = f.Close()
111+ }
112+ pinned, _, err := s.open(kindBlob, sum(blob(2)))
113+ if err != nil {
114+ t.Fatal(err)
115+ }
116+ defer func() { _ = pinned.Close() }()
117+ s.mu.Lock()
118+ s.lru.MoveToFront(s.entries[kindBlob.key(sum(blob(2))[7:])].elem) // oldest of all, but open
119+ s.mu.Unlock()
120+ now = now.Add(time.Second)
121+ put(t, s, "a") // over the cap: frees the low-water mark
122+ present := func(i int) bool {
123+ _, err := os.Stat(filepath.Join(s.dir, kindBlob.key(sum(blob(i))[7:])))
124+ return err == nil
125+ }
126+ for i := range n {
127+ want := i != 3 && i != 4 // 3 and 4 are the least recently used unpinned
128+ if present(i) != want {
129+ t.Errorf("blob %d present=%v", i, present(i))
130+ }
131+ }
132+ accounted(t, s)
133+ s.mu.Lock()
134+ free := s.max - s.total
135+ s.mu.Unlock()
136+ if free < min(lowWater, s.max/20)-minCharge {
137+ t.Fatalf("freed only to %d", free)
138+ }
139+}
140+
141+// Eviction restarts from the front after a manifest removal that fails, and
142+// does not loop on it.
143+func TestEvictionAfterFailedManifestRemoval(t *testing.T) {
144+ now := time.Unix(1_700_000_000, 0)
145+ c := storeCfg(t.TempDir(), 1<<20)
146+ c.now = func() time.Time { return now }
147+ s := mustOpen(t, c)
148+ m := `{"schemaVersion":2}`
149+ putManifest(t, s, m, "application/vnd.oci.image.index.v1+json")
150+ now = now.Add(time.Second)
151+ if err := s.setTag("library/alpine", "3", "", sum(m)); err != nil {
152+ t.Fatal(err)
153+ }
154+ now = now.Add(time.Second)
155+ put(t, s, "blob")
156+ // The manifest's .type cannot be removed: replace it with a non-empty
157+ // directory. Its tag goes first all the same.
158+ typ := filepath.Join(s.dir, kindManifest.key(sum(m)[7:])+".type")
159+ if err := os.Remove(typ); err != nil {
160+ t.Fatal(err)
161+ }
162+ if err := os.MkdirAll(filepath.Join(typ, "x"), 0700); err != nil {
163+ t.Fatal(err)
164+ }
165+ s.mu.Lock()
166+ s.evictLocked(s.max) // ask for everything
167+ _, manifestKept := s.entries[kindManifest.key(sum(m)[7:])]
168+ _, blobKept := s.entries[kindBlob.key(sum("blob")[7:])]
169+ tags := len(s.tagsOf)
170+ s.mu.Unlock()
171+ if !manifestKept || blobKept || tags != 0 {
172+ t.Fatalf("manifest kept=%v blob kept=%v tags=%d", manifestKept, blobKept, tags)
173+ }
174+ accounted(t, s)
175+}
176+
177+// The committed entry is pinned for its waiters, so an eviction between the
178+// commit and their open cannot remove it.
179+func TestCommittedEntryPinnedForWaiters(t *testing.T) {
180+ s := mustOpen(t, storeCfg(t.TempDir(), minCharge))
181+ key := kindBlob.key(sum("aaaa")[7:])
182+ s.hookCommitted = func(string) {
183+ // Ask for the whole store before the waiters are woken.
184+ s.mu.Lock()
185+ s.evictLocked(s.max)
186+ s.mu.Unlock()
187+ }
188+ f, _, _, err := s.ensure(context.Background(), kindBlob, sum("aaaa"), func(_ context.Context, w io.Writer) (string, error) {
189+ _, err := io.WriteString(w, "aaaa")
190+ return "", err
191+ }, keepalive{})
192+ if err != nil {
193+ t.Fatalf("waiter lost its entry: %v", err)
194+ }
195+ s.mu.Lock()
196+ pins := s.entries[key].pins
197+ s.mu.Unlock()
198+ if pins != 1 {
199+ t.Fatalf("pins %d: the commit pin was not released to the reader's", pins)
200+ }
201+ _ = f.Close()
202+ s.hookCommitted = nil
203+ s.mu.Lock()
204+ s.evictLocked(s.max)
205+ _, kept := s.entries[key]
206+ s.mu.Unlock()
207+ if kept {
208+ t.Fatal("unpinned entry not evictable")
209+ }
210+}
211+
212+// A blob that a response is still sending is pinned: a fill that needs its
213+// room fails, and succeeds once the response has closed the file.
214+func TestEvictionWhileServing(t *testing.T) {
215+ a, b := strings.Repeat("a", int(minCharge)), strings.Repeat("b", int(minCharge))
216+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
217+ switch {
218+ case strings.HasSuffix(r.URL.Path, sum(a)):
219+ _, _ = io.WriteString(w, a)
220+ case strings.HasSuffix(r.URL.Path, sum(b)):
221+ _, _ = io.WriteString(w, b)
222+ default:
223+ http.NotFound(w, r)
224+ }
225+ })
226+ p := proxyWith(t, srv, Config{Max: minCharge * 3 / 2})
227+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(a)); w.Code != http.StatusOK {
228+ t.Fatal(w.Code)
229+ }
230+ bw := &blockingWriter{ResponseRecorder: httptest.NewRecorder(), started: make(chan struct{}), release: make(chan struct{})}
231+ done := make(chan struct{})
232+ go func() {
233+ defer close(done)
234+ p.ServeHTTP(bw, httptest.NewRequest(http.MethodGet, "/v2/library/alpine/blobs/"+sum(a), nil))
235+ }()
236+ <-bw.started
237+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(b)); w.Code != http.StatusBadGateway {
238+ t.Fatalf("fill evicted a blob being served: %d", w.Code)
239+ }
240+ if _, err := os.Stat(filepath.Join(p.s.dir, kindBlob.key(sum(a)[7:]))); err != nil {
241+ t.Fatalf("served blob removed: %v", err)
242+ }
243+ accounted(t, p.s)
244+ close(bw.release)
245+ <-done
246+ if bw.Code != http.StatusOK || bw.Body.String() != a {
247+ t.Fatalf("served: %d", bw.Code)
248+ }
249+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(b)); w.Code != http.StatusOK || w.Body.String() != b {
250+ t.Fatalf("after release: %d", w.Code)
251+ }
252+ if _, err := os.Stat(filepath.Join(p.s.dir, kindBlob.key(sum(a)[7:]))); !os.IsNotExist(err) {
253+ t.Fatal("released blob not evicted")
254+ }
255+ accounted(t, p.s)
256+}
257+
258+// Tag files share the LRU of the store, so they cannot grow the store past
259+// Max; reopening charges them again.
260+func TestTagsChargedAndEvicted(t *testing.T) {
261+ now := time.Unix(1_700_000_000, 0)
262+ dir := t.TempDir()
263+ c := storeCfg(dir, 4*minCharge) // the manifest and 3 tags
264+ c.now = func() time.Time { return now }
265+ s := mustOpen(t, c)
266+ putManifest(t, s, manifestBody, "application/vnd.oci.image.index.v1+json")
267+ for i := range 20 {
268+ now = now.Add(time.Second)
269+ if err := s.setTag("library/alpine", fmt.Sprint("t", i), "", sum(manifestBody)); err != nil {
270+ t.Fatalf("tag %d: %v", i, err)
271+ }
272+ if n := accounted(t, s); n > 3 {
273+ t.Fatalf("%d tags over a cap of 3", n)
274+ }
275+ }
276+ if s.tag("library/alpine", "t19", "") != sum(manifestBody) || s.tag("library/alpine", "t0", "") != "" {
277+ t.Fatal("tags not evicted least recently used first")
278+ }
279+ s = mustOpen(t, storeCfg(dir, 4*minCharge))
280+ if n := accounted(t, s); n != 3 {
281+ t.Fatalf("reopened with %d tags", n)
282+ }
283+}
284+
285+// The tag index follows rewrites, and removing a manifest takes exactly its
286+// own tags along.
287+func TestTagIndex(t *testing.T) {
288+ s := mustOpen(t, storeCfg(t.TempDir(), 1<<20))
289+ manifests := []string{`{"schemaVersion":2,"a":1}`, `{"schemaVersion":2,"b":1}`}
290+ for _, m := range manifests {
291+ putManifest(t, s, m, "application/vnd.oci.image.index.v1+json")
292+ }
293+ for _, tag := range []string{"x", "y"} {
294+ if err := s.setTag("library/alpine", tag, "", sum(manifests[0])); err != nil {
295+ t.Fatal(err)
296+ }
297+ }
298+ // y moves to the second manifest.
299+ if err := s.setTag("library/alpine", "y", "", sum(manifests[1])); err != nil {
300+ t.Fatal(err)
301+ }
302+ s.mu.Lock()
303+ if !s.removeLocked(kindManifest.key(sum(manifests[0])[7:])) {
304+ t.Fatal("remove failed")
305+ }
306+ n0, n1 := len(s.tagsOf[sum(manifests[0])]), len(s.tagsOf[sum(manifests[1])])
307+ s.mu.Unlock()
308+ if s.tag("library/alpine", "x", "") != "" || s.tag("library/alpine", "y", "") != sum(manifests[1]) || n0 != 0 || n1 != 1 {
309+ t.Fatalf("index: %d %d", n0, n1)
310+ }
311+ accounted(t, s)
312+}
313+
314+// A .type file without its manifest is removed at startup.
315+func TestOrphanTypeRemoved(t *testing.T) {
316+ dir := t.TempDir()
317+ mustOpen(t, storeCfg(dir, 1<<20))
318+ orphan := filepath.Join(dir, kindManifest.key(sum("gone")[7:])+".type")
319+ if err := os.WriteFile(orphan, []byte("application/json"), 0600); err != nil {
320+ t.Fatal(err)
321+ }
322+ s := mustOpen(t, storeCfg(dir, 1<<20))
323+ if _, err := os.Stat(orphan); !os.IsNotExist(err) {
324+ t.Fatal("orphan .type kept")
325+ }
326+ accounted(t, s)
327+}
@@ -0,0 +1,327 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "fmt"
8+ "io"
9+ "net/http"
10+ "net/http/httptest"
11+ "os"
12+ "path/filepath"
13+ "strings"
14+ "testing"
15+ "time"
16+)
17+
18+// Reopening a store keeps its content, charges it again and clears the
19+// incomplete downloads in tmp/.
20+func TestStoreReopen(t *testing.T) {
21+ dir := t.TempDir()
22+ s := mustOpen(t, storeCfg(dir, 8*minCharge))
23+ put(t, s, "aaaa")
24+ putManifest(t, s, manifestBody, "application/vnd.oci.image.index.v1+json")
25+ if err := s.setTag("library/alpine", "3", normalizeAccept(""), sum(manifestBody)); err != nil {
26+ t.Fatal(err)
27+ }
28+ if err := os.WriteFile(filepath.Join(dir, "tmp", "leftover"), []byte("partial"), 0600); err != nil {
29+ t.Fatal(err)
30+ }
31+ total := s.total
32+ s = mustOpen(t, storeCfg(dir, 8*minCharge))
33+ f, _, err := s.open(kindBlob, sum("aaaa"))
34+ if err != nil {
35+ t.Fatal(err)
36+ }
37+ _ = f.Close()
38+ if got := s.tag("library/alpine", "3", normalizeAccept("")); got != sum(manifestBody) {
39+ t.Fatalf("tag after reopen: %q", got)
40+ }
41+ if ds, _ := os.ReadDir(filepath.Join(dir, "tmp")); len(ds) != 0 {
42+ t.Fatal("tmp not cleared")
43+ }
44+ if accounted(t, s); s.total != total {
45+ t.Fatalf("reopened total %d, was %d", s.total, total)
46+ }
47+}
48+
49+// Every entry is charged at least minCharge, on commit and on reopen, so tiny
50+// blobs cannot make more than Max/minCharge files.
51+func TestEntriesChargedAtLeastMinCharge(t *testing.T) {
52+ dir := t.TempDir()
53+ s := mustOpen(t, storeCfg(dir, 3*minCharge))
54+ for i := range 10 {
55+ put(t, s, fmt.Sprint("tiny-", i))
56+ }
57+ accounted(t, s)
58+ if n := len(s.entries); n > 3 {
59+ t.Fatalf("%d entries under a cap of 3 charges", n)
60+ }
61+ s = mustOpen(t, storeCfg(dir, 3*minCharge))
62+ accounted(t, s)
63+ if s.total != int64(len(s.entries))*minCharge {
64+ t.Fatalf("reopened total %d for %d entries", s.total, len(s.entries))
65+ }
66+}
67+
68+// A cap below one entry's minimum charge is refused; a blob larger than Max is
69+// refused and leaves nothing charged.
70+func TestStoreOversize(t *testing.T) {
71+ if _, err := openStore(storeCfg(t.TempDir(), minCharge-1)); err == nil {
72+ t.Fatal("cap below minCharge accepted")
73+ }
74+ s := mustOpen(t, storeCfg(t.TempDir(), minCharge))
75+ large := strings.Repeat("x", int(minCharge)+1)
76+ _, _, _, err := s.ensure(context.Background(), kindBlob, sum(large), func(_ context.Context, w io.Writer) (string, error) {
77+ _, err := io.WriteString(w, large)
78+ return "", err
79+ }, keepalive{})
80+ if err != errTooLarge {
81+ t.Fatal(err)
82+ }
83+ if s.pending != 0 || s.total != 0 {
84+ t.Fatalf("accounting: %d %d", s.pending, s.total)
85+ }
86+}
87+
88+// Eviction walks the LRU, evicts least recently used first, frees down to the
89+// low-water mark rather than the bare need, and skips pinned entries.
90+func TestEvictionLRUAndLowWater(t *testing.T) {
91+ now := time.Unix(1_700_000_000, 0)
92+ const n = 40
93+ c := storeCfg(t.TempDir(), n*minCharge) // low water: 5%, two entries
94+ c.now = func() time.Time { return now }
95+ s := mustOpen(t, c)
96+ // 40 blobs fill the store. Freeing only the need would evict one blob; the
97+ // low-water mark evicts two.
98+ blob := func(i int) string { return fmt.Sprintf("%03d", i) + strings.Repeat("x", 47) }
99+ for i := range n {
100+ now = now.Add(time.Second)
101+ put(t, s, blob(i))
102+ }
103+ // Touch 0 and 1, and keep 2 open: the least recently used are then 3, 4...
104+ for _, i := range []int{0, 1} {
105+ now = now.Add(time.Second)
106+ f, _, err := s.open(kindBlob, sum(blob(i)))
107+ if err != nil {
108+ t.Fatal(err)
109+ }
110+ _ = f.Close()
111+ }
112+ pinned, _, err := s.open(kindBlob, sum(blob(2)))
113+ if err != nil {
114+ t.Fatal(err)
115+ }
116+ defer func() { _ = pinned.Close() }()
117+ s.mu.Lock()
118+ s.lru.MoveToFront(s.entries[kindBlob.key(sum(blob(2))[7:])].elem) // oldest of all, but open
119+ s.mu.Unlock()
120+ now = now.Add(time.Second)
121+ put(t, s, "a") // over the cap: frees the low-water mark
122+ present := func(i int) bool {
123+ _, err := os.Stat(filepath.Join(s.dir, kindBlob.key(sum(blob(i))[7:])))
124+ return err == nil
125+ }
126+ for i := range n {
127+ want := i != 3 && i != 4 // 3 and 4 are the least recently used unpinned
128+ if present(i) != want {
129+ t.Errorf("blob %d present=%v", i, present(i))
130+ }
131+ }
132+ accounted(t, s)
133+ s.mu.Lock()
134+ free := s.max - s.total
135+ s.mu.Unlock()
136+ if free < min(lowWater, s.max/20)-minCharge {
137+ t.Fatalf("freed only to %d", free)
138+ }
139+}
140+
141+// Eviction restarts from the front after a manifest removal that fails, and
142+// does not loop on it.
143+func TestEvictionAfterFailedManifestRemoval(t *testing.T) {
144+ now := time.Unix(1_700_000_000, 0)
145+ c := storeCfg(t.TempDir(), 1<<20)
146+ c.now = func() time.Time { return now }
147+ s := mustOpen(t, c)
148+ m := `{"schemaVersion":2}`
149+ putManifest(t, s, m, "application/vnd.oci.image.index.v1+json")
150+ now = now.Add(time.Second)
151+ if err := s.setTag("library/alpine", "3", "", sum(m)); err != nil {
152+ t.Fatal(err)
153+ }
154+ now = now.Add(time.Second)
155+ put(t, s, "blob")
156+ // The manifest's .type cannot be removed: replace it with a non-empty
157+ // directory. Its tag goes first all the same.
158+ typ := filepath.Join(s.dir, kindManifest.key(sum(m)[7:])+".type")
159+ if err := os.Remove(typ); err != nil {
160+ t.Fatal(err)
161+ }
162+ if err := os.MkdirAll(filepath.Join(typ, "x"), 0700); err != nil {
163+ t.Fatal(err)
164+ }
165+ s.mu.Lock()
166+ s.evictLocked(s.max) // ask for everything
167+ _, manifestKept := s.entries[kindManifest.key(sum(m)[7:])]
168+ _, blobKept := s.entries[kindBlob.key(sum("blob")[7:])]
169+ tags := len(s.tagsOf)
170+ s.mu.Unlock()
171+ if !manifestKept || blobKept || tags != 0 {
172+ t.Fatalf("manifest kept=%v blob kept=%v tags=%d", manifestKept, blobKept, tags)
173+ }
174+ accounted(t, s)
175+}
176+
177+// The committed entry is pinned for its waiters, so an eviction between the
178+// commit and their open cannot remove it.
179+func TestCommittedEntryPinnedForWaiters(t *testing.T) {
180+ s := mustOpen(t, storeCfg(t.TempDir(), minCharge))
181+ key := kindBlob.key(sum("aaaa")[7:])
182+ s.hookCommitted = func(string) {
183+ // Ask for the whole store before the waiters are woken.
184+ s.mu.Lock()
185+ s.evictLocked(s.max)
186+ s.mu.Unlock()
187+ }
188+ f, _, _, err := s.ensure(context.Background(), kindBlob, sum("aaaa"), func(_ context.Context, w io.Writer) (string, error) {
189+ _, err := io.WriteString(w, "aaaa")
190+ return "", err
191+ }, keepalive{})
192+ if err != nil {
193+ t.Fatalf("waiter lost its entry: %v", err)
194+ }
195+ s.mu.Lock()
196+ pins := s.entries[key].pins
197+ s.mu.Unlock()
198+ if pins != 1 {
199+ t.Fatalf("pins %d: the commit pin was not released to the reader's", pins)
200+ }
201+ _ = f.Close()
202+ s.hookCommitted = nil
203+ s.mu.Lock()
204+ s.evictLocked(s.max)
205+ _, kept := s.entries[key]
206+ s.mu.Unlock()
207+ if kept {
208+ t.Fatal("unpinned entry not evictable")
209+ }
210+}
211+
212+// A blob that a response is still sending is pinned: a fill that needs its
213+// room fails, and succeeds once the response has closed the file.
214+func TestEvictionWhileServing(t *testing.T) {
215+ a, b := strings.Repeat("a", int(minCharge)), strings.Repeat("b", int(minCharge))
216+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
217+ switch {
218+ case strings.HasSuffix(r.URL.Path, sum(a)):
219+ _, _ = io.WriteString(w, a)
220+ case strings.HasSuffix(r.URL.Path, sum(b)):
221+ _, _ = io.WriteString(w, b)
222+ default:
223+ http.NotFound(w, r)
224+ }
225+ })
226+ p := proxyWith(t, srv, Config{Max: minCharge * 3 / 2})
227+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(a)); w.Code != http.StatusOK {
228+ t.Fatal(w.Code)
229+ }
230+ bw := &blockingWriter{ResponseRecorder: httptest.NewRecorder(), started: make(chan struct{}), release: make(chan struct{})}
231+ done := make(chan struct{})
232+ go func() {
233+ defer close(done)
234+ p.ServeHTTP(bw, httptest.NewRequest(http.MethodGet, "/v2/library/alpine/blobs/"+sum(a), nil))
235+ }()
236+ <-bw.started
237+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(b)); w.Code != http.StatusBadGateway {
238+ t.Fatalf("fill evicted a blob being served: %d", w.Code)
239+ }
240+ if _, err := os.Stat(filepath.Join(p.s.dir, kindBlob.key(sum(a)[7:]))); err != nil {
241+ t.Fatalf("served blob removed: %v", err)
242+ }
243+ accounted(t, p.s)
244+ close(bw.release)
245+ <-done
246+ if bw.Code != http.StatusOK || bw.Body.String() != a {
247+ t.Fatalf("served: %d", bw.Code)
248+ }
249+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(b)); w.Code != http.StatusOK || w.Body.String() != b {
250+ t.Fatalf("after release: %d", w.Code)
251+ }
252+ if _, err := os.Stat(filepath.Join(p.s.dir, kindBlob.key(sum(a)[7:]))); !os.IsNotExist(err) {
253+ t.Fatal("released blob not evicted")
254+ }
255+ accounted(t, p.s)
256+}
257+
258+// Tag files share the LRU of the store, so they cannot grow the store past
259+// Max; reopening charges them again.
260+func TestTagsChargedAndEvicted(t *testing.T) {
261+ now := time.Unix(1_700_000_000, 0)
262+ dir := t.TempDir()
263+ c := storeCfg(dir, 4*minCharge) // the manifest and 3 tags
264+ c.now = func() time.Time { return now }
265+ s := mustOpen(t, c)
266+ putManifest(t, s, manifestBody, "application/vnd.oci.image.index.v1+json")
267+ for i := range 20 {
268+ now = now.Add(time.Second)
269+ if err := s.setTag("library/alpine", fmt.Sprint("t", i), "", sum(manifestBody)); err != nil {
270+ t.Fatalf("tag %d: %v", i, err)
271+ }
272+ if n := accounted(t, s); n > 3 {
273+ t.Fatalf("%d tags over a cap of 3", n)
274+ }
275+ }
276+ if s.tag("library/alpine", "t19", "") != sum(manifestBody) || s.tag("library/alpine", "t0", "") != "" {
277+ t.Fatal("tags not evicted least recently used first")
278+ }
279+ s = mustOpen(t, storeCfg(dir, 4*minCharge))
280+ if n := accounted(t, s); n != 3 {
281+ t.Fatalf("reopened with %d tags", n)
282+ }
283+}
284+
285+// The tag index follows rewrites, and removing a manifest takes exactly its
286+// own tags along.
287+func TestTagIndex(t *testing.T) {
288+ s := mustOpen(t, storeCfg(t.TempDir(), 1<<20))
289+ manifests := []string{`{"schemaVersion":2,"a":1}`, `{"schemaVersion":2,"b":1}`}
290+ for _, m := range manifests {
291+ putManifest(t, s, m, "application/vnd.oci.image.index.v1+json")
292+ }
293+ for _, tag := range []string{"x", "y"} {
294+ if err := s.setTag("library/alpine", tag, "", sum(manifests[0])); err != nil {
295+ t.Fatal(err)
296+ }
297+ }
298+ // y moves to the second manifest.
299+ if err := s.setTag("library/alpine", "y", "", sum(manifests[1])); err != nil {
300+ t.Fatal(err)
301+ }
302+ s.mu.Lock()
303+ if !s.removeLocked(kindManifest.key(sum(manifests[0])[7:])) {
304+ t.Fatal("remove failed")
305+ }
306+ n0, n1 := len(s.tagsOf[sum(manifests[0])]), len(s.tagsOf[sum(manifests[1])])
307+ s.mu.Unlock()
308+ if s.tag("library/alpine", "x", "") != "" || s.tag("library/alpine", "y", "") != sum(manifests[1]) || n0 != 0 || n1 != 1 {
309+ t.Fatalf("index: %d %d", n0, n1)
310+ }
311+ accounted(t, s)
312+}
313+
314+// A .type file without its manifest is removed at startup.
315+func TestOrphanTypeRemoved(t *testing.T) {
316+ dir := t.TempDir()
317+ mustOpen(t, storeCfg(dir, 1<<20))
318+ orphan := filepath.Join(dir, kindManifest.key(sum("gone")[7:])+".type")
319+ if err := os.WriteFile(orphan, []byte("application/json"), 0600); err != nil {
320+ t.Fatal(err)
321+ }
322+ s := mustOpen(t, storeCfg(dir, 1<<20))
323+ if _, err := os.Stat(orphan); !os.IsNotExist(err) {
324+ t.Fatal("orphan .type kept")
325+ }
326+ accounted(t, s)
327+}
A internal/registryproxy/upstream.go +169 -0
@@ -0,0 +1,169 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "encoding/json"
8+ "errors"
9+ "fmt"
10+ "io"
11+ "net"
12+ "net/http"
13+ "net/url"
14+ "sync"
15+ "time"
16+)
17+
18+type upstreamError struct{ status int }
19+
20+func (e upstreamError) Error() string { return fmt.Sprintf("registry upstream status %d", e.status) }
21+
22+// unavailable reports an upstream that cannot answer now (down, timing out,
23+// failing or rate limiting): a known tag may then be served stale.
24+func unavailable(err error) bool {
25+ var e upstreamError
26+ if errors.As(err, &e) {
27+ return e.status >= http.StatusInternalServerError || e.status == http.StatusTooManyRequests
28+ }
29+ var network net.Error
30+ return errors.As(err, &network) || errors.Is(err, context.DeadlineExceeded)
31+}
32+
33+func proxyStatus(err error) int {
34+ var e upstreamError
35+ if errors.As(err, &e) && (e.status == http.StatusUnauthorized || e.status == http.StatusForbidden || e.status == http.StatusNotFound) {
36+ return http.StatusNotFound
37+ }
38+ return http.StatusBadGateway
39+}
40+
41+type bearer struct {
42+ value string
43+ until time.Time
44+}
45+
46+type upstream struct {
47+ base, auth string
48+ client *http.Client
49+ blobClient *http.Client
50+ requestTimeout, blobIdle time.Duration
51+ mu sync.Mutex
52+ tokens map[string]bearer
53+}
54+
55+func (u *upstream) token(ctx context.Context, name string, refresh bool) (string, error) {
56+ // Do not hold the token map lock over network I/O: an unavailable auth
57+ // endpoint must not queue unrelated repositories behind its timeout.
58+ // No guest credentials or challenge-controlled URLs are accepted.
59+ u.mu.Lock()
60+ t, ok := u.tokens[name]
61+ u.mu.Unlock()
62+ if ok && !refresh && time.Now().Before(t.until) {
63+ return t.value, nil
64+ }
65+ a, err := url.Parse(u.auth)
66+ if err != nil {
67+ return "", err
68+ }
69+ q := a.Query()
70+ q.Set("service", "registry.docker.io")
71+ q.Set("scope", "repository:"+name+":pull")
72+ a.RawQuery = q.Encode()
73+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.String(), nil)
74+ if err != nil {
75+ return "", err
76+ }
77+ resp, err := u.client.Do(req)
78+ if err != nil {
79+ return "", err
80+ }
81+ defer func() { _ = resp.Body.Close() }()
82+ if resp.StatusCode != http.StatusOK {
83+ return "", upstreamError{resp.StatusCode}
84+ }
85+ var data struct {
86+ Token string `json:"token"`
87+ AccessToken string `json:"access_token"`
88+ Expires int64 `json:"expires_in"`
89+ }
90+ b, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20+1))
91+ if err != nil {
92+ return "", err
93+ }
94+ if len(b) > 1<<20 {
95+ return "", errors.New("registry token response too large")
96+ }
97+ if err = json.Unmarshal(b, &data); err != nil {
98+ return "", err
99+ }
100+ if data.Token == "" {
101+ data.Token = data.AccessToken
102+ }
103+ if data.Token == "" {
104+ return "", errors.New("empty registry token")
105+ }
106+ if data.Expires <= 0 {
107+ data.Expires = 60
108+ }
109+ if data.Expires > 86400 {
110+ data.Expires = 86400
111+ }
112+ u.mu.Lock()
113+ defer u.mu.Unlock()
114+ if len(u.tokens) >= 1024 {
115+ clear(u.tokens)
116+ }
117+ u.tokens[name] = bearer{data.Token, time.Now().Add(time.Duration(data.Expires) * time.Second)}
118+ return data.Token, nil
119+}
120+
121+// authorized calls send with a bearer token for name, and once more with a
122+// fresh token after a 401 or 403. Each token request is bounded by
123+// RequestTimeout (and ctx). It returns a 200 response; any other closes the
124+// body and is an upstreamError.
125+func (u *upstream) authorized(ctx context.Context, name string, send func(tok string) (*http.Response, error)) (*http.Response, error) {
126+ for attempt := 0; attempt < 2; attempt++ {
127+ authCtx, cancel := context.WithTimeout(ctx, u.requestTimeout)
128+ tok, err := u.token(authCtx, name, attempt > 0)
129+ cancel()
130+ if err != nil {
131+ return nil, err
132+ }
133+ resp, err := send(tok)
134+ if err != nil {
135+ return nil, err
136+ }
137+ if resp.StatusCode == http.StatusOK {
138+ return resp, nil
139+ }
140+ _ = resp.Body.Close()
141+ if (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) && attempt == 0 {
142+ continue
143+ }
144+ return nil, upstreamError{resp.StatusCode}
145+ }
146+ return nil, upstreamError{http.StatusUnauthorized}
147+}
148+
149+// request is a metadata request (manifests, blob HEAD) bounded by ctx. Manifests
150+// use the client that never follows redirects; a blob HEAD uses the blob client,
151+// because Docker Hub answers it with a redirect to its CDN.
152+func (u *upstream) request(ctx context.Context, method, name string, k kind, ref, accept string) (*http.Response, error) {
153+ client := u.client
154+ if k == kindBlob {
155+ client = u.blobClient
156+ }
157+ return u.authorized(ctx, name, func(tok string) (*http.Response, error) {
158+ req, err := http.NewRequestWithContext(ctx, method, u.base+"/v2/"+name+"/"+k.dir()+"/"+ref, nil)
159+ if err != nil {
160+ return nil, err
161+ }
162+ req.Header.Set("Authorization", "Bearer "+tok)
163+ if accept != "" {
164+ req.Header.Set("Accept", accept)
165+ }
166+ // Do not forward guest Authorization, cookies, query parameters or Range.
167+ return client.Do(req)
168+ })
169+}
@@ -0,0 +1,169 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "context"
7+ "encoding/json"
8+ "errors"
9+ "fmt"
10+ "io"
11+ "net"
12+ "net/http"
13+ "net/url"
14+ "sync"
15+ "time"
16+)
17+
18+type upstreamError struct{ status int }
19+
20+func (e upstreamError) Error() string { return fmt.Sprintf("registry upstream status %d", e.status) }
21+
22+// unavailable reports an upstream that cannot answer now (down, timing out,
23+// failing or rate limiting): a known tag may then be served stale.
24+func unavailable(err error) bool {
25+ var e upstreamError
26+ if errors.As(err, &e) {
27+ return e.status >= http.StatusInternalServerError || e.status == http.StatusTooManyRequests
28+ }
29+ var network net.Error
30+ return errors.As(err, &network) || errors.Is(err, context.DeadlineExceeded)
31+}
32+
33+func proxyStatus(err error) int {
34+ var e upstreamError
35+ if errors.As(err, &e) && (e.status == http.StatusUnauthorized || e.status == http.StatusForbidden || e.status == http.StatusNotFound) {
36+ return http.StatusNotFound
37+ }
38+ return http.StatusBadGateway
39+}
40+
41+type bearer struct {
42+ value string
43+ until time.Time
44+}
45+
46+type upstream struct {
47+ base, auth string
48+ client *http.Client
49+ blobClient *http.Client
50+ requestTimeout, blobIdle time.Duration
51+ mu sync.Mutex
52+ tokens map[string]bearer
53+}
54+
55+func (u *upstream) token(ctx context.Context, name string, refresh bool) (string, error) {
56+ // Do not hold the token map lock over network I/O: an unavailable auth
57+ // endpoint must not queue unrelated repositories behind its timeout.
58+ // No guest credentials or challenge-controlled URLs are accepted.
59+ u.mu.Lock()
60+ t, ok := u.tokens[name]
61+ u.mu.Unlock()
62+ if ok && !refresh && time.Now().Before(t.until) {
63+ return t.value, nil
64+ }
65+ a, err := url.Parse(u.auth)
66+ if err != nil {
67+ return "", err
68+ }
69+ q := a.Query()
70+ q.Set("service", "registry.docker.io")
71+ q.Set("scope", "repository:"+name+":pull")
72+ a.RawQuery = q.Encode()
73+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.String(), nil)
74+ if err != nil {
75+ return "", err
76+ }
77+ resp, err := u.client.Do(req)
78+ if err != nil {
79+ return "", err
80+ }
81+ defer func() { _ = resp.Body.Close() }()
82+ if resp.StatusCode != http.StatusOK {
83+ return "", upstreamError{resp.StatusCode}
84+ }
85+ var data struct {
86+ Token string `json:"token"`
87+ AccessToken string `json:"access_token"`
88+ Expires int64 `json:"expires_in"`
89+ }
90+ b, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20+1))
91+ if err != nil {
92+ return "", err
93+ }
94+ if len(b) > 1<<20 {
95+ return "", errors.New("registry token response too large")
96+ }
97+ if err = json.Unmarshal(b, &data); err != nil {
98+ return "", err
99+ }
100+ if data.Token == "" {
101+ data.Token = data.AccessToken
102+ }
103+ if data.Token == "" {
104+ return "", errors.New("empty registry token")
105+ }
106+ if data.Expires <= 0 {
107+ data.Expires = 60
108+ }
109+ if data.Expires > 86400 {
110+ data.Expires = 86400
111+ }
112+ u.mu.Lock()
113+ defer u.mu.Unlock()
114+ if len(u.tokens) >= 1024 {
115+ clear(u.tokens)
116+ }
117+ u.tokens[name] = bearer{data.Token, time.Now().Add(time.Duration(data.Expires) * time.Second)}
118+ return data.Token, nil
119+}
120+
121+// authorized calls send with a bearer token for name, and once more with a
122+// fresh token after a 401 or 403. Each token request is bounded by
123+// RequestTimeout (and ctx). It returns a 200 response; any other closes the
124+// body and is an upstreamError.
125+func (u *upstream) authorized(ctx context.Context, name string, send func(tok string) (*http.Response, error)) (*http.Response, error) {
126+ for attempt := 0; attempt < 2; attempt++ {
127+ authCtx, cancel := context.WithTimeout(ctx, u.requestTimeout)
128+ tok, err := u.token(authCtx, name, attempt > 0)
129+ cancel()
130+ if err != nil {
131+ return nil, err
132+ }
133+ resp, err := send(tok)
134+ if err != nil {
135+ return nil, err
136+ }
137+ if resp.StatusCode == http.StatusOK {
138+ return resp, nil
139+ }
140+ _ = resp.Body.Close()
141+ if (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) && attempt == 0 {
142+ continue
143+ }
144+ return nil, upstreamError{resp.StatusCode}
145+ }
146+ return nil, upstreamError{http.StatusUnauthorized}
147+}
148+
149+// request is a metadata request (manifests, blob HEAD) bounded by ctx. Manifests
150+// use the client that never follows redirects; a blob HEAD uses the blob client,
151+// because Docker Hub answers it with a redirect to its CDN.
152+func (u *upstream) request(ctx context.Context, method, name string, k kind, ref, accept string) (*http.Response, error) {
153+ client := u.client
154+ if k == kindBlob {
155+ client = u.blobClient
156+ }
157+ return u.authorized(ctx, name, func(tok string) (*http.Response, error) {
158+ req, err := http.NewRequestWithContext(ctx, method, u.base+"/v2/"+name+"/"+k.dir()+"/"+ref, nil)
159+ if err != nil {
160+ return nil, err
161+ }
162+ req.Header.Set("Authorization", "Bearer "+tok)
163+ if accept != "" {
164+ req.Header.Set("Accept", accept)
165+ }
166+ // Do not forward guest Authorization, cookies, query parameters or Range.
167+ return client.Do(req)
168+ })
169+}
A internal/registryproxy/upstream_test.go +172 -0
@@ -0,0 +1,172 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "fmt"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "os"
11+ "path/filepath"
12+ "strings"
13+ "sync/atomic"
14+ "testing"
15+ "time"
16+)
17+
18+// A blob body has no total deadline: a transfer that keeps moving may last
19+// far longer than RequestTimeout and BlobIdleTimeout.
20+func TestBlobTransferOutlivesRequestTimeout(t *testing.T) {
21+ body := strings.Repeat("layer", 4000)
22+ srv := upstreamWith(t, slowBlob(body, 20, 25*time.Millisecond))
23+ p := proxyWith(t, srv, Config{RequestTimeout: 100 * time.Millisecond, BlobIdleTimeout: 150 * time.Millisecond})
24+ start := time.Now()
25+ w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(body))
26+ if w.Code != http.StatusOK || w.Body.String() != body {
27+ t.Fatalf("blob: %d", w.Code)
28+ }
29+ if d := time.Since(start); d < 300*time.Millisecond {
30+ t.Fatalf("transfer took %v, not longer than the timeouts", d)
31+ }
32+}
33+
34+// A blob body that stops moving is cut after BlobIdleTimeout, and auth,
35+// manifests and HEAD keep their total RequestTimeout.
36+func TestStalledUpstreamTimesOut(t *testing.T) {
37+ body := strings.Repeat("layer", 4000)
38+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
39+ if strings.Contains(r.URL.Path, "/blobs/") && r.Method == http.MethodGet {
40+ w.Header().Set("Content-Length", fmt.Sprint(len(body)))
41+ _, _ = io.WriteString(w, body[:100])
42+ w.(http.Flusher).Flush()
43+ }
44+ select { // a stalled body, or headers that never come
45+ case <-r.Context().Done():
46+ case <-time.After(5 * time.Second):
47+ }
48+ })
49+ p := proxyWith(t, srv, Config{RequestTimeout: 100 * time.Millisecond, BlobIdleTimeout: 150 * time.Millisecond})
50+ for _, tc := range []struct{ method, path string }{
51+ {http.MethodGet, "/v2/library/alpine/blobs/" + sum(body)},
52+ {http.MethodHead, "/v2/library/alpine/blobs/" + sum(body)},
53+ {http.MethodGet, "/v2/library/alpine/manifests/3"},
54+ } {
55+ start := time.Now()
56+ if w := request(t, p, tc.method, tc.path); w.Code != http.StatusBadGateway {
57+ t.Errorf("%s %s: %d", tc.method, tc.path, w.Code)
58+ }
59+ if d := time.Since(start); d > 2*time.Second {
60+ t.Errorf("%s %s: cut after %v", tc.method, tc.path, d)
61+ }
62+ }
63+ if _, err := os.Stat(filepath.Join(p.s.dir, kindBlob.key(sum(body)[7:]))); !os.IsNotExist(err) {
64+ t.Fatal("partial blob stored")
65+ }
66+}
67+
68+// Auth and manifest requests never follow a redirect.
69+func TestMetadataRedirectsNotFollowed(t *testing.T) {
70+ var elsewhere atomic.Int32
71+ other := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
72+ elsewhere.Add(1)
73+ _, _ = io.WriteString(w, `{"token":"public"}`)
74+ }))
75+ t.Cleanup(other.Close)
76+ for _, redirectAuth := range []bool{true, false} {
77+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
78+ if r.URL.Path == "/token" && !redirectAuth {
79+ _, _ = io.WriteString(w, `{"token":"public"}`)
80+ return
81+ }
82+ http.Redirect(w, r, other.URL+r.URL.Path, http.StatusTemporaryRedirect)
83+ }))
84+ p := proxyWith(t, srv, Config{})
85+ for _, ref := range []string{"3", sum(manifestBody)} {
86+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/"+ref); w.Code != http.StatusBadGateway {
87+ t.Errorf("auth redirect %v, ref %s: %d", redirectAuth, ref, w.Code)
88+ }
89+ }
90+ srv.Close()
91+ }
92+ if elsewhere.Load() != 0 {
93+ t.Fatalf("%d redirects followed", elsewhere.Load())
94+ }
95+}
96+
97+// Blob requests, HEAD included, follow the CDN redirect without the bearer
98+// token.
99+func TestBlobRedirectStripsBearer(t *testing.T) {
100+ var cdnHits atomic.Int32
101+ cdn := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
102+ cdnHits.Add(1)
103+ if r.Header.Get("Authorization") != "" {
104+ t.Error("bearer sent to the CDN")
105+ }
106+ w.Header().Set("Content-Length", fmt.Sprint(len(blobBody)))
107+ if r.Method == http.MethodGet {
108+ _, _ = io.WriteString(w, blobBody)
109+ }
110+ }))
111+ t.Cleanup(cdn.Close)
112+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
113+ if r.Header.Get("Authorization") != "Bearer public" {
114+ t.Error("no bearer on the registry")
115+ }
116+ http.Redirect(w, r, cdn.URL+"/layer", http.StatusTemporaryRedirect)
117+ })
118+ p := proxyWith(t, srv, Config{})
119+ path := "/v2/library/alpine/blobs/" + sum(blobBody)
120+ if w := request(t, p, http.MethodHead, path); w.Code != http.StatusOK || w.Header().Get("Content-Length") != fmt.Sprint(len(blobBody)) {
121+ t.Fatalf("HEAD: %d %v", w.Code, w.Header())
122+ }
123+ if w := request(t, p, http.MethodGet, path); w.Code != http.StatusOK || w.Body.String() != blobBody {
124+ t.Fatalf("blob: %d", w.Code)
125+ }
126+ if cdnHits.Load() != 2 {
127+ t.Fatalf("CDN hits %d", cdnHits.Load())
128+ }
129+}
130+
131+// A 401 or 403 is retried once with a fresh token; a second one is a 404, so
132+// that dockerd falls back to Docker Hub with the job's credentials.
133+func TestUnauthorizedRetriedOnce(t *testing.T) {
134+ for _, path := range []string{"/v2/library/alpine/manifests/3", "/v2/library/alpine/blobs/" + sum(blobBody)} {
135+ f := fake(t)
136+ p := proxyFor(t, f)
137+ f.status.Store(http.StatusUnauthorized)
138+ if w := request(t, p, http.MethodGet, path); w.Code != http.StatusNotFound {
139+ t.Fatalf("%s: %d", path, w.Code)
140+ }
141+ if f.tokens.Load() != 2 {
142+ t.Fatalf("%s: %d token requests, want a fresh one", path, f.tokens.Load())
143+ }
144+ }
145+}
146+
147+// When upstream is down, fails or rate-limits (429), a known tag is served
148+// stale; an unknown one is an error.
149+func TestStaleTagWhenUpstreamUnavailable(t *testing.T) {
150+ for _, tc := range []struct {
151+ name string
152+ status int32 // 0: the server is closed
153+ }{{"rate limited", http.StatusTooManyRequests}, {"failing", http.StatusServiceUnavailable}, {"down", 0}} {
154+ t.Run(tc.name, func(t *testing.T) {
155+ f := fake(t)
156+ p := proxyFor(t, f)
157+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/3"); w.Code != http.StatusOK {
158+ t.Fatal(w.Code)
159+ }
160+ if tc.status == 0 {
161+ f.server.Close()
162+ }
163+ f.status.Store(tc.status)
164+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/3"); w.Code != http.StatusOK || w.Body.String() != manifestBody {
165+ t.Fatalf("stale: %d %s", w.Code, w.Body)
166+ }
167+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/unknown"); w.Code != http.StatusBadGateway {
168+ t.Fatalf("unknown: %d", w.Code)
169+ }
170+ })
171+ }
172+}
@@ -0,0 +1,172 @@
1+// SPDX-License-Identifier: MIT
2+
3+package registryproxy
4+
5+import (
6+ "fmt"
7+ "io"
8+ "net/http"
9+ "net/http/httptest"
10+ "os"
11+ "path/filepath"
12+ "strings"
13+ "sync/atomic"
14+ "testing"
15+ "time"
16+)
17+
18+// A blob body has no total deadline: a transfer that keeps moving may last
19+// far longer than RequestTimeout and BlobIdleTimeout.
20+func TestBlobTransferOutlivesRequestTimeout(t *testing.T) {
21+ body := strings.Repeat("layer", 4000)
22+ srv := upstreamWith(t, slowBlob(body, 20, 25*time.Millisecond))
23+ p := proxyWith(t, srv, Config{RequestTimeout: 100 * time.Millisecond, BlobIdleTimeout: 150 * time.Millisecond})
24+ start := time.Now()
25+ w := request(t, p, http.MethodGet, "/v2/library/alpine/blobs/"+sum(body))
26+ if w.Code != http.StatusOK || w.Body.String() != body {
27+ t.Fatalf("blob: %d", w.Code)
28+ }
29+ if d := time.Since(start); d < 300*time.Millisecond {
30+ t.Fatalf("transfer took %v, not longer than the timeouts", d)
31+ }
32+}
33+
34+// A blob body that stops moving is cut after BlobIdleTimeout, and auth,
35+// manifests and HEAD keep their total RequestTimeout.
36+func TestStalledUpstreamTimesOut(t *testing.T) {
37+ body := strings.Repeat("layer", 4000)
38+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
39+ if strings.Contains(r.URL.Path, "/blobs/") && r.Method == http.MethodGet {
40+ w.Header().Set("Content-Length", fmt.Sprint(len(body)))
41+ _, _ = io.WriteString(w, body[:100])
42+ w.(http.Flusher).Flush()
43+ }
44+ select { // a stalled body, or headers that never come
45+ case <-r.Context().Done():
46+ case <-time.After(5 * time.Second):
47+ }
48+ })
49+ p := proxyWith(t, srv, Config{RequestTimeout: 100 * time.Millisecond, BlobIdleTimeout: 150 * time.Millisecond})
50+ for _, tc := range []struct{ method, path string }{
51+ {http.MethodGet, "/v2/library/alpine/blobs/" + sum(body)},
52+ {http.MethodHead, "/v2/library/alpine/blobs/" + sum(body)},
53+ {http.MethodGet, "/v2/library/alpine/manifests/3"},
54+ } {
55+ start := time.Now()
56+ if w := request(t, p, tc.method, tc.path); w.Code != http.StatusBadGateway {
57+ t.Errorf("%s %s: %d", tc.method, tc.path, w.Code)
58+ }
59+ if d := time.Since(start); d > 2*time.Second {
60+ t.Errorf("%s %s: cut after %v", tc.method, tc.path, d)
61+ }
62+ }
63+ if _, err := os.Stat(filepath.Join(p.s.dir, kindBlob.key(sum(body)[7:]))); !os.IsNotExist(err) {
64+ t.Fatal("partial blob stored")
65+ }
66+}
67+
68+// Auth and manifest requests never follow a redirect.
69+func TestMetadataRedirectsNotFollowed(t *testing.T) {
70+ var elsewhere atomic.Int32
71+ other := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
72+ elsewhere.Add(1)
73+ _, _ = io.WriteString(w, `{"token":"public"}`)
74+ }))
75+ t.Cleanup(other.Close)
76+ for _, redirectAuth := range []bool{true, false} {
77+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
78+ if r.URL.Path == "/token" && !redirectAuth {
79+ _, _ = io.WriteString(w, `{"token":"public"}`)
80+ return
81+ }
82+ http.Redirect(w, r, other.URL+r.URL.Path, http.StatusTemporaryRedirect)
83+ }))
84+ p := proxyWith(t, srv, Config{})
85+ for _, ref := range []string{"3", sum(manifestBody)} {
86+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/"+ref); w.Code != http.StatusBadGateway {
87+ t.Errorf("auth redirect %v, ref %s: %d", redirectAuth, ref, w.Code)
88+ }
89+ }
90+ srv.Close()
91+ }
92+ if elsewhere.Load() != 0 {
93+ t.Fatalf("%d redirects followed", elsewhere.Load())
94+ }
95+}
96+
97+// Blob requests, HEAD included, follow the CDN redirect without the bearer
98+// token.
99+func TestBlobRedirectStripsBearer(t *testing.T) {
100+ var cdnHits atomic.Int32
101+ cdn := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
102+ cdnHits.Add(1)
103+ if r.Header.Get("Authorization") != "" {
104+ t.Error("bearer sent to the CDN")
105+ }
106+ w.Header().Set("Content-Length", fmt.Sprint(len(blobBody)))
107+ if r.Method == http.MethodGet {
108+ _, _ = io.WriteString(w, blobBody)
109+ }
110+ }))
111+ t.Cleanup(cdn.Close)
112+ srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
113+ if r.Header.Get("Authorization") != "Bearer public" {
114+ t.Error("no bearer on the registry")
115+ }
116+ http.Redirect(w, r, cdn.URL+"/layer", http.StatusTemporaryRedirect)
117+ })
118+ p := proxyWith(t, srv, Config{})
119+ path := "/v2/library/alpine/blobs/" + sum(blobBody)
120+ if w := request(t, p, http.MethodHead, path); w.Code != http.StatusOK || w.Header().Get("Content-Length") != fmt.Sprint(len(blobBody)) {
121+ t.Fatalf("HEAD: %d %v", w.Code, w.Header())
122+ }
123+ if w := request(t, p, http.MethodGet, path); w.Code != http.StatusOK || w.Body.String() != blobBody {
124+ t.Fatalf("blob: %d", w.Code)
125+ }
126+ if cdnHits.Load() != 2 {
127+ t.Fatalf("CDN hits %d", cdnHits.Load())
128+ }
129+}
130+
131+// A 401 or 403 is retried once with a fresh token; a second one is a 404, so
132+// that dockerd falls back to Docker Hub with the job's credentials.
133+func TestUnauthorizedRetriedOnce(t *testing.T) {
134+ for _, path := range []string{"/v2/library/alpine/manifests/3", "/v2/library/alpine/blobs/" + sum(blobBody)} {
135+ f := fake(t)
136+ p := proxyFor(t, f)
137+ f.status.Store(http.StatusUnauthorized)
138+ if w := request(t, p, http.MethodGet, path); w.Code != http.StatusNotFound {
139+ t.Fatalf("%s: %d", path, w.Code)
140+ }
141+ if f.tokens.Load() != 2 {
142+ t.Fatalf("%s: %d token requests, want a fresh one", path, f.tokens.Load())
143+ }
144+ }
145+}
146+
147+// When upstream is down, fails or rate-limits (429), a known tag is served
148+// stale; an unknown one is an error.
149+func TestStaleTagWhenUpstreamUnavailable(t *testing.T) {
150+ for _, tc := range []struct {
151+ name string
152+ status int32 // 0: the server is closed
153+ }{{"rate limited", http.StatusTooManyRequests}, {"failing", http.StatusServiceUnavailable}, {"down", 0}} {
154+ t.Run(tc.name, func(t *testing.T) {
155+ f := fake(t)
156+ p := proxyFor(t, f)
157+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/3"); w.Code != http.StatusOK {
158+ t.Fatal(w.Code)
159+ }
160+ if tc.status == 0 {
161+ f.server.Close()
162+ }
163+ f.status.Store(tc.status)
164+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/3"); w.Code != http.StatusOK || w.Body.String() != manifestBody {
165+ t.Fatalf("stale: %d %s", w.Code, w.Body)
166+ }
167+ if w := request(t, p, http.MethodGet, "/v2/library/alpine/manifests/unknown"); w.Code != http.StatusBadGateway {
168+ t.Fatalf("unknown: %d", w.Code)
169+ }
170+ })
171+ }
172+}
M internal/sandbox/container/container_test.go +11 -0
@@ -349,6 +349,17 @@ func TestStartRunStop(t *testing.T) {
349349 if err != nil {
350350 t.Fatal(err)
351351 }
352+ timed := sb.(sandbox.Timed)
353+ timings := timed.StartTimings()
354+ for _, key := range []string{"socket_ms", "cache_ms", "scratch_ms", "net_ms", "container_ms", "connect_ms"} {
355+ if value, ok := timings[key]; !ok || value < 0 {
356+ t.Errorf("StartTimings[%q] = %d, present=%v", key, value, ok)
357+ }
358+ }
359+ timings["socket_ms"] = -1
360+ if timed.StartTimings()["socket_ms"] < 0 {
361+ t.Error("StartTimings returned shared storage")
362+ }
352363 out, code := stub(t, conn)
353364 if out != "hello from fake container\n" || code != 7 {
354365 t.Fatalf("out=%q code=%d", out, code)
@@ -349,6 +349,17 @@ func TestStartRunStop(t *testing.T) {
349 if err != nil {349 if err != nil {
350 t.Fatal(err)350 t.Fatal(err)
351 }351 }
352+ timed := sb.(sandbox.Timed)
353+ timings := timed.StartTimings()
354+ for _, key := range []string{"socket_ms", "cache_ms", "scratch_ms", "net_ms", "container_ms", "connect_ms"} {
355+ if value, ok := timings[key]; !ok || value < 0 {
356+ t.Errorf("StartTimings[%q] = %d, present=%v", key, value, ok)
357+ }
358+ }
359+ timings["socket_ms"] = -1
360+ if timed.StartTimings()["socket_ms"] < 0 {
361+ t.Error("StartTimings returned shared storage")
362+ }
352 out, code := stub(t, conn)363 out, code := stub(t, conn)
353 if out != "hello from fake container\n" || code != 7 {364 if out != "hello from fake container\n" || code != 7 {
354 t.Fatalf("out=%q code=%d", out, code)365 t.Fatalf("out=%q code=%d", out, code)
M internal/sandbox/container/driver.go +34 -6
@@ -22,6 +22,7 @@ import (
2222 "context"
2323 "errors"
2424 "fmt"
25+ "maps"
2526 "net"
2627 "net/netip"
2728 "os"
@@ -285,10 +286,11 @@ const (
285286 )
286287
287288 type containerSandbox struct {
288- d *Driver
289- slot int
290- id string
291- gate sandbox.BootGate
289+ d *Driver
290+ slot int
291+ id string
292+ gate sandbox.BootGate
293+ timings map[string]int64
292294
293295 mu sync.Mutex
294296 started bool
@@ -305,6 +307,10 @@ type containerSandbox struct {
305307 attached bool
306308 }
307309
310+// StartTimings returns a copy of the phase durations of a successful Start.
311+// It is valid once Start has returned, which writes them.
312+func (s *containerSandbox) StartTimings() map[string]int64 { return maps.Clone(s.timings) }
313+
308314 // ID implements sandbox.Sandbox.
309315 func (s *containerSandbox) ID() string { return s.id }
310316
@@ -334,8 +340,11 @@ func (s *containerSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (san
334340 if d.cfg.GuestForge != nil {
335341 log.Warn().Str("job", spec.JobID).Str("container", s.id).Msg(d.cfg.GuestForge.Warning())
336342 }
343+ start := time.Now()
344+ s.timings = make(map[string]int64)
337345 conn, err := s.run(bootCtx, spec)
338346 if err != nil {
347+ s.timings = nil
339348 var tail string
340349 if s.container != "" {
341350 tail, _ = d.eng.LogsTail(context.WithoutCancel(ctx), s.container, 40)
@@ -346,7 +355,11 @@ func (s *containerSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (san
346355 }
347356 return nil, errors.Join(err, cerr)
348357 }
349- log.Info().Str("container", s.container).Int("slot", s.slot).Str("job", spec.JobID).Bool("privileged", d.privileged).Msg("container running")
358+ fields := map[string]any{"total_ms": time.Since(start).Milliseconds()}
359+ for k, v := range s.timings {
360+ fields[k] = v
361+ }
362+ log.Info().Str("container", s.container).Int("slot", s.slot).Str("job", spec.JobID).Bool("privileged", d.privileged).Fields(fields).Msg("container running")
350363 return conn, nil
351364 }
352365
@@ -356,25 +369,40 @@ func (s *containerSandbox) run(ctx context.Context, spec sandbox.JobSpec) (sandb
356369 if err != nil {
357370 return nil, err
358371 }
372+ start := time.Now()
359373 accepted, err := s.runSocket()
360374 if err != nil {
361375 return nil, err
362376 }
377+ s.timings["socket_ms"] = time.Since(start).Milliseconds()
378+ start = time.Now()
363379 if err := s.runCache(spec); err != nil {
364380 return nil, err
365381 }
382+ s.timings["cache_ms"] = time.Since(start).Milliseconds()
383+ start = time.Now()
366384 if err := s.runScratch(ctx, spec); err != nil {
367385 return nil, err
368386 }
369387 labels := map[string]string{sandbox.LabelJob: spec.JobID}
388+ s.timings["scratch_ms"] = time.Since(start).Milliseconds()
389+ start = time.Now()
370390 if err := s.runNetwork(ctx, netw, labels); err != nil {
371391 return nil, err
372392 }
393+ s.timings["net_ms"] = time.Since(start).Milliseconds()
394+ start = time.Now()
373395 exited, err := s.runContainer(ctx, spec, netw, labels)
374396 if err != nil {
375397 return nil, err
376398 }
377- return s.waitExecutor(ctx, accepted, exited)
399+ s.timings["container_ms"] = time.Since(start).Milliseconds()
400+ start = time.Now()
401+ conn, err := s.waitExecutor(ctx, accepted, exited)
402+ if err == nil {
403+ s.timings["connect_ms"] = time.Since(start).Milliseconds()
404+ }
405+ return conn, err
378406 }
379407
380408 type acceptResult struct {
@@ -22,6 +22,7 @@ import (
22 "context"22 "context"
23 "errors"23 "errors"
24 "fmt"24 "fmt"
25+ "maps"
25 "net"26 "net"
26 "net/netip"27 "net/netip"
27 "os"28 "os"
@@ -285,10 +286,11 @@ const (
285 )286 )
286 287
287 type containerSandbox struct {288 type containerSandbox struct {
288- d *Driver289+ d *Driver
289- slot int290+ slot int
290- id string291+ id string
291- gate sandbox.BootGate292+ gate sandbox.BootGate
293+ timings map[string]int64
292 294
293 mu sync.Mutex295 mu sync.Mutex
294 started bool296 started bool
@@ -305,6 +307,10 @@ type containerSandbox struct {
305 attached bool307 attached bool
306 }308 }
307 309
310+// StartTimings returns a copy of the phase durations of a successful Start.
311+// It is valid once Start has returned, which writes them.
312+func (s *containerSandbox) StartTimings() map[string]int64 { return maps.Clone(s.timings) }
313+
308 // ID implements sandbox.Sandbox.314 // ID implements sandbox.Sandbox.
309 func (s *containerSandbox) ID() string { return s.id }315 func (s *containerSandbox) ID() string { return s.id }
310 316
@@ -334,8 +340,11 @@ func (s *containerSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (san
334 if d.cfg.GuestForge != nil {340 if d.cfg.GuestForge != nil {
335 log.Warn().Str("job", spec.JobID).Str("container", s.id).Msg(d.cfg.GuestForge.Warning())341 log.Warn().Str("job", spec.JobID).Str("container", s.id).Msg(d.cfg.GuestForge.Warning())
336 }342 }
343+ start := time.Now()
344+ s.timings = make(map[string]int64)
337 conn, err := s.run(bootCtx, spec)345 conn, err := s.run(bootCtx, spec)
338 if err != nil {346 if err != nil {
347+ s.timings = nil
339 var tail string348 var tail string
340 if s.container != "" {349 if s.container != "" {
341 tail, _ = d.eng.LogsTail(context.WithoutCancel(ctx), s.container, 40)350 tail, _ = d.eng.LogsTail(context.WithoutCancel(ctx), s.container, 40)
@@ -346,7 +355,11 @@ func (s *containerSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (san
346 }355 }
347 return nil, errors.Join(err, cerr)356 return nil, errors.Join(err, cerr)
348 }357 }
349- log.Info().Str("container", s.container).Int("slot", s.slot).Str("job", spec.JobID).Bool("privileged", d.privileged).Msg("container running")358+ fields := map[string]any{"total_ms": time.Since(start).Milliseconds()}
359+ for k, v := range s.timings {
360+ fields[k] = v
361+ }
362+ log.Info().Str("container", s.container).Int("slot", s.slot).Str("job", spec.JobID).Bool("privileged", d.privileged).Fields(fields).Msg("container running")
350 return conn, nil363 return conn, nil
351 }364 }
352 365
@@ -356,25 +369,40 @@ func (s *containerSandbox) run(ctx context.Context, spec sandbox.JobSpec) (sandb
356 if err != nil {369 if err != nil {
357 return nil, err370 return nil, err
358 }371 }
372+ start := time.Now()
359 accepted, err := s.runSocket()373 accepted, err := s.runSocket()
360 if err != nil {374 if err != nil {
361 return nil, err375 return nil, err
362 }376 }
377+ s.timings["socket_ms"] = time.Since(start).Milliseconds()
378+ start = time.Now()
363 if err := s.runCache(spec); err != nil {379 if err := s.runCache(spec); err != nil {
364 return nil, err380 return nil, err
365 }381 }
382+ s.timings["cache_ms"] = time.Since(start).Milliseconds()
383+ start = time.Now()
366 if err := s.runScratch(ctx, spec); err != nil {384 if err := s.runScratch(ctx, spec); err != nil {
367 return nil, err385 return nil, err
368 }386 }
369 labels := map[string]string{sandbox.LabelJob: spec.JobID}387 labels := map[string]string{sandbox.LabelJob: spec.JobID}
388+ s.timings["scratch_ms"] = time.Since(start).Milliseconds()
389+ start = time.Now()
370 if err := s.runNetwork(ctx, netw, labels); err != nil {390 if err := s.runNetwork(ctx, netw, labels); err != nil {
371 return nil, err391 return nil, err
372 }392 }
393+ s.timings["net_ms"] = time.Since(start).Milliseconds()
394+ start = time.Now()
373 exited, err := s.runContainer(ctx, spec, netw, labels)395 exited, err := s.runContainer(ctx, spec, netw, labels)
374 if err != nil {396 if err != nil {
375 return nil, err397 return nil, err
376 }398 }
377- return s.waitExecutor(ctx, accepted, exited)399+ s.timings["container_ms"] = time.Since(start).Milliseconds()
400+ start = time.Now()
401+ conn, err := s.waitExecutor(ctx, accepted, exited)
402+ if err == nil {
403+ s.timings["connect_ms"] = time.Since(start).Milliseconds()
404+ }
405+ return conn, err
378 }406 }
379 407
380 type acceptResult struct {408 type acceptResult struct {
M internal/sandbox/sandbox.go +7 -0
@@ -89,6 +89,13 @@ type Sandbox interface {
8989 Stop(ctx context.Context) error
9090 }
9191
92+// Timed is implemented by sandboxes that measure their start phases.
93+type Timed interface {
94+ // StartTimings returns a copy of the phase durations (ms) of a successful
95+ // Start, nil after a failed one. Call it only once Start has returned.
96+ StartTimings() map[string]int64
97+}
98+
9299 // Driver creates sandboxes of one mode and reaps what a previous run left.
93100 type Driver interface {
94101 // Mode is VM or Container.
@@ -89,6 +89,13 @@ type Sandbox interface {
89 Stop(ctx context.Context) error89 Stop(ctx context.Context) error
90 }90 }
91 91
92+// Timed is implemented by sandboxes that measure their start phases.
93+type Timed interface {
94+ // StartTimings returns a copy of the phase durations (ms) of a successful
95+ // Start, nil after a failed one. Call it only once Start has returned.
96+ StartTimings() map[string]int64
97+}
98+
92 // Driver creates sandboxes of one mode and reaps what a previous run left.99 // Driver creates sandboxes of one mode and reaps what a previous run left.
93 type Driver interface {100 type Driver interface {
94 // Mode is VM or Container.101 // Mode is VM or Container.
M internal/sandbox/vm/driver.go +42 -7
@@ -16,6 +16,7 @@ import (
1616 "errors"
1717 "fmt"
1818 "io/fs"
19+ "maps"
1920 "net"
2021 "net/netip"
2122 "os"
@@ -232,10 +233,11 @@ func (d *Driver) markLeaked(slot int) {
232233
233234 // vmSandbox is one VM, single use.
234235 type vmSandbox struct {
235- d *Driver
236- slot int
237- id string
238- gate sandbox.BootGate
236+ d *Driver
237+ slot int
238+ id string
239+ gate sandbox.BootGate
240+ timings map[string]int64
239241
240242 mu sync.Mutex
241243 started bool
@@ -257,6 +259,10 @@ type vmSandbox struct {
257259 attached bool
258260 }
259261
262+// StartTimings returns a copy of the phase durations of a successful Start.
263+// It is valid once Start has returned, which writes them.
264+func (s *vmSandbox) StartTimings() map[string]int64 { return maps.Clone(s.timings) }
265+
260266 // ID implements sandbox.Sandbox.
261267 func (s *vmSandbox) ID() string { return s.id }
262268
@@ -299,8 +305,11 @@ func (s *vmSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Co
299305 return nil, fmt.Errorf("write state record: %w", err)
300306 }
301307
308+ start := time.Now()
309+ s.timings = make(map[string]int64)
302310 conn, err := s.boot(bootCtx, spec)
303311 if err != nil {
312+ s.timings = nil
304313 console := ""
305314 if s.console != nil {
306315 console = s.console.String()
@@ -311,7 +320,11 @@ func (s *vmSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Co
311320 }
312321 return nil, errors.Join(err, cerr)
313322 }
314- log.Info().Str("vm", s.id).Int("slot", s.slot).Str("job", spec.JobID).Msg("micro-VM running")
323+ fields := map[string]any{"total_ms": time.Since(start).Milliseconds()}
324+ for k, v := range s.timings {
325+ fields[k] = v
326+ }
327+ log.Info().Str("vm", s.id).Int("slot", s.slot).Str("job", spec.JobID).Fields(fields).Msg("micro-VM running")
315328 return conn, nil
316329 }
317330
@@ -322,17 +335,24 @@ func (s *vmSandbox) boot(ctx context.Context, spec sandbox.JobSpec) (sandbox.Con
322335 if err != nil {
323336 return nil, err
324337 }
338+ start := time.Now()
325339 if err := s.bootNetwork(ctx, netw); err != nil {
326340 return nil, err
327341 }
342+ s.timings["net_ms"] = time.Since(start).Milliseconds() - s.timings["tap_ms"]
343+ start = time.Now()
328344 api, err := s.bootVMM(ctx, spec)
329345 if err != nil {
330346 return nil, err
331347 }
348+ s.timings["vmm_ms"] = time.Since(start).Milliseconds()
349+ start = time.Now()
332350 files, err := s.bootFiles(ctx, spec)
333351 if err != nil {
334352 return nil, err
335353 }
354+ s.timings["files_ms"] = time.Since(start).Milliseconds() - s.timings["scratch_ms"]
355+ start = time.Now()
336356 accepted, err := s.bootVsock()
337357 if err != nil {
338358 return nil, err
@@ -340,10 +360,18 @@ func (s *vmSandbox) boot(ctx context.Context, spec sandbox.JobSpec) (sandbox.Con
340360 if err := s.bootCache(spec); err != nil {
341361 return nil, err
342362 }
363+ s.timings["vsock_ms"] = time.Since(start).Milliseconds()
364+ start = time.Now()
343365 if err := api.configure(ctx, s.vmConfig(spec, netw, files)); err != nil {
344366 return nil, err
345367 }
346- return s.waitExecutor(ctx, accepted)
368+ s.timings["configure_ms"] = time.Since(start).Milliseconds()
369+ start = time.Now()
370+ conn, err := s.waitExecutor(ctx, accepted)
371+ if err == nil {
372+ s.timings["connect_ms"] = time.Since(start).Milliseconds()
373+ }
374+ return conn, err
347375 }
348376
349377 // bootNetwork creates the TAP (it must exist when Firecracker opens it) and
@@ -355,9 +383,11 @@ func (s *vmSandbox) bootNetwork(ctx context.Context, netw netpolicy.Net) error {
355383 if err := d.store.save(s.rec); err != nil {
356384 return err
357385 }
386+ start := time.Now()
358387 if err := createTap(ctx, d.opt.Run, s.tap, s.lay.uid, s.gateway); err != nil {
359388 return err
360389 }
390+ s.timings["tap_ms"] = time.Since(start).Milliseconds()
361391 s.attached = true // from here Detach is idempotent and always safe
362392 return d.net.Attach(ctx, netpolicy.Guest{Iface: s.tap, Gateway: netw.Gateway, Addr: netw.Guest, ForgePort: d.cfg.GuestForge.TCPPort()})
363393 }
@@ -402,7 +432,12 @@ func (s *vmSandbox) bootFiles(ctx context.Context, spec sandbox.JobSpec) (vmFile
402432 if s.lay.jailed {
403433 owner = s.lay.uid
404434 }
405- return f, disk.CreateScratch(ctx, d.opt.Run, s.lay.host("scratch.ext4"), spec.DiskMiB, owner)
435+ start := time.Now()
436+ err = disk.CreateScratch(ctx, d.opt.Run, s.lay.host("scratch.ext4"), spec.DiskMiB, owner)
437+ if err == nil {
438+ s.timings["scratch_ms"] = time.Since(start).Milliseconds()
439+ }
440+ return f, err
406441 }
407442
408443 // shareExternals makes the read-only disk of the host's external tools available
@@ -16,6 +16,7 @@ import (
16 "errors"16 "errors"
17 "fmt"17 "fmt"
18 "io/fs"18 "io/fs"
19+ "maps"
19 "net"20 "net"
20 "net/netip"21 "net/netip"
21 "os"22 "os"
@@ -232,10 +233,11 @@ func (d *Driver) markLeaked(slot int) {
232 233
233 // vmSandbox is one VM, single use.234 // vmSandbox is one VM, single use.
234 type vmSandbox struct {235 type vmSandbox struct {
235- d *Driver236+ d *Driver
236- slot int237+ slot int
237- id string238+ id string
238- gate sandbox.BootGate239+ gate sandbox.BootGate
240+ timings map[string]int64
239 241
240 mu sync.Mutex242 mu sync.Mutex
241 started bool243 started bool
@@ -257,6 +259,10 @@ type vmSandbox struct {
257 attached bool259 attached bool
258 }260 }
259 261
262+// StartTimings returns a copy of the phase durations of a successful Start.
263+// It is valid once Start has returned, which writes them.
264+func (s *vmSandbox) StartTimings() map[string]int64 { return maps.Clone(s.timings) }
265+
260 // ID implements sandbox.Sandbox.266 // ID implements sandbox.Sandbox.
261 func (s *vmSandbox) ID() string { return s.id }267 func (s *vmSandbox) ID() string { return s.id }
262 268
@@ -299,8 +305,11 @@ func (s *vmSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Co
299 return nil, fmt.Errorf("write state record: %w", err)305 return nil, fmt.Errorf("write state record: %w", err)
300 }306 }
301 307
308+ start := time.Now()
309+ s.timings = make(map[string]int64)
302 conn, err := s.boot(bootCtx, spec)310 conn, err := s.boot(bootCtx, spec)
303 if err != nil {311 if err != nil {
312+ s.timings = nil
304 console := ""313 console := ""
305 if s.console != nil {314 if s.console != nil {
306 console = s.console.String()315 console = s.console.String()
@@ -311,7 +320,11 @@ func (s *vmSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Co
311 }320 }
312 return nil, errors.Join(err, cerr)321 return nil, errors.Join(err, cerr)
313 }322 }
314- log.Info().Str("vm", s.id).Int("slot", s.slot).Str("job", spec.JobID).Msg("micro-VM running")323+ fields := map[string]any{"total_ms": time.Since(start).Milliseconds()}
324+ for k, v := range s.timings {
325+ fields[k] = v
326+ }
327+ log.Info().Str("vm", s.id).Int("slot", s.slot).Str("job", spec.JobID).Fields(fields).Msg("micro-VM running")
315 return conn, nil328 return conn, nil
316 }329 }
317 330
@@ -322,17 +335,24 @@ func (s *vmSandbox) boot(ctx context.Context, spec sandbox.JobSpec) (sandbox.Con
322 if err != nil {335 if err != nil {
323 return nil, err336 return nil, err
324 }337 }
338+ start := time.Now()
325 if err := s.bootNetwork(ctx, netw); err != nil {339 if err := s.bootNetwork(ctx, netw); err != nil {
326 return nil, err340 return nil, err
327 }341 }
342+ s.timings["net_ms"] = time.Since(start).Milliseconds() - s.timings["tap_ms"]
343+ start = time.Now()
328 api, err := s.bootVMM(ctx, spec)344 api, err := s.bootVMM(ctx, spec)
329 if err != nil {345 if err != nil {
330 return nil, err346 return nil, err
331 }347 }
348+ s.timings["vmm_ms"] = time.Since(start).Milliseconds()
349+ start = time.Now()
332 files, err := s.bootFiles(ctx, spec)350 files, err := s.bootFiles(ctx, spec)
333 if err != nil {351 if err != nil {
334 return nil, err352 return nil, err
335 }353 }
354+ s.timings["files_ms"] = time.Since(start).Milliseconds() - s.timings["scratch_ms"]
355+ start = time.Now()
336 accepted, err := s.bootVsock()356 accepted, err := s.bootVsock()
337 if err != nil {357 if err != nil {
338 return nil, err358 return nil, err
@@ -340,10 +360,18 @@ func (s *vmSandbox) boot(ctx context.Context, spec sandbox.JobSpec) (sandbox.Con
340 if err := s.bootCache(spec); err != nil {360 if err := s.bootCache(spec); err != nil {
341 return nil, err361 return nil, err
342 }362 }
363+ s.timings["vsock_ms"] = time.Since(start).Milliseconds()
364+ start = time.Now()
343 if err := api.configure(ctx, s.vmConfig(spec, netw, files)); err != nil {365 if err := api.configure(ctx, s.vmConfig(spec, netw, files)); err != nil {
344 return nil, err366 return nil, err
345 }367 }
346- return s.waitExecutor(ctx, accepted)368+ s.timings["configure_ms"] = time.Since(start).Milliseconds()
369+ start = time.Now()
370+ conn, err := s.waitExecutor(ctx, accepted)
371+ if err == nil {
372+ s.timings["connect_ms"] = time.Since(start).Milliseconds()
373+ }
374+ return conn, err
347 }375 }
348 376
349 // bootNetwork creates the TAP (it must exist when Firecracker opens it) and377 // bootNetwork creates the TAP (it must exist when Firecracker opens it) and
@@ -355,9 +383,11 @@ func (s *vmSandbox) bootNetwork(ctx context.Context, netw netpolicy.Net) error {
355 if err := d.store.save(s.rec); err != nil {383 if err := d.store.save(s.rec); err != nil {
356 return err384 return err
357 }385 }
386+ start := time.Now()
358 if err := createTap(ctx, d.opt.Run, s.tap, s.lay.uid, s.gateway); err != nil {387 if err := createTap(ctx, d.opt.Run, s.tap, s.lay.uid, s.gateway); err != nil {
359 return err388 return err
360 }389 }
390+ s.timings["tap_ms"] = time.Since(start).Milliseconds()
361 s.attached = true // from here Detach is idempotent and always safe391 s.attached = true // from here Detach is idempotent and always safe
362 return d.net.Attach(ctx, netpolicy.Guest{Iface: s.tap, Gateway: netw.Gateway, Addr: netw.Guest, ForgePort: d.cfg.GuestForge.TCPPort()})392 return d.net.Attach(ctx, netpolicy.Guest{Iface: s.tap, Gateway: netw.Gateway, Addr: netw.Guest, ForgePort: d.cfg.GuestForge.TCPPort()})
363 }393 }
@@ -402,7 +432,12 @@ func (s *vmSandbox) bootFiles(ctx context.Context, spec sandbox.JobSpec) (vmFile
402 if s.lay.jailed {432 if s.lay.jailed {
403 owner = s.lay.uid433 owner = s.lay.uid
404 }434 }
405- return f, disk.CreateScratch(ctx, d.opt.Run, s.lay.host("scratch.ext4"), spec.DiskMiB, owner)435+ start := time.Now()
436+ err = disk.CreateScratch(ctx, d.opt.Run, s.lay.host("scratch.ext4"), spec.DiskMiB, owner)
437+ if err == nil {
438+ s.timings["scratch_ms"] = time.Since(start).Milliseconds()
439+ }
440+ return f, err
406 }441 }
407 442
408 // shareExternals makes the read-only disk of the host's external tools available443 // shareExternals makes the read-only disk of the host's external tools available
M internal/sandbox/vm/vm_test.go +10 -0
@@ -228,6 +228,16 @@ func TestStartRunStop(t *testing.T) {
228228 if err != nil {
229229 t.Fatal(err)
230230 }
231+ timings := sb.(sandbox.Timed).StartTimings()
232+ for _, key := range []string{"tap_ms", "net_ms", "vmm_ms", "files_ms", "scratch_ms", "vsock_ms", "configure_ms", "connect_ms"} {
233+ if value, ok := timings[key]; !ok || value < 0 {
234+ t.Errorf("%s = %d, present=%v", key, value, ok)
235+ }
236+ }
237+ timings["tap_ms"] = -1
238+ if sb.(sandbox.Timed).StartTimings()["tap_ms"] < 0 {
239+ t.Error("shared timing map")
240+ }
231241 out, code := runStub(t, conn)
232242 if out != "hello from fake guest\n" || code != 0 {
233243 t.Fatalf("out=%q code=%d", out, code)
@@ -228,6 +228,16 @@ func TestStartRunStop(t *testing.T) {
228 if err != nil {228 if err != nil {
229 t.Fatal(err)229 t.Fatal(err)
230 }230 }
231+ timings := sb.(sandbox.Timed).StartTimings()
232+ for _, key := range []string{"tap_ms", "net_ms", "vmm_ms", "files_ms", "scratch_ms", "vsock_ms", "configure_ms", "connect_ms"} {
233+ if value, ok := timings[key]; !ok || value < 0 {
234+ t.Errorf("%s = %d, present=%v", key, value, ok)
235+ }
236+ }
237+ timings["tap_ms"] = -1
238+ if sb.(sandbox.Timed).StartTimings()["tap_ms"] < 0 {
239+ t.Error("shared timing map")
240+ }
231 out, code := runStub(t, conn)241 out, code := runStub(t, conn)
232 if out != "hello from fake guest\n" || code != 0 {242 if out != "hello from fake guest\n" || code != 0 {
233 t.Fatalf("out=%q code=%d", out, code)243 t.Fatalf("out=%q code=%d", out, code)
M main.go +1 -0
@@ -79,6 +79,7 @@ func execCmd(ctx context.Context) int {
7979 }
8080 code, err := guest.Exec(ctx, guest.ExecConfig{
8181 Target: target,
82+ BootMs: guest.ParseBootMs(os.Getenv("ERMITE_BOOT_MS")),
8283 Sandbox: envOr("ERMITE_SANDBOX", "container"),
8384 Version: envOr("ERMITE_VERSION", version),
8485 // A test image without the user "runner" (ERMITE_RUNNER_ALLOW_ROOT_STEPS=1);
@@ -79,6 +79,7 @@ func execCmd(ctx context.Context) int {
79 }79 }
80 code, err := guest.Exec(ctx, guest.ExecConfig{80 code, err := guest.Exec(ctx, guest.ExecConfig{
81 Target: target,81 Target: target,
82+ BootMs: guest.ParseBootMs(os.Getenv("ERMITE_BOOT_MS")),
82 Sandbox: envOr("ERMITE_SANDBOX", "container"),83 Sandbox: envOr("ERMITE_SANDBOX", "container"),
83 Version: envOr("ERMITE_VERSION", version),84 Version: envOr("ERMITE_VERSION", version),
84 // A test image without the user "runner" (ERMITE_RUNNER_ALLOW_ROOT_STEPS=1);85 // A test image without the user "runner" (ERMITE_RUNNER_ALLOW_ROOT_STEPS=1);
A registry_setup_test.go +75 -0
@@ -0,0 +1,75 @@
1+// SPDX-License-Identifier: MIT
2+
3+package main
4+
5+import (
6+ "context"
7+ "strings"
8+ "testing"
9+
10+ "bnema.dev/ermite-runner/internal/actionscache"
11+ "bnema.dev/ermite-runner/internal/cachesvc"
12+ "bnema.dev/ermite-runner/internal/sandbox"
13+)
14+
15+// The three size caps share one parser: a zero size in any unit turns the
16+// registry proxy and the action archive cache off, and is refused for the
17+// cache API, which ERMITE_RUNNER_CACHE=0 turns off. Each error names its
18+// variable.
19+func TestCacheSizeConfiguration(t *testing.T) {
20+ t.Setenv(envCache, "1")
21+ for _, tc := range []struct {
22+ registry, actions, cache string
23+ registryOn, archivesOn bool
24+ }{
25+ {"", "", "1M", true, true},
26+ {"1M", "0", "2M", true, false},
27+ {"0", "1M", "1M", false, true},
28+ {"0G", "0GiB", "1M", false, false},
29+ {" 0M ", "", "1M", false, true},
30+ {"00", "", "1M", false, true},
31+ {"0GiB", "", "1M", false, true},
32+ {"10M", "", "1M", true, true},
33+ } {
34+ t.Setenv(envRegistryMax, tc.registry)
35+ t.Setenv(actionscache.EnvMax, tc.actions)
36+ t.Setenv(cachesvc.EnvMax, tc.cache)
37+ cfg := sandbox.Config{StateDir: t.TempDir()}
38+ svc, archives, err := setupCache(context.Background(), &cfg)
39+ if err != nil {
40+ t.Fatalf("%+v: %v", tc, err)
41+ }
42+ if svc.RegistryEnabled() != tc.registryOn || (archives != nil) != tc.archivesOn {
43+ t.Fatalf("%+v: registry=%v archives=%v", tc, svc.RegistryEnabled(), archives != nil)
44+ }
45+ _ = svc.CloseAll()
46+ }
47+ t.Setenv(envRegistryMax, "")
48+ t.Setenv(actionscache.EnvMax, "")
49+ t.Setenv(cachesvc.EnvMax, "")
50+ for _, tc := range []struct{ env, value string }{
51+ {envRegistryMax, "invalid"}, {envRegistryMax, "G"}, {envRegistryMax, "-1G"},
52+ {actionscache.EnvMax, "x"},
53+ {cachesvc.EnvMax, "0"}, {cachesvc.EnvMax, "0G"}, {cachesvc.EnvMax, "-1"},
54+ } {
55+ t.Run(tc.env+"="+tc.value, func(t *testing.T) {
56+ t.Setenv(tc.env, tc.value)
57+ cfg := sandbox.Config{StateDir: t.TempDir()}
58+ svc, _, err := setupCache(context.Background(), &cfg)
59+ if err == nil {
60+ _ = svc.CloseAll()
61+ t.Fatal("accepted")
62+ }
63+ if !strings.HasPrefix(err.Error(), tc.env+": ") {
64+ t.Fatalf("error does not name %s: %v", tc.env, err)
65+ }
66+ })
67+ }
68+ t.Setenv(envCache, "0")
69+ t.Setenv(envRegistryMax, "invalid")
70+ cfg := sandbox.Config{StateDir: t.TempDir()}
71+ svc, _, err := setupCache(context.Background(), &cfg)
72+ if err != nil || svc != nil {
73+ t.Fatalf("cache off: %v %v", svc, err)
74+ }
75+}
@@ -0,0 +1,75 @@
1+// SPDX-License-Identifier: MIT
2+
3+package main
4+
5+import (
6+ "context"
7+ "strings"
8+ "testing"
9+
10+ "bnema.dev/ermite-runner/internal/actionscache"
11+ "bnema.dev/ermite-runner/internal/cachesvc"
12+ "bnema.dev/ermite-runner/internal/sandbox"
13+)
14+
15+// The three size caps share one parser: a zero size in any unit turns the
16+// registry proxy and the action archive cache off, and is refused for the
17+// cache API, which ERMITE_RUNNER_CACHE=0 turns off. Each error names its
18+// variable.
19+func TestCacheSizeConfiguration(t *testing.T) {
20+ t.Setenv(envCache, "1")
21+ for _, tc := range []struct {
22+ registry, actions, cache string
23+ registryOn, archivesOn bool
24+ }{
25+ {"", "", "1M", true, true},
26+ {"1M", "0", "2M", true, false},
27+ {"0", "1M", "1M", false, true},
28+ {"0G", "0GiB", "1M", false, false},
29+ {" 0M ", "", "1M", false, true},
30+ {"00", "", "1M", false, true},
31+ {"0GiB", "", "1M", false, true},
32+ {"10M", "", "1M", true, true},
33+ } {
34+ t.Setenv(envRegistryMax, tc.registry)
35+ t.Setenv(actionscache.EnvMax, tc.actions)
36+ t.Setenv(cachesvc.EnvMax, tc.cache)
37+ cfg := sandbox.Config{StateDir: t.TempDir()}
38+ svc, archives, err := setupCache(context.Background(), &cfg)
39+ if err != nil {
40+ t.Fatalf("%+v: %v", tc, err)
41+ }
42+ if svc.RegistryEnabled() != tc.registryOn || (archives != nil) != tc.archivesOn {
43+ t.Fatalf("%+v: registry=%v archives=%v", tc, svc.RegistryEnabled(), archives != nil)
44+ }
45+ _ = svc.CloseAll()
46+ }
47+ t.Setenv(envRegistryMax, "")
48+ t.Setenv(actionscache.EnvMax, "")
49+ t.Setenv(cachesvc.EnvMax, "")
50+ for _, tc := range []struct{ env, value string }{
51+ {envRegistryMax, "invalid"}, {envRegistryMax, "G"}, {envRegistryMax, "-1G"},
52+ {actionscache.EnvMax, "x"},
53+ {cachesvc.EnvMax, "0"}, {cachesvc.EnvMax, "0G"}, {cachesvc.EnvMax, "-1"},
54+ } {
55+ t.Run(tc.env+"="+tc.value, func(t *testing.T) {
56+ t.Setenv(tc.env, tc.value)
57+ cfg := sandbox.Config{StateDir: t.TempDir()}
58+ svc, _, err := setupCache(context.Background(), &cfg)
59+ if err == nil {
60+ _ = svc.CloseAll()
61+ t.Fatal("accepted")
62+ }
63+ if !strings.HasPrefix(err.Error(), tc.env+": ") {
64+ t.Fatalf("error does not name %s: %v", tc.env, err)
65+ }
66+ })
67+ }
68+ t.Setenv(envCache, "0")
69+ t.Setenv(envRegistryMax, "invalid")
70+ cfg := sandbox.Config{StateDir: t.TempDir()}
71+ svc, _, err := setupCache(context.Background(), &cfg)
72+ if err != nil || svc != nil {
73+ t.Fatalf("cache off: %v %v", svc, err)
74+ }
75+}

Comments1 open / 1

bnema

E2E on the head of this MR (2b9f5c9), production setup: unprivileged container under rootless Docker, vm isolation, /dev/kvm + /dev/net/tun + NET_ADMIN only, guest forge relay, the env of ermite-prod.toml. Both jobs green on two passes. setup-go: Found in cache (1.27.1). alpine:3: first pass fetched through the proxy, second pass all manifests hit=true and no blob fetched. Job total 2.6 s (container) and 6.2 s (go) on the warm pass. Not covered: rootless Podman itself (no podman on the test host).

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
l
merge request: the list of files (j k move, enter opens, esc back)
M
merge request: the merge menu; there 1–3 pick the mode, D deletes the branch, enter merges
R
merge request: mark the head read
esc
close a dialog, leave a field

Branches 4

Tags 11

↑↓ moveenter openesc close