v0.1.0-pre.024
This commit is contained in:
279
kb-onchain-transport/src/http_pool.rs
Normal file
279
kb-onchain-transport/src/http_pool.rs
Normal file
@@ -0,0 +1,279 @@
|
||||
// file: kb-onchain-transport/src/http_pool.rs
|
||||
// version: 7
|
||||
|
||||
//! HTTP endpoint pool and role-based routing.
|
||||
|
||||
/// Pool of HTTP JSON-RPC endpoints.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct HttpEndpointPool {
|
||||
clients: std::vec::Vec<crate::HttpClient>,
|
||||
next_index: std::sync::Arc<std::sync::atomic::AtomicUsize>,
|
||||
}
|
||||
|
||||
impl crate::HttpEndpointPool {
|
||||
/// Builds a pool from the active profile HTTP endpoint list.
|
||||
pub fn from_profile(profile: &kb_config::ProfileConfig) -> kb_core::Result<Self> {
|
||||
let mut clients = std::vec::Vec::new();
|
||||
for endpoint in &profile.solana.http_endpoints {
|
||||
if !endpoint.enabled {
|
||||
continue;
|
||||
}
|
||||
let client = match crate::HttpClient::new(endpoint.clone()) {
|
||||
std::result::Result::Ok(client) => client,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
clients.push(client);
|
||||
}
|
||||
return crate::HttpEndpointPool::new(clients);
|
||||
}
|
||||
|
||||
/// Creates a pool from already constructed clients.
|
||||
pub fn new(clients: std::vec::Vec<crate::HttpClient>) -> kb_core::Result<Self> {
|
||||
if clients.is_empty() {
|
||||
tracing::error!(target: crate::TRACING_TARGET, action = "create_http_pool", error_code = "http_pool_empty", "HTTP endpoint pool has no enabled endpoint");
|
||||
return std::result::Result::Err(kb_core::Error::config(
|
||||
"http endpoint pool requires at least one enabled endpoint".to_string(),
|
||||
));
|
||||
}
|
||||
tracing::debug!(target: crate::TRACING_TARGET, action = "create_http_pool", endpoint_count = clients.len(), "HTTP endpoint pool created");
|
||||
return std::result::Result::Ok(Self {
|
||||
clients,
|
||||
next_index: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
|
||||
});
|
||||
}
|
||||
|
||||
/// Returns a serializable snapshot of every endpoint in the pool.
|
||||
pub fn snapshot(&self) -> std::vec::Vec<crate::HttpPoolClientSnapshot> {
|
||||
let mut snapshots = std::vec::Vec::new();
|
||||
for client in &self.clients {
|
||||
snapshots.push(client.snapshot());
|
||||
}
|
||||
return snapshots;
|
||||
}
|
||||
|
||||
/// Selects one endpoint for the requested role and method.
|
||||
pub fn select_client_for_role_and_method(
|
||||
&self,
|
||||
required_role: &str,
|
||||
method: &str,
|
||||
) -> kb_core::Result<crate::HttpClient> {
|
||||
let request_kind = crate::request_kind_from_method(method);
|
||||
return self.select_client_for_role_and_kind(required_role, &request_kind);
|
||||
}
|
||||
|
||||
/// Selects one endpoint for the requested role and request kind.
|
||||
pub fn select_client_for_role_and_kind(
|
||||
&self,
|
||||
required_role: &str,
|
||||
request_kind: &str,
|
||||
) -> kb_core::Result<crate::HttpClient> {
|
||||
if self.clients.is_empty() {
|
||||
return std::result::Result::Err(kb_core::Error::not_connected(
|
||||
"http endpoint pool has no clients".to_string(),
|
||||
));
|
||||
}
|
||||
let start_index = self.next_index.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
let client_count = self.clients.len();
|
||||
let mut offset = 0_usize;
|
||||
while offset < client_count {
|
||||
let index = (start_index + offset) % client_count;
|
||||
let client = self.clients[index].clone();
|
||||
if client.can_handle(required_role, request_kind) {
|
||||
tracing::debug!(target: crate::TRACING_TARGET, action = "select_http_endpoint", required_role, request_kind, endpoint_name = %client.endpoint_name(), provider = %client.provider(), "selected HTTP endpoint");
|
||||
return std::result::Result::Ok(client);
|
||||
}
|
||||
offset += 1;
|
||||
}
|
||||
tracing::error!(target: crate::TRACING_TARGET, action = "select_http_endpoint", required_role, request_kind, endpoint_count = self.clients.len(), error_code = "http_endpoint_not_found", "no HTTP endpoint supports requested role and kind");
|
||||
return std::result::Result::Err(kb_core::Error::config(format!(
|
||||
"no http endpoint supports role '{}' and request kind '{}'",
|
||||
required_role, request_kind
|
||||
)));
|
||||
}
|
||||
|
||||
/// Executes one typed standard HTTP request through an endpoint selected by role.
|
||||
pub async fn execute_standard_request_for_role<Request>(
|
||||
&self,
|
||||
required_role: &str,
|
||||
request: &Request,
|
||||
) -> kb_core::Result<<Request as crate::StandardHttpRequest>::Response>
|
||||
where
|
||||
Request: crate::StandardHttpRequest,
|
||||
{
|
||||
let client = match self.select_client_for_role_and_method(
|
||||
required_role,
|
||||
<Request as crate::StandardHttpRequest>::METHOD,
|
||||
) {
|
||||
std::result::Result::Ok(client) => client,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
return client.execute_standard_request(request).await;
|
||||
}
|
||||
|
||||
/// Executes one explicitly registered standard HTTP method through the selected endpoint.
|
||||
pub async fn execute_standard_method_raw_for_role(
|
||||
&self,
|
||||
required_role: &str,
|
||||
method: &crate::StandardHttpMethodSpec,
|
||||
params: std::vec::Vec<serde_json::Value>,
|
||||
) -> kb_core::Result<serde_json::Value> {
|
||||
let client = match self.select_client_for_role_and_method(required_role, method.method) {
|
||||
std::result::Result::Ok(client) => client,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
return client.execute_standard_method_raw(method, params).await;
|
||||
}
|
||||
|
||||
/// Executes one JSON-RPC request through the selected endpoint.
|
||||
pub async fn execute_json_rpc_result_raw_for_role(
|
||||
&self,
|
||||
required_role: &str,
|
||||
method: std::string::String,
|
||||
params: std::vec::Vec<serde_json::Value>,
|
||||
) -> kb_core::Result<serde_json::Value> {
|
||||
let client = match self.select_client_for_role_and_method(required_role, &method) {
|
||||
std::result::Result::Ok(client) => client,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
return client.execute_json_rpc_result_raw(method, params).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[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(
|
||||
name: &str,
|
||||
role: &str,
|
||||
request_kinds: std::vec::Vec<std::string::String>,
|
||||
) -> kb_config::HttpEndpointConfig {
|
||||
return kb_config::HttpEndpointConfig {
|
||||
name: name.to_string(),
|
||||
enabled: true,
|
||||
provider: "test".to_string(),
|
||||
cluster: "devnet".to_string(),
|
||||
url: format!("https://{name}.invalid"),
|
||||
connect_timeout_ms: 100,
|
||||
request_timeout_ms: 100,
|
||||
max_idle_connections_per_host: 2,
|
||||
roles: std::vec![role_config(role, request_kinds)],
|
||||
};
|
||||
}
|
||||
|
||||
fn client(endpoint: kb_config::HttpEndpointConfig) -> crate::HttpClient {
|
||||
match crate::HttpClient::new(endpoint) {
|
||||
std::result::Result::Ok(client) => return client,
|
||||
std::result::Result::Err(error) => panic!("client creation failed: {error}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn new_rejects_empty_pool() {
|
||||
let result = crate::HttpEndpointPool::new(std::vec::Vec::new());
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn snapshot_lists_every_client() {
|
||||
let pool = match crate::HttpEndpointPool::new(std::vec![
|
||||
client(endpoint("a", "http_queries", std::vec!["get_version".to_string()])),
|
||||
client(endpoint("b", "http_queries", std::vec!["get_version".to_string()])),
|
||||
]) {
|
||||
std::result::Result::Ok(pool) => pool,
|
||||
std::result::Result::Err(error) => panic!("pool creation failed: {error}"),
|
||||
};
|
||||
let snapshot = pool.snapshot();
|
||||
assert_eq!(snapshot.len(), 2);
|
||||
assert_eq!(snapshot[0].endpoint_name, "a");
|
||||
assert_eq!(snapshot[1].endpoint_name, "b");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn select_client_round_robins_matching_clients() {
|
||||
let pool = match crate::HttpEndpointPool::new(std::vec![
|
||||
client(endpoint("a", "http_queries", std::vec!["get_version".to_string()])),
|
||||
client(endpoint("b", "http_queries", std::vec!["get_version".to_string()])),
|
||||
]) {
|
||||
std::result::Result::Ok(pool) => pool,
|
||||
std::result::Result::Err(error) => panic!("pool creation failed: {error}"),
|
||||
};
|
||||
let first = match pool.select_client_for_role_and_method("http_queries", "getVersion") {
|
||||
std::result::Result::Ok(client) => client,
|
||||
std::result::Result::Err(error) => panic!("selection failed: {error}"),
|
||||
};
|
||||
let second = match pool.select_client_for_role_and_method("http_queries", "getVersion") {
|
||||
std::result::Result::Ok(client) => client,
|
||||
std::result::Result::Err(error) => panic!("selection failed: {error}"),
|
||||
};
|
||||
assert_eq!(first.endpoint_name(), "a");
|
||||
assert_eq!(second.endpoint_name(), "b");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn select_client_skips_unsupported_clients() {
|
||||
let pool = match crate::HttpEndpointPool::new(std::vec![
|
||||
client(endpoint("a", "http_heavy", std::vec!["get_block".to_string()])),
|
||||
client(endpoint("b", "http_queries", std::vec!["get_version".to_string()])),
|
||||
]) {
|
||||
std::result::Result::Ok(pool) => pool,
|
||||
std::result::Result::Err(error) => panic!("pool creation failed: {error}"),
|
||||
};
|
||||
let selected = match pool.select_client_for_role_and_method("http_queries", "getVersion") {
|
||||
std::result::Result::Ok(client) => client,
|
||||
std::result::Result::Err(error) => panic!("selection failed: {error}"),
|
||||
};
|
||||
assert_eq!(selected.endpoint_name(), "b");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn standard_method_selection_uses_the_canonical_request_kind() {
|
||||
let pool = match crate::HttpEndpointPool::new(std::vec![client(endpoint(
|
||||
"a",
|
||||
"http_queries",
|
||||
std::vec!["get_version".to_string()],
|
||||
))]) {
|
||||
std::result::Result::Ok(pool) => pool,
|
||||
std::result::Result::Err(error) => panic!("pool creation failed: {error}"),
|
||||
};
|
||||
let method = match crate::standard_http_method("getVersion") {
|
||||
std::option::Option::Some(value) => value,
|
||||
std::option::Option::None => panic!("getVersion specification missing"),
|
||||
};
|
||||
let selected = match pool.select_client_for_role_and_method("http_queries", method.method) {
|
||||
std::result::Result::Ok(client) => client,
|
||||
std::result::Result::Err(error) => panic!("selection failed: {error}"),
|
||||
};
|
||||
assert_eq!(selected.endpoint_name(), "a");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn select_client_returns_error_for_missing_role() {
|
||||
let pool = match crate::HttpEndpointPool::new(std::vec![client(endpoint(
|
||||
"a",
|
||||
"http_queries",
|
||||
std::vec!["get_version".to_string()]
|
||||
)),])
|
||||
{
|
||||
std::result::Result::Ok(pool) => pool,
|
||||
std::result::Result::Err(error) => panic!("pool creation failed: {error}"),
|
||||
};
|
||||
let selected = pool.select_client_for_role_and_kind("http_heavy", "get_block");
|
||||
assert!(selected.is_err());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user