noierrdev_antoine_tx_engine/

lib.rs

use std::{
    str::FromStr,
    sync::{
        Arc,
    },
    time::Instant
};
use reqwest::{
    Client as HttpClient,
};
use serde_json::{
    json,
    Value
};
use solana_client::{
    rpc_client::RpcClient,
};
use solana_sdk::{
    signature::{Keypair,Signature,Signer},
    transaction::{ VersionedTransaction},
    instruction::{Instruction},
    hash::Hash,
    pubkey::Pubkey
};
use tokio::{
    time::{sleep, Duration},
    sync::{
        RwLock,
    }
};
use dashmap::{DashMap,DashSet};
use rand::{
    SeedableRng,
    Rng,
    rngs::StdRng
};
pub mod tx;

pub mod jito;
pub mod zero_slot;
pub mod astralane;
pub mod helius;
pub mod temperal;
pub mod node1;
pub mod blockrazor;
pub mod flashblock;
pub mod stellium;

pub mod tip;
pub mod general;
pub mod bundle;
pub mod nonce;
pub mod send;

pub struct TxEngine {
    pub rpc_client : Arc<RpcClient>,
    pub http_client : Arc<HttpClient>,
    pub recent_blockhash : Arc<RwLock<Hash>>,
    pub tx_meta_pool : Arc<DashMap<[u8;64], Value>>
}


impl TxEngine {
    pub fn new(
        rpc_client : &Arc<RpcClient>,
        http_client : &Arc<HttpClient>,
        tx_meta_pool : Arc<DashMap<[u8;64], Value>>
    ) 
    ->Self
    {
        let recent_blockhash : Hash = rpc_client.get_latest_blockhash().unwrap();
        let recent_blockhash_arc=Arc::new(RwLock::new(recent_blockhash));
        tokio::spawn({
            let rpc_client = rpc_client.clone();
            let recent_blockhash_arc = recent_blockhash_arc.clone();
            async move {
                loop {
                    match rpc_client.get_latest_blockhash() {
                        Ok(new_hash) => {
                            *recent_blockhash_arc.write().await = new_hash;
                        }
                        Err(err) => {
                            println!("Failed to refresh blockhash: {:?}", err);
                        }
                    }
                    sleep(Duration::from_secs(1)).await;
                }
            }
        });
        Self { 
            rpc_client : rpc_client.clone(), 
            http_client : http_client.clone(),
            recent_blockhash : recent_blockhash_arc,
            tx_meta_pool : tx_meta_pool
        }
    }

    pub async fn get_blockhash(&self)
    ->Hash
    {
        let recent_blockhash = *self.recent_blockhash.read().await;
        recent_blockhash
    }

    pub async fn send_sequential(&self, sender_url : &str, txs : &Vec<VersionedTransaction>, retry : u64){
        for one_tx in txs {
            let hash = one_tx.signatures[0];
            let mut hash_bytes = [0u8; 64];
            hash_bytes.copy_from_slice(hash.as_ref());
            let mut tx_meta : Value=Value::Null;
            self.tx_meta_pool.insert(hash_bytes, Value::Null);
            let mut success = false;
            for i in 0..retry {
                send::send_url_versioned(&self.http_client, &sender_url, &one_tx).await;
                let mut await_counter = 0;
                loop {
                    let value_opt = self.tx_meta_pool.get(&hash_bytes)
                        .map(|v| v.clone());
                    match value_opt {
                        Some(val) => {
                            if val == Value::Null {
                                await_counter += 1;
                            } else {
                                success=true;
                                tx_meta = val;
                                self.tx_meta_pool.remove(&hash_bytes);
                                break;
                            }
                        }
                        None => {
                            await_counter += 1;
                        }
                    }
                    if await_counter > 150 {
                        break;
                    }
                    tokio::task::yield_now().await;
                    sleep(Duration::from_millis(5)).await;
                }
                if success == true{
                    break;
                }
            }
            if !success {
                println!("Failed to land sequently!");
                break;
            }
        }
    }

    pub fn send_rpc_versioned(&self, tx : VersionedTransaction)
    ->String
    {
        let tx_hash = send::send_rpc_versioned(&self.rpc_client, &tx);
        tx_hash
    }

    pub async fn send_url_versioned(&self, sender_url : &str, tx : VersionedTransaction)
    ->String
    {
        let tx_hash = send::send_url_versioned(&self.http_client, &sender_url, &tx).await;
        tx_hash
    }

    pub async fn send_url_text(&self, sender_url : &str, tx : VersionedTransaction){
        send::send_url_text(&self.http_client, &sender_url, &tx).await;
    }

    pub fn get_nonce(&self, nonce_account : &Pubkey)
    ->Hash
    {
        let hash  :Hash = nonce::get_nonce(&self.rpc_client, &nonce_account);
        hash
    }

    pub fn get_nonce_value(&self, nonce_account : &Pubkey)
    ->Hash
    {
        let nonce_hash = nonce::get_nonce(&self.rpc_client, &nonce_account);
        nonce_hash
    }

    pub async fn send_with_durable_nonce(
        &self, 
        wallets : Vec<Arc<Keypair>>, 
        payer : &Pubkey, 
        mut ixs : Vec<Instruction>, 
        nonce_account : &Pubkey,
        jito_sender_url : &str,
        astralane_sender_url : &str,
        zero_slot_sender_url : &str,
        tip_amount : u64,
        nonce_hash : Hash
    )
    {
        ixs.insert(0,nonce::build_advance_nonce_instruction(&payer, &nonce_account));

        let ixs = Arc::new(ixs);
        let wallets = Arc::new(wallets);

        let payer = *payer;
        let http = self.http_client.clone();
        let mut tasks = Vec::with_capacity(4);

        macro_rules! spawn_sender {
            ($builder:path, $sender:path, $url:expr) => {{
                let http = http.clone();
                let wallets = wallets.clone();
                let ixs = ixs.clone();
                if $url == "" {
                    return;
                }
                let url = $url.to_string();
                let nonce_hash = nonce_hash.clone();
                tokio::spawn(async move {
                    let tx = $builder(
                        (*wallets).clone(),
                        &payer,
                        (*ixs).clone(),
                        tip_amount,
                        nonce_hash,
                    );
                    $sender(&http, &url, &tx).await;
                })
            }};
        }
        if jito_sender_url != "" {
            tasks.push(spawn_sender!(jito::build_tx_with_jito_tip, send::send_url_versioned, jito_sender_url));
        }
        if astralane_sender_url != ""{
            tasks.push(spawn_sender!(astralane::build_tx_with_astralane_tip, send::send_binary_versioned, astralane_sender_url));
        }
        if zero_slot_sender_url != "" {
            tasks.push(spawn_sender!(zero_slot::build_tx_with_zero_slot, send::send_binary_versioned, zero_slot_sender_url));
        }
        futures::future::join_all(tasks).await;
    }

    pub async fn send_json_for_tx_meta(&self, sender_url : &str, tx : &VersionedTransaction)
    ->Value
    {
        let mut meta : Value=Value::Null;
        let hash = tx.signatures[0];
        let mut hash_bytes = [0u8; 64];
        hash_bytes.copy_from_slice(hash.as_ref());
        let start = Instant::now();
        self.tx_meta_pool.insert(hash_bytes, Value::Null);
        send::send_url_versioned(
            &self.http_client,
            &sender_url,
            &tx
        ).await;
        let mut await_counter = 0;
        let mut success:bool=false;
        loop {
            let value_opt = self.tx_meta_pool.get(&hash_bytes)
                .map(|v| v.clone());
            match value_opt {
                Some(val) => {
                    if val == Value::Null {
                        await_counter += 1;
                    } else {
                        success=true;
                        meta = val;
                        self.tx_meta_pool.remove(&hash_bytes);
                        break;
                    }
                }
                None => {
                    await_counter += 1;
                }
            }
            if await_counter > 500 {
                break;
            }
            tokio::task::yield_now().await;
            sleep(Duration::from_millis(1)).await;
        }
        let elapsed = start.elapsed();
        println!("Elapsed: {:?}", elapsed);
        meta
    }


    pub async fn send_binary_for_tx_meta(&self, sender_url : &str, tx : &VersionedTransaction)
    ->Value
    {
        let mut meta : Value=Value::Null;
        let hash = tx.signatures[0];
        let mut hash_bytes = [0u8; 64];
        hash_bytes.copy_from_slice(hash.as_ref());
        let start = Instant::now();
        self.tx_meta_pool.insert(hash_bytes, Value::Null);
        send::send_binary_versioned(
            &self.http_client,
            &sender_url,
            &tx
        ).await;
        let mut await_counter = 0;
        let mut success:bool=false;
        loop {
            let value_opt = self.tx_meta_pool.get(&hash_bytes)
                .map(|v| v.clone());
            match value_opt {
                Some(val) => {
                    if val == Value::Null {
                        await_counter += 1;
                    } else {
                        success=true;
                        meta = val;
                        self.tx_meta_pool.remove(&hash_bytes);
                        break;
                    }
                }
                None => {
                    await_counter += 1;
                }
            }
            if await_counter > 500 {
                break;
            }
            tokio::task::yield_now().await;
            sleep(Duration::from_millis(1)).await;
        }
        let elapsed = start.elapsed();
        println!("Elapsed: {:?}", elapsed);
        meta
    }

    pub async fn send_tx_multiple_urls(&self, urls : Vec<&str>, tx : &VersionedTransaction){
        let http_client = &self.http_client;

        let mut tasks = vec![];
        for one_url in urls {
            let http_client = http_client.clone();
            let tx=tx.clone();
            let one_url=one_url.to_string();
            let one_task = tokio::spawn(async move {
                send::send_url_versioned(&http_client, &one_url, &tx).await;
            });
            tasks.push(one_task);
        }
        futures::future::join_all(tasks).await;
    }

    pub async fn send_jito_bundle(&self, jito_bundle_url :&str, txs : &Vec<VersionedTransaction>){
        jito::send_jito_bundle_versioned(&self.http_client, &jito_bundle_url, &txs).await;
    }

    pub async fn send_astralane_bundle(&self, astralane_bundle_url : &str, txs : &Vec<VersionedTransaction>){
        astralane::send_astralane_bundle(&self.http_client, &astralane_bundle_url, &txs).await;
    }

    pub async fn send_json_with_retry(&self, sender_url : &str, tx : &VersionedTransaction, retry : u64)
    ->Value
    {
        let mut meta : Value=Value::Null;
        let hash = tx.signatures[0];
        let mut hash_bytes = [0u8; 64];
        hash_bytes.copy_from_slice(hash.as_ref());
        self.tx_meta_pool.insert(hash_bytes, Value::Null);
        let mut success = false;
        for i in 0..retry+1 {
            send::send_url_versioned(
                &self.http_client,
                &sender_url,
                &tx
            ).await;
            let mut await_counter = 0;
            let mut success:bool=false;
            loop {
                let value_opt = self.tx_meta_pool.get(&hash_bytes)
                    .map(|v| v.clone());
                match value_opt {
                    Some(val) => {
                        if val == Value::Null {
                            await_counter += 1;
                        } else {
                            success=true;
                            meta = val;
                            self.tx_meta_pool.remove(&hash_bytes);
                            break;
                        }
                    }
                    None => {
                        await_counter += 1;
                    }
                }
                if await_counter > 500 {
                    break;
                }
                tokio::task::yield_now().await;
                sleep(Duration::from_millis(1)).await;
            }
            if success {
                break;
            }
        }
        meta
    }

    pub async fn send_txs_as_jito_bundles(
        &self,
        wallet : &Keypair,
        jito_tip_amount : u64,
        jito_bundle_url : &str,
        txs : &Vec<VersionedTransaction>,
    ){
        let recent_blockhash = self.recent_blockhash.read().await;
        let mut rng = StdRng::from_entropy();
        let random_number = rng.gen_range(0..jito::JITO_TIP_ACCOUNTS.len());
        let mut first_chunk_len = txs.len()%4;
        if first_chunk_len == 0 {
            first_chunk_len =4;
        }
        let mut first_chunk = txs[..first_chunk_len].to_vec();
        let first_jito_tip_tx = jito::build_jito_tip_tx_with_fixed_account(&wallet, random_number, jito_tip_amount, *recent_blockhash);
        first_chunk.insert(0, first_jito_tip_tx);
        self.send_jito_bundle(&jito_bundle_url, &first_chunk).await;
        for (chunk_index, chunk_txs) in txs[first_chunk_len..].chunks(4).enumerate(){
            let mut txs_in_chunk = chunk_txs.to_vec();
            let jito_tip_tx = jito::build_jito_tip_tx_with_fixed_account(&wallet, random_number, jito_tip_amount - ((chunk_index as u64)+1)*2000, *recent_blockhash);
            txs_in_chunk.insert(0, jito_tip_tx);
            self.send_jito_bundle(&jito_bundle_url, &txs_in_chunk).await;
        }
    }
}