Skip to content

Commit

Permalink
Allow Tokio Console support (#517)
Browse files Browse the repository at this point in the history
* first iteration:
add `tracing` feature of tokio, and the console-subscriber

* update on tokio version, based on:
tokio-rs/console#345 (comment)

* use `tracing` library, instead of log.

* remove unnecessary logs;
add necessary logs;
add timeout in `read_message_from_stellar` because it gets stuck

* cleanup zombie task

* Update README.md

add documentation of the `tokio-console`.

* remove duplicate `Parachain Block Listener`

* https://github.com/pendulum-chain/spacewalk/pull/517/files#r1600267305

* #517 (comment),
#517 (comment),

* https://github.com/pendulum-chain/spacewalk/actions/runs/9094123182/job/24994504193?pr=517

* remocehttps://github.com/pendulum-chain/spacewalk/actions/runs/9095519314/job/24998905803?pr=517

* https://github.com/pendulum-chain/spacewalk/actions/runs/9096987912/job/25003758078?pr=517

* https://github.com/pendulum-chain/spacewalk/actions/runs/9098087121/job/25007476694?pr=517

* https://github.com/pendulum-chain/spacewalk/actions/runs/9108826563/job/25040418069?pr=517
  • Loading branch information
b-yap authored May 22, 2024
1 parent 8331813 commit 93ddbce
Show file tree
Hide file tree
Showing 29 changed files with 474 additions and 141 deletions.
288 changes: 268 additions & 20 deletions Cargo.lock

Large diffs are not rendered by default.

39 changes: 38 additions & 1 deletion clients/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,43 @@ If the transaction submission fails giving a `tx_failed` in the `result_codes` o
due to the converted destination account not having trustlines set up for the redeemed asset.
The destination account is derived automatically from the account that called the extrinsic on-chain.

### Debugging with `tokio-console`
The vault is `tokio-console` ready, with the feature **_`allow-debugger`_**. _Remember to [set the rustflags](https://github.com/tokio-rs/console?tab=readme-ov-file#instrumenting-your-program)!_
```
RUSTFLAGS="--cfg tokio_unstable" cargo run --bin vault --features allow-debugger
```
[Install tokio-console](https://github.com/tokio-rs/console?tab=readme-ov-file#running-the-console)
and connect to the vault.
If using the [testchain](../testchain) and vault's `standalone-metadata` feature, you can use the command:
```
tokio-console http://127.0.0.1:6669
```
This will display:
<img width="1138" alt="Screenshot 2024-05-13 at 6 33 33 PM" src="https://github.com/pendulum-chain/spacewalk/assets/2826165/6681a16c-84c0-47f5-abc6-e4ba6a7dc032">
The multiple ` tokio::task clients/vault/src/system.rs ` tasks follows the tasks spawned consecutively, in [system.rs](https://github.com/pendulum-chain/spacewalk/blob/main/clients/vault/src/system.rs):

* The first 4 tasks are from `fn create_initial_tasks(...)` :
* VaultId Registration Listener
* Restart Timer
* Stellar Transaction Listener
* Parachain Block Listener
* Next 5 tasks from `fn create_issue_tasks(...)` :
* Issue Request Listener
* Issue Cancel Listener
* Issue Execute Listener
* Issue Executor
* Issue Cancel Scheduler
* Next 4 tasks from `fn create_replace_tasks(...)` :
* Request Replace Listener
* Accept Replace Listener
* Execute Replace Listener
* Replace Cancellation Scheduler
* Redeem Request Listener
* The last 2 tasks from `create_bridge_metrics_tasks(...)` :
* Bridge Metrics Listener
* Bridge Metrics Poller


## Notes on the implementation of subxt

This section is supposed to help when encountering issues with communication of vault client and parachain.
Expand Down Expand Up @@ -257,4 +294,4 @@ found [here](https://docs.rs/subxt-macro/latest/subxt_macro/#adding-derives-for-
When the compiler complains about mismatched types although the types seem to be the same, you might have to use type
substitutions.
This is done by adding the `#[subxt(substitute_type = "some type")]` attribute to the metadata module.
More documentation can be found [here](https://docs.rs/subxt-macro/latest/subxt_macro/#substituting-types).
More documentation can be found [here](https://docs.rs/subxt-macro/latest/subxt_macro/#substituting-types).
2 changes: 1 addition & 1 deletion clients/runner/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ edition = "2021"
[dependencies]
clap = { version = "4.0.17", features = ["derive"]}
hex = "0.4.3"
tokio = { version = "1.8", features = ["rt-multi-thread", "macros", "time"] }
tokio = { version = "1.37", features = ["rt-multi-thread", "macros", "time"] }
codec = { package = "parity-scale-codec", version = "3.0.0", default-features = false, features = ["derive", "full", "bit-vec"] }
sp-core = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v0.9.42" }
thiserror = "1.0.0"
Expand Down
2 changes: 1 addition & 1 deletion clients/runtime/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ log = "0.4.0"
serde = { version = "1.0.136", features = ["derive"] }
serde_json = "1.0.71"
thiserror = "1.0"
tokio = { version = "1.0", features = ["full"] }
tokio = { version = "1.37", features = ["full"] }
prometheus = { version = "0.12.0", features = ["process"] }
url = "2"

Expand Down
2 changes: 1 addition & 1 deletion clients/runtime/client/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ description = "Embed a substrate node into your subxt application."
keywords = ["parity", "substrate", "blockchain"]

[dependencies]
tokio = { version = "1.10", features = ["time", "rt-multi-thread"] }
tokio = { version = "1.37", features = ["time", "rt-multi-thread"] }
futures = { version = "0.3.9", features = ["compat"], package = "futures" }
futures01 = { package = "futures", version = "0.1.29" }
jsonrpsee = "0.16.0"
Expand Down
2 changes: 1 addition & 1 deletion clients/service/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ thiserror = "1.0"

hyper = { version = "0.14.11" }
hyper-tls = "0.5.0"
tokio = { version = "1.0", features = ["full"] }
tokio = { version = "1.37", features = ["full"] }
warp = "0.3.2"

serde = { version = "1.0.136", features = ["derive"] }
Expand Down
7 changes: 4 additions & 3 deletions clients/stellar-relay-lib/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ wallet = { path = "../wallet", features = ["testing-utils"] }

[dependencies]
hex = "0.4.3"
log = {version = "0.4.14"}
tracing = { version = "0.1", features = ["log"] }

base64 = "0.13.0"
rand = "0.8.5"
Expand All @@ -34,11 +34,12 @@ substrate-stellar-sdk = {git = "https://github.com/pendulum-chain/substrate-stel

err-derive = "0.3.1"

tokio = { version = "1.0", features = [
tokio = { version = "1.37", features = [
"macros", # allows main function to be async
"rt-multi-thread", # for multi-thread runtime
"sync", # to make channels available
"time" # for timeouts and sleep, when reconnecting
"time", # for timeouts and sleep, when reconnecting
"tracing" # for tokio console
] }
async-std = { version = "1.12.0", features = ["attributes"] }

Expand Down
2 changes: 1 addition & 1 deletion clients/stellar-relay-lib/examples/connect.rs
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
ScpStatementPledges::ScpStExternalize(_) => "ScpStExternalize",
ScpStatementPledges::ScpStNominate(_) => "ScpStNominate ",
};
log::info!(
tracing::info!(
"{} sent StellarMessage of type {} for ledger {}",
node_id,
stmt_type,
Expand Down
2 changes: 1 addition & 1 deletion clients/stellar-relay-lib/src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ impl StellarOverlayConfig {

let public_key = secret_key.get_public().to_encoding();
let public_key = std::str::from_utf8(&public_key).unwrap();
log::info!(
tracing::info!(
"connection_info(): Connecting to Stellar overlay network using public key: {public_key}"
);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ pub fn create_auth_cert(

let signature: Signature = Signature::new(keypair.create_signature(raw_sig_data).to_vec())
.map_err(|e| {
log::error!("create_auth_cert(): {e:?}");
tracing::error!("create_auth_cert(): {e:?}");
Error::AuthSignatureFailed
})?;

Expand Down Expand Up @@ -87,7 +87,7 @@ pub fn verify_remote_auth_cert(
match auth_cert_sig.try_into() {
Ok(raw_sig) => remote_pub_key.verify_signature(raw_data, &raw_sig),
Err(_) => {
log::warn!(
tracing::warn!(
"failed to convert auth cert signature of size {} to fixed array of 64.",
sig_len
);
Expand Down
11 changes: 3 additions & 8 deletions clients/stellar-relay-lib/src/connection/connector/connector.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ use substrate_stellar_sdk::{
types::{AuthenticatedMessageV0, Curve25519Public, HmacSha256Mac, MessageType},
XdrCodec,
};
use tracing::{error, trace};

use crate::{
connection::{
Expand Down Expand Up @@ -86,18 +87,12 @@ impl Connector {
body: &[u8],
) -> Result<(), Error> {
let remote_info = self.remote_info.as_ref().ok_or(Error::NoRemoteInfo)?;
log::trace!(
trace!(
"verify_auth(): remote sequence: {}, auth message sequence: {}",
remote_info.sequence(),
auth_msg.sequence
);

let auth_msg_xdr = auth_msg.to_base64_xdr();
let auth_msg_xdr =
String::from_utf8(auth_msg_xdr.clone()).unwrap_or(format!("{:?}", auth_msg_xdr));

log::debug!("verify_auth(): received auth message from Stellar Node: {auth_msg_xdr}");

if remote_info.sequence() != auth_msg.sequence {
// must be handled on main thread because workers could mix up order of messages.
return Err(Error::InvalidSequenceNumber)
Expand Down Expand Up @@ -169,7 +164,7 @@ impl Connector {

pub fn stop(&mut self) {
if let Err(e) = self.tcp_stream.shutdown(Shutdown::Both) {
log::error!("stop(): failed to shutdown tcp stream: {}", e);
error!("stop(): failed to shutdown tcp stream: {}", e);
}
}
}
Expand Down
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
use substrate_stellar_sdk::{
types::{ErrorCode, Hello, MessageType, StellarMessage},
XdrCodec,
};

use crate::connection::{
authentication::verify_remote_auth_cert,
helper::{error_to_string, time_now},
hmac::HMacKeys,
xdr_converter::parse_authenticated_message,
Connector, Error, Xdr,
};
use substrate_stellar_sdk::{
types::{ErrorCode, Hello, MessageType, StellarMessage},
XdrCodec,
};
use tracing::{error, info, trace, warn};

use crate::node::RemoteInfo;

Expand All @@ -33,13 +33,13 @@ impl Connector {

MessageType::ErrorMsg => match auth_msg.message {
StellarMessage::ErrorMsg(e) => {
log::error!(
error!(
"process_raw_message(): Received ErrorMsg during authentication: {}",
error_to_string(e.clone())
);
return Err(Error::from(e))
},
other => log::error!(
other => error!(
"process_raw_message(): Received ErrorMsg during authentication: {:?}",
other
),
Expand All @@ -50,9 +50,7 @@ impl Connector {
if self.is_handshake_created() {
self.verify_auth(&auth_msg, &data[4..(data.len() - 32)])?;
self.increment_remote_sequence()?;
log::trace!(
"process_raw_message(): Processing {msg_type:?} message: auth verified"
);
trace!("process_raw_message(): Processing {msg_type:?} message: auth verified");
}

return self.process_stellar_message(auth_msg.message, msg_type).await
Expand Down Expand Up @@ -80,29 +78,23 @@ impl Connector {
} else {
self.send_auth_message().await?;
}
log::info!("process_stellar_message(): Hello message processed successfully");
info!("process_stellar_message(): Hello message processed successfully");
},

StellarMessage::Auth(_) => {
self.process_auth_message().await?;
},

StellarMessage::ErrorMsg(e) => {
log::error!(
"process_stellar_message(): Received ErrorMsg during authentication: {e:?}"
);
error!("process_stellar_message(): Received ErrorMsg during authentication: {e:?}");
if e.code == ErrorCode::ErrConf || e.code == ErrorCode::ErrAuth {
return Err(Error::from(e))
}
return Ok(Some(StellarMessage::ErrorMsg(e)))
},

// we do not handle other messages. Return to caller
other => {
log::trace!(
"process_stellar_message(): Processing {} message: received from overlay",
String::from_utf8(other.to_base64_xdr())
.unwrap_or(format!("{:?}", other.to_base64_xdr()))
);
self.check_to_send_more(msg_type).await?;
return Ok(Some(other))
},
Expand All @@ -124,7 +116,7 @@ impl Connector {
remote.node().overlay_version,
);
} else {
log::warn!("process_auth_message(): No remote overlay version after handshake.");
warn!("process_auth_message(): No remote overlay version after handshake.");
}

self.check_to_send_more(MessageType::Auth).await
Expand Down
Loading

0 comments on commit 93ddbce

Please sign in to comment.