refactor(dash-spv): rewrite the peer-to-peer network module

Replaces the old network layer with a self-coordinating PeerNetworkManager
(broker): it owns request de-duplication, pacing, per-peer in-flight sizing,
timeouts, retries and peer hot-swap, so the sync pipelines just declare what
they want. Highlights:

- Restore sync/ from dev and re-wire it to the broker; delete the per-pipeline
  DownloadCoordinator.
- Minimal `NetworkManager` trait + in-memory `MockNetworkManager`; the client is
  generic over the network (`DashSpvClient<W, N, S>`) with the network injected.
- Strict-priority message scheduler: control, then blocks, filters, filter
  headers, block headers.
- Drop peer storage and the peer-reputation system (unused; to be redone).
- Un-Box the `NetworkMessage` variants to stay close to dev.
- Storage kept as on dev (minus the removed peer storage).
- Restore dev's integration tests and the network-dependent unit tests via the mock.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CUb3bkX9C1gBFA65GFsN53

Author: ZocoLini

dash-spv-bench/src/main.rs

@@ -152,9 +152,7 @@ async fn main() -> Result<()> {
     let wallet_probe = wallet.clone();
 
     let handler = Arc::new(BenchEventHandler::new(dashboard.clone()));
-    let network = dash_spv::network::PeerNetworkManager::new(&config)
-        .await
-        .map_err(|e| anyhow!("network new: {e}"))?;
+    let network = dash_spv::network::PeerNetworkManager::new(&config).await;
 
     let client = DashSpvClient::new(
         config,

dash-spv-ffi/src/callbacks.rs

@@ -349,6 +349,7 @@ impl FFISyncEventCallbacks {
                 start_height,
                 end_height,
                 tip_height,
+                ..
             } => {
                 if let Some(cb) = self.on_filter_headers_stored {
                     cb(*start_height, *end_height, *tip_height, self.user_data);
@@ -550,17 +551,13 @@ impl FFINetworkEventCallbacks {
         use dash_spv::network::NetworkEvent;
 
         match event {
-            NetworkEvent::PeerConnected {
-                address,
-            } => {
+            NetworkEvent::PeerConnected(address) => {
                 if let Some(cb) = self.on_peer_connected {
                     let c_addr = CString::new(address.to_string()).unwrap_or_default();
                     cb(c_addr.as_ptr(), self.user_data);
                 }
             }
-            NetworkEvent::PeerDisconnected {
-                address,
-            } => {
+            NetworkEvent::PeerDisconnected(address) => {
                 if let Some(cb) = self.on_peer_disconnected {
                     let c_addr = CString::new(address.to_string()).unwrap_or_default();
                     cb(c_addr.as_ptr(), self.user_data);
@@ -569,10 +566,9 @@ impl FFINetworkEventCallbacks {
             NetworkEvent::PeersUpdated {
                 connected_count,
                 best_height,
-                ..
             } => {
                 if let Some(cb) = self.on_peers_updated {
-                    cb(*connected_count as u32, best_height.unwrap_or(0), self.user_data);
+                    cb(*connected_count, *best_height, self.user_data);
                 }
             }
         }

dash-spv-ffi/src/client.rs

@@ -82,15 +82,15 @@ pub unsafe extern "C" fn dash_spv_ffi_client_new(
 
     let client_result = runtime.block_on(async move {
         // Construct concrete implementations for generics
-        let network = dash_spv::network::PeerNetworkManager::new(&client_config).await;
         let storage = DiskStorageManager::new(&client_config).await;
         let wallet = key_wallet_manager::WalletManager::<
             key_wallet::wallet::managed_wallet_info::ManagedWalletInfo,
         >::new(client_config.network);
         let wallet = std::sync::Arc::new(tokio::sync::RwLock::new(wallet));
 
-        match (network, storage) {
-            (Ok(network), Ok(storage)) => {
+        match storage {
+            Ok(storage) => {
+                let network = dash_spv::network::PeerNetworkManager::new(&client_config).await;
                 DashSpvClient::new(
                     client_config,
                     network,
@@ -100,8 +100,7 @@ pub unsafe extern "C" fn dash_spv_ffi_client_new(
                 )
                 .await
             }
-            (Err(e), _) => Err(e),
-            (_, Err(e)) => Err(dash_spv::SpvError::Storage(e)),
+            Err(e) => Err(dash_spv::SpvError::Storage(e)),
         }
     });
 

dash-spv/Cargo.toml

@@ -10,7 +10,7 @@ rust-version = "1.89"
 
 [dependencies]
 # Core Dash libraries
-dashcore = { path = "../dash", features = ["serde", "core-block-hash-use-x11", "message_verification", "bls", "quorum_validation"] }
+dashcore = { path = "../dash", features = ["serde", "core-block-hash-use-x11", "message_verification", "bls", "quorum_validation", "tokio"] }
 dashcore_hashes = { path = "../hashes" }
 dash-network-seeds = { path = "../dash-network-seeds" }
 key-wallet = { path = "../key-wallet" }
@@ -22,6 +22,7 @@ clap = { version = "4.0", features = ["derive", "env"] }
 
 # Async runtime
 tokio = { version = "1.0", features = ["full"] }
+parking_lot = "0.12"
 tokio-util = "0.7"
 tokio-stream = { version = "0.1", features = ["sync"] }
 async-trait = "0.1"

dash-spv/examples/filter_sync.rs

@@ -1,6 +1,5 @@
 //! BIP157 filter synchronization example.
 
-use dash_spv::network::PeerNetworkManager;
 use dash_spv::storage::DiskStorageManager;
 use dash_spv::{init_console_logging, ClientConfig, DashSpvClient, LevelFilter};
 use dashcore::Address;
@@ -25,18 +24,15 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .with_storage_path("./.tmp/filter-sync-example-storage")
         .without_masternodes(); // Skip masternode sync for this example
 
-    // Create network manager
-    let network_manager = PeerNetworkManager::new(&config).await?;
-
     // Create storage manager
     let storage_manager = DiskStorageManager::new(&config).await?;
 
     // Create wallet manager
     let wallet = Arc::new(RwLock::new(WalletManager::<ManagedWalletInfo>::new(config.network)));
 
     // Create the client
-    let client =
-        DashSpvClient::new(config, network_manager, storage_manager, wallet, vec![]).await?;
+    let network = dash_spv::network::PeerNetworkManager::new(&config).await;
+    let client = DashSpvClient::new(config, network, storage_manager, wallet, vec![]).await?;
 
     println!("Starting synchronization with filter support...");
     println!("Watching address: {:?}", watch_address);

dash-spv/examples/simple_sync.rs

@@ -1,6 +1,5 @@
 //! Simple header synchronization example.
 
-use dash_spv::network::PeerNetworkManager;
 use dash_spv::storage::DiskStorageManager;
 use dash_spv::{init_console_logging, ClientConfig, DashSpvClient, LevelFilter};
 use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
@@ -20,18 +19,15 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .without_filters() // Skip filter sync for this example
         .without_masternodes(); // Skip masternode sync for this example
 
-    // Create network manager
-    let network_manager = PeerNetworkManager::new(&config).await?;
-
     // Create storage manager
     let storage_manager = DiskStorageManager::new(&config).await?;
 
     // Create wallet manager
     let wallet = Arc::new(RwLock::new(WalletManager::<ManagedWalletInfo>::new(config.network)));
 
     // Create the client
-    let client =
-        DashSpvClient::new(config, network_manager, storage_manager, wallet, vec![]).await?;
+    let network = dash_spv::network::PeerNetworkManager::new(&config).await;
+    let client = DashSpvClient::new(config, network, storage_manager, wallet, vec![]).await?;
 
     println!("Starting header synchronization...");
 

dash-spv/examples/spv_with_wallet.rs

@@ -2,7 +2,6 @@
 //!
 //! This example shows how to integrate the SPV client with a wallet manager.
 
-use dash_spv::network::PeerNetworkManager;
 use dash_spv::storage::DiskStorageManager;
 use dash_spv::{ClientConfig, DashSpvClient, LevelFilter};
 use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
@@ -20,18 +19,15 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .with_storage_path("./.tmp/spv-with-wallet-example-storage")
         .with_validation_mode(dash_spv::ValidationMode::Full);
 
-    // Create network manager
-    let network_manager = PeerNetworkManager::new(&config).await?;
-
     // Create storage manager - use disk storage for persistence
     let storage_manager = DiskStorageManager::new(&config).await?;
 
     // Create wallet manager
     let wallet = Arc::new(RwLock::new(WalletManager::<ManagedWalletInfo>::new(config.network)));
 
     // Create the SPV client with all components
-    let client =
-        DashSpvClient::new(config, network_manager, storage_manager, wallet, vec![]).await?;
+    let network = dash_spv::network::PeerNetworkManager::new(&config).await;
+    let client = DashSpvClient::new(config, network, storage_manager, wallet, vec![]).await?;
 
     // The wallet will automatically be notified of:
     // - New blocks via process_block()

dash-spv/src/client/core.rs

@@ -35,8 +35,8 @@ pub(super) type PersistentSyncCoordinator<W> = SyncCoordinator<
 ///
 /// # Generic Design Philosophy
 ///
-/// This struct uses three generic parameters (`W`, `N`, `S`) instead of concrete types or
-/// trait objects. This design choice provides significant benefits for a library:
+/// This struct uses two generic parameters (`W`, `S`) instead of concrete types or
+/// trait objects. This design choice provides significant benefits for a library.
 ///
 /// ## Benefits of Generic Architecture
 ///
@@ -57,8 +57,7 @@ pub(super) type PersistentSyncCoordinator<W> = SyncCoordinator<
 /// - Essential for a reusable library
 ///
 /// ### 4. **Testing Without Mocks** 🧪
-/// - Test implementations (`MockNetworkManager`) are
-///   first-class types, not runtime injections
+/// - Test implementations are first-class types, not runtime injections
 /// - No conditional compilation or feature flags needed for tests
 /// - Type system ensures test and production code are compatible
 ///
@@ -70,8 +69,8 @@ pub(super) type PersistentSyncCoordinator<W> = SyncCoordinator<
 /// ## Type Parameters
 ///
 /// - `W: WalletInterface` - Handles UTXO tracking, address management, transaction processing
-/// - `N: NetworkManager` - Manages peer connections, message routing, network protocol
 /// - `S: StorageManager` - Persistent storage for headers, filters, chain state
+/// - Networking is the concrete `crate::network::PeerNetworkManager`
 /// - Event handlers are stored as `Vec<Arc<dyn EventHandler>>`
 ///
 /// ## Common Configurations
@@ -82,14 +81,6 @@ pub(super) type PersistentSyncCoordinator<W> = SyncCoordinator<
 /// // Production configuration
 /// type StandardSpvClient = DashSpvClient<
 ///     WalletManager,
-///     PeerNetworkManager,
-///     DiskStorageManager,
-/// >;
-///
-/// // Test configuration
-/// type TestSpvClient = DashSpvClient<
-///     WalletManager,
-///     MockNetworkManager,
 ///     DiskStorageManager,
 /// >;
 /// ```
@@ -105,7 +96,7 @@ pub(super) type PersistentSyncCoordinator<W> = SyncCoordinator<
 /// The generic design is an intentional, beneficial architectural choice for a library.
 pub struct DashSpvClient<W: WalletInterface, N: NetworkManager, S: StorageManager> {
     pub(super) config: Arc<RwLock<ClientConfig>>,
-    pub(super) network: Arc<Mutex<N>>,
+    pub(super) network: Arc<N>,
     pub(super) storage: Arc<Mutex<S>>,
     /// External wallet implementation (required)
     pub(super) wallet: Arc<RwLock<W>>,
@@ -114,6 +105,12 @@ pub struct DashSpvClient<W: WalletInterface, N: NetworkManager, S: StorageManage
     /// `true` while running, `false` once a stop is requested. Stored as a
     /// `watch` so a stop is observed immediately rather than polled.
     pub(super) running: Arc<watch::Sender<bool>>,
+    /// Set by `stop()` before it flips `running` false; read by `start()` under
+    /// the `running` watch lock so a stop that races startup is not lost. Without
+    /// it, a `stop()` arriving while `start()` is still connecting would no-op
+    /// (running not yet true), then `start()` would flip running true and the run
+    /// task would sync forever — hanging any `run_handle.await`.
+    pub(super) stop_requested: Arc<std::sync::atomic::AtomicBool>,
     pub(super) event_handlers: Arc<Vec<Arc<dyn super::EventHandler>>>,
 }
 
@@ -127,6 +124,7 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager> Clone for DashSpv
             masternode_engine: self.masternode_engine.clone(),
             sync_coordinator: Arc::clone(&self.sync_coordinator),
             running: Arc::clone(&self.running),
+            stop_requested: Arc::clone(&self.stop_requested),
             event_handlers: Arc::clone(&self.event_handlers),
         }
     }

dash-spv/src/client/event_handler.rs

@@ -326,9 +326,8 @@ mod tests {
         };
         handler.on_sync_event(&event);
         handler.on_network_event(&NetworkEvent::PeersUpdated {
-            connected_count: 0,
-            addresses: vec![],
-            best_height: None,
+            connected_count: 1,
+            best_height: 100,
         });
         handler.on_progress(&SyncProgress::default());
         handler.on_error("test error");
@@ -533,14 +532,8 @@ mod tests {
         );
 
         let addr: SocketAddr = "127.0.0.1:9999".parse().unwrap();
-        tx.send(NetworkEvent::PeerConnected {
-            address: addr,
-        })
-        .unwrap();
-        tx.send(NetworkEvent::PeerDisconnected {
-            address: addr,
-        })
-        .unwrap();
+        tx.send(NetworkEvent::PeerConnected(addr)).unwrap();
+        tx.send(NetworkEvent::PeerDisconnected(addr)).unwrap();
         tokio::time::sleep(std::time::Duration::from_millis(50)).await;
 
         shutdown.cancel();

dash-spv/src/client/events.rs

@@ -4,9 +4,10 @@
 //! - Event receiver management
 //! - Event emission
 
+use crate::network::NetworkManager;
 use tokio::sync::watch;
 
-use crate::network::{NetworkEvent, NetworkManager};
+use crate::network::NetworkEvent;
 use crate::storage::StorageManager;
 use crate::sync::{SyncEvent, SyncProgress};
 use key_wallet_manager::WalletInterface;
@@ -32,6 +33,6 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager> DashSpvClient<W,
 
     /// Subscribe to network events.
     pub(crate) async fn subscribe_network_events(&self) -> broadcast::Receiver<NetworkEvent> {
-        self.network.lock().await.subscribe_network_events()
+        self.network.events()
     }
 }