From d02ad9b2bac1b270c22986f4d15a3dbf7d771e75 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 23 Oct 2026 16:15:24 +0200 Subject: [PATCH 1/4] Coordinate instances sharing a Postgres database Instances started together on an empty database created the schema at the same time and all but one failed on duplicate types. Schema creation and migrations now run under a Postgres advisory lock. Eviction, reclaim, retention and the Gradle build cache sweep take a per-job advisory lock for each pass, and an instance that finds the job running elsewhere skips the pass. Before, every instance evicted the whole overshoot over max_size at once and emptied the cache. Mirror schedules record their last start in a new job_runs table, so each schedule runs once per interval instead of once per instance. Refs #543 --- deploy/charts/proxy/values.yaml | 9 +- docs/configuration.md | 14 +- internal/database/database.go | 27 ++- internal/database/database_test.go | 2 +- internal/database/lock.go | 98 ++++++++++ internal/database/lock_test.go | 239 +++++++++++++++++++++++ internal/database/schema.go | 39 ++++ internal/mirror/schedule.go | 37 ++++ internal/mirror/schedule_test.go | 56 ++++++ internal/server/eviction.go | 4 +- internal/server/exclusive.go | 21 ++ internal/server/exclusive_test.go | 60 ++++++ internal/server/gradle_cache_eviction.go | 3 +- internal/server/mirror_schedule.go | 1 + internal/server/reclaim.go | 4 +- internal/server/retention.go | 10 +- 16 files changed, 603 insertions(+), 21 deletions(-) create mode 100644 internal/database/lock.go create mode 100644 internal/database/lock_test.go create mode 100644 internal/server/exclusive.go create mode 100644 internal/server/exclusive_test.go diff --git a/deploy/charts/proxy/values.yaml b/deploy/charts/proxy/values.yaml index 9a74e475..cb7a0bfe 100644 --- a/deploy/charts/proxy/values.yaml +++ b/deploy/charts/proxy/values.yaml @@ -1,7 +1,8 @@ -# More than one replica requires config.data.database on Postgres, -# config.data.storage on object storage, deploymentStrategy: RollingUpdate, -# and a PVC access mode other than ReadWriteOnce (or persistence disabled). -# With upstream.git, each replica also needs its own git.cache_dir. +# More than one replica requires config.data.database on Postgres and +# deploymentStrategy: RollingUpdate. Storage can be object storage (with +# persistence disabled) or a file:// URL on a ReadWriteMany volume +# (persistence.accessModes: [ReadWriteMany]). With upstream.git, each +# replica also needs its own git.cache_dir. replicaCount: 1 nameOverride: "" diff --git a/docs/configuration.md b/docs/configuration.md index 4da6f634..875f8fcb 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -197,6 +197,18 @@ database: |--------|-------------|------|-------------| | `database.url` | `PROXY_DATABASE_URL` | `-database-url` | PostgreSQL connection URL | +### Several instances + +Several instances can serve one cache when they share a PostgreSQL database and storage. The storage can be S3 or a `file://` directory on a volume every instance mounts, such as NFS, CephFS or a Kubernetes `ReadWriteMany` volume. Each fetch writes its own file and renames it into place, and records are updated with compare-and-swap, so concurrent fetches and a crashed instance leave no partial files or records that don't match a file. SQLite only supports a single instance. + +Instances coordinate through PostgreSQL advisory locks: + +- Instances started together create the schema and run migrations one after the other. +- Eviction, reclaim, retention and the Gradle build cache sweep run on one instance at a time. An instance that finds a job running elsewhere skips that pass. +- A [scheduled mirror](#scheduled-refresh) runs on whichever instance reaches it first in each interval. + +Cache misses are only coalesced within an instance. Two instances missing the same file both fetch it from upstream; the extra copy is queued for deletion like any replaced file. Each instance also needs its own [`git.cache_dir`](#git-repositories). + ### Hit counts Every cache hit updates the artifact's hit count and last-access time, which the stats pages and LRU eviction use. Rather than writing each hit as its own transaction, the proxy counts hits in memory and writes them together every `hit_flush_interval`. SQLite allows one writer at a time, and the proxy uses a single SQLite connection, so with a write per hit, concurrent downloads queue behind each other. @@ -1496,7 +1508,7 @@ mirror: Each run checks the tag with a HEAD request. If the tag still points to the cached digest, nothing else is fetched. If it moved, the new manifests and blobs are fetched first and the tag is switched only when all of them are cached, so a failed run leaves the previous image pullable. Images of one schedule are mirrored one after the other. -- The first run starts within ten minutes of startup, later runs every `interval` ±10 %, so restarts and several instances don't hit the registry at once. Each instance runs the schedules on its own. +- The first run starts within ten minutes of startup, later runs every `interval` ±10 %, so restarts and several instances don't hit the registry at once. Instances sharing a PostgreSQL database run each schedule once per interval: the first to reach it records the run, and the others skip it. - Digest-pinned PURLs never move and are rejected. A `pkg:oci` digest with `tag=` is accepted; the tag is refreshed and a warning is logged when it no longer matches the digest. - The previous digest stays cached, so pulls by digest keep working, until eviction or retention removes it. - Docker Hub does not count HEAD requests against its pull limit, but every changed manifest is a pull. Set `platforms` and `upstream.auth` for `registry-1.docker.io` to stay under the anonymous limit. diff --git a/internal/database/database.go b/internal/database/database.go index f438f676..5a7e1275 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -1,6 +1,7 @@ package database import ( + "context" "fmt" "os" "path/filepath" @@ -125,18 +126,24 @@ func OpenPostgresOrCreate(url string) (*DB, error) { return nil, err } - var exists bool - err = db.Get(&exists, "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'schema_info')") - if err != nil { - _ = db.Close() - return nil, fmt.Errorf("checking schema: %w", err) - } - - if !exists { + // Instances started together on an empty database would otherwise + // create the schema at the same time and fail on each other's tables. + err = db.WithLock(context.Background(), schemaLock, func() error { + var exists bool + if err := db.Get(&exists, "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'schema_info')"); err != nil { + return fmt.Errorf("checking schema: %w", err) + } + if exists { + return nil + } if err := db.CreateSchema(); err != nil { - _ = db.Close() - return nil, fmt.Errorf("creating schema: %w", err) + return fmt.Errorf("creating schema: %w", err) } + return nil + }) + if err != nil { + _ = db.Close() + return nil, err } return db, nil diff --git a/internal/database/database_test.go b/internal/database/database_test.go index f09a86a0..e5b9899b 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", "migrations", "schema_info"} + tables := []string{"artifacts", "pending_deletes", "versions", "packages", "vulnerabilities", "metadata_cache", "metadata_dates", "job_runs", "migrations", "schema_info"} for _, table := range tables { _, _ = db.Exec("DROP TABLE IF EXISTS " + table + " CASCADE") } diff --git a/internal/database/lock.go b/internal/database/lock.go new file mode 100644 index 00000000..b245c629 --- /dev/null +++ b/internal/database/lock.go @@ -0,0 +1,98 @@ +package database + +import ( + "context" + "database/sql" + "database/sql/driver" + "fmt" + "hash/fnv" + "time" +) + +// Several instances can share one Postgres database. Advisory locks keep +// them from creating the schema, migrating it or running a background job at +// the same time. SQLite is single-instance, so there every lock is granted +// at once. + +// lockKey maps a lock name to the 64-bit key Postgres advisory locks take. +func lockKey(name string) int64 { + h := fnv.New64a() + _, _ = h.Write([]byte("git-pkgs/proxy:" + name)) + return int64(h.Sum64()) //nolint:gosec // a key, wrapping is fine +} + +// WithLock runs fn while holding the named lock, waiting for another +// instance that holds it. +func (db *DB) WithLock(ctx context.Context, name string, fn func() error) error { + if db.dialect != DialectPostgres { + return fn() + } + conn, err := db.Conn(ctx) + if err != nil { + return fmt.Errorf("taking connection for lock %s: %w", name, err) + } + defer func() { _ = conn.Close() }() + if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_lock($1)", lockKey(name)); err != nil { + return fmt.Errorf("taking lock %s: %w", name, err) + } + defer unlock(conn, name) + return fn() +} + +// TryLock takes the named lock if no other instance holds it. When ok is +// true the caller must call release once done. The lock lives in a +// connection held for that time, and Postgres frees it if the process dies. +func (db *DB) TryLock(ctx context.Context, name string) (release func(), ok bool, err error) { + if db.dialect != DialectPostgres { + return func() {}, true, nil + } + conn, err := db.Conn(ctx) + if err != nil { + return nil, false, fmt.Errorf("taking connection for lock %s: %w", name, err) + } + if err := conn.QueryRowContext(ctx, "SELECT pg_try_advisory_lock($1)", lockKey(name)).Scan(&ok); err != nil { + _ = conn.Close() + return nil, false, fmt.Errorf("taking lock %s: %w", name, err) + } + if !ok { + _ = conn.Close() + return nil, false, nil + } + return func() { + unlock(conn, name) + _ = conn.Close() + }, true, nil +} + +// unlockTimeout bounds the unlock. If it fails, closing the connection still +// ends the session that holds the lock. +const unlockTimeout = 10 * time.Second + +func unlock(conn *sql.Conn, name string) { + ctx, cancel := context.WithTimeout(context.Background(), unlockTimeout) + defer cancel() + if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", lockKey(name)); err != nil { + // Drop the connection so the session, and with it the lock, ends. + _ = conn.Raw(func(any) error { return driver.ErrBadConn }) + } +} + +// ClaimJobRun records that the named job starts now unless it started less +// than minGap ago, and reports whether the caller should run it. Instances +// sharing the database each keep their own timer for a scheduled job, and +// this lets only the first of them run it per interval. +func (db *DB) ClaimJobRun(name string, minGap time.Duration) (bool, error) { + now := time.Now().UTC() + res, err := db.Exec(db.Rebind(` + INSERT INTO job_runs (name, started_at) VALUES (?, ?) + ON CONFLICT(name) DO UPDATE SET started_at = excluded.started_at + WHERE job_runs.started_at <= ?`), name, now, now.Add(-minGap)) + if err != nil { + return false, fmt.Errorf("claiming job run %s: %w", name, err) + } + n, err := res.RowsAffected() + if err != nil { + return false, fmt.Errorf("claiming job run %s: %w", name, err) + } + return n > 0, nil +} diff --git a/internal/database/lock_test.go b/internal/database/lock_test.go new file mode 100644 index 00000000..c49906d8 --- /dev/null +++ b/internal/database/lock_test.go @@ -0,0 +1,239 @@ +package database + +import ( + "context" + "net/url" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestTryLockSQLiteAlwaysGranted(t *testing.T) { + db := createTestDB(t) + release, ok, err := db.TryLock(context.Background(), "job:test") + if err != nil || !ok { + t.Fatalf("TryLock = %v, %v; want granted", ok, err) + } + // SQLite is single-instance, so a second holder is granted too. + release2, ok, err := db.TryLock(context.Background(), "job:test") + if err != nil || !ok { + t.Fatalf("second TryLock = %v, %v; want granted", ok, err) + } + release2() + release() +} + +func TestClaimJobRun(t *testing.T) { + forEachDialect(t, func(t *testing.T, db *DB) { + claimed, err := db.ClaimJobRun("mirror-schedule:a", time.Hour) + if err != nil || !claimed { + t.Fatalf("first claim = %v, %v; want claimed", claimed, err) + } + claimed, err = db.ClaimJobRun("mirror-schedule:a", time.Hour) + if err != nil || claimed { + t.Fatalf("claim within the gap = %v, %v; want not claimed", claimed, err) + } + claimed, err = db.ClaimJobRun("mirror-schedule:b", time.Hour) + if err != nil || !claimed { + t.Fatalf("claim of another job = %v, %v; want claimed", claimed, err) + } + // A run started longer ago than the gap can be claimed again. + claimed, err = db.ClaimJobRun("mirror-schedule:a", 0) + if err != nil || !claimed { + t.Fatalf("claim after the gap = %v, %v; want claimed", claimed, err) + } + }) +} + +// forEachDialect runs fn against SQLite and, when PROXY_DATABASE_URL is set, +// against Postgres. +func forEachDialect(t *testing.T, fn func(t *testing.T, db *DB)) { + t.Run("sqlite", func(t *testing.T) { fn(t, createTestDB(t)) }) + t.Run("postgres", func(t *testing.T) { + db := createTestPostgresDB(t) + if db == nil { + t.Skip("PROXY_DATABASE_URL not set") + } + defer func() { _ = db.Close() }() + fn(t, db) + }) +} + +func openTestPostgres(t *testing.T) *DB { + t.Helper() + dbURL := os.Getenv("PROXY_DATABASE_URL") + if dbURL == "" { + t.Skip("PROXY_DATABASE_URL not set") + } + db, err := OpenPostgres(dbURL) + if err != nil { + t.Fatalf("OpenPostgres: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + return db +} + +func TestTryLockPostgresExcludesOtherInstance(t *testing.T) { + a := openTestPostgres(t) + b := openTestPostgres(t) + ctx := context.Background() + + release, ok, err := a.TryLock(ctx, "job:test-exclusive") + if err != nil || !ok { + t.Fatalf("TryLock on a = %v, %v; want granted", ok, err) + } + if _, ok, err := b.TryLock(ctx, "job:test-exclusive"); err != nil || ok { + t.Fatalf("TryLock on b while a holds it = %v, %v; want refused", ok, err) + } + // Another lock name is independent. + releaseOther, ok, err := b.TryLock(ctx, "job:test-other") + if err != nil || !ok { + t.Fatalf("TryLock on another name = %v, %v; want granted", ok, err) + } + releaseOther() + + release() + release, ok, err = b.TryLock(ctx, "job:test-exclusive") + if err != nil || !ok { + t.Fatalf("TryLock on b after a released = %v, %v; want granted", ok, err) + } + release() +} + +func TestTryLockPostgresFreedWhenSessionEnds(t *testing.T) { + a := openTestPostgres(t) + b := openTestPostgres(t) + ctx := context.Background() + + if _, ok, err := a.TryLock(ctx, "job:test-crash"); err != nil || !ok { + t.Fatalf("TryLock on a = %v, %v; want granted", ok, err) + } + // End a's session holding the lock, as a crash would. + if _, err := b.Exec(`SELECT pg_terminate_backend(pid) FROM pg_locks + WHERE locktype = 'advisory' AND pid <> pg_backend_pid()`); err != nil { + t.Fatalf("terminating a's session: %v", err) + } + waitUntil(t, func() bool { + release, ok, err := b.TryLock(ctx, "job:test-crash") + if ok { + release() + } + return err == nil && ok + }) +} + +func TestWithLockPostgresSerialises(t *testing.T) { + const instances = 4 + dbs := make([]*DB, instances) + for i := range dbs { + dbs[i] = openTestPostgres(t) + } + var inside, overlapped atomic.Int32 + var wg sync.WaitGroup + for _, db := range dbs { + wg.Go(func() { + err := db.WithLock(context.Background(), "test-serialise", func() error { + if inside.Add(1) > 1 { + overlapped.Store(1) + } + time.Sleep(20 * time.Millisecond) + inside.Add(-1) + return nil + }) + if err != nil { + t.Errorf("WithLock: %v", err) + } + }) + } + wg.Wait() + if overlapped.Load() != 0 { + t.Error("two instances held the lock at once") + } +} + +// TestOpenPostgresOrCreateConcurrent starts several instances at once on an +// empty database. Without the schema lock all but one failed on each +// other's tables. +func TestOpenPostgresOrCreateConcurrent(t *testing.T) { + admin := openTestPostgres(t) + name := "proxy_lock_test_" + strings.ReplaceAll(time.Now().UTC().Format("150405.000000"), ".", "_") + if _, err := admin.Exec("CREATE DATABASE " + name); err != nil { + t.Skipf("cannot create a scratch database: %v", err) + } + t.Cleanup(func() { _, _ = admin.Exec("DROP DATABASE IF EXISTS " + name + " WITH (FORCE)") }) + + u, err := url.Parse(os.Getenv("PROXY_DATABASE_URL")) + if err != nil { + t.Fatalf("parsing PROXY_DATABASE_URL: %v", err) + } + u.Path = "/" + name + scratch := u.String() + + const instances = 5 + errs := make([]error, instances) + var wg sync.WaitGroup + start := make(chan struct{}) + for i := range instances { + wg.Go(func() { + <-start + db, err := OpenPostgresOrCreate(scratch) + if err == nil { + err = db.MigrateSchema() + _ = db.Close() + } + errs[i] = err + }) + } + close(start) + wg.Wait() + for i, err := range errs { + if err != nil { + t.Errorf("instance %d: %v", i, err) + } + } + + db, err := OpenPostgres(scratch) + if err != nil { + t.Fatalf("OpenPostgres: %v", err) + } + defer func() { _ = db.Close() }() + var n int + if err := db.Get(&n, "SELECT COUNT(*) FROM schema_info"); err != nil || n != 1 { + t.Errorf("schema_info rows = %d, %v; want 1", n, err) + } + if err := db.Get(&n, "SELECT COUNT(*) FROM migrations"); err != nil || n != len(migrations) { + t.Errorf("migrations rows = %d, %v; want %d", n, err, len(migrations)) + } +} + +func waitUntil(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatal("timed out") + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestMigrateAddsJobRuns(t *testing.T) { + forEachDialect(t, func(t *testing.T, db *DB) { + // A database from before the job_runs migration. + if _, err := db.Exec("DROP TABLE job_runs"); err != nil { + t.Fatalf("dropping job_runs: %v", err) + } + if _, err := db.Exec(db.Rebind("DELETE FROM migrations WHERE name = ?"), "015_add_job_runs"); err != nil { + t.Fatalf("forgetting the migration: %v", err) + } + if err := db.MigrateSchema(); err != nil { + t.Fatalf("MigrateSchema: %v", err) + } + if claimed, err := db.ClaimJobRun("mirror-schedule:a", time.Hour); err != nil || !claimed { + t.Fatalf("ClaimJobRun after migration = %v, %v; want claimed", claimed, err) + } + }) +} diff --git a/internal/database/schema.go b/internal/database/schema.go index dc9fc80a..7c1922bb 100644 --- a/internal/database/schema.go +++ b/internal/database/schema.go @@ -1,6 +1,7 @@ package database import ( + "context" "fmt" "strings" "time" @@ -86,6 +87,11 @@ CREATE TABLE IF NOT EXISTS pending_deletes ( ); CREATE INDEX IF NOT EXISTS idx_pending_deletes_queued_at ON pending_deletes(queued_at); +CREATE TABLE IF NOT EXISTS job_runs ( + name TEXT NOT NULL PRIMARY KEY, + started_at DATETIME NOT NULL +); + CREATE TABLE IF NOT EXISTS vulnerabilities ( id INTEGER PRIMARY KEY, vuln_id TEXT NOT NULL, @@ -210,6 +216,11 @@ CREATE TABLE IF NOT EXISTS pending_deletes ( ); CREATE INDEX IF NOT EXISTS idx_pending_deletes_queued_at ON pending_deletes(queued_at); +CREATE TABLE IF NOT EXISTS job_runs ( + name TEXT NOT NULL PRIMARY KEY, + started_at TIMESTAMP NOT NULL +); + CREATE TABLE IF NOT EXISTS vulnerabilities ( id SERIAL PRIMARY KEY, vuln_id TEXT NOT NULL, @@ -420,6 +431,7 @@ var migrations = []migration{ {"012_add_artifact_source", migrateAddArtifactSource}, {"013_add_artifact_manifest", migrateAddArtifactManifest}, {"014_add_metadata_last_accessed", migrateAddMetadataLastAccessed}, + {"015_add_job_runs", migrateAddJobRuns}, } // isTableNotFound returns true if the error indicates a missing table. @@ -489,9 +501,19 @@ func (db *DB) recordAllMigrations() error { return nil } +// schemaLock serialises schema creation and migrations across instances +// sharing a Postgres database. +const schemaLock = "schema" + // MigrateSchema applies any unapplied migrations in order. // For a fully migrated database this executes a single SELECT query. +// Instances sharing a Postgres database run it one after the other, so each +// sees the migrations the one before it recorded. func (db *DB) MigrateSchema() error { + return db.WithLock(context.Background(), schemaLock, db.migrateSchema) +} + +func (db *DB) migrateSchema() error { applied, err := db.appliedMigrations() if err != nil { return err @@ -709,6 +731,23 @@ func migrateAddPendingDeletes(db *DB) error { return nil } +// migrateAddJobRuns creates the table that records when a scheduled job +// last started, so instances sharing the database run it once per interval. +func migrateAddJobRuns(db *DB) error { + ts := sqliteDatetime + if db.dialect == DialectPostgres { + ts = postgresTimestamp + } + query := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS job_runs ( + name TEXT NOT NULL PRIMARY KEY, + started_at %s NOT NULL + )`, ts) + if _, err := db.Exec(query); err != nil { + return fmt.Errorf("creating job_runs table: %w", err) + } + return nil +} + // migrateAddPendingDeletesQueuedAtIndex indexes the column reclaim orders // the queue by, which grows large after a retention sweep. func migrateAddPendingDeletesQueuedAtIndex(db *DB) error { diff --git a/internal/mirror/schedule.go b/internal/mirror/schedule.go index 4ab0d2b2..55d6ab4a 100644 --- a/internal/mirror/schedule.go +++ b/internal/mirror/schedule.go @@ -2,6 +2,9 @@ package mirror import ( "context" + "crypto/sha256" + "encoding/hex" + "fmt" "log/slog" "math/rand/v2" "sync" @@ -27,6 +30,11 @@ type Scheduler struct { schedules []Schedule logger *slog.Logger wg sync.WaitGroup + // Claim, when set, reports whether this instance should run a schedule + // that no instance sharing the database started within minGap. Each + // instance keeps its own timer, so without it every instance would run + // every schedule. + Claim func(key string, minGap time.Duration) (bool, error) // randN returns a value in [0, n); tests replace it. randN func(n int64) int64 } @@ -75,6 +83,20 @@ func (s *Scheduler) loop(ctx context.Context, index int, schedule Schedule) { } func (s *Scheduler) run(ctx context.Context, index int, schedule Schedule) { + if s.Claim != nil { + // Later runs come at least interval - 2*spread apart, so a run + // started more recently than that was another instance's. + minGap := schedule.Interval - 2*(schedule.Interval/scheduleJitter) //nolint:mnd // two spreads of margin + claimed, err := s.Claim(scheduleKey(schedule), minGap) + if err != nil { + s.logger.Warn("mirror schedule: failed to claim run, skipping it", "schedule", index, "error", err) + return + } + if !claimed { + s.logger.Debug("mirror schedule: another instance ran it, skipping", "schedule", index) + return + } + } started := time.Now() progress, err := s.mirror.Run(ctx, &PURLSource{PURLs: schedule.PURLs, Platforms: schedule.Platforms}) if ctx.Err() != nil { @@ -103,3 +125,18 @@ func scheduleDelay(interval time.Duration, first bool, randN func(int64) int64) spread := interval / scheduleJitter return interval - spread + time.Duration(randN(int64(2*spread)+1)) //nolint:mnd // spread on both sides } + +// scheduleKey names a schedule by what it mirrors and how often, so +// instances with the same configuration agree on it and a changed schedule +// does not inherit the old one's last run. +func scheduleKey(schedule Schedule) string { + h := sha256.New() + for _, purl := range schedule.PURLs { + _, _ = fmt.Fprintf(h, "purl %s\n", purl) + } + for _, platform := range schedule.Platforms { + _, _ = fmt.Fprintf(h, "platform %s\n", platform) + } + _, _ = fmt.Fprintf(h, "interval %s\n", schedule.Interval) + return "mirror-schedule:" + hex.EncodeToString(h.Sum(nil))[:32] +} diff --git a/internal/mirror/schedule_test.go b/internal/mirror/schedule_test.go index c869c8a1..53bdea2e 100644 --- a/internal/mirror/schedule_test.go +++ b/internal/mirror/schedule_test.go @@ -175,3 +175,59 @@ func TestSchedulerSkipsScheduleWithoutInterval(t *testing.T) { t.Errorf("calls = %d, want 0", got) } } + +func TestSchedulerSkipsRunClaimedElsewhere(t *testing.T) { + images := &scheduleImages{} + schedule := Schedule{PURLs: []string{"pkg:docker/alpine@3"}, Interval: time.Hour} + s := newTestScheduler(images, schedule) + var mu sync.Mutex + var keys []string + var gaps []time.Duration + claim := true + s.Claim = func(key string, minGap time.Duration) (bool, error) { + mu.Lock() + defer mu.Unlock() + keys = append(keys, key) + gaps = append(gaps, minGap) + return claim, nil + } + + s.run(context.Background(), 0, schedule) + if got := images.calls.Load(); got != 1 { + t.Fatalf("calls after a claimed run = %d, want 1", got) + } + claim = false + s.run(context.Background(), 0, schedule) + if got := images.calls.Load(); got != 1 { + t.Fatalf("calls after a run claimed elsewhere = %d, want 1", got) + } + s.Claim = func(string, time.Duration) (bool, error) { return false, errors.New("database down") } + s.run(context.Background(), 0, schedule) + if got := images.calls.Load(); got != 1 { + t.Fatalf("calls after a failed claim = %d, want 1", got) + } + + if keys[0] != keys[1] { + t.Errorf("claim keys differ between runs: %q, %q", keys[0], keys[1]) + } + // Runs come at least 0.9 intervals apart; the gap stays below that. + if want := 48 * time.Minute; gaps[0] != want { + t.Errorf("minGap = %s, want %s", gaps[0], want) + } +} + +func TestScheduleKey(t *testing.T) { + base := Schedule{PURLs: []string{"pkg:docker/alpine@3"}, Platforms: []string{"linux/amd64"}, Interval: time.Hour} + if scheduleKey(base) != scheduleKey(Schedule{PURLs: []string{"pkg:docker/alpine@3"}, Platforms: []string{"linux/amd64"}, Interval: time.Hour}) { + t.Error("same schedule gave different keys") + } + for name, other := range map[string]Schedule{ + "purls": {PURLs: []string{"pkg:docker/alpine@4"}, Platforms: base.Platforms, Interval: base.Interval}, + "platforms": {PURLs: base.PURLs, Platforms: []string{"linux/arm64"}, Interval: base.Interval}, + "interval": {PURLs: base.PURLs, Platforms: base.Platforms, Interval: 2 * time.Hour}, + } { + if scheduleKey(other) == scheduleKey(base) { + t.Errorf("changing %s kept the key", name) + } + } +} diff --git a/internal/server/eviction.go b/internal/server/eviction.go index 5bbae41e..66887d18 100644 --- a/internal/server/eviction.go +++ b/internal/server/eviction.go @@ -39,7 +39,9 @@ func (s *Server) startEvictionLoop(ctx context.Context) { } func (s *Server) runEviction(ctx context.Context, maxSize int64) { - evictLRU(ctx, s.db, s.storage, s.logger, maxSize) + s.runExclusive(ctx, "eviction", func() { + evictLRU(ctx, s.db, s.storage, s.logger, maxSize) + }) } // evictLRU removes the least recently used cached artifacts and metadata diff --git a/internal/server/exclusive.go b/internal/server/exclusive.go new file mode 100644 index 00000000..5067afdd --- /dev/null +++ b/internal/server/exclusive.go @@ -0,0 +1,21 @@ +package server + +import "context" + +// runExclusive runs one pass of a background job unless another instance +// sharing the database is running it. That instance does the work, so this +// pass is skipped rather than waited for. Jobs that act on shared state, such +// as eviction against max_size, would otherwise each do all of it. +func (s *Server) runExclusive(ctx context.Context, job string, run func()) { + release, ok, err := s.db.TryLock(ctx, "job:"+job) + if err != nil { + s.logger.Warn("background job: failed to take lock, skipping this pass", "job", job, "error", err) + return + } + if !ok { + s.logger.Debug("background job: running on another instance, skipping this pass", "job", job) + return + } + defer release() + run() +} diff --git a/internal/server/exclusive_test.go b/internal/server/exclusive_test.go new file mode 100644 index 00000000..abd506b2 --- /dev/null +++ b/internal/server/exclusive_test.go @@ -0,0 +1,60 @@ +package server + +import ( + "context" + "io" + "log/slog" + "os" + "path/filepath" + "testing" + + "github.com/git-pkgs/proxy/internal/database" +) + +func TestRunExclusiveSQLiteRuns(t *testing.T) { + db, err := database.Create(filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatalf("creating database: %v", err) + } + defer func() { _ = db.Close() }() + s := &Server{db: db, logger: slog.New(slog.NewTextHandler(io.Discard, nil))} + + ran := false + s.runExclusive(context.Background(), "eviction", func() { ran = true }) + if !ran { + t.Error("job did not run on SQLite") + } +} + +// TestRunExclusiveSkipsJobRunningElsewhere checks that a pass is skipped +// while another instance sharing the Postgres database runs the job, and +// runs once that instance is done. +func TestRunExclusiveSkipsJobRunningElsewhere(t *testing.T) { + url := os.Getenv("PROXY_DATABASE_URL") + if url == "" { + t.Skip("PROXY_DATABASE_URL not set") + } + open := func() *Server { + db, err := database.OpenPostgres(url) + if err != nil { + t.Fatalf("OpenPostgres: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + return &Server{db: db, logger: slog.New(slog.NewTextHandler(io.Discard, nil))} + } + a, b := open(), open() + + ranOnB := 0 + a.runExclusive(context.Background(), "eviction", func() { + b.runExclusive(context.Background(), "eviction", func() { ranOnB++ }) + // A different job is not held up. + b.runExclusive(context.Background(), "retention", func() { ranOnB += 10 }) + }) + if ranOnB != 10 { + t.Fatalf("while a ran eviction, b ran %d (want only retention, 10)", ranOnB) + } + b.runExclusive(context.Background(), "eviction", func() { ranOnB++ }) + if ranOnB != 11 { + t.Errorf("after a finished, b's eviction did not run") + } +} diff --git a/internal/server/gradle_cache_eviction.go b/internal/server/gradle_cache_eviction.go index 7f546d16..10980bb9 100644 --- a/internal/server/gradle_cache_eviction.go +++ b/internal/server/gradle_cache_eviction.go @@ -34,7 +34,7 @@ func (s *Server) startGradleBuildCacheEviction(ctx context.Context) { "max_size_bytes", maxSize, "interval", interval) - sweep := func() { + sweepOnce := func() { deletedCount, freedBytes, err := sweepGradleBuildCache(ctx, s.storage, lister, maxAge, maxSize, time.Now()) if err != nil { s.logger.Warn("gradle cache eviction sweep failed", "error", err) @@ -47,6 +47,7 @@ func (s *Server) startGradleBuildCacheEviction(ctx context.Context) { } } + sweep := func() { s.runExclusive(ctx, "gradle-cache-eviction", sweepOnce) } sweep() go func() { diff --git a/internal/server/mirror_schedule.go b/internal/server/mirror_schedule.go index 3f69e7fa..fb32f249 100644 --- a/internal/server/mirror_schedule.go +++ b/internal/server/mirror_schedule.go @@ -24,6 +24,7 @@ func (s *Server) startMirrorSchedules(ctx context.Context, proxy *handler.Proxy) schedules[i] = mirror.Schedule{PURLs: schedule.Images, Platforms: schedule.Platforms, Interval: schedule.IntervalDuration()} } s.scheduler = mirror.NewScheduler(m, schedules, s.logger) + s.scheduler.Claim = s.db.ClaimJobRun s.scheduler.Start(ctx) s.logger.Info("mirror schedules enabled", "schedules", len(schedules)) } diff --git a/internal/server/reclaim.go b/internal/server/reclaim.go index aa986a1c..9437023c 100644 --- a/internal/server/reclaim.go +++ b/internal/server/reclaim.go @@ -34,7 +34,9 @@ func (s *Server) startReclaimLoop(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - reclaimDue(ctx, s.db, s.storage, s.logger, time.Now().Add(-grace), reclaimTickBudget) + s.runExclusive(ctx, "reclaim", func() { + reclaimDue(ctx, s.db, s.storage, s.logger, time.Now().Add(-grace), reclaimTickBudget) + }) } } } diff --git a/internal/server/retention.go b/internal/server/retention.go index 9a2e9fbc..520b4cc2 100644 --- a/internal/server/retention.go +++ b/internal/server/retention.go @@ -47,18 +47,24 @@ func (s *Server) startRetentionLoop(ctx context.Context) { ticker := time.NewTicker(s.cfg.ParseRetentionSweepInterval()) defer ticker.Stop() - sweepRetention(ctx, s.db, s.logger, reg, rules, time.Now()) + s.sweepRetention(ctx, reg, rules) for { select { case <-ctx.Done(): return case <-ticker.C: - sweepRetention(ctx, s.db, s.logger, reg, rules, time.Now()) + s.sweepRetention(ctx, reg, rules) } } } +func (s *Server) sweepRetention(ctx context.Context, reg *retention.Registry, rules retention.Rules) { + s.runExclusive(ctx, "retention", func() { + sweepRetention(ctx, s.db, s.logger, reg, rules, time.Now()) + }) +} + // notCovered returns the known ecosystem keys the global default cannot reach. func notCovered(covered []string) []string { var missing []string From b907ef0abf5ad2575327f771aecdbb71c690f2f1 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 23 Oct 2026 16:23:54 +0200 Subject: [PATCH 2/4] Run the Postgres tests in CI The tests gated on PROXY_DATABASE_URL were always skipped in CI. A new job runs them against a Postgres 17 service container. --- .github/workflows/ci.yml | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 64ecd809..c8d0bc79 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,6 +31,40 @@ jobs: - name: Test run: go test -v -race ./... + postgres: + name: test (postgres) + runs-on: ubuntu-latest + services: + postgres: + image: postgres:17@sha256:b8faeff7712663c2773a190181e1246e37e364b887611c04658c28b44e2408fe + env: + POSTGRES_PASSWORD: test + POSTGRES_DB: proxy + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U postgres -d proxy" + --health-interval 2s + --health-timeout 5s + --health-retries 30 + + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Set up Go + uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: go.mod + + # The packages with tests that need PROXY_DATABASE_URL. -p 1 because + # their tests drop and recreate tables in the one shared database. + - name: Test + env: + PROXY_DATABASE_URL: postgres://postgres:test@localhost:5432/proxy?sslmode=disable + run: go test -v -race -p 1 ./internal/database/... ./internal/handler/... ./internal/server/... + lint: runs-on: ubuntu-latest steps: From 3fcdb0b652d7f20c818ca632dcfecd077ecbbbdf Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 23 Oct 2026 16:30:43 +0200 Subject: [PATCH 3/4] Close the SQLite test databases in the lock tests Windows cannot remove the temporary directory while the file is open. --- internal/database/lock_test.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/internal/database/lock_test.go b/internal/database/lock_test.go index c49906d8..00750a60 100644 --- a/internal/database/lock_test.go +++ b/internal/database/lock_test.go @@ -13,6 +13,7 @@ import ( func TestTryLockSQLiteAlwaysGranted(t *testing.T) { db := createTestDB(t) + defer func() { _ = db.Close() }() release, ok, err := db.TryLock(context.Background(), "job:test") if err != nil || !ok { t.Fatalf("TryLock = %v, %v; want granted", ok, err) @@ -51,7 +52,11 @@ func TestClaimJobRun(t *testing.T) { // forEachDialect runs fn against SQLite and, when PROXY_DATABASE_URL is set, // against Postgres. func forEachDialect(t *testing.T, fn func(t *testing.T, db *DB)) { - t.Run("sqlite", func(t *testing.T) { fn(t, createTestDB(t)) }) + t.Run("sqlite", func(t *testing.T) { + db := createTestDB(t) + defer func() { _ = db.Close() }() + fn(t, db) + }) t.Run("postgres", func(t *testing.T) { db := createTestPostgresDB(t) if db == nil { From 743289c8ca5a575bd753904b3931fee9a37e8a60 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 23 Oct 2026 19:07:32 +0200 Subject: [PATCH 4/4] Let an interrupted mirror schedule run again A schedule's claim was kept when its run was cut off, so after a restart every instance skipped it for most of an interval. The scheduler now drops the claim when the run ends with its context cancelled. --- docs/configuration.md | 2 +- internal/database/lock.go | 9 ++++++ internal/database/lock_test.go | 12 ++++++++ internal/mirror/schedule.go | 11 ++++++- internal/mirror/schedule_test.go | 48 ++++++++++++++++++++++++++++++ internal/server/mirror_schedule.go | 1 + 6 files changed, 81 insertions(+), 2 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 875f8fcb..6826bf0e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1508,7 +1508,7 @@ mirror: Each run checks the tag with a HEAD request. If the tag still points to the cached digest, nothing else is fetched. If it moved, the new manifests and blobs are fetched first and the tag is switched only when all of them are cached, so a failed run leaves the previous image pullable. Images of one schedule are mirrored one after the other. -- The first run starts within ten minutes of startup, later runs every `interval` ±10 %, so restarts and several instances don't hit the registry at once. Instances sharing a PostgreSQL database run each schedule once per interval: the first to reach it records the run, and the others skip it. +- The first run starts within ten minutes of startup, later runs every `interval` ±10 %, so restarts and several instances don't hit the registry at once. Instances sharing a PostgreSQL database run each schedule once per interval: the first to reach it records the run, and the others skip it. A run cut off by a shutdown is not recorded, so the next instance to reach the schedule runs it. - Digest-pinned PURLs never move and are rejected. A `pkg:oci` digest with `tag=` is accepted; the tag is refreshed and a warning is logged when it no longer matches the digest. - The previous digest stays cached, so pulls by digest keep working, until eviction or retention removes it. - Docker Hub does not count HEAD requests against its pull limit, but every changed manifest is a pull. Set `platforms` and `upstream.auth` for `registry-1.docker.io` to stay under the anonymous limit. diff --git a/internal/database/lock.go b/internal/database/lock.go index b245c629..f4b1f70e 100644 --- a/internal/database/lock.go +++ b/internal/database/lock.go @@ -96,3 +96,12 @@ func (db *DB) ClaimJobRun(name string, minGap time.Duration) (bool, error) { } return n > 0, nil } + +// ReleaseJobRun drops the named job's last claim, so the next attempt on any +// instance runs it. It is for a run that was interrupted before it finished. +func (db *DB) ReleaseJobRun(name string) error { + if _, err := db.Exec(db.Rebind(`DELETE FROM job_runs WHERE name = ?`), name); err != nil { + return fmt.Errorf("releasing job run %s: %w", name, err) + } + return nil +} diff --git a/internal/database/lock_test.go b/internal/database/lock_test.go index 00750a60..81292e23 100644 --- a/internal/database/lock_test.go +++ b/internal/database/lock_test.go @@ -46,6 +46,18 @@ func TestClaimJobRun(t *testing.T) { if err != nil || !claimed { t.Fatalf("claim after the gap = %v, %v; want claimed", claimed, err) } + // A released run can be claimed again within the gap. + if err := db.ReleaseJobRun("mirror-schedule:a"); err != nil { + t.Fatal(err) + } + claimed, err = db.ClaimJobRun("mirror-schedule:a", time.Hour) + if err != nil || !claimed { + t.Fatalf("claim after release = %v, %v; want claimed", claimed, err) + } + claimed, err = db.ClaimJobRun("mirror-schedule:b", time.Hour) + if err != nil || claimed { + t.Fatalf("other job after release = %v, %v; want still claimed", claimed, err) + } }) } diff --git a/internal/mirror/schedule.go b/internal/mirror/schedule.go index 55d6ab4a..89bb5121 100644 --- a/internal/mirror/schedule.go +++ b/internal/mirror/schedule.go @@ -35,6 +35,9 @@ type Scheduler struct { // instance keeps its own timer, so without it every instance would run // every schedule. Claim func(key string, minGap time.Duration) (bool, error) + // Release, when set, drops the claim of a run that was interrupted, so + // the next attempt runs it rather than waiting an interval. + Release func(key string) error // randN returns a value in [0, n); tests replace it. randN func(n int64) int64 } @@ -83,11 +86,12 @@ func (s *Scheduler) loop(ctx context.Context, index int, schedule Schedule) { } func (s *Scheduler) run(ctx context.Context, index int, schedule Schedule) { + key := scheduleKey(schedule) if s.Claim != nil { // Later runs come at least interval - 2*spread apart, so a run // started more recently than that was another instance's. minGap := schedule.Interval - 2*(schedule.Interval/scheduleJitter) //nolint:mnd // two spreads of margin - claimed, err := s.Claim(scheduleKey(schedule), minGap) + claimed, err := s.Claim(key, minGap) if err != nil { s.logger.Warn("mirror schedule: failed to claim run, skipping it", "schedule", index, "error", err) return @@ -100,6 +104,11 @@ func (s *Scheduler) run(ctx context.Context, index int, schedule Schedule) { started := time.Now() progress, err := s.mirror.Run(ctx, &PURLSource{PURLs: schedule.PURLs, Platforms: schedule.Platforms}) if ctx.Err() != nil { + if s.Claim != nil && s.Release != nil { + if err := s.Release(key); err != nil { + s.logger.Warn("mirror schedule: failed to release interrupted run", "schedule", index, "error", err) + } + } return } if err != nil { diff --git a/internal/mirror/schedule_test.go b/internal/mirror/schedule_test.go index 53bdea2e..b5b0143a 100644 --- a/internal/mirror/schedule_test.go +++ b/internal/mirror/schedule_test.go @@ -216,6 +216,54 @@ func TestSchedulerSkipsRunClaimedElsewhere(t *testing.T) { } } +// TestSchedulerReleasesInterruptedRun checks that a run cut off by shutdown +// does not count as done: the next attempt, on a restarted instance or +// another one, runs the schedule again. +func TestSchedulerReleasesInterruptedRun(t *testing.T) { + images := &scheduleImages{block: make(chan struct{})} + schedule := Schedule{PURLs: []string{"pkg:docker/alpine@3"}, Interval: time.Hour} + s := newTestScheduler(images, schedule) + // The claims of a shared database: a claimed key is not claimed again. + var mu sync.Mutex + claims := map[string]bool{} + s.Claim = func(key string, _ time.Duration) (bool, error) { + mu.Lock() + defer mu.Unlock() + if claims[key] { + return false, nil + } + claims[key] = true + return true, nil + } + s.Release = func(key string) error { + mu.Lock() + defer mu.Unlock() + delete(claims, key) + return nil + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + s.run(ctx, 0, schedule) + close(done) + }() + waitFor(t, "the run to start", func() bool { return images.calls.Load() == 1 }) + cancel() + <-done + + close(images.block) + s.run(context.Background(), 0, schedule) + if got := images.calls.Load(); got != 2 { + t.Fatalf("calls = %d, want 2: the interrupted run blocked the next one", got) + } + // A finished run keeps its claim. + s.run(context.Background(), 0, schedule) + if got := images.calls.Load(); got != 2 { + t.Fatalf("calls = %d, want 2 after a finished run", got) + } +} + func TestScheduleKey(t *testing.T) { base := Schedule{PURLs: []string{"pkg:docker/alpine@3"}, Platforms: []string{"linux/amd64"}, Interval: time.Hour} if scheduleKey(base) != scheduleKey(Schedule{PURLs: []string{"pkg:docker/alpine@3"}, Platforms: []string{"linux/amd64"}, Interval: time.Hour}) { diff --git a/internal/server/mirror_schedule.go b/internal/server/mirror_schedule.go index fb32f249..3b023375 100644 --- a/internal/server/mirror_schedule.go +++ b/internal/server/mirror_schedule.go @@ -25,6 +25,7 @@ func (s *Server) startMirrorSchedules(ctx context.Context, proxy *handler.Proxy) } s.scheduler = mirror.NewScheduler(m, schedules, s.logger) s.scheduler.Claim = s.db.ClaimJobRun + s.scheduler.Release = s.db.ReleaseJobRun s.scheduler.Start(ctx) s.logger.Info("mirror schedules enabled", "schedules", len(schedules)) }