wip(fix+test): fix common ancestor and fork branch bug, add new reorg…

… tests

src/bin.rs

@@ -76,11 +76,12 @@ async fn main() -> anyhow::Result<()> {
     // async fetch the data stream through the lib
     let block_events = block_events::subscribe_to_blocks(
         base_url.as_str(),
-        Some((
-            102,
-            BlockHash::from_str("3d3ab0efd0f8f0eb047d9e77f7ad7c6b6791c896bd8a21437da555670f799e08")
-                .unwrap(),
-        )),
+        // Some((
+        //     102,
+        //     BlockHash::from_str("3d3ab0efd0f8f0eb047d9e77f7ad7c6b6791c896bd8a21437da555670f799e08")
+        //         .unwrap(),
+        // )),
+        None,
     )
     .await?;
 

src/lib.rs

@@ -85,12 +85,13 @@ impl BlockHeadersCache {
         let mut common_ancestor = branch_candidate;
         let mut fork_branch: VecDeque<BlockExtended> = VecDeque::new();
         while !self.active_headers.contains_key(&common_ancestor.id) {
-            log::debug!("{:?}", common_ancestor);
             fork_branch.push_back(common_ancestor);
             common_ancestor = http_client
                 ._get_block(common_ancestor.prev_blockhash)
                 .await?;
         }
+        log::debug!("[common_ancestor] {:?}", common_ancestor);
+        log::debug!("[fork_branch] {:?}", fork_branch);
         Ok((common_ancestor, fork_branch))
     }
 
@@ -108,8 +109,10 @@ impl BlockHeadersCache {
             all_disconnected.push_back(disconnected_header);
             self.stale_headers
                 .insert(disconnected_hash, disconnected_header);
-            self.tip = block.id;
+            self.tip = disconnected_header.prev_blockhash;
         }
+        log::info!("[all_disconnected] {:?}", all_disconnected);
+        log::info!("[self.tip] {:?}", self.tip);
         Ok(all_disconnected)
     }
 
@@ -119,13 +122,15 @@ impl BlockHeadersCache {
     pub fn apply_fork_chain(
         &mut self,
         mut fork_branch: VecDeque<BlockExtended>,
-    ) -> anyhow::Result<BlockHash> {
+    ) -> anyhow::Result<(BlockHash, VecDeque<BlockExtended>)> {
+        let mut connected = VecDeque::new();
         while !fork_branch.is_empty() {
             let block = fork_branch.pop_front().unwrap();
+            connected.push_back(block);
             self.active_headers.insert(block.id, block);
             self.tip = block.id;
         }
-        Ok(self.tip)
+        Ok((self.tip, connected))
     }
 }
 
@@ -135,14 +140,17 @@ pub async fn subscribe_to_blocks(
     checkpoint: Option<(u64, BlockHash)>,
 ) -> anyhow::Result<Pin<Box<dyn Stream<Item = BlockEvent>>>> {
     let http_client = http::HttpClient::new(base_url, DEFAULT_CONCURRENT_REQUESTS);
+
+    let current_tip = match checkpoint {
+        Some((height, _)) => height - 1,
+        _ => http_client._get_height().await?,
+    };
+
     let cache = BlockHeadersCache {
-        tip: http_client
-            ._get_block_height(http_client._get_height().await?)
-            .await?,
+        tip: http_client._get_block_height(current_tip).await?,
         active_headers: HashMap::new(),
         stale_headers: HashMap::new(),
     };
-    log::debug!("[BlockHeadersCache] {:?}", cache);
 
     match checkpoint {
         Some(checkpoint) => {
@@ -193,10 +201,16 @@ async fn process_candidates(
             for disconnected in cache.rollback_active_chain(common_ancestor).await.unwrap().iter() {
                 yield BlockEvent::Disconnected((disconnected.height, disconnected.id));
             }
+            // TODO: (@leonardo.lima) fix order of return
 
             // iterate over forked chain candidates
             // update [`Cache`] active_headers field with candidates
-            let _ = cache.apply_fork_chain(fork_chain);
+            let (_, connected) = cache.apply_fork_chain(fork_chain).unwrap();
+            for block in connected {
+                yield BlockEvent::Connected(BlockHeader::from(block.clone()));
+            }
+            // TODO: (@leonardo.lima fix order of return)
+
         }
     };
     Ok(stream)
@@ -219,18 +233,13 @@ pub async fn fetch_blocks(
     let mut tip = http_client._get_height().await?;
     let mut height = ckpt_height;
 
-    log::debug!("tip: {}", tip);
-    log::debug!("ckpt: {:?}", checkpoint);
-
     let mut interval = Instant::now(); // it should try to update the tip every 5 minutes.
     let stream = stream! {
         while height <= tip {
             let hash = http_client._get_block_height(height).await.unwrap();
             let block = http_client._get_block(hash).await.unwrap();
 
             height += 1;
-            log::debug!("height: {}", height);
-            log::debug!("block: {:?}", block);
 
             if interval.elapsed() >= Duration::from_secs(300) {
                 interval = Instant::now();

tests/integration_tests.rs

@@ -385,17 +385,6 @@ async fn test_block_events_stream_with_checkpoint() {
     .await
     .unwrap();
 
-    // std::thread::sleep(delay);
-    // // generate 5 new blocks through bitcoind rpc-client
-    // let mut generated_blocks = VecDeque::from(
-    //     rpc_client
-    //         .generate_to_address(3, &rpc_client.get_new_address(None, None).unwrap())
-    //         .unwrap(),
-    // );
-
-    // insert the first blocks starting from checkpoint
-    // first_blocks.append(&mut generated_blocks);
-
     // consume new blocks from block-events stream
     pin_mut!(block_events);
     while !first_blocks.is_empty() {
@@ -417,4 +406,101 @@ async fn test_block_events_stream_with_checkpoint() {
 
 #[tokio::test]
 #[serial]
-async fn test_block_events_stream_with_reorg() {}
+async fn test_block_events_stream_with_reorg() {
+    let _ = env_logger::try_init();
+    let delay = Duration::from_millis(5000);
+
+    let docker = clients::Cli::docker();
+    let client = MempoolTestClient::default();
+
+    let _mariadb = docker.run(client.mariadb_database);
+
+    std::thread::sleep(delay); // there is some delay between running the docker and the port being really available
+    let mempool = docker.run(client.mempool_backend);
+
+    // get block-events stream
+    let block_events = block_events::subscribe_to_blocks(
+        build_base_url(mempool.get_host_port_ipv4(mempool.get_host_port_ipv4(8999))).as_str(),
+        None,
+    )
+    .await
+    .unwrap();
+
+    // initiate bitcoind client
+    let rpc_client = &client.bitcoind.client;
+
+    // generate 5 new blocks through bitcoind rpc-client
+    let generated_blocks = VecDeque::from(
+        rpc_client
+            .generate_to_address(5, &rpc_client.get_new_address(None, None).unwrap())
+            .unwrap(),
+    );
+    let mut new_blocks = generated_blocks.clone();
+
+    // consume new blocks from block-events stream
+    pin_mut!(block_events);
+    while !new_blocks.is_empty() {
+        let block_hash = new_blocks.pop_front().unwrap();
+        let block_event = block_events.next().await.unwrap();
+
+        // should produce a BlockEvent::Connected result for each block event
+        assert!(matches!(block_event, BlockEvent::Connected { .. }));
+
+        // should parse the BlockEvent::Connected successfully
+        let connected_block = match block_event {
+            BlockEvent::Connected(block) => block,
+            _ => unreachable!("This test is supposed to have only connected blocks, please check why it's generating disconnected and/or errors at the moment."),
+        };
+        assert_eq!(block_hash.to_owned(), connected_block.block_hash());
+    }
+
+    // invalidate last 2 blocks
+    let mut invalidated_blocks = VecDeque::new();
+    for block in generated_blocks.range(3..) {
+        rpc_client.invalidate_block(block).unwrap();
+        invalidated_blocks.push_front(block);
+    }
+    log::debug!("invalidated_blocks {:?}", invalidated_blocks);
+
+    // generate 2 new blocks
+    let mut new_blocks = VecDeque::from(
+        rpc_client
+            .generate_to_address(3, &rpc_client.get_new_address(None, None).unwrap())
+            .unwrap(),
+    );
+    log::debug!("new_blocks {:?}", new_blocks);
+
+    // should disconnect invalidated blocks
+    while !invalidated_blocks.is_empty() {
+        log::info!("len {:?}", invalidated_blocks.len());
+        let invalidated = invalidated_blocks.pop_front().unwrap();
+        let block_event = block_events.next().await.unwrap();
+
+        log::info!("{:?}", block_event);
+        // should produce a BlockEvent::Connected result for each block event
+        assert!(matches!(block_event, BlockEvent::Disconnected(..)));
+
+        // should parse the BlockEvent::Connected successfully
+        let disconnected = match block_event {
+            BlockEvent::Disconnected((_, hash)) => hash,
+            _ => unreachable!("This test is supposed to have only connected blocks, please check why it's generating disconnected and/or errors at the moment."),
+        };
+        assert_eq!(invalidated.to_owned(), disconnected);
+    }
+
+    // should connect the new created blocks
+    while !new_blocks.is_empty() {
+        let new_block = new_blocks.pop_front().unwrap();
+        let block_event = block_events.next().await.unwrap();
+
+        // should produce a BlockEvent::Connected result for each block event
+        assert!(matches!(block_event, BlockEvent::Connected { .. }));
+
+        // should parse the BlockEvent::Connected successfully
+        let connected = match block_event {
+            BlockEvent::Connected(block) => block.block_hash(),
+            _ => unreachable!("This test is supposed to have only connected blocks, please check why it's generating disconnected and/or errors at the moment."),
+        };
+        assert_eq!(new_block.to_owned(), connected);
+    }
+}