Merge pull request #2021 from blockstack/feat/events-1930

Event dispatcher updates

Author: Aaron Blankstein

.github/workflows/stacks-blockchain.yml

@@ -48,6 +48,7 @@ jobs:
   # Run net-tests
   nettest:
     runs-on: ubuntu-latest
+    if: github.event_name == 'push' && startsWith(github.event.ref, 'refs/tags/v')
     steps:
       - uses: actions/checkout@v2
       - name: Run network relay tests

docs/event-dispatcher.md

@@ -67,10 +67,41 @@ Example:
       "tx_index": 1,
       "txid": "0x738e4d44636023efa08374033428e44eca490582bd39a6e61f3b6cf749b4214c"
     }
-  ]
+   ],
+   "matured_miner_rewards": [
+    {
+      "recipient": "ST31DA6FTSJX2WGTZ69SFY11BH51NZMB0ZZ239N96",
+      "coinbase_amount": "1000",
+      "tx_fees_anchored_shared": "800",
+      "tx_fees_anchored_exclusive": "0",
+      "tx_fees_streamed_confirmed": "0",
+      "from_stacks_block_hash": "0xf5d4ce0efe1d42c963d615ce57f0d014f263a985175e4ece766eceff10e0a358",
+      "from_index_block_hash": "0x329efcbcc6daf5ac3f264522e0df50eddb5be85df6ee8a9fc2384c54274d7afc",
+    }
+   ]
 }
 ```
 
+### `POST /new_burn_block`
+
+This payload includes information about burn blocks as their sortitions are processed.
+In the event of PoX forks, a `new_burn_block` event may be triggered for a burn block
+previously processed.
+
+Example:
+
+```json
+{
+  "burn_block_hash": "0x4eaabcd105865e471f697eff5dd5bd85d47ecb5a26a3379d74fae0ae87c40904",
+  "reward_recipients": [
+    {
+      "recipient": "1C56LYirKa3PFXFsvhSESgDy2acEHVAEt6",
+      "amount": 5000
+    }
+  ],
+  "burn_amount": 12000
+}
+```
 
 ### `POST /new_mempool_tx`
 

src/chainstate/coordinator/mod.rs

@@ -14,22 +14,23 @@
 // 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::VecDeque;
+use std::collections::{HashMap, VecDeque};
 use std::convert::{TryFrom, TryInto};
 use std::time::Duration;
 
 use burnchains::{
     db::{BurnchainBlockData, BurnchainDB},
-    Burnchain, BurnchainBlockHeader, BurnchainHeaderHash, Error as BurnchainError, Txid,
+    Address, Burnchain, BurnchainBlockHeader, BurnchainHeaderHash, Error as BurnchainError, Txid,
 };
 use chainstate::burn::{
     db::sortdb::{PoxId, SortitionDB, SortitionId},
     operations::leader_block_commit::RewardSetInfo,
+    operations::BlockstackOperationType,
     BlockHeaderHash, BlockSnapshot, ConsensusHash,
 };
 use chainstate::stacks::{
     boot::STACKS_BOOT_CODE_CONTRACT_ADDRESS,
-    db::{ClarityTx, StacksChainState, StacksHeaderInfo},
+    db::{accounts::MinerReward, ClarityTx, MinerRewardInfo, StacksChainState, StacksHeaderInfo},
     events::StacksTransactionReceipt,
     Error as ChainstateError, StacksAddress, StacksBlock, StacksBlockHeader, StacksBlockId,
 };
@@ -108,6 +109,19 @@ pub trait BlockEventDispatcher {
         receipts: Vec<StacksTransactionReceipt>,
         parent: &StacksBlockId,
         winner_txid: Txid,
+        matured_rewards: Vec<MinerReward>,
+        matured_rewards_info: Option<MinerRewardInfo>,
+    );
+
+    /// called whenever a burn block is about to be
+    ///  processed for sortition. note, in the event
+    ///  of PoX forks, this will be called _multiple_
+    ///  times for the same burnchain header hash.
+    fn announce_burn_block(
+        &self,
+        burn_block: &BurnchainHeaderHash,
+        rewards: Vec<(StacksAddress, u64)>,
+        burns: u64,
     );
 
     fn dispatch_boot_receipts(&mut self, receipts: Vec<StacksTransactionReceipt>);
@@ -426,6 +440,33 @@ pub fn get_reward_cycle_info<U: RewardSetProvider>(
     }
 }
 
+fn dispatcher_announce_burn_ops<T: BlockEventDispatcher>(
+    dispatcher: &T,
+    burn_header: &BurnchainBlockHeader,
+    ops: &[BlockstackOperationType],
+) {
+    let mut reward_recipients: HashMap<_, u64> = HashMap::new();
+    let mut burn_amt = 0;
+    for op in ops.iter() {
+        if let BlockstackOperationType::LeaderBlockCommit(commit) = op {
+            let amt_per_address = commit.burn_fee / (commit.commit_outs.len() as u64);
+            for addr in commit.commit_outs.iter() {
+                if addr.is_burn() {
+                    burn_amt += amt_per_address;
+                } else {
+                    if let Some(prior_amt) = reward_recipients.get_mut(addr) {
+                        *prior_amt += amt_per_address;
+                    } else {
+                        reward_recipients.insert(addr.clone(), amt_per_address);
+                    }
+                }
+            }
+        }
+    }
+    let reward_recipients_vec = reward_recipients.into_iter().collect();
+    dispatcher.announce_burn_block(&burn_header.block_hash, reward_recipients_vec, burn_amt);
+}
+
 impl<'a, T: BlockEventDispatcher, N: CoordinatorNotices, U: RewardSetProvider>
     ChainsCoordinator<'a, T, N, U>
 {
@@ -475,6 +516,10 @@ impl<'a, T: BlockEventDispatcher, N: CoordinatorNotices, U: RewardSetProvider>
         for unprocessed_block in sortitions_to_process.drain(..) {
             let BurnchainBlockData { header, ops } = unprocessed_block;
 
+            if let Some(dispatcher) = self.dispatcher {
+                dispatcher_announce_burn_ops(dispatcher, &header, &ops);
+            }
+
             let sortition_tip_snapshot = SortitionDB::get_block_snapshot(
                 self.sortition_db.conn(),
                 &canonical_sortition_tip,
@@ -621,6 +666,8 @@ impl<'a, T: BlockEventDispatcher, N: CoordinatorNotices, U: RewardSetProvider>
                             block_receipt.tx_receipts,
                             &parent,
                             winner_txid,
+                            block_receipt.matured_rewards,
+                            block_receipt.matured_rewards_info,
                         );
                     }
 

src/chainstate/coordinator/tests.rs

@@ -24,7 +24,9 @@ use util::hash::Hash160;
 use burnchains::{db::*, *};
 use chainstate::burn::db::sortdb::{PoxId, SortitionDB, SortitionId};
 use chainstate::burn::*;
-use chainstate::stacks::db::{ClarityTx, StacksChainState, StacksHeaderInfo};
+use chainstate::stacks::db::{
+    accounts::MinerReward, ClarityTx, StacksChainState, StacksHeaderInfo,
+};
 use chainstate::stacks::index::TrieHash;
 use core;
 use monitoring::increment_stx_blocks_processed_counter;
@@ -229,14 +231,24 @@ impl BlockEventDispatcher for NullEventDispatcher {
         _receipts: Vec<StacksTransactionReceipt>,
         _parent: &StacksBlockId,
         _winner_txid: Txid,
+        _rewards: Vec<MinerReward>,
+        _rewards_info: Option<MinerRewardInfo>,
     ) {
         assert!(
             false,
             "We should never try to announce to the null dispatcher"
         );
     }
 
-    fn dispatch_boot_receipts(&mut self, receipts: Vec<StacksTransactionReceipt>) {}
+    fn announce_burn_block(
+        &self,
+        _burn_block: &BurnchainHeaderHash,
+        _rewards: Vec<(StacksAddress, u64)>,
+        _burns: u64,
+    ) {
+    }
+
+    fn dispatch_boot_receipts(&mut self, _receipts: Vec<StacksTransactionReceipt>) {}
 }
 
 pub fn make_coordinator<'a>(

src/chainstate/stacks/address.rs

@@ -36,6 +36,7 @@ use util::hash::HASH160_ENCODED_SIZE;
 use burnchains::Address;
 use burnchains::PublicKey;
 
+use address::b58;
 use address::c32::c32_address_decode;
 
 use deps::bitcoin::blockdata::opcodes::All as BtcOp;
@@ -167,6 +168,16 @@ impl StacksAddress {
         }
     }
 
+    pub fn to_b58(self) -> String {
+        let StacksAddress { version, bytes } = self;
+        let btc_version = to_b52_version_byte(version)
+            // fallback to version
+            .unwrap_or(version);
+        let mut all_bytes = vec![btc_version];
+        all_bytes.extend(bytes.0.iter());
+        b58::check_encode_slice(&all_bytes)
+    }
+
     /// Convert to PrincipalData::Standard(StandardPrincipalData)
     pub fn to_account_principal(&self) -> PrincipalData {
         PrincipalData::Standard(StandardPrincipalData(

src/chainstate/stacks/db/accounts.rs

@@ -116,30 +116,6 @@ impl FromRow<MinerPaymentSchedule> for MinerPaymentSchedule {
 }
 
 impl MinerReward {
-    pub fn empty_miner(address: &StacksAddress) -> MinerReward {
-        MinerReward {
-            address: address.clone(),
-            coinbase: 0,
-            tx_fees_anchored_shared: 0,
-            tx_fees_anchored_exclusive: 0,
-            tx_fees_streamed_produced: 0,
-            tx_fees_streamed_confirmed: 0,
-            vtxindex: 0,
-        }
-    }
-
-    pub fn empty_user(address: &StacksAddress, vtxindex: u32) -> MinerReward {
-        MinerReward {
-            address: address.clone(),
-            coinbase: 0,
-            tx_fees_anchored_shared: 0,
-            tx_fees_anchored_exclusive: 0,
-            tx_fees_streamed_produced: 0,
-            tx_fees_streamed_confirmed: 0,
-            vtxindex: vtxindex,
-        }
-    }
-
     pub fn total(&self) -> u128 {
         self.coinbase
             + self.tx_fees_anchored_shared
@@ -624,11 +600,11 @@ impl StacksChainState {
     /// scheduled miner payments to the cache.  Miner payments are immutable once written, and are
     /// keyed to the index block hash (which is globally unique), so once cached, no invalidation
     /// should be necessary.
-    pub fn find_mature_miner_rewards<'a>(
-        tx: &mut StacksDBTx<'a>,
+    pub fn find_mature_miner_rewards(
+        tx: &mut StacksDBTx,
         tip: &StacksHeaderInfo,
         mut cache: Option<&mut MinerPaymentCache>,
-    ) -> Result<Option<Vec<MinerReward>>, Error> {
+    ) -> Result<Option<(Vec<MinerReward>, MinerRewardInfo)>, Error> {
         if tip.block_height <= MINER_REWARD_MATURITY + MINER_REWARD_WINDOW {
             // no mature rewards exist
             return Ok(None);
@@ -736,6 +712,11 @@ impl StacksChainState {
             assert_eq!(*scheduled_payments.last().unwrap(), rs_miners_after);
         }
 
+        let matured_rewards_info = MinerRewardInfo {
+            from_stacks_block_hash: matured_miners.0.block_hash.clone(),
+            from_block_consensus_hash: matured_miners.0.consensus_hash.clone(),
+        };
+
         let mut rewards = vec![];
         let miner_reward =
             StacksChainState::calculate_miner_reward(&matured_miners.0, &scheduled_payments);
@@ -745,7 +726,7 @@ impl StacksChainState {
             let reward = StacksChainState::calculate_miner_reward(user_reward, &scheduled_payments);
             rewards.push(reward);
         }
-        Ok(Some(rewards))
+        Ok(Some((rewards, matured_rewards_info)))
     }
 }
 
@@ -1162,31 +1143,33 @@ mod test {
         let legacy_rewards = old_find_mature_miner_rewards(&mut tx, &parent_tip)
             .unwrap()
             .unwrap();
+
         assert_eq!(legacy_rewards, expected_rewards);
 
-        let rewards_opt =
-            StacksChainState::find_mature_miner_rewards(&mut tx, &parent_tip, None).unwrap();
-        assert!(rewards_opt.is_some());
+        let (rewards, _) = StacksChainState::find_mature_miner_rewards(&mut tx, &parent_tip, None)
+            .unwrap()
+            .unwrap();
 
-        let rewards = rewards_opt.unwrap();
         assert_eq!(rewards.len(), 2);
         assert_eq!(rewards, expected_rewards);
 
-        let rewards_cached =
+        let (rewards_cached, _) =
             StacksChainState::find_mature_miner_rewards(&mut tx, &parent_tip, Some(&mut cache))
                 .unwrap()
                 .unwrap();
+
         assert_eq!(rewards_cached, rewards);
         assert_eq!(rewards_cached, expected_rewards);
 
         let mut empty_cache = MinerPaymentCache::new();
-        let rewards_cached = StacksChainState::find_mature_miner_rewards(
+        let (rewards_cached, _) = StacksChainState::find_mature_miner_rewards(
             &mut tx,
             &parent_tip,
             Some(&mut empty_cache),
         )
         .unwrap()
         .unwrap();
+
         assert_eq!(rewards_cached, rewards);
         assert_eq!(rewards_cached, expected_rewards);
     }

src/chainstate/stacks/db/blocks.rs

@@ -3753,9 +3753,9 @@ impl StacksChainState {
     /// block's transactions.  Finally, it returns the execution costs for the microblock stream
     /// and for the anchored block (separately).
     /// Returns None if we're out of blocks to process.
-    fn append_block<'a>(
-        chainstate_tx: &mut ChainstateTx<'a>,
-        clarity_instance: &'a mut ClarityInstance,
+    fn append_block(
+        chainstate_tx: &mut ChainstateTx,
+        clarity_instance: &mut ClarityInstance,
         burn_dbconn: &dyn BurnStateDB,
         parent_chain_tip: &StacksHeaderInfo,
         chain_tip_consensus_hash: &ConsensusHash,
@@ -3778,17 +3778,19 @@ impl StacksChainState {
         let next_block_height = block.header.total_work.work;
 
         // find matured miner rewards, so we can grant them within the Clarity DB tx.
-        let matured_miner_rewards_opt = {
-            StacksChainState::find_mature_miner_rewards(
+        let (matured_rewards, matured_rewards_info) =
+            match StacksChainState::find_mature_miner_rewards(
                 &mut chainstate_tx.headers_tx,
                 parent_chain_tip,
                 Some(chainstate_tx.miner_payment_cache),
-            )?
-        };
+            )? {
+                Some((rewards, rewards_info)) => (rewards, Some(rewards_info)),
+                None => (vec![], None),
+            };
 
         let (
             scheduled_miner_reward,
-            txs_receipts,
+            tx_receipts,
             microblock_execution_cost,
             block_execution_cost,
             total_liquid_ustx,
@@ -3896,16 +3898,12 @@ impl StacksChainState {
                 .expect("BUG: microblock cost + block cost < block cost");
 
             // grant matured miner rewards
-            let new_liquid_miner_ustx =
-                if let Some(mature_miner_rewards) = matured_miner_rewards_opt {
-                    // grant in order by miner, then users
-                    StacksChainState::process_matured_miner_rewards(
-                        &mut clarity_tx,
-                        &mature_miner_rewards,
-                    )?
-                } else {
-                    0
-                };
+            let new_liquid_miner_ustx = if matured_rewards.len() > 0 {
+                // grant in order by miner, then users
+                StacksChainState::process_matured_miner_rewards(&mut clarity_tx, &matured_rewards)?
+            } else {
+                0
+            };
 
             // total burns
             let total_burnt = block_burns
@@ -3994,11 +3992,13 @@ impl StacksChainState {
         )
         .expect("FATAL: failed to advance chain tip");
 
-        chainstate_tx.log_transactions_processed(&new_tip.index_block_hash(), &txs_receipts);
+        chainstate_tx.log_transactions_processed(&new_tip.index_block_hash(), &tx_receipts);
 
         let epoch_receipt = StacksEpochReceipt {
             header: new_tip,
-            tx_receipts: txs_receipts,
+            tx_receipts,
+            matured_rewards,
+            matured_rewards_info,
             parent_microblocks_cost: microblock_execution_cost,
             anchored_block_cost: block_execution_cost,
         };