Handle cooldown lookup failures and reuse download indexes

Author: andrew

docs/configuration.md

@@ -1201,6 +1201,10 @@ Maven caveats:
 
 Helm uses each release's `created` time from the repository's `index.yaml`. Releases in cooldown are left out of the index the proxy serves, and their archives aren't served; a release without `created` is let through. Charts pulled over OCI have no publish time and aren't held back.
 
+Cargo, RubyGems, Hex, pub.dev, Conda and Helm check artifact downloads against cooldown, including files cached before the policy changed. When a publish-time lookup fails, an already-cached file is still served; an uncached download fails with 502, or 503 with `Retry-After` when the rate-limit gate holds the request. Conda checks timestamps per file because builds of one version can have different publish times.
+
+Conda repodata and Helm indexes are parsed once per cached document for download checks. With `cache_metadata` off, the parsed index is kept for one minute, including the fetch, so an install can reuse it across files. Conda streams repodata when its compressed copy exceeds `metadata_max_size`, under the same decompressed size bound as the metadata filter.
+
 RubyGems, Hex and PyPI keep their publish times outside the metadata clients read: the RubyGems versions API, the Hex HTTP API (`upstream.hex_api`) and the PyPI JSON API. The proxy fetches them alongside each compact index, package or simple page it filters, and for PyPI downloads. Cooldown fails closed: when it applies to a package and those times can't be fetched or parsed, or the metadata can't be rewritten, the request fails with 502, or with 503 and `Retry-After` while the upstream is rate limiting the proxy, instead of serving the metadata unfiltered. The same goes for npm, Composer and pub.dev metadata that can't be rewritten, and for npm tarball downloads whose publish time can't be read. A PyPI file or npm tarball that is already cached is still served when its publish time can't be read; only a download that would have to be fetched fails. A PyPI release's time is the upload time of its newest file. The proxy stores it when a download looks it up, and serves files of that release that are already cached against the stored time without reading the JSON API again. A file that isn't cached yet is always looked up, since it may have been added to the release since, and the newer time is stored. Packages exempted with a duration of `0` are served as before and don't need the times. A PyPI package the JSON API returns 404 for has no times to go by and is served unfiltered.
 
 For RubyGems and Hex, both documents go through the metadata cache. With [`cache_metadata`](#metadata-caching) they are served from it within `metadata_ttl` without asking the upstream, revalidated with their ETags after it, and filtered from the last cached copies, marked stale, while the upstream is down; without it they are still fetched once for concurrent requests. Each document has its own TTL, so the compact index or Hex package can be newer than the cached times and list a version they don't have. The times are then fetched again before filtering, and if that fails, such versions are left out until the times can be read. The filtered index or package carries an ETag of its own, so Bundler and Mix revalidate it with `304 Not Modified`. Publish times are compared with the current time on each request, and the filtered result is kept until the first version it held back leaves cooldown (see [rewritten metadata cache](#rewritten-metadata-cache)); a newly published version shows up within `metadata_ttl`, as with other cached metadata. Upstream errors are answered as for other cached metadata (404, or 502) rather than relayed.

internal/handler/cargo.go

@@ -197,11 +197,17 @@ func (h *CargoHandler) publishTime(r *http.Request, name, version string) (time.
 		return time.Time{}, false, err
 	}
 	for line := range bytes.SplitSeq(body, []byte("\n")) {
-		if vers, _, _ := lookupJSONString(line, "vers"); vers != version {
+		if len(bytes.TrimSpace(line)) == 0 {
 			continue
 		}
-		pubtime, _, _ := lookupJSONString(line, "pubtime")
-		published, err := time.Parse(time.RFC3339, pubtime)
+		var entry crateIndexEntry
+		if err := json.Unmarshal(line, &entry); err != nil {
+			return time.Time{}, false, err
+		}
+		if entry.Version != version {
+			continue
+		}
+		published, err := time.Parse(time.RFC3339, entry.PublishTime)
 		return published, err == nil, nil
 	}
 	return time.Time{}, false, nil
@@ -238,8 +244,8 @@ func (h *CargoHandler) handleDownload(w http.ResponseWriter, r *http.Request) {
 	h.proxy.Logger.Debug("cargo download request",
 		"crate", name, "version", version, "filename", filename)
 
-	if h.proxy.withholdInCooldown(w, downloadVersion{
-		ecosystem: "cargo", policyName: name, storedName: name, version: version,
+	if h.proxy.withholdInCooldown(w, r, downloadVersion{
+		ecosystem: "cargo", policyName: name, storedName: name, version: version, cacheName: name, filename: filename,
 		lookup: func() (time.Time, bool, error) { return h.publishTime(r, name, version) },
 	}) {
 		return

internal/handler/conda.go

@@ -141,18 +141,16 @@ func (h *CondaHandler) handleDownload(w http.ResponseWriter, r *http.Request) {
 
 	// Builds and subdirs of one version each have their own timestamp, so
 	// the check is per file and keeps no time in the shared versions row.
-	if h.proxy.withholdInCooldown(w, downloadVersion{
-		ecosystem: "conda", policyName: name, version: version,
+	// Subdirs distinguish cached files with the same filename.
+	cacheFilename := arch + "/" + filename
+	if h.proxy.withholdInCooldown(w, r, downloadVersion{
+		ecosystem: "conda", policyName: name, version: version, cacheName: packageName, filename: cacheFilename,
 		lookup: func() (time.Time, bool, error) { return h.publishTime(r, channel, arch, filename) },
 	}) {
 		return
 	}
 
 	upstreamURL := h.upstreamURL + r.URL.Path
-	// The same filename can hold different bytes in each subdir, so the
-	// subdir is part of the cache key.
-	cacheFilename := arch + "/" + filename
-
 	result, err := h.proxy.GetOrFetchArtifactFromURL(r.Context(), "conda", packageName, version, cacheFilename, upstreamURL)
 	if err != nil {
 		h.proxy.serveArtifactError(w, err, "failed to fetch package")
@@ -540,26 +538,21 @@ func (h *CondaHandler) metadataCacheKey(urlPath string) string {
 	return hashedCacheKey(h.upstreamURL, urlPath)
 }
 
-// publishTime reads a package file's timestamp from the subdir's
-// repodata.json, fetched with the cache key and encoding proxyCached uses.
-// The repodata is indexed once per document (see publishIndex), since a
-// subdir's index runs to hundreds of megabytes and an install downloads many
-// files from it. A compressed copy over metadata_max_size can't be cached,
-// so it is streamed from upstream for each check, as handleRepodata does.
+// publishTime streams compressed repodata when it exceeds metadata_max_size.
 func (h *CondaHandler) publishTime(r *http.Request, channel, arch, filename string) (time.Time, bool, error) {
 	repodataPath := "/" + channel + "/" + arch + "/repodata.json"
 	cacheKey := h.metadataCacheKey(repodataPath)
-	res := h.proxy.lookupMetadata(r.Context(), "conda", cacheKey, h.upstreamURL+repodataPath, "*/*", "gzip", nil)
-	var index map[string]time.Time
-	var err error
-	switch {
-	case errors.Is(res.err, ErrMetadataTooLarge):
-		index, err = h.streamRepodataIndex(r, h.upstreamURL+repodataPath)
-	case res.err != nil:
-		err = res.err
-	default:
-		index, err = publishIndex(h.proxy, "conda\x00"+cacheKey, res, h.indexRepodata)
-	}
+	index, err := publishIndex(h.proxy, "conda\x00"+cacheKey, func() metadataResult {
+		return h.proxy.lookupMetadata(r.Context(), "conda", cacheKey, h.upstreamURL+repodataPath, "*/*", "gzip", nil)
+	}, func(res metadataResult) (map[string]time.Time, error) {
+		if errors.Is(res.err, ErrMetadataTooLarge) {
+			return h.streamRepodataIndex(r, h.upstreamURL+repodataPath)
+		}
+		if res.err != nil {
+			return nil, res.err
+		}
+		return h.indexRepodata(res.body, res.contentEncoding)
+	})
 	if err != nil {
 		return time.Time{}, false, err
 	}

internal/handler/cooldown_times.go

@@ -16,6 +16,21 @@ import (
 // through.
 type publishTimes map[string]time.Time
 
+func (p *Proxy) lookupPublishTime(ctx context.Context, t timedMetadata, version string) (time.Time, bool, error) {
+	var times publishTimes
+	validate := func(body []byte) error {
+		var err error
+		times, err = t.parseTimes(body)
+		return err
+	}
+	res := p.lookupMetadata(ctx, t.ecosystem, t.timesKey, t.timesURL, t.timesAccept, "", validate)
+	if res.err != nil {
+		return time.Time{}, false, res.err
+	}
+	published, ok := times[version]
+	return published, ok && !published.IsZero(), nil
+}
+
 // timedMetadata describes metadata whose publish times live in a second
 // document, as with the RubyGems compact index and the versions API, or a
 // Hex package and the Hex HTTP API. Both are read through the metadata cache.

internal/handler/download_cooldown.go

@@ -24,21 +24,22 @@ type downloadVersion struct {
 	// keeps no time, and every check asks lookup.
 	storedName string
 	version    string
+	cacheName  string
+	filename   string
 	lookup     publishTimeLookup
 }
 
 // withholdInCooldown reports whether a download is inside the cooldown
-// window, and if so answers it with a 404. Filtering metadata is not enough on
-// its own: lockfiles record artifact URLs, so clients reach the download path
-// without reading the metadata first, and a cached artifact would be served
-// even after the policy was raised.
-//
-// A publish time never changes, so the stored versions row is read first and
-// lookup is only asked for a version the proxy has no time for yet; the time
-// it finds is stored. A version with no usable publish time is allowed
-// through, matching how the metadata filters treat it.
-func (p *Proxy) withholdInCooldown(w http.ResponseWriter, d downloadVersion) bool {
-	if !p.versionInCooldown(d) {
+// window, or a lookup failure prevents an uncached download.
+func (p *Proxy) withholdInCooldown(w http.ResponseWriter, r *http.Request, d downloadVersion) bool {
+	withheld, err := p.versionInCooldown(d)
+	if err != nil {
+		if !p.serveCachedDespiteCooldownError(w, r, d.ecosystem, d.cacheName, d.version, d.filename, err) {
+			p.writeCooldownLookupError(w, d.ecosystem, d.policyName, err)
+		}
+		return true
+	}
+	if !withheld {
 		return false
 	}
 	p.Logger.Info("cooldown: withholding "+d.ecosystem+" download",
@@ -47,31 +48,26 @@ func (p *Proxy) withholdInCooldown(w http.ResponseWriter, d downloadVersion) boo
 	return true
 }
 
-func (p *Proxy) versionInCooldown(d downloadVersion) bool {
-	if p.Cooldown == nil || !p.Cooldown.Enabled() {
-		return false
-	}
+func (p *Proxy) versionInCooldown(d downloadVersion) (bool, error) {
 	policyKey := canonicalPackagePURL(d.ecosystem, d.policyName)
-	if p.Cooldown.For(d.ecosystem, policyKey) <= 0 {
-		return false
+	if !p.cooldownApplies(d.ecosystem, d.policyName) {
+		return false, nil
 	}
 
 	var versionPURL string
 	if d.storedName != "" {
 		versionPURL = canonicalVersionPURL(d.ecosystem, d.storedName, d.version)
 		if ver, err := p.DB.GetVersionByPURL(versionPURL); err == nil && ver != nil && ver.PublishedAt.Valid {
-			return !p.Cooldown.IsAllowed(d.ecosystem, policyKey, ver.PublishedAt.Time)
+			return !p.cooldown().IsAllowed(d.ecosystem, policyKey, ver.PublishedAt.Time), nil
 		}
 	}
 
 	published, ok, err := d.lookup()
 	if err != nil {
-		p.Logger.Warn("cooldown: could not read publish time for download check",
-			"ecosystem", d.ecosystem, "name", d.policyName, "version", d.version, "error", err)
-		return false
+		return false, err
 	}
 	if !ok {
-		return false
+		return false, nil
 	}
 
 	if versionPURL != "" {
@@ -81,7 +77,7 @@ func (p *Proxy) versionInCooldown(d downloadVersion) bool {
 		}
 	}
 
-	return !p.Cooldown.IsAllowed(d.ecosystem, policyKey, published)
+	return !p.cooldown().IsAllowed(d.ecosystem, policyKey, published), nil
 }
 
 // maxPublishIndexes bounds how many documents publishIndexCache keeps an
@@ -99,27 +95,40 @@ type publishIndexCache struct {
 }
 
 type publishIndexEntry struct {
-	digest string
-	index  any
+	digest  string
+	index   any
+	expires time.Time
 }
 
-// publishIndex returns build's index of res, reusing the one built for the
-// same document under key. res.digest identifies the document; without one,
-// as when the metadata cache is off, the index is built for this call alone.
-func publishIndex[V any](p *Proxy, key string, res metadataResult, build func([]byte, string) (map[string]V, error)) (map[string]V, error) {
+// publishIndex reuses parsed documents by digest, or for a minute without one.
+func publishIndex[V any](p *Proxy, key string, load func() metadataResult, build func(metadataResult) (V, error)) (V, error) { //nolint:ireturn // Supports timestamp maps and Helm lookups.
 	c := &p.publishIndexes
-	if res.digest == "" {
-		return build(res.body, res.contentEncoding)
+	fresh := func() (V, bool) {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+		if e, ok := c.entries[key]; ok && !e.expires.IsZero() && p.clock().Before(e.expires) {
+			return e.index.(V), true
+		}
+		var zero V
+		return zero, false
 	}
-	c.mu.Lock()
-	if e, ok := c.entries[key]; ok && e.digest == res.digest {
-		c.mu.Unlock()
-		return e.index.(map[string]V), nil
+	if index, ok := fresh(); ok {
+		return index, nil
 	}
-	c.mu.Unlock()
-
-	v, err, _ := c.group.Do(key+"\x00"+res.digest, func() (any, error) {
-		index, err := build(res.body, res.contentEncoding)
+	v, err, _ := c.group.Do(key, func() (any, error) {
+		if index, ok := fresh(); ok {
+			return index, nil
+		}
+		res := load()
+		if res.err == nil && res.digest != "" {
+			c.mu.Lock()
+			entry, ok := c.entries[key]
+			c.mu.Unlock()
+			if ok && entry.digest == res.digest {
+				return entry.index, nil
+			}
+		}
+		index, err := build(res)
 		if err != nil {
 			return nil, err
 		}
@@ -134,11 +143,16 @@ func publishIndex[V any](p *Proxy, key string, res metadataResult, build func([]
 				break
 			}
 		}
-		c.entries[key] = publishIndexEntry{digest: res.digest, index: index}
+		entry := publishIndexEntry{digest: res.digest, index: index}
+		if res.digest == "" || !p.CacheMetadata {
+			entry.expires = p.clock().Add(time.Minute)
+		}
+		c.entries[key] = entry
 		return index, nil
 	})
 	if err != nil {
-		return nil, err
+		var zero V
+		return zero, err
 	}
-	return v.(map[string]V), nil
+	return v.(V), nil
 }

internal/handler/download_cooldown_cache_test.go

@@ -0,0 +1,26 @@
+package handler
+
+import (
+	"net/http"
+	"testing"
+)
+
+func TestGemDownloadSharesPublishTimeMetadata(t *testing.T) {
+	u, proxy, h := gemCacheSetup(t)
+	wantStatus(t, serveGet(t, h, gemInfoPath), http.StatusOK, "")
+	seedPackage(t, proxy.DB, proxy.Storage.(*mockStorage), "gem", "testgem", "1.0.0", "testgem-1.0.0.gem", "gem bytes")
+	wantStatus(t, serveGet(t, h, "/gems/testgem-1.0.0.gem"), http.StatusOK, "")
+	if n, _ := u.counts(gemVersionsPath); n != 1 {
+		t.Fatalf("versions API fetched %d times, want 1", n)
+	}
+}
+
+func TestHexDownloadSharesPublishTimeMetadata(t *testing.T) {
+	u, proxy, h, _ := hexCacheSetup(t, 0)
+	wantStatus(t, serveGet(t, h, hexPackagePath), http.StatusOK, "")
+	seedPackage(t, proxy.DB, proxy.Storage.(*mockStorage), "hex", "testpkg", "1.0.0", "testpkg-1.0.0.tar", "hex bytes")
+	wantStatus(t, serveGet(t, h, "/tarballs/testpkg-1.0.0.tar"), http.StatusOK, "")
+	if n, _ := u.counts(hexAPIPath); n != 1 {
+		t.Fatalf("Hex API fetched %d times, want 1", n)
+	}
+}