Let LiquidityManager take a KVStore and add Sync wrapper

We add `KVStore` to `LiquidityManager`, which will be used in the next
commits. We also add a `LiquidityManagerSync` wrapper that wraps a the
`LiquidityManager` interface which will soon become async due to usage
of the async `KVStore`.

Author: tnull

fuzz/src/lsps_message.rs

@@ -21,7 +21,7 @@ use lightning::util::test_utils::{
 };
 
 use lightning_liquidity::lsps0::ser::LSPS_MESSAGE_TYPE_ID;
-use lightning_liquidity::LiquidityManager;
+use lightning_liquidity::LiquidityManagerSync;
 
 use core::time::Duration;
 
@@ -77,12 +77,13 @@ pub fn do_test(data: &[u8]) {
 		genesis_block.header.time,
 	));
 
-	let liquidity_manager = Arc::new(LiquidityManager::new(
+	let liquidity_manager = Arc::new(LiquidityManagerSync::new(
 		Arc::clone(&keys_manager),
 		Arc::clone(&keys_manager),
 		Arc::clone(&manager),
 		None::<Arc<dyn Filter + Send + Sync>>,
 		None,
+		kv_store,
 		None,
 		None,
 	));

lightning-background-processor/Cargo.toml

@@ -31,6 +31,7 @@ possiblyrandom = { version = "0.2", path = "../possiblyrandom", default-features
 tokio = { version = "1.35", features = [ "macros", "rt", "rt-multi-thread", "sync", "time" ] }
 lightning = { version = "0.2.0", path = "../lightning", features = ["_test_utils"] }
 lightning-invoice = { version = "0.34.0", path = "../lightning-invoice" }
+lightning-liquidity = { version = "0.2.0", path = "../lightning-liquidity", default-features = false, features = ["_test_utils"] }
 lightning-persister = { version = "0.2.0", path = "../lightning-persister" }
 
 [lints]

lightning-background-processor/src/lib.rs

@@ -68,7 +68,7 @@ use lightning::util::sweep::OutputSweeperSync;
 use lightning::util::wakers::Sleeper;
 use lightning_rapid_gossip_sync::RapidGossipSync;
 
-use lightning_liquidity::ALiquidityManager;
+use lightning_liquidity::{ALiquidityManager, ALiquidityManagerSync};
 
 use core::ops::Deref;
 use core::time::Duration;
@@ -1350,7 +1350,7 @@ impl BackgroundProcessor {
 		CM::Target: AChannelManager,
 		OM::Target: AOnionMessenger,
 		PM::Target: APeerManager,
-		LM::Target: ALiquidityManager,
+		LM::Target: ALiquidityManagerSync,
 		D::Target: ChangeDestinationSourceSync,
 		O::Target: 'static + OutputSpender,
 		K::Target: 'static + KVStoreSync,
@@ -1693,7 +1693,7 @@ mod tests {
 	use lightning::util::test_utils;
 	use lightning::{get_event, get_event_msg};
 	use lightning_liquidity::utils::time::DefaultTimeProvider;
-	use lightning_liquidity::LiquidityManager;
+	use lightning_liquidity::{ALiquidityManagerSync, LiquidityManagerSync};
 	use lightning_persister::fs_store::FilesystemStore;
 	use lightning_rapid_gossip_sync::RapidGossipSync;
 	use std::collections::VecDeque;
@@ -1790,7 +1790,7 @@ mod tests {
 		IgnoringMessageHandler,
 	>;
 
-	type LM = LiquidityManager<
+	type LM = LiquidityManagerSync<
 		Arc<KeysManager>,
 		Arc<KeysManager>,
 		Arc<ChannelManager>,
@@ -2242,12 +2242,13 @@ mod tests {
 				Arc::clone(&logger),
 				Arc::clone(&keys_manager),
 			));
-			let liquidity_manager = Arc::new(LiquidityManager::new(
+			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,
 			));
@@ -2627,7 +2628,7 @@ mod tests {
 			Some(Arc::clone(&nodes[0].messenger)),
 			nodes[0].rapid_gossip_sync(),
 			Arc::clone(&nodes[0].peer_manager),
-			Some(Arc::clone(&nodes[0].liquidity_manager)),
+			Some(nodes[0].liquidity_manager.get_lm_async()),
 			Some(nodes[0].sweeper.sweeper_async()),
 			Arc::clone(&nodes[0].logger),
 			Some(Arc::clone(&nodes[0].scorer)),
@@ -3136,7 +3137,7 @@ mod tests {
 			Some(Arc::clone(&nodes[0].messenger)),
 			nodes[0].rapid_gossip_sync(),
 			Arc::clone(&nodes[0].peer_manager),
-			Some(Arc::clone(&nodes[0].liquidity_manager)),
+			Some(nodes[0].liquidity_manager.get_lm_async()),
 			Some(nodes[0].sweeper.sweeper_async()),
 			Arc::clone(&nodes[0].logger),
 			Some(Arc::clone(&nodes[0].scorer)),
@@ -3351,7 +3352,7 @@ mod tests {
 			Some(Arc::clone(&nodes[0].messenger)),
 			nodes[0].no_gossip_sync(),
 			Arc::clone(&nodes[0].peer_manager),
-			Some(Arc::clone(&nodes[0].liquidity_manager)),
+			Some(nodes[0].liquidity_manager.get_lm_async()),
 			Some(nodes[0].sweeper.sweeper_async()),
 			Arc::clone(&nodes[0].logger),
 			Some(Arc::clone(&nodes[0].scorer)),

lightning-liquidity/Cargo.toml

@@ -18,6 +18,7 @@ default = ["std", "time"]
 std = ["lightning/std"]
 time = ["std"]
 backtrace = ["dep:backtrace"]
+_test_utils = []
 
 [dependencies]
 lightning = { version = "0.2.0", path = "../lightning", default-features = false }

lightning-liquidity/src/lib.rs

@@ -73,5 +73,6 @@ mod tests;
 pub mod utils;
 
 pub use manager::{
-	ALiquidityManager, LiquidityClientConfig, LiquidityManager, LiquidityServiceConfig,
+	ALiquidityManager, ALiquidityManagerSync, LiquidityClientConfig, LiquidityManager,
+	LiquidityManagerSync, LiquidityServiceConfig,
 };