Merge pull request #830 from AloeareV/hotfix

Hotfix

darkside-tests/tests/network_interruption_tests.rs

@@ -2,25 +2,79 @@ use std::{
     collections::HashMap,
     sync::{
         atomic::{AtomicBool, Ordering},
-        Arc,
+        Arc, Mutex,
     },
     time::Duration,
 };
 
-use darkside_tests::utils::{
-    create_chainbuild_file, load_chainbuild_file,
-    scenarios::{DarksideScenario, DarksideSender},
+use darkside_tests::{
+    constants::DARKSIDE_SEED,
+    utils::{
+        create_chainbuild_file, load_chainbuild_file, prepare_darksidewalletd,
+        scenarios::{DarksideScenario, DarksideSender},
+        DarksideHandler,
+    },
 };
 use json::JsonValue;
 use tokio::time::sleep;
-use zingo_testutils::start_proxy_and_connect_lightclient;
+use zingo_testutils::{scenarios::setup::ClientBuilder, start_proxy_and_connect_lightclient};
+use zingoconfig::RegtestNetwork;
 use zingolib::{
     get_base_address,
     lightclient::PoolBalances,
     testvectors::seeds,
     wallet::{data::summaries::ValueTransferKind, Pool},
 };
 
+#[ignore]
+#[tokio::test]
+async fn interrupt_initial_tree_fetch() {
+    let darkside_handler = DarksideHandler::new(None);
+
+    let server_id = zingoconfig::construct_lightwalletd_uri(Some(format!(
+        "http://127.0.0.1:{}",
+        darkside_handler.grpc_port
+    )));
+    prepare_darksidewalletd(server_id.clone(), true)
+        .await
+        .unwrap();
+    let regtest_network = RegtestNetwork::all_upgrades_active();
+    let light_client = ClientBuilder::new(server_id, darkside_handler.darkside_dir.clone())
+        .build_client(DARKSIDE_SEED.to_string(), 0, true, regtest_network)
+        .await;
+    let mut cond_log =
+        HashMap::<&'static str, Box<dyn Fn(Arc<AtomicBool>) + Send + Sync + 'static>>::new();
+    let (sender, receiver) = std::sync::mpsc::channel();
+    let sender = Arc::new(Mutex::new(sender));
+    cond_log.insert(
+        "get_tree_state",
+        Box::new(move |_online| {
+            println!("acquiring lcok");
+            match sender.clone().lock() {
+                Ok(mutguard) => {
+                    println!("acquired lock");
+                    mutguard.send(()).unwrap();
+                    println!("Ending proxy");
+                }
+                Err(poisoned_thread) => panic!("{}", poisoned_thread),
+            };
+        }),
+    );
+    let (proxy_handle, _proxy_status) =
+        start_proxy_and_connect_lightclient(&light_client, cond_log);
+
+    let receiver = Arc::new(Mutex::new(receiver));
+    println!("made receiver");
+    std::thread::spawn(move || {
+        receiver.lock().unwrap().recv().unwrap();
+        proxy_handle.abort();
+        println!("aborted proxy");
+    });
+    println!("spawned abortion task");
+    let result = light_client.do_sync(true).await;
+    assert_eq!(result.unwrap_err(),"status: Unavailable, message: \"error trying to connect: tcp connect error: Connection refused (os error 111)\", details: [], metadata: MetadataMap { headers: {} }");
+}
+
 // Verifies that shielded transactions correctly mark notes as change
 // Also verifies:
 // - send-to-self value transfer is created
@@ -104,7 +158,7 @@ async fn shielded_note_marked_as_change_test() {
 
     // setup gRPC network interrupt conditions
     let mut conditional_logic =
-        HashMap::<&'static str, Box<dyn Fn(&Arc<AtomicBool>) + Send + Sync>>::new();
+        HashMap::<&'static str, Box<dyn Fn(Arc<AtomicBool>) + Send + Sync>>::new();
     // conditional_logic.insert(
     //     "get_block_range",
     //     Box::new(|online: &Arc<AtomicBool>| {
@@ -114,20 +168,20 @@ async fn shielded_note_marked_as_change_test() {
     // );
     conditional_logic.insert(
         "get_tree_state",
-        Box::new(|online: &Arc<AtomicBool>| {
+        Box::new(|online: Arc<AtomicBool>| {
             println!("Turning off, as we received get_tree_state call");
             online.store(false, Ordering::Relaxed);
         }),
     );
     conditional_logic.insert(
         "get_transaction",
-        Box::new(|online: &Arc<AtomicBool>| {
+        Box::new(|online: Arc<AtomicBool>| {
             println!("Turning off, as we received get_transaction call");
             online.store(false, Ordering::Relaxed);
         }),
     );
 
-    let proxy_status =
+    let (_proxy_handle, proxy_status) =
         start_proxy_and_connect_lightclient(scenario.get_lightclient(0), conditional_logic);
     tokio::task::spawn(async move {
         loop {

zingo-testutils/src/grpc_proxy.rs

@@ -61,7 +61,7 @@ pub struct ProxyServer {
     pub lightwalletd_uri: http::Uri,
     pub online: Arc<AtomicBool>,
     #[allow(clippy::type_complexity)]
-    pub conditional_operations: HashMap<&'static str, Box<dyn Fn(&Arc<AtomicBool>) + Send + Sync>>,
+    pub conditional_operations: HashMap<&'static str, Box<dyn Fn(Arc<AtomicBool>) + Send + Sync>>,
 }
 
 impl ProxyServer {
@@ -93,7 +93,7 @@ impl ProxyServer {
 
     fn passthrough_helper(&self, name: &str) {
         if let Some(fun) = self.conditional_operations.get(name) {
-            fun(&self.online)
+            fun(self.online.clone())
         }
     }
     pub fn new(lightwalletd_uri: http::Uri) -> Self {

zingo-testutils/src/lib.rs

@@ -9,6 +9,7 @@ use std::string::String;
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use std::time::Duration;
+use tokio::task::JoinHandle;
 use zcash_address::unified::{Fvk, Ufvk};
 use zingolib::wallet::keys::unified::WalletCapability;
 use zingolib::wallet::WalletBase;
@@ -1081,24 +1082,27 @@ pub mod scenarios {
 #[allow(clippy::type_complexity)]
 pub fn start_proxy_and_connect_lightclient(
     client: &LightClient,
-    conditional_operations: HashMap<&'static str, Box<dyn Fn(&Arc<AtomicBool>) + Send + Sync>>,
-) -> Arc<AtomicBool> {
+    conditional_operations: HashMap<&'static str, Box<dyn Fn(Arc<AtomicBool>) + Send + Sync>>,
+) -> (
+    JoinHandle<Result<(), tonic::transport::Error>>,
+    Arc<AtomicBool>,
+) {
     let proxy_online = Arc::new(std::sync::atomic::AtomicBool::new(true));
     let proxy_port = portpicker::pick_unused_port().unwrap();
     let proxy_uri = format!("http://localhost:{proxy_port}");
-    let _proxy_handle = ProxyServer {
+    let proxy_handle = ProxyServer {
         lightwalletd_uri: client.get_server_uri(),
         online: proxy_online.clone(),
         conditional_operations,
     }
     .serve(proxy_port);
     client.set_server(proxy_uri.parse().unwrap());
-    proxy_online
+    (proxy_handle, proxy_online)
 }
 
 pub async fn check_proxy_server_works() {
     let (_regtest_manager, _cph, ref faucet) = scenarios::faucet_default().await;
-    let proxy_status = start_proxy_and_connect_lightclient(faucet, HashMap::new());
+    let (_proxy_handle, proxy_status) = start_proxy_and_connect_lightclient(faucet, HashMap::new());
     proxy_status.store(false, std::sync::atomic::Ordering::Relaxed);
     tokio::task::spawn(async move {
         sleep(Duration::from_secs(5)).await;

zingolib/src/blaze/block_management_reorg_detection.rs

@@ -324,17 +324,23 @@ impl BlockManagementData {
         existing_blocks: Arc<RwLock<Vec<BlockData>>>,
         transaction_metadata_set: Arc<RwLock<TransactionMetadataSet>>,
     ) {
-        // First, pop the first block (which is the top block) in the existing_blocks.
-        let top_wallet_block = existing_blocks.write().await.drain(0..1).next().unwrap();
-        if top_wallet_block.height != reorg_height {
-            panic!("Wrong block reorg'd");
-        }
+        let mut existing_blocks_writelock = existing_blocks.write().await;
+        if existing_blocks_writelock.len() != 0 {
+            // First, pop the first block (which is the top block) in the existing_blocks.
+            let top_wallet_block = existing_blocks_writelock
+                .drain(0..1)
+                .next()
+                .expect("there to be blocks");
+            if top_wallet_block.height != reorg_height {
+                panic!("Wrong block reorg'd");
+            }
 
-        // Remove all wallet transactions at the height
-        transaction_metadata_set
-            .write()
-            .await
-            .remove_txns_at_height(reorg_height);
+            // Remove all wallet transactions at the height
+            transaction_metadata_set
+                .write()
+                .await
+                .remove_txns_at_height(reorg_height);
+        }
     }
 
     /// Ingest the incoming blocks, handle any reorgs, then populate the block data

zingolib/src/lightclient.rs

@@ -1444,9 +1444,10 @@ impl LightClient {
         let mut res = Err("No batches were run!".to_string());
         for (batch_num, batch_latest_block) in latest_block_batches.into_iter().enumerate() {
             res = self.sync_nth_batch(batch_latest_block, batch_num).await;
-            if res.is_err() {
+            if let Err(ref err) = res {
                 // If something went wrong during a batch, reset the wallet state to
                 // how it was before the latest batch
+                println!("sync hit error {}. Rolling back", err);
                 BlockManagementData::invalidate_block(
                     self.wallet.last_synced_height().await,
                     self.wallet.blocks.clone(),