233 lines
11 KiB
Rust
233 lines
11 KiB
Rust
// file: crates/ksp-onchain-transport-lib/src/executor.rs
|
|
// version: 2
|
|
|
|
const HTTP_REQUEST_TIMEOUT: u16 = 408;
|
|
const HTTP_TOO_MANY_REQUESTS: u16 = 429;
|
|
const HTTP_INTERNAL_SERVER_ERROR: u16 = 500;
|
|
const HTTP_BAD_GATEWAY: u16 = 502;
|
|
const HTTP_SERVICE_UNAVAILABLE: u16 = 503;
|
|
const HTTP_GATEWAY_TIMEOUT: u16 = 504;
|
|
|
|
impl crate::HttpTransportPool {
|
|
/// Executes one audited standard Solana HTTP JSON-RPC method through KSP routing, admission and bounded retry policy.
|
|
///
|
|
/// This generic transport surface intentionally returns the raw JSON result. Typed method coverage remains explicit and is provided separately by
|
|
/// method-specific KSP adapters.
|
|
pub async fn execute_standard_rpc(
|
|
&self,
|
|
role: &crate::HttpRoleName,
|
|
method: &crate::HttpRpcMethodDescriptor,
|
|
params: std::vec::Vec<serde_json::Value>,
|
|
) -> ksp_core_lib::Result<serde_json::Value> {
|
|
let support = method.ensure_runtime_supported();
|
|
if let std::result::Result::Err(error) = support {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let request_id = self.next_request_id();
|
|
let request_result = crate::JsonRpcRequest::new(request_id, method.method(), params);
|
|
let request = match request_result {
|
|
std::result::Result::Ok(request) => request,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let payload_result = request.to_json_string();
|
|
let payload = match payload_result {
|
|
std::result::Result::Ok(payload) => payload,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let request_kind = crate::HttpRequestKind::new(method.request_kind());
|
|
let timeout_result = self.common_request_timeout(role, &request_kind);
|
|
let timeout = match timeout_result {
|
|
std::result::Result::Ok(timeout) => timeout,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let started = std::time::Instant::now();
|
|
let deadline = match started.checked_add(timeout) {
|
|
std::option::Option::Some(deadline) => deadline,
|
|
std::option::Option::None => return execution_timeout(method, "HTTP JSON-RPC execution deadline could not be represented"),
|
|
};
|
|
let mut completed_retries = 0_u32;
|
|
loop {
|
|
let remaining = remaining_budget(deadline);
|
|
if remaining.is_zero() {
|
|
return execution_timeout(method, "HTTP JSON-RPC execution deadline expired before transport attempt");
|
|
}
|
|
let permit_result = self.acquire_for_request_kind_with_timeout(role, &request_kind, remaining).await;
|
|
let permit = match permit_result {
|
|
std::result::Result::Ok(permit) => permit,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let send_timeout = std::cmp::min(remaining_budget(deadline), permit.client().request_timeout());
|
|
let response_result = permit.client().post_json_rpc(payload.as_str(), send_timeout).await;
|
|
let response = match response_result {
|
|
std::result::Result::Ok(response) => response,
|
|
std::result::Result::Err(error) => {
|
|
permit.record_failure();
|
|
let cause = retry_cause_for_error(&error);
|
|
let dispatch_state = dispatch_state_for_error(&error);
|
|
let decision =
|
|
crate::evaluate_transport_retry(method, self.retry_settings(), cause, dispatch_state, completed_retries, std::option::Option::None);
|
|
drop(permit);
|
|
if let std::option::Option::Some(delay) = decision.delay() {
|
|
let waited = wait_retry_delay(delay, deadline).await;
|
|
if waited {
|
|
completed_retries = completed_retries.saturating_add(1);
|
|
continue;
|
|
}
|
|
return execution_timeout(method, "HTTP JSON-RPC retry delay exceeded the common request deadline");
|
|
}
|
|
return std::result::Result::Err(error);
|
|
},
|
|
};
|
|
let status = response.status();
|
|
if status == HTTP_TOO_MANY_REQUESTS {
|
|
let provider_retry_after = response.retry_after();
|
|
permit.record_rate_limited(provider_retry_after);
|
|
let decision = crate::evaluate_transport_retry(
|
|
method,
|
|
self.retry_settings(),
|
|
crate::HttpRetryCause::RateLimited,
|
|
crate::HttpDispatchState::DispatchedAmbiguous,
|
|
completed_retries,
|
|
provider_retry_after,
|
|
);
|
|
drop(permit);
|
|
if let std::option::Option::Some(delay) = decision.delay() {
|
|
let waited = wait_retry_delay(delay, deadline).await;
|
|
if waited {
|
|
completed_retries = completed_retries.saturating_add(1);
|
|
continue;
|
|
}
|
|
return execution_timeout(method, "HTTP JSON-RPC rate-limit retry exceeded the common request deadline");
|
|
}
|
|
return rate_limited_error(method, provider_retry_after);
|
|
}
|
|
if is_temporary_http_status(status) {
|
|
permit.record_failure();
|
|
let decision = crate::evaluate_transport_retry(
|
|
method,
|
|
self.retry_settings(),
|
|
crate::HttpRetryCause::TemporaryHttp,
|
|
crate::HttpDispatchState::DispatchedAmbiguous,
|
|
completed_retries,
|
|
std::option::Option::None,
|
|
);
|
|
drop(permit);
|
|
if let std::option::Option::Some(delay) = decision.delay() {
|
|
let waited = wait_retry_delay(delay, deadline).await;
|
|
if waited {
|
|
completed_retries = completed_retries.saturating_add(1);
|
|
continue;
|
|
}
|
|
return execution_timeout(method, "HTTP JSON-RPC temporary-status retry exceeded the common request deadline");
|
|
}
|
|
return http_status_error(method, status);
|
|
}
|
|
if !(200..300).contains(&status) {
|
|
permit.record_success();
|
|
return http_status_error(method, status);
|
|
}
|
|
let response_text = match std::str::from_utf8(response.body()) {
|
|
std::result::Result::Ok(text) => text,
|
|
std::result::Result::Err(error) => {
|
|
permit.record_failure();
|
|
return std::result::Result::Err(
|
|
ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "HTTP JSON-RPC response body is not valid UTF-8")
|
|
.with_context("rpc_method", method.method())
|
|
.with_source(error),
|
|
);
|
|
},
|
|
};
|
|
let parsed_result = crate::parse_json_rpc_response_text(response_text, request_id);
|
|
let parsed = match parsed_result {
|
|
std::result::Result::Ok(parsed) => parsed,
|
|
std::result::Result::Err(error) => {
|
|
permit.record_failure();
|
|
return std::result::Result::Err(error);
|
|
},
|
|
};
|
|
permit.record_success();
|
|
ksp_logging_lib::debug!(
|
|
target: crate::TRACING_TARGET,
|
|
endpoint_name = permit.selection().endpoint_name(),
|
|
role = permit.selection().role().as_str(),
|
|
rpc_method = method.method(),
|
|
request_id,
|
|
completed_retries,
|
|
http_status = status,
|
|
"completed Solana HTTP JSON-RPC request"
|
|
);
|
|
return parsed.into_result();
|
|
}
|
|
}
|
|
}
|
|
|
|
fn retry_cause_for_error(error: &ksp_core_lib::Error) -> crate::HttpRetryCause {
|
|
if error.code() == crate::ERROR_CODE_HTTP_CONNECTION_FAILED {
|
|
return crate::HttpRetryCause::Connection;
|
|
}
|
|
if error.code() == crate::ERROR_CODE_TIMEOUT {
|
|
return crate::HttpRetryCause::Timeout;
|
|
}
|
|
return crate::HttpRetryCause::Request;
|
|
}
|
|
|
|
fn dispatch_state_for_error(error: &ksp_core_lib::Error) -> crate::HttpDispatchState {
|
|
if error.code() == crate::ERROR_CODE_HTTP_CONNECTION_FAILED {
|
|
return crate::HttpDispatchState::NotDispatched;
|
|
}
|
|
return crate::HttpDispatchState::DispatchedAmbiguous;
|
|
}
|
|
|
|
const fn is_temporary_http_status(status: u16) -> bool {
|
|
return status == HTTP_REQUEST_TIMEOUT
|
|
|| status == HTTP_INTERNAL_SERVER_ERROR
|
|
|| status == HTTP_BAD_GATEWAY
|
|
|| status == HTTP_SERVICE_UNAVAILABLE
|
|
|| status == HTTP_GATEWAY_TIMEOUT;
|
|
}
|
|
|
|
fn remaining_budget(deadline: std::time::Instant) -> std::time::Duration {
|
|
let now = std::time::Instant::now();
|
|
if now >= deadline {
|
|
return std::time::Duration::ZERO;
|
|
}
|
|
return deadline.duration_since(now);
|
|
}
|
|
|
|
async fn wait_retry_delay(delay: std::time::Duration, deadline: std::time::Instant) -> bool {
|
|
let remaining = remaining_budget(deadline);
|
|
if remaining.is_zero() || delay >= remaining {
|
|
return false;
|
|
}
|
|
tokio::time::sleep(delay).await;
|
|
return std::time::Instant::now() < deadline;
|
|
}
|
|
|
|
fn execution_timeout(method: &crate::HttpRpcMethodDescriptor, message: &str) -> ksp_core_lib::Result<serde_json::Value> {
|
|
return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_TIMEOUT, message).with_context("rpc_method", method.method()));
|
|
}
|
|
|
|
fn rate_limited_error(
|
|
method: &crate::HttpRpcMethodDescriptor,
|
|
provider_retry_after: std::option::Option<std::time::Duration>,
|
|
) -> ksp_core_lib::Result<serde_json::Value> {
|
|
let mut error = ksp_core_lib::Error::new(crate::ERROR_CODE_RATE_LIMITED, "Solana HTTP endpoint rate-limited the JSON-RPC request")
|
|
.with_context("rpc_method", method.method());
|
|
if let std::option::Option::Some(delay) = provider_retry_after {
|
|
error = error.with_context("retry_after_seconds", delay.as_secs().to_string());
|
|
}
|
|
return std::result::Result::Err(error);
|
|
}
|
|
|
|
fn http_status_error(method: &crate::HttpRpcMethodDescriptor, status: u16) -> ksp_core_lib::Result<serde_json::Value> {
|
|
return std::result::Result::Err(
|
|
ksp_core_lib::Error::new(crate::ERROR_CODE_HTTP_REQUEST_FAILED, "Solana HTTP endpoint returned an unsuccessful status")
|
|
.with_context("rpc_method", method.method())
|
|
.with_context("http_status", status.to_string()),
|
|
);
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "../unit_tests/executor.rs"]
|
|
mod tests;
|