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