rpc: init wrapper crate + global: snapshot

crates/brk_rpc/src/inner.rs

@@ -0,0 +1,139 @@
+use bitcoincore_rpc::{Client as CoreClient, Error as RpcError, jsonrpc};
+use brk_error::Result;
+use log::info;
+use parking_lot::RwLock;
+use std::time::Duration;
+
+pub use bitcoincore_rpc::Auth;
+
+pub struct ClientInner {
+    url: String,
+    auth: Auth,
+    client: RwLock<CoreClient>,
+    max_retries: usize,
+    retry_delay: Duration,
+}
+
+impl ClientInner {
+    pub fn new(url: &str, auth: Auth, max_retries: usize, retry_delay: Duration) -> Result<Self> {
+        let client = Self::retry(max_retries, retry_delay, || {
+            CoreClient::new(url, auth.clone()).map_err(Into::into)
+        })?;
+
+        Ok(Self {
+            url: url.to_string(),
+            auth,
+            client: RwLock::new(client),
+            max_retries,
+            retry_delay,
+        })
+    }
+
+    fn recreate(&self) -> Result<()> {
+        *self.client.write() = CoreClient::new(&self.url, self.auth.clone())?;
+        Ok(())
+    }
+
+    fn is_retriable(error: &RpcError) -> bool {
+        matches!(
+            error,
+            RpcError::JsonRpc(jsonrpc::Error::Rpc(e))
+                if e.code == -32600 || e.code == 401 || e.code == -28
+        ) || matches!(error, RpcError::JsonRpc(jsonrpc::Error::Transport(_)))
+    }
+
+    fn retry<F, T>(max_retries: usize, delay: Duration, mut f: F) -> Result<T>
+    where
+        F: FnMut() -> Result<T>,
+    {
+        let mut last_error = None;
+
+        for attempt in 0..=max_retries {
+            if attempt > 0 {
+                info!(
+                    "Retrying to connect to Bitcoin Core (attempt {}/{})",
+                    attempt, max_retries
+                );
+                std::thread::sleep(delay);
+            }
+
+            match f() {
+                Ok(value) => {
+                    if attempt > 0 {
+                        info!(
+                            "Successfully connected to Bitcoin Core after {} retries",
+                            attempt
+                        );
+                    }
+                    return Ok(value);
+                }
+                Err(e) => {
+                    if attempt == 0 {
+                        info!("Could not connect to Bitcoin Core, retrying: {}", e);
+                    }
+                    last_error = Some(e);
+                }
+            }
+        }
+
+        let err = last_error.unwrap();
+        info!(
+            "Failed to connect to Bitcoin Core after {} attempts",
+            max_retries + 1
+        );
+        Err(err)
+    }
+
+    pub fn call_with_retry<F, T>(&self, f: F) -> Result<T, RpcError>
+    where
+        F: Fn(&CoreClient) -> Result<T, RpcError>,
+    {
+        for attempt in 0..=self.max_retries {
+            if attempt > 0 {
+                info!(
+                    "Trying to reconnect to Bitcoin Core (attempt {}/{})",
+                    attempt, self.max_retries
+                );
+                self.recreate().ok();
+                std::thread::sleep(self.retry_delay);
+            }
+
+            match f(&self.client.read()) {
+                Ok(value) => {
+                    if attempt > 0 {
+                        info!(
+                            "Successfully reconnected to Bitcoin Core after {} attempts",
+                            attempt
+                        );
+                    }
+                    return Ok(value);
+                }
+                Err(e) if Self::is_retriable(&e) => {
+                    if attempt == 0 {
+                        info!("Lost connection to Bitcoin Core, reconnecting...");
+                    }
+                }
+                Err(e) => return Err(e),
+            }
+        }
+
+        info!(
+            "Could not reconnect to Bitcoin Core after {} attempts",
+            self.max_retries + 1
+        );
+        Err(RpcError::JsonRpc(jsonrpc::Error::Rpc(
+            jsonrpc::error::RpcError {
+                code: -1,
+                message: "Max retries exceeded".to_string(),
+                data: None,
+            },
+        )))
+    }
+
+    pub fn call_once<F, T>(&self, f: F) -> Result<T, RpcError>
+    where
+        F: Fn(&CoreClient) -> Result<T, RpcError>,
+    {
+        f(&self.client.read())
+    }
+}

crates/brk_rpc/src/lib.rs

@@ -0,0 +1,103 @@
+use bitcoin::BlockHash;
+use bitcoincore_rpc::json::GetBlockResult;
+use bitcoincore_rpc::{Client as CoreClient, Error as RpcError, RpcApi};
+use brk_error::Result;
+use std::sync::Arc;
+use std::time::Duration;
+
+pub use bitcoincore_rpc::Auth;
+
+mod inner;
+
+use inner::ClientInner;
+
+///
+/// Bitcoin Core RPC Client
+///
+/// Free to clone (Arc)
+///
+#[derive(Clone)]
+pub struct Client(Arc<ClientInner>);
+
+impl Client {
+    pub fn new(url: &str, auth: Auth) -> Result<Self> {
+        Self::new_with(url, auth, 1_000_000, Duration::from_secs(1))
+    }
+
+    pub fn new_with(
+        url: &str,
+        auth: Auth,
+        max_retries: usize,
+        retry_delay: Duration,
+    ) -> Result<Self> {
+        Ok(Self(Arc::new(ClientInner::new(
+            url,
+            auth,
+            max_retries,
+            retry_delay,
+        )?)))
+    }
+
+    pub fn get_block_info(&self, hash: &BlockHash) -> Result<GetBlockResult> {
+        self.call(|c| c.get_block_info(hash)).map_err(Into::into)
+    }
+
+    /// Checks if a block is in the main chain (has positive confirmations)
+    pub fn is_in_main_chain(&self, hash: &BlockHash) -> Result<bool> {
+        let block_info = self.get_block_info(hash)?;
+        Ok(block_info.confirmations > 0)
+    }
+
+    pub fn get_closest_valid_height(&self, hash: BlockHash) -> Result<u64> {
+        // First, try to get block info for the hash
+        match self.get_block_info(&hash) {
+            Ok(block_info) => {
+                // Check if this block is in the main chain
+                if self.is_in_main_chain(&hash)? {
+                    // Block is in the main chain
+                    Ok(block_info.height as u64)
+                } else {
+                    // Confirmations is -1, meaning it's on a fork
+                    // We need to find where it diverged from the main chain
+
+                    // Get the previous block hash and walk backwards
+                    let mut current_hash = block_info
+                        .previousblockhash
+                        .ok_or("Genesis block has no previous block")?;
+
+                    loop {
+                        if self.is_in_main_chain(&current_hash)? {
+                            // Found a block in the main chain
+                            let current_info = self.get_block_info(&current_hash)?;
+                            return Ok(current_info.height as u64);
+                        }
+
+                        // Continue walking backwards
+                        let current_info = self.get_block_info(&current_hash)?;
+                        current_hash = current_info
+                            .previousblockhash
+                            .ok_or("Reached genesis without finding main chain")?;
+                    }
+                }
+            }
+            Err(_) => {
+                // Block not found in the node's database at all
+                Err("Block hash not found in blockchain".into())
+            }
+        }
+    }
+
+    pub fn call<F, T>(&self, f: F) -> Result<T, RpcError>
+    where
+        F: Fn(&CoreClient) -> Result<T, RpcError>,
+    {
+        self.0.call_with_retry(f)
+    }
+
+    pub fn call_once<F, T>(&self, f: F) -> Result<T, RpcError>
+    where
+        F: Fn(&CoreClient) -> Result<T, RpcError>,
+    {
+        self.0.call_once(f)
+    }
+}