From b09ac872646d293b06b3d3a09fd57837f7677f1e Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Sun, 25 Oct 2026 17:41:40 +0100 Subject: [PATCH] Read gem, Hex and conda cooldown metadata through the metadata cache With cooldown on, the RubyGems compact index and versions API, the Hex package and Hex API entry, and conda's repodata.json were fetched with plain HTTP requests on every client request: no TTL, no revalidation, no coalescing and no stale fallback, even with cache_metadata. Gem and Hex now read both documents through the metadata cache and filter the cached copies. The filtered result goes through the rewrite cache and is served with an ETag. When the times document is older than the index and misses a version, it is fetched again; if that fails the version is left out. Conda filters a fresh stored repodata.json as it reads it from storage, uses the metadata cache otherwise, and remembers the ETag of the filtered copy until a withheld entry leaves cooldown. --- docs/configuration.md | 8 +- internal/handler/conda.go | 185 ++++- .../handler/cooldown_metadata_cache_test.go | 644 ++++++++++++++++++ internal/handler/cooldown_times.go | 149 ++++ internal/handler/gem.go | 224 +++--- internal/handler/gem_test.go | 46 +- internal/handler/handler.go | 30 +- internal/handler/hex.go | 221 +++--- internal/handler/hex_test.go | 2 +- internal/handler/metadata_stream.go | 94 ++- internal/handler/relay_test.go | 11 +- internal/handler/upstream_test.go | 8 +- 12 files changed, 1306 insertions(+), 316 deletions(-) create mode 100644 internal/handler/cooldown_metadata_cache_test.go create mode 100644 internal/handler/cooldown_times.go diff --git a/docs/configuration.md b/docs/configuration.md index 5a877e38..564825b7 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1046,9 +1046,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. 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` and sharded repodata (CEP 16). 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` and sharded repodata (CEP 16). 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`. @@ -1297,7 +1299,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 @@ -1317,7 +1319,7 @@ Or via environment variable: `PROXY_ARTIFACT_MAX_SIZE=2GB`. ### Rewritten metadata cache -The npm and Composer 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. 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. 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. Such documents are dated like any other, so clients that revalidate with `If-Modified-Since`, such as Composer, get `304 Not Modified`. +The npm and Composer 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. 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. 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. Such documents are dated like any other, so clients that revalidate with `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. diff --git a/internal/handler/conda.go b/internal/handler/conda.go index 1142ccf2..1a14eb85 100644 --- a/internal/handler/conda.go +++ b/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. @@ -179,6 +191,14 @@ func (h *CondaHandler) parseFilename(filename string) (name, version string) { } // 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.proxy.Cooldown == nil || !h.proxy.Cooldown.Enabled() { h.proxyCached(w, r) @@ -186,7 +206,49 @@ func (h *CondaHandler) handleRepodata(w http.ResponseWriter, r *http.Request) { } 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) @@ -212,11 +274,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 @@ -231,7 +303,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() } @@ -243,6 +316,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. @@ -271,6 +399,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() @@ -280,7 +413,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) @@ -294,14 +427,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 diff --git a/internal/handler/cooldown_metadata_cache_test.go b/internal/handler/cooldown_metadata_cache_test.go new file mode 100644 index 00000000..08ec870d --- /dev/null +++ b/internal/handler/cooldown_metadata_cache_test.go @@ -0,0 +1,644 @@ +package handler + +import ( + "bytes" + "compress/gzip" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/git-pkgs/cooldown" +) + +// countingUpstream serves fixed documents by path with ETags, answers a +// matching If-None-Match with 304, and counts what it was asked. Setting +// down makes it answer 500, and closing gate holds requests until opened. +type countingUpstream struct { + *httptest.Server + mu sync.Mutex + docs map[string][]byte + gzipped map[string]bool + requests map[string]int + notModified map[string]int + down map[string]bool + gate chan struct{} +} + +func newCountingUpstream(t *testing.T) *countingUpstream { + t.Helper() + u := &countingUpstream{ + docs: map[string][]byte{}, gzipped: map[string]bool{}, + requests: map[string]int{}, notModified: map[string]int{}, down: map[string]bool{}, + } + u.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + u.mu.Lock() + u.requests[r.URL.Path]++ + gate := u.gate + doc, ok := u.docs[r.URL.Path] + down := u.down[r.URL.Path] + gz := u.gzipped[r.URL.Path] + u.mu.Unlock() + if gate != nil { + <-gate + } + switch { + case down: + w.WriteHeader(http.StatusInternalServerError) + return + case !ok: + w.WriteHeader(http.StatusNotFound) + return + } + sum := sha256.Sum256(doc) + etag := `"` + hex.EncodeToString(sum[:8]) + `"` + w.Header().Set("ETag", etag) + if r.Header.Get("If-None-Match") == etag { + u.mu.Lock() + u.notModified[r.URL.Path]++ + u.mu.Unlock() + w.WriteHeader(http.StatusNotModified) + return + } + if gz && strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") { + w.Header().Set("Content-Encoding", "gzip") + zw := gzip.NewWriter(w) + _, _ = zw.Write(doc) + _ = zw.Close() + return + } + _, _ = w.Write(doc) + })) + t.Cleanup(u.Close) + return u +} + +func (u *countingUpstream) set(path string, doc []byte) { + u.mu.Lock() + defer u.mu.Unlock() + u.docs[path] = doc +} + +func (u *countingUpstream) setDown(path string, down bool) { + u.mu.Lock() + defer u.mu.Unlock() + u.down[path] = down +} + +func (u *countingUpstream) counts(path string) (requests, notModified int) { + u.mu.Lock() + defer u.mu.Unlock() + return u.requests[path], u.notModified[path] +} + +// cooldownCacheProxy returns a proxy with cooldown and the metadata cache on. +func cooldownCacheProxy(t *testing.T, u *countingUpstream) *Proxy { + t.Helper() + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = u.Client() + config := &cooldown.Config{Default: "3d"} + // Config parses itself on first use, which isn't safe to race. + config.Enabled() + proxy.Cooldown = config + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + return proxy +} + +// expireCachedMetadata moves a metadata cache row's fetch time past the TTL. +func expireCachedMetadata(t *testing.T, proxy *Proxy, ecosystem, key string) { + t.Helper() + entry, err := proxy.DB.GetMetadataCache(ecosystem, key) + if err != nil || entry == nil { + t.Fatalf("metadata entry %s/%s: %v", ecosystem, key, err) + } + entry.FetchedAt.Time = time.Now().Add(-2 * proxy.MetadataTTL) + if err := proxy.DB.UpsertMetadataCache(entry); err != nil { + t.Fatalf("expiring metadata entry: %v", err) + } +} + +func getWith(t *testing.T, h http.Handler, path string, header http.Header) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodGet, path, nil) + for k, vv := range header { + req.Header[k] = vv + } + w := httptest.NewRecorder() + h.ServeHTTP(w, req) + return w +} + +const ( + gemInfoPath = "/info/testgem" + gemVersionsPath = "/api/v1/versions/testgem.json" +) + +func gemVersionsJSON(t *testing.T, versions map[string]time.Duration) []byte { + t.Helper() + var list []gemVersion + for number, age := range versions { + list = append(list, gemVersion{Number: number, Platform: "ruby", CreatedAt: time.Now().Add(-age).Format(time.RFC3339)}) + } + b, err := json.Marshal(list) + if err != nil { + t.Fatal(err) + } + return b +} + +// gemCacheSetup serves testgem with 1.0.0 published a week ago and 2.0.0 an +// hour ago. +func gemCacheSetup(t *testing.T) (*countingUpstream, *Proxy, http.Handler) { + t.Helper() + u := newCountingUpstream(t) + u.set(gemInfoPath, []byte("---\n1.0.0 |checksum:a\n2.0.0 |checksum:b\n")) + u.set(gemVersionsPath, gemVersionsJSON(t, map[string]time.Duration{"1.0.0": 7 * 24 * time.Hour, "2.0.0": time.Hour})) + proxy := cooldownCacheProxy(t, u) + return u, proxy, NewGemHandlerWithUpstream(proxy, "http://proxy.test", u.URL).Routes() +} + +const gemFiltered = "---\n1.0.0 |checksum:a\n" + +func TestGemCooldownServesFromMetadataCache(t *testing.T) { + u, _, h := gemCacheSetup(t) + for i := range 3 { + w := getWith(t, h, gemInfoPath, nil) + if w.Code != http.StatusOK || w.Body.String() != gemFiltered { + t.Fatalf("request %d: status %d, body %q", i, w.Code, w.Body) + } + } + for _, path := range []string{gemInfoPath, gemVersionsPath} { + if n, _ := u.counts(path); n != 1 { + t.Errorf("%s requested %d times, want 1 within the TTL", path, n) + } + } +} + +func TestGemCooldownRevalidatesAfterTTL(t *testing.T) { + u, proxy, h := gemCacheSetup(t) + getWith(t, h, gemInfoPath, nil) + expireCachedMetadata(t, proxy, "gem", "info/testgem") + expireCachedMetadata(t, proxy, "gem", "api/v1/versions/testgem.json") + + w := getWith(t, h, gemInfoPath, nil) + if w.Code != http.StatusOK || w.Body.String() != gemFiltered { + t.Fatalf("status %d, body %q", w.Code, w.Body) + } + for _, path := range []string{gemInfoPath, gemVersionsPath} { + if n, nm := u.counts(path); n != 2 || nm != 1 { + t.Errorf("%s: %d requests, %d answered 304; want 2 and 1", path, n, nm) + } + } +} + +func TestGemCooldownConcurrentRequestsShareFetch(t *testing.T) { + u, proxy, h := gemCacheSetup(t) + u.gate = make(chan struct{}) + var release sync.Once + t.Cleanup(func() { release.Do(func() { close(u.gate) }) }) + + const clients = 5 + var wg sync.WaitGroup + codes := make([]int, clients) + bodies := make([]string, clients) + for i := range clients { + wg.Go(func() { + w := getWith(t, h, gemInfoPath, nil) + codes[i], bodies[i] = w.Code, w.Body.String() + }) + } + waitForMetadataWaiters(t, proxy, metadataCoalesceKey("gem", "info/testgem", u.URL+gemInfoPath, "*/*", "identity", false), clients-1) + waitForMetadataWaiters(t, proxy, metadataCoalesceKey("gem", "api/v1/versions/testgem.json", u.URL+gemVersionsPath, contentTypeJSON, "", true), clients-1) + release.Do(func() { close(u.gate) }) + wg.Wait() + + for i := range clients { + if codes[i] != http.StatusOK || bodies[i] != gemFiltered { + t.Errorf("client %d: status %d, body %q", i, codes[i], bodies[i]) + } + } + for _, path := range []string{gemInfoPath, gemVersionsPath} { + if n, _ := u.counts(path); n != 1 { + t.Errorf("%s requested %d times, want 1", path, n) + } + } +} + +func TestGemCooldownStaleCopyIsFiltered(t *testing.T) { + u, proxy, h := gemCacheSetup(t) + getWith(t, h, gemInfoPath, nil) + expireCachedMetadata(t, proxy, "gem", "info/testgem") + expireCachedMetadata(t, proxy, "gem", "api/v1/versions/testgem.json") + u.setDown(gemInfoPath, true) + u.setDown(gemVersionsPath, true) + + w := getWith(t, h, gemInfoPath, nil) + if w.Code != http.StatusOK || w.Body.String() != gemFiltered { + t.Fatalf("status %d, body %q; want the stale copy, filtered", w.Code, w.Body) + } + if got := w.Header().Get("Warning"); got != staleWarning { + t.Errorf("Warning = %q, want %q", got, staleWarning) + } +} + +func TestGemCooldownFailsClosedWithoutTimes(t *testing.T) { + u, _, h := gemCacheSetup(t) + u.setDown(gemVersionsPath, true) + + w := getWith(t, h, gemInfoPath, nil) + if w.Code != http.StatusBadGateway || strings.Contains(w.Body.String(), "1.0.0") { + t.Fatalf("status %d, body %q; want 502 without the index", w.Code, w.Body) + } + // The index alone was cached; it must not be served unfiltered either. + w = getWith(t, h, gemInfoPath, nil) + if w.Code != http.StatusBadGateway { + t.Fatalf("second request: status %d, want 502", w.Code) + } +} + +func TestGemCooldownClientRevalidates(t *testing.T) { + _, _, h := gemCacheSetup(t) + first := getWith(t, h, gemInfoPath, nil) + etag := first.Header().Get("ETag") + if etag == "" { + t.Fatal("filtered index served without an ETag") + } + w := getWith(t, h, gemInfoPath, http.Header{"If-None-Match": {etag}}) + if w.Code != http.StatusNotModified { + t.Fatalf("status %d, want 304", w.Code) + } +} + +// The index can be fetched again while the versions API response is still +// within its TTL. A version published in between is then in the index but +// has no publish time, and must not be let through for that. +func TestGemCooldownRefreshesTimesOlderThanIndex(t *testing.T) { + for _, timesDown := range []bool{false, true} { + name := "times refreshed" + if timesDown { + name = "times upstream down" + } + t.Run(name, func(t *testing.T) { + u, proxy, h := gemCacheSetup(t) + getWith(t, h, gemInfoPath, nil) + + u.set(gemInfoPath, []byte("---\n1.0.0 |checksum:a\n2.0.0 |checksum:b\n3.0.0 |checksum:c\n")) + u.set(gemVersionsPath, gemVersionsJSON(t, map[string]time.Duration{ + "1.0.0": 7 * 24 * time.Hour, "2.0.0": time.Hour, "3.0.0": time.Minute, + })) + u.setDown(gemVersionsPath, timesDown) + expireCachedMetadata(t, proxy, "gem", "info/testgem") + + w := getWith(t, h, gemInfoPath, nil) + if w.Code != http.StatusOK || w.Body.String() != gemFiltered { + t.Fatalf("status %d, body %q; want %q", w.Code, w.Body, gemFiltered) + } + if n, _ := u.counts(gemVersionsPath); n != 2 { + t.Errorf("versions API requested %d times, want 2", n) + } + if got := w.Header().Get("Warning"); (got != "") != timesDown { + t.Errorf("Warning = %q with times upstream down = %v", got, timesDown) + } + }) + } +} + +const ( + hexPackagePath = "/packages/testpkg" + hexAPIPath = "/api/packages/testpkg" +) + +func hexAPIJSON(t *testing.T, releases map[string]time.Duration) []byte { + t.Helper() + var api hexPackageAPI + for version, age := range releases { + api.Releases = append(api.Releases, hexRelease{Version: version, InsertedAt: time.Now().Add(-age).Format(time.RFC3339Nano)}) + } + b, err := json.Marshal(api) + if err != nil { + t.Fatal(err) + } + return b +} + +func hexCacheSetup(t *testing.T, recent time.Duration) (*countingUpstream, *Proxy, http.Handler, []byte) { + t.Helper() + u := newCountingUpstream(t) + signed := buildHexSigned(buildHexPackage("testpkg", []string{testVersion100, "2.0.0"})) + u.set(hexPackagePath, signed) + u.set(hexAPIPath, hexAPIJSON(t, map[string]time.Duration{testVersion100: 7 * 24 * time.Hour, "2.0.0": recent})) + proxy := cooldownCacheProxy(t, u) + return u, proxy, NewHexHandlerWithUpstreams(proxy, "http://proxy.test", u.URL, u.URL).Routes(), signed +} + +// hexVersions decodes the release versions of a served package. +func hexVersions(t *testing.T, body []byte) []string { + t.Helper() + zr, err := gzip.NewReader(bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + signed, err := io.ReadAll(zr) + if err != nil { + t.Fatal(err) + } + payload, err := extractProtobufBytes(signed, 1) + if err != nil { + t.Fatal(err) + } + versions, err := packageReleaseVersions(payload) + if err != nil { + t.Fatal(err) + } + return versions +} + +func TestHexCooldownServesFromMetadataCache(t *testing.T) { + u, proxy, h, _ := hexCacheSetup(t, time.Hour) + for i := range 3 { + w := getWith(t, h, hexPackagePath, nil) + if w.Code != http.StatusOK { + t.Fatalf("request %d: status %d", i, w.Code) + } + if got := hexVersions(t, w.Body.Bytes()); len(got) != 1 || got[0] != testVersion100 { + t.Fatalf("request %d: versions %v, want [%s]", i, got, testVersion100) + } + if ce := w.Header().Get("Content-Encoding"); ce != "gzip" { + t.Errorf("Content-Encoding = %q, want gzip", ce) + } + } + for _, path := range []string{hexPackagePath, hexAPIPath} { + if n, _ := u.counts(path); n != 1 { + t.Errorf("%s requested %d times, want 1 within the TTL", path, n) + } + } + + expireCachedMetadata(t, proxy, "hex", "packages/testpkg") + expireCachedMetadata(t, proxy, "hex", hashedCacheKey(u.URL, "api/packages/testpkg")) + if w := getWith(t, h, hexPackagePath, nil); w.Code != http.StatusOK || len(hexVersions(t, w.Body.Bytes())) != 1 { + t.Fatalf("after TTL: status %d", w.Code) + } + for _, path := range []string{hexPackagePath, hexAPIPath} { + if n, nm := u.counts(path); n != 2 || nm != 1 { + t.Errorf("%s: %d requests, %d answered 304; want 2 and 1", path, n, nm) + } + } +} + +func TestHexCooldownStaleCopyIsFiltered(t *testing.T) { + u, proxy, h, _ := hexCacheSetup(t, time.Hour) + getWith(t, h, hexPackagePath, nil) + expireCachedMetadata(t, proxy, "hex", "packages/testpkg") + expireCachedMetadata(t, proxy, "hex", hashedCacheKey(u.URL, "api/packages/testpkg")) + u.setDown(hexPackagePath, true) + u.setDown(hexAPIPath, true) + + w := getWith(t, h, hexPackagePath, nil) + if w.Code != http.StatusOK { + t.Fatalf("status %d, want the stale copy", w.Code) + } + if got := hexVersions(t, w.Body.Bytes()); len(got) != 1 || got[0] != testVersion100 { + t.Errorf("versions %v, want [%s]", got, testVersion100) + } + if got := w.Header().Get("Warning"); got != staleWarning { + t.Errorf("Warning = %q, want %q", got, staleWarning) + } +} + +func TestHexCooldownFailsClosedWithoutTimes(t *testing.T) { + u, _, h, _ := hexCacheSetup(t, time.Hour) + u.setDown(hexAPIPath, true) + for i := range 2 { + if w := getWith(t, h, hexPackagePath, nil); w.Code != http.StatusBadGateway { + t.Fatalf("request %d: status %d, want 502", i, w.Code) + } + } +} + +func TestHexCooldownUnfilteredKeepsSignature(t *testing.T) { + _, _, h, signed := hexCacheSetup(t, 7*24*time.Hour) + w := getWith(t, h, hexPackagePath, nil) + if w.Code != http.StatusOK || !bytes.Equal(w.Body.Bytes(), signed) { + t.Fatalf("status %d; want upstream's signed bytes when nothing is in cooldown", w.Code) + } + etag := w.Header().Get("ETag") + if etag == "" { + t.Fatal("package served without an ETag") + } + if w := getWith(t, h, hexPackagePath, http.Header{"If-None-Match": {etag}}); w.Code != http.StatusNotModified { + t.Fatalf("revalidation: status %d, want 304", w.Code) + } +} + +const condaRepodataPath = "/conda-forge/noarch/repodata.json" + +func condaCacheSetup(t *testing.T) (*countingUpstream, *Proxy, *CondaHandler, http.Handler) { + t.Helper() + u := newCountingUpstream(t) + repodata, err := json.Marshal(map[string]any{ + "info": map[string]any{"subdir": "noarch"}, + "packages": map[string]any{ + "old-1.0.0-0.tar.bz2": map[string]any{"name": "old", "version": "1.0.0", "timestamp": time.Now().Add(-7 * 24 * time.Hour).UnixMilli()}, + "new-2.0.0-0.tar.bz2": map[string]any{"name": "new", "version": "2.0.0", "timestamp": time.Now().Add(-time.Hour).UnixMilli()}, + }, + "packages.conda": map[string]any{}, + }) + if err != nil { + t.Fatal(err) + } + u.set(condaRepodataPath, repodata) + u.gzipped[condaRepodataPath] = true + proxy := cooldownCacheProxy(t, u) + h := NewCondaHandlerWithUpstream(proxy, "http://proxy.test", u.URL) + return u, proxy, h, h.Routes() +} + +func condaPackages(t *testing.T, w *httptest.ResponseRecorder) []string { + t.Helper() + if w.Code != http.StatusOK { + t.Fatalf("status %d, body %q", w.Code, w.Body) + } + var doc struct { + Packages map[string]json.RawMessage `json:"packages"` + } + if err := json.Unmarshal(w.Body.Bytes(), &doc); err != nil { + t.Fatalf("decoding repodata: %v", err) + } + var names []string + for name := range doc.Packages { + names = append(names, name) + } + return names +} + +func assertCondaFiltered(t *testing.T, w *httptest.ResponseRecorder) { + t.Helper() + if got := condaPackages(t, w); len(got) != 1 || got[0] != "old-1.0.0-0.tar.bz2" { + t.Fatalf("packages %v, want only old-1.0.0-0.tar.bz2", got) + } +} + +func TestCondaCooldownServesFromMetadataCache(t *testing.T) { + u, proxy, h, routes := condaCacheSetup(t) + for range 3 { + assertCondaFiltered(t, getWith(t, routes, condaRepodataPath, nil)) + } + if n, _ := u.counts(condaRepodataPath); n != 1 { + t.Errorf("repodata requested %d times, want 1 within the TTL", n) + } + + expireCachedMetadata(t, proxy, "conda", h.metadataCacheKey(condaRepodataPath)) + assertCondaFiltered(t, getWith(t, routes, condaRepodataPath, nil)) + if n, nm := u.counts(condaRepodataPath); n != 2 || nm != 1 { + t.Errorf("%d requests, %d answered 304; want 2 and 1", n, nm) + } +} + +func TestCondaCooldownStaleCopyIsFiltered(t *testing.T) { + u, proxy, h, routes := condaCacheSetup(t) + assertCondaFiltered(t, getWith(t, routes, condaRepodataPath, nil)) + expireCachedMetadata(t, proxy, "conda", h.metadataCacheKey(condaRepodataPath)) + u.setDown(condaRepodataPath, true) + + w := getWith(t, routes, condaRepodataPath, nil) + assertCondaFiltered(t, w) + if got := w.Header().Get("Warning"); got != staleWarning { + t.Errorf("Warning = %q, want %q", got, staleWarning) + } +} + +func TestCondaCooldownConcurrentRequestsShareFetch(t *testing.T) { + u, proxy, h, routes := condaCacheSetup(t) + u.gate = make(chan struct{}) + var release sync.Once + t.Cleanup(func() { release.Do(func() { close(u.gate) }) }) + + const clients = 5 + var wg sync.WaitGroup + recorders := make([]*httptest.ResponseRecorder, clients) + for i := range clients { + wg.Go(func() { recorders[i] = getWith(t, routes, condaRepodataPath, nil) }) + } + key := metadataCoalesceKey("conda", h.metadataCacheKey(condaRepodataPath), u.URL+condaRepodataPath, "*/*", "gzip", false) + waitForMetadataWaiters(t, proxy, key, clients-1) + release.Do(func() { close(u.gate) }) + wg.Wait() + + for _, w := range recorders { + assertCondaFiltered(t, w) + } + if n, _ := u.counts(condaRepodataPath); n != 1 { + t.Errorf("repodata requested %d times, want 1", n) + } +} + +func TestCondaCooldownClientRevalidates(t *testing.T) { + _, _, h, routes := condaCacheSetup(t) + // The first copy is streamed as it is filtered, before its ETag is known. + assertCondaFiltered(t, getWith(t, routes, condaRepodataPath, nil)) + w := getWith(t, routes, condaRepodataPath, nil) + assertCondaFiltered(t, w) + etag := w.Header().Get("ETag") + if etag == "" { + t.Fatal("filtered repodata served without an ETag") + } + if w := getWith(t, routes, condaRepodataPath, http.Header{"If-None-Match": {etag}}); w.Code != http.StatusNotModified { + t.Fatalf("revalidation: status %d, want 304", w.Code) + } + + // Once the entry left out leaves cooldown, the copy changes and so does + // its ETag. + key := h.metadataCacheKey(condaRepodataPath) + h.repodataMu.Lock() + state := h.repodataStates[key] + h.repodataMu.Unlock() + if want := time.Now().Add(-time.Hour + 3*24*time.Hour); state.until.Sub(want).Abs() > time.Minute { + t.Errorf("ETag kept until %v, want about %v", state.until, want) + } + req := httptest.NewRequest(http.MethodGet, condaRepodataPath, nil) + req.Header.Set("If-None-Match", etag) + if h.repodataNotModified(httptest.NewRecorder(), req, key, state.digest, state.until) { + t.Error("ETag still matched after the withheld entry left cooldown") + } +} + +func TestCondaCooldownWithoutMetadataCacheStreams(t *testing.T) { + u, proxy, _, routes := condaCacheSetup(t) + proxy.CacheMetadata = false + for range 2 { + assertCondaFiltered(t, getWith(t, routes, condaRepodataPath, nil)) + } + if n, _ := u.counts(condaRepodataPath); n != 2 { + t.Errorf("repodata requested %d times, want 2 without cache_metadata", n) + } +} + +// A cached repodata.json is filtered as it is read from storage: the +// response starts while the stored document is still being read. +func TestCondaCooldownStreamsCachedRepodata(t *testing.T) { + u, proxy, _, routes := condaCacheSetup(t) + packages := map[string]any{} + for i := range 20000 { + packages[fmt.Sprintf("pkg%d-1.0.0-0.tar.bz2", i)] = map[string]any{ + "name": fmt.Sprintf("pkg%d", i), "version": "1.0.0", "timestamp": time.Now().Add(-7 * 24 * time.Hour).UnixMilli(), + } + } + repodata, err := json.Marshal(map[string]any{"info": map[string]any{}, "packages": packages}) + if err != nil { + t.Fatal(err) + } + u.set(condaRepodataPath, repodata) + // Stored uncompressed, so the document is larger than the part the + // storage hands out before stalling. + u.gzipped[condaRepodataPath] = false + if w := getWith(t, routes, condaRepodataPath, nil); w.Code != http.StatusOK { + t.Fatalf("populating cache: status %d", w.Code) + } + + gate := &stallingStorage{Storage: proxy.Storage, head: 256 << 10, release: make(chan struct{})} + proxy.Storage = gate + var release sync.Once + t.Cleanup(func() { release.Do(func() { close(gate.release) }) }) + srv := httptest.NewServer(routes) + t.Cleanup(srv.Close) + + type result struct { + resp *http.Response + err error + } + started := make(chan result, 1) + go func() { + resp, err := http.Get(srv.URL + condaRepodataPath) + if err == nil { + _, err = io.ReadFull(resp.Body, make([]byte, 1024)) + } + started <- result{resp, err} + }() + var res result + select { + case res = <-started: + case <-time.After(5 * time.Second): + release.Do(func() { close(gate.release) }) + if res := <-started; res.resp != nil { + _ = res.resp.Body.Close() + } + t.Fatal("response did not start before the cached document was read to its end") + } + release.Do(func() { close(gate.release) }) + if res.err != nil { + t.Fatal(res.err) + } + _, _ = io.Copy(io.Discard, res.resp.Body) + _ = res.resp.Body.Close() + if n, _ := u.counts(condaRepodataPath); n != 1 { + t.Errorf("repodata requested %d times, want 1", n) + } +} diff --git a/internal/handler/cooldown_times.go b/internal/handler/cooldown_times.go new file mode 100644 index 00000000..88532cd1 --- /dev/null +++ b/internal/handler/cooldown_times.go @@ -0,0 +1,149 @@ +package handler + +import ( + "bytes" + "compress/gzip" + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" + "time" +) + +// publishTimes maps each version a times document lists to its publish +// time. A version whose time didn't parse maps to the zero time and is let +// through. +type publishTimes map[string]time.Time + +// 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. +type timedMetadata struct { + ecosystem string + // docKey and docURL are the cache key and upstream URL of the document + // clients read. They match the plain metadata path's, so both share one + // cached copy. + docKey, docURL string + // timesKey and timesURL are those of the document with publish times. + timesKey, timesURL string + timesAccept string + parseTimes func([]byte) (publishTimes, error) +} + +// timedDocuments is what fetchTimedMetadata hands back. published is read +// from times and is nil when times.err is set. +type timedDocuments struct { + doc metadataResult + times metadataResult + published publishTimes + // withholdUnknown is set when doc lists versions the times document + // doesn't, and that document is a stale copy: those versions are newer + // than the copy, so they are left out rather than let through. + withholdUnknown bool +} + +// fetchTimedMetadata fetches both documents concurrently. The document +// clients read is fetched as the plain metadata path fetches it, with +// identity encoding and Accept */*. +// +// Each document has its own TTL, so a cached times document can be older +// than the document it is used to filter, and miss versions published in +// between. Those versions would be let through as having no publish time. +// listed returns the versions the document lists; when some are missing +// from a times document read from the cache, the times document is fetched +// again before filtering. +func (p *Proxy) fetchTimedMetadata(ctx context.Context, t timedMetadata, listed func(metadataResult) ([]string, error)) (timedDocuments, error) { + var res timedDocuments + var published publishTimes + validate := func(body []byte) error { + parsed, err := t.parseTimes(body) + if err == nil { + published = parsed + } + return err + } + + done := make(chan struct{}) + go func() { + defer close(done) + res.times = p.lookupMetadata(ctx, t.ecosystem, t.timesKey, t.timesURL, t.timesAccept, "", validate) + }() + res.doc = p.lookupMetadata(ctx, t.ecosystem, t.docKey, t.docURL, "*/*", "identity", nil) + <-done + + if res.doc.err != nil || res.times.err != nil { + return res, nil + } + versions, err := listed(res.doc) + if err != nil { + return res, err + } + unknown := func() bool { + for _, v := range versions { + if _, ok := published[v]; !ok { + return true + } + } + return false + } + if unknown() && res.times.fromCache { + p.Logger.Info("cooldown: publish times older than metadata, fetching them again", + "ecosystem", t.ecosystem, "key", t.timesKey) + res.times = p.refreshMetadata(ctx, t.ecosystem, t.timesKey, t.timesURL, t.timesAccept, "", validate) + if res.times.err != nil { + return res, nil + } + } + res.published = published + res.withholdUnknown = res.times.stale && unknown() + return res, nil +} + +// rewriteDigest identifies the inputs of a cooldown rewrite for the rewrite +// cache: both documents and whether unknown versions are withheld. +func (d timedDocuments) rewriteDigest() string { + h := sha256.New() + for _, res := range []metadataResult{d.doc, d.times} { + digest := res.digest + if digest == "" { + sum := sha256.Sum256(res.body) + digest = "sha256:" + hex.EncodeToString(sum[:]) + } + _, _ = fmt.Fprintf(h, "%d:%s", len(digest), digest) + } + _, _ = fmt.Fprintf(h, "%t", d.withholdUnknown) + return "sha256:" + hex.EncodeToString(h.Sum(nil)) +} + +// withheld reports whether cooldown leaves version out, given the +// package's publish times. +func (d timedDocuments) withheld(p *Proxy, ecosystem, packagePURL, version string) bool { + published, ok := d.published[version] + if !ok { + return d.withholdUnknown + } + if published.IsZero() { + return false + } + return !p.Cooldown.IsAllowed(ecosystem, packagePURL, published) +} + +// decodedMetadataBody returns res's body without its Content-Encoding. The +// documents are fetched with identity encoding, so an upstream that sends +// gzip anyway is the only case to handle. +func (p *Proxy) decodedMetadataBody(res metadataResult) ([]byte, error) { + switch strings.ToLower(strings.TrimSpace(res.contentEncoding)) { + case "", "identity": + return res.body, nil + case "gzip", "x-gzip": + zr, err := gzip.NewReader(bytes.NewReader(res.body)) + if err != nil { + return nil, err + } + defer func() { _ = zr.Close() }() + return p.ReadMetadata(zr) + default: + return nil, fmt.Errorf("unsupported content encoding %q", res.contentEncoding) + } +} diff --git a/internal/handler/gem.go b/internal/handler/gem.go index dbb6e08d..edae77d3 100644 --- a/internal/handler/gem.go +++ b/internal/handler/gem.go @@ -6,8 +6,6 @@ import ( "fmt" "net/http" "regexp" - "slices" - "strconv" "strings" "time" @@ -144,6 +142,12 @@ func gemVersionAndPlatform(s string) bool { // handleCompactIndex serves the compact index for a gem, filtering versions // based on cooldown when enabled. +// +// With cooldown the index and the versions API's publish times are read +// through the metadata cache, so a bundle install within metadata_ttl of the +// last one asks rubygems.org nothing, and one after it revalidates both with +// their ETags. The filtered index is kept in the rewrite cache and served +// with an ETag of its own. func (h *GemHandler) handleCompactIndex(w http.ResponseWriter, r *http.Request) { name := r.PathValue("name") if name == "" { @@ -157,122 +161,118 @@ func (h *GemHandler) handleCompactIndex(w http.ResponseWriter, r *http.Request) h.proxy.Logger.Info("gem compact index request with cooldown", "name", name) - indexResp, filteredVersions, versionsErr, err := h.fetchIndexAndVersions(r, name) - if err != nil { - if writeUpstreamHeld(w, err) { - return - } - h.proxy.Logger.Error("upstream compact index request failed", "error", err) - http.Error(w, "upstream request failed", http.StatusBadGateway) + docs, err := h.proxy.fetchTimedMetadata(r.Context(), h.timedIndex(name), h.indexVersions) + switch { + case docs.doc.err != nil: + h.proxy.writeMetadataFetchError(w, docs.doc.err) return - } - defer func() { _ = indexResp.Body.Close() }() - - if indexResp.StatusCode != http.StatusOK { - h.proxy.relayResponse(w, r, indexResp, nil) + case docs.times.err != nil: + h.proxy.writeCooldownLookupError(w, "gem", name, docs.times.err) + return + case err != nil: + h.proxy.Logger.Error("failed to read gem compact index", "gem", name, "error", err) + http.Error(w, "failed to read upstream response", http.StatusBadGateway) return } - if versionsErr != nil { - h.proxy.writeCooldownLookupError(w, "gem", name, versionsErr) + packagePURL := canonicalPackagePURL("gem", name) + m, err := h.proxy.cachedRewrite(r.Context(), "gem", h.proxyURL, "info/"+name, docs.doc.body, docs.rewriteDigest(), + func([]byte) ([]byte, []string, error) { + body, err := h.proxy.decodedMetadataBody(docs.doc) + if err != nil { + return nil, nil, err + } + out, kept := h.filterIndex(body, name, func(version string) bool { + return docs.withheld(h.proxy, "gem", packagePURL, version) + }) + return out, kept, nil + }) + if err != nil { + h.proxy.Logger.Error("failed to filter gem compact index", "gem", name, "error", err) + http.Error(w, "failed to read upstream response", http.StatusBadGateway) return } - h.writeFilteredIndex(w, indexResp, name, filteredVersions) -} - -// fetchIndexAndVersions fetches the compact index and versions API -// concurrently. It returns the index response and the set of versions to -// filter, the versions API's error, and the index fetch's error. -func (h *GemHandler) fetchIndexAndVersions(r *http.Request, name string) (*http.Response, map[string]bool, error, error) { - type versionsResult struct { - filtered map[string]bool - err error + // Served as identity, as upstream's index was: Bundler checks digest + // headers against the bytes it receives, and asks for ranges of it. + m.gzipped = nil + contentType := docs.doc.contentType + if contentType == "" || contentType == contentTypeJSON { + contentType = "text/plain; charset=utf-8" } + markStale(w, docs.doc.stale || docs.times.stale) + serveRewrittenMetadataAs(w, r, m, contentType) +} - versionsCh := make(chan versionsResult, 1) - go func() { - filtered, err := h.fetchFilteredVersions(r, name) - versionsCh <- versionsResult{filtered: filtered, err: err} - }() - - indexResp, err := h.fetchCompactIndex(r, name) - - versionsRes := <-versionsCh - - if err != nil { - return nil, nil, nil, err +// timedIndex describes a gem's compact index and the versions API it takes +// publish times from. +func (h *GemHandler) timedIndex(name string) timedMetadata { + return timedMetadata{ + ecosystem: "gem", + docKey: "info/" + name, + docURL: h.upstreamURL + "/info/" + name, + timesKey: "api/v1/versions/" + name + ".json", + timesURL: fmt.Sprintf("%s/api/v1/versions/%s.json", h.upstreamURL, name), + timesAccept: contentTypeJSON, + parseTimes: parseGemVersions, } - return indexResp, versionsRes.filtered, versionsRes.err, nil } -// fetchCompactIndex fetches the compact index from upstream. -func (h *GemHandler) fetchCompactIndex(r *http.Request, name string) (*http.Response, error) { - indexURL := h.upstreamURL + "/info/" + name - req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, indexURL, nil) +// indexVersions lists the versions in a compact index, as the versions API +// names them. +func (h *GemHandler) indexVersions(doc metadataResult) ([]string, error) { + body, err := h.proxy.decodedMetadataBody(doc) if err != nil { return nil, err } - // Only Accept is passed on. The client's Accept-Encoding would turn off - // the transport's transparent decompression and leave a gzip body to - // filter, and its validators belong to the filtered copy it was served, - // not to upstream's file. - if v := r.Header.Get(headerAccept); v != "" { - req.Header.Set(headerAccept, v) + var versions []string + for rest := body; len(rest) > 0; { + line, tail, _ := bytes.Cut(rest, []byte("\n")) + rest = tail + if version := gemIndexLineVersion(line); gemIndexListsVersion(version) { + versions = append(versions, string(version)) + } } - return h.proxy.HTTPClient.Do(req) + return versions, nil } -// gemFilteredIndexDroppedHeaders describe upstream's file as a whole, so -// they are wrong for a copy with versions removed. Bundler checks the digest -// headers against the body it receives. -var gemFilteredIndexDroppedHeaders = []string{ - headerContentLength, headerContentEncoding, headerETag, headerLastModified, - "Digest", "Repr-Digest", "Content-Digest", "Accept-Ranges", "Content-Range", +// gemIndexLineVersion returns the version a compact index line starts with. +func gemIndexLineVersion(line []byte) []byte { + if spaceIdx := bytes.IndexByte(line, ' '); spaceIdx > 0 { + return line[:spaceIdx] + } + return line } -// writeFilteredIndex writes the compact index response with cooldown-filtered -// versions removed. The body is read in full first, so a read error or an -// oversized index answers 502 instead of a truncated 200. -func (h *GemHandler) writeFilteredIndex(w http.ResponseWriter, resp *http.Response, name string, filtered map[string]bool) { - body, err := h.proxy.ReadMetadata(resp.Body) - if err != nil { - h.proxy.Logger.Error("failed to read gem compact index", "gem", name, "error", err) - http.Error(w, "failed to read upstream response", http.StatusBadGateway) - return - } +// gemIndexListsVersion reports whether a compact index line's first field +// is a version, rather than the "---" that starts the file or an empty line. +func gemIndexListsVersion(field []byte) bool { + return len(field) > 0 && string(field) != "---" +} +// filterIndex returns the compact index without the lines of versions +// withheld reports, and the versions it kept. +func (h *GemHandler) filterIndex(body []byte, name string, withheld func(string) bool) ([]byte, []string) { + var kept []string var out bytes.Buffer out.Grow(len(body)) for rest := body; len(rest) > 0; { line, tail, _ := bytes.Cut(rest, []byte("\n")) rest = tail - version := line - if spaceIdx := bytes.IndexByte(line, ' '); spaceIdx > 0 { - version = line[:spaceIdx] - } - if filtered[string(version)] { - h.proxy.Logger.Info("cooldown: filtering gem version", - "gem", name, "version", string(version)) - continue + version := gemIndexLineVersion(line) + if gemIndexListsVersion(version) { + if withheld(string(version)) { + h.proxy.Logger.Info("cooldown: filtering gem version", + "gem", name, "version", string(version)) + continue + } + kept = append(kept, string(version)) } - out.Write(line) out.WriteByte('\n') } - - for k, vv := range resp.Header { - if slices.ContainsFunc(gemFilteredIndexDroppedHeaders, func(d string) bool { return strings.EqualFold(k, d) }) { - continue - } - for _, v := range vv { - w.Header().Add(k, v) - } - } - w.Header().Set(headerContentLength, strconv.Itoa(out.Len())) - w.WriteHeader(http.StatusOK) - _, _ = w.Write(out.Bytes()) + return out.Bytes(), kept } // gemVersion represents a version entry from the RubyGems versions API. @@ -282,54 +282,26 @@ type gemVersion struct { CreatedAt string `json:"created_at"` } -// fetchFilteredVersions fetches the versions API and returns a set of version -// strings that should be filtered out by cooldown. -func (h *GemHandler) fetchFilteredVersions(r *http.Request, name string) (map[string]bool, error) { - versionsURL := fmt.Sprintf("%s/api/v1/versions/%s.json", h.upstreamURL, name) - req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, versionsURL, nil) - if err != nil { - return nil, err - } - - 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("versions API returned %d", resp.StatusCode) - } - - body, err := h.proxy.ReadMetadata(resp.Body) - if err != nil { - return nil, err - } +// parseGemVersions reads the publish times from a versions API response, +// keyed by version as the compact index writes it. +func parseGemVersions(body []byte) (publishTimes, error) { var versions []gemVersion if err := json.Unmarshal(body, &versions); err != nil { return nil, err } - - packagePURL := canonicalPackagePURL("gem", name) - filtered := make(map[string]bool) - + times := make(publishTimes, len(versions)) for _, v := range versions { + versionStr := v.Number + if v.Platform != "" && v.Platform != "ruby" { + versionStr = v.Number + "-" + v.Platform + } createdAt, err := time.Parse(time.RFC3339, v.CreatedAt) if err != nil { - continue - } - - if !h.proxy.Cooldown.IsAllowed("gem", packagePURL, createdAt) { - // Build version string matching compact index format - versionStr := v.Number - if v.Platform != "" && v.Platform != "ruby" { - versionStr = v.Number + "-" + v.Platform - } - filtered[versionStr] = true + createdAt = time.Time{} } + times[versionStr] = createdAt } - - return filtered, nil + return times, nil } // proxyCached forwards a metadata request with caching. diff --git a/internal/handler/gem_test.go b/internal/handler/gem_test.go index a13d819a..ba62ef8b 100644 --- a/internal/handler/gem_test.go +++ b/internal/handler/gem_test.go @@ -196,10 +196,11 @@ func TestGemCompactIndexCooldownClientAcceptsGzip(t *testing.T) { if got := w.Body.String(); got != want { t.Errorf("body = %q, want %q", got, want) } - for _, hdr := range []string{"ETag", "Repr-Digest"} { - if v := w.Header().Get(hdr); v != "" { - t.Errorf("%s = %q, upstream's value must not describe the filtered index", hdr, v) - } + if v := w.Header().Get("Repr-Digest"); v != "" { + t.Errorf("Repr-Digest = %q, upstream's value must not describe the filtered index", v) + } + if v := w.Header().Get("ETag"); v == "" || v == `"upstream"` { + t.Errorf("ETag = %q, want one of the filtered index", v) } } @@ -351,7 +352,7 @@ func TestGemCompactIndexVersionsAPIFails(t *testing.T) { } } -func TestGemFetchFilteredVersions(t *testing.T) { +func TestGemParseVersions(t *testing.T) { now := time.Now() oldTime := now.Add(-7 * 24 * time.Hour).Format(time.RFC3339) recentTime := now.Add(-1 * time.Hour).Format(time.RFC3339) @@ -360,39 +361,22 @@ func TestGemFetchFilteredVersions(t *testing.T) { {Number: "1.0.0", Platform: "ruby", CreatedAt: oldTime}, {Number: "2.0.0", Platform: "ruby", CreatedAt: recentTime}, {Number: "2.0.0", Platform: "java", CreatedAt: recentTime}, + {Number: "3.0.0", Platform: "ruby", CreatedAt: "not a time"}, }) - upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write(versionsJSON) - })) - defer upstream.Close() - proxy := testProxy() - proxy.Cooldown = &cooldown.Config{ - Default: "3d", - } - - h := &GemHandler{ - proxy: proxy, - upstreamURL: upstream.URL, - proxyURL: "http://proxy.local", - } + proxy.Cooldown = &cooldown.Config{Default: "3d"} - req := httptest.NewRequest(http.MethodGet, "/info/testgem", nil) - filtered, err := h.fetchFilteredVersions(req, "testgem") + times, err := parseGemVersions(versionsJSON) if err != nil { t.Fatal(err) } - - if filtered["1.0.0"] { - t.Error("version 1.0.0 should not be filtered (old enough)") - } - if !filtered["2.0.0"] { - t.Error("version 2.0.0 (ruby) should be filtered") - } - if !filtered["2.0.0-java"] { - t.Error("version 2.0.0-java should be filtered") + docs := timedDocuments{published: times} + purl := canonicalPackagePURL("gem", "testgem") + for version, want := range map[string]bool{"1.0.0": false, "2.0.0": true, "2.0.0-java": true, "3.0.0": false, "4.0.0": false} { + if got := docs.withheld(proxy, "gem", purl, version); got != want { + t.Errorf("withheld(%s) = %v, want %v", version, got, want) + } } _ = fmt.Sprintf // silence unused import diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 2e0d688c..9c78ad4c 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -1479,7 +1479,7 @@ func (p *Proxy) cachedMetadataState(ctx context.Context, ecosystem, cacheKey str 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 { - hit := metadataResult{body: data, contentType: ct, contentEncoding: entry.ContentEncoding.String}.describedBy(entry) + hit := metadataResult{body: data, contentType: ct, contentEncoding: entry.ContentEncoding.String, fromCache: true}.describedBy(entry) return entry, &hit } if validate != nil { @@ -1490,6 +1490,31 @@ func (p *Proxy) cachedMetadataState(ctx context.Context, ecosystem, cacheKey str return entry, nil } +// refreshMetadata is lookupMetadata for a document whose cached copy is known +// to be out of date although it is within its TTL. It skips the TTL and asks +// upstream, revalidating the cached copy with its validators, and falls back +// to that copy if upstream fails, which the result reports as stale. +// Concurrent refreshes of one document share a fetch. +func (p *Proxy) refreshMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validate func([]byte) error) metadataResult { + if containsPathTraversal(cacheKey) { + return metadataResult{err: fmt.Errorf("invalid cache key: %q", cacheKey)} + } + key := metadataCoalesceKey(ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate != nil) + "\x00refresh" + res, shared := p.coalesceMetadata(ctx, key, func(fetchCtx context.Context) metadataResult { + var entry *database.MetadataCacheEntry + if p.CacheMetadata && p.DB != nil { + entry, _ = p.DB.GetMetadataCache(ecosystem, cacheKey) + } + return p.fetchMetadataFromUpstream(fetchCtx, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate, entry) + }) + if shared && res.err == nil && validate != nil { + if err := validate(res.body); err != nil { + return metadataResult{err: err} + } + } + return res +} + // 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. @@ -1558,6 +1583,9 @@ type metadataResult struct { // stale is set when the body is the cached copy served because upstream // failed. stale bool + // fromCache is set when the body is a stored copy within its TTL, read + // without asking upstream. + fromCache bool } // describedBy returns res with the digest that entry, the cache row its body diff --git a/internal/handler/hex.go b/internal/handler/hex.go index 2d86686c..a46a7e4c 100644 --- a/internal/handler/hex.go +++ b/internal/handler/hex.go @@ -101,6 +101,11 @@ const hexAPIURL = "https://hex.pm" // handlePackages proxies the /packages/{name} endpoint, applying cooldown filtering // when enabled. Since the protobuf format has no timestamps, we fetch them from the // Hex HTTP API concurrently. +// +// Both are read through the metadata cache, so within metadata_ttl a package +// costs no request to repo.hex.pm or to the rate limited Hex API, and after +// it both are revalidated with their ETags. The filtered package is kept in +// the rewrite cache and served with an ETag of its own. func (h *HexHandler) handlePackages(w http.ResponseWriter, r *http.Request) { name := r.PathValue("name") if name == "" || !h.proxy.cooldownApplies("hex", name) { @@ -110,87 +115,102 @@ func (h *HexHandler) handlePackages(w http.ResponseWriter, r *http.Request) { h.proxy.Logger.Info("hex package request with cooldown", "name", name) - protoResp, filteredVersions, versionsErr, err := h.fetchPackageAndVersions(r, name) - if err != nil { - if writeUpstreamHeld(w, err) { - return - } - h.proxy.Logger.Error("upstream request failed", "error", err) - http.Error(w, "upstream request failed", http.StatusBadGateway) + docs, err := h.proxy.fetchTimedMetadata(r.Context(), h.timedPackage(name), h.packageVersions) + switch { + case docs.doc.err != nil: + h.proxy.writeMetadataFetchError(w, docs.doc.err) return - } - defer func() { _ = protoResp.Body.Close() }() - - if protoResp.StatusCode != http.StatusOK { - h.proxy.relayResponse(w, r, protoResp, nil) + case docs.times.err != nil: + h.proxy.writeCooldownLookupError(w, "hex", name, docs.times.err) return - } - - if versionsErr != nil { - h.proxy.writeCooldownLookupError(w, "hex", name, versionsErr) - return - } - - body, err := h.proxy.ReadMetadata(protoResp.Body) - if err != nil { + case err != nil: h.proxy.Logger.Error("failed to read hex package", "name", name, "error", err) - http.Error(w, "failed to read response", http.StatusBadGateway) - return - } - - if len(filteredVersions) == 0 { - // No versions in cooldown, pass through - w.Header().Set(headerContentType, protoResp.Header.Get(headerContentType)) - w.Header().Set("Content-Encoding", "gzip") - _, _ = w.Write(body) + http.Error(w, "failed to filter package metadata", http.StatusBadGateway) return } - filtered, err := h.filterSignedPackage(body, filteredVersions) + packagePURL := canonicalPackagePURL("hex", name) + m, err := h.proxy.cachedRewrite(r.Context(), "hex", h.proxyURL, "packages/"+name, docs.doc.body, docs.rewriteDigest(), + func([]byte) ([]byte, []string, error) { + body, err := h.proxy.decodedMetadataBody(docs.doc) + if err != nil { + return nil, nil, err + } + versions, err := h.packageVersions(docs.doc) + if err != nil { + return nil, nil, err + } + filtered := make(map[string]bool) + var kept []string + for _, version := range versions { + if docs.withheld(h.proxy, "hex", packagePURL, version) { + filtered[version] = true + h.proxy.Logger.Info("cooldown: filtering hex version", "package", name, "version", version) + continue + } + kept = append(kept, version) + } + if len(filtered) == 0 { + // Nothing in cooldown: upstream's bytes, signature included. + return body, kept, nil + } + out, err := h.filterSignedPackage(body, filtered) + return out, kept, err + }) if err != nil { h.proxy.Logger.Error("failed to filter hex package", "name", name, "error", err) http.Error(w, "failed to filter package metadata", http.StatusBadGateway) return } - w.Header().Set(headerContentType, "application/octet-stream") - w.Header().Set("Content-Encoding", "gzip") - _, _ = w.Write(filtered) + markStale(w, docs.doc.stale || docs.times.stale) + serveHexPackage(w, r, m) } -// fetchPackageAndVersions fetches the protobuf package and version timestamps -// concurrently. It returns the package response and the set of versions to -// filter, the Hex API's error, and the package fetch's error. -func (h *HexHandler) fetchPackageAndVersions(r *http.Request, name string) (*http.Response, map[string]bool, error, error) { - type versionsResult struct { - filtered map[string]bool - err error +// serveHexPackage writes a package with its validators, or 304 when they +// show the client's copy is current. The registry file is gzip itself, and +// Hex clients gunzip it however it is labelled, so it is never compressed +// again. +func serveHexPackage(w http.ResponseWriter, r *http.Request, m rewrittenMetadata) { + w.Header().Set(headerETag, m.etag) + if !m.lastModified.IsZero() { + w.Header().Set(headerLastModified, m.lastModified.UTC().Format(http.TimeFormat)) } + if notModified(r, m.etag, m.lastModified) { + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set(headerContentType, "application/octet-stream") + w.Header().Set(headerContentEncoding, "gzip") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(m.body) +} - versionsCh := make(chan versionsResult, 1) - go func() { - filtered, err := h.fetchFilteredVersions(r, name) - versionsCh <- versionsResult{filtered: filtered, err: err} - }() - - protoResp, err := h.fetchUpstreamPackage(r, name) - - versionsRes := <-versionsCh - - if err != nil { - return nil, nil, nil, err +// timedPackage describes a package's registry file and the Hex API entry +// it takes publish times from. +func (h *HexHandler) timedPackage(name string) timedMetadata { + return timedMetadata{ + ecosystem: "hex", + docKey: "packages/" + name, + docURL: h.upstreamURL + "/packages/" + name, + timesKey: hashedCacheKey(h.apiURL, "api/packages/"+name), + timesURL: fmt.Sprintf("%s/api/packages/%s", h.apiURL, name), + timesAccept: contentTypeJSON, + parseTimes: parseHexReleases, } - return protoResp, versionsRes.filtered, versionsRes.err, nil } -// fetchUpstreamPackage fetches the protobuf package from upstream. -func (h *HexHandler) fetchUpstreamPackage(r *http.Request, name string) (*http.Response, error) { - upstreamURL := h.upstreamURL + "/packages/" + name - req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, upstreamURL, nil) +// packageVersions lists the releases in a gzipped Signed package. +func (h *HexHandler) packageVersions(doc metadataResult) ([]string, error) { + body, err := h.proxy.decodedMetadataBody(doc) if err != nil { return nil, err } - return h.proxy.HTTPClient.Do(req) + payload, err := h.signedPayload(body) + if err != nil { + return nil, err + } + return packageReleaseVersions(payload) } // hexRelease represents a version entry from the Hex API. @@ -204,60 +224,26 @@ type hexPackageAPI struct { Releases []hexRelease `json:"releases"` } -// fetchFilteredVersions fetches the Hex API and returns a set of version -// strings that should be filtered out by cooldown. -func (h *HexHandler) fetchFilteredVersions(r *http.Request, name string) (map[string]bool, error) { - apiURL := fmt.Sprintf("%s/api/packages/%s", h.apiURL, name) - req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, apiURL, nil) - if err != nil { - return nil, err - } - req.Header.Set("Accept", "application/json") - - 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("hex API returned %d", resp.StatusCode) - } - - body, err := h.proxy.ReadMetadata(resp.Body) - if err != nil { - return nil, err - } +// parseHexReleases reads the publish times from a Hex API package response. +func parseHexReleases(body []byte) (publishTimes, error) { var pkg hexPackageAPI if err := json.Unmarshal(body, &pkg); err != nil { return nil, err } - - packagePURL := canonicalPackagePURL("hex", name) - filtered := make(map[string]bool) - + times := make(publishTimes, len(pkg.Releases)) for _, release := range pkg.Releases { insertedAt, err := time.Parse(time.RFC3339Nano, release.InsertedAt) if err != nil { - continue - } - - if !h.proxy.Cooldown.IsAllowed("hex", packagePURL, insertedAt) { - filtered[release.Version] = true - h.proxy.Logger.Info("cooldown: filtering hex version", - "package", name, "version", release.Version, - "published", release.InsertedAt) + insertedAt = time.Time{} } + times[release.Version] = insertedAt } - - return filtered, nil + return times, nil } -// filterSignedPackage decompresses gzipped data, decodes the Signed protobuf wrapper, -// filters releases from the Package payload, and re-encodes as gzipped protobuf -// (without the original signature since the payload has changed). -func (h *HexHandler) filterSignedPackage(gzippedData []byte, filteredVersions map[string]bool) ([]byte, error) { - // Decompress gzip, with the metadata limit applied to the decompressed size +// signedPayload decompresses a gzipped Signed message, with the metadata +// limit applied to the decompressed size, and returns its payload. +func (h *HexHandler) signedPayload(gzippedData []byte) ([]byte, error) { gr, err := gzip.NewReader(bytes.NewReader(gzippedData)) if err != nil { return nil, err @@ -273,9 +259,30 @@ func (h *HexHandler) filterSignedPackage(gzippedData []byte, filteredVersions ma if err != nil { return nil, fmt.Errorf("extracting payload: %w", err) } + return payload, nil +} + +// packageReleaseVersions lists the versions of the releases in a Package +// message. +func packageReleaseVersions(payload []byte) ([]string, error) { + var versions []string + _, err := filterPackageReleases(payload, nil, func(version string) { + versions = append(versions, version) + }) + return versions, err +} + +// filterSignedPackage decompresses gzipped data, decodes the Signed protobuf wrapper, +// filters releases from the Package payload, and re-encodes as gzipped protobuf +// (without the original signature since the payload has changed). +func (h *HexHandler) filterSignedPackage(gzippedData []byte, filteredVersions map[string]bool) ([]byte, error) { + payload, err := h.signedPayload(gzippedData) + if err != nil { + return nil, err + } // Filter releases from the Package message - filteredPayload, err := filterPackageReleases(payload, filteredVersions) + filteredPayload, err := filterPackageReleases(payload, filteredVersions, nil) if err != nil { return nil, fmt.Errorf("filtering releases: %w", err) } @@ -298,9 +305,10 @@ func (h *HexHandler) filterSignedPackage(gzippedData []byte, filteredVersions ma return buf.Bytes(), nil } -// filterPackageReleases filters releases from a Package protobuf message. +// filterPackageReleases filters releases from a Package protobuf message, +// calling seen, when not nil, with the version of each release. // Package: field 1 = releases (repeated), field 2 = name, field 3 = repository -func filterPackageReleases(payload []byte, filteredVersions map[string]bool) ([]byte, error) { +func filterPackageReleases(payload []byte, filteredVersions map[string]bool, seen func(string)) ([]byte, error) { var result []byte data := payload @@ -325,6 +333,9 @@ func filterPackageReleases(payload []byte, filteredVersions map[string]bool) ([] if num == 1 { // releases field version := extractReleaseVersion(v) + if seen != nil { + seen(version) + } if filteredVersions[version] { continue // skip this release } diff --git a/internal/handler/hex_test.go b/internal/handler/hex_test.go index d448b760..a0b0448a 100644 --- a/internal/handler/hex_test.go +++ b/internal/handler/hex_test.go @@ -95,7 +95,7 @@ func buildHexSigned(payload []byte) []byte { func TestHexFilterPackageReleases(t *testing.T) { pkg := buildHexPackage("phoenix", []string{testVersion100, "2.0.0", "3.0.0"}) - filtered, err := filterPackageReleases(pkg, map[string]bool{"2.0.0": true}) + filtered, err := filterPackageReleases(pkg, map[string]bool{"2.0.0": true}, nil) if err != nil { t.Fatal(err) } diff --git a/internal/handler/metadata_stream.go b/internal/handler/metadata_stream.go index 8bb319d9..fd4fb83e 100644 --- a/internal/handler/metadata_stream.go +++ b/internal/handler/metadata_stream.go @@ -1,12 +1,15 @@ package handler import ( + "context" "errors" + "io" "net/http" "strconv" "strings" "time" + "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/metrics" ) @@ -21,65 +24,92 @@ import ( // buffered path instead: no fresh row, a row without a digest or size, a // document over the metadata size limit, or one that can't be opened. func (p *Proxy) serveCachedMetadata(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string) bool { - if p.DB == nil || p.Storage == nil || p.MetadataTTL <= 0 || containsPathTraversal(cacheKey) { + entry := p.freshMetadataEntry(ecosystem, cacheKey) + if entry == nil { + return false + } + body, err := p.openMetadataEntry(r.Context(), ecosystem, cacheKey, entry) + if err != nil { return false } + defer func() { _ = body.Close() }() + + p.recordStreamedMetadataHit(ecosystem, entry) + + cm := p.cachedMetaFor(entry) + if writeMetadataValidators(w, r, cm) { + return true + } + contentType := contentTypeJSON + if entry.ContentType.Valid { + contentType = entry.ContentType.String + } + w.Header().Set(headerContentType, contentType) + w.Header().Set(headerContentLength, strconv.FormatInt(entry.Size.Int64, 10)) + if entry.ContentEncoding.String != "" { + w.Header().Set(headerContentEncoding, entry.ContentEncoding.String) + } + markStale(w, cm.stale) + w.WriteHeader(http.StatusOK) + if r.Method != http.MethodHead { + copyArtifactBody(w, body, entry.Size.Int64, false) + } + return true +} + +// freshMetadataEntry returns the metadata cache row for ecosystem and +// cacheKey when its document can be streamed: the row is within its TTL and +// records a sha256 digest and a size within the metadata size limit. It +// returns nil otherwise. +func (p *Proxy) freshMetadataEntry(ecosystem, cacheKey string) *database.MetadataCacheEntry { + if p.DB == nil || p.Storage == nil || p.MetadataTTL <= 0 || containsPathTraversal(cacheKey) { + return nil + } entry, err := p.DB.GetMetadataCache(ecosystem, cacheKey) if err != nil || entry == nil || !entry.FetchedAt.Valid || time.Since(entry.FetchedAt.Time) >= p.MetadataTTL { - return false + return nil } limit := p.MetadataMaxSize if limit <= 0 { limit = defaultMetadataMaxSize } - digest, ok := strings.CutPrefix(entry.ContentDigest.String, "sha256:") - if !entry.ContentDigest.Valid || !ok || !entry.Size.Valid || entry.Size.Int64 > limit { - return false + if !entry.ContentDigest.Valid || !strings.HasPrefix(entry.ContentDigest.String, "sha256:") || !entry.Size.Valid || entry.Size.Int64 > limit { + return nil } - checks, err := newIntegrityChecks(digest, "") + return entry +} + +// openMetadataEntry opens the stored document of a row freshMetadataEntry +// returned. Its bytes are checked against the row's digest as they are read: +// a mismatch fails the read at the end and drops the row, so the next +// request fetches the document again. +func (p *Proxy) openMetadataEntry(ctx context.Context, ecosystem, cacheKey string, entry *database.MetadataCacheEntry) (io.ReadCloser, error) { + checks, err := newIntegrityChecks(strings.TrimPrefix(entry.ContentDigest.String, "sha256:"), "") if err != nil { - return false + return nil, err } - stored, err := p.Storage.Open(r.Context(), entry.StoragePath) + stored, err := p.Storage.Open(ctx, entry.StoragePath) if err != nil { - return false + return nil, err } body, err := checks.wrapFailOnMismatch(stored, func(reason string) { p.Logger.Error("cached metadata failed integrity check", "ecosystem", ecosystem, "key", cacheKey, "path", entry.StoragePath, "reason", reason) metrics.RecordIntegrityFailure(ecosystem, metrics.FailureDetail{ Path: entry.StoragePath, Err: errors.New(reason)}) - // Drop the row so the next request fetches the document again. if err := p.DB.DiscardMetadataCache(ecosystem, cacheKey, entry.StoragePath, entry.ContentDigest.String); err != nil { p.Logger.Warn("failed to clear corrupt metadata from cache", "error", err) } }) if err != nil { _ = stored.Close() - return false + return nil, err } - defer func() { _ = body.Close() }() + return body, nil +} +// recordStreamedMetadataHit counts a hit on a row served by streaming. +func (p *Proxy) recordStreamedMetadataHit(ecosystem string, entry *database.MetadataCacheEntry) { p.recordMetadataRead(entry) metrics.RecordMetadataCacheHit(ecosystem, int(entry.Size.Int64)) - - cm := p.cachedMetaFor(entry) - if writeMetadataValidators(w, r, cm) { - return true - } - contentType := contentTypeJSON - if entry.ContentType.Valid { - contentType = entry.ContentType.String - } - w.Header().Set(headerContentType, contentType) - w.Header().Set(headerContentLength, strconv.FormatInt(entry.Size.Int64, 10)) - if entry.ContentEncoding.String != "" { - w.Header().Set(headerContentEncoding, entry.ContentEncoding.String) - } - markStale(w, cm.stale) - w.WriteHeader(http.StatusOK) - if r.Method != http.MethodHead { - copyArtifactBody(w, body, entry.Size.Int64, false) - } - return true } diff --git a/internal/handler/relay_test.go b/internal/handler/relay_test.go index a7a8fb79..72d20f38 100644 --- a/internal/handler/relay_test.go +++ b/internal/handler/relay_test.go @@ -52,8 +52,8 @@ func TestRelayRoutes(t *testing.T) { "/upstream", "/file", "/metadata", "/nuget/query", "/conan/v2/conans/demo/1.0/user/stable/revisions/rev/files/other.txt", "/composer/search.json", "/pypi/simple/", "/gem/api/v1/dependencies", - "/gem/info/demo", "/conda/conda-forge/noarch/repodata.json", - "/hex/packages/demo", "/swift/scope/demo/1.0.0/Package.swift", + "/conda/conda-forge/noarch/repodata.json", + "/swift/scope/demo/1.0.0/Package.swift", "/v2/library/demo/manifests/latest", "/v2/library/demo/tags/list", "/npm/-/npm/v1/keys", } { @@ -82,10 +82,9 @@ func testRelayRoute(t *testing.T, route string, truncated bool) { if !truncated { _, _ = fmt.Fprint(rw, "0\r\nX-Checksum: verified\r\nX-Late: discovered-at-eof\r\nX-Private: still-secret\r\n\r\n") } - // Two of these routes fan out a second upstream request for cooldown - // timestamps and abandon its body on a non-200, so this may be writing to - // a connection the proxy has already dropped. What the proxy made of the - // response under test is asserted downstream. + // The proxy may drop the connection before reading the whole + // response. What it made of the response under test is asserted + // downstream. _ = rw.Flush() })) defer upstream.Close() diff --git a/internal/handler/upstream_test.go b/internal/handler/upstream_test.go index 7839eaa3..9d99214f 100644 --- a/internal/handler/upstream_test.go +++ b/internal/handler/upstream_test.go @@ -1,6 +1,7 @@ package handler import ( + "context" "io" "net/http" "net/http/httptest" @@ -92,9 +93,10 @@ func TestHexHandlerUsesConfiguredAPIUpstream(t *testing.T) { upstream.URL+"/hex", upstream.URL+"/hex-api", ) - _, err := h.fetchFilteredVersions(httptest.NewRequest(http.MethodGet, "/", nil), "demo") - if err != nil { - t.Fatalf("fetchFilteredVersions failed: %v", err) + timed := h.timedPackage("demo") + res := h.proxy.lookupMetadata(context.Background(), "hex", timed.timesKey, timed.timesURL, timed.timesAccept, "", nil) + if res.err != nil { + t.Fatalf("fetching publish times failed: %v", res.err) } if requestedPath != "/hex-api/api/packages/demo" { t.Errorf("API path = %q, want %q", requestedPath, "/hex-api/api/packages/demo")