From 5a33b0fc20eeb8ccf175daf4a459d1c47b2c27f9 Mon Sep 17 00:00:00 2001 From: Joel Date: Mon, 2 Sep 2024 11:24:59 +0800 Subject: [PATCH 1/6] queyr logs via LogQuery in wath loop --- .../src/sync_manager/log_entry_fetcher.rs | 22 ++++++++++++------- node/log_entry_sync/src/sync_manager/mod.rs | 3 +-- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs b/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs index 3773b24..8031997 100644 --- a/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs +++ b/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs @@ -290,6 +290,7 @@ impl LogEntryFetcher { let contract = ZgsFlow::new(self.contract_address, self.provider.clone()); let provider = self.provider.clone(); let confirmation_delay = self.confirmation_delay; + let log_page_size = self.log_page_size; executor.spawn( async move { debug!("start_watch starts, start={}", start_block_number); @@ -305,6 +306,7 @@ impl LogEntryFetcher { confirmation_delay, &contract, &block_hash_cache, + log_page_size, ) .await { @@ -340,6 +342,7 @@ impl LogEntryFetcher { confirmation_delay: u64, contract: &ZgsFlow>>, block_hash_cache: &Arc>>>, + log_page_size: u64, ) -> Result>)>> { let latest_block_number = provider.get_block_number().await?.as_u64(); debug!( @@ -408,8 +411,11 @@ impl LogEntryFetcher { .to_block(to_block_number) .address(contract.address().into()) .filter; + let mut stream = LogQuery::new(&provider, &filter, Duration::from_millis(10)) + .with_page_size(log_page_size); let mut block_logs: BTreeMap> = BTreeMap::new(); - for log in provider.get_logs(&filter).await? { + while let Some(maybe_log) = stream.next().await { + let log = maybe_log?; let block_number = log .block_number .ok_or_else(|| anyhow!("block number missing"))? @@ -496,20 +502,20 @@ impl LogEntryFetcher { } else { None }; + for log in log_events.into_iter() { + if let Err(e) = watch_tx.send(log) { + warn!("send LogFetchProgress::Transaction failed: {:?}", e); + return Ok(progress); + } + } if let Some(p) = &new_progress { if let Err(e) = watch_tx.send(LogFetchProgress::SyncedBlock(*p)) { - warn!("send LogFetchProgress failed: {:?}", e); + warn!("send LogFetchProgress::SyncedBlock failed: {:?}", e); return Ok(progress); } else { block_hash_cache.write().await.insert(p.0, None); } } - for log in log_events.into_iter() { - if let Err(e) = watch_tx.send(log) { - warn!("send log failed: {:?}", e); - return Ok(progress); - } - } progress = new_progress; } } diff --git a/node/log_entry_sync/src/sync_manager/mod.rs b/node/log_entry_sync/src/sync_manager/mod.rs index c561676..401462f 100644 --- a/node/log_entry_sync/src/sync_manager/mod.rs +++ b/node/log_entry_sync/src/sync_manager/mod.rs @@ -339,8 +339,7 @@ impl LogSyncManager { LogFetchProgress::Transaction(tx) => { if !self.put_tx(tx.clone()).await { // Unexpected error. - error!("log sync write error"); - break; + bail!("log sync write error"); } if let Err(e) = self.event_send.send(LogSyncEvent::TxSynced { tx }) { // TODO: Do we need to wait until all receivers are initialized? From 64a34e49f7319a9dbe5c9f09cc4f838fe2254e43 Mon Sep 17 00:00:00 2001 From: Joel Date: Mon, 2 Sep 2024 11:30:33 +0800 Subject: [PATCH 2/6] fix lints error --- node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs b/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs index 8031997..ba22481 100644 --- a/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs +++ b/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs @@ -411,7 +411,7 @@ impl LogEntryFetcher { .to_block(to_block_number) .address(contract.address().into()) .filter; - let mut stream = LogQuery::new(&provider, &filter, Duration::from_millis(10)) + let mut stream = LogQuery::new(provider, &filter, Duration::from_millis(10)) .with_page_size(log_page_size); let mut block_logs: BTreeMap> = BTreeMap::new(); while let Some(maybe_log) = stream.next().await { From 553d9c9249abf97a28350e367c019e47f95f79b1 Mon Sep 17 00:00:00 2001 From: Joel Date: Mon, 2 Sep 2024 14:42:55 +0800 Subject: [PATCH 3/6] Configre for force sync log from start block number --- node/log_entry_sync/src/sync_manager/config.rs | 4 ++++ node/log_entry_sync/src/sync_manager/mod.rs | 12 ++++++++++-- node/src/config/convert.rs | 1 + node/src/config/mod.rs | 1 + 4 files changed, 16 insertions(+), 2 deletions(-) diff --git a/node/log_entry_sync/src/sync_manager/config.rs b/node/log_entry_sync/src/sync_manager/config.rs index 15d540c..d9d281f 100644 --- a/node/log_entry_sync/src/sync_manager/config.rs +++ b/node/log_entry_sync/src/sync_manager/config.rs @@ -32,6 +32,8 @@ pub struct LogSyncConfig { pub remove_finalized_block_interval_minutes: u64, // watch_loop (eth_getLogs) trigger interval pub watch_loop_wait_time_ms: u64, + // force to sync log from start block number + pub force_log_sync_from_start_block_number: bool, } #[derive(Clone)] @@ -58,6 +60,7 @@ impl LogSyncConfig { default_finalized_block_count: u64, remove_finalized_block_interval_minutes: u64, watch_loop_wait_time_ms: u64, + force_log_sync_from_start_block_number: bool, ) -> Self { Self { rpc_endpoint_url, @@ -73,6 +76,7 @@ impl LogSyncConfig { default_finalized_block_count, remove_finalized_block_interval_minutes, watch_loop_wait_time_ms, + force_log_sync_from_start_block_number, } } } diff --git a/node/log_entry_sync/src/sync_manager/mod.rs b/node/log_entry_sync/src/sync_manager/mod.rs index 401462f..f85d7d7 100644 --- a/node/log_entry_sync/src/sync_manager/mod.rs +++ b/node/log_entry_sync/src/sync_manager/mod.rs @@ -102,7 +102,14 @@ impl LogSyncManager { block_hash_cache, }; - let (mut start_block_number, mut start_block_hash) = + let (mut start_block_number, mut start_block_hash) = if log_sync_manager + .config + .force_log_sync_from_start_block_number + { + let block_number = log_sync_manager.config.start_block_number; + let block_hash = log_sync_manager.get_block(block_number.into()).await?.1; + (block_number, block_hash) + } else { match log_sync_manager.store.get_sync_progress()? { // No previous progress, so just use config. None => { @@ -112,7 +119,8 @@ impl LogSyncManager { (block_number, block_hash) } Some((block_number, block_hash)) => (block_number, block_hash), - }; + } + }; let (mut finalized_block_number, mut finalized_block_hash) = match log_sync_manager.get_block(BlockNumber::Finalized).await { diff --git a/node/src/config/convert.rs b/node/src/config/convert.rs index 450fbd0..58c0152 100644 --- a/node/src/config/convert.rs +++ b/node/src/config/convert.rs @@ -156,6 +156,7 @@ impl ZgsConfig { self.default_finalized_block_count, self.remove_finalized_block_interval_minutes, self.watch_loop_wait_time_ms, + self.force_log_sync_from_start_block_number, )) } diff --git a/node/src/config/mod.rs b/node/src/config/mod.rs index ff6ab1c..e5d2bf1 100644 --- a/node/src/config/mod.rs +++ b/node/src/config/mod.rs @@ -33,6 +33,7 @@ build_config! { (blockchain_rpc_endpoint, (String), "http://127.0.0.1:8545".to_string()) (log_contract_address, (String), "".to_string()) (log_sync_start_block_number, (u64), 0) + (force_log_sync_from_start_block_number, (bool), false) (confirmation_block_count, (u64), 12) (log_page_size, (u64), 999) (max_cache_data_size, (usize), 100 * 1024 * 1024) // 100 MB From 6f848c5cf2e8b2639a95a0694f49f1f45ab684ff Mon Sep 17 00:00:00 2001 From: Joel Date: Mon, 2 Sep 2024 14:51:48 +0800 Subject: [PATCH 4/6] update log --- node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs | 2 ++ node/log_entry_sync/src/sync_manager/mod.rs | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs b/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs index ba22481..41ffd1f 100644 --- a/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs +++ b/node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs @@ -491,6 +491,8 @@ impl LogEntryFetcher { log_events.push(submission_event_to_transaction(submit_filter)); } + + info!("synced {} events", log_events.len()); } let new_progress = if block.hash.is_some() && block.number.is_some() { diff --git a/node/log_entry_sync/src/sync_manager/mod.rs b/node/log_entry_sync/src/sync_manager/mod.rs index f85d7d7..753bfd7 100644 --- a/node/log_entry_sync/src/sync_manager/mod.rs +++ b/node/log_entry_sync/src/sync_manager/mod.rs @@ -306,7 +306,7 @@ impl LogSyncManager { async fn handle_data(&mut self, mut rx: UnboundedReceiver) -> Result<()> { while let Some(data) = rx.recv().await { - trace!("handle_data: data={:?}", data); + debug!("handle_data: data={:?}", data); match data { LogFetchProgress::SyncedBlock(( block_number, From a0b1d815cb30fb131b5c0eea79b9718e7bef154d Mon Sep 17 00:00:00 2001 From: Joel Date: Mon, 2 Sep 2024 15:00:13 +0800 Subject: [PATCH 5/6] fix warn --- node/log_entry_sync/src/sync_manager/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/node/log_entry_sync/src/sync_manager/mod.rs b/node/log_entry_sync/src/sync_manager/mod.rs index 753bfd7..241a2c5 100644 --- a/node/log_entry_sync/src/sync_manager/mod.rs +++ b/node/log_entry_sync/src/sync_manager/mod.rs @@ -5,7 +5,7 @@ use anyhow::{anyhow, bail, Result}; use ethereum_types::H256; use ethers::{prelude::Middleware, types::BlockNumber}; use futures::FutureExt; -use jsonrpsee::tracing::{debug, error, trace, warn}; +use jsonrpsee::tracing::{debug, error, warn}; use shared_types::{ChunkArray, Transaction}; use std::collections::BTreeMap; use std::fmt::Debug; From f58eed1c5bce2d227324180e5ea6272a618f5bd8 Mon Sep 17 00:00:00 2001 From: Joel Date: Fri, 6 Sep 2024 15:28:48 +0800 Subject: [PATCH 6/6] resolve conflicts --- node/log_entry_sync/src/sync_manager/mod.rs | 9 +++++++++ node/src/config/mod.rs | 2 +- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/node/log_entry_sync/src/sync_manager/mod.rs b/node/log_entry_sync/src/sync_manager/mod.rs index 90bda0b..0055fb4 100644 --- a/node/log_entry_sync/src/sync_manager/mod.rs +++ b/node/log_entry_sync/src/sync_manager/mod.rs @@ -472,6 +472,15 @@ impl LogSyncManager { async fn get_start_block_number_with_hash( log_sync_manager: &LogSyncManager, ) -> Result<(u64, H256), anyhow::Error> { + if log_sync_manager + .config + .force_log_sync_from_start_block_number + { + let block_number = log_sync_manager.config.start_block_number; + let block_hash = log_sync_manager.get_block(block_number.into()).await?.1; + return Ok((block_number, block_hash)); + } + if let Some(block_number) = log_sync_manager.store.get_log_latest_block_number()? { if let Some(Some(val)) = log_sync_manager .block_hash_cache diff --git a/node/src/config/mod.rs b/node/src/config/mod.rs index e5d2bf1..0373fca 100644 --- a/node/src/config/mod.rs +++ b/node/src/config/mod.rs @@ -34,7 +34,7 @@ build_config! { (log_contract_address, (String), "".to_string()) (log_sync_start_block_number, (u64), 0) (force_log_sync_from_start_block_number, (bool), false) - (confirmation_block_count, (u64), 12) + (confirmation_block_count, (u64), 3) (log_page_size, (u64), 999) (max_cache_data_size, (usize), 100 * 1024 * 1024) // 100 MB (cache_tx_seq_ttl, (usize), 500)