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

README.md

@@ -170,8 +170,8 @@ routewatch/
 ```sql
 -- Core tables
 asns(id, number, handle, description, first_seen, last_seen)
-prefixes_v4(prefix, first_seen)
-prefixes_v6(prefix, first_seen)
+prefixes_v4(id, prefix, mask_length, first_seen, last_seen)
+prefixes_v6(id, prefix, mask_length, first_seen, last_seen)
 
 -- Live routing tables (one per IP version)
 live_routes_v4(id, prefix, mask_length, origin_asn, peer_ip, as_path,

internal/database/database.go

@@ -760,11 +760,40 @@ func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
 	return int(deletedV4 + deletedV6), nil
 }
 
-// insertPrefixesInTable inserts the prefixes, all of one IP version, that the
-// prefix table of that version does not hold yet, each first seen at its time
-// in prefixes. A prefix already stored is left as it is. It returns how many it
-// inserted.
-func insertPrefixesInTable(
+// upsertPrefix updates the last_seen time of prefix with the statements of its
+// IP version's table, inserting the prefix when the update matched no row. It
+// reports whether it inserted it.
+func upsertPrefix(
+	ctx context.Context, updateStmt, insertStmt *sql.Stmt,
+	prefix string, timestamp time.Time,
+) (bool, error) {
+	res, err := updateStmt.ExecContext(ctx, timestamp, prefix)
+	if err != nil {
+		return false, fmt.Errorf("failed to update prefix %s: %w", prefix, err)
+	}
+
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("failed to count updated prefix %s: %w",
+			prefix, err)
+	}
+
+	if affected > 0 {
+		return false, nil
+	}
+
+	_, err = insertStmt.ExecContext(ctx, prefix, timestamp, timestamp)
+	if err != nil {
+		return false, fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
+	}
+
+	return true, nil
+}
+
+// upsertPrefixesInTable updates the last_seen time of prefixes, all of one IP
+// version, in the prefix table of that version, inserting the ones that are
+// new. It returns how many it inserted.
+func upsertPrefixesInTable(
 	ctx context.Context, tx *loggingTx, prefixes map[string]time.Time,
 	ipVersion int,
 ) (int, error) {
@@ -773,15 +802,24 @@ func insertPrefixesInTable(
 	}
 
 	family := ipv4Family
-	insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen) VALUES (?, ?) " +
-		"ON CONFLICT (prefix) DO NOTHING"
+	updateSQL := "UPDATE prefixes_v4 SET last_seen = ? WHERE prefix = ?"
+	insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen, last_seen) " +
+		"VALUES (?, ?, ?)"
 
 	if ipVersion == ipVersionV6 {
 		family = ipv6Family
-		insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen) VALUES (?, ?) " +
-			"ON CONFLICT (prefix) DO NOTHING"
+		updateSQL = "UPDATE prefixes_v6 SET last_seen = ? WHERE prefix = ?"
+		insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen, last_seen) " +
+			"VALUES (?, ?, ?)"
 	}
 
+	updateStmt, err := tx.PrepareContext(ctx, updateSQL)
+	if err != nil {
+		return 0, fmt.Errorf("failed to prepare %s update statement: %w",
+			family, err)
+	}
+	defer func() { _ = updateStmt.Close() }()
+
 	insertStmt, err := tx.PrepareContext(ctx, insertSQL)
 	if err != nil {
 		return 0, fmt.Errorf("failed to prepare %s insert statement: %w",
@@ -793,19 +831,13 @@ func insertPrefixesInTable(
 
 	// The rows are written in key order, as UpsertLiveRouteBatch writes routes.
 	for _, prefix := range slices.Sorted(maps.Keys(prefixes)) {
-		res, err := insertStmt.ExecContext(ctx, prefix, prefixes[prefix])
+		inserted, err := upsertPrefix(ctx, updateStmt, insertStmt, prefix,
+			prefixes[prefix])
 		if err != nil {
-			return 0, fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
+			return 0, err
 		}
 
-		// One row for a new prefix, none for one already stored.
-		inserted, err := res.RowsAffected()
-		if err != nil {
-			return 0, fmt.Errorf("failed to count inserted prefix %s: %w",
-				prefix, err)
-		}
-
-		if inserted > 0 {
+		if inserted {
 			newPrefixes++
 		}
 	}
@@ -813,9 +845,8 @@ func insertPrefixesInTable(
 	return newPrefixes, nil
 }
 
-// UpdatePrefixesBatch stores, in a single transaction, the prefixes that are
-// not stored yet, each first seen at its time in prefixes: the IPv4 prefixes
-// first and then the IPv6 prefixes.
+// UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a
+// single transaction, the IPv4 prefixes first and then the IPv6 prefixes.
 func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	if len(prefixes) == 0 {
 		return nil
@@ -843,12 +874,12 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 
 	ctx := context.Background()
 
-	newV4, err := insertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
+	newV4, err := upsertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
 	if err != nil {
 		return err
 	}
 
-	newV6, err := insertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
+	newV6, err := upsertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
 	if err != nil {
 		return err
 	}
@@ -1081,13 +1112,30 @@ func (d *Database) GetOrCreatePrefix(
 
 	var p Prefix
 
-	query := fmt.Sprintf("SELECT prefix, first_seen FROM %s WHERE prefix = ?",
-		tableName)
+	query := fmt.Sprintf("SELECT prefix, first_seen, last_seen "+
+		"FROM %s WHERE prefix = ?", tableName)
 
-	err = tx.QueryRow(query, prefix).Scan(&p.Prefix, &p.FirstSeen)
+	err = tx.QueryRow(query, prefix).Scan(&p.Prefix, &p.FirstSeen, &p.LastSeen)
 	if err == nil {
-		// Prefix exists; the deferred rollback ends the read.
+		// Prefix exists, update last_seen
 		p.IPVersion = ipVersion
+		updateQuery := fmt.Sprintf("UPDATE %s SET last_seen = ? WHERE prefix = ?",
+			tableName)
+
+		_, err = tx.Exec(updateQuery, timestamp, prefix)
+		if err != nil {
+			return nil, err
+		}
+
+		p.LastSeen = timestamp
+
+		err = tx.Commit()
+		if err != nil {
+			d.logger.Error("Failed to commit transaction for prefix update",
+				"prefix", prefix, "error", err)
+
+			return nil, err
+		}
 
 		return &p, nil
 	}
@@ -1097,11 +1145,12 @@ func (d *Database) GetOrCreatePrefix(
 	}
 
 	// Prefix doesn't exist, create it
-	p = Prefix{Prefix: prefix, IPVersion: ipVersion, FirstSeen: timestamp}
+	p = Prefix{Prefix: prefix, IPVersion: ipVersion,
+		FirstSeen: timestamp, LastSeen: timestamp}
 	insertQuery := fmt.Sprintf("INSERT INTO %s "+
-		"(prefix, first_seen) VALUES (?, ?)", tableName)
+		"(prefix, first_seen, last_seen) VALUES (?, ?, ?)", tableName)
 
-	_, err = tx.Exec(insertQuery, p.Prefix, p.FirstSeen)
+	_, err = tx.Exec(insertQuery, p.Prefix, p.FirstSeen, p.LastSeen)
 	if err != nil {
 		return nil, err
 	}

internal/database/database_internal_test.go

@@ -1218,8 +1218,8 @@ func TestLiveRouteUpdateSetsOriginWhenItChanges(t *testing.T) {
 }
 
 // TestUpdatePrefixesBatchKeepsFirstSeen writes a prefix, then a batch with it
-// and a new prefix. Only the new prefix is inserted and counted, and the known
-// one keeps its first_seen time.
+// and a new prefix. Only the new prefix is counted, and the known one keeps its
+// first_seen time while its last_seen time moves on.
 func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) {
 	t.Parallel()
 
@@ -1257,29 +1257,29 @@ func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) {
 	}
 
 	for _, tt := range []struct {
-		query, prefix string
-		firstSeen     time.Time
+		query, prefix       string
+		firstSeen, lastSeen time.Time
 	}{
 		{
-			query:  "SELECT first_seen FROM prefixes_v4 WHERE prefix = ?",
-			prefix: known, firstSeen: first,
+			query:  "SELECT first_seen, last_seen FROM prefixes_v4 WHERE prefix = ?",
+			prefix: known, firstSeen: first, lastSeen: later,
 		},
 		{
-			query:  "SELECT first_seen FROM prefixes_v6 WHERE prefix = ?",
-			prefix: testV6Prefix, firstSeen: later,
+			query:  "SELECT first_seen, last_seen FROM prefixes_v6 WHERE prefix = ?",
+			prefix: testV6Prefix, firstSeen: later, lastSeen: later,
 		},
 	} {
-		var firstSeen time.Time
+		var firstSeen, lastSeen time.Time
 
 		err := db.db.QueryRowContext(t.Context(), tt.query, tt.prefix).
-			Scan(&firstSeen)
+			Scan(&firstSeen, &lastSeen)
 		if err != nil {
 			t.Fatalf("reading %s: %v", tt.prefix, err)
 		}
 
-		if !firstSeen.Equal(tt.firstSeen) {
-			t.Errorf("%s: first seen %s, want %s", tt.prefix, firstSeen,
-				tt.firstSeen)
+		if !firstSeen.Equal(tt.firstSeen) || !lastSeen.Equal(tt.lastSeen) {
+			t.Errorf("%s: first seen %s, last seen %s; want %s, %s", tt.prefix,
+				firstSeen, lastSeen, tt.firstSeen, tt.lastSeen)
 		}
 	}
 }
@@ -1489,8 +1489,8 @@ func benchRoute(rng *rand.Rand, prefix, peer int, ts time.Time) *LiveRoute {
 }
 
 // writeBenchBatch writes what the prefix handler writes for a batch: the
-// announced routes, the withdrawals, and every prefix the batch names, of
-// which only the ones not stored yet are inserted.
+// announced routes, the withdrawals, and the last-seen time of every prefix
+// the batch names.
 func writeBenchBatch(
 	b *testing.B, db *Database, routes []*LiveRoute,
 	withdrawals []LiveRouteDeletion, ts time.Time,

internal/database/models.go

@@ -35,11 +35,12 @@ type ASN struct {
 }
 
 // Prefix represents an IP prefix (CIDR block) with its IP version (4 or 6)
-// and the time it was first seen.
+// and first/last seen timestamps.
 type Prefix struct {
 	Prefix    string
 	IPVersion int // 4 or 6
 	FirstSeen time.Time
+	LastSeen  time.Time
 }
 
 // Announcement represents a BGP announcement or withdrawal event,

internal/database/schema.sql

@@ -31,13 +31,15 @@ CREATE TABLE IF NOT EXISTS asns (
 -- rows are stored by prefix (WITHOUT ROWID) and need no index of their own.
 CREATE TABLE IF NOT EXISTS prefixes_v4 (
 	prefix TEXT PRIMARY KEY,
-	first_seen DATETIME NOT NULL
+	first_seen DATETIME NOT NULL,
+	last_seen DATETIME NOT NULL
 ) WITHOUT ROWID;
 
 -- IPv6 prefixes table, stored as prefixes_v4 is.
 CREATE TABLE IF NOT EXISTS prefixes_v6 (
 	prefix TEXT PRIMARY KEY,
-	first_seen DATETIME NOT NULL
+	first_seen DATETIME NOT NULL,
+	last_seen DATETIME NOT NULL
 ) WITHOUT ROWID;
 
 CREATE TABLE IF NOT EXISTS announcements (

internal/routewatch/app_integration_internal_test.go

@@ -82,7 +82,9 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	defer m.mu.Unlock()
 
 	for prefix, timestamp := range prefixes {
-		if _, exists := m.Prefixes[prefix]; !exists {
+		if p, exists := m.Prefixes[prefix]; exists {
+			p.LastSeen = timestamp
+		} else {
 			const (
 				ipVersionV4 = 4
 				ipVersionV6 = 6
@@ -97,6 +99,7 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 				Prefix:    prefix,
 				IPVersion: ipVersion,
 				FirstSeen: timestamp,
+				LastSeen:  timestamp,
 			}
 		}
 	}
@@ -112,6 +115,8 @@ func (m *mockStore) GetOrCreatePrefix(
 	defer m.mu.Unlock()
 
 	if p, exists := m.Prefixes[prefix]; exists {
+		p.LastSeen = timestamp
+
 		return p, nil
 	}
 
@@ -129,6 +134,7 @@ func (m *mockStore) GetOrCreatePrefix(
 		Prefix:    prefix,
 		IPVersion: ipVersion,
 		FirstSeen: timestamp,
+		LastSeen:  timestamp,
 	}
 	m.Prefixes[prefix] = p
 	m.PrefixCount++
@@ -364,6 +370,7 @@ func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
 				Prefix:    route.Prefix,
 				IPVersion: route.IPVersion,
 				FirstSeen: route.LastUpdated,
+				LastSeen:  route.LastUpdated,
 			}
 			m.PrefixCount++