Pass learned routes on to peers, with marks and control communities

What was to be scripted is configuration, done natively:

- import { add_communities, add_large_communities, remove_communities,
  local_pref }: every route taken from a peer is marked (where it was
  learned, say), with nobody else's word for our marks taken, and given
  its preference, which selection goes by.
- export { add_communities, add_large_communities, remove_communities,
  as_path_prepend, learned_routes }, for a peer or for all that have none
  of their own.
- control_communities: the communities by which routes say which peers
  they are passed on to, as exchanges' route servers have them (0:P, L:P,
  0:L, and the large forms L:0:P, L:1:P, L:0:0), taken off as the route
  goes on.

And what the last of these is for: with export.learned_routes a peer is
told the best route of every prefix the others announced. The rules are
BGP's and cannot be turned off: not back where it came from, not to a
peer whose AS is on the path, not between two internal peers, not past
NO_EXPORT or NO_ADVERTISE; our AS on the path and ourselves as next hop
to another network, LOCAL_PREF to our own; to a route server's client
the path, MED and next hop as they came (RFC 7947). Which routes go where
follows from the roles of RFC 9234: a route that came down or across goes
on down only, so that with roles on every peer this makes us transit for
our customers and nothing else. A peer without a role is told everything,
and the daemon says so at startup.

Each session follows the table's best routes with a reader and keeps the
prefixes it has announced, so that it can withdraw them. Routes that go
out alike go in one UPDATE: changes are gathered for 50 ms first, and a
table that is given out anew is gathered whole (at most four sessions at
a time; it is tens of megabytes). For that to work the table's journal
had to give out together what was written together: every change now
carries a time on a table-wide clock, a reader's pass reads every shard
up to one time, no further than the changes that are complete in every
shard, and a resync takes its places in the journals as of one time.

One peer sending 1.23 M routes over a session, another told of them:
in our table after 0.47 s, all heard after 2.1 s, four routes an UPDATE.
The lab has a phase, transit: FRR as our customer and BIRD as our
provider, routes each way with the right paths and marks, one kept back
by a community, and pings between their networks through us.

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

crates/integration-tests/src/test_server.rs

@@ -67,6 +67,8 @@ impl TestServer {
             http_endpoint: None,
             rpki: None,
             kernel: None,
+            export: Default::default(),
+            control: None,
         };
         customize(&mut config);
 

crates/integration-tests/tests/add_path.rs

@@ -50,6 +50,7 @@ async fn server_offering(
             import: ImportPolicy::permissive(),
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
         c.originate = originate;
         // Keep the dialer out of the way.

crates/integration-tests/tests/admission.rs

@@ -28,6 +28,7 @@ fn peer_at(endpoint: SocketAddr) -> PeerConfig {
         import: ImportPolicy::permissive(),
         role: None,
         require_role: false,
+        export: Default::default(),
     }
 }
 

crates/integration-tests/tests/aspa.rs

@@ -67,6 +67,7 @@ async fn server_with(
             import,
             role,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await?;

crates/integration-tests/tests/collision.rs

@@ -46,6 +46,7 @@ async fn server_dialing(listener: &TcpListener) -> Result<TestServer> {
             import: ImportPolicy::permissive(),
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await

crates/integration-tests/tests/debugging.rs

@@ -180,6 +180,7 @@ async fn test_refused_routes_can_be_seen_and_counted() -> Result<()> {
             },
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await?;

crates/integration-tests/tests/dialer.rs

@@ -37,6 +37,7 @@ async fn server_dialing(listener: &TcpListener, remote_asn: u32) -> Result<TestS
             import: ImportPolicy::permissive(),
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await
@@ -194,6 +195,7 @@ async fn test_inbound_from_configured_address_uses_config() -> Result<()> {
             import: ImportPolicy::permissive(),
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await?;
@@ -245,6 +247,7 @@ async fn test_local_address_is_used_as_the_source() -> Result<()> {
             import: ImportPolicy::permissive(),
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await?;
@@ -265,6 +268,7 @@ async fn test_local_address_is_used_as_the_source() -> Result<()> {
             import: ImportPolicy::permissive(),
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await?;

crates/integration-tests/tests/import_policy.rs

@@ -51,6 +51,7 @@ async fn server_with(import: ImportPolicy) -> Result<(TestServer, TcpListener)>
             import,
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await?;
@@ -281,3 +282,63 @@ async fn test_a_peer_over_its_limit_is_cut_off_and_kept_down() -> Result<()> {
         .await?;
     Ok(())
 }
+
+/// Routes from a peer are marked with where they were learned and given
+/// a preference, as the peer's `import` says; and marks that mean
+/// something to us are not taken on the peer's word.
+#[tokio::test]
+async fn test_a_peers_routes_are_marked_and_ranked_as_configured() -> Result<()> {
+    use bgp_packet::message::{CommunitiesPathAttribute, LocalPrefPathAttribute};
+    use bgp_server::marks::Marks;
+    let (server, _listener) = server_with(ImportPolicy {
+        marks: Marks::parse(
+            &["1299:*".to_string()],
+            &["1299:1001".to_string()],
+            &["1299:9:7".to_string()],
+        )
+        .unwrap(),
+        local_pref: Some(250),
+        ..default_policy()
+    })
+    .await?;
+    let mut peer = established(&server).await?;
+    peer.send(&Message::Update(UpdateMessage {
+        withdrawn_routes: vec![],
+        path_attributes: vec![
+            PathAttribute::Origin(OriginPathAttribute::IGP),
+            AsPathAttribute::from_asns(vec![PEER_ASN]),
+            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
+            PathAttribute::Communities(CommunitiesPathAttribute(vec![(3333, 5), (1299, 2002)])),
+        ],
+        announced_routes: vec![prefix("193.0.0.0/21").into(), prefix("193.0.8.0/22").into()],
+    }))
+    .await?;
+    table_after(&server, 2, 0).await?;
+    let routes = server.received_routes();
+    for route in &routes {
+        assert!(
+            route
+                .attributes
+                .contains(&PathAttribute::Communities(CommunitiesPathAttribute(vec![
+                    (3333, 5),
+                    (1299, 1001)
+                ]))),
+            "{route:?}"
+        );
+        assert!(
+            route
+                .attributes
+                .contains(&PathAttribute::LocalPref(LocalPrefPathAttribute(250))),
+            "{route:?}"
+        );
+        assert!(
+            route.attributes.iter().any(|attribute| matches!(
+                attribute,
+                PathAttribute::LargeCommunities(large) if large.communities.len() == 1
+            )),
+            "{route:?}"
+        );
+    }
+    assert_eq!(server.rib.attrs().len(), 1, "they share their attributes");
+    Ok(())
+}

crates/integration-tests/tests/negotiation.rs

@@ -193,6 +193,7 @@ async fn test_no_family_in_common_refuses_the_session() -> Result<()> {
             import: ImportPolicy::permissive(),
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
         // Keep the dialer out of the way.
         c.connect_retry = Duration::from_secs(3600);

crates/integration-tests/tests/roles.rs

@@ -58,6 +58,7 @@ async fn server_with(role: Option<PeerRole>, require_role: bool) -> Result<Setup
             import: default_policy(),
             role,
             require_role,
+            export: Default::default(),
         }];
     })
     .await?;

crates/integration-tests/tests/rpki.rs

@@ -77,6 +77,7 @@ async fn server_with(
             import,
             role: None,
             require_role: false,
+            export: Default::default(),
         }];
     })
     .await?;

crates/integration-tests/tests/transit.rs

@@ -0,0 +1,389 @@
+//! Passing routes on: what one peer announced reaches the others, as far
+//! as the rules for that let it.
+
+use std::collections::BTreeSet;
+use std::net::Ipv4Addr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use bgp_integration_tests::test_peer::TestPeer;
+use bgp_integration_tests::test_server::TestServer;
+use bgp_packet::message::{
+    AsPathAttribute, CommunitiesPathAttribute, Message, NextHopPathAttribute, OriginPathAttribute,
+    PathAttribute, RouteRefreshMessage, UpdateMessage,
+};
+use bgp_primitives::constants::AfiSafi;
+use bgp_primitives::ip_prefix::IpPrefix;
+use bgp_rib::{LocalRoute, LocalRouteAttrs};
+use bgp_server::marks::{Control, Marks};
+use bgp_server::policy::ExportPolicy;
+use eyre::Result;
+
+const LOCAL_ASN: u32 = 64601;
+const TIMEOUT: Duration = Duration::from_secs(5);
+
+fn prefix(s: &str) -> IpPrefix {
+    IpPrefix::try_from(s).unwrap()
+}
+
+/// A server that passes on to every peer what it learned from the others.
+async fn transit(customize: impl FnOnce(&mut bgp_server::config::Config)) -> Result<TestServer> {
+    TestServer::start_with(LOCAL_ASN, |config| {
+        config.export = Arc::new(ExportPolicy {
+            learned_routes: true,
+            ..ExportPolicy::default()
+        });
+        customize(config);
+    })
+    .await
+}
+
+/// The peer with AS `64700 + n`, its session up.
+async fn peer(server: &TestServer, n: u8) -> Result<TestPeer> {
+    let before = server.registry.list().len();
+    let mut peer = TestPeer::connect(server.addr).await?;
+    peer.handshake(64700 + u32::from(n), Ipv4Addr::new(10, 0, 0, n), 90)
+        .await?;
+    tokio::time::timeout(TIMEOUT, async {
+        while server.registry.list().len() != before + 1 {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await?;
+    Ok(peer)
+}
+
+async fn announce(
+    peer: &mut TestPeer,
+    prefixes: &[&str],
+    path: &[u32],
+    communities: &[(u16, u16)],
+) -> Result<()> {
+    let mut path_attributes = vec![
+        PathAttribute::Origin(OriginPathAttribute::IGP),
+        AsPathAttribute::from_asns(path.to_vec()),
+        PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
+    ];
+    if !communities.is_empty() {
+        path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute(
+            communities.to_vec(),
+        )));
+    }
+    peer.send(&Message::Update(UpdateMessage {
+        withdrawn_routes: vec![],
+        path_attributes,
+        announced_routes: prefixes.iter().map(|p| prefix(p).into()).collect(),
+    }))
+    .await
+}
+
+async fn withdraw(peer: &mut TestPeer, prefixes: &[&str]) -> Result<()> {
+    peer.send(&Message::Update(UpdateMessage {
+        withdrawn_routes: prefixes.iter().map(|p| prefix(p).into()).collect(),
+        path_attributes: vec![],
+        announced_routes: vec![],
+    }))
+    .await
+}
+
+/// What a peer has been told so far, as the prefixes it would now have
+/// and the path each came with.
+#[derive(Default, Debug)]
+struct Told {
+    routes: std::collections::BTreeMap<String, Vec<u32>>,
+    updates: usize,
+}
+
+impl Told {
+    fn take(&mut self, update: &UpdateMessage) {
+        self.updates += 1;
+        for route in &update.withdrawn_routes {
+            self.routes.remove(&route.prefix.to_string());
+        }
+        let path: Vec<u32> = update
+            .path_attributes
+            .iter()
+            .find_map(|attribute| match attribute {
+                PathAttribute::AsPath(path) => Some(
+                    path.segments
+                        .iter()
+                        .flat_map(|segment| segment.path.clone())
+                        .collect(),
+                ),
+                _ => None,
+            })
+            .unwrap_or_default();
+        for route in &update.announced_routes {
+            self.routes.insert(route.prefix.to_string(), path.clone());
+        }
+    }
+
+    fn prefixes(&self) -> BTreeSet<&str> {
+        self.routes.keys().map(String::as_str).collect()
+    }
+}
+
+/// Reads what the peer is sent until it has exactly `expected`.
+async fn told_becomes(peer: &mut TestPeer, told: &mut Told, expected: &[&str]) -> Result<()> {
+    let expected: BTreeSet<&str> = expected.iter().copied().collect();
+    let reading = async {
+        while told.prefixes() != expected {
+            let update = peer.recv_update().await?;
+            told.take(&update);
+        }
+        Ok::<(), eyre::Report>(())
+    };
+    match tokio::time::timeout(TIMEOUT, reading).await {
+        Ok(read) => read,
+        Err(_) => Err(eyre::eyre!(
+            "the peer has {:?}; expected {expected:?}",
+            told.prefixes()
+        )),
+    }
+}
+
+/// Nothing more is sent to the peer for a while.
+async fn told_nothing(peer: &mut TestPeer) {
+    let more = tokio::time::timeout(Duration::from_millis(300), peer.recv_update()).await;
+    assert!(more.is_err(), "the peer was sent {more:?}");
+}
+
+/// What one peer announces the others are told, with our AS on the path
+/// and ourselves as next hop; and what it takes back, they are told too.
+#[tokio::test]
+async fn test_routes_from_one_peer_reach_the_others() -> Result<()> {
+    let server = transit(|_| {}).await?;
+    let mut one = peer(&server, 1).await?;
+    let mut two = peer(&server, 2).await?;
+
+    announce(
+        &mut one,
+        &["10.1.0.0/16", "10.2.0.0/16", "10.3.0.0/16"],
+        &[64701, 65000],
+        &[],
+    )
+    .await?;
+    let mut told = Told::default();
+    told_becomes(
+        &mut two,
+        &mut told,
+        &["10.1.0.0/16", "10.2.0.0/16", "10.3.0.0/16"],
+    )
+    .await?;
+    assert_eq!(told.routes["10.1.0.0/16"], [LOCAL_ASN, 64701, 65000]);
+    assert_eq!(
+        told.updates, 1,
+        "routes that go out alike go in one message"
+    );
+    // Not back to where they came from.
+    told_nothing(&mut one).await;
+
+    // The next hop is ourselves: the address of our end of the session.
+    announce(&mut one, &["10.4.0.0/16"], &[64701], &[]).await?;
+    let update = tokio::time::timeout(TIMEOUT, two.recv_update()).await??;
+    assert!(
+        update
+            .path_attributes
+            .contains(&PathAttribute::NextHop(NextHopPathAttribute(
+                Ipv4Addr::LOCALHOST
+            ))),
+        "{update:?}"
+    );
+    told.take(&update);
+
+    // One is taken back; then the peer goes, and its routes with it.
+    withdraw(&mut one, &["10.2.0.0/16"]).await?;
+    told_becomes(
+        &mut two,
+        &mut told,
+        &["10.1.0.0/16", "10.3.0.0/16", "10.4.0.0/16"],
+    )
+    .await?;
+    drop(one);
+    told_becomes(&mut two, &mut told, &[]).await?;
+    Ok(())
+}
+
+/// A peer is told of the best route, and of the next best when that one
+/// goes; and a peer that comes later is told what there is.
+#[tokio::test]
+async fn test_a_peer_is_kept_at_the_best_route() -> Result<()> {
+    let server = transit(|_| {}).await?;
+    let mut near = peer(&server, 1).await?;
+    let mut far = peer(&server, 2).await?;
+    let mut watcher = peer(&server, 3).await?;
+    let mut told = Told::default();
+
+    announce(&mut far, &["10.1.0.0/16"], &[64702, 65000, 65001], &[]).await?;
+    told_becomes(&mut watcher, &mut told, &["10.1.0.0/16"]).await?;
+    assert_eq!(told.routes["10.1.0.0/16"], [LOCAL_ASN, 64702, 65000, 65001]);
+
+    // A better route for it comes: that is the one told now.
+    announce(&mut near, &["10.1.0.0/16"], &[64701, 65001], &[]).await?;
+    let update = tokio::time::timeout(TIMEOUT, watcher.recv_update()).await??;
+    told.take(&update);
+    assert_eq!(told.routes["10.1.0.0/16"], [LOCAL_ASN, 64701, 65001]);
+    assert!(
+        update.withdrawn_routes.is_empty(),
+        "replaced, not withdrawn"
+    );
+
+    // And goes again: back to the other, with no gap.
+    withdraw(&mut near, &["10.1.0.0/16"]).await?;
+    let update = tokio::time::timeout(TIMEOUT, watcher.recv_update()).await??;
+    told.take(&update);
+    assert_eq!(told.routes["10.1.0.0/16"], [LOCAL_ASN, 64702, 65000, 65001]);
+
+    // The other peer was told of the better route while it was the best,
+    // and then that it is gone: the one left is its own, which it is not
+    // told of.
+    let mut told_far = Told::default();
+    told_becomes(&mut far, &mut told_far, &["10.1.0.0/16"]).await?;
+    let gone = tokio::time::timeout(TIMEOUT, far.recv_update()).await??;
+    assert_eq!(gone.withdrawn_routes, vec![prefix("10.1.0.0/16")]);
+    told_nothing(&mut far).await;
+
+    // A peer that comes now is told what there is; and again if it asks.
+    let mut late = peer(&server, 4).await?;
+    let mut late_told = Told::default();
+    told_becomes(&mut late, &mut late_told, &["10.1.0.0/16"]).await?;
+    late.send(&Message::RouteRefresh(RouteRefreshMessage::request(
+        AfiSafi::IPV4_UNICAST,
+    )))
+    .await?;
+    let again = tokio::time::timeout(TIMEOUT, late.recv_update()).await??;
+    assert_eq!(again.announced_routes, vec![prefix("10.1.0.0/16")]);
+    Ok(())
+}
+
+/// A route is not sent to a peer whose AS is on its path already, nor
+/// anywhere if it says NO_EXPORT; and what we originate goes to everyone
+/// beside what was learned.
+#[tokio::test]
+async fn test_routes_stop_where_they_must() -> Result<()> {
+    let server = transit(|config| {
+        config.originate = vec![LocalRoute::new(
+            prefix("10.99.0.0/16"),
+            LocalRouteAttrs::default(),
+        )];
+    })
+    .await?;
+    let mut one = peer(&server, 1).await?;
+    let mut two = peer(&server, 2).await?;
+    let mut told = Told::default();
+    told_becomes(&mut two, &mut told, &["10.99.0.0/16"]).await?;
+    assert_eq!(told.routes["10.99.0.0/16"], [LOCAL_ASN]);
+
+    // Through the second peer's AS already; and not for other networks.
+    announce(&mut one, &["10.1.0.0/16"], &[64701, 64702, 65000], &[]).await?;
+    announce(&mut one, &["10.2.0.0/16"], &[64701], &[(0xFFFF, 0xFF01)]).await?;
+    announce(&mut one, &["10.3.0.0/16"], &[64701], &[]).await?;
+    told_becomes(&mut two, &mut told, &["10.99.0.0/16", "10.3.0.0/16"]).await?;
+    told_nothing(&mut two).await;
+    assert_eq!(server.rib.len(), 3, "all three were taken");
+    Ok(())
+}
+
+/// Communities on a route say which peers it is passed on to, and are
+/// gone when it is; and every route to a peer gets the marks of `export`.
+#[tokio::test]
+async fn test_communities_say_who_is_told() -> Result<()> {
+    let server = transit(|config| {
+        config.control = Some(Control {
+            local_asn: LOCAL_ASN,
+        });
+        config.export = Arc::new(ExportPolicy {
+            learned_routes: true,
+            marks: Marks::parse(&[], &["64601:7".to_string()], &[]).unwrap(),
+            prepend: 1,
+        });
+        config.originate = vec![LocalRoute::new(
+            prefix("10.99.0.0/16"),
+            LocalRouteAttrs {
+                // Our own route, and not for the second peer.
+                communities: vec![(0, 64702)],
+                ..LocalRouteAttrs::default()
+            },
+        )];
+    })
+    .await?;
+    let mut one = peer(&server, 1).await?;
+    let mut two = peer(&server, 2).await?;
+    let mut three = peer(&server, 3).await?;
+    let (mut told_two, mut told_three) = (Told::default(), Told::default());
+
+    // To everyone but the second; to the second only; to nobody; and
+    // with nothing said.
+    announce(
+        &mut one,
+        &["10.1.0.0/16"],
+        &[64701],
+        &[(0, 64702), (3356, 3)],
+    )
+    .await?;
+    announce(&mut one, &["10.2.0.0/16"], &[64701], &[(64601, 64702)]).await?;
+    announce(&mut one, &["10.3.0.0/16"], &[64701], &[(0, 64601)]).await?;
+    announce(&mut one, &["10.4.0.0/16"], &[64701], &[]).await?;
+
+    told_becomes(&mut two, &mut told_two, &["10.2.0.0/16", "10.4.0.0/16"]).await?;
+    told_becomes(
+        &mut three,
+        &mut told_three,
+        &["10.99.0.0/16", "10.1.0.0/16", "10.4.0.0/16"],
+    )
+    .await?;
+    told_nothing(&mut two).await;
+    told_nothing(&mut three).await;
+    // Our AS once for passing it on, and once more by `export`.
+    assert_eq!(
+        told_three.routes["10.1.0.0/16"],
+        [LOCAL_ASN, LOCAL_ASN, 64701]
+    );
+    assert_eq!(told_three.routes["10.99.0.0/16"], [LOCAL_ASN, LOCAL_ASN]);
+
+    // What was for us to act on is gone, and the router's mark is on.
+    announce(
+        &mut one,
+        &["10.5.0.0/16"],
+        &[64701],
+        &[(0, 64702), (3356, 3)],
+    )
+    .await?;
+    let update = tokio::time::timeout(TIMEOUT, three.recv_update()).await??;
+    assert!(
+        update
+            .path_attributes
+            .contains(&PathAttribute::Communities(CommunitiesPathAttribute(vec![
+                (3356, 3),
+                (64601, 7)
+            ]))),
+        "{update:?}"
+    );
+    Ok(())
+}
+
+/// Without `learned_routes` a peer is told what we originate and nothing
+/// else, as before there was such a thing.
+#[tokio::test]
+async fn test_nothing_learned_is_passed_on_unless_asked_for() -> Result<()> {
+    let server = TestServer::start_with(LOCAL_ASN, |config| {
+        config.originate = vec![LocalRoute::new(
+            prefix("10.99.0.0/16"),
+            LocalRouteAttrs::default(),
+        )];
+    })
+    .await?;
+    let mut one = peer(&server, 1).await?;
+    let mut two = peer(&server, 2).await?;
+    let mut told = Told::default();
+    told_becomes(&mut two, &mut told, &["10.99.0.0/16"]).await?;
+    announce(&mut one, &["10.1.0.0/16"], &[64701], &[]).await?;
+    tokio::time::timeout(TIMEOUT, async {
+        while server.rib.len() != 1 {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await?;
+    told_nothing(&mut two).await;
+    Ok(())
+}