feat(node): Add GetCFHeaders to the node handle

crates/floresta-wire/src/p2p_wire/node/mod.rs

@@ -111,6 +111,12 @@ pub enum NodeRequest {
     /// Proof hashes are the hashes needed to reconstruct the proof, while
     /// leaf data are the actual data of the leaves (i.e., the txouts).
     GetBlockProof((BlockHash, Bitmap, Bitmap)),
+
+    /// Ask for a Compact Block Filters Header
+    GetCFHeaders {
+        start_height: u32,
+        stop_hash: BlockHash,
+    },
 }
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone)]

crates/floresta-wire/src/p2p_wire/node/peer_man.rs

@@ -446,6 +446,31 @@ where
                 self.increase_banscore(peer, 5)?;
                 Ok(None)
             }
+            PeerMessages::CFHeaders(cfheaders) => {
+                let req = self.inflight_user_requests.iter().find_map(|(req, _)| {
+                    if let UserRequest::GetCFilterHeaders { stop_hash, .. } = req {
+                        if *stop_hash == cfheaders.stop_hash {
+                            return Some(req.clone());
+                        }
+                    }
+
+                    None
+                });
+
+                match req {
+                    Some(req) => {
+                        let final_req = self.inflight_user_requests.remove(&req).unwrap();
+                        let _ = final_req.2.send(NodeResponse::CFilterHeaders(cfheaders));
+                    }
+
+                    None => {
+                        warn!("Peer {peer} sent us cfheaders, but we didn't request it");
+                        self.increase_banscore(peer, 5)?;
+                    }
+                }
+
+                Ok(None)
+            }
             _ => Ok(Some(msg)),
         }
     }

crates/floresta-wire/src/p2p_wire/node/user_req.rs

@@ -183,6 +183,24 @@ where
                 let _ = responder.send(NodeResponse::TransactionBroadcastResult(Ok(txid)));
                 return;
             }
+
+            UserRequest::GetCFilterHeaders {
+                start_height,
+                stop_hash,
+            } => {
+                let req = NodeRequest::GetCFHeaders {
+                    start_height,
+                    stop_hash,
+                };
+
+                let peer = self.send_to_fast_peer(req, ServiceFlags::COMPACT_FILTERS);
+                if let Ok(peer) = peer {
+                    self.inflight_user_requests
+                        .insert(user_req, (peer, Instant::now(), responder));
+                }
+
+                return;
+            }
         };
 
         let peer = self.send_to_fast_peer(req, ServiceFlags::NONE);

crates/floresta-wire/src/p2p_wire/node_interface.rs

@@ -12,6 +12,7 @@ use bitcoin::BlockHash;
 use bitcoin::Transaction;
 use bitcoin::Txid;
 use bitcoin::p2p::ServiceFlags;
+use bitcoin::p2p::message_filter::CFHeaders;
 use floresta_mempool::mempool::MempoolError;
 use rustreexo::proof::Proof;
 use serde::Serialize;
@@ -101,6 +102,18 @@ pub enum UserRequest {
 
     /// Return address manager statistics.
     GetAddrManInfo,
+
+    /// Request compact filter headers from a peer, starting from a given height, until a stop hash
+    /// is reached.
+    GetCFilterHeaders {
+        /// The first height we are requesting filters for.
+        start_height: u32,
+
+        /// The last block where we wish filters for.
+        ///
+        /// The remote node will send min(height(stop_hash), 2_000) headers on each request.
+        stop_hash: BlockHash,
+    },
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -165,6 +178,9 @@ pub enum NodeResponse {
 
     /// Address manager statistics.
     GetAddrManInfo(ConnectionStats),
+
+    /// Compact Filters headers
+    CFilterHeaders(CFHeaders),
 }
 
 #[derive(Debug, Clone)]
@@ -342,6 +358,22 @@ impl NodeInterface {
 
         extract_variant!(GetAddrManInfo, val)
     }
+
+    /// Returns a list of Compact Block Filters headers for the requested block range.
+    pub async fn get_cfilters_headers(
+        &self,
+        start_height: u32,
+        stop_hash: BlockHash,
+    ) -> Result<CFHeaders, oneshot::error::RecvError> {
+        let val = self
+            .send_request(UserRequest::GetCFilterHeaders {
+                start_height,
+                stop_hash,
+            })
+            .await?;
+
+        extract_variant!(CFilterHeaders, val)
+    }
 }
 
 fn serialize_service_flags<S>(flags: &ServiceFlags, serializer: S) -> Result<S::Ok, S::Error>

crates/floresta-wire/src/p2p_wire/peer.rs

@@ -23,6 +23,8 @@ use bitcoin::p2p::ServiceFlags;
 use bitcoin::p2p::address::AddrV2Message;
 use bitcoin::p2p::message::NetworkMessage;
 use bitcoin::p2p::message_blockdata::Inventory;
+use bitcoin::p2p::message_filter::CFHeaders;
+use bitcoin::p2p::message_filter::GetCFHeaders;
 use bitcoin::p2p::message_network::VersionMessage;
 use floresta_common::impl_error_from;
 use floresta_mempool::Mempool;
@@ -82,6 +84,12 @@ const INV_MESSAGE_INTERVAL: Duration = Duration::from_secs(30); // 30 seconds
 /// If a peer sends more than this, we disconnect it.
 const MAX_MSGS_PER_SEC: u64 = 10_000;
 
+/// The version for BIP158 basic filter type.
+const BASIC_FILTER_VERSION: u8 = 0;
+
+/// How many filter (or filter headers) are allowed in a single message.
+const MAX_FILTERS_PER_MESSAGE: usize = 2_000;
+
 #[derive(Debug, PartialEq)]
 enum State {
     None,
@@ -404,7 +412,7 @@ impl<T: AsyncWrite + Unpin + Send + Sync> Peer<T> {
             }
             NodeRequest::GetFilter((stop_hash, start_height)) => {
                 let get_filter = bitcoin::p2p::message_filter::GetCFilters {
-                    filter_type: 0,
+                    filter_type: BASIC_FILTER_VERSION,
                     start_height,
                     stop_hash,
                 };
@@ -431,6 +439,20 @@ impl<T: AsyncWrite + Unpin + Send + Sync> Peer<T> {
                 })
                 .await?;
             }
+
+            NodeRequest::GetCFHeaders {
+                start_height,
+                stop_hash,
+            } => {
+                let get_cfheaders = GetCFHeaders {
+                    filter_type: BASIC_FILTER_VERSION,
+                    start_height,
+                    stop_hash,
+                };
+
+                self.write(NetworkMessage::GetCFHeaders(get_cfheaders))
+                    .await?;
+            }
         }
         Ok(())
     }
@@ -571,6 +593,20 @@ impl<T: AsyncWrite + Unpin + Send + Sync> Peer<T> {
                     }
                     _ => {}
                 },
+
+                NetworkMessage::CFHeaders(cfheaders) => {
+                    if cfheaders.filter_hashes.len() > MAX_FILTERS_PER_MESSAGE {
+                        return Err(PeerError::MessageTooBig);
+                    }
+
+                    if cfheaders.filter_type != BASIC_FILTER_VERSION {
+                        warn!("Unknown filter header type {}", cfheaders.filter_type);
+                        return Err(PeerError::UnexpectedMessage);
+                    }
+
+                    self.send_to_node(PeerMessages::CFHeaders(cfheaders), time);
+                }
+
                 // Explicitly ignore these messages, if something changes in the future
                 // this would cause a compile error.
                 NetworkMessage::Verack
@@ -580,7 +616,6 @@ impl<T: AsyncWrite + Unpin + Send + Sync> Peer<T> {
                 | NetworkMessage::Alert(_)
                 | NetworkMessage::BlockTxn(_)
                 | NetworkMessage::CFCheckpt(_)
-                | NetworkMessage::CFHeaders(_)
                 | NetworkMessage::CmpctBlock(_)
                 | NetworkMessage::FilterAdd(_)
                 | NetworkMessage::FilterClear
@@ -858,6 +893,9 @@ pub enum PeerMessages {
 
     /// Remote peer sent us a Utreexo proof,
     UtreexoProof(UtreexoProof),
+
+    /// Remote peer sent us compact block filter headers
+    CFHeaders(CFHeaders),
 }
 
 #[cfg(test)]