Write batches through one connection with a 1 GiB cache, in key order (closes #96)

Every write now uses one connection taken from the pool when the
database opens, under the database lock. Its 1 GiB page cache keeps
the pages the route writes touch from one batch to the next. A reading
connection's cache is emptied whenever another connection has written,
so the 9 reading connections drop to 16 MiB each. The soft and hard
heap limits rise to 1.5 and 2 GiB, above the caches; README "Memory"
adds up the new budget. Vacuum, Analyze and Checkpoint check their
context after the lock, which the kept connection does not. Batch
writes sort their rows by peer, then prefix. The benchmark also times
the largest prefix handler batch, 250,000 updates.

Model: opus-5-5

internal/database/database.go

@@ -1,18 +1,21 @@
 package database
 
 import (
+	"cmp"
 	"context"
 	"database/sql"
 	_ "embed"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"math/rand/v2"
 	"net"
 	"net/netip"
 	"os"
 	"path/filepath"
 	"runtime"
+	"slices"
 	"sync"
 	"time"
 
@@ -51,20 +54,31 @@ const (
 	minLookupMaskLengthV6 = 16
 )
 
-// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every
-// pooled connection gets them; the heap limits are process-wide and set once.
+// SQLite memory tuning (README "Memory"). cache_size and busy_timeout go in the
+// DSN so every pooled connection gets them; the connection kept for writing
+// then sets its own cache size. The heap limits are process-wide and set once.
 const (
-	// sqliteCacheSizeKiB is the per-connection page cache; negative means KiB.
-	// -65536 = 64 MiB, so at most 640 MiB across the 10-connection pool.
-	sqliteCacheSizeKiB = -65536
+	// maxConns is the size of the connection pool: the connection kept for
+	// writing and 9 reading connections.
+	maxConns = 10
+	// sqliteReadCacheSizeKiB is the page cache of each reading connection;
+	// negative means KiB. -16384 = 16 MiB, at most 144 MiB for the 9. A reading
+	// connection's cache is emptied whenever another connection has written
+	// since its last read, many times a minute, so a large one is wasted.
+	sqliteReadCacheSizeKiB = -16384
+	// sqliteWriteCacheSizeKiB is the page cache of the connection kept for
+	// writing, -1048576 = 1 GiB. It keeps its pages from one write to the next,
+	// so the index pages the route writes touch stay in memory.
+	sqliteWriteCacheSizeKiB = -1048576
 	// sqliteBusyTimeoutMs is how long a connection waits on a locked database.
 	sqliteBusyTimeoutMs = 5000
-	// sqliteSoftHeapLimitBytes (1 GiB) makes SQLite recycle its cache rather
-	// than allocate once its C heap passes this size.
-	sqliteSoftHeapLimitBytes = 1073741824
-	// sqliteHardHeapLimitBytes (1.5 GiB) fails a statement with SQLITE_NOMEM
+	// sqliteSoftHeapLimitBytes (1.5 GiB) makes SQLite recycle its cache rather
+	// than allocate once its C heap passes this size. It is above the caches
+	// together, so it does not shrink them.
+	sqliteSoftHeapLimitBytes = 1610612736
+	// sqliteHardHeapLimitBytes (2 GiB) fails a statement with SQLITE_NOMEM
 	// instead of growing the C heap without bound.
-	sqliteHardHeapLimitBytes = 1610612736
+	sqliteHardHeapLimitBytes = 2147483648
 )
 
 // Settings of the ANALYZE that Analyze runs.
@@ -105,7 +119,10 @@ var (
 
 // Database manages the SQLite database connection and operations.
 type Database struct {
-	db       *sql.DB
+	db *sql.DB
+	// writer is the connection kept for writing, taken from db when the
+	// database opens. Every write uses it while holding mu.
+	writer   *sql.Conn
 	logger   *logger.Logger
 	path     string
 	mu       sync.Mutex
@@ -143,7 +160,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 		"file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+
 			"&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate",
 		dbPath,
-		sqliteCacheSizeKiB,
+		sqliteReadCacheSizeKiB,
 		sqliteBusyTimeoutMs,
 	)
 
@@ -152,20 +169,34 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 		return nil, fmt.Errorf("failed to open database: %w", err)
 	}
 
-	err = db.PingContext(context.Background())
+	ctx := context.Background()
+
+	err = db.PingContext(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to ping database: %w", err)
 	}
 
 	// Set connection pool parameters
 	// Multiple connections allow concurrent reads while writes are serialized
-	const maxConns = 10
 	db.SetMaxOpenConns(maxConns)
 	db.SetMaxIdleConns(maxConns)
 	db.SetConnMaxLifetime(0)
 
+	writer, err := db.Conn(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open the writing connection: %w", err)
+	}
+
+	_, err = writer.ExecContext(ctx,
+		fmt.Sprintf("PRAGMA cache_size=%d", sqliteWriteCacheSizeKiB))
+	if err != nil {
+		return nil, fmt.Errorf("failed to set the writing connection's cache: %w",
+			err)
+	}
+
 	database := &Database{
-		db: db, logger: logger, path: dbPath, counts: &liveCounts{},
+		db: db, writer: writer, logger: logger, path: dbPath,
+		counts: &liveCounts{},
 	}
 
 	err = database.Initialize()
@@ -187,10 +218,11 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 // Initialize creates the database schema if it doesn't exist.
 func (d *Database) Initialize() error {
 	// Set SQLite pragmas for performance. Per-connection settings (cache_size,
-	// synchronous, busy_timeout, journal_mode) live in the DSN; temp_store is
-	// left at its default so DISTINCT temp B-trees spill to disk instead of C
-	// heap. The heap limits below are process-wide, so setting them once here is
-	// enough for the whole pool.
+	// synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the
+	// writing connection's cache_size; temp_store is left at its default so
+	// DISTINCT temp B-trees spill to disk instead of C heap. The heap limits
+	// below are process-wide, so setting them once here is enough for the whole
+	// pool.
 	pragmas := []string{
 		"PRAGMA journal_mode=WAL", // Write-Ahead Logging
 		fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes),
@@ -207,7 +239,7 @@ func (d *Database) Initialize() error {
 	// Run WAL checkpoint on startup to consolidate any existing WAL
 	var walPages, checkpointedPages, movedPages int
 
-	err := d.db.QueryRowContext(context.Background(),
+	err := d.writer.QueryRowContext(context.Background(),
 		"PRAGMA wal_checkpoint(TRUNCATE)",
 	).Scan(&walPages, &checkpointedPages, &movedPages)
 	if err != nil {
@@ -228,9 +260,9 @@ func (d *Database) Initialize() error {
 	return nil
 }
 
-// Close closes the database connection.
+// Close closes the writing connection and then the connection pool.
 func (d *Database) Close() error {
-	return d.db.Close()
+	return errors.Join(d.writer.Close(), d.db.Close())
 }
 
 // A live-route upsert sets the origin AS of the peer's route for the prefix
@@ -428,9 +460,18 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
 	}
 	defer d.rollback(tx)
 
+	// The rows are written in key order, peer then prefix, so consecutive writes
+	// land on the same index pages. The sort is stable, so of two updates of
+	// one route the later one is still written last.
+	sorted := slices.SortedStableFunc(slices.Values(routes),
+		func(a, b *LiveRoute) int {
+			return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP),
+				cmp.Compare(a.Prefix, b.Prefix))
+		})
+
 	var routesV4, routesV6 []*LiveRoute
 
-	for _, route := range routes {
+	for _, route := range sorted {
 		if route.IPVersion == ipVersionV4 {
 			routesV4 = append(routesV4, route)
 		} else {
@@ -574,9 +615,17 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 	}
 	defer d.rollback(tx)
 
+	// The rows are deleted in key order, peer then prefix, as
+	// UpsertLiveRouteBatch writes them.
+	sorted := slices.SortedFunc(slices.Values(deletions),
+		func(a, b LiveRouteDeletion) int {
+			return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP),
+				cmp.Compare(a.Prefix, b.Prefix))
+		})
+
 	var deletionsV4, deletionsV6 []LiveRouteDeletion
 
-	for _, del := range deletions {
+	for _, del := range sorted {
 		if del.IPVersion == ipVersionV4 {
 			deletionsV4 = append(deletionsV4, del)
 		} else {
@@ -780,9 +829,10 @@ func upsertPrefixesInTable(
 
 	var newPrefixes int
 
-	for prefix, timestamp := range prefixes {
+	// The rows are written in key order, as UpsertLiveRouteBatch writes routes.
+	for _, prefix := range slices.Sorted(maps.Keys(prefixes)) {
 		inserted, err := upsertPrefix(ctx, updateStmt, insertStmt, prefix,
-			timestamp)
+			prefixes[prefix])
 		if err != nil {
 			return 0, err
 		}
@@ -1411,25 +1461,25 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
 
 	// The write lock is held, so no other writer can insert this key between the
 	// update and the insert even though they are separate autocommit statements.
-	setOrigin, err := d.db.PrepareContext(ctx, originSQL)
+	setOrigin, err := d.writer.PrepareContext(ctx, originSQL)
 	if err != nil {
 		return fmt.Errorf("failed to prepare origin statement: %w", err)
 	}
 	defer func() { _ = setOrigin.Close() }()
 
-	upd, err := d.db.PrepareContext(ctx, updateSQL)
+	upd, err := d.writer.PrepareContext(ctx, updateSQL)
 	if err != nil {
 		return fmt.Errorf("failed to prepare update statement: %w", err)
 	}
 	defer func() { _ = upd.Close() }()
 
-	ins, err := d.db.PrepareContext(ctx, insertSQL)
+	ins, err := d.writer.PrepareContext(ctx, insertSQL)
 	if err != nil {
 		return fmt.Errorf("failed to prepare insert statement: %w", err)
 	}
 	defer func() { _ = ins.Close() }()
 
-	has, err := d.db.PrepareContext(ctx, lookupSQL)
+	has, err := d.writer.PrepareContext(ctx, lookupSQL)
 	if err != nil {
 		return fmt.Errorf("failed to prepare prefix lookup statement: %w", err)
 	}
@@ -2025,7 +2075,7 @@ func (d *Database) UpdateASNWHOIS(
 		WHERE asn = ?
 	`
 
-	_, err := d.db.ExecContext(ctx, query,
+	_, err := d.writer.ExecContext(ctx, query,
 		update.ASName,
 		update.OrgName,
 		update.OrgID,
@@ -2182,7 +2232,8 @@ func (d *Database) GetIPInfoContext(
 // Vacuum runs incremental vacuum to return up to 1000 unused pages to the
 // filesystem. Like Analyze and Checkpoint, it holds the database lock, so it
 // and the batch writes wait for each other instead of failing on SQLite's
-// write lock.
+// write lock, and returns ctx's error without running if ctx ended while it
+// waited: the writing connection it uses does not check ctx itself.
 func (d *Database) Vacuum(ctx context.Context) error {
 	// Free up to 1000 pages per call (~4MB with default 4KB page size)
 	// This keeps each vacuum operation quick
@@ -2191,7 +2242,12 @@ func (d *Database) Vacuum(ctx context.Context) error {
 	d.lock("Vacuum")
 	defer d.unlock()
 
-	rows, err := d.db.QueryContext(ctx,
+	err := ctx.Err()
+	if err != nil {
+		return fmt.Errorf("failed to run incremental vacuum: %w", err)
+	}
+
+	rows, err := d.writer.QueryContext(ctx,
 		fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree))
 	if err != nil {
 		return fmt.Errorf("failed to run incremental vacuum: %w", err)
@@ -2221,9 +2277,14 @@ func (d *Database) Analyze(ctx context.Context) error {
 	d.lock("Analyze")
 	defer d.unlock()
 
+	err := ctx.Err()
+	if err != nil {
+		return fmt.Errorf("failed to analyze database: %w", err)
+	}
+
 	// analysis_limit is a setting of the connection, so it is set in the same
 	// call as PRAGMA optimize, which runs both statements on one connection.
-	_, err := d.db.ExecContext(ctx, fmt.Sprintf(
+	_, err = d.writer.ExecContext(ctx, fmt.Sprintf(
 		"PRAGMA analysis_limit=%d; PRAGMA optimize=%#x",
 		analysisLimit, optimizeAllTables))
 	if err != nil {
@@ -2240,7 +2301,12 @@ func (d *Database) Checkpoint(ctx context.Context) error {
 	d.lock("Checkpoint")
 	defer d.unlock()
 
-	_, err := d.db.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)")
+	err := ctx.Err()
+	if err != nil {
+		return fmt.Errorf("failed to checkpoint WAL: %w", err)
+	}
+
+	_, err = d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)")
 	if err != nil {
 		return fmt.Errorf("failed to checkpoint WAL: %w", err)
 	}
@@ -2296,9 +2362,9 @@ func (d *Database) unlock() {
 		"held_by", lockedBy, "duration_ms", holdDuration.Milliseconds())
 }
 
-// beginTx starts a new transaction with logging
+// beginTx starts a new transaction on the writing connection, with logging
 func (d *Database) beginTx() (*loggingTx, error) {
-	tx, err := d.db.BeginTx(context.Background(), nil)
+	tx, err := d.writer.BeginTx(context.Background(), nil)
 	if err != nil {
 		return nil, err
 	}