Test: both families reach a collector over one IPv6 session

A collector at an exchange is fed over one IPv6 session that carries
both families, with ADD-PATH for both and IPv4 routes with IPv6 next
hops (RFC 8950). The integration test for that has the collector on
IPv6 loopback with those capabilities and two feeders sending 20,000
IPv4 and 1,000 IPv6 prefixes each, and checks that every route of both
families arrives with its identifier. The test helper now reads the
prefixes that come and go in the multiprotocol attributes too.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

crates/integration-tests/tests/add_path.rs

@@ -10,9 +10,13 @@ use bgp_integration_tests::announcement::{Announcement, withdrawal};
 use bgp_integration_tests::test_peer::TestPeer;
 use bgp_integration_tests::test_server::TestServer;
 use bgp_integration_tests::{peer_config, prefix, wait_until};
-use bgp_packet::message::PathAttribute;
-use bgp_packet::open::{AddPathEntry, Capability, OpenMessage, SendReceive};
+use bgp_packet::message::{
+    AsPathAttribute, Message, MpReachNlriPathAttribute, NlriNextHop, OriginPathAttribute,
+    PathAttribute, UpdateMessage,
+};
+use bgp_packet::open::{AddPathEntry, Capability, ExtendedNextHopEntry, OpenMessage, SendReceive};
 use bgp_primitives::constants::AfiSafi;
+use bgp_primitives::constants::{Afi, Safi};
 use bgp_primitives::ip_prefix::Nlri;
 use bgp_rib::{LocalRoute, LocalRouteAttrs};
 use bgp_server::config::PeerConfig;
@@ -273,6 +277,22 @@ async fn told(
         for nlri in &update.announced_routes {
             held.insert((nlri.prefix.to_string(), nlri.path_id), path.clone());
         }
+        // And what came and went in the multiprotocol attributes.
+        for attribute in &update.path_attributes {
+            match attribute {
+                PathAttribute::MpReachNlri(reach) => {
+                    for nlri in &reach.prefixes {
+                        held.insert((nlri.prefix.to_string(), nlri.path_id), path.clone());
+                    }
+                }
+                PathAttribute::MpUnreachNlri(unreach) => {
+                    for nlri in &unreach.prefixes {
+                        held.remove(&(nlri.prefix.to_string(), nlri.path_id));
+                    }
+                }
+                _ => {}
+            }
+        }
     }
     Ok(())
 }
@@ -470,3 +490,118 @@ async fn test_a_large_table_reaches_a_collector_that_was_there_first() -> Result
     assert_eq!(ids.into_iter().collect::<Vec<_>>(), [1, 2]);
     Ok(())
 }
+
+/// As a router at an exchange feeds a collector: one IPv6 session that
+/// carries both families, with ADD-PATH for both and IPv4 routes with
+/// IPv6 next hops (RFC 8950), and a table of some size from two feeders.
+#[tokio::test]
+async fn test_both_families_reach_a_collector_over_one_ipv6_session() -> Result<()> {
+    const V4: u32 = 20_000;
+    const V6: u32 = 1_000;
+    let server = TestServer::start_with(LOCAL_ASN, |c| {
+        c.listen_endpoints
+            .push(SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 0));
+        c.peers = vec![PeerConfig {
+            add_path: Some(SendReceive::Send),
+            import: ImportPolicy::permissive(),
+            export: Arc::new(ExportPolicy {
+                learned_routes: true,
+                all_paths: true,
+                ..ExportPolicy::default()
+            }),
+            ..peer_config(
+                "collector",
+                SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 1),
+                COLLECTOR_ASN,
+            )
+        }];
+        c.connect_retry = Duration::from_secs(3600);
+    })
+    .await?;
+    let mut capabilities = vec![Capability::FourByteAsn { asn: COLLECTOR_ASN }];
+    capabilities.extend(SUPPORTED.into_iter().map(Capability::multiprotocol));
+    capabilities.push(Capability::AddPath {
+        entries: [AfiSafi::IPV4_UNICAST, AfiSafi::IPV6_UNICAST]
+            .into_iter()
+            .map(|family| AddPathEntry {
+                afi: family.afi,
+                safi: family.safi,
+                send_recv: SendReceive::Receive,
+            })
+            .collect(),
+    });
+    capabilities.push(Capability::ExtendedNextHop {
+        entries: vec![ExtendedNextHopEntry {
+            afi: Afi::IPV4,
+            safi: Safi::UNICAST,
+            nh_afi: Afi::IPV6,
+        }],
+    });
+    let mut collector = TestPeer::connect(server.addrs[1]).await?;
+    collector
+        .handshake_with(
+            COLLECTOR_ASN,
+            Ipv4Addr::new(10, 0, 0, 100),
+            90,
+            capabilities,
+        )
+        .await?;
+    collector.ctx.add_path = vec![AfiSafi::IPV4_UNICAST, AfiSafi::IPV6_UNICAST];
+    server.wait_for_sessions(1).await?;
+    let session = &server.registry.list()[0];
+    assert!(session.extended_next_hop, "RFC 8950 agreed");
+    assert_eq!(session.add_path_send.iter().count(), 2);
+
+    let v4 = |n: u32| {
+        let [_, b, c, _] = (0x0a00_0000u32 + n * 256).to_be_bytes();
+        prefix(&format!("10.{b}.{c}.0/24"))
+    };
+    let v6 = |n: u32| prefix(&format!("2001:db8:{:x}::/48", 0x1000 + n));
+    let mut feeders = Vec::new();
+    for n in 0..2u32 {
+        let mut peer = TestPeer::connect(server.addr).await?;
+        peer.handshake(PEER_ASN + n, Ipv4Addr::new(10, 0, 0, n as u8 + 1), 90)
+            .await?;
+        feeders.push(peer);
+    }
+    server.wait_for_sessions(3).await?;
+    for start in (0..V4).step_by(100) {
+        for (n, feeder) in feeders.iter_mut().enumerate() {
+            feeder
+                .announce(Announcement::of(
+                    (start..start + 100).map(|i| v4(i).into()),
+                    &[PEER_ASN + n as u32, 64510],
+                ))
+                .await?;
+        }
+    }
+    for start in (0..V6).step_by(100) {
+        for (n, feeder) in feeders.iter_mut().enumerate() {
+            let update = UpdateMessage {
+                withdrawn_routes: vec![],
+                path_attributes: vec![
+                    PathAttribute::Origin(OriginPathAttribute::IGP),
+                    AsPathAttribute::from_asns(vec![PEER_ASN + n as u32, 64510]),
+                    PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
+                        family: AfiSafi::IPV6_UNICAST,
+                        next_hop: NlriNextHop::Ipv6("2001:db8::9".parse().unwrap()),
+                        prefixes: (start..start + 100).map(|i| v6(i).into()).collect(),
+                    }),
+                ],
+                announced_routes: vec![],
+            };
+            feeder.send(&Message::Update(update)).await?;
+        }
+    }
+    let mut held = Held::new();
+    told(&mut collector, &mut held, |held| {
+        held.len() == 2 * (V4 + V6) as usize
+    })
+    .await?;
+    let v6_routes = held
+        .keys()
+        .filter(|(prefix, _)| prefix.contains(':'))
+        .count();
+    assert_eq!(v6_routes, 2 * V6 as usize);
+    Ok(())
+}