freenet 0.2.40

//! Subscription topology registry for simulation testing.
//!
//! This module provides a global registry for subscription topology snapshots,
//! enabling SimNetwork to validate subscription tree structure during tests.
//!
//! # Usage
//!
//! Nodes register their subscription state using `register_topology_snapshot()`.
//! SimNetwork queries the registry using `get_topology_snapshot()` or
//! `get_all_topology_snapshots()` to validate topology correctness.
//!
//! # Thread Safety
//!
//! The registry uses `DashMap` for thread-safe access without explicit locking.
//! Multiple nodes can register snapshots concurrently.

use dashmap::DashMap;
use freenet_stdlib::prelude::{ContractInstanceId, ContractKey};
use std::{
    cell::RefCell,
    collections::{HashMap, HashSet},
    net::SocketAddr,
    sync::LazyLock,
};

/// Global registry for subscription topology snapshots.
/// Key: (network_name, peer_address)
/// Value: TopologySnapshot for that peer
static TOPOLOGY_REGISTRY: LazyLock<DashMap<(String, SocketAddr), TopologySnapshot>> =
    LazyLock::new(DashMap::new);

// Thread-local current network name: allows parallel simulation tests.
thread_local! {
    static CURRENT_NETWORK_NAME: RefCell<Option<String>> = const { RefCell::new(None) };
}

/// Set the current simulation network name (thread-local).
/// Called by SimNetwork before starting nodes.
pub fn set_current_network_name(name: &str) {
    CURRENT_NETWORK_NAME.with(|n| *n.borrow_mut() = Some(name.to_string()));
}

/// Get the current simulation network name (thread-local).
/// Returns None if not in a simulation context.
pub fn get_current_network_name() -> Option<String> {
    CURRENT_NETWORK_NAME.with(|n| n.borrow().clone())
}

/// Clear the current simulation network name (thread-local).
/// Called when SimNetwork is dropped or test ends.
pub fn clear_current_network_name() {
    CURRENT_NETWORK_NAME.with(|n| *n.borrow_mut() = None);
}

/// A snapshot of a peer's subscription topology for a contract.
#[derive(Debug, Clone)]
pub struct ContractSubscription {
    /// The contract key
    pub contract_key: ContractKey,
    /// The upstream peer (if any) - where we receive updates from
    pub upstream: Option<SocketAddr>,
    /// Downstream peers - peers that receive updates from us
    pub downstream: Vec<SocketAddr>,
    /// Whether we're hosting this contract (have it cached)
    pub is_hosting: bool,
    /// Whether we have local client subscriptions for this contract
    pub has_client_subscriptions: bool,
}

/// A snapshot of a peer's complete subscription topology.
#[derive(Debug, Clone)]
pub struct TopologySnapshot {
    /// The peer's socket address
    pub peer_addr: SocketAddr,
    /// The peer's ring location (for proximity analysis)
    pub location: f64,
    /// Subscriptions per contract
    pub contracts: HashMap<ContractInstanceId, ContractSubscription>,
    /// Timestamp when this snapshot was taken (for staleness detection)
    pub timestamp_nanos: u64,
}

impl TopologySnapshot {
    /// Create a new empty topology snapshot for a peer.
    pub fn new(peer_addr: SocketAddr, location: f64) -> Self {
        Self {
            peer_addr,
            location,
            contracts: HashMap::new(),
            timestamp_nanos: 0,
        }
    }

    /// Add or update a contract subscription.
    pub fn set_contract(
        &mut self,
        contract_id: ContractInstanceId,
        subscription: ContractSubscription,
    ) {
        self.contracts.insert(contract_id, subscription);
    }

    /// Get the subscription for a contract.
    pub fn get_contract(&self, contract_id: &ContractInstanceId) -> Option<&ContractSubscription> {
        self.contracts.get(contract_id)
    }

    /// Check if this peer has an upstream for a contract.
    pub fn has_upstream(&self, contract_id: &ContractInstanceId) -> bool {
        self.contracts
            .get(contract_id)
            .map(|s| s.upstream.is_some())
            .unwrap_or(false)
    }

    /// Check if this peer is hosting a contract.
    pub fn is_hosting(&self, contract_id: &ContractInstanceId) -> bool {
        self.contracts
            .get(contract_id)
            .map(|s| s.is_hosting)
            .unwrap_or(false)
    }
}

/// Register a topology snapshot for a peer in a network.
pub fn register_topology_snapshot(network_name: &str, snapshot: TopologySnapshot) {
    let key = (network_name.to_string(), snapshot.peer_addr);
    TOPOLOGY_REGISTRY.insert(key, snapshot);
}

/// Get the topology snapshot for a specific peer in a network.
pub fn get_topology_snapshot(
    network_name: &str,
    peer_addr: &SocketAddr,
) -> Option<TopologySnapshot> {
    let key = (network_name.to_string(), *peer_addr);
    TOPOLOGY_REGISTRY.get(&key).map(|r| r.value().clone())
}

/// Get all topology snapshots for a network.
pub fn get_all_topology_snapshots(network_name: &str) -> Vec<TopologySnapshot> {
    let mut snapshots: Vec<_> = TOPOLOGY_REGISTRY
        .iter()
        .filter(|entry| entry.key().0 == network_name)
        .map(|entry| (entry.key().1, entry.value().clone()))
        .collect();
    // Sort by socket address for deterministic ordering (critical for simulation tests)
    snapshots.sort_by(|a, b| {
        let addr_a = format!("{}", a.0);
        let addr_b = format!("{}", b.0);
        addr_a.cmp(&addr_b)
    });
    snapshots
        .into_iter()
        .map(|(_, snapshot)| snapshot)
        .collect()
}

/// Clear all topology snapshots for a network.
pub fn clear_topology_snapshots(network_name: &str) {
    TOPOLOGY_REGISTRY.retain(|key, _| key.0 != network_name);
}

/// Clear all topology snapshots (for test cleanup).
pub fn clear_all_topology_snapshots() {
    TOPOLOGY_REGISTRY.clear();
}

/// Result of topology validation.
#[derive(Debug, Default)]
pub struct TopologyValidationResult {
    /// Bidirectional cycles detected (pairs of peers)
    pub bidirectional_cycles: Vec<(SocketAddr, SocketAddr)>,
    /// Orphan hosters (peers hosting without upstream or downstream)
    pub orphan_hosters: Vec<(SocketAddr, ContractInstanceId)>,
    /// Disconnected upstream (hosters with downstream but no upstream, not a source)
    /// These are problematic because downstream peers depend on them but they can't receive updates
    pub disconnected_upstream: Vec<(SocketAddr, ContractInstanceId)>,
    /// Unreachable hosters (hosters that can't receive updates from source)
    pub unreachable_hosters: Vec<(SocketAddr, ContractInstanceId)>,
    /// Proximity violations (upstream is farther from contract than downstream)
    pub proximity_violations: Vec<ProximityViolation>,
    /// Total number of issues found
    pub issue_count: usize,
}

impl TopologyValidationResult {
    /// Check if the topology is healthy (no issues).
    pub fn is_healthy(&self) -> bool {
        self.issue_count == 0
    }
}

/// A proximity violation in upstream selection.
#[derive(Debug, Clone)]
pub struct ProximityViolation {
    pub contract_id: ContractInstanceId,
    pub downstream_addr: SocketAddr,
    pub upstream_addr: SocketAddr,
    pub downstream_location: f64,
    pub upstream_location: f64,
    pub contract_location: f64,
}

/// Validate the subscription topology for a contract across all peers in a network.
///
/// This function checks for:
/// - Bidirectional cycles that create isolated islands
/// - Orphan hosters without recovery paths
/// - Unreachable hosters that can't receive updates
/// - Proximity violations in upstream selection
pub fn validate_topology(
    network_name: &str,
    contract_id: &ContractInstanceId,
    contract_location: f64,
) -> TopologyValidationResult {
    let snapshots = get_all_topology_snapshots(network_name);
    validate_topology_from_snapshots(&snapshots, contract_id, contract_location)
}

/// Validates the subscription topology for a contract from provided snapshots.
///
/// Use this variant when you have captured snapshots and the global registry
/// may have been cleared (e.g., after SimNetwork::Drop).
///
/// Same validation as `validate_topology` but operates on the provided snapshots
/// instead of fetching from the global registry.
pub fn validate_topology_from_snapshots(
    snapshots: &[TopologySnapshot],
    contract_id: &ContractInstanceId,
    contract_location: f64,
) -> TopologyValidationResult {
    let mut result = TopologyValidationResult::default();

    // Build a map of peer -> (upstream, downstream) for this contract
    let mut subscription_graph: HashMap<SocketAddr, (Option<SocketAddr>, Vec<SocketAddr>)> =
        HashMap::new();
    let mut peer_locations: HashMap<SocketAddr, f64> = HashMap::new();
    let mut hosters: HashSet<SocketAddr> = HashSet::new();

    for snapshot in snapshots {
        peer_locations.insert(snapshot.peer_addr, snapshot.location);

        if let Some(sub) = snapshot.contracts.get(contract_id) {
            subscription_graph.insert(snapshot.peer_addr, (sub.upstream, sub.downstream.clone()));

            if sub.is_hosting {
                hosters.insert(snapshot.peer_addr);
            }
        }
    }

    // Check for bidirectional cycles
    for (&peer, (upstream, _)) in &subscription_graph {
        if let Some(upstream_addr) = upstream {
            // Check if the upstream also has us as their upstream (bidirectional)
            if let Some((their_upstream, _)) = subscription_graph.get(upstream_addr) {
                if their_upstream == &Some(peer) {
                    // Found a bidirectional cycle
                    let pair = if peer < *upstream_addr {
                        (peer, *upstream_addr)
                    } else {
                        (*upstream_addr, peer)
                    };
                    if !result.bidirectional_cycles.contains(&pair) {
                        result.bidirectional_cycles.push(pair);
                        result.issue_count += 1;
                    }
                }
            }
        }
    }

    // Source detection threshold: peers within 5% of ring distance to contract are considered sources
    const SOURCE_THRESHOLD: f64 = 0.05;

    // Check if any hoster is within SOURCE_THRESHOLD (a "proper" source)
    let has_proper_source = hosters.iter().any(|hoster| {
        peer_locations
            .get(hoster)
            .map(|loc| ring_distance(*loc, contract_location) < SOURCE_THRESHOLD)
            .unwrap_or(false)
    });

    // Find de-facto sources: when no peer is within SOURCE_THRESHOLD,
    // any hoster that is acting as a tree root (no upstream, has downstream) is a valid source.
    // This is important for Issue #2755 - topology should still be valid even without a "proper" source.
    let de_facto_sources: HashSet<SocketAddr> = if has_proper_source {
        HashSet::new()
    } else {
        // Find all hosters that are acting as tree roots (no upstream but has downstream)
        hosters
            .iter()
            .filter(|hoster| {
                subscription_graph
                    .get(*hoster)
                    .map(|(upstream, downstream)| upstream.is_none() && !downstream.is_empty())
                    .unwrap_or(false)
            })
            .copied()
            .collect()
    };

    // Check for orphan hosters and disconnected upstream
    for &hoster in &hosters {
        if let Some((upstream, downstream)) = subscription_graph.get(&hoster) {
            // Check if hoster is close to contract location (is source)
            // Use ring_distance for consistent wrap-around handling
            let is_proper_source = peer_locations
                .get(&hoster)
                .map(|loc| ring_distance(*loc, contract_location) < SOURCE_THRESHOLD)
                .unwrap_or(false);

            // Also check if this is a de-facto source (acting as tree root when no proper source exists)
            let is_de_facto_source = de_facto_sources.contains(&hoster);

            let is_source = is_proper_source || is_de_facto_source;

            // Orphan if: not source, no upstream, no downstream
            if !is_source && upstream.is_none() && downstream.is_empty() {
                result.orphan_hosters.push((hoster, *contract_id));
                result.issue_count += 1;
            }

            // Disconnected upstream: has downstream but no upstream (not a source)
            // This is problematic because downstream peers depend on us but we can't receive updates
            if !is_source && upstream.is_none() && !downstream.is_empty() {
                result.disconnected_upstream.push((hoster, *contract_id));
                result.issue_count += 1;
            }
        }
    }

    // Check for proximity violations
    for (&peer, (upstream, _)) in &subscription_graph {
        if let Some(upstream_addr) = upstream {
            let peer_loc = peer_locations.get(&peer).copied().unwrap_or(0.0);
            let upstream_loc = peer_locations.get(upstream_addr).copied().unwrap_or(0.0);

            let peer_dist = ring_distance(peer_loc, contract_location);
            let upstream_dist = ring_distance(upstream_loc, contract_location);

            // Upstream should be closer to contract than downstream (with tolerance)
            if upstream_dist > peer_dist + 0.1 {
                result.proximity_violations.push(ProximityViolation {
                    contract_id: *contract_id,
                    downstream_addr: peer,
                    upstream_addr: *upstream_addr,
                    downstream_location: peer_loc,
                    upstream_location: upstream_loc,
                    contract_location,
                });
                result.issue_count += 1;
            }
        }
    }

    // Check for unreachable hosters using BFS from source
    // Use ring_distance for consistent wrap-around handling
    let source_candidates: Vec<_> = peer_locations
        .iter()
        .filter(|(_, loc)| ring_distance(**loc, contract_location) < SOURCE_THRESHOLD)
        .map(|(addr, _)| *addr)
        .collect();

    if !source_candidates.is_empty() {
        let mut reachable: HashSet<SocketAddr> = HashSet::new();
        let mut to_visit: Vec<SocketAddr> = source_candidates;

        while let Some(peer) = to_visit.pop() {
            if reachable.insert(peer) {
                if let Some((_, downstream)) = subscription_graph.get(&peer) {
                    to_visit.extend(downstream.iter().copied());
                }
            }
        }

        for &hoster in &hosters {
            if !reachable.contains(&hoster) {
                result.unreachable_hosters.push((hoster, *contract_id));
                result.issue_count += 1;
            }
        }
    }

    result
}

/// Calculate distance on the ring (handles wrap-around).
fn ring_distance(a: f64, b: f64) -> f64 {
    let diff = (a - b).abs();
    diff.min(1.0 - diff)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn make_contract_id(seed: u8) -> ContractInstanceId {
        ContractInstanceId::new([seed; 32])
    }

    fn make_contract_key(seed: u8) -> ContractKey {
        use freenet_stdlib::prelude::CodeHash;
        ContractKey::from_id_and_code(
            ContractInstanceId::new([seed; 32]),
            CodeHash::new([seed.wrapping_add(1); 32]),
        )
    }

    #[test]
    fn test_bidirectional_cycle_detection() {
        let network = "test-bidirectional";
        clear_topology_snapshots(network);

        let peer_a: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let peer_b: SocketAddr = "10.0.2.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Create bidirectional cycle: A → B and B → A
        let mut snap_a = TopologySnapshot::new(peer_a, 0.3);
        snap_a.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: Some(peer_b),
                downstream: vec![peer_b], // B is also downstream
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        let mut snap_b = TopologySnapshot::new(peer_b, 0.4);
        snap_b.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: Some(peer_a),
                downstream: vec![peer_a], // A is also downstream
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap_a);
        register_topology_snapshot(network, snap_b);

        let result = validate_topology(network, &contract_id, 0.5);
        assert!(
            !result.bidirectional_cycles.is_empty(),
            "Should detect bidirectional cycle"
        );

        clear_topology_snapshots(network);
    }

    #[test]
    fn test_orphan_hoster_detection() {
        let network = "test-orphan";
        clear_topology_snapshots(network);

        let peer: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Create orphan hoster: hosting but no upstream, no downstream
        let mut snap = TopologySnapshot::new(peer, 0.3);
        snap.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap);

        let result = validate_topology(network, &contract_id, 0.5);
        assert!(
            !result.orphan_hosters.is_empty(),
            "Should detect orphan hoster"
        );

        clear_topology_snapshots(network);
    }

    #[test]
    fn test_de_facto_source_single_hoster() {
        // Issue #2755: When no peer is within SOURCE_THRESHOLD, a single hoster
        // acting as tree root (has downstream, no upstream) should be recognized
        // as a valid de-facto source, NOT flagged as disconnected upstream.
        let network = "test-de-facto-source";
        clear_topology_snapshots(network);

        let peer: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let downstream_peer: SocketAddr = "10.0.2.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Single hoster acting as tree root: no upstream, has downstream
        // Location 0.3, contract at 0.5, distance = 0.2 (> 0.05 threshold, not a proper source)
        // But it's the only hoster and acting as root, so it's a valid de-facto source
        let mut snap = TopologySnapshot::new(peer, 0.3);
        snap.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![downstream_peer],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap);

        let result = validate_topology(network, &contract_id, 0.5);
        assert!(
            result.disconnected_upstream.is_empty(),
            "Single hoster acting as tree root should be valid de-facto source, not disconnected"
        );
        assert!(result.is_healthy(), "Topology should be healthy");

        clear_topology_snapshots(network);
    }

    #[test]
    fn test_disconnected_upstream_with_proper_source() {
        // When a proper source EXISTS (within threshold), other hosters acting
        // as root (no upstream, has downstream) ARE disconnected upstreams.
        let network = "test-disconnected-with-source";
        clear_topology_snapshots(network);

        let source_peer: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let disconnected_peer: SocketAddr = "10.0.2.1:5000".parse().unwrap();
        let downstream_peer: SocketAddr = "10.0.3.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Proper source: within threshold (location 0.52, contract at 0.5, distance = 0.02)
        let mut source_snap = TopologySnapshot::new(source_peer, 0.52);
        source_snap.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        // Disconnected peer: not a source (location 0.3, distance = 0.2), but acting as root
        let mut disconnected_snap = TopologySnapshot::new(disconnected_peer, 0.3);
        disconnected_snap.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![downstream_peer], // Has downstream but no upstream
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, source_snap);
        register_topology_snapshot(network, disconnected_snap);

        let result = validate_topology(network, &contract_id, 0.5);
        assert!(
            !result.disconnected_upstream.is_empty(),
            "Should detect disconnected upstream when proper source exists"
        );

        clear_topology_snapshots(network);
    }

    #[test]
    fn test_ring_distance_wrap_around() {
        let network = "test-wrap-around";
        clear_topology_snapshots(network);

        let peer: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Contract at location 0.02, peer at location 0.99
        // Ring distance should be 0.03 (through wrap-around), not 0.97
        // This peer should be considered a source (within 0.05 threshold)
        let mut snap = TopologySnapshot::new(peer, 0.99);
        snap.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap);

        // Contract location at 0.02 - peer at 0.99 is within 0.05 ring distance
        let result = validate_topology(network, &contract_id, 0.02);
        assert!(
            result.orphan_hosters.is_empty(),
            "Peer at 0.99 should be considered source for contract at 0.02 (ring distance 0.03)"
        );
        assert!(
            result.disconnected_upstream.is_empty(),
            "Peer at 0.99 should be considered source, not disconnected upstream"
        );

        clear_topology_snapshots(network);
    }

    #[test]
    fn test_source_hoster_not_orphan() {
        let network = "test-source-not-orphan";
        clear_topology_snapshots(network);

        let peer: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Peer at same location as contract (0.5) - should be considered source
        let mut snap = TopologySnapshot::new(peer, 0.5);
        snap.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![], // No upstream, no downstream - but it's a source
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap);

        let result = validate_topology(network, &contract_id, 0.5);
        assert!(
            result.orphan_hosters.is_empty(),
            "Source hoster should not be flagged as orphan"
        );

        clear_topology_snapshots(network);
    }

    #[test]
    fn test_proximity_violation_detection() {
        let network = "test-proximity-violation";
        clear_topology_snapshots(network);

        let peer: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let upstream_peer: SocketAddr = "10.0.2.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Contract at 0.5
        // Downstream peer at 0.45 (distance 0.05 from contract)
        // Upstream peer at 0.8 (distance 0.3 from contract)
        // Violation: upstream is farther from contract than downstream
        let mut snap_downstream = TopologySnapshot::new(peer, 0.45);
        snap_downstream.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: Some(upstream_peer),
                downstream: vec![],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        let mut snap_upstream = TopologySnapshot::new(upstream_peer, 0.8);
        snap_upstream.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![peer],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap_downstream);
        register_topology_snapshot(network, snap_upstream);

        let result = validate_topology(network, &contract_id, 0.5);
        assert!(
            !result.proximity_violations.is_empty(),
            "Should detect proximity violation when upstream is farther from contract than downstream"
        );

        // Verify the violation details
        let violation = &result.proximity_violations[0];
        assert_eq!(violation.downstream_addr, peer);
        assert_eq!(violation.upstream_addr, upstream_peer);

        clear_topology_snapshots(network);
    }

    #[test]
    fn test_unreachable_hoster_detection() {
        let network = "test-unreachable";
        clear_topology_snapshots(network);

        let source_peer: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let reachable_peer: SocketAddr = "10.0.2.1:5000".parse().unwrap();
        let unreachable_peer: SocketAddr = "10.0.3.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Contract at 0.5
        // Source peer at 0.5 (is source, distance 0)
        // Reachable peer at 0.6 (connected to source as downstream)
        // Unreachable peer at 0.7 (not connected to anyone)

        let mut snap_source = TopologySnapshot::new(source_peer, 0.5);
        snap_source.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![reachable_peer],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        let mut snap_reachable = TopologySnapshot::new(reachable_peer, 0.6);
        snap_reachable.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: Some(source_peer),
                downstream: vec![],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        // Unreachable peer - no upstream connection, not a source
        let mut snap_unreachable = TopologySnapshot::new(unreachable_peer, 0.7);
        snap_unreachable.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None, // No upstream!
                downstream: vec![],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap_source);
        register_topology_snapshot(network, snap_reachable);
        register_topology_snapshot(network, snap_unreachable);

        let result = validate_topology(network, &contract_id, 0.5);
        assert!(
            !result.unreachable_hosters.is_empty(),
            "Should detect unreachable hoster"
        );
        assert!(
            result
                .unreachable_hosters
                .iter()
                .any(|(addr, _)| *addr == unreachable_peer),
            "Unreachable peer should be in the list"
        );

        clear_topology_snapshots(network);
    }

    #[test]
    fn test_healthy_topology_no_issues() {
        let network = "test-healthy";
        clear_topology_snapshots(network);

        let source_peer: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let downstream_peer: SocketAddr = "10.0.2.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Healthy topology: source → downstream (correct direction)
        // Contract at 0.5, source at 0.5 (is source), downstream at 0.6
        let mut snap_source = TopologySnapshot::new(source_peer, 0.5);
        snap_source.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![downstream_peer],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        let mut snap_downstream = TopologySnapshot::new(downstream_peer, 0.6);
        snap_downstream.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: Some(source_peer),
                downstream: vec![],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap_source);
        register_topology_snapshot(network, snap_downstream);

        let result = validate_topology(network, &contract_id, 0.5);
        assert!(
            result.is_healthy(),
            "Healthy topology should have no issues, got: cycles={}, orphans={}, disconnected={}, unreachable={}, proximity={}",
            result.bidirectional_cycles.len(),
            result.orphan_hosters.len(),
            result.disconnected_upstream.len(),
            result.unreachable_hosters.len(),
            result.proximity_violations.len()
        );

        clear_topology_snapshots(network);
    }

    /// Tests detection of mutual downstream relationships (Issue #2773).
    ///
    /// This scenario occurs when two peers race to subscribe and both add
    /// each other as downstream before either can establish an upstream.
    ///
    /// ## Scenario
    ///
    /// - Peer A: downstream=[B], upstream=None
    /// - Peer B: downstream=[A], upstream=None
    ///
    /// Both have each other as downstream, but neither has an upstream.
    /// This blocks both from receiving updates from the contract source.
    ///
    /// ## Current Behavior (Limitation)
    ///
    /// When no proper source exists (both peers are far from contract location),
    /// the validation logic treats ANY peer with (no upstream, has downstream)
    /// as a valid "de-facto source". This means mutual downstream relationships
    /// are NOT detected when no proper source exists nearby.
    ///
    /// ## When It IS Detected
    ///
    /// When a proper source EXISTS (peer within SOURCE_THRESHOLD), mutual
    /// downstream peers are correctly flagged as `disconnected_upstream`.
    /// This test uses a proper source to demonstrate detection.
    #[test]
    fn test_mutual_downstream_detection_with_proper_source_issue_2773() {
        let network = "test-mutual-downstream-with-source";
        clear_topology_snapshots(network);

        let source_peer: SocketAddr = "10.0.0.1:5000".parse().unwrap();
        let peer_a: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let peer_b: SocketAddr = "10.0.2.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Proper source: within threshold (location 0.51, contract at 0.5, distance = 0.01)
        let mut snap_source = TopologySnapshot::new(source_peer, 0.51);
        snap_source.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![], // Source with no subscribers yet
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        // Create mutual downstream: A has B as downstream, B has A as downstream
        // Neither has upstream. Both are far from contract (not sources).
        // Location 0.2 and 0.3, contract at 0.5:
        // - A distance: 0.3 > SOURCE_THRESHOLD (0.05)
        // - B distance: 0.2 > SOURCE_THRESHOLD (0.05)

        let mut snap_a = TopologySnapshot::new(peer_a, 0.2);
        snap_a.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,           // No upstream!
                downstream: vec![peer_b], // B is downstream
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        let mut snap_b = TopologySnapshot::new(peer_b, 0.3);
        snap_b.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,           // No upstream!
                downstream: vec![peer_a], // A is downstream
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap_source);
        register_topology_snapshot(network, snap_a);
        register_topology_snapshot(network, snap_b);

        let result = validate_topology(network, &contract_id, 0.5);

        // With a proper source, both A and B should be flagged as disconnected_upstream
        // (has downstream but no upstream, and not a source since proper source exists)
        assert_eq!(
            result.disconnected_upstream.len(),
            2,
            "Both peers should be flagged as disconnected upstream when proper source exists, found: {:?}",
            result.disconnected_upstream
        );

        // Verify both peers are in the disconnected list
        let disconnected_addrs: Vec<_> = result
            .disconnected_upstream
            .iter()
            .map(|(a, _)| *a)
            .collect();
        assert!(
            disconnected_addrs.contains(&peer_a),
            "Peer A should be in disconnected_upstream"
        );
        assert!(
            disconnected_addrs.contains(&peer_b),
            "Peer B should be in disconnected_upstream"
        );

        // This is NOT a bidirectional cycle (which requires mutual UPSTREAM)
        assert!(
            result.bidirectional_cycles.is_empty(),
            "Mutual downstream is not a bidirectional cycle (no upstream established)"
        );

        // The topology is NOT healthy
        assert!(
            !result.is_healthy(),
            "Mutual downstream should be flagged as unhealthy"
        );

        clear_topology_snapshots(network);
    }

    /// Tests that mutual downstream WITHOUT a proper source is NOT currently detected.
    ///
    /// This is a KNOWN LIMITATION. When no peer is within SOURCE_THRESHOLD, the
    /// validation treats any peer with (no upstream, has downstream) as a valid
    /// "de-facto source" for Issue #2755. But with mutual downstream, BOTH peers
    /// qualify as de-facto sources, so neither is flagged.
    ///
    /// ## Future Work
    ///
    /// The validation should detect when multiple "de-facto sources" reference
    /// each other's downstream, as this creates an isolated loop rather than
    /// a valid tree rooted at a source. This would require checking if de-facto
    /// sources form a connected component with no external upstream.
    #[test]
    fn test_mutual_downstream_without_source_not_detected_limitation() {
        let network = "test-mutual-downstream-no-source";
        clear_topology_snapshots(network);

        let peer_a: SocketAddr = "10.0.1.1:5000".parse().unwrap();
        let peer_b: SocketAddr = "10.0.2.1:5000".parse().unwrap();
        let contract_id = make_contract_id(1);
        let contract_key = make_contract_key(1);

        // Create mutual downstream: A has B as downstream, B has A as downstream
        // Neither has upstream. NO proper source exists.
        let mut snap_a = TopologySnapshot::new(peer_a, 0.2);
        snap_a.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![peer_b],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        let mut snap_b = TopologySnapshot::new(peer_b, 0.3);
        snap_b.set_contract(
            contract_id,
            ContractSubscription {
                contract_key,
                upstream: None,
                downstream: vec![peer_a],
                is_hosting: true,
                has_client_subscriptions: false,
            },
        );

        register_topology_snapshot(network, snap_a);
        register_topology_snapshot(network, snap_b);

        let result = validate_topology(network, &contract_id, 0.5);

        // KNOWN LIMITATION: Both peers are treated as valid de-facto sources
        // because no proper source exists and both have (no upstream, has downstream).
        // This means mutual downstream is NOT detected in this scenario.
        assert!(
            result.disconnected_upstream.is_empty(),
            "KNOWN LIMITATION: Mutual downstream without proper source is not detected"
        );

        // The topology appears healthy but is actually broken
        // (updates from neither peer can reach an actual source)
        assert!(
            result.is_healthy(),
            "KNOWN LIMITATION: Mutual downstream appears healthy when no source exists"
        );

        clear_topology_snapshots(network);
    }
}