diff --git a/Cargo.toml b/Cargo.toml index 55f1381..80f7507 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,9 +13,12 @@ readme = "README.md" crate-type = ["cdylib", "rlib"] [dependencies] -solana-sdk = "2.1.7" -solana-client = "2.1.7" -solana-transaction-status = "2.1.7" +solana-sdk = "2.1.13" +solana-client = "2.1.13" +solana-transaction-status = "2.1.13" +solana-rpc-client = "2.1.13" +solana-rpc-client-api = "2.1.13" +solana-program = "2.1.13" spl-token = "7.0.0" spl-associated-token-account = "6.0.0" mpl-token-metadata = "5.1.0" @@ -30,11 +33,11 @@ bs58 = "0.5.1" rand = "0.9.0" bincode = "1.3.3" anyhow = "1.0.90" -reqwest = { version = "0.11.27", features = ["json"] } +reqwest = { version = "0.12.12", features = ["json"] } tonic = { version = "0.12.3", features = ["tls", "tls-webpki-roots", "tls-roots"] } tokio = { version = "1.42.0" , features = ["full", "rt-multi-thread"]} -yellowstone-grpc-client = { version = "4.1.0" } -yellowstone-grpc-proto = { version = "4.1.1" } +yellowstone-grpc-client = { version = "5.0.0" } +yellowstone-grpc-proto = { version = "5.0.0" } rustls = { version = "0.23.20", features = ["ring"] } dotenvy = "0.15.7" pretty_env_logger = "0.5.0" @@ -42,4 +45,6 @@ log = "0.4.22" chrono = "0.4.39" regex = "1" tracing = "0.1.41" +thiserror = "2.0.11" +async-trait = "0.1.86" diff --git a/src/jito/abc.rs b/src/jito/abc.rs deleted file mode 100755 index ce03eed..0000000 --- a/src/jito/abc.rs +++ /dev/null @@ -1,235 +0,0 @@ -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 index 52deee0..9cac97c 100644 --- a/src/jito/api.rs +++ b/src/jito/api.rs @@ -6,75 +6,13 @@ use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; 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 client_builder = reqwest::Client::builder(); - 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 TipAccountResult { - pub fn from(value: Value) -> Result { - let accounts = value["result"] - .as_array() - .context("expected 'result' to be an array")? - .iter() - .map(|v| v.as_str().unwrap().to_string()) - .collect(); + pub fn from(accounts: Vec) -> Result { 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/client_error.rs b/src/jito/client_error.rs new file mode 100644 index 0000000..14cd2af --- /dev/null +++ b/src/jito/client_error.rs @@ -0,0 +1,125 @@ +pub use reqwest; +use solana_rpc_client_api::{client_error::ErrorKind, request}; +use solana_sdk::{ + signature::SignerError, transaction::TransactionError, transport::TransportError, +}; +use thiserror::Error as ThisError; + +use crate::jito::request::RpcRequest; + +#[derive(ThisError, Debug)] +#[error("{kind}")] +pub struct Error { + pub request: Option, + + #[source] + pub kind: ErrorKind, +} + +impl Error { + pub fn new_with_request(kind: ErrorKind, request: RpcRequest) -> Self { + Self { + request: Some(request), + kind, + } + } + + pub fn into_with_request(self, request: RpcRequest) -> Self { + Self { + request: Some(request), + ..self + } + } + + pub fn request(&self) -> Option<&RpcRequest> { + self.request.as_ref() + } + + pub fn kind(&self) -> &ErrorKind { + &self.kind + } + + pub fn get_transaction_error(&self) -> Option { + self.kind.get_transaction_error() + } +} + +impl From for Error { + fn from(kind: ErrorKind) -> Self { + Self { + request: None, + kind, + } + } +} + +impl From for Error { + fn from(err: TransportError) -> Self { + Self { + request: None, + kind: err.into(), + } + } +} + +impl From for TransportError { + fn from(client_error: Error) -> Self { + client_error.kind.into() + } +} + +impl From for Error { + fn from(err: std::io::Error) -> Self { + Self { + request: None, + kind: err.into(), + } + } +} + +impl From for Error { + fn from(err: reqwest::Error) -> Self { + Self { + request: None, + kind: ErrorKind::Custom(format!("Reqwest error: {}", err)), + } + } +} + +impl From for Error { + fn from(err: request::RpcError) -> Self { + Self { + request: None, + kind: err.into(), + } + } +} + +impl From for Error { + fn from(err: serde_json::error::Error) -> Self { + Self { + request: None, + kind: err.into(), + } + } +} + +impl From for Error { + fn from(err: SignerError) -> Self { + Self { + request: None, + kind: err.into(), + } + } +} + +impl From for Error { + fn from(err: TransactionError) -> Self { + Self { + request: None, + kind: err.into(), + } + } +} + +pub type Result = std::result::Result; diff --git a/src/jito/http_sender.rs b/src/jito/http_sender.rs new file mode 100644 index 0000000..44c8caf --- /dev/null +++ b/src/jito/http_sender.rs @@ -0,0 +1,218 @@ +use std::{ + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, RwLock, + }, + time::{Duration, Instant}, +}; + +use async_trait::async_trait; +use log::debug; +use reqwest::{ + self, + header::{CONTENT_TYPE, RETRY_AFTER}, + StatusCode, +}; +use solana_rpc_client_api::{ + custom_error, + error_object::RpcErrorObject, + request::{RpcError, RpcResponseErrorData}, + response::RpcSimulateTransactionResult, +}; +use tokio::time::sleep; + +use crate::jito::{client_error::Result, request::RpcRequest, rpc_sender::RpcSender}; + +pub struct HttpSender { + client: Arc, + url: String, + request_id: AtomicU64, + stats: RwLock, +} + +/// Nonblocking [`RpcSender`] over HTTP. +impl HttpSender { + /// Create an HTTP RPC sender. + /// + /// The URL is an HTTP URL, usually for port 8899, as in + /// "http://localhost:8899". The sender has a default timeout of 30 seconds. + pub fn new(url: U) -> Self { + Self::new_with_timeout(url, Duration::from_secs(30)) + } + + /// Create an HTTP RPC sender. + /// + /// The URL is an HTTP URL, usually for port 8899. + pub fn new_with_timeout(url: U, timeout: Duration) -> Self { + let client = Arc::new( + reqwest::Client::builder() + .timeout(timeout) + .pool_idle_timeout(timeout) + .build() + .expect("build rpc client"), + ); + + Self { + client, + url: url.to_string(), + request_id: AtomicU64::new(0), + stats: RwLock::new(solana_rpc_client::rpc_sender::RpcTransportStats::default()), + } + } +} + +struct StatsUpdater<'a> { + stats: &'a RwLock, + request_start_time: Instant, + rate_limited_time: Duration, +} + +impl<'a> StatsUpdater<'a> { + fn new(stats: &'a RwLock) -> Self { + Self { + stats, + request_start_time: Instant::now(), + rate_limited_time: Duration::default(), + } + } + + fn add_rate_limited_time(&mut self, duration: Duration) { + self.rate_limited_time += duration; + } +} + +impl<'a> Drop for StatsUpdater<'a> { + fn drop(&mut self) { + let mut stats = self.stats.write().unwrap(); + stats.request_count += 1; + stats.elapsed_time += Instant::now().duration_since(self.request_start_time); + stats.rate_limited_time += self.rate_limited_time; + } +} + +#[async_trait] +impl RpcSender for HttpSender { + fn get_transport_stats(&self) -> solana_rpc_client::rpc_sender::RpcTransportStats { + self.stats.read().unwrap().clone() + } + + async fn send( + &self, + request: RpcRequest, + params: serde_json::Value, + ) -> Result { + let mut stats_updater = StatsUpdater::new(&self.stats); + + let request_id = self.request_id.fetch_add(1, Ordering::Relaxed); + let request_json = request.build_request_json(request_id, params).to_string(); + + let mut too_many_requests_retries = 5; + loop { + let response = { + let client = self.client.clone(); + let request_json = request_json.clone(); + client + .post(&self.url) + .header(CONTENT_TYPE, "application/json") + .body(request_json) + .send() + .await + }?; + + if !response.status().is_success() { + if response.status() == StatusCode::TOO_MANY_REQUESTS + && too_many_requests_retries > 0 + { + let mut duration = Duration::from_millis(500); + if let Some(retry_after) = response.headers().get(RETRY_AFTER) { + if let Ok(retry_after) = retry_after.to_str() { + if let Ok(retry_after) = retry_after.parse::() { + if retry_after < 120 { + duration = Duration::from_secs(retry_after); + } + } + } + } + + too_many_requests_retries -= 1; + debug!( + "Too many requests: server responded with {:?}, {} retries left, pausing for {:?}", + response, too_many_requests_retries, duration + ); + + sleep(duration).await; + stats_updater.add_rate_limited_time(duration); + continue; + } + return Err(response.error_for_status().unwrap_err().into()); + } + + let mut json = response.json::().await?; + if json["error"].is_object() { + return match serde_json::from_value::(json["error"].clone()) { + Ok(rpc_error_object) => { + let data = match rpc_error_object.code { + solana_rpc_client_api::custom_error::JSON_RPC_SERVER_ERROR_SEND_TRANSACTION_PREFLIGHT_FAILURE => { + match serde_json::from_value::(json["error"]["data"].clone()) { + Ok(data) => RpcResponseErrorData::SendTransactionPreflightFailure(data), + Err(err) => { + debug!("Failed to deserialize RpcSimulateTransactionResult: {:?}", err); + RpcResponseErrorData::Empty + } + } + }, + custom_error::JSON_RPC_SERVER_ERROR_NODE_UNHEALTHY => { + match serde_json::from_value::(json["error"]["data"].clone()) { + Ok(custom_error::NodeUnhealthyErrorData {num_slots_behind}) => RpcResponseErrorData::NodeUnhealthy {num_slots_behind}, + Err(_err) => { + RpcResponseErrorData::Empty + } + } + }, + _ => RpcResponseErrorData::Empty + }; + + Err(RpcError::RpcResponseError { + code: rpc_error_object.code, + message: rpc_error_object.message, + data, + } + .into()) + } + Err(err) => Err(RpcError::RpcRequestError(format!( + "Failed to deserialize RPC error response: {} [{}]", + serde_json::to_string(&json["error"]).unwrap(), + err + )) + .into()), + }; + } + return Ok(json["result"].take()); + } + } + + fn url(&self) -> String { + self.url.clone() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test(flavor = "multi_thread")] + async fn http_sender_on_tokio_multi_thread() { + let http_sender = HttpSender::new("http://localhost:1234".to_string()); + let _ = http_sender + .send(RpcRequest::GetTipAccounts, serde_json::Value::Null) + .await; + } + + #[tokio::test(flavor = "current_thread")] + async fn http_sender_on_tokio_current_thread() { + let http_sender = HttpSender::new("http://localhost:1234".to_string()); + let _ = http_sender + .send(RpcRequest::GetTipAccounts, serde_json::Value::Null) + .await; + } +} diff --git a/src/jito/mod.rs b/src/jito/mod.rs index 445bdec..d4143ef 100644 --- a/src/jito/mod.rs +++ b/src/jito/mod.rs @@ -1,30 +1,32 @@ -use std::{convert::TryInto, future::Future, str::FromStr, time::Duration, fmt}; +use std::{str::FromStr, time::Duration, fmt}; use anyhow::{anyhow, Result}; use api::TipAccountResult; -use rand::{rng, seq::IteratorRandom}; -use reqwest::Client; +use rand::{seq::IteratorRandom}; use serde::Deserialize; use serde_json::{json, Value}; -use solana_sdk::{pubkey::Pubkey, transaction::Transaction}; -use tokio::{ - sync::RwLock, - time::{sleep, Instant}, +use solana_sdk::{ + pubkey::Pubkey, + transaction::{Transaction, VersionedTransaction}, }; -use tracing::{debug, error, info, warn}; - -use crate::error::{ClientError, ClientResult}; +use tokio::sync::RwLock; +use tracing::error; pub mod api; +pub mod client_error; +pub mod http_sender; +pub mod request; +pub mod rpc_client; +pub mod rpc_sender; + +use crate::jito::rpc_client::RpcClient; -#[derive(Debug)] pub struct JitoClient { base_url: String, tip_accounts: RwLock>, tips_percentile: RwLock>, tip_percentile: String, - uuid: Option, - client: Client, + client: RpcClient, } impl Clone for JitoClient { @@ -34,30 +36,12 @@ impl Clone for JitoClient { tip_accounts: RwLock::new(Vec::new()), tips_percentile: RwLock::new(None), tip_percentile: self.tip_percentile.clone(), - uuid: self.uuid.clone(), - client: Client::new(), + client: RpcClient::new(self.base_url.clone()), } } } -#[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) - } -} - const DEFAULT_TIP_PERCENTILE: &str = "25"; - -#[derive(Debug, Deserialize, Clone)] pub struct TipPercentileData { pub time: String, pub landed_tips_25th_percentile: f64, @@ -68,68 +52,23 @@ pub struct TipPercentileData { 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 { + pub fn new(jito_url: &str, uuid: Option) -> Self { Self { - base_url: base_url.to_string(), + base_url: jito_url.to_string(), tip_accounts: RwLock::new(vec![]), tips_percentile: RwLock::new(None), tip_percentile: DEFAULT_TIP_PERCENTILE.to_string(), - uuid, - client: Client::new(), + client: RpcClient::new(jito_url.to_string()), } } pub async fn get_tip_accounts(&self) -> Result { - let endpoint = if let Some(uuid) = &self.uuid { - format!("/api/v1/bundles?uuid={}", uuid) - } else { - "/api/v1/bundles".to_string() - }; - - let result = self.send_request(&endpoint, "getTipAccounts", None).await?; + let result = self.client.get_tip_accounts().await?; let tip_accounts = TipAccountResult::from(result).map_err(|e| anyhow!(e))?; Ok(tip_accounts) } - async fn send_request(&self, endpoint: &str, method: &str, params: Option) -> Result { - let url = format!("{}{}", self.base_url, endpoint); - let data = json!({ - "jsonrpc": "2.0", - "id": 1, - "method": method, - "params": params.unwrap_or(json!([])) - }); - - let response = self.client - .post(&url) - .header("Content-Type", "application/json") - .json(&data) - .send() - .await - .map_err(|e| anyhow!(format!("Request failed: {}", e)))?; - - let body = response.json::().await - .map_err(|e| anyhow!(format!("Failed to parse response: {}", e)))?; - - Ok(body) - } - pub async fn init_tip_accounts(&self) -> Result<()> { let accounts = self.get_tip_accounts().await?; let mut tip_accounts = self.tip_accounts.write().await; @@ -142,26 +81,21 @@ impl JitoClient { } 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) { + 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); })?); } } - } // 这里释放读锁 + } - // 如果账户列表为空，初始化账户列表 self.init_tip_accounts().await?; - // 重新获取读锁 let accounts = self.tip_accounts.read().await; - let mut rng = rng(); - match accounts.iter().choose(&mut rng) { + 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); })?), @@ -169,151 +103,19 @@ impl JitoClient { } } - 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 { - Err(anyhow!("jito: failed get tip")) - } - } - pub async fn send_transaction( &self, transaction: &Transaction, ) -> Result { - let wire_transaction = bincode::serialize(transaction).map_err(|e| { - anyhow!( - "Transaction serialization failed: {}", - 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 = self.uuid.clone(); - - // Send bundle using Jito SDK - let response = self.send_bundle(Some(bundle), uuid).await?; - - response["result"] - .as_str() - .map(|s| s.to_string()) - .ok_or_else(|| anyhow!("Invalid response format: missing result field")) - } - - pub async fn send_bundle(&self, params: Option, uuid: Option) -> Result { - let mut endpoint = "/api/v1/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(anyhow!("Bundle must contain at least one transaction")); - } - if transactions.len() > 5 { - return Err(anyhow!("Bundle can contain at most 5 transactions")); - } - transactions - }, - _ => return Err(anyhow!("Invalid bundle format: expected an array of transactions")), - }; - - // Wrap the transactions array in another array - let params = json!([transactions]); - - // Send the wrapped transactions array - self.send_request(&endpoint, "sendBundle", Some(params)) - .await - .map_err(|e| anyhow!(e)) - } - - 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(); - - loop { - let statuses = fetch_statuses(bundle_id.clone()).await?; - - 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, - ); - })?; - - 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); - } - _ => { - debug!("bundle_status: {:?}", bundle_status); - } - } - } else { - debug!("Finalizing bundle {}: {}", bundle_id, "None"); - } - - if start_time.elapsed() > timeout { - warn!("Loop exceeded {:?}, breaking out.", timeout); - return Err(anyhow!("Bundle status get timeout")); - } - - sleep(interval).await; - } + let bundles = vec![VersionedTransaction::from(transaction.clone())]; + 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 { diff --git a/src/jito/request.rs b/src/jito/request.rs new file mode 100644 index 0000000..6bbbfa0 --- /dev/null +++ b/src/jito/request.rs @@ -0,0 +1,55 @@ +use std::fmt; + +use serde_json::{json, Value}; + +#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy)] +pub enum RpcRequest { + Custom { method: &'static str }, + GetBundlesStatuses, + GetTipAccounts, + SendBundle, +} + +impl fmt::Display for RpcRequest { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let method = match self { + RpcRequest::Custom { method } => method, + RpcRequest::GetBundlesStatuses => "getBundleStatuses", + RpcRequest::GetTipAccounts => "getTipAccounts", + RpcRequest::SendBundle => "sendBundle", + }; + + write!(f, "{method}") + } +} + +impl RpcRequest { + pub fn build_request_json(self, id: u64, params: Value) -> Value { + let jsonrpc = "2.0"; + json!({ + "jsonrpc": jsonrpc, + "id": id, + "method": format!("{self}"), + "params": params, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_build_request_json() { + let test_request = RpcRequest::GetTipAccounts; + let request = test_request.build_request_json(1, json!([])); + assert_eq!(request["method"], "getTipAccounts"); + assert_eq!(request["params"], json!([])); + + let test_request = RpcRequest::GetBundlesStatuses; + let addr = json!("deadbeefXjn8o3yroDHxUtKsZZgoy4GPkPPXfouKNHhx"); + let request = test_request.build_request_json(1, json!([addr])); + assert_eq!(request["method"], "getBundleStatuses"); + assert_eq!(request["params"], json!([addr])); + } +} diff --git a/src/jito/rpc_client.rs b/src/jito/rpc_client.rs new file mode 100644 index 0000000..227b56b --- /dev/null +++ b/src/jito/rpc_client.rs @@ -0,0 +1,406 @@ +use std::time::Duration; + +use bincode::serialize; +use log::*; +use serde_json::{json, Value}; +use solana_rpc_client::{ + rpc_client::{RpcClientConfig, SerializableTransaction}, + rpc_sender::RpcTransportStats, +}; +use solana_rpc_client_api::{ + client_error::ErrorKind as ClientErrorKind, request::RpcError, response::Response, +}; +use solana_sdk::{bs58, commitment_config::CommitmentConfig}; +use solana_transaction_status::UiTransactionEncoding; + +use crate::jito::{ + client_error, + client_error::{Error as ClientError, Result as ClientResult}, + http_sender::HttpSender, + request::RpcRequest, + rpc_sender::*, +}; + +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, + ) -> Self { + Self { + sender: Box::new(sender), + config, + } + } + + /// 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), + RpcClientConfig::with_commitment(commitment_config), + ) + } + + /// 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), + RpcClientConfig::with_commitment(CommitmentConfig::default()), + ) + } + + /// 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], + ) -> ClientResult { + let mut serialized_encoded: Vec = Vec::with_capacity(transactions.len()); + for transaction in transactions { + 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 + { + Ok(signature_base58_str) => ClientResult::Ok(signature_base58_str), + Err(err) => { + if let ClientErrorKind::RpcError(RpcError::RpcResponseError { + code, message, .. + }) = &err.kind + { + debug!("{} {}", code, message); + } + Err(err) + } + } + } + + async fn default_cluster_transaction_encoding( + &self, + ) -> Result { + 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], + ) -> RpcResult> { + self.send(RpcRequest::GetBundlesStatuses, json!([signatures])) + .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 + } + + pub async fn send(&self, request: RpcRequest, params: Value) -> ClientResult + where + T: serde::de::DeserializeOwned, + { + assert!(params.is_array() || params.is_null()); + + let response = self + .sender + .send(request, params) + .await + .map_err(|err| err.into_with_request(request))?; + serde_json::from_value(response) + .map_err(|err| ClientError::new_with_request(err.into(), request)) + } + + pub fn get_transport_stats(&self) -> RpcTransportStats { + self.sender.get_transport_stats() + } +} + +fn serialize_and_encode(input: &T, encoding: UiTransactionEncoding) -> ClientResult +where + T: serde::ser::Serialize, +{ + let serialized = serialize(input) + .map_err(|e| ClientErrorKind::Custom(format!("Serialization failed: {e}")))?; + let encoded = match encoding { + UiTransactionEncoding::Base58 => bs58::encode(serialized).into_string(), + _ => { + return Err(ClientErrorKind::Custom(format!( + "unsupported encoding: {encoding}. Supported encodings: base58" + )) + .into()) + } + }; + 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; + use solana_sdk::{ + pubkey::Pubkey, signature::Signer, signer::keypair::Keypair, system_transaction, + transaction::VersionedTransaction, + }; + + use crate::jsonrpc_client::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( + &signer_keypair, + &signer_keypair.pubkey(), + 10000, + recent_blockhash, + ))]; + + // Add the tip + bundle.push(VersionedTransaction::from(system_transaction::transfer( + &signer_keypair, + &tip_account, + 10000, + 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/jito/rpc_sender.rs b/src/jito/rpc_sender.rs new file mode 100644 index 0000000..228a789 --- /dev/null +++ b/src/jito/rpc_sender.rs @@ -0,0 +1,21 @@ +use async_trait::async_trait; +use solana_rpc_client::rpc_sender::RpcTransportStats; + +use crate::jito::{client_error::Result, request::RpcRequest}; + +/// A transport for RPC calls. +/// +/// `RpcSender` implements the underlying transport of requests to, and +/// responses from, a Solana node, and is used primarily by [`RpcClient`]. +/// +/// [`RpcClient`]: crate::rpc_client::RpcClient +#[async_trait] +pub trait RpcSender { + async fn send( + &self, + request: RpcRequest, + params: serde_json::Value, + ) -> Result; + fn get_transport_stats(&self) -> RpcTransportStats; + fn url(&self) -> String; +}