1072 lines
48 KiB
Rust
1072 lines
48 KiB
Rust
// file: crates/ksp-config-lib/src/transport.rs
|
|
// version: 6
|
|
|
|
/// Effective standard on-chain Transport configuration resolved from Config and mapped to HTTP plus optional WebSocket and Yellowstone gRPC runtime contracts.
|
|
#[derive(Clone, Eq, PartialEq)]
|
|
pub struct ResolvedTransportConfig {
|
|
file_id: crate::ConfigFileId,
|
|
source_path: std::path::PathBuf,
|
|
profile_id: String,
|
|
selection_source: crate::ConfigProfileSelectionSource,
|
|
effective: crate::ResolvedConfigJson,
|
|
settings: ksp_onchain_transport_lib::HttpTransportSettings,
|
|
ws_settings: std::option::Option<ksp_onchain_transport_lib::WsTransportSettings>,
|
|
grpc_settings: std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcTransportSettings>,
|
|
}
|
|
|
|
impl ResolvedTransportConfig {
|
|
/// Returns the logical Config document identifier used by this runtime configuration.
|
|
#[must_use]
|
|
pub const fn file_id(&self) -> &crate::ConfigFileId {
|
|
return &self.file_id;
|
|
}
|
|
|
|
/// Returns the physical source Config document path.
|
|
#[must_use]
|
|
pub fn source_path(&self) -> &std::path::Path {
|
|
return self.source_path.as_path();
|
|
}
|
|
|
|
/// Returns the selected standard Transport profile identifier.
|
|
#[must_use]
|
|
pub fn profile_id(&self) -> &str {
|
|
return self.profile_id.as_str();
|
|
}
|
|
|
|
/// Returns the source that selected the standard Transport profile.
|
|
#[must_use]
|
|
pub const fn selection_source(&self) -> crate::ConfigProfileSelectionSource {
|
|
return self.selection_source;
|
|
}
|
|
|
|
/// Returns the detailed environment-resolved effective Config view.
|
|
///
|
|
/// The real tree is available to legitimate runtime consumers. The safe tree redacts values originating from `KSP_SECRET_*` or `KSPB_SECRET_*`
|
|
/// placeholders and is the only representation used by this type's [`std::fmt::Debug`] implementation.
|
|
#[must_use]
|
|
pub const fn effective(&self) -> &crate::ResolvedConfigJson {
|
|
return &self.effective;
|
|
}
|
|
|
|
/// Returns the validated runtime HTTP Transport settings.
|
|
///
|
|
/// This compatibility accessor keeps the HTTP contract introduced before Transport V2.
|
|
#[must_use]
|
|
pub const fn settings(&self) -> &ksp_onchain_transport_lib::HttpTransportSettings {
|
|
return &self.settings;
|
|
}
|
|
|
|
/// Returns the validated runtime HTTP Transport settings.
|
|
#[must_use]
|
|
pub const fn http_settings(&self) -> &ksp_onchain_transport_lib::HttpTransportSettings {
|
|
return &self.settings;
|
|
}
|
|
|
|
/// Returns validated WebSocket Transport settings when the selected document uses format V2 or V3.
|
|
///
|
|
/// Backward-compatible V1 HTTP-only documents return [`std::option::Option::None`].
|
|
#[must_use]
|
|
pub fn ws_settings(&self) -> std::option::Option<&ksp_onchain_transport_lib::WsTransportSettings> {
|
|
return self.ws_settings.as_ref();
|
|
}
|
|
|
|
/// Returns validated Yellowstone gRPC Transport settings when the selected V3 profile declares gRPC endpoints.
|
|
///
|
|
/// V1/V2 documents and V3 profiles without `grpc_endpoints` return [`std::option::Option::None`].
|
|
#[must_use]
|
|
pub fn grpc_settings(&self) -> std::option::Option<&ksp_onchain_transport_lib::YellowstoneGrpcTransportSettings> {
|
|
return self.grpc_settings.as_ref();
|
|
}
|
|
|
|
/// Consumes this resolved Config and returns the mapped runtime HTTP Transport settings.
|
|
#[must_use]
|
|
pub fn into_settings(self) -> ksp_onchain_transport_lib::HttpTransportSettings {
|
|
return self.settings;
|
|
}
|
|
|
|
/// Consumes this resolved Config and returns HTTP plus optional WebSocket runtime settings.
|
|
///
|
|
/// This compatibility accessor intentionally preserves the V2 tuple shape. Use [`Self::into_all_transport_settings`] when Yellowstone gRPC settings are
|
|
/// also required.
|
|
#[must_use]
|
|
pub fn into_transport_settings(
|
|
self,
|
|
) -> (ksp_onchain_transport_lib::HttpTransportSettings, std::option::Option<ksp_onchain_transport_lib::WsTransportSettings>) {
|
|
return (self.settings, self.ws_settings);
|
|
}
|
|
|
|
/// Consumes this resolved Config and returns HTTP plus optional WebSocket and Yellowstone gRPC runtime settings.
|
|
#[must_use]
|
|
pub fn into_all_transport_settings(
|
|
self,
|
|
) -> (
|
|
ksp_onchain_transport_lib::HttpTransportSettings,
|
|
std::option::Option<ksp_onchain_transport_lib::WsTransportSettings>,
|
|
std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcTransportSettings>,
|
|
) {
|
|
return (self.settings, self.ws_settings, self.grpc_settings);
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Debug for ResolvedTransportConfig {
|
|
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
return formatter
|
|
.debug_struct("ResolvedTransportConfig")
|
|
.field("file_id", &self.file_id)
|
|
.field("source_path", &self.source_path)
|
|
.field("profile_id", &self.profile_id)
|
|
.field("selection_source", &self.selection_source)
|
|
.field("effective", &self.effective)
|
|
.field("has_ws_settings", &self.ws_settings.is_some())
|
|
.field("has_grpc_settings", &self.grpc_settings.is_some())
|
|
.finish_non_exhaustive();
|
|
}
|
|
}
|
|
|
|
impl crate::ConfigDocumentEngine {
|
|
/// Loads the standard Transport document, selects a profile, resolves environment placeholders and maps HTTP plus optional WebSocket and Yellowstone
|
|
/// gRPC runtime settings.
|
|
///
|
|
/// `requested_profile = None` uses the document `default_profile`; `Some(profile_id)` requests an explicit profile. Secret endpoint URLs are allowed
|
|
/// because Transport URL wrappers own runtime redaction. V1 remains HTTP-only, V2 adds WebSocket and V3 can additionally map Yellowstone gRPC.
|
|
pub fn load_resolved_transport_config(
|
|
&self,
|
|
requested_profile: std::option::Option<&str>,
|
|
environment: &crate::ConfigEnvironment,
|
|
) -> ksp_core_lib::Result<ResolvedTransportConfig> {
|
|
let file_id = crate::ConfigFileId::new(crate::FILE_ID_STD_TRANSPORT);
|
|
let file_id = match file_id {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let profile = self.load_resolved_profile(&file_id, requested_profile);
|
|
let profile = match profile {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
return resolve_transport_profile(&profile, environment);
|
|
}
|
|
|
|
/// Maps an already resolved standard Transport profile to HTTP plus optional WebSocket and Yellowstone gRPC runtime adapters while preserving
|
|
/// selection provenance.
|
|
///
|
|
/// This entry point is intended for profiles selected by a composite. The profile must reference `cfg.std.transport`.
|
|
pub fn resolve_transport_config_profile(
|
|
&self,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
environment: &crate::ConfigEnvironment,
|
|
) -> ksp_core_lib::Result<ResolvedTransportConfig> {
|
|
if profile.file_id().as_str() != crate::FILE_ID_STD_TRANSPORT {
|
|
return std::result::Result::Err(effective_error(profile, "resolved Config profile does not reference the standard Transport document"));
|
|
}
|
|
let descriptor = self.registry().descriptor(profile.file_id());
|
|
if let std::result::Result::Err(error) = descriptor {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
return resolve_transport_profile(profile, environment);
|
|
}
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveTransportSource {
|
|
format_version: u32,
|
|
profile_id: String,
|
|
retry: EffectiveRetrySource,
|
|
#[serde(default)]
|
|
ws_defaults: std::option::Option<EffectiveWsSessionSource>,
|
|
#[serde(default)]
|
|
grpc_defaults: std::option::Option<EffectiveGrpcSessionSource>,
|
|
endpoints: std::vec::Vec<EffectiveEndpointSource>,
|
|
#[serde(default)]
|
|
ws_endpoints: std::option::Option<std::vec::Vec<EffectiveWsEndpointSource>>,
|
|
#[serde(default)]
|
|
grpc_endpoints: std::option::Option<std::vec::Vec<EffectiveGrpcEndpointSource>>,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveRetrySource {
|
|
max_retries: u32,
|
|
initial_backoff_ms: u64,
|
|
max_backoff_ms: u64,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveEndpointSource {
|
|
name: String,
|
|
enabled: bool,
|
|
provider: String,
|
|
cluster: String,
|
|
url: String,
|
|
connect_timeout_ms: u64,
|
|
request_timeout_ms: u64,
|
|
max_idle_connections_per_host: std::option::Option<usize>,
|
|
roles: std::vec::Vec<EffectiveRoleSource>,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveRoleSource {
|
|
role: String,
|
|
enabled: bool,
|
|
request_kinds: std::vec::Vec<String>,
|
|
priority: u32,
|
|
limits: EffectiveLimitsSource,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveLimitsSource {
|
|
requests_per_second: std::option::Option<u32>,
|
|
burst_capacity: std::option::Option<u32>,
|
|
max_concurrent_requests: std::option::Option<u32>,
|
|
pause_after_rate_limit_ms: std::option::Option<u64>,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveWsSessionSource {
|
|
command_timeout_ms: u64,
|
|
close_timeout_ms: u64,
|
|
reconnect: EffectiveWsReconnectSource,
|
|
resubscribe: String,
|
|
command_queue_capacity: usize,
|
|
notification_queue_capacity: usize,
|
|
max_active_subscriptions: usize,
|
|
max_pending_requests: usize,
|
|
max_message_size_bytes: usize,
|
|
max_frame_size_bytes: usize,
|
|
max_write_buffer_size_bytes: usize,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveWsReconnectSource {
|
|
max_retries: u32,
|
|
initial_backoff_ms: u64,
|
|
max_backoff_ms: u64,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveWsEndpointSource {
|
|
name: String,
|
|
enabled: bool,
|
|
provider: String,
|
|
cluster: String,
|
|
kind: String,
|
|
url: String,
|
|
#[serde(default)]
|
|
capabilities: std::option::Option<std::vec::Vec<String>>,
|
|
#[serde(default)]
|
|
session: std::option::Option<EffectiveWsSessionOverrideSource>,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveWsSessionOverrideSource {
|
|
command_timeout_ms: std::option::Option<u64>,
|
|
close_timeout_ms: std::option::Option<u64>,
|
|
reconnect: std::option::Option<EffectiveWsReconnectOverrideSource>,
|
|
resubscribe: std::option::Option<String>,
|
|
command_queue_capacity: std::option::Option<usize>,
|
|
notification_queue_capacity: std::option::Option<usize>,
|
|
max_active_subscriptions: std::option::Option<usize>,
|
|
max_pending_requests: std::option::Option<usize>,
|
|
max_message_size_bytes: std::option::Option<usize>,
|
|
max_frame_size_bytes: std::option::Option<usize>,
|
|
max_write_buffer_size_bytes: std::option::Option<usize>,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveWsReconnectOverrideSource {
|
|
max_retries: std::option::Option<u32>,
|
|
initial_backoff_ms: std::option::Option<u64>,
|
|
max_backoff_ms: std::option::Option<u64>,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveGrpcReconnectSource {
|
|
max_retries: u32,
|
|
initial_backoff_ms: u64,
|
|
max_backoff_ms: u64,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveGrpcReconnectOverrideSource {
|
|
max_retries: std::option::Option<u32>,
|
|
initial_backoff_ms: std::option::Option<u64>,
|
|
max_backoff_ms: std::option::Option<u64>,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveGrpcSessionSource {
|
|
connect_timeout_ms: u64,
|
|
unary_timeout_ms: u64,
|
|
close_timeout_ms: u64,
|
|
reconnect: EffectiveGrpcReconnectSource,
|
|
request_channel_capacity: usize,
|
|
update_channel_capacity: usize,
|
|
max_inbound_message_size_bytes: usize,
|
|
max_outbound_message_size_bytes: usize,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveGrpcSessionOverrideSource {
|
|
connect_timeout_ms: std::option::Option<u64>,
|
|
unary_timeout_ms: std::option::Option<u64>,
|
|
close_timeout_ms: std::option::Option<u64>,
|
|
reconnect: std::option::Option<EffectiveGrpcReconnectOverrideSource>,
|
|
request_channel_capacity: std::option::Option<usize>,
|
|
update_channel_capacity: std::option::Option<usize>,
|
|
max_inbound_message_size_bytes: std::option::Option<usize>,
|
|
max_outbound_message_size_bytes: std::option::Option<usize>,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveGrpcMetadataSource {
|
|
key: String,
|
|
value: String,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct EffectiveGrpcEndpointSource {
|
|
name: String,
|
|
enabled: bool,
|
|
provider: String,
|
|
cluster: String,
|
|
protocol: String,
|
|
url: String,
|
|
#[serde(default)]
|
|
metadata: std::vec::Vec<EffectiveGrpcMetadataSource>,
|
|
#[serde(default)]
|
|
secret_metadata: std::vec::Vec<EffectiveGrpcMetadataSource>,
|
|
#[serde(default)]
|
|
session: std::option::Option<EffectiveGrpcSessionOverrideSource>,
|
|
}
|
|
|
|
fn resolve_transport_profile(profile: &crate::ResolvedConfigProfile, environment: &crate::ConfigEnvironment) -> ksp_core_lib::Result<ResolvedTransportConfig> {
|
|
ksp_logging_lib::trace!(target: crate::TRACING_TARGET, profile_id = profile.profile_id(), "mapping standard Transport Config profile");
|
|
let effective = profile.resolve_effective_environment_detailed(environment);
|
|
let effective = match effective {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let source = serde_json::from_value::<EffectiveTransportSource>(effective.value().clone());
|
|
let source = match source {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
effective_error(profile, "effective Transport Config cannot be decoded into the runtime adapter contract").with_source(error),
|
|
);
|
|
},
|
|
};
|
|
if source.profile_id != profile.profile_id() {
|
|
return std::result::Result::Err(effective_error(profile, "effective Transport profile_id does not match the selected profile"));
|
|
}
|
|
let format_version = source.format_version;
|
|
let retry = ksp_onchain_transport_lib::HttpRetrySettings::new(
|
|
source.retry.max_retries,
|
|
std::time::Duration::from_millis(source.retry.initial_backoff_ms),
|
|
std::time::Duration::from_millis(source.retry.max_backoff_ms),
|
|
);
|
|
let endpoints = map_endpoints(source.endpoints, profile);
|
|
let endpoints = match endpoints {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let settings = ksp_onchain_transport_lib::HttpTransportSettings::new(endpoints, retry);
|
|
if let std::result::Result::Err(error) = settings.validate() {
|
|
return std::result::Result::Err(transport_contract_error(profile, "effective HTTP Transport settings fail the Transport runtime contract", &error));
|
|
}
|
|
let ws_settings = map_optional_ws_settings(format_version, source.ws_defaults, source.ws_endpoints, profile);
|
|
let ws_settings = match ws_settings {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let grpc_settings = map_optional_grpc_settings(format_version, source.grpc_defaults, source.grpc_endpoints, &effective, profile);
|
|
let grpc_settings = match grpc_settings {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let ws_endpoint_count = match ws_settings.as_ref() {
|
|
std::option::Option::Some(value) => value.endpoints().len(),
|
|
std::option::Option::None => 0_usize,
|
|
};
|
|
let grpc_endpoint_count = match grpc_settings.as_ref() {
|
|
std::option::Option::Some(value) => value.endpoints().len(),
|
|
std::option::Option::None => 0_usize,
|
|
};
|
|
ksp_logging_lib::debug!(
|
|
target: crate::TRACING_TARGET,
|
|
profile_id = profile.profile_id(),
|
|
format_version,
|
|
http_endpoint_count = settings.endpoints().len(),
|
|
ws_endpoint_count,
|
|
grpc_endpoint_count,
|
|
"mapped standard Transport Config to runtime settings"
|
|
);
|
|
return std::result::Result::Ok(ResolvedTransportConfig {
|
|
file_id: profile.file_id().clone(),
|
|
source_path: profile.path().to_path_buf(),
|
|
profile_id: profile.profile_id().to_owned(),
|
|
selection_source: profile.selection_source(),
|
|
effective,
|
|
settings,
|
|
ws_settings,
|
|
grpc_settings,
|
|
});
|
|
}
|
|
|
|
fn map_optional_ws_settings(
|
|
format_version: u32,
|
|
defaults: std::option::Option<EffectiveWsSessionSource>,
|
|
sources: std::option::Option<std::vec::Vec<EffectiveWsEndpointSource>>,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
) -> ksp_core_lib::Result<std::option::Option<ksp_onchain_transport_lib::WsTransportSettings>> {
|
|
return match format_version {
|
|
1 => {
|
|
if defaults.is_some() || sources.is_some() {
|
|
std::result::Result::Err(effective_error(profile, "Transport V1 must remain HTTP-only"))
|
|
} else {
|
|
ksp_logging_lib::trace!(
|
|
target: crate::TRACING_TARGET,
|
|
profile_id = profile.profile_id(),
|
|
"mapped backward-compatible Transport V1 without WebSocket settings"
|
|
);
|
|
std::result::Result::Ok(std::option::Option::None)
|
|
}
|
|
},
|
|
2 | 3 => {
|
|
let defaults = match defaults {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(effective_error(profile, "Transport V2/V3 requires ws_defaults")),
|
|
};
|
|
let sources = match sources {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(effective_error(profile, "Transport V2/V3 profile requires ws_endpoints")),
|
|
};
|
|
let endpoints = map_ws_endpoints(sources, &defaults, profile);
|
|
let endpoints = match endpoints {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let settings = ksp_onchain_transport_lib::WsTransportSettings::new(endpoints);
|
|
if let std::result::Result::Err(error) = settings.validate() {
|
|
return std::result::Result::Err(transport_contract_error(
|
|
profile,
|
|
"effective WebSocket Transport settings fail the Transport runtime contract",
|
|
&error,
|
|
));
|
|
}
|
|
std::result::Result::Ok(std::option::Option::Some(settings))
|
|
},
|
|
_ => std::result::Result::Err(effective_error(profile, "effective Transport format_version is unsupported")),
|
|
};
|
|
}
|
|
|
|
fn map_optional_grpc_settings(
|
|
format_version: u32,
|
|
defaults: std::option::Option<EffectiveGrpcSessionSource>,
|
|
sources: std::option::Option<std::vec::Vec<EffectiveGrpcEndpointSource>>,
|
|
effective: &crate::ResolvedConfigJson,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
) -> ksp_core_lib::Result<std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcTransportSettings>> {
|
|
return match format_version {
|
|
1 | 2 => {
|
|
if defaults.is_some() || sources.is_some() {
|
|
std::result::Result::Err(effective_error(profile, "Transport V1/V2 must not define Yellowstone gRPC settings"))
|
|
} else {
|
|
std::result::Result::Ok(std::option::Option::None)
|
|
}
|
|
},
|
|
3 => {
|
|
let defaults = match defaults {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(effective_error(profile, "Transport V3 requires grpc_defaults")),
|
|
};
|
|
let sources = match sources {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
|
|
};
|
|
let endpoints = map_grpc_endpoints(sources, &defaults, effective, profile);
|
|
let endpoints = match endpoints {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let settings = ksp_onchain_transport_lib::YellowstoneGrpcTransportSettings::new(endpoints);
|
|
if let std::result::Result::Err(error) = settings.validate() {
|
|
return std::result::Result::Err(transport_contract_error(
|
|
profile,
|
|
"effective Yellowstone gRPC Transport settings fail the Transport runtime contract",
|
|
&error,
|
|
));
|
|
}
|
|
std::result::Result::Ok(std::option::Option::Some(settings))
|
|
},
|
|
_ => std::result::Result::Err(effective_error(profile, "effective Transport format_version is unsupported")),
|
|
};
|
|
}
|
|
|
|
fn map_grpc_endpoints(
|
|
sources: std::vec::Vec<EffectiveGrpcEndpointSource>,
|
|
defaults: &EffectiveGrpcSessionSource,
|
|
effective: &crate::ResolvedConfigJson,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
) -> ksp_core_lib::Result<std::vec::Vec<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings>> {
|
|
let mut endpoints = std::vec::Vec::<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings>::with_capacity(sources.len());
|
|
for (endpoint_index, source) in sources.into_iter().enumerate() {
|
|
let endpoint_name = source.name.clone();
|
|
if source.protocol != "solana_yellowstone" {
|
|
return std::result::Result::Err(
|
|
effective_error(profile, "effective gRPC protocol is unsupported")
|
|
.with_context("endpoint_name", endpoint_name)
|
|
.with_context("grpc_protocol", source.protocol),
|
|
);
|
|
}
|
|
let url = ksp_onchain_transport_lib::YellowstoneGrpcEndpointUrl::parse(source.url);
|
|
let url = match url {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
transport_contract_error(profile, "effective Yellowstone gRPC endpoint URL is invalid", &error)
|
|
.with_context("endpoint_name", endpoint_name),
|
|
);
|
|
},
|
|
};
|
|
let session = map_grpc_session_settings(defaults, source.session.as_ref(), profile, endpoint_name.as_str());
|
|
let session = match session {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let metadata = map_grpc_metadata(source.metadata, source.secret_metadata, effective, profile, endpoint_name.as_str(), endpoint_index);
|
|
let metadata = match metadata {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let endpoint = ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings::new(
|
|
source.name,
|
|
source.enabled,
|
|
ksp_onchain_transport_lib::YellowstoneGrpcProviderName::new(source.provider),
|
|
ksp_onchain_transport_lib::YellowstoneGrpcClusterName::new(source.cluster),
|
|
url,
|
|
session,
|
|
);
|
|
let endpoint = endpoint.with_metadata(metadata);
|
|
let endpoint = match endpoint {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
transport_contract_error(profile, "effective Yellowstone gRPC metadata fail the Transport runtime contract", &error)
|
|
.with_context("endpoint_name", endpoint_name),
|
|
);
|
|
},
|
|
};
|
|
endpoints.push(endpoint);
|
|
}
|
|
return std::result::Result::Ok(endpoints);
|
|
}
|
|
|
|
fn map_grpc_session_settings(
|
|
defaults: &EffectiveGrpcSessionSource,
|
|
overrides: std::option::Option<&EffectiveGrpcSessionOverrideSource>,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
endpoint_name: &str,
|
|
) -> ksp_core_lib::Result<ksp_onchain_transport_lib::YellowstoneGrpcSessionSettings> {
|
|
let mut connect_timeout_ms = defaults.connect_timeout_ms;
|
|
let mut unary_timeout_ms = defaults.unary_timeout_ms;
|
|
let mut close_timeout_ms = defaults.close_timeout_ms;
|
|
let mut reconnect_max_retries = defaults.reconnect.max_retries;
|
|
let mut reconnect_initial_backoff_ms = defaults.reconnect.initial_backoff_ms;
|
|
let mut reconnect_max_backoff_ms = defaults.reconnect.max_backoff_ms;
|
|
let mut request_channel_capacity = defaults.request_channel_capacity;
|
|
let mut update_channel_capacity = defaults.update_channel_capacity;
|
|
let mut max_inbound_message_size_bytes = defaults.max_inbound_message_size_bytes;
|
|
let mut max_outbound_message_size_bytes = defaults.max_outbound_message_size_bytes;
|
|
if let std::option::Option::Some(overrides) = overrides {
|
|
if let std::option::Option::Some(value) = overrides.connect_timeout_ms {
|
|
connect_timeout_ms = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.unary_timeout_ms {
|
|
unary_timeout_ms = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.close_timeout_ms {
|
|
close_timeout_ms = value;
|
|
}
|
|
if let std::option::Option::Some(reconnect) = overrides.reconnect.as_ref() {
|
|
if let std::option::Option::Some(value) = reconnect.max_retries {
|
|
reconnect_max_retries = value;
|
|
}
|
|
if let std::option::Option::Some(value) = reconnect.initial_backoff_ms {
|
|
reconnect_initial_backoff_ms = value;
|
|
}
|
|
if let std::option::Option::Some(value) = reconnect.max_backoff_ms {
|
|
reconnect_max_backoff_ms = value;
|
|
}
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.request_channel_capacity {
|
|
request_channel_capacity = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.update_channel_capacity {
|
|
update_channel_capacity = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.max_inbound_message_size_bytes {
|
|
max_inbound_message_size_bytes = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.max_outbound_message_size_bytes {
|
|
max_outbound_message_size_bytes = value;
|
|
}
|
|
}
|
|
let reconnect = ksp_onchain_transport_lib::YellowstoneGrpcReconnectSettings::new(
|
|
reconnect_max_retries,
|
|
std::time::Duration::from_millis(reconnect_initial_backoff_ms),
|
|
std::time::Duration::from_millis(reconnect_max_backoff_ms),
|
|
);
|
|
let settings = ksp_onchain_transport_lib::YellowstoneGrpcSessionSettings::new(
|
|
std::time::Duration::from_millis(connect_timeout_ms),
|
|
std::time::Duration::from_millis(unary_timeout_ms),
|
|
std::time::Duration::from_millis(close_timeout_ms),
|
|
reconnect,
|
|
request_channel_capacity,
|
|
update_channel_capacity,
|
|
max_inbound_message_size_bytes,
|
|
max_outbound_message_size_bytes,
|
|
);
|
|
if let std::result::Result::Err(error) = settings.validate() {
|
|
return std::result::Result::Err(
|
|
transport_contract_error(profile, "effective Yellowstone gRPC session settings fail the Transport runtime contract", &error)
|
|
.with_context("endpoint_name", endpoint_name),
|
|
);
|
|
}
|
|
return std::result::Result::Ok(settings);
|
|
}
|
|
|
|
fn map_grpc_metadata(
|
|
public_sources: std::vec::Vec<EffectiveGrpcMetadataSource>,
|
|
secret_sources: std::vec::Vec<EffectiveGrpcMetadataSource>,
|
|
effective: &crate::ResolvedConfigJson,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
endpoint_name: &str,
|
|
endpoint_index: usize,
|
|
) -> ksp_core_lib::Result<std::vec::Vec<ksp_onchain_transport_lib::YellowstoneGrpcMetadataEntry>> {
|
|
let mut metadata = std::vec::Vec::<ksp_onchain_transport_lib::YellowstoneGrpcMetadataEntry>::with_capacity(public_sources.len() + secret_sources.len());
|
|
for (metadata_index, source) in public_sources.into_iter().enumerate() {
|
|
let pointer = format!("/grpc_endpoints/{endpoint_index}/metadata/{metadata_index}/value");
|
|
if let std::result::Result::Err(error) = validate_grpc_metadata_provenance(effective, pointer.as_str(), false, profile, endpoint_name) {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let entry = ksp_onchain_transport_lib::YellowstoneGrpcMetadataEntry::public(source.key, source.value);
|
|
let entry = match entry {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
transport_contract_error(profile, "effective public Yellowstone gRPC metadata are invalid", &error)
|
|
.with_context("endpoint_name", endpoint_name),
|
|
);
|
|
},
|
|
};
|
|
metadata.push(entry);
|
|
}
|
|
for (metadata_index, source) in secret_sources.into_iter().enumerate() {
|
|
let pointer = format!("/grpc_endpoints/{endpoint_index}/secret_metadata/{metadata_index}/value");
|
|
if let std::result::Result::Err(error) = validate_grpc_metadata_provenance(effective, pointer.as_str(), true, profile, endpoint_name) {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let entry = ksp_onchain_transport_lib::YellowstoneGrpcMetadataEntry::secret(source.key, source.value);
|
|
let entry = match entry {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
transport_contract_error(profile, "effective secret Yellowstone gRPC metadata are invalid", &error)
|
|
.with_context("endpoint_name", endpoint_name),
|
|
);
|
|
},
|
|
};
|
|
metadata.push(entry);
|
|
}
|
|
return std::result::Result::Ok(metadata);
|
|
}
|
|
|
|
fn validate_grpc_metadata_provenance(
|
|
effective: &crate::ResolvedConfigJson,
|
|
pointer: &str,
|
|
secret_class: bool,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
endpoint_name: &str,
|
|
) -> ksp_core_lib::Result<()> {
|
|
let provenance = match effective.provenance_at(pointer) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => {
|
|
return std::result::Result::Err(
|
|
effective_error(profile, "Yellowstone gRPC metadata provenance is unavailable")
|
|
.with_context("endpoint_name", endpoint_name)
|
|
.with_context("field", pointer),
|
|
);
|
|
},
|
|
};
|
|
let mut has_secret_environment = false;
|
|
for item in provenance {
|
|
let variable_name = match item.variable_name() {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => continue,
|
|
};
|
|
let sensitivity = crate::ConfigSensitivity::from_variable_name(variable_name);
|
|
let sensitivity = match sensitivity {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
if sensitivity.is_secret() {
|
|
has_secret_environment = true;
|
|
if !secret_class {
|
|
return std::result::Result::Err(
|
|
effective_error(profile, "public Yellowstone gRPC metadata must not derive from a secret environment variable")
|
|
.with_context("endpoint_name", endpoint_name)
|
|
.with_context("field", pointer),
|
|
);
|
|
}
|
|
} else if secret_class {
|
|
return std::result::Result::Err(
|
|
effective_error(profile, "secret Yellowstone gRPC metadata may reference only secret environment variables")
|
|
.with_context("endpoint_name", endpoint_name)
|
|
.with_context("field", pointer),
|
|
);
|
|
}
|
|
}
|
|
if secret_class && !has_secret_environment {
|
|
return std::result::Result::Err(
|
|
effective_error(profile, "secret Yellowstone gRPC metadata require secret environment provenance")
|
|
.with_context("endpoint_name", endpoint_name)
|
|
.with_context("field", pointer),
|
|
);
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
fn map_endpoints(
|
|
sources: std::vec::Vec<EffectiveEndpointSource>,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
) -> ksp_core_lib::Result<std::vec::Vec<ksp_onchain_transport_lib::HttpEndpointSettings>> {
|
|
let mut endpoints = std::vec::Vec::<ksp_onchain_transport_lib::HttpEndpointSettings>::with_capacity(sources.len());
|
|
for source in sources {
|
|
let endpoint_name = source.name.clone();
|
|
let url = ksp_onchain_transport_lib::HttpEndpointUrl::parse(source.url);
|
|
let url = match url {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
transport_contract_error(profile, "effective HTTP endpoint URL is invalid", &error).with_context("endpoint_name", endpoint_name),
|
|
);
|
|
},
|
|
};
|
|
let roles = map_roles(source.roles, profile, endpoint_name.as_str());
|
|
let roles = match roles {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
endpoints.push(ksp_onchain_transport_lib::HttpEndpointSettings::new(
|
|
source.name,
|
|
source.enabled,
|
|
ksp_onchain_transport_lib::HttpProviderName::new(source.provider),
|
|
ksp_onchain_transport_lib::HttpClusterName::new(source.cluster),
|
|
url,
|
|
std::time::Duration::from_millis(source.connect_timeout_ms),
|
|
std::time::Duration::from_millis(source.request_timeout_ms),
|
|
source.max_idle_connections_per_host,
|
|
roles,
|
|
));
|
|
}
|
|
return std::result::Result::Ok(endpoints);
|
|
}
|
|
|
|
fn map_ws_endpoints(
|
|
sources: std::vec::Vec<EffectiveWsEndpointSource>,
|
|
defaults: &EffectiveWsSessionSource,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
) -> ksp_core_lib::Result<std::vec::Vec<ksp_onchain_transport_lib::WsEndpointSettings>> {
|
|
let mut endpoints = std::vec::Vec::<ksp_onchain_transport_lib::WsEndpointSettings>::with_capacity(sources.len());
|
|
for source in sources {
|
|
let endpoint_name = source.name.clone();
|
|
let protocol = map_ws_protocol_kind(source.kind.as_str(), profile, endpoint_name.as_str());
|
|
let protocol = match protocol {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let url = ksp_onchain_transport_lib::WsEndpointUrl::parse(source.url);
|
|
let url = match url {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
return std::result::Result::Err(
|
|
transport_contract_error(profile, "effective WebSocket endpoint URL is invalid", &error).with_context("endpoint_name", endpoint_name),
|
|
);
|
|
},
|
|
};
|
|
let session = map_ws_session_settings(defaults, source.session.as_ref(), profile, endpoint_name.as_str());
|
|
let session = match session {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let mut endpoint = ksp_onchain_transport_lib::WsEndpointSettings::new(
|
|
source.name,
|
|
source.enabled,
|
|
ksp_onchain_transport_lib::WsProviderName::new(source.provider),
|
|
ksp_onchain_transport_lib::WsClusterName::new(source.cluster),
|
|
protocol,
|
|
url,
|
|
session,
|
|
);
|
|
if let std::option::Option::Some(capability_sources) = source.capabilities {
|
|
let mut capabilities = std::vec::Vec::<ksp_onchain_transport_lib::WsSubscriptionKind>::with_capacity(capability_sources.len());
|
|
for capability_source in capability_sources {
|
|
let capability = map_ws_subscription_capability(capability_source.as_str(), profile, endpoint_name.as_str());
|
|
let capability = match capability {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
capabilities.push(capability);
|
|
}
|
|
endpoint = endpoint.with_subscription_capabilities(capabilities);
|
|
}
|
|
endpoints.push(endpoint);
|
|
}
|
|
return std::result::Result::Ok(endpoints);
|
|
}
|
|
|
|
fn map_ws_protocol_kind(
|
|
value: &str,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
endpoint_name: &str,
|
|
) -> ksp_core_lib::Result<ksp_onchain_transport_lib::WsProtocolKind> {
|
|
return match value {
|
|
"solana_standard" => std::result::Result::Ok(ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard),
|
|
"helius_laserstream" => std::result::Result::Ok(ksp_onchain_transport_lib::WsProtocolKind::HeliusLaserStream),
|
|
_ => std::result::Result::Err(
|
|
effective_error(profile, "effective WebSocket protocol kind is unsupported")
|
|
.with_context("endpoint_name", endpoint_name)
|
|
.with_context("ws_kind", value),
|
|
),
|
|
};
|
|
}
|
|
|
|
fn map_ws_subscription_capability(
|
|
value: &str,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
endpoint_name: &str,
|
|
) -> ksp_core_lib::Result<ksp_onchain_transport_lib::WsSubscriptionKind> {
|
|
return match value {
|
|
"account" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::Account),
|
|
"block" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::Block),
|
|
"logs" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::Logs),
|
|
"program" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::Program),
|
|
"root" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::Root),
|
|
"signature" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::Signature),
|
|
"slot" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::Slot),
|
|
"slots_updates" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::SlotsUpdates),
|
|
"vote" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::Vote),
|
|
"helius_transaction" => std::result::Result::Ok(ksp_onchain_transport_lib::WsSubscriptionKind::HeliusTransaction),
|
|
_ => std::result::Result::Err(
|
|
effective_error(profile, "effective WebSocket subscription capability is unsupported")
|
|
.with_context("endpoint_name", endpoint_name)
|
|
.with_context("ws_capability", value),
|
|
),
|
|
};
|
|
}
|
|
|
|
fn map_ws_session_settings(
|
|
defaults: &EffectiveWsSessionSource,
|
|
overrides: std::option::Option<&EffectiveWsSessionOverrideSource>,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
endpoint_name: &str,
|
|
) -> ksp_core_lib::Result<ksp_onchain_transport_lib::WsSessionSettings> {
|
|
let mut command_timeout_ms = defaults.command_timeout_ms;
|
|
let mut close_timeout_ms = defaults.close_timeout_ms;
|
|
let mut reconnect_max_retries = defaults.reconnect.max_retries;
|
|
let mut reconnect_initial_backoff_ms = defaults.reconnect.initial_backoff_ms;
|
|
let mut reconnect_max_backoff_ms = defaults.reconnect.max_backoff_ms;
|
|
let mut resubscribe_text = defaults.resubscribe.clone();
|
|
let mut command_queue_capacity = defaults.command_queue_capacity;
|
|
let mut notification_queue_capacity = defaults.notification_queue_capacity;
|
|
let mut max_active_subscriptions = defaults.max_active_subscriptions;
|
|
let mut max_pending_requests = defaults.max_pending_requests;
|
|
let mut max_message_size_bytes = defaults.max_message_size_bytes;
|
|
let mut max_frame_size_bytes = defaults.max_frame_size_bytes;
|
|
let mut max_write_buffer_size_bytes = defaults.max_write_buffer_size_bytes;
|
|
if let std::option::Option::Some(overrides) = overrides {
|
|
if let std::option::Option::Some(value) = overrides.command_timeout_ms {
|
|
command_timeout_ms = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.close_timeout_ms {
|
|
close_timeout_ms = value;
|
|
}
|
|
if let std::option::Option::Some(reconnect) = overrides.reconnect.as_ref() {
|
|
if let std::option::Option::Some(value) = reconnect.max_retries {
|
|
reconnect_max_retries = value;
|
|
}
|
|
if let std::option::Option::Some(value) = reconnect.initial_backoff_ms {
|
|
reconnect_initial_backoff_ms = value;
|
|
}
|
|
if let std::option::Option::Some(value) = reconnect.max_backoff_ms {
|
|
reconnect_max_backoff_ms = value;
|
|
}
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.resubscribe.as_ref() {
|
|
resubscribe_text = value.clone();
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.command_queue_capacity {
|
|
command_queue_capacity = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.notification_queue_capacity {
|
|
notification_queue_capacity = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.max_active_subscriptions {
|
|
max_active_subscriptions = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.max_pending_requests {
|
|
max_pending_requests = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.max_message_size_bytes {
|
|
max_message_size_bytes = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.max_frame_size_bytes {
|
|
max_frame_size_bytes = value;
|
|
}
|
|
if let std::option::Option::Some(value) = overrides.max_write_buffer_size_bytes {
|
|
max_write_buffer_size_bytes = value;
|
|
}
|
|
}
|
|
let reconnect = ksp_onchain_transport_lib::WsReconnectSettings::new(
|
|
reconnect_max_retries,
|
|
std::time::Duration::from_millis(reconnect_initial_backoff_ms),
|
|
std::time::Duration::from_millis(reconnect_max_backoff_ms),
|
|
);
|
|
let resubscribe = map_ws_resubscribe_policy(resubscribe_text.as_str(), profile, endpoint_name);
|
|
let resubscribe = match resubscribe {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let settings = ksp_onchain_transport_lib::WsSessionSettings::new(
|
|
std::time::Duration::from_millis(command_timeout_ms),
|
|
std::time::Duration::from_millis(close_timeout_ms),
|
|
reconnect,
|
|
resubscribe,
|
|
command_queue_capacity,
|
|
notification_queue_capacity,
|
|
max_active_subscriptions,
|
|
max_pending_requests,
|
|
max_message_size_bytes,
|
|
max_frame_size_bytes,
|
|
max_write_buffer_size_bytes,
|
|
);
|
|
if let std::result::Result::Err(error) = settings.validate() {
|
|
return std::result::Result::Err(
|
|
transport_contract_error(profile, "effective WebSocket session settings fail the Transport runtime contract", &error)
|
|
.with_context("endpoint_name", endpoint_name),
|
|
);
|
|
}
|
|
return std::result::Result::Ok(settings);
|
|
}
|
|
|
|
fn map_ws_resubscribe_policy(
|
|
value: &str,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
endpoint_name: &str,
|
|
) -> ksp_core_lib::Result<ksp_onchain_transport_lib::WsResubscribePolicy> {
|
|
return match value {
|
|
"never" => std::result::Result::Ok(ksp_onchain_transport_lib::WsResubscribePolicy::Never),
|
|
"active_subscriptions" => std::result::Result::Ok(ksp_onchain_transport_lib::WsResubscribePolicy::ActiveSubscriptions),
|
|
_ => std::result::Result::Err(
|
|
effective_error(profile, "effective WebSocket resubscribe policy is unsupported")
|
|
.with_context("endpoint_name", endpoint_name)
|
|
.with_context("resubscribe", value),
|
|
),
|
|
};
|
|
}
|
|
|
|
fn map_roles(
|
|
sources: std::vec::Vec<EffectiveRoleSource>,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
endpoint_name: &str,
|
|
) -> ksp_core_lib::Result<std::vec::Vec<ksp_onchain_transport_lib::HttpEndpointRoleSettings>> {
|
|
let mut roles = std::vec::Vec::<ksp_onchain_transport_lib::HttpEndpointRoleSettings>::with_capacity(sources.len());
|
|
for source in sources {
|
|
let requests_per_second = map_non_zero(source.limits.requests_per_second, profile, "limits.requests_per_second", endpoint_name);
|
|
let requests_per_second = match requests_per_second {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let burst_capacity = map_non_zero(source.limits.burst_capacity, profile, "limits.burst_capacity", endpoint_name);
|
|
let burst_capacity = match burst_capacity {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let max_concurrent_requests = map_non_zero(source.limits.max_concurrent_requests, profile, "limits.max_concurrent_requests", endpoint_name);
|
|
let max_concurrent_requests = match max_concurrent_requests {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let pause_after_rate_limit = match source.limits.pause_after_rate_limit_ms {
|
|
std::option::Option::Some(value) => std::option::Option::Some(std::time::Duration::from_millis(value)),
|
|
std::option::Option::None => std::option::Option::None,
|
|
};
|
|
let limits = ksp_onchain_transport_lib::HttpRoleLimits::new(requests_per_second, burst_capacity, max_concurrent_requests, pause_after_rate_limit);
|
|
let mut request_kinds = std::vec::Vec::<ksp_onchain_transport_lib::HttpRequestKind>::with_capacity(source.request_kinds.len());
|
|
for request_kind in source.request_kinds {
|
|
request_kinds.push(ksp_onchain_transport_lib::HttpRequestKind::new(request_kind));
|
|
}
|
|
roles.push(ksp_onchain_transport_lib::HttpEndpointRoleSettings::new(
|
|
ksp_onchain_transport_lib::HttpRoleName::new(source.role),
|
|
source.enabled,
|
|
request_kinds,
|
|
source.priority,
|
|
limits,
|
|
));
|
|
}
|
|
return std::result::Result::Ok(roles);
|
|
}
|
|
|
|
fn map_non_zero(
|
|
value: std::option::Option<u32>,
|
|
profile: &crate::ResolvedConfigProfile,
|
|
field: &'static str,
|
|
endpoint_name: &str,
|
|
) -> ksp_core_lib::Result<std::option::Option<std::num::NonZeroU32>> {
|
|
let value = match value {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
|
|
};
|
|
let non_zero = std::num::NonZeroU32::new(value);
|
|
return match non_zero {
|
|
std::option::Option::Some(value) => std::result::Result::Ok(std::option::Option::Some(value)),
|
|
std::option::Option::None => std::result::Result::Err(
|
|
effective_error(profile, "effective Transport role limit must be greater than zero")
|
|
.with_context("field", field)
|
|
.with_context("endpoint_name", endpoint_name),
|
|
),
|
|
};
|
|
}
|
|
|
|
fn transport_contract_error(profile: &crate::ResolvedConfigProfile, reason: &'static str, transport_error: &ksp_core_lib::Error) -> ksp_core_lib::Error {
|
|
return effective_error(profile, reason)
|
|
.with_context("transport_error_domain", transport_error.code().domain())
|
|
.with_context("transport_error_code", transport_error.code().code());
|
|
}
|
|
|
|
fn effective_error(profile: &crate::ResolvedConfigProfile, reason: &'static str) -> ksp_core_lib::Error {
|
|
return ksp_core_lib::Error::new(crate::ERROR_CODE_EFFECTIVE_CONFIG_INVALID, "effective Config cannot be mapped to the requested runtime contract")
|
|
.with_context("file_id", profile.file_id().as_str())
|
|
.with_context("profile_id", profile.profile_id())
|
|
.with_context("reason", reason);
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "../unit_tests/transport.rs"]
|
|
mod tests;
|