Store the routing table in the database (part of #122)

A new routes table holds one row per prefix and origin AS: the peer set
as the bitset's bytes, and the two kept routes, each with its peer's
number, AS path and time in seconds. A new route_peers table holds each
peer's number and address.

WriteRouteChanges writes one TakeChanges result in a transaction.
ReadRouteTable reads both tables into a new routing table, through the
new Table.Load, which records no change. It fails on a row that names a
peer number missing from route_peers. Nothing uses them yet.

Model: opus-5-5

internal/database/routes.go

@@ -18,9 +18,9 @@ import (
 // columns for: the newest and the older one.
 const keptRoutesPerRow = 2
 
-// errUnknownRoutePeer is returned for a kept route whose peer has no number in
+// errUnknownRoutePeer is returned for a peer that route_peers lacks: on a
-// route_peers.
+// write, the peer of a kept route; on a read, a peer number in a row.
-var errUnknownRoutePeer = errors.New("kept route of a peer without a number")
+var errUnknownRoutePeer = errors.New("peer not in route_peers")
 
 // The statements of the routes and route_peers tables (schema.sql).
 const (
@@ -190,6 +190,7 @@ func routeRow(
 // tables hold, as startup reads it. It reads the peers first, so that the
 // table keeps their numbers, then loads each row into the table as it reads
 // it. The two reads are separate, so nothing may write the tables meanwhile.
+// It fails on a row that names a peer number route_peers lacks.
 func (d *Database) ReadRouteTable(ctx context.Context) (*routetable.Table, error) {
 	peerNumbers, err := d.readRoutePeers(ctx)
 	if err != nil {
@@ -197,8 +198,11 @@ func (d *Database) ReadRouteTable(ctx context.Context) (*routetable.Table, error
 	}
 
 	addresses := make(map[uint]string, len(peerNumbers))
+	numbers := new(bitset.BitSet)
+
 	for address, number := range peerNumbers {
 		addresses[number] = address
+		numbers.Set(number)
 	}
 
 	table := routetable.New(routetable.Params{PeerNumbers: peerNumbers})
@@ -210,7 +214,7 @@ func (d *Database) ReadRouteTable(ctx context.Context) (*routetable.Table, error
 	defer func() { _ = rows.Close() }()
 
 	for rows.Next() {
-		change, err := scanRouteRow(rows, addresses)
+		change, err := scanRouteRow(rows, addresses, numbers)
 		if err != nil {
 			return nil, err
 		}
@@ -266,9 +270,11 @@ type keptRouteColumns struct {
 }
 
 // scanRouteRow returns the content of the routes row rows is at, with each kept
-// route's peer named by its address in addresses.
+// route's peer named by its address in addresses. It fails when the row names a
+// peer number that addresses lacks. numbers holds the numbers of addresses as a
+// set, against which a row's whole peer set is checked at once.
 func scanRouteRow(
-	rows *sql.Rows, addresses map[uint]string,
+	rows *sql.Rows, addresses map[uint]string, numbers *bitset.BitSet,
 ) (routetable.Change, error) {
 	var (
 		change routetable.Change
@@ -297,6 +303,13 @@ func scanRouteRow(
 			prefix, err)
 	}
 
+	if !numbers.IsSuperSet(change.Peers) {
+		unknown, _ := change.Peers.Difference(numbers).NextSet(0)
+
+		return change, fmt.Errorf("%w: number %d in the peers of route %s",
+			errUnknownRoutePeer, unknown, prefix)
+	}
+
 	for _, columns := range kept {
 		if !columns.peer.Valid {
 			continue

internal/database/routes_internal_test.go

@@ -113,10 +113,62 @@ func TestReadRouteTableGivesBackWhatWasWritten(t *testing.T) {
 	checkReadTable(t, db, source)
 }
 
+func TestReadRouteTableGivesATableThatTakesLaterUpdates(t *testing.T) {
+	t.Parallel()
+
+	db := newRoutesTestDatabase(t)
+	source := routetable.New(routetable.Params{})
+
+	// peerA and peerB carry tablePrefixV4, peerB and peerC tablePrefixV6.
+	announceTo(source, tablePrefixV4, asA, peerA, 10)
+	announceTo(source, tablePrefixV4, asA, peerB, 20)
+	announceTo(source, tablePrefixV6, asA, peerB, 30)
+	announceTo(source, tablePrefixV6, asA, peerC, 40)
+
+	writeTableChanges(t, db, source)
+	read := checkReadTable(t, db, source)
+
+	// The same updates to both tables: a loaded peer withdraws a prefix,
+	// another one goes down, and a new peer comes.
+	for _, table := range []*routetable.Table{source, read} {
+		table.Withdraw(netip.MustParsePrefix(tablePrefixV4), peerA)
+		table.PeerDown(peerB)
+		announceTo(table, tablePrefixShortV4, asB, peerD, 50)
+	}
+
+	checkSameReads(t, read, source)
+}
+
+func TestReadRouteTableFailsOnAPeerWithoutANumber(t *testing.T) {
+	t.Parallel()
+
+	db := newRoutesTestDatabase(t)
+	table := routetable.New(routetable.Params{})
+
+	// Only the routes of peerB and peerC are kept, so only the peer set names
+	// peerA.
+	announceTo(table, tablePrefixV4, asA, peerA, 10)
+	announceTo(table, tablePrefixV4, asA, peerB, 20)
+	announceTo(table, tablePrefixV4, asA, peerC, 30)
+	writeTableChanges(t, db, table)
+
+	_, err := db.db.ExecContext(t.Context(),
+		"DELETE FROM route_peers WHERE address = ?", peerA)
+	if err != nil {
+		t.Fatalf("deleting a route peer: %v", err)
+	}
+
+	_, err = db.ReadRouteTable(t.Context())
+	if !errors.Is(err, errUnknownRoutePeer) {
+		t.Fatalf("ReadRouteTable = %v, want %v", err, errUnknownRoutePeer)
+	}
+}
+
 // checkReadTable checks that the table ReadRouteTable reads answers every read
-// as want does, for the prefixes and the origin ASes the tests use, and that
+// as want does, and that reading it recorded no change. It returns that table.
-// reading it recorded no change.
+func checkReadTable(
-func checkReadTable(t *testing.T, db *Database, want *routetable.Table) {
+	t *testing.T, db *Database, want *routetable.Table,
+) *routetable.Table {
 	t.Helper()
 
 	got, err := db.ReadRouteTable(t.Context())
@@ -124,6 +176,22 @@ func checkReadTable(t *testing.T, db *Database, want *routetable.Table) {
 		t.Fatalf("ReadRouteTable: %v", err)
 	}
 
+	checkSameReads(t, got, want)
+
+	changes, newPeers := got.TakeChanges()
+	if len(changes) != 0 || len(newPeers) != 0 {
+		t.Errorf("the read recorded changes %+v and new peers %v, want none",
+			changes, newPeers)
+	}
+
+	return got
+}
+
+// checkSameReads checks that got answers every read as want does, for the
+// prefixes and the origin ASes the tests use.
+func checkSameReads(t *testing.T, got, want *routetable.Table) {
+	t.Helper()
+
 	if !reflect.DeepEqual(got.Counts(), want.Counts()) {
 		t.Errorf("Counts() = %+v, want %+v", got.Counts(), want.Counts())
 	}
@@ -167,12 +235,6 @@ func checkReadTable(t *testing.T, db *Database, want *routetable.Table) {
 				got.OriginatedPrefixes(asn), want.OriginatedPrefixes(asn))
 		}
 	}
-
-	changes, newPeers := got.TakeChanges()
-	if len(changes) != 0 || len(newPeers) != 0 {
-		t.Errorf("the read recorded changes %+v and new peers %v, want none",
-			changes, newPeers)
-	}
 }
 
 // sortedSample returns every prefix of the IP version and mask length in

internal/routetable/routetable.go

@@ -224,9 +224,8 @@ func (t *Table) PeerDown(peer string) {
 // Unlike an update, Load records no change, so TakeChanges does not hand the
 // pair back to be written again. Loading a pair twice changes nothing.
 //
-// The peer set of change holds the peers' numbers from Params.PeerNumbers. A
+// The caller makes sure that every peer change names, in its peer set or in
-// kept route whose peer has no number there is left out, because numbering
+// its routes, has a number in Params.PeerNumbers.
-// the peer would record a new peer.
 func (t *Table) Load(change Change) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
@@ -253,10 +252,7 @@ func (t *Table) Load(change Change) {
 	// Oldest first, as they were announced, so that two routes with the same
 	// time are kept in the order they had.
 	for _, r := range slices.Backward(change.Routes) {
-		peer, known := t.peerNumbers[r.Peer]
+		peer := t.peerNumbers[r.Peer]
-		if !known {
-			continue
-		}
 
 		t.dropKeptRoute(o, peer)
 		t.keepIfNewest(o, route{

internal/routetable/routetable_test.go

@@ -522,31 +522,6 @@ func TestLoad(t *testing.T) {
 	checkChanges(t, loaded, []change{}, map[string]uint{})
 }
 
-func TestLoadLeavesOutTheRouteOfAPeerWithoutANumber(t *testing.T) {
-	t.Parallel()
-
-	prefix := netip.MustParsePrefix(prefixV4)
-	source := routetable.New(routetable.Params{})
-	announceAll(source, prefixV4, asA, 10, peerA, peerB)
-
-	// peerB has no number in the loaded table.
-	changes, _ := source.TakeChanges()
-	loaded := routetable.New(routetable.Params{
-		PeerNumbers: map[string]uint{peerA: 0},
-	})
-	loaded.Load(changes[0])
-
-	want := []routetable.Route{routeOf(peerA, asA, 10)}
-
-	origins := loaded.Origins(prefix)
-	if len(origins) != 1 || !reflect.DeepEqual(origins[0].Routes, want) {
-		t.Errorf("Origins(%s) = %+v, want one origin with routes %+v",
-			prefix, origins, want)
-	}
-
-	checkChanges(t, loaded, []change{}, map[string]uint{})
-}
-
 // TestConcurrentUse has goroutines update one table while others read it. The
 // race detector, which make test runs, fails it when a method uses the table
 // without its lock.