wip

Author: joostjager

lightning/src/chain/chainmonitor.rs

@@ -34,6 +34,7 @@ use crate::chain::transaction::{OutPoint, TransactionData};
 use crate::ln::types::ChannelId;
 use crate::sign::ecdsa::EcdsaChannelSigner;
 use crate::events::{self, Event, EventHandler, ReplayEvent};
+use crate::util::async_poll::{AsyncResult, AsyncResultNo};
 use crate::util::logger::{Logger, WithContext};
 use crate::util::errors::APIError;
 use crate::util::persist::MonitorName;
@@ -171,7 +172,7 @@ pub trait Persist<ChannelSigner: EcdsaChannelSigner> {
 	/// the archive process. Additionally, because the archive operation could be retried on
 	/// restart, this method must in that case be idempotent, ensuring it can handle scenarios where
 	/// the monitor already exists in the archive.
-	fn archive_persisted_channel(&self, monitor_name: MonitorName);
+	fn archive_persisted_channel<'a>(&'a self, monitor_name: MonitorName) -> AsyncResultNo<'a>;
 }
 
 struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> {
@@ -1136,4 +1137,3 @@ mod tests {
 		}).is_err());
 	}
 }
-

lightning/src/util/async_poll.rs

@@ -96,6 +96,9 @@ pub(crate) fn dummy_waker() -> Waker {
 	unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) }
 }
 
+pub type AsyncResultError<'a, T, E> = Pin<Box<dyn Future<Output = Result<T, E>> + 'a + Send>>;
+pub type AsyncResultNo<'a> = Pin<Box<dyn Future<Output = ()> + 'a + Send>>;
+
 /// A type alias for a future that returns a result of type T.
 #[cfg(feature = "std")]
 pub type AsyncResult<'a, T> = Pin<Box<dyn Future<Output = Result<T, ()>> + 'a + Send>>;

lightning/src/util/persist.rs

@@ -32,6 +32,8 @@ use crate::sign::{ecdsa::EcdsaChannelSigner, EntropySource, SignerProvider};
 use crate::util::logger::Logger;
 use crate::util::ser::{Readable, ReadableArgs, Writeable};
 
+use super::async_poll::{AsyncResult, AsyncResultError, AsyncResultNo};
+
 /// The alphabet of characters allowed for namespaces and keys.
 pub const KVSTORE_NAMESPACE_KEY_ALPHABET: &str =
 	"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-";
@@ -127,9 +129,9 @@ pub trait KVStore {
 	/// `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>;
+	fn read<'a>(
+		&'a self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> AsyncResultError<'a, 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
@@ -186,13 +188,13 @@ pub trait MigratableKVStore: KVStore {
 ///
 /// Will abort and return an error if any IO operation fails. Note that in this case the
 /// `target_store` might get left in an intermediate state.
-pub fn migrate_kv_store_data<S: MigratableKVStore, T: MigratableKVStore>(
+pub async fn migrate_kv_store_data<S: MigratableKVStore, T: MigratableKVStore>(
 	source_store: &mut S, target_store: &mut T,
 ) -> Result<(), io::Error> {
 	let keys_to_migrate = source_store.list_all_keys()?;
 
 	for (primary_namespace, secondary_namespace, key) in &keys_to_migrate {
-		let data = source_store.read(primary_namespace, secondary_namespace, key)?;
+		let data = source_store.read(primary_namespace, secondary_namespace, key).await?;
 		target_store.write(primary_namespace, secondary_namespace, key, &data)?;
 	}
 
@@ -254,7 +256,7 @@ where
 	}
 }
 
-impl<ChannelSigner: EcdsaChannelSigner, K: KVStore + ?Sized> Persist<ChannelSigner> for K {
+impl<ChannelSigner: EcdsaChannelSigner, K: KVStore + Sync + ?Sized> 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.
 	// Then we should return InProgress rather than UnrecoverableError, implying we should probably
@@ -289,36 +291,38 @@ impl<ChannelSigner: EcdsaChannelSigner, K: KVStore + ?Sized> Persist<ChannelSign
 		}
 	}
 
-	fn archive_persisted_channel(&self, monitor_name: MonitorName) {
-		let monitor_key = monitor_name.to_string();
-		let monitor = match self.read(
-			CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
-			CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
-			monitor_key.as_str(),
-		) {
-			Ok(monitor) => monitor,
-			Err(_) => return,
-		};
-		match self.write(
-			ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
-			ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
-			monitor_key.as_str(),
-			&monitor,
-		) {
-			Ok(()) => {},
-			Err(_e) => return,
-		};
-		let _ = self.remove(
-			CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
-			CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
-			monitor_key.as_str(),
-			true,
-		);
+	fn archive_persisted_channel<'a>(&'a self, monitor_name: MonitorName) -> AsyncResultNo<'a>{
+		Box::pin(async move {
+			let monitor_key = monitor_name.to_string();
+			let monitor = match self.read(
+				CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+				CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+				monitor_key.as_str(),
+			).await {
+				Ok(monitor) => monitor,
+				Err(_) => return,
+			};
+			match self.write(
+				ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+				ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+				monitor_key.as_str(),
+				&monitor,
+			) {
+				Ok(()) => {},
+				Err(_e) => return,
+			};
+			let _ = self.remove(
+				CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+				CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+				monitor_key.as_str(),
+				true,
+			);
+		})
 	}
 }
 
 /// Read previously persisted [`ChannelMonitor`]s from the store.
-pub fn read_channel_monitors<K: Deref, ES: Deref, SP: Deref>(
+pub async fn read_channel_monitors<K: Deref, ES: Deref, SP: Deref>(
 	kv_store: K, entropy_source: ES, signer_provider: SP,
 ) -> Result<Vec<(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>)>, io::Error>
 where
@@ -337,7 +341,7 @@ where
 				CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
 				CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
 				&stored_key,
-			)?),
+			).await?),
 			(&*entropy_source, &*signer_provider),
 		) {
 			Ok((block_hash, channel_monitor)) => {
@@ -586,15 +590,15 @@ where
 	}
 
 	/// Read a channel monitor.
-	fn read_monitor(
+	async fn read_monitor(
 		&self, monitor_name: &MonitorName, monitor_key: &str,
 	) -> Result<(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>), io::Error>
 	{
 		let mut monitor_cursor = io::Cursor::new(self.kv_store.read(
 			CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
 			CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
 			monitor_key,
-		)?);
+		).await?);
 		// Discard the sentinel bytes if found.
 		if monitor_cursor.get_ref().starts_with(MONITOR_UPDATING_PERSISTER_PREPEND_SENTINEL) {
 			monitor_cursor.set_position(MONITOR_UPDATING_PERSISTER_PREPEND_SENTINEL.len() as u64);