[rdb] rework RPF cache rebuilds around a bounded worker

This replaces the interval-driven poptrie rebuild with a single long-lived
worker that rebuilds on loc-RIB change. Mutation batches already end
with a change notification, so `notify()` now triggers the rebuild as
well. Paths that bypass notification (bestpath re-runs, peer staleness marking,
direct path removal) still trigger explicitly.

This long-lived worker keeps memory bounded with a one-slot pending queue per
address family, where requests arriving within a fixed coalescing window
merge, and distinct changed prefixes widen to a full sweep.

Completed snapshots install unconditionally. A trigger that lands during a build
leaves its request in the pending slot, so a possibly stale install is followed
by a rebuild from a newer snapshot 1 coalescing window later. Installs are
serialized in the single worker, so each is newer than the last. After updates
quiesce, the final rebuild converges to the current RIB.

Optimizations:

- Snapshots derive the compact `RpfNexthops` payload in the same pass under the
RIB lock, so full path sets are never cloned, and `get_rpf_neighbor` selects a
neighbor without cloning the path set.
- The linear-scan fallback now also matches /0 default routes.

With these changes, the rebuild-interval setting and its plumbing are removed,
including the admin API endpoints and a regeneration of the unblessed v12
OpenAPI spec.

Author: zeeshanlakhani

mg-api-types/versions/src/multicast_support/mod.rs

@@ -7,9 +7,9 @@
 //! Version `MULTICAST_SUPPORT` of the Maghemite Admin API.
 //!
 //! Adds MRIB (Multicast Routing Information Base) endpoints for static
-//! multicast route management, RPF rebuild rate control, and query of
-//! imported and selected multicast routes. Introduces the supporting wire
-//! types (validated unicast and multicast addresses, the underlay group
-//! within `ff04::/64`, multicast route keys, and route entries).
+//! multicast route management and query of imported and selected multicast
+//! routes. Introduces the supporting wire types (validated unicast and
+//! multicast addresses, the underlay group within `ff04::/64`, multicast
+//! route keys, and route entries).
 
 pub mod mrib;

mg-api-types/versions/src/multicast_support/mrib.rs

@@ -156,23 +156,6 @@ pub struct MribDeleteStaticRequest {
     pub keys: Vec<MulticastRouteKey>,
 }
 
-/// Response containing the current RPF rebuild interval.
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
-pub struct MribRpfRebuildIntervalResponse {
-    /// Minimum interval between RPF cache rebuilds in milliseconds.
-    /// A value of 0 means rate-limiting is disabled.
-    pub interval_ms: u64,
-}
-
-/// Request body for setting the RPF rebuild interval.
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
-pub struct MribRpfRebuildIntervalRequest {
-    /// Minimum interval between RPF cache rebuilds in milliseconds.
-    /// A value of 0 disables rate-limiting.
-    /// Every unicast RIB change triggers an immediate poptrie rebuild.
-    pub interval_ms: u64,
-}
-
 /// Filter for multicast route origin.
 #[derive(
     Debug, Clone, Copy, Deserialize, Serialize, JsonSchema, PartialEq, Eq,
@@ -1015,7 +998,12 @@ impl MulticastRouteKey {
 pub struct MulticastRoute {
     /// The multicast route key (S,G) or (*,G).
     pub key: MulticastRouteKey,
-    /// Expected RPF neighbor for the source (for RPF checks).
+    /// Upstream neighbor selected for RPF checks.
+    ///
+    /// This records one representative neighbor rather than the full ECMP
+    /// set. When the unicast route has multiple equal-cost paths, any active
+    /// member is valid. `None` means RPF does not apply or no active unicast
+    /// path is available.
     pub rpf_neighbor: Option<IpAddr>,
     /// Underlay multicast group address (ff04::/64).
     ///

mg-api/src/lib.rs

@@ -1568,20 +1568,4 @@ pub trait MgAdminApi {
     async fn static_list_mcast_routes(
         rqctx: RequestContext<Self::Context>,
     ) -> Result<HttpResponseOk<Vec<latest::mrib::MulticastRoute>>, HttpError>;
-
-    /// Get the RPF rebuild rate-limit interval.
-    #[endpoint { method = GET, path = "/mrib/config/rpf/rebuild-interval", versions = VERSION_MULTICAST_SUPPORT.. }]
-    async fn read_mrib_rpf_rebuild_interval(
-        rqctx: RequestContext<Self::Context>,
-    ) -> Result<
-        HttpResponseOk<latest::mrib::MribRpfRebuildIntervalResponse>,
-        HttpError,
-    >;
-
-    /// Set the RPF rebuild rate-limit interval.
-    #[endpoint { method = POST, path = "/mrib/config/rpf/rebuild-interval", versions = VERSION_MULTICAST_SUPPORT.. }]
-    async fn update_mrib_rpf_rebuild_interval(
-        rqctx: RequestContext<Self::Context>,
-        request: TypedBody<latest::mrib::MribRpfRebuildIntervalRequest>,
-    ) -> Result<HttpResponseUpdatedNoContent, HttpError>;
 }

mgadm/src/mrib.rs

@@ -19,8 +19,7 @@ use clap::{Args, Subcommand};
 use client_common::println_nopipe;
 use mg_admin_client::Client;
 use mg_admin_client::types::{
-    MribRpfRebuildIntervalRequest, MulticastRoute, MulticastRouteKey,
-    RouteOriginFilter,
+    MulticastRoute, MulticastRouteKey, RouteOriginFilter,
 };
 use mg_api_types::mrib::DEFAULT_MULTICAST_VNI;
 use mg_api_types::rdb::rib::AddressFamily;
@@ -41,9 +40,6 @@ fn parse_route_origin(s: &str) -> Result<RouteOriginFilter, String> {
 pub enum Commands {
     /// View MRIB state.
     Status(StatusCommand),
-
-    /// RPF (Reverse Path Forwarding) table configuration and lookup.
-    Rpf(RpfCommand),
 }
 
 #[derive(Debug, Args)]
@@ -109,24 +105,6 @@ pub enum StatusCmd {
     },
 }
 
-#[derive(Debug, Args)]
-pub struct RpfCommand {
-    #[command(subcommand)]
-    command: RpfCmd,
-}
-
-#[derive(Subcommand, Debug)]
-pub enum RpfCmd {
-    /// Get RPF rebuild interval.
-    GetInterval,
-
-    /// Set RPF rebuild interval.
-    SetInterval {
-        /// Rebuild interval in milliseconds
-        interval_ms: u64,
-    },
-}
-
 pub async fn commands(command: Commands, c: Client) -> Result<()> {
     match command {
         Commands::Status(status_cmd) => match status_cmd.command {
@@ -157,12 +135,6 @@ pub async fn commands(command: Commands, c: Client) -> Result<()> {
                 }
             }
         },
-        Commands::Rpf(rpf_cmd) => match rpf_cmd.command {
-            RpfCmd::GetInterval => get_rpf_interval(c).await?,
-            RpfCmd::SetInterval { interval_ms } => {
-                set_rpf_interval(c, interval_ms).await?
-            }
-        },
     }
     Ok(())
 }
@@ -229,21 +201,6 @@ async fn get_route_selected(
     Ok(())
 }
 
-async fn get_rpf_interval(c: Client) -> Result<()> {
-    let result = c.read_mrib_rpf_rebuild_interval().await?.into_inner();
-    println_nopipe!("RPF rebuild interval: {}ms", result.interval_ms);
-    Ok(())
-}
-
-async fn set_rpf_interval(c: Client, interval_ms: u64) -> Result<()> {
-    c.update_mrib_rpf_rebuild_interval(&MribRpfRebuildIntervalRequest {
-        interval_ms,
-    })
-    .await?;
-    println_nopipe!("Updated RPF rebuild interval to: {interval_ms}ms");
-    Ok(())
-}
-
 fn print_routes(routes: &[MulticastRoute]) {
     if routes.is_empty() {
         println_nopipe!("No multicast routes");
@@ -295,21 +252,19 @@ mod tests {
         ])
         .unwrap();
 
-        match cli.command {
-            Commands::Status(cmd) => match cmd.command {
-                StatusCmd::Imported {
-                    group, source, vni, ..
-                } => {
-                    assert_eq!(
-                        group,
-                        Some(IpAddr::V4(Ipv4Addr::new(225, 1, 2, 3)))
-                    );
-                    assert_eq!(source, None);
-                    assert_eq!(vni, DEFAULT_VNI);
-                }
-                _ => panic!("expected Imported"),
-            },
-            _ => panic!("expected Status command"),
+        let Commands::Status(cmd) = cli.command;
+        match cmd.command {
+            StatusCmd::Imported {
+                group, source, vni, ..
+            } => {
+                assert_eq!(
+                    group,
+                    Some(IpAddr::V4(Ipv4Addr::new(225, 1, 2, 3)))
+                );
+                assert_eq!(source, None);
+                assert_eq!(vni, DEFAULT_VNI);
+            }
+            _ => panic!("expected Imported"),
         }
     }
 
@@ -328,24 +283,22 @@ mod tests {
         ])
         .unwrap();
 
-        match cli.command {
-            Commands::Status(cmd) => match cmd.command {
-                StatusCmd::Imported {
-                    group, source, vni, ..
-                } => {
-                    assert_eq!(
-                        group,
-                        Some(IpAddr::V4(Ipv4Addr::new(225, 1, 2, 3)))
-                    );
-                    assert_eq!(
-                        source,
-                        Some(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)))
-                    );
-                    assert_eq!(vni, 100);
-                }
-                _ => panic!("expected Imported"),
-            },
-            _ => panic!("expected Status command"),
+        let Commands::Status(cmd) = cli.command;
+        match cmd.command {
+            StatusCmd::Imported {
+                group, source, vni, ..
+            } => {
+                assert_eq!(
+                    group,
+                    Some(IpAddr::V4(Ipv4Addr::new(225, 1, 2, 3)))
+                );
+                assert_eq!(
+                    source,
+                    Some(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)))
+                );
+                assert_eq!(vni, 100);
+            }
+            _ => panic!("expected Imported"),
         }
     }
 
@@ -364,28 +317,26 @@ mod tests {
         ])
         .unwrap();
 
-        match cli.command {
-            Commands::Status(cmd) => match cmd.command {
-                StatusCmd::Selected {
-                    group, source, vni, ..
-                } => {
-                    assert_eq!(
-                        group,
-                        Some(IpAddr::V6(Ipv6Addr::new(
-                            0xff0e, 0, 0, 0, 0, 0, 0, 1
-                        )))
-                    );
-                    assert_eq!(
-                        source,
-                        Some(IpAddr::V6(Ipv6Addr::new(
-                            0x2001, 0xdb8, 0, 0, 0, 0, 0, 1
-                        )))
-                    );
-                    assert_eq!(vni, 42);
-                }
-                _ => panic!("expected Selected"),
-            },
-            _ => panic!("expected Status command"),
+        let Commands::Status(cmd) = cli.command;
+        match cmd.command {
+            StatusCmd::Selected {
+                group, source, vni, ..
+            } => {
+                assert_eq!(
+                    group,
+                    Some(IpAddr::V6(Ipv6Addr::new(
+                        0xff0e, 0, 0, 0, 0, 0, 0, 1
+                    )))
+                );
+                assert_eq!(
+                    source,
+                    Some(IpAddr::V6(Ipv6Addr::new(
+                        0x2001, 0xdb8, 0, 0, 0, 0, 0, 1
+                    )))
+                );
+                assert_eq!(vni, 42);
+            }
+            _ => panic!("expected Selected"),
         }
     }
 
@@ -395,19 +346,17 @@ mod tests {
             TestCli::try_parse_from(["test", "status", "imported", "ipv4"])
                 .unwrap();
 
-        match cli.command {
-            Commands::Status(cmd) => match cmd.command {
-                StatusCmd::Imported {
-                    group,
-                    address_family,
-                    ..
-                } => {
-                    assert_eq!(group, None);
-                    assert_eq!(address_family, Some(AddressFamily::Ipv4));
-                }
-                _ => panic!("expected Imported"),
-            },
-            _ => panic!("expected Status command"),
+        let Commands::Status(cmd) = cli.command;
+        match cmd.command {
+            StatusCmd::Imported {
+                group,
+                address_family,
+                ..
+            } => {
+                assert_eq!(group, None);
+                assert_eq!(address_family, Some(AddressFamily::Ipv4));
+            }
+            _ => panic!("expected Imported"),
         }
     }
 
@@ -418,15 +367,13 @@ mod tests {
         ])
         .unwrap();
 
-        match cli.command {
-            Commands::Status(cmd) => match cmd.command {
-                StatusCmd::Imported { group, origin, .. } => {
-                    assert_eq!(group, None);
-                    assert_eq!(origin, Some(RouteOriginFilter::Dynamic));
-                }
-                _ => panic!("expected Imported"),
-            },
-            _ => panic!("expected Status command"),
+        let Commands::Status(cmd) = cli.command;
+        match cmd.command {
+            StatusCmd::Imported { group, origin, .. } => {
+                assert_eq!(group, None);
+                assert_eq!(origin, Some(RouteOriginFilter::Dynamic));
+            }
+            _ => panic!("expected Imported"),
         }
     }
 
@@ -435,38 +382,19 @@ mod tests {
         let cli =
             TestCli::try_parse_from(["test", "status", "selected"]).unwrap();
 
-        match cli.command {
-            Commands::Status(cmd) => match cmd.command {
-                StatusCmd::Selected {
-                    group,
-                    address_family,
-                    origin,
-                    ..
-                } => {
-                    assert_eq!(group, None);
-                    assert_eq!(address_family, None);
-                    assert_eq!(origin, None);
-                }
-                _ => panic!("expected Selected"),
-            },
-            _ => panic!("expected Status command"),
-        }
-    }
-
-    #[test]
-    fn test_rpf_set_interval() {
-        let cli =
-            TestCli::try_parse_from(["test", "rpf", "set-interval", "500"])
-                .unwrap();
-
-        match cli.command {
-            Commands::Rpf(cmd) => match cmd.command {
-                RpfCmd::SetInterval { interval_ms } => {
-                    assert_eq!(interval_ms, 500);
-                }
-                _ => panic!("expected SetInterval"),
-            },
-            _ => panic!("expected Rpf command"),
+        let Commands::Status(cmd) = cli.command;
+        match cmd.command {
+            StatusCmd::Selected {
+                group,
+                address_family,
+                origin,
+                ..
+            } => {
+                assert_eq!(group, None);
+                assert_eq!(address_family, None);
+                assert_eq!(origin, None);
+            }
+            _ => panic!("expected Selected"),
         }
     }
 }