fix: release peer lock before persistence

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.

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 <dev@tnull.de>"]
 homepage = "https://lightningdevkit.org/"
 license = "MIT OR Apache-2.0"

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(

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

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

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" },
 ]

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()))

src/event.rs

@@ -2146,7 +2146,8 @@ where
 							&self.network_graph,
 							&self.peer_store,
 							self.logger.clone(),
-						);
+						)
+						.await;
 					}
 				}
 			},

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<HashSet<PublicKey>>,
+	node_ids: tokio::sync::RwLock<HashSet<PublicKey>>,
 }
 
 impl RgsPeerRecoveryExclusions {
-	fn read(&self) -> RwLockReadGuard<'_, HashSet<PublicKey>> {
-		self.node_ids.read().unwrap()
+	async fn read(&self) -> tokio::sync::RwLockReadGuard<'_, HashSet<PublicKey>> {
+		self.node_ids.read().await
 	}
 
-	fn write(&self) -> RwLockWriteGuard<'_, HashSet<PublicKey>> {
-		self.node_ids.write().unwrap()
+	async fn write(&self) -> tokio::sync::RwLockWriteGuard<'_, HashSet<PublicKey>> {
+		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<L>(
+	async fn add_peer_and_clear_exclusion<L>(
 		&self, peer_store: &PeerStore<L>, 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<DynStore> = 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);
 	}
 }

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)
 	}