Skip to content

Commit

Permalink
chore(hubble): raceclient - provide winning client in result (#3156)
Browse files Browse the repository at this point in the history
  • Loading branch information
qlp authored Nov 4, 2024
2 parents 18c86b4 + 3ca6735 commit a9767d8
Show file tree
Hide file tree
Showing 6 changed files with 188 additions and 410 deletions.
137 changes: 40 additions & 97 deletions hubble/src/indexer/aptos/provider.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,10 @@ use aptos_rest_client::{
};
use url::Url;

use crate::{indexer::api::BlockHeight, race_client::RaceClient};
use crate::{
indexer::api::BlockHeight,
race_client::{RaceClient, RaceClientId, RaceClientResponse},
};

#[derive(Clone, Debug)]
pub struct Provider {
Expand All @@ -16,7 +19,13 @@ pub struct Provider {

#[derive(Clone, Debug, Copy)]
pub struct RpcProviderId {
index: usize,
race_client_id: RaceClientId,
}

impl From<RpcProviderId> for RaceClientId {
fn from(value: RpcProviderId) -> Self {
value.race_client_id
}
}

#[derive(Debug)]
Expand All @@ -26,16 +35,20 @@ pub struct RpcResult<T> {
}

impl<T> RpcResult<T> {
fn new(provider_index: usize, result: T) -> Self {
fn new(race_client_id: RaceClientId, result: T) -> Self {
Self {
provider_id: RpcProviderId {
index: provider_index,
},
provider_id: RpcProviderId { race_client_id },
response: result,
}
}
}

impl<T> From<RaceClientResponse<T>> for RpcResult<T> {
fn from(value: RaceClientResponse<T>) -> Self {
RpcResult::new(value.race_client_id, value.response)
}
}

impl Provider {
pub fn new(rpc_urls: Vec<Url>) -> Self {
Self {
Expand All @@ -48,55 +61,28 @@ impl Provider {
}
}

fn rpc_client(&self, provider_id: Option<RpcProviderId>) -> RaceClient<Client> {
Self::select_client(self.rpc_client.clone(), provider_id.map(|id| id.index))
}

fn select_client<T: Clone>(
client: RaceClient<T>,
provider_index: Option<usize>,
) -> RaceClient<T> {
match provider_index {
Some(provider_index) => RaceClient::new(vec![client.clients[provider_index].clone()]),
None => client,
}
}

// RPC
pub async fn get_index(
&self,
provider_id: Option<RpcProviderId>,
) -> Result<RpcResult<Response<IndexResponse>>, RestError> {
let result = self.rpc_client(provider_id).get_index().await?;

// TODO: improve race client to return index with result
Ok(RpcResult::new(
provider_id.map_or_else(
|| self.rpc_client.fastest_index(),
|provider_id| provider_id.index,
),
result,
))
self.rpc_client
.race(provider_id.map(Into::into), |c| c.get_index())
.await
.map(Into::into)
}

pub async fn get_block_by_height(
&self,
height: BlockHeight,
provider_id: Option<RpcProviderId>,
) -> Result<RpcResult<Response<Block>>, RestError> {
let result = self
.rpc_client(provider_id)
.get_block_by_height(height)
.await?;

// TODO: improve race client to return index with result
Ok(RpcResult::new(
provider_id.map_or_else(
|| self.rpc_client.fastest_index(),
|provider_id| provider_id.index,
),
result,
))
self.rpc_client
.race(provider_id.map(Into::into), |c| {
c.get_block_by_height(height, false)
})
.await
.map(Into::into)
}

pub async fn get_transactions(
Expand All @@ -105,67 +91,24 @@ impl Provider {
limit: u16,
provider_id: Option<RpcProviderId>,
) -> Result<RpcResult<Response<Vec<Transaction>>>, RestError> {
let result = self
.rpc_client(provider_id)
.get_transactions(start, limit)
.await?;

// TODO: improve race client to return index with result
Ok(RpcResult::new(
provider_id.map_or_else(
|| self.rpc_client.fastest_index(),
|provider_id| provider_id.index,
),
result,
))
self.rpc_client
.race(provider_id.map(Into::into), |c| {
c.get_transactions(Some(start), Some(limit))
})
.await
.map(Into::into)
}

pub async fn get_transaction_by_version(
&self,
version: u64,
provider_id: Option<RpcProviderId>,
) -> Result<RpcResult<Response<Transaction>>, RestError> {
let result = self
.rpc_client(provider_id)
.get_transaction_by_version(version)
.await?;

// TODO: improve race client to return index with result
Ok(RpcResult::new(
provider_id.map_or_else(
|| self.rpc_client.fastest_index(),
|provider_id| provider_id.index,
),
result,
))
}
}

impl RaceClient<Client> {
pub async fn get_index(&self) -> Result<Response<IndexResponse>, RestError> {
self.race(|c| c.get_index()).await
}

pub async fn get_block_by_height(
&self,
height: BlockHeight,
) -> Result<Response<Block>, RestError> {
self.race(|c| c.get_block_by_height(height, false)).await
}

pub async fn get_transactions(
&self,
start: BlockHeight,
limit: u16,
) -> Result<Response<Vec<Transaction>>, RestError> {
self.race(|c| c.get_transactions(Some(start), Some(limit)))
self.rpc_client
.race(provider_id.map(Into::into), |c| {
c.get_transaction_by_version(version)
})
.await
}

pub async fn get_transaction_by_version(
&self,
version: u64,
) -> Result<Response<Transaction>, RestError> {
self.race(|c| c.get_transaction_by_version(version)).await
.map(Into::into)
}
}
4 changes: 2 additions & 2 deletions hubble/src/indexer/eth/create_client_tracker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -52,8 +52,8 @@ pub fn schedule_create_client_checker(
let tx = provider
.get_transaction_by_hash(FixedBytes::from_str(&transaction_hash).expect("valid transaction hash"), None)
.await?
.response
.expect("transaction");
.expect("transaction")
.response;

let msg = match <IbcHandler::CreateClientCall as alloy::sol_types::SolCall>::abi_decode(&tx.input,true) {
Ok(msg) => msg,
Expand Down
11 changes: 6 additions & 5 deletions hubble/src/indexer/eth/fetcher_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -109,23 +109,24 @@ impl EthFetcherClient {
.await;

match block {
Ok(rpc_result) => match rpc_result.response {
Some(block) => {
Ok(rpc_result) => match rpc_result {
Some(result) => {
let block = result.response;
debug!(
"{}: fetched (provider index: {:?})",
selection, rpc_result.provider_id
selection, result.provider_id
);

Ok(EthBlockHandle {
reference: block.block_reference()?,
details: match mode {
FetchMode::Lazy => BlockDetails::Lazy(block),
FetchMode::Eager => BlockDetails::Eager(
self.fetch_details(&block, rpc_result.provider_id).await?,
self.fetch_details(&block, result.provider_id).await?,
),
},
eth_client: self.clone(),
provider_id: rpc_result.provider_id,
provider_id: result.provider_id,
})
}
None => {
Expand Down
Loading

0 comments on commit a9767d8

Please sign in to comment.