Merge pull request #2571 from blockstack/feat/events-for-microblocks

Feat: Added event for new microblocks

Author: pavitthrap

src/chainstate/stacks/db/unconfirmed.rs

@@ -41,6 +41,24 @@ use crate::types::chainstate::{StacksBlockHeader, StacksBlockId, StacksMicrobloc
 
 pub type UnconfirmedTxMap = HashMap<Txid, (StacksTransaction, BlockHeaderHash, u16)>;
 
+pub struct ProcessedUnconfirmedState {
+    pub total_burns: u128,
+    pub total_fees: u128,
+    // each element of this vector is a tuple, where each tuple contains a microblock
+    // sequence number, and a vector of transaction receipts for that microblock
+    pub receipts: Vec<(u16, Vec<StacksTransactionReceipt>)>,
+}
+
+impl Default for ProcessedUnconfirmedState {
+    fn default() -> Self {
+        ProcessedUnconfirmedState {
+            total_burns: 0,
+            total_fees: 0,
+            receipts: vec![],
+        }
+    }
+}
+
 pub struct UnconfirmedState {
     pub confirmed_chain_tip: StacksBlockId,
     pub unconfirmed_chain_tip: StacksBlockId,
@@ -136,10 +154,10 @@ impl UnconfirmedState {
         chainstate: &StacksChainState,
         burn_dbconn: &dyn BurnStateDB,
         mblocks: Vec<StacksMicroblock>,
-    ) -> Result<(u128, u128, Vec<StacksTransactionReceipt>), Error> {
+    ) -> Result<ProcessedUnconfirmedState, Error> {
         if self.last_mblock_seq == u16::max_value() {
             // drop them -- nothing to do
-            return Ok((0, 0, vec![]));
+            return Ok(Default::default());
         }
 
         debug!(
@@ -193,7 +211,7 @@ impl UnconfirmedState {
                     &mblock_hash, mblock.header.sequence
                 );
 
-                let (stx_fees, stx_burns, mut receipts) =
+                let (stx_fees, stx_burns, receipts) =
                     match StacksChainState::process_microblocks_transactions(
                         &mut clarity_tx,
                         &vec![mblock.clone()],
@@ -211,7 +229,7 @@ impl UnconfirmedState {
                 total_fees += stx_fees;
                 total_burns += stx_burns;
                 num_new_mblocks += 1;
-                all_receipts.append(&mut receipts);
+                all_receipts.push((seq, receipts));
 
                 last_mblock = Some(mblock_header);
                 last_mblock_seq = seq;
@@ -255,7 +273,11 @@ impl UnconfirmedState {
             self.bytes_so_far = 0;
         }
 
-        Ok((total_fees, total_burns, all_receipts))
+        Ok(ProcessedUnconfirmedState {
+            total_fees,
+            total_burns,
+            receipts: all_receipts,
+        })
     }
 
     /// Load up the Stacks microblock stream to process, composed of only the new microblocks
@@ -280,24 +302,25 @@ impl UnconfirmedState {
     }
 
     /// Update the view of the current confiremd chain tip's unconfirmed microblock state
+    /// Returns ProcessedUnconfirmedState for the microblocks newly added to the unconfirmed state
     pub fn refresh(
         &mut self,
         chainstate: &StacksChainState,
         burn_dbconn: &dyn BurnStateDB,
-    ) -> Result<(u128, u128, Vec<StacksTransactionReceipt>), Error> {
+    ) -> Result<ProcessedUnconfirmedState, Error> {
         assert!(
             !self.readonly,
             "BUG: code tried to write unconfirmed state to a read-only instance"
         );
 
         if self.last_mblock_seq == u16::max_value() {
             // no-op
-            return Ok((0, 0, vec![]));
+            return Ok(Default::default());
         }
 
         match self.load_child_microblocks(chainstate)? {
             Some(microblocks) => self.append_microblocks(chainstate, burn_dbconn, microblocks),
-            None => Ok((0, 0, vec![])),
+            None => Ok(Default::default()),
         }
     }
 
@@ -375,15 +398,15 @@ impl StacksChainState {
         &self,
         burn_dbconn: &dyn BurnStateDB,
         anchored_block_id: StacksBlockId,
-    ) -> Result<(UnconfirmedState, u128, u128, Vec<StacksTransactionReceipt>), Error> {
+    ) -> Result<(UnconfirmedState, ProcessedUnconfirmedState), Error> {
         debug!("Make new unconfirmed state off of {}", &anchored_block_id);
         let mut unconfirmed_state = UnconfirmedState::new(self, anchored_block_id)?;
-        let (fees, burns, receipts) = unconfirmed_state.refresh(self, burn_dbconn)?;
+        let processed_unconfirmed_state = unconfirmed_state.refresh(self, burn_dbconn)?;
         debug!(
             "Made new unconfirmed state off of {} (at {})",
             &anchored_block_id, &unconfirmed_state.unconfirmed_chain_tip
         );
-        Ok((unconfirmed_state, fees, burns, receipts))
+        Ok((unconfirmed_state, processed_unconfirmed_state))
     }
 
     /// Reload the unconfirmed view from a new chain tip.
@@ -395,7 +418,7 @@ impl StacksChainState {
         &mut self,
         burn_dbconn: &dyn BurnStateDB,
         canonical_tip: StacksBlockId,
-    ) -> Result<(u128, u128, Vec<StacksTransactionReceipt>), Error> {
+    ) -> Result<ProcessedUnconfirmedState, Error> {
         debug!("Reload unconfirmed state off of {}", &canonical_tip);
 
         let unconfirmed_state = self.unconfirmed_state.take();
@@ -427,7 +450,7 @@ impl StacksChainState {
             self.drop_unconfirmed_state(unconfirmed_state);
         }
 
-        let (new_unconfirmed_state, fees, burns, receipts) =
+        let (new_unconfirmed_state, processed_unconfirmed_state) =
             self.make_unconfirmed_state(burn_dbconn, canonical_tip)?;
 
         debug!(
@@ -436,19 +459,19 @@ impl StacksChainState {
         );
 
         self.unconfirmed_state = Some(new_unconfirmed_state);
-        Ok((fees, burns, receipts))
+        Ok(processed_unconfirmed_state)
     }
 
     /// Refresh the current unconfirmed chain state
     pub fn refresh_unconfirmed_state(
         &mut self,
         burn_dbconn: &dyn BurnStateDB,
-    ) -> Result<(u128, u128, Vec<StacksTransactionReceipt>), Error> {
+    ) -> Result<ProcessedUnconfirmedState, Error> {
         let mut unconfirmed_state = self.unconfirmed_state.take();
         let res = if let Some(ref mut unconfirmed_state) = unconfirmed_state {
             if !unconfirmed_state.is_readable() {
                 warn!("Unconfirmed state is not readable; it will soon be refreshed");
-                return Ok((0, 0, vec![]));
+                return Ok(Default::default());
             }
 
             debug!(
@@ -465,7 +488,7 @@ impl StacksChainState {
             res
         } else {
             warn!("No unconfirmed state instantiated");
-            Ok((0, 0, vec![]))
+            Ok(Default::default())
         };
         self.unconfirmed_state = unconfirmed_state;
         res

src/net/relay.rs

@@ -53,6 +53,7 @@ use vm::costs::ExecutionCost;
 
 use crate::chainstate::coordinator::BlockEventDispatcher;
 use crate::types::chainstate::{PoxId, SortitionId};
+use chainstate::stacks::db::unconfirmed::ProcessedUnconfirmedState;
 use types::chainstate::BurnchainHeaderHash;
 
 pub type BlocksAvailableMap = HashMap<BurnchainHeaderHash, (u64, ConsensusHash)>;
@@ -85,6 +86,7 @@ pub struct RelayerStats {
 
 pub struct ProcessedNetReceipts {
     pub mempool_txs_added: Vec<StacksTransaction>,
+    pub processed_unconfirmed_state: ProcessedUnconfirmedState,
 }
 
 /// Private trait for keeping track of messages that can be relayed, so we can identify the peers
@@ -750,7 +752,7 @@ impl Relayer {
         Ok((new_blocks, bad_neighbors))
     }
 
-    /// Prerocess all downloaded, confirmed microblock streams.
+    /// Preprocess all downloaded, confirmed microblock streams.
     /// Does not fail on invalid blocks; just logs a warning.
     /// Returns the consensus hashes for the sortitions that elected the stacks anchored blocks that produced these streams.
     fn preprocess_downloaded_microblocks(
@@ -1105,7 +1107,7 @@ impl Relayer {
     pub fn setup_unconfirmed_state(
         chainstate: &mut StacksChainState,
         sortdb: &SortitionDB,
-    ) -> Result<(), Error> {
+    ) -> Result<ProcessedUnconfirmedState, Error> {
         let (canonical_consensus_hash, canonical_block_hash) =
             SortitionDB::get_canonical_stacks_chain_tip_hash(sortdb.conn())?;
         let canonical_tip = StacksBlockHeader::make_index_block_hash(
@@ -1117,8 +1119,10 @@ impl Relayer {
             "Reload unconfirmed state off of {}/{}",
             &canonical_consensus_hash, &canonical_block_hash
         );
-        chainstate.reload_unconfirmed_state(&sortdb.index_conn(), canonical_tip)?;
-        Ok(())
+        let processed_unconfirmed_state =
+            chainstate.reload_unconfirmed_state(&sortdb.index_conn(), canonical_tip)?;
+
+        Ok(processed_unconfirmed_state)
     }
 
     /// Set up unconfirmed chain state in a read-only fashion
@@ -1142,16 +1146,23 @@ impl Relayer {
         Ok(())
     }
 
-    pub fn refresh_unconfirmed(chainstate: &mut StacksChainState, sortdb: &mut SortitionDB) {
-        if let Err(e) = Relayer::setup_unconfirmed_state(chainstate, sortdb) {
-            if let net_error::ChainstateError(ref err_msg) = e {
-                if err_msg == "Stacks chainstate error: NoSuchBlockError" {
-                    trace!("Failed to instantiate unconfirmed state: {:?}", &e);
+    pub fn refresh_unconfirmed(
+        chainstate: &mut StacksChainState,
+        sortdb: &mut SortitionDB,
+    ) -> ProcessedUnconfirmedState {
+        match Relayer::setup_unconfirmed_state(chainstate, sortdb) {
+            Ok(processed_unconfirmed_state) => processed_unconfirmed_state,
+            Err(e) => {
+                if let net_error::ChainstateError(ref err_msg) = e {
+                    if err_msg == "Stacks chainstate error: NoSuchBlockError" {
+                        trace!("Failed to instantiate unconfirmed state: {:?}", &e);
+                    } else {
+                        warn!("Failed to instantiate unconfirmed state: {:?}", &e);
+                    }
                 } else {
                     warn!("Failed to instantiate unconfirmed state: {:?}", &e);
                 }
-            } else {
-                warn!("Failed to instantiate unconfirmed state: {:?}", &e);
+                Default::default()
             }
         }
     }
@@ -1272,13 +1283,18 @@ impl Relayer {
             }
         }
 
-        let receipts = ProcessedNetReceipts { mempool_txs_added };
+        let mut processed_unconfirmed_state = Default::default();
 
         // finally, refresh the unconfirmed chainstate, if need be
         if network_result.has_microblocks() {
-            Relayer::refresh_unconfirmed(chainstate, sortdb);
+            processed_unconfirmed_state = Relayer::refresh_unconfirmed(chainstate, sortdb);
         }
 
+        let receipts = ProcessedNetReceipts {
+            mempool_txs_added,
+            processed_unconfirmed_state,
+        };
+
         Ok(receipts)
     }
 }

testnet/stacks-node/src/config.rs

@@ -1308,6 +1308,7 @@ pub enum EventKeyType {
     AssetEvent(AssetIdentifier),
     STXEvent,
     MemPoolTransactions,
+    Microblocks,
     AnyEvent,
     BurnchainBlocks,
 }
@@ -1330,6 +1331,10 @@ impl EventKeyType {
             return Some(EventKeyType::BurnchainBlocks);
         }
 
+        if raw_key == "microblocks" {
+            return Some(EventKeyType::Microblocks);
+        }
+
         let comps: Vec<_> = raw_key.split("::").collect();
         if comps.len() == 1 {
             let split: Vec<_> = comps[0].split(".").collect();