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

README.md

@@ -315,7 +315,8 @@ routewatch/
       on two collectors, a session going down on one also removes the routes the
       other still carries, until they are announced again. It also applies each
       update to the routing table in memory, and writes the table's changes to
-      the database every second.
+      the database every second. A write takes at most 20,000 changed prefix and
+      origin AS pairs, and the rest follow in the next writes.
     - **PeeringHandler**: Extracts AS peering relationships from AS paths
 
 5. **HTTP Request Flow**

TODO.md

@@ -34,8 +34,9 @@ tested.
 # Completed Steps
 
 - 2026-10-27: the prefix handler applies each update to the routing table in
-  memory, and writes the table's changes to the database every second. A failed
+  memory, and writes the table's changes to the database every second. A write
-  write is written again with the next one. Startup reads the table before the
+  takes at most 20,000 changed pairs, and the rest follow in the next writes. A
+  failed write is written again later. Startup reads the table before the
   streamer connects. The pages still read the per-peer route tables (part of
   #122)
 - 2026-10-27: the database keeps a copy of the whole routing table in a new

internal/database/routes_internal_test.go

@@ -3,6 +3,7 @@ package database
 import (
 	"errors"
 	"fmt"
+	"math"
 	"math/rand/v2"
 	"net/netip"
 	"os"
@@ -71,7 +72,7 @@ func newRoutesTestDatabase(t *testing.T) *Database {
 func writeTableChanges(t *testing.T, db *Database, table *routetable.Table) {
 	t.Helper()
 
-	changes, newPeers := table.TakeChanges()
+	changes, newPeers := table.TakeChanges(math.MaxInt)
 
 	err := db.WriteRouteChanges(changes, newPeers)
 	if err != nil {
@@ -178,7 +179,7 @@ func checkReadTable(
 
 	checkSameReads(t, got, want)
 
-	changes, newPeers := got.TakeChanges()
+	changes, newPeers := got.TakeChanges(math.MaxInt)
 	if len(changes) != 0 || len(newPeers) != 0 {
 		t.Errorf("the read recorded changes %+v and new peers %v, want none",
 			changes, newPeers)
@@ -369,7 +370,7 @@ func TestWriteRouteChangesWritesNothingWhenItFails(t *testing.T) {
 	announceTo(table, tablePrefixV6, asA, peerB, 20)
 
 	// peerB's number is missing, so its kept route cannot be written.
-	changes, newPeers := table.TakeChanges()
+	changes, newPeers := table.TakeChanges(math.MaxInt)
 	delete(newPeers, peerB)
 
 	err := db.WriteRouteChanges(changes, newPeers)

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/routetable.go

@@ -109,8 +109,9 @@ type Table struct {
 	// a full table at startup took several times as long.
 	peerPrefixes []int
 
-	// newPeers and changed are what TakeChanges hands over next: the peers
+	// newPeers and changed are what TakeChanges hands over: the peers
-	// numbered, and the prefix and origin AS pairs changed, since it last ran.
+	// 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{}
 
@@ -271,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
+// TakeChanges returns at most maxChanges of the prefix and origin AS pairs that
-// ran, in prefix order, each with its current content. It also returns the
+// changed since they were last taken, in prefix order, each with its current
-// peers numbered since then, by address. It then clears both records.
+// content. The other changed pairs wait for a later call. It also returns every
-func (t *Table) TakeChanges() ([]Change, map[string]uint) {
+// 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 {
@@ -290,15 +304,15 @@ 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 then returns those pairs
+// caller that could not write it. The next TakeChanges returns those peers
-// and peers again, with the pairs that changed since. It returns each pair with
+// again, and those pairs wait to be taken again like any changed pair. Each is
-// its content at that time, so a later change of a pair replaces the earlier.
+// 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()
@@ -310,8 +324,8 @@ func (t *Table) PutBackChanges(changes []Change, newPeers map[string]uint) {
 	maps.Copy(t.newPeers, newPeers)
 }
 
-// ChangeCount returns how many prefix and origin AS pairs changed since
+// ChangeCount returns how many changed prefix and origin AS pairs wait for
-// TakeChanges last ran: how many it would return now.
+// TakeChanges.
 func (t *Table) ChangeCount() int {
 	t.mu.RLock()
 	defer t.mu.RUnlock()

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,43 @@ 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.
@@ -488,7 +526,7 @@ func TestPutBackChanges(t *testing.T) {
 
 	table.Announce(v4, asA, routeOf(peerA, asA, 10))
 	table.Announce(v6, asA, routeOf(peerA, asA, 10))
-	table.PutBackChanges(table.TakeChanges())
+	table.PutBackChanges(table.TakeChanges(math.MaxInt))
 
 	if got := table.ChangeCount(); got != 2 {
 		t.Errorf("ChangeCount() = %d after the changes were put back, want 2",
@@ -534,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.
@@ -580,7 +618,7 @@ 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.PutBackChanges(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) },
@@ -601,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 {

internal/routewatch/prefixhandler.go

@@ -39,6 +39,16 @@ const (
 	// DO NOT reduce this timeout - larger batches are more efficient
 	prefixBatchTimeout = 1 * time.Second
 
+	// maxTableChangesPerWrite is the most changed prefix and origin AS pairs
+	// of the routing table that one flush writes; the rest wait for the next
+	// flushes. The change of one pair takes about 430 bytes of memory and 15
+	// microseconds to write. So a write holds about 9 MB and takes about a
+	// third of the one-second flush interval, which leaves the rest to the
+	// batch. The feed peaks at about 5,000 updates a second, each changing at
+	// most one pair. So the million pairs that a full-table peer's session
+	// going down changes are written in about a minute.
+	maxTableChangesPerWrite = 20000
+
 	// IP version constants
 	ipv4Version = 4
 	ipv6Version = 6
@@ -242,14 +252,15 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop stops the flush goroutine, which writes the routing table's changes and
+// Stop stops the flush goroutine, which makes one last flush before it returns,
-// the updates and peer downs the handler holds before it returns, and waits for
+// and waits for it until ctx ends. That flush writes the updates and peer downs
-// it until ctx ends. Once ctx has ended the flush goroutine skips that last
+// the handler holds, and at most maxTableChangesPerWrite of the routing table's
-// write, and a write still running is lost when the process exits. A peer whose
+// changes. Once ctx has ended the flush goroutine skips that last flush, and a
-// route delete fails in the last write is not retried either, nor is a failed
+// write still running is lost when the process exits. A peer whose route delete
-// write of the table's changes. Stop logs how many updates are not written, how
+// fails in the last flush is not retried either, nor is a failed write of the
-// many peer downs have not had their routes deleted, and how many of the table's
+// table's changes. Stop logs how many updates are not written, how many peer
-// changes are not written.
+// downs have not had their routes deleted, and how many of the table's changes
+// are not written.
 func (h *PrefixHandler) Stop(ctx context.Context) {
 	h.stopCh <- ctx
 
@@ -276,8 +287,8 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 		return
 	}
 
-	h.logger.Error("The prefix handler stopped without retrying writes that "+
+	h.logger.Error("The prefix handler stopped before it wrote everything",
-		"failed", "updates", unwritten, "peers_down", peersDown,
+		"updates", unwritten, "peers_down", peersDown,
 		"table_changes", tableChanges)
 }
 
@@ -474,12 +485,14 @@ func (h *PrefixHandler) applyToTable(update prefixUpdate) {
 	})
 }
 
-// writeTableChanges writes the changes of the routing table in memory since
+// writeTableChanges writes at most maxTableChangesPerWrite changes of the
-// the last write. When the write fails, they go back to the table, so that the
+// routing table in memory, and every peer the table numbered since the last
-// next write writes them again, each pair with its content then.
+// write.
+// When the write fails, they go back to the table, so that a later write writes
+// them again, each pair with its content then.
 func (h *PrefixHandler) writeTableChanges() {
 	h.mu.Lock()
-	changes, newPeers := h.table.TakeChanges()
+	changes, newPeers := h.table.TakeChanges(maxTableChangesPerWrite)
 	h.writingTableChanges = len(changes)
 	h.mu.Unlock()
 

internal/routewatch/prefixhandler_table_internal_test.go

@@ -26,6 +26,9 @@ const (
 
 	otherV4Prefix = "198.51.100.0/24"
 	setV4Prefix   = "203.0.113.0/24"
+
+	// firstManyPrefix is the first prefix announceManyPrefixes announces.
+	firstManyPrefix = "10.0.0.0/32"
 )
 
 // tableWriteTimeout is how long a test waits for the flush timer to write the
@@ -279,6 +282,95 @@ func TestPrefixHandlerWritesAFailedTableWriteAgain(t *testing.T) {
 	}
 }
 
+// tableWriteSizesStore is a real database that records how many changes and
+// how many new peers each write of the routing table's changes holds.
+type tableWriteSizesStore struct {
+	*database.Database
+
+	changes  []int
+	newPeers []int
+}
+
+func (s *tableWriteSizesStore) WriteRouteChanges(
+	changes []routetable.Change, newPeers map[string]uint,
+) error {
+	s.changes = append(s.changes, len(changes))
+	s.newPeers = append(s.newPeers, len(newPeers))
+
+	return s.Database.WriteRouteChanges(changes, newPeers)
+}
+
+// TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes has one new peer
+// change one and a half times as many prefix and origin AS pairs as one write
+// takes, then flushes three times. The first write must hold as many pairs as
+// one write takes, and the peer. The second must hold the other pairs, and the
+// third flush must write nothing. The database must then hold the table in
+// memory.
+func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
+	t.Parallel()
+
+	const (
+		pairs   = maxTableChangesPerWrite * 3 / 2
+		flushes = 3
+	)
+
+	db := newTestDatabase(t)
+	store := &tableWriteSizesStore{Database: db}
+	h := prefixHandlerWithoutTimer(store)
+	announceManyPrefixes(h.table, pairs)
+
+	for range flushes {
+		h.flushBatch()
+	}
+
+	wantChanges := []int{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite}
+	wantNewPeers := []int{1, 0}
+
+	if !slices.Equal(store.changes, wantChanges) ||
+		!slices.Equal(store.newPeers, wantNewPeers) {
+		t.Errorf("the writes held %v pairs and %v new peers, want %v and %v",
+			store.changes, store.newPeers, wantChanges, wantNewPeers)
+	}
+
+	read, err := db.ReadRouteTable(t.Context())
+	if err != nil {
+		t.Fatalf("ReadRouteTable: %v", err)
+	}
+
+	checkSameTable(t, read, h.table, firstManyPrefix)
+}
+
+// TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler,
+// with no deadline, whose routing table holds more changed pairs than two
+// writes take. Its last flush writes one write's worth, so Stop must log the
+// pairs that no write took.
+func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
+	t.Parallel()
+
+	// More than two writes take, so that pairs are left even when the flush
+	// timer writes once before Stop.
+	const pairs = 2*maxTableChangesPerWrite + 1
+
+	var logs bytes.Buffer
+
+	store := &tableWriteSizesStore{Database: newTestDatabase(t)}
+	table := routetable.New(routetable.Params{})
+	announceManyPrefixes(table, pairs)
+
+	h := NewPrefixHandler(store, newErrorLogger(&logs), table)
+	h.Stop(t.Context())
+
+	written := 0
+	for _, n := range store.changes {
+		written += n
+	}
+
+	if logged := lastLogged(t, &logs); logged.TableChanges != pairs-written {
+		t.Errorf("Stop logged %d table changes not written, want %d",
+			logged.TableChanges, pairs-written)
+	}
+}
+
 // readLog holds what readRouteTable logs.
 //
 //nolint:tagliatelle // field names are the keys readRouteTable logs
@@ -383,6 +475,22 @@ func peerWithdrawal(peer string, prefixes ...string) *ristypes.RISMessage {
 	}
 }
 
+// announceManyPrefixes has testPeerIP announce n prefixes in table, with the
+// path of peerAnnouncement: the /32 of each address from firstManyPrefix on.
+func announceManyPrefixes(table *routetable.Table, n int) {
+	addr := netip.MustParsePrefix(firstManyPrefix).Addr()
+	route := routetable.Route{
+		Peer:   testPeerIP,
+		ASPath: []int{testASNA, testASNB},
+		Time:   time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
+	}
+
+	for range n {
+		table.Announce(netip.PrefixFrom(addr, addr.BitLen()), testASNB, route)
+		addr = addr.Next()
+	}
+}
+
 // sessionDownOf returns the STATE message that says the BGP session of peer
 // went down.
 func sessionDownOf(peer string) *ristypes.RISMessage {