Preconnect RPC server to reduce cache miss latency

Cargo.toml

@@ -10,13 +10,17 @@ actix-tls = { version = "*", default-features = false, features = ["accept"] }
 actix-web = { version = "4.5", default-features = false, features = ["macros", "openssl"] }
 actix-web-lab = "0.20"
 async-stream = "0.3"
+bytes = "1.5"
 chrono = "0.4"
 clap = { version = "4.5", features = ["derive", "wrap_help"] }
 cpufeatures = "0.2"
 filesize = "0.2"
 filetime = "0.2"
 futures = "0.3"
 hex = "0.4"
+http-body-util = "0.1"
+hyper = { version = "1.2", features = ["client", "http1"] }
+hyper-util = { version = "0.1", features = ["tokio"] }
 inquire = "0.6"
 log = { version = "0.4", features = ["std"] }
 mime = "0.3"
@@ -28,6 +32,7 @@ rand = { version = "0.8", default-features = false, features = ["small_rng"] }
 regex = "1.10"
 reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "stream", "socks"] }
 scopeguard = "1.2"
+socket2 = "0.5"
 tempfile = "3.10"
 tokio = { version = "1", features = ["full", "parking_lot"] }
 tokio-stream = { version = "0.1", default-features = false, features = ["fs"] }

src/error.rs

@@ -1,12 +1,14 @@
 use core::fmt;
 
+#[allow(clippy::enum_variant_names)]
 #[derive(Debug)]
 pub enum Error {
     VersionTooOld,
     ApiResponseFail { fail_code: String, message: String },
     ConnectTestFail,
     InitSettingsMissing(String),
     HashMismatch { expected: [u8; 20], actual: [u8; 20] },
+    ServerError { status: u16, body: Option<String> },
 }
 
 impl Error {
@@ -28,6 +30,7 @@ impl fmt::Display for Error {
             Error::ConnectTestFail => write!(f, "Connect test failed"),
             Error::InitSettingsMissing(settings) => write!(f, "Missing init settings: {settings}"),
             Error::HashMismatch { expected, actual } => write!(f, "Hash missmatch. Expected={expected:?}, Actual={actual:?}"),
+            Error::ServerError { status, body } => write!(f, "Status={status}, Body={}", body.clone().unwrap_or_default()),
         }
     }
 }

src/main.rs

@@ -61,6 +61,7 @@ mod logger;
 mod middleware;
 mod route;
 mod rpc;
+mod rpc_http_client;
 mod util;
 
 #[cfg(not(target_env = "msvc"))]

src/rpc.rs

@@ -23,11 +23,7 @@ use parking_lot::{RwLock, RwLockUpgradableReadGuard};
 use rand::prelude::SliceRandom;
 use reqwest::{IntoUrl, Url};
 
-use crate::{
-    error::Error,
-    gallery_downloader::GalleryMeta,
-    util::{create_http_client, string_to_hash},
-};
+use crate::{error::Error, gallery_downloader::GalleryMeta, rpc_http_client::RPCHttpClient, util::string_to_hash};
 
 const API_VERSION: i32 = 160; // For server check capabilities.
 const DEFAULT_SERVER: &str = "rpc.hentaiathome.net";
@@ -39,7 +35,7 @@ pub struct RPCClient {
     clock_offset: AtomicI64,
     id: i32,
     key: String,
-    reqwest: reqwest::Client,
+    http_client: RPCHttpClient,
     rpc_servers: RwLock<Vec<String>>,
     running: AtomicBool,
     settings: Arc<Settings>,
@@ -128,7 +124,7 @@ impl RPCClient {
             clock_offset: AtomicI64::new(0),
             id,
             key: key.to_string(),
-            reqwest: create_http_client(Duration::from_secs(600), None),
+            http_client: RPCHttpClient::new(Duration::from_secs(600)),
             rpc_servers: RwLock::new(vec![]),
             running: AtomicBool::new(false),
             settings: Arc::new(Settings {
@@ -206,11 +202,9 @@ impl RPCClient {
     pub async fn get_cert(&self) -> Option<ParsedPkcs12_2> {
         let _provider = Provider::try_load(None, "legacy", true).unwrap();
         let cert = self
-            .reqwest
+            .http_client
             .get(self.build_url("get_cert", "", None))
-            .send()
-            .and_then(|res| async { res.error_for_status() })
-            .and_then(|res| res.bytes())
+            .and_then(|res| self.http_client.to_bytes(res))
             .await
             .ok()
             .and_then(|data| Pkcs12::from_der(&data[..]).ok())
@@ -474,10 +468,9 @@ The program will now terminate.
                     return Ok(response);
                 }
                 Err(err) => {
-                    if err.is_connect() || err.is_timeout() || err.status().map_or(false, |s| s.is_server_error()) {
-                        self.change_server();
-                    }
-                    error = Box::new(err);
+                    error!("Send request error: {}", err);
+                    self.change_server();
+                    error = err;
                 }
             }
             retry -= 1;
@@ -486,19 +479,14 @@ The program will now terminate.
         Err(error)
     }
 
-    async fn send_request<U: IntoUrl>(&self, url: U) -> Result<String, reqwest::Error> {
-        let res = self.reqwest.get(url).timeout(Duration::from_secs(600)).send().await?;
-
-        if let Err(err) = res.error_for_status_ref() {
-            warn!(
-                "Server response error: code={}, body={}",
-                res.status(),
-                res.text().await.unwrap_or_default()
-            );
-            return Err(err);
+    async fn send_request<U: IntoUrl>(&self, url: U) -> Result<String, RequestError> {
+        match self.http_client.get(url).await {
+            Ok(res) => self.http_client.to_text(res).await,
+            Err(err) => {
+                warn!("Server response error: {}", err);
+                Err(err)
+            }
         }
-
-        res.text().await
     }
 
     fn build_url(&self, action: &str, additional: &str, endpoint: Option<&str>) -> Url {