Compare commits

...

4 Commits

Author SHA1 Message Date
Peter Zhang
4127333297 update next tx seq 2024-09-02 11:15:51 +08:00
Peter Zhang
b82183bf09 update next tx seq 2024-09-02 10:43:04 +08:00
Peter Zhang
e75f58c1ee update next tx seq 2024-09-02 09:48:37 +08:00
Peter Zhang
0734ec9886 add log 2024-09-02 07:32:52 +08:00
2 changed files with 14 additions and 7 deletions

View File

@ -1,6 +1,6 @@
use ethers::prelude::{Filter, JsonRpcClient, Log, Middleware, Provider, ProviderError, U64}; use ethers::prelude::{Filter, JsonRpcClient, Log, Middleware, Provider, ProviderError, U64};
use futures_core::stream::Stream; use futures_core::stream::Stream;
use jsonrpsee::tracing::trace; use jsonrpsee::tracing::{debug, error, trace};
use std::future::Future; use std::future::Future;
use std::time::Duration; use std::time::Duration;
use std::{ use std::{
@ -152,6 +152,7 @@ where
} }
} }
LogQueryState::LoadLogs((from_block, fut)) => { LogQueryState::LoadLogs((from_block, fut)) => {
debug!("LoadLogs: loading logs from block={:?}", from_block);
match futures_util::ready!(fut.as_mut().poll(ctx)) { match futures_util::ready!(fut.as_mut().poll(ctx)) {
Ok(logs) => { Ok(logs) => {
self.current_logs = VecDeque::from(logs); self.current_logs = VecDeque::from(logs);
@ -163,6 +164,10 @@ where
if err.to_string().contains(msg) { if err.to_string().contains(msg) {
self.from_block = *from_block; self.from_block = *from_block;
self.page_size /= 2; self.page_size /= 2;
error!(
"log_query: page size to large at {} reducing page size to {}",
self.from_block.unwrap(), self.page_size
);
rewake_with_new_state!(ctx, self, LogQueryState::Consume); rewake_with_new_state!(ctx, self, LogQueryState::Consume);
} }
} }

View File

@ -5,7 +5,7 @@ use anyhow::{anyhow, bail, Result};
use ethereum_types::H256; use ethereum_types::H256;
use ethers::{prelude::Middleware, types::BlockNumber}; use ethers::{prelude::Middleware, types::BlockNumber};
use futures::FutureExt; use futures::FutureExt;
use jsonrpsee::tracing::{debug, error, trace, warn}; use jsonrpsee::tracing::{debug, error, warn};
use shared_types::{ChunkArray, Transaction}; use shared_types::{ChunkArray, Transaction};
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::fmt::Debug; use std::fmt::Debug;
@ -56,7 +56,7 @@ impl LogSyncManager {
store: Arc<dyn Store>, store: Arc<dyn Store>,
) -> Result<(broadcast::Sender<LogSyncEvent>, oneshot::Receiver<()>)> { ) -> Result<(broadcast::Sender<LogSyncEvent>, oneshot::Receiver<()>)> {
let next_tx_seq = store.next_tx_seq(); let next_tx_seq = store.next_tx_seq();
debug!("LogSyncManager spawn next_tx_seq: {}", next_tx_seq);
let executor_clone = executor.clone(); let executor_clone = executor.clone();
let mut shutdown_sender = executor.shutdown_sender(); let mut shutdown_sender = executor.shutdown_sender();
@ -251,8 +251,8 @@ impl LogSyncManager {
} }
std::cmp::Ordering::Greater => { std::cmp::Ordering::Greater => {
error!( error!(
"Unexpected transaction seq: next={} get={}", "Unexpected transaction tx={:?}",
self.next_tx_seq, tx.seq tx,
); );
false false
} }
@ -298,7 +298,7 @@ impl LogSyncManager {
async fn handle_data(&mut self, mut rx: UnboundedReceiver<LogFetchProgress>) -> Result<()> { async fn handle_data(&mut self, mut rx: UnboundedReceiver<LogFetchProgress>) -> Result<()> {
while let Some(data) = rx.recv().await { while let Some(data) = rx.recv().await {
trace!("handle_data: data={:?}", data); debug!("handle_data: data={:?}", data);
match data { match data {
LogFetchProgress::SyncedBlock(( LogFetchProgress::SyncedBlock((
block_number, block_number,
@ -387,7 +387,7 @@ impl LogSyncManager {
} }
} }
self.data_cache.garbage_collect(self.next_tx_seq); self.data_cache.garbage_collect(self.next_tx_seq);
self.next_tx_seq += 1; self.next_tx_seq = self.store.next_tx_seq();
true true
} }
} }
@ -427,6 +427,7 @@ impl LogSyncManager {
&executor_clone, &executor_clone,
Duration::from_millis(self.config.recover_query_delay), Duration::from_millis(self.config.recover_query_delay),
); );
self.next_tx_seq = self.store.next_tx_seq();
self.handle_data(recover_rx).await?; self.handle_data(recover_rx).await?;
} }
@ -437,6 +438,7 @@ impl LogSyncManager {
self.config.default_finalized_block_count, self.config.default_finalized_block_count,
self.config.remove_finalized_block_interval_minutes, self.config.remove_finalized_block_interval_minutes,
); );
Ok(()) Ok(())
} }
} }