Read persisted LSPS1ServiceHandler state on startup

.. we read the persisted state in `LiquidityManager::new`

Author: tnull

lightning-liquidity/src/lsps1/mod.rs

@@ -13,6 +13,6 @@ pub mod client;
 pub mod event;
 pub mod msgs;
 #[cfg(lsps1_service)]
-mod peer_state;
+pub(crate) mod peer_state;
 #[cfg(lsps1_service)]
 pub mod service;

lightning-liquidity/src/lsps1/peer_state.rs

@@ -23,7 +23,7 @@ use lightning::util::hash_tables::new_hash_map;
 use core::fmt;
 
 #[derive(Default)]
-pub(super) struct PeerState {
+pub(crate) struct PeerState {
 	outbound_channels_by_order_id: HashMap<LSPS1OrderId, ChannelOrder>,
 	pending_requests: HashMap<LSPSRequestId, LSPS1Request>,
 	needs_persist: bool,

lightning-liquidity/src/lsps1/service.rs

@@ -36,7 +36,7 @@ use crate::persist::{
 	LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, LSPS1_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use crate::prelude::hash_map::Entry;
-use crate::prelude::{new_hash_map, HashMap};
+use crate::prelude::HashMap;
 use crate::sync::{Arc, Mutex, RwLock};
 use crate::utils;
 use crate::utils::async_poll::dummy_waker;
@@ -90,17 +90,17 @@ where
 {
 	/// Constructs a `LSPS1ServiceHandler`.
 	pub(crate) fn new(
-		entropy_source: ES, pending_messages: Arc<MessageQueue>,
-		pending_events: Arc<EventQueue<K>>, channel_manager: CM, kv_store: K, time_provider: TP,
-		config: LSPS1ServiceConfig,
+		per_peer_state: HashMap<PublicKey, Mutex<PeerState>>, entropy_source: ES,
+		pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue<K>>,
+		channel_manager: CM, kv_store: K, time_provider: TP, config: LSPS1ServiceConfig,
 	) -> Self {
 		Self {
 			entropy_source,
 			_channel_manager: channel_manager,
 			kv_store,
 			pending_messages,
 			pending_events,
-			per_peer_state: RwLock::new(new_hash_map()),
+			per_peer_state: RwLock::new(per_peer_state),
 			persistence_in_flight: AtomicUsize::new(0),
 			time_provider,
 			config,

lightning-liquidity/src/manager.rs

@@ -23,6 +23,8 @@ use crate::lsps5::client::{LSPS5ClientConfig, LSPS5ClientHandler};
 use crate::lsps5::msgs::LSPS5Message;
 use crate::lsps5::service::{LSPS5ServiceConfig, LSPS5ServiceHandler};
 use crate::message_queue::MessageQueue;
+#[cfg(lsps1_service)]
+use crate::persist::read_lsps1_service_peer_states;
 use crate::persist::{
 	read_event_queue, read_lsps2_service_peer_states, read_lsps5_service_peer_states,
 };
@@ -475,24 +477,32 @@ where
 		});
 
 		#[cfg(lsps1_service)]
-		let lsps1_service_handler = service_config.as_ref().and_then(|config| {
-			if let Some(number) =
-				<LSPS1ServiceHandler<ES, CM, K, TP> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
-			{
-				supported_protocols.push(number);
-			}
-			config.lsps1_service_config.as_ref().map(|config| {
-				LSPS1ServiceHandler::new(
+		let lsps1_service_handler = if let Some(service_config) = service_config.as_ref() {
+			if let Some(lsps1_service_config) = service_config.lsps1_service_config.as_ref() {
+				if let Some(number) =
+					<LSPS1ServiceHandler<ES, CM, K, TP> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
+				{
+					supported_protocols.push(number);
+				}
+
+				let peer_states = read_lsps1_service_peer_states(kv_store.clone()).await?;
+
+				Some(LSPS1ServiceHandler::new(
+					peer_states,
 					entropy_source.clone(),
 					Arc::clone(&pending_messages),
 					Arc::clone(&pending_events),
 					channel_manager.clone(),
 					kv_store.clone(),
 					time_provider,
-					config.clone(),
-				)
-			})
-		});
+					lsps1_service_config.clone(),
+				))
+			} else {
+				None
+			}
+		} else {
+			None
+		};
 
 		let lsps0_client_handler = LSPS0ClientHandler::new(
 			entropy_source.clone(),

lightning-liquidity/src/persist.rs

@@ -10,6 +10,8 @@
 //! Types and utils for persistence.
 
 use crate::events::{EventQueueDeserWrapper, LiquidityEvent};
+#[cfg(lsps1_service)]
+use crate::lsps1::peer_state::PeerState as LSPS1ServicePeerState;
 use crate::lsps2::service::PeerState as LSPS2ServicePeerState;
 use crate::lsps5::service::PeerState as LSPS5ServicePeerState;
 use crate::prelude::{new_hash_map, HashMap};
@@ -90,6 +92,51 @@ where
 	Ok(Some(queue.0))
 }
 
+#[cfg(lsps1_service)]
+pub(crate) async fn read_lsps1_service_peer_states<K: Deref>(
+	kv_store: K,
+) -> Result<HashMap<PublicKey, Mutex<LSPS1ServicePeerState>>, lightning::io::Error>
+where
+	K::Target: KVStore,
+{
+	let mut res = new_hash_map();
+
+	for stored_key in kv_store
+		.list(
+			LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+			LSPS1_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+		)
+		.await?
+	{
+		let mut reader = Cursor::new(
+			kv_store
+				.read(
+					LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+					LSPS1_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+					&stored_key,
+				)
+				.await?,
+		);
+
+		let peer_state = LSPS1ServicePeerState::read(&mut reader).map_err(|_| {
+			lightning::io::Error::new(
+				lightning::io::ErrorKind::InvalidData,
+				"Failed to deserialize LSPS1 peer state",
+			)
+		})?;
+
+		let key = PublicKey::from_str(&stored_key).map_err(|_| {
+			lightning::io::Error::new(
+				lightning::io::ErrorKind::InvalidData,
+				"Failed to deserialize stored key entry",
+			)
+		})?;
+
+		res.insert(key, Mutex::new(peer_state));
+	}
+	Ok(res)
+}
+
 pub(crate) async fn read_lsps2_service_peer_states<K: Deref>(
 	kv_store: K,
 ) -> Result<HashMap<PublicKey, Mutex<LSPS2ServicePeerState>>, lightning::io::Error>