[mutlicast] DDM multicast exchange: V4 protocol, MRIB sync, M2P programming

Adds multicast group subscription distribution to the DDM exchange
protocol with a V4 version bump (frozen V3 types for wire compat).

Key changes:
- V4 exchange protocol with multicast support (V3 peers are unaffected)
- UnderlayMulticastIpv6 validated newtype moved to mg-common (ff04::/64) (moved from rdb types)
- MRIB->DDM sync in mg-lower/mrib.rs
- OPTE M2P table programming on learned multicast routes
- Atomic update_imported_mcast on Db (single lock for import/delete/diff, which is a bit different from the tunnel work)
- Collapsed send_update dispatch
- Shared pull handler helpers (collect_underlay_tunnel, collect_multicast)
- MulticastPathHop constructor
- Some serde round-trip and validation tests, including for version handling

Stacked on zl/mrib (MRIB: Multicast RIB implementation [#675](#675)).

Author: zeeshanlakhani

Cargo.lock

@@ -39,3 +39,25 @@ impl std::hash::Hash for types::TunnelOrigin {
         self.metric.hash(state);
     }
 }
+
+impl std::cmp::PartialEq for types::MulticastOrigin {
+    fn eq(&self, other: &Self) -> bool {
+        self.overlay_group.eq(&other.overlay_group)
+            && self.underlay_group.eq(&other.underlay_group)
+            && self.vni.eq(&other.vni)
+            && self.source.eq(&other.source)
+    }
+}
+
+impl std::cmp::Eq for types::MulticastOrigin {}
+
+/// Metric is excluded from identity so that metric changes update
+/// an existing entry rather than creating a duplicate.
+impl std::hash::Hash for types::MulticastOrigin {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.overlay_group.hash(state);
+        self.underlay_group.hash(state);
+        self.vni.hash(state);
+        self.source.hash(state);
+    }
+}

ddm-admin-client/src/lib.rs

@@ -10,7 +10,7 @@ use dropshot::Path;
 use dropshot::RequestContext;
 use dropshot::TypedBody;
 use dropshot_api_manager_types::api_versions;
-use mg_common::net::TunnelOrigin;
+use mg_common::net::{MulticastOrigin, TunnelOrigin};
 use oxnet::Ipv6Net;
 use std::collections::{HashMap, HashSet};
 
@@ -26,6 +26,7 @@ api_versions!([
     // |  example for the next person.
     // v
     // (next_int, IDENT),
+    (2, MULTICAST_SUPPORT),
     (1, INITIAL),
 ]);
 
@@ -100,6 +101,44 @@ pub trait DdmAdminApi {
         request: TypedBody<HashSet<TunnelOrigin>>,
     ) -> Result<HttpResponseUpdatedNoContent, HttpError>;
 
+    #[endpoint {
+        method = GET,
+        path = "/originated_multicast_groups",
+        versions = VERSION_MULTICAST_SUPPORT..
+    }]
+    async fn get_originated_multicast_groups(
+        ctx: RequestContext<Self::Context>,
+    ) -> Result<HttpResponseOk<HashSet<MulticastOrigin>>, HttpError>;
+
+    #[endpoint {
+        method = GET,
+        path = "/multicast_groups",
+        versions = VERSION_MULTICAST_SUPPORT..
+    }]
+    async fn get_multicast_groups(
+        ctx: RequestContext<Self::Context>,
+    ) -> Result<HttpResponseOk<HashSet<latest::db::MulticastRoute>>, HttpError>;
+
+    #[endpoint {
+        method = PUT,
+        path = "/multicast_group",
+        versions = VERSION_MULTICAST_SUPPORT..
+    }]
+    async fn advertise_multicast_groups(
+        ctx: RequestContext<Self::Context>,
+        request: TypedBody<HashSet<MulticastOrigin>>,
+    ) -> Result<HttpResponseUpdatedNoContent, HttpError>;
+
+    #[endpoint {
+        method = DELETE,
+        path = "/multicast_group",
+        versions = VERSION_MULTICAST_SUPPORT..
+    }]
+    async fn withdraw_multicast_groups(
+        ctx: RequestContext<Self::Context>,
+        request: TypedBody<HashSet<MulticastOrigin>>,
+    ) -> Result<HttpResponseUpdatedNoContent, HttpError>;
+
     #[endpoint { method = PUT, path = "/sync" }]
     async fn sync(
         ctx: RequestContext<Self::Context>,

ddm-api/src/lib.rs

@@ -15,9 +15,12 @@ pub mod db {
     pub use crate::v1::db::PeerStatus;
     pub use crate::v1::db::RouterKind;
     pub use crate::v1::db::TunnelRoute;
+    pub use crate::v2::db::MulticastRoute;
 }
 
 pub mod exchange {
     pub use crate::v1::exchange::PathVector;
     pub use crate::v1::exchange::PathVectorV2;
+    pub use crate::v2::exchange::MulticastPathHop;
+    pub use crate::v2::exchange::MulticastPathVector;
 }

ddm-types/versions/src/latest.rs

@@ -32,3 +32,5 @@
 pub mod latest;
 #[path = "initial/mod.rs"]
 pub mod v1;
+#[path = "multicast_support/mod.rs"]
+pub mod v2;

ddm-types/versions/src/lib.rs

@@ -0,0 +1,62 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+use std::net::Ipv6Addr;
+
+use mg_common::net::MulticastOrigin;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::v2::exchange::MulticastPathHop;
+
+/// A multicast route learned via DDM.
+///
+/// Carries both the group origin and the path vector from the
+/// originating subscriber through intermediate transit routers.
+/// The path enables loop detection and (in multi-rack topologies)
+/// replication optimizations per [RFD 488] in the future.
+///
+/// Equality and hashing consider only `origin` and `nexthop` so that
+/// a route update with a longer path replaces the existing entry in
+/// hash-based collections.
+///
+/// [RFD 488]: https://rfd.shared.oxide.computer/rfd/0488
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
+pub struct MulticastRoute {
+    /// The multicast group origin information.
+    pub origin: MulticastOrigin,
+
+    /// Underlay nexthop address (DDM peer that advertised this route).
+    /// Used to associate the route with a peer for expiration.
+    pub nexthop: Ipv6Addr,
+
+    /// Path vector from the originating subscriber outward.
+    /// Each hop records the router that redistributed this
+    /// subscription announcement. Used for loop detection on pull
+    /// and for future replication optimization in multi-rack
+    /// topologies.
+    #[serde(default)]
+    pub path: Vec<MulticastPathHop>,
+}
+
+impl PartialEq for MulticastRoute {
+    fn eq(&self, other: &Self) -> bool {
+        self.origin == other.origin && self.nexthop == other.nexthop
+    }
+}
+
+impl Eq for MulticastRoute {}
+
+impl std::hash::Hash for MulticastRoute {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.origin.hash(state);
+        self.nexthop.hash(state);
+    }
+}
+
+impl From<MulticastRoute> for MulticastOrigin {
+    fn from(x: MulticastRoute) -> Self {
+        x.origin
+    }
+}

ddm-types/versions/src/multicast_support/db.rs

@@ -0,0 +1,76 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use std::net::Ipv6Addr;
+
+/// A single hop in the multicast path, carrying metadata needed for
+/// replication optimization.
+///
+/// Unlike unicast paths which only need hostnames, multicast hops carry
+/// additional information for computing optimal replication points per
+/// [RFD 488].
+///
+/// [RFD 488]: https://rfd.shared.oxide.computer/rfd/0488
+#[derive(
+    Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize, JsonSchema,
+)]
+pub struct MulticastPathHop {
+    /// Router identifier (hostname).
+    pub router_id: String,
+
+    /// The underlay address of this router (for replication targeting).
+    pub underlay_addr: Ipv6Addr,
+
+    /// Number of downstream subscribers reachable via this hop.
+    /// Used for load-aware replication decisions in multi-rack
+    /// topologies.
+    #[serde(default)]
+    pub downstream_subscriber_count: u32,
+}
+
+impl MulticastPathHop {
+    /// Create a hop with the given router identity and a zero subscriber
+    /// count. The count will be populated once transit routers track
+    /// downstream subscriber counts for load-aware replication (RFD 488).
+    pub fn new(router_id: String, underlay_addr: Ipv6Addr) -> Self {
+        Self {
+            router_id,
+            underlay_addr,
+            downstream_subscriber_count: 0,
+        }
+    }
+}
+
+/// Multicast group subscription announcement propagating through DDM.
+///
+/// The path records the sequence of routers from the original subscriber
+/// toward the current receiving router. Currently, this is used for loop
+/// detection: if our router_id appears in the path, the announcement has
+/// already traversed us and is dropped. The path structure also carries
+/// topology information for future replication optimizations (RFD 488).
+#[derive(
+    Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize, JsonSchema,
+)]
+pub struct MulticastPathVector {
+    /// The multicast group origin information.
+    pub origin: mg_common::net::MulticastOrigin,
+
+    /// The path from the original subscriber to the current router.
+    /// Ordered from subscriber outward (subscriber router first).
+    pub path: Vec<MulticastPathHop>,
+}
+
+impl MulticastPathVector {
+    /// Append a hop to this path vector.
+    pub fn with_hop(&self, hop: MulticastPathHop) -> Self {
+        let mut path = self.path.clone();
+        path.push(hop);
+        Self {
+            origin: self.origin.clone(),
+            path,
+        }
+    }
+}

ddm-types/versions/src/multicast_support/exchange.rs

@@ -0,0 +1,9 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+//! Types from API version 2 (MULTICAST_SUPPORT) that add multicast
+//! group management to the DDM admin API.
+
+pub mod db;
+pub mod exchange;

ddm-types/versions/src/multicast_support/mod.rs

@@ -7,7 +7,7 @@ use crate::sm::{AdminEvent, Event, PrefixSet, SmContext};
 use ddm_api::DdmAdminApi;
 use ddm_api::ddm_admin_api_mod;
 use ddm_types::admin::{EnableStatsRequest, ExpirePathParams, PrefixMap};
-use ddm_types::db::{PeerInfo, TunnelRoute};
+use ddm_types::db::{MulticastRoute, PeerInfo, TunnelRoute};
 use ddm_types::exchange::PathVector;
 use dropshot::ApiDescription;
 use dropshot::ApiDescriptionBuildErrors;
@@ -21,7 +21,7 @@ use dropshot::Path;
 use dropshot::RequestContext;
 use dropshot::TypedBody;
 use mg_common::lock;
-use mg_common::net::TunnelOrigin;
+use mg_common::net::{MulticastOrigin, TunnelOrigin};
 use oxnet::Ipv6Net;
 use slog::{Logger, error, info};
 use std::collections::{HashMap, HashSet};
@@ -333,6 +333,71 @@ impl DdmAdminApi for DdmAdminApiImpl {
         Ok(HttpResponseUpdatedNoContent())
     }
 
+    async fn get_originated_multicast_groups(
+        ctx: RequestContext<Self::Context>,
+    ) -> Result<HttpResponseOk<HashSet<MulticastOrigin>>, HttpError> {
+        let ctx = lock!(ctx.context());
+        let originated = ctx
+            .db
+            .originated_mcast()
+            .map_err(|e| HttpError::for_internal_error(e.to_string()))?;
+        Ok(HttpResponseOk(originated))
+    }
+
+    async fn get_multicast_groups(
+        ctx: RequestContext<Self::Context>,
+    ) -> Result<HttpResponseOk<HashSet<MulticastRoute>>, HttpError> {
+        let ctx = lock!(ctx.context());
+        let imported = ctx.db.imported_mcast();
+        Ok(HttpResponseOk(imported))
+    }
+
+    async fn advertise_multicast_groups(
+        ctx: RequestContext<Self::Context>,
+        request: TypedBody<HashSet<MulticastOrigin>>,
+    ) -> Result<HttpResponseUpdatedNoContent, HttpError> {
+        let ctx = lock!(ctx.context());
+        let groups = request.into_inner();
+        slog::info!(ctx.log, "advertise multicast groups: {groups:#?}");
+        ctx.db
+            .originate_mcast(&groups)
+            .map_err(|e| HttpError::for_internal_error(e.to_string()))?;
+
+        for e in &ctx.event_channels {
+            e.send(Event::Admin(AdminEvent::Announce(PrefixSet::Multicast(
+                groups.clone(),
+            ))))
+            .map_err(|e| {
+                HttpError::for_internal_error(format!("admin event send: {e}"))
+            })?;
+        }
+
+        Ok(HttpResponseUpdatedNoContent())
+    }
+
+    async fn withdraw_multicast_groups(
+        ctx: RequestContext<Self::Context>,
+        request: TypedBody<HashSet<MulticastOrigin>>,
+    ) -> Result<HttpResponseUpdatedNoContent, HttpError> {
+        let ctx = lock!(ctx.context());
+        let groups = request.into_inner();
+        slog::info!(ctx.log, "withdraw multicast groups: {groups:#?}");
+        ctx.db
+            .withdraw_mcast(&groups)
+            .map_err(|e| HttpError::for_internal_error(e.to_string()))?;
+
+        for e in &ctx.event_channels {
+            e.send(Event::Admin(AdminEvent::Withdraw(PrefixSet::Multicast(
+                groups.clone(),
+            ))))
+            .map_err(|e| {
+                HttpError::for_internal_error(format!("admin event send: {e}"))
+            })?;
+        }
+
+        Ok(HttpResponseUpdatedNoContent())
+    }
+
     async fn sync(
         ctx: RequestContext<Self::Context>,
     ) -> Result<HttpResponseUpdatedNoContent, HttpError> {