export: give the leave to gather back once the table is gathered

The last change held a session's leave to gather a table, of which
there are four, until its table had gone out. A peer that reads at the
pace of its receive window takes hours over a table, and two such
peers held two of the four for that long, keeping other sessions from
gathering theirs: they waited for a leave before their first UPDATE.
The leave is for the memory a gather takes; what is still held once
the table is gathered is let go of as it goes out, as it was before
the gather was reworked.

The mesh benchmark's new peer can also come once the first router has
its own peers' routes and the other router's are still arriving
(BENCH_RECEIVER_EARLY=2), as a router's sessions come up after a
restart while its peers' tables arrive; the walk of a full table while
the table fills costs the arriving routes nothing measurable in the
VM.

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

crates/bench/benches/mesh.rs

@@ -17,7 +17,10 @@
 //! second; `BENCH_SCALE_DOWN` shrinks the table. The routers peer over
 //! IPv6 loopback and the feeders come over IPv4 loopback; the new peer
 //! comes from 127.0.0.2, which Linux has and macOS does not, so that
-//! part is skipped on macOS.
+//! part is skipped on macOS. With `BENCH_RECEIVER_EARLY=1` the new peer
+//! comes as the feeders start instead, so that its exporter goes
+//! through the table while the table is being filled, as a router's
+//! do after a restart.
 
 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::sync::Arc;
@@ -32,6 +35,7 @@ use bgp_packet::open::{AddPathEntry, Capability, SendReceive};
 use bgp_server::config::PeerConfig;
 use bgp_server::family::SUPPORTED;
 use bgp_server::policy::{ExportPolicy, ImportPolicy};
+use bgp_server::session::SessionRegistry;
 use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
 
 /// An AS number that is in none of the synthetic paths.
@@ -158,7 +162,21 @@ fn main() {
         let own_a = per_peer * peers_a as usize;
         let own_b = per_peer * peers_b as usize;
 
+        let receiver_routes = shape.ipv4_prefixes * peers_a as usize;
         let started = Instant::now();
+        // Early: as the feeders start, or (2) once A has its own peers'
+        // routes and B's are still coming, as a router's sessions come
+        // up after a restart while its peers' tables arrive.
+        let when = env("BENCH_RECEIVER_EARLY", 0);
+        let early_receiver = (when > 0).then(|| {
+            let (addr, registry, rib) = (a.addr, a.registry.clone(), Arc::clone(&a.rib));
+            tokio::spawn(async move {
+                while when == 2 && rib.len() < own_a {
+                    tokio::time::sleep(Duration::from_millis(5)).await;
+                }
+                resync(addr, registry, receiver_routes).await
+            })
+        });
         let mut sending = Vec::new();
         for (mut session, stream) in feed_a.into_iter().chain(feed_b) {
             sending.push(tokio::spawn(async move {
@@ -235,20 +253,24 @@ fn main() {
         for session in sending {
             feeders.push(session.await.expect("a feeder is done"));
         }
-        resync(&a, shape.ipv4_prefixes * peers_a as usize).await;
+        match early_receiver {
+            Some(receiver) => receiver.await.expect("the receiver is done"),
+            None => resync(a.addr, a.registry.clone(), receiver_routes).await,
+        }
         drop(feeders);
     });
 }
 
-/// A new peer takes router `a`'s table, `routes` routes with every path:
-/// how long until the first UPDATE, and until all of it is sent.
-async fn resync(a: &TestServer, routes: usize) {
+/// A new peer takes the table of the router at `addr`, whose sessions
+/// are in `registry`: `routes` routes with every path. How long until
+/// the first UPDATE, and until all of it is sent.
+async fn resync(addr: SocketAddr, registry: SessionRegistry, routes: usize) {
     let socket = tokio::net::TcpSocket::new_v4().expect("a socket");
     if let Err(e) = socket.bind(SocketAddr::new(RECEIVER.into(), 0)) {
         println!("A new peer taking A's routes: skipped, {RECEIVER} is not ours ({e})");
         return;
     }
-    let stream = socket.connect(a.addr).await.expect("A listens");
+    let stream = socket.connect(addr).await.expect("A listens");
     let mut session = TestPeer::from_stream(stream);
     let mut capabilities = vec![Capability::FourByteAsn { asn: ASN }];
     capabilities.extend(SUPPORTED.into_iter().map(Capability::multiprotocol));
@@ -328,8 +350,7 @@ async fn resync(a: &TestServer, routes: usize) {
     let deadline = Duration::from_secs(env("BENCH_DEADLINE", 600) as u64);
     let advertised = loop {
         tokio::time::sleep(Duration::from_millis(5)).await;
-        let advertised = a
-            .registry
+        let advertised = registry
             .list()
             .iter()
             .find(|s| s.peer.addr.ip() == RECEIVER)
@@ -338,8 +359,7 @@ async fn resync(a: &TestServer, routes: usize) {
             break advertised;
         }
         if started.elapsed() > deadline {
-            let sessions: Vec<String> = a
-                .registry
+            let sessions: Vec<String> = registry
                 .list()
                 .iter()
                 .map(|s| {