diff --git a/crates/bench/benches/full_table.rs b/crates/bench/benches/full_table.rs
index 5148813..4621971 100644
--- a/crates/bench/benches/full_table.rs
+++ b/crates/bench/benches/full_table.rs
@@ -18,6 +18,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::{Duration, Instant};
 
 use bgp_bench::memory::{Region, describe};
+use bgp_bench::report::{mib, per_second, row};
 use bgp_bench::{Prefixes, Shape, announcements, encoded, env, wire_context};
 use bgp_integration_tests::test_peer::TestPeer;
 use bgp_integration_tests::test_server::TestServer;
@@ -26,23 +27,6 @@ use bgp_primitives::ip_prefix::IpPrefix;
 use bgp_rib::table::{PeerIndex, PeerInfo, RibTable};
 use bytes::BytesMut;
 
-fn mib(bytes: usize) -> f64 {
-    bytes as f64 / (1024.0 * 1024.0)
-}
-
-fn per_second(count: usize, elapsed: Duration) -> String {
-    let rate = count as f64 / elapsed.as_secs_f64();
-    if rate >= 1e6 {
-        format!("{:.2} M/s", rate / 1e6)
-    } else {
-        format!("{:.0} k/s", rate / 1e3)
-    }
-}
-
-fn row(what: &str, elapsed: Duration, detail: String) {
-    println!("{what:<44} {:>9.3} s   {detail}", elapsed.as_secs_f64());
-}
-
 fn main() {
     let peers = env("BENCH_PEERS", 4) as u32;
     let shape = Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1));
@@ -110,6 +94,8 @@ fn main() {
                 .handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 90)
                 .await
                 .expect("a session");
+            // The same bytes `table_stream` would make, from the messages
+            // already encoded for the decoding above.
             let stream: Vec<u8> = messages.iter().flat_map(|m| m.iter().copied()).collect();
             senders.push((session, stream));
         }
diff --git a/crates/bench/benches/mesh.rs b/crates/bench/benches/mesh.rs
index 9c09c32..1946db7 100644
--- a/crates/bench/benches/mesh.rs
+++ b/crates/bench/benches/mesh.rs
@@ -32,17 +32,21 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, Instant};
 
-use bgp_bench::{Prefixes, Shape, announcements, encoded, env};
+use bgp_bench::report::row;
+use bgp_bench::{Prefixes, Shape, announcements, encoded, env, table_stream};
 use bgp_integration_tests::peer_config;
-use bgp_integration_tests::test_peer::TestPeer;
+use bgp_integration_tests::test_peer::{TestPeer, collector_capabilities};
 use bgp_integration_tests::test_server::TestServer;
 use bgp_packet::attributes::PathAttribute;
-use bgp_packet::open::{AddPathEntry, Capability, SendReceive};
+use bgp_packet::message::msg_type;
+use bgp_packet::open::SendReceive;
 use bgp_server::config::PeerConfig;
 use bgp_server::family::SUPPORTED;
 use bgp_server::policy::{ExportPolicy, ImportPolicy, PeerRole};
 use bgp_server::session::SessionRegistry;
 use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
+use tokio::net::tcp::OwnedWriteHalf;
+use tokio::task::JoinHandle;
 
 /// An AS number that is in none of the synthetic paths.
 const ASN: u32 = 4_200_000_000;
@@ -58,6 +62,9 @@ fn collector_address(n: u32) -> Ipv4Addr {
 /// The collectors' AS numbers start here.
 const COLLECTOR_ASN: u32 = 65000;
 
+/// How often the tables are looked at while they fill.
+const POLL: Duration = Duration::from_millis(5);
+
 /// What the routers send each other, and the receiver: every path.
 fn every_path() -> Arc<ExportPolicy> {
     Arc::new(ExportPolicy {
@@ -77,12 +84,12 @@ fn free_port() -> u16 {
 
 /// A router of the mesh, listening for feeders on IPv4 loopback and for
 /// the other router on IPv6 loopback at `port`, which it dials at
-/// `other`.
+/// `other_port`, knowing it as `other_name`.
 async fn router(
-    name: &'static str,
+    other_name: &'static str,
     id: u8,
     port: u16,
-    other: u16,
+    other_port: u16,
     with_receiver: bool,
     collectors: u32,
 ) -> TestServer {
@@ -99,8 +106,8 @@ async fn router(
             import: ImportPolicy::permissive(),
             export: every_path(),
             ..peer_config(
-                name,
-                SocketAddr::new(Ipv6Addr::LOCALHOST.into(), other),
+                other_name,
+                SocketAddr::new(Ipv6Addr::LOCALHOST.into(), other_port),
                 ASN,
             )
         }];
@@ -144,20 +151,22 @@ async fn feeders(
 ) -> Vec<(TestPeer, Vec<u8>)> {
     let mut sessions = Vec::new();
     for peer in first..first + count {
-        let mut updates = announcements(prefixes, shape, peer);
-        if marked {
+        let stream = if marked {
             // As having come across from a peer (RFC 9234): not for the
             // collectors, which are peers too.
+            let mut updates = announcements(prefixes, shape, peer);
             for update in &mut updates {
                 update
                     .path_attributes
                     .push(PathAttribute::OnlyToCustomer(64512 + peer));
             }
-        }
-        let stream: Vec<u8> = encoded(&updates)
-            .iter()
-            .flat_map(|m| m.iter().copied())
-            .collect();
+            encoded(&updates)
+                .iter()
+                .flat_map(|m| m.iter().copied())
+                .collect()
+        } else {
+            table_stream(prefixes, shape, peer)
+        };
         let mut session = TestPeer::connect(router.addr).await.expect("a connection");
         // No hold timer: a feeder has nothing to say after its table.
         session
@@ -169,39 +178,112 @@ async fn feeders(
     sessions
 }
 
-fn main() {
-    // The routers' logs, with `BENCH_LOG` as a `RUST_LOG` filter.
-    if let Ok(filter) = std::env::var("BENCH_LOG") {
-        tracing_subscriber::fmt()
-            .with_env_filter(filter)
-            .with_writer(std::io::stderr)
-            .init();
-    }
-    let peers_a = env("BENCH_PEERS", 4) as u32;
-    let peers_b = env("BENCH_PEERS_B", 2) as u32;
-    let shape = Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1));
-    let per_peer = shape.prefixes();
-    println!(
-        "== Two routers of AS{ASN}, every path both ways: A fed by {peers_a} peers, B by {peers_b}, {} prefixes each",
-        per_peer
-    );
-    let prefixes = Prefixes::generate(shape);
+/// Every session of `registry`, one line each, for the panic when a
+/// deadline passes: where the routes have got to, and what each session
+/// was to be sent.
+fn sessions_report(registry: &SessionRegistry) -> Vec<String> {
+    registry
+        .list()
+        .iter()
+        .map(|s| {
+            format!(
+                "{} {:?} export {:?} add_path_send {:?} routes {} rejected {} advertised {}",
+                s.peer,
+                s.name,
+                s.export,
+                s.add_path_send,
+                s.counters.routes(),
+                s.counters.rejected(),
+                s.counters.advertised()
+            )
+        })
+        .collect()
+}
 
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .build()
-        .expect("a runtime");
-    runtime.block_on(async {
+/// How big a run is, from the environment.
+struct Settings {
+    peers_a: u32,
+    peers_b: u32,
+    shape: Shape,
+    collectors: u32,
+}
+
+impl Settings {
+    fn from_env() -> Self {
+        Self {
+            peers_a: env("BENCH_PEERS", 4) as u32,
+            peers_b: env("BENCH_PEERS_B", 2) as u32,
+            shape: Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1)),
+            collectors: env("BENCH_COLLECTORS", 0) as u32,
+        }
+    }
+
+    /// How many routes A's own feeders bring, and B's.
+    fn own_routes(&self) -> (usize, usize) {
+        let per_peer = self.shape.prefixes();
+        (
+            per_peer * self.peers_a as usize,
+            per_peer * self.peers_b as usize,
+        )
+    }
+
+    /// How many routes the receiver is to be sent: A's IPv4 routes from
+    /// its own peers.
+    fn receiver_routes(&self) -> usize {
+        self.shape.ipv4_prefixes * self.peers_a as usize
+    }
+}
+
+/// The two routers, up and peered, with the collectors' sessions once
+/// they are connected.
+struct Mesh {
+    settings: Settings,
+    prefixes: Prefixes,
+    a: TestServer,
+    b: TestServer,
+    /// The collectors' connections, drained, held until the end.
+    collectors: Vec<(OwnedWriteHalf, JoinHandle<()>)>,
+}
+
+/// The feeders sending, from when they began.
+struct Feeding {
+    started: Instant,
+    /// Each hands its session back once its table is written.
+    sending: Vec<JoinHandle<TestPeer>>,
+}
+
+/// When each router had what: its own peers' routes, and the other's too.
+struct Tables {
+    a_own: Duration,
+    b_own: Duration,
+    a_all: Duration,
+    b_all: Duration,
+}
+
+impl Mesh {
+    /// Starts the routers, with the collectors and receiver configured at
+    /// A, and waits for them to peer.
+    async fn start(settings: Settings, prefixes: Prefixes) -> Self {
         let (port_a, port_b) = (free_port(), free_port());
-        let collectors = env("BENCH_COLLECTORS", 0) as u32;
-        let a = router("b", 1, port_a, port_b, true, collectors).await;
+        let a = router("b", 1, port_a, port_b, true, settings.collectors).await;
         let b = router("a", 2, port_b, port_a, false, 0).await;
         a.wait_for_sessions(1).await.expect("the mesh session at A");
         b.wait_for_sessions(1).await.expect("the mesh session at B");
-        // The collectors are there before the feeders, as a router's
-        // peers are before its tables arrive after a restart.
-        let mut draining = Vec::new();
-        for n in 1..=collectors {
+        Self {
+            settings,
+            prefixes,
+            a,
+            b,
+            collectors: Vec::new(),
+        }
+    }
+
+    /// Connects the collectors to A. They are there before the feeders, as
+    /// a router's peers are before its tables arrive after a restart. An
+    /// address that is not this machine's ends the connecting, with a
+    /// word.
+    async fn connect_collectors(&mut self) {
+        for n in 1..=self.settings.collectors {
             let socket = tokio::net::TcpSocket::new_v4().expect("a socket");
             if let Err(e) = socket.bind(SocketAddr::new(collector_address(n).into(), 0)) {
                 println!(
@@ -210,61 +292,82 @@ fn main() {
                 );
                 break;
             }
-            let stream = socket.connect(a.addr).await.expect("A listens");
+            let stream = socket.connect(self.a.addr).await.expect("A listens");
             let mut session = TestPeer::from_stream(stream);
             session
                 .handshake(COLLECTOR_ASN + n, Ipv4Addr::new(192, 0, 3, n as u8), 0)
                 .await
                 .expect("A takes a collector's session");
-            let (mut reader, writer) = session.into_stream().into_split();
-            draining.push((
-                writer,
-                tokio::spawn(async move {
-                    let mut sink = vec![0u8; 1 << 16];
-                    while matches!(reader.read(&mut sink).await, Ok(n) if n > 0) {}
-                }),
-            ));
+            self.collectors.push(session.drain());
         }
-        if collectors > 0 {
-            a.wait_for_sessions(1 + draining.len())
+        if !self.collectors.is_empty() {
+            self.a
+                .wait_for_sessions(1 + self.collectors.len())
                 .await
                 .expect("the collectors' sessions at A");
         }
+    }
 
-        let feed_a = feeders(&a, &prefixes, shape, 1, peers_a, collectors > 0).await;
-        let feed_b = feeders(&b, &prefixes, shape, 1 + peers_a, peers_b, false).await;
-        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;
+    /// Connects the feeders to both routers and starts them sending.
+    async fn feed(&self) -> Feeding {
+        let Settings {
+            peers_a,
+            peers_b,
+            shape,
+            ..
+        } = self.settings;
+        let marked = !self.collectors.is_empty();
+        let feed_a = feeders(&self.a, &self.prefixes, shape, 1, peers_a, marked).await;
+        let feed_b = feeders(&self.b, &self.prefixes, shape, 1 + peers_a, peers_b, false).await;
         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 sending = feed_a
+            .into_iter()
+            .chain(feed_b)
+            .map(|(mut session, stream)| {
+                tokio::spawn(async move {
+                    session.send_raw(&stream).await.expect("the router reads");
+                    session
+                })
             })
-        });
-        let mut sending = Vec::new();
-        for (mut session, stream) in feed_a.into_iter().chain(feed_b) {
-            sending.push(tokio::spawn(async move {
-                session.send_raw(&stream).await.expect("the router reads");
-                session
-            }));
+            .collect();
+        Feeding { started, sending }
+    }
+
+    /// The receiver taking A's table while it fills, if `BENCH_RECEIVER_EARLY`
+    /// asks for that: as the feeders start (1), or once A has its own
+    /// peers' routes and B's are still coming (2), as a router's sessions
+    /// come up after a restart while its peers' tables arrive.
+    fn early_receiver(&self, deadline: Duration) -> Option<JoinHandle<()>> {
+        let when = env("BENCH_RECEIVER_EARLY", 0);
+        if when == 0 {
+            return None;
         }
+        let (own_a, _) = self.settings.own_routes();
+        let routes = self.settings.receiver_routes();
+        let (addr, registry, rib) = (
+            self.a.addr,
+            self.a.registry.clone(),
+            Arc::clone(&self.a.rib),
+        );
+        Some(tokio::spawn(async move {
+            while when == 2 && rib.len() < own_a {
+                tokio::time::sleep(POLL).await;
+            }
+            resync(addr, registry, routes, deadline).await
+        }))
+    }
+
+    /// Waits until both routers have every route, noting when each had
+    /// what; or panics once `deadline` has passed since `started`.
+    async fn wait_for_tables(&self, started: Instant, deadline: Duration) -> Tables {
+        let (own_a, own_b) = self.settings.own_routes();
         let mut a_own = None;
         let mut b_own = None;
         let mut a_all = None;
         let mut b_all = None;
-        while a_all.is_none() || b_all.is_none() {
-            tokio::time::sleep(Duration::from_millis(5)).await;
-            let (at_a, at_b) = (a.rib.len(), b.rib.len());
+        let (a_all, b_all) = loop {
+            tokio::time::sleep(POLL).await;
+            let (at_a, at_b) = (self.a.rib.len(), self.b.rib.len());
             let now = started.elapsed();
             if a_own.is_none() && at_a >= own_a {
                 a_own = Some(now);
@@ -278,192 +381,279 @@ fn main() {
             if b_all.is_none() && at_b >= own_a + own_b {
                 b_all = Some(now);
             }
-            if now > Duration::from_secs(env("BENCH_DEADLINE", 600) as u64) {
-                let sessions = |router: &TestServer| -> Vec<String> {
-                    router
-                        .registry
-                        .list()
-                        .iter()
-                        .map(|s| {
-                            format!(
-                                "{} routes {} rejected {}",
-                                s.peer,
-                                s.counters.routes(),
-                                s.counters.rejected()
-                            )
-                        })
-                        .collect()
-                };
+            if let (Some(a_all), Some(b_all)) = (a_all, b_all) {
+                break (a_all, b_all);
+            }
+            if now > deadline {
                 panic!(
                     "not all routes arrived: A has {at_a} {:?}, B has {at_b} {:?}",
-                    sessions(&a),
-                    sessions(&b)
+                    sessions_report(&self.a.registry),
+                    sessions_report(&self.b.registry)
                 );
             }
-        }
-        let row = |what: &str, when: Option<Duration>, routes: usize| {
-            let when = when.unwrap_or_default();
-            println!(
-                "{what:<52} {:>8.3} s   {:.2} M routes/s",
-                when.as_secs_f64(),
-                routes as f64 / when.as_secs_f64() / 1e6
-            );
         };
-        row("A has its own peers' routes", a_own, own_a);
-        row("B has its own peers' routes", b_own, own_b);
+        Tables {
+            a_own: a_own.unwrap_or_default(),
+            b_own: b_own.unwrap_or_default(),
+            a_all,
+            b_all,
+        }
+    }
+
+    /// One row for each of the four moments of `tables`, with the rate the
+    /// routes that made the moment came at.
+    fn report(&self, tables: &Tables) {
+        let (own_a, own_b) = self.settings.own_routes();
+        let rate = |when: Duration, routes: usize| {
+            format!("{:.2} M routes/s", routes as f64 / when.as_secs_f64() / 1e6)
+        };
+        row(
+            "A has its own peers' routes",
+            tables.a_own,
+            rate(tables.a_own, own_a),
+        );
+        row(
+            "B has its own peers' routes",
+            tables.b_own,
+            rate(tables.b_own, own_b),
+        );
         row(
             "A has B's routes too (B's exporter, A's session)",
-            a_all,
-            own_b,
+            tables.a_all,
+            rate(tables.a_all, own_b),
         );
         row(
             "B has A's routes too (A's exporter, B's session)",
-            b_all,
-            own_a,
+            tables.b_all,
+            rate(tables.b_all, own_a),
         );
-        // The feeders' sessions stay up until the end: a feeder going
-        // would have its routes withdrawn from both tables meanwhile.
+    }
+}
+
+impl Feeding {
+    /// Waits for every feeder to have written its table, and returns the
+    /// sessions, which are to stay up until the end: a feeder going would
+    /// have its routes withdrawn from both tables meanwhile.
+    async fn finished(self) -> Vec<TestPeer> {
         let mut feeders = Vec::new();
-        for session in sending {
+        for session in self.sending {
             feeders.push(session.await.expect("a feeder is done"));
         }
+        feeders
+    }
+}
+
+fn main() {
+    // The routers' logs, with `BENCH_LOG` as a `RUST_LOG` filter.
+    if let Ok(filter) = std::env::var("BENCH_LOG") {
+        tracing_subscriber::fmt()
+            .with_env_filter(filter)
+            .with_writer(std::io::stderr)
+            .init();
+    }
+    let settings = Settings::from_env();
+    let deadline = Duration::from_secs(env("BENCH_DEADLINE", 600) as u64);
+    println!(
+        "== Two routers of AS{ASN}, every path both ways: A fed by {} peers, B by {}, {} prefixes each",
+        settings.peers_a,
+        settings.peers_b,
+        settings.shape.prefixes()
+    );
+    let prefixes = Prefixes::generate(settings.shape);
+
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .expect("a runtime");
+    runtime.block_on(async {
+        let mut mesh = Mesh::start(settings, prefixes).await;
+        mesh.connect_collectors().await;
+        let early_receiver = mesh.early_receiver(deadline);
+        let feeding = mesh.feed().await;
+        let tables = mesh.wait_for_tables(feeding.started, deadline).await;
+        mesh.report(&tables);
+        let feeders = feeding.finished().await;
         match early_receiver {
             Some(receiver) => receiver.await.expect("the receiver is done"),
-            None => resync(a.addr, a.registry.clone(), receiver_routes).await,
+            None => {
+                resync(
+                    mesh.a.addr,
+                    mesh.a.registry.clone(),
+                    mesh.settings.receiver_routes(),
+                    deadline,
+                )
+                .await
+            }
         }
         drop(feeders);
-        drop(draining);
+        drop(mesh);
     });
 }
 
+/// The new peer taking a router's table: counting the UPDATEs it is sent
+/// while keeping the session alive.
+struct Receiver {
+    started: Instant,
+    /// When the first UPDATE arrived, in microseconds after `started`; 0
+    /// until it has.
+    first: Arc<AtomicU64>,
+    updates: Arc<AtomicU64>,
+    counting: JoinHandle<()>,
+    keepalives: JoinHandle<()>,
+}
+
+impl Receiver {
+    /// A session with the router at `addr` from [`RECEIVER`], as a
+    /// collector of every path; or `None`, with a word, where that address
+    /// is not this machine's to connect from.
+    async fn connect(addr: SocketAddr) -> Option<TestPeer> {
+        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 None;
+        }
+        let stream = socket.connect(addr).await.expect("A listens");
+        let mut session = TestPeer::from_stream(stream);
+        session
+            .handshake_with(
+                ASN,
+                Ipv4Addr::new(192, 0, 2, 9),
+                90,
+                collector_capabilities(ASN, &SUPPORTED, false),
+            )
+            .await
+            .expect("A takes the receiver's session");
+        Some(session)
+    }
+
+    /// Starts counting what `session` is sent, from now. Every millionth
+    /// UPDATE is noted, as is a NOTIFICATION or the connection closing:
+    /// what went wrong, if the table never all arrives.
+    fn count_updates(session: TestPeer) -> Self {
+        let started = Instant::now();
+        let (reader, mut writer) = session.into_stream().into_split();
+        // KEEPALIVEs, so that the session outlives the hold time.
+        let keepalives = tokio::spawn(async move {
+            let keepalive = TestPeer::keepalive_bytes();
+            loop {
+                tokio::time::sleep(Duration::from_secs(20)).await;
+                if writer.write_all(&keepalive).await.is_err() {
+                    break;
+                }
+            }
+        });
+        let first = Arc::new(AtomicU64::new(0));
+        let updates = Arc::new(AtomicU64::new(0));
+        let counting = tokio::spawn({
+            let (first, updates) = (Arc::clone(&first), Arc::clone(&updates));
+            async move {
+                let mut reader = BufReader::with_capacity(1 << 20, reader);
+                let mut header = [0u8; 19];
+                let mut body = vec![0u8; 4096];
+                while reader.read_exact(&mut header).await.is_ok() {
+                    let length = usize::from(u16::from_be_bytes([header[16], header[17]]));
+                    if reader.read_exact(&mut body[..length - 19]).await.is_err() {
+                        break;
+                    }
+                    match header[18] {
+                        msg_type::UPDATE => {
+                            let _ = first.compare_exchange(
+                                0,
+                                started.elapsed().as_micros() as u64,
+                                Ordering::Relaxed,
+                                Ordering::Relaxed,
+                            );
+                            let n = updates.fetch_add(1, Ordering::Relaxed) + 1;
+                            if n % 1_000_000 == 0 {
+                                println!(
+                                    "  the receiver has read {n} UPDATEs after {:.2} s",
+                                    started.elapsed().as_secs_f64()
+                                );
+                            }
+                        }
+                        msg_type::NOTIFICATION => println!(
+                            "the receiver was sent a NOTIFICATION, error {} subcode {}, after {} UPDATEs",
+                            body[0],
+                            body[1],
+                            updates.load(Ordering::Relaxed)
+                        ),
+                        _ => {}
+                    }
+                }
+                println!(
+                    "the receiver's connection closed after {} UPDATEs",
+                    updates.load(Ordering::Relaxed)
+                );
+            }
+        });
+        Self {
+            started,
+            first,
+            updates,
+            counting,
+            keepalives,
+        }
+    }
+
+    /// Waits until the router whose sessions are in `registry` has
+    /// advertised `routes` routes to the receiver, and returns how many it
+    /// has; or panics once `deadline` has passed.
+    async fn wait_for_advertised(
+        &self,
+        registry: &SessionRegistry,
+        routes: usize,
+        deadline: Duration,
+    ) -> u64 {
+        loop {
+            tokio::time::sleep(POLL).await;
+            let advertised = registry
+                .list()
+                .iter()
+                .find(|s| s.peer.addr.ip() == RECEIVER)
+                .map_or(0, |s| s.counters.advertised());
+            if advertised as usize >= routes {
+                return advertised;
+            }
+            if self.started.elapsed() > deadline {
+                panic!(
+                    "the receiver has {advertised} of {routes} routes and read {} UPDATEs; sessions at A: {:#?}",
+                    self.updates.load(Ordering::Relaxed),
+                    sessions_report(registry)
+                );
+            }
+        }
+    }
+
+    /// Prints the row for the table being all there, `advertised` routes
+    /// of it, and stops counting.
+    fn report(self, advertised: u64) {
+        let took = self.started.elapsed();
+        self.counting.abort();
+        self.keepalives.abort();
+        let first = Duration::from_micros(self.first.load(Ordering::Relaxed));
+        row(
+            "A new peer has A's IPv4 routes (A's exporter)",
+            took,
+            format!(
+                "{:.2} M routes/s   first UPDATE after {:.0} ms, {} UPDATEs",
+                advertised as f64 / took.as_secs_f64() / 1e6,
+                first.as_secs_f64() * 1e3,
+                self.updates.load(Ordering::Relaxed),
+            ),
+        );
+    }
+}
+
 /// 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})");
+async fn resync(addr: SocketAddr, registry: SessionRegistry, routes: usize, deadline: Duration) {
+    let Some(session) = Receiver::connect(addr).await else {
         return;
-    }
-    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));
-    capabilities.push(Capability::AddPath {
-        entries: SUPPORTED
-            .iter()
-            .map(|family| AddPathEntry {
-                afi: family.afi,
-                safi: family.safi,
-                send_recv: SendReceive::Receive,
-            })
-            .collect(),
-    });
-    session
-        .handshake_with(ASN, Ipv4Addr::new(192, 0, 2, 9), 90, capabilities)
-        .await
-        .expect("A takes the receiver's session");
-    let started = Instant::now();
-    let (reader, mut writer) = session.into_stream().into_split();
-    // KEEPALIVEs, so that the session outlives the hold time.
-    let keepalives = tokio::spawn(async move {
-        let mut keepalive = [0xffu8; 19];
-        keepalive[16..19].copy_from_slice(&[0, 19, 4]);
-        loop {
-            tokio::time::sleep(Duration::from_secs(20)).await;
-            if writer.write_all(&keepalive).await.is_err() {
-                break;
-            }
-        }
-    });
-    // When the first UPDATE arrived, in microseconds; 0 until it has.
-    let first = Arc::new(AtomicU64::new(0));
-    let updates = Arc::new(AtomicU64::new(0));
-    let counting = tokio::spawn({
-        let (first, updates) = (Arc::clone(&first), Arc::clone(&updates));
-        async move {
-            let mut reader = BufReader::with_capacity(1 << 20, reader);
-            let mut header = [0u8; 19];
-            let mut body = vec![0u8; 4096];
-            while reader.read_exact(&mut header).await.is_ok() {
-                let length = usize::from(u16::from_be_bytes([header[16], header[17]]));
-                if reader.read_exact(&mut body[..length - 19]).await.is_err() {
-                    break;
-                }
-                match header[18] {
-                    // An UPDATE (RFC 4271 §4.1).
-                    2 => {
-                        let _ = first.compare_exchange(
-                            0,
-                            started.elapsed().as_micros() as u64,
-                            Ordering::Relaxed,
-                            Ordering::Relaxed,
-                        );
-                        let n = updates.fetch_add(1, Ordering::Relaxed) + 1;
-                        if n % 1_000_000 == 0 {
-                            println!(
-                                "  the receiver has read {n} UPDATEs after {:.2} s",
-                                started.elapsed().as_secs_f64()
-                            );
-                        }
-                    }
-                    3 => println!(
-                        "the receiver was sent a NOTIFICATION, error {} subcode {}, after {} UPDATEs",
-                        body[0],
-                        body[1],
-                        updates.load(Ordering::Relaxed)
-                    ),
-                    _ => {}
-                }
-            }
-            println!(
-                "the receiver's connection closed after {} UPDATEs",
-                updates.load(Ordering::Relaxed)
-            );
-        }
-    });
-    let deadline = Duration::from_secs(env("BENCH_DEADLINE", 600) as u64);
-    let advertised = loop {
-        tokio::time::sleep(Duration::from_millis(5)).await;
-        let advertised = registry
-            .list()
-            .iter()
-            .find(|s| s.peer.addr.ip() == RECEIVER)
-            .map_or(0, |s| s.counters.advertised());
-        if advertised as usize >= routes {
-            break advertised;
-        }
-        if started.elapsed() > deadline {
-            let sessions: Vec<String> = registry
-                .list()
-                .iter()
-                .map(|s| {
-                    format!(
-                        "{} {:?} export {:?} add_path_send {:?} advertised {}",
-                        s.peer,
-                        s.name,
-                        s.export,
-                        s.add_path_send,
-                        s.counters.advertised()
-                    )
-                })
-                .collect();
-            panic!(
-                "the receiver has {advertised} of {routes} routes and read {} UPDATEs; sessions at A: {sessions:#?}",
-                updates.load(Ordering::Relaxed)
-            );
-        }
     };
-    let took = started.elapsed();
-    counting.abort();
-    keepalives.abort();
-    let first = Duration::from_micros(first.load(Ordering::Relaxed));
-    println!(
-        "{:<52} {:>8.3} s   {:.2} M routes/s   first UPDATE after {:.0} ms, {} UPDATEs",
-        "A new peer has A's IPv4 routes (A's exporter)",
-        took.as_secs_f64(),
-        advertised as f64 / took.as_secs_f64() / 1e6,
-        first.as_secs_f64() * 1e3,
-        updates.load(Ordering::Relaxed),
-    );
+    let receiver = Receiver::count_updates(session);
+    let advertised = receiver
+        .wait_for_advertised(&registry, routes, deadline)
+        .await;
+    receiver.report(advertised);
 }
diff --git a/crates/bench/benches/real_table.rs b/crates/bench/benches/real_table.rs
index 93451dd..aabecba 100644
--- a/crates/bench/benches/real_table.rs
+++ b/crates/bench/benches/real_table.rs
@@ -11,30 +11,20 @@ use std::net::Ipv4Addr;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use bgp_bench::memory::Region;
 use bgp_bench::mrt::{Dump, parse_attributes};
+use bgp_bench::report::{mib, per_second, row};
 use bgp_packet::message::{NlriNextHop, PathAttribute};
 use bgp_primitives::ip_prefix::IpPrefix;
 use bgp_rib::table::{Event, PeerIndex, PeerInfo, RibTable};
 
 /// Bytes allocated and not freed since `region` began; zero when memory
-/// is not being counted (see `bgp_bench::memory`).
-fn retained(region: &Region) -> i64 {
-    region.held().unwrap_or(0)
-}
-
-fn mib(bytes: i64) -> f64 {
-    bytes as f64 / (1024.0 * 1024.0)
-}
-
-fn per_second(count: usize, elapsed: Duration) -> String {
-    format!("{:.2} M/s", count as f64 / elapsed.as_secs_f64() / 1e6)
-}
-
-fn row(what: &str, elapsed: Duration, detail: String) {
-    println!("{what:<44} {:>9.3} s   {detail}", elapsed.as_secs_f64());
+/// is not being counted (see `bgp_bench::memory`), and zero too in the
+/// odd case of more freed than allocated, which is nothing held.
+fn retained(region: &Region) -> usize {
+    usize::try_from(region.held().unwrap_or(0)).unwrap_or(0)
 }
 
 fn dump_path() -> Option<PathBuf> {
diff --git a/crates/bench/examples/feed.rs b/crates/bench/examples/feed.rs
index 4a50945..c991665 100644
--- a/crates/bench/examples/feed.rs
+++ b/crates/bench/examples/feed.rs
@@ -32,32 +32,18 @@
 use std::net::{Ipv4Addr, SocketAddr};
 use std::time::{Duration, Instant};
 
-use bgp_bench::{Prefixes, Shape, announcements, encoded, env};
-use bgp_integration_tests::test_peer::TestPeer;
-use bgp_packet::open::{AddPathEntry, Capability, ExtendedNextHopEntry, SendReceive};
-use bgp_primitives::constants::{Afi, AfiSafi, Safi};
-use bgp_server::family::SUPPORTED;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use bgp_bench::report::mib;
+use bgp_bench::{Prefixes, Shape, env, table_stream};
+use bgp_integration_tests::test_peer::{TestPeer, collector_capabilities};
+use bgp_primitives::constants::AfiSafi;
+use tokio::io::AsyncWriteExt;
 use tokio::net::TcpSocket;
 use tokio::net::tcp::OwnedWriteHalf;
 
-/// A KEEPALIVE message: the marker, the length (19) and the type (4).
-const KEEPALIVE: [u8; 19] = {
-    let mut bytes = [0xff; 19];
-    bytes[16] = 0;
-    bytes[17] = 19;
-    bytes[18] = 4;
-    bytes
-};
-
 /// How often KEEPALIVEs go out while the sessions are held open: well
 /// inside the 90 s hold time.
 const KEEPALIVE_EVERY: Duration = Duration::from_secs(20);
 
-fn mib(bytes: usize) -> f64 {
-    bytes as f64 / (1024.0 * 1024.0)
-}
-
 #[tokio::main]
 async fn main() {
     let to: SocketAddr = std::env::var("FEED_TO")
@@ -73,12 +59,7 @@ async fn main() {
     let started = Instant::now();
     let prefixes = Prefixes::generate(shape);
     let tables: Vec<Vec<u8>> = (1..=peers)
-        .map(|peer| {
-            encoded(&announcements(&prefixes, shape, peer))
-                .iter()
-                .flat_map(|message| message.iter().copied())
-                .collect()
-        })
+        .map(|peer| table_stream(&prefixes, shape, peer))
         .collect();
     let routes = shape.prefixes() * peers as usize;
     let bytes: usize = tables.iter().map(Vec::len).sum();
@@ -109,38 +90,21 @@ async fn main() {
         }
         let stream = socket.connect(to).await.expect("the daemon listens");
         let mut session = TestPeer::from_stream(stream);
-        let mut capabilities = vec![Capability::FourByteAsn { asn: 212_232 }];
-        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,
-            }],
-        });
         session
-            .handshake_with(212_232, Ipv4Addr::new(10, 0, 0, 200), 90, capabilities)
+            .handshake_with(
+                212_232,
+                Ipv4Addr::new(10, 0, 0, 200),
+                90,
+                collector_capabilities(
+                    212_232,
+                    &[AfiSafi::IPV4_UNICAST, AfiSafi::IPV6_UNICAST],
+                    true,
+                ),
+            )
             .await
             .expect("the daemon takes the collector's session");
-        let (mut reader, writer) = session.into_stream().into_split();
-        draining.push(tokio::spawn(async move {
-            let mut sink = [0u8; 65536];
-            while let Ok(n) = reader.read(&mut sink).await {
-                if n == 0 {
-                    break;
-                }
-            }
-        }));
+        let (writer, sink) = session.drain();
+        draining.push(sink);
         println!("collector session established");
         collector_writer = Some(writer);
     }
@@ -166,7 +130,11 @@ async fn main() {
             )
             .await
             .expect("the daemon takes the session");
-        sessions.push(session.into_stream());
+        // Whatever the daemon sends back, KEEPALIVEs as a rule, is read
+        // and dropped, so that the writer is never held up.
+        let (writer, sink) = session.drain();
+        draining.push(sink);
+        sessions.push(writer);
     }
     println!("{peers} session(s) established; sending");
 
@@ -174,19 +142,8 @@ async fn main() {
     let senders: Vec<_> = sessions
         .into_iter()
         .zip(tables)
-        .map(|(stream, table)| {
+        .map(|(mut writer, table)| {
             tokio::spawn(async move {
-                let (mut reader, mut writer) = stream.into_split();
-                // Whatever the daemon sends back, KEEPALIVEs as a rule, is
-                // read and dropped, so that its writer is never held up.
-                tokio::spawn(async move {
-                    let mut sink = [0u8; 65536];
-                    while let Ok(n) = reader.read(&mut sink).await {
-                        if n == 0 {
-                            break;
-                        }
-                    }
-                });
                 let started = Instant::now();
                 writer.write_all(&table).await.expect("the daemon reads");
                 (writer, started.elapsed())
@@ -210,12 +167,13 @@ async fn main() {
         routes as f64 / elapsed.as_secs_f64() / 1e6
     );
 
+    let keepalive = TestPeer::keepalive_bytes();
     let deadline = Instant::now() + hold;
     while Instant::now() < deadline {
         tokio::time::sleep(KEEPALIVE_EVERY.min(deadline - Instant::now())).await;
         for writer in &mut writers {
             // A session the daemon closed is left alone.
-            let _ = writer.write_all(&KEEPALIVE).await;
+            let _ = writer.write_all(&keepalive).await;
         }
     }
     println!("sessions closed after {} s", hold.as_secs());
diff --git a/crates/bench/src/lib.rs b/crates/bench/src/lib.rs
index e74a061..d848a88 100644
--- a/crates/bench/src/lib.rs
+++ b/crates/bench/src/lib.rs
@@ -16,6 +16,7 @@
 
 pub mod memory;
 pub mod mrt;
+pub mod report;
 
 // The benchmarks allocate as the daemon does (bin/bgp/src/main.rs), with
 // jemalloc; the C library's allocator would put its own locks in the
@@ -266,6 +267,16 @@ pub fn announcements(prefixes: &Prefixes, shape: Shape, peer: u32) -> Vec<Update
     updates
 }
 
+/// The whole table as the peer numbered `peer` sends it, one stream of
+/// bytes: [`announcements`], [`encoded`] and run together, for a session
+/// that writes the table in one go.
+pub fn table_stream(prefixes: &Prefixes, shape: Shape, peer: u32) -> Vec<u8> {
+    encoded(&announcements(prefixes, shape, peer))
+        .iter()
+        .flat_map(|message| message.iter().copied())
+        .collect()
+}
+
 /// `updates` as they are on the wire, on a session with four-octet AS
 /// numbers.
 pub fn encoded(updates: &[UpdateMessage]) -> Vec<BytesMut> {
@@ -300,3 +311,68 @@ pub fn env(name: &str, default: usize) -> usize {
         .and_then(|v| v.parse().ok())
         .unwrap_or(default)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A small table, enough for every prefix length and attribute to
+    /// come up.
+    fn shape() -> Shape {
+        Shape::INTERNET.scaled_down(1000)
+    }
+
+    /// The benchmarks are compared from one run to the next, so the table
+    /// they are fed must not change between runs: the generator is
+    /// seeded, and nothing in it depends on where things are in memory.
+    #[test]
+    fn the_table_is_the_same_every_run() {
+        let shape = shape();
+        let once = Prefixes::generate(shape);
+        let again = Prefixes::generate(shape);
+        assert_eq!(once.ipv4, again.ipv4);
+        assert_eq!(once.ipv6, again.ipv6);
+        assert_eq!(once.ipv4.len(), shape.ipv4_prefixes);
+        assert_eq!(once.ipv6.len(), shape.ipv6_prefixes);
+        for peer in [1, 2] {
+            assert_eq!(
+                table_stream(&once, shape, peer),
+                table_stream(&again, shape, peer)
+            );
+        }
+        assert_ne!(
+            table_stream(&once, shape, 1),
+            table_stream(&once, shape, 2),
+            "each peer has attributes of its own"
+        );
+    }
+
+    /// What a peer of the benchmarks sends is well formed, and says what
+    /// it was meant to: every UPDATE reads back as it was made.
+    #[test]
+    fn every_announcement_decodes_to_itself() {
+        let shape = shape();
+        let prefixes = Prefixes::generate(shape);
+        let updates = announcements(&prefixes, shape, 1);
+        let ctx = wire_context();
+        let announced: usize = updates
+            .iter()
+            .map(|update| {
+                update.announced_routes.len()
+                    + update
+                        .path_attributes
+                        .iter()
+                        .map(|attribute| match attribute {
+                            PathAttribute::MpReachNlri(reach) => reach.prefixes.len(),
+                            _ => 0,
+                        })
+                        .sum::<usize>()
+            })
+            .sum();
+        assert_eq!(announced, shape.prefixes());
+        for (update, wire) in updates.iter().zip(encoded(&updates)) {
+            let decoded = Message::decode(&ctx, &wire).expect("well formed");
+            assert_eq!(decoded, Message::Update(update.clone()));
+        }
+    }
+}
diff --git a/crates/bench/src/mrt.rs b/crates/bench/src/mrt.rs
index 586ad5b..308ca29 100644
--- a/crates/bench/src/mrt.rs
+++ b/crates/bench/src/mrt.rs
@@ -10,7 +10,7 @@ use std::io::{self, BufReader, Read};
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 use std::path::Path;
 
-use bgp_packet::message::{NlriNextHop, PathAttribute};
+use bgp_packet::message::{NlriNextHop, PathAttribute, attr_type, split_attribute};
 use bgp_primitives::constants::IpVersion;
 use bgp_primitives::ip_prefix::IpPrefix;
 use flate2::read::MultiGzDecoder;
@@ -21,7 +21,6 @@ const TABLE_DUMP_V2: u16 = 13;
 const PEER_INDEX_TABLE: u16 = 1;
 const RIB_IPV4_UNICAST: u16 = 2;
 const RIB_IPV6_UNICAST: u16 = 4;
-const MP_REACH_NLRI: u8 = 14;
 
 /// A peer of the collector.
 #[derive(Debug, Clone)]
@@ -205,32 +204,23 @@ pub fn parse_attributes(mut wire: &[u8]) -> Option<(Vec<PathAttribute>, Option<N
     let mut attributes = Vec::new();
     let mut next_hop = None;
     while !wire.is_empty() {
-        if wire.get(1) == Some(&MP_REACH_NLRI) {
-            let extended = wire[0] & 0x10 != 0;
-            let (length, header) = if extended {
-                (
-                    usize::from(u16::from_be_bytes([*wire.get(2)?, *wire.get(3)?])),
-                    4,
-                )
-            } else {
-                (usize::from(*wire.get(2)?), 3)
+        let (raw, rest) = split_attribute(wire).ok()?;
+        if raw.type_code == attr_type::MP_REACH_NLRI {
+            // The value is the next hop's length and the next hop, and
+            // nothing of the family, reserved octet and prefixes that
+            // follow in an UPDATE, so the attribute's own parser would
+            // refuse it.
+            let (length, address) = raw.value.split_first()?;
+            if usize::from(*length) != address.len() {
+                return None;
+            }
+            next_hop = Some(NlriNextHop::from_bytes(address)?);
+        } else {
+            let ([], attribute) = PathAttribute::from_wire(&ctx, raw.wire).ok()? else {
+                return None;
             };
-            let value = wire.get(header..header + length)?;
-            let address = |bytes: &[u8]| Some(Ipv6Addr::from(<[u8; 16]>::try_from(bytes).ok()?));
-            next_hop = match value {
-                [4, a, b, c, d] => Some(NlriNextHop::Ipv4(Ipv4Addr::new(*a, *b, *c, *d))),
-                [16, rest @ ..] => Some(NlriNextHop::Ipv6(address(rest)?)),
-                [32, rest @ ..] if rest.len() == 32 => Some(NlriNextHop::Ipv6WithLl {
-                    global: address(&rest[..16])?,
-                    link_local: address(&rest[16..])?,
-                }),
-                _ => return None,
-            };
-            wire = &wire[header + length..];
-            continue;
+            attributes.push(attribute);
         }
-        let (rest, attribute) = PathAttribute::from_wire(&ctx, wire).ok()?;
-        attributes.push(attribute);
         wire = rest;
     }
     Some((attributes, next_hop))
diff --git a/crates/bench/src/report.rs b/crates/bench/src/report.rs
new file mode 100644
index 0000000..8f3ba01
--- /dev/null
+++ b/crates/bench/src/report.rs
@@ -0,0 +1,31 @@
+//! How the benchmarks print what they measured: one row a measurement,
+//! what was measured in a column wide enough for the longest of them, the
+//! time, and whatever else there is to say about it. The rows are what a
+//! run is compared against, so their shape is settled here once.
+
+use std::time::Duration;
+
+/// How wide the column that says what was measured is. The widest row is
+/// the mesh benchmark's, so that is what every benchmark lines up to.
+pub const WIDTH: usize = 52;
+
+/// `bytes` in mebibytes.
+pub fn mib(bytes: usize) -> f64 {
+    bytes as f64 / (1024.0 * 1024.0)
+}
+
+/// `count` things in `elapsed`, as a rate: millions a second where there
+/// are that many, thousands otherwise.
+pub fn per_second(count: usize, elapsed: Duration) -> String {
+    let rate = count as f64 / elapsed.as_secs_f64();
+    if rate >= 1e6 {
+        format!("{:.2} M/s", rate / 1e6)
+    } else {
+        format!("{:.0} k/s", rate / 1e3)
+    }
+}
+
+/// Prints one row: `what` took `elapsed`, and `detail` says more.
+pub fn row(what: &str, elapsed: Duration, detail: impl std::fmt::Display) {
+    println!("{what:<WIDTH$} {:>8.3} s   {detail}", elapsed.as_secs_f64());
+}
diff --git a/crates/fib/src/kernel.rs b/crates/fib/src/kernel.rs
index 5e14cab..f33f4db 100644
--- a/crates/fib/src/kernel.rs
+++ b/crates/fib/src/kernel.rs
@@ -498,9 +498,53 @@ struct Task {
     /// them means looking at every route, and is not done once there are
     /// none to find.
     keeping: bool,
+    /// What `follow` takes out of the table, and what it makes of it. Kept
+    /// from one wake to the next so that their room is not found anew each
+    /// time the table has something, which is many times a second while
+    /// it fills.
+    news: Vec<News>,
+    ops: Vec<Op>,
 }
 
 impl Task {
+    /// A task that is to keep the kernel in step with `rib`, talking to it
+    /// over `control` and `resolution` and writing routes over `routes`
+    /// from a task of its own. Returned with it is where that task
+    /// reports what became of the routes.
+    fn new(
+        rib: Arc<RibTable>,
+        fib: Fib,
+        control: Socket,
+        routes: Socket,
+        resolution: Option<Socket>,
+    ) -> (Self, mpsc::UnboundedReceiver<Report>) {
+        let (queue, batches) = mpsc::channel(QUEUE);
+        let (reports, heard) = mpsc::unbounded_channel();
+        let config = fib.config();
+        let task = Self {
+            config,
+            rib,
+            fib,
+            control,
+            resolution,
+            queue: Some(queue),
+            writer: tokio::spawn(write(routes, batches, reports)),
+            topology: Topology::new(config.resolution_table, config.resolve_via_default),
+            resolved: FastHashMap::default(),
+            plan: Plan::default(),
+            nexthops: config.nexthop_objects.then(FastHashMap::default),
+            owned: HashSet::new(),
+            awaiting: None,
+            marks: 0,
+            synced: false,
+            mended_at: None,
+            keeping: false,
+            news: Vec::with_capacity(BATCH),
+            ops: Vec::with_capacity(BATCH),
+        };
+        (task, heard)
+    }
+
     /// Whether a route the kernel has is one of ours.
     fn is_ours(&self, route: &Route) -> bool {
         route.protocol == PROTOCOL_BGP
@@ -631,6 +675,19 @@ impl Task {
         Ok(true)
     }
 
+    /// Learns what the kernel has that bears on our routes: the routes
+    /// that say the way to next hops, its nexthop objects, and the routes
+    /// of ours it has from before, which are kept for the peers to announce
+    /// again if `keep` says so. Done at the start, and again whenever what
+    /// the kernel has is no longer known.
+    async fn learn(&mut self, keep: bool) -> Result<(), FibError> {
+        self.learn_routes().await?;
+        let nexthops = self.kernel_nexthops().await?;
+        self.adopt_routes(&nexthops, keep).await?;
+        self.use_nexthops(&nexthops);
+        Ok(())
+    }
+
     /// Finds the routes of ours that the kernel has from before.
     async fn adopt_routes(
         &mut self,
@@ -751,17 +808,21 @@ impl Task {
         Ok(Ok((change, route)))
     }
 
-    /// Hands `ops` over to be written to the kernel, in order. With
-    /// `whole_table` they complete a whole table, and the kernel is in
-    /// step once it has done them.
-    async fn send(&mut self, ops: Vec<Op>, whole_table: bool) -> Result<(), FibError> {
+    /// Hands `ops` over to be written to the kernel, in order, and leaves
+    /// the vector empty for the caller to fill again. With `whole_table`
+    /// they complete a whole table, and the kernel is in step once it has
+    /// done them.
+    async fn send(&mut self, ops: &mut Vec<Op>, whole_table: bool) -> Result<(), FibError> {
         let mut batch = Batch {
             ops: Vec::with_capacity(ops.len()),
             requests: Vec::with_capacity(ops.len()),
             mark: None,
         };
-        let mut ops = std::collections::VecDeque::from(ops);
-        while let Some(op) = ops.pop_front() {
+        // By place rather than by draining: what takes a refused op's
+        // place is put at the end, and is dealt with in its turn.
+        let mut at = 0;
+        while let Some(op) = ops.get(at).copied() {
+            at += 1;
             match self.request(&op).await? {
                 Ok(request) => {
                     CHANGES
@@ -777,6 +838,7 @@ impl Task {
                 Err(errno) => ops.extend(self.refused(&op, errno)?),
             }
         }
+        ops.clear();
         if whole_table {
             self.marks += 1;
             batch.mark = Some(self.marks);
@@ -850,7 +912,12 @@ impl Task {
     /// hands over what brings the kernel in step with them. Returns whether
     /// the reader has read everything the table has for it.
     async fn follow(&mut self, reader: &mut Reader) -> Result<bool, FibError> {
-        let mut news = Vec::with_capacity(BATCH);
+        // Taken out of `self` while they are worked through, since working
+        // through them is what the rest of `self` is for.
+        let mut news = std::mem::take(&mut self.news);
+        let mut ops = std::mem::take(&mut self.ops);
+        news.clear();
+        ops.clear();
         // Only what is needed is taken while the table is held.
         let progress = reader.poll(BATCH, |event| {
             news.push(match event {
@@ -872,9 +939,8 @@ impl Task {
                 Event::Routes { .. } => return,
             });
         });
-        let mut ops = Vec::with_capacity(news.len());
         let mut whole_table = false;
-        for item in news {
+        for item in news.drain(..) {
             match item {
                 News::Reset => {
                     self.plan.reset();
@@ -928,7 +994,10 @@ impl Task {
                 News::Withdrawn(prefix) => ops.extend(self.plan.withdrawn(prefix)),
             }
         }
-        self.send(ops, whole_table).await?;
+        let sent = self.send(&mut ops, whole_table).await;
+        self.news = news;
+        self.ops = ops;
+        sent?;
         Ok(progress.caught_up)
     }
 
@@ -957,7 +1026,7 @@ impl Task {
             }
             Report::Refused { op, errno } => {
                 let instead = self.refused(&op, errno)?;
-                self.send(instead.into_iter().collect(), false).await?;
+                self.send(&mut instead.into_iter().collect(), false).await?;
                 // A refused route may have named a nexthop object that
                 // went away. Working the routes out again mends that.
                 let due = matches!(op, Op::Install { .. })
@@ -1026,18 +1095,14 @@ impl Task {
         self.flush(heard).await?;
         self.synced = false;
         self.plan = Plan::default();
-        self.learn_routes().await?;
-        let nexthops = self.kernel_nexthops().await?;
         // While routes of an earlier run are being kept for the peers to
         // announce again, they are kept through this too.
-        self.adopt_routes(&nexthops, self.keeping).await?;
-        self.use_nexthops(&nexthops);
-        Ok(())
+        self.learn(self.keeping).await
     }
 
     /// Waits until the kernel has done everything handed over so far.
     async fn flush(&mut self, heard: &mut mpsc::UnboundedReceiver<Report>) -> Result<(), FibError> {
-        self.send(vec![], true).await?;
+        self.send(&mut Vec::new(), true).await?;
         while let Some(mark) = self.awaiting {
             match heard.recv().await {
                 Some(Report::Done(done)) if done == mark => self.awaiting = None,
@@ -1090,8 +1155,8 @@ impl Task {
             self.plan = Plan::default();
             self.adopt_routes(&HashMap::new(), false).await?;
         }
-        let ops = self.plan.clear();
-        self.send(ops, false).await?;
+        let mut ops = self.plan.clear();
+        self.send(&mut ops, false).await?;
         self.finish_writer().await
     }
 
@@ -1153,38 +1218,15 @@ pub(crate) async fn run(
         resolution,
         mut hints,
     } = sockets;
-    let (queue, batches) = mpsc::channel(QUEUE);
-    let (reports, mut heard) = mpsc::unbounded_channel();
-    let config = fib.config();
-    let mut task = Task {
-        config,
-        rib,
-        fib,
-        control,
-        resolution,
-        queue: Some(queue),
-        writer: tokio::spawn(write(routes, batches, reports)),
-        topology: Topology::new(config.resolution_table, config.resolve_via_default),
-        resolved: FastHashMap::default(),
-        plan: Plan::default(),
-        nexthops: config.nexthop_objects.then(FastHashMap::default),
-        owned: HashSet::new(),
-        awaiting: None,
-        marks: 0,
-        synced: false,
-        mended_at: None,
-        keeping: false,
-    };
+    let (mut task, mut heard) = Task::new(rib, fib, control, routes, resolution);
+    let config = task.config;
     // The socket for news was bound before this: what changes while the
     // interfaces are being learned is told there.
     task.learn_interfaces().await?;
-    task.learn_routes().await?;
-    let nexthops = task.kernel_nexthops().await?;
     // Kept for a while, if there is to be any keeping: our table is empty
     // until the peers have sent theirs again.
     let keep = !config.stale_time.is_zero();
-    task.adopt_routes(&nexthops, keep).await?;
-    task.use_nexthops(&nexthops);
+    task.learn(keep).await?;
     let mut release_at = None;
     if task.plan.installed() > 0 {
         info!(
@@ -1284,12 +1326,12 @@ pub(crate) async fn run(
                 // taken out only to be put back.
                 while !task.follow(&mut reader).await? {}
                 task.keeping = false;
-                let stale = task.plan.release();
+                let mut stale = task.plan.release();
                 info!(
                     removed = stale.len(),
                     "Routes of an earlier run that were not announced again are removed"
                 );
-                task.send(stale, false).await?;
+                task.send(&mut stale, false).await?;
                 task.publish(false);
             }
             ended = &mut task.writer => return Err(writer_error(ended)),
diff --git a/crates/fib/src/netlink.rs b/crates/fib/src/netlink.rs
index c7b1f2d..de89914 100644
--- a/crates/fib/src/netlink.rs
+++ b/crates/fib/src/netlink.rs
@@ -29,9 +29,13 @@ pub const KIND_UNICAST: u8 = 1;
 /// `RTN_BLACKHOLE`: a route that discards the packets for its prefix.
 pub const KIND_BLACKHOLE: u8 = 6;
 
-const HEADER_LEN: usize = 16;
+/// The netlink header: length, type, flags, sequence number and the
+/// sender's port, four octets each but for the two-octet type and flags.
+/// The socket diagnostics family (`sockdiag`) frames its messages the
+/// same way.
+pub(crate) const HEADER_LEN: usize = 16;
 
-mod message_type {
+pub(crate) mod message_type {
     pub const ERROR: u16 = 2;
     pub const DONE: u16 = 3;
     pub const NEW_LINK: u16 = 16;
@@ -48,7 +52,7 @@ mod message_type {
     pub const GET_NEXTHOP: u16 = 106;
 }
 
-mod flag {
+pub(crate) mod flag {
     pub const REQUEST: u16 = 0x1;
     pub const ACK: u16 = 0x4;
     /// Send the sender what everyone else is told of the change.
@@ -76,8 +80,8 @@ pub mod group {
 /// joined once the socket is bound.
 pub const GROUP_NEXTHOP: u32 = 32;
 
-const AF_INET: u8 = 2;
-const AF_INET6: u8 = 10;
+pub(crate) const AF_INET: u8 = 2;
+pub(crate) const AF_INET6: u8 = 10;
 
 /// `RT_TABLE_COMPAT`: the table's number is in an attribute, being too
 /// large for the octet the fixed part has for it.
@@ -219,22 +223,31 @@ pub enum NetlinkError {
     Truncated { message_type: u16 },
 }
 
-fn ne_u16(bytes: &[u8]) -> u16 {
+// The kernel writes its numbers in the machine's byte order. Each reader
+// takes the first octets of what it is given, so that a caller can pass
+// the rest of a buffer rather than cut the field out first.
+pub(crate) fn ne_u16(bytes: &[u8]) -> u16 {
     u16::from_ne_bytes([bytes[0], bytes[1]])
 }
 
-fn ne_u32(bytes: &[u8]) -> u32 {
+pub(crate) fn ne_u32(bytes: &[u8]) -> u32 {
     u32::from_ne_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])
 }
 
+pub(crate) fn ne_u64(bytes: &[u8]) -> u64 {
+    u64::from_ne_bytes([
+        bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
+    ])
+}
+
 /// Rounds a length up to the four octets everything is aligned to.
-fn aligned(length: usize) -> usize {
+pub(crate) fn aligned(length: usize) -> usize {
     (length + 3) & !3
 }
 
 /// The attributes in `bytes`: each its type and value. Stops at what is
 /// not an attribute.
-fn attributes(mut bytes: &[u8]) -> impl Iterator<Item = (u16, &[u8])> {
+pub(crate) fn attributes(mut bytes: &[u8]) -> impl Iterator<Item = (u16, &[u8])> {
     std::iter::from_fn(move || {
         let header = bytes.first_chunk::<4>()?;
         let length = usize::from(ne_u16(header));
diff --git a/crates/fib/src/sockdiag.rs b/crates/fib/src/sockdiag.rs
index 636badc..7e0ad20 100644
--- a/crates/fib/src/sockdiag.rs
+++ b/crates/fib/src/sockdiag.rs
@@ -7,6 +7,10 @@
 use std::io;
 use std::net::{IpAddr, SocketAddr};
 
+use crate::netlink::{
+    AF_INET, AF_INET6, HEADER_LEN, attributes, flag, message_type, ne_u16, ne_u32, ne_u64,
+};
+
 /// A TCP connection as the kernel sees it. Times are in microseconds,
 /// sizes in octets, rates in octets per second, unless said otherwise.
 /// Fields the running kernel does not report are zero.
@@ -90,17 +94,12 @@ impl TcpConnection {
 }
 
 const SOCK_DIAG_BY_FAMILY: u16 = 20;
-const NLMSG_ERROR: u16 = 2;
-const NLM_F_REQUEST: u16 = 1;
 const IPPROTO_TCP: u8 = 6;
-const AF_INET: u8 = 2;
-const AF_INET6: u8 = 10;
 const INET_DIAG_INFO: u16 = 2;
 const INET_DIAG_CONG: u16 = 4;
 /// `idiag_ext` bits asking for those two.
 const EXTENSIONS: u8 = (1 << (INET_DIAG_INFO - 1)) | (1 << (INET_DIAG_CONG - 1));
 const NO_COOKIE: u32 = u32::MAX;
-const HEADER_LEN: usize = 16;
 const REQUEST_LEN: usize = HEADER_LEN + 56;
 const DIAG_MSG_LEN: usize = 72;
 
@@ -113,7 +112,7 @@ pub fn request(local: SocketAddr, peer: SocketAddr, sequence: u32) -> Vec<u8> {
     let mut buf = Vec::with_capacity(REQUEST_LEN);
     buf.extend_from_slice(&(REQUEST_LEN as u32).to_ne_bytes());
     buf.extend_from_slice(&SOCK_DIAG_BY_FAMILY.to_ne_bytes());
-    buf.extend_from_slice(&NLM_F_REQUEST.to_ne_bytes());
+    buf.extend_from_slice(&flag::REQUEST.to_ne_bytes());
     buf.extend_from_slice(&sequence.to_ne_bytes());
     buf.extend_from_slice(&0u32.to_ne_bytes());
     // inet_diag_req_v2
@@ -162,11 +161,11 @@ pub fn parse(buf: &[u8]) -> io::Result<Answer> {
         .get(..HEADER_LEN)
         .ok_or_else(|| io::Error::other("short netlink answer"))?;
     let length = ne_u32(&header[0..4]) as usize;
-    let message_type = ne_u16(&header[4..6]);
+    let kind = ne_u16(&header[4..6]);
     let message = buf
         .get(HEADER_LEN..length.min(buf.len()))
         .ok_or_else(|| io::Error::other("short netlink answer"))?;
-    if message_type == NLMSG_ERROR {
+    if kind == message_type::ERROR {
         let errno = -(message
             .first_chunk::<4>()
             .map_or(0, |e| i32::from_ne_bytes(*e)));
@@ -177,7 +176,7 @@ pub fn parse(buf: &[u8]) -> io::Result<Answer> {
             Answer::Refused(errno)
         });
     }
-    if message_type != SOCK_DIAG_BY_FAMILY || message.len() < DIAG_MSG_LEN {
+    if kind != SOCK_DIAG_BY_FAMILY || message.len() < DIAG_MSG_LEN {
         return Err(io::Error::other("not a socket diagnostic"));
     }
     let mut connection = TcpConnection {
@@ -186,14 +185,7 @@ pub fn parse(buf: &[u8]) -> io::Result<Answer> {
         send_queue: ne_u32(&message[60..64]),
         ..TcpConnection::default()
     };
-    let mut attributes = &message[DIAG_MSG_LEN..];
-    while attributes.len() >= 4 {
-        let length = usize::from(ne_u16(&attributes[0..2]));
-        let kind = ne_u16(&attributes[2..4]);
-        if length < 4 || length > attributes.len() {
-            break;
-        }
-        let value = &attributes[4..length];
+    for (kind, value) in attributes(&message[DIAG_MSG_LEN..]) {
         match kind {
             INET_DIAG_INFO => fill_from_tcp_info(&mut connection, value),
             INET_DIAG_CONG => {
@@ -202,7 +194,6 @@ pub fn parse(buf: &[u8]) -> io::Result<Answer> {
             }
             _ => {}
         }
-        attributes = &attributes[(length + 3) & !3..];
     }
     Ok(Answer::Connection(Box::new(connection)))
 }
@@ -250,20 +241,6 @@ fn fill_from_tcp_info(c: &mut TcpConnection, info: &[u8]) {
     c.receive_window = u32_at(232);
 }
 
-fn ne_u16(bytes: &[u8]) -> u16 {
-    u16::from_ne_bytes([bytes[0], bytes[1]])
-}
-
-fn ne_u32(bytes: &[u8]) -> u32 {
-    u32::from_ne_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])
-}
-
-fn ne_u64(bytes: &[u8]) -> u64 {
-    let mut array = [0u8; 8];
-    array.copy_from_slice(bytes);
-    u64::from_ne_bytes(array)
-}
-
 /// Asks the kernel about the TCP connection from `local` to `peer`.
 /// `Ok(None)` if there is no such connection, or on an operating system
 /// that cannot be asked.
@@ -304,9 +281,10 @@ pub async fn tcp_connection(
 mod tests {
     use super::*;
 
-    /// A reply as the kernel writes it, for an established connection
-    /// with a `tcp_info` of the full modern length.
-    fn reply() -> Vec<u8> {
+    /// A reply as the kernel writes it, for an established connection,
+    /// with the first `info_len` octets of a `tcp_info` of the full
+    /// modern length: an older kernel sends a shorter one.
+    fn reply(info_len: usize) -> Vec<u8> {
         let mut info = vec![0u8; 248];
         info[0] = 1; // established
         info[6] = 0x7a; // snd_wscale 10, rcv_wscale 7
@@ -326,6 +304,7 @@ mod tests {
         put64(&mut info, 168, 1_000_000); // busy
         put32(&mut info, 228, 131072); // snd_wnd
         put32(&mut info, 232, 262144); // rcv_wnd
+        info.truncate(info_len);
         let cong = b"cubic\0";
         let mut message = vec![0u8; DIAG_MSG_LEN];
         message[1] = 1;
@@ -345,7 +324,9 @@ mod tests {
         let mut buf = Vec::new();
         buf.extend_from_slice(&(length as u32).to_ne_bytes());
         buf.extend_from_slice(&SOCK_DIAG_BY_FAMILY.to_ne_bytes());
-        buf.extend_from_slice(&[0, 0, 1, 0, 0, 0, 0, 0, 0, 0]);
+        buf.extend_from_slice(&0u16.to_ne_bytes()); // flags
+        buf.extend_from_slice(&1u32.to_ne_bytes()); // sequence
+        buf.extend_from_slice(&0u32.to_ne_bytes()); // port
         buf.extend_from_slice(&message);
         buf.extend_from_slice(&attrs);
         buf
@@ -353,7 +334,7 @@ mod tests {
 
     #[test]
     fn a_reply_is_read_field_by_field() {
-        let Answer::Connection(c) = parse(&reply()).unwrap() else {
+        let Answer::Connection(c) = parse(&reply(248)).unwrap() else {
             panic!("a connection");
         };
         assert_eq!(c.state_name(), "established");
@@ -367,12 +348,25 @@ mod tests {
         assert_eq!((c.send_window, c.receive_window), (131072, 262144));
     }
 
+    /// Linux 4.x sends 104 octets of `tcp_info`, up to `tcpi_total_retrans`:
+    /// what is there is read, and what came later is left at zero rather
+    /// than read from the attribute after it.
+    #[test]
+    fn a_short_tcp_info_from_an_old_kernel_leaves_later_fields_zero() {
+        let Answer::Connection(c) = parse(&reply(104)).unwrap() else {
+            panic!("a connection");
+        };
+        assert_eq!(c.congestion.as_deref(), Some("cubic"));
+        assert_eq!((c.mss, c.rtt, c.cwnd), (1448, 2300, 42));
+        assert_eq!((c.busy, c.send_window), (0, 0));
+    }
+
     #[test]
     fn no_such_connection_and_a_refusal_are_told_apart() {
         let error = |errno: i32| {
             let mut buf = vec![0u8; 20];
             buf[0..4].copy_from_slice(&20u32.to_ne_bytes());
-            buf[4..6].copy_from_slice(&NLMSG_ERROR.to_ne_bytes());
+            buf[4..6].copy_from_slice(&message_type::ERROR.to_ne_bytes());
             buf[16..20].copy_from_slice(&(-errno).to_ne_bytes());
             buf
         };
diff --git a/crates/integration-tests/src/test_peer.rs b/crates/integration-tests/src/test_peer.rs
index 747d721..6da4d9c 100644
--- a/crates/integration-tests/src/test_peer.rs
+++ b/crates/integration-tests/src/test_peer.rs
@@ -2,16 +2,55 @@ use std::net::{Ipv4Addr, SocketAddr};
 use std::time::Duration;
 
 use bgp_packet::message::{BGP_HEADER_LEN, Message, NotificationMessage, UpdateMessage};
-use bgp_packet::open::{Capability, OpenMessage, OpenOption};
+use bgp_packet::open::{
+    AddPathEntry, Capability, ExtendedNextHopEntry, OpenMessage, OpenOption, SendReceive,
+};
 use bgp_primitives::constants::{Afi, AfiSafi, Safi};
 use bgp_primitives::parser::ParserContext;
 use bytes::BytesMut;
 use eyre::{Result, bail, eyre};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::tcp::OwnedWriteHalf;
 use tokio::net::{TcpListener, TcpStream};
+use tokio::task::JoinHandle;
 
 use crate::TIMEOUT;
 
+/// The capabilities a route collector offers in its OPEN: four-octet AS
+/// numbers as `asn`, each of `families`, and to be sent several routes
+/// for a prefix (ADD-PATH, RFC 7911) in each of them; with
+/// `ipv4_over_ipv6`, IPv4 routes with IPv6 next hops too (RFC 8950), as a
+/// collector on an IPv6-only session wants. For the tests and benchmarks
+/// that play a collector taking a router's whole table.
+pub fn collector_capabilities(
+    asn: u32,
+    families: &[AfiSafi],
+    ipv4_over_ipv6: bool,
+) -> Vec<Capability> {
+    let mut capabilities = vec![Capability::FourByteAsn { asn }];
+    capabilities.extend(families.iter().copied().map(Capability::multiprotocol));
+    capabilities.push(Capability::AddPath {
+        entries: families
+            .iter()
+            .map(|family| AddPathEntry {
+                afi: family.afi,
+                safi: family.safi,
+                send_recv: SendReceive::Receive,
+            })
+            .collect(),
+    });
+    if ipv4_over_ipv6 {
+        capabilities.push(Capability::ExtendedNextHop {
+            entries: vec![ExtendedNextHopEntry {
+                afi: Afi::IPV4,
+                safi: Safi::UNICAST,
+                nh_afi: Afi::IPV6,
+            }],
+        });
+    }
+    capabilities
+}
+
 /// A minimal BGP peer for use in tests. Connects over TCP and can exchange
 /// framed BGP messages using the `bgp-packet` wire format.
 pub struct TestPeer {
@@ -57,6 +96,32 @@ impl TestPeer {
         self.stream
     }
 
+    /// The connection with a task reading and dropping whatever the server
+    /// sends over it, and the half that is left for writing: for a peer
+    /// that has nothing to hear, as one feeding a table or taking one by
+    /// the megabyte has not, and whose writer must never be held up by
+    /// what it has not read. The task ends with the connection, or when
+    /// its handle is dropped.
+    pub fn drain(self) -> (OwnedWriteHalf, JoinHandle<()>) {
+        let (mut reader, writer) = self.stream.into_split();
+        let sink = tokio::spawn(async move {
+            let mut sink = vec![0u8; 1 << 16];
+            while matches!(reader.read(&mut sink).await, Ok(n) if n > 0) {}
+        });
+        (writer, sink)
+    }
+
+    /// A KEEPALIVE as it goes on the wire, for whoever writes to the
+    /// connection without the framing and has to keep the session alive
+    /// past its hold time.
+    pub fn keepalive_bytes() -> Vec<u8> {
+        let mut wire = BytesMut::new();
+        Message::KeepAlive
+            .to_wire(&ParserContext::default(), &mut wire)
+            .expect("a KEEPALIVE is nineteen octets");
+        wire.to_vec()
+    }
+
     /// Our end of the connection: what the server sees the peer as.
     pub fn local_addr(&self) -> Result<SocketAddr> {
         Ok(self.stream.local_addr()?)
@@ -145,7 +210,9 @@ impl TestPeer {
     pub async fn recv_open(&mut self) -> Result<OpenMessage> {
         match self.recv().await? {
             Message::Open(o) => {
-                // Enable 4-octet ASN parsing if the peer advertised it.
+                // AS numbers are four octets wide only if both sides
+                // offered it (RFC 6793 §4.1): the server's offer is read
+                // here, ours was noted when the OPEN went out.
                 let four_byte = o.options.iter().any(|opt| {
                     matches!(opt, OpenOption::Capabilities(caps)
                         if caps.iter().any(|c| matches!(c, Capability::FourByteAsn { .. })))
@@ -243,7 +310,8 @@ impl TestPeer {
 
     async fn finish_handshake(&mut self) -> Result<OpenMessage> {
         let server_open = self.recv_open().await?;
-        // Expect server KEEPALIVE
+        // The server's KEEPALIVE says it accepted our OPEN; anything else
+        // is its refusal, or a server out of step.
         match self.recv().await? {
             Message::KeepAlive => {}
             other => {
@@ -253,12 +321,15 @@ impl TestPeer {
                 ));
             }
         }
-        // Send our KEEPALIVE
+        // Ours says the same of its OPEN, and the session is established.
         self.send(&Message::KeepAlive).await?;
         Ok(server_open)
     }
 
-    /// Send raw bytes without framing — useful for testing invalid input.
+    /// Writes `bytes` as they are, with no framing of its own: for a test
+    /// that sends what no message would be, and for the benchmarks, which
+    /// send a table encoded once as one stream rather than message by
+    /// message.
     pub async fn send_raw(&mut self, bytes: &[u8]) -> Result<()> {
         self.stream.write_all(bytes).await?;
         Ok(())
diff --git a/crates/integration-tests/tests/add_path.rs b/crates/integration-tests/tests/add_path.rs
index 4113a7c..618739d 100644
--- a/crates/integration-tests/tests/add_path.rs
+++ b/crates/integration-tests/tests/add_path.rs
@@ -7,16 +7,15 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use bgp_integration_tests::announcement::{Announcement, withdrawal};
-use bgp_integration_tests::test_peer::TestPeer;
+use bgp_integration_tests::test_peer::{TestPeer, collector_capabilities};
 use bgp_integration_tests::test_server::TestServer;
 use bgp_integration_tests::{peer_config, prefix, wait_until};
 use bgp_packet::message::{
     AsPathAttribute, Message, MpReachNlriPathAttribute, NlriNextHop, OriginPathAttribute,
     PathAttribute, UpdateMessage,
 };
-use bgp_packet::open::{AddPathEntry, Capability, ExtendedNextHopEntry, OpenMessage, SendReceive};
+use bgp_packet::open::{AddPathEntry, Capability, 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;
@@ -350,25 +349,16 @@ async fn test_every_route_goes_to_a_collector_told_apart_by_path_identifiers() -
     b.announce(route(&[PEER_ASN + 1, 64510])).await?;
     server.wait_for_routes(2).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,
+            collector_capabilities(COLLECTOR_ASN, &SUPPORTED, false),
         )
         .await?;
-    collector.ctx.add_path = vec![AfiSafi::IPV4_UNICAST];
+    collector.ctx.add_path = SUPPORTED.to_vec();
 
     // Both routes, told apart.
     let mut held = Held::new();
@@ -437,25 +427,16 @@ async fn test_a_large_table_reaches_a_collector_that_was_there_first() -> Result
         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,
+            collector_capabilities(COLLECTOR_ASN, &SUPPORTED, false),
         )
         .await?;
-    collector.ctx.add_path = vec![AfiSafi::IPV4_UNICAST];
+    collector.ctx.add_path = SUPPORTED.to_vec();
     server.wait_for_sessions(1).await?;
 
     // Two feeders, each sending the same prefixes a hundred to an UPDATE.
@@ -518,35 +499,16 @@ async fn test_both_families_reach_a_collector_over_one_ipv6_session() -> Result<
         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,
+            collector_capabilities(COLLECTOR_ASN, &SUPPORTED, true),
         )
         .await?;
-    collector.ctx.add_path = vec![AfiSafi::IPV4_UNICAST, AfiSafi::IPV6_UNICAST];
+    collector.ctx.add_path = SUPPORTED.to_vec();
     server.wait_for_sessions(1).await?;
     let session = &server.registry.list()[0];
     assert!(session.extended_next_hop, "RFC 8950 agreed");
diff --git a/crates/packet/src/attributes.rs b/crates/packet/src/attributes.rs
index b971703..f233973 100644
--- a/crates/packet/src/attributes.rs
+++ b/crates/packet/src/attributes.rs
@@ -18,7 +18,6 @@ use bgp_primitives::constants::{AfiSafi, IpVersion, Safi};
 use bgp_primitives::ip_prefix::{IpPrefix, Nlri};
 use bgp_primitives::parser::{BgpParserError, EncodeError, ParserContext};
 
-use crate::message::write_nlri;
 use crate::wire::{
     Parsed, list, malformed, parse_afi_safi, parse_asn, put_afi_safi, put_asn, put_length, records,
     whole,
@@ -833,6 +832,19 @@ pub(crate) fn parse_nlri_list(
     list(buf, |buf| parse_nlri(version, with_path_id, buf))
 }
 
+/// Writes prefixes as `parse_nlri_list` reads them: for each, its path
+/// identifier if the session uses them for the family (RFC 7911 §3), its
+/// length in bits, and the bytes that length covers.
+pub fn put_nlri_list(nlri: &[Nlri], with_path_id: bool, out: &mut BytesMut) {
+    for n in nlri {
+        if with_path_id {
+            out.put_u32(n.path_id);
+        }
+        out.put_u8(n.prefix.length());
+        out.put(n.prefix.bytes());
+    }
+}
+
 /// The IP version of the prefixes that make up the NLRI of `family`.
 ///
 /// Any family can be named in MP_REACH_NLRI and MP_UNREACH_NLRI, but only
@@ -897,7 +909,7 @@ impl MpReachNlriPathAttribute {
         out.put_u8(self.next_hop.wire_len());
         self.next_hop.to_wire(out);
         out.put_u8(0);
-        write_nlri(&self.prefixes, ctx.has_add_path(self.family), out);
+        put_nlri_list(&self.prefixes, ctx.has_add_path(self.family), out);
     }
 }
 
@@ -919,6 +931,6 @@ impl MpUnreachNlriPathAttribute {
 
     pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) {
         put_afi_safi(self.family, out);
-        write_nlri(&self.prefixes, ctx.has_add_path(self.family), out);
+        put_nlri_list(&self.prefixes, ctx.has_add_path(self.family), out);
     }
 }
diff --git a/crates/packet/src/message.rs b/crates/packet/src/message.rs
index b34b67e..99b5f68 100644
--- a/crates/packet/src/message.rs
+++ b/crates/packet/src/message.rs
@@ -15,7 +15,7 @@ use crate::errors::DecodeError;
 use crate::errors::MsgHeaderSubcode;
 use crate::errors::OpenMsgSubcode;
 use crate::open::OpenMessage;
-use crate::wire::{fill_length, put_length};
+use crate::wire::{body_has_path_id, fill_length, put_length};
 
 /// A sentinel AS number used to denote that the actual AS number is provided as
 /// a 4 byte value in the capabilities instead.
@@ -30,10 +30,15 @@ pub const BGP_HEADER_LEN: u16 = 19;
 /// Maximum BGP message length per RFC 4271 §4.1.
 pub const BGP_MAX_MSG_LEN: u16 = 4096;
 
+/// The sixteen octets of ones that open every message (RFC 4271 §4.1).
+/// Whoever frames a message by hand wants [`begin_message`] and
+/// [`finish_message`] rather than this.
 pub const BGP_MARKER: [u8; 16] = [0xFF; 16];
 
-/// BGP message type codes (RFC 4271 §4.1, RFC 2918).
-mod msg_type {
+/// BGP message type codes (RFC 4271 §4.1, RFC 2918), for whoever frames
+/// a message with [`begin_message`] or tells messages apart by their
+/// header.
+pub mod msg_type {
     pub const OPEN: u8 = 1;
     pub const UPDATE: u8 = 2;
     pub const NOTIFICATION: u8 = 3;
@@ -162,23 +167,10 @@ impl Message {
     /// cannot be encoded, `out` is left as it was.
     pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> {
         let start = out.len();
-        out.put_slice(&BGP_MARKER);
-        // The length, filled in once the body is written. It counts the
-        // header too.
-        let length_at = out.len();
-        out.put_u16(0);
-        out.put_u8(self.type_code());
-        let written = self.body_to_wire(ctx, out).and_then(|()| {
-            let length = out.len() - start;
-            put_length(
-                out,
-                length_at,
-                2,
-                length,
-                BGP_MAX_MSG_LEN.into(),
-                "BGP message",
-            )
-        });
+        let length_at = begin_message(out, self.type_code());
+        let written = self
+            .body_to_wire(ctx, out)
+            .and_then(|()| finish_message(out, length_at));
         if let Err(e) = written {
             out.truncate(start);
             return Err(e);
@@ -215,6 +207,35 @@ impl Message {
     }
 }
 
+/// Writes the header of a message of type `kind` to the end of `out`,
+/// with its length left at zero, and returns where the length goes. The
+/// body follows, and then [`finish_message`] fills the length in: for
+/// whoever builds a message out of pieces that are not a [`Message`], as
+/// an exporter streaming a table does.
+pub fn begin_message(out: &mut BytesMut, kind: u8) -> usize {
+    out.put_slice(&BGP_MARKER);
+    let length_at = out.len();
+    out.put_u16(0);
+    out.put_u8(kind);
+    length_at
+}
+
+/// Fills in the length of the message whose header [`begin_message`] put
+/// at `length_at`, counting the header too, once the body is written. A
+/// message longer than [`BGP_MAX_MSG_LEN`] cannot be sent, and is an
+/// error; `out` is left for the caller to cut back.
+pub fn finish_message(out: &mut BytesMut, length_at: usize) -> Result<(), EncodeError> {
+    let length = out.len() - length_at + BGP_MARKER.len();
+    put_length(
+        out,
+        length_at,
+        2,
+        length,
+        BGP_MAX_MSG_LEN.into(),
+        "BGP message",
+    )
+}
+
 /// Decodes the body of an OPEN, with the NOTIFICATION that reports a
 /// failure (RFC 4271 §6.2).
 fn decode_open(body: &[u8]) -> Result<OpenMessage, DecodeError> {
@@ -256,7 +277,10 @@ pub struct Decoded {
     pub message: Message,
     /// The errors in an UPDATE that were handled without resetting the
     /// session (RFC 7606). Empty for a well-formed UPDATE and for every
-    /// other message.
+    /// other message. These are
+    /// [`DecodedUpdate::errors`](crate::update_decode::DecodedUpdate::errors),
+    /// named here for the one message that can have them, since this is
+    /// the result for any message.
     pub update_errors: Vec<UpdateError>,
     /// See [`crate::update_decode::DecodedUpdate::attributes_wire`].
     pub attributes_wire: Option<Vec<u8>>,
@@ -332,14 +356,13 @@ impl UpdateMessage {
 
     /// Writes the body of the UPDATE (what follows the 19-octet header).
     pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> {
-        // The prefixes in the body are IPv4 unicast (RFC 4271 §4.3).
-        let with_path_id = ctx.has_add_path(AfiSafi::IPV4_UNICAST);
+        let with_path_id = body_has_path_id(ctx);
 
         // Each field's two-octet length is filled in once the field is
         // written.
         let withdrawn = out.len();
         out.put_u16(0);
-        write_nlri(&self.withdrawn_routes, with_path_id, out);
+        put_nlri_list(&self.withdrawn_routes, with_path_id, out);
         fill_length(out, withdrawn, 2, u16::MAX.into(), "withdrawn routes")?;
 
         let attributes = out.len();
@@ -350,23 +373,14 @@ impl UpdateMessage {
         fill_length(out, attributes, 2, u16::MAX.into(), "path attributes")?;
 
         // The NLRI take up the rest of the message.
-        write_nlri(&self.announced_routes, with_path_id, out);
+        put_nlri_list(&self.announced_routes, with_path_id, out);
         Ok(())
     }
 }
 
-/// Writes prefixes as an UPDATE names them: for each, its path identifier
-/// if the session uses them for the family (RFC 7911 §3), its length in
-/// bits, and the bytes that length covers.
-pub fn write_nlri(nlri: &[Nlri], with_path_id: bool, out: &mut BytesMut) {
-    for n in nlri {
-        if with_path_id {
-            out.put_u32(n.path_id);
-        }
-        out.put_u8(n.prefix.length());
-        out.put(n.prefix.bytes());
-    }
-}
+/// The name [`put_nlri_list`] went by before it moved next to the parser
+/// it is the pair of; the exporter still uses it.
+pub use crate::attributes::put_nlri_list as write_nlri;
 
 /// A parser failure as text for a log line.
 pub(crate) fn describe(e: &nom::Err<BgpParserError<&[u8]>>) -> String {
@@ -409,7 +423,7 @@ mod tests {
         str::FromStr,
     };
 
-    use bytes::BytesMut;
+    use bytes::{BufMut, BytesMut};
     use eyre::{Result, eyre};
 
     use crate::message::{
@@ -1133,6 +1147,31 @@ mod tests {
         Ok(())
     }
 
+    /// A message framed by hand, as the exporter frames an UPDATE it
+    /// builds from pieces, is the message `to_wire` writes; and one that
+    /// grows past the maximum is refused with its length filled in.
+    #[test]
+    fn a_message_framed_by_hand_is_the_one_to_wire_writes() {
+        use crate::message::{begin_message, finish_message, msg_type};
+        let ctx = default_v6_context();
+        let mut whole = BytesMut::new();
+        Message::Update(UpdateMessage::end_of_rib(AfiSafi::IPV4_UNICAST))
+            .to_wire(&ctx, &mut whole)
+            .unwrap();
+
+        let mut by_hand = BytesMut::new();
+        let length_at = begin_message(&mut by_hand, msg_type::UPDATE);
+        assert_eq!(length_at, 16);
+        by_hand.put_u32(0);
+        finish_message(&mut by_hand, length_at).unwrap();
+        assert_eq!(by_hand, whole);
+
+        let mut too_long = BytesMut::new();
+        let length_at = begin_message(&mut too_long, msg_type::UPDATE);
+        too_long.put_bytes(0, 4096);
+        assert!(finish_message(&mut too_long, length_at).is_err());
+    }
+
     #[test]
     fn test_message_rejects_bad_marker() {
         let ctx = default_v6_context();
diff --git a/crates/packet/src/update_decode.rs b/crates/packet/src/update_decode.rs
index 191bc99..a4aaa50 100644
--- a/crates/packet/src/update_decode.rs
+++ b/crates/packet/src/update_decode.rs
@@ -19,7 +19,7 @@
 //! be acted on, with treat-as-withdraw and attribute discard already carried
 //! out, together with a report of what was wrong, for the log.
 
-use bgp_primitives::constants::{AfiSafi, IpVersion};
+use bgp_primitives::constants::IpVersion;
 use bgp_primitives::ip_prefix::Nlri;
 use bgp_primitives::parser::ParserContext;
 
@@ -29,6 +29,7 @@ use crate::message::{
     MpUnreachNlriPathAttribute, NotAnAttribute, PathAttribute, RawAttribute, UpdateMessage,
     defined_flags, describe, parse_nlri_list, split_attribute,
 };
+use crate::wire::body_has_path_id;
 
 /// How an error in an UPDATE was handled, short of resetting the session.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -62,7 +63,9 @@ pub struct UpdateError {
 pub struct DecodedUpdate {
     /// The UPDATE after treat-as-withdraw and attribute discard.
     pub update: UpdateMessage,
-    /// Empty when the UPDATE was well formed.
+    /// What was wrong, in the order met; empty when the UPDATE was well
+    /// formed. Where a message of any type was decoded this is
+    /// [`Decoded::update_errors`](crate::message::Decoded::update_errors).
     pub errors: Vec<UpdateError>,
     /// The attributes in `update`, as they were on the wire, for a table
     /// that keeps attributes in that form and need not write them out
@@ -102,9 +105,7 @@ fn reset(subcode: UpdateMsgSubcode, data: Vec<u8>, detail: String) -> DecodeErro
 /// error here cannot be recovered from: the prefixes are what
 /// treat-as-withdraw would act on.
 fn prefixes(ctx: &ParserContext, field: &str, buf: &[u8]) -> Result<Vec<Nlri>, DecodeError> {
-    // The prefixes in the body are IPv4 unicast (RFC 4271 §4.3).
-    let with_path_id = ctx.has_add_path(AfiSafi::IPV4_UNICAST);
-    match parse_nlri_list(IpVersion::Ipv4, with_path_id, buf) {
+    match parse_nlri_list(IpVersion::Ipv4, body_has_path_id(ctx), buf) {
         Ok((_, prefixes)) => Ok(prefixes),
         Err(e) => Err(reset(
             UpdateMsgSubcode::InvalidNetworkField,
@@ -234,7 +235,10 @@ struct AttributeReader<'a> {
     attributes: Vec<PathAttribute>,
     errors: Vec<UpdateError>,
     /// The bytes of the attributes that are kept and are a property of a
-    /// route: all but MP_REACH_NLRI and MP_UNREACH_NLRI.
+    /// route: all but MP_REACH_NLRI and MP_UNREACH_NLRI. Left empty on a
+    /// session with two-octet AS numbers, whose bytes no table keeps (see
+    /// [`DecodedUpdate::attributes_wire`]), so that nothing is copied or
+    /// reserved for them.
     kept_wire: Vec<u8>,
     /// The type codes met so far, one bit each.
     seen: [u64; 4],
@@ -246,7 +250,7 @@ impl<'a> AttributeReader<'a> {
             ctx,
             attributes: Vec::new(),
             errors: Vec::new(),
-            kept_wire: Vec::with_capacity(wire_len),
+            kept_wire: Vec::with_capacity(if ctx.four_octet_asn { wire_len } else { 0 }),
             seen: [0; 4],
         }
     }
@@ -291,7 +295,7 @@ impl<'a> AttributeReader<'a> {
                 route_property,
             } = self.judge(&raw)?
             {
-                if route_property {
+                if route_property && self.ctx.four_octet_asn {
                     self.kept_wire.extend_from_slice(raw.wire);
                 }
                 self.attributes.push(attribute);
@@ -494,6 +498,7 @@ mod tests {
     use std::net::Ipv4Addr;
 
     use crate::message::PathAttributeFlags;
+    use bgp_primitives::constants::AfiSafi;
     use bgp_primitives::ip_prefix::IpPrefix;
 
     use super::*;
diff --git a/crates/packet/src/wire.rs b/crates/packet/src/wire.rs
index ddc9d20..18ab7a0 100644
--- a/crates/packet/src/wire.rs
+++ b/crates/packet/src/wire.rs
@@ -9,7 +9,7 @@ use nom::IResult;
 use nom::number::complete::{be_u8, be_u16, be_u32};
 
 use bgp_primitives::constants::{Afi, AfiSafi, Safi};
-use bgp_primitives::parser::{BgpParserError, EncodeError};
+use bgp_primitives::parser::{BgpParserError, EncodeError, ParserContext};
 
 use crate::message::AS_TRANS;
 
@@ -91,6 +91,15 @@ pub(crate) fn put_asn(four_octet: bool, asn: u32, out: &mut BytesMut) {
     }
 }
 
+/// Whether the prefixes in the body of an UPDATE carry path identifiers
+/// on a session with `ctx`. The body holds IPv4 unicast routes and nothing
+/// else (RFC 4271 §4.3), so that is the family whose ADD-PATH agreement
+/// decides it; every other family's prefixes are in MP_REACH_NLRI and
+/// MP_UNREACH_NLRI, which name their family.
+pub(crate) fn body_has_path_id(ctx: &ParserContext) -> bool {
+    ctx.has_add_path(AfiSafi::IPV4_UNICAST)
+}
+
 /// Parses an address family as the multiprotocol attributes and most
 /// capabilities name it: a two-octet AFI and a one-octet SAFI.
 pub(crate) fn parse_afi_safi(buf: &[u8]) -> Parsed<'_, AfiSafi> {