Merge pull request fedimint#6202 from tvolk131/remove_block_in_place

chore: remove block_in_place usage in devimint external

devimint/src/external.rs

@@ -22,6 +22,7 @@ use itertools::Itertools;
 use ln_gateway::envs::FM_CLN_EXTENSION_LISTEN_ADDRESS_ENV;
 use tokio::fs;
 use tokio::sync::{MappedMutexGuard, Mutex, MutexGuard};
+use tokio::task::spawn_blocking;
 use tokio::time::Instant;
 use tonic_lnd::lnrpc::{ChanInfoRequest, GetInfoRequest, ListChannelsRequest};
 use tonic_lnd::Client as LndClient;
@@ -112,9 +113,7 @@ impl Bitcoind {
         debug!("Setting up bitcoind");
         // create RPC wallet
         for attempt in 0.. {
-            match fedimint_core::runtime::block_in_place(|| {
-                client.create_wallet("", None, None, None, None)
-            }) {
+            match block_in_place(|| client.create_wallet("", None, None, None, None)) {
                 Ok(_) => {
                     break;
                 }
@@ -166,6 +165,7 @@ impl Bitcoind {
     pub async fn poll_ready(&self) -> anyhow::Result<()> {
         poll("bitcoind rpc ready", || async {
             self.get_block_count()
+                .await
                 .map_err(ControlFlow::Continue::<anyhow::Error, _>)?;
             Ok(())
         })
@@ -184,16 +184,17 @@ impl Bitcoind {
     /// Fedimint's IBitcoindRpc considers block count the total number of
     /// blocks, where bitcoind's rpc returns the height. Since the genesis
     /// block has height 0, we need to add 1 to get the total block count.
-    pub fn get_block_count(&self) -> Result<u64> {
-        Ok(block_in_place(|| self.client.get_block_count())? + 1)
+    pub async fn get_block_count(&self) -> Result<u64> {
+        let client = self.client.clone();
+        Ok(spawn_blocking(move || client.get_block_count()).await?? + 1)
     }
 
     pub async fn mine_blocks_no_wait(&self, block_num: u64) -> Result<u64> {
         let start_time = Instant::now();
         debug!(target: LOG_DEVIMINT, ?block_num, "Mining bitcoin blocks");
         let addr = self.get_new_address().await?;
-        let initial_block_count = self.get_block_count()?;
-        self.generate_to_address(block_num, &addr).await?;
+        let initial_block_count = self.get_block_count().await?;
+        self.generate_to_address(block_num, addr).await?;
         debug!(target: LOG_DEVIMINT,
             elapsed_ms = %start_time.elapsed().as_millis(),
             ?block_num, "Mined blocks (no wait)");
@@ -205,9 +206,9 @@ impl Bitcoind {
         let start_time = Instant::now();
         debug!(target: LOG_DEVIMINT, ?block_num, "Mining bitcoin blocks");
         let addr = self.get_new_address().await?;
-        let initial_block_count = self.get_block_count()?;
-        self.generate_to_address(block_num, &addr).await?;
-        while self.get_block_count()? < initial_block_count + block_num {
+        let initial_block_count = self.get_block_count().await?;
+        self.generate_to_address(block_num, addr).await?;
+        while self.get_block_count().await? < initial_block_count + block_num {
             trace!(target: LOG_DEVIMINT, ?block_num, "Waiting for blocks to be mined");
             sleep(Duration::from_millis(100)).await;
         }
@@ -225,70 +226,66 @@ impl Bitcoind {
         let tx = self
             .wallet_client()
             .await?
-            .send_to_address(&bitcoin::Address::from_str(&addr)?.assume_checked(), amount)
+            .send_to_address(bitcoin::Address::from_str(&addr)?.assume_checked(), amount)
             .await?;
         Ok(tx)
     }
 
-    pub async fn get_txout_proof(&self, txid: &bitcoin::Txid) -> Result<String> {
-        let proof = self.get_tx_out_proof(&[*txid], None).await?;
+    pub async fn get_txout_proof(&self, txid: bitcoin::Txid) -> Result<String> {
+        let client = self.wallet_client().await?.clone();
+        let proof = spawn_blocking(move || client.client.get_tx_out_proof(&[txid], None)).await??;
         Ok(proof.encode_hex())
     }
 
-    pub fn get_raw_transaction(&self, txid: &bitcoin::Txid) -> Result<String> {
-        let tx = block_in_place(|| self.client.get_raw_transaction(txid, None))?;
+    pub async fn get_raw_transaction(&self, txid: bitcoin::Txid) -> Result<String> {
+        let client = self.client.clone();
+        let tx = spawn_blocking(move || client.get_raw_transaction(&txid, None)).await??;
         let bytes = tx.consensus_encode_to_vec();
         Ok(bytes.encode_hex())
     }
 
     pub async fn get_new_address(&self) -> Result<Address> {
-        let client = &self.wallet_client().await?;
-        let addr = block_in_place(|| client.client.get_new_address(None, None))?.assume_checked();
+        let client = self.wallet_client().await?.clone();
+        let addr = spawn_blocking(move || client.client.get_new_address(None, None))
+            .await??
+            .assume_checked();
         Ok(addr)
     }
 
     pub async fn generate_to_address(
         &self,
         block_num: u64,
-        address: &Address,
+        address: Address,
     ) -> Result<Vec<BlockHash>> {
-        let client = &self.wallet_client().await?;
-        Ok(block_in_place(|| {
-            client.client.generate_to_address(block_num, address)
-        })?)
-    }
-
-    pub async fn get_tx_out_proof(
-        &self,
-        txid: &[bitcoin::Txid],
-        block_hash: Option<&BlockHash>,
-    ) -> anyhow::Result<Vec<u8>> {
-        let client = &self.wallet_client().await?;
-        Ok(block_in_place(|| {
-            client.client.get_tx_out_proof(txid, block_hash)
-        })?)
+        let client = self.wallet_client().await?.clone();
+        Ok(
+            spawn_blocking(move || client.client.generate_to_address(block_num, &address))
+                .await??,
+        )
     }
 
-    pub fn get_blockchain_info(&self) -> anyhow::Result<GetBlockchainInfoResult> {
-        Ok(block_in_place(|| self.client.get_blockchain_info())?)
+    pub async fn get_blockchain_info(&self) -> anyhow::Result<GetBlockchainInfoResult> {
+        let client = self.client.clone();
+        Ok(spawn_blocking(move || client.get_blockchain_info()).await??)
     }
 
     pub async fn send_to_address(
         &self,
-        addr: &Address,
+        addr: Address,
         amount: bitcoin::Amount,
     ) -> anyhow::Result<bitcoin::Txid> {
-        let client = self.wallet_client().await?;
-        Ok(block_in_place(|| {
+        let client = self.wallet_client().await?.clone();
+        Ok(spawn_blocking(move || {
             client
                 .client
-                .send_to_address(addr, amount, None, None, None, None, None, None)
-        })?)
+                .send_to_address(&addr, amount, None, None, None, None, None, None)
+        })
+        .await??)
     }
 
     pub(crate) async fn get_balances(&self) -> anyhow::Result<GetBalancesResult> {
-        let client = self.wallet_client().await?;
-        Ok(block_in_place(|| client.client.get_balances())?)
+        let client = self.wallet_client().await?.clone();
+        Ok(spawn_blocking(move || client.client.get_balances()).await??)
     }
 
     pub(crate) fn get_jsonrpc_client(&self) -> &bitcoincore_rpc::jsonrpc::Client {
@@ -398,6 +395,7 @@ impl Lightningd {
             let btc_height = self
                 .bitcoind
                 .get_blockchain_info()
+                .await
                 .context("bitcoind getblockchaininfo")
                 .map_err(ControlFlow::Continue)?
                 .blocks;
@@ -958,7 +956,7 @@ pub async fn open_channels_between_gateways(
             loop {
                 // Bitcoind's getrawtransaction RPC call will return an error if the transaction
                 // is not known.
-                if bitcoind.get_raw_transaction(txid).is_ok() {
+                if bitcoind.get_raw_transaction(*txid).await.is_ok() {
                     break;
                 }
 

devimint/src/federation.rs

@@ -767,7 +767,7 @@ impl Federation {
 
     pub async fn await_block_sync(&self) -> Result<u64> {
         let finality_delay = self.get_finality_delay()?;
-        let block_count = self.bitcoind.get_block_count()?;
+        let block_count = self.bitcoind.get_block_count().await?;
         let expected = block_count.saturating_sub(finality_delay.into());
         cmd!(
             self.internal_client().await?,

devimint/src/gatewayd.rs

@@ -377,7 +377,11 @@ impl Gatewayd {
 
     pub async fn wait_for_chain_sync(&self, bitcoind: &Bitcoind) -> Result<()> {
         poll("lightning node block processing", || async {
-            let block_height = bitcoind.get_block_count().map_err(ControlFlow::Continue)? - 1;
+            let block_height = bitcoind
+                .get_block_count()
+                .await
+                .map_err(ControlFlow::Continue)?
+                - 1;
             cmd!(
                 self,
                 "lightning",

devimint/src/tests.rs

@@ -1088,7 +1088,8 @@ pub async fn cli_tests(dev_fed: DevFed) -> Result<()> {
     let tx_hex = poll("Waiting for transaction in mempool", || async {
         // TODO: distinguish errors from not found
         bitcoind
-            .get_raw_transaction(&txid)
+            .get_raw_transaction(txid)
+            .await
             .context("getrawtransaction")
             .map_err(ControlFlow::Continue)
     })
@@ -2034,7 +2035,8 @@ pub async fn recoverytool_test(dev_fed: DevFed) -> Result<()> {
             .expect("txid should be parsable");
         let tx_hex = poll("Waiting for transaction in mempool", || async {
             bitcoind
-                .get_raw_transaction(&txid)
+                .get_raw_transaction(txid)
+                .await
                 .context("getrawtransaction")
                 .map_err(ControlFlow::Continue)
         })

modules/fedimint-wallet-tests/src/bin/circular-deposit-test.rs

@@ -43,7 +43,8 @@ async fn assert_withdrawal(
     let txid: Txid = withdraw_res["txid"].as_str().unwrap().parse().unwrap();
     let tx_hex = poll("Waiting for transaction in mempool", || async {
         bitcoind
-            .get_raw_transaction(&txid)
+            .get_raw_transaction(txid)
+            .await
             .context("getrawtransaction")
             .map_err(ControlFlow::Continue)
     })