diff --git a/Cargo.toml b/Cargo.toml index 33755e8..24cfc5d 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ solana-security-txt = "1.1.1" solana-entry = "2.1.16" solana-rpc-client-nonce-utils = "2.1.16" solana-perf = "2.1.16" +solana-metrics = "2.1.16" spl-token = "8.0.0" spl-token-2022 = { version = "8.0.0", features = ["no-entrypoint"] } diff --git a/src/common/types.rs b/src/common/types.rs index 6c02e37..8b6b6f4 100755 --- a/src/common/types.rs +++ b/src/common/types.rs @@ -3,12 +3,15 @@ use std::sync::Arc; use solana_client::rpc_client::RpcClient; use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Keypair}; use serde::Deserialize; -use crate::{constants::pumpfun::trade::{DEFAULT_BUY_TIP_FEE, DEFAULT_COMPUTE_UNIT_LIMIT, DEFAULT_COMPUTE_UNIT_PRICE, DEFAULT_RPC_UNIT_LIMIT, DEFAULT_RPC_UNIT_PRICE, DEFAULT_SELL_TIP_FEE}, swqos::FeeClient}; +use crate::{constants::pumpfun::trade::{DEFAULT_BUY_TIP_FEE, DEFAULT_COMPUTE_UNIT_LIMIT, DEFAULT_COMPUTE_UNIT_PRICE, DEFAULT_RPC_UNIT_LIMIT, DEFAULT_RPC_UNIT_PRICE, DEFAULT_SELL_TIP_FEE}, swqos::SwqosClient}; #[derive(Debug, Clone, PartialEq)] -pub enum FeeType { +pub enum SwqosType { Jito, NextBlock, + ZeroSlot, + Nozomi, + Rpc, } #[derive(Debug, Clone)] @@ -104,11 +107,11 @@ pub struct MethodArgs { pub payer: Arc, pub rpc: Arc, pub nonblocking_rpc: Arc, - pub jito_client: Arc, + pub jito_client: Arc, } impl MethodArgs { - pub fn new(payer: Arc, rpc: Arc, nonblocking_rpc: Arc, jito_client: Arc) -> Self { + pub fn new(payer: Arc, rpc: Arc, nonblocking_rpc: Arc, jito_client: Arc) -> Self { Self { payer, rpc, nonblocking_rpc, jito_client } } } diff --git a/src/grpc/shred_stream.rs b/src/grpc/shred_stream.rs index c2d5faa..750cdae 100755 --- a/src/grpc/shred_stream.rs +++ b/src/grpc/shred_stream.rs @@ -18,8 +18,8 @@ use crate::common::pumpswap::logs_events::PumpSwapEvent; use crate::common::pumpfun::logs_filters::LogFilter; use crate::common::pumpswap::logs_filters::LogFilter as PumpswapLogFilter; use crate::common::raydium::logs_filters::LogFilter as RaydiumLogFilter; -use crate::swqos::jito_grpc::shredstream::shredstream_proxy_client::ShredstreamProxyClient; -use crate::swqos::jito_grpc::shredstream::SubscribeEntriesRequest; +use crate::protos::shredstream::shredstream_proxy_client::ShredstreamProxyClient; +use crate::protos::shredstream::SubscribeEntriesRequest; const CHANNEL_SIZE: usize = 1000; diff --git a/src/lib.rs b/src/lib.rs index 2ea4bf2..95cde5e 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,11 +9,12 @@ pub mod swqos; pub mod pumpfun; pub mod pumpswap; pub mod trading; +pub mod protos; use std::sync::Arc; use std::sync::Mutex; -use swqos::{FeeClient, JitoClient, NextBlockClient, NozomiClient, SolRpcClient, ZeroSlotClient}; +use swqos::SwqosClient; use rustls::crypto::{ring::default_provider, CryptoProvider}; use solana_hash::Hash; use solana_sdk::{ @@ -30,6 +31,11 @@ use constants::trade_type::{COPY_BUY, SNIPER_BUY}; use constants::trade_platform::{PUMPFUN, PUMPFUN_SWAP, RAYDIUM}; use accounts::BondingCurveAccount; +use crate::swqos::jito::JitoClient; +use crate::swqos::nextblock::NextBlockClient; +use crate::swqos::nozomi::NozomiClient; +use crate::swqos::solana_rpc::SolRpcClient; +use crate::swqos::zeroslot::ZeroSlotClient; use crate::trading::core::params::PumpFunParams; use crate::trading::core::params::PumpFunSellParams; use crate::trading::core::params::PumpSwapParams; @@ -40,7 +46,7 @@ use crate::trading::SellWithTipParams; pub struct SolanaTrade { pub payer: Arc, pub rpc: Arc, - pub fee_clients: Vec>, + pub swqos_clients: Vec>, pub priority_fee: PriorityFee, pub cluster: Cluster, } @@ -52,7 +58,7 @@ impl Clone for SolanaTrade { Self { payer: self.payer.clone(), rpc: self.rpc.clone(), - fee_clients: self.fee_clients.clone(), + swqos_clients: self.swqos_clients.clone(), priority_fee: self.priority_fee.clone(), cluster: self.cluster.clone(), } @@ -76,14 +82,14 @@ impl SolanaTrade { cluster.clone().commitment ); let rpc = Arc::new(rpc); - let mut fee_clients: Vec> = vec![]; + let mut swqos_clients: Vec> = vec![]; if cluster.clone().use_jito { let jito_client = JitoClient::new( cluster.clone().rpc_url, cluster.clone().block_engine_url ).await.expect("Failed to create Jito client"); - fee_clients.push(Arc::new(jito_client)); + swqos_clients.push(Arc::new(jito_client)); } if cluster.clone().use_zeroslot { @@ -93,7 +99,7 @@ impl SolanaTrade { cluster.clone().zeroslot_auth_token ); - fee_clients.push(Arc::new(zeroslot_client)); + swqos_clients.push(Arc::new(zeroslot_client)); } if cluster.clone().use_nozomi { @@ -103,7 +109,7 @@ impl SolanaTrade { cluster.clone().nozomi_auth_token ); - fee_clients.push(Arc::new(nozomi_client)); + swqos_clients.push(Arc::new(nozomi_client)); } if cluster.clone().use_nextblock { @@ -113,18 +119,18 @@ impl SolanaTrade { cluster.clone().nextblock_auth_token ); - fee_clients.push(Arc::new(nextblock_client)); + swqos_clients.push(Arc::new(nextblock_client)); } if cluster.clone().use_rpc { let rpc_client = SolRpcClient::new(rpc.clone()); - fee_clients.push(Arc::new(rpc_client)); + swqos_clients.push(Arc::new(rpc_client)); } let instance = Self { payer, rpc, - fee_clients, + swqos_clients, priority_fee: cluster.clone().priority_fee, cluster: cluster.clone(), }; @@ -192,7 +198,7 @@ impl SolanaTrade { ) -> Result<(), anyhow::Error> { pumpfun::create::create_and_buy_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), payer, mint, ipfs, @@ -350,7 +356,7 @@ impl SolanaTrade { priority_fee.buy_tip_fees = vec![custom_buy_tip_fee.unwrap(),custom_buy_tip_fee.unwrap(),custom_buy_tip_fee.unwrap(),custom_buy_tip_fee.unwrap()]; } pumpfun::buy::buy_with_tip( - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -384,7 +390,7 @@ impl SolanaTrade { .as_any() .downcast_ref::() { pumpfun::buy::buy_with_tip( - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -402,7 +408,7 @@ impl SolanaTrade { .downcast_ref::() { pumpswap::buy::buy_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -441,7 +447,7 @@ impl SolanaTrade { } if trade_platform == PUMPFUN { pumpfun::buy::buy_with_tip( - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -456,7 +462,7 @@ impl SolanaTrade { } else if trade_platform == PUMPFUN_SWAP { pumpswap::buy::buy_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -595,7 +601,7 @@ impl SolanaTrade { if trade_platform == PUMPFUN { pumpfun::sell::sell_by_percent_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -608,7 +614,7 @@ impl SolanaTrade { } else if trade_platform == PUMPFUN_SWAP { pumpswap::sell::sell_by_percent_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -639,7 +645,7 @@ impl SolanaTrade { if trade_platform == PUMPFUN { pumpfun::sell::sell_by_amount_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -651,7 +657,7 @@ impl SolanaTrade { } else if trade_platform == PUMPFUN_SWAP { pumpswap::sell::sell_by_amount_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -681,7 +687,7 @@ impl SolanaTrade { ) -> Result<(), anyhow::Error> { pumpfun::sell::sell_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -808,7 +814,7 @@ impl SolanaTrade { .downcast_ref::() { pumpfun::sell::sell_by_percent_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -824,7 +830,7 @@ impl SolanaTrade { .downcast_ref::() { pumpswap::sell::sell_by_percent_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -858,7 +864,7 @@ impl SolanaTrade { .downcast_ref::() { pumpfun::sell::sell_by_amount_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, @@ -873,7 +879,7 @@ impl SolanaTrade { .downcast_ref::() { pumpswap::sell::sell_by_amount_with_tip( self.rpc.clone(), - self.fee_clients.clone(), + self.swqos_clients.clone(), self.payer.clone(), mint, creator, diff --git a/src/swqos/jito_grpc/auth.rs b/src/protos/auth.rs similarity index 100% rename from src/swqos/jito_grpc/auth.rs rename to src/protos/auth.rs diff --git a/src/swqos/jito_grpc/block.rs b/src/protos/block.rs similarity index 100% rename from src/swqos/jito_grpc/block.rs rename to src/protos/block.rs diff --git a/src/swqos/jito_grpc/block_engine.rs b/src/protos/block_engine.rs similarity index 100% rename from src/swqos/jito_grpc/block_engine.rs rename to src/protos/block_engine.rs diff --git a/src/swqos/jito_grpc/bundle.rs b/src/protos/bundle.rs similarity index 100% rename from src/swqos/jito_grpc/bundle.rs rename to src/protos/bundle.rs diff --git a/src/swqos/jito_grpc/convert.rs b/src/protos/convert.rs similarity index 99% rename from src/swqos/jito_grpc/convert.rs rename to src/protos/convert.rs index 9ea2332..c0e9e1b 100755 --- a/src/swqos/jito_grpc/convert.rs +++ b/src/protos/convert.rs @@ -11,7 +11,7 @@ use solana_sdk::{ transaction::VersionedTransaction, }; -use crate::swqos::jito_grpc::{ +use crate::protos::{ packet::{ Meta as ProtoMeta, Packet as ProtoPacket, PacketBatch as ProtoPacketBatch, PacketFlags as ProtoPacketFlags, diff --git a/src/swqos/jito_grpc/mod.rs b/src/protos/mod.rs similarity index 71% rename from src/swqos/jito_grpc/mod.rs rename to src/protos/mod.rs index ab4a948..c0a3978 100755 --- a/src/swqos/jito_grpc/mod.rs +++ b/src/protos/mod.rs @@ -9,3 +9,6 @@ pub mod shared; pub mod shredstream; pub mod trace_shred; pub mod convert; +pub mod nextblock_grpc; +pub mod searcher_client; +pub mod token_authenticator; diff --git a/src/swqos/api.rs b/src/protos/nextblock_grpc.rs similarity index 100% rename from src/swqos/api.rs rename to src/protos/nextblock_grpc.rs diff --git a/src/swqos/jito_grpc/packet.rs b/src/protos/packet.rs similarity index 100% rename from src/swqos/jito_grpc/packet.rs rename to src/protos/packet.rs diff --git a/src/swqos/jito_grpc/relayer.rs b/src/protos/relayer.rs similarity index 100% rename from src/swqos/jito_grpc/relayer.rs rename to src/protos/relayer.rs diff --git a/src/swqos/jito_grpc/searcher.rs b/src/protos/searcher.rs similarity index 100% rename from src/swqos/jito_grpc/searcher.rs rename to src/protos/searcher.rs diff --git a/src/swqos/searcher_client.rs b/src/protos/searcher_client.rs similarity index 98% rename from src/swqos/searcher_client.rs rename to src/protos/searcher_client.rs index 3dc78fa..bdeeac6 100755 --- a/src/swqos/searcher_client.rs +++ b/src/protos/searcher_client.rs @@ -3,7 +3,7 @@ use std::{ time::{Duration, Instant}, }; -use crate::swqos::jito_grpc::{ +use crate::protos::{ bundle::{ Bundle, BundleResult, }, @@ -25,8 +25,7 @@ use yellowstone_grpc_client::ClientTlsConfig; use crate::swqos::common::poll_transaction_confirmation; use crate::common::SolanaRpcClient; - -use super::TradeType; +use crate::swqos::TradeType; #[derive(Debug, Error)] pub enum BlockEngineConnectionError { diff --git a/src/swqos/jito_grpc/shared.rs b/src/protos/shared.rs similarity index 100% rename from src/swqos/jito_grpc/shared.rs rename to src/protos/shared.rs diff --git a/src/swqos/jito_grpc/shredstream.rs b/src/protos/shredstream.rs similarity index 100% rename from src/swqos/jito_grpc/shredstream.rs rename to src/protos/shredstream.rs diff --git a/src/swqos/token_authenticator.rs b/src/protos/token_authenticator.rs similarity index 99% rename from src/swqos/token_authenticator.rs rename to src/protos/token_authenticator.rs index 917efb9..debb4ca 100755 --- a/src/swqos/token_authenticator.rs +++ b/src/protos/token_authenticator.rs @@ -3,7 +3,7 @@ use std::{ time::{Duration, SystemTime}, }; -use jito_protos::auth::{ +use crate::protos::auth::{ auth_service_client::AuthServiceClient, GenerateAuthChallengeRequest, GenerateAuthTokensRequest, RefreshAccessTokenRequest, Role, Token, }; diff --git a/src/swqos/jito_grpc/trace_shred.rs b/src/protos/trace_shred.rs similarity index 100% rename from src/swqos/jito_grpc/trace_shred.rs rename to src/protos/trace_shred.rs diff --git a/src/pumpfun/buy.rs b/src/pumpfun/buy.rs index 3790a01..f23325a 100755 --- a/src/pumpfun/buy.rs +++ b/src/pumpfun/buy.rs @@ -1,7 +1,7 @@ use crate::accounts::BondingCurveAccount; use crate::{ common::{PriorityFee, SolanaRpcClient}, - swqos::FeeClient, + swqos::SwqosClient, trading::{core::params::PumpFunParams, factory::Protocol, BuyParams, TradeFactory}, }; use solana_hash::Hash; @@ -49,7 +49,7 @@ pub async fn buy( } pub async fn buy_with_tip( - fee_clients: Vec>, + swqos_clients: Vec>, payer: Arc, mint: Pubkey, creator: Pubkey, @@ -82,7 +82,7 @@ pub async fn buy_with_tip( data_size_limit: MAX_LOADED_ACCOUNTS_DATA_SIZE_LIMIT, protocol_params, }; - let buy_with_tip_params = buy_params.with_tip(fee_clients); + let buy_with_tip_params = buy_params.with_tip(swqos_clients); // 执行买入 executor.buy_with_tip(buy_with_tip_params).await?; Ok(()) diff --git a/src/pumpfun/create.rs b/src/pumpfun/create.rs index 0f3b529..89a9e17 100755 --- a/src/pumpfun/create.rs +++ b/src/pumpfun/create.rs @@ -16,7 +16,7 @@ use spl_associated_token_account::instruction::create_associated_token_account; use crate::{ common::{PriorityFee, SolanaRpcClient}, constants, instruction, - ipfs::TokenMetadataIPFS, swqos::{FeeClient, TradeType}, + ipfs::TokenMetadataIPFS, swqos::{SwqosClient, TradeType}, }; use crate::pumpfun::common::{ @@ -86,7 +86,7 @@ pub async fn create_and_buy( pub async fn create_and_buy_with_tip( rpc: Arc, - fee_clients: Vec>, + fee_clients: Vec>, payer: Arc, mint: Keypair, ipfs: TokenMetadataIPFS, diff --git a/src/pumpfun/sell.rs b/src/pumpfun/sell.rs index 546e4e8..b8875da 100755 --- a/src/pumpfun/sell.rs +++ b/src/pumpfun/sell.rs @@ -3,7 +3,7 @@ use crate::trading::{ }; use crate::{ common::{PriorityFee, SolanaRpcClient}, - swqos::FeeClient, + swqos::SwqosClient, }; use anyhow::anyhow; use solana_hash::Hash; @@ -99,7 +99,7 @@ pub async fn sell_by_amount( pub async fn sell_by_percent_with_tip( rpc: Arc, - fee_clients: Vec>, + fee_clients: Vec>, payer: Arc, mint: Pubkey, creator: Pubkey, @@ -129,7 +129,7 @@ pub async fn sell_by_percent_with_tip( pub async fn sell_by_amount_with_tip( rpc: Arc, - fee_clients: Vec>, + fee_clients: Vec>, payer: Arc, mint: Pubkey, creator: Pubkey, @@ -158,7 +158,7 @@ pub async fn sell_by_amount_with_tip( /// Sell tokens using Jito pub async fn sell_with_tip( rpc: Arc, - fee_clients: Vec>, + fee_clients: Vec>, payer: Arc, mint: Pubkey, creator: Pubkey, diff --git a/src/pumpswap/buy.rs b/src/pumpswap/buy.rs index e519311..409b732 100755 --- a/src/pumpswap/buy.rs +++ b/src/pumpswap/buy.rs @@ -2,7 +2,7 @@ use solana_hash::Hash; use solana_sdk::{pubkey::Pubkey, signature::Keypair}; use std::sync::Arc; -use crate::swqos::FeeClient; +use crate::swqos::SwqosClient; use crate::trading::{core::params::PumpSwapParams, factory::Protocol, BuyParams, TradeFactory}; use crate::{common::PriorityFee, SolanaRpcClient}; @@ -62,7 +62,7 @@ pub async fn buy( // Buy tokens using a MEV service pub async fn buy_with_tip( rpc: Arc, - fee_clients: Vec>, + swqos_clients: Vec>, payer: Arc, mint: Pubkey, creator: Pubkey, @@ -104,7 +104,7 @@ pub async fn buy_with_tip( data_size_limit: MAX_LOADED_ACCOUNTS_DATA_SIZE_LIMIT, protocol_params, }; - let buy_with_tip_params = buy_params.with_tip(fee_clients); + let buy_with_tip_params = buy_params.with_tip(swqos_clients); // 执行买入 executor.buy_with_tip(buy_with_tip_params).await?; Ok(()) diff --git a/src/pumpswap/sell.rs b/src/pumpswap/sell.rs index 87532df..ece9211 100755 --- a/src/pumpswap/sell.rs +++ b/src/pumpswap/sell.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use crate::common::{PriorityFee, SolanaRpcClient}; use crate::pumpswap::common::get_token_balance; -use crate::swqos::FeeClient; +use crate::swqos::SwqosClient; use crate::trading::{core::params::PumpSwapParams, factory::Protocol, SellParams, TradeFactory}; // Sell tokens to a Pumpswap pool @@ -140,7 +140,7 @@ pub async fn sell_by_amount( // Sell tokens using a MEV service pub async fn sell_with_tip( rpc: Arc, - fee_clients: Vec>, + swqos_clients: Vec>, payer: Arc, mint: Pubkey, creator: Pubkey, @@ -179,7 +179,7 @@ pub async fn sell_with_tip( recent_blockhash, protocol_params, }; - let sell_with_tip_params = sell_params.with_tip(fee_clients); + let sell_with_tip_params = sell_params.with_tip(swqos_clients); // 执行卖出交易 executor.sell_with_tip(sell_with_tip_params).await?; Ok(()) @@ -188,7 +188,7 @@ pub async fn sell_with_tip( // Sell tokens by percentage using a MEV service pub async fn sell_by_percent_with_tip( rpc: Arc, - fee_clients: Vec>, + swqos_clients: Vec>, payer: Arc, mint: Pubkey, creator: Pubkey, @@ -212,7 +212,7 @@ pub async fn sell_by_percent_with_tip( let amount = balance_u64 * percent / 100; sell_with_tip( rpc, - fee_clients, + swqos_clients, payer, mint, creator, @@ -233,7 +233,7 @@ pub async fn sell_by_percent_with_tip( // Sell tokens by amount using a MEV service pub async fn sell_by_amount_with_tip( rpc: Arc, - fee_clients: Vec>, + swqos_clients: Vec>, payer: Arc, mint: Pubkey, creator: Pubkey, @@ -255,7 +255,7 @@ pub async fn sell_by_amount_with_tip( sell_with_tip( rpc, - fee_clients, + swqos_clients, payer, mint, creator, diff --git a/src/swqos/jito.rs b/src/swqos/jito.rs new file mode 100755 index 0000000..f46cd2b --- /dev/null +++ b/src/swqos/jito.rs @@ -0,0 +1,66 @@ + +use tonic::transport::Channel; +use tokio::sync::Mutex; + +use rand::{rng, seq::IteratorRandom}; +use anyhow::{anyhow, Result}; +use std::sync::Arc; +use solana_sdk::{transaction::VersionedTransaction, signature::Signature}; +use crate::protos::searcher::searcher_service_client::SearcherServiceClient; +use crate::protos::searcher_client::{self, get_searcher_client_no_auth, send_bundle_with_confirmation}; +use crate::swqos::{ClientType, TradeType}; +use crate::swqos::SwqosClientTrait; + +use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::JITO_TIP_ACCOUNTS}; + + +pub struct JitoClient { + pub rpc_client: Arc, + pub searcher_client: Arc>>, +} + +#[async_trait::async_trait] +impl SwqosClientTrait for JitoClient { + async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { + self.send_bundle_with_confirmation(trade_type, &vec![transaction.clone()]).await?.first().cloned().ok_or(anyhow!("Failed to send transaction")) + } + + async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { + self.send_bundle_with_confirmation(trade_type, transactions).await + } + + fn get_tip_account(&self) -> Result { + if let Some(acc) = JITO_TIP_ACCOUNTS.iter().choose(&mut rng()) { + Ok(acc.to_string()) + } else { + Err(anyhow!("no valid tip accounts found")) + } + } + + fn get_client_type(&self) -> ClientType { + ClientType::Jito + } +} + +impl JitoClient { + pub async fn new(rpc_url: String, block_engine_url: String) -> Result { + let rpc_client = SolanaRpcClient::new(rpc_url); + let searcher_client = get_searcher_client_no_auth(block_engine_url.as_str()).await?; + Ok(Self { rpc_client: Arc::new(rpc_client), searcher_client: Arc::new(Mutex::new(searcher_client)) }) + } + + pub async fn send_bundle_with_confirmation( + &self, + trade_type: TradeType, + transactions: &Vec, + ) -> Result> { + send_bundle_with_confirmation(self.rpc_client.clone(), trade_type, &transactions, self.searcher_client.clone()).await + } + + pub async fn send_bundle_no_wait( + &self, + transactions: &Vec, + ) -> Result> { + searcher_client::send_bundle_no_wait(&transactions, self.searcher_client.clone()).await + } +} \ No newline at end of file diff --git a/src/swqos/mod.rs b/src/swqos/mod.rs index 1e9ad67..9adc709 100755 --- a/src/swqos/mod.rs +++ b/src/swqos/mod.rs @@ -1,35 +1,17 @@ -use api::api_client::ApiClient; -use common::{poll_transaction_confirmation, serialize_smart_transaction_and_encode, serialize_transaction_and_encode}; -use solana_client::rpc_config::RpcSendTransactionConfig; -use crate::swqos::jito_grpc::searcher::searcher_service_client::SearcherServiceClient; -use reqwest::Client; -use searcher_client::{get_searcher_client_no_auth, send_bundle_with_confirmation}; -use serde_json::json; -use tonic::transport::Channel; -use yellowstone_grpc_client::Interceptor; -use std::{sync::Arc, time::Instant}; -use tokio::sync::{Mutex, RwLock}; -use solana_sdk::{commitment_config::CommitmentLevel, signature::Signature}; - -use std::str::FromStr; -use rustls::crypto::{ring::default_provider, CryptoProvider}; - -use tonic::{service::interceptor::InterceptedService, transport::Uri, Status}; -use std::time::Duration; -use solana_transaction_status::UiTransactionEncoding; -use tonic::transport::ClientTlsConfig; - -use anyhow::{anyhow, Result}; -use rand::{rng, seq::{IndexedRandom, IteratorRandom}}; -use solana_sdk::transaction::VersionedTransaction; - -use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::{JITO_TIP_ACCOUNTS, NEXTBLOCK_TIP_ACCOUNTS, ZEROSLOT_TIP_ACCOUNTS, NOZOMI_TIP_ACCOUNTS}}; - -pub mod api; pub mod common; -pub mod searcher_client; -pub mod jito_grpc; +pub mod solana_rpc; +pub mod jito; +pub mod nextblock; +pub mod zeroslot; +pub mod nozomi; +pub mod types; + +use solana_sdk::signature::Signature; +use solana_sdk::transaction::VersionedTransaction; +use tokio::sync::RwLock; + +use anyhow::Result; lazy_static::lazy_static! { static ref TIP_ACCOUNT_CACHE: RwLock> = RwLock::new(Vec::new()); @@ -64,458 +46,12 @@ pub enum ClientType { Rpc, } -pub type FeeClient = dyn FeeClientTrait + Send + Sync + 'static; +pub type SwqosClient = dyn SwqosClientTrait + Send + Sync + 'static; #[async_trait::async_trait] -pub trait FeeClientTrait { +pub trait SwqosClientTrait { async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result; async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result>; fn get_tip_account(&self) -> Result; fn get_client_type(&self) -> ClientType; -} - -#[derive(Clone)] -pub struct SolRpcClient { - pub rpc_client: Arc, -} - -#[async_trait::async_trait] -impl FeeClientTrait for SolRpcClient { - async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { - let signature = self.rpc_client.send_transaction_with_config(transaction, RpcSendTransactionConfig{ - skip_preflight: true, - preflight_commitment: Some(CommitmentLevel::Processed), - encoding: Some(UiTransactionEncoding::Base64), - max_retries: Some(3), - min_context_slot: Some(0), - }).await?; - - let start_time = Instant::now(); - match poll_transaction_confirmation(&self.rpc_client, signature).await { - Ok(_) => (), - Err(_) => (), - } - println!(" signature: {:?}", signature); - println!(" rpc{}确认: {:?}", trade_type, start_time.elapsed()); - - Ok(signature) - } - - async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { - let mut signatures = Vec::new(); - for transaction in transactions { - let signature = self.send_transaction(trade_type, transaction).await?; - signatures.push(signature); - } - Ok(signatures) - } - - fn get_tip_account(&self) -> Result { - Ok("".to_string()) - } - - fn get_client_type(&self) -> ClientType { - ClientType::Rpc - } -} - -impl SolRpcClient { - pub fn new(rpc_client: Arc) -> Self { - Self { rpc_client } - } -} - -pub struct JitoClient { - pub rpc_client: Arc, - pub searcher_client: Arc>>, -} - -#[async_trait::async_trait] -impl FeeClientTrait for JitoClient { - async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { - self.send_bundle_with_confirmation(trade_type, &vec![transaction.clone()]).await?.first().cloned().ok_or(anyhow!("Failed to send transaction")) - } - - async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result, anyhow::Error> { - self.send_bundle_with_confirmation(trade_type, transactions).await - } - - fn get_tip_account(&self) -> Result { - if let Some(acc) = JITO_TIP_ACCOUNTS.iter().choose(&mut rng()) { - Ok(acc.to_string()) - } else { - Err(anyhow!("no valid tip accounts found")) - } - } - - fn get_client_type(&self) -> ClientType { - ClientType::Jito - } -} - -impl JitoClient { - pub async fn new(rpc_url: String, block_engine_url: String) -> Result { - let rpc_client = SolanaRpcClient::new(rpc_url); - let searcher_client = get_searcher_client_no_auth(block_engine_url.as_str()).await?; - Ok(Self { rpc_client: Arc::new(rpc_client), searcher_client: Arc::new(Mutex::new(searcher_client)) }) - } - - pub async fn send_bundle_with_confirmation( - &self, - trade_type: TradeType, - transactions: &Vec, - ) -> Result, anyhow::Error> { - send_bundle_with_confirmation(self.rpc_client.clone(), trade_type, &transactions, self.searcher_client.clone()).await - } - - pub async fn send_bundle_no_wait( - &self, - transactions: &Vec, - ) -> Result, anyhow::Error> { - searcher_client::send_bundle_no_wait(&transactions, self.searcher_client.clone()).await - } -} - -#[derive(Clone)] -pub struct MyInterceptor { - auth_token: String, -} - -impl MyInterceptor { - pub fn new(auth_token: String) -> Self { - Self { auth_token } - } -} - -impl Interceptor for MyInterceptor { - fn call(&mut self, mut request: tonic::Request<()>) -> Result, Status> { - request.metadata_mut().insert( - "authorization", - tonic::metadata::MetadataValue::from_str(&self.auth_token) - .map_err(|_| Status::invalid_argument("Invalid auth token"))? - ); - Ok(request) - } -} - -#[derive(Clone)] -pub struct NextBlockClient { - pub rpc_client: Arc, - pub client: ApiClient>, -} - -#[async_trait::async_trait] -impl FeeClientTrait for NextBlockClient { - async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { - self.send_transaction(trade_type, transaction).await - } - - async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result, anyhow::Error> { - self.send_transactions(trade_type, transactions).await - } - - fn get_tip_account(&self) -> Result { - let tip_account = *NEXTBLOCK_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| NEXTBLOCK_TIP_ACCOUNTS.first()).unwrap(); - Ok(tip_account.to_string()) - } - - fn get_client_type(&self) -> ClientType { - ClientType::NextBlock - } -} - -impl NextBlockClient { - pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self { - if CryptoProvider::get_default().is_none() { - let _ = default_provider() - .install_default() - .map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e)); - } - - let endpoint = endpoint.parse::().unwrap(); - let tls = ClientTlsConfig::new().with_native_roots(); - let channel = Channel::builder(endpoint) - .tls_config(tls).expect("Failed to create TLS config") - .tcp_keepalive(Some(Duration::from_secs(60))) - .http2_keep_alive_interval(Duration::from_secs(30)) - .keep_alive_while_idle(true) - .timeout(Duration::from_secs(30)) - .connect_timeout(Duration::from_secs(10)) - .connect_lazy(); - - let client = ApiClient::with_interceptor(channel, MyInterceptor::new(auth_token)); - let rpc_client = SolanaRpcClient::new(rpc_url); - Self { rpc_client: Arc::new(rpc_client), client } - } - - pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { - let start_time = Instant::now(); - let (content, signature) = serialize_smart_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?; - - self.client.clone().post_submit_v2(api::PostSubmitRequest { - transaction: Some(api::TransactionMessage { - content, - is_cleanup: false, - }), - skip_pre_flight: true, - front_running_protection: Some(true), - experimental_front_running_protection: Some(true), - snipe_transaction: Some(true), - }).await?; - - println!(" nextblock{}提交: {:?}", trade_type, start_time.elapsed()); - - let start_time: Instant = Instant::now(); - let timeout: Duration = Duration::from_secs(10); - while Instant::now().duration_since(start_time) < timeout { - match poll_transaction_confirmation(&self.rpc_client, signature).await { - Ok(_) => break, - Err(_) => continue, - } - } - - println!(" nextblock{}确认: {:?}", trade_type, start_time.elapsed()); - - Ok(signature) - } - - pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result, anyhow::Error> { - let mut entries = Vec::new(); - let encoding = UiTransactionEncoding::Base64; - - let mut signatures = Vec::new(); - for transaction in transactions { - let (content, signature) = serialize_smart_transaction_and_encode(transaction, encoding).await?; - entries.push(api::PostSubmitRequestEntry { - transaction: Some(api::TransactionMessage { - content, - is_cleanup: false, - }), - skip_pre_flight: true, - }); - signatures.push(signature); - } - - self.client.clone().post_submit_batch_v2(api::PostSubmitBatchRequest { - entries, - submit_strategy: api::SubmitStrategy::PSubmitAll as i32, - use_bundle: Some(true), - front_running_protection: Some(true), - }).await?; - - let start_time: Instant = Instant::now(); - for signature in signatures.clone() { - match poll_transaction_confirmation(&self.rpc_client, signature).await { - Ok(_) => continue, - Err(_) => continue, - } - } - - println!(" nextblock{}确认: {:?}", trade_type, start_time.elapsed()); - - Ok(signatures) - } -} - -#[derive(Clone)] -pub struct ZeroSlotClient { - pub endpoint: String, - pub auth_token: String, - pub rpc_client: Arc, - pub http_client: Client, -} - -#[async_trait::async_trait] -impl FeeClientTrait for ZeroSlotClient { - async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { - self.send_transaction(trade_type, transaction).await - } - - async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result, anyhow::Error> { - self.send_transactions(trade_type, transactions).await - } - - fn get_tip_account(&self) -> Result { - let tip_account = *ZEROSLOT_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| ZEROSLOT_TIP_ACCOUNTS.first()).unwrap(); - Ok(tip_account.to_string()) - } - - fn get_client_type(&self) -> ClientType { - ClientType::ZeroSlot - } -} - -impl ZeroSlotClient { - pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self { - let rpc_client = SolanaRpcClient::new(rpc_url); - let http_client = Client::builder() - .pool_idle_timeout(Duration::from_secs(60)) - .pool_max_idle_per_host(64) - .tcp_keepalive(Some(Duration::from_secs(1200))) - .http2_keep_alive_interval(Duration::from_secs(15)) - .timeout(Duration::from_secs(10)) - .connect_timeout(Duration::from_secs(5)) - .build() - .unwrap(); - Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client } - } - - pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { - let start_time = Instant::now(); - let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?; - println!(" 交易编码base64: {:?}", start_time.elapsed()); - - let request_body = serde_json::to_string(&json!({ - "jsonrpc": "2.0", - "id": 1, - "method": "sendTransaction", - "params": [ - content, - { "encoding": "base64", "skipPreflight": true } - ] - }))?; - - let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20); - url.push_str(&self.endpoint); - url.push_str("/?api-key="); - url.push_str(&self.auth_token); - - // 4. 直接使用 `text().await?`，避免 `json().await?` 的异步 JSON 解析 - let response_text = self.http_client.post(&url) - .body(request_body) // 直接传字符串，避免 `json()` 开销 - .header("Content-Type", "application/json") // 显式指定 JSON 头 - .send() - .await? - .text() - .await?; - - // 5. 用 `serde_json::from_str()` 解析 JSON，减少 `.json().await?` 额外等待 - if let Ok(response_json) = serde_json::from_str::(&response_text) { - if response_json.get("result").is_some() { - println!(" 0slot{}提交: {:?}", trade_type, start_time.elapsed()); - } else if let Some(_error) = response_json.get("error") { - eprintln!(" 0slot{}提交失败: {:?}", trade_type, _error); - } - } - - let start_time: Instant = Instant::now(); - match poll_transaction_confirmation(&self.rpc_client, signature).await { - Ok(_) => (), - Err(_) => (), - } - - println!(" 0slot{}确认: {:?}", trade_type, start_time.elapsed()); - - Ok(signature) - } - - pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result, anyhow::Error> { - let mut signatures = Vec::new(); - for transaction in transactions { - let signature = self.send_transaction(trade_type, transaction).await?; - signatures.push(signature); - } - Ok(signatures) - } -} - -#[derive(Clone)] -pub struct NozomiClient { - pub rpc_client: Arc, - pub endpoint: String, - pub auth_token: String, - pub http_client: Client, -} - -#[async_trait::async_trait] -impl FeeClientTrait for NozomiClient { - async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { - self.send_transaction(trade_type, transaction).await - } - - async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result, anyhow::Error> { - self.send_transactions(trade_type, transactions).await - } - - fn get_tip_account(&self) -> Result { - let tip_account = *NOZOMI_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| NOZOMI_TIP_ACCOUNTS.first()).unwrap(); - Ok(tip_account.to_string()) - } - - fn get_client_type(&self) -> ClientType { - ClientType::Nozomi - } -} - -impl NozomiClient { - pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self { - let rpc_client = SolanaRpcClient::new(rpc_url); - let http_client = Client::builder() - .pool_idle_timeout(Duration::from_secs(60)) - .pool_max_idle_per_host(64) - .tcp_keepalive(Some(Duration::from_secs(1200))) - .http2_keep_alive_interval(Duration::from_secs(15)) - .timeout(Duration::from_secs(10)) - .connect_timeout(Duration::from_secs(5)) - .build() - .unwrap(); - Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client } - } - - pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { - let start_time = Instant::now(); - let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?; - println!(" 交易编码base64: {:?}", start_time.elapsed()); - - // 按照 Nozomi 文档要求构建请求体 - let request_body = serde_json::to_string(&json!({ - "jsonrpc": "2.0", - "id": 1, - "method": "sendTransaction", - "params": [ - content, - { "encoding": "base64" } - ] - }))?; - - let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20); - url.push_str(&self.endpoint); - url.push_str("/?c="); - url.push_str(&self.auth_token); - - let response_text = self.http_client.post(&url) - .body(request_body) - .header("Content-Type", "application/json") - .send() - .await? - .text() - .await?; - - if let Ok(response_json) = serde_json::from_str::(&response_text) { - if response_json.get("result").is_some() { - println!(" nozomi{}提交: {:?}", trade_type, start_time.elapsed()); - } else if let Some(_error) = response_json.get("error") { - // eprintln!("nozomi交易提交失败: {:?}", _error); - } - } - - let start_time: Instant = Instant::now(); - match poll_transaction_confirmation(&self.rpc_client, signature).await { - Ok(_) => (), - Err(_) => (), - } - - println!(" nozomi{}确认: {:?}", trade_type, start_time.elapsed()); - - Ok(signature) - } - - pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result, anyhow::Error> { - let mut signatures = Vec::new(); - for transaction in transactions { - let signature = self.send_transaction(trade_type, transaction).await?; - signatures.push(signature); - } - Ok(signatures) - } } \ No newline at end of file diff --git a/src/swqos/nextblock.rs b/src/swqos/nextblock.rs new file mode 100755 index 0000000..ba990fd --- /dev/null +++ b/src/swqos/nextblock.rs @@ -0,0 +1,166 @@ +use crate::protos::nextblock_grpc; +use crate::protos::nextblock_grpc::api_client::ApiClient; +use crate::swqos::common::{poll_transaction_confirmation, serialize_smart_transaction_and_encode}; +use rand::seq::IndexedRandom; +use rustls::crypto::ring::default_provider; +use rustls::crypto::CryptoProvider; +use tonic::transport::Channel; +use yellowstone_grpc_client::Interceptor; +use std::str::FromStr; +use std::{sync::Arc, time::Instant}; + +use solana_sdk::{signature::Signature}; + +use tonic::{service::interceptor::InterceptedService, transport::Uri, Status}; +use std::time::Duration; +use solana_transaction_status::UiTransactionEncoding; +use tonic::transport::ClientTlsConfig; + +use anyhow::Result; +use solana_sdk::transaction::VersionedTransaction; +use crate::swqos::{ClientType, TradeType}; +use crate::swqos::SwqosClientTrait; + +use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::NEXTBLOCK_TIP_ACCOUNTS}; + + +#[derive(Clone)] +pub struct MyInterceptor { + auth_token: String, +} + +impl MyInterceptor { + pub fn new(auth_token: String) -> Self { + Self { auth_token } + } +} + +impl Interceptor for MyInterceptor { + fn call(&mut self, mut request: tonic::Request<()>) -> Result, Status> { + request.metadata_mut().insert( + "authorization", + tonic::metadata::MetadataValue::from_str(&self.auth_token) + .map_err(|_| Status::invalid_argument("Invalid auth token"))? + ); + Ok(request) + } +} + +#[derive(Clone)] +pub struct NextBlockClient { + pub rpc_client: Arc, + pub client: ApiClient>, +} + +#[async_trait::async_trait] +impl SwqosClientTrait for NextBlockClient { + async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { + self.send_transaction(trade_type, transaction).await + } + + async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { + self.send_transactions(trade_type, transactions).await + } + + fn get_tip_account(&self) -> Result { + let tip_account = *NEXTBLOCK_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| NEXTBLOCK_TIP_ACCOUNTS.first()).unwrap(); + Ok(tip_account.to_string()) + } + + fn get_client_type(&self) -> ClientType { + ClientType::NextBlock + } +} + +impl NextBlockClient { + pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self { + if CryptoProvider::get_default().is_none() { + let _ = default_provider() + .install_default() + .map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e)); + } + + let endpoint = endpoint.parse::().unwrap(); + let tls = ClientTlsConfig::new().with_native_roots(); + let channel = Channel::builder(endpoint) + .tls_config(tls).expect("Failed to create TLS config") + .tcp_keepalive(Some(Duration::from_secs(60))) + .http2_keep_alive_interval(Duration::from_secs(30)) + .keep_alive_while_idle(true) + .timeout(Duration::from_secs(30)) + .connect_timeout(Duration::from_secs(10)) + .connect_lazy(); + + let client = ApiClient::with_interceptor(channel, MyInterceptor::new(auth_token)); + let rpc_client = SolanaRpcClient::new(rpc_url); + Self { rpc_client: Arc::new(rpc_client), client } + } + + pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { + let start_time = Instant::now(); + let (content, signature) = serialize_smart_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?; + + self.client.clone().post_submit_v2(nextblock_grpc::PostSubmitRequest { + transaction: Some(nextblock_grpc::TransactionMessage { + content, + is_cleanup: false, + }), + skip_pre_flight: true, + front_running_protection: Some(true), + experimental_front_running_protection: Some(true), + snipe_transaction: Some(true), + }).await?; + + println!(" nextblock{}提交: {:?}", trade_type, start_time.elapsed()); + + let start_time: Instant = Instant::now(); + let timeout: Duration = Duration::from_secs(10); + while Instant::now().duration_since(start_time) < timeout { + match poll_transaction_confirmation(&self.rpc_client, signature).await { + Ok(_) => break, + Err(_) => continue, + } + } + + println!(" nextblock{}确认: {:?}", trade_type, start_time.elapsed()); + + Ok(signature) + } + + pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { + let mut entries = Vec::new(); + let encoding = UiTransactionEncoding::Base64; + + let mut signatures = Vec::new(); + for transaction in transactions { + let (content, signature) = serialize_smart_transaction_and_encode(transaction, encoding).await?; + entries.push(nextblock_grpc::PostSubmitRequestEntry { + transaction: Some(nextblock_grpc::TransactionMessage { + content, + is_cleanup: false, + }), + skip_pre_flight: true, + }); + signatures.push(signature); + } + + self.client.clone().post_submit_batch_v2(nextblock_grpc::PostSubmitBatchRequest { + entries, + submit_strategy: nextblock_grpc::SubmitStrategy::PSubmitAll as i32, + use_bundle: Some(true), + front_running_protection: Some(true), + }).await?; + + let start_time: Instant = Instant::now(); + for signature in signatures.clone() { + match poll_transaction_confirmation(&self.rpc_client, signature).await { + Ok(_) => continue, + Err(_) => continue, + } + } + + println!(" nextblock{}确认: {:?}", trade_type, start_time.elapsed()); + + Ok(signatures) + } +} \ No newline at end of file diff --git a/src/swqos/nozomi.rs b/src/swqos/nozomi.rs new file mode 100755 index 0000000..062d7c8 --- /dev/null +++ b/src/swqos/nozomi.rs @@ -0,0 +1,120 @@ + +use crate::swqos::common::{poll_transaction_confirmation, serialize_transaction_and_encode}; +use rand::seq::IndexedRandom; +use reqwest::Client; +use serde_json::json; +use std::{sync::Arc, time::Instant}; + +use solana_sdk::{signature::Signature}; + +use std::time::Duration; +use solana_transaction_status::UiTransactionEncoding; + +use anyhow::Result; +use solana_sdk::transaction::VersionedTransaction; +use crate::swqos::{ClientType, TradeType}; +use crate::swqos::SwqosClientTrait; + +use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::NOZOMI_TIP_ACCOUNTS}; + + +#[derive(Clone)] +pub struct NozomiClient { + pub rpc_client: Arc, + pub endpoint: String, + pub auth_token: String, + pub http_client: Client, +} + +#[async_trait::async_trait] +impl SwqosClientTrait for NozomiClient { + async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { + self.send_transaction(trade_type, transaction).await + } + + async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { + self.send_transactions(trade_type, transactions).await + } + + fn get_tip_account(&self) -> Result { + let tip_account = *NOZOMI_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| NOZOMI_TIP_ACCOUNTS.first()).unwrap(); + Ok(tip_account.to_string()) + } + + fn get_client_type(&self) -> ClientType { + ClientType::Nozomi + } +} + +impl NozomiClient { + pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self { + let rpc_client = SolanaRpcClient::new(rpc_url); + let http_client = Client::builder() + .pool_idle_timeout(Duration::from_secs(60)) + .pool_max_idle_per_host(64) + .tcp_keepalive(Some(Duration::from_secs(1200))) + .http2_keep_alive_interval(Duration::from_secs(15)) + .timeout(Duration::from_secs(10)) + .connect_timeout(Duration::from_secs(5)) + .build() + .unwrap(); + Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client } + } + + pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { + let start_time = Instant::now(); + let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?; + println!(" 交易编码base64: {:?}", start_time.elapsed()); + + // 按照 Nozomi 文档要求构建请求体 + let request_body = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "sendTransaction", + "params": [ + content, + { "encoding": "base64" } + ] + }))?; + + let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20); + url.push_str(&self.endpoint); + url.push_str("/?c="); + url.push_str(&self.auth_token); + + let response_text = self.http_client.post(&url) + .body(request_body) + .header("Content-Type", "application/json") + .send() + .await? + .text() + .await?; + + if let Ok(response_json) = serde_json::from_str::(&response_text) { + if response_json.get("result").is_some() { + println!(" nozomi{}提交: {:?}", trade_type, start_time.elapsed()); + } else if let Some(_error) = response_json.get("error") { + // eprintln!("nozomi交易提交失败: {:?}", _error); + } + } + + let start_time: Instant = Instant::now(); + match poll_transaction_confirmation(&self.rpc_client, signature).await { + Ok(_) => (), + Err(_) => (), + } + + println!(" nozomi{}确认: {:?}", trade_type, start_time.elapsed()); + + Ok(signature) + } + + pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { + let mut signatures = Vec::new(); + for transaction in transactions { + let signature = self.send_transaction(trade_type, transaction).await?; + signatures.push(signature); + } + Ok(signatures) + } +} \ No newline at end of file diff --git a/src/swqos/solana_rpc.rs b/src/swqos/solana_rpc.rs new file mode 100755 index 0000000..c282439 --- /dev/null +++ b/src/swqos/solana_rpc.rs @@ -0,0 +1,64 @@ +use std::{sync::Arc, time::Instant}; + +use solana_client::rpc_config::RpcSendTransactionConfig; +use solana_sdk::{ + commitment_config::CommitmentLevel, + signature::Signature, + transaction::VersionedTransaction, +}; +use solana_transaction_status::UiTransactionEncoding; + +use crate::{common::SolanaRpcClient, swqos::{common::poll_transaction_confirmation, ClientType, TradeType}}; +use crate::swqos::SwqosClientTrait; +use anyhow::Result; + +#[derive(Clone)] +pub struct SolRpcClient { + pub rpc_client: Arc, +} + +#[async_trait::async_trait] +impl SwqosClientTrait for SolRpcClient { + async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { + let signature = self.rpc_client.send_transaction_with_config(transaction, RpcSendTransactionConfig{ + skip_preflight: true, + preflight_commitment: Some(CommitmentLevel::Processed), + encoding: Some(UiTransactionEncoding::Base64), + max_retries: Some(3), + min_context_slot: Some(0), + }).await?; + + let start_time = Instant::now(); + match poll_transaction_confirmation(&self.rpc_client, signature).await { + Ok(_) => (), + Err(_) => (), + } + println!(" signature: {:?}", signature); + println!(" rpc{}确认: {:?}", trade_type, start_time.elapsed()); + + Ok(signature) + } + + async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { + let mut signatures = Vec::new(); + for transaction in transactions { + let signature = self.send_transaction(trade_type, transaction).await?; + signatures.push(signature); + } + Ok(signatures) + } + + fn get_tip_account(&self) -> Result { + Ok("".to_string()) + } + + fn get_client_type(&self) -> ClientType { + ClientType::Rpc + } +} + +impl SolRpcClient { + pub fn new(rpc_client: Arc) -> Self { + Self { rpc_client } + } +} \ No newline at end of file diff --git a/src/swqos/types.rs b/src/swqos/types.rs new file mode 100755 index 0000000..e69de29 diff --git a/src/swqos/zeroslot.rs b/src/swqos/zeroslot.rs new file mode 100755 index 0000000..f98e0e5 --- /dev/null +++ b/src/swqos/zeroslot.rs @@ -0,0 +1,119 @@ +use crate::swqos::common::{poll_transaction_confirmation, serialize_transaction_and_encode}; +use rand::seq::IndexedRandom; +use reqwest::Client; +use serde_json::json; +use std::{sync::Arc, time::Instant}; + +use std::time::Duration; +use solana_transaction_status::UiTransactionEncoding; + +use anyhow::Result; +use solana_sdk::signature::Signature; +use solana_sdk::transaction::VersionedTransaction; +use crate::swqos::{ClientType, TradeType}; +use crate::swqos::SwqosClientTrait; + +use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::ZEROSLOT_TIP_ACCOUNTS}; + + +#[derive(Clone)] +pub struct ZeroSlotClient { + pub endpoint: String, + pub auth_token: String, + pub rpc_client: Arc, + pub http_client: Client, +} + +#[async_trait::async_trait] +impl SwqosClientTrait for ZeroSlotClient { + async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { + self.send_transaction(trade_type, transaction).await + } + + async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { + self.send_transactions(trade_type, transactions).await + } + + fn get_tip_account(&self) -> Result { + let tip_account = *ZEROSLOT_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| ZEROSLOT_TIP_ACCOUNTS.first()).unwrap(); + Ok(tip_account.to_string()) + } + + fn get_client_type(&self) -> ClientType { + ClientType::ZeroSlot + } +} + +impl ZeroSlotClient { + pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self { + let rpc_client = SolanaRpcClient::new(rpc_url); + let http_client = Client::builder() + .pool_idle_timeout(Duration::from_secs(60)) + .pool_max_idle_per_host(64) + .tcp_keepalive(Some(Duration::from_secs(1200))) + .http2_keep_alive_interval(Duration::from_secs(15)) + .timeout(Duration::from_secs(10)) + .connect_timeout(Duration::from_secs(5)) + .build() + .unwrap(); + Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client } + } + + pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result { + let start_time = Instant::now(); + let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?; + println!(" 交易编码base64: {:?}", start_time.elapsed()); + + let request_body = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "sendTransaction", + "params": [ + content, + { "encoding": "base64", "skipPreflight": true } + ] + }))?; + + let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20); + url.push_str(&self.endpoint); + url.push_str("/?api-key="); + url.push_str(&self.auth_token); + + // 4. 直接使用 `text().await?`，避免 `json().await?` 的异步 JSON 解析 + let response_text = self.http_client.post(&url) + .body(request_body) // 直接传字符串，避免 `json()` 开销 + .header("Content-Type", "application/json") // 显式指定 JSON 头 + .send() + .await? + .text() + .await?; + + // 5. 用 `serde_json::from_str()` 解析 JSON，减少 `.json().await?` 额外等待 + if let Ok(response_json) = serde_json::from_str::(&response_text) { + if response_json.get("result").is_some() { + println!(" 0slot{}提交: {:?}", trade_type, start_time.elapsed()); + } else if let Some(_error) = response_json.get("error") { + eprintln!(" 0slot{}提交失败: {:?}", trade_type, _error); + } + } + + let start_time: Instant = Instant::now(); + match poll_transaction_confirmation(&self.rpc_client, signature).await { + Ok(_) => (), + Err(_) => (), + } + + println!(" 0slot{}确认: {:?}", trade_type, start_time.elapsed()); + + Ok(signature) + } + + pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec) -> Result> { + let mut signatures = Vec::new(); + for transaction in transactions { + let signature = self.send_transaction(trade_type, transaction).await?; + signatures.push(signature); + } + Ok(signatures) + } +} \ No newline at end of file diff --git a/src/trading/core/executor.rs b/src/trading/core/executor.rs index 00f441b..c2130f6 100755 --- a/src/trading/core/executor.rs +++ b/src/trading/core/executor.rs @@ -107,7 +107,7 @@ impl TradeExecutor for GenericTradeExecutor { // 并行执行交易 parallel_execute_with_tips( - params.fee_clients, + params.swqos_clients, params.payer, instructions, params.priority_fee, @@ -180,7 +180,7 @@ impl TradeExecutor for GenericTradeExecutor { // 并行执行交易 parallel_execute_with_tips( - params.fee_clients, + params.swqos_clients, params.payer, instructions, params.priority_fee, diff --git a/src/trading/core/parallel.rs b/src/trading/core/parallel.rs index 3591587..dbd2bb6 100755 --- a/src/trading/core/parallel.rs +++ b/src/trading/core/parallel.rs @@ -6,7 +6,7 @@ use tokio::task::JoinHandle; use crate::{ common::PriorityFee, - swqos::{ClientType, FeeClient, TradeType}, + swqos::{ClientType, SwqosClient, TradeType}, trading::common::{ build_rpc_transaction, build_sell_tip_transaction_with_priority_fee, build_sell_transaction, build_tip_transaction_with_priority_fee, @@ -15,7 +15,7 @@ use crate::{ /// 并行执行交易的通用函数 pub async fn parallel_execute_with_tips( - fee_clients: Vec>, + swqos_clients: Vec>, payer: Arc, instructions: Vec, priority_fee: PriorityFee, @@ -27,8 +27,8 @@ pub async fn parallel_execute_with_tips( let cores = core_affinity::get_core_ids().unwrap(); let mut handles: Vec>> = vec![]; - for i in 0..fee_clients.len() { - let fee_client = fee_clients[i].clone(); + for i in 0..swqos_clients.len() { + let swqos_client = swqos_clients[i].clone(); let payer = payer.clone(); let instructions = instructions.clone(); let mut priority_fee = priority_fee.clone(); @@ -37,7 +37,7 @@ pub async fn parallel_execute_with_tips( let handle = tokio::spawn(async move { core_affinity::set_for_current(core_id); let transaction = if matches!(trade_type, TradeType::Sell) - && fee_client.get_client_type() == ClientType::Rpc + && swqos_client.get_client_type() == ClientType::Rpc { build_sell_transaction( payer, @@ -48,9 +48,9 @@ pub async fn parallel_execute_with_tips( ) .await? } else if matches!(trade_type, TradeType::Sell) - && fee_client.get_client_type() != ClientType::Rpc + && swqos_client.get_client_type() != ClientType::Rpc { - let tip_account = fee_client.get_tip_account()?; + let tip_account = swqos_client.get_tip_account()?; let tip_account = Arc::new(Pubkey::from_str(&tip_account).map_err(|e| anyhow!(e))?); build_sell_tip_transaction_with_priority_fee( payer, @@ -61,7 +61,7 @@ pub async fn parallel_execute_with_tips( recent_blockhash, ) .await? - } else if fee_client.get_client_type() == ClientType::Rpc { + } else if swqos_client.get_client_type() == ClientType::Rpc { build_rpc_transaction( payer, &priority_fee, @@ -72,7 +72,7 @@ pub async fn parallel_execute_with_tips( ) .await? } else { - let tip_account = fee_client.get_tip_account()?; + let tip_account = swqos_client.get_tip_account()?; let tip_account = Arc::new(Pubkey::from_str(&tip_account).map_err(|e| anyhow!(e))?); priority_fee.buy_tip_fee = priority_fee.buy_tip_fees[i]; @@ -88,7 +88,7 @@ pub async fn parallel_execute_with_tips( .await? }; - fee_client + swqos_client .send_transaction(trade_type, &transaction) .await?; Ok::<(), anyhow::Error>(()) diff --git a/src/trading/core/params.rs b/src/trading/core/params.rs index 4abd55f..9a67ea3 100755 --- a/src/trading/core/params.rs +++ b/src/trading/core/params.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use super::traits::ProtocolParams; use crate::common::{PriorityFee, SolanaRpcClient}; -use crate::swqos::FeeClient; +use crate::swqos::SwqosClient; use crate::accounts::BondingCurveAccount; /// 通用买入参数 @@ -27,7 +27,7 @@ pub struct BuyParams { #[derive(Clone)] pub struct BuyWithTipParams { pub rpc: Option>, - pub fee_clients: Vec>, + pub swqos_clients: Vec>, pub payer: Arc, pub mint: Pubkey, pub creator: Pubkey, @@ -59,7 +59,7 @@ pub struct SellParams { #[derive(Clone)] pub struct SellWithTipParams { pub rpc: Option>, - pub fee_clients: Vec>, + pub swqos_clients: Vec>, pub payer: Arc, pub mint: Pubkey, pub creator: Pubkey, @@ -124,10 +124,10 @@ impl ProtocolParams for PumpSwapParams { impl BuyParams { /// 转换为BuyWithTipParams - pub fn with_tip(self, fee_clients: Vec>) -> BuyWithTipParams { + pub fn with_tip(self, swqos_clients: Vec>) -> BuyWithTipParams { BuyWithTipParams { rpc: self.rpc, - fee_clients, + swqos_clients, payer: self.payer, mint: self.mint, creator: self.creator, @@ -144,10 +144,10 @@ impl BuyParams { impl SellParams { /// 转换为SellWithTipParams - pub fn with_tip(self, fee_clients: Vec>) -> SellWithTipParams { + pub fn with_tip(self, swqos_clients: Vec>) -> SellWithTipParams { SellWithTipParams { rpc: self.rpc, - fee_clients, + swqos_clients, payer: self.payer, mint: self.mint, creator: self.creator,