Keep blobs of scheduled images from retention

docs/configuration.md

@@ -111,7 +111,7 @@ storage:
     sweep_interval: "10m"
 ```
 
-Each cached file ages on its own, counted from the last time the proxy served it, or from when it was fetched if it has not been served since. A newer release of the same package does not shorten the life of older versions. Durations take a `d` suffix for days, and `"0"` or an empty value means never evict by age. With no retention configured, artifacts stay until `max_size` evicts them. `sweep_interval` must be at least `1m`.
+Each cached file ages on its own, counted from the last time the proxy served it, or from when it was fetched if it has not been served since. The blobs of an image that `mirror.schedules` keeps current also count from the last run that found them cached, so the schedule keeps them; for `max_size` eviction such a run doesn't count as use. A newer release of the same package does not shorten the life of older versions. Durations take a `d` suffix for days, and `"0"` or an empty value means never evict by age. With no retention configured, artifacts stay until `max_size` evicts them. `sweep_interval` must be at least `1m`.
 
 A package rule wins over an ecosystem rule, which wins over `default`. Package rules are keyed by package PURL without a version, written in one of the forms the table below lists for the ecosystem. The ecosystems and package rules can only name ecosystems that support retention in the running version; anything else, including a typo, stops the proxy at startup with an error naming the setting. `default` applies only to supported ecosystems, except `generic`, which requires an explicit rule. Support is added one ecosystem at a time, so a later release can make an existing `default` cover more ecosystems. On startup the proxy logs which ecosystems retention supports (`covered`) and which ones `default` does not reach (`default_not_applied`), including `generic`.
 

internal/database/retention.go

@@ -17,28 +17,33 @@ type RetentionCandidate struct {
 	Size           sql.NullInt64 `db:"size"`
 	FetchedAt      sql.NullTime  `db:"fetched_at"`
 	LastAccessedAt sql.NullTime  `db:"last_accessed_at"`
+	MirroredAt     sql.NullTime  `db:"mirrored_at"`
 	Ecosystem      string        `db:"ecosystem"`
 	PackagePURL    string        `db:"package_purl"`
 }
 
-// ExpiredBefore reports whether the artifact was neither fetched nor
-// accessed at or after cutoff. Both are checked: clearing a record keeps its
-// last access time, so a record fetched again after an eviction has a fresh
-// fetch time next to an old access time.
+// ExpiredBefore reports whether the artifact was neither fetched, accessed
+// nor found by a mirror run at or after cutoff. Fetch and access are both
+// checked: clearing a record keeps its last access time, so a record fetched
+// again after an eviction has a fresh fetch time next to an old access time.
 func (c RetentionCandidate) ExpiredBefore(cutoff time.Time) bool {
 	if !c.FetchedAt.Valid || !c.FetchedAt.Time.Before(cutoff) {
 		return false
 	}
+	if c.MirroredAt.Valid && !c.MirroredAt.Time.Before(cutoff) {
+		return false
+	}
 	return !c.LastAccessedAt.Valid || c.LastAccessedAt.Time.Before(cutoff)
 }
 
-// expiredCondition selects cached artifacts fetched before the cutoff and not
-// accessed since. It takes the cutoff twice.
+// expiredCondition selects cached artifacts fetched before the cutoff and
+// neither accessed nor mirrored since. It takes the cutoff three times.
 const expiredCondition = `
 	storage_path IS NOT NULL
 	AND source <> 'local'
 	AND fetched_at < ?
-	AND (last_accessed_at IS NULL OR last_accessed_at < ?)`
+	AND (last_accessed_at IS NULL OR last_accessed_at < ?)
+	AND (mirrored_at IS NULL OR mirrored_at < ?)`
 
 // MaxArtifactID returns the highest artifact id, or zero for an empty table.
 func (db *DB) MaxArtifactID() (int64, error) {
@@ -56,7 +61,7 @@ func (db *DB) GetRetentionCandidates(fromID, toID int64, cutoff time.Time) ([]Re
 	var candidates []RetentionCandidate
 	query := db.Rebind(`
 		SELECT a.id, a.version_purl, a.filename, a.storage_path, a.size,
-		       a.fetched_at, a.last_accessed_at, p.ecosystem, p.purl AS package_purl
+		       a.fetched_at, a.last_accessed_at, a.mirrored_at, p.ecosystem, p.purl AS package_purl
 		FROM (
 			SELECT * FROM artifacts
 			WHERE id > ? AND id <= ? AND` + expiredCondition + `
@@ -65,7 +70,7 @@ func (db *DB) GetRetentionCandidates(fromID, toID int64, cutoff time.Time) ([]Re
 		JOIN packages p ON p.purl = v.package_purl
 		ORDER BY a.id
 	`)
-	if err := db.Select(&candidates, query, fromID, toID, cutoff.UTC(), cutoff.UTC()); err != nil {
+	if err := db.Select(&candidates, query, fromID, toID, cutoff.UTC(), cutoff.UTC(), cutoff.UTC()); err != nil {
 		return nil, err
 	}
 	return candidates, nil
@@ -89,7 +94,7 @@ func (db *DB) ClearExpiredArtifact(versionPURL, filename, storagePath string, cu
 		SET storage_path = NULL, content_hash = NULL, size = NULL,
 		    content_type = NULL, fetched_at = NULL, updated_at = ?
 		WHERE version_purl = ? AND filename = ? AND storage_path = ? AND` + expiredCondition)
-	res, err := tx.Exec(query, time.Now(), versionPURL, filename, storagePath, cutoff.UTC(), cutoff.UTC())
+	res, err := tx.Exec(query, time.Now(), versionPURL, filename, storagePath, cutoff.UTC(), cutoff.UTC(), cutoff.UTC())
 	if err != nil {
 		return false, err
 	}
@@ -103,6 +108,16 @@ func (db *DB) ClearExpiredArtifact(versionPURL, filename, storagePath string, cu
 	return true, tx.Commit()
 }
 
+// MarkArtifactMirrored records that a mirror run found the artifact cached,
+// which keeps it from retention without counting as a cache hit.
+func (db *DB) MarkArtifactMirrored(versionPURL, filename string) error {
+	_, err := db.Exec(db.Rebind(`
+		UPDATE artifacts SET mirrored_at = ?
+		WHERE version_purl = ? AND filename = ? AND storage_path IS NOT NULL
+	`), time.Now().UTC(), versionPURL, filename)
+	return err
+}
+
 // GetArtifactEcosystem returns the packages.ecosystem value of the package an
 // artifact version belongs to, or "" when it has no package record.
 func (db *DB) GetArtifactEcosystem(versionPURL string) (string, error) {

internal/database/retention_test.go

@@ -341,18 +341,20 @@ func TestRetentionCandidateExpiredBefore(t *testing.T) {
 	recent := sql.NullTime{Time: now, Valid: true}
 
 	tests := []struct {
-		name              string
-		fetched, accessed sql.NullTime
-		want              bool
+		name                        string
+		fetched, accessed, mirrored sql.NullTime
+		want                        bool
 	}{
-		{"fetched and accessed before", old, old, true},
-		{"never served", old, sql.NullTime{}, true},
-		{"accessed since", old, recent, false},
-		{"fetched again since", recent, old, false},
-		{"not cached", sql.NullTime{}, old, false},
+		{"fetched and accessed before", old, old, sql.NullTime{}, true},
+		{"never served", old, sql.NullTime{}, sql.NullTime{}, true},
+		{"accessed since", old, recent, sql.NullTime{}, false},
+		{"fetched again since", recent, old, sql.NullTime{}, false},
+		{"not cached", sql.NullTime{}, old, sql.NullTime{}, false},
+		{"mirrored since", old, old, recent, false},
+		{"mirrored before", old, old, old, true},
 	}
 	for _, tt := range tests {
-		c := RetentionCandidate{FetchedAt: tt.fetched, LastAccessedAt: tt.accessed}
+		c := RetentionCandidate{FetchedAt: tt.fetched, LastAccessedAt: tt.accessed, MirroredAt: tt.mirrored}
 		if got := c.ExpiredBefore(cutoff); got != tt.want {
 			t.Errorf("%s: ExpiredBefore() = %v, want %v", tt.name, got, tt.want)
 		}

internal/database/schema.go

@@ -75,7 +75,8 @@ CREATE TABLE IF NOT EXISTS artifacts (
 	created_at DATETIME,
 	updated_at DATETIME,
 	source TEXT NOT NULL DEFAULT 'upstream',
-	manifest TEXT
+	manifest TEXT,
+	mirrored_at DATETIME
 );
 CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_version_filename ON artifacts(version_purl, filename);
 CREATE INDEX IF NOT EXISTS idx_artifacts_storage_path ON artifacts(storage_path);
@@ -252,7 +253,8 @@ CREATE TABLE IF NOT EXISTS artifacts (
 	created_at TIMESTAMP,
 	updated_at TIMESTAMP,
 	source TEXT NOT NULL DEFAULT 'upstream',
-	manifest TEXT
+	manifest TEXT,
+	mirrored_at TIMESTAMP
 );
 CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_version_filename ON artifacts(version_purl, filename);
 CREATE INDEX IF NOT EXISTS idx_artifacts_storage_path ON artifacts(storage_path);
@@ -388,7 +390,8 @@ CREATE TABLE IF NOT EXISTS artifacts (
 	created_at DATETIME,
 	updated_at DATETIME,
 	source TEXT NOT NULL DEFAULT 'upstream',
-	manifest TEXT
+	manifest TEXT,
+	mirrored_at DATETIME
 );
 CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_version_filename ON artifacts(version_purl, filename);
 CREATE INDEX IF NOT EXISTS idx_artifacts_storage_path ON artifacts(storage_path);
@@ -413,7 +416,8 @@ CREATE TABLE IF NOT EXISTS artifacts (
 	created_at TIMESTAMP,
 	updated_at TIMESTAMP,
 	source TEXT NOT NULL DEFAULT 'upstream',
-	manifest TEXT
+	manifest TEXT,
+	mirrored_at TIMESTAMP
 );
 CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_version_filename ON artifacts(version_purl, filename);
 CREATE INDEX IF NOT EXISTS idx_artifacts_storage_path ON artifacts(storage_path);
@@ -538,6 +542,7 @@ var migrations = []migration{
 	{"019_add_artifact_filename_index", migrateAddArtifactFilenameIndex},
 	{"020_add_metadata_snapshots", migrateAddMetadataSnapshots},
 	{"021_add_projects", migrateAddProjects},
+	{"022_add_artifact_mirrored_at", migrateAddArtifactMirroredAt},
 }
 
 // isTableNotFound returns true if the error indicates a missing table.
@@ -1080,6 +1085,26 @@ func migrateAddArtifactSource(db *DB) error {
 	return nil
 }
 
+// migrateAddArtifactMirroredAt records when a mirror run last found an
+// artifact cached, which retention counts as use and LRU eviction doesn't.
+func migrateAddArtifactMirroredAt(db *DB) error {
+	hasCol, err := db.HasColumn("artifacts", "mirrored_at")
+	if err != nil {
+		return fmt.Errorf("checking column mirrored_at: %w", err)
+	}
+	if hasCol {
+		return nil
+	}
+	ts := sqliteDatetime
+	if db.dialect == DialectPostgres {
+		ts = postgresTimestamp
+	}
+	if _, err := db.Exec(fmt.Sprintf("ALTER TABLE artifacts ADD COLUMN mirrored_at %s", ts)); err != nil {
+		return fmt.Errorf("adding column mirrored_at to artifacts: %w", err)
+	}
+	return nil
+}
+
 // migrateAddArtifactManifest stores the manifest read from a pushed archive,
 // which the metadata of its ecosystem lists the version with.
 func migrateAddArtifactManifest(db *DB) error {

internal/database/types.go

@@ -151,6 +151,9 @@ type Artifact struct {
 	UpdatedAt      time.Time      `db:"updated_at" json:"updated_at"`
 	Source         string         `db:"source" json:"source"`
 	Manifest       sql.NullString `db:"manifest" json:"manifest,omitempty"`
+	// MirroredAt is when a mirror run last found the artifact cached. It
+	// keeps the artifact from retention but not from LRU eviction.
+	MirroredAt sql.NullTime `db:"mirrored_at" json:"-"`
 }
 
 func (a *Artifact) sourceOrUpstream() string {

internal/handler/container_mirror.go

@@ -502,6 +502,11 @@ func (m *imageMirror) mirrorBlob(ctx context.Context, digest string) (int64, boo
 		return 0, false, fmt.Errorf("blob %s: %w", digest, err)
 	}
 	if cached {
+		// Not a cache hit, so LRU eviction still sees the image as unused,
+		// but retention keeps it while the schedule does.
+		if err := proxy.MarkMirrored("oci", m.cacheName, digest, digest); err != nil {
+			return 0, false, fmt.Errorf("blob %s: %w", digest, err)
+		}
 		return 0, false, nil
 	}
 	result, err := proxy.GetOrFetchArtifactFromURLWithDigest(ctx, "oci", m.cacheName, digest, digest,

internal/handler/container_mirror_test.go

@@ -488,6 +488,43 @@ func TestMirrorImageCountsNoCacheHits(t *testing.T) {
 	}
 }
 
+// A scheduled image that no client pulls keeps its blobs: a mirror run that
+// finds them cached holds them back from retention, though not from LRU
+// eviction, which TestMirrorImageCountsNoCacheHits covers.
+func TestMirrorImageKeepsCachedBlobsFromRetention(t *testing.T) {
+	registry, _ := newFakeImageRegistry(t)
+	proxy := newMirrorTestProxy(t, registry)
+	h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL)
+	ref := ImageRef{Name: "team/app", References: []string{"multi"}}
+
+	if _, err := h.MirrorImage(context.Background(), ref, nil); err != nil {
+		t.Fatalf("first MirrorImage: %v", err)
+	}
+	old := time.Now().Add(-30 * 24 * time.Hour).UTC()
+	if _, err := proxy.DB.Exec(proxy.DB.Rebind("UPDATE artifacts SET fetched_at = ?, last_accessed_at = ?"), old, old); err != nil {
+		t.Fatal(err)
+	}
+	cutoff := time.Now().Add(-14 * 24 * time.Hour)
+	maxID, err := proxy.DB.MaxArtifactID()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if candidates, err := proxy.DB.GetRetentionCandidates(0, maxID, cutoff); err != nil || len(candidates) == 0 {
+		t.Fatalf("before the second run: %d candidates, err %v; want the aged blobs", len(candidates), err)
+	}
+
+	if _, err := h.MirrorImage(context.Background(), ref, nil); err != nil {
+		t.Fatalf("second MirrorImage: %v", err)
+	}
+	candidates, err := proxy.DB.GetRetentionCandidates(0, maxID, cutoff)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(candidates) != 0 {
+		t.Errorf("after the second run %d blobs are retention candidates, want none", len(candidates))
+	}
+}
+
 func TestMirrorImageRefetchesBlobMissingFromStorage(t *testing.T) {
 	registry, image := newFakeImageRegistry(t)
 	proxy := newMirrorTestProxy(t, registry)

internal/handler/handler.go

@@ -451,6 +451,16 @@ func (p *Proxy) HasCachedArtifact(ctx context.Context, ecosystem, name, version,
 	return p.Storage.Exists(ctx, artifact.StoragePath)
 }
 
+// MarkMirrored records that a mirror run found an artifact cached, which
+// keeps it from retention without counting as a cache hit.
+func (p *Proxy) MarkMirrored(ecosystem, name, version, filename string) error {
+	_, versionPURL, err := packagePURLStrings(ecosystem, name, version)
+	if err != nil {
+		return err
+	}
+	return p.DB.MarkArtifactMirrored(versionPURL, filename)
+}
+
 // ClearCachedArtifact clears an artifact cache record after an external
 // integrity check fails, and queues its stored bytes for deletion.
 func (p *Proxy) ClearCachedArtifact(ecosystem, name, version, filename string) error {