Rework sync checking and initial reorg tests

Author: buffrr

node/src/node.rs

@@ -26,6 +26,7 @@ pub trait BlockSource {
     fn get_block(&self, hash: &BlockHash) -> Result<Block, BitcoinRpcError>;
     fn get_median_time(&self) -> Result<u64, BitcoinRpcError>;
     fn get_block_count(&self) -> Result<u64, BitcoinRpcError>;
+    fn get_best_chain(&self) -> Result<ChainAnchor, BitcoinRpcError>;
 }
 
 #[derive(Debug, Clone)]

node/src/source.rs

@@ -320,6 +320,29 @@ impl BlockFetcher {
         self.job_id.fetch_add(1, Ordering::SeqCst);
     }
 
+    fn should_sync(
+        source: &BitcoinBlockSource,
+        start: ChainAnchor,
+    ) -> Result<Option<ChainAnchor>, BlockFetchError> {
+        let tip = source.get_best_chain()?;
+        if start.height > tip.height {
+            return Err(BlockFetchError::BlockMismatch);
+        }
+
+        // Ensure start block is still in the best chain
+        let start_hash: BlockHash = source.get_block_hash(start.height)?;
+        if start.hash != start_hash {
+            return Err(BlockFetchError::BlockMismatch);
+        }
+
+        // If the chain didn't advance and the hashes match, no rescan is needed
+        if tip.height == start.height && tip.hash == start.hash {
+            return Ok(None);
+        }
+
+        Ok(Some(tip))
+    }
+
     pub fn start(&self, mut checkpoint: ChainAnchor) {
         self.stop();
 
@@ -343,27 +366,28 @@ impl BlockFetcher {
                 }
                 last_check = Instant::now();
 
-                let tip: u32 = match task_src.get_block_count() {
-                    Ok(t) => t as _,
+                let tip = match BlockFetcher::should_sync(&task_src, checkpoint) {
+                    Ok(t) => t,
                     Err(e) => {
-                        _ = task_sender.send(BlockEvent::Error(BlockFetchError::RpcError(e)));
+                        _ = task_sender.send(BlockEvent::Error(e));
                         return;
                     }
                 };
 
-                if tip > checkpoint.height {
+                if let Some(tip) = tip {
                     let res = Self::run_workers(
                         job_id,
                         current_task.clone(),
                         task_src.clone(),
                         task_sender.clone(),
                         checkpoint,
-                        tip,
+                        tip.height,
                         num_workers,
                     );
 
                     match res {
                         Ok(new_tip) => {
+                            info!("new tip set: {}", new_tip.hash);
                             checkpoint = new_tip;
                         }
                         Err(e) => {
@@ -724,4 +748,22 @@ impl BlockSource for BitcoinBlockSource {
             .rpc
             .send_json_blocking(&self.client, &self.rpc.get_block_count())?)
     }
+
+    fn get_best_chain(&self) -> Result<ChainAnchor, BitcoinRpcError> {
+        #[derive(Deserialize)]
+        struct Info {
+            #[serde(rename = "blocks")]
+            height: u64,
+            #[serde(rename = "bestblockhash")]
+            hash: BlockHash,
+        }
+        let info: Info = self
+            .rpc
+            .send_json_blocking(&self.client, &self.rpc.get_blockchain_info())?;
+
+        Ok(ChainAnchor {
+            hash: info.hash,
+            height: info.height as _,
+        })
+    }
 }

node/tests/fetcher_tests.rs

@@ -26,10 +26,7 @@ fn test_block_fetching_from_bitcoin_rpc() -> Result<()> {
         &rig.bitcoind.rpc_url(),
         BitcoinRpcAuth::UserPass("user".to_string(), "password".to_string()),
     ));
-    let (fetcher, receiver) = BlockFetcher::new(
-        fetcher_rpc.clone(),
-        8
-    );
+    let (fetcher, receiver) = BlockFetcher::new(fetcher_rpc.clone(), 8);
     fetcher.start(ChainAnchor { hash, height: 0 });
 
     let timeout = Duration::from_secs(5);

node/tests/reorg_tests.rs

@@ -0,0 +1,22 @@
+use std::time::Duration;
+
+use spaced::rpc::RpcClient;
+use testutil::TestRig;
+
+#[tokio::test]
+async fn it_should_resync_after_reorg_at_same_height() -> anyhow::Result<()> {
+    let rig = TestRig::new().await?;
+    rig.mine_blocks(38, None).await?;
+    rig.wait_until_synced().await?;
+
+    let info = rig.spaced.client.get_server_info().await?;
+    assert_eq!(info.tip.height, 38);
+    assert_eq!(info.tip.hash, rig.get_best_block_hash().await?);
+
+    let reorged = rig.reorg(1).await?;
+    assert_eq!(reorged.len(), 1);
+
+    rig.wait_until_tip(reorged[0], Duration::from_secs(2))
+        .await?;
+    Ok(())
+}

testutil/src/lib.rs

@@ -19,6 +19,7 @@ use ::spaced::{
 use anyhow::Result;
 use bitcoind::{
     anyhow,
+    anyhow::anyhow,
     bitcoincore_rpc::{
         bitcoincore_rpc_json::{GetBlockTemplateModes, GetBlockTemplateRules},
         RpcApi,
@@ -90,6 +91,22 @@ impl TestRig {
         }
     }
 
+    /// Waits until spaced tip == specified best_hash or specified time out
+    pub async fn wait_until_tip(&self, tip: BlockHash, timeout: Duration) -> Result<()> {
+        let start_time = tokio::time::Instant::now();
+
+        loop {
+            let info = self.spaced.client.get_server_info().await?;
+            if info.tip.hash == tip {
+                return Ok(());
+            }
+            if start_time.elapsed() >= timeout {
+                return Err(anyhow!("Rimed out waiting for tip {:?}", tip));
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+    }
+
     /// Waits until named wallet tip == bitcoind tip
     pub async fn wait_until_wallet_synced(&self, wallet_name: &str) -> anyhow::Result<()> {
         loop {