Merge pull request 'Flag Error 60: better handling of errors, forgotten check for wallettx version' (#27) from lucretius/silentdragonlite-cli:mem into dev

Reviewed-on: https://git.hush.is/hush/silentdragonlite-cli/pulls/27

lib/src/grpcconnector.rs

@@ -97,52 +97,64 @@ pub fn get_coinsupply(uri: http::Uri, no_cert: bool) -> Result<Coinsupply, Strin
     let mut rt = tokio::runtime::Runtime::new().map_err(|e| e.to_string())?;
    
     rt.block_on(get_coinsupply_info(&uri, no_cert)).map_err( |e| e.to_string())
-   // tokio::runtime::current_thread::Runtime::new().unwrap().block_on(runner)
 }
 
-async fn get_block_range<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, pool: ThreadPool, c: F) 
-    -> Result<(), Box<dyn std::error::Error>> 
+async fn get_block_range<F : 'static + std::marker::Send>(
+    uri: &http::Uri, 
+    start_height: u64, 
+    end_height: u64, 
+    no_cert: bool, 
+    pool: ThreadPool, 
+    c: F
+) -> Result<(), Box<dyn std::error::Error>> 
 where F : Fn(&[u8], u64) {
     let mut client = get_client(uri, no_cert).await?;
 
-    let bs = BlockId{ height: start_height, hash: vec!()};
-    let be = BlockId{ height: end_height,   hash: vec!()};
+    let bs = BlockId { height: start_height, hash: vec![] };
+    let be = BlockId { height: end_height, hash: vec![] };
 
-    let request = Request::new(BlockRange{ start: Some(bs), end: Some(be) });
+    let request = Request::new(BlockRange { start: Some(bs), end: Some(be) });
 
-     // Channel where the blocks are sent. A None signifies end of all blocks
-     let (tx, rx) = channel::<Option<CompactBlock>>();
-
-     // Channel that the processor signals it is done, so the method can return
-     let (ftx, frx) = channel();
- 
-     // The processor runs on a different thread, so that the network calls don't
-     // block on this
-     pool.execute(move || {
-         while let Some(block) = rx.recv().unwrap() {
-             use prost::Message;
-             let mut encoded_buf = vec![];
- 
-             block.encode(&mut encoded_buf).unwrap();
-             c(&encoded_buf, block.height);
-         }
- 
-         ftx.send(Ok(())).unwrap();
-     });
+    let (tx, rx) = channel::<Option<CompactBlock>>();
+    let (ftx, frx) = channel();
 
+    pool.execute(move || {
+        while let Ok(Some(block)) = rx.recv() {
+            use prost::Message;
+            let mut encoded_buf = vec![];
+    
+            if let Err(e) = block.encode(&mut encoded_buf) {
+                eprintln!("Error encoding block: {:?}", e);
+                break;
+            }
+    
+            c(&encoded_buf, block.height);
+        }
+    
+        if let Err(e) = ftx.send(Ok(())) {
+            eprintln!("Error sending completion signal: {:?}", e);
+        }
+    });
+    
     let mut response = client.get_block_range(request).await?.into_inner();
-    //println!("{:?}", response);
+
     while let Some(block) = response.message().await? {
-        tx.send(Some(block)).unwrap();
+        if let Err(e) = tx.send(Some(block)) {
+            eprintln!("Error sending block to channel: {:?}", e);
+            break;
+        }
+    }
+
+    if let Err(e) = tx.send(None) {
+        eprintln!("Error sending end signal to channel: {:?}", e);
     }
-    tx.send(None).unwrap();
 
-    // Wait for the processor to exit
     frx.iter().take(1).collect::<Result<Vec<()>, String>>()?;
 
     Ok(())
 }
 
+
 pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, pool: ThreadPool, c: F) -> Result<(), String>
     where F : Fn(&[u8], u64) {
     
@@ -151,7 +163,6 @@ pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_heig
         Err(e) => {
             let es = format!("Error creating runtime {:?}", e);
             error!("{}", es);
-            eprintln!("{}", e);
             return Err(es);
         }
     };
@@ -161,31 +172,49 @@ pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_heig
         Err(e) => {
             let e = format!("Error fetching blocks {:?}", e);
             error!("{}", e);
-            eprintln!("{}", e);
             Err(e)
         }
     }
 }
 
-
 // get_address_txids GRPC call
-async fn get_address_txids<F : 'static + std::marker::Send>(uri: &http::Uri, address: String, 
-        start_height: u64, end_height: u64, no_cert: bool, c: F) -> Result<(), Box<dyn std::error::Error>>
-    where F : Fn(&[u8], u64) {
+async fn get_address_txids<F : 'static + std::marker::Send>(
+    uri: &http::Uri, 
+    address: String, 
+    start_height: u64, 
+    end_height: u64, 
+    no_cert: bool, 
+    c: F
+) -> Result<(), Box<dyn std::error::Error>>
+where F : Fn(&[u8], u64) {
+
+    let mut client = match get_client(uri, no_cert).await {
+        Ok(client) => client,
+        Err(e) => {
+            eprintln!("Error creating client: {:?}", e);
+            return Err(e.into());
+        }
+    };
 
-    let mut client = get_client(uri, no_cert).await?;
     let start = Some(BlockId{ height: start_height, hash: vec!()});
-    let end   = Some(BlockId{ height: end_height,   hash: vec!()});
+    let end = Some(BlockId{ height: end_height, hash: vec!()});
 
-    let request = Request::new(TransparentAddressBlockFilter{ address, range: Some(BlockRange{start, end}) });
+    let request = Request::new(TransparentAddressBlockFilter{ address, range: Some(BlockRange{ start, end }) });
+
+    let maybe_response = match client.get_address_txids(request).await {
+        Ok(response) => response,
+        Err(e) => {
+            eprintln!("Error getting address txids: {:?}", e);
+            return Err(e.into());
+        }
+    };
 
-    let maybe_response = client.get_address_txids(request).await?;
     let mut response = maybe_response.into_inner();
 
     while let Some(tx) = response.message().await? {
         c(&tx.data, tx.height);
     }
-
+    
     Ok(())
 }
 
@@ -223,16 +252,21 @@ where
     Ok(())
 }
 
-pub fn fetch_transparent_txids<F : 'static + std::marker::Send>(uri: &http::Uri, address: String, 
-        start_height: u64, end_height: u64, no_cert: bool, c: F) -> Result<(), String>
-    where F : Fn(&[u8], u64) {
-    
+pub fn fetch_transparent_txids<F : 'static + std::marker::Send>(
+    uri: &http::Uri, 
+    address: String, 
+    start_height: u64, 
+    end_height: u64, 
+    no_cert: bool, 
+    c: F
+) -> Result<(), String>
+where F : Fn(&[u8], u64) {
+
     let mut rt = match tokio::runtime::Runtime::new() {
         Ok(r) => r,
         Err(e) => {
             let e = format!("Error creating runtime {:?}", e);
             error!("{}", e);
-            eprintln!("{}", e);
             return Err(e);
         }
     };
@@ -242,13 +276,11 @@ pub fn fetch_transparent_txids<F : 'static + std::marker::Send>(uri: &http::Uri,
         Err(e) => {
             let e = format!("Error with get_address_txids runtime {:?}", e);
             error!("{}", e);
-            eprintln!("{}", e);
-            Err(e)
+            return Err(e)
         }
     }
 }
 
-
 // get_transaction GRPC call
 async fn get_transaction(uri: &http::Uri, txid: TxId, no_cert: bool) 
     -> Result<RawTransaction, Box<dyn std::error::Error>> {
@@ -266,7 +298,6 @@ pub fn fetch_full_tx(uri: &http::Uri, txid: TxId, no_cert: bool) -> Result<Vec<u
         Err(e) => {
             let errstr = format!("Error creating runtime {}", e.to_string());
             error!("{}", errstr);
-            eprintln!("{}", errstr);
             return Err(errstr);
         }
     };
@@ -276,7 +307,6 @@ pub fn fetch_full_tx(uri: &http::Uri, txid: TxId, no_cert: bool) -> Result<Vec<u
         Err(e) => {
             let errstr = format!("Error in get_transaction runtime {}", e.to_string());
             error!("{}", errstr);
-            eprintln!("{}", errstr);
             Err(errstr)
         }
     }    

lib/src/lightclient.rs

@@ -1233,7 +1233,7 @@ pub fn start_mempool_monitor(lc: Arc<LightClient>) -> Result<(), String> {
                     }
                     // Sleep exponentially backing off
                     std::thread::sleep(std::time::Duration::from_secs((2 as u64).pow(retry_count)));
-                    println!("Sync error {}\nRetry count {}", e, retry_count);
+                   // println!("Sync error {}\nRetry count {}", e, retry_count);
                 }
             }
         }
@@ -1477,47 +1477,83 @@ pub fn start_mempool_monitor(lc: Arc<LightClient>) -> Result<(), String> {
             }        
         }
     }
-
-    fn scan_taddress_txids(&self, pool: &ThreadPool, block_times: Arc<RwLock<HashMap<u64, u32>>>, start_height: u64, end_height: u64, no_cert: bool) -> Result<Vec<()>, String> {
-        // Copy over addresses so as to not lock up the wallet, which we'll use inside the callback below. 
-        let addresses =  self.wallet.read().unwrap()
+    fn scan_taddress_txids(
+        &self, 
+        pool: &ThreadPool, 
+        block_times: Arc<RwLock<HashMap<u64, u32>>>, 
+        start_height: u64, 
+        end_height: u64, 
+        no_cert: bool
+    ) -> Result<Vec<()>, String> {
+        let addresses = self.wallet.read()
+            .map_err(|e| format!("Failed to read wallet: {:?}", e))?
             .get_all_taddresses().iter()
-            .map(|a| a.clone())
+            .cloned()
             .collect::<Vec<String>>();
-
-        // Create a channel so the fetch_transparent_txids can send the results back
+    
         let (ctx, crx) = channel();
         let num_addresses = addresses.len();
-
+    
         for address in addresses {
+            let address_clone = address.clone();
             let wallet = self.wallet.clone();
-
             let pool = pool.clone();
             let server_uri = self.get_server_uri();
             let ctx = ctx.clone();
-
             let block_times = block_times.clone();
-
+    
             pool.execute(move || {
-                // Fetch the transparent transactions for this address, and send the results 
-                // via the channel
-                let r = fetch_transparent_txids(&server_uri, address, start_height, end_height,no_cert,
-                move |tx_bytes: &[u8], height: u64| {
-                    let tx = Transaction::read(tx_bytes).unwrap();
-
-                    // Scan this Tx for transparent inputs and outputs
-                    let datetime = block_times.read().unwrap().get(&height).map(|v| *v).unwrap_or(0);
-                    wallet.read().unwrap().scan_full_tx(&tx, height as i32, datetime as u64); 
-                });
-                ctx.send(r).unwrap();
+              
+                let r = fetch_transparent_txids(
+                    &server_uri, 
+                    address, 
+                    start_height, 
+                    end_height, 
+                    no_cert,
+                    move |tx_bytes: &[u8], height: u64| {
+                        let tx_result = Transaction::read(tx_bytes)
+                            .map_err(|e| format!("Failed to read transaction: {:?}", e));
+    
+                        match tx_result {
+                            Ok(tx) => {
+                                let datetime_result = block_times.read()
+                                    .map_err(|e| format!("Failed to read block times: {:?}", e))
+                                    .and_then(|bt| bt.get(&height).cloned().ok_or_else(|| format!("No datetime for height: {}", height)));
+    
+                                match datetime_result {
+                                    Ok(datetime) => {
+                                        match wallet.read().map_err(|e| format!("Failed to read wallet: {:?}", e)) {
+                                            Ok(w) => {
+                                                w.scan_full_tx(&tx, height as i32, datetime as u64);
+                                            },
+                                            Err(e) => {
+                                                println!("Error reading wallet: {}", e);
+                                            },
+                                        }
+                                    },
+                                    Err(e) => {
+                                        println!("Error processing transaction: {}", e);
+                                    },
+                                }
+                            },
+                            Err(e) => {
+                                println!("Error reading transaction: {}", e);
+                            },
+                        }
+                    }
+                );
+    
+                match ctx.send(r) {
+                    Ok(_) => info!("Successfully sent data for address: {}", address_clone),
+                    Err(e) => println!("Failed to send data for address: {}: {:?}", address_clone, e),
+                }
             });
         }
-
-        // Collect all results from the transparent fetches, and make sure everything was OK. 
-        // If it was not, we return an error, which will go back to the retry
-        crx.iter().take(num_addresses).collect::<Result<Vec<()>, String>>()
-    } 
-
+    
+        crx.iter().take(num_addresses).collect()
+    }
+    
+ 
     fn scan_fill_fulltxs(&self, pool: &ThreadPool, decoy_txids: Vec<(TxId, i32)>) -> Result<Vec<()>, String> {
 
         // We need to first copy over the Txids from the wallet struct, because

lib/src/lightwallet.rs

@@ -1470,26 +1470,43 @@ pub fn scan_full_tx(&self, tx: &Transaction, height: i32, datetime: u64) {
                         // Do it in a short scope because of the write lock.
                         {
                             info!("A sapling output was sent in {}", tx.txid());
-                            let mut txs = self.txs.write().unwrap();
-                            if txs.get(&tx.txid()).unwrap().outgoing_metadata.iter()
-                                    .find(|om| om.address == address && om.value == note.value  && om.memo == memo)
-                                    .is_some() {
-                                warn!("Duplicate outgoing metadata");
-                                continue;
-                            }
+                            match self.txs.write() {
+                                Ok(mut txs) => {
+                                    match txs.get(&tx.txid()) {
+                                        Some(wtx) => {
+                                            if wtx.outgoing_metadata.iter()
+                                                .any(|om| om.address == address && om.value == note.value && om.memo == memo)
+                                            {
+                                                warn!("Duplicate outgoing metadata");
+                                                continue;
+                                            }
                             
-                            // Write the outgoing metadata
-                            txs.get_mut(&tx.txid()).unwrap()
-                                .outgoing_metadata
-                                .push(OutgoingTxMetadata{
-                                    address, value: note.value, memo,
-                                });
+                                            // Write the outgoing metadata
+                                            txs.get_mut(&tx.txid()).unwrap()
+                                                .outgoing_metadata
+                                                .push(OutgoingTxMetadata {
+                                                    address,
+                                                    value: note.value,
+                                                    memo,
+                                                });
+                                        },
+                                        None => {
+                                            error!("Can not find any entry for txid : {}", tx.txid());
+                                            continue; 
+                                        }
+                                    }
+                                },
+                                Err(poisoned) => {
+                                    error!("Lock is poisoned: {}", poisoned);
+                                    return; 
+                                }
+                            }
                         }
                     },
                     None => {}
             };
         }
-    }
+    }                            
     // Mark this Tx as scanned
     {
         let mut txs = self.txs.write().unwrap();

lib/src/lightwallet/data.rs

@@ -440,7 +440,7 @@ pub struct WalletTx {
 
 impl WalletTx {
     pub fn serialized_version() -> u64 {
-        return 4;
+        return 5;
     }
 
     pub fn new(height: i32, datetime: u64, txid: &TxId) -> Self {
@@ -460,10 +460,9 @@ impl WalletTx {
 
     pub fn read<R: Read>(mut reader: R) -> io::Result<Self> {
         let version = reader.read_u64::<LittleEndian>()?;
-        assert!(version <= WalletTx::serialized_version());
+        assert!(version <= WalletTx::serialized_version(), "Version mismatch. Please restore with your Seed");
 
         let block = reader.read_i32::<LittleEndian>()?;
-
         let datetime = if version >= 4 {
             reader.read_u64::<LittleEndian>()?
         } else {
@@ -472,23 +471,24 @@ impl WalletTx {
 
         let mut txid_bytes = [0u8; 32];
         reader.read_exact(&mut txid_bytes)?;
-
         let txid = TxId{0: txid_bytes};
 
         let notes = Vector::read(&mut reader, |r| SaplingNoteData::read(r))?;
         let utxos = Vector::read(&mut reader, |r| Utxo::read(r))?;
-
         let total_shielded_value_spent = reader.read_u64::<LittleEndian>()?;
         let total_transparent_value_spent = reader.read_u64::<LittleEndian>()?;
-
-        // Outgoing metadata was only added in version 2
         let outgoing_metadata = Vector::read(&mut reader, |r| OutgoingTxMetadata::read(r))?;
 
-        let incoming_metadata = Vector::read(&mut reader, |r| IncomingTxMetadata::read(r))?;
+        // Read incoming_metadata only if version is 5 or higher
+        let incoming_metadata = if version >= 5 {
+            Vector::read(&mut reader, |r| IncomingTxMetadata::read(r))?
+        } else {
+            vec![]
+        };
 
         let full_tx_scanned = reader.read_u8()? > 0;
-            
-        Ok(WalletTx{
+
+        Ok(WalletTx {
             block,
             datetime,
             txid,
@@ -502,6 +502,7 @@ impl WalletTx {
         })
     }
 
+
     pub fn write<W: Write>(&self, mut writer: W) -> io::Result<()> {
         writer.write_u64::<LittleEndian>(WalletTx::serialized_version())?;