v0.2.9-pre.003
This commit is contained in:
@@ -1,80 +1,96 @@
|
||||
// file: crates/ksp-onchain-transport-lib/src/grpc_channel.rs
|
||||
// version: 2
|
||||
// version: 3
|
||||
|
||||
/// Prepared Yellowstone gRPC channel owned by KSP Transport.
|
||||
/// Prepared or connected Yellowstone gRPC channel owned by KSP Transport.
|
||||
///
|
||||
/// `pre.002` intentionally prepares a lazy HTTP/2 channel without performing network I/O. TLS configuration, provider-neutral metadata injection and live
|
||||
/// connection/error semantics are added by `pre.003`. The underlying Tonic channel and upstream Yellowstone protobuf types are never exposed publicly.
|
||||
/// The underlying Tonic channel, endpoint URL and request metadata remain private. Callers use the KSP-owned typed Yellowstone surfaces layered on this
|
||||
/// physical channel instead of receiving a raw Tonic escape hatch.
|
||||
#[derive(Clone)]
|
||||
pub struct YellowstoneGrpcChannel {
|
||||
endpoint_name: std::string::String,
|
||||
provider: crate::YellowstoneGrpcProviderName,
|
||||
cluster: crate::YellowstoneGrpcClusterName,
|
||||
_channel: tonic::transport::Channel,
|
||||
channel: tonic::transport::Channel,
|
||||
metadata: std::vec::Vec<crate::YellowstoneGrpcMetadataEntry>,
|
||||
session: crate::YellowstoneGrpcSessionSettings,
|
||||
}
|
||||
|
||||
impl YellowstoneGrpcChannel {
|
||||
/// Prepares one lazy Yellowstone gRPC channel from validated Transport-owned endpoint settings.
|
||||
/// Prepares one lazy HTTP/2 channel without establishing a network connection.
|
||||
///
|
||||
/// This operation performs no network connection. HTTPS endpoints remain syntactically accepted here; TLS is deliberately completed in `pre.003`
|
||||
/// before any live call is allowed.
|
||||
pub fn prepare(endpoint: &crate::YellowstoneGrpcEndpointSettings) -> ksp_core_lib::Result<Self> {
|
||||
if let std::result::Result::Err(error) = endpoint.validate() {
|
||||
/// HTTPS endpoints receive the KSP TLS configuration immediately, so invalid local TLS setup is rejected before a typed client is created. Tonic lazy
|
||||
/// channels require an active Tokio runtime even though no socket is opened yet.
|
||||
pub fn prepare(settings: &crate::YellowstoneGrpcEndpointSettings) -> ksp_core_lib::Result<Self> {
|
||||
if let std::result::Result::Err(error) = settings.validate() {
|
||||
return std::result::Result::Err(error);
|
||||
}
|
||||
if !endpoint.enabled() {
|
||||
if !settings.enabled() {
|
||||
return std::result::Result::Err(
|
||||
ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_SETTINGS, "disabled Yellowstone gRPC endpoint cannot prepare a channel")
|
||||
.with_context("field", "grpc_endpoint.enabled")
|
||||
.with_context("endpoint_name", endpoint.name()),
|
||||
ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, "disabled Yellowstone gRPC endpoint cannot prepare a channel")
|
||||
.with_context("endpoint_name", settings.name()),
|
||||
);
|
||||
}
|
||||
if tokio::runtime::Handle::try_current().is_err() {
|
||||
ksp_logging_lib::warn!(
|
||||
target: crate::TRACING_TARGET,
|
||||
endpoint_name = endpoint.name(),
|
||||
provider = endpoint.provider().as_str(),
|
||||
cluster = endpoint.cluster().as_str(),
|
||||
"Yellowstone gRPC channel preparation requires an active Tokio runtime"
|
||||
);
|
||||
return std::result::Result::Err(grpc_channel_error(settings, "Yellowstone gRPC channel preparation requires an active Tokio runtime"));
|
||||
}
|
||||
let endpoint = match build_tonic_endpoint(settings) {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let channel = endpoint.connect_lazy();
|
||||
ksp_logging_lib::debug!(
|
||||
target: crate::TRACING_TARGET,
|
||||
endpoint_name = settings.name(),
|
||||
provider = settings.provider().as_str(),
|
||||
cluster = settings.cluster().as_str(),
|
||||
tls = settings.url().uses_tls(),
|
||||
metadata_count = settings.metadata().len(),
|
||||
"prepared lazy Yellowstone gRPC channel"
|
||||
);
|
||||
return std::result::Result::Ok(Self::from_parts(settings, channel));
|
||||
}
|
||||
|
||||
/// Establishes one Yellowstone gRPC HTTP/2 channel with bounded connect timeout and configured TLS roots.
|
||||
pub async fn connect(settings: &crate::YellowstoneGrpcEndpointSettings) -> ksp_core_lib::Result<Self> {
|
||||
if let std::result::Result::Err(error) = settings.validate() {
|
||||
return std::result::Result::Err(error);
|
||||
}
|
||||
if !settings.enabled() {
|
||||
return std::result::Result::Err(
|
||||
ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, "Yellowstone gRPC channel requires an active Tokio runtime")
|
||||
.with_context("endpoint_name", endpoint.name()),
|
||||
ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, "disabled Yellowstone gRPC endpoint cannot connect")
|
||||
.with_context("endpoint_name", settings.name()),
|
||||
);
|
||||
}
|
||||
let tonic_endpoint = match tonic::transport::Endpoint::from_shared(endpoint.url().as_str().to_owned()) {
|
||||
if tokio::runtime::Handle::try_current().is_err() {
|
||||
return std::result::Result::Err(grpc_channel_error(settings, "Yellowstone gRPC connection requires an active Tokio runtime"));
|
||||
}
|
||||
let endpoint = match build_tonic_endpoint(settings) {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let channel = match endpoint.connect().await {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
ksp_logging_lib::warn!(
|
||||
target: crate::TRACING_TARGET,
|
||||
endpoint_name = endpoint.name(),
|
||||
provider = endpoint.provider().as_str(),
|
||||
cluster = endpoint.cluster().as_str(),
|
||||
"failed to prepare Yellowstone gRPC channel URI"
|
||||
);
|
||||
return std::result::Result::Err(
|
||||
ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, "Yellowstone gRPC channel could not be prepared")
|
||||
.with_context("endpoint_name", endpoint.name()),
|
||||
endpoint_name = settings.name(),
|
||||
provider = settings.provider().as_str(),
|
||||
cluster = settings.cluster().as_str(),
|
||||
tls = settings.url().uses_tls(),
|
||||
"Yellowstone gRPC channel connection failed"
|
||||
);
|
||||
return std::result::Result::Err(grpc_channel_error(settings, "Yellowstone gRPC channel connection failed"));
|
||||
},
|
||||
};
|
||||
let tonic_endpoint = tonic_endpoint
|
||||
.connect_timeout(endpoint.session().connect_timeout())
|
||||
.timeout(endpoint.session().unary_timeout())
|
||||
.buffer_size(endpoint.session().request_channel_capacity());
|
||||
let channel = tonic_endpoint.connect_lazy();
|
||||
ksp_logging_lib::debug!(
|
||||
target: crate::TRACING_TARGET,
|
||||
endpoint_name = endpoint.name(),
|
||||
provider = endpoint.provider().as_str(),
|
||||
cluster = endpoint.cluster().as_str(),
|
||||
"prepared lazy Yellowstone gRPC channel"
|
||||
endpoint_name = settings.name(),
|
||||
provider = settings.provider().as_str(),
|
||||
cluster = settings.cluster().as_str(),
|
||||
tls = settings.url().uses_tls(),
|
||||
"connected Yellowstone gRPC channel"
|
||||
);
|
||||
return std::result::Result::Ok(Self {
|
||||
endpoint_name: endpoint.name().to_owned(),
|
||||
provider: endpoint.provider().clone(),
|
||||
cluster: endpoint.cluster().clone(),
|
||||
_channel: channel,
|
||||
});
|
||||
return std::result::Result::Ok(Self::from_parts(settings, channel));
|
||||
}
|
||||
|
||||
/// Returns the safe logical endpoint name.
|
||||
@@ -83,17 +99,34 @@ impl YellowstoneGrpcChannel {
|
||||
return self.endpoint_name.as_str();
|
||||
}
|
||||
|
||||
/// Returns the safe open provider descriptor.
|
||||
/// Returns the open provider descriptor.
|
||||
#[must_use]
|
||||
pub const fn provider(&self) -> &crate::YellowstoneGrpcProviderName {
|
||||
return &self.provider;
|
||||
}
|
||||
|
||||
/// Returns the safe open cluster descriptor.
|
||||
/// Returns the open cluster descriptor.
|
||||
#[must_use]
|
||||
pub const fn cluster(&self) -> &crate::YellowstoneGrpcClusterName {
|
||||
return &self.cluster;
|
||||
}
|
||||
|
||||
/// Creates the standard Solana Yellowstone unary facade over this physical channel.
|
||||
#[must_use]
|
||||
pub fn standard_unary_client(&self) -> crate::SolanaYellowstoneGrpcUnaryClient {
|
||||
return crate::SolanaYellowstoneGrpcUnaryClient::new(self.channel.clone(), self.metadata.clone(), self.session.clone());
|
||||
}
|
||||
|
||||
fn from_parts(settings: &crate::YellowstoneGrpcEndpointSettings, channel: tonic::transport::Channel) -> Self {
|
||||
return Self {
|
||||
endpoint_name: settings.name().to_owned(),
|
||||
provider: settings.provider().clone(),
|
||||
cluster: settings.cluster().clone(),
|
||||
channel,
|
||||
metadata: settings.metadata().to_vec(),
|
||||
session: settings.session().clone(),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for YellowstoneGrpcChannel {
|
||||
@@ -103,10 +136,39 @@ impl std::fmt::Debug for YellowstoneGrpcChannel {
|
||||
.field("endpoint_name", &self.endpoint_name)
|
||||
.field("provider", &self.provider)
|
||||
.field("cluster", &self.cluster)
|
||||
.field("metadata_count", &self.metadata.len())
|
||||
.field("channel", &"<private>")
|
||||
.finish();
|
||||
}
|
||||
}
|
||||
|
||||
fn build_tonic_endpoint(settings: &crate::YellowstoneGrpcEndpointSettings) -> ksp_core_lib::Result<tonic::transport::Endpoint> {
|
||||
let endpoint = match tonic::transport::Endpoint::from_shared(settings.url().as_str().to_owned()) {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(grpc_channel_error(settings, "Yellowstone gRPC endpoint URI cannot be prepared by the HTTP/2 transport"));
|
||||
},
|
||||
};
|
||||
let endpoint = endpoint.connect_timeout(settings.session().connect_timeout()).buffer_size(settings.session().request_channel_capacity());
|
||||
if settings.url().uses_tls() {
|
||||
let tls = tonic::transport::ClientTlsConfig::new().with_webpki_roots().timeout(settings.session().connect_timeout());
|
||||
return match endpoint.tls_config(tls) {
|
||||
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
||||
std::result::Result::Err(_) => {
|
||||
std::result::Result::Err(grpc_channel_error(settings, "Yellowstone gRPC TLS configuration failed before connection"))
|
||||
},
|
||||
};
|
||||
}
|
||||
return std::result::Result::Ok(endpoint);
|
||||
}
|
||||
|
||||
fn grpc_channel_error(settings: &crate::YellowstoneGrpcEndpointSettings, message: &str) -> ksp_core_lib::Error {
|
||||
return ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, message)
|
||||
.with_context("endpoint_name", settings.name())
|
||||
.with_context("provider", settings.provider().as_str())
|
||||
.with_context("cluster", settings.cluster().as_str());
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "../unit_tests/grpc_channel.rs"]
|
||||
mod tests;
|
||||
|
||||
Reference in New Issue
Block a user