Make live route writes cheaper so the prefix handler keeps up (closes #96)

The live route tables lose their random UUID key; the row id the
prefix-length sample draws is declared as the integer key. The prefix
tables are keyed by prefix and keep only first_seen; a batch inserts only
prefixes not stored yet. A route update sets only the columns that
change. Each kept index names the queries using it. Every write goes
through one connection kept for writing, with a 1 GiB page cache; reading
connections get 16 MiB, heap limits rise to 1.5 and 2 GiB (README
"Memory"). Batch writes go in key order. A benchmark run by hand writes
250,000 updates on 5 million routes faster than the feed's 10,000 a
second; a live run follows.

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(id, prefix, mask_length, first_seen, last_seen)
-prefixes_v6(id, prefix, mask_length, first_seen, last_seen)
+prefixes_v4(prefix, first_seen)
+prefixes_v6(prefix, first_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,40 +760,11 @@ func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
 	return int(deletedV4 + deletedV6), nil
 }
 
-// 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(
+// 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(
 	ctx context.Context, tx *loggingTx, prefixes map[string]time.Time,
 	ipVersion int,
 ) (int, error) {
@@ -802,24 +773,15 @@ func upsertPrefixesInTable(
 	}
 
 	family := ipv4Family
-	updateSQL := "UPDATE prefixes_v4 SET last_seen = ? WHERE prefix = ?"
-	insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen, last_seen) " +
-		"VALUES (?, ?, ?)"
+	insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen) VALUES (?, ?) " +
+		"ON CONFLICT (prefix) DO NOTHING"
 
 	if ipVersion == ipVersionV6 {
 		family = ipv6Family
-		updateSQL = "UPDATE prefixes_v6 SET last_seen = ? WHERE prefix = ?"
-		insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen, last_seen) " +
-			"VALUES (?, ?, ?)"
+		insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen) VALUES (?, ?) " +
+			"ON CONFLICT (prefix) DO NOTHING"
 	}
 
-	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",
@@ -831,13 +793,19 @@ func upsertPrefixesInTable(
 
 	// 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,
-			prefixes[prefix])
+		res, err := insertStmt.ExecContext(ctx, prefix, prefixes[prefix])
 		if err != nil {
-			return 0, err
+			return 0, fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
 		}
 
-		if inserted {
+		// 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 {
 			newPrefixes++
 		}
 	}
@@ -845,8 +813,9 @@ func upsertPrefixesInTable(
 	return newPrefixes, nil
 }
 
-// UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a
-// single transaction, the IPv4 prefixes first and then the IPv6 prefixes.
+// 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.
 func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	if len(prefixes) == 0 {
 		return nil
@@ -874,12 +843,12 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 
 	ctx := context.Background()
 
-	newV4, err := upsertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
+	newV4, err := insertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
 	if err != nil {
 		return err
 	}
 
-	newV6, err := upsertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
+	newV6, err := insertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
 	if err != nil {
 		return err
 	}
@@ -1112,30 +1081,13 @@ func (d *Database) GetOrCreatePrefix(
 
 	var p Prefix
 
-	query := fmt.Sprintf("SELECT prefix, first_seen, last_seen "+
-		"FROM %s WHERE prefix = ?", tableName)
+	query := fmt.Sprintf("SELECT prefix, first_seen FROM %s WHERE prefix = ?",
+		tableName)
 
-	err = tx.QueryRow(query, prefix).Scan(&p.Prefix, &p.FirstSeen, &p.LastSeen)
+	err = tx.QueryRow(query, prefix).Scan(&p.Prefix, &p.FirstSeen)
 	if err == nil {
-		// Prefix exists, update last_seen
+		// Prefix exists; the deferred rollback ends the read.
 		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
 	}
@@ -1145,12 +1097,11 @@ func (d *Database) GetOrCreatePrefix(
 	}
 
 	// Prefix doesn't exist, create it
-	p = Prefix{Prefix: prefix, IPVersion: ipVersion,
-		FirstSeen: timestamp, LastSeen: timestamp}
+	p = Prefix{Prefix: prefix, IPVersion: ipVersion, FirstSeen: timestamp}
 	insertQuery := fmt.Sprintf("INSERT INTO %s "+
-		"(prefix, first_seen, last_seen) VALUES (?, ?, ?)", tableName)
+		"(prefix, first_seen) VALUES (?, ?)", tableName)
 
-	_, err = tx.Exec(insertQuery, p.Prefix, p.FirstSeen, p.LastSeen)
+	_, err = tx.Exec(insertQuery, p.Prefix, p.FirstSeen)
 	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 counted, and the known one keeps its
-// first_seen time while its last_seen time moves on.
+// and a new prefix. Only the new prefix is inserted and counted, and the known
+// one keeps its first_seen time.
 func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) {
 	t.Parallel()
 
@@ -1257,29 +1257,29 @@ func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) {
 	}
 
 	for _, tt := range []struct {
-		query, prefix       string
-		firstSeen, lastSeen time.Time
+		query, prefix string
+		firstSeen     time.Time
 	}{
 		{
-			query:  "SELECT first_seen, last_seen FROM prefixes_v4 WHERE prefix = ?",
-			prefix: known, firstSeen: first, lastSeen: later,
+			query:  "SELECT first_seen FROM prefixes_v4 WHERE prefix = ?",
+			prefix: known, firstSeen: first,
 		},
 		{
-			query:  "SELECT first_seen, last_seen FROM prefixes_v6 WHERE prefix = ?",
-			prefix: testV6Prefix, firstSeen: later, lastSeen: later,
+			query:  "SELECT first_seen FROM prefixes_v6 WHERE prefix = ?",
+			prefix: testV6Prefix, firstSeen: later,
 		},
 	} {
-		var firstSeen, lastSeen time.Time
+		var firstSeen time.Time
 
 		err := db.db.QueryRowContext(t.Context(), tt.query, tt.prefix).
-			Scan(&firstSeen, &lastSeen)
+			Scan(&firstSeen)
 		if err != nil {
 			t.Fatalf("reading %s: %v", tt.prefix, err)
 		}
 
-		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)
+		if !firstSeen.Equal(tt.firstSeen) {
+			t.Errorf("%s: first seen %s, want %s", tt.prefix, firstSeen,
+				tt.firstSeen)
 		}
 	}
 }
@@ -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 the last-seen time of every prefix
-// the batch names.
+// announced routes, the withdrawals, and every prefix the batch names, of
+// which only the ones not stored yet are inserted.
 func writeBenchBatch(
 	b *testing.B, db *Database, routes []*LiveRoute,
 	withdrawals []LiveRouteDeletion, ts time.Time,

internal/database/models.go

@@ -35,12 +35,11 @@ type ASN struct {
 }
 
 // Prefix represents an IP prefix (CIDR block) with its IP version (4 or 6)
-// and first/last seen timestamps.
+// and the time it was first seen.
 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,15 +31,13 @@ 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,
-	last_seen DATETIME NOT NULL
+	first_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,
-	last_seen DATETIME NOT NULL
+	first_seen DATETIME NOT NULL
 ) WITHOUT ROWID;
 
 CREATE TABLE IF NOT EXISTS announcements (

internal/routewatch/app_integration_internal_test.go

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