sync wrapper wip

Author: joostjager

lightning-background-processor/src/lib.rs

@@ -40,7 +40,9 @@ use lightning::sign::ChangeDestinationSourceSync;
 use lightning::sign::OutputSpender;
 use lightning::util::async_poll::FutureSpawner;
 use lightning::util::logger::Logger;
-use lightning::util::persist::{KVStore, Persister};
+use lightning::util::persist::{
+	KVStore, KVStoreSync, KVStoreSyncWrapper, Persister, PersisterSync,
+};
 use lightning::util::sweep::OutputSweeper;
 #[cfg(feature = "std")]
 use lightning::util::sweep::OutputSweeperSync;
@@ -995,7 +997,9 @@ impl BackgroundProcessor {
 		D: 'static + Deref,
 		O: 'static + Deref,
 		K: 'static + Deref,
-		OS: 'static + Deref<Target = OutputSweeperSync<T, D, F, CF, K, L, O>> + Send,
+		OS: 'static
+			+ Deref<Target = OutputSweeperSync<T, D, F, CF, KVStoreSyncWrapper<K>, L, O>>
+			+ Send,
 		FS: FutureSpawner,
 	>(
 		persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
@@ -1009,14 +1013,14 @@ impl BackgroundProcessor {
 		F::Target: 'static + FeeEstimator,
 		L::Target: 'static + Logger,
 		P::Target: 'static + Persist<<CM::Target as AChannelManager>::Signer>,
-		PS::Target: 'static + Persister<'a, CM, L, S>,
+		PS::Target: 'static + PersisterSync<'a, CM, L, S>,
 		CM::Target: AChannelManager,
 		OM::Target: AOnionMessenger,
 		PM::Target: APeerManager,
 		LM::Target: ALiquidityManager,
 		D::Target: ChangeDestinationSourceSync,
 		O::Target: 'static + OutputSpender,
-		K::Target: 'static + KVStore,
+		K::Target: 'static + KVStoreSync,
 	{
 		let stop_thread = Arc::new(AtomicBool::new(false));
 		let stop_thread_clone = stop_thread.clone();
@@ -1033,9 +1037,9 @@ impl BackgroundProcessor {
 						.expect("Time should be sometime after 1970");
 					if update_scorer(scorer, &event, duration_since_epoch) {
 						log_trace!(logger, "Persisting scorer after update");
-						// if let Err(e) = persister.persist_scorer(&scorer).await {
-						// 	log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
-						// }
+						if let Err(e) = persister.persist_scorer(&scorer) {
+							log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
+						}
 					}
 				}
 				event_handler.handle_event(event)

lightning/src/util/persist.rs

@@ -167,6 +167,91 @@ pub trait KVStore {
 	) -> Result<Vec<String>, io::Error>;
 }
 
+/// Provides a synchronous interface to the [`KVStore`] trait.
+pub trait KVStoreSync {
+	/// Returns the data stored for the given `primary_namespace`, `secondary_namespace`, and
+	/// `key`.
+	///
+	/// Returns an [`ErrorKind::NotFound`] if the given `key` could not be found in the given
+	/// `primary_namespace` and `secondary_namespace`.
+	///
+	/// [`ErrorKind::NotFound`]: io::ErrorKind::NotFound
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> Result<Vec<u8>, io::Error>;
+	/// Persists the given data under the given `key`.
+	///
+	/// Will create the given `primary_namespace` and `secondary_namespace` if not already present
+	/// in the store.
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8],
+	) -> Result<(), io::Error>;
+	/// Removes any data that had previously been persisted under the given `key`.
+	///
+	/// If the `lazy` flag is set to `true`, the backend implementation might choose to lazily
+	/// remove the given `key` at some point in time after the method returns, e.g., as part of an
+	/// eventual batch deletion of multiple keys. As a consequence, subsequent calls to
+	/// [`KVStore::list`] might include the removed key until the changes are actually persisted.
+	///
+	/// Note that while setting the `lazy` flag reduces the I/O burden of multiple subsequent
+	/// `remove` calls, it also influences the atomicity guarantees as lazy `remove`s could
+	/// potentially get lost on crash after the method returns. Therefore, this flag should only be
+	/// set for `remove` operations that can be safely replayed at a later time.
+	///
+	/// Returns successfully if no data will be stored for the given `primary_namespace`,
+	/// `secondary_namespace`, and `key`, independently of whether it was present before its
+	/// invokation or not.
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> Result<(), io::Error>;
+	/// Returns a list of keys that are stored under the given `secondary_namespace` in
+	/// `primary_namespace`.
+	///
+	/// Returns the keys in arbitrary order, so users requiring a particular order need to sort the
+	/// returned keys. Returns an empty list if `primary_namespace` or `secondary_namespace` is unknown.
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> Result<Vec<String>, io::Error>;
+}
+
+/// A wrapper around a [`KVStoreSync`] that implements the [`KVStore`] trait.
+pub struct KVStoreSyncWrapper<K: KVStoreSync>(Arc<K>);
+
+impl<K: KVStoreSync> KVStoreSyncWrapper<K> {
+	/// Constructs a new [`KVStoreSyncWrapper`].
+	pub fn new(kv_store: Arc<K>) -> Self {
+		Self(kv_store)
+	}
+}
+
+impl<K: KVStoreSync> KVStore for KVStoreSyncWrapper<K> {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> AsyncResultType<'static, Vec<u8>, io::Error> {
+		let res = self.0.read(primary_namespace, secondary_namespace, key);
+
+		Box::pin(async move { res })
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8],
+	) -> AsyncResultType<'static, (), io::Error> {
+		todo!()
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> Result<(), io::Error> {
+		todo!()
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> Result<Vec<String>, io::Error> {
+		todo!()
+	}
+}
+
 /// Provides additional interface methods that are required for [`KVStore`]-to-[`KVStore`]
 /// data migration.
 pub trait MigratableKVStore: KVStore {
@@ -275,6 +360,61 @@ where
 	}
 }
 
+/// Trait that handles persisting a [`ChannelManager`], [`NetworkGraph`], and [`WriteableScore`] to disk.
+///
+/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
+pub trait PersisterSync<'a, CM: Deref, L: Deref, S: Deref>
+where
+	CM::Target: 'static + AChannelManager,
+	L::Target: 'static + Logger,
+	S::Target: WriteableScore<'a>,
+{
+	/// Persist the given ['ChannelManager'] to disk, returning an error if persistence failed.
+	///
+	/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
+	fn persist_manager(&self, channel_manager: &CM) -> Result<(), io::Error>;
+
+	/// Persist the given [`NetworkGraph`] to disk, returning an error if persistence failed.
+	fn persist_graph(&self, network_graph: &NetworkGraph<L>) -> Result<(), io::Error>;
+
+	/// Persist the given [`WriteableScore`] to disk, returning an error if persistence failed.
+	fn persist_scorer(&self, scorer: &S) -> Result<(), io::Error>;
+}
+
+impl<'a, A: KVStoreSync + ?Sized, CM: Deref, L: Deref, S: Deref> PersisterSync<'a, CM, L, S> for A
+where
+	CM::Target: 'static + AChannelManager,
+	L::Target: 'static + Logger,
+	S::Target: WriteableScore<'a>,
+{
+	fn persist_manager(&self, channel_manager: &CM) -> Result<(), io::Error> {
+		self.write(
+			CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+			CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
+			CHANNEL_MANAGER_PERSISTENCE_KEY,
+			&channel_manager.get_cm().encode(),
+		)
+	}
+
+	fn persist_graph(&self, network_graph: &NetworkGraph<L>) -> Result<(), io::Error> {
+		self.write(
+			NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
+			NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
+			NETWORK_GRAPH_PERSISTENCE_KEY,
+			&network_graph.encode(),
+		)
+	}
+
+	fn persist_scorer(&self, scorer: &S) -> Result<(), io::Error> {
+		self.write(
+			SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
+			SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
+			SCORER_PERSISTENCE_KEY,
+			&scorer.encode(),
+		)
+	}
+}
+
 impl<ChannelSigner: EcdsaChannelSigner, K: KVStore + ?Sized + Sync + Send + 'static>
 	Persist<ChannelSigner> for Arc<K>
 {