From 61171af619381f915d570887a23e3a4b3ec94e71 Mon Sep 17 00:00:00 2001 From: William Date: Wed, 12 Feb 2025 20:12:14 +0800 Subject: [PATCH] update jito --- Cargo.toml | 3 +- src/jito/abc.rs | 235 ++++++++++++++++++++++++++++++++++++ src/jito/api.rs | 83 +++++++++++++ src/jito/mod.rs | 308 +++++++++++++++++++++++++++++++++--------------- src/lib.rs | 6 +- 5 files changed, 533 insertions(+), 102 deletions(-) create mode 100755 src/jito/abc.rs create mode 100644 src/jito/api.rs mode change 100755 => 100644 src/jito/mod.rs diff --git a/Cargo.toml b/Cargo.toml index f940bd1..55f1381 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,7 @@ futures = "0.3.31" futures-util = "0.3.31" base64 = "0.22.1" bs58 = "0.5.1" -rand = "0.8.5" +rand = "0.9.0" bincode = "1.3.3" anyhow = "1.0.90" reqwest = { version = "0.11.27", features = ["json"] } @@ -41,4 +41,5 @@ pretty_env_logger = "0.5.0" log = "0.4.22" chrono = "0.4.39" regex = "1" +tracing = "0.1.41" diff --git a/src/jito/abc.rs b/src/jito/abc.rs new file mode 100755 index 0000000..ce03eed --- /dev/null +++ b/src/jito/abc.rs @@ -0,0 +1,235 @@ +use std::fmt; +use std::str::FromStr; +use rand::seq::SliceRandom; +use reqwest::Client; +use serde_json::{json, Value}; + +use solana_sdk::{ + pubkey::Pubkey, + transaction::Transaction, +}; + +use crate::error::{ClientError, ClientResult}; + +#[derive(Clone, Debug)] +pub struct JitoClient { + base_url: String, + uuid: Option, + client: Client, +} + +#[derive(Debug)] +pub struct PrettyJsonValue(pub Value); + +impl fmt::Display for PrettyJsonValue { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", serde_json::to_string_pretty(&self.0).unwrap()) + } +} + +impl From for PrettyJsonValue { + fn from(value: Value) -> Self { + PrettyJsonValue(value) + } +} + +impl JitoClient { + pub fn new(base_url: &str, uuid: Option) -> Self { + Self { + base_url: base_url.to_string(), + uuid, + client: Client::new(), + } + } + + async fn send_request(&self, endpoint: &str, method: &str, params: Option) -> ClientResult { + let url = format!("{}{}", self.base_url, endpoint); + + let data = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": method, + "params": params.unwrap_or(json!([])) + }); + + // println!("Sending request to: {}", url); + // println!("Request body: {}", serde_json::to_string_pretty(&data).unwrap()); + + let response = self.client + .post(&url) + .header("Content-Type", "application/json") + .json(&data) + .send() + .await + .map_err(|e| ClientError::Other(format!("Request failed: {}", e)))?; + + let status = response.status(); + // println!("Response status: {}", status); + + let body = response.json::().await + .map_err(|e| ClientError::Other(format!("Failed to parse response: {}", e)))?; + // println!("Response body: {}", serde_json::to_string_pretty(&body).unwrap()); + + Ok(body) + } + + pub async fn get_tip_accounts(&self) -> ClientResult { + let endpoint = if let Some(uuid) = &self.uuid { + format!("/bundles?uuid={}", uuid) + } else { + "/bundles".to_string() + }; + + self.send_request(&endpoint, "getTipAccounts", None).await + } + + // Get a random tip account + pub async fn get_tip_account(&self) -> ClientResult { + let tip_accounts_response = self.get_tip_accounts().await?; + + let tip_accounts = tip_accounts_response["result"] + .as_array() + .ok_or_else(|| ClientError::Other("Failed to parse tip accounts as array".to_string()))?; + + if tip_accounts.is_empty() { + return Err(ClientError::Other("No tip accounts available".to_string())); + } + + let random_account = tip_accounts + .choose(&mut rand::thread_rng()) + .ok_or_else(|| ClientError::Other("Failed to choose random tip account".to_string()))?; + + let address = random_account + .as_str() + .ok_or_else(|| ClientError::Other("Failed to parse tip account as string".to_string()))?; + + Pubkey::from_str(address) + .map_err(|e| ClientError::Other(format!("Failed to parse pubkey: {}", e))) + } + + pub async fn get_bundle_statuses(&self, bundle_uuids: Vec) -> ClientResult { + let endpoint = if let Some(uuid) = &self.uuid { + format!("/bundles?uuid={}", uuid) + } else { + "/bundles".to_string() + }; + + // Construct the params as a list within a list + let params = json!([bundle_uuids]); + + self.send_request(&endpoint, "getBundleStatuses", Some(params)) + .await + } + + pub async fn send_transaction( + &self, + transaction: &Transaction, + ) -> ClientResult { + let wire_transaction = bincode::serialize(transaction).map_err(|e| { + ClientError::Parse( + "Transaction serialization failed".to_string(), + e.to_string(), + ) + })?; + + let serialized_tx = bs58::encode(&wire_transaction).into_string(); + + // Prepare bundle for submission (array of transactions) + let bundle = json!([serialized_tx]); + + // UUID for the bundle + let uuid = None; + + // Send bundle using Jito SDK + // println!("Sending bundle with 1 transaction..."); + let response = self.send_bundle(Some(bundle), uuid).await?; + + response["result"] + .as_str() + .map(|s| s.to_string()) + .ok_or_else(|| ClientError::Parse( + "Invalid response format".to_string(), + "Missing result field".to_string(), + )) + } + + pub async fn send_bundle(&self, params: Option, uuid: Option<&str>) -> ClientResult { + let mut endpoint = "/bundles".to_string(); + + if let Some(uuid) = uuid { + endpoint = format!("{}?uuid={}", endpoint, uuid); + } + + // Ensure params is an array of transactions + let transactions = match params { + Some(Value::Array(transactions)) => { + if transactions.is_empty() { + return Err(ClientError::Other("Bundle must contain at least one transaction".to_string())); + } + if transactions.len() > 5 { + return Err(ClientError::Other("Bundle can contain at most 5 transactions".to_string())); + } + transactions + }, + _ => return Err(ClientError::Other("Invalid bundle format: expected an array of transactions".to_string())), + }; + + // Wrap the transactions array in another array + let params = json!([transactions]); + + // Send the wrapped transactions array + self.send_request(&endpoint, "sendBundle", Some(params)) + .await + } + + pub async fn send_txn(&self, params: Option, bundle_only: bool) -> ClientResult { + let mut query_params = Vec::new(); + + if bundle_only { + query_params.push("bundleOnly=true".to_string()); + } + + let endpoint = if query_params.is_empty() { + "/transactions".to_string() + } else { + format!("/transactions?{}", query_params.join("&")) + }; + + // Construct params as an array instead of an object + let params = match params { + Some(Value::Object(map)) => { + let tx = map.get("tx").and_then(Value::as_str).unwrap_or_default(); + let skip_preflight = map.get("skipPreflight").and_then(Value::as_bool).unwrap_or(false); + json!([ + tx, + { + "encoding": "base64", + "skipPreflight": skip_preflight + } + ]) + }, + _ => json!([]), + }; + + self.send_request(&endpoint, "sendTransaction", Some(params)).await + } + + pub async fn get_in_flight_bundle_statuses(&self, bundle_uuids: Vec) -> ClientResult { + let endpoint = if let Some(uuid) = &self.uuid { + format!("/bundles?uuid={}", uuid) + } else { + "/bundles".to_string() + }; + + // Construct the params as a list within a list + let params = json!([bundle_uuids]); + + self.send_request(&endpoint, "getInflightBundleStatuses", Some(params)) + .await + } + + // Helper method to convert Value to PrettyJsonValue + pub fn prettify(value: Value) -> PrettyJsonValue { + PrettyJsonValue(value) + } +} \ No newline at end of file diff --git a/src/jito/api.rs b/src/jito/api.rs new file mode 100644 index 0000000..9ae8133 --- /dev/null +++ b/src/jito/api.rs @@ -0,0 +1,83 @@ +use std::env; + +use anyhow::{Context, Result}; +use reqwest::Proxy; +use serde::{Deserialize, Serialize}; +use std::convert::TryFrom; + +use super::TipPercentileData; + +#[derive(Serialize)] +struct RpcRequest { + jsonrpc: String, + id: u32, + method: String, + params: Vec<()>, +} + +#[derive(Deserialize, Debug)] +pub struct RpcResponse { + pub jsonrpc: String, + pub id: u32, + pub result: serde_json::Value, +} + +pub async fn get_tip_accounts(block_engine_url: &str) -> Result { + let mut client_builder = reqwest::Client::builder(); + if let Ok(http_proxy) = env::var("HTTP_PROXY") { + let proxy = Proxy::all(http_proxy)?; + client_builder = client_builder.proxy(proxy); + } + let client = client_builder.build()?; + let request_body = RpcRequest { + jsonrpc: "2.0".to_string(), + id: 1, + method: "getTipAccounts".to_string(), + params: vec![], + }; + let result = client + .post(format!("{}/api/v1/bundles", block_engine_url)) + .json(&request_body) + .send() + .await? + .json::() + .await?; + + Ok(result) +} +/// tip accounts +#[derive(Debug)] +pub struct TipAccountResult { + pub accounts: Vec, +} + +impl TryFrom for TipAccountResult { + type Error = anyhow::Error; + fn try_from(value: RpcResponse) -> Result { + let accounts = value + .result + .as_array() + .context("expected 'result' to be an array")? + .iter() + .map(|v| v.as_str().unwrap().to_string()) + .collect(); + Ok(TipAccountResult { accounts }) + } +} + +pub async fn get_tip_amounts() -> Result> { + let mut client_builder = reqwest::Client::builder(); + if let Ok(http_proxy) = env::var("HTTP_PROXY") { + let proxy = Proxy::all(http_proxy)?; + client_builder = client_builder.proxy(proxy); + } + let client = client_builder.build()?; + + let result = client + .get("https://bundles.jito.wtf/api/v1/bundles/tip_floor") + .send() + .await? + .json::>() + .await?; + Ok(result) +} diff --git a/src/jito/mod.rs b/src/jito/mod.rs old mode 100755 new mode 100644 index ce03eed..3bf0109 --- a/src/jito/mod.rs +++ b/src/jito/mod.rs @@ -1,23 +1,45 @@ -use std::fmt; -use std::str::FromStr; -use rand::seq::SliceRandom; -use reqwest::Client; -use serde_json::{json, Value}; +use std::{convert::TryInto, future::Future, str::FromStr, time::Duration, fmt}; -use solana_sdk::{ - pubkey::Pubkey, - transaction::Transaction, +use anyhow::{anyhow, Result}; +use api::{get_tip_accounts, TipAccountResult}; +use rand::{rng, seq::IteratorRandom}; +use reqwest::Client; +use serde::Deserialize; +use serde_json::{json, Value}; +use solana_sdk::{pubkey::Pubkey, transaction::Transaction}; +use tokio::{ + sync::RwLock, + time::{sleep, Instant}, }; +use tracing::{debug, error, info, warn}; use crate::error::{ClientError, ClientResult}; -#[derive(Clone, Debug)] +pub mod api; + +#[derive(Debug)] pub struct JitoClient { base_url: String, + tip_accounts: RwLock>, + tips_percentile: RwLock>, + tip_percentile: String, uuid: Option, client: Client, } +impl Clone for JitoClient { + fn clone(&self) -> Self { + Self { + base_url: self.base_url.clone(), + tip_accounts: RwLock::new(Vec::new()), + tips_percentile: RwLock::new(None), + tip_percentile: self.tip_percentile.clone(), + uuid: self.uuid.clone(), + client: Client::new(), + } + } +} + #[derive(Debug)] pub struct PrettyJsonValue(pub Value); @@ -33,10 +55,41 @@ impl From for PrettyJsonValue { } } +const DEFAULT_TIP_PERCENTILE: &str = "25"; + +#[derive(Debug, Deserialize, Clone)] +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, +} + +#[derive(Deserialize, Debug)] +pub struct BundleStatus { + pub bundle_id: String, + pub transactions: Vec, + pub slot: u64, + pub confirmation_status: String, + pub err: ErrorStatus, +} + +#[derive(Deserialize, Debug)] +pub struct ErrorStatus { + #[serde(rename = "Ok")] + pub ok: Option<()>, +} + impl JitoClient { pub fn new(base_url: &str, uuid: Option) -> Self { Self { base_url: base_url.to_string(), + tip_accounts: RwLock::new(vec![]), + tips_percentile: RwLock::new(None), + tip_percentile: DEFAULT_TIP_PERCENTILE.to_string(), uuid, client: Client::new(), } @@ -52,9 +105,6 @@ impl JitoClient { "params": params.unwrap_or(json!([])) }); - // println!("Sending request to: {}", url); - // println!("Request body: {}", serde_json::to_string_pretty(&data).unwrap()); - let response = self.client .post(&url) .header("Content-Type", "application/json") @@ -63,62 +113,74 @@ impl JitoClient { .await .map_err(|e| ClientError::Other(format!("Request failed: {}", e)))?; - let status = response.status(); - // println!("Response status: {}", status); - let body = response.json::().await .map_err(|e| ClientError::Other(format!("Failed to parse response: {}", e)))?; - // println!("Response body: {}", serde_json::to_string_pretty(&body).unwrap()); Ok(body) } - pub async fn get_tip_accounts(&self) -> ClientResult { - let endpoint = if let Some(uuid) = &self.uuid { - format!("/bundles?uuid={}", uuid) - } else { - "/bundles".to_string() - }; + pub async fn init_tip_accounts(&self) -> Result<()> { + let accounts: TipAccountResult = get_tip_accounts(&self.base_url).await?.try_into()?; + let mut tip_accounts = self.tip_accounts.write().await; - self.send_request(&endpoint, "getTipAccounts", None).await + accounts + .accounts + .iter() + .for_each(|account| tip_accounts.push(account.to_string())); + Ok(()) } - // Get a random tip account - pub async fn get_tip_account(&self) -> ClientResult { - let tip_accounts_response = self.get_tip_accounts().await?; - - let tip_accounts = tip_accounts_response["result"] - .as_array() - .ok_or_else(|| ClientError::Other("Failed to parse tip accounts as array".to_string()))?; + pub async fn get_tip_account(&self) -> Result { + // 第一次尝试获取读锁 + { + let accounts = self.tip_accounts.read().await; + if !accounts.is_empty() { + let mut rng = rng(); + if let Some(acc) = accounts.iter().choose(&mut rng) { + return Ok(Pubkey::from_str(acc).inspect_err(|err| { + error!("jito: failed to parse Pubkey: {:?}", err); + })?); + } + } + } // 这里释放读锁 - if tip_accounts.is_empty() { - return Err(ClientError::Other("No tip accounts available".to_string())); + // 如果账户列表为空，初始化账户列表 + self.init_tip_accounts().await?; + + // 重新获取读锁 + let accounts = self.tip_accounts.read().await; + let mut rng = rng(); + match accounts.iter().choose(&mut 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")), } - - let random_account = tip_accounts - .choose(&mut rand::thread_rng()) - .ok_or_else(|| ClientError::Other("Failed to choose random tip account".to_string()))?; - - let address = random_account - .as_str() - .ok_or_else(|| ClientError::Other("Failed to parse tip account as string".to_string()))?; - - Pubkey::from_str(address) - .map_err(|e| ClientError::Other(format!("Failed to parse pubkey: {}", e))) } - pub async fn get_bundle_statuses(&self, bundle_uuids: Vec) -> ClientResult { - let endpoint = if let Some(uuid) = &self.uuid { - format!("/bundles?uuid={}", uuid) + pub async fn init_tip_amounts(&self) -> Result<()> { + let tip_percentiles = api::get_tip_amounts().await?; + *self.tips_percentile.write().await = tip_percentiles.first().cloned(); + + Ok(()) + } + + // unit sol + pub async fn get_tip_value(&self) -> Result { + let tips = self.tips_percentile.read().await; + + if let Some(ref data) = *tips { + match self.tip_percentile.as_str() { + "25" => Ok(data.landed_tips_25th_percentile), + "50" => Ok(data.landed_tips_50th_percentile), + "75" => Ok(data.landed_tips_75th_percentile), + "95" => Ok(data.landed_tips_95th_percentile), + "99" => Ok(data.landed_tips_99th_percentile), + _ => Err(anyhow!("jito: invalid TIP_PERCENTILE value")), + } } else { - "/bundles".to_string() - }; - - // Construct the params as a list within a list - let params = json!([bundle_uuids]); - - self.send_request(&endpoint, "getBundleStatuses", Some(params)) - .await + Err(anyhow!("jito: failed get tip")) + } } pub async fn send_transaction( @@ -138,10 +200,9 @@ impl JitoClient { let bundle = json!([serialized_tx]); // UUID for the bundle - let uuid = None; + let uuid = self.uuid.clone(); // Send bundle using Jito SDK - // println!("Sending bundle with 1 transaction..."); let response = self.send_bundle(Some(bundle), uuid).await?; response["result"] @@ -153,7 +214,7 @@ impl JitoClient { )) } - pub async fn send_bundle(&self, params: Option, uuid: Option<&str>) -> ClientResult { + pub async fn send_bundle(&self, params: Option, uuid: Option) -> ClientResult { let mut endpoint = "/bundles".to_string(); if let Some(uuid) = uuid { @@ -182,54 +243,105 @@ impl JitoClient { .await } - pub async fn send_txn(&self, params: Option, bundle_only: bool) -> ClientResult { - let mut query_params = Vec::new(); + pub async fn wait_for_bundle_confirmation( + &self, + fetch_statuses: F, + bundle_id: String, + interval: Duration, + timeout: Duration, + ) -> Result> + where + F: Fn(String) -> Fut, + Fut: Future>>, + { + let start_time = Instant::now(); - if bundle_only { - query_params.push("bundleOnly=true".to_string()); - } + loop { + let statuses = fetch_statuses(bundle_id.clone()).await?; - let endpoint = if query_params.is_empty() { - "/transactions".to_string() - } else { - format!("/transactions?{}", query_params.join("&")) - }; + if let Some(status) = statuses.first() { + let bundle_status: BundleStatus = + serde_json::from_value(status.clone()).inspect_err(|err| { + error!( + "Failed to parse JSON when get_bundle_statuses, err: {}", + err, + ); + })?; - // Construct params as an array instead of an object - let params = match params { - Some(Value::Object(map)) => { - let tx = map.get("tx").and_then(Value::as_str).unwrap_or_default(); - let skip_preflight = map.get("skipPreflight").and_then(Value::as_bool).unwrap_or(false); - json!([ - tx, - { - "encoding": "base64", - "skipPreflight": skip_preflight + debug!("{:?}", bundle_status); + match bundle_status.confirmation_status.as_str() { + "finalized" | "confirmed" => { + info!( + "Finalized bundle {}: {}", + bundle_id, bundle_status.confirmation_status + ); + bundle_status + .transactions + .iter() + .for_each(|tx| info!("https://solscan.io/tx/{}", tx)); + return Ok(bundle_status.transactions); } - ]) - }, - _ => json!([]), - }; + _ => { + debug!("bundle_status: {:?}", bundle_status); + } + } + } else { + debug!("Finalizing bundle {}: {}", bundle_id, "None"); + } - self.send_request(&endpoint, "sendTransaction", Some(params)).await + if start_time.elapsed() > timeout { + warn!("Loop exceeded {:?}, breaking out.", timeout); + return Err(anyhow!("Bundle status get timeout")); + } + + sleep(interval).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} + })] } - pub async fn get_in_flight_bundle_statuses(&self, bundle_uuids: Vec) -> ClientResult { - let endpoint = if let Some(uuid) = &self.uuid { - format!("/bundles?uuid={}", uuid) - } else { - "/bundles".to_string() - }; - - // Construct the params as a list within a list - let params = json!([bundle_uuids]); - - self.send_request(&endpoint, "getInflightBundleStatuses", Some(params)) - .await + #[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()); + } } - // Helper method to convert Value to PrettyJsonValue - pub fn prettify(value: Value) -> PrettyJsonValue { - PrettyJsonValue(value) + #[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()); } -} \ No newline at end of file +} diff --git a/src/lib.rs b/src/lib.rs index 49f15ed..f22ab9b 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,6 +7,7 @@ pub mod jito; pub mod grpc; pub mod common; +use anyhow::anyhow; use solana_client::rpc_client::RpcClient; use solana_sdk::{ commitment_config::CommitmentConfig, @@ -25,7 +26,6 @@ use spl_associated_token_account::{ use common::{logs_data::TradeInfo, logs_events::PumpfunEvent, logs_subscribe}; use common::logs_subscribe::SubscriptionHandle; -use common::logs_events::DexEvent; use spl_token::instruction::close_account; use std::sync::Arc; @@ -264,7 +264,7 @@ impl PumpFun { utils::calculate_with_slippage_buy(max_sol_cost, slippage_basis_points.unwrap_or(DEFAULT_SLIPPAGE)); let mut instructions = vec![]; - let tip_account = jito_client.get_tip_account().await?; + let tip_account = jito_client.get_tip_account().await.map_err(|e| anyhow!(e)).unwrap(); let ata = get_associated_token_address(&self.payer.pubkey(), mint); if self.rpc.get_account(&ata).is_err() { instructions.push(create_associated_token_account( @@ -452,7 +452,7 @@ impl PumpFun { ); let mut instructions = vec![]; - let tip_account = jito_client.get_tip_account().await?; + let tip_account = jito_client.get_tip_account().await.map_err(|e| ClientError::Other(e.to_string()))?; instructions.push(instruction::sell( &self.payer.clone(), mint,