test: production-shaped push/pull benchmark with per-backend request counts

TestBenchRealImages pushes and pulls three images whose layer sizes are copied from real manifests in the production appview database (the median, p75 and p90 images by layer count) and reports, per operation, wall time and the number of requests to the registry, the fake PDS, the hold and S3, broken down by endpoint. Skipped unless BENCH_PROFILES is set, so the integration target does not run it. BENCH_LAT_{PDS,HOLD,S3} inject per-request latency, which is what makes byte-path changes visible in-process; request counts are the reliable signal either way. internal/reqcount counts and delays requests through a handler wrapper and a client-side RoundTripper. testharness.WithBackendTap wraps the PDS and S3 handlers and puts a counting reverse proxy in front of the hold; testpds.WithMiddleware is the hook that makes the PDS side possible. The bench showed a pull costs three hold calls per blob, not two: distribution installs its notifications listener unconditionally and it re-Stats every blob after ServeBlob to build the pull event. The backlog's presign memoization item is rewritten with the measured numbers. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WTdBxLFU5TpwmqVdVsN1wq
2026-09-19 08:44:14 +00:00 · 2026-09-11 17:05:19 -05:00
parent 65db7945b2
commit bf4e63e810
5 changed files with 438 additions and 13 deletions
@@ -28,20 +28,29 @@ are pipelined so one part is in flight while the next fills.

 ### Presign memoization within a request

-**Problem.** Distribution calls `Stat` then `ServeBlob` for every blob GET and
-HEAD. Each asks the hold for a presigned URL, so a client fetch costs two hold
-calls, each doing full token validation and crew lookup.
+**Problem.** Every blob GET costs three hold calls, each doing full token
+validation, a captain record read and a presign. Distribution's blob handler
+calls `Stat` then `ServeBlob`, and distribution's notifications listener
+(`notifications.Listen` wraps every repository unconditionally, whether or not
+any endpoint is configured) calls `Stat` a third time after `ServeBlob` to
+build the pull event. Measured 2026-09-11 with `TestBenchRealImages`: a p90
+pull of 22 layers is 22 registry GETs and 66 hold calls, 44 of them HEAD
+presigns whose URL is never used.

 **Where.** `ProxyBlobStore.Stat` and `ServeBlob` in
 `pkg/appview/storage/proxy_blob_store.go`.

-**Fix.** The store is built per request, so a small map keyed by digest and
-method inside it is safe. Stat requests the URL and size once; ServeBlob reuses
-the URL if the method matches. S3 signs the HTTP method, so a HEAD from Docker
-cannot reuse a GET-signed URL; either key the cache by method or have the hold
-return both URLs in one response.
+**Fix.** The store is built per request, so a small memo keyed by digest and
+method inside it is safe. Stat reads the request method from the context
+(`storage.HTTPRequestMethod`, already set by the auth middleware), presigns
+for that method when it is GET or HEAD, and remembers the URL and size.
+ServeBlob reuses the URL when the digest and method match; the listener's
+second Stat is answered from the memo. S3 signs the HTTP method, so a HEAD
+URL cannot serve a GET; anything other than GET or HEAD must still presign
+HEAD, since the hold refuses `method=PUT` on the read path.

-**Impact.** One hold call per blob fetch instead of two.
+**Impact.** One hold call per blob fetch instead of three. On the p90 pull
+that is 66 hold calls down to 22.

 ### Stat from the appview's own layers table

@@ -0,0 +1,174 @@
+// Package reqcount counts HTTP requests per backend and can inject a fixed
+// latency in front of each, so an in-process benchmark of the ATCR stack can
+// report round trips per operation. Round-trip counts are the reliable signal
+// in-process: with no latency injected, wall time is dominated by gofakes3's
+// in-memory copies and crane's client-side gzip, not by ATCR code.
+package reqcount
+
+import (
+	"net/http"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Tap wraps handlers and counts what passes through them.
+type Tap struct {
+	mu      sync.Mutex
+	counts  map[string]map[string]int
+	latency map[string]time.Duration
+}
+
+// New returns an empty Tap.
+func New() *Tap {
+	return &Tap{
+		counts:  make(map[string]map[string]int),
+		latency: make(map[string]time.Duration),
+	}
+}
+
+// SetLatency makes every request to backend sleep for d before it is served.
+func (t *Tap) SetLatency(backend string, d time.Duration) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.latency[backend] = d
+}
+
+// Wrap returns next wrapped with counting and latency for backend.
+func (t *Tap) Wrap(backend string, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		key := classify(backend, r)
+		t.mu.Lock()
+		m := t.counts[backend]
+		if m == nil {
+			m = make(map[string]int)
+			t.counts[backend] = m
+		}
+		m[key]++
+		d := t.latency[backend]
+		t.mu.Unlock()
+		if d > 0 {
+			time.Sleep(d)
+		}
+		next.ServeHTTP(w, r)
+	})
+}
+
+// Reset clears every count.
+func (t *Tap) Reset() {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.counts = make(map[string]map[string]int)
+}
+
+// Snapshot returns a copy of the counts: backend -> key -> requests.
+func (t *Tap) Snapshot() map[string]map[string]int {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	out := make(map[string]map[string]int, len(t.counts))
+	for b, m := range t.counts {
+		c := make(map[string]int, len(m))
+		for k, n := range m {
+			c[k] = n
+		}
+		out[b] = c
+	}
+	return out
+}
+
+// Total sums one backend's counts.
+func Total(m map[string]int) int {
+	n := 0
+	for _, v := range m {
+		n += v
+	}
+	return n
+}
+
+// Keys returns a backend's keys sorted by count, largest first.
+func Keys(m map[string]int) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		if m[keys[i]] != m[keys[j]] {
+			return m[keys[i]] > m[keys[j]]
+		}
+		return keys[i] < keys[j]
+	})
+	return keys
+}
+
+// classify turns a request into a short, stable key for its backend.
+func classify(backend string, r *http.Request) string {
+	q := r.URL.Query()
+	switch backend {
+	case "s3":
+		switch {
+		case q.Has("uploads"):
+			return "POST initiate"
+		case q.Has("uploadId") && q.Has("partNumber"):
+			return "PUT part"
+		case q.Has("uploadId") && r.Method == http.MethodPost:
+			return "POST complete"
+		case q.Has("uploadId") && r.Method == http.MethodDelete:
+			return "DELETE abort"
+		case r.Header.Get("X-Amz-Copy-Source") != "":
+			return "PUT copy"
+		}
+		return r.Method + " object"
+	case "hold":
+		p := strings.TrimPrefix(r.URL.Path, "/xrpc/")
+		if strings.HasSuffix(p, "sync.getBlob") && q.Get("method") != "" {
+			p += "?method=" + q.Get("method")
+		}
+		return r.Method + " " + p
+	default:
+		return r.Method + " " + strings.TrimPrefix(r.URL.Path, "/xrpc/")
+	}
+}
+
+// Transport wraps an http.RoundTripper so client-side requests are counted
+// under backend, keyed by method and a coarse path class. It is meant for the
+// registry itself, which sits in front of the tapped backends.
+func (t *Tap) Transport(backend string, base http.RoundTripper) http.RoundTripper {
+	if base == nil {
+		base = http.DefaultTransport
+	}
+	return roundTripFunc(func(r *http.Request) (*http.Response, error) {
+		t.mu.Lock()
+		m := t.counts[backend]
+		if m == nil {
+			m = make(map[string]int)
+			t.counts[backend] = m
+		}
+		m[r.Method+" "+pathClass(r.URL.Path)]++
+		t.mu.Unlock()
+		return base.RoundTrip(r)
+	})
+}
+
+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
+
+func pathClass(p string) string {
+	switch {
+	case !strings.HasPrefix(p, "/v2/") && !strings.HasPrefix(p, "/auth/"):
+		// A redirect the client followed, to S3 or elsewhere.
+		return "redirect"
+	case strings.Contains(p, "/blobs/uploads"):
+		return "blobs/uploads"
+	case strings.Contains(p, "/blobs/"):
+		return "blobs"
+	case strings.Contains(p, "/manifests/"):
+		return "manifests"
+	case strings.HasPrefix(p, "/auth/"):
+		return "auth"
+	case p == "/v2/" || p == "/v2":
+		return "ping"
+	}
+	return "other"
+}
@@ -9,6 +9,8 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"net/http/httputil"
+	"net/url"
 	"os"
 	"path/filepath"
 	"strings"
@@ -36,11 +38,20 @@ import (
 type Option func(*options)

 type options struct {
+	tap         func(backend string, next http.Handler) http.Handler
 	quota       *quota.Config
 	billing     *billing.Config
 	privateHold bool
 }

+// WithBackendTap wraps every backend the stack talks to: the fake PDS handler
+// ("pds"), the gofakes3 handler ("s3"), and a reverse proxy placed in front of
+// the hold ("hold") so the appview's XRPC calls pass through it. Benchmarks
+// use it to count round trips per operation and to inject latency.
+func WithBackendTap(tap func(backend string, next http.Handler) http.Handler) Option {
+	return func(o *options) { o.tap = tap }
+}
+
 // WithPrivateHold builds the hold with captain.Public = false. Reads then
 // require the owner or a crew member, so anonymous pulls are refused and a
 // PDS-known stranger is refused too — the mirror of the default public hold,
@@ -114,7 +125,13 @@ func New(t *testing.T, opts ...Option) *Harness {
 	h := &Harness{t: t}

 	// 1. Fake PDS.
-	h.PDS = testpds.New(t)
+	var pdsOpts []testpds.Option
+	if o.tap != nil {
+		pdsOpts = append(pdsOpts, testpds.WithMiddleware(func(next http.Handler) http.Handler {
+			return o.tap("pds", next)
+		}))
+	}
+	h.PDS = testpds.New(t, pdsOpts...)
 	atproto.SetDirectory(h.PDS.Directory())
 	t.Cleanup(func() {
 		// Reset to a fresh default so a later non-test process won't see our
@@ -128,7 +145,11 @@ func New(t *testing.T, opts ...Option) *Harness {
 		t.Fatalf("create test bucket: %v", err)
 	}
 	faker := gofakes3.New(backend)
-	s3ts := httptest.NewServer(faker.Server())
+	var s3Handler = faker.Server()
+	if o.tap != nil {
+		s3Handler = o.tap("s3", s3Handler)
+	}
+	s3ts := httptest.NewServer(s3Handler)
 	t.Cleanup(s3ts.Close)
 	h.S3URL = s3ts.URL

@@ -139,6 +160,21 @@ func New(t *testing.T, opts ...Option) *Harness {
 	}
 	holdAddr := holdListener.Addr().String()
 	holdPublicURL := "http://" + holdAddr
+	if o.tap != nil {
+		// Everything that addresses the hold by its public URL (the appview,
+		// the DID document, the did:web itself) goes through a counting
+		// reverse proxy; the hold keeps serving on its own listener behind it.
+		proxyListener, err := net.Listen("tcp", "127.0.0.1:0")
+		if err != nil {
+			t.Fatalf("hold proxy listen: %v", err)
+		}
+		target, _ := url.Parse("http://" + holdAddr)
+		proxy := httputil.NewSingleHostReverseProxy(target)
+		proxySrv := &http.Server{Handler: o.tap("hold", proxy)}
+		go func() { _ = proxySrv.Serve(proxyListener) }()
+		t.Cleanup(func() { _ = proxySrv.Close() })
+		holdPublicURL = "http://" + proxyListener.Addr().String()
+	}
 	h.HoldURL = holdPublicURL
 	h.HoldDID = atprotodid.GenerateDIDFromURL(holdPublicURL)

@@ -31,12 +31,29 @@ type Server struct {
 	didHostEsc string               // percent-encoded host:port for synthesized DIDs
 }

+// Option configures a fake PDS.
+type Option func(*serverOptions)
+
+type serverOptions struct {
+	middleware func(http.Handler) http.Handler
+}
+
+// WithMiddleware wraps the PDS's handler, for request counting or latency
+// injection in benchmarks.
+func WithMiddleware(mw func(http.Handler) http.Handler) Option {
+	return func(o *serverOptions) { o.middleware = mw }
+}
+
 // New starts a fake PDS bound to a random port via httptest.NewServer. The
 // server is torn down automatically via t.Cleanup. Callers should immediately
 // install s.Directory() with atproto.SetDirectory() so DID resolution short-
 // circuits through the in-memory store.
-func New(t *testing.T) *Server {
+func New(t *testing.T, opts ...Option) *Server {
 	t.Helper()
+	var o serverOptions
+	for _, opt := range opts {
+		opt(&o)
+	}
 	s := &Server{
 		t:        t,
 		dir:      newDirectory(),
@@ -58,7 +75,11 @@ func New(t *testing.T) *Server {
 	mux.HandleFunc("/xrpc/com.atproto.repo.uploadBlob", s.handleUploadBlob)
 	mux.HandleFunc("/xrpc/com.atproto.sync.getBlob", s.handleSyncGetBlob)

-	s.httptest = httptest.NewServer(mux)
+	var handler http.Handler = mux
+	if o.middleware != nil {
+		handler = o.middleware(handler)
+	}
+	s.httptest = httptest.NewServer(handler)
 	t.Cleanup(s.httptest.Close)

 	s.didHostEsc = didWebForHost(strings.TrimPrefix(s.httptest.URL, "http://"))
@@ -0,0 +1,185 @@
+//go:build integration
+
+package integration
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/go-containerregistry/pkg/crane"
+	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/empty"
+	"github.com/google/go-containerregistry/pkg/v1/mutate"
+	"github.com/google/go-containerregistry/pkg/v1/random"
+	"github.com/google/go-containerregistry/pkg/v1/types"
+
+	"atcr.io/internal/reqcount"
+	"atcr.io/internal/testharness"
+)
+
+// Layer sizes in MB copied from three real manifests in the production
+// appview database (2026-09-10): the median image, the p75 image, and the
+// p90 image by layer count.
+var benchProfiles = map[string][]float64{
+	"median": {3.63, 1.29, 0.01, 36.32, 1.64, 0.34},
+	"p75":    {3.95, 0.44, 13.09, 0.01, 17.34, 48.04, 0.15, 0.02, 0.01, 0.15, 0.01},
+	"p90":    {3.47, 47.69, 1.20, 0.01, 1.82, 0.01, 0.03, 0.01, 0.01, 0.01, 0.01, 27.52, 15.71, 0.01, 0.07, 0.01, 0.01, 0.02, 0.01, 0.01, 0.01, 0.01},
+}
+
+// TestBenchRealImages pushes and pulls production-shaped images through the
+// in-process stack and reports round trips per backend and wall time.
+//
+// Skipped unless BENCH_PROFILES is set (a comma list of profile names, or
+// "all"). BENCH_LAT_PDS, BENCH_LAT_HOLD and BENCH_LAT_S3 inject per-request
+// latency in milliseconds. BENCH_OUT appends one JSON line per operation.
+//
+//	BENCH_PROFILES=all BENCH_LAT_PDS=50 BENCH_LAT_HOLD=5 BENCH_LAT_S3=20 \
+//	  go test -tags=integration,testmode -run TestBenchRealImages -v -count=1 ./test/integration/
+func TestBenchRealImages(t *testing.T) {
+	want := os.Getenv("BENCH_PROFILES")
+	if want == "" {
+		t.Skip("set BENCH_PROFILES to run")
+	}
+	var names []string
+	if want == "all" {
+		names = []string{"median", "p75", "p90"}
+	} else {
+		names = strings.Split(want, ",")
+	}
+
+	tap := reqcount.New()
+	for _, b := range []string{"pds", "hold", "s3"} {
+		if ms, _ := strconv.Atoi(os.Getenv("BENCH_LAT_" + strings.ToUpper(b))); ms > 0 {
+			tap.SetLatency(b, time.Duration(ms)*time.Millisecond)
+		}
+	}
+
+	h := testharness.New(t, testharness.WithBackendTap(func(backend string, next http.Handler) http.Handler {
+		return tap.Wrap(backend, next)
+	}))
+	// crane, with a counting transport so client-to-registry requests are
+	// reported under "registry" next to the backends.
+	craneOpts := func(creds testharness.Auth) []crane.Option {
+		return []crane.Option{
+			crane.WithAuth(toAuthn(creds)),
+			crane.Insecure,
+			crane.WithTransport(tap.Transport("registry", nil)),
+		}
+	}
+
+	var out *os.File
+	if p := os.Getenv("BENCH_OUT"); p != "" {
+		f, err := os.OpenFile(p, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
+		if err != nil {
+			t.Fatalf("open BENCH_OUT: %v", err)
+		}
+		defer f.Close()
+		out = f
+	}
+
+	for _, profile := range names {
+		sizes, ok := benchProfiles[profile]
+		if !ok {
+			t.Fatalf("unknown profile %q", profile)
+		}
+		sailor := h.AddSailor(profile + ".test")
+		creds := h.RegistryCreds(sailor)
+
+		// Warm the per-user caches (hold prefs row, service token) with a
+		// tiny push so the measured ops see steady state, like a returning
+		// user's second push.
+		warm, err := random.Image(1<<10, 1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		warmRef, _ := name.ParseReference(fmt.Sprintf("%s/%s/warm:latest", h.AppViewHostPort(), sailor.Handle()), name.Insecure)
+		if err := crane.Push(warm, warmRef.String(), craneOpts(creds)...); err != nil {
+			t.Fatalf("warm push: %v", err)
+		}
+
+		img := buildImage(t, sizes)
+		ref, _ := name.ParseReference(fmt.Sprintf("%s/%s/bench:%s", h.AppViewHostPort(), sailor.Handle(), profile), name.Insecure)
+
+		tap.Reset()
+		start := time.Now()
+		if err := crane.Push(img, ref.String(), craneOpts(creds)...); err != nil {
+			t.Fatalf("%s push: %v", profile, err)
+		}
+		report(t, out, profile, "push", time.Since(start), tap.Snapshot())
+
+		tap.Reset()
+		start = time.Now()
+		if err := pullAll(ref.String(), craneOpts(creds)); err != nil {
+			t.Fatalf("%s pull: %v", profile, err)
+		}
+		report(t, out, profile, "pull", time.Since(start), tap.Snapshot())
+	}
+}
+
+// pullAll fetches the manifest and reads every layer's bytes, the same shape
+// as the crane client in clients.go.
+func pullAll(ref string, opts []crane.Option) error {
+	img, err := crane.Pull(ref, opts...)
+	if err != nil {
+		return err
+	}
+	layers, err := img.Layers()
+	if err != nil {
+		return err
+	}
+	for _, l := range layers {
+		rc, err := l.Compressed()
+		if err != nil {
+			return err
+		}
+		_, cerr := io.Copy(io.Discard, rc)
+		rc.Close()
+		if cerr != nil {
+			return cerr
+		}
+	}
+	return nil
+}
+
+func buildImage(t *testing.T, sizesMB []float64) v1.Image {
+	t.Helper()
+	layers := make([]v1.Layer, 0, len(sizesMB))
+	for _, mb := range sizesMB {
+		l, err := random.Layer(int64(mb*1024*1024), types.DockerLayer)
+		if err != nil {
+			t.Fatalf("random layer: %v", err)
+		}
+		layers = append(layers, l)
+	}
+	img, err := mutate.AppendLayers(empty.Image, layers...)
+	if err != nil {
+		t.Fatalf("append layers: %v", err)
+	}
+	return img
+}
+
+func report(t *testing.T, out *os.File, profile, op string, wall time.Duration, snap map[string]map[string]int) {
+	t.Helper()
+	pds, hold, s3 := reqcount.Total(snap["pds"]), reqcount.Total(snap["hold"]), reqcount.Total(snap["s3"])
+	reg := reqcount.Total(snap["registry"])
+	t.Logf("%-6s %-4s %6dms  registry=%-3d pds=%-3d hold=%-3d s3=%-3d", profile, op, wall.Milliseconds(), reg, pds, hold, s3)
+	for _, b := range []string{"registry", "hold", "s3", "pds"} {
+		for _, k := range reqcount.Keys(snap[b]) {
+			t.Logf("    %-4s %4d  %s", b, snap[b][k], k)
+		}
+	}
+	if out != nil {
+		_ = json.NewEncoder(out).Encode(map[string]any{
+			"profile": profile, "op": op, "wall_ms": wall.Milliseconds(),
+			"registry": reg, "pds": pds, "hold": hold, "s3": s3, "detail": snap,
+		})
+	}
+}