878 lines
40 KiB
Rust
878 lines
40 KiB
Rust
// file: crates/ksp-store-postgres-lib/src/runtime.rs
|
|
// version: 18
|
|
|
|
const APPLICATION_NAME: &str = "ksp-store";
|
|
const MAX_CONNECTION_URI_BYTES: usize = 4_096;
|
|
const SHUTDOWN_POLL_INTERVAL_MS: u64 = 10;
|
|
|
|
/// Safe PostgreSQL runtime counters exported only through the narrow backend bridge.
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct PostgresBackendRuntimeSnapshot {
|
|
pool_available: u32,
|
|
pool_capacity: u32,
|
|
pool_size: u32,
|
|
pool_waiting: u32,
|
|
}
|
|
|
|
impl PostgresBackendRuntimeSnapshot {
|
|
/// Creates a safe pool-counter projection from already bounded values.
|
|
#[must_use]
|
|
pub(crate) const fn new(pool_capacity: u32, pool_size: u32, pool_available: u32, pool_waiting: u32) -> Self {
|
|
return Self { pool_available, pool_capacity, pool_size, pool_waiting };
|
|
}
|
|
|
|
/// Returns the number of currently available pooled PostgreSQL clients.
|
|
#[must_use]
|
|
pub const fn pool_available(&self) -> u32 {
|
|
return self.pool_available;
|
|
}
|
|
|
|
/// Returns the configured maximum pooled PostgreSQL client count.
|
|
#[must_use]
|
|
pub const fn pool_capacity(&self) -> u32 {
|
|
return self.pool_capacity;
|
|
}
|
|
|
|
/// Returns the current pooled PostgreSQL client count.
|
|
#[must_use]
|
|
pub const fn pool_size(&self) -> u32 {
|
|
return self.pool_size;
|
|
}
|
|
|
|
/// Returns the number of tasks currently waiting for a pooled PostgreSQL client.
|
|
#[must_use]
|
|
pub const fn pool_waiting(&self) -> u32 {
|
|
return self.pool_waiting;
|
|
}
|
|
}
|
|
|
|
/// Safe PostgreSQL readiness projection returned to the common Store facade.
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct PostgresBackendHealthSnapshot {
|
|
error_kind: std::option::Option<crate::PostgresBackendErrorKind>,
|
|
migration_version: std::option::Option<u64>,
|
|
pending_migration_count: u32,
|
|
ready: bool,
|
|
runtime: PostgresBackendRuntimeSnapshot,
|
|
}
|
|
|
|
impl PostgresBackendHealthSnapshot {
|
|
/// Creates one successful safe readiness projection.
|
|
#[must_use]
|
|
pub(crate) const fn ready(runtime: PostgresBackendRuntimeSnapshot, migration_version: u64, pending_migration_count: u32) -> Self {
|
|
return Self {
|
|
error_kind: std::option::Option::None,
|
|
migration_version: std::option::Option::Some(migration_version),
|
|
pending_migration_count,
|
|
ready: true,
|
|
runtime,
|
|
};
|
|
}
|
|
|
|
/// Creates one failed safe readiness projection from a classified backend error.
|
|
#[must_use]
|
|
pub(crate) const fn not_ready(
|
|
runtime: PostgresBackendRuntimeSnapshot,
|
|
migration_version: std::option::Option<u64>,
|
|
pending_migration_count: u32,
|
|
error_kind: crate::PostgresBackendErrorKind,
|
|
) -> Self {
|
|
return Self {
|
|
error_kind: std::option::Option::Some(error_kind),
|
|
migration_version,
|
|
pending_migration_count,
|
|
ready: false,
|
|
runtime,
|
|
};
|
|
}
|
|
|
|
/// Returns the safe backend error classification when readiness could not be proven.
|
|
#[must_use]
|
|
pub const fn error_kind(&self) -> std::option::Option<crate::PostgresBackendErrorKind> {
|
|
return self.error_kind;
|
|
}
|
|
|
|
/// Returns whether the latest bounded PostgreSQL probe proved readiness.
|
|
#[must_use]
|
|
pub const fn is_ready(&self) -> bool {
|
|
return self.ready;
|
|
}
|
|
|
|
/// Returns the migration version observed by the readiness probe when available.
|
|
#[must_use]
|
|
pub const fn migration_version(&self) -> std::option::Option<u64> {
|
|
return self.migration_version;
|
|
}
|
|
|
|
/// Returns the number of embedded migrations newer than the observed applied version.
|
|
#[must_use]
|
|
pub const fn pending_migration_count(&self) -> u32 {
|
|
return self.pending_migration_count;
|
|
}
|
|
|
|
/// Returns the safe PostgreSQL pool counters captured for this probe.
|
|
#[must_use]
|
|
pub const fn runtime(&self) -> &PostgresBackendRuntimeSnapshot {
|
|
return &self.runtime;
|
|
}
|
|
}
|
|
|
|
/// 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,
|
|
migration_lock_timeout: std::time::Duration,
|
|
migration_timeout: std::time::Duration,
|
|
network: ksp_store_api::RawNetworkId,
|
|
recycle_timeout: std::time::Duration,
|
|
schema_autocreate: bool,
|
|
schema_autoupdate: bool,
|
|
tls_mode: PostgresBackendTlsMode,
|
|
wait_timeout: std::time::Duration,
|
|
}
|
|
|
|
impl PostgresBackendSettings {
|
|
/// Creates the physical PostgreSQL settings bridge using the legacy single migration switch.
|
|
#[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,
|
|
auto_migrate: bool,
|
|
migration_timeout: std::time::Duration,
|
|
migration_lock_timeout: std::time::Duration,
|
|
) -> Self {
|
|
return Self::with_schema_policy(
|
|
network,
|
|
connection_uri,
|
|
max_connections,
|
|
connect_timeout,
|
|
wait_timeout,
|
|
create_timeout,
|
|
recycle_timeout,
|
|
tls_mode,
|
|
auto_migrate,
|
|
auto_migrate,
|
|
migration_timeout,
|
|
migration_lock_timeout,
|
|
);
|
|
}
|
|
|
|
/// Creates the physical PostgreSQL settings bridge with independent schema creation and update policies.
|
|
#[must_use]
|
|
pub fn with_schema_policy(
|
|
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,
|
|
schema_autocreate: bool,
|
|
schema_autoupdate: bool,
|
|
migration_timeout: std::time::Duration,
|
|
migration_lock_timeout: std::time::Duration,
|
|
) -> Self {
|
|
return Self {
|
|
connect_timeout,
|
|
connection_uri: connection_uri.into(),
|
|
create_timeout,
|
|
max_connections,
|
|
migration_lock_timeout,
|
|
migration_timeout,
|
|
network,
|
|
recycle_timeout,
|
|
schema_autocreate,
|
|
schema_autoupdate,
|
|
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("schema_autocreate", &self.schema_autocreate)
|
|
.field("schema_autoupdate", &self.schema_autoupdate)
|
|
.field("max_connections", &self.max_connections)
|
|
.field("migration_timeout", &self.migration_timeout)
|
|
.field("migration_lock_timeout", &self.migration_lock_timeout)
|
|
.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;
|
|
let mut client = match probe {
|
|
std::result::Result::Ok(value) => value,
|
|
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"
|
|
);
|
|
let bootstrap_result = crate::bootstrap(
|
|
&mut client,
|
|
settings.network(),
|
|
settings.schema_autocreate,
|
|
settings.schema_autoupdate,
|
|
settings.migration_timeout,
|
|
settings.migration_lock_timeout,
|
|
)
|
|
.await;
|
|
if let std::result::Result::Err(error) = bootstrap_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
drop(client);
|
|
ksp_logging_lib::debug!(
|
|
target: crate::TRACING_TARGET,
|
|
network = settings.network().as_str(),
|
|
schema_autocreate = settings.schema_autocreate,
|
|
schema_autoupdate = settings.schema_autoupdate,
|
|
"PostgreSQL Store migration/bootstrap foundation verified"
|
|
);
|
|
return std::result::Result::Ok(Self { network: settings.network, pool });
|
|
}
|
|
|
|
/// Returns a safe synchronous snapshot of bounded pool counters without performing PostgreSQL I/O.
|
|
#[must_use]
|
|
pub fn runtime_snapshot(&self) -> PostgresBackendRuntimeSnapshot {
|
|
return runtime_snapshot_from_status(self.pool.status());
|
|
}
|
|
|
|
/// Runs a bounded lightweight PostgreSQL readiness probe and returns only safe classified diagnostics.
|
|
pub async fn health(&self) -> PostgresBackendHealthSnapshot {
|
|
return crate::probe_health(&self.pool).await;
|
|
}
|
|
|
|
/// Reads one complete canonical RAW account state without exposing physical PostgreSQL row types.
|
|
pub async fn get_raw_account_state(
|
|
&self,
|
|
reference: &ksp_store_api::RawAccountStateReference,
|
|
) -> std::result::Result<std::option::Option<ksp_store_api::RawAccountState>, crate::PostgresBackendError> {
|
|
return crate::get_raw_account_state(&self.pool, &self.network, reference).await;
|
|
}
|
|
|
|
/// Lists deterministic canonical RAW account-state references with a backend-owned opaque continuation cursor.
|
|
pub async fn list_raw_account_states(
|
|
&self,
|
|
query: &ksp_store_api::RawAccountStateQuery,
|
|
) -> std::result::Result<ksp_store_api::RawPage<ksp_store_api::RawAccountStateReference>, crate::PostgresBackendError> {
|
|
return crate::list_raw_account_states(&self.pool, &self.network, query).await;
|
|
}
|
|
|
|
/// Inspects one data-free random-access RAW account-state window with exact logical counts.
|
|
pub async fn inspect_raw_account_states(
|
|
&self,
|
|
query: &ksp_store_api::RawAccountStateInspectionQuery,
|
|
) -> std::result::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawAccountStateSummary>, crate::PostgresBackendError> {
|
|
return crate::inspect_raw_account_states(&self.pool, &self.network, query).await;
|
|
}
|
|
|
|
/// Inspects one safe random-access RAW account-observation window with exact logical counts.
|
|
pub async fn inspect_raw_account_observations(
|
|
&self,
|
|
query: &ksp_store_api::RawAccountObservationInspectionQuery,
|
|
) -> std::result::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawAccountObservationSummary>, crate::PostgresBackendError> {
|
|
return crate::inspect_raw_account_observations(&self.pool, &self.network, query).await;
|
|
}
|
|
|
|
/// Reads one persisted RAW account observation by producer-owned idempotence key.
|
|
pub async fn get_raw_account_observation(
|
|
&self,
|
|
observation_key: &ksp_store_api::RawObservationKey,
|
|
) -> std::result::Result<std::option::Option<ksp_store_api::RawAccountObservation>, crate::PostgresBackendError> {
|
|
return crate::get_raw_account_observation(&self.pool, &self.network, observation_key).await;
|
|
}
|
|
|
|
/// Persists one complete RAW account state and its acquisition observation atomically.
|
|
pub async fn persist_raw_account_acquisition(
|
|
&self,
|
|
state: ksp_store_api::RawAccountState,
|
|
observation: ksp_store_api::RawAccountObservation,
|
|
) -> std::result::Result<ksp_store_api::RawAcquisitionWriteOutcome, crate::PostgresBackendError> {
|
|
return crate::persist_raw_account_acquisition(&self.pool, &self.network, state, observation).await;
|
|
}
|
|
|
|
/// Persists one additional RAW account observation for an already durable account state.
|
|
pub async fn record_raw_account_observation(
|
|
&self,
|
|
observation: ksp_store_api::RawAccountObservation,
|
|
) -> std::result::Result<ksp_store_api::RawObservationWriteOutcome, crate::PostgresBackendError> {
|
|
return crate::record_raw_account_observation(&self.pool, &self.network, observation).await;
|
|
}
|
|
|
|
/// Reads one canonical RAW transaction without exposing physical PostgreSQL row types.
|
|
pub async fn get_raw_transaction(
|
|
&self,
|
|
reference: &ksp_store_api::RawTransactionReference,
|
|
) -> std::result::Result<std::option::Option<ksp_store_api::RawTransaction>, crate::PostgresBackendError> {
|
|
return crate::get_raw_transaction(&self.pool, &self.network, reference).await;
|
|
}
|
|
|
|
/// Lists deterministic canonical RAW transaction references with a backend-owned opaque continuation cursor.
|
|
pub async fn list_raw_transactions(
|
|
&self,
|
|
query: &ksp_store_api::RawTransactionQuery,
|
|
) -> std::result::Result<ksp_store_api::RawPage<ksp_store_api::RawTransactionReference>, crate::PostgresBackendError> {
|
|
return crate::list_raw_transactions(&self.pool, &self.network, query).await;
|
|
}
|
|
|
|
/// Inspects one payload-free random-access RAW transaction window with exact logical counts.
|
|
pub async fn inspect_raw_transactions(
|
|
&self,
|
|
query: &ksp_store_api::RawTransactionInspectionQuery,
|
|
) -> std::result::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawTransactionSummary>, crate::PostgresBackendError> {
|
|
return crate::inspect_raw_transactions(&self.pool, &self.network, query).await;
|
|
}
|
|
|
|
/// Inspects one safe random-access RAW transaction-observation window with exact logical counts.
|
|
pub async fn inspect_raw_transaction_observations(
|
|
&self,
|
|
query: &ksp_store_api::RawTransactionObservationInspectionQuery,
|
|
) -> std::result::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawTransactionObservationSummary>, crate::PostgresBackendError> {
|
|
return crate::inspect_raw_transaction_observations(&self.pool, &self.network, query).await;
|
|
}
|
|
|
|
/// Reads one persisted RAW transaction observation by producer-owned idempotence key.
|
|
pub async fn get_raw_transaction_observation(
|
|
&self,
|
|
observation_key: &ksp_store_api::RawObservationKey,
|
|
) -> std::result::Result<std::option::Option<ksp_store_api::RawTransactionObservation>, crate::PostgresBackendError> {
|
|
return crate::get_raw_transaction_observation(&self.pool, &self.network, observation_key).await;
|
|
}
|
|
|
|
/// Reads the retention state of one canonical RAW transaction identity.
|
|
pub async fn get_raw_transaction_retention_state(
|
|
&self,
|
|
reference: &ksp_store_api::RawTransactionReference,
|
|
) -> std::result::Result<std::option::Option<ksp_store_api::RawRetentionState>, crate::PostgresBackendError> {
|
|
return crate::get_raw_transaction_retention_state(&self.pool, &self.network, reference).await;
|
|
}
|
|
|
|
/// Reads the minimal durable tombstone only when one RAW transaction is purged.
|
|
pub async fn get_raw_transaction_tombstone(
|
|
&self,
|
|
reference: &ksp_store_api::RawTransactionReference,
|
|
) -> std::result::Result<std::option::Option<ksp_store_api::RawTransactionTombstone>, crate::PostgresBackendError> {
|
|
return crate::get_raw_transaction_tombstone(&self.pool, &self.network, reference).await;
|
|
}
|
|
|
|
/// Persists one canonical RAW transaction and its acquisition observation atomically.
|
|
pub async fn persist_raw_transaction_acquisition(
|
|
&self,
|
|
raw_transaction: ksp_store_api::RawTransaction,
|
|
observation: ksp_store_api::RawTransactionObservation,
|
|
mode: ksp_store_api::RawTransactionAcquisitionMode,
|
|
) -> std::result::Result<ksp_store_api::RawAcquisitionWriteOutcome, crate::PostgresBackendError> {
|
|
return crate::persist_raw_transaction_acquisition(&self.pool, &self.network, raw_transaction, observation, mode).await;
|
|
}
|
|
|
|
/// Persists one additional acquisition observation for an existing RAW transaction.
|
|
pub async fn record_raw_transaction_observation(
|
|
&self,
|
|
observation: ksp_store_api::RawTransactionObservation,
|
|
) -> std::result::Result<ksp_store_api::RawObservationWriteOutcome, crate::PostgresBackendError> {
|
|
return crate::record_raw_transaction_observation(&self.pool, &self.network, observation).await;
|
|
}
|
|
|
|
/// Applies one policy-authorized atomic RAW transaction retention transition.
|
|
pub async fn transition_raw_transaction_retention(
|
|
&self,
|
|
transition: ksp_store_api::RawTransactionRetentionTransition,
|
|
) -> std::result::Result<ksp_store_api::RawRetentionWriteOutcome, crate::PostgresBackendError> {
|
|
return crate::transition_raw_transaction_retention(&self.pool, &self.network, transition).await;
|
|
}
|
|
|
|
/// 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 ksp_store_api::RawAccountObservationInspectionRead for PostgresBackend {
|
|
fn inspect_raw_account_observations<'a>(
|
|
&'a self,
|
|
query: &'a ksp_store_api::RawAccountObservationInspectionQuery,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawAccountObservationSummary>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::inspect_raw_account_observations(self, query).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawAccountObservationRead for PostgresBackend {
|
|
fn get_raw_account_observation<'a>(
|
|
&'a self,
|
|
observation_key: &'a ksp_store_api::RawObservationKey,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawAccountObservation>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::get_raw_account_observation(self, observation_key).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawAccountObservationWrite for PostgresBackend {
|
|
fn record_raw_account_observation<'a>(
|
|
&'a self,
|
|
observation: ksp_store_api::RawAccountObservation,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawObservationWriteOutcome>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::record_raw_account_observation(self, observation).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawAccountStateRead for PostgresBackend {
|
|
fn get_raw_account_state<'a>(
|
|
&'a self,
|
|
reference: &'a ksp_store_api::RawAccountStateReference,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawAccountState>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::get_raw_account_state(self, reference).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
|
|
fn list_raw_account_states<'a>(
|
|
&'a self,
|
|
query: &'a ksp_store_api::RawAccountStateQuery,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawPage<ksp_store_api::RawAccountStateReference>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::list_raw_account_states(self, query).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawAccountStateInspectionRead for PostgresBackend {
|
|
fn inspect_raw_account_states<'a>(
|
|
&'a self,
|
|
query: &'a ksp_store_api::RawAccountStateInspectionQuery,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawAccountStateSummary>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::inspect_raw_account_states(self, query).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawAccountStateWrite for PostgresBackend {
|
|
fn persist_raw_account_acquisition<'a>(
|
|
&'a self,
|
|
state: ksp_store_api::RawAccountState,
|
|
observation: ksp_store_api::RawAccountObservation,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawAcquisitionWriteOutcome>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::persist_raw_account_acquisition(self, state, observation).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawTransactionRead for PostgresBackend {
|
|
fn get_raw_transaction<'a>(
|
|
&'a self,
|
|
reference: &'a ksp_store_api::RawTransactionReference,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawTransaction>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::get_raw_transaction(self, reference).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
|
|
fn list_raw_transactions<'a>(
|
|
&'a self,
|
|
query: &'a ksp_store_api::RawTransactionQuery,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawPage<ksp_store_api::RawTransactionReference>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::list_raw_transactions(self, query).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawTransactionInspectionRead for PostgresBackend {
|
|
fn inspect_raw_transactions<'a>(
|
|
&'a self,
|
|
query: &'a ksp_store_api::RawTransactionInspectionQuery,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawTransactionSummary>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::inspect_raw_transactions(self, query).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawTransactionWrite for PostgresBackend {
|
|
fn persist_raw_transaction_acquisition<'a>(
|
|
&'a self,
|
|
transaction: ksp_store_api::RawTransaction,
|
|
observation: ksp_store_api::RawTransactionObservation,
|
|
mode: ksp_store_api::RawTransactionAcquisitionMode,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawAcquisitionWriteOutcome>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::persist_raw_transaction_acquisition(self, transaction, observation, mode).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawTransactionObservationInspectionRead for PostgresBackend {
|
|
fn inspect_raw_transaction_observations<'a>(
|
|
&'a self,
|
|
query: &'a ksp_store_api::RawTransactionObservationInspectionQuery,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawTransactionObservationSummary>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::inspect_raw_transaction_observations(self, query).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawTransactionObservationRead for PostgresBackend {
|
|
fn get_raw_transaction_observation<'a>(
|
|
&'a self,
|
|
observation_key: &'a ksp_store_api::RawObservationKey,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawTransactionObservation>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::get_raw_transaction_observation(self, observation_key).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawTransactionObservationWrite for PostgresBackend {
|
|
fn record_raw_transaction_observation<'a>(
|
|
&'a self,
|
|
observation: ksp_store_api::RawTransactionObservation,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawObservationWriteOutcome>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::record_raw_transaction_observation(self, observation).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawTransactionRetentionRead for PostgresBackend {
|
|
fn get_raw_transaction_retention_state<'a>(
|
|
&'a self,
|
|
reference: &'a ksp_store_api::RawTransactionReference,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawRetentionState>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::get_raw_transaction_retention_state(self, reference).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
|
|
fn get_raw_transaction_tombstone<'a>(
|
|
&'a self,
|
|
reference: &'a ksp_store_api::RawTransactionReference,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawTransactionTombstone>>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::get_raw_transaction_tombstone(self, reference).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
impl ksp_store_api::RawTransactionRetentionWrite for PostgresBackend {
|
|
fn transition_raw_transaction_retention<'a>(
|
|
&'a self,
|
|
transition: ksp_store_api::RawTransactionRetentionTransition,
|
|
) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawRetentionWriteOutcome>> {
|
|
return std::boxed::Box::pin(async move {
|
|
let result = PostgresBackend::transition_raw_transaction_retention(self, transition).await;
|
|
return result.map_err(map_capability_error);
|
|
});
|
|
}
|
|
}
|
|
|
|
fn map_capability_error(error: crate::PostgresBackendError) -> ksp_store_api::Error {
|
|
let code = match error.kind() {
|
|
crate::PostgresBackendErrorKind::ConfigInvalid => ksp_store_api::ErrorCode::new("store", "postgres_config_invalid"),
|
|
crate::PostgresBackendErrorKind::ConnectFailed => ksp_store_api::ErrorCode::new("store", "postgres_connect_failed"),
|
|
crate::PostgresBackendErrorKind::Conflict => ksp_store_api::ERROR_CODE_RAW_CONFLICT,
|
|
crate::PostgresBackendErrorKind::DataInvalid => ksp_store_api::ErrorCode::new("store", "postgres_data_invalid"),
|
|
crate::PostgresBackendErrorKind::HealthFailed => ksp_store_api::ErrorCode::new("store", "postgres_health_failed"),
|
|
crate::PostgresBackendErrorKind::MigrationFailed => ksp_store_api::ErrorCode::new("store", "postgres_migration_failed"),
|
|
crate::PostgresBackendErrorKind::MigrationMismatch => ksp_store_api::ErrorCode::new("store", "postgres_migration_mismatch"),
|
|
crate::PostgresBackendErrorKind::PageLimitUnsupported => ksp_store_api::ErrorCode::new("store", "postgres_page_limit_unsupported"),
|
|
crate::PostgresBackendErrorKind::PoolTimeout => ksp_store_api::ErrorCode::new("store", "postgres_pool_timeout"),
|
|
crate::PostgresBackendErrorKind::QueryInvalid => ksp_store_api::ERROR_CODE_RAW_QUERY_INVALID,
|
|
crate::PostgresBackendErrorKind::ReadFailed => ksp_store_api::ErrorCode::new("store", "postgres_read_failed"),
|
|
crate::PostgresBackendErrorKind::ReferenceNotFound => ksp_store_api::ErrorCode::new("store", "raw_reference_not_found"),
|
|
crate::PostgresBackendErrorKind::RetentionCompactionUnsupported => crate::ERROR_CODE_POSTGRES_RETENTION_COMPACTION_UNSUPPORTED,
|
|
crate::PostgresBackendErrorKind::SchemaNewer => ksp_store_api::ErrorCode::new("store", "postgres_schema_newer"),
|
|
crate::PostgresBackendErrorKind::ShutdownTimeout => ksp_store_api::ErrorCode::new("store", "shutdown_timeout"),
|
|
crate::PostgresBackendErrorKind::TlsFailed => ksp_store_api::ErrorCode::new("store", "postgres_tls_failed"),
|
|
crate::PostgresBackendErrorKind::WriteFailed => ksp_store_api::ErrorCode::new("store", "postgres_write_failed"),
|
|
crate::PostgresBackendErrorKind::WrongNetwork => ksp_store_api::ErrorCode::new("store", "wrong_network"),
|
|
};
|
|
return ksp_store_api::Error::new(code, "PostgreSQL Store capability operation failed")
|
|
.with_context("backend", "postgres")
|
|
.with_context("phase", error.phase());
|
|
}
|
|
|
|
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));
|
|
}
|
|
|
|
/// Maps one Deadpool acquisition error into a redacted backend classification.
|
|
pub(crate) 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"),
|
|
};
|
|
}
|
|
|
|
/// Converts Deadpool status into a bounded safe backend runtime projection.
|
|
pub(crate) fn runtime_snapshot_from_status(status: deadpool_postgres::Status) -> PostgresBackendRuntimeSnapshot {
|
|
return PostgresBackendRuntimeSnapshot::new(
|
|
bounded_count(status.max_size),
|
|
bounded_count(status.size),
|
|
bounded_count(status.available),
|
|
bounded_count(status.waiting),
|
|
);
|
|
}
|
|
|
|
fn bounded_count(value: usize) -> u32 {
|
|
if value > u32::MAX as usize {
|
|
return u32::MAX;
|
|
}
|
|
return value as u32;
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "../unit_tests/runtime.rs"]
|
|
mod tests;
|