// 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::(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::(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 { 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;