From 456aafde73a31c886f6e5a6a2cf0dcb0d2bd17b5 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 19 Oct 2026 08:18:58 +0000 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 those routes stayed in the live tables for good. The streamer now hands STATE messages to the handlers that want them. When the prefix handler gets one whose state is down, it drops the peer's updates still in its batch and notes the peer; the next flush deletes all of the peer's live routes, IPv4 and IPv6, before writing the batch, so updates the peer sends after the down are kept. The new store method keeps the in-memory route counts and prefix distribution exact, and an index on peer_ip lets it find the peer's routes without reading the whole table. Model: opus-5-5 --- README.md | 5 +- internal/database/counts_test.go | 76 +++++++++++++ internal/database/database.go | 84 ++++++++++++++ internal/database/interface.go | 1 + internal/database/schema.sql | 6 +- internal/ristypes/ris.go | 18 +++ internal/routewatch/app_integration_test.go | 6 + internal/routewatch/prefixhandler.go | 70 +++++++++--- internal/routewatch/prefixhandler_test.go | 115 ++++++++++++++++++++ internal/streamer/streamer.go | 8 +- internal/streamer/streamer_test.go | 46 ++++++++ 11 files changed, 414 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index 838f498..730044b 100644 --- a/README.md +++ b/README.md @@ -128,7 +128,10 @@ routewatch/ 4. **Handler Details** - **ASHandler**: Tracks all ASNs seen in AS paths - **PeerHandler**: Records BGP peer information - - **PrefixHandler**: Maintains live routing table (upserts on announcement, deletes on withdrawal) + - **PrefixHandler**: Maintains live routing table (upserts on announcement, deletes on withdrawal, + deletes all of a peer's routes when its BGP session goes down). Routes are kept per peer IP, so + when one peer IP has sessions on two collectors, a session going down on one also removes the + routes the other still carries, until they are announced again. - **PeeringHandler**: Extracts AS peering relationships from AS paths 5. **HTTP Request Flow** diff --git a/internal/database/counts_test.go b/internal/database/counts_test.go index b3259ba..d0be676 100644 --- a/internal/database/counts_test.go +++ b/internal/database/counts_test.go @@ -349,6 +349,82 @@ 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) + } + + if err := db.DeleteLiveRoutesOfPeer(peerIP); err != nil { + t.Fatalf("DeleteLiveRoutesOfPeer: %v", err) + } + + 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 6461336..4a6bdbf 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -530,6 +530,90 @@ 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. +func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) error { + d.lock("DeleteLiveRoutesOfPeer") + defer d.unlock() + + tx, err := d.beginTx() + if err != nil { + return 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 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 fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w", peerIP, err) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + + d.counts.addRoutes(-int(deletedV4), -int(deletedV6)) + d.counts.addToDistribution(goneMaskLengthsV4, goneMaskLengthsV6, -1) + + return 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..09aa094 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) 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 a45c855..c87ce21 100644 --- a/internal/database/schema.sql +++ b/internal/database/schema.sql @@ -130,6 +130,8 @@ CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origi CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated); -- Index to optimize prefix distribution queries CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix); +-- Index to find a peer's routes when its BGP session goes down +CREATE INDEX IF NOT EXISTS idx_live_routes_v4_peer_ip ON live_routes_v4(peer_ip); -- Indexes for live_routes_v6 table CREATE INDEX IF NOT EXISTS idx_live_routes_v6_prefix ON live_routes_v6(prefix); @@ -137,4 +139,6 @@ CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_asn ON live_routes_v6(origin_asn); CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated); -- Index to optimize prefix distribution queries -CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix); \ No newline at end of file +CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix); +-- Index to find a peer's routes when its BGP session goes down +CREATE INDEX IF NOT EXISTS idx_live_routes_v6_peer_ip ON live_routes_v6(peer_ip); \ No newline at end of file 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..d5234b8 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) error { + // Simple mock - just return nil + return nil +} + // GetOrCreateASNBatch mock implementation func (m *mockStore) GetOrCreateASNBatch(asns map[int]time.Time) error { m.mu.Lock() diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 5c76410..1081cc4 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" @@ -40,17 +41,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 +96,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 +109,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 +182,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 +219,47 @@ 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. + for _, peer := range peersDown { + if err := h.db.DeleteLiveRoutesOfPeer(peer); err != nil { + h.logger.Error("Failed to delete the live routes of a peer whose session went down", + "peer", peer, "error", err) + + continue + } + h.logger.Info("Deleted the live routes of a peer whose session went down", "peer", peer) + } + // 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 +335,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..3a7c378 100644 --- a/internal/routewatch/prefixhandler_test.go +++ b/internal/routewatch/prefixhandler_test.go @@ -288,6 +288,121 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) { } } +// TestPrefixHandlerDeletesAPeersRoutesWhenItsSessionGoesDown runs +// announcements from two peers through the prefix handler into a real +// database, then a batch in which one of the peers announces a prefix and its +// session goes down. The down deletes every live route of that peer, IPv4 and +// IPv6, the one announced just before it included, and keeps the other peer's +// routes for the same prefixes. An announcement the peer sends after the down, +// in the same batch, is kept. The in-memory route counts and prefix +// distribution must match the route tables. +func TestPrefixHandlerDeletesAPeersRoutesWhenItsSessionGoesDown(t *testing.T) { + const ( + otherPeerIP = "2001:db8:ffff::2" + // Both peers announce the shared prefixes, only testPeerIP the others. + sharedV4 = "198.51.100.0/24" + sharedV6 = "2001:db8:1::/48" + onlyV4 = "203.0.113.0/24" + onlyV6 = "2001:db8:2::/48" + beforeDown = "192.0.2.0/24" + afterDown = "2001:db8:3::/48" + // The mask lengths of the IPv4 and of the IPv6 prefixes. + maskLengthV4 = 24 + maskLengthV6 = 48 + ) + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + announce := func(peer string, prefixes ...string) *ristypes.RISMessage { + return &ristypes.RISMessage{ + Type: ristypes.MessageTypeUpdate, + Peer: peer, + Path: ristypes.ASPath{testASNA, testASNB}, + ParsedTimestamp: ts, + Announcements: []ristypes.RISAnnouncement{{NextHop: peer, Prefixes: prefixes}}, + } + } + down := &ristypes.RISMessage{ + Type: ristypes.MessageTypeState, + Peer: testPeerIP, + ParsedTimestamp: ts, + State: ristypes.StateDown, + } + + 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{announce(testPeerIP, beforeDown), down}, + }, + { + name: "the peer announces a prefix after the down", + batch: []*ristypes.RISMessage{ + announce(testPeerIP, beforeDown), down, announce(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(announce(testPeerIP, sharedV4, sharedV6, onlyV4, onlyV6)) + h.HandleMessage(announce(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++ + } + + 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(t, db, ipv4Version, 1, + []database.PrefixDistribution{{MaskLength: maskLengthV4, Count: 1}}) + assertCounts(t, db, ipv6Version, wantRoutesV6, + []database.PrefixDistribution{{MaskLength: maskLengthV6, Count: wantRoutesV6}}) + }) + } +} + // 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..13a3836 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -629,7 +629,7 @@ func (s *Streamer) stream(ctx context.Context) error { // Process based on message type switch msg.Type { - case "UPDATE": + case ristypes.MessageTypeUpdate: // Process BGP UPDATE messages // Will be dispatched to handlers case "RIS_PEER_STATE": @@ -668,9 +668,9 @@ func (s *Streamer) stream(ctx context.Context) error { ) continue - case "STATE": - // Peer state changes - silently ignore - continue + case ristypes.MessageTypeState: + // The BGP session with the peer changed state, for example went + // down. Dispatched to the handlers that want it. default: s.logger.Warn("Unknown message type, skipping", "type", msg.Type, diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_test.go index d2da82c..2bdab92 100644 --- a/internal/streamer/streamer_test.go +++ b/internal/streamer/streamer_test.go @@ -108,6 +108,52 @@ 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") + } +} + // 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