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();
@@ -1180,16 +1184,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();
 
@@ -1202,8 +1204,8 @@ impl<PersistedState: ChainStore> UpdatableChainstate for ChainState<PersistedSta
             current_acc: acc,
             final_height,
             assume_valid: false,
-            initial_height,
             current_height: initial_height,
+            initial_height,
         };
 
         Ok(PartialChainState(UnsafeCell::new(inner)))

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

@@ -141,7 +141,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(
@@ -155,6 +154,8 @@ pub trait UpdatableChainstate {
     /// This mimics the behavour 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
@@ -205,6 +206,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

@@ -22,6 +22,7 @@
 //!     threads, as long as the origin thread gives away the ownership.
 use bitcoin::BlockHash;
 use floresta_common::prelude::*;
+
 extern crate alloc;
 
 use core::cell::UnsafeCell;
@@ -50,14 +51,10 @@ 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,
+    pub(crate) initial_height: u32,
     /// The height we are syncing up to, trying to push more blocks than this will
     /// result in an error.
     pub(crate) final_height: u32,
@@ -96,19 +93,21 @@ 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.initial_height {
+            return None;
+        }
+
+        let pos = (height - self.initial_height) as usize;
+        if pos >= self.blocks.len() {
+            return None;
+        }
+        self.blocks.get(pos)
     }
 
     #[cfg(feature = "bitcoinconsensus")]
@@ -198,6 +197,7 @@ impl PartialChainStateInner {
                 block.block_hash()
             );
         }
+
         self.update_state(height, acc);
 
         Ok(height)
@@ -215,6 +215,7 @@ impl PartialChainStateInner {
                 BlockValidationErrors::BadMerkleRoot,
             ));
         }
+
         if height >= self.chain_params().params.bip34_height
             && block.bip34_block_height() != Ok(height as u64)
         {
@@ -322,6 +323,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> {
@@ -381,7 +386,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)
@@ -391,8 +395,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)?,
         ))
     }
 
@@ -616,10 +620,13 @@ mod tests {
             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);
+        println!("{}", blocks1.last().unwrap().block_hash());
+        println!("{:?}", blocks2_headers.get(1));
 
         let blocks2_headers = blocks2_headers.iter().map(|block| block.header).collect();
 
@@ -634,6 +641,7 @@ mod tests {
                 .process_block(block, proof, inputs, del_hashes)
                 .unwrap();
         }
+
         // The state after 100 blocks, computed ahead of time.
         let roots = [
             "a2f1e6db842e13c7480c8d80f29ca2db5f9b96e1b428ebfdbd389676d7619081",
@@ -662,7 +670,7 @@ mod tests {
             final_height: 150,
             blocks: blocks2_headers,
             error: None,
-            initial_height: 100,
+            initial_height: 100, // we count the last block in the previous chunk
         }
         .into();
 

crates/floresta-electrum/src/electrum_protocol.rs

@@ -912,6 +912,7 @@ mod test {
     use floresta_watch_only::kv_database::KvDatabase;
     use floresta_watch_only::merkle::MerkleProof;
     use floresta_watch_only::AddressCache;
+    use floresta_wire::address_man::AddressMan;
     use floresta_wire::mempool::Mempool;
     use floresta_wire::node::UtreexoNode;
     use floresta_wire::running_node::RunningNode;
@@ -926,6 +927,7 @@ mod test {
     use tokio::io::AsyncWriteExt;
     use tokio::net::TcpListener;
     use tokio::net::TcpStream;
+    use tokio::sync::RwLock;
     use tokio::task;
     use tokio::time::timeout;
     use tokio_rustls::rustls::Certificate;
@@ -1063,6 +1065,8 @@ mod test {
                 chain.clone(),
                 Arc::new(tokio::sync::RwLock::new(Mempool::new())),
                 None,
+                Arc::new(RwLock::new(true)),
+                AddressMan::default(),
             );
 
         let node_interface = chain_provider.get_handle();

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>,
@@ -397,6 +397,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::NodeHash;
 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 {
@@ -726,7 +723,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

@@ -15,6 +15,7 @@ use std::time::Instant;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
 
+use bitcoin::constants::genesis_block;
 use bitcoin::p2p::address::AddrV2;
 use bitcoin::p2p::address::AddrV2Message;
 use bitcoin::p2p::ServiceFlags;
@@ -169,6 +170,7 @@ pub struct NodeCommon<Chain: BlockchainInterface + UpdatableChainstate> {
     pub(crate) config: UtreexoNodeConfig,
     pub(crate) block_filters: Option<Arc<NetworkFilters<FlatFiltersStore>>>,
     pub(crate) last_filter: BlockHash,
+    pub(crate) kill_signal: Arc<RwLock<bool>>,
 }
 
 pub struct UtreexoNode<Context, Chain: BlockchainInterface + UpdatableChainstate>(
@@ -207,12 +209,17 @@ where
         chain: Chain,
         mempool: Arc<RwLock<Mempool>>,
         block_filters: Option<Arc<NetworkFilters<FlatFiltersStore>>>,
+        kill_signal: Arc<RwLock<bool>>,
+        address_man: AddressMan,
     ) -> Self {
         let (node_tx, node_rx) = unbounded_channel();
         let socks5 = config.proxy.map(Socks5StreamBuilder::new);
         UtreexoNode(
             NodeCommon {
-                last_filter: chain.get_block_hash(0).unwrap(),
+                kill_signal,
+                last_filter: chain
+                    .get_block_hash(0)
+                    .unwrap_or_else(|_| genesis_block(config.network).block_hash()),
                 block_filters,
                 inflight: HashMap::new(),
                 peer_id_count: 0,
@@ -225,7 +232,7 @@ where
                 network: config.network.into(),
                 node_rx,
                 node_tx,
-                address_man: AddressMan::default(),
+                address_man,
                 last_headers_request: Instant::now(),
                 last_tip_update: Instant::now(),
                 last_connection: Instant::now(),
@@ -701,6 +708,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
@@ -712,6 +724,7 @@ where
             "attempting connection with address={:?} kind={:?}",
             address, kind
         );
+
         let (peer_id, address) = address?;
         let now = SystemTime::now()
             .duration_since(UNIX_EPOCH)