chore: fix conflict.

Author: l-monninger

migration/util/migrator-types/src/migrator/movement_aptos_migrator.rs

@@ -2,7 +2,7 @@ use anyhow::Context;
 use aptos_config::config::NodeConfig;
 use aptos_rest_client::Client as MovementAptosRestClient;
 use kestrel::WaitCondition;
-use movement_aptos_core::MovementAptos;
+use movement_aptos_core::{runtime, MovementAptos};
 use mtma_node_types::executor::MovementAptosNode;
 
 /// An enum supporting different types of runners.
@@ -12,7 +12,7 @@ use mtma_node_types::executor::MovementAptosNode;
 #[derive(Clone)]
 pub enum Runner {
 	/// [MovementAptos] runner.
-	MovementAptos(MovementAptos),
+	MovementAptos(MovementAptos<runtime::TokioTest>),
 }
 
 /// The [MovementAptos] migration struct as would be presented in the criterion.

util/movement-aptos/movement-aptos-core/src/config.rs

@@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 use std::str::FromStr;
 
-use crate::movement_aptos::MovementAptos;
+use crate::movement_aptos::{runtime, MovementAptos};
 use aptos_node::create_single_node_test_config;
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
@@ -66,11 +66,6 @@ pub struct Config {
 	#[orfile(config)]
 	#[clap(long)]
 	pub log_file: Option<PathBuf>,
-
-	/// Whether to create a global rayon pool.
-	#[orfile(config)]
-	#[clap(long)]
-	pub create_global_rayon_pool: bool,
 }
 
 impl Config {
@@ -90,19 +85,15 @@ impl Config {
 		)
 		.map_err(|e| ConfigError::Internal(e.into()))?;
 
-		Ok(Config {
-			node_config: NodeConfigWrapper(node_config),
-			log_file: None,
-			create_global_rayon_pool: false,
-		})
+		Ok(Config { node_config: NodeConfigWrapper(node_config), log_file: None })
 	}
 
 	/// Builds the config into a [MovementAptos] runner.
-	pub fn build(&self) -> Result<MovementAptos, ConfigError> {
-		Ok(MovementAptos::new(
+	pub fn build(&self) -> Result<MovementAptos<runtime::TokioTest>, ConfigError> {
+		Ok(MovementAptos::<runtime::TokioTest>::try_new(
 			self.node_config.0.clone(),
 			self.log_file.clone(),
-			self.create_global_rayon_pool,
-		))
+		)
+		.map_err(|e| ConfigError::Internal(e.into()))?)
 	}
 }

util/movement-aptos/movement-aptos-core/src/movement_aptos.rs

@@ -2,8 +2,11 @@ use aptos_config::config::NodeConfig;
 use kestrel::State;
 use std::path::PathBuf;
 pub mod rest_api;
+pub mod runtime;
 pub use rest_api::RestApi;
 
+use runtime::Runtime;
+
 /// Errors thrown when running [MovementAptos].
 #[derive(Debug, thiserror::Error)]
 pub enum MovementAptosError {
@@ -12,28 +15,41 @@ pub enum MovementAptosError {
 }
 
 #[derive(Clone)]
-pub struct MovementAptos {
+pub struct MovementAptos<R>
+where
+	R: Runtime,
+{
 	/// The [NodeConfig] for the Aptos node.
 	pub node_config: NodeConfig,
 	/// The path to the log file.
 	pub log_file: Option<PathBuf>,
-	/// Whether to create a global rayon pool.
-	pub create_global_rayon_pool: bool,
+	/// The runtime for the Aptos node.
+	pub runtime: std::marker::PhantomData<R>,
 	/// The [MovementAptosRestApi] for the Aptos node.
 	pub rest_api: State<RestApi>,
 }
 
-impl MovementAptos {
-	pub fn new(
+impl<R> MovementAptos<R>
+where
+	R: Runtime,
+{
+	/// If you have something that marks your ability to get a runtime, you can use this.
+	pub fn new(node_config: NodeConfig, log_file: Option<PathBuf>, _runtime: R) -> Self {
+		Self { node_config, log_file, runtime: std::marker::PhantomData, rest_api: State::new() }
+	}
+
+	/// Checks runtime availability and creates a new [MovementAptos].
+	pub fn try_new(
 		node_config: NodeConfig,
 		log_file: Option<PathBuf>,
-		create_global_rayon_pool: bool,
-	) -> Self {
-		Self { node_config, log_file, create_global_rayon_pool, rest_api: State::new() }
+	) -> Result<Self, anyhow::Error> {
+		let runtime = R::try_new()?;
+		let movement_aptos = MovementAptos::new(node_config, log_file, runtime);
+		Ok(movement_aptos)
 	}
 
 	pub fn from_config(config: NodeConfig) -> Result<Self, anyhow::Error> {
-		let movement_aptos = MovementAptos::new(config, None, false);
+		let movement_aptos = MovementAptos::new(config, None, R::try_new()?);
 		Ok(movement_aptos)
 	}
 
@@ -43,15 +59,37 @@ impl MovementAptos {
 	}
 
 	/// Runs the internal node logic
-	pub(crate) fn run_node(&self) -> Result<(), MovementAptosError> {
-		aptos_node::start(
-			self.node_config.clone(),
-			self.log_file.clone(),
-			self.create_global_rayon_pool,
-		)
-		.map_err(|e| MovementAptosError::Internal(e.into()))?;
-
-		Ok(())
+	pub(crate) async fn run_node(&self) -> Result<(), MovementAptosError> {
+		// Clone necessary data for the closure
+		let node_config = self.node_config.clone();
+		let log_file = self.log_file.clone();
+
+		// Spawn the blocking task
+		let blocking_task_result = tokio::task::spawn_blocking(move || {
+			// This closure runs on a blocking thread
+			aptos_node::start(
+				node_config,
+				log_file,
+				R::create_global_rayon_pool(), // Assuming R is in scope and its result is Send
+			)
+			// The closure should return the direct result from aptos_node::start.
+			// The error type from aptos_node::start (let's call it AptosNodeError)
+			// needs to be Send + 'static for the closure.
+		})
+		.await;
+
+		match blocking_task_result {
+			Ok(Ok(())) => Ok(()), // aptos_node::start succeeded
+			Ok(Err(aptos_node_err)) => {
+				// aptos_node::start failed. We need aptos_node_err to be convertible
+				// into the Box<dyn Error> for MovementAptosError::Internal.
+				Err(MovementAptosError::Internal(aptos_node_err.into()))
+			}
+			Err(join_err) => {
+				// spawn_blocking task failed (e.g., panicked or was cancelled by Tokio)
+				Err(MovementAptosError::Internal(Box::new(join_err)))
+			}
+		}
 	}
 
 	/// Runs the node and fills state.
@@ -66,18 +104,22 @@ impl MovementAptos {
 
 		let runner = self.clone();
 		let runner_task = kestrel::task(async move {
-			runner.run_node()?;
+			runner.run_node().await?;
 			Ok::<_, MovementAptosError>(())
 		});
 
 		// rest api state
 		let rest_api_state = self.rest_api.clone();
 		let rest_api_polling = kestrel::task(async move {
 			loop {
+				tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+				println!("POLLING REST API: {:?}", rest_api);
 				// wait for the rest api to be ready
 				let response = reqwest::get(rest_api.rest_api_url.clone())
 					.await
 					.map_err(|e| MovementAptosError::Internal(e.into()))?;
+
+				println!("REST API RESPONSE: {:?}", response);
 				if response.status().is_success() {
 					rest_api_state.write().set(rest_api).await;
 					break;
@@ -102,7 +144,7 @@ mod tests {
 	use rand::thread_rng;
 	use std::path::Path;
 
-	#[tokio::test]
+	#[tokio::test(flavor = "multi_thread")]
 	async fn test_movement_aptos() -> Result<(), anyhow::Error> {
 		// open in a new db
 		let unique_id = uuid::Uuid::new_v4();
@@ -128,11 +170,20 @@ mod tests {
 			rng,
 		)?;
 
-		let movement_aptos = MovementAptos::new(node_config, None, false);
+		let movement_aptos = MovementAptos::<runtime::TokioTest>::try_new(node_config, None)?;
 		let rest_api_state = movement_aptos.rest_api().read().clone();
-		movement_aptos.run().await?;
+
+		let movement_aptos_task = kestrel::task(async move {
+			movement_aptos.run().await?;
+			Ok::<_, MovementAptosError>(())
+		});
+
 		rest_api_state.wait_for(tokio::time::Duration::from_secs(30)).await?;
 
+		println!("ENDING MOVEMENT APTOS");
+
+		kestrel::end!(movement_aptos_task)?;
+
 		Ok(())
 	}
 }

util/movement-aptos/movement-aptos-core/src/movement_aptos/runtime.rs

@@ -0,0 +1,42 @@
+pub mod delegated;
+pub mod native;
+pub mod tokio_test;
+
+pub use delegated::*;
+pub use native::*;
+pub use tokio_test::*;
+
+use std::fmt::Debug;
+
+/// Errors thrown when attempting to use a runtime.
+#[derive(Debug, thiserror::Error)]
+pub enum RuntimeError {
+	#[error("encountered internal error while using Movement Aptos Runtime: {0}")]
+	Internal(#[source] Box<dyn std::error::Error + Send + Sync>),
+	#[error("requested Movement Aptos Runtime is unavailable: {0}")]
+	Unavailable(#[source] Box<dyn std::error::Error + Send + Sync>),
+}
+
+/// Trait for a runtime that can be used to run [MovementAptos].
+///
+/// A runtime knows statically whether it is multithreaded or not.
+pub trait Runtime: Sized + Clone + Debug + Send + Sync + 'static {
+	/// Try to create a new runtime.
+	fn try_new() -> Result<Self, RuntimeError>;
+
+	/// Returns whether to create a global rayon pool.
+	fn create_global_rayon_pool() -> bool;
+}
+
+/// Returns true if the current runtime is multithreaded.
+fn is_multithreaded_runtime() -> bool {
+	std::panic::catch_unwind(|| {
+		tokio::task::block_in_place(|| {});
+	})
+	.is_ok()
+}
+
+/// Returns true if the current runtime is a tokio runtime.
+fn is_in_tokio_runtime() -> bool {
+	tokio::runtime::Handle::try_current().is_ok()
+}

util/movement-aptos/movement-aptos-core/src/movement_aptos/runtime/delegated.rs

@@ -0,0 +1,21 @@
+use super::{Runtime, RuntimeError};
+
+/// Delegated runtime refers to a runtime where the global rayon pool is created outside of the runner.
+///
+/// This is useful when we may be calling from other tasks, i.e., not at the main thread.
+#[derive(Debug, Clone)]
+pub struct Delegated;
+
+impl Runtime for Delegated {
+	/// Try to create a new runtime.
+	///
+	/// There are no restrictions on the surrounding environment here.
+	fn try_new() -> Result<Self, RuntimeError> {
+		Ok(Self)
+	}
+
+	/// Whether to create a global rayon pool.
+	fn create_global_rayon_pool() -> bool {
+		false
+	}
+}

util/movement-aptos/movement-aptos-core/src/movement_aptos/runtime/native.rs

@@ -0,0 +1,43 @@
+use super::{is_in_tokio_runtime, Runtime, RuntimeError};
+
+/// Native runtime refers to a runtime where the global rayon pool is created within the runner.
+#[derive(Debug, Clone)]
+pub struct Native;
+
+impl Runtime for Native {
+	/// Try to create a new runtime.
+	///
+	/// There are no restrictions on the surrounding environment here.
+	fn try_new() -> Result<Self, RuntimeError> {
+		if is_in_tokio_runtime() {
+			return Err(RuntimeError::Unavailable(Box::new(std::io::Error::new(
+				std::io::ErrorKind::Other,
+				"Native runtime is not available in a tokio runtime",
+			))));
+		}
+
+		Ok(Self)
+	}
+
+	/// Whether to create a global rayon pool.
+	fn create_global_rayon_pool() -> bool {
+		true
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[tokio::test]
+	async fn test_native_runtime_should_fail_in_tokio_runtime() -> Result<(), anyhow::Error> {
+		assert!(Native::try_new().is_err());
+		Ok(())
+	}
+
+	#[test]
+	fn test_native_runtime_should_succeed_outside_of_tokio_runtime() -> Result<(), anyhow::Error> {
+		assert!(Native::try_new().is_ok());
+		Ok(())
+	}
+}

util/movement-aptos/movement-aptos-core/src/movement_aptos/runtime/tokio_test.rs

@@ -0,0 +1,47 @@
+use super::{is_in_tokio_runtime, is_multithreaded_runtime, Runtime, RuntimeError};
+
+/// Tokio test runtime.
+#[derive(Debug, Clone)]
+pub struct TokioTest;
+
+impl Runtime for TokioTest {
+	/// Try to create a new runtime.
+	fn try_new() -> Result<Self, RuntimeError> {
+		if !is_in_tokio_runtime() || !is_multithreaded_runtime() {
+			return Err(RuntimeError::Unavailable(Box::new(std::io::Error::new(
+				std::io::ErrorKind::Other,
+				"Tokio test runtime is not multithreaded use #[tokio::test(flavor = \"multi_thread\")] instead",
+			))));
+		}
+
+		Ok(Self)
+	}
+
+	/// Whether to create a global rayon pool.
+	fn create_global_rayon_pool() -> bool {
+		false
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[tokio::test(flavor = "multi_thread")]
+	async fn test_tokio_test_should_succeed_in_multi_thread() -> Result<(), anyhow::Error> {
+		TokioTest::try_new()?;
+		Ok(())
+	}
+
+	#[tokio::test]
+	async fn test_tokio_test_should_fail_in_single_thread() -> Result<(), anyhow::Error> {
+		assert!(TokioTest::try_new().is_err());
+		Ok(())
+	}
+
+	#[test]
+	fn test_tokio_test_should_fail_outside_of_tokio_test() -> Result<(), anyhow::Error> {
+		assert!(TokioTest::try_new().is_err());
+		Ok(())
+	}
+}