Introduce Runtime object allowng to detect outer runtime context

Instead of holding an `Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>`
and dealing with stuff like `tokio::task::block_in_place` at all
callsites, we introduce a `Runtime` object that takes care of the state
transitions, and allows to detect and reuse an outer runtime context.

We also adjust the `with_runtime` API to take a `tokio::runtime::Handle`
rather than an `Arc<Runtime>`.

Author: tnull

bindings/ldk_node.udl

@@ -64,7 +64,7 @@ dictionary LogRecord {
 
 [Trait, WithForeign]
 interface LogWriter {
-    void log(LogRecord record);
+	void log(LogRecord record);
 };
 
 interface Builder {
@@ -160,8 +160,8 @@ interface Node {
 
 [Enum]
 interface Bolt11InvoiceDescription {
-  Hash(string hash);
-  Direct(string description);
+	Hash(string hash);
+	Direct(string description);
 };
 
 interface Bolt11Payment {
@@ -252,6 +252,7 @@ interface LSPS1Liquidity {
 enum NodeError {
 	"AlreadyRunning",
 	"NotRunning",
+	"RuntimeSetupFailed",
 	"OnchainTxCreationFailed",
 	"ConnectionFailed",
 	"InvoiceCreationFailed",

src/builder.rs

@@ -27,6 +27,7 @@ use crate::liquidity::{
 use crate::logger::{log_error, log_info, LdkLogger, LogLevel, LogWriter, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::peer_store::PeerStore;
+use crate::runtime::Runtime;
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
 	ChainMonitor, ChannelManager, DynStore, GossipSync, Graph, KeysManager, MessageRouter,
@@ -1101,7 +1102,7 @@ fn build_with_store_internal(
 		},
 	};
 
-	let runtime = Arc::new(RwLock::new(None));
+	let runtime = Arc::new(Runtime::new(Arc::clone(&logger)));
 
 	// Initialize the ChainMonitor
 	let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(

src/chain/electrum.rs

@@ -15,6 +15,7 @@ use crate::fee_estimator::{
 	ConfirmationTarget,
 };
 use crate::logger::{log_bytes, log_error, log_info, log_trace, LdkLogger, Logger};
+use crate::runtime::Runtime;
 
 use lightning::chain::{Confirm, Filter, WatchedOutput};
 use lightning::util::ser::Writeable;
@@ -46,15 +47,14 @@ pub(crate) struct ElectrumRuntimeClient {
 	electrum_client: Arc<ElectrumClient>,
 	bdk_electrum_client: Arc<BdkElectrumClient<ElectrumClient>>,
 	tx_sync: Arc<ElectrumSyncClient<Arc<Logger>>>,
-	runtime: Arc<tokio::runtime::Runtime>,
+	runtime: Arc<Runtime>,
 	config: Arc<Config>,
 	logger: Arc<Logger>,
 }
 
 impl ElectrumRuntimeClient {
 	pub(crate) fn new(
-		server_url: String, runtime: Arc<tokio::runtime::Runtime>, config: Arc<Config>,
-		logger: Arc<Logger>,
+		server_url: String, runtime: Arc<Runtime>, config: Arc<Config>, logger: Arc<Logger>,
 	) -> Result<Self, Error> {
 		let electrum_config = ElectrumConfigBuilder::new()
 			.retry(ELECTRUM_CLIENT_NUM_RETRIES)
@@ -88,7 +88,7 @@ impl ElectrumRuntimeClient {
 		let now = Instant::now();
 
 		let tx_sync = Arc::clone(&self.tx_sync);
-		let spawn_fut = self.runtime.spawn_blocking(move || tx_sync.sync(confirmables));
+		let spawn_fut = self.runtime.spawn_blocking(move || tx_sync.sync(confirmables))?;
 		let timeout_fut =
 			tokio::time::timeout(Duration::from_secs(LDK_WALLET_SYNC_TIMEOUT_SECS), spawn_fut);
 
@@ -130,7 +130,7 @@ impl ElectrumRuntimeClient {
 				BDK_ELECTRUM_CLIENT_BATCH_SIZE,
 				true,
 			)
-		});
+		})?;
 		let wallet_sync_timeout_fut =
 			tokio::time::timeout(Duration::from_secs(BDK_WALLET_SYNC_TIMEOUT_SECS), spawn_fut);
 
@@ -159,7 +159,7 @@ impl ElectrumRuntimeClient {
 
 		let spawn_fut = self.runtime.spawn_blocking(move || {
 			bdk_electrum_client.sync(request, BDK_ELECTRUM_CLIENT_BATCH_SIZE, true)
-		});
+		})?;
 		let wallet_sync_timeout_fut =
 			tokio::time::timeout(Duration::from_secs(BDK_WALLET_SYNC_TIMEOUT_SECS), spawn_fut);
 
@@ -185,8 +185,13 @@ impl ElectrumRuntimeClient {
 		let txid = tx.compute_txid();
 		let tx_bytes = tx.encode();
 
-		let spawn_fut =
-			self.runtime.spawn_blocking(move || electrum_client.transaction_broadcast(&tx));
+		let spawn_fut = if let Ok(spawn_fut) =
+			self.runtime.spawn_blocking(move || electrum_client.transaction_broadcast(&tx))
+		{
+			spawn_fut
+		} else {
+			return;
+		};
 
 		let timeout_fut =
 			tokio::time::timeout(Duration::from_secs(TX_BROADCAST_TIMEOUT_SECS), spawn_fut);
@@ -233,7 +238,7 @@ impl ElectrumRuntimeClient {
 			batch.estimate_fee(num_blocks);
 		}
 
-		let spawn_fut = self.runtime.spawn_blocking(move || electrum_client.batch_call(&batch));
+		let spawn_fut = self.runtime.spawn_blocking(move || electrum_client.batch_call(&batch))?;
 
 		let timeout_fut = tokio::time::timeout(
 			Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),

src/chain/mod.rs

@@ -24,6 +24,7 @@ use crate::fee_estimator::{
 };
 use crate::io::utils::write_node_metrics;
 use crate::logger::{log_bytes, log_error, log_info, log_trace, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, NodeMetrics};
 
@@ -126,7 +127,7 @@ impl ElectrumRuntimeStatus {
 	}
 
 	pub(crate) fn start(
-		&mut self, server_url: String, runtime: Arc<tokio::runtime::Runtime>, config: Arc<Config>,
+		&mut self, server_url: String, runtime: Arc<Runtime>, config: Arc<Config>,
 		logger: Arc<Logger>,
 	) -> Result<(), Error> {
 		match self {
@@ -311,7 +312,7 @@ impl ChainSource {
 		}
 	}
 
-	pub(crate) fn start(&self, runtime: Arc<tokio::runtime::Runtime>) -> Result<(), Error> {
+	pub(crate) fn start(&self, runtime: Arc<Runtime>) -> Result<(), Error> {
 		match self {
 			Self::Electrum { server_url, electrum_runtime_status, config, logger, .. } => {
 				electrum_runtime_status.write().unwrap().start(

src/error.rs

@@ -5,6 +5,8 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
+use crate::runtime::RuntimeError;
+
 use bdk_chain::bitcoin::psbt::ExtractTxError as BdkExtractTxError;
 use bdk_chain::local_chain::CannotConnectError as BdkChainConnectionError;
 use bdk_chain::tx_graph::CalculateFeeError as BdkChainCalculateFeeError;
@@ -20,6 +22,8 @@ pub enum Error {
 	AlreadyRunning,
 	/// Returned when trying to stop [`crate::Node`] while it is not running.
 	NotRunning,
+	/// An attempt to setup a runtime has failed.
+	RuntimeSetupFailed,
 	/// An on-chain transaction could not be created.
 	OnchainTxCreationFailed,
 	/// A network connection has been closed.
@@ -127,6 +131,7 @@ impl fmt::Display for Error {
 		match *self {
 			Self::AlreadyRunning => write!(f, "Node is already running."),
 			Self::NotRunning => write!(f, "Node is not running."),
+			Self::RuntimeSetupFailed => write!(f, "Failed to setup a runtime."),
 			Self::OnchainTxCreationFailed => {
 				write!(f, "On-chain transaction could not be created.")
 			},
@@ -199,6 +204,16 @@ impl fmt::Display for Error {
 
 impl std::error::Error for Error {}
 
+impl From<RuntimeError> for Error {
+	fn from(runtime_error: RuntimeError) -> Self {
+		match runtime_error {
+			RuntimeError::SetupFailed => Self::RuntimeSetupFailed,
+			RuntimeError::AlreadyRunning => Self::AlreadyRunning,
+			RuntimeError::NotRunning => Self::NotRunning,
+		}
+	}
+}
+
 impl From<BdkSignerError> for Error {
 	fn from(_: BdkSignerError) -> Self {
 		Self::OnchainTxSigningFailed

src/event.rs

@@ -29,6 +29,8 @@ use crate::io::{
 };
 use crate::logger::{log_debug, log_error, log_info, LdkLogger};
 
+use crate::runtime::{Runtime, RuntimeError};
+
 use lightning::events::bump_transaction::BumpTransactionEvent;
 use lightning::events::{ClosureReason, PaymentPurpose, ReplayEvent};
 use lightning::events::{Event as LdkEvent, PaymentFailureReason};
@@ -53,7 +55,7 @@ use core::future::Future;
 use core::task::{Poll, Waker};
 use std::collections::VecDeque;
 use std::ops::Deref;
-use std::sync::{Arc, Condvar, Mutex, RwLock};
+use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
 
 /// An event emitted by [`Node`], which should be handled by the user.
@@ -451,7 +453,7 @@ where
 	liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 	payment_store: Arc<PaymentStore>,
 	peer_store: Arc<PeerStore<L>>,
-	runtime: Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>>,
+	runtime: Arc<Runtime>,
 	logger: L,
 	config: Arc<Config>,
 }
@@ -466,8 +468,8 @@ where
 		channel_manager: Arc<ChannelManager>, connection_manager: Arc<ConnectionManager<L>>,
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<Graph>,
 		liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
-		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>,
-		runtime: Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>>, logger: L, config: Arc<Config>,
+		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>, runtime: Arc<Runtime>,
+		logger: L, config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -1049,16 +1051,16 @@ where
 				let forwarding_channel_manager = self.channel_manager.clone();
 				let min = time_forwardable.as_millis() as u64;
 
-				let runtime_lock = self.runtime.read().unwrap();
-				debug_assert!(runtime_lock.is_some());
+				let future = async move {
+					let millis_to_sleep = thread_rng().gen_range(min..min * 5) as u64;
+					tokio::time::sleep(Duration::from_millis(millis_to_sleep)).await;
 
-				if let Some(runtime) = runtime_lock.as_ref() {
-					runtime.spawn(async move {
-						let millis_to_sleep = thread_rng().gen_range(min..min * 5) as u64;
-						tokio::time::sleep(Duration::from_millis(millis_to_sleep)).await;
+					forwarding_channel_manager.process_pending_htlc_forwards();
+				};
 
-						forwarding_channel_manager.process_pending_htlc_forwards();
-					});
+				if let Err(RuntimeError::NotRunning) = self.runtime.spawn(future) {
+					log_error!(self.logger, "Tried spawing a future while the runtime wasn't available. This should never happen.");
+					debug_assert!(false, "Tried spawing a future while the runtime wasn't available. This should never happen.");
 				}
 			},
 			LdkEvent::SpendableOutputs { outputs, channel_id } => {
@@ -1419,30 +1421,29 @@ where
 				debug_assert!(false, "We currently don't handle BOLT12 invoices manually, so this event should never be emitted.");
 			},
 			LdkEvent::ConnectionNeeded { node_id, addresses } => {
-				let runtime_lock = self.runtime.read().unwrap();
-				debug_assert!(runtime_lock.is_some());
-
-				if let Some(runtime) = runtime_lock.as_ref() {
-					let spawn_logger = self.logger.clone();
-					let spawn_cm = Arc::clone(&self.connection_manager);
-					runtime.spawn(async move {
-						for addr in &addresses {
-							match spawn_cm.connect_peer_if_necessary(node_id, addr.clone()).await {
-								Ok(()) => {
-									return;
-								},
-								Err(e) => {
-									log_error!(
-										spawn_logger,
-										"Failed to establish connection to peer {}@{}: {}",
-										node_id,
-										addr,
-										e
-									);
-								},
-							}
+				let spawn_logger = self.logger.clone();
+				let spawn_cm = Arc::clone(&self.connection_manager);
+				let future = async move {
+					for addr in &addresses {
+						match spawn_cm.connect_peer_if_necessary(node_id, addr.clone()).await {
+							Ok(()) => {
+								return;
+							},
+							Err(e) => {
+								log_error!(
+									spawn_logger,
+									"Failed to establish connection to peer {}@{}: {}",
+									node_id,
+									addr,
+									e
+								);
+							},
 						}
-					});
+					}
+				};
+				if let Err(RuntimeError::NotRunning) = self.runtime.spawn(future) {
+					log_error!(self.logger, "Tried spawing a future while the runtime wasn't available. This should never happen.");
+					debug_assert!(false, "Tried spawing a future while the runtime wasn't available. This should never happen.");
 				}
 			},
 			LdkEvent::BumpTransaction(bte) => {

src/gossip.rs

@@ -8,14 +8,15 @@
 use crate::chain::ChainSource;
 use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, LdkLogger, Logger};
+use crate::runtime::{Runtime, RuntimeError};
 use crate::types::{GossipSync, Graph, P2PGossipSync, PeerManager, RapidGossipSync, UtxoLookup};
 use crate::Error;
 
 use lightning_block_sync::gossip::{FutureSpawner, GossipVerifier};
 
 use std::future::Future;
 use std::sync::atomic::{AtomicU32, Ordering};
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::Duration;
 
 pub(crate) enum GossipSource {
@@ -63,7 +64,7 @@ impl GossipSource {
 
 	pub(crate) fn set_gossip_verifier(
 		&self, chain_source: Arc<ChainSource>, peer_manager: Arc<PeerManager>,
-		runtime: Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>>,
+		runtime: Arc<Runtime>,
 	) {
 		match self {
 			Self::P2PNetwork { gossip_sync, logger } => {
@@ -133,28 +134,21 @@ impl GossipSource {
 }
 
 pub(crate) struct RuntimeSpawner {
-	runtime: Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl RuntimeSpawner {
-	pub(crate) fn new(
-		runtime: Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>>, logger: Arc<Logger>,
-	) -> Self {
+	pub(crate) fn new(runtime: Arc<Runtime>, logger: Arc<Logger>) -> Self {
 		Self { runtime, logger }
 	}
 }
 
 impl FutureSpawner for RuntimeSpawner {
 	fn spawn<T: Future<Output = ()> + Send + 'static>(&self, future: T) {
-		let rt_lock = self.runtime.read().unwrap();
-		if rt_lock.is_none() {
+		if let Err(RuntimeError::NotRunning) = self.runtime.spawn(future) {
 			log_error!(self.logger, "Tried spawing a future while the runtime wasn't available. This should never happen.");
 			debug_assert!(false, "Tried spawing a future while the runtime wasn't available. This should never happen.");
-			return;
 		}
-
-		let runtime = rt_lock.as_ref().unwrap();
-		runtime.spawn(future);
 	}
 }