Use crate::runtime::Runtime for VssStore

.. which now gives us cleaner reuse/handling of outer runtime contexts,
cleanup on `Drop`, etc.

Author: tnull

src/builder.rs

@@ -601,10 +601,11 @@ impl NodeBuilder {
 		let vss_seed_bytes: [u8; 32] = vss_xprv.private_key.secret_bytes();
 
 		let vss_store =
-			VssStore::new(vss_url, store_id, vss_seed_bytes, header_provider).map_err(|e| {
-				log_error!(logger, "Failed to setup VssStore: {}", e);
-				BuildError::KVStoreSetupFailed
-			})?;
+			VssStore::new(vss_url, store_id, vss_seed_bytes, header_provider, Arc::clone(&logger))
+				.map_err(|e| {
+					log_error!(logger, "Failed to setup VssStore: {}", e);
+					BuildError::KVStoreSetupFailed
+				})?;
 		build_with_store_internal(
 			config,
 			self.chain_data_source_config.as_ref(),

src/io/vss_store.rs

@@ -6,6 +6,9 @@
 // accordance with one or both of these licenses.
 
 use crate::io::utils::check_namespace_key_validity;
+use crate::logger::Logger;
+use crate::runtime::Runtime;
+
 use bitcoin::hashes::{sha256, Hash, HashEngine, Hmac, HmacEngine};
 use lightning::io::{self, Error, ErrorKind};
 use lightning::util::persist::KVStore;
@@ -15,7 +18,6 @@ use rand::RngCore;
 use std::panic::RefUnwindSafe;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::runtime::Runtime;
 use vss_client::client::VssClient;
 use vss_client::error::VssError;
 use vss_client::headers::VssHeaderProvider;
@@ -49,9 +51,10 @@ pub struct VssStore {
 impl VssStore {
 	pub(crate) fn new(
 		base_url: String, store_id: String, vss_seed: [u8; 32],
-		header_provider: Arc<dyn VssHeaderProvider>,
+		header_provider: Arc<dyn VssHeaderProvider>, logger: Arc<Logger>,
 	) -> io::Result<Self> {
-		let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build()?;
+		let runtime = Runtime::new(logger);
+		runtime.start()?;
 		let (data_encryption_key, obfuscation_master_key) =
 			derive_data_encryption_and_obfuscation_keys(&vss_seed);
 		let key_obfuscator = KeyObfuscator::new(obfuscation_master_key);
@@ -136,19 +139,16 @@ impl KVStore for VssStore {
 			store_id: self.store_id.clone(),
 			key: self.build_key(primary_namespace, secondary_namespace, key)?,
 		};
-
-		let resp =
-			tokio::task::block_in_place(|| self.runtime.block_on(self.client.get_object(&request)))
-				.map_err(|e| {
-					let msg = format!(
-						"Failed to read from key {}/{}/{}: {}",
-						primary_namespace, secondary_namespace, key, e
-					);
-					match e {
-						VssError::NoSuchKeyError(..) => Error::new(ErrorKind::NotFound, msg),
-						_ => Error::new(ErrorKind::Other, msg),
-					}
-				})?;
+		let resp = self.runtime.block_on(self.client.get_object(&request))?.map_err(|e| {
+			let msg = format!(
+				"Failed to read from key {}/{}/{}: {}",
+				primary_namespace, secondary_namespace, key, e
+			);
+			match e {
+				VssError::NoSuchKeyError(..) => Error::new(ErrorKind::NotFound, msg),
+				_ => Error::new(ErrorKind::Other, msg),
+			}
+		})?;
 		// unwrap safety: resp.value must be always present for a non-erroneous VSS response, otherwise
 		// it is an API-violation which is converted to [`VssError::InternalServerError`] in [`VssClient`]
 		let storable = Storable::decode(&resp.value.unwrap().value[..]).map_err(|e| {
@@ -179,14 +179,13 @@ impl KVStore for VssStore {
 			delete_items: vec![],
 		};
 
-		tokio::task::block_in_place(|| self.runtime.block_on(self.client.put_object(&request)))
-			.map_err(|e| {
-				let msg = format!(
-					"Failed to write to key {}/{}/{}: {}",
-					primary_namespace, secondary_namespace, key, e
-				);
-				Error::new(ErrorKind::Other, msg)
-			})?;
+		self.runtime.block_on(self.client.put_object(&request))?.map_err(|e| {
+			let msg = format!(
+				"Failed to write to key {}/{}/{}: {}",
+				primary_namespace, secondary_namespace, key, e
+			);
+			Error::new(ErrorKind::Other, msg)
+		})?;
 
 		Ok(())
 	}
@@ -204,30 +203,29 @@ impl KVStore for VssStore {
 			}),
 		};
 
-		tokio::task::block_in_place(|| self.runtime.block_on(self.client.delete_object(&request)))
-			.map_err(|e| {
-				let msg = format!(
-					"Failed to delete key {}/{}/{}: {}",
-					primary_namespace, secondary_namespace, key, e
-				);
-				Error::new(ErrorKind::Other, msg)
-			})?;
+		self.runtime.block_on(self.client.delete_object(&request))?.map_err(|e| {
+			let msg = format!(
+				"Failed to delete key {}/{}/{}: {}",
+				primary_namespace, secondary_namespace, key, e
+			);
+			Error::new(ErrorKind::Other, msg)
+		})?;
 		Ok(())
 	}
 
 	fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> {
 		check_namespace_key_validity(primary_namespace, secondary_namespace, None, "list")?;
 
-		let keys = tokio::task::block_in_place(|| {
-			self.runtime.block_on(self.list_all_keys(primary_namespace, secondary_namespace))
-		})
-		.map_err(|e| {
-			let msg = format!(
-				"Failed to retrieve keys in namespace: {}/{} : {}",
-				primary_namespace, secondary_namespace, e
-			);
-			Error::new(ErrorKind::Other, msg)
-		})?;
+		let keys = self
+			.runtime
+			.block_on(self.list_all_keys(primary_namespace, secondary_namespace))?
+			.map_err(|e| {
+				let msg = format!(
+					"Failed to retrieve keys in namespace: {}/{} : {}",
+					primary_namespace, secondary_namespace, e
+				);
+				Error::new(ErrorKind::Other, msg)
+			})?;
 
 		Ok(keys)
 	}
@@ -269,7 +267,21 @@ mod tests {
 	use vss_client::headers::FixedHeaders;
 
 	#[test]
-	fn read_write_remove_list_persist() {
+	fn vss_read_write_remove_list_persist() {
+		let vss_base_url = std::env::var("TEST_VSS_BASE_URL").unwrap();
+		let mut rng = thread_rng();
+		let rand_store_id: String = (0..7).map(|_| rng.sample(Alphanumeric) as char).collect();
+		let mut vss_seed = [0u8; 32];
+		rng.fill_bytes(&mut vss_seed);
+		let header_provider = Arc::new(FixedHeaders::new(HashMap::new()));
+		let vss_store =
+			VssStore::new(vss_base_url, rand_store_id, vss_seed, header_provider).unwrap();
+
+		do_read_write_remove_list_persist(&vss_store);
+	}
+
+	#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+	async fn vss_read_write_remove_list_persist_in_runtime_context() {
 		let vss_base_url = std::env::var("TEST_VSS_BASE_URL").unwrap();
 		let mut rng = thread_rng();
 		let rand_store_id: String = (0..7).map(|_| rng.sample(Alphanumeric) as char).collect();

src/runtime.rs

@@ -153,9 +153,9 @@ impl fmt::Display for RuntimeError {
 	}
 }
 
-impl From<RuntimeError> for std::io::Error {
+impl From<RuntimeError> for lightning::io::Error {
 	fn from(runtime_error: RuntimeError) -> Self {
 		let msg = format!("Runtime error: {}", runtime_error);
-		std::io::Error::new(std::io::ErrorKind::Other, msg)
+		lightning::io::Error::new(lightning::io::ErrorKind::Other, msg)
 	}
 }