Add tests for LSPS5 client and service.

Author: martinsaposnic

lightning-background-processor/src/lib.rs

@@ -647,6 +647,7 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// # use std::sync::atomic::{AtomicBool, Ordering};
 /// # use std::time::SystemTime;
 /// # use lightning_background_processor::{process_events_async, GossipSync};
+/// # use lightning_liquidity::lsps5::service::TimeProvider;
 /// # struct Logger {}
 /// # impl lightning::util::logger::Logger for Logger {
 /// #     fn log(&self, _record: lightning::util::logger::Record) {}
@@ -658,6 +659,16 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// #     fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) }
 /// #     fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
 /// # }
+/// #
+/// # use core::time::Duration;
+/// # struct DefaultTimeProvider;
+/// #
+/// # impl TimeProvider for DefaultTimeProvider {
+/// #    fn duration_since_epoch(&self) -> Duration {
+/// #        use std::time::{SystemTime, UNIX_EPOCH};
+/// #        SystemTime::now().duration_since(UNIX_EPOCH).expect("system time before Unix epoch")
+/// #    }
+/// # }
 /// # struct EventHandler {}
 /// # impl EventHandler {
 /// #     async fn handle_event(&self, _: lightning::events::Event) -> Result<(), ReplayEvent> { Ok(()) }
@@ -673,7 +684,7 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// # type P2PGossipSync<UL> = lightning::routing::gossip::P2PGossipSync<Arc<NetworkGraph>, Arc<UL>, Arc<Logger>>;
 /// # type ChannelManager<B, F, FE> = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor<B, F, FE>, B, FE, Logger>;
 /// # type OnionMessenger<B, F, FE> = lightning::onion_message::messenger::OnionMessenger<Arc<lightning::sign::KeysManager>, Arc<lightning::sign::KeysManager>, Arc<Logger>, Arc<ChannelManager<B, F, FE>>, Arc<lightning::onion_message::messenger::DefaultMessageRouter<Arc<NetworkGraph>, Arc<Logger>, Arc<lightning::sign::KeysManager>>>, Arc<ChannelManager<B, F, FE>>, lightning::ln::peer_handler::IgnoringMessageHandler, lightning::ln::peer_handler::IgnoringMessageHandler, lightning::ln::peer_handler::IgnoringMessageHandler>;
-/// # type LiquidityManager<B, F, FE> = lightning_liquidity::LiquidityManager<Arc<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>>;
+/// # type LiquidityManager<B, F, FE> = lightning_liquidity::LiquidityManager<Arc<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>, Arc<DefaultTimeProvider>>;
 /// # type Scorer = RwLock<lightning::routing::scoring::ProbabilisticScorer<Arc<NetworkGraph>, Arc<Logger>>>;
 /// # type PeerManager<B, F, FE, UL> = lightning::ln::peer_handler::SimpleArcPeerManager<SocketDescriptor, ChainMonitor<B, F, FE>, B, FE, Arc<UL>, Logger>;
 /// #
@@ -1285,8 +1296,12 @@ mod tests {
 		IgnoringMessageHandler,
 	>;
 
-	type LM =
-		LiquidityManager<Arc<KeysManager>, Arc<ChannelManager>, Arc<dyn Filter + Sync + Send>>;
+	type LM = LiquidityManager<
+		Arc<KeysManager>,
+		Arc<ChannelManager>,
+		Arc<dyn Filter + Sync + Send>,
+		Arc<DefaultTimeProvider>,
+	>;
 
 	struct Node {
 		node: Arc<ChannelManager>,

lightning-liquidity/src/lsps5/client.rs

@@ -570,3 +570,210 @@ where
 		self.handle_message(message, lsp_node_id)
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	#![cfg(all(test, feature = "time"))]
+	use core::time::Duration;
+
+	use super::*;
+	use crate::{
+		lsps0::ser::LSPSRequestId,
+		lsps5::{msgs::SetWebhookResponse, service::DefaultTimeProvider},
+		tests::utils::TestEntropy,
+	};
+	use bitcoin::{key::Secp256k1, secp256k1::SecretKey};
+
+	fn setup_test_client(
+		time_provider: Arc<dyn TimeProvider>,
+	) -> (
+		LSPS5ClientHandler<Arc<TestEntropy>, Arc<dyn TimeProvider>>,
+		Arc<MessageQueue>,
+		Arc<EventQueue>,
+		PublicKey,
+		PublicKey,
+	) {
+		let test_entropy_source = Arc::new(TestEntropy {});
+		let message_queue = Arc::new(MessageQueue::new());
+		let event_queue = Arc::new(EventQueue::new());
+		let client = LSPS5ClientHandler::new(
+			test_entropy_source,
+			message_queue.clone(),
+			event_queue.clone(),
+			LSPS5ClientConfig::default(),
+			time_provider,
+		);
+
+		let secp = Secp256k1::new();
+		let secret_key_1 = SecretKey::from_slice(&[42u8; 32]).unwrap();
+		let secret_key_2 = SecretKey::from_slice(&[43u8; 32]).unwrap();
+		let peer_1 = PublicKey::from_secret_key(&secp, &secret_key_1);
+		let peer_2 = PublicKey::from_secret_key(&secp, &secret_key_2);
+
+		(client, message_queue, event_queue, peer_1, peer_2)
+	}
+
+	#[test]
+	fn test_per_peer_state_isolation() {
+		let (client, _, _, peer_1, peer_2) = setup_test_client(Arc::new(DefaultTimeProvider));
+
+		let req_id_1 = client
+			.set_webhook(peer_1, "test-app-1".to_string(), "https://example.com/hook1".to_string())
+			.unwrap();
+		let req_id_2 = client
+			.set_webhook(peer_2, "test-app-2".to_string(), "https://example.com/hook2".to_string())
+			.unwrap();
+
+		{
+			let outer_state_lock = client.per_peer_state.read().unwrap();
+
+			let peer_1_state = outer_state_lock.get(&peer_1).unwrap().lock().unwrap();
+			assert!(peer_1_state.pending_set_webhook_requests.contains_key(&req_id_1));
+
+			let peer_2_state = outer_state_lock.get(&peer_2).unwrap().lock().unwrap();
+			assert!(peer_2_state.pending_set_webhook_requests.contains_key(&req_id_2));
+		}
+	}
+
+	#[test]
+	fn test_pending_request_tracking() {
+		let (client, _, _, peer, _) = setup_test_client(Arc::new(DefaultTimeProvider));
+		const APP_NAME: &str = "test-app";
+		const WEBHOOK_URL: &str = "https://example.com/hook";
+		let lsps5_app_name = LSPS5AppName::from_string(APP_NAME.to_string()).unwrap();
+		let lsps5_webhook_url = LSPS5WebhookUrl::from_string(WEBHOOK_URL.to_string()).unwrap();
+		let set_req_id =
+			client.set_webhook(peer, APP_NAME.to_string(), WEBHOOK_URL.to_string()).unwrap();
+		let list_req_id = client.list_webhooks(peer);
+		let remove_req_id = client.remove_webhook(peer, "test-app".to_string()).unwrap();
+
+		{
+			let outer_state_lock = client.per_peer_state.read().unwrap();
+			let peer_state = outer_state_lock.get(&peer).unwrap().lock().unwrap();
+			assert_eq!(
+				peer_state.pending_set_webhook_requests.get(&set_req_id).unwrap(),
+				&(
+					lsps5_app_name.clone(),
+					lsps5_webhook_url,
+					peer_state.pending_set_webhook_requests.get(&set_req_id).unwrap().2.clone()
+				)
+			);
+
+			assert!(peer_state.pending_list_webhooks_requests.contains_key(&list_req_id));
+
+			assert_eq!(
+				peer_state.pending_remove_webhook_requests.get(&remove_req_id).unwrap().0,
+				lsps5_app_name
+			);
+		}
+	}
+
+	#[test]
+	fn test_handle_response_clears_pending_state() {
+		let (client, _, _, peer, _) = setup_test_client(Arc::new(DefaultTimeProvider));
+
+		let req_id = client
+			.set_webhook(peer, "test-app".to_string(), "https://example.com/hook".to_string())
+			.unwrap();
+
+		let response = LSPS5Response::SetWebhook(SetWebhookResponse {
+			num_webhooks: 1,
+			max_webhooks: 5,
+			no_change: false,
+		});
+		let response_msg = LSPS5Message::Response(req_id.clone(), response);
+
+		{
+			let outer_state_lock = client.per_peer_state.read().unwrap();
+			let peer_state = outer_state_lock.get(&peer).unwrap().lock().unwrap();
+			assert!(peer_state.pending_set_webhook_requests.contains_key(&req_id));
+		}
+
+		client.handle_message(response_msg, &peer).unwrap();
+
+		{
+			let outer_state_lock = client.per_peer_state.read().unwrap();
+			let peer_state = outer_state_lock.get(&peer).unwrap().lock().unwrap();
+			assert!(!peer_state.pending_set_webhook_requests.contains_key(&req_id));
+		}
+	}
+
+	#[test]
+	fn test_cleanup_expired_responses() {
+		let (client, _, _, _, _) = setup_test_client(Arc::new(DefaultTimeProvider));
+		let time_provider = &client.time_provider;
+		const OLD_APP_NAME: &str = "test-app-old";
+		const NEW_APP_NAME: &str = "test-app-new";
+		const WEBHOOK_URL: &str = "https://example.com/hook";
+		let lsps5_old_app_name = LSPS5AppName::from_string(OLD_APP_NAME.to_string()).unwrap();
+		let lsps5_new_app_name = LSPS5AppName::from_string(NEW_APP_NAME.to_string()).unwrap();
+		let lsps5_webhook_url = LSPS5WebhookUrl::from_string(WEBHOOK_URL.to_string()).unwrap();
+		let now = time_provider.duration_since_epoch();
+		let mut peer_state = PeerState::new(Duration::from_secs(1800), time_provider.clone());
+		peer_state.last_cleanup = Some(LSPSDateTime::new_from_duration_since_epoch(
+			now.checked_sub(Duration::from_secs(120)).unwrap(),
+		));
+
+		let old_request_id = LSPSRequestId("test:request:old".to_string());
+		let new_request_id = LSPSRequestId("test:request:new".to_string());
+
+		// Add an old request (should be removed during cleanup)
+		peer_state.pending_set_webhook_requests.insert(
+			old_request_id.clone(),
+			(
+				lsps5_old_app_name,
+				lsps5_webhook_url.clone(),
+				LSPSDateTime::new_from_duration_since_epoch(
+					now.checked_sub(Duration::from_secs(7200)).unwrap(),
+				),
+			), // 2 hours old
+		);
+
+		// Add a recent request (should be kept)
+		peer_state.pending_set_webhook_requests.insert(
+			new_request_id.clone(),
+			(
+				lsps5_new_app_name,
+				lsps5_webhook_url,
+				LSPSDateTime::new_from_duration_since_epoch(
+					now.checked_sub(Duration::from_secs(600)).unwrap(),
+				),
+			), // 10 minutes old
+		);
+
+		peer_state.cleanup_expired_responses();
+
+		assert!(!peer_state.pending_set_webhook_requests.contains_key(&old_request_id));
+		assert!(peer_state.pending_set_webhook_requests.contains_key(&new_request_id));
+
+		let cleanup_age = if let Some(last_cleanup) = peer_state.last_cleanup {
+			LSPSDateTime::new_from_duration_since_epoch(time_provider.duration_since_epoch())
+				.abs_diff(last_cleanup)
+		} else {
+			0
+		};
+		assert!(cleanup_age < 10);
+	}
+
+	#[test]
+	fn test_unknown_request_id_handling() {
+		let (client, _message_queue, _, peer, _) = setup_test_client(Arc::new(DefaultTimeProvider));
+
+		let _valid_req = client
+			.set_webhook(peer, "test-app".to_string(), "https://example.com/hook".to_string())
+			.unwrap();
+
+		let unknown_req_id = LSPSRequestId("unknown:request:id".to_string());
+		let response = LSPS5Response::SetWebhook(SetWebhookResponse {
+			num_webhooks: 1,
+			max_webhooks: 5,
+			no_change: false,
+		});
+		let response_msg = LSPS5Message::Response(unknown_req_id, response);
+
+		let result = client.handle_message(response_msg, &peer);
+		assert!(result.is_err());
+		let error = result.unwrap_err();
+		assert!(error.err.to_lowercase().contains("unknown request id"));
+	}
+}

lightning-liquidity/tests/common/mod.rs

@@ -1,16 +1,16 @@
-#![cfg(test)]
+#![cfg(all(test, feature = "time"))]
 // TODO: remove these flags and unused code once we know what we'll need.
 #![allow(dead_code)]
 #![allow(unused_imports)]
 #![allow(unused_macros)]
 
-use lightning::chain::Filter;
-use lightning::sign::EntropySource;
-
 use bitcoin::blockdata::constants::{genesis_block, ChainHash};
 use bitcoin::blockdata::transaction::Transaction;
+use bitcoin::secp256k1::SecretKey;
 use bitcoin::Network;
+
 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
+use lightning::chain::Filter;
 use lightning::chain::{chainmonitor, BestBlock, Confirm};
 use lightning::ln::channelmanager;
 use lightning::ln::channelmanager::ChainParameters;
@@ -24,6 +24,7 @@ use lightning::onion_message::messenger::DefaultMessageRouter;
 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 use lightning::routing::router::{CandidateRouteHop, DefaultRouter, Path};
 use lightning::routing::scoring::{ChannelUsage, ScoreLookUp, ScoreUpdate};
+use lightning::sign::EntropySource;
 use lightning::sign::{InMemorySigner, KeysManager};
 use lightning::util::config::UserConfig;
 use lightning::util::persist::{
@@ -34,10 +35,13 @@ use lightning::util::persist::{
 	SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use lightning::util::test_utils;
+
+use lightning_liquidity::lsps5::service::TimeProvider;
 use lightning_liquidity::{LiquidityClientConfig, LiquidityManager, LiquidityServiceConfig};
 use lightning_persister::fs_store::FilesystemStore;
 
 use std::collections::{HashMap, VecDeque};
+use std::ops::Deref;
 use std::path::PathBuf;
 use std::sync::atomic::AtomicBool;
 use std::sync::mpsc::SyncSender;
@@ -67,7 +71,7 @@ type LockingWrapper<T> = lightning::routing::scoring::MultiThreadedLockableScore
 #[cfg(not(c_bindings))]
 type LockingWrapper<T> = std::sync::Mutex<T>;
 
-type ChannelManager = channelmanager::ChannelManager<
+pub(crate) type ChannelManager = channelmanager::ChannelManager<
 	Arc<ChainMonitor>,
 	Arc<test_utils::TestBroadcaster>,
 	Arc<KeysManager>,
@@ -127,13 +131,20 @@ pub(crate) struct Node {
 					Arc<KeysManager>,
 					Arc<ChannelManager>,
 					Arc<dyn Filter + Send + Sync>,
+					Arc<dyn TimeProvider>,
 				>,
 			>,
 			Arc<KeysManager>,
 		>,
 	>,
-	pub(crate) liquidity_manager:
-		Arc<LiquidityManager<Arc<KeysManager>, Arc<ChannelManager>, Arc<dyn Filter + Send + Sync>>>,
+	pub(crate) liquidity_manager: Arc<
+		LiquidityManager<
+			Arc<KeysManager>,
+			Arc<ChannelManager>,
+			Arc<dyn Filter + Send + Sync>,
+			Arc<dyn TimeProvider>,
+		>,
+	>,
 	pub(crate) chain_monitor: Arc<ChainMonitor>,
 	pub(crate) kv_store: Arc<FilesystemStore>,
 	pub(crate) tx_broadcaster: Arc<test_utils::TestBroadcaster>,
@@ -400,7 +411,7 @@ fn get_full_filepath(filepath: String, filename: String) -> String {
 
 pub(crate) fn create_liquidity_node(
 	i: usize, persist_dir: &str, network: Network, service_config: Option<LiquidityServiceConfig>,
-	client_config: Option<LiquidityClientConfig>,
+	client_config: Option<LiquidityClientConfig>, time_provider: Arc<dyn TimeProvider>,
 ) -> Node {
 	let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
 	let fee_estimator = Arc::new(test_utils::TestFeeEstimator::new(253));
@@ -451,16 +462,16 @@ pub(crate) fn create_liquidity_node(
 		Some(chain_source.clone()),
 		logger.clone(),
 	));
-
-	let liquidity_manager = Arc::new(LiquidityManager::new(
-		Arc::clone(&keys_manager),
-		Arc::clone(&channel_manager),
+	let liquidity_manager = Arc::new(LiquidityManager::new_with_custom_time_provider(
+		keys_manager.clone(),
+		channel_manager.clone(),
 		None::<Arc<dyn Filter + Send + Sync>>,
-		Some(chain_params),
+		Some(chain_params.clone()),
 		service_config,
 		client_config,
-		None,
+		time_provider,
 	));
+
 	let msg_handler = MessageHandler {
 		chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(
 			ChainHash::using_genesis_block(Network::Testnet),
@@ -489,14 +500,29 @@ pub(crate) fn create_liquidity_node(
 }
 
 pub(crate) fn create_service_and_client_nodes(
-	persist_dir: &str, service_config: LiquidityServiceConfig, client_config: LiquidityClientConfig,
+	persist_dir: &str, service_config: LiquidityServiceConfig,
+	client_config: LiquidityClientConfig, time_provider: Arc<dyn TimeProvider>,
 ) -> (Node, Node) {
 	let persist_temp_path = env::temp_dir().join(persist_dir);
 	let persist_dir = persist_temp_path.to_string_lossy().to_string();
 	let network = Network::Bitcoin;
 
-	let service_node = create_liquidity_node(1, &persist_dir, network, Some(service_config), None);
-	let client_node = create_liquidity_node(2, &persist_dir, network, None, Some(client_config));
+	let service_node = create_liquidity_node(
+		1,
+		&persist_dir,
+		network,
+		Some(service_config),
+		None,
+		time_provider.clone(),
+	);
+	let client_node = create_liquidity_node(
+		2,
+		&persist_dir,
+		network,
+		None,
+		Some(client_config),
+		time_provider.clone(),
+	);
 
 	service_node
 		.channel_manager