Merge pull request fedimint#6229 from tvolk131/ldk-node_runtime

fix: ldk gateway node relies on global runtime

fedimint-testing/src/fixtures.rs

@@ -182,12 +182,11 @@ impl Fixtures {
 
         let lightning_builder: Arc<dyn LightningBuilder + Send + Sync> =
             Arc::new(FakeLightningBuilder);
+        // Note: This runtime isn't used by `FakeLightningBuilder`. It is immediately
+        // dropped.
+        let runtime = Arc::new(tokio::runtime::Runtime::new().unwrap());
+        let ln_client: Arc<dyn ILnRpcClient> = lightning_builder.build(runtime).await.into();
 
-        // Module tests do not use the webserver, so any port is ok
-        let listen: SocketAddr = "127.0.0.1:9000".parse().unwrap();
-        let address: SafeUrl = format!("http://{listen}").parse().unwrap();
-
-        let ln_client: Arc<dyn ILnRpcClient> = lightning_builder.build().await.into();
         let (lightning_public_key, lightning_alias, lightning_network, _, _) = ln_client
             .parsed_node_info()
             .await
@@ -199,6 +198,10 @@ impl Fixtures {
             lightning_network,
         };
 
+        // Module tests do not use the webserver, so any port is ok
+        let listen: SocketAddr = "127.0.0.1:9000".parse().unwrap();
+        let address: SafeUrl = format!("http://{listen}").parse().unwrap();
+
         Gateway::new_with_custom_registry(
             lightning_builder,
             client_builder,

fedimint-testing/src/gateway.rs

@@ -1,5 +1,6 @@
 use std::fmt::{Display, Formatter};
 use std::str::FromStr;
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use ln_gateway::lightning::{ILnRpcClient, LightningBuilder};
@@ -43,7 +44,10 @@ pub struct FakeLightningBuilder;
 
 #[async_trait]
 impl LightningBuilder for FakeLightningBuilder {
-    async fn build(&self) -> Box<dyn ILnRpcClient> {
+    async fn build(&self, runtime: Arc<tokio::runtime::Runtime>) -> Box<dyn ILnRpcClient> {
+        // Runtimes must be dropped in a blocking context. Removing this line can lead
+        // to panics.
+        fedimint_core::runtime::block_in_place(|| drop(runtime));
         Box::new(FakeLightningTest::new())
     }
 }

gateway/ln-gateway/src/bin/gatewayd.rs

@@ -8,6 +8,8 @@
 //! clients to request routing of payments through the Lightning Network.
 //! The API also has endpoints for managing the gateway.
 
+use std::sync::Arc;
+
 use fedimint_core::fedimint_build_code_version_env;
 use fedimint_core::task::TaskGroup;
 use fedimint_core::util::handle_version_hash_command;
@@ -22,16 +24,18 @@ use tracing::info;
 // rocksdb suffers from memory fragmentation when using standard allocator
 static GLOBAL: Jemalloc = Jemalloc;
 
-#[tokio::main]
-async fn main() -> Result<(), anyhow::Error> {
-    handle_version_hash_command(fedimint_build_code_version_env!());
-    TracingSetup::default().init()?;
-    let tg = TaskGroup::new();
-    tg.install_kill_handler();
-    let gatewayd = Gateway::new_with_default_modules().await?;
-    let shutdown_receiver = gatewayd.clone().run(tg).await?;
-    shutdown_receiver.await;
-    gatewayd.unannounce_from_all_federations().await;
-    info!("Gatewayd exiting...");
-    Ok(())
+fn main() -> Result<(), anyhow::Error> {
+    let runtime = Arc::new(tokio::runtime::Runtime::new()?);
+    runtime.block_on(async {
+        handle_version_hash_command(fedimint_build_code_version_env!());
+        TracingSetup::default().init()?;
+        let tg = TaskGroup::new();
+        tg.install_kill_handler();
+        let gatewayd = Gateway::new_with_default_modules().await?;
+        let shutdown_receiver = gatewayd.clone().run(tg, runtime.clone()).await?;
+        shutdown_receiver.await;
+        gatewayd.unannounce_from_all_federations().await;
+        info!("Gatewayd exiting...");
+        Ok(())
+    })
 }

gateway/ln-gateway/src/lib.rs

@@ -413,10 +413,14 @@ impl Gateway {
     /// timer, loads the federation clients from the persisted config,
     /// begins listening for intercepted payments, and starts the webserver
     /// to service requests.
-    pub async fn run(self, tg: TaskGroup) -> anyhow::Result<TaskShutdownToken> {
+    pub async fn run(
+        self,
+        tg: TaskGroup,
+        runtime: Arc<tokio::runtime::Runtime>,
+    ) -> anyhow::Result<TaskShutdownToken> {
         self.register_clients_timer(&tg);
         self.load_clients().await?;
-        self.start_gateway(&tg);
+        self.start_gateway(&tg, runtime);
         // start webserver last to avoid handling requests before fully initialized
         let handle = tg.make_handle();
         run_webserver(Arc::new(self), tg).await?;
@@ -426,7 +430,7 @@ impl Gateway {
 
     /// Begins the task for listening for intercepted payments from the
     /// lightning node.
-    fn start_gateway(&self, task_group: &TaskGroup) {
+    fn start_gateway(&self, task_group: &TaskGroup, runtime: Arc<tokio::runtime::Runtime>) {
         const PAYMENT_STREAM_RETRY_SECONDS: u64 = 5;
 
         let self_copy = self.clone();
@@ -442,7 +446,7 @@ impl Gateway {
                     }
 
                     let payment_stream_task_group = tg.make_subgroup();
-                    let lnrpc_route = self_copy.lightning_builder.build().await;
+                    let lnrpc_route = self_copy.lightning_builder.build(runtime.clone()).await;
 
                     debug!("Establishing lightning payment stream...");
                     let (stream, ln_client) = match lnrpc_route.route_htlcs(&payment_stream_task_group).await

gateway/ln-gateway/src/lightning/ldk.rs

@@ -68,6 +68,7 @@ impl GatewayLdkClient {
         network: Network,
         lightning_port: u16,
         mnemonic: Mnemonic,
+        runtime: Arc<tokio::runtime::Runtime>,
     ) -> anyhow::Result<Self> {
         // In devimint, gateways must allow for other gateways to open channels to them.
         // To ensure this works, we must set a node alias to signal to ldk-node that we
@@ -100,9 +101,7 @@ impl GatewayLdkClient {
         node_builder.set_storage_dir_path(data_dir_str.to_string());
 
         let node = Arc::new(node_builder.build()?);
-        // TODO: Call `start_with_runtime()` instead of `start()`.
-        // See https://github.com/fedimint/fedimint/issues/6159
-        node.start().map_err(|e| {
+        node.start_with_runtime(runtime).map_err(|e| {
             error!(?e, "Failed to start LDK Node");
             LightningRpcError::FailedToConnect
         })?;

gateway/ln-gateway/src/lightning/mod.rs

@@ -343,7 +343,8 @@ pub enum LightningMode {
 
 #[async_trait]
 pub trait LightningBuilder {
-    async fn build(&self) -> Box<dyn ILnRpcClient>;
+    async fn build(&self, runtime: Arc<tokio::runtime::Runtime>) -> Box<dyn ILnRpcClient>;
+
     fn lightning_mode(&self) -> Option<LightningMode> {
         None
     }
@@ -359,7 +360,7 @@ pub struct GatewayLightningBuilder {
 
 #[async_trait]
 impl LightningBuilder for GatewayLightningBuilder {
-    async fn build(&self) -> Box<dyn ILnRpcClient> {
+    async fn build(&self, runtime: Arc<tokio::runtime::Runtime>) -> Box<dyn ILnRpcClient> {
         match self.lightning_mode.clone() {
             LightningMode::Cln { cln_extension_addr } => {
                 Box::new(NetworkLnRpcClient::new(cln_extension_addr))
@@ -386,6 +387,7 @@ impl LightningBuilder for GatewayLightningBuilder {
                     network,
                     lightning_port,
                     self.mnemonic.clone(),
+                    runtime,
                 )
                 .unwrap(),
             ),