feat: backfill now works

crates/floresta-chain/src/pruned_utreexo/chain_state.rs

@@ -1004,6 +1004,10 @@ impl<PersistedState: ChainStore> UpdatableChainstate for ChainState<PersistedSta
         self.reorg(new_tip)
     }
 
+    fn get_acc(&self) -> Stump {
+        self.acc()
+    }
+
     fn mark_block_as_valid(&self, block: BlockHash) -> Result<(), BlockchainError> {
         let header = self.get_disk_block_header(&block)?;
         let height = header.height().unwrap();
@@ -1183,16 +1187,14 @@ impl<PersistedState: ChainStore> UpdatableChainstate for ChainState<PersistedSta
         final_height: u32,
         acc: Stump,
     ) -> Result<super::partial_chain::PartialChainState, BlockchainError> {
-        let blocks = (initial_height..=final_height)
-            .flat_map(|height| {
+        let blocks = (0..=final_height)
+            .map(|height| {
                 let hash = self
                     .get_block_hash(height)
                     .expect("Block should be present");
-                self.get_disk_block_header(&hash)
-            })
-            .filter_map(|header| match header {
-                DiskBlockHeader::FullyValid(header, _) => Some(header),
-                _ => None,
+                *self
+                    .get_disk_block_header(&hash)
+                    .expect("Block should be present")
             })
             .collect();
 
@@ -1205,7 +1207,6 @@ impl<PersistedState: ChainStore> UpdatableChainstate for ChainState<PersistedSta
             current_acc: acc,
             final_height,
             assume_valid: false,
-            initial_height,
             current_height: initial_height,
         };
 

crates/floresta-chain/src/pruned_utreexo/mod.rs

@@ -142,7 +142,6 @@ pub trait UpdatableChainstate {
     /// [PartialChainState] to completion by downloading blocks inside that chainstate's range.
     /// If all goes right, it'll end without error, and you should mark blocks in this range as
     /// valid.
-    ///
     /// Since this chainstate may start from a height with an existing UTXO set, you need to
     /// provide a [Stump] for that block.
     fn get_partial_chain(
@@ -156,6 +155,8 @@ pub trait UpdatableChainstate {
     /// This mimics the behaviour of checking every block before this block, and continues
     /// from this point
     fn mark_chain_as_assumed(&self, acc: Stump, tip: BlockHash) -> Result<bool, BlockchainError>;
+    /// Returns the current accumulator
+    fn get_acc(&self) -> Stump;
 }
 
 /// [ChainStore] is a trait defining how we interact with our chain database. This definitions
@@ -206,6 +207,10 @@ impl<T: UpdatableChainstate> UpdatableChainstate for Arc<T> {
         T::flush(self)
     }
 
+    fn get_acc(&self) -> Stump {
+        T::get_acc(self)
+    }
+
     fn toggle_ibd(&self, is_ibd: bool) {
         T::toggle_ibd(self, is_ibd)
     }

crates/floresta-chain/src/pruned_utreexo/partial_chain.rs

@@ -51,11 +51,6 @@ pub(crate) struct PartialChainStateInner {
     /// and to build the accumulator. We assume this is sorted by height, and
     /// should contains all blocks in this interval.
     pub(crate) blocks: Vec<BlockHeader>,
-    /// The height this interval starts at. This [initial_height, final_height), so
-    /// if we break the interval at height 100, the first interval will be [0, 100)
-    /// and the second interval will be [100, 200). And the initial height of the
-    /// second interval will be 99.
-    pub(crate) initial_height: u32,
     /// The height we are on right now, this is used to keep track of the progress
     /// of the sync.
     pub(crate) current_height: u32,
@@ -97,19 +92,17 @@ unsafe impl Send for PartialChainState {}
 unsafe impl Sync for PartialChainState {}
 
 impl PartialChainStateInner {
-    /// Returns the height we have synced up to so far
-    pub fn current_height(&self) -> u32 {
-        self.current_height
-    }
-
     /// Whether or not we have synced up to the final height
     pub fn is_sync(&self) -> bool {
         self.current_height == self.final_height
     }
 
     pub fn get_block(&self, height: u32) -> Option<&BlockHeader> {
-        let index = height - self.initial_height;
-        self.blocks.get(index as usize)
+        if height >= self.blocks.len() as u32 {
+            return None;
+        }
+
+        self.blocks.get(height as usize)
     }
 
     #[cfg(feature = "bitcoinconsensus")]
@@ -199,6 +192,7 @@ impl PartialChainStateInner {
                 block.block_hash()
             );
         }
+
         self.update_state(height, acc);
 
         Ok(height)
@@ -216,6 +210,7 @@ impl PartialChainStateInner {
                 BlockValidationErrors::BadMerkleRoot,
             ));
         }
+
         if height >= self.chain_params().params.bip34_height
             && block.bip34_block_height() != Ok(height as u64)
         {
@@ -323,6 +318,10 @@ impl UpdatableChainstate for PartialChainState {
         self.inner().current_acc.roots.clone()
     }
 
+    fn get_acc(&self) -> Stump {
+        self.inner().current_acc.clone()
+    }
+
     //these are no-ops, you can call them, but they won't do anything
 
     fn flush(&self) -> Result<(), BlockchainError> {
@@ -382,7 +381,6 @@ impl BlockchainInterface for PartialChainState {
     }
 
     fn get_block_hash(&self, height: u32) -> Result<bitcoin::BlockHash, BlockchainError> {
-        let height = height - self.inner().initial_height;
         self.inner()
             .blocks
             .get(height as usize)
@@ -392,8 +390,8 @@ impl BlockchainInterface for PartialChainState {
 
     fn get_best_block(&self) -> Result<(u32, bitcoin::BlockHash), Self::Error> {
         Ok((
-            self.inner().current_height(),
-            self.get_block_hash(self.inner().current_height())?,
+            self.inner().final_height,
+            self.get_block_hash(self.inner().final_height)?,
         ))
     }
 
@@ -552,7 +550,6 @@ mod tests {
             final_height: 1,
             blocks,
             error: None,
-            initial_height: 0,
         }
         .into()
     }
@@ -578,7 +575,6 @@ mod tests {
             final_height: 100,
             blocks: parsed_blocks.iter().map(|block| block.header).collect(),
             error: None,
-            initial_height: 0,
         }
         .into();
         parsed_blocks.remove(0);
@@ -604,7 +600,8 @@ mod tests {
             parsed_blocks.push(block);
         }
         // The file contains 150 blocks, we split them into two chains.
-        let (blocks1, blocks2) = parsed_blocks.split_at(101);
+        let split = parsed_blocks.clone();
+        let (blocks1, blocks2) = split.split_at(101);
         let mut chainstate1 = PartialChainStateInner {
             assume_valid: true,
             consensus: Consensus {
@@ -613,28 +610,26 @@ mod tests {
             current_height: 0,
             current_acc: Stump::default(),
             final_height: 100,
-            blocks: blocks1.iter().map(|block| block.header).collect(),
+            blocks: parsed_blocks.iter().map(|block| block.header).collect(),
             error: None,
-            initial_height: 0,
         };
+
         // We need to add the last block of the first chain to the second chain, so that
         // the second chain can validate all its blocks.
-        let mut blocks2_headers = vec![blocks1.last().unwrap()];
-        blocks2_headers.extend(blocks2);
-
-        let blocks2_headers = blocks2_headers.iter().map(|block| block.header).collect();
-
-        let mut blocks1 = blocks1.iter();
-        blocks1.next();
+        for (height, block) in blocks1.iter().enumerate() {
+            // skip the genesis block
+            if height == 0 {
+                continue;
+            }
 
-        for block in blocks1 {
             let proof = Proof::default();
             let inputs = HashMap::new();
             let del_hashes = vec![];
             chainstate1
                 .process_block(block, proof, inputs, del_hashes)
                 .unwrap();
         }
+
         // The state after 100 blocks, computed ahead of time.
         let roots = [
             "a2f1e6db842e13c7480c8d80f29ca2db5f9b96e1b428ebfdbd389676d7619081",
@@ -661,9 +656,8 @@ mod tests {
             current_height: 100,
             current_acc: acc2,
             final_height: 150,
-            blocks: blocks2_headers,
+            blocks: parsed_blocks.iter().map(|block| block.header).collect(),
             error: None,
-            initial_height: 100,
         }
         .into();
 

crates/floresta-electrum/src/electrum_protocol.rs

@@ -928,6 +928,7 @@ mod test {
     use tokio::net::TcpListener;
     use tokio::net::TcpStream;
     use tokio::sync::Mutex;
+    use tokio::sync::RwLock;
     use tokio::task;
     use tokio::time::timeout;
     use tokio_rustls::rustls::Certificate;
@@ -1065,6 +1066,7 @@ mod test {
                 chain.clone(),
                 Arc::new(Mutex::new(Mempool::new(Pollard::default(), 0))),
                 None,
+                Arc::new(RwLock::new(false)),
             )
             .unwrap();
 

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

@@ -155,7 +155,7 @@ impl LocalAddress {
 }
 
 /// A module that keeps track of know addresses and serve them to our node to connect
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct AddressMan {
     addresses: HashMap<usize, LocalAddress>,
     good_addresses: Vec<usize>,
@@ -411,6 +411,7 @@ impl AddressMan {
 
         let idx = rand::random::<usize>() % peers.len();
         let utreexo_peer = peers.get(idx)?;
+
         Some((*utreexo_peer, self.addresses.get(utreexo_peer)?.to_owned()))
     }
 

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

@@ -45,7 +45,6 @@
 //! downloading the actual blocks and validating them.
 use std::collections::HashMap;
 use std::collections::HashSet;
-use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
 
@@ -61,7 +60,6 @@ use log::info;
 use log::warn;
 use rustreexo::accumulator::node_hash::BitcoinNodeHash;
 use rustreexo::accumulator::stump::Stump;
-use tokio::sync::RwLock;
 use tokio::time::timeout;
 
 use super::error::WireError;
@@ -170,8 +168,7 @@ where
             .and_modify(|e| *e = last)
             .or_insert(last);
 
-        self.request_headers(headers.last().unwrap().block_hash(), peer)
-            .await
+        self.request_headers(last, peer).await
     }
 
     /// Takes a serialized accumulator and parses it into a Stump
@@ -681,7 +678,7 @@ where
         Ok(())
     }
 
-    pub async fn run(&mut self, stop_signal: Arc<RwLock<bool>>) -> Result<(), WireError> {
+    pub async fn run(&mut self) -> Result<(), WireError> {
         info!("Starting ibd, selecting the best chain");
 
         loop {
@@ -730,7 +727,7 @@ where
 
             try_and_log!(self.check_for_timeout().await);
 
-            if *stop_signal.read().await {
+            if *self.kill_signal.read().await {
                 break;
             }
         }

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

@@ -183,6 +183,7 @@ pub struct NodeCommon<Chain: BlockchainInterface + UpdatableChainstate> {
     pub(crate) config: UtreexoNodeConfig,
     pub(crate) datadir: String,
     pub(crate) network: Network,
+    pub(crate) kill_signal: Arc<tokio::sync::RwLock<bool>>,
 }
 
 pub struct UtreexoNode<Chain: BlockchainInterface + UpdatableChainstate, Context> {
@@ -221,6 +222,7 @@ where
         chain: Chain,
         mempool: Arc<Mutex<Mempool>>,
         block_filters: Option<Arc<NetworkFilters<FlatFiltersStore>>>,
+        kill_signal: Arc<tokio::sync::RwLock<bool>>,
     ) -> Result<Self, WireError> {
         let (node_tx, node_rx) = unbounded_channel();
         let socks5 = config.proxy.map(Socks5StreamBuilder::new);
@@ -259,10 +261,11 @@ where
                 last_get_address_request: Instant::now(),
                 last_send_addresses: Instant::now(),
                 datadir: config.datadir.clone(),
-                socks5,
                 max_banscore: config.max_banscore,
+                socks5,
                 fixed_peer,
                 config,
+                kill_signal,
             },
             context: T::default(),
         })
@@ -964,6 +967,11 @@ where
 
     pub(crate) async fn create_connection(&mut self, kind: ConnectionKind) -> Option<()> {
         let required_services = self.get_required_services();
+        debug!(
+            "openning a new connection with required services: {:?}",
+            required_services
+        );
+
         let address = match &self.fixed_peer {
             Some(address) => Some((0, address.clone())),
             None => self
@@ -975,6 +983,7 @@ where
             "attempting connection with address={:?} kind={:?}",
             address, kind
         );
+
         let (peer_id, address) = address?;
         let now = SystemTime::now()
             .duration_since(UNIX_EPOCH)