Count every peer the feed has heard from in bgp_peer_count (closes #74)

bgp_peer_count grew only on OPEN messages, so it started at 0 while
updates arrived from hundreds of peers whose sessions were already up, and
it kept peers whose sessions had gone down. Now any message from a peer
adds it to the streamer's peer set, and a NOTIFICATION or a STATE message
whose state is down removes it without adding it first. The rule is in
updatePeers, called for every message.

Model: opus-5-5

internal/streamer/streamer.go

@@ -64,6 +64,10 @@ const (
 	backpressureSlope     = 2.0 // Slope for linear drop probability increase
 )
 
+// messageTypeNotification is the type of a BGP NOTIFICATION message, which ends
+// the session with the peer.
+const messageTypeNotification = "NOTIFICATION"
+
 var (
 	errAlreadyRunning   = errors.New("streamer already running")
 	errUnexpectedStatus = errors.New("unexpected status code")
@@ -665,8 +669,8 @@ func (s *Streamer) readStream(
 }
 
 // processLine passes one line of the stream to the raw handler, parses it,
-// and dispatches an UPDATE or STATE message to the handlers. An error ends the
-// connection.
+// updates the peer set (updatePeers), and dispatches an UPDATE or STATE
+// message to the handlers. An error ends the connection.
 func (s *Streamer) processLine(ctx context.Context, line []byte) error {
 	// Update metrics with decompressed message size
 	s.updateMetrics(len(line))
@@ -711,6 +715,8 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
 	// Parse the timestamp
 	msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()
 
+	peerCount := s.updatePeers(&msg)
+
 	// Process based on message type
 	switch msg.Type {
 	case ristypes.MessageTypeUpdate:
@@ -723,13 +729,6 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
 		// BGP keepalive messages - silently process
 		return nil
 	case "OPEN":
-		// BGP open messages - track peer as active
-		s.bgpPeersMu.Lock()
-		s.bgpPeers[msg.Peer] = true
-		peerCount := len(s.bgpPeers)
-		s.bgpPeersMu.Unlock()
-		s.metrics.SetBGPPeerCount(peerCount)
-
 		s.logger.Info("BGP session opened",
 			"peer", msg.Peer,
 			"peer_asn", msg.PeerASN,
@@ -737,14 +736,7 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
 		)
 
 		return nil
-	case "NOTIFICATION":
-		// BGP notification messages (session closed)
-		s.bgpPeersMu.Lock()
-		delete(s.bgpPeers, msg.Peer)
-		peerCount := len(s.bgpPeers)
-		s.bgpPeersMu.Unlock()
-		s.metrics.SetBGPPeerCount(peerCount)
-
+	case messageTypeNotification:
 		s.logger.Warn("BGP notification",
 			"peer", msg.Peer,
 			"peer_asn", msg.PeerASN,
@@ -766,6 +758,29 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
 	return s.dispatchMessage(ctx, &msg)
 }
 
+// updatePeers adds the peer msg came from to the peer set, or removes it if
+// msg ends the peer's session: a NOTIFICATION, or a STATE message whose state
+// is down. It records the size of the set as the BGP peer count and returns
+// it.
+func (s *Streamer) updatePeers(msg *ristypes.RISMessage) int {
+	sessionEnded := msg.Type == messageTypeNotification ||
+		(msg.Type == ristypes.MessageTypeState && msg.State == ristypes.StateDown)
+
+	s.bgpPeersMu.Lock()
+
+	if sessionEnded {
+		delete(s.bgpPeers, msg.Peer)
+	} else {
+		s.bgpPeers[msg.Peer] = true
+	}
+
+	peerCount := len(s.bgpPeers)
+	s.bgpPeersMu.Unlock()
+	s.metrics.SetBGPPeerCount(peerCount)
+
+	return peerCount
+}
+
 // dispatchMessage queues msg for each handler that wants it, unless
 // queueMessage drops it for that handler.
 func (s *Streamer) dispatchMessage(