diff --git a/README.md b/README.md index a324c538..5fdde39e 100644 --- a/README.md +++ b/README.md @@ -734,6 +734,11 @@ signature verification keeps working. Indexes use the metadata cache (`metadata_ttl`, stale fallback); `.apk` packages are stored in the shared artifact cache and remain available when the upstream is unreachable. +Set `upstream.apk_snapshots` (e.g. `"90d"`) to keep replaced indexes and serve +them by date at `/apk/alpine@2026-10-01/v3.22/main`, so pinned package versions +stay installable after the mirror moves on. See +[Alpine APK snapshots](docs/configuration.md#alpine-apk-snapshots). + To proxy other mirrors or private repositories, configure named upstreams under `upstream.apk` (this replaces the built-in default; re-add `alpine` if you still want it): diff --git a/docs/configuration.md b/docs/configuration.md index 3e94531a..e6ee78b9 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -323,6 +323,7 @@ Each upstream used by a built-in package route can be set in YAML or JSON under | `upstream.rpm` | `PROXY_UPSTREAM_RPM` | `https://dl.fedoraproject.org/pub/fedora/linux` | | `upstream.homebrew_api` | `PROXY_UPSTREAM_HOMEBREW_API` | `https://formulae.brew.sh/api` | | `upstream.homebrew_artifact` | `PROXY_UPSTREAM_HOMEBREW_ARTIFACT` | `https://ghcr.io` | +| `upstream.apk_snapshots` | `PROXY_UPSTREAM_APK_SNAPSHOTS` | disabled; see [Alpine APK snapshots](#alpine-apk-snapshots) | Private, ULA, CGNAT, and loopback addresses are rejected by default. Add each private upstream hostname or IP address to `upstream.allow_private_hosts`. The matching environment variable accepts a comma-separated list. Loopback upstreams also require `upstream.allow_loopback: true`. That setting permits upstream requests and redirects to reach any loopback address. @@ -576,6 +577,8 @@ with the metadata TTL and served byte-for-byte unchanged so apk signature verification keeps working; `.apk` packages use the shared artifact cache. When `upstream.apk` is empty, a single repository named `alpine` pointing at the official mirror is available; configuring any entry replaces that default. +Earlier index versions can be kept and served by date; see +[Alpine APK snapshots](#alpine-apk-snapshots). ```yaml upstream: @@ -623,6 +626,53 @@ as signatures and keys, is passed through uncached. When `upstream.freebsd` is e repository named `freebsd` pointing at `https://pkg.freebsd.org` is available; configuring any entry replaces that default. +### Alpine APK snapshots + +```yaml +upstream: + # Keep replaced APK indexes for 90 days. + apk_snapshots: "90d" +``` + +Alpine mirrors only keep the current build of each package, so after an +upgrade `apk add curl=8.5.0-r0` fails because the index no longer lists that +build. With `upstream.apk_snapshots` set, the proxy keeps every version of an +index (and of a detached signature) it stores, and serves them under a dated +repository name: + +``` +http://localhost:8080/apk/alpine@2026-10-01/v3.22/main +``` + +`/apk/{name}@{YYYY-MM-DD}/…` returns the index that was current at the end of +that day (UTC). The kept files are the signed upstream bytes, so apk verifies +them as usual. A day before the first stored version of an index, or a day +that has not started yet, is `404`. Today's snapshot can still change until the +day ends. Packages under a snapshot path are the same as under the plain path: +they come from the artifact cache, or from upstream if not cached; a build the +upstream has dropped and the proxy never cached is `404`. A configured +repository whose name contains `@` takes precedence over a snapshot path. + +Snapshots are recorded when an index is requested through the plain path, so +they only exist for repositories clients use. A version counts from the first +time the proxy served it, which can be up to `metadata_ttl` after upstream +published it. A version is deleted once it was replaced more than the +configured duration ago; the newest version of each index is always kept. +Durations take a `d` suffix; empty or `"0"` disables snapshots. Snapshots need +`cache_metadata`. Their files live under `_snapshots/` in storage and do not +count towards `storage.max_size`. + +Turning snapshots off, or `cache_metadata` off, leaves the kept versions in +place. To remove them, first set a short duration such as `"1h"` and let the +hourly prune delete everything but the newest version of each index. + +Snapshots keep indexes, not packages. A package listed by an old snapshot can +still be evicted by `storage.max_size` or [retention](#retention), and is then +only available while upstream still has it. Retention counts from a package's +last download, so a pinned build nobody installed for longer than the `alpine` +retention is evicted even while a snapshot lists it. Use package rules with +`"0"` for builds that must stay installable. + ### Go checksum database ```yaml diff --git a/internal/config/config.go b/internal/config/config.go index c09a1bae..c956edb0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -773,6 +773,12 @@ type UpstreamConfig struct { // Default when empty: {"alpine": "https://dl-cdn.alpinelinux.org/alpine"}. APK map[string]string `json:"apk" yaml:"apk"` + // APKSnapshots keeps earlier versions of Alpine APK indexes for this + // long after upstream replaced them, and serves them at + // /apk/{name}@{YYYY-MM-DD}/. Accepts a "d" suffix, e.g. "90d". + // Empty or "0" disables snapshots. + APKSnapshots string `json:"apk_snapshots" yaml:"apk_snapshots"` + // Arch maps repository names to Arch Linux pacman mirror base URLs, // served at /arch/{name}/. The remaining request path mirrors the // upstream layout, e.g. /arch/archlinux/core/os/x86_64/core.db. @@ -881,6 +887,11 @@ func (u *UpstreamConfig) Validate() error { if err := validateNamedUpstreams("upstream.apk", u.APK); err != nil { return err } + if u.APKSnapshots != "" { + if _, err := parseRetentionDuration("upstream.apk_snapshots", u.APKSnapshots); err != nil { + return err + } + } if err := validateNamedUpstreams("upstream.arch", u.Arch); err != nil { return err } @@ -1462,6 +1473,7 @@ func (c *Config) LoadFromEnv() { setEnvString(&c.Upstream.Swift, "PROXY_UPSTREAM_SWIFT") setEnvString(&c.Upstream.OCIDefault, "PROXY_UPSTREAM_OCI_DEFAULT") setEnvString(&c.Upstream.Debian, "PROXY_UPSTREAM_DEBIAN") + setEnvString(&c.Upstream.APKSnapshots, "PROXY_UPSTREAM_APK_SNAPSHOTS") setEnvString(&c.Upstream.RPM, "PROXY_UPSTREAM_RPM") setEnvString(&c.Upstream.HomebrewAPI, "PROXY_UPSTREAM_HOMEBREW_API") setEnvString(&c.Upstream.HomebrewArtifact, "PROXY_UPSTREAM_HOMEBREW_ARTIFACT") @@ -1959,6 +1971,20 @@ func (c *Config) ParseGradleBuildCacheMaxUploadSize() int64 { return size } +// ParseAPKSnapshots returns how long earlier Alpine APK indexes are kept +// after upstream replaced them. Returns 0 when snapshots are disabled or the +// value is invalid. +func (c *Config) ParseAPKSnapshots() time.Duration { + if c.Upstream.APKSnapshots == "" { + return 0 + } + d, err := parseDayDuration(c.Upstream.APKSnapshots) + if err != nil || d < 0 { + return 0 + } + return d +} + // ParseGradleBuildCacheMaxAge returns age-based eviction threshold. // Returns 0 when disabled or invalid. func (c *Config) ParseGradleBuildCacheMaxAge() time.Duration { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index a0132cfd..b8c8fbfe 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -2057,6 +2057,41 @@ upstream: } } +func TestAPKSnapshots(t *testing.T) { + cfg := Default() + if got := cfg.ParseAPKSnapshots(); got != 0 { + t.Errorf("default ParseAPKSnapshots = %v, want 0", got) + } + + for value, want := range map[string]time.Duration{ + "90d": 90 * 24 * time.Hour, + "12h": 12 * time.Hour, + "0": 0, + } { + cfg.Upstream.APKSnapshots = value + if err := cfg.Validate(); err != nil { + t.Errorf("Validate(%q): %v", value, err) + } + if got := cfg.ParseAPKSnapshots(); got != want { + t.Errorf("ParseAPKSnapshots(%q) = %v, want %v", value, got, want) + } + } + + for _, value := range []string{"forever", "-1d"} { + cfg.Upstream.APKSnapshots = value + if err := cfg.Validate(); err == nil { + t.Errorf("Validate(%q) succeeded, want an error", value) + } + } + + cfg = Default() + t.Setenv("PROXY_UPSTREAM_APK_SNAPSHOTS", "30d") + cfg.LoadFromEnv() + if cfg.Upstream.APKSnapshots != "30d" { + t.Errorf("APKSnapshots from env = %q, want 30d", cfg.Upstream.APKSnapshots) + } +} + func TestCooldownShortest(t *testing.T) { for _, tc := range []struct { name string diff --git a/internal/database/database_test.go b/internal/database/database_test.go index f8956a63..a61f91d5 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -613,7 +613,7 @@ func createTestPostgresDB(t *testing.T) *DB { // Drop and recreate every table CreateSchema creates for clean test state; // leftover migration records make the next CreateSchema fail on the // migrations primary key. - tables := []string{"artifacts", "pending_deletes", "versions", "packages", "vulnerabilities", "metadata_cache", "metadata_dates", "job_runs", "fetch_leases", "upstream_holds", "migrations", "schema_info"} + tables := []string{"artifacts", "pending_deletes", "versions", "packages", "vulnerabilities", "metadata_cache", "metadata_dates", "job_runs", "fetch_leases", "upstream_holds", "metadata_snapshots", "migrations", "schema_info"} for _, table := range tables { _, _ = db.Exec("DROP TABLE IF EXISTS " + table + " CASCADE") } diff --git a/internal/database/metadata_snapshots.go b/internal/database/metadata_snapshots.go new file mode 100644 index 00000000..268baf74 --- /dev/null +++ b/internal/database/metadata_snapshots.go @@ -0,0 +1,156 @@ +package database + +import ( + "database/sql" + "fmt" + "math" + "time" +) + +// MetadataSnapshot is one kept version of a metadata document, so a request +// for a past date can be answered with the document that was current then. +// It is valid from ValidFromMS (Unix milliseconds, the time the proxy first +// stored it) until the next snapshot of the same ecosystem and name. +type MetadataSnapshot struct { + ID int64 `db:"id"` + Ecosystem string `db:"ecosystem"` + Name string `db:"name"` + StoragePath string `db:"storage_path"` + ContentType sql.NullString `db:"content_type"` + ContentEncoding sql.NullString `db:"content_encoding"` + ContentDigest string `db:"content_digest"` + Size int64 `db:"size"` + ValidFromMS int64 `db:"valid_from_ms"` +} + +// ValidFrom returns the time the snapshot became current. +func (s *MetadataSnapshot) ValidFrom() time.Time { + return time.UnixMilli(s.ValidFromMS) +} + +const metadataSnapshotColumns = `id, ecosystem, name, storage_path, content_type, content_encoding, + content_digest, size, valid_from_ms` + +// MetadataSnapshotAt returns the snapshot of ecosystem and name that was +// current at at: the newest one that became current before it. It returns nil +// when there is none. +func (db *DB) MetadataSnapshotAt(ecosystem, name string, at time.Time) (*MetadataSnapshot, error) { + return db.metadataSnapshotBefore(ecosystem, name, at.UnixMilli()) +} + +// LatestMetadataSnapshot returns the newest snapshot of ecosystem and name, +// or nil when there is none. +func (db *DB) LatestMetadataSnapshot(ecosystem, name string) (*MetadataSnapshot, error) { + return db.metadataSnapshotBefore(ecosystem, name, math.MaxInt64) +} + +func (db *DB) metadataSnapshotBefore(ecosystem, name string, beforeMS int64) (*MetadataSnapshot, error) { + var snap MetadataSnapshot + err := db.Get(&snap, db.Rebind(`SELECT `+metadataSnapshotColumns+` + FROM metadata_snapshots + WHERE ecosystem = ? AND name = ? AND valid_from_ms < ? + ORDER BY valid_from_ms DESC, id DESC + LIMIT 1`), ecosystem, name, beforeMS) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("reading metadata snapshot: %w", err) + } + return &snap, nil +} + +// AddMetadataSnapshot records snap as the version of its document current +// from snap.ValidFromMS, unless the snapshot current at that time already has +// its digest, and reports whether it did. On Postgres it first takes an +// advisory lock on the document, so two replicas can't both read the previous +// digest and both insert. The casts let Postgres type the parameters of the +// INSERT ... SELECT. +func (db *DB) AddMetadataSnapshot(snap *MetadataSnapshot) (bool, error) { + tx, err := db.Beginx() + if err != nil { + return false, fmt.Errorf("recording metadata snapshot: %w", err) + } + defer func() { _ = tx.Rollback() }() + if db.dialect == DialectPostgres { + if _, err := tx.Exec("SELECT pg_advisory_xact_lock($1)", metadataSnapshotLockKey(snap.Ecosystem, snap.Name)); err != nil { + return false, fmt.Errorf("locking metadata snapshots: %w", err) + } + } + res, err := tx.Exec(db.Rebind(` + INSERT INTO metadata_snapshots (ecosystem, name, storage_path, content_type, content_encoding, + content_digest, size, valid_from_ms) + SELECT CAST(? AS TEXT), CAST(? AS TEXT), CAST(? AS TEXT), CAST(? AS TEXT), CAST(? AS TEXT), + CAST(? AS TEXT), CAST(? AS BIGINT), CAST(? AS BIGINT) + WHERE COALESCE(( + SELECT content_digest FROM metadata_snapshots + WHERE ecosystem = ? AND name = ? AND valid_from_ms <= CAST(? AS BIGINT) + ORDER BY valid_from_ms DESC, id DESC + LIMIT 1 + ), '') <> CAST(? AS TEXT) + `), snap.Ecosystem, snap.Name, snap.StoragePath, snap.ContentType, snap.ContentEncoding, + snap.ContentDigest, snap.Size, snap.ValidFromMS, + snap.Ecosystem, snap.Name, snap.ValidFromMS, snap.ContentDigest) + if err != nil { + return false, fmt.Errorf("recording metadata snapshot: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return false, err + } + if err := tx.Commit(); err != nil { + return false, fmt.Errorf("recording metadata snapshot: %w", err) + } + return n > 0, nil +} + +// metadataSnapshotLockKey is the advisory lock AddMetadataSnapshot takes for +// a document. +func metadataSnapshotLockKey(ecosystem, name string) int64 { + return lockKey("metadata-snapshot:" + ecosystem + "\x00" + name) +} + +// PruneMetadataSnapshots deletes the snapshots of ecosystem that stopped +// being current before cutoff, that is those replaced by a newer snapshot +// that became current before cutoff, and queues their blobs for deletion. The +// newest snapshot of each document is always kept. It returns the number of +// snapshots deleted. +func (db *DB) PruneMetadataSnapshots(ecosystem string, cutoff time.Time) (int, error) { + tx, err := db.Beginx() + if err != nil { + return 0, err + } + defer func() { _ = tx.Rollback() }() + + var expired []struct { + ID int64 `db:"id"` + StoragePath string `db:"storage_path"` + } + err = tx.Select(&expired, db.Rebind(` + SELECT s.id, s.storage_path FROM metadata_snapshots s + WHERE s.ecosystem = ? + AND EXISTS ( + SELECT 1 FROM metadata_snapshots n + WHERE n.ecosystem = s.ecosystem AND n.name = s.name + AND (n.valid_from_ms > s.valid_from_ms OR (n.valid_from_ms = s.valid_from_ms AND n.id > s.id)) + AND n.valid_from_ms < ? + ) + `), ecosystem, cutoff.UnixMilli()) + if err != nil { + return 0, fmt.Errorf("listing expired metadata snapshots: %w", err) + } + + queuedAt := time.Now().UTC() + for _, row := range expired { + if _, err := tx.Exec(db.Rebind(`DELETE FROM metadata_snapshots WHERE id = ?`), row.ID); err != nil { + return 0, fmt.Errorf("deleting metadata snapshot: %w", err) + } + if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), row.StoragePath, queuedAt); err != nil { + return 0, fmt.Errorf("queueing metadata snapshot blob: %w", err) + } + } + if err := tx.Commit(); err != nil { + return 0, err + } + return len(expired), nil +} diff --git a/internal/database/metadata_snapshots_test.go b/internal/database/metadata_snapshots_test.go new file mode 100644 index 00000000..71865b98 --- /dev/null +++ b/internal/database/metadata_snapshots_test.go @@ -0,0 +1,190 @@ +package database + +import ( + "testing" + "time" +) + +func addSnapshot(t *testing.T, db *DB, name, digest string, validFrom time.Time) bool { + t.Helper() + added, err := db.AddMetadataSnapshot(&MetadataSnapshot{ + Ecosystem: "alpine", + Name: name, + StoragePath: "_snapshots/alpine/" + name + "/" + digest, + ContentDigest: digest, + Size: 1, + ValidFromMS: validFrom.UnixMilli(), + }) + if err != nil { + t.Fatalf("AddMetadataSnapshot: %v", err) + } + return added +} + +func snapshotDigestAt(t *testing.T, db *DB, name string, at time.Time) string { + t.Helper() + snap, err := db.MetadataSnapshotAt("alpine", name, at) + if err != nil { + t.Fatalf("MetadataSnapshotAt: %v", err) + } + if snap == nil { + return "" + } + return snap.ContentDigest +} + +func TestMetadataSnapshots_AddAndLookup(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + day := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC) + if !addSnapshot(t, db, "idx", "sha256:a", day) { + t.Fatal("first snapshot not added") + } + if addSnapshot(t, db, "idx", "sha256:a", day.Add(time.Hour)) { + t.Error("unchanged digest added again") + } + if !addSnapshot(t, db, "idx", "sha256:b", day.Add(48*time.Hour)) { + t.Fatal("changed digest not added") + } + // A rollback to an earlier version is a new snapshot. + if !addSnapshot(t, db, "idx", "sha256:a", day.Add(96*time.Hour)) { + t.Fatal("rolled back digest not added") + } + + for _, tc := range []struct { + at time.Time + want string + }{ + {day, ""}, + {day.Add(time.Minute), "sha256:a"}, + {day.Add(72 * time.Hour), "sha256:b"}, + {day.Add(100 * time.Hour), "sha256:a"}, + } { + if got := snapshotDigestAt(t, db, "idx", tc.at); got != tc.want { + t.Errorf("snapshot at %s = %q, want %q", tc.at, got, tc.want) + } + } + latest, err := db.LatestMetadataSnapshot("alpine", "idx") + if err != nil || latest == nil || !latest.ValidFrom().Equal(day.Add(96*time.Hour)) { + t.Errorf("LatestMetadataSnapshot = %+v, %v", latest, err) + } + }) +} + +// TestMetadataSnapshots_RecordedLate adds a version after a newer one was +// recorded, as a request that fetched it first but recorded it last does. It +// is compared with the version current at its own time, so it keeps its place +// in the history. +func TestMetadataSnapshots_RecordedLate(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + day := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC) + addSnapshot(t, db, "idx", "sha256:a", day) + addSnapshot(t, db, "idx", "sha256:c", day.Add(2*time.Hour)) + if !addSnapshot(t, db, "idx", "sha256:b", day.Add(time.Hour)) { + t.Fatal("version recorded after a newer one not added") + } + if addSnapshot(t, db, "idx", "sha256:b", day.Add(90*time.Minute)) { + t.Error("version already current at that time added again") + } + for _, tc := range []struct { + at time.Time + want string + }{ + {day.Add(30 * time.Minute), "sha256:a"}, + {day.Add(100 * time.Minute), "sha256:b"}, + {day.Add(3 * time.Hour), "sha256:c"}, + } { + if got := snapshotDigestAt(t, db, "idx", tc.at); got != tc.want { + t.Errorf("snapshot at %s = %q, want %q", tc.at, got, tc.want) + } + } + }) +} + +func TestMetadataSnapshots_Prune(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC) + addSnapshot(t, db, "idx", "sha256:a", now.AddDate(0, 0, -100)) + addSnapshot(t, db, "idx", "sha256:b", now.AddDate(0, 0, -60)) + addSnapshot(t, db, "idx", "sha256:c", now.AddDate(0, 0, -10)) + // The only version of another index stays however old it is. + addSnapshot(t, db, "other", "sha256:x", now.AddDate(0, 0, -200)) + + pruned, err := db.PruneMetadataSnapshots("alpine", now.AddDate(0, 0, -30)) + if err != nil { + t.Fatalf("PruneMetadataSnapshots: %v", err) + } + // a was replaced 60 days ago, b only 10 days ago. + if pruned != 1 { + t.Errorf("pruned = %d, want 1", pruned) + } + if got := snapshotDigestAt(t, db, "idx", now.AddDate(0, 0, -90)); got != "" { + t.Errorf("snapshot of a pruned day = %q, want none", got) + } + if got := snapshotDigestAt(t, db, "idx", now.AddDate(0, 0, -50)); got != "sha256:b" { + t.Errorf("snapshot 50 days ago = %q, want sha256:b", got) + } + if got := snapshotDigestAt(t, db, "other", now); got != "sha256:x" { + t.Errorf("only snapshot of other = %q, want sha256:x", got) + } + + due, err := db.GetDuePendingDeletes(time.Now().Add(time.Hour), 10) + if err != nil { + t.Fatal(err) + } + if len(due) != 1 || due[0] != "_snapshots/alpine/idx/sha256:a" { + t.Errorf("queued deletes = %v, want the pruned blob", due) + } + }) +} + +// TestMetadataSnapshots_ConcurrentPostgres records a version while another +// replica's transaction has recorded the same one and not committed yet, as +// two replicas can. The second must wait and then find it, so one row +// remains. Without the lock both inserts succeed. +func TestMetadataSnapshots_ConcurrentPostgres(t *testing.T) { + db := createTestPostgresDB(t) + if db == nil { + t.Skip("PROXY_DATABASE_URL not set, skipping postgres test") + } + defer func() { _ = db.Close() }() + const name = "v3.22/main/x86_64/APKINDEX.tar.gz" + + other, err := db.Beginx() + if err != nil { + t.Fatal(err) + } + defer func() { _ = other.Rollback() }() + if _, err := other.Exec("SELECT pg_advisory_xact_lock($1)", metadataSnapshotLockKey("alpine", name)); err != nil { + t.Fatal(err) + } + if _, err := other.Exec(`INSERT INTO metadata_snapshots (ecosystem, name, storage_path, content_digest, size, valid_from_ms) + VALUES ('alpine', $1, 'other', 'sha256:a', 1, $2)`, name, time.Now().UnixMilli()); err != nil { + t.Fatal(err) + } + + done := make(chan bool, 1) + go func() { done <- addSnapshot(t, db, name, "sha256:a", time.Now()) }() + select { + case added := <-done: + t.Fatalf("recorded without waiting for the other replica (added=%v)", added) + case <-time.After(200 * time.Millisecond): + } + if err := other.Commit(); err != nil { + t.Fatal(err) + } + if added := <-done; added { + t.Error("recorded a version the other replica had just recorded") + } + var n int + if err := db.Get(&n, `SELECT COUNT(*) FROM metadata_snapshots WHERE name = $1`, name); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Errorf("%d snapshot rows, want 1", n) + } + + // Going back to an earlier digest is still recorded. + if !addSnapshot(t, db, name, "sha256:b", time.Now()) || !addSnapshot(t, db, name, "sha256:a", time.Now().Add(time.Millisecond)) { + t.Error("a change back to an earlier digest was not recorded") + } +} diff --git a/internal/database/queries.go b/internal/database/queries.go index d7150e80..973bd082 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -645,6 +645,7 @@ func (db *DB) GetDuePendingDeletes(cutoff time.Time, limit int) ([]string, error WHERE queued_at < ? AND NOT EXISTS (SELECT 1 FROM artifacts WHERE artifacts.storage_path = pending_deletes.path) AND NOT EXISTS (SELECT 1 FROM metadata_cache WHERE metadata_cache.storage_path = pending_deletes.path) + AND NOT EXISTS (SELECT 1 FROM metadata_snapshots WHERE metadata_snapshots.storage_path = pending_deletes.path) ORDER BY queued_at LIMIT ? `) diff --git a/internal/database/schema.go b/internal/database/schema.go index f0ec6516..6d31ff42 100644 --- a/internal/database/schema.go +++ b/internal/database/schema.go @@ -154,6 +154,20 @@ CREATE TABLE IF NOT EXISTS upstream_holds ( until_ms INTEGER NOT NULL ); +CREATE TABLE IF NOT EXISTS metadata_snapshots ( + id INTEGER PRIMARY KEY, + ecosystem TEXT NOT NULL, + name TEXT NOT NULL, + storage_path TEXT NOT NULL, + content_type TEXT, + content_encoding TEXT, + content_digest TEXT NOT NULL, + size INTEGER NOT NULL, + valid_from_ms INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_metadata_snapshots_key ON metadata_snapshots(ecosystem, name, valid_from_ms); +CREATE INDEX IF NOT EXISTS idx_metadata_snapshots_path ON metadata_snapshots(storage_path); + CREATE TABLE IF NOT EXISTS migrations ( name TEXT NOT NULL PRIMARY KEY, applied_at DATETIME NOT NULL @@ -297,6 +311,20 @@ CREATE TABLE IF NOT EXISTS upstream_holds ( until_ms BIGINT NOT NULL ); +CREATE TABLE IF NOT EXISTS metadata_snapshots ( + id SERIAL PRIMARY KEY, + ecosystem TEXT NOT NULL, + name TEXT NOT NULL, + storage_path TEXT NOT NULL, + content_type TEXT, + content_encoding TEXT, + content_digest TEXT NOT NULL, + size BIGINT NOT NULL, + valid_from_ms BIGINT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_metadata_snapshots_key ON metadata_snapshots(ecosystem, name, valid_from_ms); +CREATE INDEX IF NOT EXISTS idx_metadata_snapshots_path ON metadata_snapshots(storage_path); + CREATE TABLE IF NOT EXISTS migrations ( name TEXT NOT NULL PRIMARY KEY, applied_at TIMESTAMP NOT NULL @@ -468,6 +496,7 @@ var migrations = []migration{ {"017_add_artifact_content_hash_index", migrateAddArtifactContentHashIndex}, {"018_add_upstream_holds", migrateAddUpstreamHolds}, {"019_add_artifact_filename_index", migrateAddArtifactFilenameIndex}, + {"020_add_metadata_snapshots", migrateAddMetadataSnapshots}, } // isTableNotFound returns true if the error indicates a missing table. @@ -850,6 +879,36 @@ func migrateAddArtifactFilenameIndex(db *DB) error { return nil } +// migrateAddMetadataSnapshots creates the table of kept earlier versions of +// metadata documents, which dated snapshot requests are served from. +func migrateAddMetadataSnapshots(db *DB) error { + pk, integer := "id INTEGER PRIMARY KEY", "INTEGER" + if db.dialect == DialectPostgres { + pk, integer = "id SERIAL PRIMARY KEY", "BIGINT" + } + statements := []string{ + fmt.Sprintf(`CREATE TABLE IF NOT EXISTS metadata_snapshots ( + %s, + ecosystem TEXT NOT NULL, + name TEXT NOT NULL, + storage_path TEXT NOT NULL, + content_type TEXT, + content_encoding TEXT, + content_digest TEXT NOT NULL, + size %s NOT NULL, + valid_from_ms %s NOT NULL + )`, pk, integer, integer), + `CREATE INDEX IF NOT EXISTS idx_metadata_snapshots_key ON metadata_snapshots(ecosystem, name, valid_from_ms)`, + `CREATE INDEX IF NOT EXISTS idx_metadata_snapshots_path ON metadata_snapshots(storage_path)`, + } + for _, query := range statements { + if _, err := db.Exec(query); err != nil { + return fmt.Errorf("creating metadata_snapshots: %w", err) + } + } + return nil +} + // migrateAddMetadataDates creates the table of Last-Modified dates for // rewritten metadata, which replicas sharing the database agree on. func migrateAddMetadataDates(db *DB) error { diff --git a/internal/handler/apk.go b/internal/handler/apk.go index 9024fe0e..5de909a5 100644 --- a/internal/handler/apk.go +++ b/internal/handler/apk.go @@ -6,6 +6,7 @@ import ( "net/http" "regexp" "strings" + "time" "github.com/git-pkgs/proxy/internal/retention" "github.com/git-pkgs/purl" @@ -51,6 +52,8 @@ type APKHandler struct { proxy *Proxy proxyURL string repositories map[string]string + // snapshots: see WithSnapshots. + snapshots time.Duration } // NewAPKHandler creates an Alpine APK repository protocol handler. @@ -87,14 +90,18 @@ func (h *APKHandler) Routes() http.Handler { return } - repository, rest, ok := strings.Cut(path, "/") - upstreamURL, found := h.repositories[repository] + segment, rest, ok := strings.Cut(path, "/") + repository, upstreamURL, day, snapshot, found := h.resolveRepository(segment) if !ok || rest == "" || !found { http.NotFound(w, r) return } switch { + case snapshot && (isAPKIndex(rest) || isAPKSignature(rest)): + // A dated snapshot serves the kept version of the index; packages + // and other files below are the same as under the plain path. + h.serveSnapshot(w, r, repository, upstreamURL, rest, day) case isAPKIndex(rest) || isAPKSignature(rest): // Indexes and detached signatures are signed upstream metadata. // Cache them with the metadata TTL and serve the stored bytes @@ -160,8 +167,12 @@ func (h *APKHandler) handlePackageDownload(w http.ResponseWriter, r *http.Reques // metadata cache. Stored bytes are re-served verbatim, which keeps embedded // and detached signatures valid. func (h *APKHandler) handleMetadata(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { - h.proxy.ProxyCached(w, r, upstreamURL+"/"+path, apkEcosystem, - h.metadataCacheKey(repository, upstreamURL, path), "*/*") + cacheKey := h.metadataCacheKey(repository, upstreamURL, path) + if h.snapshotsEnabled() { + h.serveMetadataWithSnapshot(w, r, upstreamURL+"/"+path, cacheKey) + return + } + h.proxy.ProxyCached(w, r, upstreamURL+"/"+path, apkEcosystem, cacheKey, "*/*") } // metadataCacheKey derives the metadata cache key from the repository name, diff --git a/internal/handler/apk_snapshots.go b/internal/handler/apk_snapshots.go new file mode 100644 index 00000000..401dc009 --- /dev/null +++ b/internal/handler/apk_snapshots.go @@ -0,0 +1,239 @@ +package handler + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "database/sql" + "encoding/hex" + "io" + "net/http" + "strconv" + "strings" + "time" + + "github.com/git-pkgs/proxy/internal/database" +) + +// apkSnapshotDateLayout is the date format of a snapshot path segment. +const apkSnapshotDateLayout = "2006-01-02" + +// WithSnapshots keeps earlier versions of the indexes this handler serves and +// makes them available under /apk/{name}@{YYYY-MM-DD}/, as signed upstream +// bytes that apk verifies as usual. keep is how long a version is kept after +// it was replaced; 0 disables snapshots. +func (h *APKHandler) WithSnapshots(keep time.Duration) *APKHandler { + h.snapshots = keep + return h +} + +// snapshotsEnabled reports whether snapshots are recorded and served. They +// need the metadata cache, which stores the documents they are copied from. +func (h *APKHandler) snapshotsEnabled() bool { + return h.snapshots > 0 && h.proxy.CacheMetadata && h.proxy.DB != nil && h.proxy.Storage != nil +} + +// resolveRepository maps the first path segment to a configured repository. +// A segment of the form {name}@{YYYY-MM-DD} names a snapshot of repository +// name for that day, when snapshots are enabled and no repository is +// configured under the full segment. ok is false for anything else. +func (h *APKHandler) resolveRepository(segment string) (repository, upstreamURL string, day time.Time, snapshot, ok bool) { + if upstreamURL, found := h.repositories[segment]; found { + return segment, upstreamURL, time.Time{}, false, true + } + if !h.snapshotsEnabled() { + return "", "", time.Time{}, false, false + } + at := strings.LastIndexByte(segment, '@') + if at < 0 { + return "", "", time.Time{}, false, false + } + upstreamURL, found := h.repositories[segment[:at]] + if !found { + return "", "", time.Time{}, false, false + } + day, err := time.Parse(apkSnapshotDateLayout, segment[at+1:]) + if err != nil { + return "", "", time.Time{}, false, false + } + return segment[:at], upstreamURL, day, true, true +} + +// serveSnapshot answers a request for an index or signature under a snapshot +// path with the version that was current at the end of day. A day for which +// no version was stored, or one that has not begun yet, is 404. +func (h *APKHandler) serveSnapshot(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string, day time.Time) { + end := day.AddDate(0, 0, 1) + if day.After(time.Now()) { + http.NotFound(w, r) + return + } + snap, err := h.proxy.DB.MetadataSnapshotAt(apkEcosystem, h.metadataCacheKey(repository, upstreamURL, path), end) + if err != nil { + h.proxy.Logger.Error("apk snapshot lookup failed", "repository", repository, "path", path, "error", err) + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + if snap == nil { + http.NotFound(w, r) + return + } + + if writeMetadataValidators(w, r, cachedMeta{etag: `"` + snap.ContentDigest + `"`, lastModified: snap.ValidFrom()}) { + return + } + body, err := h.proxy.Storage.Open(r.Context(), snap.StoragePath) + if err != nil { + h.proxy.Logger.Error("apk snapshot unreadable", "repository", repository, "path", path, + "storage_path", snap.StoragePath, "error", err) + http.Error(w, "snapshot unavailable", http.StatusInternalServerError) + return + } + defer func() { _ = body.Close() }() + + if snap.ContentType.String != "" { + w.Header().Set(headerContentType, snap.ContentType.String) + } + w.Header().Set(headerContentLength, strconv.FormatInt(snap.Size, 10)) + if snap.ContentEncoding.String != "" { + w.Header().Set(headerContentEncoding, snap.ContentEncoding.String) + } + w.WriteHeader(http.StatusOK) + if r.Method != http.MethodHead { + _, _ = io.Copy(w, body) + } +} + +// serveMetadataWithSnapshot is ProxyCached, recording each version it serves. +func (h *APKHandler) serveMetadataWithSnapshot(w http.ResponseWriter, r *http.Request, upstream, cacheKey string) { + ctx := context.WithoutCancel(r.Context()) + h.proxy.proxyCachedWithEncoding(w, r, upstream, apkEcosystem, cacheKey, "identity", &metadataHooks{ + hit: func(entry *database.MetadataCacheEntry, seen time.Time) { + h.recordStoredSnapshot(ctx, cacheKey, entry, seen) + }, + fetched: func(body []byte, contentType, contentEncoding string) { + h.recordServedSnapshot(ctx, cacheKey, body, contentType, contentEncoding, time.Now()) + }, + }, "*/*") +} + +// recordServedSnapshot keeps body, the document a request is serving, as the +// version current from observed, unless the version current then already has +// its bytes. +func (h *APKHandler) recordServedSnapshot(ctx context.Context, cacheKey string, body []byte, contentType, contentEncoding string, observed time.Time) { + sum := sha256.Sum256(body) + digest := "sha256:" + hex.EncodeToString(sum[:]) + if !h.snapshotNeeded(cacheKey, digest, observed) { + return + } + storagePath := apkSnapshotStoragePath(cacheKey, digest, observed) + if _, _, err := h.proxy.Storage.Store(ctx, storagePath, bytes.NewReader(body)); err != nil { + h.proxy.Logger.Warn("failed to store apk snapshot", "path", storagePath, "error", err) + h.discardSnapshotBlob(ctx, storagePath) + return + } + h.addSnapshot(ctx, &database.MetadataSnapshot{ + Ecosystem: apkEcosystem, + Name: cacheKey, + StoragePath: storagePath, + ContentType: sql.NullString{String: contentType, Valid: contentType != ""}, + ContentEncoding: sql.NullString{String: contentEncoding, Valid: contentEncoding != ""}, + ContentDigest: digest, + Size: int64(len(body)), + ValidFromMS: observed.UnixMilli(), + }) +} + +// recordStoredSnapshot keeps the cached document entry describes as the +// version current from observed, unless the version current then already +// has its digest. +func (h *APKHandler) recordStoredSnapshot(ctx context.Context, cacheKey string, entry *database.MetadataCacheEntry, observed time.Time) { + if !entry.ContentDigest.Valid || !h.snapshotNeeded(cacheKey, entry.ContentDigest.String, observed) { + return + } + h.storeSnapshot(ctx, cacheKey, entry, observed) +} + +// snapshotNeeded reports whether a version with digest seen at observed +// differs from the version current then. It is only a shortcut that saves +// copying the document: AddMetadataSnapshot makes the same check. +func (h *APKHandler) snapshotNeeded(cacheKey, digest string, observed time.Time) bool { + current, err := h.proxy.DB.MetadataSnapshotAt(apkEcosystem, cacheKey, observed.Add(time.Millisecond)) + return err == nil && (current == nil || current.ContentDigest != digest) +} + +// storeSnapshot records the cached document entry as the version current +// from validFrom. Concurrent requests can both get here for the same +// version; each copies to a path of its own, so the one whose row is not +// added only deletes its own copy. +func (h *APKHandler) storeSnapshot(ctx context.Context, cacheKey string, entry *database.MetadataCacheEntry, validFrom time.Time) { + digest := entry.ContentDigest.String + storagePath := apkSnapshotStoragePath(cacheKey, digest, validFrom) + size, ok := h.copySnapshotBlob(ctx, entry.StoragePath, storagePath, digest) + if !ok { + return + } + h.addSnapshot(ctx, &database.MetadataSnapshot{ + Ecosystem: apkEcosystem, + Name: cacheKey, + StoragePath: storagePath, + ContentType: entry.ContentType, + ContentEncoding: entry.ContentEncoding, + ContentDigest: digest, + Size: size, + ValidFromMS: validFrom.UnixMilli(), + }) +} + +// addSnapshot adds the row for a snapshot whose blob is stored, and deletes +// the blob when the row isn't added, as when another request recorded the +// same version first. +func (h *APKHandler) addSnapshot(ctx context.Context, snap *database.MetadataSnapshot) { + added, err := h.proxy.DB.AddMetadataSnapshot(snap) + if err != nil { + h.proxy.Logger.Warn("failed to record apk snapshot", "key", snap.Name, "error", err) + } + if err != nil || !added { + h.discardSnapshotBlob(ctx, snap.StoragePath) + } +} + +// copySnapshotBlob copies the cached document at from to to and reports its +// size, provided the copied bytes still have digest. A concurrent fetch can +// replace the cached document after its row was read; that version is +// recorded by the request that stored it. +func (h *APKHandler) copySnapshotBlob(ctx context.Context, from, to, digest string) (int64, bool) { + src, err := h.proxy.Storage.Open(ctx, from) + if err != nil { + return 0, false + } + defer func() { _ = src.Close() }() + size, hash, err := h.proxy.Storage.Store(ctx, to, src) + if err != nil { + h.proxy.Logger.Warn("failed to store apk snapshot", "path", to, "error", err) + return 0, false + } + if "sha256:"+hash != digest { + h.discardSnapshotBlob(ctx, to) + return 0, false + } + return size, true +} + +func (h *APKHandler) discardSnapshotBlob(ctx context.Context, path string) { + if err := h.proxy.Storage.Delete(ctx, path); err != nil { + h.proxy.Logger.Warn("failed to discard apk snapshot blob", "path", path, "error", err) + } +} + +// apkSnapshotStoragePath returns a new path to store one kept version of the +// document cached under cacheKey at. The time it became current keeps a +// version that returns after an upgrade was rolled back apart from its +// earlier copy, and a random suffix keeps two requests recording the same +// version in the same millisecond apart. +func apkSnapshotStoragePath(cacheKey, digest string, validFrom time.Time) string { + return "_snapshots/" + apkEcosystem + "/" + cacheKey + "/" + + strconv.FormatInt(validFrom.UnixMilli(), 10) + "-" + strings.TrimPrefix(digest, "sha256:") + + "-" + rand.Text() +} diff --git a/internal/handler/apk_snapshots_test.go b/internal/handler/apk_snapshots_test.go new file mode 100644 index 00000000..83bb26de --- /dev/null +++ b/internal/handler/apk_snapshots_test.go @@ -0,0 +1,382 @@ +package handler + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/registries/fetch" +) + +const apkSnapshotIndexPath = "/v3.22/main/x86_64/APKINDEX.tar.gz" + +// apkSnapshotUpstream serves the index held in index, initially "index v1", +// and counts package downloads. +type apkSnapshotUpstream struct { + index atomic.Value + downloads atomic.Int32 + server *httptest.Server +} + +func newAPKSnapshotUpstream(t *testing.T) *apkSnapshotUpstream { + t.Helper() + u := &apkSnapshotUpstream{} + u.index.Store("index v1") + u.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case apkSnapshotIndexPath: + _, _ = w.Write([]byte(u.index.Load().(string))) + case "/v3.22/main/x86_64/curl-8.5.0-r0.apk": + u.downloads.Add(1) + _, _ = w.Write([]byte("curl 8.5.0")) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(u.server.Close) + return u +} + +// newSnapshotAPKHandler returns an APK handler with snapshots enabled whose +// metadata TTL is 0, so every index request asks upstream. +func newSnapshotAPKHandler(t *testing.T, upstreamURL string) (*APKHandler, *database.DB) { + t.Helper() + proxy, db, _, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = 0 + proxy.HTTPClient = http.DefaultClient + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(http.DefaultClient), fetch.WithMaxRetries(0)) + proxy.Fetcher = fetcher + t.Cleanup(func() { _ = fetcher.Close() }) + h := NewAPKHandler(proxy, "http://proxy.example", map[string]string{"alpine": upstreamURL}). + WithSnapshots(30 * 24 * time.Hour) + return h, db +} + +func snapshotCount(t *testing.T, db *database.DB) int { + t.Helper() + var n int + if err := db.Get(&n, `SELECT COUNT(*) FROM metadata_snapshots`); err != nil { + t.Fatal(err) + } + return n +} + +func expectAPKResponse(t *testing.T, h *APKHandler, target string, status int, body string) { + t.Helper() + w := serveAPKRequest(h, target) + if w.Code != status { + t.Fatalf("%s: status = %d, want %d: %s", target, w.Code, status, w.Body.String()) + } + if status == http.StatusOK && w.Body.String() != body { + t.Errorf("%s: body = %q, want %q", target, w.Body.String(), body) + } +} + +// TestAPKSnapshots_ServeIndexCurrentAtDate records two versions of an index +// through the plain path and checks that each dated path serves the version +// that was current at the end of that day. +func TestAPKSnapshots_ServeIndexCurrentAtDate(t *testing.T) { + upstream := newAPKSnapshotUpstream(t) + h, db := newSnapshotAPKHandler(t, upstream.server.URL) + + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") + // Pretend v1 was first stored on 2026-01-10. + recorded := time.Date(2026, 1, 10, 12, 0, 0, 0, time.UTC) + if _, err := db.Exec(`UPDATE metadata_snapshots SET valid_from_ms = ?`, recorded.UnixMilli()); err != nil { + t.Fatal(err) + } + + upstream.index.Store("index v2") + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v2") + // An unchanged index adds no snapshot. + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v2") + if got := snapshotCount(t, db); got != 2 { + t.Fatalf("snapshots = %d, want 2", got) + } + + today := time.Now().UTC().Format(apkSnapshotDateLayout) + future := time.Now().UTC().AddDate(0, 0, 2).Format(apkSnapshotDateLayout) + expectAPKResponse(t, h, "/alpine@2026-01-09"+apkSnapshotIndexPath, http.StatusNotFound, "") + expectAPKResponse(t, h, "/alpine@2026-01-10"+apkSnapshotIndexPath, http.StatusOK, "index v1") + expectAPKResponse(t, h, "/alpine@"+today+apkSnapshotIndexPath, http.StatusOK, "index v2") + expectAPKResponse(t, h, "/alpine@"+future+apkSnapshotIndexPath, http.StatusNotFound, "") + expectAPKResponse(t, h, "/alpine@2026-13-01"+apkSnapshotIndexPath, http.StatusNotFound, "") + expectAPKResponse(t, h, "/other@2026-01-10"+apkSnapshotIndexPath, http.StatusNotFound, "") + expectAPKResponse(t, h, "/alpine@2026-01-10/v3.22/community/x86_64/APKINDEX.tar.gz", http.StatusNotFound, "") +} + +func TestAPKSnapshots_ConditionalRequest(t *testing.T) { + upstream := newAPKSnapshotUpstream(t) + h, _ := newSnapshotAPKHandler(t, upstream.server.URL) + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") + + target := "/alpine@" + time.Now().UTC().Format(apkSnapshotDateLayout) + apkSnapshotIndexPath + first := serveAPKRequest(h, target) + etag := first.Header().Get("ETag") + if first.Code != http.StatusOK || etag == "" { + t.Fatalf("status = %d, ETag = %q, want 200 with an ETag", first.Code, etag) + } + req := httptest.NewRequest(http.MethodGet, target, nil) + req.Header.Set("If-None-Match", etag) + w := httptest.NewRecorder() + h.Routes().ServeHTTP(w, req) + if w.Code != http.StatusNotModified { + t.Errorf("conditional request: status = %d, want 304", w.Code) + } +} + +// TestAPKSnapshots_PackagesShareTheCache checks that a package requested +// under a snapshot path is the cached artifact of the plain path. +func TestAPKSnapshots_PackagesShareTheCache(t *testing.T) { + upstream := newAPKSnapshotUpstream(t) + h, _ := newSnapshotAPKHandler(t, upstream.server.URL) + + pkg := "/v3.22/main/x86_64/curl-8.5.0-r0.apk" + expectAPKResponse(t, h, "/alpine"+pkg, http.StatusOK, "curl 8.5.0") + expectAPKResponse(t, h, "/alpine@2026-01-10"+pkg, http.StatusOK, "curl 8.5.0") + if got := upstream.downloads.Load(); got != 1 { + t.Errorf("upstream downloads = %d, want 1", got) + } + expectAPKResponse(t, h, "/alpine@2026-01-10/v3.22/main/x86_64/gone-1.0-r0.apk", http.StatusNotFound, "") +} + +func TestAPKSnapshots_DisabledByDefault(t *testing.T) { + upstream := newAPKSnapshotUpstream(t) + h, db := newSnapshotAPKHandler(t, upstream.server.URL) + h.WithSnapshots(0) + + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") + if got := snapshotCount(t, db); got != 0 { + t.Errorf("snapshots = %d, want 0", got) + } + today := time.Now().UTC().Format(apkSnapshotDateLayout) + expectAPKResponse(t, h, "/alpine@"+today+apkSnapshotIndexPath, http.StatusNotFound, "") +} + +// TestAPKSnapshots_ConfiguredNameWins checks that a repository whose +// configured name contains '@' is served as configured, not as a snapshot. +func TestAPKSnapshots_ConfiguredNameWins(t *testing.T) { + upstream := newAPKSnapshotUpstream(t) + h, _ := newSnapshotAPKHandler(t, upstream.server.URL) + h.repositories["alpine@2026-01-10"] = upstream.server.URL + + expectAPKResponse(t, h, "/alpine@2026-01-10"+apkSnapshotIndexPath, http.StatusOK, "index v1") +} + +// TestAPKSnapshots_SameVersionRecordedTwice checks that two requests +// recording the same version in the same millisecond, as coalesced fetches +// after an upstream update can, leave the recorded snapshot readable: the +// request whose row is not added must not delete the other's blob. +func TestAPKSnapshots_SameVersionRecordedTwice(t *testing.T) { + u := newAPKSnapshotUpstream(t) + h, db := newSnapshotAPKHandler(t, u.server.URL) + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") + + key := h.metadataCacheKey("alpine", u.server.URL, apkSnapshotIndexPath[1:]) + entry, err := db.GetMetadataCache(apkEcosystem, key) + if err != nil || entry == nil { + t.Fatalf("cached index row = %v, %v", entry, err) + } + if _, err := db.Exec(`DELETE FROM metadata_snapshots`); err != nil { + t.Fatal(err) + } + validFrom := time.Now() + h.storeSnapshot(t.Context(), key, entry, validFrom) + h.storeSnapshot(t.Context(), key, entry, validFrom) + + if n := snapshotCount(t, db); n != 1 { + t.Fatalf("%d snapshots recorded, want 1", n) + } + snap, err := db.LatestMetadataSnapshot(apkEcosystem, key) + if err != nil || snap == nil { + t.Fatalf("latest snapshot = %v, %v", snap, err) + } + body, err := h.proxy.Storage.Open(t.Context(), snap.StoragePath) + if err != nil { + t.Fatalf("recorded snapshot blob unreadable: %v", err) + } + _ = body.Close() + day := validFrom.UTC().Format(apkSnapshotDateLayout) + expectAPKResponse(t, h, "/alpine@"+day+apkSnapshotIndexPath, http.StatusOK, "index v1") +} + +// pausingWriter blocks the first Write until release is closed, after +// signalling on writing. +type pausingWriter struct { + *httptest.ResponseRecorder + once sync.Once + writing chan struct{} + release chan struct{} +} + +func (w *pausingWriter) Write(b []byte) (int, error) { + w.once.Do(func() { + close(w.writing) + <-w.release + }) + return w.ResponseRecorder.Write(b) +} + +// TestAPKSnapshots_OverlappingRequests pauses a request while it writes +// index v1, lets a second request fetch and serve v2, and then lets the first +// one finish. Both versions must be kept, in the order they were fetched, +// with readable blobs. +func TestAPKSnapshots_OverlappingRequests(t *testing.T) { + u := newAPKSnapshotUpstream(t) + h, db := newSnapshotAPKHandler(t, u.server.URL) + target := "/alpine" + apkSnapshotIndexPath + + first := &pausingWriter{ + ResponseRecorder: httptest.NewRecorder(), + writing: make(chan struct{}), + release: make(chan struct{}), + } + done := make(chan struct{}) + go func() { + defer close(done) + h.Routes().ServeHTTP(first, httptest.NewRequest(http.MethodGet, target, nil)) + }() + select { + case <-first.writing: + case <-done: + t.Fatal("first request finished without writing a body") + } + + u.index.Store("index v2") + expectAPKResponse(t, h, target, http.StatusOK, "index v2") + close(first.release) + <-done + if first.Code != http.StatusOK || first.Body.String() != "index v1" { + t.Fatalf("first request = %d %q, want 200 with index v1", first.Code, first.Body.String()) + } + + var snaps []database.MetadataSnapshot + if err := db.Select(&snaps, `SELECT `+ + `id, ecosystem, name, storage_path, content_type, content_encoding, content_digest, size, valid_from_ms `+ + `FROM metadata_snapshots ORDER BY valid_from_ms, id`); err != nil { + t.Fatal(err) + } + if len(snaps) != 2 { + t.Fatalf("%d snapshots recorded, want 2", len(snaps)) + } + for i, want := range []string{"index v1", "index v2"} { + body, err := h.proxy.Storage.Open(t.Context(), snaps[i].StoragePath) + if err != nil { + t.Fatalf("snapshot %d blob unreadable: %v", i, err) + } + got, _ := io.ReadAll(body) + _ = body.Close() + if string(got) != want { + t.Errorf("snapshot %d = %q, want %q", i, got, want) + } + if snaps[i].Size != int64(len(want)) { + t.Errorf("snapshot %d size = %d, want %d", i, snaps[i].Size, len(want)) + } + } +} + +// TestAPKSnapshots_ConcurrentReplicas records the same index version from +// two replicas sharing a Postgres database, many times at once. One snapshot +// row must remain, with a readable blob, and the other copies are deleted. +func TestAPKSnapshots_ConcurrentReplicas(t *testing.T) { + a, b, db := sharedInstances(t) + u := newAPKSnapshotUpstream(t) + handlers := make([]*APKHandler, 0, 2) + for _, p := range []*Proxy{a, b} { + p.CacheMetadata = true + p.HTTPClient = http.DefaultClient + handlers = append(handlers, NewAPKHandler(p, "http://proxy.example", map[string]string{"alpine": u.server.URL}). + WithSnapshots(30*24*time.Hour)) + } + expectAPKResponse(t, handlers[0], "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") + + key := handlers[0].metadataCacheKey("alpine", u.server.URL, apkSnapshotIndexPath[1:]) + entry, err := db.GetMetadataCache(apkEcosystem, key) + if err != nil || entry == nil { + t.Fatalf("cached index row = %v, %v", entry, err) + } + if _, err := db.Exec(`DELETE FROM metadata_snapshots`); err != nil { + t.Fatal(err) + } + // The request above recorded a copy of its own, which the test leaves be. + store := a.Storage.(*mockStorage) + store.mu.Lock() + before := map[string]bool{} + for path := range store.files { + before[path] = true + } + store.mu.Unlock() + + validFrom := time.Now() + var wg sync.WaitGroup + for i := range 16 { + wg.Go(func() { handlers[i%2].storeSnapshot(t.Context(), key, entry, validFrom) }) + } + wg.Wait() + + if n := snapshotCount(t, db); n != 1 { + t.Fatalf("%d snapshots recorded, want 1", n) + } + snap, err := db.LatestMetadataSnapshot(apkEcosystem, key) + if err != nil || snap == nil { + t.Fatalf("latest snapshot = %v, %v", snap, err) + } + store.mu.Lock() + defer store.mu.Unlock() + if got := string(store.files[snap.StoragePath]); got != "index v1" { + t.Errorf("snapshot blob = %q, want the index", got) + } + for path := range store.files { + if strings.HasPrefix(path, "_snapshots/") && path != snap.StoragePath && !before[path] { + t.Errorf("copy %s of a snapshot that wasn't recorded is left in storage", path) + } + } +} + +// TestAPKSnapshots_CacheHit records an index cached before snapshots were +// enabled when a request is served from the cache within its TTL, and adds +// no row for a later hit on the same version. +func TestAPKSnapshots_CacheHit(t *testing.T) { + u := newAPKSnapshotUpstream(t) + h, db := newSnapshotAPKHandler(t, u.server.URL) + h.proxy.MetadataTTL = 5 * time.Minute + h.WithSnapshots(0) + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") + if n := snapshotCount(t, db); n != 0 { + t.Fatalf("%d snapshots recorded with snapshots off, want 0", n) + } + + h.WithSnapshots(30 * 24 * time.Hour) + // Upstream moved on; a hit still serves the cached v1. + u.index.Store("index v2") + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") + if n := snapshotCount(t, db); n != 1 { + t.Fatalf("%d snapshots recorded on a cache hit, want 1", n) + } + key := h.metadataCacheKey("alpine", u.server.URL, apkSnapshotIndexPath[1:]) + snap, err := db.LatestMetadataSnapshot(apkEcosystem, key) + if err != nil || snap == nil { + t.Fatalf("latest snapshot = %v, %v", snap, err) + } + body, err := h.proxy.Storage.Open(t.Context(), snap.StoragePath) + if err != nil { + t.Fatalf("recorded snapshot blob unreadable: %v", err) + } + got, _ := io.ReadAll(body) + _ = body.Close() + if string(got) != "index v1" { + t.Fatalf("snapshot blob = %q, want %q", got, "index v1") + } + + expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") + if n := snapshotCount(t, db); n != 1 { + t.Fatalf("%d snapshots after a second hit on the same version, want 1", n) + } +} diff --git a/internal/handler/conda.go b/internal/handler/conda.go index 3cbaf937..ca8fd2fb 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -540,7 +540,7 @@ func (h *CondaHandler) proxyCached(w http.ResponseWriter, r *http.Request) { if strings.HasSuffix(r.URL.Path, ".json") { acceptEncoding = "gzip" } - h.proxy.proxyCachedWithEncoding(w, r, h.upstreamURL+r.URL.Path, "conda", cacheKey, acceptEncoding, "*/*") + h.proxy.proxyCachedWithEncoding(w, r, h.upstreamURL+r.URL.Path, "conda", cacheKey, acceptEncoding, nil, "*/*") } // proxyUpstream forwards a request to Anaconda without caching. diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 4b7ebe72..7fef62e7 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -2151,7 +2151,7 @@ func (p *Proxy) cachedMetaFor(entry *database.MetadataCacheEntry) cachedMeta { // When metadata caching is disabled, the response is streamed directly to avoid buffering // large metadata responses (e.g. npm packages with many versions) in memory. func (p *Proxy) ProxyCached(w http.ResponseWriter, r *http.Request, upstreamURL, ecosystem, cacheKey string, acceptHeaders ...string) { - p.proxyCachedWithEncoding(w, r, upstreamURL, ecosystem, cacheKey, "identity", acceptHeaders...) + p.proxyCachedWithEncoding(w, r, upstreamURL, ecosystem, cacheKey, "identity", nil, acceptHeaders...) } // proxyCachedWithEncoding is ProxyCached with an explicit upstream Accept-Encoding. @@ -2159,14 +2159,30 @@ func (p *Proxy) ProxyCached(w http.ResponseWriter, r *http.Request, upstreamURL, // compressed for large, non-hash-pinned metadata whose clients decode gzip // (Homebrew API and Conda repodata). The stored bytes and Content-Encoding are // replayed verbatim either way. -func (p *Proxy) proxyCachedWithEncoding(w http.ResponseWriter, r *http.Request, upstreamURL, ecosystem, cacheKey, acceptEncoding string, acceptHeaders ...string) { +// metadataHooks are optional callbacks for proxyCachedWithEncoding: hit runs +// after a fresh cached row is served, with that row and when it was read, and +// fetched runs with a fetched document before it is written. +type metadataHooks struct { + hit func(entry *database.MetadataCacheEntry, seen time.Time) + fetched func(body []byte, contentType, contentEncoding string) +} + +func (p *Proxy) proxyCachedWithEncoding(w http.ResponseWriter, r *http.Request, upstreamURL, ecosystem, cacheKey, acceptEncoding string, hooks *metadataHooks, acceptHeaders ...string) { if !p.CacheMetadata { // Stream directly without buffering when caching is off. p.proxyMetadataStream(w, r, upstreamURL, acceptEncoding, acceptHeaders...) return } - if p.serveCachedMetadata(w, r, ecosystem, cacheKey) { + if hooks != nil && hooks.hit != nil { + if entry := p.freshMetadataEntry(ecosystem, cacheKey); entry != nil { + seen := time.Now() + if p.serveCachedMetadata(w, r, ecosystem, cacheKey) { + hooks.hit(entry, seen) + return + } + } + } else if p.serveCachedMetadata(w, r, ecosystem, cacheKey) { return } @@ -2176,6 +2192,9 @@ func (p *Proxy) proxyCachedWithEncoding(w http.ResponseWriter, r *http.Request, return } + if hooks != nil && hooks.fetched != nil { + hooks.fetched(body, contentType, contentEncoding) + } p.writeMetadataCachedResponseWithEncoding(w, r, ecosystem, cacheKey, body, contentType, contentEncoding) } diff --git a/internal/handler/homebrew.go b/internal/handler/homebrew.go index 0af67afc..4e40d02b 100644 --- a/internal/handler/homebrew.go +++ b/internal/handler/homebrew.go @@ -64,7 +64,7 @@ func (h *HomebrewHandler) Routes() http.Handler { if strings.HasPrefix(requestPath, "analytics/") { acceptEncoding = "identity" } - h.proxy.proxyCachedWithEncoding(w, r, upstreamURL, homebrewMetadataEcosystem, homebrewMetadataCacheKey(requestPath, r.URL.RawQuery), acceptEncoding, "*/*") + h.proxy.proxyCachedWithEncoding(w, r, upstreamURL, homebrewMetadataEcosystem, homebrewMetadataCacheKey(requestPath, r.URL.RawQuery), acceptEncoding, nil, "*/*") }) } diff --git a/internal/handler/proxy_cached_encoding_test.go b/internal/handler/proxy_cached_encoding_test.go index 35c8793b..45eaef32 100644 --- a/internal/handler/proxy_cached_encoding_test.go +++ b/internal/handler/proxy_cached_encoding_test.go @@ -56,7 +56,7 @@ func (u *gzipWhenAskedUpstream) sawAcceptEncoding() string { func serveGzip(proxy *Proxy, upstreamURL string) *httptest.ResponseRecorder { w := httptest.NewRecorder() r := httptest.NewRequest(http.MethodGet, "/index.json", nil) - proxy.proxyCachedWithEncoding(w, r, upstreamURL, "gzip-test", "index", "gzip", "*/*") + proxy.proxyCachedWithEncoding(w, r, upstreamURL, "gzip-test", "index", "gzip", nil, "*/*") return w } @@ -187,7 +187,7 @@ func TestProxyCachedWithEncoding_UpsertFailureDiscardsBlob(t *testing.T) { // Seed an identity row + plain blob, as every key has before the gzip rollout. w := httptest.NewRecorder() proxy.proxyCachedWithEncoding(w, httptest.NewRequest(http.MethodGet, "/index.json", nil), - upstream.URL+"/index.json", "gzip-test", "index", "identity", "*/*") + upstream.URL+"/index.json", "gzip-test", "index", "identity", nil, "*/*") if w.Code != http.StatusOK { t.Fatalf("seed status = %d, want 200", w.Code) } @@ -250,7 +250,7 @@ func TestProxyCachedWithEncoding_StaleFallbackRereadsRow(t *testing.T) { w := httptest.NewRecorder() proxy.proxyCachedWithEncoding(w, httptest.NewRequest(http.MethodGet, "/index.json", nil), - upstream.URL+"/index.json", "gzip-test", "index", "identity", "*/*") + upstream.URL+"/index.json", "gzip-test", "index", "identity", nil, "*/*") if w.Code != http.StatusOK { t.Fatalf("seed status = %d, want 200", w.Code) } diff --git a/internal/server/apk_snapshots.go b/internal/server/apk_snapshots.go new file mode 100644 index 00000000..ee8e3c98 --- /dev/null +++ b/internal/server/apk_snapshots.go @@ -0,0 +1,48 @@ +package server + +import ( + "context" + "time" +) + +// apkSnapshotPruneInterval is how often snapshots kept past +// upstream.apk_snapshots are deleted. +const apkSnapshotPruneInterval = time.Hour + +// startAPKSnapshotPruning deletes APK index snapshots that were replaced more +// than upstream.apk_snapshots ago, once at startup and then every hour. Their +// blobs go through the pending-delete queue like evicted cache entries. +func (s *Server) startAPKSnapshotPruning(ctx context.Context) { + keep := s.cfg.ParseAPKSnapshots() + if keep <= 0 { + return + } + if !s.cfg.CacheMetadata { + s.logger.Warn("upstream.apk_snapshots needs cache_metadata, snapshots disabled") + return + } + + ticker := time.NewTicker(apkSnapshotPruneInterval) + defer ticker.Stop() + for { + s.pruneAPKSnapshots(ctx, keep, time.Now()) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +func (s *Server) pruneAPKSnapshots(ctx context.Context, keep time.Duration, now time.Time) { + s.runExclusive(ctx, "apk-snapshot-prune", func() { + n, err := s.db.PruneMetadataSnapshots("alpine", now.Add(-keep)) + if err != nil { + s.logger.Warn("apk snapshots: prune failed", "error", err) + return + } + if n > 0 { + s.logger.Info("apk snapshots: pruned", "deleted", n) + } + }) +} diff --git a/internal/server/apk_snapshots_test.go b/internal/server/apk_snapshots_test.go new file mode 100644 index 00000000..191e0737 --- /dev/null +++ b/internal/server/apk_snapshots_test.go @@ -0,0 +1,63 @@ +package server + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "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/fetch" + "github.com/go-chi/chi/v5" +) + +// TestAPKSnapshotsConfigured checks that upstream.apk_snapshots enables the +// dated APK paths, with the digest the file storage records for the cached +// index matching the copy kept as a snapshot. +func TestAPKSnapshotsConfigured(t *testing.T) { + const index = "signed index" + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, index) + })) + defer upstream.Close() + + 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) + } + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + proxy := handler.NewProxy(db, store, fetch.NewFetcher(), fetch.NewResolver(), logger) + proxy.HTTPClient = upstream.Client() + proxy.CacheMetadata = true + + cfg := config.Default() + cfg.BaseURL = "http://proxy.test" + cfg.Upstream.APK = map[string]string{"alpine": upstream.URL} + cfg.Upstream.APKSnapshots = "30d" + s := &Server{cfg: cfg, logger: logger} + r := chi.NewRouter() + s.mountProtocolHandlers(r, proxy) + + today := time.Now().UTC().Format("2006-01-02") + for _, target := range []string{ + "/apk/alpine/v3.22/main/x86_64/APKINDEX.tar.gz", + "/apk/alpine@" + today + "/v3.22/main/x86_64/APKINDEX.tar.gz", + } { + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, target, nil)) + if w.Code != http.StatusOK || w.Body.String() != index { + t.Fatalf("%s: status = %d, body = %q, want 200 %q", target, w.Code, w.Body.String(), index) + } + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 69363ca8..f40c0543 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -233,6 +233,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()) @@ -280,25 +307,8 @@ func (s *Server) serve(listener net.Listener) error { return err } - 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 @@ -428,6 +438,7 @@ func (s *Server) serve(listener net.Listener) error { s.background.Go(func() { s.startEvictionLoop(bgCtx) }) retentionResolvers := s.retentionResolvers(proxy) s.background.Go(func() { s.startRetentionLoop(bgCtx, retentionResolvers) }) + s.background.Go(func() { s.startAPKSnapshotPruning(bgCtx) }) s.background.Go(func() { s.startReclaimLoop(bgCtx) }) if listener != nil { @@ -493,7 +504,8 @@ func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) { containerHandler := s.newContainerHandler(proxy) helmHandler := handler.NewHelmHandlerWithOCIRegistries( proxy, s.cfg.BaseURL, s.cfg.Upstream.Helm, s.cfg.Upstream.OCIDefault, s.cfg.Upstream.OCI) - apkHandler := handler.NewAPKHandler(proxy, s.cfg.BaseURL, s.cfg.Upstream.APK) + apkHandler := handler.NewAPKHandler(proxy, s.cfg.BaseURL, s.cfg.Upstream.APK). + WithSnapshots(s.cfg.ParseAPKSnapshots()) archHandler := handler.NewArchHandler(proxy, s.cfg.BaseURL, s.cfg.Upstream.Arch) freebsdHandler := handler.NewFreeBSDHandler(proxy, s.cfg.BaseURL, s.cfg.Upstream.FreeBSD) debianHandler := handler.NewDebianHandler(