Connection timeout (#765)

impls/src/adapters/tor.rs

@@ -12,15 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use grin_wallet_config::TorConfig;
-use serde::Serialize;
-use serde_json::{json, Value};
-use std::collections::HashMap;
-use std::convert::TryFrom;
-use std::net::SocketAddr;
-use std::path::PathBuf;
-use std::sync::Arc;
-
 use crate::client_utils::{Client, ClientError};
 use crate::libwallet::slate_versions::{SlateVersion, VersionedSlate};
 use crate::libwallet::{Error, Slate};
@@ -30,6 +21,14 @@ use crate::tor::process::TorProcess;
 use crate::tor::proxy::TorProxy;
 use crate::tor::{config as tor_config, Tor};
 use crate::SlateSender;
+use grin_wallet_config::TorConfig;
+use serde::Serialize;
+use serde_json::{json, Value};
+use std::collections::HashMap;
+use std::convert::TryFrom;
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::sync::Arc;
 
 #[derive(Clone)]
 pub struct TorSlateSender {
@@ -104,7 +103,7 @@ impl TorSlateSender {
 		path.push("torrc");
 		tor.torrc_path(path.to_str().unwrap())
 			.working_dir(tor_dir.to_str().unwrap())
-			.timeout(20)
+			.timeout(config.bootstrap_timeout().as_secs())
 			.completion_percent(100)
 			.launch()
 			.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
@@ -183,14 +182,15 @@ impl TorSlateSender {
 				SocketAddr::V4(self.config.socks_proxy_addr.parse().map_err(|_| {
 					ClientError::Internal("Socks proxy address is not set".to_string())
 				})?);
-			let client = Client::with_proxy(socks_proxy_addr, "socks5h://")
+			let timeout = self.config.request_timeout();
+			let client = Client::with_proxy(socks_proxy_addr, "socks5h://", timeout)
 				.map_err(|_| ClientError::Internal("Unable to create http client".into()))?;
 			let req = client.create_post_request(url, None, &input)?;
 			let res = client.send_request(req)?;
 			res
 		} else {
 			if let Some(client) = &self.tor.client {
-				tor_post(client.clone(), &input, url)
+				tor_post(client.clone(), &self.config, &input, url)
 					.map_err(|e| ClientError::RequestError(format!("{:?}", e)))?
 			} else {
 				return Err(ClientError::Internal("Tor is not configured".to_string()));

impls/src/client_utils/client.rs

@@ -52,22 +52,26 @@ pub struct Client {
 
 impl Client {
 	/// New client
-	pub fn new() -> Result<Self, Error> {
-		Self::build(None)
+	pub fn new(request_timeout: Duration) -> Result<Self, Error> {
+		Self::build(None, request_timeout)
 	}
 
-	pub fn with_proxy(socks_proxy_addr: SocketAddr, scheme: &'static str) -> Result<Self, Error> {
-		Self::build(Some((socks_proxy_addr, scheme)))
+	pub fn with_proxy(
+		socks_proxy_addr: SocketAddr,
+		scheme: &'static str,
+		request_timeout: Duration,
+	) -> Result<Self, Error> {
+		Self::build(Some((socks_proxy_addr, scheme)), request_timeout)
 	}
 
-	fn build(proxy: Option<(SocketAddr, &str)>) -> Result<Self, Error> {
+	fn build(proxy: Option<(SocketAddr, &str)>, request_timeout: Duration) -> Result<Self, Error> {
 		let mut headers = HeaderMap::new();
 		headers.insert(USER_AGENT, HeaderValue::from_static("grin-client"));
 		headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
 		headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
 
 		let mut builder = ClientBuilder::new()
-			.timeout(Duration::from_secs(20))
+			.timeout(request_timeout)
 			.use_rustls_tls()
 			.default_headers(headers);
 

impls/src/node_clients/http.rs

@@ -14,7 +14,9 @@
 
 //! Client functions, implementations of the NodeClient trait
 
+use super::resp_types::*;
 use crate::api::{self, LocatedTxKernel, OutputListing, OutputPrintable};
+use crate::client_utils::json_rpc::*;
 use crate::client_utils::{Client, RUNTIME};
 use crate::core::core::{Transaction, TxKernel};
 use crate::libwallet;
@@ -26,9 +28,7 @@ use futures::TryStreamExt;
 use std::collections::HashMap;
 use std::env;
 use std::net::SocketAddr;
-
-use super::resp_types::*;
-use crate::client_utils::json_rpc::*;
+use std::time::Duration;
 
 const ENDPOINT: &str = "/v2/foreign";
 
@@ -45,8 +45,9 @@ impl HTTPNodeClient {
 	pub fn new(
 		node_url: &str,
 		node_api_secret: Option<String>,
+		request_timeout: Duration,
 	) -> Result<HTTPNodeClient, libwallet::Error> {
-		Self::new_proxy(node_url, node_api_secret, None)
+		Self::new_proxy(node_url, node_api_secret, None, request_timeout)
 	}
 
 	/// Create a new client with proxy
@@ -54,16 +55,17 @@ impl HTTPNodeClient {
 		node_url: &str,
 		node_api_secret: Option<String>,
 		proxy: Option<(SocketAddr, &'static str)>,
+		request_timeout: Duration,
 	) -> Result<HTTPNodeClient, libwallet::Error> {
 		let client = if let Some((a, s)) = proxy {
-			Client::with_proxy(a, s)
+			Client::with_proxy(a, s, request_timeout)
 		} else {
-			Client::new()
+			Client::new(request_timeout)
 		};
 		Ok(HTTPNodeClient {
 			client: client.map_err(|_| libwallet::Error::Node)?,
 			node_url: node_url.to_owned(),
-			node_api_secret: node_api_secret,
+			node_api_secret,
 			node_version_info: None,
 		})
 	}
@@ -107,18 +109,25 @@ impl NodeClient for HTTPNodeClient {
 	fn node_url(&self) -> &str {
 		&self.node_url
 	}
-	fn node_api_secret(&self) -> Option<String> {
-		self.node_api_secret.clone()
-	}
-
 	fn set_node_url(&mut self, node_url: &str) {
 		self.node_url = node_url.to_owned();
 	}
 
+	fn node_api_secret(&self) -> Option<String> {
+		self.node_api_secret.clone()
+	}
+
 	fn set_node_api_secret(&mut self, node_api_secret: Option<String>) {
 		self.node_api_secret = node_api_secret;
 	}
 
+	/// Posts a transaction to a grin node
+	fn post_tx(&self, tx: &Transaction, fluff: bool) -> Result<(), libwallet::Error> {
+		let params = json!([tx, fluff]);
+		self.send_json_request::<serde_json::Value>("push_transaction", &params)?;
+		Ok(())
+	}
+
 	fn get_version_info(&mut self) -> Option<NodeVersionInfo> {
 		if let Some(v) = self.node_version_info.as_ref() {
 			return Some(v.clone());
@@ -135,30 +144,23 @@ impl NodeClient for HTTPNodeClient {
 				// If node isn't available, allow offline functions
 				// unfortunately have to parse string due to error structure
 				let err_string = format!("{}", e);
-				if err_string.contains("404") {
-					return Some(NodeVersionInfo {
+				return if err_string.contains("404") {
+					Some(NodeVersionInfo {
 						node_version: "1.0.0".into(),
 						block_header_version: 1,
 						verified: Some(false),
-					});
+					})
 				} else {
 					error!("Unable to contact Node to get version info: {}, check your node is running", e);
 					warn!("Warning: a) Node is offline, or b) 'node_api_secret_path' in 'grin-wallet.toml' is set incorrectly");
-					return None;
-				}
+					None
+				};
 			}
 		};
 		self.node_version_info = Some(retval.clone());
 		Some(retval)
 	}
 
-	/// Posts a transaction to a grin node
-	fn post_tx(&self, tx: &Transaction, fluff: bool) -> Result<(), libwallet::Error> {
-		let params = json!([tx, fluff]);
-		self.send_json_request::<serde_json::Value>("push_transaction", &params)?;
-		Ok(())
-	}
-
 	/// Return the chain tip from a given node
 	fn get_chain_tip(&self) -> Result<(u64, String), libwallet::Error> {
 		let result = self.send_json_request::<GetTipResp>("get_tip", &serde_json::Value::Null)?;
@@ -222,7 +224,7 @@ impl NodeClient for HTTPNodeClient {
 			.collect();
 
 		// going to leave this here even though we're moving
-		// to the json RPC api to keep the functionality of
+		// to the JSON RPC api to keep the functionality of
 		// parallelizing larger requests. Will raise default
 		// from 200 to 500, however
 		let chunk_default = 500;
@@ -420,8 +422,8 @@ mod tests {
 		}
 	}
 
-	// Wallet will "push" a transaction to node, serializing the transaction as json.
-	// We are testing the json structure is what we expect here.
+	// Wallet will "push" a transaction to node, serializing the transaction as JSON.
+	// We are testing the JSON structure is what we expect here.
 	#[test]
 	fn test_transaction_json_ser_deser() {
 		let tx1 = tx1i1o_v2_compatible();

impls/src/tor/arti.rs

@@ -160,12 +160,10 @@ pub fn start_tor_client(tor_dir: &str, config: TorConfig) -> Result<Tor, Error>
 	})
 }
 
-/// Tor request timeout in milliseconds.
-const REQUEST_TIMEOUT_MS: u64 = 60000;
-
 /// Make POST request with provided client.
 pub fn tor_post<IN>(
 	client: Arc<TorClient<TokioNativeTlsRuntime>>,
+	tor_config: &TorConfig,
 	input: &IN,
 	url: &str,
 ) -> Result<String, Error>
@@ -183,12 +181,13 @@ where
 		Some(h) => h,
 	}
 	.to_string();
+	let timeout = tor_config.request_timeout();
 	let res: Result<String, Error> = thread::spawn(move || {
 		let c = client.clone();
 		client.runtime().block_on(async move {
 			let res = c
 				.runtime()
-				.timeout(Duration::from_millis(REQUEST_TIMEOUT_MS), async {
+				.timeout(timeout, async {
 					let stream = c
 						.connect((host, url.port_u16().unwrap_or(80)))
 						.await
@@ -238,21 +237,18 @@ where
 	res
 }
 
-/// Client Bootstrap timeout in milliseconds.
-const BOOTSTRAP_TIMEOUT_MS: u64 = 60000;
-
 /// Create Tor client.
 fn init_client(
 	state_path: &PathBuf,
 	cache_path: &PathBuf,
-	config: TorConfig,
+	tor_config: TorConfig,
 	reuse_client: bool,
 ) -> Result<(Arc<TorClient<TokioNativeTlsRuntime>>, TorClientConfig), Error> {
 	let mut builder = TorClientConfigBuilder::from_directories(&state_path, cache_path);
 	builder.address_filter().allow_onion_addrs(true);
 
 	// Configure bridge.
-	if let Some(bridge_line) = config.bridge.bridge_line {
+	if let Some(bridge_line) = tor_config.bridge.bridge_line.as_ref() {
 		let bridge: BridgeConfigBuilder = bridge_line
 			.parse()
 			.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
@@ -266,7 +262,11 @@ fn init_client(
 			}
 			Some(t) => {
 				// Now configure bridge transport. (Requires the "pt-client" feature)
-				let bin_path = config.bridge.bridge_bin_path.unwrap_or(t.to_owned());
+				let bin_path = tor_config
+					.bridge
+					.bridge_bin_path
+					.clone()
+					.unwrap_or(t.to_owned());
 				let mut transport = TransportConfigBuilder::default();
 				transport
 					.protocols(vec![t
@@ -294,7 +294,7 @@ fn init_client(
 				*cached_client_config = None;
 			}
 		}
-		let res = launch_client(config.clone());
+		let res = launch_client(config.clone(), &tor_config);
 		return match res {
 			Ok(client) => {
 				cached_client_config.replace((client.clone(), config.clone()));
@@ -304,7 +304,7 @@ fn init_client(
 		};
 	}
 
-	let res = launch_client(config.clone());
+	let res = launch_client(config.clone(), &tor_config);
 	match res {
 		Ok(client) => Ok((client, config)),
 		Err(e) => Err(e),
@@ -312,13 +312,17 @@ fn init_client(
 }
 
 /// Launch tor client from provided configuration.
-fn launch_client(config: TorClientConfig) -> Result<Arc<TorClient<TokioNativeTlsRuntime>>, Error> {
+fn launch_client(
+	client_config: TorClientConfig,
+	tor_config: &TorConfig,
+) -> Result<Arc<TorClient<TokioNativeTlsRuntime>>, Error> {
 	let r = runtime()?;
 	let client = TorClient::with_runtime(r)
-		.config(config)
+		.config(client_config)
 		.create_unbootstrapped()
 		.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
 	let c = client.clone();
+	let timeout = tor_config.bootstrap_timeout();
 	let res = client.runtime().block_on(async move {
 		let bootstrap = async || {
 			return match c.bootstrap().await {
@@ -339,11 +343,7 @@ fn launch_client(config: TorClientConfig) -> Result<Arc<TorClient<TokioNativeTls
 				Err(e) => Err(e),
 			};
 		};
-		match c
-			.runtime()
-			.timeout(Duration::from_millis(BOOTSTRAP_TIMEOUT_MS), bootstrap())
-			.await
-		{
+		match c.runtime().timeout(timeout, bootstrap()).await {
 			Ok(r) => match r {
 				Err(e) => Err(Error::TorProcess(format!("{:?}", e))),
 				Ok(_) => Ok(c),

impls/src/tor/process.rs

@@ -51,13 +51,13 @@ extern crate timer;
 
 use regex::Regex;
 use std::fs::{self, File};
-use std::io;
 use std::io::Write;
 use std::io::{BufRead, BufReader};
 use std::path::{Path, MAIN_SEPARATOR};
 use std::process::{Child, ChildStdout, Command, Stdio};
 use std::sync::mpsc::channel;
 use std::thread;
+use std::{cmp, io};
 use sysinfo::{Process, ProcessExt, System, SystemExt};
 
 #[cfg(windows)]
@@ -83,7 +83,7 @@ pub struct TorProcess {
 	args: Vec<String>,
 	torrc_path: Option<String>,
 	completion_percent: u8,
-	timeout: u32,
+	timeout: u64,
 	working_dir: Option<String>,
 	pub stdout: Option<BufReader<ChildStdout>>,
 	pub process: Option<Child>,
@@ -96,8 +96,8 @@ impl TorProcess {
 			tor_cmd: TOR_EXE_NAME.to_string(),
 			args: vec![],
 			torrc_path: None,
-			completion_percent: 100 as u8,
-			timeout: 0 as u32,
+			completion_percent: 100,
+			timeout: 0,
 			working_dir: None,
 			stdout: None,
 			process: None,
@@ -137,7 +137,7 @@ impl TorProcess {
 		self
 	}
 
-	pub fn timeout(&mut self, timeout: u32) -> &mut Self {
+	pub fn timeout(&mut self, timeout: u64) -> &mut Self {
 		self.timeout = timeout;
 		self
 	}
@@ -198,10 +198,11 @@ impl TorProcess {
 		let stdout_timeout_tx = stdout_tx.clone();
 
 		let timer = timer::Timer::new();
-		let _guard =
-			timer.schedule_with_delay(chrono::Duration::seconds(self.timeout as i64), move || {
-				stdout_timeout_tx.send(Err(Error::Timeout)).unwrap_or(());
-			});
+		// Keep this below chrono's DateTime range used by timer.
+		let timeout = chrono::Duration::seconds(cmp::min(self.timeout, u32::MAX as u64) as i64);
+		let _guard = timer.schedule_with_delay(timeout, move || {
+			stdout_timeout_tx.send(Err(Error::Timeout)).unwrap_or(());
+		});
 		let stdout_thread = thread::spawn(move || {
 			stdout_tx
 				.send(Self::parse_tor_stdout(stdout, completion_percent))
@@ -225,8 +226,8 @@ impl TorProcess {
 		mut stdout: BufReader<ChildStdout>,
 		completion_perc: u8,
 	) -> Result<BufReader<ChildStdout>, Error> {
-		let re_bootstrap = Regex::new(r"^\[notice\] Bootstrapped (?P<perc>[0-9]+)%(.*): ")
-			.map_err(Error::Regex)?;
+		let re_bootstrap =
+			Regex::new(r"^\[notice] Bootstrapped (?P<perc>[0-9]+)%(.*): ").map_err(Error::Regex)?;
 
 		let timestamp_len = "May 16 02:50:08.792".len();
 		let mut warnings = Vec::new();