see time on ci based on number of stackslib mutants

stackslib/src/chainstate/nakamoto/coordinator/mod.rs

@@ -681,6 +681,7 @@ impl<
     /// with Some(pox-anchor-block-hash) until the reward cycle info is processed in the sortition
     /// DB.
     pub fn handle_new_nakamoto_stacks_block(&mut self) -> Result<Option<BlockHeaderHash>, Error> {
+        debug!("Handle new Nakamoto block");
         let canonical_sortition_tip = self.canonical_sortition_tip.clone().expect(
             "FAIL: processing a new Stacks block, but don't have a canonical sortition tip",
         );

stackslib/src/chainstate/nakamoto/coordinator/tests.rs

@@ -404,7 +404,7 @@ fn replay_reward_cycle(
             &sortdb,
             &mut sort_handle,
             &mut node.chainstate,
-            block.clone(),
+            &block,
             None,
         )
         .unwrap_or(false);

stackslib/src/chainstate/nakamoto/mod.rs

@@ -1835,7 +1835,7 @@ impl NakamotoChainState {
     /// Insert a Nakamoto block into the staging blocks DB
     pub(crate) fn store_block(
         staging_db_tx: &NakamotoStagingBlocksTx,
-        block: NakamotoBlock,
+        block: &NakamotoBlock,
         burn_attachable: bool,
     ) -> Result<(), ChainstateError> {
         let block_id = block.block_id();
@@ -1894,7 +1894,7 @@ impl NakamotoChainState {
     /// Returns true if we stored the block; false if not.
     pub fn accept_block(
         config: &ChainstateConfig,
-        block: NakamotoBlock,
+        block: &NakamotoBlock,
         db_handle: &mut SortitionHandleConn,
         staging_db_tx: &NakamotoStagingBlocksTx,
         headers_conn: &Connection,
@@ -1927,14 +1927,14 @@ impl NakamotoChainState {
 
         // it's okay if this fails because we might not have the parent block yet.  It will be
         // checked on `::append_block()`
-        let expected_burn_opt = Self::get_expected_burns(db_handle, headers_conn, &block)?;
+        let expected_burn_opt = Self::get_expected_burns(db_handle, headers_conn, block)?;
 
         // this block must be consistent with its miner's leader-key and block-commit, and must
         // contain only transactions that are valid in this epoch.
         if let Err(e) = Self::validate_nakamoto_block_burnchain(
             db_handle,
             expected_burn_opt,
-            &block,
+            block,
             config.mainnet,
             config.chain_id,
         ) {
@@ -1958,9 +1958,8 @@ impl NakamotoChainState {
         // same sortition history as `db_handle` (and thus it must be burn_attachable)
         let burn_attachable = true;
 
-        let _block_id = block.block_id();
         Self::store_block(staging_db_tx, block, burn_attachable)?;
-        test_debug!("Stored Nakamoto block {}", &_block_id);
+        test_debug!("Stored Nakamoto block {}", &block.block_id());
         Ok(true)
     }
 

stackslib/src/chainstate/nakamoto/tests/mod.rs

@@ -1069,7 +1069,7 @@ pub fn test_load_store_update_nakamoto_blocks() {
             300,
         )
         .unwrap();
-        NakamotoChainState::store_block(&staging_tx, nakamoto_block.clone(), false).unwrap();
+        NakamotoChainState::store_block(&staging_tx, &nakamoto_block, false).unwrap();
 
         // tenure has one block
         assert_eq!(
@@ -1102,7 +1102,7 @@ pub fn test_load_store_update_nakamoto_blocks() {
         )
         .unwrap();
 
-        NakamotoChainState::store_block(&staging_tx, nakamoto_block_2.clone(), false).unwrap();
+        NakamotoChainState::store_block(&staging_tx, &nakamoto_block_2, false).unwrap();
 
         // tenure has two blocks
         assert_eq!(
@@ -1123,7 +1123,7 @@ pub fn test_load_store_update_nakamoto_blocks() {
         );
 
         // store, but do not process, a block
-        NakamotoChainState::store_block(&staging_tx, nakamoto_block_3.clone(), false).unwrap();
+        NakamotoChainState::store_block(&staging_tx, &nakamoto_block_3, false).unwrap();
 
         staging_tx.commit().unwrap();
         tx.commit().unwrap();

stackslib/src/chainstate/nakamoto/tests/node.rs

@@ -632,7 +632,7 @@ impl TestStacksNode {
                 sortdb,
                 &mut sort_handle,
                 chainstate,
-                nakamoto_block.clone(),
+                &nakamoto_block,
                 None,
             ) {
                 Ok(accepted) => accepted,
@@ -1159,7 +1159,7 @@ impl<'a> TestPeer<'a> {
                 &sortdb,
                 &mut sort_handle,
                 &mut node.chainstate,
-                block,
+                &block,
                 None,
             )
             .unwrap();

stackslib/src/net/chat.rs

@@ -128,6 +128,8 @@ pub struct NeighborStats {
     pub transaction_push_rx_counts: VecDeque<(u64, u64)>,
     /// (timestamp, num bytes)
     pub stackerdb_push_rx_counts: VecDeque<(u64, u64)>,
+    /// (timestamp, num bytes)
+    pub nakamoto_block_push_rx_counts: VecDeque<(u64, u64)>,
     pub relayed_messages: HashMap<NeighborAddress, RelayStats>,
 }
 
@@ -152,6 +154,7 @@ impl NeighborStats {
             microblocks_push_rx_counts: VecDeque::new(),
             transaction_push_rx_counts: VecDeque::new(),
             stackerdb_push_rx_counts: VecDeque::new(),
+            nakamoto_block_push_rx_counts: VecDeque::new(),
             relayed_messages: HashMap::new(),
         }
     }
@@ -214,6 +217,17 @@ impl NeighborStats {
         }
     }
 
+    /// Record that we recently received a Nakamoto blcok push of the given size.
+    /// Keeps track of the last `NUM_BANDWIDTH_POINTS` such events, so we can estimate the current
+    /// bandwidth consumed by Nakamoto block pushes
+    pub fn add_nakamoto_block_push(&mut self, message_size: u64) -> () {
+        self.nakamoto_block_push_rx_counts
+            .push_back((get_epoch_time_secs(), message_size));
+        while self.nakamoto_block_push_rx_counts.len() > NUM_BANDWIDTH_POINTS {
+            self.nakamoto_block_push_rx_counts.pop_front();
+        }
+    }
+
     pub fn add_relayer(&mut self, addr: &NeighborAddress, num_bytes: u64) -> () {
         if let Some(stats) = self.relayed_messages.get_mut(addr) {
             stats.num_messages += 1;
@@ -298,6 +312,14 @@ impl NeighborStats {
         NeighborStats::get_bandwidth(&self.stackerdb_push_rx_counts, BANDWIDTH_POINT_LIFETIME)
     }
 
+    /// Get a peer's total nakamoto block bandwidth usage
+    pub fn get_nakamoto_block_push_bandwidth(&self) -> f64 {
+        NeighborStats::get_bandwidth(
+            &self.nakamoto_block_push_rx_counts,
+            BANDWIDTH_POINT_LIFETIME,
+        )
+    }
+
     /// Determine how many of a particular message this peer has received
     pub fn get_message_recv_count(&self, msg_id: StacksMessageID) -> u64 {
         *(self.msg_rx_counts.get(&msg_id).unwrap_or(&0))
@@ -2217,6 +2239,45 @@ impl ConversationP2P {
         Ok(None)
     }
 
+    /// Validate a pushed Nakamoto block list.
+    /// Update bandwidth accounting, but forward the blocks along if we can accept them.
+    /// Possibly return a reply handle for a NACK if we throttle the remote sender
+    fn validate_nakamoto_block_push(
+        &mut self,
+        network: &PeerNetwork,
+        preamble: &Preamble,
+        relayers: Vec<RelayData>,
+    ) -> Result<Option<ReplyHandleP2P>, net_error> {
+        assert!(preamble.payload_len > 1); // don't count 1-byte type prefix
+
+        let local_peer = network.get_local_peer();
+        let chain_view = network.get_chain_view();
+
+        if !self.process_relayers(local_peer, preamble, &relayers) {
+            warn!(
+                "Drop pushed Nakamoto blocks -- invalid relayers {:?}",
+                &relayers
+            );
+            self.stats.msgs_err += 1;
+            return Err(net_error::InvalidMessage);
+        }
+
+        self.stats
+            .add_nakamoto_block_push((preamble.payload_len as u64) - 1);
+
+        if self.connection.options.max_nakamoto_block_push_bandwidth > 0
+            && self.stats.get_nakamoto_block_push_bandwidth()
+                > (self.connection.options.max_nakamoto_block_push_bandwidth as f64)
+        {
+            debug!("Neighbor {:?} exceeded max Nakamoto block push bandwidth of {} bytes/sec (currently at {})", &self.to_neighbor_key(), self.connection.options.max_nakamoto_block_push_bandwidth, self.stats.get_nakamoto_block_push_bandwidth());
+            return self
+                .reply_nack(local_peer, chain_view, preamble, NackErrorCodes::Throttled)
+                .and_then(|handle| Ok(Some(handle)));
+        }
+
+        Ok(None)
+    }
+
     /// Handle an inbound authenticated p2p data-plane message.
     /// Return the message if not handled
     fn handle_data_message(
@@ -2305,6 +2366,21 @@ impl ConversationP2P {
                     }
                 }
             }
+            StacksMessageType::NakamotoBlocks(_) => {
+                // not handled here, but do some accounting -- we can't receive too many
+                // Nakamoto blocks per second
+                match self.validate_nakamoto_block_push(
+                    network,
+                    &msg.preamble,
+                    msg.relayers.clone(),
+                )? {
+                    Some(handle) => Ok(handle),
+                    None => {
+                        // will forward upstream
+                        return Ok(Some(msg));
+                    }
+                }
+            }
             _ => {
                 // all else will forward upstream
                 return Ok(Some(msg));
@@ -6603,6 +6679,54 @@ mod test {
         assert_eq!(bw_stats.get_stackerdb_push_bandwidth(), 110.0);
     }
 
+    #[test]
+    fn test_neighbor_stats_nakamoto_block_push_bandwidth() {
+        let mut stats = NeighborStats::new(false);
+
+        assert_eq!(stats.get_nakamoto_block_push_bandwidth(), 0.0);
+
+        stats.add_nakamoto_block_push(100);
+        assert_eq!(stats.get_nakamoto_block_push_bandwidth(), 0.0);
+
+        // this should all happen in one second
+        let bw_stats = loop {
+            let mut bw_stats = stats.clone();
+            let start = get_epoch_time_secs();
+
+            for _ in 0..(NUM_BANDWIDTH_POINTS - 1) {
+                bw_stats.add_nakamoto_block_push(100);
+            }
+
+            let end = get_epoch_time_secs();
+            if end == start {
+                break bw_stats;
+            }
+        };
+
+        assert_eq!(
+            bw_stats.get_nakamoto_block_push_bandwidth(),
+            (NUM_BANDWIDTH_POINTS as f64) * 100.0
+        );
+
+        // space some out; make sure it takes 11 seconds
+        let bw_stats = loop {
+            let mut bw_stats = NeighborStats::new(false);
+            let start = get_epoch_time_secs();
+            for _ in 0..11 {
+                bw_stats.add_nakamoto_block_push(100);
+                sleep_ms(1001);
+            }
+
+            let end = get_epoch_time_secs();
+            if end == start + 11 {
+                break bw_stats;
+            }
+        };
+
+        // 100 bytes/sec
+        assert_eq!(bw_stats.get_nakamoto_block_push_bandwidth(), 110.0);
+    }
+
     #[test]
     fn test_sign_relay_forward_message() {
         let conn_opts = ConnectionOptions::default();

stackslib/src/net/codec.rs

@@ -41,6 +41,7 @@ use stacks_common::util::secp256k1::{
 
 use crate::burnchains::{BurnchainView, PrivateKey, PublicKey};
 use crate::chainstate::burn::ConsensusHash;
+use crate::chainstate::nakamoto::NakamotoBlock;
 use crate::chainstate::stacks::{
     StacksBlock, StacksMicroblock, StacksPublicKey, StacksTransaction, MAX_BLOCK_LEN,
 };
@@ -353,6 +354,37 @@ impl NakamotoInvData {
     }
 }
 
+impl StacksMessageCodec for NakamotoBlocksData {
+    #[cfg_attr(test, mutants::skip)]
+    fn consensus_serialize<W: Write>(&self, fd: &mut W) -> Result<(), codec_error> {
+        write_next(fd, &self.blocks)?;
+        Ok(())
+    }
+
+    fn consensus_deserialize<R: Read>(fd: &mut R) -> Result<Self, codec_error> {
+        let blocks: Vec<NakamotoBlock> = {
+            // loose upper-bound
+            let mut bound_read = BoundReader::from_reader(fd, MAX_MESSAGE_LEN as u64);
+            read_next_at_most::<_, NakamotoBlock>(&mut bound_read, NAKAMOTO_BLOCKS_PUSHED_MAX)
+        }?;
+
+        // only valid if there are no dups
+        let mut present = HashSet::new();
+        for block in blocks.iter() {
+            if present.contains(&block.block_id()) {
+                // no dups allowed
+                return Err(codec_error::DeserializeError(
+                    "Invalid NakamotoBlocksData: duplicate block".to_string(),
+                ));
+            }
+
+            present.insert(block.block_id());
+        }
+
+        Ok(NakamotoBlocksData { blocks })
+    }
+}
+
 impl StacksMessageCodec for GetPoxInv {
     fn consensus_serialize<W: Write>(&self, fd: &mut W) -> Result<(), codec_error> {
         write_next(fd, &self.consensus_hash)?;
@@ -930,6 +962,7 @@ impl StacksMessageType {
             StacksMessageType::StackerDBPushChunk(ref _m) => StacksMessageID::StackerDBPushChunk,
             StacksMessageType::GetNakamotoInv(ref _m) => StacksMessageID::GetNakamotoInv,
             StacksMessageType::NakamotoInv(ref _m) => StacksMessageID::NakamotoInv,
+            StacksMessageType::NakamotoBlocks(ref _m) => StacksMessageID::NakamotoBlocks,
         }
     }
 
@@ -964,6 +997,7 @@ impl StacksMessageType {
             StacksMessageType::StackerDBPushChunk(ref _m) => "StackerDBPushChunk",
             StacksMessageType::GetNakamotoInv(ref _m) => "GetNakamotoInv",
             StacksMessageType::NakamotoInv(ref _m) => "NakamotoInv",
+            StacksMessageType::NakamotoBlocks(ref _m) => "NakamotoBlocks",
         }
     }
 
@@ -1071,6 +1105,15 @@ impl StacksMessageType {
             StacksMessageType::NakamotoInv(ref m) => {
                 format!("NakamotoInv({:?})", &m.tenures)
             }
+            StacksMessageType::NakamotoBlocks(ref m) => {
+                format!(
+                    "NakamotoBlocks({:?})",
+                    m.blocks
+                        .iter()
+                        .map(|block| block.block_id())
+                        .collect::<Vec<_>>()
+                )
+            }
         }
     }
 }
@@ -1122,6 +1165,7 @@ impl StacksMessageCodec for StacksMessageID {
             }
             x if x == StacksMessageID::GetNakamotoInv as u8 => StacksMessageID::GetNakamotoInv,
             x if x == StacksMessageID::NakamotoInv as u8 => StacksMessageID::NakamotoInv,
+            x if x == StacksMessageID::NakamotoBlocks as u8 => StacksMessageID::NakamotoBlocks,
             _ => {
                 return Err(codec_error::DeserializeError(
                     "Unknown message ID".to_string(),
@@ -1166,6 +1210,7 @@ impl StacksMessageCodec for StacksMessageType {
             StacksMessageType::StackerDBPushChunk(ref m) => write_next(fd, m)?,
             StacksMessageType::GetNakamotoInv(ref m) => write_next(fd, m)?,
             StacksMessageType::NakamotoInv(ref m) => write_next(fd, m)?,
+            StacksMessageType::NakamotoBlocks(ref m) => write_next(fd, m)?,
         }
         Ok(())
     }
@@ -1276,6 +1321,10 @@ impl StacksMessageCodec for StacksMessageType {
                 let m: NakamotoInvData = read_next(fd)?;
                 StacksMessageType::NakamotoInv(m)
             }
+            StacksMessageID::NakamotoBlocks => {
+                let m: NakamotoBlocksData = read_next(fd)?;
+                StacksMessageType::NakamotoBlocks(m)
+            }
             StacksMessageID::Reserved => {
                 return Err(codec_error::DeserializeError(
                     "Unsupported message ID 'reserved'".to_string(),