Merge pull request #4 from rustaceanrob/rescan

node: add rescan

example/rescan.rs

@@ -38,7 +38,7 @@ async fn main() {
  .unwrap(),
  ))
  // The number of connections we would like to maintain
- .num_required_peers(2)
+ .num_required_peers(3)
  // Create the node and client
  .build_node()
  .await;
@@ -75,6 +75,7 @@ async fn main() {
  sender.add_scripts(new_scripts).await.unwrap();
  // // Tell the node to look for these new scripts
  sender.rescan().await.unwrap();
+ tracing::info!("Starting rescan");
  loop {
  if let Ok(message) = receiver.recv().await {
  match message {

src/chain/chain.rs

@@ -626,6 +626,11 @@ impl Chain {
  self.scripts.insert(script);
  }
  }
+
+ //
+ pub(crate) async fn clear_filters(&mut self) {
+ self.filter_chain.clear_cache().await;
+ }
 }
 
 #[cfg(test)]

src/filters/filter_chain.rs

@@ -41,18 +41,4 @@ impl FilterChain {
  pub(crate) fn last_stop_hash_request(&mut self) -> &Option<BlockHash> {
  &self.prev_stophash_request
  }
-
- // pub(crate) fn filter_at_height(&self, height: usize) -> Option<Filter> {
- // let adjusted_height = self.adjusted_height(height);
- // match adjusted_height {
- // Some(height) => {
- // if let Some(filter) = self.chain.get(height) {
- // Some(filter.clone())
- // } else {
- // None
- // }
- // }
- // None => None,
- // }
- // }
 }

src/node/node.rs

@@ -196,7 +196,7 @@ impl Node {
  node_map.send_random(block_request).await;
  }
  // If we have a transaction to broadcast and we are connected to peers, we should broadcast it
- if node_map.live().ge(&1) && !tx_broadcaster.is_empty() {
+ if node_map.live().ge(&self.required_peers) && !tx_broadcaster.is_empty() {
  let transaction = tx_broadcaster.next().unwrap();
  match transaction.broadcast_policy {
  TxBroadcastPolicy::AllPeers => {
@@ -260,7 +260,7 @@ impl Node {
  PeerMessage::Filter(filter) => {
  match self.handle_filter(peer_thread.nonce, filter).await {
  Some(response) => {
- node_map.broadcast(response).await;
+ node_map.send_message(peer_thread.nonce, response).await;
  }
  None => continue,
  }
@@ -302,7 +302,11 @@ impl Node {
  ClientMessage::Shutdown => return Ok(()),
  ClientMessage::Broadcast(transaction) => tx_broadcaster.add(transaction),
  ClientMessage::AddScripts(scripts) => self.add_scripts(scripts).await,
- ClientMessage::Rescan => return Ok(()),
+ ClientMessage::Rescan => {
+ if let Some(response) = self.rescan().await {
+ node_map.broadcast(response).await;
+ }
+ },
  }
  }
  }
@@ -483,13 +487,13 @@ impl Node {
  }
  }
  CFHeaderSyncResult::Dispute(_) => {
- // Request the block from the peer
+ // TODO: Request the filter and block from the peer
  self.dialog
  .send_warning(
  "Found a conflict while peers are sending filter headers".into(),
  )
  .await;
- None
+ Some(MainThreadMessage::Disconnect)
  }
  },
  Err(e) => {
@@ -587,6 +591,29 @@ impl Node {
  }
  }
 
+ // Add more scripts to the chain to look for. Does not imply a rescan.
+ async fn add_scripts(&mut self, scripts: HashSet<ScriptBuf>) {
+ let mut chain = self.chain.lock().await;
+ chain.put_scripts(scripts);
+ }
+
+ // Clear the filter hash cache and redownload the filters.
+ async fn rescan(&mut self) -> Option<MainThreadMessage> {
+ let mut state = self.state.write().await;
+ let mut chain = self.chain.lock().await;
+ match *state {
+ NodeState::Behind => None,
+ NodeState::HeadersSynced => None,
+ _ => {
+ chain.clear_filters().await;
+ *state = NodeState::FilterHeadersSynced;
+ Some(MainThreadMessage::GetFilters(
+ chain.next_filter_message().await,
+ ))
+ }
+ }
+ }
+
  // First we seach the whitelist for peers that we trust. Then, depending on the state
  // we either need to catch up on block headers or we may start requesting filters and blocks.
  // When requesting filters, we try to select peers that have signaled for CF support.
@@ -666,9 +693,4 @@ impl Node {
  }
  }
  }
-
- async fn add_scripts(&mut self, scripts: HashSet<ScriptBuf>) {
- let mut chain = self.chain.lock().await;
- chain.put_scripts(scripts);
- }
 }