diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index 2eee1e6..db4561c 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -97,6 +97,34 @@ type ProxyBlobStore struct { ctx *RegistryContext // All context and services holdURL string // Resolved HTTP URL for XRPC requests httpClient *http.Client + + // presigns memoizes the hold's presign answers for the life of this store, + // which is the life of one registry request (RoutingRepository.Blobs builds + // the store once per request under a sync.Once, so Stat and ServeBlob for + // the same HTTP request share this instance). + // + // One blob GET used to cost three identical round trips to the hold. + // distribution's blob handler calls Stat and then ServeBlob, and the + // notifications listener — installed unconditionally — calls Stat a third + // time after ServeBlob returns to build its BlobPulled event. A 22 layer + // pull was 66 com.atproto.sync.getBlob calls for 22 redirects. + // + // Keyed by method as well as digest because S3 SigV4 signs the HTTP verb: a + // URL presigned for HEAD answers 403 to a GET. Stat therefore presigns for + // the verb the client is actually going to send (see Stat), which is what + // makes ServeBlob's lookup a hit rather than a second call. + // + // Only successful presigns are remembered, so nothing here can change the + // error semantics of a later call. + presignMu sync.Mutex + presigns map[presignKey]presignedBlob +} + +// presignKey identifies a memoized presign. The method is part of the identity, +// not an attribute: presigned URLs are verb-bound. +type presignKey struct { + digest digest.Digest + method string } // NewProxyBlobStore creates a new proxy blob store @@ -171,9 +199,22 @@ func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribu return distribution.Descriptor{}, err } - method := "HEAD" + // Presign for the verb the client is about to send, not always for HEAD. + // distribution calls Stat immediately before ServeBlob, so on a blob GET the + // two want the same URL — but only if they asked for the same verb, because + // SigV4 signs it. Asking for GET here is what lets ServeBlob be free. + // + // Anything that is not a read (a push's existence check arrives as a PUT, + // POST or PATCH, and manifest verification stats blobs mid-PUT) falls back + // to HEAD. That is not just a default: the hold refuses method=PUT on the + // read path unless the caller has write access, and skips the size lookup + // for it, so Stat must never ask for a write verb. + method := http.MethodHead + if m, ok := ctx.Value(HTTPRequestMethod).(string); ok && (m == http.MethodGet || m == http.MethodHead) { + method = m + } - blob, err := p.getPresignedURL(ctx, method, dgst) + blob, err := p.presign(ctx, method, dgst) if err != nil { // Preserve an authorization verdict. distribution calls Stat before // ServeBlob on both GET and HEAD, so flattening everything to @@ -202,7 +243,28 @@ func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribu // No size in the response: the hold predates the field. Fall back to the // original behaviour and read Content-Length off the presigned URL, so a // new AppView keeps working against a hold that has not been upgraded. - req, err := http.NewRequestWithContext(ctx, method, blob.URL, nil) + // + // The probe is always a HEAD, and when the URL above was signed for GET it + // cannot be used for it: sending the HEAD to a GET-signed URL is a 403, and + // sending a GET instead would download the entire blob just to read a + // header. So ask the hold for a HEAD-signed URL — the one extra call is + // paid only by the old-hold path, and only for the first Stat of a digest. + probe := blob + if method != http.MethodHead { + probe, err = p.presign(ctx, http.MethodHead, dgst) + if err != nil { + return distribution.Descriptor{}, distribution.ErrBlobUnknown + } + if probe.Size != nil { + return distribution.Descriptor{ + Digest: dgst, + Size: *probe.Size, + MediaType: "application/octet-stream", + }, nil + } + } + + req, err := http.NewRequestWithContext(ctx, http.MethodHead, probe.URL, nil) if err != nil { return distribution.Descriptor{}, distribution.ErrBlobUnknown } @@ -242,7 +304,9 @@ func (p *ProxyBlobStore) Get(ctx context.Context, dgst digest.Digest) ([]byte, e method := "GET" - blob, err := p.getPresignedURL(ctx, method, dgst) + // Through the memo: on a blob GET, Stat already presigned for GET, so this + // is usually free too. + blob, err := p.presign(ctx, method, dgst) if err != nil { return nil, err } @@ -277,7 +341,9 @@ func (p *ProxyBlobStore) Open(ctx context.Context, dgst digest.Digest) (io.ReadS method := "GET" - blob, err := p.getPresignedURL(ctx, method, dgst) + // Through the memo: on a blob GET, Stat already presigned for GET, so this + // is usually free too. + blob, err := p.presign(ctx, method, dgst) if err != nil { return nil, err } @@ -366,7 +432,9 @@ func (p *ProxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r return err } - blob, err := p.getPresignedURL(ctx, r.Method, dgst) + // Normally free: Stat ran moments ago on this same store and presigned for + // this same verb, so this is a memo hit and the redirect costs no hold call. + blob, err := p.presign(ctx, r.Method, dgst) if err != nil { return err } @@ -466,6 +534,39 @@ type presignedBlob struct { Size *int64 } +// presign is getPresignedURL through the per-request memo: the first caller for +// a (digest, method) pair pays the hold round trip and every later caller in the +// same request is answered from memory. See the presigns field for why one +// request asks the same question up to three times. +// +// Failures are not remembered. A presign fails for reasons that belong to the +// call that made it (authorization, a missing blob, a hold that is down), and +// caching those would mean one call's verdict silently deciding another's. +func (p *ProxyBlobStore) presign(ctx context.Context, method string, dgst digest.Digest) (presignedBlob, error) { + key := presignKey{digest: dgst, method: method} + + p.presignMu.Lock() + blob, hit := p.presigns[key] + p.presignMu.Unlock() + if hit { + return blob, nil + } + + blob, err := p.getPresignedURL(ctx, method, dgst) + if err != nil { + return presignedBlob{}, err + } + + p.presignMu.Lock() + if p.presigns == nil { + p.presigns = make(map[presignKey]presignedBlob) + } + p.presigns[key] = blob + p.presignMu.Unlock() + + return blob, nil +} + // getPresignedURL asks the hold for a presigned URL for a blob operation, and // for reads gets the blob's size back with it. func (p *ProxyBlobStore) getPresignedURL(ctx context.Context, operation string, dgst digest.Digest) (presignedBlob, error) { diff --git a/pkg/appview/storage/proxy_blob_store_test.go b/pkg/appview/storage/proxy_blob_store_test.go index 1cb9449..5e5d1dc 100644 --- a/pkg/appview/storage/proxy_blob_store_test.go +++ b/pkg/appview/storage/proxy_blob_store_test.go @@ -3203,3 +3203,247 @@ func TestCreate_RefusesDuplicateID(t *testing.T) { t.Errorf("Expected 1 writer tracked in globalUploads, got %d", tracked) } } + +// presignRecorder is a fake hold that answers com.atproto.sync.getBlob and +// remembers every presign it was asked for, so a test can count hold calls and +// see which verb each one asked to sign. +type presignRecorder struct { + mu sync.Mutex + calls []string // ":", in order + s3URL string + reportSz *int64 // nil reproduces a hold older than the size field + status int // non-zero overrides the 200 +} + +func (h *presignRecorder) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + h.mu.Lock() + h.calls = append(h.calls, q.Get("method")+":"+q.Get("cid")) + h.mu.Unlock() + + if h.status != 0 { + w.WriteHeader(h.status) + return + } + + // Sign the verb into the URL the way SigV4 does, so a test can prove the + // client is handed a URL signed for the verb it is about to send. + body := map[string]any{ + "url": h.s3URL + "/blob?X-Amz-SignedMethod=" + q.Get("method"), + } + if h.reportSz != nil { + body["size"] = *h.reportSz + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(body) + } +} + +func (h *presignRecorder) seen() []string { + h.mu.Lock() + defer h.mu.Unlock() + return append([]string(nil), h.calls...) +} + +// newPresignStore wires a ProxyBlobStore to a presignRecorder. +func newPresignStore(t *testing.T, size *int64, status int) (*ProxyBlobStore, *presignRecorder) { + t.Helper() + + hold := &presignRecorder{s3URL: "https://s3.example.invalid", reportSz: size, status: status} + holdServer := httptest.NewServer(hold.handler()) + t.Cleanup(holdServer.Close) + + store := NewProxyBlobStore(&RegistryContext{ + DID: "did:plc:test", + HoldDID: "did:web:hold.example.com", + Repository: "test-repo", + ServiceToken: "test-service-token", + }) + store.holdURL = holdServer.URL + return store, hold +} + +// TestPresignMemo_OneHoldCallPerBlobRead is the point of the memo. distribution +// calls Stat, then ServeBlob, and then the notifications listener calls Stat +// again after ServeBlob returns — three identical questions per blob, which used +// to be three round trips to the hold. All three must now cost one, and the +// redirect must carry a URL signed for the verb the client actually sent. +func TestPresignMemo_OneHoldCallPerBlobRead(t *testing.T) { + for _, method := range []string{http.MethodGet, http.MethodHead} { + t.Run(method, func(t *testing.T) { + size := int64(4096) + store, hold := newPresignStore(t, &size, 0) + + dgst := digest.FromString("memo-blob-" + method) + ctx := context.WithValue(context.Background(), HTTPRequestMethod, method) + + desc, err := store.Stat(ctx, dgst) + if err != nil { + t.Fatalf("Stat() failed: %v", err) + } + if desc.Size != size { + t.Errorf("Stat size = %d, want %d", desc.Size, size) + } + + rec := httptest.NewRecorder() + req := httptest.NewRequest(method, "/v2/test-repo/blobs/"+dgst.String(), nil) + if err := store.ServeBlob(ctx, rec, req, dgst); err != nil { + t.Fatalf("ServeBlob() failed: %v", err) + } + if rec.Code != http.StatusTemporaryRedirect { + t.Fatalf("ServeBlob status = %d, want %d", rec.Code, http.StatusTemporaryRedirect) + } + if loc := rec.Header().Get("Location"); !strings.Contains(loc, "X-Amz-SignedMethod="+method) { + t.Errorf("redirect URL %q is not signed for %s", loc, method) + } + + // The notifications listener's post-ServeBlob Stat, which runs on a + // WithoutCancel copy of the same context. + if _, err := store.Stat(context.WithoutCancel(ctx), dgst); err != nil { + t.Fatalf("listener Stat() failed: %v", err) + } + + calls := hold.seen() + if len(calls) != 1 { + t.Fatalf("expected exactly 1 hold presign for a %s, got %d: %v", method, len(calls), calls) + } + if calls[0] != method+":"+dgst.String() { + t.Errorf("hold was asked for %q, want %q", calls[0], method+":"+dgst.String()) + } + }) + } +} + +// TestPresignMemo_KeyedByDigest proves the memo is not a single slot: two blobs +// read in the same request each get their own presign. +func TestPresignMemo_KeyedByDigest(t *testing.T) { + size := int64(10) + store, hold := newPresignStore(t, &size, 0) + + ctx := context.WithValue(context.Background(), HTTPRequestMethod, http.MethodGet) + first := digest.FromString("blob-one") + second := digest.FromString("blob-two") + + for _, dgst := range []digest.Digest{first, first, second, second} { + if _, err := store.Stat(ctx, dgst); err != nil { + t.Fatalf("Stat(%s) failed: %v", dgst, err) + } + } + + want := []string{"GET:" + first.String(), "GET:" + second.String()} + if got := hold.seen(); len(got) != 2 || got[0] != want[0] || got[1] != want[1] { + t.Errorf("hold calls = %v, want %v", got, want) + } +} + +// TestPresignMemo_PushStatPresignsHead pins the constraint the memo must not +// break: the hold refuses method=PUT on the read path unless the caller has +// write access, so a push's existence check (Docker HEADs the blob inside a +// request whose method is POST/PUT/PATCH) must still ask for a HEAD. +func TestPresignMemo_PushStatPresignsHead(t *testing.T) { + for _, method := range []string{http.MethodPut, http.MethodPost, http.MethodPatch, ""} { + name := method + if name == "" { + name = "no-method-in-context" + } + t.Run(name, func(t *testing.T) { + size := int64(7) + store, hold := newPresignStore(t, &size, 0) + + ctx := context.Background() + if method != "" { + ctx = context.WithValue(ctx, HTTPRequestMethod, method) + } + + dgst := digest.FromString("push-existence-check") + if _, err := store.Stat(ctx, dgst); err != nil { + t.Fatalf("Stat() failed: %v", err) + } + + calls := hold.seen() + if len(calls) != 1 || calls[0] != "HEAD:"+dgst.String() { + t.Errorf("hold calls = %v, want one HEAD presign", calls) + } + }) + } +} + +// TestPresignMemo_MissingSizeProbesWithHead covers the old-hold fallback under +// the memo. Stat presigned for GET because the client is doing a GET, but the +// size probe is still a HEAD, and it must go to a HEAD-signed URL — sending a +// GET there would download the whole blob just to read Content-Length. +func TestPresignMemo_MissingSizeProbesWithHead(t *testing.T) { + const blobSize = 8192 + + var mu sync.Mutex + var s3 []string // " " + + s3Server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + s3 = append(s3, r.Method+" "+r.URL.Query().Get("X-Amz-SignedMethod")) + mu.Unlock() + w.Header().Set("Content-Length", strconv.Itoa(blobSize)) + w.WriteHeader(http.StatusOK) + })) + defer s3Server.Close() + + // An old hold: a url and nothing else. + store, hold := newPresignStore(t, nil, 0) + hold.s3URL = s3Server.URL + + ctx := context.WithValue(context.Background(), HTTPRequestMethod, http.MethodGet) + dgst := digest.FromString("unsized-blob-on-a-get") + + desc, err := store.Stat(ctx, dgst) + if err != nil { + t.Fatalf("Stat() failed: %v", err) + } + if desc.Size != blobSize { + t.Errorf("size = %d, want %d from Content-Length", desc.Size, blobSize) + } + + mu.Lock() + got := append([]string(nil), s3...) + mu.Unlock() + if len(got) != 1 || got[0] != "HEAD HEAD" { + t.Fatalf("S3 requests = %v, want exactly one HEAD to a HEAD-signed URL", got) + } + + // The GET presign Stat started with is still memoized, so the ServeBlob that + // follows costs nothing and still gets a GET-signed URL. + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/v2/test-repo/blobs/"+dgst.String(), nil) + if err := store.ServeBlob(ctx, rec, req, dgst); err != nil { + t.Fatalf("ServeBlob() failed: %v", err) + } + if loc := rec.Header().Get("Location"); !strings.Contains(loc, "X-Amz-SignedMethod=GET") { + t.Errorf("redirect URL %q is not signed for GET", loc) + } + + calls := hold.seen() + want := []string{"GET:" + dgst.String(), "HEAD:" + dgst.String()} + if len(calls) != 2 || calls[0] != want[0] || calls[1] != want[1] { + t.Errorf("hold calls = %v, want %v (the GET presign plus the HEAD size probe)", calls, want) + } +} + +// TestPresignMemo_DoesNotCacheFailures keeps the memo out of the error path: a +// hold 404 is still ErrBlobUnknown, on the first call and on every one after. +func TestPresignMemo_DoesNotCacheFailures(t *testing.T) { + store, hold := newPresignStore(t, nil, http.StatusNotFound) + + ctx := context.WithValue(context.Background(), HTTPRequestMethod, http.MethodGet) + dgst := digest.FromString("missing-blob") + + for i := range 2 { + if _, err := store.Stat(ctx, dgst); !errors.Is(err, distribution.ErrBlobUnknown) { + t.Fatalf("Stat() #%d = %v, want ErrBlobUnknown", i+1, err) + } + } + + // Both attempts reached the hold: nothing was answered from a poisoned memo. + if calls := hold.seen(); len(calls) != 2 { + t.Errorf("hold calls = %v, want 2 (failures are never memoized)", calls) + } +}