blvm-node 0.1.90

//! Network manager: coordinates peers, transports (TCP, Quinn, Iroh), and network-facing state.

use crate::network::protocol::cmd;
use crate::network::protocol::{AddrMessage, NetworkAddress, ProtocolMessage, ProtocolParser};
use crate::network::tcp_transport::TcpTransport;
use crate::network::transport::{Transport, TransportAddr, TransportListener, TransportPreference};
use crate::node::mempool::MempoolManager;
use crate::storage::Storage;
use crate::utils::{current_timestamp, current_timestamp_nanos};
use anyhow::Result;
use blvm_consensus::Network;
use blvm_protocol::mempool::Mempool;
use blvm_protocol::{BitcoinProtocolEngine, ConsensusProof, UtxoSet};
use std::collections::HashMap;
use std::collections::HashSet;
use std::net::IpAddr;
use std::net::SocketAddr;
use std::sync::{
    Arc,
    atomic::{AtomicBool, AtomicU64, Ordering},
};
use std::time::SystemTime;
use tokio::sync::{Mutex, RwLock, mpsc};
use tracing::{debug, error, info, warn};

fn process_ban_path(dir: &std::path::Path) -> std::path::PathBuf {
    dir.join("banlist.dat")
}

/// Read the process-local ban file. A missing file is an empty map.
/// Every stored key is `SocketAddr::new(ip, 0)`.
fn load_process_bans(dir: &std::path::Path) -> HashMap<SocketAddr, u64> {
    let Ok(bytes) = std::fs::read(process_ban_path(dir)) else {
        return HashMap::new();
    };
    if bytes.len() < 4 {
        return HashMap::new();
    }
    let count = u32::from_le_bytes(bytes[0..4].try_into().unwrap()) as usize;
    let mut offset = 4;
    let mut bans = HashMap::new();
    for _ in 0..count {
        if offset >= bytes.len() {
            break;
        }
        let family = bytes[offset];
        offset += 1;
        let ip = if family == 4 && offset + 4 <= bytes.len() {
            let octets: [u8; 4] = bytes[offset..offset + 4].try_into().unwrap();
            offset += 4;
            IpAddr::V4(octets.into())
        } else if family == 6 && offset + 16 <= bytes.len() {
            let octets: [u8; 16] = bytes[offset..offset + 16].try_into().unwrap();
            offset += 16;
            IpAddr::V6(octets.into())
        } else {
            break;
        };
        if offset + 8 > bytes.len() {
            break;
        }
        let until = u64::from_le_bytes(bytes[offset..offset + 8].try_into().unwrap());
        offset += 8;
        bans.insert(SocketAddr::new(ip, 0), until);
    }
    bans
}

fn write_process_bans(dir: &std::path::Path, bans: &HashMap<SocketAddr, u64>) {
    let mut bytes = Vec::with_capacity(4 + bans.len() * 16);
    bytes.extend_from_slice(&(bans.len() as u32).to_le_bytes());
    for (addr, until) in bans {
        match addr.ip() {
            IpAddr::V4(ip) => {
                bytes.push(4);
                bytes.extend_from_slice(&ip.octets());
            }
            IpAddr::V6(ip) => {
                bytes.push(6);
                bytes.extend_from_slice(&ip.octets());
            }
        }
        bytes.extend_from_slice(&until.to_le_bytes());
    }
    let _ = std::fs::write(process_ban_path(dir), bytes);
}

fn outbound_pick_tried() -> bool {
    SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|elapsed| elapsed.subsec_nanos() & 1 == 1)
        .unwrap_or(false)
}

use super::NetworkMessage;
use super::lan_discovery;
use super::peer;
use super::peer_manager::{PeerByteRateLimiter, PeerManager, PeerRateLimiter};
use super::transport;
use hex;
/// Get wire command string for ProtocolMessage (for MessageReceived event)
fn protocol_message_type(msg: &ProtocolMessage) -> &'static str {
    match msg {
        ProtocolMessage::Version(_) => cmd::VERSION,
        ProtocolMessage::Verack => cmd::VERACK,
        ProtocolMessage::SendHeaders => cmd::SENDHEADERS,
        ProtocolMessage::WtxidRelay => cmd::WTXIDRELAY,
        ProtocolMessage::Ping(_) => cmd::PING,
        ProtocolMessage::Pong(_) => cmd::PONG,
        ProtocolMessage::GetHeaders(_) => cmd::GETHEADERS,
        ProtocolMessage::Headers(_) => cmd::HEADERS,
        ProtocolMessage::GetBlocks(_) => cmd::GETBLOCKS,
        ProtocolMessage::Block(_) => cmd::BLOCK,
        ProtocolMessage::GetData(_) => cmd::GETDATA,
        ProtocolMessage::Inv(_) => cmd::INV,
        ProtocolMessage::NotFound(_) => cmd::NOTFOUND,
        ProtocolMessage::Reject(_) => cmd::REJECT,
        ProtocolMessage::Tx(_) => cmd::TX,
        ProtocolMessage::FeeFilter(_) => cmd::FEEFILTER,
        ProtocolMessage::MemPool => cmd::MEMPOOL,
        ProtocolMessage::SendCmpct(_) => cmd::SENDCMPCT,
        ProtocolMessage::CmpctBlock(_) => cmd::CMPCTBLOCK,
        ProtocolMessage::GetBlockTxn(_) => cmd::GETBLOCKTXN,
        ProtocolMessage::BlockTxn(_) => cmd::BLOCKTXN,
        ProtocolMessage::GetUTXOSet(_) => cmd::GETUTXOSET,
        ProtocolMessage::UTXOSet(_) => cmd::UTXOSET,
        ProtocolMessage::GetUTXOProof(_) => cmd::GETUTXOPROOF,
        ProtocolMessage::UTXOProof(_) => cmd::UTXOPROOF,
        ProtocolMessage::GetFilteredBlock(_) => cmd::GETFILTEREDBLOCK,
        ProtocolMessage::FilteredBlock(_) => cmd::FILTEREDBLOCK,
        ProtocolMessage::GetCfilters(_) => cmd::GETCFILTERS,
        ProtocolMessage::Cfilter(_) => cmd::CFILTER,
        ProtocolMessage::GetCfheaders(_) => cmd::GETCFHEADERS,
        ProtocolMessage::Cfheaders(_) => cmd::CFHEADERS,
        ProtocolMessage::GetCfcheckpt(_) => cmd::GETCFCHECKPT,
        ProtocolMessage::Cfcheckpt(_) => cmd::CFCHECKPT,
        ProtocolMessage::GetPaymentRequest(_) => cmd::GETPAYMENTREQUEST,
        ProtocolMessage::PaymentRequest(_) => cmd::PAYMENTREQUEST,
        ProtocolMessage::Payment(_) => cmd::PAYMENT,
        ProtocolMessage::PaymentACK(_) => cmd::PAYMENTACK,
        #[cfg(feature = "ctv")]
        ProtocolMessage::PaymentProof(_) => cmd::PAYMENTPROOF,
        ProtocolMessage::SettlementNotification(_) => cmd::SETTLEMENTNOTIFICATION,
        ProtocolMessage::SendPkgTxn(_) => cmd::SENDPKGTXN,
        ProtocolMessage::PkgTxn(_) => cmd::PKGTXN,
        ProtocolMessage::PkgTxnReject(_) => cmd::PKGTXNREJECT,
        ProtocolMessage::SendPackages(_) => cmd::SENDPACKAGES,
        ProtocolMessage::AncPkgInfo(_) => cmd::ANCPKGINFO,
        ProtocolMessage::GetPkgTxns(_) => cmd::GETPKGTXNS,
        ProtocolMessage::PkgTxns(_) => cmd::PKGTXNS,
        ProtocolMessage::GetBanList(_) => cmd::GETBANLIST,
        ProtocolMessage::BanList(_) => cmd::BANLIST,
        ProtocolMessage::GetAddr => cmd::GETADDR,
        ProtocolMessage::Addr(_) => cmd::ADDR,
        ProtocolMessage::AddrV2(_) => cmd::ADDRV2,
        ProtocolMessage::GetModule(_) => cmd::GETMODULE,
        ProtocolMessage::Module(_) => cmd::MODULE,
        ProtocolMessage::GetModuleByHash(_) => cmd::GETMODULEBYHASH,
        ProtocolMessage::ModuleByHash(_) => cmd::MODULEBYHASH,
        ProtocolMessage::ModuleInv(_) => cmd::MODULEINV,
        ProtocolMessage::GetModuleList(_) => cmd::GETMODULELIST,
        ProtocolMessage::ModuleList(_) => cmd::MODULELIST,
        ProtocolMessage::MeshPacket(_) => cmd::MESH,
    }
}

fn sorted_denylist_snapshot(
    set: &HashSet<blvm_protocol::Hash>,
    max: usize,
) -> (u64, bool, Vec<blvm_protocol::Hash>) {
    let total_count = set.len() as u64;
    let mut hashes: Vec<_> = set.iter().copied().collect();
    hashes.sort();
    let truncated = hashes.len() > max;
    hashes.truncate(max);
    (total_count, truncated, hashes)
}

/// Network manager that coordinates all network operations
///
/// Supports multiple transports (TCP, Quinn, Iroh) based on configuration.
pub struct NetworkManager {
    peer_manager: Arc<Mutex<PeerManager>>,
    tcp_transport: TcpTransport,
    #[cfg(feature = "quinn")]
    quinn_transport: std::sync::Mutex<Option<Arc<crate::network::quinn_transport::QuinnTransport>>>,
    #[cfg(feature = "iroh")]
    iroh_transport:
        std::sync::Mutex<Option<std::sync::Arc<crate::network::iroh_transport::IrohTransport>>>,
    transport_preference: TransportPreference,
    peer_tx: mpsc::UnboundedSender<NetworkMessage>,
    peer_rx: Arc<tokio::sync::Mutex<mpsc::UnboundedReceiver<NetworkMessage>>>,
    /// Block filter service for BIP157/158
    filter_service: crate::network::filter_service::BlockFilterService,
    /// Peer inventory tracking (inv relay)
    inventory: Arc<std::sync::Mutex<super::inventory::InventoryManager>>,
    /// Consensus engine for mempool acceptance
    consensus: ConsensusProof,
    /// Shared UTXO snapshot for mempool fee checks (synced from chainstate storage).
    utxo_set: Arc<Mutex<UtxoSet>>,
    /// Shared mempool
    mempool: Arc<Mutex<Mempool>>,
    /// Protocol engine for network message processing
    protocol_engine: Option<Arc<BitcoinProtocolEngine>>,
    /// Storage for chain state access
    storage: Option<Arc<Storage>>,
    /// Mempool manager for transaction access
    mempool_manager: Option<Arc<MempoolManager>>,
    /// Module registry for serving modules via P2P
    module_registry:
        Arc<tokio::sync::Mutex<Option<Arc<crate::module::registry::client::ModuleRegistry>>>>,
    /// Node API. Paid GetModule asks blvm-payment through this.
    node_api: Arc<tokio::sync::Mutex<Option<Arc<dyn crate::module::traits::NodeAPI>>>>,
    /// Node payment address script (for module downloads - 10% fee)
    node_payment_script: Arc<tokio::sync::Mutex<Option<Vec<u8>>>>,
    /// Module encryption for encrypted module serving
    module_encryption:
        Arc<tokio::sync::Mutex<Option<Arc<crate::module::encryption::ModuleEncryption>>>>,
    /// Modules directory for encrypted/decrypted module storage
    modules_dir: Arc<tokio::sync::Mutex<Option<std::path::PathBuf>>>,
    /// Event publisher for module event notifications (optional)
    event_publisher:
        Arc<tokio::sync::Mutex<Option<Arc<crate::node::event_publisher::EventPublisher>>>>,
    /// Peer state storage (per-connection state)
    /// Read-heavy: many reads to check peer state, fewer writes when updating state
    peer_states: Arc<RwLock<HashMap<SocketAddr, blvm_protocol::network::PeerState>>>,
    /// Persistent peer list (peers to connect to on startup)
    persistent_peers: Arc<Mutex<HashSet<SocketAddr>>>,
    /// Eclipse attack prevention: track peer diversity
    /// Maps IP address prefixes (first 3 octets) to connection count
    /// Prevents too many connections from same IP range
    peer_diversity: Arc<Mutex<HashMap<[u8; 3], usize>>>,
    /// Network active state (true = enabled, false = disabled)
    network_active: Arc<Mutex<bool>>,
    /// Ban list (banned peers with unban timestamp)
    /// Read-heavy: many reads to check if peer is banned, fewer writes when banning/unbanning
    ban_list: Arc<RwLock<HashMap<SocketAddr, u64>>>, // addr -> unban timestamp
    /// Directory of the process-local ban file. Absent keeps the map in memory.
    ban_store_dir: Option<std::path::PathBuf>,
    /// Per-IP connection count (to prevent Sybil attacks)
    connections_per_ip: Arc<Mutex<HashMap<std::net::IpAddr, usize>>>,
    /// Per-peer message rate limiting (token bucket)
    peer_message_rates: Arc<Mutex<HashMap<SocketAddr, PeerRateLimiter>>>,
    /// Iroh handshake state, keyed by endpoint bytes.
    iroh_peer_states: Arc<Mutex<HashMap<Vec<u8>, blvm_protocol::network::PeerState>>>,
    /// Iroh bans, keyed by endpoint bytes. TCP bans stay on `ban_list`.
    iroh_bans: Arc<RwLock<HashMap<Vec<u8>, u64>>>,
    /// Iroh message-rate entries, keyed by endpoint bytes.
    iroh_message_rates: Arc<Mutex<HashMap<Vec<u8>, PeerRateLimiter>>>,
    /// Per-peer transaction rate limiting (separate from message rate limiting)
    peer_tx_rate_limiters: Arc<Mutex<HashMap<SocketAddr, PeerRateLimiter>>>,
    /// Per-peer transaction byte rate limiting (bytes per second)
    peer_tx_byte_rate_limiters: Arc<Mutex<HashMap<SocketAddr, PeerByteRateLimiter>>>,
    /// Mempool policy configuration for rate limits
    mempool_policy_config: Option<Arc<crate::config::MempoolPolicyConfig>>,
    /// Spam ban configuration
    spam_ban_config: Option<Arc<crate::config::SpamBanConfig>>,
    /// Pending blocks queue (blocks received via BlockReceived message)
    /// Separate from main message channel to avoid draining other messages
    pending_blocks: Arc<std::sync::Mutex<std::collections::VecDeque<Vec<u8>>>>,
    /// Service bits a loaded module asked this node to advertise, keyed by module name.
    module_service_bits: Arc<std::sync::Mutex<HashMap<String, u64>>>,
    /// Compact blocks waiting on `blocktxn` for the positions `getblocktxn` asked for.
    pub(crate) pending_compact:
        Arc<std::sync::Mutex<HashMap<[u8; 32], crate::network::compact_blocks::CompactAssembly>>>,
    /// Per-peer transaction getdata, capped separately from the mempool.
    tx_requests: Arc<std::sync::Mutex<crate::network::tx_inflight::TxRequestTracker>>,
    /// Network statistics
    /// Optimization: Use AtomicU64 for lock-free updates
    bytes_sent: Arc<AtomicU64>,
    bytes_received: Arc<AtomicU64>,
    /// Mapping from SocketAddr to TransportAddr (for Iroh peers that use placeholder SocketAddr)
    socket_to_transport: Arc<Mutex<HashMap<SocketAddr, TransportAddr>>>,
    /// Request ID counter for async request-response patterns
    /// Optimization: Use AtomicU64 for lock-free updates (eliminates need for block_in_place)
    request_id_counter: Arc<AtomicU64>,
    /// Pending async requests with metadata
    /// Key: request_id, Value: (sender, peer_addr, timestamp)
    pending_requests: Arc<Mutex<HashMap<u64, PendingRequest>>>,
    /// Pending Headers requests (for IBD) - supports pipelining with queue per peer
    /// Key: peer_addr, Value: queue of sender channels (FIFO order)
    pending_headers_requests: Arc<
        Mutex<
            HashMap<
                SocketAddr,
                std::collections::VecDeque<
                    tokio::sync::oneshot::Sender<Vec<blvm_protocol::BlockHeader>>,
                >,
            >,
        >,
    >,
    /// Pending Block requests (for IBD)
    /// Key: (peer_ip, block_hash) - uses IpAddr to match regardless of port (inbound vs outbound)
    ///
    /// Value is a `Vec` of senders, not a single sender: during IBD the coordinator's stall
    /// recovery legitimately issues several overlapping requests for the *same* (peer, block)
    /// (e.g. "up to 3 fetchers/height" plus periodic requeues). A single-sender map made each
    /// new registration `insert`-overwrite (and drop) the previous request's sender, killing an
    /// in-flight download's receiver; worse, `complete_block_request` would then deliver the
    /// arriving block to whichever lone sender survived — frequently one whose receiver had
    /// already been dropped — silently losing the block (`let _ = sender.send(..)`). With the
    /// relentless requeue storm this trapped a needed height forever (observed: h=575630 routed
    /// every ~600ms yet timing out for 30s, validation frozen). Keeping every sender and
    /// delivering to the first *live* one guarantees the currently-waiting download receives it.
    /// IBD GetData delivery: `(block, witnesses, optional original P2P payload)`.
    /// Third field is W5/N1 wire-bytes store input (`data[24..]` from the frame).
    pending_block_requests: Arc<
        Mutex<
            HashMap<
                (IpAddr, blvm_protocol::Hash),
                Vec<
                    tokio::sync::oneshot::Sender<(
                        blvm_protocol::Block,
                        Vec<Vec<blvm_protocol::segwit::Witness>>,
                        Option<Vec<u8>>,
                    )>,
                >,
            >,
        >,
    >,
    /// DoS protection manager
    dos_protection: Arc<super::dos_protection::DosProtectionManager>,
    /// IBD bandwidth protection manager
    ibd_protection: Arc<super::ibd_protection::IbdProtectionManager>,
    /// Unified bandwidth protection manager (extends IBD protection)
    bandwidth_protection: Arc<super::bandwidth_protection::BandwidthProtectionManager>,
    /// Replay protection for custom protocol messages
    replay_protection: Arc<super::replay_protection::ReplayProtection>,
    /// Pending ban shares (for periodic sharing)
    pending_ban_shares: Arc<Mutex<Vec<(SocketAddr, u64, String)>>>, // (addr, unban_timestamp, reason)
    /// Ban list sharing configuration. `None` means sharing is off and GetBanList / BanList stay unhandled.
    pub(crate) ban_list_sharing_config: Option<crate::config::BanListSharingConfig>,
    /// Spam violation tracking per peer (for spam-specific banning)
    /// Maps SocketAddr -> violation count
    peer_spam_violations: Arc<Mutex<HashMap<SocketAddr, usize>>>,
    /// Governance message relay configuration
    #[cfg(feature = "governance")]
    governance_config: Option<crate::config::GovernanceConfig>,
    /// Address database for peer discovery
    /// Read-heavy: many reads to query addresses, fewer writes when adding addresses
    address_database: Arc<RwLock<super::address_db::AddressDatabase>>,
    /// Last time we sent an addr message to each peer (Unix timestamp).
    last_addr_sent: Arc<Mutex<HashMap<SocketAddr, u64>>>,
    /// Enable self-advertisement (send own address to peers)
    enable_self_advertisement: bool,
    /// Bind TCP, Quinn, and Iroh listeners. False still constructs the manager with a socket address.
    listen: bool,
    /// Request timeout configuration
    request_timeout_config: Arc<crate::config::RequestTimeoutConfig>,
    /// Protocol limits (message size, addr, inv, headers)
    protocol_limits: Arc<crate::config::ProtocolLimitsConfig>,
    /// Background task interval configuration
    background_task_config: Arc<crate::config::BackgroundTaskConfig>,
    /// Relay section. Absent config uses the same defaults as `RelayPolicies::default`.
    relay_policies: super::relay::RelayPolicies,
    /// Running relay, including Dandelion stem parameters when that relay is enabled.
    relay_manager: std::sync::Mutex<super::relay::RelayManager>,
    /// Tip-follow `getheaders` wait. Taken from `ibd.headers_timeout_secs` (default 30).
    headers_probe_timeout_secs: u64,
    /// Outbound target, addr relay interval, and addr message size.
    network_timing: Arc<crate::config::NetworkTimingConfig>,
    /// Message token bucket `(burst, rate)` when `[peer_rate_limiting]` is set.
    peer_message_rate: Option<(u32, u32)>,
    /// Advertise and serve BIP157 compact filters.
    serve_compact_filters: bool,
    /// Peer reconnection queue (exponential backoff)
    /// Maps SocketAddr to (attempts, last_attempt_timestamp, quality_score)
    peer_reconnection_queue: Arc<Mutex<HashMap<SocketAddr, (u32, u64, f64)>>>,
    /// Block hashes modules merged into the full-block serve denylist (see `merge_block_serve_denylist`).
    block_serve_denylist: Arc<parking_lot::RwLock<HashSet<blvm_protocol::Hash>>>,
    /// Txids merged into the full-tx serve denylist (see `merge_tx_serve_denylist`).
    tx_serve_denylist: Arc<parking_lot::RwLock<HashSet<blvm_protocol::Hash>>>,
    /// When true, refuse all full-block answers on `getdata` (operational maintenance).
    block_serve_maintenance: Arc<AtomicBool>,

    /// IPs evicted from IBD because they consistently failed to serve historical blocks
    /// (NODE_NETWORK_LIMITED / pruned).  `spawn_outbound_reconnect_attempt` will not
    /// re-connect to any address whose IP is in this set.
    pub(crate) ibd_evicted_ips: Arc<std::sync::RwLock<HashSet<std::net::IpAddr>>>,

    /// Nonces we placed in our own outbound Version messages.
    /// Any incoming Version that echoes one of these nonces is a self-connection.
    pub(crate) local_version_nonces: Arc<std::sync::Mutex<HashSet<u64>>>,
    /// Peers that have already received a GetAddr response this connection.
    /// We answer GetAddr at most once per connection (Core behaviour).
    pub(crate) getaddr_responded: Arc<std::sync::Mutex<HashSet<SocketAddr>>>,
    /// When false, P2P Stratum TLV demux is skipped (requires `stratum-v2` feature).
    #[cfg(feature = "stratum-v2")]
    p2p_stratum_demux_enabled: bool,
}

/// Pending request metadata (pub(crate) for utxo_commitments_client)
pub(crate) struct PendingRequest {
    /// Channel to send response to
    sender: tokio::sync::oneshot::Sender<Vec<u8>>,
    /// Peer address that sent the request
    peer_addr: SocketAddr,
    /// Timestamp when request was registered (Unix timestamp)
    timestamp: u64,
    /// Request priority (0 = normal, higher = more important)
    priority: u8,
    /// Number of retry attempts
    retry_count: u8,
}

impl PendingRequest {
    pub(crate) fn timestamp(&self) -> u64 {
        self.timestamp
    }

    pub(crate) fn send_response(self, data: Vec<u8>) {
        let _ = self.sender.send(data);
    }
}

impl NetworkManager {
    /// Create a new network manager with default TCP-only transport
    pub fn new(_listen_addr: SocketAddr) -> Self {
        Self::with_config(_listen_addr, 100, TransportPreference::TCP_ONLY, None)
    }

    /// Create a new network manager with configuration
    pub fn with_config(
        _listen_addr: SocketAddr,
        max_peers: usize,
        preference: TransportPreference,
        config: Option<&crate::config::NodeConfig>,
    ) -> Self {
        let (peer_tx, peer_rx) = mpsc::unbounded_channel();
        let peer_rx = Arc::new(tokio::sync::Mutex::new(peer_rx));

        // Use config for DoS protection
        let dos_config_default = crate::config::DosProtectionConfig::default();
        let dos_config = config
            .and_then(|c| c.dos_protection.as_ref())
            .unwrap_or(&dos_config_default);

        let dos_protection = Arc::new(
            super::dos_protection::DosProtectionManager::with_ban_settings(
                dos_config.max_connections_per_window,
                dos_config.window_seconds,
                dos_config.max_message_queue_size,
                dos_config.max_active_connections,
                dos_config.auto_ban_threshold,
                dos_config.ban_duration_seconds,
            ),
        );

        // Initialize IBD protection (bandwidth exhaustion attack mitigation)
        let ibd_protection = if let Some(ibd_config) =
            config.and_then(|c| c.ibd_protection.as_ref())
        {
            let mut ibd_protection_config = super::ibd_protection::IbdProtectionConfig::default();
            // A product that does not fit stays at the default byte cap.
            let assign_gb = |slot: &mut u64, gb: u64| {
                if let Some(bytes) = crate::config::bandwidth_gb_to_bytes(gb) {
                    *slot = bytes;
                }
            };
            assign_gb(
                &mut ibd_protection_config.max_bandwidth_per_peer_per_day,
                ibd_config.max_bandwidth_per_peer_per_day_gb,
            );
            assign_gb(
                &mut ibd_protection_config.max_bandwidth_per_peer_per_hour,
                ibd_config.max_bandwidth_per_peer_per_hour_gb,
            );
            assign_gb(
                &mut ibd_protection_config.max_bandwidth_per_ip_per_day,
                ibd_config.max_bandwidth_per_ip_per_day_gb,
            );
            assign_gb(
                &mut ibd_protection_config.max_bandwidth_per_ip_per_hour,
                ibd_config.max_bandwidth_per_ip_per_hour_gb,
            );
            assign_gb(
                &mut ibd_protection_config.max_bandwidth_per_subnet_per_day,
                ibd_config.max_bandwidth_per_subnet_per_day_gb,
            );
            assign_gb(
                &mut ibd_protection_config.max_bandwidth_per_subnet_per_hour,
                ibd_config.max_bandwidth_per_subnet_per_hour_gb,
            );
            ibd_protection_config.max_concurrent_ibd_serving =
                ibd_config.max_concurrent_ibd_serving;
            ibd_protection_config.ibd_request_cooldown_seconds =
                ibd_config.ibd_request_cooldown_seconds;
            ibd_protection_config.suspicious_reconnection_threshold =
                ibd_config.suspicious_reconnection_threshold;
            ibd_protection_config.reputation_ban_threshold = ibd_config.reputation_ban_threshold;
            ibd_protection_config.reputation_penalty_per_request =
                ibd_config.reputation_penalty_per_request;
            ibd_protection_config.enable_emergency_throttle = ibd_config.enable_emergency_throttle;
            ibd_protection_config.emergency_throttle_percent =
                ibd_config.emergency_throttle_percent;
            Arc::new(super::ibd_protection::IbdProtectionManager::with_config(
                ibd_protection_config,
            ))
        } else {
            Arc::new(super::ibd_protection::IbdProtectionManager::new())
        };

        // Initialize unified bandwidth protection (extends IBD protection)
        let bandwidth_protection = Arc::new(
            super::bandwidth_protection::BandwidthProtectionManager::new(Arc::clone(
                &ibd_protection,
            )),
        );

        // Use config for address database
        let addr_db_config_default = crate::config::AddressDatabaseConfig::default();
        let addr_db_config = config
            .and_then(|c| c.address_database.as_ref())
            .unwrap_or(&addr_db_config_default);

        let address_database = Arc::new(RwLock::new(
            super::address_db::AddressDatabase::with_expiration(
                addr_db_config.max_addresses,
                addr_db_config.expiration_seconds,
            ),
        ));

        // Use config for request timeouts
        let timeout_config_default = crate::config::RequestTimeoutConfig::default();
        let timeout_config = config
            .and_then(|c| c.request_timeouts.as_ref())
            .unwrap_or(&timeout_config_default);
        let request_timeout_config = Arc::new(timeout_config.clone());

        // Use config for protocol limits
        let limits_config_default = crate::config::ProtocolLimitsConfig::default();
        let mut limits_config = config
            .and_then(|c| c.protocol_limits.clone())
            .unwrap_or(limits_config_default);
        limits_config.clamp_message_ceiling();
        let tcp_transport =
            TcpTransport::with_max_message_length(limits_config.max_protocol_message_length);
        let protocol_limits = Arc::new(limits_config);

        // Use config for background task intervals
        let bg_config_default = crate::config::BackgroundTaskConfig::default();
        let bg_config = config
            .and_then(|c| c.background_tasks.as_ref())
            .unwrap_or(&bg_config_default);
        let background_task_config = Arc::new(bg_config.clone());
        let mut relay_policies = config
            .and_then(|c| c.relay.as_ref())
            .map(super::relay::RelayPolicies::from_relay_config)
            .unwrap_or_default();
        #[cfg(feature = "dandelion")]
        if config.and_then(|c| c.dandelion.as_ref()).is_some()
            && config
                .and_then(|c| c.relay.as_ref())
                .map(|relay| relay.enable_dandelion)
                .unwrap_or(true)
        {
            // A `[dandelion]` section turns stem relay on unless `[relay] enable_dandelion`
            // is explicitly false (including `--disable-dandelion`).
            relay_policies.enable_dandelion = true;
        }
        #[cfg(feature = "dandelion")]
        let dandelion_cfg = config.and_then(|c| c.dandelion.clone());
        #[cfg(feature = "dandelion")]
        let relay_manager =
            std::sync::Mutex::new(super::relay::RelayManager::with_policies_and_dandelion(
                relay_policies.clone(),
                dandelion_cfg.as_ref(),
            ));
        #[cfg(not(feature = "dandelion"))]
        let relay_manager = std::sync::Mutex::new(super::relay::RelayManager::with_policies(
            relay_policies.clone(),
        ));
        let network_timing = Arc::new(
            config
                .and_then(|c| c.network_timing.clone())
                .unwrap_or_default(),
        );
        let peer_message_rate = config.and_then(|c| {
            c.peer_rate_limiting
                .as_ref()
                .map(|limits| (limits.default_burst, limits.default_rate))
        });
        let serve_compact_filters = config.map(|c| c.serve_compact_filters).unwrap_or(false);
        let ban_store_dir = config
            .and_then(|c| c.storage.as_ref())
            .map(|storage| std::path::PathBuf::from(&storage.data_dir));
        let loaded_bans = ban_store_dir
            .as_ref()
            .map(|dir| load_process_bans(dir))
            .unwrap_or_default();
        let headers_probe_timeout_secs = config
            .and_then(|c| c.ibd.as_ref())
            .map(|ibd| ibd.headers_timeout_secs)
            .unwrap_or(30);

        #[cfg(feature = "stratum-v2")]
        let p2p_stratum_demux_enabled =
            crate::config::p2p_stratum_demux_enabled(config.and_then(|c| c.stratum_v2.as_ref()));

        Self {
            peer_manager: Arc::new(Mutex::new(PeerManager::new(max_peers))),
            peer_diversity: Arc::new(Mutex::new(HashMap::new())),
            tcp_transport,
            #[cfg(feature = "quinn")]
            quinn_transport: std::sync::Mutex::new(None),
            #[cfg(feature = "iroh")]
            iroh_transport: std::sync::Mutex::new(None),
            transport_preference: preference,
            peer_tx,
            peer_rx,
            filter_service: crate::network::filter_service::BlockFilterService::new(),
            inventory: Arc::new(std::sync::Mutex::new(
                super::inventory::InventoryManager::new(),
            )),
            consensus: ConsensusProof::new(),
            utxo_set: Arc::new(Mutex::new(UtxoSet::default())),
            mempool: Arc::new(Mutex::new(Mempool::new())),
            protocol_engine: None,
            storage: None,
            mempool_manager: None,
            module_registry: Arc::new(tokio::sync::Mutex::new(None)),
            node_api: Arc::new(tokio::sync::Mutex::new(None)),
            node_payment_script: Arc::new(tokio::sync::Mutex::new(None)),
            module_encryption: Arc::new(tokio::sync::Mutex::new(None)),
            modules_dir: Arc::new(tokio::sync::Mutex::new(None)),
            event_publisher: Arc::new(tokio::sync::Mutex::new(None)),
            peer_states: Arc::new(RwLock::new(HashMap::new())),
            persistent_peers: Arc::new(Mutex::new(HashSet::new())),
            network_active: Arc::new(Mutex::new(true)),
            ban_list: Arc::new(RwLock::new(loaded_bans)),
            ban_store_dir,
            connections_per_ip: Arc::new(Mutex::new(HashMap::new())),
            peer_message_rates: Arc::new(Mutex::new(HashMap::new())),
            iroh_peer_states: Arc::new(Mutex::new(HashMap::new())),
            iroh_bans: Arc::new(RwLock::new(HashMap::new())),
            iroh_message_rates: Arc::new(Mutex::new(HashMap::new())),
            peer_tx_rate_limiters: Arc::new(Mutex::new(HashMap::new())),
            peer_tx_byte_rate_limiters: Arc::new(Mutex::new(HashMap::new())),
            mempool_policy_config: config
                .and_then(|c| c.mempool.as_ref())
                .map(|c| Arc::new(c.clone())),
            spam_ban_config: config
                .and_then(|c| c.spam_ban.as_ref())
                .map(|c| Arc::new(c.clone())),
            peer_spam_violations: Arc::new(Mutex::new(HashMap::new())),
            pending_blocks: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new())),
            module_service_bits: Arc::new(std::sync::Mutex::new(HashMap::new())),
            pending_compact: Arc::new(std::sync::Mutex::new(HashMap::new())),
            tx_requests: Arc::new(std::sync::Mutex::new(
                crate::network::tx_inflight::TxRequestTracker::default(),
            )),
            bytes_sent: Arc::new(AtomicU64::new(0)),
            bytes_received: Arc::new(AtomicU64::new(0)),
            socket_to_transport: Arc::new(Mutex::new(HashMap::new())),
            request_id_counter: Arc::new(AtomicU64::new(0)),
            pending_requests: Arc::new(Mutex::new(HashMap::new())),
            pending_headers_requests: Arc::new(Mutex::new(HashMap::new())),
            pending_block_requests: Arc::new(Mutex::new(HashMap::new())),
            dos_protection,
            ibd_protection,
            bandwidth_protection,
            replay_protection: Arc::new({
                let rp_config = config.and_then(|c| c.replay_protection.as_ref());
                if let Some(c) = rp_config {
                    super::replay_protection::ReplayProtection::with_config(
                        std::time::Duration::from_secs(c.cleanup_interval_secs),
                        std::time::Duration::from_secs(c.message_id_expiration_secs),
                        std::time::Duration::from_secs(c.request_id_expiration_secs),
                        c.future_tolerance_secs,
                    )
                } else {
                    super::replay_protection::ReplayProtection::new()
                }
            }),
            pending_ban_shares: Arc::new(Mutex::new(Vec::new())),
            ban_list_sharing_config: config.and_then(|c| c.ban_list_sharing.clone()),
            #[cfg(feature = "governance")]
            governance_config: config.and_then(|c| c.governance.clone()),
            address_database,
            last_addr_sent: Arc::new(Mutex::new(HashMap::new())),
            enable_self_advertisement: config.map(|c| c.enable_self_advertisement).unwrap_or(true),
            listen: config.map(|c| c.listen).unwrap_or(true),
            request_timeout_config,
            protocol_limits,
            background_task_config,
            relay_policies,
            relay_manager,
            headers_probe_timeout_secs,
            network_timing,
            peer_message_rate,
            serve_compact_filters,
            peer_reconnection_queue: Arc::new(Mutex::new(HashMap::new())),
            block_serve_denylist: Arc::new(parking_lot::RwLock::new(HashSet::new())),
            tx_serve_denylist: Arc::new(parking_lot::RwLock::new(HashSet::new())),
            block_serve_maintenance: Arc::new(AtomicBool::new(false)),
            ibd_evicted_ips: Arc::new(std::sync::RwLock::new(HashSet::new())),
            local_version_nonces: Arc::new(std::sync::Mutex::new(HashSet::new())),
            getaddr_responded: Arc::new(std::sync::Mutex::new(HashSet::new())),
            #[cfg(feature = "stratum-v2")]
            p2p_stratum_demux_enabled,
        }
    }

    /// Set dependencies for protocol message processing
    pub fn with_dependencies(
        mut self,
        protocol_engine: Arc<BitcoinProtocolEngine>,
        storage: Arc<Storage>,
        mempool_manager: Arc<MempoolManager>,
    ) -> Self {
        self.protocol_engine = Some(protocol_engine);
        self.storage = Some(storage);
        self.mempool_manager = Some(mempool_manager);
        self
    }

    /// Set module registry for serving modules via P2P
    pub fn with_module_registry(
        mut self,
        module_registry: Arc<crate::module::registry::client::ModuleRegistry>,
    ) -> Self {
        *self.module_registry.blocking_lock() = Some(module_registry);
        self
    }

    /// Set module registry for serving modules via P2P
    pub async fn set_module_registry(
        &self,
        module_registry: Arc<crate::module::registry::client::ModuleRegistry>,
    ) {
        *self.module_registry.lock().await = Some(module_registry);
    }

    /// Remember the node API so paid GetModule can ask blvm-payment.
    pub async fn set_node_api(&self, api: Arc<dyn crate::module::traits::NodeAPI>) {
        *self.node_api.lock().await = Some(api);
    }

    /// Set node payment address script (for module downloads)
    pub async fn set_node_payment_script(&self, script: Option<Vec<u8>>) {
        *self.node_payment_script.lock().await = script;
    }

    /// Set module encryption for encrypted module serving
    pub async fn set_module_encryption(
        &self,
        encryption: Arc<crate::module::encryption::ModuleEncryption>,
    ) {
        *self.module_encryption.lock().await = Some(encryption);
    }

    /// Set modules directory for encrypted/decrypted module storage
    pub async fn set_modules_dir(&self, modules_dir: std::path::PathBuf) {
        *self.modules_dir.lock().await = Some(modules_dir);
    }

    /// Set event publisher for network event notifications
    pub async fn set_event_publisher(
        &self,
        event_publisher: Option<Arc<crate::node::event_publisher::EventPublisher>>,
    ) {
        *self.event_publisher.lock().await = event_publisher;
    }

    /// Stable string identity for a P2P transport (TCP/QUIC `SocketAddr` display), for module-visible peer maps.
    pub(crate) fn companion_udp_p2p_peer_key(transport: &TransportAddr) -> Option<String> {
        match transport {
            TransportAddr::Tcp(s) => Some(s.to_string()),
            #[cfg(feature = "quinn")]
            TransportAddr::Quinn(s) => Some(s.to_string()),
            #[cfg(feature = "iroh")]
            TransportAddr::Iroh(_) => None,
        }
    }

    /// Derived companion UDP address: same IP as P2P, port = P2P port + 1 (Commons NODE_FIBRE convention).
    pub(crate) fn companion_udp_addr_for_p2p_peer(transport: &TransportAddr) -> Option<SocketAddr> {
        match transport {
            TransportAddr::Tcp(s) => Some(SocketAddr::new(s.ip(), s.port().saturating_add(1))),
            #[cfg(feature = "quinn")]
            TransportAddr::Quinn(s) => Some(SocketAddr::new(s.ip(), s.port().saturating_add(1))),
            #[cfg(feature = "iroh")]
            TransportAddr::Iroh(_) => None,
        }
    }

    /// After a peer `version`: if they advertise the commons auxiliary-block-relay service bit (`NODE_FIBRE`),
    /// publish the companion UDP binding for loadable modules (e.g. `blvm-fibre`).
    pub(crate) async fn publish_companion_udp_peer_after_handshake(
        &self,
        transport: &TransportAddr,
        version_msg: &crate::network::protocol::VersionMessage,
    ) {
        if !version_msg.supports_fibre() {
            return;
        }
        let Some(p2p_peer_addr) = Self::companion_udp_p2p_peer_key(transport) else {
            return;
        };
        let Some(udp) = Self::companion_udp_addr_for_p2p_peer(transport) else {
            return;
        };
        let udp_addr_str = udp.to_string();

        let ep_guard = self.event_publisher.lock().await;
        if let Some(ep) = ep_guard.as_ref() {
            let ep = Arc::clone(ep);
            drop(ep_guard);
            ep.publish_companion_udp_peer_registered(&p2p_peer_addr, &udp_addr_str)
                .await;
        }
    }

    /// On disconnect: if the peer had advertised `NODE_FIBRE`, clear the binding for modules.
    pub(crate) async fn clear_companion_udp_peer_on_disconnect(&self, transport: &TransportAddr) {
        let Some(p2p_peer_addr) = Self::companion_udp_p2p_peer_key(transport) else {
            return;
        };
        let had_node_fibre = {
            let pm = self.peer_manager.lock().await;
            pm.get_peer(transport)
                .map(|p| p.has_service(crate::network::protocol::NODE_FIBRE))
                .unwrap_or(false)
        };
        if !had_node_fibre {
            return;
        }

        let ep_guard = self.event_publisher.lock().await;
        if let Some(ep) = ep_guard.as_ref() {
            let ep = Arc::clone(ep);
            drop(ep_guard);
            ep.publish_companion_udp_peer_unregistered(&p2p_peer_addr)
                .await;
        }
    }

    /// Check if address is local (loopback, private, link-local)
    pub(crate) fn is_local_address(addr: &SocketAddr) -> bool {
        match addr.ip() {
            IpAddr::V4(ip) => ip.is_loopback() || ip.is_private() || ip.is_link_local(),
            IpAddr::V6(ip) => ip.is_loopback() || ip.is_unspecified(),
        }
    }

    /// Check if address is onion (Tor). Always returns false.
    /// Onion detection would require NetworkAddress (AddrV2) layer; SocketAddr is IP:port only.
    /// Onion connections arrive via proxy; we see the proxy's address, not the .onion hostname.
    pub(crate) fn is_onion_address(_addr: &SocketAddr) -> bool {
        false
    }

    /// Handle protocol violation: disconnect peer, optionally ban.
    /// Returns Ok(()) if peer has manual/NoBan (don't disconnect, but don't process message).
    /// Returns Err if we disconnected (caller should return early).
    pub(crate) async fn disconnect_for_protocol_violation(
        &self,
        peer_addr: SocketAddr,
        violation_msg: &str,
        ban_for_violation: bool,
    ) -> Result<(), anyhow::Error> {
        let mut pm = self.peer_manager.lock().await;
        let should_disconnect = if let Some(peer) = pm.get_peer(&TransportAddr::Tcp(peer_addr)) {
            !peer.is_manual() && !peer.has_noban_permission()
        } else {
            true
        };
        drop(pm);

        if should_disconnect {
            if ban_for_violation
                && !Self::is_local_address(&peer_addr)
                && !Self::is_onion_address(&peer_addr)
            {
                let ban_until = current_timestamp() + 24 * 60 * 60;
                let mut ban_list = self.ban_list.write().await;
                ban_list.insert(SocketAddr::new(peer_addr.ip(), 0), ban_until);
                let snapshot = ban_list.clone();
                drop(ban_list);
                self.persist_process_bans(&snapshot);
            } else if ban_for_violation {
                warn!(
                    "Disconnecting local/onion peer {} for {} (not banning)",
                    peer_addr, violation_msg
                );
            }
            {
                let mut pm = self.peer_manager.lock().await;
                if let Some(peer) = pm.get_peer_mut(&TransportAddr::Tcp(peer_addr)) {
                    peer.shutdown_connection();
                }
            }
            let _ = self
                .peer_tx
                .send(NetworkMessage::PeerDisconnected(TransportAddr::Tcp(
                    peer_addr,
                )));
            Err(anyhow::anyhow!("{violation_msg}"))
        } else {
            warn!(
                "Peer {} has manual/NoBan permission, not disconnecting for {}",
                peer_addr, violation_msg
            );
            Ok(())
        }
    }

    /// Check Tx message rate limits and spam. Returns true if the tx should be dropped.
    async fn should_drop_tx(&self, peer_addr: SocketAddr, data: &[u8]) -> bool {
        use super::bandwidth_protection::ServiceType;
        let tx_bytes = data.len() as u64;

        match self
            .bandwidth_protection
            .check_service_request(ServiceType::TransactionRelay, peer_addr)
            .await
        {
            Ok(true) => {}
            Ok(false) => {
                warn!(
                    "Transaction relay bandwidth limit exceeded for IP {} (peer {}), dropping transaction",
                    peer_addr.ip(),
                    peer_addr
                );
                return true;
            }
            Err(e) => {
                warn!("Bandwidth check error for transaction relay: {}", e);
            }
        }

        let (burst, rate) = self
            .mempool_policy_config
            .as_ref()
            .map(|cfg| (cfg.tx_rate_limit_burst, cfg.tx_rate_limit_per_sec))
            .unwrap_or((10, 1));

        let should_process = {
            let mut tx_rates = self.peer_tx_rate_limiters.lock().await;
            let limiter = tx_rates
                .entry(peer_addr)
                .or_insert_with(|| PeerRateLimiter::new(burst, rate));
            limiter.check_and_consume()
        };

        if !should_process {
            warn!(
                "Transaction rate limit exceeded for peer {}, dropping transaction",
                peer_addr
            );
            return true;
        }

        let (byte_burst, byte_rate) = self
            .mempool_policy_config
            .as_ref()
            .map(|cfg| (cfg.tx_byte_rate_burst, cfg.tx_byte_rate_limit))
            .unwrap_or((1_000_000, 100_000));

        let should_process_bytes = {
            let mut byte_rates = self.peer_tx_byte_rate_limiters.lock().await;
            let limiter = byte_rates
                .entry(peer_addr)
                .or_insert_with(|| PeerByteRateLimiter::new(byte_burst, byte_rate));
            limiter.check_and_consume(tx_bytes)
        };

        if !should_process_bytes {
            warn!(
                "Transaction byte rate limit exceeded for peer {} ({} bytes), dropping transaction",
                peer_addr, tx_bytes
            );
            return true;
        }

        self.bandwidth_protection
            .record_service_bandwidth(ServiceType::TransactionRelay, peer_addr, tx_bytes)
            .await;

        use blvm_protocol::spam_filter::SpamFilter;
        if data.len() <= 4 {
            return false;
        }
        if let Ok(tx_msg) = bincode::deserialize::<crate::network::protocol::TxMessage>(&data[4..])
        {
            let tx = &tx_msg.transaction;
            let spam_filter = SpamFilter::new();
            let witness_ref = if tx_msg.witnesses.is_empty() {
                None
            } else {
                Some(tx_msg.witnesses.as_slice())
            };
            let result = spam_filter.is_spam_with_witness(tx, witness_ref, None);

            if result.is_spam {
                let mut violations = self.peer_spam_violations.lock().await;
                let violation_count = violations.entry(peer_addr).or_insert(0);
                *violation_count += 1;
                let current_count = *violation_count;
                drop(violations);

                if let Some(ban_config) = self.spam_ban_config.as_ref() {
                    if current_count >= ban_config.spam_ban_threshold {
                        let unban_timestamp = crate::utils::current_timestamp()
                            + ban_config.spam_ban_duration_seconds;
                        warn!(
                            "Auto-banning peer {} for spam violations ({} violations, unban at {})",
                            peer_addr, current_count, unban_timestamp
                        );
                        self.ban_peer(peer_addr, unban_timestamp);
                        return true;
                    }
                }

                debug!(
                    "Spam transaction from peer {} (violation count: {})",
                    peer_addr, current_count
                );
            }
        }

        false
    }

    /// Evict extra outbound peers if we have too many
    /// Standard policy protects up to MAX_OUTBOUND_PEERS_TO_PROTECT peers
    /// based on block announcement recency
    #[allow(dead_code)]
    async fn evict_extra_outbound_peers(&self) {
        use crate::network::peer_manager::MAX_OUTBOUND_PEERS_TO_PROTECT_FROM_DISCONNECT;

        let mut pm = self.peer_manager.lock().await;

        // Get all outbound peers with their last block announcement time
        let mut outbound_peers: Vec<(TransportAddr, u64)> = pm
            .peers()
            .iter()
            .filter(|(_, peer)| peer.is_outbound() && !peer.is_manual())
            .map(|(addr, peer)| {
                let last_announce = peer.last_block_announcement().unwrap_or(0);
                (addr.clone(), last_announce)
            })
            .collect();

        if outbound_peers.len() <= MAX_OUTBOUND_PEERS_TO_PROTECT_FROM_DISCONNECT {
            return; // Not too many peers, no eviction needed
        }

        // Sort by last announcement (oldest first) - peers with no announcements (0) come first
        outbound_peers.sort_by_key(|(_, time)| *time);

        // Disconnect oldest peers (keep best MAX_OUTBOUND_PEERS_TO_PROTECT_FROM_DISCONNECT)
        let peers_to_evict = outbound_peers.len() - MAX_OUTBOUND_PEERS_TO_PROTECT_FROM_DISCONNECT;
        for (addr, last_announce) in outbound_peers.iter().take(peers_to_evict) {
            let now = current_timestamp();
            let time_ago = if *last_announce > 0 {
                now.saturating_sub(*last_announce)
            } else {
                0
            };
            warn!(
                "Evicting extra outbound peer {:?} (last block announcement: {} seconds ago)",
                addr, time_ago
            );
            drop(pm); // Release lock before sending message
            {
                let mut pm = self.peer_manager.lock().await;
                if let Some(peer) = pm.get_peer_mut(addr) {
                    peer.shutdown_connection();
                }
            }
            let _ = self
                .peer_tx
                .send(NetworkMessage::PeerDisconnected(addr.clone()));
            pm = self.peer_manager.lock().await; // Re-acquire lock for next iteration
        }
    }

    /// Create a new network manager with transport preference
    pub fn with_transport_preference(
        listen_addr: SocketAddr,
        max_peers: usize,
        preference: TransportPreference,
    ) -> Self {
        Self::with_config(listen_addr, max_peers, preference, None)
    }

    /// Get transport preference
    pub fn transport_preference(&self) -> TransportPreference {
        self.transport_preference
    }

    /// Get message channel sender (for FIBRE and other integrations)
    pub fn message_sender(&self) -> mpsc::UnboundedSender<NetworkMessage> {
        self.peer_tx.clone()
    }

    /// Whether `start` binds the P2P listeners.
    pub fn listens(&self) -> bool {
        self.listen
    }

    /// Start the network manager
    pub async fn start(&self, listen_addr: SocketAddr) -> Result<()> {
        info!(
            "Starting network manager with transport preference: {:?}",
            self.transport_preference
        );

        #[cfg(feature = "quinn")]
        super::startup::init_quinn_transport(self).await?;

        #[cfg(feature = "iroh")]
        super::startup::init_iroh_transport(self).await?;

        if self.listen {
            super::startup::start_tcp_listener(self, listen_addr).await?;

            #[cfg(feature = "quinn")]
            super::startup::start_quinn_listener(self, listen_addr).await?;

            #[cfg(feature = "iroh")]
            super::startup::start_iroh_listener(self, listen_addr).await?;
        }

        // Start periodic ban cleanup task
        self.start_ban_cleanup_task();
        self.start_ban_list_share_task();

        // Start pending request cleanup task
        self.start_request_cleanup_task();

        // Start DoS protection cleanup task
        self.start_dos_protection_cleanup_task();

        // Start peer reconnection task
        self.start_peer_reconnection_task();

        // Start periodic ping task (sends ping every 2 minutes)
        self.start_ping_task();

        // Start ping timeout checking task (checks every 30 seconds)
        self.start_ping_timeout_check_task();

        // Start chain sync timeout checking task (checks every 60 seconds)
        self.start_chain_sync_timeout_check_task();

        // Start outbound peer eviction task (checks every 5 minutes)
        self.start_outbound_peer_eviction_task();

        // Note: Peer connection initialization (DNS seeds, persistent peers, etc.)
        // should be called separately via initialize_peer_connections() after start()
        // This allows the caller to provide config, network type, and target peer count

        Ok(())
    }

    /// Generate a new request ID for async request-response patterns
    ///
    /// Optimization: Uses AtomicU64 for lock-free operation (no async locks needed)
    pub fn generate_request_id(&self) -> u64 {
        self.request_id_counter.fetch_add(1, Ordering::Relaxed)
    }

    /// Register a pending request and return the response receiver
    /// Returns (request_id, response_receiver)
    pub fn register_request(
        &self,
        peer_addr: SocketAddr,
    ) -> (u64, tokio::sync::oneshot::Receiver<Vec<u8>>) {
        self.register_request_with_priority(peer_addr, 0)
    }

    /// Register a pending request with priority and return the response receiver
    /// Returns (request_id, response_receiver)
    /// Priority: 0 = normal, higher = more important
    pub fn register_request_with_priority(
        &self,
        peer_addr: SocketAddr,
        priority: u8,
    ) -> (u64, tokio::sync::oneshot::Receiver<Vec<u8>>) {
        let request_id = self.generate_request_id();
        let (tx, rx) = tokio::sync::oneshot::channel();

        let timestamp = current_timestamp();

        let pending_req = PendingRequest {
            sender: tx,
            peer_addr,
            timestamp,
            priority,
            retry_count: 0,
        };

        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                self.pending_requests
                    .lock()
                    .await
                    .insert(request_id, pending_req);
            })
        });
        (request_id, rx)
    }

    /// Complete a pending request by sending the response
    pub fn complete_request(&self, request_id: u64, response: Vec<u8>) -> bool {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut pending = self.pending_requests.lock().await;
                if let Some(pending_req) = pending.remove(&request_id) {
                    pending_req.send_response(response);
                    true
                } else {
                    false
                }
            })
        })
    }

    /// Cancel a pending request
    pub fn cancel_request(&self, request_id: u64) -> bool {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut pending = self.pending_requests.lock().await;
                pending.remove(&request_id).is_some()
            })
        })
    }

    /// Get pending requests for a specific peer
    pub fn get_pending_requests_for_peer(&self, peer_addr: SocketAddr) -> Vec<u64> {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let pending = self.pending_requests.lock().await;
                pending
                    .iter()
                    .filter(|(_, req)| req.peer_addr == peer_addr)
                    .map(|(id, _)| *id)
                    .collect()
            })
        })
    }

    /// Register a pending Headers request (supports pipelining - multiple requests per peer)
    /// Returns a receiver that will receive the Headers response
    pub fn register_headers_request(
        &self,
        peer_addr: SocketAddr,
    ) -> tokio::sync::oneshot::Receiver<Vec<blvm_protocol::BlockHeader>> {
        let (tx, rx) = tokio::sync::oneshot::channel();
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                self.pending_headers_requests
                    .lock()
                    .await
                    .entry(peer_addr)
                    .or_insert_with(std::collections::VecDeque::new)
                    .push_back(tx);
            })
        });
        rx
    }

    /// Complete a pending Headers request (FIFO — oldest *live* waiter).
    ///
    /// A 30s timeout drops the oneshot rx but left the tx at the front.
    /// The next Headers then went to a closed waiter and IBD sat (R-48
    /// 640k / R-47 ~840k). Skip closed senders.
    pub fn complete_headers_request(
        &self,
        peer_addr: SocketAddr,
        headers: Vec<blvm_protocol::BlockHeader>,
    ) -> bool {
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut pending = self.pending_headers_requests.lock().await;
                let Some(queue) = pending.get_mut(&peer_addr) else {
                    return false;
                };
                while let Some(sender) = queue.pop_front() {
                    if sender.is_closed() {
                        continue;
                    }
                    let _ = sender.send(headers);
                    if queue.is_empty() {
                        pending.remove(&peer_addr);
                    }
                    return true;
                }
                pending.remove(&peer_addr);
                false
            })
        })
    }

    /// Get the number of pending header requests for a peer
    pub fn pending_headers_count(&self, peer_addr: SocketAddr) -> usize {
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                self.pending_headers_requests
                    .lock()
                    .await
                    .get(&peer_addr)
                    .map(|q| q.len())
                    .unwrap_or(0)
            })
        })
    }

    /// Key for block request matching: IpAddr so inbound (diff port) matches outbound
    fn block_request_key(addr: SocketAddr) -> IpAddr {
        let ip = addr.ip();
        if let std::net::IpAddr::V6(v6) = ip {
            if let Some(v4) = v6.to_ipv4_mapped() {
                return std::net::IpAddr::V4(v4);
            }
        }
        ip
    }

    /// Returns true if `addr` maps to a connected peer (TCP socket or transport mapping).
    pub async fn is_peer_connected(&self, addr: SocketAddr) -> bool {
        let pm = self.peer_manager.lock().await;
        if let Some(transport_addr) = pm.find_transport_addr_by_socket(addr) {
            return pm
                .get_peer(&transport_addr)
                .is_some_and(|p| p.is_connected());
        }
        pm.get_peer(&TransportAddr::Tcp(addr))
            .is_some_and(|p| p.is_connected())
    }

    /// P0-A A7: transport up and Bitcoin handshake complete (VerAck received).
    pub async fn peer_ibd_ready(&self, addr: SocketAddr) -> bool {
        if !self.is_peer_connected(addr).await {
            return false;
        }
        self.peer_states()
            .read()
            .await
            .get(&addr)
            .is_some_and(|s| s.handshake_complete)
    }

    /// Returns the service flags of a connected peer, or 0 if not connected / not yet received.
    pub async fn peer_services(&self, addr: SocketAddr) -> u64 {
        let pm = self.peer_manager.lock().await;
        let ta = pm
            .find_transport_addr_by_socket(addr)
            .unwrap_or(TransportAddr::Tcp(addr));
        pm.get_peer(&ta).map(|p| p.services()).unwrap_or(0)
    }

    /// Register a pending Block request
    /// Returns a receiver that will receive the Block response
    pub fn register_block_request(
        &self,
        peer_addr: SocketAddr,
        block_hash: blvm_protocol::Hash,
    ) -> tokio::sync::oneshot::Receiver<(
        blvm_protocol::Block,
        Vec<Vec<blvm_protocol::segwit::Witness>>,
        Option<Vec<u8>>,
    )> {
        let key = (Self::block_request_key(peer_addr), block_hash);
        let (tx, rx) = tokio::sync::oneshot::channel();
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                // Append (do not overwrite): concurrent same-(peer,block) requests must each
                // keep their own sender so an in-flight download's receiver is never dropped.
                self.pending_block_requests
                    .lock()
                    .await
                    .entry(key)
                    .or_default()
                    .push(tx);
            })
        });
        rx
    }

    fn block_fetch_timeout(config: &crate::config::RequestTimeoutConfig) -> std::time::Duration {
        std::time::Duration::from_secs(config.async_request_timeout_seconds)
    }

    /// Ask a connected peer for one full block via existing GetData (not persisted).
    pub async fn request_block(
        &self,
        block_hash: blvm_protocol::Hash,
    ) -> Option<(
        blvm_protocol::Block,
        Vec<Vec<blvm_protocol::segwit::Witness>>,
    )> {
        use crate::network::inventory::MSG_WITNESS_BLOCK;
        use crate::network::protocol::{GetDataMessage, InventoryVector, ProtocolMessage};

        let fetch_timeout = Self::block_fetch_timeout(self.request_timeout_config());

        let peers = self.peer_addresses();
        for peer_addr in peers {
            if !self.is_peer_connected(peer_addr).await {
                continue;
            }
            let block_rx = self.register_block_request(peer_addr, block_hash);
            let inventory = vec![InventoryVector {
                inv_type: MSG_WITNESS_BLOCK,
                hash: block_hash,
            }];
            let wire_msg =
                match ProtocolParser::serialize_message(&ProtocolMessage::GetData(GetDataMessage {
                    inventory,
                })) {
                    Ok(msg) => msg,
                    Err(_) => {
                        self.cancel_block_request(peer_addr, block_hash);
                        continue;
                    }
                };
            if self.send_to_peer(peer_addr, wire_msg).await.is_err() {
                self.cancel_block_request(peer_addr, block_hash);
                continue;
            }
            match tokio::time::timeout(fetch_timeout, block_rx).await {
                Ok(Ok((block, witnesses, _wire))) => return Some((block, witnesses)),
                _ => self.cancel_block_request(peer_addr, block_hash),
            }
        }
        None
    }

    /// Ask **one** connected peer for a full block via GetData. Not persisted.
    /// Used by the non-owner IBD probe — do not loop other peers.
    pub async fn request_block_from_peer(
        &self,
        peer_addr: SocketAddr,
        block_hash: blvm_protocol::Hash,
        timeout: std::time::Duration,
    ) -> Option<blvm_protocol::Block> {
        use crate::network::inventory::MSG_WITNESS_BLOCK;
        use crate::network::protocol::{GetDataMessage, InventoryVector, ProtocolMessage};

        if !self.is_peer_connected(peer_addr).await {
            return None;
        }
        let block_rx = self.register_block_request(peer_addr, block_hash);
        let inventory = vec![InventoryVector {
            inv_type: MSG_WITNESS_BLOCK,
            hash: block_hash,
        }];
        let wire_msg =
            match ProtocolParser::serialize_message(&ProtocolMessage::GetData(GetDataMessage {
                inventory,
            })) {
                Ok(msg) => msg,
                Err(_) => {
                    self.cancel_block_request(peer_addr, block_hash);
                    return None;
                }
            };
        if self.send_to_peer(peer_addr, wire_msg).await.is_err() {
            self.cancel_block_request(peer_addr, block_hash);
            return None;
        }
        match tokio::time::timeout(timeout, block_rx).await {
            Ok(Ok((block, _witnesses, _wire))) => Some(block),
            _ => {
                self.cancel_block_request(peer_addr, block_hash);
                None
            }
        }
    }

    /// Register multiple block download requests in a single mutex acquisition.
    ///
    /// Equivalent to calling [`register_block_request`] N times but acquires
    /// `pending_block_requests` only once, reducing lock overhead for batched `getdata`
    /// messages. Returns one oneshot receiver per hash in the same order.
    pub fn register_block_requests_batch(
        &self,
        peer_addr: SocketAddr,
        block_hashes: &[blvm_protocol::Hash],
    ) -> Vec<
        tokio::sync::oneshot::Receiver<(
            blvm_protocol::Block,
            Vec<Vec<blvm_protocol::segwit::Witness>>,
            Option<Vec<u8>>,
        )>,
    > {
        let key_prefix = Self::block_request_key(peer_addr);
        let mut rxs = Vec::with_capacity(block_hashes.len());
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut pending = self.pending_block_requests.lock().await;
                for &hash in block_hashes {
                    let key = (key_prefix, hash);
                    let (tx, rx) = tokio::sync::oneshot::channel();
                    // Append (do not overwrite): see `register_block_request` / the
                    // `pending_block_requests` field doc for why concurrent same-key requests
                    // must coexist instead of clobbering each other.
                    pending.entry(key).or_default().push(tx);
                    rxs.push(rx);
                }
            })
        });
        rxs
    }

    /// Drop a pending block request (e.g. GetData failed, timed out, or chunk aborted).
    ///
    /// Prunes only senders whose receiver has already been dropped (`is_closed()`) — i.e. the
    /// caller's own abandoned request — and never a *live* sibling. During IBD the coordinator's
    /// stall recovery routinely has several overlapping requests for the same (peer, block) in
    /// flight; removing the whole key here (as a single-value map would) would drop a concurrently
    /// waiting download's receiver, so the block then arrives to no live receiver and is lost —
    /// exactly the deterministic stall observed (block "routed" repeatedly yet the gap height never
    /// reaches validation). Keeping live siblings guarantees the still-waiting download gets it.
    pub fn cancel_block_request(&self, peer_addr: SocketAddr, block_hash: blvm_protocol::Hash) {
        let key = (Self::block_request_key(peer_addr), block_hash);
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut pending = self.pending_block_requests.lock().await;
                if let Some(senders) = pending.get_mut(&key) {
                    senders.retain(|s| !s.is_closed());
                    if senders.is_empty() {
                        pending.remove(&key);
                        crate::node::parallel_ibd::wire_hash_gate::forget_want(block_hash);
                    }
                }
            })
        });
    }

    /// HASH_FETCH extras only: drop the live GetData so a losing peer unblocks
    /// when another peer already landed the body. Stall recovery must keep
    /// using [`Self::cancel_block_request`] (live siblings stay).
    pub fn cancel_block_request_force(
        &self,
        peer_addr: SocketAddr,
        block_hash: blvm_protocol::Hash,
    ) {
        let key = (Self::block_request_key(peer_addr), block_hash);
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut pending = self.pending_block_requests.lock().await;
                pending.remove(&key);
                crate::node::parallel_ibd::wire_hash_gate::forget_want(block_hash);
            })
        });
    }

    /// Complete a pending Block request (no original wire payload).
    pub fn complete_block_request(
        &self,
        peer_addr: SocketAddr,
        block_hash: blvm_protocol::Hash,
        block: blvm_protocol::Block,
        witnesses: Vec<Vec<blvm_protocol::segwit::Witness>>,
    ) -> bool {
        self.complete_block_request_with_wire(peer_addr, block_hash, block, witnesses, None)
    }

    /// Complete a pending Block request, optionally attaching the original P2P payload (W5/N1).
    pub fn complete_block_request_with_wire(
        &self,
        peer_addr: SocketAddr,
        block_hash: blvm_protocol::Hash,
        block: blvm_protocol::Block,
        witnesses: Vec<Vec<blvm_protocol::segwit::Witness>>,
        wire_payload: Option<Vec<u8>>,
    ) -> bool {
        let key = (Self::block_request_key(peer_addr), block_hash);
        let wire_bytes = wire_payload.as_ref().map(|p| p.len() as u64).unwrap_or(0);
        let handled = tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut pending = self.pending_block_requests.lock().await;
                let Some(senders) = pending.remove(&key) else {
                    crate::node::parallel_ibd::body_dup::note_loser_branch_missing();
                    if crate::node::parallel_ibd::late_body::try_admit(
                        peer_addr,
                        block_hash,
                        block,
                        witnesses,
                        wire_payload,
                    ) {
                        crate::node::parallel_ibd::body_dup::note_discard(
                            crate::node::parallel_ibd::body_dup::DiscardReason::LateAdmit,
                            None,
                            wire_bytes,
                        );
                        return true;
                    }
                    crate::node::parallel_ibd::body_dup::note_discard(
                        crate::node::parallel_ibd::body_dup::DiscardReason::RequeueLoser,
                        None,
                        wire_bytes,
                    );
                    return false;
                };
                // Deliver a copy to every live receiver. A closed sibling is skipped.
                // Removing the key is safe only after each live receiver has the block.
                let live: Vec<_> = senders
                    .into_iter()
                    .filter(|sender| !sender.is_closed())
                    .collect();
                if live.is_empty() {
                    crate::node::parallel_ibd::body_dup::note_loser_branch_dropped();
                    if crate::node::parallel_ibd::late_body::try_admit(
                        peer_addr,
                        block_hash,
                        block,
                        witnesses,
                        wire_payload,
                    ) {
                        crate::node::parallel_ibd::body_dup::note_discard(
                            crate::node::parallel_ibd::body_dup::DiscardReason::LateAdmit,
                            None,
                            wire_bytes,
                        );
                        return true;
                    }
                    crate::node::parallel_ibd::body_dup::note_discard(
                        crate::node::parallel_ibd::body_dup::DiscardReason::RequeueLoser,
                        None,
                        wire_bytes,
                    );
                    return false;
                }
                let mut payload = Some((block, witnesses, wire_payload));
                let mut delivered = false;
                let last = live.len().saturating_sub(1);
                for (i, sender) in live.into_iter().enumerate() {
                    let message = if i == last {
                        payload.take().expect("payload")
                    } else {
                        payload.as_ref().expect("payload").clone()
                    };
                    if sender.send(message).is_ok() {
                        delivered = true;
                    }
                }
                delivered
            })
        });
        // After admit lookup: `want` is the header index try_admit reads.
        crate::node::parallel_ibd::wire_hash_gate::forget_want(block_hash);
        handled
    }

    /// Clean up expired requests (older than max_age_seconds)
    pub fn cleanup_expired_requests(&self, max_age_seconds: u64) -> usize {
        let now = current_timestamp();

        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut pending = self.pending_requests.lock().await;
                let expired: Vec<u64> = pending
                    .iter()
                    .filter(|(_, req)| now.saturating_sub(req.timestamp) > max_age_seconds)
                    .map(|(id, _)| *id)
                    .collect();

                for id in &expired {
                    pending.remove(id);
                }

                expired.len()
            })
        })
    }

    /// Get the number of connected peers
    pub fn peer_count(&self) -> usize {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let pm = self.peer_manager.lock().await;
                pm.peer_count()
            })
        })
    }

    /// Get all peer addresses (as SocketAddr for backward compatibility)
    pub fn peer_addresses(&self) -> Vec<SocketAddr> {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let pm = self.peer_manager.lock().await;
                pm.peer_socket_addresses()
            })
        })
    }

    /// Connected IBD addrs only (no DNS / connect expansion).
    ///
    /// Coordinator ready-refresh must use this under `tokio::time::timeout`.
    /// [`Self::peer_addresses_for_ibd`] is `block_in_place` + `block_on` and
    /// cannot be cancelled — R-240 CRAWL silent after TIP_FAILOVER @107124,
    /// zero `[IBD_READY_REFRESH_TIMEOUT]`.
    pub async fn peer_addresses_for_ibd_connected(&self) -> Vec<SocketAddr> {
        let pm = self.peer_manager.lock().await;
        pm.peer_socket_addresses_for_ibd()
    }

    /// Get peer addresses suitable for IBD full-history block download.
    ///
    /// Excludes peers advertising `NODE_NETWORK_LIMITED` (BIP159 pruned nodes), which
    /// cannot serve historical blocks from years ago. Falls back to all peers if none
    /// qualify, so IBD never stalls due to missing full-history peers.
    pub fn peer_addresses_for_ibd(&self) -> Vec<SocketAddr> {
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                use blvm_protocol::service_flags::standard::{NODE_NETWORK, NODE_NETWORK_LIMITED};

                // Fast path: at least one full-history peer already connected.
                // A peer is full-history if it advertises NODE_NETWORK (0x1) — the flag that
                // means "I have the complete blockchain".  Modern Bitcoin Core 25+ sets both
                // NODE_NETWORK and NODE_NETWORK_LIMITED; filtering on LIMITED alone incorrectly
                // excludes all such peers. Only peers with NODE_NETWORK_LIMITED but no
                // NODE_NETWORK are genuinely pruned (serve recent blocks only).
                {
                    let pm = self.peer_manager.lock().await;
                    let addrs = pm.peer_socket_addresses_for_ibd();
                    // Empty set: `Iterator::all` is vacuously true — do NOT treat that as
                    // "all peers are pruned" and burn ~10s on DNS/connect expansion.
                    // Local-disk / sequential IBD often has zero TCP peers; the coordinator
                    // ready-refresh used to call this every ~1s and stall tip crawl at ~10 BPS.
                    if pm.peers().is_empty() {
                        return addrs;
                    }
                    // "all limited" = no peer has NODE_NETWORK set.
                    let all_limited = pm.peers().iter().all(|(_, p)| {
                        let s = p.services();
                        s != 0 && (s & NODE_NETWORK) == 0
                    });
                    if !all_limited {
                        return addrs;
                    }
                }

                // All connected peers are NODE_NETWORK_LIMITED (pruned).  Try to expand
                // the peer set with fresh addresses from the database, then wait briefly
                // for handshakes to complete so we can pick full-history peers.
                warn!(
                    "IBD: all connected peers are NODE_NETWORK_LIMITED; \
                     connecting to more peers from address database"
                );
                let current: Vec<std::net::SocketAddr> = {
                    let pm = self.peer_manager.lock().await;
                    pm.peer_socket_addresses()
                };
                let ban_list = self.ban_list().read().await.clone();

                // Check how many non-limited candidates exist.  If all candidates in the DB are
                // already marked limited, use the archive-node DNS seeds (x1. prefix = NODE_NETWORK
                // service-bit filter) to populate specifically full-history peers.  Regular seeds
                // return mostly pruned nodes; archive seeds return only archival nodes that keep the
                // complete blockchain and can serve historical IBD blocks.
                let fresh_non_limited = {
                    let db = self.address_database().read().await;
                    db.count_fresh_non_limited()
                };
                if fresh_non_limited < 4 {
                    warn!(
                        "IBD: only {} non-limited addresses in DB — seeding from archive (x1-filtered) DNS seeds",
                        fresh_non_limited
                    );
                    // Prefer archive seeds; fall back to standard seeds if archive seeds fail.
                    if let Ok(n) = self.discover_archive_peers_from_dns().await {
                        if n == 0 {
                            let (net_name, port) =
                                crate::network::protocol::ProtocolParser::dns_seed_network();
                            let _ = self
                                .discover_peers_from_dns(net_name, port, &Default::default())
                                .await;
                        }
                    } else {
                        let (net_name, port) =
                            crate::network::protocol::ProtocolParser::dns_seed_network();
                        let _ = self
                            .discover_peers_from_dns(net_name, port, &Default::default())
                            .await;
                    }
                }

                let candidates: Vec<std::net::SocketAddr> = {
                    let db = self.address_database().read().await;
                    // Use IBD-biased ordering (full-history first, then unknown, then limited).
                    let fresh = db.select_outbound(128, outbound_pick_tried());
                    let filtered = db.filter_addresses(fresh, &ban_list, &current);
                    filtered
                        .iter()
                        .map(|a| db.network_addr_to_socket(a))
                        .collect()
                };
                // Connect in parallel so we don't block shutdown on sequential 10-s timeouts.
                use futures::StreamExt as _;
                let mut stream = futures::stream::FuturesUnordered::new();
                for sock in candidates.into_iter().take(16) {
                    stream.push(async move { (sock, self.connect_to_peer(sock).await) });
                }
                let mut connected = 0usize;
                while let Some((sock, result)) = stream.next().await {
                    match result {
                        Ok(()) => {
                            connected += 1;
                            if connected >= 8 {
                                break;
                            }
                        }
                        Err(e) => {
                            debug!("IBD peer expansion: failed to connect to {}: {}", sock, e);
                        }
                    }
                }
                if connected > 0 {
                    // Give new connections time to complete version/verack handshake.
                    tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
                }
                // Re-check — if we found full-history peers, use them; else fall back.
                let pm = self.peer_manager.lock().await;
                pm.peer_socket_addresses_for_ibd()
            })
        })
    }

    /// Evict a peer that has proven it cannot serve historical IBD blocks (e.g. consistent
    /// block-timeout failures indicating `NODE_NETWORK_LIMITED`/pruned).
    ///
    /// This:
    /// 1. Writes `NODE_NETWORK_LIMITED` into the address-database entry so the peer is skipped
    ///    on future `peer_addresses_for_ibd` and `get_fresh_addresses` calls.
    /// 2. Disconnects the peer, freeing its connection slot.
    /// 3. Attempts to connect a replacement from the address database, preferring addresses that
    ///    do not have `NODE_NETWORK_LIMITED` set.
    ///
    /// Clear the IBD evicted-IP set so a fresh IBD session can try all peers again.
    ///
    /// Must be called at the start of each IBD attempt (e.g. in `try_catch_up_ibd`).
    /// Without this, a failed IBD run permanently poisons the eviction set, preventing
    /// the reconnection task from ever re-establishing connections to those peers, even
    /// though they may have recovered or the node may be connecting from a different
    /// perspective.  The address database already durably records `NODE_NETWORK_LIMITED`
    /// for confirmed pruned peers, so clearing the in-memory set is safe.
    pub fn clear_ibd_evicted_ips(&self) {
        let mut evicted = self.ibd_evicted_ips.write().unwrap();
        let prev_count = evicted.len();
        evicted.clear();
        if prev_count > 0 {
            info!(
                "IBD: cleared {} evicted-IP entries at session end",
                prev_count
            );
        }
    }

    /// Called from the IBD download worker when `MAX_CONSECUTIVE_FAILURES` are all block-timeouts.
    pub fn evict_ibd_peer(&self, peer_addr: SocketAddr) {
        use blvm_protocol::service_flags::standard::NODE_NETWORK_LIMITED;
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                if crate::network::ibd_peers_pinned() {
                    warn!(
                        "IBD: pinned set — disconnect {} and reconnect from BLVM_IBD_PIN_PEERS (no DNS)",
                        peer_addr
                    );
                    let _ = self.peer_tx.send(NetworkMessage::PeerDisconnected(
                        TransportAddr::Tcp(peer_addr),
                    ));
                    tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
                    for addr in crate::network::ibd_pin_peers() {
                        let current: Vec<SocketAddr> = {
                            let pm = self.peer_manager.lock().await;
                            pm.peer_socket_addresses()
                        };
                        if current.contains(&addr) {
                            continue;
                        }
                        if self.connect_to_peer(addr).await.is_ok() {
                            info!("IBD: pinned replacement connected: {}", addr);
                            break;
                        }
                    }
                    return;
                }
                // Record this IP for the rest of the IBD session. Service flags stay as stored.
                // spawn_outbound_reconnect_attempt skips the IP until the session guard clears the set.
                {
                    let mut evicted = self.ibd_evicted_ips.write().unwrap();
                    evicted.insert(peer_addr.ip());
                }
                // 3. Disconnect the peer (fires PeerDisconnected → cancel pending block reqs).
                let _ = self.peer_tx.send(NetworkMessage::PeerDisconnected(
                    TransportAddr::Tcp(peer_addr),
                ));
                warn!(
                    "IBD: evicted peer {} for this download session (slot freed for replacement)",
                    peer_addr
                );
                // 3. Brief pause so the disconnect propagates before we count current peers.
                tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;

                // 4. Connect a replacement. Prefer entries that don't have NODE_NETWORK_LIMITED.
                let current: Vec<SocketAddr> = {
                    let pm = self.peer_manager.lock().await;
                    pm.peer_socket_addresses()
                };
                let ban_list = self.ban_list().read().await.clone();
                let candidates: Vec<SocketAddr> = {
                    let db = self.address_database().read().await;
                    // IBD-ordered: full-history first, unknown second, limited last.
                    let fresh = db.select_outbound(128, outbound_pick_tried());
                    let filtered = db.filter_addresses(fresh, &ban_list, &current);
                    // Split into non-limited (preferred) and limited (fallback).
                    let mut preferred = Vec::new();
                    let mut fallback = Vec::new();
                    for a in &filtered {
                        let sock = db.network_addr_to_socket(a);
                        // services==0 means we don't know yet — include optimistically.
                        // services with LIMITED bit set → skip if we have alternatives.
                        let entry_services = db.services_for_socket(sock);
                        if entry_services != 0 && (entry_services & NODE_NETWORK_LIMITED) != 0 {
                            fallback.push(sock);
                        } else {
                            preferred.push(sock);
                        }
                    }
                    let mut out = preferred;
                    out.extend(fallback);
                    out
                };
                // If no candidates, trigger archive DNS re-seed (x1. prefix = NODE_NETWORK filter)
                // and recompute from the freshly seeded address database.
                let candidates = if candidates.is_empty() {
                    warn!(
                        "IBD: no replacement candidates after evicting {} — seeding from archive DNS seeds",
                        peer_addr
                    );
                    // Use archive seeds first (full-history nodes); fall back to standard seeds.
                    if self.discover_archive_peers_from_dns().await.unwrap_or(0) == 0 {
                        let (net_name, port) =
                            crate::network::protocol::ProtocolParser::dns_seed_network();
                        let _ = self
                            .discover_peers_from_dns(net_name, port, &Default::default())
                            .await;
                    }
                    // Recompute candidates from the freshly seeded address database.
                    let current_after: Vec<SocketAddr> = {
                        let pm = self.peer_manager.lock().await;
                        pm.peer_socket_addresses()
                    };
                    let ban_list2 = self.ban_list().read().await.clone();
                    let db = self.address_database().read().await;
                    let fresh2 = db.select_outbound(128, outbound_pick_tried());
                    let filtered2 = db.filter_addresses(fresh2, &ban_list2, &current_after);
                    let evicted2 = self.ibd_evicted_ips.read().unwrap();
                    let mut preferred2 = Vec::new();
                    let mut fallback2 = Vec::new();
                    for a in &filtered2 {
                        let sock = db.network_addr_to_socket(a);
                        if evicted2.contains(&sock.ip()) {
                            continue;
                        }
                        let svc = db.services_for_socket(sock);
                        if svc != 0 && (svc & NODE_NETWORK_LIMITED) != 0 {
                            fallback2.push(sock);
                        } else {
                            preferred2.push(sock);
                        }
                    }
                    let mut out2 = preferred2;
                    out2.extend(fallback2);
                    out2
                } else {
                    candidates
                };
                for sock in candidates.into_iter().take(3) {
                    match self.connect_to_peer(sock).await {
                        Ok(()) => {
                            info!("IBD: replacement peer connected: {}", sock);
                            break;
                        }
                        Err(e) => {
                            debug!("IBD: replacement candidate {} failed: {}", sock, e);
                        }
                    }
                }
            })
        });
    }

    /// Get all peer addresses (as TransportAddr)
    pub fn peer_transport_addresses(&self) -> Vec<TransportAddr> {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let pm = self.peer_manager.lock().await;
                pm.peer_addresses()
            })
        })
    }

    /// Get the highest start_height (best block height) from all connected peers
    /// Returns None if no peers are connected or no peers have reported a start_height
    pub fn get_highest_peer_start_height(&self) -> Option<u64> {
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current()
                .block_on(async { self.get_highest_peer_start_height_async().await })
        })
    }

    /// Async variant for use from the node run loop (avoids block_in_place inside async tasks).
    pub async fn get_highest_peer_start_height_async(&self) -> Option<u64> {
        let pm = self.peer_manager.lock().await;
        let mut max_height: Option<u64> = None;
        for (_addr, peer) in pm.peers().iter() {
            let from_version = peer.start_height().max(0) as u64;
            let from_updates = peer.best_block_height().unwrap_or(0);
            let peer_height = from_version.max(from_updates);
            if peer_height > 0 {
                max_height = Some(max_height.map_or(peer_height, |m| m.max(peer_height)));
            }
        }
        max_height
    }

    /// Broadcast a message to all peers
    pub async fn broadcast(&self, message: Vec<u8>) -> Result<()> {
        // Get peer addresses first, then drop lock before async operations
        let peer_addrs = {
            let pm = self.peer_manager.lock().await;
            pm.peer_addresses()
        };

        // Send to each peer using transport address (avoids needing to clone Peer)
        for addr in peer_addrs {
            let addr_clone = addr.clone();
            if let Err(e) = self.send_to_peer_by_transport(addr, message.clone()).await {
                warn!(
                    "Failed to broadcast message to peer {:?}: {}",
                    addr_clone, e
                );
            }
        }
        Ok(())
    }

    /// Broadcast to reliable peers first, then others
    /// Uses peer quality to prioritize reliable peers for critical messages
    pub async fn broadcast_with_quality_priority(&self, message: Vec<u8>) -> Result<()> {
        // Get reliable peers first, then drop lock before async operations
        let reliable_peers = {
            let pm = self.peer_manager.lock().await;
            pm.select_reliable_peers()
        };

        // Send to reliable peers first
        for addr in &reliable_peers {
            if let Err(e) = self
                .send_to_peer_by_transport(addr.clone(), message.clone())
                .await
            {
                warn!("Failed to send to reliable peer {:?}: {}", addr, e);
            }
        }

        // Then send to remaining peers
        let pm = self.peer_manager.lock().await;
        let all_peers = pm.peer_addresses();
        let remaining_peers: Vec<_> = all_peers
            .iter()
            .filter(|addr| !reliable_peers.contains(addr))
            .collect();
        drop(pm);

        for addr in remaining_peers {
            if let Err(e) = self
                .send_to_peer_by_transport(addr.clone(), message.clone())
                .await
            {
                warn!("Failed to send to peer {:?}: {}", addr, e);
            }
        }

        Ok(())
    }

    /// Send to best peer (highest quality score)
    /// Returns the peer address used, or error if no peers available
    pub async fn send_to_best_peer(&self, message: Vec<u8>) -> Result<TransportAddr> {
        let pm = self.peer_manager.lock().await;
        let best_peers = pm.select_best_peers(1);
        drop(pm);

        if let Some(addr) = best_peers.first() {
            self.send_to_peer_by_transport(addr.clone(), message)
                .await?;
            Ok(addr.clone())
        } else {
            Err(anyhow::anyhow!("No peers available"))
        }
    }

    /// Send to reliable peer only (for critical operations)
    /// Returns error if no reliable peers available
    pub async fn send_to_reliable_peer(&self, message: Vec<u8>) -> Result<TransportAddr> {
        let pm = self.peer_manager.lock().await;
        let reliable_peers = pm.select_reliable_peers();
        drop(pm);

        if reliable_peers.is_empty() {
            return Err(anyhow::anyhow!("No reliable peers available"));
        }

        // Select best reliable peer
        let pm = self.peer_manager.lock().await;
        let best_reliable = reliable_peers.iter().max_by(|a, b| {
            let score_a = pm.get_peer(a).map(|p| p.quality_score()).unwrap_or(0.0);
            let score_b = pm.get_peer(b).map(|p| p.quality_score()).unwrap_or(0.0);
            score_a
                .partial_cmp(&score_b)
                .unwrap_or(std::cmp::Ordering::Equal)
        });
        drop(pm);

        if let Some(addr) = best_reliable {
            self.send_to_peer_by_transport(addr.clone(), message)
                .await?;
            Ok(addr.clone())
        } else {
            Err(anyhow::anyhow!("No reliable peers available"))
        }
    }

    /// Send a message to a specific peer (by SocketAddr - for TCP/Quinn)
    /// For Iroh peers, uses the socket_to_transport mapping
    pub async fn send_to_peer(&self, addr: SocketAddr, message: Vec<u8>) -> Result<()> {
        // Try to find transport address (TCP, Quinn, or Iroh mapping)
        let transport_addr = {
            let pm = self.peer_manager.lock().await;
            pm.find_transport_addr_by_socket(addr).or_else(|| {
                // Check socket_to_transport mapping (synchronous access needed)
                // Note: This is a fallback, so we use try_lock to avoid blocking
                // If we can't get the lock immediately, return None
                self.socket_to_transport
                    .try_lock()
                    .ok()
                    .and_then(|map| map.get(&addr).cloned())
            })
        };

        if let Some(transport_addr) = transport_addr {
            self.send_to_peer_by_transport(transport_addr, message)
                .await
        } else {
            // Fallback: try TCP (for backward compatibility)
            self.send_to_peer_by_transport(TransportAddr::Tcp(addr), message)
                .await
        }
    }

    /// Send a message to a specific peer (by TransportAddr - supports all transports)
    /// Returns error if peer doesn't exist or is disconnected
    pub async fn send_to_peer_by_transport(
        &self,
        addr: TransportAddr,
        message: Vec<u8>,
    ) -> Result<()> {
        let message_len = message.len();

        // Check if peer exists and get sender channel before dropping lock
        let sender = {
            let pm = self.peer_manager.lock().await;
            if let Some(peer) = pm.get_peer(&addr) {
                if !peer.is_connected() {
                    return Err(anyhow::anyhow!("Peer {addr:?} is disconnected"));
                }
                peer.send_tx.clone()
            } else {
                // Peer doesn't exist - return error so caller knows to try another peer
                return Err(anyhow::anyhow!("Peer {addr:?} not found"));
            }
        };

        // Track bytes sent only if peer exists (async-safe)
        self.track_bytes_sent(message_len as u64).await;

        // Extract message type for event publishing (before message is moved)
        let msg_type = if message_len >= 16 {
            String::from_utf8_lossy(&message[4..16])
                .trim_end_matches('\0')
                .to_string()
        } else {
            "raw".to_string()
        };

        // Bandwidth Protection: Record bandwidth for various message types
        // Check message command (Bitcoin wire format: 4-byte magic + 12-byte command + 4-byte length + payload)
        if message_len >= 16 {
            let command = msg_type.as_str();

            // Get SocketAddr for bandwidth protection (needed for tracking)
            let peer_socket_addr_opt: Option<SocketAddr> = match &addr {
                TransportAddr::Tcp(sock) => Some(*sock),
                #[cfg(feature = "quinn")]
                TransportAddr::Quinn(sock) => Some(*sock),
                #[cfg(feature = "iroh")]
                TransportAddr::Iroh(_) => {
                    // For Iroh, try to get from socket_to_transport mapping
                    // If not found, skip bandwidth tracking (Iroh uses different addressing)
                    self.socket_to_transport
                        .lock()
                        .await
                        .iter()
                        .find_map(|(sock, ta)| if *ta == addr { Some(*sock) } else { None })
                }
            };

            if let Some(peer_socket_addr) = peer_socket_addr_opt {
                // Record bandwidth for IBD-serving messages
                if command == cmd::HEADERS || command == cmd::BLOCK {
                    // Record bandwidth for IBD protection
                    self.ibd_protection
                        .record_bandwidth(peer_socket_addr, message_len as u64)
                        .await;
                    debug!(
                        "IBD protection: Recorded {} bytes for {} message to {}",
                        message_len, command, peer_socket_addr
                    );
                }
                // Record bandwidth for transaction relay
                else if command == "tx" {
                    use super::bandwidth_protection::ServiceType;
                    // Record transaction relay bandwidth (per-IP/subnet limits)
                    self.bandwidth_protection
                        .record_service_bandwidth(
                            ServiceType::TransactionRelay,
                            peer_socket_addr,
                            message_len as u64,
                        )
                        .await;
                    debug!(
                        "Transaction relay: Recorded {} bytes to {}",
                        message_len, peer_socket_addr
                    );
                }
            }
        }

        // Send message without holding the lock (unbounded channel, so send won't block)
        sender
            .send(message)
            .map_err(|e| anyhow::anyhow!("Failed to send message to peer: {e}"))?;

        // Publish MessageSent event for module subscribers
        if let Some(ref ep) = *self.event_publisher.lock().await {
            let addr_str = addr.to_string();
            let ep_clone = Arc::clone(ep);
            let msg_type_clone = msg_type.clone();
            tokio::spawn(async move {
                ep_clone
                    .publish_message_sent(&addr_str, &msg_type_clone, message_len)
                    .await;
            });
        }

        // Record send after message is sent
        let mut pm = self.peer_manager.lock().await;
        if let Some(peer) = pm.get_peer_mut(&addr) {
            peer.record_send(message_len);
        }
        Ok(())
    }

    pub async fn ping_all_peers(&self) -> Result<()> {
        use crate::network::protocol::{PingMessage, ProtocolMessage, ProtocolParser};
        // Generate nonce for ping
        let nonce = current_timestamp_nanos();

        let ping_msg = ProtocolMessage::Ping(PingMessage { nonce });
        let wire_msg = ProtocolParser::serialize_message(&ping_msg)?;

        let peer_addrs = {
            let mut pm = self.peer_manager.lock().await;
            // Record ping in peer state before sending
            for (addr, peer) in pm.peers_mut().iter_mut() {
                peer.record_ping_sent(nonce);
            }
            pm.peer_addresses()
        };

        for addr in peer_addrs {
            // Convert TransportAddr to SocketAddr for send_to_peer, or use send_to_peer_by_transport
            let addr_clone = addr.clone();
            match addr_clone {
                super::transport::TransportAddr::Tcp(sock) => {
                    if let Err(e) = self.send_to_peer(sock, wire_msg.clone()).await {
                        warn!("Failed to ping peer {}: {}", sock, e);
                    }
                }
                #[cfg(feature = "quinn")]
                crate::network::transport::TransportAddr::Quinn(sock) => {
                    if let Err(e) = self
                        .send_to_peer_by_transport(addr_clone.clone(), wire_msg.clone())
                        .await
                    {
                        warn!("Failed to ping peer {}: {}", sock, e);
                    }
                }
                #[cfg(feature = "iroh")]
                crate::network::transport::TransportAddr::Iroh(_) => {
                    if let Err(e) = self
                        .send_to_peer_by_transport(addr_clone.clone(), wire_msg.clone())
                        .await
                    {
                        warn!("Failed to ping peer {}: {}", addr_clone, e);
                    }
                }
            }
        }

        Ok(())
    }

    /// Try to receive a block message (non-blocking)
    /// Returns Some(block_data) if a block was received, None otherwise
    /// Try to receive a block from the pending blocks queue
    /// This does NOT consume other message types from the main channel
    pub fn try_recv_block(&self) -> Option<Vec<u8>> {
        // Use the dedicated pending_blocks queue instead of draining the main channel
        // This preserves PeerConnected, RawMessageReceived, etc. for proper processing
        let mut blocks = self.pending_blocks.lock().ok()?;
        blocks.pop_front()
    }

    /// Queue a block for processing (called when BlockReceived is received in process_messages)
    pub fn queue_block(&self, data: Vec<u8>) {
        if let Ok(mut blocks) = self.pending_blocks.lock() {
            blocks.push_back(data);
        }
    }

    pub(crate) fn pending_block_queue_len(&self) -> usize {
        self.pending_blocks
            .lock()
            .map(|blocks| blocks.len())
            .unwrap_or(0)
    }

    /// Process all pending network messages without blocking
    /// Returns the number of messages processed
    pub async fn process_pending_messages(&self) -> Result<usize> {
        let mut processed = 0;

        loop {
            // Try to receive a message without blocking
            let message = {
                let mut rx = self.peer_rx.lock().await;
                match rx.try_recv() {
                    Ok(msg) => {
                        debug!(
                            "process_pending_messages: received message {:?}",
                            match &msg {
                                NetworkMessage::PeerConnected(addr) =>
                                    format!("PeerConnected({addr:?})"),
                                NetworkMessage::PeerDisconnected(addr) =>
                                    format!("PeerDisconnected({addr:?})"),
                                NetworkMessage::RawMessageReceived(data, addr) => format!(
                                    "RawMessageReceived({} bytes from {})",
                                    data.len(),
                                    addr
                                ),
                                _ => "other".to_string(),
                            }
                        );
                        Some(msg)
                    }
                    Err(tokio::sync::mpsc::error::TryRecvError::Empty) => None,
                    Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
                        warn!("process_pending_messages: channel disconnected");
                        return Ok(processed);
                    }
                }
            };

            let message = match message {
                Some(msg) => msg,
                None => break, // No more pending messages
            };

            if let NetworkMessage::RawMessageReceived(data, _) = &message {
                crate::network::parse_offload::release_peer_rx(data.len());
            }
            // Process the message (reuse the same processing logic)
            self.handle_network_message(message).await;
            processed += 1;
        }

        Ok(processed)
    }

    /// Handle a single network message (extracted for reuse)
    async fn handle_network_message(&self, message: NetworkMessage) {
        match message {
            NetworkMessage::PeerConnected(addr) => {
                info!("Peer connected: {:?}", addr);

                // Send Version message to initiate handshake
                let socket_addr = match &addr {
                    TransportAddr::Tcp(sock) => Some(*sock),
                    #[cfg(feature = "quinn")]
                    TransportAddr::Quinn(sock) => Some(*sock),
                    #[cfg(feature = "iroh")]
                    TransportAddr::Iroh(_) => None,
                };

                if let Some(peer_socket) = socket_addr {
                    let start_height = if let Some(ref storage) = self.storage {
                        storage.chain().get_height().ok().flatten().unwrap_or(0) as i32
                    } else {
                        0
                    };

                    let peer_ip = match peer_socket.ip() {
                        std::net::IpAddr::V4(ip) => {
                            let mut addr = [0u8; 16];
                            addr[10] = 0xff;
                            addr[11] = 0xff;
                            addr[12..16].copy_from_slice(&ip.octets());
                            addr
                        }
                        std::net::IpAddr::V6(ip) => ip.octets(),
                    };
                    let addr_recv = crate::network::protocol::NetworkAddress {
                        time: 0,
                        services: 0,
                        ip: peer_ip,
                        port: peer_socket.port(),
                    };
                    let addr_from = crate::network::protocol::NetworkAddress {
                        time: 0,
                        services: 0,
                        ip: [0u8; 16],
                        port: 0,
                    };

                    let nonce: u64 = rand::random();
                    // Track this nonce so we can detect self-connections.
                    self.local_version_nonces
                        .lock()
                        .unwrap_or_else(|e| e.into_inner())
                        .insert(nonce);

                    let version_msg = self.create_version_message(
                        70015,
                        0,
                        crate::utils::current_timestamp() as i64,
                        addr_recv,
                        addr_from,
                        nonce,
                        format!("/Bitcoin Commons:{}/", env!("CARGO_PKG_VERSION")),
                        start_height,
                        true,
                    );

                    match ProtocolParser::serialize_message(&ProtocolMessage::Version(version_msg))
                    {
                        Ok(wire_msg) => {
                            if let Err(e) =
                                self.send_to_peer_by_transport(addr.clone(), wire_msg).await
                            {
                                warn!("Failed to send Version message to {:?}: {}", addr, e);
                            } else {
                                info!("Sent Version message to {:?}", addr);
                            }
                        }
                        Err(e) => {
                            warn!("Failed to serialize Version message for {:?}: {}", addr, e);
                        }
                    }
                }

                // Publish peer connected event
                let event_publisher_guard = self.event_publisher.lock().await;
                if let Some(ref event_publisher) = *event_publisher_guard {
                    let addr_str = addr.to_string();
                    let transport_type = match &addr {
                        TransportAddr::Tcp(_) => "tcp",
                        #[cfg(feature = "quinn")]
                        TransportAddr::Quinn(_) => "quinn",
                        #[cfg(feature = "iroh")]
                        TransportAddr::Iroh(_) => "iroh",
                    };

                    let pm = self.peer_manager.lock().await;
                    let (services, version) = if let Some(peer) = pm.get_peer(&addr) {
                        (peer.services(), peer.version())
                    } else {
                        (0, 0)
                    };
                    drop(pm);

                    let event_pub_clone = Arc::clone(event_publisher);
                    tokio::spawn(async move {
                        event_pub_clone
                            .publish_peer_connected(&addr_str, transport_type, services, version)
                            .await;
                    });
                }
            }
            NetworkMessage::PeerDisconnected(addr) => {
                info!("Peer disconnected (during pending processing): {:?}", addr);
                let handshake_done = {
                    let pm = self.peer_manager.lock().await;
                    pm.get_peer(&addr).is_some_and(|peer| peer.version() > 0)
                };
                if let TransportAddr::Tcp(sa) = addr {
                    if !handshake_done {
                        crate::network::peer_connections::note_dial_failure(sa);
                    }
                    crate::node::parallel_ibd::ibd_peer_gone(&sa.to_string());
                }
                self.clear_companion_udp_peer_on_disconnect(&addr).await;
                let mut pm = self.peer_manager.lock().await;
                pm.remove_peer(&addr);
                drop(pm);

                // Cancel all in-flight block requests for this peer so their
                // oneshot receivers resolve immediately with RecvError rather
                // than waiting for the 30 s download timeout.  Each 30 s wait
                // blocks the ordered pipeline; at ~13 disconnects/minute this
                // was wasting ~6× as much wall-clock time as real block work.
                if let TransportAddr::Tcp(sa) = addr {
                    self.cancel_pending_block_requests_for_disconnected_peer(sa)
                        .await;
                }

                let event_publisher_guard = self.event_publisher.lock().await;
                if let Some(ref event_publisher) = *event_publisher_guard {
                    let addr_str = addr.to_string();
                    let event_pub_clone = Arc::clone(event_publisher);
                    tokio::spawn(async move {
                        event_pub_clone
                            .publish_peer_disconnected(&addr_str, "disconnected")
                            .await;
                    });
                }
            }
            NetworkMessage::RawMessageReceived(data, peer) => {
                // Process raw message - this will handle Version messages and send VerAck
                if let Err(e) = self.handle_incoming_identity(peer.clone(), data).await {
                    debug!("Error processing raw message from {}: {}", peer, e);
                }
            }
            _ => {
                // Other message types - handle minimally during handshake
                debug!("Received other message type during handshake phase");
            }
        }
    }

    /// Process incoming network messages
    pub async fn process_messages(self: &Arc<Self>) -> Result<()> {
        // Track message queue size manually (unbounded channel doesn't have len())
        let mut message_count = 0u64;
        let mut last_metrics_update = std::time::SystemTime::now();

        // Restructured to work with Arc<Mutex>: lock, receive one message, unlock, process
        loop {
            // Lock, receive one message, unlock
            let message = {
                let mut rx = self.peer_rx.lock().await;
                rx.recv().await
            };

            let message = match message {
                Some(msg) => msg,
                None => {
                    // Channel closed
                    warn!("Network message channel closed");
                    break;
                }
            };

            // Process message (no lock held)
            message_count += 1;

            // Update metrics periodically (every 100 messages or 10 seconds)
            let now = std::time::SystemTime::now();
            let should_update = message_count % 100 == 0
                || now
                    .duration_since(last_metrics_update)
                    .unwrap_or_else(|_| std::time::Duration::from_secs(0))
                    .as_secs()
                    >= 10;

            if should_update {
                let pm = self.peer_manager.lock().await;
                let active_connections = pm.peer_count();
                let bytes_received = self.bytes_received.load(Ordering::Relaxed);
                let bytes_sent = self.bytes_sent.load(Ordering::Relaxed);

                // Approximate queue size (messages processed since last check)
                let queue_size = message_count as usize;

                // Check message queue size limit
                if !self
                    .dos_protection
                    .check_message_queue_size(queue_size)
                    .await
                {
                    warn!(
                        "Message queue size limit exceeded (processed {} messages), potential DoS",
                        queue_size
                    );
                    // Optionally detect DoS attack
                    if self.dos_protection.detect_dos_attack().await {
                        warn!("DoS attack detected - message queue and connections at high levels");
                        // Could trigger automatic mitigation here (e.g., increase rate limits, ban aggressive IPs)
                    }
                    // Reset counter to prevent false positives
                    // Note: Counter will be reset again after metrics update (line 2043),
                    // but this early reset prevents false positives in DoS detection.
                    // The assignment is intentional - value is used in next loop iteration.
                    #[allow(unused_assignments)] // Used in next loop iteration
                    {
                        message_count = 0;
                    }
                }

                self.dos_protection
                    .update_metrics(active_connections, queue_size, bytes_received, bytes_sent)
                    .await;

                last_metrics_update = now;
                message_count = 0; // Reset after update for next period
            }

            crate::network::parse_offload::note_peer_rx_dequeue();
            if let NetworkMessage::RawMessageReceived(data, _) = &message {
                crate::network::parse_offload::release_peer_rx(data.len());
            }

            // Process getdata inline (KEEP tc220 path). Spawn+SEM was forensics-only and
            // coincided with tip30 collapse; parallel W5 serve_getdata_request remains.
            // R-302: inbound BLOCK parse may leave this loop when PARSE_OFFLOAD=1.
            // That is not the serve path — do not spawn getdata handling here.
            let message = match crate::network::parse_offload::take_inbound_block(self, message) {
                Some(m) => m,
                None => continue,
            };
            super::network_message_dispatch::handle_network_message(self, message).await?;
        }
        Ok(())
    }

    // NOTE: Network message handling match extracted to network_message_dispatch::handle_network_message

    /// An unrecognized command is not a broken peer. Checksum and size errors are.
    pub(crate) fn disconnect_on_unrecognized_command(err: &anyhow::Error) -> bool {
        !format!("{err:#}").contains("Unknown command")
    }

    /// Parse incoming TCP wire message and process with protocol layer
    ///
    /// This function:
    /// 1. Converts ProtocolMessage to blvm_protocol::network::NetworkMessage
    /// 2. Gets or creates PeerState for this peer
    /// 3. Creates NodeChainAccess from storage modules
    /// 4. Calls blvm_protocol::network::process_network_message()
    /// 5. Converts NetworkResponse back to wire format and enqueues for sending
    pub async fn handle_incoming_wire_tcp(
        &self,
        peer_addr: SocketAddr,
        data: Vec<u8>,
    ) -> Result<()> {
        // Track bytes received (async-safe)
        self.track_bytes_received(data.len() as u64).await;

        // Check if peer is banned
        if self.is_banned(peer_addr) {
            warn!("Rejecting message from banned peer: {}", peer_addr);
            return Ok(()); // Silently drop messages from banned peers
        }

        // Check for mesh packet magic bytes (0x4D, 0x45, 0x53, 0x48 = "MESH")
        // Mesh packets are handled separately and routed to mesh module
        if data.len() >= 4 && data[0..4] == [0x4D, 0x45, 0x53, 0x48] {
            debug!(
                "Detected mesh packet from {}: {} bytes",
                peer_addr,
                data.len()
            );
            // Route mesh packet to mesh module via event system
            // The packet is sent to the message queue, which then publishes MeshPacketReceived events
            // that the mesh module can subscribe to (see handle_network_messages for event publishing)
            let _ = self
                .peer_tx
                .send(NetworkMessage::MeshPacketReceived(data, peer_addr));
            return Ok(());
        }

        // Check for Stratum V2 TLV message format
        // Stratum V2 messages start with [4-byte length][2-byte tag][4-byte length][payload]
        // Valid Stratum V2 tags are in range 0x0001-0x0032 (see blvm-stratum-v2/src/messages.rs)
        #[cfg(feature = "stratum-v2")]
        if self.p2p_stratum_demux_enabled && data.len() >= 10 {
            // Try to parse as Stratum V2 TLV
            let length_prefix = u32::from_le_bytes([data[0], data[1], data[2], data[3]]);
            if length_prefix >= 6 && length_prefix <= 1024 * 1024 {
                // Reasonable size limits
                let tag = u16::from_le_bytes([data[4], data[5]]);
                // Check if tag is a valid Stratum V2 message type
                // Valid tags: 0x0001-0x0003 (setup), 0x0010-0x0012 (channel), 0x0020-0x0021 (job), 0x0030-0x0032 (shares)
                if (tag >= 0x0001 && tag <= 0x0003)
                    || (tag >= 0x0010 && tag <= 0x0012)
                    || (tag >= 0x0020 && tag <= 0x0021)
                    || (tag >= 0x0030 && tag <= 0x0032)
                {
                    debug!(
                        "Detected Stratum V2 message from {}: tag={:04x}, {} bytes",
                        peer_addr,
                        tag,
                        data.len()
                    );
                    // Forward TLV part only (handle_message uses decode_raw, expects [tag][len][payload])
                    let _ = self.peer_tx.send(NetworkMessage::StratumV2MessageReceived(
                        data[4..].to_vec(),
                        peer_addr,
                    ));
                    return Ok(());
                }
            }
        }

        // R-304: hash 80-byte header before deserialize. Default-off is a no-op.
        if crate::network::parse_offload::is_block_frame(&data) {
            if let Some(hash) =
                crate::node::parallel_ibd::wire_hash_gate::try_skip_obsolete_block_frame(&data)
            {
                self.cancel_block_request_force(peer_addr, hash);
                return Ok(());
            }
        }

        let parsed = {
            let t0 = std::time::Instant::now();
            let parsed =
                match ProtocolParser::parse_message(&data) {
                    Ok(parsed) => parsed,
                    Err(err) => {
                        // An unrecognized command is ignored. Disconnecting here
                        // drops a peer that sent version and verack, then a command
                        // this node does not parse.
                        if !Self::disconnect_on_unrecognized_command(&err) {
                            info!("ignoring unrecognized message from {peer_addr}: {err:#}");
                            return Ok(());
                        }
                        let _ = self.peer_tx().send(NetworkMessage::PeerDisconnected(
                            TransportAddr::Tcp(peer_addr),
                        ));
                        return Err(err);
                    }
                };
            if matches!(parsed, ProtocolMessage::Block(_)) {
                crate::node::parallel_ibd::note_block_parse(t0.elapsed().as_millis() as u64, false);
            }
            parsed
        };

        // Publish MessageReceived event for module subscribers
        if let Some(ref ep) = *self.event_publisher.lock().await {
            let msg_type = protocol_message_type(&parsed);
            let peer_str = peer_addr.to_string();
            let size = data.len();
            let ep_clone = Arc::clone(ep);
            tokio::spawn(async move {
                ep_clone
                    .publish_message_received(&peer_str, msg_type, size, 0)
                    .await;
            });
        }

        if let ProtocolMessage::Tx(_) = &parsed {
            if self.should_drop_tx(peer_addr, &data).await {
                return Ok(());
            }
        }

        if let ProtocolMessage::Block(block_msg) = parsed {
            return self
                .handle_block_wire_message(peer_addr, block_msg, data)
                .await;
        }

        self.dispatch_protocol_message(peer_addr, &parsed, data)
            .await
    }

    fn pool_finality(&self) -> (u64, Option<blvm_protocol::types::TimeContext>) {
        let Some(storage) = self.storage.as_ref() else {
            return (0, None);
        };
        let Some((height, median_time_past)) = crate::node::mempool::next_block_finality(storage)
        else {
            return (0, None);
        };
        (
            height,
            Some(blvm_protocol::types::TimeContext {
                network_time: median_time_past,
                median_time_past,
            }),
        )
    }

    /// Submit validated transactions to the mempool.
    /// Each witness is the stack list from the package bytes. `None` is a
    /// transaction that carried no witness.
    pub(crate) async fn submit_transactions_to_mempool(
        &self,
        txs: &[(
            blvm_protocol::Transaction,
            Option<Vec<blvm_protocol::segwit::Witness>>,
        )],
    ) -> Result<()> {
        if let Some(ref mempool_manager) = self.mempool_manager {
            let flags = mempool_manager.add_package_with_witnesses(txs.to_vec())?;
            for ((tx, _), admitted) in txs.iter().zip(flags) {
                if !admitted {
                    continue;
                }
                let txid = blvm_protocol::block::calculate_tx_id(tx);
                let fee = mempool_manager.cached_fee_basis(&txid);
                self.announce_accepted_transaction(txid, fee).await;
            }
            return Ok(());
        }
        let utxo_lock = self.utxo_set.lock().await;
        let mempool_lock = self.mempool.lock().await;
        let (height, time_context) = self.pool_finality();
        for (tx, witnesses) in txs {
            let _ = self.consensus.accept_to_memory_pool_with_witness(
                tx,
                witnesses.as_deref(),
                &utxo_lock,
                &mempool_lock,
                height,
                time_context,
                Network::Mainnet,
            );
        }
        Ok(())
    }

    /// Submit one transaction, keeping its witness stacks when they were on the wire.
    pub(crate) async fn submit_transaction_with_witness(
        &self,
        tx: blvm_protocol::Transaction,
        witnesses: Option<Vec<blvm_protocol::segwit::Witness>>,
    ) -> Result<()> {
        if let Some(ref mempool_manager) = self.mempool_manager {
            // Route through MempoolManager so policy (min-fee, spam filter,
            // ancestor limits, RBF) is enforced. add_transaction uses interior
            // mutability (&self) so this is safe from Arc context.
            let txid = blvm_protocol::block::calculate_tx_id(&tx);
            match mempool_manager.add_transaction_with_witness(tx, witnesses) {
                Ok(true) => {
                    let fee = mempool_manager.cached_fee_basis(&txid);
                    self.announce_accepted_transaction(txid, fee).await;
                }
                Ok(false) => {}
                Err(e) => {
                    tracing::debug!("Transaction rejected by mempool: {}", e);
                }
            }
        } else {
            // Fallback: no MempoolManager configured, accept via consensus layer.
            let utxo_lock = self.utxo_set.lock().await;
            let mempool_lock = self.mempool.lock().await;
            let (height, time_context) = self.pool_finality();
            let _ = self.consensus.accept_to_memory_pool_with_witness(
                &tx,
                witnesses.as_deref(),
                &utxo_lock,
                &mempool_lock,
                height,
                time_context,
                Network::Mainnet,
            );
        }
        Ok(())
    }

    /// Get peer manager reference (locked)
    ///
    /// Note: This is now async and uses tokio::sync::Mutex to avoid deadlocks
    pub async fn peer_manager(&self) -> tokio::sync::MutexGuard<'_, PeerManager> {
        self.peer_manager.lock().await
    }

    /// Get current connected peer addresses as SocketAddr
    /// Returns addresses for TCP/Quinn peers only (Iroh peers use different addressing)
    pub async fn get_connected_peer_addresses(&self) -> Vec<SocketAddr> {
        let pm = self.peer_manager.lock().await;
        pm.peer_socket_addresses()
    }

    /// Check eclipse prevention for an IP address
    /// Returns true if connection is allowed, false if it would violate eclipse prevention
    pub fn check_eclipse_prevention(&self, ip: std::net::IpAddr) -> bool {
        let prefix = self.get_ip_prefix(ip);
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let diversity = self.peer_diversity.lock().await;
                let count = diversity.get(&prefix).copied().unwrap_or(0);
                count < 3 // Allow max 3 connections from same IP prefix
            })
        })
    }

    /// Get IP prefix (first 3 octets) for eclipse prevention
    pub fn get_ip_prefix(&self, ip: std::net::IpAddr) -> [u8; 3] {
        match ip {
            std::net::IpAddr::V4(ipv4) => {
                let octets = ipv4.octets();
                [octets[0], octets[1], octets[2]]
            }
            std::net::IpAddr::V6(ipv6) => {
                let octets = ipv6.octets();
                [octets[0], octets[1], octets[2]]
            }
        }
    }

    /// Remove peer diversity tracking for an IP address
    pub fn remove_peer_diversity(&self, ip: std::net::IpAddr) {
        let prefix = self.get_ip_prefix(ip);
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut diversity = self.peer_diversity.lock().await;
                if let Some(count) = diversity.get_mut(&prefix) {
                    if *count > 0 {
                        *count -= 1;
                    }
                    if *count == 0 {
                        diversity.remove(&prefix);
                    }
                }
            })
        })
    }

    /// Increment the eclipse-prevention diversity count for an IP.
    /// Must be called when a connection to `ip` is successfully established.
    pub(crate) async fn add_peer_diversity(&self, ip: std::net::IpAddr) {
        let prefix = self.get_ip_prefix(ip);
        *self.peer_diversity.lock().await.entry(prefix).or_insert(0) += 1;
    }

    /// Get filter service reference
    pub fn filter_service(&self) -> &crate::network::filter_service::BlockFilterService {
        &self.filter_service
    }

    pub(crate) fn inventory(&self) -> &Arc<std::sync::Mutex<super::inventory::InventoryManager>> {
        &self.inventory
    }

    /// Record inventory announced by a peer (`inv` message).
    pub(crate) fn handle_inventory_received(
        &self,
        data: &[u8],
        peer_addr: SocketAddr,
    ) -> Result<()> {
        let inv_msg = match ProtocolParser::parse_message(data) {
            Ok(ProtocolMessage::Inv(msg)) => msg,
            _ => return Ok(()),
        };
        let peer = peer_addr.to_string();
        self.inventory
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .add_inventory(&peer, &inv_msg.inventory)
    }

    /// Add a persistent peer (will be connected to on startup)
    pub fn add_persistent_peer(&self, addr: SocketAddr) {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut peers = self.persistent_peers.lock().await;
                peers.insert(addr);
            })
        })
    }

    /// Remove a persistent peer
    pub fn remove_persistent_peer(&self, addr: SocketAddr) {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut peers = self.persistent_peers.lock().await;
                peers.remove(&addr);
            })
        })
    }

    /// Get list of persistent peers (async version for RPC)
    pub async fn get_persistent_peers(&self) -> HashSet<SocketAddr> {
        self.persistent_peers.lock().await.clone()
    }

    /// Get list of persistent peers (sync version)
    pub fn get_persistent_peers_sync(&self) -> Vec<SocketAddr> {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let peers = self.persistent_peers.lock().await;
                peers.iter().cloned().collect()
            })
        })
    }

    fn persist_process_bans(&self, bans: &HashMap<SocketAddr, u64>) {
        let Some(dir) = &self.ban_store_dir else {
            return;
        };
        write_process_bans(dir, bans);
    }

    /// Ban a peer (with optional unban timestamp, 0 = permanent)
    pub fn ban_peer(&self, addr: SocketAddr, unban_timestamp: u64) {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut ban_list = self.ban_list.write().await;
                let key = SocketAddr::new(addr.ip(), 0);
                if unban_timestamp == 0 {
                    // Permanent ban (use max timestamp)
                    ban_list.insert(key, u64::MAX);
                } else {
                    ban_list.insert(key, unban_timestamp);
                }
                let snapshot = ban_list.clone();
                drop(ban_list);
                self.persist_process_bans(&snapshot);

                // Publish PeerBanned event for module subscribers
                let ban_duration_secs = if unban_timestamp == 0 || unban_timestamp == u64::MAX {
                    0 // Permanent
                } else {
                    unban_timestamp.saturating_sub(crate::utils::current_timestamp())
                };
                if let Some(ref ep) = *self.event_publisher.lock().await {
                    ep.publish_peer_banned(&addr.to_string(), "ban", ban_duration_secs)
                        .await;
                }
                if let Some(cfg) = self.ban_list_sharing_config.as_ref() {
                    if cfg.shares()
                        && matches!(cfg.share_mode, crate::config::BanShareMode::Immediate)
                    {
                        super::handlers::ban_list::broadcast_ban_list(
                            self.ban_list().as_ref(),
                            self.peer_manager_mutex().as_ref(),
                            cfg.min_ban_duration_to_share,
                            self.event_publisher(),
                        )
                        .await;
                    }
                }
            })
        })
    }

    /// Unban a peer
    pub fn unban_peer(&self, addr: SocketAddr) {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut ban_list = self.ban_list.write().await;
                ban_list.remove(&SocketAddr::new(addr.ip(), 0));
                drop(ban_list);

                // Publish PeerUnbanned event for module subscribers
                if let Some(ref ep) = *self.event_publisher.lock().await {
                    ep.publish_peer_unbanned(&addr.to_string()).await;
                }
            })
        })
    }

    /// Clear all bans
    pub fn clear_bans(&self) {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let mut ban_list = self.ban_list.write().await;
                ban_list.clear();
            })
        })
    }

    /// Send our own address to peers (self-advertisement)
    pub async fn advertise_self(&self, listen_addr: SocketAddr, services: u64) -> Result<()> {
        if !self.enable_self_advertisement {
            return Ok(()); // Self-advertisement disabled
        }

        use crate::network::protocol::{
            AddrMessage, NetworkAddress, ProtocolMessage, ProtocolParser,
        };
        use std::net::IpAddr;

        // Convert SocketAddr to NetworkAddress
        let ip_bytes = match listen_addr.ip() {
            IpAddr::V4(ipv4) => {
                // IPv4-mapped IPv6 format
                let mut bytes = [0u8; 16];
                bytes[10] = 0xff;
                bytes[11] = 0xff;
                bytes[12..16].copy_from_slice(&ipv4.octets());
                bytes
            }
            IpAddr::V6(ipv6) => ipv6.octets(),
        };

        let our_addr = NetworkAddress {
            time: 0,
            services,
            ip: ip_bytes,
            port: listen_addr.port(),
        };

        // Create Addr message with just our address
        let addr_msg = AddrMessage {
            addresses: vec![our_addr.clone()],
        };
        let relay_msg = ProtocolMessage::Addr(addr_msg);
        let wire_msg = ProtocolParser::serialize_message(&relay_msg)?;

        // Send to all connected peers
        let peer_addrs: Vec<SocketAddr> = {
            let pm = self.peer_manager.lock().await;
            pm.peer_socket_addresses()
        };

        for peer_addr in peer_addrs {
            if let Err(e) = self.send_to_peer(peer_addr, wire_msg.clone()).await {
                warn!("Failed to advertise self to {}: {}", peer_addr, e);
            }
        }

        // Also store our own address in database
        {
            let mut db = self.address_database.write().await;
            db.add_address(our_addr, services);
        }

        Ok(())
    }

    /// Get list of banned peers
    pub fn get_banned_peers(&self) -> Vec<(SocketAddr, u64)> {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let ban_list = self.ban_list.read().await;
                ban_list
                    .iter()
                    .map(|(addr, timestamp)| (*addr, *timestamp))
                    .collect()
            })
        })
    }

    /// Get peer addresses (async version for RPC)
    pub async fn get_peer_addresses(&self) -> Vec<TransportAddr> {
        let pm = self.peer_manager.lock().await;
        pm.peer_addresses()
    }

    /// Set network active state
    pub async fn set_network_active(&self, active: bool) -> Result<()> {
        let mut state = self.network_active.lock().await;
        *state = active;
        info!("Network active state set to: {}", active);
        Ok(())
    }

    /// Get network active state
    pub fn is_network_active(&self) -> bool {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async { *self.network_active.lock().await })
        })
    }

    /// Check if a peer is banned
    pub fn is_banned(&self, addr: SocketAddr) -> bool {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let ban_list = self.ban_list.read().await;
                if let Some(&unban_timestamp) = ban_list.get(&SocketAddr::new(addr.ip(), 0)) {
                    if unban_timestamp == u64::MAX {
                        return true; // Permanent ban
                    }
                    // Check if ban has expired
                    let now = current_timestamp();
                    if now < unban_timestamp {
                        return true; // Still banned
                    } else {
                        // Ban expired, remove it
                        drop(ban_list);
                        self.unban_peer(addr);
                        return false;
                    }
                }
                false
            })
        })
    }

    /// Ban one Iroh endpoint. A second endpoint is a different key.
    #[cfg(feature = "iroh")]
    pub async fn ban_identity(&self, id: &super::transport::PeerIdentity) {
        let super::transport::PeerIdentity::Iroh(endpoint) = id else {
            return;
        };
        self.iroh_bans
            .write()
            .await
            .insert(endpoint.clone(), u64::MAX);
    }

    /// Whether this Iroh endpoint is banned. TCP bans stay on `is_banned`.
    #[cfg(feature = "iroh")]
    pub async fn identity_is_banned(&self, id: &super::transport::PeerIdentity) -> bool {
        let super::transport::PeerIdentity::Iroh(endpoint) = id else {
            return false;
        };
        self.iroh_bans.read().await.contains_key(endpoint)
    }

    /// Record a handshake entry for this Iroh connection alone.
    #[cfg(feature = "iroh")]
    pub async fn remember_handshake(&self, id: &super::transport::PeerIdentity) {
        let super::transport::PeerIdentity::Iroh(endpoint) = id else {
            return;
        };
        self.iroh_peer_states
            .lock()
            .await
            .entry(endpoint.clone())
            .or_insert_with(blvm_protocol::network::PeerState::new);
    }

    #[cfg(feature = "iroh")]
    pub async fn handshake_known(&self, id: &super::transport::PeerIdentity) -> bool {
        let super::transport::PeerIdentity::Iroh(endpoint) = id else {
            return false;
        };
        self.iroh_peer_states.lock().await.contains_key(endpoint)
    }

    /// One rate-limit entry per Iroh endpoint.
    #[cfg(feature = "iroh")]
    pub async fn note_message_rate(&self, id: &super::transport::PeerIdentity) {
        let super::transport::PeerIdentity::Iroh(endpoint) = id else {
            return;
        };
        self.iroh_message_rates
            .lock()
            .await
            .entry(endpoint.clone())
            .or_insert_with(|| PeerRateLimiter::new(1, 1));
    }

    #[cfg(feature = "iroh")]
    pub async fn identity_rate_entries(
        &self,
        left: &super::transport::PeerIdentity,
        right: &super::transport::PeerIdentity,
    ) -> usize {
        let rates = self.iroh_message_rates.lock().await;
        [left, right]
            .into_iter()
            .filter(|id| match id {
                super::transport::PeerIdentity::Iroh(endpoint) => rates.contains_key(endpoint),
                super::transport::PeerIdentity::Socket(_) => false,
            })
            .count()
    }

    /// A socket used only to keep an Iroh frame on the existing TCP parse path.
    /// It is not `0.0.0.0:0`, and it is not the handshake, ban, or rate key.
    #[cfg(feature = "iroh")]
    fn iroh_dispatch_socket(endpoint: &[u8]) -> SocketAddr {
        let mut ip = [240, 0, 0, 1];
        for (i, byte) in endpoint.iter().enumerate() {
            ip[1 + (i % 3)] ^= *byte;
        }
        let port = u16::from(endpoint.first().copied().unwrap_or(1)).saturating_add(1);
        SocketAddr::from((ip, port))
    }

    /// Ban, rate-limit, and handshake lookups for a queued frame.
    /// An Iroh frame keeps its endpoint. A TCP frame keeps its socket.
    pub(crate) async fn handle_incoming_identity(
        &self,
        peer: super::transport::PeerIdentity,
        data: Vec<u8>,
    ) -> Result<()> {
        match peer {
            super::transport::PeerIdentity::Socket(addr) => {
                self.handle_incoming_wire_tcp(addr, data).await
            }
            #[cfg(feature = "iroh")]
            super::transport::PeerIdentity::Iroh(endpoint) => {
                let id = super::transport::PeerIdentity::Iroh(endpoint.clone());
                if self.identity_is_banned(&id).await {
                    return Ok(());
                }
                self.note_message_rate(&id).await;
                self.remember_handshake(&id).await;
                let addr = Self::iroh_dispatch_socket(&endpoint);
                self.handle_incoming_wire_tcp(addr, data).await
            }
        }
    }

    /// Track bytes sent (async-safe)
    ///
    /// Optimization: Uses AtomicU64 for lock-free operation
    /// Get bytes sent (for tests)
    pub fn bytes_sent(&self) -> u64 {
        self.bytes_sent.load(Ordering::Relaxed)
    }

    pub async fn track_bytes_sent(&self, bytes: u64) {
        self.bytes_sent.fetch_add(bytes, Ordering::Relaxed);
    }

    /// Track bytes received (async-safe)
    ///
    /// Optimization: Uses AtomicU64 for lock-free operation
    pub async fn track_bytes_received(&self, bytes: u64) {
        self.bytes_received.fetch_add(bytes, Ordering::Relaxed);
    }

    /// Get DoS protection manager reference
    pub fn dos_protection(&self) -> &Arc<super::dos_protection::DosProtectionManager> {
        &self.dos_protection
    }

    /// Get peer manager reference (for module API access)
    pub fn peer_manager_ref(&self) -> &Arc<Mutex<PeerManager>> {
        &self.peer_manager
    }

    pub(crate) fn ban_list(&self) -> &Arc<RwLock<HashMap<SocketAddr, u64>>> {
        &self.ban_list
    }

    pub(crate) fn address_database(&self) -> &Arc<RwLock<super::address_db::AddressDatabase>> {
        &self.address_database
    }

    pub(crate) fn peer_tx(&self) -> &mpsc::UnboundedSender<NetworkMessage> {
        &self.peer_tx
    }

    pub(crate) fn last_addr_sent(&self) -> &Arc<Mutex<HashMap<SocketAddr, u64>>> {
        &self.last_addr_sent
    }

    pub(crate) fn bandwidth_protection(
        &self,
    ) -> &Arc<super::bandwidth_protection::BandwidthProtectionManager> {
        &self.bandwidth_protection
    }

    pub(crate) fn storage(&self) -> &Option<Arc<Storage>> {
        &self.storage
    }

    /// When set, used for protocol feature gates (e.g. segwit) when serving `getdata` blocks.
    pub(crate) fn protocol_engine(&self) -> Option<&Arc<BitcoinProtocolEngine>> {
        self.protocol_engine.as_ref()
    }

    /// Confirmed + mempool transaction lookup for `getdata` (`MSG_TX`).
    pub(crate) fn mempool_manager(&self) -> Option<&Arc<MempoolManager>> {
        self.mempool_manager.as_ref()
    }

    /// Merge block hashes into the full-block serve denylist (modules via [`crate::module::traits::NodeAPI`]).
    pub(crate) fn merge_block_serve_denylist(&self, hashes: &[blvm_protocol::Hash]) {
        if hashes.is_empty() {
            return;
        }
        let mut g = self.block_serve_denylist.write();
        for h in hashes {
            g.insert(*h);
        }
    }

    pub(crate) fn is_block_serve_denied(&self, hash: &blvm_protocol::Hash) -> bool {
        self.block_serve_denylist.read().contains(hash)
    }

    pub(crate) fn block_serve_maintenance_mode(&self) -> bool {
        self.block_serve_maintenance.load(Ordering::Relaxed)
    }

    pub(crate) fn set_block_serve_maintenance_mode(&self, enabled: bool) {
        self.block_serve_maintenance
            .store(enabled, Ordering::Relaxed);
    }

    pub(crate) fn block_serve_denylist_snapshot(
        &self,
    ) -> crate::module::traits::BlockServeDenylistSnapshot {
        let (total_count, truncated, hashes) = sorted_denylist_snapshot(
            &self.block_serve_denylist.read(),
            crate::module::traits::SERVE_DENYLIST_SNAPSHOT_MAX_HASHES,
        );
        crate::module::traits::BlockServeDenylistSnapshot {
            total_count,
            truncated,
            hashes,
        }
    }

    pub(crate) fn clear_block_serve_denylist(&self) {
        self.block_serve_denylist.write().clear();
    }

    pub(crate) fn replace_block_serve_denylist(&self, hashes: &[blvm_protocol::Hash]) {
        let mut g = self.block_serve_denylist.write();
        g.clear();
        for h in hashes {
            g.insert(*h);
        }
    }

    /// Merge txids that must not be answered with a full `tx` on `getdata`.
    pub(crate) fn merge_tx_serve_denylist(&self, hashes: &[blvm_protocol::Hash]) {
        if hashes.is_empty() {
            return;
        }
        let mut g = self.tx_serve_denylist.write();
        for h in hashes {
            g.insert(*h);
        }
    }

    pub(crate) fn is_tx_serve_denied(&self, hash: &blvm_protocol::Hash) -> bool {
        self.tx_serve_denylist.read().contains(hash)
    }

    pub(crate) fn tx_serve_denylist_snapshot(
        &self,
    ) -> crate::module::traits::TxServeDenylistSnapshot {
        let (total_count, truncated, hashes) = sorted_denylist_snapshot(
            &self.tx_serve_denylist.read(),
            crate::module::traits::SERVE_DENYLIST_SNAPSHOT_MAX_HASHES,
        );
        crate::module::traits::TxServeDenylistSnapshot {
            total_count,
            truncated,
            hashes,
        }
    }

    pub(crate) fn clear_tx_serve_denylist(&self) {
        self.tx_serve_denylist.write().clear();
    }

    pub(crate) fn replace_tx_serve_denylist(&self, hashes: &[blvm_protocol::Hash]) {
        let mut g = self.tx_serve_denylist.write();
        g.clear();
        for h in hashes {
            g.insert(*h);
        }
    }

    pub(crate) fn replay_protection(&self) -> &Arc<super::replay_protection::ReplayProtection> {
        &self.replay_protection
    }

    pub(crate) fn module_registry(
        &self,
    ) -> &Arc<tokio::sync::Mutex<Option<Arc<crate::module::registry::client::ModuleRegistry>>>>
    {
        &self.module_registry
    }

    pub(crate) fn node_api(
        &self,
    ) -> &Arc<tokio::sync::Mutex<Option<Arc<dyn crate::module::traits::NodeAPI>>>> {
        &self.node_api
    }

    pub(crate) fn module_encryption(
        &self,
    ) -> &Arc<tokio::sync::Mutex<Option<Arc<crate::module::encryption::ModuleEncryption>>>> {
        &self.module_encryption
    }

    pub(crate) fn modules_dir(&self) -> &Arc<tokio::sync::Mutex<Option<std::path::PathBuf>>> {
        &self.modules_dir
    }

    pub(crate) fn node_payment_script(&self) -> &Arc<tokio::sync::Mutex<Option<Vec<u8>>>> {
        &self.node_payment_script
    }

    pub(crate) fn event_publisher(
        &self,
    ) -> &Arc<tokio::sync::Mutex<Option<Arc<crate::node::event_publisher::EventPublisher>>>> {
        &self.event_publisher
    }

    pub(crate) fn ibd_protection(&self) -> &Arc<super::ibd_protection::IbdProtectionManager> {
        &self.ibd_protection
    }

    pub(crate) fn peer_manager_mutex(&self) -> &Arc<Mutex<PeerManager>> {
        &self.peer_manager
    }

    pub(crate) fn peer_reconnection_queue(
        &self,
    ) -> &Arc<Mutex<HashMap<SocketAddr, (u32, u64, f64)>>> {
        &self.peer_reconnection_queue
    }

    pub(crate) fn persistent_peers_lock(&self) -> &Arc<Mutex<HashSet<SocketAddr>>> {
        &self.persistent_peers
    }

    pub(crate) fn connections_per_ip(&self) -> &Arc<Mutex<HashMap<std::net::IpAddr, usize>>> {
        &self.connections_per_ip
    }

    pub(crate) fn peer_message_rates(&self) -> &Arc<Mutex<HashMap<SocketAddr, PeerRateLimiter>>> {
        &self.peer_message_rates
    }

    pub(crate) fn peer_tx_rate_limiters(
        &self,
    ) -> &Arc<Mutex<HashMap<SocketAddr, PeerRateLimiter>>> {
        &self.peer_tx_rate_limiters
    }

    pub(crate) fn peer_tx_byte_rate_limiters(
        &self,
    ) -> &Arc<Mutex<HashMap<SocketAddr, PeerByteRateLimiter>>> {
        &self.peer_tx_byte_rate_limiters
    }

    pub(crate) fn tcp_transport(&self) -> &TcpTransport {
        &self.tcp_transport
    }

    #[cfg(feature = "quinn")]
    pub(crate) fn quinn_transport(
        &self,
    ) -> &std::sync::Mutex<Option<Arc<crate::network::quinn_transport::QuinnTransport>>> {
        &self.quinn_transport
    }

    #[cfg(feature = "iroh")]
    pub(crate) fn iroh_transport(
        &self,
    ) -> &std::sync::Mutex<Option<std::sync::Arc<crate::network::iroh_transport::IrohTransport>>>
    {
        &self.iroh_transport
    }

    /// Socket-to-transport mapping (for utxo_commitments_client)
    pub(crate) fn socket_to_transport(
        &self,
    ) -> &Arc<Mutex<HashMap<SocketAddr, super::transport::TransportAddr>>> {
        &self.socket_to_transport
    }

    /// Transaction requests outstanding to one peer.
    pub(crate) fn tx_requests(
        &self,
    ) -> &std::sync::Mutex<crate::network::tx_inflight::TxRequestTracker> {
        &self.tx_requests
    }

    /// Peer states (for utxo_commitments_client)
    pub(crate) fn peer_states(
        &self,
    ) -> &Arc<RwLock<HashMap<SocketAddr, blvm_protocol::network::PeerState>>> {
        &self.peer_states
    }

    /// Request timeout config (for utxo_commitments_client)
    pub(crate) fn request_timeout_config(&self) -> &Arc<crate::config::RequestTimeoutConfig> {
        &self.request_timeout_config
    }

    /// Protocol limits (for wire_dispatch, addr handler, etc.)
    pub(crate) fn protocol_limits(&self) -> &Arc<crate::config::ProtocolLimitsConfig> {
        &self.protocol_limits
    }

    /// Background task interval config (for background_tasks)
    pub(crate) fn background_task_config(&self) -> &Arc<crate::config::BackgroundTaskConfig> {
        &self.background_task_config
    }

    pub(crate) fn relay_policies(&self) -> &super::relay::RelayPolicies {
        &self.relay_policies
    }

    pub(crate) fn relay_manager(&self) -> &std::sync::Mutex<super::relay::RelayManager> {
        &self.relay_manager
    }

    pub(crate) fn network_timing(&self) -> &Arc<crate::config::NetworkTimingConfig> {
        &self.network_timing
    }

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

    /// Consume one message token when `[peer_rate_limiting]` is set.
    pub(crate) async fn allow_peer_message(&self, peer: SocketAddr) -> bool {
        let Some((burst, rate)) = self.peer_message_rate else {
            return true;
        };
        let mut rates = self.peer_message_rates().lock().await;
        let limiter = rates
            .entry(peer)
            .or_insert_with(|| PeerRateLimiter::new(burst, rate));
        limiter.check_and_consume()
    }

    pub(crate) fn ban_sharing_active(&self) -> bool {
        self.ban_list_sharing_config
            .as_ref()
            .is_some_and(|cfg| cfg.shares())
    }

    pub(crate) fn headers_probe_timeout_secs(&self) -> u64 {
        self.headers_probe_timeout_secs
    }

    /// Pending requests (for utxo_commitments_client)
    pub(crate) fn pending_requests(&self) -> &Arc<Mutex<HashMap<u64, PendingRequest>>> {
        &self.pending_requests
    }

    /// Drop pending GetData oneshots for a TCP peer that just left.
    ///
    /// Handshake `PeerDisconnected` already did this. The live dispatch path
    /// (`handle_peer_disconnected`) did not — R-242 RST of covering hero
    /// `3.136.178.225` left inflight GetData parked until the 30–45s deadline,
    /// then TIP_WALK_PROMOTE retitled the corpse as owner of 186264.
    pub async fn cancel_pending_block_requests_for_disconnected_peer(&self, sa: SocketAddr) {
        let peer_ip = Self::block_request_key(sa);
        let mut pending = self.pending_block_requests.lock().await;
        pending.retain(|(ip, _), _| *ip != peer_ip);
    }

    /// Pending block requests (for utxo_commitments_client)
    pub(crate) fn pending_block_requests(
        &self,
    ) -> &Arc<
        Mutex<
            HashMap<
                (IpAddr, blvm_protocol::Hash),
                Vec<
                    tokio::sync::oneshot::Sender<(
                        blvm_protocol::Block,
                        Vec<Vec<blvm_protocol::segwit::Witness>>,
                        Option<Vec<u8>>,
                    )>,
                >,
            >,
        >,
    > {
        &self.pending_block_requests
    }

    /// Get network statistics
    pub async fn get_network_stats(&self) -> crate::node::metrics::NetworkMetrics {
        let sent = self.bytes_sent.load(Ordering::Relaxed);
        let received = self.bytes_received.load(Ordering::Relaxed);
        let active_connections = {
            let pm = self.peer_manager.lock().await;
            pm.peer_count()
        };
        let banned_peers_count = {
            let ban_list = self.ban_list.read().await;
            ban_list.len()
        };
        let _resource_metrics = self.dos_protection.get_metrics().await;

        crate::node::metrics::NetworkMetrics {
            peer_count: active_connections,
            bytes_sent: sent,
            bytes_received: received,
            messages_sent: 0,     // Would need to track this
            messages_received: 0, // Would need to track this
            active_connections,
            banned_peers: banned_peers_count,
            connection_attempts: 0, // Would need to track this
            connection_failures: 0, // Would need to track this
            dos_protection: crate::node::metrics::DosMetrics {
                connection_rate_violations: 0,   // Would need to track this
                auto_bans: 0,                    // Would need to track this
                message_queue_overflows: 0,      // Would need to track this
                active_connection_limit_hits: 0, // Would need to track this
                resource_exhaustion_events: 0,   // Would need to track this
            },
        }
    }

    /// Get network statistics (legacy method for backward compatibility)
    pub fn get_network_stats_legacy(&self) -> (u64, u64) {
        // Use block_in_place to avoid blocking async runtime
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let sent = self.bytes_sent.load(Ordering::Relaxed);
                let received = self.bytes_received.load(Ordering::Relaxed);
                (sent, received)
            })
        })
    }

    /// Returns the shared UTXO set Arc so other components (e.g. MempoolManager) can be wired in.
    pub fn utxo_set_arc(&self) -> Arc<Mutex<UtxoSet>> {
        Arc::clone(&self.utxo_set)
    }

    /// Replace the shared UTXO snapshot (called after loading chainstate or connecting a block).
    pub async fn replace_utxo_snapshot(&self, snapshot: UtxoSet) {
        *self.utxo_set.lock().await = snapshot;
    }

    /// Load the current UTXO set from storage into the shared snapshot.
    pub async fn sync_utxo_from_storage(&self) -> Result<()> {
        let Some(ref storage) = self.storage else {
            return Ok(());
        };
        // Filling the shared set from the table hides checkpoint coins and
        // keeps spent table coins spendable. Misses are read from the chain.
        let snapshot = if storage.open_engine_resume_snapshot()?.is_some() {
            UtxoSet::default()
        } else {
            storage.utxos().get_all_utxos()?
        };
        self.replace_utxo_snapshot(snapshot).await;
        Ok(())
    }

    /// Service bits a version message from this node carries before any caller-supplied extras.
    pub fn advertised_services(&self) -> u64 {
        use blvm_protocol::bip157::NODE_COMPACT_FILTERS;
        use blvm_protocol::service_flags::standard;

        // NODE_WITNESS stays set. The genesis body decides history bits.
        // Present: NODE_NETWORK. Absent: NODE_NETWORK_LIMITED.
        // No storage keeps the old branch and advertises NODE_NETWORK.
        let mut services_with_filters = standard::NODE_WITNESS;
        if self.genesis_body_absent() {
            services_with_filters |= standard::NODE_NETWORK_LIMITED;
        } else {
            services_with_filters |= standard::NODE_NETWORK;
        }

        if self.serve_compact_filters {
            services_with_filters |= NODE_COMPACT_FILTERS;
        }

        #[cfg(feature = "utxo-commitments")]
        {
            services_with_filters |= crate::network::protocol::NODE_UTXO_COMMITMENTS;
        }

        if self.ban_sharing_active() {
            services_with_filters |= crate::network::protocol::NODE_BAN_LIST_SHARING;
        }

        services_with_filters |= crate::network::protocol::NODE_PACKAGE_RELAY;

        if let Ok(bits) = self.module_service_bits.lock() {
            for extra in bits.values() {
                services_with_filters |= *extra;
            }
        }

        #[cfg(feature = "governance")]
        if self.governance_config.is_some() {
            services_with_filters |= crate::network::protocol::NODE_GOVERNANCE;
        }

        #[cfg(feature = "dandelion")]
        if self.relay_policies.enable_dandelion {
            services_with_filters |= crate::network::protocol::NODE_DANDELION;
        }

        self.apply_history_service_bits(services_with_filters)
    }

    /// Record `NODE_FIBRE` for `module_name`. Other bits are dropped.
    pub fn set_module_service_bits(&self, module_name: &str, bits: u64) {
        let allowed = bits & crate::network::protocol::NODE_FIBRE;
        if module_name.is_empty() || allowed == 0 {
            return;
        }
        if let Ok(mut map) = self.module_service_bits.lock() {
            map.insert(module_name.to_string(), allowed);
        }
    }

    /// Drop service bits recorded for `module_name`.
    pub fn clear_module_service_bits(&self, module_name: &str) {
        if let Ok(mut map) = self.module_service_bits.lock() {
            map.remove(module_name);
        }
    }

    /// Storage with no genesis body cannot serve the full chain.
    /// Missing storage is not that case: the caller still advertises `NODE_NETWORK`.
    fn genesis_body_absent(&self) -> bool {
        let Some(storage) = self.storage.as_ref() else {
            return false;
        };
        let Ok(Some(hash)) = storage.blocks().get_hash_by_height(0) else {
            return true;
        };
        !storage.blocks().has_block_body(&hash).unwrap_or(false)
    }

    /// A stored genesis body keeps `NODE_NETWORK`. A missing one keeps `NODE_NETWORK_LIMITED`.
    fn apply_history_service_bits(&self, services: u64) -> u64 {
        use blvm_protocol::service_flags::standard;
        if self.genesis_body_absent() {
            (services & !standard::NODE_NETWORK) | standard::NODE_NETWORK_LIMITED
        } else {
            (services & !standard::NODE_NETWORK_LIMITED) | standard::NODE_NETWORK
        }
    }

    /// Relay floor as BTC per kvB. An unset mempool policy is 100 sat/kvB.
    pub fn relay_fee_btc_per_kvb(&self) -> f64 {
        let sats = self
            .mempool_manager
            .as_ref()
            .map(|pool| pool.min_relay_fee_sat_per_kvb())
            .unwrap_or(100);
        sats as f64 / 100_000_000.0
    }

    /// Create version message with service flags for all supported features.
    ///
    /// Sets service flags based on:
    /// - Standard Bitcoin flags:
    ///   - NODE_NETWORK (bit 0) - Always set for full nodes
    ///   - NODE_WITNESS (bit 3) - SegWit support (always enabled)
    /// - BIP157: NODE_COMPACT_FILTERS when `serve_compact_filters` is set
    /// - UTXO Commitments: NODE_UTXO_COMMITMENTS (if feature enabled)
    /// - Ban List Sharing: NODE_BAN_LIST_SHARING (if config enabled)
    /// - Dandelion: NODE_DANDELION (if feature enabled)
    /// - Package relay: `NODE_PACKAGE_RELAY` (`1 << 25`) because `sendpackages`,
    ///   `ancpkginfo`, `getpkgtxns`, and `pkgtxns` round-trip
    /// - FIBRE: `NODE_FIBRE` (`1 << 26`) stays clear
    pub fn create_version_message(
        &self,
        version: i32,
        services: u64,
        timestamp: i64,
        addr_recv: crate::network::protocol::NetworkAddress,
        addr_from: crate::network::protocol::NetworkAddress,
        nonce: u64,
        user_agent: String,
        start_height: i32,
        relay: bool,
    ) -> crate::network::protocol::VersionMessage {
        crate::network::protocol::VersionMessage {
            version,
            services: self.apply_history_service_bits(self.advertised_services() | services),
            timestamp,
            addr_recv,
            addr_from,
            nonce,
            user_agent,
            start_height,
            relay,
        }
    }
}

// NetworkManager: internal state uses Arc<Mutex>/Arc<RwLock> and is Send + Sync when those hold Send types.

#[cfg(test)]
mod pruning_version_services {
    use super::NetworkManager;
    use crate::config::{PruningConfig, PruningMode};
    use crate::node::mempool::MempoolManager;
    use crate::storage::Storage;
    use crate::storage::database::default_backend;
    use blvm_protocol::service_flags::standard::{NODE_NETWORK, NODE_NETWORK_LIMITED};
    use blvm_protocol::{BitcoinProtocolEngine, ProtocolVersion};
    use std::net::SocketAddr;
    use std::sync::Arc;

    fn version_services(storage: Arc<Storage>) -> u64 {
        let listen: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let protocol = Arc::new(BitcoinProtocolEngine::new(ProtocolVersion::Regtest).unwrap());
        let mempool = Arc::new(MempoolManager::new());
        let manager = NetworkManager::new(listen).with_dependencies(protocol, storage, mempool);
        let addr = crate::network::protocol::NetworkAddress {
            time: 0,
            services: 0,
            ip: [0u8; 16],
            port: 0,
        };
        manager
            .create_version_message(
                70015,
                NODE_NETWORK,
                0,
                addr.clone(),
                addr,
                1,
                String::new(),
                0,
                true,
            )
            .services
    }

    #[tokio::test]
    async fn genesis_body_sets_the_history_bit() {
        use crate::network::protocol::{NODE_FIBRE, NODE_PACKAGE_RELAY};

        let listen: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let bare = NetworkManager::new(listen).advertised_services();
        assert_ne!(bare & NODE_NETWORK, 0);
        assert_eq!(bare & NODE_NETWORK_LIMITED, 0);
        assert_ne!(bare & NODE_PACKAGE_RELAY, 0);
        assert_eq!(bare & NODE_FIBRE, 0);

        let listen: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let manager = NetworkManager::new(listen);
        let addr = crate::network::protocol::NetworkAddress {
            time: 0,
            services: 0,
            ip: [0u8; 16],
            port: 0,
        };
        let version_bits = |nm: &NetworkManager| {
            nm.create_version_message(
                70015,
                0,
                0,
                addr.clone(),
                addr.clone(),
                1,
                String::new(),
                0,
                true,
            )
            .services
        };
        assert_eq!(version_bits(&manager) & NODE_FIBRE, 0);
        manager.set_module_service_bits("relay-mod", NODE_FIBRE);
        assert_ne!(version_bits(&manager) & NODE_FIBRE, 0);
        manager.clear_module_service_bits("relay-mod");
        assert_eq!(version_bits(&manager) & NODE_FIBRE, 0);

        let empty = tempfile::TempDir::new().unwrap();
        let mut config = PruningConfig::default();
        config.mode = PruningMode::Normal {
            keep_from_height: 0,
            min_recent_blocks: 288,
        };
        let empty_storage = Arc::new(
            Storage::with_backend_and_pruning(empty.path(), default_backend(), Some(config))
                .unwrap(),
        );
        let empty_bits = version_services(Arc::clone(&empty_storage));
        assert_eq!(empty_bits & NODE_NETWORK, 0);
        assert_ne!(empty_bits & NODE_NETWORK_LIMITED, 0);
        assert_ne!(empty_bits & NODE_PACKAGE_RELAY, 0);
        assert_eq!(empty_bits & NODE_FIBRE, 0);

        let genesis = blvm_consensus::block::genesis_block(blvm_protocol::types::Network::Regtest);
        let genesis_hash = empty_storage.blocks().get_block_hash(&genesis);
        empty_storage.blocks().store_block(&genesis).unwrap();
        empty_storage
            .blocks()
            .store_height(0, &genesis_hash)
            .unwrap();
        let with_genesis = version_services(empty_storage);
        assert_ne!(with_genesis & NODE_NETWORK, 0);
        assert_eq!(with_genesis & NODE_NETWORK_LIMITED, 0);
        assert_ne!(with_genesis & (1 << 25), 0);
        assert_eq!(with_genesis & (1 << 26), 0);
    }

    #[tokio::test]
    async fn compact_filter_bit_is_clear_unless_the_flag_is_set() {
        use blvm_protocol::bip157::NODE_COMPACT_FILTERS;
        let listen: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let off = NetworkManager::new(listen);
        assert_eq!(off.advertised_services() & NODE_COMPACT_FILTERS, 0);

        let mut config = crate::config::NodeConfig::default();
        assert!(!config.serve_compact_filters);
        let from_default = NetworkManager::with_config(
            listen,
            8,
            crate::network::transport::TransportPreference::TCP_ONLY,
            Some(&config),
        );
        assert_eq!(from_default.advertised_services() & NODE_COMPACT_FILTERS, 0);

        config.serve_compact_filters = true;
        let on = NetworkManager::with_config(
            listen,
            8,
            crate::network::transport::TransportPreference::TCP_ONLY,
            Some(&config),
        );
        assert_ne!(on.advertised_services() & NODE_COMPACT_FILTERS, 0);

        let omitted: crate::config::NodeConfig = toml::from_str("").unwrap();
        assert!(!omitted.serve_compact_filters);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn block_completion_delivers_to_every_live_waiter() {
        use blvm_protocol::Block;
        let manager = NetworkManager::new("127.0.0.1:0".parse().unwrap());
        let peer: SocketAddr = "1.2.3.4:8333".parse().unwrap();
        let hash = [9u8; 32];
        let mut receivers = manager.register_block_requests_batch(peer, &[hash, hash]);
        let block = Block {
            header: blvm_protocol::BlockHeader {
                version: 1,
                prev_block_hash: [0u8; 32],
                merkle_root: [1u8; 32],
                timestamp: 1,
                bits: 0x207fffff,
                nonce: 1,
            },
            transactions: Vec::new().into(),
        };
        assert!(manager.complete_block_request(peer, hash, block, Vec::new()));
        let (first, _, _) = receivers.remove(0).try_recv().expect("older waiter");
        let (second, _, _) = receivers.remove(0).try_recv().expect("newer waiter");
        assert_eq!(first.header.nonce, 1);
        assert_eq!(second.header.nonce, 1);
    }
}

#[cfg(test)]
mod block_fetch_timeout_tests {
    use super::NetworkManager;
    use crate::config::RequestTimeoutConfig;
    use std::time::Duration;

    #[test]
    fn request_block_waits_for_the_configured_async_timeout() {
        let wait = NetworkManager::block_fetch_timeout(&RequestTimeoutConfig::default());
        assert_eq!(wait, Duration::from_secs(300));
        assert_ne!(wait, Duration::from_secs(30));
    }

    #[test]
    fn an_unknown_command_does_not_drop_the_peer() {
        let unknown = anyhow::anyhow!("Unknown command: sendaddrv2");
        let bad_checksum = anyhow::anyhow!("Invalid checksum");
        assert!(!NetworkManager::disconnect_on_unrecognized_command(
            &unknown
        ));
        assert!(NetworkManager::disconnect_on_unrecognized_command(
            &bad_checksum
        ));
    }
}

#[cfg(test)]
mod ban_reload_tests {
    use super::NetworkManager;
    use crate::config::{NodeConfig, StorageConfig};
    use std::net::SocketAddr;

    fn manager(dir: &std::path::Path) -> NetworkManager {
        let mut config = NodeConfig::default();
        let mut storage = StorageConfig::default();
        storage.data_dir = dir.to_string_lossy().into_owned();
        config.storage = Some(storage);
        NetworkManager::with_config(
            "127.0.0.1:0".parse().unwrap(),
            8,
            crate::network::transport::TransportPreference::TCP_ONLY,
            Some(&config),
        )
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_ban_is_present_after_a_new_manager_loads_the_directory() {
        let dir = std::env::temp_dir().join(format!(
            "blvm-f53-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();

        let fresh = manager(&dir);
        assert!(
            fresh.get_banned_peers().is_empty(),
            "a missing ban file starts empty"
        );

        let setban: SocketAddr = "203.0.113.9:8333".parse().unwrap();
        let until = 1_900_000_000u64;
        fresh.ban_peer(setban, until);

        let loaded = manager(&dir);
        let key = SocketAddr::new(setban.ip(), 0);
        let banned = loaded.get_banned_peers();
        assert!(
            banned.iter().any(|(addr, ts)| *addr == key && *ts == until),
            "setban was not reloaded: {banned:?}"
        );

        let violation_dir = dir.join("violation");
        std::fs::create_dir_all(&violation_dir).unwrap();
        let violator: SocketAddr = "198.51.100.8:8333".parse().unwrap();
        let before = crate::utils::current_timestamp();
        let first = manager(&violation_dir);
        let _ = first
            .disconnect_for_protocol_violation(violator, "bad-message", true)
            .await;
        let after = crate::utils::current_timestamp();
        let vkey = SocketAddr::new(violator.ip(), 0);
        let found = manager(&violation_dir)
            .get_banned_peers()
            .into_iter()
            .find(|(addr, _)| *addr == vkey);
        let Some((_, ts)) = found else {
            panic!("protocol-violation ban was not reloaded");
        };
        let day = 24 * 60 * 60;
        assert!(ts >= before + day && ts <= after + day);
    }
}