Files
khadhroony-solana-project/crates/ksp-store-lib/src/store.rs
2026-08-30 14:10:10 +02:00

502 lines
22 KiB
Rust

// file: crates/ksp-store-lib/src/store.rs
// version: 6
/// Opaque common Store runtime facade.
///
/// A successful value is returned only after the selected compiled backend has completed its bounded physical opening path. PostgreSQL pool, client, TLS and
/// driver types remain private to the backend crate.
pub struct Store {
backend_kind: crate::StoreBackendKind,
network: ksp_store_api::RawNetworkId,
#[cfg(feature = "postgres")]
runtime: StoreRuntime,
shutdown_timeout: std::time::Duration,
}
impl Store {
/// Validates settings, selects the requested backend and opens one ready Store instance for exactly one logical network.
///
/// A known backend whose Cargo feature is absent is rejected before any I/O. A successful PostgreSQL result proves that one physical pooled connection has
/// been established under the typed TLS and timeout policy.
pub async fn open(settings: crate::StoreSettings) -> ksp_store_api::Result<Self> {
let validation = settings.validate();
if let std::result::Result::Err(error) = validation {
return std::result::Result::Err(error);
}
let backend_kind = settings.backend_kind();
let network = settings.network().clone();
let shutdown_timeout = settings.shutdown_timeout();
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
backend = backend_kind.code(),
network = network.as_str(),
"opening Store runtime"
);
return match settings.backend() {
crate::StoreBackendSettings::Postgres(postgres) => open_postgres(backend_kind, network, shutdown_timeout, postgres).await,
};
}
/// Returns a safe synchronous runtime snapshot without performing backend I/O.
#[must_use]
pub fn runtime_snapshot(&self) -> crate::StoreRuntimeSnapshot {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => map_postgres_runtime_snapshot(backend.runtime_snapshot(), self.backend_kind, self.network.clone()),
};
}
#[cfg(not(feature = "postgres"))]
{
return crate::StoreRuntimeSnapshot::new(self.backend_kind, self.network.clone(), 0, 0, 0, 0);
}
}
/// Runs the selected backend's lightweight bounded readiness probe and returns only portable redacted diagnostics.
pub async fn health(&self) -> crate::StoreHealthSnapshot {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let snapshot = backend.health().await;
map_postgres_health_snapshot(snapshot, self.backend_kind, self.network.clone())
},
};
}
#[cfg(not(feature = "postgres"))]
{
return crate::StoreHealthSnapshot::new(
crate::StoreHealthState::NotReady,
self.runtime_snapshot(),
std::option::Option::None,
0,
std::option::Option::Some(crate::ERROR_CODE_BACKEND_NOT_COMPILED),
);
}
}
/// Explicitly closes the Store runtime, consumes its facade handle and applies the configured bounded shutdown deadline.
pub async fn close(self) -> ksp_store_api::Result<()> {
let backend_kind = self.backend_kind;
#[cfg(feature = "postgres")]
{
let network = self.network;
let shutdown_timeout = self.shutdown_timeout;
let result = match self.runtime {
StoreRuntime::Postgres(backend) => backend.close(shutdown_timeout).await,
};
return match result {
std::result::Result::Ok(()) => {
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
backend = backend_kind.code(),
network = network.as_str(),
"Store runtime closed"
);
std::result::Result::Ok(())
},
std::result::Result::Err(error) => std::result::Result::Err(map_postgres_error(error, backend_kind, network.as_str())),
};
}
#[cfg(not(feature = "postgres"))]
{
return std::result::Result::Err(unavailable_runtime_error(backend_kind));
}
}
}
impl std::fmt::Debug for Store {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter
.debug_struct("Store")
.field("backend_kind", &self.backend_kind)
.field("network", &self.network)
.field("shutdown_timeout", &self.shutdown_timeout)
.finish_non_exhaustive();
}
}
impl ksp_store_api::RawTransactionRead for Store {
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>>> {
let network_check = validate_operation_network(&self.network, reference.network(), self.backend_kind);
if let std::result::Result::Err(error) = network_check {
return std::boxed::Box::pin(async move {
return std::result::Result::Err(error);
});
}
return std::boxed::Box::pin(async move {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let result = backend.get_raw_transaction(reference).await;
result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
},
};
}
#[cfg(not(feature = "postgres"))]
{
let _ = reference;
return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
}
});
}
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>>> {
let network_check = validate_operation_network(&self.network, query.network(), self.backend_kind);
if let std::result::Result::Err(error) = network_check {
return std::boxed::Box::pin(async move {
return std::result::Result::Err(error);
});
}
return std::boxed::Box::pin(async move {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let result = backend.list_raw_transactions(query).await;
result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
},
};
}
#[cfg(not(feature = "postgres"))]
{
let _ = query;
return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
}
});
}
}
impl ksp_store_api::RawTransactionWrite for Store {
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>> {
let transaction_network = validate_operation_network(&self.network, transaction.reference().network(), self.backend_kind);
if let std::result::Result::Err(error) = transaction_network {
return std::boxed::Box::pin(async move {
return std::result::Result::Err(error);
});
}
let observation_network = validate_operation_network(&self.network, observation.transaction().network(), self.backend_kind);
if let std::result::Result::Err(error) = observation_network {
return std::boxed::Box::pin(async move {
return std::result::Result::Err(error);
});
}
return std::boxed::Box::pin(async move {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let result = backend.persist_raw_transaction_acquisition(transaction, observation, mode).await;
result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
},
};
}
#[cfg(not(feature = "postgres"))]
{
let _ = transaction;
let _ = observation;
let _ = mode;
return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
}
});
}
}
impl ksp_store_api::RawTransactionObservationRead for Store {
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 {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let result = backend.get_raw_transaction_observation(observation_key).await;
result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
},
};
}
#[cfg(not(feature = "postgres"))]
{
let _ = observation_key;
return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
}
});
}
}
impl ksp_store_api::RawTransactionObservationWrite for Store {
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>> {
let network_check = validate_operation_network(&self.network, observation.transaction().network(), self.backend_kind);
if let std::result::Result::Err(error) = network_check {
return std::boxed::Box::pin(async move {
return std::result::Result::Err(error);
});
}
return std::boxed::Box::pin(async move {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let result = backend.record_raw_transaction_observation(observation).await;
result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
},
};
}
#[cfg(not(feature = "postgres"))]
{
let _ = observation;
return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
}
});
}
}
impl ksp_store_api::RawTransactionRetentionRead for Store {
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>>> {
let network_check = validate_operation_network(&self.network, reference.network(), self.backend_kind);
if let std::result::Result::Err(error) = network_check {
return std::boxed::Box::pin(async move {
return std::result::Result::Err(error);
});
}
return std::boxed::Box::pin(async move {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let result = backend.get_raw_transaction_retention_state(reference).await;
result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
},
};
}
#[cfg(not(feature = "postgres"))]
{
let _ = reference;
return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
}
});
}
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>>> {
let network_check = validate_operation_network(&self.network, reference.network(), self.backend_kind);
if let std::result::Result::Err(error) = network_check {
return std::boxed::Box::pin(async move {
return std::result::Result::Err(error);
});
}
return std::boxed::Box::pin(async move {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let result = backend.get_raw_transaction_tombstone(reference).await;
result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
},
};
}
#[cfg(not(feature = "postgres"))]
{
let _ = reference;
return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
}
});
}
}
impl ksp_store_api::RawTransactionRetentionWrite for Store {
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>> {
let network_check = validate_operation_network(&self.network, transition.reference().network(), self.backend_kind);
if let std::result::Result::Err(error) = network_check {
return std::boxed::Box::pin(async move {
return std::result::Result::Err(error);
});
}
return std::boxed::Box::pin(async move {
#[cfg(feature = "postgres")]
{
return match &self.runtime {
StoreRuntime::Postgres(backend) => {
let result = backend.transition_raw_transaction_retention(transition).await;
result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
},
};
}
#[cfg(not(feature = "postgres"))]
{
let _ = transition;
return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
}
});
}
}
fn validate_operation_network(
store_network: &ksp_store_api::RawNetworkId,
operation_network: &ksp_store_api::RawNetworkId,
backend_kind: crate::StoreBackendKind,
) -> ksp_store_api::Result<()> {
if store_network != operation_network {
return std::result::Result::Err(
ksp_store_api::Error::new(crate::ERROR_CODE_WRONG_NETWORK, "Store operation targeted a different logical network")
.with_context("backend", backend_kind.code())
.with_context("network", store_network.as_str()),
);
}
return std::result::Result::Ok(());
}
#[cfg(feature = "postgres")]
enum StoreRuntime {
Postgres(ksp_store_postgres_lib::PostgresBackend),
}
#[cfg(feature = "postgres")]
async fn open_postgres(
backend_kind: crate::StoreBackendKind,
network: ksp_store_api::RawNetworkId,
shutdown_timeout: std::time::Duration,
settings: &crate::PostgresStoreSettings,
) -> ksp_store_api::Result<Store> {
let bootstrap = settings.bootstrap();
let pool = settings.pool();
let tls_mode = match settings.tls_mode() {
crate::PostgresTlsMode::Disabled => ksp_store_postgres_lib::PostgresBackendTlsMode::Disabled,
crate::PostgresTlsMode::VerifyFull => ksp_store_postgres_lib::PostgresBackendTlsMode::VerifyFull,
};
let backend_settings = ksp_store_postgres_lib::PostgresBackendSettings::with_schema_policy(
network.clone(),
settings.connection_uri(),
pool.max_connections(),
pool.connect_timeout(),
pool.wait_timeout(),
pool.create_timeout(),
pool.recycle_timeout(),
tls_mode,
bootstrap.schema_autocreate(),
bootstrap.schema_autoupdate(),
bootstrap.migration_timeout(),
bootstrap.migration_lock_timeout(),
);
let opened = ksp_store_postgres_lib::PostgresBackend::open(backend_settings).await;
return match opened {
std::result::Result::Ok(backend) => {
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
backend = backend_kind.code(),
network = network.as_str(),
"Store backend is physically ready"
);
std::result::Result::Ok(Store { backend_kind, network, runtime: StoreRuntime::Postgres(backend), shutdown_timeout })
},
std::result::Result::Err(error) => std::result::Result::Err(map_postgres_error(error, backend_kind, network.as_str())),
};
}
#[cfg(not(feature = "postgres"))]
async fn open_postgres(
backend_kind: crate::StoreBackendKind,
_network: ksp_store_api::RawNetworkId,
_shutdown_timeout: std::time::Duration,
_settings: &crate::PostgresStoreSettings,
) -> ksp_store_api::Result<Store> {
return std::result::Result::Err(unavailable_runtime_error(backend_kind));
}
#[cfg(feature = "postgres")]
fn map_postgres_error(error: ksp_store_postgres_lib::PostgresBackendError, backend_kind: crate::StoreBackendKind, network: &str) -> ksp_store_api::Error {
let code = postgres_error_code(error.kind());
return ksp_store_api::Error::new(code, "PostgreSQL Store backend operation failed")
.with_context("backend", backend_kind.code())
.with_context("network", network)
.with_context("phase", error.phase());
}
#[cfg(feature = "postgres")]
fn map_postgres_runtime_snapshot(
snapshot: ksp_store_postgres_lib::PostgresBackendRuntimeSnapshot,
backend_kind: crate::StoreBackendKind,
network: ksp_store_api::RawNetworkId,
) -> crate::StoreRuntimeSnapshot {
return crate::StoreRuntimeSnapshot::new(
backend_kind,
network,
snapshot.pool_capacity(),
snapshot.pool_size(),
snapshot.pool_available(),
snapshot.pool_waiting(),
);
}
#[cfg(feature = "postgres")]
fn map_postgres_health_snapshot(
snapshot: ksp_store_postgres_lib::PostgresBackendHealthSnapshot,
backend_kind: crate::StoreBackendKind,
network: ksp_store_api::RawNetworkId,
) -> crate::StoreHealthSnapshot {
let state = if snapshot.is_ready() { crate::StoreHealthState::Ready } else { crate::StoreHealthState::NotReady };
let error_code = snapshot.error_kind().map(|kind| return postgres_error_code(kind));
let runtime = map_postgres_runtime_snapshot(snapshot.runtime().clone(), backend_kind, network);
return crate::StoreHealthSnapshot::new(state, runtime, snapshot.migration_version(), snapshot.pending_migration_count(), error_code);
}
#[cfg(feature = "postgres")]
fn postgres_error_code(kind: ksp_store_postgres_lib::PostgresBackendErrorKind) -> ksp_store_api::ErrorCode {
return match kind {
ksp_store_postgres_lib::PostgresBackendErrorKind::ConfigInvalid => crate::ERROR_CODE_POSTGRES_CONFIG_INVALID,
ksp_store_postgres_lib::PostgresBackendErrorKind::ConnectFailed => crate::ERROR_CODE_POSTGRES_CONNECT_FAILED,
ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict => ksp_store_api::ERROR_CODE_RAW_CONFLICT,
ksp_store_postgres_lib::PostgresBackendErrorKind::DataInvalid => crate::ERROR_CODE_POSTGRES_DATA_INVALID,
ksp_store_postgres_lib::PostgresBackendErrorKind::HealthFailed => crate::ERROR_CODE_POSTGRES_HEALTH_FAILED,
ksp_store_postgres_lib::PostgresBackendErrorKind::PoolTimeout => crate::ERROR_CODE_POSTGRES_POOL_TIMEOUT,
ksp_store_postgres_lib::PostgresBackendErrorKind::MigrationFailed => crate::ERROR_CODE_POSTGRES_MIGRATION_FAILED,
ksp_store_postgres_lib::PostgresBackendErrorKind::MigrationMismatch => crate::ERROR_CODE_POSTGRES_MIGRATION_MISMATCH,
ksp_store_postgres_lib::PostgresBackendErrorKind::PageLimitUnsupported => crate::ERROR_CODE_POSTGRES_PAGE_LIMIT_UNSUPPORTED,
ksp_store_postgres_lib::PostgresBackendErrorKind::QueryInvalid => ksp_store_api::ERROR_CODE_RAW_QUERY_INVALID,
ksp_store_postgres_lib::PostgresBackendErrorKind::ReadFailed => crate::ERROR_CODE_POSTGRES_READ_FAILED,
ksp_store_postgres_lib::PostgresBackendErrorKind::ReferenceNotFound => crate::ERROR_CODE_RAW_REFERENCE_NOT_FOUND,
ksp_store_postgres_lib::PostgresBackendErrorKind::RetentionCompactionUnsupported => crate::ERROR_CODE_POSTGRES_RETENTION_COMPACTION_UNSUPPORTED,
ksp_store_postgres_lib::PostgresBackendErrorKind::SchemaNewer => crate::ERROR_CODE_POSTGRES_SCHEMA_NEWER,
ksp_store_postgres_lib::PostgresBackendErrorKind::ShutdownTimeout => crate::ERROR_CODE_SHUTDOWN_TIMEOUT,
ksp_store_postgres_lib::PostgresBackendErrorKind::TlsFailed => crate::ERROR_CODE_POSTGRES_TLS_FAILED,
ksp_store_postgres_lib::PostgresBackendErrorKind::WriteFailed => crate::ERROR_CODE_POSTGRES_WRITE_FAILED,
ksp_store_postgres_lib::PostgresBackendErrorKind::WrongNetwork => crate::ERROR_CODE_WRONG_NETWORK,
_ => crate::ERROR_CODE_BACKEND_OPEN_FAILED,
};
}
#[cfg(not(feature = "postgres"))]
fn unavailable_runtime_error(backend_kind: crate::StoreBackendKind) -> ksp_store_api::Error {
return ksp_store_api::Error::new(crate::ERROR_CODE_BACKEND_NOT_COMPILED, "Selected Store backend is not compiled")
.with_context("backend", backend_kind.code());
}
#[cfg(test)]
#[path = "../unit_tests/store.rs"]
mod tests;