Merge branch 'main' into marvin/nonce

Cargo.lock

@@ -335,6 +335,7 @@ name = "amm_program"
 version = "0.1.0"
 dependencies = [
  "amm_core",
+ "nssa",
  "nssa_core",
  "token_core",
 ]
@@ -3713,6 +3714,7 @@ dependencies = [
  "serde",
  "serde_json",
  "storage",
+ "tempfile",
  "tokio",
  "url",
 ]
@@ -5530,7 +5532,6 @@ dependencies = [
 name = "nssa"
 version = "0.1.0"
 dependencies = [
- "amm_core",
  "anyhow",
  "borsh",
  "env_logger",

bedrock_client/src/lib.rs

@@ -31,9 +31,9 @@ impl Default for BackoffConfig {
     }
 }
 
-// Simple wrapper
-// maybe extend in the future for our purposes
-// `Clone` is cheap because `CommonHttpClient` is internally reference counted (`Arc`).
+/// Simple wrapper
+/// maybe extend in the future for our purposes
+/// `Clone` is cheap because `CommonHttpClient` is internally reference counted (`Arc`).
 #[derive(Clone)]
 pub struct BedrockClient {
     http_client: CommonHttpClient,
@@ -62,10 +62,22 @@ impl BedrockClient {
         })
     }
 
-    pub async fn post_transaction(&self, tx: SignedMantleTx) -> Result<(), Error> {
-        Retry::spawn(self.backoff_strategy(), || {
-            self.http_client
+    pub async fn post_transaction(&self, tx: SignedMantleTx) -> Result<Result<(), Error>, Error> {
+        Retry::spawn(self.backoff_strategy(), || async {
+            match self
+                .http_client
                 .post_transaction(self.node_url.clone(), tx.clone())
+                .await
+            {
+                Ok(()) => Ok(Ok(())),
+                Err(err) => match err {
+                    // Retry arm.
+                    // Retrying only reqwest errors: mainly connected to http.
+                    Error::Request(_) => Err(err),
+                    // Returning non-retryable error
+                    Error::Server(_) | Error::Client(_) | Error::Url(_) => Ok(Err(err)),
+                },
+            }
         })
         .await
     }

indexer/core/Cargo.toml

@@ -25,3 +25,7 @@ url.workspace = true
 logos-blockchain-core.workspace = true
 serde_json.workspace = true
 async-stream.workspace = true
+
+[dev-dependencies]
+tempfile.workspace = true
+

indexer/core/src/block_store.rs

@@ -8,10 +8,12 @@ use common::{
 };
 use nssa::{Account, AccountId, V02State};
 use storage::indexer::RocksDBIO;
+use tokio::sync::RwLock;
 
 #[derive(Clone)]
 pub struct IndexerStore {
     dbio: Arc<RocksDBIO>,
+    current_state: Arc<RwLock<V02State>>,
 }
 
 impl IndexerStore {
@@ -25,9 +27,11 @@ impl IndexerStore {
         initial_state: &V02State,
     ) -> Result<Self> {
         let dbio = RocksDBIO::open_or_create(location, genesis_block, initial_state)?;
+        let current_state = dbio.final_state()?;
 
         Ok(Self {
             dbio: Arc::new(dbio),
+            current_state: Arc::new(RwLock::new(current_state)),
         })
     }
 
@@ -93,22 +97,31 @@ impl IndexerStore {
         Ok(self.dbio.calculate_state_for_id(block_id)?)
     }
 
-    pub fn final_state(&self) -> Result<V02State> {
+    /// Recalculation of final state directly from DB.
+    ///
+    /// Used for indexer healthcheck.
+    pub fn recalculate_final_state(&self) -> Result<V02State> {
         Ok(self.dbio.final_state()?)
     }
 
-    pub fn get_account_final(&self, account_id: &AccountId) -> Result<Account> {
-        Ok(self.final_state()?.get_account_by_id(*account_id))
+    pub async fn account_current_state(&self, account_id: &AccountId) -> Result<Account> {
+        Ok(self
+            .current_state
+            .read()
+            .await
+            .get_account_by_id(*account_id))
     }
 
-    pub fn put_block(&self, mut block: Block, l1_header: HeaderId) -> Result<()> {
-        let mut final_state = self.dbio.final_state()?;
+    pub async fn put_block(&self, mut block: Block, l1_header: HeaderId) -> Result<()> {
+        {
+            let mut state_guard = self.current_state.write().await;
 
-        for transaction in &block.body.transactions {
-            transaction
-                .clone()
-                .transaction_stateless_check()?
-                .execute_check_on_state(&mut final_state)?;
+            for transaction in &block.body.transactions {
+                transaction
+                    .clone()
+                    .transaction_stateless_check()?
+                    .execute_check_on_state(&mut state_guard)?;
+            }
         }
 
         // ToDo: Currently we are fetching only finalized blocks
@@ -119,3 +132,95 @@ impl IndexerStore {
         Ok(self.dbio.put_block(&block, l1_header.into())?)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use nssa::{AccountId, PublicKey};
+    use tempfile::tempdir;
+
+    use super::*;
+
+    fn genesis_block() -> Block {
+        common::test_utils::produce_dummy_block(1, None, vec![])
+    }
+
+    fn acc1_sign_key() -> nssa::PrivateKey {
+        nssa::PrivateKey::try_new([1; 32]).unwrap()
+    }
+
+    fn acc2_sign_key() -> nssa::PrivateKey {
+        nssa::PrivateKey::try_new([2; 32]).unwrap()
+    }
+
+    fn acc1() -> AccountId {
+        AccountId::from(&PublicKey::new_from_private_key(&acc1_sign_key()))
+    }
+
+    fn acc2() -> AccountId {
+        AccountId::from(&PublicKey::new_from_private_key(&acc2_sign_key()))
+    }
+
+    #[test]
+    fn correct_startup() {
+        let home = tempdir().unwrap();
+
+        let storage = IndexerStore::open_db_with_genesis(
+            home.as_ref(),
+            &genesis_block(),
+            &nssa::V02State::new_with_genesis_accounts(&[(acc1(), 10000), (acc2(), 20000)], &[]),
+        )
+        .unwrap();
+
+        let block = storage.get_block_at_id(1).unwrap();
+        let final_id = storage.get_last_block_id().unwrap();
+
+        assert_eq!(block.header.hash, genesis_block().header.hash);
+        assert_eq!(final_id, 1);
+    }
+
+    #[tokio::test]
+    async fn state_transition() {
+        let home = tempdir().unwrap();
+
+        let storage = IndexerStore::open_db_with_genesis(
+            home.as_ref(),
+            &genesis_block(),
+            &nssa::V02State::new_with_genesis_accounts(&[(acc1(), 10000), (acc2(), 20000)], &[]),
+        )
+        .unwrap();
+
+        let mut prev_hash = genesis_block().header.hash;
+
+        let from = acc1();
+        let to = acc2();
+        let sign_key = acc1_sign_key();
+
+        for i in 2..10 {
+            let tx = common::test_utils::create_transaction_native_token_transfer(
+                from,
+                i - 2,
+                to,
+                10,
+                &sign_key,
+            );
+
+            let next_block = common::test_utils::produce_dummy_block(
+                u64::try_from(i).unwrap(),
+                Some(prev_hash),
+                vec![tx],
+            );
+            prev_hash = next_block.header.hash;
+
+            storage
+                .put_block(next_block, HeaderId::from([u8::try_from(i).unwrap(); 32]))
+                .await
+                .unwrap();
+        }
+
+        let acc1_val = storage.account_current_state(&acc1()).await.unwrap();
+        let acc2_val = storage.account_current_state(&acc2()).await.unwrap();
+
+        assert_eq!(acc1_val.balance, 9920);
+        assert_eq!(acc2_val.balance, 20080);
+    }
+}

indexer/core/src/lib.rs

@@ -124,8 +124,8 @@ impl IndexerCore {
                     l2_blocks_parsed_ids.sort_unstable();
                     info!("Parsed {} L2 blocks with ids {:?}", l2_block_vec.len(), l2_blocks_parsed_ids);
 
-                    for l2_block in l2_block_vec {
-                        self.store.put_block(l2_block.clone(), l1_header)?;
+                        for l2_block in l2_block_vec {
+                            self.store.put_block(l2_block.clone(), l1_header).await?;
 
                         yield Ok(l2_block);
                     }
@@ -158,7 +158,7 @@ impl IndexerCore {
                     info!("Parsed {} L2 blocks with ids {:?}", l2_block_vec.len(), l2_blocks_parsed_ids);
 
                     for l2_block in l2_block_vec {
-                        self.store.put_block(l2_block.clone(), header)?;
+                        self.store.put_block(l2_block.clone(), header).await?;
 
                         yield Ok(l2_block);
                     }

indexer/service/src/service.rs

@@ -74,7 +74,8 @@ impl indexer_service_rpc::RpcServer for IndexerService {
         Ok(self
             .indexer
             .store
-            .get_account_final(&account_id.into())
+            .account_current_state(&account_id.into())
+            .await
             .map_err(db_error)?
             .into())
     }
@@ -131,7 +132,11 @@ impl indexer_service_rpc::RpcServer for IndexerService {
 
     async fn healthcheck(&self) -> Result<(), ErrorObjectOwned> {
         // Checking, that indexer can calculate last state
-        let _ = self.indexer.store.final_state().map_err(db_error)?;
+        let _ = self
+            .indexer
+            .store
+            .recalculate_final_state()
+            .map_err(db_error)?;
 
         Ok(())
     }

integration_tests/tests/indexer.rs

@@ -1,19 +1,14 @@
 #![expect(
-    clippy::shadow_unrelated,
     clippy::tests_outside_test_module,
     reason = "We don't care about these in tests"
 )]
 
 use std::time::Duration;
 
-use anyhow::{Context as _, Result};
+use anyhow::Result;
 use indexer_service_rpc::RpcClient as _;
-use integration_tests::{
-    TIME_TO_WAIT_FOR_BLOCK_SECONDS, TestContext, format_private_account_id,
-    format_public_account_id, verify_commitment_is_in_state,
-};
+use integration_tests::{TIME_TO_WAIT_FOR_BLOCK_SECONDS, TestContext, format_public_account_id};
 use log::info;
-use nssa::AccountId;
 use tokio::test;
 use wallet::cli::{Command, programs::native_token_transfer::AuthTransferSubcommand};
 
@@ -120,36 +115,6 @@ async fn indexer_state_consistency() -> Result<()> {
     assert_eq!(acc_1_balance.balance, 9900);
     assert_eq!(acc_2_balance.balance, 20100);
 
-    let from: AccountId = ctx.existing_private_accounts()[0];
-    let to: AccountId = ctx.existing_private_accounts()[1];
-
-    let command = Command::AuthTransfer(AuthTransferSubcommand::Send {
-        from: format_private_account_id(from),
-        to: Some(format_private_account_id(to)),
-        to_npk: None,
-        to_vpk: None,
-        amount: 100,
-    });
-
-    wallet::cli::execute_subcommand(ctx.wallet_mut(), command).await?;
-
-    info!("Waiting for next block creation");
-    tokio::time::sleep(Duration::from_secs(TIME_TO_WAIT_FOR_BLOCK_SECONDS)).await;
-
-    let new_commitment1 = ctx
-        .wallet()
-        .get_private_account_commitment(from)
-        .context("Failed to get private account commitment for sender")?;
-    assert!(verify_commitment_is_in_state(new_commitment1, ctx.sequencer_client()).await);
-
-    let new_commitment2 = ctx
-        .wallet()
-        .get_private_account_commitment(to)
-        .context("Failed to get private account commitment for receiver")?;
-    assert!(verify_commitment_is_in_state(new_commitment2, ctx.sequencer_client()).await);
-
-    info!("Successfully transferred privately to owned account");
-
     // WAIT
     info!("Waiting for indexer to parse blocks");
     tokio::time::sleep(std::time::Duration::from_millis(L2_TO_L1_TIMEOUT_MILLIS)).await;

nssa/Cargo.toml

@@ -28,7 +28,6 @@ risc0-binfmt = "3.0.2"
 
 [dev-dependencies]
 token_core.workspace = true
-amm_core.workspace = true
 test_program_methods.workspace = true
 
 env_logger.workspace = true
@@ -38,3 +37,4 @@ test-case = "3.3.1"
 [features]
 default = []
 prove = ["risc0-zkvm/prove"]
+test-utils = []

nssa/core/src/account.rs

@@ -84,13 +84,15 @@ impl BorshDeserialize for Nonce {
     }
 }
 
+pub type Balance = u128;
+
 /// Account to be used both in public and private contexts.
 #[derive(
     Default, Clone, Eq, PartialEq, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
 )]
 pub struct Account {
     pub program_owner: ProgramId,
-    pub balance: u128,
+    pub balance: Balance,
     pub data: Data,
     pub nonce: Nonce,
 }

programs/amm/Cargo.toml

@@ -8,6 +8,10 @@ license = { workspace = true }
 workspace = true
 
 [dependencies]
+nssa = { workspace = true, optional = true, features = ["test-utils"], default-features = true }
 nssa_core.workspace = true
 token_core.workspace = true
 amm_core.workspace = true
+
+[features]
+nssa = ["dep:nssa"]

programs/amm/src/lib.rs

@@ -14,4 +14,5 @@ pub mod new_definition;
 pub mod remove;
 pub mod swap;
 
+#[cfg(test)]
 mod tests;

sequencer_core/src/block_settlement_client.rs

@@ -102,7 +102,8 @@ impl BlockSettlementClientTrait for BlockSettlementClient {
         self.client
             .post_transaction(tx)
             .await
-            .context("Failed to post transaction to Bedrock")?;
+            .context("Failed to post transaction to Bedrock after retries")?
+            .context("Failed to post transaction to Bedrock with non-retryable error")?;
 
         log::debug!("Posted block to Bedrock with parent id {parent_id:?} and msg id: {msg_id:?}");