Find best server

Cargo.toml

@@ -36,7 +36,7 @@ flexi_logger = {version="0.22.3", features = ["compress"]}
 serde = {version = "1.0.126", features = ["derive"]}
 serde_json = "1.0.64"
 bincode = "1.3.3"
-tokio = { version = "^1.6", features = ["macros", "rt-multi-thread"] }
+tokio = { version = "^1.6", features = ["macros", "rt-multi-thread", "time"] }
 tokio-stream = "0.1.7"
 protobuf = "3.0.2"
 hex = "0.4.3"

binding.h

@@ -11,12 +11,16 @@ typedef char bool;
 #endif
 typedef void *DartPostCObjectFnType;
 
+#define QR_DATA_SIZE 256
+
+#define MAX_OUTPUTS_PER_CHUNK 200000
+
 void dummy_export(void);
 
-void deallocate_str(char *s);
-
 void dart_post_cobject(DartPostCObjectFnType ptr);
 
+void deallocate_str(char *s);
+
 bool get_error(void);
 
 char *get_error_msg(void);
@@ -104,4 +108,6 @@ char *split_data(uint32_t id, char *data);
 
 char *merge_data(char *drop);
 
-char *get_tx_summary(char *drop);
+char *get_tx_summary(char *tx);
+
+char *get_best_server(char **servers, uint32_t count);

src/api/dart_ffi.rs

@@ -553,3 +553,21 @@ pub unsafe extern "C" fn get_tx_summary(tx: *mut c_char) -> *mut c_char {
     };
     to_c_str(log_string(res()))
 }
+
+#[tokio::main]
+#[no_mangle]
+pub async unsafe extern "C" fn get_best_server(
+    servers: *mut *mut c_char,
+    count: u32,
+) -> *mut c_char {
+    let mut cservers = vec![];
+    for i in 0..count {
+        let ptr = *servers.offset(i as isize);
+        let s = CStr::from_ptr(ptr).to_string_lossy();
+        cservers.push(s.to_string());
+    }
+    let best_server = crate::get_best_server(&cservers)
+        .await
+        .unwrap_or(String::new());
+    to_c_str(best_server)
+}

src/chain.rs

@@ -5,13 +5,17 @@ use crate::lw_rpc::compact_tx_streamer_client::CompactTxStreamerClient;
 use crate::lw_rpc::*;
 use crate::scan::{Blocks, MAX_OUTPUTS_PER_CHUNK};
 use ff::PrimeField;
+use futures::{future, FutureExt};
 use log::info;
 use rayon::prelude::*;
 use std::collections::HashMap;
 use std::marker::PhantomData;
+use std::time::Duration;
 use std::time::Instant;
 use thiserror::Error;
 use tokio::sync::mpsc::Sender;
+use tokio::time::timeout;
+use tokio_stream::StreamExt;
 use tonic::transport::{Certificate, Channel, ClientTlsConfig};
 use tonic::Request;
 use zcash_note_encryption::batch::try_compact_note_decryption;
@@ -477,6 +481,29 @@ pub async fn connect_lightwalletd(url: &str) -> anyhow::Result<CompactTxStreamer
     Ok(client)
 }
 
+async fn get_height(server: String) -> Option<(String, u32)> {
+    let mut client = connect_lightwalletd(&server).await.ok()?;
+    let height = get_latest_height(&mut client).await.ok()?;
+    log::info!("{} {}", server, height);
+    Some((server, height))
+}
+
+pub async fn get_best_server(servers: &[String]) -> Option<String> {
+    let mut server_heights = vec![];
+    for s in servers.iter() {
+        let server_height =
+            tokio::spawn(timeout(Duration::from_secs(1), get_height(s.to_string()))).boxed();
+        server_heights.push(server_height);
+    }
+    let server_heights = future::try_join_all(server_heights).await.ok()?;
+    let best_server = server_heights
+        .into_iter()
+        .filter_map(|h| h.unwrap_or(None))
+        .max_by_key(|(_, h)| *h)
+        .map(|x| x.0);
+    best_server
+}
+
 // pub async fn sync(
 //     network: &Network,
 //     vks: HashMap<u32, AccountViewKey>,

src/lib.rs

@@ -62,8 +62,8 @@ pub fn hex_to_hash(hex: &str) -> anyhow::Result<[u8; 32]> {
 
 pub use crate::builder::advance_tree;
 pub use crate::chain::{
-    calculate_tree_state_v2, connect_lightwalletd, download_chain, get_latest_height, ChainError,
-    DecryptNode,
+    calculate_tree_state_v2, connect_lightwalletd, download_chain, get_best_server,
+    get_latest_height, ChainError, DecryptNode,
 };
 pub use crate::coinconfig::{
     init_coin, set_active, set_active_account, set_coin_lwd_url, CoinConfig,

src/main/rpc.rs

@@ -11,7 +11,7 @@ use std::collections::HashMap;
 use thiserror::Error;
 use warp_api_ffi::api::payment::{Recipient, RecipientMemo};
 use warp_api_ffi::api::payment_uri::PaymentURI;
-use warp_api_ffi::{AccountRec, CoinConfig, RaptorQDrops, Tx, TxRec};
+use warp_api_ffi::{get_best_server, AccountRec, CoinConfig, RaptorQDrops, Tx, TxRec};
 
 #[derive(Debug, Error)]
 pub enum Error {
@@ -49,6 +49,15 @@ async fn main() -> anyhow::Result<()> {
     env_logger::init();
     let _ = dotenv::dotenv();
 
+    let server = get_best_server(&[
+        "https://lwdv3.zecwallet.co:443".to_string(),
+        "https://zuul.free2z.cash:9067".to_string(),
+        "https://mainnet.lightwalletd.com:9067".to_string(),
+    ])
+    .await
+    .unwrap();
+    log::info!("Best server = {}", server);
+
     let rocket = rocket::build();
     let figment = rocket.figment();
     let zec: HashMap<String, String> = figment.extract_inner("zec")?;

src/scan.rs

@@ -59,14 +59,7 @@ pub async fn scan_all(network: &Network, fvks: &[ExtendedFullViewingKey]) -> any
     });
 
     let start = Instant::now();
-    download_chain(
-        &mut client,
-        start_height,
-        end_height,
-        None,
-        blocks_tx,
-    )
-    .await?;
+    download_chain(&mut client, start_height, end_height, None, blocks_tx).await?;
     info!("Download chain: {} ms", start.elapsed().as_millis());
 
     info!("Total: {} ms", total_start.elapsed().as_millis());
@@ -326,7 +319,13 @@ pub async fn sync_async(
                 for w in witnesses.iter() {
                     DbAdapter::store_witnesses(&db_transaction, w, height, w.id_note)?;
                 }
-                DbAdapter::store_block(&mut db_transaction, height, &block.hash, block.time, &tree)?;
+                DbAdapter::store_block(
+                    &mut db_transaction,
+                    height,
+                    &block.hash,
+                    block.time,
+                    &tree,
+                )?;
                 db_transaction.commit()?;
             }
         }