Re-vendor the shared files from sneak/prompts at 0653579 (closes #52)

Takes .golangci.yml, .dockerignore, .gitignore, .editorconfig, the
workflow and REPO_POLICIES.md from sneak/prompts 0653579, with this
repository's own entries after the shared content. Lint (golangci-lint
v2.14.0) and test are Dockerfile phases that script/lint and script/test
build alone with a new CACHEBUST value, keeping Go's and golangci-lint's
caches; the image build depends on both. make fmt and make fmt-check also
format the markdown with prettier, pinned in yarn.lock. The agent
guidance moves to AGENTS.md. The Go code is fixed for what the new
configuration finds, and the AS and prefix detail reads now return the
rows' error after their loops.

Model: opus-5-5

README.md

@@ -132,7 +132,10 @@ routewatch/
     - **ASHandler**: Tracks all ASNs seen in AS paths
     - **PeerHandler**: Records BGP peer information
     - **PrefixHandler**: Maintains live routing table (upserts on announcement,
-      deletes on withdrawal)
+      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**
@@ -241,7 +244,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_internal_test.go

@@ -427,6 +427,114 @@ 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) {
+	t.Parallel()
+
+	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
+
+	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),
+	})
+	if 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)
+		}
+	}
+
+	for _, prefix := range []string{sharedV4, sharedV6} {
+		routes, _, err := db.GetPrefixDetailsContext(context.Background(), 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)
+		}
+	}
+
+	checkOneLiveRoutePerIPVersion(t, db)
+}
+
+// checkOneLiveRoutePerIPVersion checks that the database has one IPv4 live
+// route, for a /24, and one IPv6 live route, for a /32, both as counted in
+// memory and as read from the route tables.
+func checkOneLiveRoutePerIPVersion(t *testing.T, db *Database) {
+	t.Helper()
+
+	ctx := context.Background()
+
+	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

@@ -570,6 +570,108 @@ 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(
+	ctx context.Context, tx *loggingTx,
+	onlyRouteMaskLengthsSQL, deleteSQL, peerIP string,
+) (int64, []int, error) {
+	rows, err := tx.QueryContext(ctx, onlyRouteMaskLengthsSQL, peerIP)
+	if err != nil {
+		return 0, nil, err
+	}
+	defer func() { _ = rows.Close() }()
+
+	var goneMaskLengths []int
+
+	for rows.Next() {
+		var maskLength int
+
+		err := rows.Scan(&maskLength)
+		if err != nil {
+			return 0, nil, err
+		}
+
+		goneMaskLengths = append(goneMaskLengths, maskLength)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return 0, nil, err
+	}
+
+	res, err := tx.ExecContext(ctx, 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 d.rollback(tx)
+
+	ctx := context.Background()
+
+	deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, 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(ctx, tx,
+		peerOnlyRouteMaskLengthsV6SQL, deletePeerLiveRoutesV6SQL, peerIP)
+	if err != nil {
+		return 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w",
+			peerIP, err)
+	}
+
+	err = tx.Commit()
+	if 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
+}
+
 // upsertPrefix updates the last_seen time of prefix with the statements of its
 // IP version's table, inserting the prefix when it is new. It reports whether
 // it inserted it.

internal/database/interface.go

@@ -59,6 +59,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) (

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/database/slowquery.go

@@ -53,6 +53,17 @@ func (tx *loggingTx) QueryRowContext(
 	return tx.Tx.QueryRowContext(ctx, query, args...)
 }
 
+// QueryContext wraps sql.Tx.QueryContext to log slow queries
+func (tx *loggingTx) QueryContext(
+	ctx context.Context, query string, args ...any,
+) (*sql.Rows, error) {
+	start := time.Now()
+	defer logSlowQuery(tx.logger, query, start)
+
+	//nolint:sqlclosecheck // returns the rows, which the caller closes
+	return tx.Tx.QueryContext(ctx, query, args...)
+}
+
 // Exec wraps sql.Tx.Exec to log slow queries
 func (tx *loggingTx) Exec(query string, args ...any) (sql.Result, error) {
 	start := time.Now()

internal/ristypes/ris.go

@@ -7,6 +7,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.
@@ -93,6 +109,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_internal_test.go

@@ -393,6 +393,12 @@ func (m *mockStore) DeleteLiveRouteBatch(_ []database.LiveRouteDeletion) error {
 	return nil
 }
 
+// DeleteLiveRoutesOfPeer mock implementation
+func (m *mockStore) DeleteLiveRoutesOfPeer(_ 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/ashandler.go

@@ -85,7 +85,7 @@ func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 // the AS path information needed to track autonomous systems.
 func (h *ASHandler) WantsMessage(messageType string) bool {
 	// We only care about UPDATE messages for the database
-	return messageType == updateMessageType
+	return messageType == ristypes.MessageTypeUpdate
 }
 
 // QueueCapacity returns the recommended message queue size for this handler.

internal/routewatch/handler.go

@@ -7,10 +7,6 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
 )
 
-// updateMessageType is the RIS message type of BGP UPDATE messages, the only
-// type the database handlers process.
-const updateMessageType = "UPDATE"
-
 // SimpleHandler is a basic implementation of streamer.MessageHandler that
 // filters messages by type and delegates processing to a callback function.
 // It provides a simple way to handle specific RIS message types without

internal/routewatch/peerhandler.go

@@ -1,8 +1,8 @@
 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.
+// monitoring. It tracks BGP peers from incoming UPDATE messages and maintains
+// peer state in the database.
 
 import (
 	"strconv"
@@ -34,7 +34,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.
@@ -85,11 +85,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/peeringhandler.go

@@ -72,7 +72,7 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler
 // the AS path information needed to extract peering relationships.
 func (h *PeeringHandler) WantsMessage(messageType string) bool {
 	// We only care about UPDATE messages that have AS paths
-	return messageType == updateMessageType
+	return messageType == ristypes.MessageTypeUpdate
 }
 
 // QueueCapacity returns the buffer size for the handler's message queue.

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,21 @@ const (
 )
 
 // PrefixHandler tracks BGP prefixes and maintains a live routing table in the
-// database. Routes are added on announcement and deleted on withdrawal.
+// database. 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 holds the peers whose BGP session went down since the batch
+	// was taken.
+	peersDown []string
 	// batchTaken is signalled when the flush goroutine takes batch; its L is
 	// &mu.
 	batchTaken  sync.Cond
@@ -96,8 +102,10 @@ 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 == updateMessageType
+	// 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
@@ -108,6 +116,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
 
@@ -179,6 +195,21 @@ func (h *PrefixHandler) Stop() {
 	h.wg.Wait()
 }
 
+// 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() {
@@ -204,34 +235,40 @@ 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.
+	h.deleteLiveRoutesOfPeers(peersDown)
+
 	// 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
@@ -290,6 +327,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(),
@@ -298,6 +336,28 @@ func (h *PrefixHandler) flushBatch() {
 	h.lastFlush = time.Now()
 }
 
+// deleteLiveRoutesOfPeers deletes all live routes of each of peers, the peers
+// whose BGP session went down. Most downs come from peers that have no routes,
+// so only a delete that removed routes is logged.
+func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) {
+	for _, peer := range peers {
+		deleted, err := h.db.DeleteLiveRoutesOfPeer(peer)
+		if err != nil {
+			h.logger.Error(
+				"Failed to delete the live routes of a peer whose session went down",
+				"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)
+		}
+	}
+}
+
 // writeBatch writes the collected route upserts, route deletions and prefix
 // updates to the database and returns how many routes it wrote.
 func (h *PrefixHandler) writeBatch(

internal/routewatch/prefixhandler_internal_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"sync"
 	"testing"
 	"time"
 
@@ -263,6 +264,238 @@ 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) {
+	t.Parallel()
+
+	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) {
+	t.Parallel()
+
+	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(),
+			},
+		},
+		{
+			name: "the peer announces a prefix after the down",
+			batch: []*ristypes.RISMessage{
+				peerAnnouncement(testPeerIP, beforeDown),
+				sessionDown(),
+				peerAnnouncement(testPeerIP, afterDown),
+			},
+			wantAfterDown: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			db := handleBatches(t, [][]*ristypes.RISMessage{
+				{
+					peerAnnouncement(testPeerIP,
+						sharedV4, sharedV6, onlyV4, onlyV6),
+					peerAnnouncement(otherPeerIP, sharedV4, sharedV6),
+				},
+				tt.batch,
+			})
+
+			// 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) {
+	t.Parallel()
+
+	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
+	)
+
+	// Each message is handled and then flushed as its own batch.
+	db := handleBatches(t, [][]*ristypes.RISMessage{
+		{peerAnnouncement(testPeerIP, beforeDown)},
+		{sessionDown()},
+		{peerAnnouncement(testPeerIP, afterDown)},
+		{peerAnnouncement(otherPeerIP, otherPeers)},
+	})
+
+	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
+}
+
+// UpsertLiveRouteBatch holds the first call until release is closed, then
+// writes routes to the database.
+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) {
+	t.Parallel()
+
+	const (
+		beforeDown = "192.0.2.0/24"
+		afterDown  = "198.51.100.0/24"
+		maskLength = 24
+	)
+
+	db := newTestDatabase(t)
+	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())
+		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},
+	})
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {
@@ -301,6 +534,30 @@ func handleBatches(
 	return db
 }
 
+// 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 the BGP session of the peer
+// testPeerIP went down.
+func sessionDown() *ristypes.RISMessage {
+	return &ristypes.RISMessage{
+		Type:  ristypes.MessageTypeState,
+		Peer:  testPeerIP,
+		State: ristypes.StateDown,
+	}
+}
+
 // checkLivePeers checks that exactly wantPeers have a live route for the IPv4
 // prefix, and that the in-memory route count and prefix distribution match
 // the route table.
@@ -310,20 +567,7 @@ func checkLivePeers(
 ) {
 	t.Helper()
 
-	routes, _, err := db.GetPrefixDetailsContext(context.Background(), prefix)
-	if err != nil && !errors.Is(err, database.ErrNoRoute) {
-		t.Fatalf("GetPrefixDetailsContext(%s): %v", prefix, err)
-	}
-
-	gotPeers := make([]string, 0, len(routes))
-	for _, route := range routes {
-		gotPeers = append(gotPeers, route.PeerIP)
-	}
-
-	if !slices.Equal(gotPeers, wantPeers) {
-		t.Errorf("peers with a live route for %s = %v, want %v",
-			prefix, gotPeers, wantPeers)
-	}
+	assertPeers(t, db, map[string][]string{prefix: wantPeers})
 
 	var wantDistribution []database.PrefixDistribution
 	if len(wantPeers) > 0 {
@@ -335,6 +579,31 @@ func checkLivePeers(
 	assertCounts(t, db, ipv4Version, len(wantPeers), wantDistribution)
 }
 
+// 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)
+		}
+
+		got := make([]string, 0, len(routes))
+		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)
+		}
+	}
+}
+
 // checkLiveOrigins checks that each of prefixes has one live route, with the
 // origin AS wantOrigins gives for it, or none when wantOrigins leaves it out.
 func checkLiveOrigins(

internal/streamer/streamer.go

@@ -51,10 +51,12 @@ const (
 	minBackoffDelay   = 5 * time.Second
 	maxBackoffDelay   = 320 * time.Second
 	// Reconnect when no line arrives for this long
-	streamIdleTimeout     = 60 * time.Second
-	metricsLogInterval    = 10 * time.Second
-	bytesPerKB            = 1024
-	bytesPerMB            = 1024 * 1024
+	streamIdleTimeout  = 60 * time.Second
+	metricsLogInterval = 10 * time.Second
+	bytesPerKB         = 1024
+	bytesPerMB         = 1024 * 1024
+	// Well above an extended BGP message (RFC 8654) in hex plus its prefixes
+	maxLineSize           = bytesPerMB
 	maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers
 
 	// Backpressure constants
@@ -88,7 +90,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
 }
 
@@ -112,6 +115,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
@@ -618,6 +624,7 @@ func (s *Streamer) readStream(
 	ctx, connCtx context.Context, reader io.Reader, idleTimer *time.Timer,
 ) error {
 	scanner := bufio.NewScanner(reader)
+	scanner.Buffer(nil, maxLineSize)
 
 	for scanner.Scan() {
 		idleTimer.Reset(s.idleTimeout)
@@ -658,7 +665,7 @@ func (s *Streamer) readStream(
 }
 
 // processLine passes one line of the stream to the raw handler, parses it,
-// and dispatches an UPDATE message to the handlers. An error ends the
+// and dispatches an UPDATE or STATE message to the handlers. An error ends the
 // connection.
 func (s *Streamer) processLine(ctx context.Context, line []byte) error {
 	// Update metrics with decompressed message size
@@ -706,7 +713,7 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) 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":
@@ -745,9 +752,9 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
 		)
 
 		return nil
-	case "STATE":
-		// Peer state changes - silently ignore
-		return nil
+	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,
@@ -759,8 +766,8 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
 	return s.dispatchMessage(ctx, &msg)
 }
 
-// dispatchMessage queues msg for each handler that wants it, dropping it for
-// a handler whose queue is full or under backpressure.
+// dispatchMessage queues msg for each handler that wants it, unless
+// queueMessage drops it for that handler.
 func (s *Streamer) dispatchMessage(
 	ctx context.Context, msg *ristypes.RISMessage,
 ) error {
@@ -780,30 +787,20 @@ func (s *Streamer) dispatchMessage(
 			continue
 		}
 
-		// Check if we should drop due to backpressure
-		if s.shouldDropForBackpressure(info) {
+		if !s.queueMessage(info, msg) {
 			atomic.AddUint64(&info.metrics.droppedCount, 1)
 			s.totalDropped.Add(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)
-			s.totalDropped.Add(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()
@@ -811,6 +808,52 @@ func (s *Streamer) dispatchMessage(
 	return nil
 }
 
+// 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 {

internal/streamer/streamer_internal_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"runtime"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -129,6 +130,250 @@ func flush(w http.ResponseWriter) {
 	}
 }
 
+// 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) {
+	t.Parallel()
+
+	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{})
+
+	err := s.stream(context.Background())
+	if !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) {
+	t.Parallel()
+
+	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) {
+	t.Parallel()
+
+	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{})
+
+	err := s.stream(context.Background())
+	if !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)
+	}
+
+	got := len(s.handlers[1].queue)
+	if 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) {
+	t.Parallel()
+
+	const (
+		lines   = 2
+		maxWait = 5 * time.Second
+	)
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			_, _ = io.WriteString(w, stateDownLine+stateDownLine)
+
+			flush(w)
+			<-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
@@ -237,6 +482,42 @@ func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
 	}
 }
 
+// TestStreamReadsLongLine has the server send one UPDATE line of about
+// 200 KiB, padded with raw hex the way RIS Live writes an extended BGP
+// message, and then end the response. The line is longer than a
+// bufio.Scanner reads by default. The streamer must hand the message to the
+// handler's queue and read on until the server ends the stream, instead of
+// dropping the connection at the long line.
+func TestStreamReadsLongLine(t *testing.T) {
+	t.Parallel()
+
+	const rawHexLength = 200 * bytesPerKB
+
+	longLine := `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` +
+		`"peer_asn":"64496","timestamp":1700000000,"raw":"` +
+		strings.Repeat("0", rawHexLength) + `"}}` + "\n"
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = io.WriteString(w, longLine)
+		}))
+	defer srv.Close()
+
+	s := New(logger.New(), metrics.New())
+	s.url = srv.URL
+	s.RegisterHandler(updateHandler{})
+
+	err := s.stream(context.Background())
+	if !errors.Is(err, errStreamEnded) {
+		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+	}
+
+	queued := len(s.handlers[0].queue)
+	if queued != 1 {
+		t.Errorf("handler queue holds %d messages, want 1", queued)
+	}
+}
+
 // checkReconnect runs a streamer against a fake RIS Live server. The first
 // response is served by firstResponse; every later one carries one UPDATE
 // line and then stays open. The streamer must come back for a second