Phase 2 of Shielded Sync refactor

.changelog/unreleased/improvements/3405-phase-1-shielded-sync-refactor.md

@@ -0,0 +1,7 @@
+ - Implements phase 1 of Issue [\#3385](https://github.com/anoma/namada/issues/3385)
+   - When fetching notes, connections and related failures should not halt shielded sync. Instead, the process
+     should be restarted
+   - If fetching is interrupted, the data fetched should be persisted locally so that progress isn't lost.
+   -  A trait for fetching behavior should be added to provide modularity
+
+      ([\#3498](https://github.com/anoma/namada/pull/3498))

.changelog/unreleased/improvements/3456-masp-ss-phase-2.md

@@ -0,0 +1,4 @@
+- Implement the phase 2 masp shielded sync client. This client uses
+  a [`namada-masp-indexer`](https://github.com/anoma/namada-masp-
+  indexer) instance to query the state of the shielded context.
+  ([\#3456](https://github.com/anoma/namada/pull/3456))

crates/apps_lib/src/cli.rs

@@ -3213,8 +3213,6 @@ pub mod args {
             Err(_) => config::get_default_namada_folder(),
         }),
     );
-    pub const BATCH_SIZE_OPT: ArgDefault<u64> =
-        arg_default("batch-size", DefaultFn(|| 1));
     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> =
@@ -3468,6 +3466,7 @@ pub mod args {
     pub const WASM_CHECKSUMS_PATH: Arg<PathBuf> = arg("wasm-checksums-path");
     pub const WASM_DIR: ArgOpt<PathBuf> = arg_opt("wasm-dir");
     pub const WEBSITE_OPT: ArgOpt<String> = arg_opt("website");
+    pub const WITH_INDEXER: ArgOpt<String> = arg_opt("with-indexer");
     pub const TX_PATH: Arg<PathBuf> = arg("tx-path");
     pub const TX_PATH_OPT: ArgOpt<PathBuf> = TX_PATH.opt();
 
@@ -6575,27 +6574,23 @@ pub mod args {
     impl Args for ShieldedSync<CliTypes> {
         fn parse(matches: &ArgMatches) -> Self {
             let ledger_address = CONFIG_RPC_LEDGER_ADDRESS.parse(matches);
-            let batch_size = BATCH_SIZE_OPT.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 = SPENDING_KEYS.parse(matches);
             let viewing_keys = VIEWING_KEYS.parse(matches);
+            let with_indexer = WITH_INDEXER.parse(matches);
             Self {
                 ledger_address,
-                batch_size,
                 start_query_height,
                 last_query_height,
                 spending_keys,
                 viewing_keys,
+                with_indexer,
             }
         }
 
         fn def(app: App) -> App {
             app.arg(CONFIG_RPC_LEDGER_ADDRESS.def().help(LEDGER_ADDRESS_ABOUT))
-                .arg(BATCH_SIZE_OPT.def().help(wrap!(
-                    "Optional batch size which determines how many txs to \
-                     fetch before caching locally. Default is 1."
-                )))
                 .arg(BLOCK_HEIGHT_TO_OPT.def().help(wrap!(
                     "Option block height to sync up to. Default is latest."
                 )))
@@ -6612,6 +6607,11 @@ pub mod args {
                     "List of new viewing keys with which to check note \
                      ownership. These will be added to the shielded context."
                 )))
+                .arg(WITH_INDEXER.def().help(wrap!(
+                    "Address of a `namada-masp-indexer` live instance. If \
+                     present, the shielded sync will be performed using data \
+                     retrieved from the given indexer."
+                )))
         }
     }
 
@@ -6626,7 +6626,6 @@ pub mod args {
 
             Ok(ShieldedSync {
                 ledger_address: chain_ctx.get(&self.ledger_address),
-                batch_size: self.batch_size,
                 start_query_height: self.start_query_height,
                 last_query_height: self.last_query_height,
                 spending_keys: self
@@ -6639,6 +6638,7 @@ pub mod args {
                     .iter()
                     .map(|vk| chain_ctx.get_cached(vk))
                     .collect(),
+                with_indexer: self.with_indexer.map(|_| ()),
             })
         }
     }
@@ -6962,6 +6962,7 @@ pub mod args {
         type Data = PathBuf;
         type EthereumAddress = String;
         type Keypair = WalletKeypair;
+        type MaspIndexerAddress = String;
         type PaymentAddress = WalletPaymentAddr;
         type PublicKey = WalletPublicKey;
         type SpendingKey = WalletSpendingKey;

crates/apps_lib/src/cli/client.rs

@@ -12,7 +12,7 @@ use crate::cli::cmds::*;
 use crate::client::{rpc, tx, utils};
 
 impl CliApi {
-    pub async fn handle_client_command<C, IO: Io>(
+    pub async fn handle_client_command<C, IO: Io + Send + Sync>(
         client: Option<C>,
         cmd: cli::NamadaClient,
         io: IO,
@@ -336,13 +336,16 @@ impl CliApi {
                         tx::submit_validator_metadata_change(&namada, args)
                             .await?;
                     }
-                    Sub::ShieldedSync(ShieldedSync(args)) => {
+                    Sub::ShieldedSync(ShieldedSync(mut args)) => {
+                        let indexer_addr = args.with_indexer.take();
                         let args = args.to_sdk(&mut ctx)?;
                         let chain_ctx = ctx.take_chain_or_exit();
                         let client = client.unwrap_or_else(|| {
                             C::from_tendermint_address(&args.ledger_address)
                         });
-                        client.wait_until_node_is_synced(&io).await?;
+                        if indexer_addr.is_none() {
+                            client.wait_until_node_is_synced(&io).await?;
+                        }
                         let vks = chain_ctx
                             .wallet
                             .get_viewing_keys()
@@ -361,8 +364,8 @@ impl CliApi {
                         crate::client::masp::syncing(
                             chain_ctx.shielded,
                             &client,
+                            indexer_addr.as_ref().map(|s| s.as_ref()),
                             &io,
-                            args.batch_size,
                             args.start_query_height,
                             args.last_query_height,
                             &sks,