Fetch each cache miss once across instances

.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:

README.md

@@ -1474,6 +1474,7 @@ The proxy exposes Prometheus metrics at `GET /metrics`. All metric names are pre
 | `proxy_metadata_cache_entries` | gauge | `ecosystem` | Number of cached metadata documents per ecosystem. |
 | `proxy_artifacts_evicted_total` | counter | `reason`, `ecosystem` | Artifacts evicted by the size limit (`lru`) or by age (`retention`) |
 | `proxy_metadata_evicted_total` | counter | `ecosystem` | Cached metadata documents evicted by the size limit |
+| `proxy_fetch_lease_waits_total` | counter | `kind` | Cache misses (`artifact` or `metadata`) that waited for another instance sharing the database to fetch them |
 | `proxy_upstream_fetch_duration_seconds` | histogram | `ecosystem` | Time spent fetching from upstream |
 | `proxy_upstream_errors_total` | counter | `ecosystem`, `error_type` | Upstream fetch failures |
 | `proxy_storage_operation_duration_seconds` | histogram | `operation` | Storage read/write latency |

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: ""

docs/configuration.md

@@ -197,6 +197,20 @@ 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 fetched from upstream once across instances. The instance that fetches holds a lease on the miss in the `fetch_leases` table and renews it while it downloads. Another instance missing the same file waits until it is stored and serves the stored copy; `proxy_fetch_lease_waits_total` counts those waits. If the fetch fails, the instances waiting on it stop waiting together: each serves its stale metadata copy if it has one, and otherwise fetches itself, so a failing upstream costs every instance one attempt at the same time rather than one after another. A lease of an instance that stopped expires after 30 seconds. Metadata is shared this way only with [`cache_metadata`](#metadata-caching) and a `metadata_ttl` above zero, since without a stored copy there is nothing to wait for. A cache hit costs no extra query; a miss costs one.
+
+Each instance 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 +1510,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.

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

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", "fetch_leases", "migrations", "schema_info"}
 	for _, table := range tables {
 		_, _ = db.Exec("DROP TABLE IF EXISTS " + table + " CASCADE")
 	}

internal/database/fetch_leases.go

@@ -0,0 +1,94 @@
+package database
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+)
+
+// Fetch leases let instances sharing a Postgres database fetch each cache
+// miss from upstream once. The instance that fetches holds the lease for the
+// miss's key and renews it until the result is stored; the others wait for
+// the stored result. Expiry uses the database clock, so instances whose
+// clocks disagree still agree on when a lease has run out. SQLite is
+// single-instance, so these are only used on Postgres.
+
+// AcquireFetchLease takes the lease on key for owner unless another owner
+// holds one that has neither expired nor failed, and reports whether it did.
+func (db *DB) AcquireFetchLease(key, owner string, ttl time.Duration) (bool, error) {
+	res, err := db.Exec(`
+		INSERT INTO fetch_leases (key, owner, expires_at)
+		VALUES ($1, $2, now() + $3 * interval '1 millisecond')
+		ON CONFLICT (key) DO UPDATE SET owner = excluded.owner, expires_at = excluded.expires_at, failed = FALSE
+		WHERE fetch_leases.expires_at < now() OR fetch_leases.failed`, key, owner, ttl.Milliseconds())
+	if err != nil {
+		return false, fmt.Errorf("acquiring fetch lease: %w", err)
+	}
+	n, err := res.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("acquiring fetch lease: %w", err)
+	}
+	return n > 0, nil
+}
+
+// RenewFetchLease extends owner's lease on key by ttl from now, and reports
+// whether owner still held it.
+func (db *DB) RenewFetchLease(key, owner string, ttl time.Duration) (bool, error) {
+	res, err := db.Exec(`
+		UPDATE fetch_leases SET expires_at = now() + $3 * interval '1 millisecond'
+		WHERE key = $1 AND owner = $2`, key, owner, ttl.Milliseconds())
+	if err != nil {
+		return false, fmt.Errorf("renewing fetch lease: %w", err)
+	}
+	n, err := res.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("renewing fetch lease: %w", err)
+	}
+	return n > 0, nil
+}
+
+// FetchLeaseHolder returns the owner of the lease on key, and whether its
+// fetch failed. ok is false when no lease is held or it has expired.
+func (db *DB) FetchLeaseHolder(key string) (owner string, failed, ok bool, err error) {
+	err = db.QueryRow(`
+		SELECT owner, failed FROM fetch_leases
+		WHERE key = $1 AND expires_at >= now()`, key).Scan(&owner, &failed)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", false, false, nil
+	}
+	if err != nil {
+		return "", false, false, fmt.Errorf("reading fetch lease: %w", err)
+	}
+	return owner, failed, true, nil
+}
+
+// FailFetchLease marks owner's lease on key as failed. Instances waiting on
+// it stop waiting, and the next AcquireFetchLease takes it at once.
+func (db *DB) FailFetchLease(key, owner string) error {
+	if _, err := db.Exec(`UPDATE fetch_leases SET failed = TRUE WHERE key = $1 AND owner = $2`, key, owner); err != nil {
+		return fmt.Errorf("failing fetch lease: %w", err)
+	}
+	return nil
+}
+
+// ReleaseFetchLease drops owner's lease on key.
+func (db *DB) ReleaseFetchLease(key, owner string) error {
+	if _, err := db.Exec(`DELETE FROM fetch_leases WHERE key = $1 AND owner = $2`, key, owner); err != nil {
+		return fmt.Errorf("releasing fetch lease: %w", err)
+	}
+	return nil
+}
+
+// DeleteExpiredFetchLeases removes leases that expired more than grace ago,
+// left by instances that stopped while fetching.
+func (db *DB) DeleteExpiredFetchLeases(grace time.Duration) (int64, error) {
+	if db.dialect != DialectPostgres {
+		return 0, nil
+	}
+	res, err := db.Exec(`DELETE FROM fetch_leases WHERE expires_at < now() - $1 * interval '1 millisecond'`, grace.Milliseconds())
+	if err != nil {
+		return 0, fmt.Errorf("deleting expired fetch leases: %w", err)
+	}
+	return res.RowsAffected()
+}

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
+}