From 42a071bd7e25bbcf26bfba0d0f73a3d9d8e6f68d Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 10 Nov 2026 16:54:20 +0100 Subject: [PATCH 1/2] Keep the blobs of the newest container images with keep_latest storage.retention.keep_latest now also applies to oci. Blobs are cached under their digest, so there is no version to order. Instead the proxy keeps the blobs of the N images of a repository that a tag most recently pointed at. To know which blobs an image needs, the proxy records what each manifest it serves or mirrors references, in a new image_manifests table. The retention sweep follows index -> platform manifests -> config and layers from the newest tagged images and keeps those blobs. It also prunes the records of images that are no longer among the newest. Nothing is recorded unless keep_latest is set and a retention rule can evict oci blobs. Images pulled only by digest are not counted. --- config.example.yaml | 4 +- docs/architecture.md | 11 + docs/configuration.md | 12 +- internal/database/database_test.go | 2 +- internal/database/image_manifests.go | 170 ++++++++++ internal/database/image_manifests_test.go | 190 +++++++++++ internal/database/schema.go | 40 +++ internal/handler/container.go | 2 +- internal/handler/container_image_refs.go | 97 ++++++ internal/handler/container_image_refs_test.go | 192 +++++++++++ internal/handler/container_manifest.go | 12 +- internal/handler/container_mirror.go | 2 + internal/handler/handler.go | 6 + internal/handler/profile.go | 2 +- internal/handler/profile_test.go | 1 + internal/retention/keep_latest_test.go | 5 +- internal/retention/retention.go | 11 +- internal/server/retention.go | 75 ++++- .../server/retention_keep_latest_oci_test.go | 301 ++++++++++++++++++ internal/server/retention_keep_latest_test.go | 4 +- internal/server/server.go | 1 + 21 files changed, 1116 insertions(+), 24 deletions(-) create mode 100644 internal/database/image_manifests.go create mode 100644 internal/database/image_manifests_test.go create mode 100644 internal/handler/container_image_refs.go create mode 100644 internal/handler/container_image_refs_test.go create mode 100644 internal/server/retention_keep_latest_oci_test.go diff --git a/config.example.yaml b/config.example.yaml index 240e5646..ce4d8f11 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -117,7 +117,9 @@ storage: # How often to look for expired artifacts; at least "1m". sweep_interval: "10m" # Keep this many of the newest cached versions of each package from - # eviction by age (cargo, golang, npm and pypi). 0 keeps none. + # eviction by age (cargo, golang, npm and pypi). For oci it keeps the + # blobs of this many images per repository, those a tag most recently + # pointed at. 0 keeps none. # keep_latest: 2 # Store fetched artifacts. Set to false to stream every download from diff --git a/docs/architecture.md b/docs/architecture.md index 1d174266..c498cc7f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -205,10 +205,21 @@ metadata_dates ( last_modified INTEGER NOT NULL, -- Unix seconds; BIGINT on Postgres PRIMARY KEY (ecosystem, proxy_url, name) ) + +image_manifests ( + package_purl TEXT NOT NULL, -- the repository, as its blobs are cached + digest TEXT NOT NULL, + refs TEXT NOT NULL, -- digests it references, one per line + tagged_at INTEGER NOT NULL, -- Unix ms a tag last pointed at it, 0 if none did; BIGINT on Postgres + recorded_at INTEGER NOT NULL, -- Unix ms; BIGINT on Postgres + PRIMARY KEY (package_purl, digest) +) ``` `metadata_dates` holds the `Last-Modified` served with rewritten metadata, such as npm packuments, Composer documents, PyPI simple pages and Helm indexes. A package's date is the second it started serving its current representation, and a new representation is only dated in a later second. Keeping it in the database means replicas sharing one serve the same date, and never give one representation's date to another, so `If-Modified-Since` cannot return 304 for a body the client does not have. +`image_manifests` is filled only with `storage.retention.keep_latest` set and a retention rule for `oci`. It holds which manifests and blobs each served container manifest references, so the retention sweep can keep the blobs of the newest images of a repository. The sweep also drops the rows of images that are no longer among the newest. + On PostgreSQL, `INTEGER PRIMARY KEY` becomes `SERIAL`, `DATETIME` becomes `TIMESTAMP`, `INTEGER DEFAULT 0` booleans become `BOOLEAN DEFAULT FALSE`, and size/count columns use `BIGINT`. The `MigrateSchema()` function handles backward compatibility with older git-pkgs databases by running named migrations that add missing columns and tables. See [migrations.md](migrations.md) for how to add new schema changes. diff --git a/docs/configuration.md b/docs/configuration.md index 1ac89a6b..35066b8a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -71,7 +71,7 @@ storage: | `storage.cache_artifacts` | `PROXY_STORAGE_CACHE_ARTIFACTS` | - | Store fetched artifacts (default: true); `false` streams them from upstream | | `storage.retention.default` | `PROXY_STORAGE_RETENTION_DEFAULT` | - | Evict artifacts not served for this long (e.g. "30d"); see [Retention](#retention) | | `storage.retention.sweep_interval` | `PROXY_STORAGE_RETENTION_SWEEP_INTERVAL` | - | How often expired artifacts are looked for (default: "10m", at least "1m") | -| `storage.retention.keep_latest` | - | - | Keep this many of the newest cached versions of each package from eviction by age (default: 0); see [Retention](#retention) | +| `storage.retention.keep_latest` | - | - | Keep this many of the newest cached versions of each package, or images of each repository, from eviction by age (default: 0); see [Retention](#retention) | `storage.max_size` counts cached artifacts and cached metadata documents (`cache_metadata`). When the total goes over the limit, the least recently served entries of either kind are evicted first; an evicted metadata document is fetched from upstream again on the next request for it. Pushed artifacts are not counted or evicted. An artifact replaced by a refetch stays in storage for at least an hour, or `storage.direct_serve_ttl` if longer, so requests already reading it can finish, and storage use can exceed the limit by what was replaced in that time. @@ -127,7 +127,15 @@ storage: keep_latest: 2 ``` -With this, the files of the two highest cached versions of a package are never evicted by age, and older versions expire by their rule as before. Versions are ordered by the ecosystem's version rules, not by when they were fetched, and a prerelease counts like any other version. Only versions in the cache count, not those upstream, so the kept ones need not be the latest releases. `keep_latest` applies to `cargo`, `golang`, `npm` and `pypi`; the other ecosystems ignore it, since many of them cache files under a digest or a build rather than a version. On startup the proxy logs the ecosystems it reaches (`keep_latest_covered`). `storage.max_size` can still evict a kept version. +With this, the files of the two highest cached versions of a package are never evicted by age, and older versions expire by their rule as before. Versions are ordered by the ecosystem's version rules, not by when they were fetched, and a prerelease counts like any other version. Only versions in the cache count, not those upstream, so the kept ones need not be the latest releases. `keep_latest` applies to `cargo`, `golang`, `npm`, `pypi` and, by image, `oci`; the other ecosystems ignore it, since many of them cache files under a digest or a build rather than a version. On startup the proxy logs the ecosystems it reaches (`keep_latest_covered`). `storage.max_size` can still evict a kept version. + +For `oci` there are no versions to order, so `keep_latest` counts images instead: it keeps the blobs of the images of a repository that a tag most recently pointed at. With `keep_latest: 2` and one tag such as `beta` in use, that is the image `beta` points at now and the one it pointed at before. The count is per repository, not per tag, so with several tags in use set it at least that high. A blob that a kept image shares with an older one stays. + +To know which blobs an image needs, the proxy records what each manifest it serves references, in the `image_manifests` table, while `keep_latest` is set and a retention rule applies to `oci`. This has a few consequences: + +- An image only counts once its tag was pulled or mirrored with `keep_latest` set, and its blobs are known once its platform manifest was pulled too. Until then they expire by their rule as before. +- An image pulled only by digest is not counted and not kept. +- Homebrew bottles and Helm charts pulled over OCI are images here: the newest versions of a bottle or chart are those whose tags were last asked for. Generic retention is opt-in because release assets may disappear upstream and cannot then be downloaded again after eviction. An existing `default` never starts evicting generic downloads. Set `ecosystems.generic: "30d"` to cover all generic artifacts, or configure only selected GitHub repositories: diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 500ad0d8..8a2e0574 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", "metadata_snapshots", "projects", "project_components", "migrations", "schema_info"} + tables := []string{"artifacts", "pending_deletes", "versions", "packages", "vulnerabilities", "metadata_cache", "metadata_dates", "image_manifests", "job_runs", "fetch_leases", "upstream_holds", "metadata_snapshots", "projects", "project_components", "migrations", "schema_info"} for _, table := range tables { _, _ = db.Exec("DROP TABLE IF EXISTS " + table + " CASCADE") } diff --git a/internal/database/image_manifests.go b/internal/database/image_manifests.go new file mode 100644 index 00000000..ae729909 --- /dev/null +++ b/internal/database/image_manifests.go @@ -0,0 +1,170 @@ +package database + +import ( + "database/sql" + "errors" + "fmt" + "strings" + "time" + + "github.com/jmoiron/sqlx" +) + +const ( + // imageWalkDepth is how many levels of references NewestImageDigests + // follows below an image: nested indexes, a platform manifest, its blobs. + imageWalkDepth = 4 + // imageWalkBatch is how many digests one query looks up. + imageWalkBatch = 500 +) + +// ImageManifest is what one container image manifest of a repository +// references. +type ImageManifest struct { + // PackagePURL is the repository as its blobs are cached. + PackagePURL string + Digest string + // Refs are the digests of the manifests or blobs it references. + Refs []string + // TaggedAt is when a tag was last found to point at the manifest, or + // zero for one only asked for by digest. + TaggedAt time.Time +} + +type imageManifestRow struct { + Digest string `db:"digest"` + Refs string `db:"refs"` +} + +// RecordImageManifest stores a manifest's references. For one recorded +// before it only moves the tag time forward. +func (db *DB) RecordImageManifest(m ImageManifest) error { + tagged := int64(0) + if !m.TaggedAt.IsZero() { + tagged = m.TaggedAt.UnixMilli() + } + _, err := db.Exec(db.Rebind(` + INSERT INTO image_manifests (package_purl, digest, refs, tagged_at, recorded_at) VALUES (?, ?, ?, ?, ?) + ON CONFLICT (package_purl, digest) DO UPDATE SET tagged_at = excluded.tagged_at + WHERE image_manifests.tagged_at < excluded.tagged_at + `), m.PackagePURL, m.Digest, strings.Join(m.Refs, "\n"), tagged, time.Now().UnixMilli()) + return err +} + +// ImageManifestTaggedAt returns when a tag was last found to point at a +// manifest, zero if none was, and whether the manifest is recorded at all. +func (db *DB) ImageManifestTaggedAt(packagePURL, digest string) (time.Time, bool, error) { + var tagged int64 + err := db.Get(&tagged, db.Rebind(` + SELECT tagged_at FROM image_manifests WHERE package_purl = ? AND digest = ? + `), packagePURL, digest) + if errors.Is(err, sql.ErrNoRows) { + return time.Time{}, false, nil + } + if err != nil || tagged == 0 { + return time.Time{}, err == nil, err + } + return time.UnixMilli(tagged), true, nil +} + +// NewestImageDigests returns the digests of the n images of a repository +// that a tag was most recently found to point at, and of every manifest and +// blob they reference. +func (db *DB) NewestImageDigests(packagePURL string, n int) (map[string]struct{}, error) { + digests, _, err := db.walkNewestImages(packagePURL, n) + return digests, err +} + +// ImageManifestRepositories returns the repositories that have recorded +// manifests. +func (db *DB) ImageManifestRepositories() ([]string, error) { + var repositories []string + err := db.Select(&repositories, `SELECT DISTINCT package_purl FROM image_manifests`) + return repositories, err +} + +// PruneImageManifests removes the manifests of a repository that its n +// newest images do not reference and that were recorded before the given +// time. The time leaves a manifest recorded shortly before the one that +// references it in place. +func (db *DB) PruneImageManifests(packagePURL string, n int, before time.Time) error { + _, manifests, err := db.walkNewestImages(packagePURL, n) + if err != nil { + return err + } + query := `DELETE FROM image_manifests WHERE package_purl = ? AND recorded_at < ?` + args := []any{packagePURL, before.UnixMilli()} + if len(manifests) > 0 { + query, args, err = sqlx.In(query+` AND digest NOT IN (?)`, packagePURL, before.UnixMilli(), manifests) + if err != nil { + return fmt.Errorf("building prune query: %w", err) + } + } + _, err = db.Exec(db.Rebind(query), args...) + return err +} + +// walkNewestImages returns every digest the n newest images of a repository +// reference, themselves included, and those among them that are recorded +// manifests. +func (db *DB) walkNewestImages(packagePURL string, n int) (map[string]struct{}, []string, error) { + digests := map[string]struct{}{} + if n <= 0 { + return digests, nil, nil + } + var rows []imageManifestRow + err := db.Select(&rows, db.Rebind(` + SELECT digest, refs FROM image_manifests + WHERE package_purl = ? AND tagged_at > 0 + ORDER BY tagged_at DESC, digest + LIMIT ? + `), packagePURL, n) + if err != nil { + return nil, nil, err + } + + var manifests []string + for _, row := range rows { + digests[row.Digest] = struct{}{} + } + for depth := 0; len(rows) > 0; depth++ { + var next []string + for _, row := range rows { + manifests = append(manifests, row.Digest) + for _, ref := range strings.Split(row.Refs, "\n") { + if _, seen := digests[ref]; seen || ref == "" { + continue + } + digests[ref] = struct{}{} + next = append(next, ref) + } + } + if depth == imageWalkDepth { + break + } + if rows, err = db.imageManifestRows(packagePURL, next); err != nil { + return nil, nil, err + } + } + return digests, manifests, nil +} + +// imageManifestRows returns the recorded manifests among digests. +func (db *DB) imageManifestRows(packagePURL string, digests []string) ([]imageManifestRow, error) { + var rows []imageManifestRow + for start := 0; start < len(digests); start += imageWalkBatch { + batch := digests[start:min(start+imageWalkBatch, len(digests))] + query, args, err := sqlx.In(` + SELECT digest, refs FROM image_manifests WHERE package_purl = ? AND digest IN (?) + `, packagePURL, batch) + if err != nil { + return nil, fmt.Errorf("building image manifest query: %w", err) + } + var found []imageManifestRow + if err := db.Select(&found, db.Rebind(query), args...); err != nil { + return nil, err + } + rows = append(rows, found...) + } + return rows, nil +} diff --git a/internal/database/image_manifests_test.go b/internal/database/image_manifests_test.go new file mode 100644 index 00000000..34777c27 --- /dev/null +++ b/internal/database/image_manifests_test.go @@ -0,0 +1,190 @@ +package database + +import ( + "slices" + "testing" + "time" +) + +const testImageRepo = "pkg:oci/library%2Fnginx" + +func recordImage(t *testing.T, db *DB, repo, digest string, taggedAt time.Time, refs ...string) { + t.Helper() + err := db.RecordImageManifest(ImageManifest{PackagePURL: repo, Digest: digest, Refs: refs, TaggedAt: taggedAt}) + if err != nil { + t.Fatalf("RecordImageManifest(%s): %v", digest, err) + } +} + +func newestImageDigests(t *testing.T, db *DB, repo string, n int) []string { + t.Helper() + set, err := db.NewestImageDigests(repo, n) + if err != nil { + t.Fatalf("NewestImageDigests: %v", err) + } + digests := make([]string, 0, len(set)) + for d := range set { + digests = append(digests, d) + } + slices.Sort(digests) + return digests +} + +// The newest images are the manifests a tag was last found to point at. What +// they reference is followed through the manifests recorded for the same +// repository, so an index leads to its platform manifests and their blobs. +func TestNewestImageDigestsFollowsReferences(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + recordImage(t, db, testImageRepo, "sha256:index-old", now.Add(-2*time.Hour), "sha256:arm-old", "sha256:amd-old") + recordImage(t, db, testImageRepo, "sha256:arm-old", time.Time{}, "sha256:config-old", "sha256:layer-shared", "sha256:layer-old") + recordImage(t, db, testImageRepo, "sha256:index-new", now.Add(-time.Hour), "sha256:arm-new", "sha256:amd-new") + recordImage(t, db, testImageRepo, "sha256:arm-new", time.Time{}, "sha256:config-new", "sha256:layer-shared", "sha256:layer-new") + // Another repository's image is newer still and must not count here. + recordImage(t, db, "pkg:oci/library%2Fredis", "sha256:redis", now, "sha256:layer-redis") + + want := []string{"sha256:amd-new", "sha256:arm-new", "sha256:config-new", "sha256:index-new", "sha256:layer-new", "sha256:layer-shared"} + if got := newestImageDigests(t, db, testImageRepo, 1); !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want %v", got, want) + } + if got := newestImageDigests(t, db, testImageRepo, 2); len(got) != 11 { + t.Errorf("newest 2 = %v, want all 11 digests of both images", got) + } + if got := newestImageDigests(t, db, testImageRepo, 0); len(got) != 0 { + t.Errorf("newest 0 = %v, want none", got) + } + }) +} + +// A manifest only asked for by digest is not an image of its own. +func TestNewestImageDigestsIgnoresUntaggedManifests(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + recordImage(t, db, testImageRepo, "sha256:pinned", time.Time{}, "sha256:layer-pinned") + if got := newestImageDigests(t, db, testImageRepo, 5); len(got) != 0 { + t.Errorf("newest = %v, want none", got) + } + }) +} + +// Recording a manifest again moves its tag time forward only, and a request +// by digest does not make it lose the time a tag gave it. +func TestRecordImageManifestKeepsLatestTagTime(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now().Truncate(time.Millisecond) + recordImage(t, db, testImageRepo, "sha256:a", now.Add(-time.Hour), "sha256:layer-a") + recordImage(t, db, testImageRepo, "sha256:b", now.Add(-30*time.Minute), "sha256:layer-b") + + if at, ok, err := db.ImageManifestTaggedAt(testImageRepo, "sha256:missing"); err != nil || ok || !at.IsZero() { + t.Fatalf("ImageManifestTaggedAt(missing) = %v, %v, %v; want zero, false, nil", at, ok, err) + } + + recordImage(t, db, testImageRepo, "sha256:a", time.Time{}, "sha256:layer-a") + recordImage(t, db, testImageRepo, "sha256:a", now.Add(-2*time.Hour), "sha256:layer-a") + at, ok, err := db.ImageManifestTaggedAt(testImageRepo, "sha256:a") + if err != nil || !ok || !at.Equal(now.Add(-time.Hour)) { + t.Fatalf("ImageManifestTaggedAt(a) = %v, %v, %v; want %v, true, nil", at, ok, err, now.Add(-time.Hour)) + } + if got, want := newestImageDigests(t, db, testImageRepo, 1), []string{"sha256:b", "sha256:layer-b"}; !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want %v", got, want) + } + + // A tag found on a again makes it the newest image. + recordImage(t, db, testImageRepo, "sha256:a", now, "sha256:layer-a") + if got, want := newestImageDigests(t, db, testImageRepo, 1), []string{"sha256:a", "sha256:layer-a"}; !slices.Equal(got, want) { + t.Errorf("newest 1 after retag = %v, want %v", got, want) + } + }) +} + +// A manifest that references itself, or a chain longer than any real index, +// ends the walk instead of looping. +func TestNewestImageDigestsStopsOnCycles(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + recordImage(t, db, testImageRepo, "sha256:a", now, "sha256:b") + recordImage(t, db, testImageRepo, "sha256:b", time.Time{}, "sha256:a", "sha256:layer") + if got, want := newestImageDigests(t, db, testImageRepo, 1), []string{"sha256:a", "sha256:b", "sha256:layer"}; !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want %v", got, want) + } + }) +} + +// Pruning drops what the newest images no longer reference, once it is older +// than the grace time, so a platform manifest recorded just after its index +// is not lost. +func TestPruneImageManifests(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + recordImage(t, db, testImageRepo, "sha256:index-old", now.Add(-2*time.Hour), "sha256:arm-old") + recordImage(t, db, testImageRepo, "sha256:arm-old", time.Time{}, "sha256:layer-old") + recordImage(t, db, testImageRepo, "sha256:index-new", now.Add(-time.Hour), "sha256:arm-new") + recordImage(t, db, testImageRepo, "sha256:arm-new", time.Time{}, "sha256:layer-new") + recordImage(t, db, testImageRepo, "sha256:pinned", time.Time{}, "sha256:layer-pinned") + recordImage(t, db, "pkg:oci/library%2Fredis", "sha256:redis-old", now.Add(-3*time.Hour), "sha256:layer-redis") + + // Everything was recorded just now, so nothing is old enough yet. + if err := db.PruneImageManifests(testImageRepo, 1, now.Add(-time.Minute)); err != nil { + t.Fatalf("PruneImageManifests: %v", err) + } + if got := newestImageDigests(t, db, testImageRepo, 2); len(got) != 6 { + t.Fatalf("after early prune, newest 2 = %v, want both images", got) + } + + if err := db.PruneImageManifests(testImageRepo, 1, now.Add(time.Minute)); err != nil { + t.Fatalf("PruneImageManifests: %v", err) + } + want := []string{"sha256:arm-new", "sha256:index-new", "sha256:layer-new"} + if got := newestImageDigests(t, db, testImageRepo, 2); !slices.Equal(got, want) { + t.Errorf("after prune, newest 2 = %v, want %v", got, want) + } + for _, digest := range []string{"sha256:index-old", "sha256:arm-old", "sha256:pinned"} { + if _, ok, err := db.ImageManifestTaggedAt(testImageRepo, digest); err != nil || ok { + t.Errorf("%s still recorded after prune (ok=%v, err=%v)", digest, ok, err) + } + } + if got := newestImageDigests(t, db, "pkg:oci/library%2Fredis", 1); len(got) != 2 { + t.Errorf("other repository after prune = %v, want its image kept", got) + } + }) +} + +// A database from before the table existed gets it from the migration. +func TestMigrateAddsImageManifests(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + if _, err := db.Exec("DROP TABLE image_manifests"); err != nil { + t.Fatal(err) + } + if _, err := db.Exec("DELETE FROM migrations WHERE name = '024_add_image_manifests'"); err != nil { + t.Fatal(err) + } + if err := db.MigrateSchema(); err != nil { + t.Fatalf("MigrateSchema: %v", err) + } + recordImage(t, db, testImageRepo, "sha256:a", time.Now(), "sha256:layer-a") + if got := newestImageDigests(t, db, testImageRepo, 1); len(got) != 2 { + t.Errorf("newest 1 after migration = %v, want the image", got) + } + }) +} + +// The repositories with records are listed once each, for pruning. +func TestImageManifestRepositories(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + if got, err := db.ImageManifestRepositories(); err != nil || len(got) != 0 { + t.Fatalf("ImageManifestRepositories() = %v, %v; want none", got, err) + } + const redis = "pkg:oci/library%2Fredis" + recordImage(t, db, testImageRepo, "sha256:a", time.Now(), "sha256:layer-a") + recordImage(t, db, testImageRepo, "sha256:b", time.Time{}, "sha256:layer-b") + recordImage(t, db, redis, "sha256:c", time.Time{}, "sha256:layer-c") + + got, err := db.ImageManifestRepositories() + if err != nil { + t.Fatalf("ImageManifestRepositories: %v", err) + } + slices.Sort(got) + if want := []string{testImageRepo, redis}; !slices.Equal(got, want) { + t.Errorf("ImageManifestRepositories() = %v, want %v", got, want) + } + }) +} diff --git a/internal/database/schema.go b/internal/database/schema.go index bca37bd4..1c298fab 100644 --- a/internal/database/schema.go +++ b/internal/database/schema.go @@ -151,6 +151,15 @@ CREATE TABLE IF NOT EXISTS metadata_dates ( PRIMARY KEY (ecosystem, proxy_url, name) ); +CREATE TABLE IF NOT EXISTS image_manifests ( + package_purl TEXT NOT NULL, + digest TEXT NOT NULL, + refs TEXT NOT NULL, + tagged_at INTEGER NOT NULL DEFAULT 0, + recorded_at INTEGER NOT NULL, + PRIMARY KEY (package_purl, digest) +); + CREATE TABLE IF NOT EXISTS upstream_holds ( host TEXT NOT NULL PRIMARY KEY, until_ms INTEGER NOT NULL @@ -330,6 +339,15 @@ CREATE TABLE IF NOT EXISTS metadata_dates ( PRIMARY KEY (ecosystem, proxy_url, name) ); +CREATE TABLE IF NOT EXISTS image_manifests ( + package_purl TEXT NOT NULL, + digest TEXT NOT NULL, + refs TEXT NOT NULL, + tagged_at BIGINT NOT NULL DEFAULT 0, + recorded_at BIGINT NOT NULL, + PRIMARY KEY (package_purl, digest) +); + CREATE TABLE IF NOT EXISTS upstream_holds ( host TEXT NOT NULL PRIMARY KEY, until_ms BIGINT NOT NULL @@ -546,6 +564,7 @@ var migrations = []migration{ {"021_add_projects", migrateAddProjects}, {"022_add_artifact_mirrored_at", migrateAddArtifactMirroredAt}, {"023_add_metadata_upstream_url", migrateAddMetadataUpstreamURL}, + {"024_add_image_manifests", migrateAddImageManifests}, } // isTableNotFound returns true if the error indicates a missing table. @@ -1030,6 +1049,27 @@ func migrateAddMetadataDates(db *DB) error { return nil } +// migrateAddImageManifests creates the table of what container image +// manifests reference, which retention's keep_latest reads. +func migrateAddImageManifests(db *DB) error { + millis := "INTEGER" + if db.dialect == DialectPostgres { + millis = "BIGINT" + } + query := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS image_manifests ( + package_purl TEXT NOT NULL, + digest TEXT NOT NULL, + refs TEXT NOT NULL, + tagged_at %s NOT NULL DEFAULT 0, + recorded_at %s NOT NULL, + PRIMARY KEY (package_purl, digest) + )`, millis, millis) + if _, err := db.Exec(query); err != nil { + return fmt.Errorf("creating image_manifests table: %w", err) + } + return nil +} + // EnsureMetadataCacheTable creates the metadata_cache table if it doesn't exist. func (db *DB) EnsureMetadataCacheTable() error { has, err := db.HasTable("metadata_cache") diff --git a/internal/handler/container.go b/internal/handler/container.go index 40f5d0f2..c0453595 100644 --- a/internal/handler/container.go +++ b/internal/handler/container.go @@ -70,7 +70,7 @@ func isDockerHubKey(key string) bool { // Package rules write the repository name as a path, pkg:oci/library/nginx, // and are mapped to that form. func init() { - retention.Default.Register(retention.Spec{Key: "oci", CanonicalPackage: ociRetentionKey}) + retention.Default.Register(retention.Spec{Key: "oci", CanonicalPackage: ociRetentionKey, KeepLatestImages: true}) } // ociRetentionKey maps a package rule to the stored package PURL. It refuses diff --git a/internal/handler/container_image_refs.go b/internal/handler/container_image_refs.go new file mode 100644 index 00000000..fbe523cd --- /dev/null +++ b/internal/handler/container_image_refs.go @@ -0,0 +1,97 @@ +package handler + +import ( + "encoding/json" + "regexp" + "slices" + "time" + + "github.com/git-pkgs/proxy/internal/database" +) + +// maxImageManifestRefs is how many references of one manifest are recorded. +// A manifest with more is not recorded. +const maxImageManifestRefs = 4096 + +// imageSubjectTagPattern matches the tags that hold what refers to an image +// rather than an image of their own: the referrers tag schema, and the +// sha256-.sig, .att and .sbom tags of cosign. +var imageSubjectTagPattern = regexp.MustCompile(`^sha256-[a-f0-9]{64}(\.[a-z0-9]+)?$`) + +// recordImageManifest notes which manifests and blobs a served manifest +// references, and for a tag that it pointed at the manifest, so retention +// can keep the blobs of the newest images of a repository. It does nothing +// unless storage.retention.keep_latest covers container images. The +// retention sweep drops the records of images that are no longer among the +// newest. +func (h *ContainerHandler) recordImageManifest(cacheName, reference string, manifest *cachedContainerManifest, stale bool) { + if h.proxy.KeepLatestImages <= 0 || h.proxy.DB == nil { + return + } + // A manifest asked for by digest is recorded under that digest, the one + // the manifest referencing it names. + digest, tagged := reference, time.Time{} + if !manifestDigestReferencePattern.MatchString(reference) { + digest = manifest.contentDigest + if !imageSubjectTagPattern.MatchString(reference) { + // The time upstream last confirmed the tag, which a stale copy + // served during an outage does not move. + if tagged = manifest.fetchedAt; tagged.IsZero() && !stale { + tagged = time.Now() + } + } + } + if !mirrorDigestPattern.MatchString(digest) { + return + } + + packagePURL := canonicalPackagePURL("oci", cacheName) + known, recorded, err := h.proxy.DB.ImageManifestTaggedAt(packagePURL, digest) + if err != nil { + h.proxy.Logger.Warn("failed to read image manifest record", "package_purl", packagePURL, "digest", digest, "error", err) + return + } + // Tag times are stored in milliseconds. + if recorded && tagged.UnixMilli() <= known.UnixMilli() { + return + } + var refs []string + if !recorded { + // The answer to a HEAD for a manifest that is not cached has no + // body to read references from. + if refs = imageManifestRefs(manifest.body); len(refs) == 0 { + return + } + } + record := database.ImageManifest{PackagePURL: packagePURL, Digest: digest, Refs: refs, TaggedAt: tagged} + if err := h.proxy.DB.RecordImageManifest(record); err != nil { + h.proxy.Logger.Warn("failed to record image manifest", "package_purl", packagePURL, "digest", digest, "error", err) + } +} + +// imageManifestRefs returns the digests of the manifests an index lists, or +// of the config and layers of an image manifest. It returns nil for a +// document that names none, or more than maxImageManifestRefs. +func imageManifestRefs(body []byte) []string { + var manifest ociManifest + if len(body) == 0 || json.Unmarshal(body, &manifest) != nil { + return nil + } + descriptors := slices.Concat(manifest.Manifests, manifest.Layers) + if manifest.Config != nil { + descriptors = append(descriptors, *manifest.Config) + } + seen := make(map[string]struct{}, len(descriptors)) + var refs []string + for _, descriptor := range descriptors { + if _, dup := seen[descriptor.Digest]; dup || !mirrorDigestPattern.MatchString(descriptor.Digest) { + continue + } + seen[descriptor.Digest] = struct{}{} + refs = append(refs, descriptor.Digest) + } + if len(refs) > maxImageManifestRefs { + return nil + } + return refs +} diff --git a/internal/handler/container_image_refs_test.go b/internal/handler/container_image_refs_test.go new file mode 100644 index 00000000..9b2dac6b --- /dev/null +++ b/internal/handler/container_image_refs_test.go @@ -0,0 +1,192 @@ +package handler + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +const imageRefsRepo = "team/app" + +func imageRefsHandler(t *testing.T, keepLatest int) (*ContainerHandler, *fakeImageRegistry, fakeImage) { + t.Helper() + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + proxy.KeepLatestImages = keepLatest + return NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL), registry, image +} + +func pullImageManifest(t *testing.T, h *ContainerHandler, reference string) { + t.Helper() + if response := pull(t, h, "/"+imageRefsRepo+"/manifests/"+reference, containerdManifestAccept); response.Code != http.StatusOK { + t.Fatalf("manifest %s: status %d: %s", reference, response.Code, response.Body.String()) + } +} + +// newestImageDigests returns what the newest image of the repository needs. +func newestImageDigests(t *testing.T, h *ContainerHandler) map[string]struct{} { + t.Helper() + digests, err := h.proxy.DB.NewestImageDigests(canonicalPackagePURL("oci", imageRefsRepo), 1) + if err != nil { + t.Fatalf("NewestImageDigests: %v", err) + } + return digests +} + +func imageTaggedAt(t *testing.T, h *ContainerHandler, digest string) (time.Time, bool) { + t.Helper() + at, ok, err := h.proxy.DB.ImageManifestTaggedAt(canonicalPackagePURL("oci", imageRefsRepo), digest) + if err != nil { + t.Fatalf("ImageManifestTaggedAt: %v", err) + } + return at, ok +} + +// moveTag points a tag of the fake registry at another manifest. +func (r *fakeImageRegistry) moveTag(tag, digest string) { + r.mu.Lock() + defer r.mu.Unlock() + r.manifests["/v2/"+imageRefsRepo+"/manifests/"+tag] = r.manifests["/v2/"+imageRefsRepo+"/manifests/"+digest] +} + +// A pull by tag makes the manifest an image of the repository, and the +// manifests pulled by digest after it say which blobs it needs. +func TestContainerPullRecordsWhatAnImageReferences(t *testing.T) { + h, _, image := imageRefsHandler(t, 1) + + pullImageManifest(t, h, "multi") + if at, ok := imageTaggedAt(t, h, image.indexDigest); !ok || at.IsZero() { + t.Fatalf("index after a pull by tag: recorded=%v tagged=%v, want both", ok, at) + } + pullImageManifest(t, h, image.arm64Digest) + if at, ok := imageTaggedAt(t, h, image.arm64Digest); !ok || !at.IsZero() { + t.Fatalf("platform manifest after a pull by digest: recorded=%v tagged=%v, want recorded and untagged", ok, at) + } + + digests := newestImageDigests(t, h) + for name, digest := range map[string]string{ + "index": image.indexDigest, "arm64 manifest": image.arm64Digest, + "amd64 manifest": image.amd64Digest, "arm64 layer": image.arm64Layer, + } { + if _, ok := digests[digest]; !ok { + t.Errorf("%s %s is not among the newest image's digests", name, digest) + } + } + // The amd64 manifest was never pulled, so its blobs are unknown. + if _, ok := digests[image.amd64Layer]; ok { + t.Errorf("amd64 layer is among the digests though its manifest was not pulled") + } +} + +// Without keep_latest nothing is recorded. +func TestContainerPullRecordsNothingWithoutKeepLatest(t *testing.T) { + h, _, image := imageRefsHandler(t, 0) + pullImageManifest(t, h, "multi") + pullImageManifest(t, h, image.arm64Digest) + if _, ok := imageTaggedAt(t, h, image.indexDigest); ok { + t.Errorf("index recorded though keep_latest is off") + } + if _, ok := imageTaggedAt(t, h, image.arm64Digest); ok { + t.Errorf("platform manifest recorded though keep_latest is off") + } +} + +// Referrers and cosign tags hold what refers to an image. They are recorded +// like a pull by digest, so they don't count as images of their own. +func TestContainerPullOfReferrersTagIsNotAnImage(t *testing.T) { + h, registry, image := imageRefsHandler(t, 1) + subject := "sha256-" + image.indexDigest[len("sha256:"):] + for _, tag := range []string{subject, subject + ".sig", subject + ".att"} { + registry.moveTag(tag, image.singleDigest) + pullImageManifest(t, h, tag) + } + if at, ok := imageTaggedAt(t, h, image.singleDigest); !ok || !at.IsZero() { + t.Errorf("manifest behind referrers tags: recorded=%v tagged=%v, want recorded and untagged", ok, at) + } +} + +// A tag that is asked for again moves the image's tag time forward, also +// when the client only sends a HEAD, as containerd does for an image it has. +func TestContainerPullAgainMovesTagTimeForward(t *testing.T) { + for _, method := range []string{http.MethodGet, http.MethodHead} { + t.Run(method, func(t *testing.T) { + h, _, image := imageRefsHandler(t, 1) + pullImageManifest(t, h, "single") + db := h.proxy.DB + if _, err := db.Exec(db.Rebind("UPDATE image_manifests SET tagged_at = ?"), 1000); err != nil { + t.Fatal(err) + } + + request := httptest.NewRequest(method, "/"+imageRefsRepo+"/manifests/single", nil) + request.Header.Set("Accept", containerdManifestAccept) + response := httptest.NewRecorder() + h.Routes().ServeHTTP(response, request) + if response.Code != http.StatusOK { + t.Fatalf("status %d", response.Code) + } + if at, _ := imageTaggedAt(t, h, image.singleDigest); time.Since(at) > time.Minute { + t.Errorf("tag time after the second request = %v, want now", at) + } + }) + } +} + +// When a tag moves, the image it points at now is the newest, and the one +// it pointed at before no longer is. +func TestContainerMovedTagMakesTheNewImageNewest(t *testing.T) { + h, registry, image := imageRefsHandler(t, 1) + registry.moveTag("stable", image.singleDigest) + pullImageManifest(t, h, "stable") + if _, ok := newestImageDigests(t, h)[image.singleLayer]; !ok { + t.Fatalf("layer of the first image is not kept") + } + db := h.proxy.DB + if _, err := db.Exec(db.Rebind("UPDATE image_manifests SET tagged_at = ?"), 1000); err != nil { + t.Fatal(err) + } + + registry.moveTag("stable", image.indexDigest) + pullImageManifest(t, h, "stable") + pullImageManifest(t, h, image.arm64Digest) + + digests := newestImageDigests(t, h) + if _, ok := digests[image.arm64Layer]; !ok { + t.Errorf("layer of the image the tag moved to is not kept") + } + if _, ok := digests[image.singleLayer]; ok { + t.Errorf("layer of the image the tag moved away from is still kept") + } +} + +// A scheduled mirror run resolves tags itself, and records them like a pull. +func TestMirrorImageRecordsWhatAnImageReferences(t *testing.T) { + h, _, image := imageRefsHandler(t, 1) + platform, err := ParsePlatform("linux/arm64") + if err != nil { + t.Fatal(err) + } + ref := ImageRef{Name: imageRefsRepo, References: []string{"multi"}} + if _, err := h.MirrorImage(context.Background(), ref, []Platform{platform}); err != nil { + t.Fatalf("MirrorImage: %v", err) + } + if at, ok := imageTaggedAt(t, h, image.indexDigest); !ok || at.IsZero() { + t.Fatalf("index after a mirror run: recorded=%v tagged=%v, want both", ok, at) + } + if _, ok := newestImageDigests(t, h)[image.arm64Layer]; !ok { + t.Errorf("arm64 layer is not among the mirrored image's digests") + } + + // A later run that finds the tag unchanged still moves its time forward. + db := h.proxy.DB + if _, err := db.Exec(db.Rebind("UPDATE image_manifests SET tagged_at = ? WHERE tagged_at > 0"), 1000); err != nil { + t.Fatal(err) + } + if _, err := h.MirrorImage(context.Background(), ref, []Platform{platform}); err != nil { + t.Fatalf("second MirrorImage: %v", err) + } + if at, _ := imageTaggedAt(t, h, image.indexDigest); time.Since(at) > time.Minute { + t.Errorf("tag time after the second run = %v, want now", at) + } +} diff --git a/internal/handler/container_manifest.go b/internal/handler/container_manifest.go index 6d129d0a..fa857f37 100644 --- a/internal/handler/container_manifest.go +++ b/internal/handler/container_manifest.go @@ -63,14 +63,14 @@ func (h *ContainerHandler) serveManifest(w http.ResponseWriter, r *http.Request, } cached, _ := h.loadContainerManifestForAccept(r.Context(), registryURL, name, reference, req.accept, req.cacheAccept) if cached != nil && (req.immutable() || h.containerManifestFresh(cached)) { - h.writeManifest(w, r, cacheName, cached, false) + h.writeManifest(w, r, cacheName, reference, cached, false) return } outcome := h.coalescedManifest(r.Context(), req) switch { case outcome.manifest != nil: - h.writeManifest(w, r, cacheName, outcome.manifest, false) + h.writeManifest(w, r, cacheName, reference, outcome.manifest, false) case outcome.relay != nil: outcome.relay.write(h.proxy, w, r) case outcome.digestInvalid: @@ -83,7 +83,7 @@ func (h *ContainerHandler) serveManifest(w http.ResponseWriter, r *http.Request, func (h *ContainerHandler) serveStaleManifestOrError(w http.ResponseWriter, r *http.Request, cacheName, reference string, cached *cachedContainerManifest, err error) { if cached != nil { h.proxy.Logger.Warn("upstream manifest fetch failed, serving stale cache", "error", err) - h.writeManifest(w, r, cacheName, cached, true) + h.writeManifest(w, r, cacheName, reference, cached, true) return } if referrersTagPattern.MatchString(reference) { @@ -109,11 +109,13 @@ func (h *ContainerHandler) manifestDenied(w http.ResponseWriter, cacheName, refe return true } -// writeManifest writes manifest unless the denylist names its digest. -func (h *ContainerHandler) writeManifest(w http.ResponseWriter, r *http.Request, cacheName string, manifest *cachedContainerManifest, stale bool) { +// writeManifest writes the manifest that reference resolved to, unless the +// denylist names its digest. +func (h *ContainerHandler) writeManifest(w http.ResponseWriter, r *http.Request, cacheName, reference string, manifest *cachedContainerManifest, stale bool) { if h.manifestDenied(w, cacheName, manifest.contentDigest) { return } + h.recordImageManifest(cacheName, reference, manifest, stale) writeContainerManifest(w, r, manifest, stale) } diff --git a/internal/handler/container_mirror.go b/internal/handler/container_mirror.go index f19ac736..19a52fc7 100644 --- a/internal/handler/container_mirror.go +++ b/internal/handler/container_mirror.go @@ -255,6 +255,8 @@ func (m *imageMirror) mirrorTag(ctx context.Context, tag string) (string, error) return digest, nil } m.h.storeContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, tag, accept, cacheAccept, manifest) + // The walk recorded the manifest as one asked for by digest. + m.h.recordImageManifest(m.cacheName, tag, manifest, false) return digest, nil } diff --git a/internal/handler/handler.go b/internal/handler/handler.go index d9284e24..700325b4 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -248,6 +248,12 @@ type Proxy struct { DirectServeBaseURL string HTTPClient *http.Client + // KeepLatestImages is storage.retention.keep_latest where a retention + // rule applies to container images. Above zero, the container handler + // records what each served manifest references, so retention can keep + // the blobs of that many images per repository. + KeepLatestImages int + // RateLimits is the gate under the upstream clients. Artifact fetches // check it before calling Fetcher, so held hosts never reach its breaker. RateLimits UpstreamHolds diff --git a/internal/handler/profile.go b/internal/handler/profile.go index ceff0a30..cc8bb8e7 100644 --- a/internal/handler/profile.go +++ b/internal/handler/profile.go @@ -14,7 +14,7 @@ func (p *Proxy) WithCooldown(policy CooldownPolicy, shortest time.Duration) *Pro GradleReadOnly: p.GradleReadOnly, HostedEnabled: p.HostedEnabled, GradleMaxUploadSize: p.GradleMaxUploadSize, NPMFullMetadata: p.NPMFullMetadata, DirectServe: p.DirectServe, DirectServeTTL: p.DirectServeTTL, - DirectServeBaseURL: p.DirectServeBaseURL, HTTPClient: p.HTTPClient, + DirectServeBaseURL: p.DirectServeBaseURL, HTTPClient: p.HTTPClient, KeepLatestImages: p.KeepLatestImages, RateLimits: p.RateLimits, HostedNPMScopes: p.HostedNPMScopes, HostedComposerVendors: p.HostedComposerVendors, StreamArtifacts: p.StreamArtifacts, Scanners: p.Scanners, ScanSigningKey: p.ScanSigningKey, ScanFetchBaseURL: p.ScanFetchBaseURL, diff --git a/internal/handler/profile_test.go b/internal/handler/profile_test.go index ca163ee6..8729672f 100644 --- a/internal/handler/profile_test.go +++ b/internal/handler/profile_test.go @@ -35,6 +35,7 @@ func TestWithCooldownCopiesProxyFields(t *testing.T) { p.DirectServe = true p.DirectServeTTL = time.Minute p.DirectServeBaseURL = "https://storage.example.com" + p.KeepLatestImages = 2 p.RateLimits = registryclient.NewRateLimitGate(http.DefaultTransport) p.HostedNPMScopes = []string{"@example"} p.HostedComposerVendors = []string{"example"} diff --git a/internal/retention/keep_latest_test.go b/internal/retention/keep_latest_test.go index 5d26eabe..c5fcfa8d 100644 --- a/internal/retention/keep_latest_test.go +++ b/internal/retention/keep_latest_test.go @@ -56,8 +56,9 @@ func TestRegistryKeepLatestKeys(t *testing.T) { reg.Register(Spec{Key: "npm", CompareVersions: VersionOrder("npm")}) reg.Register(Spec{Key: "gem"}) reg.Register(Spec{Key: "cargo", CompareVersions: VersionOrder("cargo")}) + reg.Register(Spec{Key: "oci", KeepLatestImages: true}) - if got := reg.KeepLatestKeys(); !slices.Equal(got, []string{"cargo", "npm"}) { - t.Errorf("KeepLatestKeys() = %v, want [cargo npm]", got) + if got := reg.KeepLatestKeys(); !slices.Equal(got, []string{"cargo", "npm", "oci"}) { + t.Errorf("KeepLatestKeys() = %v, want [cargo npm oci]", got) } } diff --git a/internal/retention/retention.go b/internal/retention/retention.go index 8379dc5d..4a1acd8e 100644 --- a/internal/retention/retention.go +++ b/internal/retention/retention.go @@ -63,6 +63,12 @@ type Spec struct { // applies to the ecosystem. Nil means the stored versions cannot be // ordered, as with digests and hashes. CompareVersions func(a, b string) int + + // KeepLatestImages marks an ecosystem that stores the blobs of container + // images under their digests and records which of them each image + // references. Rules.KeepLatest then keeps the blobs of the newest + // images of a repository. + KeepLatestImages bool } // VersionOrder returns a CompareVersions for an ecosystem whose handler @@ -173,7 +179,7 @@ func (r *Registry) Keys() []string { func (r *Registry) KeepLatestKeys() []string { var keys []string for _, key := range r.Keys() { - if s, _ := r.Lookup(key); s.CompareVersions != nil { + if s, _ := r.Lookup(key); s.CompareVersions != nil || s.KeepLatestImages { keys = append(keys, key) } } @@ -258,7 +264,8 @@ type Rules struct { Sources map[string]map[string]time.Duration // KeepLatest is how many of the highest cached versions of each package // are never evicted by age, in ecosystems whose Spec sets - // CompareVersions. Zero keeps none. + // CompareVersions. Where it sets KeepLatestImages, it is how many of the + // newest images of each repository keep their blobs. Zero keeps none. KeepLatest int } diff --git a/internal/server/retention.go b/internal/server/retention.go index d7798497..5003dfd7 100644 --- a/internal/server/retention.go +++ b/internal/server/retention.go @@ -4,8 +4,10 @@ import ( "context" "log/slog" "slices" + "strings" "time" + "github.com/git-pkgs/proxy/internal/config" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/metrics" "github.com/git-pkgs/proxy/internal/retention" @@ -15,6 +17,11 @@ import ( // retentionWindow is how many artifact ids one query scans. const retentionWindow = 5000 +// imageRecordGrace is how long a manifest record is safe from pruning, so +// one written while a sweep runs, or shortly before the tag that leads to +// it is recorded, is not lost. +const imageRecordGrace = time.Hour + // retentionMaxClears caps how many artifacts one sweep clears, so a first // sweep over a large, old cache spreads its deletes and the refetches that // follow over several intervals. A variable so tests can lower it. @@ -96,6 +103,8 @@ func sweepRetention(ctx context.Context, db *database.DB, logger *slog.Logger, return } + pruneImageRecords(db, logger, rules.KeepLatest, now) + // Without the buffered hits written, an artifact being served right now // can look expired, so skip this sweep rather than clear it. if err := db.FlushHits(); err != nil { @@ -152,6 +161,41 @@ func sweepRetention(ctx context.Context, db *database.DB, logger *slog.Logger, } } +// keepLatestImages returns storage.retention.keep_latest if a retention rule +// can evict container image blobs, and zero otherwise: without such a rule +// nothing reads what the container handler would record. +func keepLatestImages(cfg *config.Config) int { + rules, err := cfg.RetentionRules(retention.Default) + if err != nil || rules.KeepLatest <= 0 { + return 0 + } + if rules.For("oci", "") > 0 { + return rules.KeepLatest + } + for packagePURL, d := range rules.Packages { + if d > 0 && strings.HasPrefix(packagePURL, "pkg:oci/") { + return rules.KeepLatest + } + } + return 0 +} + +// pruneImageRecords drops the container manifest records that the keep +// newest images of their repository do not reference, which is all of them +// once keep_latest is off. +func pruneImageRecords(db *database.DB, logger *slog.Logger, keep int, now time.Time) { + repositories, err := db.ImageManifestRepositories() + if err != nil { + logger.Warn("retention: failed to list image manifest records", "error", err) + return + } + for _, packagePURL := range repositories { + if err := db.PruneImageManifests(packagePURL, keep, now.Add(-imageRecordGrace)); err != nil { + logger.Warn("retention: failed to prune image manifest records", "package_purl", packagePURL, "error", err) + } + } +} + // clearIfExpired clears one candidate when its own rule says it has expired // and keep does not hold it back, and queues its object for deletion, // reporting whether it did. @@ -180,8 +224,8 @@ func clearIfExpired(db *database.DB, logger *slog.Logger, reg *retention.Registr } // latestKeeper answers, for one sweep, whether an artifact belongs to one of -// the highest cached versions of its package, which keep_latest holds back -// from eviction by age. +// the highest cached versions of its package, or to one of the newest images +// of its repository, which keep_latest holds back from eviction by age. type latestKeeper struct { db *database.DB logger *slog.Logger @@ -197,13 +241,14 @@ func newLatestKeeper(db *database.DB, logger *slog.Logger, reg *retention.Regist } // keeps reports whether c, an artifact of the ecosystem key, is a file of -// one of the highest cached versions of its package. +// one of the highest cached versions of its package, or a blob of one of the +// newest images of its repository. func (k *latestKeeper) keeps(key string, c database.RetentionCandidate) bool { if k.n <= 0 { return false } spec, _ := k.reg.Lookup(key) - if spec.CompareVersions == nil { + if spec.CompareVersions == nil && !spec.KeepLatestImages { return false } kept, ok := k.kept[c.PackagePURL] @@ -212,7 +257,7 @@ func (k *latestKeeper) keeps(key string, c database.RetentionCandidate) bool { if kept, err = k.newest(spec, c.PackagePURL); err != nil { // Not knowing which versions are the highest, keep the artifact // for this sweep rather than evict one of them. - k.logger.Warn("retention: failed to list cached versions, keeping artifact", + k.logger.Warn("retention: failed to list the versions to keep, keeping artifact", "package_purl", c.PackagePURL, "error", err) return true } @@ -223,8 +268,16 @@ func (k *latestKeeper) keeps(key string, c database.RetentionCandidate) bool { } // newest returns the version PURLs of the highest cached versions of a -// package. +// package. For a repository of container images, whose versions are blob +// digests, these are the blobs its newest images reference. func (k *latestKeeper) newest(spec retention.Spec, packagePURL string) (map[string]struct{}, error) { + var imageDigests map[string]struct{} + if spec.KeepLatestImages { + var err error + if imageDigests, err = k.db.NewestImageDigests(packagePURL, k.n); err != nil { + return nil, err + } + } purls, err := k.db.CachedVersionPURLs(packagePURL) if err != nil { return nil, err @@ -242,7 +295,15 @@ func (k *latestKeeper) newest(spec retention.Spec, packagePURL string) (map[stri byVersion[p.Version] = append(byVersion[p.Version], versionPURL) } kept := map[string]struct{}{} - for _, v := range retention.NewestVersions(spec.CompareVersions, versions, k.n) { + if spec.KeepLatestImages { + versions = slices.DeleteFunc(versions, func(digest string) bool { + _, referenced := imageDigests[digest] + return !referenced + }) + } else { + versions = retention.NewestVersions(spec.CompareVersions, versions, k.n) + } + for _, v := range versions { for _, versionPURL := range byVersion[v] { kept[versionPURL] = struct{}{} } diff --git a/internal/server/retention_keep_latest_oci_test.go b/internal/server/retention_keep_latest_oci_test.go new file mode 100644 index 00000000..bf6f96de --- /dev/null +++ b/internal/server/retention_keep_latest_oci_test.go @@ -0,0 +1,301 @@ +package server + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync" + "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/retention" + "github.com/git-pkgs/purl" + "github.com/git-pkgs/registries/fetch" +) + +const ( + keepImagesRepo = "team/app" + keepImagesIndexType = "application/vnd.oci.image.index.v1+json" + keepImagesImageType = "application/vnd.oci.image.manifest.v1+json" +) + +// keepImagesRegistry is a registry with one repository whose tags can be +// moved between images. +type keepImagesRegistry struct { + srv *httptest.Server + mu sync.Mutex + // manifests holds the manifest documents by digest, tags the digest + // each tag points at. + manifests map[string]string + types map[string]string + tags map[string]string + blobs map[string]string +} + +// keepImage is one image of the repository: an index with one platform +// manifest, which has a config and a layer. +type keepImage struct { + index, manifest string + config, layer ociBlob +} + +func newKeepImagesRegistry(t *testing.T) *keepImagesRegistry { + t.Helper() + r := &keepImagesRegistry{manifests: map[string]string{}, types: map[string]string{}, tags: map[string]string{}, blobs: map[string]string{}} + r.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + r.mu.Lock() + defer r.mu.Unlock() + const manifestsPrefix, blobsPrefix = "/v2/" + keepImagesRepo + "/manifests/", "/v2/" + keepImagesRepo + "/blobs/" + if len(req.URL.Path) > len(blobsPrefix) && req.URL.Path[:len(blobsPrefix)] == blobsPrefix { + if content, ok := r.blobs[req.URL.Path[len(blobsPrefix):]]; ok { + _, _ = io.WriteString(w, content) + return + } + } + if len(req.URL.Path) > len(manifestsPrefix) && req.URL.Path[:len(manifestsPrefix)] == manifestsPrefix { + digest := req.URL.Path[len(manifestsPrefix):] + if tagged, ok := r.tags[digest]; ok { + digest = tagged + } + if body, ok := r.manifests[digest]; ok { + w.Header().Set("Content-Type", r.types[digest]) + w.Header().Set("Docker-Content-Digest", digest) + if req.Method != http.MethodHead { + _, _ = io.WriteString(w, body) + } + return + } + } + http.NotFound(w, req) + })) + t.Cleanup(r.srv.Close) + return r +} + +func (r *keepImagesRegistry) addManifest(t *testing.T, mediaType string, doc map[string]any) string { + t.Helper() + doc["schemaVersion"], doc["mediaType"] = 2, mediaType + body, err := json.Marshal(doc) + if err != nil { + t.Fatal(err) + } + digest := ociBlob{content: string(body)}.digest() + r.manifests[digest], r.types[digest] = string(body), mediaType + return digest +} + +// addImage adds an image whose blobs are named after it. +func (r *keepImagesRegistry) addImage(t *testing.T, name string) keepImage { + t.Helper() + r.mu.Lock() + defer r.mu.Unlock() + image := keepImage{ + config: ociBlob{keepImagesRepo, name + " config"}, + layer: ociBlob{keepImagesRepo, name + " layer"}, + } + for _, b := range []ociBlob{image.config, image.layer} { + r.blobs[b.digest()] = b.content + } + image.manifest = r.addManifest(t, keepImagesImageType, map[string]any{ + "config": map[string]any{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": image.config.digest()}, + "layers": []map[string]any{{"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "digest": image.layer.digest()}}, + }) + image.index = r.addManifest(t, keepImagesIndexType, map[string]any{ + "manifests": []map[string]any{{ + "mediaType": keepImagesImageType, "digest": image.manifest, + "platform": map[string]string{"os": "linux", "architecture": "arm64"}, + }}, + }) + return image +} + +func (r *keepImagesRegistry) tag(tag string, image keepImage) { + r.mu.Lock() + defer r.mu.Unlock() + r.tags[tag] = image.index +} + +// newKeepImagesProxy serves the registry through a proxy that keeps the +// blobs of keepLatest images per repository. +func newKeepImagesProxy(t *testing.T, registry *keepImagesRegistry, keepLatest int) (*httptest.Server, *database.DB) { + t.Helper() + db, store := setupEvictionTest(t) + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(registry.srv.Client()), fetch.WithMaxRetries(0)) + t.Cleanup(func() { _ = fetcher.Close() }) + proxy := handler.NewProxy(db, store, fetcher, fetch.NewResolver(), discardLogger()) + proxy.HTTPClient = registry.srv.Client() + proxy.KeepLatestImages = keepLatest + srv := httptest.NewServer(handler.NewContainerHandlerWithRegistry(proxy, "http://proxy.test", registry.srv.URL).Routes()) + t.Cleanup(srv.Close) + return srv, db +} + +// pullImage pulls a tag the way a client does: the tag, the platform +// manifest by digest, then its blobs. +func pullImage(t *testing.T, proxyURL, tag string, image keepImage) { + t.Helper() + base := proxyURL + "/" + keepImagesRepo + ociGet(t, base+"/manifests/"+tag) + ociGet(t, base+"/manifests/"+image.manifest) + ociGet(t, base+"/blobs/"+image.config.digest()) + ociGet(t, base+"/blobs/"+image.layer.digest()) +} + +// ageImages makes every cached blob, every record and every tag sighting +// look old, with the images given first seen longest ago. +func ageImages(t *testing.T, db *database.DB, now time.Time, images ...keepImage) { + t.Helper() + recorded := now.Add(-30 * retentionDay).UnixMilli() + if _, err := db.Exec(db.Rebind(`UPDATE image_manifests SET recorded_at = ?`), recorded); err != nil { + t.Fatalf("backdating records: %v", err) + } + for i, image := range images { + old := now.Add(-time.Duration(30-i) * retentionDay) + ociBackdate(t, db, image.config, old, old) + ociBackdate(t, db, image.layer, old, old) + query := db.Rebind(`UPDATE image_manifests SET tagged_at = ? WHERE digest = ? AND tagged_at > 0`) + if _, err := db.Exec(query, old.UnixMilli(), image.index); err != nil { + t.Fatalf("backdating image %d: %v", i, err) + } + } +} + +func imageCached(t *testing.T, db *database.DB, image keepImage) bool { + t.Helper() + config := ociArtifact(t, db, keepImagesRepo, image.config).StoragePath.Valid + layer := ociArtifact(t, db, keepImagesRepo, image.layer).StoragePath.Valid + if config != layer { + t.Errorf("config cached = %v, layer cached = %v, want the same", config, layer) + } + return layer +} + +// With keep_latest, the blobs of the images a tag most recently pointed at +// outlive their rule. Blobs of images the tag left longer ago are evicted. +func TestSweepRetentionKeepsBlobsOfNewestImages(t *testing.T) { + for _, tt := range []struct { + keepLatest int + wantCached [3]bool + }{ + {keepLatest: 0, wantCached: [3]bool{false, false, false}}, + {keepLatest: 1, wantCached: [3]bool{false, false, true}}, + {keepLatest: 2, wantCached: [3]bool{false, true, true}}, + } { + registry := newKeepImagesRegistry(t) + srv, db := newKeepImagesProxy(t, registry, tt.keepLatest) + images := []keepImage{registry.addImage(t, "first"), registry.addImage(t, "second"), registry.addImage(t, "third")} + for _, image := range images { + registry.tag("beta", image) + pullImage(t, srv.URL, "beta", image) + } + now := time.Now() + ageImages(t, db, now, images...) + + rules := retention.Rules{Ecosystems: map[string]time.Duration{"oci": retentionDay}, KeepLatest: tt.keepLatest} + sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now) + + for i, image := range images { + if got := imageCached(t, db, image); got != tt.wantCached[i] { + t.Errorf("keep_latest %d: image %d cached = %v, want %v", tt.keepLatest, i+1, got, tt.wantCached[i]) + } + } + } +} + +// A sweep drops the records of images that are no longer among the newest, +// so the table holds a few images per repository, and all records once +// keep_latest is turned off. +func TestSweepRetentionPrunesImageRecords(t *testing.T) { + registry := newKeepImagesRegistry(t) + srv, db := newKeepImagesProxy(t, registry, 1) + images := []keepImage{registry.addImage(t, "first"), registry.addImage(t, "second")} + for _, image := range images { + registry.tag("beta", image) + pullImage(t, srv.URL, "beta", image) + } + now := time.Now() + ageImages(t, db, now, images...) + recorded := func(digest string) bool { + t.Helper() + _, ok, err := db.ImageManifestTaggedAt(purl.MakePURLString("oci", keepImagesRepo, ""), digest) + if err != nil { + t.Fatalf("ImageManifestTaggedAt: %v", err) + } + return ok + } + + rules := retention.Rules{Ecosystems: map[string]time.Duration{"oci": retentionDay}, KeepLatest: 1} + sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now) + for _, digest := range []string{images[0].index, images[0].manifest} { + if recorded(digest) { + t.Errorf("manifest %s of the older image is still recorded", digest) + } + } + for _, digest := range []string{images[1].index, images[1].manifest} { + if !recorded(digest) { + t.Errorf("manifest %s of the newest image is no longer recorded", digest) + } + } + + rules.KeepLatest = 0 + sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now) + if recorded(images[1].index) { + t.Error("records are left after keep_latest was turned off") + } +} + +// Manifests are recorded only when a retention rule can evict container +// blobs, since nothing reads the records otherwise. +func TestKeepLatestImagesNeedsAnOCIRule(t *testing.T) { + for _, tt := range []struct { + name string + def string + ecosystems map[string]string + packages map[string]string + keepLatest int + want int + }{ + {name: "no keep_latest", ecosystems: map[string]string{"oci": "1d"}, want: 0}, + {name: "no rule", keepLatest: 2, want: 0}, + {name: "rule for another ecosystem", ecosystems: map[string]string{"npm": "7d"}, keepLatest: 2, want: 0}, + {name: "oci rule", ecosystems: map[string]string{"oci": "1d"}, keepLatest: 2, want: 2}, + {name: "default", def: "30d", keepLatest: 2, want: 2}, + {name: "default with oci exempt", def: "30d", ecosystems: map[string]string{"oci": "0"}, keepLatest: 2, want: 0}, + {name: "rule for one repository", packages: map[string]string{"pkg:oci/library/nginx": "3d"}, keepLatest: 2, want: 2}, + {name: "rule for another ecosystem's package", packages: map[string]string{"pkg:npm/lodash": "3d"}, keepLatest: 2, want: 0}, + } { + t.Run(tt.name, func(t *testing.T) { + cfg := config.Default() + cfg.Storage.Retention.Default = tt.def + cfg.Storage.Retention.Ecosystems = tt.ecosystems + cfg.Storage.Retention.Packages = tt.packages + cfg.Storage.Retention.KeepLatest = tt.keepLatest + if got := keepLatestImages(cfg); got != tt.want { + t.Errorf("keepLatestImages() = %d, want %d", got, tt.want) + } + }) + } +} + +// An image pulled only by digest is not one of the newest images, so its +// blobs are evicted by age as without keep_latest. +func TestSweepRetentionKeepLatestIgnoresImagesPulledByDigest(t *testing.T) { + registry := newKeepImagesRegistry(t) + srv, db := newKeepImagesProxy(t, registry, 1) + image := registry.addImage(t, "pinned") + pullImage(t, srv.URL, image.index, image) + now := time.Now() + ageImages(t, db, now, image) + + rules := retention.Rules{Ecosystems: map[string]time.Duration{"oci": retentionDay}, KeepLatest: 1} + sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now) + + if imageCached(t, db, image) { + t.Error("blobs of an image pulled only by digest were kept") + } +} diff --git a/internal/server/retention_keep_latest_test.go b/internal/server/retention_keep_latest_test.go index 9ee138ba..dd809854 100644 --- a/internal/server/retention_keep_latest_test.go +++ b/internal/server/retention_keep_latest_test.go @@ -206,9 +206,9 @@ func TestSweepRetentionWithoutKeepLatestEvictsNewestToo(t *testing.T) { } // The ecosystems whose handlers store plain release versions support -// keep_latest. +// keep_latest, and so does oci, which keeps by image. func TestKeepLatestEcosystems(t *testing.T) { - want := []string{"cargo", "golang", "npm", "pypi"} + want := []string{"cargo", "golang", "npm", "oci", "pypi"} if got := retention.Default.KeepLatestKeys(); !slices.Equal(got, want) { t.Errorf("KeepLatestKeys() = %v, want %v", got, want) } diff --git a/internal/server/server.go b/internal/server/server.go index c836ee34..9f995ca9 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -304,6 +304,7 @@ func (s *Server) serve(listener net.Listener) error { return fmt.Errorf("configuring scanners: %w", err) } configureMetadata(proxy, s.cfg) + proxy.KeepLatestImages = keepLatestImages(s.cfg) proxy.GradleReadOnly = s.cfg.Gradle.BuildCache.ReadOnly proxy.HostedEnabled = s.cfg.Hosted.Enabled proxy.HostedNPMScopes = s.cfg.Hosted.NPMScopes From fff46418ca95a10a9f79baf4c326492b23ec1189 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 10 Nov 2026 18:24:59 +0100 Subject: [PATCH 2/2] Count a tag's index and platform manifest as one image A client that accepts no image index is served one platform's manifest for a tag, so the tag was recorded under two digests and keep_latest could keep that one platform and drop the rest of the index. A tagged manifest that another tagged one references now counts as part of it. proxy mirror now records the images it mirrors, as serve does. --- docs/configuration.md | 1 + internal/database/image_manifests.go | 76 ++++++++++++++++--- internal/database/image_manifests_test.go | 38 ++++++++++ .../server/retention_keep_latest_oci_test.go | 13 ++++ internal/server/server.go | 1 + 5 files changed, 118 insertions(+), 11 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 35066b8a..6850d11a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -135,6 +135,7 @@ To know which blobs an image needs, the proxy records what each manifest it serv - An image only counts once its tag was pulled or mirrored with `keep_latest` set, and its blobs are known once its platform manifest was pulled too. Until then they expire by their rule as before. - An image pulled only by digest is not counted and not kept. +- A multi-platform tag counts as one image, also when a client that accepts no image index was served one platform's manifest for it. - Homebrew bottles and Helm charts pulled over OCI are images here: the newest versions of a bottle or chart are those whose tags were last asked for. Generic retention is opt-in because release assets may disappear upstream and cannot then be downloaded again after eviction. An existing `default` never starts evicting generic downloads. Set `ecosystems.generic: "30d"` to cover all generic artifacts, or configure only selected GitHub repositories: diff --git a/internal/database/image_manifests.go b/internal/database/image_manifests.go index ae729909..4a26758d 100644 --- a/internal/database/image_manifests.go +++ b/internal/database/image_manifests.go @@ -1,9 +1,12 @@ package database import ( + "cmp" "database/sql" "errors" "fmt" + "maps" + "slices" "strings" "time" @@ -32,8 +35,13 @@ type ImageManifest struct { } type imageManifestRow struct { - Digest string `db:"digest"` - Refs string `db:"refs"` + Digest string `db:"digest"` + Refs string `db:"refs"` + TaggedAt int64 `db:"tagged_at"` +} + +func (r imageManifestRow) refs() []string { + return strings.Fields(r.Refs) } // RecordImageManifest stores a manifest's references. For one recorded @@ -112,13 +120,7 @@ func (db *DB) walkNewestImages(packagePURL string, n int) (map[string]struct{}, if n <= 0 { return digests, nil, nil } - var rows []imageManifestRow - err := db.Select(&rows, db.Rebind(` - SELECT digest, refs FROM image_manifests - WHERE package_purl = ? AND tagged_at > 0 - ORDER BY tagged_at DESC, digest - LIMIT ? - `), packagePURL, n) + rows, err := db.newestImages(packagePURL, n) if err != nil { return nil, nil, err } @@ -131,8 +133,8 @@ func (db *DB) walkNewestImages(packagePURL string, n int) (map[string]struct{}, var next []string for _, row := range rows { manifests = append(manifests, row.Digest) - for _, ref := range strings.Split(row.Refs, "\n") { - if _, seen := digests[ref]; seen || ref == "" { + for _, ref := range row.refs() { + if _, seen := digests[ref]; seen { continue } digests[ref] = struct{}{} @@ -149,6 +151,58 @@ func (db *DB) walkNewestImages(packagePURL string, n int) (map[string]struct{}, return digests, manifests, nil } +// newestImages returns the n tagged manifests of a repository whose tag was +// seen last. A registry serves a tag as an index to one client and as one +// platform's manifest to a client that accepts no index, so a tagged +// manifest that another tagged one references is that image seen another +// way: it is no image of its own and lends the other its tag time. +func (db *DB) newestImages(packagePURL string, n int) ([]imageManifestRow, error) { + var rows []imageManifestRow + err := db.Select(&rows, db.Rebind(` + SELECT digest, refs, tagged_at FROM image_manifests + WHERE package_purl = ? AND tagged_at > 0 + `), packagePURL) + if err != nil { + return nil, err + } + + tagged := make(map[string]int64, len(rows)) + for _, row := range rows { + tagged[row.Digest] = row.TaggedAt + } + seen := maps.Clone(tagged) + referenced := map[string]struct{}{} + for _, row := range rows { + for _, ref := range row.refs() { + at, ok := tagged[ref] + if !ok || ref == row.Digest { + continue + } + referenced[ref] = struct{}{} + seen[row.Digest] = max(seen[row.Digest], at) + } + } + + images := rows[:0:0] + for _, row := range rows { + if _, ok := referenced[row.Digest]; !ok { + images = append(images, row) + } + } + if len(images) == 0 { + // Manifests that only reference each other; no registry serves these. + images = rows + } + slices.SortFunc(images, func(a, b imageManifestRow) int { + return cmp.Or( + cmp.Compare(seen[b.Digest], seen[a.Digest]), + cmp.Compare(b.TaggedAt, a.TaggedAt), + cmp.Compare(a.Digest, b.Digest), + ) + }) + return images[:min(n, len(images))], nil +} + // imageManifestRows returns the recorded manifests among digests. func (db *DB) imageManifestRows(packagePURL string, digests []string) ([]imageManifestRow, error) { var rows []imageManifestRow diff --git a/internal/database/image_manifests_test.go b/internal/database/image_manifests_test.go index 34777c27..b6074cb7 100644 --- a/internal/database/image_manifests_test.go +++ b/internal/database/image_manifests_test.go @@ -56,6 +56,44 @@ func TestNewestImageDigestsFollowsReferences(t *testing.T) { }) } +// A registry may serve a tag as an index to one client and as one platform's +// manifest to a client that accepts no index. Both are then tagged, and +// count as one image, with the time the tag was last seen through either. +func TestNewestImageDigestsCountsATagsRepresentationsOnce(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + recordImage(t, db, testImageRepo, "sha256:index-other", now.Add(-2*time.Hour), "sha256:arm-other") + recordImage(t, db, testImageRepo, "sha256:index", now.Add(-3*time.Hour), "sha256:arm", "sha256:amd") + recordImage(t, db, testImageRepo, "sha256:arm", time.Time{}, "sha256:layer-arm") + recordImage(t, db, testImageRepo, "sha256:amd", now.Add(-time.Hour), "sha256:layer-amd") + + want := []string{"sha256:amd", "sha256:arm", "sha256:index", "sha256:layer-amd", "sha256:layer-arm"} + if got := newestImageDigests(t, db, testImageRepo, 1); !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want the whole index %v", got, want) + } + if got := newestImageDigests(t, db, testImageRepo, 2); len(got) != 7 { + t.Errorf("newest 2 = %v, want all 7 digests of both images", got) + } + }) +} + +// A platform manifest two indexes share lends both its tag time, and the +// index tagged later is then the newer image. +func TestNewestImageDigestsOrdersImagesSharingAManifest(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + // The digests are named so that a tie would pick the old index. + recordImage(t, db, testImageRepo, "sha256:a-index-old", now.Add(-3*time.Hour), "sha256:arm", "sha256:amd-old") + recordImage(t, db, testImageRepo, "sha256:b-index-new", now.Add(-2*time.Hour), "sha256:arm", "sha256:amd-new") + recordImage(t, db, testImageRepo, "sha256:arm", now.Add(-time.Hour), "sha256:layer-arm") + + want := []string{"sha256:amd-new", "sha256:arm", "sha256:b-index-new", "sha256:layer-arm"} + if got := newestImageDigests(t, db, testImageRepo, 1); !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want %v", got, want) + } + }) +} + // A manifest only asked for by digest is not an image of its own. func TestNewestImageDigestsIgnoresUntaggedManifests(t *testing.T) { runWithBothDatabases(t, func(t *testing.T, db *DB) { diff --git a/internal/server/retention_keep_latest_oci_test.go b/internal/server/retention_keep_latest_oci_test.go index bf6f96de..50739b24 100644 --- a/internal/server/retention_keep_latest_oci_test.go +++ b/internal/server/retention_keep_latest_oci_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "io" + "log/slog" "net/http" "net/http/httptest" "sync" @@ -282,6 +283,18 @@ func TestKeepLatestImagesNeedsAnOCIRule(t *testing.T) { } } +// proxy mirror runs in a process of its own and has to record the images it +// mirrors as serve does, or the sweep would not know their blobs. +func TestMirrorProxyRecordsImagesForKeepLatest(t *testing.T) { + cfg := config.Default() + cfg.Storage.Retention.Ecosystems = map[string]string{"oci": "1d"} + cfg.Storage.Retention.KeepLatest = 2 + proxy, _ := NewMirrorProxy(cfg, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + if proxy.KeepLatestImages != 2 { + t.Errorf("KeepLatestImages = %d, want 2", proxy.KeepLatestImages) + } +} + // An image pulled only by digest is not one of the newest images, so its // blobs are evicted by age as without keep_latest. func TestSweepRetentionKeepLatestIgnoresImagesPulledByDigest(t *testing.T) { diff --git a/internal/server/server.go b/internal/server/server.go index 807e6e46..39d43c9d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -633,6 +633,7 @@ func NewMirrorProxy(cfg *config.Config, db *database.DB, store storage.Storage, configureMirrorResolvers(proxy, cfg.Upstream) proxy.HTTPClient = metadataClient proxy.RateLimits = s.rateLimits + proxy.KeepLatestImages = keepLatestImages(cfg) return proxy, s.newContainerHandler(proxy) }