Check cooldown on artifact downloads for cargo, gem, hex, pub, conda and helm

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/database/database_test.go

@@ -1204,3 +1204,46 @@ func TestGetFetchedVersionPURLs(t *testing.T) {
 		}
 	})
 }
+
+// Publish times are written in UTC whatever zone the caller's time is in, so
+// a TIMESTAMP column without a zone reads back the same instant.
+func TestVersionPublishedAtIsUTC(t *testing.T) {
+	tokyo, err := time.LoadLocation("Asia/Tokyo")
+	if err != nil {
+		t.Skipf("time zone data unavailable: %v", err)
+	}
+	prev := time.Local
+	time.Local = tokyo
+	t.Cleanup(func() { time.Local = prev })
+
+	runWithBothDatabases(t, func(t *testing.T, db *DB) {
+		written := time.Date(2026, 10, 10, 21, 0, 1, 0, tokyo)
+		want := written.UTC()
+		check := func(purl string) {
+			t.Helper()
+			got, err := db.GetVersionByPURL(purl)
+			if err != nil || got == nil || !got.PublishedAt.Valid {
+				t.Fatalf("GetVersionByPURL(%s) = %+v, %v", purl, got, err)
+			}
+			// Postgres reads TIMESTAMP back as a fixed +0000 zone rather
+			// than time.UTC, so compare the offset.
+			if _, offset := got.PublishedAt.Time.Zone(); !got.PublishedAt.Time.Equal(want) || offset != 0 {
+				t.Errorf("%s PublishedAt read back as %v, want %v", purl, got.PublishedAt.Time, want)
+			}
+		}
+
+		if err := db.SetVersionPublishedAt("pkg:cargo/demo@1.0.0", "pkg:cargo/demo", written); err != nil {
+			t.Fatal(err)
+		}
+		check("pkg:cargo/demo@1.0.0")
+
+		if err := db.UpsertVersion(&Version{
+			PURL:        "pkg:cargo/demo@2.0.0",
+			PackagePURL: "pkg:cargo/demo",
+			PublishedAt: sql.NullTime{Time: written, Valid: true},
+		}); err != nil {
+			t.Fatal(err)
+		}
+		check("pkg:cargo/demo@2.0.0")
+	})
+}

internal/database/queries.go

@@ -185,7 +185,7 @@ func (db *DB) upsertVersion(exec execer, v *Version) error {
 
 	_, err := exec.Exec(query,
 		v.PURL, v.PackagePURL, v.License, v.Integrity,
-		v.PublishedAt, v.Yanked, v.EnrichedAt, now, now,
+		inUTC(v.PublishedAt), v.Yanked, v.EnrichedAt, now, now,
 	)
 	if err != nil {
 		return fmt.Errorf("upserting version: %w", err)
@@ -218,7 +218,7 @@ func (db *DB) SetVersionPublishedAt(versionPURL, packagePURL string, publishedAt
 		`
 	}
 
-	_, err := db.Exec(query, versionPURL, packagePURL, publishedAt, now, now)
+	_, err := db.Exec(query, versionPURL, packagePURL, publishedAt.UTC(), now, now)
 	if err != nil {
 		return fmt.Errorf("setting version publish time: %w", err)
 	}

internal/handler/cargo.go

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"bufio"
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -186,6 +187,32 @@ func (h *CargoHandler) applyCooldownFiltering(downstreamResponse http.ResponseWr
 	}
 }
 
+// publishTime reads a crate version's pubtime from the sparse index, sharing
+// the index cache with handleIndex. Entries published before crates.io
+// recorded pubtime have none.
+func (h *CargoHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) {
+	upstreamURL := fmt.Sprintf("%s/%s", h.indexURL, h.buildIndexPath(name))
+	body, _, err := h.proxy.FetchOrCacheMetadata(r.Context(), "cargo", name, upstreamURL, "text/plain")
+	if err != nil {
+		return time.Time{}, false, err
+	}
+	for line := range bytes.SplitSeq(body, []byte("\n")) {
+		if len(bytes.TrimSpace(line)) == 0 {
+			continue
+		}
+		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
+}
+
 // buildIndexPath builds the sparse index path for a crate name.
 func (h *CargoHandler) buildIndexPath(name string) string {
 	name = strings.ToLower(name)
@@ -217,6 +244,16 @@ 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, r, downloadVersion{
+		ecosystem: "cargo", policyName: name, storedName: name, version: version, cacheName: name, filename: filename,
+		lookup: func() (time.Time, bool, bool, error) {
+			published, ok, err := h.publishTime(r, name, version)
+			return published, ok, false, err
+		},
+	}) {
+		return
+	}
+
 	downloadURL := fmt.Sprintf(
 		"%s/%s/%s",
 		h.downloadURL,

internal/handler/conda.go

@@ -139,11 +139,21 @@ func (h *CondaHandler) handleDownload(w http.ResponseWriter, r *http.Request) {
 	h.proxy.Logger.Debug("conda download request",
 		"channel", channel, "arch", arch, "name", name, "version", version)
 
-	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.
+	// 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.
+	// 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, bool, error) {
+			published, ok, err := h.publishTime(r, channel, arch, filename)
+			return published, ok, false, err
+		},
+	}) {
+		return
+	}
 
+	upstreamURL := h.upstreamURL + r.URL.Path
 	result, err := h.proxy.GetOrFetchArtifactFromURL(r.Context(), "conda", packageName, version, cacheFilename, upstreamURL)
 	if err != nil {
 		h.proxy.serveArtifactError(w, err, "failed to fetch package")
@@ -531,6 +541,96 @@ func (h *CondaHandler) metadataCacheKey(urlPath string) string {
 	return hashedCacheKey(h.upstreamURL, urlPath)
 }
 
+// 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)
+	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
+	}
+	published, ok := index[filename]
+	return published, ok, nil
+}
+
+// indexRepodata indexes a cached repodata.json body.
+func (h *CondaHandler) indexRepodata(body []byte, encoding string) (map[string]time.Time, error) {
+	var reader io.Reader = bytes.NewReader(body)
+	if strings.EqualFold(strings.TrimSpace(encoding), "gzip") {
+		decoded, err := gzip.NewReader(reader)
+		if err != nil {
+			return nil, err
+		}
+		reader = decoded
+	}
+	return readRepodataIndex(reader)
+}
+
+// streamRepodataIndex indexes repodata.json as it streams from upstream.
+func (h *CondaHandler) streamRepodataIndex(r *http.Request, upstreamURL string) (map[string]time.Time, error) {
+	req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, upstreamURL, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set(headerAcceptEncoding, "gzip")
+	resp, err := h.proxy.HTTPClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("repodata returned %d", resp.StatusCode)
+	}
+	var reader io.Reader = resp.Body
+	if strings.EqualFold(strings.TrimSpace(resp.Header.Get(headerContentEncoding)), "gzip") {
+		decoded, err := gzip.NewReader(resp.Body)
+		if err != nil {
+			return nil, err
+		}
+		defer func() { _ = decoded.Close() }()
+		reader = decoded
+	}
+	return readRepodataIndex(reader)
+}
+
+// readRepodataIndex maps each package file in a repodata.json to its
+// timestamp, leaving out files without one. It reads the document as it
+// streams, under the same bound as filterRepodata, so the decompressed size
+// is not limited by metadata_max_size.
+func readRepodataIndex(r io.Reader) (map[string]time.Time, error) {
+	index := make(map[string]time.Time)
+	err := jsonstream.CopyObject(json.NewDecoder(io.LimitReader(r, maxRepodataSize)), io.Discard, func(m *jsonstream.Member) error {
+		if m.Key != "packages" && m.Key != "packages.conda" {
+			return m.Copy()
+		}
+		return m.Object(func(entry *jsonstream.Member) error {
+			raw, err := entry.Raw()
+			if err != nil {
+				return err
+			}
+			var e repodataEntry
+			if json.Unmarshal(raw, &e) == nil && e.Timestamp != 0 {
+				index[entry.Key] = time.Unix(int64(e.Timestamp)/condaTimestampDivisor, 0)
+			}
+			return nil
+		})
+	})
+	if err != nil {
+		return nil, err
+	}
+	return index, nil
+}
+
 // proxyCached forwards a metadata request with caching.
 func (h *CondaHandler) proxyCached(w http.ResponseWriter, r *http.Request) {
 	cacheKey := h.metadataCacheKey(r.URL.Path)

internal/handler/cooldown_times.go

@@ -7,6 +7,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
+	"slices"
 	"strings"
 	"time"
 )
@@ -16,6 +17,26 @@ import (
 // through.
 type publishTimes map[string]time.Time
 
+func (p *Proxy) lookupPublishTime(ctx context.Context, t timedMetadata, version string, listed func(metadataResult) ([]string, error)) (time.Time, bool, bool, error) {
+	var versions []string
+	docs, err := p.fetchTimedMetadata(ctx, t, func(doc metadataResult) ([]string, error) {
+		var err error
+		versions, err = listed(doc)
+		return versions, err
+	})
+	switch {
+	case docs.doc.err != nil:
+		return time.Time{}, false, false, docs.doc.err
+	case docs.times.err != nil:
+		return time.Time{}, false, false, docs.times.err
+	case err != nil:
+		return time.Time{}, false, false, err
+	}
+	published, ok := docs.published[version]
+	withholdUnknown := !ok && docs.withholdUnknown && slices.Contains(versions, version)
+	return published, ok && !published.IsZero(), withholdUnknown, 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.