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

@@ -69,6 +69,8 @@ use lightning::util::wakers::Sleeper;
 use lightning_rapid_gossip_sync::RapidGossipSync;
 
 use lightning_liquidity::ALiquidityManager;
+#[cfg(feature = "std")]
+use lightning_liquidity::ALiquidityManagerSync;
 
 use core::ops::Deref;
 use core::time::Duration;
@@ -424,6 +426,31 @@ pub const NO_LIQUIDITY_MANAGER: Option<
 				CM = &DynChannelManager,
 				Filter = dyn chain::Filter,
 				C = &dyn chain::Filter,
+				KVStore = dyn lightning::util::persist::KVStore,
+				K = &dyn lightning::util::persist::KVStore,
+				TimeProvider = dyn lightning_liquidity::utils::time::TimeProvider,
+				TP = &dyn lightning_liquidity::utils::time::TimeProvider,
+			> + Send
+			+ Sync,
+	>,
+> = None;
+
+/// When initializing a background processor without a liquidity manager, this can be used to avoid
+/// specifying a concrete `LiquidityManagerSync` type.
+#[cfg(all(not(c_bindings), feature = "std"))]
+pub const NO_LIQUIDITY_MANAGER_SYNC: Option<
+	Arc<
+		dyn ALiquidityManagerSync<
+				EntropySource = dyn EntropySource,
+				ES = &dyn EntropySource,
+				NodeSigner = dyn lightning::sign::NodeSigner,
+				NS = &dyn lightning::sign::NodeSigner,
+				AChannelManager = DynChannelManager,
+				CM = &DynChannelManager,
+				Filter = dyn chain::Filter,
+				C = &dyn chain::Filter,
+				KVStoreSync = dyn lightning::util::persist::KVStoreSync,
+				KS = &dyn lightning::util::persist::KVStoreSync,
 				TimeProvider = dyn lightning_liquidity::utils::time::TimeProvider,
 				TP = &dyn lightning_liquidity::utils::time::TimeProvider,
 			> + Send
@@ -731,7 +758,7 @@ use futures_util::{dummy_waker, Joiner, OptionalSelector, Selector, SelectorOutp
 /// # 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<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>, Arc<DefaultTimeProvider>>;
+/// # type LiquidityManager<B, F, FE> = lightning_liquidity::LiquidityManager<Arc<lightning::sign::KeysManager>, Arc<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>, Arc<Store>, 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, F, StoreSync>;
 /// # type OutputSweeper<B, D, FE, F, O> = lightning::util::sweep::OutputSweeper<Arc<B>, Arc<D>, Arc<FE>, Arc<F>, Arc<Store>, Arc<Logger>, Arc<O>>;
@@ -1450,7 +1477,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,
@@ -1793,7 +1820,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;
@@ -1890,11 +1917,12 @@ mod tests {
 		IgnoringMessageHandler,
 	>;
 
-	type LM = LiquidityManager<
+	type LM = LiquidityManagerSync<
 		Arc<KeysManager>,
 		Arc<KeysManager>,
 		Arc<ChannelManager>,
 		Arc<dyn Filter + Sync + Send>,
+		Arc<Persister>,
 		Arc<DefaultTimeProvider>,
 	>;
 
@@ -2342,12 +2370,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),
 				None,
 				None,
 			));
@@ -2727,7 +2756,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)),
@@ -3236,7 +3265,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)),
@@ -3451,7 +3480,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)),
@@ -3500,7 +3529,7 @@ mod tests {
 			crate::NO_ONION_MESSENGER,
 			nodes[0].no_gossip_sync(),
 			Arc::clone(&nodes[0].peer_manager),
-			crate::NO_LIQUIDITY_MANAGER,
+			crate::NO_LIQUIDITY_MANAGER_SYNC,
 			Some(Arc::clone(&nodes[0].sweeper)),
 			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,
 };