diff --git a/README.md b/README.md
index 838f498..730044b 100644
--- a/README.md
+++ b/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**
diff --git a/internal/database/counts_test.go b/internal/database/counts_test.go
index b3259ba..5cd62b1 100644
--- a/internal/database/counts_test.go
+++ b/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
diff --git a/internal/database/database.go b/internal/database/database.go
index 29de8e3..7005042 100644
--- a/internal/database/database.go
+++ b/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 {
diff --git a/internal/database/interface.go b/internal/database/interface.go
index bbbe7f4..0311f05 100644
--- a/internal/database/interface.go
+++ b/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)
diff --git a/internal/database/schema.sql b/internal/database/schema.sql
index bf8a037..ef2f5dd 100644
--- a/internal/database/schema.sql
+++ b/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
diff --git a/internal/ristypes/ris.go b/internal/ristypes/ris.go
index a6831a4..e8130b6 100644
--- a/internal/ristypes/ris.go
+++ b/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.
diff --git a/internal/routewatch/app_integration_test.go b/internal/routewatch/app_integration_test.go
index 634e9c6..a23d39a 100644
--- a/internal/routewatch/app_integration_test.go
+++ b/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()
diff --git a/internal/routewatch/peerhandler.go b/internal/routewatch/peerhandler.go
index 9bd1f5d..74a32fc 100644
--- a/internal/routewatch/peerhandler.go
+++ b/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
-	// when a queue is full, so this bounds memory. Batches still flush on a
-	// timer (peerBatchTimeout).
+	// about 4 seconds of feed at peak. The streamer drops messages rather than
+	// blocks when a queue is full (a STATE message waits for room), so this
+	// bounds memory. Batches still flush on a timer (peerBatchTimeout).
 	peerHandlerQueueSize = 20000
 
 	// peerBatchSize is the number of peer updates to batch together
diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go
index 5c76410..c824439 100644
--- a/internal/routewatch/prefixhandler.go
+++ b/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 waits for
+	// 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(),
diff --git a/internal/routewatch/prefixhandler_test.go b/internal/routewatch/prefixhandler_test.go
index c1ce235..c535d3b 100644
--- a/internal/routewatch/prefixhandler_test.go
+++ b/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.
diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go
index 04e4429..29b370c 100644
--- a/internal/streamer/streamer.go
+++ b/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":
-			// 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 +693,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(connCtx, 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 +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
diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_test.go
index d2da82c..111b5f7 100644
--- a/internal/streamer/streamer_test.go
+++ b/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