Refresh the most used packages in the background

config.example.yaml

@@ -120,8 +120,8 @@ storage:
   # Store fetched artifacts. Set to false to stream every download from
   # upstream without storing it; metadata filtering, cooldown and the
   # denylist still apply. Useful when another cache sits in front of the
-  # proxy. false is incompatible with direct_serve, scanning, mirror_api and
-  # mirror.schedules.
+  # proxy. false is incompatible with direct_serve, scanning, mirror_api,
+  # mirror.schedules and mirror.refresh.
   cache_artifacts: true
 
   # Redirect cached artifact downloads to presigned storage URLs (HTTP 302)
@@ -467,6 +467,16 @@ gradle:
 #     - images: ["pkg:docker/library/alpine@3.20", "pkg:docker/library/nginx@stable"]
 #       interval: "6h"
 #       platforms: ["linux/amd64"]
+#
+# Mirror new releases of the most used packages in the background, so the
+# first install after a release doesn't wait on the upstream. Cooldown and
+# the denylist apply.
+#   refresh:
+#     interval: "24h"
+#     top: 50              # packages per ecosystem, by recent downloads
+#     since: "14d"
+#     max_versions: 200    # per run; default no cap
+#     max_bytes: "2GB"     # per run; default no cap
 
 # Health endpoint configuration.
 health:

docs/configuration.md

@@ -91,7 +91,7 @@ storage:
 
 Every download is a fresh upstream fetch, and concurrent requests for the same artifact are not combined. Artifacts with a digest known up front (OCI blobs, Swift archives, Helm charts) are verified while streaming: the response is sent chunked, and on a mismatch the connection is aborted before the response completes so the client never receives a tampered artifact as a good one. The same happens when the upstream connection fails mid-download.
 
-`cache_artifacts: false` cannot be combined with `scanning.enabled`, `storage.direct_serve`, `mirror_api` or `mirror.schedules`, which all need stored artifacts, and the `mirror` command refuses to run with it.
+`cache_artifacts: false` cannot be combined with `scanning.enabled`, `storage.direct_serve`, `mirror_api`, `mirror.schedules` or `mirror.refresh`, which all need stored artifacts, and the `mirror` command refuses to run with it.
 
 ### Retention
 
@@ -1949,6 +1949,37 @@ Each run checks the tag with a HEAD request. If the tag still points to the cach
 - Mirrored layers are scanned when `scanning.enabled` is set. A blocked layer fails the run and keeps the previous image.
 - Schedules are read at startup; changes need a restart. They do not need `mirror_api`.
 
+### Refreshing the most used packages
+
+For packages, the cache learns about a new release only when a client asks for it, and that first install waits on the upstream or fails while it is down. `mirror.refresh` mirrors new releases of the packages clients download most, so they are cached before anyone asks:
+
+```yaml
+mirror:
+  refresh:
+    interval: 24h
+    top: 50
+    since: 14d
+    max_versions: 200
+    max_bytes: 2GB
+```
+
+| Setting | Env | Description |
+|---------|-----|-------------|
+| `mirror.refresh.interval` | - | How often the refresh runs, at least `1m` |
+| `mirror.refresh.top` | - | Packages per ecosystem to refresh, the ones with the most downloads among those downloaded within `since`. Default: `50` |
+| `mirror.refresh.since` | - | How recently a package must have been downloaded, with a `d` suffix for days. Default: `14d` |
+| `mirror.refresh.max_versions` | - | Stop a run after mirroring this many new versions. Default: no cap |
+| `mirror.refresh.max_bytes` | - | Stop a run once it has downloaded this much, such as `2GB`. It is checked before each version, so the last one can go over it. Default: no cap |
+
+For each of those packages a run lists the releases on the configured upstream and mirrors, for each major line a version is cached on, the newest release of that line. A major line is the major version, or `0.minor` below 1.0. Prereleases and yanked, retracted or deprecated releases are skipped, and so are releases cooldown still withholds or the denylist names, so a run mirrors what clients would be served rather than the newest published. A release that leaves cooldown later is mirrored by a later run.
+
+- PyPI and conda versions publish many files. Only the kinds clients fetched of earlier versions are mirrored: wheels with the same python and ABI tags for the same OS and architecture, whatever glibc or macOS version they need (and, for `abi3` wheels, whatever minimum Python version), and the sdist, or conda packages of the same subdir, build variant (such as `py312` or `_cpython`) and file type.
+- Debian, Alpine and RPM clients find new versions in repository indexes. A run fetches the `InRelease`, `APKINDEX.tar.gz` and `repomd.xml` files clients fetched within `since` again, reads the `Packages` and primary indexes they point at for the components and architectures of the cached packages, and mirrors the newest file each index lists of a package and architecture when it isn't cached but an older version is. Suites sharing a pool, such as bookworm and trixie, are refreshed each on its own that way. Files are checked against the SHA-256 the index lists, where it lists one. Indexes are found by the URL the metadata cache records with them, which caches from before this release lack until they are next revalidated. This requires `cache_metadata`.
+- Container images are not refreshed; list the tags to follow in `mirror.schedules`. Arch Linux and FreeBSD packages, Julia packages, which are cached by tree hash, and other ecosystems the mirror can't list versions of are left out too.
+- Hit counts are kept per file since it was cached, so `top` ranks by all downloads of the files used within `since`.
+- Runs are spread and shared between instances like schedules: the first run comes within ten minutes of startup, later ones every `interval` ±10 %, and instances sharing a PostgreSQL database take turns. A run fetches one version at a time.
+- What a run mirrors is evicted by `storage.max_size` and retention like any other file. With `max_versions` and `max_bytes` a run stays bounded when many packages release at once.
+
 ## Docker
 
 ### SQLite with Local Storage

go.mod

@@ -29,6 +29,7 @@ require (
 	github.com/prometheus/client_golang v1.25.0
 	github.com/prometheus/client_model v0.6.3
 	github.com/swaggo/swag v1.16.6
+	github.com/ulikunitz/xz v0.5.17
 	github.com/vmihailenco/msgpack/v5 v5.4.1
 	gocloud.dev v0.47.0
 	golang.org/x/mod v0.42.0
@@ -108,7 +109,6 @@ require (
 	github.com/rogpeppe/go-internal v1.16.0 // indirect
 	github.com/rs/dnscache v0.0.0-20230804202142-fc85eb664529 // indirect
 	github.com/rubyist/circuitbreaker v2.2.1+incompatible // indirect
-	github.com/ulikunitz/xz v0.5.17 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel v1.46.0 // indirect

internal/config/config.go

@@ -1661,6 +1661,8 @@ func (c *Config) validateCacheArtifacts() error {
 		return fmt.Errorf("storage.cache_artifacts: false cannot be combined with mirror_api: mirrored artifacts would never be served")
 	case len(c.Mirror.Schedules) > 0:
 		return fmt.Errorf("storage.cache_artifacts: false cannot be combined with mirror.schedules: mirrored artifacts would never be served")
+	case c.Mirror.Refresh != nil:
+		return fmt.Errorf("storage.cache_artifacts: false cannot be combined with mirror.refresh: mirrored artifacts would never be served")
 	case c.Hosted.Enabled:
 		return fmt.Errorf("storage.cache_artifacts: false cannot be combined with hosted.enabled: pushed packages would never be served")
 	}

internal/config/mirror.go

@@ -15,6 +15,105 @@ type MirrorConfig struct {
 	// Schedules re-mirror container image tags at a fixed interval, so a
 	// moved tag is picked up even when nobody pulls it.
 	Schedules []MirrorSchedule `json:"schedules" yaml:"schedules"`
+
+	// Refresh, when set, keeps the most used packages up to date in the
+	// background.
+	Refresh *MirrorRefresh `json:"refresh" yaml:"refresh"`
+}
+
+const (
+	defaultMirrorRefreshTop   = 50
+	defaultMirrorRefreshSince = 14 * 24 * time.Hour
+)
+
+// MirrorRefresh mirrors new releases of the packages clients use most, so
+// they are cached before the first client asks for them.
+type MirrorRefresh struct {
+	// Interval is how often the refresh runs, at least one minute (e.g.
+	// "24h").
+	Interval string `json:"interval" yaml:"interval"`
+
+	// Top is how many packages per ecosystem are refreshed, the ones with
+	// the most downloads among those used within Since. Default: 50.
+	Top int `json:"top" yaml:"top"`
+
+	// Since is how recently a package must have been downloaded to be
+	// refreshed (e.g. "14d"). Default: 14d.
+	Since string `json:"since" yaml:"since"`
+
+	// MaxVersions caps how many new versions one run mirrors. 0 means no
+	// cap.
+	MaxVersions int `json:"max_versions" yaml:"max_versions"`
+
+	// MaxBytes caps how much one run downloads (e.g. "2GB"). Empty means no
+	// cap.
+	MaxBytes string `json:"max_bytes" yaml:"max_bytes"`
+
+	interval time.Duration
+	since    time.Duration
+	maxBytes int64
+}
+
+// IntervalDuration returns the parsed Interval. Validate must run first.
+func (r *MirrorRefresh) IntervalDuration() time.Duration { return r.interval }
+
+// TopPackages returns Top, or its default.
+func (r *MirrorRefresh) TopPackages() int {
+	if r.Top == 0 {
+		return defaultMirrorRefreshTop
+	}
+	return r.Top
+}
+
+// SinceDuration returns the parsed Since, or its default. Validate must run
+// first.
+func (r *MirrorRefresh) SinceDuration() time.Duration { return r.since }
+
+// MaxVersionsPerRun returns MaxVersions; 0 means no cap.
+func (r *MirrorRefresh) MaxVersionsPerRun() int { return r.MaxVersions }
+
+// MaxBytesPerRun returns the parsed MaxBytes; 0 means no cap. Validate must
+// run first.
+func (r *MirrorRefresh) MaxBytesPerRun() int64 { return r.maxBytes }
+
+func (r *MirrorRefresh) validate() error {
+	d, err := time.ParseDuration(r.Interval)
+	if err != nil {
+		return fmt.Errorf("invalid mirror.refresh.interval %q: %w", r.Interval, err)
+	}
+	if d < minMirrorScheduleInterval {
+		return fmt.Errorf("invalid mirror.refresh.interval %q: must be at least %s", r.Interval, minMirrorScheduleInterval)
+	}
+	r.interval = d
+	if r.Top < 0 {
+		return fmt.Errorf("invalid mirror.refresh.top %d: must not be negative", r.Top)
+	}
+	r.since = defaultMirrorRefreshSince
+	if r.Since != "" {
+		since, err := parseDayDuration(r.Since)
+		if err != nil {
+			return fmt.Errorf("invalid mirror.refresh.since %q: %w", r.Since, err)
+		}
+		if since <= 0 {
+			return fmt.Errorf("invalid mirror.refresh.since %q: must be positive", r.Since)
+		}
+		r.since = since
+	}
+	if r.MaxVersions < 0 {
+		return fmt.Errorf("invalid mirror.refresh.max_versions %d: must not be negative", r.MaxVersions)
+	}
+	r.maxBytes = 0
+	if r.MaxBytes != "" {
+		n, err := ParseSize(r.MaxBytes)
+		if err != nil {
+			return fmt.Errorf("invalid mirror.refresh.max_bytes %q: %w", r.MaxBytes, err)
+		}
+		if n < 0 {
+			return fmt.Errorf("invalid mirror.refresh.max_bytes %q: must not be negative", r.MaxBytes)
+		}
+		r.maxBytes = n
+	}
+	return nil
 }
 
 // MirrorSchedule is a set of image PURLs mirrored every Interval.
@@ -45,6 +144,9 @@ func (m *MirrorConfig) Validate() error {
 			return err
 		}
 	}
+	if m.Refresh != nil {
+		return m.Refresh.validate()
+	}
 	return nil
 }
 

internal/config/mirror_test.go

@@ -87,3 +87,85 @@ mirror:
 		t.Error("Validate() accepted a 30s interval in the second schedule")
 	}
 }
+
+func TestMirrorRefreshValidate(t *testing.T) {
+	tests := []struct {
+		name    string
+		refresh *MirrorRefresh
+		wantErr bool
+	}{
+		{name: "defaults", refresh: &MirrorRefresh{Interval: "24h"}},
+		{name: "all fields", refresh: &MirrorRefresh{Interval: "12h", Top: 20, Since: "14d", MaxVersions: 100, MaxBytes: "2GB"}},
+		{name: "missing interval", refresh: &MirrorRefresh{}, wantErr: true},
+		{name: "interval under a minute", refresh: &MirrorRefresh{Interval: "10s"}, wantErr: true},
+		{name: "negative top", refresh: &MirrorRefresh{Interval: "1h", Top: -1}, wantErr: true},
+		{name: "bad since", refresh: &MirrorRefresh{Interval: "1h", Since: "recently"}, wantErr: true},
+		{name: "negative since", refresh: &MirrorRefresh{Interval: "1h", Since: "-1d"}, wantErr: true},
+		{name: "negative max_versions", refresh: &MirrorRefresh{Interval: "1h", MaxVersions: -1}, wantErr: true},
+		{name: "bad max_bytes", refresh: &MirrorRefresh{Interval: "1h", MaxBytes: "lots"}, wantErr: true},
+		{name: "negative max_bytes", refresh: &MirrorRefresh{Interval: "1h", MaxBytes: "-1GB"}, wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := MirrorConfig{Refresh: tt.refresh}
+			err := m.Validate()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("Validate() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestMirrorRefreshDefaults(t *testing.T) {
+	m := MirrorConfig{Refresh: &MirrorRefresh{Interval: "24h"}}
+	if err := m.Validate(); err != nil {
+		t.Fatalf("Validate: %v", err)
+	}
+	r := m.Refresh
+	if r.IntervalDuration() != 24*time.Hour || r.TopPackages() != 50 || r.SinceDuration() != 14*24*time.Hour ||
+		r.MaxVersionsPerRun() != 0 || r.MaxBytesPerRun() != 0 {
+		t.Errorf("got interval %v top %d since %v max_versions %d max_bytes %d, want 24h 50 336h 0 0",
+			r.IntervalDuration(), r.TopPackages(), r.SinceDuration(), r.MaxVersionsPerRun(), r.MaxBytesPerRun())
+	}
+}
+
+func TestValidateRejectsMirrorRefreshWithoutCachedArtifacts(t *testing.T) {
+	cfg := Default()
+	cfg.Mirror.Refresh = &MirrorRefresh{Interval: "24h"}
+	cfg.Storage.CacheArtifacts = false
+	if err := cfg.Validate(); err == nil {
+		t.Error("Validate() accepted mirror.refresh with storage.cache_artifacts: false")
+	}
+}
+
+func TestLoadYAMLMirrorRefresh(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "config.yaml")
+	data := `
+mirror:
+  refresh:
+    interval: 12h
+    top: 10
+    since: 7d
+    max_versions: 200
+    max_bytes: 1GB
+`
+	if err := os.WriteFile(path, []byte(data), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	cfg, err := Load(path)
+	if err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	r := cfg.Mirror.Refresh
+	if r == nil {
+		t.Fatal("mirror.refresh not loaded")
+	}
+	if err := cfg.Validate(); err != nil {
+		t.Fatalf("Validate: %v", err)
+	}
+	if r.IntervalDuration() != 12*time.Hour || r.TopPackages() != 10 || r.SinceDuration() != 7*24*time.Hour ||
+		r.MaxVersionsPerRun() != 200 || r.MaxBytesPerRun() != 1<<30 {
+		t.Errorf("got interval %v top %d since %v max_versions %d max_bytes %d",
+			r.IntervalDuration(), r.TopPackages(), r.SinceDuration(), r.MaxVersionsPerRun(), r.MaxBytesPerRun())
+	}
+}

internal/database/metadata_upstream_url_test.go

@@ -0,0 +1,90 @@
+package database
+
+import (
+	"database/sql"
+	"path/filepath"
+	"reflect"
+	"testing"
+	"time"
+)
+
+func TestMigrationAddsMetadataUpstreamURL(t *testing.T) {
+	dbPath := filepath.Join(t.TempDir(), "existing.db")
+	sqlDB, err := sql.Open("sqlite", dbPath)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if _, err := sqlDB.Exec(schemaSQLite); err != nil {
+		t.Fatalf("create schema: %v", err)
+	}
+	if _, err := sqlDB.Exec(`ALTER TABLE metadata_cache DROP COLUMN upstream_url`); err != nil {
+		t.Fatalf("drop upstream_url column: %v", err)
+	}
+	if _, err := sqlDB.Exec(`DELETE FROM migrations WHERE name = '023_add_metadata_upstream_url'`); err != nil {
+		t.Fatalf("forget migration: %v", err)
+	}
+	if err := sqlDB.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+
+	db, err := Open(dbPath)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+	if err := db.MigrateSchema(); err != nil {
+		t.Fatalf("MigrateSchema: %v", err)
+	}
+	has, err := db.HasColumn("metadata_cache", "upstream_url")
+	if err != nil || !has {
+		t.Errorf("metadata_cache.upstream_url after migration = %v, %v", has, err)
+	}
+}
+
+func upsertMetadataURL(t *testing.T, db *DB, ecosystem, name, url string) {
+	t.Helper()
+	entry := &MetadataCacheEntry{Ecosystem: ecosystem, Name: name, StoragePath: "_metadata/" + name}
+	if url != "" {
+		entry.UpstreamURL = sql.NullString{String: url, Valid: true}
+	}
+	if err := db.UpsertMetadataCache(entry); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestUpsertMetadataCacheKeepsUpstreamURL(t *testing.T) {
+	runWithBothDatabases(t, func(t *testing.T, db *DB) {
+		upsertMetadataURL(t, db, "apk", "k1", "https://dl-cdn.alpinelinux.org/alpine/v3.20/main/aarch64/APKINDEX.tar.gz")
+		// A writer that doesn't know the URL leaves it as it was.
+		upsertMetadataURL(t, db, "apk", "k1", "")
+		got, err := db.GetMetadataCache("apk", "k1")
+		if err != nil || got == nil {
+			t.Fatalf("GetMetadataCache: %+v, %v", got, err)
+		}
+		if got.UpstreamURL.String != "https://dl-cdn.alpinelinux.org/alpine/v3.20/main/aarch64/APKINDEX.tar.gz" {
+			t.Errorf("upstream_url = %q", got.UpstreamURL.String)
+		}
+	})
+}
+
+func TestRecentMetadataDocuments(t *testing.T) {
+	runWithBothDatabases(t, func(t *testing.T, db *DB) {
+		upsertMetadataURL(t, db, "apk", "k1", "https://a.test/v3.20/main/aarch64/APKINDEX.tar.gz")
+		upsertMetadataURL(t, db, "apk", "k2", "https://a.test/v3.19/main/aarch64/APKINDEX.tar.gz")
+		upsertMetadataURL(t, db, "apk", "k3", "")
+		upsertMetadataURL(t, db, "debian", "k4", "https://d.test/dists/bookworm/InRelease")
+		old := time.Now().Add(-30 * 24 * time.Hour).UTC()
+		if _, err := db.Exec(db.Rebind("UPDATE metadata_cache SET last_accessed_at = ?, updated_at = ? WHERE name = ?"), old, old, "k2"); err != nil {
+			t.Fatal(err)
+		}
+
+		got, err := db.RecentMetadataDocuments("apk", time.Now().Add(-14*24*time.Hour))
+		if err != nil {
+			t.Fatal(err)
+		}
+		want := []MetadataDocument{{Name: "k1", UpstreamURL: "https://a.test/v3.20/main/aarch64/APKINDEX.tar.gz"}}
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("RecentMetadataDocuments = %+v, want %+v", got, want)
+		}
+	})
+}