Read persisted LSPS2 service state in LiquidityManager::new

We read any previously-persisted state upon construction of
`LiquidityManager`.

Author: tnull

fuzz/src/lsps_message.rs

@@ -86,7 +86,7 @@ pub fn do_test(data: &[u8]) {
 		kv_store,
 		None,
 		None,
-	));
+	).unwrap());
 	let mut reader = data;
 	if let Ok(Some(msg)) = liquidity_manager.read(LSPS_MESSAGE_TYPE_ID, &mut reader) {
 		let secp = Secp256k1::signing_only();

lightning-background-processor/src/lib.rs

@@ -2242,16 +2242,19 @@ mod tests {
 				Arc::clone(&logger),
 				Arc::clone(&keys_manager),
 			));
-			let liquidity_manager = Arc::new(LiquidityManagerSync::new(
-				Arc::clone(&keys_manager),
-				Arc::clone(&keys_manager),
-				Arc::clone(&manager),
-				None,
-				None,
-				Arc::clone(&kv_store) as Arc<dyn KVStoreSync + Sync + Send>,
-				None,
-				None,
-			));
+			let liquidity_manager = Arc::new(
+				LiquidityManagerSync::new(
+					Arc::clone(&keys_manager),
+					Arc::clone(&keys_manager),
+					Arc::clone(&manager),
+					None,
+					None,
+					Arc::clone(&kv_store) as Arc<dyn KVStoreSync + Sync + Send>,
+					None,
+					None,
+				)
+				.unwrap(),
+			);
 			let node = Node {
 				node: manager,
 				p2p_gossip_sync,

lightning-liquidity/src/lsps2/service.rs

@@ -468,7 +468,7 @@ impl OutboundJITChannel {
 	}
 }
 
-struct PeerState {
+pub(crate) struct PeerState {
 	outbound_channels_by_intercept_scid: HashMap<u64, OutboundJITChannel>,
 	intercept_scid_by_user_channel_id: HashMap<u128, u64>,
 	intercept_scid_by_channel_id: HashMap<ChannelId, u64>,
@@ -593,15 +593,35 @@ where
 {
 	/// Constructs a `LSPS2ServiceHandler`.
 	pub(crate) fn new(
-		pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue>, channel_manager: CM,
+		peer_states: Vec<(PublicKey, PeerState)>, pending_messages: Arc<MessageQueue>,
+		pending_events: Arc<EventQueue>, channel_manager: CM,
 		kv_store: Arc<dyn KVStore + Send + Sync>, config: LSPS2ServiceConfig,
 	) -> Self {
+		let mut peer_by_intercept_scid = new_hash_map();
+		let mut peer_by_channel_id = new_hash_map();
+		for (node_id, peer_state) in peer_states.iter() {
+			for (intercept_scid, _) in peer_state.outbound_channels_by_intercept_scid.iter() {
+				let res = peer_by_intercept_scid.insert(*intercept_scid, *node_id);
+				debug_assert!(res.is_none(), "Intercept SCIDs should never collide");
+			}
+
+			for (channel_id, _) in peer_state.intercept_scid_by_channel_id.iter() {
+				let res = peer_by_channel_id.insert(*channel_id, *node_id);
+				debug_assert!(res.is_none(), "Channel IDs should never collide");
+			}
+		}
+
+		let per_peer_state = peer_states
+			.into_iter()
+			.map(|(k, v)| (k, Mutex::new(v)))
+			.collect::<HashMap<PublicKey, Mutex<PeerState>>>();
+
 		Self {
 			pending_messages,
 			pending_events,
-			per_peer_state: RwLock::new(new_hash_map()),
-			peer_by_intercept_scid: RwLock::new(new_hash_map()),
-			peer_by_channel_id: RwLock::new(new_hash_map()),
+			per_peer_state: RwLock::new(per_peer_state),
+			peer_by_intercept_scid: RwLock::new(peer_by_intercept_scid),
+			peer_by_channel_id: RwLock::new(peer_by_channel_id),
 			total_pending_requests: AtomicUsize::new(0),
 			channel_manager,
 			kv_store,

lightning-liquidity/src/manager.rs

@@ -24,6 +24,7 @@ use crate::lsps5::client::{LSPS5ClientConfig, LSPS5ClientHandler};
 use crate::lsps5::msgs::LSPS5Message;
 use crate::lsps5::service::{LSPS5ServiceConfig, LSPS5ServiceHandler};
 use crate::message_queue::MessageQueue;
+use crate::persist::read_lsps2_service_peer_states;
 
 use crate::lsps1::client::{LSPS1ClientConfig, LSPS1ClientHandler};
 use crate::lsps1::msgs::LSPS1Message;
@@ -35,6 +36,7 @@ use crate::lsps2::msgs::LSPS2Message;
 use crate::lsps2::service::{LSPS2ServiceConfig, LSPS2ServiceHandler};
 use crate::prelude::{new_hash_map, new_hash_set, HashMap, HashSet};
 use crate::sync::{Arc, Mutex, RwLock};
+use crate::utils::async_poll::dummy_waker;
 #[cfg(feature = "time")]
 use crate::utils::time::DefaultTimeProvider;
 use crate::utils::time::TimeProvider;
@@ -57,6 +59,7 @@ use bitcoin::secp256k1::PublicKey;
 use core::future::Future as StdFuture;
 use core::ops::Deref;
 use core::pin::Pin;
+use core::task;
 
 const LSPS_FEATURE_BIT: usize = 729;
 
@@ -281,12 +284,14 @@ where
 	C::Target: Filter,
 {
 	/// Constructor for the [`LiquidityManager`] using the default system clock
-	pub fn new(
+	///
+	/// Will read persisted service states from the given [`KVStore`].
+	pub async fn new(
 		entropy_source: ES, node_signer: NS, channel_manager: CM, chain_source: Option<C>,
 		chain_params: Option<ChainParameters>, kv_store: Arc<dyn KVStore + Send + Sync>,
 		service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>,
-	) -> Self {
+	) -> Result<Self, lightning::io::Error> {
 		let time_provider = Arc::new(DefaultTimeProvider);
 		Self::new_with_custom_time_provider(
 			entropy_source,
@@ -299,6 +304,7 @@ where
 			client_config,
 			time_provider,
 		)
+		.await
 	}
 }
 
@@ -318,16 +324,18 @@ where
 {
 	/// Constructor for the [`LiquidityManager`] with a custom time provider.
 	///
+	/// Will read persisted service states from the given [`KVStore`].
+	///
 	/// This should be used on non-std platforms where access to the system time is not
 	/// available.
 	/// Sets up the required protocol message handlers based on the given
 	/// [`LiquidityClientConfig`] and [`LiquidityServiceConfig`].
-	pub fn new_with_custom_time_provider(
+	pub async fn new_with_custom_time_provider(
 		entropy_source: ES, node_signer: NS, channel_manager: CM, chain_source: Option<C>,
 		chain_params: Option<ChainParameters>, kv_store: Arc<dyn KVStore + Send + Sync>,
 		service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>, time_provider: TP,
-	) -> Self {
+	) -> Result<Self, lightning::io::Error> {
 		let pending_messages = Arc::new(MessageQueue::new());
 		let pending_events = Arc::new(EventQueue::new(Arc::clone(&kv_store)));
 		let ignored_peers = RwLock::new(new_hash_set());
@@ -344,22 +352,30 @@ where
 				)
 			})
 		});
-		let lsps2_service_handler = service_config.as_ref().and_then(|config| {
-			config.lsps2_service_config.as_ref().map(|config| {
+
+		let lsps2_service_handler = if let Some(service_config) = service_config.as_ref() {
+			if let Some(lsps2_service_config) = service_config.lsps2_service_config.as_ref() {
 				if let Some(number) =
 					<LSPS2ServiceHandler<CM> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
 				{
 					supported_protocols.push(number);
 				}
-				LSPS2ServiceHandler::new(
+
+				let peer_states = read_lsps2_service_peer_states(Arc::clone(&kv_store)).await?;
+				Some(LSPS2ServiceHandler::new(
+					peer_states,
 					Arc::clone(&pending_messages),
 					Arc::clone(&pending_events),
 					channel_manager.clone(),
 					Arc::clone(&kv_store),
-					config.clone(),
-				)
-			})
-		});
+					lsps2_service_config.clone(),
+				))
+			} else {
+				None
+			}
+		} else {
+			None
+		};
 
 		let lsps5_client_handler = client_config.as_ref().and_then(|config| {
 			config.lsps5_client_config.as_ref().map(|config| {
@@ -434,7 +450,7 @@ where
 			None
 		};
 
-		Self {
+		Ok(Self {
 			pending_messages,
 			pending_events,
 			request_id_to_method_map: Mutex::new(new_hash_map()),
@@ -452,7 +468,7 @@ where
 			_client_config: client_config,
 			best_block: RwLock::new(chain_params.map(|chain_params| chain_params.best_block)),
 			_chain_source: chain_source,
-		}
+		})
 	}
 
 	/// Returns a reference to the LSPS0 client-side handler.
@@ -990,9 +1006,10 @@ where
 		chain_params: Option<ChainParameters>, kv_store_sync: Arc<dyn KVStoreSync + Send + Sync>,
 		service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>,
-	) -> Self {
+	) -> Result<Self, lightning::io::Error> {
 		let kv_store = Arc::new(KVStoreSyncWrapper(kv_store_sync));
-		let inner = Arc::new(LiquidityManager::new(
+
+		let mut fut = Box::pin(LiquidityManager::new(
 			entropy_source,
 			node_signer,
 			channel_manager,
@@ -1002,7 +1019,17 @@ where
 			service_config,
 			client_config,
 		));
-		Self { inner }
+
+		let mut waker = dummy_waker();
+		let mut ctx = task::Context::from_waker(&mut waker);
+		let inner = match fut.as_mut().poll(&mut ctx) {
+			task::Poll::Ready(result) => result,
+			task::Poll::Pending => {
+				// In a sync context, we can't wait for the future to complete.
+				unreachable!("LiquidityManager::new should not be pending in a sync context");
+			},
+		}?;
+		Ok(Self { inner: Arc::new(inner) })
 	}
 }
 
@@ -1028,9 +1055,9 @@ where
 		chain_params: Option<ChainParameters>, kv_store_sync: Arc<dyn KVStoreSync + Send + Sync>,
 		service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>, time_provider: TP,
-	) -> Self {
+	) -> Result<Self, lightning::io::Error> {
 		let kv_store = Arc::new(KVStoreSyncWrapper(kv_store_sync));
-		let inner = Arc::new(LiquidityManager::new_with_custom_time_provider(
+		let mut fut = Box::pin(LiquidityManager::new_with_custom_time_provider(
 			entropy_source,
 			node_signer,
 			channel_manager,
@@ -1041,7 +1068,17 @@ where
 			client_config,
 			time_provider,
 		));
-		Self { inner }
+
+		let mut waker = dummy_waker();
+		let mut ctx = task::Context::from_waker(&mut waker);
+		let inner = match fut.as_mut().poll(&mut ctx) {
+			task::Poll::Ready(result) => result,
+			task::Poll::Pending => {
+				// In a sync context, we can't wait for the future to complete.
+				unreachable!("LiquidityManager::new should not be pending in a sync context");
+			},
+		}?;
+		Ok(Self { inner: Arc::new(inner) })
 	}
 
 	/// Returns a reference to the LSPS0 client-side handler.

lightning-liquidity/src/persist.rs

@@ -9,6 +9,19 @@
 
 //! Types and utils for persistence.
 
+use crate::lsps2::service::PeerState as LSPS2ServicePeerState;
+
+use lightning::io::Cursor;
+use lightning::util::persist::KVStore;
+use lightning::util::ser::Readable;
+
+use bitcoin::secp256k1::PublicKey;
+
+use alloc::vec::Vec;
+
+use core::ops::Deref;
+use core::str::FromStr;
+
 /// The primary namespace under which the [`LiquidityManager`] will be persisted.
 ///
 /// [`LiquidityManager`]: crate::LiquidityManager
@@ -33,3 +46,47 @@ pub const LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps2_service";
 ///
 /// [`LSPS5ServiceHandler`]: crate::lsps5::service::LSPS5ServiceHandler
 pub const LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps5_service";
+
+pub(crate) async fn read_lsps2_service_peer_states<K: Deref>(
+	kv_store: K,
+) -> Result<Vec<(PublicKey, LSPS2ServicePeerState)>, lightning::io::Error>
+where
+	K::Target: KVStore,
+{
+	let mut res = Vec::new();
+
+	for stored_key in kv_store
+		.list(
+			LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+			LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+		)
+		.await?
+	{
+		let mut reader = Cursor::new(
+			kv_store
+				.read(
+					LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+					LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+					&stored_key,
+				)
+				.await?,
+		);
+
+		let peer_state = LSPS2ServicePeerState::read(&mut reader).map_err(|_| {
+			lightning::io::Error::new(
+				lightning::io::ErrorKind::InvalidData,
+				"Failed to deserialize LSPS2 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.push((key, peer_state));
+	}
+	Ok(res)
+}

lightning-liquidity/tests/common/mod.rs

@@ -38,7 +38,8 @@ pub(crate) fn create_service_and_client_nodes<'a, 'b, 'c>(
 		Some(service_config),
 		None,
 		Arc::clone(&time_provider),
-	);
+	)
+	.unwrap();
 
 	let client_kv_store = Arc::new(TestStore::new(false));
 	let client_lm = LiquidityManagerSync::new_with_custom_time_provider(
@@ -51,7 +52,8 @@ pub(crate) fn create_service_and_client_nodes<'a, 'b, 'c>(
 		None,
 		Some(client_config),
 		time_provider,
-	);
+	)
+	.unwrap();
 
 	let mut iter = nodes.into_iter();
 	let service_node = LiquidityNode::new(iter.next().unwrap(), service_lm);