f Switch to runtime building on initialization

Author: tnull

bindings/ldk_node.udl

@@ -252,7 +252,6 @@ interface LSPS1Liquidity {
 enum NodeError {
 	"AlreadyRunning",
 	"NotRunning",
-	"RuntimeSetupFailed",
 	"OnchainTxCreationFailed",
 	"ConnectionFailed",
 	"InvoiceCreationFailed",
@@ -331,6 +330,7 @@ enum BuildError {
 	"InvalidListeningAddresses",
 	"InvalidAnnouncementAddresses",
 	"InvalidNodeAlias",
+	"RuntimeSetupFailed",
 	"ReadFailed",
 	"WriteFailed",
 	"StoragePathAccessFailed",

src/builder.rs

@@ -155,6 +155,8 @@ pub enum BuildError {
 	InvalidAnnouncementAddresses,
 	/// The provided alias is invalid.
 	InvalidNodeAlias,
+	/// An attempt to setup a runtime has failed.
+	RuntimeSetupFailed,
 	/// We failed to read data from the [`KVStore`].
 	///
 	/// [`KVStore`]: lightning::util::persist::KVStore
@@ -192,6 +194,7 @@ impl fmt::Display for BuildError {
 			Self::InvalidAnnouncementAddresses => {
 				write!(f, "Given announcement addresses are invalid.")
 			},
+			Self::RuntimeSetupFailed => write!(f, "Failed to setup a runtime."),
 			Self::ReadFailed => write!(f, "Failed to read from store."),
 			Self::WriteFailed => write!(f, "Failed to write to store."),
 			Self::StoragePathAccessFailed => write!(f, "Failed to access the given storage path."),
@@ -223,6 +226,7 @@ pub struct NodeBuilder {
 	gossip_source_config: Option<GossipSourceConfig>,
 	liquidity_source_config: Option<LiquiditySourceConfig>,
 	log_writer_config: Option<LogWriterConfig>,
+	runtime_handle: Option<tokio::runtime::Handle>,
 }
 
 impl NodeBuilder {
@@ -239,16 +243,27 @@ impl NodeBuilder {
 		let gossip_source_config = None;
 		let liquidity_source_config = None;
 		let log_writer_config = None;
+		let runtime_handle = None;
 		Self {
 			config,
 			entropy_source_config,
 			chain_data_source_config,
 			gossip_source_config,
 			liquidity_source_config,
 			log_writer_config,
+			runtime_handle,
 		}
 	}
 
+	/// Configures the [`Node`] instance to (re-)use a specific `tokio` runtime.
+	///
+	/// If not provided, the node will spawn its own runtime or reuse any outer runtime context it
+	/// can detect.
+	pub fn set_runtime(&mut self, runtime_handle: tokio::runtime::Handle) -> &mut Self {
+		self.runtime_handle = Some(runtime_handle);
+		self
+	}
+
 	/// Configures the [`Node`] instance to source its wallet entropy from a seed file on disk.
 	///
 	/// If the given file does not exist a new random seed file will be generated and
@@ -583,6 +598,15 @@ impl NodeBuilder {
 	) -> Result<Node, BuildError> {
 		let logger = setup_logger(&self.log_writer_config, &self.config)?;
 
+		let runtime = if let Some(handle) = self.runtime_handle.as_ref() {
+			Arc::new(Runtime::with_handle(handle.clone()))
+		} else {
+			Arc::new(Runtime::new().map_err(|e| {
+				log_error!(logger, "Failed to setup tokio runtime: {}", e);
+				BuildError::RuntimeSetupFailed
+			})?)
+		};
+
 		let seed_bytes = seed_bytes_from_config(
 			&self.config,
 			self.entropy_source_config.as_ref(),
@@ -611,6 +635,7 @@ impl NodeBuilder {
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
 			seed_bytes,
+			runtime,
 			logger,
 			Arc::new(vss_store),
 		)
@@ -620,6 +645,15 @@ impl NodeBuilder {
 	pub fn build_with_store(&self, kv_store: Arc<DynStore>) -> Result<Node, BuildError> {
 		let logger = setup_logger(&self.log_writer_config, &self.config)?;
 
+		let runtime = if let Some(handle) = self.runtime_handle.as_ref() {
+			Arc::new(Runtime::with_handle(handle.clone()))
+		} else {
+			Arc::new(Runtime::new().map_err(|e| {
+				log_error!(logger, "Failed to setup tokio runtime: {}", e);
+				BuildError::RuntimeSetupFailed
+			})?)
+		};
+
 		let seed_bytes = seed_bytes_from_config(
 			&self.config,
 			self.entropy_source_config.as_ref(),
@@ -633,6 +667,7 @@ impl NodeBuilder {
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
 			seed_bytes,
+			runtime,
 			logger,
 			kv_store,
 		)
@@ -935,7 +970,7 @@ fn build_with_store_internal(
 	config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
 	gossip_source_config: Option<&GossipSourceConfig>,
 	liquidity_source_config: Option<&LiquiditySourceConfig>, seed_bytes: [u8; 64],
-	logger: Arc<Logger>, kv_store: Arc<DynStore>,
+	runtime: Arc<Runtime>, logger: Arc<Logger>, kv_store: Arc<DynStore>,
 ) -> Result<Node, BuildError> {
 	if let Err(err) = may_announce_channel(&config) {
 		if config.announcement_addresses.is_some() {
@@ -1102,8 +1137,6 @@ fn build_with_store_internal(
 		},
 	};
 
-	let runtime = Arc::new(Runtime::new(Arc::clone(&logger)));
-
 	// Initialize the ChainMonitor
 	let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(
 		Some(Arc::clone(&chain_source)),
@@ -1496,6 +1529,8 @@ fn build_with_store_internal(
 	let (stop_sender, _) = tokio::sync::watch::channel(());
 	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
+	let is_running = Arc::new(RwLock::new(false));
+
 	Ok(Node {
 		runtime,
 		stop_sender,
@@ -1521,6 +1556,7 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		is_running,
 		is_listening,
 		node_metrics,
 	})

src/chain/electrum.rs

@@ -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,18 +185,8 @@ impl ElectrumRuntimeClient {
 		let txid = tx.compute_txid();
 		let tx_bytes = tx.encode();
 
-		let spawn_fut = if let Ok(spawn_fut) =
-			self.runtime.spawn_blocking(move || electrum_client.transaction_broadcast(&tx))
-		{
-			spawn_fut
-		} else {
-			debug_assert!(
-				false,
-				"Failed to broadcast due to runtime being unavailable. This should never happen."
-			);
-			return;
-		};
-
+		let spawn_fut =
+			self.runtime.spawn_blocking(move || electrum_client.transaction_broadcast(&tx));
 		let timeout_fut =
 			tokio::time::timeout(Duration::from_secs(TX_BROADCAST_TIMEOUT_SECS), spawn_fut);
 
@@ -242,7 +232,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/error.rs

@@ -5,8 +5,6 @@
 // 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;
@@ -22,8 +20,6 @@ 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.
@@ -131,7 +127,6 @@ 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.")
 			},
@@ -204,16 +199,6 @@ 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,7 +29,7 @@ use crate::io::{
 };
 use crate::logger::{log_debug, log_error, log_info, LdkLogger};
 
-use crate::runtime::{Runtime, RuntimeError};
+use crate::runtime::Runtime;
 
 use lightning::events::bump_transaction::BumpTransactionEvent;
 use lightning::events::{ClosureReason, PaymentPurpose, ReplayEvent};
@@ -1058,10 +1058,7 @@ where
 					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.");
-				}
+				self.runtime.spawn(future);
 			},
 			LdkEvent::SpendableOutputs { outputs, channel_id } => {
 				match self.output_sweeper.track_spendable_outputs(outputs, channel_id, true, None) {
@@ -1441,10 +1438,7 @@ where
 						}
 					}
 				};
-				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.");
-				}
+				self.runtime.spawn(future);
 			},
 			LdkEvent::BumpTransaction(bte) => {
 				match bte {

src/gossip.rs

@@ -7,8 +7,8 @@
 
 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::logger::{log_trace, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{GossipSync, Graph, P2PGossipSync, PeerManager, RapidGossipSync, UtxoLookup};
 use crate::Error;
 
@@ -22,7 +22,6 @@ use std::time::Duration;
 pub(crate) enum GossipSource {
 	P2PNetwork {
 		gossip_sync: Arc<P2PGossipSync>,
-		logger: Arc<Logger>,
 	},
 	RapidGossipSync {
 		gossip_sync: Arc<RapidGossipSync>,
@@ -39,7 +38,7 @@ impl GossipSource {
 			None::<Arc<UtxoLookup>>,
 			Arc::clone(&logger),
 		));
-		Self::P2PNetwork { gossip_sync, logger }
+		Self::P2PNetwork { gossip_sync }
 	}
 
 	pub fn new_rgs(
@@ -67,9 +66,9 @@ impl GossipSource {
 		runtime: Arc<Runtime>,
 	) {
 		match self {
-			Self::P2PNetwork { gossip_sync, logger } => {
+			Self::P2PNetwork { gossip_sync } => {
 				if let Some(utxo_source) = chain_source.as_utxo_source() {
-					let spawner = RuntimeSpawner::new(Arc::clone(&runtime), Arc::clone(&logger));
+					let spawner = RuntimeSpawner::new(Arc::clone(&runtime));
 					let gossip_verifier = Arc::new(GossipVerifier::new(
 						utxo_source,
 						spawner,
@@ -135,20 +134,16 @@ impl GossipSource {
 
 pub(crate) struct RuntimeSpawner {
 	runtime: Arc<Runtime>,
-	logger: Arc<Logger>,
 }
 
 impl RuntimeSpawner {
-	pub(crate) fn new(runtime: Arc<Runtime>, logger: Arc<Logger>) -> Self {
-		Self { runtime, logger }
+	pub(crate) fn new(runtime: Arc<Runtime>) -> Self {
+		Self { runtime }
 	}
 }
 
 impl FutureSpawner for RuntimeSpawner {
 	fn spawn<T: Future<Output = ()> + Send + 'static>(&self, future: T) {
-		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.");
-		}
+		self.runtime.spawn(future);
 	}
 }