Merge pull request #5197 from stacks-network/fix/5193-stackerdb-decoherence

Fix/5193 stackerdb decoherence

Author: wileyj

stackslib/src/net/api/getneighbors.rs

@@ -54,6 +54,8 @@ pub struct RPCNeighbor {
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(with = "serde_opt_vec_qci")]
     pub stackerdbs: Option<Vec<QualifiedContractIdentifier>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub age: Option<u64>,
 }
 
 /// Serialize and deserialize `Option<Vec<QualifiedContractIdentifier>>`
@@ -95,6 +97,7 @@ impl RPCNeighbor {
         pkh: Hash160,
         auth: bool,
         stackerdbs: Vec<QualifiedContractIdentifier>,
+        age: Option<u64>,
     ) -> RPCNeighbor {
         RPCNeighbor {
             network_id: nk.network_id,
@@ -104,6 +107,7 @@ impl RPCNeighbor {
             public_key_hash: pkh,
             authenticated: auth,
             stackerdbs: Some(stackerdbs),
+            age,
         }
     }
 }
@@ -138,6 +142,7 @@ impl RPCNeighborsInfo {
                     Hash160::from_node_public_key(&n.public_key),
                     true,
                     stackerdb_contract_ids,
+                    None,
                 )
             })
             .collect();
@@ -164,6 +169,7 @@ impl RPCNeighborsInfo {
                     Hash160::from_node_public_key(&n.public_key),
                     true,
                     stackerdb_contract_ids,
+                    None,
                 )
             })
             .collect();
@@ -185,13 +191,15 @@ impl RPCNeighborsInfo {
                     naddr.public_key_hash,
                     convo.is_authenticated(),
                     convo.get_stackerdb_contract_ids().to_vec(),
+                    Some(convo.age()),
                 ));
             } else {
                 inbound.push(RPCNeighbor::from_neighbor_key_and_pubkh(
                     nk,
                     naddr.public_key_hash,
                     convo.is_authenticated(),
                     convo.get_stackerdb_contract_ids().to_vec(),
+                    Some(convo.age()),
                 ));
             }
         }

stackslib/src/net/chat.rs

@@ -609,6 +609,10 @@ impl ConversationP2P {
         }
     }
 
+    pub fn age(&self) -> u64 {
+        get_epoch_time_secs().saturating_sub(self.instantiated)
+    }
+
     pub fn set_public_key(&mut self, pubkey_opt: Option<Secp256k1PublicKey>) -> () {
         self.connection.set_public_key(pubkey_opt);
     }

stackslib/src/net/inv/epoch2x.rs

@@ -973,6 +973,9 @@ pub struct InvState {
 
     /// What's the last reward cycle we _started_ the inv scan at?
     pub block_sortition_start: u64,
+
+    /// event IDs of connections we established, so they don't get pruned
+    pinned: HashSet<usize>,
 }
 
 impl InvState {
@@ -994,11 +997,13 @@ impl InvState {
             num_inv_syncs: 0,
 
             block_sortition_start: 0,
+            pinned: HashSet::new(),
         }
     }
 
     fn reset_sync_peers(
         &mut self,
+        network: &PeerNetwork,
         peers: HashSet<NeighborKey>,
         bootstrap_peers: &HashSet<NeighborKey>,
         max_neighbors: usize,
@@ -1042,6 +1047,24 @@ impl InvState {
             added,
             &peers
         );
+
+        // if we're still connected to these peers, then keep them pinned
+        self.pinned.clear();
+        for peer in peers.iter() {
+            if let Some(event_id) = network.get_event_id(&peer) {
+                self.pinned.insert(event_id);
+            }
+        }
+    }
+
+    /// Pin a connection
+    pub fn pin_connection(&mut self, event_id: usize) {
+        self.pinned.insert(event_id);
+    }
+
+    /// Get the set of connections this state machine is using
+    pub fn get_pinned_connections(&self) -> &HashSet<usize> {
+        &self.pinned
     }
 
     pub fn get_peer_status(&self, nk: &NeighborKey) -> NodeStatus {
@@ -1801,6 +1824,7 @@ impl PeerNetwork {
     /// Start requesting the next batch of PoX inventories
     fn inv_getpoxinv_begin(
         &mut self,
+        pins: &mut HashSet<usize>,
         sortdb: &SortitionDB,
         nk: &NeighborKey,
         stats: &mut NeighborBlockStats,
@@ -1821,6 +1845,8 @@ impl PeerNetwork {
         };
 
         let payload = StacksMessageType::GetPoxInv(getpoxinv);
+        let event_id_opt = self.get_event_id(&nk);
+
         let message = self.sign_for_neighbor(nk, payload)?;
         let request = self
             .send_neighbor_message(nk, message, request_timeout)
@@ -1830,6 +1856,10 @@ impl PeerNetwork {
             })?;
 
         stats.getpoxinv_begin(request, target_pox_reward_cycle);
+        if let Some(event_id) = event_id_opt {
+            pins.insert(event_id);
+        }
+
         Ok(())
     }
 
@@ -1988,6 +2018,7 @@ impl PeerNetwork {
     /// Start requesting the next batch of block inventories
     fn inv_getblocksinv_begin(
         &mut self,
+        pins: &mut HashSet<usize>,
         sortdb: &SortitionDB,
         nk: &NeighborKey,
         stats: &mut NeighborBlockStats,
@@ -2008,6 +2039,7 @@ impl PeerNetwork {
 
         let num_blocks_expected = getblocksinv.num_blocks;
         let payload = StacksMessageType::GetBlocksInv(getblocksinv);
+        let event_id_opt = self.get_event_id(nk);
         let message = self.sign_for_neighbor(nk, payload)?;
         let request = self
             .send_neighbor_message(nk, message, request_timeout)
@@ -2017,6 +2049,9 @@ impl PeerNetwork {
             })?;
 
         stats.getblocksinv_begin(request, target_block_reward_cycle, num_blocks_expected);
+        if let Some(event_id) = event_id_opt {
+            pins.insert(event_id);
+        }
         Ok(())
     }
 
@@ -2114,6 +2149,7 @@ impl PeerNetwork {
     /// Run a single state-machine to completion
     fn inv_sync_run(
         &mut self,
+        pins: &mut HashSet<usize>,
         sortdb: &SortitionDB,
         nk: &NeighborKey,
         stats: &mut NeighborBlockStats,
@@ -2130,13 +2166,13 @@ impl PeerNetwork {
             debug!("Inv sync state is {:?}", &stats.state);
             let again = match stats.state {
                 InvWorkState::GetPoxInvBegin => self
-                    .inv_getpoxinv_begin(sortdb, nk, stats, request_timeout)
+                    .inv_getpoxinv_begin(pins, sortdb, nk, stats, request_timeout)
                     .and_then(|_| Ok(true))?,
                 InvWorkState::GetPoxInvFinish => {
                     self.inv_getpoxinv_try_finish(sortdb, nk, stats, ibd)?
                 }
                 InvWorkState::GetBlocksInvBegin => self
-                    .inv_getblocksinv_begin(sortdb, nk, stats, request_timeout)
+                    .inv_getblocksinv_begin(pins, sortdb, nk, stats, request_timeout)
                     .and_then(|_| Ok(true))?,
                 InvWorkState::GetBlocksInvFinish => {
                     self.inv_getblocksinv_try_finish(nk, stats, ibd)?
@@ -2231,9 +2267,10 @@ impl PeerNetwork {
     ) -> (bool, bool, Vec<NeighborKey>, Vec<NeighborKey>) {
         PeerNetwork::with_inv_state(self, |network, inv_state| {
             debug!(
-                "{:?}: Inventory state has {} block stats tracked",
+                "{:?}: Inventory state has {} block stats tracked on connections {:?}",
                 &network.local_peer,
-                inv_state.block_stats.len()
+                inv_state.block_stats.len(),
+                inv_state.pinned,
             );
 
             let mut all_done = true;
@@ -2261,6 +2298,7 @@ impl PeerNetwork {
                 return (true, true, vec![], vec![]);
             }
 
+            let mut new_pins = HashSet::new();
             for (nk, stats) in inv_state.block_stats.iter_mut() {
                 debug!(
                     "{:?}: inv state-machine for {:?} is in state {:?}, at PoX {},target={}; blocks {},target={}; status {:?}, done={}",
@@ -2275,7 +2313,7 @@ impl PeerNetwork {
                     stats.done
                 );
                 if !stats.done {
-                    match network.inv_sync_run(sortdb, nk, stats, inv_state.request_timeout, ibd) {
+                    match network.inv_sync_run(&mut new_pins, sortdb, nk, stats, inv_state.request_timeout, ibd) {
                         Ok(d) => d,
                         Err(net_error::StaleView) => {
                             // stop work on this state machine -- it needs to be restarted.
@@ -2341,6 +2379,9 @@ impl PeerNetwork {
                     }
                 }
             }
+            let _ = new_pins
+                .into_iter()
+                .map(|event_id| inv_state.pin_connection(event_id));
 
             if all_done {
                 let mut new_sync_peers = network.get_outbound_sync_peers();
@@ -2450,6 +2491,7 @@ impl PeerNetwork {
                 }
 
                 inv_state.reset_sync_peers(
+                    network,
                     good_sync_peers_set,
                     &bootstrap_peers,
                     network.connection_opts.num_neighbors as usize,

stackslib/src/net/inv/nakamoto.rs

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
 use stacks_common::bitvec::BitVec;
 use stacks_common::types::chainstate::StacksBlockId;
@@ -557,6 +557,10 @@ impl<NC: NeighborComms> NakamotoInvStateMachine<NC> {
         self.comms.reset();
     }
 
+    pub fn get_pinned_connections(&self) -> &HashSet<usize> {
+        self.comms.get_pinned_connections()
+    }
+
     /// Remove state for a particular neighbor
     pub fn del_peer(&mut self, peer: &NeighborAddress) {
         self.inventories.remove(peer);

stackslib/src/net/neighbors/comms.rs

@@ -50,6 +50,8 @@ pub trait NeighborComms {
     fn get_connecting<NK: ToNeighborKey>(&self, network: &PeerNetwork, nk: &NK) -> Option<usize>;
     /// Remove a neighbor from connecting state
     fn remove_connecting<NK: ToNeighborKey>(&mut self, network: &PeerNetwork, nk: &NK);
+    /// Remove a neighbor from connecting state due to an error
+    fn remove_connecting_error<NK: ToNeighborKey>(&mut self, network: &PeerNetwork, nk: &NK);
     /// Mark a neighbor as dead (inactive, unreachable, etc.)
     fn add_dead<NK: ToNeighborKey>(&mut self, network: &PeerNetwork, nk: &NK);
     /// Mark a neighbor as broken (in protocol violation)
@@ -150,7 +152,7 @@ pub trait NeighborComms {
             // is the peer network still working?
             if !network.is_connecting(event_id) {
                 debug!("{:?}: Failed to connect to {:?} (event {} no longer connecting; assumed timed out)", network.get_local_peer(), event_id, &nk);
-                self.remove_connecting(network, &nk);
+                self.remove_connecting_error(network, &nk);
                 return Err(net_error::PeerNotConnected);
             }
 
@@ -518,7 +520,13 @@ impl NeighborComms for PeerNetworkComms {
             .map(|event_ref| *event_ref)
     }
 
+    /// Remove a connecting neighbor because it connected
     fn remove_connecting<NK: ToNeighborKey>(&mut self, network: &PeerNetwork, nk: &NK) {
+        self.connecting.remove(&nk.to_neighbor_key(network));
+    }
+
+    /// Remove a connecting neighbor due to an error.  The connection will be unpinned.
+    fn remove_connecting_error<NK: ToNeighborKey>(&mut self, network: &PeerNetwork, nk: &NK) {
         let event_id_opt = self.connecting.remove(&nk.to_neighbor_key(network));
         if let Some(event_id) = event_id_opt {
             self.unpin_connection(event_id);

stackslib/src/net/p2p.rs

@@ -2399,7 +2399,7 @@ impl PeerNetwork {
     }
 
     /// Prune inbound and outbound connections if we can
-    fn prune_connections(&mut self) -> () {
+    pub(crate) fn prune_connections(&mut self) -> () {
         if cfg!(test) && self.connection_opts.disable_network_prune {
             return;
         }
@@ -2443,6 +2443,22 @@ impl PeerNetwork {
                 }
             }
 
+            // if we're in the middle of epoch2 inv sync, then don't prune any connections it
+            // established
+            if let Some(inv_state) = self.inv_state.as_ref() {
+                if inv_state.get_pinned_connections().contains(event_id) {
+                    safe.insert(*event_id);
+                }
+            }
+
+            // if we're in the middle of nakamoto inv sync, then don't prune any connections it
+            // established
+            if let Some(nakamoto_inv) = self.inv_state_nakamoto.as_ref() {
+                if nakamoto_inv.get_pinned_connections().contains(event_id) {
+                    safe.insert(*event_id);
+                }
+            }
+
             // if we're running stacker DBs, then don't prune any outbound connections it
             // established
             if let Some(stacker_db_syncs) = self.stacker_db_syncs.as_ref() {
@@ -2454,6 +2470,7 @@ impl PeerNetwork {
             }
         }
 
+        debug!("Pinned connections: {:?}", &safe);
         self.prune_frontier(&safe);
     }