Use a single long-lived RPC connection when authenticated

The prior system spawned a new connection per request to enable parallelism,
yet kept hitting hyper::IncompleteMessages I couldn't track down. This
attempts to resolve those by a long-lived socket.

Halves the amount of requests per-authenticated RPC call, and accordingly is
likely still better overall.

I don't believe this is resolved yet but this is still worth pushing.

coins/monero/Cargo.toml

@@ -54,6 +54,7 @@ serde_json = { version = "1", default-features = false, features = ["alloc"] }
 base58-monero = { version = "2", default-features = false, features = ["check"] }
 
 # Used for the provided HTTP RPC
+async-recursion = { version = "1", optional = true }
 digest_auth = { version = "0.3", default-features = false, optional = true }
 simple-request = { path = "../../common/request", version = "0.1", default-features = false, optional = true }
 tokio = { version = "1", default-features = false, optional = true }
@@ -100,7 +101,7 @@ std = [
   "base58-monero/std",
 ]
 
-http-rpc = ["digest_auth", "simple-request", "tokio"]
+http-rpc = ["async-recursion", "digest_auth", "simple-request", "tokio"]
 multisig = ["transcript", "frost", "dleq", "std"]
 binaries = ["tokio/rt-multi-thread", "tokio/macros", "http-rpc"]
 experimental = []

coins/monero/src/bin/reserialize_chain.rs

@@ -271,14 +271,15 @@ async fn main() {
   }
   let nodes = if specified_nodes.is_empty() { default_nodes } else { specified_nodes };
 
-  let rpc = |url: String| {
+  let rpc = |url: String| async move {
     HttpRpc::new(url.clone())
+      .await
       .unwrap_or_else(|_| panic!("couldn't create HttpRpc connected to {url}"))
   };
-  let main_rpc = rpc(nodes[0].clone());
+  let main_rpc = rpc(nodes[0].clone()).await;
   let mut rpcs = vec![];
   for i in 0 .. async_parallelism {
-    rpcs.push(Arc::new(rpc(nodes[i % nodes.len()].clone())));
+    rpcs.push(Arc::new(rpc(nodes[i % nodes.len()].clone()).await));
   }
 
   let mut rpc_i = 0;

coins/monero/src/rpc/http.rs

@@ -1,25 +1,30 @@
-use std::io::Read;
+use std::{sync::Arc, io::Read};
 
 use async_trait::async_trait;
 
-use digest_auth::AuthContext;
+use tokio::sync::Mutex;
+
+use digest_auth::{WwwAuthenticateHeader, AuthContext};
 use simple_request::{
-  hyper::{header::HeaderValue, Request},
-  Client,
+  hyper::{StatusCode, header::HeaderValue, Request},
+  Response, Client,
 };
 
 use crate::rpc::{RpcError, RpcConnection, Rpc};
 
 #[derive(Clone, Debug)]
 enum Authentication {
-  // If unauthenticated, reuse a single client
+  // If unauthenticated, use a single client
   Unauthenticated(Client),
-  // If authenticated, don't reuse clients so that each connection makes its own connection
+  // If authenticated, use a single client which supports being locked and tracks its nonce
   // This ensures that if a nonce is requested, another caller doesn't make a request invalidating
   // it
-  // We could acquire a mutex over the client, yet creating a new client is preferred for the
-  // possibility of parallelism
-  Authenticated { username: String, password: String },
+  Authenticated {
+    username: String,
+    password: String,
+    #[allow(clippy::type_complexity)]
+    connection: Arc<Mutex<(Option<(WwwAuthenticateHeader, u64)>, Client)>>,
+  },
 }
 
 /// An HTTP(S) transport for the RPC.
@@ -32,11 +37,29 @@ pub struct HttpRpc {
 }
 
 impl HttpRpc {
+  fn digest_auth_challenge(
+    response: &Response,
+  ) -> Result<Option<(WwwAuthenticateHeader, u64)>, RpcError> {
+    Ok(if let Some(header) = response.headers().get("www-authenticate") {
+      Some((
+        digest_auth::parse(
+          header
+            .to_str()
+            .map_err(|_| RpcError::InvalidNode("www-authenticate header wasn't a string"))?,
+        )
+        .map_err(|_| RpcError::InvalidNode("invalid digest-auth response"))?,
+        0,
+      ))
+    } else {
+      None
+    })
+  }
+
   /// Create a new HTTP(S) RPC connection.
   ///
   /// A daemon requiring authentication can be used via including the username and password in the
   /// URL.
-  pub fn new(mut url: String) -> Result<Rpc<HttpRpc>, RpcError> {
+  pub async fn new(mut url: String) -> Result<Rpc<HttpRpc>, RpcError> {
     let authentication = if url.contains('@') {
       // Parse out the username and password
       let url_clone = url;
@@ -61,9 +84,24 @@ impl HttpRpc {
       if split_userpass.len() > 2 {
         Err(RpcError::ConnectionError("invalid amount of passwords".to_string()))?;
       }
+
+      let client = Client::without_connection_pool(url.clone())
+        .map_err(|_| RpcError::ConnectionError("invalid URL".to_string()))?;
+      // Obtain the initial challenge, which also somewhat validates this connection
+      let challenge = Self::digest_auth_challenge(
+        &client
+          .request(
+            Request::post(url.clone())
+              .body(vec![].into())
+              .map_err(|e| RpcError::ConnectionError(format!("couldn't make request: {e:?}")))?,
+          )
+          .await
+          .map_err(|e| RpcError::ConnectionError(format!("{e:?}")))?,
+      )?;
       Authentication::Authenticated {
         username: split_userpass[0].to_string(),
         password: split_userpass.get(1).unwrap_or(&"").to_string(),
+        connection: Arc::new(Mutex::new((challenge, client))),
       }
     } else {
       Authentication::Unauthenticated(Client::with_connection_pool())
@@ -74,60 +112,96 @@ impl HttpRpc {
 }
 
 impl HttpRpc {
-  async fn inner_post(&self, route: &str, body: Vec<u8>) -> Result<Vec<u8>, RpcError> {
-    let request = |uri| Request::post(uri).body(body.clone().into()).unwrap();
+  #[async_recursion::async_recursion]
+  async fn inner_post(
+    &self,
+    route: &str,
+    body: Vec<u8>,
+    recursing: bool,
+  ) -> Result<Vec<u8>, RpcError> {
+    let request_fn = |uri| {
+      Request::post(uri)
+        .body(body.clone().into())
+        .map_err(|e| RpcError::ConnectionError(format!("couldn't make request: {e:?}")))
+    };
 
-    let mut connection = None;
     let response = match &self.authentication {
       Authentication::Unauthenticated(client) => client
-        .request(request(self.url.clone() + "/" + route))
+        .request(request_fn(self.url.clone() + "/" + route)?)
         .await
         .map_err(|e| RpcError::ConnectionError(format!("{e:?}")))?,
-      Authentication::Authenticated { username, password } => {
-        // This Client will drop and replace its connection on error, when monero-serai requires
-        // a single socket for the lifetime of this function
-        // Since dropping the connection will raise an error, and this function aborts on any
-        // error, this is fine
-        let client = Client::without_connection_pool(self.url.clone())
-          .map_err(|_| RpcError::ConnectionError("invalid URL".to_string()))?;
-        let mut response = client
-          .request(request("/".to_string() + route))
-          .await
-          .map_err(|e| RpcError::ConnectionError(format!("{e:?}")))?;
+      Authentication::Authenticated { username, password, connection } => {
+        let mut connection_lock = connection.lock().await;
+
+        let mut request = request_fn("/".to_string() + route)?;
+
+        // If we don't have an auth challenge, obtain one
+        if connection_lock.0.is_none() {
+          connection_lock.0 = Self::digest_auth_challenge(
+            &connection_lock
+              .1
+              .request(request)
+              .await
+              .map_err(|e| RpcError::ConnectionError(format!("{e:?}")))?,
+          )?;
+          request = request_fn("/".to_string() + route)?;
+        }
+
+        // Insert the challenge response, if we have a challenge
+        if let Some((challenge, cnonce)) = connection_lock.0.as_mut() {
+          // Update the cnonce
+          // Overflow isn't a concern as this is a u64
+          *cnonce += 1;
+
+          let mut context = AuthContext::new_post::<_, _, _, &[u8]>(
+            username,
+            password,
+            "/".to_string() + route,
+            None,
+          );
+          context.set_custom_cnonce(hex::encode(cnonce.to_le_bytes()));
 
-        // Only provide authentication if this daemon actually expects it
-        if let Some(header) = response.headers().get("www-authenticate") {
-          let mut request = request("/".to_string() + route);
           request.headers_mut().insert(
             "Authorization",
             HeaderValue::from_str(
-              &digest_auth::parse(
-                header
-                  .to_str()
-                  .map_err(|_| RpcError::InvalidNode("www-authenticate header wasn't a string"))?,
-              )
-              .map_err(|_| RpcError::InvalidNode("invalid digest-auth response"))?
-              .respond(&AuthContext::new_post::<_, _, _, &[u8]>(
-                username,
-                password,
-                "/".to_string() + route,
-                None,
-              ))
-              .map_err(|_| RpcError::InvalidNode("couldn't respond to digest-auth challenge"))?
-              .to_header_string(),
+              &challenge
+                .respond(&context)
+                .map_err(|_| RpcError::InvalidNode("couldn't respond to digest-auth challenge"))?
+                .to_header_string(),
             )
             .unwrap(),
           );
-
-          // Make the request with the response challenge
-          response = client
-            .request(request)
-            .await
-            .map_err(|e| RpcError::ConnectionError(format!("{e:?}")))?;
         }
 
-        // Store the client so it's not dropped yet
-        connection = Some(client);
+        let response_result = connection_lock
+          .1
+          .request(request)
+          .await
+          .map_err(|e| RpcError::ConnectionError(format!("{e:?}")));
+
+        // If the connection entered an error state, drop the cached challenge as challenges are
+        // per-connection
+        // We don't need to create a new connection as simple-request will for us
+        if response_result.is_err() {
+          connection_lock.0 = None;
+        }
+        let response = response_result?;
+
+        // If we need to re-auth due to this token being stale, recursively re-call this function,
+        // unless we're already recursing
+        if (!recursing) && (response.status() == StatusCode::UNAUTHORIZED) {
+          if let Some(header) = response.headers().get("www-authenticate") {
+            if header
+              .to_str()
+              .map_err(|_| RpcError::InvalidNode("www-authenticate header wasn't a string"))?
+              .contains("stale")
+            {
+              connection_lock.0 = None;
+              drop(connection_lock);
+              return self.inner_post(route, body, true).await;
+            }
+          }
+        }
 
         response
       }
@@ -163,8 +237,6 @@ impl HttpRpc {
       .read_to_end(&mut res)
       .unwrap();
 
-    drop(connection);
-
     Ok(res)
   }
 }
@@ -173,7 +245,7 @@ impl HttpRpc {
 impl RpcConnection for HttpRpc {
   async fn post(&self, route: &str, body: Vec<u8>) -> Result<Vec<u8>, RpcError> {
     // TODO: Make this timeout configurable
-    tokio::time::timeout(core::time::Duration::from_secs(30), self.inner_post(route, body))
+    tokio::time::timeout(core::time::Duration::from_secs(30), self.inner_post(route, body, false))
       .await
       .map_err(|e| RpcError::ConnectionError(format!("{e:?}")))?
   }

coins/monero/tests/runner.rs

@@ -86,7 +86,7 @@ pub fn check_weight_and_fee(tx: &Transaction, fee_rate: Fee) {
 }
 
 pub async fn rpc() -> Rpc<HttpRpc> {
-  let rpc = HttpRpc::new("http://127.0.0.1:18081".to_string()).unwrap();
+  let rpc = HttpRpc::new("http://127.0.0.1:18081".to_string()).await.unwrap();
 
   // Only run once
   if rpc.get_height().await.unwrap() != 1 {

coins/monero/tests/wallet2_compatibility.rs

@@ -35,7 +35,7 @@ async fn make_integrated_address(rpc: &Rpc<HttpRpc>, payment_id: [u8; 8]) -> Str
 }
 
 async fn initialize_rpcs() -> (Rpc<HttpRpc>, Rpc<HttpRpc>, String) {
-  let wallet_rpc = HttpRpc::new("http://127.0.0.1:6061".to_string()).unwrap();
+  let wallet_rpc = HttpRpc::new("http://127.0.0.1:6061".to_string()).await.unwrap();
   let daemon_rpc = runner::rpc().await;
 
   #[derive(Debug, Deserialize)]

common/request/src/lib.rs

@@ -26,7 +26,7 @@ pub enum Error {
   InvalidUri,
   MissingHost,
   InconsistentHost,
-  SslError,
+  SslError(Box<dyn Send + Sync + std::error::Error>),
   Hyper(hyper::Error),
 }
 
@@ -116,7 +116,7 @@ impl Client {
         // If there's not a connection...
         if connection_lock.is_none() {
           let (requester, connection) = hyper::client::conn::http1::handshake(
-            https_builder.clone().call(host.clone()).await.map_err(|_| Error::SslError)?,
+            https_builder.clone().call(host.clone()).await.map_err(Error::SslError)?,
           )
           .await
           .map_err(Error::Hyper)?;

processor/src/main.rs

@@ -650,7 +650,7 @@ async fn main() {
     #[cfg(feature = "bitcoin")]
     NetworkId::Bitcoin => run(db, Bitcoin::new(url).await, coordinator).await,
     #[cfg(feature = "monero")]
-    NetworkId::Monero => run(db, Monero::new(url), coordinator).await,
+    NetworkId::Monero => run(db, Monero::new(url).await, coordinator).await,
     _ => panic!("spawning a processor for an unsupported network"),
   }
 }

processor/src/networks/monero.rs

@@ -191,8 +191,8 @@ fn map_rpc_err(err: RpcError) -> NetworkError {
 }
 
 impl Monero {
-  pub fn new(url: String) -> Monero {
-    Monero { rpc: HttpRpc::new(url).unwrap() }
+  pub async fn new(url: String) -> Monero {
+    Monero { rpc: HttpRpc::new(url).await.unwrap() }
   }
 
   fn view_pair(spend: EdwardsPoint) -> ViewPair {

processor/src/tests/literal/mod.rs

@@ -114,13 +114,15 @@ mod monero {
 
   async fn monero(ops: &DockerOperations) -> Monero {
     let handle = ops.handle("serai-dev-monero").host_port(18081).unwrap();
-    let monero = Monero::new(format!("http://serai:seraidex@{}:{}", handle.0, handle.1));
+    let url = format!("http://serai:seraidex@{}:{}", handle.0, handle.1);
     for _ in 0 .. 60 {
-      if monero.get_latest_block_number().await.is_ok() {
+      if monero_serai::rpc::HttpRpc::new(url.clone()).await.is_ok() {
         break;
       }
       tokio::time::sleep(core::time::Duration::from_secs(1)).await;
     }
+
+    let monero = Monero::new(url).await;
     while monero.get_latest_block_number().await.unwrap() < 150 {
       monero.mine_block().await;
     }

tests/full-stack/src/lib.rs

@@ -171,7 +171,7 @@ impl Handles {
     // If the RPC server has yet to start, sleep for up to 60s until it does
     for _ in 0 .. 60 {
       tokio::time::sleep(Duration::from_secs(1)).await;
-      let Ok(client) = HttpRpc::new(rpc.clone()) else { continue };
+      let Ok(client) = HttpRpc::new(rpc.clone()).await else { continue };
       if client.get_height().await.is_err() {
         continue;
       }