[multicast] Adapt multicast code to new dendrite DPD API

nexus/db-queries/src/db/datastore/multicast/members.rs

@@ -39,9 +39,8 @@ use crate::db::pagination::paginated;
 /// - **SSM (232.0.0.0/8, ff3x::/32)**: Always use `specific_sources` per RFC 4607.
 ///   The `has_any_source_member` flag is ignored because API validation
 ///   prevents SSM joins without sources.
-/// - **ASM**: Currently always passes `None` to DPD (Dendrite doesn't support
-///   ASM filtering yet). TODO: if `has_any_source_member` is true, skip
-///   switch-level filtering; otherwise use `specific_sources`.
+/// - **ASM**: If `has_any_source_member` is true, passes `None` to DPD
+///   (no switch-level filtering). Otherwise uses `specific_sources`.
 /// - **OPTE**: Always uses per-member source lists for fine-grained filtering,
 ///   regardless of switch-level behavior.
 ///
@@ -57,8 +56,8 @@ pub struct SourceFilterState {
 
     /// True if any member has empty `source_ips` (wants any source).
     ///
-    /// For ASM groups: currently unused (Dendrite doesn't support ASM filtering).
-    /// TODO: when true, switch-level filtering will be disabled.
+    /// For ASM groups: when true, switch-level source filtering is disabled
+    /// (sources passed as `None` to Dendrite).
     /// For SSM groups: ignored per RFC 4607 (API validation prevents SSM joins
     /// without sources).
     pub has_any_source_member: bool,

nexus/src/app/background/tasks/multicast/groups.rs

@@ -109,21 +109,19 @@ fn dpd_state_matches_tag(
     dpd_group: &dpd_client::types::MulticastGroupExternalResponse,
     db_group: &MulticastGroup,
 ) -> bool {
-    match (&dpd_group.tag, &db_group.tag) {
-        (dpd_tag, Some(db_tag)) => dpd_tag == db_tag,
-        _ => false,
+    match &db_group.tag {
+        Some(db_tag) => dpd_group.tag.as_str() == db_tag,
+        None => false,
     }
 }
 
 /// Check if DPD sources match the expected state based on source filter.
 ///
 /// Source filtering logic per RFC 4607 (mirrors dataplane code):
-/// - SSM (232/8, ff3x::/32): MUST have specific sources. `has_any_source_member`
-///   is ignored because API validation prevents SSM joins without sources.
-/// - ASM: Currently expects `None` (Dendrite doesn't support ASM filtering yet).
-///
-/// TODO: Once Dendrite accepts ASM source filtering, enable it for ASM groups
-/// where `has_any_source_member=false`.
+/// - SSM (232/8, ff3x::/32): always expect specific sources. API validation
+///   prevents SSM joins without sources.
+/// - ASM: expect specific sources when all members specify sources,
+///   otherwise `None` to allow any source at the switch level.
 fn dpd_state_matches_sources(
     dpd_group: &dpd_client::types::MulticastGroupExternalResponse,
     source_filter: &SourceFilterState,
@@ -134,12 +132,11 @@ fn dpd_state_matches_sources(
 
     // Expected DPD state based on source filter logic (RFC 4607)
     let expected_sources = if is_ssm_address(group_ip) {
-        // SSM: always expect specific sources
         Some(&source_filter.specific_sources)
-    } else {
-        // ASM: Dendrite doesn't support ASM filtering yet
-        // TODO: check `has_any_source_member` to enable/disable filtering
+    } else if source_filter.has_any_source_member {
         None
+    } else {
+        Some(&source_filter.specific_sources)
     };
 
     match (dpd_sources, expected_sources) {
@@ -152,7 +149,7 @@ fn dpd_state_matches_sources(
                 .into_iter()
                 .filter_map(|src| match src {
                     dpd_client::types::IpSrc::Exact(ip) => Some(ip),
-                    _ => None, // Subnet matching removed in follow-up Dendrite TODO
+                    _ => None,
                 })
                 .collect();
             dpd_ips.sort();
@@ -1051,8 +1048,8 @@ mod tests {
 
     #[test]
     fn test_dpd_state_matches_sources_asm_address() {
-        // ASM address (not 232.x.x.x) - should always expect None from DPD
-        // regardless of specific_sources (Dendrite limitation, see TODO)
+        // ASM address with all members specifying sources: expect those
+        // sources in DPD.
         let source_filter = SourceFilterState {
             specific_sources: BTreeSet::from(["10.0.0.1"
                 .parse::<IpAddr>()
@@ -1062,15 +1059,15 @@ mod tests {
 
         let group = create_group("224.1.1.1"); // ASM address (not 232.x.x.x)
 
-        // DPD has None (correct for ASM)
-        let dpd_group = create_dpd_group(None);
-        assert!(dpd_state_matches_sources(&dpd_group, &source_filter, &group));
-
-        // DPD has sources (mismatch: ASM should have none)
+        // DPD has matching sources (correct)
         let dpd_group =
             create_dpd_group(Some(vec![dpd_client::types::IpSrc::Exact(
                 "10.0.0.1".parse().unwrap(),
             )]));
+        assert!(dpd_state_matches_sources(&dpd_group, &source_filter, &group));
+
+        // DPD has None (mismatch: ASM with all-specific should have sources)
+        let dpd_group = create_dpd_group(None);
         assert!(!dpd_state_matches_sources(&dpd_group, &source_filter, &group));
     }