Add LSPS2ServiceHandler persistence

We add simple `persist` call to `LSPS2ServiceHandler` that sequentially
persist all the peer states under a key that encodes their node id.

Author: tnull

lightning-liquidity/src/lib.rs

@@ -65,6 +65,7 @@ pub mod lsps2;
 pub mod lsps5;
 mod manager;
 pub mod message_queue;
+pub mod persist;
 #[allow(dead_code)]
 #[allow(unused_imports)]
 mod sync;
@@ -73,6 +74,6 @@ mod tests;
 pub mod utils;
 
 pub use manager::{
-	ALiquidityManager, ALiquidityManagerSync, LiquidityClientConfig, LiquidityManager, LiquidityManagerSync,
-	LiquidityServiceConfig,
+	ALiquidityManager, ALiquidityManagerSync, LiquidityClientConfig, LiquidityManager,
+	LiquidityManagerSync, LiquidityServiceConfig,
 };

lightning-liquidity/src/lsps2/service.rs

@@ -9,11 +9,15 @@
 
 //! Contains the main bLIP-52 / LSPS2 server-side object, [`LSPS2ServiceHandler`].
 
+use alloc::boxed::Box;
 use alloc::string::{String, ToString};
 use alloc::vec::Vec;
+use lightning::util::persist::KVStore;
 
 use core::cmp::Ordering as CmpOrdering;
+use core::future::Future;
 use core::ops::Deref;
+use core::pin::Pin;
 use core::sync::atomic::{AtomicUsize, Ordering};
 
 use crate::events::EventQueue;
@@ -28,6 +32,9 @@ use crate::lsps2::utils::{
 	compute_opening_fee, is_expired_opening_fee_params, is_valid_opening_fee_params,
 };
 use crate::message_queue::{MessageQueue, MessageQueueNotifierGuard};
+use crate::persist::{
+	LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+};
 use crate::prelude::hash_map::Entry;
 use crate::prelude::{new_hash_map, HashMap};
 use crate::sync::{Arc, Mutex, MutexGuard, RwLock};
@@ -38,6 +45,7 @@ use lightning::ln::msgs::{ErrorAction, LightningError};
 use lightning::ln::types::ChannelId;
 use lightning::util::errors::APIError;
 use lightning::util::logger::Level;
+use lightning::util::ser::Writeable;
 use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum};
 
 use lightning_types::payment::PaymentHash;
@@ -569,6 +577,7 @@ where
 	CM::Target: AChannelManager,
 {
 	channel_manager: CM,
+	kv_store: Arc<dyn KVStore + Send + Sync>,
 	pending_messages: Arc<MessageQueue>,
 	pending_events: Arc<EventQueue>,
 	per_peer_state: RwLock<HashMap<PublicKey, Mutex<PeerState>>>,
@@ -585,7 +594,7 @@ where
 	/// Constructs a `LSPS2ServiceHandler`.
 	pub(crate) fn new(
 		pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue>, channel_manager: CM,
-		config: LSPS2ServiceConfig,
+		kv_store: Arc<dyn KVStore + Send + Sync>, config: LSPS2ServiceConfig,
 	) -> Self {
 		Self {
 			pending_messages,
@@ -595,6 +604,7 @@ where
 			peer_by_channel_id: RwLock::new(new_hash_map()),
 			total_pending_requests: AtomicUsize::new(0),
 			channel_manager,
+			kv_store,
 			config,
 		}
 	}
@@ -1442,6 +1452,45 @@ where
 		);
 	}
 
+	fn persist_peer_state(
+		&self, counterparty_node_id: PublicKey, peer_state: &PeerState,
+	) -> Pin<Box<dyn Future<Output = Result<(), lightning::io::Error>> + Send>> {
+		let key = counterparty_node_id.to_string();
+		let encoded = peer_state.encode();
+		self.kv_store.write(
+			LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+			LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+			&key,
+			encoded,
+		)
+	}
+
+	pub(crate) fn persist(
+		&self,
+	) -> Pin<Box<dyn Future<Output = Result<(), lightning::io::Error>> + Send>> {
+		let outer_state_lock = self.per_peer_state.read().unwrap();
+		let mut futures = Vec::new();
+		for (counterparty_node_id, inner_state_lock) in outer_state_lock.iter() {
+			let peer_state_lock = inner_state_lock.lock().unwrap();
+			let fut = self.persist_peer_state(*counterparty_node_id, &*peer_state_lock);
+			futures.push(fut);
+		}
+
+		// TODO: We should eventually persist in parallel, however, when we do, we probably want to
+		// introduce some batching to upper-bound the number of requests inflight at any given
+		// time.
+		Box::pin(async move {
+			let mut ret = Ok(());
+			for fut in futures {
+				let res = fut.await;
+				if res.is_err() {
+					ret = res;
+				}
+			}
+			ret
+		})
+	}
+
 	pub(crate) fn peer_disconnected(&self, counterparty_node_id: PublicKey) {
 		let mut outer_state_lock = self.per_peer_state.write().unwrap();
 		let is_prunable =

lightning-liquidity/src/manager.rs

@@ -7,6 +7,7 @@
 // You may not use this file except in accordance with one or both of these
 // licenses.
 
+use alloc::boxed::Box;
 use alloc::string::ToString;
 use alloc::vec::Vec;
 
@@ -53,7 +54,9 @@ use lightning_types::features::{InitFeatures, NodeFeatures};
 
 use bitcoin::secp256k1::PublicKey;
 
+use core::future::Future as StdFuture;
 use core::ops::Deref;
+use core::pin::Pin;
 
 const LSPS_FEATURE_BIT: usize = 729;
 
@@ -173,7 +176,9 @@ pub trait ALiquidityManagerSync {
 	type TP: Deref<Target = Self::TimeProvider> + Clone;
 	/// Returns the inner async [`LiquidityManager`] for testing purposes.
 	#[cfg(any(test, feature = "_test_utils"))]
-	fn get_lm_async(&self) -> Arc<LiquidityManager<Self::ES, Self::NS, Self::CM, Self::C, Self::TP>>;
+	fn get_lm_async(
+		&self,
+	) -> Arc<LiquidityManager<Self::ES, Self::NS, Self::CM, Self::C, Self::TP>>;
 	/// Returns a reference to the actual [`LiquidityManager`] object.
 	fn get_lm(&self) -> &LiquidityManagerSync<Self::ES, Self::NS, Self::CM, Self::C, Self::TP>;
 }
@@ -204,7 +209,9 @@ where
 	type TP = TP;
 	/// Returns the inner async [`LiquidityManager`] for testing purposes.
 	#[cfg(any(test, feature = "_test_utils"))]
-	fn get_lm_async(&self) -> Arc<LiquidityManager<Self::ES, Self::NS, Self::CM, Self::C, Self::TP>> {
+	fn get_lm_async(
+		&self,
+	) -> Arc<LiquidityManager<Self::ES, Self::NS, Self::CM, Self::C, Self::TP>> {
 		Arc::clone(&self.inner)
 	}
 	fn get_lm(&self) -> &LiquidityManagerSync<ES, NS, CM, C, TP> {
@@ -262,7 +269,6 @@ pub struct LiquidityManager<
 	_client_config: Option<LiquidityClientConfig>,
 	best_block: RwLock<Option<BestBlock>>,
 	_chain_source: Option<C>,
-	kv_store: Arc<dyn KVStore + Send + Sync>,
 }
 
 #[cfg(feature = "time")]
@@ -349,6 +355,7 @@ where
 					Arc::clone(&pending_messages),
 					Arc::clone(&pending_events),
 					channel_manager.clone(),
+					Arc::clone(&kv_store),
 					config.clone(),
 				)
 			})
@@ -444,7 +451,6 @@ where
 			_client_config: client_config,
 			best_block: RwLock::new(chain_params.map(|chain_params| chain_params.best_block)),
 			_chain_source: chain_source,
-			kv_store,
 		}
 	}
 
@@ -483,7 +489,7 @@ where
 
 	/// Returns a reference to the LSPS2 server-side handler.
 	///
-	/// The returned handler allows to initiate the LSPS2 service-side flow.
+	/// The returned hendler allows to initiate the LSPS2 service-side flow.
 	pub fn lsps2_service_handler(&self) -> Option<&LSPS2ServiceHandler<CM>> {
 		self.lsps2_service_handler.as_ref()
 	}
@@ -559,6 +565,31 @@ where
 		self.pending_events.get_and_clear_pending_events()
 	}
 
+	/// Persists the state of the service handlers towards the given [`KVStore`] implementation.
+	///
+	/// This will be regularly called by LDK's background processor if necessary and only needs to
+	/// be called manually if it's not utilized.
+	pub fn persist(
+		&self,
+	) -> Pin<Box<dyn StdFuture<Output = Result<(), lightning::io::Error>> + Send>> {
+		let mut futures = Vec::new();
+		if let Some(lsps2_service_handler) = self.lsps2_service_handler.as_ref() {
+			futures.push(lsps2_service_handler.persist());
+		}
+
+		// TODO: We should eventually persist in parallel.
+		Box::pin(async move {
+			let mut ret = Ok(());
+			for fut in futures {
+				let res = fut.await;
+				if res.is_err() {
+					ret = res;
+				}
+			}
+			ret
+		})
+	}
+
 	fn handle_lsps_message(
 		&self, msg: LSPSMessage, sender_node_id: &PublicKey,
 	) -> Result<(), lightning::ln::msgs::LightningError> {

lightning-liquidity/src/persist.rs

@@ -0,0 +1,20 @@
+// This file is Copyright its original authors, visible in version control
+// history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
+// or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
+// You may not use this file except in accordance with one or both of these
+// licenses.
+
+//! Types and utils for persistence.
+
+/// The primary namespace under which the [`LiquidityManager`] will be persisted.
+///
+/// [`LiquidityManager`]: crate::LiquidityManager
+pub const LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE: &str = "liquidity";
+
+/// The secondary namespace under which the [`LSPS2ServiceHandler`] data will be persisted.
+///
+/// [`LSPS2ServiceHandler`]: crate::lsps2::service::LSPS2ServiceHandler
+pub const LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps2_service";