Sasedev/khadhroony-bobobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.4.3

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-04-22 16:01:19 +02:00
parent 073266a104
commit 23dab2df85
8 changed files with 779 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
kb_lib/src/http_client.rs
View File

@@ -134,7 +134,7 @@ impl KbHttpTokenBucket {
}
}
#[derive(Debug)]
#[derive(Clone, Debug)]
enum KbHttpEndpointLifecycleState {
Active,
PausedUntil(std::time::Instant),
@@ -174,6 +174,7 @@ pub struct HttpClient {
client: reqwest::Client,
next_request_id: std::sync::Arc<std::sync::atomic::AtomicU64>,
runtime: std::sync::Arc<tokio::sync::Mutex<KbHttpRuntimeState>>,
concurrency_limiter: std::sync::Arc<tokio::sync::Semaphore>,
}
impl HttpClient {
@@ -236,6 +237,12 @@ impl HttpClient {
endpoint.name
)));
}
if endpoint.max_concurrent_requests_per_endpoint == 0 {
return Err(crate::KbError::Config(format!(
"http endpoint '{}' must have max_concurrent_requests_per_endpoint > 0",
endpoint.name
)));
}
let resolved_url_result = endpoint.resolved_url();
let resolved_url = match resolved_url_result {
Ok(resolved_url) => resolved_url,
@@ -272,6 +279,9 @@ impl HttpClient {
runtime: std::sync::Arc::new(tokio::sync::Mutex::new(KbHttpRuntimeState::new(
&endpoint,
))),
concurrency_limiter: std::sync::Arc::new(tokio::sync::Semaphore::new(
endpoint.max_concurrent_requests_per_endpoint,
)),
})
}
@@ -290,6 +300,19 @@ impl HttpClient {
&self.endpoint
}
/// Returns whether this endpoint supports the requested logical role.
pub fn supports_role(&self, required_role: &str) -> bool {
if required_role.trim().is_empty() {
return true;
}
self.endpoint.roles.iter().any(|role| role == required_role)
}
/// Returns the currently available concurrency slots for this endpoint.
pub fn available_concurrency_slots(&self) -> usize {
self.concurrency_limiter.available_permits()
}
/// Returns the next request identifier and increments the internal counter.
pub fn next_request_id(&self) -> u64 {
self.next_request_id
@@ -300,12 +323,13 @@ impl HttpClient {
pub async fn endpoint_status(&self) -> KbHttpEndpointStatus {
let mut runtime_guard = self.runtime.lock().await;
kb_http_normalize_runtime_lifecycle(&mut runtime_guard);
match &runtime_guard.lifecycle {
match runtime_guard.lifecycle.clone() {
KbHttpEndpointLifecycleState::Active => KbHttpEndpointStatus::Active,
KbHttpEndpointLifecycleState::Disabled => KbHttpEndpointStatus::Disabled,
KbHttpEndpointLifecycleState::PausedUntil(deadline) => {
let now = std::time::Instant::now();
if *deadline <= now {
if deadline <= now {
runtime_guard.lifecycle = KbHttpEndpointLifecycleState::Active;
KbHttpEndpointStatus::Active
} else {
@@ -386,6 +410,16 @@ impl HttpClient {
Ok(body) => body,
Err(error) => return Err(error),
};
let concurrency_permit_result = self.concurrency_limiter.clone().acquire_owned().await;
let _concurrency_permit = match concurrency_permit_result {
Ok(concurrency_permit) => concurrency_permit,
Err(error) => {
return Err(crate::KbError::Http(format!(
"cannot acquire concurrency slot for endpoint '{}' method '{}': {}",
self.endpoint.name, request.method, error
)));
}
};
tracing::debug!(
endpoint_name = %self.endpoint.name,
endpoint_url = %self.resolved_url,
@@ -411,6 +445,10 @@ impl HttpClient {
}
};
let status = response.status();
let retry_after_header = response
.headers()
.get(reqwest::header::RETRY_AFTER)
.cloned();
let text_result = response.text().await;
let text = match text_result {
Ok(text) => text,
@@ -422,7 +460,13 @@ impl HttpClient {
}
};
if status.as_u16() == 429 {
let pause_duration_ms = self.endpoint.pause_after_http_429_ms.unwrap_or(1500);
let pause_duration_ms = match retry_after_header
.as_ref()
.and_then(kb_http_retry_after_to_pause_ms)
{
Some(retry_after_ms) => retry_after_ms,
None => self.endpoint.pause_after_http_429_ms.unwrap_or(1500),
};
self.pause_for(pause_duration_ms).await;
return Err(crate::KbError::Http(format!(
"http status 429 returned by endpoint '{}' method '{}'; endpoint paused for {} ms; body='{}'",
@@ -750,6 +794,38 @@ impl HttpClient {
.await
}
/// Raw helper for `sendTransaction`.
pub async fn send_transaction_raw(
&self,
encoded_transaction: std::string::String,
config: std::option::Option<serde_json::Value>,
) -> Result<serde_json::Value, crate::KbError> {
let params = kb_build_first_string_optional_config_params(encoded_transaction, config);
self.execute_json_rpc_result_raw("sendTransaction".to_string(), params)
.await
}
/// Typed helper for `sendTransaction`.
pub async fn send_transaction(
&self,
encoded_transaction: std::string::String,
config: std::option::Option<solana_rpc_client_api::config::RpcSendTransactionConfig>,
) -> Result<std::string::String, crate::KbError> {
let config_value_result =
kb_serialize_optional_json_value(config, "sendTransaction config");
let config_value = match config_value_result {
Ok(config_value) => config_value,
Err(error) => return Err(error),
};
let params =
kb_build_first_string_optional_config_params(encoded_transaction, config_value);
self.execute_json_rpc_request_typed::<std::string::String>(
"sendTransaction".to_string(),
params,
)
.await
}
async fn acquire_rate_limit_slot_for_method_class(
&self,
method_class: KbHttpMethodClass,
@@ -758,7 +834,7 @@ impl HttpClient {
let wait_duration_option = {
let mut runtime_guard = self.runtime.lock().await;
kb_http_normalize_runtime_lifecycle(&mut runtime_guard);
match runtime_guard.lifecycle {
match runtime_guard.lifecycle.clone() {
KbHttpEndpointLifecycleState::Disabled => {
return Err(crate::KbError::Http(format!(
"http endpoint '{}' is disabled",
@@ -961,13 +1037,10 @@ fn kb_http_effective_burst_capacity(
}
fn kb_http_normalize_runtime_lifecycle(runtime: &mut KbHttpRuntimeState) {
match runtime.lifecycle {
KbHttpEndpointLifecycleState::PausedUntil(deadline) => {
if deadline <= std::time::Instant::now() {
runtime.lifecycle = KbHttpEndpointLifecycleState::Active;
}
if let KbHttpEndpointLifecycleState::PausedUntil(deadline) = runtime.lifecycle.clone() {
if deadline <= std::time::Instant::now() {
runtime.lifecycle = KbHttpEndpointLifecycleState::Active;
}
_ => {}
}
}
@@ -1007,6 +1080,35 @@ fn kb_http_consume_rate_limit_token(
Some(std::time::Duration::from_secs_f64(wait_seconds.max(0.001)))
}
fn kb_http_retry_after_to_pause_ms(
header_value: &reqwest::header::HeaderValue,
) -> std::option::Option<u64> {
let header_text_result = header_value.to_str();
let header_text = match header_text_result {
Ok(header_text) => header_text.trim(),
Err(_) => {
return None;
}
};
let seconds_result = header_text.parse::<u64>();
if let Ok(seconds) = seconds_result {
return Some(seconds.saturating_mul(1000));
}
let parsed_date_result = chrono::DateTime::parse_from_rfc2822(header_text);
let parsed_date = match parsed_date_result {
Ok(parsed_date) => parsed_date.with_timezone(&chrono::Utc),
Err(_) => {
return None;
}
};
let now = chrono::Utc::now();
let delta_ms = parsed_date.signed_duration_since(now).num_milliseconds();
if delta_ms <= 0 {
return Some(0);
}
Some(delta_ms as u64)
}
fn kb_http_shorten_text(input: &str, max_chars: usize) -> std::string::String {
let char_count = input.chars().count();
if char_count <= max_chars {
@@ -1082,7 +1184,7 @@ mod tests {
"id": id
}).to_string();
let response_text = format!(
"HTTP/1.1 429 TOO MANY REQUESTS\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
"HTTP/1.1 429 TOO MANY REQUESTS\r\nContent-Type: application/json\r\nRetry-After: 1\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
response_body.len(),
response_body
);
@@ -1251,7 +1353,8 @@ mod tests {
connect_timeout_ms: 2000,
request_timeout_ms: 2000,
max_idle_connections_per_host: 4,
pause_after_http_429_ms: Some(50),
pause_after_http_429_ms: Some(1500),
max_concurrent_requests_per_endpoint: 2,
}
}
@@ -1391,6 +1494,11 @@ mod tests {
.await
.expect("get_transaction must succeed");
assert!(transaction.is_none());
let sent_signature = client
.send_transaction("AAAA".to_string(), None)
.await
.expect("send_transaction must succeed");
assert_eq!(sent_signature, "signature-test".to_string());
let observed_methods = server.observed_methods_snapshot().await;
assert!(observed_methods.iter().any(|method| method == "getHealth"));
assert!(observed_methods.iter().any(|method| method == "getVersion"));
@@ -1416,6 +1524,11 @@ mod tests {
.iter()
.any(|method| method == "getTransaction")
);
assert!(
observed_methods
.iter()
.any(|method| method == "sendTransaction")
);
server.shutdown().await;
}
@@ -1424,6 +1537,7 @@ mod tests {
let server = TestHttpServer::spawn().await;
let endpoint = make_http_endpoint(server.url.clone());
let client = crate::HttpClient::new(endpoint).expect("client creation must succeed");
let _ = client
.get_health_raw()
.await
@@ -1467,6 +1581,10 @@ mod tests {
)
.await
.expect("get_transaction_raw must succeed");
let _ = client
.send_transaction_raw("AAAA".to_string(), None)
.await
.expect("send_transaction_raw must succeed");
let observed_methods = server.observed_methods_snapshot().await;
assert!(observed_methods.iter().any(|method| method == "getHealth"));
assert!(observed_methods.iter().any(|method| method == "getVersion"));
@@ -1502,6 +1620,11 @@ mod tests {
.iter()
.any(|method| method == "getTransaction")
);
assert!(
observed_methods
.iter()
.any(|method| method == "sendTransaction")
);
server.shutdown().await;
}
@@ -1523,7 +1646,7 @@ mod tests {
panic!("unexpected status after 429: {other:?}");
}
}
tokio::time::sleep(std::time::Duration::from_millis(70)).await;
tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
let resumed_status = client.endpoint_status().await;
assert_eq!(resumed_status, crate::KbHttpEndpointStatus::Active);
server.shutdown().await;