Remove a peer's live routes when its session goes down (closes #56)

RIS sends no withdrawals for the routes of a peer whose BGP session goes
down, so routewatch kept them.

The streamer now hands STATE messages to the prefix handler. They are never
dropped for backpressure: one that finds the queue full is held and queued
first once there is room, and the handler's other messages are dropped
meanwhile, so none overtakes it. Nothing waits. On a down, the prefix
handler drops the peer's pending updates and its next flush deletes all the
peer's live routes before the batch's other writes, keeping the in-memory
counts exact. The live route tables are unique on peer and prefix, which
also finds a peer's routes.

Model: opus-5-5

README.md

@@ -128,7 +128,10 @@ routewatch/
 4. **Handler Details**
    - **ASHandler**: Tracks all ASNs seen in AS paths
    - **PeerHandler**: Records BGP peer information
-   - **PrefixHandler**: Maintains live routing table (upserts on announcement, deletes on withdrawal)
+   - **PrefixHandler**: Maintains live routing table (upserts on announcement, deletes on withdrawal,
+     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.
    - **PeeringHandler**: Extracts AS peering relationships from AS paths
 
 5. **HTTP Request Flow**
@@ -232,7 +235,9 @@ What happens at each limit:
   out-of-memory error, and the handler logs it and drops that batch. The process
   keeps running.
 - Handler queues: each of the four handler queues holds at most 20,000 messages.
-  When a queue fills, the streamer drops messages instead of blocking.
+  When a queue fills, the streamer drops messages instead of blocking, except a
+  `STATE` message, which it holds, in feed order, until there is room: a few
+  hundred bytes each, about ten a second.
 - Handler batches: while the AS, peer or prefix handler writes a batch to the
   database, the updates that arrive collect in its next batch, which stops at
   ten times the batch size: 300,000 updates for the AS handler

internal/database/counts_test.go

@@ -349,6 +349,89 @@ func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
 	return db.DeleteLiveRouteBatch(deletions)
 }
 
+// TestDeleteLiveRoutesOfPeer gives two peers IPv4 and IPv6 live routes, some
+// for the same prefixes, and deletes all of one peer's routes. The other peer's
+// routes stay, and the route counts and prefix distribution held in memory
+// equal what is read from the route tables.
+func TestDeleteLiveRoutesOfPeer(t *testing.T) {
+	const (
+		peerIP      = "192.0.2.1"
+		otherPeerIP = "192.0.2.2"
+		sharedV4    = "198.51.100.0/24"
+		sharedV6    = "2001:db8::/32"
+	)
+
+	db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+	ofPeer := func(peer string, route *LiveRoute) *LiveRoute {
+		route.PeerIP = peer
+
+		return route
+	}
+	onlyV6 := mkV6Route("2001:db8:1::/48", 64503, ts)
+	onlyV6.MaskLength = 48
+	if err := db.UpsertLiveRouteBatch([]*LiveRoute{
+		ofPeer(peerIP, mkV4Route(t, sharedV4, 64500, ts)),
+		ofPeer(otherPeerIP, mkV4Route(t, sharedV4, 64500, ts)),
+		ofPeer(peerIP, mkV4Route(t, "203.0.113.0/24", 64501, ts)),
+		ofPeer(peerIP, mkV6Route(sharedV6, 64502, ts)),
+		ofPeer(otherPeerIP, mkV6Route(sharedV6, 64502, ts)),
+		ofPeer(peerIP, onlyV6),
+	}); err != nil {
+		t.Fatalf("UpsertLiveRouteBatch: %v", err)
+	}
+
+	// peerIP has two IPv4 and two IPv6 routes; a second delete finds none.
+	for _, want := range []int{4, 0} {
+		deleted, err := db.DeleteLiveRoutesOfPeer(peerIP)
+		if err != nil {
+			t.Fatalf("DeleteLiveRoutesOfPeer: %v", err)
+		}
+		if deleted != want {
+			t.Errorf("DeleteLiveRoutesOfPeer deleted %d routes, want %d", deleted, want)
+		}
+	}
+
+	ctx := context.Background()
+	for _, prefix := range []string{sharedV4, sharedV6} {
+		routes, _, err := db.GetPrefixDetailsContext(ctx, prefix)
+		if err != nil {
+			t.Fatalf("GetPrefixDetailsContext(%s): %v", prefix, err)
+		}
+		if len(routes) != 1 || routes[0].PeerIP != otherPeerIP {
+			t.Errorf("live routes for %s = %+v, want only the one from %s", prefix, routes, otherPeerIP)
+		}
+	}
+
+	stats, err := db.GetStatsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetStatsContext: %v", err)
+	}
+	tableV4, tableV6, err := db.GetLiveRouteCountsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetLiveRouteCountsContext: %v", err)
+	}
+	if stats.IPv4Routes != 1 || tableV4 != 1 || stats.IPv6Routes != 1 || tableV6 != 1 {
+		t.Errorf("live routes (IPv4, IPv6): (%d, %d) in memory, (%d, %d) in the tables, want (1, 1)",
+			stats.IPv4Routes, stats.IPv6Routes, tableV4, tableV6)
+	}
+	queryV4, queryV6, err := db.GetPrefixDistributionContext(ctx)
+	if err != nil {
+		t.Fatalf("GetPrefixDistributionContext: %v", err)
+	}
+	wantV4 := []PrefixDistribution{{MaskLength: 24, Count: 1}}
+	wantV6 := []PrefixDistribution{{MaskLength: 32, Count: 1}}
+	assertDistribution(t, "IPv4 distribution", stats.IPv4PrefixDistribution, wantV4)
+	assertDistribution(t, "IPv6 distribution", stats.IPv6PrefixDistribution, wantV6)
+	assertDistribution(t, "IPv4 distribution query", queryV4, wantV4)
+	assertDistribution(t, "IPv6 distribution query", queryV6, wantV6)
+}
+
 // TestStatsRouteTimestamps checks the oldest/newest route timestamps are read
 // from the right rows across both tables and parse into time.Time. The old
 // MIN/MAX union query read its result into *time.Time, which the driver could

internal/database/database.go

@@ -542,6 +542,91 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 	return nil
 }
 
+// Deleting all live routes of a peer. The first query of each table selects the
+// mask length of each of the peer's routes whose prefix no other peer has a
+// live route for: the delete takes those prefixes out of the prefix
+// distribution.
+const (
+	peerOnlyRouteMaskLengthsV4SQL = `SELECT r.mask_length FROM live_routes_v4 AS r WHERE r.peer_ip = ?
+		AND NOT EXISTS (SELECT 1 FROM live_routes_v4 AS o WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
+	deletePeerLiveRoutesV4SQL = `DELETE FROM live_routes_v4 WHERE peer_ip = ?`
+
+	peerOnlyRouteMaskLengthsV6SQL = `SELECT r.mask_length FROM live_routes_v6 AS r WHERE r.peer_ip = ?
+		AND NOT EXISTS (SELECT 1 FROM live_routes_v6 AS o WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
+	deletePeerLiveRoutesV6SQL = `DELETE FROM live_routes_v6 WHERE peer_ip = ?`
+)
+
+// deletePeerRoutes deletes all live routes of peerIP from one table, given that
+// table's two statements above. It returns how many routes it deleted and the
+// mask length of each prefix it left with no live route.
+func deletePeerRoutes(tx *loggingTx, onlyRouteMaskLengthsSQL, deleteSQL, peerIP string) (
+	deleted int64, goneMaskLengths []int, err error) {
+	rows, err := tx.Query(onlyRouteMaskLengthsSQL, peerIP)
+	if err != nil {
+		return 0, nil, err
+	}
+	defer func() { _ = rows.Close() }()
+	for rows.Next() {
+		var maskLength int
+		if err := rows.Scan(&maskLength); err != nil {
+			return 0, nil, err
+		}
+		goneMaskLengths = append(goneMaskLengths, maskLength)
+	}
+	if err := rows.Err(); err != nil {
+		return 0, nil, err
+	}
+
+	res, err := tx.Exec(deleteSQL, peerIP)
+	if err != nil {
+		return 0, nil, err
+	}
+	deleted, err = res.RowsAffected()
+	if err != nil {
+		return 0, nil, err
+	}
+
+	return deleted, goneMaskLengths, nil
+}
+
+// DeleteLiveRoutesOfPeer deletes every live route of the peer, IPv4 and IPv6,
+// in one transaction, as when the peer's BGP session goes down. It returns how
+// many routes it deleted.
+func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
+	d.lock("DeleteLiveRoutesOfPeer")
+	defer d.unlock()
+
+	tx, err := d.beginTx()
+	if err != nil {
+		return 0, fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer func() {
+		if err := tx.Rollback(); err != nil && err != sql.ErrTxDone {
+			d.logger.Error("Failed to rollback transaction", "error", err)
+		}
+	}()
+
+	deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(tx,
+		peerOnlyRouteMaskLengthsV4SQL, deletePeerLiveRoutesV4SQL, peerIP)
+	if err != nil {
+		return 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w", peerIP, err)
+	}
+	deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(tx,
+		peerOnlyRouteMaskLengthsV6SQL, deletePeerLiveRoutesV6SQL, peerIP)
+	if err != nil {
+		return 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w", peerIP, err)
+	}
+
+	if err := tx.Commit(); err != nil {
+		return 0, fmt.Errorf("failed to commit transaction: %w", err)
+	}
+
+	d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
+	d.counts.addToDistribution(goneMaskLengthsV4, goneMaskLengthsV6, -1)
+
+	return int(deletedV4 + deletedV6), nil
+}
+
 // UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a single transaction
 func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	if len(prefixes) == 0 {

internal/database/interface.go

@@ -57,6 +57,7 @@ type Store interface {
 	UpsertLiveRouteBatch(routes []*LiveRoute) error
 	DeleteLiveRoute(prefix, peerIP string) error
 	DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
+	DeleteLiveRoutesOfPeer(peerIP string) (deleted int, err error)
 	GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
 	GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
 	GetLiveRouteCounts() (ipv4Count, ipv6Count int, err error)

internal/database/schema.sql

@@ -107,7 +107,10 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 (
 	as_path TEXT NOT NULL, -- JSON array
 	next_hop TEXT NOT NULL,
 	last_updated DATETIME NOT NULL,
-	UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix
+	-- A peer has at most one route for a prefix. peer_ip comes first so that
+	-- this key also finds all of a peer's routes, when its BGP session goes
+	-- down; lookups by prefix use the prefix index below.
+	UNIQUE(peer_ip, prefix)
 );
 
 -- IPv6 routing table maintained by PrefixHandler
@@ -120,7 +123,8 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 (
 	as_path TEXT NOT NULL, -- JSON array
 	next_hop TEXT NOT NULL,
 	last_updated DATETIME NOT NULL,
-	UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix
+	-- As in live_routes_v4: peer_ip first, to find all of a peer's routes.
+	UNIQUE(peer_ip, prefix)
 );
 
 -- Indexes for live_routes_v4 table

internal/ristypes/ris.go

@@ -6,6 +6,22 @@ import (
 	"time"
 )
 
+// Message types and the session state the handlers act on, as the feed writes
+// them in a message's type and state fields.
+const (
+	// MessageTypeUpdate is a BGP UPDATE: announcements and withdrawals.
+	MessageTypeUpdate = "UPDATE"
+
+	// MessageTypeState is a change in the state of the BGP session between
+	// the collector and the peer.
+	MessageTypeState = "STATE"
+
+	// StateDown is the state of a STATE message whose session has ended: the
+	// collector no longer has any of the peer's routes, and sends no
+	// withdrawals for them.
+	StateDown = "down"
+)
+
 // ASPath represents a BGP AS path as a slice of AS numbers.
 // It handles JSON unmarshaling of both simple arrays and nested AS sets,
 // flattening any nested structures into a single sequence of AS numbers.
@@ -85,6 +101,8 @@ type RISMessage struct {
 	Announcements []RISAnnouncement `json:"announcements,omitempty"`
 	Withdrawals   []string          `json:"withdrawals,omitempty"`
 	Raw           string            `json:"-"`
+	// State is set on STATE messages: the session's new state, such as StateDown.
+	State string `json:"state,omitempty"`
 }
 
 // RISAnnouncement represents a BGP route announcement within a RIS message.

internal/routewatch/app_integration_test.go

@@ -364,6 +364,12 @@ func (m *mockStore) DeleteLiveRouteBatch(deletions []database.LiveRouteDeletion)
 	return nil
 }
 
+// DeleteLiveRoutesOfPeer mock implementation
+func (m *mockStore) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
+	// Simple mock - deletes nothing
+	return 0, nil
+}
+
 // GetOrCreateASNBatch mock implementation
 func (m *mockStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
 	m.mu.Lock()

internal/routewatch/peerhandler.go

@@ -1,7 +1,7 @@
 package routewatch
 
 // peerhandler.go provides batched peer tracking functionality for BGP route monitoring.
-// It tracks BGP peers from all incoming RIS messages and maintains peer state in the database.
+// It tracks BGP peers from incoming UPDATE messages and maintains peer state in the database.
 
 import (
 	"strconv"
@@ -33,7 +33,7 @@ const (
 	peerBatchTimeout = 2 * time.Second
 )
 
-// PeerHandler tracks BGP peers from all message types using batched operations.
+// PeerHandler tracks BGP peers from UPDATE messages using batched operations.
 // It maintains a queue of peer updates and periodically flushes them to the database
 // in batches to improve performance. The handler deduplicates peer updates within
 // each batch, keeping only the most recent update for each peer IP address.
@@ -82,11 +82,11 @@ func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler {
 	return h
 }
 
-// WantsMessage returns true for all message types since peer information
-// is extracted from every RIS message regardless of type. This satisfies
-// the MessageHandler interface.
-func (h *PeerHandler) WantsMessage(_ string) bool {
-	return true
+// WantsMessage returns true for UPDATE messages only. A peer whose session
+// never comes up sends STATE messages and nothing else, so a STATE message is
+// not a sign of a peer that carries routes.
+func (h *PeerHandler) WantsMessage(messageType string) bool {
+	return messageType == ristypes.MessageTypeUpdate
 }
 
 // QueueCapacity returns the desired queue capacity for this handler.

internal/routewatch/prefixhandler.go

@@ -3,6 +3,7 @@ package routewatch
 import (
 	"net"
 	"net/netip"
+	"slices"
 	"strings"
 	"sync"
 	"time"
@@ -16,9 +17,10 @@ import (
 
 const (
 	// prefixHandlerQueueSize is the queue capacity for prefix tracking
-	// operations, about 4 seconds of feed at peak. The streamer drops rather
-	// than blocks when a queue is full, so this bounds memory. Batches still
-	// flush on a timer (prefixBatchTimeout).
+	// operations, about 4 seconds of feed at peak. The streamer drops updates
+	// rather than blocks when a queue is full (a STATE message is held until
+	// there is room), so this bounds memory. Batches still flush on a timer
+	// (prefixBatchTimeout).
 	prefixHandlerQueueSize = 20000
 
 	// prefixBatchSize is the number of prefix updates to batch together
@@ -40,17 +42,19 @@ const (
 )
 
 // PrefixHandler tracks BGP prefixes and maintains a live routing table in the database.
-// Routes are added on announcement and deleted on withdrawal.
+// Routes are added on announcement and deleted on withdrawal, and all routes of
+// a peer are deleted when its BGP session goes down.
 type PrefixHandler struct {
 	db      database.Store
 	logger  *logger.Logger
 	metrics *metrics.Tracker
 
-	// Batching. mu guards batch only: the flush goroutine takes the batch
-	// under mu and writes it after releasing mu, so HandleMessage waits for
-	// a database write only while batch is at prefixBatchMaxSize.
+	// Batching. mu guards batch and peersDown only: the flush goroutine takes
+	// both 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
+	peersDown   []string      // peers whose BGP session went down since the batch was taken
 	batchTaken  sync.Cond     // signalled when the flush goroutine takes batch; its L is &mu
 	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
 	lastFlush   time.Time     // used only by the flush goroutine
@@ -93,8 +97,9 @@ func (h *PrefixHandler) SetMetricsTracker(metrics *metrics.Tracker) {
 
 // WantsMessage returns true if this handler wants to process messages of the given type
 func (h *PrefixHandler) WantsMessage(messageType string) bool {
-	// We only care about UPDATE messages for the routing table
-	return messageType == "UPDATE"
+	// UPDATE messages change routes; a STATE message can say a peer's
+	// session went down, which removes all of its routes.
+	return messageType == ristypes.MessageTypeUpdate || messageType == ristypes.MessageTypeState
 }
 
 // QueueCapacity returns the desired queue capacity for this handler
@@ -105,6 +110,14 @@ func (h *PrefixHandler) QueueCapacity() int {
 
 // HandleMessage processes a message to track prefix information
 func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
+	if msg.Type == ristypes.MessageTypeState {
+		if msg.State == ristypes.StateDown {
+			h.peerSessionDown(msg.Peer)
+		}
+
+		return
+	}
+
 	// Use the pre-parsed timestamp
 	timestamp := msg.ParsedTimestamp
 
@@ -170,6 +183,19 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
+// 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 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.
+func (h *PrefixHandler) peerSessionDown(peer string) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	h.batch = slices.DeleteFunc(h.batch, func(update prefixUpdate) bool { return update.peer == peer })
+	h.peersDown = append(h.peersDown, peer)
+}
+
 // flushLoop runs in a goroutine, the only one that writes the batch: it
 // flushes when the batch is full, periodically, and once more on Stop.
 func (h *PrefixHandler) flushLoop() {
@@ -194,33 +220,53 @@ func (h *PrefixHandler) flushLoop() {
 	}
 }
 
-// takeBatch returns the pending updates and leaves an empty batch in their
-// place for HandleMessage to fill.
-func (h *PrefixHandler) takeBatch() []prefixUpdate {
+// takeBatch returns the pending updates and the peers whose session went down
+// since the last flush, and leaves an empty batch and no peers in their place
+// for HandleMessage to fill.
+func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	if len(h.batch) == 0 {
-		return nil
+	if len(h.batch) == 0 && len(h.peersDown) == 0 {
+		return nil, nil
 	}
-	batch := h.batch
+	batch, peersDown := h.batch, h.peersDown
 	h.batch = make([]prefixUpdate, 0, prefixBatchSize)
+	h.peersDown = nil
 	h.batchTaken.Broadcast()
 
-	return batch
+	return batch, peersDown
 }
 
 // flushBatch takes the pending batch and writes it to the database, without
 // holding the mutex during the write.
 func (h *PrefixHandler) flushBatch() {
-	batch := h.takeBatch()
-	if len(batch) == 0 {
+	batch, peersDown := h.takeBatch()
+	if len(batch) == 0 && len(peersDown) == 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. Most downs
+	// come from peers that have no routes, so only a delete that removed
+	// routes is logged.
+	for _, peer := range peersDown {
+		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",
+				"peer", peer, "error", err)
+
+			continue
+		}
+		if deleted > 0 {
+			h.logger.Info("Deleted the live routes of a peer whose session went down",
+				"peer", peer, "routes", deleted)
+		}
+	}
+
 	// 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
@@ -296,6 +342,7 @@ func (h *PrefixHandler) flushBatch() {
 	elapsed := time.Since(startTime)
 	h.logger.Debug("Flushed prefix batch",
 		"batch_size", batchSize,
+		"peers_down", len(peersDown),
 		"unique_prefixes", len(prefixesToUpdate),
 		"success", successCount,
 		"duration_ms", elapsed.Milliseconds(),

internal/routewatch/prefixhandler_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"sync"
 	"testing"
 	"time"
 
@@ -288,6 +289,256 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) {
 	}
 }
 
+// TestStateMessagesGoToThePrefixHandlerNotThePeerHandler checks that STATE
+// messages go to the prefix handler, which acts on a down, and not to the peer
+// handler, which would record a peer whose session never comes up as active.
+func TestStateMessagesGoToThePrefixHandlerNotThePeerHandler(t *testing.T) {
+	if !(&PrefixHandler{}).WantsMessage(ristypes.MessageTypeState) {
+		t.Error("the prefix handler does not want STATE messages")
+	}
+	if (&PeerHandler{}).WantsMessage(ristypes.MessageTypeState) {
+		t.Error("the peer handler wants STATE messages")
+	}
+	if !(&PeerHandler{}).WantsMessage(ristypes.MessageTypeUpdate) {
+		t.Error("the peer handler does not want UPDATE messages")
+	}
+}
+
+// TestPrefixHandlerDeletesAPeersRoutesWhenItsSessionGoesDown runs
+// announcements from two peers through the prefix handler into a real
+// database, then a batch in which one of the peers announces a prefix and its
+// session goes down. The down deletes every live route of that peer, IPv4 and
+// IPv6, the one announced just before it included, and keeps the other peer's
+// routes for the same prefixes. An announcement the peer sends after the down,
+// in the same batch, is kept. The in-memory route counts and prefix
+// distribution must match the route tables.
+func TestPrefixHandlerDeletesAPeersRoutesWhenItsSessionGoesDown(t *testing.T) {
+	const (
+		otherPeerIP = "2001:db8:ffff::2"
+		// Both peers announce the shared prefixes, only testPeerIP the others.
+		sharedV4   = "198.51.100.0/24"
+		sharedV6   = "2001:db8:1::/48"
+		onlyV4     = "203.0.113.0/24"
+		onlyV6     = "2001:db8:2::/48"
+		beforeDown = "192.0.2.0/24"
+		afterDown  = "2001:db8:3::/48"
+		// The mask lengths of the IPv4 and of the IPv6 prefixes.
+		maskLengthV4 = 24
+		maskLengthV6 = 48
+	)
+
+	tests := []struct {
+		name string
+		// Handled and flushed as one batch after both peers' announcements.
+		batch []*ristypes.RISMessage
+		// Whether testPeerIP is left with a live route for afterDown.
+		wantAfterDown bool
+	}{
+		{
+			name:  "the down ends the batch",
+			batch: []*ristypes.RISMessage{peerAnnouncement(testPeerIP, beforeDown), sessionDown(testPeerIP)},
+		},
+		{
+			name: "the peer announces a prefix after the down",
+			batch: []*ristypes.RISMessage{
+				peerAnnouncement(testPeerIP, beforeDown),
+				sessionDown(testPeerIP),
+				peerAnnouncement(testPeerIP, afterDown),
+			},
+			wantAfterDown: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
+			if err != nil {
+				t.Fatalf("failed to create database: %v", err)
+			}
+			defer func() { _ = db.Close() }()
+
+			// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+			h := &PrefixHandler{db: db, logger: logger.New()}
+			h.HandleMessage(peerAnnouncement(testPeerIP, sharedV4, sharedV6, onlyV4, onlyV6))
+			h.HandleMessage(peerAnnouncement(otherPeerIP, sharedV4, sharedV6))
+			h.flushBatch()
+			for _, msg := range tt.batch {
+				h.HandleMessage(msg)
+			}
+			h.flushBatch()
+
+			// The peers left with a live route for each prefix.
+			wantPeers := map[string][]string{
+				sharedV4:   {otherPeerIP},
+				sharedV6:   {otherPeerIP},
+				onlyV4:     nil,
+				onlyV6:     nil,
+				beforeDown: nil,
+				afterDown:  nil,
+			}
+			wantRoutesV6 := 1
+			if tt.wantAfterDown {
+				wantPeers[afterDown] = []string{testPeerIP}
+				wantRoutesV6++
+			}
+
+			assertPeers(t, db, wantPeers)
+			assertCounts(t, db, ipv4Version, 1,
+				[]database.PrefixDistribution{{MaskLength: maskLengthV4, Count: 1}})
+			assertCounts(t, db, ipv6Version, wantRoutesV6,
+				[]database.PrefixDistribution{{MaskLength: maskLengthV6, Count: wantRoutesV6}})
+		})
+	}
+}
+
+// TestPrefixHandlerKeepsWhatAPeerAnnouncesAfterTheFlushOfItsDown runs, through
+// the prefix handler into a real database, one flush after another: a peer's
+// announcement, its session going down, a new announcement from the peer, and
+// another peer's announcement. The down deletes the peer's routes on the flush
+// that handles it and on no later flush, so the route the peer announced after
+// it is kept.
+func TestPrefixHandlerKeepsWhatAPeerAnnouncesAfterTheFlushOfItsDown(t *testing.T) {
+	const (
+		otherPeerIP = "2001:db8:ffff::2"
+		beforeDown  = "192.0.2.0/24"
+		afterDown   = "198.51.100.0/24"
+		otherPeers  = "203.0.113.0/24"
+		maskLength  = 24
+	)
+
+	db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+	h := &PrefixHandler{db: db, logger: logger.New()}
+	for _, msg := range []*ristypes.RISMessage{
+		peerAnnouncement(testPeerIP, beforeDown),
+		sessionDown(testPeerIP),
+		peerAnnouncement(testPeerIP, afterDown),
+		peerAnnouncement(otherPeerIP, otherPeers),
+	} {
+		h.HandleMessage(msg)
+		h.flushBatch()
+	}
+
+	assertPeers(t, db, map[string][]string{
+		beforeDown: nil,
+		afterDown:  {testPeerIP},
+		otherPeers: {otherPeerIP},
+	})
+	assertCounts(t, db, ipv4Version, 2, []database.PrefixDistribution{{MaskLength: maskLength, Count: 2}})
+}
+
+// firstWriteHeldStore is a real database whose first live route batch write
+// waits until the test closes release. writeStarted is closed when it starts.
+type firstWriteHeldStore struct {
+	*database.Database
+
+	writeStarted chan struct{}
+	release      chan struct{}
+	once         sync.Once
+}
+
+func (s *firstWriteHeldStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
+	s.once.Do(func() {
+		close(s.writeStarted)
+		<-s.release
+	})
+
+	return s.Database.UpsertLiveRouteBatch(routes)
+}
+
+// TestPrefixHandlerHandlesADownWhileAFlushWrites holds the write of a flush
+// that stores a peer's route and meanwhile hands the prefix handler the peer's
+// session going down, then a new announcement from the peer. The next flush
+// must delete the route the held write stored and keep the one announced after
+// the down.
+func TestPrefixHandlerHandlesADownWhileAFlushWrites(t *testing.T) {
+	const (
+		beforeDown = "192.0.2.0/24"
+		afterDown  = "198.51.100.0/24"
+		maskLength = 24
+	)
+
+	db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+	store := &firstWriteHeldStore{Database: db, writeStarted: make(chan struct{}), release: make(chan struct{})}
+
+	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+	h := &PrefixHandler{db: store, logger: logger.New()}
+	h.HandleMessage(peerAnnouncement(testPeerIP, beforeDown))
+	flushed := make(chan struct{})
+	go func() {
+		h.flushBatch()
+		close(flushed)
+	}()
+	select {
+	case <-store.writeStarted:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the flush did not start its write")
+	}
+
+	if !returnsInTime(func() {
+		h.HandleMessage(sessionDown(testPeerIP))
+		h.HandleMessage(peerAnnouncement(testPeerIP, afterDown))
+	}) {
+		t.Fatal("HandleMessage did not return while a database write was held")
+	}
+	close(store.release)
+	select {
+	case <-flushed:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the flush did not finish after its write was let go")
+	}
+	h.flushBatch()
+
+	assertPeers(t, db, map[string][]string{beforeDown: nil, afterDown: {testPeerIP}})
+	assertCounts(t, db, ipv4Version, 1, []database.PrefixDistribution{{MaskLength: maskLength, Count: 1}})
+}
+
+// peerAnnouncement returns an UPDATE message in which peer announces prefixes
+// with origin AS testASNB.
+func peerAnnouncement(peer string, prefixes ...string) *ristypes.RISMessage {
+	return &ristypes.RISMessage{
+		Type:            ristypes.MessageTypeUpdate,
+		Peer:            peer,
+		Path:            ristypes.ASPath{testASNA, testASNB},
+		ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
+		Announcements:   []ristypes.RISAnnouncement{{NextHop: peer, Prefixes: prefixes}},
+	}
+}
+
+// sessionDown returns the STATE message that says peer's BGP session went down.
+func sessionDown(peer string) *ristypes.RISMessage {
+	return &ristypes.RISMessage{Type: ristypes.MessageTypeState, Peer: peer, State: ristypes.StateDown}
+}
+
+// assertPeers checks, for each prefix in wantPeers, that the peers with a live
+// route for it are the ones listed, in peer IP order; nil means none.
+func assertPeers(t *testing.T, db *database.Database, wantPeers map[string][]string) {
+	t.Helper()
+
+	for prefix, want := range wantPeers {
+		routes, _, err := db.GetPrefixDetailsContext(context.Background(), prefix)
+		if err != nil && !errors.Is(err, database.ErrNoRoute) {
+			t.Fatalf("GetPrefixDetailsContext(%s): %v", prefix, err)
+		}
+		var got []string
+		for _, route := range routes {
+			got = append(got, route.PeerIP)
+		}
+		if !slices.Equal(got, want) {
+			t.Errorf("peers with a live route for %s = %v, want %v", prefix, got, want)
+		}
+	}
+}
+
 // assertCounts checks that the live route count and prefix distribution of one
 // IP version are wantRoutes and wantDistribution, both as held in memory and as
 // read from the route table.

internal/streamer/streamer.go

@@ -78,7 +78,8 @@ type MessageHandler interface {
 	// QueueCapacity returns the desired queue capacity for this handler.
 	// Handlers that process quickly can have larger queues to buffer bursts.
 	// When the queue fills up, messages will be dropped according to the
-	// backpressure algorithm.
+	// backpressure algorithm, except STATE messages, which are held until
+	// there is room.
 	QueueCapacity() int
 }
 
@@ -102,6 +103,9 @@ type handlerInfo struct {
 	handler MessageHandler
 	queue   chan *ristypes.RISMessage
 	metrics handlerMetrics
+	// heldState holds, in feed order, the STATE messages that found queue
+	// full (see queueMessage). Only the read loop uses it.
+	heldState []*ristypes.RISMessage
 }
 
 // Streamer manages a connection to the RIPE RIS Live streaming API for receiving
@@ -629,7 +633,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 
 		// Process based on message type
 		switch msg.Type {
-		case "UPDATE":
+		case ristypes.MessageTypeUpdate:
 			// Process BGP UPDATE messages
 			// Will be dispatched to handlers
 		case "RIS_PEER_STATE":
@@ -668,9 +672,9 @@ func (s *Streamer) stream(ctx context.Context) error {
 			)
 
 			continue
-		case "STATE":
-			// Peer state changes - silently ignore
-			continue
+		case ristypes.MessageTypeState:
+			// The BGP session with the peer changed state, for example went
+			// down. Dispatched to the handlers that want it.
 		default:
 			s.logger.Warn("Unknown message type, skipping",
 				"type", msg.Type,
@@ -693,30 +697,20 @@ func (s *Streamer) stream(ctx context.Context) error {
 				continue
 			}
 
-			// Check if we should drop due to backpressure
-			if s.shouldDropForBackpressure(info) {
+			if !s.queueMessage(info, &msg) {
 				atomic.AddUint64(&info.metrics.droppedCount, 1)
 				atomic.AddUint64(&s.totalDropped, 1)
 
 				continue
 			}
 
-			// Try to queue the message
-			select {
-			case info.queue <- &msg:
-				// Message queued successfully
-				// Update high water mark if needed
-				queueLen := len(info.queue)
-				info.metrics.mu.Lock()
-				if queueLen > info.metrics.queueHighWaterMark {
-					info.metrics.queueHighWaterMark = queueLen
-				}
-				info.metrics.mu.Unlock()
-			default:
-				// Queue is full, drop the message
-				atomic.AddUint64(&info.metrics.droppedCount, 1)
-				atomic.AddUint64(&s.totalDropped, 1)
+			// Update high water mark if needed
+			queueLen := len(info.queue)
+			info.metrics.mu.Lock()
+			if queueLen > info.metrics.queueHighWaterMark {
+				info.metrics.queueHighWaterMark = queueLen
 			}
+			info.metrics.mu.Unlock()
 		}
 		s.mu.RUnlock()
 	}
@@ -736,6 +730,50 @@ func (s *Streamer) stream(ctx context.Context) error {
 	return errStreamEnded
 }
 
+// queueMessage puts msg on the handler's queue and reports false if it dropped
+// msg instead. It never waits for room, so a full queue holds up neither the
+// read loop nor Stop.
+//
+// A message is dropped while the queue is filling up (shouldDropForBackpressure)
+// or full, except a STATE message: one whose state is down is the only notice
+// that all of a peer's routes are gone. A STATE message skips the random drop,
+// and one that finds the queue full is held in info.heldState. Held messages
+// go into the queue, oldest first, before anything else for the handler, as
+// soon as there is room; while any is still held, the handler's other messages
+// are dropped, so none of them overtakes a held down.
+//
+// The held list stays small: a STATE message is a few hundred bytes and the
+// feed sends about ten a second, so a queue full for a minute holds about 600.
+func (s *Streamer) queueMessage(info *handlerInfo, msg *ristypes.RISMessage) bool {
+	for len(info.heldState) > 0 && trySend(info.queue, info.heldState[0]) {
+		info.heldState = info.heldState[1:]
+	}
+
+	if msg.Type == ristypes.MessageTypeState {
+		if len(info.heldState) > 0 || !trySend(info.queue, msg) {
+			info.heldState = append(info.heldState, msg)
+		}
+
+		return true
+	}
+
+	if len(info.heldState) > 0 || s.shouldDropForBackpressure(info) {
+		return false
+	}
+
+	return trySend(info.queue, msg)
+}
+
+// trySend puts msg on queue if there is room and reports whether it did.
+func trySend(queue chan *ristypes.RISMessage, msg *ristypes.RISMessage) bool {
+	select {
+	case queue <- msg:
+		return true
+	default:
+		return false
+	}
+}
+
 // shouldDropForBackpressure determines if a message should be dropped based on queue utilization
 func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
 	// Calculate queue utilization

internal/streamer/streamer_test.go

@@ -108,6 +108,204 @@ func (h updateHandler) HandleMessage(*ristypes.RISMessage) {
 	}
 }
 
+// stateDownLine is the STATE message RIS Live sends when the collector's BGP
+// session with the peer stateDownPeer goes down.
+const (
+	stateDownPeer = "192.0.2.1"
+	stateDownLine = `{"type":"ris_message","data":{"type":"STATE","peer":"` + stateDownPeer + `",` +
+		`"peer_asn":"64496","timestamp":1700000000,"state":"down"}}` + "\n"
+)
+
+// stateHandler wants STATE messages only. Tests read what it is handed from
+// its queue, so HandleMessage does nothing.
+type stateHandler struct{}
+
+func (stateHandler) WantsMessage(messageType string) bool {
+	return messageType == ristypes.MessageTypeState
+}
+func (stateHandler) QueueCapacity() int                 { return 10 }
+func (stateHandler) HandleMessage(*ristypes.RISMessage) {}
+
+// TestStreamHandsStateMessagesToHandlersThatWantThem has the server send one
+// STATE message whose state is down. The streamer must queue it, with its
+// peer and state, for a handler that wants STATE messages.
+func TestStreamHandsStateMessagesToHandlersThatWantThem(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine)
+	}))
+	defer srv.Close()
+
+	s := New(logger.New(), metrics.New())
+	s.url = srv.URL
+	s.RegisterHandler(stateHandler{})
+
+	if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
+		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+	}
+
+	select {
+	case msg := <-s.handlers[0].queue:
+		if msg.Type != ristypes.MessageTypeState || msg.Peer != stateDownPeer || msg.State != ristypes.StateDown {
+			t.Errorf("queued message has type %q, peer %q, state %q; want %q, %q, %q",
+				msg.Type, msg.Peer, msg.State, ristypes.MessageTypeState, stateDownPeer, ristypes.StateDown)
+		}
+	default:
+		t.Fatal("the STATE message was not queued for the handler that wants it")
+	}
+}
+
+// TestHeldStateMessagesKeepTheirPlaceInTheQueue hands queueMessage two STATE
+// messages and two updates while the queue is full, making room in between.
+// Neither STATE message may be dropped: each must reach the queue after
+// everything queued before it and before anything handed over after it. The
+// update handed over while a STATE message is held is dropped.
+func TestHeldStateMessagesKeepTheirPlaceInTheQueue(t *testing.T) {
+	s := New(logger.New(), metrics.New())
+	s.RegisterHandler(stateHandler{})
+	info := s.handlers[0]
+
+	before := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate}
+	for len(info.queue) < cap(info.queue) {
+		info.queue <- before
+	}
+	firstDown := &ristypes.RISMessage{Type: ristypes.MessageTypeState, State: ristypes.StateDown}
+	secondDown := &ristypes.RISMessage{Type: ristypes.MessageTypeState, State: ristypes.StateDown}
+	updateWhileHeld := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate}
+	after := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate}
+
+	if !s.queueMessage(info, firstDown) {
+		t.Fatal("the first STATE message was dropped on a full queue")
+	}
+	<-info.queue // the handler makes room for one message
+	if !s.queueMessage(info, secondDown) {
+		t.Fatal("the second STATE message was dropped")
+	}
+	if s.queueMessage(info, updateWhileHeld) {
+		t.Fatal("an update was queued while a STATE message was held")
+	}
+
+	// The handler takes every message queued before the first STATE message.
+	for range cap(info.queue) - 1 {
+		if <-info.queue != before {
+			t.Fatal("a message overtook those queued before the first STATE message")
+		}
+	}
+	if !s.queueMessage(info, after) {
+		t.Fatal("an update was dropped on a queue with room")
+	}
+
+	if <-info.queue != firstDown {
+		t.Fatal("the first STATE message did not come next")
+	}
+	if <-info.queue != secondDown {
+		t.Fatal("the second STATE message did not come after the first")
+	}
+	if <-info.queue != after {
+		t.Fatal("the update handed over after the STATE messages did not come after them")
+	}
+	if len(info.queue) != 0 {
+		t.Errorf("%d messages left on the queue, want none", len(info.queue))
+	}
+}
+
+// newStreamerWithFullStateQueue returns a streamer reading from url with one
+// handler, a stateHandler whose queue is full.
+func newStreamerWithFullStateQueue(url string) *Streamer {
+	s := New(logger.New(), metrics.New())
+	s.url = url
+	s.RegisterHandler(stateHandler{})
+	queue := s.handlers[0].queue
+	for len(queue) < cap(queue) {
+		queue <- &ristypes.RISMessage{Type: ristypes.MessageTypeState}
+	}
+
+	return s
+}
+
+// TestOtherHandlersKeepReceivingWhileAStateMessageIsHeld has the server send a
+// STATE message whose state is down, then an UPDATE, while the queue of the
+// handler that wants STATE messages is full. The STATE message must be held
+// for that handler, and the read loop must go on to queue the UPDATE for the
+// handler that wants it.
+func TestOtherHandlersKeepReceivingWhileAStateMessageIsHeld(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine+updateLine)
+	}))
+	defer srv.Close()
+
+	s := newStreamerWithFullStateQueue(srv.URL)
+	s.RegisterHandler(updateHandler{})
+
+	if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
+		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+	}
+
+	held := s.handlers[0].heldState
+	if len(held) != 1 || held[0].Peer != stateDownPeer || held[0].State != ristypes.StateDown {
+		t.Errorf("held %d STATE messages, want the down from %s alone", len(held), stateDownPeer)
+	}
+	if got := len(s.handlers[1].queue); got != 1 {
+		t.Errorf("%d messages queued for the handler that wants UPDATE messages, want 1", got)
+	}
+}
+
+// TestStopReturnsAtOnceWithStateMessagesHeld has the server send two STATE
+// messages and hold the response open while the queue of the handler that
+// wants them is full. Stop must return at once and end the stream.
+func TestStopReturnsAtOnceWithStateMessagesHeld(t *testing.T) {
+	const (
+		lines   = 2
+		maxWait = 5 * time.Second
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine+stateDownLine)
+		w.(http.Flusher).Flush()
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	s := newStreamerWithFullStateQueue(srv.URL)
+	// The raw handler sees each line before the read loop hands it to the
+	// queues, so once it has seen the second line, the first is held.
+	linesSeen := make(chan struct{}, lines)
+	s.RegisterRawHandler(func(string) { linesSeen <- struct{}{} })
+
+	// As in TestStopBeforeMessageReachesQueues, setting cancel as Start does
+	// lets Stop cancel the stream run here.
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancel = cancel
+	streamErr := make(chan error, 1)
+	go func() { streamErr <- s.stream(ctx) }()
+
+	for range lines {
+		select {
+		case <-linesSeen:
+		case <-time.After(maxWait):
+			t.Fatalf("the read loop did not read %d lines within %s", lines, maxWait)
+		}
+	}
+
+	stopped := make(chan struct{})
+	go func() {
+		s.Stop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(maxWait):
+		t.Fatalf("Stop did not return within %s while a STATE message was held", maxWait)
+	}
+
+	select {
+	case err := <-streamErr:
+		if !errors.Is(err, context.Canceled) {
+			t.Errorf("stream returned %v, want %v", err, context.Canceled)
+		}
+	case <-time.After(maxWait):
+		t.Fatalf("stream did not end within %s of Stop", maxWait)
+	}
+}
+
 // TestStopBeforeMessageReachesQueues stops the streamer after the read loop
 // has checked for cancellation but before it hands the message to the handler
 // queues. That is the gap a stop from another goroutine can land in, and it