Implement Append Entries to Log

chaos/experiments/leader-election.sh

@@ -86,7 +86,7 @@ check_leader() {
     local grpc_address=$1
 
     # Send the gRPC request using grpcurl and capture both stdout and stderr
-    response=$(grpcurl -import-path static/proto -proto append_entry.proto -plaintext -d '{"term": 5, "prevLogIndex": 0, "prevLogTerm": 4, "leader_id": "leader_id_value", "block_entry": {"number": 1, "hash": "ZXh0cmFfZGF0YV92YWx1ZQ==", "transactions_root": "ZXh0cmFfZGF0YV92YWx1ZQ==", "gas_used": 999, "gas_limit": 999, "bloom": "ZXh0cmFfZGF0YV92YWx1ZQ==", "timestamp": 123456789, "parent_hash": "ZXh0cmFfZGF0YV92YWx1ZQ==", "author": "ZXh0cmFfZGF0YV92YWx1ZQ==", "extra_data": "ZXh0cmFfZGF0YV92YWx1ZQ==", "miner": "ZXh0cmFfZGF0YV92YWx1ZQ==", "receipts_root": "ZXh0cmFfZGF0YV92YWx1ZQ==", "uncle_hash": "ZXh0cmFfZGF0YV92YWx1ZQ==", "size": 12345, "state_root": "ZXh0cmFfZGF0YV92YWx1ZQ==", "transaction_hashes": []}}' "$grpc_address" append_entry.AppendEntryService/AppendBlockCommit 2>&1)
+    response=$(grpcurl -import-path static/proto -proto append_entry.proto -plaintext -d '{"term": 999999999, "prevLogIndex": 0, "prevLogTerm": 0, "leader_id": "leader_id_value", "block_entry": {"number": 1, "hash": "ZXh0cmFfZGF0YV92YWx1ZQ==", "transactions_root": "ZXh0cmFfZGF0YV92YWx1ZQ==", "gas_used": 999, "gas_limit": 999, "bloom": "ZXh0cmFfZGF0YV92YWx1ZQ==", "timestamp": 123456789, "parent_hash": "ZXh0cmFfZGF0YV92YWx1ZQ==", "author": "ZXh0cmFfZGF0YV92YWx1ZQ==", "extra_data": "ZXh0cmFfZGF0YV92YWx1ZQ==", "miner": "ZXh0cmFfZGF0YV92YWx1ZQ==", "receipts_root": "ZXh0cmFfZGF0YV92YWx1ZQ==", "uncle_hash": "ZXh0cmFfZGF0YV92YWx1ZQ==", "size": 12345, "state_root": "ZXh0cmFfZGF0YV92YWx1ZQ==", "transaction_hashes": []}}' "$grpc_address" append_entry.AppendEntryService/AppendBlockCommit 2>&1)
 
     # Check the response for specific strings to determine the node status
     if [[ "$response" == *"append_transaction_executions called on leader node"* ]]; then
@@ -113,7 +113,7 @@ find_leader() {
 
 # Function to remove rocks-path directory
 remove_rocks_path() {
-    rm -rf data/
+    find . -type d -name "tmp_rocks_*" -print0 | xargs -0 rm -rf
 }
 
 # Function to run the election test

src/bin/run_with_importer.rs

@@ -24,6 +24,7 @@ async fn run(config: RunWithImporterConfig) -> anyhow::Result<()> {
     let miner = config.miner.init_external_mode(Arc::clone(&storage), None)?;
     let consensus = Consensus::new(
         Arc::clone(&storage),
+        config.storage.perm_storage.rocks_path_prefix.clone(),
         config.clone().candidate_peers.clone(),
         Some(config.clone()),
         config.address,

src/config.rs

@@ -846,7 +846,6 @@ pub struct PermanentStorageConfig {
     #[arg(long = "perm-storage", env = "PERM_STORAGE")]
     pub perm_storage_kind: PermanentStorageKind,
 
-    #[cfg(feature = "rocks")]
     /// RocksDB storage path prefix to execute multiple local Stratus instances.
     #[arg(long = "rocks-path-prefix", env = "ROCKS_PATH_PREFIX")]
     pub rocks_path_prefix: Option<String>,

src/eth/consensus/append_log_entries_storage.rs

@@ -7,15 +7,32 @@ use rocksdb::Options;
 use rocksdb::DB;
 
 use super::log_entry::LogEntry;
+use super::log_entry::LogEntryData;
 
 pub struct AppendLogEntriesStorage {
     db: DB,
 }
 
 impl AppendLogEntriesStorage {
-    pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
+    pub fn new(path: Option<String>) -> Result<Self> {
         let mut opts = Options::default();
         opts.create_if_missing(true);
+
+        let path = if let Some(prefix) = path {
+            // run some checks on the given prefix
+            assert!(!prefix.is_empty(), "given prefix for RocksDB is empty, try not providing the flag");
+            if Path::new(&prefix).is_dir() || Path::new(&prefix).iter().count() > 1 {
+                tracing::warn!(?prefix, "given prefix for RocksDB might put it in another folder");
+            }
+
+            let path = format!("{prefix}-log-entries-rocksdb");
+            tracing::info!("starting rocksdb log entries storage - at custom path: '{:?}'", path);
+            path
+        } else {
+            tracing::info!("starting rocksdb log entries storage - at default path: 'data/log-entries-rocksdb'"); // TODO: keep inside data?
+            "data/log-entries-rocksdb".to_string()
+        };
+
         let db = DB::open(&opts, path).context("Failed to open RocksDB")?;
         Ok(Self { db })
     }
@@ -81,6 +98,35 @@ impl AppendLogEntriesStorage {
             None => Ok(0), // Default to 0 if not set
         }
     }
+
+    pub fn save_log_entry(&self, index: u64, term: u64, data: LogEntryData, entry_type: &str) -> Result<()> {
+        tracing::debug!(index, term, "Creating {} log entry", entry_type);
+        let log_entry = LogEntry { term, index, data };
+        tracing::debug!(index = log_entry.index, term = log_entry.term, "{} log entry created", entry_type);
+
+        tracing::debug!("Checking for existing {} entry at new index", entry_type);
+        match self.get_entry(log_entry.index) {
+            Ok(Some(existing_entry)) =>
+                if existing_entry.term != log_entry.term {
+                    tracing::warn!(
+                        index = log_entry.index,
+                        "Conflicting entry found, deleting existing entry and all that follow it"
+                    );
+                    self.delete_entries_from(log_entry.index)?;
+                },
+            Ok(None) => {
+                // No existing entry at this index, proceed to save the new entry
+            }
+            Err(e) => {
+                tracing::error!(index = log_entry.index, "Error retrieving entry: {}", e);
+                return Err(anyhow::anyhow!("Error retrieving entry: {}", e));
+            }
+        }
+
+        tracing::debug!("Appending new {} log entry", entry_type);
+        self.save_entry(&log_entry)
+            .map_err(|e| anyhow::anyhow!("Failed to append {} log entry: {}", entry_type, e))
+    }
 }
 
 #[cfg(test)]
@@ -89,12 +135,11 @@ mod tests {
 
     use super::*;
     use crate::eth::consensus::tests::factories::*;
-    use crate::eth::consensus::LogEntryData;
 
     fn setup_storage() -> AppendLogEntriesStorage {
         let temp_dir = TempDir::new().unwrap();
-        let temp_path = temp_dir.path();
-        AppendLogEntriesStorage::new(temp_path).unwrap()
+        let temp_path = temp_dir.path().to_str().expect("Failed to get temp path").to_string();
+        AppendLogEntriesStorage::new(Some(temp_path)).unwrap()
     }
 
     #[test]
@@ -192,4 +237,91 @@ mod tests {
             panic!("Expected TransactionExecutionEntries");
         }
     }
+
+    #[test]
+    fn test_save_log_entry_no_conflict() {
+        let storage = setup_storage();
+        let index = 1;
+        let term = 1;
+        let log_entry_data = create_mock_log_entry_data_block();
+
+        storage.save_log_entry(index, term, log_entry_data.clone(), "block").unwrap();
+        let retrieved_entry = storage.get_entry(index).unwrap().unwrap();
+
+        assert_eq!(retrieved_entry.index, index);
+        assert_eq!(retrieved_entry.term, term);
+
+        if let LogEntryData::BlockEntry(ref block) = retrieved_entry.data {
+            if let LogEntryData::BlockEntry(ref expected_block) = log_entry_data {
+                assert_eq!(block.hash, expected_block.hash);
+                assert_eq!(block.number, expected_block.number);
+                assert_eq!(block.parent_hash, expected_block.parent_hash);
+                assert_eq!(block.uncle_hash, expected_block.uncle_hash);
+                assert_eq!(block.transactions_root, expected_block.transactions_root);
+                assert_eq!(block.state_root, expected_block.state_root);
+                assert_eq!(block.receipts_root, expected_block.receipts_root);
+                assert_eq!(block.miner, expected_block.miner);
+                assert_eq!(block.extra_data, expected_block.extra_data);
+                assert_eq!(block.size, expected_block.size);
+                assert_eq!(block.gas_limit, expected_block.gas_limit);
+                assert_eq!(block.gas_used, expected_block.gas_used);
+                assert_eq!(block.timestamp, expected_block.timestamp);
+                assert_eq!(block.bloom, expected_block.bloom);
+                assert_eq!(block.author, expected_block.author);
+                assert_eq!(block.transaction_hashes, expected_block.transaction_hashes);
+            } else {
+                panic!("Expected BlockEntry");
+            }
+        } else {
+            panic!("Expected BlockEntry");
+        }
+    }
+
+    #[test]
+    fn test_save_log_entry_with_conflict() {
+        let storage = setup_storage();
+        let index = 1;
+        let term = 1;
+        let conflicting_term = 2;
+        let log_entry_data = create_mock_log_entry_data_block();
+
+        // Save initial log entry
+        storage.save_log_entry(index, term, log_entry_data.clone(), "block").unwrap();
+
+        // Save conflicting log entry at the same index but with a different term
+        storage.save_log_entry(index, conflicting_term, log_entry_data.clone(), "block").unwrap();
+
+        // Assert no entries exist after the conflicting entry's index, confirming that the conflicting entry and all that follow it were deleted
+        assert!(storage.get_entry(index + 1).unwrap().is_none());
+
+        // Retrieve the entry at the index and assert it matches the conflicting term entry
+        let retrieved_entry = storage.get_entry(index).unwrap().unwrap();
+        assert_eq!(retrieved_entry.index, index);
+        assert_eq!(retrieved_entry.term, conflicting_term);
+
+        if let LogEntryData::BlockEntry(ref block) = retrieved_entry.data {
+            if let LogEntryData::BlockEntry(ref expected_block) = log_entry_data {
+                assert_eq!(block.hash, expected_block.hash);
+                assert_eq!(block.number, expected_block.number);
+                assert_eq!(block.parent_hash, expected_block.parent_hash);
+                assert_eq!(block.uncle_hash, expected_block.uncle_hash);
+                assert_eq!(block.transactions_root, expected_block.transactions_root);
+                assert_eq!(block.state_root, expected_block.state_root);
+                assert_eq!(block.receipts_root, expected_block.receipts_root);
+                assert_eq!(block.miner, expected_block.miner);
+                assert_eq!(block.extra_data, expected_block.extra_data);
+                assert_eq!(block.size, expected_block.size);
+                assert_eq!(block.gas_limit, expected_block.gas_limit);
+                assert_eq!(block.gas_used, expected_block.gas_used);
+                assert_eq!(block.timestamp, expected_block.timestamp);
+                assert_eq!(block.bloom, expected_block.bloom);
+                assert_eq!(block.author, expected_block.author);
+                assert_eq!(block.transaction_hashes, expected_block.transaction_hashes);
+            } else {
+                panic!("Expected BlockEntry");
+            }
+        } else {
+            panic!("Expected BlockEntry");
+        }
+    }
 }

src/eth/consensus/mod.rs

@@ -63,6 +63,8 @@ use append_entry::RequestVoteRequest;
 use append_entry::StatusCode;
 use append_entry::TransactionExecutionEntry;
 
+#[cfg(feature = "rocks")]
+use self::append_log_entries_storage::AppendLogEntriesStorage;
 use self::log_entry::LogEntryData;
 use super::primitives::TransactionExecution;
 use super::primitives::TransactionInput;
@@ -164,6 +166,8 @@ pub struct Consensus {
     broadcast_sender: broadcast::Sender<LogEntryData>, //propagates the blocks
     importer_config: Option<RunWithImporterConfig>,    //HACK this is used with sync online only
     storage: Arc<StratusStorage>,
+    #[cfg(feature = "rocks")]
+    log_entries_storage: Arc<AppendLogEntriesStorage>,
     peers: Arc<RwLock<HashMap<PeerAddress, PeerTuple>>>,
     #[allow(dead_code)]
     direct_peers: Vec<String>,
@@ -180,8 +184,10 @@ pub struct Consensus {
 }
 
 impl Consensus {
+    #[allow(clippy::too_many_arguments)] //TODO: refactor into consensus config
     pub async fn new(
         storage: Arc<StratusStorage>,
+        log_storage_path: Option<String>,
         direct_peers: Vec<String>,
         importer_config: Option<RunWithImporterConfig>,
         jsonrpc_address: SocketAddr,
@@ -197,6 +203,8 @@ impl Consensus {
         let consensus = Self {
             broadcast_sender,
             storage,
+            #[cfg(feature = "rocks")]
+            log_entries_storage: Arc::new(AppendLogEntriesStorage::new(log_storage_path).unwrap()),
             peers,
             direct_peers,
             current_term: AtomicU64::new(0),
@@ -385,19 +393,48 @@ impl Consensus {
     }
 
     fn initialize_transaction_execution_queue(consensus: Arc<Consensus>) {
-        //TODO add data to consensus-log-transactions
-        //TODO rediscover followers on comunication error
-        //XXX FIXME deal with the scenario where a transactionHash arrives after the block, in this case before saving the block LogEntry, it should ALWAYS wait for all transaction hashes
-        //TODO maybe check if I'm currently the leader?
+        // TODO: add data to consensus-log-transactions
+        // TODO: rediscover followers on communication error
+        // XXX FIXME: deal with the scenario where a transactionHash arrives after the block;
+        // in this case, before saving the block LogEntry, it should ALWAYS wait for all transaction hashes
+        // TODO: maybe check if I'm currently the leader?
+
         spawn_named("consensus::transaction_execution_queue", async move {
             let interval = Duration::from_millis(40);
             loop {
                 tokio::time::sleep(interval).await;
+
                 if consensus.is_leader() {
                     let mut queue = consensus.transaction_execution_queue.lock().await;
                     let executions = queue.drain(..).collect::<Vec<_>>();
                     drop(queue);
 
+                    #[cfg(feature = "rocks")]
+                    {
+                        tracing::debug!(executions_len = executions.len(), "Processing transaction executions");
+                        if !executions.is_empty() {
+                            let last_index = consensus.log_entries_storage.get_last_index().unwrap_or(0);
+                            tracing::debug!(last_index, "Last index fetched");
+
+                            let current_term = consensus.current_term.load(Ordering::SeqCst);
+                            tracing::debug!(current_term, "Current term loaded");
+
+                            match consensus.log_entries_storage.save_log_entry(
+                                last_index + 1,
+                                current_term,
+                                LogEntryData::TransactionExecutionEntries(executions.clone()),
+                                "transaction",
+                            ) {
+                                Ok(_) => {
+                                    tracing::debug!("Transaction execution entry saved successfully");
+                                }
+                                Err(e) => {
+                                    tracing::error!("Failed to save transaction execution entry: {:?}", e);
+                                }
+                            }
+                        }
+                    }
+
                     let peers = consensus.peers.read().await;
                     for (_, (peer, _)) in peers.iter() {
                         let mut peer_clone = peer.clone();
@@ -420,14 +457,14 @@ impl Consensus {
             loop {
                 tokio::select! {
                     Ok(tx) = rx_pending_txs.recv() => {
+                        tracing::debug!("Attempting to receive transaction execution");
                         if consensus.is_leader() {
                             tracing::info!(tx_hash = %tx.hash(), "received transaction execution to send to followers");
                             if tx.is_local() {
                                 tracing::debug!(tx_hash = %tx.hash(), "skipping local transaction because only external transactions are supported for now");
                                 continue;
                             }
 
-                            //TODO XXX save transaction to appendEntries log
                             let transaction = vec![tx.to_append_entry_transaction()];
                             let transaction_entry = LogEntryData::TransactionExecutionEntries(transaction);
                             if consensus.broadcast_sender.send(transaction_entry).is_err() {
@@ -437,13 +474,44 @@ impl Consensus {
                     }
                     Ok(block) = rx_blocks.recv() => {
                         if consensus.is_leader() {
-                            tracing::info!(number = block.header.number.as_u64(), "received block to send to followers");
-
-                            //TODO save block to appendEntries log
-                            //TODO before saving check if all transaction_hashes are already in the log
-                            let block_entry = LogEntryData::BlockEntry(block.header.to_append_entry_block_header(Vec::new()));
-                            if consensus.broadcast_sender.send(block_entry).is_err() {
-                                tracing::error!("failed to broadcast block");
+                            tracing::info!(number = block.header.number.as_u64(), "Leader received block to send to followers");
+
+                            #[cfg(feature = "rocks")]
+                            {
+                                //TODO: before saving check if all transaction_hashes are already in the log
+                                let last_index = consensus.log_entries_storage.get_last_index().unwrap_or(0);
+                                tracing::debug!(last_index, "Last index fetched");
+
+                                let current_term = consensus.current_term.load(Ordering::SeqCst);
+                                tracing::debug!(current_term, "Current term loaded");
+
+                                let transaction_hashes: Vec<Vec<u8>> = block.transactions.iter().map(|tx| tx.input.hash.to_string().into_bytes()).collect();
+
+                                match consensus.log_entries_storage.save_log_entry(
+                                    last_index + 1,
+                                    current_term,
+                                    LogEntryData::BlockEntry(block.header.to_append_entry_block_header(transaction_hashes.clone())),
+                                    "block",
+                                ) {
+                                    Ok(_) => {
+                                        tracing::debug!("Block entry saved successfully");
+                                        let block_entry = LogEntryData::BlockEntry(block.header.to_append_entry_block_header(transaction_hashes));
+                                        if consensus.broadcast_sender.send(block_entry).is_err() {
+                                            tracing::error!("Failed to broadcast block");
+                                        }
+                                    }
+                                    Err(e) => {
+                                        tracing::error!("Failed to save block entry: {:?}", e);
+                                    }
+                                }
+                            }
+                            #[cfg(not(feature = "rocks"))]
+                            {
+                                let transaction_hashes: Vec<Vec<u8>> = block.transactions.iter().map(|tx| tx.input.hash.to_string().into_bytes()).collect();
+                                let block_entry = LogEntryData::BlockEntry(block.header.to_append_entry_block_header(transaction_hashes));
+                                if consensus.broadcast_sender.send(block_entry).is_err() {
+                                    tracing::error!("Failed to broadcast block");
+                                }
                             }
                         }
                     }