Write the routing table every second and read it at startup (part of #122)

The prefix handler applies each update to the routing table in memory as
it handles it, sorted into announcements and withdrawals as for the live
route tables. Each flush first writes at most 20,000 of the table's
changed pairs, with every new peer; the rest follow in the next flushes.
A failed write puts its changes and new peers back in the table. Stop
makes one last flush, and logs the changed pairs left.

Startup reads the table before the streamer connects, and logs its pairs,
peers and time. The per-peer counts are a slice, so the read takes about
a third less time. The health check's start period is 60 seconds, so a
full table's read fits before upaas looks.

Model: opus-5-5

internal/routetable/benchmark_test.go

@@ -2,6 +2,7 @@ package routetable_test
 
 import (
 	"fmt"
+	"math"
 	"math/rand/v2"
 	"net/netip"
 	"runtime"
@@ -77,7 +78,7 @@ func BenchmarkApplyMixedBatch(b *testing.B) {
 
 	// The record of changes is handed over every second while routewatch
 	// runs, so it is not part of the table's size.
-	table.TakeChanges()
+	table.TakeChanges(math.MaxInt)
 
 	bytesPerPair := float64(heapBytes()-before) / benchPrefixes
 

internal/routetable/reads.go

@@ -31,6 +31,11 @@ type OriginatedPrefix struct {
 
 // Counts are the table's figures for the stats.
 type Counts struct {
+	// Pairs counts the prefix and origin AS pairs that some peer carries.
+	Pairs int
+	// Peers counts the peers that have a number: those given in
+	// Params.PeerNumbers and those numbered since.
+	Peers        int
 	IPv4Prefixes int
 	IPv6Prefixes int
 	// IPv4PrefixesByLength[n] and IPv6PrefixesByLength[n] are how many
@@ -160,6 +165,7 @@ func (t *Table) Counts() Counts {
 	defer t.mu.RUnlock()
 
 	counts := Counts{
+		Peers:                   len(t.peerNumbers),
 		IPv4PrefixesByLength:    t.ipv4.prefixesByLength(),
 		IPv6PrefixesByLength:    t.ipv6.prefixesByLength(),
 		IPv4Routes:              t.ipv4.routes,
@@ -175,6 +181,10 @@ func (t *Table) Counts() Counts {
 		counts.IPv6Prefixes += n
 	}
 
+	for _, prefixes := range t.byOrigin {
+		counts.Pairs += len(prefixes)
+	}
+
 	oldest, found := t.keptTimes.Min()
 	if found {
 		counts.OldestKeptRoute = &oldest.time

internal/routetable/reads_test.go

@@ -194,6 +194,8 @@ func TestCounts(t *testing.T) {
 	announceAll(table, "2001:db8::/32", asA, 5, peerA)
 	announceAll(table, "2001:db8::/32", asA, 25, peerC)
 	checkCounts(t, table, routetable.Counts{
+		Pairs:                3,
+		Peers:                3,
 		IPv4Prefixes:         2,
 		IPv6Prefixes:         1,
 		IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}),
@@ -209,6 +211,8 @@ func TestCounts(t *testing.T) {
 	announceAll(table, prefixV4, asA, 30, peerC)
 	table.Withdraw(netip.MustParsePrefix("2001:db8::/32"), peerA)
 	checkCounts(t, table, routetable.Counts{
+		Pairs:                3,
+		Peers:                3,
 		IPv4Prefixes:         2,
 		IPv6Prefixes:         1,
 		IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}),
@@ -219,9 +223,11 @@ func TestCounts(t *testing.T) {
 		NewestKeptRoute:      timeAt(30),
 	})
 
-	// peerC was the last peer of the IPv6 prefix.
+	// peerC was the last peer of the IPv6 prefix. It keeps its number.
 	table.PeerDown(peerC)
 	checkCounts(t, table, routetable.Counts{
+		Pairs:                2,
+		Peers:                3,
 		IPv4Prefixes:         2,
 		IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}),
 		IPv6PrefixesByLength: byLength(ipv6Bits, nil),

internal/routetable/routetable.go

@@ -7,6 +7,7 @@ package routetable
 
 import (
 	"cmp"
+	"maps"
 	"net/netip"
 	"slices"
 	"sync"
@@ -102,11 +103,15 @@ type Table struct {
 	peerNumbers    map[string]uint
 	peerAddresses  map[uint]string
 	nextPeerNumber uint
-	// peerPrefixes counts the prefixes each peer carries, by its number.
-	peerPrefixes map[uint]int
+	// peerPrefixes[n] counts the prefixes the peer numbered n carries, for
+	// each n below nextPeerNumber. It is a slice, not a map, because Load
+	// counts once for every peer of every pair it loads: with a map, reading
+	// a full table at startup took several times as long.
+	peerPrefixes []int
 
-	// newPeers and changed are what TakeChanges hands over next: the peers
-	// numbered, and the prefix and origin AS pairs changed, since it last ran.
+	// newPeers and changed are what TakeChanges hands over: the peers
+	// numbered since it last ran, and the prefix and origin AS pairs changed
+	// since it last took them.
 	newPeers map[string]uint
 	changed  map[pair]struct{}
 
@@ -127,7 +132,6 @@ func New(params Params) *Table {
 		byOrigin:      make(map[int]map[netip.Prefix]*origin),
 		peerNumbers:   make(map[string]uint, len(params.PeerNumbers)),
 		peerAddresses: make(map[uint]string, len(params.PeerNumbers)),
-		peerPrefixes:  make(map[uint]int),
 		newPeers:      make(map[string]uint),
 		changed:       make(map[pair]struct{}),
 		keptTimes: btree.NewG(keptTimesDegree, func(a, b keptTime) bool {
@@ -141,6 +145,8 @@ func New(params Params) *Table {
 		t.nextPeerNumber = max(t.nextPeerNumber, number+1)
 	}
 
+	t.peerPrefixes = make([]int, t.nextPeerNumber)
+
 	return t
 }
 
@@ -240,13 +246,14 @@ func (t *Table) Load(change Change) {
 	o := t.findOrAddOrigin(prefix, change.OriginASN)
 	wasWithoutKeptRoute := o.withoutKeptRoute()
 
-	for peer := range change.Peers.EachSet() {
-		if !o.peers.Test(peer) {
-			o.peers.Set(peer)
+	// The peers not yet in the peer set join it in one union, which takes
+	// less time than setting their bits one at a time.
+	joining := change.Peers.Difference(&o.peers)
+	o.peers.InPlaceUnion(joining)
+	t.familyOf(prefix.Addr()).routes += int(joining.Count())
 
-			t.familyOf(prefix.Addr()).routes++
-			t.peerPrefixes[peer]++
-		}
+	for peer := range joining.EachSet() {
+		t.peerPrefixes[peer]++
 	}
 
 	// Oldest first, as they were announced, so that two routes with the same
@@ -265,16 +272,29 @@ func (t *Table) Load(change Change) {
 	t.countOriginWithoutKeptRoute(o, wasWithoutKeptRoute)
 }
 
-// TakeChanges returns the prefix and origin AS pairs that changed since it last
-// ran, in prefix order, each with its current content. It also returns the
-// peers numbered since then, by address. It then clears both records.
-func (t *Table) TakeChanges() ([]Change, map[string]uint) {
+// TakeChanges returns at most maxChanges of the prefix and origin AS pairs that
+// changed since they were last taken, in prefix order, each with its current
+// content. The other changed pairs wait for a later call. It also returns every
+// peer numbered since it last ran, by address, so a pair taken later names only
+// peers already returned. It then forgets what it returned.
+func (t *Table) TakeChanges(maxChanges int) ([]Change, map[string]uint) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
-	changes := make([]Change, 0, len(t.changed))
+	changes := make([]Change, 0, min(maxChanges, len(t.changed)))
 	for key := range t.changed {
+		if len(changes) == maxChanges {
+			break
+		}
+
 		changes = append(changes, t.currentContent(key))
+		delete(t.changed, key)
+	}
+
+	// A map keeps the memory of its largest size, so an emptied one is
+	// replaced.
+	if len(t.changed) == 0 {
+		t.changed = make(map[pair]struct{})
 	}
 
 	slices.SortFunc(changes, func(a, b Change) int {
@@ -284,11 +304,35 @@ func (t *Table) TakeChanges() ([]Change, map[string]uint) {
 
 	newPeers := t.newPeers
 	t.newPeers = make(map[string]uint)
-	t.changed = make(map[pair]struct{})
 
 	return changes, newPeers
 }
 
+// PutBackChanges records again what one TakeChanges call returned, for a
+// caller that could not write it. The next TakeChanges returns those peers
+// again, and those pairs wait to be taken again like any changed pair. Each is
+// then taken with its content at that time, so a later change of a pair
+// replaces the earlier.
+func (t *Table) PutBackChanges(changes []Change, newPeers map[string]uint) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	for _, change := range changes {
+		t.changed[pair{prefix: change.Prefix, asn: change.OriginASN}] = struct{}{}
+	}
+
+	maps.Copy(t.newPeers, newPeers)
+}
+
+// ChangeCount returns how many changed prefix and origin AS pairs wait for
+// TakeChanges.
+func (t *Table) ChangeCount() int {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	return len(t.changed)
+}
+
 // numberPeer returns the peer's number, and gives it the next one the first
 // time the table sees it.
 func (t *Table) numberPeer(address string) uint {
@@ -301,6 +345,7 @@ func (t *Table) numberPeer(address string) uint {
 	t.nextPeerNumber++
 	t.peerNumbers[address] = number
 	t.peerAddresses[number] = address
+	t.peerPrefixes = append(t.peerPrefixes, 0)
 	t.newPeers[address] = number
 
 	return number

internal/routetable/routetable_test.go

@@ -1,6 +1,7 @@
 package routetable_test
 
 import (
+	"math"
 	"net/netip"
 	"reflect"
 	"slices"
@@ -438,7 +439,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	// peerB to v4 are kept; peerA keeps only its bit.
 	announceAll(table, prefixV4, asA, 10, peerA, peerB, peerC)
 	table.Announce(v6, asA, routeOf(peerB, asA, 10))
-	table.TakeChanges()
+	table.TakeChanges(math.MaxInt)
 
 	// peerA announces again, with a time older than both kept routes.
 	table.Announce(v4, asA, routeOf(peerA, asA, 15))
@@ -476,6 +477,80 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	checkChanges(t, table, []change{}, map[string]uint{})
 }
 
+// TestTakeChangesTakesAtMostMaxChanges changes three prefix and origin AS
+// pairs, then takes the changes two at a time. The first call must return two
+// pairs and both new peers, and the second the third pair and no peer.
+func TestTakeChangesTakesAtMostMaxChanges(t *testing.T) {
+	t.Parallel()
+
+	const maxChanges = 2
+
+	table := routetable.New(routetable.Params{})
+	v4 := netip.MustParsePrefix(prefixV4)
+	v6 := netip.MustParsePrefix(prefixV6)
+
+	table.Announce(v4, asA, routeOf(peerA, asA, 10))
+	table.Announce(v4, asB, routeOf(peerB, asB, 10))
+	table.Announce(v6, asA, routeOf(peerA, asA, 10))
+
+	first, firstPeers := table.TakeChanges(maxChanges)
+	second, secondPeers := table.TakeChanges(maxChanges)
+
+	if len(first) != maxChanges || len(second) != 1 {
+		t.Fatalf("TakeChanges(%d) returned %d and then %d pairs, want %d and 1",
+			maxChanges, len(first), len(second), maxChanges)
+	}
+
+	wantPeers := map[string]uint{peerA: 0, peerB: 1}
+	if !reflect.DeepEqual(firstPeers, wantPeers) || len(secondPeers) != 0 {
+		t.Errorf("TakeChanges(%d) returned the new peers %v and then %v, "+
+			"want %v and none", maxChanges, firstPeers, secondPeers, wantPeers)
+	}
+
+	for _, c := range first {
+		if c.Prefix == second[0].Prefix && c.OriginASN == second[0].OriginASN {
+			t.Errorf("both calls returned %s with AS%d", c.Prefix, c.OriginASN)
+		}
+	}
+}
+
+// TestPutBackChanges takes a table's changes and puts them back, as a caller
+// does whose write failed. The next TakeChanges must return them again, with
+// each pair's content then, and with the changes since.
+func TestPutBackChanges(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{})
+	v4 := netip.MustParsePrefix(prefixV4)
+	v6 := netip.MustParsePrefix(prefixV6)
+
+	table.Announce(v4, asA, routeOf(peerA, asA, 10))
+	table.Announce(v6, asA, routeOf(peerA, asA, 10))
+	table.PutBackChanges(table.TakeChanges(math.MaxInt))
+
+	if got := table.ChangeCount(); got != 2 {
+		t.Errorf("ChangeCount() = %d after the changes were put back, want 2",
+			got)
+	}
+
+	// peerA withdraws v4, so v4's earlier content must not come back; peerB
+	// announces v6, and is a new peer as well.
+	table.Withdraw(v4, peerA)
+	table.Announce(v6, asA, routeOf(peerB, asA, 20))
+
+	checkChanges(t, table, []change{
+		{Prefix: prefixV4, OriginASN: asA, Removed: true},
+		{Prefix: prefixV6, OriginASN: asA, Peers: []uint{0, 1},
+			Routes: []routetable.Route{
+				routeOf(peerB, asA, 20), routeOf(peerA, asA, 10),
+			}},
+	}, map[string]uint{peerA: 0, peerB: 1})
+
+	if got := table.ChangeCount(); got != 0 {
+		t.Errorf("ChangeCount() = %d after TakeChanges, want 0", got)
+	}
+}
+
 func TestLoad(t *testing.T) {
 	t.Parallel()
 
@@ -497,7 +572,7 @@ func TestLoad(t *testing.T) {
 	source.Withdraw(other, peerB)
 	source.Withdraw(other, peerC)
 
-	changes, peers := source.TakeChanges()
+	changes, peers := source.TakeChanges(math.MaxInt)
 	loaded := routetable.New(routetable.Params{PeerNumbers: peers})
 
 	// Loading a pair twice changes nothing.
@@ -543,7 +618,8 @@ func TestConcurrentUse(t *testing.T) {
 		func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) },
 		func(int) { table.Withdraw(prefix, peerA) },
 		func(int) { table.PeerDown(peerB) },
-		func(int) { table.TakeChanges() },
+		func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) },
+		func(int) { table.ChangeCount() },
 		func(int) { table.Lookup(prefix.Addr()) },
 		func(int) { table.Origins(prefix) },
 		func(int) { table.OriginatedPrefixes(asA) },
@@ -563,14 +639,15 @@ func TestConcurrentUse(t *testing.T) {
 	wg.Wait()
 }
 
-// checkChanges checks that TakeChanges returns want and wantPeers.
+// checkChanges checks that TakeChanges, taking every change, returns want and
+// wantPeers.
 func checkChanges(
 	t *testing.T, table *routetable.Table, want []change,
 	wantPeers map[string]uint,
 ) {
 	t.Helper()
 
-	changes, newPeers := table.TakeChanges()
+	changes, newPeers := table.TakeChanges(math.MaxInt)
 
 	got := make([]change, 0, len(changes))
 	for _, c := range changes {