From c68324f17d6b87b3a90e1d4a4563b8212a054622 Mon Sep 17 00:00:00 2001 From: montehurd Date: Fri, 2 Oct 2026 22:40:15 -0700 Subject: [PATCH 1/2] Coalesce concurrent metadata fetches Concurrent cache misses for one artifact already share a single upstream fetch (#329), but metadata misses did not: every request for a package's metadata went to the registry, even when the same document was already being fetched. In CI that is the common case. Composer resolves versions on every job, since MediaWiki core and its extensions commit no composer.lock, and each package is one /p2/ document, so a burst of jobs fetched each document once per job. Metadata misses now follow the same pattern as artifacts: the first caller fetches, the rest wait for its result, and the caller that takes the key rechecks the cache first so a fetch that just finished is not repeated. The key includes the Accept and Accept-Encoding headers, which change the bytes upstream returns (npm's abbreviated and full documents), and whether the caller validates the response. One difference from artifacts: the shared fetch is detached from the first caller's cancellation. Artifact fetches keep it for scanning and mirroring, neither of which applies to metadata, and otherwise one job disconnecting would fail every job waiting on the same document. runScan detaches the same way, and the HTTP client's timeout still bounds the fetch. This covers every ecosystem that goes through fetchOrCacheMetadata (Composer, npm, PyPI, Maven, Cargo, Swift, Pub, Helm, NuGet), with metadata caching on or off. Streamed metadata with caching off is not coalesced. On an 8 vCPU test instance with caching off, 10 concurrent clients requesting the npm metadata of 1,263 packages went from 12,630 upstream requests to 1,898. With cache_metadata on, 20 clients requesting 1,538 packages made exactly 1,538. --- internal/handler/handler.go | 161 ++++++++++--- internal/handler/metadata_coalesce_test.go | 249 +++++++++++++++++++++ 2 files changed, 383 insertions(+), 27 deletions(-) create mode 100644 internal/handler/metadata_coalesce_test.go diff --git a/internal/handler/handler.go b/internal/handler/handler.go index b04f941..5951828 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 @@ -915,28 +920,10 @@ func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, u 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 +932,54 @@ 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) + return 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) + }) +} + +// 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 +990,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 +1010,85 @@ 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. 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 { + 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()} + case <-f.done: + return f.res + } + } + + // 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 } 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) + } +} From 4856311e2450dceb6c90672048df35fd88a9e812 Mon Sep 17 00:00:00 2001 From: montehurd Date: Sat, 3 Oct 2026 15:47:53 -0700 Subject: [PATCH 2/2] Run each caller's validate on a shared metadata fetch A caller that joined another's metadata fetch got the shared bytes without running its own validate. NuGet's cooldown path uses validate to decode the document for the request, so a joined download found no publication date and was allowed, bypassing the cooldown. coalesceMetadata now reports whether the caller joined, and coalescedMetadataMiss runs that caller's validate on the shared bytes. Validation before caching is unchanged: the first caller still runs it inside the fetch. TestNuGetCooldownConcurrentDownloads reproduces the bypass through NuGetHandler.Routes() with two overlapping downloads of a version inside its cooldown. --- internal/handler/handler.go | 27 +++++++++---- internal/handler/nuget_cooldown_test.go | 54 +++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 7 deletions(-) diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 5951828..e9abd5a 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -914,7 +914,11 @@ 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) @@ -940,7 +944,7 @@ func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, u // 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) - return p.coalesceMetadata(ctx, key, func(fetchCtx context.Context) metadataResult { + 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 @@ -951,6 +955,14 @@ func (p *Proxy) coalescedMetadataMiss(ctx context.Context, ecosystem, cacheKey, } 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 @@ -1049,13 +1061,14 @@ func metadataCoalesceKey(ecosystem, cacheKey, upstreamURL, accept, acceptEncodin } // coalesceMetadata runs fetch at most once for concurrent callers sharing key -// and gives each the result. It follows coalesceFetch with one difference: +// 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 { +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) @@ -1072,9 +1085,9 @@ func (p *Proxy) coalesceMetadata(ctx context.Context, key string, fetch func(con if joined { select { case <-ctx.Done(): - return metadataResult{err: ctx.Err()} + return metadataResult{err: ctx.Err()}, true case <-f.done: - return f.res + return f.res, true } } @@ -1088,7 +1101,7 @@ func (p *Proxy) coalesceMetadata(ctx context.Context, key string, fetch func(con close(f.done) }() f.res = fetch(context.WithoutCancel(ctx)) - return f.res + 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/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 }{