fix(spv): fix shutdown deadlock and ordering

dash-spv/CLAUDE.md

@@ -168,6 +168,9 @@ Use domain-specific error types:
 - **Built on**: `dashcore` library with Dash-specific features enabled
 - **Async runtime**: Tokio with full feature set
 
+### Logging Convention
+The project uses the `tracing` framework for structured logging. New code should use `tracing::` macros (`tracing::info!`, `tracing::debug!`, `tracing::warn!`, `tracing::error!`).
+
 ## Key Implementation Details
 
 ### Storage Architecture

dash-spv/src/client/events.rs

@@ -26,6 +26,13 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager> DashSpvClient<W,
     }
 
     /// Subscribe to sync events from the sync coordinator.
+    ///
+    /// **Important:** Consumers MUST monitor for [`SyncEvent::ManagerExited`] events.
+    /// When a manager exits due to a fatal network error, it will not restart
+    /// automatically. The consumer should restart the SPV client to recover.
+    ///
+    /// The overall sync state (via [`Self::subscribe_progress`] / [`Self::progress`])
+    /// will transition to [`SyncState::Error`] when any manager exits fatally.
     pub fn subscribe_sync_events(&self) -> broadcast::Receiver<SyncEvent> {
         self.sync_coordinator.subscribe_events()
     }

dash-spv/src/client/lifecycle.rs

@@ -193,6 +193,12 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager> DashSpvClient<W,
             }
         }
 
+        // Shut down sync coordinator: signals cancellation and waits for manager
+        // tasks to drain before we tear down the network and storage layers.
+        if let Err(e) = self.sync_coordinator.shutdown().await {
+            tracing::warn!("Error shutting down sync coordinator: {}", e);
+        }
+
         // Disconnect from network
         self.network.disconnect().await?;
 

dash-spv/src/error.rs

@@ -210,10 +210,26 @@ pub enum SyncError {
     #[error("Timeout error: {0}")]
     Timeout(String),
 
-    /// Network-related errors (e.g., connection failures, protocol errors)
+    /// Network-related errors (e.g., connection failures, protocol errors).
+    ///
+    /// For transient network issues, use this variant or [`SyncError::Timeout`].
+    /// For fatal, unrecoverable conditions (e.g., request channel closed),
+    /// use [`SyncError::FatalNetwork`] instead.
     #[error("Network error: {0}")]
     Network(String),
 
+    /// Fatal network condition that causes the manager task loop to exit immediately.
+    ///
+    /// Unlike [`SyncError::Network`], which is logged and allows the loop to continue,
+    /// this variant signals an unrecoverable failure (e.g., the request channel is closed)
+    /// and causes the manager to stop permanently.
+    ///
+    /// **Consumer responsibility:** When this error occurs, [`SyncEvent::ManagerExited`]
+    /// is emitted and [`SyncState::Error`] is set on the affected manager's progress.
+    /// The consumer must restart the SPV client to recover the affected sync phase.
+    #[error("Fatal network error (manager exiting): {0}")]
+    FatalNetwork(String),
+
     /// Validation errors for data received during sync (e.g., invalid headers, invalid proofs)
     /// Use this for data validation errors, not state errors
     #[error("Validation error: {0}")]
@@ -240,7 +256,7 @@ impl SyncError {
             SyncError::Timeout(_) => "timeout",
             SyncError::Validation(_) => "validation",
             SyncError::MissingDependency(_) => "dependency",
-            SyncError::Network(_) => "network",
+            SyncError::Network(_) | SyncError::FatalNetwork(_) => "network",
             SyncError::Storage(_) => "storage",
             SyncError::Headers2DecompressionFailed(_) => "headers2",
             SyncError::MasternodeSyncFailed(_) => "masternode",

dash-spv/src/network/manager.rs

@@ -240,31 +240,46 @@ impl PeerNetworkManager {
         let message_dispatcher = self.message_dispatcher.clone();
         let network_event_sender = self.network_event_sender.clone();
 
-        // Spawn connection task
-        let mut tasks = self.tasks.lock().await;
+        // Spawn connection task — use select to avoid blocking on the lock during shutdown
+        let mut tasks = tokio::select! {
+            guard = self.tasks.lock() => guard,
+            _ = self.shutdown_token.cancelled() => {
+                self.pool.remove_peer(&addr).await;
+                return;
+            }
+        };
         tasks.spawn(async move {
-            log::debug!("Attempting to connect to {}", addr);
+            tracing::debug!("Attempting to connect to {}", addr);
 
-            match Peer::connect(addr, CONNECTION_TIMEOUT.as_secs(), network).await {
+            let connect_result = tokio::select! {
+                result = Peer::connect(addr, CONNECTION_TIMEOUT.as_secs(), network) => result,
+                _ = shutdown_token.cancelled() => {
+                    tracing::debug!("Connection to {} cancelled by shutdown", addr);
+                    pool.remove_peer(&addr).await;
+                    return;
+                }
+            };
+
+            match connect_result {
                 Ok(mut peer) => {
                     // Perform handshake
                     let mut handshake_manager =
                         HandshakeManager::new(network, mempool_strategy, user_agent);
                     match handshake_manager.perform_handshake(&mut peer).await {
                         Ok(_) => {
-                            log::info!("Successfully connected to {}", addr);
+                            tracing::info!("Successfully connected to {}", addr);
 
                             // Request addresses from the peer for discovery
                             if let Err(e) = peer.send_message(NetworkMessage::GetAddr).await {
-                                log::warn!("Failed to send GetAddr to {}: {}", addr, e);
+                                tracing::warn!("Failed to send GetAddr to {}: {}", addr, e);
                             }
 
                             // Record successful connection
                             reputation_manager.record_successful_connection(addr).await;
 
                             // Add to pool
                             if let Err(e) = pool.add_peer(addr, peer).await {
-                                log::error!("Failed to add peer to pool: {}", e);
+                                tracing::error!("Failed to add peer to pool: {}", e);
                                 return;
                             }
 
@@ -784,6 +799,7 @@ impl PeerNetworkManager {
                         if !this.pool.is_connected(addr).await
                             && !this.pool.is_connecting(addr).await
                         {
+                            if this.shutdown_token.is_cancelled() { break; }
                             log::info!("Reconnecting to exclusive peer: {}", addr);
                             this.connect_to_peer(*addr).await;
                         }
@@ -803,6 +819,7 @@ impl PeerNetworkManager {
                             if !this.pool.is_connected(&addr).await
                                 && !this.pool.is_connecting(&addr).await
                             {
+                                if this.shutdown_token.is_cancelled() { break; }
                                 this.connect_to_peer(addr).await;
                                 attempted += 1;
                                 if attempted >= needed {
@@ -1232,11 +1249,16 @@ impl PeerNetworkManager {
             log::warn!("Failed to save reputation data on shutdown: {}", e);
         }
 
-        // Wait for tasks to complete
-        let mut tasks = self.tasks.lock().await;
+        // Take tasks out of the mutex so we don't hold the lock while draining.
+        // This prevents a deadlock where a task (e.g. maintenance loop) tries to
+        // acquire self.tasks via connect_to_peer() while we hold the lock here.
+        let mut tasks = {
+            let mut guard = self.tasks.lock().await;
+            std::mem::take(&mut *guard)
+        };
         while let Some(result) = tasks.join_next().await {
             if let Err(e) = result {
-                log::error!("Task join error: {}", e);
+                tracing::error!("Task join error: {}", e);
             }
         }
 

dash-spv/src/sync/events.rs

@@ -109,7 +109,9 @@ pub enum SyncEvent {
         height: u32,
     },
 
-    /// A manager encountered a recoverable error.
+    /// A manager encountered a non-fatal error (logged, loop continues).
+    ///
+    /// For fatal errors that cause the manager to exit, see [`SyncEvent::ManagerExited`].
     ///
     /// Emitted by: Any manager
     /// Consumed by: Coordinator (for logging/monitoring)
@@ -120,6 +122,20 @@ pub enum SyncEvent {
         error: String,
     },
 
+    /// A manager task has exited due to a fatal error and will not restart.
+    ///
+    /// This is a **critical event** that consumers MUST handle. The affected sync
+    /// phase has stopped permanently. The client should be restarted to recover.
+    ///
+    /// Emitted by: Any manager (on `SyncError::FatalNetwork`)
+    /// Consumer action required: Restart the SPV client or take corrective action.
+    ManagerExited {
+        /// Which manager exited
+        manager: ManagerIdentifier,
+        /// Reason for exit
+        error: String,
+    },
+
     /// ChainLock received and processed.
     ///
     /// Emitted by: `ChainLockManager`
@@ -218,6 +234,13 @@ impl SyncEvent {
             } => {
                 format!("ManagerError({}, {})", manager, error)
             }
+            SyncEvent::ManagerExited {
+                manager,
+                error,
+                ..
+            } => {
+                format!("ManagerExited({}, {})", manager, error)
+            }
             SyncEvent::ChainLockReceived {
                 chain_lock,
                 validated,