Merge pull request #21 from rustaceanrob/dep-reorg

test(lib): depth two reorg, stale tip

example/memory.rs

@@ -29,7 +29,7 @@ async fn main() {
     let (height, hash) = SIGNET_HEADER_CP.last().unwrap();
     let anchor = HeaderCheckpoint::new(*height, BlockHash::from_str(hash).unwrap());
     // Define a peer to connect to
-    let peer = IpAddr::V4(Ipv4Addr::new(95, 217, 198, 121));
+    let peer = IpAddr::V4(Ipv4Addr::new(23, 137, 57, 100));
     // Limited devices may not save any peers to disk
     let peer_store = StatelessPeerStore::new();
     // Create a new node builder

scripts/mine.sh

@@ -0,0 +1,8 @@
+
+RPC_USER="test"
+RPC_PASSWORD="kyoto"
+CHAIN="regtest"
+WALLET="test_kyoto"
+bitcoin-cli -chain=$CHAIN -rpcuser=$RPC_USER -rpcpassword=$RPC_PASSWORD createwallet $WALLET
+NEW_ADDRESS=$(bitcoin-cli -chain=$CHAIN -rpcuser=$RPC_USER -rpcpassword=$RPC_PASSWORD getnewaddress)
+bitcoin-cli -chain=$CHAIN -rpcuser=$RPC_USER -rpcpassword=$RPC_PASSWORD generatetoaddress 2500 $NEW_ADDRESS

src/chain/header_chain.rs

@@ -218,7 +218,7 @@ impl HeaderChain {
                     .insert(current_anchor + 1 + index as u32, *header);
             }
         }
-        reorged
+        reorged.into_iter().rev().collect()
     }
 
     fn remove(&mut self, height: &u32) {
@@ -415,7 +415,7 @@ mod tests {
         assert_eq!(reorged.len(), 2);
         assert_eq!(
             reorged.iter().map(|f| f.header).collect::<Vec<Header>>(),
-            vec![block_2, block_1]
+            vec![block_1, block_2]
         );
         assert_eq!(vec![new_block_1, new_block_2], chain.values());
         let no_org = chain.extend(&batch_2);

tests/node.rs

@@ -7,6 +7,7 @@ use std::{
 use bitcoin::{consensus::serialize, Amount, ScriptBuf};
 use bitcoincore_rpc::{json::CreateRawTransactionInput, RpcApi};
 use kyoto::{
+    chain::checkpoints::HeaderCheckpoint,
     node::{client::Client, node::Node},
     TxBroadcast,
 };
@@ -55,6 +56,21 @@ async fn new_node_sql(addrs: HashSet<ScriptBuf>) -> (Node, Client) {
     (node, client)
 }
 
+async fn new_node_anchor_sql(
+    addrs: HashSet<ScriptBuf>,
+    checkpoint: HeaderCheckpoint,
+) -> (Node, Client) {
+    let host = (IpAddr::from(Ipv4Addr::new(0, 0, 0, 0)), PORT);
+    let builder = kyoto::node::builder::NodeBuilder::new(bitcoin::Network::Regtest);
+    let (node, client) = builder
+        .add_peers(vec![host])
+        .add_scripts(addrs)
+        .anchor_checkpoint(checkpoint)
+        .build_node()
+        .await;
+    (node, client)
+}
+
 // This test may be run as much as required without altering Bitcoin Core's database.
 #[tokio::test]
 async fn test_reorg() {
@@ -239,7 +255,7 @@ async fn test_long_chain() {
 
 // This test requires a clean Bitcoin Core regtest instance or unchange headers from Bitcoin Core since the last test.
 #[tokio::test]
-async fn test_sql() {
+async fn test_sql_reorg() {
     let rpc_result = initialize_client();
     // If we can't fetch the genesis block then bitcoind is not running. Just exit.
     if let Err(_) = rpc_result {
@@ -327,3 +343,191 @@ async fn test_sql() {
     client.shutdown().await.unwrap();
     rpc.stop().unwrap();
 }
+
+// This test requires a clean Bitcoin Core regtest instance or unchange headers from Bitcoin Core since the last test.
+#[tokio::test]
+async fn test_two_deep_reorg() {
+    let rpc_result = initialize_client();
+    // If we can't fetch the genesis block then bitcoind is not running. Just exit.
+    if let Err(_) = rpc_result {
+        println!("Bitcoin Core is not running. Skipping this test...");
+        return;
+    }
+    let rpc = rpc_result.unwrap();
+    // Mine some blocks.
+    let miner = rpc.get_new_address(None, None).unwrap().assume_checked();
+    rpc.generate_to_address(10, &miner).unwrap();
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    let best = rpc.get_best_block_hash().unwrap();
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    let mut scripts = HashSet::new();
+    let other = rpc.get_new_address(None, None).unwrap().assume_checked();
+    scripts.insert(other.into());
+    let (mut node, mut client) = new_node_sql(scripts.clone()).await;
+    tokio::task::spawn(async move { node.run().await });
+    let (_, mut recv) = client.split();
+    while let Ok(message) = recv.recv().await {
+        match message {
+            kyoto::node::messages::NodeMessage::Dialog(d) => println!("{d}"),
+            kyoto::node::messages::NodeMessage::Warning(e) => println!("{e}"),
+            kyoto::node::messages::NodeMessage::Synced(update) => {
+                println!("Done");
+                assert_eq!(update.tip().hash, best);
+                break;
+            }
+            _ => {}
+        }
+    }
+    client.shutdown().await.unwrap();
+    // Reorganize the blocks
+    let old_height = rpc.get_block_count().unwrap();
+    let old_best = best;
+    rpc.invalidate_block(&best).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let best = rpc.get_best_block_hash().unwrap();
+    rpc.invalidate_block(&best).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    rpc.generate_to_address(3, &miner).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let best = rpc.get_best_block_hash().unwrap();
+    // Make sure the reorganization is caught after a cold start
+    let (mut node, mut client) = new_node_sql(scripts.clone()).await;
+    tokio::task::spawn(async move { node.run().await });
+    let (_, mut recv) = client.split();
+    while let Ok(message) = recv.recv().await {
+        match message {
+            kyoto::node::messages::NodeMessage::Dialog(d) => println!("{d}"),
+            kyoto::node::messages::NodeMessage::Warning(e) => println!("{e}"),
+            kyoto::node::messages::NodeMessage::BlocksDisconnected(blocks) => {
+                assert_eq!(blocks.len(), 2);
+                assert_eq!(blocks.last().unwrap().header.block_hash(), old_best);
+                assert_eq!(old_height as u32, blocks.last().unwrap().height);
+            }
+            kyoto::node::messages::NodeMessage::Synced(update) => {
+                println!("Done");
+                assert_eq!(update.tip().hash, best);
+                break;
+            }
+            _ => {}
+        }
+    }
+    client.shutdown().await.unwrap();
+    // Mine more blocks
+    rpc.generate_to_address(2, &miner).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let best = rpc.get_best_block_hash().unwrap();
+    // Make sure the node does not have any corrupted headers
+    let (mut node, mut client) = new_node_sql(scripts.clone()).await;
+    tokio::task::spawn(async move { node.run().await });
+    let (_, mut recv) = client.split();
+    // The node properly syncs after persisting a reorg
+    while let Ok(message) = recv.recv().await {
+        match message {
+            kyoto::node::messages::NodeMessage::Dialog(d) => println!("{d}"),
+            kyoto::node::messages::NodeMessage::Warning(e) => println!("{e}"),
+            kyoto::node::messages::NodeMessage::Synced(update) => {
+                println!("Done");
+                assert_eq!(update.tip().hash, best);
+                break;
+            }
+            _ => {}
+        }
+    }
+    client.shutdown().await.unwrap();
+    rpc.stop().unwrap();
+}
+
+// This test requires a clean Bitcoin Core regtest instance or unchange headers from Bitcoin Core since the last test.
+#[tokio::test]
+#[ignore = "broken"]
+async fn test_sql_stale_anchor() {
+    let rpc = bitcoincore_rpc::Client::new(
+        HOST,
+        bitcoincore_rpc::Auth::UserPass(RPC_USER.into(), RPC_PASSWORD.into()),
+    )
+    .unwrap();
+    // Do a call that will only fail if we are not connected to RPC.
+    if let Err(_) = rpc.get_best_block_hash() {
+        println!("Bitcoin Core is not running. Skipping this test...");
+    }
+    // Get an address and the tip of the chain.
+    let miner = rpc.get_new_address(None, None).unwrap().assume_checked();
+    let best = rpc.get_best_block_hash().unwrap();
+    let mut scripts = HashSet::new();
+    let other = rpc.get_new_address(None, None).unwrap().assume_checked();
+    scripts.insert(other.into());
+    let (mut node, mut client) = new_node_sql(scripts.clone()).await;
+    tokio::task::spawn(async move { node.run().await });
+    let (_, mut recv) = client.split();
+    while let Ok(message) = recv.recv().await {
+        match message {
+            kyoto::node::messages::NodeMessage::Dialog(d) => println!("{d}"),
+            kyoto::node::messages::NodeMessage::Warning(e) => println!("{e}"),
+            kyoto::node::messages::NodeMessage::Synced(update) => {
+                println!("Done");
+                assert_eq!(update.tip().hash, best);
+                break;
+            }
+            _ => {}
+        }
+    }
+    client.shutdown().await.unwrap();
+    // Reorganize the blocks
+    let old_best = best;
+    let old_height = rpc.get_block_count().unwrap();
+    rpc.invalidate_block(&best).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    rpc.generate_to_address(2, &miner).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let best = rpc.get_best_block_hash().unwrap();
+    // Spin up the node on a cold start with a stale tip
+    let (mut node, mut client) = new_node_anchor_sql(
+        scripts.clone(),
+        HeaderCheckpoint::new(old_height as u32, old_best),
+    )
+    .await;
+    tokio::task::spawn(async move { node.run().await });
+    let (_, mut recv) = client.split();
+    // Ensure SQL is able to catch the fork by loading in headers from the database
+    while let Ok(message) = recv.recv().await {
+        match message {
+            kyoto::node::messages::NodeMessage::Dialog(d) => println!("{d}"),
+            kyoto::node::messages::NodeMessage::Warning(e) => println!("{e}"),
+            kyoto::node::messages::NodeMessage::BlocksDisconnected(blocks) => {
+                assert_eq!(blocks.len(), 1);
+                assert_eq!(blocks.first().unwrap().header.block_hash(), old_best);
+                assert_eq!(old_height as u32, blocks.first().unwrap().height);
+            }
+            kyoto::node::messages::NodeMessage::Synced(update) => {
+                println!("Done");
+                assert_eq!(update.tip().hash, best);
+                break;
+            }
+            _ => {}
+        }
+    }
+    client.shutdown().await.unwrap();
+    // Mine more blocks
+    rpc.generate_to_address(2, &miner).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let best = rpc.get_best_block_hash().unwrap();
+    // Make sure the node does not have any corrupted headers
+    let (mut node, mut client) = new_node_sql(scripts.clone()).await;
+    tokio::task::spawn(async move { node.run().await });
+    let (_, mut recv) = client.split();
+    // The node properly syncs after persisting a reorg
+    while let Ok(message) = recv.recv().await {
+        match message {
+            kyoto::node::messages::NodeMessage::Dialog(d) => println!("{d}"),
+            kyoto::node::messages::NodeMessage::Warning(e) => println!("{e}"),
+            kyoto::node::messages::NodeMessage::Synced(update) => {
+                println!("Done");
+                assert_eq!(update.tip().hash, best);
+                break;
+            }
+            _ => {}
+        }
+    }
+    client.shutdown().await.unwrap();
+    rpc.stop().unwrap();
+}