diff --git a/docs/HOLD_XRPC_ENDPOINTS.md b/docs/HOLD_XRPC_ENDPOINTS.md index 6cde6fb..a7d1564 100644 --- a/docs/HOLD_XRPC_ENDPOINTS.md +++ b/docs/HOLD_XRPC_ENDPOINTS.md @@ -33,6 +33,39 @@ This document lists all XRPC endpoints implemented in the Hold service (`pkg/hol |----------|--------|-------------| | `/xrpc/com.atproto.sync.getBlob` | GET/HEAD | Get blob (routes OCI vs ATProto) | +#### getBlob response shape + +The endpoint routes on the `cid` parameter and answers differently per branch. + +**ATProto blob** (`cid` is a CID): 307 redirect to a presigned URL. + +**OCI blob** (`cid` starts with `sha256:`): JSON. + +```json +{ + "url": "https://s3.example.com/...?X-Amz-Signature=...", + "size": 27129344 +} +``` + +| Field | Type | Notes | +|---|---|---| +| `url` | string | Presigned URL for the requested `method` (GET, HEAD or PUT). Always present on a 200. | +| `size` | int64 | Blob size in bytes. Present on GET and HEAD only, and omitted when the size could not be resolved. | + +`size` lets the AppView build an OCI descriptor without a second round trip: `ProxyBlobStore.Stat` used to fetch a presigned HEAD URL here and then HEAD it against S3 purely to read `Content-Length`. It is resolved from the hold's records index (any layer with an `io.atcr.hold.layer` record) and falls back to a `HeadObject` (image config blobs, and layers whose manifest notification has not landed yet). + +The field is additive. A client that reads only `url` behaves exactly as before, and an AppView that gets no `size` falls back to HEADing the presigned URL. `size` is never sent for a PUT presign: the object does not exist yet. + +On GET and HEAD, a blob that is not in storage is answered with **404** and a JSON error body instead of a presigned URL: + +```json +{ + "error": "BlobUnknown", + "message": "blob not found" +} +``` + ### Owner/Crew Admin Required | Endpoint | Method | Description | diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index 1210c71..d66c8b8 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -143,7 +143,7 @@ func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribu method := "HEAD" - url, err := p.getPresignedURL(ctx, method, dgst) + blob, err := p.getPresignedURL(ctx, method, dgst) if err != nil { // Preserve an authorization verdict. distribution calls Stat before // ServeBlob on both GET and HEAD, so flattening everything to @@ -157,8 +157,22 @@ func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribu return distribution.Descriptor{}, distribution.ErrBlobUnknown } - // Make HEAD request to presigned URL - req, err := http.NewRequestWithContext(ctx, method, url, nil) + if blob.Size != nil { + // The hold reported the size, so the descriptor is complete and the + // blob's bytes never have to be touched. This is the whole point of the + // field: Stat is called before every GET and HEAD of a blob, and the + // round trip below was buying nothing but Content-Length. + return distribution.Descriptor{ + Digest: dgst, + Size: *blob.Size, + MediaType: "application/octet-stream", + }, nil + } + + // 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) if err != nil { return distribution.Descriptor{}, distribution.ErrBlobUnknown } @@ -198,13 +212,13 @@ func (p *ProxyBlobStore) Get(ctx context.Context, dgst digest.Digest) ([]byte, e method := "GET" - url, err := p.getPresignedURL(ctx, method, dgst) + blob, err := p.getPresignedURL(ctx, method, dgst) if err != nil { return nil, err } // Download the blob from presigned URL - req, err := http.NewRequestWithContext(ctx, method, url, nil) + req, err := http.NewRequestWithContext(ctx, method, blob.URL, nil) if err != nil { return nil, err } @@ -233,13 +247,13 @@ func (p *ProxyBlobStore) Open(ctx context.Context, dgst digest.Digest) (io.ReadS method := "GET" - url, err := p.getPresignedURL(ctx, method, dgst) + blob, err := p.getPresignedURL(ctx, method, dgst) if err != nil { return nil, err } // Download the blob from presigned URL - req, err := http.NewRequestWithContext(ctx, method, url, nil) + req, err := http.NewRequestWithContext(ctx, method, blob.URL, nil) if err != nil { return nil, err } @@ -319,13 +333,13 @@ func (p *ProxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r return err } - url, err := p.getPresignedURL(ctx, r.Method, dgst) + blob, err := p.getPresignedURL(ctx, r.Method, dgst) if err != nil { return err } // Redirect to presigned URL - http.Redirect(w, r, url, http.StatusTemporaryRedirect) + http.Redirect(w, r, blob.URL, http.StatusTemporaryRedirect) return nil } @@ -388,8 +402,21 @@ func (p *ProxyBlobStore) Resume(ctx context.Context, id string) (distribution.Bl return writer, nil } -// getPresignedURL returns the XRPC endpoint URL for blob operations -func (p *ProxyBlobStore) getPresignedURL(ctx context.Context, operation string, dgst digest.Digest) (string, error) { +// presignedBlob is the hold's answer to a getBlob presign request. +type presignedBlob struct { + // URL is the presigned S3 URL for the requested operation. + URL string + // Size is the blob's byte size as reported by the hold, or nil when the + // hold did not report one. A pointer rather than a plain int64 so that + // "the hold said nothing" stays distinguishable from "the hold said zero": + // a hold older than the size field reports nothing, and callers must fall + // back rather than believe in a zero-length blob. + Size *int64 +} + +// 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) { // Use XRPC endpoint: /xrpc/com.atproto.sync.getBlob?did={userDID}&cid={digest} // The 'did' parameter is the USER's DID (whose blob we're fetching), not the hold service DID // Per migration doc: hold accepts OCI digest directly as cid parameter (checks for sha256: prefix) @@ -398,44 +425,55 @@ func (p *ProxyBlobStore) getPresignedURL(ctx context.Context, operation string, req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil) if err != nil { - return "", fmt.Errorf("failed to create request: %w", err) + return presignedBlob{}, fmt.Errorf("failed to create request: %w", err) } resp, err := p.doAuthenticatedRequest(ctx, req) if err != nil { // Don't wrap errcode errors - return them directly if _, ok := err.(errcode.Error); ok { - return "", err + return presignedBlob{}, err } - return "", fmt.Errorf("failed to get presigned URL: %w", err) + return presignedBlob{}, fmt.Errorf("failed to get presigned URL: %w", err) } defer resp.Body.Close() if resp.StatusCode == http.StatusForbidden && p.ctx.Anonymous { // Stale local captain cache let an anonymous request through, but the // hold says private. Surface a 401 so the client re-authenticates. - return "", errcode.ErrorCodeUnauthorized.WithMessage("authentication required") + return presignedBlob{}, errcode.ErrorCodeUnauthorized.WithMessage("authentication required") + } + + if resp.StatusCode == http.StatusNotFound { + // The hold checked its storage and the blob is not there. Return the + // sentinel rather than a generic failure: Stat passes it through as + // blob-unknown, and Get and Open hand their callers the error the + // distribution interface documents for a missing blob. + return presignedBlob{}, distribution.ErrBlobUnknown } if resp.StatusCode != http.StatusOK { bodyBytes, _ := io.ReadAll(resp.Body) - return "", fmt.Errorf("hold service returned error: status %d, body: %s", resp.StatusCode, string(bodyBytes)) + return presignedBlob{}, fmt.Errorf("hold service returned error: status %d, body: %s", resp.StatusCode, string(bodyBytes)) } - // Parse JSON response to get presigned HEAD URL + // Parse JSON response to get the presigned URL, and the size when the hold + // reports one. Size is a pointer so an older hold, which sends no size + // field at all, is not read as a zero-length blob. var result struct { - URL string `json:"url"` + URL string `json:"url"` + Size *int64 `json:"size"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return "", fmt.Errorf("failed to parse hold service response: %w", err) + return presignedBlob{}, fmt.Errorf("failed to parse hold service response: %w", err) } if result.URL == "" { - return "", fmt.Errorf("hold service returned empty URL") + return presignedBlob{}, fmt.Errorf("hold service returned empty URL") } - slog.Debug("Got presigned HEAD URL from hold service", "component", "proxy_blob_store", "url", result.URL) - return result.URL, nil + slog.Debug("Got presigned URL from hold service", "component", "proxy_blob_store", "url", result.URL, "size_reported", result.Size != nil) + return presignedBlob{URL: result.URL, Size: result.Size}, nil } // startMultipartUpload initiates a multipart upload via XRPC initiateUpload endpoint diff --git a/pkg/appview/storage/proxy_blob_store_test.go b/pkg/appview/storage/proxy_blob_store_test.go index fb59a18..4607225 100644 --- a/pkg/appview/storage/proxy_blob_store_test.go +++ b/pkg/appview/storage/proxy_blob_store_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/base64" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -2099,3 +2100,156 @@ func TestMultipartEndpoints_CorrectURLs(t *testing.T) { }) } } + +// TestStat_UsesHoldReportedSize pins that a hold which reports the size ends +// the Stat there: the descriptor is built from it and S3 is never touched. +func TestStat_UsesHoldReportedSize(t *testing.T) { + cases := []struct { + name string + size int64 + }{ + {"normal blob", 4096}, + // Zero is a real size, and it must not be read as "the hold said + // nothing". That is why the field is decoded into a pointer. + {"empty blob", 0}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var s3Requests int64 + var mu sync.Mutex + + s3Server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + mu.Lock() + s3Requests++ + mu.Unlock() + w.WriteHeader(http.StatusOK) + })) + defer s3Server.Close() + + holdServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "url": s3Server.URL + "/blob?X-Amz-Signature=fake", + "size": tc.size, + }) + })) + defer 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 + + dgst := digest.FromString("sized-blob") + desc, err := store.Stat(context.Background(), dgst) + if err != nil { + t.Fatalf("Stat() failed: %v", err) + } + + if desc.Size != tc.size { + t.Errorf("Expected size %d from the hold, got %d", tc.size, desc.Size) + } + if desc.Digest != dgst { + t.Errorf("Expected digest %s, got %s", dgst, desc.Digest) + } + + mu.Lock() + defer mu.Unlock() + if s3Requests != 0 { + t.Errorf("Expected no request to S3 when the hold reports the size, got %d", s3Requests) + } + }) + } +} + +// TestStat_FallsBackToHeadWhenSizeAbsent pins the rollout direction that +// matters: a new AppView against a hold that predates the size field must keep +// working, by HEADing the presigned URL exactly as it always did. +func TestStat_FallsBackToHeadWhenSizeAbsent(t *testing.T) { + const blobSize = 8192 + + var mu sync.Mutex + var s3Methods []string + + s3Server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + s3Methods = append(s3Methods, r.Method) + mu.Unlock() + + w.Header().Set("Content-Length", strconv.Itoa(blobSize)) + w.WriteHeader(http.StatusOK) + })) + defer s3Server.Close() + + holdServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + // An old hold answers with the url and nothing else. + json.NewEncoder(w).Encode(map[string]string{ + "url": s3Server.URL + "/blob?X-Amz-Signature=fake", + }) + })) + defer 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 + + desc, err := store.Stat(context.Background(), digest.FromString("unsized-blob")) + if err != nil { + t.Fatalf("Stat() failed: %v", err) + } + + if desc.Size != blobSize { + t.Errorf("Expected size %d from Content-Length, got %d", blobSize, desc.Size) + } + + mu.Lock() + defer mu.Unlock() + if len(s3Methods) != 1 { + t.Fatalf("Expected exactly 1 request to S3, got %d: %v", len(s3Methods), s3Methods) + } + if s3Methods[0] != http.MethodHead { + t.Errorf("Expected a HEAD to the presigned URL, got %s", s3Methods[0]) + } +} + +// TestStat_HoldNotFoundIsBlobUnknown pins the error mapping for a hold that +// checked storage and found nothing. It must read as a missing blob, not as a +// server error and not as an authentication problem. +func TestStat_HoldNotFoundIsBlobUnknown(t *testing.T) { + holdServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + json.NewEncoder(w).Encode(map[string]string{ + "error": "BlobUnknown", + "message": "blob not found", + }) + })) + defer 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 + + _, err := store.Stat(context.Background(), digest.FromString("missing-blob")) + if !errors.Is(err, distribution.ErrBlobUnknown) { + t.Errorf("Expected ErrBlobUnknown for a hold 404, got %v", err) + } + + // Get and Open get the same sentinel rather than a generic failure. + if _, err := store.Get(context.Background(), digest.FromString("missing-blob")); !errors.Is(err, distribution.ErrBlobUnknown) { + t.Errorf("Expected ErrBlobUnknown from Get for a hold 404, got %v", err) + } +} diff --git a/pkg/hold/pds/records.go b/pkg/hold/pds/records.go index 7d3dc09..a9caf69 100644 --- a/pkg/hold/pds/records.go +++ b/pkg/hold/pds/records.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "database/sql" + "errors" "fmt" "log/slog" "strings" @@ -43,6 +44,7 @@ CREATE TABLE IF NOT EXISTS records ( ); CREATE INDEX IF NOT EXISTS idx_records_collection_rkey ON records(collection, rkey); CREATE INDEX IF NOT EXISTS idx_records_collection_did ON records(collection, did); +CREATE INDEX IF NOT EXISTS idx_records_digest ON records(digest); ` // NewRecordsIndex creates or opens a records index @@ -420,6 +422,39 @@ func splitStatements(sql string) []string { return out } +// SizeForDigest returns the indexed byte size of a blob digest. +// +// Only layer records carry a digest (see extractLayerFieldsFromRecord), so this +// answers for anything that has been through a manifest notification. The +// second return distinguishes "the index has no size for this digest" from a +// genuine zero. A digest can appear on several layer records (the same layer +// referenced by more than one manifest, or pushed by more than one user); they +// all describe the same content-addressed bytes, so any one of them answers. +// +// idx_records_digest keeps this a lookup rather than a scan. It is created by +// the schema statements, which run on every open, so an existing database picks +// the index up without a rebuild. +func (ri *RecordsIndex) SizeForDigest(digest string) (int64, bool, error) { + if digest == "" { + return 0, false, nil + } + + var size sql.NullInt64 + err := ri.db.QueryRow(` + SELECT size FROM records WHERE digest = ? AND size IS NOT NULL LIMIT 1 + `, digest).Scan(&size) + if errors.Is(err, sql.ErrNoRows) { + return 0, false, nil + } + if err != nil { + return 0, false, fmt.Errorf("failed to look up size for digest: %w", err) + } + if !size.Valid { + return 0, false, nil + } + return size.Int64, true, nil +} + // QuotaForDID returns unique blob count and total size for a single DID in a collection. // Uses SQL aggregation over the denormalized digest/size columns. func (ri *RecordsIndex) QuotaForDID(collection, did string) (uniqueBlobs int, totalSize int64, err error) { diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index 9b96f8b..3119983 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -1276,6 +1276,31 @@ func (h *XRPCHandler) handleGetOCIBlob(w http.ResponseWriter, r *http.Request, d } } + // Resolve the blob's size for reads so the caller does not have to go to S3 + // itself just to fill in a descriptor. A read presign is the AppView's Stat, + // and Stat used to cost a second round trip: the presigned HEAD URL was + // fetched here and then HEADed against S3 purely to read Content-Length. + // + // Writes are excluded: the object being presigned for PUT does not exist yet. + var size int64 + var sizeKnown bool + if operation != http.MethodPut { + var err error + size, sizeKnown, err = h.ociBlobSize(r.Context(), digest) + if errors.Is(err, errOCIBlobMissing) { + // Storage says the object is not there, so there is nothing to hand + // out a read capability for. Say so here rather than presigning a + // URL whose only possible answer is a 404 from S3. + slog.Debug("OCI blob not present in storage", "digest", digest, "operation", operation) + render.Status(r, http.StatusNotFound) + render.JSON(w, r, map[string]string{ + "error": "BlobUnknown", + "message": "blob not found", + }) + return + } + } + // Generate presigned URL (use empty DID for content-addressed storage) presignedURL, err := h.GetPresignedURL(r.Context(), operation, digest, "") if err != nil { @@ -1290,12 +1315,64 @@ func (h *XRPCHandler) handleGetOCIBlob(w http.ResponseWriter, r *http.Request, d slog.Debug("Returning presigned URL for OCI blob", "operation", operation, "digest", digest, - "url", presignedURL) + "url", presignedURL, + "size_known", sizeKnown) - // Return JSON response with presigned URL (AppView expects this format) - render.JSON(w, r, map[string]string{ + // Return JSON response with presigned URL (AppView expects this format). + // + // "size" is additive: a client that only reads "url" behaves exactly as + // before, and it is omitted entirely when the size could not be resolved so + // that absent stays distinguishable from zero. + response := map[string]any{ "url": presignedURL, - }) + } + if sizeKnown { + response["size"] = size + } + render.JSON(w, r, response) +} + +// errOCIBlobMissing reports that storage was asked about an OCI blob and +// answered that it does not exist. It is distinct from a lookup that merely +// failed: a failed lookup leaves the size unknown, a missing object is a 404. +var errOCIBlobMissing = errors.New("oci blob not found in storage") + +// ociBlobSize resolves the byte size of an OCI blob, cheapest source first. +// +// The records index knows the size of every layer that has been through a +// manifest notification, and answers from SQLite without touching the network. +// Anything else (an image config blob, which never gets a layer record, or a +// layer whose manifest notification has not landed yet) falls back to a +// HeadObject against S3. +// +// A lookup failure that is not a missing object leaves the size unknown rather +// than failing the request: the caller can still be handed a presigned URL, and +// an AppView that gets no size falls back to HEADing that URL as it always did. +func (h *XRPCHandler) ociBlobSize(ctx context.Context, digest string) (int64, bool, error) { + if idx := h.pds.RecordsIndex(); idx != nil { + size, ok, err := idx.SizeForDigest(digest) + if err != nil { + slog.Warn("Records index size lookup failed", "error", err, "digest", digest) + } else if ok { + return size, true, nil + } + } + + if h.s3Service.Client == nil { + // No S3 client: GetPresignedURL falls back to the XRPC proxy, and there + // is nothing to ask about the object. + return 0, false, nil + } + + size, err := h.s3Service.Stat(ctx, s3.BlobPath(digest)) + if err != nil { + if s3.IsNotFound(err) { + return 0, false, errOCIBlobMissing + } + slog.Warn("Failed to stat OCI blob for size", "error", err, "digest", digest) + return 0, false, nil + } + return size, true, nil } // handleGetATProtoBlob handles standard ATProto blob requests diff --git a/pkg/hold/pds/xrpc_blob_presign_test.go b/pkg/hold/pds/xrpc_blob_presign_test.go index 3074d2a..8c5a5b6 100644 --- a/pkg/hold/pds/xrpc_blob_presign_test.go +++ b/pkg/hold/pds/xrpc_blob_presign_test.go @@ -390,6 +390,12 @@ func TestGetBlob_ReadOperationsStillWork(t *testing.T) { t.Run(tc.name, func(t *testing.T) { handler, mockS3Client, _ := setupTestXRPCHandlerWithMockS3(t) + // An OCI read presign resolves the blob's size before it hands out + // a URL, and answers 404 for an object that is not in storage. + if strings.HasPrefix(tc.cid, "sha256:") { + seedOCIBlob(t, handler, mockS3Client, tc.cid, 512) + } + params := map[string]string{ "did": testHoldDID, "cid": tc.cid, diff --git a/pkg/hold/pds/xrpc_blob_size_test.go b/pkg/hold/pds/xrpc_blob_size_test.go new file mode 100644 index 0000000..4589da0 --- /dev/null +++ b/pkg/hold/pds/xrpc_blob_size_test.go @@ -0,0 +1,270 @@ +package pds + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "atcr.io/pkg/atproto" + "atcr.io/pkg/s3" +) + +// getBlob answers a read presign with the blob's size so the AppView can build +// an OCI descriptor without a second round trip. Before that field existed, +// every Stat (which distribution performs ahead of every blob GET and HEAD) +// fetched a presigned HEAD URL here and then HEADed S3 through it purely to +// read Content-Length. +// +// The size has two sources, and which one answers is the whole point: the +// records index knows every layer that has been through a manifest +// notification and costs a local SQLite lookup, while an image config blob +// (which never gets a layer record) has to be asked about with a HeadObject. +// These tests pin both, and pin that an object storage says nothing about is +// answered with a 404 rather than a presigned URL that could only 404 later. + +// ociBlobKey returns the S3 key a content-addressed OCI blob lands on for the +// handler's S3 service, prefix included. +func ociBlobKey(handler *XRPCHandler, digest string) string { + key := strings.TrimPrefix(s3.BlobPath(digest), "/") + if handler.s3Service.PathPrefix != "" { + key = handler.s3Service.PathPrefix + "/" + key + } + return key +} + +// seedOCIBlob puts size bytes at the digest's content-addressed key, so the +// handler's HeadObject finds an object there. +func seedOCIBlob(t *testing.T, handler *XRPCHandler, mock *s3.MockS3Client, digest string, size int) { + t.Helper() + mock.Objects[ociBlobKey(handler, digest)] = make([]byte, size) +} + +// getOCIBlobJSON performs a getBlob read for an OCI digest and decodes the +// JSON body. size is nil when the response carried no size field. +func getOCIBlobJSON(t *testing.T, handler *XRPCHandler, digest, method string) (code int, url string, size *int64) { + t.Helper() + + params := map[string]string{ + "did": testHoldDID, + "cid": digest, + } + if method != "" { + params["method"] = method + } + + w := httptest.NewRecorder() + handler.HandleGetBlob(w, makeXRPCGetRequest(atproto.SyncGetBlob, params)) + + var body struct { + URL string `json:"url"` + Size *int64 `json:"size"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("Failed to parse getBlob response %q: %v", w.Body.String(), err) + } + return w.Code, body.URL, body.Size +} + +// setupTestXRPCHandlerWithRecordsIndex is setupTestXRPCHandlerWithMockS3 with a +// file-backed database, which is what gives the PDS a records index: the +// in-memory setups leave HoldPDS.recordsIndex nil, so they can only exercise +// the HeadObject fallback. +func setupTestXRPCHandlerWithRecordsIndex(t *testing.T) (*XRPCHandler, *s3.MockS3Client, context.Context) { + t.Helper() + + ctx := context.Background() + tmpDir := t.TempDir() + + keyPath := filepath.Join(tmpDir, "signing-key") + if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil { + t.Fatalf("Failed to copy shared signing key: %v", err) + } + + pds, err := NewHoldPDS(ctx, testHoldDID, "https://hold.example.com", "https://atcr.io", + filepath.Join(tmpDir, "hold.db"), keyPath, false) + if err != nil { + t.Fatalf("Failed to create test PDS: %v", err) + } + t.Cleanup(func() { pds.Close() }) + + // Bootstrap writes to stdout; drain it the way the other setups do. + oldStdout := os.Stdout + r, w, _ := os.Pipe() + os.Stdout = w + err = pds.Bootstrap(ctx, nil, BootstrapConfig{OwnerDID: testHoldOwnerDID, Public: true}) + w.Close() + os.Stdout = oldStdout + io.ReadAll(r) + if err != nil { + t.Fatalf("Failed to bootstrap PDS: %v", err) + } + + if pds.RecordsIndex() == nil { + t.Fatal("Expected a records index on a file-backed PDS") + } + + mockS3Client := s3.NewMockS3Client("https://mock-s3.example.com") + handler := NewXRPCHandler(pds, s3.S3Service{ + Client: mockS3Client, + Bucket: "test-bucket", + PathPrefix: "test-prefix", + }, nil, &mockPDSClient{}, nil) + + return handler, mockS3Client, ctx +} + +// TestGetBlob_OCISizeFromRecordsIndex pins that a layer with a record is +// answered from the index. Nothing is seeded into S3: if the handler fell +// through to HeadObject it would find no object and answer 404, so a 200 with +// the indexed size is proof the index answered. +func TestGetBlob_OCISizeFromRecordsIndex(t *testing.T) { + for _, method := range []string{"", "GET", "HEAD"} { + name := method + if name == "" { + name = "default" + } + t.Run(name, func(t *testing.T) { + handler, mockS3Client, _ := setupTestXRPCHandlerWithRecordsIndex(t) + + const indexedSize = int64(4096) + err := handler.pds.RecordsIndex().IndexRecord( + atproto.LayerCollection, "3lktestrkey", "bafyreitestcid", + "did:plc:pusher", testOCIDigest, indexedSize) + if err != nil { + t.Fatalf("Failed to index layer record: %v", err) + } + + code, url, size := getOCIBlobJSON(t, handler, testOCIDigest, method) + + if code != http.StatusOK { + t.Fatalf("Expected 200, got %d", code) + } + if url == "" { + t.Error("Expected a presigned url in the response") + } + if size == nil { + t.Fatal("Expected a size field in the response") + } + if *size != indexedSize { + t.Errorf("Expected size %d from the index, got %d", indexedSize, *size) + } + if n := len(mockS3Client.RealHeadObjectCalls); n != 0 { + t.Errorf("Expected no HeadObject when the index answers, got %d: %+v", + n, mockS3Client.RealHeadObjectCalls) + } + }) + } +} + +// TestGetBlob_OCISizeFromHeadObject pins the fallback. An image config blob +// never gets a layer record, and neither does a layer whose manifest +// notification has not landed yet, so the size has to come from storage. +func TestGetBlob_OCISizeFromHeadObject(t *testing.T) { + handler, mockS3Client, _ := setupTestXRPCHandlerWithRecordsIndex(t) + + const objectSize = 1234 + seedOCIBlob(t, handler, mockS3Client, testOCIDigest, objectSize) + + code, url, size := getOCIBlobJSON(t, handler, testOCIDigest, "GET") + + if code != http.StatusOK { + t.Fatalf("Expected 200, got %d", code) + } + if url == "" { + t.Error("Expected a presigned url in the response") + } + if size == nil { + t.Fatal("Expected a size field in the response") + } + if *size != objectSize { + t.Errorf("Expected size %d from HeadObject, got %d", objectSize, *size) + } + if n := len(mockS3Client.RealHeadObjectCalls); n != 1 { + t.Errorf("Expected exactly 1 HeadObject when the index misses, got %d", n) + } +} + +// TestGetBlob_OCIMissingObjectReturns404 pins that a blob storage has never +// heard of is reported as missing here. Handing back a presigned URL for it +// only moves the 404 to S3 and costs the caller another round trip to find +// out. +func TestGetBlob_OCIMissingObjectReturns404(t *testing.T) { + for _, method := range []string{"GET", "HEAD"} { + t.Run(method, func(t *testing.T) { + handler, mockS3Client, _ := setupTestXRPCHandlerWithRecordsIndex(t) + + code, url, size := getOCIBlobJSON(t, handler, testOCIDigest, method) + + if code != http.StatusNotFound { + t.Fatalf("Expected 404 for a blob that is not in storage, got %d", code) + } + if url != "" { + t.Errorf("Expected no presigned url for a missing blob, got %q", url) + } + if size != nil { + t.Errorf("Expected no size for a missing blob, got %d", *size) + } + if n := len(mockS3Client.GetObjectCalls) + len(mockS3Client.HeadObjectCalls); n != 0 { + t.Errorf("Expected no presign for a missing blob, got %d call(s)", n) + } + }) + } +} + +// TestGetBlob_OCIPutSkipsSizeLookup pins that the write path is left alone. +// The object being presigned for PUT does not exist yet, so asking storage +// about it would cost a round trip only to answer "missing", and a 404 would +// make the upload impossible. +func TestGetBlob_OCIPutSkipsSizeLookup(t *testing.T) { + handler, mockS3Client, _ := setupTestXRPCHandlerWithRecordsIndex(t) + + w := httptest.NewRecorder() + handler.HandleGetBlob(w, authorizeAs(t, putBlobRequest(testOCIDigest), testHoldOwnerDID)) + + if w.Code != http.StatusOK { + t.Fatalf("Expected 200 for an authorized PUT presign, got %d (body: %s)", w.Code, w.Body.String()) + } + + var body struct { + URL string `json:"url"` + Size *int64 `json:"size"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("Failed to parse response %q: %v", w.Body.String(), err) + } + if body.URL == "" { + t.Error("Expected a presigned url for an authorized PUT") + } + if body.Size != nil { + t.Errorf("Expected no size on a PUT presign, got %d", *body.Size) + } + if n := len(mockS3Client.RealHeadObjectCalls); n != 0 { + t.Errorf("Expected no HeadObject on the PUT path, got %d", n) + } +} + +// TestGetBlob_ATProtoUnaffectedBySizeLookup pins that the ATProto blob branch +// is untouched: it still redirects, and it is not made to pay for a size it +// does not report. +func TestGetBlob_ATProtoUnaffectedBySizeLookup(t *testing.T) { + handler, mockS3Client, _ := setupTestXRPCHandlerWithRecordsIndex(t) + + w := httptest.NewRecorder() + handler.HandleGetBlob(w, makeXRPCGetRequest(atproto.SyncGetBlob, map[string]string{ + "did": testHoldDID, + "cid": testBlobCID, + })) + + if w.Code != http.StatusTemporaryRedirect { + t.Fatalf("Expected 307 for an ATProto blob, got %d (body: %s)", w.Code, w.Body.String()) + } + if n := len(mockS3Client.RealHeadObjectCalls); n != 0 { + t.Errorf("Expected no HeadObject on the ATProto branch, got %d", n) + } +} diff --git a/pkg/hold/pds/xrpc_test.go b/pkg/hold/pds/xrpc_test.go index c0230e5..92bffbc 100644 --- a/pkg/hold/pds/xrpc_test.go +++ b/pkg/hold/pds/xrpc_test.go @@ -2327,6 +2327,9 @@ func TestHandleGetBlob_MockS3_SHA256Digest(t *testing.T) { holdDID := "did:web:hold.example.com" digest := testOCIDigest // OCI digest format + // A read presign now resolves the blob's size, so the object has to exist. + seedOCIBlob(t, handler, mockS3Client, digest, 512) + req := makeXRPCGetRequest(atproto.SyncGetBlob, map[string]string{ "did": holdDID, "cid": digest, @@ -2341,19 +2344,24 @@ func TestHandleGetBlob_MockS3_SHA256Digest(t *testing.T) { } // Parse JSON response - var response map[string]string + // The response carries the blob's size alongside the url, so it no longer + // decodes into a map of strings. + var response struct { + URL string `json:"url"` + Size *int64 `json:"size"` + } if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil { t.Fatalf("Failed to parse JSON response: %v", err) } // Verify URL field points to mock S3 - if response["url"] == "" { + if response.URL == "" { t.Error("Expected url field in response") } // URL should be from mock S3 server - if !strings.Contains(response["url"], "mock-s3.example.com") { - t.Errorf("Expected mock S3 URL, got: %s", response["url"]) + if !strings.Contains(response.URL, "mock-s3.example.com") { + t.Errorf("Expected mock S3 URL, got: %s", response.URL) } // Verify GetObjectPresignable was called @@ -2369,6 +2377,9 @@ func TestHandleGetBlob_MockS3_HeadMethod(t *testing.T) { holdDID := "did:web:hold.example.com" digest := testOCIDigest // OCI digest format + // A read presign now resolves the blob's size, so the object has to exist. + seedOCIBlob(t, handler, mockS3Client, digest, 512) + // Use HEAD instead of GET with method query param req := makeXRPCGetRequest(atproto.SyncGetBlob, map[string]string{ "did": holdDID, @@ -2385,19 +2396,24 @@ func TestHandleGetBlob_MockS3_HeadMethod(t *testing.T) { } // Parse JSON response - var response map[string]string + // The response carries the blob's size alongside the url, so it no longer + // decodes into a map of strings. + var response struct { + URL string `json:"url"` + Size *int64 `json:"size"` + } if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil { t.Fatalf("Failed to parse JSON response: %v", err) } // Verify URL field points to mock S3 - if response["url"] == "" { + if response.URL == "" { t.Error("Expected url field in response") } // URL should be from mock S3 server - if !strings.Contains(response["url"], "mock-s3.example.com") { - t.Errorf("Expected mock S3 URL, got: %s", response["url"]) + if !strings.Contains(response.URL, "mock-s3.example.com") { + t.Errorf("Expected mock S3 URL, got: %s", response.URL) } // Verify HeadObjectPresignable was called @@ -2451,11 +2467,14 @@ func TestHandleGetBlob(t *testing.T) { // TestHandleGetBlob_SHA256Digest tests getBlob with OCI sha256 digest format // Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-blob func TestHandleGetBlob_SHA256Digest(t *testing.T) { - handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + handler, mockS3Client, _ := setupTestXRPCHandlerWithBlobs(t) holdDID := "did:web:hold.example.com" digest := testOCIDigest // OCI digest format + // A read presign now resolves the blob's size, so the object has to exist. + seedOCIBlob(t, handler, mockS3Client, digest, 512) + req := makeXRPCGetRequest(atproto.SyncGetBlob, map[string]string{ "did": holdDID, "cid": digest, @@ -2470,13 +2489,18 @@ func TestHandleGetBlob_SHA256Digest(t *testing.T) { } // Parse JSON response - var response map[string]string + // The response carries the blob's size alongside the url, so it no longer + // decodes into a map of strings. + var response struct { + URL string `json:"url"` + Size *int64 `json:"size"` + } if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil { t.Fatalf("Failed to parse JSON response: %v", err) } // Verify URL field exists (will be XRPC proxy URL since we don't have S3 client) - if response["url"] == "" { + if response.URL == "" { t.Error("Expected url field in response") } } diff --git a/pkg/s3/mock.go b/pkg/s3/mock.go index 997945e..235e0dc 100644 --- a/pkg/s3/mock.go +++ b/pkg/s3/mock.go @@ -49,6 +49,12 @@ type MockS3Client struct { HeadObjectCalls []HeadObjectCall PutObjectCalls []PutObjectCall + // RealHeadObjectCalls records HeadObject calls. HeadObjectCalls above + // records PresignHeadObject, which is a different thing: one asks storage + // about an object, the other only mints a URL. Tests that care whether a + // code path actually touched S3 need to count the former. + RealHeadObjectCalls []HeadObjectCall + // Error injection for testing error handling CreateMultipartError error CompleteError error @@ -118,6 +124,7 @@ func NewMockS3Client(testServerURL string) *MockS3Client { GetObjectCalls: []GetObjectCall{}, HeadObjectCalls: []HeadObjectCall{}, PutObjectCalls: []PutObjectCall{}, + RealHeadObjectCalls: []HeadObjectCall{}, } } @@ -209,9 +216,17 @@ func (m *MockS3Client) HeadObject(ctx context.Context, input *awss3.HeadObjectIn } key := aws.ToString(input.Key) + m.RealHeadObjectCalls = append(m.RealHeadObjectCalls, HeadObjectCall{ + Bucket: aws.ToString(input.Bucket), + Key: key, + }) + data, ok := m.Objects[key] if !ok { - return nil, fmt.Errorf("NoSuchKey: object %s not found", key) + // Return the error type the SDK actually produces for a missing object + // on HeadObject, so callers that branch on s3.IsNotFound are exercised + // here the same way they are against real storage. + return nil, &s3types.NotFound{Message: aws.String(fmt.Sprintf("object %s not found", key))} } size := int64(len(data)) diff --git a/pkg/s3/types.go b/pkg/s3/types.go index 7e53219..ab850cf 100644 --- a/pkg/s3/types.go +++ b/pkg/s3/types.go @@ -5,17 +5,21 @@ package s3 import ( "bytes" "context" + "errors" "fmt" "io" "log/slog" + "net/http" "net/url" "strings" "time" "github.com/aws/aws-sdk-go-v2/aws" + awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http" "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/credentials" awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" ) // S3Client defines the S3 operations used by the hold service. @@ -282,6 +286,41 @@ func (s *S3Service) Stat(ctx context.Context, blobPath string) (int64, error) { return 0, nil } +// IsNotFound reports whether err is S3 saying the object does not exist, as +// opposed to any other failure (credentials, network, throttling, a 5xx). +// +// The distinction matters wherever a missing object is a legitimate answer +// rather than an outage: only a genuine "not there" should be turned into a +// 404, and a transient error must never be mistaken for one. +// +// HeadObject has no response body, so the SDK deserializes its errors from the +// HTTP status alone and a 404 arrives as *s3types.NotFound. GetObject and the +// other body-carrying operations report NoSuchKey instead. The raw response +// check is the belt and braces for S3-compatible backends whose error payload +// does not map onto either type. +func IsNotFound(err error) bool { + if err == nil { + return false + } + + var notFound *s3types.NotFound + if errors.As(err, ¬Found) { + return true + } + + var noSuchKey *s3types.NoSuchKey + if errors.As(err, &noSuchKey) { + return true + } + + var respErr *awshttp.ResponseError + if errors.As(err, &respErr) && respErr.HTTPStatusCode() == http.StatusNotFound { + return true + } + + return false +} + // PutBytes uploads data to blobPath with the given content type. func (s *S3Service) PutBytes(ctx context.Context, blobPath string, data []byte, contentType string) error { key := s.s3Key(blobPath)