From c31a8333ec979d7c6cea7462dd32cab37544fb60 Mon Sep 17 00:00:00 2001 From: imotai Date: Mon, 31 Jul 2023 17:05:51 +0800 Subject: [PATCH 1/4] fix: handle the arweave-rs exception --- Cargo.toml | 2 +- sdk/tests/client_v2.test.ts | 2 +- src/node/src/node_test_base.rs | 53 +++++++++++++++++++-------------- src/node/src/recover.rs | 7 ++--- src/node/src/rollup_executor.rs | 21 ++++++++++++- 5 files changed, 54 insertions(+), 31 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index f13163cd4..36cb07a07 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,6 @@ fastcrypto = {git = "https://github.com/MystenLabs/fastcrypto", rev = "306465d4f ethers = {git="https://github.com/imotai/ethers-rs", rev="d526191b7972e8cf4412fee8b71cbf42e0ce7995"} tonic = {git="https://github.com/hyperium/tonic", rev="ae7580160431cd25c1eecda4c85014ef6ce8d93f"} tonic-web = {git="https://github.com/hyperium/tonic", rev="ae7580160431cd25c1eecda4c85014ef6ce8d93f"} -arweave-rs = {git="https://github.com/imotai/arweave-rs", rev="c46183c4cef05004685e53a869606d18abcbcdf2"} +arweave-rs = {git="https://github.com/imotai/arweave-rs", rev="ce7bcaadc614afb0e991953da6861ff173dce6ea"} ejdb2rs = {git="https://github.com/imotai/ejdb2rs", rev="dfe9a00c66986c072b8375ce68086b9c98ded4fe"} serde_json= "1.0" diff --git a/sdk/tests/client_v2.test.ts b/sdk/tests/client_v2.test.ts index 982e392ff..39fbd313b 100644 --- a/sdk/tests/client_v2.test.ts +++ b/sdk/tests/client_v2.test.ts @@ -63,7 +63,7 @@ describe('test db3.js client module', () => { async function createTestClient() { const db3_account = createRandomAccount() const client = createClient( - 'http://127.0.0.1:26619', + 'http://scoll.rollup.testnet.db3.network', 'http://127.0.0.1:26639', db3_account ) diff --git a/src/node/src/node_test_base.rs b/src/node/src/node_test_base.rs index 0c0420c19..7c982a4aa 100644 --- a/src/node/src/node_test_base.rs +++ b/src/node/src/node_test_base.rs @@ -144,9 +144,35 @@ pub mod tests { } } + pub fn add_mutations(storage: &MutationStore, rows: u64) -> u64 { + let payload: Vec = vec![1]; + let signature: &str = "0xasdasdsad"; + let (_id, block, _order) = storage + .generate_mutation_block_and_order(payload.as_ref(), signature) + .unwrap(); + for _i in 0..rows { + let (_id, block, order) = storage + .generate_mutation_block_and_order(payload.as_ref(), signature) + .unwrap(); + let result = storage.add_mutation( + payload.as_ref(), + signature, + "", + &DB3Address::ZERO, + 1, + block, + order, + 1, + MutationAction::CreateDocumentDb, + ); + assert_eq!(true, result.is_ok()); + } + block + } + pub async fn setup_for_smoke_test( tmp_dir_path: &TempDir, - ) -> Result<(RollupExecutor, Recover)> { + ) -> Result<(RollupExecutor, Recover, MutationStore)> { let ( state_config, system_store_config, @@ -164,29 +190,10 @@ pub mod tests { let result = system_store.update_config(&SystemRole::DataRollupNode, &system_config); let db_store = DBStoreV2::new(db_config)?; assert_eq!(true, result.is_ok()); - for _i in 0..3 { - let payload: Vec = vec![1]; - let signature: &str = "0xasdasdsad"; - let (_id, block, order) = storage - .generate_mutation_block_and_order(payload.as_ref(), signature) - .unwrap(); - println!("block: {}, order: {}", block, order); - let result = storage.add_mutation( - payload.as_ref(), - signature, - "", - &DB3Address::ZERO, - 1, - block, - order, - 1, - MutationAction::CreateDocumentDb, - ); - assert_eq!(true, result.is_ok()); - } + Self::add_mutations(&storage, 3); let (_, _) = storage.increase_block_return_last_state()?; let rollup_executor = - RollupExecutor::new(rollup_config, storage, system_store.clone()).await?; + RollupExecutor::new(rollup_config, storage.clone(), system_store.clone()).await?; let rollup_recover = Recover::new( recover_rollup_config, db_store.clone(), @@ -194,7 +201,7 @@ pub mod tests { None, ) .await?; - Ok((rollup_executor, rollup_recover)) + Ok((rollup_executor, rollup_recover, storage)) } } } diff --git a/src/node/src/recover.rs b/src/node/src/recover.rs index 41390dac2..31d8bcb0a 100644 --- a/src/node/src/recover.rs +++ b/src/node/src/recover.rs @@ -17,17 +17,14 @@ use crate::ar_toolbox::ArToolBox; use crate::mutation_utils::MutationUtil; -use bytes::BytesMut; -use db3_crypto::id::TxId; use db3_error::{DB3Error, Result}; -use db3_proto::db3_mutation_v2_proto::{MutationAction, MutationBody, MutationHeader}; +use db3_proto::db3_mutation_v2_proto::MutationAction; use db3_storage::ar_fs::{ArFileSystem, ArFileSystemConfig}; use db3_storage::db_store_v2::DBStoreV2; use db3_storage::meta_store_client::MetaStoreClient; use db3_storage::mutation_store::MutationStore; use db3_storage::system_store::{SystemRole, SystemStore}; use ethers::prelude::Signer; -use prost::Message; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use tracing::{debug, info}; @@ -258,7 +255,7 @@ mod tests { let tmp_dir_path = TempDir::new("test_fetch_arware_tx_from_block").expect("create temp dir"); match NodeTestBase::setup_for_smoke_test(&tmp_dir_path).await { - Ok((rollup_executor, recover)) => { + Ok((rollup_executor, recover, _storage)) => { let result = rollup_executor.process().await; assert_eq!(true, result.is_ok()); let result = recover.fetch_arweave_tx_from_block(0).await; diff --git a/src/node/src/rollup_executor.rs b/src/node/src/rollup_executor.rs index 24f97a039..95525dea9 100644 --- a/src/node/src/rollup_executor.rs +++ b/src/node/src/rollup_executor.rs @@ -348,6 +348,7 @@ impl RollupExecutor { Ok(()) } } + #[cfg(test)] mod tests { use crate::node_test_base::tests::NodeTestBase; @@ -357,7 +358,7 @@ mod tests { async fn test_rollup_smoke_test() { let tmp_dir_path = TempDir::new("test_rollup_smoke_test").expect("create temp dir"); match NodeTestBase::setup_for_smoke_test(&tmp_dir_path).await { - Ok((rollup_executor, recover)) => { + Ok((rollup_executor, recover, storage)) => { let result = rollup_executor.process().await; assert_eq!(true, result.is_ok()); let result = recover.get_latest_arweave_tx().await; @@ -365,6 +366,24 @@ mod tests { let tx = result.unwrap(); println!("the tx is {}", tx); assert!(!tx.is_empty()); + let result = storage.get_last_rollup_record(); + assert_eq!(true, result.is_ok()); + let record = result.unwrap().unwrap(); + println!( + "start block {} end block {}", + record.start_block, record.end_block + ); + let block = NodeTestBase::add_mutations(&storage, 10); + let result = rollup_executor.process().await; + assert_eq!(true, result.is_ok()); + let result = storage.get_last_rollup_record(); + assert_eq!(true, result.is_ok()); + let record = result.unwrap().unwrap(); + println!( + "start block {} end block {}", + record.start_block, record.end_block + ); + assert_eq!(record.end_block, block); } Err(e) => { assert!(false, "{e}"); From c9f8a51cf3de7b872913bfce26cbe12d7a9d27eb Mon Sep 17 00:00:00 2001 From: imotai Date: Mon, 31 Jul 2023 17:08:16 +0800 Subject: [PATCH 2/4] fix: revert the client_v2.test.ts --- sdk/src/index.ts | 2 +- sdk/tests/client_v2.test.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/src/index.ts b/sdk/src/index.ts index a365d62a7..aaba2b9b0 100644 --- a/sdk/src/index.ts +++ b/sdk/src/index.ts @@ -59,7 +59,7 @@ export { getDatabase, getCollection, addIndex, - deleteEventDatabase + deleteEventDatabase, } from './store/database_v2' export { Index, IndexType } from './proto/db3_database_v2' diff --git a/sdk/tests/client_v2.test.ts b/sdk/tests/client_v2.test.ts index 39fbd313b..982e392ff 100644 --- a/sdk/tests/client_v2.test.ts +++ b/sdk/tests/client_v2.test.ts @@ -63,7 +63,7 @@ describe('test db3.js client module', () => { async function createTestClient() { const db3_account = createRandomAccount() const client = createClient( - 'http://scoll.rollup.testnet.db3.network', + 'http://127.0.0.1:26619', 'http://127.0.0.1:26639', db3_account ) From 076f3cef2cde65500c1807e2d1e229aee671a7fc Mon Sep 17 00:00:00 2001 From: imotai Date: Mon, 31 Jul 2023 21:26:39 +0800 Subject: [PATCH 3/4] fix: fix the empty log bugs --- src/event/src/event_processor.rs | 158 +++++++++++--------- src/event/src/meta_store_event_processor.rs | 135 +++++++++-------- src/node/src/rollup_executor.rs | 11 +- 3 files changed, 164 insertions(+), 140 deletions(-) diff --git a/src/event/src/event_processor.rs b/src/event/src/event_processor.rs index 68a3eb171..7932b8c82 100644 --- a/src/event/src/event_processor.rs +++ b/src/event/src/event_processor.rs @@ -29,6 +29,7 @@ use ethers::{ use std::collections::HashSet; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; +use tokio::time::{sleep, Duration}; use tracing::{info, warn}; #[derive(Debug)] @@ -43,7 +44,6 @@ pub struct EventProcessorConfig { pub struct EventProcessor { config: EventProcessorConfig, - provider: Arc>, running: Arc, db_store: DBStoreV2, block_number: Arc, @@ -56,13 +56,8 @@ unsafe impl Send for EventProcessor {} impl EventProcessor { pub async fn new(config: EventProcessorConfig, db_store: DBStoreV2) -> Result { info!("new event processor with config {:?}", config); - let provider = Provider::::connect(&config.evm_node_url) - .await - .map_err(|e| DB3Error::StoreEventError(format!("{e}")))?; - let provider_arc = Arc::new(provider); Ok(Self { config, - provider: provider_arc, running: Arc::new(AtomicBool::new(false)), db_store, block_number: Arc::new(AtomicU64::new(0)), @@ -91,8 +86,6 @@ impl EventProcessor { } pub async fn start(&self) -> Result<()> { - let abi: Abi = serde_json::from_str(self.config.abi.as_str()) - .map_err(|e| DB3Error::StoreEventError(format!("{e}")))?; self.running .store(true, std::sync::atomic::Ordering::Relaxed); let address = self @@ -102,28 +95,10 @@ impl EventProcessor { .map_err(|e| DB3Error::StoreEventError(format!("{e}")))?; let db_addr = DB3Address::from_hex(self.config.db_addr.as_str()) .map_err(|e| DB3Error::StoreEventError(format!("{e}")))?; - let filter = match self.config.start_block == 0 { - true => Filter::new().address(address), - false => { - info!( - "start process contract from block {} with address {}", - self.config.start_block, self.config.contract_addr - ); - Filter::new() - .from_block(self.config.start_block) - .address(address) - } - }; - let mut stream = self - .provider - .subscribe_logs(&filter) - .await + let abi: Abi = serde_json::from_str(self.config.abi.as_str()) .map_err(|e| DB3Error::StoreEventError(format!("{e}")))?; - info!( - "event processor for contract {}", - self.config.contract_addr.as_str() - ); - while let Some(log) = stream.next().await { + let local_evm_node_url = self.config.evm_node_url.to_string(); + loop { if !self.running.load(Ordering::Relaxed) { info!( "stop event processor for contract {}", @@ -131,59 +106,100 @@ impl EventProcessor { ); break; } - if let Some(number) = log.block_number { - if number.as_u64() % 10 == 0 { + let provider = + Provider::::connect_with_reconnects(local_evm_node_url.as_str(), 100) + .await + .map_err(|e| DB3Error::StoreEventError(format!("{e}")))?; + let provider_arc = Arc::new(provider); + let filter = match self.config.start_block == 0 { + true => Filter::new().address(address), + false => { info!( - "contract {} sync status block {} event number {}", - self.config.contract_addr.as_str(), - self.block_number.load(Ordering::Relaxed), - self.event_number.load(Ordering::Relaxed) + "start process contract from block {} with address {}", + self.config.start_block, self.config.contract_addr ); + Filter::new() + .from_block(self.config.start_block) + .address(address) } - self.block_number.store(number.as_u64(), Ordering::Relaxed) - } - for e in abi.events() { - // verify - let event_signature = log - .topics - .get(0) - .ok_or(DB3Error::StoreEventError(format!("")))?; + }; - if event_signature != &e.signature() { - continue; - } - if !self.config.target_events.contains(e.name.as_str()) { - continue; - } - let raw_log = RawLog { - topics: log.topics.clone(), - data: log.data.to_vec(), - }; - if let Ok(log_entry) = e.parse_log(raw_log) { - let json_value = Self::log_to_doc(&log_entry); - match serde_json::to_string(&json_value) { - Ok(value) => { - let values = vec![value.to_string()]; - if let Err(e) = self.db_store.add_docs( - &db_addr, - &DB3Address::ZERO, - e.name.as_str(), - &values, - None, - ) { - warn!("fail to write json doc {} for {e}", value.as_str()); - } else { - self.event_number.fetch_add(1, Ordering::Relaxed); + if let Ok(mut stream) = provider_arc.clone().subscribe_logs(&filter).await { + loop { + if let Some(log) = stream.next().await { + if !self.running.load(Ordering::Relaxed) { + info!( + "stop event processor for contract {}", + self.config.contract_addr.as_str() + ); + break; + } + if let Some(number) = log.block_number { + if number.as_u64() % 10 == 0 { + info!( + "contract {} sync status block {} event number {}", + self.config.contract_addr.as_str(), + self.block_number.load(Ordering::Relaxed), + self.event_number.load(Ordering::Relaxed) + ); } + self.block_number.store(number.as_u64(), Ordering::Relaxed) } - Err(e) => { - warn!("fail to convert to json for {e} "); + for e in abi.events() { + // verify + let event_signature = log + .topics + .get(0) + .ok_or(DB3Error::StoreEventError(format!("")))?; + if event_signature != &e.signature() { + continue; + } + if !self.config.target_events.contains(e.name.as_str()) { + continue; + } + let raw_log = RawLog { + topics: log.topics.clone(), + data: log.data.to_vec(), + }; + if let Ok(log_entry) = e.parse_log(raw_log) { + let json_value = Self::log_to_doc(&log_entry); + match serde_json::to_string(&json_value) { + Ok(value) => { + let values = vec![value.to_string()]; + if let Err(e) = self.db_store.add_docs( + &db_addr, + &DB3Address::ZERO, + e.name.as_str(), + &values, + None, + ) { + warn!( + "fail to write json doc {} for {e}", + value.as_str() + ); + } else { + self.event_number.fetch_add(1, Ordering::Relaxed); + } + } + Err(e) => { + warn!("fail to convert to json for {e} "); + } + } + break; + } } + } else { + warn!("empty log from stream, sleep 5 seconds and reconnect to it"); + sleep(Duration::from_millis(5 * 1000)).await; + break; } - break; } + } else { + warn!("fail to subscribe the log, sleep 5 seconds and reconnect to it"); + sleep(Duration::from_millis(5 * 1000)).await; } } + Ok(()) } diff --git a/src/event/src/meta_store_event_processor.rs b/src/event/src/meta_store_event_processor.rs index 6ef8b68c3..c2eda0773 100644 --- a/src/event/src/meta_store_event_processor.rs +++ b/src/event/src/meta_store_event_processor.rs @@ -347,10 +347,6 @@ impl MetaStoreEventProcessor { evm_node_url ); self.last_running.store(Some(running.clone())); - let provider = Provider::::connect_with_reconnects(evm_node_url, 100) - .await - .map_err(|e| DB3Error::StoreEventError(format!("{e}")))?; - let provider = Arc::new(provider); let address = contract_addr .parse::
() .map_err(|_| DB3Error::InvalidAddress)?; @@ -391,81 +387,88 @@ impl MetaStoreEventProcessor { } }; let local_contract_addr = contract_addr.to_string(); + let local_evm_node_url = evm_node_url.to_string(); task::spawn(async move { loop { if !running.load(Ordering::Relaxed) { break; } - match provider.subscribe_logs(&filter).await { - Ok(mut stream) => loop { - if !running.load(Ordering::Relaxed) { - info!( - "stop event processor for contract {}", - local_contract_addr.as_str() - ); - break; - } - match stream.next().await { - Some(log) => { + if let Ok(provider) = + Provider::::connect_with_reconnects(local_evm_node_url.as_str(), 100).await + { + let provider = Arc::new(provider); + match provider.subscribe_logs(&filter).await { + Ok(mut stream) => loop { + if !running.load(Ordering::Relaxed) { info!( - "block number {:?} transacion {:?} sender address {:?} ", - log.block_number, log.transaction_hash, log.address + "stop event processor for contract {}", + local_contract_addr.as_str() ); - if let (Ok(Some(transaction)), Ok(event_signature)) = ( - provider - .get_transaction(log.transaction_hash.unwrap()) - .await - .map_err(|e| { - DB3Error::StoreEventError(format!( - "fail to get transaction {e}" - )) - }), - log.topics.get(0).ok_or(DB3Error::StoreEventError(format!( - "fail to get topics" - ))), - ) { - if event_signature == &(CreateDatabaseFilter::signature()) { - if let Ok(_) = Self::handle_create_doc_database( - &log, - &transaction, - network, - &wallet, - &local_state_store, - &local_storage, - &local_db_store, - &mutation_type, - ) - .await - {} - } else if event_signature - == &(CreateCollectionFilter::signature()) - { - if let Ok(_) = Self::handle_create_collection( - &log, - &transaction, - network, - &wallet, - &local_state_store, - &local_storage, - &local_db_store, - &mutation_type, - ) - .await - {} + break; + } + match stream.next().await { + Some(log) => { + info!( + "block number {:?} transacion {:?} sender address {:?} ", + log.block_number, log.transaction_hash, log.address + ); + if let (Ok(Some(transaction)), Ok(event_signature)) = ( + provider + .get_transaction(log.transaction_hash.unwrap()) + .await + .map_err(|e| { + DB3Error::StoreEventError(format!( + "fail to get transaction {e}" + )) + }), + log.topics.get(0).ok_or(DB3Error::StoreEventError( + format!("fail to get topics"), + )), + ) { + if event_signature == &(CreateDatabaseFilter::signature()) { + if let Ok(_) = Self::handle_create_doc_database( + &log, + &transaction, + network, + &wallet, + &local_state_store, + &local_storage, + &local_db_store, + &mutation_type, + ) + .await + {} + } else if event_signature + == &(CreateCollectionFilter::signature()) + { + if let Ok(_) = Self::handle_create_collection( + &log, + &transaction, + network, + &wallet, + &local_state_store, + &local_storage, + &local_db_store, + &mutation_type, + ) + .await + {} + } } } + None => { + warn!("empty log from stream, sleep 5 seconds and reconnect to it"); + sleep(Duration::from_millis(5 * 1000)).await; + break; + } } - None => { - warn!("empty log from stream"); - sleep(Duration::from_millis(5 * 1000)).await; - } + }, + Err(e) => { + warn!("fail get stream for error {e}"); } - }, - Err(e) => { - warn!("fail get stream for error {e}"); } + sleep(Duration::from_millis(5 * 1000)).await; } - sleep(Duration::from_millis(5 * 1000)).await; } warn!("the meta contract event listener exits"); }); diff --git a/src/node/src/rollup_executor.rs b/src/node/src/rollup_executor.rs index 95525dea9..2956eeaf9 100644 --- a/src/node/src/rollup_executor.rs +++ b/src/node/src/rollup_executor.rs @@ -259,7 +259,7 @@ impl RollupExecutor { { let network_id = self.network_id.load(Ordering::Relaxed); self.storage.flush_state()?; - let (last_start_block, last_end_block, tx) = + let (_last_start_block, last_end_block, tx) = match self.storage.get_last_rollup_record()? { Some(r) => (r.start_block, r.end_block, r.arweave_tx.to_string()), _ => (0_u64, 0_u64, "".to_string()), @@ -275,7 +275,7 @@ impl RollupExecutor { last_end_block ); self.pending_start_block - .store(last_start_block, Ordering::Relaxed); + .store(last_end_block, Ordering::Relaxed); self.pending_end_block .store(current_block, Ordering::Relaxed); let mutations = self @@ -373,7 +373,11 @@ mod tests { "start block {} end block {}", record.start_block, record.end_block ); + let result = storage.increase_block_return_last_state(); + assert_eq!(true, result.is_ok()); let block = NodeTestBase::add_mutations(&storage, 10); + let result = storage.increase_block_return_last_state(); + assert_eq!(true, result.is_ok()); let result = rollup_executor.process().await; assert_eq!(true, result.is_ok()); let result = storage.get_last_rollup_record(); @@ -383,7 +387,8 @@ mod tests { "start block {} end block {}", record.start_block, record.end_block ); - assert_eq!(record.end_block, block); + assert_eq!(record.end_block, 3); + assert_eq!(record.start_block, 1); } Err(e) => { assert!(false, "{e}"); From d15eb0dd149e25300cc7c41ce13b9926a82adf42 Mon Sep 17 00:00:00 2001 From: imotai Date: Mon, 7 Aug 2023 15:24:54 +0800 Subject: [PATCH 4/4] feat: add chatgpt test case --- src/storage/src/key_store.rs | 42 ++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/src/storage/src/key_store.rs b/src/storage/src/key_store.rs index 8cd145cae..4407b2aa8 100644 --- a/src/storage/src/key_store.rs +++ b/src/storage/src/key_store.rs @@ -98,8 +98,50 @@ mod tests { use super::*; use ethers::prelude::{LocalWallet, Signer}; use std::ops::Deref; + use std::fs; use tempdir::TempDir; + #[test] + fn test_has_key_existing() { + let tmp_dir_path = TempDir::new("tmp_dir").expect("create temp dir"); + let real_path = tmp_dir_path.path().to_str().unwrap().to_string(); + let keystore = KeyStore::new(KeyStoreConfig { key_root_path: real_path.clone() }); + let key = "test_key"; + let value = b"value"; + keystore.write_key(key, value).unwrap(); + assert_eq!(keystore.has_key(key), true); + } + + #[test] + fn test_has_key_non_existing() { + let tmp_dir_path = TempDir::new("tmp_dir").expect("create temp dir"); + let real_path = tmp_dir_path.path().to_str().unwrap().to_string(); + let keystore = KeyStore::new(KeyStoreConfig { key_root_path: real_path.clone() }); + let key = "test_key"; + assert_eq!(keystore.has_key(key), false); + } + + #[test] + fn test_has_key_invalid_input() { + let tmp_dir_path = TempDir::new("tmp_dir").expect("create temp dir"); + let real_path = tmp_dir_path.path().to_str().unwrap().to_string(); + let keystore = KeyStore::new(KeyStoreConfig { key_root_path: real_path.clone() }); + let empty_key = ""; + assert_eq!(keystore.has_key(empty_key), false); + } + + #[test] + fn test_has_key_file_error() { + let tmp_dir_path = TempDir::new("tmp_dir").expect("create temp dir"); + let real_path = tmp_dir_path.path().to_str().unwrap().to_string(); + let keystore = KeyStore::new(KeyStoreConfig { key_root_path: real_path.clone() }); + let mut invalid_file = std::fs::File::create(format!("{}/INVALID", real_path)).unwrap(); + invalid_file.write_all(b"invalid content").unwrap(); // create an invalid file + let invalid_key = "INVALID"; + assert_eq!(keystore.has_key(invalid_key), false); + } + + #[test] fn evm_account_key_store_smoke_test() { let tmp_dir_path = TempDir::new("key store").expect("create temp dir");