f Return trait objects from KVStore, update io::util deser methods

Author: tnull

src/event.rs

@@ -3,7 +3,7 @@ use crate::{
 	PaymentInfo, PaymentInfoStorage, PaymentStatus, Wallet,
 };
 
-use crate::io::KVStoreUnpersister;
+use crate::io::{KVStoreUnpersister, EVENT_QUEUE_PERSISTENCE_KEY};
 use crate::logger::{log_error, log_info, Logger};
 
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
@@ -24,9 +24,6 @@ use std::ops::Deref;
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
 
-/// The event queue will be persisted under this key.
-pub(crate) const EVENTS_PERSISTENCE_KEY: &str = "events";
-
 /// An event emitted by [`Node`], which should be handled by the user.
 ///
 /// [`Node`]: [`crate::Node`]
@@ -86,7 +83,7 @@ impl_writeable_tlv_based_enum!(Event,
 	};
 );
 
-pub(crate) struct EventQueue<K: Deref>
+pub struct EventQueue<K: Deref>
 where
 	K::Target: KVStorePersister,
 {
@@ -134,7 +131,7 @@ where
 
 	fn persist_queue(&self, locked_queue: &VecDeque<Event>) -> Result<(), Error> {
 		self.persister
-			.persist(EVENTS_PERSISTENCE_KEY, &EventQueueSerWrapper(locked_queue))
+			.persist(EVENT_QUEUE_PERSISTENCE_KEY, &EventQueueSerWrapper(locked_queue))
 			.map_err(|_| Error::PersistenceFailed)?;
 		Ok(())
 	}
@@ -572,7 +569,7 @@ mod tests {
 		}
 
 		// Check we can read back what we persisted.
-		let persisted_bytes = persister.get_persisted_bytes(EVENTS_PERSISTENCE_KEY).unwrap();
+		let persisted_bytes = persister.get_persisted_bytes(EVENT_QUEUE_PERSISTENCE_KEY).unwrap();
 		let deser_event_queue =
 			EventQueue::read(&mut &persisted_bytes[..], Arc::clone(&persister)).unwrap();
 		assert_eq!(deser_event_queue.next_event(), expected_event);

src/io/fs_store.rs

@@ -46,19 +46,19 @@ impl FilesystemStore {
 	}
 }
 
-impl KVStore<FilesystemReader, FilesystemWriter> for FilesystemStore {
-	fn read(&self, namespace: &str, key: &str) -> std::io::Result<FilesystemReader> {
+impl KVStore for FilesystemStore {
+	fn read(&self, namespace: &str, key: &str) -> std::io::Result<Box<dyn Read>> {
 		let mut dest_file = self.dest_dir.clone();
 		dest_file.push(namespace);
 		dest_file.push(key);
-		FilesystemReader::new(dest_file)
+		FilesystemReader::new(dest_file).map(|r| Box::new(r) as Box<dyn Read>)
 	}
 
-	fn write(&self, namespace: &str, key: &str) -> std::io::Result<FilesystemWriter> {
+	fn write(&self, namespace: &str, key: &str) -> std::io::Result<Box<dyn TransactionalWrite>> {
 		let mut dest_file = self.dest_dir.clone();
 		dest_file.push(namespace);
 		dest_file.push(key);
-		FilesystemWriter::new(dest_file)
+		FilesystemWriter::new(dest_file).map(|w| Box::new(w) as Box<dyn TransactionalWrite>)
 	}
 
 	fn remove(&self, namespace: &str, key: &str) -> std::io::Result<bool> {

src/io/mod.rs

@@ -1,22 +1,31 @@
 pub(crate) mod fs_store;
 pub(crate) mod utils;
 
-use lightning::chain::channelmonitor::ChannelMonitor;
-use lightning::chain::keysinterface::{EntropySource, SignerProvider};
-use lightning::util::ser::ReadableArgs;
-
-use bitcoin::hash_types::{BlockHash, Txid};
-use bitcoin::hashes::hex::FromHex;
-
 use std::io::{Read, Write};
-use std::ops::Deref;
 
-// The namespace and key LDK uses for persisting
-pub(crate) const CHANNEL_MANAGER_PERSISTENCE_KEY: &str = "manager";
+// The namespacs and keys LDK uses for persisting
 pub(crate) const CHANNEL_MANAGER_PERSISTENCE_NAMESPACE: &str = "";
+pub(crate) const CHANNEL_MANAGER_PERSISTENCE_KEY: &str = "manager";
 
 pub(crate) const CHANNEL_MONITOR_PERSISTENCE_NAMESPACE: &str = "monitors";
 
+pub(crate) const NETWORK_GRAPH_PERSISTENCE_NAMESPACE: &str = "";
+pub(crate) const NETWORK_GRAPH_PERSISTENCE_KEY: &str = "network_graph";
+
+pub(crate) const SCORER_PERSISTENCE_NAMESPACE: &str = "";
+pub(crate) const SCORER_PERSISTENCE_KEY: &str = "scorer";
+
+/// The event queue will be persisted under this key.
+pub(crate) const EVENT_QUEUE_PERSISTENCE_NAMESPACE: &str = "";
+pub(crate) const EVENT_QUEUE_PERSISTENCE_KEY: &str = "events";
+
+/// The peer information will be persisted under this key.
+pub(crate) const PEER_INFO_PERSISTENCE_NAMESPACE: &str = "";
+pub(crate) const PEER_INFO_PERSISTENCE_KEY: &str = "peers";
+
+/// The payment information will be persisted under this prefix.
+pub(crate) const PAYMENT_INFO_PERSISTENCE_NAMESPACE: &str = "payments";
+
 /// Provides an interface that allows to store and retrieve persisted values that are associated
 /// with given keys.
 ///
@@ -26,71 +35,24 @@ pub(crate) const CHANNEL_MONITOR_PERSISTENCE_NAMESPACE: &str = "monitors";
 ///
 /// Keys and namespaces are required to be valid ASCII strings and the empty namespace (`""`) is
 /// assumed to be valid namespace.
-pub trait KVStore<R: Read, TW: TransactionalWrite> {
+pub trait KVStore {
 	/// Returns a [`Read`] for the given `key` from which [`Readable`]s may be read.
 	///
 	/// [`Readable`]: lightning::util::ser::Readable
-	fn read(&self, namespace: &str, key: &str) -> std::io::Result<R>;
+	fn read(&self, namespace: &str, key: &str) -> std::io::Result<Box<dyn Read>>;
 	/// Returns a [`TransactionalWrite`] for the given `key` to which [`Writeable`]s may be written.
 	///
 	/// Note that [`TransactionalWrite::commit`] MUST be called to commit the written data, otherwise
 	/// the changes won't be persisted.
 	///
 	/// [`Writeable`]: lightning::util::ser::Writeable
-	fn write(&self, namespace: &str, key: &str) -> std::io::Result<TW>;
+	fn write(&self, namespace: &str, key: &str) -> std::io::Result<Box<dyn TransactionalWrite>>;
 	/// Removes any data that had previously been persisted under the given `key`.
 	///
 	/// Returns `true` if the key was present, and `false` otherwise.
 	fn remove(&self, namespace: &str, key: &str) -> std::io::Result<bool>;
 	/// Returns a list of keys that are stored under the given `namespace`.
 	fn list(&self, namespace: &str) -> std::io::Result<Vec<String>>;
-	/// Read the persisted [`ChannelMonitor`]s from the store.
-	fn read_channelmonitors<ES: Deref, SP: Deref>(
-		&self, entropy_source: ES, signer_provider: SP,
-	) -> std::io::Result<Vec<(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::Signer>)>>
-	where
-		ES::Target: EntropySource + Sized,
-		SP::Target: SignerProvider + Sized,
-	{
-		let mut res = Vec::new();
-
-		for stored_key in self.list(CHANNEL_MONITOR_PERSISTENCE_NAMESPACE)? {
-			let txid = Txid::from_hex(stored_key.split_at(64).0).map_err(|_| {
-				std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid tx ID in stored key")
-			})?;
-
-			let index: u16 = stored_key.split_at(65).1.parse().map_err(|_| {
-				std::io::Error::new(
-					std::io::ErrorKind::InvalidData,
-					"Invalid tx index in stored key",
-				)
-			})?;
-
-			match <(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::Signer>)>::read(
-				&mut self.read(CHANNEL_MONITOR_PERSISTENCE_NAMESPACE, &stored_key)?,
-				(&*entropy_source, &*signer_provider),
-			) {
-				Ok((block_hash, channel_monitor)) => {
-					if channel_monitor.get_funding_txo().0.txid != txid
-						|| channel_monitor.get_funding_txo().0.index != index
-					{
-						return Err(std::io::Error::new(
-							std::io::ErrorKind::InvalidData,
-							"ChannelMonitor was stored under the wrong key",
-						));
-					}
-					res.push((block_hash, channel_monitor));
-				}
-				Err(e) => {
-					return Err(std::io::Error::new(
-						std::io::ErrorKind::InvalidData,
-						format!("Failed to deserialize ChannelMonitor: {}", e),
-					))
-				}
-			}
-		}
-		Ok(res)
-	}
 }
 
 /// A [`Write`] asserting data consistency.

src/io/utils.rs

@@ -1,15 +1,27 @@
-use crate::payment_store::{PaymentInfo, PAYMENT_INFO_PERSISTENCE_NAMESPACE};
-use crate::{Config, FilesystemLogger, NetworkGraph, Scorer, WALLET_KEYS_SEED_LEN};
+use super::*;
+use crate::WALLET_KEYS_SEED_LEN;
 
+use crate::peer_store::PeerInfoStorage;
+use crate::{EventQueue, PaymentInfo};
+
+use lightning::chain::channelmonitor::ChannelMonitor;
+use lightning::chain::keysinterface::{EntropySource, SignerProvider};
+use lightning::routing::gossip::NetworkGraph;
 use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringParameters};
+use lightning::util::logger::Logger;
+use lightning::util::persist::KVStorePersister;
 use lightning::util::ser::{Readable, ReadableArgs};
 
+use bitcoin::hash_types::{BlockHash, Txid};
+use bitcoin::hashes::hex::FromHex;
 use rand::{thread_rng, RngCore};
 
 use std::fs;
-use std::io::{BufReader, Write};
+use std::io::Write;
+use std::ops::Deref;
 use std::path::Path;
-use std::sync::Arc;
+
+use super::KVStore;
 
 pub(crate) fn read_or_generate_seed_file(keys_seed_path: &str) -> [u8; WALLET_KEYS_SEED_LEN] {
 	if Path::new(&keys_seed_path).exists() {
@@ -33,53 +45,125 @@ pub(crate) fn read_or_generate_seed_file(keys_seed_path: &str) -> [u8; WALLET_KE
 	}
 }
 
-pub(crate) fn read_network_graph(config: &Config, logger: Arc<FilesystemLogger>) -> NetworkGraph {
-	let ldk_data_dir = format!("{}/ldk", config.storage_dir_path);
-	let network_graph_path = format!("{}/network_graph", ldk_data_dir);
-
-	if let Ok(file) = fs::File::open(network_graph_path) {
-		if let Ok(graph) = NetworkGraph::read(&mut BufReader::new(file), Arc::clone(&logger)) {
-			return graph;
-		}
-	}
-
-	NetworkGraph::new(config.network, logger)
-}
-
-pub(crate) fn read_scorer(
-	config: &Config, network_graph: Arc<NetworkGraph>, logger: Arc<FilesystemLogger>,
-) -> Scorer {
-	let ldk_data_dir = format!("{}/ldk", config.storage_dir_path);
-	let scorer_path = format!("{}/scorer", ldk_data_dir);
+/// 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,
+) -> std::io::Result<Vec<(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::Signer>)>>
+where
+	K::Target: KVStore,
+	ES::Target: EntropySource + Sized,
+	SP::Target: SignerProvider + Sized,
+{
+	let mut res = Vec::new();
 
-	let params = ProbabilisticScoringParameters::default();
-	if let Ok(file) = fs::File::open(scorer_path) {
-		let args = (params.clone(), Arc::clone(&network_graph), Arc::clone(&logger));
-		if let Ok(scorer) = ProbabilisticScorer::read(&mut BufReader::new(file), args) {
-			return scorer;
-		}
-	}
-	ProbabilisticScorer::new(params, network_graph, logger)
-}
+	for stored_key in kv_store.list(CHANNEL_MONITOR_PERSISTENCE_NAMESPACE)? {
+		let txid = Txid::from_hex(stored_key.split_at(64).0).map_err(|_| {
+			std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid tx ID in stored key")
+		})?;
 
-pub(crate) fn read_payment_info(config: &Config) -> Vec<PaymentInfo> {
-	let ldk_data_dir = format!("{}/ldk", config.storage_dir_path);
-	let payment_store_path = format!("{}/{}", ldk_data_dir, PAYMENT_INFO_PERSISTENCE_NAMESPACE);
-	let mut payments = Vec::new();
+		let index: u16 = stored_key.split_at(65).1.parse().map_err(|_| {
+			std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid tx index in stored key")
+		})?;
 
-	if let Ok(res) = fs::read_dir(payment_store_path) {
-		for entry in res {
-			if let Ok(entry) = entry {
-				if entry.path().is_file() {
-					if let Ok(mut f) = fs::File::open(entry.path()) {
-						if let Ok(payment_info) = PaymentInfo::read(&mut f) {
-							payments.push(payment_info);
-						}
-					}
+		match <(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::Signer>)>::read(
+			&mut kv_store.read(CHANNEL_MONITOR_PERSISTENCE_NAMESPACE, &stored_key)?,
+			(&*entropy_source, &*signer_provider),
+		) {
+			Ok((block_hash, channel_monitor)) => {
+				if channel_monitor.get_funding_txo().0.txid != txid
+					|| channel_monitor.get_funding_txo().0.index != index
+				{
+					return Err(std::io::Error::new(
+						std::io::ErrorKind::InvalidData,
+						"ChannelMonitor was stored under the wrong key",
+					));
 				}
+				res.push((block_hash, channel_monitor));
+			}
+			Err(e) => {
+				return Err(std::io::Error::new(
+					std::io::ErrorKind::InvalidData,
+					format!("Failed to deserialize ChannelMonitor: {}", e),
+				))
 			}
 		}
 	}
+	Ok(res)
+}
+/// Read a previously persisted [`NetworkGraph`] from the store.
+pub(crate) fn read_network_graph<K: Deref, L: Deref>(
+	kv_store: K, logger: L,
+) -> Result<NetworkGraph<L>, std::io::Error>
+where
+	K::Target: KVStore,
+	L::Target: Logger,
+{
+	let mut reader =
+		kv_store.read(NETWORK_GRAPH_PERSISTENCE_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY)?;
+	let graph = NetworkGraph::read(&mut reader, logger).map_err(|_| {
+		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize NetworkGraph")
+	})?;
+	Ok(graph)
+}
+/// 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,
+) -> Result<ProbabilisticScorer<G, L>, std::io::Error>
+where
+	K::Target: KVStore,
+	L::Target: Logger,
+{
+	let params = ProbabilisticScoringParameters::default();
+	let mut reader = kv_store.read(SCORER_PERSISTENCE_NAMESPACE, SCORER_PERSISTENCE_KEY)?;
+	let args = (params, network_graph, logger);
+	let scorer = ProbabilisticScorer::read(&mut reader, args).map_err(|_| {
+		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
+	})?;
+	Ok(scorer)
+}
+/// Read previously persisted events from the store.
+pub(crate) fn read_event_queue<K: Deref>(kv_store: K) -> Result<EventQueue<K>, std::io::Error>
+where
+	K::Target: KVStore + KVStorePersister,
+{
+	let mut reader =
+		kv_store.read(EVENT_QUEUE_PERSISTENCE_NAMESPACE, EVENT_QUEUE_PERSISTENCE_KEY)?;
+	let event_queue = EventQueue::read(&mut reader, kv_store).map_err(|_| {
+		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize EventQueue")
+	})?;
+	Ok(event_queue)
+}
+/// Read previously persisted peer info from the store.
+pub(crate) fn read_peer_info<K: Deref>(kv_store: K) -> Result<PeerInfoStorage<K>, std::io::Error>
+where
+	K::Target: KVStore + KVStorePersister,
+{
+	let mut reader = kv_store.read(PEER_INFO_PERSISTENCE_NAMESPACE, PEER_INFO_PERSISTENCE_KEY)?;
+	let peer_info = PeerInfoStorage::read(&mut reader, kv_store).map_err(|_| {
+		std::io::Error::new(
+			std::io::ErrorKind::InvalidData,
+			"Failed to deserialize PeerInfoStorage",
+		)
+	})?;
+	Ok(peer_info)
+}
+/// Read previously persisted payments information from the store.
+pub(crate) fn read_payment_info<K: Deref>(kv_store: K) -> Result<Vec<PaymentInfo>, std::io::Error>
+where
+	K::Target: KVStore + KVStorePersister,
+{
+	let mut res = Vec::new();
 
-	payments
+	for stored_key in kv_store.list(PAYMENT_INFO_PERSISTENCE_NAMESPACE)? {
+		let payment_info =
+			PaymentInfo::read(&mut kv_store.read(PAYMENT_INFO_PERSISTENCE_NAMESPACE, &stored_key)?)
+				.map_err(|_| {
+					std::io::Error::new(
+						std::io::ErrorKind::InvalidData,
+						"Failed to deserialize PaymentInfo",
+					)
+				})?;
+		res.push(payment_info);
+	}
+	Ok(res)
 }