integrity.rs (8811B)
1 //! Explicit SQLite integrity validation and passive result reporting. 2 3 use radroots_storage::{ 4 Error, 5 status::{IntegrityHealth, IntegrityStatus}, 6 }; 7 use sqlx::SqlitePool; 8 9 use crate::SqliteStorage; 10 11 impl SqliteStorage { 12 /// Returns the last recorded integrity result without running maintenance, 13 /// querying SQLite pragmas, or mutating either owned database. 14 pub async fn integrity(&self) -> Result<IntegrityStatus, Error> { 15 self.lifecycle.integrity() 16 } 17 18 /// Explicitly validates physical and referential integrity for both owned 19 /// databases and records the result under a caller-supplied timestamp. 20 pub async fn check_integrity(&self, checked_at_unix_ms: u64) -> Result<IntegrityStatus, Error> { 21 if checked_at_unix_ms == 0 { 22 return Err(Error::InvalidIntegrityStatus); 23 } 24 self.lifecycle.require_open()?; 25 26 let outcomes = [ 27 check_member(&self.pool).await, 28 check_member(&self.private_pool).await, 29 ]; 30 let verified_members = outcomes 31 .iter() 32 .filter(|outcome| **outcome == MemberOutcome::Verified) 33 .count(); 34 let failed_members = outcomes.len().saturating_sub(verified_members); 35 let health = if outcomes.contains(&MemberOutcome::Corrupt) { 36 IntegrityHealth::Corrupt 37 } else if outcomes.contains(&MemberOutcome::Unavailable) { 38 IntegrityHealth::Degraded 39 } else { 40 IntegrityHealth::Healthy 41 }; 42 let status = IntegrityStatus::new( 43 health, 44 Some(checked_at_unix_ms), 45 u32::try_from(verified_members).map_err(|_| Error::InvalidIntegrityStatus)?, 46 u32::try_from(failed_members).map_err(|_| Error::InvalidIntegrityStatus)?, 47 )?; 48 self.lifecycle.record_integrity(status) 49 } 50 } 51 52 pub(crate) fn unknown() -> Result<IntegrityStatus, Error> { 53 IntegrityStatus::new(IntegrityHealth::Unknown, None, 0, 0) 54 } 55 56 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 57 pub(crate) enum MemberOutcome { 58 Verified, 59 Corrupt, 60 Unavailable, 61 } 62 63 async fn check_member(pool: &SqlitePool) -> MemberOutcome { 64 let mut connection = match pool.acquire().await { 65 Ok(connection) => connection, 66 Err(_) => return MemberOutcome::Unavailable, 67 }; 68 check_connection(&mut connection).await 69 } 70 71 pub(crate) async fn check_connection(connection: &mut sqlx::SqliteConnection) -> MemberOutcome { 72 let integrity = match sqlx::query_scalar::<_, String>("PRAGMA integrity_check") 73 .fetch_all(&mut *connection) 74 .await 75 { 76 Ok(rows) => rows, 77 Err(_) => return MemberOutcome::Unavailable, 78 }; 79 if integrity.as_slice() != ["ok"] { 80 return MemberOutcome::Corrupt; 81 } 82 match sqlx::query("PRAGMA foreign_key_check") 83 .fetch_all(&mut *connection) 84 .await 85 { 86 Ok(rows) if rows.is_empty() => MemberOutcome::Verified, 87 Ok(_) => MemberOutcome::Corrupt, 88 Err(_) => MemberOutcome::Unavailable, 89 } 90 } 91 92 #[cfg(test)] 93 #[cfg_attr(coverage_nightly, coverage(off))] 94 mod policy_tests { 95 use serde::Deserialize; 96 97 const POLICY: &str = include_str!("../../../contracts/storage/integrity_policy_v1.toml"); 98 99 #[derive(Deserialize)] 100 struct Policy { 101 schema_version: u32, 102 members: Vec<String>, 103 invocation: String, 104 checks: Vec<String>, 105 timestamp: String, 106 initial_passive_health: String, 107 passive_status_runs_checks: bool, 108 recording: String, 109 closed_backend: String, 110 healthy: String, 111 corrupt: String, 112 degraded: String, 113 } 114 115 #[test] 116 fn implementation_matches_the_governed_integrity_policy() { 117 let policy = toml::from_str::<Policy>(POLICY).expect("integrity policy"); 118 assert_eq!(policy.schema_version, 1); 119 assert_eq!(policy.members, ["runtime.sqlite", "private.sqlite"]); 120 assert_eq!(policy.invocation, "explicit_only"); 121 assert_eq!( 122 policy.checks, 123 ["pragma_integrity_check", "pragma_foreign_key_check"] 124 ); 125 assert_eq!(policy.timestamp, "caller_supplied_positive_unix_ms"); 126 assert_eq!(policy.initial_passive_health, "unknown"); 127 assert!(!policy.passive_status_runs_checks); 128 assert_eq!(policy.recording, "latest_monotonic_checked_at"); 129 assert_eq!(policy.closed_backend, "reject"); 130 assert_eq!(policy.healthy, "all_members_verified"); 131 assert_eq!( 132 policy.corrupt, 133 "one_or_more_members_failed_a_completed_check" 134 ); 135 assert_eq!( 136 policy.degraded, 137 "one_or_more_members_could_not_complete_checks" 138 ); 139 } 140 } 141 142 #[cfg(test)] 143 #[cfg_attr(coverage_nightly, coverage(off))] 144 mod tests { 145 use radroots_storage::{ 146 Error, 147 event::SourceGeneration, 148 status::{EventStoreMode, IntegrityHealth}, 149 }; 150 use sqlx::sqlite::SqlitePoolOptions; 151 152 use crate::{OpenMode, OpenOptions, Paths, SqliteStorage}; 153 154 fn generation(byte: u8) -> SourceGeneration { 155 SourceGeneration::new([byte; 32]).expect("source generation") 156 } 157 158 async fn create(directory: &std::path::Path) -> SqliteStorage { 159 let paths = Paths::from_directory(directory).expect("owned paths"); 160 SqliteStorage::open( 161 OpenOptions::new(paths, OpenMode::Create) 162 .with_source_generation(generation(83), 8_300) 163 .expect("source generation"), 164 ) 165 .await 166 .expect("create storage") 167 } 168 169 #[tokio::test] 170 async fn explicit_checks_record_healthy_and_corrupt_results_monotonically() { 171 let directory = tempfile::tempdir().expect("temporary directory"); 172 let store = create(directory.path()).await; 173 174 assert_eq!( 175 store.check_integrity(0).await, 176 Err(Error::InvalidIntegrityStatus) 177 ); 178 let healthy = store.check_integrity(100).await.expect("healthy check"); 179 assert_eq!(healthy.health(), IntegrityHealth::Healthy); 180 assert_eq!(healthy.checked_at_unix_ms(), Some(100)); 181 assert_eq!(healthy.verified_members(), 2); 182 assert_eq!(healthy.failed_members(), 0); 183 assert_eq!(store.integrity().await.expect("passive result"), healthy); 184 185 let mut private = store 186 .private_pool 187 .acquire() 188 .await 189 .expect("private connection"); 190 sqlx::raw_sql( 191 "PRAGMA foreign_keys = OFF; 192 CREATE TABLE integrity_parent (id INTEGER PRIMARY KEY); 193 CREATE TABLE integrity_child ( 194 parent_id INTEGER NOT NULL REFERENCES integrity_parent(id) 195 ); 196 INSERT INTO integrity_child (parent_id) VALUES (1);", 197 ) 198 .execute(&mut *private) 199 .await 200 .expect("inject referential corruption"); 201 drop(private); 202 203 let corrupt = store.check_integrity(200).await.expect("corrupt check"); 204 assert_eq!(corrupt.health(), IntegrityHealth::Corrupt); 205 assert_eq!(corrupt.checked_at_unix_ms(), Some(200)); 206 assert_eq!(corrupt.verified_members(), 1); 207 assert_eq!(corrupt.failed_members(), 1); 208 assert_eq!( 209 store 210 .storage_status() 211 .await 212 .expect("passive storage status") 213 .integrity(), 214 corrupt 215 ); 216 assert_eq!( 217 store.check_integrity(199).await, 218 Err(Error::InvalidIntegrityStatus) 219 ); 220 assert_eq!(store.integrity().await.expect("latest result"), corrupt); 221 222 store.close().await.expect("close storage"); 223 assert_eq!( 224 store.check_integrity(300).await, 225 Err(Error::BackendUnavailable) 226 ); 227 } 228 229 #[tokio::test] 230 async fn incomplete_member_checks_record_degraded_status() { 231 let runtime = SqlitePoolOptions::new() 232 .max_connections(1) 233 .connect("sqlite::memory:") 234 .await 235 .expect("runtime pool"); 236 let private = SqlitePoolOptions::new() 237 .max_connections(1) 238 .connect("sqlite::memory:") 239 .await 240 .expect("private pool"); 241 let store = SqliteStorage::with_private_pool( 242 runtime, 243 private.clone(), 244 generation(84), 245 EventStoreMode::ReadWrite, 246 ); 247 private.close().await; 248 249 let degraded = store.check_integrity(400).await.expect("degraded check"); 250 assert_eq!(degraded.health(), IntegrityHealth::Degraded); 251 assert_eq!(degraded.verified_members(), 1); 252 assert_eq!(degraded.failed_members(), 1); 253 assert_eq!(store.integrity().await.expect("recorded result"), degraded); 254 } 255 }