Merge pull request #106 from movementlabsxyz/l-monninger/expose-faucet-api

feat: expose `movement-aptos` faucet API

Author: l-monninger

Cargo.lock

@@ -37,8 +37,8 @@ impl MovementAptosMigrator {
 	}
 
 	/// Builds a [MovementAptosMigrator] from a [NodeConfig].
-	pub fn from_config(config: NodeConfig) -> Result<Self, anyhow::Error> {
-		let movement_aptos = MovementAptos::from_config(config)?;
+	pub fn try_from_config(config: NodeConfig) -> Result<Self, anyhow::Error> {
+		let movement_aptos = MovementAptos::try_from_config(config, None)?;
 		let runner = Runner::MovementAptos(movement_aptos);
 		Ok(Self::new(runner))
 	}
@@ -47,7 +47,7 @@ impl MovementAptosMigrator {
 	pub fn from_movement_aptos_node(node: MovementAptosNode) -> Result<Self, anyhow::Error> {
 		// get the config from the node
 		let config = node.node_config()?;
-		Ok(Self::from_config(config)?)
+		Ok(Self::try_from_config(config)?)
 	}
 
 	/// Rest Api url for the runner.
@@ -68,6 +68,24 @@ impl MovementAptosMigrator {
 		}
 	}
 
+	/// Faucet Api url for the runner.
+	pub async fn wait_for_faucet_api_url(
+		&self,
+		condition: impl Into<WaitCondition>,
+	) -> Result<String, anyhow::Error> {
+		match &self.runner {
+			Runner::MovementAptos(movement_aptos) => {
+				let faucet_api = movement_aptos
+					.faucet_api()
+					.read()
+					.wait_for(condition)
+					.await
+					.context("failed to wait for Movement Aptos faucet api")?;
+				Ok(faucet_api.listen_url().to_string())
+			}
+		}
+	}
+
 	/// Waits for the rest client to be ready.
 	pub async fn wait_for_rest_client_ready(
 		&self,
@@ -83,15 +101,17 @@ impl MovementAptosMigrator {
 	/// Waits for the rest faucet client to be ready.
 	pub async fn wait_for_faucet_client_ready(
 		&self,
-		condition: impl Into<WaitCondition>,
+		condition: impl Into<WaitCondition> + Clone,
 	) -> Result<MovementAptosFaucetClient, anyhow::Error> {
-		// let rest_api_url = self.wait_for_rest_api_url(condition).await?;
-		// let faucet_client =
-		// 	MovementAptosFaucetClient::new(rest_api_url.parse().map_err(|e| {
-		// 		anyhow::anyhow!("failed to parse Movement Aptos rest api url: {}", e)
-		// 	})?);
-		// Ok(faucet_client)
-		todo!()
+		let rest_api_url = self.wait_for_rest_api_url(condition.clone()).await?.parse().map_err(|e| {
+			anyhow::anyhow!("failed to parse Movement Aptos rest api url: {}", e)
+		})?;
+		let faucet_api_url = self.wait_for_faucet_api_url(condition).await?.parse().map_err(|e| {
+			anyhow::anyhow!("failed to parse Movement Aptos faucet api url: {}", e)
+		})?;
+		let faucet_client =
+			MovementAptosFaucetClient::new(rest_api_url, faucet_api_url);
+		Ok(faucet_client)
 	}
 
 	/// Gets a [MovementAptosNode] from the runner.
@@ -105,7 +125,7 @@ impl TryFrom<NodeConfig> for MovementAptosMigrator {
 	type Error = anyhow::Error;
 
 	fn try_from(config: NodeConfig) -> Result<Self, Self::Error> {
-		Ok(Self::from_config(config)?)
+		Ok(Self::try_from_config(config)?)
 	}
 }
 

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

@@ -30,6 +30,7 @@ uuid = { workspace = true }
 serde_yaml = { workspace = true }
 tracing = { workspace = true }
 mtma-types = { workspace = true }
+portpicker = { workspace = true }
 
 [dev-dependencies]
 uuid = { workspace = true }

util/movement-aptos/core/Cargo.toml

@@ -63,6 +63,10 @@ pub struct Config {
 	#[clap(long)]
 	pub node_config: NodeConfigWrapper,
 
+	/// The faucet port to use.
+	#[clap(long)]
+	pub faucet_port: Option<u16>,
+
 	/// The log file to use.
 	#[orfile(config)]
 	#[clap(long)]
@@ -88,7 +92,7 @@ impl Config {
 
 		node_config.base.working_dir = Some(db_dir.clone());
 
-		Ok(Config { node_config: NodeConfigWrapper(node_config), log_file: None })
+		Ok(Config { node_config: NodeConfigWrapper(node_config), log_file: None, faucet_port: None })
 	}
 
 	/// Builds the config into a [MovementAptos] runner.
@@ -107,6 +111,7 @@ impl Config {
 
 		Ok(MovementAptos::<runtime::TokioTest>::new(
 			self.node_config.node_config().clone(),
+			self.faucet_port.unwrap_or_default(),
 			false,
 			workspace_dir,
 		))

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

@@ -3,12 +3,14 @@ use mtma_types::movement_aptos::aptos_config::config::NodeConfig;
 use std::path::PathBuf;
 pub mod rest_api;
 use kestrel::process::{command::Command, ProcessOperations};
+pub mod faucet_api;
 pub mod runtime;
 use anyhow::Context;
 pub use rest_api::RestApi;
+pub use faucet_api::FaucetApi;
 use runtime::Runtime;
 use std::marker::PhantomData;
-use tracing::{info, warn};
+use tracing::{info, debug, warn};
 
 /// Errors thrown when running [MovementAptos].
 #[derive(Debug, thiserror::Error)]
@@ -24,6 +26,8 @@ where
 {
 	/// The [NodeConfig]
 	pub node_config: NodeConfig,
+	/// The faucet port
+	pub faucet_port: u16,
 	/// Whether or not to multiprocess
 	pub multiprocess: bool,
 	/// The workspace for the multiprocessing should it occur
@@ -32,6 +36,8 @@ where
 	pub workspace: PathBuf,
 	/// The rest api state
 	pub rest_api: State<RestApi>,
+	/// The faucet api state
+	pub faucet_api: State<FaucetApi>,
 	/// The marker for the runtime
 	pub runtime: PhantomData<R>,
 }
@@ -41,26 +47,37 @@ 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, multiprocess: bool, workspace: PathBuf) -> Self {
-		Self { node_config, multiprocess, workspace, rest_api: State::new(), runtime: PhantomData }
+	pub fn new(node_config: NodeConfig, faucet_port: u16, multiprocess: bool, workspace: PathBuf) -> Self {
+		Self { node_config, faucet_port, multiprocess, workspace, rest_api: State::new(), faucet_api: State::new(), runtime: PhantomData }
 	}
 
 	/// Constructs a new [MovementAptos] from a [NodeConfig].
-	pub fn from_config(config_path: NodeConfig) -> Result<Self, MovementAptosError> {
+	pub fn try_from_config(config_path: NodeConfig, faucet_port: Option<u16>) -> Result<Self, MovementAptosError> {
+
+		let faucet_port = match faucet_port {
+			Some(port) => port,
+			None => portpicker::pick_unused_port().context("Failed to pick unused port").map_err(|e| MovementAptosError::Internal(e.into()))?
+		};
+
 		let workspace = config_path
 			.base
 			.working_dir
 			.clone()
 			.context("Working directory not set")
 			.map_err(|e| MovementAptosError::Internal(e.into()))?;
-		Ok(Self::new(config_path, true, workspace))
+		Ok(Self::new(config_path, faucet_port, true, workspace))
 	}
 
 	/// Borrow sthe rest api state
 	pub fn rest_api(&self) -> &State<RestApi> {
 		&self.rest_api
 	}
 
+	/// Borrow the faucet api state
+	pub fn faucet_api(&self) -> &State<FaucetApi> {
+		&self.faucet_api
+	}
+
 	/// Borrows the [NodeConfig]
 	pub fn node_config(&self) -> &NodeConfig {
 		&self.node_config
@@ -109,6 +126,8 @@ where
 			.await
 			.map_err(|e| MovementAptosError::Internal(e.into()))?;
 
+		// note: we could grab the entirety of the run-localnet [Args] struct and expose it, replacing the test dir and config path as is reasonable. 
+		// But, we will do that ad hoc.
 		let command = Command::line(
 			"aptos",
 			vec![
@@ -118,6 +137,8 @@ where
 				&self.workspace.to_string_lossy(),
 				"--config-path",
 				&config_path.to_string_lossy(),
+				"--faucet-port",
+				&self.faucet_port.to_string(),
 			],
 			Some(&self.workspace),
 			false,
@@ -143,6 +164,7 @@ where
 	/// 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) };
+		let faucet_api = FaucetApi { faucet_api_url: format!("http://127.0.0.1:{}", self.faucet_port) };
 
 		let runner = self.clone();
 		let runner_task = kestrel::task(async move {
@@ -157,23 +179,52 @@ where
 			tokio::time::sleep(std::time::Duration::from_secs(5)).await;
 			loop {
 				tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-				info!("POLLING REST API: {:?}", rest_api);
+				debug!("Polling rest api: {:?}", rest_api);
 				// wait for the rest api to be ready
 				match reqwest::get(rest_api.rest_api_url.clone())
 					.await
 					.map_err(|e| MovementAptosError::Internal(e.into()))
 				{
 					Ok(response) => {
-						info!("REST API RESPONSE: {:?}", response);
+						debug!("Received response from rest api: {:?}", response);
 						if response.status().is_success() {
 							rest_api_state.write().set(rest_api).await;
 							break;
 						} else {
-							warn!("REST API RESPONSE: {:?}", response);
+							warn!("Failed to poll rest api: {:?}", response);
+						}
+					}
+					Err(e) => {
+						warn!("Encountered error while polling rest api: {:?}", e);
+					}
+				}
+			}
+
+			Ok::<_, MovementAptosError>(())
+		});
+
+		// faucet api state
+		let faucet_api_state = self.faucet_api.clone();
+		let faucet_api_polling = kestrel::task(async move {
+			loop {
+				tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+				debug!("Polling faucet api: {:?}", faucet_api);
+				// wait for the faucet api to be ready
+				match reqwest::get(faucet_api.faucet_api_url.clone())
+					.await
+					.map_err(|e| MovementAptosError::Internal(e.into()))
+				{
+					Ok(response) => {
+						debug!("Received response from faucet api: {:?}", response);
+						if response.status().is_success() {
+							faucet_api_state.write().set(faucet_api).await;
+							break;
+						} else {
+							warn!("Failed to poll faucet api: {:?}", response);
 						}
 					}
 					Err(e) => {
-						warn!("REST API ERROR: {:?}", e);
+						warn!("Encountered error while polling faucet api: {:?}", e);
 					}
 				}
 			}
@@ -184,6 +235,7 @@ where
 		// await the runner
 		runner_task.await.map_err(|e| MovementAptosError::Internal(e.into()))??;
 		rest_api_polling.await.map_err(|e| MovementAptosError::Internal(e.into()))??;
+		faucet_api_polling.await.map_err(|e| MovementAptosError::Internal(e.into()))??;
 
 		Ok(())
 	}
@@ -208,6 +260,7 @@ mod tests {
 			unique_id.to_string().split('-').next().unwrap()
 		));
 		let db_dir = working_dir.join("0");
+		let faucet_port = portpicker::pick_unused_port().context("Failed to pick unused port").map_err(|e| MovementAptosError::Internal(e.into()))?;
 
 		// create parent dirs
 		{
@@ -229,15 +282,19 @@ mod tests {
 		node_config.storage.dir = db_dir.clone();
 
 		let movement_aptos =
-			MovementAptos::<runtime::Delegated>::new(node_config, true, working_dir);
+			MovementAptos::<runtime::Delegated>::new(node_config, faucet_port, true, working_dir);
+
+		// extract the states for which we will wait
 		let rest_api_state = movement_aptos.rest_api().read().clone();
+		let faucet_api_state = movement_aptos.faucet_api().read().clone();
 
 		let movement_aptos_task = kestrel::task(async move {
 			movement_aptos.run().await?;
 			Ok::<_, MovementAptosError>(())
 		});
 
 		rest_api_state.wait_for(tokio::time::Duration::from_secs(120)).await?;
+		faucet_api_state.wait_for(tokio::time::Duration::from_secs(120)).await?;
 
 		kestrel::end!(movement_aptos_task)?;
 

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

@@ -0,0 +1,16 @@
+#[derive(Debug, Clone)]
+pub struct FaucetApi {
+	/// The Rest Api url.
+	pub faucet_api_url: String,
+}
+
+impl FaucetApi {
+	pub fn new(faucet_api_url: String) -> Self {
+		Self { faucet_api_url }
+	}
+
+	/// Borrow the rest api listen url.
+	pub fn listen_url(&self) -> &str {
+		&self.faucet_api_url
+	}
+}

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

@@ -108,6 +108,7 @@ Run run with all parameters passed explicitly as CLI flags. See Orfile documenta
 ###### **Options:**
 
 * `--node-config <NODE_CONFIG>` — The node config to use
+* `--faucet-port <FAUCET_PORT>` — The faucet port to use
 * `--log-file <LOG_FILE>` — The log file to use