Merge pull request #256 from rustaceanrob/fee-rate-1-9

Request broadcast minimum fee rate from client

example/signet.rs

@@ -70,6 +70,9 @@ async fn main() {
                         Event::Synced(update) => {
                             tracing::info!("Synced chain up to block {}",update.tip().height);
                             tracing::info!("Chain tip: {}",update.tip().hash);
+                            // Request information from the node
+                            let fee = sender.broadcast_min_feerate().await.unwrap();
+                            tracing::info!("Minimum transaction broadcast fee rate: {}", fee);
                             break;
                         },
                         Event::Block(indexed_block) => {
@@ -79,7 +82,6 @@ async fn main() {
                         Event::BlocksDisconnected(_) => {
                             tracing::warn!("Some blocks were reorganized")
                         },
-                        Event::FeeFilter(fee_rate) => tracing::info!("Minimum broadcast fee: {fee_rate}"),
                         _ => (),
                     }
                 }

example/testnet4.rs

@@ -72,7 +72,6 @@ async fn main() {
                         Event::BlocksDisconnected(_) => {
                             tracing::warn!("Some blocks were reorganized")
                         },
-                        Event::FeeFilter(fee_rate) => tracing::info!("Minimum broadcast fee: {fee_rate}"),
                         _ => (),
                     }
                 }

src/core/client.rs

@@ -1,9 +1,9 @@
-use bitcoin::block::Header;
 #[cfg(feature = "filter-control")]
 use bitcoin::BlockHash;
 #[cfg(not(feature = "filter-control"))]
 use bitcoin::ScriptBuf;
 use bitcoin::Transaction;
+use bitcoin::{block::Header, FeeRate};
 use std::{collections::BTreeMap, ops::Range, time::Duration};
 use tokio::sync::mpsc;
 pub use tokio::sync::mpsc::Receiver;
@@ -15,7 +15,7 @@ use crate::{Event, Log, TrustedPeer, TxBroadcast};
 #[cfg(feature = "filter-control")]
 use super::{error::FetchBlockError, messages::BlockRequest, BlockReceiver, IndexedBlock};
 use super::{
-    error::{ClientError, FetchHeaderError},
+    error::{ClientError, FetchFeeRateError, FetchHeaderError},
     messages::{BatchHeaderRequest, ClientMessage, HeaderRequest},
 };
 
@@ -133,6 +133,20 @@ impl EventSender {
             .map_err(|_| ClientError::SendError)
     }
 
+    /// A connection has a minimum transaction fee requirement to enter its mempool. For proper transaction propagation,
+    /// transactions should have a fee rate at least as high as the maximum fee filter received.
+    /// This method returns the maximum fee rate requirement of all connected peers.
+    ///
+    /// For more information, refer to BIP133
+    pub async fn broadcast_min_feerate(&self) -> Result<FeeRate, FetchFeeRateError> {
+        let (tx, rx) = tokio::sync::oneshot::channel::<FeeRate>();
+        self.ntx
+            .send(ClientMessage::GetBroadcastMinFeeRate(tx))
+            .await
+            .map_err(|_| FetchFeeRateError::SendError)?;
+        rx.await.map_err(|_| FetchFeeRateError::RecvError)
+    }
+
     /// Add more Bitcoin [`ScriptBuf`] to watch for. Does not rescan the filters.
     /// If the script was already present in the node's collection, no change will occur.
     ///

src/core/error.rs

@@ -202,3 +202,29 @@ impl core::fmt::Display for FetchBlockError {
 }
 
 impl_sourceless_error!(FetchBlockError);
+
+/// Errors that occur when fetching the minimum fee rate to broadcast a transaction.
+#[derive(Debug)]
+pub enum FetchFeeRateError {
+    /// The channel to the node was likely closed and dropped from memory.
+    /// This implies the node is not running.
+    SendError,
+    /// The channel to the client was likely closed by the node and dropped from memory.
+    RecvError,
+}
+
+impl core::fmt::Display for FetchFeeRateError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            FetchFeeRateError::SendError => {
+                write!(f, "the receiver of this message was dropped from memory.")
+            }
+            FetchFeeRateError::RecvError => write!(
+                f,
+                "the channel to the client was likely closed by the node and dropped from memory."
+            ),
+        }
+    }
+}
+
+impl_sourceless_error!(FetchFeeRateError);

src/core/messages.rs

@@ -64,11 +64,6 @@ pub enum Event {
     BlocksDisconnected(Vec<DisconnectedHeader>),
     /// A problem occured sending a transaction. Either the remote node disconnected or the transaction was rejected.
     TxBroadcastFailure(FailurePayload),
-    /// A connection has a minimum transaction fee requirement to enter its mempool. For proper transaction propagation,
-    /// transactions should have a fee rate at least as high as the maximum fee filter received.
-    ///
-    /// For more information, refer to BIP133.
-    FeeFilter(FeeRate),
     /// A compact block filter with associated height and block hash.
     #[cfg(feature = "filter-control")]
     IndexedFilter(IndexedFilter),
@@ -176,6 +171,8 @@ pub(crate) enum ClientMessage {
     GetHeader(HeaderRequest),
     /// Request a range of headers.
     GetHeaderBatch(BatchHeaderRequest),
+    /// Request the broadcast minimum fee rate.
+    GetBroadcastMinFeeRate(FeeRateSender),
 }
 
 type HeaderSender = tokio::sync::oneshot::Sender<Result<Header, FetchHeaderError>>;
@@ -209,6 +206,8 @@ impl BatchHeaderRequest {
 
 pub(crate) type BlockSender = tokio::sync::oneshot::Sender<Result<IndexedBlock, FetchBlockError>>;
 
+pub(crate) type FeeRateSender = tokio::sync::oneshot::Sender<FeeRate>;
+
 #[cfg(feature = "filter-control")]
 #[derive(Debug)]
 pub(crate) struct BlockRequest {

src/core/node.rs

@@ -277,6 +277,10 @@ impl<H: HeaderStore, P: PeerStore> Node<H, P> {
                                         .send_warning(Warning::TransactionRejected).await;
                                     self.dialog.send_event(Event::TxBroadcastFailure(payload)).await;
                                 }
+                                PeerMessage::FeeFilter(feerate) => {
+                                    let mut peer_map = self.peer_map.lock().await;
+                                    peer_map.set_broadcast_min(peer_thread.nonce, feerate);
+                                }
                                 _ => continue,
                             }
                         },
@@ -332,6 +336,14 @@ impl<H: HeaderStore, P: PeerStore> Node<H, P> {
                                 if send_result.is_err() {
                                     self.dialog.send_warning(Warning::ChannelDropped).await
                                 };
+                            },
+                            ClientMessage::GetBroadcastMinFeeRate(request) => {
+                                let peer_map = self.peer_map.lock().await;
+                                let fee_rate = peer_map.broadcast_min();
+                                let send_result = request.send(fee_rate);
+                                if send_result.is_err() {
+                                    self.dialog.send_warning(Warning::ChannelDropped).await
+                                };
                             }
                         }
                     }

src/core/peer_map.rs

@@ -8,7 +8,7 @@ use std::{
 use bitcoin::{
     key::rand,
     p2p::{address::AddrV2, ServiceFlags},
-    Network,
+    FeeRate, Network,
 };
 use rand::{rngs::StdRng, seq::IteratorRandom, SeedableRng};
 use tokio::{
@@ -50,6 +50,7 @@ pub(crate) struct ManagedPeer {
     address: AddrV2,
     port: u16,
     service_flags: Option<ServiceFlags>,
+    broadcast_min: FeeRate,
     ptx: Sender<MainThreadMessage>,
     handle: JoinHandle<Result<(), PeerError>>,
 }
@@ -198,6 +199,7 @@ impl<P: PeerStore> PeerMap<P> {
                 service_flags: None,
                 address: loaded_peer.addr,
                 port: loaded_peer.port,
+                broadcast_min: FeeRate::BROADCAST_MIN,
                 net_time: 0,
                 ptx,
                 handle,
@@ -206,6 +208,13 @@ impl<P: PeerStore> PeerMap<P> {
         Ok(())
     }
 
+    // Set the minimum fee rate this peer will accept
+    pub fn set_broadcast_min(&mut self, nonce: u32, fee_rate: FeeRate) {
+        if let Some(peer) = self.map.get_mut(&nonce) {
+            peer.broadcast_min = fee_rate;
+        }
+    }
+
     // Set the services of a peer
     pub fn set_services(&mut self, nonce: u32, flags: ServiceFlags) {
         if let Some(peer) = self.map.get_mut(&nonce) {
@@ -230,6 +239,15 @@ impl<P: PeerStore> PeerMap<P> {
         self.heights.values().max()
     }
 
+    // The minimum fee rate to successfully broadcast a transaction to all peers
+    pub fn broadcast_min(&self) -> FeeRate {
+        self.map
+            .values()
+            .map(|peer| peer.broadcast_min)
+            .max()
+            .unwrap_or(FeeRate::BROADCAST_MIN)
+    }
+
     // Send a message to the specified peer
     pub async fn send_message(&mut self, nonce: u32, message: MainThreadMessage) {
         if let Some(peer) = self.map.get(&nonce) {

src/network/peer.rs

@@ -21,7 +21,6 @@ use crate::{
         PeerTimeoutConfig,
     },
     network::outbound_messages::V1OutboundMessage,
-    Event,
 };
 
 use super::{
@@ -297,7 +296,13 @@ impl Peer {
             }
             PeerMessage::Pong(_) => Ok(()),
             PeerMessage::FeeFilter(fee) => {
-                self.dialog.send_event(Event::FeeFilter(fee)).await;
+                self.main_thread_sender
+                    .send(PeerThreadMessage {
+                        nonce: self.nonce,
+                        message: PeerMessage::FeeFilter(fee),
+                    })
+                    .await
+                    .map_err(|_| PeerError::ThreadChannel)?;
                 Ok(())
             }
             PeerMessage::Reject(payload) => {

tests/core.rs

@@ -250,7 +250,7 @@ async fn test_mine_after_reorg() {
 }
 
 #[tokio::test]
-async fn test_long_chain() {
+async fn test_various_client_methods() {
     let rpc_result = start_bitcoind(false);
     // If we can't fetch the genesis block then bitcoind is not running. Just exit.
     if rpc_result.is_err() {
@@ -275,6 +275,7 @@ async fn test_long_chain() {
     } = client;
     sync_assert(&best, &mut channel, &mut log).await;
     let batch = sender.get_header_range(10_000..10_002).await.unwrap();
+    let _ = sender.broadcast_min_feerate().await.unwrap();
     assert!(batch.is_empty());
     sender.shutdown().await.unwrap();
     rpc.stop().unwrap();