From 0ec9bd6a412e9641f6cf0160e2dda9f4d642776c Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Fri, 2 Oct 2026 20:00:10 +0300 Subject: [PATCH 1/4] Recover inconsistent BMP state through automatic snapshot reconciliation Assisted-by: OpenAI Codex --- src/units/bmp_tcp_in/mod.rs | 1 + src/units/bmp_tcp_in/reconciliation.rs | 149 ++++++++++++++++++ src/units/bmp_tcp_in/router_handler.rs | 46 +++++- src/units/bmp_tcp_in/state_machine/machine.rs | 49 ++++++ src/units/bmp_tcp_in/unit.rs | 16 ++ 5 files changed, 260 insertions(+), 1 deletion(-) create mode 100644 src/units/bmp_tcp_in/reconciliation.rs diff --git a/src/units/bmp_tcp_in/mod.rs b/src/units/bmp_tcp_in/mod.rs index 0056b9a..c741511 100644 --- a/src/units/bmp_tcp_in/mod.rs +++ b/src/units/bmp_tcp_in/mod.rs @@ -1,5 +1,6 @@ mod io; mod metrics; +pub mod reconciliation; mod router_handler; mod state_machine; mod status_reporter; diff --git a/src/units/bmp_tcp_in/reconciliation.rs b/src/units/bmp_tcp_in/reconciliation.rs new file mode 100644 index 0000000..699bb4b --- /dev/null +++ b/src/units/bmp_tcp_in/reconciliation.rs @@ -0,0 +1,149 @@ +//! Automatic snapshot renewal for BMP state that can no longer be trusted. +//! RFC 7854 sections 3.2, 3.3 and 5: dropping TCP restarts monitoring; the +//! exporter replays established peers and their current Adj-RIBs-In. + +use serde::Deserialize; +use std::time::Duration; +use tokio::time::Instant; + +#[derive(Clone, Copy, Debug, Deserialize)] +#[serde(default, try_from = "ConfigValues")] +pub struct Config { + pub interval_secs: u64, + pub min_session_secs: u64, + pub replay_grace_secs: u64, + pub max_peer_states: usize, +} + +impl Default for Config { + fn default() -> Self { + Self { + interval_secs: 6 * 60 * 60, + min_session_secs: 60, + replay_grace_secs: 5 * 60, + max_peer_states: 65_536, + } + } +} + +#[derive(Deserialize)] +#[serde(default, deny_unknown_fields)] +struct ConfigValues { + interval_secs: u64, + min_session_secs: u64, + replay_grace_secs: u64, + max_peer_states: usize, +} + +impl Default for ConfigValues { + fn default() -> Self { + let c = Config::default(); + Self { + interval_secs: c.interval_secs, + min_session_secs: c.min_session_secs, + replay_grace_secs: c.replay_grace_secs, + max_peer_states: c.max_peer_states, + } + } +} + +impl TryFrom for Config { + type Error = &'static str; + fn try_from(c: ConfigValues) -> Result { + if c.min_session_secs == 0 + || c.min_session_secs > c.replay_grace_secs + || c.replay_grace_secs >= c.interval_secs + || c.interval_secs > 7 * 24 * 60 * 60 + || c.max_peer_states == 0 + { + return Err("BMP reconciliation requires 0 < min_session_secs <= replay_grace_secs < interval_secs <= 604800 and max_peer_states > 0"); + } + Ok(Self { + interval_secs: c.interval_secs, + min_session_secs: c.min_session_secs, + replay_grace_secs: c.replay_grace_secs, + max_peer_states: c.max_peer_states, + }) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Reason { + Periodic, + UnmatchedPeerDown, + PossibleReplacement, + PeerStateLimit, +} + +impl std::fmt::Display for Reason { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::Periodic => "periodic snapshot renewal", + Self::UnmatchedPeerDown => "unmatched Peer Down", + Self::PossibleReplacement => { + "possible unnumbered peer replacement" + } + Self::PeerStateLimit => "peer-state limit reached", + }) + } +} + +pub struct Reconciliation { + config: Config, + started: Instant, + deadline: Instant, + reason: Reason, +} + +impl Reconciliation { + pub fn new(config: Config, now: Instant, jitter_seed: u64) -> Self { + // Spread periodic refreshes over an additional 0..10% interval. + let jitter = jitter_seed % (config.interval_secs / 10 + 1); + Self { + config, + started: now, + deadline: now + + Duration::from_secs(config.interval_secs + jitter), + reason: Reason::Periodic, + } + } + + pub fn observe( + &mut self, + now: Instant, + reason: Option, + peer_states: usize, + ) { + if peer_states >= self.config.max_peer_states { + // Capacity takes precedence over the minimum session lifetime. + self.deadline = now; + self.reason = Reason::PeerStateLimit; + return; + } + let Some(reason) = reason else { return }; + if reason == Reason::PossibleReplacement + && now + < self.started + + Duration::from_secs(self.config.replay_grace_secs) + { + // Peer Ups during startup can be legitimate parallel sessions. + // There is no global end-of-snapshot marker in BMP. Do not use + // one peer's EOR as proof all peers have been enumerated. + return; + } + let deadline = now.max( + self.started + Duration::from_secs(self.config.min_session_secs), + ); + if deadline < self.deadline { + self.deadline = deadline; + self.reason = reason; + } + } + + pub fn deadline(&self) -> Instant { + self.deadline + } + pub fn reason(&self) -> Reason { + self.reason + } +} diff --git a/src/units/bmp_tcp_in/router_handler.rs b/src/units/bmp_tcp_in/router_handler.rs index a9ff76e..52ec2d7 100644 --- a/src/units/bmp_tcp_in/router_handler.rs +++ b/src/units/bmp_tcp_in/router_handler.rs @@ -72,6 +72,7 @@ pub struct RouterHandler { /// (see the unit's `forward_raw_updates` config). Captured at connection /// accept. forward_raw_updates: bool, + reconciliation: std::sync::Mutex, } impl RouterHandler { @@ -90,6 +91,7 @@ impl RouterHandler { ingress_register: Arc, ignore_post_policy_routes: bool, forward_raw_updates: bool, + reconciliation: super::reconciliation::Config, ) -> Self { Self { gate, @@ -106,6 +108,13 @@ impl RouterHandler { ingress_register, ignore_post_policy_routes, forward_raw_updates, + reconciliation: std::sync::Mutex::new( + super::reconciliation::Reconciliation::new( + reconciliation, + tokio::time::Instant::now(), + uuid::Uuid::new_v4().as_u128() as u64, + ), + ), } } @@ -153,6 +162,13 @@ impl RouterHandler { ingress_register: Default::default(), ignore_post_policy_routes: false, forward_raw_updates: false, + reconciliation: std::sync::Mutex::new( + super::reconciliation::Reconciliation::new( + Default::default(), + tokio::time::Instant::now(), + 0, + ), + ), }; (mock, gate_agent, parent_gate) @@ -213,7 +229,26 @@ impl RouterHandler { loop { // Read the incoming TCP stream, extracting BMP messages. - match stream.next().await { + let (deadline, reason) = { + let policy = self.reconciliation.lock().unwrap(); + (policy.deadline(), policy.reason()) + }; + if tokio::time::Instant::now() >= deadline { + log::warn!("Reconciling BMP speaker {router_addr}: {reason}; closing transport for a fresh snapshot"); + break; + } + // Cancelling a partial read is safe here: the transport is dropped, + // never reused. The timer also renews completely silent sessions. + let next = match tokio::time::timeout_at(deadline, stream.next()) + .await + { + Ok(next) => next, + Err(_) => { + log::warn!("Reconciling BMP speaker {router_addr}: {reason}; closing transport for a fresh snapshot"); + break; + } + }; + match next { Err(err) => { // There was a problem reading from the BMP stream. let bmp_state_lock = self.state_machine.lock().await; @@ -364,6 +399,9 @@ impl RouterHandler { } } + // Close TCP/TLS even if downstream withdrawal processing stalls. + // A timed-out partial read already dropped its owned transport. + drop(stream); let bmp_state_lock = self.state_machine.lock().await; self.status_reporter.router_connection_lost( @@ -615,7 +653,13 @@ impl RouterHandler { self.status_reporter .message_processed(bmp_state.router_id()); + let reason = bmp_state.reconciliation_reason(&msg); let mut res = bmp_state.process_msg(received, msg, trace_id); + self.reconciliation.lock().unwrap().observe( + tokio::time::Instant::now(), + reason, + res.next_state.peer_state_count(), + ); match res.message_type { MessageType::InvalidMessage { .. } => { diff --git a/src/units/bmp_tcp_in/state_machine/machine.rs b/src/units/bmp_tcp_in/state_machine/machine.rs index feab47f..e172d27 100644 --- a/src/units/bmp_tcp_in/state_machine/machine.rs +++ b/src/units/bmp_tcp_in/state_machine/machine.rs @@ -285,6 +285,55 @@ where } impl BmpState { + pub(in crate::units::bmp_tcp_in) fn peer_states( + &self, + ) -> Option<&PeerStates> { + match self { + Self::Dumping(s) => Some(&s.details.peer_states), + Self::Updating(s) => Some(&s.details.peer_states), + _ => None, + } + } + + pub(in crate::units::bmp_tcp_in) fn peer_state_count(&self) -> usize { + self.peer_states().map_or(0, |p| p.0.len()) + } + + /// Evidence to refresh the speaker's snapshot, never authority to + /// delete a guessed peer. Policy siblings share a logical identity. + pub(in crate::units::bmp_tcp_in) fn reconciliation_reason( + &self, + msg: &BmpMsg, + ) -> Option { + use super::super::reconciliation::Reason; + let peers = self.peer_states()?; + let pph = match msg { + BmpMsg::PeerDownNotification(m) => m.per_peer_header(), + BmpMsg::PeerUpNotification(m) => m.per_peer_header(), + _ => return None, + }; + let identity = |k: &PerPeerHeader| { + k.peer_type() == pph.peer_type() + && k.distinguisher() == pph.distinguisher() + && k.asn() == pph.asn() + && k.bgp_id() == pph.bgp_id() + }; + let exact = peers + .0 + .keys() + .any(|k| identity(k) && k.address() == pph.address()); + match msg { + BmpMsg::PeerDownNotification(_) if !exact => Some(Reason::UnmatchedPeerDown), + BmpMsg::PeerUpNotification(_) if !exact + && matches!(pph.address(), IpAddr::V6(a) if a.is_unicast_link_local()) + && peers.0.keys().any(|k| identity(k) + && matches!(k.address(), IpAddr::V6(a) if a.is_unicast_link_local())) => { + Some(Reason::PossibleReplacement) + } + _ => None, + } + } + pub fn ingress_id(&self) -> ingress::IngressId { match self { BmpState::Initiating(v) => v.ingress_id, diff --git a/src/units/bmp_tcp_in/unit.rs b/src/units/bmp_tcp_in/unit.rs index 7a89733..d795ef1 100644 --- a/src/units/bmp_tcp_in/unit.rs +++ b/src/units/bmp_tcp_in/unit.rs @@ -164,6 +164,10 @@ pub struct BmpTcpIn { /// connections accepted after a (re)configure. #[serde(default = "BmpTcpIn::default_forward_raw_updates")] pub forward_raw_updates: bool, + + /// Always-on snapshot reconciliation; timing and capacity bounds. + #[serde(default)] + pub reconciliation: super::reconciliation::Config, } impl BmpTcpIn { @@ -243,6 +247,7 @@ impl BmpTcpIn { ingress_register, self.ignore_post_policy_routes, self.forward_raw_updates, + self.reconciliation, ) .run::<_, _, StandardTcpStream, BmpTcpInRunner>(Arc::new( StandardTcpListenerFactory, @@ -309,6 +314,7 @@ struct BmpTcpInRunner { ingress_register: Arc, ignore_post_policy_routes: bool, forward_raw_updates: bool, + reconciliation: super::reconciliation::Config, } impl BmpTcpInRunner { @@ -337,6 +343,7 @@ impl BmpTcpInRunner { ingress_register: Arc, ignore_post_policy_routes: bool, forward_raw_updates: bool, + reconciliation: super::reconciliation::Config, ) -> Self { Self { component, @@ -357,6 +364,7 @@ impl BmpTcpInRunner { ingress_register, ignore_post_policy_routes, forward_raw_updates, + reconciliation, } } @@ -388,6 +396,7 @@ impl BmpTcpInRunner { roto_metrics: Default::default(), ignore_post_policy_routes: false, forward_raw_updates: false, + reconciliation: Default::default(), }; (runner, gate_agent) @@ -660,6 +669,7 @@ impl BmpTcpInRunner { self.ingress_register.clone(), self.ignore_post_policy_routes, self.forward_raw_updates, + self.reconciliation, ); (child_name, router_handler, router_ingress_id) @@ -694,6 +704,7 @@ impl BmpTcpInRunner { new_ignore_post_policy_routes, forward_raw_updates: new_forward_raw_updates, + reconciliation: new_reconciliation, }), } => { // Runtime reconfiguration of this unit has @@ -717,6 +728,7 @@ impl BmpTcpInRunner { new_ignore_post_policy_routes; self.forward_raw_updates = new_forward_raw_updates; + self.reconciliation = new_reconciliation; if rebind { // Trigger re-binding to the new listen port. @@ -1127,6 +1139,7 @@ mod tests { tracing_mode: Default::default(), ignore_post_policy_routes: false, forward_raw_updates: false, + reconciliation: Default::default(), }; let new_config = Unit::BmpTcpIn(new_config); agent.reconfigure(new_config, new_gate).await.unwrap(); @@ -1197,6 +1210,7 @@ mod tests { tracing_mode: Default::default(), ignore_post_policy_routes: false, forward_raw_updates: false, + reconciliation: Default::default(), }; let new_config = Unit::BmpTcpIn(new_config); agent.reconfigure(new_config, new_gate).await.unwrap(); @@ -1271,6 +1285,7 @@ mod tests { tracing_mode: Default::default(), ignore_post_policy_routes: false, forward_raw_updates: false, + reconciliation: Default::default(), }; let new_config = Unit::BmpTcpIn(new_config); agent.reconfigure(new_config, new_gate).await.unwrap(); @@ -1434,6 +1449,7 @@ mod tests { roto_metrics: Default::default(), ignore_post_policy_routes: false, forward_raw_updates: false, + reconciliation: Default::default(), }; (runner, gate_agent, status_reporter) From a816a125385f5de57c3714075ec3fa26130cde0f Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Fri, 2 Oct 2026 20:05:44 +0300 Subject: [PATCH 2/4] Add BMP reconciliation lifecycle and authoritative replay simulations Assisted-by: OpenAI Codex --- scripts/bmp-peer-churn.py | 302 ++++++++++++++++++++ src/units/bmp_tcp_in/reconciliation.rs | 104 +++++++ src/units/bmp_tcp_in/router_handler.rs | 44 +++ src/units/bmp_tcp_in/state_machine/tests.rs | 158 +++++++++- src/units/bmp_tcp_in/unit.rs | 34 +++ 5 files changed, 640 insertions(+), 2 deletions(-) create mode 100644 scripts/bmp-peer-churn.py diff --git a/scripts/bmp-peer-churn.py b/scripts/bmp-peer-churn.py new file mode 100644 index 0000000..9ac6336 --- /dev/null +++ b/scripts/bmp-peer-churn.py @@ -0,0 +1,302 @@ +#!/usr/bin/env python3 +"""Exercise automatic BMP reconciliation against isolated real daemons. + +Reconnection replays the exporter's authoritative current peers and routes. +Includes issue #11 churn, parallel sessions, silent disappearance, partial +frames and capacity limits. Python stdlib only; exit 0 on success, 1 on failure. +""" + +import argparse +import ipaddress +import json +import os +from pathlib import Path +import socket +import struct +import subprocess +import tempfile +import time +import urllib.error +import urllib.request + + +def bmp(kind, body): + return struct.pack("!BIB", 3, len(body) + 6, kind) + body + + +def bgp(kind, body): + return b"\xff" * 16 + struct.pack("!HB", len(body) + 19, kind) + body + + +def header(address): + # Global instance, IPv6 peer, pre-policy, fixed ASN and BGP identifier. + return (bytes([0, 0x80]) + bytes(8) + + socket.inet_pton(socket.AF_INET6, address) + + struct.pack("!I", 65001) + socket.inet_aton("10.0.0.7") + + bytes(8)) + + +def peer_up(address): + caps = bytes([1, 4, 0, 1, 0, 1, 65, 4]) + struct.pack("!I", 65001) + opt = bytes([2, len(caps)]) + caps + def open_msg(asn, identifier): + return bgp(1, struct.pack("!BHH", 4, asn, 0) + + socket.inet_aton(identifier) + bytes([len(opt)]) + opt) + return bmp(3, header(address) + + socket.inet_pton(socket.AF_INET6, "fe80::ffff") + + struct.pack("!HH", 179, 33001) + + open_msg(65001, "10.0.0.254") + + open_msg(65001, "10.0.0.7")) + + +def announce(address): + # One identical IPv4 prefix via every incarnation of the IPv6 peer. + attrs = bytes([0x40, 1, 1, 0, 0x40, 2, 6, 2, 1]) + attrs += struct.pack("!I", 65001) + attrs += bytes([0x40, 3, 4]) + socket.inet_aton("10.0.0.7") + update = bgp(2, bytes(2) + struct.pack("!H", len(attrs)) + + attrs + bytes([32, 10, 0, 0, 7])) + return bmp(0, header(address) + update) + + +def free_port(): + # Small bind race, as with other local integration harnesses. Startup + # checks below fail explicitly if another process takes the port. + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def wait_until(read, predicate, timeout, description): + end = time.monotonic() + timeout + last = None + while time.monotonic() < end: + last = read() + if predicate(last): + return last + time.sleep(0.1) + raise RuntimeError(f"timeout waiting for {description}: {last}") + + +class Exporter: + """Authoritative exporter: reconnects replay only current sessions.""" + + def __init__(self, port, snapshot): + self.port = port + self.snapshot = snapshot + self.current = [] + self.socket = None + self.replays = 0 + + def connect(self): + self.socket = socket.create_connection(("127.0.0.1", self.port), timeout=5) + name = b"peer-churn-feeder" + self.socket.sendall(bmp(4, struct.pack("!HH", 2, len(name)) + name)) + self.replays += 1 + for address in self.current: + self.socket.sendall(peer_up(address) + announce(address)) + # Empty IPv4 UPDATE = End-of-RIB for this peer. + self.socket.sendall(bmp(0, header(address) + bgp(2, bytes(4)))) + wait_until(self.snapshot, + lambda s: set(s["addresses"]) == set(self.current), + 5, "authoritative replay") + + def send(self, data): + try: + self.socket.sendall(data) + except (BrokenPipeError, ConnectionResetError): + # Reset can precede the new Peer Up; replay supplies it instead. + pass + + def expect_reset(self, timeout=5): + self.socket.settimeout(timeout) + try: + data = self.socket.recv(1) + if data: + raise RuntimeError("collector unexpectedly sent BMP data") + except ConnectionResetError: + pass + except socket.timeout as error: + raise RuntimeError("collector failed to reconcile") from error + self.socket.close() + self.socket = None + + def expect_open(self, seconds): + self.socket.settimeout(seconds) + try: + self.socket.recv(1) + except socket.timeout: + return + except ConnectionResetError: + pass + raise RuntimeError("unexpected reconnect: parallel replay/control must remain connected") + + def close(self): + if self.socket: + self.socket.close() + self.socket = None + + +def simulate(binary, scenario, cycles, grace): + http_port, bmp_port = free_port(), free_port() + while bmp_port == http_port: + bmp_port = free_port() + # Production defaults stay enabled. Shorten only deadlines for the test. + interval = 6 if scenario in {"silent", "partial"} else 120 + limit = 3 if scenario == "limit" else 65536 + with tempfile.TemporaryDirectory(prefix="netom-peer-churn-") as directory: + directory = Path(directory) + config = directory / "netom.conf" + config.write_text(f'''http_listen = ["127.0.0.1:{http_port}"] +[units.bmp-in] +type = "bmp-tcp-in" +listen = "127.0.0.1:{bmp_port}" +reconciliation = {{ interval_secs = {interval}, min_session_secs = 1, replay_grace_secs = 1, max_peer_states = {limit} }} +[units.rib] +type = "rib" +sources = ["bmp-in"] +[targets.null] +type = "null-out" +sources = ["rib"] +''') + env = dict(os.environ, NETOM_RIB_GC_INTERVAL_SECS="1", + NETOM_RIB_REAP_EVERY_TICKS="1") + with (directory / "netom.log").open("w+") as log: + process = subprocess.Popen([str(binary), "-c", str(config)], + stdout=log, stderr=log, env=env) + + def get(path): + if process.poll() is not None: + raise RuntimeError(f"daemon exited: {process.returncode}") + with urllib.request.urlopen( + f"http://127.0.0.1:{http_port}{path}", timeout=3) as res: + return res.read().decode() + + def snapshot(): + peers = [p for p in json.loads(get("/api/v1/ingresses"))["data"] + if p.get("ingress_type") == "bgpViaBmp"] + records = None + for line in get("/metrics").splitlines(): + name = line.split("{", 1)[0].split(" ", 1)[0] + if name == "netom_rib_unit_num_items_total": + records = (records or 0) + int(float(line.rsplit(" ", 1)[1])) + if records is None: + raise RuntimeError("RIB record metric missing") + return {"peers": len(peers), "records": records, + "addresses": sorted(p["remote_addr"] for p in peers + if p.get("state") == "Connected")} + + exporter = Exporter(bmp_port, snapshot) + try: + def ready(): + try: + return bool(get("/metrics")) + except (OSError, urllib.error.URLError): + return False + wait_until(ready, bool, 30, "HTTP startup") + exporter.current = ["fe80::1"] + if scenario == "parallel": + exporter.current.append("fe80::2") + exporter.connect() + wait_until(snapshot, lambda s: s["records"] >= len(exporter.current), + 5, "initial routes") + if scenario in {"matching", "mismatched", "missing"}: + for i in range(1, cycles + 1): + # Move beyond the replay grace before introducing a new + # address. During replay it is deliberately not guessed + # to be a replacement. + if scenario != "matching": + time.sleep(1.1) + old = exporter.current[0] + address = str(ipaddress.IPv6Address(int(ipaddress.IPv6Address("fe80::1")) + i)) + exporter.current = [address] + if scenario != "missing": + down = old if scenario == "matching" else "::" + exporter.send(bmp(2, header(down) + bytes([4]))) + exporter.send(peer_up(address) + announce(address)) + if scenario != "matching": + exporter.expect_reset() + exporter.connect() + else: + wait_until(snapshot, lambda s: s["addresses"] == [address], + 5, "matching Peer Down replacement") + if i in {1, cycles // 2, cycles}: + print(f"{scenario}: replacements={i}, replays={exporter.replays}: {snapshot()}", flush=True) + elif scenario == "parallel": + # Startup siblings must coexist without an anomaly reset. + exporter.expect_open(1.3) + exporter.current.append("fe80::3") + exporter.send(peer_up("fe80::3") + announce("fe80::3")) + exporter.expect_reset() + exporter.connect() # replay all three legitimate peers + exporter.expect_open(2) + elif scenario in {"silent", "partial"}: + exporter.current = [] # real peer vanished; no Peer Down + if scenario == "partial": + exporter.send(bytes([3, 0, 0])) # stalled common header + exporter.expect_reset(timeout=9) + exporter.connect() # empty authoritative snapshot + elif scenario == "limit": + # Startup growth must still be bounded by the capacity guard. + exporter.send(peer_up("fe80::2") + announce("fe80::2") + + peer_up("fe80::3")) + exporter.expect_reset() + exporter.connect() # only fe80::1 still exists + + expected = len(exporter.current) + state = wait_until(snapshot, lambda s: s["peers"] == expected + and s["records"] == expected + and set(s["addresses"]) == set(exporter.current), + grace, "reclamation after authoritative replay") + # The control and parallel case must remain live; no false pass + # because the collector disconnected and dropped everything. + if expected: + exporter.expect_open(0.2) + print(f"{scenario}: PASS, replays={exporter.replays}, after GC: {state}", flush=True) + exporter.close() + wait_until(snapshot, lambda s: s["peers"] == 0 and s["records"] == 0, + grace, "final BMP disconnect cleanup") + return True + except Exception: + log.flush() + log.seek(0) + print("Daemon log (tail):\n" + log.read()[-6000:]) + raise + finally: + exporter.close() + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--binary", type=Path, help="existing netom binary; otherwise build release") + parser.add_argument("--cycles", type=int, default=5, help="peer replacements (default: 5)") + parser.add_argument("--grace", type=float, default=10, help="GC wait deadline in seconds (minimum: 4)") + scenarios = ["matching", "mismatched", "missing", "parallel", "silent", "partial", "limit"] + parser.add_argument("--scenario", choices=["all"] + scenarios, default="all") + args = parser.parse_args() + if args.cycles < 1 or not 4 <= args.grace < float("inf"): + parser.error("cycles must be positive and grace must be finite and >= 4 seconds") + root = Path(__file__).resolve().parent.parent + if args.binary is None: + subprocess.run(["cargo", "build", "--release", "--bin", "netom"], cwd=root, check=True) + args.binary = root / "target/release/netom" + if args.scenario != "all": + scenarios = [args.scenario] + for scenario in scenarios: + simulate(args.binary.resolve(), scenario, args.cycles, args.grace) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except (OSError, RuntimeError, ValueError, subprocess.SubprocessError) as error: + print(f"FAIL: {error}", flush=True) + raise SystemExit(1) diff --git a/src/units/bmp_tcp_in/reconciliation.rs b/src/units/bmp_tcp_in/reconciliation.rs index 699bb4b..df401b4 100644 --- a/src/units/bmp_tcp_in/reconciliation.rs +++ b/src/units/bmp_tcp_in/reconciliation.rs @@ -147,3 +147,107 @@ impl Reconciliation { self.reason } } + +#[cfg(test)] +mod tests { + use super::*; + + fn config() -> Config { + Config { + interval_secs: 100, + min_session_secs: 5, + replay_grace_secs: 10, + max_peer_states: 8, + } + } + + #[test] + fn startup_parallel_peers_do_not_cause_a_reconnect_loop() { + let now = Instant::now(); + // Repeat the same authoritative replay over successive connections. + for generation in 0..3 { + let start = now + Duration::from_secs(generation * 100); + let mut p = Reconciliation::new(config(), start, 0); + p.observe( + start + Duration::from_secs(1), + Some(Reason::PossibleReplacement), + 2, + ); + p.observe( + start + Duration::from_secs(9), + Some(Reason::PossibleReplacement), + 3, + ); + assert_eq!(p.deadline(), start + Duration::from_secs(100)); + assert_eq!(p.reason(), Reason::Periodic); + } + } + + #[test] + fn anomalies_coalesce_without_postponing_recovery() { + let now = Instant::now(); + let mut p = Reconciliation::new(config(), now, 0); + p.observe(now, Some(Reason::UnmatchedPeerDown), 1); + assert_eq!(p.deadline(), now + Duration::from_secs(5)); + for second in 1..5 { + p.observe( + now + Duration::from_secs(second), + Some(Reason::UnmatchedPeerDown), + 1, + ); + assert_eq!(p.deadline(), now + Duration::from_secs(5)); + } + assert_eq!(p.reason(), Reason::UnmatchedPeerDown); + } + + #[test] + fn incremental_replacement_refreshes_without_guessing_a_victim() { + let now = Instant::now(); + let mut p = Reconciliation::new(config(), now, 0); + let after_replay = now + Duration::from_secs(10); + p.observe(after_replay, Some(Reason::PossibleReplacement), 2); + assert_eq!(p.deadline(), after_replay); + assert_eq!(p.reason(), Reason::PossibleReplacement); + } + + #[test] + fn capacity_overrides_cooldown_and_periodic_jitter_is_bounded() { + let now = Instant::now(); + for seed in [0, 1, 9, 10, 11, u64::MAX] { + let mut p = Reconciliation::new(config(), now, seed); + assert!(p.deadline() >= now + Duration::from_secs(100)); + assert!(p.deadline() <= now + Duration::from_secs(110)); + p.observe(now, None, 8); + assert_eq!(p.deadline(), now); + assert_eq!(p.reason(), Reason::PeerStateLimit); + } + } + + #[test] + fn configuration_cannot_disable_reconciliation() { + #[derive(Deserialize)] + struct Wrapper { + reconciliation: Config, + } + let defaults: Wrapper = toml::from_str("[reconciliation]").unwrap(); + assert_eq!(defaults.reconciliation.interval_secs, 21600); + for fields in [ + "interval_secs = 0", + "interval_secs = 604801", + "min_session_secs = 0", + "replay_grace_secs = 0", + "max_peer_states = 0", + "enabled = false", + "min_session_secs = 301", + "replay_grace_secs = 21600", + ] { + assert!( + toml::from_str::(&format!( + "[reconciliation]\n{fields}" + )) + .is_err(), + "{fields}" + ); + } + } +} diff --git a/src/units/bmp_tcp_in/router_handler.rs b/src/units/bmp_tcp_in/router_handler.rs index 52ec2d7..7bf22cd 100644 --- a/src/units/bmp_tcp_in/router_handler.rs +++ b/src/units/bmp_tcp_in/router_handler.rs @@ -808,6 +808,50 @@ mod tests { const SYS_DESCR: &str = "some-sys-desc"; const OTHER_SYS_NAME: &str = "other-sys-name"; + #[tokio::test(start_paused = true)] + async fn reconciliation_closes_idle_and_partial_transports() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + for partial in [false, true] { + let (runner, _agent, _parent) = RouterHandler::mock(); + *runner.reconciliation.lock().unwrap() = + super::super::reconciliation::Reconciliation::new( + super::super::reconciliation::Config { + interval_secs: 3, + min_session_secs: 1, + replay_grace_secs: 1, + max_peer_states: 8, + }, + tokio::time::Instant::now(), + 0, + ); + let (mut writer, reader) = tokio::io::duplex(128); + if partial { + writer.write_all(&[3, 0, 0]).await.unwrap(); + } + let register = Arc::new(ingress::Register::default()); + let id = register.register(); + runner + .read_from_router( + reader, + "127.0.0.1:12345".parse().unwrap(), + id, + register.clone(), + Arc::default(), + Arc::default(), + ) + .await; + assert_eq!( + writer.read(&mut [0]).await.unwrap(), + 0, + "transport must close on timer" + ); + assert_eq!( + register.get(id).unwrap().state, + Some(IngressState::Disconnected) + ); + } + } + #[tokio::test(flavor = "multi_thread")] async fn terminate_on_loss_of_parent_gate() { let (runner, _gate_agent, parent_gate) = RouterHandler::mock(); diff --git a/src/units/bmp_tcp_in/state_machine/tests.rs b/src/units/bmp_tcp_in/state_machine/tests.rs index ffd62cb..c4574bb 100644 --- a/src/units/bmp_tcp_in/state_machine/tests.rs +++ b/src/units/bmp_tcp_in/state_machine/tests.rs @@ -2144,6 +2144,148 @@ fn route_attribute_changes() { // BGP protocol." } +#[test] +fn reconciliation_detects_inconsistency_without_retiring_parallel_peers() { + use crate::ingress::register::{IngressState, IngressType}; + use crate::units::bmp_tcp_in::reconciliation::Reason; + let register = Arc::default(); + let mut state = mk_test_processor_with_register(®ister) + .process_msg( + Instant::now(), + mk_initiation_msg(TEST_ROUTER_SYS_NAME, TEST_ROUTER_SYS_DESC), + None, + ) + .next_state; + let mut first = mk_per_peer_header("fe80::1", 65001); + first.peer_flags = 0x80; + state = state + .process_msg( + Instant::now(), + mk_addpath_peer_up_notification_msg(&first), + None, + ) + .next_state; + state = state + .process_msg( + Instant::now(), + mk_addpath_v4_route_monitoring_msg(&first, &[1, 2]), + None, + ) + .next_state; + let mut post = mk_per_peer_header("fe80::1", 65001); + post.peer_flags = 0xc0; + let policy_up = mk_peer_up_notification_msg(&post, false).0; + assert_eq!(state.reconciliation_reason(&policy_up), None); + state = state + .process_msg( + Instant::now(), + mk_addpath_v4_route_monitoring_msg(&post, &[3]), + None, + ) + .next_state; + let mut replacement = mk_per_peer_header("fe80::2", 65001); + replacement.peer_flags = 0x80; + let up = mk_peer_up_notification_msg(&replacement, false).0; + assert_eq!( + state.reconciliation_reason(&up), + Some(Reason::PossibleReplacement) + ); + // Detection must not disconnect either identity or its policy/path children. + state = state + .process_msg(Instant::now(), up.clone(), None) + .next_state; + assert_eq!(state.peer_state_count(), 3); + assert_eq!(state.reconciliation_reason(&up), None); + let peers: Vec<_> = register + .cloned_info() + .into_iter() + .filter(|(_, i)| { + matches!( + i.ingress_type, + Some(IngressType::BgpViaBmp | IngressType::BgpPath) + ) + }) + .collect(); + assert_eq!(peers.len(), 6); + assert!(peers + .iter() + .all(|(_, i)| i.state == Some(IngressState::Connected))); + let mut down = mk_per_peer_header("::", 65001); + down.peer_flags = 0x80; + assert_eq!( + state.reconciliation_reason(&mk_peer_down_notification_msg(&down)), + Some(Reason::UnmatchedPeerDown) + ); + assert_eq!( + state.reconciliation_reason(&mk_peer_down_notification_msg(&post)), + None + ); + let BmpState::Dumping(inner) = state else { + panic!("expected dumping") + }; + let state = BmpState::Updating(inner.into()); + assert_eq!( + state.reconciliation_reason(&mk_peer_down_notification_msg(&down)), + Some(Reason::UnmatchedPeerDown) + ); + let withdrawn = state.disconnect_into_register(®ister); + let ids: std::collections::HashSet<_> = + withdrawn.iter().map(|(id, _)| *id).collect(); + assert_eq!(ids, peers.iter().map(|(id, _)| *id).collect()); + assert!(peers.iter().all(|(id, _)| register.get(*id).unwrap().state + == Some(IngressState::Disconnected))); + // A new speaker/snapshot has no knowledge of the old peers, even with a shared register. + let fresh = mk_test_processor_with_register(®ister) + .process_msg( + Instant::now(), + mk_initiation_msg("other", TEST_ROUTER_SYS_DESC), + None, + ) + .next_state; + assert_eq!(fresh.reconciliation_reason(&up), None); +} + +#[test] +fn reconciliation_does_not_conflate_peer_scopes() { + let mut first = mk_per_peer_header("fe80::1", 65001); + first.peer_flags = 0x80; + let state = mk_test_processor() + .process_msg( + Instant::now(), + mk_initiation_msg(TEST_ROUTER_SYS_NAME, TEST_ROUTER_SYS_DESC), + None, + ) + .next_state + .process_msg( + Instant::now(), + mk_peer_up_notification_msg(&first, false).0, + None, + ) + .next_state; + for case in 0..5 { + let mut next = mk_per_peer_header("fe80::2", 65001); + next.peer_flags = 0x80; + match case { + 0 => next.peer_as = Asn::from_u32(65002), + 1 => next.peer_bgp_id = [4, 3, 2, 1], + 2 => next.peer_distinguisher = [1; 8], + 3 => { + next.peer_type = + routecore::bmp::message::PeerType::RdInstance.into() + } + 4 => next.peer_address = "2001:db8::1".parse().unwrap(), + _ => unreachable!(), + } + assert_eq!( + state.reconciliation_reason( + &mk_peer_up_notification_msg(&next, false).0 + ), + None, + "case {case}" + ); + } +} + // --- Test helpers ----------------------------------------------------------------------------------------------- // RFC 4724 Graceful Restart Mechanism for BGP @@ -2176,7 +2318,13 @@ fn mk_peer_up_notification_msg( ) -> (BmpMsg, PerPeerHeader) { let bytes = crate::bgp::encode::mk_peer_up_notification_msg( pph, - "10.0.0.1".parse().unwrap(), + (if pph.peer_flags & 0x80 != 0 { + "fe80::ffff" + } else { + "10.0.0.1" + }) + .parse() + .unwrap(), 11019, 4567, 111, @@ -2332,7 +2480,13 @@ fn mk_addpath_peer_up_notification_msg( let bytes = crate::bgp::encode::mk_peer_up_notification_msg_with_capabilities( pph, - "10.0.0.1".parse().unwrap(), + (if pph.peer_flags & 0x80 != 0 { + "fe80::ffff" + } else { + "10.0.0.1" + }) + .parse() + .unwrap(), 11019, 4567, 111, diff --git a/src/units/bmp_tcp_in/unit.rs b/src/units/bmp_tcp_in/unit.rs index d795ef1..5592026 100644 --- a/src/units/bmp_tcp_in/unit.rs +++ b/src/units/bmp_tcp_in/unit.rs @@ -952,11 +952,45 @@ mod tests { ) { let (mut runner, agent) = BmpTcpInRunner::_mock(); runner.connection = config.connection; + runner.reconciliation = config.reconciliation; let states = runner.router_states.clone(); let task = tokio::spawn(runner.run::<_, _, crate::common::net::StandardTcpStream, BmpTcpInRunner>(Arc::new(crate::common::net::StandardTcpListenerFactory))); (agent, task, states) } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn active_reconciliation_redials_after_partial_snapshot() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = + tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let config = mk_config_from_toml(&format!("connect = '{}'\nreconciliation = {{ interval_secs = 3, min_session_secs = 1, replay_grace_secs = 1 }}", listener.local_addr().unwrap())).unwrap(); + let (agent, task, states) = start_active(config); + let (mut first, _) = + timeout(Duration::from_secs(2), listener.accept()) + .await + .unwrap() + .unwrap(); + first.write_all(&[3, 0, 0]).await.unwrap(); + assert_eq!( + timeout(Duration::from_secs(5), first.read(&mut [0])) + .await + .unwrap() + .unwrap(), + 0 + ); + let (_second, _) = timeout(Duration::from_secs(3), listener.accept()) + .await + .unwrap() + .unwrap(); + agent.terminate().await; + timeout(Duration::from_secs(2), task) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert!(states.is_empty()); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn active_reconnects_after_eof_and_cleans_up_on_shutdown() { let listener = From 9a851d64c2e308e238743fdd5203665583889fe5 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Fri, 2 Oct 2026 20:10:35 +0300 Subject: [PATCH 3/4] Document BMP reconciliation defaults, RFC basis and monitoring caveats Assisted-by: OpenAI Codex --- TESTING.md | 52 ++++++++++++++++++++++++++ docs/bmp-tcp-in.md | 92 +++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 143 insertions(+), 1 deletion(-) diff --git a/TESTING.md b/TESTING.md index a21517a..e64bc8a 100644 --- a/TESTING.md +++ b/TESTING.md @@ -80,3 +80,55 @@ validity states. The `e2e-flowspec` GitHub workflow runs both scripts on demand (workflow_dispatch). + +## BMP reconciliation simulations + +`python3 scripts/bmp-peer-churn.py` builds and runs isolated real daemons, then +acts as a BMP exporter over loopback TCP. No FRR, containers, external data, or +third-party Python packages are required. It exercises the automatic recovery +for [issue #11](https://github.com/FastNetMon/netom/issues/11). + +```sh +python3 scripts/bmp-peer-churn.py --binary target/debug/netom +python3 scripts/bmp-peer-churn.py --binary target/debug/netom --cycles 20 +``` + +The exporter keeps an independent list of actually established peers. After a +collector-initiated disconnect, it reconnects and replays only those peers and +their routes, including End-of-RIB. Retired identities are not replayed. + +Seven scenarios run by default; select one with `--scenario NAME`: + +- `matching`: correctly addressed Peer Down, ordinary cleanup control. +- `mismatched`: Peer Down with `::`, then a new link-local peer address. +- `missing`: a new peer address with no preceding Peer Down. +- `parallel`: legitimate parallel peers coexist in startup; adding another + later causes one reconciliation, and all three survive the new snapshot + beyond its startup window without another anomaly reset. +- `silent`: a peer vanishes without any further BMP message; periodic renewal + removes it after the authoritative empty replay. +- `partial`: the same disappearance with a stalled, incomplete BMP frame. +- `limit`: startup churn reaches the peer-state cap; reconnect restores the + sole actually established peer. + +The script shortens reconciliation timings in the test configuration and GC +intervals in the test daemon's environment. It verifies current peer addresses, +retained peer entries and route records after GC, then checks complete cleanup +when the exporter closes. The table is constant-size during churn. A timeout, +missing metric, failed replay, unexpected parallel-session reset, or retained +state makes the script exit 1; success exits 0. These are object-count checks, +not RSS thresholds: allocator caching can retain pages after objects are freed. + +Rust regressions are part of the normal suite: + +```sh +cargo test --lib units::bmp_tcp_in +``` + +They cover deadline coalescing, startup replay, periodic jitter, capacity, +configuration validation, peer identity scopes, policy/ADD-PATH teardown, +idle/partial transport cancellation, and active-connector redial. Live network +tests require permission to bind loopback sockets. + +See [automatic reconciliation](docs/bmp-tcp-in.md#automatic-reconciliation-and-stale-peer-retention) +for the RFC 7854 references, defaults, and monitoring-gap/replay caveat. diff --git a/docs/bmp-tcp-in.md b/docs/bmp-tcp-in.md index b7d7607..50f2c81 100644 --- a/docs/bmp-tcp-in.md +++ b/docs/bmp-tcp-in.md @@ -56,9 +56,99 @@ connect, handshake, idle reads and retry delays. Downstream backpressure can still delay cleanup; the daemon's shutdown/drain rules apply. The existing filtering and tracing settings apply to active input as well. -`ignore_post_policy_routes` and `forward_raw_updates` changes apply to the next +`ignore_post_policy_routes`, `forward_raw_updates`, and `reconciliation` changes apply to the next connection. Listening address changes preserve existing accepted connections. +## Automatic reconciliation and stale-peer retention + +Reconciliation is enabled for every BMP input, for both `listen` and `connect`. +No opt-in is needed. The former experimental `recover_unnumbered_peers` heuristic +has been replaced: Netom does not infer that a shared BGP identifier authorizes +it to delete another peer. + +An exporter can omit Peer Down or send it with an address that no longer matches +Peer Up. [Issue #11](https://github.com/FastNetMon/netom/issues/11) describes FRR +reporting `::` after losing an unnumbered peer's old link-local address. The +collector otherwise retains that peer as connected, including all of its routes. +GC cannot reclaim state that still appears live. + +Netom repairs the observation by closing **that speaker's BMP transport** and +rebuilding from a fresh snapshot on reconnection. It uses normal disconnect +processing to withdraw every peer, policy view and ADD-PATH child owned by the +connection. Their disconnected register entries and records are reclaimed by +normal RIB GC if they are not rebound. Other speakers' state is not withdrawn. + +This follows [RFC 7854 §3.2](https://www.rfc-editor.org/rfc/rfc7854.html#section-3.2): +a collector restarts monitoring by dropping the TCP connection. Under +[§3.3](https://www.rfc-editor.org/rfc/rfc7854.html#section-3.3), reconnection starts +with Initiation, Peer Up for each established monitored peer, and its current +Adj-RIB-In. [§5](https://www.rfc-editor.org/rfc/rfc7854.html#section-5) describes +that snapshot and per-peer End-of-RIB markers. BMP does not provide a collector +request to refresh one peer, or a global marker proving all peers have been +listed. + +Refresh triggers and default bounds: + +| Condition | Behavior | +| --- | --- | +| Accepted Peer Down cannot match a logical peer (ignoring policy flags) | Schedule reconciliation, no earlier than 60 seconds after connection start. Repeated anomalies do not postpone it. | +| A new link-local address has the same ASN, BGP identifier, peer type and distinguisher as an existing link-local peer | After the startup window, reconcile the whole speaker; do not guess which session is obsolete. | +| Peer Ups arrive within the first 300 seconds | Accept parallel identities as part of startup/replay. Possible-replacement detection is suppressed; unmatched Peer Down and capacity protection still apply. | +| No detectable anomaly, including a completely silent connection | Renew every 6 hours, plus independently selected 0–10% jitter. This bounds the age of an unreconciled observation. | +| The connection reaches 65,536 FSM peer states, including policy views | Close immediately, even within the minimum session lifetime. This is a per-connection peer-state limit, **not a total process-memory or route-count limit**. | + +The first anomaly after the minimum lifetime can close the connection immediately. +Anomalies early in a new connection are coalesced until that minimum. Replay +collisions within the startup window do not cause a new reset, so legitimate +parallel sessions in an ordinary fresh snapshot survive. A timer also interrupts +an idle transport or a partial BMP frame; a cancelled partial read is discarded +with its transport and is never resumed. + +The defaults can be tuned without disabling reconciliation: + +```toml +[units.bmp-in] +type = "bmp-tcp-in" +listen = "0.0.0.0:11019" +reconciliation = { interval_secs = 21600, min_session_secs = 60, replay_grace_secs = 300, max_peer_states = 65536 } +``` + +Validation requires `0 < min_session_secs <= replay_grace_secs < interval_secs +<= 604800` and `max_peer_states > 0`. The maximum interval is seven days before +jitter. Changes take effect on the next BMP connection. Unmatched messages +rejected by a Roto filter do not trigger anomaly reconciliation; periodic and +capacity guards remain active. + +### Operational caveat + +This recovery has a cost: it withdraws the speaker's routes from Netom and its +downstream consumers until replay restores them. It does **not** reset the +router's BGP sessions, but it causes a monitoring gap and a full-table replay, +which consumes exporter CPU, bandwidth, and collector work. Consumers may see +withdrawal/reannouncement events. This is not an atomic or gap-free snapshot +replacement, and downstream backpressure can delay withdrawal/GC processing. + +For passive inputs, the exporter must reconnect. For active inputs, Netom's +normal retry/backoff reconnects. Replay depends on the exporter supplying a +current snapshot; an incremental-only exporter cannot restore missing routes, +and an exporter that replays obsolete peers cannot be corrected by reconciliation. +No timeout can distinguish an idle live peer from a dead peer whose Peer Down +was never sent, so Netom never deletes a peer merely for having no updates. +A missed disappearance during startup may remain visible until periodic renewal. + +The startup window is a conservative time allowance, not proof that replay has +finished. Size it for exporters with slow or very large initial dumps; parallel +Peer Ups arriving after it can cause an additional refresh. A legitimate feed +above the peer-state limit will repeatedly hit that limit until capacity is +raised or the feed is split. The hard limit overrides cooldown; the active +connector/exporter's reconnect backoff must still be respected. Neither the +peer limit nor periodic renewal promises a fixed byte limit for all collector +memory. Physical reclamation also waits for the normal RIB GC sweeps. + +Each refresh logs its reason and speaker address. `scripts/bmp-peer-churn.py` +exercises current-state replay, disappearance, parallel peers, partial frames, +and capacity protection; see [TESTING.md](../TESTING.md). + ## History and initial data TCP and TLS use the same BMP parser, peer identities, ADD-PATH handling and From f7c1830e01b6a84239942a54a7f237e2e2515817 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Fri, 2 Oct 2026 20:35:33 +0300 Subject: [PATCH 4/4] Record successful BMP reconciliation design verification Assisted-by: OpenAI Codex --- TESTING.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/TESTING.md b/TESTING.md index e64bc8a..dc7e680 100644 --- a/TESTING.md +++ b/TESTING.md @@ -132,3 +132,15 @@ tests require permission to bind loopback sockets. See [automatic reconciliation](docs/bmp-tcp-in.md#automatic-reconciliation-and-stale-peer-retention) for the RFC 7854 references, defaults, and monitoring-gap/replay caveat. + +Validation on 2026-10-02 (implementation `0ec9bd6`, regressions `a816a12`): + +- `cargo test --lib --offline`: 354 passed, 31 ignored, no failures. +- `cargo build --bin netom --offline`: passed (existing unused `set_rib` warning). +- `python3 scripts/bmp-peer-churn.py --binary target/debug/netom --cycles 20`: + all seven scenarios passed. Each churn scenario ended with one peer and one + record after GC; the parallel replay retained all three peers/records; silent + and partial-frame recovery reclaimed all obsolete peers/records; capacity + recovery restored the one current peer. Every final disconnect cleaned up. +- Changed Rust files passed rustfmt checks; the Python driver passed syntax + compilation; `git diff --check` passed.