Add API endpoint to query specific MASP txs by (block height, block index)

Cargo.toml

@@ -7,7 +7,7 @@ authors = ["Heliax AG <[email protected]>"]
 edition = "2024"
 license = "GPL-3.0-or-later"
 readme = "README.md"
-version = "1.2.1"
+version = "1.3.0"
 
 [workspace.dependencies]
 anyhow = "1.0.75"
@@ -23,9 +23,9 @@ diesel_migrations = { version = "2.2.0", default-features = false, features = [
 futures = "0.3.30"
 itertools = "0.13.0"
 lazy_static = "1.4.0"
-namada_core = { version = "0.47.1" }
-namada_sdk = { version = "0.47.1", default-features = false, features = ["std", "async-send", "download-params"] }
-namada_tx = { version = "0.47.1" }
+namada_sdk = { version = "0.149.1", default-features = false, features = ["std", "async-send", "download-params"] }
+namada_tx = "0.149.1"
+namada_core = "0.149.1"
 orm = { path = "orm" }
 rayon = "1.10.0"
 serde = { version = "1.0.138", features = [ "derive" ] }
@@ -42,7 +42,7 @@ tower-http = { version = "0.4.4", features = [ "compression-full", "limit", "tra
 tracing = "0.1"
 tracing-appender = "0.2.0"
 tracing-subscriber = { version = "0.3", features = [ "env-filter" ] }
-tryhard = { version = "0.5.1" }
 validator = { version = "0.16.0", features = ["derive"] }
-vergen = "8.0.0"
 xorf = { version = "0.11.0", features = ["serde"]}
+tryhard = { version = "0.5.1" }
+vergen = "8.0.0"

chain/Cargo.toml

@@ -26,13 +26,11 @@ orm.workspace = true
 rayon.workspace = true
 shared.workspace = true
 tendermint-rpc.workspace = true
-tendermint.workspace = true
 tokio-retry.workspace = true
 tokio.workspace = true 
 tracing-subscriber.workspace = true
 tracing.workspace = true
 tryhard.workspace = true
-itertools.workspace = true
 
 [build-dependencies]
 vergen = { workspace = true, features = ["build", "git", "gitcl"] }

chain/run.sh

@@ -1 +1 @@
-cargo run --release -- --tendermint-url http://127.0.0.1:27657 --database-url postgres://postgres:[email protected]:5435/namada-masp-indexer --chain-id local.61271083678d6bb47c35eda1 --checksums-filepath ../artifacts/checksums.json
+cargo run --release -- --cometbft-url http://127.0.0.1:27657 --database-url postgres://postgres:[email protected]:5435/masp-indexer

chain/src/entity/commitment_tree.rs

@@ -28,6 +28,10 @@ impl InnerCommitmentTree {
         }
     }
 
+    fn is_dirty(&self) -> bool {
+        self.transactional.is_dirty()
+    }
+
     fn rollback(&mut self) {
         self.transactional.rollback();
     }
@@ -68,6 +72,10 @@ impl CommitmentTree {
         Self(Arc::new(Mutex::new(InnerCommitmentTree::new(tree))))
     }
 
+    pub fn is_dirty(&self) -> bool {
+        self.0.lock().unwrap().is_dirty()
+    }
+
     pub fn rollback(&self) {
         self.0.lock().unwrap().rollback()
     }

chain/src/entity/tx_notes_index.rs

@@ -1,13 +1,13 @@
 use std::collections::BTreeMap;
 
 use orm::notes_index::NotesIndexInsertDb;
-use shared::indexed_tx::IndexedTx;
+use shared::indexed_tx::{IndexedTx, MaspIndexedTx};
 
 #[derive(Default, Clone, Debug)]
-pub struct TxNoteMap(BTreeMap<IndexedTx, usize>);
+pub struct TxNoteMap(BTreeMap<MaspIndexedTx, usize>);
 
 impl TxNoteMap {
-    pub fn insert(&mut self, indexed_tx: IndexedTx, note_pos: usize) {
+    pub fn insert(&mut self, indexed_tx: MaspIndexedTx, note_pos: usize) {
         self.0.insert(indexed_tx, note_pos);
     }
 
@@ -20,18 +20,25 @@ impl TxNoteMap {
             .iter()
             .map(
                 |(
-                    &IndexedTx {
-                        block_height,
-                        block_index,
-                        batch_index,
-                        ..
+                    &MaspIndexedTx {
+                        indexed_tx:
+                            IndexedTx {
+                                block_height,
+                                block_index,
+                                masp_tx_index,
+                            },
+                        kind,
                     },
                     &note_pos,
                 )| NotesIndexInsertDb {
                     block_index: block_index.0 as i32,
                     note_position: note_pos as i32,
                     block_height: block_height.0 as i32,
-                    masp_tx_index: batch_index as i32,
+                    masp_tx_index: masp_tx_index.0 as i32,
+                    is_masp_fee_payment: matches!(
+                        kind,
+                        shared::indexed_tx::MaspTxKind::FeePayment
+                    ),
                 },
             )
             .collect()

chain/src/main.rs

@@ -3,18 +3,17 @@ pub mod config;
 pub mod entity;
 pub mod services;
 
-use std::collections::HashSet;
+use std::collections::BTreeMap;
 use std::env;
 use std::sync::Arc;
 use std::sync::atomic::{self, AtomicBool};
 use std::time::Duration;
 
 use anyhow::Context;
 use clap::Parser;
-use shared::block::Block;
 use shared::error::{IntoMainError, MainError};
 use shared::height::{BlockHeight, FollowingHeights};
-use shared::indexed_tx::IndexedTx;
+use shared::transaction::Transaction;
 use tendermint_rpc::HttpClient;
 use tendermint_rpc::client::CompatMode;
 use tokio::signal;
@@ -218,8 +217,6 @@ async fn build_and_commit_masp_data_at_height(
         return Ok(());
     }
 
-    let start = Instant::now();
-
     // NB: rollback changes from previous failed commit attempts
     witness_map.rollback();
     commitment_tree.rollback();
@@ -242,7 +239,9 @@ async fn build_and_commit_masp_data_at_height(
         return Err(MainError);
     }
 
-    let block_data = {
+    let mut checkpoint = Instant::now();
+
+    let (block_data, num_transactions) = {
         tracing::info!(
             %block_height,
             "Fetching block data from CometBFT"
@@ -251,62 +250,61 @@ async fn build_and_commit_masp_data_at_height(
             cometbft_service::query_masp_txs_in_block(&client, block_height)
                 .await
                 .into_rpc_error()?;
-        tracing::info!(
-            %block_height,
-            "Acquired block data from CometBFT"
-        );
-        block_data
+        with_time_taken(&mut checkpoint, |time_taken| {
+            tracing::info!(
+                time_taken,
+                %block_height,
+                "Acquired block data from CometBFT"
+            );
+        });
+        let num_transactions = block_data.transactions.len();
+        (block_data, num_transactions)
     };
 
-    let mut shielded_txs = Vec::new();
+    let mut shielded_txs = BTreeMap::new();
     let mut tx_notes_index = TxNoteMap::default();
 
-    let ordered_txs =
-        lookup_valid_commitment_tree(&client, &commitment_tree, &block_data)
-            .await?;
-
-    let anything_to_commit = !ordered_txs.is_empty();
-
-    commitment_tree.rollback();
+    tracing::info!(
+        %block_height,
+        num_transactions,
+        "Processing new masp transactions...",
+    );
 
-    for (new_masp_tx_index, mut indexed_tx) in
-        ordered_txs.into_iter().enumerate()
+    for (masp_indexed_tx, Transaction { masp_tx, .. }) in
+        block_data.transactions.into_iter()
     {
-        let masp_tx = block_data.get_masp_tx(indexed_tx).unwrap();
-
-        indexed_tx.masp_tx_index = new_masp_tx_index.into();
-
         masp_service::update_witness_map(
             &commitment_tree,
             &mut tx_notes_index,
             &witness_map,
-            indexed_tx,
-            masp_tx,
+            masp_indexed_tx,
+            &masp_tx,
         )
         .into_masp_error()?;
 
-        shielded_txs.push((indexed_tx, masp_tx.clone()));
+        shielded_txs.insert(masp_indexed_tx, masp_tx);
     }
 
-    if anything_to_commit {
-        masp_service::query_witness_map_anchor_existence(
-            &witness_map,
-            commitment_tree.root(),
-            number_of_witness_map_roots_to_check,
-        )
-        .into_masp_error()?;
-    }
-
-    let first_checkpoint = Instant::now();
+    with_time_taken(&mut checkpoint, |time_taken| {
+        tracing::info!(
+            %block_height,
+            num_transactions,
+            time_taken,
+            "Processed new masp transactions",
+        );
+    });
 
-    tracing::info!(
-        %block_height,
-        num_transactions = block_data.transactions.len(),
-        time_taken = first_checkpoint.duration_since(start).as_secs_f64(),
-        "Processed new masp transactions...",
-    );
+    validate_masp_state(
+        &mut checkpoint,
+        &client,
+        &commitment_tree,
+        &witness_map,
+        number_of_witness_map_roots_to_check,
+    )
+    .await?;
 
     db_service::commit(
+        &mut checkpoint,
         &conn_obj,
         chain_state,
         commitment_tree,
@@ -317,83 +315,60 @@ async fn build_and_commit_masp_data_at_height(
     .await
     .into_db_error()?;
 
-    let second_checkpoint = Instant::now();
-
-    tracing::info!(
-        block_height = %chain_state.block_height,
-        time_taken = second_checkpoint
-            .duration_since(first_checkpoint)
-            .as_secs_f64(),
-        "Committed new block"
-    );
-
     Ok(())
 }
 
-async fn lookup_valid_commitment_tree(
+async fn validate_masp_state(
+    checkpoint: &mut Instant,
     client: &HttpClient,
     commitment_tree: &CommitmentTree,
-    block: &Block,
-) -> Result<Vec<IndexedTx>, MainError> {
-    use itertools::Itertools;
-
-    let all_indexed_txs: Vec<_> = block.indexed_txs().collect();
-
-    let mut correct_order = Vec::with_capacity(all_indexed_txs.len());
-    let mut fee_unshields = HashSet::with_capacity(all_indexed_txs.len());
-
-    // Guess the set of fee unshieldings at the current height
-    let fee_unshield_sets = all_indexed_txs.iter().copied().powerset();
-
-    for fee_unshield_set in fee_unshield_sets {
-        // Start a new attempt at guessing the root of
-        // the commitment tree
-        commitment_tree.rollback();
-        correct_order.clear();
-        fee_unshields.clear();
-
-        tracing::info!(
-            ?fee_unshield_set,
-            "Checking subset of masp fee unshields to build cmt tree"
-        );
-
-        for indexed_tx in fee_unshield_set {
-            let masp_tx = block.get_masp_tx(indexed_tx).unwrap();
+    witness_map: &WitnessMap,
+    number_of_witness_map_roots_to_check: usize,
+) -> Result<(), MainError> {
+    if commitment_tree.is_dirty() && number_of_witness_map_roots_to_check > 0 {
+        tracing::info!("Validating MASP state...");
 
-            masp_service::update_commitment_tree(commitment_tree, masp_tx)
-                .into_masp_error()?;
+        let tree_root = tokio::task::block_in_place(|| commitment_tree.root());
 
-            correct_order.push(indexed_tx);
-            fee_unshields.insert(indexed_tx);
-        }
+        let commitment_tree_check_fut = async {
+            cometbft_service::query_commitment_tree_anchor_existence(
+                client, tree_root,
+            )
+            .await
+            .into_rpc_error()
+        };
+
+        let witness_map = witness_map.clone();
+        let witness_map_check_fut = async move {
+            tokio::task::spawn_blocking(move || {
+                masp_service::query_witness_map_anchor_existence(
+                    &witness_map,
+                    tree_root,
+                    number_of_witness_map_roots_to_check,
+                )
+                .into_masp_error()
+            })
+            .await
+            .context("Failed to join Tokio task")
+            .into_tokio_join_error()?
+        };
 
-        for indexed_tx in all_indexed_txs
-            .iter()
-            .copied()
-            // We filter fee unshields out of this loop
-            .filter(|indexed_tx| !fee_unshields.contains(indexed_tx))
-        {
-            let masp_tx = block.get_masp_tx(indexed_tx).unwrap();
+        futures::try_join!(commitment_tree_check_fut, witness_map_check_fut)?;
 
-            masp_service::update_commitment_tree(commitment_tree, masp_tx)
-                .into_masp_error()?;
+        with_time_taken(checkpoint, |time_taken| {
+            tracing::info!(time_taken, "Validated MASP state");
+        });
+    }
 
-            correct_order.push(indexed_tx);
-        }
+    Ok(())
+}
 
-        if cometbft_service::query_commitment_tree_anchor_existence(
-            client,
-            commitment_tree.root(),
-        )
-        .await
-        .into_masp_error()?
-        {
-            return Ok(correct_order);
-        }
-    }
+fn with_time_taken<F, T>(checkpoint: &mut Instant, callback: F) -> T
+where
+    F: FnOnce(f64) -> T,
+{
+    let last_checkpoint = std::mem::replace(checkpoint, Instant::now());
+    let time_taken = last_checkpoint.elapsed().as_secs_f64();
 
-    Err(anyhow::anyhow!(
-        "Couldn't find a valid permutation of fee unshieldings"
-    ))
-    .into_masp_error()
+    callback(time_taken)
 }