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 those routes stayed in the live tables for good.

The streamer now hands STATE messages to the handlers that want them and
never drops one for backpressure: it waits for room in a full queue until
the idle timer drops the connection. When the prefix handler gets one whose
state is down, it drops the peer's updates still in its batch and notes the
peer; the next flush deletes all of the peer's live routes, IPv4 and IPv6,
before writing the batch, so updates the peer sends after the down are
kept. The live route tables' unique key now leads with peer_ip, so it 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**

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

@@ -15,9 +15,9 @@ import (
 
 const (
 	// peerHandlerQueueSize is the queue capacity for peer tracking operations,
-	// about 4 seconds of feed at peak. The streamer drops rather than blocks
+	// about 4 seconds of feed at peak. The streamer drops messages rather than
-	// when a queue is full, so this bounds memory. Batches still flush on a
+	// blocks when a queue is full (a STATE message waits for room), so this
-	// timer (peerBatchTimeout).
+	// bounds memory. Batches still flush on a timer (peerBatchTimeout).
 	peerHandlerQueueSize = 20000
 
 	// peerBatchSize is the number of peer updates to batch together

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
+	// operations, about 4 seconds of feed at peak. The streamer drops updates
-	// than blocks when a queue is full, so this bounds memory. Batches still
+	// rather than blocks when a queue is full (a STATE message waits for
-	// flush on a timer (prefixBatchTimeout).
+	// 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
+	// Batching. mu guards batch and peersDown only: the flush goroutine takes
-	// under mu and writes it after releasing mu, so HandleMessage waits for
+	// both under mu and writes them after releasing mu, so HandleMessage
-	// a database write only while batch is at prefixBatchMaxSize.
+	// 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
+	// UPDATE messages change routes; a STATE message can say a peer's
-	return messageType == "UPDATE"
+	// 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
+// takeBatch returns the pending updates and the peers whose session went down
-// place for HandleMessage to fill.
+// since the last flush, and leaves an empty batch and no peers in their place
-func (h *PrefixHandler) takeBatch() []prefixUpdate {
+// for HandleMessage to fill.
+func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	if len(h.batch) == 0 {
+	if len(h.batch) == 0 && len(h.peersDown) == 0 {
-		return nil
+		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()
+	batch, peersDown := h.takeBatch()
-	if len(batch) == 0 {
+	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,241 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) {
 	}
 }
 
+// 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,7 @@ 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 wait for room.
 	QueueCapacity() int
 }
 
@@ -629,7 +629,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 +668,9 @@ func (s *Streamer) stream(ctx context.Context) error {
 			)
 
 			continue
-		case "STATE":
+		case ristypes.MessageTypeState:
-			// Peer state changes - silently ignore
+			// The BGP session with the peer changed state, for example went
-			continue
+			// down. Dispatched to the handlers that want it.
 		default:
 			s.logger.Warn("Unknown message type, skipping",
 				"type", msg.Type,
@@ -693,30 +693,20 @@ func (s *Streamer) stream(ctx context.Context) error {
 				continue
 			}
 
-			// Check if we should drop due to backpressure
+			if !s.queueMessage(connCtx, info, &msg) {
-			if s.shouldDropForBackpressure(info) {
 				atomic.AddUint64(&info.metrics.droppedCount, 1)
 				atomic.AddUint64(&s.totalDropped, 1)
 
 				continue
 			}
 
-			// Try to queue the message
+			// Update high water mark if needed
-			select {
+			queueLen := len(info.queue)
-			case info.queue <- &msg:
+			info.metrics.mu.Lock()
-				// Message queued successfully
+			if queueLen > info.metrics.queueHighWaterMark {
-				// Update high water mark if needed
+				info.metrics.queueHighWaterMark = queueLen
-				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)
 			}
+			info.metrics.mu.Unlock()
 		}
 		s.mu.RUnlock()
 	}
@@ -736,6 +726,34 @@ func (s *Streamer) stream(ctx context.Context) error {
 	return errStreamEnded
 }
 
+// queueMessage puts msg on the handler's queue and reports whether it did. 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, so a STATE message waits for room
+// instead. The wait ends if the idle timer drops the connection by cancelling
+// connCtx, so the read loop never waits longer than its idle timeout.
+func (s *Streamer) queueMessage(connCtx context.Context, info *handlerInfo, msg *ristypes.RISMessage) bool {
+	if msg.Type == ristypes.MessageTypeState {
+		select {
+		case info.queue <- msg:
+			return true
+		case <-connCtx.Done():
+			return false
+		}
+	}
+
+	if s.shouldDropForBackpressure(info) {
+		return false
+	}
+
+	select {
+	case info.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,143 @@ 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")
+	}
+}
+
+// streamWithFullStateQueue runs stream, in a goroutine, against a streamer
+// reading from url whose stateHandler's queue is full of STATE messages with
+// no peer. It returns the queue and a channel that gets what stream returns.
+func streamWithFullStateQueue(url string, idleTimeout time.Duration) (
+	queue chan *ristypes.RISMessage, streamErr chan error) {
+	s := New(logger.New(), metrics.New())
+	s.url = url
+	s.idleTimeout = idleTimeout
+	s.RegisterHandler(stateHandler{})
+	queue = s.handlers[0].queue
+	for len(queue) < cap(queue) {
+		queue <- &ristypes.RISMessage{Type: ristypes.MessageTypeState}
+	}
+
+	streamErr = make(chan error, 1)
+	go func() { streamErr <- s.stream(context.Background()) }()
+
+	return queue, streamErr
+}
+
+// TestStreamWaitsForRoomForAStateMessage has the server send one STATE message
+// whose state is down while the queue of the handler that wants it is full. A
+// full queue is where the random drop always drops, and where a message that
+// cannot wait is dropped. The streamer must not drop it: it waits until the
+// handler takes a message off its queue, then queues it.
+func TestStreamWaitsForRoomForAStateMessage(t *testing.T) {
+	const (
+		stillWaitingAfter = 100 * time.Millisecond
+		maxWait           = 5 * time.Second
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine)
+	}))
+	defer srv.Close()
+
+	queue, streamErr := streamWithFullStateQueue(srv.URL, streamIdleTimeout)
+
+	select {
+	case err := <-streamErr:
+		t.Fatalf("stream returned %v while the queue was full: it dropped the STATE message", err)
+	case <-time.After(stillWaitingAfter):
+	}
+
+	<-queue // the handler takes a message, which makes room
+	select {
+	case err := <-streamErr:
+		if !errors.Is(err, errStreamEnded) {
+			t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+		}
+	case <-time.After(maxWait):
+		t.Fatalf("stream did not queue the STATE message within %s of there being room", maxWait)
+	}
+
+	var last *ristypes.RISMessage
+	for len(queue) > 0 {
+		last = <-queue
+	}
+	if last.Peer != stateDownPeer || last.State != ristypes.StateDown {
+		t.Errorf("last queued message has peer %q, state %q; want %q, %q",
+			last.Peer, last.State, stateDownPeer, ristypes.StateDown)
+	}
+}
+
+// TestStreamStopsWaitingForRoomAtTheIdleTimeout has the server send one STATE
+// message and then hold the response open, while the queue of the handler
+// that wants it stays full. The wait for room must not stall the read loop
+// past its idle timeout: the stream ends as when no line arrives.
+func TestStreamStopsWaitingForRoomAtTheIdleTimeout(t *testing.T) {
+	const (
+		idleTimeout = 50 * time.Millisecond
+		maxWait     = 5 * time.Second
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine)
+		w.(http.Flusher).Flush()
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	_, streamErr := streamWithFullStateQueue(srv.URL, idleTimeout)
+
+	select {
+	case err := <-streamErr:
+		if err == nil || errors.Is(err, errStreamEnded) {
+			t.Errorf("stream returned %v, want the error of a stream dropped as idle", err)
+		}
+	case <-time.After(maxWait):
+		t.Fatalf("stream still waiting for room %s after the idle timeout of %s", maxWait, idleTimeout)
+	}
+}
+
 // 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