diff --git a/crates/integration-tests/tests/add_path.rs b/crates/integration-tests/tests/add_path.rs index 7c96002..b03b93e 100644 --- a/crates/integration-tests/tests/add_path.rs +++ b/crates/integration-tests/tests/add_path.rs @@ -389,3 +389,84 @@ async fn test_every_route_goes_to_a_collector_told_apart_by_path_identifiers() - ); Ok(()) } + +/// A collector that is connected before the table arrives is told the +/// whole table, however large: the walk of the table that a new follower +/// starts with takes many batches with a table of some size, and goes on +/// while the feeders keep sending. +#[tokio::test] +async fn test_a_large_table_reaches_a_collector_that_was_there_first() -> Result<()> { + const PREFIXES: u32 = 40_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(Ipv4Addr::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: vec![AddPathEntry { + afi: AfiSafi::IPV4_UNICAST.afi, + safi: AfiSafi::IPV4_UNICAST.safi, + send_recv: SendReceive::Receive, + }], + }); + let mut collector = TestPeer::connect(server.addr).await?; + collector + .handshake_with( + COLLECTOR_ASN, + Ipv4Addr::new(10, 0, 0, 100), + 90, + capabilities, + ) + .await?; + collector.ctx.add_path = vec![AfiSafi::IPV4_UNICAST]; + server.wait_for_sessions(1).await?; + + // Two feeders, each sending the same prefixes a hundred to an UPDATE. + let nth = |n: u32| { + let [_, b, c, _] = (0x0a00_0000u32 + n * 256).to_be_bytes(); + prefix(&format!("10.{b}.{c}.0/24")) + }; + let mut feeders = Vec::new(); + for n in 0..2u32 { + let mut peer = TestPeer::connect(server.addrs[1]).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..PREFIXES).step_by(100) { + for (n, feeder) in feeders.iter_mut().enumerate() { + feeder + .announce(Announcement::of( + (start..start + 100).map(|i| nth(i).into()), + &[PEER_ASN + n as u32, 64510], + )) + .await?; + } + } + let mut held = Held::new(); + told(&mut collector, &mut held, |held| { + held.len() == 2 * PREFIXES as usize + }) + .await?; + let ids: std::collections::BTreeSet = held.keys().map(|(_, id)| *id).collect(); + assert_eq!(ids.into_iter().collect::>(), [1, 2]); + Ok(()) +} diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 1eec721..0bdda15 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -768,6 +768,16 @@ impl Exporter { for item in news { match item { News::Reset => { + if learned.synced { + // The table was changing faster than this exporter + // read it, and the journal let the oldest changes + // go: the whole table is gone through again, and + // what is not heard of again is withdrawn then. + info!( + peer = %session.peer, + "The table changed too fast to follow for the peer; going through it again" + ); + } // Whatever was announced and is not heard of again by // the time the table is through is withdrawn then. let known = learned.announced.take(); @@ -945,6 +955,12 @@ impl Exporter { if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) { return Err(over_limit(session, limit, routes)); } + // Counted as it goes out, not after: a whole table takes a while + // to reach a peer that reads it at the pace of the Internet. + let note = |announced: usize| { + self.session.counters.set_advertised(announced); + self.advertised_metric.set(announced as i64); + }; for (version, routes) in [IpVersion::Ipv4, IpVersion::Ipv6] .into_iter() .zip(withdrawn) @@ -959,6 +975,12 @@ impl Exporter { session.send(Message::Update(update)).await?; } } + let before = routes + - batches + .values() + .map(|(_, routes)| routes.len()) + .sum::(); + let mut announced = 0; for ((_, version), (dressed, routes)) in batches { ROUTES_EXPORTED .with_label_values(&["announced"]) @@ -973,6 +995,8 @@ impl Exporter { for update in updates { session.send(Message::Update(update)).await?; } + announced += routes.len(); + note(before + announced); } self.note_advertised(); Ok(())