From b830861815bc65a1520c9e511407df45d1310d5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B8rn=20Olav?= Date: Sat, 5 Sep 2026 07:43:55 +0200 Subject: [PATCH 1/5] archive: walk ancestors on the header height --- lib/archive/iter.rs | 3 +- lib/archive/mod.rs | 95 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 1 deletion(-) diff --git a/lib/archive/iter.rs b/lib/archive/iter.rs index 6cddac56..2343fa04 100644 --- a/lib/archive/iter.rs +++ b/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), }), }; diff --git a/lib/archive/mod.rs b/lib/archive/mod.rs index a1b565f2..35c66fbe 100644 --- a/lib/archive/mod.rs +++ b/lib/archive/mod.rs @@ -1655,3 +1655,98 @@ 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))?; + } + let chain_tip = main_header_info(CHAIN_LEN - 1).block_hash; + let walk = + |rwtxn: &sneed::RwTxn| -> Result, crate::archive::Error> { + let mut connected: Vec = Vec::new(); + while connected.last() != Some(&(CHAIN_LEN - 1)) { + let start_height = + connected.last().map_or(0, |height| height + 1); + let batch: Vec = 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::>(); + 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); + + Ok(()) + } +} From 26c89640c1646c475f921f3e9579370e20c94528 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B8rn=20Olav?= Date: Sat, 5 Sep 2026 07:44:08 +0200 Subject: [PATCH 2/5] archive: read the parent of a mainchain header --- lib/archive/mod.rs | 36 ++++++++++++++++++++++++++++++++++++ lib/node/mainchain_task.rs | 22 ++++------------------ 2 files changed, 40 insertions(+), 18 deletions(-) diff --git a/lib/archive/mod.rs b/lib/archive/mod.rs index 35c66fbe..23e808ad 100644 --- a/lib/archive/mod.rs +++ b/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, 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>( @@ -1747,6 +1761,28 @@ pub(crate) mod test { } 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(()) } } diff --git a/lib/node/mainchain_task.rs b/lib/node/mainchain_task.rs index ce3a104b..a21d259a 100644 --- a/lib/node/mainchain_task.rs +++ b/lib/node/mainchain_task.rs @@ -239,17 +239,8 @@ where if main_state_tip_info.block_hash == common_ancestor { break; } - let main_state_prev_tip_info = if main_state_tip_info - .prev_block_hash - == bitcoin::BlockHash::all_zeros() - { - None - } else { - Some(archive.get_main_header_info( - &rwtxn, - &main_state_tip_info.prev_block_hash, - )?) - }; + let main_state_prev_tip_info = archive + .main_parent_header_info(&rwtxn, &main_state_tip_info)?; let bmm_commitment = archive .get_main_block_info(&rwtxn, &main_state_tip_info.block_hash)? .bmm_commitment; @@ -436,13 +427,8 @@ where let bmm_commitment = archive .get_main_block_info(&rwtxn, &block_hash)? .bmm_commitment; - let parent_info = if header_info.prev_block_hash - != bitcoin::BlockHash::all_zeros() - { - Some(archive.get_main_header_info(&rwtxn, &block_hash)?) - } else { - None - }; + let parent_info = + archive.main_parent_header_info(&rwtxn, &header_info)?; let () = archive .side_tips() .disconnect_mainchain_tip( From 33297e8a6ac2371847d268f84cf915a32b7faecc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B8rn=20Olav?= Date: Sat, 5 Sep 2026 07:44:12 +0200 Subject: [PATCH 3/5] net: keep a peer while the mainchain tip is unset --- lib/archive/mod.rs | 9 +++++++++ lib/net/peer/task.rs | 7 ++++++- 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/lib/archive/mod.rs b/lib/archive/mod.rs index 23e808ad..5900a372 100644 --- a/lib/archive/mod.rs +++ b/lib/archive/mod.rs @@ -1726,6 +1726,15 @@ pub(crate) mod test { 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, crate::archive::Error> { diff --git a/lib/net/peer/task.rs b/lib/net/peer/task.rs index 95736989..27c6ae67 100644 --- a/lib/net/peer/task.rs +++ b/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); } From 604071f61285bd466bbbe1c575cdf56a552511d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B8rn=20Olav?= Date: Sat, 5 Sep 2026 07:44:23 +0200 Subject: [PATCH 4/5] node: connect to the mainchain node again --- lib/node/error.rs | 2 -- lib/node/mainchain_task.rs | 44 +++++++++++++++++++++++++------------- lib/node/net_task.rs | 13 ++++++----- 3 files changed, 37 insertions(+), 22 deletions(-) diff --git a/lib/node/error.rs b/lib/node/error.rs index 1d30f085..5f1cd752 100644 --- a/lib/node/error.rs +++ b/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)")] diff --git a/lib/node/mainchain_task.rs b/lib/node/mainchain_task.rs index a21d259a..f86be8b4 100644 --- a/lib/node/mainchain_task.rs +++ b/lib/node/mainchain_task.rs @@ -480,7 +480,7 @@ where } } - async fn run(mut self) -> Result<(), Error> { + async fn run_once(&mut self) -> Result<(), Error> { let (best_main_tip, block_event_stream) = Self::subscribe_block_events(&mut self.mainchain).await?; if !Self::request_ancestor_infos( @@ -525,12 +525,13 @@ where } let block_event_stream = block_event_stream.map_ok(MailboxItem::BlockEvent); - let request_stream = self.request_rx.map(|(request, response_tx)| { - Ok(MailboxItem::Request { - request, - response_tx, - }) - }); + let request_stream = + (&mut self.request_rx).map(|(request, response_tx)| { + Ok(MailboxItem::Request { + request, + response_tx, + }) + }); let mut mailbox_stream = futures::stream::select(block_event_stream, request_stream); @@ -562,6 +563,26 @@ where } Ok(()) } + + /// Run the task, and start it again after it stops. The mainchain node can + /// stop at any time, and the node must connect to it again. + async fn run(mut self) { + const RECONNECT_DELAY: Duration = Duration::from_secs(5); + + loop { + match self.run_once().await { + Ok(()) => { + tracing::warn!("Mainchain task: the event stream closed") + } + Err(err) => tracing::error!( + "Mainchain task error: {:#}", + ErrorChain::new(&err) + ), + } + tokio::time::sleep(RECONNECT_DELAY).await; + tracing::info!("Mainchain task: connecting to the mainchain node"); + } + } } /// Handle to the task to communicate with mainchain node. @@ -595,14 +616,7 @@ impl MainchainTaskHandle { request_rx, event_tx, }; - let task = spawn(async move { - if let Err(err) = task.run().await { - tracing::error!( - "Mainchain task error: {:#}", - ErrorChain::new(&err) - ); - } - }); + let task = spawn(task.run()); let task_handle = MainchainTaskHandle { task: Arc::new(task), request_tx, diff --git a/lib/node/net_task.rs b/lib/node/net_task.rs index c675a7e7..a9bcf803 100644 --- a/lib/node/net_task.rs +++ b/lib/node/net_task.rs @@ -1069,15 +1069,18 @@ impl NetTask { peer, peer_state_id, ) => { + if self.ctxt.mainchain_task.request(request).is_err() { + tracing::warn!( + ?request, + %peer, + "the mainchain task took no request" + ); + continue; + } mainchain_task_request_sources .entry(request) .or_default() .insert((peer, peer_state_id)); - let () = self - .ctxt - .mainchain_task - .request(request) - .map_err(|_| Error::SendMainchainTaskRequest)?; } MailboxItem::MainchainTaskEvent(event) => { let () = Self::handle_mainchain_task_event( From 3484928c15046f0b0528764c05706950121fb3a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B8rn=20Olav?= Date: Sat, 5 Sep 2026 07:44:38 +0200 Subject: [PATCH 5/5] app: wait for the CUSF mainchain services --- Cargo.lock | 1 + app/Cargo.toml | 3 ++ app/app.rs | 97 +++++++++++++++++++++++++++++++++++++++++++------- 3 files changed, 89 insertions(+), 12 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 20fa9750..f373b649 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6455,6 +6455,7 @@ dependencies = [ "mimalloc", "parking_lot", "poll-promise", + "reserve-port", "rustreexo", "serde", "shlex", diff --git a/app/Cargo.toml b/app/Cargo.toml index 52b5d344..f2464ada 100644 --- a/app/Cargo.toml +++ b/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 diff --git a/app/app.rs b/app/app.rs index 6b756e80..d2af358d 100644 --- a/app/app.rs +++ b/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, - source: Box, - }, #[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 { // 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(()) + } +}