status.rs (14777B)
1 //! Storage capability, health, and integrity status contracts. 2 3 use radroots_transport::BoxFuture; 4 5 use crate::{Error, event::SourceGeneration}; 6 7 /// Passive backend-level status capability independent of backup workflows. 8 pub trait StorageStatusProvider: Send + Sync { 9 fn storage_status(&self) -> BoxFuture<'_, Result<StorageStatus, Error>>; 10 } 11 12 /// Storage-engine family needed to interpret durability-specific status. 13 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 14 #[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))] 15 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 16 pub enum StorageBackend { 17 Memory, 18 Sqlite, 19 } 20 21 impl StorageBackend { 22 pub const fn as_str(self) -> &'static str { 23 match self { 24 Self::Memory => "memory", 25 Self::Sqlite => "sqlite", 26 } 27 } 28 } 29 30 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 31 #[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))] 32 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 33 pub enum StorageOpenMode { 34 ReadOnly, 35 ReadWriteExisting, 36 Create, 37 } 38 39 impl StorageOpenMode { 40 pub const fn as_str(self) -> &'static str { 41 match self { 42 Self::ReadOnly => "read_only", 43 Self::ReadWriteExisting => "read_write_existing", 44 Self::Create => "create", 45 } 46 } 47 } 48 49 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 50 #[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))] 51 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 52 pub enum WriterPolicy { 53 NoWriter, 54 AdvisoryProcessLock, 55 } 56 57 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 58 #[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))] 59 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 60 pub enum ShutdownState { 61 Open, 62 Closing, 63 Closed, 64 } 65 66 impl ShutdownState { 67 pub const fn as_str(self) -> &'static str { 68 match self { 69 Self::Open => "open", 70 Self::Closing => "closing", 71 Self::Closed => "closed", 72 } 73 } 74 } 75 76 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 77 #[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))] 78 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 79 pub enum IntegrityHealth { 80 Healthy, 81 Degraded, 82 Corrupt, 83 Unknown, 84 } 85 86 impl IntegrityHealth { 87 pub const fn as_str(self) -> &'static str { 88 match self { 89 Self::Healthy => "healthy", 90 Self::Degraded => "degraded", 91 Self::Corrupt => "corrupt", 92 Self::Unknown => "unknown", 93 } 94 } 95 } 96 97 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 98 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 99 pub struct IntegrityStatus { 100 health: IntegrityHealth, 101 checked_at_unix_ms: Option<u64>, 102 verified_members: u32, 103 failed_members: u32, 104 } 105 106 impl IntegrityStatus { 107 pub fn new( 108 health: IntegrityHealth, 109 checked_at_unix_ms: Option<u64>, 110 verified_members: u32, 111 failed_members: u32, 112 ) -> Result<Self, Error> { 113 if matches!(checked_at_unix_ms, Some(0)) 114 || (health == IntegrityHealth::Healthy && failed_members != 0) 115 || (health == IntegrityHealth::Corrupt && failed_members == 0) 116 || (health == IntegrityHealth::Unknown && checked_at_unix_ms.is_some()) 117 { 118 return Err(Error::InvalidIntegrityStatus); 119 } 120 Ok(Self { 121 health, 122 checked_at_unix_ms, 123 verified_members, 124 failed_members, 125 }) 126 } 127 pub const fn health(self) -> IntegrityHealth { 128 self.health 129 } 130 pub const fn checked_at_unix_ms(self) -> Option<u64> { 131 self.checked_at_unix_ms 132 } 133 pub const fn verified_members(self) -> u32 { 134 self.verified_members 135 } 136 pub const fn failed_members(self) -> u32 { 137 self.failed_members 138 } 139 } 140 141 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 142 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 143 pub struct StorageStatus { 144 backend: StorageBackend, 145 open_mode: StorageOpenMode, 146 writer_policy: WriterPolicy, 147 shutdown: ShutdownState, 148 integrity: IntegrityStatus, 149 wal_enabled: bool, 150 busy_timeout_ms: u32, 151 } 152 153 impl StorageStatus { 154 pub fn new( 155 backend: StorageBackend, 156 open_mode: StorageOpenMode, 157 writer_policy: WriterPolicy, 158 shutdown: ShutdownState, 159 integrity: IntegrityStatus, 160 wal_enabled: bool, 161 busy_timeout_ms: u32, 162 ) -> Result<Self, Error> { 163 let valid_engine_status = match backend { 164 StorageBackend::Memory => { 165 writer_policy == WriterPolicy::NoWriter && !wal_enabled && busy_timeout_ms == 0 166 } 167 StorageBackend::Sqlite => { 168 (open_mode == StorageOpenMode::ReadOnly && writer_policy == WriterPolicy::NoWriter) 169 || (open_mode != StorageOpenMode::ReadOnly 170 && writer_policy == WriterPolicy::AdvisoryProcessLock 171 && wal_enabled 172 && busy_timeout_ms != 0) 173 } 174 }; 175 if !valid_engine_status { 176 return Err(Error::InvalidStorageStatus); 177 } 178 Ok(Self { 179 backend, 180 open_mode, 181 writer_policy, 182 shutdown, 183 integrity, 184 wal_enabled, 185 busy_timeout_ms, 186 }) 187 } 188 pub const fn backend(self) -> StorageBackend { 189 self.backend 190 } 191 pub const fn open_mode(self) -> StorageOpenMode { 192 self.open_mode 193 } 194 pub const fn writer_policy(self) -> WriterPolicy { 195 self.writer_policy 196 } 197 pub const fn shutdown(self) -> ShutdownState { 198 self.shutdown 199 } 200 pub const fn integrity(self) -> IntegrityStatus { 201 self.integrity 202 } 203 pub const fn wal_enabled(self) -> bool { 204 self.wal_enabled 205 } 206 pub const fn busy_timeout_ms(self) -> u32 { 207 self.busy_timeout_ms 208 } 209 } 210 211 /// Current event-store operating mode. 212 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 213 #[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))] 214 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 215 pub enum EventStoreMode { 216 ReadOnly, 217 ReadWrite, 218 } 219 220 /// Current health of the canonical event source. 221 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 222 #[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))] 223 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 224 pub enum EventStoreHealth { 225 Available, 226 Degraded, 227 Unavailable, 228 } 229 230 /// Passive event-store capability and cardinality report. 231 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] 232 #[derive(Clone, Debug, Eq, PartialEq)] 233 pub struct EventStoreStatus { 234 generation: SourceGeneration, 235 mode: EventStoreMode, 236 health: EventStoreHealth, 237 raw_events: u64, 238 verified_events: u64, 239 visible_events: u64, 240 } 241 242 impl EventStoreStatus { 243 /// Creates a consistent event-store status report. 244 pub const fn new( 245 generation: SourceGeneration, 246 mode: EventStoreMode, 247 health: EventStoreHealth, 248 raw_events: u64, 249 verified_events: u64, 250 visible_events: u64, 251 ) -> Result<Self, Error> { 252 if verified_events > raw_events || visible_events > verified_events { 253 return Err(Error::CorruptStoredEvent); 254 } 255 Ok(Self { 256 generation, 257 mode, 258 health, 259 raw_events, 260 verified_events, 261 visible_events, 262 }) 263 } 264 265 pub const fn generation(&self) -> SourceGeneration { 266 self.generation 267 } 268 269 pub const fn mode(&self) -> EventStoreMode { 270 self.mode 271 } 272 273 pub const fn health(&self) -> EventStoreHealth { 274 self.health 275 } 276 277 pub const fn raw_events(&self) -> u64 { 278 self.raw_events 279 } 280 281 pub const fn verified_events(&self) -> u64 { 282 self.verified_events 283 } 284 285 pub const fn visible_events(&self) -> u64 { 286 self.visible_events 287 } 288 } 289 290 #[cfg(test)] 291 mod tests { 292 use super::*; 293 294 #[test] 295 fn storage_status_labels_are_explicit_and_stable() { 296 assert_eq!(StorageBackend::Memory.as_str(), "memory"); 297 assert_eq!(StorageBackend::Sqlite.as_str(), "sqlite"); 298 assert_eq!(StorageOpenMode::ReadOnly.as_str(), "read_only"); 299 assert_eq!( 300 StorageOpenMode::ReadWriteExisting.as_str(), 301 "read_write_existing" 302 ); 303 assert_eq!(StorageOpenMode::Create.as_str(), "create"); 304 assert_eq!(ShutdownState::Open.as_str(), "open"); 305 assert_eq!(ShutdownState::Closing.as_str(), "closing"); 306 assert_eq!(ShutdownState::Closed.as_str(), "closed"); 307 assert_eq!(IntegrityHealth::Healthy.as_str(), "healthy"); 308 assert_eq!(IntegrityHealth::Degraded.as_str(), "degraded"); 309 assert_eq!(IntegrityHealth::Corrupt.as_str(), "corrupt"); 310 assert_eq!(IntegrityHealth::Unknown.as_str(), "unknown"); 311 } 312 313 fn integrity() -> IntegrityStatus { 314 IntegrityStatus::new(IntegrityHealth::Healthy, Some(1), 3, 0).unwrap() 315 } 316 317 #[test] 318 fn integrity_status_covers_every_invariant_and_accessor() { 319 assert_eq!( 320 IntegrityStatus::new(IntegrityHealth::Healthy, Some(0), 0, 0), 321 Err(Error::InvalidIntegrityStatus) 322 ); 323 assert_eq!( 324 IntegrityStatus::new(IntegrityHealth::Healthy, Some(1), 0, 1), 325 Err(Error::InvalidIntegrityStatus) 326 ); 327 assert_eq!( 328 IntegrityStatus::new(IntegrityHealth::Corrupt, Some(1), 1, 0), 329 Err(Error::InvalidIntegrityStatus) 330 ); 331 assert_eq!( 332 IntegrityStatus::new(IntegrityHealth::Unknown, Some(1), 0, 0), 333 Err(Error::InvalidIntegrityStatus) 334 ); 335 336 let status = integrity(); 337 assert_eq!(status.health(), IntegrityHealth::Healthy); 338 assert_eq!(status.checked_at_unix_ms(), Some(1)); 339 assert_eq!(status.verified_members(), 3); 340 assert_eq!(status.failed_members(), 0); 341 assert!(IntegrityStatus::new(IntegrityHealth::Degraded, None, 0, 1).is_ok()); 342 assert!(IntegrityStatus::new(IntegrityHealth::Corrupt, None, 0, 1).is_ok()); 343 assert!(IntegrityStatus::new(IntegrityHealth::Unknown, None, 0, 0).is_ok()); 344 } 345 346 #[test] 347 fn storage_status_covers_memory_and_sqlite_policy_matrix() { 348 let memory = StorageStatus::new( 349 StorageBackend::Memory, 350 StorageOpenMode::Create, 351 WriterPolicy::NoWriter, 352 ShutdownState::Open, 353 integrity(), 354 false, 355 0, 356 ) 357 .unwrap(); 358 assert_eq!(memory.backend(), StorageBackend::Memory); 359 assert_eq!(memory.open_mode(), StorageOpenMode::Create); 360 assert_eq!(memory.writer_policy(), WriterPolicy::NoWriter); 361 assert_eq!(memory.shutdown(), ShutdownState::Open); 362 assert_eq!(memory.integrity(), integrity()); 363 assert!(!memory.wal_enabled()); 364 assert_eq!(memory.busy_timeout_ms(), 0); 365 366 for (writer, wal, timeout) in [ 367 (WriterPolicy::AdvisoryProcessLock, false, 0), 368 (WriterPolicy::NoWriter, true, 0), 369 (WriterPolicy::NoWriter, false, 1), 370 ] { 371 assert_eq!( 372 StorageStatus::new( 373 StorageBackend::Memory, 374 StorageOpenMode::ReadOnly, 375 writer, 376 ShutdownState::Closed, 377 integrity(), 378 wal, 379 timeout, 380 ), 381 Err(Error::InvalidStorageStatus) 382 ); 383 } 384 385 assert!( 386 StorageStatus::new( 387 StorageBackend::Sqlite, 388 StorageOpenMode::ReadOnly, 389 WriterPolicy::NoWriter, 390 ShutdownState::Closing, 391 integrity(), 392 false, 393 0, 394 ) 395 .is_ok() 396 ); 397 assert!( 398 StorageStatus::new( 399 StorageBackend::Sqlite, 400 StorageOpenMode::ReadWriteExisting, 401 WriterPolicy::AdvisoryProcessLock, 402 ShutdownState::Open, 403 integrity(), 404 true, 405 1, 406 ) 407 .is_ok() 408 ); 409 for (mode, writer, wal, timeout) in [ 410 ( 411 StorageOpenMode::ReadOnly, 412 WriterPolicy::AdvisoryProcessLock, 413 false, 414 0, 415 ), 416 (StorageOpenMode::Create, WriterPolicy::NoWriter, true, 1), 417 ( 418 StorageOpenMode::Create, 419 WriterPolicy::AdvisoryProcessLock, 420 false, 421 1, 422 ), 423 ( 424 StorageOpenMode::Create, 425 WriterPolicy::AdvisoryProcessLock, 426 true, 427 0, 428 ), 429 ] { 430 assert_eq!( 431 StorageStatus::new( 432 StorageBackend::Sqlite, 433 mode, 434 writer, 435 ShutdownState::Open, 436 integrity(), 437 wal, 438 timeout, 439 ), 440 Err(Error::InvalidStorageStatus) 441 ); 442 } 443 } 444 445 #[test] 446 fn event_store_status_covers_bounds_and_accessors() { 447 let generation = SourceGeneration::new([1; 32]).unwrap(); 448 assert_eq!( 449 EventStoreStatus::new( 450 generation, 451 EventStoreMode::ReadOnly, 452 EventStoreHealth::Unavailable, 453 1, 454 2, 455 0, 456 ), 457 Err(Error::CorruptStoredEvent) 458 ); 459 assert_eq!( 460 EventStoreStatus::new( 461 generation, 462 EventStoreMode::ReadWrite, 463 EventStoreHealth::Degraded, 464 2, 465 1, 466 2, 467 ), 468 Err(Error::CorruptStoredEvent) 469 ); 470 let status = EventStoreStatus::new( 471 generation, 472 EventStoreMode::ReadWrite, 473 EventStoreHealth::Available, 474 3, 475 2, 476 1, 477 ) 478 .unwrap(); 479 assert_eq!(status.generation(), generation); 480 assert_eq!(status.mode(), EventStoreMode::ReadWrite); 481 assert_eq!(status.health(), EventStoreHealth::Available); 482 assert_eq!(status.raw_events(), 3); 483 assert_eq!(status.verified_events(), 2); 484 assert_eq!(status.visible_events(), 1); 485 } 486 }