cleanup

Author: Apostlex0

lightning-block-sync/src/http.rs

@@ -8,14 +8,11 @@ use bitreq::RequestExt;
 
 use std::convert::TryFrom;
 use std::fmt;
-use std::net::{SocketAddr, ToSocketAddrs};
 use std::time::Duration;
 
-/// Timeout for reading the first byte of a response. This is separate from the general read
-/// timeout as it is not uncommon for Bitcoin Core to be blocked waiting on UTXO cache flushes for
-/// upwards of 10 minutes on slow devices (e.g. RPis with SSDs over USB). Note that we always retry
-/// once when we time out, so the maximum time we allow Bitcoin Core to block for is twice this
-/// value.
+/// Timeout for requests. This is set to a high value as it is not uncommon for Bitcoin Core to be
+/// blocked waiting on UTXO cache flushes for upwards of 10 minutes on slow devices (e.g. RPis with
+/// SSDs over USB).
 const TCP_STREAM_RESPONSE_TIMEOUT: Duration = Duration::from_secs(300);
 
 /// Maximum HTTP message body size in bytes. Enough for a hex-encoded block in JSON format and any
@@ -48,7 +45,7 @@ impl fmt::Display for HttpClientError {
 		match self {
 			HttpClientError::Transport(e) => write!(f, "transport error: {}", e),
 			HttpClientError::Http(e) => write!(f, "HTTP error: {}", e),
-			HttpClientError::Io(e) => write!(f, "I/O error: {}", e),
+			HttpClientError::Io(e) => write!(f, "Response parsing/conversion error: {}", e),
 		}
 	}
 }
@@ -130,49 +127,38 @@ const MAX_CONNECTIONS: usize = 10;
 
 /// Client for making HTTP requests.
 pub(crate) struct HttpClient {
-	address: SocketAddr,
+	host: String,
+	port: u16,
 	#[cfg(feature = "tokio")]
 	client: bitreq::Client,
 }
 
 impl HttpClient {
-	/// Opens a connection to an HTTP endpoint.
-	pub fn connect<E: ToSocketAddrs>(endpoint: E) -> Result<Self, HttpClientError> {
-		let address = match endpoint.to_socket_addrs()?.next() {
-			None => {
-				return Err(std::io::Error::new(
-					std::io::ErrorKind::InvalidInput,
-					"could not resolve to any addresses",
-				)
-				.into());
-			},
-			Some(address) => address,
-		};
-
-		Ok(Self {
-			address,
+	/// Creates a new HTTP client for the given endpoint.
+	///
+	/// DNS resolution is deferred until the first request is made.
+	pub fn new(endpoint: &HttpEndpoint) -> Self {
+		Self {
+			host: endpoint.host().to_string(),
+			port: endpoint.port(),
 			#[cfg(feature = "tokio")]
 			client: bitreq::Client::new(MAX_CONNECTIONS),
-		})
+		}
 	}
 
 	/// Sends a `GET` request for a resource identified by `uri`.
 	///
 	/// Returns the response body in `F` format.
 	#[allow(dead_code)]
-	pub async fn get<F>(&mut self, uri: &str) -> Result<F, HttpClientError>
+	pub async fn get<F>(&self, uri: &str) -> Result<F, HttpClientError>
 	where
 		F: TryFrom<Vec<u8>, Error = std::io::Error>,
 	{
-		let address = self.address;
-		let response_body = self
-			.send_request_with_retry(|| {
-				let url = format!("http://{}{}", address, uri);
-				bitreq::get(url)
-					.with_timeout(TCP_STREAM_RESPONSE_TIMEOUT.as_secs())
-					.with_max_body_size(Some(MAX_HTTP_MESSAGE_BODY_SIZE))
-			})
-			.await?;
+		let url = format!("http://{}:{}{}", self.host, self.port, uri);
+		let request = bitreq::get(url)
+			.with_timeout(TCP_STREAM_RESPONSE_TIMEOUT.as_secs())
+			.with_max_body_size(Some(MAX_HTTP_MESSAGE_BODY_SIZE));
+		let response_body = self.send_request(request).await?;
 		F::try_from(response_body).map_err(HttpClientError::Io)
 	}
 
@@ -183,57 +169,22 @@ impl HttpClient {
 	/// format.
 	#[allow(dead_code)]
 	pub async fn post<F>(
-		&mut self, uri: &str, auth: &str, content: serde_json::Value,
+		&self, uri: &str, auth: &str, content: serde_json::Value,
 	) -> Result<F, HttpClientError>
 	where
 		F: TryFrom<Vec<u8>, Error = std::io::Error>,
 	{
-		let address = self.address;
-		let content = content.to_string();
-		let response_body = self
-			.send_request_with_retry(|| {
-				let url = format!("http://{}{}", address, uri);
-				bitreq::post(url)
-					.with_header("Authorization", auth)
-					.with_header("Content-Type", "application/json")
-					.with_timeout(TCP_STREAM_RESPONSE_TIMEOUT.as_secs())
-					.with_max_body_size(Some(MAX_HTTP_MESSAGE_BODY_SIZE))
-					.with_body(content.clone())
-			})
-			.await?;
+		let url = format!("http://{}:{}{}", self.host, self.port, uri);
+		let request = bitreq::post(url)
+			.with_header("Authorization", auth)
+			.with_header("Content-Type", "application/json")
+			.with_timeout(TCP_STREAM_RESPONSE_TIMEOUT.as_secs())
+			.with_max_body_size(Some(MAX_HTTP_MESSAGE_BODY_SIZE))
+			.with_body(content.to_string());
+		let response_body = self.send_request(request).await?;
 		F::try_from(response_body).map_err(HttpClientError::Io)
 	}
 
-	/// Sends an HTTP request message and reads the response, returning its body. Attempts to
-	/// reconnect and retry only on transport failures (not on HTTP errors like 500/404).
-	async fn send_request_with_retry(
-		&mut self, build_request: impl Fn() -> bitreq::Request,
-	) -> Result<Vec<u8>, HttpClientError> {
-		match self.send_request(build_request()).await {
-			Ok(bytes) => Ok(bytes),
-			Err(HttpClientError::Http(e)) => {
-				// Don't retry on HTTP errors (non-2xx responses)
-				Err(HttpClientError::Http(e))
-			},
-			Err(HttpClientError::Io(e)) => {
-				// Don't retry on I/O errors (e.g., response parsing failures).
-				Err(HttpClientError::Io(e))
-			},
-			Err(HttpClientError::Transport(_)) => {
-				// Reconnect and retry on transport failures. This can happen if the connection
-				// was closed after the keep-alive limits are reached, or generally if the
-				// request timed out due to Bitcoin Core being stuck on a long-running operation
-				// or its RPC queue being full.
-				#[cfg(feature = "tokio")]
-				tokio::time::sleep(Duration::from_millis(100)).await;
-				#[cfg(not(feature = "tokio"))]
-				std::thread::sleep(Duration::from_millis(100));
-				*self = Self::connect(self.address)?;
-				self.send_request(build_request()).await
-			},
-		}
-	}
-
 	/// Sends an HTTP request message and reads the response, returning its body.
 	async fn send_request(&self, request: bitreq::Request) -> Result<Vec<u8>, HttpClientError> {
 		#[cfg(feature = "tokio")]
@@ -446,12 +397,12 @@ pub(crate) mod client_tests {
 		}
 	}
 
-	#[test]
-	fn connect_with_no_socket_address() {
-		match HttpClient::connect(&vec![][..]) {
-			Err(HttpClientError::Io(e)) => {
-				assert_eq!(e.kind(), std::io::ErrorKind::InvalidInput)
-			},
+	#[tokio::test]
+	async fn connect_with_invalid_host() {
+		let endpoint = HttpEndpoint::for_host("invalid.host.example".to_string()).with_port(80);
+		let client = HttpClient::new(&endpoint);
+		match client.get::<JsonResponse>("/foo").await {
+			Err(HttpClientError::Transport(_)) => {},
 			Err(e) => panic!("Unexpected error type: {:?}", e),
 			Ok(_) => panic!("Expected error"),
 		}
@@ -461,7 +412,7 @@ pub(crate) mod client_tests {
 	async fn read_error() {
 		let server = HttpServer::responding_with_server_error("foo");
 
-		let mut client = HttpClient::connect(&server.endpoint()).unwrap();
+		let client = HttpClient::new(&server.endpoint());
 		match client.get::<JsonResponse>("/foo").await {
 			Err(HttpClientError::Http(http_error)) => {
 				assert_eq!(http_error.status_code, 500);
@@ -478,7 +429,7 @@ pub(crate) mod client_tests {
 		let content = MessageBody::Content(body.clone());
 		let server = HttpServer::responding_with_ok::<String>(content);
 
-		let mut client = HttpClient::connect(&server.endpoint()).unwrap();
+		let client = HttpClient::new(&server.endpoint());
 		match client.get::<BinaryResponse>("/foo").await {
 			Err(e) => panic!("Unexpected error: {:?}", e),
 			Ok(bytes) => assert_eq!(bytes.0, body.as_bytes()),
@@ -489,7 +440,7 @@ pub(crate) mod client_tests {
 	async fn reconnect_closed_connection() {
 		let server = HttpServer::responding_with_ok::<String>(MessageBody::Empty);
 
-		let mut client = HttpClient::connect(&server.endpoint()).unwrap();
+		let client = HttpClient::new(&server.endpoint());
 		assert!(client.get::<BinaryResponse>("/foo").await.is_ok());
 		match client.get::<BinaryResponse>("/foo").await {
 			Err(e) => panic!("Unexpected error: {:?}", e),

lightning-block-sync/src/rest.rs

@@ -35,10 +35,10 @@ impl RestClient {
 	{
 		let uri = format!("{}/{}", self.endpoint.path().trim_end_matches("/"), resource_path);
 		let reserved_client = self.client.lock().unwrap().take();
-		let mut client = if let Some(client) = reserved_client {
+		let client = if let Some(client) = reserved_client {
 			client
 		} else {
-			HttpClient::connect(&self.endpoint)?
+			HttpClient::new(&self.endpoint)
 		};
 		let res = client.get::<F>(&uri).await?.try_into().map_err(HttpClientError::Io);
 		*self.client.lock().unwrap() = Some(client);

lightning-block-sync/src/rpc.rs

@@ -118,10 +118,10 @@ impl RpcClient {
 		});
 
 		let reserved_client = self.client.lock().unwrap().take();
-		let mut client = if let Some(client) = reserved_client {
+		let client = if let Some(client) = reserved_client {
 			client
 		} else {
-			HttpClient::connect(&self.endpoint)?
+			HttpClient::new(&self.endpoint)
 		};
 		let http_response = client.post::<JsonResponse>(&uri, &self.basic_auth, content).await;
 		*self.client.lock().unwrap() = Some(client);