Switch to async traits for bitcoind and KVStore outside of monitors

With LDK 0.2, most traits now offer async variants. In the previous
commit we ignored these but here we migrate to them everywhere
except in the `ChannelMonitor` persistence.

Author: TheBlueMatt

Cargo.toml

@@ -13,7 +13,7 @@ lightning-block-sync = { version = "0.2.0-beta1", features = [ "rpc-client", "to
 lightning-dns-resolver = { version = "0.3.0-beta1" }
 lightning-invoice = { version = "0.34.0-beta1" }
 lightning-net-tokio = { version = "0.2.0-beta1" }
-lightning-persister = { version = "0.2.0-beta1" }
+lightning-persister = { version = "0.2.0-beta1", features = [ "tokio" ] }
 lightning-background-processor = { version = "0.2.0-beta1" }
 lightning-rapid-gossip-sync = { version = "0.2.0-beta1" }
 lightning-macros = { version = "0.2.0-beta1" }

src/bitcoind_client.rs

@@ -16,10 +16,10 @@ use bitcoin::key::XOnlyPublicKey;
 use bitcoin::psbt::Psbt;
 use bitcoin::{Network, OutPoint, TxOut, WPubkeyHash};
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
-use lightning::events::bump_transaction::Utxo;
-use lightning::events::bump_transaction::sync::WalletSourceSync;
+use lightning::events::bump_transaction::{Utxo, WalletSource};
 use lightning::log_error;
-use lightning::sign::ChangeDestinationSourceSync;
+use lightning::sign::ChangeDestinationSource;
+use lightning::util::async_poll::AsyncResult;
 use lightning::util::logger::Logger;
 use lightning_block_sync::http::HttpEndpoint;
 use lightning_block_sync::rpc::RpcClient;
@@ -32,7 +32,7 @@ use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-use tokio::runtime::{self, Runtime};
+use tokio::runtime::Handle;
 
 pub struct BitcoindClient {
 	pub(crate) bitcoind_rpc_client: Arc<RpcClient>,
@@ -42,8 +42,7 @@ pub struct BitcoindClient {
 	rpc_user: String,
 	rpc_password: String,
 	fees: Arc<HashMap<ConfirmationTarget, AtomicU32>>,
-	main_runtime_handle: runtime::Handle,
-	inner_runtime: Arc<Runtime>,
+	main_runtime_handle: Handle,
 	logger: Arc<FilesystemLogger>,
 }
 
@@ -71,7 +70,7 @@ const MIN_FEERATE: u32 = 253;
 impl BitcoindClient {
 	pub(crate) async fn new(
 		host: String, port: u16, rpc_user: String, rpc_password: String, network: Network,
-		handle: runtime::Handle, logger: Arc<FilesystemLogger>,
+		handle: Handle, logger: Arc<FilesystemLogger>,
 	) -> std::io::Result<Self> {
 		let http_endpoint = HttpEndpoint::for_host(host.clone()).with_port(port);
 		let rpc_credentials =
@@ -100,15 +99,6 @@ impl BitcoindClient {
 		fees.insert(ConfirmationTarget::ChannelCloseMinimum, AtomicU32::new(MIN_FEERATE));
 		fees.insert(ConfirmationTarget::OutputSpendingFee, AtomicU32::new(MIN_FEERATE));
 
-		let mut builder = runtime::Builder::new_multi_thread();
-		let runtime =
-			builder.enable_all().worker_threads(1).thread_name("rpc-worker").build().unwrap();
-		let inner_runtime = Arc::new(runtime);
-		// Tokio will panic if we drop a runtime while in another runtime. Because the entire
-		// application runs inside a tokio runtime, we have to ensure this runtime is never
-		// `drop`'d, which we do by leaking an Arc reference.
-		std::mem::forget(Arc::clone(&inner_runtime));
-
 		let client = Self {
 			bitcoind_rpc_client: Arc::new(bitcoind_rpc_client),
 			host,
@@ -118,7 +108,6 @@ impl BitcoindClient {
 			network,
 			fees: Arc::new(fees),
 			main_runtime_handle: handle.clone(),
-			inner_runtime,
 			logger,
 		};
 		BitcoindClient::poll_for_fee_estimates(
@@ -131,7 +120,7 @@ impl BitcoindClient {
 
 	fn poll_for_fee_estimates(
 		fees: Arc<HashMap<ConfirmationTarget, AtomicU32>>, rpc_client: Arc<RpcClient>,
-		handle: tokio::runtime::Handle,
+		handle: Handle,
 	) {
 		handle.spawn(async move {
 			loop {
@@ -241,39 +230,6 @@ impl BitcoindClient {
 		});
 	}
 
-	fn run_future_in_blocking_context<F: Future + Send + 'static>(&self, future: F) -> F::Output
-	where
-		F::Output: Send + 'static,
-	{
-		// Tokio deliberately makes it nigh impossible to block on a future in a sync context that
-		// is running in an async task (which makes it really hard to interact with sync code that
-		// has callbacks in an async project).
-		//
-		// Reading the docs, it *seems* like
-		// `tokio::task::block_in_place(tokio::runtime::Handle::spawn(future))` should do the
-		// trick, and 99.999% of the time it does! But tokio has a "non-stealable I/O driver" - if
-		// the task we're running happens to, by sheer luck, be holding the "I/O driver" when we go
-		// into a `block_in_place` call, and the inner future requires I/O (which of course it
-		// does, its a future!), the whole thing will come to a grinding halt as no other thread is
-		// allowed to poll I/O until the blocked one finishes.
-		//
-		// This is, of course, nuts, and an almost trivial performance penalty of occasional
-		// additional wakeups would solve this, but tokio refuses to do so because any performance
-		// penalty at all would be too much (tokio issue #4730).
-		//
-		// Instead, we have to do a rather insane dance - we have to spawn the `future` we want to
-		// run on a *different* (threaded) tokio runtime (doing the `block_in_place` dance to avoid
-		// blocking too many threads on the main runtime). We want to block on that `future` being
-		// run on the other runtime's threads, but tokio only provides `block_on` to do so, which
-		// runs the `future` itself on the current thread, panicing if this thread is already a
-		// part of a tokio runtime (which in this case it is - the main tokio runtime). Thus, we
-		// have to `spawn` the `future` on the secondary runtime and then `block_on` the resulting
-		// `JoinHandle` on the main runtime.
-		tokio::task::block_in_place(move || {
-			self.main_runtime_handle.block_on(self.inner_runtime.spawn(future)).unwrap()
-		})
-	}
-
 	pub fn get_new_rpc_client(&self) -> RpcClient {
 		let http_endpoint = HttpEndpoint::for_host(self.host.clone()).with_port(self.port);
 		let rpc_credentials = base64::encode(format!("{}:{}", self.rpc_user, self.rpc_password));
@@ -406,60 +362,64 @@ impl BroadcasterInterface for BitcoindClient {
 	}
 }
 
-impl ChangeDestinationSourceSync for BitcoindClient {
-	fn get_change_destination_script(&self) -> Result<ScriptBuf, ()> {
-		let future = self.get_new_address();
-		Ok(self.run_future_in_blocking_context(async move { future.await.script_pubkey() }))
+impl ChangeDestinationSource for BitcoindClient {
+	fn get_change_destination_script<'a>(&'a self) -> AsyncResult<'a, ScriptBuf> {
+		Box::pin(async move {
+			Ok(self.get_new_address().await.script_pubkey())
+		})
 	}
 }
 
-impl WalletSourceSync for BitcoindClient {
-	fn list_confirmed_utxos(&self) -> Result<Vec<Utxo>, ()> {
-		let future = self.list_unspent();
-		let utxos = self.run_future_in_blocking_context(async move { future.await.0 });
-		Ok(utxos
-			.into_iter()
-			.filter_map(|utxo| {
-				let outpoint = OutPoint { txid: utxo.txid, vout: utxo.vout };
-				let value = bitcoin::Amount::from_sat(utxo.amount);
-				match utxo.address.witness_program() {
-					Some(prog) if prog.is_p2wpkh() => {
-						WPubkeyHash::from_slice(prog.program().as_bytes())
-							.map(|wpkh| Utxo::new_v0_p2wpkh(outpoint, value, &wpkh))
-							.ok()
-					},
-					Some(prog) if prog.is_p2tr() => {
-						// TODO: Add `Utxo::new_v1_p2tr` upstream.
-						XOnlyPublicKey::from_slice(prog.program().as_bytes())
-							.map(|_| Utxo {
-								outpoint,
-								output: TxOut {
-									value,
-									script_pubkey: utxo.address.script_pubkey(),
-								},
-								satisfaction_weight: 1 /* empty script_sig */ * WITNESS_SCALE_FACTOR as u64 +
-									1 /* witness items */ + 1 /* schnorr sig len */ + 64, /* schnorr sig */
-							})
-							.ok()
-					},
-					_ => None,
-				}
-			})
-			.collect())
+impl WalletSource for BitcoindClient {
+	fn list_confirmed_utxos<'a>(&'a self) -> AsyncResult<'a, Vec<Utxo>> {
+		Box::pin(async move {
+			let utxos = self.list_unspent().await.0;
+			Ok(utxos
+				.into_iter()
+				.filter_map(|utxo| {
+					let outpoint = OutPoint { txid: utxo.txid, vout: utxo.vout };
+					let value = bitcoin::Amount::from_sat(utxo.amount);
+					match utxo.address.witness_program() {
+						Some(prog) if prog.is_p2wpkh() => {
+							WPubkeyHash::from_slice(prog.program().as_bytes())
+								.map(|wpkh| Utxo::new_v0_p2wpkh(outpoint, value, &wpkh))
+								.ok()
+						},
+						Some(prog) if prog.is_p2tr() => {
+							// TODO: Add `Utxo::new_v1_p2tr` upstream.
+							XOnlyPublicKey::from_slice(prog.program().as_bytes())
+								.map(|_| Utxo {
+									outpoint,
+									output: TxOut {
+										value,
+										script_pubkey: utxo.address.script_pubkey(),
+									},
+									satisfaction_weight: 1 /* empty script_sig */ * WITNESS_SCALE_FACTOR as u64 +
+										1 /* witness items */ + 1 /* schnorr sig len */ + 64, /* schnorr sig */
+								})
+								.ok()
+						},
+						_ => None,
+					}
+				})
+				.collect())
+		})
 	}
 
-	fn get_change_script(&self) -> Result<ScriptBuf, ()> {
-		let future = self.get_new_address();
-		Ok(self.run_future_in_blocking_context(async move { future.await.script_pubkey() }))
+	fn get_change_script<'a>(&'a self) -> AsyncResult<'a, ScriptBuf> {
+		Box::pin(async move {
+			Ok(self.get_new_address().await.script_pubkey())
+		})
 	}
 
-	fn sign_psbt(&self, tx: Psbt) -> Result<Transaction, ()> {
-		let mut tx_bytes = Vec::new();
-		let _ = tx.unsigned_tx.consensus_encode(&mut tx_bytes).map_err(|_| ());
-		let tx_hex = hex_utils::hex_str(&tx_bytes);
-		let future = self.sign_raw_transaction_with_wallet(tx_hex);
-		let signed_tx = self.run_future_in_blocking_context(async move { future.await });
-		let signed_tx_bytes = hex_utils::to_vec(&signed_tx.hex).ok_or(())?;
-		Transaction::consensus_decode(&mut signed_tx_bytes.as_slice()).map_err(|_| ())
+	fn sign_psbt<'a>(&'a self, tx: Psbt) -> AsyncResult<'a, Transaction> {
+		Box::pin(async move {
+			let mut tx_bytes = Vec::new();
+			let _ = tx.unsigned_tx.consensus_encode(&mut tx_bytes).map_err(|_| ());
+			let tx_hex = hex_utils::hex_str(&tx_bytes);
+			let signed_tx = self.sign_raw_transaction_with_wallet(tx_hex).await;
+			let signed_tx_bytes = hex_utils::to_vec(&signed_tx.hex).ok_or(())?;
+			Transaction::consensus_decode(&mut signed_tx_bytes.as_slice()).map_err(|_| ())
+		})
 	}
 }