Merge pull request #3578 from anoma/tiago+jacob/masp-ss-client-phase-3

Shielded sync phase 3 client

.changelog/unreleased/SDK/3578-masp-ss-client-phase-3.md

@@ -0,0 +1,2 @@
+- Parallelize the shielded sync implementation in Namada.
+  ([\#3578](https://github.com/anoma/namada/pull/3578))

Cargo.toml

@@ -109,6 +109,7 @@ expectrl = "0.7.0"
 eyre = "0.6.12"
 fd-lock = "3.0.12"
 flate2 = "1.0.22"
+flume = "0.11.0"
 fs_extra = "1.2.0"
 futures = "0.3"
 git2 = { version = "0.18.1", default-features = false }
@@ -118,9 +119,11 @@ ibc-testkit = { git = "https://github.com/heliaxdev/cosmos-ibc-rs", rev = "1dd9b
 ics23 = "0.11.0"
 index-set = { git = "https://github.com/heliaxdev/index-set", tag = "v0.8.1", features = ["serialize-borsh", "serialize-serde"] }
 indexmap = { git = "https://github.com/heliaxdev/indexmap", tag = "2.2.4-heliax-1", features = ["borsh-schema", "serde"] }
+init-once = "0.6.0"
 itertools = "0.12.1"
 jubjub = "0.10"
 k256 = { version = "0.13.0", default-features = false, features = ["ecdsa", "pkcs8", "precomputed-tables", "serde", "std"]}
+kdam = "0.5.2"
 konst = { version = "0.3.8", default-features = false }
 lazy_static = "1.4.0"
 # TODO: upstreamed in https://github.com/ledger-community/rust-ledger/pull/9
@@ -152,7 +155,7 @@ prost = "0.12.0"
 prost-types = "0.12.0"
 rand = {version = "0.8", default-features = false}
 rand_core = {version = "0.6", default-features = false}
-rayon = "=1.5.3"
+rayon = "1.5.3"
 regex = "1.10.2"
 reqwest = "0.11.4"
 ripemd = "0.1"
@@ -194,6 +197,7 @@ tracing = "0.1.30"
 tracing-appender = "0.2.2"
 tracing-log = "0.2.0"
 tracing-subscriber = {version = "0.3.7", default-features = false, features = ["env-filter", "fmt"]}
+typed-builder = "0.19.1"
 wasmparser = "0.107.0"
 wasm-instrument = {version = "0.4.0", features = ["sign_ext"]}
 wasmer = "4.3.5"
@@ -204,6 +208,7 @@ wasmer-types = "4.3.5"
 wasmer-vm = "4.3.5"
 wasmtimer = "0.2.0"
 winapi = "0.3.9"
+xorf = { version = "0.11.0", features = ["serde"] }
 yansi = "0.5.1"
 zeroize = { version = "1.5.5", features = ["zeroize_derive"] }
 

crates/apps_lib/Cargo.toml

@@ -54,6 +54,7 @@ fd-lock.workspace = true
 flate2.workspace = true
 futures.workspace = true
 itertools.workspace = true
+kdam.workspace = true
 lazy_static = { workspace = true, optional = true }
 linkme = { workspace = true, optional = true }
 ledger-lib = { workspace = true }

crates/apps_lib/src/cli.rs

@@ -3165,6 +3165,7 @@ pub mod args {
     use namada_sdk::ibc::core::host::types::identifiers::{ChannelId, PortId};
     use namada_sdk::keccak::KeccakHash;
     use namada_sdk::key::*;
+    use namada_sdk::masp::utils::RetryStrategy;
     use namada_sdk::masp::{MaspEpoch, PaymentAddress};
     use namada_sdk::storage::{self, BlockHeight, Epoch};
     use namada_sdk::time::DateTimeUtc;
@@ -3214,8 +3215,6 @@ pub mod args {
     pub const BIRTHDAY: ArgOpt<BlockHeight> = arg_opt("birthday");
     pub const BLOCK_HEIGHT: Arg<BlockHeight> = arg("block-height");
     pub const BLOCK_HEIGHT_OPT: ArgOpt<BlockHeight> = arg_opt("height");
-    pub const BLOCK_HEIGHT_FROM_OPT: ArgOpt<BlockHeight> =
-        arg_opt("from-height");
     pub const BLOCK_HEIGHT_TO_OPT: ArgOpt<BlockHeight> = arg_opt("to-height");
     pub const BRIDGE_POOL_GAS_AMOUNT: ArgDefault<token::DenominatedAmount> =
         arg_default(
@@ -3353,6 +3352,8 @@ pub mod args {
     pub const MASP_EPOCH: ArgOpt<MaspEpoch> = arg_opt("masp-epoch");
     pub const MAX_COMMISSION_RATE_CHANGE: Arg<Dec> =
         arg("max-commission-rate-change");
+    pub const MAX_CONCURRENT_FETCHES: ArgDefault<usize> =
+        arg_default("max-concurrent-fetches", DefaultFn(|| 100));
     pub const MAX_ETH_GAS: ArgOpt<u64> = arg_opt("max_eth-gas");
     pub const MEMO_OPT: ArgOpt<String> = arg_opt("memo");
     pub const MIGRATION_PATH: ArgOpt<PathBuf> = arg_opt("migration-path");
@@ -3406,7 +3407,7 @@ pub mod args {
     pub const REFUND_TARGET: ArgOpt<WalletTransferTarget> =
         arg_opt("refund-target");
     pub const RELAYER: Arg<Address> = arg("relayer");
-    pub const SAFE_MODE: ArgFlag = flag("safe-mode");
+    pub const RETRIES: ArgOpt<u64> = arg_opt("retries");
     pub const SCHEME: ArgDefault<SchemeType> =
         arg_default("scheme", DefaultFn(|| SchemeType::Ed25519));
     pub const SHELL: Arg<Shell> = arg("shell");
@@ -3466,6 +3467,8 @@ pub mod args {
     pub const VIEWING_KEYS: ArgMulti<WalletViewingKey, GlobStar> =
         arg_multi("viewing-keys");
     pub const VP: ArgOpt<String> = arg_opt("vp");
+    pub const WAIT_FOR_LAST_QUERY_HEIGHT: ArgFlag =
+        flag("wait-for-last-query-height");
     pub const WALLET_ALIAS_FORCE: ArgFlag = flag("wallet-alias-force");
     pub const WASM_CHECKSUMS_PATH: Arg<PathBuf> = arg("wasm-checksums-path");
     pub const WASM_DIR: ArgOpt<PathBuf> = arg_opt("wasm-dir");
@@ -4099,14 +4102,12 @@ pub mod args {
                 gas_price: self.gas_price,
                 eth_addr: self.eth_addr,
                 sync: self.sync,
-                safe_mode: self.safe_mode,
             }
         }
     }
 
     impl Args for RelayBridgePoolProof<CliTypes> {
         fn parse(matches: &ArgMatches) -> Self {
-            let safe_mode = SAFE_MODE.parse(matches);
             let ledger_address = LEDGER_ADDRESS.parse(matches);
             let hashes = HASH_LIST.parse(matches);
             let relayer = RELAYER.parse(matches);
@@ -4137,16 +4138,11 @@ pub mod args {
                 eth_rpc_endpoint,
                 eth_addr,
                 confirmations,
-                safe_mode,
             }
         }
 
         fn def(app: App) -> App {
             app.arg(LEDGER_ADDRESS.def().help(LEDGER_ADDRESS_ABOUT))
-                .arg(SAFE_MODE.def().help(wrap!(
-                    "Safe mode overrides keyboard interrupt signals, to \
-                     ensure Ethereum transfers aren't canceled midway through."
-                )))
                 .arg(HASH_LIST.def().help(wrap!(
                     "Whitespace separated Keccak hash list of transfers in \
                      the Bridge pool."
@@ -4282,15 +4278,13 @@ pub mod args {
                 sync: self.sync,
                 retry_dur: self.retry_dur,
                 success_dur: self.success_dur,
-                safe_mode: self.safe_mode,
             }
         }
     }
 
     impl Args for ValidatorSetUpdateRelay<CliTypes> {
         fn parse(matches: &ArgMatches) -> Self {
             let ledger_address = LEDGER_ADDRESS.parse(matches);
-            let safe_mode = SAFE_MODE.parse(matches);
             let daemon = DAEMON_MODE.parse(matches);
             let epoch = EPOCH.parse(matches);
             let gas = ETH_GAS.parse(matches);
@@ -4315,16 +4309,11 @@ pub mod args {
                 eth_addr,
                 retry_dur,
                 success_dur,
-                safe_mode,
             }
         }
 
         fn def(app: App) -> App {
             app.arg(LEDGER_ADDRESS.def().help(LEDGER_ADDRESS_ABOUT))
-                .arg(SAFE_MODE.def().help(wrap!(
-                    "Safe mode overrides keyboard interrupt signals, to \
-                     ensure Ethereum transfers aren't canceled midway through."
-                )))
                 .arg(DAEMON_MODE.def().help(wrap!(
                     "Run in daemon mode, which will continuously perform \
                      validator set updates."
@@ -6601,18 +6590,26 @@ pub mod args {
     impl Args for ShieldedSync<CliTypes> {
         fn parse(matches: &ArgMatches) -> Self {
             let ledger_address = CONFIG_RPC_LEDGER_ADDRESS.parse(matches);
-            let start_query_height = BLOCK_HEIGHT_FROM_OPT.parse(matches);
             let last_query_height = BLOCK_HEIGHT_TO_OPT.parse(matches);
             let spending_keys = DATED_SPENDING_KEYS.parse(matches);
             let viewing_keys = DATED_VIEWING_KEYS.parse(matches);
             let with_indexer = WITH_INDEXER.parse(matches);
+            let wait_for_last_query_height =
+                WAIT_FOR_LAST_QUERY_HEIGHT.parse(matches);
+            let max_concurrent_fetches = MAX_CONCURRENT_FETCHES.parse(matches);
+            let retry_strategy = match RETRIES.parse(matches) {
+                Some(times) => RetryStrategy::Times(times),
+                None => RetryStrategy::Forever,
+            };
             Self {
                 ledger_address,
-                start_query_height,
                 last_query_height,
                 spending_keys,
                 viewing_keys,
                 with_indexer,
+                wait_for_last_query_height,
+                max_concurrent_fetches,
+                retry_strategy,
             }
         }
 
@@ -6621,11 +6618,6 @@ pub mod args {
                 .arg(BLOCK_HEIGHT_TO_OPT.def().help(wrap!(
                     "Option block height to sync up to. Default is latest."
                 )))
-                .arg(
-                    BLOCK_HEIGHT_FROM_OPT
-                        .def()
-                        .help(wrap!("Option block height to sync from.")),
-                )
                 .arg(DATED_SPENDING_KEYS.def().help(wrap!(
                     "List of new spending keys with which to check note \
                      ownership. These will be added to the shielded context. \
@@ -6643,6 +6635,18 @@ pub mod args {
                      present, the shielded sync will be performed using data \
                      retrieved from the given indexer."
                 )))
+                .arg(WAIT_FOR_LAST_QUERY_HEIGHT.def().help(wrap!(
+                    "Wait until the last height to sync is available instead \
+                     of returning early from the shielded sync."
+                )))
+                .arg(MAX_CONCURRENT_FETCHES.def().help(wrap!(
+                    "Maximum number of fetch jobs that will ever execute \
+                     concurrently during the shielded sync."
+                )))
+                .arg(RETRIES.def().help(wrap!(
+                    "Maximum number of times to retry fetching. If no \
+                     argument is provided, defaults to retrying forever."
+                )))
         }
     }
 
@@ -6656,8 +6660,9 @@ pub mod args {
             let chain_ctx = ctx.borrow_mut_chain_or_exit();
 
             Ok(ShieldedSync {
+                max_concurrent_fetches: self.max_concurrent_fetches,
+                wait_for_last_query_height: self.wait_for_last_query_height,
                 ledger_address: chain_ctx.get(&self.ledger_address),
-                start_query_height: self.start_query_height,
                 last_query_height: self.last_query_height,
                 spending_keys: self
                     .spending_keys
@@ -6669,7 +6674,8 @@ pub mod args {
                     .iter()
                     .map(|vk| chain_ctx.get_cached(vk))
                     .collect(),
-                with_indexer: self.with_indexer.map(|_| ()),
+                with_indexer: self.with_indexer,
+                retry_strategy: self.retry_strategy,
             })
         }
     }

crates/apps_lib/src/cli/api.rs

@@ -8,7 +8,7 @@ use crate::facade::tendermint_rpc::{HttpClient, Url as TendermintUrl};
 
 /// Trait for clients that can be used with the CLI.
 #[async_trait::async_trait(?Send)]
-pub trait CliClient: Client + Sync {
+pub trait CliClient: Client + Send + Sync + 'static {
     fn from_tendermint_address(address: &TendermintUrl) -> Self;
     async fn wait_until_node_is_synced(
         &self,