blockworks-foundation
/
BankingStageErrorsTrackingSidecar
mirror of https://github.com/blockworks-foundation/BankingStageErrorsTrackingSidecar.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

cargo fmt and minor changes

This commit is contained in:
godmodegalactus 2024-01-26 11:46:39 +01:00
parent 2b16651f2b
commit edcdfa85d9
No known key found for this signature in database
GPG Key ID: 22DA4A30887FDA3C
3 changed files with 111 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/alt_store.rs
View File

@ -1,3 +1,4 @@
use crate::block_info::TransactionAccount;
use dashmap::DashMap; use dashmap::DashMap;
use itertools::Itertools; use itertools::Itertools;
use prometheus::{opts, register_int_gauge, IntGauge}; use prometheus::{opts, register_int_gauge, IntGauge};
@ -6,7 +7,6 @@ use solana_address_lookup_table_program::state::AddressLookupTable;
use solana_rpc_client::nonblocking::rpc_client::RpcClient; use solana_rpc_client::nonblocking::rpc_client::RpcClient;
use solana_sdk::{account::ReadableAccount, commitment_config::CommitmentConfig, pubkey::Pubkey}; use solana_sdk::{account::ReadableAccount, commitment_config::CommitmentConfig, pubkey::Pubkey};
use std::{collections::HashSet, sync::Arc, time::Duration}; use std::{collections::HashSet, sync::Arc, time::Duration};
use crate::block_info::TransactionAccount;
use tokio::sync::RwLock; use tokio::sync::RwLock;
lazy_static::lazy_static! { lazy_static::lazy_static! {
@ -34,10 +34,10 @@ impl ALTStore {
let alts_list = { let alts_list = {
let lk = self.under_loading.read().await; let lk = self.under_loading.read().await;
alts_list alts_list
.iter() .iter()
.filter(|x| !self.map.contains_key(x) && !lk.contains(x)) .filter(|x| !self.map.contains_key(x) && !lk.contains(x))
.cloned() .cloned()
.collect_vec() .collect_vec()
}; };
if alts_list.is_empty() { if alts_list.is_empty() {
@ -54,12 +54,14 @@ impl ALTStore {
let rpc_client = self.rpc_client.clone(); let rpc_client = self.rpc_client.clone();
let this = self.clone(); let this = self.clone();
tokio::spawn(async move { tokio::spawn(async move {
if let Ok(Ok(multiple_accounts)) = tokio::time::timeout( Duration::from_secs(30), rpc_client if let Ok(Ok(multiple_accounts)) = tokio::time::timeout(
.get_multiple_accounts_with_commitment( Duration::from_secs(30),
rpc_client.get_multiple_accounts_with_commitment(
&batch, &batch,
CommitmentConfig::processed(), CommitmentConfig::processed(),
)) ),
.await )
.await
{ {
for (index, acc) in multiple_accounts.value.iter().enumerate() { for (index, acc) in multiple_accounts.value.iter().enumerate() {
if let Some(acc) = acc { if let Some(acc) = acc {
@ -79,7 +81,6 @@ impl ALTStore {
self.finished_loading(&alts_list).await; self.finished_loading(&alts_list).await;
ALTS_IN_STORE.set(alts_list.len() as i64); ALTS_IN_STORE.set(alts_list.len() as i64);
} }
pub fn save_account(&self, address: &Pubkey, data: &[u8]) { pub fn save_account(&self, address: &Pubkey, data: &[u8]) {
@ -164,9 +165,8 @@ impl ALTStore {
write_accounts: &Vec<u8>, write_accounts: &Vec<u8>,
read_account: &Vec<u8>, read_account: &Vec<u8>,
) -> Vec<TransactionAccount> { ) -> Vec<TransactionAccount> {
let mut times = 0; let mut times = 0;
const MAX_TIMES_RETRY : usize = 100; const MAX_TIMES_RETRY: usize = 100;
while self.is_loading_contains(alt).await { while self.is_loading_contains(alt).await {
if times > MAX_TIMES_RETRY { if times > MAX_TIMES_RETRY {
break; break;
@ -203,14 +203,14 @@ impl ALTStore {
} }
} }
async fn is_loading(&self, alts_list: &Vec<Pubkey>) { async fn is_loading(&self, alts_list: &Vec<Pubkey>) {
let mut write = self.under_loading.write().await; let mut write = self.under_loading.write().await;
for alt in alts_list { for alt in alts_list {
write.insert(alt.clone()); write.insert(alt.clone());
} }
} }
async fn finished_loading(&self, alts_list: &Vec<Pubkey>) { async fn finished_loading(&self, alts_list: &Vec<Pubkey>) {
let mut write = self.under_loading.write().await; let mut write = self.under_loading.write().await;
for alt in alts_list { for alt in alts_list {
write.remove(alt); write.remove(alt);
@ -225,14 +225,15 @@ impl ALTStore {
#[derive(Serialize, Deserialize)] #[derive(Serialize, Deserialize)]
pub struct BinaryALTData { pub struct BinaryALTData {
pub data: Vec<(Pubkey, Vec<Pubkey>)> pub data: Vec<(Pubkey, Vec<Pubkey>)>,
} }
impl BinaryALTData { impl BinaryALTData {
pub fn new(map: &Arc<DashMap<Pubkey, Vec<Pubkey>>>) -> Self { pub fn new(map: &Arc<DashMap<Pubkey, Vec<Pubkey>>>) -> Self {
let data = map.iter().map(|x| (x.key().clone(), x.value().clone())).collect_vec(); let data = map
Self { .iter()
data .map(|x| (x.key().clone(), x.value().clone()))
} .collect_vec();
Self { data }
} }
} }

166
src/block_info.rs
View File

@ -118,8 +118,6 @@ pub struct BlockInfo {
pub transactions: Vec<BlockTransactionInfo>, pub transactions: Vec<BlockTransactionInfo>,
} }
impl BlockInfo { impl BlockInfo {
pub async fn process_versioned_message( pub async fn process_versioned_message(
atl_store: &Arc<ALTStore>, atl_store: &Arc<ALTStore>,
@ -381,97 +379,103 @@ impl BlockInfo {
let mut total_cu_requested: u64 = 0; let mut total_cu_requested: u64 = 0;
let mut prio_fees_in_block = vec![]; let mut prio_fees_in_block = vec![];
let mut lookup_tables = HashSet::new(); let mut lookup_tables = HashSet::new();
let sigs_and_messages = block.transactions.iter().filter_map( |transaction| { let sigs_and_messages = block
let Some(tx) = &transaction.transaction else { .transactions
return None; .iter()
}; .filter_map(|transaction| {
let Some(tx) = &transaction.transaction else {
return None;
};
let Some(message) = &tx.message else { let Some(message) = &tx.message else {
return None; return None;
}; };
let Some(header) = &message.header else { let Some(header) = &message.header else {
return None; return None;
}; };
let Some(meta) = &transaction.meta else { let Some(meta) = &transaction.meta else {
return None; return None;
}; };
let signature = Signature::try_from(&tx.signatures[0][0..64]) let signature = Signature::try_from(&tx.signatures[0][0..64])
.unwrap() .unwrap()
.to_string(); .to_string();
let message = VersionedMessage::V0(v0::Message { let message = VersionedMessage::V0(v0::Message {
header: MessageHeader { header: MessageHeader {
num_required_signatures: header.num_required_signatures as u8, num_required_signatures: header.num_required_signatures as u8,
num_readonly_signed_accounts: header.num_readonly_signed_accounts as u8, num_readonly_signed_accounts: header.num_readonly_signed_accounts as u8,
num_readonly_unsigned_accounts: header.num_readonly_unsigned_accounts as u8, num_readonly_unsigned_accounts: header.num_readonly_unsigned_accounts as u8,
}, },
account_keys: message account_keys: message
.account_keys .account_keys
.clone() .clone()
.into_iter() .into_iter()
.map(|key| { .map(|key| {
let bytes: [u8; 32] = let bytes: [u8; 32] =
key.try_into().unwrap_or(Pubkey::default().to_bytes()); key.try_into().unwrap_or(Pubkey::default().to_bytes());
Pubkey::new_from_array(bytes) Pubkey::new_from_array(bytes)
}) })
.collect(), .collect(),
recent_blockhash: solana_sdk::hash::Hash::new(&message.recent_blockhash), recent_blockhash: solana_sdk::hash::Hash::new(&message.recent_blockhash),
instructions: message instructions: message
.instructions .instructions
.clone() .clone()
.into_iter() .into_iter()
.map(|ix| CompiledInstruction { .map(|ix| CompiledInstruction {
program_id_index: ix.program_id_index as u8, program_id_index: ix.program_id_index as u8,
accounts: ix.accounts, accounts: ix.accounts,
data: ix.data, data: ix.data,
}) })
.collect(), .collect(),
address_table_lookups: message address_table_lookups: message
.address_table_lookups .address_table_lookups
.clone() .clone()
.into_iter() .into_iter()
.map(|table| { .map(|table| {
let bytes: [u8; 32] = table let bytes: [u8; 32] = table
.account_key .account_key
.try_into() .try_into()
.unwrap_or(Pubkey::default().to_bytes()); .unwrap_or(Pubkey::default().to_bytes());
let account_key = Pubkey::new_from_array(bytes); let account_key = Pubkey::new_from_array(bytes);
lookup_tables.insert(account_key.clone()); lookup_tables.insert(account_key.clone());
MessageAddressTableLookup { MessageAddressTableLookup {
account_key, account_key,
writable_indexes: table.writable_indexes, writable_indexes: table.writable_indexes,
readonly_indexes: table.readonly_indexes, readonly_indexes: table.readonly_indexes,
} }
}) })
.collect(), .collect(),
}); });
Some((signature, message, meta, transaction.is_vote)) Some((signature, message, meta, transaction.is_vote))
}).collect_vec(); })
.collect_vec();
atl_store.load_all_alts(lookup_tables.iter().cloned().collect_vec()).await; atl_store
.load_all_alts(lookup_tables.iter().cloned().collect_vec())
.await;
let mut block_transactions = vec![]; let mut block_transactions = vec![];
for (signature, message, meta, is_vote) in sigs_and_messages { for (signature, message, meta, is_vote) in sigs_and_messages {
let tx = Self::process_versioned_message( let tx = Self::process_versioned_message(
&atl_store, &atl_store,
&signature, &signature,
slot, slot,
&message, &message,
&mut prio_fees_in_block, &mut prio_fees_in_block,
&mut writelocked_accounts, &mut writelocked_accounts,
&mut readlocked_accounts, &mut readlocked_accounts,
meta.compute_units_consumed.unwrap_or(0), meta.compute_units_consumed.unwrap_or(0),
&mut total_cu_requested, &mut total_cu_requested,
is_vote, is_vote,
meta.err.is_none(), meta.err.is_none(),
) )
.await; .await;
if let Some(tx) = tx { if let Some(tx) = tx {
block_transactions.push(tx); block_transactions.push(tx);
} }
}; }
let heavily_locked_accounts = let heavily_locked_accounts =
Self::calculate_account_usage(&writelocked_accounts, &readlocked_accounts); Self::calculate_account_usage(&writelocked_accounts, &readlocked_accounts);

8
src/main.rs
View File

@ -13,7 +13,10 @@ use std::{
}, },
time::Duration, time::Duration,
}; };
use tokio::{io::{AsyncReadExt, AsyncWriteExt}, time::Instant}; use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
time::Instant,
};
use crate::prometheus_sync::PrometheusSync; use crate::prometheus_sync::PrometheusSync;
use block_info::BlockInfo; use block_info::BlockInfo;
@ -207,7 +210,7 @@ async fn start_tracking_blocks(
// .collect_vec(); // .collect_vec();
// ALT store from binary // ALT store from binary
// let atl_store = { // let atl_store = {
// let alt_store = Arc::new(alt_store::ALTStore::new(rpc_client)); // let alt_store = Arc::new(alt_store::ALTStore::new(rpc_client));
// let mut alts_file = tokio::fs::File::open(alts_list).await.unwrap(); // let mut alts_file = tokio::fs::File::open(alts_list).await.unwrap();
// let mut buf = vec![]; // let mut buf = vec![];
@ -325,7 +328,6 @@ async fn start_tracking_blocks(
} }
} }
#[tokio::main(worker_threads=1)]
async fn main() -> anyhow::Result<()> { async fn main() -> anyhow::Result<()> {
tracing_subscriber::fmt::init(); tracing_subscriber::fmt::init();