export: count routes as they go out, and say when the table is walked again

The number of routes advertised to a peer was written once a flush was
over, after every UPDATE of it had been handed to the session. The first
flush to a route collector is the whole table, two million routes or
more to a peer that reads at the pace of the Internet, so the status
page showed the peer holding one route for as long as that took, which
read as nothing going out.

Now the count is written after each batch of a flush, so that the page
and `bgp_peer_advertised_routes` show the table going out.

Also, an exporter that had the whole table and is given it again,
because the table changed faster than the exporter read it and the
journal let the oldest changes go, now logs that it is going through the
table again, at info. Until now that was invisible, and a collector
session during a full-table load from several peers is where it can
happen: the exporter is held up sending while the shards' journals,
cut back to twice their rows, run past its place.

Testing: a new integration test has a collector connected before two
feeders send 40,000 prefixes each, and checks it is told all 80,000
routes with their identifiers, which takes the exporter's initial walk
of the table through many batches; the advertiser and ADD-PATH tests
pass.

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

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<u32> = held.keys().map(|(_, id)| *id).collect();
+    assert_eq!(ids.into_iter().collect::<Vec<_>>(), [1, 2]);
+    Ok(())
+}

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::<usize>();
+        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(())