v0.3.2-pre.005
This commit is contained in:
303
crates/ksp-store-postgres-lib/src/runtime.rs
Normal file
303
crates/ksp-store-postgres-lib/src/runtime.rs
Normal file
@@ -0,0 +1,303 @@
|
||||
// file: crates/ksp-store-postgres-lib/src/runtime.rs
|
||||
// version: 1
|
||||
|
||||
const APPLICATION_NAME: &str = "ksp-store";
|
||||
const MAX_CONNECTION_URI_BYTES: usize = 4_096;
|
||||
const SHUTDOWN_POLL_INTERVAL_MS: u64 = 10;
|
||||
|
||||
/// TLS mode accepted by the physical PostgreSQL backend bridge.
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
#[non_exhaustive]
|
||||
pub enum PostgresBackendTlsMode {
|
||||
/// Disable TLS for the selected PostgreSQL target.
|
||||
Disabled,
|
||||
/// Require TLS with system-root trust and server-identity verification.
|
||||
VerifyFull,
|
||||
}
|
||||
|
||||
impl PostgresBackendTlsMode {
|
||||
/// Returns the stable safe TLS mode code used only in diagnostics.
|
||||
#[must_use]
|
||||
pub const fn code(&self) -> &'static str {
|
||||
return match self {
|
||||
Self::Disabled => "disabled",
|
||||
Self::VerifyFull => "verify_full",
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// Physical settings consumed only by the PostgreSQL backend crate.
|
||||
pub struct PostgresBackendSettings {
|
||||
connect_timeout: std::time::Duration,
|
||||
connection_uri: std::string::String,
|
||||
create_timeout: std::time::Duration,
|
||||
max_connections: u32,
|
||||
network: ksp_store_api::RawNetworkId,
|
||||
recycle_timeout: std::time::Duration,
|
||||
tls_mode: PostgresBackendTlsMode,
|
||||
wait_timeout: std::time::Duration,
|
||||
}
|
||||
|
||||
impl PostgresBackendSettings {
|
||||
/// Creates the physical PostgreSQL settings bridge from already validated facade-owned values.
|
||||
#[must_use]
|
||||
pub fn new(
|
||||
network: ksp_store_api::RawNetworkId,
|
||||
connection_uri: impl std::convert::Into<std::string::String>,
|
||||
max_connections: u32,
|
||||
connect_timeout: std::time::Duration,
|
||||
wait_timeout: std::time::Duration,
|
||||
create_timeout: std::time::Duration,
|
||||
recycle_timeout: std::time::Duration,
|
||||
tls_mode: PostgresBackendTlsMode,
|
||||
) -> Self {
|
||||
return Self {
|
||||
connect_timeout,
|
||||
connection_uri: connection_uri.into(),
|
||||
create_timeout,
|
||||
max_connections,
|
||||
network,
|
||||
recycle_timeout,
|
||||
tls_mode,
|
||||
wait_timeout,
|
||||
};
|
||||
}
|
||||
|
||||
/// Returns the logical network bound to this one backend instance.
|
||||
#[must_use]
|
||||
pub const fn network(&self) -> &ksp_store_api::RawNetworkId {
|
||||
return &self.network;
|
||||
}
|
||||
|
||||
/// Returns the selected safe TLS mode.
|
||||
#[must_use]
|
||||
pub const fn tls_mode(&self) -> PostgresBackendTlsMode {
|
||||
return self.tls_mode;
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for PostgresBackendSettings {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
return formatter
|
||||
.debug_struct("PostgresBackendSettings")
|
||||
.field("network", &self.network)
|
||||
.field("connection_uri", &"<redacted>")
|
||||
.field("max_connections", &self.max_connections)
|
||||
.field("connect_timeout", &self.connect_timeout)
|
||||
.field("wait_timeout", &self.wait_timeout)
|
||||
.field("create_timeout", &self.create_timeout)
|
||||
.field("recycle_timeout", &self.recycle_timeout)
|
||||
.field("tls_mode", &self.tls_mode)
|
||||
.finish();
|
||||
}
|
||||
}
|
||||
|
||||
/// Opaque physical PostgreSQL backend owning the bounded Deadpool connection pool.
|
||||
pub struct PostgresBackend {
|
||||
network: ksp_store_api::RawNetworkId,
|
||||
pool: deadpool_postgres::Pool,
|
||||
}
|
||||
|
||||
impl PostgresBackend {
|
||||
/// Parses and normalizes one supplied URI, builds a bounded pool and proves one physical connection before returning readiness.
|
||||
pub async fn open(settings: PostgresBackendSettings) -> std::result::Result<Self, crate::PostgresBackendError> {
|
||||
let normalized = normalized_config(&settings);
|
||||
let pg_config = match normalized {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
ksp_logging_lib::debug!(
|
||||
target: crate::TRACING_TARGET,
|
||||
network = settings.network().as_str(),
|
||||
tls_mode = settings.tls_mode().code(),
|
||||
max_connections = settings.max_connections,
|
||||
"opening PostgreSQL Store backend pool"
|
||||
);
|
||||
let pool_result = match settings.tls_mode {
|
||||
PostgresBackendTlsMode::Disabled => build_pool(pg_config, tokio_postgres::NoTls, &settings),
|
||||
PostgresBackendTlsMode::VerifyFull => {
|
||||
let tls_result = build_verified_tls();
|
||||
let tls = match tls_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
build_pool(pg_config, tls, &settings)
|
||||
},
|
||||
};
|
||||
let pool = match pool_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let probe = pool.get().await;
|
||||
match probe {
|
||||
std::result::Result::Ok(client) => drop(client),
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(map_pool_error(error)),
|
||||
}
|
||||
ksp_logging_lib::debug!(
|
||||
target: crate::TRACING_TARGET,
|
||||
network = settings.network().as_str(),
|
||||
tls_mode = settings.tls_mode().code(),
|
||||
"PostgreSQL Store backend established initial physical connection"
|
||||
);
|
||||
return std::result::Result::Ok(Self { network: settings.network, pool });
|
||||
}
|
||||
|
||||
/// Explicitly closes the pool and waits for all owned pooled objects to drain inside the supplied bound.
|
||||
pub async fn close(self, timeout: std::time::Duration) -> std::result::Result<(), crate::PostgresBackendError> {
|
||||
self.pool.close();
|
||||
let drain = async {
|
||||
loop {
|
||||
if self.pool.status().size == 0 {
|
||||
return;
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_millis(SHUTDOWN_POLL_INTERVAL_MS)).await;
|
||||
}
|
||||
};
|
||||
let result = tokio::time::timeout(timeout, drain).await;
|
||||
return match result {
|
||||
std::result::Result::Ok(()) => {
|
||||
ksp_logging_lib::debug!(target: crate::TRACING_TARGET, network = self.network.as_str(), "PostgreSQL Store backend pool closed");
|
||||
std::result::Result::Ok(())
|
||||
},
|
||||
std::result::Result::Err(_) => {
|
||||
std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ShutdownTimeout, "pool_drain"))
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for PostgresBackend {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
return formatter.debug_struct("PostgresBackend").field("network", &self.network).field("state", &"open").finish();
|
||||
}
|
||||
}
|
||||
|
||||
impl std::ops::Drop for PostgresBackend {
|
||||
fn drop(&mut self) {
|
||||
self.pool.close();
|
||||
}
|
||||
}
|
||||
|
||||
fn normalized_config(settings: &PostgresBackendSettings) -> std::result::Result<tokio_postgres::Config, crate::PostgresBackendError> {
|
||||
if settings.connection_uri.is_empty() || settings.connection_uri.len() > MAX_CONNECTION_URI_BYTES {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConfigInvalid, "connection_uri"));
|
||||
}
|
||||
let parsed = settings.connection_uri.parse::<tokio_postgres::Config>();
|
||||
let mut config = match parsed {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConfigInvalid, "connection_uri"));
|
||||
},
|
||||
};
|
||||
if config.get_options().is_some() {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConfigInvalid, "server_options"));
|
||||
}
|
||||
if config.get_hosts().is_empty() && config.get_hostaddrs().is_empty() {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConfigInvalid, "host"));
|
||||
}
|
||||
if settings.tls_mode == PostgresBackendTlsMode::VerifyFull {
|
||||
if config.get_hosts().is_empty() {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConfigInvalid, "tls_server_identity"));
|
||||
}
|
||||
for host in config.get_hosts() {
|
||||
if !is_tcp_host(host) {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConfigInvalid, "tls_server_identity"));
|
||||
}
|
||||
}
|
||||
}
|
||||
config.application_name(APPLICATION_NAME);
|
||||
config.connect_timeout(settings.connect_timeout);
|
||||
config.ssl_negotiation(tokio_postgres::config::SslNegotiation::Postgres);
|
||||
match settings.tls_mode {
|
||||
PostgresBackendTlsMode::Disabled => {
|
||||
config.ssl_mode(tokio_postgres::config::SslMode::Disable);
|
||||
},
|
||||
PostgresBackendTlsMode::VerifyFull => {
|
||||
config.ssl_mode(tokio_postgres::config::SslMode::Require);
|
||||
},
|
||||
}
|
||||
return std::result::Result::Ok(config);
|
||||
}
|
||||
|
||||
fn is_tcp_host(host: &tokio_postgres::config::Host) -> bool {
|
||||
return match host {
|
||||
tokio_postgres::config::Host::Tcp(_) => true,
|
||||
#[cfg(unix)]
|
||||
tokio_postgres::config::Host::Unix(_) => false,
|
||||
};
|
||||
}
|
||||
|
||||
fn build_pool<T>(
|
||||
pg_config: tokio_postgres::Config,
|
||||
tls: T,
|
||||
settings: &PostgresBackendSettings,
|
||||
) -> std::result::Result<deadpool_postgres::Pool, crate::PostgresBackendError>
|
||||
where
|
||||
T: tokio_postgres::tls::MakeTlsConnect<tokio_postgres::Socket> + std::clone::Clone + std::marker::Send + std::marker::Sync + 'static,
|
||||
T::Stream: std::marker::Send + std::marker::Sync,
|
||||
T::TlsConnect: std::marker::Send + std::marker::Sync,
|
||||
<T::TlsConnect as tokio_postgres::tls::TlsConnect<tokio_postgres::Socket>>::Future: std::marker::Send,
|
||||
{
|
||||
let manager_config = deadpool_postgres::ManagerConfig { recycling_method: deadpool_postgres::RecyclingMethod::Verified };
|
||||
let manager = deadpool_postgres::Manager::from_config(pg_config, tls, manager_config);
|
||||
let built = deadpool_postgres::Pool::builder(manager)
|
||||
.max_size(settings.max_connections as usize)
|
||||
.wait_timeout(std::option::Option::Some(settings.wait_timeout))
|
||||
.create_timeout(std::option::Option::Some(settings.create_timeout))
|
||||
.recycle_timeout(std::option::Option::Some(settings.recycle_timeout))
|
||||
.runtime(deadpool_postgres::Runtime::Tokio1)
|
||||
.build();
|
||||
return match built {
|
||||
std::result::Result::Ok(pool) => std::result::Result::Ok(pool),
|
||||
std::result::Result::Err(_) => std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConfigInvalid, "pool_build")),
|
||||
};
|
||||
}
|
||||
|
||||
fn build_verified_tls() -> std::result::Result<tokio_postgres_rustls::MakeRustlsConnect, crate::PostgresBackendError> {
|
||||
let native = rustls_native_certs::load_native_certs();
|
||||
let native_error_count = native.errors.len();
|
||||
if native.certs.is_empty() {
|
||||
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, native_error_count, "no system TLS roots available for PostgreSQL verify_full");
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::TlsFailed, "native_roots"));
|
||||
}
|
||||
let mut roots = rustls::RootCertStore::empty();
|
||||
let (added, ignored) = roots.add_parsable_certificates(native.certs);
|
||||
if added == 0 {
|
||||
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, native_error_count, ignored, "system TLS roots could not be admitted for PostgreSQL verify_full");
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::TlsFailed, "native_roots"));
|
||||
}
|
||||
if native_error_count > 0 || ignored > 0 {
|
||||
ksp_logging_lib::debug!(target: crate::TRACING_TARGET, added, ignored, native_error_count, "loaded PostgreSQL system TLS roots with partial diagnostics");
|
||||
}
|
||||
let provider = std::sync::Arc::new(rustls::crypto::aws_lc_rs::default_provider());
|
||||
let builder_result = rustls::ClientConfig::builder_with_provider(provider).with_safe_default_protocol_versions();
|
||||
let builder = match builder_result {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::TlsFailed, "protocol_versions"));
|
||||
},
|
||||
};
|
||||
let client_config = builder.with_root_certificates(roots).with_no_client_auth();
|
||||
return std::result::Result::Ok(tokio_postgres_rustls::MakeRustlsConnect::new(client_config));
|
||||
}
|
||||
|
||||
fn map_pool_error(error: deadpool_postgres::PoolError) -> crate::PostgresBackendError {
|
||||
return match error {
|
||||
deadpool_postgres::PoolError::Timeout(timeout_type) => crate::PostgresBackendError::new(
|
||||
crate::PostgresBackendErrorKind::PoolTimeout,
|
||||
match timeout_type {
|
||||
deadpool_postgres::TimeoutType::Wait => "pool_wait",
|
||||
deadpool_postgres::TimeoutType::Create => "pool_create",
|
||||
deadpool_postgres::TimeoutType::Recycle => "pool_recycle",
|
||||
},
|
||||
),
|
||||
deadpool_postgres::PoolError::Backend(_) => crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConnectFailed, "physical_connect"),
|
||||
deadpool_postgres::PoolError::Closed => crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConnectFailed, "pool_closed"),
|
||||
deadpool_postgres::PoolError::NoRuntimeSpecified => crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConfigInvalid, "pool_runtime"),
|
||||
deadpool_postgres::PoolError::PostCreateHook(_) => crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ConnectFailed, "pool_post_create"),
|
||||
};
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "../unit_tests/runtime.rs"]
|
||||
mod tests;
|
||||
Reference in New Issue
Block a user