Expire cooldown rewrites when a withheld version becomes available

docs/configuration.md

@@ -1141,7 +1141,7 @@ 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 an hour (see [rewritten metadata cache](#rewritten-metadata-cache)), so a version leaving cooldown can show up that much later; 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.
+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.
 
 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. When `latest` is in cooldown, the packument tags the newest version out of cooldown instead, and `npm dist-tag ls` shows the same tags.
 
@@ -1420,7 +1420,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 expires, because cooldown filtering depends on the current time: it is kept for a 24th of the shortest cooldown configured, at least a minute and at most an hour, so a version shows up at most that long after its cooldown ends. With a cooldown of a day or more that is an hour; with `"2h"` somewhere, five minutes. 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.
+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 expires when the first version it held back leaves cooldown, since versions it let through stay allowed and a new upstream version changes the document. A rewrite that held nothing back is kept for the shortest cooldown configured, at least a minute and at most a day. 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 kept gzip-compressed, when that makes it smaller, and served with `Content-Encoding: gzip` to clients that accept it, as npm, pnpm, Yarn and Composer do. A large packument is compressed once rather than on every request, by the proxy or by a reverse proxy in front of it. Only the compressed copy is kept, typically an eighth of the document, and it is decompressed for the clients that don't accept gzip. Documents rewritten outside the cache are served uncompressed.
 

internal/handler/cargo.go

@@ -171,7 +171,7 @@ func (h *CargoHandler) applyCooldownFiltering(downstreamResponse http.ResponseWr
 
 		cratePURL := canonicalPackagePURL("cargo", crate.Name)
 
-		if !h.proxy.Cooldown.IsAllowed("cargo", cratePURL, publishedAt) {
+		if !h.proxy.cooldown().IsAllowed("cargo", cratePURL, publishedAt) {
 			h.proxy.Logger.Info("cooldown: filtering cargo version",
 				"crate", crate.Name, "version", crate.Version,
 				"published", crate.PublishTime)

internal/handler/composer.go

@@ -394,7 +394,7 @@ func (h *ComposerHandler) shouldFilterVersion(packagePURL, packageName, version
 		return false
 	}
 
-	if !h.proxy.Cooldown.IsAllowed("composer", packagePURL, publishedAt) {
+	if !h.proxy.cooldown().IsAllowed("composer", packagePURL, publishedAt) {
 		h.proxy.Logger.Info("cooldown: filtering composer version",
 			"package", packageName, "version", version)
 		return true
@@ -731,7 +731,7 @@ func (h *ComposerHandler) serveCachedRelease(w http.ResponseWriter, r *http.Requ
 		if err != nil || ver == nil || !ver.PublishedAt.Valid {
 			return false
 		}
-		if !h.proxy.Cooldown.IsAllowed("composer", canonicalPackagePURL("composer", packageName), ver.PublishedAt.Time) {
+		if !h.proxy.cooldown().IsAllowed("composer", canonicalPackagePURL("composer", packageName), ver.PublishedAt.Time) {
 			h.proxy.Logger.Info("cooldown: withholding composer archive",
 				"package", packageName, "version", version)
 			http.Error(w, "version not found", http.StatusNotFound)

internal/handler/conda.go

@@ -493,7 +493,7 @@ func (f *repodataFilter) inCooldown(filename string, raw []byte) bool {
 		return false
 	}
 	publishedAt := time.Unix(int64(entry.Timestamp)/condaTimestampDivisor, 0)
-	decision := f.h.proxy.Cooldown.Evaluate("conda", canonicalPackagePURL("conda", entry.Name), publishedAt, f.now)
+	decision := f.h.proxy.cooldown().Evaluate("conda", canonicalPackagePURL("conda", entry.Name), publishedAt, f.now)
 	if decision.Allowed {
 		return false
 	}
@@ -517,7 +517,7 @@ func (h *CondaHandler) recordInCooldown(filename, name, version string, timestam
 		return false
 	}
 	publishedAt := time.Unix(int64(timestamp)/condaTimestampDivisor, 0)
-	if h.proxy.Cooldown.IsAllowed("conda", canonicalPackagePURL("conda", name), publishedAt) {
+	if h.proxy.cooldown().IsAllowed("conda", canonicalPackagePURL("conda", name), publishedAt) {
 		return false
 	}
 	h.proxy.Logger.Info("cooldown: filtering conda package",

internal/handler/cooldown_expiry.go

@@ -0,0 +1,164 @@
+package handler
+
+import (
+	"sync"
+	"time"
+
+	"github.com/git-pkgs/cooldown"
+)
+
+// A rewrite made with cooldown on stays right until a version it held back
+// leaves cooldown: versions it let through stay allowed as time passes, and
+// new versions upstream change the document and so the cache key. So the
+// cached rewrite expires when the earliest version it withheld becomes
+// available, instead of after a fixed time.
+//
+// Handlers ask cooldown through Proxy.cooldown, which tells every rewrite
+// running at that moment when a withheld version becomes available. A
+// rewrite may hear of versions another rewrite withheld at the same time;
+// that only makes it expire earlier than it has to, never later.
+
+// maxCooldownRewriteTTL is the longest a rewrite made with cooldown on is
+// served, even when it held nothing back.
+const maxCooldownRewriteTTL = 24 * time.Hour
+
+// minCooldownRewriteTTL is the shortest cap on a rewrite made with cooldown
+// on, however short the cooldown.
+const minCooldownRewriteTTL = time.Minute
+
+// cooldownRewriteTTL returns the longest a rewrite made with cooldown on is
+// served: the shortest cooldown configured, between a minute and a day. It
+// only matters when a rewrite held nothing back, or held back a version
+// whose availability is unknown.
+func (p *Proxy) cooldownRewriteTTL() time.Duration {
+	ttl := p.ShortestCooldown
+	switch {
+	case ttl <= 0 || ttl > maxCooldownRewriteTTL:
+		return maxCooldownRewriteTTL
+	case ttl < minCooldownRewriteTTL:
+		return minCooldownRewriteTTL
+	}
+	return ttl
+}
+
+// cooldownWatch holds the rewrites running with cooldown on.
+type cooldownWatch struct {
+	mu     sync.Mutex
+	active map[*cooldownRecorder]struct{}
+}
+
+// cooldownRecorder collects when the versions withheld during one rewrite
+// become available.
+type cooldownRecorder struct {
+	// earliest is when the first withheld version becomes available, or
+	// zero if none was withheld.
+	earliest time.Time
+	// unknown is set when a version was withheld without a time it becomes
+	// available.
+	unknown bool
+}
+
+// watchCooldown starts recording the versions cooldown holds back for a
+// rewrite about to run. unwatchCooldown ends it.
+func (p *Proxy) watchCooldown() *cooldownRecorder {
+	rec := &cooldownRecorder{}
+	w := &p.cooldownWatch
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.active == nil {
+		w.active = make(map[*cooldownRecorder]struct{})
+	}
+	w.active[rec] = struct{}{}
+	return rec
+}
+
+func (p *Proxy) unwatchCooldown(rec *cooldownRecorder) {
+	w := &p.cooldownWatch
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	delete(w.active, rec)
+}
+
+// cooldownExpiry returns when a rewrite that started at start and recorded
+// into rec should expire: when the first version it held back leaves
+// cooldown, and no later than cooldownRewriteTTL after start.
+func (p *Proxy) cooldownExpiry(rec *cooldownRecorder, start time.Time) time.Time {
+	w := &p.cooldownWatch
+	w.mu.Lock()
+	earliest, unknown := rec.earliest, rec.unknown
+	w.mu.Unlock()
+
+	limit := start.Add(p.cooldownRewriteTTL())
+	if unknown || earliest.IsZero() || earliest.After(limit) {
+		return limit
+	}
+	return earliest
+}
+
+// withheld tells the running rewrites that a version is held back until
+// availableAt, zero when that is unknown.
+func (p *Proxy) withheld(availableAt time.Time) {
+	w := &p.cooldownWatch
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	for rec := range w.active {
+		switch {
+		case availableAt.IsZero():
+			rec.unknown = true
+		case rec.earliest.IsZero() || availableAt.Before(rec.earliest):
+			rec.earliest = availableAt
+		}
+	}
+}
+
+// cooldown returns the proxy's cooldown policy, telling the running rewrites
+// about each version it holds back. Handlers use it rather than p.Cooldown.
+func (p *Proxy) cooldown() CooldownPolicy { //nolint:ireturn // nil when cooldown is not configured, as p.Cooldown is
+	if p.Cooldown == nil {
+		return nil
+	}
+	return recordingCooldown{p: p}
+}
+
+// recordingCooldown is the proxy's cooldown policy, reporting the versions it
+// holds back to the running rewrites.
+type recordingCooldown struct {
+	p *Proxy
+}
+
+func (r recordingCooldown) IsAllowed(ecosystem, packagePURL string, publishedAt time.Time) bool {
+	if r.p.Cooldown.IsAllowed(ecosystem, packagePURL, publishedAt) {
+		return true
+	}
+	r.p.withheld(r.availableAt(ecosystem, packagePURL, publishedAt))
+	return false
+}
+
+func (r recordingCooldown) Evaluate(ecosystem, packagePURL string, publishedAt, evaluatedAt time.Time) cooldown.Decision {
+	d := r.p.Cooldown.Evaluate(ecosystem, packagePURL, publishedAt, evaluatedAt)
+	if !d.Allowed {
+		r.p.withheld(d.AvailableAt)
+	}
+	return d
+}
+
+func (r recordingCooldown) For(ecosystem, packagePURL string) time.Duration {
+	return r.p.Cooldown.For(ecosystem, packagePURL)
+}
+
+func (r recordingCooldown) Enabled() bool {
+	return r.p.Cooldown.Enabled()
+}
+
+// availableAt is when a version IsAllowed held back leaves cooldown, or zero
+// when that is unknown.
+func (r recordingCooldown) availableAt(ecosystem, packagePURL string, publishedAt time.Time) time.Time {
+	if publishedAt.IsZero() {
+		return time.Time{}
+	}
+	d := r.p.Cooldown.For(ecosystem, packagePURL)
+	if d <= 0 {
+		return time.Time{}
+	}
+	return publishedAt.Add(d)
+}