From 1abb8c029a3783197f5856831dc0a5f5b921f658 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 27 Oct 2026 16:18:26 +0100 Subject: [PATCH 1/5] Serve dated snapshots of Alpine APK indexes Alpine mirrors keep only the current build of each package, so once a package is upgraded `apk add curl=8.5.0-r0` fails: the current index no longer lists the old build, even when the proxy still has it cached. With upstream.apk_snapshots set (e.g. "90d"), each version of an index or detached signature the metadata cache stores is copied to _snapshots/alpine/ and recorded in a new metadata_snapshots table with the time it was first served. /apk/{name}@{YYYY-MM-DD}/ serves the version that was current at the end of that day; packages below it are the plain path's requests. The kept bytes are the signed upstream bytes, so apk verifies them unchanged. An hourly job deletes versions replaced longer ago than the configured duration and queues their blobs on the pending-delete queue; the newest version of each index stays. Packages listed by kept snapshots are not protected from eviction or retention; the docs say so. --- README.md | 5 + docs/configuration.md | 46 +++++ internal/config/config.go | 26 +++ internal/config/config_test.go | 35 ++++ internal/database/database_test.go | 2 +- internal/database/metadata_snapshots.go | 136 ++++++++++++++ internal/database/metadata_snapshots_test.go | 108 +++++++++++ internal/database/queries.go | 1 + internal/database/schema.go | 59 ++++++ internal/handler/apk.go | 20 +- internal/handler/apk_snapshots.go | 187 +++++++++++++++++++ internal/handler/apk_snapshots_test.go | 168 +++++++++++++++++ internal/server/apk_snapshots.go | 48 +++++ internal/server/apk_snapshots_test.go | 63 +++++++ internal/server/server.go | 4 +- 15 files changed, 902 insertions(+), 6 deletions(-) create mode 100644 internal/database/metadata_snapshots.go create mode 100644 internal/database/metadata_snapshots_test.go create mode 100644 internal/handler/apk_snapshots.go create mode 100644 internal/handler/apk_snapshots_test.go create mode 100644 internal/server/apk_snapshots.go create mode 100644 internal/server/apk_snapshots_test.go diff --git a/README.md b/README.md index 20c6d5db..c360866e 100644 --- a/README.md +++ b/README.md @@ -722,6 +722,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 f10e4b23..f2faf462 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -314,6 +314,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. @@ -566,6 +567,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: @@ -613,6 +616,49 @@ 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`. + +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 a6606994..6b905417 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -744,6 +744,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. @@ -852,6 +858,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 } @@ -1431,6 +1442,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") @@ -1928,6 +1940,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 7590a64f..e3f2dc23 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -2052,3 +2052,38 @@ upstream: t.Errorf("generic_paths = %v", cfg.Upstream.GenericPaths) } } + +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) + } +} 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..b174aab4 --- /dev/null +++ b/internal/database/metadata_snapshots.go @@ -0,0 +1,136 @@ +package database + +import ( + "database/sql" + "fmt" + "math" + "time" +) + +// Metadata snapshots keep earlier versions of a metadata document after +// upstream replaced it, so a request for a past date can be answered with the +// document that was current then. Each row is one version, valid from +// valid_from_ms (Unix milliseconds, the time the proxy first stored it) until +// the next row for the same ecosystem and name. + +// MetadataSnapshot is one kept version of a metadata document. +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 current version of its document, +// unless the newest snapshot already has its digest, and reports whether it +// did. The check and the insert are one statement, so concurrent callers +// recording the same document add it once. The casts let Postgres type the +// parameters of the INSERT ... SELECT. +func (db *DB) AddMetadataSnapshot(snap *MetadataSnapshot) (bool, error) { + res, err := db.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 = ? + 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.ContentDigest) + if err != nil { + return false, fmt.Errorf("recording metadata snapshot: %w", err) + } + n, err := res.RowsAffected() + return n > 0, err +} + +// 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..ae859fd0 --- /dev/null +++ b/internal/database/metadata_snapshots_test.go @@ -0,0 +1,108 @@ +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) + } + }) +} + +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) + } + }) +} 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..071ae3a9 100644 --- a/internal/handler/apk.go +++ b/internal/handler/apk.go @@ -1,11 +1,13 @@ package handler import ( + "context" "crypto/sha256" "encoding/hex" "net/http" "regexp" "strings" + "time" "github.com/git-pkgs/proxy/internal/retention" "github.com/git-pkgs/purl" @@ -51,6 +53,9 @@ type APKHandler struct { proxy *Proxy proxyURL string repositories map[string]string + // snapshots is how long replaced index versions are kept; 0 disables + // snapshots. See WithSnapshots. + snapshots time.Duration } // NewAPKHandler creates an Alpine APK repository protocol handler. @@ -87,14 +92,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 +169,11 @@ 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) + h.proxy.ProxyCached(w, r, upstreamURL+"/"+path, apkEcosystem, cacheKey, "*/*") + if h.snapshotsEnabled() { + h.recordSnapshot(context.WithoutCancel(r.Context()), 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..cca04d6d --- /dev/null +++ b/internal/handler/apk_snapshots.go @@ -0,0 +1,187 @@ +package handler + +import ( + "context" + "io" + "net/http" + "strconv" + "strings" + "time" + + "github.com/git-pkgs/proxy/internal/database" +) + +// Dated snapshots of APK indexes. +// +// Alpine mirrors keep only the current build of each package, so once a +// package is upgraded the index no longer lists the old build and +// `apk add name=old-version` fails. With snapshots enabled, every version of an +// index (and of a detached signature) the proxy stores is kept, and +// /apk/{name}@{YYYY-MM-DD}/... serves the version that was current at the end +// of that day (UTC). The kept bytes are the signed upstream bytes, so apk +// verifies them as usual. Packages under a snapshot path are the same +// requests as under the plain path. + +// 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}/. 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) + } +} + +// recordSnapshot keeps the document now cached under cacheKey as a snapshot +// when it differs from the newest one kept. It runs after the document was +// served, so a failure only means this version is recorded on a later +// request instead. +func (h *APKHandler) recordSnapshot(ctx context.Context, cacheKey string) { + entry, err := h.proxy.DB.GetMetadataCache(apkEcosystem, cacheKey) + if err != nil || entry == nil || !entry.ContentDigest.Valid { + return + } + digest := entry.ContentDigest.String + latest, err := h.proxy.DB.LatestMetadataSnapshot(apkEcosystem, cacheKey) + if err != nil || (latest != nil && latest.ContentDigest == digest) { + return + } + + validFrom := time.Now() + storagePath := apkSnapshotStoragePath(cacheKey, digest, validFrom) + size, ok := h.copySnapshotBlob(ctx, entry.StoragePath, storagePath, digest) + if !ok { + return + } + added, err := h.proxy.DB.AddMetadataSnapshot(&database.MetadataSnapshot{ + Ecosystem: apkEcosystem, + Name: cacheKey, + StoragePath: storagePath, + ContentType: entry.ContentType, + ContentEncoding: entry.ContentEncoding, + ContentDigest: digest, + Size: size, + ValidFromMS: validFrom.UnixMilli(), + }) + if err != nil { + h.proxy.Logger.Warn("failed to record apk snapshot", "key", cacheKey, "error", err) + } + if err != nil || !added { + // Another request recorded this version first. + h.discardSnapshotBlob(ctx, 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 is where one kept version of the document cached +// under cacheKey is stored. The time it became current keeps a version that +// returns after an upgrade was rolled back apart from its earlier copy. +func apkSnapshotStoragePath(cacheKey, digest string, validFrom time.Time) string { + return "_snapshots/" + apkEcosystem + "/" + cacheKey + "/" + + strconv.FormatInt(validFrom.UnixMilli(), 10) + "-" + strings.TrimPrefix(digest, "sha256:") +} diff --git a/internal/handler/apk_snapshots_test.go b/internal/handler/apk_snapshots_test.go new file mode 100644 index 00000000..50291c13 --- /dev/null +++ b/internal/handler/apk_snapshots_test.go @@ -0,0 +1,168 @@ +package handler + +import ( + "net/http" + "net/http/httptest" + "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") +} 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 12619a6a..fb378acb 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -425,6 +425,7 @@ func (s *Server) serve(listener net.Listener) error { s.background.Go(func() { backfillMetadataAccess(bgCtx, s.db, s.logger, metadataBackfillBatch) }) s.background.Go(func() { s.startEvictionLoop(bgCtx) }) s.background.Go(func() { s.startRetentionLoop(bgCtx) }) + s.background.Go(func() { s.startAPKSnapshotPruning(bgCtx) }) s.background.Go(func() { s.startReclaimLoop(bgCtx) }) if listener != nil { @@ -489,7 +490,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( From 77525b0d78cb01ca4bbad82614233ace33a80dc1 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 30 Oct 2026 10:26:21 +0100 Subject: [PATCH 2/5] Store each snapshot recording at a path of its own Two requests recording the same index version in the same millisecond stored to one path, and the one whose row was not added deleted the blob the other's row points at. The path now ends in a random suffix. Also trims comments that restated the PR, notes in the docs that turning snapshots off leaves them in place, and moves the git cache setup out of serve, which lint found too long after the merge. --- docs/configuration.md | 4 +++ internal/database/metadata_snapshots.go | 11 +++--- internal/handler/apk.go | 3 +- internal/handler/apk_snapshots.go | 38 ++++++++++---------- internal/handler/apk_snapshots_test.go | 37 +++++++++++++++++++ internal/server/server.go | 48 +++++++++++++++---------- 6 files changed, 95 insertions(+), 46 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 17444315..2d7542ee 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -662,6 +662,10 @@ 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 diff --git a/internal/database/metadata_snapshots.go b/internal/database/metadata_snapshots.go index b174aab4..95edacc0 100644 --- a/internal/database/metadata_snapshots.go +++ b/internal/database/metadata_snapshots.go @@ -7,13 +7,10 @@ import ( "time" ) -// Metadata snapshots keep earlier versions of a metadata document after -// upstream replaced it, so a request for a past date can be answered with the -// document that was current then. Each row is one version, valid from -// valid_from_ms (Unix milliseconds, the time the proxy first stored it) until -// the next row for the same ecosystem and name. - -// MetadataSnapshot is one kept version of a metadata document. +// 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"` diff --git a/internal/handler/apk.go b/internal/handler/apk.go index 071ae3a9..83fee975 100644 --- a/internal/handler/apk.go +++ b/internal/handler/apk.go @@ -53,8 +53,7 @@ type APKHandler struct { proxy *Proxy proxyURL string repositories map[string]string - // snapshots is how long replaced index versions are kept; 0 disables - // snapshots. See WithSnapshots. + // snapshots: see WithSnapshots. snapshots time.Duration } diff --git a/internal/handler/apk_snapshots.go b/internal/handler/apk_snapshots.go index cca04d6d..0f29c85d 100644 --- a/internal/handler/apk_snapshots.go +++ b/internal/handler/apk_snapshots.go @@ -2,6 +2,7 @@ package handler import ( "context" + "crypto/rand" "io" "net/http" "strconv" @@ -11,23 +12,13 @@ import ( "github.com/git-pkgs/proxy/internal/database" ) -// Dated snapshots of APK indexes. -// -// Alpine mirrors keep only the current build of each package, so once a -// package is upgraded the index no longer lists the old build and -// `apk add name=old-version` fails. With snapshots enabled, every version of an -// index (and of a detached signature) the proxy stores is kept, and -// /apk/{name}@{YYYY-MM-DD}/... serves the version that was current at the end -// of that day (UTC). The kept bytes are the signed upstream bytes, so apk -// verifies them as usual. Packages under a snapshot path are the same -// requests as under the plain path. - // 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}/. keep is how long a -// version is kept after it was replaced; 0 disables snapshots. +// 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 @@ -125,7 +116,15 @@ func (h *APKHandler) recordSnapshot(ctx context.Context, cacheKey string) { return } - validFrom := time.Now() + h.storeSnapshot(ctx, cacheKey, entry, time.Now()) +} + +// 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 { @@ -178,10 +177,13 @@ func (h *APKHandler) discardSnapshotBlob(ctx context.Context, path string) { } } -// apkSnapshotStoragePath is where one kept version of the document cached -// under cacheKey is stored. The time it became current keeps a version that -// returns after an upgrade was rolled back apart from its earlier copy. +// 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:") + 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 index 50291c13..52b5ff7e 100644 --- a/internal/handler/apk_snapshots_test.go +++ b/internal/handler/apk_snapshots_test.go @@ -166,3 +166,40 @@ func TestAPKSnapshots_ConfiguredNameWins(t *testing.T) { 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") +} diff --git a/internal/server/server.go b/internal/server/server.go index ff06d917..41c4c5a7 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -234,6 +234,33 @@ func (s *Server) Start(listeners ...net.Listener) error { return s.serve(listener) } +// configureGitCache sets up the cache of upstream git repositories, when any +// are configured. +func (s *Server) configureGitCache(client *http.Client) error { + if len(s.cfg.Upstream.Git) == 0 { + return nil + } + upstreamDirs := make(map[string]string, len(s.cfg.Upstream.Git)) + for name, upstream := range s.cfg.Upstream.Git { + upstreamDirs[name] = gitcache.UpstreamDir(name, upstream.URL) + } + gitCache, err := gitcache.New(gitcache.Options{ + CacheDir: s.cfg.Git.CacheDir, + MaxSize: s.cfg.Git.MaxSizeBytes(), + MaxRepoSize: s.cfg.Git.MaxRepoSizeBytes(), + RefreshWait: s.cfg.Git.RefreshWaitDuration(), + StaleMaxAge: s.cfg.Git.StaleMaxAgeDuration(), + Client: client, + Logger: s.logger, + Upstreams: upstreamDirs, + }) + if err != nil { + return fmt.Errorf("configuring git cache: %w", err) + } + s.git = gitCache + return nil +} + func (s *Server) serve(listener net.Listener) error { metadataClient, artifactClient := s.newUpstreamClients(s.cfg.ParseArtifactTimeout()) @@ -283,25 +310,8 @@ func (s *Server) serve(listener net.Listener) error { proxy.DirectServeBaseURL = s.cfg.Storage.DirectServeBaseURL proxy.StreamArtifacts = !s.cfg.Storage.CacheArtifacts - if len(s.cfg.Upstream.Git) > 0 { - upstreamDirs := make(map[string]string, len(s.cfg.Upstream.Git)) - for name, upstream := range s.cfg.Upstream.Git { - upstreamDirs[name] = gitcache.UpstreamDir(name, upstream.URL) - } - gitCache, err := gitcache.New(gitcache.Options{ - CacheDir: s.cfg.Git.CacheDir, - MaxSize: s.cfg.Git.MaxSizeBytes(), - MaxRepoSize: s.cfg.Git.MaxRepoSizeBytes(), - RefreshWait: s.cfg.Git.RefreshWaitDuration(), - StaleMaxAge: s.cfg.Git.StaleMaxAgeDuration(), - Client: artifactClient, - Logger: s.logger, - Upstreams: upstreamDirs, - }) - if err != nil { - return fmt.Errorf("configuring git cache: %w", err) - } - s.git = gitCache + if err := s.configureGitCache(artifactClient); err != nil { + return err } // Create router with Chi From ddba210a60d082fedf2c98bac8dc1bd60a1d6b40 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 30 Oct 2026 15:31:31 +0100 Subject: [PATCH 3/5] Serialize snapshot recording per document on Postgres AddMetadataSnapshot takes a transaction-scoped advisory lock on the document before its check and insert, so two replicas recording the same version can't both read the previous digest and both insert. SQLite runs one write at a time, so it keeps the single statement. --- internal/database/metadata_snapshots.go | 36 ++++++++++-- internal/database/metadata_snapshots_test.go | 52 +++++++++++++++++ internal/handler/apk_snapshots_test.go | 60 ++++++++++++++++++++ 3 files changed, 143 insertions(+), 5 deletions(-) diff --git a/internal/database/metadata_snapshots.go b/internal/database/metadata_snapshots.go index 95edacc0..fba774a6 100644 --- a/internal/database/metadata_snapshots.go +++ b/internal/database/metadata_snapshots.go @@ -62,11 +62,25 @@ func (db *DB) metadataSnapshotBefore(ecosystem, name string, beforeMS int64) (*M // AddMetadataSnapshot records snap as the current version of its document, // unless the newest snapshot already has its digest, and reports whether it -// did. The check and the insert are one statement, so concurrent callers -// recording the same document add it once. The casts let Postgres type the -// parameters of the INSERT ... SELECT. +// did. On Postgres, replicas recording the same document take an advisory +// lock on it first, so their check and insert don't interleave: without it, +// two transactions can both read the previous digest and both insert. SQLite +// runs one write at a time, so the single statement is enough there. A +// document going back to an earlier digest is recorded again, since only the +// newest snapshot is compared. The casts let Postgres type the parameters of +// the INSERT ... SELECT. func (db *DB) AddMetadataSnapshot(snap *MetadataSnapshot) (bool, error) { - res, err := db.Exec(db.Rebind(` + 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), @@ -84,7 +98,19 @@ func (db *DB) AddMetadataSnapshot(snap *MetadataSnapshot) (bool, error) { return false, fmt.Errorf("recording metadata snapshot: %w", err) } n, err := res.RowsAffected() - return n > 0, err + 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 diff --git a/internal/database/metadata_snapshots_test.go b/internal/database/metadata_snapshots_test.go index ae859fd0..0b4919f2 100644 --- a/internal/database/metadata_snapshots_test.go +++ b/internal/database/metadata_snapshots_test.go @@ -106,3 +106,55 @@ func TestMetadataSnapshots_Prune(t *testing.T) { } }) } + +// 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/handler/apk_snapshots_test.go b/internal/handler/apk_snapshots_test.go index 52b5ff7e..38d4b089 100644 --- a/internal/handler/apk_snapshots_test.go +++ b/internal/handler/apk_snapshots_test.go @@ -3,6 +3,8 @@ package handler import ( "net/http" "net/http/httptest" + "strings" + "sync" "sync/atomic" "testing" "time" @@ -203,3 +205,61 @@ func TestAPKSnapshots_SameVersionRecordedTwice(t *testing.T) { day := validFrom.UTC().Format(apkSnapshotDateLayout) expectAPKResponse(t, h, "/alpine@"+day+apkSnapshotIndexPath, http.StatusOK, "index v1") } + +// 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) + } + } +} From 04f732e14744592bea58c1176a64b3f32c887d69 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 30 Oct 2026 19:26:03 +0100 Subject: [PATCH 4/5] Record APK snapshots from the document a request serves A snapshot was recorded after the response by reading the metadata cache again, so a refresh that replaced the cached copy in between lost the version the request served. Record it from the served bytes before responding instead, and compare a new snapshot with the version current at its own time, so one recorded late keeps its place in the history. --- internal/database/metadata_snapshots.go | 21 +++-- internal/database/metadata_snapshots_test.go | 30 ++++++ internal/handler/apk.go | 6 +- internal/handler/apk_snapshots.go | 98 +++++++++++++++++--- internal/handler/apk_snapshots_test.go | 76 +++++++++++++++ 5 files changed, 204 insertions(+), 27 deletions(-) diff --git a/internal/database/metadata_snapshots.go b/internal/database/metadata_snapshots.go index fba774a6..4df1fa15 100644 --- a/internal/database/metadata_snapshots.go +++ b/internal/database/metadata_snapshots.go @@ -60,15 +60,18 @@ func (db *DB) metadataSnapshotBefore(ecosystem, name string, beforeMS int64) (*M return &snap, nil } -// AddMetadataSnapshot records snap as the current version of its document, -// unless the newest snapshot already has its digest, and reports whether it -// did. On Postgres, replicas recording the same document take an advisory -// lock on it first, so their check and insert don't interleave: without it, -// two transactions can both read the previous digest and both insert. SQLite +// 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. Comparing with the version +// current at that time rather than the newest one keeps a version a request +// recorded late, after a newer one, in its place in the history. On +// Postgres, replicas recording the same document take an advisory lock on it +// first, so their check and insert don't interleave: without it, two +// transactions can both read the previous digest and both insert. SQLite // runs one write at a time, so the single statement is enough there. A // document going back to an earlier digest is recorded again, since only the -// newest snapshot is compared. The casts let Postgres type the parameters of -// the INSERT ... SELECT. +// snapshot current at that time is compared. 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 { @@ -87,13 +90,13 @@ func (db *DB) AddMetadataSnapshot(snap *MetadataSnapshot) (bool, error) { CAST(? AS TEXT), CAST(? AS BIGINT), CAST(? AS BIGINT) WHERE COALESCE(( SELECT content_digest FROM metadata_snapshots - WHERE ecosystem = ? AND name = ? + 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.ContentDigest) + snap.Ecosystem, snap.Name, snap.ValidFromMS, snap.ContentDigest) if err != nil { return false, fmt.Errorf("recording metadata snapshot: %w", err) } diff --git a/internal/database/metadata_snapshots_test.go b/internal/database/metadata_snapshots_test.go index 0b4919f2..71865b98 100644 --- a/internal/database/metadata_snapshots_test.go +++ b/internal/database/metadata_snapshots_test.go @@ -70,6 +70,36 @@ func TestMetadataSnapshots_AddAndLookup(t *testing.T) { }) } +// 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) diff --git a/internal/handler/apk.go b/internal/handler/apk.go index 83fee975..5de909a5 100644 --- a/internal/handler/apk.go +++ b/internal/handler/apk.go @@ -1,7 +1,6 @@ package handler import ( - "context" "crypto/sha256" "encoding/hex" "net/http" @@ -169,10 +168,11 @@ func (h *APKHandler) handlePackageDownload(w http.ResponseWriter, r *http.Reques // and detached signatures valid. func (h *APKHandler) handleMetadata(w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string) { cacheKey := h.metadataCacheKey(repository, upstreamURL, path) - h.proxy.ProxyCached(w, r, upstreamURL+"/"+path, apkEcosystem, cacheKey, "*/*") if h.snapshotsEnabled() { - h.recordSnapshot(context.WithoutCancel(r.Context()), cacheKey) + 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 index 0f29c85d..d581980d 100644 --- a/internal/handler/apk_snapshots.go +++ b/internal/handler/apk_snapshots.go @@ -1,8 +1,12 @@ package handler import ( + "bytes" "context" "crypto/rand" + "crypto/sha256" + "database/sql" + "encoding/hex" "io" "net/http" "strconv" @@ -101,22 +105,80 @@ func (h *APKHandler) serveSnapshot(w http.ResponseWriter, r *http.Request, repos } } -// recordSnapshot keeps the document now cached under cacheKey as a snapshot -// when it differs from the newest one kept. It runs after the document was -// served, so a failure only means this version is recorded on a later -// request instead. -func (h *APKHandler) recordSnapshot(ctx context.Context, cacheKey string) { - entry, err := h.proxy.DB.GetMetadataCache(apkEcosystem, cacheKey) - if err != nil || entry == nil || !entry.ContentDigest.Valid { +// serveMetadataWithSnapshot is ProxyCached for an index or signature that +// keeps each version it serves as a snapshot. A version is recorded from the +// document the request serves, not by reading the cache again afterwards: a +// concurrent refresh can replace the cached copy at any time, and the version +// it replaced would then never be recorded. +func (h *APKHandler) serveMetadataWithSnapshot(w http.ResponseWriter, r *http.Request, upstream, cacheKey string) { + p := h.proxy + ctx := context.WithoutCancel(r.Context()) + + // A cached copy within its TTL was recorded by the request that fetched + // it; this only catches one stored before snapshots were enabled, or + // whose recording failed. + if entry := p.freshMetadataEntry(apkEcosystem, cacheKey); entry != nil { + seen := time.Now() + if p.serveCachedMetadata(w, r, apkEcosystem, cacheKey) { + h.recordStoredSnapshot(ctx, cacheKey, entry, seen) + return + } + } + + body, contentType, contentEncoding, err := p.fetchOrCacheMetadata(r.Context(), apkEcosystem, cacheKey, upstream, "identity", nil, "*/*") + if err != nil { + p.writeMetadataFetchError(w, err) return } - digest := entry.ContentDigest.String - latest, err := h.proxy.DB.LatestMetadataSnapshot(apkEcosystem, cacheKey) - if err != nil || (latest != nil && latest.ContentDigest == digest) { + // Record before responding, so the version is kept by the time a + // client has it. + h.recordServedSnapshot(ctx, cacheKey, body, contentType, contentEncoding, time.Now()) + p.writeMetadataCachedResponseWithEncoding(w, r, apkEcosystem, cacheKey, body, contentType, contentEncoding) +} + +// 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(), + }) +} - h.storeSnapshot(ctx, cacheKey, entry, time.Now()) +// 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 @@ -130,7 +192,7 @@ func (h *APKHandler) storeSnapshot(ctx context.Context, cacheKey string, entry * if !ok { return } - added, err := h.proxy.DB.AddMetadataSnapshot(&database.MetadataSnapshot{ + h.addSnapshot(ctx, &database.MetadataSnapshot{ Ecosystem: apkEcosystem, Name: cacheKey, StoragePath: storagePath, @@ -140,12 +202,18 @@ func (h *APKHandler) storeSnapshot(ctx context.Context, cacheKey string, entry * 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", cacheKey, "error", err) + h.proxy.Logger.Warn("failed to record apk snapshot", "key", snap.Name, "error", err) } if err != nil || !added { - // Another request recorded this version first. - h.discardSnapshotBlob(ctx, storagePath) + h.discardSnapshotBlob(ctx, snap.StoragePath) } } diff --git a/internal/handler/apk_snapshots_test.go b/internal/handler/apk_snapshots_test.go index 38d4b089..b0f109df 100644 --- a/internal/handler/apk_snapshots_test.go +++ b/internal/handler/apk_snapshots_test.go @@ -1,6 +1,7 @@ package handler import ( + "io" "net/http" "net/http/httptest" "strings" @@ -206,6 +207,81 @@ func TestAPKSnapshots_SameVersionRecordedTwice(t *testing.T) { 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. From 143eef7f05d6b356e604d6a88bd89c2c7554f5e6 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Sun, 1 Nov 2026 12:48:37 +0100 Subject: [PATCH 5/5] Test snapshots on cache hits and record them through proxyCachedWithEncoding TestAPKSnapshots_CacheHit covers the cache-hit branch with a metadata TTL above 0. proxyCachedWithEncoding takes optional hooks for a cache hit and a fetched body, so the APK handler no longer keeps a copy of it. --- internal/database/metadata_snapshots.go | 16 +++----- internal/handler/apk_snapshots.go | 34 ++++----------- internal/handler/apk_snapshots_test.go | 41 +++++++++++++++++++ internal/handler/conda.go | 2 +- internal/handler/handler.go | 25 +++++++++-- internal/handler/homebrew.go | 2 +- .../handler/proxy_cached_encoding_test.go | 6 +-- 7 files changed, 81 insertions(+), 45 deletions(-) diff --git a/internal/database/metadata_snapshots.go b/internal/database/metadata_snapshots.go index 4df1fa15..268baf74 100644 --- a/internal/database/metadata_snapshots.go +++ b/internal/database/metadata_snapshots.go @@ -61,17 +61,11 @@ func (db *DB) metadataSnapshotBefore(ecosystem, name string, beforeMS int64) (*M } // 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. Comparing with the version -// current at that time rather than the newest one keeps a version a request -// recorded late, after a newer one, in its place in the history. On -// Postgres, replicas recording the same document take an advisory lock on it -// first, so their check and insert don't interleave: without it, two -// transactions can both read the previous digest and both insert. SQLite -// runs one write at a time, so the single statement is enough there. A -// document going back to an earlier digest is recorded again, since only the -// snapshot current at that time is compared. The casts let Postgres type the -// parameters of the INSERT ... SELECT. +// 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 { diff --git a/internal/handler/apk_snapshots.go b/internal/handler/apk_snapshots.go index d581980d..401dc009 100644 --- a/internal/handler/apk_snapshots.go +++ b/internal/handler/apk_snapshots.go @@ -105,35 +105,17 @@ func (h *APKHandler) serveSnapshot(w http.ResponseWriter, r *http.Request, repos } } -// serveMetadataWithSnapshot is ProxyCached for an index or signature that -// keeps each version it serves as a snapshot. A version is recorded from the -// document the request serves, not by reading the cache again afterwards: a -// concurrent refresh can replace the cached copy at any time, and the version -// it replaced would then never be recorded. +// serveMetadataWithSnapshot is ProxyCached, recording each version it serves. func (h *APKHandler) serveMetadataWithSnapshot(w http.ResponseWriter, r *http.Request, upstream, cacheKey string) { - p := h.proxy ctx := context.WithoutCancel(r.Context()) - - // A cached copy within its TTL was recorded by the request that fetched - // it; this only catches one stored before snapshots were enabled, or - // whose recording failed. - if entry := p.freshMetadataEntry(apkEcosystem, cacheKey); entry != nil { - seen := time.Now() - if p.serveCachedMetadata(w, r, apkEcosystem, cacheKey) { + h.proxy.proxyCachedWithEncoding(w, r, upstream, apkEcosystem, cacheKey, "identity", &metadataHooks{ + hit: func(entry *database.MetadataCacheEntry, seen time.Time) { h.recordStoredSnapshot(ctx, cacheKey, entry, seen) - return - } - } - - body, contentType, contentEncoding, err := p.fetchOrCacheMetadata(r.Context(), apkEcosystem, cacheKey, upstream, "identity", nil, "*/*") - if err != nil { - p.writeMetadataFetchError(w, err) - return - } - // Record before responding, so the version is kept by the time a - // client has it. - h.recordServedSnapshot(ctx, cacheKey, body, contentType, contentEncoding, time.Now()) - p.writeMetadataCachedResponseWithEncoding(w, r, apkEcosystem, cacheKey, body, contentType, contentEncoding) + }, + 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 diff --git a/internal/handler/apk_snapshots_test.go b/internal/handler/apk_snapshots_test.go index b0f109df..83bb26de 100644 --- a/internal/handler/apk_snapshots_test.go +++ b/internal/handler/apk_snapshots_test.go @@ -339,3 +339,44 @@ func TestAPKSnapshots_ConcurrentReplicas(t *testing.T) { } } } + +// 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) }