Backport Thunder 0.12.1, bump version for release 0.12.1

Cargo.toml

@@ -7,7 +7,7 @@ authors = ["Ash Manning <[email protected]>"]
 edition = "2024"
 license-file = "LICENSE.txt"
 publish = false
-version = "0.12.0"
+version = "0.12.1"
 
 [workspace.dependencies]
 anyhow = "1.0.72"

app/cli.rs

@@ -9,6 +9,8 @@ use clap::{Arg, Parser};
 use plain_bitassets::types::{Network, THIS_SIDECHAIN};
 use url::{Host, Url};
 
+use crate::util::saturating_pred_level;
+
 const fn ipv4_socket_addr(ipv4_octets: [u8; 4], port: u16) -> SocketAddr {
     let [a, b, c, d] = ipv4_octets;
     let ipv4 = Ipv4Addr::new(a, b, c, d);
@@ -179,12 +181,17 @@ impl Cli {
                 }
             }
         };
+        let log_level = if self.headless {
+            self.log_level
+        } else {
+            saturating_pred_level(self.log_level)
+        };
         Ok(Config {
             datadir: self.datadir.0,
             file_log_level: self.file_log_level,
             headless: self.headless,
             log_dir,
-            log_level: self.log_level,
+            log_level,
             mainchain_grpc_url,
             mnemonic_seed_phrase_path: self.mnemonic_seed_phrase_path,
             net_addr: self.net_addr,

app/main.rs

@@ -17,17 +17,8 @@ mod rpc_server;
 mod util;
 
 use line_buffer::{LineBuffer, LineBufferWriter};
+use util::saturating_pred_level;
 
-/// Saturating predecessor of a log level
-fn saturating_pred_level(log_level: tracing::Level) -> tracing::Level {
-    match log_level {
-        tracing::Level::TRACE => tracing::Level::DEBUG,
-        tracing::Level::DEBUG => tracing::Level::INFO,
-        tracing::Level::INFO => tracing::Level::WARN,
-        tracing::Level::WARN => tracing::Level::ERROR,
-        tracing::Level::ERROR => tracing::Level::ERROR,
-    }
-}
 /// The empty string target `""` can be used to set a default level.
 fn targets_directive_str<'a, Targets>(targets: Targets) -> String
 where

app/util.rs

@@ -64,3 +64,14 @@ where
         Self(Some(Self::promise(stream)))
     }
 }
+
+/// Saturating predecessor of a log level
+pub fn saturating_pred_level(log_level: tracing::Level) -> tracing::Level {
+    match log_level {
+        tracing::Level::TRACE => tracing::Level::DEBUG,
+        tracing::Level::DEBUG => tracing::Level::INFO,
+        tracing::Level::INFO => tracing::Level::WARN,
+        tracing::Level::WARN => tracing::Level::ERROR,
+        tracing::Level::ERROR => tracing::Level::ERROR,
+    }
+}

lib/net/mod.rs

@@ -23,6 +23,7 @@ use crate::{
 
 mod peer;
 
+pub(crate) use peer::error::mailbox::Error as PeerConnectionMailboxError;
 use peer::{
     Connection, ConnectionContext as PeerConnectionCtxt,
     ConnectionHandle as PeerConnectionHandle,
@@ -266,6 +267,20 @@ impl Net {
         }
     }
 
+    /// Apply the provided function to the peer connection handle,
+    /// if it exists.
+    pub fn try_with_active_peer_connection<F, T>(
+        &self,
+        addr: SocketAddr,
+        f: F,
+    ) -> Option<T>
+    where
+        F: FnMut(&PeerConnectionHandle) -> T,
+    {
+        let active_peers_read = self.active_peers.read();
+        active_peers_read.get(&addr).map(f)
+    }
+
     // TODO: This should have more context.
     // Last received message, connection state, etc.
     pub fn get_active_peers(&self) -> Vec<Peer> {

lib/net/peer/channel_pool.rs

@@ -10,9 +10,12 @@ use async_lock::{Semaphore, SemaphoreGuardArc};
 use futures::{FutureExt as _, Stream, StreamExt as _, future::Either, stream};
 use tokio::task::AbortHandle;
 
-use crate::net::peer::{
-    Connection, PeerResponseItem, error, join_set,
-    message::{Heartbeat, Request},
+use crate::{
+    net::peer::{
+        Connection, PeerResponseItem, error,
+        message::{Heartbeat, Request},
+    },
+    util::join_set,
 };
 
 /// Type tags for channel limiters

lib/net/peer/error.rs

@@ -188,7 +188,7 @@ pub(in crate::net::peer) mod request_queue {
     }
 }
 
-pub(in crate::net::peer) mod mailbox {
+pub mod mailbox {
     #[derive(thiserror::Error, Debug)]
     pub enum Error {
         #[error("Heartbeat timeout")]

lib/net/peer/mailbox.rs

@@ -1,5 +1,10 @@
 //! Mailbox for a peer connection task
 
+use std::sync::{
+    Arc,
+    atomic::{self, AtomicBool},
+};
+
 use futures::{
     Stream, StreamExt as _, TryFutureExt as _, channel::mpsc, stream,
 };
@@ -65,9 +70,12 @@ pub struct Receiver {
 }
 
 impl Receiver {
+    /// `received_msg_successfully` is set to `True` if a valid message is
+    /// received successfully.
     pub fn into_stream(
         self,
         connection: Connection,
+        received_msg_successfully: &Arc<AtomicBool>,
     ) -> impl Stream<Item = MailboxItem> + Unpin {
         let (peer_response_tx, peer_response_rx) = mpsc::unbounded();
         let internal_message_stream =
@@ -81,10 +89,14 @@ impl Receiver {
             .map(|err| MailboxItem::Error(err.into()));
         let peer_request_stream = stream::try_unfold((), move |()| {
             let conn = connection.clone();
+            let received_msg_successfully = received_msg_successfully.clone();
             let fut = async move {
                 let item = timeout(
                     Connection::HEARTBEAT_TIMEOUT_INTERVAL,
-                    conn.receive_request(),
+                    conn.receive_request().inspect_ok(|_| {
+                        received_msg_successfully
+                            .store(true, atomic::Ordering::SeqCst);
+                    }),
                 )
                 .map_err(|_| Error::HeartbeatTimeout)
                 .await??;
@@ -96,8 +108,12 @@ impl Receiver {
             Ok(peer_request) => MailboxItem::PeerRequest(peer_request),
             Err(err) => MailboxItem::Error(err),
         });
-        let peer_response_stream =
-            peer_response_rx.map(MailboxItem::PeerResponse);
+        let peer_response_stream = peer_response_rx.map(|resp| {
+            if resp.response.is_ok() {
+                received_msg_successfully.store(true, atomic::Ordering::SeqCst);
+            }
+            MailboxItem::PeerResponse(resp)
+        });
         stream::select_all([
             internal_message_stream.boxed(),
             heartbeat_stream.boxed(),

lib/net/peer/message.rs

@@ -62,9 +62,9 @@ pub struct GetHeadersRequest {
 impl GetHeadersRequest {
     /// Limit bytes to read in a response to a request
     pub const fn read_response_limit(&self) -> NonZeroUsize {
-        // 1KB limit per header
+        // 2KB limit per header
         const READ_HEADER_LIMIT: NonZeroUsize =
-            NonZeroUsize::new(1024).unwrap();
+            NonZeroUsize::new(2048).unwrap();
         let expected_headers = self.height.expect(
             "GetHeaders height should always be Some in an outbound request",
         ) as usize