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.

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

feat: add redelegation info #96

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion chain/run.sh
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
. ../.env
export TENDERMINT_URL
export DATABASE_URL
cargo run -- --initial-query-retry-time 5
cargo run
3 changes: 0 additions & 3 deletions chain/src/config.rs
Original file line number Diff line number Diff line change
@@ -23,9 +23,6 @@ pub struct AppConfig {
#[clap(long, env)]
pub database_url: String,

#[clap(long, env)]
pub initial_query_retry_time: u64,

#[command(flatten)]
pub verbosity: Verbosity<InfoLevel>,
}
65 changes: 33 additions & 32 deletions chain/src/main.rs
Original file line number Diff line number Diff line change
@@ -1,15 +1,13 @@
use std::convert::identity;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Context;
use chain::app_state::AppState;
use chain::config::AppConfig;
use chain::repository;
use chain::services::db::get_pos_crawler_state;
use chain::services::namada::{
query_all_balances, query_all_bonds_and_unbonds, query_all_proposals,
query_bonds, query_last_block_height,
query_bonds, query_last_block_height, query_redelegations,
};
use chain::services::{
db as db_service, namada as namada_service,
@@ -29,7 +27,6 @@ use shared::crawler_state::ChainCrawlerState;
use shared::error::{AsDbError, AsRpcError, ContextDbInteractError, MainError};
use shared::id::Id;
use tendermint_rpc::HttpClient;
use tokio::time::sleep;
use tracing::Level;
use tracing_subscriber::FmtSubscriber;

@@ -74,7 +71,7 @@ async fn main() -> Result<(), MainError> {
.context_db_interact_error()
.into_db_error()?;

initial_query(&client, &conn, config.initial_query_retry_time).await?;
initial_query(&client, &conn).await?;

let crawler_state = db_service::get_chain_crawler_state(&conn)
.await
@@ -179,9 +176,16 @@ async fn crawling_fn(
tracing::info!("Creating {} governance votes...", proposals_votes.len());

let addresses = block.bond_addresses();
let bonds = query_bonds(&client, addresses).await.into_rpc_error()?;
let bonds = query_bonds(&client, addresses.clone())
.await
.into_rpc_error()?;
tracing::info!("Updating bonds for {} addresses", bonds.len());

let redelegations = query_redelegations(&client, addresses)
.await
.into_rpc_error()?;
tracing::info!("Updating redelegations for {} addresses", bonds.len());

let bonds_updates = bonds
.iter()
.cloned()
@@ -247,6 +251,11 @@ async fn crawling_fn(
repository::pos::insert_bonds(transaction_conn, bonds_updates)?;

repository::pos::insert_unbonds(transaction_conn, unbonds)?;
repository::pos::insert_redelegations(
transaction_conn,
redelegations,
)?;
repository::pos::clear_redelegations(transaction_conn, epoch)?;
repository::pos::remove_withdraws(
transaction_conn,
epoch,
@@ -286,42 +295,30 @@ async fn crawling_fn(
async fn initial_query(
client: &HttpClient,
conn: &Object,
initial_query_retry_time: u64,
) -> Result<(), MainError> {
tracing::info!("Querying initial data...");
let block_height =
query_last_block_height(client).await.into_rpc_error()?;
let mut epoch =
namada_service::get_epoch_at_block_height(client, block_height)
.await
.into_rpc_error()?;
let epoch = namada_service::get_epoch_at_block_height(client, block_height)
.await
.into_rpc_error()?;
let first_block_in_epoch = namada_service::get_first_block_in_epoch(client)
.await
.into_rpc_error()?;

loop {
let pos_crawler_state =
get_pos_crawler_state(conn).await.into_db_error();

match pos_crawler_state {
// >= in case epochs are really short
Ok(pos_crawler_state)
if pos_crawler_state.last_processed_epoch >= epoch =>
{
// We assign pos crawler epoch as epoch to process
epoch = pos_crawler_state.last_processed_epoch;
break;
}
_ => {}
}

tracing::info!("Waiting for PoS service update...");

sleep(Duration::from_secs(initial_query_retry_time)).await;
}

let balances = query_all_balances(client).await.into_rpc_error()?;

let validators_set =
namada_service::get_validator_addresses_at_epoch(client, epoch)
.await
.into_rpc_error()?;

tracing::info!("Querying redelegations...");
let redelegations =
namada_service::query_all_redelegations(client, validators_set)
.await
.into_rpc_error()?;

tracing::info!("Querying bonds and unbonds...");
let (bonds, unbonds) = query_all_bonds_and_unbonds(client, None, None)
.await
@@ -373,6 +370,10 @@ async fn initial_query(

repository::pos::insert_bonds(transaction_conn, bonds)?;
repository::pos::insert_unbonds(transaction_conn, unbonds)?;
repository::pos::insert_redelegations(
transaction_conn,
redelegations,
)?;

repository::crawler_state::upsert_crawler_state(
transaction_conn,
Loading
Loading