Rewrite Task of listening Stellar Messages

.github/workflows/ci-main.yml

@@ -27,7 +27,7 @@ jobs:
       - name: Check Build
         run: |
           bash ./scripts/cmd-all build check "--release"
-          
+
       - name: Clippy -- Libraries and Binaries
         run: |
           bash ./scripts/cmd-all clippy "clippy --lib --bins" "--release -- -W clippy::all -A clippy::style -A forgetting_copy_types -A forgetting_references"

clients/service/src/lib.rs

@@ -10,7 +10,10 @@ use async_trait::async_trait;
 use futures::{future::Either, Future, FutureExt};
 use governor::{Quota, RateLimiter};
 use nonzero_ext::*;
-use tokio::{sync::RwLock, time::sleep};
+use tokio::{
+	sync::{broadcast::error::TryRecvError, RwLock},
+	time::sleep,
+};
 pub use warp;
 
 pub use cli::{LoggingFormat, MonitoringConfig, RestartPolicy, ServiceConfig};
@@ -170,17 +173,38 @@ impl<Config: Clone + Send + 'static, F: Fn()> ConnectionManager<Config, F> {
 	}
 }
 
-pub async fn wait_or_shutdown<F, E>(shutdown_tx: ShutdownSender, future2: F) -> Result<(), E>
+pub async fn wait_or_shutdown<F, E>(
+	shutdown_tx: ShutdownSender,
+	future2: F,
+	// a consumer that receives a precheck signal to start a task.
+	precheck_signal: Option<tokio::sync::broadcast::Receiver<()>>,
+) -> Result<(), E>
 where
 	F: Future<Output = Result<(), E>>,
 {
+	if let Some(mut precheck_signal) = precheck_signal {
+		loop {
+			match precheck_signal.try_recv() {
+				// Received a signal to start the task
+				Ok(_) => break,
+				Err(TryRecvError::Empty) =>
+					tracing::trace!("wait_or_shutdown precheck signal: waiting..."),
+				// Precheck signal failed. Cannot start the task.
+				Err(e) => {
+					tracing::error!("Error receiving precheck signal: {:?}", e);
+					return Ok(());
+				},
+			}
+		}
+	}
+
 	match run_cancelable(shutdown_tx.subscribe(), future2).await {
 		TerminationStatus::Cancelled => {
-			tracing::trace!("Received shutdown signal");
+			tracing::trace!("wait_or_shutdown(): Received shutdown signal");
 			Ok(())
 		},
 		TerminationStatus::Completed(res) => {
-			tracing::trace!("Sending shutdown signal");
+			tracing::trace!("wait_or_shutdown(): Sending shutdown signal");
 			let _ = shutdown_tx.send(());
 			res
 		},

clients/stellar-relay-lib/Cargo.toml

@@ -10,9 +10,10 @@ name = "stellar_relay_lib"
 path = "src/lib.rs"
 
 [dev-dependencies]
-env_logger = "0.9.0"
+ntest = "0.9.0"
 serial_test = "0.9.0"
 wallet = { path = "../wallet", features = ["testing-utils"] }
+console-subscriber = { version = "0.3.0" }
 
 [dependencies]
 hex = "0.4.3"

clients/stellar-relay-lib/examples/connect.rs

@@ -1,14 +1,14 @@
 use stellar_relay_lib::{
 	connect_to_stellar_overlay_network,
 	sdk::types::{ScpStatementPledges, StellarMessage},
-	StellarOverlayConfig,
+	Error, StellarOverlayConfig,
 };
 
 use wallet::keys::get_source_secret_key_from_env;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-	env_logger::init();
+	console_subscriber::init();
 
 	let args: Vec<String> = std::env::args().collect();
 	let arg_network = if args.len() > 1 { &args[1] } else { "testnet" };
@@ -22,33 +22,48 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
 	let secret_key = get_source_secret_key_from_env(arg_network == "mainnet");
 
-	let mut overlay_connection = connect_to_stellar_overlay_network(cfg, &secret_key).await?;
-
-	while let Ok(Some(msg)) = overlay_connection.listen() {
-		match msg {
-			StellarMessage::ScpMessage(msg) => {
-				let node_id = msg.statement.node_id.to_encoding();
-				let node_id = base64::encode(&node_id);
-				let slot = msg.statement.slot_index;
-
-				let stmt_type = match msg.statement.pledges {
-					ScpStatementPledges::ScpStPrepare(_) => "ScpStPrepare",
-					ScpStatementPledges::ScpStConfirm(_) => "ScpStConfirm",
-					ScpStatementPledges::ScpStExternalize(_) => "ScpStExternalize",
-					ScpStatementPledges::ScpStNominate(_) => "ScpStNominate ",
-				};
-				tracing::info!(
-					"{} sent StellarMessage of type {} for ledger {}",
-					node_id,
-					stmt_type,
-					slot
-				);
+	let mut overlay_connection = connect_to_stellar_overlay_network(cfg, secret_key).await?;
+
+	loop {
+		match overlay_connection.listen().await {
+			Ok(Some(msg)) => match msg {
+				StellarMessage::Hello(_) => {
+					tracing::info!("received Hello message");
+				},
+
+				StellarMessage::ScpMessage(msg) => {
+					let node_id = msg.statement.node_id.to_encoding();
+					let node_id = base64::encode(&node_id);
+					let slot = msg.statement.slot_index;
+
+					let stmt_type = match msg.statement.pledges {
+						ScpStatementPledges::ScpStPrepare(_) => "ScpStPrepare",
+						ScpStatementPledges::ScpStConfirm(_) => "ScpStConfirm",
+						ScpStatementPledges::ScpStExternalize(_) => "ScpStExternalize",
+						ScpStatementPledges::ScpStNominate(_) => "ScpStNominate ",
+					};
+					tracing::info!(
+						"{} sent StellarMessage of type {} for ledger {}",
+						node_id,
+						stmt_type,
+						slot
+					);
+				},
+				_ => {
+					let _ = overlay_connection.send_to_node(StellarMessage::GetPeers).await;
+				},
+			},
+			Ok(None) => {},
+			Err(Error::Timeout) => {
+				tracing::warn!("took more than a second to respond");
 			},
-			_ => {
-				let _ = overlay_connection.send_to_node(StellarMessage::GetPeers).await;
+			Err(e) => {
+				tracing::error!("Error: {:?}", e);
+				break
 			},
 		}
 	}
 
+	tracing::info!("ooops, connection stopped ");
 	Ok(())
 }

clients/stellar-relay-lib/src/config.rs

@@ -35,13 +35,16 @@ impl StellarOverlayConfig {
 
 	#[allow(dead_code)]
 	pub(crate) fn node_info(&self) -> NodeInfo {
-		self.node_info.clone().into()
+		NodeInfo::new(&self.node_info)
 	}
 
 	#[allow(dead_code)]
-	pub(crate) fn connection_info(&self, secret_key: &str) -> Result<ConnectionInfo, Error> {
+	pub(crate) fn connection_info(
+		&self,
+		secret_key_as_string: String,
+	) -> Result<ConnectionInfo, Error> {
 		let cfg = &self.connection_info;
-		let secret_key = SecretKey::from_encoding(secret_key)?;
+		let secret_key = SecretKey::from_encoding(secret_key_as_string)?;
 
 		let public_key = secret_key.get_public().to_encoding();
 		let public_key = std::str::from_utf8(&public_key).unwrap();
@@ -128,9 +131,9 @@ impl ConnectionInfoCfg {
 /// Returns the `StellarOverlayConnection` if connection is a success, otherwise an Error
 pub async fn connect_to_stellar_overlay_network(
 	cfg: StellarOverlayConfig,
-	secret_key: &str,
+	secret_key_as_string: String,
 ) -> Result<StellarOverlayConnection, Error> {
-	let conn_info = cfg.connection_info(secret_key)?;
+	let conn_info = cfg.connection_info(secret_key_as_string)?;
 	let local_node = cfg.node_info;
 
 	StellarOverlayConnection::connect(local_node.into(), conn_info).await

clients/stellar-relay-lib/src/connection/connector/connector.rs

@@ -282,7 +282,7 @@ mod test {
 		let cfg =
 			StellarOverlayConfig::try_from_path(cfg_file_path).expect("should create a config");
 		let node_info = cfg.node_info();
-		let conn_info = cfg.connection_info(&secret_key).expect("should create a connection info");
+		let conn_info = cfg.connection_info(secret_key).expect("should create a connection info");
 		// this is a channel to communicate with the connection/config (this needs renaming)
 
 		let connector = Connector::start(node_info.clone(), conn_info.clone())
@@ -333,7 +333,7 @@ mod test {
 			cert: new_auth_cert,
 			nonce: [0; 32],
 		};
-		connector.set_remote(RemoteInfo::new(&hello));
+		connector.set_remote(RemoteInfo::new(hello));
 
 		assert!(connector.remote().is_some());
 	}
@@ -357,7 +357,7 @@ mod test {
 			cert: new_auth_cert,
 			nonce: [0; 32],
 		};
-		connector.set_remote(RemoteInfo::new(&hello));
+		connector.set_remote(RemoteInfo::new(hello));
 		assert_eq!(connector.remote().unwrap().sequence(), 0);
 
 		connector.increment_remote_sequence().unwrap();
@@ -385,7 +385,7 @@ mod test {
 			cert: new_auth_cert,
 			nonce: [0; 32],
 		};
-		let remote = RemoteInfo::new(&hello);
+		let remote = RemoteInfo::new(hello);
 		let remote_nonce = remote.nonce();
 		connector.set_remote(remote.clone());
 

clients/stellar-relay-lib/src/connection/connector/message_creation.rs

@@ -3,7 +3,10 @@ use crate::connection::{
 	Connector, Error,
 };
 use substrate_stellar_sdk::{
-	types::{AuthenticatedMessage, AuthenticatedMessageV0, HmacSha256Mac, StellarMessage},
+	compound_types::LimitedString,
+	types::{
+		AuthenticatedMessage, AuthenticatedMessageV0, ErrorCode, HmacSha256Mac, StellarMessage,
+	},
 	XdrCodec,
 };
 
@@ -75,3 +78,14 @@ impl Connector {
 		)
 	}
 }
+
+/// Create our own error to send over to the user/outsider.
+pub(crate) fn crate_specific_error() -> StellarMessage {
+	let error = "Stellar Relay Error".as_bytes().to_vec();
+	let error = substrate_stellar_sdk::types::Error {
+		code: ErrorCode::ErrMisc,
+		msg: LimitedString::new(error).expect("should return a valid LimitedString"),
+	};
+
+	StellarMessage::ErrorMsg(error)
+}