Merge f456773b72 into 53449e1faa

* Half log_page_size when it queries too many logs.

* fmt.

* Increase log broadcast channel size.

node/log_entry_sync/src/sync_manager/log_query.rs

@@ -14,6 +14,8 @@ use thiserror::Error;
 pub(crate) type PinBoxFut<'a, T> =
     Pin<Box<dyn Future<Output = Result<T, ProviderError>> + Send + 'a>>;
 
+const TOO_MANY_LOGS_ERROR_MSG: &str = "query returned more than";
+
 /// A log query provides streaming access to historical logs via a paginated
 /// request. For streaming access to future logs, use [`Middleware::watch`] or
 /// [`Middleware::subscribe_logs`]
@@ -21,6 +23,9 @@ pub struct LogQuery<'a, P> {
     provider: &'a Provider<P>,
     filter: Filter,
     from_block: Option<U64>,
+
+    expected_page_size: u64,
+    /// It may be smaller than `expected_page_size` if the server cannot return all the logs.
     page_size: u64,
     current_logs: VecDeque<Log>,
     last_block: Option<U64>,
@@ -31,7 +36,8 @@ pub struct LogQuery<'a, P> {
 enum LogQueryState<'a> {
     Initial,
     LoadLastBlock(PinBoxFut<'a, U64>),
-    LoadLogs(PinBoxFut<'a, Vec<Log>>),
+    /// `(from_block, get_logs_fut)`. `from_block` is used to resume if the request fails.
+    LoadLogs((Option<U64>, PinBoxFut<'a, Vec<Log>>)),
     Consume,
 }
 
@@ -45,6 +51,7 @@ where
             provider,
             filter: filter.clone(),
             from_block: filter.get_from_block(),
+            expected_page_size: 10000,
             page_size: 10000,
             current_logs: VecDeque::new(),
             last_block: None,
@@ -56,6 +63,7 @@ where
     /// set page size for pagination
     pub fn with_page_size(mut self, page_size: u64) -> Self {
         self.page_size = page_size;
+        self.expected_page_size = page_size;
         self
     }
 }
@@ -98,7 +106,7 @@ where
                         tokio::time::sleep(delay).await;
                         provider.get_logs(&filter).await
                     });
-                    rewake_with_new_state!(ctx, self, LogQueryState::LoadLogs(fut));
+                    rewake_with_new_state!(ctx, self, LogQueryState::LoadLogs((None, fut)));
                 } else {
                     // if paginatable, load last block
                     let fut = match self.filter.get_to_block() {
@@ -134,18 +142,33 @@ where
                             tokio::time::sleep(delay).await;
                             provider.get_logs(&filter).await
                         });
-                        rewake_with_new_state!(ctx, self, LogQueryState::LoadLogs(fut));
+                        rewake_with_new_state!(
+                            ctx,
+                            self,
+                            LogQueryState::LoadLogs((Some(from_block), fut))
+                        );
                     }
                     Err(err) => Poll::Ready(Some(Err(LogQueryError::LoadLastBlockError(err)))),
                 }
             }
-            LogQueryState::LoadLogs(fut) => match futures_util::ready!(fut.as_mut().poll(ctx)) {
+            LogQueryState::LoadLogs((from_block, fut)) => {
-                Ok(logs) => {
+                match futures_util::ready!(fut.as_mut().poll(ctx)) {
-                    self.current_logs = VecDeque::from(logs);
+                    Ok(logs) => {
-                    rewake_with_new_state!(ctx, self, LogQueryState::Consume);
+                        self.current_logs = VecDeque::from(logs);
+                        self.page_size = self.expected_page_size;
+                        rewake_with_new_state!(ctx, self, LogQueryState::Consume);
+                    }
+                    Err(err) => {
+                        if err.to_string().contains(TOO_MANY_LOGS_ERROR_MSG) {
+                            self.from_block = *from_block;
+                            self.page_size /= 2;
+                            rewake_with_new_state!(ctx, self, LogQueryState::Consume);
+                        } else {
+                            Poll::Ready(Some(Err(LogQueryError::LoadLogsError(err))))
+                        }
+                    }
                 }
-                Err(err) => Poll::Ready(Some(Err(LogQueryError::LoadLogsError(err)))),
+            }
-            },
             LogQueryState::Consume => {
                 let log = self.current_logs.pop_front();
                 if log.is_none() {
@@ -183,7 +206,11 @@ where
                             provider.get_logs(&filter).await
                         });
 
-                        rewake_with_new_state!(ctx, self, LogQueryState::LoadLogs(fut));
+                        rewake_with_new_state!(
+                            ctx,
+                            self,
+                            LogQueryState::LoadLogs((Some(from_block), fut))
+                        );
                     }
                 } else {
                     Poll::Ready(log.map(Ok))

node/log_entry_sync/src/sync_manager/mod.rs

@@ -21,7 +21,8 @@ use tokio::sync::{oneshot, RwLock};
 const RETRY_WAIT_MS: u64 = 500;
 
 // A RPC query can return at most 10000 entries.
-const BROADCAST_CHANNEL_CAPACITY: usize = 10000;
+// Each tx has less than 10KB, so the cache size should be acceptable.
+const BROADCAST_CHANNEL_CAPACITY: usize = 25000;
 const CATCH_UP_END_GAP: u64 = 10;
 
 #[derive(Clone, Debug)]

scripts/update_config.sh

@@ -1,31 +0,0 @@
-#!/bin/bash
-
-MINER_KEY=""
-MINE_CONTRACT=""
-BLOCKCHAIN_RPC=""
-FLOW_CONTRACT=""
-BLOCK_NUMBER=0
-PUBLIC_IP=$(curl -s https://ipinfo.io/ip)
-
-FILE=run/config.toml
-
-# enable sync
-sed -in-place='' 's/# \[sync\]/\[sync\]/g' $FILE
-# enable auto_sync
-sed -in-place='' 's/# auto_sync_enabled = false/auto_sync_enabled = true/g' $FILE
-# reduce timeout for finding peers
-sed -in-place='' 's/# find_peer_timeout = .*/find_peer_timeout = "10s"/g' $FILE
-# set public ip
-sed -in-place='' "s/# network_enr_address = .*/network_enr_address = \"$PUBLIC_IP\"/g" $FILE
-# set miner key
-sed -in-place='' "s/miner_key = \"\"/miner_key = \"$MINER_KEY\"/g" $FILE
-# set miner contract address
-sed -in-place='' "s/mine_contract_address = .*/mine_contract_address = \"$MINE_CONTRACT\"/g" $FILE
-# set blockchain rpc endpoint
-sed -in-place='' "s|blockchain_rpc_endpoint = .*|blockchain_rpc_endpoint = \"$BLOCKCHAIN_RPC\"|g" $FILE
-# set flow contract address
-sed -in-place='' "s/log_contract_address = .*/log_contract_address = \"$FLOW_CONTRACT\"/g" $FILE
-# set contract deployed block number
-sed -in-place='' "s/log_sync_start_block_number = .*/log_sync_start_block_number = $BLOCK_NUMBER/g" $FILE
-# update the boot node ids
-sed -in-place='' 's|network_boot_nodes = .*|network_boot_nodes = ["/ip4/54.219.26.22/udp/1234/p2p/16Uiu2HAmTVDGNhkHD98zDnJxQWu3i1FL1aFYeh9wiQTNu4pDCgps","/ip4/52.52.127.117/udp/1234/p2p/16Uiu2HAkzRjxK2gorngB1Xq84qDrT4hSVznYDHj6BkbaE4SGx9oS","/ip4/18.167.69.68/udp/1234/p2p/16Uiu2HAm2k6ua2mGgvZ8rTMV8GhpW71aVzkQWy7D37TTDuLCpgmX"]|g' $FILE