Merge branch 'develop' into feat/pessimistic-estimator-fix

stacks-network · Jan 7, 2022 · 8750b69 · 8750b69
2 parents 37fcaac + fd3f28d
commit 8750b69
Show file tree

Hide file tree

Showing 26 changed files with 4,680 additions and 1,096 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -63,6 +63,7 @@ slog-term = "2.6.0"
 slog-json = { version = "2.3.0", optional = true }
 chrono = "0.4.19"
 libc = "0.2.82"
+siphasher = "0.3.7"
 
 [target.'cfg(unix)'.dependencies]
 nix = "0.23"

diff --git a/clarity/Cargo.toml b/clarity/Cargo.toml
@@ -40,6 +40,7 @@ slog-term = "2.6.0"
 slog-json = { version = "2.3.0", optional = true }
 chrono = "0.4.19"
 libc = "0.2.82"
+siphasher = "0.3.7"
 
 [target.'cfg(unix)'.dependencies]
 nix = "0.23"

diff --git a/src/chainstate/stacks/db/blocks.rs b/src/chainstate/stacks/db/blocks.rs
@@ -47,12 +47,14 @@ use chainstate::stacks::{
     C32_ADDRESS_VERSION_TESTNET_MULTISIG, C32_ADDRESS_VERSION_TESTNET_SINGLESIG,
 };
 use clarity_vm::clarity::{ClarityBlockConnection, ClarityConnection, ClarityInstance};
+use core::mempool::MemPoolDB;
 use core::mempool::MAXIMUM_MEMPOOL_TX_CHAINING;
 use core::*;
 use cost_estimates::EstimatorError;
 use net::BlocksInvData;
 use net::Error as net_error;
 use net::ExtendedStacksHeader;
+use net::MemPoolSyncData;
 use util::db::u64_to_sql;
 use util::db::Error as db_error;
 use util::db::{
@@ -528,6 +530,18 @@ impl StreamCursor {
         }))
     }
 
+    pub fn new_tx_stream(tx_query: MemPoolSyncData, max_txs: u64, height: u64) -> StreamCursor {
+        StreamCursor::MempoolTxs(TxStreamData {
+            tx_query,
+            last_randomized_txid: Txid([0u8; 32]),
+            tx_buf: vec![],
+            tx_buf_ptr: 0,
+            num_txs: 0,
+            max_txs: max_txs,
+            height: height,
+        })
+    }
+
     fn stream_one_byte<W: Write>(fd: &mut W, b: u8) -> Result<u64, Error> {
         loop {
             match fd.write(&[b]) {
@@ -560,6 +574,8 @@ impl StreamCursor {
             StreamCursor::Block(ref stream) => stream.offset(),
             StreamCursor::Microblocks(ref stream) => stream.offset(),
             StreamCursor::Headers(ref stream) => stream.offset(),
+            // no-op for mempool txs
+            StreamCursor::MempoolTxs(..) => 0,
         }
     }
 
@@ -568,11 +584,14 @@ impl StreamCursor {
             StreamCursor::Block(ref mut stream) => stream.add_bytes(nw),
             StreamCursor::Microblocks(ref mut stream) => stream.add_bytes(nw),
             StreamCursor::Headers(ref mut stream) => stream.add_bytes(nw),
+            // no-op fo mempool txs
+            StreamCursor::MempoolTxs(..) => (),
         }
     }
 
     pub fn stream_to<W: Write>(
         &mut self,
+        mempool: &MemPoolDB,
         chainstate: &mut StacksChainState,
         fd: &mut W,
         count: u64,
@@ -592,6 +611,7 @@ impl StreamCursor {
                         .and_then(|bytes_sent| Ok(bytes_sent + num_written))
                 }
             }
+            StreamCursor::MempoolTxs(ref mut tx_stream) => mempool.stream_txs(fd, tx_stream, count),
             StreamCursor::Headers(ref mut stream) => {
                 let mut num_written = 0;
                 if stream.total_bytes == 0 {
@@ -9140,35 +9160,59 @@ pub mod test {
         stream: &mut StreamCursor,
         count: u64,
     ) -> Result<Vec<u8>, chainstate_error> {
+        let mempool = MemPoolDB::open_test(
+            chainstate.mainnet,
+            chainstate.chain_id,
+            &chainstate.root_path,
+        )
+        .unwrap();
         let mut bytes = vec![];
-        stream.stream_to(chainstate, &mut bytes, count).map(|nr| {
-            assert_eq!(bytes.len(), nr as usize);
-            bytes
-        })
+        stream
+            .stream_to(&mempool, chainstate, &mut bytes, count)
+            .map(|nr| {
+                assert_eq!(bytes.len(), nr as usize);
+                bytes
+            })
     }
 
     fn stream_unconfirmed_microblocks_to_vec(
         chainstate: &mut StacksChainState,
         stream: &mut StreamCursor,
         count: u64,
     ) -> Result<Vec<u8>, chainstate_error> {
+        let mempool = MemPoolDB::open_test(
+            chainstate.mainnet,
+            chainstate.chain_id,
+            &chainstate.root_path,
+        )
+        .unwrap();
         let mut bytes = vec![];
-        stream.stream_to(chainstate, &mut bytes, count).map(|nr| {
-            assert_eq!(bytes.len(), nr as usize);
-            bytes
-        })
+        stream
+            .stream_to(&mempool, chainstate, &mut bytes, count)
+            .map(|nr| {
+                assert_eq!(bytes.len(), nr as usize);
+                bytes
+            })
     }
 
     fn stream_confirmed_microblocks_to_vec(
         chainstate: &mut StacksChainState,
         stream: &mut StreamCursor,
         count: u64,
     ) -> Result<Vec<u8>, chainstate_error> {
+        let mempool = MemPoolDB::open_test(
+            chainstate.mainnet,
+            chainstate.chain_id,
+            &chainstate.root_path,
+        )
+        .unwrap();
         let mut bytes = vec![];
-        stream.stream_to(chainstate, &mut bytes, count).map(|nr| {
-            assert_eq!(bytes.len(), nr as usize);
-            bytes
-        })
+        stream
+            .stream_to(&mempool, chainstate, &mut bytes, count)
+            .map(|nr| {
+                assert_eq!(bytes.len(), nr as usize);
+                bytes
+            })
     }
 
     fn decode_microblock_stream(mblock_bytes: &Vec<u8>) -> Vec<StacksMicroblock> {

diff --git a/src/chainstate/stacks/db/mod.rs b/src/chainstate/stacks/db/mod.rs
@@ -58,6 +58,7 @@ use clarity_vm::clarity::{
 use core::*;
 use net::atlas::BNS_CHARS_REGEX;
 use net::Error as net_error;
+use net::MemPoolSyncData;
 use util::db::Error as db_error;
 use util::db::{
     query_count, query_row, tx_begin_immediate, tx_busy_handler, DBConn, DBTx, FromColumn, FromRow,
@@ -485,6 +486,7 @@ pub enum StreamCursor {
     Block(BlockStreamData),
     Microblocks(MicroblockStreamData),
     Headers(HeaderStreamData),
+    MempoolTxs(TxStreamData),
 }
 
 #[derive(Debug, PartialEq, Clone)]
@@ -538,6 +540,23 @@ pub struct HeaderStreamData {
     corked: bool,
 }
 
+#[derive(Debug, PartialEq, Clone)]
+pub struct TxStreamData {
+    /// Mempool sync data requested
+    pub tx_query: MemPoolSyncData,
+    /// last txid loaded
+    pub last_randomized_txid: Txid,
+    /// serialized transaction buffer that's being sent
+    pub tx_buf: Vec<u8>,
+    pub tx_buf_ptr: usize,
+    /// number of transactions sent so far
+    pub num_txs: u64,
+    /// maximum we can send
+    pub max_txs: u64,
+    /// height of the chain at time of query
+    pub height: u64,
+}
+
 pub const CHAINSTATE_VERSION: &'static str = "2";
 
 const CHAINSTATE_INITIAL_SCHEMA: &'static [&'static str] = &[

diff --git a/src/codec/macros.rs b/src/codec/macros.rs
@@ -35,3 +35,10 @@ macro_rules! impl_byte_array_message_codec {
         }
     };
 }
+
+// macro for determining how big an inv bitvec can be, given its bitlen
+macro_rules! BITVEC_LEN {
+    ($bitvec:expr) => {
+        (($bitvec) / 8 + if ($bitvec) % 8 > 0 { 1 } else { 0 }) as u32
+    };
+}