335 lines
15 KiB
Rust
335 lines
15 KiB
Rust
// file: crates/ksp-config-lib/src/transport.rs
|
|
// version: 2
|
|
|
|
/// Effective standard HTTP Transport configuration resolved from Config and mapped to the Transport runtime contract.
|
|
#[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,
|
|
}
|
|
|
|
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.
|
|
#[must_use]
|
|
pub const fn settings(&self) -> &ksp_onchain_transport_lib::HttpTransportSettings {
|
|
return &self.settings;
|
|
}
|
|
|
|
/// 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;
|
|
}
|
|
}
|
|
|
|
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)
|
|
.finish_non_exhaustive();
|
|
}
|
|
}
|
|
|
|
impl crate::ConfigDocumentEngine {
|
|
/// Loads the standard HTTP Transport document, selects a profile, resolves environment placeholders and maps it to Transport runtime settings.
|
|
///
|
|
/// `requested_profile = None` uses the document `default_profile`; `Some(profile_id)` requests an explicit profile. Secret endpoint URLs are allowed
|
|
/// because the Transport URL wrapper owns runtime redaction. Invalid environment-resolved values are reported as
|
|
/// [`crate::ERROR_CODE_EFFECTIVE_CONFIG_INVALID`] without copying endpoint URL values into ordinary error context.
|
|
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 the runtime HTTP Transport adapter while preserving its 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,
|
|
endpoints: std::vec::Vec<EffectiveEndpointSource>,
|
|
}
|
|
|
|
#[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>,
|
|
}
|
|
|
|
fn resolve_transport_profile(profile: &crate::ResolvedConfigProfile, environment: &crate::ConfigEnvironment) -> ksp_core_lib::Result<ResolvedTransportConfig> {
|
|
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.format_version != 1 {
|
|
return std::result::Result::Err(effective_error(profile, "effective Transport format_version is unsupported"));
|
|
}
|
|
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 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);
|
|
let validation = settings.validate();
|
|
if let std::result::Result::Err(error) = validation {
|
|
return std::result::Result::Err(transport_contract_error(profile, "effective Transport settings fail the Transport runtime contract", &error));
|
|
}
|
|
ksp_logging_lib::debug!(
|
|
target: crate::TRACING_TARGET,
|
|
profile_id = profile.profile_id(),
|
|
endpoint_count = settings.endpoints().len(),
|
|
"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,
|
|
});
|
|
}
|
|
|
|
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 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_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;
|