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?