chore: rewrite movement aptos embedded runner.

Author: l-monninger

Cargo.lock

@@ -18,6 +18,7 @@ members = [
     "checks/e2e/citeria/*",
     # util
     "util/movement/*",
+    "util/movement-aptos/*",
 
 ]
 
@@ -108,7 +109,8 @@ aptos-db = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "2
 aptos_schemadb = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "2941ee2b5876ab93cfb8010154c9466d7d593ed9" }
 aptos-config = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "2941ee2b5876ab93cfb8010154c9466d7d593ed9" }
 aptos-db-indexer = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "2941ee2b5876ab93cfb8010154c9466d7d593ed9" }
-# model checking and verification
+aptos-node = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "2941ee2b5876ab93cfb8010154c9466d7d593ed9" }
+aptos-cached-packages = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "2941ee2b5876ab93cfb8010154c9466d7d593ed9" }
 
 
 # secure-signing

Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "movement-aptos-core"
+version = { workspace = true }
+edition = { workspace = true }
+license = { workspace = true }
+authors = { workspace = true }
+homepage = { workspace = true }
+publish = { workspace = true }
+rust-version = { workspace = true }
+
+[dependencies]
+thiserror = { workspace = true }
+serde = { workspace = true }
+anyhow = { workspace = true }
+hex = { workspace = true }
+syncador = { workspace = true }
+tokio = { workspace = true }
+aptos-node = { workspace = true }
+aptos-config = { workspace = true }
+serde_json = { workspace = true }
+clap = { workspace = true }
+jsonlvar = { workspace = true }
+orfile = { workspace = true }
+kestrel = { workspace = true }
+reqwest = { workspace = true }
+[dev-dependencies]
+uuid = { workspace = true }
+chrono = { workspace = true }
+aptos-cached-packages = { workspace = true }
+rand = { workspace = true }
+
+[lints]
+workspace = true

util/movement-aptos/movement-aptos-core/Cargo.toml

@@ -0,0 +1,2 @@
+# `movement-core`
+The core lib for running an embedded Movement Full Node. 

util/movement-aptos/movement-aptos-core/README.md

@@ -0,0 +1,64 @@
+use aptos_config::config::NodeConfig;
+use clap::Parser;
+use jsonlvar::Jsonl;
+use orfile::Orfile;
+use serde::{Deserialize, Serialize};
+use std::path::PathBuf;
+use std::str::FromStr;
+
+use crate::movement_aptos::MovementAptos;
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct NodeConfigWrapper(NodeConfig);
+
+impl FromStr for NodeConfigWrapper {
+	type Err = ConfigError;
+
+	fn from_str(s: &str) -> Result<Self, Self::Err> {
+		// if "default" return the default config
+		if s == "default" {
+			return Ok(NodeConfigWrapper(NodeConfig::default()));
+		}
+
+		// otherwise, parse the json
+		let node_config: NodeConfig =
+			serde_json::from_str(s).map_err(|e| ConfigError::Internal(e.into()))?;
+		Ok(NodeConfigWrapper(node_config))
+	}
+}
+
+/// Errors thrown when parsing an [Eth] network.
+#[derive(Debug, thiserror::Error)]
+pub enum ConfigError {
+	#[error("movment-core Config encountered an internal error: {0}")]
+	Internal(#[source] Box<dyn std::error::Error + Send + Sync>),
+}
+
+#[derive(Parser, Debug, Serialize, Deserialize, Clone, Jsonl, Orfile)]
+#[clap(help_expected = true)]
+pub struct Config {
+	/// The node config to use.
+	#[clap(long)]
+	pub node_config: NodeConfigWrapper,
+
+	/// The log file to use.
+	#[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 {
+	/// Builds the config into a [MovementAptos] runner.
+	pub fn build(&self) -> Result<MovementAptos, ConfigError> {
+		Ok(MovementAptos::new(
+			self.node_config.0.clone(),
+			self.log_file.clone(),
+			self.create_global_rayon_pool,
+		))
+	}
+}

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

@@ -0,0 +1,5 @@
+pub mod config;
+pub mod movement_aptos;
+
+pub use config::*;
+pub use movement_aptos::*;

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

@@ -0,0 +1,135 @@
+use aptos_config::config::NodeConfig;
+use kestrel::State;
+use std::path::PathBuf;
+pub mod rest_api;
+use kestrel::fulfill::custom::Custom;
+pub use rest_api::RestApi;
+use rest_api::WaitForRestApi;
+
+/// Errors thrown when running [MovementAptos].
+#[derive(Debug, thiserror::Error)]
+pub enum MovementAptosError {
+	#[error("Movement Aptos failed to run with error: {0}")]
+	Internal(#[source] Box<dyn std::error::Error + Send + Sync>),
+}
+
+#[derive(Clone)]
+pub struct MovementAptos {
+	/// 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 [MovementAptosRestApi] for the Aptos node.
+	pub rest_api: State<RestApi>,
+}
+
+impl MovementAptos {
+	pub fn 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() }
+	}
+
+	/// Borrow the rest api state
+	pub fn rest_api(&self) -> &State<RestApi> {
+		&self.rest_api
+	}
+
+	/// 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(())
+	}
+
+	/// Runs the node and fills state.
+	pub async fn run(&self) -> Result<(), MovementAptosError> {
+		let rest_api = RestApi {
+			rest_api_url: format!(
+				"http://{}:{}",
+				self.node_config.api.address.ip(),
+				self.node_config.api.address.port()
+			),
+		};
+
+		let runner = self.clone();
+		let runner_task = kestrel::task(async move {
+			runner.run_node()?;
+			Ok::<_, MovementAptosError>(())
+		});
+
+		// rest api state
+		let rest_api_state = self.rest_api.clone();
+		let rest_api_polling = kestrel::task(async move {
+			loop {
+				// 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()))?;
+				if response.status().is_success() {
+					rest_api_state.write().set(rest_api).await;
+					break;
+				}
+			}
+
+			Ok::<_, MovementAptosError>(())
+		});
+
+		// await the runner
+		runner_task.await.map_err(|e| MovementAptosError::Internal(e.into()))??;
+		rest_api_polling.await.map_err(|e| MovementAptosError::Internal(e.into()))??;
+
+		Ok(())
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use aptos_node::create_single_node_test_config;
+	use rand::thread_rng;
+	use std::path::Path;
+
+	#[tokio::test]
+	async fn test_movement_aptos() -> Result<(), anyhow::Error> {
+		// open in a new db
+		let unique_id = uuid::Uuid::new_v4();
+		let timestamp = chrono::Utc::now().timestamp_millis();
+		let db_dir = Path::new(".debug").join(format!(
+			"movement-aptos-db-{}-{}",
+			timestamp,
+			unique_id.to_string().split('-').next().unwrap()
+		));
+
+		// create parent dirs
+		std::fs::create_dir_all(db_dir.clone())?;
+
+		let rng = thread_rng();
+		let node_config = create_single_node_test_config(
+			&None,
+			&None,
+			db_dir.as_path(),
+			true,
+			false,
+			true,
+			&aptos_cached_packages::head_release_bundle().clone(),
+			rng,
+		)?;
+
+		let movement_aptos = MovementAptos::new(node_config, None, false);
+		let rest_api_state = movement_aptos.rest_api().read().clone();
+		movement_aptos.run().await?;
+		rest_api_state.wait_for(tokio::time::Duration::from_secs(30)).await?;
+
+		Ok(())
+	}
+}

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

@@ -0,0 +1,39 @@
+use kestrel::{fulfill::custom::CustomProcessor, fulfill::FulfillError};
+use tokio::sync::mpsc::Receiver;
+
+#[derive(Debug, Clone)]
+pub struct RestApi {
+	/// The Rest Api url.
+	pub rest_api_url: String,
+}
+
+pub struct WaitForRestApi {
+	/// The expected [RestApi]
+	pub rest_api: RestApi,
+}
+
+impl WaitForRestApi {
+	pub fn new(rest_api: RestApi) -> Self {
+		Self { rest_api }
+	}
+}
+
+impl CustomProcessor<RestApi> for WaitForRestApi {
+	async fn process_receiver(
+		&self,
+		receiver: &mut Receiver<String>,
+	) -> Result<Option<RestApi>, FulfillError> {
+		loop {
+			// use the receiver to pace the polling
+			receiver.recv().await;
+
+			// keep curling the rest api until the faucet is ready
+			let response = reqwest::get(self.rest_api.rest_api_url.clone())
+				.await
+				.map_err(|e| FulfillError::Internal(e.into()))?;
+			if response.status().is_success() {
+				return Ok(Some(self.rest_api.clone()));
+			}
+		}
+	}
+}