From 825afa2ebe51ffd805ce7bdd8ce7d489f310628a Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 19 Oct 2026 11:40:31 +0200 Subject: [PATCH] Remove a peer's live routes when its session goes down (closes #56) RIS sends no withdrawals for the routes of a peer whose BGP session goes down, so routewatch kept them. The streamer now hands STATE messages to the prefix handler. They are never dropped for backpressure: one that finds the queue full is held and queued first once there is room, and the handler's other messages are dropped meanwhile, so none overtakes it. Nothing waits. On a down, the prefix handler drops the peer's pending updates and its next flush deletes all the peer's live routes before the batch's other writes, keeping the in-memory counts exact. The live route tables are unique on peer and prefix, which also finds a peer's routes. Model: opus-5-5 --- README.md | 9 +- internal/database/counts_test.go | 83 +++++++ internal/database/database.go | 85 +++++++ internal/database/interface.go | 1 + internal/database/schema.sql | 8 +- internal/ristypes/ris.go | 18 ++ internal/routewatch/app_integration_test.go | 6 + internal/routewatch/peerhandler.go | 14 +- internal/routewatch/prefixhandler.go | 83 +++++-- internal/routewatch/prefixhandler_test.go | 251 ++++++++++++++++++++ internal/streamer/streamer.go | 82 +++++-- internal/streamer/streamer_test.go | 198 +++++++++++++++ 12 files changed, 787 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index 838f498..0c2adac 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** @@ -232,7 +235,9 @@ What happens at each limit: out-of-memory error, and the handler logs it and drops that batch. The process keeps running. - Handler queues: each of the four handler queues holds at most 20,000 messages. - When a queue fills, the streamer drops messages instead of blocking. + When a queue fills, the streamer drops messages instead of blocking, except a + `STATE` message, which it holds, in feed order, until there is room: a few + hundred bytes each, about ten a second. - Handler batches: while the AS, peer or prefix handler writes a batch to the database, the updates that arrive collect in its next batch, which stops at ten times the batch size: 300,000 updates for the AS handler 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..61510f0 100644 --- a/internal/routewatch/peerhandler.go +++ b/internal/routewatch/peerhandler.go @@ -1,7 +1,7 @@ package routewatch // peerhandler.go provides batched peer tracking functionality for BGP route monitoring. -// It tracks BGP peers from all incoming RIS messages and maintains peer state in the database. +// It tracks BGP peers from incoming UPDATE messages and maintains peer state in the database. import ( "strconv" @@ -33,7 +33,7 @@ const ( peerBatchTimeout = 2 * time.Second ) -// PeerHandler tracks BGP peers from all message types using batched operations. +// PeerHandler tracks BGP peers from UPDATE messages using batched operations. // It maintains a queue of peer updates and periodically flushes them to the database // in batches to improve performance. The handler deduplicates peer updates within // each batch, keeping only the most recent update for each peer IP address. @@ -82,11 +82,11 @@ func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler { return h } -// WantsMessage returns true for all message types since peer information -// is extracted from every RIS message regardless of type. This satisfies -// the MessageHandler interface. -func (h *PeerHandler) WantsMessage(_ string) bool { - return true +// WantsMessage returns true for UPDATE messages only. A peer whose session +// never comes up sends STATE messages and nothing else, so a STATE message is +// not a sign of a peer that carries routes. +func (h *PeerHandler) WantsMessage(messageType string) bool { + return messageType == ristypes.MessageTypeUpdate } // QueueCapacity returns the desired queue capacity for this handler. diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 5c76410..4aec6ca 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 is held until + // there is room), so this bounds memory. Batches still flush on a timer + // (prefixBatchTimeout). prefixHandlerQueueSize = 20000 // prefixBatchSize is the number of prefix updates to batch together @@ -40,17 +42,19 @@ const ( ) // PrefixHandler tracks BGP prefixes and maintains a live routing table in the database. -// Routes are added on announcement and deleted on withdrawal. +// Routes are added on announcement and deleted on withdrawal, and all routes of +// a peer are deleted when its BGP session goes down. type PrefixHandler struct { db database.Store logger *logger.Logger metrics *metrics.Tracker - // Batching. mu guards batch only: the flush goroutine takes the batch - // under mu and writes it after releasing mu, so HandleMessage waits for - // a database write only while batch is at prefixBatchMaxSize. + // Batching. mu guards batch and peersDown only: the flush goroutine takes + // both under mu and writes them after releasing mu, so HandleMessage + // waits for a database write only while batch is at prefixBatchMaxSize. mu sync.Mutex batch []prefixUpdate + peersDown []string // peers whose BGP session went down since the batch was taken batchTaken sync.Cond // signalled when the flush goroutine takes batch; its L is &mu batchFullCh chan struct{} // wakes the flush goroutine when batch is full lastFlush time.Time // used only by the flush goroutine @@ -93,8 +97,9 @@ func (h *PrefixHandler) SetMetricsTracker(metrics *metrics.Tracker) { // WantsMessage returns true if this handler wants to process messages of the given type func (h *PrefixHandler) WantsMessage(messageType string) bool { - // We only care about UPDATE messages for the routing table - return messageType == "UPDATE" + // UPDATE messages change routes; a STATE message can say a peer's + // session went down, which removes all of its routes. + return messageType == ristypes.MessageTypeUpdate || messageType == ristypes.MessageTypeState } // QueueCapacity returns the desired queue capacity for this handler @@ -105,6 +110,14 @@ func (h *PrefixHandler) QueueCapacity() int { // HandleMessage processes a message to track prefix information func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { + if msg.Type == ristypes.MessageTypeState { + if msg.State == ristypes.StateDown { + h.peerSessionDown(msg.Peer) + } + + return + } + // Use the pre-parsed timestamp timestamp := msg.ParsedTimestamp @@ -170,6 +183,19 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { } } +// peerSessionDown notes that the peer's BGP session went down, so that the next +// flush deletes all of the peer's live routes before it writes the batch. The +// peer's updates still in the batch came before the down, which replaces them, +// so they are dropped; updates the peer sends later stay in the batch and are +// written after the delete. +func (h *PrefixHandler) peerSessionDown(peer string) { + h.mu.Lock() + defer h.mu.Unlock() + + h.batch = slices.DeleteFunc(h.batch, func(update prefixUpdate) bool { return update.peer == peer }) + h.peersDown = append(h.peersDown, peer) +} + // flushLoop runs in a goroutine, the only one that writes the batch: it // flushes when the batch is full, periodically, and once more on Stop. func (h *PrefixHandler) flushLoop() { @@ -194,33 +220,53 @@ func (h *PrefixHandler) flushLoop() { } } -// takeBatch returns the pending updates and leaves an empty batch in their -// place for HandleMessage to fill. -func (h *PrefixHandler) takeBatch() []prefixUpdate { +// takeBatch returns the pending updates and the peers whose session went down +// since the last flush, and leaves an empty batch and no peers in their place +// for HandleMessage to fill. +func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) { h.mu.Lock() defer h.mu.Unlock() - if len(h.batch) == 0 { - return nil + if len(h.batch) == 0 && len(h.peersDown) == 0 { + return nil, nil } - batch := h.batch + batch, peersDown := h.batch, h.peersDown h.batch = make([]prefixUpdate, 0, prefixBatchSize) + h.peersDown = nil h.batchTaken.Broadcast() - return batch + return batch, peersDown } // flushBatch takes the pending batch and writes it to the database, without // holding the mutex during the write. func (h *PrefixHandler) flushBatch() { - batch := h.takeBatch() - if len(batch) == 0 { + batch, peersDown := h.takeBatch() + if len(batch) == 0 && len(peersDown) == 0 { return } startTime := time.Now() batchSize := len(batch) + // The routes of a peer whose session went down are deleted first: the + // batch holds only the updates the peer sent after the down. Most downs + // come from peers that have no routes, so only a delete that removed + // routes is logged. + for _, peer := range peersDown { + deleted, err := h.db.DeleteLiveRoutesOfPeer(peer) + if err != nil { + h.logger.Error("Failed to delete the live routes of a peer whose session went down", + "peer", peer, "error", err) + + continue + } + if deleted > 0 { + h.logger.Info("Deleted the live routes of a peer whose session went down", + "peer", peer, "routes", deleted) + } + } + // For each prefix and peer, keep only the last update the feed delivered. // Updates from different peers for the same prefix are all kept. Arrival // order decides, not the timestamp: timestamps are whole seconds, so @@ -296,6 +342,7 @@ func (h *PrefixHandler) flushBatch() { elapsed := time.Since(startTime) h.logger.Debug("Flushed prefix batch", "batch_size", batchSize, + "peers_down", len(peersDown), "unique_prefixes", len(prefixesToUpdate), "success", successCount, "duration_ms", elapsed.Milliseconds(), diff --git a/internal/routewatch/prefixhandler_test.go b/internal/routewatch/prefixhandler_test.go index c1ce235..2575a49 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,256 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) { } } +// TestStateMessagesGoToThePrefixHandlerNotThePeerHandler checks that STATE +// messages go to the prefix handler, which acts on a down, and not to the peer +// handler, which would record a peer whose session never comes up as active. +func TestStateMessagesGoToThePrefixHandlerNotThePeerHandler(t *testing.T) { + if !(&PrefixHandler{}).WantsMessage(ristypes.MessageTypeState) { + t.Error("the prefix handler does not want STATE messages") + } + if (&PeerHandler{}).WantsMessage(ristypes.MessageTypeState) { + t.Error("the peer handler wants STATE messages") + } + if !(&PeerHandler{}).WantsMessage(ristypes.MessageTypeUpdate) { + t.Error("the peer handler does not want UPDATE messages") + } +} + +// TestPrefixHandlerDeletesAPeersRoutesWhenItsSessionGoesDown runs +// announcements from two peers through the prefix handler into a real +// database, then a batch in which one of the peers announces a prefix and its +// session goes down. The down deletes every live route of that peer, IPv4 and +// IPv6, the one announced just before it included, and keeps the other peer's +// routes for the same prefixes. An announcement the peer sends after the down, +// in the same batch, is kept. The in-memory route counts and prefix +// distribution must match the route tables. +func TestPrefixHandlerDeletesAPeersRoutesWhenItsSessionGoesDown(t *testing.T) { + const ( + otherPeerIP = "2001:db8:ffff::2" + // Both peers announce the shared prefixes, only testPeerIP the others. + sharedV4 = "198.51.100.0/24" + sharedV6 = "2001:db8:1::/48" + onlyV4 = "203.0.113.0/24" + onlyV6 = "2001:db8:2::/48" + beforeDown = "192.0.2.0/24" + afterDown = "2001:db8:3::/48" + // The mask lengths of the IPv4 and of the IPv6 prefixes. + maskLengthV4 = 24 + maskLengthV6 = 48 + ) + + tests := []struct { + name string + // Handled and flushed as one batch after both peers' announcements. + batch []*ristypes.RISMessage + // Whether testPeerIP is left with a live route for afterDown. + wantAfterDown bool + }{ + { + name: "the down ends the batch", + batch: []*ristypes.RISMessage{peerAnnouncement(testPeerIP, beforeDown), sessionDown(testPeerIP)}, + }, + { + name: "the peer announces a prefix after the down", + batch: []*ristypes.RISMessage{ + peerAnnouncement(testPeerIP, beforeDown), + sessionDown(testPeerIP), + peerAnnouncement(testPeerIP, afterDown), + }, + wantAfterDown: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + // Built without NewPrefixHandler's flush timer, so the test flushes itself. + h := &PrefixHandler{db: db, logger: logger.New()} + h.HandleMessage(peerAnnouncement(testPeerIP, sharedV4, sharedV6, onlyV4, onlyV6)) + h.HandleMessage(peerAnnouncement(otherPeerIP, sharedV4, sharedV6)) + h.flushBatch() + for _, msg := range tt.batch { + h.HandleMessage(msg) + } + h.flushBatch() + + // The peers left with a live route for each prefix. + wantPeers := map[string][]string{ + sharedV4: {otherPeerIP}, + sharedV6: {otherPeerIP}, + onlyV4: nil, + onlyV6: nil, + beforeDown: nil, + afterDown: nil, + } + wantRoutesV6 := 1 + if tt.wantAfterDown { + wantPeers[afterDown] = []string{testPeerIP} + wantRoutesV6++ + } + + assertPeers(t, db, wantPeers) + assertCounts(t, db, ipv4Version, 1, + []database.PrefixDistribution{{MaskLength: maskLengthV4, Count: 1}}) + assertCounts(t, db, ipv6Version, wantRoutesV6, + []database.PrefixDistribution{{MaskLength: maskLengthV6, Count: wantRoutesV6}}) + }) + } +} + +// TestPrefixHandlerKeepsWhatAPeerAnnouncesAfterTheFlushOfItsDown runs, through +// the prefix handler into a real database, one flush after another: a peer's +// announcement, its session going down, a new announcement from the peer, and +// another peer's announcement. The down deletes the peer's routes on the flush +// that handles it and on no later flush, so the route the peer announced after +// it is kept. +func TestPrefixHandlerKeepsWhatAPeerAnnouncesAfterTheFlushOfItsDown(t *testing.T) { + const ( + otherPeerIP = "2001:db8:ffff::2" + beforeDown = "192.0.2.0/24" + afterDown = "198.51.100.0/24" + otherPeers = "203.0.113.0/24" + maskLength = 24 + ) + + db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + // Built without NewPrefixHandler's flush timer, so the test flushes itself. + h := &PrefixHandler{db: db, logger: logger.New()} + for _, msg := range []*ristypes.RISMessage{ + peerAnnouncement(testPeerIP, beforeDown), + sessionDown(testPeerIP), + peerAnnouncement(testPeerIP, afterDown), + peerAnnouncement(otherPeerIP, otherPeers), + } { + h.HandleMessage(msg) + h.flushBatch() + } + + assertPeers(t, db, map[string][]string{ + beforeDown: nil, + afterDown: {testPeerIP}, + otherPeers: {otherPeerIP}, + }) + assertCounts(t, db, ipv4Version, 2, []database.PrefixDistribution{{MaskLength: maskLength, Count: 2}}) +} + +// firstWriteHeldStore is a real database whose first live route batch write +// waits until the test closes release. writeStarted is closed when it starts. +type firstWriteHeldStore struct { + *database.Database + + writeStarted chan struct{} + release chan struct{} + once sync.Once +} + +func (s *firstWriteHeldStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error { + s.once.Do(func() { + close(s.writeStarted) + <-s.release + }) + + return s.Database.UpsertLiveRouteBatch(routes) +} + +// TestPrefixHandlerHandlesADownWhileAFlushWrites holds the write of a flush +// that stores a peer's route and meanwhile hands the prefix handler the peer's +// session going down, then a new announcement from the peer. The next flush +// must delete the route the held write stored and keep the one announced after +// the down. +func TestPrefixHandlerHandlesADownWhileAFlushWrites(t *testing.T) { + const ( + beforeDown = "192.0.2.0/24" + afterDown = "198.51.100.0/24" + maskLength = 24 + ) + + db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + store := &firstWriteHeldStore{Database: db, writeStarted: make(chan struct{}), release: make(chan struct{})} + + // Built without NewPrefixHandler's flush timer, so the test flushes itself. + h := &PrefixHandler{db: store, logger: logger.New()} + h.HandleMessage(peerAnnouncement(testPeerIP, beforeDown)) + flushed := make(chan struct{}) + go func() { + h.flushBatch() + close(flushed) + }() + select { + case <-store.writeStarted: + case <-time.After(handleMessageTimeout): + t.Fatal("the flush did not start its write") + } + + if !returnsInTime(func() { + h.HandleMessage(sessionDown(testPeerIP)) + h.HandleMessage(peerAnnouncement(testPeerIP, afterDown)) + }) { + t.Fatal("HandleMessage did not return while a database write was held") + } + close(store.release) + select { + case <-flushed: + case <-time.After(handleMessageTimeout): + t.Fatal("the flush did not finish after its write was let go") + } + h.flushBatch() + + assertPeers(t, db, map[string][]string{beforeDown: nil, afterDown: {testPeerIP}}) + assertCounts(t, db, ipv4Version, 1, []database.PrefixDistribution{{MaskLength: maskLength, Count: 1}}) +} + +// peerAnnouncement returns an UPDATE message in which peer announces prefixes +// with origin AS testASNB. +func peerAnnouncement(peer string, prefixes ...string) *ristypes.RISMessage { + return &ristypes.RISMessage{ + Type: ristypes.MessageTypeUpdate, + Peer: peer, + Path: ristypes.ASPath{testASNA, testASNB}, + ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), + Announcements: []ristypes.RISAnnouncement{{NextHop: peer, Prefixes: prefixes}}, + } +} + +// sessionDown returns the STATE message that says peer's BGP session went down. +func sessionDown(peer string) *ristypes.RISMessage { + return &ristypes.RISMessage{Type: ristypes.MessageTypeState, Peer: peer, State: ristypes.StateDown} +} + +// assertPeers checks, for each prefix in wantPeers, that the peers with a live +// route for it are the ones listed, in peer IP order; nil means none. +func assertPeers(t *testing.T, db *database.Database, wantPeers map[string][]string) { + t.Helper() + + for prefix, want := range wantPeers { + routes, _, err := db.GetPrefixDetailsContext(context.Background(), prefix) + if err != nil && !errors.Is(err, database.ErrNoRoute) { + t.Fatalf("GetPrefixDetailsContext(%s): %v", prefix, err) + } + var got []string + for _, route := range routes { + got = append(got, route.PeerIP) + } + if !slices.Equal(got, want) { + t.Errorf("peers with a live route for %s = %v, want %v", prefix, got, want) + } + } +} + // assertCounts checks that the live route count and prefix distribution of one // IP version are wantRoutes and wantDistribution, both as held in memory and as // read from the route table. diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index 04e4429..6e4e18a 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -78,7 +78,8 @@ type MessageHandler interface { // QueueCapacity returns the desired queue capacity for this handler. // Handlers that process quickly can have larger queues to buffer bursts. // When the queue fills up, messages will be dropped according to the - // backpressure algorithm. + // backpressure algorithm, except STATE messages, which are held until + // there is room. QueueCapacity() int } @@ -102,6 +103,9 @@ type handlerInfo struct { handler MessageHandler queue chan *ristypes.RISMessage metrics handlerMetrics + // heldState holds, in feed order, the STATE messages that found queue + // full (see queueMessage). Only the read loop uses it. + heldState []*ristypes.RISMessage } // Streamer manages a connection to the RIPE RIS Live streaming API for receiving @@ -629,7 +633,7 @@ func (s *Streamer) stream(ctx context.Context) error { // Process based on message type switch msg.Type { - case "UPDATE": + case ristypes.MessageTypeUpdate: // Process BGP UPDATE messages // Will be dispatched to handlers case "RIS_PEER_STATE": @@ -668,9 +672,9 @@ func (s *Streamer) stream(ctx context.Context) error { ) continue - case "STATE": - // Peer state changes - silently ignore - continue + case ristypes.MessageTypeState: + // The BGP session with the peer changed state, for example went + // down. Dispatched to the handlers that want it. default: s.logger.Warn("Unknown message type, skipping", "type", msg.Type, @@ -693,30 +697,20 @@ func (s *Streamer) stream(ctx context.Context) error { continue } - // Check if we should drop due to backpressure - if s.shouldDropForBackpressure(info) { + if !s.queueMessage(info, &msg) { atomic.AddUint64(&info.metrics.droppedCount, 1) atomic.AddUint64(&s.totalDropped, 1) continue } - // Try to queue the message - select { - case info.queue <- &msg: - // Message queued successfully - // Update high water mark if needed - queueLen := len(info.queue) - info.metrics.mu.Lock() - if queueLen > info.metrics.queueHighWaterMark { - info.metrics.queueHighWaterMark = queueLen - } - info.metrics.mu.Unlock() - default: - // Queue is full, drop the message - atomic.AddUint64(&info.metrics.droppedCount, 1) - atomic.AddUint64(&s.totalDropped, 1) + // Update high water mark if needed + queueLen := len(info.queue) + info.metrics.mu.Lock() + if queueLen > info.metrics.queueHighWaterMark { + info.metrics.queueHighWaterMark = queueLen } + info.metrics.mu.Unlock() } s.mu.RUnlock() } @@ -736,6 +730,50 @@ func (s *Streamer) stream(ctx context.Context) error { return errStreamEnded } +// queueMessage puts msg on the handler's queue and reports false if it dropped +// msg instead. It never waits for room, so a full queue holds up neither the +// read loop nor Stop. +// +// A message is dropped while the queue is filling up (shouldDropForBackpressure) +// or full, except a STATE message: one whose state is down is the only notice +// that all of a peer's routes are gone. A STATE message skips the random drop, +// and one that finds the queue full is held in info.heldState. Held messages +// go into the queue, oldest first, before anything else for the handler, as +// soon as there is room; while any is still held, the handler's other messages +// are dropped, so none of them overtakes a held down. +// +// The held list stays small: a STATE message is a few hundred bytes and the +// feed sends about ten a second, so a queue full for a minute holds about 600. +func (s *Streamer) queueMessage(info *handlerInfo, msg *ristypes.RISMessage) bool { + for len(info.heldState) > 0 && trySend(info.queue, info.heldState[0]) { + info.heldState = info.heldState[1:] + } + + if msg.Type == ristypes.MessageTypeState { + if len(info.heldState) > 0 || !trySend(info.queue, msg) { + info.heldState = append(info.heldState, msg) + } + + return true + } + + if len(info.heldState) > 0 || s.shouldDropForBackpressure(info) { + return false + } + + return trySend(info.queue, msg) +} + +// trySend puts msg on queue if there is room and reports whether it did. +func trySend(queue chan *ristypes.RISMessage, msg *ristypes.RISMessage) bool { + select { + case queue <- msg: + return true + default: + return false + } +} + // shouldDropForBackpressure determines if a message should be dropped based on queue utilization func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool { // Calculate queue utilization diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_test.go index d2da82c..90e355c 100644 --- a/internal/streamer/streamer_test.go +++ b/internal/streamer/streamer_test.go @@ -108,6 +108,204 @@ func (h updateHandler) HandleMessage(*ristypes.RISMessage) { } } +// stateDownLine is the STATE message RIS Live sends when the collector's BGP +// session with the peer stateDownPeer goes down. +const ( + stateDownPeer = "192.0.2.1" + stateDownLine = `{"type":"ris_message","data":{"type":"STATE","peer":"` + stateDownPeer + `",` + + `"peer_asn":"64496","timestamp":1700000000,"state":"down"}}` + "\n" +) + +// stateHandler wants STATE messages only. Tests read what it is handed from +// its queue, so HandleMessage does nothing. +type stateHandler struct{} + +func (stateHandler) WantsMessage(messageType string) bool { + return messageType == ristypes.MessageTypeState +} +func (stateHandler) QueueCapacity() int { return 10 } +func (stateHandler) HandleMessage(*ristypes.RISMessage) {} + +// TestStreamHandsStateMessagesToHandlersThatWantThem has the server send one +// STATE message whose state is down. The streamer must queue it, with its +// peer and state, for a handler that wants STATE messages. +func TestStreamHandsStateMessagesToHandlersThatWantThem(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, stateDownLine) + })) + defer srv.Close() + + s := New(logger.New(), metrics.New()) + s.url = srv.URL + s.RegisterHandler(stateHandler{}) + + if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) { + t.Fatalf("stream returned %v, want %v", err, errStreamEnded) + } + + select { + case msg := <-s.handlers[0].queue: + if msg.Type != ristypes.MessageTypeState || msg.Peer != stateDownPeer || msg.State != ristypes.StateDown { + t.Errorf("queued message has type %q, peer %q, state %q; want %q, %q, %q", + msg.Type, msg.Peer, msg.State, ristypes.MessageTypeState, stateDownPeer, ristypes.StateDown) + } + default: + t.Fatal("the STATE message was not queued for the handler that wants it") + } +} + +// TestHeldStateMessagesKeepTheirPlaceInTheQueue hands queueMessage two STATE +// messages and two updates while the queue is full, making room in between. +// Neither STATE message may be dropped: each must reach the queue after +// everything queued before it and before anything handed over after it. The +// update handed over while a STATE message is held is dropped. +func TestHeldStateMessagesKeepTheirPlaceInTheQueue(t *testing.T) { + s := New(logger.New(), metrics.New()) + s.RegisterHandler(stateHandler{}) + info := s.handlers[0] + + before := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate} + for len(info.queue) < cap(info.queue) { + info.queue <- before + } + firstDown := &ristypes.RISMessage{Type: ristypes.MessageTypeState, State: ristypes.StateDown} + secondDown := &ristypes.RISMessage{Type: ristypes.MessageTypeState, State: ristypes.StateDown} + updateWhileHeld := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate} + after := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate} + + if !s.queueMessage(info, firstDown) { + t.Fatal("the first STATE message was dropped on a full queue") + } + <-info.queue // the handler makes room for one message + if !s.queueMessage(info, secondDown) { + t.Fatal("the second STATE message was dropped") + } + if s.queueMessage(info, updateWhileHeld) { + t.Fatal("an update was queued while a STATE message was held") + } + + // The handler takes every message queued before the first STATE message. + for range cap(info.queue) - 1 { + if <-info.queue != before { + t.Fatal("a message overtook those queued before the first STATE message") + } + } + if !s.queueMessage(info, after) { + t.Fatal("an update was dropped on a queue with room") + } + + if <-info.queue != firstDown { + t.Fatal("the first STATE message did not come next") + } + if <-info.queue != secondDown { + t.Fatal("the second STATE message did not come after the first") + } + if <-info.queue != after { + t.Fatal("the update handed over after the STATE messages did not come after them") + } + if len(info.queue) != 0 { + t.Errorf("%d messages left on the queue, want none", len(info.queue)) + } +} + +// newStreamerWithFullStateQueue returns a streamer reading from url with one +// handler, a stateHandler whose queue is full. +func newStreamerWithFullStateQueue(url string) *Streamer { + s := New(logger.New(), metrics.New()) + s.url = url + s.RegisterHandler(stateHandler{}) + queue := s.handlers[0].queue + for len(queue) < cap(queue) { + queue <- &ristypes.RISMessage{Type: ristypes.MessageTypeState} + } + + return s +} + +// TestOtherHandlersKeepReceivingWhileAStateMessageIsHeld has the server send a +// STATE message whose state is down, then an UPDATE, while the queue of the +// handler that wants STATE messages is full. The STATE message must be held +// for that handler, and the read loop must go on to queue the UPDATE for the +// handler that wants it. +func TestOtherHandlersKeepReceivingWhileAStateMessageIsHeld(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, stateDownLine+updateLine) + })) + defer srv.Close() + + s := newStreamerWithFullStateQueue(srv.URL) + s.RegisterHandler(updateHandler{}) + + if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) { + t.Fatalf("stream returned %v, want %v", err, errStreamEnded) + } + + held := s.handlers[0].heldState + if len(held) != 1 || held[0].Peer != stateDownPeer || held[0].State != ristypes.StateDown { + t.Errorf("held %d STATE messages, want the down from %s alone", len(held), stateDownPeer) + } + if got := len(s.handlers[1].queue); got != 1 { + t.Errorf("%d messages queued for the handler that wants UPDATE messages, want 1", got) + } +} + +// TestStopReturnsAtOnceWithStateMessagesHeld has the server send two STATE +// messages and hold the response open while the queue of the handler that +// wants them is full. Stop must return at once and end the stream. +func TestStopReturnsAtOnceWithStateMessagesHeld(t *testing.T) { + const ( + lines = 2 + maxWait = 5 * time.Second + ) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, stateDownLine+stateDownLine) + w.(http.Flusher).Flush() + <-r.Context().Done() + })) + defer srv.Close() + + s := newStreamerWithFullStateQueue(srv.URL) + // The raw handler sees each line before the read loop hands it to the + // queues, so once it has seen the second line, the first is held. + linesSeen := make(chan struct{}, lines) + s.RegisterRawHandler(func(string) { linesSeen <- struct{}{} }) + + // As in TestStopBeforeMessageReachesQueues, setting cancel as Start does + // lets Stop cancel the stream run here. + ctx, cancel := context.WithCancel(context.Background()) + s.cancel = cancel + streamErr := make(chan error, 1) + go func() { streamErr <- s.stream(ctx) }() + + for range lines { + select { + case <-linesSeen: + case <-time.After(maxWait): + t.Fatalf("the read loop did not read %d lines within %s", lines, maxWait) + } + } + + stopped := make(chan struct{}) + go func() { + s.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(maxWait): + t.Fatalf("Stop did not return within %s while a STATE message was held", maxWait) + } + + select { + case err := <-streamErr: + if !errors.Is(err, context.Canceled) { + t.Errorf("stream returned %v, want %v", err, context.Canceled) + } + case <-time.After(maxWait): + t.Fatalf("stream did not end within %s of Stop", maxWait) + } +} + // TestStopBeforeMessageReachesQueues stops the streamer after the read loop // has checked for cancellation but before it hands the message to the handler // queues. That is the gap a stop from another goroutine can land in, and it