472 lines
20 KiB
Rust
472 lines
20 KiB
Rust
use std::{
|
|
sync::{
|
|
atomic::{AtomicBool, Ordering},
|
|
Arc,
|
|
},
|
|
time::Duration,
|
|
};
|
|
|
|
use chrono::Utc;
|
|
use dashmap::DashMap;
|
|
use log::{debug, warn};
|
|
use solana_client::{nonblocking::rpc_client::RpcClient, rpc_config::RpcBlockConfig};
|
|
use solana_lite_rpc_core::notifications::NotificationMsg;
|
|
use solana_sdk::{
|
|
commitment_config::{CommitmentConfig, CommitmentLevel},
|
|
signature::Signature,
|
|
slot_history::Slot,
|
|
};
|
|
use solana_transaction_status::{
|
|
option_serializer::OptionSerializer, RewardType, TransactionDetails, UiConfirmedBlock,
|
|
UiTransactionEncoding,
|
|
};
|
|
|
|
use crate::states::{BlockData, TransactionConfirmRecord, TransactionSendRecord};
|
|
|
|
use tokio::{
|
|
sync::broadcast::Sender, sync::mpsc::UnboundedReceiver, task::JoinHandle, time::Instant,
|
|
};
|
|
|
|
pub async fn process_blocks(
|
|
block: &UiConfirmedBlock,
|
|
tx_confirm_records: Sender<TransactionConfirmRecord>,
|
|
tx_block_data: Sender<BlockData>,
|
|
transaction_map: Arc<DashMap<Signature, (TransactionSendRecord, Instant)>>,
|
|
slot: u64,
|
|
commitment: CommitmentLevel,
|
|
) {
|
|
let mut mm_transaction_count: u64 = 0;
|
|
let rewards = block.rewards.as_ref().unwrap();
|
|
let slot_leader = match rewards
|
|
.iter()
|
|
.find(|r| r.reward_type == Some(RewardType::Fee))
|
|
{
|
|
Some(x) => x.pubkey.clone(),
|
|
None => "".to_string(),
|
|
};
|
|
|
|
if let Some(transactions) = &block.transactions {
|
|
let nb_transactions = transactions.len();
|
|
let mut mm_cu_consumed: u64 = 0;
|
|
let mut total_cu_consumed: u64 = 0;
|
|
for solana_transaction_status::EncodedTransactionWithStatusMeta {
|
|
transaction, meta, ..
|
|
} in transactions
|
|
{
|
|
let tx_cu_consumed =
|
|
meta.as_ref()
|
|
.map_or(0, |meta| match meta.compute_units_consumed {
|
|
OptionSerializer::Some(cu_consumed) => cu_consumed,
|
|
_ => 0,
|
|
});
|
|
let transaction = match transaction.decode() {
|
|
Some(tx) => tx,
|
|
None => {
|
|
continue;
|
|
}
|
|
};
|
|
for signature in &transaction.signatures {
|
|
// add CU in counter
|
|
total_cu_consumed = total_cu_consumed.saturating_add(tx_cu_consumed);
|
|
if let Some((_, (transaction_record, _))) = transaction_map.remove(signature) {
|
|
mm_transaction_count += 1;
|
|
mm_cu_consumed = mm_cu_consumed.saturating_add(tx_cu_consumed);
|
|
|
|
match tx_confirm_records.send(TransactionConfirmRecord {
|
|
signature: transaction_record.signature.to_string(),
|
|
confirmed_slot: Some(slot),
|
|
confirmed_at: Some(Utc::now().to_string()),
|
|
sent_at: transaction_record.sent_at.to_string(),
|
|
sent_slot: transaction_record.sent_slot,
|
|
successful: if let Some(meta) = &meta {
|
|
meta.status.is_ok()
|
|
} else {
|
|
false
|
|
},
|
|
error: if let Some(meta) = &meta {
|
|
meta.err.as_ref().map(|x| x.to_string())
|
|
} else {
|
|
None
|
|
},
|
|
block_hash: Some(block.blockhash.clone()),
|
|
market: transaction_record.market.map(|x| x.to_string()),
|
|
market_maker: transaction_record.market_maker.map(|x| x.to_string()),
|
|
keeper_instruction: transaction_record.keeper_instruction,
|
|
slot_processed: Some(slot),
|
|
slot_leader: Some(slot_leader.clone()),
|
|
timed_out: false,
|
|
priority_fees: transaction_record.priority_fees,
|
|
}) {
|
|
Ok(_) => {}
|
|
Err(e) => {
|
|
warn!("Tx confirm record channel broken {}", e.to_string());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// push block data
|
|
{
|
|
let _ = tx_block_data.send(BlockData {
|
|
block_hash: block.blockhash.clone(),
|
|
block_leader: slot_leader,
|
|
block_slot: slot,
|
|
block_time: if let Some(time) = block.block_time {
|
|
time as u64
|
|
} else {
|
|
0
|
|
},
|
|
number_of_mango_simulation_txs: mm_transaction_count,
|
|
total_transactions: nb_transactions as u64,
|
|
cu_consumed: total_cu_consumed,
|
|
cu_consumed_by_mango_simulations: mm_cu_consumed,
|
|
commitment,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn get_blocks_with_retry(
|
|
client: Arc<RpcClient>,
|
|
start_block: u64,
|
|
commitment_confirmation: CommitmentConfig,
|
|
) -> Result<Vec<Slot>, ()> {
|
|
const N_TRY_REQUEST_BLOCKS: u64 = 4;
|
|
for _ in 0..N_TRY_REQUEST_BLOCKS {
|
|
let block_slots = client
|
|
.get_blocks_with_commitment(start_block, None, commitment_confirmation)
|
|
.await;
|
|
|
|
match block_slots {
|
|
Ok(slots) => {
|
|
return Ok(slots);
|
|
}
|
|
Err(error) => {
|
|
warn!("Failed to download blocks: {}, retry", error);
|
|
}
|
|
}
|
|
}
|
|
Err(())
|
|
}
|
|
|
|
pub fn confirmation_by_lite_rpc_notification_stream(
|
|
tx_record_rx: UnboundedReceiver<TransactionSendRecord>,
|
|
notification_stream: UnboundedReceiver<NotificationMsg>,
|
|
tx_confirm_records: tokio::sync::broadcast::Sender<TransactionConfirmRecord>,
|
|
tx_block_data: tokio::sync::broadcast::Sender<BlockData>,
|
|
exit_signal: Arc<AtomicBool>,
|
|
) -> Vec<JoinHandle<()>> {
|
|
let transaction_map: Arc<DashMap<String, (TransactionSendRecord, Instant)>> =
|
|
Arc::new(DashMap::new());
|
|
|
|
let confirming_task = {
|
|
let transaction_map = transaction_map.clone();
|
|
let tx_confirm_records = tx_confirm_records.clone();
|
|
let exit_signal = exit_signal.clone();
|
|
tokio::spawn(async move {
|
|
let mut tx_record_rx = tx_record_rx;
|
|
let mut notification_stream = notification_stream;
|
|
|
|
while !transaction_map.is_empty() || !exit_signal.load(Ordering::Relaxed) {
|
|
tokio::select! {
|
|
transaction_record = tx_record_rx.recv() => {
|
|
if let Some(transaction_record) = transaction_record{
|
|
transaction_map
|
|
.insert(transaction_record.signature.to_string(), (transaction_record, Instant::now()));
|
|
}
|
|
|
|
},
|
|
notification = notification_stream.recv() => {
|
|
if let Some(notification) = notification {
|
|
|
|
match notification {
|
|
NotificationMsg::BlockNotificationMsg(block_notification) => {
|
|
if block_notification.commitment != CommitmentLevel::Finalized {
|
|
continue;
|
|
}
|
|
let _ = tx_block_data.send(BlockData {
|
|
block_hash: block_notification.blockhash.to_string(),
|
|
block_leader: block_notification.block_leader,
|
|
block_slot: block_notification.slot,
|
|
block_time: block_notification.block_time,
|
|
number_of_mango_simulation_txs: block_notification.transaction_found,
|
|
total_transactions: block_notification.total_transactions,
|
|
cu_consumed: block_notification.total_cu_consumed,
|
|
cu_consumed_by_mango_simulations: block_notification.cu_consumed_by_txs,
|
|
commitment: block_notification.commitment,
|
|
});
|
|
}
|
|
NotificationMsg::UpdateTransactionMsg(tx_update_notifications) => {
|
|
|
|
for tx_notification in tx_update_notifications {
|
|
if tx_notification.commitment != CommitmentLevel::Finalized {
|
|
continue;
|
|
}
|
|
|
|
if let Some(value) = transaction_map.get(&tx_notification.signature) {
|
|
let (tx_sent_record, _) = value.clone();
|
|
let error = match &tx_notification.transaction_status {
|
|
Err(e) => {
|
|
Some(e.to_string())
|
|
},
|
|
_ => None
|
|
};
|
|
let _ = tx_confirm_records.send(TransactionConfirmRecord {
|
|
signature: tx_notification.signature.clone(),
|
|
confirmed_slot: Some(tx_notification.slot),
|
|
confirmed_at: Some(Utc::now().to_string()),
|
|
sent_at: tx_sent_record.sent_at.to_string(),
|
|
sent_slot: tx_sent_record.sent_slot,
|
|
successful: tx_notification.transaction_status.is_ok(),
|
|
error,
|
|
block_hash: Some(tx_notification.blockhash),
|
|
market: tx_sent_record.market.map(|x| x.to_string()),
|
|
market_maker: tx_sent_record.market_maker.map(|x| x.to_string()),
|
|
keeper_instruction: tx_sent_record.keeper_instruction.clone(),
|
|
slot_processed: Some(tx_notification.slot),
|
|
slot_leader: Some(tx_notification.leader.to_string()),
|
|
timed_out: false,
|
|
priority_fees: tx_sent_record.priority_fees,
|
|
});
|
|
}
|
|
|
|
transaction_map.remove(&tx_notification.signature);
|
|
}
|
|
},
|
|
_ => {
|
|
// others do nothing
|
|
}
|
|
}
|
|
}
|
|
},
|
|
_ = tokio::time::sleep(Duration::from_secs(1)) => {
|
|
// timeout
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
log::info!("stopped processing the transactions");
|
|
})
|
|
};
|
|
|
|
let cleaner_jh = {
|
|
let transaction_map = transaction_map;
|
|
let exit_signal = exit_signal;
|
|
let tx_confirm_records = tx_confirm_records;
|
|
tokio::spawn(async move {
|
|
loop {
|
|
tokio::time::sleep(Duration::from_secs(60)).await;
|
|
{
|
|
let mut to_remove = vec![];
|
|
|
|
for tx_data in transaction_map.iter() {
|
|
let sent_record = &tx_data.0;
|
|
let instant = tx_data.1;
|
|
let signature = tx_data.key();
|
|
let remove = instant.elapsed() > Duration::from_secs(120);
|
|
|
|
// add to timeout if not retaining
|
|
if remove {
|
|
let _ = tx_confirm_records.send(TransactionConfirmRecord {
|
|
signature: signature.to_string(),
|
|
confirmed_slot: None,
|
|
confirmed_at: None,
|
|
sent_at: sent_record.sent_at.to_string(),
|
|
sent_slot: sent_record.sent_slot,
|
|
successful: false,
|
|
error: Some("timeout".to_string()),
|
|
block_hash: None,
|
|
market: sent_record.market.map(|x| x.to_string()),
|
|
market_maker: sent_record.market_maker.map(|x| x.to_string()),
|
|
keeper_instruction: sent_record.keeper_instruction.clone(),
|
|
slot_processed: None,
|
|
slot_leader: None,
|
|
timed_out: true,
|
|
priority_fees: sent_record.priority_fees,
|
|
});
|
|
to_remove.push(signature.clone());
|
|
}
|
|
}
|
|
|
|
for signature in to_remove {
|
|
transaction_map.remove(&signature);
|
|
}
|
|
|
|
// if exit and all the transactions are processed
|
|
if exit_signal.load(Ordering::Relaxed) && transaction_map.is_empty() {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
})
|
|
};
|
|
vec![confirming_task, cleaner_jh]
|
|
}
|
|
|
|
#[deprecated]
|
|
pub fn confirmations_by_blocks(
|
|
client: Arc<RpcClient>,
|
|
mut tx_record_rx: UnboundedReceiver<TransactionSendRecord>,
|
|
tx_confirm_records: tokio::sync::broadcast::Sender<TransactionConfirmRecord>,
|
|
tx_block_data: tokio::sync::broadcast::Sender<BlockData>,
|
|
from_slot: u64,
|
|
exit_signal: Arc<AtomicBool>,
|
|
) -> Vec<JoinHandle<()>> {
|
|
let transaction_map = Arc::new(DashMap::new());
|
|
|
|
let map_filler_jh = {
|
|
let transaction_map = transaction_map.clone();
|
|
let exit_signal = exit_signal.clone();
|
|
tokio::spawn(async move {
|
|
loop {
|
|
match tokio::time::timeout(tokio::time::Duration::from_secs(1), tx_record_rx.recv())
|
|
.await
|
|
{
|
|
Ok(tx_record) => match tx_record {
|
|
Some(tx_record) => {
|
|
debug!(
|
|
"add to queue len={} sig={}",
|
|
transaction_map.len() + 1,
|
|
tx_record.signature
|
|
);
|
|
transaction_map
|
|
.insert(tx_record.signature, (tx_record, Instant::now()));
|
|
}
|
|
None => {
|
|
exit_signal.store(true, Ordering::Relaxed);
|
|
break;
|
|
}
|
|
},
|
|
Err(_) => {
|
|
// on timeout just check if services are being stopped
|
|
if exit_signal.load(Ordering::Relaxed) {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
})
|
|
};
|
|
|
|
let cleaner_jh = {
|
|
let transaction_map = transaction_map.clone();
|
|
let exit_signal = exit_signal.clone();
|
|
let tx_confirm_records = tx_confirm_records.clone();
|
|
tokio::spawn(async move {
|
|
loop {
|
|
tokio::time::sleep(Duration::from_secs(60)).await;
|
|
{
|
|
let mut to_remove = vec![];
|
|
|
|
for tx_data in transaction_map.iter() {
|
|
let sent_record = &tx_data.0;
|
|
let instant = tx_data.1;
|
|
let signature = tx_data.key();
|
|
let remove = instant.elapsed() > Duration::from_secs(120);
|
|
|
|
// add to timeout if not retaining
|
|
if remove {
|
|
let _ = tx_confirm_records.send(TransactionConfirmRecord {
|
|
signature: signature.to_string(),
|
|
confirmed_slot: None,
|
|
confirmed_at: None,
|
|
sent_at: sent_record.sent_at.to_string(),
|
|
sent_slot: sent_record.sent_slot,
|
|
successful: false,
|
|
error: Some("timeout".to_string()),
|
|
block_hash: None,
|
|
market: sent_record.market.map(|x| x.to_string()),
|
|
market_maker: sent_record.market_maker.map(|x| x.to_string()),
|
|
keeper_instruction: sent_record.keeper_instruction.clone(),
|
|
slot_processed: None,
|
|
slot_leader: None,
|
|
timed_out: true,
|
|
priority_fees: sent_record.priority_fees,
|
|
});
|
|
to_remove.push(*signature);
|
|
}
|
|
}
|
|
|
|
for signature in to_remove {
|
|
transaction_map.remove(&signature);
|
|
}
|
|
|
|
// if exit and all the transactions are processed
|
|
if exit_signal.load(Ordering::Relaxed) && transaction_map.len() == 0 {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
})
|
|
};
|
|
|
|
let block_confirmation_jh = {
|
|
let exit_signal = exit_signal;
|
|
tokio::spawn(async move {
|
|
let mut start_block = from_slot;
|
|
let mut start_instant = tokio::time::Instant::now();
|
|
let refresh_in = Duration::from_secs(10);
|
|
let commitment_confirmation = CommitmentConfig {
|
|
commitment: CommitmentLevel::Confirmed,
|
|
};
|
|
loop {
|
|
if exit_signal.load(Ordering::Relaxed) && transaction_map.len() == 0 {
|
|
break;
|
|
}
|
|
|
|
let wait_duration = tokio::time::Instant::now() - start_instant;
|
|
if wait_duration < refresh_in {
|
|
tokio::time::sleep(refresh_in - wait_duration).await;
|
|
}
|
|
start_instant = tokio::time::Instant::now();
|
|
|
|
let block_slots =
|
|
get_blocks_with_retry(client.clone(), start_block, commitment_confirmation)
|
|
.await;
|
|
if block_slots.is_err() {
|
|
break;
|
|
}
|
|
|
|
let block_slots = block_slots.unwrap();
|
|
if block_slots.is_empty() {
|
|
continue;
|
|
}
|
|
start_block = *block_slots.last().unwrap() + 1;
|
|
|
|
let blocks = block_slots.iter().map(|slot| {
|
|
client.get_block_with_config(
|
|
*slot,
|
|
RpcBlockConfig {
|
|
encoding: Some(UiTransactionEncoding::Base64),
|
|
transaction_details: Some(TransactionDetails::Full),
|
|
rewards: Some(true),
|
|
commitment: Some(commitment_confirmation),
|
|
max_supported_transaction_version: Some(0),
|
|
},
|
|
)
|
|
});
|
|
let blocks = futures::future::join_all(blocks).await;
|
|
for block_slot in blocks.iter().zip(block_slots) {
|
|
let block = match block_slot.0 {
|
|
Ok(x) => x,
|
|
Err(_) => continue,
|
|
};
|
|
let tx_confirm_records = tx_confirm_records.clone();
|
|
let tx_block_data = tx_block_data.clone();
|
|
let transaction_map = transaction_map.clone();
|
|
process_blocks(
|
|
block,
|
|
tx_confirm_records,
|
|
tx_block_data,
|
|
transaction_map,
|
|
block_slot.1,
|
|
commitment_confirmation.commitment,
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
})
|
|
};
|
|
vec![map_filler_jh, cleaner_jh, block_confirmation_jh]
|
|
}
|