384 lines
17 KiB
Rust
384 lines
17 KiB
Rust
// file: kb-onchain-transport/src/http_client.rs
|
|
// version: 11
|
|
|
|
//! HTTP JSON-RPC client for standard Solana RPC endpoints.
|
|
|
|
/// Local HTTP method class used for routing diagnostics.
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Serialize)]
|
|
pub enum HttpMethodClass {
|
|
/// Standard RPC reads and generic methods.
|
|
GeneralRpc,
|
|
/// Transaction submission methods.
|
|
SendTransaction,
|
|
/// Resource-intensive read methods.
|
|
HeavyRead,
|
|
}
|
|
|
|
/// Snapshot of one pooled HTTP endpoint.
|
|
#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct HttpPoolClientSnapshot {
|
|
/// Logical endpoint name.
|
|
pub endpoint_name: std::string::String,
|
|
/// Provider name.
|
|
pub provider: std::string::String,
|
|
/// Endpoint URL.
|
|
pub endpoint_url: std::string::String,
|
|
/// Supported roles.
|
|
pub roles: std::vec::Vec<crate::EndpointRoleSnapshot>,
|
|
/// Status string.
|
|
pub status: std::string::String,
|
|
}
|
|
|
|
/// HTTP JSON-RPC client bound to one configured endpoint.
|
|
#[derive(Clone, Debug)]
|
|
pub struct HttpClient {
|
|
endpoint: kb_config::HttpEndpointConfig,
|
|
client: reqwest::Client,
|
|
next_request_id: std::sync::Arc<std::sync::atomic::AtomicU64>,
|
|
}
|
|
|
|
impl crate::HttpClient {
|
|
/// Creates a new HTTP client bound to one endpoint.
|
|
pub fn new(endpoint: kb_config::HttpEndpointConfig) -> kb_core::Result<Self> {
|
|
if !endpoint.enabled {
|
|
tracing::error!(target: crate::TRACING_TARGET, action = "create_http_client", endpoint_name = %endpoint.name, provider = %endpoint.provider, error_code = "http_endpoint_disabled", "cannot create HTTP client for disabled endpoint");
|
|
return std::result::Result::Err(kb_core::Error::config(format!(
|
|
"http endpoint '{}' is disabled",
|
|
endpoint.name
|
|
)));
|
|
}
|
|
let timeout = std::time::Duration::from_millis(endpoint.request_timeout_ms);
|
|
let connect_timeout = std::time::Duration::from_millis(endpoint.connect_timeout_ms);
|
|
let client_result = reqwest::Client::builder()
|
|
.timeout(timeout)
|
|
.connect_timeout(connect_timeout)
|
|
.pool_max_idle_per_host(endpoint.max_idle_connections_per_host as usize)
|
|
.build();
|
|
let client = match client_result {
|
|
std::result::Result::Ok(client) => client,
|
|
std::result::Result::Err(error) => {
|
|
tracing::error!(target: crate::TRACING_TARGET, action = "create_http_client", endpoint_name = %endpoint.name, provider = %endpoint.provider, error = %error, "HTTP client construction failed");
|
|
return std::result::Result::Err(kb_core::Error::http(format!(
|
|
"cannot build http client for endpoint '{}': {error}",
|
|
endpoint.name
|
|
)));
|
|
},
|
|
};
|
|
tracing::debug!(target: crate::TRACING_TARGET, action = "create_http_client", endpoint_name = %endpoint.name, provider = %endpoint.provider, role_count = endpoint.roles.len(), "HTTP client created");
|
|
return std::result::Result::Ok(Self {
|
|
endpoint,
|
|
client,
|
|
next_request_id: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)),
|
|
});
|
|
}
|
|
|
|
/// Returns the endpoint name.
|
|
pub fn endpoint_name(&self) -> &str {
|
|
return self.endpoint.name.as_str();
|
|
}
|
|
|
|
/// Returns the provider name.
|
|
pub fn provider(&self) -> &str {
|
|
return self.endpoint.provider.as_str();
|
|
}
|
|
|
|
/// Returns the endpoint URL.
|
|
pub fn endpoint_url(&self) -> &str {
|
|
return self.endpoint.url.as_str();
|
|
}
|
|
|
|
/// Returns the endpoint configuration.
|
|
pub fn endpoint_config(&self) -> &kb_config::HttpEndpointConfig {
|
|
return &self.endpoint;
|
|
}
|
|
|
|
/// Returns true when this endpoint supports the required role and request kind.
|
|
pub fn can_handle(&self, required_role: &str, request_kind: &str) -> bool {
|
|
if !self.endpoint.enabled {
|
|
return false;
|
|
}
|
|
for role in &self.endpoint.roles {
|
|
if crate::role_matches(role, required_role, request_kind) {
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/// Returns a serializable endpoint snapshot.
|
|
pub fn snapshot(&self) -> crate::HttpPoolClientSnapshot {
|
|
let mut roles = std::vec::Vec::new();
|
|
for role in &self.endpoint.roles {
|
|
roles.push(crate::EndpointRoleSnapshot::from_config(role));
|
|
}
|
|
return crate::HttpPoolClientSnapshot {
|
|
endpoint_name: self.endpoint.name.clone(),
|
|
provider: self.endpoint.provider.clone(),
|
|
endpoint_url: self.endpoint.url.clone(),
|
|
roles,
|
|
status: "active".to_string(),
|
|
};
|
|
}
|
|
|
|
/// Classifies a Solana HTTP method into a broad local class.
|
|
pub fn classify_method(method: &str) -> crate::HttpMethodClass {
|
|
let standard = crate::standard_http_method(method);
|
|
return match standard {
|
|
std::option::Option::Some(specification) => specification.method_class(),
|
|
std::option::Option::None => crate::HttpMethodClass::GeneralRpc,
|
|
};
|
|
}
|
|
|
|
/// Executes one typed standard HTTP request and decodes its method-specific result.
|
|
pub async fn execute_standard_request<Request>(
|
|
&self,
|
|
request: &Request,
|
|
) -> kb_core::Result<<Request as crate::StandardHttpRequest>::Response>
|
|
where
|
|
Request: crate::StandardHttpRequest,
|
|
{
|
|
let specification =
|
|
match crate::standard_http_method(<Request as crate::StandardHttpRequest>::METHOD) {
|
|
std::option::Option::Some(specification) => specification,
|
|
std::option::Option::None => {
|
|
return std::result::Result::Err(kb_core::Error::invalid_state(format!(
|
|
"typed standard request '{}' is absent from the canonical registry",
|
|
<Request as crate::StandardHttpRequest>::METHOD
|
|
)));
|
|
},
|
|
};
|
|
if specification.contract != crate::StandardRpcContract::TypedAdapter {
|
|
return std::result::Result::Err(kb_core::Error::invalid_state(format!(
|
|
"standard request '{}' is not declared as a typed adapter",
|
|
<Request as crate::StandardHttpRequest>::METHOD
|
|
)));
|
|
}
|
|
let params = match request.params() {
|
|
std::result::Result::Ok(params) => params,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let raw = match self
|
|
.execute_json_rpc_result_raw(
|
|
<Request as crate::StandardHttpRequest>::METHOD.to_string(),
|
|
params,
|
|
)
|
|
.await
|
|
{
|
|
std::result::Result::Ok(raw) => raw,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return match serde_json::from_value::<<Request as crate::StandardHttpRequest>::Response>(
|
|
raw,
|
|
) {
|
|
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
|
std::result::Result::Err(error) => {
|
|
std::result::Result::Err(kb_core::Error::json(format!(
|
|
"cannot decode standard JSON-RPC result for '{}': {error}",
|
|
<Request as crate::StandardHttpRequest>::METHOD
|
|
)))
|
|
},
|
|
};
|
|
}
|
|
|
|
/// Executes one explicitly registered standard HTTP method and returns its raw result.
|
|
pub async fn execute_standard_method_raw(
|
|
&self,
|
|
method: &crate::StandardHttpMethodSpec,
|
|
params: std::vec::Vec<serde_json::Value>,
|
|
) -> kb_core::Result<serde_json::Value> {
|
|
return self.execute_json_rpc_result_raw(method.method.to_string(), params).await;
|
|
}
|
|
|
|
/// Executes one JSON-RPC request and returns the raw result value.
|
|
pub async fn execute_json_rpc_result_raw(
|
|
&self,
|
|
method: std::string::String,
|
|
params: std::vec::Vec<serde_json::Value>,
|
|
) -> kb_core::Result<serde_json::Value> {
|
|
let request_id = self.next_request_id.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
|
let parameter_count = params.len();
|
|
let method_class = crate::HttpClient::classify_method(method.as_str());
|
|
tracing::debug!(target: crate::TRACING_TARGET, action = "execute_http_json_rpc", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id, method = %method, method_class = ?method_class, parameter_count, "send HTTP JSON-RPC request");
|
|
let request = crate::JsonRpcRequest::new_with_u64_id(request_id, method.clone(), params);
|
|
let response_result =
|
|
self.client.post(self.endpoint.url.as_str()).json(&request).send().await;
|
|
let response = match response_result {
|
|
std::result::Result::Ok(response) => response,
|
|
std::result::Result::Err(error) => {
|
|
tracing::error!(target: crate::TRACING_TARGET, action = "execute_http_json_rpc", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id, method = %method, error = %error, "HTTP JSON-RPC transport failed");
|
|
return std::result::Result::Err(kb_core::Error::http(format!(
|
|
"http json-rpc request '{}' failed on endpoint '{}': {error}",
|
|
method, self.endpoint.name
|
|
)));
|
|
},
|
|
};
|
|
let status = response.status();
|
|
let text_result = response.text().await;
|
|
let text = match text_result {
|
|
std::result::Result::Ok(text) => text,
|
|
std::result::Result::Err(error) => {
|
|
tracing::error!(target: crate::TRACING_TARGET, action = "execute_http_json_rpc", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id, method = %method, http_status = %status, error = %error, "HTTP JSON-RPC response body read failed");
|
|
return std::result::Result::Err(kb_core::Error::http(format!(
|
|
"cannot read http json-rpc response '{}' from endpoint '{}': {error}",
|
|
method, self.endpoint.name
|
|
)));
|
|
},
|
|
};
|
|
if !status.is_success() {
|
|
tracing::error!(target: crate::TRACING_TARGET, action = "execute_http_json_rpc", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id, method = %method, http_status = %status, response_byte_length = text.len(), "HTTP JSON-RPC endpoint returned non-success status");
|
|
return std::result::Result::Err(kb_core::Error::http(format!(
|
|
"http json-rpc endpoint '{}' returned status {}: {}",
|
|
self.endpoint.name, status, text
|
|
)));
|
|
}
|
|
let parsed = match crate::parse_json_rpc_text(&text) {
|
|
std::result::Result::Ok(parsed) => parsed,
|
|
std::result::Result::Err(error) => {
|
|
tracing::error!(target: crate::TRACING_TARGET, action = "execute_http_json_rpc", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id, method = %method, http_status = %status, response_byte_length = text.len(), error = %error, "HTTP JSON-RPC response parsing failed");
|
|
return std::result::Result::Err(error);
|
|
},
|
|
};
|
|
return match parsed {
|
|
crate::JsonRpcResponse::Success(success) => {
|
|
tracing::debug!(target: crate::TRACING_TARGET, action = "execute_http_json_rpc", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id, method = %method, http_status = %status, response_byte_length = text.len(), outcome = "success", "HTTP JSON-RPC request completed");
|
|
std::result::Result::Ok(success.result)
|
|
},
|
|
crate::JsonRpcResponse::Error(error_response) => {
|
|
tracing::error!(target: crate::TRACING_TARGET, action = "execute_http_json_rpc", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id, method = %method, rpc_error_code = error_response.error.code, rpc_error_message = %error_response.error.message, "HTTP JSON-RPC endpoint returned an RPC error");
|
|
std::result::Result::Err(kb_core::Error::http(format!(
|
|
"json-rpc error {} from '{}': {}",
|
|
error_response.error.code, self.endpoint.name, error_response.error.message
|
|
)))
|
|
},
|
|
crate::JsonRpcResponse::Notification(_) => {
|
|
tracing::error!(target: crate::TRACING_TARGET, action = "execute_http_json_rpc", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id, method = %method, "HTTP JSON-RPC response was an unexpected notification");
|
|
std::result::Result::Err(kb_core::Error::http(
|
|
"http json-rpc response cannot be a notification".to_string(),
|
|
))
|
|
},
|
|
};
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
fn role_config(
|
|
role: &str,
|
|
request_kinds: std::vec::Vec<std::string::String>,
|
|
) -> kb_config::EndpointRoleConfig {
|
|
return kb_config::EndpointRoleConfig {
|
|
role: role.to_string(),
|
|
enabled: true,
|
|
request_kinds,
|
|
priority: 1,
|
|
requests_per_second: 10,
|
|
burst_capacity: 10,
|
|
max_concurrent_requests: 4,
|
|
max_subscriptions: 16,
|
|
pause_after_rate_limit_ms: 1500,
|
|
};
|
|
}
|
|
|
|
fn endpoint(enabled: bool) -> kb_config::HttpEndpointConfig {
|
|
return kb_config::HttpEndpointConfig {
|
|
name: "http_a".to_string(),
|
|
enabled,
|
|
provider: "test".to_string(),
|
|
cluster: "devnet".to_string(),
|
|
url: "https://example.invalid".to_string(),
|
|
connect_timeout_ms: 100,
|
|
request_timeout_ms: 100,
|
|
max_idle_connections_per_host: 2,
|
|
roles: std::vec![
|
|
role_config("http_queries", std::vec!["get_version".to_string()]),
|
|
role_config("http_heavy", std::vec!["get_block".to_string()]),
|
|
role_config("http_any", std::vec!["*".to_string()]),
|
|
],
|
|
};
|
|
}
|
|
|
|
#[test]
|
|
fn new_rejects_disabled_endpoint() {
|
|
let result = crate::HttpClient::new(endpoint(false));
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn can_handle_matches_exact_role_and_kind() {
|
|
let client = match crate::HttpClient::new(endpoint(true)) {
|
|
std::result::Result::Ok(client) => client,
|
|
std::result::Result::Err(error) => panic!("client creation failed: {error}"),
|
|
};
|
|
assert!(client.can_handle("http_queries", "get_version"));
|
|
assert!(!client.can_handle("http_queries", "get_block"));
|
|
}
|
|
|
|
#[test]
|
|
fn can_handle_matches_wildcard_kind() {
|
|
let client = match crate::HttpClient::new(endpoint(true)) {
|
|
std::result::Result::Ok(client) => client,
|
|
std::result::Result::Err(error) => panic!("client creation failed: {error}"),
|
|
};
|
|
assert!(client.can_handle("http_any", "send_transaction"));
|
|
}
|
|
|
|
#[test]
|
|
fn snapshot_preserves_endpoint_metadata() {
|
|
let client = match crate::HttpClient::new(endpoint(true)) {
|
|
std::result::Result::Ok(client) => client,
|
|
std::result::Result::Err(error) => panic!("client creation failed: {error}"),
|
|
};
|
|
let snapshot = client.snapshot();
|
|
assert_eq!(snapshot.endpoint_name, "http_a");
|
|
assert_eq!(snapshot.provider, "test");
|
|
assert_eq!(snapshot.endpoint_url, "https://example.invalid");
|
|
assert_eq!(snapshot.roles.len(), 3);
|
|
}
|
|
|
|
#[test]
|
|
fn classify_method_detects_transaction_submission() {
|
|
for method in ["requestAirdrop", "sendTransaction"] {
|
|
assert_eq!(
|
|
crate::HttpClient::classify_method(method),
|
|
crate::HttpMethodClass::SendTransaction
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn classify_method_detects_heavy_reads() {
|
|
assert_eq!(
|
|
crate::HttpClient::classify_method("getBlock"),
|
|
crate::HttpMethodClass::HeavyRead
|
|
);
|
|
assert_eq!(
|
|
crate::HttpClient::classify_method("getProgramAccounts"),
|
|
crate::HttpMethodClass::HeavyRead
|
|
);
|
|
assert_eq!(
|
|
crate::HttpClient::classify_method("getSignaturesForAddress"),
|
|
crate::HttpMethodClass::HeavyRead
|
|
);
|
|
assert_eq!(
|
|
crate::HttpClient::classify_method("simulateTransaction"),
|
|
crate::HttpMethodClass::HeavyRead
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn every_standard_method_uses_its_declared_routing_class() {
|
|
for method in &crate::STANDARD_HTTP_METHODS {
|
|
assert_eq!(crate::HttpClient::classify_method(method.method), method.method_class());
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn classify_method_defaults_to_general_rpc() {
|
|
assert_eq!(
|
|
crate::HttpClient::classify_method("getVersion"),
|
|
crate::HttpMethodClass::GeneralRpc
|
|
);
|
|
}
|
|
}
|