Refactor cbf, better rpc error logging and hard code checkpoints

buffrr · buffrr · commit 09913cf770e3 · 2025-05-06T09:40:24.000-07:00
diff --git a/client/src/app.rs b/client/src/app.rs
@@ -100,8 +100,7 @@ impl App {
     }
 
     pub async fn run(&mut self, args: Vec<String>) -> anyhow::Result<()> {
-        let shutdown_receiver = self.shutdown.subscribe();
-        let spaced = Args::configure(args, shutdown_receiver).await?;
+        let spaced = Args::configure(args).await?;
         self.setup_rpc_services(&spaced).await;
         self.setup_sync_service(spaced).await;
 
diff --git a/client/src/cbf.rs b/client/src/cbf.rs
@@ -29,6 +29,7 @@ pub struct CompactFilterSync {
     total_filters: u32,
     wait: Option<Instant>,
     state: SyncState,
+    filters_queued: bool
 }
 
 enum SyncState {
@@ -63,6 +64,7 @@ impl CompactFilterSync {
             total_filters: 0,
             wait: None,
             state: SyncState::SyncChecks,
+            filters_queued: false,
         };
         cbf.load_scripts(wallet);
         cbf
@@ -123,6 +125,11 @@ impl CompactFilterSync {
                     return Ok(());
                 }
                 if info.filters != info.filter_headers {
+                    if !self.filters_queued {
+                        source.queue_filters()?;
+                        self.filters_queued = true;
+                    }
+
                     info!("Filters syncing, retrying...");
                     *progress = WalletProgressUpdate::CbfFilterSync {
                         total: info.filter_headers.unwrap_or(0),
diff --git a/client/src/client.rs b/client/src/client.rs
@@ -33,6 +33,7 @@ pub trait BlockSource {
     fn get_blockchain_info(&self) -> Result<BlockchainInfo, BitcoinRpcError>;
     fn get_block_filter_by_height(&self, height: u32) -> Result<Option<BlockFilterRpc>, BitcoinRpcError>;
     fn queue_blocks(&self, heights: Vec<u32>) -> Result<(), BitcoinRpcError>;
+    fn queue_filters(&self) -> Result<(), BitcoinRpcError>;
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/client/src/config.rs b/client/src/config.rs
@@ -101,7 +101,7 @@ impl ExtendedNetwork {
 impl Args {
     /// Configures spaced node by processing command line arguments
     /// and configuration files
-    pub async fn configure(args: Vec<String>, shutdown: tokio::sync::broadcast::Receiver<()>) -> anyhow::Result<Spaced> {
+    pub async fn configure(args: Vec<String>) -> anyhow::Result<Spaced> {
         let mut args =  Args::try_parse_from(args)?;
         let default_dirs = get_default_node_dirs();
 
@@ -147,7 +147,7 @@ impl Args {
             !args.bitcoin_rpc_light
         );
 
-        let genesis = Spaced::genesis(&rpc, args.chain, shutdown).await?;
+        let genesis = Spaced::genesis(args.chain);
 
         fs::create_dir_all(data_dir.clone())?;
 
diff --git a/client/src/rpc.rs b/client/src/rpc.rs
@@ -642,7 +642,7 @@ impl WalletManager {
             .rpc
             .send_json(&client, &self.rpc.get_block_count())
             .await?;
-        let height = std::cmp::max(count - 20, 0) as u32;
+        let height = std::cmp::max(count - 1, 0) as u32;
 
         let hash = self
             .rpc
diff --git a/client/src/source.rs b/client/src/source.rs
@@ -8,7 +8,7 @@ use std::{
     },
     time::Duration,
 };
-
+use std::error::Error;
 use base64::Engine;
 use bitcoin::{Block, BlockHash, Txid};
 use hex::FromHexError;
@@ -190,6 +190,11 @@ impl BitcoinRpc {
         self.make_request("queueblocks", params)
     }
 
+    pub fn queue_filters(&self) -> BitcoinRpcRequest {
+        let params = serde_json::json!([]);
+        self.make_request("queuefilters", params)
+    }
+
     pub fn get_mempool_entry(&self, txid: &Txid) -> BitcoinRpcRequest {
         let params = serde_json::json!([txid]);
 
@@ -276,7 +281,7 @@ impl BitcoinRpc {
             {
                 Ok(res) => return Self::clean_rpc_response(res).await,
                 Err(e) if e.is_temporary() && attempt < max_retries - 1 => {
-                    error!("Rpc: {} - retrying in {:?}...", e, delay);
+                    log_rpc_error(&request.body, &e, delay);
                     last_error = Some(e.into());
                     tokio::time::sleep(delay).await;
                     delay *= 2;
@@ -302,15 +307,14 @@ impl BitcoinRpc {
             if let Some(auth) = self.auth_token.as_ref() {
                 builder = builder.header("Authorization", format!("Basic {}", auth));
             }
-
             match builder
                 .json(&request.body)
                 .send()
                 .map_err(BitcoinRpcError::from)
             {
                 Ok(res) => return Self::clean_rpc_response_blocking(res),
                 Err(e) if e.is_temporary() && attempt < max_retries - 1 => {
-                    error!("Rpc: {} - retrying in {:?}...", e, delay);
+                    log_rpc_error(&request.body, &e, delay);
                     last_error = Some(e.into());
                     std::thread::sleep(delay);
                     delay *= 2;
@@ -811,7 +815,28 @@ impl std::error::Error for BitcoinRpcError {
 
 impl ErrorForRpc for reqwest::Response {
     async fn error_for_rpc<T: DeserializeOwned>(self) -> Result<T, BitcoinRpcError> {
-        let rpc_res: JsonRpcResponse<T> = self.json().await?;
+        let text = self
+            .text()
+            .await
+            .map_err(|e| BitcoinRpcError::Other(format!("Could not read response body: {}", e)))?;
+
+        // Try to deserialize as JsonRpcResponse<T>
+        let rpc_res: JsonRpcResponse<T> = match serde_json::from_str(&text) {
+            Ok(rpc_res) => rpc_res,
+            Err(e) => {
+                // Try to decode without result
+                let error_res: Option<JsonRpcResponse<Option<String>>> = serde_json::from_str(&text).ok();
+                if let Some(error_res) = error_res {
+                    if let Some(error) = error_res.error {
+                        return Err(BitcoinRpcError::Rpc(error));
+                    }
+                }
+                return Err(BitcoinRpcError::Other(
+                    format!("Expected a JSON response, got '{}': {}", text, e),
+                ));
+            }
+        };
+
         if let Some(e) = rpc_res.error {
             return Err(BitcoinRpcError::Rpc(e));
         }
@@ -822,7 +847,28 @@ impl ErrorForRpc for reqwest::Response {
 
 impl ErrorForRpcBlocking for reqwest::blocking::Response {
     fn error_for_rpc<T: DeserializeOwned>(self) -> Result<T, BitcoinRpcError> {
-        let rpc_res: JsonRpcResponse<T> = self.json()?;
+        let text = self.text().map_err(|e| BitcoinRpcError::Other(
+            format!("Could not read response body: {}", e),
+        ))?;
+
+        // Attempt to deserialize the text as JSON
+        let rpc_res: JsonRpcResponse<T> = match serde_json::from_str(&text)
+             {
+            Ok(rpc_res) => rpc_res,
+            Err(e) => {
+                // try to decode without result
+                let error_res : Option<JsonRpcResponse<Option<String>>> = serde_json::from_str(&text).ok();
+                if let Some(error_res) = error_res {
+                    if let Some(error) = error_res.error {
+                        return Err(BitcoinRpcError::Rpc(error));
+                    }
+                }
+                return Err(BitcoinRpcError::Other(
+                    format!("Expected a JSON response, got '{}': {}", text, e),
+                ))
+            }
+        };
+
         if let Some(e) = rpc_res.error {
             return Err(BitcoinRpcError::Rpc(e));
         }
@@ -905,7 +951,6 @@ impl BlockSource for BitcoinBlockSource {
             .send_json_blocking(&self.client, &self.rpc.get_block_count())?)
     }
 
-
     fn get_best_chain(&self, tip: Option<u32>, expected_chain: Network) -> Result<Option<ChainAnchor>, BitcoinRpcError> {
         #[derive(Deserialize)]
         struct Info {
@@ -981,4 +1026,25 @@ impl BlockSource for BitcoinBlockSource {
             .send_json_blocking::<()>(&self.client, &self.rpc.queue_blocks(heights))?;
         Ok(())
     }
+
+    fn queue_filters(&self) -> anyhow::Result<(), BitcoinRpcError> {
+        self
+            .rpc
+            .send_json_blocking::<()>(&self.client, &self.rpc.queue_filters())?;
+        Ok(())
+    }
+}
+
+fn log_rpc_error(request: &Value, e: &BitcoinRpcError, delay: Duration) {
+    let rpc_method = serde_json::to_string(&request.get("method"))
+        .unwrap_or("".to_string());
+    let rpc_params = serde_json::to_string(&request.get("params"))
+        .unwrap_or("".to_string());
+    let src = match e {
+        BitcoinRpcError::Transport(e) =>
+            e.source().map(|s| format!("({:?})", s)),
+        _ => None
+    }.unwrap_or("".to_string());
+
+    error!("Rpc {}{}: {}{} - retrying in {:?}...", rpc_method, rpc_params, e, src, delay);
 }
diff --git a/client/src/spaces.rs b/client/src/spaces.rs
@@ -3,7 +3,7 @@ use std::{net::SocketAddr, path::PathBuf, time::Duration};
 use anyhow::{anyhow, Context};
 use log::{info, warn};
 use spaces_protocol::{
-    bitcoin::{hashes::Hash, Block, BlockHash},
+    bitcoin::{Block, BlockHash},
     constants::ChainAnchor,
     hasher::BaseHash,
 };
@@ -255,86 +255,15 @@ impl Spaced {
         Ok(())
     }
 
-    pub async fn genesis(
-        rpc: &BitcoinRpc,
+    pub fn genesis(
         network: ExtendedNetwork,
-        mut shutdown: tokio::sync::broadcast::Receiver<()>
-    ) -> anyhow::Result<ChainAnchor> {
-        let mut anchor = match network {
+    ) -> ChainAnchor {
+        match network {
             ExtendedNetwork::Testnet => ChainAnchor::TESTNET(),
             ExtendedNetwork::Testnet4 => ChainAnchor::TESTNET4(),
             ExtendedNetwork::Regtest => ChainAnchor::REGTEST(),
             ExtendedNetwork::Mainnet => ChainAnchor::MAINNET(),
             _ => panic!("unsupported network"),
-        };
-
-
-
-        // Wait for the RPC node to be ready
-        let mut attempts = 0;
-        let mut last_error = BitcoinRpcError::Other("Unknown error".to_string());
-        loop {
-            if shutdown.try_recv().is_ok() {
-                return Err(anyhow!("Fetching activation height terminated: shutdown requested"))
-            }
-            if attempts > 5 {
-                return Err(anyhow!(
-                        "Could not retrieve activation height: {}",
-                        last_error
-                    ));
-            }
-
-            let rpc_task = rpc.clone();
-            let net_task = network.fallback_network();
-            let best_chain = tokio::task::spawn_blocking(move || {
-                let source = BitcoinBlockSource::new(rpc_task);
-                    source.get_best_chain(Some(anchor.height), net_task)
-            }).await.expect("join");
-
-            match best_chain {
-                Ok(Some(tip)) => {
-                    info!("Connect to RPC node (tip: {})", tip.height);
-                    if anchor.hash != BlockHash::all_zeros() {
-                        break;
-                    }
-
-                    // Pull the activation block hash
-                    let client = reqwest::Client::new();
-                    anchor.hash = match rpc
-                        .send_json(&client, &rpc.get_block_hash(anchor.height))
-                        .await
-                    {
-                        Ok(hash) => hash,
-                        Err(e) => {
-                            warn!("Fetching height {}:{}, retrying in 1s ...", anchor.height, e);
-                            last_error = e;
-                            match &last_error {
-                                BitcoinRpcError::Rpc(_) => {}
-                                _ => attempts += 1,
-                            }
-                            continue;
-                        }
-                    };
-
-                    break;
-                }
-                Ok(None) => {
-                    warn!("Connected RPC node is still syncing, waiting 5s ...");
-                    tokio::time::sleep(Duration::from_secs(5)).await;
-                }
-                Err(e) => {
-                    warn!("Error fetching blockchain info: {}, retrying in 1s ...", e);
-                    last_error = e;
-                    tokio::time::sleep(Duration::from_secs(1)).await;
-                    match &last_error {
-                        BitcoinRpcError::Rpc(_) => {}
-                        _ => attempts += 1,
-                    }
-                }
-            }
         }
-
-
-        Ok(anchor)
     }
 }
diff --git a/client/src/wallets.rs b/client/src/wallets.rs
diff --git a/protocol/src/constants.rs b/protocol/src/constants.rs

Original file line number	Diff line number	Diff line change
`@@ -100,8 +100,7 @@ impl App {`
`100`	`100`	`}`
`101`	`101`
`102`	`102`	`pub async fn run(&mut self, args: Vec<String>) -> anyhow::Result<()> {`
`103`		`- let shutdown_receiver = self.shutdown.subscribe();`
`104`		`- let spaced = Args::configure(args, shutdown_receiver).await?;`
	`103`	`+ let spaced = Args::configure(args).await?;`
`105`	`104`	`self.setup_rpc_services(&spaced).await;`
`106`	`105`	`self.setup_sync_service(spaced).await;`
`107`	`106`
Original file line number	Diff line number	Diff line change
`@@ -33,6 +33,7 @@ pub trait BlockSource {`
`33`	`33`	`fn get_blockchain_info(&self) -> Result<BlockchainInfo, BitcoinRpcError>;`
`34`	`34`	`fn get_block_filter_by_height(&self, height: u32) -> Result<Option<BlockFilterRpc>, BitcoinRpcError>;`
`35`	`35`	`fn queue_blocks(&self, heights: Vec<u32>) -> Result<(), BitcoinRpcError>;`
	`36`	`+ fn queue_filters(&self) -> Result<(), BitcoinRpcError>;`
`36`	`37`	`}`
`37`	`38`
`38`	`39`	`#[derive(Debug, Clone, Serialize, Deserialize)]`