Coordinate instances sharing a Postgres database

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

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

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

internal/database/lock.go

@@ -0,0 +1,107 @@
+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
+}
+
+// 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
+}