Follow npm and Packagist change feeds to revalidate changed metadata

README.md

@@ -202,7 +202,7 @@ the proxy, which fetches them from the package's registry and caches them;
 attestations of versions held back by cooldown or blocked by the denylist are
 not served.
 
-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
 
@@ -409,7 +409,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++)
 
@@ -1553,6 +1553,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 |
@@ -1623,7 +1624,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.
 

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"
 

docs/configuration.md

@@ -297,6 +297,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` |
@@ -314,6 +315,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` |
@@ -1461,6 +1463,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. With `cache_metadata` and the metadata rewrite cache on, an npm packument is streamed from the upstream into storage and rewritten from there in two passes, so it isn't held in memory and the limit doesn't apply to it; the rewritten copy is kept gzip-compressed. Requests for a single version (`/<package>/<version>`) and tarball downloads read the version from the stored copy the same way. `metadata_stream_max_size` (1 GB by default) caps a document streamed this way; one over it isn't stored, and the stored copy, if any, is served instead.

internal/config/config.go

@@ -619,6 +619,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"`
@@ -716,6 +722,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"`
@@ -1014,6 +1026,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
 }
 
@@ -1453,6 +1471,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")
@@ -1470,6 +1489,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")

internal/config/config_test.go

@@ -1164,6 +1164,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()
 
@@ -1752,6 +1763,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 {

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)
+	}
+}

internal/database/queries.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 	"time"
@@ -1359,6 +1360,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 {