diff --git a/docs/PERFORMANCE_BACKLOG.md b/docs/PERFORMANCE_BACKLOG.md index 0d8704b..070747c 100644 --- a/docs/PERFORMANCE_BACKLOG.md +++ b/docs/PERFORMANCE_BACKLOG.md @@ -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 diff --git a/internal/reqcount/reqcount.go b/internal/reqcount/reqcount.go new file mode 100644 index 0000000..f20dd0c --- /dev/null +++ b/internal/reqcount/reqcount.go @@ -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" +} diff --git a/internal/testharness/harness.go b/internal/testharness/harness.go index e43fdf0..028922a 100644 --- a/internal/testharness/harness.go +++ b/internal/testharness/harness.go @@ -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) diff --git a/pkg/testpds/server.go b/pkg/testpds/server.go index 5808a2e..73f74d9 100644 --- a/pkg/testpds/server.go +++ b/pkg/testpds/server.go @@ -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://")) diff --git a/test/integration/bench_real_test.go b/test/integration/bench_real_test.go new file mode 100644 index 0000000..d693115 --- /dev/null +++ b/test/integration/bench_real_test.go @@ -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, + }) + } +}