smeb_y/0g-storage-node
1
0
Fork 0
mirror of https://github.com/0glabs/0g-storage-node.git synced 2024-11-10 10:05:17 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

queyr logs via LogQuery in wath loop

Browse Source
This commit is contained in:
Joel 2024-09-02 11:24:59 +08:00
parent e3c199d361
commit 5a33b0fc20
2 changed files with 15 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs
View File

@ -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<Provider<RetryClient<Http>>>,
block_hash_cache: &Arc<RwLock<BTreeMap<u64, Option<BlockHashAndSubmissionIndex>>>>,
log_page_size: u64,
) -> Result<Option<(u64, H256, Option<Option<u64>>)>> {
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<u64, Vec<Log>> = 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;
}
}

3
node/log_entry_sync/src/sync_manager/mod.rs
View File

@ -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?