From f240d3023fccf8057ff3d80b4e67ce388a280155 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Sat, 24 Oct 2026 13:13:12 +0200 Subject: [PATCH 1/2] Stream metadata cache hits for ProxyCached With cache_metadata on, every metadata cache hit read the whole stored document into its own buffer before any coalescing, since coalesceMetadata only covers upstream misses. metadata_max_size bounds one request, not the sum: 30 concurrent hits on a 20 MB document held about 820 MB. The handlers that go through ProxyCached serve the stored document unchanged, so a hit there no longer needs the bytes in memory. serveCachedMetadata streams the blob from storage to the response instead, with the row's Content-Type, Content-Encoding, size, ETag and Last-Modified, and answers If-None-Match and If-Modified-Since with 304 as before. The bytes are checked against the row's content digest as they are sent, and a mismatch aborts the response like a corrupt cached artifact. A row without a digest or size, a document over the size limit, or a blob that can't be opened takes the buffered path as before. Handlers that rewrite or filter metadata keep reading it into memory, and the rewrite cache stays in front of them. writeMetadataValidators is split out of writeMetadataCachedResponseWithEncoding so both paths share the conditional request handling. --- internal/handler/handler.go | 43 +++-- internal/handler/metadata_stream.go | 77 +++++++++ internal/handler/metadata_stream_test.go | 202 +++++++++++++++++++++++ 3 files changed, 309 insertions(+), 13 deletions(-) create mode 100644 internal/handler/metadata_stream.go create mode 100644 internal/handler/metadata_stream_test.go diff --git a/internal/handler/handler.go b/internal/handler/handler.go index c06ec2d4..164ad265 100644 --- a/internal/handler/handler.go +++ b/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 diff --git a/internal/handler/metadata_stream.go b/internal/handler/metadata_stream.go new file mode 100644 index 00000000..2fb994ad --- /dev/null +++ b/internal/handler/metadata_stream.go @@ -0,0 +1,77 @@ +package handler + +import ( + "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 the way a corrupt cached artifact does. +// +// 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) + }) + 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 +} diff --git a/internal/handler/metadata_stream_test.go b/internal/handler/metadata_stream_test.go new file mode 100644 index 00000000..2874d9f9 --- /dev/null +++ b/internal/handler/metadata_stream_test.go @@ -0,0 +1,202 @@ +package handler + +import ( + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "slices" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/storage" +) + +// stallingStorage hands out cached metadata readers that return the first +// head bytes and then block until release closes, so a test can see whether a +// response starts before the whole document has been read. +type stallingStorage struct { + storage.Storage + head int + release chan struct{} +} + +func (s *stallingStorage) Open(ctx context.Context, path string) (io.ReadCloser, error) { + rc, err := s.Storage.Open(ctx, path) + if err != nil || !strings.HasPrefix(path, "_metadata/") { + return rc, err + } + return &stallingReader{ReadCloser: rc, left: s.head, release: s.release}, nil +} + +type stallingReader struct { + io.ReadCloser + left int + release chan struct{} +} + +func (r *stallingReader) Read(p []byte) (int, error) { + if r.left == 0 { + <-r.release + return r.ReadCloser.Read(p) + } + if len(p) > r.left { + p = p[:r.left] + } + n, err := r.ReadCloser.Read(p) + r.left -= n + return n, err +} + +// cachedMetadataServer serves ProxyCached for one document, cached on the +// first request. +func cachedMetadataServer(t *testing.T, doc []byte) (*Proxy, *mockStorage, *httptest.Server) { + t.Helper() + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set(headerContentType, "application/x-test") + w.Header().Set(headerETag, `"v1"`) + _, _ = w.Write(doc) + })) + t.Cleanup(upstream.Close) + + proxy, _, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.HTTPClient = upstream.Client() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + proxy.ProxyCached(w, r, upstream.URL+"/doc", "test", "doc", "*/*") + })) + t.Cleanup(srv.Close) + + resp, err := http.Get(srv.URL) + if err != nil { + t.Fatalf("populating cache: %v", err) + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("populating cache: status %d", resp.StatusCode) + } + return proxy, store, srv +} + +func TestConcurrentMetadataCacheHitsStream(t *testing.T) { + const callers = 10 + doc := bytes.Repeat([]byte("0123456789abcdef"), 1<<18) // 4 MB + proxy, store, srv := cachedMetadataServer(t, doc) + + gate := &stallingStorage{Storage: store, head: 64 << 10, release: make(chan struct{})} + proxy.Storage = gate + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(gate.release) }) } + t.Cleanup(release) + + // Every hit has to start its response while its reader is stalled after + // the first 64 KB. A hit that reads the document into memory first can't + // send anything until release. + type result struct { + resp *http.Response + first []byte + err error + } + started := make(chan result, callers) + for range callers { + go func() { + resp, err := http.Get(srv.URL) + if err != nil { + started <- result{err: err} + return + } + first := make([]byte, 1024) + _, err = io.ReadFull(resp.Body, first) + started <- result{resp: resp, first: first, err: err} + }() + } + results := make([]result, 0, callers) + timeout := time.After(5 * time.Second) + streamed := 0 + for len(results) < callers { + select { + case res := <-started: + results = append(results, res) + case <-timeout: + // Let the stalled hits finish, so the server can shut down. + release() + timeout = nil + streamed = len(results) + } + } + release() + t.Cleanup(func() { + for _, res := range results { + if res.resp != nil { + _ = res.resp.Body.Close() + } + } + }) + if timeout == nil { + t.Fatalf("%d of %d concurrent cache hits started their response before the document was read to its end", streamed, callers) + } + + for _, res := range results { + if res.err != nil { + t.Fatalf("cache hit: %v", res.err) + } + rest, err := io.ReadAll(res.resp.Body) + _ = res.resp.Body.Close() + if err != nil { + t.Fatalf("reading cache hit: %v", err) + } + if body := slices.Concat(res.first, rest); !bytes.Equal(body, doc) { + t.Fatalf("cache hit returned %d bytes, want the %d-byte document", len(body), len(doc)) + } + if got := res.resp.Header.Get(headerContentType); got != "application/x-test" { + t.Errorf("Content-Type = %q, want application/x-test", got) + } + if got := res.resp.Header.Get(headerContentLength); got != strconv.Itoa(len(doc)) { + t.Errorf("Content-Length = %q, want %d", got, len(doc)) + } + if got := res.resp.Header.Get(headerETag); got != `"v1"` { + t.Errorf("ETag = %q, want %q", got, `"v1"`) + } + } + + req, _ := http.NewRequest(http.MethodGet, srv.URL, nil) + req.Header.Set("If-None-Match", `"v1"`) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("conditional cache hit: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusNotModified { + t.Errorf("conditional cache hit: status %d, want 304", resp.StatusCode) + } +} + +func TestCachedMetadataHitAbortsOnDigestMismatch(t *testing.T) { + doc := bytes.Repeat([]byte("metadata "), 1<<14) + _, store, srv := cachedMetadataServer(t, doc) + + corrupt := bytes.Clone(doc) + corrupt[len(corrupt)-1] ^= 0xff + store.mu.Lock() + store.files[metadataStoragePath("test", "doc")] = corrupt + store.mu.Unlock() + + resp, err := http.Get(srv.URL) + if err != nil { + t.Fatalf("cache hit: %v", err) + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err == nil { + t.Fatalf("cache hit completed with %d bytes of a document that doesn't match its digest, want the response aborted", len(body)) + } + if len(body) >= len(doc) { + t.Errorf("aborted response still delivered %d of %d bytes", len(body), len(doc)) + } +} From 6c9505f3220c39d97fdd02acce000ad0ea788efc Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Sun, 25 Oct 2026 12:32:20 +0100 Subject: [PATCH 2/2] Drop cached metadata that fails its integrity check A streamed cache hit whose bytes don't match the row's digest now records an integrity failure and deletes the row, queueing its blob for deletion, as a corrupt cached artifact is handled. The next request fetches the document from upstream instead of aborting until the TTL runs out. --- internal/database/cache_eviction.go | 26 +++++++++++++++++ internal/handler/metadata_stream.go | 10 ++++++- internal/handler/metadata_stream_test.go | 37 +++++++++++++++++++----- 3 files changed, 64 insertions(+), 9 deletions(-) diff --git a/internal/database/cache_eviction.go b/internal/database/cache_eviction.go index e14df8d3..33bcdae7 100644 --- a/internal/database/cache_eviction.go +++ b/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 { diff --git a/internal/handler/metadata_stream.go b/internal/handler/metadata_stream.go index 2fb994ad..8bb319d9 100644 --- a/internal/handler/metadata_stream.go +++ b/internal/handler/metadata_stream.go @@ -1,6 +1,7 @@ package handler import ( + "errors" "net/http" "strconv" "strings" @@ -13,7 +14,8 @@ import ( // 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 the way a corrupt cached artifact does. +// 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 @@ -45,6 +47,12 @@ func (p *Proxy) serveCachedMetadata(w http.ResponseWriter, r *http.Request, ecos 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() diff --git a/internal/handler/metadata_stream_test.go b/internal/handler/metadata_stream_test.go index 2874d9f9..702eea86 100644 --- a/internal/handler/metadata_stream_test.go +++ b/internal/handler/metadata_stream_test.go @@ -10,6 +10,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "testing" "time" @@ -52,22 +53,24 @@ func (r *stallingReader) Read(p []byte) (int, error) { return n, err } -// cachedMetadataServer serves ProxyCached for one document, cached on the -// first request. -func cachedMetadataServer(t *testing.T, doc []byte) (*Proxy, *mockStorage, *httptest.Server) { +// cachedMetadataServer serves doc through ProxyCached and caches it with one +// request. upstreamHits counts the requests that reached the upstream. +func cachedMetadataServer(t *testing.T, doc []byte) (proxy *Proxy, store *mockStorage, srv *httptest.Server, upstreamHits *atomic.Int32) { t.Helper() + upstreamHits = &atomic.Int32{} upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstreamHits.Add(1) w.Header().Set(headerContentType, "application/x-test") w.Header().Set(headerETag, `"v1"`) _, _ = w.Write(doc) })) t.Cleanup(upstream.Close) - proxy, _, store, _ := setupTestProxy(t) + proxy, _, store, _ = setupTestProxy(t) proxy.CacheMetadata = true proxy.MetadataTTL = time.Hour proxy.HTTPClient = upstream.Client() - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { proxy.ProxyCached(w, r, upstream.URL+"/doc", "test", "doc", "*/*") })) t.Cleanup(srv.Close) @@ -81,13 +84,13 @@ func cachedMetadataServer(t *testing.T, doc []byte) (*Proxy, *mockStorage, *http if resp.StatusCode != http.StatusOK { t.Fatalf("populating cache: status %d", resp.StatusCode) } - return proxy, store, srv + return proxy, store, srv, upstreamHits } func TestConcurrentMetadataCacheHitsStream(t *testing.T) { const callers = 10 doc := bytes.Repeat([]byte("0123456789abcdef"), 1<<18) // 4 MB - proxy, store, srv := cachedMetadataServer(t, doc) + proxy, store, srv, _ := cachedMetadataServer(t, doc) gate := &stallingStorage{Storage: store, head: 64 << 10, release: make(chan struct{})} proxy.Storage = gate @@ -179,7 +182,7 @@ func TestConcurrentMetadataCacheHitsStream(t *testing.T) { func TestCachedMetadataHitAbortsOnDigestMismatch(t *testing.T) { doc := bytes.Repeat([]byte("metadata "), 1<<14) - _, store, srv := cachedMetadataServer(t, doc) + _, store, srv, upstreamHits := cachedMetadataServer(t, doc) corrupt := bytes.Clone(doc) corrupt[len(corrupt)-1] ^= 0xff @@ -199,4 +202,22 @@ func TestCachedMetadataHitAbortsOnDigestMismatch(t *testing.T) { if len(body) >= len(doc) { t.Errorf("aborted response still delivered %d of %d bytes", len(body), len(doc)) } + + // The corrupt row is dropped, so the next request fetches the document + // from upstream again instead of aborting until the TTL runs out. + resp, err = http.Get(srv.URL) + if err != nil { + t.Fatalf("request after abort: %v", err) + } + defer func() { _ = resp.Body.Close() }() + body, err = io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("request after abort: %v", err) + } + if !bytes.Equal(body, doc) { + t.Errorf("request after abort got %d bytes, want the upstream document (%d bytes)", len(body), len(doc)) + } + if got := upstreamHits.Load(); got != 2 { + t.Errorf("upstream requests = %d, want 2 (populate, then refetch after the corrupt hit)", got) + } }