diff --git a/CHANGELOG.md b/CHANGELOG.md index aa56a549b..2bf341b43 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,8 @@ -# 0.7.0-rc.60 (Synonym Fork) +# 0.7.0-rc.61 (Synonym Fork) ## Bug Fixes +- Moved peer persistence to async KV storage so slow writes no longer hold the peer-store lock. - Prevented Electrum runtime self-drop crashes and unbounded shutdown waits. - Scaled Electrum full-scan timeouts for additional wallets with the configured stop gap and batch size. diff --git a/Cargo.toml b/Cargo.toml index 53103a3d9..99b289442 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ exclude = ["bindings/uniffi-bindgen"] [package] name = "ldk-node" -version = "0.7.0-rc.60" +version = "0.7.0-rc.61" authors = ["Elias Rohrer "] homepage = "https://lightningdevkit.org/" license = "MIT OR Apache-2.0" diff --git a/Package.swift b/Package.swift index 8bc8e4241..f8c3eb73b 100644 --- a/Package.swift +++ b/Package.swift @@ -3,8 +3,8 @@ import PackageDescription -let tag = "v0.7.0-rc.60" -let checksum = "15d23d3252422fd0f6f77c388c65c877fd95c5ef8c805ac4f7255368a4efda2b" +let tag = "v0.7.0-rc.61" +let checksum = "8662cdf7f52725ceab0ebf68b7ded139dfbbb0b4b5ec4e451cb5f7cce505ce68" 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 5fc9217cd..a65660994 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.60 +version=0.7.0-rc.61 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 a7c7d6621..540eb2f36 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 2d5cb2b3e..8808437c1 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 10e61335f..5d4b4b9bd 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 75e4f1c9b..d345cadb4 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.60 +version=0.7.0-rc.61 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 f34dd2907..06bcba837 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 0f27feffd..734cfac7b 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 23225f566..8ff7ea357 100644 --- a/bindings/python/pyproject.toml +++ b/bindings/python/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "ldk_node" -version = "0.7.0-rc.60" +version = "0.7.0-rc.61" authors = [ { name="Elias Rohrer", email="dev@tnull.de" }, ] diff --git a/src/builder.rs b/src/builder.rs index 1810c9978..07cbcaec1 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -2590,12 +2590,12 @@ fn build_with_store_internal( }, }; - persist_missing_channel_peers( + runtime.block_on(persist_missing_channel_peers( channel_manager.list_channels().into_iter().map(|channel| channel.counterparty.node_id), &network_graph, &peer_store, Arc::clone(&logger), - ); + )); let om_mailbox = if let Some(AsyncPaymentsRole::Server) = async_payments_role { Some(Arc::new(OnionMessageMailbox::new())) diff --git a/src/event.rs b/src/event.rs index a70570ef0..8398ad7a7 100644 --- a/src/event.rs +++ b/src/event.rs @@ -2146,7 +2146,8 @@ where &self.network_graph, &self.peer_store, self.logger.clone(), - ); + ) + .await; } } }, diff --git a/src/lib.rs b/src/lib.rs index 50f0591cf..d2693239c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -108,7 +108,7 @@ use std::default::Default; use std::net::ToSocketAddrs; use std::ops::Deref; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; -use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard}; +use std::sync::{Arc, Mutex, RwLock}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; pub use balance::{AddressTypeBalance, BalanceDetails, LightningBalance, PendingSweepBalance}; @@ -228,53 +228,53 @@ pub struct Node { #[derive(Default)] pub(crate) struct RgsPeerRecoveryExclusions { - node_ids: RwLock>, + node_ids: tokio::sync::RwLock>, } impl RgsPeerRecoveryExclusions { - fn read(&self) -> RwLockReadGuard<'_, HashSet> { - self.node_ids.read().unwrap() + async fn read(&self) -> tokio::sync::RwLockReadGuard<'_, HashSet> { + self.node_ids.read().await } - fn write(&self) -> RwLockWriteGuard<'_, HashSet> { - self.node_ids.write().unwrap() + async fn write(&self) -> tokio::sync::RwLockWriteGuard<'_, HashSet> { + self.node_ids.write().await } - fn exclude_after_disconnect(&self, node_id: PublicKey) { - self.write().insert(node_id); + async fn exclude_after_disconnect(&self, node_id: PublicKey) { + self.write().await.insert(node_id); } - fn exclude_after_last_channel_close(&self, node_id: PublicKey) { - self.write().insert(node_id); + async fn exclude_after_last_channel_close(&self, node_id: PublicKey) { + self.write().await.insert(node_id); } #[cfg(test)] - fn clear_after_persistent_connect(&self, node_id: &PublicKey) { - self.write().remove(node_id); + async fn clear_after_persistent_connect(&self, node_id: &PublicKey) { + self.write().await.remove(node_id); } #[cfg(test)] - fn clear_after_channel_open(&self, node_id: &PublicKey) { - self.write().remove(node_id); + async fn clear_after_channel_open(&self, node_id: &PublicKey) { + self.write().await.remove(node_id); } - fn add_peer_and_clear_exclusion( + async fn add_peer_and_clear_exclusion( &self, peer_store: &PeerStore, peer_info: PeerInfo, ) -> Result<(), Error> where L: Deref, L::Target: LdkLogger, { - let mut excluded_node_ids = self.write(); + let mut excluded_node_ids = self.write().await; let node_id = peer_info.node_id; - peer_store.add_peer(peer_info)?; + peer_store.add_peer(peer_info).await?; excluded_node_ids.remove(&node_id); Ok(()) } #[cfg(test)] - fn contains(&self, node_id: &PublicKey) -> bool { - self.read().contains(node_id) + async fn contains(&self, node_id: &PublicKey) -> bool { + self.read().await.contains(node_id) } } @@ -353,9 +353,9 @@ impl Node { gossip_sync_logger, "Background sync of RGS gossip data finished in {}ms.", now.elapsed().as_millis() - ); + ); let peer_recovery_exclusions = - gossip_peer_recovery_exclusions.read(); + gossip_peer_recovery_exclusions.read().await; persist_missing_channel_peers_excluding( gossip_channel_manager .list_channels() @@ -365,7 +365,8 @@ impl Node { &gossip_peer_store, &peer_recovery_exclusions, Arc::clone(&gossip_sync_logger), - ); + ) + .await; { let mut locked_node_metrics = gossip_node_metrics.write().unwrap(); locked_node_metrics.latest_rgs_snapshot_timestamp = Some(updated_timestamp); @@ -1442,8 +1443,10 @@ impl Node { // Persist first so the address is updated even if the connection attempt // races with an in-flight reconnection loop attempt at the old address. if persist { - self.rgs_peer_recovery_exclusions - .add_peer_and_clear_exclusion(self.peer_store.as_ref(), peer_info.clone())?; + self.runtime.block_on( + self.rgs_peer_recovery_exclusions + .add_peer_and_clear_exclusion(self.peer_store.as_ref(), peer_info.clone()), + )?; } let con_node_id = peer_info.node_id; @@ -1477,8 +1480,10 @@ impl Node { log_info!(self.logger, "Disconnecting peer {}..", counterparty_node_id); - self.rgs_peer_recovery_exclusions.exclude_after_disconnect(counterparty_node_id); - match self.peer_store.remove_peer(&counterparty_node_id) { + self.runtime.block_on( + self.rgs_peer_recovery_exclusions.exclude_after_disconnect(counterparty_node_id), + ); + match self.runtime.block_on(self.peer_store.remove_peer(&counterparty_node_id)) { Ok(()) => {}, Err(e) => { log_error!(self.logger, "Failed to remove peer {}: {}", counterparty_node_id, e) @@ -1542,8 +1547,10 @@ impl Node { "Initiated channel creation with peer {}. ", peer_info.node_id ); - self.rgs_peer_recovery_exclusions - .add_peer_and_clear_exclusion(self.peer_store.as_ref(), peer_info)?; + self.runtime.block_on( + self.rgs_peer_recovery_exclusions + .add_peer_and_clear_exclusion(self.peer_store.as_ref(), peer_info), + )?; Ok(UserChannelId(user_channel_id)) }, Err(e) => { @@ -1964,9 +1971,11 @@ impl Node { // Check if this was the last open channel, if so, forget the peer. if open_channels.len() == 1 { - self.rgs_peer_recovery_exclusions - .exclude_after_last_channel_close(counterparty_node_id); - self.peer_store.remove_peer(&counterparty_node_id)?; + self.runtime.block_on( + self.rgs_peer_recovery_exclusions + .exclude_after_last_channel_close(counterparty_node_id), + ); + self.runtime.block_on(self.peer_store.remove_peer(&counterparty_node_id))?; } } @@ -2624,8 +2633,7 @@ pub(crate) fn total_anchor_channels_reserve_sats( #[cfg(test)] mod tests { use std::str::FromStr; - use std::sync::{mpsc, Arc}; - use std::thread; + use std::sync::Arc; use std::time::Duration; use super::*; @@ -2647,54 +2655,51 @@ mod tests { } } - #[test] - fn rgs_peer_recovery_exclusions_follow_disconnect_and_connect_transitions() { + #[tokio::test] + async fn rgs_peer_recovery_exclusions_follow_disconnect_and_connect_transitions() { let exclusions = RgsPeerRecoveryExclusions::default(); let node_id = test_node_id(); - exclusions.exclude_after_disconnect(node_id); - assert!(exclusions.contains(&node_id)); + exclusions.exclude_after_disconnect(node_id).await; + assert!(exclusions.contains(&node_id).await); - exclusions.clear_after_persistent_connect(&node_id); - assert!(!exclusions.contains(&node_id)); + exclusions.clear_after_persistent_connect(&node_id).await; + assert!(!exclusions.contains(&node_id).await); - exclusions.exclude_after_last_channel_close(node_id); - assert!(exclusions.contains(&node_id)); + exclusions.exclude_after_last_channel_close(node_id).await; + assert!(exclusions.contains(&node_id).await); - exclusions.clear_after_channel_open(&node_id); - assert!(!exclusions.contains(&node_id)); + exclusions.clear_after_channel_open(&node_id).await; + assert!(!exclusions.contains(&node_id).await); } - #[test] - fn rgs_peer_recovery_read_guard_serializes_disconnect_exclusion() { + #[tokio::test] + async fn rgs_peer_recovery_read_guard_serializes_disconnect_exclusion() { let exclusions = Arc::new(RgsPeerRecoveryExclusions::default()); let node_id = test_node_id(); - let recovery_read_guard = exclusions.read(); + let recovery_read_guard = exclusions.read().await; - let (attempting_write_tx, attempting_write_rx) = mpsc::channel(); - let (write_done_tx, write_done_rx) = mpsc::channel(); + let (attempting_write_tx, attempting_write_rx) = tokio::sync::oneshot::channel(); let writer_exclusions = Arc::clone(&exclusions); - let writer = thread::spawn(move || { + let mut writer = tokio::spawn(async move { attempting_write_tx.send(()).unwrap(); - writer_exclusions.exclude_after_disconnect(node_id); - write_done_tx.send(()).unwrap(); + writer_exclusions.exclude_after_disconnect(node_id).await; }); - attempting_write_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + attempting_write_rx.await.unwrap(); assert!( - write_done_rx.recv_timeout(Duration::from_millis(200)).is_err(), + tokio::time::timeout(Duration::from_millis(200), &mut writer).await.is_err(), "disconnect exclusion must wait while RGS recovery holds the live read guard" ); assert!(!recovery_read_guard.contains(&node_id)); drop(recovery_read_guard); - write_done_rx.recv_timeout(Duration::from_secs(1)).unwrap(); - writer.join().unwrap(); - assert!(exclusions.contains(&node_id)); + writer.await.unwrap(); + assert!(exclusions.contains(&node_id).await); } - #[test] - fn rgs_peer_recovery_persistent_peer_updates_take_exclusion_lock_first() { + #[tokio::test] + async fn rgs_peer_recovery_persistent_peer_updates_take_exclusion_lock_first() { let store: Arc = Arc::new(crate::io::test_utils::InMemoryStore::new()); let logger = Arc::new(Logger::new_log_facade()); let peer_store = Arc::new(PeerStore::new(store, logger)); @@ -2704,46 +2709,39 @@ mod tests { let peer_info = test_peer_info(node_id, 9738); let other_peer_info = test_peer_info(other_node_id, 9739); - exclusions.exclude_after_disconnect(node_id); - let recovery_read_guard = exclusions.read(); + exclusions.exclude_after_disconnect(node_id).await; + let recovery_read_guard = exclusions.read().await; - let (attempting_clear_tx, attempting_clear_rx) = mpsc::channel(); - let (clear_done_tx, clear_done_rx) = mpsc::channel(); + let (attempting_clear_tx, attempting_clear_rx) = tokio::sync::oneshot::channel(); let clear_exclusions = Arc::clone(&exclusions); let clear_peer_store = Arc::clone(&peer_store); let peer_info_for_clear = peer_info.clone(); - let clearer = thread::spawn(move || { + let mut clearer = tokio::spawn(async move { attempting_clear_tx.send(()).unwrap(); clear_exclusions .add_peer_and_clear_exclusion(clear_peer_store.as_ref(), peer_info_for_clear) + .await .unwrap(); - clear_done_tx.send(()).unwrap(); }); - attempting_clear_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + attempting_clear_rx.await.unwrap(); assert!( - clear_done_rx.recv_timeout(Duration::from_millis(200)).is_err(), + tokio::time::timeout(Duration::from_millis(200), &mut clearer).await.is_err(), "persistent peer update must wait for the recovery read guard before taking peer_store" ); - let (peer_store_write_tx, peer_store_write_rx) = mpsc::channel(); - let concurrent_peer_store = Arc::clone(&peer_store); - let other_peer_info_for_write = other_peer_info.clone(); - let peer_store_writer = thread::spawn(move || { - concurrent_peer_store.add_peer(other_peer_info_for_write).unwrap(); - peer_store_write_tx.send(()).unwrap(); - }); - peer_store_write_rx.recv_timeout(Duration::from_secs(1)).unwrap(); - peer_store_writer.join().unwrap(); + tokio::time::timeout(Duration::from_secs(1), peer_store.add_peer(other_peer_info.clone())) + .await + .expect("peer store must remain available while exclusion acquisition is pending") + .unwrap(); assert_eq!(peer_store.get_peer(&other_node_id), Some(other_peer_info)); assert!(peer_store.get_peer(&node_id).is_none()); assert!(recovery_read_guard.contains(&node_id)); drop(recovery_read_guard); - clear_done_rx.recv_timeout(Duration::from_secs(1)).unwrap(); - clearer.join().unwrap(); + clearer.await.unwrap(); assert_eq!(peer_store.get_peer(&node_id), Some(peer_info)); - assert!(!exclusions.contains(&node_id)); + assert!(!exclusions.contains(&node_id).await); } } diff --git a/src/payment/bolt11.rs b/src/payment/bolt11.rs index 4c96c485f..3670afc1c 100644 --- a/src/payment/bolt11.rs +++ b/src/payment/bolt11.rs @@ -814,7 +814,7 @@ impl Bolt11Payment { self.payment_store.insert(payment)?; // Persist LSP peer to make sure we reconnect on restart. - self.peer_store.add_peer(peer_info)?; + self.runtime.block_on(self.peer_store.add_peer(peer_info))?; Ok(invoice) } diff --git a/src/peer_store.rs b/src/peer_store.rs index 0c2131a49..9d9771681 100644 --- a/src/peer_store.rs +++ b/src/peer_store.rs @@ -12,7 +12,7 @@ use std::sync::{Arc, RwLock}; use bitcoin::secp256k1::PublicKey; use lightning::impl_writeable_tlv_based; use lightning::routing::gossip::NodeId; -use lightning::util::persist::KVStoreSync; +use lightning::util::persist::KVStore; use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer}; use crate::io::{ @@ -28,6 +28,7 @@ where L::Target: LdkLogger, { peers: RwLock>, + mutation_lock: tokio::sync::Mutex<()>, kv_store: Arc, logger: L, } @@ -38,56 +39,90 @@ where { pub(crate) fn new(kv_store: Arc, logger: L) -> Self { let peers = RwLock::new(HashMap::new()); - Self { peers, kv_store, logger } + let mutation_lock = tokio::sync::Mutex::new(()); + Self { peers, mutation_lock, kv_store, logger } } - pub(crate) fn add_peer(&self, peer_info: PeerInfo) -> Result<(), Error> { - let mut locked_peers = self.peers.write().unwrap(); + pub(crate) async fn add_peer(&self, peer_info: PeerInfo) -> Result<(), Error> { + let _guard = self.mutation_lock.lock().await; + let node_id = peer_info.node_id; + let (previous_peer, data) = { + let mut locked_peers = self.peers.write().expect("lock"); + if let Some(existing) = locked_peers.get(&node_id) { + if existing.address == peer_info.address { + return Ok(()); + } + log_info!( + self.logger, + "Updating socket address for peer {}: {} -> {}", + node_id, + existing.address, + peer_info.address + ); + } - if let Some(existing) = locked_peers.get(&peer_info.node_id) { - if existing.address == peer_info.address { - return Ok(()); + let previous_peer = locked_peers.insert(node_id, peer_info); + let data = PeerStoreSerWrapper(&locked_peers).encode(); + (previous_peer, data) + }; + + if let Err(e) = self.persist_peers(data).await { + let mut locked_peers = self.peers.write().expect("lock"); + if let Some(previous_peer) = previous_peer { + locked_peers.insert(node_id, previous_peer); + } else { + locked_peers.remove(&node_id); } - log_info!( - self.logger, - "Updating socket address for peer {}: {} -> {}", - peer_info.node_id, - existing.address, - peer_info.address - ); + return Err(e); } - - let mut updated_peers = locked_peers.clone(); - updated_peers.insert(peer_info.node_id, peer_info); - self.persist_peers(&updated_peers)?; - *locked_peers = updated_peers; Ok(()) } - pub(crate) fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> { - let mut locked_peers = self.peers.write().unwrap(); - - locked_peers.remove(node_id); - self.persist_peers(&*locked_peers) + pub(crate) async fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> { + let _guard = self.mutation_lock.lock().await; + let (removed_peer, data) = { + let mut locked_peers = self.peers.write().expect("lock"); + let removed_peer = locked_peers.remove(node_id); + let data = PeerStoreSerWrapper(&locked_peers).encode(); + (removed_peer, data) + }; + + if let Err(e) = self.persist_peers(data).await { + if let Some(peer_info) = removed_peer { + self.peers.write().expect("lock").insert(*node_id, peer_info); + } + return Err(e); + } + Ok(()) } + /// Returns the current in-memory peer set. + /// + /// The async mutation lock serializes `add_peer` and `remove_peer`, but this synchronous + /// reader cannot wait on it. Until peer-store reads are async, callers may observe peer + /// changes that are still being persisted. pub(crate) fn list_peers(&self) -> Vec { - self.peers.read().unwrap().values().cloned().collect() + self.peers.read().expect("lock").values().cloned().collect() } + /// Returns the current in-memory peer info for `node_id`. + /// + /// The async mutation lock serializes `add_peer` and `remove_peer`, but this synchronous + /// reader cannot wait on it. Until peer-store reads are async, callers may observe peer + /// changes that are still being persisted. pub(crate) fn get_peer(&self, node_id: &PublicKey) -> Option { - self.peers.read().unwrap().get(node_id).cloned() + self.peers.read().expect("lock").get(node_id).cloned() } - fn persist_peers(&self, locked_peers: &HashMap) -> Result<(), Error> { - let data = PeerStoreSerWrapper(&*locked_peers).encode(); - KVStoreSync::write( + async fn persist_peers(&self, data: Vec) -> Result<(), Error> { + KVStore::write( &*self.kv_store, PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PEER_INFO_PERSISTENCE_KEY, data, ) + .await .map_err(|e| { log_error!( self.logger, @@ -114,7 +149,8 @@ where let (kv_store, logger) = args; let read_peers: PeerStoreDeserWrapper = Readable::read(reader)?; let peers: RwLock> = RwLock::new(read_peers.0); - Ok(Self { peers, kv_store, logger }) + let mutation_lock = tokio::sync::Mutex::new(()); + Ok(Self { peers, mutation_lock, kv_store, logger }) } } @@ -160,7 +196,7 @@ impl_writeable_tlv_based!(PeerInfo, { (2, address, required), }); -pub(crate) fn persist_missing_channel_peers( +pub(crate) async fn persist_missing_channel_peers( counterparty_node_ids: I, network_graph: &Graph, peer_store: &PeerStore, logger: L, ) where L: Deref, @@ -174,9 +210,10 @@ pub(crate) fn persist_missing_channel_peers( &HashSet::new(), logger, ) + .await } -pub(crate) fn persist_missing_channel_peers_excluding( +pub(crate) async fn persist_missing_channel_peers_excluding( counterparty_node_ids: I, network_graph: &Graph, peer_store: &PeerStore, excluded_node_ids: &HashSet, logger: L, ) where @@ -184,28 +221,29 @@ pub(crate) fn persist_missing_channel_peers_excluding( L::Target: LdkLogger, I: IntoIterator, { - let graph = network_graph.read_only(); - let mut seen = HashSet::new(); - let missing_peers = counterparty_node_ids - .into_iter() - .filter_map(|counterparty_node_id| { - if !seen.insert(counterparty_node_id) - || excluded_node_ids.contains(&counterparty_node_id) - || peer_store.get_peer(&counterparty_node_id).is_some() - { - return None; - } - - graph - .nodes() - .get(&NodeId::from_pubkey(&counterparty_node_id)) - .and_then(|node_info| node_info.announcement_info.as_ref()) - .and_then(|announcement_info| announcement_info.addresses().first()) - .cloned() - .map(|address| PeerInfo { node_id: counterparty_node_id, address }) - }) - .collect::>(); - drop(graph); + let missing_peers = { + let graph = network_graph.read_only(); + let mut seen = HashSet::new(); + counterparty_node_ids + .into_iter() + .filter_map(|counterparty_node_id| { + if !seen.insert(counterparty_node_id) + || excluded_node_ids.contains(&counterparty_node_id) + || peer_store.get_peer(&counterparty_node_id).is_some() + { + return None; + } + + graph + .nodes() + .get(&NodeId::from_pubkey(&counterparty_node_id)) + .and_then(|node_info| node_info.announcement_info.as_ref()) + .and_then(|announcement_info| announcement_info.addresses().first()) + .cloned() + .map(|address| PeerInfo { node_id: counterparty_node_id, address }) + }) + .collect::>() + }; for peer_info in missing_peers { let node_id = peer_info.node_id; @@ -213,7 +251,7 @@ pub(crate) fn persist_missing_channel_peers_excluding( continue; } - match peer_store.add_peer(peer_info) { + match peer_store.add_peer(peer_info).await { Ok(()) => { log_info!(logger, "Persisted peer {} from channel counterparty", node_id) }, @@ -232,21 +270,22 @@ mod tests { use std::str::FromStr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; + use std::time::Duration; use bitcoin::Network; use lightning::io; use lightning::ln::msgs::UnsignedNodeAnnouncement; use lightning::routing::gossip::{NodeAlias, NodeId}; use lightning::types::features::{ChannelFeatures, NodeFeatures}; - use lightning::util::persist::KVStore; + use lightning::util::persist::{KVStore, KVStoreSync}; use lightning::util::test_utils::TestLogger; use super::*; use crate::io::test_utils::InMemoryStore; use crate::logger::Logger; - #[test] - fn peer_info_persistence() { + #[tokio::test] + async fn peer_info_persistence() { let store: Arc = Arc::new(InMemoryStore::new()); let logger = Arc::new(TestLogger::new()); let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); @@ -257,22 +296,24 @@ mod tests { .unwrap(); let address = SocketAddress::from_str("127.0.0.1:9738").unwrap(); let expected_peer_info = PeerInfo { node_id, address }; - assert!(KVStoreSync::read( + assert!(KVStore::read( &*store, PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PEER_INFO_PERSISTENCE_KEY, ) + .await .is_err()); - peer_store.add_peer(expected_peer_info.clone()).unwrap(); + peer_store.add_peer(expected_peer_info.clone()).await.unwrap(); // Check we can read back what we persisted. - let persisted_bytes = KVStoreSync::read( + let persisted_bytes = KVStore::read( &*store, PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PEER_INFO_PERSISTENCE_KEY, ) + .await .unwrap(); let deser_peer_store = PeerStore::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap(); @@ -283,8 +324,8 @@ mod tests { assert_eq!(deser_peer_store.get_peer(&node_id), Some(expected_peer_info)); } - #[test] - fn peer_address_updated_on_readd() { + #[tokio::test] + async fn peer_address_updated_on_readd() { let store: Arc = Arc::new(InMemoryStore::new()); let logger = Arc::new(TestLogger::new()); let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); @@ -296,28 +337,29 @@ mod tests { let old_address = SocketAddress::from_str("34.65.186.40:9735").unwrap(); let new_address = SocketAddress::from_str("34.65.153.174:9735").unwrap(); - peer_store.add_peer(PeerInfo { node_id, address: old_address.clone() }).unwrap(); + peer_store.add_peer(PeerInfo { node_id, address: old_address.clone() }).await.unwrap(); assert_eq!(peer_store.get_peer(&node_id).unwrap().address, old_address); - peer_store.add_peer(PeerInfo { node_id, address: new_address.clone() }).unwrap(); + peer_store.add_peer(PeerInfo { node_id, address: new_address.clone() }).await.unwrap(); assert_eq!(peer_store.get_peer(&node_id).unwrap().address, new_address); assert_eq!(peer_store.list_peers().len(), 1); - let persisted_bytes = KVStoreSync::read( + let persisted_bytes = KVStore::read( &*store, PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PEER_INFO_PERSISTENCE_KEY, ) + .await .unwrap(); let deser_peer_store = PeerStore::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap(); assert_eq!(deser_peer_store.get_peer(&node_id).unwrap().address, new_address); } - #[test] - fn peer_same_address_skips_persist() { + #[tokio::test] + async fn peer_same_address_skips_persist() { let store: Arc = Arc::new(InMemoryStore::new()); let logger = Arc::new(TestLogger::new()); let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); @@ -328,15 +370,15 @@ mod tests { .unwrap(); let address = SocketAddress::from_str("127.0.0.1:9738").unwrap(); - peer_store.add_peer(PeerInfo { node_id, address: address.clone() }).unwrap(); + peer_store.add_peer(PeerInfo { node_id, address: address.clone() }).await.unwrap(); - peer_store.add_peer(PeerInfo { node_id, address }).unwrap(); + peer_store.add_peer(PeerInfo { node_id, address }).await.unwrap(); assert_eq!(peer_store.list_peers().len(), 1); } - #[test] - fn peer_add_persistence_failure_leaves_peer_retryable() { - let store: Arc = Arc::new(FailFirstWriteStore::new()); + #[tokio::test] + async fn peer_add_persistence_failure_leaves_peer_retryable() { + let store: Arc = Arc::new(FailableWriteStore::new(true)); let logger = Arc::new(TestLogger::new()); let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); @@ -347,15 +389,135 @@ mod tests { let address = SocketAddress::from_str("127.0.0.1:9738").unwrap(); let peer_info = PeerInfo { node_id, address }; - assert!(matches!(peer_store.add_peer(peer_info.clone()), Err(Error::PersistenceFailed))); + assert!(matches!( + peer_store.add_peer(peer_info.clone()).await, + Err(Error::PersistenceFailed) + )); assert!(peer_store.get_peer(&node_id).is_none()); - peer_store.add_peer(peer_info.clone()).unwrap(); + peer_store.add_peer(peer_info.clone()).await.unwrap(); assert_eq!(peer_store.get_peer(&node_id), Some(peer_info)); } - #[test] - fn missing_channel_peer_is_persisted_from_graph() { + #[tokio::test] + async fn peer_address_update_persistence_failure_leaves_update_retryable() { + let store = Arc::new(FailableWriteStore::new(false)); + let dyn_store: Arc = store.clone(); + let logger = Arc::new(TestLogger::new()); + let peer_store = PeerStore::new(dyn_store, Arc::clone(&logger)); + + let node_id = PublicKey::from_str( + "0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993", + ) + .unwrap(); + let old_address = SocketAddress::from_str("34.65.186.40:9735").unwrap(); + let new_address = SocketAddress::from_str("34.65.153.174:9735").unwrap(); + let old_peer_info = PeerInfo { node_id, address: old_address }; + let new_peer_info = PeerInfo { node_id, address: new_address }; + + peer_store.add_peer(old_peer_info.clone()).await.unwrap(); + store.fail_next_write(); + assert!(matches!( + peer_store.add_peer(new_peer_info.clone()).await, + Err(Error::PersistenceFailed) + )); + assert_eq!(peer_store.get_peer(&node_id), Some(old_peer_info)); + + peer_store.add_peer(new_peer_info.clone()).await.unwrap(); + assert_eq!(peer_store.get_peer(&node_id), Some(new_peer_info)); + } + + #[tokio::test] + async fn peer_remove_persistence_failure_leaves_removal_retryable() { + let store = Arc::new(FailableWriteStore::new(false)); + let dyn_store: Arc = store.clone(); + let logger = Arc::new(TestLogger::new()); + let peer_store = PeerStore::new(dyn_store, Arc::clone(&logger)); + + let node_id = PublicKey::from_str( + "0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993", + ) + .unwrap(); + let address = SocketAddress::from_str("127.0.0.1:9738").unwrap(); + let peer_info = PeerInfo { node_id, address }; + + peer_store.add_peer(peer_info.clone()).await.unwrap(); + store.fail_next_write(); + assert!(matches!(peer_store.remove_peer(&node_id).await, Err(Error::PersistenceFailed))); + assert_eq!(peer_store.get_peer(&node_id), Some(peer_info)); + + peer_store.remove_peer(&node_id).await.unwrap(); + assert!(peer_store.get_peer(&node_id).is_none()); + } + + #[tokio::test] + async fn peer_reads_continue_while_async_persistence_is_pending() { + let store = Arc::new(BlockingWriteStore::new()); + let dyn_store: Arc = store.clone(); + let logger = Arc::new(TestLogger::new()); + let peer_store = Arc::new(PeerStore::new(dyn_store, Arc::clone(&logger))); + + let first_node_id = PublicKey::from_str( + "0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993", + ) + .unwrap(); + let second_node_id = PublicKey::from_str( + "02eec7245d6b7d2ccb30380bfbe2a3648cd7a942653f5aa340edcea1f283686619", + ) + .unwrap(); + let first_peer_info = PeerInfo { + node_id: first_node_id, + address: SocketAddress::from_str("127.0.0.1:9738").unwrap(), + }; + let second_peer_info = PeerInfo { + node_id: second_node_id, + address: SocketAddress::from_str("127.0.0.1:9739").unwrap(), + }; + + let first_writer_store = Arc::clone(&peer_store); + let first_writer_peer = first_peer_info.clone(); + let first_writer = + tokio::spawn(async move { first_writer_store.add_peer(first_writer_peer).await }); + store.wait_for_write().await; + + let reader_store = Arc::clone(&peer_store); + let observed_peers = tokio::time::timeout( + Duration::from_millis(200), + tokio::task::spawn_blocking(move || reader_store.list_peers()), + ) + .await + .expect("peer reads must not wait for persistence") + .unwrap(); + assert_eq!(observed_peers, vec![first_peer_info]); + + let second_writer_store = Arc::clone(&peer_store); + let mut second_writer = + tokio::spawn(async move { second_writer_store.add_peer(second_peer_info).await }); + assert!( + tokio::time::timeout(Duration::from_millis(200), &mut second_writer).await.is_err(), + "peer mutations must remain serialized while persistence is pending" + ); + + store.release_write(); + first_writer.await.unwrap().unwrap(); + second_writer.await.unwrap().unwrap(); + assert_eq!(peer_store.list_peers().len(), 2); + + let persisted_bytes = KVStore::read( + &*store, + PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PEER_INFO_PERSISTENCE_KEY, + ) + .await + .unwrap(); + let persisted_peer_store = + PeerStore::read(&mut &persisted_bytes[..], (store, logger)).unwrap(); + assert_eq!(persisted_peer_store.list_peers().len(), 2); + } + + #[tokio::test] + async fn missing_channel_peer_is_persisted_from_graph() { let store: Arc = Arc::new(InMemoryStore::new()); let logger = Arc::new(Logger::new_log_facade()); let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); @@ -394,15 +556,15 @@ mod tests { }) .unwrap(); - persist_missing_channel_peers(vec![node_id], &network_graph, &peer_store, logger); + persist_missing_channel_peers(vec![node_id], &network_graph, &peer_store, logger).await; let peer = peer_store.get_peer(&node_id).unwrap(); assert_eq!(peer.node_id, node_id); assert_eq!(peer.address, address); } - #[test] - fn missing_channel_peer_without_announced_address_is_skipped() { + #[tokio::test] + async fn missing_channel_peer_without_announced_address_is_skipped() { let store: Arc = Arc::new(InMemoryStore::new()); let logger = Arc::new(Logger::new_log_facade()); let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); @@ -413,13 +575,13 @@ mod tests { ) .unwrap(); - persist_missing_channel_peers(vec![node_id], &network_graph, &peer_store, logger); + persist_missing_channel_peers(vec![node_id], &network_graph, &peer_store, logger).await; assert!(peer_store.get_peer(&node_id).is_none()); } - #[test] - fn missing_channel_peer_is_persisted_after_graph_retry() { + #[tokio::test] + async fn missing_channel_peer_is_persisted_after_graph_retry() { let store: Arc = Arc::new(InMemoryStore::new()); let logger = Arc::new(Logger::new_log_facade()); let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); @@ -440,7 +602,8 @@ mod tests { &network_graph, &peer_store, Arc::clone(&logger), - ); + ) + .await; assert!(peer_store.get_peer(&node_id).is_none()); network_graph @@ -466,15 +629,15 @@ mod tests { }) .unwrap(); - persist_missing_channel_peers(vec![node_id], &network_graph, &peer_store, logger); + persist_missing_channel_peers(vec![node_id], &network_graph, &peer_store, logger).await; let peer = peer_store.get_peer(&node_id).unwrap(); assert_eq!(peer.node_id, node_id); assert_eq!(peer.address, address); } - #[test] - fn excluded_missing_channel_peer_is_not_persisted_from_graph() { + #[tokio::test] + async fn excluded_missing_channel_peer_is_not_persisted_from_graph() { let store: Arc = Arc::new(InMemoryStore::new()); let logger = Arc::new(Logger::new_log_facade()); let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); @@ -520,24 +683,29 @@ mod tests { &peer_store, &excluded_node_ids, logger, - ); + ) + .await; assert!(peer_store.get_peer(&node_id).is_none()); } - struct FailFirstWriteStore { + struct FailableWriteStore { persisted_bytes: Mutex>>>, fail_next_write: AtomicBool, } - impl FailFirstWriteStore { - fn new() -> Self { + impl FailableWriteStore { + fn new(fail_next_write: bool) -> Self { Self { persisted_bytes: Mutex::new(HashMap::new()), - fail_next_write: AtomicBool::new(true), + fail_next_write: AtomicBool::new(fail_next_write), } } + fn fail_next_write(&self) { + self.fail_next_write.store(true, Ordering::Relaxed); + } + fn read_internal( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, ) -> io::Result> { @@ -589,7 +757,7 @@ mod tests { } } - impl KVStore for FailFirstWriteStore { + impl KVStore for FailableWriteStore { fn read( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, ) -> Pin>> + Send + 'static>> { @@ -619,7 +787,7 @@ mod tests { } } - impl KVStoreSync for FailFirstWriteStore { + impl KVStoreSync for FailableWriteStore { fn read( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, ) -> io::Result> { @@ -644,4 +812,94 @@ mod tests { self.list_internal(primary_namespace, secondary_namespace) } } + + struct BlockingWriteStore { + store: InMemoryStore, + block_next_write: AtomicBool, + write_started: Arc, + release_write: Arc, + } + + impl BlockingWriteStore { + fn new() -> Self { + Self { + store: InMemoryStore::new(), + block_next_write: AtomicBool::new(true), + write_started: Arc::new(tokio::sync::Notify::new()), + release_write: Arc::new(tokio::sync::Notify::new()), + } + } + + async fn wait_for_write(&self) { + self.write_started.notified().await; + } + + fn release_write(&self) { + self.release_write.notify_one(); + } + } + + impl KVStore for BlockingWriteStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> Pin>> + Send + 'static>> { + KVStore::read(&self.store, primary_namespace, secondary_namespace, key) + } + + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, + ) -> Pin> + Send + 'static>> { + let write = + KVStore::write(&self.store, primary_namespace, secondary_namespace, key, buf); + if self.block_next_write.swap(false, Ordering::AcqRel) { + let write_started = Arc::clone(&self.write_started); + let release_write = Arc::clone(&self.release_write); + Box::pin(async move { + write_started.notify_one(); + release_write.notified().await; + write.await + }) + } else { + write + } + } + + fn remove( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> Pin> + Send + 'static>> { + KVStore::remove(&self.store, primary_namespace, secondary_namespace, key, lazy) + } + + fn list( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> Pin>> + Send + 'static>> { + KVStore::list(&self.store, primary_namespace, secondary_namespace) + } + } + + impl KVStoreSync for BlockingWriteStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> io::Result> { + KVStoreSync::read(&self.store, primary_namespace, secondary_namespace, key) + } + + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, + ) -> io::Result<()> { + KVStoreSync::write(&self.store, primary_namespace, secondary_namespace, key, buf) + } + + fn remove( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> io::Result<()> { + KVStoreSync::remove(&self.store, primary_namespace, secondary_namespace, key, lazy) + } + + fn list( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> io::Result> { + KVStoreSync::list(&self.store, primary_namespace, secondary_namespace) + } + } }