Read SQLite through a separate read-only pool

docs/architecture.md

@@ -96,6 +96,8 @@ the full artifact digest because they read only part of the artifact.
 
 SQLite or PostgreSQL database for cache metadata. SQLite uses `modernc.org/sqlite` (pure Go, no CGO). PostgreSQL uses `lib/pq`.
 
+For SQLite, `DB` holds two pools: one connection for writes and a pool of read-only (`query_only`) connections for reads. `Get`, `Select`, `Query` and `QueryRow` go to the reader; `Exec`, transactions and everything else go to the writer, so a read inside a transaction sees the transaction's writes. A write sent to the reader fails instead of running. WAL mode lets the readers run while the writer commits. PostgreSQL and in-memory SQLite use one pool for both.
+
 The schema is compatible with [git-pkgs](https://github.com/git-pkgs) databases. The proxy adds the `artifacts` and `vulnerabilities` tables on top of the shared `packages` and `versions` tables, so both tools can point at the same database.
 
 **Tables:**

docs/configuration.md

@@ -185,6 +185,8 @@ database:
 | `database.driver` | `PROXY_DATABASE_DRIVER` | `-database-driver` | `sqlite` or `postgres` |
 | `database.path` | `PROXY_DATABASE_PATH` | `-database-path` | SQLite file path |
 
+The database runs in WAL mode. Writes go through a single connection, since SQLite allows one writer at a time. Reads use a separate pool of read-only connections (one per CPU, between 2 and 8), so request lookups don't wait behind writes, stats queries or eviction scans. An in-memory database (`:memory:`) uses one connection for both.
+
 ### PostgreSQL
 
 ```yaml
@@ -213,7 +215,7 @@ 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.
+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 writes through a single SQLite connection, so with a write per hit, concurrent downloads queue behind each other.
 
 ```yaml
 database:

internal/database/database.go

@@ -2,9 +2,12 @@ package database
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
+	"strings"
 	"time"
 
 	"github.com/jmoiron/sqlx"
@@ -33,8 +36,33 @@ const (
 	DialectPostgres Dialect = "postgres"
 )
 
+// SQLite DSN parameters. Every connection a pool opens applies them, so a
+// connection database/sql reopens gets them too. The reader pool sets
+// query_only, which makes a write sent to it fail instead of being run.
+const (
+	sqliteWriterParams = "?_busy_timeout=5000&_journal_mode=WAL&_synchronous=NORMAL"
+	sqliteReaderParams = "?_busy_timeout=5000&_query_only=1"
+)
+
+// Size bounds of the SQLite reader pool. At least two, so a slow stats or
+// eviction scan does not block request lookups even on one CPU.
+const (
+	sqliteMinReaders = 2
+	sqliteMaxReaders = 8
+)
+
+// DB is the proxy database. The embedded pool takes every write. For SQLite
+// it holds a single connection, as SQLite allows one writer at a time, and
+// reads go to a separate read-only pool so they don't queue behind writes or
+// behind each other (WAL mode lets readers run alongside the writer). For
+// Postgres and in-memory SQLite both are the same pool.
+//
+// Get, Select, Query and QueryRow (and their Context forms) use the reader.
+// Exec, transactions and everything else use the writer, so a read inside a
+// transaction goes through the transaction and sees its own writes.
 type DB struct {
 	*sqlx.DB
+	reader  *sqlx.DB
 	dialect Dialect
 	path    string
 	hits    *hitBatch
@@ -76,24 +104,51 @@ func Open(path string) (*DB, error) {
 		}
 	}
 
-	// Add busy_timeout to handle concurrent writes
-	sqlDB, err := sqlx.Open("sqlite", path+"?_busy_timeout=5000")
+	writer, err := sqlx.Open("sqlite", path+sqliteWriterParams)
 	if err != nil {
 		return nil, fmt.Errorf("opening database: %w", err)
 	}
+	// One connection serializes writes, so they never wait on SQLite's
+	// busy timeout for each other.
+	writer.SetMaxOpenConns(1)
 
-	// Limit connections to 1 for SQLite to serialize writes
-	sqlDB.SetMaxOpenConns(1)
+	// Connecting applies journal_mode=WAL before any reader opens the file.
+	if err := writer.Ping(); err != nil {
+		_ = writer.Close()
+		return nil, fmt.Errorf("opening database: %w", err)
+	}
 
-	db := &DB{DB: sqlDB, dialect: DialectSQLite, path: path}
-	if err := db.OptimizeForReads(); err != nil {
-		_ = sqlDB.Close()
-		return nil, fmt.Errorf("optimizing database: %w", err)
+	db := &DB{DB: writer, reader: writer, dialect: DialectSQLite, path: path}
+	if isMemoryPath(path) {
+		// Each connection to an in-memory database gets its own empty
+		// database, so the reads have to share the writer's connection.
+		return db, nil
 	}
 
+	reader, err := sqlx.Open("sqlite", path+sqliteReaderParams)
+	if err != nil {
+		_ = writer.Close()
+		return nil, fmt.Errorf("opening database reader: %w", err)
+	}
+	n := sqliteReaderConns()
+	reader.SetMaxOpenConns(n)
+	reader.SetMaxIdleConns(n)
+	db.reader = reader
+
 	return db, nil
 }
 
+func sqliteReaderConns() int {
+	return min(max(runtime.GOMAXPROCS(0), sqliteMinReaders), sqliteMaxReaders)
+}
+
+// isMemoryPath reports whether path names an in-memory database, as
+// ":memory:" or a URI such as "file::memory:?cache=shared" or
+// "file:x?mode=memory".
+func isMemoryPath(path string) bool {
+	return path == "" || strings.Contains(path, ":memory:") || strings.Contains(path, "mode=memory")
+}
+
 func OpenOrCreate(path string) (*DB, error) {
 	if Exists(path) {
 		return Open(path)
@@ -117,7 +172,7 @@ func OpenPostgres(url string) (*DB, error) {
 		return nil, fmt.Errorf("connecting to postgres: %w", err)
 	}
 
-	return &DB{DB: sqlDB, dialect: DialectPostgres}, nil
+	return &DB{DB: sqlDB, reader: sqlDB, dialect: DialectPostgres}, nil
 }
 
 func OpenPostgresOrCreate(url string) (*DB, error) {
@@ -179,3 +234,56 @@ func (db *DB) Path() string {
 func (db *DB) Rebind(query string) string {
 	return db.DB.Rebind(query)
 }
+
+// Get runs a read on the reader pool.
+func (db *DB) Get(dest any, query string, args ...any) error {
+	return db.reader.Get(dest, query, args...)
+}
+
+// GetContext runs a read on the reader pool.
+func (db *DB) GetContext(ctx context.Context, dest any, query string, args ...any) error {
+	return db.reader.GetContext(ctx, dest, query, args...)
+}
+
+// Select runs a read on the reader pool.
+func (db *DB) Select(dest any, query string, args ...any) error {
+	return db.reader.Select(dest, query, args...)
+}
+
+// SelectContext runs a read on the reader pool.
+func (db *DB) SelectContext(ctx context.Context, dest any, query string, args ...any) error {
+	return db.reader.SelectContext(ctx, dest, query, args...)
+}
+
+// Query runs a read on the reader pool.
+func (db *DB) Query(query string, args ...any) (*sql.Rows, error) {
+	return db.reader.Query(query, args...)
+}
+
+// QueryContext runs a read on the reader pool.
+func (db *DB) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
+	return db.reader.QueryContext(ctx, query, args...)
+}
+
+// QueryRow runs a read on the reader pool.
+func (db *DB) QueryRow(query string, args ...any) *sql.Row {
+	return db.reader.QueryRow(query, args...)
+}
+
+// QueryRowContext runs a read on the reader pool.
+func (db *DB) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {
+	return db.reader.QueryRowContext(ctx, query, args...)
+}
+
+// closePools closes the reader pool, if separate, then the writer. The
+// writer goes last so that, as the last connection, it checkpoints the WAL.
+func (db *DB) closePools() error {
+	var readerErr error
+	if db.reader != nil && db.reader != db.DB {
+		readerErr = db.reader.Close()
+	}
+	if err := db.DB.Close(); err != nil {
+		return err
+	}
+	return readerErr
+}

internal/database/hits.go

@@ -211,5 +211,5 @@ func (db *DB) Close() error {
 		b.stopOnce.Do(func() { close(b.stop) })
 		<-b.done
 	}
-	return db.DB.Close()
+	return db.closePools()
 }

internal/database/read_pool_bench_test.go

@@ -0,0 +1,157 @@
+package database
+
+import (
+	"fmt"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// benchArtifacts is the artifact count the read pool benchmarks seed: about
+// what a busy cache holds, and enough for the stats queries to take a while.
+const (
+	benchPackages  = 30_000
+	benchVersions  = 100_000
+	benchArtifacts = 300_000
+)
+
+// seedBenchDB creates a database at path with benchArtifacts artifacts
+// spread over three ecosystems, written in one statement per table.
+func seedBenchDB(b *testing.B, path string) {
+	b.Helper()
+	db, err := Create(path)
+	if err != nil {
+		b.Fatalf("Create: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+	stmts := []string{
+		fmt.Sprintf(`WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i+1 FROM n WHERE i+1 < %d)
+			INSERT INTO packages (purl, ecosystem, name, created_at, updated_at)
+			SELECT 'pkg:' || e || '/p' || i, e, 'p' || i, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
+			FROM (SELECT i, CASE i %% 3 WHEN 0 THEN 'npm' WHEN 1 THEN 'pypi' ELSE 'cargo' END AS e FROM n)`, benchPackages),
+		fmt.Sprintf(`WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i+1 FROM n WHERE i+1 < %d)
+			INSERT INTO versions (purl, package_purl, integrity, created_at, updated_at)
+			SELECT p.purl || '@' || n.i, p.purl, 'sha256-x', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
+			FROM n JOIN packages p ON p.id = (n.i %% %d) + 1`, benchVersions, benchPackages),
+		fmt.Sprintf(`WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i+1 FROM n WHERE i+1 < %d)
+			INSERT INTO artifacts (version_purl, filename, upstream_url, storage_path, content_hash, size,
+			                       content_type, fetched_at, hit_count, last_accessed_at, created_at, updated_at)
+			SELECT v.purl, 'f' || n.i, 'https://example.org/' || n.i, 'blobs/' || n.i, 'h', 1000 + n.i,
+			       'application/octet-stream', CURRENT_TIMESTAMP, n.i %% 7, CURRENT_TIMESTAMP,
+			       CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
+			FROM n JOIN versions v ON v.id = (n.i %% %d) + 1`, benchArtifacts, benchVersions),
+	}
+	for _, s := range stmts {
+		if _, err := db.Exec(s); err != nil {
+			b.Fatalf("seeding: %v", err)
+		}
+	}
+}
+
+// openBenchDB opens path. With oneConn it routes reads to the writer, as
+// before the read pool existed.
+func openBenchDB(b *testing.B, path string, oneConn bool) *DB {
+	b.Helper()
+	db, err := Open(path)
+	if err != nil {
+		b.Fatalf("Open: %v", err)
+	}
+	if oneConn {
+		_ = db.reader.Close()
+		db.reader = db.DB
+	}
+	b.Cleanup(func() { _ = db.Close() })
+	return db
+}
+
+// benchKey returns the lookup arguments for artifact i, as seeded above.
+func benchKey(i int) (pkgPURL, versionPURL, filename string) {
+	v := i % benchVersions
+	p := v % benchPackages
+	eco := [...]string{"npm", "pypi", "cargo"}[p%3]
+	pkgPURL = fmt.Sprintf("pkg:%s/p%d", eco, p)
+	return pkgPURL, fmt.Sprintf("%s@%d", pkgPURL, v), fmt.Sprintf("f%d", i)
+}
+
+func lookup(b *testing.B, db *DB, i int) {
+	pkgPURL, versionPURL, filename := benchKey(i % benchArtifacts)
+	art, err := db.GetCachedArtifact(pkgPURL, versionPURL, filename)
+	if err != nil || art == nil {
+		b.Fatalf("GetCachedArtifact(%s, %s) = %v, %v", versionPURL, filename, art, err)
+	}
+}
+
+var poolModes = []struct {
+	name    string
+	oneConn bool
+}{
+	{"one-connection", true},
+	{"read-pool", false},
+}
+
+// BenchmarkSQLiteParallelLookups runs GetCachedArtifact from parallel
+// goroutines, as concurrent downloads do.
+func BenchmarkSQLiteParallelLookups(b *testing.B) {
+	path := filepath.Join(b.TempDir(), "bench.db")
+	seedBenchDB(b, path)
+	for _, mode := range poolModes {
+		b.Run(mode.name, func(b *testing.B) {
+			db := openBenchDB(b, path, mode.oneConn)
+			var next atomic.Int64
+			b.ResetTimer()
+			b.RunParallel(func(pb *testing.PB) {
+				for pb.Next() {
+					lookup(b, db, int(next.Add(7919)))
+				}
+			})
+		})
+	}
+}
+
+// BenchmarkSQLiteLookupDuringStats runs GetCachedArtifact one at a time
+// while another goroutine keeps running GetEcosystemStats and GetCacheStats,
+// as the metrics updater and the dashboard do. It reports the slowest lookup.
+func BenchmarkSQLiteLookupDuringStats(b *testing.B) {
+	path := filepath.Join(b.TempDir(), "bench.db")
+	seedBenchDB(b, path)
+	for _, mode := range poolModes {
+		b.Run(mode.name, func(b *testing.B) {
+			db := openBenchDB(b, path, mode.oneConn)
+			stop := make(chan struct{})
+			var wg sync.WaitGroup
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				for {
+					select {
+					case <-stop:
+						return
+					default:
+					}
+					if _, err := db.GetEcosystemStats(); err != nil {
+						b.Errorf("GetEcosystemStats: %v", err)
+						return
+					}
+					if _, err := db.GetCacheStats(); err != nil {
+						b.Errorf("GetCacheStats: %v", err)
+						return
+					}
+				}
+			}()
+
+			var slowest time.Duration
+			b.ResetTimer()
+			for i := range b.N {
+				start := time.Now()
+				lookup(b, db, i*7919)
+				slowest = max(slowest, time.Since(start))
+			}
+			b.StopTimer()
+			close(stop)
+			wg.Wait()
+			b.ReportMetric(float64(slowest.Microseconds())/1000, "max-ms")
+		})
+	}
+}