diff --git a/internal/handler/handler.go b/internal/handler/handler.go index b04f941..e9abd5a 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -198,6 +198,11 @@ type Proxy struct { fetchMu sync.Mutex inFlight map[string]*inflightFetch + // inFlightMeta does the same for metadata misses. Keyed by + // metadataCoalesceKey. + metaMu sync.Mutex + inFlightMeta map[string]*inflightMetadata + // rewrites caches metadata documents after their handler rewrites them. // Nil leaves every request to rewrite its own copy. rewrites *rewriteCache @@ -909,34 +914,20 @@ func (p *Proxy) FetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, u // replayed as sent. The ProxyCached path uses "identity" for signed indexes and // "gzip" where both hops should stay compressed. // validate, when supplied, runs before caching or serving a document. Validation -// failures follow the same stale-cache fallback path as upstream failures. +// failures follow the same stale-cache fallback path as upstream failures. It +// runs for every caller, including one that shares another caller's fetch, so +// it can also decode the document into request-local state; a caller that +// joined a fetch gets its own validation error directly. It must not modify the +// body, which joined callers share. func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, acceptEncoding string, validate func([]byte) error, acceptHeaders ...string) ([]byte, string, string, error) { if containsPathTraversal(cacheKey) { return nil, "", "", fmt.Errorf("invalid cache key: %q", cacheKey) } - storagePath := metadataStoragePath(ecosystem, cacheKey) - - // Check for existing cache entry (for ETag revalidation and TTL) - var entry *database.MetadataCacheEntry - if p.CacheMetadata && p.DB != nil { - entry, _ = p.DB.GetMetadataCache(ecosystem, cacheKey) - } - // Serve from cache if within TTL (skip upstream entirely) - if entry != nil && p.MetadataTTL > 0 && entry.FetchedAt.Valid { - if time.Since(entry.FetchedAt.Time) < p.MetadataTTL { - data, ct, readErr := p.readCachedMetadata(ctx, entry, validate) - if readErr == nil { - metrics.RecordCacheHit(ecosystem) - return data, ct, entry.ContentEncoding.String, nil - } - if validate != nil { - // Do not revalidate an unusable cached body with its ETag. - entry = nil - } - // Cache file missing/unreadable, fall through to upstream - } + if _, hit := p.cachedMetadataState(ctx, ecosystem, cacheKey, validate); hit != nil { + metrics.RecordCacheHit(ecosystem) + return hit.body, hit.contentType, hit.contentEncoding, nil } p.recordMetadataCacheMiss(ecosystem) @@ -945,7 +936,62 @@ func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, u accept = acceptHeaders[0] } - // Try upstream + res := p.coalescedMetadataMiss(ctx, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate) + return res.body, res.contentType, res.contentEncoding, res.err +} + +// coalescedMetadataMiss handles a metadata cache miss, sharing one upstream +// fetch among concurrent callers with the same key. +func (p *Proxy) coalescedMetadataMiss(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validate func([]byte) error) metadataResult { + key := metadataCoalesceKey(ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate != nil) + res, shared := p.coalesceMetadata(ctx, key, func(fetchCtx context.Context) metadataResult { + // The caller's lookup ran before it took the key, so a fetch that + // finished in between has already refreshed the row. Recheck it + // rather than fetching again, and revalidate against the row as it + // is now. + entry, hit := p.cachedMetadataState(fetchCtx, ecosystem, cacheKey, validate) + if hit != nil { + return *hit + } + return p.fetchMetadataFromUpstream(fetchCtx, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate, entry) + }) + // The fetch ran the first caller's validate. A caller that joined it runs + // its own on the shared bytes, since validate may also decode them for it. + if shared && res.err == nil && validate != nil { + if err := validate(res.body); err != nil { + return metadataResult{err: err} + } + } + return res +} + +// cachedMetadataState reads the cache row for a metadata lookup. hit is set +// when the row is within TTL and its bytes are usable. Otherwise entry is the +// row to revalidate against with its ETag: nil when there is none, or when +// validate rejected the cached body, since revalidating an unusable body would +// keep it. +func (p *Proxy) cachedMetadataState(ctx context.Context, ecosystem, cacheKey string, validate func([]byte) error) (*database.MetadataCacheEntry, *metadataResult) { + if !p.CacheMetadata || p.DB == nil { + return nil, nil + } + entry, _ := p.DB.GetMetadataCache(ecosystem, cacheKey) + if entry != nil && p.MetadataTTL > 0 && entry.FetchedAt.Valid && time.Since(entry.FetchedAt.Time) < p.MetadataTTL { + data, ct, err := p.readCachedMetadata(ctx, entry, validate) + if err == nil { + return entry, &metadataResult{body: data, contentType: ct, contentEncoding: entry.ContentEncoding.String} + } + if validate != nil { + return nil, nil + } + // Cache file missing/unreadable, fall through to upstream + } + return entry, nil +} + +// fetchMetadataFromUpstream fetches metadata after a cache miss, caches it, +// and falls back to the cached copy if upstream fails. entry is the row to +// revalidate against, or nil. +func (p *Proxy) fetchMetadataFromUpstream(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validate func([]byte) error, entry *database.MetadataCacheEntry) metadataResult { meta, err := p.fetchUpstreamMetadata(ctx, upstreamURL, entry, accept, acceptEncoding) if errors.Is(err, errStale304) { // 304 but cached file is gone; retry without ETag @@ -956,14 +1002,14 @@ func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, u } if err == nil { if p.CacheMetadata { - p.cacheMetadataBlob(ctx, ecosystem, cacheKey, storagePath, meta) + p.cacheMetadataBlob(ctx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), meta) } - return meta.body, meta.contentType, meta.contentEncoding, nil + return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding} } // Upstream failed -- fall back to cache if available if !p.CacheMetadata || entry == nil { - return nil, "", "", fmt.Errorf("upstream failed and no cached metadata: %w", err) + return metadataResult{err: fmt.Errorf("upstream failed and no cached metadata: %w", err)} } p.Logger.Warn("upstream metadata fetch failed, checking cache", @@ -976,12 +1022,86 @@ func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, u data, ct, readErr := p.readCachedMetadata(ctx, entry, validate) if readErr != nil { - return nil, "", "", fmt.Errorf("upstream failed and cached metadata unusable (%v): %w", readErr, err) + return metadataResult{err: fmt.Errorf("upstream failed and cached metadata unusable (%v): %w", readErr, err)} } p.Logger.Info("serving metadata from cache", "ecosystem", ecosystem, "key", cacheKey) - return data, ct, entry.ContentEncoding.String, nil + return metadataResult{body: data, contentType: ct, contentEncoding: entry.ContentEncoding.String} +} + +// metadataResult is what a metadata lookup hands back. A shared fetch gives +// every waiter the same body, so callers must treat it as read-only. +type metadataResult struct { + body []byte + contentType string + contentEncoding string + err error +} + +// inflightMetadata is one metadata fetch that concurrent callers share. res is +// written before done closes and read only after, so the close is the handoff. +type inflightMetadata struct { + done chan struct{} + res metadataResult + + // waiters counts callers that joined the fetch, guarded by metaMu. Tests + // read it to know every caller has joined before the fetch finishes. + waiters int +} + +// metadataCoalesceKey identifies metadata requests that can share one fetch. +// Accept and Accept-Encoding are part of it because they change the bytes +// upstream returns: npm serves an abbreviated or a full document for the same +// package, and the cached-proxy paths ask for identity or gzip. Whether the +// caller validates is part of it so an unvalidated caller is never handed a +// result that skipped validation, or the reverse. +func metadataCoalesceKey(ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validated bool) string { + return strings.Join([]string{ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, strconv.FormatBool(validated)}, "\x00") +} + +// coalesceMetadata runs fetch at most once for concurrent callers sharing key +// and gives each the result, reporting whether this caller joined another's +// fetch rather than running it. It follows coalesceFetch with one difference: +// fetch runs on a context detached from the first caller's cancellation. A +// metadata fetch has no scan or mirror that depends on the caller aborting it, +// and CI jobs asking for the same Composer or npm metadata would otherwise all +// fail when the first of them disconnects. It stays bounded by the HTTP +// client's timeout. Waiters still leave when their own context ends. +func (p *Proxy) coalesceMetadata(ctx context.Context, key string, fetch func(context.Context) metadataResult) (metadataResult, bool) { + p.metaMu.Lock() + if p.inFlightMeta == nil { + p.inFlightMeta = make(map[string]*inflightMetadata) + } + f, joined := p.inFlightMeta[key] + if joined { + f.waiters++ + } else { + f = &inflightMetadata{done: make(chan struct{})} + p.inFlightMeta[key] = f + } + p.metaMu.Unlock() + + if joined { + select { + case <-ctx.Done(): + return metadataResult{err: ctx.Err()}, true + case <-f.done: + return f.res, true + } + } + + // Set before running so a panicking fetch leaves waiters with an error + // rather than an empty body. + f.res = metadataResult{err: errSharedFetchAbandoned} + defer func() { + p.metaMu.Lock() + delete(p.inFlightMeta, key) + p.metaMu.Unlock() + close(f.done) + }() + f.res = fetch(context.WithoutCancel(ctx)) + return f.res, false } func (p *Proxy) readCachedMetadata(ctx context.Context, entry *database.MetadataCacheEntry, validate func([]byte) error) ([]byte, string, error) { diff --git a/internal/handler/metadata_coalesce_test.go b/internal/handler/metadata_coalesce_test.go new file mode 100644 index 0000000..af8e2eb --- /dev/null +++ b/internal/handler/metadata_coalesce_test.go @@ -0,0 +1,249 @@ +package handler + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" +) + +// metadataUpstream is a stub registry that counts metadata requests. It +// signals entered when the first request arrives and holds every request until +// release is closed, so a test can keep the first fetch in flight while others +// arrive. +type metadataUpstream struct { + *httptest.Server + calls atomic.Int64 + entered chan struct{} + release chan struct{} + once sync.Once +} + +func newMetadataUpstream(t *testing.T, status int) *metadataUpstream { + t.Helper() + u := &metadataUpstream{entered: make(chan struct{}), release: make(chan struct{})} + u.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + u.calls.Add(1) + u.once.Do(func() { close(u.entered) }) + <-u.release + if status != http.StatusOK { + w.WriteHeader(status) + return + } + w.Header().Set("Content-Type", "application/json") + // Echo the Accept header so a test can tell which variant it got. + _, _ = w.Write([]byte(`{"accept":"` + r.Header.Get("Accept") + `"}`)) + })) + t.Cleanup(u.Close) + return u +} + +func metadataTestProxy(t *testing.T, u *metadataUpstream) *Proxy { + t.Helper() + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = u.Client() + return proxy +} + +// waitForMetadataWaiters returns once n callers are waiting on the fetch for +// key, so a test can release upstream knowing every caller joined it. +func waitForMetadataWaiters(t *testing.T, p *Proxy, key string, n int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + p.metaMu.Lock() + f := p.inFlightMeta[key] + joined := 0 + if f != nil { + joined = f.waiters + } + p.metaMu.Unlock() + if joined >= n { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("timed out waiting for %d callers to join the metadata fetch", n) +} + +type metadataCall struct { + body []byte + err error +} + +// startMetadataCalls starts n FetchOrCacheMetadata calls for one package. +func startMetadataCalls(p *Proxy, ctx context.Context, n int, url, accept string) []chan metadataCall { + out := make([]chan metadataCall, n) + for i := range out { + out[i] = make(chan metadataCall, 1) + go func(ch chan metadataCall) { + body, _, err := p.FetchOrCacheMetadata(ctx, "npm", "left-pad", url, accept) + ch <- metadataCall{body, err} + }(out[i]) + } + return out +} + +// TestFetchOrCacheMetadata_ConcurrentMissesCoalesce asserts that concurrent +// misses for one package make a single upstream request: the CI shape, where +// parallel jobs resolve the same Composer or npm dependencies at once. +func TestFetchOrCacheMetadata_ConcurrentMissesCoalesce(t *testing.T) { + u := newMetadataUpstream(t, http.StatusOK) + p := metadataTestProxy(t, u) + const n = 10 + + calls := startMetadataCalls(p, context.Background(), 1, u.URL, contentTypeJSON) + <-u.entered + calls = append(calls, startMetadataCalls(p, context.Background(), n-1, u.URL, contentTypeJSON)...) + key := metadataCoalesceKey("npm", "left-pad", u.URL, contentTypeJSON, "", false) + waitForMetadataWaiters(t, p, key, n-1) + close(u.release) + + for _, ch := range calls { + c := <-ch + if c.err != nil { + t.Fatalf("FetchOrCacheMetadata: %v", c.err) + } + if string(c.body) != `{"accept":"application/json"}` { + t.Errorf("body = %s", c.body) + } + } + if got := u.calls.Load(); got != 1 { + t.Errorf("upstream requests = %d, want 1", got) + } +} + +// TestFetchOrCacheMetadata_DifferentAcceptDoNotShare asserts that requests +// for different variants of one package, such as npm's abbreviated and full +// documents, each get their own fetch and their own bytes. +func TestFetchOrCacheMetadata_DifferentAcceptDoNotShare(t *testing.T) { + u := newMetadataUpstream(t, http.StatusOK) + p := metadataTestProxy(t, u) + const abbreviated = "application/vnd.npm.install-v1+json" + + full := startMetadataCalls(p, context.Background(), 1, u.URL, contentTypeJSON) + <-u.entered + abbr := startMetadataCalls(p, context.Background(), 1, u.URL, abbreviated) + deadline := time.Now().Add(5 * time.Second) + for u.calls.Load() < 2 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + close(u.release) + + if c := <-full[0]; c.err != nil || string(c.body) != `{"accept":"application/json"}` { + t.Errorf("full document: body %s, err %v", c.body, c.err) + } + if c := <-abbr[0]; c.err != nil || string(c.body) != `{"accept":"`+abbreviated+`"}` { + t.Errorf("abbreviated document: body %s, err %v", c.body, c.err) + } + if got := u.calls.Load(); got != 2 { + t.Errorf("upstream requests = %d, want 2", got) + } +} + +// TestFetchOrCacheMetadata_FirstCallerLeavingDoesNotFailOthers asserts that +// when the caller running the shared fetch disconnects, the fetch continues +// and everyone waiting on it still gets the metadata. +func TestFetchOrCacheMetadata_FirstCallerLeavingDoesNotFailOthers(t *testing.T) { + u := newMetadataUpstream(t, http.StatusOK) + p := metadataTestProxy(t, u) + const n = 5 + + leaderCtx, cancelLeader := context.WithCancel(context.Background()) + leader := startMetadataCalls(p, leaderCtx, 1, u.URL, contentTypeJSON) + <-u.entered + waiters := startMetadataCalls(p, context.Background(), n, u.URL, contentTypeJSON) + key := metadataCoalesceKey("npm", "left-pad", u.URL, contentTypeJSON, "", false) + waitForMetadataWaiters(t, p, key, n) + cancelLeader() + close(u.release) + + <-leader[0] + for _, ch := range waiters { + if c := <-ch; c.err != nil { + t.Errorf("waiter failed after the first caller left: %v", c.err) + } + } + if got := u.calls.Load(); got != 1 { + t.Errorf("upstream requests = %d, want 1", got) + } +} + +// TestFetchOrCacheMetadata_WaiterLeavingKeepsFetch asserts that a waiter whose +// client disconnects returns its own context error without disturbing the +// fetch the others are waiting on. +func TestFetchOrCacheMetadata_WaiterLeavingKeepsFetch(t *testing.T) { + u := newMetadataUpstream(t, http.StatusOK) + p := metadataTestProxy(t, u) + + leader := startMetadataCalls(p, context.Background(), 1, u.URL, contentTypeJSON) + <-u.entered + waiterCtx, cancelWaiter := context.WithCancel(context.Background()) + leaving := startMetadataCalls(p, waiterCtx, 1, u.URL, contentTypeJSON) + key := metadataCoalesceKey("npm", "left-pad", u.URL, contentTypeJSON, "", false) + waitForMetadataWaiters(t, p, key, 1) + cancelWaiter() + + if c := <-leaving[0]; !errors.Is(c.err, context.Canceled) { + t.Errorf("leaving waiter err = %v, want context.Canceled", c.err) + } + close(u.release) + if c := <-leader[0]; c.err != nil { + t.Errorf("first caller failed: %v", c.err) + } +} + +// TestFetchOrCacheMetadata_SharedNotFound asserts that an upstream 404 reaches +// every caller sharing the fetch as ErrUpstreamNotFound, from one request. +func TestFetchOrCacheMetadata_SharedNotFound(t *testing.T) { + u := newMetadataUpstream(t, http.StatusNotFound) + p := metadataTestProxy(t, u) + const n = 5 + + calls := startMetadataCalls(p, context.Background(), 1, u.URL, contentTypeJSON) + <-u.entered + calls = append(calls, startMetadataCalls(p, context.Background(), n-1, u.URL, contentTypeJSON)...) + key := metadataCoalesceKey("npm", "left-pad", u.URL, contentTypeJSON, "", false) + waitForMetadataWaiters(t, p, key, n-1) + close(u.release) + + for _, ch := range calls { + if c := <-ch; !errors.Is(c.err, ErrUpstreamNotFound) { + t.Errorf("err = %v, want ErrUpstreamNotFound", c.err) + } + } + if got := u.calls.Load(); got != 1 { + t.Errorf("upstream requests = %d, want 1", got) + } +} + +// TestCoalescedMetadataMiss_ServesRowCommittedSinceLookup asserts that a +// caller which missed the cache, but finds the row fresh once it takes the +// key, serves that row instead of fetching again. That is a fetch finishing +// between a caller's lookup and its turn at the key. +func TestCoalescedMetadataMiss_ServesRowCommittedSinceLookup(t *testing.T) { + u := newMetadataUpstream(t, http.StatusOK) + close(u.release) + p := metadataTestProxy(t, u) + p.CacheMetadata = true + p.MetadataTTL = time.Hour + + if _, _, err := p.FetchOrCacheMetadata(context.Background(), "npm", "left-pad", u.URL); err != nil { + t.Fatalf("priming fetch: %v", err) + } + + res := p.coalescedMetadataMiss(context.Background(), "npm", "left-pad", u.URL, contentTypeJSON, "", nil) + if res.err != nil { + t.Fatalf("coalescedMetadataMiss: %v", res.err) + } + if string(res.body) != `{"accept":"application/json"}` { + t.Errorf("body = %s", res.body) + } + if got := u.calls.Load(); got != 1 { + t.Errorf("upstream requests = %d, want 1 (the priming fetch only)", got) + } +} diff --git a/internal/handler/nuget_cooldown_test.go b/internal/handler/nuget_cooldown_test.go index 74258da..3160bd5 100644 --- a/internal/handler/nuget_cooldown_test.go +++ b/internal/handler/nuget_cooldown_test.go @@ -3,12 +3,15 @@ package handler import ( "bytes" "compress/gzip" + "crypto/sha256" "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" "strings" + "sync" "sync/atomic" "testing" "time" @@ -170,6 +173,57 @@ func TestNuGetCooldownColdDownload(t *testing.T) { } } +// TestNuGetCooldownConcurrentDownloads asserts that a download which joins +// another request's metadata fetch still applies the cooldown. Each caller +// decodes the leaf in its own validate callback, so a waiter on the shared +// fetch must run its own rather than finding no document and allowing it. +func TestNuGetCooldownConcurrentDownloads(t *testing.T) { + p, db, store, fetcher := setupTestProxy(t) + p.Cooldown = &cooldown.Config{Default: "14d"} + seedPackage(t, db, store, "nuget", "testpkg", "2.0.0", "testpkg.2.0.0.nupkg", "cached package") + entered, release := make(chan struct{}), make(chan struct{}) + var enteredOnce, releaseOnce sync.Once + releaseUpstream := func() { releaseOnce.Do(func() { close(release) }) } + var calls atomic.Int64 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + enteredOnce.Do(func() { close(entered) }) + <-release + _ = json.NewEncoder(w).Encode(map[string]string{"published": time.Now().Add(-time.Hour).Format(time.RFC3339)}) + })) + // Cleanups run last first: release the held request before closing the + // server, so a failed wait can't hang the test. + t.Cleanup(upstream.Close) + t.Cleanup(releaseUpstream) + routes := NewNuGetHandlerWithUpstreams(p, "http://proxy.test", upstream.URL, upstream.URL).Routes() + + statuses := make(chan int, 2) + download := func() { + w := httptest.NewRecorder() + routes.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v3-flatcontainer/testpkg/2.0.0/testpkg.2.0.0.nupkg", nil)) + statuses <- w.Code + } + go download() + <-entered + go download() + target := upstream.URL + nugetRegistrationPath + "testpkg/2.0.0.json" + key := metadataCoalesceKey("nuget", fmt.Sprintf("_cooldown/%x", sha256.Sum256([]byte(target))), target, contentTypeJSON, "", true) + waitForMetadataWaiters(t, p, key, 1) + releaseUpstream() + + for range 2 { + if status := <-statuses; status != http.StatusNotFound { + t.Errorf("download status = %d, want %d", status, http.StatusNotFound) + } + } + if got := calls.Load(); got != 1 { + t.Errorf("upstream metadata requests = %d, want 1", got) + } + if fetcher.fetchCalled { + t.Error("blocked downloads must not fetch artifacts") + } +} + func TestNuGetRegistrationServiceAliases(t *testing.T) { h := NewNuGetHandler(nugetTestProxy(), "http://proxy.test") for _, tt := range []struct{ service, path string }{