401 fix issues of connecting to stellar overlay network on testnet (#414

)

* a 3rd try of #409

* cargo fmt and fix warnings in
https://github.com/pendulum-chain/spacewalk/actions/runs/6459806136/job/17536291785?pr=414

* update the `substrate-stellar-sdk` version

* #414 (comment),
#414 (comment),
#414 (comment),
#414 (comment)

clients/runtime/client/Cargo.toml

@@ -25,7 +25,7 @@ thiserror = "1.0.23"
 
 subxt = "0.25.0"
 
-sc-client-db = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v0.9.40" }
+sc-client-db = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v0.9.40", default-features = false }
 sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v0.9.40", default-features = false }
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v0.9.40", default-features = false }
 sc-service = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v0.9.40", default-features = false }

clients/stellar-relay-lib/Cargo.toml

@@ -43,4 +43,9 @@ tokio = { version = "1.0", features = [
 ] }
 
 [features]
-default = []
+std = [
+    "substrate-stellar-sdk/std"
+]
+default = [
+    "std"
+]

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

@@ -1,19 +1,16 @@
 use rand::Rng;
 use sha2::{Digest, Sha256};
 use std::time::{SystemTime, UNIX_EPOCH};
-use substrate_stellar_sdk::{
-	types::{TransactionSet, Uint256},
-	SecretKey, XdrCodec,
-};
+use substrate_stellar_sdk::{types::Uint256, SecretKey};
 
-/// a helpful macro to unwrap an `Ok` or return immediately.
+/// a helpful macro to log an error (if it occurs) and return immediately.
 macro_rules! log_error {
 	// expression, return value, extra log
 	($res:expr, $log:expr) => {
-		$res.map_err(|e| {
+		if let Err(e) = $res {
 			log::error!("{:?}: {e:?}", $log);
-			e
-		})?;
+			return
+		}
 	};
 }
 
@@ -41,15 +38,3 @@ pub fn time_now() -> u64 {
 		u64::MAX
 	})
 }
-
-//todo: this has to be moved somewhere.
-pub fn compute_non_generic_tx_set_content_hash(tx_set: &TransactionSet) -> [u8; 32] {
-	let mut hasher = Sha256::new();
-	hasher.update(tx_set.previous_ledger_hash);
-
-	tx_set.txes.get_vec().iter().for_each(|envlp| {
-		hasher.update(envlp.to_xdr());
-	});
-
-	hasher.finalize().as_slice().try_into().unwrap()
-}

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

@@ -152,7 +152,7 @@ pub(crate) async fn receiving_service(
 	actions_sender: mpsc::Sender<ConnectorActions>,
 	timeout_in_secs: u64,
 	retries: u8,
-) -> Result<(), Error> {
+) {
 	let mut retry = 0;
 	let mut retry_read = 0;
 	let mut proc_id = 0;
@@ -171,9 +171,8 @@ pub(crate) async fn receiving_service(
 		{
 			Ok(Ok(0)) => {
 				if retry_read >= retries {
-					return Err(Error::ReadFailed(format!(
-						"Failed to read messages from the stream. Received 0 size more than {retries} times"
-					)))
+					log::error!("proc_id: {proc_id}. Failed to read messages from the stream. Received 0 size more than {retries} times");
+					return
 				}
 				retry_read += 1;
 			},
@@ -213,18 +212,20 @@ pub(crate) async fn receiving_service(
 				retry = 0;
 				retry_read = 0;
 				// let's read the continuation number of bytes from the previous message.
-				read_unfinished_message(
-					&mut r_stream,
-					&actions_sender,
-					&mut lack_bytes_from_prev,
-					&mut proc_id,
-					&mut readbuf,
-				)
-				.await?;
+				log_error!(
+					read_unfinished_message(
+						&mut r_stream,
+						&actions_sender,
+						&mut lack_bytes_from_prev,
+						&mut proc_id,
+						&mut readbuf,
+					).await,
+					format!("proc_id:{proc_id}. Error occurred while reading unfinished stellar message")
+				);
 			},
 			Ok(Err(e)) => {
 				log::error!("proc_id: {proc_id}. Error occurred while reading the stream: {e:?}");
-				return Err(Error::ConnectionFailed(e.to_string()))
+				return
 			},
 			Err(elapsed) => {
 				log::error!(
@@ -233,9 +234,8 @@ pub(crate) async fn receiving_service(
 				);
 
 				if retry >= retries {
-					return Err(Error::ConnectionFailed(
-						"TIMED OUT reading for messages from the stream".to_string(),
-					))
+					log::error!("proc_id: {proc_id}. Exhausted maximum retries for reading messages from Stellar Node.");
+					return
 				}
 				retry += 1;
 			},
@@ -285,21 +285,19 @@ pub(crate) async fn connection_handler(
 	mut connector: Connector,
 	mut actions_receiver: mpsc::Receiver<ConnectorActions>,
 	mut w_stream: tcp::OwnedWriteHalf,
-) -> Result<(), Error> {
+) {
 	let mut timeout_counter = 0;
 	loop {
 		match timeout(Duration::from_secs(connector.timeout_in_secs), actions_receiver.recv()).await
 		{
 			Ok(Some(ConnectorActions::Disconnect)) => {
-				w_stream.shutdown().await.map_err(|e| {
-					log::error!("Failed to shutdown write half of stream: {:?}", e);
-
-					Error::ConnectionFailed("Failed to disconnect tcp stream".to_string())
-				})?;
-
+				log_error!(
+					w_stream.shutdown().await,
+					format!("Failed to shutdown write half of stream:")
+				);
 				drop(connector);
 				drop(actions_receiver);
-				return Ok(())
+				return
 			},
 
 			Ok(Some(action)) => {
@@ -317,11 +315,11 @@ pub(crate) async fn connection_handler(
 			Err(elapsed) => {
 				log::error!("Connection timed out after {} seconds", elapsed.to_string());
 				if timeout_counter >= connector.retries {
-					connector.send_to_user(StellarRelayMessage::Timeout).await?;
-					return Err(Error::ConnectionFailed(format!(
-						"Timed out! elapsed time: {:?}",
-						elapsed.to_string()
-					)))
+					log_error!(
+						connector.send_to_user(StellarRelayMessage::Timeout).await,
+						format!("Connection Timed out:")
+					);
+					return
 				}
 				timeout_counter += 1;
 			},

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

@@ -37,6 +37,7 @@ pub(crate) fn from_authenticated_message(message: &AuthenticatedMessage) -> Resu
 	message_to_bytes(message)
 }
 
+// todo: move to substrate-stellar-sdk
 /// To easily convert any bytes to a Stellar type.
 ///
 /// # Examples

clients/stellar-relay-lib/src/tests/mod.rs

@@ -1,18 +1,15 @@
 use std::{sync::Arc, time::Duration};
 use substrate_stellar_sdk::{
 	types::{ScpStatementExternalize, ScpStatementPledges, StellarMessage},
-	Hash,
+	Hash, IntoHash,
 };
 
 use crate::{
 	node::NodeInfo, ConnectionInfo, StellarOverlayConfig, StellarOverlayConnection,
 	StellarRelayMessage,
 };
 use serial_test::serial;
-use tokio::{
-	sync::Mutex,
-	time::{sleep, timeout},
-};
+use tokio::{sync::Mutex, time::timeout};
 
 fn secret_key(is_mainnet: bool) -> String {
 	let path = if is_mainnet {
@@ -110,14 +107,16 @@ async fn stellar_overlay_should_receive_tx_set() {
 		scp_value[0..32].try_into().unwrap()
 	}
 
-	let (node_info, conn_info) = overlay_infos(true);
+	let (node_info, conn_info) = overlay_infos(false);
 	let overlay_connection = Arc::new(Mutex::new(
 		StellarOverlayConnection::connect(node_info, conn_info).await.unwrap(),
 	));
 
 	let ov_conn = overlay_connection.clone();
-	let tx_set_vec = Arc::new(Mutex::new(vec![]));
-	let tx_set_vec_clone = tx_set_vec.clone();
+	let tx_set_hashes = Arc::new(Mutex::new(vec![]));
+	let actual_tx_set_hashes = Arc::new(Mutex::new(vec![]));
+	let tx_set_hashes_clone = tx_set_hashes.clone();
+	let actual_tx_set_hashes_clone = actual_tx_set_hashes.clone();
 
 	timeout(Duration::from_secs(500), async move {
 		let mut ov_conn_locked = ov_conn.lock().await;
@@ -128,16 +127,24 @@ async fn stellar_overlay_should_receive_tx_set() {
 					StellarMessage::ScpMessage(msg) =>
 						if let ScpStatementPledges::ScpStExternalize(stmt) = &msg.statement.pledges
 						{
-							let txset_hash = get_tx_set_hash(stmt);
+							let tx_set_hash = get_tx_set_hash(stmt);
+							tx_set_hashes_clone.lock().await.push(tx_set_hash.clone());
 							ov_conn_locked
-								.send(StellarMessage::GetTxSet(txset_hash))
+								.send(StellarMessage::GetTxSet(tx_set_hash))
 								.await
 								.unwrap();
-							// let it sleep to wait for the `TxSet` message to appear
-							sleep(Duration::from_secs(10)).await;
 						},
 					StellarMessage::TxSet(set) => {
-						tx_set_vec_clone.lock().await.push(set);
+						let tx_set_hash = set.into_hash().expect("should return a hash");
+						actual_tx_set_hashes_clone.lock().await.push(tx_set_hash);
+
+						ov_conn_locked.disconnect().await.expect("failed to disconnect");
+						break
+					},
+					StellarMessage::GeneralizedTxSet(set) => {
+						let tx_set_hash = set.into_hash().expect("should return a hash");
+						actual_tx_set_hashes_clone.lock().await.push(tx_set_hash);
+
 						ov_conn_locked.disconnect().await.expect("failed to disconnect");
 						break
 					},
@@ -150,9 +157,14 @@ async fn stellar_overlay_should_receive_tx_set() {
 	.await
 	.expect("time has elapsed");
 
-	//arrange
 	//ensure that we receive some tx set from stellar node
-	assert!(!tx_set_vec.lock().await.is_empty());
+	let expected_hashes = tx_set_hashes.lock().await;
+	assert!(!expected_hashes.is_empty());
+
+	let actual_hashes = actual_tx_set_hashes.lock().await;
+	assert!(!actual_hashes.is_empty());
+
+	assert!(expected_hashes.contains(&actual_hashes[0]))
 }
 
 #[tokio::test]

clients/vault/Cargo.toml

@@ -6,6 +6,10 @@ name = "vault"
 version = "0.0.1"
 
 [features]
+std = [
+    "stellar-relay-lib/std"
+]
+
 integration = [
     "wallet/testing-utils"
 ]
@@ -53,8 +57,8 @@ jsonrpc-core-client = { version = "18.0.0", features = ["http", "tls"] }
 runtime = { path = "../runtime" }
 service = { path = "../service" }
 wallet = { path = "../wallet" }
-stellar-relay-lib = { package = "stellar-relay-lib", path = "../stellar-relay-lib" }
-primitives = { path = "../../primitives", package = "spacewalk-primitives" }
+stellar-relay-lib = { package = "stellar-relay-lib", path = "../stellar-relay-lib", default-features = false }
+primitives = { path = "../../primitives", package = "spacewalk-primitives", default-features = false }
 
 # Substrate dependencies
 sp-arithmetic = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v0.9.40" }

clients/vault/resources/samples/generalized_tx_set

@@ -0,0 +1 @@
+AAAAAVUcNbdctOS+OftZngTJY07YUUqb4P1I/owUmgdMuzscAAAAAgAAAAAAAAABAAAAAAAAAAEAAAAAAAAAZAAAAAEAAAACAAAAAMgOELhl9VFf5x0pG1aY8Mm/QQcnigdQ9MgWM1F8c6HSAAAAZAAZMt8AADrRAAAAAQAAAAAAAAAAAAAAAGUb2+8AAAABAAAAG3RzOjIwMjMtMTAtMDNUMDk6MTU6MDEuNTkyWgAAAAABAAAAAAAAAAEAAAAAf4MDV2AZH1oB1nouL9LSGUHGGafzcb48GXQyWFd9zswAAAAAAAAAAACYloAAAAAAAAAAAXxzodIAAABAEq3w/8HQ6kjqooVJPjg1TquL2pMOT+P9P7a3HpdqUYyFyJ8F32igbhIu3jvIJkafhDTosuL/rid2BxmScxhfDwAAAAAAAAAA

clients/vault/src/oracle/agent.rs

@@ -16,7 +16,7 @@ use crate::oracle::{
 	collector::ScpMessageCollector,
 	errors::Error,
 	types::{Slot, StellarMessageSender},
-	Proof,
+	AddTxSet, Proof,
 };
 
 pub struct OracleAgent {
@@ -42,8 +42,14 @@ async fn handle_message(
 			StellarMessage::ScpMessage(env) => {
 				collector.write().await.handle_envelope(env, message_sender).await?;
 			},
-			StellarMessage::TxSet(set) => {
-				collector.read().await.handle_tx_set(set);
+			StellarMessage::TxSet(set) =>
+				if let Err(e) = collector.read().await.add_txset(set) {
+					tracing::error!(e);
+				},
+			StellarMessage::GeneralizedTxSet(set) => {
+				if let Err(e) = collector.read().await.add_txset(set) {
+					tracing::error!(e);
+				}
 			},
 			_ => {},
 		},