Have Node take a genric KVStore parameter

Previously, we had `Node` take a concrete `FilesystemStore` struct. Here
we switch to a generic `KVStore` type parameter. To this end we switched
from `K: Deref` to concrete `Arc<K>` in all of the modules to avoid
confusion of the type paramters or requirements to track `Arc`-ed and
non-`Arc`ed version of the `K: KVStore` paramter. Moreover, as Uniffi
doesn't support exposing generics we now expose a concretized `LDKNode`
type alias in bindings, which will use `SqliteStore` in the future.

Note that going the generic route was necessary as `dyn KVStore`
wasn't an opion due to the incompatibility of `dyn` with associated
types. In this case the incompatibility is in regard to
`KVStore::Reader` and I opted to go this route over forcing any
implementation of the trait to allocating and returning the same
concrete value (e.g., `Vec<u8>`) which could potentially have
significant performance impacts over returning a buffered reader for
example.

Author: tnull

bindings/ldk_node.udl

@@ -13,10 +13,10 @@ interface Builder {
 	constructor();
 	[Name=from_config]
 	constructor(Config config);
-	Node build();
+	LDKNode build();
 };
 
-interface Node {
+interface LDKNode {
 	[Throws=NodeError]
 	void start();
 	[Throws=NodeError]

src/event.rs

@@ -106,23 +106,21 @@ impl_writeable_tlv_based_enum!(Event,
 	};
 );
 
-pub struct EventQueue<K: Deref, L: Deref>
+pub struct EventQueue<K: KVStore + Sync + Send, L: Deref>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	queue: Mutex<VecDeque<Event>>,
 	notifier: Condvar,
-	kv_store: K,
+	kv_store: Arc<K>,
 	logger: L,
 }
 
-impl<K: Deref, L: Deref> EventQueue<K, L>
+impl<K: KVStore + Sync + Send, L: Deref> EventQueue<K, L>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
-	pub(crate) fn new(kv_store: K, logger: L) -> Self {
+	pub(crate) fn new(kv_store: Arc<K>, logger: L) -> Self {
 		let queue: Mutex<VecDeque<Event>> = Mutex::new(VecDeque::new());
 		let notifier = Condvar::new();
 		Self { queue, notifier, kv_store, logger }
@@ -173,14 +171,13 @@ where
 	}
 }
 
-impl<K: Deref, L: Deref> ReadableArgs<(K, L)> for EventQueue<K, L>
+impl<K: KVStore + Sync + Send, L: Deref> ReadableArgs<(Arc<K>, L)> for EventQueue<K, L>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	#[inline]
 	fn read<R: lightning::io::Read>(
-		reader: &mut R, args: (K, L),
+		reader: &mut R, args: (Arc<K>, L),
 	) -> Result<Self, lightning::ln::msgs::DecodeError> {
 		let (kv_store, logger) = args;
 		let read_queue: EventQueueDeserWrapper = Readable::read(reader)?;
@@ -217,14 +214,13 @@ impl Writeable for EventQueueSerWrapper<'_> {
 	}
 }
 
-pub(crate) struct EventHandler<K: Deref + Clone, L: Deref>
+pub(crate) struct EventHandler<K: KVStore + Sync + Send, L: Deref>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	wallet: Arc<Wallet<bdk::database::SqliteDatabase>>,
 	event_queue: Arc<EventQueue<K, L>>,
-	channel_manager: Arc<ChannelManager>,
+	channel_manager: Arc<ChannelManager<K>>,
 	network_graph: Arc<NetworkGraph>,
 	keys_manager: Arc<KeysManager>,
 	payment_store: Arc<PaymentStore<K, L>>,
@@ -233,14 +229,13 @@ where
 	_config: Arc<Config>,
 }
 
-impl<K: Deref + Clone, L: Deref> EventHandler<K, L>
+impl<K: KVStore + Sync + Send + 'static, L: Deref> EventHandler<K, L>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	pub fn new(
 		wallet: Arc<Wallet<bdk::database::SqliteDatabase>>, event_queue: Arc<EventQueue<K, L>>,
-		channel_manager: Arc<ChannelManager>, network_graph: Arc<NetworkGraph>,
+		channel_manager: Arc<ChannelManager<K>>, network_graph: Arc<NetworkGraph>,
 		keys_manager: Arc<KeysManager>, payment_store: Arc<PaymentStore<K, L>>,
 		runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>, logger: L, _config: Arc<Config>,
 	) -> Self {

src/io/fs_store.rs

@@ -38,13 +38,14 @@ fn path_to_windows_str<T: AsRef<OsStr>>(path: T) -> Vec<winapi::shared::ntdef::W
 	path.as_ref().encode_wide().chain(Some(0)).collect()
 }
 
+/// A [`KVStore`] implementation that writes to and reads from the file system.
 pub struct FilesystemStore {
 	dest_dir: PathBuf,
 	locks: Mutex<HashMap<(String, String), Arc<RwLock<()>>>>,
 }
 
 impl FilesystemStore {
-	pub fn new(dest_dir: PathBuf) -> Self {
+	pub(crate) fn new(dest_dir: PathBuf) -> Self {
 		let locks = Mutex::new(HashMap::new());
 		Self { dest_dir, locks }
 	}

src/io/mod.rs

@@ -1,6 +1,8 @@
 pub(crate) mod fs_store;
 pub(crate) mod utils;
 
+pub use fs_store::FilesystemStore;
+
 use lightning::util::persist::KVStorePersister;
 
 use std::io::Read;

src/io/utils.rs

@@ -20,6 +20,7 @@ use std::fs;
 use std::io::Write;
 use std::ops::Deref;
 use std::path::Path;
+use std::sync::Arc;
 
 use super::KVStore;
 
@@ -46,11 +47,10 @@ pub(crate) fn read_or_generate_seed_file(keys_seed_path: &str) -> [u8; WALLET_KE
 }
 
 /// Read previously persisted [`ChannelMonitor`]s from the store.
-pub(crate) fn read_channel_monitors<K: Deref, ES: Deref, SP: Deref>(
-	kv_store: K, entropy_source: ES, signer_provider: SP,
+pub(crate) fn read_channel_monitors<K: KVStore + Sync + Send, ES: Deref, SP: Deref>(
+	kv_store: Arc<K>, entropy_source: ES, signer_provider: SP,
 ) -> std::io::Result<Vec<(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::Signer>)>>
 where
-	K::Target: KVStore,
 	ES::Target: EntropySource + Sized,
 	SP::Target: SignerProvider + Sized,
 {
@@ -92,11 +92,10 @@ where
 }
 
 /// Read a previously persisted [`NetworkGraph`] from the store.
-pub(crate) fn read_network_graph<K: Deref, L: Deref>(
-	kv_store: K, logger: L,
+pub(crate) fn read_network_graph<K: KVStore + Sync + Send, L: Deref>(
+	kv_store: Arc<K>, logger: L,
 ) -> Result<NetworkGraph<L>, std::io::Error>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	let mut reader =
@@ -108,11 +107,10 @@ where
 }
 
 /// Read a previously persisted [`Scorer`] from the store.
-pub(crate) fn read_scorer<K: Deref, G: Deref<Target = NetworkGraph<L>>, L: Deref>(
-	kv_store: K, network_graph: G, logger: L,
+pub(crate) fn read_scorer<K: KVStore + Sync + Send, G: Deref<Target = NetworkGraph<L>>, L: Deref>(
+	kv_store: Arc<K>, network_graph: G, logger: L,
 ) -> Result<ProbabilisticScorer<G, L>, std::io::Error>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	let params = ProbabilisticScoringParameters::default();
@@ -125,11 +123,10 @@ where
 }
 
 /// Read previously persisted events from the store.
-pub(crate) fn read_event_queue<K: Deref, L: Deref>(
-	kv_store: K, logger: L,
+pub(crate) fn read_event_queue<K: KVStore + Sync + Send, L: Deref>(
+	kv_store: Arc<K>, logger: L,
 ) -> Result<EventQueue<K, L>, std::io::Error>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	let mut reader =
@@ -141,11 +138,10 @@ where
 }
 
 /// Read previously persisted peer info from the store.
-pub(crate) fn read_peer_info<K: Deref, L: Deref>(
-	kv_store: K, logger: L,
+pub(crate) fn read_peer_info<K: KVStore + Sync + Send, L: Deref>(
+	kv_store: Arc<K>, logger: L,
 ) -> Result<PeerStore<K, L>, std::io::Error>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	let mut reader = kv_store.read(PEER_INFO_PERSISTENCE_NAMESPACE, PEER_INFO_PERSISTENCE_KEY)?;
@@ -156,10 +152,9 @@ where
 }
 
 /// Read previously persisted payments information from the store.
-pub(crate) fn read_payments<K: Deref>(kv_store: K) -> Result<Vec<PaymentDetails>, std::io::Error>
-where
-	K::Target: KVStore,
-{
+pub(crate) fn read_payments<K: KVStore + Sync + Send>(
+	kv_store: Arc<K>,
+) -> Result<Vec<PaymentDetails>, std::io::Error> {
 	let mut res = Vec::new();
 
 	for stored_key in kv_store.list(PAYMENT_INFO_PERSISTENCE_NAMESPACE)? {
@@ -174,10 +169,9 @@ where
 	Ok(res)
 }
 
-pub(crate) fn read_rgs_latest_sync_timestamp<K: Deref>(kv_store: K) -> Result<u32, std::io::Error>
-where
-	K::Target: KVStore,
-{
+pub(crate) fn read_rgs_latest_sync_timestamp<K: KVStore + Sync + Send>(
+	kv_store: Arc<K>,
+) -> Result<u32, std::io::Error> {
 	let mut reader =
 		kv_store.read(RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE, RGS_LATEST_SYNC_TIMESTAMP_KEY)?;
 	u32::read(&mut reader).map_err(|_| {
@@ -188,24 +182,24 @@ where
 	})
 }
 
-pub(crate) fn write_rgs_latest_sync_timestamp<K: Deref, L: Deref>(
-	updated_timestamp: u32, kv_store: K, logger: L,
+pub(crate) fn write_rgs_latest_sync_timestamp<K: KVStore + Sync + Send, L: Deref>(
+	updated_timestamp: u32, kv_store: Arc<K>, logger: L,
 ) -> Result<(), Error>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
 	let data = updated_timestamp.encode();
-	kv_store.write(RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE, RGS_LATEST_SYNC_TIMESTAMP_KEY, &data)
+	kv_store
+		.write(RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE, RGS_LATEST_SYNC_TIMESTAMP_KEY, &data)
 		.map_err(|e| {
-		log_error!(
-			logger,
-			"Writing data to key {}/{} failed due to: {}",
-			RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE,
-			RGS_LATEST_SYNC_TIMESTAMP_KEY,
-			e
-		);
-		Error::PersistenceFailed
-	})?;
+			log_error!(
+				logger,
+				"Writing data to key {}/{} failed due to: {}",
+				RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE,
+				RGS_LATEST_SYNC_TIMESTAMP_KEY,
+				e
+			);
+			Error::PersistenceFailed
+		})?;
 	Ok(())
 }

src/lib.rs

@@ -78,7 +78,7 @@ mod error;
 mod event;
 mod gossip;
 mod hex_utils;
-mod io;
+pub mod io;
 mod logger;
 mod payment_store;
 mod peer_store;
@@ -312,8 +312,18 @@ impl Builder {
 		self
 	}
 
+	/// Builds a [`Node`] instance with a [`FilesystemStore`] backend and according to the options
+	/// previously configured.
+	pub fn build(&self) -> Arc<Node<FilesystemStore>> {
+		let ldk_data_dir = format!("{}/ldk", self.config.storage_dir_path);
+		let kv_store = Arc::new(FilesystemStore::new(ldk_data_dir.clone().into()));
+		self.build_with_store(kv_store)
+	}
+
 	/// Builds a [`Node`] instance according to the options previously configured.
-	pub fn build(&self) -> Arc<Node> {
+	pub fn build_with_store<K: KVStore + Sync + Send + 'static>(
+		&self, kv_store: Arc<K>,
+	) -> Arc<Node<K>> {
 		let config = Arc::new(self.config.clone());
 
 		let ldk_data_dir = format!("{}/ldk", config.storage_dir_path);
@@ -384,10 +394,8 @@ impl Builder {
 			Arc::clone(&logger),
 		));
 
-		let kv_store = Arc::new(FilesystemStore::new(ldk_data_dir.clone().into()));
-
 		// Initialize the ChainMonitor
-		let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(
+		let chain_monitor: Arc<ChainMonitor<K>> = Arc::new(chainmonitor::ChainMonitor::new(
 			Some(Arc::clone(&tx_sync)),
 			Arc::clone(&wallet),
 			Arc::clone(&logger),
@@ -484,7 +492,7 @@ impl Builder {
 					channel_monitor_references,
 				);
 				let (_hash, channel_manager) =
-					<(BlockHash, ChannelManager)>::read(&mut reader, read_args)
+					<(BlockHash, ChannelManager<K>)>::read(&mut reader, read_args)
 						.expect("Failed to read channel manager from store");
 				channel_manager
 			} else {
@@ -653,30 +661,34 @@ impl Builder {
 	}
 }
 
+/// This type alias is required as Uniffi doesn't support generics, i.e., we can only expose the
+/// concretized types via this aliasing hack.
+type LDKNode = Node<FilesystemStore>;
+
 /// The main interface object of LDK Node, wrapping the necessary LDK and BDK functionalities.
 ///
 /// Needs to be initialized and instantiated through [`Builder::build`].
-pub struct Node {
+pub struct Node<K: KVStore + Sync + Send + 'static> {
 	runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
 	stop_running: Arc<AtomicBool>,
 	config: Arc<Config>,
 	wallet: Arc<Wallet<bdk::database::SqliteDatabase>>,
 	tx_sync: Arc<EsploraSyncClient<Arc<FilesystemLogger>>>,
-	event_queue: Arc<EventQueue<Arc<FilesystemStore>, Arc<FilesystemLogger>>>,
-	channel_manager: Arc<ChannelManager>,
-	chain_monitor: Arc<ChainMonitor>,
-	peer_manager: Arc<PeerManager>,
+	event_queue: Arc<EventQueue<K, Arc<FilesystemLogger>>>,
+	channel_manager: Arc<ChannelManager<K>>,
+	chain_monitor: Arc<ChainMonitor<K>>,
+	peer_manager: Arc<PeerManager<K>>,
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<NetworkGraph>,
 	gossip_source: Arc<GossipSource>,
-	kv_store: Arc<FilesystemStore>,
+	kv_store: Arc<K>,
 	logger: Arc<FilesystemLogger>,
 	scorer: Arc<Mutex<Scorer>>,
-	peer_store: Arc<PeerStore<Arc<FilesystemStore>, Arc<FilesystemLogger>>>,
-	payment_store: Arc<PaymentStore<Arc<FilesystemStore>, Arc<FilesystemLogger>>>,
+	peer_store: Arc<PeerStore<K, Arc<FilesystemLogger>>>,
+	payment_store: Arc<PaymentStore<K, Arc<FilesystemLogger>>>,
 }
 
-impl Node {
+impl<K: KVStore + Sync + Send + 'static> Node<K> {
 	/// Starts the necessary background tasks, such as handling events coming from user input,
 	/// LDK/BDK, and the peer-to-peer network.
 	///
@@ -1558,14 +1570,14 @@ impl Node {
 	}
 }
 
-impl Drop for Node {
+impl<K: KVStore + Sync + Send + 'static> Drop for Node<K> {
 	fn drop(&mut self) {
 		let _ = self.stop();
 	}
 }
 
-async fn connect_peer_if_necessary(
-	pubkey: PublicKey, peer_addr: SocketAddr, peer_manager: Arc<PeerManager>,
+async fn connect_peer_if_necessary<K: KVStore + Sync + Send + 'static + 'static>(
+	pubkey: PublicKey, peer_addr: SocketAddr, peer_manager: Arc<PeerManager<K>>,
 	logger: Arc<FilesystemLogger>,
 ) -> Result<(), Error> {
 	for (node_pubkey, _addr) in peer_manager.get_peer_node_ids() {
@@ -1577,8 +1589,8 @@ async fn connect_peer_if_necessary(
 	do_connect_peer(pubkey, peer_addr, peer_manager, logger).await
 }
 
-async fn do_connect_peer(
-	pubkey: PublicKey, peer_addr: SocketAddr, peer_manager: Arc<PeerManager>,
+async fn do_connect_peer<K: KVStore + Sync + Send + 'static + 'static>(
+	pubkey: PublicKey, peer_addr: SocketAddr, peer_manager: Arc<PeerManager<K>>,
 	logger: Arc<FilesystemLogger>,
 ) -> Result<(), Error> {
 	log_info!(logger, "Connecting to peer: {}@{}", pubkey, peer_addr);