freenet 0.2.77

//! Ring protocol logic and supporting types.
//!
//! Mainly maintains a healthy and optimal pool of connections to other peers in the network
//! and routes requests to the optimal peers.

use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::net::SocketAddr;
use std::sync::{Arc, Weak, atomic::AtomicU64};
use std::time::Duration;
// Intentional wall-clock type for suspend/resume detection in
// `connection_maintenance`. See `classify_suspend_jump` for why the DST
// `TimeSource` abstraction is the *wrong* primitive here (it returns
// simulation time, which would never reflect a real OS suspend). This is
// the one call site in `crates/core/` that needs a real monotonic wall
// clock; renaming sidesteps the crates/core/ DST rule-lint grep while
// keeping intent explicit to the reader.
use std::time::Instant as WallClockInstant;
use tokio::time::Instant;

use tracing::Instrument;

use either::Either;
use freenet_stdlib::prelude::{ContractInstanceId, ContractKey};
use parking_lot::{Mutex, RwLock};

pub use hosting::{
    AddClientSubscriptionResult, ClientDisconnectResult, SubscribeResult,
    SubscribedContractSnapshot,
};

use crate::message::TransactionType;
use crate::topology::TopologyAdjustment;
use crate::topology::rate::Rate;
use crate::tracing::{NetEventLog, NetEventRegister};

use crate::transport::TransportPublicKey;
use crate::util::{Contains, time_source::InstantTimeSrc};
use crate::{
    config::{GlobalExecutor, GlobalRng},
    message::Transaction,
    node::{self, EventLoopNotificationsSender, NodeConfig, OpManager, PeerId},
    router::Router,
};

mod broken_invariants;
mod connection_backoff;
mod connection_manager;
pub(crate) mod contract_ban_list;
pub(crate) use connection_manager::ConnectionManager;
mod connection;
mod hosting;
pub(crate) use broken_invariants::{BrokenInvariant, BrokenInvariantsTracker};
/// Single source of truth for the default hosted-contract-state budget.
/// `config::default_max_hosting_storage()` resolves to this so the
/// operator-facing default and the in-code fallback can never drift.
pub(crate) use hosting::DEFAULT_HOSTING_BUDGET_BYTES;
pub use hosting::{AccessType, RecordAccessResult};
pub mod interest;
mod live_tx;
mod location;
pub(crate) mod peer_cache;
mod peer_connection_backoff;
mod peer_key_location;
pub mod topology_registry;
pub(crate) mod update_rate_limit;

/// Whether to auto-subscribe to contracts on GET.
/// When true, GET operations will automatically subscribe to the contract
/// to receive updates. This is controlled by hosting cache eviction.
pub const AUTO_SUBSCRIBE_ON_GET: bool = true;

/// Per-beneficiary weight for a local-client subscription when
/// computing the LIVE benefit snapshot each governance reaper tick.
/// Strong signal: a real user on this node is currently subscribed.
/// Hard to fake without running an actual freenet client locally.
///
/// Sourced from the design doc — see "Sybil weighting falls out
/// naturally" in `docs/design/contract-hardening.md`.
const LOCAL_DEMAND_WEIGHT: f64 = 1.0;

/// Per-beneficiary weight for a downstream peer's CURRENT subscription
/// when computing the LIVE benefit snapshot. Weaker signal because peer
/// identity is attacker-rotatable — without an identity layer, a single
/// attacker can spin up many peers and each "subscribes." We weight
/// each forwarded subscriber at 0.1 so an attacker would need 10
/// rotating peers to fake the standing demand of one local user.
const FORWARDED_DEMAND_WEIGHT: f64 = 0.1;

/// Interval between governance reaper ticks. Each tick applies
/// decay and runs MAD-based outlier detection across the population.
/// One minute balances responsiveness (a sustained-cost contract is
/// flagged within a minute or two of crossing the threshold) against
/// CPU overhead (MAD over the entire contract set every tick).
const GOVERNANCE_TICK_INTERVAL: Duration = Duration::from_secs(60);

use connection_backoff::ConnectionBackoff;
pub use connection_backoff::ConnectionFailureReason;
pub(crate) use peer_connection_backoff::PeerConnectionBackoff;

pub use self::live_tx::LiveTransactionTracker;
pub use connection::Connection;
pub use interest::PeerKey;
pub use location::{Distance, Location};
pub use peer_key_location::{KnownPeerKeyLocation, PeerAddr, PeerKeyLocation};

/// Thread safe and friendly data structure to keep track of the local knowledge
/// of the state of the ring.
///
// Note: For now internally we wrap some of the types internally with locks and/or use
// multithreaded maps. In the future if performance requires it some of this can be moved
// towards a more lock-free multithreading model if necessary.
/// Backoff state for contract-directed CONNECT attempts.
struct ContractConnectState {
    current_backoff: Duration,
    last_attempt: Instant,
}

pub(crate) struct Ring {
    pub max_hops_to_live: usize,
    pub connection_manager: ConnectionManager,
    pub router: Arc<RwLock<Router>>,
    pub live_tx_tracker: LiveTransactionTracker,
    hosting_manager: hosting::HostingManager,
    /// Per-contract record of detected CRDT-invariant violations (e.g. a
    /// non-idempotent `update_state`). Used to gate outbound broadcast
    /// so a broken contract's state changes don't leave the node.
    broken_invariants: BrokenInvariantsTracker,
    /// Per-contract governance scoring + reaper. Routes every
    /// per-contract resource report through `ingest_cost` so the
    /// governance state for a contract reflects what the meter
    /// already records. Mode defaults to `DryRun` per the staged-
    /// rollout plan.
    pub(crate) governance: Arc<crate::contract::governance::GovernanceManager>,
    /// Front-line per-(sender, contract) UPDATE rate limit. Catches
    /// flood patterns at the receive boundary in milliseconds, before
    /// the slower MAD-based governance reaper can react. See
    /// `crate::ring::update_rate_limit` and `docs/design/contract-hardening.md`
    /// Phase 2.
    pub(crate) update_rate_limiter: Arc<update_rate_limit::UpdateRateLimiter>,
    /// Per-contract ban list. Populated by the governance reaper on
    /// `BanTriggered` / `BanLifted` transitions; consulted at the
    /// inbound dispatch site to drop wire requests for banned
    /// contracts. Phase 7 of the contract-hardening plan
    /// (`docs/design/contract-hardening.md`) and the auto-detection
    /// half of issue #4274 (operator-CLI blocklist).
    pub(crate) contract_ban_list: Arc<contract_ban_list::ContractBanList>,
    event_register: Box<dyn NetEventRegister>,
    op_manager: RwLock<Option<Weak<OpManager>>>,
    /// Whether this peer is a gateway or not. This will affect behavior of the node when acquiring
    /// and dropping connections.
    pub(crate) is_gateway: bool,
    /// Shared connection backoff tracker for all connection failure types.
    connection_backoff: Arc<parking_lot::Mutex<ConnectionBackoff>>,
    /// Per-contract backoff for contract-directed CONNECT attempts.
    contract_connect_backoff: Mutex<HashMap<ContractKey, ContractConnectState>>,
    /// Injectable time source used by `connection_maintenance`. Using `util::TimeSource`
    /// (which returns `tokio::time::Instant`) lets tests supply `SharedMockTimeSource` for
    /// fine-grained control without pausing the entire tokio runtime.
    pub(crate) time_source: Arc<dyn crate::util::time_source::TimeSource + Send + Sync>,
    /// Directory for persisting the peer address cache. When set, the peer cache
    /// is periodically saved here and loaded on startup for fast reconnection.
    pub(crate) peer_cache_dir: Option<std::path::PathBuf>,
}

// /// A data type that represents the fact that a peer has been blacklisted
// /// for some action. Has to be coupled with that action
// #[derive(Debug)]
// struct Blacklisted {
//     since: Instant,
//     peer: PeerKey,
// }

/// Guard that ensures `complete_subscription_request` is called even if the
/// subscription task panics. This prevents contracts from being stuck in
/// `pending_subscription_requests` forever.
pub(crate) struct SubscriptionRecoveryGuard {
    op_manager: Arc<OpManager>,
    contract_key: ContractKey,
    completed: bool,
}

impl SubscriptionRecoveryGuard {
    pub(crate) fn new(op_manager: Arc<OpManager>, contract_key: ContractKey) -> Self {
        Self {
            op_manager,
            contract_key,
            completed: false,
        }
    }

    pub(crate) fn complete(mut self, success: bool) {
        self.op_manager
            .ring
            .complete_subscription_request(&self.contract_key, success);
        self.completed = true;
    }
}

impl Drop for SubscriptionRecoveryGuard {
    fn drop(&mut self) {
        if !self.completed {
            // Task panicked or was cancelled before completion - treat as failure
            tracing::warn!(
                contract = %self.contract_key,
                "Subscription recovery task terminated unexpectedly, marking as failed"
            );
            self.op_manager
                .ring
                .complete_subscription_request(&self.contract_key, false);
        }
    }
}

/// State of the `Ring -> OpManager` weak back-reference, as observed by
/// the `connection_maintenance` loop.
///
/// The startup-vs-shutdown distinction is the whole point: a `None` slot
/// (not attached yet) must keep waiting, while a `Some(weak)` that no longer
/// upgrades (attached then dropped) must terminate the loop. Conflating the
/// two leaves the maintenance task spinning forever on a dead `Weak` —
/// the #3308 zombie.
enum OpManagerState<T> {
    /// `attach_op_manager` has not run yet — normal startup window. The
    /// maintenance loop should keep waiting for attachment.
    NotAttached,
    /// The `OpManager` is alive and was successfully upgraded.
    Live(Arc<T>),
    /// The back-reference was attached once but the owning `Arc` has since
    /// been dropped. Nothing re-attaches it, so the loop owner is dead and
    /// the maintenance task must terminate instead of spinning on a `Weak`
    /// that can never upgrade again (#3308).
    Detached,
}

/// Classify a `RwLock<Option<Weak<T>>>` back-reference into the three states
/// the maintenance loop cares about. Generic over `T` so it can be
/// unit-tested with a stand-in `Arc` instead of a full `OpManager` (#3308).
fn classify_op_manager_ref<T>(slot: &RwLock<Option<Weak<T>>>) -> OpManagerState<T> {
    let upgraded = slot.read().as_ref().map(|weak| weak.clone().upgrade());
    match upgraded {
        None => OpManagerState::NotAttached,
        Some(Some(strong)) => OpManagerState::Live(strong),
        Some(None) => OpManagerState::Detached,
    }
}

/// Result of pruning a connection.
#[derive(Debug, Default)]
pub struct PruneConnectionResult {
    /// Orphaned transactions that need to be retried or failed.
    pub orphaned_transactions: Vec<Transaction>,
    /// True if this prune caused us to drop below the readiness threshold.
    pub became_unready: bool,
}

impl Ring {
    pub const DEFAULT_MIN_CONNECTIONS: usize = 25;

    pub const DEFAULT_MAX_CONNECTIONS: usize = 200;

    const DEFAULT_MAX_UPSTREAM_BANDWIDTH: Rate = Rate::new_per_second(1_000_000.0);

    const DEFAULT_MAX_DOWNSTREAM_BANDWIDTH: Rate = Rate::new_per_second(1_000_000.0);

    /// Above this number of remaining hops, randomize which node a message which be forwarded to.
    const DEFAULT_RAND_WALK_ABOVE_HTL: usize = 7;

    /// Max hops to be performed for certain operations (e.g. propagating connection of a peer in the network).
    pub const DEFAULT_MAX_HOPS_TO_LIVE: usize = 10;

    pub fn new<ER: NetEventRegister + Clone>(
        config: &NodeConfig,
        event_loop_notifier: EventLoopNotificationsSender,
        event_register: ER,
        is_gateway: bool,
        connection_manager: ConnectionManager,
        task_monitor: &crate::node::background_task_monitor::BackgroundTaskMonitor,
    ) -> anyhow::Result<Arc<Self>> {
        let live_tx_tracker = LiveTransactionTracker::new();

        let max_hops_to_live = if let Some(v) = config.max_hops_to_live {
            v
        } else {
            Self::DEFAULT_MAX_HOPS_TO_LIVE
        };

        let router = Arc::new(RwLock::new(Router::new(&[])));
        crate::node::network_status::set_router(router.clone());
        task_monitor.register(
            "refresh_router",
            GlobalExecutor::spawn(Self::refresh_router(router.clone(), event_register.clone())),
        );

        // Interval for topology snapshot registration (1 second in test mode)
        // Registers subscription topology with the global registry for validation
        #[cfg(any(test, feature = "testing"))]
        const TOPOLOGY_SNAPSHOT_INTERVAL: Duration = Duration::from_secs(1);

        // Just initialize with a fake location, this will be later updated when the peer has an actual location assigned.
        let peer_cache_dir = if is_gateway {
            // Gateways don't need peer cache — peers connect to them.
            None
        } else {
            Some(config.config.data_dir())
        };
        let time_source: Arc<dyn crate::util::time_source::TimeSource + Send + Sync> =
            Arc::new(InstantTimeSrc::new());
        // Production always passes `None`, which resolves to
        // `GovernanceConfig::default()`. The override exists only so
        // simulation tests can inject compressed timescales + lower
        // `min_samples` to drive the rate-limit → MAD → evict → ban
        // chain within a paused-time sim. See issue #4301.
        let governance_config = config
            .governance_config_override
            .clone()
            .unwrap_or_default();
        let governance = Arc::new(crate::contract::governance::GovernanceManager::new(
            governance_config,
            time_source.clone(),
        ));
        let ring = Ring {
            max_hops_to_live,
            router,
            connection_manager,
            hosting_manager: hosting::HostingManager::new(config.config.max_hosting_storage),
            broken_invariants: BrokenInvariantsTracker::new(time_source.clone()),
            governance,
            update_rate_limiter: Arc::new(update_rate_limit::UpdateRateLimiter::new(
                time_source.clone(),
            )),
            contract_ban_list: Arc::new(contract_ban_list::ContractBanList::new(
                time_source.clone(),
            )),
            live_tx_tracker: live_tx_tracker.clone(),
            event_register: Box::new(event_register),
            op_manager: RwLock::new(None),
            is_gateway,
            connection_backoff: Arc::new(Mutex::new(ConnectionBackoff::new())),
            contract_connect_backoff: Mutex::new(HashMap::new()),
            time_source,
            peer_cache_dir,
        };

        if let Some(loc) = config.location {
            if config.own_addr.is_none() && is_gateway {
                return Err(anyhow::anyhow!("own_addr is required for gateways"));
            }
            ring.connection_manager.update_location(Some(loc));
        }

        let ring = Arc::new(ring);
        let current_span = tracing::Span::current();
        let span = if current_span.is_none() {
            tracing::info_span!("connection_maintenance")
        } else {
            tracing::info_span!(parent: current_span, "connection_maintenance")
        };

        task_monitor.register(
            "connection_maintenance",
            GlobalExecutor::spawn({
                let fut = ring
                    .clone()
                    .connection_maintenance(event_loop_notifier, live_tx_tracker)
                    .instrument(span);
                async move {
                    if let Err(e) = fut.await {
                        tracing::error!(error = %e, "connection_maintenance exited with error");
                    }
                }
            }),
        );

        // Spawn periodic subscription state telemetry task
        task_monitor.register(
            "emit_subscription_state_telemetry",
            GlobalExecutor::spawn(Self::emit_subscription_state_telemetry(
                ring.clone(),
                Self::SUBSCRIPTION_STATE_INTERVAL,
            )),
        );

        // Spawn periodic subscription recovery task to fix "orphaned hosters"
        // (peers that have contracts cached but aren't in the subscription tree)
        task_monitor.register(
            "recover_orphaned_subscriptions",
            GlobalExecutor::spawn(Self::recover_orphaned_subscriptions(
                ring.clone(),
                Self::SUBSCRIPTION_RECOVERY_INTERVAL,
            )),
        );

        // Spawn periodic GET subscription cache sweep task
        // Cleans up expired GET-triggered subscriptions to maintain bounded memory
        task_monitor.register(
            "sweep_get_subscription_cache",
            GlobalExecutor::spawn(Self::sweep_get_subscription_cache(
                ring.clone(),
                Self::GET_SUBSCRIPTION_SWEEP_INTERVAL,
            )),
        );

        // Spawn periodic topology snapshot registration task (test mode only)
        // This allows SimNetwork to validate subscription topology during tests
        #[cfg(any(test, feature = "testing"))]
        task_monitor.register(
            "register_topology_snapshots",
            GlobalExecutor::spawn(Self::register_topology_snapshots_periodically(
                ring.clone(),
                TOPOLOGY_SNAPSHOT_INTERVAL,
            )),
        );

        // Spawn periodic router model snapshot telemetry (every 5 minutes)
        task_monitor.register(
            "emit_router_snapshot_telemetry",
            GlobalExecutor::spawn(Self::emit_router_snapshot_telemetry(
                ring.clone(),
                Duration::from_secs(60 * 5),
            )),
        );

        // Spawn periodic contract-directed CONNECT task.
        // When a peer is a "subscription root" (closest to contract among neighbors),
        // it sends CONNECTs toward the contract's ring location to merge disconnected
        // subscription subtrees.
        const CONTRACT_CONNECT_INTERVAL: Duration = Duration::from_secs(30);
        task_monitor.register(
            "contract_directed_connects",
            GlobalExecutor::spawn(Self::contract_directed_connects(
                ring.clone(),
                CONTRACT_CONNECT_INTERVAL,
            )),
        );

        // Spawn periodic interest heartbeat task.
        // Sends full Interests { hashes } to each connected peer to keep
        // interest entries alive and prevent the death spiral where expired
        // entries block broadcast delivery.
        task_monitor.register(
            "interest_heartbeat",
            GlobalExecutor::spawn(Self::interest_heartbeat(ring.clone())),
        );

        // Spawn periodic governance reaper tick.
        // Computes per-contract state from accumulated cost/benefit
        // samples and logs `ReaperDecision`s. Mode defaults to DryRun
        // per the staged-rollout plan, so this task only ever logs —
        // no eviction actions until the operator (or release default)
        // flips to Enforce. The dashboard reads `governance.score_snapshot`
        // independently of the tick; the tick is only the engine that
        // updates state.
        task_monitor.register(
            "governance_reaper",
            GlobalExecutor::spawn(Self::governance_reaper_loop(ring.clone())),
        );

        Ok(ring)
    }

    pub fn attach_op_manager(&self, op_manager: &Arc<OpManager>) {
        self.op_manager.write().replace(Arc::downgrade(op_manager));
    }

    fn upgrade_op_manager(&self) -> Option<Arc<OpManager>> {
        self.op_manager
            .read()
            .as_ref()
            .and_then(|weak| weak.clone().upgrade())
    }

    /// Classify the current state of the `OpManager` back-reference.
    ///
    /// The maintenance loop must distinguish "not attached yet" (normal
    /// startup window, before [`Ring::attach_op_manager`] runs) from
    /// "attached then dropped" (the node has been torn down and the
    /// `Arc<OpManager>` is gone). The first case must keep waiting; the
    /// second must terminate the loop so it doesn't become a zombie
    /// spinning forever on a `Weak` that can never upgrade again (#3308).
    ///
    /// This is the sole production instantiation of the generic
    /// [`classify_op_manager_ref`] (with `T = OpManager`); the unit tests
    /// instantiate it with a stand-in `Arc` to exercise the same logic.
    fn op_manager_state(&self) -> OpManagerState<OpManager> {
        classify_op_manager_ref(&self.op_manager)
    }

    pub fn is_gateway(&self) -> bool {
        self.is_gateway
    }

    pub fn open_connections(&self) -> usize {
        self.connection_manager.connection_count()
    }

    /// Record a connection failure to the backoff tracker.
    pub fn record_connection_failure(&self, target: Location, reason: ConnectionFailureReason) {
        let mut backoff = self.connection_backoff.lock();
        backoff.record_failure_with_reason(target, reason);
    }

    /// Record a successful connection to clear backoff.
    pub fn record_connection_success(&self, target: Location) {
        let mut backoff = self.connection_backoff.lock();
        backoff.record_success(target);
    }

    /// Check if a target is currently in backoff.
    pub fn is_in_connection_backoff(&self, target: Location) -> bool {
        self.connection_backoff.lock().is_in_backoff(target)
    }

    /// Periodic cleanup of expired backoff entries.
    pub fn cleanup_connection_backoff(&self) {
        self.connection_backoff.lock().cleanup_expired();
    }

    /// Reset all connection backoff state. Used during isolation recovery
    /// when the node has had zero ring connections for an extended period.
    pub fn reset_all_connection_backoff(&self) {
        self.connection_backoff.lock().clear();
    }

    // ==================== Contract-Directed CONNECT ====================

    /// Initial backoff before retrying a contract-directed CONNECT.
    const INITIAL_CONTRACT_CONNECT_BACKOFF: Duration = Duration::from_secs(30);
    /// Maximum backoff cap for contract-directed CONNECTs (24 hours).
    const MAX_CONTRACT_CONNECT_BACKOFF: Duration = Duration::from_secs(24 * 60 * 60);
    /// Maximum contract-directed CONNECTs per cycle.
    const MAX_CONTRACT_CONNECTS_PER_CYCLE: usize = 2;

    /// Returns true if this peer is the closest to the contract among its connected neighbors
    /// (i.e., it's a subscription root for this contract).
    fn is_subscription_root(&self, contract_key: &ContractKey) -> bool {
        if !self.is_hosting_contract(contract_key) {
            return false;
        }
        let contract_location = Location::from(contract_key);
        let my_location = match self.connection_manager.own_location().location() {
            Some(loc) => loc,
            None => return false,
        };
        let my_distance = my_location.distance(contract_location);

        let connections = self.connection_manager.get_connections_by_location();
        for (_loc, conns) in connections.iter() {
            for conn in conns {
                if let Some(peer_loc) = conn.location.location() {
                    if peer_loc.distance(contract_location) < my_distance {
                        return false;
                    }
                }
            }
        }
        true
    }

    /// Check if a contract-directed CONNECT is currently in backoff.
    fn is_in_contract_connect_backoff(&self, contract_key: &ContractKey) -> bool {
        let backoff = self.contract_connect_backoff.lock();
        if let Some(state) = backoff.get(contract_key) {
            state.last_attempt.elapsed() < state.current_backoff
        } else {
            false
        }
    }

    /// Record a contract-directed CONNECT attempt. Doubles the backoff (up to cap).
    fn record_contract_connect_attempt(&self, contract_key: &ContractKey) {
        let mut backoff = self.contract_connect_backoff.lock();
        let now = self.time_source.now();
        let state = backoff
            .entry(*contract_key)
            .or_insert_with(|| ContractConnectState {
                current_backoff: Self::INITIAL_CONTRACT_CONNECT_BACKOFF,
                last_attempt: now,
            });
        state.last_attempt = now;
        state.current_backoff = (state.current_backoff * 2).min(Self::MAX_CONTRACT_CONNECT_BACKOFF);
    }

    /// Periodic task: when this peer is a subscription root for a contract,
    /// initiate a CONNECT toward the contract's ring location to merge
    /// disconnected subscription subtrees.
    async fn contract_directed_connects(ring: Arc<Self>, interval: Duration) {
        // Random initial delay to prevent thundering herd.
        let initial_delay = Duration::from_secs(GlobalRng::random_range(30u64..=60u64));
        tokio::time::sleep(initial_delay).await;

        let mut tick_interval = tokio::time::interval(interval);
        tick_interval.tick().await; // skip first immediate tick

        loop {
            tick_interval.tick().await;

            // Skip if we have too few connections to be meaningful.
            let conn_count = ring.connection_manager.connection_count();
            if conn_count < 2 {
                continue;
            }

            let contracts = ring.hosting_contract_keys();
            if contracts.is_empty() {
                continue;
            }

            let Some(op_manager) = ring.upgrade_op_manager() else {
                continue;
            };

            let mut connects_this_cycle = 0;

            for contract_key in &contracts {
                if connects_this_cycle >= Self::MAX_CONTRACT_CONNECTS_PER_CYCLE {
                    break;
                }

                if !ring.is_subscription_root(contract_key) {
                    // No longer root — a closer connection now exists.
                    // If we had backoff state, we previously sent a CONNECT as root.
                    // Now that a closer peer is connected, expire the subscription
                    // so it re-routes through the closer peer on the next renewal cycle.
                    let had_backoff = ring
                        .contract_connect_backoff
                        .lock()
                        .remove(contract_key)
                        .is_some();
                    if had_backoff {
                        ring.force_subscription_renewal(contract_key);
                        tracing::info!(
                            contract = %contract_key,
                            "No longer subscription root after contract-directed CONNECT; \
                             expired subscription to re-route through closer peer"
                        );
                    }
                    continue;
                }

                let contract_location = Location::from(contract_key);
                let my_location = ring.connection_manager.own_location().location();
                let my_distance = my_location.map(|l| l.distance(contract_location).as_f64());

                if ring.is_in_contract_connect_backoff(contract_key) {
                    continue;
                }

                // Emit telemetry for the root detection.
                crate::tracing::telemetry::send_standalone_event(
                    "subscription_root_detected",
                    serde_json::json!({
                        "contract": contract_key.to_string(),
                        "contract_location": contract_location.as_f64(),
                        "neighbor_count": conn_count,
                        "my_distance": my_distance,
                    }),
                );

                ring.record_contract_connect_attempt(contract_key);

                let backoff_secs = {
                    let b = ring.contract_connect_backoff.lock();
                    b.get(contract_key)
                        .map(|s| s.current_backoff.as_secs())
                        .unwrap_or(0)
                };

                tracing::info!(
                    contract = %contract_key,
                    %contract_location,
                    backoff_secs,
                    "Initiating contract-directed CONNECT as subscription root"
                );

                crate::tracing::telemetry::send_standalone_event(
                    "contract_directed_connect",
                    serde_json::json!({
                        "contract": contract_key.to_string(),
                        "contract_location": contract_location.as_f64(),
                        "my_distance": my_distance,
                        "backoff_secs": backoff_secs,
                    }),
                );

                let skip_list = HashSet::new();
                match ring
                    .acquire_new(
                        contract_location,
                        &skip_list,
                        &op_manager.to_event_listener,
                        &ring.live_tx_tracker,
                        &op_manager,
                    )
                    .await
                {
                    Ok(Some(tx)) => {
                        tracing::debug!(
                            %tx,
                            contract = %contract_key,
                            "Contract-directed CONNECT initiated"
                        );
                        connects_this_cycle += 1;
                    }
                    Ok(None) => {
                        tracing::debug!(
                            contract = %contract_key,
                            "Contract-directed CONNECT: no routing target found"
                        );
                    }
                    Err(e) => {
                        tracing::warn!(
                            contract = %contract_key,
                            error = %e,
                            "Contract-directed CONNECT failed"
                        );
                    }
                }
            }
        }
    }

    /// Periodic heartbeat: send `Interests { hashes }` to each connected peer.
    ///
    /// This prevents the death spiral where interest entries expire because no
    /// broadcasts are flowing, which in turn prevents broadcasts from ever
    /// flowing again. By periodically re-sending the full interest set, we
    /// Periodic governance reaper loop. Ticks every
    /// `GOVERNANCE_TICK_INTERVAL` and:
    ///
    ///   1. Calls `governance_tick(interval)` which snapshots live
    ///      benefit, applies cost decay, and runs the MAD detector
    ///      over the per-contract distribution.
    ///   2. Logs each `ReaperDecision` at INFO level with the from→to
    ///      state and reason. In `DryRun` mode (the default) this is
    ///      the only effect; in `Enforce` mode the caller would also
    ///      emit `EvictContract` events here. Enforce-mode wiring
    ///      lands in a subsequent commit.
    ///   3. Logs the network-norms summary (median, MAD, threshold,
    ///      sample size) at DEBUG so a busy node doesn't spam INFO.
    ///
    /// **Cancellation safety:** the loop's only `.await` points are
    /// `tokio::time::sleep` / `interval.tick()`. No channel sends, no
    /// long-running operations — the design-doc rule that "the
    /// dashboard reflects the back-end" is respected here too: the
    /// reaper writes governance state; readers consume it
    /// asynchronously.
    async fn governance_reaper_loop(ring: Arc<Self>) {
        // Initial delay so a fleet-wide restart doesn't have every node
        // ticking in lockstep. Derived from the node's own ring location
        // (already random + node-distinct) rather than from `GlobalRng`,
        // because consuming from `GlobalRng` at startup shifts the
        // deterministic seed state and breaks simulation tests whose
        // routing/topology decisions are RNG-driven (the
        // `test_get_routing_coverage_low_htl` regression was caught by
        // CI on PR #4270). 30-90 second offset; the tick itself runs
        // every minute below.
        let own_loc = ring
            .connection_manager
            .own_location()
            .location()
            .map(|l| l.as_f64())
            .unwrap_or(0.5);
        let initial_delay_secs = 30 + ((own_loc * 60.0) as u64);
        let initial_delay = Duration::from_secs(initial_delay_secs);
        tokio::time::sleep(initial_delay).await;

        let mut interval = tokio::time::interval(GOVERNANCE_TICK_INTERVAL);
        interval.tick().await; // Skip first immediate tick.
        let mut last_tick = ring.time_source.now();

        loop {
            interval.tick().await;
            let now = ring.time_source.now();
            let elapsed = now.saturating_duration_since(last_tick);
            last_tick = now;

            let result = ring.governance_tick(elapsed);

            // Cycle the UPDATE rate limiter's idle-entry cleanup on
            // the same cadence. Keeps the `(sender, contract)` map
            // bounded; CLEANUP_AGE is 5 minutes so this drops entries
            // for pairs that haven't tried an UPDATE since then.
            ring.update_rate_limiter.cleanup();

            // Phase 7 ban-list maintenance. Defense-in-depth — the
            // BanLifted decisions below explicitly unban, but if any
            // entry slipped through the reaper (e.g. mode flipped off
            // mid-window), cleanup catches it on the next tick.
            ring.contract_ban_list.cleanup();

            // Sweep expired broken-invariant flags on the same cadence so a
            // suppressed contract (especially a false-positive) recovers
            // after BROKEN_INVARIANT_TTL instead of being bricked forever.
            ring.broken_invariants.cleanup();

            // Network-norms summary at DEBUG — useful when debugging
            // calibration but too noisy for INFO on a healthy node.
            tracing::debug!(
                median = ?result.median_log_ratio,
                mad = ?result.mad,
                threshold = ?result.threshold,
                sample_size = result.sample_size,
                capacity_ceiling_binding = result.capacity_ceiling_binding,
                skip_reason = ?result.skip_reason,
                "governance reaper tick",
            );

            // Phase 7 enforcement wiring: BanTriggered → ban list;
            // BanLifted → unban. Done AFTER the cleanup() so a
            // freshly-added entry (with expiry = now + ban_ttl) is
            // not immediately swept by the same tick's cleanup. The
            // two calls are sequential in this task — no concurrent
            // race — the ordering is purely to defend the
            // not-quite-expired-yet invariant.
            Self::apply_ban_decisions(
                &ring.contract_ban_list,
                &result.decisions,
                ring.time_source.now() + ring.governance.ban_ttl(),
            );

            // Decisions at INFO — these are the dashboard-relevant
            // events. Empty in healthy state, surfaces every
            // transition during incidents.
            for decision in result.decisions {
                tracing::info!(
                    contract = %decision.key,
                    from = ?decision.from,
                    to = ?decision.to,
                    reason = ?decision.reason,
                    actionable = decision.actionable,
                    "governance state transition",
                );
            }
        }
    }

    /// Translate a batch of governance decisions into ban-list
    /// mutations. Pulled out of the reaper loop so the wiring is
    /// directly testable: a missing or reversed branch here would
    /// silently break Phase 7 enforcement even with the
    /// `GovernanceManager` emitting correct transitions.
    ///
    /// **`actionable` filter:** non-actionable decisions (DryRun
    /// mode) are skipped. Today the `GovernanceManager` only emits
    /// `BanTriggered` in Enforce mode so this is a defense-in-depth
    /// guard — but a future "shadow" mode that wants to surface
    /// would-have-banned transitions in the dashboard must not
    /// silently begin enforcing them through this wiring.
    pub(crate) fn apply_ban_decisions(
        ban_list: &contract_ban_list::ContractBanList,
        decisions: &[crate::contract::governance::ReaperDecision],
        ban_expiry: tokio::time::Instant,
    ) {
        use crate::contract::governance::TransitionReason;
        for decision in decisions {
            if !decision.actionable {
                continue;
            }
            #[allow(clippy::wildcard_enum_match_arm)]
            match decision.reason {
                TransitionReason::BanTriggered => {
                    ban_list.ban(
                        decision.key,
                        ban_expiry,
                        contract_ban_list::BanReason::AutoMad,
                    );
                }
                TransitionReason::BanLifted => {
                    ban_list.unban(&decision.key);
                }
                _ => {}
            }
        }
    }

    /// keep entries alive on the remote side.
    ///
    /// Sends are spread evenly across the interval to avoid bursts.
    async fn interest_heartbeat(ring: Arc<Self>) {
        use crate::ring::interest::INTEREST_HEARTBEAT_INTERVAL;

        // Random initial delay to prevent synchronized heartbeats across peers
        let initial_delay = Duration::from_secs(GlobalRng::random_range(15u64..=45u64));
        tokio::time::sleep(initial_delay).await;

        let mut interval = tokio::time::interval(INTEREST_HEARTBEAT_INTERVAL);
        interval.tick().await; // Skip first immediate tick

        loop {
            interval.tick().await;

            let Some(op_manager) = ring.upgrade_op_manager() else {
                continue;
            };

            // Get our current interest hashes
            let hashes = op_manager.interest_manager.get_all_interest_hashes();
            if hashes.is_empty() {
                continue;
            }

            // Get all connected peer addresses (deduplicated)
            let connections = ring.connection_manager.get_connections_by_location();
            let peer_addrs: Vec<std::net::SocketAddr> = {
                let mut seen = HashSet::new();
                connections
                    .values()
                    .flat_map(|conns| conns.iter())
                    .filter_map(|conn| conn.location.socket_addr())
                    .filter(|addr| seen.insert(*addr))
                    .collect()
            };

            if peer_addrs.is_empty() {
                continue;
            }

            let num_peers = peer_addrs.len();
            // num_peers >= 1 guaranteed by the is_empty() check above
            let spread_delay = INTEREST_HEARTBEAT_INTERVAL / num_peers as u32;

            tracing::debug!(
                num_peers,
                num_hashes = hashes.len(),
                "Interest heartbeat: sending Interests to peers"
            );

            let sender = op_manager.to_event_listener.notifications_sender();
            let mut peers_sent = 0usize;
            for (i, peer_addr) in peer_addrs.into_iter().enumerate() {
                let message = crate::message::InterestMessage::Interests {
                    hashes: hashes.clone(),
                };
                if let Err(e) = sender
                    .send(either::Either::Right(
                        crate::message::NodeEvent::SendInterestMessage {
                            target: peer_addr,
                            message,
                        },
                    ))
                    .await
                {
                    // Channel send failure means the receiver is dropped (node
                    // shutting down). No point sending to remaining peers.
                    tracing::debug!(
                        peer = %peer_addr,
                        error = %e,
                        "Interest heartbeat: failed to queue message"
                    );
                    break;
                }
                peers_sent += 1;

                // Spread sends evenly across the interval (skip delay after last)
                if i + 1 < num_peers {
                    tokio::time::sleep(spread_delay).await;
                }
            }

            crate::tracing::telemetry::send_standalone_event(
                "interest_heartbeat_cycle",
                serde_json::json!({
                    "peers_sent": peers_sent,
                    "interest_hashes": hashes.len(),
                }),
            );
        }
    }

    /// Register events with the event system.
    /// This is used by operations to emit failure and other events.
    pub async fn register_events<'a>(
        &self,
        events: either::Either<
            crate::tracing::NetEventLog<'a>,
            Vec<crate::tracing::NetEventLog<'a>>,
        >,
    ) {
        self.event_register.register_events(events).await;
    }

    /// Maximum number of route events to load from the AOF event log on startup
    /// and during periodic refresh. Caps memory use while retaining enough history
    /// for the isotonic estimators to converge.
    const ROUTER_HISTORY_LIMIT: usize = 10_000;

    async fn refresh_router<ER: NetEventRegister>(router: Arc<RwLock<Router>>, register: ER) {
        // Load routing history immediately on startup so the router doesn't
        // start cold — without this, peers route suboptimally for ~5 minutes
        // until the first periodic refresh.
        match register.get_router_events(Self::ROUTER_HISTORY_LIMIT).await {
            Ok(history) if !history.is_empty() => {
                tracing::info!(
                    events = history.len(),
                    "Restored routing history from event log"
                );
                *router.write() = Router::new(&history);
            }
            Ok(_) => {
                tracing::debug!("No routing history to restore on startup");
            }
            Err(error) => {
                tracing::warn!(%error, "Failed to load routing history on startup, starting cold");
            }
        }

        let mut interval = tokio::time::interval(Duration::from_secs(60 * 5));
        interval.tick().await;
        loop {
            interval.tick().await;
            let history = match register.get_router_events(Self::ROUTER_HISTORY_LIMIT).await {
                Ok(h) => h,
                Err(error) => {
                    // get_router_events errors are transient and recoverable —
                    // e.g. file-descriptor exhaustion ("os error 24") when the
                    // AOF segment file can't be opened during a connection-churn
                    // spike. Router refresh is a best-effort optimization, so we
                    // keep the existing in-memory router in place and retry on
                    // the next interval rather than killing the node. (A genuine
                    // panic inside this task is still caught by the
                    // BackgroundTaskMonitor — only this expected, transient read
                    // failure is swallowed here.)
                    tracing::error!(
                        error = %error,
                        "Failed to refresh routing history from event log; \
                         keeping existing router and retrying on next interval"
                    );
                    continue;
                }
            };
            if !history.is_empty() {
                *router.write() = Router::new(&history);
            }
        }
    }

    /// Periodically emit a router model snapshot as an EventKind::RouterSnapshot event.
    ///
    /// This captures the isotonic regression curves and model state, including the
    /// connect forward estimator if available via OpManager.
    async fn emit_router_snapshot_telemetry(ring: Arc<Self>, interval_duration: Duration) {
        let mut interval = tokio::time::interval(interval_duration);
        // Skip the first immediate tick
        interval.tick().await;

        loop {
            interval.tick().await;

            let mut snapshot = ring.router.read().snapshot();

            // Try to include connect forward estimator data.
            // Drop the read guard immediately after `snapshot()` so unrelated
            // consumers don't queue behind us for the four field assignments
            // (clippy: `significant_drop_tightening`).
            if let Some(op_manager) = ring.upgrade_op_manager() {
                let (curve, data_range, events, adjustments) =
                    op_manager.connect_forward_estimator.read().snapshot();
                snapshot.connect_forward_curve = Some(curve);
                snapshot.connect_forward_data_range = Some(data_range);
                snapshot.connect_forward_events = Some(events);
                snapshot.connect_forward_peer_adjustments = Some(adjustments);
            }

            // Node-health gauges (#4440): make fd-exhaustion headroom observable
            // on the existing snapshot cadence. Best-effort — `None` where the
            // platform can't report it (see `read_fd_usage`).
            let (open_fds, fd_soft_limit) = read_fd_usage();
            snapshot.open_fds = open_fds;
            snapshot.fd_soft_limit = fd_soft_limit;

            // Compiled-WASM module-cache occupancy + eviction gauges (#4440),
            // read from the process-global the caches publish into (they live
            // behind the contract-handler channel, unreachable from here).
            let mc = crate::wasm_runtime::MODULE_CACHE_METRICS.snapshot();
            snapshot.contract_module_cache_entries = Some(mc.contract_entries);
            snapshot.contract_module_cache_total_bytes = Some(mc.contract_total_bytes);
            snapshot.contract_module_cache_budget_bytes = Some(mc.contract_budget_bytes);
            snapshot.contract_module_cache_evictions_total = Some(mc.contract_evictions_total);
            snapshot.delegate_module_cache_entries = Some(mc.delegate_entries);
            snapshot.delegate_module_cache_total_bytes = Some(mc.delegate_total_bytes);
            snapshot.delegate_module_cache_budget_bytes = Some(mc.delegate_budget_bytes);
            snapshot.delegate_module_cache_evictions_total = Some(mc.delegate_evictions_total);

            tracing::info!(
                failure_events = snapshot.failure_events,
                success_events = snapshot.success_events,
                prediction_active = snapshot.prediction_active,
                consider_n_closest_peers = snapshot.consider_n_closest_peers,
                open_fds = ?snapshot.open_fds,
                fd_soft_limit = ?snapshot.fd_soft_limit,
                contract_module_cache_entries = mc.contract_entries,
                contract_module_cache_total_bytes = mc.contract_total_bytes,
                contract_module_cache_evictions_total = mc.contract_evictions_total,
                "router_snapshot"
            );

            if let Some(event) = NetEventLog::router_snapshot(&ring, snapshot) {
                ring.event_register
                    .register_events(Either::Left(event))
                    .await;
            }
        }
    }

    /// Periodically emit subscription_state telemetry events for all active subscriptions.
    ///
    /// This enables the telemetry dashboard to reconstruct historical subscription trees
    /// and show accurate subscription state at any point in time.
    async fn emit_subscription_state_telemetry(ring: Arc<Self>, interval_duration: Duration) {
        let mut interval = tokio::time::interval(interval_duration);
        // Skip the first immediate tick
        interval.tick().await;

        loop {
            interval.tick().await;

            // Get subscription states from the new lease-based model
            let subscription_states = ring.get_subscription_states();

            if subscription_states.is_empty() {
                continue;
            }

            tracing::debug!(
                subscription_count = subscription_states.len(),
                "Emitting periodic subscription state telemetry"
            );

            // Log subscription states (simplified - no upstream/downstream in new model)
            for (key, has_client, is_active, _expires_at) in subscription_states {
                tracing::trace!(
                    %key,
                    has_client_subscription = has_client,
                    is_active_subscription = is_active,
                    "Subscription state"
                );
            }
        }
    }

    /// Maximum renewal tasks spawned per tick.
    ///
    /// At 30s intervals, this yields up to 20 renewals/minute. Since
    /// `contracts_needing_renewal()` returns contracts expiring within the
    /// 2-minute renewal window, this effectively handles ~40 concurrent
    /// subscriptions before renewals can't keep up. The mid-cycle channel
    /// capacity check (RENEWAL_STOP_CAPACITY_FRACTION) provides backpressure
    /// if the network can't absorb this many.
    const MAX_RECOVERY_ATTEMPTS_PER_INTERVAL: usize = 10;

    /// Skip renewal cycle when channel remaining capacity falls below this
    /// fraction of max (i.e. channel is more than 50% full).
    const RENEWAL_DEFER_CAPACITY_FRACTION: usize = 2; // channel_max / 2

    /// Stop spawning mid-cycle when remaining capacity falls below this
    /// fraction of max (i.e. channel is more than 75% full).
    const RENEWAL_STOP_CAPACITY_FRACTION: usize = 4; // channel_max / 4

    /// Interval for periodic subscription state telemetry snapshots.
    pub(crate) const SUBSCRIPTION_STATE_INTERVAL: Duration = Duration::from_secs(60);

    /// Interval for periodic subscription recovery attempts.
    ///
    /// This recovers "orphaned hosters" - peers that have contracts in cache
    /// but failed to establish subscription (no upstream in subscription tree).
    pub(crate) const SUBSCRIPTION_RECOVERY_INTERVAL: Duration = Duration::from_secs(30);

    /// Interval for periodic GET subscription cache sweep.
    pub(crate) const GET_SUBSCRIPTION_SWEEP_INTERVAL: Duration = Duration::from_secs(60);

    /// Periodically attempt to recover "orphaned hosters" - contracts we're hosting
    /// but don't have an upstream subscription for.
    ///
    /// This can happen when:
    /// - The initial subscription after GET/PUT failed (network issues, timeout)
    /// - Our upstream peer disconnected and we haven't found a new one
    /// - A race condition left us hosting without subscription
    ///
    /// The task respects existing backoff mechanisms to avoid subscription spam.
    ///
    /// **Connection gating (#3676):** This task skips renewal cycles entirely when
    /// the node has zero ring connections. Without this gate, disconnected peers
    /// generate a subscribe retry storm — thousands of subscribe requests per cycle
    /// that all fail immediately because there's no one to send them to. Telemetry
    /// showed 3 peers with 0 connections generating 96% of all subscribe traffic.
    async fn recover_orphaned_subscriptions(ring: Arc<Self>, interval_duration: Duration) {
        // Wait indefinitely for the first ring connection before starting
        // subscription recovery. The per-cycle connection check below is the
        // real gate; this just avoids running the loop body with no peers.
        let mut wait_logged = false;
        loop {
            tokio::time::sleep(Duration::from_millis(500)).await;
            if ring.open_connections() > 0 {
                tracing::info!(
                    hosted_contracts = ring.hosting_contract_keys().len(),
                    "Ring connection established, starting subscription recovery"
                );
                break;
            }
            // Log periodically so operators can diagnose stuck nodes.
            if !wait_logged {
                wait_logged = true;
                tracing::info!(
                    hosted_contracts = ring.hosting_contract_keys().len(),
                    "Waiting for ring connection before starting subscription recovery"
                );
            }
        }

        // Small jitter (2-5s) after first connection to let the ring stabilize
        // slightly before flooding with subscribe requests.
        let jitter = Duration::from_secs(GlobalRng::random_range(2u64..=5u64));
        tokio::time::sleep(jitter).await;

        let mut interval = tokio::time::interval(interval_duration);
        // Skip the first immediate tick — we run the first pass immediately
        // below (no tick wait) so client subscriptions get prompt renewal.
        interval.tick().await;

        let mut first_pass = true;

        loop {
            if first_pass {
                first_pass = false;
            } else {
                interval.tick().await;
            }

            // Always run expiry sweeps, even when disconnected. Stale
            // subscriptions and downstream subscribers must be cleaned up
            // to keep interest manager counts accurate. Only the renewal
            // spawning (below) is gated on having connections.
            //
            // First, expire any stale subscriptions
            let expired = ring.expire_stale_subscriptions();
            if !expired.is_empty() {
                tracing::debug!(
                    expired_count = expired.len(),
                    "Expired {} stale subscriptions",
                    expired.len()
                );
            }

            // Expire stale downstream subscribers and decrement interest manager
            let ds_expired = ring.expire_stale_downstream_subscribers();
            if !ds_expired.is_empty() {
                tracing::debug!(
                    expired_count = ds_expired.len(),
                    "Expired stale downstream subscribers"
                );

                if let Some(op_manager) = ring.upgrade_op_manager() {
                    for (contract, expired_count) in &ds_expired {
                        // Decrement interest manager for each expired peer
                        for _ in 0..*expired_count {
                            op_manager
                                .interest_manager
                                .remove_downstream_subscriber(contract);
                        }

                        // Send Unsubscribe upstream if no remaining interest
                        if ring.should_unsubscribe_upstream(contract) {
                            let op_mgr = op_manager.clone();
                            let contract = *contract;
                            GlobalExecutor::spawn(async move {
                                op_mgr.send_unsubscribe_upstream(&contract).await;
                            });
                        }
                    }
                }
            }

            // Gate: skip renewal spawning if we have no ring connections (#3676).
            // Subscribe requests require connected peers to route through.
            // Without this, disconnected peers flood the notification channel
            // with doomed subscribe requests every 30 seconds.
            if ring.open_connections() == 0 {
                tracing::debug!("Skipping subscription renewal: no ring connections");
                continue;
            }

            // Get contracts that need subscription renewal (have client subscriptions)
            let mut contracts_needing_renewal = ring.contracts_needing_renewal();

            if contracts_needing_renewal.is_empty() {
                tracing::debug!(
                    hosted = ring.hosting_contract_keys().len(),
                    "No contracts needing subscription renewal"
                );
                continue;
            }

            tracing::info!(
                needing_renewal = contracts_needing_renewal.len(),
                hosted = ring.hosting_contract_keys().len(),
                "Starting subscription renewal cycle"
            );

            // Shuffle to prevent starvation: without this, the same failing contracts
            // (first N in iteration order) would always be tried first, blocking later
            // contracts from ever being attempted when they hit the batch limit.
            GlobalRng::shuffle(&mut contracts_needing_renewal);

            // Get op_manager to spawn subscription requests
            let Some(op_manager) = ring.upgrade_op_manager() else {
                tracing::debug!("OpManager not available for subscription renewal");
                continue;
            };

            // Backpressure: reduce batch size when the notification channel is
            // congested, but never skip entirely. Renewals are critical-path —
            // skipping a full cycle when the channel is busy lets subscriptions
            // expire, which causes cascading failures as the subscription tree
            // thins out and remaining renewals take longer paths.
            let sender = op_manager.to_event_listener.notifications_sender();
            let channel_remaining = sender.capacity();
            let channel_max = sender.max_capacity();

            let batch_limit =
                if channel_remaining < channel_max / Self::RENEWAL_DEFER_CAPACITY_FRACTION {
                    // Channel >50% full: allow a reduced batch (quarter of normal)
                    // so critical renewals still get through. Always attempt at least 1.
                    let reduced = (Self::MAX_RECOVERY_ATTEMPTS_PER_INTERVAL / 4).max(1);
                    tracing::warn!(
                        channel_remaining,
                        channel_max,
                        batch_limit = reduced,
                        contracts = contracts_needing_renewal.len(),
                        "Notification channel >50% full, reducing renewal batch size"
                    );
                    reduced
                } else {
                    Self::MAX_RECOVERY_ATTEMPTS_PER_INTERVAL
                };

            let mut attempted = 0;
            let mut skipped = 0;

            for contract in contracts_needing_renewal {
                // Limit concurrent renewal attempts to avoid overwhelming the network
                if attempted >= batch_limit {
                    tracing::debug!(
                        limit = batch_limit,
                        "Reached max renewal attempts for this interval, remaining will be tried next cycle"
                    );
                    break;
                }

                // Stop early if the channel is filling from our own spawns.
                let remaining_now = sender.capacity();
                if remaining_now < channel_max / Self::RENEWAL_STOP_CAPACITY_FRACTION {
                    tracing::warn!(
                        channel_remaining = remaining_now,
                        attempted,
                        "Notification channel >75% full during renewal spawning, stopping early"
                    );
                    break;
                }

                // Phase 7 egress gate (#4373). Don't renew a subscription
                // for a contract we have banned: a renewal re-registers
                // interest via the same outbound-SUBSCRIBE machinery as a
                // client-initiated request, but unlike the four
                // `start_client_*` originator entry points this scheduler
                // doesn't pass through `reject_if_contract_banned`. Without
                // this check the node keeps emitting outbound SUBSCRIBE
                // renewals for a banned-but-still-subscribed contract on
                // every maintenance cycle until the ban TTL lifts. Checked
                // before `can_request_subscription` so a banned contract is
                // skipped regardless of its spam-backoff state.
                if ring.contract_ban_list.is_banned(contract.id()) {
                    tracing::debug!(
                        %contract,
                        phase = "subscription_renewal_banned_skip",
                        "skipping subscription renewal for banned contract"
                    );
                    skipped += 1;
                    continue;
                }

                // Check spam prevention (respects exponential backoff and pending checks)
                if !ring.can_request_subscription(&contract) {
                    skipped += 1;
                    continue;
                }

                // Mark as pending and spawn subscription request
                if ring.mark_subscription_pending(contract) {
                    attempted += 1;

                    // Spread tasks across the interval to avoid thundering-herd bursts.
                    let jitter_ms = GlobalRng::random_range(0u64..=15_000);

                    let op_manager_clone = op_manager.clone();
                    let contract_key = contract;

                    GlobalExecutor::spawn(async move {
                        tokio::time::sleep(Duration::from_millis(jitter_ms)).await;
                        // Guard ensures complete_subscription_request is called even on panic
                        let guard =
                            SubscriptionRecoveryGuard::new(op_manager_clone.clone(), contract_key);

                        let instance_id = *contract_key.id();
                        // Renewal driver: same machinery as
                        // client-initiated SUBSCRIBE, with delivery
                        // returned to this task instead of via
                        // `result_router_tx`. `is_renewal=true` so
                        // the responder skips sending state.
                        //
                        // Outer renewal-cycle timeout: cap each renewal task at
                        // a fraction of the renewal interval so a stuck driver
                        // (slow peer, blocked op_execution_sender) self-cancels
                        // within one cycle instead of leaking past
                        // `mark_subscription_pending` semantics. The driver's
                        // per-attempt `OPERATION_TTL` (60 s) bounds individual
                        // peer waits; this outer timeout bounds total task
                        // lifetime. Without it, sustained slow-peer load could
                        // accumulate background renewal tasks faster than they
                        // complete (90+ contracts × multi-attempt ×
                        // OPERATION_TTL on a gateway).
                        let renewal_tx = crate::message::Transaction::new::<
                            crate::operations::subscribe::SubscribeMsg,
                        >();
                        let renewal_deadline = Self::SUBSCRIPTION_RECOVERY_INTERVAL
                            .saturating_sub(Duration::from_secs(5));
                        let outcome_enum = match tokio::time::timeout(
                            renewal_deadline,
                            crate::operations::subscribe::run_renewal_subscribe(
                                op_manager_clone.clone(),
                                instance_id,
                                renewal_tx,
                            ),
                        )
                        .await
                        {
                            Ok(outcome) => outcome,
                            Err(_) => crate::operations::subscribe::RenewalOutcome::Failed {
                                reason: format!(
                                    "renewal task exceeded {}s cycle deadline",
                                    renewal_deadline.as_secs()
                                ),
                            },
                        };

                        let (outcome, error_msg) = match outcome_enum {
                            crate::operations::subscribe::RenewalOutcome::Success => {
                                tracing::info!(
                                    %contract_key,
                                    "Subscription renewal succeeded"
                                );
                                guard.complete(true);
                                ("success", None)
                            }
                            crate::operations::subscribe::RenewalOutcome::ChannelCongestion => {
                                // Channel congestion is a local resource issue, not a
                                // protocol failure. Don't penalize with backoff — just
                                // clear the pending mark so the contract is eligible on
                                // the next cycle.
                                tracing::warn!(
                                    %contract_key,
                                    "Subscription renewal skipped (channel full), will retry next cycle"
                                );
                                guard.complete(true);
                                ("dropped_channel_full", None)
                            }
                            crate::operations::subscribe::RenewalOutcome::Failed { reason } => {
                                tracing::debug!(
                                    %contract_key,
                                    error = %reason,
                                    "Subscription renewal failed (will retry with backoff)"
                                );
                                guard.complete(false);
                                ("failed", Some(reason))
                            }
                        };

                        crate::tracing::telemetry::send_standalone_event(
                            "subscription_renewal_outcome",
                            serde_json::json!({
                                "contract": contract_key.to_string(),
                                "outcome": outcome,
                                "error": error_msg,
                            }),
                        );
                    });
                }
            }

            if attempted > 0 || skipped > 0 {
                tracing::info!(
                    attempted,
                    skipped_rate_limited = skipped,
                    "Subscription renewal cycle complete"
                );
            }
        }
    }

    /// Background task to sweep expired entries from the GET subscription cache.
    ///
    /// When contracts are evicted (past max entries and beyond TTL), this task
    /// cleans up the local subscription state. The upstream peer will eventually
    /// prune us when updates fail to deliver.
    async fn sweep_get_subscription_cache(ring: Arc<Self>, interval_duration: Duration) {
        // Add random initial delay to prevent synchronized sweeps across peers
        let initial_delay = Duration::from_secs(GlobalRng::random_range(10u64..=30u64));
        tokio::time::sleep(initial_delay).await;

        let mut interval = tokio::time::interval(interval_duration);
        interval.tick().await; // Skip first immediate tick

        loop {
            interval.tick().await;

            // Sweep expired entries from GET subscription cache
            let expired = ring.sweep_expired_get_subscriptions();

            // Do NOT early-continue when `expired` is empty: pending
            // reclamation retries below must run every cycle, otherwise
            // entries queued by the in-use / queue-full skip points stay
            // leaked indefinitely whenever the cache is under budget
            // (the common case). See Codex r10 P2.
            if !expired.is_empty() {
                tracing::debug!(
                    expired_count = expired.len(),
                    "GET subscription cache sweep found expired entries"
                );
            }

            // Reclaim on-disk storage for the expired contracts so the hosting
            // budget is a real disk bound. The sweep task only holds an
            // `Arc<Ring>`; reach the `OpManager` the same way
            // `recover_orphaned_subscriptions` does, via the weak back-reference.
            // `reclaim_evicted_contract` re-checks the subscription gate per
            // key before emitting the eviction event.
            let op_manager = ring.upgrade_op_manager();
            if op_manager.is_none() {
                // The weak back-reference is dropped only during node shutdown;
                // surface the skipped reclamation so an unexpected `None` (and
                // the resulting on-disk leak for this cycle) is observable.
                tracing::debug!(
                    expired_count = expired.len(),
                    "OpManager unavailable during GET subscription sweep — \
                     on-disk reclamation skipped for the expired contracts this cycle"
                );
            }

            // Clean up local subscription state for each expired contract.
            // Note: contracts with client subscriptions are protected from eviction
            // by the should_retain predicate in sweep_expired_hosting().
            // The `expected_generation` snapshot is captured atomically with
            // the eviction decision in `HostingCache::record_access` /
            // `sweep_expired`; it is re-checked at deletion time by
            // `RuntimePool::remove_contract` to close the re-host race.
            for (key, expected_generation) in expired {
                ring.unsubscribe(&key);
                tracing::info!(
                    %key,
                    "Cleaned up expired hosting subscription from local state"
                );
                if let Some(op_manager) = &op_manager {
                    crate::operations::reclaim_evicted_contract(
                        op_manager,
                        key,
                        expected_generation,
                    );
                }
            }

            // Retry pending reclamations queued by the two skip points
            // (fair-queue rejection of `EvictContract` and the
            // `contract_in_use` skip in `RuntimePool::remove_contract`).
            // The snapshot iterates without holding the DashMap shard
            // guards. Each retry routes through
            // `reclaim_evicted_contract`, which re-checks
            // `contract_in_use` — entries that are still in use stay in
            // the queue (no event emitted) and will be retried next
            // cycle. Successful reclamations clear their pending entry
            // in `RuntimePool::remove_contract`.
            if let Some(op_manager) = &op_manager {
                let pending = ring.pending_reclamation_snapshot();
                if !pending.is_empty() {
                    tracing::debug!(
                        pending_count = pending.len(),
                        "Retrying pending reclamations from previous skipped \
                         `EvictContract` events"
                    );
                    for (key, expected_generation) in pending {
                        crate::operations::reclaim_evicted_contract(
                            op_manager,
                            key,
                            expected_generation,
                        );
                    }
                }
            }
        }
    }

    /// Periodically register topology snapshots for simulation testing.
    ///
    /// This task only runs when `CURRENT_NETWORK_NAME` is set (i.e., during SimNetwork tests).
    /// It allows SimNetwork to validate subscription topology by querying the global registry.
    #[cfg(any(test, feature = "testing"))]
    async fn register_topology_snapshots_periodically(
        ring: Arc<Self>,
        interval_duration: Duration,
    ) {
        use topology_registry::{get_current_network_name, register_topology_snapshot};

        tracing::info!("Topology snapshot registration task started");

        // Add small initial delay to let network stabilize (use short delay in tests)
        tokio::time::sleep(Duration::from_millis(100)).await;

        let mut interval = tokio::time::interval(interval_duration);
        interval.tick().await; // Skip first immediate tick

        loop {
            interval.tick().await;

            // Only register if we're in a simulation context
            let Some(network_name) = get_current_network_name() else {
                tracing::debug!("Topology snapshot: no network name set, skipping");
                continue;
            };

            let Some(peer_addr) = ring.connection_manager.get_own_addr() else {
                tracing::debug!("Topology snapshot: no peer address yet, skipping");
                continue;
            };

            // Use get_stored_location() for consistency with set_upstream distance check.
            // This ensures topology validation uses the same location as the tie-breaker.
            let location = ring
                .connection_manager
                .get_stored_location()
                .map(|l| l.as_f64())
                .unwrap_or(0.0);

            let snapshot = ring
                .hosting_manager
                .generate_topology_snapshot(peer_addr, location);
            let contract_count = snapshot.contracts.len();
            register_topology_snapshot(&network_name, snapshot);

            tracing::info!(
                %peer_addr,
                location,
                network = %network_name,
                contract_count,
                "Registered topology snapshot"
            );
        }
    }

    /// Record an access to a contract in the hosting cache.
    ///
    /// This adds or refreshes the contract in the unified hosting cache.
    /// ALL contracts in the hosting cache get subscription renewal.
    ///
    /// Returns a `RecordAccessResult` containing:
    /// - `is_new`: Whether this contract was newly added (vs. refreshed existing)
    /// - `evicted`: Contracts that were evicted to make room
    pub fn host_contract(
        &self,
        key: ContractKey,
        size_bytes: u64,
        access_type: AccessType,
    ) -> RecordAccessResult {
        self.hosting_manager
            .record_contract_access(key, size_bytes, access_type)
    }

    /// Record a GET access to a contract in the hosting cache.
    ///
    /// Returns a `RecordAccessResult` indicating whether this was a new addition
    /// and which contracts were evicted (if any).
    pub fn record_get_access(&self, key: ContractKey, size_bytes: u64) -> RecordAccessResult {
        self.host_contract(key, size_bytes, AccessType::Get)
    }

    /// Whether this node is hosting this contract (has it in cache).
    #[inline]
    pub fn is_hosting_contract(&self, key: &ContractKey) -> bool {
        self.hosting_manager.is_hosting_contract(key)
    }

    /// Set the storage reference for hosting metadata persistence.
    ///
    /// Must be called after executor creation. This enables automatic
    /// cleanup of persisted metadata when contracts are evicted.
    pub fn set_hosting_storage(&self, storage: crate::contract::storages::Storage) {
        self.hosting_manager.set_storage(storage);
    }

    /// Drop the ring's clones of the redb `Storage` handle (hosting metadata +
    /// broken-invariants persistence). Called on node shutdown so the redb
    /// `Database` Arc count can fall to zero and release the on-disk file lock
    /// — otherwise an in-process restart against the same data dir deadlocks on
    /// the still-held lock. See issue #4401.
    pub(crate) fn clear_redb_storage(&self) {
        self.hosting_manager.clear_storage();
        self.broken_invariants.clear_storage();
    }

    /// Load hosting cache from persisted storage.
    ///
    /// Call this during startup after storage is available to restore
    /// the hosting cache from the previous run. Also migrates legacy contracts
    /// that have state but no hosting metadata.
    ///
    /// # Arguments
    /// * `storage` - The storage backend
    /// * `code_hash_lookup` - Function to look up CodeHash from ContractInstanceId.
    ///   Uses ContractStore which has the id->code_hash mapping.
    #[cfg(feature = "redb")]
    pub fn load_hosting_cache<F>(
        &self,
        storage: &crate::contract::storages::Storage,
        code_hash_lookup: F,
    ) -> Result<usize, redb::Error>
    where
        F: Fn(
            &freenet_stdlib::prelude::ContractInstanceId,
        ) -> Option<freenet_stdlib::prelude::CodeHash>,
    {
        self.hosting_manager
            .load_from_storage(storage, code_hash_lookup)
    }

    /// Load hosting cache from persisted storage (sqlite version).
    ///
    /// Also migrates legacy contracts that have state but no hosting metadata.
    #[cfg(all(feature = "sqlite", not(feature = "redb")))]
    pub async fn load_hosting_cache<F>(
        &self,
        storage: &crate::contract::storages::Storage,
        code_hash_lookup: F,
    ) -> Result<usize, crate::contract::storages::sqlite::SqlDbError>
    where
        F: Fn(
            &freenet_stdlib::prelude::ContractInstanceId,
        ) -> Option<freenet_stdlib::prelude::CodeHash>,
    {
        self.hosting_manager
            .load_from_storage(storage, code_hash_lookup)
            .await
    }

    pub fn record_request(
        &self,
        recipient: PeerKeyLocation,
        target: Location,
        request_type: TransactionType,
    ) {
        self.connection_manager
            .topology_manager
            .write()
            .record_request(recipient, target, request_type);
    }

    /// Add a connection to the ring topology.
    ///
    /// Returns `true` if this connection caused us to cross the readiness threshold
    /// (i.e., we just became ready to accept non-CONNECT operations).
    /// Returns `false` if the connection was rejected (e.g., capacity cap) or we
    /// were already ready.
    pub async fn add_connection(&self, loc: Location, peer: PeerId, was_reserved: bool) -> bool {
        tracing::info!(
            peer = %peer,
            peer_location = %loc,
            this = ?self.connection_manager.get_own_addr(),
            was_reserved = %was_reserved,
            "Adding connection to peer"
        );
        let min_ready = self.connection_manager.min_ready_connections;
        let was_ready = min_ready == 0 || self.connection_manager.connection_count() >= min_ready;

        let addr = peer.socket_addr();
        let pub_key = peer.pub_key().clone();
        let added = self
            .connection_manager
            .add_connection(loc, addr, pub_key, was_reserved);
        if !added {
            tracing::warn!(
                peer = %peer,
                peer_location = %loc,
                "Ring rejected connection - not updating caches or logging connection event"
            );
            return false;
        }
        if let Some(own_loc) = self.connection_manager.own_location().location() {
            crate::node::network_status::set_own_location(own_loc.as_f64());
        }
        // ConnectEvent::Connected telemetry is emitted by the CONNECT state
        // machine with proper transaction context; not duplicated here (#3578).
        self.refresh_density_request_cache();

        let is_ready = self.connection_manager.is_self_ready();
        // Return true only if we just crossed the threshold
        !was_ready && is_ready
    }

    pub fn update_connection_identity(&self, old_peer: &PeerId, new_peer: PeerId) {
        if self.connection_manager.update_peer_identity(
            old_peer.socket_addr(),
            new_peer.socket_addr(),
            new_peer.pub_key().clone(),
        ) {
            self.refresh_density_request_cache();
        }
    }

    fn refresh_density_request_cache(&self) {
        let cbl = self.connection_manager.get_connections_by_location();
        let topology_manager = &mut self.connection_manager.topology_manager.write();
        let _refreshed = topology_manager.refresh_cache(&cbl);
    }

    /// Returns a filtered iterator for peers that are not connected to this node already.
    pub fn is_not_connected<'a>(
        &self,
        peers: impl Iterator<Item = &'a PeerKeyLocation>,
    ) -> impl Iterator<Item = &'a PeerKeyLocation> + Send {
        let mut filtered = Vec::new();
        for peer in peers {
            if let Some(addr) = peer.socket_addr() {
                if !self.connection_manager.has_connection_or_pending(addr) {
                    filtered.push(peer);
                }
            } else {
                // If address is unknown, include the peer
                filtered.push(peer);
            }
        }
        filtered.into_iter()
    }

    /// Return the most optimal peer for hosting a given contract.
    ///
    /// This function only considers connected peers, not the node itself.
    #[inline]
    pub fn closest_potentially_hosting(
        &self,
        contract_key: &ContractKey,
        skip_list: impl Contains<std::net::SocketAddr>,
    ) -> Option<PeerKeyLocation> {
        let router = self.router.read();
        let target = Location::from(contract_key);
        let (peer, decision) = self
            .connection_manager
            .routing_with_telemetry(target, None, skip_list, &router);

        if let Some(decision) = &decision {
            tracing::debug!(
                target_location = %target.as_f64(),
                strategy = ?decision.strategy,
                num_candidates = decision.candidates.len(),
                total_routing_events = decision.total_routing_events,
                selected = peer.is_some(),
                "routing_decision"
            );
        }

        peer
    }

    /// Get k best peers for hosting a contract, ranked by routing predictions.
    /// Accepts either &ContractKey or &ContractInstanceId (both implement From<&T> for Location).
    pub fn k_closest_potentially_hosting<K>(
        &self,
        contract_id: &K,
        skip_list: impl Contains<std::net::SocketAddr> + Clone,
        k: usize,
    ) -> Vec<PeerKeyLocation>
    where
        for<'a> Location: From<&'a K>,
    {
        // Router read-lock is only needed for the final `select_*` call;
        // building the candidate list does not need it. Acquire it late to
        // keep the critical section short (clippy: `significant_drop_tightening`).
        let target_location = Location::from(contract_id);

        let mut seen = HashSet::new();
        let mut candidates: Vec<PeerKeyLocation> = Vec::new();
        let mut not_ready_fallback: Vec<PeerKeyLocation> = Vec::new();
        let mut skipped_not_ready: usize = 0;
        let mut skipped_transient: usize = 0;

        let connections = self.connection_manager.get_connections_by_location();
        // Sort keys for deterministic iteration order (HashMap iteration is non-deterministic)
        // This ensures the `seen.insert()` check behaves consistently across runs
        let mut sorted_keys: Vec<_> = connections.keys().collect();
        sorted_keys.sort();
        for loc in sorted_keys {
            let conns = connections.get(loc).expect("key exists");
            // Sort connections for deterministic iteration order
            let mut sorted_conns: Vec<_> = conns.iter().collect();
            sorted_conns.sort_by_key(|c| c.location.clone());
            for conn in sorted_conns {
                if let Some(addr) = conn.location.socket_addr() {
                    if skip_list.has_element(addr) || !seen.insert(addr) {
                        continue;
                    }
                    // Skip transient peers — these are short-TTL connections used for
                    // CONNECT coordination, not stable routing targets. PUT/UPDATE
                    // already exclude them via `ConnectionManager::routing_candidates`
                    // (see connection_manager.rs:1578); previously GET/SUBSCRIBE
                    // could route through them and waste hops on a forwarder that
                    // was about to be dropped. Issue #4222 / #3570.
                    if self.connection_manager.is_transient(addr) {
                        tracing::debug!(
                            %addr,
                            target_location = %target_location.as_f64(),
                            "k_closest: skipping transient peer"
                        );
                        skipped_transient += 1;
                        continue;
                    }
                    // Skip peers that haven't advertised readiness, but collect them
                    // as fallback candidates in case all peers fail the readiness check.
                    if !self.connection_manager.is_peer_ready(addr) {
                        tracing::debug!(
                            %addr,
                            target_location = %target_location.as_f64(),
                            "k_closest: skipping peer not yet ready"
                        );
                        not_ready_fallback.push(conn.location.clone());
                        skipped_not_ready += 1;
                        continue;
                    }
                } else {
                    // Addressless candidates bypass every filter above (skip
                    // list, dedup, transient, readiness) because all of them
                    // key on the socket addr. If the router then selects one,
                    // the GET advance helpers can't use it as a wire target
                    // and the hop dies. Keep the inclusion (the candidate may
                    // gain an addr by send time) but make it visible (#4361).
                    tracing::debug!(
                        peer = ?conn.location,
                        target_location = %target_location.as_f64(),
                        "k_closest: including addressless candidate (bypasses addr-keyed filters)"
                    );
                }
                candidates.push(conn.location.clone());
            }
        }

        // If all connected peers failed the readiness check, fall back to using them anyway.
        // This prevents GET/SUBSCRIBE operations from failing with EmptyRing when the node
        // is connected but peers haven't yet sent ReadyState messages (e.g., early after
        // connecting, or in network topologies where the min_ready_connections threshold is
        // never satisfied). A warn-level log is emitted so operators know gating was bypassed.
        // Note: ConnectionManager::routing_candidates has the same fallback for PUT/UPDATE.
        if candidates.is_empty() && !not_ready_fallback.is_empty() {
            tracing::warn!(
                count = not_ready_fallback.len(),
                target_location = %target_location.as_f64(),
                "k_closest: no ready peers available, falling back to not-yet-ready peers to avoid EmptyRing"
            );
            candidates = not_ready_fallback;
        }

        // If the entire connection set was filtered out and transients carried
        // weight in that filtering, warn the operator. There is no transient
        // fallback by design — transient peers are short-TTL coordination slots
        // and routing through them just wastes hops — but a sustained
        // empty-candidate state means the local node has no viable routing
        // options for this hop, which is operator-visible information that
        // would otherwise only surface as an `EmptyRing` higher up.
        if candidates.is_empty() && skipped_transient > 0 {
            tracing::warn!(
                skipped_transient,
                target_location = %target_location.as_f64(),
                "k_closest: no viable peers — all eligible connections were transient \
                 (no fallback by design; routing will fail this hop)"
            );
        }

        // Sort candidates for deterministic input to select_k_best_peers
        candidates.sort();

        // Note: We intentionally do NOT fall back to known_locations here.
        // known_locations may contain peers we're not currently connected to,
        // and attempting to route to them would require establishing a new connection
        // which may fail (especially in NAT scenarios without coordination).
        // It's better to return fewer candidates than unreachable ones.

        let (selected, decision) = self.router.read().select_k_best_peers_with_telemetry(
            candidates.iter(),
            target_location,
            k,
        );
        // `selected` borrows from `candidates`, not from the router guard, so
        // the read lock is released here before the tracing/collect below.

        tracing::debug!(
            target_location = %target_location.as_f64(),
            strategy = ?decision.strategy,
            num_candidates = decision.candidates.len(),
            total_routing_events = decision.total_routing_events,
            selected_count = selected.len(),
            "routing_decision"
        );

        tracing::debug!(
            target_location = %target_location.as_f64(),
            candidates_found = selected.len(),
            skipped_not_ready,
            skipped_transient,
            "k_closest_potentially_hosting result"
        );

        selected.into_iter().cloned().collect()
    }

    pub fn routing_finished(&self, event: crate::router::RouteEvent) {
        self.connection_manager
            .topology_manager
            .write()
            .report_outbound_request(event.peer.clone(), event.contract_location);

        // Update peer health tracking based on routing outcome.
        if let Some(addr) = event.peer.socket_addr() {
            let mut health = self.connection_manager.peer_health.lock();
            match &event.outcome {
                crate::router::RouteOutcome::Success { .. }
                | crate::router::RouteOutcome::SuccessUntimed => {
                    health.record_success(addr);
                }
                crate::router::RouteOutcome::Failure => {
                    health.record_failure(addr);
                }
            }
        }

        self.router.write().add_event(event);
    }

    // ==================== Subscription Management (Lease-Based) ====================

    /// Subscribe to a contract with a lease.
    ///
    /// Creates a new subscription or renews an existing one. The subscription
    /// will expire after `SUBSCRIPTION_LEASE_DURATION` unless renewed.
    pub fn subscribe(&self, contract: ContractKey) -> SubscribeResult {
        self.hosting_manager.subscribe(contract)
    }

    /// Unsubscribe from a contract.
    ///
    /// Removes the active subscription. The contract may still be hosted
    /// (in the hosting cache) until evicted by LRU.
    pub fn unsubscribe(&self, contract: &ContractKey) {
        self.hosting_manager.unsubscribe(contract)
    }

    /// Check if we have an active (non-expired) subscription to a contract.
    pub fn is_subscribed(&self, contract: &ContractKey) -> bool {
        self.hosting_manager.is_subscribed(contract)
    }

    /// Get all contracts with active subscriptions.
    pub fn get_subscribed_contracts(&self) -> Vec<ContractKey> {
        self.hosting_manager.get_subscribed_contracts()
    }

    /// Force-expire a contract's subscription so it gets renewed through the
    /// current best route on the next recovery cycle.
    fn force_subscription_renewal(&self, contract: &ContractKey) {
        self.hosting_manager.force_subscription_renewal(contract);
    }

    /// Expire stale subscriptions and return the contracts that were expired.
    ///
    /// Should be called periodically by a background task.
    pub fn expire_stale_subscriptions(&self) -> Vec<ContractKey> {
        self.hosting_manager.expire_stale_subscriptions()
    }

    // ==================== Downstream Subscriber Tracking ====================

    pub fn add_downstream_subscriber(&self, contract: &ContractKey, peer: PeerKey) -> bool {
        let outcome = self
            .hosting_manager
            .add_downstream_subscriber(contract, peer);
        // No governance demand is ingested here anymore. Benefit is a
        // LIVE SNAPSHOT read fresh each reaper tick from the hosting
        // manager's standing subscriber count (see
        // `Ring::governance_tick` and
        // `HostingManager::downstream_subscriber_count`), not an
        // accumulator fed by subscribe events. That makes the
        // Sybil-on-renewal concern moot: renewals merely extend a lease
        // the snapshot already counts, so they cannot inflate benefit —
        // the count reflects the CURRENT lease-valid subscriber set, no
        // matter how often each peer renews.
        //
        // Downstream demand is still weighted low (FORWARDED_DEMAND_WEIGHT
        // = 0.1) when the snapshot is computed, because peer identities
        // are attacker-rotatable: without an identity layer a single
        // attacker can spin up many peers that each "subscribe", so each
        // forwarded subscriber is worth one tenth of a real local client.
        // Caller-facing bool preserved for backward compat: Rejected
        // → false; NewAdd or Renewal → true (the peer is tracked).
        !matches!(
            outcome,
            crate::ring::hosting::AddSubscriberOutcome::Rejected
        )
    }

    #[allow(dead_code)] // Only used in tests
    pub fn renew_downstream_subscriber(&self, contract: &ContractKey, peer: &PeerKey) -> bool {
        self.hosting_manager
            .renew_downstream_subscriber(contract, peer)
    }

    pub fn remove_downstream_subscriber(&self, contract: &ContractKey, peer: &PeerKey) -> bool {
        self.hosting_manager
            .remove_downstream_subscriber(contract, peer)
    }

    pub fn has_downstream_subscribers(&self, contract: &ContractKey) -> bool {
        self.hosting_manager.has_downstream_subscribers(contract)
    }

    /// Whether something still depends on this node hosting `contract` — a
    /// live local client subscription or a downstream peer subscriber.
    ///
    /// Used to gate hosting-cache eviction reclamation: a contract that is in
    /// use must not have its on-disk state/code deleted. See
    /// `HostingManager::contract_in_use` for why an active upstream network
    /// subscription alone does NOT make the contract in-use (the renewal
    /// machinery would refresh the lease unboundedly).
    pub(crate) fn contract_in_use(&self, contract: &ContractKey) -> bool {
        self.hosting_manager.contract_in_use(contract)
    }

    /// Single helper for every state-write chokepoint. Does the three
    /// things a chokepoint MUST do, in order:
    ///
    /// 1. Bump the per-contract write generation (closes the
    ///    `EvictContract` re-host race — see
    ///    `HostingManager::state_generation`).
    /// 2. Refresh the hosting-cache snapshot of that generation so
    ///    already-hosted contracts don't leak on eviction after this
    ///    write — see `HostingCache::refresh_entry_generation`.
    /// 3. Report `state_size` bytes against this contract on the
    ///    `StateBytesWritten` axis of the topology meter, feeding the
    ///    governance scoring layer (see `crate::governance`).
    ///
    /// Every state-write chokepoint in the executor MUST go through
    /// this helper, NOT call the three primitives by hand. The
    /// "manually-mirrored side effects after a task-per-tx migration"
    /// pattern in `.claude/rules/bug-prevention-patterns.md` lists this
    /// exact failure mode: one site drops the report and governance
    /// silently undercounts that path for months before anyone notices.
    pub(crate) fn commit_state_write(&self, contract: &ContractKey, state_size: usize) {
        let new_gen = self.hosting_manager.bump_state_generation(contract);
        self.hosting_manager
            .refresh_cache_generation(contract, new_gen);
        self.report_contract_resource_usage(
            *contract.id(),
            crate::topology::meter::ResourceType::StateBytesWritten,
            state_size as f64,
        );
    }

    /// Read the current state-write generation for `contract` (0 if never written).
    pub(crate) fn state_generation(&self, contract: &ContractKey) -> u64 {
        self.hosting_manager.state_generation(contract)
    }

    /// Forget the state-write generation entry for `contract` after a
    /// successful disk reclamation. Keeps the generation map bounded.
    pub(crate) fn forget_state_generation(&self, contract: &ContractKey) {
        self.hosting_manager.forget_state_generation(contract)
    }

    /// Add `contract` to the pending-reclamation retry queue with the
    /// captured `expected_generation`. Called from the two skip points
    /// that drop an `EvictContract` event before it can complete (fair
    /// queue rejection, `contract_in_use` skip in `RuntimePool::remove_contract`).
    /// See `HostingManager::pending_reclamation_add` for the queue's
    /// invariants and how the periodic sweep retries entries.
    pub(crate) fn pending_reclamation_add(&self, contract: ContractKey, expected_generation: u64) {
        self.hosting_manager
            .pending_reclamation_add(contract, expected_generation)
    }

    /// Remove `contract` from the pending-reclamation retry queue after
    /// a successful disk reclamation. See
    /// `HostingManager::pending_reclamation_remove`.
    pub(crate) fn pending_reclamation_remove(&self, contract: &ContractKey) {
        self.hosting_manager.pending_reclamation_remove(contract)
    }

    /// Snapshot the pending-reclamation queue for the periodic sweep
    /// to iterate without holding any DashMap shard guard. See
    /// `HostingManager::pending_reclamation_snapshot`.
    pub(crate) fn pending_reclamation_snapshot(&self) -> Vec<(ContractKey, u64)> {
        self.hosting_manager.pending_reclamation_snapshot()
    }

    pub fn expire_stale_downstream_subscribers(&self) -> Vec<(ContractKey, usize)> {
        self.hosting_manager.expire_stale_downstream_subscribers()
    }

    pub fn should_unsubscribe_upstream(&self, contract: &ContractKey) -> bool {
        self.hosting_manager.should_unsubscribe_upstream(contract)
    }

    /// Check if this node is actively receiving updates for a contract.
    ///
    /// Returns true only when we have an active network subscription or local
    /// client subscriptions. The hosting LRU cache alone is not sufficient,
    /// since cached state may be stale after subscription expiry.
    pub fn is_receiving_updates(&self, contract: &ContractKey) -> bool {
        self.hosting_manager.is_receiving_updates(contract)
    }

    /// Get contracts that need subscription renewal.
    ///
    /// Returns contracts where:
    /// - We have an active subscription that will expire soon, OR
    /// - We have client subscriptions but no active network subscription, OR
    /// - We have hosted contracts without active subscriptions (THE FIX)
    pub fn contracts_needing_renewal(&self) -> Vec<ContractKey> {
        self.hosting_manager.contracts_needing_renewal()
    }

    // ==================== Client Subscription Management ====================

    /// Register a client subscription for a contract (WebSocket client subscribed).
    ///
    /// Returns information about the operation for telemetry.
    pub fn add_client_subscription(
        &self,
        instance_id: &ContractInstanceId,
        client_id: crate::client_events::ClientId,
    ) -> AddClientSubscriptionResult {
        // No governance demand is ingested here anymore. A local client
        // subscription is the strong demand signal (full
        // LOCAL_DEMAND_WEIGHT = 1.0), but benefit is now a LIVE SNAPSHOT
        // read fresh each reaper tick from
        // `HostingManager::local_client_count` (see
        // `Ring::governance_tick`), not an accumulator fed here. The
        // snapshot counts the currently-subscribed client set, so an
        // idempotent re-subscribe cannot inflate benefit and there is no
        // need to gate on `is_new_for_client` for scoring purposes.
        self.hosting_manager
            .add_client_subscription(instance_id, client_id)
    }

    /// Remove a client from all its subscriptions (used when client disconnects).
    ///
    /// Returns a [`ClientDisconnectResult`] with:
    /// - `affected_contracts`: all contracts where the client was subscribed (for cleanup)
    pub fn remove_client_from_all_subscriptions(
        &self,
        client_id: crate::client_events::ClientId,
    ) -> ClientDisconnectResult {
        self.hosting_manager
            .remove_client_from_all_subscriptions(client_id)
    }

    /// Get all hosted contract keys from the hosting cache.
    pub fn hosting_contract_keys(&self) -> Vec<ContractKey> {
        self.hosting_manager.hosting_contract_keys()
    }

    /// Get the cached state size in bytes for a hosted contract.
    pub fn hosting_contract_size(&self, key: &ContractKey) -> u64 {
        self.hosting_manager.hosting_contract_size(key)
    }

    /// Get the number of contracts in the hosting cache.
    /// This is the actual count of contracts this node is caching/hosting.
    pub fn hosting_contracts_count(&self) -> usize {
        self.hosting_manager.hosting_contracts_count()
    }

    /// Get subscription state for all contracts (for telemetry).
    ///
    /// Returns: (contract, has_client_subscription, is_active_subscription, expires_at)
    pub fn get_subscription_states(&self) -> Vec<(ContractKey, bool, bool, Option<Instant>)> {
        self.hosting_manager.get_subscription_states()
    }

    /// Snapshot of every active subscription for the local-peer dashboard.
    /// Reads directly from the canonical lease map.
    pub fn dashboard_subscription_snapshot(&self) -> Vec<SubscribedContractSnapshot> {
        self.hosting_manager.dashboard_subscription_snapshot()
    }

    /// Snapshot of per-contract governance state for the local-peer
    /// dashboard. Reads directly from the canonical `GovernanceManager`
    /// state — no mirror, no cache, no derived recomputation. If a
    /// state appears here it's because the manager computed it from
    /// real meter samples and real subscription events.
    pub fn dashboard_governance_snapshot(&self) -> crate::node::network_status::GovernanceSnapshot {
        use crate::contract::governance as gov;
        use crate::node::network_status as ns;

        let now = self.time_source.now();
        let mode = match self.governance.mode() {
            gov::GovernanceMode::Off => ns::GovernanceModeSnapshot::Off,
            gov::GovernanceMode::DryRun => ns::GovernanceModeSnapshot::DryRun,
            gov::GovernanceMode::Enforce => ns::GovernanceModeSnapshot::Enforce,
        };

        let map_state = |s: gov::GovernanceState| match s {
            gov::GovernanceState::Normal => ns::GovernanceStateSnapshot::Normal,
            gov::GovernanceState::Borderline => ns::GovernanceStateSnapshot::Borderline,
            gov::GovernanceState::WouldEvict => ns::GovernanceStateSnapshot::WouldEvict,
            gov::GovernanceState::Evicted => ns::GovernanceStateSnapshot::Evicted,
            gov::GovernanceState::Banned => ns::GovernanceStateSnapshot::Banned,
        };
        let map_reason = |r: gov::TransitionReason| match r {
            gov::TransitionReason::FirstSeen => ns::GovernanceTransitionReasonSnapshot::FirstSeen,
            gov::TransitionReason::BorderlineEntered => {
                ns::GovernanceTransitionReasonSnapshot::BorderlineEntered
            }
            gov::TransitionReason::ThresholdCrossed => {
                ns::GovernanceTransitionReasonSnapshot::ThresholdCrossed
            }
            gov::TransitionReason::Evicted => ns::GovernanceTransitionReasonSnapshot::Evicted,
            gov::TransitionReason::BanTriggered => {
                ns::GovernanceTransitionReasonSnapshot::BanTriggered
            }
            gov::TransitionReason::Recovered => ns::GovernanceTransitionReasonSnapshot::Recovered,
            gov::TransitionReason::BanLifted => ns::GovernanceTransitionReasonSnapshot::BanLifted,
        };

        // Only iterate flagged contracts for the dashboard mirror —
        // the renderer hides Normal anyway. Avoids cloning thousands of
        // entries per refresh on a busy node. See `iter_flagged_scores`.
        let contracts: Vec<ns::ContractGovernanceEntry> = self
            .governance
            .iter_flagged_scores()
            .into_iter()
            .map(|(id, score)| {
                let instance_id = id.to_string();
                let instance_id_short = if instance_id.chars().count() > 12 {
                    let trunc: String = instance_id.chars().take(12).collect();
                    format!("{trunc}...")
                } else {
                    instance_id.clone()
                };
                let history = score
                    .history
                    .iter()
                    .map(|t| ns::GovernanceTransitionEntry {
                        secs_ago: now.saturating_duration_since(t.at).as_secs(),
                        from: map_state(t.from),
                        to: map_state(t.to),
                        reason: map_reason(t.reason),
                    })
                    .collect();
                ns::ContractGovernanceEntry {
                    instance_id,
                    instance_id_short,
                    state: map_state(score.state),
                    cost_used: score.cost_used,
                    benefit_score: score.benefit_score,
                    log_ratio: score.log_ratio(self.governance.benefit_floor()),
                    age_secs: now.saturating_duration_since(score.first_seen).as_secs(),
                    last_transition_secs_ago: now
                        .saturating_duration_since(score.last_transition)
                        .as_secs(),
                    history,
                }
            })
            .collect();

        let norms = match self.governance.latest_norms() {
            Some(n) => ns::NetworkNorms {
                median_log_ratio: n.median_log_ratio,
                mad: n.mad,
                threshold: n.threshold,
                sample_size: n.sample_size,
                capacity_ceiling_binding: n.capacity_ceiling_binding,
                skip_reason: n.skip_reason.map(|r| match r {
                    crate::governance::SkipReason::InsufficientSamples => {
                        ns::GovernanceSkipReasonSnapshot::InsufficientSamples
                    }
                    crate::governance::SkipReason::MadCollapsed => {
                        ns::GovernanceSkipReasonSnapshot::MadCollapsed
                    }
                    crate::governance::SkipReason::NoExtractableRatios => {
                        ns::GovernanceSkipReasonSnapshot::NoExtractableRatios
                    }
                }),
            },
            None => ns::NetworkNorms::default(),
        };

        // state_by_id: map ContractInstanceId → state for the
        // Subscribed Contracts table's Gov column cross-reference.
        // Walking iter_flagged_scores() once for the `contracts`
        // list above gave us the flagged set; for unflagged
        // contracts the absence from this map means "Normal".
        let state_by_id: std::collections::HashMap<String, ns::GovernanceStateSnapshot> = contracts
            .iter()
            .map(|c| (c.instance_id.clone(), c.state))
            .collect();

        let observed_count = self.governance.len();
        let min_samples = self.governance.outlier_min_samples();
        let last_tick_at = self.governance.latest_norms().map(|n| n.at);

        ns::GovernanceSnapshot {
            mode,
            contracts,
            observed_count,
            min_samples,
            norms,
            last_tick_at,
            state_by_id,
        }
    }

    /// Snapshot of the contract ban list for the local-peer dashboard
    /// (#4302). Reads directly from the canonical `contract_ban_list` —
    /// no mirror, no cache. The count, capacity-rejection counter, and
    /// per-entry list (key + reason + time remaining) all come from the
    /// list's own accessors, so the panel can't drift the way a mirrored
    /// counter would.
    pub fn dashboard_ban_list_snapshot(&self) -> crate::node::network_status::BanListSnapshot {
        use crate::node::network_status as ns;
        use crate::ring::contract_ban_list::BanReason;

        let map_reason = |r: BanReason| match r {
            BanReason::AutoMad => ns::BanReasonSnapshot::AutoMad,
            BanReason::Operator => ns::BanReasonSnapshot::Operator,
        };

        let mut entries: Vec<ns::BanListEntry> = self
            .contract_ban_list
            .snapshot()
            .into_iter()
            .map(|e| ns::BanListEntry {
                instance_id: e.contract.to_string(),
                reason: map_reason(e.reason),
                expires_in_secs: e.remaining.as_secs(),
            })
            .collect();
        // Stable display order: soonest-to-lift first, then by id so the
        // dashboard doesn't reshuffle rows between refreshes (DashMap
        // iteration order is unspecified).
        entries.sort_by(|a, b| {
            a.expires_in_secs
                .cmp(&b.expires_in_secs)
                .then_with(|| a.instance_id.cmp(&b.instance_id))
        });

        // Count LIVE entries — derive from the filtered snapshot, not
        // `ContractBanList::len()`. `len()` includes entries that have
        // expired but not yet been swept by `cleanup()`/`unban()`;
        // `snapshot()` filters those out with the same `now < expires_at`
        // predicate as `is_banned`. Using `len()` here would let the
        // count tile read "1 contract banned" while the entry list (and
        // the wire boundary) show zero — an inconsistency in the
        // expiry-before-sweep window. (Codex review on #4464.)
        ns::BanListSnapshot {
            count: entries.len(),
            capacity_rejected_total: self.contract_ban_list.capacity_rejected_total(),
            entries,
        }
    }

    /// Record that a state update was observed for `contract`.
    /// No-op if the contract is not currently subscribed.
    pub fn record_contract_update(&self, contract: &ContractKey) {
        self.hosting_manager.record_contract_update(contract)
    }

    /// True if `contract` has been flagged as violating a CRDT invariant
    /// (e.g. non-idempotent `update_state`). Callers on the broadcast path
    /// must skip emission when this returns true to prevent the broken
    /// contract from generating a propagation storm.
    pub fn is_contract_broken(&self, contract: &ContractKey) -> bool {
        self.broken_invariants.is_broken(contract.id())
    }

    /// Mark `contract` as broken with `kind`. Idempotent. See
    /// [`broken_invariants`] module docs.
    pub(crate) fn record_broken_invariant(&self, contract: ContractKey, kind: BrokenInvariant) {
        self.broken_invariants.record(*contract.id(), kind);
    }

    /// Wire persistent storage for the broken-invariants tracker. Called
    /// once at executor wiring time.
    pub(crate) fn set_broken_invariants_storage(
        &self,
        storage: crate::contract::storages::Storage,
    ) {
        self.broken_invariants.set_storage(storage);
    }

    /// Report a per-contract resource sample to the topology meter.
    ///
    /// Lightweight non-blocking write: takes the topology manager's
    /// write lock briefly to insert into the running-average store.
    /// Safe to call from executor commit paths — does NOT use channels
    /// (cf. `.claude/rules/channel-safety.md`); the May 2026 deadlock
    /// pattern (`#4145`) is not reachable through this surface.
    ///
    /// Used by `Executor::commit_state_update` to attribute state-write
    /// bytes to a contract, and (in follow-up work) by the WASM call
    /// wrappers for CPU/fuel and the broadcast dispatcher for fanout
    /// cost. Per-contract attribution lets the shared governance
    /// outlier-detection module (`crate::governance`) score contracts
    /// against the network's observed cost-per-benefit distribution.
    pub(crate) fn report_contract_resource_usage(
        &self,
        contract_id: freenet_stdlib::prelude::ContractInstanceId,
        resource: crate::topology::meter::ResourceType,
        amount: f64,
    ) {
        // Use the injectable `TimeSource` (rather than `Instant::now()`
        // directly) so deterministic simulation tests can drive this
        // path. Per `.claude/rules/code-style.md` "Need current time?
        // → USE: TimeSource trait" — the executor commit path will reach
        // here from inside simulated nodes once the governance scoring
        // integration tests land, and reading wall-clock there would
        // break determinism.
        let now = self.time_source.now();
        let mut topo = self.connection_manager.topology_manager.write();
        topo.report_resource_usage(
            &crate::topology::meter::AttributionSource::Contract(contract_id),
            resource,
            amount,
            now,
        );
        drop(topo);
        // Also feed the governance manager — the meter stores rates
        // for telemetry/the dashboard's bandwidth view, while the
        // governance manager aggregates these samples into the
        // cost/benefit ratio that drives state. Both ingest in
        // parallel from this one entry point so there's no risk of
        // them diverging (governance reading from a stale meter
        // snapshot, etc).
        //
        // Per-resource weight: identity (1.0) for now. Future tuning
        // could weight CPU-µs and fanout-cost higher than
        // state-bytes, but until we have live data to calibrate
        // against, unit weights match what the design doc proposed
        // as a starting point.
        let weight = resource_weight(resource);
        self.governance.ingest_cost(contract_id, amount * weight);
    }

    // Note: there is intentionally no `ingest_contract_demand` method
    // here, and demand is no longer pushed on subscribe events at all.
    // Benefit is a LIVE SNAPSHOT pulled each reaper tick by
    // `governance_tick` from the hosting manager's standing subscriber
    // counts (`local_client_count` + `downstream_subscriber_count`).
    // A push entry point would re-introduce the accumulate-on-event
    // model this redesign removed (#4296).

    /// Run one governance reaper tick. Caller (the periodic
    /// `governance_reaper_loop` task) is expected to feed
    /// `tick_interval` = wall-time since the previous tick for cost
    /// decay. Returns the `ReaperTickResult` for the caller to act on
    /// (emit `EvictContract` events for actionable decisions, log
    /// dry-run decisions, surface stats on the dashboard).
    ///
    /// Before ticking, this builds the LIVE benefit snapshot for every
    /// contract the governance manager is tracking: for each tracked
    /// contract id, `LOCAL_DEMAND_WEIGHT × current local-client count +
    /// FORWARDED_DEMAND_WEIGHT × current (non-expired) downstream
    /// subscriber count`, read fresh from the hosting manager. This is
    /// the cost-per-current-beneficiary denominator — a popular
    /// contract keeps a high benefit because its beneficiaries are
    /// counted live, not decayed from old subscribe events.
    pub(crate) fn governance_tick(
        &self,
        tick_interval: Duration,
    ) -> crate::contract::governance::ReaperTickResult {
        // Single-pass live benefit snapshot: one iteration over
        // `client_subscriptions` and one over `downstream_subscribers`
        // (see `HostingManager::beneficiary_counts`), then filter to the
        // contracts governance is tracking. This replaces the previous
        // shape that deep-cloned every `ContractScore` (incl. its
        // history Vec) via `iter_scores()` and then re-scanned the whole
        // `downstream_subscribers` map per contract — O(N×M) per 60s
        // tick. `tracked_ids()` returns keys only (no score clone).
        let tracked = self.governance.tracked_ids();
        let mut benefits = self
            .hosting_manager
            .beneficiary_counts(LOCAL_DEMAND_WEIGHT, FORWARDED_DEMAND_WEIGHT);
        // Keep only tracked contracts (a contract with beneficiaries but
        // no ingested cost has no score and must not enter the map —
        // preserves the prior per-contract behavior).
        benefits.retain(|id, _| tracked.contains(id));
        self.governance.tick(tick_interval, &benefits)
    }

    /// Test-only accessor: current local-client beneficiary count for a
    /// contract, read from the hosting manager. Used by the governance
    /// sim e2e tests to assert the live benefit snapshot reflects real
    /// subscriptions.
    #[cfg(all(test, feature = "simulation_tests"))]
    pub(crate) fn hosting_manager_local_client_count(
        &self,
        instance_id: &ContractInstanceId,
    ) -> usize {
        self.hosting_manager.local_client_count(instance_id)
    }

    /// Test-only accessor: current (non-expired) downstream subscriber
    /// count for a contract, read from the hosting manager.
    #[cfg(all(test, feature = "simulation_tests"))]
    pub(crate) fn hosting_manager_downstream_subscriber_count(
        &self,
        instance_id: &ContractInstanceId,
    ) -> usize {
        self.hosting_manager
            .downstream_subscriber_count(instance_id)
    }
}

/// Per-resource weight for cost aggregation. Defaults to 1.0 for all
/// dimensions so total cost = simple sum of meter samples. Hooks here
/// for future tuning if any one dimension proves to dominate the
/// signal-to-noise ratio.
fn resource_weight(resource: crate::topology::meter::ResourceType) -> f64 {
    use crate::topology::meter::ResourceType::*;
    match resource {
        InboundBandwidthBytes | OutboundBandwidthBytes => 1.0,
        ExecCpuMicros | ExecFuelUnits => 1.0,
        StateBytesWritten | BroadcastFanoutCost => 1.0,
    }
}

/// Bridge the transport layer's per-peer wire-byte counters into the
/// topology meter so `TopologyManager::adjust_topology` can make load-aware
/// connection decisions in production (#3453).
///
/// `TRANSPORT_METRICS.per_peer_snapshot()` exposes *cumulative* sent/received
/// byte counts per peer socket address. The topology meter, by contrast,
/// accumulates each reported sample into a sliding-window sum and divides by
/// the elapsed wall-time (`now − oldest_sample`) to produce a rate. We diff
/// the current snapshot against the previous tick's counts (`prev`) to recover
/// the bytes transferred *during the interval*, and feed that delta as one
/// sample per direction:
///   - received bytes → `InboundBandwidthBytes`
///   - sent bytes     → `OutboundBandwidthBytes`
///
/// `interval_start` MUST be the time of the *previous* maintenance tick, not
/// the current one. The meter divides the windowed byte sum by
/// `query_time − oldest_sample_time` with a 1-second floor
/// (`RunningAverage::get_rate_at_time`). `adjust_topology` queries the meter at
/// the current tick time immediately after this call, so timestamping the
/// interval's whole byte delta at the interval START makes the very first
/// sample divide by the real interval length (e.g. ~60s) instead of flooring
/// to 1s. Timestamping at the interval END would treat a full interval's bytes
/// as if they were transferred in a single second, overstating the rate by up
/// to the tick interval (~60×) and triggering spurious connection removals
/// under ordinary traffic (#3453 review).
///
/// Only deltas with a resolvable `PeerKeyLocation` are reported: an address
/// that doesn't (yet) map to a ring peer — e.g. a transient handshake
/// connection — is skipped rather than attributed to a phantom source, so we
/// don't pollute the meter with samples that can't be matched against the
/// node's actual neighbors.
///
/// Three things are bounded as peers churn (#3453 review):
///   - `prev` is pruned of peers absent from `snapshot`, so it stays bounded
///     by the transport metrics table size (`MAX_TRACKED_PEERS`).
///   - The topology meter's per-source bandwidth meters AND
///     `source_creation_times` are pruned to the set of peers resolved this
///     tick via `retain_peer_sources`, so neither grows without bound (and
///     `adjust_topology`, which iterates `source_creation_times` every tick,
///     stays O(live peers)).
///
/// This takes the topology manager's write lock once per reported delta plus
/// once for the retain. It performs no channel sends and no `.await`, so it is
/// safe to call inline on the `connection_maintenance` tick
/// (cf. `.claude/rules/channel-safety.md`).
fn feed_peer_bandwidth_to_meter(
    connection_manager: &ConnectionManager,
    snapshot: &[(SocketAddr, u64, u64)],
    prev: &mut HashMap<SocketAddr, (u64, u64)>,
    interval_start: Instant,
) {
    use crate::topology::meter::{AttributionSource, ResourceType};

    // Peers resolved to a ring `PeerKeyLocation` this tick. Used to bound the
    // meter / source_creation_times to the live connection set below.
    let mut live_peers: std::collections::HashSet<PeerKeyLocation> =
        std::collections::HashSet::new();

    for &(addr, cum_sent, cum_recv) in snapshot {
        // Cumulative counters only ever increase, but use saturating
        // subtraction so a counter reset (e.g. the per-peer slot was evicted
        // and re-created since the last tick) can never underflow into a
        // huge bogus delta.
        let (prev_sent, prev_recv) = prev.get(&addr).copied().unwrap_or((0, 0));
        let sent_delta = cum_sent.saturating_sub(prev_sent);
        let recv_delta = cum_recv.saturating_sub(prev_recv);

        // Always record the latest cumulative counts so the next tick diffs
        // against them, even when this tick produced no usable delta.
        prev.insert(addr, (cum_sent, cum_recv));

        let Some(peer) = connection_manager.get_peer_by_addr(addr) else {
            // No ring peer for this address (transient/handshake connection,
            // or just-torn-down). Skip rather than attribute to a phantom.
            continue;
        };
        // Track every resolvable peer (even with a zero delta this tick) as
        // live so retain below doesn't evict a currently-connected, currently
        // idle peer's accumulated samples.
        live_peers.insert(peer.clone());

        if sent_delta == 0 && recv_delta == 0 {
            continue;
        }

        let source = AttributionSource::Peer(peer);
        let mut topo = connection_manager.topology_manager.write();
        if recv_delta > 0 {
            topo.report_resource_usage(
                &source,
                ResourceType::InboundBandwidthBytes,
                recv_delta as f64,
                interval_start,
            );
        }
        if sent_delta > 0 {
            topo.report_resource_usage(
                &source,
                ResourceType::OutboundBandwidthBytes,
                sent_delta as f64,
                interval_start,
            );
        }
    }

    // Bound the meter / source_creation_times to the live peer set so they
    // don't accumulate departed peers forever (#3453 review). Non-Peer
    // (Contract/Delegate) sources are retained by `retain_peer_sources`.
    connection_manager
        .topology_manager
        .write()
        .retain_peer_sources(&live_peers);

    // Drop prior-tick entries for peers that vanished from the snapshot so
    // `prev` stays bounded as peers churn (#3453). After the loop above,
    // every snapshot address is present in `prev`, so `prev.len() >
    // snapshot.len()` holds exactly when some prior peer is no longer in the
    // snapshot — the only case where a prune is needed.
    if prev.len() > snapshot.len() {
        let live: std::collections::HashSet<SocketAddr> =
            snapshot.iter().map(|&(addr, _, _)| addr).collect();
        prev.retain(|addr, _| live.contains(addr));
    }
}

impl Ring {
    // ==================== Subscription Retry Spam Prevention ====================

    /// Check if a subscription request can be made for a contract.
    /// Returns false if request is already pending or in backoff period.
    pub fn can_request_subscription(&self, contract: &ContractKey) -> bool {
        self.hosting_manager.can_request_subscription(contract)
    }

    /// Mark a subscription request as in-flight.
    /// Returns false if already pending.
    pub fn mark_subscription_pending(&self, contract: ContractKey) -> bool {
        self.hosting_manager.mark_subscription_pending(contract)
    }

    /// Mark a subscription request as completed.
    /// If success is false, applies exponential backoff.
    pub fn complete_subscription_request(&self, contract: &ContractKey, success: bool) {
        self.hosting_manager
            .complete_subscription_request(contract, success)
    }

    // ==================== Hosting Cache Management ====================

    /// Touch a contract in the hosting cache (refresh TTL without adding).
    ///
    /// Called when a user GET serves a hosted contract from local cache.
    pub fn touch_hosting(&self, key: &ContractKey) {
        self.hosting_manager.touch_hosting(key)
    }

    /// Mark a contract as accessed by a local client (HTTP/WebSocket).
    pub fn mark_local_client_access(&self, key: &ContractKey) {
        self.hosting_manager.mark_local_client_access(key)
    }

    /// Check if a contract was accessed by a local client.
    pub fn has_local_client_access(&self, key: &ContractKey) -> bool {
        self.hosting_manager.has_local_client_access(key)
    }

    /// Sweep for expired entries in the hosting cache.
    ///
    /// Returns `(ContractKey, write_generation)` pairs for contracts
    /// evicted from this cache. Contracts with client subscriptions
    /// (or downstream/network subscribers) are protected from eviction.
    /// The generation snapshot is carried through `EvictContract` so the
    /// deletion-time guard can detect a re-host race.
    pub fn sweep_expired_hosting(&self) -> Vec<(ContractKey, u64)> {
        self.hosting_manager.sweep_expired_hosting()
    }

    // ==================== Legacy GET Auto-Subscription (delegating to hosting cache) ====================
    /// Sweep for expired entries (delegated to hosting cache).
    ///
    /// Returns `(ContractKey, write_generation)` pairs for contracts
    /// evicted from the cache. Contracts with client subscriptions are
    /// protected from eviction.
    pub fn sweep_expired_get_subscriptions(&self) -> Vec<(ContractKey, u64)> {
        // Delegate to hosting cache
        self.sweep_expired_hosting()
    }

    // ==================== Connection Pruning ====================

    /// Prune a peer connection.
    ///
    /// Returns orphaned transactions that need to be retried or failed.
    /// In the new lease-based subscription model, subscriptions are not tied to specific
    /// peers, so no subscription pruning is needed when a peer disconnects.
    pub async fn prune_connection(&self, peer: PeerId) -> PruneConnectionResult {
        use crate::tracing::DisconnectReason;

        tracing::debug!(%peer, "Removing connection");
        crate::node::network_status::record_peer_disconnected(peer.socket_addr());
        let orphaned_transactions = self
            .live_tx_tracker
            .prune_transactions_from_peer(peer.socket_addr());

        if !orphaned_transactions.is_empty() {
            tracing::debug!(
                %peer,
                orphaned_count = orphaned_transactions.len(),
                "Connection pruned with orphaned transactions"
            );
        }

        let min_ready = self.connection_manager.min_ready_connections;
        let was_ready = self.connection_manager.is_self_ready();

        // Capture connection duration before pruning
        let connection_duration_ms = self
            .connection_manager
            .get_connection_duration_ms(peer.socket_addr());

        // This case would be when a connection is being open, so peer location hasn't been recorded yet
        let Some(_loc) = self
            .connection_manager
            .prune_alive_connection(peer.socket_addr())
        else {
            return PruneConnectionResult {
                orphaned_transactions,
                became_unready: false,
            };
        };

        if let Some(event) = NetEventLog::disconnected_with_context(
            self,
            &peer,
            DisconnectReason::Pruned,
            connection_duration_ms,
            None, // bytes_sent not tracked yet
            None, // bytes_received not tracked yet
        ) {
            self.event_register
                .register_events(Either::Left(event))
                .await;
        }

        let is_ready = self.connection_manager.is_self_ready();
        let became_unready = min_ready > 0 && was_ready && !is_ready;

        PruneConnectionResult {
            orphaned_transactions,
            became_unready,
        }
    }

    async fn connection_maintenance(
        self: Arc<Self>,
        notifier: EventLoopNotificationsSender,
        live_tx_tracker: LiveTransactionTracker,
    ) -> anyhow::Result<()> {
        let is_gateway = self.is_gateway;
        tracing::info!(is_gateway, "Connection maintenance task starting");
        #[cfg(not(test))]
        const CHECK_TICK_DURATION: Duration = Duration::from_secs(60);
        #[cfg(test)]
        const CHECK_TICK_DURATION: Duration = Duration::from_secs(2);

        // Faster tick when below min_connections, so initial mesh formation
        // doesn't bottleneck on the 60-second steady-state interval.
        #[cfg(not(test))]
        const FAST_CHECK_TICK_DURATION: Duration = Duration::from_secs(5);
        #[cfg(test)]
        const FAST_CHECK_TICK_DURATION: Duration = Duration::from_secs(1);

        const REGENERATE_DENSITY_MAP_INTERVAL: Duration = Duration::from_secs(60);

        /// Base number of concurrent connection acquisition attempts (steady-state).
        const BASE_CONCURRENT_CONNECTIONS: usize = 3;

        let mut check_interval = tokio::time::interval(CHECK_TICK_DURATION);
        check_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        let mut refresh_density_map = tokio::time::interval(REGENERATE_DENSITY_MAP_INTERVAL);
        refresh_density_map.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

        // if the peer is just starting wait a bit before
        // we even attempt acquiring more connections
        tokio::time::sleep(Duration::from_secs(2)).await;

        let mut pending_conn_adds = BTreeSet::new();
        let mut last_backoff_cleanup = self.time_source.now();
        let mut last_health_check = self.time_source.now();
        let mut last_peer_cache_save = self.time_source.now();
        const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(300);
        // How often to snapshot the peer cache to disk.
        const PEER_CACHE_SAVE_INTERVAL: Duration = Duration::from_secs(30);
        const BACKOFF_CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
        /// Duration of zero ring connections before escalating recovery.
        /// Uses a shorter threshold initially (before first successful connection)
        /// so that cold-start failures after OS restart recover faster (#3737).
        const ISOLATION_ESCALATION_THRESHOLD: Duration = Duration::from_secs(120);
        const INITIAL_ISOLATION_ESCALATION_THRESHOLD: Duration = Duration::from_secs(30);
        /// Max time to hold a deferred swap drop before abandoning it.
        const DEFERRED_SWAP_DROP_TTL: Duration = Duration::from_secs(120);

        /// How often to probe a gateway for version discovery (#3677).
        #[cfg(not(test))]
        const GATEWAY_VERSION_PROBE_INTERVAL: Duration = Duration::from_secs(4 * 3600);
        #[cfg(test)]
        const GATEWAY_VERSION_PROBE_INTERVAL: Duration = Duration::from_secs(10);
        const GATEWAY_PROBE_JITTER_FACTOR: f64 = 0.2;

        // Deferred swap drops: (addr, queued_at) using time_source for
        // deterministic simulation support.
        let mut deferred_swap_drops: Vec<(SocketAddr, tokio::time::Instant)> = Vec::new();
        let mut zero_connections_since: Option<Instant> = None;
        // Track whether we've ever had ring connections. Before the first
        // successful connection, use a shorter isolation escalation threshold
        // for faster recovery from cold-start failures (#3737).
        let mut ever_had_connections = false;

        // Prior-tick cumulative per-peer wire-byte counts, keyed by the
        // transport socket address. Each maintenance tick we diff the current
        // `TRANSPORT_METRICS.per_peer_snapshot()` against this to derive the
        // bytes transferred *during the tick*, then feed that delta into the
        // topology meter so `adjust_topology` can make load-aware decisions
        // (#3453). Bounded: entries for peers absent from the latest snapshot
        // are pruned each tick, so this never outgrows the transport metrics
        // table (capped at `MAX_TRACKED_PEERS`).
        //
        // Seed with the current cumulative counts so the first maintenance
        // tick attributes only the bytes transferred *during* that first
        // interval, not the entire history accumulated before the loop
        // started (which would over-count any pre-maintenance bootstrap
        // traffic into a single inflated first sample).
        let mut prev_peer_bandwidth: HashMap<SocketAddr, (u64, u64)> =
            crate::transport::metrics::TRANSPORT_METRICS
                .per_peer_snapshot()
                .into_iter()
                .map(|(addr, sent, recv)| (addr, (sent, recv)))
                .collect();
        // Time of the previous bandwidth feed (the start of the interval whose
        // byte delta the next feed reports). Seeded with "now" so the first
        // feed's delta is divided by the real first-interval length rather than
        // flooring to the meter's 1s minimum (#3453 review — see
        // `feed_peer_bandwidth_to_meter`).
        let mut prev_bandwidth_tick = self.time_source.now();

        // Gateway version probe: random initial delay to prevent thundering herd.
        // The loop guard (`!is_gateway`) ensures gateways never probe themselves.
        let initial_probe_delay_secs =
            GlobalRng::random_u64() % GATEWAY_VERSION_PROBE_INTERVAL.as_secs();
        let mut next_gateway_probe =
            self.time_source.now() + Duration::from_secs(initial_probe_delay_secs);

        // Adaptive fast-tick backoff: increase the fast-tick interval when
        // connection count stops growing, to avoid hammering the network
        // with CONNECTs indefinitely (#3578).
        let mut last_conn_count: usize = 0;
        let mut no_progress_ticks: u32 = 0;
        // After this many consecutive no-progress ticks, start doubling
        // the fast-tick interval.
        const FAST_TICK_BACKOFF_THRESHOLD: u32 = 6; // 30s at 5s/tick

        // Maximum fast-tick multiplier: caps backoff at the normal tick rate.
        // Derived from tick ratio so invariant holds in both prod and test cfg.
        const MAX_FAST_TICK_MULTIPLIER: u32 =
            (CHECK_TICK_DURATION.as_secs() / FAST_CHECK_TICK_DURATION.as_secs()) as u32;

        // Suspend/resume detection.
        //
        // We compare two clocks that differ only under OS suspend:
        //
        //   * `boot_time::Instant` → CLOCK_BOOTTIME on Linux. Advances while
        //     the machine is suspended.
        //   * `std::time::Instant` → CLOCK_MONOTONIC on Linux. Does *not*
        //     advance while the machine is suspended.
        //
        // The delta between them across a loop iteration is the amount of
        // time the machine was suspended. Using just `boot_elapsed` alone
        // also counts scheduler stalls and virtual-time jumps in simulation
        // tests (`tokio::time::start_paused(true)`), which previously caused
        // false positives: the Apr 2026 nightly logs showed
        // `boot_elapsed_secs=139` on CI runners under load, which tripped the
        // 30s test-only threshold and fired `DropAllConnections` mid-test,
        // wiping in-flight GET ops and tripping the `debug_assert!` in
        // `GetMsg::Request` handling. Comparing against a monotonic baseline
        // eliminates that: a heavy-CPU stall advances *both* clocks equally
        // (delta ≈ 0) and doesn't trip the detector, while a real suspend
        // advances only boot time (delta ≈ suspend duration) and does.
        let mut last_boot_time = boot_time::Instant::now();
        let mut last_mono_time = WallClockInstant::now();
        // 2x the check tick is plenty of headroom to tell a real suspend
        // (minutes) from the sub-tick jitter that a healthy monotonic clock
        // can still exhibit relative to CLOCK_BOOTTIME.
        const SUSPEND_DETECTION_THRESHOLD: Duration = CHECK_TICK_DURATION.saturating_mul(2);

        let mut this_peer = None;
        'maintenance: loop {
            // Update clock tracking at the top of every iteration (including
            // early-continue paths) so elapsed time doesn't accumulate during
            // startup. The four clock operations below MUST stay back-to-back
            // with no intervening `.await` — any suspension point between
            // them lets the two clocks drift relative to one another for
            // reasons unrelated to suspend/resume, which would poison the
            // delta in `classify_suspend_jump` below. Keep them as a block.
            let boot_elapsed = last_boot_time.elapsed();
            let mono_elapsed = last_mono_time.elapsed();
            last_boot_time = boot_time::Instant::now();
            last_mono_time = WallClockInstant::now();
            let suspend_jump = classify_suspend_jump(boot_elapsed, mono_elapsed);
            // Diagnostic: a small monotonic-ahead skew is normal non-atomic
            // read jitter; a large one would indicate a virtualization TSC
            // anomaly or a monotonic clock going backwards. Surface it so
            // nobody has to rediscover the clock-ordering assumption the
            // hard way.
            if let Some(skew) = mono_elapsed.checked_sub(boot_elapsed)
                && skew > Duration::from_millis(100)
            {
                tracing::warn!(
                    mono_ahead_ms = skew.as_millis() as u64,
                    boot_elapsed_ms = boot_elapsed.as_millis() as u64,
                    mono_elapsed_ms = mono_elapsed.as_millis() as u64,
                    "connection_maintenance: monotonic clock is significantly \
                     ahead of boot clock — possible virtualization TSC anomaly \
                     or monotonic clock regression; suspend detection is \
                     saturated to zero for this iteration"
                );
            }

            let op_manager = match self.op_manager_state() {
                OpManagerState::Live(op_manager) => op_manager,
                OpManagerState::NotAttached => {
                    // Still in the startup window before attach_op_manager;
                    // wait for the owner to wire itself up.
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    continue;
                }
                OpManagerState::Detached => {
                    // The OpManager was attached and then dropped (node
                    // shutdown). Stop the maintenance work instead of spinning
                    // forever on a Weak that can never upgrade again (#3308).
                    // We do NOT `return Ok(())` here: this task is registered
                    // with `BackgroundTaskMonitor`, whose `wait_for_any_exit`
                    // treats any clean return as a fatal "task exited
                    // unexpectedly". Per the #4292 convention, a monitored
                    // long-lived task whose work has ended must hand a
                    // non-completing future to the monitor — so we break out
                    // of the loop and park below.
                    tracing::info!(
                        is_gateway,
                        "OpManager dropped; connection maintenance loop ending (parking)"
                    );
                    break 'maintenance;
                }
            };
            let Some(this_addr) = &this_peer else {
                let Some(addr) = self.connection_manager.get_own_addr() else {
                    tokio::time::sleep(Duration::from_secs(1)).await;
                    continue;
                };
                this_peer = Some(addr);
                continue;
            };
            // avoid connecting to the same peer multiple times
            let mut skip_list = HashSet::new();
            skip_list.insert(*this_addr);

            // Resets both connection (location-based) and gateway (address-based)
            // backoff state, and clears stale pending reservations for gateways.
            // Used during isolation recovery to ensure all gateways are retryable
            // when the node has zero ring connections (#3319).
            // Wakes any task sleeping on gateway backoff
            // (`initial_join_procedure`) so it can retry immediately.
            let reset_all_backoff = || {
                self.reset_all_connection_backoff();
                op_manager.gateway_backoff.lock().clear();
                op_manager.gateway_backoff_cleared.notify_waiters();
                // Also clear stale pending reservations for gateways — without this,
                // gateways appear "connected/pending" via has_connection_or_pending()
                // even after backoff is reset, blocking retry attempts (#3319).
                let gateway_addrs: Vec<_> = op_manager
                    .configured_gateways
                    .iter()
                    .filter_map(|gw| gw.socket_addr())
                    .collect();
                self.connection_manager
                    .clear_pending_reservations_for(&gateway_addrs);
            };

            // Suspend/resume detection: if boot time advanced much more than
            // monotonic time, the machine was suspended. Both `suspend_jump`
            // and the underlying clock reads happen at the top of the loop so
            // they include early-continue time.
            if suspend_jump > SUSPEND_DETECTION_THRESHOLD {
                tracing::warn!(
                    boot_elapsed_secs = boot_elapsed.as_secs(),
                    mono_elapsed_secs = mono_elapsed.as_secs(),
                    suspend_jump_secs = suspend_jump.as_secs(),
                    "Detected suspend/resume (boot-time jump) — dropping all connections and clearing state"
                );
                reset_all_backoff();
                // Clear recently-failed addresses since they may be reachable again.
                self.connection_manager.cleanup_all_failed_addrs();
                // Drop all connections (including transient gateway connections).
                // After suspend, transport sockets are dead but connection entries
                // persist as zombies — keepalive tasks exit on socket error but
                // don't trigger connection cleanup. The bootstrap loop then sends
                // CONNECT messages into dead sockets that never reach the gateway.
                notifier
                    .notifications_sender
                    .send(Either::Right(crate::message::NodeEvent::DropAllConnections))
                    .await
                    .map_err(|error| {
                        tracing::debug!(?error, "Failed to send DropAllConnections");
                        error
                    })?;
                zero_connections_since = None;
            }

            // Periodic cleanup of expired backoff entries
            if last_backoff_cleanup.elapsed() > BACKOFF_CLEANUP_INTERVAL {
                self.cleanup_connection_backoff();
                last_backoff_cleanup = self.time_source.now();
            }

            // Clean up stale pending reservations to prevent permanent isolation
            // when CONNECT operations fail to complete cleanly.
            let stale_removed = self.connection_manager.cleanup_stale_reservations();
            if stale_removed > 0 {
                tracing::warn!(
                    stale_removed,
                    "Cleaned up stale reservations and orphaned location entries"
                );
            }

            // Capture a single `now` for all TTL/cleanup checks in this tick so that
            // they all see the same moment rather than drifting across calls. Using
            // `self.time_source` (rather than `Instant::now()` directly) allows tests
            // to supply a `SharedMockTimeSource` without pausing the whole tokio runtime.
            let tick_now = self.time_source.now();

            // Expire old NAT traversal failure entries
            self.connection_manager.cleanup_stale_failed_addrs();

            // Expire acceptor reliability entries for peers whose TTL has elapsed
            self.connection_manager
                .cleanup_expired_acceptor_stats(tick_now);

            // Clean up expired transient connections
            let expired_transients = self.connection_manager.cleanup_expired_transients();
            if expired_transients > 0 {
                tracing::debug!(
                    expired_transients,
                    "Cleaned up expired transient connections"
                );
            }

            // Periodic peer health check: evict peers with sustained routing failures.
            if last_health_check.elapsed() > HEALTH_CHECK_INTERVAL {
                last_health_check = self.time_source.now();
                let current_ring = self.connection_manager.connection_count();
                let unhealthy = self
                    .connection_manager
                    .peer_health
                    .lock()
                    .unhealthy_peers(self.connection_manager.min_connections, current_ring);
                for addr in unhealthy {
                    tracing::warn!(
                        peer = %addr,
                        "Evicting unhealthy peer (sustained routing failures)"
                    );
                    if let Err(e) = notifier
                        .notifications_sender
                        .send(Either::Right(crate::message::NodeEvent::DropConnection(
                            addr,
                        )))
                        .await
                    {
                        tracing::debug!(error = ?e, "Failed to send DropConnection for unhealthy peer");
                    }
                }
            }

            // Periodically save peer cache for fast reconnection after restart.
            if last_peer_cache_save.elapsed() > PEER_CACHE_SAVE_INTERVAL {
                last_peer_cache_save = self.time_source.now();
                if let Some(ref dir) = self.peer_cache_dir {
                    let cache = peer_cache::PeerCache::snapshot_from(
                        &self.connection_manager,
                        self.time_source.as_ref(),
                    );
                    if !cache.peers.is_empty() {
                        if let Err(e) = cache.save(dir) {
                            tracing::warn!(error = %e, "Failed to save peer cache");
                        }
                    }
                }
            }

            // Isolation recovery: when we have zero ring connections for too long,
            // reset all backoff state so we can retry aggressively (#2928).
            let current_conn_count = self.connection_manager.connection_count();
            // Expose to update check task for version mismatch decisions (#3204).
            crate::transport::set_open_connection_count(current_conn_count);
            if current_conn_count == 0 {
                // Use shorter threshold before first successful connection so
                // cold-start failures (e.g., after OS restart) recover in ~30s
                // instead of ~120s. Once the node has connected at least once,
                // use the steady-state threshold. See #3737.
                let threshold = if ever_had_connections {
                    ISOLATION_ESCALATION_THRESHOLD
                } else {
                    INITIAL_ISOLATION_ESCALATION_THRESHOLD
                };
                if let Some(since) = zero_connections_since {
                    if since.elapsed() > threshold {
                        tracing::warn!(
                            is_gateway,
                            isolated_for_secs = since.elapsed().as_secs(),
                            threshold_secs = threshold.as_secs(),
                            ever_connected = ever_had_connections,
                            "Node isolated with zero ring connections — resetting all backoff state"
                        );
                        reset_all_backoff();
                        zero_connections_since = Some(self.time_source.now());
                    }
                } else {
                    zero_connections_since = Some(self.time_source.now());
                    tracing::warn!(
                        is_gateway,
                        "Zero ring connections detected — starting isolation timer"
                    );
                }
            } else if zero_connections_since.take().is_some() {
                ever_had_connections = true;
                tracing::info!(
                    connections = current_conn_count,
                    "Recovered from zero-connection state"
                );
            }

            // Periodic gateway version probe: initiate a CONNECT to a gateway so the
            // transport handshake exchanges version info, even when the ring is full (#3677).
            //
            // The combined predicate (`!is_gateway && !configured_gateways.is_empty() &&
            // now >= next_probe`) is extracted into `should_probe_gateway` so each branch
            // is unit-testable. Without the empty-gateways guard merged into the same
            // condition, the index/modulo on the next line would panic.
            if should_probe_gateway(
                is_gateway,
                !op_manager.configured_gateways.is_empty(),
                self.time_source.now(),
                next_gateway_probe,
            ) {
                let gw_index =
                    GlobalRng::random_u64() as usize % op_manager.configured_gateways.len();
                let gateway = &op_manager.configured_gateways[gw_index];

                if let Err(e) =
                    crate::operations::connect::gateway_version_probe(gateway, &op_manager).await
                {
                    tracing::debug!(
                        error = %e,
                        gateway = %gateway,
                        "Gateway version probe failed, will retry next cycle"
                    );
                }

                let base_secs = GATEWAY_VERSION_PROBE_INTERVAL.as_secs() as f64;
                let jitter_range = base_secs * GATEWAY_PROBE_JITTER_FACTOR;
                let uniform_01 = (GlobalRng::random_u64() as f64) / (u64::MAX as f64);
                let jittered_secs = base_secs + jitter_range * (2.0 * uniform_01 - 1.0);
                next_gateway_probe =
                    self.time_source.now() + Duration::from_secs_f64(jittered_secs.max(1.0));
            }

            // Gateway bootstrap fallback: at zero connections, acquire_new always
            // fails (no routing candidates). Connect to gateways directly (#3219).
            //
            // Note: initial_join_procedure may also be attempting gateway connections
            // concurrently. is_not_connected provides best-effort dedup via pending
            // reservations (should_accept creates one on each join_ring_request), so
            // the second caller typically sees the gateway as "pending" and skips it.
            // Occasional duplicate CONNECTs are harmless — the connect state machine
            // handles them gracefully.
            if current_conn_count == 0 && !op_manager.configured_gateways.is_empty() {
                let eligible: Vec<_> = {
                    let backoff = op_manager.gateway_backoff.lock();
                    self.is_not_connected(op_manager.configured_gateways.iter())
                        .filter(|gw| {
                            gw.socket_addr()
                                .map(|addr| !backoff.is_in_backoff(addr))
                                .unwrap_or(false) // skip gateways without addresses
                        })
                        .cloned()
                        .collect()
                };

                if eligible.is_empty() {
                    tracing::debug!(
                        total_gateways = op_manager.configured_gateways.len(),
                        "Zero connections — all gateways connected/pending or in backoff"
                    );
                } else {
                    let attempt_count = eligible.len().min(BASE_CONCURRENT_CONNECTIONS);
                    tracing::info!(
                        eligible = eligible.len(),
                        attempting = attempt_count,
                        "Zero connections — attempting gateway bootstrap"
                    );
                    for gw in eligible.iter().take(BASE_CONCURRENT_CONNECTIONS) {
                        match crate::operations::connect::join_ring_request(gw, &op_manager, None)
                            .await
                        {
                            Ok(()) => tracing::debug!(gateway = %gw, "Gateway bootstrap initiated"),
                            Err(e) => {
                                tracing::warn!(gateway = %gw, error = %e, "Gateway bootstrap failed")
                            }
                        }
                    }
                }
            }

            // Scale concurrent connection limit based on deficit to min_connections.
            // During bootstrap (far below min_connections), allow more parallel attempts
            // to avoid stalling when slots fill with slow/timing-out transactions.
            let max_concurrent = calculate_max_concurrent_connections(
                current_conn_count,
                self.connection_manager.min_connections,
            );

            // Drain pending connections, initiating multiple attempts per tick
            // (up to max_concurrent) for faster mesh formation. Counts only this
            // node's own in-flight acquisitions, NOT CONNECTs it is relaying for
            // others (#4348).
            let mut active_count = live_tx_tracker.active_acquisition_transaction_count();
            // Under-min nodes bypass per-target backoff to escape the straggler
            // trap (#4348); see `should_respect_location_backoff`.
            let respect_backoff = should_respect_location_backoff(
                current_conn_count,
                self.connection_manager.min_connections,
            );
            while let Some(ideal_location) = pending_conn_adds.pop_first() {
                if respect_backoff && self.is_in_connection_backoff(ideal_location) {
                    tracing::debug!(
                        target_location = %ideal_location,
                        "Skipping connection attempt - target in backoff"
                    );
                    // Intentionally not re-queued: adjust_topology will re-request
                    // this location on the next cycle if still below min_connections.
                    continue;
                }
                if active_count >= max_concurrent {
                    tracing::debug!(
                        active_connections = active_count,
                        max_concurrent,
                        target_location = %ideal_location,
                        "At max concurrent connections, re-queuing location"
                    );
                    pending_conn_adds.insert(ideal_location);
                    break;
                }
                tracing::debug!(
                    active_connections = active_count,
                    max_concurrent,
                    target_location = %ideal_location,
                    "Attempting to acquire new connection"
                );
                let tx = self
                    .acquire_new(
                        ideal_location,
                        &skip_list,
                        &notifier,
                        &live_tx_tracker,
                        &op_manager,
                    )
                    .await
                    .map_err(|error| {
                        tracing::error!(
                            ?error,
                            "FATAL: Connection maintenance task failed - shutting down"
                        );
                        error
                    })?;
                if tx.is_none() {
                    let conns = self.connection_manager.connection_count();
                    tracing::debug!(
                        connections = conns,
                        target_location = %ideal_location,
                        "acquire_new returned None - likely no peers to query through"
                    );
                    // Don't record a backoff against the target location here.
                    // acquire_new returning None means we have insufficient routing
                    // candidates locally — the target location itself is fine.
                    // Backing off the target would block future attempts when we
                    // gain more connections and could actually route to it.
                    // adjust_topology will re-request this location on the next tick.
                } else {
                    active_count += 1;
                    tracing::info!(
                        active_connections = active_count,
                        "Successfully initiated connection acquisition"
                    );
                }
            }

            let current_connections = self.connection_manager.connection_count();
            let pending_connection_targets = pending_conn_adds.len();
            let peers = self.connection_manager.get_connections_by_location();
            let connections_considered: usize = peers.values().map(|c| c.len()).sum();

            let mut neighbor_locations: BTreeMap<_, Vec<_>> = peers
                .iter()
                .map(|(loc, conns)| {
                    let conns: Vec<_> = conns
                        .iter()
                        .filter(|conn| {
                            conn.location
                                .socket_addr()
                                .map(|addr| !live_tx_tracker.has_live_connection(addr))
                                .unwrap_or(true)
                        })
                        .cloned()
                        .collect();
                    (*loc, conns)
                })
                .filter(|(_, conns)| !conns.is_empty())
                .collect();

            if neighbor_locations.is_empty() && connections_considered > 0 {
                tracing::debug!(
                    current_connections,
                    connections_considered,
                    live_tx_peers = live_tx_tracker.len(),
                    "Neighbor filtering removed all candidates; using all connections"
                );

                neighbor_locations = peers
                    .iter()
                    .map(|(loc, conns)| (*loc, conns.clone()))
                    .filter(|(_, conns)| !conns.is_empty())
                    .collect();
            }

            if current_connections > self.connection_manager.max_connections {
                // When over capacity, consider all connections for removal regardless of live_tx filter.
                neighbor_locations = peers.clone();
            }

            tracing::debug!(
                current_connections,
                candidates = peers.len(),
                live_tx_peers = live_tx_tracker.len(),
                "Evaluating topology maintenance"
            );

            // Feed the per-peer bandwidth measured since the previous tick into
            // the topology meter BEFORE adjust_topology reads it. Without this
            // the meter sees no peer-attributed bandwidth samples in production,
            // so `calculate_usage_proportion` always reports ~0% usage and
            // `adjust_topology` perpetually takes the "add connections" branch —
            // the load-aware add/hold/remove logic never engages (#3453).
            let bandwidth_tick_now = self.time_source.now();
            feed_peer_bandwidth_to_meter(
                &self.connection_manager,
                &crate::transport::metrics::TRANSPORT_METRICS.per_peer_snapshot(),
                &mut prev_peer_bandwidth,
                // Timestamp this interval's byte delta at the PREVIOUS tick
                // (interval start) so the meter divides by the real interval
                // length, not the 1s floor. See feed_peer_bandwidth_to_meter.
                prev_bandwidth_tick,
            );
            prev_bandwidth_tick = bandwidth_tick_now;

            let adjustment = self
                .connection_manager
                .topology_manager
                .write()
                .adjust_topology(
                    &neighbor_locations,
                    &self.connection_manager.own_location().location(),
                    self.time_source.now(),
                    current_connections,
                );

            tracing::debug!(
                adjustment = ?adjustment,
                current_connections,
                is_gateway,
                pending_adds = pending_connection_targets,
                "Topology adjustment result"
            );

            match adjustment {
                TopologyAdjustment::AddConnections(target_locs) => {
                    let allowed = calculate_allowed_connection_additions(
                        current_connections,
                        pending_connection_targets,
                        self.connection_manager.min_connections,
                        self.connection_manager.max_connections,
                        target_locs.len(),
                    );

                    if allowed == 0 {
                        tracing::debug!(
                            requested = target_locs.len(),
                            current_connections,
                            pending = pending_connection_targets,
                            min_connections = self.connection_manager.min_connections,
                            max_connections = self.connection_manager.max_connections,
                            "Skipping queuing new connection targets – backlog already satisfies capacity constraints"
                        );
                    } else {
                        let total_pending_after = pending_connection_targets + allowed;
                        tracing::debug!(
                            requested = target_locs.len(),
                            allowed,
                            total_pending_after,
                            "Queuing additional connection targets"
                        );
                        pending_conn_adds.extend(target_locs.into_iter().take(allowed));
                    }
                }
                TopologyAdjustment::RemoveConnections(should_disconnect_peers) => {
                    for peer in should_disconnect_peers {
                        if let Some(addr) = peer.socket_addr() {
                            notifier
                                .notifications_sender
                                .send(Either::Right(crate::message::NodeEvent::DropConnection(
                                    addr,
                                )))
                                .await
                                .map_err(|error| {
                                    tracing::debug!(
                                        error = ?error,
                                        "Shutting down connection maintenance task"
                                    );
                                    error
                                })?;
                        }
                    }
                }
                TopologyAdjustment::SwapConnection {
                    remove,
                    add_location,
                } => {
                    // Connect-first swap: defer the drop until the replacement
                    // connects (connection_count > min_connections), preventing
                    // undershoot that would block future swaps. Deferred drops
                    // expire after DEFERRED_SWAP_DROP_TTL if replacement fails.
                    if let Some(addr) = remove.socket_addr() {
                        tracing::info!(
                            remove_peer = %remove,
                            add_target = %add_location,
                            "Topology swap: queuing replacement connection (drop deferred)"
                        );
                        pending_conn_adds.insert(add_location);
                        // Deduplicate: don't queue the same peer twice if
                        // consecutive swaps select it before the first executes.
                        if !deferred_swap_drops.iter().any(|(a, _)| *a == addr) {
                            deferred_swap_drops.push((addr, tick_now));
                        }
                    } else {
                        tracing::warn!(
                            remove_peer = %remove,
                            "Topology swap skipped: peer has no socket address"
                        );
                    }
                }
                TopologyAdjustment::NoChange => {}
            }

            // Execute deferred swap drops: only drop as many peers as we
            // have headroom above min_connections to avoid undershooting.
            // Expire stale entries whose replacement never connected.
            {
                let before_len = deferred_swap_drops.len();
                deferred_swap_drops.retain(|(_, queued_at)| {
                    tick_now.saturating_duration_since(*queued_at) < DEFERRED_SWAP_DROP_TTL
                });
                let expired = before_len - deferred_swap_drops.len();
                if expired > 0 {
                    tracing::debug!(
                        expired,
                        "Deferred swap drops expired (replacement never connected)"
                    );
                }

                if !deferred_swap_drops.is_empty() {
                    let fresh_count = self.connection_manager.connection_count();
                    let min_conn = self.connection_manager.min_connections;
                    let n_to_drop = deferred_swap_drops_to_execute(
                        fresh_count,
                        min_conn,
                        deferred_swap_drops.len(),
                    );
                    for (addr, _) in deferred_swap_drops.drain(..n_to_drop) {
                        tracing::info!(
                            peer = %addr,
                            connections = fresh_count,
                            "Executing deferred swap drop (replacement connected)"
                        );
                        notifier
                            .notifications_sender
                            .send(Either::Right(crate::message::NodeEvent::DropConnection(
                                addr,
                            )))
                            .await
                            .map_err(|error| {
                                tracing::debug!(
                                    ?error,
                                    "Shutting down connection maintenance task"
                                );
                                error
                            })?;
                    }
                }
            }

            let needs_fast_tick = current_connections < self.connection_manager.min_connections;

            if needs_fast_tick {
                // Adaptive backoff: reset on any connection count change
                // (gain OR loss), otherwise slow down. A loss means topology
                // changed and we should re-enter aggressive mode.
                if current_connections != last_conn_count {
                    no_progress_ticks = 0;
                } else {
                    no_progress_ticks = no_progress_ticks.saturating_add(1);
                }
                last_conn_count = current_connections;

                let multiplier = if no_progress_ticks <= FAST_TICK_BACKOFF_THRESHOLD {
                    1u32
                } else {
                    let excess = no_progress_ticks - FAST_TICK_BACKOFF_THRESHOLD;
                    2u32.saturating_pow(excess).min(MAX_FAST_TICK_MULTIPLIER)
                };
                // Apply ±20% jitter to prevent synchronized CONNECT bursts
                // across peers that bootstrapped simultaneously.
                let jitter: f64 = crate::config::GlobalRng::random_range(0.8..=1.2);
                let adaptive_duration =
                    FAST_CHECK_TICK_DURATION.mul_f64(multiplier as f64 * jitter);

                if multiplier > 1 {
                    tracing::debug!(
                        current_connections,
                        min_connections = self.connection_manager.min_connections,
                        no_progress_ticks,
                        tick_interval_secs = adaptive_duration.as_secs(),
                        "Fast-tick backed off due to no connection progress"
                    );
                }

                // Uses sleep() instead of the check_interval so we don't need a
                // second Interval object. We reset check_interval on transition
                // back to steady-state to avoid an immediate burst tick.
                crate::deterministic_select! {
                  _ = refresh_density_map.tick() => {
                    self.refresh_density_request_cache();
                  },
                  _ = tokio::time::sleep(adaptive_duration) => {},
                }
            } else {
                // Reached min_connections: reset backoff for next time.
                no_progress_ticks = 0;
                last_conn_count = current_connections;

                // Reset the interval on transition from fast to normal tick so
                // accumulated missed ticks don't cause an immediate burst.
                check_interval.reset();
                crate::deterministic_select! {
                  _ = refresh_density_map.tick() => {
                    self.refresh_density_request_cache();
                  },
                  _ = check_interval.tick() => {},
                }
            }
        }

        // Intentional teardown parking (#4292): the loop only breaks once the
        // OpManager has been dropped (node shutdown), so there is no more work
        // to do. This task is registered with `BackgroundTaskMonitor`; parking
        // on a never-resolving future — rather than returning `Ok(())` — keeps
        // `wait_for_any_exit` from misreading orderly teardown as a fatal
        // background-task exit. The unreachable `Ok(())` after it keeps the
        // `anyhow::Result<()>` signature.
        std::future::pending::<()>().await;
        Ok(())
    }

    #[tracing::instrument(level = "debug", skip(self, notifier, live_tx_tracker, op_manager), fields(peer = %self.connection_manager.pub_key))]
    async fn acquire_new(
        &self,
        ideal_location: Location,
        skip_list: &HashSet<SocketAddr>,
        notifier: &EventLoopNotificationsSender,
        live_tx_tracker: &LiveTransactionTracker,
        op_manager: &Arc<OpManager>,
    ) -> anyhow::Result<Option<Transaction>> {
        let current_connections = self.connection_manager.connection_count();
        let is_gateway = self.is_gateway;

        tracing::debug!(
            current_connections,
            is_gateway,
            target_location = %ideal_location,
            "acquire_new: attempting to find peer to query"
        );

        let query_target = {
            // The router read-lock is only needed for the single
            // `select_k_best_peers_with_telemetry` call; routing_candidates
            // takes its own connection-manager locks. Acquire late to keep
            // the critical section short (clippy: `significant_drop_tightening`).
            let num_connections = self.connection_manager.num_connections();
            tracing::debug!(
                target_location = %ideal_location,
                num_connections,
                skip_list_size = skip_list.len(),
                self_addr = ?self.connection_manager.get_own_addr(),
                "Looking for peer to route through"
            );
            // CONNECT operations bypass readiness gating — peers need to route
            // through ANY ring connection to acquire new connections, even if those
            // connections haven't advertised readiness yet. Without this, peers get
            // stuck below the readiness threshold: they can't initiate CONNECTs
            // because routing() filters out all their "not ready" connections,
            // but they can't become ready without more connections.
            let candidates = self.connection_manager.routing_candidates(
                ideal_location,
                None,
                skip_list,
                false, // bypass readiness — this is a CONNECT for connection acquisition
            );
            let selected = if !candidates.is_empty() {
                let (selected, _) = self.router.read().select_k_best_peers_with_telemetry(
                    candidates.iter(),
                    ideal_location,
                    1,
                );
                selected.into_iter().next().cloned()
            } else {
                None
            };
            if let Some(target) = selected {
                tracing::debug!(
                    query_target = %target,
                    target_location = %ideal_location,
                    "connection_maintenance selected routing target"
                );
                target
            } else {
                tracing::warn!(
                    current_connections,
                    is_gateway,
                    target_location = %ideal_location,
                    "acquire_new: no routing candidates found - cannot find peer to query"
                );
                return Ok(None);
            }
        };

        let joiner = self.connection_manager.own_location();
        tracing::debug!(
            this_peer = %joiner,
            query_target_peer = %query_target,
            target_location = %ideal_location,
            "Sending connect request via connection_maintenance"
        );

        // Driver computes ttl / target_connections / exclude_addrs internally
        // (see start_client_connect at op_ctx_task.rs). acquire_new only
        // needs to allocate the joiner tx, register it with
        // live_tx_tracker, and spawn the driver task.
        let _ = notifier;
        let tx = Transaction::new::<crate::operations::connect::ConnectMsg>();

        let gateway_addr = match query_target.socket_addr() {
            Some(addr) => addr,
            None => {
                tracing::warn!(
                    target_location = %ideal_location,
                    "acquire_new: selected query target has no socket address, skipping spawn"
                );
                return Ok(None);
            }
        };

        // Register tx with the live transaction tracker BEFORE spawning the
        // driver. Otherwise the driver's first Response could land on the
        // bypass before the registration completes. (The acquisition-throttle
        // registration is done inside `start_client_connect`, shared by every
        // self-initiated CONNECT — ring acquisition, gateway join, and probe.)
        live_tx_tracker.add_transaction(gateway_addr, tx);

        let op_manager_spawn = op_manager.clone();
        let gateway = query_target;
        let joiner_for_driver = joiner;
        GlobalExecutor::spawn(async move {
            if let Err(err) = crate::operations::connect::op_ctx_task::start_client_connect(
                tx,
                gateway,
                gateway_addr,
                &op_manager_spawn,
                joiner_for_driver,
                ideal_location,
                None,
            )
            .await
            {
                tracing::debug!(
                    %tx,
                    %err,
                    "acquire_new: CONNECT driver completed with error"
                );
            }
        });

        tracing::debug!(tx = %tx, "Connect request sent");
        Ok(Some(tx))
    }

    /// Register a topology snapshot for this peer with the global registry.
    ///
    /// This should be called periodically during simulation tests to enable
    /// topology validation. The snapshot captures the current subscription
    /// state for all contracts.
    #[cfg(any(test, feature = "testing"))]
    #[allow(dead_code)] // Used by SimNetwork tests
    pub fn register_topology_snapshot(&self, network_name: &str) {
        let Some(peer_addr) = self.connection_manager.get_own_addr() else {
            return;
        };
        // Use get_stored_location() for consistency with set_upstream distance check.
        let location = self
            .connection_manager
            .get_stored_location()
            .map(|l| l.as_f64())
            .unwrap_or(0.0);

        let snapshot = self
            .hosting_manager
            .generate_topology_snapshot(peer_addr, location);
        topology_registry::register_topology_snapshot(network_name, snapshot);
    }

    /// Get a topology snapshot for this peer without registering it.
    #[cfg(any(test, feature = "testing"))]
    #[allow(dead_code)] // Used by SimNetwork tests
    pub fn get_topology_snapshot(&self) -> Option<topology_registry::TopologySnapshot> {
        let peer_addr = self.connection_manager.get_own_addr()?;
        // Use get_stored_location() for consistency with set_upstream distance check.
        let location = self
            .connection_manager
            .get_stored_location()
            .map(|l| l.as_f64())
            .unwrap_or(0.0);

        Some(
            self.hosting_manager
                .generate_topology_snapshot(peer_addr, location),
        )
    }
}

/// Calculate the maximum number of concurrent connection acquisition attempts.
///
/// During bootstrap (below `min_connections`), scales up from the base to allow
/// more parallel attempts, preventing stalls when slots fill with slow transactions.
/// Once at or above `min_connections`, returns the base value.
fn calculate_max_concurrent_connections(
    current_connections: usize,
    min_connections: usize,
) -> usize {
    /// Base concurrent connection slots (steady-state).
    const BASE: usize = 3;
    /// How many missing connections map to one additional concurrent slot.
    const CONNECTIONS_PER_EXTRA_SLOT: usize = 3;

    if current_connections >= min_connections {
        return BASE;
    }
    let deficit = min_connections - current_connections;
    // Cap at half of min_connections, floored at BASE.
    let bootstrap_cap = (min_connections / 2).max(BASE);
    (BASE + deficit / CONNECTIONS_PER_EXTRA_SLOT).min(bootstrap_cap)
}

/// Whether `connection_maintenance` should honor per-target-location connection
/// backoff for a node with `current_connections` open connections.
///
/// Backoff is honored only once the node has reached `min_connections`. A node
/// still below min MUST keep probing even recently-rejected ring regions: its
/// under-connection is a local capacity problem, not the target's fault (cf. the
/// ring.md "Backoff Target Must Match Failure Cause" rule), and the 30s→600s
/// location backoff stamped on a `Rejected` would otherwise trap a poorly
/// positioned node permanently below min — the straggler tail in #4348. This
/// mirrors the zero-connection re-bootstrap escape, extended to the whole
/// under-min regime; at/above min, steady-state backoff is unchanged.
///
/// Storm safety: bypassing location backoff below min does NOT let a stuck node
/// hammer indefinitely. The maintenance loop's adaptive fast-tick backoff still
/// stretches the retry cadence toward the steady ~60s `CHECK_TICK` after
/// consecutive no-progress ticks (connection count not changing), per-tick
/// attempts remain bounded by [`calculate_max_concurrent_connections`], and the
/// separate gateway re-bootstrap backoff (`gateway_backoff`) is untouched.
fn should_respect_location_backoff(current_connections: usize, min_connections: usize) -> bool {
    current_connections >= min_connections
}

fn calculate_allowed_connection_additions(
    current_connections: usize,
    pending_connections: usize,
    min_connections: usize,
    max_connections: usize,
    requested: usize,
) -> usize {
    if requested == 0 {
        return 0;
    }

    let effective_connections = current_connections.saturating_add(pending_connections);
    if effective_connections >= max_connections {
        return 0;
    }

    let mut available_capacity = max_connections - effective_connections;

    if current_connections < min_connections {
        let deficit_to_min = min_connections.saturating_sub(effective_connections);
        available_capacity = available_capacity.min(deficit_to_min);
    }

    available_capacity.min(requested)
}

/// Compute how many deferred swap-drops can be executed this tick.
///
/// A deferred drop is safe to execute only when a replacement peer has actually
/// connected.  The guard: `current_connections > min_connections + pending_drops`
/// — meaning we have at least one extra connection above what's needed to cover
/// all pending drops plus the minimum.  Each drop we commit to reduces the
/// effective count by one, so we re-evaluate per element.
///
/// Drops are sent as async events; `connection_count()` won't reflect them until
/// the events are processed.  We track the decrement locally instead.
fn deferred_swap_drops_to_execute(
    current_connections: usize,
    min_connections: usize,
    pending_drops: usize,
) -> usize {
    let mut effective_count = current_connections;
    let mut n_to_drop = 0usize;
    for _ in 0..pending_drops {
        let remaining_pending = pending_drops - n_to_drop;
        if effective_count > min_connections.saturating_add(remaining_pending) {
            n_to_drop += 1;
            effective_count = effective_count.saturating_sub(1);
        } else {
            break;
        }
    }
    n_to_drop
}

/// Amount of wall time the machine was suspended across a maintenance loop
/// iteration, derived from the two clocks the caller samples at the top of
/// each loop pass.
///
/// ## Clock pairing
///
/// On Linux/Android/openBSD/L4Re, `boot_time::Instant` uses `CLOCK_BOOTTIME`
/// (advances during suspend) and `std::time::Instant` uses `CLOCK_MONOTONIC`
/// (does not advance during suspend). The delta across a loop iteration is
/// the amount of time the machine was suspended:
///
///   * Real suspend → boot advances by the suspend duration, monotonic
///     stays flat, delta ≈ suspend duration.
///   * Scheduler stall / heavy CPU work → both advance equally, delta ≈ 0.
///   * Virtual-time jumps under `tokio::time::start_paused(true)` → neither
///     wall clock is touched by tokio's virtual clock; both still advance
///     by whatever real wall time elapsed during the iteration, delta ≈ 0.
///
/// Using just `boot_elapsed` alone against a fixed threshold conflates all
/// three cases, which caused spurious `DropAllConnections` under CI load
/// (the 2026-04-14 nightly logs showed `boot_elapsed_secs=139` tripping the
/// old 30s test-only threshold mid-test).
///
/// ## Non-Linux platforms
///
/// On macOS / FreeBSD / Emscripten the `boot_time` crate resolves to the
/// same clock Rust's `std::time::Instant` uses (`mach_continuous_time` on
/// Darwin; `CLOCK_MONOTONIC` fallback on FreeBSD and Emscripten per the
/// boot_time-0.1.3 source). Both sides of the subtraction therefore advance
/// together and the delta stays near zero — the detector becomes a safe
/// no-op on those platforms rather than a false-positive source. Freenet
/// production gateways and CI runners are Linux, where the detector is
/// load-bearing; the no-op elsewhere is an intentional safe-degradation.
///
/// ## Monotonic-ahead diagnostic
///
/// `saturating_sub` intentionally clamps to zero if `mono_elapsed >
/// boot_elapsed`. Small (O(ns)..O(µs)) negative deltas are normal — the
/// two clocks are read back-to-back, not atomically, so scheduling jitter
/// between the two reads shows up as skew. A *large* negative delta would
/// indicate something pathological (virtualization TSC jump after live
/// migration, or a monotonic clock going backwards); the caller logs a
/// diagnostic warning in that case instead of silently swallowing the
/// signal.
#[inline]
fn classify_suspend_jump(boot_elapsed: Duration, mono_elapsed: Duration) -> Duration {
    boot_elapsed.saturating_sub(mono_elapsed)
}

/// Best-effort snapshot of this process's open file-descriptor count and the
/// `RLIMIT_NOFILE` soft limit, for the node-health telemetry gauges (#4440).
///
/// fd exhaustion (open fds reaching the soft limit → `EMFILE`) drove the
/// v0.2.73 gateway crash-loop and was invisible to central telemetry at the
/// time. Emitting these on the existing `router_snapshot` cadence makes the
/// headroom observable to operators.
///
/// Returns `(open_fds, fd_soft_limit)`. Either element is `None` on a platform
/// where it can't be read cheaply: the open-fd count is Linux-only (via
/// `/proc/self/fd`); the soft limit is any-unix (via `getrlimit`).
fn read_fd_usage() -> (Option<u64>, Option<u64>) {
    (read_open_fd_count(), read_fd_soft_limit())
}

/// Count this process's open file descriptors by enumerating `/proc/self/fd`.
///
/// Returns `None` if the process is already at its fd limit — `read_dir` itself
/// needs a descriptor and fails with `EMFILE` at the cliff. The companion
/// [`read_fd_soft_limit`] still reports in that case, and the 5-minute cadence
/// captures the *approach* to the limit, which is the actionable signal.
#[cfg(target_os = "linux")]
fn read_open_fd_count() -> Option<u64> {
    // `read_dir` itself holds one descriptor open for the duration of the
    // iteration, so it is included in the entry count; subtract it to report
    // the steady-state number of descriptors the process actually holds.
    let count = std::fs::read_dir("/proc/self/fd").ok()?.count() as u64;
    Some(count.saturating_sub(1))
}

#[cfg(not(target_os = "linux"))]
fn read_open_fd_count() -> Option<u64> {
    None
}

/// Read the current `RLIMIT_NOFILE` soft limit (the ceiling that triggers
/// `EMFILE`). Mirrors the `getrlimit` read in `bin/freenet.rs::raise_fd_limit`.
#[cfg(unix)]
fn read_fd_soft_limit() -> Option<u64> {
    let mut limits = libc::rlimit {
        rlim_cur: 0,
        rlim_max: 0,
    };
    // SAFETY: `getrlimit` with a valid resource id and an exclusively-borrowed,
    // initialized out-param is sound; the return code is checked before the
    // struct is read.
    if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limits) } != 0 {
        return None;
    }
    // `rlim_t` is `u64` on linux-gnu (CI target) but varies across unix targets;
    // normalize to `u64`. The cast is redundant on linux-gnu, hence the scoped
    // allow rather than a bare `as u64` that trips clippy.
    #[allow(clippy::unnecessary_cast)]
    Some(limits.rlim_cur as u64)
}

#[cfg(not(unix))]
fn read_fd_soft_limit() -> Option<u64> {
    None
}

#[cfg(test)]
mod fd_usage_tests {
    //! Validates the node-health fd gauges (#4440): the open-fd count and the
    //! `RLIMIT_NOFILE` soft limit must report sane values on Linux so the
    //! fd-exhaustion headroom signal that was missing during the v0.2.73
    //! incident is trustworthy.

    #[cfg(target_os = "linux")]
    #[test]
    fn read_fd_usage_reports_sane_values_on_linux() {
        let (open, soft) = super::read_fd_usage();
        let open = open.expect("linux reports an open-fd count via /proc/self/fd");
        let soft = soft.expect("unix reports the RLIMIT_NOFILE soft limit");
        // A running process always holds at least stdin/stdout/stderr.
        assert!(open > 0, "expected a positive open-fd count, got {open}");
        // The kernel enforces open-fds <= soft limit, so this is invariant, not
        // a heuristic — it pins that we read the two values the right way round.
        assert!(
            soft >= open,
            "open fds ({open}) must not exceed the soft limit ({soft})"
        );
    }
}

#[cfg(test)]
mod resource_meter_bridge_tests {
    //! Regression tests for #3453: the topology resource meter was never fed
    //! per-peer bandwidth data in production, so `adjust_topology` always saw
    //! ~0% usage and perpetually took the "add connections" branch. These
    //! tests drive `feed_peer_bandwidth_to_meter` — the production bridge from
    //! `TRANSPORT_METRICS.per_peer_snapshot()` into the meter — and assert the
    //! meter actually receives the samples, exercising the load-aware
    //! remove/hold/add decision logic that was previously dead in production.

    // `super::*` brings in ConnectionManager, TopologyAdjustment, Location,
    // SocketAddr, HashMap, BTreeMap, Duration, and tokio's Instant.
    use super::*;
    use crate::topology::meter::{AttributionSource, ResourceType};
    use crate::transport::TransportKeypair;

    /// Mirror of `topology::constants::SOURCE_RAMP_UP_DURATION` (5 min). That
    /// constant is `pub(super)` to the topology module, so we restate it here.
    /// Samples reported older than this window use their measured attributed
    /// rate rather than the P50 ramp-up extrapolation.
    const SOURCE_RAMP_UP_DURATION: Duration = Duration::from_secs(5 * 60);

    /// Add `n` ring connections to `cm`, returning their socket addresses in
    /// the order added. Each gets a distinct loopback address and location.
    fn add_connections(cm: &ConnectionManager, n: usize) -> Vec<SocketAddr> {
        let mut addrs = Vec::with_capacity(n);
        for i in 0..n {
            let addr: SocketAddr = format!("127.0.0.1:{}", 9000 + i).parse().unwrap();
            // Spread locations across the ring so each peer is a distinct
            // neighbor location (adjust_topology keys neighbors by location).
            let loc = Location::new((i as f64 + 0.5) / n as f64);
            let keypair = TransportKeypair::new();
            assert!(cm.add_connection(loc, addr, keypair.public().clone(), false));
            addrs.push(addr);
        }
        addrs
    }

    /// Drive the bridge once per simulated second, placing the first sample at
    /// `first_sample` and one sample per second thereafter for `ticks` seconds.
    /// `per_sec_recv[i]` is the bytes-received delta attributed to `addrs[i]`
    /// each second; cumulative counters are simulated by accumulating the
    /// per-second deltas (mirrors production, where each maintenance tick feeds
    /// one delta per peer). Returns the final `prev` map for inspection.
    ///
    /// Callers that want the meter to report a *measured* rate (rather than the
    /// ramp-up P50 estimate) must arrange two things, both governed by the
    /// sample timestamps relative to the eventual `adjust_topology` query time
    /// `q`:
    ///   1. The first sample (the source's creation time) must be older than
    ///      `SOURCE_RAMP_UP_DURATION` before `q`, or `extrapolated_usage`
    ///      treats the source as ramping up and substitutes the network P50.
    ///   2. The samples must extend up to just before `q`, because the meter's
    ///      rate is `sum(retained samples) / (q − oldest_retained_sample)`.
    ///      A window that ends far in the past inflates the denominator and
    ///      dilutes the rate toward zero. The meter retains only the most
    ///      recent `RUNNING_AVERAGE_WINDOW` (100) samples.
    fn drive_bridge(
        cm: &ConnectionManager,
        addrs: &[SocketAddr],
        per_sec_recv: &[u64],
        first_sample: Instant,
        ticks: u64,
    ) -> HashMap<SocketAddr, (u64, u64)> {
        let mut prev: HashMap<SocketAddr, (u64, u64)> = HashMap::new();
        let mut cumulative: Vec<u64> = vec![0; addrs.len()];
        for t in 0..ticks {
            for (i, c) in cumulative.iter_mut().enumerate() {
                *c += per_sec_recv[i];
            }
            let snapshot: Vec<(SocketAddr, u64, u64)> = addrs
                .iter()
                .zip(cumulative.iter())
                .map(|(&addr, &recv)| (addr, 0u64, recv))
                .collect();
            let at = first_sample + Duration::from_secs(t);
            feed_peer_bandwidth_to_meter(cm, &snapshot, &mut prev, at);
        }
        prev
    }

    /// The core regression: with the bridge feeding high per-peer bandwidth,
    /// `adjust_topology` reaches the resource-overload "remove" branch. Before
    /// the fix nothing fed the meter, so this path was unreachable in
    /// production and the node would instead always try to ADD connections.
    #[test_log::test]
    fn bridge_feeds_meter_so_overloaded_node_removes() {
        // test_default: min_connections=4, max_connections=10,
        // max_upstream/downstream = 1_000_000 bytes/sec.
        let cm = ConnectionManager::test_default();
        let addrs = add_connections(&cm, 6);

        // `extrapolated_usage` SUMS the per-source rates across all peers, so
        // 200_000 B/s inbound per peer × 6 peers = 1_200_000 B/s = 1.2× the
        // 1_000_000 downstream limit → usage proportion 1.2 > 0.9 → remove.
        let per_sec_recv = vec![200_000u64; 6];

        // Choose the query time `now`, then lay samples from
        // (now − RAMP_UP − 30s) up to (now − 1s): the first sample is older
        // than the ramp-up window (so the source reports its measured rate),
        // and the retained 100-sample tail ends ~1s before `now` (so the rate
        // isn't diluted by a stale denominator). See `drive_bridge` docs.
        let now = Instant::now();
        let first_sample = now - SOURCE_RAMP_UP_DURATION - Duration::from_secs(30);
        let ticks = SOURCE_RAMP_UP_DURATION.as_secs() + 29; // last sample ≈ now − 1s
        drive_bridge(&cm, &addrs, &per_sec_recv, first_sample, ticks);

        let mut neighbor_locations = BTreeMap::new();
        let conns = cm.get_connections_by_location();
        for (loc, c) in &conns {
            neighbor_locations.insert(*loc, c.clone());
        }

        let adjustment =
            cm.topology_manager
                .write()
                .adjust_topology(&neighbor_locations, &None, now, 6);

        assert!(
            matches!(adjustment, TopologyAdjustment::RemoveConnections(_)),
            "overloaded node fed via the bridge must remove a connection, got {adjustment:?}"
        );
    }

    /// Demonstrates the bug: with NO bridge call (the production state before
    /// this fix), the meter is empty and `adjust_topology` on a node above
    /// min_connections takes the low-usage "add" branch. This is the symptom
    /// #3453 describes; the test above proves the bridge cures it.
    #[test_log::test]
    fn empty_meter_never_removes() {
        let cm = ConnectionManager::test_default();
        let _addrs = add_connections(&cm, 6);

        let mut neighbor_locations = BTreeMap::new();
        let conns = cm.get_connections_by_location();
        for (loc, c) in &conns {
            neighbor_locations.insert(*loc, c.clone());
        }

        let adjustment = cm.topology_manager.write().adjust_topology(
            &neighbor_locations,
            &None,
            Instant::now(),
            6,
        );

        assert!(
            !matches!(adjustment, TopologyAdjustment::RemoveConnections(_)),
            "with an unfed meter the remove branch must be unreachable (the #3453 bug), got {adjustment:?}"
        );
    }

    /// The bridge must attribute samples to the right `ResourceType` and only
    /// for peers that resolve to a ring `PeerKeyLocation`.
    #[test_log::test]
    fn bridge_records_inbound_and_outbound_for_known_peer() {
        let cm = ConnectionManager::test_default();
        let addrs = add_connections(&cm, 1);
        let peer = cm
            .get_peer_by_addr(addrs[0])
            .expect("peer is a ring connection");

        let at = Instant::now();
        let mut prev = HashMap::new();
        // First snapshot establishes the baseline; cumulative counts start here.
        feed_peer_bandwidth_to_meter(&cm, &[(addrs[0], 1_000, 2_000)], &mut prev, at);

        let source = AttributionSource::Peer(peer);
        // `attributed_usage_rate` takes `&mut self` (it can refresh the
        // meter's cached estimate), so a write guard is required.
        let mut topo = cm.topology_manager.write();
        // First tick: delta == cumulative (prev was 0), so both directions get
        // a sample equal to the cumulative count.
        let outbound = topo.attributed_usage_rate(
            &source,
            &ResourceType::OutboundBandwidthBytes,
            at + Duration::from_secs(1),
        );
        let inbound = topo.attributed_usage_rate(
            &source,
            &ResourceType::InboundBandwidthBytes,
            at + Duration::from_secs(1),
        );
        drop(topo);

        assert_eq!(outbound.expect("outbound recorded").per_second(), 1_000.0);
        assert_eq!(inbound.expect("inbound recorded").per_second(), 2_000.0);
    }

    /// An address with no ring connection (e.g. a transient handshake) must be
    /// skipped, not attributed to a phantom source. `prev` is still updated so
    /// the next tick diffs correctly if the peer later resolves.
    #[test_log::test]
    fn bridge_skips_unresolvable_address() {
        let cm = ConnectionManager::test_default();
        // Six real peers above min_connections so the only thing that could
        // push usage into the remove branch is bandwidth attribution.
        let _addrs = add_connections(&cm, 6);
        // An address that is NOT a ring connection.
        let unknown: SocketAddr = "127.0.0.1:55555".parse().unwrap();

        // Feed enormous traffic for the unknown address across a full window.
        let now = Instant::now();
        let window_end = now - SOURCE_RAMP_UP_DURATION - Duration::from_secs(30);
        let mut prev = HashMap::new();
        let mut cum = 0u64;
        for t in 0..120u64 {
            cum += 10_000_000; // far above any limit, IF it were attributed
            feed_peer_bandwidth_to_meter(
                &cm,
                &[(unknown, 0, cum)],
                &mut prev,
                window_end - Duration::from_secs(120 - t),
            );
        }

        let mut neighbor_locations = BTreeMap::new();
        for (loc, c) in &cm.get_connections_by_location() {
            neighbor_locations.insert(*loc, c.clone());
        }
        let adjustment =
            cm.topology_manager
                .write()
                .adjust_topology(&neighbor_locations, &None, now, 6);

        // The unknown address contributed no meter sample, so usage stays ~0
        // and the node must NOT remove — proving the skip.
        assert!(
            !matches!(adjustment, TopologyAdjustment::RemoveConnections(_)),
            "traffic for an unresolvable address must not be attributed, got {adjustment:?}"
        );
        // But prev is still updated so a future tick computes the delta from here.
        assert_eq!(prev.get(&unknown).copied(), Some((0, cum)));
    }

    /// A counter that goes backwards (per-peer slot evicted then re-created,
    /// resetting cumulative counts) must not underflow into a giant bogus
    /// delta. saturating_sub clamps it to zero.
    #[test_log::test]
    fn bridge_handles_counter_reset_without_underflow() {
        let cm = ConnectionManager::test_default();
        let addrs = add_connections(&cm, 1);
        let peer = cm.get_peer_by_addr(addrs[0]).unwrap();
        let source = AttributionSource::Peer(peer);

        let t0 = Instant::now();
        let mut prev = HashMap::new();
        // Establish a high cumulative count.
        feed_peer_bandwidth_to_meter(&cm, &[(addrs[0], 0, 1_000_000)], &mut prev, t0);
        // Counter reset: cumulative drops to a small value. Delta must be 0,
        // not u64::MAX - something.
        let t1 = t0 + Duration::from_secs(1);
        feed_peer_bandwidth_to_meter(&cm, &[(addrs[0], 0, 10)], &mut prev, t1);

        let rate = cm
            .topology_manager
            .write()
            .attributed_usage_rate(
                &source,
                &ResourceType::InboundBandwidthBytes,
                t1 + Duration::from_secs(1),
            )
            .unwrap()
            .per_second();
        // Only the first tick's 1_000_000 sample is present; the reset tick
        // contributed a 0 delta (no sample). Sum=1_000_000 over a ~2s window.
        assert!(
            rate > 0.0 && rate.is_finite(),
            "counter reset must clamp to a finite, non-underflowed rate, got {rate}"
        );
    }

    /// `prev` must stay bounded as peers churn: entries for peers absent from
    /// the latest snapshot are pruned so it never outgrows the transport
    /// metrics table.
    #[test_log::test]
    fn bridge_prunes_departed_peers_from_prev() {
        let cm = ConnectionManager::test_default();
        let addrs = add_connections(&cm, 3);

        let at = Instant::now();
        let mut prev = HashMap::new();
        let full: Vec<_> = addrs.iter().map(|&a| (a, 1u64, 1u64)).collect();
        feed_peer_bandwidth_to_meter(&cm, &full, &mut prev, at);
        assert_eq!(prev.len(), 3);

        // Next tick: only one peer remains in the snapshot. The other two must
        // be pruned from prev.
        feed_peer_bandwidth_to_meter(
            &cm,
            &[(addrs[0], 2, 2)],
            &mut prev,
            at + Duration::from_secs(1),
        );
        assert_eq!(prev.len(), 1, "departed peers must be pruned from prev");
        assert!(prev.contains_key(&addrs[0]));
    }

    /// #3453 review (P1): the per-interval byte delta must be timestamped at
    /// the interval START so the meter divides it by the real interval length,
    /// not the 1-second floor in `RunningAverage::get_rate_at_time`. A single
    /// tick reporting one interval's worth of bytes must yield
    /// `bytes / interval`, NOT `bytes / 1s` (which would overstate the rate by
    /// the interval length and trigger spurious removals).
    #[test_log::test]
    fn bridge_timestamps_delta_at_interval_start_no_inflation() {
        let cm = ConnectionManager::test_default();
        let addrs = add_connections(&cm, 1);
        let peer = cm.get_peer_by_addr(addrs[0]).unwrap();
        let source = AttributionSource::Peer(peer);

        // 60_000 bytes transferred over a 60s interval = 1000 B/s. Production
        // passes the PREVIOUS tick time as `interval_start`.
        let interval = Duration::from_secs(60);
        let interval_start = Instant::now();
        let query_time = interval_start + interval;

        let mut prev = HashMap::new();
        feed_peer_bandwidth_to_meter(&cm, &[(addrs[0], 0, 60_000)], &mut prev, interval_start);

        let rate = cm
            .topology_manager
            .write()
            .attributed_usage_rate(&source, &ResourceType::InboundBandwidthBytes, query_time)
            .expect("inbound recorded")
            .per_second();

        // Correct: 60_000 / 60s = 1000 B/s. The interval-END bug would give
        // 60_000 / 1s = 60_000 B/s (60× inflation).
        assert!(
            (rate - 1000.0).abs() < 1.0,
            "interval delta must be divided by the real interval, got {rate} B/s (expected ~1000)"
        );
        assert!(
            rate < 2000.0,
            "rate must not be inflated toward the 1s-floor (delta/1s = 60000), got {rate}"
        );
    }

    /// #3453 review (P2): peers that leave the live set must be pruned from the
    /// topology meter (and `source_creation_times`), not accumulate forever.
    /// After a peer departs, a subsequent bridge tick that no longer resolves
    /// it must drop its meter samples.
    #[test_log::test]
    fn bridge_prunes_departed_peer_from_meter() {
        let cm = ConnectionManager::test_default();
        let addrs = add_connections(&cm, 2);
        let peer0 = cm.get_peer_by_addr(addrs[0]).unwrap();
        let peer1 = cm.get_peer_by_addr(addrs[1]).unwrap();
        let src0 = AttributionSource::Peer(peer0);
        let src1 = AttributionSource::Peer(peer1);

        let t0 = Instant::now();
        let mut prev = HashMap::new();
        // Tick 1: both peers transfer bytes — both get meter entries.
        feed_peer_bandwidth_to_meter(
            &cm,
            &[(addrs[0], 0, 10_000), (addrs[1], 0, 10_000)],
            &mut prev,
            t0,
        );
        let q = t0 + Duration::from_secs(60);
        {
            let mut topo = cm.topology_manager.write();
            assert!(
                topo.attributed_usage_rate(&src0, &ResourceType::InboundBandwidthBytes, q)
                    .is_some(),
                "peer0 should have a meter entry after tick 1"
            );
            assert!(
                topo.attributed_usage_rate(&src1, &ResourceType::InboundBandwidthBytes, q)
                    .is_some(),
                "peer1 should have a meter entry after tick 1"
            );
        }

        // peer1 disconnects from the ring; only peer0 remains a resolvable
        // connection. The transport snapshot drops the departed peer too.
        cm.prune_alive_connection(addrs[1]);
        assert!(
            cm.get_peer_by_addr(addrs[1]).is_none(),
            "peer1 must no longer resolve after prune"
        );

        // Tick 2: only peer0 present in the snapshot. peer1 is no longer in the
        // live set, so its meter samples must be pruned.
        feed_peer_bandwidth_to_meter(&cm, &[(addrs[0], 0, 20_000)], &mut prev, t0);
        {
            let mut topo = cm.topology_manager.write();
            assert!(
                topo.attributed_usage_rate(&src0, &ResourceType::InboundBandwidthBytes, q)
                    .is_some(),
                "live peer0 must keep its meter entry"
            );
            assert!(
                topo.attributed_usage_rate(&src1, &ResourceType::InboundBandwidthBytes, q)
                    .is_none(),
                "departed peer1 must be pruned from the meter"
            );
        }
    }
}

#[cfg(test)]
mod k_closest_source_tests {
    //! Source-scrape pin tests for `Ring::k_closest_potentially_hosting`.
    //!
    //! Constructing a real `Ring` for behavioral tests requires significant
    //! scaffolding (NodeConfig, EventLoopNotificationsSender, NetEventRegister,
    //! BackgroundTaskMonitor) that does not currently exist in the test suite.
    //! These tests instead scrape the production source to ensure load-bearing
    //! filter invariants stay wired in — a future refactor that silently drops
    //! the transient filter (issue #4222) or the readiness fallback should fail
    //! CI rather than silently regress production routing behavior.
    //!
    //! Behavioral coverage exists transitively via the simulation_integration
    //! tests; a dedicated integration test for the transient filter is filed
    //! as a follow-up to the Ring test-scaffolding work.
    fn production_source() -> &'static str {
        const FULL: &str = include_str!("ring.rs");
        // ring.rs has inline `#[cfg(test)]` annotations on individual const
        // declarations inside `connection_maintenance` (lines ~1894–1940), so
        // a plain `find("#[cfg(test)]")` would cut off the file mid-impl and
        // miss the function we want to scrape. Anchor on the first *top-level*
        // test module declaration instead.
        let cutoff = FULL
            .find("\n#[cfg(test)]\nmod ")
            .expect("ring.rs must have a top-level #[cfg(test)] mod section");
        &FULL[..cutoff]
    }

    fn extract_fn_body<'a>(source: &'a str, signature_prefix: &str) -> &'a str {
        let start = source
            .find(signature_prefix)
            .unwrap_or_else(|| panic!("could not find {signature_prefix}"));
        let brace = source[start..].find('{').expect("fn sig must have body");
        let body_start = start + brace + 1;
        let bytes = source.as_bytes();
        let mut depth: i32 = 1;
        let mut i = body_start;
        while i < bytes.len() {
            match bytes[i] {
                b'{' => depth += 1,
                b'}' => {
                    depth -= 1;
                    if depth == 0 {
                        return &source[body_start..i];
                    }
                }
                _ => {}
            }
            i += 1;
        }
        panic!("unbalanced braces while extracting {signature_prefix}");
    }

    /// Issue #4222: `k_closest_potentially_hosting` must skip transient peers
    /// so GET/SUBSCRIBE doesn't route through about-to-be-dropped connections.
    /// The PUT/UPDATE path already filters them via `routing_candidates`; this
    /// pin makes sure the GET path stays in sync.
    #[test]
    fn k_closest_potentially_hosting_filters_transient_peers() {
        let src = production_source();
        let body = extract_fn_body(src, "pub fn k_closest_potentially_hosting<K>(");
        assert!(
            body.contains("is_transient(addr)"),
            "k_closest_potentially_hosting must call is_transient(addr) on each \
             candidate connection. If the filter was deliberately removed, also \
             update .claude/rules/ring.md (which documents the filter rule) and \
             this test. Issue #4222 / #3570."
        );
        assert!(
            body.contains("skipped_transient"),
            "k_closest_potentially_hosting must surface a skipped_transient \
             counter so operators can see when the filter is removing peers."
        );
    }

    /// The existing readiness fallback must remain — without it, a node whose
    /// peers haven't yet sent ReadyState would fail every GET with EmptyRing.
    /// Pinning it here so a future refactor of the filter chain can't silently
    /// drop the fallback.
    #[test]
    fn k_closest_potentially_hosting_preserves_not_ready_fallback() {
        let src = production_source();
        let body = extract_fn_body(src, "pub fn k_closest_potentially_hosting<K>(");
        assert!(
            body.contains("not_ready_fallback"),
            "k_closest_potentially_hosting must keep the not-yet-ready peer \
             fallback so cold-start nodes don't fail every GET with EmptyRing."
        );
    }
}

#[cfg(test)]
mod renewal_ban_gate_source_tests {
    //! Source-scrape pin test for the subscription-renewal ban gate (#4373).
    //!
    //! `recover_orphaned_subscriptions` spawns a `run_renewal_subscribe`
    //! driver for every contract in `contracts_needing_renewal()`. That
    //! driver emits an outbound SUBSCRIBE using the same machinery as a
    //! client-initiated request, but — unlike the four `start_client_*`
    //! originator entry points — the renewal scheduler does NOT route
    //! through `operations::reject_if_contract_banned`. So before #4373 a
    //! contract that was banned while the node still held an active
    //! subscription kept emitting outbound SUBSCRIBE renewals on every
    //! maintenance cycle until the ban TTL lifted.
    //!
    //! The fix adds an `is_banned` gate in the renewal loop. As the
    //! `k_closest_source_tests` module above documents, building a real
    //! `Ring` for a behavioral test requires scaffolding that does not yet
    //! exist in this suite, so — mirroring the `*_dispatch_gates_banned_contracts`
    //! pins in `contract_ban_list.rs` — this test scrapes the production
    //! source to ensure the gate stays wired in and keeps running BEFORE
    //! the renewal is spawned. A refactor that drops or reorders the gate
    //! would fail CI rather than silently re-open the egress leak.

    fn production_source() -> &'static str {
        const FULL: &str = include_str!("ring.rs");
        let cutoff = FULL
            .find("\n#[cfg(test)]\nmod ")
            .expect("ring.rs must have a top-level #[cfg(test)] mod section");
        &FULL[..cutoff]
    }

    fn extract_fn_body<'a>(source: &'a str, signature_prefix: &str) -> &'a str {
        let start = source
            .find(signature_prefix)
            .unwrap_or_else(|| panic!("could not find {signature_prefix}"));
        let brace = source[start..].find('{').expect("fn sig must have body");
        let body_start = start + brace + 1;
        let bytes = source.as_bytes();
        let mut depth: i32 = 1;
        let mut i = body_start;
        while i < bytes.len() {
            match bytes[i] {
                b'{' => depth += 1,
                b'}' => {
                    depth -= 1;
                    if depth == 0 {
                        return &source[body_start..i];
                    }
                }
                _ => {}
            }
            i += 1;
        }
        panic!("unbalanced braces while extracting {signature_prefix}");
    }

    #[test]
    fn renewal_loop_gates_banned_contracts_before_spawning() {
        let src = production_source();
        let body = extract_fn_body(
            src,
            "async fn recover_orphaned_subscriptions(ring: Arc<Self>",
        );

        let gate_pos = body.find("contract_ban_list.is_banned").expect(
            "the subscription-renewal loop must gate on \
             `contract_ban_list.is_banned` so a banned-but-still-subscribed \
             contract stops emitting outbound SUBSCRIBE renewals (#4373). If \
             this gate was removed, the egress leak is back.",
        );

        // The gate must precede the spam-prevention check: a banned contract
        // is skipped regardless of its `can_request_subscription` backoff
        // state, so order matters.
        let can_request_pos = body
            .find("can_request_subscription(&contract)")
            .expect("renewal loop must still consult can_request_subscription");
        assert!(
            gate_pos < can_request_pos,
            "ban gate (offset {gate_pos}) must run BEFORE the \
             can_request_subscription spam check (offset {can_request_pos}) so \
             a banned contract is always skipped"
        );

        // The gate must precede the renewal spawn. Both `mark_subscription_pending`
        // (which flips per-contract pending state) and `run_renewal_subscribe`
        // (the outbound-egress driver) must only be reached for non-banned
        // contracts.
        let mark_pending_pos = body
            .find("mark_subscription_pending(contract)")
            .expect("renewal loop must still call mark_subscription_pending");
        assert!(
            gate_pos < mark_pending_pos,
            "ban gate (offset {gate_pos}) must run BEFORE \
             mark_subscription_pending (offset {mark_pending_pos})"
        );
        let spawn_pos = body
            .find("run_renewal_subscribe(")
            .expect("renewal loop must still spawn run_renewal_subscribe");
        assert!(
            gate_pos < spawn_pos,
            "ban gate (offset {gate_pos}) must run BEFORE the \
             run_renewal_subscribe outbound-egress spawn (offset {spawn_pos})"
        );
    }
}

#[cfg(test)]
mod renewal_ban_gate_behavior_tests {
    //! Behavioral coverage for the predicate the renewal-loop ban gate
    //! relies on (#4373): a contract on the `ContractBanList` reports
    //! `is_banned == true` (so the loop's `continue` fires), and an
    //! un-banned contract reports `false` (so renewal proceeds). This
    //! exercises the exact `contract_ban_list.is_banned(contract.id())`
    //! call the loop makes, including the `ContractKey::id()` projection,
    //! against the real ban-list type and a controllable time source.

    use super::contract_ban_list::{BanReason, ContractBanList};
    // `TimeSource` is needed in scope for the `ts.now()` trait method.
    use crate::util::time_source::{SharedMockTimeSource, TimeSource};
    use freenet_stdlib::prelude::{CodeHash, ContractInstanceId, ContractKey};
    use std::sync::Arc;
    use std::time::Duration;

    fn mk_key(byte: u8) -> ContractKey {
        // A full `ContractKey` (not a bare instance id) so the test
        // exercises the same `contract.id()` projection the renewal loop's
        // `is_banned(contract.id())` gate performs.
        ContractKey::from_id_and_code(
            ContractInstanceId::new([byte; 32]),
            CodeHash::new([byte; 32]),
        )
    }

    #[test]
    fn banned_contract_is_skipped_for_renewal_while_unbanned_proceeds() {
        let ts = SharedMockTimeSource::new();
        let ban_list = ContractBanList::new(Arc::new(ts.clone()));

        let banned = mk_key(1);
        let healthy = mk_key(2);

        // Ban one contract; leave the other alone.
        ban_list.ban(
            *banned.id(),
            ts.now() + Duration::from_secs(60),
            BanReason::AutoMad,
        );

        // The renewal loop's gate is `if is_banned(contract.id()) { continue }`.
        // Banned -> skipped; un-banned -> proceeds to renewal.
        assert!(
            ban_list.is_banned(banned.id()),
            "a banned contract must be skipped for subscription renewal (#4373)"
        );
        assert!(
            !ban_list.is_banned(healthy.id()),
            "a non-banned contract must still be eligible for renewal"
        );

        // Once the ban TTL lifts, the formerly-banned contract becomes
        // eligible for renewal again — the gate is time-bounded, matching
        // the issue's "self-resolves when the ban TTL expires" note.
        ts.advance_time(Duration::from_secs(61));
        assert!(
            !ban_list.is_banned(banned.id()),
            "after the ban TTL expires the contract is eligible for renewal again"
        );
    }
}

#[cfg(test)]
mod suspend_jump_tests {
    use super::classify_suspend_jump;
    use std::time::Duration;

    /// Scheduler stall: both clocks advance equally (the loop task was
    /// starved, but the machine wasn't suspended). The detector must NOT
    /// treat this as a suspend event — that was the root cause of the
    /// 2026-04-14 nightly `test_get_reliability_with_latency` false
    /// positive at `boot_elapsed_secs=139`.
    #[test]
    fn scheduler_stall_is_not_suspend() {
        let boot = Duration::from_secs(139);
        let mono = Duration::from_secs(139);
        assert_eq!(classify_suspend_jump(boot, mono), Duration::ZERO);
    }

    /// Real OS suspend: CLOCK_BOOTTIME advances by the suspend duration,
    /// CLOCK_MONOTONIC stays essentially flat. The detector must surface
    /// the full suspend duration so the caller can compare it against the
    /// threshold and trigger the DropAllConnections recovery path.
    #[test]
    fn real_suspend_is_detected() {
        let boot = Duration::from_secs(3600);
        let mono = Duration::from_millis(50);
        assert_eq!(
            classify_suspend_jump(boot, mono),
            Duration::from_secs(3600) - Duration::from_millis(50)
        );
    }

    /// Healthy tick: both clocks advance by roughly the tick interval and
    /// the delta is near zero.
    #[test]
    fn healthy_tick_is_not_suspend() {
        let boot = Duration::from_millis(2050);
        let mono = Duration::from_millis(2048);
        assert_eq!(classify_suspend_jump(boot, mono), Duration::from_millis(2));
    }

    /// Monotonic exceeding boot (can happen from scheduling jitter between
    /// the two non-atomic clock reads, or from virtualization TSC anomalies)
    /// must saturate to zero rather than underflow. A large skew here is
    /// logged separately by `connection_maintenance` as a diagnostic.
    #[test]
    fn monotonic_ahead_of_boot_saturates_to_zero() {
        let boot = Duration::from_millis(100);
        let mono = Duration::from_millis(101);
        assert_eq!(classify_suspend_jump(boot, mono), Duration::ZERO);
    }

    /// Threshold comparison: the value the live code compares against the
    /// detection threshold must be the delta, not `boot_elapsed` alone.
    /// A 139s scheduler stall at a 30s threshold would trip the old
    /// detector; with the delta it does not.
    #[test]
    fn threshold_comparison_rejects_scheduler_stall() {
        let threshold = Duration::from_secs(30);
        let boot = Duration::from_secs(139);
        let mono = Duration::from_secs(139);
        assert!(classify_suspend_jump(boot, mono) <= threshold);
    }

    /// Threshold comparison: a real suspend of two full check ticks must
    /// exceed the 2x-tick detection threshold.
    #[test]
    fn threshold_comparison_accepts_real_suspend() {
        let threshold = Duration::from_secs(4);
        let boot = Duration::from_secs(300);
        let mono = Duration::from_millis(3);
        assert!(classify_suspend_jump(boot, mono) > threshold);
    }
}

/// Predicate controlling when `connection_maintenance` fires a gateway version probe.
///
/// Extracted as a free function so each branch (gateway role, empty configuration,
/// timing) can be exercised in unit tests without standing up an `OpManager`. The
/// caller relies on `has_configured_gateways` to gate the modulo-indexing into
/// `op_manager.configured_gateways` — flipping that flag at the call site (rather
/// than inside this function) keeps the borrow of the gateway slice local to the
/// caller.
#[inline]
fn should_probe_gateway(
    is_gateway: bool,
    has_configured_gateways: bool,
    now: Instant,
    next_probe: Instant,
) -> bool {
    !is_gateway && has_configured_gateways && now >= next_probe
}

#[cfg(test)]
mod gateway_version_probe_predicate_tests {
    use super::should_probe_gateway;
    use std::time::Duration;
    use tokio::time::Instant;

    #[test]
    fn fires_on_non_gateway_with_configured_gateways_at_due_time() {
        let now = Instant::now();
        let next_probe = now - Duration::from_secs(1);
        assert!(should_probe_gateway(false, true, now, next_probe));
    }

    #[test]
    fn skipped_when_running_as_gateway() {
        // Gateways must never probe themselves — they are the destination, not
        // the originator. This is the protection that keeps the gateway loop
        // from generating phantom CONNECTs.
        let now = Instant::now();
        let next_probe = now - Duration::from_secs(1);
        assert!(!should_probe_gateway(true, true, now, next_probe));
    }

    #[test]
    fn skipped_when_no_gateways_are_configured() {
        // The empty-gateways case is the boundary that previously sat in an
        // inner `if` and was unreachable from any test. Merging it into the
        // predicate makes the modulo-indexing in the caller structurally safe.
        let now = Instant::now();
        let next_probe = now - Duration::from_secs(1);
        assert!(!should_probe_gateway(false, false, now, next_probe));
    }

    #[test]
    fn skipped_before_due_time() {
        let now = Instant::now();
        let next_probe = now + Duration::from_secs(60);
        assert!(!should_probe_gateway(false, true, now, next_probe));
    }

    #[test]
    fn fires_at_exact_due_time_boundary() {
        // `>=` semantics: a now equal to next_probe should fire, not skip.
        let now = Instant::now();
        assert!(should_probe_gateway(false, true, now, now));
    }

    #[test]
    fn gateway_with_no_configured_gateways_is_still_skipped() {
        let now = Instant::now();
        assert!(!should_probe_gateway(true, false, now, now));
    }
}

#[cfg(test)]
mod max_concurrent_connections_tests {
    use super::calculate_max_concurrent_connections;

    #[test]
    fn at_min_connections_returns_base() {
        assert_eq!(calculate_max_concurrent_connections(25, 25), 3);
    }

    #[test]
    fn above_min_connections_returns_base() {
        assert_eq!(calculate_max_concurrent_connections(30, 25), 3);
    }

    #[test]
    fn large_deficit_scales_up() {
        // deficit=15, 3 + 15/3 = 8, cap = 25/2 = 12 → 8
        assert_eq!(calculate_max_concurrent_connections(10, 25), 8);
    }

    #[test]
    fn full_deficit_capped_at_half_min() {
        // deficit=25, 3 + 25/3 = 11, cap = 25/2 = 12 → 11
        assert_eq!(calculate_max_concurrent_connections(0, 25), 11);
    }

    #[test]
    fn small_deficit_adds_nothing() {
        // deficit=2, 3 + 2/3 = 3, cap = 25/2 = 12 → 3
        assert_eq!(calculate_max_concurrent_connections(23, 25), 3);
    }

    #[test]
    fn very_small_min_connections() {
        // min_conns=1, deficit=1, 3 + 0 = 3, cap = max(0, 3) = 3 → 3
        assert_eq!(calculate_max_concurrent_connections(0, 1), 3);
        // min_conns=2, deficit=2, 3 + 0 = 3, cap = max(1, 3) = 3 → 3
        assert_eq!(calculate_max_concurrent_connections(0, 2), 3);
    }

    #[test]
    fn high_min_connections_scales_cap() {
        // deficit=50, 3 + 50/3 = 19, cap = 50/2 = 25 → 19
        assert_eq!(calculate_max_concurrent_connections(0, 50), 19);
    }
}

#[cfg(test)]
mod respect_location_backoff_tests {
    use super::should_respect_location_backoff;

    /// Regression guard for the under-min backoff escape (#4348): a node below
    /// min_connections must bypass per-target location backoff so it cannot be
    /// trapped below min by a stamped `Rejected` backoff; at/above min, backoff
    /// is respected as before. Pins the inequality direction and boundary.
    #[test]
    fn below_min_bypasses_backoff() {
        assert!(!should_respect_location_backoff(0, 10));
        assert!(!should_respect_location_backoff(7, 10));
        assert!(!should_respect_location_backoff(9, 10));
    }

    #[test]
    fn at_or_above_min_respects_backoff() {
        assert!(should_respect_location_backoff(10, 10)); // exactly min
        assert!(should_respect_location_backoff(11, 10));
        assert!(should_respect_location_backoff(20, 10));
    }
}

#[cfg(test)]
mod pending_additions_tests {
    use super::calculate_allowed_connection_additions;

    #[test]
    fn respects_minimum_when_backlog_exists() {
        let allowed = calculate_allowed_connection_additions(1, 24, 25, 200, 24);
        assert_eq!(allowed, 0, "Backlog should satisfy minimum deficit");
    }

    #[test]
    fn permits_requests_until_minimum_is_met() {
        let allowed = calculate_allowed_connection_additions(1, 0, 25, 200, 24);
        assert_eq!(allowed, 24);
    }

    #[test]
    fn caps_additions_at_available_capacity() {
        let allowed = calculate_allowed_connection_additions(190, 5, 25, 200, 10);
        assert_eq!(allowed, 5);
    }

    #[test]
    fn respects_requested_when_capacity_allows() {
        let allowed = calculate_allowed_connection_additions(50, 0, 25, 200, 3);
        assert_eq!(allowed, 3);
    }
}

#[cfg(test)]
mod op_manager_state_tests {
    use super::{OpManagerState, classify_op_manager_ref};
    use parking_lot::RwLock;
    use std::sync::{Arc, Weak};

    // Stand-in for `OpManager`: `classify_op_manager_ref` is generic over the
    // pointee, so we exercise the exact production logic without building a
    // full node. The three slot states map 1:1 onto the maintenance loop's
    // startup / running / shutdown decisions (#3308).

    #[test]
    fn empty_slot_is_not_attached() {
        // Mirrors the startup window before `attach_op_manager` runs.
        let slot: RwLock<Option<Weak<u32>>> = RwLock::new(None);
        assert!(matches!(
            classify_op_manager_ref(&slot),
            OpManagerState::NotAttached
        ));
    }

    #[test]
    fn live_weak_upgrades_to_live() {
        let owner = Arc::new(7u32);
        let slot: RwLock<Option<Weak<u32>>> = RwLock::new(Some(Arc::downgrade(&owner)));
        match classify_op_manager_ref(&slot) {
            OpManagerState::Live(got) => assert_eq!(*got, 7),
            OpManagerState::NotAttached | OpManagerState::Detached => {
                panic!("expected Live while the owner Arc is alive")
            }
        }
    }

    #[test]
    fn dropped_owner_is_detached_not_not_attached() {
        // The #3308 regression: the slot was attached (`Some(weak)`) but the
        // owning Arc has been dropped. This MUST classify as `Detached`
        // (terminate the loop), NOT `NotAttached` (which spins forever).
        let owner = Arc::new(7u32);
        let weak = Arc::downgrade(&owner);
        let slot: RwLock<Option<Weak<u32>>> = RwLock::new(Some(weak));
        drop(owner);
        assert!(
            matches!(classify_op_manager_ref(&slot), OpManagerState::Detached),
            "a dropped owner must be Detached so connection_maintenance exits \
             instead of zombie-spinning (#3308)"
        );
    }

    /// End-to-end wiring guard for the maintenance loop's `Detached` arm.
    ///
    /// The pure classifier tests above only pin the `slot -> state` mapping.
    /// They would all stay green if a refactor swapped the loop's `Detached`
    /// arm back to `continue` (the original #3308 bug) or to a bare
    /// `return Ok(())` (the #4292 monitor-convention violation). This test
    /// pins the *action*: a maintenance-shaped task that polls
    /// `classify_op_manager_ref`, breaks on `Detached`, and then parks must,
    /// once its owner `Arc` is dropped, hand a *non-completing* future to a
    /// real `BackgroundTaskMonitor` — `wait_for_any_exit` must NOT fire.
    ///
    /// Mirrors `reference_ping::spawn_with_unbindable_target_does_not_exit`.
    #[tokio::test(start_paused = true)]
    async fn detached_maintenance_task_parks_without_tripping_monitor() {
        use crate::node::background_task_monitor::BackgroundTaskMonitor;
        use std::time::Duration;

        // Shared back-reference slot, exactly like `Ring::op_manager`.
        let owner = Arc::new(0u32);
        let slot: Arc<RwLock<Option<Weak<u32>>>> =
            Arc::new(RwLock::new(Some(Arc::downgrade(&owner))));

        let task_slot = Arc::clone(&slot);
        let handle = tokio::spawn(async move {
            // Reproduces the maintenance loop's arm-to-action wiring without
            // standing up a full Ring: poll the back-reference, act per state.
            loop {
                match classify_op_manager_ref(&task_slot) {
                    // Owner still alive: keep running (the Live work is a
                    // no-op here; the point is that it does NOT exit).
                    OpManagerState::Live(_) => {
                        tokio::time::sleep(Duration::from_millis(50)).await;
                    }
                    // Startup window: keep waiting.
                    OpManagerState::NotAttached => {
                        tokio::time::sleep(Duration::from_millis(50)).await;
                    }
                    // Owner dropped: stop the loop, then park (#3308 / #4292).
                    OpManagerState::Detached => break,
                }
            }
            // Teardown parking: must NOT return cleanly, or the monitor fires.
            std::future::pending::<()>().await;
        });

        let monitor = BackgroundTaskMonitor::new();
        monitor.register("connection_maintenance_sim", handle);

        // Let the task observe `Live` at least once and start polling.
        tokio::time::advance(Duration::from_millis(100)).await;
        tokio::task::yield_now().await;

        // Drop the owner: the weak now fails to upgrade -> `Detached`.
        drop(owner);

        // Give the task time to observe `Detached`, break, and park.
        tokio::time::advance(Duration::from_millis(200)).await;
        tokio::task::yield_now().await;

        // The monitor must NOT fire: orderly teardown parks instead of
        // returning. A `continue` (zombie spin) or a `return Ok(())`
        // (monitor-convention violation) would both fail this assertion —
        // the latter by resolving `wait_for_any_exit`.
        let exit = monitor.wait_for_any_exit();
        tokio::pin!(exit);
        let still_parked = tokio::time::timeout(Duration::from_millis(100), &mut exit)
            .await
            .is_err();
        assert!(
            still_parked,
            "Detached maintenance task must park, not exit: a clean return \
             would trip BackgroundTaskMonitor::wait_for_any_exit and crash \
             the node (#3308 / #4292)"
        );
    }
}

#[cfg(test)]
mod refresh_router_tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;

    use either::Either;
    use futures::FutureExt;
    use parking_lot::RwLock;

    use crate::ring::PeerKeyLocation;
    use crate::ring::location::Location;
    use crate::router::{RouteEvent, RouteOutcome, Router};
    use crate::tracing::{NetEventLog, NetEventRegister};

    /// Mock register that returns pre-populated RouteEvents on startup.
    #[derive(Clone)]
    struct WarmStartRegister {
        events: Vec<RouteEvent>,
    }

    impl NetEventRegister for WarmStartRegister {
        fn register_events<'a>(
            &'a self,
            _events: Either<NetEventLog<'a>, Vec<NetEventLog<'a>>>,
        ) -> futures::future::BoxFuture<'a, ()> {
            async {}.boxed()
        }

        fn notify_of_time_out(
            &mut self,
            _tx: crate::message::Transaction,
            _op_type: &str,
            _target_peer: Option<String>,
        ) -> futures::future::BoxFuture<'_, ()> {
            async {}.boxed()
        }

        fn trait_clone(&self) -> Box<dyn NetEventRegister> {
            Box::new(self.clone())
        }

        fn get_router_events(
            &self,
            number: usize,
        ) -> futures::future::BoxFuture<'_, anyhow::Result<Vec<RouteEvent>>> {
            let events = self.events.iter().take(number).cloned().collect();
            async move { Ok(events) }.boxed()
        }
    }

    fn make_route_events(count: usize) -> Vec<RouteEvent> {
        (0..count)
            .map(|_| RouteEvent {
                peer: PeerKeyLocation::random(),
                contract_location: Location::random(),
                outcome: RouteOutcome::Success {
                    time_to_response_start: Duration::from_millis(50),
                    payload_size: 1000,
                    payload_transfer_time: Duration::from_millis(100),
                },
                op_type: None,
            })
            .collect()
    }

    #[tokio::test]
    async fn refresh_router_loads_history_on_startup() {
        let events = make_route_events(100);
        let register = WarmStartRegister {
            events: events.clone(),
        };
        let router = Arc::new(RwLock::new(Router::new(&[])));

        // Verify the router starts empty
        let snapshot = router.read().snapshot();
        assert_eq!(snapshot.failure_events, 0);
        assert_eq!(snapshot.success_events, 0);

        // Run refresh_router with a timeout so it doesn't loop forever.
        // The startup load happens before the interval loop, so it completes
        // within the first few milliseconds.
        tokio::time::timeout(
            Duration::from_millis(100),
            super::Ring::refresh_router(router.clone(), register),
        )
        .await
        .ok(); // timeout is expected — the function loops forever

        // Verify the router was populated from the historical events
        let snapshot = router.read().snapshot();
        assert_eq!(
            snapshot.success_events, 100,
            "Router should have been populated with startup history"
        );
    }

    #[tokio::test]
    async fn refresh_router_handles_empty_history() {
        let register = WarmStartRegister { events: vec![] };
        let router = Arc::new(RwLock::new(Router::new(&[])));

        tokio::time::timeout(
            Duration::from_millis(100),
            super::Ring::refresh_router(router.clone(), register),
        )
        .await
        .ok();

        // Router should remain empty
        let snapshot = router.read().snapshot();
        assert_eq!(snapshot.failure_events, 0);
        assert_eq!(snapshot.success_events, 0);
    }

    /// Mock register whose `get_router_events` fails the first `fail_first`
    /// times it is called, then returns the configured events on every
    /// subsequent call. Records the total call count so tests can confirm the
    /// refresh loop keeps polling past a transient error rather than dying.
    #[derive(Clone)]
    struct FlakyRegister {
        events: Vec<RouteEvent>,
        fail_first: usize,
        calls: Arc<AtomicUsize>,
    }

    impl NetEventRegister for FlakyRegister {
        fn register_events<'a>(
            &'a self,
            _events: Either<NetEventLog<'a>, Vec<NetEventLog<'a>>>,
        ) -> futures::future::BoxFuture<'a, ()> {
            async {}.boxed()
        }

        fn notify_of_time_out(
            &mut self,
            _tx: crate::message::Transaction,
            _op_type: &str,
            _target_peer: Option<String>,
        ) -> futures::future::BoxFuture<'_, ()> {
            async {}.boxed()
        }

        fn trait_clone(&self) -> Box<dyn NetEventRegister> {
            Box::new(self.clone())
        }

        fn get_router_events(
            &self,
            number: usize,
        ) -> futures::future::BoxFuture<'_, anyhow::Result<Vec<RouteEvent>>> {
            let call = self.calls.fetch_add(1, Ordering::SeqCst);
            let fail = call < self.fail_first;
            let events: Vec<RouteEvent> = self.events.iter().take(number).cloned().collect();
            async move {
                if fail {
                    // Mimics the production failure: an AOF segment read failing
                    // under fd exhaustion ("No file descriptors available").
                    Err(anyhow::anyhow!(
                        "No file descriptors available (os error 24)"
                    ))
                } else {
                    Ok(events)
                }
            }
            .boxed()
        }
    }

    /// Regression test: a transient `get_router_events` error inside the
    /// periodic refresh loop must NOT terminate the task (which, as a monitored
    /// background task, would be node-fatal and crash-loop the gateway).
    ///
    /// The mock fails the startup load plus the first few periodic ticks, then
    /// succeeds. We assert the loop kept polling past the errors (call count >
    /// the number of injected failures) AND eventually recovered by loading the
    /// history — both impossible if the error arm had `return`ed.
    #[tokio::test(start_paused = true)]
    async fn refresh_router_survives_transient_get_router_events_error() {
        let events = make_route_events(100);
        let calls = Arc::new(AtomicUsize::new(0));
        // Fail the startup load (call 0) and the first 3 periodic ticks
        // (calls 1..=3), then succeed from call 4 onward.
        let register = FlakyRegister {
            events: events.clone(),
            fail_first: 4,
            calls: calls.clone(),
        };
        let router = Arc::new(RwLock::new(Router::new(&[])));

        // The periodic loop ticks every 5 minutes; under start_paused the
        // timeout auto-advances virtual time, so ~40 min covers >5 ticks and
        // lets the loop poll well past the injected failures and recover.
        tokio::time::timeout(
            Duration::from_secs(60 * 40),
            super::Ring::refresh_router(router.clone(), register),
        )
        .await
        .ok(); // timeout is expected — the loop runs forever when healthy

        // The loop must have kept polling past every injected failure rather
        // than returning on the first Err.
        let total_calls = calls.load(Ordering::SeqCst);
        assert!(
            total_calls > 4,
            "refresh loop should keep polling past transient errors, but only \
             called get_router_events {total_calls} times (<= the 4 injected failures), \
             implying it returned/died instead of retrying"
        );

        // And it must have recovered: once get_router_events started succeeding,
        // the router got populated from the historical events.
        let snapshot = router.read().snapshot();
        assert_eq!(
            snapshot.success_events, 100,
            "router should have recovered and loaded history after the transient \
             errors cleared"
        );
    }
}

#[cfg(test)]
mod deferred_swap_drop_tests {
    use super::deferred_swap_drops_to_execute;

    /// 3-node ring: current=2, min=1, pending=1.
    /// The original bug: headroom = 2-1 = 1 > 0, so the old code fired the drop,
    /// kicking the only other peer and leaving the node isolated.
    /// Fixed: guard requires current > min + pending (2 > 1+1 = 2) → false.
    #[test]
    fn three_node_ring_no_drop_without_replacement() {
        assert_eq!(deferred_swap_drops_to_execute(2, 1, 1), 0);
    }

    /// Replacement connected in a 3-node ring: current=3, min=1, pending=1.
    /// 3 > 1+1=2 → true. One drop allowed.
    #[test]
    fn three_node_ring_drop_when_replacement_connected() {
        assert_eq!(deferred_swap_drops_to_execute(3, 1, 1), 1);
    }

    /// At min_connections with no pending drops: current=10, min=10, pending=0.
    /// No pending drops means nothing to execute.
    #[test]
    fn no_pending_drops_returns_zero() {
        assert_eq!(deferred_swap_drops_to_execute(10, 10, 0), 0);
    }

    /// Large network, no replacements yet: current=12, min=10, pending=3.
    /// 12 > 10+3=13 → false. None dropped.
    #[test]
    fn large_network_no_drop_without_replacement() {
        assert_eq!(deferred_swap_drops_to_execute(12, 10, 3), 0);
    }

    /// Large network, exactly enough for one replacement: current=13, min=10, pending=3.
    /// i=0: remaining=3, 13 > 13 → false. Still none.
    #[test]
    fn large_network_boundary_no_drop() {
        assert_eq!(deferred_swap_drops_to_execute(13, 10, 3), 0);
    }

    /// Large network, one replacement connected: current=14, min=10, pending=3.
    /// i=0: remaining=3, 14 > 13 → true (effective=13, n=1)
    /// i=1: remaining=2, 13 > 12 → true (effective=12, n=2)
    /// i=2: remaining=1, 12 > 11 → true (effective=11, n=3)
    /// All 3 dropped; after drops effective=11 ≥ min=10.
    #[test]
    fn large_network_all_replacements_connected() {
        assert_eq!(deferred_swap_drops_to_execute(14, 10, 3), 3);
    }

    /// current exactly at min with one pending: current=10, min=10, pending=1.
    /// 10 > 10+1=11 → false. No drop.
    #[test]
    fn at_min_with_pending_no_drop() {
        assert_eq!(deferred_swap_drops_to_execute(10, 10, 1), 0);
    }

    /// current = min + 1 with one pending: current=11, min=10, pending=1.
    /// 11 > 11 → false. No drop (1 extra but still need the pending slot covered).
    #[test]
    fn one_above_min_with_pending_no_drop() {
        assert_eq!(deferred_swap_drops_to_execute(11, 10, 1), 0);
    }

    /// current = min + 2 with one pending: current=12, min=10, pending=1.
    /// 12 > 11 → true. One drop allowed.
    #[test]
    fn two_above_min_with_one_pending_drops_one() {
        assert_eq!(deferred_swap_drops_to_execute(12, 10, 1), 1);
    }

    /// Overflow-safe: current=0, min=0, pending=0.
    #[test]
    fn all_zero_returns_zero() {
        assert_eq!(deferred_swap_drops_to_execute(0, 0, 0), 0);
    }

    /// current < min (below minimum, e.g. still bootstrapping): no drops.
    #[test]
    fn below_min_connections_no_drop() {
        assert_eq!(deferred_swap_drops_to_execute(5, 10, 2), 0);
    }
}

/// Source-grep pin test: lock down the set of bare `Instant::now()` call
/// sites in this file so a future change can't silently reintroduce a
/// wall-clock time read on the connection-maintenance path.
///
/// All production time reads in `ring.rs` go through `self.time_source.now()`
/// (the injectable `TimeSource`) so simulation/governance tests can drive the
/// connection-maintenance loop under `tokio::time::start_paused(true)` (or a
/// `SharedMockTimeSource`) deterministically. See #4277 and the #4260 fix that
/// migrated the first such call site (`report_contract_resource_usage`).
///
/// Two categories are deliberately exempt and are NOT bare `Instant::now()`:
///   * `boot_time::Instant::now()` / `WallClockInstant::now()` — the OS
///     suspend/resume detector in `connection_maintenance`, which by design
///     needs real wall-clock time (see `.claude/rules/code-style.md`).
///     These carry a type prefix, so they don't match a *bare* `Instant::now()`.
///
/// The only remaining bare `Instant::now()` call sites live in two
/// `#[cfg(test)]` modules:
///   * `gateway_version_probe_predicate_tests` (6 sites) — constructs inputs
///     for the pure `should_probe_gateway` predicate (no production time read).
///   * `resource_meter_bridge_tests` (8 sites) — drives the #3453 bandwidth
///     bridge directly with synthetic timestamps; `feed_peer_bandwidth_to_meter`
///     itself takes the time as a parameter, and the production caller in
///     `connection_maintenance` supplies `self.time_source.now()`.
///
/// If you add a new production time read, route it through
/// `self.time_source.now()` rather than bumping this count.
#[cfg(test)]
mod instant_now_pin_test {
    /// Number of bare `Instant::now()` call sites expected in this file, all
    /// in test code. Production code must use `self.time_source.now()`.
    /// 6 in `gateway_version_probe_predicate_tests` + 8 in
    /// `resource_meter_bridge_tests` (#3453).
    const EXPECTED_BARE_INSTANT_NOW: usize = 14;

    #[test]
    fn no_unexpected_bare_instant_now_call_sites() {
        let src = include_str!("ring.rs");
        // Build the needle from fragments so this test's own source line does
        // not itself contain a verbatim bare call-site token that it would count.
        let needle = format!("Instant{}now()", "::");
        let mut count = 0usize;
        for line in src.lines() {
            // Strip a `//` line comment so doc/comment mentions (including this
            // test's own docs) don't count.
            let code = match line.split_once("//") {
                Some((before, _)) => before,
                None => line,
            };
            // Count bare occurrences — i.e. not preceded by an identifier or
            // path separator (excludes `boot_time::Instant::now()` and
            // `WallClockInstant::now()`).
            let bytes = code.as_bytes();
            let mut idx = 0;
            while let Some(pos) = code[idx..].find(&needle) {
                let abs = idx + pos;
                let prev_is_pathy = abs
                    .checked_sub(1)
                    .map(|p| {
                        let c = bytes[p];
                        c == b':' || c == b'_' || c.is_ascii_alphanumeric()
                    })
                    .unwrap_or(false);
                if !prev_is_pathy {
                    count += 1;
                }
                idx = abs + needle.len();
            }
        }
        assert_eq!(
            count, EXPECTED_BARE_INSTANT_NOW,
            "Unexpected number of bare wall-clock time-read call sites in ring.rs. \
             Production code must read time via `self.time_source.now()` so the \
             connection-maintenance loop stays deterministic under start_paused \
             (#4277). If you intentionally added or removed a TEST-only call site, \
             update EXPECTED_BARE_INSTANT_NOW; if this fired on PRODUCTION code, \
             migrate it to `self.time_source.now()` instead of bumping the count."
        );
    }
}

#[derive(thiserror::Error, Debug)]
pub(crate) enum RingError {
    #[error(transparent)]
    ConnError(#[from] Box<node::ConnectionError>),
    /// Retained for completeness; client_events maps it to a stable
    /// `ClientError::EmptyRing`. Currently unconstructed.
    #[error("No ring connections found")]
    #[allow(dead_code)]
    EmptyRing,
    #[error("Ran out of, or haven't found any, hosting peers for contract {0}")]
    NoHostingPeers(ContractInstanceId),
    #[error("Peer has not joined the network yet (no ring location established)")]
    PeerNotJoined,
}