Add LSPS5ServiceHandler persistence

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

Author: tnull

lightning-liquidity/src/lsps5/service.rs

@@ -17,6 +17,9 @@ use crate::lsps5::msgs::{
 	SetWebhookRequest, SetWebhookResponse, WebhookNotification, WebhookNotificationMethod,
 };
 use crate::message_queue::MessageQueue;
+use crate::persist::{
+	LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+};
 use crate::prelude::*;
 use crate::sync::{Arc, Mutex, RwLock, RwLockWriteGuard};
 use crate::utils::time::TimeProvider;
@@ -28,10 +31,15 @@ use lightning::ln::channelmanager::AChannelManager;
 use lightning::ln::msgs::{ErrorAction, LightningError};
 use lightning::sign::NodeSigner;
 use lightning::util::logger::Level;
+use lightning::util::persist::KVStore;
+use lightning::util::ser::Writeable;
 
+use core::future::Future;
 use core::ops::Deref;
+use core::pin::Pin;
 use core::time::Duration;
 
+use alloc::boxed::Box;
 use alloc::string::String;
 use alloc::vec::Vec;
 
@@ -131,6 +139,7 @@ where
 	time_provider: TP,
 	channel_manager: CM,
 	node_signer: NS,
+	kv_store: Arc<dyn KVStore + Send + Sync>,
 	last_pruning: Mutex<Option<LSPSDateTime>>,
 }
 
@@ -143,7 +152,8 @@ where
 	/// Constructs a `LSPS5ServiceHandler` using the given time provider.
 	pub(crate) fn new_with_time_provider(
 		event_queue: Arc<EventQueue>, pending_messages: Arc<MessageQueue>, channel_manager: CM,
-		node_signer: NS, config: LSPS5ServiceConfig, time_provider: TP,
+		kv_store: Arc<dyn KVStore + Send + Sync>, node_signer: NS, config: LSPS5ServiceConfig,
+		time_provider: TP,
 	) -> Self {
 		assert!(config.max_webhooks_per_client > 0, "`max_webhooks_per_client` must be > 0");
 		Self {
@@ -154,6 +164,7 @@ where
 			time_provider,
 			channel_manager,
 			node_signer,
+			kv_store,
 			last_pruning: Mutex::new(None),
 		}
 	}
@@ -186,6 +197,44 @@ 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,
+			LSPS5_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, peer_state) in outer_state_lock.iter() {
+			let fut = self.persist_peer_state(*counterparty_node_id, peer_state);
+			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
+		})
+	}
+
 	fn check_prune_stale_webhooks<'a>(
 		&self, outer_state_lock: &mut RwLockWriteGuard<'a, HashMap<PublicKey, PeerState>>,
 	) {

lightning-liquidity/src/manager.rs

@@ -384,6 +384,7 @@ where
 					Arc::clone(&pending_events),
 					Arc::clone(&pending_messages),
 					channel_manager.clone(),
+					Arc::clone(&kv_store),
 					node_signer,
 					config.clone(),
 					time_provider,
@@ -577,6 +578,10 @@ where
 			futures.push(lsps2_service_handler.persist());
 		}
 
+		if let Some(lsps5_service_handler) = self.lsps5_service_handler.as_ref() {
+			futures.push(lsps5_service_handler.persist());
+		}
+
 		// TODO: We should eventually persist in parallel.
 		Box::pin(async move {
 			let mut ret = Ok(());

lightning-liquidity/src/persist.rs

@@ -18,3 +18,8 @@ pub const LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE: &str = "liquidity";
 ///
 /// [`LSPS2ServiceHandler`]: crate::lsps2::service::LSPS2ServiceHandler
 pub const LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps2_service";
+
+/// The secondary namespace under which the [`LSPS5ServiceHandler`] data will be persisted.
+///
+/// [`LSPS5ServiceHandler`]: crate::lsps5::service::LSPS5ServiceHandler
+pub const LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps5_service";