node: recover the mainchain task

app/Cargo.toml

@@ -51,6 +51,9 @@ features = ["glow", "wayland", "web_screen_reader", "x11"]
 workspace = true
 features = ["macros", "rt-multi-thread", "signal"]
 
+[dev-dependencies]
+reserve-port = { workspace = true }
+
 [lints]
 workspace = true
 

app/app.rs

@@ -1,4 +1,4 @@
-use std::{borrow::BorrowMut, collections::HashMap, sync::Arc};
+use std::{borrow::BorrowMut, collections::HashMap, sync::Arc, time::Duration};
 
 use fallible_iterator::FallibleIterator as _;
 use futures::{StreamExt, TryFutureExt};
@@ -44,11 +44,6 @@ pub enum Error {
     NoCusfMainchainWalletClient,
     #[error("Failed to request mainchain ancestor info for {block_hash}")]
     RequestMainchainAncestorInfos { block_hash: bitcoin::BlockHash },
-    #[error("Unable to verify existence of CUSF mainchain service(s) at {url}")]
-    VerifyMainchainServices {
-        url: Box<url::Url>,
-        source: Box<tonic::Status>,
-    },
     #[error("wallet error")]
     Wallet(#[from] wallet::Error),
 }
@@ -214,6 +209,27 @@ impl App {
         Ok(res)
     }
 
+    /// Ask the mainchain node for its services until it answers. The node may
+    /// start before the mainchain node.
+    async fn wait_for_proto_support(
+        transport: tonic::transport::channel::Channel,
+        url: &url::Url,
+    ) -> ProtoSupport {
+        const RETRY_DELAY: Duration = Duration::from_secs(5);
+
+        loop {
+            match Self::check_proto_support(transport.clone()).await {
+                Ok(proto_support) => return proto_support,
+                Err(status) => {
+                    tracing::warn!(
+                        %url, %status, "Waiting for CUSF mainchain service(s)"
+                    );
+                    tokio::time::sleep(RETRY_DELAY).await;
+                }
+            }
+        }
+    }
+
     pub fn new(config: &Config) -> Result<Self, Error> {
         // Node launches some tokio tasks for p2p networking, that is why we need a tokio runtime
         // here.
@@ -244,12 +260,11 @@ impl App {
         .concurrency_limit(256)
         .connect_lazy();
         let (cusf_mainchain, cusf_mainchain_miner, cusf_mainchain_wallet) = {
-            let ProtoSupport { miner, wallet } = runtime
-                .block_on(Self::check_proto_support(transport.clone()))
-                .map_err(|err| Error::VerifyMainchainServices {
-                    url: Box::new(config.mainchain_grpc_url.clone()),
-                    source: Box::new(err),
-                })?;
+            let ProtoSupport { miner, wallet } =
+                runtime.block_on(Self::wait_for_proto_support(
+                    transport.clone(),
+                    &config.mainchain_grpc_url,
+                ));
             let mining_client = if miner {
                 Some(mainchain::MiningClient::new(transport.clone()))
             } else {
@@ -660,3 +675,61 @@ impl Drop for App {
         self.task.abort()
     }
 }
+
+#[cfg(test)]
+mod test {
+    use std::{net::SocketAddr, time::Duration};
+
+    use thunder::types::proto::mainchain::generated::validator_service_server;
+    use tokio::time::timeout;
+    use tonic_health::ServingStatus;
+
+    use crate::app::App;
+
+    fn transport(addr: SocketAddr) -> tonic::transport::channel::Channel {
+        tonic::transport::channel::Channel::from_shared(format!(
+            "http://{addr}"
+        ))
+        .unwrap()
+        .connect_lazy()
+    }
+
+    async fn serve_validator_service(addr: SocketAddr) {
+        let (health_reporter, health_service) =
+            tonic_health::server::health_reporter();
+        let () = health_reporter
+            .set_service_status(
+                validator_service_server::SERVICE_NAME,
+                ServingStatus::Serving,
+            )
+            .await;
+        tokio::spawn(
+            tonic::transport::Server::builder()
+                .add_service(health_service)
+                .serve(addr),
+        );
+    }
+
+    /// The node may start before the mainchain node, so it waits for the
+    /// validator service instead of an error.
+    #[tokio::test]
+    async fn wait_for_the_validator_service() -> anyhow::Result<()> {
+        let reserved =
+            reserve_port::ReservedSocketAddr::reserve_random_socket_addr()?;
+        let addr = reserved.socket_addr();
+        let url = format!("http://{addr}").parse()?;
+        let mut proto_support =
+            Box::pin(App::wait_for_proto_support(transport(addr), &url));
+        assert!(
+            timeout(Duration::from_secs(1), &mut proto_support)
+                .await
+                .is_err()
+        );
+        let () = serve_validator_service(addr).await;
+        let proto_support =
+            timeout(Duration::from_secs(30), proto_support).await?;
+        assert!(!proto_support.miner);
+        assert!(!proto_support.wallet);
+        Ok(())
+    }
+}

lib/archive/iter.rs

@@ -202,7 +202,8 @@ pub mod mainchain_ancestors_rev {
                 None => self.inner.insert(Inner {
                     end_height: self
                         .archive
-                        .get_main_height(self.rotxn, self.end_block_hash)?,
+                        .get_main_header_info(self.rotxn, &self.end_block_hash)?
+                        .height,
                     buffer: Vec::with_capacity(MAX_BATCH_SIZE as usize),
                 }),
             };

lib/archive/mod.rs

@@ -1266,6 +1266,20 @@ impl Archive {
         )
     }
 
+    /// Header info of the parent of `header_info`, or `None` for the genesis
+    /// block.
+    pub fn main_parent_header_info(
+        &self,
+        rotxn: &RoTxn,
+        header_info: &BlockHeaderInfo,
+    ) -> Result<Option<BlockHeaderInfo>, Error> {
+        if header_info.prev_block_hash == bitcoin::BlockHash::all_zeros() {
+            return Ok(None);
+        }
+        self.get_main_header_info(rotxn, &header_info.prev_block_hash)
+            .map(Some)
+    }
+
     /// Return a fallible iterator over ancestors of a mainchain block,
     /// starting with the specified block's header
     pub fn main_ancestors<'a>(
@@ -1655,3 +1669,129 @@ impl Archive {
         }
     }
 }
+
+#[cfg(test)]
+pub(crate) mod test {
+    use bitcoin::hashes::Hash as _;
+    use fallible_iterator::FallibleIterator as _;
+
+    use crate::{archive::Archive, types::proto::mainchain::BlockHeaderInfo};
+
+    pub(crate) fn temp_env(
+        test_name: &str,
+    ) -> anyhow::Result<(temp_dir::TempDir, sneed::Env)> {
+        let nanos = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)?
+            .as_nanos();
+        let temp_dir = temp_dir::TempDir::with_prefix(format!(
+            "thunder-{test_name}-{}-{nanos}",
+            std::process::id()
+        ))?;
+        let mut opts = heed::EnvOpenOptions::new();
+        opts.map_size(64 * 1024 * 1024).max_dbs(Archive::NUM_DBS);
+        let env = unsafe { sneed::Env::open(&opts, temp_dir.path()) }?;
+        Ok((temp_dir, env))
+    }
+
+    pub(crate) fn main_header_info(height: u32) -> BlockHeaderInfo {
+        let block_hash = {
+            let mut bytes = [0u8; 32];
+            bytes[0] = 0xff;
+            bytes[1..5].copy_from_slice(&height.to_le_bytes());
+            bitcoin::BlockHash::from_byte_array(bytes)
+        };
+        let prev_block_hash = if height == 0 {
+            bitcoin::BlockHash::all_zeros()
+        } else {
+            main_header_info(height - 1).block_hash
+        };
+        BlockHeaderInfo {
+            block_hash,
+            prev_block_hash,
+            height,
+            work: bitcoin::Work::from_le_bytes([1; 32]),
+        }
+    }
+
+    /// The walk reads each block one time, whatever height the block index
+    /// holds for it.
+    #[test]
+    fn no_batch_repeats_a_block() -> anyhow::Result<()> {
+        const CHAIN_LEN: u32 = 100;
+        const BATCH_SIZE: usize = 32;
+        let (_temp_dir, env) = temp_env("no-batch-repeats-a-block")?;
+        let archive = Archive::new(&env)?;
+        let mut rwtxn = env.write_txn()?;
+        for height in 0..CHAIN_LEN {
+            archive
+                .put_main_header_info(&mut rwtxn, &main_header_info(height))?;
+        }
+        // A node that never synced its mainchain state holds no tip, and the
+        // peer comparison must read that as "no tip", not as a block hash.
+        assert!(
+            archive
+                .side_tips()
+                .get_mainchain_tip(&rwtxn)?
+                .tip_info
+                .is_none()
+        );
+        let chain_tip = main_header_info(CHAIN_LEN - 1).block_hash;
+        let walk =
+            |rwtxn: &sneed::RwTxn| -> Result<Vec<u32>, crate::archive::Error> {
+                let mut connected: Vec<u32> = Vec::new();
+                while connected.last() != Some(&(CHAIN_LEN - 1)) {
+                    let start_height =
+                        connected.last().map_or(0, |height| height + 1);
+                    let batch: Vec<BlockHeaderInfo> = archive
+                        .main_ancestor_header_infos_rev(
+                            rwtxn,
+                            chain_tip,
+                            start_height,
+                        )
+                        .take(BATCH_SIZE)
+                        .collect()?;
+                    assert!(
+                        !batch.is_empty(),
+                        "batch from {start_height} is empty"
+                    );
+                    connected.extend(batch.iter().map(|info| info.height));
+                }
+                Ok(connected)
+            };
+        let heights = (0..CHAIN_LEN).collect::<Vec<_>>();
+        assert_eq!(walk(&rwtxn)?, heights);
+        for height in 0..CHAIN_LEN {
+            let block_hash = main_header_info(height).block_hash;
+            archive.main_block_hash_to_height.put(
+                &mut rwtxn,
+                &block_hash,
+                &(height + 1),
+            )?;
+        }
+        assert_eq!(walk(&rwtxn)?, heights);
+
+        // A disconnect must return the mainchain tip to the parent, so the
+        // block it removed connects again.
+        let genesis = main_header_info(0);
+        let child = main_header_info(1);
+        assert!(archive.main_parent_header_info(&rwtxn, &genesis)?.is_none());
+        let parent = archive.main_parent_header_info(&rwtxn, &child)?;
+        assert_eq!(
+            parent.map(|info| info.block_hash),
+            Some(genesis.block_hash)
+        );
+        archive
+            .side_tips()
+            .connect_mainchain_tip(&mut rwtxn, genesis, None)?;
+        archive
+            .side_tips()
+            .connect_mainchain_tip(&mut rwtxn, child, None)?;
+        archive
+            .side_tips()
+            .disconnect_mainchain_tip(&mut rwtxn, parent, None)?;
+        archive
+            .side_tips()
+            .connect_mainchain_tip(&mut rwtxn, child, None)?;
+        Ok(())
+    }
+}

lib/net/peer/task.rs

@@ -73,10 +73,15 @@ impl ConnectionTask {
                 .side_tips()
                 .get_mainchain_tip(&rotxn)
                 .map_err(archive::Error::from)?;
+            // A node that never synced its mainchain state holds no tip to
+            // compare the peer against.
+            let Some(side_tips_tip_info) = side_tips_tip.tip_info else {
+                return Ok(None);
+            };
             if !ctxt.archive.is_main_descendant(
                 &rotxn,
                 peer_tip_info.tip.main_block_hash,
-                side_tips_tip.block_hash(),
+                side_tips_tip_info.block_hash,
             )? {
                 return Ok(None);
             }

lib/node/error.rs

@@ -130,8 +130,6 @@ pub mod net_task {
         ReceiveMainchainTaskResponse,
         #[error("Receive reorg result cancelled (oneshot)")]
         ReceiveReorgResultOneshot(#[source] oneshot::Canceled),
-        #[error("Send mainchain task request failed")]
-        SendMainchainTaskRequest,
         #[error("Send new tip ready failed")]
         SendNewTipReady(#[source] mpsc::SendError),
         #[error("Send reorg result error (oneshot)")]