fix: peer retention and connections (#7123)

Description
---
- Fixed peer retention in the database by improving delete all stale
peers logic.
- Improved error responses when peers could not be found in the
database.
- Improved DHT connection pools sync with peer database.

Motivation and Context
---
DHT neighbour and random pools were not in sync with actual connections
and peers in the peer db.

Issue - Why `PeerManagerError(PeerNotFoundError)`
```
2025-05-28 04:49:55.101009900 [comms::dht::connectivity] ERROR Error refreshing neighbour peer pool: PeerManagerError(PeerNotFoundError)
2025-05-28 04:49:55.101120100 [comms::connectivity::manager] TRACE Request (14743808475136314793): GetAllowList(Sender { inner: Some(Inner { state: State { is_complete: false, is_closed: false, is_rx_task_set: true, is_tx_task_set: false } }) })
2025-05-28 04:49:55.101160300 [comms::connectivity::manager] TRACE Request (14743808475136314793) done
2025-05-28 04:49:55.104823200 [comms::dht::connectivity] ERROR Error refreshing random peer pool: PeerManagerError(PeerNotFoundError)
```
Issue  -  Why `0 connected` but `active DHT connections: 10/12`
```
2025-05-28 04:49:55.104929100 [comms::dht::connectivity] DEBUG DHT connectivity status: neighbour pool: 8/8 (0 connected), random pool: 4/4 (0 connected, last refreshed 12777s ago), active DHT connections: 10/12
```
Issue - Why `Inbound pipeline returned an error: 'The requested peer
does not exist'`
```
2025-05-28 10:42:21.447513700 [comms::pipeline::inbound] WARN  Inbound pipeline returned an error: 'The requested peer does not exist'
```
How Has This Been Tested?
---
- Adapted unit test for improved delete all stale peers logic.
- System-level testing [**TBD**]

What process can a PR reviewer use to test or verify this change?
---
- Code review.
- System-level testing

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
<!-- BREAKING CHANGE: Description what the user should do, e.g. delete a
database, resync the chain -->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Improved error messages now include detailed peer identifiers for
easier troubleshooting.
- Introduced distinct soft and hard peer deletion methods for better
peer lifecycle management.
- Support for duplicate peer addresses has been added, allowing more
flexible network setups.

- **Bug Fixes**
- Enhanced error handling to skip peers with inaccessible addresses
without failing operations.

- **Database Migration**
- Removed the unique constraint on peer addresses to support multiple
peers sharing the same address.

- **Tests**
- Added and updated tests to validate peer deletion behaviors and
duplicate address support.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: hansieodendaal

applications/minotari_node/src/commands/command/test_peer_liveness.rs

@@ -86,7 +86,7 @@ impl HandleCommand<ArgsTestPeerLiveness> for CommandContext {
         let address_clone = args.address.clone();
 
         // Remove the peer from the peer manager (not the peer db)
-        let _res = peer_manager.delete_peer(&node_id).await;
+        let _res = peer_manager.soft_delete_peer(&node_id).await;
 
         // Create a new peer with the given address, if the peer exists, this will merge the given address
         let peer = Peer::new(

comms/core/src/connection_manager/manager.rs

@@ -510,7 +510,7 @@ where
                 warn!(target: LOG_TARGET, "Peer not found for dial");
                 if let Some(reply) = reply {
                     let _result = reply.send(Err(ConnectionManagerError::PeerManagerError(
-                        PeerManagerError::PeerNotFoundError,
+                        PeerManagerError::peer_not_found(&node_id),
                     )));
                 }
             },

comms/core/src/connection_manager/tests/manager.rs

@@ -78,7 +78,7 @@ async fn connect_to_nonexistent_peer() {
     rt_handle.spawn(connection_manager.run());
 
     let err = requester.dial_peer(NodeId::default()).await.unwrap_err();
-    unpack_enum!(ConnectionManagerError::PeerManagerError(PeerManagerError::PeerNotFoundError) = err);
+    unpack_enum!(ConnectionManagerError::PeerManagerError(PeerManagerError::PeerNotFound(_)) = err);
 
     shutdown.trigger();
 }

comms/core/src/connectivity/manager.rs

@@ -467,7 +467,12 @@ impl ConnectivityManagerActor {
 
     async fn delete_stale_peers_from_db(&mut self, task_id: u64) {
         let start = Instant::now();
-        match tokio::time::timeout(STALE_PEER_DELETE_TIMEOUT, self.peer_manager.delete_all_stale_peers()).await {
+        match tokio::time::timeout(
+            STALE_PEER_DELETE_TIMEOUT,
+            self.peer_manager.hard_delete_all_stale_peers(),
+        )
+        .await
+        {
             Ok(res) => match res {
                 Ok(deleted) => {
                     let len = deleted.len();
@@ -713,7 +718,7 @@ impl ConnectivityManagerActor {
                             .map(|d| format!("{}s ago", d.as_secs()))
                             .unwrap_or_else(|| "Never".to_string()),
                     );
-                    self.peer_manager.delete_peer(node_id).await?;
+                    self.peer_manager.soft_delete_peer(node_id).await?;
                 }
             }
         }
@@ -1108,7 +1113,7 @@ impl ConnectivityManagerActor {
             format_duration(duration),
             reason
         );
-        self.peer_manager.ban_peer_by_node_id(node_id, duration, reason).await?;
+        let ban_result = self.peer_manager.ban_peer_by_node_id(node_id, duration, reason).await;
 
         #[cfg(feature = "metrics")]
         super::metrics::banned_peers_counter().inc();
@@ -1123,6 +1128,7 @@ impl ConnectivityManagerActor {
                 "Disconnected banned peer {}. The peer connection status is {}", node_id, status
             );
         }
+        ban_result?;
         Ok(())
     }
 

comms/core/src/peer_manager/error.rs

@@ -22,9 +22,11 @@
 
 // use std::sync::PoisonError;
 
+use std::fmt;
+
 use multiaddr::Multiaddr;
 use tari_common_sqlite::error::StorageError;
-use tari_utilities::hex::HexError;
+use tari_utilities::hex::{Hex, HexError};
 use thiserror::Error;
 use tokio::task::JoinError;
 
@@ -33,8 +35,8 @@ use crate::peer_manager::NodeId;
 /// Error type for [PeerManager](super::PeerManager).
 #[derive(Debug, Error, Clone)]
 pub enum PeerManagerError {
-    #[error("The requested peer does not exist")]
-    PeerNotFoundError,
+    #[error("The requested peer does not exist: {0}")]
+    PeerNotFound(DisplayVec),
     #[error("DB Data inconsistency: {0}")]
     DataInconsistency(String),
     #[error("The peer has been banned")]
@@ -68,7 +70,6 @@ pub enum PeerManagerError {
     #[error("Process error: `{0}`")]
     ProcessError(String),
 }
-
 impl From<JoinError> for PeerManagerError {
     fn from(err: JoinError) -> Self {
         PeerManagerError::JoinError(err.to_string())
@@ -81,13 +82,6 @@ impl From<StorageError> for PeerManagerError {
     }
 }
 
-impl PeerManagerError {
-    /// Returns true if this error indicates that the peer is not found, otherwise false
-    pub fn is_peer_not_found(&self) -> bool {
-        matches!(self, PeerManagerError::PeerNotFoundError)
-    }
-}
-
 impl From<HexError> for PeerManagerError {
     fn from(value: HexError) -> Self {
         PeerManagerError::HexError(value.to_string())
@@ -99,3 +93,55 @@ impl From<std::io::Error> for PeerManagerError {
         PeerManagerError::StorageError(value.to_string())
     }
 }
+
+/// Display helper struct for a vector of strings.
+#[derive(Debug, Error, Clone)]
+pub struct DisplayVec(Vec<NodeId>);
+
+impl fmt::Display for DisplayVec {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "[{}]",
+            self.0.iter().map(|v| v.to_hex()).collect::<Vec<_>>().join(", ")
+        )
+    }
+}
+
+impl PeerManagerError {
+    pub fn peer_not_found(peer: &NodeId) -> Self {
+        PeerManagerError::PeerNotFound(DisplayVec(vec![peer.clone()]))
+    }
+
+    pub fn peers_not_found<T>(peers: T) -> Self
+    where T: AsRef<[NodeId]> {
+        PeerManagerError::PeerNotFound(DisplayVec(peers.as_ref().to_vec()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_peer_not_found() {
+        let peers = [
+            NodeId::from_hex("abb1556d806c2ff042f433ca0a").unwrap(),
+            NodeId::from_hex("ba9ab662a6d974c5a607562326").unwrap(),
+            NodeId::from_hex("97676095b1901327bdc36e8cb6").unwrap(),
+        ];
+
+        let error = PeerManagerError::peers_not_found(&peers);
+        assert_eq!(
+            error.to_string(),
+            "The requested peer does not exist: [abb1556d806c2ff042f433ca0a, ba9ab662a6d974c5a607562326, \
+             97676095b1901327bdc36e8cb6]"
+        );
+
+        let error = PeerManagerError::peer_not_found(&peers[0]);
+        assert_eq!(
+            error.to_string(),
+            "The requested peer does not exist: [abb1556d806c2ff042f433ca0a]"
+        );
+    }
+}

comms/core/src/peer_manager/manager.rs

@@ -80,9 +80,9 @@ impl PeerManager {
         Ok(peer_id)
     }
 
-    /// The peer with the specified public_key will be removed from the PeerManager
-    pub async fn delete_peer(&self, node_id: &NodeId) -> Result<(), PeerManagerError> {
-        self.peer_storage_sql.delete_peer(node_id)?;
+    /// The peer with the specified node id will be soft deleted (marked as deleted)
+    pub async fn soft_delete_peer(&self, node_id: &NodeId) -> Result<(), PeerManagerError> {
+        self.peer_storage_sql.soft_delete_peer(node_id)?;
         #[cfg(feature = "metrics")]
         {
             let count = self.count().await;
@@ -93,8 +93,8 @@ impl PeerManager {
     }
 
     /// Delete all stale peers, removing them from the database and returning their node_ids
-    pub async fn delete_all_stale_peers(&self) -> Result<Vec<NodeId>, PeerManagerError> {
-        let deleted_peers = self.peer_storage_sql.delete_all_stale_peers()?;
+    pub async fn hard_delete_all_stale_peers(&self) -> Result<Vec<NodeId>, PeerManagerError> {
+        let deleted_peers = self.peer_storage_sql.hard_delete_all_stale_peers()?;
         Ok(deleted_peers)
     }
 
@@ -184,7 +184,7 @@ impl PeerManager {
     pub async fn direct_identity_node_id(&self, node_id: &NodeId) -> Result<Option<Peer>, PeerManagerError> {
         match self.peer_storage_sql.direct_identity_node_id(node_id) {
             Ok(peer) => Ok(Some(peer)),
-            Err(PeerManagerError::PeerNotFoundError) | Err(PeerManagerError::BannedPeer) => Ok(None),
+            Err(PeerManagerError::PeerNotFound(_)) | Err(PeerManagerError::BannedPeer) => Ok(None),
             Err(err) => Err(err),
         }
     }
@@ -196,7 +196,7 @@ impl PeerManager {
     ) -> Result<Option<Peer>, PeerManagerError> {
         match self.peer_storage_sql.direct_identity_public_key(public_key) {
             Ok(peer) => Ok(Some(peer)),
-            Err(PeerManagerError::PeerNotFoundError) | Err(PeerManagerError::BannedPeer) => Ok(None),
+            Err(PeerManagerError::PeerNotFound(_)) | Err(PeerManagerError::BannedPeer) => Ok(None),
             Err(err) => Err(err),
         }
     }
@@ -296,7 +296,7 @@ impl PeerManager {
         let peer = self
             .find_by_node_id(node_id)
             .await?
-            .ok_or(PeerManagerError::PeerNotFoundError)?;
+            .ok_or(PeerManagerError::peer_not_found(node_id))?;
         Ok(peer.features)
     }
 
@@ -308,7 +308,7 @@ impl PeerManager {
         let peer = self
             .find_by_node_id(node_id)
             .await?
-            .ok_or(PeerManagerError::PeerNotFoundError)?;
+            .ok_or(PeerManagerError::peer_not_found(node_id))?;
         Ok(peer.addresses)
     }
 
@@ -324,7 +324,7 @@ impl PeerManager {
         }
         let peers = self.get_peers_by_node_ids(node_ids).await?;
         if peers.is_empty() {
-            return Err(PeerManagerError::PeerNotFoundError);
+            return Err(PeerManagerError::peers_not_found(node_ids));
         }
         let results = peers.into_iter().map(|p| (p.node_id, p.addresses)).collect::<Vec<_>>();
         Ok(results)

comms/core/src/peer_manager/or_not_found.rs

@@ -20,20 +20,20 @@
 //  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use crate::peer_manager::PeerManagerError;
+use crate::peer_manager::{NodeId, PeerManagerError};
 
 /// Extension trait for Result<Option<T>, PeerManagerError>.
 pub trait OrNotFound {
     type Value;
     type Error;
-    fn or_not_found(self) -> Result<Self::Value, Self::Error>;
+    fn or_not_found(self, node_id: &NodeId) -> Result<Self::Value, Self::Error>;
 }
 
 impl<T> OrNotFound for Result<Option<T>, PeerManagerError> {
     type Error = PeerManagerError;
     type Value = T;
 
-    fn or_not_found(self) -> Result<Self::Value, Self::Error> {
-        self.and_then(|val| val.ok_or(PeerManagerError::PeerNotFoundError))
+    fn or_not_found(self, node_id: &NodeId) -> Result<Self::Value, Self::Error> {
+        self.and_then(|val| val.ok_or(PeerManagerError::peer_not_found(node_id)))
     }
 }

comms/core/src/peer_manager/peer_storage_sql.rs

@@ -99,9 +99,9 @@ impl PeerStorageSql {
             .add_or_update_online_peer(pubkey, node_id, addresses, peer_features, source)?)
     }
 
-    /// The peer with the specified public_key will be removed from the PeerManager
-    pub fn delete_peer(&self, node_id: &NodeId) -> Result<(), PeerManagerError> {
-        self.peer_db.set_deleted_at(node_id)?;
+    /// The peer with the specified node id will be soft deleted (marked as deleted)
+    pub fn soft_delete_peer(&self, node_id: &NodeId) -> Result<(), PeerManagerError> {
+        self.peer_db.soft_delete_peer(node_id)?;
         Ok(())
     }
 
@@ -159,7 +159,7 @@ impl PeerStorageSql {
     pub fn direct_identity_node_id(&self, node_id: &NodeId) -> Result<Peer, PeerManagerError> {
         let peer = self
             .get_peer_by_node_id(node_id)?
-            .ok_or(PeerManagerError::PeerNotFoundError)?;
+            .ok_or(PeerManagerError::peer_not_found(node_id))?;
 
         if peer.is_banned() {
             Err(PeerManagerError::BannedPeer)
@@ -172,7 +172,7 @@ impl PeerStorageSql {
     pub fn direct_identity_public_key(&self, public_key: &CommsPublicKey) -> Result<Peer, PeerManagerError> {
         let peer = self
             .find_by_public_key(public_key)?
-            .ok_or(PeerManagerError::PeerNotFoundError)?;
+            .ok_or(PeerManagerError::peer_not_found(&NodeId::from_public_key(public_key)))?;
 
         if peer.is_banned() {
             Err(PeerManagerError::BannedPeer)
@@ -244,10 +244,10 @@ impl PeerStorageSql {
     }
 
     /// Delete all stale peers, removing them from the database and returning their node_ids
-    pub fn delete_all_stale_peers(&self) -> Result<Vec<NodeId>, PeerManagerError> {
+    pub fn hard_delete_all_stale_peers(&self) -> Result<Vec<NodeId>, PeerManagerError> {
         Ok(self
             .peer_db
-            .delete_all_stale_peers(STALE_PEER_THRESHOLD_DURATION, MAX_NEIGHBOUR_WALLET_PEER_COUNT)?)
+            .hard_delete_all_stale_peers(STALE_PEER_THRESHOLD_DURATION, MAX_NEIGHBOUR_WALLET_PEER_COUNT)?)
     }
 
     /// Compile a random list of communication node peers of size _n_ that are not banned or offline
@@ -330,7 +330,7 @@ impl PeerStorageSql {
         let node_id = NodeId::from_key(public_key);
         self.peer_db
             .set_banned(&node_id, duration, reason)?
-            .ok_or(PeerManagerError::PeerNotFoundError)
+            .ok_or(PeerManagerError::peer_not_found(&NodeId::from_public_key(public_key)))
     }
 
     /// Ban the peer for the given duration
@@ -342,13 +342,13 @@ impl PeerStorageSql {
     ) -> Result<NodeId, PeerManagerError> {
         self.peer_db
             .set_banned(node_id, duration, reason)?
-            .ok_or(PeerManagerError::PeerNotFoundError)
+            .ok_or(PeerManagerError::peer_not_found(node_id))
     }
 
     pub fn is_peer_banned(&self, node_id: &NodeId) -> Result<bool, PeerManagerError> {
         let peer = self
             .get_peer_by_node_id(node_id)?
-            .ok_or(PeerManagerError::PeerNotFoundError)?;
+            .ok_or(PeerManagerError::peer_not_found(node_id))?;
         Ok(peer.is_banned())
     }
 
@@ -597,7 +597,7 @@ mod test {
         peer_storage.find_by_public_key(&peer3.public_key).unwrap().unwrap();
 
         // Test delete of border case peer
-        assert!(peer_storage.delete_peer(&peer3.node_id).is_ok());
+        assert!(peer_storage.soft_delete_peer(&peer3.node_id).is_ok());
 
         // It is a logical delete, so there should still be 3 peers in the db
         assert_eq!(peer_storage.peer_db.size(), 3);