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_internal_test.go

@@ -9,6 +9,7 @@ import (
 	"math/rand/v2"
 	"net/netip"
 	"slices"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -33,6 +34,9 @@ const vacuumTestRoutes = 2000
 // once so each is a distinct SQLite connection that parsed the DSN.
 const heldConnections = 5
 
+// bytesPerKiB turns a cache_size in KiB into bytes.
+const bytesPerKiB = 1024
+
 // testPeerIP is the peer every route in the IP lookup test is learned from.
 const testPeerIP = "192.0.2.254"
 
@@ -336,7 +340,7 @@ func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) {
 
 // TestConnectionPoolPragmas holds several pooled connections open at once and
 // checks each one carries the per-connection settings from the DSN, plus the
-// process-wide hard heap limit, and sees the new file with auto_vacuum
+// process-wide heap limits, and sees the new file with auto_vacuum
 // incremental.
 func TestConnectionPoolPragmas(t *testing.T) {
 	t.Parallel()
@@ -371,9 +375,9 @@ func TestConnectionPoolPragmas(t *testing.T) {
 
 	for i, c := range conns {
 		cacheSize := readPragma(ctx, t, c, i, "cache_size")
-		if cacheSize != sqliteCacheSizeKiB {
+		if cacheSize != sqliteReadCacheSizeKiB {
 			t.Errorf("conn %d: cache_size = %d, want %d",
-				i, cacheSize, sqliteCacheSizeKiB)
+				i, cacheSize, sqliteReadCacheSizeKiB)
 		}
 
 		busyTimeout := readPragma(ctx, t, c, i, "busy_timeout")
@@ -388,6 +392,12 @@ func TestConnectionPoolPragmas(t *testing.T) {
 				i, tempStore, tempStoreMemory)
 		}
 
+		softHeapLimit := readPragma(ctx, t, c, i, "soft_heap_limit")
+		if softHeapLimit != sqliteSoftHeapLimitBytes {
+			t.Errorf("conn %d: soft_heap_limit = %d, want %d",
+				i, softHeapLimit, sqliteSoftHeapLimitBytes)
+		}
+
 		hardHeapLimit := readPragma(ctx, t, c, i, "hard_heap_limit")
 		if hardHeapLimit != sqliteHardHeapLimitBytes {
 			t.Errorf("conn %d: hard_heap_limit = %d, want %d",
@@ -402,6 +412,51 @@ func TestConnectionPoolPragmas(t *testing.T) {
 	}
 }
 
+// TestWritingConnectionCache checks that the connection kept for writing has
+// the large cache, and that the soft heap limit is above the caches of all the
+// connections together: SQLite shrinks the caches to stay under it.
+func TestWritingConnectionCache(t *testing.T) {
+	t.Parallel()
+
+	db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	ctx := t.Context()
+
+	var writeCacheSize, readCacheSize, softHeapLimit int64
+
+	err = db.writer.QueryRowContext(ctx, "PRAGMA cache_size").
+		Scan(&writeCacheSize)
+	if err != nil {
+		t.Fatalf("failed to read the writing connection's cache_size: %v", err)
+	}
+
+	if writeCacheSize != sqliteWriteCacheSizeKiB {
+		t.Errorf("writing connection: cache_size = %d, want %d",
+			writeCacheSize, sqliteWriteCacheSizeKiB)
+	}
+
+	err = db.db.QueryRowContext(ctx, "PRAGMA cache_size").Scan(&readCacheSize)
+	if err != nil {
+		t.Fatalf("failed to read a reading connection's cache_size: %v", err)
+	}
+
+	err = db.db.QueryRowContext(ctx, "PRAGMA soft_heap_limit").
+		Scan(&softHeapLimit)
+	if err != nil {
+		t.Fatalf("failed to read soft_heap_limit: %v", err)
+	}
+
+	cachesBytes := -bytesPerKiB * (writeCacheSize + (maxConns-1)*readCacheSize)
+	if softHeapLimit <= cachesBytes {
+		t.Errorf("soft_heap_limit = %d, not above the %d bytes of the caches",
+			softHeapLimit, cachesBytes)
+	}
+}
+
 // readPragma reads the integer value of PRAGMA name on connection i, c.
 func readPragma(
 	ctx context.Context, t *testing.T, c *sql.Conn, i int, name string,
@@ -1229,6 +1284,93 @@ func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) {
 	}
 }
 
+// TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out
+// of key order, the last one a second announcement of the first, and checks
+// that the rows were inserted in key order, peer then prefix (each new row's id
+// is one above the last), and that the route announced twice carries the later
+// announcement.
+func TestUpsertLiveRouteBatchWritesInKeyOrder(t *testing.T) {
+	t.Parallel()
+
+	db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	const (
+		firstOrigin, laterOrigin = 64500, 64501
+		// Three prefixes, in text order.
+		firstPrefix  = "192.0.2.0/24"
+		secondPrefix = "198.51.100.0/24"
+		thirdPrefix  = "203.0.113.0/24"
+	)
+
+	type row struct {
+		peer, prefix string
+		origin       int
+	}
+
+	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+
+	announcements := []row{
+		{secondV4PeerIP, secondPrefix, firstOrigin},
+		{testV4PeerIP, thirdPrefix, firstOrigin},
+		{secondV4PeerIP, firstPrefix, firstOrigin},
+		{testV4PeerIP, secondPrefix, firstOrigin},
+		{secondV4PeerIP, secondPrefix, laterOrigin},
+	}
+
+	routes := make([]*LiveRoute, 0, len(announcements))
+
+	for _, announced := range announcements {
+		route := mkV4Route(t, announced.prefix, announced.origin, ts)
+		route.PeerIP = announced.peer
+		routes = append(routes, route)
+	}
+
+	err = db.UpsertLiveRouteBatch(routes)
+	if err != nil {
+		t.Fatalf("UpsertLiveRouteBatch: %v", err)
+	}
+
+	rows, err := db.db.QueryContext(t.Context(),
+		"SELECT peer_ip, prefix, origin_asn FROM live_routes_v4 ORDER BY id")
+	if err != nil {
+		t.Fatalf("reading the routes: %v", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var stored []row
+
+	for rows.Next() {
+		var r row
+
+		err := rows.Scan(&r.peer, &r.prefix, &r.origin)
+		if err != nil {
+			t.Fatalf("reading a route: %v", err)
+		}
+
+		stored = append(stored, r)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		t.Fatalf("reading the routes: %v", err)
+	}
+
+	want := []row{
+		{testV4PeerIP, secondPrefix, firstOrigin},
+		{testV4PeerIP, thirdPrefix, firstOrigin},
+		{secondV4PeerIP, firstPrefix, firstOrigin},
+		{secondV4PeerIP, secondPrefix, laterOrigin},
+	}
+
+	if !slices.Equal(stored, want) {
+		t.Errorf("routes in insertion order = %+v, want %+v", stored, want)
+	}
+}
+
 // Parameters for BenchmarkPrefixBatchOnFiveMillionRoutes.
 const (
 	// benchPeers and benchPrefixes size the table the batches are written to:
@@ -1238,6 +1380,10 @@ const (
 	// benchBatchSize is how many updates each batch holds, as many as a full
 	// prefix handler batch (prefixBatchSize in internal/routewatch).
 	benchBatchSize = 25000
+	// benchMaxBatchSize is the largest batch the prefix handler writes, the
+	// one it holds while a slow write runs (prefixBatchMaxSize in
+	// internal/routewatch).
+	benchMaxBatchSize = 250000
 	// benchWithdrawalEvery makes every fifth update of a timed batch a
 	// withdrawal; the others are announcements.
 	benchWithdrawalEvery = 5
@@ -1250,13 +1396,14 @@ const (
 )
 
 // BenchmarkPrefixBatchOnFiveMillionRoutes times the writes the prefix handler
-// makes for one full batch, 25,000 announcements and withdrawals of routes
-// picked at random, to a database that already holds 5 million live routes:
-// 500,000 IPv4 prefixes, each from 10 peers. It builds that database first,
-// one batch at a time, which takes some minutes and a few GB of disk under
-// TMPDIR. It is run by hand, not by make test, for example in the test image:
+// makes for one batch of announcements and withdrawals of routes picked at
+// random, to a database that already holds 5 million live routes: 500,000
+// IPv4 prefixes, each from 10 peers. It times a full batch of 25,000 updates,
+// then the largest batch, 250,000. It builds that database first, one batch at
+// a time, which takes some minutes and a few GB of disk under TMPDIR. It is
+// run by hand, not by make test, for example in the test image:
 //
-//	go test -run '^$' -bench FiveMillion -benchtime 20x ./internal/database/
+//	go test -run '^$' -bench FiveMillion -benchtime 10x ./internal/database/
 func BenchmarkPrefixBatchOnFiveMillionRoutes(b *testing.B) {
 	db, err := New(&config.Config{StateDir: b.TempDir()}, logger.New())
 	if err != nil {
@@ -1280,28 +1427,32 @@ func BenchmarkPrefixBatchOnFiveMillionRoutes(b *testing.B) {
 		}
 	}
 
-	for b.Loop() {
-		ts = ts.Add(time.Second)
+	for _, batchSize := range []int{benchBatchSize, benchMaxBatchSize} {
+		b.Run(strconv.Itoa(batchSize), func(b *testing.B) {
+			for b.Loop() {
+				ts = ts.Add(time.Second)
 
-		var (
-			routes      []*LiveRoute
-			withdrawals []LiveRouteDeletion
-		)
+				var (
+					routes      []*LiveRoute
+					withdrawals []LiveRouteDeletion
+				)
 
-		for i := range benchBatchSize {
-			route := benchRoute(rng, rng.IntN(benchPrefixes),
-				rng.IntN(benchPeers), ts)
-			if i%benchWithdrawalEvery == 0 {
-				withdrawals = append(withdrawals, LiveRouteDeletion{
-					Prefix: route.Prefix, PeerIP: route.PeerIP,
-					IPVersion: ipVersionV4,
-				})
-			} else {
-				routes = append(routes, route)
+				for i := range batchSize {
+					route := benchRoute(rng, rng.IntN(benchPrefixes),
+						rng.IntN(benchPeers), ts)
+					if i%benchWithdrawalEvery == 0 {
+						withdrawals = append(withdrawals, LiveRouteDeletion{
+							Prefix: route.Prefix, PeerIP: route.PeerIP,
+							IPVersion: ipVersionV4,
+						})
+					} else {
+						routes = append(routes, route)
+					}
+				}
+
+				writeBenchBatch(b, db, routes, withdrawals, ts)
 			}
-		}
-
-		writeBenchBatch(b, db, routes, withdrawals, ts)
+		})
 	}
 }