Use FilesystemStore in BackgroundProcessor

We switch our BP over to use `FilesystemStore`, which also gives us test
coverage and ensures the compatibility.

Author: tnull

lightning-background-processor/Cargo.toml

@@ -29,4 +29,4 @@ lightning-rapid-gossip-sync = { version = "0.0.116", path = "../lightning-rapid-
 tokio = { version = "1.14", features = [ "macros", "rt", "rt-multi-thread", "sync", "time" ] }
 lightning = { version = "0.0.116", path = "../lightning", features = ["_test_utils"] }
 lightning-invoice = { version = "0.24.0", path = "../lightning-invoice" }
-lightning-persister = { version = "0.0.116", path = "../lightning-persister" }
+lightning-storage = { version = "0.0.116", path = "../lightning-storage" }

lightning-background-processor/src/lib.rs

@@ -507,6 +507,17 @@ use core::task;
 /// #     fn write(&self, namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
 /// #     fn remove(&self, namespace: &str, key: &str) -> io::Result<()> { Ok(()) }
 /// #     fn list(&self, namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
+/// # use lightning::io;
+/// # use std::sync::{Arc, Mutex};
+/// # use std::sync::atomic::{AtomicBool, Ordering};
+/// # use lightning_background_processor::{process_events_async, GossipSync};
+/// # struct MyStore {}
+/// # impl lightning::util::persist::KVStore for MyStore {
+/// #     type Reader = io::Cursor<Vec<u8>>;
+/// #     fn read(&self, namespace: &str, key: &str) -> io::Result<Self::Reader> { Ok(io::Cursor::new(Vec::new())) }
+/// #     fn write(&self, namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
+/// #     fn remove(&self, namespace: &str, key: &str) -> io::Result<()> { Ok(()) }
+/// #     fn list(&self, namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
 /// # }
 /// # struct MyEventHandler {}
 /// # impl MyEventHandler {
@@ -868,7 +879,7 @@ mod tests {
 	use lightning::util::ser::Writeable;
 	use lightning::util::test_utils;
 	use lightning::util::persist::{KVStore, CHANNEL_MANAGER_PERSISTENCE_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY, SCORER_PERSISTENCE_NAMESPACE, SCORER_PERSISTENCE_KEY};
-	use lightning_persister::fs_store::{FilesystemStore, FilesystemReader};
+	use lightning_storage::fs_store::{FilesystemStore, FilesystemReader};
 	use std::collections::VecDeque;
 	use std::{fs, env};
 	use std::path::PathBuf;

lightning/src/util/persist.rs

@@ -100,6 +100,8 @@ pub trait Persister<'a, M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F:
 }
 
 
+impl<'a, A: KVStore, M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, L: Deref, S: WriteableScore<'a>> Persister<'a, M, T, ES, NS, SP, F, R, L, S> for A
+
 impl<'a, A: KVStore, M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, L: Deref, S: WriteableScore<'a>> Persister<'a, M, T, ES, NS, SP, F, R, L, S> for A
 	where M::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
 		T::Target: 'static + BroadcasterInterface,
@@ -110,29 +112,38 @@ impl<'a, A: KVStore, M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Der
 		R::Target: 'static + Router,
 		L::Target: 'static + Logger,
 {
+	/// Persist the given ['ChannelManager'] to disk, returning an error if persistence failed.
 	/// Persist the given ['ChannelManager'] to disk, returning an error if persistence failed.
 	fn persist_manager(&self, channel_manager: &ChannelManager<M, T, ES, NS, SP, F, R, L>) -> Result<(), io::Error> {
 		self.write(CHANNEL_MANAGER_PERSISTENCE_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY, &channel_manager.encode())
+		self.write(CHANNEL_MANAGER_PERSISTENCE_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY, &channel_manager.encode())
 	}
 
+	/// Persist the given [`NetworkGraph`] to disk, returning an error if persistence failed.
 	/// Persist the given [`NetworkGraph`] to disk, returning an error if persistence failed.
 	fn persist_graph(&self, network_graph: &NetworkGraph<L>) -> Result<(), io::Error> {
 		self.write(NETWORK_GRAPH_PERSISTENCE_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY, &network_graph.encode())
+		self.write(NETWORK_GRAPH_PERSISTENCE_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY, &network_graph.encode())
 	}
 
+	/// Persist the given [`WriteableScore`] to disk, returning an error if persistence failed.
 	/// Persist the given [`WriteableScore`] to disk, returning an error if persistence failed.
 	fn persist_scorer(&self, scorer: &S) -> Result<(), io::Error> {
 		self.write(SCORER_PERSISTENCE_NAMESPACE, SCORER_PERSISTENCE_KEY, &scorer.encode())
+		self.write(SCORER_PERSISTENCE_NAMESPACE, SCORER_PERSISTENCE_KEY, &scorer.encode())
 	}
 }
 
+impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore> Persist<ChannelSigner> for K {
 impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore> Persist<ChannelSigner> for K {
 	// TODO: We really need a way for the persister to inform the user that its time to crash/shut
 	// down once these start returning failure.
 	// A PermanentFailure implies we should probably just shut down the node since we're
 	// force-closing channels without even broadcasting!
 
 	fn persist_new_channel(&self, funding_txo: OutPoint, monitor: &ChannelMonitor<ChannelSigner>, _update_id: MonitorUpdateId) -> chain::ChannelMonitorUpdateStatus {
+		let key = format!("{}_{}", funding_txo.txid.to_hex(), funding_txo.index);
+		match self.write(CHANNEL_MONITOR_PERSISTENCE_NAMESPACE, &key, &monitor.encode()) {
 		let key = format!("{}_{}", funding_txo.txid.to_hex(), funding_txo.index);
 		match self.write(CHANNEL_MONITOR_PERSISTENCE_NAMESPACE, &key, &monitor.encode()) {
 			Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
@@ -141,6 +152,8 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore> Persist<ChannelSign
 	}
 
 	fn update_persisted_channel(&self, funding_txo: OutPoint, _update: Option<&ChannelMonitorUpdate>, monitor: &ChannelMonitor<ChannelSigner>, _update_id: MonitorUpdateId) -> chain::ChannelMonitorUpdateStatus {
+		let key = format!("{}_{}", funding_txo.txid.to_hex(), funding_txo.index);
+		match self.write(CHANNEL_MONITOR_PERSISTENCE_NAMESPACE, &key, &monitor.encode()) {
 		let key = format!("{}_{}", funding_txo.txid.to_hex(), funding_txo.index);
 		match self.write(CHANNEL_MONITOR_PERSISTENCE_NAMESPACE, &key, &monitor.encode()) {
 			Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,