Merge pull request 2140-dev#362 from rustaceanrob/download-policy-5-5

Remove `FilterSyncPolicy`

Author: rustaceanrob

src/builder.rs

@@ -16,7 +16,7 @@ use crate::{
     chain::checkpoints::HeaderCheckpoint,
     db::traits::{HeaderStore, PeerStore},
 };
-use crate::{FilterSyncPolicy, LogLevel, PeerStoreSizeConfig, TrustedPeer};
+use crate::{LogLevel, PeerStoreSizeConfig, TrustedPeer};
 
 #[cfg(feature = "rusqlite")]
 /// The default node returned from the [`NodeBuilder`].
@@ -211,14 +211,6 @@ impl NodeBuilder {
         self
     }
 
-    /// Stop the node from downloading and checking compact block filters until an explicit command by the client is made.
-    /// This is only useful if the scripts to check for may not be known do to some expensive computation, like in a silent
-    /// payments context.
-    pub fn halt_filter_download(mut self) -> Self {
-        self.config.filter_sync_policy = FilterSyncPolicy::Halt;
-        self
-    }
-
     /// Consume the node builder and receive a [`Node`] and [`Client`].
     ///
     /// # Errors

src/client.rs

@@ -237,18 +237,6 @@ impl Requester {
             .map_err(|_| ClientError::SendError)
     }
 
-    /// Explicitly start the block filter syncing process. Note that the node will automatically download and check
-    /// filters unless the policy is to explicitly halt.
-    ///
-    /// # Errors
-    ///
-    /// If the node has stopped running.
-    pub fn continue_download(&self) -> Result<(), ClientError> {
-        self.ntx
-            .send(ClientMessage::ContinueDownload)
-            .map_err(|_| ClientError::SendError)
-    }
-
     /// Check if the node is running.
     pub fn is_running(&self) -> bool {
         self.ntx.send(ClientMessage::NoOp).is_ok()

src/config.rs

@@ -5,7 +5,7 @@ use bitcoin::ScriptBuf;
 use crate::{
     chain::checkpoints::HeaderCheckpoint,
     network::{dns::DnsResolver, ConnectionType},
-    FilterSyncPolicy, LogLevel, PeerStoreSizeConfig, PeerTimeoutConfig, TrustedPeer,
+    LogLevel, PeerStoreSizeConfig, PeerTimeoutConfig, TrustedPeer,
 };
 
 const REQUIRED_PEERS: u8 = 1;
@@ -20,7 +20,6 @@ pub(crate) struct NodeConfig {
     pub connection_type: ConnectionType,
     pub target_peer_size: PeerStoreSizeConfig,
     pub peer_timeout_config: PeerTimeoutConfig,
-    pub filter_sync_policy: FilterSyncPolicy,
     pub log_level: LogLevel,
 }
 
@@ -36,7 +35,6 @@ impl Default for NodeConfig {
             connection_type: Default::default(),
             target_peer_size: PeerStoreSizeConfig::default(),
             peer_timeout_config: PeerTimeoutConfig::default(),
-            filter_sync_policy: Default::default(),
             log_level: Default::default(),
         }
     }

src/lib.rs

@@ -392,16 +392,6 @@ pub enum LogLevel {
     Warning,
 }
 
-/// Should the node immediately download filters or wait for a command
-#[derive(Debug, Default)]
-pub enum FilterSyncPolicy {
-    /// The node will wait for an explicit command to start downloading and checking filters
-    Halt,
-    /// Filters are downloaded immediately after CBF headers are synced.
-    #[default]
-    Continue,
-}
-
 /// The state of the node with respect to connected peers.
 #[derive(Debug, Clone, Copy)]
 pub enum NodeState {

src/messages.rs

@@ -141,10 +141,6 @@ pub(crate) enum ClientMessage {
     AddScript(ScriptBuf),
     /// Starting at the configured anchor checkpoint, look for block inclusions with newly added scripts.
     Rescan,
-    /// If the [`FilterSyncPolicy`] is set to `Halt`, issuing this command will
-    /// start the filter download and checking process. Otherwise, this command will not have any effect
-    /// on node operation.
-    ContinueDownload,
     /// Explicitly request a block from the node.
     #[cfg(feature = "filter-control")]
     GetBlock(BlockRequest),

src/node.rs

@@ -28,7 +28,7 @@ use crate::{
     db::traits::{HeaderStore, PeerStore},
     error::FetchHeaderError,
     network::{peer_map::PeerMap, LastBlockMonitor, PeerId},
-    FilterSyncPolicy, NodeState, RejectPayload, TxBroadcastPolicy,
+    NodeState, RejectPayload, TxBroadcastPolicy,
 };
 
 use super::{
@@ -60,7 +60,6 @@ pub struct Node<H: HeaderStore, P: PeerStore> {
     dialog: Arc<Dialog>,
     client_recv: Arc<Mutex<UnboundedReceiver<ClientMessage>>>,
     peer_recv: Arc<Mutex<Receiver<PeerThreadMessage>>>,
-    filter_sync_policy: Arc<RwLock<FilterSyncPolicy>>,
 }
 
 impl<H: HeaderStore, P: PeerStore> Node<H, P> {
@@ -80,7 +79,6 @@ impl<H: HeaderStore, P: PeerStore> Node<H, P> {
             connection_type,
             target_peer_size,
             peer_timeout_config,
-            filter_sync_policy,
             log_level,
         } = config;
         // Set up a communication channel between the node and client
@@ -137,7 +135,6 @@ impl<H: HeaderStore, P: PeerStore> Node<H, P> {
                 dialog,
                 client_recv: Arc::new(Mutex::new(crx)),
                 peer_recv: Arc::new(Mutex::new(mrx)),
-                filter_sync_policy: Arc::new(RwLock::new(filter_sync_policy)),
             },
             client,
         )
@@ -255,11 +252,6 @@ impl<H: HeaderStore, P: PeerStore> Node<H, P> {
                                     self.broadcast(response).await;
                                 }
                             },
-                            ClientMessage::ContinueDownload => {
-                                if let Some(response) = self.start_filter_download().await {
-                                    self.broadcast(response).await
-                                }
-                            },
                             #[cfg(feature = "filter-control")]
                             ClientMessage::GetBlock(hash) => {
                                 let mut state = self.state.write().await;
@@ -477,10 +469,7 @@ impl<H: HeaderStore, P: PeerStore> Node<H, P> {
                 chain.next_cf_header_message(),
             ));
         } else if !chain.is_filters_synced() {
-            let filter_download = self.filter_sync_policy.read().await;
-            if matches!(*filter_download, FilterSyncPolicy::Continue) {
-                return Some(MainThreadMessage::GetFilters(chain.next_filter_message()));
-            }
+            return Some(MainThreadMessage::GetFilters(chain.next_filter_message()));
         }
         None
     }
@@ -732,22 +721,6 @@ impl<H: HeaderStore, P: PeerStore> Node<H, P> {
         }
     }
 
-    // Continue the filter syncing process by explicit command
-    async fn start_filter_download(&self) -> Option<MainThreadMessage> {
-        let mut download_policy = self.filter_sync_policy.write().await;
-        *download_policy = FilterSyncPolicy::Continue;
-        drop(download_policy);
-        let current_state = self.state.read().await;
-        match *current_state {
-            NodeState::Behind => None,
-            NodeState::HeadersSynced => None,
-            _ => {
-                let mut chain = self.chain.lock().await;
-                self.next_stateful_message(chain.deref_mut()).await
-            }
-        }
-    }
-
     // When the application starts, fetch any headers we know about from the database.
     async fn fetch_headers(&self) -> Result<(), NodeError<H::Error, P::Error>> {
         crate::log!(self.dialog, "Attempting to load headers from the database");

tests/core.rs

@@ -530,60 +530,6 @@ async fn stop_reorg_start_on_orphan() {
     rpc.stop().unwrap();
 }
 
-#[tokio::test]
-#[allow(clippy::collapsible_match)]
-async fn halting_download_works() {
-    let (bitcoind, socket_addr) = start_bitcoind(true).unwrap();
-    let rpc = &bitcoind.client;
-    let tempdir = tempfile::TempDir::new().unwrap().into_path();
-
-    let miner = rpc.new_address().unwrap();
-    mine_blocks(rpc, &miner, 10, 1).await;
-    let best = best_hash(rpc);
-    let mut scripts = HashSet::new();
-    let other = rpc.new_address().unwrap();
-    scripts.insert(other.into());
-
-    let host = (IpAddr::V4(*socket_addr.ip()), Some(socket_addr.port()));
-    let builder = kyoto::builder::NodeBuilder::new(bitcoin::Network::Regtest);
-    let (node, client) = builder
-        .add_peers(vec![host.into()])
-        .add_scripts(scripts)
-        .data_dir(tempdir)
-        .halt_filter_download()
-        .build()
-        .unwrap();
-
-    tokio::task::spawn(async move { node.run().await });
-    let Client {
-        requester,
-        log_rx: _,
-        info_rx: mut info,
-        warn_rx: _,
-        event_rx: mut channel,
-    } = client;
-    while let Some(message) = info.recv().await {
-        if let kyoto::Info::StateChange(node_state) = message {
-            if let kyoto::NodeState::FilterHeadersSynced = node_state {
-                println!("Sleeping for one second...");
-                tokio::time::sleep(Duration::from_secs(1)).await;
-                requester.continue_download().unwrap();
-                break;
-            }
-        }
-    }
-
-    while let Some(message) = channel.recv().await {
-        if let kyoto::messages::Event::Synced(update) = message {
-            println!("Done");
-            assert_eq!(update.tip().hash, best);
-            break;
-        }
-    }
-    requester.shutdown().unwrap();
-    rpc.stop().unwrap();
-}
-
 #[tokio::test]
 async fn signet_syncs() {
     let tempdir = tempfile::TempDir::new().unwrap().into_path();