Stream metadata cache hits for ProxyCached

internal/database/cache_eviction.go

@@ -145,6 +145,32 @@ func (db *DB) EvictMetadataCache(ecosystem, name, storagePath string, lastAccess
 	return true, tx.Commit()
 }
 
+// DiscardMetadataCache deletes a metadata cache row whose stored document
+// failed its integrity check, if it still points at storagePath with digest,
+// and queues the path for deletion in the same transaction. A row a newer
+// fetch has rewritten is left alone.
+func (db *DB) DiscardMetadataCache(ecosystem, name, storagePath, digest string) error {
+	tx, err := db.Beginx()
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback() }()
+	res, err := tx.Exec(db.Rebind(`
+		DELETE FROM metadata_cache
+		WHERE ecosystem = ? AND name = ? AND storage_path = ? AND content_digest = ?`),
+		ecosystem, name, storagePath, digest)
+	if err != nil {
+		return err
+	}
+	if n, err := res.RowsAffected(); err != nil || n == 0 {
+		return err
+	}
+	if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), storagePath, time.Now().UTC()); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
 // RecordMetadataHit marks a cached metadata document as served now. Like
 // RecordArtifactHit it is buffered when BatchHits is on.
 func (db *DB) RecordMetadataHit(ecosystem, name string) error {

internal/handler/handler.go

@@ -1861,6 +1861,11 @@ func (p *Proxy) lookupCachedMeta(ecosystem, cacheKey string) cachedMeta {
 	if err != nil || entry == nil {
 		return cachedMeta{}
 	}
+	return p.cachedMetaFor(entry)
+}
+
+// cachedMetaFor returns the cache validators and freshness state of entry.
+func (p *Proxy) cachedMetaFor(entry *database.MetadataCacheEntry) cachedMeta {
 	var cm cachedMeta
 	if entry.ETag.Valid {
 		cm.etag = entry.ETag.String
@@ -1896,6 +1901,10 @@ func (p *Proxy) proxyCachedWithEncoding(w http.ResponseWriter, r *http.Request,
 		return
 	}
 
+	if p.serveCachedMetadata(w, r, ecosystem, cacheKey) {
+		return
+	}
+
 	body, contentType, contentEncoding, err := p.fetchOrCacheMetadata(r.Context(), ecosystem, cacheKey, upstreamURL, acceptEncoding, nil, acceptHeaders...)
 	if err != nil {
 		p.writeMetadataFetchError(w, err)
@@ -1931,7 +1940,25 @@ func (p *Proxy) writeMetadataCachedResponse(w http.ResponseWriter, r *http.Reque
 // mislabel the bytes.
 func (p *Proxy) writeMetadataCachedResponseWithEncoding(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string, body []byte, contentType, contentEncoding string) {
 	cm := p.lookupCachedMeta(ecosystem, cacheKey)
+	if writeMetadataValidators(w, r, cm) {
+		return
+	}
 
+	w.Header().Set(headerContentType, contentType)
+	w.Header().Set(headerContentLength, strconv.Itoa(len(body)))
+	if contentEncoding != "" {
+		w.Header().Set(headerContentEncoding, contentEncoding)
+	}
+	markStale(w, cm.stale)
+	w.WriteHeader(http.StatusOK)
+	if r.Method != http.MethodHead {
+		_, _ = w.Write(body)
+	}
+}
+
+// writeMetadataValidators sets cm's ETag and Last-Modified on w and answers
+// a conditional request they satisfy with 304, reporting whether it did.
+func writeMetadataValidators(w http.ResponseWriter, r *http.Request, cm cachedMeta) bool {
 	if cm.etag != "" {
 		w.Header().Set(headerETag, cm.etag)
 	}
@@ -1940,27 +1967,17 @@ func (p *Proxy) writeMetadataCachedResponseWithEncoding(w http.ResponseWriter, r
 	}
 	if ifNoneMatchHits(r.Header.Get("If-None-Match"), cm.etag) {
 		w.WriteHeader(http.StatusNotModified)
-		return
+		return true
 	}
 	if !cm.lastModified.IsZero() {
 		if ims := r.Header.Get("If-Modified-Since"); ims != "" {
 			if t, err := http.ParseTime(ims); err == nil && !cm.lastModified.After(t) {
 				w.WriteHeader(http.StatusNotModified)
-				return
+				return true
 			}
 		}
 	}
-
-	w.Header().Set(headerContentType, contentType)
-	w.Header().Set(headerContentLength, strconv.Itoa(len(body)))
-	if contentEncoding != "" {
-		w.Header().Set(headerContentEncoding, contentEncoding)
-	}
-	markStale(w, cm.stale)
-	w.WriteHeader(http.StatusOK)
-	if r.Method != http.MethodHead {
-		_, _ = w.Write(body)
-	}
+	return false
 }
 
 // proxyMetadataStream forwards an upstream metadata response by streaming it to the client

internal/handler/metadata_stream.go

@@ -0,0 +1,85 @@
+package handler
+
+import (
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/git-pkgs/proxy/internal/metrics"
+)
+
+// serveCachedMetadata answers a ProxyCached request from a fresh metadata
+// cache row by streaming the stored document, rather than reading it into
+// memory first, so concurrent hits on a large document don't each hold a copy
+// of it. The bytes are checked against the row's digest as they are sent, and
+// a mismatch aborts the response and drops the row, the way a corrupt cached
+// artifact is handled.
+//
+// It reports false, having written nothing, when the request should take the
+// buffered path instead: no fresh row, a row without a digest or size, a
+// document over the metadata size limit, or one that can't be opened.
+func (p *Proxy) serveCachedMetadata(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string) bool {
+	if p.DB == nil || p.Storage == nil || p.MetadataTTL <= 0 || containsPathTraversal(cacheKey) {
+		return false
+	}
+	entry, err := p.DB.GetMetadataCache(ecosystem, cacheKey)
+	if err != nil || entry == nil || !entry.FetchedAt.Valid || time.Since(entry.FetchedAt.Time) >= p.MetadataTTL {
+		return false
+	}
+	limit := p.MetadataMaxSize
+	if limit <= 0 {
+		limit = defaultMetadataMaxSize
+	}
+	digest, ok := strings.CutPrefix(entry.ContentDigest.String, "sha256:")
+	if !entry.ContentDigest.Valid || !ok || !entry.Size.Valid || entry.Size.Int64 > limit {
+		return false
+	}
+	checks, err := newIntegrityChecks(digest, "")
+	if err != nil {
+		return false
+	}
+	stored, err := p.Storage.Open(r.Context(), entry.StoragePath)
+	if err != nil {
+		return false
+	}
+	body, err := checks.wrapFailOnMismatch(stored, func(reason string) {
+		p.Logger.Error("cached metadata failed integrity check",
+			"ecosystem", ecosystem, "key", cacheKey, "path", entry.StoragePath, "reason", reason)
+		metrics.RecordIntegrityFailure(ecosystem, metrics.FailureDetail{
+			Path: entry.StoragePath, Err: errors.New(reason)})
+		// Drop the row so the next request fetches the document again.
+		if err := p.DB.DiscardMetadataCache(ecosystem, cacheKey, entry.StoragePath, entry.ContentDigest.String); err != nil {
+			p.Logger.Warn("failed to clear corrupt metadata from cache", "error", err)
+		}
+	})
+	if err != nil {
+		_ = stored.Close()
+		return false
+	}
+	defer func() { _ = body.Close() }()
+
+	p.recordMetadataRead(entry)
+	metrics.RecordMetadataCacheHit(ecosystem, int(entry.Size.Int64))
+
+	cm := p.cachedMetaFor(entry)
+	if writeMetadataValidators(w, r, cm) {
+		return true
+	}
+	contentType := contentTypeJSON
+	if entry.ContentType.Valid {
+		contentType = entry.ContentType.String
+	}
+	w.Header().Set(headerContentType, contentType)
+	w.Header().Set(headerContentLength, strconv.FormatInt(entry.Size.Int64, 10))
+	if entry.ContentEncoding.String != "" {
+		w.Header().Set(headerContentEncoding, entry.ContentEncoding.String)
+	}
+	markStale(w, cm.stale)
+	w.WriteHeader(http.StatusOK)
+	if r.Method != http.MethodHead {
+		copyArtifactBody(w, body, entry.Size.Int64, false)
+	}
+	return true
+}