Client masp txs pre-caching

.changelog/unreleased/improvements/2458-masp-scanning.md

@@ -0,0 +1,2 @@
+- Simplified the transaction fetching algorithm to enable it to be saved to
+  storage more frequently. ([\#2458](https://github.com/anoma/namada/pull/2458))

Makefile

@@ -193,7 +193,7 @@ test-integration-save-proofs:
 # Run integration tests without specifying any pre-built MASP proofs option
 test-integration-slow:
 	RUST_BACKTRACE=$(RUST_BACKTRACE) \
-	$(cargo) +$(nightly) test integration::$(TEST_FILTER)  --features integration \
+	$(cargo) +$(nightly) test $(jobs) integration::$(TEST_FILTER)  --features integration \
 	-Z unstable-options \
 	-- \
 	--test-threads=1 \

crates/apps/src/lib/bench_utils.rs

@@ -117,6 +117,8 @@ const BERTHA_SPENDING_KEY: &str = "bertha_spending";
 
 const FILE_NAME: &str = "shielded.dat";
 const TMP_FILE_NAME: &str = "shielded.tmp";
+const SPECULATIVE_FILE_NAME: &str = "speculative_shielded.dat";
+const SPECULATIVE_TMP_FILE_NAME: &str = "speculative_shielded.tmp";
 
 /// For `tracing_subscriber`, which fails if called more than once in the same
 /// process
@@ -669,6 +671,31 @@ impl ShieldedUtils for BenchShieldedUtils {
         Ok(())
     }
 
+    /// Try to load the last saved speculative shielded context from the given
+    /// context directory. If this fails, then leave the current context
+    /// unchanged.
+    async fn load_speculative<U: ShieldedUtils>(
+        &self,
+        ctx: &mut ShieldedContext<U>,
+    ) -> std::io::Result<()> {
+        // Try to load shielded context from file
+        let mut ctx_file = File::open(
+            self.context_dir
+                .0
+                .path()
+                .to_path_buf()
+                .join(SPECULATIVE_FILE_NAME),
+        )?;
+        let mut bytes = Vec::new();
+        ctx_file.read_to_end(&mut bytes)?;
+        // Fill the supplied context with the deserialized object
+        *ctx = ShieldedContext {
+            utils: ctx.utils.clone(),
+            ..ShieldedContext::<U>::deserialize(&mut &bytes[..])?
+        };
+        Ok(())
+    }
+
     /// Save this shielded context into its associated context directory
     async fn save<U: ShieldedUtils>(
         &self,
@@ -695,12 +722,56 @@ impl ShieldedUtils for BenchShieldedUtils {
         // Atomicity is required to prevent other client instances from reading
         // corrupt data.
         std::fs::rename(
-            tmp_path.clone(),
+            tmp_path,
             self.context_dir.0.path().to_path_buf().join(FILE_NAME),
         )?;
-        // Finally, remove our temporary file to allow future saving of shielded
-        // contexts.
-        std::fs::remove_file(tmp_path)?;
+
+        // Remove the speculative file if present since it's state is
+        // overwritten by the confirmed one we just saved
+        let _ = std::fs::remove_file(SPECULATIVE_FILE_NAME);
+        Ok(())
+    }
+
+    /// Save this speculative shielded context into its associated context
+    /// directory
+    async fn save_speculative<U: ShieldedUtils>(
+        &self,
+        ctx: &ShieldedContext<U>,
+    ) -> std::io::Result<()> {
+        // TODO: use mktemp crate?
+        let tmp_path = self
+            .context_dir
+            .0
+            .path()
+            .to_path_buf()
+            .join(SPECULATIVE_TMP_FILE_NAME);
+        {
+            // First serialize the shielded context into a temporary file.
+            // Inability to create this file implies a simultaneuous write
+            // is in progress. In this case, immediately
+            // fail. This is unproblematic because the data
+            // intended to be stored can always be re-fetched
+            // from the blockchain.
+            let mut ctx_file = OpenOptions::new()
+                .write(true)
+                .create_new(true)
+                .open(tmp_path.clone())?;
+            let mut bytes = Vec::new();
+            ctx.serialize(&mut bytes)
+                .expect("cannot serialize shielded context");
+            ctx_file.write_all(&bytes[..])?;
+        }
+        // Atomically update the old shielded context file with new data.
+        // Atomicity is required to prevent other client instances from
+        // reading corrupt data.
+        std::fs::rename(
+            tmp_path,
+            self.context_dir
+                .0
+                .path()
+                .to_path_buf()
+                .join(SPECULATIVE_FILE_NAME),
+        )?;
         Ok(())
     }
 }
@@ -983,9 +1054,13 @@ impl BenchShieldedCtx {
             .wallet
             .find_spending_key(ALBERT_SPENDING_KEY, None)
             .unwrap();
-        async_runtime
-            .block_on(self.shielded.fetch(
+        self.shielded = async_runtime
+            .block_on(crate::client::masp::syncing(
+                self.shielded,
                 &self.shell,
+                &StdIo,
+                1,
+                None,
                 &[spending_key.into()],
                 &[],
             ))

crates/apps/src/lib/cli.rs

@@ -267,6 +267,7 @@ pub mod cmds {
                 .subcommand(QueryMetaData::def().display_order(5))
                 // Actions
                 .subcommand(SignTx::def().display_order(6))
+                .subcommand(ShieldedSync::def().display_order(6))
                 .subcommand(GenIbcShieldedTransafer::def().display_order(6))
                 // Utils
                 .subcommand(Utils::def().display_order(7))
@@ -346,6 +347,7 @@ pub mod cmds {
             let add_to_eth_bridge_pool =
                 Self::parse_with_ctx(matches, AddToEthBridgePool);
             let sign_tx = Self::parse_with_ctx(matches, SignTx);
+            let shielded_sync = Self::parse_with_ctx(matches, ShieldedSync);
             let gen_ibc_shielded =
                 Self::parse_with_ctx(matches, GenIbcShieldedTransafer);
             let utils = SubCmd::parse(matches).map(Self::WithoutContext);
@@ -397,6 +399,7 @@ pub mod cmds {
                 .or(query_metadata)
                 .or(query_account)
                 .or(sign_tx)
+                .or(shielded_sync)
                 .or(gen_ibc_shielded)
                 .or(utils)
         }
@@ -483,6 +486,7 @@ pub mod cmds {
         QueryValidatorState(QueryValidatorState),
         QueryRewards(QueryRewards),
         SignTx(SignTx),
+        ShieldedSync(ShieldedSync),
         GenIbcShieldedTransafer(GenIbcShieldedTransafer),
     }
 
@@ -1344,6 +1348,29 @@ pub mod cmds {
         }
     }
 
+    #[derive(Clone, Debug)]
+    pub struct ShieldedSync(pub args::ShieldedSync<args::CliTypes>);
+
+    impl SubCmd for ShieldedSync {
+        const CMD: &'static str = "shielded-sync";
+
+        fn parse(matches: &ArgMatches) -> Option<Self> {
+            matches
+                .subcommand_matches(Self::CMD)
+                .map(|matches| ShieldedSync(args::ShieldedSync::parse(matches)))
+        }
+
+        fn def() -> App {
+            App::new(Self::CMD)
+                .about(
+                    "Sync the local shielded context with MASP notes owned by \
+                     the provided viewing / spending keys up to an optional \
+                     specified block height.",
+                )
+                .add_args::<args::ShieldedSync<args::CliTypes>>()
+        }
+    }
+
     #[derive(Clone, Debug)]
     pub struct Bond(pub args::Bond<args::CliTypes>);
 
@@ -2878,6 +2905,8 @@ 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 BRIDGE_POOL_GAS_AMOUNT: ArgDefault<token::DenominatedAmount> =
@@ -3061,6 +3090,8 @@ pub mod args {
     pub const SIGNATURES: ArgMulti<PathBuf, GlobStar> = arg_multi("signatures");
     pub const SOURCE: Arg<WalletAddress> = arg("source");
     pub const SOURCE_OPT: ArgOpt<WalletAddress> = SOURCE.opt();
+    pub const SPENDING_KEYS: ArgMulti<WalletSpendingKey, GlobStar> =
+        arg_multi("spending-keys");
     pub const STEWARD: Arg<WalletAddress> = arg("steward");
     pub const SOURCE_VALIDATOR: Arg<WalletAddress> = arg("source-validator");
     pub const STORAGE_KEY: Arg<storage::Key> = arg("storage-key");
@@ -3097,6 +3128,8 @@ pub mod args {
     pub const VALUE: Arg<String> = arg("value");
     pub const VOTER_OPT: ArgOpt<WalletAddress> = arg_opt("voter");
     pub const VIEWING_KEY: Arg<WalletViewingKey> = arg("key");
+    pub const VIEWING_KEYS: ArgMulti<WalletViewingKey, GlobStar> =
+        arg_multi("viewing-keys");
     pub const VP: ArgOpt<String> = arg_opt("vp");
     pub const WALLET_ALIAS_FORCE: ArgFlag = flag("wallet-alias-force");
     pub const WASM_CHECKSUMS_PATH: Arg<PathBuf> = arg("wasm-checksums-path");
@@ -5608,6 +5641,63 @@ pub mod args {
         }
     }
 
+    impl Args for ShieldedSync<CliTypes> {
+        fn parse(matches: &ArgMatches) -> Self {
+            let ledger_address = LEDGER_ADDRESS.parse(matches);
+            let batch_size = BATCH_SIZE_OPT.parse(matches);
+            let last_query_height = BLOCK_HEIGHT_OPT.parse(matches);
+            let spending_keys = SPENDING_KEYS.parse(matches);
+            let viewing_keys = VIEWING_KEYS.parse(matches);
+            Self {
+                ledger_address,
+                batch_size,
+                last_query_height,
+                spending_keys,
+                viewing_keys,
+            }
+        }
+
+        fn def(app: App) -> App {
+            app.arg(LEDGER_ADDRESS.def().help(LEDGER_ADDRESS_ABOUT))
+                .arg(BATCH_SIZE_OPT.def().help(
+                    "Optional batch size which determines how many txs to \
+                     fetch before caching locally. Default is 1.",
+                ))
+                .arg(BLOCK_HEIGHT_OPT.def().help(
+                    "Option block height to sync up to. Default is latest.",
+                ))
+                .arg(SPENDING_KEYS.def().help(
+                    "List of new spending keys with which to check note \
+                     ownership. These will be added to the shielded context.",
+                ))
+                .arg(VIEWING_KEYS.def().help(
+                    "List of new viewing keys with which to check note \
+                     ownership. These will be added to the shielded context.",
+                ))
+        }
+    }
+
+    impl CliToSdk<ShieldedSync<SdkTypes>> for ShieldedSync<CliTypes> {
+        fn to_sdk(self, ctx: &mut Context) -> ShieldedSync<SdkTypes> {
+            let chain_ctx = ctx.borrow_mut_chain_or_exit();
+            ShieldedSync {
+                ledger_address: self.ledger_address,
+                batch_size: self.batch_size,
+                last_query_height: self.last_query_height,
+                spending_keys: self
+                    .spending_keys
+                    .iter()
+                    .map(|sk| chain_ctx.get_cached(sk))
+                    .collect(),
+                viewing_keys: self
+                    .viewing_keys
+                    .iter()
+                    .map(|vk| chain_ctx.get_cached(vk))
+                    .collect(),
+            }
+        }
+    }
+
     impl CliToSdk<GenIbcShieldedTransafer<SdkTypes>>
         for GenIbcShieldedTransafer<CliTypes>
     {
@@ -5892,6 +5982,7 @@ pub mod args {
         type EthereumAddress = String;
         type Keypair = WalletKeypair;
         type PublicKey = WalletPublicKey;
+        type SpendingKey = WalletSpendingKey;
         type TendermintAddress = tendermint_config::net::Address;
         type TransferSource = WalletTransferSource;
         type TransferTarget = WalletTransferTarget;

crates/apps/src/lib/cli/client.rs

@@ -1,4 +1,5 @@
 use color_eyre::eyre::Result;
+use masp_primitives::zip32::ExtendedFullViewingKey;
 use namada::types::io::Io;
 use namada_sdk::{Namada, NamadaImpl};
 
@@ -315,6 +316,39 @@ impl CliApi {
                         tx::submit_validator_metadata_change(&namada, args)
                             .await?;
                     }
+                    Sub::ShieldedSync(ShieldedSync(args)) => {
+                        let client = client.unwrap_or_else(|| {
+                            C::from_tendermint_address(&args.ledger_address)
+                        });
+                        client.wait_until_node_is_synced(&io).await?;
+                        let args = args.to_sdk(&mut ctx);
+                        let chain_ctx = ctx.take_chain_or_exit();
+                        let vks = chain_ctx
+                            .wallet
+                            .get_viewing_keys()
+                            .values()
+                            .copied()
+                            .map(|vk| ExtendedFullViewingKey::from(vk).fvk.vk)
+                            .chain(args.viewing_keys.into_iter().map(|vk| {
+                                ExtendedFullViewingKey::from(vk).fvk.vk
+                            }))
+                            .collect::<Vec<_>>();
+                        let sks = args
+                            .spending_keys
+                            .into_iter()
+                            .map(|sk| sk.into())
+                            .collect::<Vec<_>>();
+                        crate::client::masp::syncing(
+                            chain_ctx.shielded,
+                            &client,
+                            &io,
+                            args.batch_size,
+                            args.last_query_height,
+                            &sks,
+                            &vks,
+                        )
+                        .await?;
+                    }
                     // Eth bridge
                     Sub::AddToEthBridgePool(args) => {
                         let args = args.0;