Merge pull request #49 from RCasatta/timeouts

RCasatta · web-flow · commit 819ef5ec6b9a · 2025-09-17T12:13:36.000+02:00
Timeouts and logging
diff --git a/src/fetch.rs b/src/fetch.rs
@@ -57,7 +57,8 @@ const BS: &str = "https://blockstream.info";
 const LOCAL: &str = "http://127.0.0.1";
 
 impl Client {
-    pub fn new(args: &Arguments) -> Client {
+    pub fn new(args: &Arguments) -> Result<Client> {
+        args.is_valid()?;
         let esplora_url = match args.network {
             Network::Liquid => args
                 .esplora_url
@@ -89,13 +90,19 @@ impl Client {
             node_url.unwrap_or(format!("{LOCAL}:{port}"))
         };
         log::info!("connecting to {base_url}");
-        Client {
-            client: reqwest::Client::new(),
+        let client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(args.request_timeout_seconds))
+            .connect_timeout(Duration::from_secs(args.request_timeout_seconds)) // Connection establishment timeout
+            .build()
+            .with_context(|| "Failed to create HTTP client with timeout")?;
+
+        Ok(Client {
+            client,
             use_esplora,
             base_url,
             esplora_url,
             rpc_user_password: args.rpc_user_password.clone(),
-        }
+        })
     }
 
     // `curl http://127.0.0.1:7041/rest/blockhashbyheight/0.hex`
@@ -488,9 +495,10 @@ mod test {
         let _ = env_logger::try_init();
         let mut args = Arguments::default();
         args.use_esplora = true;
+        args.request_timeout_seconds = 30;
         for network in [Network::Bitcoin, Network::Liquid, Network::LiquidTestnet] {
             args.network = network;
-            let client = Client::new(&args);
+            let client = Client::new(&args).unwrap();
             tokio::time::sleep(std::time::Duration::from_secs(1)).await;
             test(client, network).await;
         }
@@ -505,7 +513,9 @@ mod test {
         args.use_esplora = false;
         args.network = Network::ElementsRegtest;
         args.node_url = Some(elementsd.rpc_url());
-        let client = Client::new(&args);
+        args.request_timeout_seconds = 10;
+        args.rpc_user_password = Some(elementsd.params.cookie_file.to_string_lossy().to_string());
+        let client = Client::new(&args).unwrap();
         test(client, args.network).await;
     }
 
@@ -535,7 +545,7 @@ mod test {
         let mut args = Arguments::default();
         args.use_esplora = false;
         args.network = network;
-        Client::new(&args)
+        Client::new(&args).unwrap()
     }
 
     #[tokio::test]
@@ -548,7 +558,9 @@ mod test {
         args.use_esplora = false;
         args.network = Network::BitcoinRegtest;
         args.node_url = Some(bitcoind.rpc_url());
-        let client = Client::new(&args);
+        args.request_timeout_seconds = 10;
+        args.rpc_user_password = Some(bitcoind.params.cookie_file.to_string_lossy().to_string());
+        let client = Client::new(&args).unwrap();
         test(client, args.network).await;
     }
 
diff --git a/src/server/mod.rs b/src/server/mod.rs
@@ -117,6 +117,10 @@ pub struct Arguments {
     /// Cache control duration in seconds for waterfalls endpoints. Set to 0 to disable cache control headers.
     #[arg(env, long, default_value = "5")]
     pub cache_control_seconds: u32,
+
+    /// Timeout in seconds for connect and for reques HTTP requests and  to the node or esplora
+    #[arg(env, long, default_value = "30")]
+    pub request_timeout_seconds: u64,
 }
 
 impl Arguments {
@@ -125,6 +129,10 @@ impl Arguments {
             Err(Error::String(
                 "When using the node you must specify --rpc-user-password".to_string(),
             ))
+        } else if self.request_timeout_seconds == 0 {
+            Err(Error::String(
+                "Request timeout must be greater than 0".to_string(),
+            ))
         } else {
             Ok(())
         }
@@ -264,7 +272,6 @@ pub async fn inner_main(
     shutdown_signal: impl Future<Output = ()>,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     log::info!("starting waterfalls");
-    args.is_valid()?;
 
     let store = get_store(&args)?;
 
@@ -307,7 +314,8 @@ pub async fn inner_main(
 
     let h1 = {
         let state = state.clone();
-        let client: Client = Client::new(&args);
+        let client: Client =
+            Client::new(&args).unwrap_or_else(|e| error_panic!("Failed to create client: {e}"));
         let shutdown_rx = shutdown_tx.subscribe();
         tokio::spawn(async move {
             let shutdown_future = async {
@@ -328,7 +336,8 @@ pub async fn inner_main(
 
     let h2 = {
         let state = state.clone();
-        let client = Client::new(&args);
+        let client =
+            Client::new(&args).unwrap_or_else(|e| error_panic!("Failed to create client: {e}"));
         let shutdown_rx = shutdown_tx.subscribe();
         tokio::spawn(async move {
             let shutdown_future = async {
@@ -353,7 +362,7 @@ pub async fn inner_main(
     log::info!("Starting on http://{addr}");
 
     let listener = TcpListener::bind(addr).await?;
-    let client = Client::new(&args);
+    let client = Client::new(&args)?;
     let client = Arc::new(Mutex::new(client));
     let mut signal = std::pin::pin!(shutdown_signal);
 
diff --git a/src/store/db.rs b/src/store/db.rs
@@ -113,7 +113,9 @@ impl DBStore {
                     block_opts.set_pin_l0_filter_and_index_blocks_in_cache(true);
 
                     if name == HISTORY_CF {
-                        // bloom filter are useful only for gets with a key miss, which happens a lot for history and never for utxo
+                        // bloom filter are useful only for gets with a key miss,
+                        // which happens a lot for history and never for utxo.
+                        // TODO actually not true for utxo_only query
                         block_opts.set_bloom_filter(10.0, true);
                     }
 
diff --git a/src/test_env.rs b/src/test_env.rs
@@ -70,6 +70,7 @@ async fn inner_launch_with_node(node: BitcoinD, path: Option<PathBuf>, family: F
         node_url: Some(node.rpc_url()),
         derivation_cache_capacity: 10000,
         cache_control_seconds: 0,
+        request_timeout_seconds: 10,
         ..Default::default()
     };
     let available_port = get_available_port().unwrap();
diff --git a/src/threads/blocks.rs b/src/threads/blocks.rs
@@ -47,9 +47,7 @@ async fn get_next_block_to_index(
     match last_indexed.as_ref() {
         Some(last) => {
             match client.get_next(&last, family).await {
-                Ok(ChainStatus::NewBlock(next)) => {
-                    return Some(next);
-                }
+                Ok(ChainStatus::NewBlock(next)) => Some(next),
                 Ok(ChainStatus::Reorg) => {
                     log::warn!("reorg happened! {last:?} removed from the chain");
                     let previous_height = last.height - 1;
@@ -64,7 +62,7 @@ async fn get_next_block_to_index(
                         BlockMeta::new(previous_height, blocks_hash_ts.0, blocks_hash_ts.1);
                     *last_indexed = Some(previous_block_meta);
                     state.store.reorg();
-                    return None;
+                    None
                 }
                 Ok(ChainStatus::Tip) => {
                     // Signal initial sync completion the first time we hit the tip
@@ -73,23 +71,33 @@ async fn get_next_block_to_index(
                         log::info!("Initial block download completed, signaling mempool thread");
                     }
                     sleep(Duration::from_secs(1)).await;
-                    return None;
+                    None
                 }
                 Err(e) => {
                     log::warn!("error getting next block {e}, sleeping for 1 second and retrying");
                     sleep(Duration::from_secs(1)).await;
-                    return None;
+                    None
                 }
             }
         }
         None => {
-            if let Ok(Some(next)) = client.block_hash(0).await {
-                return Some(BlockMeta::new(0, next, 0)); // TODO timestamp
+            match client.block_hash(0).await {
+                Ok(Some(next)) => {
+                    Some(BlockMeta::new(0, next, 0)) // TODO timestamp
+                }
+                Ok(None) => {
+                    log::info!("block hash 0 is not found");
+                    sleep(Duration::from_secs(1)).await;
+                    None
+                }
+                Err(e) => {
+                    log::warn!("error getting next block {e}, sleeping for 1 second and retrying");
+                    sleep(Duration::from_secs(1)).await;
+                    None
+                }
             }
         }
     }
-    sleep(Duration::from_secs(1)).await;
-    None
 }
 
 pub async fn index(

Original file line number	Diff line number	Diff line change
`@@ -113,7 +113,9 @@ impl DBStore {`
`113`	`113`	`block_opts.set_pin_l0_filter_and_index_blocks_in_cache(true);`
`114`	`114`
`115`	`115`	`if name == HISTORY_CF {`
`116`		`- // bloom filter are useful only for gets with a key miss, which happens a lot for history and never for utxo`
	`116`	`+ // bloom filter are useful only for gets with a key miss,`
	`117`	`+ // which happens a lot for history and never for utxo.`
	`118`	`+ // TODO actually not true for utxo_only query`
`117`	`119`	`block_opts.set_bloom_filter(10.0, true);`
`118`	`120`	`}`
`119`	`121`