From 0ec9bd6a412e9641f6cf0160e2dda9f4d642776c Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Fri, 2 Oct 2026 20:00:10 +0300 Subject: [PATCH 1/7] 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/7] 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/7] 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/7] 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. From 66071c6b3db3495faddb5fd4418a305b60d0253e Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 20 Oct 2026 20:50:47 +0300 Subject: [PATCH 5/7] Make periodic BMP reconciliation opt-in --- Changelog.md | 11 ++++ TESTING.md | 23 ++++++- docs/bmp-tcp-in.md | 40 ++++++++---- scripts/bmp-peer-churn.py | 15 +++-- src/units/bmp_tcp_in/reconciliation.rs | 88 ++++++++++++++++++++------ src/units/bmp_tcp_in/router_handler.rs | 58 +++++++++++++---- 6 files changed, 183 insertions(+), 52 deletions(-) diff --git a/Changelog.md b/Changelog.md index 1e53192..b885197 100644 --- a/Changelog.md +++ b/Changelog.md @@ -27,6 +27,17 @@ Released yyyy-mm-dd. ### New +* `bmp-tcp-in` recovers inconsistent peer state after an unmatched Peer Down, + a possible unnumbered-peer replacement, or the peer-state capacity limit by + renewing the affected speaker's BMP connection and replaying its snapshot. + This addresses stale peer retention described in issue #11 without guessing + which same-router-ID peer to delete. The `reconciliation` configuration + controls cooldown, startup grace, capacity, and optional periodic renewal. + Periodic renewal is disabled by default (`interval_secs = 0`). + Recovery temporarily withdraws the speaker's monitored routes and incurs a + full-table replay; it does not reset the router's BGP sessions. Exporters + must reconnect and provide current state. See [BMP input](docs/bmp-tcp-in.md). + * `netom-cli show ip bgp detail` (and `show ipv6 bgp detail`) prints every attribute of each path: peer and AS, how it was learned (ingress, BMP, pre/post-policy, ADD-PATH path id), status and RPKI, origin, AS path, diff --git a/TESTING.md b/TESTING.md index dc7e680..9b36354 100644 --- a/TESTING.md +++ b/TESTING.md @@ -97,20 +97,23 @@ 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`: +Eight scenarios run by default; select one with `--scenario NAME`: +- `healthy`: an idle established peer stays connected with periodic renewal off. - `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 +- `silent`: a peer vanishes without any further BMP message; opt-in 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. +Only `silent` and `partial` enable a periodic timer. Other scenarios leave +`interval_secs` unspecified to exercise the production default of no timer. 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 @@ -125,7 +128,8 @@ Rust regressions are part of the normal suite: cargo test --lib units::bmp_tcp_in ``` -They cover deadline coalescing, startup replay, periodic jitter, capacity, +They cover default transports staying open past eight simulated days, +deadline coalescing, startup replay, opt-in 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. @@ -133,6 +137,19 @@ 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-20 (periodic renewal opt-in): + +- `cargo test --locked --lib units::bmp_tcp_in`: 65 passed, 16 ignored. +- `cargo build --locked --bin netom`: passed with existing deprecation and + unused `set_rib` warnings. +- `python3 scripts/bmp-peer-churn.py --binary target/debug/netom --cycles 20`: + all eight scenarios passed. The healthy and matching-Peer-Down scenarios + required only their initial snapshot; matching cleanup reclaimed all retired + identities without a BMP reconnect. Anomaly and capacity cases recovered with + periodic renewal off. Silent and partial cases recovered with it opted in. +- Changed Rust files passed rustfmt checks; Python syntax and + `git diff --check` passed. + Validation on 2026-10-02 (implementation `0ec9bd6`, regressions `a816a12`): - `cargo test --lib --offline`: 354 passed, 31 ignored, no failures. diff --git a/docs/bmp-tcp-in.md b/docs/bmp-tcp-in.md index 50f2c81..b37eea5 100644 --- a/docs/bmp-tcp-in.md +++ b/docs/bmp-tcp-in.md @@ -61,8 +61,9 @@ 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 +Anomaly and capacity recovery are enabled for every BMP input, for both `listen` +and `connect`. Periodic renewal is disabled by default. 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. @@ -72,6 +73,11 @@ 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. +The exporter should preserve the established session's identity through teardown +and send Peer Down with the same peer address used for Peer Up. Correctly addressed +Peer Down notifications use ordinary per-peer cleanup without restarting BMP. +Reconciliation is a fallback for inconsistent observations, not an exporter fix. + 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 @@ -94,7 +100,7 @@ Refresh triggers and default bounds: | 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. | +| No detectable anomaly, including a completely silent connection | Keep the connection open by default. If `interval_secs` is nonzero, renew after that interval plus independently selected 0–10% jitter. | | 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. @@ -104,20 +110,23 @@ 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: +Default settings (anomaly/capacity recovery without periodic renewal): ```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 } +reconciliation = { interval_secs = 0, 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. +To opt into a periodic snapshot every six hours plus jitter, set +`interval_secs = 21600`. Setting it to `0` disables only periodic renewal. +Validation requires `0 < min_session_secs <= replay_grace_secs < 604800` and +`max_peer_states > 0`. A nonzero interval must satisfy +`replay_grace_secs < interval_secs <= 604800` (seven days before jitter). +Changes take effect on the next BMP connection. Unmatched messages rejected by +a Roto filter do not trigger anomaly reconciliation; capacity protection and +any configured periodic timer remain active. ### Operational caveat @@ -132,9 +141,14 @@ 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. +Without an independently configured liveness expectation, silence cannot +distinguish an idle live peer from a dead peer whose Peer Down was never sent. +Netom never deletes a peer merely for having no updates. BMP statistics reports +can provide per-peer activity evidence, but are optional and have no negotiated +heartbeat interval. This implementation does not infer peer expiry from missing +statistics. A silent disappearance, including one missed during startup, may +remain visible until a later anomaly, disconnect, or explicitly enabled periodic +renewal obtains a new snapshot. 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 diff --git a/scripts/bmp-peer-churn.py b/scripts/bmp-peer-churn.py index 9ac6336..8caa294 100644 --- a/scripts/bmp-peer-churn.py +++ b/scripts/bmp-peer-churn.py @@ -115,6 +115,7 @@ def expect_reset(self, timeout=5): if data: raise RuntimeError("collector unexpectedly sent BMP data") except ConnectionResetError: + # A TCP reset is an expected successful reconciliation outcome. pass except socket.timeout as error: raise RuntimeError("collector failed to reconcile") from error @@ -128,6 +129,7 @@ def expect_open(self, seconds): except socket.timeout: return except ConnectionResetError: + # Report resets through the common unexpected-reconnect error below. pass raise RuntimeError("unexpected reconnect: parallel replay/control must remain connected") @@ -141,8 +143,9 @@ 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 + # Only silent/partial recovery opts into periodic renewal. Other scenarios + # exercise the production default: anomalies/capacity only. + periodic = "interval_secs = 6, " if scenario in {"silent", "partial"} else "" limit = 3 if scenario == "limit" else 65536 with tempfile.TemporaryDirectory(prefix="netom-peer-churn-") as directory: directory = Path(directory) @@ -151,7 +154,7 @@ def simulate(binary, scenario, cycles, grace): [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} }} +reconciliation = {{ {periodic}min_session_secs = 1, replay_grace_secs = 1, max_peer_states = {limit} }} [units.rib] type = "rib" sources = ["bmp-in"] @@ -200,7 +203,9 @@ def ready(): exporter.connect() wait_until(snapshot, lambda s: s["records"] >= len(exporter.current), 5, "initial routes") - if scenario in {"matching", "mismatched", "missing"}: + if scenario == "healthy": + exporter.expect_open(7) + elif 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 @@ -278,7 +283,7 @@ def main(): 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"] + scenarios = ["healthy", "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"): diff --git a/src/units/bmp_tcp_in/reconciliation.rs b/src/units/bmp_tcp_in/reconciliation.rs index df401b4..ef8cea0 100644 --- a/src/units/bmp_tcp_in/reconciliation.rs +++ b/src/units/bmp_tcp_in/reconciliation.rs @@ -18,7 +18,8 @@ pub struct Config { impl Default for Config { fn default() -> Self { Self { - interval_secs: 6 * 60 * 60, + // Zero disables periodic renewal; anomaly/capacity recovery stays active. + interval_secs: 0, min_session_secs: 60, replay_grace_secs: 5 * 60, max_peer_states: 65_536, @@ -52,11 +53,13 @@ impl TryFrom for Config { 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.replay_grace_secs >= 7 * 24 * 60 * 60 + || (c.interval_secs != 0 + && 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"); + return Err("BMP reconciliation requires 0 < min_session_secs <= replay_grace_secs < 604800, max_peer_states > 0, and interval_secs = 0 (periodic renewal disabled) or replay_grace_secs < interval_secs <= 604800"); } Ok(Self { interval_secs: c.interval_secs, @@ -91,7 +94,7 @@ impl std::fmt::Display for Reason { pub struct Reconciliation { config: Config, started: Instant, - deadline: Instant, + deadline: Option, reason: Reason, } @@ -102,8 +105,9 @@ impl Reconciliation { Self { config, started: now, - deadline: now - + Duration::from_secs(config.interval_secs + jitter), + deadline: (config.interval_secs != 0).then(|| { + now + Duration::from_secs(config.interval_secs + jitter) + }), reason: Reason::Periodic, } } @@ -116,7 +120,7 @@ impl Reconciliation { ) { if peer_states >= self.config.max_peer_states { // Capacity takes precedence over the minimum session lifetime. - self.deadline = now; + self.deadline = Some(now); self.reason = Reason::PeerStateLimit; return; } @@ -134,13 +138,13 @@ impl Reconciliation { let deadline = now.max( self.started + Duration::from_secs(self.config.min_session_secs), ); - if deadline < self.deadline { - self.deadline = deadline; + if self.deadline.is_none_or(|pending| deadline < pending) { + self.deadline = Some(deadline); self.reason = reason; } } - pub fn deadline(&self) -> Instant { + pub fn deadline(&self) -> Option { self.deadline } pub fn reason(&self) -> Reason { @@ -161,6 +165,40 @@ mod tests { } } + #[test] + fn default_has_no_timer_but_anomalies_and_capacity_still_recover() { + let now = Instant::now(); + let mut p = Reconciliation::new(Config::default(), now, u64::MAX); + p.observe(now + Duration::from_secs(30 * 86400), None, 1); + assert_eq!(p.deadline(), None); + + for reason in [Reason::UnmatchedPeerDown, Reason::PossibleReplacement] + { + let mut p = Reconciliation::new(Config::default(), now, 0); + if reason == Reason::PossibleReplacement { + p.observe(now, Some(reason), 2); + assert_eq!(p.deadline(), None, "startup peers must coexist"); + } + let observed = now + Duration::from_secs(300); + p.observe(observed, Some(reason), 2); + assert_eq!(p.deadline(), Some(observed)); + assert_eq!(p.reason(), reason); + } + + let mut p = Reconciliation::new(Config::default(), now, 0); + p.observe(now, Some(Reason::UnmatchedPeerDown), 1); + assert_eq!(p.deadline(), Some(now + Duration::from_secs(60))); + p.observe( + now + Duration::from_secs(1), + Some(Reason::UnmatchedPeerDown), + 1, + ); + assert_eq!(p.deadline(), Some(now + Duration::from_secs(60))); + p.observe(now, None, 65_536); + assert_eq!(p.deadline(), Some(now), "capacity overrides cooldown"); + assert_eq!(p.reason(), Reason::PeerStateLimit); + } + #[test] fn startup_parallel_peers_do_not_cause_a_reconnect_loop() { let now = Instant::now(); @@ -178,7 +216,7 @@ mod tests { Some(Reason::PossibleReplacement), 3, ); - assert_eq!(p.deadline(), start + Duration::from_secs(100)); + assert_eq!(p.deadline(), Some(start + Duration::from_secs(100))); assert_eq!(p.reason(), Reason::Periodic); } } @@ -188,14 +226,14 @@ mod tests { 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)); + assert_eq!(p.deadline(), Some(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.deadline(), Some(now + Duration::from_secs(5))); } assert_eq!(p.reason(), Reason::UnmatchedPeerDown); } @@ -206,7 +244,7 @@ mod tests { 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.deadline(), Some(after_replay)); assert_eq!(p.reason(), Reason::PossibleReplacement); } @@ -215,31 +253,41 @@ mod tests { 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)); + assert!(p.deadline().unwrap() >= now + Duration::from_secs(100)); + assert!(p.deadline().unwrap() <= now + Duration::from_secs(110)); p.observe(now, None, 8); - assert_eq!(p.deadline(), now); + assert_eq!(p.deadline(), Some(now)); assert_eq!(p.reason(), Reason::PeerStateLimit); } } #[test] - fn configuration_cannot_disable_reconciliation() { + fn configuration_allows_optional_periodic_renewal() { #[derive(Deserialize)] struct Wrapper { reconciliation: Config, } let defaults: Wrapper = toml::from_str("[reconciliation]").unwrap(); - assert_eq!(defaults.reconciliation.interval_secs, 21600); + assert_eq!(defaults.reconciliation.interval_secs, 0); for fields in [ "interval_secs = 0", + "interval_secs = 21600", + "interval_secs = 604800", + ] { + assert!(toml::from_str::(&format!( + "[reconciliation]\n{fields}" + )) + .is_ok()); + } + for fields in [ "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", + "replay_grace_secs = 604800", + "interval_secs = 300", ] { assert!( toml::from_str::(&format!( diff --git a/src/units/bmp_tcp_in/router_handler.rs b/src/units/bmp_tcp_in/router_handler.rs index 7bf22cd..292998a 100644 --- a/src/units/bmp_tcp_in/router_handler.rs +++ b/src/units/bmp_tcp_in/router_handler.rs @@ -233,20 +233,22 @@ impl RouterHandler { 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(_) => { + let next = if let Some(deadline) = deadline { + 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: its transport is dropped, + // never reused. Anomaly deadlines also interrupt idle reads. + 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; + } + } + } else { + stream.next().await }; match next { Err(err) => { @@ -808,6 +810,40 @@ 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_default_keeps_idle_and_partial_transports_open() { + use tokio::io::AsyncWriteExt; + for partial in [false, true] { + let (runner, _agent, _parent) = RouterHandler::mock(); + 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(); + let read = runner.read_from_router( + reader, + "127.0.0.1:12345".parse().unwrap(), + id, + register.clone(), + Arc::default(), + Arc::default(), + ); + tokio::pin!(read); + tokio::select! { + _ = &mut read => panic!("default must not renew a quiet transport"), + _ = tokio::time::sleep(Duration::from_secs(8 * 86400)) => {} + } + // Ordinary disconnect still ends monitoring with no timer configured. + drop(writer); + timeout(Duration::from_secs(1), read).await.unwrap(); + assert_eq!( + register.get(id).unwrap().state, + Some(IngressState::Disconnected) + ); + } + } + #[tokio::test(start_paused = true)] async fn reconciliation_closes_idle_and_partial_transports() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; From 3d924f95d341dd163d4717e039cfbf983fdbd0e8 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 20 Oct 2026 21:02:29 +0300 Subject: [PATCH 6/7] Fix testing guide link in Sphinx manual --- docs/bmp-tcp-in.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/bmp-tcp-in.md b/docs/bmp-tcp-in.md index b37eea5..86acedc 100644 --- a/docs/bmp-tcp-in.md +++ b/docs/bmp-tcp-in.md @@ -161,7 +161,8 @@ 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). +and capacity protection; see +[TESTING.md](https://github.com/FastNetMon/netom/blob/main/TESTING.md#bmp-reconciliation-simulations). ## History and initial data From ee42ac1ad190b4f23a9fb4d6225fcc3e549dc210 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 20 Oct 2026 21:07:26 +0300 Subject: [PATCH 7/7] Remove obsolete experimental option reference from BMP manual --- docs/bmp-tcp-in.md | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/docs/bmp-tcp-in.md b/docs/bmp-tcp-in.md index 86acedc..adc2306 100644 --- a/docs/bmp-tcp-in.md +++ b/docs/bmp-tcp-in.md @@ -62,10 +62,9 @@ connection. Listening address changes preserve existing accepted connections. ## Automatic reconciliation and stale-peer retention Anomaly and capacity recovery are enabled for every BMP input, for both `listen` -and `connect`. Periodic renewal is disabled by default. 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. +and `connect`. Periodic renewal is disabled by default. Netom does not delete +an existing peer solely because a new peer shares its BGP identifier; legitimate +parallel sessions can share that identifier. 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