Read gem, Hex and conda cooldown metadata through the metadata cache

docs/configuration.md

@@ -1091,9 +1091,11 @@ Maven caveats:
 
 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, but the filtered result is kept for up to a minute (see [rewritten metadata cache](#rewritten-metadata-cache)), so a version leaving cooldown can show up a minute late; 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.
+
 With cooldown on, the proxy asks npm for full packuments, since the abbreviated ones it uses otherwise carry no publish times. Abbreviated packuments cached before cooldown was turned on are fetched again in full rather than served, including as a stale copy when the upstream is down.
 
-Conda cooldown covers `repodata.json`, `current_repodata.json`, `repodata_from_packages.json` and sharded repodata (CEP 16). Other forms of the repodata, such as `repodata.json.zst`, `repodata.json.bz2` and `repodata.jlap` patches, would list versions still in cooldown, so with cooldown on they are answered 404 and conda and mamba fall back to `repodata.json`. Clients cache shards by the hash in the shard index, so with cooldown on the proxy serves a rewritten index whose shard hashes change every hour and filters each shard as it is fetched. Versions leaving cooldown show up within the hour, and clients fetch the shards they need again once an hour. If a channel's shard index points packages or shards somewhere the proxy doesn't route, the proxy answers 404 for it and clients fall back to `repodata.json`.
+Conda cooldown covers `repodata.json`, `current_repodata.json`, `repodata_from_packages.json` and sharded repodata (CEP 16). Other forms of the repodata, such as `repodata.json.zst`, `repodata.json.bz2` and `repodata.jlap` patches, would list versions still in cooldown, so with cooldown on they are answered 404 and conda and mamba fall back to `repodata.json`. With `cache_metadata`, `repodata.json` is read through the metadata cache as it is without cooldown: within `metadata_ttl` the stored copy is filtered as it is read from storage, after it the upstream is asked with its ETag, and while the upstream is down the stale copy is filtered. Once a document has been filtered, the proxy remembers the ETag of the result until the first entry it left out leaves cooldown, so a client revalidating with `If-None-Match` gets `304 Not Modified` without the document being filtered or sent again. The first response for each document has no ETag, as it is sent while it is filtered. Without `cache_metadata`, `repodata.json` is fetched and filtered on every request. Clients cache shards by the hash in the shard index, so with cooldown on the proxy serves a rewritten index whose shard hashes change every hour and filters each shard as it is fetched. Versions leaving cooldown show up within the hour, and clients fetch the shards they need again once an hour. If a channel's shard index points packages or shards somewhere the proxy doesn't route, the proxy answers 404 for it and clients fall back to `repodata.json`.
 
 Note: Hex cooldown requires disabling registry signature verification since the proxy re-encodes the protobuf payload without the original signature. Set `HEX_NO_VERIFY_REPO_ORIGIN=1` or configure your repo with `no_verify: true`.
 
@@ -1346,7 +1348,7 @@ When upstream is unreachable and the cached entry is past its TTL, the proxy ser
 
 ### Metadata size limit
 
-Upstream metadata responses are buffered in memory before being rewritten and served. `metadata_max_size` caps that buffer to protect against OOM from a misbehaving upstream. Some npm packages with thousands of versions (for example `renovate`) exceed the 100 MB default, so raise this if you see `metadata response exceeds size limit` in the logs. Conda `repodata.json` with cooldown enabled is filtered as it streams and isn't held in memory, so this limit doesn't apply to it.
+Upstream metadata responses are buffered in memory before being rewritten and served. `metadata_max_size` caps that buffer to protect against OOM from a misbehaving upstream. Some npm packages with thousands of versions (for example `renovate`) exceed the 100 MB default, so raise this if you see `metadata response exceeds size limit` in the logs. Conda `repodata.json` with cooldown enabled is filtered as it streams and isn't held in memory decompressed. With `cache_metadata` the limit applies to the compressed document as it does without cooldown; one over it is filtered as it streams from the upstream and not cached.
 
 ```yaml
 metadata_max_size: "100MB"   # default
@@ -1366,7 +1368,7 @@ Or via environment variable: `PROXY_ARTIFACT_MAX_SIZE=2GB`.
 
 ### Rewritten metadata cache
 
-The npm, Composer, PyPI and Helm handlers rewrite every metadata document they serve, so that download URLs point at the proxy. That means decoding the whole document and encoding it again, and Composer's minified documents are also expanded, which together cost milliseconds per typical package and far more for very large ones; a large Helm `index.yaml` takes over a hundred milliseconds. The proxy keeps rewritten documents in memory, so each distinct upstream document is rewritten once and requests arriving while it is being rewritten wait for that rewrite. New bytes from upstream are rewritten again. A PyPI simple page is cached per representation (HTML or JSON) and per set of versions filtered out of it. With version cooldown enabled, a cached rewrite is kept for one minute only, because cooldown filtering depends on the current time: a version leaving cooldown shows up at most a minute late. Cached documents carry an `ETag` of their own, and are dated like any other, so clients that revalidate with `If-None-Match` or `If-Modified-Since`, such as Composer, get `304 Not Modified`.
+The npm, Composer, PyPI and Helm handlers rewrite every metadata document they serve, so that download URLs point at the proxy. That means decoding the whole document and encoding it again, and Composer's minified documents are also expanded, which together cost milliseconds per typical package and far more for very large ones; a large Helm `index.yaml` takes over a hundred milliseconds. The proxy keeps rewritten documents in memory, so each distinct upstream document is rewritten once and requests arriving while it is being rewritten wait for that rewrite. New bytes from upstream are rewritten again. A PyPI simple page is cached per representation (HTML or JSON) and per set of versions filtered out of it. With version cooldown enabled, a cached rewrite is kept for one minute only, because cooldown filtering depends on the current time: a version leaving cooldown shows up at most a minute late. Cached documents carry an `ETag` of their own, and are dated like any other, so clients that revalidate with `If-None-Match` or `If-Modified-Since`, such as Composer, get `304 Not Modified`. The RubyGems compact index and Hex packages filtered by cooldown are kept the same way; they are never served gzip-compressed.
 
 Each cached document is also kept gzip-compressed, when that makes it smaller, and served with `Content-Encoding: gzip` to clients that accept it. A large packument is compressed once rather than on every request, by the proxy or by a reverse proxy in front of it. The compressed copy counts against the cache size and typically adds about an eighth to each entry. Documents rewritten outside the cache are served uncompressed.
 

internal/handler/conda.go

@@ -2,12 +2,19 @@ package handler
 
 import (
 	"bufio"
+	"bytes"
 	"compress/gzip"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
+	"errors"
+	"fmt"
+	"hash"
 	"io"
 	"net/http"
 	"slices"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/git-pkgs/proxy/internal/jsonstream"
@@ -55,6 +62,11 @@ type CondaHandler struct {
 	proxy       *Proxy
 	upstreamURL string
 	proxyURL    string
+
+	// repodataStates holds, per metadata cache key, what is known of the
+	// last filtered copy of a repodata document. Guarded by repodataMu.
+	repodataMu     sync.Mutex
+	repodataStates map[string]repodataState
 }
 
 // NewCondaHandler creates a new Conda protocol handler.
@@ -215,14 +227,64 @@ func refuseUnfilteredRepodata(w http.ResponseWriter) {
 }
 
 // handleRepodata proxies repodata.json, applying cooldown filtering when enabled.
+//
+// With cache_metadata the document is read through the metadata cache, as
+// without cooldown, so within metadata_ttl it is filtered from storage
+// without asking upstream, and after it revalidated with its ETag. A stored
+// copy is streamed through the filter rather than read into memory; a
+// document just fetched is held compressed, as the metadata cache holds every
+// document it fetches. The decompressed document, hundreds of MB for
+// conda-forge, is never held whole.
 func (h *CondaHandler) handleRepodata(w http.ResponseWriter, r *http.Request) {
 	if !h.cooldownEnabled() {
 		h.proxyCached(w, r)
 		return
 	}
 
 	upstreamURL := h.upstreamURL + r.URL.Path
+	p := h.proxy
+	if !p.CacheMetadata || p.DB == nil || p.Storage == nil {
+		h.streamRepodata(w, r, upstreamURL)
+		return
+	}
+
+	cacheKey := h.metadataCacheKey(r.URL.Path)
+	now := time.Now()
+	if entry := p.freshMetadataEntry("conda", cacheKey); entry != nil {
+		if body, err := p.openMetadataEntry(r.Context(), "conda", cacheKey, entry); err == nil {
+			defer func() { _ = body.Close() }()
+			p.recordStreamedMetadataHit("conda", entry)
+			digest := entry.ContentDigest.String
+			if !h.repodataNotModified(w, r, cacheKey, digest, now) {
+				h.writeFilteredRepodata(w, upstreamURL, cacheKey, digest, body, entry.ContentEncoding.String, now)
+			}
+			return
+		}
+	}
 
+	// The same key and encoding as proxyCached, so with and without cooldown
+	// the proxy keeps one copy of the document.
+	res := p.lookupMetadata(r.Context(), "conda", cacheKey, upstreamURL, "*/*", "gzip", nil)
+	if errors.Is(res.err, ErrMetadataTooLarge) {
+		// Too large to cache even compressed: stream it as without the cache.
+		p.Logger.Warn("conda repodata exceeds the metadata size limit, streaming it uncached", "url", upstreamURL)
+		h.streamRepodata(w, r, upstreamURL)
+		return
+	}
+	if res.err != nil {
+		p.writeMetadataFetchError(w, res.err)
+		return
+	}
+	markStale(w, res.stale)
+	if h.repodataNotModified(w, r, cacheKey, res.digest, now) {
+		return
+	}
+	h.writeFilteredRepodata(w, upstreamURL, cacheKey, res.digest, bytes.NewReader(res.body), res.contentEncoding, now)
+}
+
+// streamRepodata filters repodata.json as it streams from upstream, without
+// the metadata cache.
+func (h *CondaHandler) streamRepodata(w http.ResponseWriter, r *http.Request, upstreamURL string) {
 	h.proxy.Logger.Debug("fetching repodata for cooldown filtering", "url", upstreamURL)
 
 	req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, upstreamURL, nil)
@@ -248,11 +310,21 @@ func (h *CondaHandler) handleRepodata(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Explicitly requesting gzip disables the transport's automatic decoding.
+	// Explicitly requesting gzip disables the transport's automatic decoding,
+	// so writeFilteredRepodata decodes it.
+	h.writeFilteredRepodata(w, upstreamURL, "", "", resp.Body, resp.Header.Get(headerContentEncoding), time.Now())
+}
+
+// writeFilteredRepodata writes repodata.json from src, decoded by
+// contentEncoding, without the entries in cooldown at now. digest identifies
+// the document in the metadata cache under cacheKey, and is empty when it
+// isn't cached; when known, the ETag of the filtered copy is remembered for
+// the next request.
+func (h *CondaHandler) writeFilteredRepodata(w http.ResponseWriter, upstreamURL, cacheKey, digest string, src io.Reader, contentEncoding string, now time.Time) {
 	// Cooldown needs JSON, so decode here and bound the decompressed size too.
-	var reader io.Reader = resp.Body
-	if strings.EqualFold(strings.TrimSpace(resp.Header.Get(headerContentEncoding)), "gzip") {
-		decoded, err := gzip.NewReader(resp.Body)
+	reader := src
+	if strings.EqualFold(strings.TrimSpace(contentEncoding), "gzip") {
+		decoded, err := gzip.NewReader(src)
 		if err != nil {
 			http.Error(w, "failed to decode response", http.StatusBadGateway)
 			return
@@ -267,7 +339,8 @@ func (h *CondaHandler) handleRepodata(w http.ResponseWriter, r *http.Request) {
 	out := &countingWriter{w: w}
 	buffered := bufio.NewWriterSize(out, repodataWriteBuffer)
 	w.Header().Set(headerContentType, "application/json")
-	err = h.filterRepodata(io.LimitReader(reader, maxRepodataSize), buffered)
+	filter := h.newRepodataFilter(now)
+	err := h.filterRepodataWith(io.LimitReader(reader, maxRepodataSize), buffered, filter)
 	if err == nil {
 		err = buffered.Flush()
 	}
@@ -279,6 +352,61 @@ func (h *CondaHandler) handleRepodata(w http.ResponseWriter, r *http.Request) {
 		}
 		panic(http.ErrAbortHandler)
 	}
+	if digest != "" {
+		h.rememberRepodata(cacheKey, digest, filter)
+	}
+}
+
+// repodataState is what the proxy remembers of the last filtered copy of a
+// cached repodata document: which document it was made from, its ETag, and
+// until when filtering that document again gives the same copy.
+type repodataState struct {
+	digest string
+	etag   string
+	// until is when the first entry left out leaves cooldown, or zero when
+	// none was left out.
+	until time.Time
+}
+
+// repodataNotModified sets the ETag of the filtered copy of the document
+// identified by digest when the proxy knows it, and answers a conditional
+// request it matches with 304, reporting whether it did. The ETag is known
+// once the document has been filtered, until an entry it left out leaves
+// cooldown. Filtering a large document costs seconds, and sending it
+// hundreds of MB, so this is what makes revalidating clients cheap.
+func (h *CondaHandler) repodataNotModified(w http.ResponseWriter, r *http.Request, cacheKey, digest string, now time.Time) bool {
+	if digest == "" {
+		return false
+	}
+	h.repodataMu.Lock()
+	state, ok := h.repodataStates[cacheKey]
+	h.repodataMu.Unlock()
+	if !ok || state.digest != digest || (!state.until.IsZero() && !now.Before(state.until)) {
+		return false
+	}
+	w.Header().Set(headerETag, state.etag)
+	if ifNoneMatchHits(r.Header.Get("If-None-Match"), state.etag) {
+		w.WriteHeader(http.StatusNotModified)
+		return true
+	}
+	return false
+}
+
+// rememberRepodata records the ETag of a filtered copy of the document
+// identified by digest. The copy depends only on the document, the startup
+// configuration and which entries were left out, and those stay left out
+// until the first of them leaves cooldown.
+func (h *CondaHandler) rememberRepodata(cacheKey, digest string, filter *repodataFilter) {
+	if filter.unknownUntil {
+		return
+	}
+	etag := keptVersionsETag(digest, "", cacheKey, []string{hex.EncodeToString(filter.withheld.Sum(nil))})
+	h.repodataMu.Lock()
+	defer h.repodataMu.Unlock()
+	if h.repodataStates == nil {
+		h.repodataStates = make(map[string]repodataState)
+	}
+	h.repodataStates[cacheKey] = repodataState{digest: digest, etag: etag, until: filter.until}
 }
 
 // countingWriter records whether anything has reached the response yet.
@@ -307,6 +435,11 @@ const (
 // packages and packages.conda that are still inside their cooldown. Kept
 // entries and every other member are written with their upstream bytes.
 func (h *CondaHandler) filterRepodata(r io.Reader, w io.Writer) error {
+	return h.filterRepodataWith(r, w, h.newRepodataFilter(time.Now()))
+}
+
+// filterRepodataWith is filterRepodata deciding by filter.
+func (h *CondaHandler) filterRepodataWith(r io.Reader, w io.Writer, filter *repodataFilter) error {
 	return jsonstream.CopyObject(json.NewDecoder(r), w, func(m *jsonstream.Member) error {
 		if m.Key != "packages" && m.Key != "packages.conda" {
 			return m.Copy()
@@ -316,7 +449,7 @@ func (h *CondaHandler) filterRepodata(r io.Reader, w io.Writer) error {
 			if err != nil {
 				return err
 			}
-			if h.inCooldown(entry.Key, raw) {
+			if filter.inCooldown(entry.Key, raw) {
 				return nil
 			}
 			return entry.Write(raw)
@@ -330,14 +463,50 @@ type repodataEntry struct {
 	Timestamp float64 `json:"timestamp"`
 }
 
+// repodataFilter decides which repodata entries cooldown leaves out, all as
+// of one time, and records what it left out, which with the document fixes
+// the filtered copy.
+type repodataFilter struct {
+	h   *CondaHandler
+	now time.Time
+	// withheld hashes the filenames left out.
+	withheld hash.Hash
+	// until is when the first entry left out leaves cooldown.
+	until time.Time
+	// unknownUntil is set when an entry was left out without a time it
+	// leaves cooldown, so until can't be relied on.
+	unknownUntil bool
+}
+
+func (h *CondaHandler) newRepodataFilter(now time.Time) *repodataFilter {
+	return &repodataFilter{h: h, now: now, withheld: sha256.New()}
+}
+
 // inCooldown reports whether a repodata entry was published too recently.
 // Entries without a name or timestamp are kept.
-func (h *CondaHandler) inCooldown(filename string, raw []byte) bool {
+func (f *repodataFilter) inCooldown(filename string, raw []byte) bool {
 	var entry repodataEntry
 	if err := json.Unmarshal(raw, &entry); err != nil {
 		return false
 	}
-	return h.recordInCooldown(filename, entry.Name, entry.Version, entry.Timestamp)
+	if entry.Name == "" || entry.Timestamp == 0 {
+		return false
+	}
+	publishedAt := time.Unix(int64(entry.Timestamp)/condaTimestampDivisor, 0)
+	decision := f.h.proxy.Cooldown.Evaluate("conda", canonicalPackagePURL("conda", entry.Name), publishedAt, f.now)
+	if decision.Allowed {
+		return false
+	}
+	f.h.proxy.Logger.Info("cooldown: filtering conda package",
+		"name", entry.Name, "version", entry.Version, "filename", filename)
+	_, _ = fmt.Fprintf(f.withheld, "%d:%s", len(filename), filename)
+	switch {
+	case decision.AvailableAt.IsZero():
+		f.unknownUntil = true
+	case f.until.IsZero() || decision.AvailableAt.Before(f.until):
+		f.until = decision.AvailableAt
+	}
+	return true
 }
 
 // recordInCooldown reports whether a package record with a millisecond