102 lines
4.5 KiB
Rust
102 lines
4.5 KiB
Rust
// file: crates/ksp-store-postgres-lib/src/health.rs
|
|
// version: 1
|
|
|
|
const DEFAULT_HEALTH_TIMEOUT_MS: u64 = 5_000;
|
|
const MIGRATION_VERSION_SQL: &str = "SELECT COALESCE(MAX(version), -1)::BIGINT FROM ksp_store_schema_migrations";
|
|
const READINESS_SQL: &str = "SELECT 1::BIGINT";
|
|
|
|
/// Runs one bounded lightweight readiness probe and returns only safe classified diagnostics.
|
|
pub(crate) async fn probe_health(pool: &deadpool_postgres::Pool) -> crate::PostgresBackendHealthSnapshot {
|
|
let runtime = crate::runtime_snapshot_from_status(pool.status());
|
|
let timeouts = pool.timeouts();
|
|
let timeout = match timeouts.wait {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => std::time::Duration::from_millis(DEFAULT_HEALTH_TIMEOUT_MS),
|
|
};
|
|
let bounded = tokio::time::timeout(timeout, probe_health_inner(pool, runtime.clone())).await;
|
|
return match bounded {
|
|
std::result::Result::Ok(snapshot) => snapshot,
|
|
std::result::Result::Err(_) => {
|
|
crate::PostgresBackendHealthSnapshot::not_ready(runtime, std::option::Option::None, 0, crate::PostgresBackendErrorKind::HealthFailed)
|
|
},
|
|
};
|
|
}
|
|
|
|
async fn probe_health_inner(pool: &deadpool_postgres::Pool, runtime: crate::PostgresBackendRuntimeSnapshot) -> crate::PostgresBackendHealthSnapshot {
|
|
let client_result = pool.get().await;
|
|
let client = match client_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => {
|
|
let classified = crate::map_pool_error(error);
|
|
return crate::PostgresBackendHealthSnapshot::not_ready(runtime, std::option::Option::None, 0, classified.kind());
|
|
},
|
|
};
|
|
let readiness_result = client.query_one(READINESS_SQL, &[]).await;
|
|
let readiness_row = match readiness_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => {
|
|
return crate::PostgresBackendHealthSnapshot::not_ready(runtime, std::option::Option::None, 0, crate::PostgresBackendErrorKind::HealthFailed);
|
|
},
|
|
};
|
|
let readiness_value = match readiness_row.try_get::<usize, i64>(0) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => {
|
|
return crate::PostgresBackendHealthSnapshot::not_ready(runtime, std::option::Option::None, 0, crate::PostgresBackendErrorKind::HealthFailed);
|
|
},
|
|
};
|
|
if readiness_value != 1 {
|
|
return crate::PostgresBackendHealthSnapshot::not_ready(runtime, std::option::Option::None, 0, crate::PostgresBackendErrorKind::HealthFailed);
|
|
}
|
|
let migration_result = client.query_one(MIGRATION_VERSION_SQL, &[]).await;
|
|
let migration_row = match migration_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => {
|
|
return crate::PostgresBackendHealthSnapshot::not_ready(runtime, std::option::Option::None, 0, crate::PostgresBackendErrorKind::HealthFailed);
|
|
},
|
|
};
|
|
let migration_value = migration_row.try_get::<usize, i64>(0);
|
|
let migration_version = match migration_value {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => {
|
|
return crate::PostgresBackendHealthSnapshot::not_ready(runtime, std::option::Option::None, 0, crate::PostgresBackendErrorKind::HealthFailed);
|
|
},
|
|
};
|
|
let expected = crate::current_migration_version();
|
|
if migration_version < 0 || migration_version < expected {
|
|
let observed = nonnegative_version(migration_version);
|
|
let pending = pending_migration_count(migration_version, expected);
|
|
return crate::PostgresBackendHealthSnapshot::not_ready(runtime, observed, pending, crate::PostgresBackendErrorKind::MigrationMismatch);
|
|
}
|
|
if migration_version > expected {
|
|
return crate::PostgresBackendHealthSnapshot::not_ready(
|
|
runtime,
|
|
nonnegative_version(migration_version),
|
|
0,
|
|
crate::PostgresBackendErrorKind::SchemaNewer,
|
|
);
|
|
}
|
|
return crate::PostgresBackendHealthSnapshot::ready(runtime, migration_version as u64, 0);
|
|
}
|
|
|
|
fn nonnegative_version(value: i64) -> std::option::Option<u64> {
|
|
if value < 0 {
|
|
return std::option::Option::None;
|
|
}
|
|
return std::option::Option::Some(value as u64);
|
|
}
|
|
|
|
fn pending_migration_count(observed: i64, expected: i64) -> u32 {
|
|
if observed >= expected {
|
|
return 0;
|
|
}
|
|
let delta = expected.saturating_sub(observed);
|
|
if delta > i64::from(u32::MAX) {
|
|
return u32::MAX;
|
|
}
|
|
return delta as u32;
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "../unit_tests/health.rs"]
|
|
mod tests;
|