diff --git a/README.md b/README.md index 6aa8d16c..fc990a02 100644 --- a/README.md +++ b/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. 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 24289fe3..3174f719 100644 --- a/docs/configuration.md +++ b/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 (`//`) 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. diff --git a/internal/config/config.go b/internal/config/config.go index 92d5fcf4..f1137c0f 100644 --- a/internal/config/config.go +++ b/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") diff --git a/internal/config/config_test.go b/internal/config/config_test.go index b9fbe05a..91e9d84e 100644 --- a/internal/config/config_test.go +++ b/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 { 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 cacec802..f70c5ebf 100644 --- a/internal/database/queries.go +++ b/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 { diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 2acde79d..85eeef46 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", @@ -319,6 +327,7 @@ func init() { CachedArtifacts, ArtifactsEvicted, MetadataEvicted, + MetadataExpiredByChangeFeed, FetchLeaseWaits, UpstreamFetchDuration, UpstreamErrors, @@ -418,6 +427,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..556650f2 --- /dev/null +++ b/internal/server/change_feeds.go @@ -0,0 +1,108 @@ +package server + +import ( + "context" + "log/slog" + "net/http" + "time" + + "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. +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, 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, changes.NewNPM(c)) + } + if s.cfg.Upstream.ComposerChangeFeed { + feeds = append(feeds, changes.NewPackagist(c, changes.WithURL(changes.PackagistFeedURL(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. 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 { + next, err := readChangeFeed(ctx, db, logger, feed, cursor) + cursor = next + if 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 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 cursor, err + } + cutoff := time.Now() + ecosystem := feed.Ecosystem() + var expired []string + switch { + 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(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)) + 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 new file mode 100644 index 00000000..366c1aed --- /dev/null +++ b/internal/server/change_feeds_test.go @@ -0,0 +1,319 @@ +package server + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "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" +) + +// 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)) + 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")), + } + 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") + 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 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) + } + } + + 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) 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 "", 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) + } +} + +// 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 19673e41..36c43f87 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -406,6 +406,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"}}