Migrate LSPS4 to async/sync persistence

Author: Fernando Ledesma

lightning-background-processor/src/lib.rs

@@ -438,6 +438,8 @@ pub const NO_LIQUIDITY_MANAGER: Option<
 				                           + Send
 				                           + Sync,
 				T = &(dyn BroadcasterInterface + Send + Sync),
+				Logger = dyn lightning::util::logger::Logger + Send + Sync,
+				L = &(dyn lightning::util::logger::Logger + Send + Sync),
 			> + Send
 			+ Sync,
 	>,

lightning-liquidity/src/lsps4/htlc_store.rs

@@ -8,8 +8,6 @@
 use lightning::util::ser::{Readable, Writeable};
 use lightning::{impl_writeable_tlv_based, log_error};
 use lightning::ln::channelmanager::InterceptId;
-use lightning::ln::types::ChannelId;
-
 use lightning::util::logger::Logger;
 use lightning::util::persist::KVStoreSync;
 use lightning_types::payment::PaymentHash;
@@ -112,15 +110,18 @@ where L::Target: Logger, KV::Target: KVStoreSync {
 	) -> Result<Self, io::Error> {
 		let mut htlcs = Vec::new();
 
-		for stored_key in kv_store.list(
+		let stored_keys = kv_store.list(
 			INTERCEPTED_HTLC_STORE_PERSISTENCE_PRIMARY_NAMESPACE,
 			INTERCEPTED_HTLC_STORE_PERSISTENCE_SECONDARY_NAMESPACE,
-		)? {
-			let mut reader = Cursor::new(kv_store.read(
+		)?;
+
+		for stored_key in stored_keys {
+			let data = kv_store.read(
 				INTERCEPTED_HTLC_STORE_PERSISTENCE_PRIMARY_NAMESPACE,
 				INTERCEPTED_HTLC_STORE_PERSISTENCE_SECONDARY_NAMESPACE,
 				&stored_key,
-			)?);
+			)?;
+			let mut reader = Cursor::new(data);
 			let htlc = InterceptedHtlc::read(&mut reader).map_err(|e| {
 				log_error!(logger, "Failed to deserialize InterceptedHtlc: {}", e);
 				io::Error::new(
@@ -138,13 +139,19 @@ where L::Target: Logger, KV::Target: KVStoreSync {
 	}
 
 	pub(crate) fn insert(&self, htlc: InterceptedHtlc) -> Result<bool, io::Error> {
-		let mut locked_htlcs = self.htlcs.lock().unwrap();
-
-		if locked_htlcs.contains_key(&htlc.id()) {
-			return Ok(false);
+		// Check if already exists
+		{
+			let locked_htlcs = self.htlcs.lock().unwrap();
+			if locked_htlcs.contains_key(&htlc.id()) {
+				return Ok(false);
+			}
 		}
 
+		// Persist first (outside the lock)
 		self.persist(&htlc)?;
+
+		// Then insert into the map
+		let mut locked_htlcs = self.htlcs.lock().unwrap();
 		let updated = locked_htlcs.insert(htlc.id(), htlc).is_some();
 		Ok(updated)
 	}

lightning-liquidity/src/lsps4/scid_store.rs

@@ -7,21 +7,16 @@
 
 use lightning::util::ser::{Readable, Writeable};
 use lightning::{impl_writeable_tlv_based, log_error};
-use lightning::ln::channelmanager::InterceptId;
-use lightning::ln::types::ChannelId;
-
 use lightning::util::logger::Logger;
 use lightning::util::persist::KVStoreSync;
-use lightning_types::payment::PaymentHash;
 
 use bitcoin::secp256k1::PublicKey;
 
 use lightning::io::{self, Cursor};
 
 use std::collections::HashMap;
 use std::ops::Deref;
-use std::time::{SystemTime, UNIX_EPOCH};
-use crate::sync::{Arc, Mutex, MutexGuard, RwLock};
+use crate::sync::RwLock;
 
 
 use crate::lsps4::utils;
@@ -81,15 +76,18 @@ where L::Target: Logger, KV::Target: KVStoreSync {
 	) -> Result<Self, io::Error> {
 		let mut scids = Vec::new();
 
-		for stored_key in kv_store.list(
+		let stored_keys = kv_store.list(
 			INTERCEPT_SCID_STORE_PERSISTENCE_PRIMARY_NAMESPACE,
 			INTERCEPT_SCID_STORE_PERSISTENCE_SECONDARY_NAMESPACE,
-		)? {
-			let mut reader = Cursor::new(kv_store.read(
+		)?;
+
+		for stored_key in stored_keys {
+			let data = kv_store.read(
 				INTERCEPT_SCID_STORE_PERSISTENCE_PRIMARY_NAMESPACE,
 				INTERCEPT_SCID_STORE_PERSISTENCE_SECONDARY_NAMESPACE,
 				&stored_key,
-			)?);
+			)?;
+			let mut reader = Cursor::new(data);
 			let scid = ScidWithPeer::read(&mut reader).map_err(|e| {
 				log_error!(logger, "Failed to deserialize InterceptScid: {}", e);
 				io::Error::new(
@@ -110,12 +108,26 @@ where L::Target: Logger, KV::Target: KVStoreSync {
 	}
 
 	pub(crate) fn insert(&self, scid: ScidWithPeer) -> Result<bool, io::Error> {
-		let mut locked_peer_by_scid = self.peer_by_scid.write().unwrap();
-		let mut locked_scid_by_peer = self.scid_by_peer.write().unwrap();
+		use lightning::log_info;
+		log_info!(self.logger, "[LSPS4 ScidStore] Inserting SCID {} for peer {}", scid.scid(), scid.peer_id());
 
+		// Persist first
 		self.persist(&scid)?;
+
+		// Then insert into the maps
+		let mut locked_peer_by_scid = self.peer_by_scid.write().unwrap();
+		let mut locked_scid_by_peer = self.scid_by_peer.write().unwrap();
 		let updated = locked_peer_by_scid.insert(scid.scid(), scid.peer_id().clone()).is_some();
 		locked_scid_by_peer.insert(scid.peer_id().clone(), scid.scid());
+
+		log_info!(
+			self.logger,
+			"[LSPS4 ScidStore] Successfully inserted SCID {} for peer {} (was_update: {})",
+			scid.scid(),
+			scid.peer_id(),
+			updated
+		);
+
 		Ok(updated)
 	}
 
@@ -170,10 +182,26 @@ where L::Target: Logger, KV::Target: KVStoreSync {
 	}
 
 	pub fn get_peer(&self, scid: u64) -> Option<PublicKey> {
-		self.peer_by_scid.read().unwrap().get(&scid).cloned()
+		use lightning::log_debug;
+		let result = self.peer_by_scid.read().unwrap().get(&scid).cloned();
+		log_debug!(
+			self.logger,
+			"[LSPS4 ScidStore] get_peer({}) = {:?}",
+			scid,
+			result
+		);
+		result
 	}
 
 	pub fn get_scid(&self, peer_id: &PublicKey) -> Option<u64> {
-		self.scid_by_peer.read().unwrap().get(peer_id).cloned()
+		use lightning::log_debug;
+		let result = self.scid_by_peer.read().unwrap().get(peer_id).cloned();
+		log_debug!(
+			self.logger,
+			"[LSPS4 ScidStore] get_scid({}) = {:?}",
+			peer_id,
+			result
+		);
+		result
 	}
 }