Merge pull request #330 from rustaceanrob/cf-headers-4-12

Replace `CFHeaderChain` and `FilterChain`

Author: rustaceanrob

src/chain/chain.rs

@@ -7,7 +7,7 @@ use bitcoin::{
     Network, Work,
 };
 
-use super::IndexedHeader;
+use super::{FilterCommitment, IndexedHeader};
 
 const LOCATOR_INDEX: &[u32] = &[1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024];
 
@@ -110,7 +110,8 @@ pub(crate) struct BlockNode {
     pub height: Height,
     pub header: Header,
     pub acc_work: Work,
-    pub filter_hash: Option<FilterHash>,
+    pub filter_commitment: Option<FilterCommitment>,
+    pub filter_checked: bool,
 }
 
 impl BlockNode {
@@ -119,7 +120,8 @@ impl BlockNode {
             height,
             header,
             acc_work,
-            filter_hash: None,
+            filter_commitment: None,
+            filter_checked: false,
         }
     }
 }
@@ -404,6 +406,10 @@ impl BlockTree {
         self.headers.get(&hash).map(|node| node.height.to_u32())
     }
 
+    pub(crate) fn header_at_hash(&self, hash: BlockHash) -> Option<Header> {
+        self.headers.get(&hash).map(|node| node.header)
+    }
+
     pub(crate) fn height(&self) -> u32 {
         self.active_tip.height.to_u32()
     }
@@ -417,13 +423,82 @@ impl BlockTree {
     }
 
     pub(crate) fn filter_hash(&self, block_hash: BlockHash) -> Option<FilterHash> {
-        self.headers.get(&block_hash)?.filter_hash
+        Some(
+            self.headers
+                .get(&block_hash)?
+                .filter_commitment?
+                .filter_hash,
+        )
+    }
+
+    pub(crate) fn filter_commitment(&self, block_hash: BlockHash) -> Option<&FilterCommitment> {
+        self.headers.get(&block_hash)?.filter_commitment.as_ref()
     }
 
     pub(crate) fn filter_hash_at_height(&self, height: impl Into<Height>) -> Option<FilterHash> {
         let height = height.into();
         let hash = self.canonical_hashes.get(&height)?;
-        self.headers.get(hash)?.filter_hash
+        Some(self.headers.get(hash)?.filter_commitment?.filter_hash)
+    }
+
+    pub(crate) fn set_commitment(&mut self, commitment: FilterCommitment, hash: BlockHash) {
+        if let Some(node) = self.headers.get_mut(&hash) {
+            node.filter_commitment = Some(commitment)
+        }
+    }
+
+    pub(crate) fn check_filter(&mut self, hash: BlockHash) {
+        if let Some(node) = self.headers.get_mut(&hash) {
+            node.filter_checked = true
+        }
+    }
+
+    pub(crate) fn reset_all_filters(&mut self) {
+        let mut curr = self.tip_hash();
+        while let Some(node) = self.headers.get_mut(&curr) {
+            match self.headers.get_mut(&curr) {
+                Some(node) => {
+                    node.filter_checked = false;
+                    curr = node.header.prev_blockhash;
+                }
+                None => break,
+            }
+        }
+        for fork in &self.candidate_forks {
+            curr = fork.hash;
+            while let Some(node) = self.headers.get_mut(&curr) {
+                match self.headers.get_mut(&curr) {
+                    Some(node) => {
+                        if !node.filter_checked {
+                            break;
+                        }
+                        node.filter_checked = false;
+                        curr = node.header.prev_blockhash;
+                    }
+                    None => break,
+                }
+            }
+        }
+    }
+
+    pub(crate) fn filter_headers_synced(&self) -> bool {
+        self.iter_data()
+            .map(|node| node.filter_commitment)
+            .all(|commitment| commitment.is_some())
+    }
+
+    pub(crate) fn filters_synced(&self) -> bool {
+        self.iter_data().all(|node| node.filter_checked)
+    }
+
+    pub(crate) fn total_filters_synced(&self) -> u32 {
+        self.iter_data().filter(|node| node.filter_checked).count() as u32
+    }
+
+    pub(crate) fn total_filter_headers_synced(&self) -> u32 {
+        self.iter_data()
+            .filter(|node| node.filter_commitment.is_some())
+            .count() as u32
     }
 
     pub(crate) fn locators(&self) -> Vec<BlockHash> {
@@ -446,20 +521,27 @@ impl BlockTree {
         self.headers.len()
     }
 
-    pub(crate) fn iter(&self) -> BlockTreeIterator {
-        BlockTreeIterator {
+    pub(crate) fn iter_data(&self) -> BlockNodeIterator {
+        BlockNodeIterator {
+            block_tree: self,
+            current: self.active_tip.hash,
+        }
+    }
+
+    pub(crate) fn iter_headers(&self) -> BlockHeaderIterator {
+        BlockHeaderIterator {
             block_tree: self,
             current: self.active_tip.hash,
         }
     }
 }
 
-pub(crate) struct BlockTreeIterator<'a> {
+pub(crate) struct BlockHeaderIterator<'a> {
     block_tree: &'a BlockTree,
     current: BlockHash,
 }
 
-impl Iterator for BlockTreeIterator<'_> {
+impl Iterator for BlockHeaderIterator<'_> {
     type Item = IndexedHeader;
 
     fn next(&mut self) -> Option<Self::Item> {
@@ -469,6 +551,21 @@ impl Iterator for BlockTreeIterator<'_> {
     }
 }
 
+pub(crate) struct BlockNodeIterator<'a> {
+    block_tree: &'a BlockTree,
+    current: BlockHash,
+}
+
+impl<'a> Iterator for BlockNodeIterator<'a> {
+    type Item = &'a BlockNode;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let node = self.block_tree.headers.get(&self.current)?;
+        self.current = node.header.prev_blockhash;
+        Some(node)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use bitcoin::consensus::deserialize;

src/chain/graph.rs

@@ -14,9 +14,9 @@ pub(crate) mod header_batch;
 
 use std::collections::HashMap;
 
-use bitcoin::block::Header;
+use bitcoin::{block::Header, BlockHash, FilterHash, FilterHeader};
 
-use crate::network::PeerId;
+use crate::{filters::cfheader_batch::CFHeaderBatch, network::PeerId};
 
 type Height = u32;
 
@@ -35,6 +35,81 @@ impl IndexedHeader {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub(crate) struct FilterCommitment {
+    pub header: FilterHeader,
+    pub filter_hash: FilterHash,
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct FilterRequestState {
+    pub last_filter_request: Option<FilterRequest>,
+    pub last_filter_header_request: Option<FilterHeaderRequest>,
+    pub pending_batch: Option<(PeerId, CFHeaderBatch)>,
+    pub agreement_state: FilterHeaderAgreements,
+}
+
+impl FilterRequestState {
+    pub(crate) fn new(required: u8) -> Self {
+        Self {
+            last_filter_request: None,
+            last_filter_header_request: None,
+            pending_batch: None,
+            agreement_state: FilterHeaderAgreements::new(required),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct FilterRequest {
+    #[allow(unused)]
+    pub start_height: u32,
+    pub stop_hash: BlockHash,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct FilterHeaderRequest {
+    pub start_height: u32,
+    pub stop_hash: BlockHash,
+    pub expected_prev_filter_header: Option<FilterHeader>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct FilterHeaderAgreements {
+    current: u8,
+    required: u8,
+}
+
+impl FilterHeaderAgreements {
+    pub(crate) fn new(required: u8) -> Self {
+        Self {
+            current: 0,
+            required,
+        }
+    }
+
+    pub(crate) fn got_agreement(&mut self) {
+        self.current += 1;
+    }
+
+    pub(crate) fn enough_agree(&self) -> bool {
+        self.current.ge(&self.required)
+    }
+
+    pub(crate) fn reset_agreements(&mut self) {
+        self.current = 0;
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum CFHeaderChanges {
+    AddedToQueue,
+    Extended,
+    // Unfortunately, auditing each peer by reconstruction the filter would be costly in network
+    // and compute. Instead it is easier to disconnect from all peers and try again.
+    Conflict,
+}
+
 #[derive(Debug)]
 pub(crate) struct HeightMonitor {
     map: HashMap<PeerId, Height>,

src/chain/mod.rs

@@ -1,6 +1,6 @@
 use tokio::sync::mpsc::{Sender, UnboundedSender};
 
-use super::messages::{Event, Info, Progress, Warning};
+use super::messages::{Event, Info, Warning};
 use crate::LogLevel;
 
 #[derive(Debug, Clone)]
@@ -33,32 +33,6 @@ impl Dialog {
         let _ = self.log_tx.send(dialog.into()).await;
     }
 
-    pub(crate) async fn chain_update(
-        &self,
-        num_headers: u32,
-        num_cf_headers: u32,
-        num_filters: u32,
-        best_height: u32,
-    ) {
-        if matches!(self.log_level, LogLevel::Debug | LogLevel::Info) {
-            let _ = self
-                .info_tx
-                .send(Info::Progress(Progress::new(
-                    num_cf_headers,
-                    num_filters,
-                    best_height,
-                )))
-                .await;
-        }
-        if matches!(self.log_level, LogLevel::Debug) {
-            let message = format!(
-                "Headers ({}/{}) Compact Filter Headers ({}/{}) Filters ({}/{})",
-                num_headers, best_height, num_cf_headers, best_height, num_filters, best_height
-            );
-            let _ = self.log_tx.send(message).await;
-        }
-    }
-
     pub(crate) fn send_warning(&self, warning: Warning) {
         let _ = self.warn_tx.send(warning);
     }

src/dialog.rs

@@ -1,7 +1,10 @@
-use bitcoin::{p2p::message_filter::CFHeaders, BlockHash, FilterHash, FilterHeader};
+use bitcoin::{p2p::message_filter::CFHeaders, BlockHash, FilterHeader};
 
+use crate::chain::FilterCommitment;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct CFHeaderBatch {
-    inner: Vec<(FilterHeader, FilterHash)>,
+    inner: Vec<FilterCommitment>,
     prev_filter_header: FilterHeader,
     stop_hash: BlockHash,
 }
@@ -11,11 +14,14 @@ impl CFHeaderBatch {
     // the CFHeader message may make it easier to detect
     // faulty peers sooner
     pub(crate) fn new(batch: CFHeaders) -> Self {
-        let mut headers: Vec<(FilterHeader, FilterHash)> = vec![];
+        let mut headers: Vec<FilterCommitment> = vec![];
         let mut prev_header = batch.previous_filter_header;
         for hash in batch.filter_hashes {
             let next_header = hash.filter_header(&prev_header);
-            headers.push((next_header, hash));
+            headers.push(FilterCommitment {
+                header: next_header,
+                filter_hash: hash,
+            });
             prev_header = next_header;
         }
         Self {
@@ -29,26 +35,17 @@ impl CFHeaderBatch {
         &self.prev_filter_header
     }
 
-    pub(crate) fn stop_hash(&self) -> &BlockHash {
-        &self.stop_hash
-    }
-
-    pub(crate) fn len(&self) -> usize {
-        self.inner.len()
+    pub(crate) fn stop_hash(&self) -> BlockHash {
+        self.stop_hash
     }
 
-    #[allow(dead_code)]
-    pub(crate) fn inner(&self) -> Vec<(FilterHeader, FilterHash)> {
-        self.inner.clone()
+    pub(crate) fn len(&self) -> u32 {
+        self.inner.len() as u32
     }
 
-    pub(crate) fn take_inner(&mut self) -> Vec<(FilterHeader, FilterHash)> {
+    pub(crate) fn take_inner(&mut self) -> Vec<FilterCommitment> {
         core::mem::take(&mut self.inner)
     }
-
-    pub(crate) fn last_header(&self) -> Option<FilterHeader> {
-        self.inner.last().map(|(header, _)| *header)
-    }
 }
 
 impl From<CFHeaders> for CFHeaderBatch {