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 }
@@ -178,14 +176,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)?;
@@ -222,14 +219,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>>,
@@ -238,14 +234,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;
@@ -33,8 +35,8 @@ pub(crate) const LATEST_RGS_SYNC_TIMESTAMP_NAMESPACE: &str = "";
 pub(crate) const LATEST_RGS_SYNC_TIMESTAMP_KEY: &str = "latest_rgs_sync_timestamp";
 
 /// The last time we broadcast a node announcement will be persisted under this key.
-pub(crate) const LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE: &str = "";
-pub(crate) const LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY: &str = "latest_node_ann_bcast_timestamp";
+pub(crate) const LATEST_NODE_ANN_BCAST_TIMESTAMP_NAMESPACE: &str = "";
+pub(crate) const LATEST_NODE_ANN_BCAST_TIMESTAMP_KEY: &str = "latest_node_ann_bcast_timestamp";
 
 /// Provides an interface that allows to store and retrieve persisted values that are associated
 /// with given keys.

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_latest_rgs_sync_timestamp<K: Deref>(kv_store: K) -> Result<u32, std::io::Error>
-where
-	K::Target: KVStore,
-{
+pub(crate) fn read_latest_rgs_sync_timestamp<K: KVStore + Sync + Send>(
+	kv_store: Arc<K>,
+) -> Result<u32, std::io::Error> {
 	let mut reader =
 		kv_store.read(LATEST_RGS_SYNC_TIMESTAMP_NAMESPACE, LATEST_RGS_SYNC_TIMESTAMP_KEY)?;
 	u32::read(&mut reader).map_err(|_| {
@@ -188,11 +182,10 @@ where
 	})
 }
 
-pub(crate) fn write_latest_rgs_sync_timestamp<K: Deref, L: Deref>(
-	updated_timestamp: u32, kv_store: K, logger: L,
+pub(crate) fn write_latest_rgs_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();
@@ -211,14 +204,11 @@ where
 	Ok(())
 }
 
-pub(crate) fn read_latest_node_ann_bcast_timestamp<K: Deref>(
-	kv_store: K,
-) -> Result<u64, std::io::Error>
-where
-	K::Target: KVStore,
-{
+pub(crate) fn read_latest_node_ann_bcast_timestamp<K: KVStore + Sync + Send>(
+	kv_store: Arc<K>,
+) -> Result<u64, std::io::Error> {
 	let mut reader = kv_store
-		.read(LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE, LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY)?;
+		.read(LATEST_NODE_ANN_BCAST_TIMESTAMP_NAMESPACE, LATEST_NODE_ANN_BCAST_TIMESTAMP_KEY)?;
 	u64::read(&mut reader).map_err(|_| {
 		std::io::Error::new(
 			std::io::ErrorKind::InvalidData,
@@ -227,43 +217,28 @@ where
 	})
 }
 
-pub(crate) fn write_latest_node_ann_bcast_timestamp<K: Deref, L: Deref>(
-	updated_timestamp: u64, kv_store: K, logger: L,
+pub(crate) fn write_latest_node_ann_bcast_timestamp<K: KVStore + Sync + Send, L: Deref>(
+	updated_timestamp: u64, kv_store: Arc<K>, logger: L,
 ) -> Result<(), Error>
 where
-	K::Target: KVStore,
 	L::Target: Logger,
 {
-	let mut writer = kv_store
-		.write(LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE, LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY)
+	let data = updated_timestamp.encode();
+	kv_store
+		.write(
+			LATEST_NODE_ANN_BCAST_TIMESTAMP_NAMESPACE,
+			LATEST_NODE_ANN_BCAST_TIMESTAMP_KEY,
+			&data,
+		)
 		.map_err(|e| {
 			log_error!(
 				logger,
-				"Getting writer for key {}/{} failed due to: {}",
-				LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE,
-				LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY,
+				"Writing data to key {}/{} failed due to: {}",
+				LATEST_NODE_ANN_BCAST_TIMESTAMP_NAMESPACE,
+				LATEST_NODE_ANN_BCAST_TIMESTAMP_KEY,
 				e
 			);
 			Error::PersistenceFailed
 		})?;
-	updated_timestamp.write(&mut writer).map_err(|e| {
-		log_error!(
-			logger,
-			"Writing data to key {}/{} failed due to: {}",
-			LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE,
-			LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY,
-			e
-		);
-		Error::PersistenceFailed
-	})?;
-	writer.commit().map_err(|e| {
-		log_error!(
-			logger,
-			"Committing data to key {}/{} failed due to: {}",
-			LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE,
-			LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY,
-			e
-		);
-		Error::PersistenceFailed
-	})
+	Ok(())
 }