Merge branch 'Pravdyvy/indexer-state-management' into Pravdyvy/bedrock-parsing-from-start-of-a-channel

common/src/block.rs

@@ -109,15 +109,15 @@ impl From<Block> for HashableBlockData {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+/// Helper struct for account (de-)serialization
-/// Helperstruct for account serialization
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AccountInitialData {
     pub account_id: AccountId,
     pub balance: u128,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+/// Helper struct to (de-)serialize initial commitments
-/// Helperstruct to initialize commitments
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct CommitmentsInitialData {
     pub npk: nssa_core::NullifierPublicKey,
     pub account: nssa_core::account::Account,

common/src/transaction.rs

@@ -1,5 +1,3 @@
-use std::fmt::Display;
-
 use borsh::{BorshDeserialize, BorshSerialize};
 use log::warn;
 use nssa::{AccountId, V02State};
@@ -22,6 +20,54 @@ impl NSSATransaction {
             NSSATransaction::ProgramDeployment(tx) => tx.hash(),
         })
     }
+
+    pub fn affected_public_account_ids(&self) -> Vec<AccountId> {
+        match self {
+            NSSATransaction::ProgramDeployment(tx) => tx.affected_public_account_ids(),
+            NSSATransaction::Public(tx) => tx.affected_public_account_ids(),
+            NSSATransaction::PrivacyPreserving(tx) => tx.affected_public_account_ids(),
+        }
+    }
+
+    // TODO: Introduce type-safe wrapper around checked transaction, e.g. AuthenticatedTransaction
+    pub fn transaction_stateless_check(self) -> Result<Self, TransactionMalformationError> {
+        // Stateless checks here
+        match self {
+            NSSATransaction::Public(tx) => {
+                if tx.witness_set().is_valid_for(tx.message()) {
+                    Ok(NSSATransaction::Public(tx))
+                } else {
+                    Err(TransactionMalformationError::InvalidSignature)
+                }
+            }
+            NSSATransaction::PrivacyPreserving(tx) => {
+                if tx.witness_set().signatures_are_valid_for(tx.message()) {
+                    Ok(NSSATransaction::PrivacyPreserving(tx))
+                } else {
+                    Err(TransactionMalformationError::InvalidSignature)
+                }
+            }
+            NSSATransaction::ProgramDeployment(tx) => Ok(NSSATransaction::ProgramDeployment(tx)),
+        }
+    }
+
+    pub fn execute_check_on_state(
+        self,
+        state: &mut V02State,
+    ) -> Result<Self, nssa::error::NssaError> {
+        match &self {
+            NSSATransaction::Public(tx) => state.transition_from_public_transaction(tx),
+            NSSATransaction::PrivacyPreserving(tx) => {
+                state.transition_from_privacy_preserving_transaction(tx)
+            }
+            NSSATransaction::ProgramDeployment(tx) => {
+                state.transition_from_program_deployment_transaction(tx)
+            }
+        }
+        .inspect_err(|err| warn!("Error at transition {err:#?}"))?;
+
+        Ok(self)
+    }
 }
 
 impl From<nssa::PublicTransaction> for NSSATransaction {
@@ -36,16 +82,6 @@ impl From<nssa::PrivacyPreservingTransaction> for NSSATransaction {
     }
 }
 
-impl NSSATransaction {
-    pub fn affected_public_account_ids(&self) -> Vec<AccountId> {
-        match self {
-            NSSATransaction::ProgramDeployment(tx) => tx.affected_public_account_ids(),
-            NSSATransaction::Public(tx) => tx.affected_public_account_ids(),
-            NSSATransaction::PrivacyPreserving(tx) => tx.affected_public_account_ids(),
-        }
-    }
-}
-
 impl From<nssa::ProgramDeploymentTransaction> for NSSATransaction {
     fn from(value: nssa::ProgramDeploymentTransaction) -> Self {
         Self::ProgramDeployment(value)
@@ -61,58 +97,10 @@ pub enum TxKind {
     ProgramDeployment,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, thiserror::Error)]
 pub enum TransactionMalformationError {
+    #[error("Invalid signature(s)")]
     InvalidSignature,
+    #[error("Failed to decode transaction with hash: {tx:?}")]
     FailedToDecode { tx: HashType },
 }
-
-impl Display for TransactionMalformationError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{self:#?}")
-    }
-}
-
-impl std::error::Error for TransactionMalformationError {}
-
-// TODO: Introduce type-safe wrapper around checked transaction, e.g. AuthenticatedTransaction
-pub fn transaction_pre_check(
-    tx: NSSATransaction,
-) -> Result<NSSATransaction, TransactionMalformationError> {
-    // Stateless checks here
-    match tx {
-        NSSATransaction::Public(tx) => {
-            if tx.witness_set().is_valid_for(tx.message()) {
-                Ok(NSSATransaction::Public(tx))
-            } else {
-                Err(TransactionMalformationError::InvalidSignature)
-            }
-        }
-        NSSATransaction::PrivacyPreserving(tx) => {
-            if tx.witness_set().signatures_are_valid_for(tx.message()) {
-                Ok(NSSATransaction::PrivacyPreserving(tx))
-            } else {
-                Err(TransactionMalformationError::InvalidSignature)
-            }
-        }
-        NSSATransaction::ProgramDeployment(tx) => Ok(NSSATransaction::ProgramDeployment(tx)),
-    }
-}
-
-pub fn execute_check_transaction_on_state(
-    state: &mut V02State,
-    tx: NSSATransaction,
-) -> Result<NSSATransaction, nssa::error::NssaError> {
-    match &tx {
-        NSSATransaction::Public(tx) => state.transition_from_public_transaction(tx),
-        NSSATransaction::PrivacyPreserving(tx) => {
-            state.transition_from_privacy_preserving_transaction(tx)
-        }
-        NSSATransaction::ProgramDeployment(tx) => {
-            state.transition_from_program_deployment_transaction(tx)
-        }
-    }
-    .inspect_err(|err| warn!("Error at transition {err:#?}"))?;
-
-    Ok(tx)
-}

indexer/core/src/block_store.rs

@@ -2,10 +2,7 @@ use std::{path::Path, sync::Arc};
 
 use anyhow::Result;
 use bedrock_client::HeaderId;
-use common::{
+use common::{block::Block, transaction::NSSATransaction};
-    block::Block,
-    transaction::{NSSATransaction, execute_check_transaction_on_state, transaction_pre_check},
-};
 use nssa::{Account, AccountId, V02State};
 use storage::indexer::RocksDBIO;
 
@@ -107,10 +104,10 @@ impl IndexerStore {
         let mut final_state = self.dbio.final_state()?;
 
         for transaction in &block.body.transactions {
-            execute_check_transaction_on_state(
+            transaction
-                &mut final_state,
+                .clone()
-                transaction_pre_check(transaction.clone())?,
+                .transaction_stateless_check()?
-            )?;
+                .execute_check_on_state(&mut final_state)?;
         }
 
         Ok(self.dbio.put_block(block, l1_header.into())?)

indexer/service/Dockerfile

@@ -43,6 +43,10 @@ RUN useradd -m -u 1000 -s /bin/bash indexer_service_user && \
 # Copy binary from builder
 COPY --from=builder --chown=indexer_service_user:indexer_service_user /indexer_service/target/release/indexer_service /usr/local/bin/indexer_service
 
+# Copy entrypoint script
+COPY indexer/service/docker-entrypoint.sh /docker-entrypoint.sh
+RUN chmod +x /docker-entrypoint.sh
+
 # Expose default port
 EXPOSE 8779
 
@@ -60,7 +64,9 @@ HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
 # Run the application
 ENV RUST_LOG=info
 
-USER indexer_service_user
+USER root
+
+ENTRYPOINT ["/docker-entrypoint.sh"]
 
 WORKDIR /indexer_service
 CMD ["indexer_service", "/etc/indexer_service/indexer_config.json"]

indexer/service/docker-compose.yml

@@ -10,3 +10,5 @@ services:
     volumes:
       # Mount configuration
       - ./configs/indexer_config.json:/etc/indexer_service/indexer_config.json
+      # Mount data folder
+      - ./data:/var/lib/indexer_service

indexer/service/docker-entrypoint.sh

@@ -0,0 +1,29 @@
+#!/bin/sh
+
+# This is an entrypoint script for the sequencer_runner Docker container,
+# it's not meant to be executed outside of the container.
+
+set -e
+
+CONFIG="/etc/sequencer_runner/sequencer_config.json"
+
+# Check config file exists
+if [ ! -f "$CONFIG" ]; then
+  echo "Config file not found: $CONFIG" >&2
+  exit 1
+fi
+
+# Parse home dir
+HOME_DIR=$(jq -r '.home' "$CONFIG")
+
+if [ -z "$HOME_DIR" ] || [ "$HOME_DIR" = "null" ]; then
+  echo "'home' key missing in config" >&2
+  exit 1
+fi
+
+# Give permissions to the data directory and switch to non-root user
+if [ "$(id -u)" = "0" ]; then
+  mkdir -p "$HOME_DIR"
+  chown -R sequencer_user:sequencer_user "$HOME_DIR"
+  exec gosu sequencer_user "$@"
+fi

indexer/service/src/service.rs

@@ -49,9 +49,7 @@ impl indexer_service_rpc::RpcServer for IndexerService {
     }
 
     async fn get_last_finalized_block_id(&self) -> Result<BlockId, ErrorObjectOwned> {
-        self.indexer.store.get_last_block_id().map_err(|err| {
+        self.indexer.store.get_last_block_id().map_err(db_error)
-            ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
-        })
     }
 
     async fn get_block_by_id(&self, block_id: BlockId) -> Result<Block, ErrorObjectOwned> {
@@ -59,9 +57,7 @@ impl indexer_service_rpc::RpcServer for IndexerService {
             .indexer
             .store
             .get_block_at_id(block_id)
-            .map_err(|err| {
+            .map_err(db_error)?
-                ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
-            })?
             .into())
     }
 
@@ -70,9 +66,7 @@ impl indexer_service_rpc::RpcServer for IndexerService {
             .indexer
             .store
             .get_block_by_hash(block_hash.0)
-            .map_err(|err| {
+            .map_err(db_error)?
-                ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
-            })?
             .into())
     }
 
@@ -81,9 +75,7 @@ impl indexer_service_rpc::RpcServer for IndexerService {
             .indexer
             .store
             .get_account_final(&account_id.into())
-            .map_err(|err| {
+            .map_err(db_error)?
-                ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
-            })?
             .into())
     }
 
@@ -92,9 +84,7 @@ impl indexer_service_rpc::RpcServer for IndexerService {
             .indexer
             .store
             .get_transaction_by_hash(tx_hash.0)
-            .map_err(|err| {
+            .map_err(db_error)?
-                ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
-            })?
             .into())
     }
 
@@ -103,9 +93,7 @@ impl indexer_service_rpc::RpcServer for IndexerService {
             .indexer
             .store
             .get_block_batch(offset as u64, limit as u64)
-            .map_err(|err| {
+            .map_err(db_error)?;
-                ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
-            })?;
 
         let mut block_res = vec![];
 
@@ -126,9 +114,7 @@ impl indexer_service_rpc::RpcServer for IndexerService {
             .indexer
             .store
             .get_transactions_by_account(account_id.value, offset as u64, limit as u64)
-            .map_err(|err| {
+            .map_err(db_error)?;
-                ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
-            })?;
 
         let mut tx_res = vec![];
 
@@ -141,9 +127,7 @@ 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(|err| {
+        let _ = self.indexer.store.final_state().map_err(db_error)?;
-            ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
-        })?;
 
         Ok(())
     }
@@ -292,3 +276,7 @@ pub fn not_yet_implemented_error() -> ErrorObjectOwned {
         Option::<String>::None,
     )
 }
+
+fn db_error(err: anyhow::Error) -> ErrorObjectOwned {
+    ErrorObjectOwned::owned(-32001, "DBError".to_string(), Some(format!("{err:#?}")))
+}

sequencer_core/src/block_settlement_client.rs

@@ -99,11 +99,7 @@ impl BlockSettlementClientTrait for BlockSettlementClient {
             Some(Op::ChannelInscribe(inscribe)) => (inscribe.parent, inscribe.id()),
             _ => panic!("Expected ChannelInscribe op"),
         };
-        log::info!(">>>>>>>>>>>>>>>>>>>>>>");
+        log::info!("Posted block to Bedrock with parent id {parent_id:?} and msg id: {msg_id:?}");
-        log::info!("Posted block to Bedrock");
-        log::info!(">>>>>> parent id: {parent_id:?}");
-        log::info!(">>>>>> msg id: {msg_id:?}");
-        log::info!(">>>>>>>>>>>>>>>>>>>>>>");
         self.bedrock_client
             .post_transaction(tx)
             .await

sequencer_core/src/lib.rs

@@ -351,9 +351,8 @@ mod tests {
     use base58::ToBase58;
     use bedrock_client::BackoffConfig;
     use common::{
-        block::AccountInitialData,
+        block::AccountInitialData, test_utils::sequencer_sign_key_for_testing,
-        test_utils::sequencer_sign_key_for_testing,
+        transaction::NSSATransaction,
-        transaction::{NSSATransaction, transaction_pre_check},
     };
     use logos_blockchain_core::mantle::ops::channel::ChannelId;
     use mempool::MemPoolHandle;
@@ -515,7 +514,7 @@ mod tests {
     #[test]
     fn test_transaction_pre_check_pass() {
         let tx = common::test_utils::produce_dummy_empty_transaction();
-        let result = transaction_pre_check(tx);
+        let result = tx.transaction_stateless_check();
 
         assert!(result.is_ok());
     }
@@ -532,7 +531,7 @@ mod tests {
         let tx = common::test_utils::create_transaction_native_token_transfer(
             acc1, 0, acc2, 10, sign_key1,
         );
-        let result = transaction_pre_check(tx);
+        let result = tx.transaction_stateless_check();
 
         assert!(result.is_ok());
     }
@@ -551,7 +550,7 @@ mod tests {
         );
 
         // Signature is valid, stateless check pass
-        let tx = transaction_pre_check(tx).unwrap();
+        let tx = tx.transaction_stateless_check().unwrap();
 
         // Signature is not from sender. Execution fails
         let result = sequencer.execute_check_transaction_on_state(tx);
@@ -575,7 +574,7 @@ mod tests {
             acc1, 0, acc2, 10000000, sign_key1,
         );
 
-        let result = transaction_pre_check(tx);
+        let result = tx.transaction_stateless_check();
 
         // Passed pre-check
         assert!(result.is_ok());

sequencer_rpc/src/process.rs

@@ -20,7 +20,7 @@ use common::{
             SendTxResponse,
         },
     },
-    transaction::{NSSATransaction, transaction_pre_check},
+    transaction::NSSATransaction,
 };
 use itertools::Itertools as _;
 use log::warn;
@@ -94,8 +94,9 @@ impl<BC: BlockSettlementClientTrait, IC: IndexerClientTrait> JsonHandler<BC, IC>
         let tx = borsh::from_slice::<NSSATransaction>(&send_tx_req.transaction).unwrap();
         let tx_hash = tx.hash();
 
-        let authenticated_tx =
+        let authenticated_tx = tx
-            transaction_pre_check(tx).inspect_err(|err| warn!("Error at pre_check {err:#?}"))?;
+            .transaction_stateless_check()
+            .inspect_err(|err| warn!("Error at pre_check {err:#?}"))?;
 
         // TODO: Do we need a timeout here? It will be usable if we have too many transactions to
         // process

storage/src/indexer.rs

@@ -1,9 +1,6 @@
 use std::{collections::HashMap, ops::Div, path::Path, sync::Arc};
 
-use common::{
+use common::{block::Block, transaction::NSSATransaction};
-    block::Block,
-    transaction::{NSSATransaction, execute_check_transaction_on_state, transaction_pre_check},
-};
 use nssa::V02State;
 use rocksdb::{
     BoundColumnFamily, ColumnFamilyDescriptor, DBWithThreadMode, MultiThreaded, Options, WriteBatch,
@@ -652,19 +649,19 @@ impl RocksDBIO {
                 let block = self.get_block(id)?;
 
                 for transaction in block.body.transactions {
-                    execute_check_transaction_on_state(
+                    transaction
-                        &mut breakpoint,
+                        .transaction_stateless_check()
-                        transaction_pre_check(transaction).map_err(|err| {
+                        .map_err(|err| {
                             DbError::db_interaction_error(format!(
                                 "transaction pre check failed with err {err:?}"
                             ))
-                        })?,
+                        })?
-                    )
+                        .execute_check_on_state(&mut breakpoint)
-                    .map_err(|err| {
+                        .map_err(|err| {
-                        DbError::db_interaction_error(format!(
+                            DbError::db_interaction_error(format!(
-                            "transaction execution failed with err {err:?}"
+                                "transaction execution failed with err {err:?}"
-                        ))
+                            ))
-                    })?;
+                        })?;
                 }
             }