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

@@ -2370,16 +2370,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),
-				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),
+					None,
+					None,
+				)
+				.unwrap(),
+			);
 			let node = Node {
 				node: manager,
 				p2p_gossip_sync,

lightning-liquidity/src/lsps2/service.rs

@@ -465,7 +465,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>,
@@ -592,20 +592,48 @@ where
 {
 	/// Constructs a `LSPS2ServiceHandler`.
 	pub(crate) fn new(
-		pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue<K>>,
-		channel_manager: CM, kv_store: K, config: LSPS2ServiceConfig,
-	) -> Self {
-		Self {
+		per_peer_state: HashMap<PublicKey, Mutex<PeerState>>, pending_messages: Arc<MessageQueue>,
+		pending_events: Arc<EventQueue<K>>, channel_manager: CM, kv_store: K,
+		config: LSPS2ServiceConfig,
+	) -> Result<Self, lightning::io::Error> {
+		let mut peer_by_intercept_scid = new_hash_map();
+		let mut peer_by_channel_id = new_hash_map();
+		for (node_id, peer_state) in per_peer_state.iter() {
+			let peer_state_lock = peer_state.lock().unwrap();
+			for (intercept_scid, _) in peer_state_lock.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");
+				if res.is_some() {
+					return Err(lightning::io::Error::new(
+						lightning::io::ErrorKind::InvalidData,
+						"Failed to read LSPS2 peer state due to data inconsistencies: Intercept SCIDs should never collide",
+					));
+				}
+			}
+
+			for (channel_id, _) in peer_state_lock.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");
+				if res.is_some() {
+					return Err(lightning::io::Error::new(
+							lightning::io::ErrorKind::InvalidData,
+							"Failed to read LSPS2 peer state due to data inconsistencies: Channel IDs should never collide",
+					));
+				}
+			}
+		}
+
+		Ok(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,
 			config,
-		}
+		})
 	}
 
 	/// Returns a reference to the used config.

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;
@@ -327,12 +328,14 @@ where
 	K::Target: KVStore,
 {
 	/// 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: K,
 		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,
@@ -345,6 +348,7 @@ where
 			client_config,
 			time_provider,
 		)
+		.await
 	}
 }
 
@@ -366,16 +370,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: K,
 		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(kv_store.clone()));
 		let ignored_peers = RwLock::new(new_hash_set());
@@ -392,22 +398,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, K> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
 				{
 					supported_protocols.push(number);
 				}
-				LSPS2ServiceHandler::new(
+
+				let peer_states = read_lsps2_service_peer_states(kv_store.clone()).await?;
+				Some(LSPS2ServiceHandler::new(
+					peer_states,
 					Arc::clone(&pending_messages),
 					Arc::clone(&pending_events),
 					channel_manager.clone(),
 					kv_store.clone(),
-					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| {
@@ -482,7 +496,7 @@ where
 			None
 		};
 
-		Self {
+		Ok(Self {
 			pending_messages,
 			pending_events,
 			request_id_to_method_map: Mutex::new(new_hash_map()),
@@ -500,7 +514,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.
@@ -1038,9 +1052,10 @@ where
 		chain_params: Option<ChainParameters>, kv_store_sync: KS,
 		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,
@@ -1050,7 +1065,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) })
 	}
 }
 
@@ -1078,9 +1103,9 @@ where
 		chain_params: Option<ChainParameters>, kv_store_sync: KS,
 		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,
@@ -1091,7 +1116,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 crate::prelude::{new_hash_map, HashMap};
+use crate::sync::Mutex;
+
+use lightning::io::Cursor;
+use lightning::util::persist::KVStore;
+use lightning::util::ser::Readable;
+
+use bitcoin::secp256k1::PublicKey;
+
+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<HashMap<PublicKey, Mutex<LSPS2ServicePeerState>>, 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,
+			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.insert(key, Mutex::new(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);