gavindiaz/solana-streamer
1
0
Fork 0
mirror of https://github.com/0xfnzero/solana-streamer.git synced 2026-08-13 08:58:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor the code to optimize performance.

Browse Source
This commit is contained in:
William
2025-02-15 18:53:17 +08:00
parent 7cf908ae5d
commit 38f1ce3fad
5 changed files with 189 additions and 580 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+3
View File
@@ -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"
src/grpc/mod.rs
+13 -115
View File
@@ -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<String, SubscribeRequestFilterTransactions>;
// 常量定义
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<String>) -> 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::<TransactionPretty>(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<F>(&self, callback: F, bot_wallet: Option<Pubkey>) -> 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<TransactionPretty>,
@@ -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::<SwapBaseInLog>(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<SwapType> {
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(())
}
src/jito/mod.rs
+20 -83
View File
@@ -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<Vec<String>>,
tips_percentile: RwLock<Option<TipPercentileData>>,
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<String>) -> Self {
pub fn new(jito_url: &str, _uuid: Option<String>) -> 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<TipAccountResult> {
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<Value> {
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());
}
}
src/jito/rpc_client.rs
+2 -207
View File
@@ -22,33 +22,13 @@ use crate::jito::{
};
pub type RpcResult<T> = client_error::Result<Response<T>>;
/// 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<dyn RpcSender + Send + Sync + 'static>,
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<T: RpcSender + Send + Sync + 'static>(
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<Vec<String>> {
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);
}
}
src/lib.rs
+151 -175
View File
@@ -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<HashMap<Pubkey, Arc<accounts::GlobalAccount>>> = RwLock::new(HashMap::new());
static ref BONDING_CURVE_CACHE: RwLock<HashMap<Pubkey, Arc<accounts::BondingCurveAccount>>> = RwLock::new(HashMap::new());
}
/// Priority fee configuration
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PriorityFee {
pub limit: Option<u32>,
@@ -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<CommitmentConfig>,
@@ -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<Signature, anyhow::Error> {
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<u64>,
priority_fee: Option<PriorityFee>,
) -> Result<Signature, anyhow::Error> {
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<u64>,
priority_fee: Option<PriorityFee>,
) -> Result<Signature, anyhow::Error> {
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<u64>,
jito_fee: Option<u64>,
jito_fee: Option<f64>,
) -> Result<String, anyhow::Error> {
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<u64>,
priority_fee: Option<PriorityFee>,
) -> 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::<u64>()
@@ -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::<u64>() / fees.len() as u64;
let average_fees = if fees.is_empty() {
DEFAULT_COMPUTE_UNIT_PRICE
} else {
fees.iter()
.map(|fee| fee.prioritization_fee)
.sum::<u64>() / 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<u64>,
priority_fee: Option<PriorityFee>,
) -> 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<u64>,
jito_fee: Option<u64>,
jito_fee: Option<f64>,
) -> Result<String, 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);
@@ -477,7 +483,7 @@ impl PumpFun {
mint: &Pubkey,
amount_token: Option<u64>,
slippage_basis_points: Option<u64>,
jito_fee: Option<u64>,
jito_fee: Option<f64>,
) -> Result<String, anyhow::Error> {
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<PriorityFee>) -> Vec<Instruction> {
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<u64, anyhow::Error> {
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<u64, anyhow::Error> {
self.rpc.get_balance(account).map_err(|_| anyhow!("Failed to get SOL balance"))
}
#[inline]
pub fn get_payer_token_balance(&self, mint: &Pubkey) -> Result<u64, anyhow::Error> {
self.get_token_balance(&self.payer.pubkey(), mint)
}
#[inline]
pub fn get_payer_sol_balance(&self) -> Result<u64, anyhow::Error> {
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<Pubkey> = 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<Pubkey> = 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> {
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<accounts::GlobalAccount, anyhow::Error> {
#[inline]
pub async fn get_global_account(&self) -> Result<Arc<accounts::GlobalAccount>, 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<accounts::BondingCurveAccount, anyhow::Error> {
) -> Result<Arc<accounts::BondingCurveAccount>, 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<F>(
&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>) -> 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<u64, &'static str> {
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<f64, anyhow::Error> {
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<f64, anyhow::Error> {
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))
}
}