Keep the blobs of the newest container images with keep_latest

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

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.

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,16 @@ 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.
+- 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:
 

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")
 	}

internal/database/image_manifests.go

@@ -0,0 +1,224 @@
+package database
+
+import (
+	"cmp"
+	"database/sql"
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"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"`
+	TaggedAt int64  `db:"tagged_at"`
+}
+
+func (r imageManifestRow) refs() []string {
+	return strings.Fields(r.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
+	}
+	rows, err := db.newestImages(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 row.refs() {
+				if _, seen := digests[ref]; seen {
+					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
+}
+
+// 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
+	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
+}