diff --git a/CHANGELOG.md b/CHANGELOG.md index 492d092d5b..1abfe1ccac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,9 @@ -# 0.7.0-rc.65 (Synonym Fork) +# 0.7.0-rc.66 (Synonym Fork) ## Bug Fixes +- Prevent native SIGABRT crashes when stopping and rebuilding the node by making runtime teardown deterministic. +- Keep exported payment and liquidity handles from calling into a shutting-down runtime, refuse restart while detached work is still live, and stop Electrum confirm gating from blocking or panicking shutdown. - Add keep consumer rules for JNA types UniFFI needs under R8. # 0.7.0-rc.64 (Synonym Fork) diff --git a/Cargo.toml b/Cargo.toml index 6880c298ab..aff8af778d 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ exclude = ["bindings/uniffi-bindgen"] [package] name = "ldk-node" -version = "0.7.0-rc.65" +version = "0.7.0-rc.66" authors = ["Elias Rohrer "] homepage = "https://lightningdevkit.org/" license = "MIT OR Apache-2.0" @@ -62,6 +62,7 @@ base64 = { version = "0.22.1", default-features = false, features = ["std"] } rand = { version = "0.9.2", default-features = false, features = ["std", "thread_rng", "os_rng"] } chrono = { version = "0.4", default-features = false, features = ["clock"] } tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync", "macros" ] } +tokio-util = { version = "0.7.10", default-features = false, features = ["rt"] } esplora-client = { version = "0.12", default-features = false, features = ["tokio", "async-https-rustls"] } electrum-client = { version = "0.24.0", default-features = false, features = ["proxy", "use-rustls-ring"] } libc = "0.2" diff --git a/Package.swift b/Package.swift index 9a5fab81a5..f8cc852f0e 100644 --- a/Package.swift +++ b/Package.swift @@ -3,8 +3,8 @@ import PackageDescription -let tag = "v0.7.0-rc.65" -let checksum = "c9e1b4fd124434a955f75a41044f5d3f2efbd14b1bb49276cc64bc174a43daee" +let tag = "v0.7.0-rc.66" +let checksum = "21ac13bfdc9fdd3099a688bd0053f8b14c74b5957943b2f09624886e65556a8e" let url = "https://github.com/synonymdev/ldk-node/releases/download/\(tag)/LDKNodeFFI.xcframework.zip" let package = Package( diff --git a/bindings/kotlin/ldk-node-android/gradle.properties b/bindings/kotlin/ldk-node-android/gradle.properties index 3011072e17..694c826d8a 100644 --- a/bindings/kotlin/ldk-node-android/gradle.properties +++ b/bindings/kotlin/ldk-node-android/gradle.properties @@ -3,4 +3,4 @@ android.useAndroidX=true android.enableJetifier=true kotlin.code.style=official group=com.synonym -version=0.7.0-rc.65 +version=0.7.0-rc.66 diff --git a/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/arm64-v8a/libldk_node.so b/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/arm64-v8a/libldk_node.so index 73d8879c2c..d6447f3f2e 100755 Binary files a/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/arm64-v8a/libldk_node.so and b/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/arm64-v8a/libldk_node.so differ diff --git a/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/armeabi-v7a/libldk_node.so b/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/armeabi-v7a/libldk_node.so index d692ff4057..5d503a817a 100755 Binary files a/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/armeabi-v7a/libldk_node.so and b/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/armeabi-v7a/libldk_node.so differ diff --git a/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/x86_64/libldk_node.so b/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/x86_64/libldk_node.so index 5497e06e77..a432cf6f4b 100755 Binary files a/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/x86_64/libldk_node.so and b/bindings/kotlin/ldk-node-android/lib/src/main/jniLibs/x86_64/libldk_node.so differ diff --git a/bindings/kotlin/ldk-node-jvm/gradle.properties b/bindings/kotlin/ldk-node-jvm/gradle.properties index 7f5aaf74dc..34dc4dc264 100644 --- a/bindings/kotlin/ldk-node-jvm/gradle.properties +++ b/bindings/kotlin/ldk-node-jvm/gradle.properties @@ -1,4 +1,4 @@ org.gradle.jvmargs=-Xmx1536m kotlin.code.style=official group=com.synonym -version=0.7.0-rc.65 +version=0.7.0-rc.66 diff --git a/bindings/kotlin/ldk-node-jvm/lib/src/main/resources/darwin-aarch64/libldk_node.dylib b/bindings/kotlin/ldk-node-jvm/lib/src/main/resources/darwin-aarch64/libldk_node.dylib index ec9ac932b3..dfdaeae63d 100644 Binary files a/bindings/kotlin/ldk-node-jvm/lib/src/main/resources/darwin-aarch64/libldk_node.dylib and b/bindings/kotlin/ldk-node-jvm/lib/src/main/resources/darwin-aarch64/libldk_node.dylib differ diff --git a/bindings/kotlin/ldk-node-jvm/lib/src/main/resources/darwin-x86-64/libldk_node.dylib b/bindings/kotlin/ldk-node-jvm/lib/src/main/resources/darwin-x86-64/libldk_node.dylib index 209af59479..d91a777c25 100644 Binary files a/bindings/kotlin/ldk-node-jvm/lib/src/main/resources/darwin-x86-64/libldk_node.dylib and b/bindings/kotlin/ldk-node-jvm/lib/src/main/resources/darwin-x86-64/libldk_node.dylib differ diff --git a/bindings/python/pyproject.toml b/bindings/python/pyproject.toml index bdef977dae..93f0f4095b 100644 --- a/bindings/python/pyproject.toml +++ b/bindings/python/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "ldk_node" -version = "0.7.0-rc.65" +version = "0.7.0-rc.66" authors = [ { name="Elias Rohrer", email="dev@tnull.de" }, ] diff --git a/src/builder.rs b/src/builder.rs index 0fcc8d536d..67cb9afec8 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -9,7 +9,7 @@ use std::collections::{HashMap, HashSet}; use std::convert::TryInto; use std::default::Default; use std::path::PathBuf; -use std::sync::atomic::AtomicU32; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64}; use std::sync::{Arc, Mutex, Once, RwLock}; use std::time::SystemTime; use std::{fmt, fs}; @@ -804,12 +804,12 @@ impl NodeBuilder { 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(), Arc::clone(&logger))) + Runtime::with_handle(handle.clone(), Arc::clone(&logger)) } else { - Arc::new(Runtime::new(Arc::clone(&logger)).map_err(|e| { + Runtime::new(Arc::clone(&logger)).map_err(|e| { log_error!(logger, "Failed to setup tokio runtime: {}", e); BuildError::RuntimeSetupFailed - })?) + })? }; let seed_bytes = seed_bytes_from_config( @@ -855,12 +855,12 @@ impl NodeBuilder { 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(), Arc::clone(&logger))) + Runtime::with_handle(handle.clone(), Arc::clone(&logger)) } else { - Arc::new(Runtime::new(Arc::clone(&logger)).map_err(|e| { + Runtime::new(Arc::clone(&logger)).map_err(|e| { log_error!(logger, "Failed to setup tokio runtime: {}", e); BuildError::RuntimeSetupFailed - })?) + })? }; let seed_bytes = seed_bytes_from_config( @@ -1570,7 +1570,7 @@ fn build_additional_wallets( fn apply_channel_data_migration( migration: &ChannelDataMigration, kv_store: &Arc, keys_manager: &K, - logger: &Arc, runtime: &Arc, + logger: &Arc, runtime: &Runtime, ) -> Result<(), BuildError> where K: EntropySource + SignerProvider, @@ -1725,7 +1725,7 @@ fn build_with_store_internal( gossip_source_config: Option<&GossipSourceConfig>, liquidity_source_config: Option<&LiquiditySourceConfig>, pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>, - async_payments_role: Option, seed_bytes: [u8; 64], runtime: Arc, + async_payments_role: Option, seed_bytes: [u8; 64], runtime: Runtime, logger: Arc, kv_store: Arc, channel_data_migration: Option<&ChannelDataMigration>, ) -> Result { @@ -2440,7 +2440,7 @@ fn build_with_store_internal( gossip_source.set_gossip_verifier( Arc::clone(&chain_source), Arc::clone(&peer_manager), - Arc::clone(&runtime), + runtime.control(), ); let connection_manager = @@ -2580,7 +2580,6 @@ fn build_with_store_internal( let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone()); Ok(Node { - runtime, stop_sender, background_processor_stop_sender, config, @@ -2607,12 +2606,16 @@ fn build_with_store_internal( peer_store, rgs_peer_recovery_exclusions: Arc::new(RgsPeerRecoveryExclusions::default()), payment_store, + lifecycle_lock: Mutex::new(()), is_running, + background_processor_failed: Arc::new(AtomicBool::new(false)), + background_processor_generation: Arc::new(AtomicU64::new(0)), node_metrics, om_mailbox, async_payments_role, runtime_sync_intervals: Arc::new(RwLock::new(RuntimeSyncIntervals::default())), local_rgs_timestamp, + runtime, }) } @@ -2918,13 +2921,11 @@ mod tests { (serialized, monitor_key, update_id, seed) } - fn make_test_deps( - seed: &[u8; 32], - ) -> (Arc, LdkKeysManager, Arc, Arc) { + fn make_test_deps(seed: &[u8; 32]) -> (Arc, LdkKeysManager, Arc, Runtime) { let store: Arc = Arc::new(InMemoryStore::new()); let keys_manager = LdkKeysManager::new(seed, 0, 0, false); let logger = Arc::new(Logger::new_log_facade()); - let runtime = Arc::new(Runtime::new(Arc::clone(&logger)).unwrap()); + let runtime = Runtime::new(Arc::clone(&logger)).unwrap(); (store, keys_manager, logger, runtime) } @@ -3112,7 +3113,7 @@ mod tests { let store: Arc = Arc::new(FailingReadStore); let keys_manager = LdkKeysManager::new(&seed, 0, 0, false); let logger = Arc::new(Logger::new_log_facade()); - let runtime = Arc::new(Runtime::new(Arc::clone(&logger)).unwrap()); + let runtime = Runtime::new(Arc::clone(&logger)).unwrap(); // Migration should fail because the store returns an IO error on read // (fail-closed: non-NotFound errors refuse migration). @@ -3234,7 +3235,7 @@ mod tests { let store: Arc = Arc::new(FailingReadStore); let keys_manager = LdkKeysManager::new(&[42u8; 32], 0, 0, false); let logger = Arc::new(Logger::new_log_facade()); - let runtime = Arc::new(Runtime::new(Arc::clone(&logger)).unwrap()); + let runtime = Runtime::new(Arc::clone(&logger)).unwrap(); let migration = ChannelDataMigration { channel_manager: Some(vec![0x01, 0x02, 0x03]), diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs index 4e994f6176..2e8a8cb181 100644 --- a/src/chain/electrum.rs +++ b/src/chain/electrum.rs @@ -6,7 +6,8 @@ // accordance with one or both of these licenses. use std::collections::HashMap; -use std::sync::{Arc, Mutex, RwLock}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, RwLock, Weak}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use bdk_chain::bdk_core::spk_client::{ @@ -166,7 +167,17 @@ impl ElectrumChainSource { } pub(super) fn stop(&self) { - self.electrum_runtime_status.write().unwrap().stop(); + let client = self.electrum_runtime_status.write().unwrap().stop(); + if let Some(client) = client { + client.begin_shutdown(); + } + } + + pub(super) fn begin_shutdown(&self) { + let client = self.electrum_runtime_status.read().unwrap().client(); + if let Some(client) = client { + client.begin_shutdown(); + } } pub(crate) async fn sync_onchain_wallet( @@ -432,7 +443,7 @@ impl ElectrumChainSource { let sync_cman = Arc::clone(&channel_manager); let sync_cmon = Arc::clone(&chain_monitor); let sync_sweeper = Arc::clone(&output_sweeper); - let confirmables = vec![ + let confirmables: Vec> = vec![ sync_cman as Arc, sync_cmon as Arc, sync_sweeper as Arc, @@ -449,7 +460,8 @@ impl ElectrumChainSource { return Err(Error::TxSyncFailed); }; - let res = electrum_client.sync_confirmables(confirmables).await; + let confirmable = electrum_client.wrap_confirmables(&confirmables); + let res = electrum_client.sync_confirmables(vec![confirmable]).await; if let Ok(_) = res { let unix_time_secs_opt = @@ -593,8 +605,11 @@ impl ElectrumRuntimeStatus { Ok(()) } - pub(super) fn stop(&mut self) { - *self = Self::new() + pub(super) fn stop(&mut self) -> Option> { + match core::mem::replace(self, Self::new()) { + Self::Started(client) => Some(client), + Self::Stopped { .. } => None, + } } fn client(&self) -> Option> { @@ -630,6 +645,7 @@ struct ElectrumRuntimeClient { runtime_handle: Handle, config: Arc, logger: Arc, + confirm_gate: Arc, } impl ElectrumRuntimeClient { @@ -670,7 +686,28 @@ impl ElectrumRuntimeClient { }, )?, ); - Ok(Self { electrum_client, bdk_electrum_client, tx_sync, runtime_handle, config, logger }) + Ok(Self { + electrum_client, + bdk_electrum_client, + tx_sync, + runtime_handle, + config, + logger, + confirm_gate: Arc::new(ConfirmGate::new()), + }) + } + + fn begin_shutdown(&self) { + self.confirm_gate.deactivate(); + } + + fn wrap_confirmables( + &self, confirmables: &[Arc], + ) -> Arc { + Arc::new(ShutdownAwareConfirm::new( + Arc::downgrade(&self.confirm_gate), + confirmables.iter().map(Arc::downgrade).collect(), + )) } pub(crate) async fn get_address_balance(&self, address: &bitcoin::Address) -> Option { @@ -723,6 +760,10 @@ impl ElectrumRuntimeClient { Error::TxSyncFailed })?; + if !self.confirm_gate.is_active() { + return Err(Error::TxSyncFailed); + } + log_info!( self.logger, "Sync of Lightning wallet finished in {}ms.", @@ -916,6 +957,91 @@ impl ElectrumRuntimeClient { } } +struct ConfirmGate { + active: AtomicBool, +} + +impl ConfirmGate { + fn new() -> Self { + Self { active: AtomicBool::new(true) } + } + + fn deactivate(&self) { + self.active.store(false, Ordering::Release); + } + + fn is_active(&self) -> bool { + self.active.load(Ordering::Acquire) + } +} + +struct ShutdownAwareConfirm { + gate: Weak, + confirmables: Vec>, +} + +impl ShutdownAwareConfirm { + fn new(gate: Weak, confirmables: Vec>) -> Self { + Self { gate, confirmables } + } + + fn with_confirmables( + &self, inactive_result: T, f: impl FnOnce(&[Arc]) -> T, + ) -> T { + let Some(gate) = self.gate.upgrade() else { + return inactive_result; + }; + if !gate.is_active() { + return inactive_result; + } + + let Some(confirmables): Option>> = + self.confirmables.iter().map(Weak::upgrade).collect() + else { + return inactive_result; + }; + if !gate.is_active() { + return inactive_result; + } + f(&confirmables) + } +} + +impl Confirm for ShutdownAwareConfirm { + fn transactions_confirmed( + &self, header: &bitcoin::block::Header, + txdata: &lightning::chain::transaction::TransactionData<'_>, height: u32, + ) { + self.with_confirmables((), |confirmables| { + for confirmable in confirmables { + confirmable.transactions_confirmed(header, txdata, height); + } + }) + } + + fn transaction_unconfirmed(&self, txid: &Txid) { + self.with_confirmables((), |confirmables| { + for confirmable in confirmables { + confirmable.transaction_unconfirmed(txid); + } + }) + } + + fn best_block_updated(&self, header: &bitcoin::block::Header, height: u32) { + self.with_confirmables((), |confirmables| { + for confirmable in confirmables { + confirmable.best_block_updated(header, height); + } + }) + } + + fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option)> { + self.with_confirmables(Vec::new(), |confirmables| { + confirmables.iter().flat_map(|confirmable| confirmable.get_relevant_txids()).collect() + }) + } +} + impl Filter for ElectrumRuntimeClient { fn register_tx(&self, txid: &Txid, script_pubkey: &Script) { self.tx_sync.register_tx(txid, script_pubkey) @@ -928,15 +1054,100 @@ impl Filter for ElectrumRuntimeClient { #[cfg(test)] mod tests { use std::net::TcpListener; + use std::panic::{catch_unwind, AssertUnwindSafe}; use std::process::Command; - use std::sync::mpsc::sync_channel; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::mpsc::{self, sync_channel}; use std::thread; + use std::time::Instant; + + use bitcoin::blockdata::constants::genesis_block; use super::*; use crate::runtime::Runtime; const RUNTIME_SELF_DROP_CHILD_ENV: &str = "LDK_NODE_ELECTRUM_RUNTIME_SELF_DROP_CHILD"; + struct RecordingConfirm { + calls: AtomicUsize, + relevant_txid: Txid, + } + + impl RecordingConfirm { + fn new(relevant_txid: Txid) -> Self { + Self { calls: AtomicUsize::new(0), relevant_txid } + } + } + + impl Confirm for RecordingConfirm { + fn transactions_confirmed( + &self, _header: &bitcoin::block::Header, + _txdata: &lightning::chain::transaction::TransactionData<'_>, _height: u32, + ) { + self.calls.fetch_add(1, Ordering::AcqRel); + } + + fn transaction_unconfirmed(&self, _txid: &Txid) { + self.calls.fetch_add(1, Ordering::AcqRel); + } + + fn best_block_updated(&self, _header: &bitcoin::block::Header, _height: u32) { + self.calls.fetch_add(1, Ordering::AcqRel); + } + + fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option)> { + vec![(self.relevant_txid, 0, None)] + } + } + + struct BlockingConfirm { + calls: AtomicUsize, + started: Mutex>>, + release: Mutex>, + } + + impl Confirm for BlockingConfirm { + fn transactions_confirmed( + &self, _header: &bitcoin::block::Header, + _txdata: &lightning::chain::transaction::TransactionData<'_>, _height: u32, + ) { + } + + fn transaction_unconfirmed(&self, _txid: &Txid) {} + + fn best_block_updated(&self, _header: &bitcoin::block::Header, _height: u32) { + self.calls.fetch_add(1, Ordering::AcqRel); + if let Some(started) = self.started.lock().unwrap().take() { + started.send(()).unwrap(); + } + self.release.lock().unwrap().recv().unwrap(); + } + + fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option)> { + Vec::new() + } + } + + struct PanickingConfirm; + + impl Confirm for PanickingConfirm { + fn transactions_confirmed( + &self, _header: &bitcoin::block::Header, + _txdata: &lightning::chain::transaction::TransactionData<'_>, _height: u32, + ) { + } + + fn transaction_unconfirmed(&self, _txid: &Txid) {} + + fn best_block_updated(&self, _header: &bitcoin::block::Header, _height: u32) { + panic!("confirm callback"); + } + + fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option)> { + Vec::new() + } + } + #[test] fn inflight_electrum_worker_does_not_own_runtime_lifecycle() { if std::env::var_os(RUNTIME_SELF_DROP_CHILD_ENV).is_some() { @@ -973,7 +1184,7 @@ mod tests { }); let logger = Arc::new(Logger::new_log_facade()); - let runtime = Arc::new(Runtime::new(Arc::clone(&logger)).unwrap()); + let runtime = Runtime::new(Arc::clone(&logger)).unwrap(); let client = Arc::new( ElectrumRuntimeClient::new( server_url, @@ -1009,6 +1220,86 @@ mod tests { thread::sleep(Duration::from_millis(500)); } + #[test] + fn confirm_callbacks_are_ignored_after_shutdown() { + let block = genesis_block(Network::Regtest); + let txid = block.txdata[0].compute_txid(); + let delegate = Arc::new(RecordingConfirm::new(txid)); + let delegate_dyn: Arc = delegate.clone(); + let gate = Arc::new(ConfirmGate::new()); + let confirm = + ShutdownAwareConfirm::new(Arc::downgrade(&gate), vec![Arc::downgrade(&delegate_dyn)]); + + confirm.best_block_updated(&block.header, 0); + assert_eq!(delegate.calls.load(Ordering::Acquire), 1); + assert_eq!(confirm.get_relevant_txids(), vec![(txid, 0, None)]); + + gate.deactivate(); + confirm.transactions_confirmed(&block.header, &[], 0); + confirm.transaction_unconfirmed(&txid); + confirm.best_block_updated(&block.header, 0); + assert_eq!(delegate.calls.load(Ordering::Acquire), 1); + assert!(confirm.get_relevant_txids().is_empty()); + } + + #[test] + fn deactivate_does_not_wait_for_in_flight_confirm() { + let block = genesis_block(Network::Regtest); + let txid = block.txdata[0].compute_txid(); + let (started_sender, started_receiver) = mpsc::sync_channel(1); + let (release_sender, release_receiver) = mpsc::sync_channel(1); + let blocking = Arc::new(BlockingConfirm { + calls: AtomicUsize::new(0), + started: Mutex::new(Some(started_sender)), + release: Mutex::new(release_receiver), + }); + let trailing = Arc::new(RecordingConfirm::new(txid)); + let blocking_dyn: Arc = blocking.clone(); + let trailing_dyn: Arc = trailing.clone(); + let gate = Arc::new(ConfirmGate::new()); + let confirm = Arc::new(ShutdownAwareConfirm::new( + Arc::downgrade(&gate), + vec![Arc::downgrade(&blocking_dyn), Arc::downgrade(&trailing_dyn)], + )); + + let callback = { + let confirm = Arc::clone(&confirm); + thread::spawn(move || confirm.best_block_updated(&block.header, 0)) + }; + started_receiver.recv().unwrap(); + + let start = Instant::now(); + gate.deactivate(); + assert!(start.elapsed() < Duration::from_millis(50)); + + release_sender.send(()).unwrap(); + callback.join().unwrap(); + + assert_eq!(blocking.calls.load(Ordering::Acquire), 1); + assert_eq!(trailing.calls.load(Ordering::Acquire), 1); + confirm.best_block_updated(&block.header, 0); + assert_eq!(blocking.calls.load(Ordering::Acquire), 1); + assert_eq!(trailing.calls.load(Ordering::Acquire), 1); + } + + #[test] + fn confirm_callback_panic_does_not_poison_shutdown() { + let block = genesis_block(Network::Regtest); + let panicking: Arc = Arc::new(PanickingConfirm); + let gate = Arc::new(ConfirmGate::new()); + let confirm = + ShutdownAwareConfirm::new(Arc::downgrade(&gate), vec![Arc::downgrade(&panicking)]); + + let panicked = catch_unwind(AssertUnwindSafe(|| { + confirm.best_block_updated(&block.header, 0); + })); + assert!(panicked.is_err()); + + gate.deactivate(); + assert!(!gate.is_active()); + confirm.best_block_updated(&block.header, 0); + } + #[test] fn additional_full_scan_settings_do_not_change_primary_defaults() { let default_config = ElectrumSyncConfig::default(); diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 07aaf62527..4a20065f59 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -730,6 +730,18 @@ impl ChainSource { } } + pub(crate) fn begin_shutdown(&self) { + match &self.kind { + ChainSourceKind::Electrum(electrum_chain_source) => { + electrum_chain_source.begin_shutdown() + }, + _ => { + // Other chain sources don't leave synchronous callbacks running after their + // driving future is cancelled. + }, + } + } + pub(crate) fn as_utxo_source(&self) -> Option> { match &self.kind { ChainSourceKind::Bitcoind(bitcoind_chain_source) => { diff --git a/src/event.rs b/src/event.rs index 8398ad7a75..0ba8cefab1 100644 --- a/src/event.rs +++ b/src/event.rs @@ -48,7 +48,7 @@ use crate::payment::store::{ PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, }; use crate::peer_store::persist_missing_channel_peers; -use crate::runtime::Runtime; +use crate::runtime::RuntimeControl; use crate::types::{CustomTlvRecord, DynStore, OnionMessenger, PaymentStore, Sweeper, Wallet}; use crate::{ hex_utils, BumpTransactionEventHandler, ChannelManager, Error, Graph, PeerStore, UserChannelId, @@ -1138,7 +1138,7 @@ where liquidity_source: Option>>>, payment_store: Arc, peer_store: Arc>, - runtime: Arc, + runtime: Arc, logger: L, config: Arc, static_invoice_store: Option, @@ -1158,7 +1158,7 @@ where liquidity_source: Option>>>, payment_store: Arc, peer_store: Arc>, static_invoice_store: Option, onion_messenger: Arc, - om_mailbox: Option>, runtime: Arc, logger: L, + om_mailbox: Option>, runtime: Arc, logger: L, config: Arc, ) -> Self { Self { diff --git a/src/gossip.rs b/src/gossip.rs index 04710e5033..6b77eff6e9 100644 --- a/src/gossip.rs +++ b/src/gossip.rs @@ -16,7 +16,7 @@ use lightning_block_sync::gossip::GossipVerifier; use crate::chain::ChainSource; use crate::config::RGS_SYNC_TIMEOUT_SECS; use crate::logger::{log_trace, LdkLogger, Logger}; -use crate::runtime::Runtime; +use crate::runtime::RuntimeControl; use crate::types::{GossipSync, Graph, P2PGossipSync, PeerManager, RapidGossipSync, UtxoLookup}; use crate::Error; @@ -63,7 +63,7 @@ impl GossipSource { pub(crate) fn set_gossip_verifier( &self, chain_source: Arc, peer_manager: Arc, - runtime: Arc, + runtime: Arc, ) { match self { Self::P2PNetwork { gossip_sync } => { @@ -133,11 +133,11 @@ impl GossipSource { } pub(crate) struct RuntimeSpawner { - runtime: Arc, + runtime: Arc, } impl RuntimeSpawner { - pub(crate) fn new(runtime: Arc) -> Self { + pub(crate) fn new(runtime: Arc) -> Self { Self { runtime } } } diff --git a/src/io/vss_store.rs b/src/io/vss_store.rs index 2fd1ab2cae..bca3779455 100644 --- a/src/io/vss_store.rs +++ b/src/io/vss_store.rs @@ -11,7 +11,7 @@ use std::future::Future; #[cfg(test)] use std::panic::RefUnwindSafe; use std::pin::Pin; -use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -37,6 +37,7 @@ use vss_client::util::retry::{ use vss_client::util::storable_builder::{EntropySource, StorableBuilder}; use crate::io::utils::check_namespace_key_validity; +use crate::runtime::StoreRuntime; type CustomRetryPolicy = FilteredRetryPolicy< JitteredRetryPolicy< @@ -73,13 +74,8 @@ pub struct VssStore { // Version counter to ensure that writes are applied in the correct order. It is assumed that read and list // operations aren't sensitive to the order of execution. next_version: AtomicU64, - // A VSS-internal runtime we use to avoid any deadlocks we could hit when waiting on a spawned - // blocking task to finish while the blocked thread had acquired the reactor. In particular, - // this works around a previously-hit case where a concurrent call to - // `PeerManager::process_pending_events` -> `ChannelManager::get_and_clear_pending_msg_events` - // would deadlock when trying to acquire sync `Mutex` locks that are held by the thread - // currently being blocked waiting on the VSS operation to finish. - internal_runtime: Option, + // A VSS-internal runtime that drives VSS I/O independently from the node runtime. + internal_runtime: Option>, } impl VssStore { @@ -88,17 +84,8 @@ impl VssStore { header_provider: Arc, ) -> io::Result { let next_version = AtomicU64::new(1); - let internal_runtime = tokio::runtime::Builder::new_multi_thread() - .enable_all() - .thread_name_fn(|| { - static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0); - let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst); - format!("ldk-node-vss-runtime-{}", id) - }) - .worker_threads(INTERNAL_RUNTIME_WORKERS) - .max_blocking_threads(INTERNAL_RUNTIME_WORKERS) - .build() - .unwrap(); + let internal_runtime = + Arc::new(StoreRuntime::new("ldk-node-vss-runtime", INTERNAL_RUNTIME_WORKERS, "VSS")?); let (data_encryption_key, obfuscation_master_key) = derive_data_encryption_and_obfuscation_keys(&vss_seed); @@ -140,6 +127,10 @@ impl VssStore { Ok(Self { inner, next_version, internal_runtime: Some(internal_runtime) }) } + fn internal_runtime(&self) -> Arc { + Arc::clone(self.internal_runtime.as_ref().expect("VSS runtime must be available")) + } + // Same logic as for the obfuscated keys below, but just for locking, using the plaintext keys fn build_locking_key( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, @@ -185,7 +176,7 @@ impl KVStoreSync for VssStore { .read_internal(&inner.blocking_client, primary_namespace, secondary_namespace, key) .await }; - tokio::task::block_in_place(move || internal_runtime.block_on(fut)) + tokio::task::block_in_place(move || internal_runtime.handle().block_on(fut)) } fn write( @@ -216,7 +207,7 @@ impl KVStoreSync for VssStore { ) .await }; - tokio::task::block_in_place(move || internal_runtime.block_on(fut)) + tokio::task::block_in_place(move || internal_runtime.handle().block_on(fut)) } fn remove( @@ -250,7 +241,7 @@ impl KVStoreSync for VssStore { internal_runtime.spawn(async { fut.await }); Ok(()) } else { - tokio::task::block_in_place(move || internal_runtime.block_on(fut)) + tokio::task::block_in_place(move || internal_runtime.handle().block_on(fut)) } } @@ -268,7 +259,7 @@ impl KVStoreSync for VssStore { .list_internal(&inner.blocking_client, primary_namespace, secondary_namespace) .await }; - tokio::task::block_in_place(move || internal_runtime.block_on(fut)) + tokio::task::block_in_place(move || internal_runtime.handle().block_on(fut)) } } @@ -280,10 +271,16 @@ impl KVStore for VssStore { let secondary_namespace = secondary_namespace.to_string(); let key = key.to_string(); let inner = Arc::clone(&self.inner); + let runtime = self.internal_runtime(); Box::pin(async move { - inner - .read_internal(&inner.async_client, primary_namespace, secondary_namespace, key) - .await + let task = runtime.spawn(async move { + inner + .read_internal(&inner.async_client, primary_namespace, secondary_namespace, key) + .await + }); + task.await.map_err(|e| { + io::Error::new(io::ErrorKind::Other, format!("VSS runtime task failed: {}", e)) + })? }) } fn write( @@ -295,19 +292,25 @@ impl KVStore for VssStore { let secondary_namespace = secondary_namespace.to_string(); let key = key.to_string(); let inner = Arc::clone(&self.inner); + let runtime = self.internal_runtime(); Box::pin(async move { - inner - .write_internal( - &inner.async_client, - inner_lock_ref, - locking_key, - version, - primary_namespace, - secondary_namespace, - key, - buf, - ) - .await + let task = runtime.spawn(async move { + inner + .write_internal( + &inner.async_client, + inner_lock_ref, + locking_key, + version, + primary_namespace, + secondary_namespace, + key, + buf, + ) + .await + }); + task.await.map_err(|e| { + io::Error::new(io::ErrorKind::Other, format!("VSS runtime task failed: {}", e)) + })? }) } fn remove( @@ -319,6 +322,7 @@ impl KVStore for VssStore { let secondary_namespace = secondary_namespace.to_string(); let key = key.to_string(); let inner = Arc::clone(&self.inner); + let runtime = self.internal_runtime(); let fut = async move { inner .remove_internal( @@ -332,12 +336,19 @@ impl KVStore for VssStore { ) .await }; - if lazy { - tokio::task::spawn(async { fut.await }); - Box::pin(async { Ok(()) }) - } else { - Box::pin(async { fut.await }) - } + Box::pin(async move { + if lazy { + runtime.spawn(async move { + let _ = fut.await; + }); + Ok(()) + } else { + let task = runtime.spawn(fut); + task.await.map_err(|e| { + io::Error::new(io::ErrorKind::Other, format!("VSS runtime task failed: {}", e)) + })? + } + }) } fn list( &self, primary_namespace: &str, secondary_namespace: &str, @@ -345,16 +356,27 @@ impl KVStore for VssStore { let primary_namespace = primary_namespace.to_string(); let secondary_namespace = secondary_namespace.to_string(); let inner = Arc::clone(&self.inner); + let runtime = self.internal_runtime(); Box::pin(async move { - inner.list_internal(&inner.async_client, primary_namespace, secondary_namespace).await + let task = runtime.spawn(async move { + inner + .list_internal(&inner.async_client, primary_namespace, secondary_namespace) + .await + }); + task.await.map_err(|e| { + io::Error::new(io::ErrorKind::Other, format!("VSS runtime task failed: {}", e)) + })? }) } } impl Drop for VssStore { fn drop(&mut self) { - let internal_runtime = self.internal_runtime.take(); - tokio::task::block_in_place(move || drop(internal_runtime)); + if let Some(runtime) = self.internal_runtime.take() { + if let Ok(runtime) = Arc::try_unwrap(runtime) { + runtime.shutdown_background(); + } + } } } diff --git a/src/lib.rs b/src/lib.rs index 4d9d5ec529..5b51005963 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -107,7 +107,7 @@ use std::collections::{HashMap, HashSet}; use std::default::Default; use std::net::ToSocketAddrs; use std::ops::Deref; -use std::sync::atomic::AtomicU32; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, RwLock}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -187,7 +187,6 @@ uniffi::include_scaffolding!("ldk_node"); /// /// Needs to be initialized and instantiated through [`Builder::build`]. pub struct Node { - runtime: Arc, stop_sender: tokio::sync::watch::Sender<()>, background_processor_stop_sender: tokio::sync::watch::Sender<()>, config: Arc, @@ -214,13 +213,34 @@ pub struct Node { peer_store: Arc>>, rgs_peer_recovery_exclusions: Arc, payment_store: Arc, + lifecycle_lock: Mutex<()>, is_running: Arc>, + background_processor_failed: Arc, + background_processor_generation: Arc, node_metrics: Arc>, om_mailbox: Option>, async_payments_role: Option, runtime_sync_intervals: Arc>, /// Shared RGS timestamp used by LocalGraphStore to persist the timestamp alongside the graph. local_rgs_timestamp: Arc, + // Keep the runtime owner last so all runtime-dependent node fields are dropped first. + runtime: Runtime, +} + +fn handle_background_processor_failure( + e: &lightning::io::Error, logger: &Arc, is_running: &RwLock, + background_processor_failed: &AtomicBool, runtime: &runtime::RuntimeControl, + chain_source: &ChainSource, generation: u64, current_generation: &AtomicU64, +) { + if generation != current_generation.load(Ordering::Acquire) { + return; + } + log_error!(logger, "Event processing failed: {}", e); + background_processor_failed.store(true, Ordering::Release); + *is_running.write().unwrap() = false; + runtime.close_task_admission(); + chain_source.begin_shutdown(); + runtime.cancel_tracked_tasks(); } #[derive(Default)] @@ -299,11 +319,13 @@ impl Node { /// After this returns, the [`Node`] instance can be controlled via the provided API methods in /// a thread-safe manner. pub fn start(&self) -> Result<(), Error> { - // Acquire a run lock and hold it until we're setup. - let mut is_running_lock = self.is_running.write().unwrap(); - if *is_running_lock { + let _lifecycle_lock = self.lifecycle_lock.lock().unwrap(); + if *self.is_running.read().unwrap() { return Err(Error::AlreadyRunning); } + if self.background_processor_failed.load(Ordering::Acquire) { + return Err(Error::PersistenceFailed); + } log_info!( self.logger, @@ -312,6 +334,14 @@ impl Node { self.config.network ); + if !self.runtime.allow_task_spawns() { + log_error!( + self.logger, + "Refusing to start while previous background work is still running against this node state.", + ); + return Err(Error::AlreadyRunning); + } + // Start up any runtime-dependant chain sources (e.g. Electrum) // Electrum needs execution access without owning the runtime lifecycle. let runtime_handle = self.runtime.handle().clone(); @@ -402,7 +432,7 @@ impl Node { Arc::clone(&self.node_metrics), Arc::clone(&self.kv_store), Arc::clone(&self.logger), - Arc::clone(&self.runtime), + self.runtime.control(), self.stop_sender.subscribe(), ); } @@ -458,7 +488,7 @@ impl Node { let logger = Arc::clone(&listening_logger); let peer_mgr = Arc::clone(&peer_manager_connection_handler); let mut stop_listen = self.stop_sender.subscribe(); - let runtime = Arc::clone(&self.runtime); + let runtime = self.runtime.control(); self.runtime.spawn_cancellable_background_task(async move { loop { tokio::select! { @@ -640,7 +670,7 @@ impl Node { static_invoice_store, Arc::clone(&self.onion_messenger), self.om_mailbox.clone(), - Arc::clone(&self.runtime), + self.runtime.control(), Arc::clone(&self.logger), Arc::clone(&self.config), )); @@ -665,6 +695,14 @@ impl Node { let background_logger = Arc::clone(&self.logger); let background_error_logger = Arc::clone(&self.logger); let background_scorer = Arc::clone(&self.scorer); + let background_is_running = Arc::clone(&self.is_running); + let background_processor_failed = Arc::clone(&self.background_processor_failed); + let background_processor_generation = + self.background_processor_generation.fetch_add(1, Ordering::AcqRel).wrapping_add(1); + let current_background_processor_generation = + Arc::clone(&self.background_processor_generation); + let background_runtime = self.runtime.control(); + let background_chain_source = Arc::clone(&self.chain_source); let stop_bp = self.background_processor_stop_sender.subscribe(); let sleeper_logger = Arc::clone(&self.logger); let sleeper = move |d| { @@ -686,29 +724,39 @@ impl Node { }) }; - self.runtime.spawn_background_processor_task(async move { - process_events_async( - background_persister, - |e| background_event_handler.handle_event(e), - background_chain_mon, - background_chan_man, - Some(background_onion_messenger), - background_gossip_sync, - background_peer_man, - background_liquidity_man_opt, - Some(background_sweeper), - background_logger, - Some(background_scorer), - sleeper, - true, - || Some(SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap()), - ) - .await - .unwrap_or_else(|e| { - log_error!(background_error_logger, "Failed to process events: {}", e); - panic!("Failed to process events"); - }); - }); + self.runtime.spawn_background_processor_task( + async move { + process_events_async( + background_persister, + |e| background_event_handler.handle_event(e), + background_chain_mon, + background_chan_man, + Some(background_onion_messenger), + background_gossip_sync, + background_peer_man, + background_liquidity_man_opt, + Some(background_sweeper), + background_logger, + Some(background_scorer), + sleeper, + true, + || Some(SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap()), + ) + .await + }, + move |e| { + handle_background_processor_failure( + e, + &background_error_logger, + &background_is_running, + &background_processor_failed, + &background_runtime, + &background_chain_source, + background_processor_generation, + ¤t_background_processor_generation, + ); + }, + ); if let Some(liquidity_source) = self.liquidity_source.as_ref() { let mut stop_liquidity_handler = self.stop_sender.subscribe(); @@ -731,7 +779,11 @@ impl Node { } log_info!(self.logger, "Startup complete."); - *is_running_lock = true; + *self.is_running.write().unwrap() = true; + if self.background_processor_failed.load(Ordering::Acquire) { + *self.is_running.write().unwrap() = false; + return Err(Error::PersistenceFailed); + } Ok(()) } @@ -765,13 +817,24 @@ impl Node { /// /// After this returns most API methods will return [`Error::NotRunning`]. pub fn stop(&self) -> Result<(), Error> { - let mut is_running_lock = self.is_running.write().unwrap(); - if !*is_running_lock { + let _lifecycle_lock = self.lifecycle_lock.lock().unwrap(); + if !*self.is_running.read().unwrap() + && !self.background_processor_failed.load(Ordering::Acquire) + { return Err(Error::NotRunning); } + *self.is_running.write().unwrap() = false; + self.background_processor_generation.fetch_add(1, Ordering::AcqRel); log_info!(self.logger, "Shutting down LDK Node with node ID {}...", self.node_id()); + // Prevent all task groups from accepting work that could outlive this shutdown generation. + self.runtime.close_task_admission(); + + // Prevent blocking Electrum syncs from making any further callbacks before persistence + // tasks stop accepting work. + self.chain_source.begin_shutdown(); + // Stop background tasks. self.stop_sender .send(()) @@ -784,7 +847,6 @@ impl Node { "Failed to send shutdown signal. This should never happen: {}", e ); - debug_assert!(false); }); // Cancel cancellable background tasks @@ -813,18 +875,25 @@ impl Node { "Failed to send shutdown signal. This should never happen: {}", e ); - debug_assert!(false); }); // Finally, wait until background processing stopped, at least until a timeout is reached. - self.runtime.wait_on_background_processor_task(); + let background_processor_result = self.runtime.wait_on_background_processor_task(); #[cfg(tokio_unstable)] self.runtime.log_metrics(); log_info!(self.logger, "Shutdown complete."); - *is_running_lock = false; - Ok(()) + let background_processor_failed = + self.background_processor_failed.swap(false, Ordering::AcqRel); + match background_processor_result { + Err(e) => { + log_error!(self.logger, "Failed to process events during shutdown: {}", e); + Err(Error::PersistenceFailed) + }, + Ok(()) if background_processor_failed => Err(Error::PersistenceFailed), + Ok(()) => Ok(()), + } } /// Returns the status of the [`Node`]. @@ -950,7 +1019,7 @@ impl Node { #[cfg(not(feature = "uniffi"))] pub fn bolt11_payment(&self) -> Bolt11Payment { Bolt11Payment::new( - Arc::clone(&self.runtime), + self.runtime.control(), Arc::clone(&self.channel_manager), Arc::clone(&self.connection_manager), self.liquidity_source.clone(), @@ -969,7 +1038,7 @@ impl Node { #[cfg(feature = "uniffi")] pub fn bolt11_payment(&self) -> Arc { Arc::new(Bolt11Payment::new( - Arc::clone(&self.runtime), + self.runtime.control(), Arc::clone(&self.channel_manager), Arc::clone(&self.connection_manager), self.liquidity_source.clone(), @@ -1136,10 +1205,11 @@ impl Node { #[cfg(not(feature = "uniffi"))] pub fn lsps1_liquidity(&self) -> LSPS1Liquidity { LSPS1Liquidity::new( - Arc::clone(&self.runtime), + self.runtime.control(), Arc::clone(&self.wallet), Arc::clone(&self.connection_manager), self.liquidity_source.clone(), + Arc::clone(&self.is_running), Arc::clone(&self.logger), ) } @@ -1150,10 +1220,11 @@ impl Node { #[cfg(feature = "uniffi")] pub fn lsps1_liquidity(&self) -> Arc { Arc::new(LSPS1Liquidity::new( - Arc::clone(&self.runtime), + self.runtime.control(), Arc::clone(&self.wallet), Arc::clone(&self.connection_manager), self.liquidity_source.clone(), + Arc::clone(&self.is_running), Arc::clone(&self.logger), )) } @@ -2505,6 +2576,51 @@ mod tests { } } + #[test] + fn background_processor_failure_marks_node_unhealthy() { + let config = Config { network: Network::Regtest, ..Config::default() }; + let mut builder = NodeBuilder::from_config(config); + builder.set_chain_source_esplora("http://127.0.0.1:1".to_string(), None); + builder.set_entropy_seed_bytes([42u8; 64]); + builder.set_log_facade_logger(); + let node = builder.build().unwrap(); + *node.is_running.write().unwrap() = true; + node.background_processor_generation.store(1, Ordering::Release); + let error = io::Error::new(io::ErrorKind::Other, "persistence failed"); + + handle_background_processor_failure( + &error, + &node.logger, + &node.is_running, + &node.background_processor_failed, + &node.runtime, + &node.chain_source, + 1, + &node.background_processor_generation, + ); + + assert!(!node.status().is_running); + assert!(node.background_processor_failed.load(Ordering::Acquire)); + assert!(matches!(node.start(), Err(Error::PersistenceFailed))); + assert!(matches!(node.stop(), Err(Error::PersistenceFailed))); + assert!(!node.background_processor_failed.load(Ordering::Acquire)); + + *node.is_running.write().unwrap() = true; + handle_background_processor_failure( + &error, + &node.logger, + &node.is_running, + &node.background_processor_failed, + &node.runtime, + &node.chain_source, + 1, + &node.background_processor_generation, + ); + assert!(node.status().is_running); + assert!(!node.background_processor_failed.load(Ordering::Acquire)); + *node.is_running.write().unwrap() = false; + } + #[test] fn disconnect_propagates_peer_removal_persistence_failure() { let config = Config { network: Network::Regtest, ..Config::default() }; diff --git a/src/liquidity.rs b/src/liquidity.rs index 24d3f3d218..4212997668 100644 --- a/src/liquidity.rs +++ b/src/liquidity.rs @@ -43,7 +43,7 @@ use crate::builder::BuildError; use crate::chain::ChainSource; use crate::connection::ConnectionManager; use crate::logger::{log_debug, log_error, log_info, LdkLogger, Logger}; -use crate::runtime::Runtime; +use crate::runtime::RuntimeControl; use crate::types::{ Broadcaster, ChannelManager, DynStore, KeysManager, LiquidityManager, PeerManager, Wallet, }; @@ -1457,20 +1457,22 @@ pub(crate) struct LSPS2BuyResponse { /// [`Bolt11Payment::receive_via_jit_channel`]: crate::payment::Bolt11Payment::receive_via_jit_channel #[derive(Clone)] pub struct LSPS1Liquidity { - runtime: Arc, + runtime: Arc, wallet: Arc, connection_manager: Arc>>, liquidity_source: Option>>>, + is_running: Arc>, logger: Arc, } impl LSPS1Liquidity { pub(crate) fn new( - runtime: Arc, wallet: Arc, + runtime: Arc, wallet: Arc, connection_manager: Arc>>, - liquidity_source: Option>>>, logger: Arc, + liquidity_source: Option>>>, is_running: Arc>, + logger: Arc, ) -> Self { - Self { runtime, wallet, connection_manager, liquidity_source, logger } + Self { runtime, wallet, connection_manager, liquidity_source, is_running, logger } } /// Connects to the configured LSP and places an order for an inbound channel. @@ -1481,6 +1483,9 @@ impl LSPS1Liquidity { &self, lsp_balance_sat: u64, client_balance_sat: u64, channel_expiry_blocks: u32, announce_channel: bool, ) -> Result { + if !*self.is_running.read().unwrap() { + return Err(Error::NotRunning); + } let liquidity_source = self.liquidity_source.as_ref().ok_or(Error::LiquiditySourceUnavailable)?; @@ -1493,16 +1498,16 @@ impl LSPS1Liquidity { // We need to use our main runtime here as a local runtime might not be around to poll // connection futures going forward. - self.runtime.block_on(async move { + self.runtime.try_block_on(async move { con_cm.connect_peer_if_necessary(con_node_id, con_addr).await - })?; + })??; log_info!(self.logger, "Connected to LSP {}@{}. ", lsp_node_id, lsp_address); let refund_address = self.wallet.get_new_address()?; let liquidity_source = Arc::clone(&liquidity_source); - let response = self.runtime.block_on(async move { + let response = self.runtime.try_block_on(async move { liquidity_source .lsps1_request_channel( lsp_balance_sat, @@ -1512,13 +1517,16 @@ impl LSPS1Liquidity { refund_address, ) .await - })?; + })??; Ok(response) } /// Connects to the configured LSP and checks for the status of a previously-placed order. pub fn check_order_status(&self, order_id: LSPS1OrderId) -> Result { + if !*self.is_running.read().unwrap() { + return Err(Error::NotRunning); + } let liquidity_source = self.liquidity_source.as_ref().ok_or(Error::LiquiditySourceUnavailable)?; @@ -1531,14 +1539,14 @@ impl LSPS1Liquidity { // We need to use our main runtime here as a local runtime might not be around to poll // connection futures going forward. - self.runtime.block_on(async move { + self.runtime.try_block_on(async move { con_cm.connect_peer_if_necessary(con_node_id, con_addr).await - })?; + })??; let liquidity_source = Arc::clone(&liquidity_source); - let response = self - .runtime - .block_on(async move { liquidity_source.lsps1_check_order_status(order_id).await })?; + let response = self.runtime.try_block_on(async move { + liquidity_source.lsps1_check_order_status(order_id).await + })??; Ok(response) } } diff --git a/src/payment/bolt11.rs b/src/payment/bolt11.rs index 3670afc1c7..0e57449584 100644 --- a/src/payment/bolt11.rs +++ b/src/payment/bolt11.rs @@ -36,7 +36,7 @@ use crate::payment::store::{ PaymentStatus, }; use crate::peer_store::{PeerInfo, PeerStore}; -use crate::runtime::Runtime; +use crate::runtime::RuntimeControl; use crate::types::{ChannelManager, PaymentStore, Router}; use crate::ProbeHandle; #[cfg(not(feature = "uniffi"))] @@ -56,7 +56,7 @@ type Bolt11InvoiceDescription = crate::ffi::Bolt11InvoiceDescription; /// [BOLT 11]: https://github.com/lightning/bolts/blob/master/11-payment-encoding.md /// [`Node::bolt11_payment`]: crate::Node::bolt11_payment pub struct Bolt11Payment { - runtime: Arc, + runtime: Arc, channel_manager: Arc, connection_manager: Arc>>, liquidity_source: Option>>>, @@ -70,7 +70,7 @@ pub struct Bolt11Payment { impl Bolt11Payment { pub(crate) fn new( - runtime: Arc, channel_manager: Arc, + runtime: Arc, channel_manager: Arc, connection_manager: Arc>>, liquidity_source: Option>>>, payment_store: Arc, peer_store: Arc>>, @@ -728,6 +728,10 @@ impl Bolt11Payment { expiry_secs: u32, max_total_lsp_fee_limit_msat: Option, max_proportional_lsp_fee_limit_ppm_msat: Option, payment_hash: Option, ) -> Result { + if !*self.is_running.read().unwrap() { + return Err(Error::NotRunning); + } + let liquidity_source = self.liquidity_source.as_ref().ok_or(Error::LiquiditySourceUnavailable)?; @@ -742,15 +746,15 @@ impl Bolt11Payment { // We need to use our main runtime here as a local runtime might not be around to poll // connection futures going forward. - self.runtime.block_on(async move { + self.runtime.try_block_on(async move { con_cm.connect_peer_if_necessary(con_node_id, con_addr).await - })?; + })??; log_info!(self.logger, "Connected to LSP {}@{}. ", peer_info.node_id, peer_info.address); let liquidity_source = Arc::clone(&liquidity_source); let (invoice, lsp_total_opening_fee, lsp_prop_opening_fee) = - self.runtime.block_on(async move { + self.runtime.try_block_on(async move { if let Some(amount_msat) = amount_msat { liquidity_source .lsps2_receive_to_jit_channel( @@ -773,7 +777,7 @@ impl Bolt11Payment { .await .map(|(invoice, prop_fee)| (invoice, None, Some(prop_fee))) } - })?; + })??; // Register payment in payment store. let payment_hash = PaymentHash(invoice.payment_hash().to_byte_array()); @@ -814,7 +818,7 @@ impl Bolt11Payment { self.payment_store.insert(payment)?; // Persist LSP peer to make sure we reconnect on restart. - self.runtime.block_on(self.peer_store.add_peer(peer_info))?; + self.runtime.try_block_on(self.peer_store.add_peer(peer_info))??; Ok(invoice) } diff --git a/src/runtime.rs b/src/runtime.rs index d00734a31f..0fa4cf1d7c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,231 +1,900 @@ // This file is Copyright its original authors, visible in version control history. // -// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in // accordance with one or both of these licenses. use std::future::Future; -use std::sync::{Arc, Mutex}; +use std::io; +use std::ops::Deref; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; use std::time::Duration; -use tokio::task::{JoinHandle, JoinSet}; +use lightning::io::{Error as LdkIoError, ErrorKind as LdkIoErrorKind}; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; use crate::config::{ BACKGROUND_TASK_SHUTDOWN_TIMEOUT_SECS, LDK_EVENT_HANDLER_SHUTDOWN_TIMEOUT_SECS, }; +use crate::error::Error; use crate::logger::{log_debug, log_error, log_trace, LdkLogger, Logger}; +fn mutex_lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn rwlock_read(lock: &RwLock) -> std::sync::RwLockReadGuard<'_, T> { + lock.read().unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn rwlock_write(lock: &RwLock) -> std::sync::RwLockWriteGuard<'_, T> { + lock.write().unwrap_or_else(std::sync::PoisonError::into_inner) +} + +/// Owns the Tokio runtime when ldk-node creates one. +/// +/// This type is intentionally non-cloneable and is held only by `Node`. All spawned work and +/// exported child handles receive [`RuntimeControl`], which cannot keep an owned runtime alive. pub(crate) struct Runtime { - mode: RuntimeMode, - background_tasks: Mutex>, - cancellable_background_tasks: Mutex>, - background_processor_task: Mutex>>, - logger: Arc, + owned_runtime: Option, + control: Arc, } impl Runtime { - pub fn new(logger: Arc) -> Result { - let mode = match tokio::runtime::Handle::try_current() { - Ok(handle) => RuntimeMode::Handle(handle), + pub fn new(logger: Arc) -> io::Result { + match tokio::runtime::Handle::try_current() { + Ok(handle) => Ok(Self::with_handle(handle, logger)), Err(_) => { - let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build()?; - RuntimeMode::Owned(rt) + let mut runtime_builder = tokio::runtime::Builder::new_multi_thread(); + runtime_builder.enable_all(); + runtime_builder.thread_name_fn(|| { + static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0); + let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst); + format!("ldk-node-runtime-{}", id) + }); + let runtime = runtime_builder.build()?; + Ok(Self::with_owned_runtime(runtime, logger)) }, - }; - let background_tasks = Mutex::new(JoinSet::new()); - let cancellable_background_tasks = Mutex::new(JoinSet::new()); - let background_processor_task = Mutex::new(None); - - Ok(Self { - mode, - background_tasks, - cancellable_background_tasks, - background_processor_task, - logger, - }) + } } pub fn with_handle(handle: tokio::runtime::Handle, logger: Arc) -> Self { - let mode = RuntimeMode::Handle(handle); - let background_tasks = Mutex::new(JoinSet::new()); - let cancellable_background_tasks = Mutex::new(JoinSet::new()); - let background_processor_task = Mutex::new(None); + Self { owned_runtime: None, control: Arc::new(RuntimeControl::new(handle, logger)) } + } + + fn with_owned_runtime(runtime: tokio::runtime::Runtime, logger: Arc) -> Self { + let control = Arc::new(RuntimeControl::new(runtime.handle().clone(), logger)); + Self { owned_runtime: Some(runtime), control } + } + + pub(crate) fn control(&self) -> Arc { + Arc::clone(&self.control) + } +} + +impl Deref for Runtime { + type Target = RuntimeControl; + + fn deref(&self) -> &Self::Target { + &self.control + } +} + +impl Drop for Runtime { + fn drop(&mut self) { + self.control.cancel_all_tasks(); + let _ = self + .control + .drain_detached_work(Duration::from_secs(BACKGROUND_TASK_SHUTDOWN_TIMEOUT_SECS)); + self.control.close_runtime_use(); + if let Some(runtime) = self.owned_runtime.take() { + runtime.shutdown_background(); + } + } +} + +/// Cloneable runtime access that never owns the Tokio runtime. +pub(crate) struct RuntimeControl { + handle: tokio::runtime::Handle, + runtime_use: RwLock, + background_tasks: Mutex, + cancellable_background_tasks: Mutex, + background_processor_task: Mutex>>>, + leftover_task_trackers: Mutex>, + leftover_processor_task: Mutex>>>, + logger: Arc, +} + +struct TrackedTasks { + tasks: TaskTracker, + cancellation_token: CancellationToken, +} + +impl TrackedTasks { + fn new() -> Self { + Self { tasks: TaskTracker::new(), cancellation_token: CancellationToken::new() } + } + fn close(&self) { + self.tasks.close(); + } + + fn cancel(&self) { + self.cancellation_token.cancel(); + } +} + +impl Drop for TrackedTasks { + fn drop(&mut self) { + self.close(); + self.cancel(); + } +} + +impl RuntimeControl { + fn new(handle: tokio::runtime::Handle, logger: Arc) -> Self { Self { - mode, - background_tasks, - cancellable_background_tasks, - background_processor_task, + handle, + runtime_use: RwLock::new(true), + background_tasks: Mutex::new(TrackedTasks::new()), + cancellable_background_tasks: Mutex::new(TrackedTasks::new()), + background_processor_task: Mutex::new(None), + leftover_task_trackers: Mutex::new(Vec::new()), + leftover_processor_task: Mutex::new(None), logger, } } + pub fn allow_task_spawns(&self) -> bool { + self.allow_task_spawns_with_timeout(Duration::from_secs( + BACKGROUND_TASK_SHUTDOWN_TIMEOUT_SECS, + )) + } + + pub(crate) fn allow_task_spawns_with_timeout(&self, timeout: Duration) -> bool { + if !self.drain_detached_work(timeout) { + log_error!( + self.logger, + "Refusing a new task generation while previous work is still running." + ); + return false; + } + Self::reset_closed_task_generation(&self.background_tasks); + Self::reset_closed_task_generation(&self.cancellable_background_tasks); + true + } + + fn reset_closed_task_generation(tasks: &Mutex) { + let mut tasks = mutex_lock(tasks); + if tasks.tasks.is_closed() { + *tasks = TrackedTasks::new(); + } + } + + pub fn close_task_admission(&self) { + mutex_lock(&self.background_tasks).close(); + mutex_lock(&self.cancellable_background_tasks).close(); + } + + fn cancel_all_tasks(&self) { + self.cancel_tracked_tasks(); + if let Some(task) = mutex_lock(&self.background_processor_task).as_ref() { + task.abort(); + } + if let Some(task) = mutex_lock(&self.leftover_processor_task).as_ref() { + task.abort(); + } + } + + pub(crate) fn cancel_tracked_tasks(&self) { + for tasks in [&self.background_tasks, &self.cancellable_background_tasks] { + let tasks = mutex_lock(tasks); + tasks.close(); + tasks.cancel(); + } + } + + fn close_runtime_use(&self) { + *rwlock_write(&self.runtime_use) = false; + } + + fn detach_task_tracker(&self, tasks: TaskTracker) { + if !tasks.is_empty() { + mutex_lock(&self.leftover_task_trackers).push(tasks); + } + } + + fn drain_detached_work(&self, timeout: Duration) -> bool { + let trackers = std::mem::take(&mut *mutex_lock(&self.leftover_task_trackers)); + let mut processor = mutex_lock(&self.leftover_processor_task).take(); + let drained = match self.try_block_on(async { + tokio::time::timeout(timeout, async { + for tracker in &trackers { + tracker.wait().await; + } + if let Some(task) = &mut processor { + let _ = task.await; + } + }) + .await + .is_ok() + }) { + Ok(drained) => drained, + Err(_) => false, + }; + if drained { + true + } else { + mutex_lock(&self.leftover_task_trackers).extend(trackers); + if let Some(task) = processor { + if !task.is_finished() { + *mutex_lock(&self.leftover_processor_task) = Some(task); + } + } + false + } + } + pub fn spawn_background_task(&self, future: F) where F: Future + Send + 'static, { - let mut background_tasks = self.background_tasks.lock().unwrap(); - let runtime_handle = self.handle(); - // Since it seems to make a difference to `tokio` (see - // https://docs.rs/tokio/latest/tokio/time/fn.timeout.html#panics) we make sure the futures - // are always put in an `async` / `.await` closure. - background_tasks.spawn_on(async { future.await }, runtime_handle); + self.spawn_tracked_task(&self.background_tasks, "background", future); } pub fn spawn_cancellable_background_task(&self, future: F) where F: Future + Send + 'static, { - let mut cancellable_background_tasks = self.cancellable_background_tasks.lock().unwrap(); - let runtime_handle = self.handle(); - // Since it seems to make a difference to `tokio` (see - // https://docs.rs/tokio/latest/tokio/time/fn.timeout.html#panics) we make sure the futures - // are always put in an `async` / `.await` closure. - cancellable_background_tasks.spawn_on(async { future.await }, runtime_handle); + self.spawn_tracked_task( + &self.cancellable_background_tasks, + "cancellable background", + future, + ); } - pub fn spawn_background_processor_task(&self, future: F) + fn spawn_tracked_task(&self, tasks: &Mutex, task_name: &str, future: F) where F: Future + Send + 'static, { - let mut background_processor_task = self.background_processor_task.lock().unwrap(); - debug_assert!(background_processor_task.is_none(), "Expected no background processor_task"); + let tasks = mutex_lock(tasks); + if tasks.tasks.is_closed() { + log_trace!(self.logger, "Ignoring {} task spawned during shutdown.", task_name); + return; + } + + let cancellation_token = tasks.cancellation_token.clone(); + let _ = tasks.tasks.spawn_on( + async move { + tokio::select! { + biased; + _ = cancellation_token.cancelled() => {}, + _ = future => {}, + } + }, + &self.handle, + ); + } + + pub fn spawn_background_processor_task(&self, future: F, on_failure: C) + where + F: Future> + Send + 'static, + C: FnOnce(&LdkIoError) + Send + 'static, + { + let mut background_processor_task = mutex_lock(&self.background_processor_task); + debug_assert!(background_processor_task.is_none(), "Expected no background processor task"); + *background_processor_task = Some(self.handle.spawn(async move { + let result = future.await; + if let Err(e) = &result { + on_failure(e); + } + result + })); + } - let runtime_handle = self.handle(); - let handle = runtime_handle.spawn(future); - *background_processor_task = Some(handle); + pub fn try_block_on(&self, future: F) -> Result { + let runtime_available = rwlock_read(&self.runtime_use); + if !*runtime_available { + return Err(Error::NotRunning); + } + let handle = tokio::runtime::Handle::try_current().unwrap_or_else(|_| self.handle.clone()); + Ok(tokio::task::block_in_place(move || handle.block_on(async { future.await }))) } pub fn block_on(&self, future: F) -> F::Output { - // While we generally decided not to overthink via which call graph users would enter our - // runtime context, we'd still try to reuse whatever current context would be present - // during `block_on`, as this is the context `block_in_place` would operate on. So we try - // to detect the outer context here, and otherwise use whatever was set during - // initialization. - let handle = tokio::runtime::Handle::try_current().unwrap_or(self.handle().clone()); - // Since it seems to make a difference to `tokio` (see - // https://docs.rs/tokio/latest/tokio/time/fn.timeout.html#panics) we make sure the futures - // are always put in an `async` / `.await` closure. - tokio::task::block_in_place(move || handle.block_on(async { future.await })) + self.try_block_on(future).expect("owned runtime must outlive owner-driven work") } pub fn abort_cancellable_background_tasks(&self) { - let mut tasks = core::mem::take(&mut *self.cancellable_background_tasks.lock().unwrap()); - debug_assert!(tasks.len() > 0, "Expected some cancellable background_tasks"); - tasks.abort_all(); - self.block_on(async { while let Some(_) = tasks.join_next().await {} }) + self.abort_cancellable_background_tasks_with_timeout(Duration::from_secs( + BACKGROUND_TASK_SHUTDOWN_TIMEOUT_SECS, + )); + } + + fn abort_cancellable_background_tasks_with_timeout(&self, shutdown_timeout: Duration) { + let tasks = { + let tasks = mutex_lock(&self.cancellable_background_tasks); + tasks.close(); + tasks.cancel(); + tasks.tasks.clone() + }; + let timed_out = self.block_on(async { + tokio::time::timeout(shutdown_timeout, tasks.wait()).await.is_err() + }); + if timed_out { + log_error!( + self.logger, + "Detaching cancellable background tasks after cancellation timed out." + ); + self.detach_task_tracker(tasks); + } else { + log_debug!(self.logger, "Stopped all cancellable background tasks."); + } } pub fn wait_on_background_tasks(&self) { - let mut tasks = core::mem::take(&mut *self.background_tasks.lock().unwrap()); - debug_assert!(tasks.len() > 0, "Expected some background_tasks"); - self.block_on(async { - loop { - let timeout_fut = tokio::time::timeout( - Duration::from_secs(BACKGROUND_TASK_SHUTDOWN_TIMEOUT_SECS), - tasks.join_next_with_id(), - ); - match timeout_fut.await { - Ok(Some(Ok((id, _)))) => { - log_trace!(self.logger, "Stopped background task with id {}", id); - }, - Ok(Some(Err(e))) => { - tasks.abort_all(); - log_trace!(self.logger, "Stopping background task failed: {}", e); - break; - }, - Ok(None) => { - log_debug!(self.logger, "Stopped all background tasks"); - break; - }, - Err(e) => { - tasks.abort_all(); - log_error!(self.logger, "Stopping background task timed out: {}", e); - break; - }, - } - } - }) - } - - pub fn wait_on_background_processor_task(&self) { - if let Some(background_processor_task) = - self.background_processor_task.lock().unwrap().take() - { - let abort_handle = background_processor_task.abort_handle(); - // Since it seems to make a difference to `tokio` (see - // https://docs.rs/tokio/latest/tokio/time/fn.timeout.html#panics) we make sure the futures - // are always put in an `async` / `.await` closure. - let timeout_res = self.block_on(async { - tokio::time::timeout( - Duration::from_secs(LDK_EVENT_HANDLER_SHUTDOWN_TIMEOUT_SECS), - background_processor_task, - ) - .await + self.wait_on_background_tasks_with_timeout(Duration::from_secs( + BACKGROUND_TASK_SHUTDOWN_TIMEOUT_SECS, + )); + } + + fn wait_on_background_tasks_with_timeout(&self, shutdown_timeout: Duration) { + let (tasks, cancellation_token) = { + let tasks = mutex_lock(&self.background_tasks); + tasks.close(); + (tasks.tasks.clone(), tasks.cancellation_token.clone()) + }; + + let timed_out = self.block_on(async { + tokio::time::timeout(shutdown_timeout, tasks.wait()).await.is_err() + }); + if timed_out { + log_error!(self.logger, "Stopping background tasks timed out."); + cancellation_token.cancel(); + let cancellation_timed_out = self.block_on(async { + tokio::time::timeout(shutdown_timeout, tasks.wait()).await.is_err() }); + if cancellation_timed_out { + log_error!(self.logger, "Detaching background tasks after cancellation timed out."); + self.detach_task_tracker(tasks); + return; + } + } + log_debug!(self.logger, "Stopped all background tasks."); + } - match timeout_res { - Ok(stop_res) => match stop_res { - Ok(()) => log_debug!(self.logger, "Stopped background processing of events."), - Err(e) => { - abort_handle.abort(); - log_error!( - self.logger, - "Stopping event handling failed. This should never happen: {}", - e - ); - panic!("Stopping event handling failed. This should never happen."); - }, - }, + pub fn wait_on_background_processor_task(&self) -> Result<(), LdkIoError> { + self.wait_on_background_processor_task_with_timeout(Duration::from_secs( + LDK_EVENT_HANDLER_SHUTDOWN_TIMEOUT_SECS, + )) + } + + fn wait_on_background_processor_task_with_timeout( + &self, shutdown_timeout: Duration, + ) -> Result<(), LdkIoError> { + let Some(mut task) = mutex_lock(&self.background_processor_task).take() else { + log_error!(self.logger, "Skipped waiting for missing background processor task."); + return Ok(()); + }; + + enum ProcessorWait { + Finished(Result<(), LdkIoError>), + Detached, + } + + let outcome = self.block_on(async { + match tokio::time::timeout(shutdown_timeout, &mut task).await { + Ok(Ok(result)) => ProcessorWait::Finished(result), + Ok(Err(e)) => ProcessorWait::Finished(Err(LdkIoError::new( + LdkIoErrorKind::Other, + format!("Event processor task failed: {}", e), + ))), Err(e) => { - abort_handle.abort(); log_error!(self.logger, "Stopping event handling timed out: {}", e); + task.abort(); + match tokio::time::timeout(shutdown_timeout, &mut task).await { + Err(e) => { + log_error!( + self.logger, + "Detaching event processor after cancellation timed out: {}", + e + ); + ProcessorWait::Detached + }, + Ok(Ok(result)) => ProcessorWait::Finished(result), + Ok(Err(e)) if e.is_cancelled() => ProcessorWait::Finished(Ok(())), + Ok(Err(e)) => ProcessorWait::Finished(Err(LdkIoError::new( + LdkIoErrorKind::Other, + format!("Event processor task failed: {}", e), + ))), + } }, } - } else { - debug_assert!(false, "Expected a background processing task"); + }); + + let result = match outcome { + ProcessorWait::Finished(result) => result, + ProcessorWait::Detached => { + *mutex_lock(&self.leftover_processor_task) = Some(task); + Ok(()) + }, }; + + if result.is_ok() { + log_debug!(self.logger, "Stopped background processing of events."); + } + result } #[cfg(tokio_unstable)] pub fn log_metrics(&self) { - let runtime_handle = self.handle(); log_trace!( self.logger, "Active runtime tasks left prior to shutdown: {}", - runtime_handle.metrics().active_tasks_count() + self.handle.metrics().active_tasks_count() ); } pub(crate) fn handle(&self) -> &tokio::runtime::Handle { - match &self.mode { - RuntimeMode::Owned(rt) => rt.handle(), - RuntimeMode::Handle(handle) => handle, + &self.handle + } +} + +impl Drop for RuntimeControl { + fn drop(&mut self) { + if let Ok(task) = self.background_processor_task.get_mut() { + if let Some(task) = task.take() { + task.abort(); + } } } } -enum RuntimeMode { - Owned(tokio::runtime::Runtime), - Handle(tokio::runtime::Handle), +/// Runtime used by async store backends while ldk-node still exposes synchronous APIs. +/// +/// Store I/O uses an independent runtime so synchronous node APIs cannot block the worker that +/// must drive the persistence future they are waiting for. +pub(crate) struct StoreRuntime { + runtime: Option, +} + +impl StoreRuntime { + pub(crate) fn new( + thread_name_prefix: &'static str, worker_threads: usize, runtime_name: &'static str, + ) -> io::Result { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .thread_name_fn(move || { + static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0); + let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst); + format!("{}-{}", thread_name_prefix, id) + }) + .worker_threads(worker_threads) + .max_blocking_threads(worker_threads) + .build() + .map_err(|e| { + io::Error::new( + io::ErrorKind::Other, + format!("Failed to build {runtime_name} runtime: {e}"), + ) + })?; + Ok(Self { runtime: Some(runtime) }) + } + + pub(crate) fn handle(&self) -> &tokio::runtime::Handle { + self.runtime.as_ref().expect("store runtime must be available").handle() + } + + pub(crate) fn spawn(&self, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + self.handle().spawn(future) + } + + pub(crate) fn shutdown_background(mut self) { + if let Some(runtime) = self.runtime.take() { + runtime.shutdown_background(); + } + } +} + +impl Drop for StoreRuntime { + fn drop(&mut self) { + if let Some(runtime) = self.runtime.take() { + runtime.shutdown_background(); + } + } } #[cfg(test)] mod tests { + use std::process::Command; + use std::thread; + + use tokio::sync::{mpsc, oneshot}; + use super::*; + const RUNTIME_SELF_DROP_CHILD_ENV: &str = "LDK_NODE_RUNTIME_SELF_DROP_CHILD"; + const STORE_RUNTIME_SELF_DROP_CHILD_ENV: &str = "LDK_NODE_STORE_RUNTIME_SELF_DROP_CHILD"; + const EXPORTED_HANDLE_DROP_CHILD_ENV: &str = "LDK_NODE_EXPORTED_HANDLE_DROP_CHILD"; + + struct DropNotifier(Option>); + + impl Drop for DropNotifier { + fn drop(&mut self) { + if let Some(sender) = self.0.take() { + let _ = sender.send(()); + } + } + } + + fn test_runtime() -> Runtime { + Runtime::new(Arc::new(Logger::new_log_facade())).unwrap() + } + + fn test_runtime_with_workers(worker_threads: usize) -> Runtime { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(worker_threads) + .enable_all() + .build() + .unwrap(); + Runtime::with_owned_runtime(runtime, Arc::new(Logger::new_log_facade())) + } + + #[test] + fn runtime_control_does_not_retain_runtime_owner() { + let runtime = test_runtime(); + let control = runtime.control(); + let weak_control = Arc::downgrade(&control); + + drop(runtime); + + assert!(weak_control.upgrade().is_some()); + let (late_sender, mut late_receiver) = oneshot::channel(); + control.spawn_background_task(async move { + let _ = late_sender.send(()); + }); + assert!(matches!(late_receiver.try_recv(), Err(oneshot::error::TryRecvError::Closed))); + drop(control); + assert!(weak_control.upgrade().is_none()); + } + + #[test] + fn completed_cancellable_tasks_are_released_before_shutdown() { + const TASK_COUNT: usize = 64; + + let runtime = test_runtime(); + let (completion_sender, mut completion_receiver) = mpsc::channel(TASK_COUNT); + for _ in 0..TASK_COUNT { + let completion_sender = completion_sender.clone(); + runtime.spawn_cancellable_background_task(async move { + completion_sender.send(()).await.unwrap(); + }); + } + drop(completion_sender); + + runtime.block_on(async { + for _ in 0..TASK_COUNT { + completion_receiver.recv().await.unwrap(); + } + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if runtime.cancellable_background_tasks.lock().unwrap().tasks.is_empty() { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + }); + } + + #[test] + fn late_task_spawns_are_not_polled_after_shutdown_starts() { + let runtime = test_runtime(); + runtime.close_task_admission(); + + let (background_sender, background_receiver) = oneshot::channel(); + runtime.spawn_background_task(async move { + let _ = background_sender.send(()); + }); + let (cancellable_sender, cancellable_receiver) = oneshot::channel(); + runtime.spawn_cancellable_background_task(async move { + let _ = cancellable_sender.send(()); + }); + + assert!(runtime.block_on(background_receiver).is_err()); + assert!(runtime.block_on(cancellable_receiver).is_err()); + } + + #[test] + fn timed_out_background_task_is_drained_before_shutdown_returns() { + let runtime = test_runtime(); + let (started_sender, started_receiver) = oneshot::channel(); + let (dropped_sender, dropped_receiver) = oneshot::channel(); + runtime.spawn_background_task(async move { + let _drop_notifier = DropNotifier(Some(dropped_sender)); + let _ = started_sender.send(()); + std::future::pending::<()>().await; + }); + runtime.block_on(started_receiver).unwrap(); + + runtime.wait_on_background_tasks_with_timeout(Duration::from_millis(10)); + + runtime.block_on(dropped_receiver).unwrap(); + } + + #[test] + fn non_cooperative_background_task_does_not_block_shutdown() { + let runtime = test_runtime_with_workers(2); + let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1); + let (release_sender, release_receiver) = std::sync::mpsc::sync_channel(1); + let (finished_sender, finished_receiver) = std::sync::mpsc::sync_channel(1); + runtime.spawn_background_task(async move { + started_sender.send(()).unwrap(); + release_receiver.recv().unwrap(); + finished_sender.send(()).unwrap(); + }); + started_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + + let start = std::time::Instant::now(); + runtime.wait_on_background_tasks_with_timeout(Duration::from_millis(10)); + assert!(start.elapsed() < Duration::from_secs(1)); + + release_sender.send(()).unwrap(); + finished_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + } + + #[test] + fn non_cooperative_cancellable_task_does_not_block_shutdown() { + let runtime = test_runtime_with_workers(2); + let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1); + let (release_sender, release_receiver) = std::sync::mpsc::sync_channel(1); + let (finished_sender, finished_receiver) = std::sync::mpsc::sync_channel(1); + runtime.spawn_cancellable_background_task(async move { + started_sender.send(()).unwrap(); + release_receiver.recv().unwrap(); + finished_sender.send(()).unwrap(); + }); + started_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + + let start = std::time::Instant::now(); + runtime.abort_cancellable_background_tasks_with_timeout(Duration::from_millis(10)); + assert!(start.elapsed() < Duration::from_secs(1)); + + release_sender.send(()).unwrap(); + finished_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + } + + #[test] + fn timed_out_event_processor_is_drained_before_shutdown_returns() { + let runtime = test_runtime(); + let (started_sender, started_receiver) = oneshot::channel(); + let (dropped_sender, dropped_receiver) = oneshot::channel(); + runtime.spawn_background_processor_task( + async move { + let _drop_notifier = DropNotifier(Some(dropped_sender)); + let _ = started_sender.send(()); + std::future::pending::>().await + }, + |_| {}, + ); + runtime.block_on(started_receiver).unwrap(); + + runtime.wait_on_background_processor_task_with_timeout(Duration::from_millis(10)).unwrap(); + + runtime.block_on(dropped_receiver).unwrap(); + } + + #[test] + fn non_cooperative_event_processor_does_not_block_shutdown() { + let runtime = test_runtime_with_workers(2); + let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1); + let (release_sender, release_receiver) = std::sync::mpsc::sync_channel(1); + let (finished_sender, finished_receiver) = std::sync::mpsc::sync_channel(1); + runtime.spawn_background_processor_task( + async move { + started_sender.send(()).unwrap(); + release_receiver.recv().unwrap(); + finished_sender.send(()).unwrap(); + Ok(()) + }, + |_| {}, + ); + started_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + + let start = std::time::Instant::now(); + runtime.wait_on_background_processor_task_with_timeout(Duration::from_millis(10)).unwrap(); + assert!(start.elapsed() < Duration::from_secs(1)); + + release_sender.send(()).unwrap(); + finished_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + } + + #[test] + fn event_processor_persistence_failure_is_returned() { + let runtime = test_runtime(); + let (failure_sender, failure_receiver) = oneshot::channel(); + runtime.spawn_background_processor_task( + async { Err(LdkIoError::new(LdkIoErrorKind::Other, "persistence failed")) }, + move |_| { + let _ = failure_sender.send(()); + }, + ); + + runtime.block_on(failure_receiver).unwrap(); + + let error = runtime.wait_on_background_processor_task().unwrap_err(); + + assert_eq!(error.kind(), LdkIoErrorKind::Other); + assert!(error.to_string().contains("persistence failed")); + } + + #[test] + fn repeated_one_worker_runtime_self_drop_is_abort_free() { + if std::env::var_os(RUNTIME_SELF_DROP_CHILD_ENV).is_some() { + run_repeated_one_worker_runtime_self_drop(); + return; + } + + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "runtime::tests::repeated_one_worker_runtime_self_drop_is_abort_free", + "--nocapture", + ]) + .env(RUNTIME_SELF_DROP_CHILD_ENV, "1") + .status() + .unwrap(); + + assert!(status.success(), "owned runtime self-drop aborted the subprocess"); + } + + fn run_repeated_one_worker_runtime_self_drop() { + for _ in 0..100 { + let tokio_runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .unwrap(); + let runtime = + Runtime::with_owned_runtime(tokio_runtime, Arc::new(Logger::new_log_facade())); + let exported_control = runtime.control(); + let handle = exported_control.handle().clone(); + let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1); + exported_control.spawn_background_task(async move { + started_sender.send(()).unwrap(); + std::future::pending::<()>().await; + }); + started_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + exported_control.close_task_admission(); + exported_control.wait_on_background_tasks_with_timeout(Duration::from_millis(1)); + let (dropped_sender, dropped_receiver) = std::sync::mpsc::sync_channel(1); + + handle.spawn(async move { + drop(runtime); + dropped_sender.send(()).unwrap(); + }); + + dropped_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + drop(exported_control); + } + } + #[test] - fn runtime_handle_does_not_retain_runtime_owner() { - let runtime = Arc::new(Runtime::new(Arc::new(Logger::new_log_facade())).unwrap()); - let weak_runtime = Arc::downgrade(&runtime); + fn store_runtime_self_drop_is_abort_free() { + if std::env::var_os(STORE_RUNTIME_SELF_DROP_CHILD_ENV).is_some() { + let runtime = StoreRuntime::new("vss-test-runtime", 1, "VSS test").unwrap(); + let handle = runtime.handle().clone(); + let (dropped_sender, dropped_receiver) = std::sync::mpsc::sync_channel(1); + handle.spawn(async move { + drop(runtime); + dropped_sender.send(()).unwrap(); + }); + dropped_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + return; + } + + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "runtime::tests::store_runtime_self_drop_is_abort_free", + "--nocapture", + ]) + .env(STORE_RUNTIME_SELF_DROP_CHILD_ENV, "1") + .status() + .unwrap(); - let runtime_handle = runtime.handle().clone(); + assert!(status.success(), "VSS store runtime self-drop aborted the subprocess"); + } - assert_eq!(Arc::strong_count(&runtime), 1); + #[test] + fn exported_control_block_on_after_runtime_drop_returns_not_running() { + let runtime = test_runtime(); + let control = runtime.control(); drop(runtime); - assert!(weak_runtime.upgrade().is_none()); + assert_eq!(control.try_block_on(async { 1u8 }), Err(Error::NotRunning)); + } + + #[test] + fn in_flight_exported_block_on_completes_before_runtime_drop() { + if std::env::var_os(EXPORTED_HANDLE_DROP_CHILD_ENV).is_some() { + let runtime = test_runtime(); + let control = runtime.control(); + let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1); + let (done_sender, done_receiver) = std::sync::mpsc::sync_channel(1); + thread::spawn(move || { + let result = control.try_block_on(async { + started_sender.send(()).unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + 1u8 + }); + done_sender.send(result).unwrap(); + }); + started_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + drop(runtime); + assert_eq!(done_receiver.recv_timeout(Duration::from_secs(2)).unwrap(), Ok(1)); + return; + } + + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "runtime::tests::in_flight_exported_block_on_completes_before_runtime_drop", + "--nocapture", + ]) + .env(EXPORTED_HANDLE_DROP_CHILD_ENV, "1") + .status() + .unwrap(); + + assert!(status.success(), "exported handle block_on aborted during runtime teardown"); + } + + #[test] + fn detached_background_work_blocks_a_new_task_generation() { + let runtime = test_runtime_with_workers(2); + let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1); + let (release_sender, release_receiver) = std::sync::mpsc::sync_channel(1); + runtime.spawn_background_task(async move { + started_sender.send(()).unwrap(); + release_receiver.recv().unwrap(); + }); + started_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + + runtime.wait_on_background_tasks_with_timeout(Duration::from_millis(10)); + assert!(!runtime.allow_task_spawns_with_timeout(Duration::from_millis(20))); + + release_sender.send(()).unwrap(); + assert!(runtime.allow_task_spawns_with_timeout(Duration::from_secs(1))); + } + + #[test] + fn detached_event_processor_blocks_a_new_task_generation() { + let runtime = test_runtime_with_workers(2); + let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1); + let (release_sender, release_receiver) = std::sync::mpsc::sync_channel(1); + runtime.spawn_background_processor_task( + async move { + started_sender.send(()).unwrap(); + release_receiver.recv().unwrap(); + Ok(()) + }, + |_| {}, + ); + started_receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + + runtime.wait_on_background_processor_task_with_timeout(Duration::from_millis(10)).unwrap(); + assert!(!runtime.allow_task_spawns_with_timeout(Duration::from_millis(20))); - drop(runtime_handle); + release_sender.send(()).unwrap(); + assert!(runtime.allow_task_spawns_with_timeout(Duration::from_secs(1))); } } diff --git a/src/scoring.rs b/src/scoring.rs index ee70f0087c..314ca7bbe5 100644 --- a/src/scoring.rs +++ b/src/scoring.rs @@ -11,14 +11,14 @@ use crate::config::{ }; use crate::io::utils::write_external_pathfinding_scores_to_cache; use crate::logger::LdkLogger; -use crate::runtime::Runtime; +use crate::runtime::RuntimeControl; use crate::{write_node_metrics, DynStore, Logger, NodeMetrics, Scorer}; /// Start a background task that periodically downloads scores via an external url and merges them into the local /// pathfinding scores. pub fn setup_background_pathfinding_scores_sync( url: String, scorer: Arc>, node_metrics: Arc>, - kv_store: Arc, logger: Arc, runtime: Arc, + kv_store: Arc, logger: Arc, runtime: Arc, mut stop_receiver: tokio::sync::watch::Receiver<()>, ) { log_info!(logger, "External scores background syncing enabled from {}", url);