Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Orchestrator thread keepalive #524

Open
wants to merge 11 commits into
base: main
Choose a base branch
from
+191 −1,849
Open

Orchestrator thread keepalive #524

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
bd594c1
Delete rust test runner
cbrit Jun 6, 2023
9dc7247
Task keep-alive loops
cbrit Jun 8, 2023
e31df4f
Move cosmos address derivation to beginning of orchestrator startup e…
cbrit Jun 8, 2023
180e699
Gravity ID retrieval in relayer binary startup
cbrit Jun 8, 2023
15a327f
Change batch size config value to usize to avoid fallable conversion
cbrit Jun 8, 2023
8ef1c96
Remove unused function check_for_eth
cbrit Jun 8, 2023
7298f80
Move block_delay query to startup
cbrit Jun 8, 2023
ba9c55a
Replace unwrap() with ? in signer and oracle loops where needed
cbrit Jun 8, 2023
ae54a0c
Add panic-handling and task restart to sender, signer, and oracle thr…
cbrit Jun 8, 2023
c7d1a56
Handle relayer panics, make error messages consistent
cbrit Jun 8, 2023
3f5ed47
Remove unecessary clone of cosmos address
cbrit Aug 1, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 0 additions & 32 deletions orchestrator/Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 1 addition & 3 deletions orchestrator/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,14 +12,13 @@ description = """
"""

[workspace]
default-members = ["gorc", "orchestrator", "test_runner"]
default-members = ["gorc", "orchestrator"]
members = [
"orchestrator",
"cosmos_gravity",
"ethereum_gravity",
"gravity_utils",
"gravity_proto_build",
"test_runner",
"gravity_proto",
"relayer",
"register_delegate_keys",
Expand All @@ -34,7 +33,6 @@ cosmos_gravity = { path = "./cosmos_gravity" }
ethereum_gravity = { path = "./ethereum_gravity" }
gravity_utils = { path = "./gravity_utils" }
gravity_proto_build = { path = "./gravity_proto_build" }
test_runner = { path = "./test_runner" }
gravity_proto = { path = "./gravity_proto" }
register_delegate_keys = { path = "./register_delegate_keys" }
gorc = { path = "./gorc" }
Expand Down
9 changes: 2 additions & 7 deletions orchestrator/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,10 +51,6 @@ This is to build the relayer logic (i.e. cosmos to ethereum) as a seperate binar

Supporting bash scripts for this library

### test_runner/

A binary which runs tests against a cosmos chain


## CLI

Expand All @@ -67,7 +63,6 @@ client deploy-erc20-representation --cosmos-grpc=<url> --cosmos-prefix=<prefix>
orchestrator --cosmos-phrase=<key> --ethereum-key=<key> --cosmos-grpc=<url> --address-prefix=<prefix> --ethereum-rpc=<url> --fees=<denom> --contract-address=<addr>
register-delegate-key --validator-phrase=<key> --address-prefix=<prefix> [--cosmos-phrase=<key>] [--ethereum-key=<key>] --cosmos-grpc=<url> --fees=<denom>
relayer --ethereum-key=<key> --cosmos-grpc=<url> --address-prefix=<prefix> --ethereum-rpc=<url> --contract-address=<addr> --cosmos-grpc=<gurl>
test_runner
```

## PROPOSED
Expand Down Expand Up @@ -105,7 +100,7 @@ gorc
update [name] [new-name]
list
show [name]
cosmos
cosmos
add [name]
import [name] [mnemnoic]
delete [name]
Expand All @@ -117,7 +112,7 @@ gorc
```json
[gravity]
contract = "0x6b175474e89094c44da98b954eedeac495271d0f"

[ethereum]
key = "testkey"
rpc = "http://localhost:8545"
Expand Down
27 changes: 6 additions & 21 deletions orchestrator/cosmos_gravity/src/build.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
use deep_space::Contact;
use deep_space::Msg;
use deep_space::{Address as CosmosAddress, Msg};
use ethereum_gravity::types::EthClient;
use ethers::prelude::*;
use ethers::utils::keccak256;
Expand All @@ -14,20 +13,16 @@ use lazy_static::lazy_static;
use regex::Regex;
use std::collections::BTreeMap;

use crate::crypto::PrivateKey as CosmosPrivateKey;

lazy_static! {
static ref DENOM_REGEX: Regex = Regex::new("^[a-zA-Z][a-zA-Z0-9/-]{2,127}$").unwrap();
}

pub async fn signer_set_tx_confirmation_messages(
contact: &Contact,
cosmos_address: CosmosAddress,
eth_client: EthClient,
valsets: Vec<Valset>,
cosmos_key: CosmosPrivateKey,
gravity_id: String,
) -> Vec<Msg> {
let cosmos_address = cosmos_key.to_address(&contact.get_prefix()).unwrap();
let ethereum_address = eth_client.address();

let mut msgs = Vec::new();
Expand All @@ -52,13 +47,11 @@ pub async fn signer_set_tx_confirmation_messages(
}

pub async fn batch_tx_confirmation_messages(
contact: &Contact,
cosmos_address: CosmosAddress,
eth_client: EthClient,
batches: Vec<TransactionBatch>,
cosmos_key: CosmosPrivateKey,
gravity_id: String,
) -> Vec<Msg> {
let cosmos_address = cosmos_key.to_address(&contact.get_prefix()).unwrap();
let ethereum_address = eth_client.address();

let mut msgs = Vec::new();
Expand All @@ -84,13 +77,11 @@ pub async fn batch_tx_confirmation_messages(
}

pub async fn contract_call_tx_confirmation_messages(
contact: &Contact,
cosmos_address: CosmosAddress,
eth_client: EthClient,
logic_calls: Vec<LogicCall>,
cosmos_key: CosmosPrivateKey,
gravity_id: String,
) -> Vec<Msg> {
let cosmos_address = cosmos_key.to_address(&contact.get_prefix()).unwrap();
let ethereum_address = eth_client.address();

let mut msgs = Vec::new();
Expand All @@ -117,12 +108,9 @@ pub async fn contract_call_tx_confirmation_messages(
}

pub async fn ethereum_vote_height_messages(
contact: &Contact,
cosmos_key: CosmosPrivateKey,
cosmos_address: CosmosAddress,
ethereum_height: U64,
) -> Vec<Msg> {
let cosmos_address = cosmos_key.to_address(&contact.get_prefix()).unwrap();

let msg = proto::MsgEthereumHeightVote {
ethereum_height: ethereum_height.as_u64(),
signer: cosmos_address.to_string(),
Expand All @@ -136,16 +124,13 @@ pub async fn ethereum_vote_height_messages(
}

pub fn ethereum_event_messages(
contact: &Contact,
cosmos_key: CosmosPrivateKey,
cosmos_address: CosmosAddress,
deposits: Vec<SendToCosmosEvent>,
batches: Vec<TransactionBatchExecutedEvent>,
erc20_deploys: Vec<Erc20DeployedEvent>,
logic_calls: Vec<LogicCallExecutedEvent>,
valsets: Vec<ValsetUpdatedEvent>,
) -> Vec<Msg> {
let cosmos_address = cosmos_key.to_address(&contact.get_prefix()).unwrap();

// This sorts oracle messages by event nonce before submitting them. It's not a pretty implementation because
// we're missing an intermediary layer of abstraction. We could implement 'EventTrait' and then implement sort
// for it, but then when we go to transform 'EventTrait' objects into GravityMsg enum values we'll have all sorts
Expand Down
53 changes: 42 additions & 11 deletions orchestrator/cosmos_gravity/src/send.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,10 @@ use gravity_proto::gravity as proto;
use gravity_utils::error::GravityError;
use gravity_utils::ethereum::format_eth_address;
use prost::Message;
use tokio::sync::Mutex;
use std::cmp;
use std::collections::HashSet;
use std::sync::Arc;
use std::{result::Result, time::Duration};

use crate::crypto::PrivateKey as CosmosPrivateKey;
Expand Down Expand Up @@ -67,7 +69,7 @@ pub async fn update_gravity_delegate_addresses(
};
let msg = Msg::new("/gravity.v1.MsgDelegateKeys", msg);

send_messages(contact, cosmos_key, gas_price, vec![msg], gas_adjustment).await
send_messages(contact.to_owned(), cosmos_key, gas_price, vec![msg], gas_adjustment).await
}

/// Sends tokens from Cosmos to Ethereum. These tokens will not be sent immediately instead
Expand All @@ -78,7 +80,7 @@ pub async fn send_to_eth(
amount: Coin,
bridge_fee: Coin,
gas_price: (f64, String),
contact: &Contact,
contact: Contact,
gas_adjustment: f64,
) -> Result<TxResponse, GravityError> {
if amount.denom != bridge_fee.denom {
Expand All @@ -90,7 +92,7 @@ pub async fn send_to_eth(
)));
}

let cosmos_address = cosmos_key.to_address(&contact.get_prefix()).unwrap();
let cosmos_address = cosmos_key.to_address(&contact.get_prefix())?;

let msg = proto::MsgSendToEthereum {
sender: cosmos_address.to_string(),
Expand All @@ -103,13 +105,13 @@ pub async fn send_to_eth(
}

pub async fn send_messages(
contact: &Contact,
contact: Contact,
cosmos_key: CosmosPrivateKey,
gas_price: (f64, String),
messages: Vec<Msg>,
gas_adjustment: f64,
) -> Result<TxResponse, GravityError> {
let cosmos_address = cosmos_key.to_address(&contact.get_prefix()).unwrap();
let cosmos_address = cosmos_key.to_address(&contact.get_prefix())?;

let fee_amount = Coin {
denom: gas_price.1.clone(),
Expand Down Expand Up @@ -149,19 +151,19 @@ pub async fn send_messages(
Ok(contact.wait_for_tx(response, TIMEOUT).await?)
}

pub async fn send_main_loop(
contact: &Contact,
pub async fn run_sender(
contact: Contact,
cosmos_key: CosmosPrivateKey,
gas_price: (f64, String),
mut rx: tokio::sync::mpsc::Receiver<Vec<Msg>>,
rx: Arc<Mutex<tokio::sync::mpsc::Receiver<Vec<Msg>>>>,
gas_adjustment: f64,
msg_batch_size: usize,
) {
while let Some(messages) = rx.recv().await {
while let Some(messages) = rx.lock().await.recv().await {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we creating a lock around the tokio receiver? The async channels should be able to be filled and drained without having to have synchronous locking.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It comes down to satisfying the borrow checker while passing in the receiver to run_sender inside of the loop inside of send_main_loop.

To get the compiler to let me pass the receiver into each loop iteration (each cosmos sender thread restart) I have to wrap the receiver in an atomic reference counter so that I can clone without making a duplicate of the receiver. However, Arc<T> doesn't have the DerefMut trait, and the recv() method requires that the value be a mutable reference. To get a mutable reference of the contained object in the Arc, I have to guarantee that one and only one mutable reference exists at a time, which is where the lock comes in.

for msg_chunk in messages.chunks(msg_batch_size) {
let batch = msg_chunk.to_vec();
match send_messages(
contact,
contact.clone(),
cosmos_key,
gas_price.to_owned(),
msg_chunk.to_vec(),
Expand All @@ -180,7 +182,7 @@ pub async fn send_main_loop(
for msg in batch {
let msg_vec = vec![msg];
match send_messages(
contact,
contact.clone(),
cosmos_key,
gas_price.to_owned(),
msg_vec.clone(),
Expand All @@ -198,6 +200,35 @@ pub async fn send_main_loop(
}
}

/// manages the cosmos message sender task. `run_sender` should never return, but in case there are system-caused
/// panics, we want to make sure we attempt to restart it. if this function ever returns, the receiver will be dropped
/// and all tasks containing senders will panic.
pub async fn send_main_loop(
contact: Contact,
cosmos_key: CosmosPrivateKey,
gas_price: (f64, String),
rx: tokio::sync::mpsc::Receiver<Vec<Msg>>,
gas_adjustment: f64,
msg_batch_size: usize,
) {
let rx = Arc::new(Mutex::new(rx));
loop {
info!("starting cosmos sender");
if let Err(err) = tokio::task::spawn(
run_sender(
contact.clone(),
cosmos_key,
gas_price.to_owned(),
rx.clone(),
gas_adjustment,
msg_batch_size,
)
).await {
error!("cosmos sender failed with: {err:?}");
}
}
}

fn log_send_error(messages: &Vec<Msg>, err: GravityError) {
let msg_types = messages
.iter()
Expand Down
2 changes: 1 addition & 1 deletion orchestrator/gorc/src/commands/cosmos_to_eth.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ impl Runnable for CosmosToEthCmd {
amount.clone(),
bridge_fee.clone(),
config.cosmos.gas_price.as_tuple(),
&contact,
contact,
1.0
)
.await;
Expand Down
6 changes: 5 additions & 1 deletion orchestrator/gorc/src/commands/orchestrator/start.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,11 @@ impl Runnable for StartCommand {
self.orchestrator_only,
config.cosmos.msg_batch_size,
)
.await;
.await
.unwrap_or_else(|e| {
error!("orchestrator exited with error: {}", e);
std::process::exit(1);
});
})
.unwrap_or_else(|e| {
status_err!("executor exited with error: {}", e);
Expand Down
2 changes: 1 addition & 1 deletion orchestrator/gorc/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ pub struct CosmosSection {
pub grpc: String,
pub prefix: String,
pub gas_adjustment: f64,
pub msg_batch_size: u32,
pub msg_batch_size: usize,
pub gas_price: GasPrice,
}

Expand Down
Loading