fix(comms+dht): mark peers as online inbound connection,join (#5741)

Description
---
- mark peers as online inbound connection and on join msg
- fix DHT peer query logic

Motivation and Context
---
Peers were not being marked as online whenever we have proof that they
are online, specifically in the inbound connection and join msg.

The peer query to exclude peers if we have attempted them but never
succeeded was incorrect and filtered out most peers. This was changed to

```rust
 // we have tried to connect to this peer, and we have never made a successful attempt at connection
if peer.last_connect_attempt().is_some() && peer.last_seen().is_none() {
    return false;
}
```

How Has This Been Tested?
---
Manually, node reconnects quickly to peers on startup

What process can a PR reviewer use to test or verify this change?
---

<!-- 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 -->

comms/core/src/connection_manager/common.rs

@@ -206,6 +206,9 @@ pub(super) fn create_or_update_peer_from_validated_peer_identity(
                     peer_identity_claim: peer_identity.claim.clone(),
                 });
 
+            peer.addresses
+                .mark_all_addresses_as_last_seen_now(&peer_identity.claim.addresses);
+
             peer.features = peer_identity.claim.features;
             peer.supported_protocols = peer_identity.metadata.supported_protocols.clone();
             peer.user_agent = peer_identity.metadata.user_agent.clone();

comms/core/src/connectivity/manager.rs

@@ -473,7 +473,7 @@ impl ConnectivityManagerActor {
         }
     }
 
-    fn mark_peer_succeeded(&mut self, node_id: NodeId) {
+    fn mark_connection_success(&mut self, node_id: NodeId) {
         let entry = self.get_connection_stat_mut(node_id);
         entry.set_connection_success();
     }
@@ -635,7 +635,7 @@ impl ConnectivityManagerActor {
         match (old_status, new_status) {
             (_, Connected) => match self.pool.get_connection(&node_id).cloned() {
                 Some(conn) => {
-                    self.mark_peer_succeeded(node_id.clone());
+                    self.mark_connection_success(conn.peer_node_id().clone());
                     self.publish_event(ConnectivityEvent::PeerConnected(conn.into()));
                 },
                 None => unreachable!(

comms/core/src/net_address/mutliaddresses_with_stats.rs

@@ -174,6 +174,21 @@ impl MultiaddressesWithStats {
         }
     }
 
+    /// Mark all addresses as seen. Returns true if all addresses are contained in this instance, otherwise false
+    pub fn mark_all_addresses_as_last_seen_now(&mut self, addresses: &[Multiaddr]) -> bool {
+        let mut all_exist = true;
+        for address in addresses {
+            match self.find_address_mut(address) {
+                Some(addr) => {
+                    addr.mark_last_seen_now().mark_last_attempted_now();
+                },
+                None => all_exist = false,
+            }
+        }
+        self.sort_addresses();
+        all_exist
+    }
+
     /// Mark that a connection could not be established with the specified net address
     ///
     /// Returns true if the address is contained in this instance, otherwise false

comms/dht/src/connectivity/mod.rs

@@ -489,10 +489,6 @@ impl DhtConnectivity {
     }
 
     async fn handle_new_peer_connected(&mut self, conn: PeerConnection) -> Result<(), DhtConnectivityError> {
-        // We can only mark the peer as seen if we know which address we are were about to connect to (Outbound).
-        if let Some(addr) = conn.known_address() {
-            self.peer_manager.mark_last_seen(conn.peer_node_id(), addr).await?;
-        }
         if conn.peer_features().is_client() {
             debug!(
                 target: LOG_TARGET,
@@ -749,6 +745,7 @@ impl DhtConnectivity {
         let peer_manager = &self.peer_manager;
         let node_id = self.node_identity.node_id();
         let connected = self.connected_peers_iter().collect::<Vec<_>>();
+
         // Fetch to all n nearest neighbour Communication Nodes
         // which are eligible for connection.
         // Currently that means:
@@ -779,7 +776,7 @@ impl DhtConnectivity {
                     return false;
                 }
                 // we have tried to connect to this peer, and we have never made a successful attempt at connection
-                if peer.offline_since().is_none() && peer.last_connect_attempt().is_some() {
+                if peer.last_connect_attempt().is_some() && peer.last_seen().is_none() {
                     return false;
                 }
 

comms/dht/src/peer_validator.rs

@@ -113,6 +113,7 @@ impl<'a> PeerValidator<'a> {
             peer.update_addresses(&claim.addresses, &PeerAddressSource::FromDiscovery {
                 peer_identity_claim: claim.clone(),
             });
+            peer.addresses.mark_all_addresses_as_last_seen_now(&claim.addresses);
         }
 
         Ok(peer)