DROPME try

Author: tnull

.github/workflows/vss-integration.yml

@@ -75,7 +75,7 @@ jobs:
           cd ldk-node
           export TEST_VSS_BASE_URL="http://localhost:8080/vss"
           RUSTFLAGS="--cfg vss_test" cargo build --verbose --color always
-          RUSTFLAGS="--cfg vss_test" cargo test --test integration_tests_vss
+          RUSTFLAGS="--cfg vss_test --cfg tokio_unstable" cargo test --test integration_tests_vss -- --nocapture
 
       - name: Cleanup
         run: |

src/io/vss_store.rs

@@ -136,6 +136,8 @@ impl KVStore for VssStore {
 			store_id: self.store_id.clone(),
 			key: self.build_key(primary_namespace, secondary_namespace, key)?,
 		};
+
+		println!("READ: {}/{}/{}", primary_namespace, secondary_namespace, key);
 		let resp = self.runtime.block_on(self.client.get_object(&request)).map_err(|e| {
 			let msg = format!(
 				"Failed to read from key {}/{}/{}: {}",
@@ -146,6 +148,7 @@ impl KVStore for VssStore {
 				_ => Error::new(ErrorKind::Other, msg),
 			}
 		})?;
+		println!("READ DONE: {}/{}/{}", primary_namespace, secondary_namespace, key);
 		// 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| {
@@ -176,14 +179,28 @@ impl KVStore for VssStore {
 			delete_items: vec![],
 		};
 
-		self.runtime.block_on(self.client.put_object(&request)).map_err(|e| {
+		println!("WRITE: {}/{}/{}", primary_namespace, secondary_namespace, key);
+		let res = self.runtime.block_on(async move {
+			tokio::time::timeout(Duration::from_secs(100), self.client.put_object(&request))
+				.await
+				.map_err(|e| {
+					let msg = format!(
+						"Failed to write to key {}/{}/{}: {}",
+						primary_namespace, secondary_namespace, key, e
+					);
+					Error::new(ErrorKind::Other, msg)
+				})
+		});
+
+		println!("WRITE DONE: {}/{}/{}: {:?}", primary_namespace, secondary_namespace, key, res);
+
+		res?.map_err(|e| {
 			let msg = format!(
 				"Failed to write to key {}/{}/{}: {}",
 				primary_namespace, secondary_namespace, key, e
 			);
 			Error::new(ErrorKind::Other, msg)
 		})?;
-
 		Ok(())
 	}
 
@@ -200,19 +217,22 @@ impl KVStore for VssStore {
 			}),
 		};
 
+		println!("REMOVE: {}/{}/{}", primary_namespace, secondary_namespace, key);
 		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)
 		})?;
+		println!("REMOVE DONE: {}/{}/{}", primary_namespace, secondary_namespace, key);
 		Ok(())
 	}
 
 	fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> {
 		check_namespace_key_validity(primary_namespace, secondary_namespace, None, "list")?;
 
+		println!("LIST: {}/{}", primary_namespace, secondary_namespace);
 		let keys = self
 			.runtime
 			.block_on(self.list_all_keys(primary_namespace, secondary_namespace))
@@ -223,6 +243,7 @@ impl KVStore for VssStore {
 				);
 				Error::new(ErrorKind::Other, msg)
 			})?;
+		println!("LIST DONE: {}/{}", primary_namespace, secondary_namespace);
 
 		Ok(keys)
 	}

src/runtime.rs

@@ -55,6 +55,15 @@ impl Runtime {
 		// to detect the outer context here, and otherwise use whatever was set during
 		// initialization.
 		let handle = tokio::runtime::Handle::try_current().unwrap_or(self.handle());
+		#[cfg(tokio_unstable)]
+		{
+			println!("Tokio blocking queue depth: {}", handle.metrics().blocking_queue_depth());
+			println!(
+				"Tokio num_blocking_threads {} / idle_blocking_threads {}",
+				handle.metrics().num_blocking_threads(),
+				handle.metrics().num_idle_blocking_threads()
+			);
+		}
 		tokio::task::block_in_place(move || handle.block_on(future))
 	}
 

tests/common/logging.rs

@@ -22,11 +22,12 @@ impl Default for TestLogWriter {
 
 pub(crate) struct MockLogFacadeLogger {
 	logs: Arc<Mutex<Vec<String>>>,
+	prefix: String,
 }
 
 impl MockLogFacadeLogger {
-	pub fn new() -> Self {
-		Self { logs: Arc::new(Mutex::new(Vec::new())) }
+	pub fn new(prefix: String) -> Self {
+		Self { prefix, logs: Arc::new(Mutex::new(Vec::new())) }
 	}
 
 	pub fn retrieve_logs(&self) -> Vec<String> {
@@ -48,6 +49,7 @@ impl LogFacadeLog for MockLogFacadeLogger {
 			record.line().unwrap(),
 			record.args()
 		);
+		println!("{}: {}", self.prefix, message);
 		self.logs.lock().unwrap().push(message);
 	}
 
@@ -95,8 +97,10 @@ impl<'a> From<MockLogRecord<'a>> for LogFacadeRecord<'a> {
 	}
 }
 
-pub(crate) fn init_log_logger(level: LogFacadeLevelFilter) -> Arc<MockLogFacadeLogger> {
-	let logger = Arc::new(MockLogFacadeLogger::new());
+pub(crate) fn init_log_logger(
+	prefix: String, level: LogFacadeLevelFilter,
+) -> Arc<MockLogFacadeLogger> {
+	let logger = Arc::new(MockLogFacadeLogger::new(prefix));
 	log::set_boxed_logger(Box::new(logger.clone())).unwrap();
 	log::set_max_level(level);
 

tests/integration_tests_rust.rs

@@ -1275,7 +1275,7 @@ fn facade_logging() {
 	let (_bitcoind, electrsd) = setup_bitcoind_and_electrsd();
 	let chain_source = TestChainSource::Esplora(&electrsd);
 
-	let logger = init_log_logger(LevelFilter::Trace);
+	let logger = init_log_logger("".to_owned(), LevelFilter::Trace);
 	let mut config = random_config(false);
 	config.log_writer = TestLogWriter::LogFacade;
 

tests/integration_tests_vss.rs

@@ -9,17 +9,23 @@
 
 mod common;
 
+use common::logging::{init_log_logger, TestLogWriter};
 use ldk_node::Builder;
+use log::LevelFilter;
 use std::collections::HashMap;
 
 #[test]
 fn channel_full_cycle_with_vss_store() {
 	let (bitcoind, electrsd) = common::setup_bitcoind_and_electrsd();
 	println!("== Node A ==");
 	let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap());
-	let config_a = common::random_config(true);
+	let mut config_a = common::random_config(true);
+	let prefix_a = "A".to_string();
+	let logger_a = init_log_logger(prefix_a, LevelFilter::Trace);
+	config_a.log_writer = TestLogWriter::LogFacade;
 	let mut builder_a = Builder::from_config(config_a.node_config);
 	builder_a.set_chain_source_esplora(esplora_url.clone(), None);
+	builder_a.set_log_facade_logger();
 	let vss_base_url = std::env::var("TEST_VSS_BASE_URL").unwrap();
 	let node_a = builder_a
 		.build_with_vss_store_and_fixed_headers(
@@ -31,9 +37,13 @@ fn channel_full_cycle_with_vss_store() {
 	node_a.start().unwrap();
 
 	println!("\n== Node B ==");
-	let config_b = common::random_config(true);
+	let mut config_b = common::random_config(true);
+	//let prefix_b = "B".to_string();
+	//let logger_b = init_log_logger(prefix_b, LevelFilter::Trace);
+	//config_a.log_writer = TestLogWriter::LogFacade;
 	let mut builder_b = Builder::from_config(config_b.node_config);
 	builder_b.set_chain_source_esplora(esplora_url.clone(), None);
+	//builder_b.set_log_facade_logger();
 	let node_b = builder_b
 		.build_with_vss_store_and_fixed_headers(
 			vss_base_url,