Dockerfile

@@ -139,8 +139,13 @@ EXPOSE 8080
 
 COPY ./entrypoint.sh /entrypoint.sh
 
-# Health check using the health endpoint, on the port PORT names
+# Health check using the health endpoint, on the port PORT names. During the
-HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
+# start period, Docker 25 and later checks every 5 seconds, and the first pass
+# makes the container healthy. After it, Docker checks every 30 seconds. The
+# start period covers the 60 seconds after which upaas reads the health
+# (README "Running under upaas"). Startup reads the routing table first,
+# which takes about 25 seconds when the table is full.
+HEALTHCHECK --interval=30s --timeout=5s --start-period=60s --retries=3 \
     CMD curl -sf "http://localhost:${PORT:-8080}/.well-known/healthcheck.json" || exit 1
 
 ENTRYPOINT ["/bin/bash", "/entrypoint.sh" ]

README.md

@@ -285,6 +285,8 @@ routewatch/
 
 2. **Initialization** (`internal/routewatch/app.go`)
     - Database created with SQLite schema (WAL mode, page caches as in "Memory")
+    - The routing table read from the database into memory, before the streamer
+      connects
     - Message handlers registered with the streamer
     - HTTP server started on configured port
 
@@ -311,7 +313,10 @@ routewatch/
       holds AS 0 anywhere, deletes all of a peer's routes when its BGP session
       goes down). Routes are kept per peer IP, so when one peer IP has sessions
       on two collectors, a session going down on one also removes the routes the
-      other still carries, until they are announced again.
+      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. 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**
@@ -495,7 +500,9 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app needs:
 - Health check: the image's `HEALTHCHECK` requests
   `/.well-known/healthcheck.json` on the container port. upaas reads the
   container's health 60 seconds after a deploy and fails the deploy unless it is
-  `healthy`.
+  `healthy`. For the first 60 seconds, Docker 25 and later runs the check every
+  5 seconds, and the first pass makes the container healthy. Startup first reads
+  the routing table, which takes about 25 seconds when the table is full.
 - Stop grace: the container needs 9 seconds between SIGTERM and SIGKILL. upaas
   gives 10 (`stopTimeoutSeconds` in its `internal/docker/client.go`), as
   `docker stop` does by default. The daemon allows 8 seconds for the handlers to

TODO.md

@@ -33,6 +33,12 @@ 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 write
+  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
   `routes` table, one row per prefix and origin AS. The peers' numbers are in a
   new `route_peers` table. One function writes the pairs that changed in memory,
@@ -254,5 +260,5 @@ tested.
 # Future Steps
 
 - the rest of https://git.eeqj.de/sneak/routewatch/issues/122, one PR each: the
-  prefix handler writing to the routing table, the pages and the API reading
+  pages and the API reading from the routing table, then removing the per-peer
-  from it, then removing the per-peer route tables
+  route tables

internal/database/interface.go

@@ -7,6 +7,8 @@ package database
 import (
 	"context"
 	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
 )
 
 // Stats contains database statistics including counts for ASNs, prefixes,
@@ -69,6 +71,10 @@ type Store interface {
 	GetLiveRouteCounts() (ipv4Count, ipv6Count int, err error)
 	GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error)
 
+	// Routing table operations: the database's copy of the table in memory
+	WriteRouteChanges(changes []routetable.Change, newPeers map[string]uint) error
+	ReadRouteTable(ctx context.Context) (*routetable.Table, error)
+
 	// IP lookup operations
 	GetIPInfo(ip string) (*IPInfo, error)
 	GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error)

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/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[n] counts the prefixes the peer numbered n carries, for
-	peerPrefixes map[uint]int
+	// 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
+	// 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{}
 
@@ -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() {
+	// The peers not yet in the peer set join it in one union, which takes
-		if !o.peers.Test(peer) {
+	// less time than setting their bits one at a time.
-			o.peers.Set(peer)
+	joining := change.Peers.Difference(&o.peers)
+	o.peers.InPlaceUnion(joining)
+	t.familyOf(prefix.Addr()).routes += int(joining.Count())
 
-			t.familyOf(prefix.Addr()).routes++
+	for peer := range joining.EachSet() {
-			t.peerPrefixes[peer]++
+		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
+// 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 {
@@ -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 {

internal/routewatch/app.go

@@ -8,6 +8,7 @@ package routewatch
 import (
 	"context"
 	"errors"
+	"fmt"
 	"path/filepath"
 	"sync"
 	"time"
@@ -17,6 +18,7 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/database"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/internal/metrics"
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
 	"git.eeqj.de/sneak/routewatch/internal/server"
 	"git.eeqj.de/sneak/routewatch/internal/streamer"
 
@@ -92,6 +94,14 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 	// Register database handler to process BGP UPDATE messages
 	if rw.config.EnableBatchedDatabaseWrites {
 		rw.logger.Info("Using batched database handlers for improved performance")
+
+		// The routing table is read before the streamer connects, so that
+		// the prefix handler applies the feed's updates to all of it.
+		table, err := rw.readRouteTable(ctx)
+		if err != nil {
+			return err
+		}
+
 		// ASHandler maintains the asns table
 		rw.dbHandler = NewASHandler(rw.db, rw.logger)
 		rw.streamer.RegisterHandler(rw.dbHandler)
@@ -100,8 +110,9 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 		rw.peerHandler = NewPeerHandler(rw.db, rw.logger)
 		rw.streamer.RegisterHandler(rw.peerHandler)
 
-		// PrefixHandler maintains the prefixes and live_routes tables
+		// PrefixHandler maintains the prefixes and live_routes tables, and
-		rw.prefixHandler = NewPrefixHandler(rw.db, rw.logger)
+		// the routing table in memory and its copy in the routes table
+		rw.prefixHandler = NewPrefixHandler(rw.db, rw.logger, table)
 		rw.prefixHandler.SetMetricsTracker(rw.streamer.GetMetricsTracker())
 		rw.streamer.RegisterHandler(rw.prefixHandler)
 
@@ -254,6 +265,27 @@ func (rw *RouteWatch) closeDatabase(ctx context.Context) {
 	}
 }
 
+// readRouteTable reads the routing table from the database and logs how many
+// prefix and origin AS pairs and peers it holds and how long the read took.
+// ctx is fx's start context, which ends 15 seconds after the start. Reading a
+// full table can take longer, so the read does not stop with ctx.
+func (rw *RouteWatch) readRouteTable(
+	ctx context.Context,
+) (*routetable.Table, error) {
+	start := time.Now()
+
+	table, err := rw.db.ReadRouteTable(context.WithoutCancel(ctx))
+	if err != nil {
+		return nil, fmt.Errorf("failed to read the routing table: %w", err)
+	}
+
+	counts := table.Counts()
+	rw.logger.Info("Read the routing table", "pairs", counts.Pairs,
+		"peers", counts.Peers, "duration_ms", time.Since(start).Milliseconds())
+
+	return table, nil
+}
+
 // getModule provides all fx dependencies
 func getModule() []fx.Option {
 	return []fx.Option{

internal/routewatch/app_integration_internal_test.go

@@ -19,6 +19,7 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/internal/metrics"
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
 	"git.eeqj.de/sneak/routewatch/internal/server"
 	"git.eeqj.de/sneak/routewatch/internal/streamer"
 )
@@ -39,6 +40,9 @@ type mockStore struct {
 	Prefixes map[string]*database.Prefix
 	Peerings map[string]bool // key is "from_to"
 	Routes   map[string]bool // key is "prefix_origin_peer"
+	// TablePrefixes holds the prefixes that the written changes of the
+	// routing table leave some peer carrying.
+	TablePrefixes map[string]bool
 
 	// Track IP versions
 	IPv4Prefixes int
@@ -48,10 +52,11 @@ type mockStore struct {
 // newMockStore creates a new mock store
 func newMockStore() *mockStore {
 	return &mockStore{
-		ASNs:     make(map[int]*database.ASN),
+		ASNs:          make(map[int]*database.ASN),
-		Prefixes: make(map[string]*database.Prefix),
+		Prefixes:      make(map[string]*database.Prefix),
-		Peerings: make(map[string]bool),
+		Peerings:      make(map[string]bool),
-		Routes:   make(map[string]bool),
+		Routes:        make(map[string]bool),
+		TablePrefixes: make(map[string]bool),
 	}
 }
 
@@ -395,6 +400,30 @@ func (m *mockStore) DeleteLiveRoutesOfPeer(_ string) (int, error) {
 	return 0, nil
 }
 
+// WriteRouteChanges records the prefixes the changes leave some peer carrying,
+// whatever their origin AS.
+func (m *mockStore) WriteRouteChanges(
+	changes []routetable.Change, _ map[string]uint,
+) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	for _, change := range changes {
+		if change.Removed {
+			delete(m.TablePrefixes, change.Prefix.String())
+		} else {
+			m.TablePrefixes[change.Prefix.String()] = true
+		}
+	}
+
+	return nil
+}
+
+// ReadRouteTable returns an empty routing table.
+func (m *mockStore) ReadRouteTable(_ context.Context) (*routetable.Table, error) {
+	return routetable.New(routetable.Params{}), nil
+}
+
 // GetOrCreateASNBatch mock implementation
 func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error {
 	m.mu.Lock()
@@ -566,7 +595,8 @@ func announcementFeed(prefixes []string) string {
 // handler's worker waits on the first one until the server sees the streamer
 // close the connection, which only Shutdown does, so all three are still in
 // the streamer when Shutdown starts. Each prefix must be in the database when
-// Shutdown returns.
+// Shutdown returns, both in the prefix tables and in the routing table's
+// writes.
 func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 	t.Parallel()
 
@@ -600,7 +630,7 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 		Logger:   log,
 		Config:   cfg,
 	})
-	rw.prefixHandler = NewPrefixHandler(store, log)
+	rw.prefixHandler = NewPrefixHandler(store, log, routetable.New(routetable.Params{}))
 	s.RegisterHandler(gatedPrefixHandler{
 		PrefixHandler: rw.prefixHandler, release: release,
 	})
@@ -640,9 +670,11 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 	defer store.mu.Unlock()
 
 	for _, prefix := range prefixes {
-		if _, ok := store.Prefixes[prefix]; !ok {
+		_, inPrefixTables := store.Prefixes[prefix]
-			t.Errorf("%s was queued in the streamer when Shutdown started "+
+		if !inPrefixTables || !store.TablePrefixes[prefix] {
-				"but is not in the database", prefix)
+			t.Errorf("%s was queued in the streamer when Shutdown started; "+
+				"written to the prefix tables %t, in the routing table's "+
+				"writes %t", prefix, inPrefixTables, store.TablePrefixes[prefix])
 		}
 	}
 }

internal/routewatch/batchwrite_internal_test.go

@@ -13,6 +13,7 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/database"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
 )
 
 const (
@@ -119,10 +120,11 @@ func (s *blockingStore) releaseWrites() {
 //
 //nolint:tagliatelle // field names are the keys the handlers log
 type stopLog struct {
-	Msg       string `json:"msg"`
+	Msg          string `json:"msg"`
-	Updates   int    `json:"updates"`
+	Updates      int    `json:"updates"`
-	PeersDown int    `json:"peers_down"`
+	PeersDown    int    `json:"peers_down"`
-	Paths     int    `json:"paths"`
+	TableChanges int    `json:"table_changes"`
+	Paths        int    `json:"paths"`
 }
 
 // newErrorLogger returns a logger that writes only errors, as JSON lines, to
@@ -313,6 +315,9 @@ func stopWhileAWriteIsBlocked(
 	}
 }
 
+// TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline also checks the
+// routing table's changes: the flush whose delete is held wrote those of the
+// fill, so Stop must log the one of the extra prefix as not written.
 func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
@@ -321,7 +326,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	var logs bytes.Buffer
 
 	store := newBlockingStore()
-	h := NewPrefixHandler(store, newErrorLogger(&logs))
+	h := NewPrefixHandler(store, newErrorLogger(&logs),
+		routetable.New(routetable.Params{}))
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	// The peer's session goes down first, so the write held is the delete of
 	// its routes, and Stop must log that peer down too.
@@ -337,9 +343,51 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	}
 
 	logged := lastLogged(t, &logs)
-	if logged.Updates != prefixBatchSize+1 || logged.PeersDown != 1 {
+	if logged.Updates != prefixBatchSize+1 || logged.PeersDown != 1 ||
-		t.Errorf("Stop logged %d updates and %d peer downs not written, "+
+		logged.TableChanges != 1 {
-			"want %d and 1", logged.Updates, logged.PeersDown, prefixBatchSize+1)
+		t.Errorf("Stop logged %d updates, %d peer downs and %d table changes "+
+			"not written, want %d, 1 and 1", logged.Updates, logged.PeersDown,
+			logged.TableChanges, prefixBatchSize+1)
+	}
+}
+
+// tableWriteBlockingStore is a blockingStore whose writes of the routing
+// table's changes also wait until the test calls releaseWrites.
+type tableWriteBlockingStore struct {
+	*blockingStore
+}
+
+func (s tableWriteBlockingStore) WriteRouteChanges(
+	changes []routetable.Change, newPeers map[string]uint,
+) error {
+	s.waitForRelease()
+
+	return s.mockStore.WriteRouteChanges(changes, newPeers)
+}
+
+// TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress holds the write
+// of the routing table's changes that a full batch starts, then hands the
+// prefix handler one more prefix and stops it with a deadline that passes
+// meanwhile. Stop must log as not written both the change the held write took
+// and the one since.
+func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T) {
+	t.Parallel()
+
+	const extraPrefix = "198.51.100.0/24"
+
+	var logs bytes.Buffer
+
+	blocking := newBlockingStore()
+	h := NewPrefixHandler(tableWriteBlockingStore{blockingStore: blocking},
+		newErrorLogger(&logs), routetable.New(routetable.Params{}))
+	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
+	stopWhileAWriteIsBlocked(t, blocking, h, h.done,
+		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
+		prefixAnnouncement([]string{extraPrefix}))
+
+	if logged := lastLogged(t, &logs); logged.TableChanges != 2 {
+		t.Errorf("Stop logged %d table changes not written, want 2",
+			logged.TableChanges)
 	}
 }
 
@@ -455,7 +503,7 @@ func TestPrefixHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 	const extraPrefix = "198.51.100.0/24"
 
 	store := newBlockingStore()
-	h := NewPrefixHandler(store, logger.New())
+	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	handleWhileAWriteIsBlocked(t, store, h,
 		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -515,7 +563,7 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	const extraPrefix = "198.51.100.0/24"
 
 	store := newBlockingStore()
-	h := NewPrefixHandler(store, logger.New())
+	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchMaxSize)
 	handleAtMaxBatchSize(t, store, h,
@@ -592,7 +640,10 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			name:      "prefix",
 			batchSize: prefixBatchSize,
 			handle: func(n int) chan struct{} {
-				h := &PrefixHandler{batchFullCh: make(chan struct{}, 1)}
+				h := &PrefixHandler{
+					batchFullCh: make(chan struct{}, 1),
+					table:       routetable.New(routetable.Params{}),
+				}
 				h.HandleMessage(prefixAnnouncement(
 					slices.Repeat([]string{testV4Prefix}, n)))
 

internal/routewatch/prefixhandler.go

@@ -13,6 +13,7 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/internal/metrics"
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
 )
 
 const (
@@ -38,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
@@ -47,15 +58,19 @@ const (
 // database. Routes are added on announcement and deleted on withdrawal or on an
 // announcement whose AS path is empty or holds AS 0 anywhere, and all routes of
 // a peer are deleted when its BGP session goes down.
+//
+// It applies the same updates to the routing table in memory as it handles
+// them, and writes the table's changes to the database with each flush.
 type PrefixHandler struct {
 	db      database.Store
 	logger  *logger.Logger
 	metrics *metrics.Tracker
+	table   *routetable.Table
 
-	// Batching. mu guards batch, held, peersDown, writing and deleting only:
+	// Batching. mu guards batch, held, peersDown, writing, deleting and
-	// the flush goroutine takes them under mu and writes them after releasing
+	// writingTableChanges only: the flush goroutine takes them under mu and
-	// mu, so HandleMessage waits for a database write only while batch is at
+	// writes them after releasing mu, so HandleMessage waits for a database
-	// prefixBatchMaxSize.
+	// write only while batch is at prefixBatchMaxSize.
 	mu    sync.Mutex
 	batch []prefixUpdate
 	// held holds the updates that the peers whose route delete failed in the
@@ -74,6 +89,9 @@ type PrefixHandler struct {
 	// deleting is how many peers from peersDown the flush goroutine took and
 	// has not yet deleted the routes of.
 	deleting int
+	// writingTableChanges is how many changes the flush goroutine took from
+	// the routing table and has not finished writing.
+	writingTableChanges int
 	// batchTaken is signalled when the flush goroutine takes batch; its L is
 	// &mu.
 	batchTaken  sync.Cond
@@ -97,11 +115,26 @@ type prefixUpdate struct {
 	endsInSet bool
 }
 
-// NewPrefixHandler creates a new batched prefix tracking handler
+// announcesRoute says whether u gives the peer a route to the prefix: an
-func NewPrefixHandler(db database.Store, logger *logger.Logger) *PrefixHandler {
+// announcement whose path gives an origin, or ends in an AS set of several
+// members, which gives origin 0 (none). Any other update removes the peer's
+// route for the prefix: a withdrawal, or an announcement whose path is empty
+// or holds AS 0 anywhere (the decoder empties such a path). Such an
+// announcement replaces the peer's route, as any announcement does, with none
+// (RFC 7607 treats AS 0 in a path as a withdrawal).
+func (u prefixUpdate) announcesRoute() bool {
+	return u.messageType == "announcement" && (u.originASN > 0 || u.endsInSet)
+}
+
+// NewPrefixHandler creates a new batched prefix tracking handler, which applies
+// the updates to table, the routing table read from db at startup.
+func NewPrefixHandler(
+	db database.Store, logger *logger.Logger, table *routetable.Table,
+) *PrefixHandler {
 	h := &PrefixHandler{
 		db:          db,
 		logger:      logger,
+		table:       table,
 		batch:       make([]prefixUpdate, 0, prefixBatchSize),
 		batchFullCh: make(chan struct{}, 1),
 		lastFlush:   time.Now(),
@@ -165,18 +198,22 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 		h.batchTaken.Wait()
 	}
 
-	// Withdrawals are queued before announcements, because the batch keeps
+	// Withdrawals are queued, and applied to the routing table, before
-	// each peer's last update for a prefix: a message that lists a prefix as
+	// announcements, because the batch keeps each peer's last update for a
-	// both withdrawn and announced keeps the route (RFC 4271 section 4.3).
+	// prefix: a message that lists a prefix as both withdrawn and announced
-	// A withdrawal names only the prefix and peer: the message's AS path, and
+	// keeps the route (RFC 4271 section 4.3). A withdrawal names only the
-	// so its origin AS, belongs to the announced prefixes.
+	// prefix and peer: the message's AS path, and so its origin AS, belongs
+	// to the announced prefixes.
 	for _, prefix := range msg.Withdrawals {
-		h.batch = append(h.batch, prefixUpdate{
+		update := prefixUpdate{
 			prefix:      canonicalPrefix(prefix),
 			peer:        msg.Peer,
 			messageType: "withdrawal",
 			timestamp:   timestamp,
-		})
+		}
+		h.batch = append(h.batch, update)
+		h.applyToTable(update)
+
 		// Record withdrawal in metrics
 		if h.metrics != nil {
 			h.metrics.RecordWithdrawal()
@@ -185,7 +222,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 
 	for _, announcement := range msg.Announcements {
 		for _, prefix := range announcement.Prefixes {
-			h.batch = append(h.batch, prefixUpdate{
+			update := prefixUpdate{
 				prefix:      canonicalPrefix(prefix),
 				originASN:   originASN,
 				peer:        msg.Peer,
@@ -193,7 +230,10 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 				timestamp:   timestamp,
 				path:        path,
 				endsInSet:   len(msg.Path.Set) > 0,
-			})
+			}
+			h.batch = append(h.batch, update)
+			h.applyToTable(update)
+
 			// Record announcement in metrics
 			if h.metrics != nil {
 				h.metrics.RecordAnnouncement()
@@ -212,12 +252,15 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop stops the flush goroutine, which writes the updates and peer downs the
+// Stop stops the flush goroutine, which makes one last flush before it returns,
-// handler holds before it returns, and waits for it until ctx ends. Once ctx
+// and waits for it until ctx ends. That flush writes the updates and peer downs
-// has ended the flush goroutine skips that last write, and a write still
+// the handler holds, and at most maxTableChangesPerWrite of the routing table's
-// running is lost when the process exits. A peer whose route delete fails in
+// changes. Once ctx has ended the flush goroutine skips that last flush, and a
-// the last write is not retried either. Stop logs how many updates are not
+// write still running is lost when the process exits. A peer whose route delete
-// written and how many peer downs have not had their routes deleted.
+// fails in the last flush is not retried either, nor is a failed write of the
+// table's changes. Stop logs how many updates are not written, how many peer
+// 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
 
@@ -229,22 +272,24 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 	h.mu.Lock()
 	unwritten := h.writing + len(h.batch) + len(h.held)
 	peersDown := h.deleting + len(h.peersDown)
+	tableChanges := h.writingTableChanges + h.table.ChangeCount()
 	h.mu.Unlock()
 
-	if unwritten == 0 && peersDown == 0 {
+	if unwritten == 0 && peersDown == 0 && tableChanges == 0 {
 		return
 	}
 
 	if ctx.Err() != nil {
 		h.logger.Error("Stop deadline passed before the prefix handler wrote "+
-			"its updates", "updates", unwritten, "peers_down", peersDown)
+			"its updates", "updates", unwritten, "peers_down", peersDown,
+			"table_changes", tableChanges)
 
 		return
 	}
 
-	h.logger.Error("The prefix handler stopped without deleting the routes "+
+	h.logger.Error("The prefix handler stopped before it wrote everything",
-		"of peers whose session went down", "updates", unwritten,
+		"updates", unwritten, "peers_down", peersDown,
-		"peers_down", peersDown)
+		"table_changes", tableChanges)
 }
 
 // peerSessionDown notes that the peer's BGP session went down, so that the next
@@ -252,11 +297,14 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 // peer's updates still in the batch or held came before the down, which
 // replaces them, so they are dropped; updates the peer sends later stay in the
 // batch and are written after the delete. A peer already in peersDown is not
-// added again: one delete removes all of its routes.
+// added again: one delete removes all of its routes. In the routing table in
+// memory the down takes effect at once.
 func (h *PrefixHandler) peerSessionDown(peer string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
+	h.table.PeerDown(peer)
+
 	isPeers := func(update prefixUpdate) bool { return update.peer == peer }
 	h.batch = slices.DeleteFunc(h.batch, isPeers)
 	h.held = slices.DeleteFunc(h.held, isPeers)
@@ -321,9 +369,11 @@ func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
 	return batch, peersDown
 }
 
-// flushBatch takes the pending batch and writes it to the database, without
+// flushBatch writes the routing table's changes, then takes the pending batch
-// holding the mutex during the write.
+// and writes it to the database, without holding the mutex during the writes.
 func (h *PrefixHandler) flushBatch() {
+	h.writeTableChanges()
+
 	batch, peersDown := h.takeBatch()
 	if len(batch) == 0 && len(peersDown) == 0 {
 		return
@@ -370,15 +420,7 @@ func (h *PrefixHandler) flushBatch() {
 	)
 
 	for _, update := range lastUpdates {
-		// An announcement is stored when its path gives an origin, or when it
+		if update.announcesRoute() {
-		// ends in an AS set of several members: that route is stored with
-		// origin 0 (none). Any other update deletes the peer's route for the
-		// prefix: a withdrawal, or an announcement whose path is empty or
-		// holds AS 0 anywhere (the decoder empties such a path). Such an
-		// announcement replaces the peer's route, as any announcement does,
-		// with none (RFC 7607 treats AS 0 in a path as a withdrawal).
-		if update.messageType == "announcement" &&
-			(update.originASN > 0 || update.endsInSet) {
 			// Create live route for batch upsert
 			route := h.createLiveRoute(update)
 			if route != nil {
@@ -421,6 +463,59 @@ func (h *PrefixHandler) flushBatch() {
 	h.lastFlush = time.Now()
 }
 
+// applyToTable applies update to the routing table in memory: as an
+// announcement when it gives the peer a route, otherwise as a withdrawal. A
+// prefix that does not parse is left out; flushBatch logs it.
+func (h *PrefixHandler) applyToTable(update prefixUpdate) {
+	prefix, err := netip.ParsePrefix(update.prefix)
+	if err != nil {
+		return
+	}
+
+	if !update.announcesRoute() {
+		h.table.Withdraw(prefix, update.peer)
+
+		return
+	}
+
+	h.table.Announce(prefix, update.originASN, routetable.Route{
+		Peer:   update.peer,
+		ASPath: update.path,
+		Time:   update.timestamp,
+	})
+}
+
+// writeTableChanges writes at most maxTableChangesPerWrite changes of the
+// routing table in memory, and every peer the table numbered since the last
+// 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(maxTableChangesPerWrite)
+	h.writingTableChanges = len(changes)
+	h.mu.Unlock()
+
+	if len(changes) == 0 && len(newPeers) == 0 {
+		return
+	}
+
+	err := h.db.WriteRouteChanges(changes, newPeers)
+
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	h.writingTableChanges = 0
+
+	if err != nil {
+		h.logger.Error("Failed to write the routing table's changes; the next "+
+			"flush retries", "changes", len(changes), "new_peers", len(newPeers),
+			"error", err)
+
+		h.table.PutBackChanges(changes, newPeers)
+	}
+}
+
 // deleteLiveRoutesOfPeers deletes all live routes of each of peers, the peers
 // whose BGP session went down, and returns those whose delete failed. Most
 // downs come from peers that have no routes, so only a delete that removed

internal/routewatch/prefixhandler_internal_test.go

@@ -16,6 +16,7 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/database"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
 )
 
 const testPeerIP = "2001:db8:ffff::1"
@@ -40,9 +41,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 	t.Parallel()
 
 	db := newTestDatabase(t)
-
+	h := prefixHandlerWithoutTimer(db)
-	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
-	h := &PrefixHandler{db: db, logger: logger.New()}
 
 	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
 	h.HandleMessage(&ristypes.RISMessage{
@@ -462,8 +461,7 @@ func TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse(t *testing.T) {
 		t.Fatalf("failed to create database: %v", err)
 	}
 
-	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+	h := prefixHandlerWithoutTimer(db)
-	h := &PrefixHandler{db: db, logger: logger.New()}
 
 	withdrawal := &ristypes.RISMessage{
 		Type:            ristypes.MessageTypeUpdate,
@@ -694,8 +692,7 @@ func TestPrefixHandlerHandlesADownWhileAFlushWrites(t *testing.T) {
 		release:      make(chan struct{}),
 	}
 
-	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+	h := prefixHandlerWithoutTimer(store)
-	h := &PrefixHandler{db: store, logger: logger.New()}
 	h.HandleMessage(peerAnnouncement(testPeerIP, beforeDown))
 
 	flushed := make(chan struct{})
@@ -832,10 +829,7 @@ func TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes(t *testing.T) {
 
 			db := newTestDatabase(t)
 			store := &firstPeerDeleteFailsStore{Database: db}
-
+			h := prefixHandlerWithoutTimer(store)
-			// Built without NewPrefixHandler's flush timer, so the test
-			// flushes itself.
-			h := &PrefixHandler{db: store, logger: logger.New()}
 			store.whileFailing = func() {
 				for _, msg := range tt.whileFailing {
 					h.HandleMessage(msg)
@@ -897,11 +891,7 @@ func TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing(
 	otherPeersPrefixes := []string{testV4Prefix, testV6Prefix, "2001:db8:2::/48"}
 
 	db := newTestDatabase(t)
-
+	h := prefixHandlerWithoutTimer(peerDeleteAlwaysFailsStore{Database: db})
-	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
-	h := &PrefixHandler{
-		db: peerDeleteAlwaysFailsStore{Database: db}, logger: logger.New(),
-	}
 	h.HandleMessage(sessionDown())
 
 	// The peer announces a /32 for each address from 10.0.0.0 on, each
@@ -953,7 +943,7 @@ func TestPrefixHandlerStopLogsAFailedDeleteWithoutTheDeadline(t *testing.T) {
 	var logs bytes.Buffer
 
 	h := NewPrefixHandler(peerDeleteAlwaysFailsStore{Database: newTestDatabase(t)},
-		newErrorLogger(&logs))
+		newErrorLogger(&logs), routetable.New(routetable.Params{}))
 	h.HandleMessage(sessionDown())
 	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 	h.Stop(t.Context())
@@ -979,7 +969,8 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) {
 
 	var logs bytes.Buffer
 
-	h := NewPrefixHandler(newTestDatabase(t), newErrorLogger(&logs))
+	h := NewPrefixHandler(newTestDatabase(t), newErrorLogger(&logs),
+		routetable.New(routetable.Params{}))
 	h.HandleMessage(sessionDown())
 	h.Stop(t.Context())
 
@@ -989,6 +980,17 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) {
 	}
 }
 
+// prefixHandlerWithoutTimer returns a prefix handler that writes to store and
+// applies the updates to an empty routing table. It is built without
+// NewPrefixHandler's flush goroutine, so the test flushes itself.
+func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
+	return &PrefixHandler{
+		db:     store,
+		logger: logger.New(),
+		table:  routetable.New(routetable.Params{}),
+	}
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {
@@ -1012,9 +1014,7 @@ func handleBatches(
 	t.Helper()
 
 	db := newTestDatabase(t)
-
+	h := prefixHandlerWithoutTimer(db)
-	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
-	h := &PrefixHandler{db: db, logger: logger.New()}
 
 	for _, batch := range batches {
 		for _, msg := range batch {

internal/routewatch/prefixhandler_table_internal_test.go

@@ -0,0 +1,535 @@
+package routewatch
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"net/netip"
+	"reflect"
+	"slices"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/database"
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+)
+
+// More peers and prefixes for the tests of the routing table in memory.
+const (
+	secondPeerIP = "2001:db8:ffff::2"
+	thirdPeerIP  = "2001:db8:ffff::3"
+	fourthPeerIP = "192.0.2.4"
+	fifthPeerIP  = "192.0.2.5"
+
+	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
+// routing table's changes: a few ticks of it.
+const tableWriteTimeout = 5 * prefixBatchTimeout
+
+// tableCase is one test of the routing table in memory: the messages handed to
+// the prefix handler, and for each prefix the peers that carry it with each
+// origin AS after them.
+type tableCase struct {
+	name     string
+	messages []*ristypes.RISMessage
+	want     map[string]map[int][]string
+}
+
+// TestPrefixHandlerAppliesUpdatesToTheRoutingTable hands the prefix handler
+// announcements, withdrawals and a peer's session going down.
+func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
+	t.Parallel()
+
+	announceAndWithdraw := peerAnnouncement(testPeerIP, testV4Prefix)
+	announceAndWithdraw.Withdrawals = []string{testV4Prefix}
+
+	checkTableCases(t, []tableCase{
+		{
+			name:     "an announcement",
+			messages: []*ristypes.RISMessage{peerAnnouncement(testPeerIP, testV4Prefix)},
+			want:     map[string]map[int][]string{testV4Prefix: {testASNB: {testPeerIP}}},
+		},
+		{
+			name: "a withdrawal",
+			messages: []*ristypes.RISMessage{
+				peerAnnouncement(testPeerIP, testV4Prefix),
+				peerWithdrawal(testPeerIP, testV4Prefix),
+			},
+			want: map[string]map[int][]string{testV4Prefix: {}},
+		},
+		{
+			// RFC 4271 section 4.3: the announcement counts, not the withdrawal.
+			name:     "a message that announces and withdraws the prefix",
+			messages: []*ristypes.RISMessage{announceAndWithdraw},
+			want:     map[string]map[int][]string{testV4Prefix: {testASNB: {testPeerIP}}},
+		},
+		{
+			name: "a peer's session going down",
+			messages: []*ristypes.RISMessage{
+				peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix),
+				peerAnnouncement(secondPeerIP, testV4Prefix),
+				sessionDown(),
+			},
+			want: map[string]map[int][]string{
+				testV4Prefix: {testASNB: {secondPeerIP}},
+				testV6Prefix: {},
+			},
+		},
+	})
+}
+
+// TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable hands the
+// prefix handler announcements whose AS path gives no origin. One whose path
+// holds AS 0 removes the peer's route, as a withdrawal does; one whose path ends
+// in an AS set of several members is kept with origin 0.
+func TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable(t *testing.T) {
+	t.Parallel()
+
+	announceWithPath := func(path ristypes.ASPath) *ristypes.RISMessage {
+		msg := peerAnnouncement(testPeerIP, testV4Prefix)
+		msg.Path = path
+
+		return msg
+	}
+
+	checkTableCases(t, []tableCase{
+		{
+			name: "a path whose last AS is 0",
+			messages: []*ristypes.RISMessage{
+				peerAnnouncement(testPeerIP, testV4Prefix),
+				announceWithPath(ristypes.ASPath{Sequence: []int{testASNA, 0}}),
+			},
+			want: map[string]map[int][]string{testV4Prefix: {}},
+		},
+		{
+			name: "a path with AS 0 before its end",
+			messages: []*ristypes.RISMessage{
+				peerAnnouncement(testPeerIP, testV4Prefix),
+				announceWithPath(decodePath(t, testASNA, 0, testASNB)),
+			},
+			want: map[string]map[int][]string{testV4Prefix: {}},
+		},
+		{
+			name: "a path that ends in an AS set",
+			messages: []*ristypes.RISMessage{
+				announceWithPath(decodePath(t, testASNA, []int{testASNC, testASND})),
+			},
+			want: map[string]map[int][]string{testV4Prefix: {0: {testPeerIP}}},
+		},
+	})
+}
+
+// checkTableCases hands the messages of each case to a new prefix handler and
+// checks its routing table right after, with no flush.
+func checkTableCases(t *testing.T, tests []tableCase) {
+	t.Helper()
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			h := prefixHandlerWithoutTimer(newMockStore())
+			for _, msg := range tt.messages {
+				h.HandleMessage(msg)
+			}
+
+			for prefix, want := range tt.want {
+				got := tableOrigins(h.table, prefix)
+				if !reflect.DeepEqual(got, want) {
+					t.Errorf("peers of %s by origin AS = %v, want %v",
+						prefix, got, want)
+				}
+			}
+		})
+	}
+}
+
+// tableWritesStore is a real database that hands each write of the routing
+// table's changes to written once it is made.
+type tableWritesStore struct {
+	*database.Database
+
+	written chan []routetable.Change
+}
+
+func (s tableWritesStore) WriteRouteChanges(
+	changes []routetable.Change, newPeers map[string]uint,
+) error {
+	err := s.Database.WriteRouteChanges(changes, newPeers)
+	s.written <- changes
+
+	return err
+}
+
+// TestPrefixHandlerWritesTheTableChangesEachSecond hands a running prefix
+// handler an announcement and then its withdrawal, each far short of a full
+// batch. The flush timer must write the change of each to the database.
+func TestPrefixHandlerWritesTheTableChangesEachSecond(t *testing.T) {
+	t.Parallel()
+
+	// The two writes the test waits for, and room for one more.
+	const writesBuffered = 3
+
+	store := tableWritesStore{
+		Database: newTestDatabase(t),
+		written:  make(chan []routetable.Change, writesBuffered),
+	}
+	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
+
+	defer h.Stop(t.Context())
+
+	prefix := netip.MustParsePrefix(testV4Prefix)
+
+	for _, step := range []struct {
+		msg         *ristypes.RISMessage
+		wantRemoved bool
+	}{
+		{msg: peerAnnouncement(testPeerIP, testV4Prefix)},
+		{msg: peerWithdrawal(testPeerIP, testV4Prefix), wantRemoved: true},
+	} {
+		h.HandleMessage(step.msg)
+
+		select {
+		case changes := <-store.written:
+			if len(changes) != 1 || changes[0].Prefix != prefix ||
+				changes[0].OriginASN != testASNB ||
+				changes[0].Removed != step.wantRemoved {
+				t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+
+					"removed %t", changes, prefix, testASNB, step.wantRemoved)
+			}
+		case <-time.After(tableWriteTimeout):
+			t.Fatalf("the routing table's change was not written within %s",
+				tableWriteTimeout)
+		}
+	}
+}
+
+// errTestTableWrite is the error a test store's failing write of the routing
+// table's changes returns.
+var errTestTableWrite = errors.New("test: the write of the table's changes failed")
+
+// firstTableWriteFailsStore is a real database whose first write of the
+// routing table's changes fails, writing nothing. writes counts the writes
+// asked for.
+type firstTableWriteFailsStore struct {
+	*database.Database
+
+	writes int
+}
+
+func (s *firstTableWriteFailsStore) WriteRouteChanges(
+	changes []routetable.Change, newPeers map[string]uint,
+) error {
+	s.writes++
+	if s.writes == 1 {
+		return errTestTableWrite
+	}
+
+	return s.Database.WriteRouteChanges(changes, newPeers)
+}
+
+// TestPrefixHandlerWritesAFailedTableWriteAgain runs two flushes into a
+// database whose first write of the routing table's changes fails. Before the
+// first, two new peers announce a prefix each; before the second, one of them
+// announces its prefix again with another path to the same origin AS. The
+// second write must hold the changes and new peers of the first, with the later
+// path, so the database holds the table that is in memory.
+func TestPrefixHandlerWritesAFailedTableWriteAgain(t *testing.T) {
+	t.Parallel()
+
+	db := newTestDatabase(t)
+	store := &firstTableWriteFailsStore{Database: db}
+	h := prefixHandlerWithoutTimer(store)
+
+	first := peerAnnouncement(testPeerIP, testV4Prefix)
+	h.HandleMessage(first)
+	h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix))
+	h.flushBatch()
+
+	laterPath := []int{testASNC, testASNB}
+	later := peerAnnouncement(testPeerIP, testV4Prefix)
+	later.Path = ristypes.ASPath{Sequence: laterPath}
+	later.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
+	h.HandleMessage(later)
+	h.flushBatch()
+
+	if store.writes != 2 {
+		t.Fatalf("the store was asked for %d writes of the table's changes, "+
+			"want 2", store.writes)
+	}
+
+	read, err := db.ReadRouteTable(t.Context())
+	if err != nil {
+		t.Fatalf("ReadRouteTable: %v", err)
+	}
+
+	checkSameTable(t, read, h.table, testV4Prefix, otherV4Prefix)
+
+	origins := read.Origins(netip.MustParsePrefix(testV4Prefix))
+	if len(origins) != 1 || len(origins[0].Routes) != 1 ||
+		!slices.Equal(origins[0].Routes[0].ASPath, laterPath) {
+		t.Errorf("Origins(%s) = %+v, want one route with the path %v",
+			testV4Prefix, origins, laterPath)
+	}
+}
+
+// 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
+type readLog struct {
+	Msg        string `json:"msg"`
+	Pairs      int    `json:"pairs"`
+	Peers      int    `json:"peers"`
+	DurationMs *int64 `json:"duration_ms"`
+}
+
+// TestARestartReadsTheRoutingTableBack runs announcements, a withdrawal and a
+// peer's session going down through a prefix handler into a database, then
+// reads the table back as startup does. The table read must answer as the one
+// in memory, and the log must give its pairs and peers. A new handler on the
+// table read then takes later updates and writes them as it stops: the
+// database must then hold what the first table holds after the same updates.
+func TestARestartReadsTheRoutingTableBack(t *testing.T) {
+	t.Parallel()
+
+	prefixes := []string{testV4Prefix, testV6Prefix, otherV4Prefix, setV4Prefix}
+
+	db := newTestDatabase(t)
+	h := prefixHandlerWithoutTimer(db)
+
+	viaOtherOrigin := peerAnnouncement(thirdPeerIP, testV4Prefix)
+	viaOtherOrigin.Path = ristypes.ASPath{Sequence: []int{testASNA, testASNC}}
+	endsInSet := peerAnnouncement(secondPeerIP, setV4Prefix)
+	endsInSet.Path = decodePath(t, testASNA, []int{testASNC, testASND})
+
+	for _, msg := range []*ristypes.RISMessage{
+		peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix),
+		peerAnnouncement(secondPeerIP, testV4Prefix),
+		viaOtherOrigin,
+		endsInSet,
+		peerAnnouncement(thirdPeerIP, otherV4Prefix),
+		peerWithdrawal(thirdPeerIP, otherV4Prefix),
+		peerAnnouncement(fourthPeerIP, otherV4Prefix),
+		sessionDownOf(fourthPeerIP),
+	} {
+		h.HandleMessage(msg)
+	}
+
+	h.flushBatch()
+
+	var logs bytes.Buffer
+
+	rw := &RouteWatch{
+		db:     db,
+		logger: &logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logs, nil))},
+	}
+
+	table, err := rw.readRouteTable(t.Context())
+	if err != nil {
+		t.Fatalf("readRouteTable: %v", err)
+	}
+
+	checkSameTable(t, table, h.table, prefixes...)
+
+	// testV4Prefix with two origin ASes, testV6Prefix and setV4Prefix; four
+	// peers, the last of which carries nothing.
+	const wantPairs, wantPeers = 4, 4
+
+	var logged readLog
+
+	err = json.Unmarshal(bytes.TrimSpace(logs.Bytes()), &logged)
+	if err != nil || logged.Pairs != wantPairs || logged.Peers != wantPeers ||
+		logged.DurationMs == nil {
+		t.Errorf("readRouteTable logged %q, want %d pairs, %d peers and the "+
+			"duration", logs.String(), wantPairs, wantPeers)
+	}
+
+	restarted := NewPrefixHandler(db, logger.New(), table)
+
+	for _, msg := range []*ristypes.RISMessage{
+		peerWithdrawal(testPeerIP, testV6Prefix),
+		sessionDownOf(secondPeerIP),
+		peerAnnouncement(fourthPeerIP, otherV4Prefix),
+		peerAnnouncement(fifthPeerIP, setV4Prefix),
+	} {
+		// The first handler only applies the updates to its table.
+		h.HandleMessage(msg)
+		restarted.HandleMessage(msg)
+	}
+
+	restarted.Stop(t.Context())
+
+	read, err := db.ReadRouteTable(t.Context())
+	if err != nil {
+		t.Fatalf("ReadRouteTable after the restart: %v", err)
+	}
+
+	checkSameTable(t, read, h.table, prefixes...)
+}
+
+// peerWithdrawal returns an UPDATE message in which peer withdraws prefixes.
+func peerWithdrawal(peer string, prefixes ...string) *ristypes.RISMessage {
+	return &ristypes.RISMessage{
+		Type:            ristypes.MessageTypeUpdate,
+		Peer:            peer,
+		ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
+		Withdrawals:     prefixes,
+	}
+}
+
+// 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 {
+	return &ristypes.RISMessage{
+		Type:  ristypes.MessageTypeState,
+		Peer:  peer,
+		State: ristypes.StateDown,
+	}
+}
+
+// tableOrigins returns, for each origin AS of prefix in table, the peers that
+// carry the prefix with it.
+func tableOrigins(table *routetable.Table, prefix string) map[int][]string {
+	origins := make(map[int][]string)
+	for _, o := range table.Origins(netip.MustParsePrefix(prefix)) {
+		origins[o.ASN] = o.Peers
+	}
+
+	return origins
+}
+
+// checkSameTable checks that got answers as want does for each of prefixes, and
+// that the two tables have the same counts.
+func checkSameTable(
+	t *testing.T, got, want *routetable.Table, prefixes ...string,
+) {
+	t.Helper()
+
+	for _, text := range prefixes {
+		prefix := netip.MustParsePrefix(text)
+
+		gotOrigins, wantOrigins := got.Origins(prefix), want.Origins(prefix)
+		if !reflect.DeepEqual(gotOrigins, wantOrigins) {
+			t.Errorf("Origins(%s) = %+v, want %+v", text, gotOrigins, wantOrigins)
+		}
+	}
+
+	gotCounts, wantCounts := got.Counts(), want.Counts()
+	if !reflect.DeepEqual(gotCounts, wantCounts) {
+		t.Errorf("Counts() = %+v, want %+v", gotCounts, wantCounts)
+	}
+}