From 92b1fb20a6d451693d4d5347092f505a9a398d10 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 27 Oct 2026 16:31:47 +0100 Subject: [PATCH 1/4] Follow npm and Packagist change feeds to revalidate changed metadata Within metadata_ttl cached npm and Composer metadata is served without asking upstream, so a new release can take that long to show up. Both registries publish a feed of the packages that changed. upstream.npm_change_feed and upstream.composer_change_feed (off by default, require cache_metadata) start a reader per feed that runs every minute and clears fetched_at on the cached rows of listed packages that were fetched before the read. The rows keep their document and ETag, so the next request revalidates them upstream. Routed packages are cached under other keys and never match. A Packagist resync expires all cached Composer metadata. Failed reads back off up to 15 minutes. The feed position is kept in memory, so after a restart the reader starts from the feed's current position. --- README.md | 7 +- config.example.yaml | 9 + docs/configuration.md | 22 ++ internal/changefeed/changefeed.go | 261 +++++++++++++++++ internal/changefeed/changefeed_test.go | 198 +++++++++++++ internal/config/config.go | 20 ++ internal/config/config_test.go | 35 +++ internal/database/metadata_cache_test.go | 90 ++++++ internal/database/queries.go | 46 +++ internal/metrics/metrics.go | 16 ++ internal/server/analytics_coverage_test.go | 40 +-- internal/server/analytics_runtime.go | 2 + internal/server/analytics_runtime_test.go | 18 ++ internal/server/change_feeds.go | 94 ++++++ internal/server/change_feeds_test.go | 272 ++++++++++++++++++ internal/server/server.go | 1 + internal/server/server_test.go | 7 - .../templates/components/runtime_metrics.html | 1 + 18 files changed, 1110 insertions(+), 29 deletions(-) create mode 100644 internal/changefeed/changefeed.go create mode 100644 internal/changefeed/changefeed_test.go create mode 100644 internal/server/change_feeds.go create mode 100644 internal/server/change_feeds_test.go diff --git a/README.md b/README.md index 20c6d5db..4f0c033c 100644 --- a/README.md +++ b/README.md @@ -188,7 +188,7 @@ therefore come from upstream's database, not from the proxy's own vulnerability data, and versions withheld by [cooldown](#version-cooldown) are not excluded from the report. -To serve private packages through the same registry URL, see [Private package routes](docs/configuration.md#private-package-routes). +To serve private packages through the same registry URL, see [Private package routes](docs/configuration.md#private-package-routes). To see new releases before `metadata_ttl` runs out, see [Change feeds](docs/configuration.md#change-feeds). ### Cargo @@ -387,7 +387,7 @@ Or set globally: composer config -g repositories.proxy composer http://localhost:8080/composer ``` -To serve private packages through the same URL, see [Private package routes](docs/configuration.md#private-package-routes). +To serve private packages through the same URL, see [Private package routes](docs/configuration.md#private-package-routes). To see new releases before `metadata_ttl` runs out, see [Change feeds](docs/configuration.md#change-feeds). ### Conan (C/C++) @@ -1521,6 +1521,7 @@ The proxy exposes Prometheus metrics at `GET /metrics`. All metric names are pre | `proxy_metadata_cache_entries` | gauge | `ecosystem` | Number of cached metadata documents per ecosystem. | | `proxy_artifacts_evicted_total` | counter | `reason`, `ecosystem` | Artifacts evicted by the size limit (`lru`) or by age (`retention`) | | `proxy_metadata_evicted_total` | counter | `ecosystem` | Cached metadata documents evicted by the size limit | +| `proxy_metadata_expired_by_change_feed_total` | counter | `ecosystem` | Cached metadata documents set to revalidate because an upstream [change feed](docs/configuration.md#change-feeds) listed them | | `proxy_fetch_lease_waits_total` | counter | `kind` | Cache misses (`artifact` or `metadata`) that waited for another instance sharing the database to fetch them | | `proxy_upstream_fetch_duration_seconds` | histogram | `ecosystem` | Time spent fetching from upstream | | `proxy_upstream_errors_total` | counter | `ecosystem`, `error_type` | Upstream fetch failures | @@ -1591,7 +1592,7 @@ The `kind` label on `proxy_cache_hits_total` and `proxy_cache_misses_total` keep `ecosystem` means three slightly different things across `/metrics`, and queries that join across them need to know which. -**From the package record, normalized.** The six `proxy_ecosystem_*` gauges, `proxy_cache_hits_total`, `proxy_cache_misses_total`, the three `proxy_metadata_cache_*` metrics, `proxy_artifacts_evicted_total`, `proxy_metadata_evicted_total`, `proxy_integrity_failures_total` and the scan metrics. Aliases collapse here: `gem` reads as `rubygems`, `composer` as `packagist`, `go` as `golang`. +**From the package record, normalized.** The six `proxy_ecosystem_*` gauges, `proxy_cache_hits_total`, `proxy_cache_misses_total`, the three `proxy_metadata_cache_*` metrics, `proxy_artifacts_evicted_total`, `proxy_metadata_evicted_total`, `proxy_metadata_expired_by_change_feed_total`, `proxy_integrity_failures_total` and the scan metrics. Aliases collapse here: `gem` reads as `rubygems`, `composer` as `packagist`, `go` as `golang`. **From the request path.** `proxy_requests_total`, `proxy_request_duration_seconds` and `proxy_response_bytes_total`. The names mostly coincide with the normalized ones -- these also report `rubygems`, `packagist` and `golang` -- but the Debian route reports `debian` where the package record says `deb`, and any path that is not a package endpoint reports `other`, which corresponds to no ecosystem at all. diff --git a/config.example.yaml b/config.example.yaml index d3b6da49..8ad37595 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -189,6 +189,10 @@ upstream: # npm_routes: # "@example/*": "https://npm.example.com/registry" + # Follow the npm registry's change feed, so a new release is fetched on the + # next request instead of after metadata_ttl. Requires cache_metadata. + # npm_change_feed: true + # Cargo sparse index URL cargo: "https://index.crates.io" @@ -245,6 +249,11 @@ upstream: # composer_routes: # "example/*": "https://composer.example.com/packages" + # Follow the change feed of the Packagist at composer, so a new release is + # fetched on the next request instead of after metadata_ttl. Requires + # cache_metadata. + # composer_change_feed: true + # Conan registry URL conan: "https://center.conan.io" diff --git a/docs/configuration.md b/docs/configuration.md index f10e4b23..e18263fe 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -287,6 +287,7 @@ Each upstream used by a built-in package route can be set in YAML or JSON under | `upstream.allow_loopback` | `PROXY_UPSTREAM_ALLOW_LOOPBACK` | `false` | | `upstream.npm` | `PROXY_UPSTREAM_NPM` | `https://registry.npmjs.org` | | `upstream.npm_full_metadata` | `PROXY_UPSTREAM_NPM_FULL_METADATA` | `false` | +| `upstream.npm_change_feed` | `PROXY_UPSTREAM_NPM_CHANGE_FEED` | `false` (see [change feeds](#change-feeds)) | | `upstream.cargo` | `PROXY_UPSTREAM_CARGO` | `https://index.crates.io` | | `upstream.cargo_download` | `PROXY_UPSTREAM_CARGO_DOWNLOAD` | `https://static.crates.io/crates` | | `upstream.gem` | `PROXY_UPSTREAM_GEM` | `https://rubygems.org` | @@ -303,6 +304,7 @@ Each upstream used by a built-in package route can be set in YAML or JSON under | `upstream.nuget_search` | `PROXY_UPSTREAM_NUGET_SEARCH` | `https://azuresearch-usnc.nuget.org` | | `upstream.composer` | `PROXY_UPSTREAM_COMPOSER` | `https://packagist.org` | | `upstream.composer_repository` | `PROXY_UPSTREAM_COMPOSER_REPOSITORY` | `https://repo.packagist.org` | +| `upstream.composer_change_feed` | `PROXY_UPSTREAM_COMPOSER_CHANGE_FEED` | `false` (see [change feeds](#change-feeds)) | | `upstream.conan` | `PROXY_UPSTREAM_CONAN` | `https://center.conan.io` | | `upstream.chef` | `PROXY_UPSTREAM_CHEF` | `https://supermarket.chef.io` | | `upstream.conda` | `PROXY_UPSTREAM_CONDA` | `https://conda.anaconda.org` | @@ -1346,6 +1348,26 @@ Set to `"0"` to always revalidate with upstream (ETag-based conditional requests When upstream is unreachable and the cached entry is past its TTL, the proxy serves the stale cached copy with a `Warning: 110 - "Response is Stale"` header so clients can tell the data may be outdated. +### Change feeds + +Within `metadata_ttl` a new release isn't seen. For npm and Composer the proxy can instead follow the registry's change feed, which lists the packages that changed, and revalidate just those: + +```yaml +cache_metadata: true +upstream: + npm_change_feed: true + composer_change_feed: true +``` + +Or via environment variables: `PROXY_UPSTREAM_NPM_CHANGE_FEED=true`, `PROXY_UPSTREAM_COMPOSER_CHANGE_FEED=true`. Both require `cache_metadata`, and do nothing with a `metadata_ttl` of `"0"`. + +Every minute the proxy reads what changed since its last read. Cached metadata of a listed package that was fetched before the read is marked as needing revalidation, and the next request for it asks the upstream with its `ETag` (a `304` when nothing changed). Artifacts are not touched. Listed packages the proxy hasn't cached cost nothing beyond the read. The feed is read in full, so the upstream doesn't learn which packages the proxy has cached. Packages in `upstream.npm_routes` or `upstream.composer_routes` are cached under other keys and never matched. + +- Composer reads `{upstream.composer}/metadata/changes.json`, which packagist.org advertises as `metadata-changes-url`. Both `vendor/package` and its `vendor/package~dev` file are covered. Packagist keeps 24 hours of changes; if the proxy's position is older than that, Packagist asks for a resync and all cached Composer metadata is revalidated. repo.packagist.org sends `Cache-Control: max-age=900`, but Packagist purges its CDN when a package changes, so revalidation gets the new file within seconds; the proxy doesn't apply that header either way. +- npm reads `https://replicate.npmjs.com/registry/_changes`, the public registry's feed, whatever `upstream.npm` is. It is only useful when `upstream.npm` is registry.npmjs.org or serves the same packages. Packuments (`/npm/{name}`) are covered; version and dist-tag documents (`/npm/{name}/latest`) still wait out `metadata_ttl`. A busy minute on npm lists a few thousand packages; the proxy reads up to 100,000 per pass and continues where it stopped on the next. + +A new release therefore shows up about a minute after the registry lists it rather than up to `metadata_ttl` later. The feed position is kept in memory: after a restart the proxy starts from the feed's current position, and changes made while it was down are covered by `metadata_ttl` as before. While a feed can't be read the proxy logs a warning and retries with a growing delay, up to 15 minutes, and `metadata_ttl` applies meanwhile. A request whose upstream fetch overlaps the change and the feed read can still store the old document for one TTL. Each instance sharing a database reads the feeds itself. `proxy_metadata_expired_by_change_feed_total` counts the cached documents marked, by ecosystem, and debug logging names each one. + ### 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 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. diff --git a/internal/changefeed/changefeed.go b/internal/changefeed/changefeed.go new file mode 100644 index 00000000..2e2d48f3 --- /dev/null +++ b/internal/changefeed/changefeed.go @@ -0,0 +1,261 @@ +// Package changefeed reads the feeds in which public registries list the +// packages that changed since a given point, so the proxy can revalidate its +// cached metadata for them before metadata_ttl runs out. +package changefeed + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "slices" + "strconv" + "strings" +) + +// Feed is one registry's change feed. It remembers how far it has read, in +// memory only: after a restart it starts again from the registry's current +// position. +type Feed interface { + // Ecosystem is the metadata cache ecosystem the changed names belong to. + Ecosystem() string + // Changes returns what changed since the previous call. The first call + // only finds the feed's current position and reports no changes. On error + // the position stays where it was, so the next call asks again. + Changes(ctx context.Context) (Changes, error) +} + +// Changes is one read of a feed. +type Changes struct { + // Names are the cache keys of the documents that changed. + Names []string + // All is set when the feed can't say what changed, so every document of + // the ecosystem has to be treated as changed. + All bool +} + +// maxFeedResponse bounds a feed response read into memory. A full page of +// the npm feed is about 1.5 MB. +const maxFeedResponse = 64 << 20 + +// getJSON fetches u and decodes its JSON body into v. A status other than 200 +// is an error unless allowStatus also accepts it. +func getJSON(ctx context.Context, client *http.Client, u string, v any, allowStatus ...int) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return fmt.Errorf("creating request: %w", err) + } + req.Header.Set("Accept", "application/json") + resp, err := client.Do(req) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK && !slices.Contains(allowStatus, resp.StatusCode) { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxFeedResponse)) + return fmt.Errorf("%s: unexpected status %d", redact(u), resp.StatusCode) + } + body, err := io.ReadAll(io.LimitReader(resp.Body, maxFeedResponse+1)) + if err != nil { + return fmt.Errorf("%s: reading response: %w", redact(u), err) + } + if len(body) > maxFeedResponse { + return fmt.Errorf("%s: response larger than %d bytes", redact(u), maxFeedResponse) + } + if err := json.Unmarshal(body, v); err != nil { + return fmt.Errorf("%s: decoding response: %w", redact(u), err) + } + return nil +} + +// redact drops credentials and the query from u for error messages. +func redact(u string) string { + parsed, err := url.Parse(u) + if err != nil { + return "feed" + } + parsed.User = nil + parsed.RawQuery = "" + return parsed.String() +} + +// withQuery returns base with the query parameters in q added. +func withQuery(base string, q url.Values) string { + sep := "?" + if strings.Contains(base, "?") { + sep = "&" + } + return base + sep + q.Encode() +} + +// ComposerFeedURL is where a Packagist instance at apiURL, such as +// https://packagist.org, publishes its metadata change feed. Packagist +// advertises it as "metadata-changes-url" in its packages.json. +func ComposerFeedURL(apiURL string) string { + return strings.TrimSuffix(apiURL, "/") + "/metadata/changes.json" +} + +// ComposerFeed reads Packagist's metadata change feed +// (https://packagist.org/apidoc#track-package-updates). Its position is a +// timestamp in units of 1/10000 s. Asked without one it answers 400 with the +// current timestamp; asked with one older than its 24 hours of history it +// answers with a single "resync" action. Each update names the file that +// changed, "vendor/package" or "vendor/package~dev", which is also the key +// the Composer handler caches it under; a delete names the package, and +// covers both files. +type ComposerFeed struct { + client *http.Client + url string + since int64 +} + +// NewComposerFeed returns the feed at feedURL, see ComposerFeedURL. +func NewComposerFeed(client *http.Client, feedURL string) *ComposerFeed { + return &ComposerFeed{client: client, url: feedURL} +} + +func (f *ComposerFeed) Ecosystem() string { return "composer" } + +type composerChanges struct { + Actions []struct { + Type string `json:"type"` + Package string `json:"package"` + } `json:"actions"` + Timestamp int64 `json:"timestamp"` + Error string `json:"error"` +} + +func (f *ComposerFeed) Changes(ctx context.Context) (Changes, error) { + if f.since == 0 { + var start composerChanges + if err := getJSON(ctx, f.client, f.url, &start, http.StatusBadRequest); err != nil { + return Changes{}, err + } + if start.Timestamp <= 0 { + return Changes{}, errors.New("composer change feed: no timestamp to start from") + } + f.since = start.Timestamp + return Changes{}, nil + } + + var page composerChanges + u := withQuery(f.url, url.Values{"since": {strconv.FormatInt(f.since, 10)}}) + if err := getJSON(ctx, f.client, u, &page); err != nil { + return Changes{}, err + } + if page.Error != "" || page.Timestamp <= 0 { + return Changes{}, fmt.Errorf("composer change feed: %q", page.Error) + } + var changes Changes + for _, a := range page.Actions { + switch a.Type { + case "update": + changes.Names = append(changes.Names, a.Package) + case "delete": + // A deleted package takes its dev versions with it. + changes.Names = append(changes.Names, a.Package, a.Package+"~dev") + case "resync": + // The position is too old for the feed to list what changed. + changes.All = true + } + } + f.since = page.Timestamp + return changes, nil +} + +// NPMFeedURL is the change feed of the public npm registry. +const NPMFeedURL = "https://replicate.npmjs.com/registry/_changes" + +// npmPageLimit is the most changes the npm feed returns in one response. +const npmPageLimit = 10000 + +// npmMaxPages bounds how many pages one Changes call reads. The rest is read +// on the next call. +const npmMaxPages = 10 + +// NPMFeed reads the npm registry's CouchDB-style changes feed at +// replicate.npmjs.com. Its position is a sequence, which npm asks followers +// to treat as opaque; the database document at the feed's parent URL, with +// its trailing slash (without it the URL is 404), gives the current one as +// update_seq. A page holds at most 10000 changes (1000 when no limit is +// given). Each result's id is a package name, which is also the key the npm +// handler caches its packument under. +type NPMFeed struct { + client *http.Client + url string + // since is the sequence exactly as the feed returned it. + since json.RawMessage +} + +// NewNPMFeed returns the feed at feedURL, see NPMFeedURL. +func NewNPMFeed(client *http.Client, feedURL string) *NPMFeed { + return &NPMFeed{client: client, url: feedURL} +} + +func (f *NPMFeed) Ecosystem() string { return "npm" } + +type npmChanges struct { + Results []struct { + ID string `json:"id"` + } `json:"results"` + LastSeq json.RawMessage `json:"last_seq"` +} + +func (f *NPMFeed) Changes(ctx context.Context) (Changes, error) { + if f.since == nil { + var info struct { + UpdateSeq json.RawMessage `json:"update_seq"` + } + if err := getJSON(ctx, f.client, strings.TrimSuffix(f.url, "_changes"), &info); err != nil { + return Changes{}, err + } + if len(info.UpdateSeq) == 0 { + return Changes{}, errors.New("npm change feed: no update_seq to start from") + } + f.since = info.UpdateSeq + return Changes{}, nil + } + + var changes Changes + for range npmMaxPages { + var page npmChanges + u := withQuery(f.url, url.Values{ + "since": {seqParam(f.since)}, + "limit": {strconv.Itoa(npmPageLimit)}, + }) + if err := getJSON(ctx, f.client, u, &page); err != nil { + // Changes already read are reported, and the position kept + // after them, so they aren't lost to a later page's failure. + if len(changes.Names) > 0 { + return changes, nil + } + return Changes{}, err + } + if len(page.LastSeq) == 0 { + return changes, errors.New("npm change feed: response without last_seq") + } + for _, r := range page.Results { + if r.ID != "" && !strings.HasPrefix(r.ID, "_design/") { + changes.Names = append(changes.Names, r.ID) + } + } + f.since = page.LastSeq + if len(page.Results) < npmPageLimit { + break + } + } + return changes, nil +} + +// seqParam returns a sequence as the since parameter takes it: a number as +// is, a string without its quotes. +func seqParam(seq json.RawMessage) string { + var s string + if json.Unmarshal(seq, &s) == nil { + return s + } + return string(seq) +} diff --git a/internal/changefeed/changefeed_test.go b/internal/changefeed/changefeed_test.go new file mode 100644 index 00000000..21a47655 --- /dev/null +++ b/internal/changefeed/changefeed_test.go @@ -0,0 +1,198 @@ +package changefeed + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" +) + +// fakePackagist serves a change feed the way packagist.org does: without +// since, 400 with the current timestamp; with an old since, a resync; with a +// known one, the actions recorded after it. +type fakePackagist struct { + mu sync.Mutex + now int64 + actions map[int64]string // since -> JSON actions + asked []string +} + +func (p *fakePackagist) ServeHTTP(w http.ResponseWriter, r *http.Request) { + p.mu.Lock() + defer p.mu.Unlock() + since := r.URL.Query().Get("since") + p.asked = append(p.asked, since) + w.Header().Set("Content-Type", "application/json") + if since == "" { + w.WriteHeader(http.StatusBadRequest) + _, _ = fmt.Fprintf(w, `{"error":"Invalid or missing since query parameter","timestamp":%d}`, p.now) + return + } + n, _ := strconv.ParseInt(since, 10, 64) + actions, ok := p.actions[n] + if !ok { + actions = `[{"type":"resync","package":"*","time":1}]` + } + _, _ = fmt.Fprintf(w, `{"actions":%s,"timestamp":%d}`, actions, p.now) +} + +func TestComposerFeed(t *testing.T) { + p := &fakePackagist{now: 100, actions: map[int64]string{ + 100: `[{"type":"update","package":"acme/lib","time":1},{"type":"update","package":"acme/lib~dev","time":1},{"type":"delete","package":"acme/gone","time":1}]`, + 200: `[]`, + }} + srv := httptest.NewServer(p) + defer srv.Close() + feed := NewComposerFeed(srv.Client(), ComposerFeedURL(srv.URL+"/")) + ctx := context.Background() + + if c, err := feed.Changes(ctx); err != nil || c.Names != nil || c.All { + t.Fatalf("first read = %+v, %v; want the start position only", c, err) + } + p.mu.Lock() + p.now = 200 + p.mu.Unlock() + c, err := feed.Changes(ctx) + if err != nil { + t.Fatal(err) + } + if want := []string{"acme/lib", "acme/lib~dev", "acme/gone", "acme/gone~dev"}; !slices.Equal(c.Names, want) || c.All { + t.Errorf("changes = %+v, want %v", c, want) + } + if c, err := feed.Changes(ctx); err != nil || len(c.Names) != 0 || c.All { + t.Errorf("changes without news = %+v, %v", c, err) + } + + // A position the feed no longer has history for gets a resync. + feed.since = 50 + if c, err := feed.Changes(ctx); err != nil || !c.All { + t.Errorf("changes after a resync = %+v, %v, want All", c, err) + } + if want := []string{"", "100", "200", "50"}; !slices.Equal(p.asked, want) { + t.Errorf("asked since %q, want %q", p.asked, want) + } +} + +func TestComposerFeedErrorKeepsPosition(t *testing.T) { + var fail atomic.Bool + fail.Store(true) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if fail.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + if r.URL.Query().Get("since") != "7" { + t.Errorf("since = %q, want 7", r.URL.Query().Get("since")) + } + _, _ = w.Write([]byte(`{"actions":[{"type":"update","package":"acme/lib","time":1}],"timestamp":8}`)) + })) + defer srv.Close() + feed := &ComposerFeed{client: srv.Client(), url: srv.URL, since: 7} + if _, err := feed.Changes(context.Background()); err == nil { + t.Fatal("want an error from a failing feed") + } + if feed.since != 7 { + t.Fatalf("since = %d after an error, want 7", feed.since) + } + fail.Store(false) + if c, err := feed.Changes(context.Background()); err != nil || !slices.Equal(c.Names, []string{"acme/lib"}) { + t.Errorf("changes after recovery = %+v, %v", c, err) + } +} + +// fakeReplicate serves a CouchDB-style changes feed over seqs 1..len(ids). +type fakeReplicate struct { + mu sync.Mutex + ids []string + limit int + asked []string +} + +func (f *fakeReplicate) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + if r.URL.Path == "/registry/" { + _, _ = fmt.Fprintf(w, `{"db_name":"registry","update_seq":%d}`, len(f.ids)) + return + } + if r.URL.Path != "/registry/_changes" { + w.WriteHeader(http.StatusNotFound) + return + } + since, _ := strconv.Atoi(r.URL.Query().Get("since")) + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + f.asked = append(f.asked, r.URL.Query().Get("since")) + if f.limit > 0 { + limit = min(limit, f.limit) + } + var results []string + last := since + for seq := since + 1; seq <= len(f.ids) && len(results) < limit; seq++ { + results = append(results, fmt.Sprintf(`{"seq":%d,"id":%q,"changes":[{"rev":"1-a"}]}`, seq, f.ids[seq-1])) + last = seq + } + _, _ = fmt.Fprintf(w, `{"results":[%s],"last_seq":%d}`, strings.Join(results, ","), last) +} + +func TestNPMFeed(t *testing.T) { + f := &fakeReplicate{ids: []string{"old"}} + srv := httptest.NewServer(f) + defer srv.Close() + feed := NewNPMFeed(srv.Client(), srv.URL+"/registry/_changes") + ctx := context.Background() + + if c, err := feed.Changes(ctx); err != nil || c.Names != nil { + t.Fatalf("first read = %+v, %v; want the start position only", c, err) + } + f.mu.Lock() + f.ids = append(f.ids, "lodash", "@scope/pkg", "_design/app") + f.mu.Unlock() + c, err := feed.Changes(ctx) + if err != nil { + t.Fatal(err) + } + if want := []string{"lodash", "@scope/pkg"}; !slices.Equal(c.Names, want) { + t.Errorf("changes = %v, want %v", c.Names, want) + } + if c, err := feed.Changes(ctx); err != nil || len(c.Names) != 0 { + t.Errorf("changes without news = %+v, %v", c, err) + } + if want := []string{"1", "4"}; !slices.Equal(f.asked, want) { + t.Errorf("asked since %q, want %q", f.asked, want) + } +} + +// TestNPMFeedPages checks that a backlog longer than a page is read page by +// page, and that one read stops after npmMaxPages. +func TestNPMFeedPages(t *testing.T) { + f := &fakeReplicate{limit: npmPageLimit} + srv := httptest.NewServer(f) + defer srv.Close() + feed := &NPMFeed{client: srv.Client(), url: srv.URL + "/registry/_changes", since: []byte("0")} + f.ids = make([]string, npmPageLimit*(npmMaxPages+1)+5) + for i := range f.ids { + f.ids[i] = "pkg-" + strconv.Itoa(i) + } + + c, err := feed.Changes(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(c.Names) != npmPageLimit*npmMaxPages { + t.Errorf("first read returned %d names, want %d", len(c.Names), npmPageLimit*npmMaxPages) + } + c, err = feed.Changes(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(c.Names) != npmPageLimit+5 { + t.Errorf("second read returned %d names, want %d", len(c.Names), npmPageLimit+5) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index a6606994..1787dc27 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -590,6 +590,12 @@ type UpstreamConfig struct { // wins. NPMRoutes map[string]string `json:"npm_routes" yaml:"npm_routes"` + // NPMChangeFeed follows the public npm registry's change feed and + // revalidates cached packuments of packages listed in it on their next + // request, rather than serving them until metadata_ttl runs out. + // Requires cache_metadata. Default: false. + NPMChangeFeed bool `json:"npm_change_feed" yaml:"npm_change_feed"` + // Cargo is the upstream cargo index URL. // Default: https://index.crates.io Cargo string `json:"cargo" yaml:"cargo"` @@ -682,6 +688,12 @@ type UpstreamConfig struct { // matching pattern wins. ComposerRoutes map[string]string `json:"composer_routes" yaml:"composer_routes"` + // ComposerChangeFeed follows the metadata change feed of + // upstream.composer and revalidates cached metadata of packages listed + // in it on their next request, rather than serving it until + // metadata_ttl runs out. Requires cache_metadata. Default: false. + ComposerChangeFeed bool `json:"composer_change_feed" yaml:"composer_change_feed"` + // Conan is the upstream Conan registry URL. // Default: https://center.conan.io Conan string `json:"conan" yaml:"conan"` @@ -969,6 +981,12 @@ func (c *Config) validateUpstream() error { if c.Upstream.GoSumDBsOffline && !c.CacheMetadata { return fmt.Errorf("upstream.go_sumdbs_offline requires cache_metadata") } + if c.Upstream.NPMChangeFeed && !c.CacheMetadata { + return fmt.Errorf("upstream.npm_change_feed requires cache_metadata") + } + if c.Upstream.ComposerChangeFeed && !c.CacheMetadata { + return fmt.Errorf("upstream.composer_change_feed requires cache_metadata") + } return nil } @@ -1407,6 +1425,7 @@ func (c *Config) LoadFromEnv() { setEnvBool(&c.Upstream.AllowLoopback, "PROXY_UPSTREAM_ALLOW_LOOPBACK") setEnvString(&c.Upstream.NPM, "PROXY_UPSTREAM_NPM") setEnvBool(&c.Upstream.NPMFullMetadata, "PROXY_UPSTREAM_NPM_FULL_METADATA") + setEnvBool(&c.Upstream.NPMChangeFeed, "PROXY_UPSTREAM_NPM_CHANGE_FEED") setEnvBool(&c.Upstream.GoSumDBsOffline, "PROXY_UPSTREAM_GO_SUMDBS_OFFLINE") setEnvString(&c.Upstream.Cargo, "PROXY_UPSTREAM_CARGO") setEnvString(&c.Upstream.CargoDownload, "PROXY_UPSTREAM_CARGO_DOWNLOAD") @@ -1423,6 +1442,7 @@ func (c *Config) LoadFromEnv() { setEnvString(&c.Upstream.NuGetSearch, "PROXY_UPSTREAM_NUGET_SEARCH") setEnvString(&c.Upstream.Composer, "PROXY_UPSTREAM_COMPOSER") setEnvString(&c.Upstream.ComposerRepository, "PROXY_UPSTREAM_COMPOSER_REPOSITORY") + setEnvBool(&c.Upstream.ComposerChangeFeed, "PROXY_UPSTREAM_COMPOSER_CHANGE_FEED") setEnvString(&c.Upstream.Conan, "PROXY_UPSTREAM_CONAN") setEnvString(&c.Upstream.Chef, "PROXY_UPSTREAM_CHEF") setEnvString(&c.Upstream.Conda, "PROXY_UPSTREAM_CONDA") diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 7590a64f..0730e018 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1137,6 +1137,17 @@ func TestLoadMetadataTTLFromEnv(t *testing.T) { } } +func TestLoadChangeFeedsFromEnv(t *testing.T) { + cfg := Default() + t.Setenv("PROXY_UPSTREAM_NPM_CHANGE_FEED", "true") + t.Setenv("PROXY_UPSTREAM_COMPOSER_CHANGE_FEED", "true") + cfg.LoadFromEnv() + + if !cfg.Upstream.NPMChangeFeed || !cfg.Upstream.ComposerChangeFeed { + t.Errorf("change feeds = npm %t, composer %t, want both on", cfg.Upstream.NPMChangeFeed, cfg.Upstream.ComposerChangeFeed) + } +} + func TestParseGradleBuildCacheConfig(t *testing.T) { cfg := Default() @@ -1725,6 +1736,30 @@ func TestValidateGoSumDBs(t *testing.T) { cfg.Upstream.GoSumDBsOffline = true }, }, + { + name: "npm change feed requires metadata caching", + modify: func(cfg *Config) { + cfg.CacheMetadata = false + cfg.Upstream.NPMChangeFeed = true + }, + wantErr: true, + }, + { + name: "Composer change feed requires metadata caching", + modify: func(cfg *Config) { + cfg.CacheMetadata = false + cfg.Upstream.ComposerChangeFeed = true + }, + wantErr: true, + }, + { + name: "change feeds with metadata caching", + modify: func(cfg *Config) { + cfg.CacheMetadata = true + cfg.Upstream.NPMChangeFeed = true + cfg.Upstream.ComposerChangeFeed = true + }, + }, } for _, tt := range tests { diff --git a/internal/database/metadata_cache_test.go b/internal/database/metadata_cache_test.go index 0e19f6c7..ee16c252 100644 --- a/internal/database/metadata_cache_test.go +++ b/internal/database/metadata_cache_test.go @@ -2,7 +2,9 @@ package database import ( "database/sql" + "fmt" "path/filepath" + "slices" "testing" "time" ) @@ -448,3 +450,91 @@ func TestRefreshMetadataCache(t *testing.T) { t.Errorf("refresh of a missing row = %v, %v, want false", refreshed, err) } } + +// TestExpireMetadataCache checks that expiring clears fetched_at only on the +// named rows of the ecosystem fetched before the cutoff, and keeps their +// documents and validators. +func TestExpireMetadataCache(t *testing.T) { + db := setupMetadataCacheDB(t) + cutoff := time.Now().UTC().Truncate(time.Second) + rows := []struct { + ecosystem, name string + fetched time.Time + }{ + {"composer", "acme/lib", cutoff.Add(-time.Minute)}, + {"composer", "acme/lib~dev", cutoff.Add(-time.Minute)}, + {"composer", "acme/new", cutoff.Add(time.Second)}, + {"composer", "acme/other", cutoff.Add(-time.Minute)}, + {"composer", "routes/abc/acme/lib", cutoff.Add(-time.Minute)}, + {testEcosystemNPM, "acme/lib", cutoff.Add(-time.Minute)}, + } + for _, r := range rows { + err := db.UpsertMetadataCache(&MetadataCacheEntry{ + Ecosystem: r.ecosystem, + Name: r.name, + StoragePath: "_metadata/" + r.ecosystem + "/" + r.name, + ETag: sql.NullString{String: `"abc"`, Valid: true}, + ContentDigest: sql.NullString{String: "sha256:aaaa", Valid: true}, + FetchedAt: sql.NullTime{Time: r.fetched, Valid: true}, + }) + if err != nil { + t.Fatal(err) + } + } + + expired, err := db.ExpireMetadataCache("composer", []string{"acme/lib", "acme/lib~dev", "acme/new", "acme/missing"}, cutoff) + if err != nil { + t.Fatal(err) + } + slices.Sort(expired) + if want := []string{"acme/lib", "acme/lib~dev"}; !slices.Equal(expired, want) { + t.Errorf("expired = %v, want %v", expired, want) + } + got, err := db.GetMetadataCache("composer", "acme/lib") + if err != nil { + t.Fatal(err) + } + if got.FetchedAt.Valid || got.ETag.String != `"abc"` || got.ContentDigest.String != "sha256:aaaa" { + t.Errorf("expired row = %+v, want no fetched_at and the same validators", got) + } + for _, r := range []struct{ ecosystem, name string }{{"composer", "acme/new"}, {"composer", "acme/other"}, {testEcosystemNPM, "acme/lib"}} { + if got, _ := db.GetMetadataCache(r.ecosystem, r.name); !got.FetchedAt.Valid { + t.Errorf("%s %s expired, want it left alone", r.ecosystem, r.name) + } + } + + expired, err = db.ExpireAllMetadataCache("composer", cutoff) + if err != nil { + t.Fatal(err) + } + if want := []string{"acme/other"}; !slices.Equal(expired, want) { + t.Errorf("expired all = %v, want %v (routed and newer rows kept)", expired, want) + } +} + +// TestExpireMetadataCacheBatches checks that more names than one statement +// takes are all expired. +func TestExpireMetadataCacheBatches(t *testing.T) { + db := setupMetadataCacheDB(t) + cutoff := time.Now().UTC() + names := make([]string, expireMetadataBatch+2) + for i := range names { + names[i] = fmt.Sprintf("pkg-%d", i) + } + for _, name := range []string{names[0], names[len(names)-1]} { + err := db.UpsertMetadataCache(&MetadataCacheEntry{ + Ecosystem: testEcosystemNPM, Name: name, StoragePath: "_metadata/npm/" + name, + FetchedAt: sql.NullTime{Time: cutoff.Add(-time.Minute), Valid: true}, + }) + if err != nil { + t.Fatal(err) + } + } + expired, err := db.ExpireMetadataCache(testEcosystemNPM, names, cutoff) + if err != nil { + t.Fatal(err) + } + if len(expired) != 2 { + t.Errorf("expired = %v, want both stored names", expired) + } +} diff --git a/internal/database/queries.go b/internal/database/queries.go index d7150e80..d97ef255 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -4,7 +4,9 @@ import ( "database/sql" "errors" "fmt" + "slices" "sort" + "strings" "time" "github.com/git-pkgs/artifacts" @@ -1357,6 +1359,50 @@ func (db *DB) ForgetMetadataDocument(ecosystem, name, from, to string) error { return err } +// expireMetadataBatch is how many names one ExpireMetadataCache statement +// covers, well below the bind parameter limits of SQLite and Postgres. +const expireMetadataBatch = 500 + +// ExpireMetadataCache clears fetched_at on the metadata cache rows for +// ecosystem and names that were fetched before cutoff, and returns the names +// of the rows it changed. The rows keep their documents and validators, so +// the next lookup of each revalidates it upstream instead of serving it as +// fresh. It is for an upstream that reports which documents changed. +func (db *DB) ExpireMetadataCache(ecosystem string, names []string, cutoff time.Time) ([]string, error) { + var expired []string + for batch := range slices.Chunk(names, expireMetadataBatch) { + args := []any{time.Now(), ecosystem, cutoff.UTC()} + for _, name := range batch { + args = append(args, name) + } + var got []string + err := db.DB.Select(&got, db.Rebind(` + UPDATE metadata_cache SET fetched_at = NULL, updated_at = ? + WHERE ecosystem = ? AND fetched_at < ? + AND name IN (`+strings.TrimSuffix(strings.Repeat("?, ", len(batch)), ", ")+`) + RETURNING name + `), args...) + if err != nil { + return expired, err + } + expired = append(expired, got...) + } + return expired, nil +} + +// ExpireAllMetadataCache is ExpireMetadataCache for every row of ecosystem, +// except those of packages routed to another registry (named "routes/..."), +// for an upstream that can't say which of its documents changed. +func (db *DB) ExpireAllMetadataCache(ecosystem string, cutoff time.Time) ([]string, error) { + var expired []string + err := db.DB.Select(&expired, db.Rebind(` + UPDATE metadata_cache SET fetched_at = NULL, updated_at = ? + WHERE ecosystem = ? AND fetched_at < ? AND name NOT LIKE 'routes/%' + RETURNING name + `), time.Now(), ecosystem, cutoff.UTC()) + return expired, err +} + // UpsertMetadataCache writes entry and marks it accessed now, since a // document is stored because a request asked for it. func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error { diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 708b5b30..984a607a 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -86,6 +86,14 @@ var ( []string{"ecosystem"}, ) + MetadataExpiredByChangeFeed = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "proxy_metadata_expired_by_change_feed_total", + Help: "Cached metadata documents set to revalidate because an upstream change feed listed them, by ecosystem", + }, + []string{"ecosystem"}, + ) + FetchLeaseWaits = prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "proxy_fetch_lease_waits_total", @@ -311,6 +319,7 @@ func init() { CachedArtifacts, ArtifactsEvicted, MetadataEvicted, + MetadataExpiredByChangeFeed, FetchLeaseWaits, UpstreamFetchDuration, UpstreamErrors, @@ -409,6 +418,13 @@ func RecordMetadataEvicted(ecosystem string) { MetadataEvicted.WithLabelValues(purl.NormalizeEcosystem(ecosystem)).Inc() } +// RecordMetadataExpiredByChangeFeed counts n metadata documents an upstream +// change feed listed as changed. ecosystem is the metadata cache's, +// normalized like the cache hit counter. +func RecordMetadataExpiredByChangeFeed(ecosystem string, n int) { + MetadataExpiredByChangeFeed.WithLabelValues(purl.NormalizeEcosystem(ecosystem)).Add(float64(n)) +} + // RecordFetchLeaseWait counts a cache miss that found another instance // sharing the database already fetching it. func RecordFetchLeaseWait(kind string) { diff --git a/internal/server/analytics_coverage_test.go b/internal/server/analytics_coverage_test.go index 82e5ca21..02e67e2c 100644 --- a/internal/server/analytics_coverage_test.go +++ b/internal/server/analytics_coverage_test.go @@ -34,25 +34,26 @@ var metricSurface = map[string]string{ "proxy_client_response_bytes_total": "Runtime: Request sources -- By client", // Registry-derived, shown in the Runtime card. - "proxy_requests_total": "Runtime: Requests + Responses by status", - "proxy_request_duration_seconds": "Runtime: Mean latency", - "proxy_active_requests": "Runtime: In flight", - "proxy_cache_hits_total": "Runtime: Artifact and metadata cache lookups + artifact hit rate", - "proxy_cache_misses_total": "Runtime: Artifact and metadata cache lookups + artifact hit rate", - "proxy_metadata_cache_served_bytes_total": "Runtime: Metadata cache lookups -- served from cache", - "proxy_upstream_fetch_duration_seconds": "Runtime: Upstream fetches + mean fetch", - "proxy_upstream_errors_total": "Runtime: Upstream errors", - "proxy_storage_operation_duration_seconds": "Runtime: Storage operations", - "proxy_storage_errors_total": "Runtime: Storage errors", - "proxy_integrity_failures_total": "Runtime: Integrity failures", - "proxy_health_probe_failures_total": "Runtime: Health probe failures", - "proxy_circuit_breaker_state": "Runtime: Circuit breakers", - "proxy_circuit_breaker_trips_total": "Runtime: Circuit breaker trips", - "proxy_artifacts_evicted_total": "Runtime: Evicted artifacts", - "proxy_metadata_evicted_total": "Runtime: Evicted metadata", - "proxy_scan_duration_seconds": "Runtime: Pre-cache scanning -- Scans", - "proxy_scan_blocked_total": "Runtime: Pre-cache scanning -- Artifacts blocked", - "proxy_scan_errors_total": "Runtime: Pre-cache scanning -- Scan errors", + "proxy_requests_total": "Runtime: Requests + Responses by status", + "proxy_request_duration_seconds": "Runtime: Mean latency", + "proxy_active_requests": "Runtime: In flight", + "proxy_cache_hits_total": "Runtime: Artifact and metadata cache lookups + artifact hit rate", + "proxy_cache_misses_total": "Runtime: Artifact and metadata cache lookups + artifact hit rate", + "proxy_metadata_cache_served_bytes_total": "Runtime: Metadata cache lookups -- served from cache", + "proxy_upstream_fetch_duration_seconds": "Runtime: Upstream fetches + mean fetch", + "proxy_upstream_errors_total": "Runtime: Upstream errors", + "proxy_storage_operation_duration_seconds": "Runtime: Storage operations", + "proxy_storage_errors_total": "Runtime: Storage errors", + "proxy_integrity_failures_total": "Runtime: Integrity failures", + "proxy_health_probe_failures_total": "Runtime: Health probe failures", + "proxy_circuit_breaker_state": "Runtime: Circuit breakers", + "proxy_circuit_breaker_trips_total": "Runtime: Circuit breaker trips", + "proxy_artifacts_evicted_total": "Runtime: Evicted artifacts", + "proxy_metadata_evicted_total": "Runtime: Evicted metadata", + "proxy_metadata_expired_by_change_feed_total": "Runtime: Metadata changed upstream", + "proxy_scan_duration_seconds": "Runtime: Pre-cache scanning -- Scans", + "proxy_scan_blocked_total": "Runtime: Pre-cache scanning -- Artifacts blocked", + "proxy_scan_errors_total": "Runtime: Pre-cache scanning -- Scan errors", } // TestEveryMetricIsSurfaced fails when a metric is registered but has no home @@ -80,6 +81,7 @@ func TestEveryMetricIsSurfaced(t *testing.T) { metrics.UpdateCacheStats(1, 1) metrics.RecordArtifactEvicted("retention", "npm") metrics.RecordMetadataEvicted("npm") + metrics.RecordMetadataExpiredByChangeFeed("npm", 1) metrics.UpdateEcosystemStats([]metrics.EcosystemStats{{Ecosystem: "npm", DownloadedBytes: 1}}) metrics.UpdateMetadataCacheStats([]metrics.MetadataCacheStats{{Ecosystem: "npm", Entries: 1, Size: 1}}) diff --git a/internal/server/analytics_runtime.go b/internal/server/analytics_runtime.go index 57738795..103e33eb 100644 --- a/internal/server/analytics_runtime.go +++ b/internal/server/analytics_runtime.go @@ -35,6 +35,7 @@ type RuntimeView struct { ProbeFailures []LabelledCount Evictions []LabelledCount MetadataEvictions []LabelledCount + ChangeFeedExpired []LabelledCount Breakers []BreakerRow BreakerTrips []LabelledCount @@ -161,6 +162,7 @@ func runtimeView(snap *metrics.Snapshot, failureDetails bool) RuntimeView { v.ProbeFailures = failures("proxy_health_probe_failures_total", "step") v.Evictions = countRows(snap, "proxy_artifacts_evicted_total", "reason", "ecosystem") v.MetadataEvictions = countRows(snap, "proxy_metadata_evicted_total", "ecosystem") + v.ChangeFeedExpired = countRows(snap, "proxy_metadata_expired_by_change_feed_total", "ecosystem") v.Breakers = breakerRows(snap) v.BreakerTrips = failures("proxy_circuit_breaker_trips_total", "registry") diff --git a/internal/server/analytics_runtime_test.go b/internal/server/analytics_runtime_test.go index 66c154e1..03bc2360 100644 --- a/internal/server/analytics_runtime_test.go +++ b/internal/server/analytics_runtime_test.go @@ -170,6 +170,24 @@ func TestRuntimeViewShowsMetadataEvictions(t *testing.T) { } } +func TestRuntimeViewShowsChangeFeedExpiries(t *testing.T) { + reg := prometheus.NewRegistry() + expired := prometheus.NewCounterVec( + prometheus.CounterOpts{Name: "proxy_metadata_expired_by_change_feed_total", Help: "t"}, []string{"ecosystem"}) + reg.MustRegister(expired) + expired.WithLabelValues("packagist").Add(2) + + snap, err := metrics.GatherFrom(reg) + if err != nil { + t.Fatalf("GatherFrom: %v", err) + } + v := runtimeView(snap, false) + + if len(v.ChangeFeedExpired) != 1 || v.ChangeFeedExpired[0].Label != "packagist" || v.ChangeFeedExpired[0].Count != "2" { + t.Errorf("ChangeFeedExpired = %+v, want packagist = 2", v.ChangeFeedExpired) + } +} + // A counter sitting at zero carries no information and would bury the rows // that do, so it is left out. func TestRuntimeViewDropsZeroCounters(t *testing.T) { diff --git a/internal/server/change_feeds.go b/internal/server/change_feeds.go new file mode 100644 index 00000000..059bd2f7 --- /dev/null +++ b/internal/server/change_feeds.go @@ -0,0 +1,94 @@ +package server + +import ( + "context" + "log/slog" + "net/http" + "time" + + "github.com/git-pkgs/proxy/internal/changefeed" + "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/proxy/internal/metrics" +) + +// changeFeedInterval is how often an upstream change feed is read. +const changeFeedInterval = time.Minute + +// changeFeedMaxBackoff caps the wait between reads of a failing feed. +const changeFeedMaxBackoff = 15 * time.Minute + +// startChangeFeeds follows the change feeds the configuration turns on. Each +// instance sharing a database follows them itself: the reads are small, and +// expiring a row twice does no harm. +func (s *Server) startChangeFeeds(ctx context.Context, client *http.Client, metadataTTL time.Duration) { + var feeds []changefeed.Feed + if s.cfg.Upstream.NPMChangeFeed { + feeds = append(feeds, changefeed.NewNPMFeed(client, changefeed.NPMFeedURL)) + } + if s.cfg.Upstream.ComposerChangeFeed { + feeds = append(feeds, changefeed.NewComposerFeed(client, changefeed.ComposerFeedURL(s.cfg.Upstream.Composer))) + } + if len(feeds) == 0 { + return + } + if !s.cfg.CacheMetadata || metadataTTL <= 0 { + // Every request already revalidates, so there is nothing to expire. + s.logger.Info("change feeds not followed: metadata is not served from cache without revalidating") + return + } + for _, feed := range feeds { + s.background.Go(func() { followChangeFeed(ctx, s.db, s.logger, feed, changeFeedInterval) }) + } +} + +// followChangeFeed reads feed every interval until ctx ends and expires the +// cached metadata it lists. After a failed read it waits twice as long as +// before, up to changeFeedMaxBackoff. +func followChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, feed changefeed.Feed, interval time.Duration) { + wait := interval + for { + if err := readChangeFeed(ctx, db, logger, feed); err != nil { + if ctx.Err() != nil { + return + } + wait = min(wait+wait, changeFeedMaxBackoff) + logger.Warn("change feed: read failed", "ecosystem", feed.Ecosystem(), "error", err, "retry_in", wait) + } else { + wait = interval + } + timer := time.NewTimer(wait) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } +} + +// readChangeFeed reads feed once and expires the cached metadata of what it +// lists, so the next request for it revalidates upstream. Only rows fetched +// before the read finished are expired: a row fetched after it was fetched +// after every change the read returned. If expiring fails, the feed has +// already moved past the changes, and those rows wait out metadata_ttl. +func readChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, feed changefeed.Feed) error { + changes, err := feed.Changes(ctx) + if err != nil { + return err + } + cutoff := time.Now() + ecosystem := feed.Ecosystem() + var expired []string + switch { + case changes.All: + logger.Info("change feed: upstream asked for a full resync, revalidating all cached metadata", "ecosystem", ecosystem) + expired, err = db.ExpireAllMetadataCache(ecosystem, cutoff) + case len(changes.Names) > 0: + expired, err = db.ExpireMetadataCache(ecosystem, changes.Names, cutoff) + } + for _, name := range expired { + logger.Debug("change feed: cached metadata changed upstream", "ecosystem", ecosystem, "key", name) + } + metrics.RecordMetadataExpiredByChangeFeed(ecosystem, len(expired)) + return err +} diff --git a/internal/server/change_feeds_test.go b/internal/server/change_feeds_test.go new file mode 100644 index 00000000..90441aca --- /dev/null +++ b/internal/server/change_feeds_test.go @@ -0,0 +1,272 @@ +package server + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/changefeed" + "github.com/git-pkgs/proxy/internal/config" + "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/proxy/internal/handler" + "github.com/git-pkgs/proxy/internal/storage" + "github.com/git-pkgs/registries/fetch" + "github.com/go-chi/chi/v5" +) + +// changingRegistry serves npm packuments, Composer metadata, and both +// registries' change feeds, all from one set of documents that a test can +// change. The feeds list every document changed since the previous read. +type changingRegistry struct { + *httptest.Server + mu sync.Mutex + docs map[string]string // path -> body + changed []string // paths changed since the feed was last read + feedSeq int + fetches map[string]int +} + +func newChangingRegistry(t *testing.T) *changingRegistry { + t.Helper() + reg := &changingRegistry{docs: map[string]string{}, fetches: map[string]int{}} + reg.Server = httptest.NewServer(http.HandlerFunc(reg.serve)) + t.Cleanup(reg.Close) + return reg +} + +func (reg *changingRegistry) set(path, body string) { + reg.mu.Lock() + defer reg.mu.Unlock() + if _, ok := reg.docs[path]; ok { + reg.changed = append(reg.changed, path) + } + reg.docs[path] = body +} + +func (reg *changingRegistry) fetched(path string) int { + reg.mu.Lock() + defer reg.mu.Unlock() + return reg.fetches[path] +} + +// takeChanged returns the changed paths with prefix and forgets them. +func (reg *changingRegistry) takeChanged(prefix string) []string { + var taken, kept []string + for _, p := range reg.changed { + if name, ok := strings.CutPrefix(p, prefix); ok { + taken = append(taken, name) + } else { + kept = append(kept, p) + } + } + reg.changed = kept + return taken +} + +func (reg *changingRegistry) serve(w http.ResponseWriter, r *http.Request) { + reg.mu.Lock() + defer reg.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/metadata/changes.json": + if r.URL.Query().Get("since") == "" { + w.WriteHeader(http.StatusBadRequest) + _, _ = io.WriteString(w, `{"error":"missing since","timestamp":1}`) + return + } + var actions []string + for _, name := range reg.takeChanged("/p2/") { + actions = append(actions, fmt.Sprintf(`{"type":"update","package":%q,"time":1}`, strings.TrimSuffix(name, ".json"))) + } + _, _ = fmt.Fprintf(w, `{"actions":[%s],"timestamp":2}`, strings.Join(actions, ",")) + return + case "/replicate/registry/": + _, _ = fmt.Fprintf(w, `{"update_seq":%d}`, reg.feedSeq) + return + case "/replicate/registry/_changes": + var results []string + for _, name := range reg.takeChanged("/npm/") { + reg.feedSeq++ + results = append(results, fmt.Sprintf(`{"seq":%d,"id":%q}`, reg.feedSeq, name)) + } + _, _ = fmt.Fprintf(w, `{"results":[%s],"last_seq":%d}`, strings.Join(results, ","), reg.feedSeq) + return + } + reg.fetches[r.URL.Path]++ + body, ok := reg.docs[r.URL.Path] + if !ok { + http.NotFound(w, r) + return + } + etag := fmt.Sprintf(`"%x"`, sha256.Sum256([]byte(body))) + if r.Header.Get("If-None-Match") == etag { + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set("ETag", etag) + _, _ = io.WriteString(w, body) +} + +func composerDoc(versions ...string) string { + var entries []string + for _, v := range versions { + entries = append(entries, fmt.Sprintf(`{"name":"acme/lib","version":%q,"version_normalized":"%s.0","dist":{"type":"zip","url":"https://example.test/acme/lib/%s.zip","reference":"r%s"}}`, v, v, v, v)) + } + return `{"packages":{"acme/lib":[` + strings.Join(entries, ",") + `]},"minified":"composer/2.0"}` +} + +func npmDoc(versions ...string) string { + var entries []string + for _, v := range versions { + entries = append(entries, fmt.Sprintf(`%q:{"name":"left-pad","version":%q,"dist":{"tarball":"https://registry.example.test/left-pad/-/left-pad-%s.tgz"}}`, v, v, v)) + } + return `{"name":"left-pad","dist-tags":{"latest":"` + versions[len(versions)-1] + `"},"versions":{` + strings.Join(entries, ",") + `}}` +} + +// changeFeedRouter mounts the npm and Composer handlers on reg, with +// metadata cached for an hour. +func changeFeedRouter(t *testing.T, reg *changingRegistry) (http.Handler, *database.DB) { + t.Helper() + db, err := database.Create(t.TempDir() + "/test.db") + if err != nil { + t.Fatalf("creating database: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + store, err := storage.OpenBucket(context.Background(), "file://"+t.TempDir()) + if err != nil { + t.Fatalf("opening storage: %v", err) + } + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(reg.Client()), fetch.WithMaxRetries(0)) + t.Cleanup(func() { _ = fetcher.Close() }) + proxy := handler.NewProxy(db, store, fetcher, fetch.NewResolver(), slog.New(slog.NewTextHandler(io.Discard, nil))) + proxy.HTTPClient = reg.Client() + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.SetMetadataRewriteCacheSize(1 << 20) + + cfg := config.Default() + cfg.BaseURL = "http://proxy.example" + cfg.Upstream.NPM = reg.URL + "/npm" + cfg.Upstream.Composer = reg.URL + cfg.Upstream.ComposerRepository = reg.URL + s := &Server{cfg: cfg} + r := chi.NewRouter() + s.mountProtocolHandlers(r, proxy) + return r, db +} + +func getBody(t *testing.T, h http.Handler, path string) string { + t.Helper() + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil)) + if w.Code != http.StatusOK { + t.Fatalf("GET %s = %d: %s", path, w.Code, w.Body.String()) + } + return w.Body.String() +} + +// TestChangeFeedRevalidatesChangedMetadata checks the whole path: cached +// metadata is served within its TTL even after upstream changed, until the +// change feed lists the package, after which the next request revalidates +// and serves the new release. Packages the feed doesn't list stay cached. +func TestChangeFeedRevalidatesChangedMetadata(t *testing.T) { + reg := newChangingRegistry(t) + reg.set("/p2/acme/lib.json", composerDoc("1.0.0")) + reg.set("/p2/acme/other.json", `{"packages":{"acme/other":[]},"minified":"composer/2.0"}`) + reg.set("/npm/left-pad", npmDoc("1.0.0")) + h, db := changeFeedRouter(t, reg) + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + feeds := []changefeed.Feed{ + changefeed.NewComposerFeed(reg.Client(), changefeed.ComposerFeedURL(reg.URL)), + changefeed.NewNPMFeed(reg.Client(), reg.URL+"/replicate/registry/_changes"), + } + for _, feed := range feeds { + if err := readChangeFeed(context.Background(), db, logger, feed); err != nil { + t.Fatalf("starting %s feed: %v", feed.Ecosystem(), err) + } + } + + getBody(t, h, "/composer/p2/acme/lib.json") + getBody(t, h, "/composer/p2/acme/other.json") + getBody(t, h, "/npm/left-pad") + + reg.set("/p2/acme/lib.json", composerDoc("1.0.0", "1.1.0")) + reg.set("/npm/left-pad", npmDoc("1.0.0", "1.1.0")) + if body := getBody(t, h, "/composer/p2/acme/lib.json"); strings.Contains(body, "1.1.0") { + t.Fatal("composer metadata revalidated before the feed listed it; the test proves nothing") + } + if body := getBody(t, h, "/npm/left-pad"); strings.Contains(body, "1.1.0") { + t.Fatal("npm packument revalidated before the feed listed it; the test proves nothing") + } + + for _, feed := range feeds { + if err := readChangeFeed(context.Background(), db, logger, feed); err != nil { + t.Fatalf("reading %s feed: %v", feed.Ecosystem(), err) + } + } + + if body := getBody(t, h, "/composer/p2/acme/lib.json"); !strings.Contains(body, "1.1.0") { + t.Errorf("composer metadata after the feed listed it = %s, want 1.1.0", body) + } + if body := getBody(t, h, "/npm/left-pad"); !strings.Contains(body, "1.1.0") { + t.Errorf("npm packument after the feed listed it = %s, want 1.1.0", body) + } + getBody(t, h, "/composer/p2/acme/other.json") + if n := reg.fetched("/p2/acme/other.json"); n != 1 { + t.Errorf("unlisted package fetched %d times, want once", n) + } + if n := reg.fetched("/p2/acme/lib.json"); n != 2 { + t.Errorf("listed package fetched %d times, want twice", n) + } +} + +// countingFeed fails its first reads and then reports no changes. +type countingFeed struct { + reads atomic.Int32 + failures int32 +} + +func (f *countingFeed) Ecosystem() string { return "npm" } + +func (f *countingFeed) Changes(context.Context) (changefeed.Changes, error) { + if f.reads.Add(1) <= f.failures { + return changefeed.Changes{}, errors.New("feed down") + } + return changefeed.Changes{}, nil +} + +// TestFollowChangeFeedStopsWithContext checks that a failing feed is retried +// rather than given up on, and that following stops when ctx ends. +func TestFollowChangeFeedStopsWithContext(t *testing.T) { + feed := &countingFeed{failures: 2} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + followChangeFeed(ctx, nil, slog.New(slog.NewTextHandler(io.Discard, nil)), feed, time.Millisecond) + }() + deadline := time.After(5 * time.Second) + for feed.reads.Load() < 4 { + select { + case <-deadline: + t.Fatalf("feed read %d times, want it read again after failing", feed.reads.Load()) + case <-time.After(time.Millisecond): + } + } + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("followChangeFeed did not return after ctx ended") + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 12619a6a..8d3a746d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -397,6 +397,7 @@ func (s *Server) serve(listener net.Listener) error { }) } s.startMirrorSchedules(bgCtx, proxy) + s.startChangeFeeds(bgCtx, proxy.HTTPClient, proxy.MetadataTTL) // Hosted package endpoints (opt-in via hosted.enabled or PROXY_HOSTED_ENABLED) if s.cfg.Hosted.Enabled { diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 292643c1..06f6b239 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -633,13 +633,6 @@ func TestSwiftCachedVersionPURLUsesStoredPackagePURL(t *testing.T) { } } -func min(a, b int) int { - if a < b { - return a - } - return b -} - func TestNPMPackageMetadata(t *testing.T) { ts := newTestServer(t) defer ts.close() diff --git a/internal/server/templates/components/runtime_metrics.html b/internal/server/templates/components/runtime_metrics.html index 18a3140c..b5427b7f 100644 --- a/internal/server/templates/components/runtime_metrics.html +++ b/internal/server/templates/components/runtime_metrics.html @@ -60,6 +60,7 @@

Runtime

{{template "runtime_counts" dict "Title" "Health probe failures" "Rows" .ProbeFailures "Empty" "None"}} {{template "runtime_counts" dict "Title" "Evicted artifacts" "Rows" .Evictions "Empty" "None"}} {{template "runtime_counts" dict "Title" "Evicted metadata" "Rows" .MetadataEvictions "Empty" "None"}} + {{template "runtime_counts" dict "Title" "Metadata changed upstream" "Rows" .ChangeFeedExpired "Empty" "None"}} {{template "runtime_counts" dict "Title" "Circuit breaker trips" "Rows" .BreakerTrips "Empty" "None"}}
From c4615ff56164f410211ede3732fadf07e4d6c244 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 30 Oct 2026 10:09:21 +0100 Subject: [PATCH 2/4] Read change feeds through git-pkgs/registries The feed readers move to registries' changes package. The proxy keeps the cursor and moves it past the changes only once their metadata is expired, so a failed expiry reads them again. --- internal/changefeed/changefeed.go | 261 ------------------------- internal/changefeed/changefeed_test.go | 198 ------------------- internal/server/change_feeds.go | 54 +++-- internal/server/change_feeds_test.go | 73 +++++-- internal/server/server.go | 48 +++-- 5 files changed, 123 insertions(+), 511 deletions(-) delete mode 100644 internal/changefeed/changefeed.go delete mode 100644 internal/changefeed/changefeed_test.go diff --git a/internal/changefeed/changefeed.go b/internal/changefeed/changefeed.go deleted file mode 100644 index 2e2d48f3..00000000 --- a/internal/changefeed/changefeed.go +++ /dev/null @@ -1,261 +0,0 @@ -// Package changefeed reads the feeds in which public registries list the -// packages that changed since a given point, so the proxy can revalidate its -// cached metadata for them before metadata_ttl runs out. -package changefeed - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "io" - "net/http" - "net/url" - "slices" - "strconv" - "strings" -) - -// Feed is one registry's change feed. It remembers how far it has read, in -// memory only: after a restart it starts again from the registry's current -// position. -type Feed interface { - // Ecosystem is the metadata cache ecosystem the changed names belong to. - Ecosystem() string - // Changes returns what changed since the previous call. The first call - // only finds the feed's current position and reports no changes. On error - // the position stays where it was, so the next call asks again. - Changes(ctx context.Context) (Changes, error) -} - -// Changes is one read of a feed. -type Changes struct { - // Names are the cache keys of the documents that changed. - Names []string - // All is set when the feed can't say what changed, so every document of - // the ecosystem has to be treated as changed. - All bool -} - -// maxFeedResponse bounds a feed response read into memory. A full page of -// the npm feed is about 1.5 MB. -const maxFeedResponse = 64 << 20 - -// getJSON fetches u and decodes its JSON body into v. A status other than 200 -// is an error unless allowStatus also accepts it. -func getJSON(ctx context.Context, client *http.Client, u string, v any, allowStatus ...int) error { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) - if err != nil { - return fmt.Errorf("creating request: %w", err) - } - req.Header.Set("Accept", "application/json") - resp, err := client.Do(req) - if err != nil { - return err - } - defer func() { _ = resp.Body.Close() }() - if resp.StatusCode != http.StatusOK && !slices.Contains(allowStatus, resp.StatusCode) { - _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxFeedResponse)) - return fmt.Errorf("%s: unexpected status %d", redact(u), resp.StatusCode) - } - body, err := io.ReadAll(io.LimitReader(resp.Body, maxFeedResponse+1)) - if err != nil { - return fmt.Errorf("%s: reading response: %w", redact(u), err) - } - if len(body) > maxFeedResponse { - return fmt.Errorf("%s: response larger than %d bytes", redact(u), maxFeedResponse) - } - if err := json.Unmarshal(body, v); err != nil { - return fmt.Errorf("%s: decoding response: %w", redact(u), err) - } - return nil -} - -// redact drops credentials and the query from u for error messages. -func redact(u string) string { - parsed, err := url.Parse(u) - if err != nil { - return "feed" - } - parsed.User = nil - parsed.RawQuery = "" - return parsed.String() -} - -// withQuery returns base with the query parameters in q added. -func withQuery(base string, q url.Values) string { - sep := "?" - if strings.Contains(base, "?") { - sep = "&" - } - return base + sep + q.Encode() -} - -// ComposerFeedURL is where a Packagist instance at apiURL, such as -// https://packagist.org, publishes its metadata change feed. Packagist -// advertises it as "metadata-changes-url" in its packages.json. -func ComposerFeedURL(apiURL string) string { - return strings.TrimSuffix(apiURL, "/") + "/metadata/changes.json" -} - -// ComposerFeed reads Packagist's metadata change feed -// (https://packagist.org/apidoc#track-package-updates). Its position is a -// timestamp in units of 1/10000 s. Asked without one it answers 400 with the -// current timestamp; asked with one older than its 24 hours of history it -// answers with a single "resync" action. Each update names the file that -// changed, "vendor/package" or "vendor/package~dev", which is also the key -// the Composer handler caches it under; a delete names the package, and -// covers both files. -type ComposerFeed struct { - client *http.Client - url string - since int64 -} - -// NewComposerFeed returns the feed at feedURL, see ComposerFeedURL. -func NewComposerFeed(client *http.Client, feedURL string) *ComposerFeed { - return &ComposerFeed{client: client, url: feedURL} -} - -func (f *ComposerFeed) Ecosystem() string { return "composer" } - -type composerChanges struct { - Actions []struct { - Type string `json:"type"` - Package string `json:"package"` - } `json:"actions"` - Timestamp int64 `json:"timestamp"` - Error string `json:"error"` -} - -func (f *ComposerFeed) Changes(ctx context.Context) (Changes, error) { - if f.since == 0 { - var start composerChanges - if err := getJSON(ctx, f.client, f.url, &start, http.StatusBadRequest); err != nil { - return Changes{}, err - } - if start.Timestamp <= 0 { - return Changes{}, errors.New("composer change feed: no timestamp to start from") - } - f.since = start.Timestamp - return Changes{}, nil - } - - var page composerChanges - u := withQuery(f.url, url.Values{"since": {strconv.FormatInt(f.since, 10)}}) - if err := getJSON(ctx, f.client, u, &page); err != nil { - return Changes{}, err - } - if page.Error != "" || page.Timestamp <= 0 { - return Changes{}, fmt.Errorf("composer change feed: %q", page.Error) - } - var changes Changes - for _, a := range page.Actions { - switch a.Type { - case "update": - changes.Names = append(changes.Names, a.Package) - case "delete": - // A deleted package takes its dev versions with it. - changes.Names = append(changes.Names, a.Package, a.Package+"~dev") - case "resync": - // The position is too old for the feed to list what changed. - changes.All = true - } - } - f.since = page.Timestamp - return changes, nil -} - -// NPMFeedURL is the change feed of the public npm registry. -const NPMFeedURL = "https://replicate.npmjs.com/registry/_changes" - -// npmPageLimit is the most changes the npm feed returns in one response. -const npmPageLimit = 10000 - -// npmMaxPages bounds how many pages one Changes call reads. The rest is read -// on the next call. -const npmMaxPages = 10 - -// NPMFeed reads the npm registry's CouchDB-style changes feed at -// replicate.npmjs.com. Its position is a sequence, which npm asks followers -// to treat as opaque; the database document at the feed's parent URL, with -// its trailing slash (without it the URL is 404), gives the current one as -// update_seq. A page holds at most 10000 changes (1000 when no limit is -// given). Each result's id is a package name, which is also the key the npm -// handler caches its packument under. -type NPMFeed struct { - client *http.Client - url string - // since is the sequence exactly as the feed returned it. - since json.RawMessage -} - -// NewNPMFeed returns the feed at feedURL, see NPMFeedURL. -func NewNPMFeed(client *http.Client, feedURL string) *NPMFeed { - return &NPMFeed{client: client, url: feedURL} -} - -func (f *NPMFeed) Ecosystem() string { return "npm" } - -type npmChanges struct { - Results []struct { - ID string `json:"id"` - } `json:"results"` - LastSeq json.RawMessage `json:"last_seq"` -} - -func (f *NPMFeed) Changes(ctx context.Context) (Changes, error) { - if f.since == nil { - var info struct { - UpdateSeq json.RawMessage `json:"update_seq"` - } - if err := getJSON(ctx, f.client, strings.TrimSuffix(f.url, "_changes"), &info); err != nil { - return Changes{}, err - } - if len(info.UpdateSeq) == 0 { - return Changes{}, errors.New("npm change feed: no update_seq to start from") - } - f.since = info.UpdateSeq - return Changes{}, nil - } - - var changes Changes - for range npmMaxPages { - var page npmChanges - u := withQuery(f.url, url.Values{ - "since": {seqParam(f.since)}, - "limit": {strconv.Itoa(npmPageLimit)}, - }) - if err := getJSON(ctx, f.client, u, &page); err != nil { - // Changes already read are reported, and the position kept - // after them, so they aren't lost to a later page's failure. - if len(changes.Names) > 0 { - return changes, nil - } - return Changes{}, err - } - if len(page.LastSeq) == 0 { - return changes, errors.New("npm change feed: response without last_seq") - } - for _, r := range page.Results { - if r.ID != "" && !strings.HasPrefix(r.ID, "_design/") { - changes.Names = append(changes.Names, r.ID) - } - } - f.since = page.LastSeq - if len(page.Results) < npmPageLimit { - break - } - } - return changes, nil -} - -// seqParam returns a sequence as the since parameter takes it: a number as -// is, a string without its quotes. -func seqParam(seq json.RawMessage) string { - var s string - if json.Unmarshal(seq, &s) == nil { - return s - } - return string(seq) -} diff --git a/internal/changefeed/changefeed_test.go b/internal/changefeed/changefeed_test.go deleted file mode 100644 index 21a47655..00000000 --- a/internal/changefeed/changefeed_test.go +++ /dev/null @@ -1,198 +0,0 @@ -package changefeed - -import ( - "context" - "fmt" - "net/http" - "net/http/httptest" - "slices" - "strconv" - "strings" - "sync" - "sync/atomic" - "testing" -) - -// fakePackagist serves a change feed the way packagist.org does: without -// since, 400 with the current timestamp; with an old since, a resync; with a -// known one, the actions recorded after it. -type fakePackagist struct { - mu sync.Mutex - now int64 - actions map[int64]string // since -> JSON actions - asked []string -} - -func (p *fakePackagist) ServeHTTP(w http.ResponseWriter, r *http.Request) { - p.mu.Lock() - defer p.mu.Unlock() - since := r.URL.Query().Get("since") - p.asked = append(p.asked, since) - w.Header().Set("Content-Type", "application/json") - if since == "" { - w.WriteHeader(http.StatusBadRequest) - _, _ = fmt.Fprintf(w, `{"error":"Invalid or missing since query parameter","timestamp":%d}`, p.now) - return - } - n, _ := strconv.ParseInt(since, 10, 64) - actions, ok := p.actions[n] - if !ok { - actions = `[{"type":"resync","package":"*","time":1}]` - } - _, _ = fmt.Fprintf(w, `{"actions":%s,"timestamp":%d}`, actions, p.now) -} - -func TestComposerFeed(t *testing.T) { - p := &fakePackagist{now: 100, actions: map[int64]string{ - 100: `[{"type":"update","package":"acme/lib","time":1},{"type":"update","package":"acme/lib~dev","time":1},{"type":"delete","package":"acme/gone","time":1}]`, - 200: `[]`, - }} - srv := httptest.NewServer(p) - defer srv.Close() - feed := NewComposerFeed(srv.Client(), ComposerFeedURL(srv.URL+"/")) - ctx := context.Background() - - if c, err := feed.Changes(ctx); err != nil || c.Names != nil || c.All { - t.Fatalf("first read = %+v, %v; want the start position only", c, err) - } - p.mu.Lock() - p.now = 200 - p.mu.Unlock() - c, err := feed.Changes(ctx) - if err != nil { - t.Fatal(err) - } - if want := []string{"acme/lib", "acme/lib~dev", "acme/gone", "acme/gone~dev"}; !slices.Equal(c.Names, want) || c.All { - t.Errorf("changes = %+v, want %v", c, want) - } - if c, err := feed.Changes(ctx); err != nil || len(c.Names) != 0 || c.All { - t.Errorf("changes without news = %+v, %v", c, err) - } - - // A position the feed no longer has history for gets a resync. - feed.since = 50 - if c, err := feed.Changes(ctx); err != nil || !c.All { - t.Errorf("changes after a resync = %+v, %v, want All", c, err) - } - if want := []string{"", "100", "200", "50"}; !slices.Equal(p.asked, want) { - t.Errorf("asked since %q, want %q", p.asked, want) - } -} - -func TestComposerFeedErrorKeepsPosition(t *testing.T) { - var fail atomic.Bool - fail.Store(true) - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if fail.Load() { - w.WriteHeader(http.StatusServiceUnavailable) - return - } - if r.URL.Query().Get("since") != "7" { - t.Errorf("since = %q, want 7", r.URL.Query().Get("since")) - } - _, _ = w.Write([]byte(`{"actions":[{"type":"update","package":"acme/lib","time":1}],"timestamp":8}`)) - })) - defer srv.Close() - feed := &ComposerFeed{client: srv.Client(), url: srv.URL, since: 7} - if _, err := feed.Changes(context.Background()); err == nil { - t.Fatal("want an error from a failing feed") - } - if feed.since != 7 { - t.Fatalf("since = %d after an error, want 7", feed.since) - } - fail.Store(false) - if c, err := feed.Changes(context.Background()); err != nil || !slices.Equal(c.Names, []string{"acme/lib"}) { - t.Errorf("changes after recovery = %+v, %v", c, err) - } -} - -// fakeReplicate serves a CouchDB-style changes feed over seqs 1..len(ids). -type fakeReplicate struct { - mu sync.Mutex - ids []string - limit int - asked []string -} - -func (f *fakeReplicate) ServeHTTP(w http.ResponseWriter, r *http.Request) { - f.mu.Lock() - defer f.mu.Unlock() - w.Header().Set("Content-Type", "application/json") - if r.URL.Path == "/registry/" { - _, _ = fmt.Fprintf(w, `{"db_name":"registry","update_seq":%d}`, len(f.ids)) - return - } - if r.URL.Path != "/registry/_changes" { - w.WriteHeader(http.StatusNotFound) - return - } - since, _ := strconv.Atoi(r.URL.Query().Get("since")) - limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) - f.asked = append(f.asked, r.URL.Query().Get("since")) - if f.limit > 0 { - limit = min(limit, f.limit) - } - var results []string - last := since - for seq := since + 1; seq <= len(f.ids) && len(results) < limit; seq++ { - results = append(results, fmt.Sprintf(`{"seq":%d,"id":%q,"changes":[{"rev":"1-a"}]}`, seq, f.ids[seq-1])) - last = seq - } - _, _ = fmt.Fprintf(w, `{"results":[%s],"last_seq":%d}`, strings.Join(results, ","), last) -} - -func TestNPMFeed(t *testing.T) { - f := &fakeReplicate{ids: []string{"old"}} - srv := httptest.NewServer(f) - defer srv.Close() - feed := NewNPMFeed(srv.Client(), srv.URL+"/registry/_changes") - ctx := context.Background() - - if c, err := feed.Changes(ctx); err != nil || c.Names != nil { - t.Fatalf("first read = %+v, %v; want the start position only", c, err) - } - f.mu.Lock() - f.ids = append(f.ids, "lodash", "@scope/pkg", "_design/app") - f.mu.Unlock() - c, err := feed.Changes(ctx) - if err != nil { - t.Fatal(err) - } - if want := []string{"lodash", "@scope/pkg"}; !slices.Equal(c.Names, want) { - t.Errorf("changes = %v, want %v", c.Names, want) - } - if c, err := feed.Changes(ctx); err != nil || len(c.Names) != 0 { - t.Errorf("changes without news = %+v, %v", c, err) - } - if want := []string{"1", "4"}; !slices.Equal(f.asked, want) { - t.Errorf("asked since %q, want %q", f.asked, want) - } -} - -// TestNPMFeedPages checks that a backlog longer than a page is read page by -// page, and that one read stops after npmMaxPages. -func TestNPMFeedPages(t *testing.T) { - f := &fakeReplicate{limit: npmPageLimit} - srv := httptest.NewServer(f) - defer srv.Close() - feed := &NPMFeed{client: srv.Client(), url: srv.URL + "/registry/_changes", since: []byte("0")} - f.ids = make([]string, npmPageLimit*(npmMaxPages+1)+5) - for i := range f.ids { - f.ids[i] = "pkg-" + strconv.Itoa(i) - } - - c, err := feed.Changes(context.Background()) - if err != nil { - t.Fatal(err) - } - if len(c.Names) != npmPageLimit*npmMaxPages { - t.Errorf("first read returned %d names, want %d", len(c.Names), npmPageLimit*npmMaxPages) - } - c, err = feed.Changes(context.Background()) - if err != nil { - t.Fatal(err) - } - if len(c.Names) != npmPageLimit+5 { - t.Errorf("second read returned %d names, want %d", len(c.Names), npmPageLimit+5) - } -} diff --git a/internal/server/change_feeds.go b/internal/server/change_feeds.go index 059bd2f7..556650f2 100644 --- a/internal/server/change_feeds.go +++ b/internal/server/change_feeds.go @@ -6,9 +6,10 @@ import ( "net/http" "time" - "github.com/git-pkgs/proxy/internal/changefeed" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/metrics" + "github.com/git-pkgs/registries/changes" + "github.com/git-pkgs/registries/client" ) // changeFeedInterval is how often an upstream change feed is read. @@ -20,13 +21,14 @@ const changeFeedMaxBackoff = 15 * time.Minute // startChangeFeeds follows the change feeds the configuration turns on. Each // instance sharing a database follows them itself: the reads are small, and // expiring a row twice does no harm. -func (s *Server) startChangeFeeds(ctx context.Context, client *http.Client, metadataTTL time.Duration) { - var feeds []changefeed.Feed +func (s *Server) startChangeFeeds(ctx context.Context, httpClient *http.Client, metadataTTL time.Duration) { + c := changes.WithClient(client.NewClient(client.WithHTTPClient(httpClient))) + var feeds []changes.Feed if s.cfg.Upstream.NPMChangeFeed { - feeds = append(feeds, changefeed.NewNPMFeed(client, changefeed.NPMFeedURL)) + feeds = append(feeds, changes.NewNPM(c)) } if s.cfg.Upstream.ComposerChangeFeed { - feeds = append(feeds, changefeed.NewComposerFeed(client, changefeed.ComposerFeedURL(s.cfg.Upstream.Composer))) + feeds = append(feeds, changes.NewPackagist(c, changes.WithURL(changes.PackagistFeedURL(s.cfg.Upstream.Composer)))) } if len(feeds) == 0 { return @@ -43,11 +45,15 @@ func (s *Server) startChangeFeeds(ctx context.Context, client *http.Client, meta // followChangeFeed reads feed every interval until ctx ends and expires the // cached metadata it lists. After a failed read it waits twice as long as -// before, up to changeFeedMaxBackoff. -func followChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, feed changefeed.Feed, interval time.Duration) { +// before, up to changeFeedMaxBackoff. The position in the feed is kept in +// memory: after a restart the feed is followed from its current position. +func followChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, feed changes.Feed, interval time.Duration) { wait := interval + var cursor changes.Cursor for { - if err := readChangeFeed(ctx, db, logger, feed); err != nil { + next, err := readChangeFeed(ctx, db, logger, feed, cursor) + cursor = next + if err != nil { if ctx.Err() != nil { return } @@ -66,29 +72,37 @@ func followChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, } } -// readChangeFeed reads feed once and expires the cached metadata of what it -// lists, so the next request for it revalidates upstream. Only rows fetched -// before the read finished are expired: a row fetched after it was fetched -// after every change the read returned. If expiring fails, the feed has -// already moved past the changes, and those rows wait out metadata_ttl. -func readChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, feed changefeed.Feed) error { - changes, err := feed.Changes(ctx) +// readChangeFeed reads feed once after cursor, or finds its current position +// when cursor is empty, and expires the cached metadata of what it lists, so +// the next request for it revalidates upstream. Only rows fetched before the +// read finished are expired: a row fetched after it was fetched after every +// change the read returned. It returns the cursor to read on from, which +// moves past the changes only once they are expired, so a failed expiry +// reads them again. +func readChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, feed changes.Feed, cursor changes.Cursor) (changes.Cursor, error) { + if cursor == "" { + return feed.Start(ctx) + } + found, next, err := feed.Changes(ctx, cursor) if err != nil { - return err + return cursor, err } cutoff := time.Now() ecosystem := feed.Ecosystem() var expired []string switch { - case changes.All: + case found.All: logger.Info("change feed: upstream asked for a full resync, revalidating all cached metadata", "ecosystem", ecosystem) expired, err = db.ExpireAllMetadataCache(ecosystem, cutoff) - case len(changes.Names) > 0: - expired, err = db.ExpireMetadataCache(ecosystem, changes.Names, cutoff) + case len(found.Names) > 0: + expired, err = db.ExpireMetadataCache(ecosystem, found.Names, cutoff) } for _, name := range expired { logger.Debug("change feed: cached metadata changed upstream", "ecosystem", ecosystem, "key", name) } metrics.RecordMetadataExpiredByChangeFeed(ecosystem, len(expired)) - return err + if err != nil { + return cursor, err + } + return next, nil } diff --git a/internal/server/change_feeds_test.go b/internal/server/change_feeds_test.go index 90441aca..366c1aed 100644 --- a/internal/server/change_feeds_test.go +++ b/internal/server/change_feeds_test.go @@ -9,17 +9,19 @@ import ( "log/slog" "net/http" "net/http/httptest" + "path/filepath" "strings" "sync" "sync/atomic" "testing" "time" - "github.com/git-pkgs/proxy/internal/changefeed" "github.com/git-pkgs/proxy/internal/config" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/handler" "github.com/git-pkgs/proxy/internal/storage" + "github.com/git-pkgs/registries/changes" + "github.com/git-pkgs/registries/client" "github.com/git-pkgs/registries/fetch" "github.com/go-chi/chi/v5" ) @@ -186,14 +188,18 @@ func TestChangeFeedRevalidatesChangedMetadata(t *testing.T) { reg.set("/npm/left-pad", npmDoc("1.0.0")) h, db := changeFeedRouter(t, reg) logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - feeds := []changefeed.Feed{ - changefeed.NewComposerFeed(reg.Client(), changefeed.ComposerFeedURL(reg.URL)), - changefeed.NewNPMFeed(reg.Client(), reg.URL+"/replicate/registry/_changes"), + c := changes.WithClient(client.NewClient(client.WithHTTPClient(reg.Client()), client.WithMaxRetries(0))) + feeds := []changes.Feed{ + changes.NewPackagist(c, changes.WithURL(changes.PackagistFeedURL(reg.URL))), + changes.NewNPM(c, changes.WithURL(reg.URL+"/replicate/registry/_changes")), } - for _, feed := range feeds { - if err := readChangeFeed(context.Background(), db, logger, feed); err != nil { - t.Fatalf("starting %s feed: %v", feed.Ecosystem(), err) + cursors := make([]changes.Cursor, len(feeds)) + for i, feed := range feeds { + cursor, err := readChangeFeed(context.Background(), db, logger, feed, "") + if err != nil || cursor == "" { + t.Fatalf("starting %s feed: cursor %q, %v", feed.Ecosystem(), cursor, err) } + cursors[i] = cursor } getBody(t, h, "/composer/p2/acme/lib.json") @@ -209,9 +215,10 @@ func TestChangeFeedRevalidatesChangedMetadata(t *testing.T) { t.Fatal("npm packument revalidated before the feed listed it; the test proves nothing") } - for _, feed := range feeds { - if err := readChangeFeed(context.Background(), db, logger, feed); err != nil { - t.Fatalf("reading %s feed: %v", feed.Ecosystem(), err) + for i, feed := range feeds { + next, err := readChangeFeed(context.Background(), db, logger, feed, cursors[i]) + if err != nil || next == cursors[i] { + t.Fatalf("reading %s feed: cursor %q -> %q, %v", feed.Ecosystem(), cursors[i], next, err) } } @@ -238,11 +245,51 @@ type countingFeed struct { func (f *countingFeed) Ecosystem() string { return "npm" } -func (f *countingFeed) Changes(context.Context) (changefeed.Changes, error) { +func (f *countingFeed) Start(context.Context) (changes.Cursor, error) { + return f.read() +} + +func (f *countingFeed) Changes(_ context.Context, cursor changes.Cursor) (changes.Changes, changes.Cursor, error) { + if _, err := f.read(); err != nil { + return changes.Changes{}, cursor, err + } + return changes.Changes{}, cursor, nil +} + +func (f *countingFeed) read() (changes.Cursor, error) { if f.reads.Add(1) <= f.failures { - return changefeed.Changes{}, errors.New("feed down") + return "", errors.New("feed down") + } + return "1", nil +} + +// fixedFeed reports the same names on every read. +type fixedFeed struct{ names []string } + +func (fixedFeed) Ecosystem() string { return "npm" } +func (fixedFeed) Start(context.Context) (changes.Cursor, error) { return "1", nil } +func (f fixedFeed) Changes(context.Context, changes.Cursor) (changes.Changes, changes.Cursor, error) { + return changes.Changes{Names: f.names}, "2", nil +} + +// TestReadChangeFeedKeepsCursorWhenExpiryFails checks that the cursor moves +// past the changes only once they are expired, so changes whose expiry +// failed are read again. +func TestReadChangeFeedKeepsCursorWhenExpiryFails(t *testing.T) { + db, err := database.Create(filepath.Join(t.TempDir(), "proxy.db")) + if err != nil { + t.Fatal(err) + } + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + feed := fixedFeed{names: []string{"left-pad"}} + + if next, err := readChangeFeed(context.Background(), db, logger, feed, "1"); err != nil || next != "2" { + t.Fatalf("read = %q, %v; want cursor 2", next, err) + } + _ = db.Close() + if next, err := readChangeFeed(context.Background(), db, logger, feed, "1"); err == nil || next != "1" { + t.Errorf("read with a failing expiry = %q, %v; want an error and cursor 1", next, err) } - return changefeed.Changes{}, nil } // TestFollowChangeFeedStopsWithContext checks that a failing feed is retried diff --git a/internal/server/server.go b/internal/server/server.go index a5c7b075..574ce3a7 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -234,6 +234,33 @@ func (s *Server) Start(listeners ...net.Listener) error { return s.serve(listener) } +// configureGitCache sets up the cache of upstream git repositories, when any +// are configured. +func (s *Server) configureGitCache(client *http.Client) error { + if len(s.cfg.Upstream.Git) == 0 { + return nil + } + upstreamDirs := make(map[string]string, len(s.cfg.Upstream.Git)) + for name, upstream := range s.cfg.Upstream.Git { + upstreamDirs[name] = gitcache.UpstreamDir(name, upstream.URL) + } + gitCache, err := gitcache.New(gitcache.Options{ + CacheDir: s.cfg.Git.CacheDir, + MaxSize: s.cfg.Git.MaxSizeBytes(), + MaxRepoSize: s.cfg.Git.MaxRepoSizeBytes(), + RefreshWait: s.cfg.Git.RefreshWaitDuration(), + StaleMaxAge: s.cfg.Git.StaleMaxAgeDuration(), + Client: client, + Logger: s.logger, + Upstreams: upstreamDirs, + }) + if err != nil { + return fmt.Errorf("configuring git cache: %w", err) + } + s.git = gitCache + return nil +} + func (s *Server) serve(listener net.Listener) error { metadataClient, artifactClient := s.newUpstreamClients(s.cfg.ParseArtifactTimeout()) @@ -283,25 +310,8 @@ func (s *Server) serve(listener net.Listener) error { proxy.DirectServeBaseURL = s.cfg.Storage.DirectServeBaseURL proxy.StreamArtifacts = !s.cfg.Storage.CacheArtifacts - if len(s.cfg.Upstream.Git) > 0 { - upstreamDirs := make(map[string]string, len(s.cfg.Upstream.Git)) - for name, upstream := range s.cfg.Upstream.Git { - upstreamDirs[name] = gitcache.UpstreamDir(name, upstream.URL) - } - gitCache, err := gitcache.New(gitcache.Options{ - CacheDir: s.cfg.Git.CacheDir, - MaxSize: s.cfg.Git.MaxSizeBytes(), - MaxRepoSize: s.cfg.Git.MaxRepoSizeBytes(), - RefreshWait: s.cfg.Git.RefreshWaitDuration(), - StaleMaxAge: s.cfg.Git.StaleMaxAgeDuration(), - Client: artifactClient, - Logger: s.logger, - Upstreams: upstreamDirs, - }) - if err != nil { - return fmt.Errorf("configuring git cache: %w", err) - } - s.git = gitCache + if err := s.configureGitCache(artifactClient); err != nil { + return err } // Create router with Chi From ba95de17f30d9068b2fa61b8dad8eb10a76495ad Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Fri, 30 Oct 2026 14:41:45 +0000 Subject: [PATCH 3/4] Handle Swift registry identities in cache and vulnerability APIs --- go.mod | 2 +- go.sum | 4 +- internal/enrichment/enrichment.go | 9 ++-- internal/enrichment/enrichment_test.go | 11 +++-- internal/packageurl/packageurl.go | 24 ++++------- internal/packageurl/packageurl_test.go | 25 +++++------ internal/server/api.go | 9 ++++ internal/server/swift_vulns_test.go | 57 ++++++++++++++++++++++++++ 8 files changed, 102 insertions(+), 39 deletions(-) create mode 100644 internal/server/swift_vulns_test.go diff --git a/go.mod b/go.mod index 4a164c29..79978b17 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/git-pkgs/archives v0.10.0 github.com/git-pkgs/artifacts v0.2.1 github.com/git-pkgs/cooldown v0.3.0 - github.com/git-pkgs/enrichment v0.7.4 + github.com/git-pkgs/enrichment v0.7.5 github.com/git-pkgs/gcs v0.1.0 github.com/git-pkgs/integrity v0.2.0 github.com/git-pkgs/magic v0.6.0 diff --git a/go.sum b/go.sum index 983c87e1..7ed7c297 100644 --- a/go.sum +++ b/go.sum @@ -123,8 +123,8 @@ github.com/git-pkgs/artifacts v0.2.1 h1:VwdxR4yTDaqBZ34h0slxQBVyr2Xfa+QGOKCKviMx github.com/git-pkgs/artifacts v0.2.1/go.mod h1:Otosgq52pXT5UNN7lh6s/lszpWKVDO8XrOjN1M70/IA= github.com/git-pkgs/cooldown v0.3.0 h1:eAl2CQFwqXxluoVP9hHmropAf1VoV3gSxLMkN+NEu8A= github.com/git-pkgs/cooldown v0.3.0/go.mod h1:FnTg8wufEALATC9Sm/9AHjLdA+u9eVn96e9BqXXjGB0= -github.com/git-pkgs/enrichment v0.7.4 h1:owiCKjITG8tNxBikcyUCDuDf1i3C/+Yoa5yJseuqIjA= -github.com/git-pkgs/enrichment v0.7.4/go.mod h1:ZN/cH4PnmzsapEVPvmAwREP7+ox+RX0yfR6KvoB8xKE= +github.com/git-pkgs/enrichment v0.7.5 h1:jbfFzf7+HSS7el8AuMri4vHXq2izIIZDVIU0uYI5LFM= +github.com/git-pkgs/enrichment v0.7.5/go.mod h1:TbxC/nGQN96YCbdoiDs8a9z3gk2zXmLVS1989/jvnsU= github.com/git-pkgs/gcs v0.1.0 h1:E3awGtsO0xZyHT9FUfEwMHjMkRxw41Bh+c7lgTmPvBo= github.com/git-pkgs/gcs v0.1.0/go.mod h1:bdkCFD66ryaWnU8MBhokVA3WkJfyAEchm9qec5woBpE= github.com/git-pkgs/integrity v0.2.0 h1:sxpAk7EUllZnTYttOoPTW/Yfe5J2imglpuJiuBUZOxM= diff --git a/internal/enrichment/enrichment.go b/internal/enrichment/enrichment.go index 4d847af5..2246e394 100644 --- a/internal/enrichment/enrichment.go +++ b/internal/enrichment/enrichment.go @@ -100,9 +100,6 @@ func (s *Service) BulkEnrichPackages(ctx context.Context, packages []struct{ Eco // CheckVulnerabilities queries for vulnerabilities affecting a package version. func (s *Service) CheckVulnerabilities(ctx context.Context, ecosystem, name, version string) ([]VulnInfo, error) { - if packageurl.Make(ecosystem, name, version) == nil { - return nil, nil - } return s.vulns.Check(ctx, ecosystem, name, version) } @@ -137,7 +134,8 @@ type EnrichmentResult struct { } // EnrichFull fetches package, version and vulnerability data for a package -// version concurrently. Failed lookups are logged and left empty. +// version concurrently. Unsupported vulnerability identities return an error; +// other failed lookups are logged and left empty. func (s *Service) EnrichFull(ctx context.Context, ecosystem, name, version string) (*EnrichmentResult, error) { result := &EnrichmentResult{} @@ -147,6 +145,9 @@ func (s *Service) EnrichFull(ctx context.Context, ecosystem, name, version strin wg.Go(func() { result.Version, verErr = s.EnrichVersion(ctx, ecosystem, name, version) }) wg.Go(func() { result.Vulnerabilities, vulnErr = s.CheckVulnerabilities(ctx, ecosystem, name, version) }) wg.Wait() + if errors.Is(vulnErr, shared.ErrUnsupportedVulnerabilityIdentity) { + return nil, vulnErr + } if pkgErr != nil { s.logger.Debug("failed to enrich package", "ecosystem", ecosystem, "name", name, "error", pkgErr) diff --git a/internal/enrichment/enrichment_test.go b/internal/enrichment/enrichment_test.go index a854e5a8..f01bb291 100644 --- a/internal/enrichment/enrichment_test.go +++ b/internal/enrichment/enrichment_test.go @@ -2,10 +2,12 @@ package enrichment import ( "context" + "errors" "log/slog" "os" "testing" + shared "github.com/git-pkgs/enrichment" "github.com/git-pkgs/registries" ) @@ -26,7 +28,7 @@ func TestNew(t *testing.T) { } } -func TestSwiftRegistryIdentitySkipsPURLDependentLookups(t *testing.T) { +func TestSwiftRegistryIdentityLookups(t *testing.T) { svc := New(slog.New(slog.NewTextHandler(os.Stdout, nil))) ctx := context.Background() @@ -41,8 +43,11 @@ func TestSwiftRegistryIdentitySkipsPURLDependentLookups(t *testing.T) { } vulnerabilities, err := svc.CheckVulnerabilities(ctx, "swift", "apple/example", "1.2.3") - if err != nil || vulnerabilities != nil { - t.Errorf("CheckVulnerabilities() = %#v, %v; want nil, nil", vulnerabilities, err) + if !errors.Is(err, shared.ErrUnsupportedVulnerabilityIdentity) || vulnerabilities != nil { + t.Errorf("CheckVulnerabilities() = %#v, %v; want unsupported identity", vulnerabilities, err) + } + if _, err := svc.EnrichFull(ctx, "swift", "apple/example", "1.2.3"); !errors.Is(err, shared.ErrUnsupportedVulnerabilityIdentity) { + t.Errorf("EnrichFull error = %v, want unsupported identity", err) } latest, err := svc.GetLatestVersion(ctx, "swift", "apple/example") diff --git a/internal/packageurl/packageurl.go b/internal/packageurl/packageurl.go index 71109426..be0e9ab7 100644 --- a/internal/packageurl/packageurl.go +++ b/internal/packageurl/packageurl.go @@ -29,25 +29,19 @@ func WithVersionString(packagePURL, version string) string { } // MakeCacheStrings returns package and version PURLs suitable for artifact -// cache records. Swift registry identities use an explicit generic PURL until -// their source repository has been resolved. The result is independent of the -// configured upstream so cache entries survive an upstream.swift change, -// matching every other ecosystem. +// cache records. Swift registry identities keep their generic storage PURLs. +// Cache keys are independent of the configured upstream. func MakeCacheStrings(ecosystem, name, version string) (packagePURL, versionPURL string) { + if purl.NormalizeEcosystem(ecosystem) == "swift" { + if identity, ok := swiftRegistryIdentity(name); ok { + pkg := purl.New("generic", "swift-registry", identity, "", nil) + return pkg.String(), pkg.WithVersion(version).String() + } + } if pkg := Make(ecosystem, name, ""); pkg != nil { return pkg.String(), pkg.WithVersion(version).String() } - if purl.NormalizeEcosystem(ecosystem) != "swift" { - return "", "" - } - - identity, ok := swiftRegistryIdentity(name) - if !ok { - return "", "" - } - - pkg := purl.New("generic", "swift-registry", identity, "", nil) - return pkg.String(), pkg.WithVersion(version).String() + return "", "" } func swiftRegistryIdentity(name string) (string, bool) { diff --git a/internal/packageurl/packageurl_test.go b/internal/packageurl/packageurl_test.go index 8982288c..5a46776c 100644 --- a/internal/packageurl/packageurl_test.go +++ b/internal/packageurl/packageurl_test.go @@ -2,17 +2,12 @@ package packageurl import "testing" -func TestMakeSwiftRegistryIdentityUnsupported(t *testing.T) { - identities := []string{"apple.swift-argument-parser", "apple/swift-argument-parser"} - for _, identity := range identities { - t.Run(identity, func(t *testing.T) { - if got := Make("swift", identity, "1.8.2"); got != nil { - t.Errorf("Make() = %q, want nil", got.String()) - } - if got := MakeString("swift", identity, "1.8.2"); got != "" { - t.Errorf("MakeString() = %q, want empty string", got) - } - }) +func TestMakeSwiftRegistryPathUnsupported(t *testing.T) { + if got := Make("swift", "apple/swift-argument-parser", "1.8.2"); got != nil { + t.Errorf("Make() = %q, want nil", got.String()) + } + if got := MakeString("swift", "apple/swift-argument-parser", "1.8.2"); got != "" { + t.Errorf("MakeString() = %q, want empty string", got) } } @@ -49,9 +44,11 @@ func TestMakeCacheStringsSwiftRegistryIdentity(t *testing.T) { t.Errorf("version PURL = %q, want %q", versionPURL, wantVersion) } - dottedPackage, dottedVersion := MakeCacheStrings("swift", "apple.example", "1.2.3") - if dottedPackage != packagePURL || dottedVersion != versionPURL { - t.Errorf("dotted identity cache PURLs = %q, %q; want %q, %q", dottedPackage, dottedVersion, packagePURL, versionPURL) + for _, identity := range []string{"apple.example", "APPLE.EXAMPLE", "apple/example"} { + dottedPackage, dottedVersion := MakeCacheStrings("Swift", identity, "1.2.3") + if dottedPackage != packagePURL || dottedVersion != versionPURL { + t.Errorf("%q cache PURLs = %q, %q; want %q, %q", identity, dottedPackage, dottedVersion, packagePURL, versionPURL) + } } } diff --git a/internal/server/api.go b/internal/server/api.go index b2043672..22bfb8a1 100644 --- a/internal/server/api.go +++ b/internal/server/api.go @@ -2,6 +2,7 @@ package server import ( "encoding/json" + "errors" "net/http" "strings" @@ -219,6 +220,10 @@ func (h *APIHandler) getPackage(w http.ResponseWriter, r *http.Request, ecosyste func (h *APIHandler) getVersion(w http.ResponseWriter, r *http.Request, ecosystem, name, version string) { result, err := h.enrichment.EnrichFull(r.Context(), ecosystem, name, version) + if errors.Is(err, shared.ErrUnsupportedVulnerabilityIdentity) { + badRequest(w, "package identity is not supported for vulnerability lookups") + return + } if err != nil { writeError(w, http.StatusBadGateway, ErrCodeUpstream, "failed to enrich version") return @@ -303,6 +308,10 @@ func (h *APIHandler) HandleVulnsPath(w http.ResponseWriter, r *http.Request) { } vulns, err := h.enrichment.CheckVulnerabilities(r.Context(), ecosystem, name, version) + if errors.Is(err, shared.ErrUnsupportedVulnerabilityIdentity) { + badRequest(w, "package identity is not supported for vulnerability lookups") + return + } if err != nil { writeError(w, http.StatusBadGateway, ErrCodeUpstream, "failed to check vulnerabilities") return diff --git a/internal/server/swift_vulns_test.go b/internal/server/swift_vulns_test.go new file mode 100644 index 00000000..f28caaf3 --- /dev/null +++ b/internal/server/swift_vulns_test.go @@ -0,0 +1,57 @@ +package server + +import ( + "encoding/json" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/git-pkgs/proxy/internal/enrichment" + "github.com/go-chi/chi/v5" +) + +func TestSwiftRegistryVulnerabilityErrors(t *testing.T) { + h := NewAPIHandler(enrichment.New(slog.Default()), nil) + r := chi.NewRouter() + r.Get("/api/vulns/{ecosystem}/*", h.HandleVulnsPath) + r.Get("/api/package/{ecosystem}/*", h.HandlePackagePath) + transport := &swiftVulnsTransport{} + original := http.DefaultTransport + http.DefaultTransport = transport + t.Cleanup(func() { http.DefaultTransport = original }) + + for _, path := range []string{ + "/api/vulns/swift/apple.swift-nio", + "/api/vulns/swift/apple.swift-nio/2.0.0", + "/api/vulns/Swift/Apple.Swift-NIO/2.0.0", + "/api/package/swift/apple.swift-nio/2.0.0", + "/api/package/Swift/Apple.Swift-NIO/2.0.0", + } { + t.Run(path, func(t *testing.T) { + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil)) + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, body = %s", w.Code, w.Body) + } + var response ErrorResponse + if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Code != ErrCodeBadRequest || response.Message != "package identity is not supported for vulnerability lookups" { + t.Fatalf("unexpected response: %+v", response) + } + }) + } + if transport.requests != 0 { + t.Errorf("made %d upstream requests for unsupported identities", transport.requests) + } +} + +type swiftVulnsTransport struct{ requests int } + +func (s *swiftVulnsTransport) RoundTrip(r *http.Request) (*http.Response, error) { + s.requests++ + return nil, fmt.Errorf("unexpected upstream request: %s", r.URL) +} From f32b46b71ae1f1f9cfd92e21b2a9bfb3deaa72d2 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Fri, 30 Oct 2026 18:02:30 +0000 Subject: [PATCH 4/4] Preserve Go cache and policy behavior when upgrading PURL --- docs/configuration.md | 3 ++ go.mod | 4 +- go.sum | 8 +-- internal/config/config.go | 3 +- internal/database/rekey_artifact.go | 40 ++++++++++++++ internal/database/rekey_artifact_test.go | 69 ++++++++++++++++++++++++ internal/denylist/denylist.go | 3 +- internal/handler/cooldown_expiry.go | 7 ++- internal/handler/cooldown_fail_closed.go | 2 +- internal/handler/denylist_test.go | 6 +++ internal/handler/go.go | 60 +++++++++++++++------ internal/handler/go_case_test.go | 69 ++++++++++++++++++++++++ internal/handler/go_cooldown.go | 8 +-- internal/handler/go_policy_test.go | 47 ++++++++++++++++ internal/handler/handler.go | 6 +++ internal/packageurl/packageurl.go | 9 ++++ internal/retention/retention.go | 11 ++-- internal/server/retention_golang_test.go | 4 +- 18 files changed, 320 insertions(+), 39 deletions(-) create mode 100644 internal/database/rekey_artifact.go create mode 100644 internal/database/rekey_artifact_test.go create mode 100644 internal/handler/go_policy_test.go diff --git a/docs/configuration.md b/docs/configuration.md index e1a4f9ef..b20685f0 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1119,6 +1119,8 @@ Durations support days (`7d`), hours (`48h`), and minutes (`30m`). Set to `0` to Package PURL keys are normalized to canonical form before matching, so `pkg:npm/@babel/core` and `pkg:npm/%40babel/core` are equivalent, as are `pkg:pypi/Django` and `pkg:pypi/django`. If both forms configure the same package, the canonical entry wins. +Go module names in exact cooldown overrides are matched case-insensitively. If both lowercase and mixed-case keys are configured, the lowercase entry wins. Go package patterns match lowercase module names. + `package_patterns` matches canonical, versionless PURLs using `*` for zero or more characters and `?` for one character. Neither wildcard crosses `/` separators. Character classes (`[...]`) and backslash escapes are rejected at startup. For example, `"pkg:npm/@example/*"` matches packages under the `@example` npm scope. Patterns accept `@` as an alias for `%40`; other characters must use their canonical PURL form. Equivalent patterns with different durations are rejected at startup. Equal durations, such as `1d` and `24h`, are accepted. Exact `packages` entries take precedence over patterns. When several patterns match, longer patterns win after excluding `*` and `?` from the length. Ties use lexical order of the normalized patterns. @@ -1180,6 +1182,7 @@ Names and versions are compared the way each ecosystem treats them: `pkg:composer/symfony/console@6.1.7` also covers `v6.1.7`. - Go versions are matched as written, `pkg:golang/example.com/mod@v1.0.0-RC1`, though download URLs escape uppercase letters (`v1.0.0-!r!c1`). + Go module names in denylist entries are matched case-insensitively. - A Swift entry names the registry scope and package, ignoring case: `pkg:swift/apple/swift-nio@2.62.0`. The stored form, `pkg:generic/swift-registry/apple.swift-nio@2.62.0`, works too. diff --git a/go.mod b/go.mod index 79978b17..2ce65e22 100644 --- a/go.mod +++ b/go.mod @@ -16,8 +16,8 @@ require ( github.com/git-pkgs/gcs v0.1.0 github.com/git-pkgs/integrity v0.2.0 github.com/git-pkgs/magic v0.6.0 - github.com/git-pkgs/purl v0.1.24 - github.com/git-pkgs/registries v0.12.0 + github.com/git-pkgs/purl v0.2.1 + github.com/git-pkgs/registries v0.13.0 github.com/git-pkgs/sbom v0.1.7 github.com/git-pkgs/spdx v0.3.3 github.com/go-chi/chi/v5 v5.5.0 diff --git a/go.sum b/go.sum index 7ed7c297..c3221612 100644 --- a/go.sum +++ b/go.sum @@ -133,10 +133,10 @@ github.com/git-pkgs/magic v0.6.0 h1:fINPe67FEIaFRVlZzhOIDClMkm+gftBroBENa8bHSxQ= github.com/git-pkgs/magic v0.6.0/go.mod h1:SXOqcsNmbmpZjJZHEEWnwxprbsFvpwWgTAZrgXp4Jm4= github.com/git-pkgs/pom v0.1.8 h1:mXCuWw6XkrKKZb6IurRZyIznYLPhV5AYg5ZO38Uj/l8= github.com/git-pkgs/pom v0.1.8/go.mod h1:sQBeYRzjCUn77fcABPn6I/a1aBHFxm55jTmkNxYtGtw= -github.com/git-pkgs/purl v0.1.24 h1:JMEB1uNWI+baEpl6AeVx+Kfq1c+xkou+THxa5KJ3MrU= -github.com/git-pkgs/purl v0.1.24/go.mod h1:l8rCfvt5iXVrFIcoQ8+vCLOaHhrtEW2id0LENP+IzMw= -github.com/git-pkgs/registries v0.12.0 h1:fhxYaR5iR9Zlx6doLkjlXurVqPxkYcXmNFiIrFeV3/k= -github.com/git-pkgs/registries v0.12.0/go.mod h1:bKxgdy/PwnzQXL2FnOktmSQHTKk6B85MA4VtvvYItPc= +github.com/git-pkgs/purl v0.2.1 h1:b46hRlOMbrG6aXjjFJdYixGi5WTh0TLlAnH9iNeuOS0= +github.com/git-pkgs/purl v0.2.1/go.mod h1:l8rCfvt5iXVrFIcoQ8+vCLOaHhrtEW2id0LENP+IzMw= +github.com/git-pkgs/registries v0.13.0 h1:5i9dHnbiXseIbXLIZ61YnuFySJ9dQX+KEu1MrmwB2BU= +github.com/git-pkgs/registries v0.13.0/go.mod h1:1g/jXVi1M/3jWvW1Wom1NMnPrhglJ7UVEZ215l2Zh0M= github.com/git-pkgs/sbom v0.1.7 h1:4eboy/Yed/SC9ZcC/xT50iywHPzWbmCU7azDb5T1RbA= github.com/git-pkgs/sbom v0.1.7/go.mod h1:KIkZX0v31l62SeFsi2qsQXcZRVrAXvIdPQ4x4en+G2M= github.com/git-pkgs/spdx v0.3.3 h1:xXlUoZzjDSkwxC6mWHfaW+kc8a0+LH6d+gGHuZWftqI= diff --git a/internal/config/config.go b/internal/config/config.go index 5e901505..c09a1bae 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -77,6 +77,7 @@ import ( "github.com/git-pkgs/cooldown" "github.com/git-pkgs/proxy/internal/denylist" + "github.com/git-pkgs/proxy/internal/packageurl" "github.com/git-pkgs/proxy/internal/retention" "github.com/git-pkgs/purl" "gopkg.in/yaml.v3" @@ -261,7 +262,7 @@ func (c *CooldownConfig) NormalizedPackages() map[string]string { for _, key := range keys { canonical := key if parsed, err := purl.Parse(key); err == nil { - canonical = parsed.String() + canonical = packageurl.PolicyPackageKey(parsed.String()) } if _, exists := normalized[canonical]; exists && key != canonical { continue diff --git a/internal/database/rekey_artifact.go b/internal/database/rekey_artifact.go new file mode 100644 index 00000000..68a8dbd4 --- /dev/null +++ b/internal/database/rekey_artifact.go @@ -0,0 +1,40 @@ +package database + +import ( + "fmt" + "time" +) + +// RekeyCachedArtifact moves one unchanged upstream record without replacing +// an existing destination. Access history and stored bytes are preserved. +func (db *DB) RekeyCachedArtifact(old *Artifact, pkg *Package, version *Version) (bool, error) { + tx, err := db.Beginx() + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback() }() + now := time.Now() + _, err = tx.Exec(db.Rebind(`INSERT INTO packages (purl, ecosystem, name, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) ON CONFLICT(purl) DO NOTHING`), pkg.PURL, pkg.Ecosystem, pkg.Name, now, now) + if err != nil { + return false, fmt.Errorf("creating artifact package: %w", err) + } + _, err = tx.Exec(db.Rebind(`INSERT INTO versions (purl, package_purl, created_at, updated_at) + VALUES (?, ?, ?, ?) ON CONFLICT(purl) DO NOTHING`), version.PURL, version.PackagePURL, now, now) + if err != nil { + return false, fmt.Errorf("creating artifact version: %w", err) + } + result, err := tx.Exec(db.Rebind(`UPDATE artifacts SET version_purl = ?, updated_at = ? + WHERE id = ? AND version_purl = ? AND storage_path = ? AND upstream_url = ? AND source = ? + AND NOT EXISTS (SELECT 1 FROM artifacts WHERE version_purl = ? AND filename = ?)`), + version.PURL, now, old.ID, old.VersionPURL, old.StoragePath, old.UpstreamURL, SourceUpstream, + version.PURL, old.Filename) + if err != nil { + return false, fmt.Errorf("moving cached artifact: %w", err) + } + count, err := result.RowsAffected() + if err != nil || count == 0 { + return false, err + } + return true, tx.Commit() +} diff --git a/internal/database/rekey_artifact_test.go b/internal/database/rekey_artifact_test.go new file mode 100644 index 00000000..d5fd2704 --- /dev/null +++ b/internal/database/rekey_artifact_test.go @@ -0,0 +1,69 @@ +package database + +import ( + "database/sql" + "testing" +) + +func TestRekeyCachedArtifact(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + for _, scenario := range []string{"move", "destination exists", "source changed", "local source"} { + t.Run(scenario, func(t *testing.T) { + oldPackage := &Package{PURL: "pkg:golang/example.com/old", Ecosystem: "golang", Name: "example.com/Old"} + oldVersion := &Version{PURL: oldPackage.PURL + "@v1.0.0", PackagePURL: oldPackage.PURL} + old := &Artifact{VersionPURL: oldVersion.PURL, Filename: scenario + ".zip", + UpstreamURL: "https://proxy.example/example.com/!old/@v/v1.0.0.zip", + StoragePath: sql.NullString{String: scenario + "/old.zip", Valid: true}} + if err := db.RecordCachedArtifact(oldPackage, oldVersion, old); err != nil { + t.Fatal(err) + } + old, err := db.GetArtifact(oldVersion.PURL, old.Filename) + if err != nil || old == nil { + t.Fatalf("source artifact = %+v, %v", old, err) + } + pkg := &Package{PURL: "pkg:golang/example.com/Old", Ecosystem: "golang", Name: "example.com/Old"} + version := &Version{PURL: pkg.PURL + "@v1.0.0", PackagePURL: pkg.PURL} + prepareRekeyConflict(t, db, scenario, old, pkg, version) + moved, err := db.RekeyCachedArtifact(old, pkg, version) + if err != nil || moved != (scenario == "move") { + t.Fatalf("RekeyCachedArtifact = %v, %v", moved, err) + } + checkRekeyResult(t, db, scenario, old, version) + }) + } + }) +} + +func checkRekeyResult(t *testing.T, db *DB, scenario string, old *Artifact, version *Version) { + t.Helper() + current, err := db.GetArtifact(version.PURL, old.Filename) + if err != nil { + t.Fatal(err) + } + if scenario == "move" && (current == nil || current.ID != old.ID || current.StoragePath != old.StoragePath) { + t.Errorf("moved artifact = %+v, old = %+v", current, old) + } + if scenario == "destination exists" && (current == nil || current.StoragePath.String != "destination.zip") { + t.Errorf("destination replaced: %+v", current) + } + if paths := allPendingDeletes(t, db); len(paths) != 0 { + t.Errorf("rekey queued stored bytes for deletion: %v", paths) + } +} + +func prepareRekeyConflict(t *testing.T, db *DB, scenario string, old *Artifact, pkg *Package, version *Version) { + t.Helper() + var err error + switch scenario { + case "destination exists": + err = db.RecordCachedArtifact(pkg, version, &Artifact{VersionPURL: version.PURL, Filename: old.Filename, + StoragePath: sql.NullString{String: "destination.zip", Valid: true}}) + case "source changed": + _, err = db.Exec(db.Rebind(`UPDATE artifacts SET storage_path = ? WHERE id = ?`), "replacement.zip", old.ID) + case "local source": + _, err = db.Exec(db.Rebind(`UPDATE artifacts SET source = ? WHERE id = ?`), SourceLocal, old.ID) + } + if err != nil { + t.Fatal(err) + } +} diff --git a/internal/denylist/denylist.go b/internal/denylist/denylist.go index 15bb9e5e..76f2de5f 100644 --- a/internal/denylist/denylist.go +++ b/internal/denylist/denylist.go @@ -5,6 +5,7 @@ import ( "fmt" "strings" + "github.com/git-pkgs/proxy/internal/packageurl" "github.com/git-pkgs/purl" ) @@ -138,5 +139,5 @@ func identify(pkg *purl.PURL) identity { channel = c.Qualifier("channel") } key := purl.New(c.Type, c.Namespace, c.Name, "", nil).String() - return identity{key: key, channel: channel, version: c.Version} + return identity{key: packageurl.PolicyPackageKey(key), channel: channel, version: c.Version} } diff --git a/internal/handler/cooldown_expiry.go b/internal/handler/cooldown_expiry.go index 1163bde9..c31ace8e 100644 --- a/internal/handler/cooldown_expiry.go +++ b/internal/handler/cooldown_expiry.go @@ -5,6 +5,7 @@ import ( "time" "github.com/git-pkgs/cooldown" + "github.com/git-pkgs/proxy/internal/packageurl" ) // A rewrite made with cooldown on stays right until a version it held back @@ -127,6 +128,7 @@ type recordingCooldown struct { } func (r recordingCooldown) IsAllowed(ecosystem, packagePURL string, publishedAt time.Time) bool { + packagePURL = packageurl.PolicyPackageKey(packagePURL) if r.p.Cooldown.IsAllowed(ecosystem, packagePURL, publishedAt) { return true } @@ -135,6 +137,7 @@ func (r recordingCooldown) IsAllowed(ecosystem, packagePURL string, publishedAt } func (r recordingCooldown) Evaluate(ecosystem, packagePURL string, publishedAt, evaluatedAt time.Time) cooldown.Decision { + packagePURL = packageurl.PolicyPackageKey(packagePURL) d := r.p.Cooldown.Evaluate(ecosystem, packagePURL, publishedAt, evaluatedAt) if !d.Allowed { r.p.withheld(d.AvailableAt) @@ -143,7 +146,7 @@ func (r recordingCooldown) Evaluate(ecosystem, packagePURL string, publishedAt, } func (r recordingCooldown) For(ecosystem, packagePURL string) time.Duration { - return r.p.Cooldown.For(ecosystem, packagePURL) + return r.p.Cooldown.For(ecosystem, packageurl.PolicyPackageKey(packagePURL)) } func (r recordingCooldown) Enabled() bool { @@ -156,7 +159,7 @@ func (r recordingCooldown) availableAt(ecosystem, packagePURL string, publishedA if publishedAt.IsZero() { return time.Time{} } - d := r.p.Cooldown.For(ecosystem, packagePURL) + d := r.For(ecosystem, packagePURL) if d <= 0 { return time.Time{} } diff --git a/internal/handler/cooldown_fail_closed.go b/internal/handler/cooldown_fail_closed.go index da600c1e..0dd2c360 100644 --- a/internal/handler/cooldown_fail_closed.go +++ b/internal/handler/cooldown_fail_closed.go @@ -5,7 +5,7 @@ import "net/http" // cooldownApplies reports whether cooldown withholds new versions of the // package, taking package overrides and exemptions into account. func (p *Proxy) cooldownApplies(ecosystem, name string) bool { - return p.cooldownEnabled() && p.Cooldown.For(ecosystem, canonicalPackagePURL(ecosystem, name)) > 0 + return p.cooldownEnabled() && p.cooldown().For(ecosystem, canonicalPackagePURL(ecosystem, name)) > 0 } // mustFilterMetadata reports whether the package's metadata may have diff --git a/internal/handler/denylist_test.go b/internal/handler/denylist_test.go index bb41e38a..663d0c3d 100644 --- a/internal/handler/denylist_test.go +++ b/internal/handler/denylist_test.go @@ -288,6 +288,12 @@ func TestDenylistMatchesEcosystemSpellings(t *testing.T) { {"go case-escaped version", "pkg:golang/github.com/example/mod@v1.0.0-RC1", "/github.com/example/mod/@v/v1.0.0-!r!c1.zip", "golang", "github.com/example/mod", "v1.0.0-!r!c1", "mod@v1.0.0-!r!c1.zip", func(p *Proxy) http.Handler { return NewGoHandler(p, "http://proxy.test").Routes() }}, + {"go lowercase rule", "pkg:golang/github.com/example/mod@v1.0.0", "/github.com/!example/mod/@v/v1.0.0.zip", + "golang", "github.com/Example/mod", "v1.0.0", goZipFilename("github.com/Example/mod", "v1.0.0"), + func(p *Proxy) http.Handler { return NewGoHandler(p, "http://proxy.test").Routes() }}, + {"go mixed-case rule", "pkg:golang/github.com/Example/mod@v1.0.0", "/github.com/example/mod/@v/v1.0.0.zip", + "golang", "github.com/example/mod", "v1.0.0", "mod@v1.0.0.zip", + func(p *Proxy) http.Handler { return NewGoHandler(p, "http://proxy.test").Routes() }}, {"oci repository path", "pkg:oci/library/alpine@" + ociTestDigest, "/library/alpine/blobs/" + ociTestDigest, "oci", "library/alpine", ociTestDigest, ociTestDigest, func(p *Proxy) http.Handler { diff --git a/internal/handler/go.go b/internal/handler/go.go index 1af00edc..08dae575 100644 --- a/internal/handler/go.go +++ b/internal/handler/go.go @@ -4,11 +4,15 @@ import ( "crypto/sha256" "encoding/hex" "errors" + "fmt" "net/http" + "net/url" "strings" "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/proxy/internal/packageurl" "github.com/git-pkgs/proxy/internal/retention" + "github.com/git-pkgs/purl" "github.com/git-pkgs/registries/fetch" gomodule "golang.org/x/mod/module" ) @@ -18,15 +22,13 @@ const ( asciiCaseOffset = 32 // difference between lowercase and uppercase ASCII letters ) -// Module zips are stored under pkg:golang/{module path}, which the PURL -// rules lowercase, so a package rule covers a module whatever case either -// is written in. A major version suffix (/v2) is part of the module path -// and so a module of its own. A key in the escaped form proxy URLs and the -// module cache use (github.com/!azure) does not survive the default -// mapping's round trip and is refused; the handler stores decoded paths, -// so such a key could never match. func init() { - retention.Default.Register(retention.Spec{Key: "golang", CanonicalPackage: retention.DefaultCanonical("golang")}) + retention.Default.Register(retention.Spec{Key: "golang", CanonicalPackage: golangRetentionKey}) +} + +func golangRetentionKey(p *purl.PURL) (string, bool) { + key, ok := retention.DefaultCanonical("golang")(p) + return packageurl.PolicyPackageKey(key), ok } // GoHandler handles Go module proxy protocol requests. @@ -200,13 +202,7 @@ func (h *GoHandler) handleDownload(w http.ResponseWriter, r *http.Request, modul ServeArtifactRequest(w, r, result) } -// goZipFilename is the name a module zip is cached under. Module paths are -// case-sensitive, but the zip is stored under the module's PURL, which -// lowercases golang names, so github.com/Sirupsen/logrus and -// github.com/sirupsen/logrus share one package. A path with uppercase -// letters therefore gets a hash of its exact spelling in the filename. An -// all-lowercase path keeps the plain name, so existing cache entries stay -// valid. +// Keep the module spelling hash used by caches with lowercased PURL keys. func goZipFilename(module, version string) string { base := lastComponent(module) + "@" + version if module == strings.ToLower(module) { @@ -220,6 +216,40 @@ func goZipFilename(module, version string) string { // filename carries: enough to tell apart paths that differ only in case. const goModuleHashBytes = 8 +// An old cache key may combine case-distinct modules. The upstream path +// identifies which module owns an individual archive. +func (p *Proxy) rekeyLegacyGoArtifact(pkgPURL, versionPURL, filename string) (bool, error) { + if !strings.HasPrefix(versionPURL, "pkg:golang/") { + return false, nil + } + parsed, err := purl.Parse(versionPURL) + if err != nil || parsed.FullName() == strings.ToLower(parsed.FullName()) { + return false, nil + } + name := parsed.FullName() + _, legacyVersion, err := packagePURLStrings("golang", strings.ToLower(name), parsed.Version) + if err != nil { + return false, err + } + record, err := p.DB.GetArtifact(legacyVersion, filename) + if err != nil { + return false, fmt.Errorf("reading legacy Go artifact: %w", err) + } + if record == nil || !record.StoragePath.Valid || record.Source != database.SourceUpstream { + return false, nil + } + escaped, err := gomodule.EscapePath(name) + if err != nil { + return false, nil + } + upstream, err := url.Parse(record.UpstreamURL) + if err != nil || !strings.HasSuffix(upstream.Path, "/"+escaped+"/@v/"+parsed.Version+".zip") { + return false, nil + } + return p.DB.RekeyCachedArtifact(record, &database.Package{PURL: pkgPURL, Ecosystem: "golang", Name: name}, + &database.Version{PURL: versionPURL, PackagePURL: pkgPURL}) +} + // discardOtherModulesZip drops a cached zip that another module stored // under filename. Before module paths kept their case, a module with // uppercase letters was cached under the plain filename of its lowercase diff --git a/internal/handler/go_case_test.go b/internal/handler/go_case_test.go index 9451caa8..91bac444 100644 --- a/internal/handler/go_case_test.go +++ b/internal/handler/go_case_test.go @@ -9,6 +9,7 @@ import ( "sync/atomic" "testing" + "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/registries/fetch" ) @@ -115,6 +116,74 @@ func TestGoZipFromAnotherUpstreamHostIsKept(t *testing.T) { } } +func TestGoLegacyCacheKeepsCaseDistinctModules(t *testing.T) { + upstream, counts := goCaseUpstream(t) + h, proxy := newGoCaseHandler(t, upstream) + seedGoZip(t, proxy, upstream.URL+lowerLogrusPath, "zip of "+lowerLogrusPath) + old := seedLegacyUpperGoZip(t, proxy, "https://old-proxy.example"+upperLogrusPath, "zip of "+upperLogrusPath) + upstream.Close() + + for range 2 { + for _, path := range []string{upperLogrusPath, lowerLogrusPath} { + if w := getGo(h, path); w.Code != http.StatusOK || w.Body.String() != "zip of "+path { + t.Fatalf("GET %s: status = %d, body = %q", path, w.Code, w.Body) + } + } + } + current, err := proxy.DB.GetArtifact("pkg:golang/github.com/Sirupsen/logrus@v1.0.6", old.Filename) + if err != nil || current == nil { + t.Fatalf("moved artifact = %+v, %v", current, err) + } + if current.ID != old.ID || current.StoragePath != old.StoragePath || current.FetchedAt != old.FetchedAt || current.HitCount != old.HitCount+2 { + t.Errorf("artifact history changed: before=%+v, after=%+v", old, current) + } + legacy, err := proxy.DB.GetArtifact(old.VersionPURL, old.Filename) + if err != nil || legacy != nil { + t.Fatalf("legacy artifact = %+v, %v", legacy, err) + } + for path, count := range counts { + if count.Load() != 0 { + t.Errorf("unexpected upstream request: %s", path) + } + } +} + +func TestGoLegacyCacheRequiresExactModuleOrigin(t *testing.T) { + for _, origin := range []string{"", "https://proxy.example" + lowerLogrusPath} { + t.Run(origin, func(t *testing.T) { + upstream, counts := goCaseUpstream(t) + h, proxy := newGoCaseHandler(t, upstream) + seedLegacyUpperGoZip(t, proxy, origin, "another module") + if w := getGo(h, upperLogrusPath); w.Code != http.StatusOK || w.Body.String() != "zip of "+upperLogrusPath { + t.Fatalf("status = %d, body = %q", w.Code, w.Body) + } + if counts[upperLogrusPath].Load() != 1 { + t.Fatal("archive with unproven identity was reused") + } + }) + } +} + +func seedLegacyUpperGoZip(t *testing.T, proxy *Proxy, upstreamURL, content string) *database.Artifact { + t.Helper() + const packagePURL = "pkg:golang/github.com/sirupsen/logrus" + const versionPURL = packagePURL + "@v1.0.6" + const storagePath = "golang/logrus/v1.0.6/legacy/upper.zip" + filename := goZipFilename("github.com/Sirupsen/logrus", "v1.0.6") + if _, _, err := proxy.Storage.Store(context.Background(), storagePath, strings.NewReader(content)); err != nil { + t.Fatal(err) + } + artifact := testArtifact(content, versionPURL, filename, "application/zip") + if err := proxy.updateCacheDB("golang", "github.com/Sirupsen/logrus", packagePURL, upstreamURL, storagePath, artifact, ""); err != nil { + t.Fatal(err) + } + old, err := proxy.DB.GetArtifact(versionPURL, filename) + if err != nil || old == nil { + t.Fatalf("seeded artifact = %+v, %v", old, err) + } + return old +} + // seedGoZip caches content as github.com/sirupsen/logrus v1.0.6 under the // plain filename, recorded as fetched from upstreamURL. func seedGoZip(t *testing.T, proxy *Proxy, upstreamURL, content string) { diff --git a/internal/handler/go_cooldown.go b/internal/handler/go_cooldown.go index 892e77b3..efcc1e19 100644 --- a/internal/handler/go_cooldown.go +++ b/internal/handler/go_cooldown.go @@ -71,12 +71,8 @@ func (h *GoHandler) fetchGoMetadata(ctx context.Context, escapedModule, module, return body, err } -// goTimeKey returns the versions-table PURLs a module's times are stored -// under. golang PURLs lowercase the name, so the key uses the escaped module -// path (github.com/!azure for github.com/Azure), which keeps modules that -// differ only in case apart and is the plain path for lowercase modules. -// Only canonical versions are stored: a query such as a branch name can -// resolve to another commit later. +// Keep escaped module paths in publish-time keys for cache compatibility. +// Only canonical versions are stored; branch queries can resolve differently. func (h *GoHandler) goTimeKey(module, version string) (versionPURL, packagePURL string, ok bool) { if h.proxy.DB == nil || gomodule.CanonicalVersion(version) != version { return "", "", false diff --git a/internal/handler/go_policy_test.go b/internal/handler/go_policy_test.go new file mode 100644 index 00000000..0ce981f1 --- /dev/null +++ b/internal/handler/go_policy_test.go @@ -0,0 +1,47 @@ +package handler + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/git-pkgs/cooldown" + "github.com/git-pkgs/proxy/internal/config" +) + +func TestGoCooldownOverridesIgnoreModuleCase(t *testing.T) { + for _, tc := range []struct { + name, rule, fallback, override string + status int + }{ + {"lowercase restriction", "example.com/module", "0", "7d", http.StatusForbidden}, + {"mixed-case restriction", "example.com/Module", "0", "7d", http.StatusForbidden}, + {"lowercase exemption", "example.com/module", "7d", "0", http.StatusOK}, + {"mixed-case exemption", "example.com/Module", "7d", "0", http.StatusOK}, + } { + t.Run(tc.name, func(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, ".info") { + _ = json.NewEncoder(w).Encode(goInfo{Version: "v1.0.0", Time: time.Now()}) + return + } + _, _ = w.Write([]byte("module zip")) + })) + defer upstream.Close() + h, proxy := newGoCaseHandler(t, upstream) + cfg := config.CooldownConfig{Packages: map[string]string{"pkg:golang/" + tc.rule: tc.override}} + proxy.Cooldown = &cooldown.Config{Default: tc.fallback, Packages: cfg.NormalizedPackages()} + for _, path := range []string{ + "/example.com/!module/@v/v1.0.0.zip", "/example.com/module/@v/v1.0.0.zip", + "/example.com/!module/@v/v1.0.0.info", "/example.com/module/@v/v1.0.0.info", + } { + if response := getGo(h, path); response.Code != tc.status { + t.Errorf("GET %s = %d %s, want %d", path, response.Code, response.Body, tc.status) + } + } + }) + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 39510164..4b7ebe72 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -498,6 +498,12 @@ func (p *Proxy) lookupCache(ctx context.Context, pkgPURL, versionPURL, filename if p.StreamArtifacts { return nil, nil } + if cached, err := p.readStored(ctx, pkgPURL, versionPURL, filename); err != nil || cached != nil { + return cached, err + } + if moved, err := p.rekeyLegacyGoArtifact(pkgPURL, versionPURL, filename); err != nil || !moved { + return nil, err + } return p.readStored(ctx, pkgPURL, versionPURL, filename) } diff --git a/internal/packageurl/packageurl.go b/internal/packageurl/packageurl.go index be0e9ab7..e01f4b2a 100644 --- a/internal/packageurl/packageurl.go +++ b/internal/packageurl/packageurl.go @@ -18,6 +18,15 @@ func MakeString(ecosystem, name, version string) string { return purl.MakePURLString(ecosystem, name, version) } +// PolicyPackageKey maps a canonical versionless PURL to its policy key. +// Go policies match module names case-insensitively for compatibility. +func PolicyPackageKey(packagePURL string) string { + if strings.HasPrefix(packagePURL, "pkg:golang/") { + return strings.ToLower(packagePURL) + } + return packagePURL +} + // WithVersionString returns a package PURL with its version replaced. It // returns an empty string when packagePURL is invalid. func WithVersionString(packagePURL, version string) string { diff --git a/internal/retention/retention.go b/internal/retention/retention.go index a8ff8359..df8e5383 100644 --- a/internal/retention/retention.go +++ b/internal/retention/retention.go @@ -14,6 +14,7 @@ import ( "sync" "time" + "github.com/git-pkgs/proxy/internal/packageurl" "github.com/git-pkgs/purl" ) @@ -41,8 +42,8 @@ type Spec struct { // derived from Key that belong to this ecosystem. ExtraDBEcosystems []string - // CanonicalPackage turns a package PURL from the config into the - // package PURL the handler stores in the cache, reporting false when the + // CanonicalPackage turns a package PURL from the config into a + // policy key, reporting false when the // PURL does not belong to this ecosystem. Nil means the ecosystem accepts // no package overrides. CanonicalPackage func(p *purl.PURL) (string, bool) @@ -150,8 +151,7 @@ func (r *Registry) KeyForDBEcosystem(ecosystem string) (string, bool) { return key, ok } -// Canonical returns the stored package PURL a configured package PURL -// stands for, asking the registered specs in key order. +// Canonical returns the policy key for a configured package PURL. func (r *Registry) Canonical(p *purl.PURL) (string, bool) { for _, key := range r.Keys() { s, _ := r.Lookup(key) @@ -173,7 +173,7 @@ type Rules struct { Default time.Duration // Ecosystems holds rules keyed by registered ecosystem key. Ecosystems map[string]time.Duration - // Packages holds rules keyed by the package PURL stored in the cache. + // Packages holds rules keyed by package PURL, with Go keys lowercased. Packages map[string]time.Duration } @@ -181,6 +181,7 @@ type Rules struct { // stored package PURL: the package rule, else the ecosystem rule, else the // default. Generic artifacts require an explicit package or ecosystem rule. func (r Rules) For(key, packagePURL string) time.Duration { + packagePURL = packageurl.PolicyPackageKey(packagePURL) if d, ok := r.Packages[packagePURL]; ok { return d } diff --git a/internal/server/retention_golang_test.go b/internal/server/retention_golang_test.go index f74eaf41..8ff03579 100644 --- a/internal/server/retention_golang_test.go +++ b/internal/server/retention_golang_test.go @@ -188,8 +188,8 @@ func TestGolangRetentionPackageRules(t *testing.T) { if err := db.Get(&stored, db.Rebind(`SELECT package_purl FROM versions WHERE purl = ?`), zip.VersionPURL); err != nil { t.Fatalf("reading stored package PURL: %v", err) } - if stored != "pkg:golang/github.com/burntsushi/toml" { - t.Errorf("stored package PURL = %q, want pkg:golang/github.com/burntsushi/toml", stored) + if stored != "pkg:golang/github.com/BurntSushi/toml" { + t.Errorf("stored package PURL = %q, want pkg:golang/github.com/BurntSushi/toml", stored) } for _, keys := range []map[string]string{