diff --git a/Cargo.toml b/Cargo.toml index 80f7507..848bc5f 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -47,4 +47,7 @@ regex = "1" tracing = "0.1.41" thiserror = "2.0.11" async-trait = "0.1.86" +lazy_static = "1.5.0" +once_cell = "1.20.3" + diff --git a/src/grpc/mod.rs b/src/grpc/mod.rs index 3910e7c..8148d6f 100755 --- a/src/grpc/mod.rs +++ b/src/grpc/mod.rs @@ -15,29 +15,16 @@ use solana_transaction_status::{ option_serializer::OptionSerializer, EncodedTransactionWithStatusMeta, UiTransactionEncoding, }; -use crate::common::logs_events::{PumpfunEvent, RaydiumEvent}; -use crate::common::logs_data::SwapBaseInLog; +use crate::common::logs_events::PumpfunEvent; use crate::error::{ClientError, ClientResult}; -// 类型别名定义 type TransactionsFilterMap = HashMap; -// 常量定义 -const AMM_V4: Pubkey = pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"); const PUMP_PROGRAM_ID: Pubkey = pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"); const CONNECT_TIMEOUT: u64 = 10; const REQUEST_TIMEOUT: u64 = 60; const CHANNEL_SIZE: usize = 1000; -// 枚举定义 -#[derive(Debug)] -pub enum SwapType { - Pump, - Raydium, -} - -// 结构体定义 -#[allow(dead_code)] pub struct TransactionPretty { pub slot: u64, pub signature: Signature, @@ -122,37 +109,6 @@ impl YellowstoneGrpc { Ok(client.subscribe_with_request(Some(subscribe_request)).await) } - pub async fn subscribe_accounts(&self, accounts: Vec) -> ClientResult<()> { - let transactions = self.get_subscribe_request_filter(accounts, vec![], vec![]); - let (mut subscribe_tx, mut stream) = self.connect(transactions).await? - .map_err(|e| ClientError::Other(format!("Failed to subscribe: {:?}", e)))?; - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); - - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Err(e) = Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await { - error!("Error handling message: {:?}", e); - break; - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); - - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = Self::process_transaction(transaction_pretty, &|_| {}, None).await { - error!("Error processing transaction: {:?}", e); - } - } - Ok(()) - } - pub async fn subscribe_pumpfun(&self, callback: F, bot_wallet: Option) -> ClientResult<()> where F: Fn(PumpfunEvent) + Send + Sync + 'static, @@ -211,7 +167,6 @@ impl YellowstoneGrpc { transactions } - async fn handle_stream_message( msg: SubscribeUpdate, tx: &mut mpsc::Sender, @@ -258,78 +213,21 @@ impl YellowstoneGrpc { &vec![] }; - if let Ok(swap_type) = Self::get_swap_type(&trade_raw) { - match swap_type { - SwapType::Raydium => { - let event = RaydiumEvent::parse_logs::(logs); - info!("RaydiumEvent {:#?}", event); - } - SwapType::Pump => { - let (create_event, trade_event) = PumpfunEvent::parse_logs(logs); - if let Some(create_event) = create_event { - callback(PumpfunEvent::NewToken(create_event)); - } - if let Some(trade_event) = trade_event { - if let Some(bot_wallet_pubkey) = bot_wallet { - if trade_event.user == bot_wallet_pubkey { - callback(PumpfunEvent::NewBotTrade(trade_event)); - } else { - callback(PumpfunEvent::NewUserTrade(trade_event)); - } - } else { - callback(PumpfunEvent::NewUserTrade(trade_event)); - } - } + let (create_event, trade_event) = PumpfunEvent::parse_logs(logs); + if let Some(create_event) = create_event { + callback(PumpfunEvent::NewToken(create_event)); + } + if let Some(trade_event) = trade_event { + if let Some(bot_wallet_pubkey) = bot_wallet { + if trade_event.user == bot_wallet_pubkey { + callback(PumpfunEvent::NewBotTrade(trade_event)); + } else { + callback(PumpfunEvent::NewUserTrade(trade_event)); } + } else { + callback(PumpfunEvent::NewUserTrade(trade_event)); } } Ok(()) } - - pub fn get_swap_type(trade_raw: &EncodedTransactionWithStatusMeta) -> ClientResult { - let transaction = trade_raw.transaction.decode() - .ok_or_else(|| ClientError::Other("Failed to decode transaction".to_string()))?; - - let account_keys = transaction.message.static_account_keys(); - let program_index = account_keys - .iter() - .position(|item| item == &AMM_V4 || item == &PUMP_PROGRAM_ID) - .ok_or_else(|| ClientError::Other("swap type program_id not found".to_string()))?; - - match account_keys[program_index] { - AMM_V4 => Ok(SwapType::Raydium), - PUMP_PROGRAM_ID => Ok(SwapType::Pump), - _ => Err(ClientError::Other("Invalid program_id".to_string())) - } - } -} - -async fn test_subscribe_pumpfun() -> ClientResult<()> { - // 创建YellowstoneGrpc实例 - let endpoint = "https://grpc.mainnet.solana.com".to_string(); - let client = YellowstoneGrpc::new(endpoint); - - // 定义回调函数 - let callback = |event: PumpfunEvent| { - match event { - PumpfunEvent::NewToken(token_info) => { - println!("收到新代币事件: {:?}", token_info); - }, - PumpfunEvent::NewUserTrade(trade_info) => { - println!("收到用户交易事件: {:?}", trade_info); - }, - PumpfunEvent::NewBotTrade(trade_info) => { - println!("收到机器人交易事件: {:?}", trade_info); - }, - PumpfunEvent::Error(err) => { - println!("收到错误: {}", err); - } - } - }; - - // 订阅事件 - let bot_wallet = None; // 可以设置为Some(bot_pubkey)来区分机器人交易 - client.subscribe_pumpfun(callback, bot_wallet).await?; - - Ok(()) } diff --git a/src/jito/mod.rs b/src/jito/mod.rs index d4143ef..7151c10 100755 --- a/src/jito/mod.rs +++ b/src/jito/mod.rs @@ -1,10 +1,8 @@ -use std::{str::FromStr, time::Duration, fmt}; +use std::str::FromStr; use anyhow::{anyhow, Result}; use api::TipAccountResult; -use rand::{seq::IteratorRandom}; -use serde::Deserialize; -use serde_json::{json, Value}; +use rand::seq::IteratorRandom; use solana_sdk::{ pubkey::Pubkey, transaction::{Transaction, VersionedTransaction}, @@ -24,8 +22,6 @@ use crate::jito::rpc_client::RpcClient; pub struct JitoClient { base_url: String, tip_accounts: RwLock>, - tips_percentile: RwLock>, - tip_percentile: String, client: RpcClient, } @@ -34,49 +30,29 @@ impl Clone for JitoClient { Self { base_url: self.base_url.clone(), tip_accounts: RwLock::new(Vec::new()), - tips_percentile: RwLock::new(None), - tip_percentile: self.tip_percentile.clone(), client: RpcClient::new(self.base_url.clone()), } } } -const DEFAULT_TIP_PERCENTILE: &str = "25"; -pub struct TipPercentileData { - pub time: String, - pub landed_tips_25th_percentile: f64, - pub landed_tips_50th_percentile: f64, - pub landed_tips_75th_percentile: f64, - pub landed_tips_95th_percentile: f64, - pub landed_tips_99th_percentile: f64, - pub ema_landed_tips_50th_percentile: f64, -} - impl JitoClient { - pub fn new(jito_url: &str, uuid: Option) -> Self { + pub fn new(jito_url: &str, _uuid: Option) -> Self { Self { base_url: jito_url.to_string(), tip_accounts: RwLock::new(vec![]), - tips_percentile: RwLock::new(None), - tip_percentile: DEFAULT_TIP_PERCENTILE.to_string(), client: RpcClient::new(jito_url.to_string()), } } pub async fn get_tip_accounts(&self) -> Result { let result = self.client.get_tip_accounts().await?; - let tip_accounts = TipAccountResult::from(result).map_err(|e| anyhow!(e))?; - Ok(tip_accounts) + TipAccountResult::from(result).map_err(|e| anyhow!(e)) } pub async fn init_tip_accounts(&self) -> Result<()> { let accounts = self.get_tip_accounts().await?; let mut tip_accounts = self.tip_accounts.write().await; - - accounts - .accounts - .iter() - .for_each(|account| tip_accounts.push(account.to_string())); + *tip_accounts = accounts.accounts.iter().map(|a| a.to_string()).collect(); Ok(()) } @@ -85,9 +61,11 @@ impl JitoClient { let accounts = self.tip_accounts.read().await; if !accounts.is_empty() { if let Some(acc) = accounts.iter().choose(&mut rand::thread_rng()) { - return Ok(Pubkey::from_str(acc).inspect_err(|err| { - error!("jito: failed to parse Pubkey: {:?}", err); - })?); + return Pubkey::from_str(acc) + .map_err(|err| { + error!("jito: failed to parse Pubkey: {:?}", err); + anyhow!("Invalid pubkey format") + }); } } } @@ -95,12 +73,16 @@ impl JitoClient { self.init_tip_accounts().await?; let accounts = self.tip_accounts.read().await; - match accounts.iter().choose(&mut rand::thread_rng()) { - Some(acc) => Ok(Pubkey::from_str(acc).inspect_err(|err| { - error!("jito: failed to parse Pubkey: {:?}", err); - })?), - None => Err(anyhow!("jito: no tip accounts available")), - } + accounts + .iter() + .choose(&mut rand::rng()) + .ok_or_else(|| anyhow!("jito: no tip accounts available")) + .and_then(|acc| { + Pubkey::from_str(acc).map_err(|err| { + error!("jito: failed to parse Pubkey: {:?}", err); + anyhow!("Invalid pubkey format") + }) + }) } pub async fn send_transaction( @@ -111,48 +93,3 @@ impl JitoClient { Ok(self.client.send_bundle(&bundles).await?) } } - -#[cfg(test)] -mod tests { - use std::time::Duration; - use serde_json::{json, Value}; - use super::*; - - fn generate_statuses(bundle_id: String, confirmation_status: &str) -> Vec { - vec![json!({ - "bundle_id": bundle_id, - "transactions": ["tx1", "tx2"], - "slot": 12345, - "confirmation_status": confirmation_status, - "err": {"Ok": null} - })] - } - - #[tokio::test] - async fn test_success_confirmation() { - let client = JitoClient::new("http://localhost:8899", None); - for &status in &["finalized", "confirmed"] { - let wait_result = client.wait_for_bundle_confirmation( - |id| async { Ok(generate_statuses(id, status)) }, - "6e4b90284778a40633b56e4289202ea79e62d2296bb3d45398bb93f6c9ec083d".to_string(), - Duration::from_secs(1), - Duration::from_secs(1), - ) - .await; - assert!(wait_result.is_ok()); - } - } - - #[tokio::test] - async fn test_error_confirmation() { - let client = JitoClient::new("http://localhost:8899", None); - let wait_result = client.wait_for_bundle_confirmation( - |id| async { Ok(generate_statuses(id, "processed")) }, - "6e4b90284778a40633b56e4289202ea79e62d2296bb3d45398bb93f6c9ec083d".to_string(), - Duration::from_secs(1), - Duration::from_secs(2), - ) - .await; - assert!(wait_result.is_err()); - } -} diff --git a/src/jito/rpc_client.rs b/src/jito/rpc_client.rs index 227b56b..a43c170 100755 --- a/src/jito/rpc_client.rs +++ b/src/jito/rpc_client.rs @@ -22,33 +22,13 @@ use crate::jito::{ }; pub type RpcResult = client_error::Result>; -/// A client of a remote Jito block engine node. -/// -/// `RpcClient` communicates with a block engine node over [JSON-RPC] -/// It is the primary Rust interface for querying and transacting with the network -/// from external programs. -/// -/// This is modeled very closely with the solan RpcClient with similar error types. -/// You can treat the client similar to the Solana RpcClient with the difference being -/// the RpcClient supports the block engine apis. -/// -/// The client can be used with jito block engine proxy server for authentication -/// The client can be used as is to send requests unauthenticated to the jito block engine as well -/// -/// Please note that the commitment level is not used at the moment. Support will be added -/// later on to specify and use the commitment levels. + pub struct RpcClient { sender: Box, config: RpcClientConfig, } impl RpcClient { - /// Create an `RpcClient` from an [`RpcSender`] and an [`RpcClientConfig`]. - /// - /// This is the basic constructor, allowing construction with any type of - /// `RpcSender`. Most applications should use one of the other constructors, - /// such as [`RpcClient::new`], [`RpcClient::new_with_commitment`] or - /// [`RpcClient::new_with_timeout`]. pub fn new_sender( sender: T, config: RpcClientConfig, @@ -59,46 +39,10 @@ impl RpcClient { } } - /// Create an HTTP `RpcClient`. - /// - /// The URL is an HTTP URL, usually for port 8899, as in - /// "http://localhost:8899". - /// - /// The client has a default timeout of 30 seconds, and a default [commitment - /// level][cl] of [`Finalized`](CommitmentLevel::Finalized). - /// - /// [cl]: https://docs.solana.com/developing/clients/jsonrpc-api#configuring-state-commitment - /// - /// # Examples - /// - /// ``` - /// # use jito_block_engine_json_rpc_client::jsonrpc_client::rpc_client::RpcClient; - /// let url = "http://localhost:8899".to_string(); - /// let client = RpcClient::new(url); - /// ``` pub fn new(url: String) -> Self { Self::new_with_commitment(url, CommitmentConfig::default()) } - /// Create an HTTP `RpcClient` with specified [commitment level][cl]. - /// - /// Please note the client is not currently implemented to support commitment level configs - /// - /// [cl]: https://docs.solana.com/developing/clients/jsonrpc-api#configuring-state-commitment - /// - /// The URL is an HTTP URL, usually for port 8899, as in - /// "http://localhost:8899". - /// - /// The client has a default timeout of 30 seconds, and a user-specified - /// [`CommitmentLevel`] via [`CommitmentConfig`]. - /// - /// # Examples - /// - /// # use solana_sdk::commitment_config::CommitmentConfig; - /// # use jito_block_engine_json_rpc_client::jsonrpc_client::rpc_client::RpcClient; - /// let url = "http://localhost:8899".to_string(); - /// let commitment_config = CommitmentConfig::processed(); - /// let client = RpcClient::new_with_commitment(url, commitment_config); fn new_with_commitment(url: String, commitment_config: CommitmentConfig) -> Self { Self::new_sender( HttpSender::new(url), @@ -106,25 +50,6 @@ impl RpcClient { ) } - /// Create an HTTP `RpcClient` with specified timeout. - /// - /// The URL is an HTTP URL, usually for port 8899, as in - /// "http://localhost:8899". - /// - /// The client has and a default [commitment level][cl] of - /// [`Finalized`](CommitmentLevel::Finalized). - /// - /// [cl]: https://docs.solana.com/developing/clients/jsonrpc-api#configuring-state-commitment - /// - /// # Examples - /// - /// ``` - /// # use std::time::Duration; - /// # use jito_block_engine_json_rpc_client::jsonrpc_client::rpc_client::RpcClient; - /// let url = "http://localhost::8899".to_string(); - /// let timeout = Duration::from_secs(1); - /// let client = RpcClient::new_with_timeout(url, timeout); - /// ``` pub fn new_with_timeout(url: String, timeout: Duration) -> Self { Self::new_sender( HttpSender::new_with_timeout(url, timeout), @@ -132,60 +57,14 @@ impl RpcClient { ) } - /// Get the configured url of the client's sender pub fn url(&self) -> String { self.sender.url() } - /// Get the configured default [commitment level][cl]. - /// - /// [cl]: https://docs.solana.com/developing/clients/jsonrpc-api#configuring-state-commitment - /// - /// The commitment config may be specified during construction, and - /// determines how thoroughly committed a transaction must be when waiting - /// for its confirmation or otherwise checking for confirmation. If not - /// specified, the default commitment level is - /// [`Finalized`](CommitmentLevel::Finalized). - /// - /// The default commitment level is overridden when calling methods that - /// explicitly provide a [`CommitmentConfig`], like - /// [`RpcClient::confirm_transaction_with_commitment`]. pub fn commitment(&self) -> CommitmentConfig { self.config.commitment_config } - /// Submits a bundle of signed transactions to the network. - /// - /// This returns a bundle_id on success and will return an error - /// code on failure. The error code will only be on the basic - /// validation of the bundle and publishing the bundle to the mempool - /// For the bundle status regarding whether it landed or not, get_bundle_statuses - /// should be used with the bundle id. - /// - /// Example excerpt can be as below - /// - /// use jito_block_engine_json_rpc_client::jsonrpc_client::rpc_client::RpcClient; - /// use solana_program::hash::Hash; - /// use solana_sdk::{pubkey::Pubkey, signer::keypair::Keypair}; - /// - /// let base = 0; - /// let MAX_BUNDLE_LEN = 5; - /// let searcher_keypair = Keypair::new(); - /// let recent_blockhash = Hash::new_unique(); - /// - /// let mut bundle: Vec<_> = (0..(MAX_BUNDLE_LEN) as u64) - /// .map(|amount| { - /// VersionedTransaction::from(system_transaction::transfer( - /// &searcher_keypair, - /// &searcher_keypair.pubkey(), - /// base + amount, - /// recent_blockhash, - /// )) - /// }) - /// .collect(); - /// - /// let rpc_client = RpcClient::new(SERVER_URL.to_owned()); - /// let response = rpc_client.send_bundle(&bundle).await; pub async fn send_bundle( &self, transactions: &[impl SerializableTransaction], @@ -195,9 +74,6 @@ impl RpcClient { let encoding = self.default_cluster_transaction_encoding().await?; serialized_encoded.push(serialize_and_encode(transaction, encoding)?); } - //The bundle may or may - // not have been submitted to the cluster, so callers should verify the success of - // the correct transaction signature independently. match self .send(RpcRequest::SendBundle, json!([serialized_encoded])) .await @@ -221,35 +97,6 @@ impl RpcClient { Ok(UiTransactionEncoding::Base58) } - /// Gets the statuses of a list of bundle ids. - /// - /// Returns the statuses of a list of signatures. Each signature must be a bundle_id. - /// bundle ids are sha256 hashes of their tx signatures (we get it after a sendBundle) - /// This method currently will provide information regarding whether the bundle - /// landed or not. - /// The behavior is similar to the solana rpc method getSignatureStatuses - /// https://docs.solana.com/api/http#getsignaturestatuses - /// - /// If the bundle_id is not found or the all of the transactions in the bundle has not landed, - /// we return null. If found and landed, we return the context information including the slot - /// at which the request was made and result with the bundle_id(s) and the transactions with the - /// slot and confirmation status. At this point, its assumed that all transactions within a bundle - /// will have the same slot number and confirmation status. - /// - /// The confirmation status of a bundle is the confirmation status of the transactions. - /// This api does not provide a commitment level to configure, but will return the commitment level - /// as returned by the rpc. The rpc used to fetch bulk transaction status does not provide a commitment - /// level configuration option either. - /// - /// Example excerpt can be as below - /// - /// use jito_block_engine_json_rpc_client::jsonrpc_client::rpc_client::RpcClient; - /// - /// let SERVER_URL = "http://localhost:8899"; - /// let bundle_id = "bundle_id".to_owned(); - /// - /// let rpc_client = RpcClient::new(SERVER_URL.to_owned()); - /// let response = rpc_client.get_bundle_statuses(&[bundle_id.clone()]).await; pub async fn get_bundle_statuses( &self, signatures: &[String], @@ -258,7 +105,6 @@ impl RpcClient { .await } - /// Returns the tip accounts to be used for tip payments. pub async fn get_tip_accounts(&self) -> ClientResult> { self.send(RpcRequest::GetTipAccounts, Value::Null).await } @@ -301,7 +147,6 @@ where Ok(encoded) } -// Sample tests. Can use these as a reference point on how to use the api and what to expect #[cfg(test)] mod rpc_client_tests { use solana_program::hash::Hash; @@ -310,41 +155,22 @@ mod rpc_client_tests { transaction::VersionedTransaction, }; - use crate::jsonrpc_client::rpc_client::RpcClient; + use crate::jito::rpc_client::RpcClient; - // Use the proxy server url here. const SERVER_URL: &str = "http://0.0.0.0:8080/api/v1/bundles"; #[tokio::test] pub async fn get_tip_accounts() { - // Let's try the same with the rpc client let rpc_client = RpcClient::new(SERVER_URL.to_owned()); - let tip_accounts = rpc_client.get_tip_accounts().await; - - // Sample output. Pick only randomly to not have contention - // ["9ttgPBBhRYFuQccdR1DSnb7hydsWANoDsV3P9kaGMCEh", - // "EoW3SUQap7ZeynXQ2QJ847aerhxbPVr843uMeTfc9dxM", - // "4xgEmT58RwTNsF5xm2RMYCnR1EVukdK8a1i2qFjnJFu3", - // "B1mrQSpdeMU9gCvkJ6VsXVVoYjRGkNA7TtjMyqxrhecH", - // "aTtUk2DHgLhKZRDjePq6eiHRKC1XXFMBiSUfQ2JNDbN", - // "9n3d1K5YD2vECAbRFhFFGYNNjiXtHXJWn9F31t89vsAV", - // "ARTtviJkLLt6cHGQDydfo1Wyk6M4VGZdKZ2ZhdnJL336", - // "E2eSqe33tuhAHKTrwky5uEjaVqnb2T9ns6nHHUrN8588"] println!("{:?}", tip_accounts); } #[tokio::test] pub async fn send_bundle() { let rpc_client = RpcClient::new(SERVER_URL.to_owned()); - - // Use your own keypair to sign let signer_keypair = Keypair::new(); - - // Get the latest blockhash from solana cluster. Can use https://docs.solana.com/api/http#getlatestblockhash let recent_blockhash = Hash::new_unique(); - - // Use the get_tip_accounts to randomly select a tip account to send tips to let tip_account = Pubkey::try_from("DCN82qDxJAQuSqHhv2BJuAgi41SPeKZB5ioBCTMNDrCC").unwrap(); let mut bundle: Vec<_> = vec![VersionedTransaction::from(system_transaction::transfer( @@ -354,7 +180,6 @@ mod rpc_client_tests { recent_blockhash, ))]; - // Add the tip bundle.push(VersionedTransaction::from(system_transaction::transfer( &signer_keypair, &tip_account, @@ -362,45 +187,15 @@ mod rpc_client_tests { recent_blockhash, ))); let response = rpc_client.send_bundle(&bundle).await; - - // If successful, the bundle_id can be retrieved. Else, an error code will be provided println!("{:?}", response); } #[tokio::test] pub async fn get_bundle_statuses() { let rpc_client = RpcClient::new(SERVER_URL.to_owned()); - - // Use the bundle id you got from send_bundle let bundle_id = "6e4b90284778a40633b56e4289202ea79e62d2296bb3d45398bb93f6c9ec083d".to_owned(); - let response = rpc_client.get_bundle_statuses(&[bundle_id]).await; - - // Sample success output: - // Response { - // context: RpcResponseContext { - // slot: 0, api_version: None }, - // value: [Object { - // "bundle_id": String("6e4b90284778a40633b56e4289202ea79e62d2296bb3d45398bb93f6c9ec083d"), - // "transactions": Array [String("4DGCuaKc2oue4Z8YC6mBwyg3oPAFG64BfxDtMbqDU3Du9zr26oVSuZcjSnJqTnHnKYFJ4AdPuq5kUrWKwTFLKtW6"), - // String("srrgfKABYeaKazZjBmpuPKySJ8qgqezYaCdDnB9nhED5CFhviZ1wgcs5vEKnAK9L2ytRauWG9czGoKRxajpZ1YR")], - // "slot": Number(240632575), - // "confirmation_status": String("finalized"), - // "err": Object {"Ok": Null}}] } - // - // Sample retryable error output: - // Response { - // context: RpcResponseContext { - // slot: 0, api_version: None }, - // value: [Object { - // "bundle_id": String("6e4b90284778a40633b56e4289202ea79e62d2296bb3d45398bb93f6c9ec083d"), - // "transactions": Array [String("4DGCuaKc2oue4Z8YC6mBwyg3oPAFG64BfxDtMbqDU3Du9zr26oVSuZcjSnJqTnHnKYFJ4AdPuq5kUrWKwTFLKtW6"), - // String("srrgfKABYeaKazZjBmpuPKySJ8qgqezYaCdDnB9nhED5CFhviZ1wgcs5vEKnAK9L2ytRauWG9czGoKRxajpZ1YR")], - // "slot": Number(612529), - // "confirmation_status": Null, - // "err": Object {"Err": Object {"Retryable": String("Failed to retrieve information from solana cluster")}}}] } - // If unknown bundle, the response would be null println!("{:?}", response); } } diff --git a/src/lib.rs b/src/lib.rs index 607751d..9416855 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,16 +8,12 @@ pub mod grpc; pub mod common; use anyhow::anyhow; -use solana_client::{connection_cache::ConnectionCache, rpc_client::RpcClient, rpc_config::{RpcSendTransactionConfig, RpcSimulateTransactionConfig}, send_and_confirm_transactions_in_parallel::{send_and_confirm_transactions_in_parallel, SendAndConfirmConfig}, tpu_client::{TpuClient, TpuClientConfig}}; +use solana_client::{ + rpc_client::RpcClient, + rpc_config::RpcSimulateTransactionConfig +}; use solana_sdk::{ - commitment_config::CommitmentConfig, - pubkey::Pubkey, - signature::{Keypair, Signature}, - signer::Signer, - instruction::Instruction, - system_instruction, - compute_budget::ComputeBudgetInstruction, - transaction::Transaction, + commitment_config::CommitmentConfig, compute_budget::ComputeBudgetInstruction, instruction::Instruction, native_token::sol_to_lamports, pubkey::Pubkey, signature::{Keypair, Signature}, signer::Signer, system_instruction, transaction::Transaction }; use spl_associated_token_account::{ get_associated_token_address, @@ -30,6 +26,8 @@ use spl_token::instruction::close_account; use std::sync::Arc; use std::time::Instant; +use std::collections::HashMap; +use tokio::sync::RwLock; use crate::jito::JitoClient; @@ -39,10 +37,14 @@ use borsh::BorshDeserialize; const DEFAULT_SLIPPAGE: u64 = 1000; // 10% const DEFAULT_COMPUTE_UNIT_LIMIT: u32 = 78000; const DEFAULT_COMPUTE_UNIT_PRICE: u64 = 500000; -const JITO_TIP_AMOUNT: u64 = 5644005; -// const WS_URL: &str = "ws://127.0.0.1:8900"; +const JITO_TIP_AMOUNT: f64 = 0.00006; + +// Cache +lazy_static::lazy_static! { + static ref ACCOUNT_CACHE: RwLock>> = RwLock::new(HashMap::new()); + static ref BONDING_CURVE_CACHE: RwLock>> = RwLock::new(HashMap::new()); +} -/// Priority fee configuration #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct PriorityFee { pub limit: Option, @@ -75,7 +77,7 @@ impl Clone for PumpFun { } impl PumpFun { - /// Create a new PumpFun client instance + #[inline] pub fn new( rpc_url: String, commitment: Option, @@ -96,10 +98,6 @@ impl PumpFun { } } - pub fn get_rpc(&self) -> &RpcClient { - &self.rpc - } - /// Create a new token pub async fn create( &self, @@ -109,7 +107,7 @@ impl PumpFun { ) -> Result { let ipfs = utils::create_token_metadata(metadata) .await - .map_err(|_| anyhow!("Failed to upload metadata"))?; + .map_err(|e| anyhow!("Failed to upload metadata: {}", e))?; let mut instructions = self.create_priority_fee_instructions(priority_fee); @@ -145,11 +143,15 @@ impl PumpFun { slippage_basis_points: Option, priority_fee: Option, ) -> Result { + if amount_sol == 0 { + return Err(anyhow!("Amount cannot be zero")); + } + let ipfs = utils::create_token_metadata(metadata) .await - .map_err(|e| anyhow!(e.to_string()))?; + .map_err(|e| anyhow!("Failed to upload metadata: {}", e))?; - let global_account = self.get_global_account()?; + let global_account = self.get_global_account().await?; let buy_amount = global_account.get_initial_buy_price(amount_sol); let buy_amount_with_slippage = utils::calculate_with_slippage_buy(amount_sol, slippage_basis_points.unwrap_or(DEFAULT_SLIPPAGE)); @@ -207,8 +209,12 @@ impl PumpFun { slippage_basis_points: Option, priority_fee: Option, ) -> Result { - let global_account = self.get_global_account()?; - let bonding_curve_account = self.get_bonding_curve_account(mint)?; + if amount_sol == 0 { + return Err(anyhow!("Amount cannot be zero")); + } + + let global_account = self.get_global_account().await?; + let bonding_curve_account = self.get_bonding_curve_account(mint).await?; let buy_amount = bonding_curve_account .get_buy_price(amount_sol) .map_err(|e| anyhow!(e))?; @@ -256,19 +262,23 @@ impl PumpFun { buy_token_amount: u64, max_sol_cost: u64, slippage_basis_points: Option, - jito_fee: Option, + jito_fee: Option, ) -> Result { + if buy_token_amount == 0 || max_sol_cost == 0 { + return Err(anyhow!("Amount cannot be zero")); + } + let start_time = Instant::now(); let jito_client = self.jito_client.as_ref() .ok_or_else(|| anyhow!("Jito client not found"))?; - let global_account = self.get_global_account()?; + let global_account = self.get_global_account().await?; let buy_amount_with_slippage = utils::calculate_with_slippage_buy(max_sol_cost, slippage_basis_points.unwrap_or(DEFAULT_SLIPPAGE)); let mut instructions = self.create_priority_fee_instructions(None); - let tip_account = jito_client.get_tip_account().await.map_err(|e| anyhow!(e)).unwrap(); + let tip_account = jito_client.get_tip_account().await.map_err(|e| anyhow!(e))?; let ata = get_associated_token_address(&self.payer.pubkey(), mint); if self.rpc.get_account(&ata).is_err() { instructions.push(create_associated_token_account( @@ -294,7 +304,7 @@ impl PumpFun { system_instruction::transfer( &self.payer.pubkey(), &tip_account, - jito_fee, + sol_to_lamports(jito_fee), ), ); @@ -320,7 +330,6 @@ impl PumpFun { slippage_basis_points: Option, priority_fee: Option, ) -> Result<(), anyhow::Error> { - // 获取代币账户余额 let ata = get_associated_token_address(&self.payer.pubkey(), mint); let balance = self.rpc.get_token_account_balance(&ata)?; let balance_u64 = balance.amount.parse::() @@ -328,12 +337,11 @@ impl PumpFun { let amount = amount_token.unwrap_or(balance_u64); if amount == 0 { - return Err(anyhow!("Balance is 0")); + return Err(anyhow!("Amount cannot be zero")); } - // 计算最小SOL输出 - let global_account = self.get_global_account()?; - let bonding_curve_account = self.get_bonding_curve_account(mint)?; + let global_account = self.get_global_account().await?; + let bonding_curve_account = self.get_bonding_curve_account(mint).await?; let min_sol_output = bonding_curve_account .get_sell_price(amount, global_account.fee_basis_points) .map_err(|e| anyhow!(e))?; @@ -342,7 +350,6 @@ impl PumpFun { slippage_basis_points.unwrap_or(DEFAULT_SLIPPAGE), ); - // 构建指令 let mut instructions = vec![ ComputeBudgetInstruction::set_compute_unit_limit(1_400_000), ComputeBudgetInstruction::set_compute_unit_price(0), @@ -364,12 +371,10 @@ impl PumpFun { &self.payer.pubkey(), &self.payer.pubkey(), &[&self.payer.pubkey()], - ).unwrap()); + )?); - // 获取最新区块哈希 let commitment_config = CommitmentConfig::confirmed(); - let recent_blockhash = self.rpc.get_latest_blockhash_with_commitment(commitment_config) - .map_err(|_| anyhow!("Failed to get latest blockhash"))? + let recent_blockhash = self.rpc.get_latest_blockhash_with_commitment(commitment_config)? .0; let simulate_tx = Transaction::new_signed_with_payer( @@ -379,7 +384,6 @@ impl PumpFun { recent_blockhash, ); - // 模拟交易 let config = RpcSimulateTransactionConfig { sig_verify: true, commitment: Some(commitment_config), @@ -393,12 +397,15 @@ impl PumpFun { return Err(anyhow!("Simulation failed: {:?}", result.err)); } - // 更新计算单元和优先费用 let result_cu = result.units_consumed.ok_or_else(|| anyhow!("No compute units consumed"))?; let fees = self.rpc.get_recent_prioritization_fees(&[])?; - let average_fees = fees.iter() - .map(|fee| fee.prioritization_fee) - .sum::() / fees.len() as u64; + let average_fees = if fees.is_empty() { + DEFAULT_COMPUTE_UNIT_PRICE + } else { + fees.iter() + .map(|fee| fee.prioritization_fee) + .sum::() / fees.len() as u64 + }; let unit_price = match priority_fee { None => average_fees, @@ -417,7 +424,6 @@ impl PumpFun { recent_blockhash, ); - // 发送交易 self.rpc.send_and_confirm_transaction(&transaction)?; Ok(()) } @@ -430,8 +436,8 @@ impl PumpFun { slippage_basis_points: Option, priority_fee: Option, ) -> Result<(), anyhow::Error> { - if percent > 100 { - return Err(anyhow!("Percentage must be between 0 and 100")); + if percent == 0 || percent > 100 { + return Err(anyhow!("Percentage must be between 1 and 100")); } let ata = get_associated_token_address(&self.payer.pubkey(), mint); @@ -452,10 +458,10 @@ impl PumpFun { mint: &Pubkey, percent: u64, slippage_basis_points: Option, - jito_fee: Option, + jito_fee: Option, ) -> Result { - if percent > 100 { - return Err(anyhow!("Percentage must be between 0 and 100")); + if percent == 0 || percent > 100 { + return Err(anyhow!("Percentage must be between 1 and 100")); } let ata = get_associated_token_address(&self.payer.pubkey(), mint); @@ -477,7 +483,7 @@ impl PumpFun { mint: &Pubkey, amount_token: Option, slippage_basis_points: Option, - jito_fee: Option, + jito_fee: Option, ) -> Result { let start_time = Instant::now(); @@ -494,8 +500,8 @@ impl PumpFun { return Err(anyhow!("Amount cannot be zero")); } - let global_account = self.get_global_account()?; - let bonding_curve_account = self.get_bonding_curve_account(mint)?; + let global_account = self.get_global_account().await?; + let bonding_curve_account = self.get_bonding_curve_account(mint).await?; let min_sol_output = bonding_curve_account .get_sell_price(amount, global_account.fee_basis_points) .map_err(|e| anyhow!(e))?; @@ -522,13 +528,14 @@ impl PumpFun { &self.payer.pubkey(), &self.payer.pubkey(), &[&self.payer.pubkey()], - ).unwrap()); + )?); + let jito_fee = jito_fee.unwrap_or(JITO_TIP_AMOUNT); instructions.push( system_instruction::transfer( &self.payer.pubkey(), &tip_account, - jito_fee.unwrap_or(JITO_TIP_AMOUNT/10), + sol_to_lamports(jito_fee), ), ); @@ -546,9 +553,40 @@ impl PumpFun { Ok(signature) } + pub async fn transfer_sol(&self, receive_wallet: &Pubkey, amount: u64) -> Result<(), anyhow::Error> { + if amount == 0 { + return Err(anyhow!("Amount cannot be zero")); + } + + let balance = self.get_payer_sol_balance()?; + if balance < amount { + return Err(anyhow!("Insufficient balance")); + } + + let transfer_instruction = system_instruction::transfer( + &self.payer.pubkey(), + receive_wallet, + amount, + ); + + let recent_blockhash = self.rpc.get_latest_blockhash()?; + + let transaction = Transaction::new_signed_with_payer( + &[transfer_instruction], + Some(&self.payer.pubkey()), + &[&self.payer.clone()], + recent_blockhash, + ); + + self.rpc.send_and_confirm_transaction(&transaction)?; + + Ok(()) + } + // Helper methods + #[inline] fn create_priority_fee_instructions(&self, priority_fee: Option) -> Vec { - let mut instructions = Vec::new(); + let mut instructions = Vec::with_capacity(2); let fee = priority_fee.unwrap_or(PriorityFee::default()); if let Some(limit) = fee.limit { instructions.push(ComputeBudgetInstruction::set_compute_unit_limit(limit)); @@ -560,11 +598,17 @@ impl PumpFun { instructions } - // Public interface methods + #[inline] + pub fn get_rpc(&self) -> &RpcClient { + &self.rpc + } + + #[inline] pub fn get_payer_pubkey(&self) -> Pubkey { self.payer.pubkey() } + #[inline] pub fn get_token_balance(&self, account: &Pubkey, mint: &Pubkey) -> Result { let ata = get_associated_token_address(account, mint); if self.rpc.get_account(&ata).is_err() { @@ -576,27 +620,38 @@ impl PumpFun { .map_err(|_| anyhow!("Failed to parse token balance")) } + #[inline] pub fn get_sol_balance(&self, account: &Pubkey) -> Result { self.rpc.get_balance(account).map_err(|_| anyhow!("Failed to get SOL balance")) } + #[inline] pub fn get_payer_token_balance(&self, mint: &Pubkey) -> Result { self.get_token_balance(&self.payer.pubkey(), mint) } + #[inline] pub fn get_payer_sol_balance(&self) -> Result { self.get_sol_balance(&self.payer.pubkey()) } - // PDA related methods + #[inline] pub fn get_global_pda() -> Pubkey { - Pubkey::find_program_address(&[constants::seeds::GLOBAL_SEED], &constants::accounts::PUMPFUN).0 + static GLOBAL_PDA: once_cell::sync::Lazy = once_cell::sync::Lazy::new(|| { + Pubkey::find_program_address(&[constants::seeds::GLOBAL_SEED], &constants::accounts::PUMPFUN).0 + }); + *GLOBAL_PDA } + #[inline] pub fn get_mint_authority_pda() -> Pubkey { - Pubkey::find_program_address(&[constants::seeds::MINT_AUTHORITY_SEED], &constants::accounts::PUMPFUN).0 + static MINT_AUTHORITY_PDA: once_cell::sync::Lazy = once_cell::sync::Lazy::new(|| { + Pubkey::find_program_address(&[constants::seeds::MINT_AUTHORITY_SEED], &constants::accounts::PUMPFUN).0 + }); + *MINT_AUTHORITY_PDA } + #[inline] pub fn get_bonding_curve_pda(mint: &Pubkey) -> Option { Pubkey::try_find_program_address( &[constants::seeds::BONDING_CURVE_SEED, mint.as_ref()], @@ -604,6 +659,7 @@ impl PumpFun { ).map(|(pubkey, _)| pubkey) } + #[inline] pub fn get_metadata_pda(mint: &Pubkey) -> Pubkey { Pubkey::find_program_address( &[ @@ -615,26 +671,49 @@ impl PumpFun { ).0 } - // Account related methods - pub fn get_global_account(&self) -> Result { + #[inline] + pub async fn get_global_account(&self) -> Result, anyhow::Error> { let global = Self::get_global_pda(); + + // Try cache first + if let Some(account) = ACCOUNT_CACHE.read().await.get(&global) { + return Ok(account.clone()); + } + + // Cache miss, fetch from RPC let account = self.rpc.get_account(&global)?; - accounts::GlobalAccount::try_from_slice(&account.data) - .map_err(|e| anyhow!(e)) + let global_account = Arc::new(accounts::GlobalAccount::try_from_slice(&account.data)?); + + // Update cache + ACCOUNT_CACHE.write().await.insert(global, global_account.clone()); + + Ok(global_account) } - pub fn get_bonding_curve_account( + #[inline] + pub async fn get_bonding_curve_account( &self, mint: &Pubkey, - ) -> Result { + ) -> Result, anyhow::Error> { let bonding_curve_pda = Self::get_bonding_curve_pda(mint) .ok_or(anyhow!("Bonding curve not found"))?; + + // Try cache first + if let Some(account) = BONDING_CURVE_CACHE.read().await.get(&bonding_curve_pda) { + return Ok(account.clone()); + } + + // Cache miss, fetch from RPC let account = self.rpc.get_account(&bonding_curve_pda)?; - accounts::BondingCurveAccount::try_from_slice(&account.data) - .map_err(|e| anyhow!(e)) + let bonding_curve = Arc::new(accounts::BondingCurveAccount::try_from_slice(&account.data)?); + + // Update cache + BONDING_CURVE_CACHE.write().await.insert(bonding_curve_pda, bonding_curve.clone()); + + Ok(bonding_curve) } - // Subscription related methods + #[inline] pub async fn tokens_subscription( &self, ws_url: &str, @@ -648,138 +727,35 @@ impl PumpFun { logs_subscribe::tokens_subscription(ws_url, commitment, callback, bot_wallet).await } + #[inline] pub async fn stop_subscription(&self, subscription_handle: SubscriptionHandle) { subscription_handle.shutdown().await; } - pub async fn transfer_sol(&self, recieve_wallet: &Pubkey, amount: u64) -> Result<(), anyhow::Error> { - let mut instructions = vec![]; - let transfer_instruction = system_instruction::transfer( - &self.payer.pubkey(), // 付款方地址 - recieve_wallet, // 收款方地址 - amount, // 转账金额 - ); - instructions.push(transfer_instruction); - - let recent_blockhash = self.rpc.get_latest_blockhash()?; - - let transaction = Transaction::new_signed_with_payer( - &instructions, - Some(&self.payer.pubkey()), - &[&self.payer.clone()], - recent_blockhash, - ); - - self.rpc.send_and_confirm_transaction(&transaction)?; - - Ok(()) - } - + #[inline] pub fn get_buy_amount_with_slippage(&self, amount_sol: u64, slippage_basis_points: Option) -> u64 { utils::calculate_with_slippage_buy(amount_sol, slippage_basis_points.unwrap_or(DEFAULT_SLIPPAGE)) } + #[inline] pub fn get_token_price(&self, virtual_sol_reserves: u64, virtual_token_reserves: u64) -> f64 { let v_sol = virtual_sol_reserves as f64 / 100_000_000.0; let v_tokens = virtual_token_reserves as f64 / 100_000.0; - let token_price = v_sol / v_tokens; - token_price + v_sol / v_tokens } + #[inline] pub fn get_buy_price(&self, amount: u64, trade_info: &TradeInfo) -> Result { if amount == 0 { return Ok(0); } - // Calculate the product of virtual reserves using u128 to avoid overflow let n: u128 = (trade_info.virtual_sol_reserves as u128) * (trade_info.virtual_token_reserves as u128); - - // Calculate the new virtual sol reserves after the purchase let i: u128 = (trade_info.virtual_sol_reserves as u128) + (amount as u128); - - // Calculate the new virtual token reserves after the purchase let r: u128 = n / i + 1; - - // Calculate the amount of tokens to be purchased let s: u128 = (trade_info.virtual_token_reserves as u128) - r; - - // Convert back to u64 and return the minimum of calculated tokens and real reserves let s_u64 = s as u64; - Ok(if s_u64 < trade_info.real_token_reserves { - s_u64 - } else { - trade_info.real_token_reserves - }) - } - - pub async fn get_token_price_in_usdc(&self, token_amount: f64) -> Result { - if token_amount == 0.0 { - return Ok(0.0); - } - let url = "https://api.jup.ag/price/v2?ids=So11111111111111111111111111111111111111112"; - let response: serde_json::Value = reqwest::get(url) - .await - .map_err(|e: reqwest::Error| anyhow!(e))? - .json() - .await - .map_err(|e: reqwest::Error| anyhow!(e))?; - - let sol_price_str = response["data"]["So11111111111111111111111111111111111111112"]["price"] - .as_str() - .ok_or(anyhow!("Failed to find SOL price as a string"))?; - - let sol_price_in_usdc: f64 = sol_price_str - .parse() - .map_err(|e: std::num::ParseFloatError| anyhow!(e))?; - - let token_price_in_usdc = sol_price_in_usdc * token_amount; - Ok(token_price_in_usdc) - } - - pub async fn get_sol_price_in_usdc(&self) -> Result { - let url = "https://api.jup.ag/price/v2?ids=So11111111111111111111111111111111111111112"; - let response: serde_json::Value = reqwest::get(url) - .await - .map_err(|_| anyhow!("Failed to install crypto provider"))? - .json() - .await - .map_err(|_| anyhow!("Failed to install crypto provider"))?; - - let sol_price_str = response["data"]["So11111111111111111111111111111111111111112"]["price"] - .as_str() - .ok_or(anyhow!("Failed to find SOL price as a string"))?; - - let sol_price_in_usdc: f64 = sol_price_str - .parse() - .map_err(|_| anyhow!("Failed to parse SOL price as a string"))?; - - Ok(sol_price_in_usdc) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_new_client() { - let payer = Arc::new(Keypair::new()); - let client = PumpFun::new(Cluster::Devnet, None, Arc::clone(&payer), None); - assert_eq!(client.payer.pubkey(), payer.pubkey()); - } - - #[test] - fn test_get_pdas() { - let mint = Keypair::new(); - let global_pda = PumpFun::get_global_pda(); - let mint_authority_pda = PumpFun::get_mint_authority_pda(); - let bonding_curve_pda = PumpFun::get_bonding_curve_pda(&mint.pubkey()); - let metadata_pda = PumpFun::get_metadata_pda(&mint.pubkey()); - - assert!(global_pda != Pubkey::default()); - assert!(mint_authority_pda != Pubkey::default()); - assert!(bonding_curve_pda.is_some()); - assert!(metadata_pda != Pubkey::default()); + Ok(s_u64.min(trade_info.real_token_reserves)) } }