Remove the per-peer route tables (closes #122)

The routing table in memory now holds every route, so the two per-peer route
tables go, with every write, delete, count and read that served them. The
prefix handler applies each update to the table, writes the table's changes
every second, and stores each new prefix with its first-seen time. The peer
delete, its retry on the next flush and the updates held for it go too. The
types and errors of the server's answers move to internal/server.

The image's memory settings fit a full table of about 1 GB: GOMEMLIMIT of
2.5 GiB, a 512 MiB page cache for the writing connection, and SQLite heap
limits of 768 MiB and 1 GiB. The README describes the new table, its
database copy and the memory budget.

Model: opus-5-5

internal/routewatch/app.go

@@ -109,8 +109,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, and
-		// the routing table in memory and its copy in the routes table
+		// PrefixHandler keeps the routing table in memory current, and writes
+		// its changes to the routes table and new prefixes to the prefix
+		// tables
 		rw.prefixHandler = NewPrefixHandler(rw.db, rw.logger, rw.table)
 		rw.prefixHandler.SetMetricsTracker(rw.streamer.GetMetricsTracker())
 		rw.streamer.RegisterHandler(rw.prefixHandler)
@@ -125,8 +126,6 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 		return errNonBatchedNotImplemented
 	}
 
-	// No longer need routing table handler - PrefixHandler maintains live_routes table
-
 	// Start streaming
 	err := rw.streamer.Start(ctx)
 	if err != nil {

internal/routewatch/app_integration_internal_test.go

@@ -33,14 +33,12 @@ type mockStore struct {
 	ASNCount        int
 	PrefixCount     int
 	PeeringCount    int
-	RouteCount      int
 	WithdrawalCount int
 
 	// Track unique items
 	ASNs     map[int]*database.ASN
 	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
@@ -56,7 +54,6 @@ func newMockStore() *mockStore {
 		ASNs:          make(map[int]*database.ASN),
 		Prefixes:      make(map[string]*database.Prefix),
 		Peerings:      make(map[string]bool),
-		Routes:        make(map[string]bool),
 		TablePrefixes: make(map[string]bool),
 	}
 }
@@ -107,6 +104,13 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 				IPVersion: ipVersion,
 				FirstSeen: timestamp,
 			}
+			m.PrefixCount++
+
+			if ipVersion == ipVersionV4 {
+				m.IPv4Prefixes++
+			} else {
+				m.IPv6Prefixes++
+			}
 		}
 	}
 
@@ -207,65 +211,6 @@ func (m *mockStore) GetStatsContext(_ context.Context) (database.Stats, error) {
 	return m.GetStats()
 }
 
-// UpsertLiveRoute mock implementation
-func (m *mockStore) UpsertLiveRoute(_ *database.LiveRoute) error {
-	// Simple mock - just return nil
-	return nil
-}
-
-// DeleteLiveRoute mock implementation
-func (m *mockStore) DeleteLiveRoute(_, _ string) error {
-	// Simple mock - just return nil
-	return nil
-}
-
-// GetPrefixDistribution mock implementation
-func (m *mockStore) GetPrefixDistribution() (
-	[]database.PrefixDistribution, []database.PrefixDistribution, error,
-) {
-	// Return empty distributions for now
-	return nil, nil, nil
-}
-
-// GetPrefixDistributionContext mock implementation with context support
-func (m *mockStore) GetPrefixDistributionContext(_ context.Context) (
-	[]database.PrefixDistribution, []database.PrefixDistribution, error,
-) {
-	return m.GetPrefixDistribution()
-}
-
-// GetLiveRouteCounts mock implementation
-func (m *mockStore) GetLiveRouteCounts() (int, int, error) {
-	// Return mock counts
-	return m.RouteCount / 2, m.RouteCount / 2, nil
-}
-
-// GetLiveRouteCountsContext mock implementation with context support
-func (m *mockStore) GetLiveRouteCountsContext(_ context.Context) (int, int, error) {
-	return m.GetLiveRouteCounts()
-}
-
-// GetASDetails mock implementation
-func (m *mockStore) GetASDetails(asn int) (*database.ASN, []database.LiveRoute, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	// Check if ASN exists
-	if asnInfo, exists := m.ASNs[asn]; exists {
-		// Return empty prefixes for now
-		return asnInfo, []database.LiveRoute{}, nil
-	}
-
-	return nil, nil, database.ErrNoRoute
-}
-
-// GetASDetailsContext mock implementation with context support
-func (m *mockStore) GetASDetailsContext(
-	_ context.Context, asn int,
-) (*database.ASN, []database.LiveRoute, error) {
-	return m.GetASDetails(asn)
-}
-
 // GetAS returns the stored AS, or only its number when none is stored.
 func (m *mockStore) GetAS(_ context.Context, asn int) (*database.ASN, bool, error) {
 	m.mu.Lock()
@@ -292,35 +237,6 @@ func (m *mockStore) GetPrefixFirstSeen(
 	return time.Time{}, nil
 }
 
-// GetPrefixDetails mock implementation
-func (m *mockStore) GetPrefixDetails(
-	_ string,
-) ([]database.LiveRoute, map[int]database.ASN, error) {
-	// Return empty routes for now
-	return []database.LiveRoute{}, map[int]database.ASN{}, nil
-}
-
-// GetPrefixDetailsContext mock implementation with context support
-func (m *mockStore) GetPrefixDetailsContext(
-	_ context.Context, prefix string,
-) ([]database.LiveRoute, map[int]database.ASN, error) {
-	return m.GetPrefixDetails(prefix)
-}
-
-func (m *mockStore) GetRandomPrefixesByLength(
-	_, _, _ int,
-) ([]database.LiveRoute, error) {
-	// Return empty routes for now
-	return []database.LiveRoute{}, nil
-}
-
-// GetRandomPrefixesByLengthContext mock implementation with context support
-func (m *mockStore) GetRandomPrefixesByLengthContext(
-	_ context.Context, maskLength, ipVersion, limit int,
-) ([]database.LiveRoute, error) {
-	return m.GetRandomPrefixesByLength(maskLength, ipVersion, limit)
-}
-
 // GetASPeers mock implementation
 func (m *mockStore) GetASPeers(_ int) ([]database.ASPeer, error) {
 	// Return empty peers for now
@@ -334,32 +250,6 @@ func (m *mockStore) GetASPeersContext(
 	return m.GetASPeers(asn)
 }
 
-// GetIPInfo mock implementation
-func (m *mockStore) GetIPInfo(ip string) (*database.IPInfo, error) {
-	return m.GetIPInfoContext(context.Background(), ip)
-}
-
-// GetIPInfoContext mock implementation with context support
-func (m *mockStore) GetIPInfoContext(
-	_ context.Context, ip string,
-) (*database.IPInfo, error) {
-	now := time.Now()
-
-	return &database.IPInfo{
-		IP:          ip,
-		Netblock:    "8.8.8.0/24",
-		MaskLength:  24,
-		IPVersion:   4,
-		NumPeers:    3,
-		ASN:         15169,
-		Handle:      "GOOGLE",
-		Description: "Google LLC",
-		CountryCode: "US",
-		FirstSeen:   now.Add(-24 * time.Hour),
-		LastSeen:    now,
-	}, nil
-}
-
 // GetNextStaleASN mock implementation
 func (m *mockStore) GetNextStaleASN(_ context.Context, _ time.Duration) (int, error) {
 	return 0, database.ErrNoStaleASN
@@ -387,46 +277,6 @@ func (m *mockStore) GetWHOISStats(
 	}, nil
 }
 
-// UpsertLiveRouteBatch mock implementation
-func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	for _, route := range routes {
-		// Track prefix
-		if _, exists := m.Prefixes[route.Prefix]; !exists {
-			m.Prefixes[route.Prefix] = &database.Prefix{
-				Prefix:    route.Prefix,
-				IPVersion: route.IPVersion,
-				FirstSeen: route.LastUpdated,
-			}
-			m.PrefixCount++
-
-			if route.IPVersion == 4 {
-				m.IPv4Prefixes++
-			} else {
-				m.IPv6Prefixes++
-			}
-		}
-
-		m.RouteCount++
-	}
-
-	return nil
-}
-
-// DeleteLiveRouteBatch mock implementation
-func (m *mockStore) DeleteLiveRouteBatch(_ []database.LiveRouteDeletion) error {
-	// Simple mock - just return nil
-	return nil
-}
-
-// DeleteLiveRoutesOfPeer mock implementation
-func (m *mockStore) DeleteLiveRoutesOfPeer(_ string) (int, error) {
-	// Simple mock - deletes nothing
-	return 0, nil
-}
-
 // WriteRouteChanges records the prefixes the changes leave some peer carrying,
 // whatever their origin AS.
 func (m *mockStore) WriteRouteChanges(

internal/routewatch/ashandler_internal_test.go

@@ -29,9 +29,9 @@ func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
 	h.HandleMessage(second)
 	h.flushBatch()
 
-	asn, _, err := db.GetASDetails(testASNA)
+	asn, _, err := db.GetAS(t.Context(), testASNA)
 	if err != nil {
-		t.Fatalf("GetASDetails(%d): %v", testASNA, err)
+		t.Fatalf("GetAS(%d): %v", testASNA, err)
 	}
 
 	if !asn.FirstSeen.Equal(first.ParsedTimestamp) {

internal/routewatch/batchwrite_internal_test.go

@@ -81,9 +81,9 @@ func TestFlushLoopWritesOnEveryTick(t *testing.T) {
 }
 
 // blockingStore is a mockStore whose batch writes for the AS, peer and prefix
-// handlers, delete of a peer's routes and peering write wait until the test
-// calls releaseWrites. writeStarted is closed when the first of them starts. It
-// records what each batch write was given.
+// handlers and peering write wait until the test calls releaseWrites.
+// writeStarted is closed when the first of them starts. It records what each
+// batch write was given.
 type blockingStore struct {
 	*mockStore
 
@@ -109,13 +109,13 @@ func newBlockingStore() *blockingStore {
 	}
 }
 
-func (s *blockingStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
+func (s *blockingStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	s.waitForRelease()
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	for _, route := range routes {
-		s.prefixes[route.Prefix] = true
+	for prefix := range prefixes {
+		s.prefixes[prefix] = true
 	}
 
 	return nil
@@ -145,12 +145,6 @@ func (s *blockingStore) UpdatePeerBatch(peers map[string]database.PeerUpdate) er
 	return nil
 }
 
-func (s *blockingStore) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
-	s.waitForRelease()
-
-	return s.mockStore.DeleteLiveRoutesOfPeer(peerIP)
-}
-
 func (s *blockingStore) RecordPeering(
 	asA, asB int, seen database.SeenTimes,
 ) error {
@@ -176,7 +170,6 @@ func (s *blockingStore) releaseWrites() {
 type stopLog struct {
 	Msg          string `json:"msg"`
 	Updates      int    `json:"updates"`
-	PeersDown    int    `json:"peers_down"`
 	TableChanges int    `json:"table_changes"`
 	Paths        int    `json:"paths"`
 }
@@ -374,10 +367,10 @@ 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. The flush
-// timer must not fire. A flush it started before the fill would be the one
-// held, and it would not have written the fill's change.
+// routing table's changes: the flush whose prefix write is held wrote those of
+// the fill, so Stop must log the one of the extra prefix as not written. The
+// flush timer must not fire. A flush it started before the fill would be the
+// one held, and it would not have written the fill's change.
 func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
@@ -389,10 +382,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	h := newPrefixHandler(store, newErrorLogger(&logs),
 		routetable.New(routetable.Params{}), longerThanAnyTest)
 	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.
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
-		[]*ristypes.RISMessage{sessionDown(), prefixAnnouncement(fill)},
+		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
 		prefixAnnouncement([]string{extraPrefix}))
 
 	store.mu.Lock()
@@ -403,11 +394,10 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	}
 
 	logged := lastLogged(t, &logs)
-	if logged.Updates != prefixBatchSize+1 || logged.PeersDown != 1 ||
-		logged.TableChanges != 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)
+	if logged.Updates != prefixBatchSize+1 || logged.TableChanges != 1 {
+		t.Errorf("Stop logged %d updates and %d table changes not written, "+
+			"want %d and 1", logged.Updates, logged.TableChanges,
+			prefixBatchSize+1)
 	}
 }
 

internal/routewatch/prefixhandler.go

@@ -2,10 +2,8 @@ package routewatch
 
 import (
 	"context"
-	"net"
 	"net/netip"
 	"slices"
-	"strings"
 	"sync"
 	"time"
 
@@ -28,11 +26,9 @@ const (
 	prefixBatchSize = 25000
 
 	// prefixBatchMaxSize is the most prefix updates the batch holds while the
-	// flush goroutine writes the previous one: ten batches, about 60 MiB
+	// flush goroutine writes the previous one: ten batches, about 15 MiB
 	// (README "Memory"). At this size HandleMessage waits until the flush
-	// goroutine takes the batch. On top of these, the handler holds at most
-	// prefixBatchSize updates, about 6 MiB, that a flush kept for the peers
-	// whose route delete failed (held).
+	// goroutine takes the batch.
 	prefixBatchMaxSize = 250000
 
 	// prefixBatchTimeout is the maximum time to wait before flushing a batch
@@ -48,47 +44,31 @@ const (
 	// 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
 )
 
-// PrefixHandler tracks BGP prefixes and maintains a live routing table in the
-// 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.
+// PrefixHandler keeps the routing table in memory current from the feed. It
+// applies each update to the table as it handles it. An announcement gives
+// the peer a route to the prefix. A withdrawal removes the peer's route, as
+// does an announcement whose AS path is empty or holds AS 0 anywhere. A peer
+// whose BGP session goes down leaves every prefix.
 //
-// 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.
+// Each flush writes the table's changes to the database. It also stores each
+// prefix the updates name with the time it was first seen.
 type PrefixHandler struct {
 	db      database.Store
 	logger  *logger.Logger
 	metrics *metrics.Tracker
 	table   *routetable.Table
 
-	// Batching. mu guards batch, held, peersDown, writing, deleting and
-	// writingTableChanges only: the flush goroutine takes them under mu and
-	// writes them after releasing mu, so HandleMessage waits for a database
-	// write only while batch is at prefixBatchMaxSize.
+	// Batching. mu guards batch, writing and writingTableChanges only: the
+	// flush goroutine takes them under mu and writes them after releasing
+	// mu, so HandleMessage waits for a database 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
-	// last flush sent after their session went down, for the next flush to
-	// write after it retries the delete. They are not in batch, so they do
-	// not count toward a full batch: a delete that keeps failing is retried
-	// at the pace of the flush timer.
-	held []prefixUpdate
-	// peersDown holds, each once, the peers whose BGP session went down
-	// since the batch was taken, and those whose route delete failed in the
-	// last flush.
-	peersDown []string
+	batch []seenPrefix
 	// writing is how many updates the flush goroutine took and has not
 	// finished writing.
 	writing int
-	// 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
@@ -102,6 +82,15 @@ type PrefixHandler struct {
 	done   chan struct{} // closed when the flush goroutine returns
 }
 
+// seenPrefix is what the batch keeps of an update: the prefix it names and its
+// time. A flush needs only these to store each new prefix's first-seen time.
+type seenPrefix struct {
+	prefix    string
+	timestamp time.Time
+}
+
+// prefixUpdate is one prefix of an update message, as applyToTable applies it
+// to the routing table.
 type prefixUpdate struct {
 	prefix      string
 	originASN   int
@@ -144,7 +133,7 @@ func newPrefixHandler(
 		db:          db,
 		logger:      logger,
 		table:       table,
-		batch:       make([]prefixUpdate, 0, prefixBatchSize),
+		batch:       make([]seenPrefix, 0, prefixBatchSize),
 		batchFullCh: make(chan struct{}, 1),
 		stopCh:      make(chan context.Context, 1),
 		done:        make(chan struct{}),
@@ -181,7 +170,7 @@ func (h *PrefixHandler) QueueCapacity() int {
 func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	if msg.Type == ristypes.MessageTypeState {
 		if msg.State == ristypes.StateDown {
-			h.peerSessionDown(msg.Peer)
+			h.table.PeerDown(msg.Peer)
 		}
 
 		return
@@ -206,12 +195,11 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 		h.batchTaken.Wait()
 	}
 
-	// Withdrawals are queued, and applied to the routing table, before
-	// announcements, because the batch keeps each peer's last update for a
-	// prefix: a message that lists a prefix as both withdrawn and announced
-	// keeps the route (RFC 4271 section 4.3). A withdrawal names only the
-	// prefix and peer: the message's AS path, and so its origin AS, belongs
-	// to the announced prefixes.
+	// Withdrawals are applied to the routing table before announcements, so
+	// a message that lists a prefix as both withdrawn and announced keeps the
+	// route (RFC 4271 section 4.3). A withdrawal names only the prefix and
+	// peer: the message's AS path, and so its origin AS, belongs to the
+	// announced prefixes.
 	for _, prefix := range msg.Withdrawals {
 		update := prefixUpdate{
 			prefix:      canonicalPrefix(prefix),
@@ -219,7 +207,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 			messageType: "withdrawal",
 			timestamp:   timestamp,
 		}
-		h.batch = append(h.batch, update)
+		h.batch = append(h.batch, seenPrefix{update.prefix, update.timestamp})
 		h.applyToTable(update)
 
 		// Record withdrawal in metrics
@@ -239,7 +227,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 				path:        path,
 				endsInSet:   len(msg.Path.Set) > 0,
 			}
-			h.batch = append(h.batch, update)
+			h.batch = append(h.batch, seenPrefix{update.prefix, update.timestamp})
 			h.applyToTable(update)
 
 			// Record announcement in metrics
@@ -261,14 +249,12 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 }
 
 // Stop stops the flush goroutine, which makes one last flush before it returns,
-// and waits for it until ctx ends. That flush writes the updates and peer downs
-// the handler holds, and at most maxTableChangesPerWrite of the routing table's
-// changes. Once ctx has ended the flush goroutine skips that last flush, and a
-// write still running is lost when the process exits. A peer whose route delete
-// 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.
+// and waits for it until ctx ends. That flush writes at most
+// maxTableChangesPerWrite of the routing table's changes, and the prefixes of
+// the updates the handler holds. Once ctx has ended the flush goroutine skips
+// that last flush, and a write still running is lost when the process exits. A
+// failed write of the table's changes is not retried either. Stop logs how many
+// updates and how many of the table's changes are not written.
 func (h *PrefixHandler) Stop(ctx context.Context) {
 	h.stopCh <- ctx
 
@@ -278,176 +264,96 @@ 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)
+	unwritten := h.writing + len(h.batch)
 	tableChanges := h.writingTableChanges + h.table.ChangeCount()
 	h.mu.Unlock()
 
-	if unwritten == 0 && peersDown == 0 && tableChanges == 0 {
+	if unwritten == 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,
-			"table_changes", tableChanges)
+			"its updates", "updates", unwritten, "table_changes", tableChanges)
 
 		return
 	}
 
 	h.logger.Error("The prefix handler stopped before it wrote everything",
-		"updates", unwritten, "peers_down", peersDown,
-		"table_changes", tableChanges)
+		"updates", unwritten, "table_changes", tableChanges)
 }
 
-// peerSessionDown notes that the peer's BGP session went down, so that the next
-// flush deletes all of the peer's live routes before it writes the batch. The
-// 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. 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)
-
-	if !slices.Contains(h.peersDown, peer) {
-		h.peersDown = append(h.peersDown, peer)
-	}
-}
-
-// takeBatch returns the pending updates, the held ones first, and the peers in
-// peersDown, and leaves an empty batch, no held updates and no peers in their
+// takeBatch returns the pending updates and leaves an empty batch in their
 // place for HandleMessage to fill. The updates count in writing until
-// flushBatch has written them, and the peers in deleting until their routes
-// are deleted.
-func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
+// flushBatch has written them.
+func (h *PrefixHandler) takeBatch() []seenPrefix {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	if len(h.held) == 0 && len(h.batch) == 0 && len(h.peersDown) == 0 {
-		return nil, nil
-	}
-
-	batch, peersDown := h.batch, h.peersDown
-	if len(h.held) > 0 {
-		batch = slices.Concat(h.held, h.batch)
+	if len(h.batch) == 0 {
+		return nil
 	}
 
+	batch := h.batch
 	h.writing = len(batch)
-	h.deleting = len(peersDown)
-	h.held = nil
-	h.batch = make([]prefixUpdate, 0, prefixBatchSize)
-	h.peersDown = nil
+	h.batch = make([]seenPrefix, 0, prefixBatchSize)
 	h.batchTaken.Broadcast()
 
-	return batch, peersDown
+	return batch
 }
 
 // flushBatch writes the routing table's changes, then takes the pending batch
-// and writes it to the database, without holding the mutex during the writes.
+// and stores the prefixes it names, without holding the mutex during the
+// writes.
 func (h *PrefixHandler) flushBatch() {
 	h.writeTableChanges()
 
-	batch, peersDown := h.takeBatch()
-	if len(batch) == 0 && len(peersDown) == 0 {
+	batch := h.takeBatch()
+	if len(batch) == 0 {
 		return
 	}
 
 	startTime := time.Now()
-	batchSize := len(batch)
-
-	// The routes of a peer whose session went down are deleted first: the
-	// batch holds only the updates the peer sent after the down. A peer whose
-	// delete failed is kept, with those updates, for the next flush to retry.
-	failedPeers := h.deleteLiveRoutesOfPeers(peersDown)
-	if len(failedPeers) > 0 {
-		batch = h.keepForNextFlush(batch, failedPeers)
-	}
-
-	// For each prefix and peer, keep only the last update the feed delivered.
-	// Updates from different peers for the same prefix are all kept. Arrival
-	// order decides, not the timestamp: timestamps are whole seconds, so
-	// updates within one second carry the same one.
-	type prefixAndPeer struct{ prefix, peer string }
-
-	lastUpdates := make(map[prefixAndPeer]prefixUpdate)
 
 	// Each prefix the batch names, announced or withdrawn, with the earliest
-	// timestamp among all of its updates, not only each peer's last one: the
-	// prefix tables store only first_seen, written when the prefix is new and
-	// never changed after.
+	// timestamp among its updates: the prefix tables store only first_seen,
+	// written when the prefix is new and never changed after.
 	prefixesToUpdate := make(map[string]time.Time)
 
 	for _, update := range batch {
-		lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
-
 		seen, exists := prefixesToUpdate[update.prefix]
 		if !exists || update.timestamp.Before(seen) {
 			prefixesToUpdate[update.prefix] = update.timestamp
 		}
 	}
 
-	// Collect routes to upsert and delete
-	var (
-		routesToUpsert []*database.LiveRoute
-		routesToDelete []database.LiveRouteDeletion
-	)
-
-	for _, update := range lastUpdates {
-		if update.announcesRoute() {
-			// Create live route for batch upsert
-			route := h.createLiveRoute(update)
-			if route != nil {
-				routesToUpsert = append(routesToUpsert, route)
-			}
-		} else {
-			// Parse CIDR to get IP version
-			_, ipVersion, err := parseCIDR(update.prefix)
-			if err != nil {
-				h.logger.Error("Failed to parse CIDR for route deletion",
-					"prefix", update.prefix, "error", err)
-
-				continue
-			}
-
-			// Create deletion record for batch delete
-			routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
-				Prefix:    update.prefix,
-				PeerIP:    update.peer,
-				IPVersion: ipVersion,
-			})
-		}
+	err := h.db.UpdatePrefixesBatch(prefixesToUpdate)
+	if err != nil {
+		h.logger.Error("Failed to update prefix batch",
+			"error", err, "count", len(prefixesToUpdate))
 	}
 
-	successCount := h.writeBatch(routesToUpsert, routesToDelete, prefixesToUpdate)
-
 	h.mu.Lock()
 	h.writing = 0
 	h.mu.Unlock()
 
-	elapsed := time.Since(startTime)
 	h.logger.Debug("Flushed prefix batch",
-		"batch_size", batchSize,
-		"peers_down", len(peersDown),
+		"batch_size", len(batch),
 		"unique_prefixes", len(prefixesToUpdate),
-		"success", successCount,
-		"duration_ms", elapsed.Milliseconds(),
+		"duration_ms", time.Since(startTime).Milliseconds(),
 	)
 }
 
 // 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.
+// prefix that does not parse is logged and left out. Each announcement counts
+// in the IPv4 or IPv6 route update rate.
 func (h *PrefixHandler) applyToTable(update prefixUpdate) {
 	prefix, err := netip.ParsePrefix(update.prefix)
 	if err != nil {
+		h.logger.Error("Failed to parse prefix",
+			"prefix", update.prefix, "error", err)
+
 		return
 	}
 
@@ -457,6 +363,14 @@ func (h *PrefixHandler) applyToTable(update prefixUpdate) {
 		return
 	}
 
+	if h.metrics != nil {
+		if prefix.Addr().Is4() {
+			h.metrics.RecordIPv4Update()
+		} else {
+			h.metrics.RecordIPv6Update()
+		}
+	}
+
 	h.table.Announce(prefix, update.originASN, routetable.Route{
 		Peer:   update.peer,
 		ASPath: update.path,
@@ -495,147 +409,11 @@ func (h *PrefixHandler) writeTableChanges() {
 	}
 }
 
-// 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
-// routes is logged. A peer whose delete succeeded no longer counts in deleting;
-// a failed one counts until keepForNextFlush puts it back in peersDown.
-func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) []string {
-	var failed []string
-
-	for _, peer := range peers {
-		deleted, err := h.db.DeleteLiveRoutesOfPeer(peer)
-		if err != nil {
-			h.logger.Error(
-				"Failed to delete the live routes of a peer whose session "+
-					"went down; the next flush retries",
-				"peer", peer, "routes_deleted", deleted, "error", err)
-
-			failed = append(failed, peer)
-
-			continue
-		}
-
-		h.mu.Lock()
-		h.deleting--
-		h.mu.Unlock()
-
-		if deleted > 0 {
-			h.logger.Info(
-				"Deleted the live routes of a peer whose session went down",
-				"peer", peer, "routes", deleted)
-		}
-	}
-
-	return failed
-}
-
-// keepForNextFlush puts the peers whose route delete failed back in peersDown,
-// and their updates from batch in held, so that the next flush retries each
-// delete before it writes the peer's updates; written now, they would be
-// deleted by the retry. The updates of a peer whose session went down again
-// since the batch was taken came before that down, so they are dropped, as
-// peerSessionDown drops them, and the peer is already in peersDown. It returns
-// the updates of the other peers, for this flush to write.
-//
-// While a delete keeps failing, the held updates would grow with every flush,
-// so only the last prefixBatchSize of them are kept and the older ones are
-// dropped. This never leaves in place a route the peer withdrew: a prefix with
-// a kept update keeps its last one, and the retry deletes all of the peer's
-// routes, so a prefix whose updates were all dropped has no route from the peer
-// until the peer announces it again.
-func (h *PrefixHandler) keepForNextFlush(
-	batch []prefixUpdate, failedPeers []string,
-) []prefixUpdate {
-	var kept, toWrite []prefixUpdate
-
-	for _, update := range batch {
-		if slices.Contains(failedPeers, update.peer) {
-			kept = append(kept, update)
-		} else {
-			toWrite = append(toWrite, update)
-		}
-	}
-
-	h.mu.Lock()
-	defer h.mu.Unlock()
-
-	kept = slices.DeleteFunc(kept, func(update prefixUpdate) bool {
-		return slices.Contains(h.peersDown, update.peer)
-	})
-
-	if len(kept) > prefixBatchSize {
-		h.logger.Error(
-			"Dropped the oldest updates of peers whose route delete failed",
-			"peers", failedPeers, "dropped", len(kept)-prefixBatchSize)
-
-		// A copy, so that the dropped updates are freed.
-		kept = slices.Clone(kept[len(kept)-prefixBatchSize:])
-	}
-
-	// takeBatch left held empty, and only this flush fills it. The kept
-	// updates now count in held, not in writing, and the failed peers in
-	// peersDown, not in deleting.
-	h.held = kept
-	h.writing = len(toWrite)
-	h.deleting = 0
-
-	for _, peer := range failedPeers {
-		if !slices.Contains(h.peersDown, peer) {
-			h.peersDown = append(h.peersDown, peer)
-		}
-	}
-
-	return toWrite
-}
-
-// writeBatch writes the collected route upserts, route deletions and prefix
-// updates to the database and returns how many routes it wrote.
-func (h *PrefixHandler) writeBatch(
-	routesToUpsert []*database.LiveRoute,
-	routesToDelete []database.LiveRouteDeletion,
-	prefixesToUpdate map[string]time.Time,
-) int {
-	successCount := 0
-
-	if len(routesToUpsert) > 0 {
-		err := h.db.UpsertLiveRouteBatch(routesToUpsert)
-		if err != nil {
-			h.logger.Error("Failed to upsert route batch",
-				"error", err, "count", len(routesToUpsert))
-		} else {
-			successCount += len(routesToUpsert)
-		}
-	}
-
-	if len(routesToDelete) > 0 {
-		err := h.db.DeleteLiveRouteBatch(routesToDelete)
-		if err != nil {
-			h.logger.Error("Failed to delete route batch",
-				"error", err, "count", len(routesToDelete))
-		} else {
-			successCount += len(routesToDelete)
-		}
-	}
-
-	// Update prefix tables
-	if len(prefixesToUpdate) > 0 {
-		err := h.db.UpdatePrefixesBatch(prefixesToUpdate)
-		if err != nil {
-			h.logger.Error("Failed to update prefix batch",
-				"error", err, "count", len(prefixesToUpdate))
-		}
-	}
-
-	return successCount
-}
-
 // canonicalPrefix returns prefix in the text form net/netip prints, the form
-// the IP lookup builds when it looks a prefix up. The feed sends IPv6
+// the prefix tables store and the IP lookup reads them in. The feed sends IPv6
 // announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
-// ("2001:db8:0:0:0:0:0:0/32"); stored as received, a withdrawal would not match
-// the route its announcement stored. A prefix that does not parse is returned
-// unchanged, and the batch flush reports it.
+// ("2001:db8:0:0:0:0:0:0/32"). A prefix that does not parse is returned
+// unchanged, and applyToTable logs it.
 func canonicalPrefix(prefix string) string {
 	p, err := netip.ParsePrefix(prefix)
 	if err != nil {
@@ -644,55 +422,3 @@ func canonicalPrefix(prefix string) string {
 
 	return p.Masked().String()
 }
-
-// parseCIDR returns the mask length and the IP version of a prefix string
-func parseCIDR(prefix string) (int, int, error) {
-	_, ipNet, err := net.ParseCIDR(prefix)
-	if err != nil {
-		return 0, 0, err
-	}
-
-	ones, _ := ipNet.Mask.Size()
-	if strings.Contains(prefix, ":") {
-		return ones, ipv6Version, nil
-	}
-
-	return ones, ipv4Version, nil
-}
-
-// createLiveRoute creates a LiveRoute from a prefix update
-func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute {
-	// Parse CIDR to get mask length
-	maskLength, ipVersion, err := parseCIDR(update.prefix)
-	if err != nil {
-		h.logger.Error("Failed to parse CIDR",
-			"prefix", update.prefix,
-			"error", err,
-		)
-
-		return nil
-	}
-
-	// Track route update metrics
-	if h.metrics != nil {
-		if ipVersion == ipv4Version {
-			h.metrics.RecordIPv4Update()
-		} else {
-			h.metrics.RecordIPv6Update()
-		}
-	}
-
-	// Create live route record
-	liveRoute := &database.LiveRoute{
-		Prefix:      update.prefix,
-		MaskLength:  maskLength,
-		IPVersion:   ipVersion,
-		OriginASN:   update.originASN,
-		PeerIP:      update.peer,
-		ASPath:      update.path,
-		NextHop:     update.peer, // Using peer as next hop
-		LastUpdated: update.timestamp,
-	}
-
-	return liveRoute
-}

internal/routewatch/prefixhandler_table_internal_test.go

@@ -8,6 +8,7 @@ import (
 	"net/netip"
 	"reflect"
 	"slices"
+	"strings"
 	"testing"
 	"time"
 
@@ -84,13 +85,25 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
 				testV6Prefix: {},
 			},
 		},
+		{
+			name: "an announcement after the peer's session went down",
+			messages: []*ristypes.RISMessage{
+				peerAnnouncement(testPeerIP, testV4Prefix),
+				sessionDown(),
+				peerAnnouncement(testPeerIP, otherV4Prefix),
+			},
+			want: map[string]map[int][]string{
+				testV4Prefix:  {},
+				otherV4Prefix: {testASNB: {testPeerIP}},
+			},
+		},
 	})
 }
 
 // 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.
+// is empty or 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()
 
@@ -110,6 +123,14 @@ func TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable(t *testing.T)
 			},
 			want: map[string]map[int][]string{testV4Prefix: {}},
 		},
+		{
+			name: "an empty path",
+			messages: []*ristypes.RISMessage{
+				peerAnnouncement(testPeerIP, testV4Prefix),
+				announceWithPath(ristypes.ASPath{}),
+			},
+			want: map[string]map[int][]string{testV4Prefix: {}},
+		},
 		{
 			name: "a path with AS 0 before its end",
 			messages: []*ristypes.RISMessage{
@@ -128,6 +149,55 @@ func TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable(t *testing.T)
 	})
 }
 
+// TestPrefixHandlerKeepsThePathOfAPathEndingInASet announces a prefix with AS
+// paths that end in an AS set. Any member of a set of several may be the
+// origin, so the routing table keeps the route with origin 0 (none); a set of
+// one member counts as that AS. The kept AS path lists the set's members after
+// the rest of the path.
+func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		path       []any
+		wantOrigin int
+		wantPath   []int
+	}{
+		{
+			name:     "a set of several members",
+			path:     []any{testASNA, testASNB, []int{testASNC, testASND}},
+			wantPath: []int{testASNA, testASNB, testASNC, testASND},
+		},
+		{
+			name:       "a set of one member",
+			path:       []any{testASNA, testASNB, []int{testASNC}},
+			wantOrigin: testASNC,
+			wantPath:   []int{testASNA, testASNB, testASNC},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			msg := peerAnnouncement(testPeerIP, testV4Prefix)
+			msg.Path = decodePath(t, tt.path...)
+
+			h := prefixHandlerWithoutTimer(newMockStore())
+			h.HandleMessage(msg)
+
+			origins := h.table.Origins(netip.MustParsePrefix(testV4Prefix))
+			if len(origins) != 1 || origins[0].ASN != tt.wantOrigin ||
+				len(origins[0].Routes) != 1 ||
+				!slices.Equal(origins[0].Routes[0].ASPath, tt.wantPath) {
+				t.Errorf("Origins(%s) = %+v, want one route with origin AS%d "+
+					"and AS path %v", testV4Prefix, origins, tt.wantOrigin,
+					tt.wantPath)
+			}
+		})
+	}
+}
+
 // 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) {
@@ -343,8 +413,9 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
 // TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler,
 // with no deadline, whose routing table holds more changed pairs than one write
 // takes. Its last flush writes one write's worth, so Stop must log the pairs
-// that no write took. The flush timer must not fire, or its writes could leave
-// no pairs for Stop to log.
+// that no write took, and must not say that the stop deadline passed. The
+// flush timer must not fire, or its writes could leave no pairs for Stop to
+// log.
 func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	t.Parallel()
 
@@ -364,7 +435,13 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 		written += n
 	}
 
-	if logged := lastLogged(t, &logs); logged.TableChanges != pairs-written {
+	logged := lastLogged(t, &logs)
+	if strings.Contains(logged.Msg, "deadline") {
+		t.Errorf("Stop logged %q although its deadline had not passed",
+			logged.Msg)
+	}
+
+	if logged.TableChanges != pairs-written {
 		t.Errorf("Stop logged %d table changes not written, want %d",
 			logged.TableChanges, pairs-written)
 	}