1use std::path::{Path, PathBuf};
61use std::sync::atomic::{AtomicBool, Ordering};
62use std::sync::Arc;
63use std::time::Duration;
64
65use khive_storage::error::StorageError;
66use khive_storage::types::StorageResult;
67use khive_storage::StorageCapability;
68use rusqlite::Connection;
69use serde::Serialize;
70
71use crate::checkpoint;
72use crate::pool::ConnectionPool;
73
74#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
81pub struct CheckpointProbe {
82 pub busy: i64,
83 pub log_frames: i64,
84 pub checkpointed_frames: i64,
85}
86
87impl CheckpointProbe {
88 pub fn pin_depth(&self) -> i64 {
91 (self.log_frames - self.checkpointed_frames).max(0)
92 }
93}
94
95pub fn checkpoint_probe(conn: &Connection) -> rusqlite::Result<CheckpointProbe> {
109 conn.query_row("PRAGMA wal_checkpoint(PASSIVE)", [], |row| {
110 Ok(CheckpointProbe {
111 busy: row.get(0)?,
112 log_frames: row.get(1)?,
113 checkpointed_frames: row.get(2)?,
114 })
115 })
116}
117
118#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
124pub struct CheckpointCounters {
125 pub last_observed_wal_pages: Option<u64>,
126 pub truncate_attempts: u64,
127 pub truncate_consecutive_failures: u64,
128 pub checkpoint_skipped_ticks: u64,
129 pub checkpoint_consecutive_skips: u64,
130 pub checkpoint_last_skip_wal_pages: Option<u64>,
131 pub checkpoint_pressure_elevated_ticks: u64,
132 pub checkpoint_pressure_episodes_started: u64,
133 pub checkpoint_pressure_episodes_recovered: u64,
134 pub checkpoint_lifecycle_append_attempts: u64,
135 pub checkpoint_lifecycle_append_failures: u64,
136 pub checkpoint_lifecycle_enqueue_drops: u64,
137 pub read_tx_max_age_evictions: u64,
141}
142
143pub fn checkpoint_counters() -> CheckpointCounters {
145 CheckpointCounters {
146 last_observed_wal_pages: checkpoint::last_observed_wal_pages(),
147 truncate_attempts: checkpoint::truncate_attempts(),
148 truncate_consecutive_failures: checkpoint::truncate_consecutive_failures(),
149 checkpoint_skipped_ticks: checkpoint::checkpoint_skipped_ticks(),
150 checkpoint_consecutive_skips: checkpoint::checkpoint_consecutive_skips(),
151 checkpoint_last_skip_wal_pages: checkpoint::checkpoint_last_skip_wal_pages(),
152 checkpoint_pressure_elevated_ticks: checkpoint::checkpoint_pressure_elevated_ticks(),
153 checkpoint_pressure_episodes_started: checkpoint::checkpoint_pressure_episodes_started(),
154 checkpoint_pressure_episodes_recovered: checkpoint::checkpoint_pressure_episodes_recovered(
155 ),
156 checkpoint_lifecycle_append_attempts: checkpoint::checkpoint_lifecycle_append_attempts(),
157 checkpoint_lifecycle_append_failures: checkpoint::checkpoint_lifecycle_append_failures(),
158 checkpoint_lifecycle_enqueue_drops: checkpoint::checkpoint_lifecycle_enqueue_drops(),
159 read_tx_max_age_evictions: checkpoint::read_tx_max_age_evictions(),
160 }
161}
162
163#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
170pub struct BuildIdentity {
171 pub version: String,
172 pub build_hash: Option<String>,
173}
174
175impl BuildIdentity {
176 pub fn from_env(version: &str, build_hash: Option<&str>) -> Self {
178 Self {
179 version: version.to_string(),
180 build_hash: build_hash.map(str::to_string),
181 }
182 }
183}
184
185#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
187pub struct WalFileState {
188 pub wal_path: String,
189 pub wal_size_bytes: Option<u64>,
192 pub unavailable_reason: Option<String>,
193}
194
195pub fn wal_file_state(db_path: &Path) -> WalFileState {
197 let wal_path = wal_sidecar_path(db_path);
198 match std::fs::metadata(&wal_path) {
199 Ok(md) => WalFileState {
200 wal_path: wal_path.display().to_string(),
201 wal_size_bytes: Some(md.len()),
202 unavailable_reason: None,
203 },
204 Err(e) => WalFileState {
205 wal_path: wal_path.display().to_string(),
206 wal_size_bytes: None,
207 unavailable_reason: Some(e.to_string()),
208 },
209 }
210}
211
212fn wal_sidecar_path(db_path: &Path) -> PathBuf {
215 let mut s = db_path.as_os_str().to_os_string();
216 s.push("-wal");
217 PathBuf::from(s)
218}
219
220#[derive(Debug, Clone, PartialEq, Serialize)]
229pub struct WalPinAttribution {
230 pub status: WalPinAttributionStatus,
232 pub status_reasons: Vec<String>,
234 pub census: WalPinCensus,
236 pub available: bool,
239 pub unavailable_reason: Option<String>,
240 #[serde(skip_serializing)]
242 pub census_holder_pids: Vec<u32>,
243 #[serde(skip_serializing)]
244 pub census_uninspectable_pids: Vec<u32>,
245 #[serde(skip_serializing)]
246 pub census_truncated: bool,
247 #[serde(skip_serializing)]
248 pub census_is_complete: bool,
249 pub reporting: Vec<WalPinHolder>,
251 pub registered_silent_pids: Vec<u32>,
253 pub unknown_pids: Vec<u32>,
255 pub census_pids_without_attribution: Vec<u32>,
257 pub fully_attributed: bool,
260 pub sidecar_entries: Vec<serde_json::Value>,
262 #[serde(skip_serializing_if = "Option::is_none")]
264 pub sidecar_listing_truncated: Option<bool>,
265 #[serde(skip_serializing_if = "Option::is_none")]
267 pub sidecar_entries_cleanup_would_reap: Option<usize>,
268}
269
270#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
276#[serde(rename_all = "snake_case")]
277pub enum WalPinAttributionStatus {
278 Complete,
280 Degraded,
282 Unavailable,
284}
285
286#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
294#[serde(tag = "status", rename_all = "snake_case")]
295pub enum WalPinCensus {
296 Complete {
298 holder_pids: Vec<u32>,
300 },
301 Incomplete {
303 holder_pids: Vec<u32>,
305 uninspectable_pids: Vec<u32>,
307 truncated: bool,
309 reason: String,
311 },
312 Unavailable {
314 reason: String,
316 },
317}
318
319#[derive(Debug, Clone, PartialEq, Serialize)]
323pub struct WalPinHolder {
324 pub pid: u32,
325 pub process_role: String,
326 pub current_oldest_tx_age_secs: f64,
327 pub oldest_tx_label: Option<String>,
328 pub attribution_is_evidence_backed: bool,
329}
330
331impl WalPinAttribution {
332 fn unavailable(reason: impl Into<String>) -> Self {
333 let reason = reason.into();
334 Self {
335 status: WalPinAttributionStatus::Unavailable,
336 status_reasons: vec![reason.clone()],
337 census: WalPinCensus::Unavailable {
338 reason: reason.clone(),
339 },
340 available: false,
341 unavailable_reason: Some(reason),
342 census_holder_pids: Vec::new(),
343 census_uninspectable_pids: Vec::new(),
344 census_truncated: false,
345 census_is_complete: false,
346 reporting: Vec::new(),
347 registered_silent_pids: Vec::new(),
348 unknown_pids: Vec::new(),
349 census_pids_without_attribution: Vec::new(),
350 fully_attributed: false,
351 sidecar_entries: Vec::new(),
352 sidecar_listing_truncated: None,
353 sidecar_entries_cleanup_would_reap: None,
354 }
355 }
356}
357
358#[cfg(unix)]
359fn wal_pin_attribution_from_census(census: crate::walpin::CensusResult) -> WalPinAttribution {
360 const SIDECAR_REASON: &str = "sidecar-to-holder reconciliation not available: this tree's \
361 khive-db has no non-destructive sidecar reconciliation primitive; walpin::enumerate_live \
362 and walpin::housekeep_live both perform mutating cleanup, so diagnostics called neither \
363 and collected only the read-only OS holder census below";
364
365 let census_is_complete = census.is_complete();
366 let mut census_holder_pids: Vec<u32> = census.holders.iter().copied().collect();
367 census_holder_pids.sort_unstable();
368 let mut census_uninspectable_pids = census.uninspectable_pids;
369 census_uninspectable_pids.sort_unstable();
370 census_uninspectable_pids.dedup();
371 let census_truncated = census.truncated;
372
373 let mut status_reasons = vec![SIDECAR_REASON.to_string()];
374 let census = if census_is_complete {
375 WalPinCensus::Complete {
376 holder_pids: census_holder_pids.clone(),
377 }
378 } else {
379 let mut causes = Vec::new();
380 if census_truncated {
381 causes.push("the OS process walk was truncated".to_string());
382 }
383 if !census_uninspectable_pids.is_empty() {
384 causes.push(format!(
385 "{} PID(s) could not be inspected",
386 census_uninspectable_pids.len()
387 ));
388 }
389 let reason = format!(
390 "OS holder census is incomplete: {}; additional database holders cannot be ruled out",
391 causes.join("; ")
392 );
393 status_reasons.push(reason.clone());
394 WalPinCensus::Incomplete {
395 holder_pids: census_holder_pids.clone(),
396 uninspectable_pids: census_uninspectable_pids.clone(),
397 truncated: census_truncated,
398 reason,
399 }
400 };
401
402 WalPinAttribution {
403 status: WalPinAttributionStatus::Degraded,
404 unavailable_reason: Some(status_reasons.join("; ")),
405 status_reasons,
406 census,
407 available: false,
408 census_holder_pids,
409 census_uninspectable_pids,
410 census_truncated,
411 census_is_complete,
412 reporting: Vec::new(),
413 registered_silent_pids: Vec::new(),
414 unknown_pids: Vec::new(),
415 census_pids_without_attribution: Vec::new(),
416 fully_attributed: false,
417 sidecar_entries: Vec::new(),
418 sidecar_listing_truncated: None,
419 sidecar_entries_cleanup_would_reap: None,
420 }
421}
422
423#[cfg(unix)]
428pub fn wal_pin_attribution(db_path: &Path, _sweep_interval: Duration) -> WalPinAttribution {
429 use crate::walpin;
430
431 let census = match walpin::census_holders(db_path) {
432 Ok(c) => c,
433 Err(e) => return WalPinAttribution::unavailable(format!("census_holders failed: {e}")),
434 };
435 wal_pin_attribution_from_census(census)
436}
437
438#[cfg(not(unix))]
439pub fn wal_pin_attribution(_db_path: &Path, _sweep_interval: Duration) -> WalPinAttribution {
440 WalPinAttribution::unavailable("WAL-pin attribution requires a Unix platform")
441}
442
443#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
457pub struct WriterContentionDiagnostics {
458 pub writer_acquisitions: u64,
461 pub pooled_writer_acquisitions: u64,
463 pub standalone_writer_acquisitions: u64,
465 pub writer_task_acquisitions: u64,
467 pub writer_acquisition_timeouts: u64,
469 pub writer_task_begin_busy: u64,
474 pub writer_task_begin_errors: u64,
477 pub writer_task_request_failures: u64,
482 pub writer_task_side_effects_unknown: u64,
485 pub audit_append_failures: Option<u64>,
497 pub audit_append_failures_unavailable_reason: Option<String>,
499 pub audit_batch_flush_failures: Option<u64>,
504 pub audit_batch_flush_failures_unavailable_reason: Option<String>,
506 pub audit_degraded_rows: Option<u64>,
509 pub audit_degraded_rows_unavailable_reason: Option<String>,
511 pub audit_degraded: Option<bool>,
515 pub audit_degraded_unavailable_reason: Option<String>,
517 pub audit_admission_refused_obligations: Option<u64>,
531 pub audit_admission_refused_obligations_unavailable_reason: Option<String>,
533 pub audit_admission_unresolved_obligations: Option<u64>,
545 pub audit_admission_unresolved_obligations_unavailable_reason: Option<String>,
548}
549
550#[derive(Debug, Clone, Copy, PartialEq, Eq)]
555pub struct RuntimeAuditBatchMetrics {
556 pub flush_failures: u64,
559 pub degraded_rows: u64,
561 pub degraded: bool,
563 pub admission_refused_obligations: u64,
568 pub admission_unresolved_obligations: u64,
575}
576
577impl WriterContentionDiagnostics {
578 fn snapshot(
579 pool: &ConnectionPool,
580 audit_append_failures: Option<u64>,
581 runtime_audit_batch_metrics: Option<RuntimeAuditBatchMetrics>,
582 ) -> Self {
583 let writer = pool.writer_acquisition_snapshot();
584 let unavailable_reason =
585 || Some("no audit-batch control is registered with this runtime instance".to_string());
586 Self {
587 writer_acquisitions: writer.acquisitions,
588 pooled_writer_acquisitions: writer.pooled_acquisitions,
589 standalone_writer_acquisitions: writer.standalone_acquisitions,
590 writer_task_acquisitions: writer.writer_task_acquisitions,
591 writer_acquisition_timeouts: writer.timeouts,
592 writer_task_begin_busy: writer.writer_task_begin_busy,
593 writer_task_begin_errors: writer.writer_task_begin_errors,
594 writer_task_request_failures: writer.writer_task_request_failures,
595 writer_task_side_effects_unknown: writer.writer_task_side_effects_unknown,
596 audit_append_failures,
597 audit_append_failures_unavailable_reason: audit_append_failures.is_none().then(|| {
598 "runtime audit instrumentation was not supplied to khive-db diagnostics".to_string()
599 }),
600 audit_batch_flush_failures: runtime_audit_batch_metrics.map(|m| m.flush_failures),
601 audit_batch_flush_failures_unavailable_reason: runtime_audit_batch_metrics
602 .is_none()
603 .then(unavailable_reason)
604 .flatten(),
605 audit_degraded_rows: runtime_audit_batch_metrics.map(|m| m.degraded_rows),
606 audit_degraded_rows_unavailable_reason: runtime_audit_batch_metrics
607 .is_none()
608 .then(unavailable_reason)
609 .flatten(),
610 audit_degraded: runtime_audit_batch_metrics.map(|m| m.degraded),
611 audit_degraded_unavailable_reason: runtime_audit_batch_metrics
612 .is_none()
613 .then(unavailable_reason)
614 .flatten(),
615 audit_admission_refused_obligations: runtime_audit_batch_metrics
616 .map(|m| m.admission_refused_obligations),
617 audit_admission_refused_obligations_unavailable_reason: runtime_audit_batch_metrics
618 .is_none()
619 .then(unavailable_reason)
620 .flatten(),
621 audit_admission_unresolved_obligations: runtime_audit_batch_metrics
622 .map(|m| m.admission_unresolved_obligations),
623 audit_admission_unresolved_obligations_unavailable_reason: runtime_audit_batch_metrics
624 .is_none()
625 .then(unavailable_reason)
626 .flatten(),
627 }
628 }
629}
630
631#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
639pub struct GraphEdgeIntegrity {
640 pub duplicate_edge_id_groups: i64,
641 pub graph_edges_rows: i64,
642 pub graph_edges_seq_rows: i64,
643 pub pre_v14_duplicate_edge_state_detected: bool,
644}
645
646fn graph_edge_integrity(conn: &Connection) -> rusqlite::Result<GraphEdgeIntegrity> {
647 conn.query_row(
648 "SELECT
649 (SELECT COUNT(*) FROM (
650 SELECT id FROM graph_edges GROUP BY id HAVING COUNT(*) > 1
651 )),
652 (SELECT COUNT(*) FROM graph_edges),
653 (SELECT COUNT(*) FROM graph_edges_seq)",
654 [],
655 |row| {
656 let duplicate_edge_id_groups = row.get(0)?;
657 Ok(GraphEdgeIntegrity {
658 duplicate_edge_id_groups,
659 graph_edges_rows: row.get(1)?,
660 graph_edges_seq_rows: row.get(2)?,
661 pre_v14_duplicate_edge_state_detected: duplicate_edge_id_groups > 0,
662 })
663 },
664 )
665}
666
667#[derive(Debug, Clone, PartialEq, Serialize)]
669pub struct DbDiagnostics {
670 pub build: BuildIdentity,
671 pub db_path: Option<String>,
674 pub wal_file: Option<WalFileState>,
675 pub checkpoint_counters: CheckpointCounters,
676 pub checkpoint_probe: Option<CheckpointProbe>,
677 pub checkpoint_probe_error: Option<String>,
678 pub writer_contention: WriterContentionDiagnostics,
680 pub graph_edge_integrity: Option<GraphEdgeIntegrity>,
681 pub graph_edge_integrity_error: Option<String>,
682 pub wal_pin: WalPinAttribution,
683}
684
685pub fn collect(
700 pool: &ConnectionPool,
701 build: BuildIdentity,
702 sweep_interval: Duration,
703) -> DbDiagnostics {
704 collect_inner(pool, build, sweep_interval, None, None)
705}
706
707pub fn collect_with_audit_append_failures(
710 pool: &ConnectionPool,
711 build: BuildIdentity,
712 sweep_interval: Duration,
713 audit_append_failures: u64,
714) -> DbDiagnostics {
715 collect_inner(
716 pool,
717 build,
718 sweep_interval,
719 Some(audit_append_failures),
720 None,
721 )
722}
723
724pub async fn collect_with_audit_append_failures_interruptibly(
734 pool: Arc<ConnectionPool>,
735 build: BuildIdentity,
736 sweep_interval: Duration,
737 audit_append_failures: u64,
738) -> StorageResult<DbDiagnostics> {
739 collect_with_runtime_audit_metrics_interruptibly(
740 pool,
741 build,
742 sweep_interval,
743 audit_append_failures,
744 None,
745 )
746 .await
747}
748
749pub async fn collect_with_runtime_audit_metrics_interruptibly(
756 pool: Arc<ConnectionPool>,
757 build: BuildIdentity,
758 sweep_interval: Duration,
759 audit_append_failures: u64,
760 runtime_audit_batch_metrics: Option<RuntimeAuditBatchMetrics>,
761) -> StorageResult<DbDiagnostics> {
762 crate::ensure_request_read_active("db_diagnostics")?;
763 let counters = checkpoint_counters();
764 let writer_contention = WriterContentionDiagnostics::snapshot(
765 &pool,
766 Some(audit_append_failures),
767 runtime_audit_batch_metrics,
768 );
769
770 let Some(path) = pool.config().path.clone() else {
771 crate::ensure_request_read_active("db_diagnostics")?;
772 return Ok(DbDiagnostics {
773 build,
774 db_path: None,
775 wal_file: None,
776 checkpoint_counters: counters,
777 checkpoint_probe: None,
778 checkpoint_probe_error: Some(
779 "in-memory database: no WAL file and no checkpoint to probe".to_string(),
780 ),
781 writer_contention,
782 graph_edge_integrity: None,
783 graph_edge_integrity_error: Some(
784 "in-memory database: no durable graph-edge ledger to inspect".to_string(),
785 ),
786 wal_pin: WalPinAttribution::unavailable(
787 "in-memory database: no file for the OS holder census",
788 ),
789 });
790 };
791
792 let inspection_pool = Arc::clone(&pool);
793 let inspection = crate::read_cancellation::run_interruptible_read(
794 StorageCapability::Sql,
795 "db_diagnostics.sqlite",
796 move |scope| inspect_pool_interruptibly(&inspection_pool, scope),
797 )
798 .await?;
799 crate::ensure_request_read_active("db_diagnostics")?;
800 let (wal_file, wal_pin) =
801 inspect_file_state_interruptibly(path.clone(), sweep_interval).await?;
802 crate::ensure_request_read_active("db_diagnostics")?;
803
804 Ok(DbDiagnostics {
805 build,
806 db_path: Some(path.display().to_string()),
807 wal_file: Some(wal_file),
808 checkpoint_counters: counters,
809 checkpoint_probe: inspection.checkpoint_probe,
810 checkpoint_probe_error: inspection.checkpoint_probe_error,
811 writer_contention,
812 graph_edge_integrity: inspection.graph_edge_integrity,
813 graph_edge_integrity_error: inspection.graph_edge_integrity_error,
814 wal_pin,
815 })
816}
817
818fn collect_inner(
819 pool: &ConnectionPool,
820 build: BuildIdentity,
821 sweep_interval: Duration,
822 audit_append_failures: Option<u64>,
823 runtime_audit_batch_metrics: Option<RuntimeAuditBatchMetrics>,
824) -> DbDiagnostics {
825 let counters = checkpoint_counters();
826 let writer_contention = WriterContentionDiagnostics::snapshot(
827 pool,
828 audit_append_failures,
829 runtime_audit_batch_metrics,
830 );
831
832 let Some(path) = pool.config().path.clone() else {
833 return DbDiagnostics {
834 build,
835 db_path: None,
836 wal_file: None,
837 checkpoint_counters: counters,
838 checkpoint_probe: None,
839 checkpoint_probe_error: Some(
840 "in-memory database: no WAL file and no checkpoint to probe".to_string(),
841 ),
842 writer_contention,
843 graph_edge_integrity: None,
844 graph_edge_integrity_error: Some(
845 "in-memory database: no durable graph-edge ledger to inspect".to_string(),
846 ),
847 wal_pin: WalPinAttribution::unavailable(
848 "in-memory database: no file for the OS holder census",
849 ),
850 };
851 };
852
853 let inspection = inspect_pool(pool);
854
855 DbDiagnostics {
856 build,
857 db_path: Some(path.display().to_string()),
858 wal_file: Some(wal_file_state(&path)),
859 checkpoint_counters: counters,
860 checkpoint_probe: inspection.checkpoint_probe,
861 checkpoint_probe_error: inspection.checkpoint_probe_error,
862 writer_contention,
863 graph_edge_integrity: inspection.graph_edge_integrity,
864 graph_edge_integrity_error: inspection.graph_edge_integrity_error,
865 wal_pin: wal_pin_attribution(&path, sweep_interval),
866 }
867}
868
869struct PoolInspection {
870 checkpoint_probe: Option<CheckpointProbe>,
871 checkpoint_probe_error: Option<String>,
872 graph_edge_integrity: Option<GraphEdgeIntegrity>,
873 graph_edge_integrity_error: Option<String>,
874}
875
876fn inspect_pool_interruptibly(
877 pool: &ConnectionPool,
878 scope: &crate::read_cancellation::InterruptibleReadScope,
879) -> StorageResult<PoolInspection> {
880 scope.ensure_active()?;
881 let conn = match pool.open_standalone_writer_untracked() {
882 Ok(conn) => conn,
883 Err(e) => {
884 scope.ensure_active()?;
885 let reason = format!("guarded standalone open refused: {e}");
886 return Ok(PoolInspection {
887 checkpoint_probe: None,
888 checkpoint_probe_error: Some(reason.clone()),
889 graph_edge_integrity: None,
890 graph_edge_integrity_error: Some(reason),
891 });
892 }
893 };
894 scope.ensure_active()?;
899
900 let (checkpoint_probe, checkpoint_probe_error) = match checkpoint_probe(&conn) {
902 Ok(probe) => (Some(probe), None),
903 Err(e) => (
904 None,
905 Some(format!("PRAGMA wal_checkpoint(PASSIVE) failed: {e}")),
906 ),
907 };
908 #[cfg(test)]
909 if TEST_PAUSE_AFTER_PASSIVE.load(Ordering::SeqCst) {
910 TEST_REACHED_AFTER_PASSIVE.store(true, Ordering::SeqCst);
911 while TEST_PAUSE_AFTER_PASSIVE.load(Ordering::SeqCst) && !scope.should_stop() {
912 std::thread::yield_now();
913 }
914 }
915 scope.ensure_active()?;
916
917 let integrity = scope.run(&conn, || Ok(graph_edge_integrity(&conn)))?;
920 let (graph_edge_integrity, graph_edge_integrity_error) = match integrity {
921 Ok(integrity) => (Some(integrity), None),
922 Err(e) => (
923 None,
924 Some(format!("graph-edge integrity query failed: {e}")),
925 ),
926 };
927
928 Ok(PoolInspection {
929 checkpoint_probe,
930 checkpoint_probe_error,
931 graph_edge_integrity,
932 graph_edge_integrity_error,
933 })
934}
935
936#[cfg(test)]
937static TEST_PAUSE_AFTER_PASSIVE: AtomicBool = AtomicBool::new(false);
938#[cfg(test)]
939static TEST_REACHED_AFTER_PASSIVE: AtomicBool = AtomicBool::new(false);
940
941struct StopCensusOnDrop {
942 stopped: Arc<AtomicBool>,
943 armed: bool,
944}
945
946impl Drop for StopCensusOnDrop {
947 fn drop(&mut self) {
948 if self.armed {
949 self.stopped.store(true, Ordering::SeqCst);
950 }
951 }
952}
953
954async fn inspect_file_state_interruptibly(
955 path: PathBuf,
956 _sweep_interval: Duration,
957) -> StorageResult<(WalFileState, WalPinAttribution)> {
958 const OPERATION: &str = "db_diagnostics.wal_holder_census";
959 crate::ensure_request_read_active(OPERATION)?;
960 let stopped = Arc::new(AtomicBool::new(false));
961 let worker_stopped = Arc::clone(&stopped);
962 let mut stop_on_drop = StopCensusOnDrop {
963 stopped: Arc::clone(&stopped),
964 armed: true,
965 };
966 let mut worker = tokio::task::spawn_blocking(move || {
967 let wal_file = wal_file_state(&path);
968 if worker_stopped.load(Ordering::SeqCst) {
969 return Err(std::io::Error::new(
970 std::io::ErrorKind::Interrupted,
971 "WAL holder census cancelled",
972 ));
973 }
974 #[cfg(unix)]
975 let attribution = match crate::walpin::census_holders_until(&path, || {
976 worker_stopped.load(Ordering::SeqCst)
977 }) {
978 Ok(census) => wal_pin_attribution_from_census(census),
979 Err(error) if error.kind() == std::io::ErrorKind::Interrupted => return Err(error),
980 Err(error) => WalPinAttribution::unavailable(format!("census_holders failed: {error}")),
981 };
982 #[cfg(not(unix))]
983 let attribution = wal_pin_attribution(&path, _sweep_interval);
984 Ok((wal_file, attribution))
985 });
986
987 tokio::select! {
988 joined = &mut worker => {
989 stop_on_drop.armed = false;
990 let result = joined
991 .map_err(|error| StorageError::driver(StorageCapability::Sql, OPERATION, error))?
992 .map_err(|error| StorageError::driver(StorageCapability::Sql, OPERATION, error))?;
993 crate::ensure_request_read_active(OPERATION)?;
994 Ok(result)
995 }
996 _ = crate::wait_for_request_read_cancellation() => {
997 stopped.store(true, Ordering::SeqCst);
998 if tokio::time::timeout(crate::sqlite_interrupt_grace_from_env(), &mut worker)
999 .await
1000 .is_err()
1001 {
1002 worker.abort();
1003 }
1004 stop_on_drop.armed = false;
1005 Err(StorageError::Timeout { operation: OPERATION.into() })
1006 }
1007 }
1008}
1009
1010fn inspect_pool(pool: &ConnectionPool) -> PoolInspection {
1017 let conn = match pool.open_standalone_writer_untracked() {
1018 Ok(conn) => conn,
1019 Err(e) => {
1020 let reason = format!("guarded standalone open refused: {e}");
1021 return PoolInspection {
1022 checkpoint_probe: None,
1023 checkpoint_probe_error: Some(reason.clone()),
1024 graph_edge_integrity: None,
1025 graph_edge_integrity_error: Some(reason),
1026 };
1027 }
1028 };
1029
1030 let (checkpoint_probe, checkpoint_probe_error) = match checkpoint_probe(&conn) {
1031 Ok(probe) => (Some(probe), None),
1032 Err(e) => (
1033 None,
1034 Some(format!("PRAGMA wal_checkpoint(PASSIVE) failed: {e}")),
1035 ),
1036 };
1037 let (graph_edge_integrity, graph_edge_integrity_error) = match graph_edge_integrity(&conn) {
1038 Ok(integrity) => (Some(integrity), None),
1039 Err(e) => (
1040 None,
1041 Some(format!("graph-edge integrity query failed: {e}")),
1042 ),
1043 };
1044
1045 PoolInspection {
1046 checkpoint_probe,
1047 checkpoint_probe_error,
1048 graph_edge_integrity,
1049 graph_edge_integrity_error,
1050 }
1051}
1052
1053#[cfg(test)]
1054mod tests {
1055 use serial_test::serial;
1056
1057 use super::*;
1058 use crate::pool::{ConnectionPool, PoolConfig};
1059
1060 fn seeded_pool(dir: &tempfile::TempDir) -> (ConnectionPool, PathBuf) {
1061 let path = dir.path().join("diag.db");
1062 let pool = ConnectionPool::new(PoolConfig {
1063 path: Some(path.clone()),
1064 ..PoolConfig::default()
1065 })
1066 .expect("pool open");
1067 {
1068 let writer = pool.try_writer().expect("writer");
1069 writer
1070 .conn()
1071 .execute_batch(
1072 "CREATE TABLE t (x INTEGER); \
1073 CREATE TABLE graph_edges (
1074 namespace TEXT NOT NULL,
1075 id TEXT NOT NULL,
1076 PRIMARY KEY (namespace, id)
1077 ); \
1078 CREATE TABLE graph_edges_seq (
1079 seq INTEGER PRIMARY KEY AUTOINCREMENT,
1080 edge_id TEXT NOT NULL UNIQUE
1081 ); \
1082 INSERT INTO t VALUES (1), (2), (3);",
1083 )
1084 .expect("seed writes");
1085 }
1086 (pool, path)
1087 }
1088
1089 #[test]
1090 fn dropping_census_future_guard_requests_cooperative_stop() {
1091 let stopped = Arc::new(AtomicBool::new(false));
1092 let guard = StopCensusOnDrop {
1093 stopped: Arc::clone(&stopped),
1094 armed: true,
1095 };
1096
1097 drop(guard);
1098
1099 assert!(
1100 stopped.load(Ordering::SeqCst),
1101 "dropping diagnostics while its census worker is live must stop the PID/fd walk"
1102 );
1103 }
1104
1105 #[tokio::test]
1106 async fn runtime_audit_batch_fields_are_additive_and_unavailable_without_a_control() {
1107 let dir = tempfile::tempdir().expect("tempdir");
1108 let (pool, _path) = seeded_pool(&dir);
1109 let pool = Arc::new(pool);
1110
1111 let without_control = collect_with_audit_append_failures_interruptibly(
1112 Arc::clone(&pool),
1113 BuildIdentity::from_env("test", None),
1114 Duration::from_secs(30),
1115 0,
1116 )
1117 .await
1118 .expect("diagnostics succeed");
1119 assert!(without_control
1120 .writer_contention
1121 .audit_batch_flush_failures
1122 .is_none());
1123 assert!(
1124 without_control
1125 .writer_contention
1126 .audit_batch_flush_failures_unavailable_reason
1127 .is_some(),
1128 "no audit-batch control was supplied, so the field must carry a reason, not a \
1129 fabricated zero"
1130 );
1131 assert!(without_control
1132 .writer_contention
1133 .audit_degraded_rows
1134 .is_none());
1135 assert!(without_control.writer_contention.audit_degraded.is_none());
1136
1137 let with_control = collect_with_runtime_audit_metrics_interruptibly(
1138 Arc::clone(&pool),
1139 BuildIdentity::from_env("test", None),
1140 Duration::from_secs(30),
1141 0,
1142 Some(RuntimeAuditBatchMetrics {
1143 flush_failures: 3,
1144 degraded_rows: 7,
1145 degraded: true,
1146 admission_refused_obligations: 5,
1147 admission_unresolved_obligations: 2,
1148 }),
1149 )
1150 .await
1151 .expect("diagnostics succeed");
1152 assert_eq!(
1153 with_control.writer_contention.audit_batch_flush_failures,
1154 Some(3)
1155 );
1156 assert!(with_control
1157 .writer_contention
1158 .audit_batch_flush_failures_unavailable_reason
1159 .is_none());
1160 assert_eq!(with_control.writer_contention.audit_degraded_rows, Some(7));
1161 assert_eq!(with_control.writer_contention.audit_degraded, Some(true));
1162 assert_eq!(
1163 with_control
1164 .writer_contention
1165 .audit_admission_refused_obligations,
1166 Some(5),
1167 "an operator must be able to read the admission-refused obligation count from \
1168 db_diagnostics without a test-only feature gate (ADR-103 Amendment 3)"
1169 );
1170 assert!(with_control
1171 .writer_contention
1172 .audit_admission_refused_obligations_unavailable_reason
1173 .is_none());
1174 assert_eq!(
1175 with_control
1176 .writer_contention
1177 .audit_admission_unresolved_obligations,
1178 Some(2),
1179 "an operator must be able to distinguish enqueued-but-unresolved rows from \
1180 confirmed-refused rows (ADR-103 Amendment 3)"
1181 );
1182 assert!(with_control
1183 .writer_contention
1184 .audit_admission_unresolved_obligations_unavailable_reason
1185 .is_none());
1186 assert!(without_control
1187 .writer_contention
1188 .audit_admission_refused_obligations
1189 .is_none());
1190 assert!(without_control
1191 .writer_contention
1192 .audit_admission_refused_obligations_unavailable_reason
1193 .is_some());
1194 assert!(without_control
1195 .writer_contention
1196 .audit_admission_unresolved_obligations
1197 .is_none());
1198 assert!(without_control
1199 .writer_contention
1200 .audit_admission_unresolved_obligations_unavailable_reason
1201 .is_some());
1202
1203 assert_eq!(
1206 with_control.writer_contention.writer_acquisitions,
1207 without_control.writer_contention.writer_acquisitions
1208 );
1209 }
1210
1211 #[test]
1212 fn writer_task_pool_sourced_counters_are_always_populated_directly() {
1213 let dir = tempfile::tempdir().expect("tempdir");
1214 let (pool, _path) = seeded_pool(&dir);
1215
1216 let report = collect(
1217 &pool,
1218 BuildIdentity::from_env("9.9.9", None),
1219 Duration::from_secs(30),
1220 );
1221
1222 assert_eq!(report.writer_contention.writer_task_request_failures, 0);
1225 assert_eq!(report.writer_contention.writer_task_side_effects_unknown, 0);
1226 }
1227
1228 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1229 #[serial]
1230 async fn request_cancellation_after_passive_stops_before_graph_and_census() {
1231 let dir = tempfile::tempdir().expect("tempdir");
1232 let (pool, _) = seeded_pool(&dir);
1233 let pool = Arc::new(pool);
1234 TEST_REACHED_AFTER_PASSIVE.store(false, Ordering::SeqCst);
1235 TEST_PAUSE_AFTER_PASSIVE.store(true, Ordering::SeqCst);
1236 let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);
1237 let diagnostic_pool = Arc::clone(&pool);
1238 let task = tokio::spawn(crate::scope_request_read_cancellation(
1239 cancel_rx,
1240 async move {
1241 collect_with_audit_append_failures_interruptibly(
1242 diagnostic_pool,
1243 BuildIdentity::from_env("test", None),
1244 Duration::from_secs(30),
1245 0,
1246 )
1247 .await
1248 },
1249 ));
1250
1251 tokio::time::timeout(Duration::from_secs(1), async {
1252 while !TEST_REACHED_AFTER_PASSIVE.load(Ordering::SeqCst) {
1253 tokio::task::yield_now().await;
1254 }
1255 })
1256 .await
1257 .expect("diagnostics never completed its admitted PASSIVE phase");
1258 cancel_tx.send(true).unwrap();
1259 let result = tokio::time::timeout(Duration::from_secs(1), task)
1260 .await
1261 .expect("cancelled diagnostics did not stop promptly")
1262 .expect("diagnostics task panicked");
1263 TEST_PAUSE_AFTER_PASSIVE.store(false, Ordering::SeqCst);
1264 assert!(matches!(result, Err(StorageError::Timeout { .. })));
1265
1266 let one: i64 = pool
1267 .reader()
1268 .expect("diagnostics returned its connection")
1269 .conn()
1270 .query_row("SELECT 1", [], |row| row.get(0))
1271 .unwrap();
1272 assert_eq!(one, 1);
1273 }
1274
1275 #[test]
1276 fn checkpoint_probe_returns_a_well_formed_triple_on_a_file_backed_db() {
1277 let dir = tempfile::tempdir().expect("tempdir");
1278 let (pool, _path) = seeded_pool(&dir);
1279 let conn = pool
1280 .open_standalone_writer_untracked()
1281 .expect("standalone probe connection");
1282
1283 let probe = checkpoint_probe(&conn).expect("probe must succeed on a WAL database");
1284
1285 assert!(
1286 probe.busy == 0 || probe.busy == 1,
1287 "busy is a 0/1 flag, got {}",
1288 probe.busy
1289 );
1290 assert!(
1291 probe.log_frames >= 0,
1292 "a WAL database must report a non-negative frame count, got {}",
1293 probe.log_frames
1294 );
1295 assert!(
1296 probe.checkpointed_frames >= 0,
1297 "checkpointed frames must be non-negative, got {}",
1298 probe.checkpointed_frames
1299 );
1300 assert!(
1301 probe.checkpointed_frames <= probe.log_frames,
1302 "a PASSIVE pass cannot checkpoint more frames than the WAL holds: {probe:?}"
1303 );
1304 assert!(probe.pin_depth() >= 0, "pin depth clamps at 0: {probe:?}");
1305 }
1306
1307 #[test]
1310 #[serial(checkpoint_skip_metrics)]
1311 fn checkpoint_probe_does_not_perturb_the_adr091_counters() {
1312 crate::checkpoint::reset_checkpoint_metrics_for_tests();
1313 let dir = tempfile::tempdir().expect("tempdir");
1314 let (pool, _path) = seeded_pool(&dir);
1315 let conn = pool
1316 .open_standalone_writer_untracked()
1317 .expect("standalone probe connection");
1318
1319 let before = checkpoint_counters();
1320 for _ in 0..3 {
1321 checkpoint_probe(&conn).expect("probe must succeed");
1322 }
1323 let after = checkpoint_counters();
1324
1325 assert_eq!(
1326 before, after,
1327 "checkpoint_probe must leave every ADR-091 counter untouched"
1328 );
1329 }
1330
1331 #[test]
1332 fn wal_file_state_reports_the_sidecar_size_for_a_live_db() {
1333 let dir = tempfile::tempdir().expect("tempdir");
1334 let (_pool, path) = seeded_pool(&dir);
1335
1336 let state = wal_file_state(&path);
1337 assert!(
1338 state.wal_path.ends_with("diag.db-wal"),
1339 "WAL path is the db path plus a -wal suffix, got {}",
1340 state.wal_path
1341 );
1342 assert!(
1343 state.wal_size_bytes.is_some(),
1344 "a seeded WAL database must have a stat-able -wal file: {state:?}"
1345 );
1346 assert!(state.unavailable_reason.is_none(), "{state:?}");
1347 }
1348
1349 #[test]
1350 fn wal_file_state_degrades_with_a_reason_when_the_sidecar_is_absent() {
1351 let dir = tempfile::tempdir().expect("tempdir");
1352 let state = wal_file_state(&dir.path().join("never-created.db"));
1353 assert!(state.wal_size_bytes.is_none());
1354 assert!(
1355 state.unavailable_reason.is_some(),
1356 "an absent WAL file must carry a reason, not a silent zero: {state:?}"
1357 );
1358 }
1359
1360 #[test]
1361 fn collect_on_a_file_backed_db_carries_build_identity_and_every_counter() {
1362 let dir = tempfile::tempdir().expect("tempdir");
1363 let (pool, _path) = seeded_pool(&dir);
1364
1365 let report = collect(
1366 &pool,
1367 BuildIdentity::from_env("9.9.9", Some("deadbeef")),
1368 Duration::from_secs(30),
1369 );
1370
1371 assert_eq!(report.build.version, "9.9.9");
1372 assert_eq!(report.build.build_hash.as_deref(), Some("deadbeef"));
1373 assert!(report.db_path.is_some());
1374 assert!(
1375 report.checkpoint_probe.is_some(),
1376 "file-backed collect must land a probe; error was {:?}",
1377 report.checkpoint_probe_error
1378 );
1379 assert!(
1380 report.wal_file.as_ref().and_then(|w| w.wal_size_bytes) >= Some(0),
1381 "wal_size_bytes must be a non-negative byte count when present"
1382 );
1383
1384 let json = serde_json::to_value(&report).expect("report serializes");
1385 let counters = json
1386 .get("checkpoint_counters")
1387 .expect("counters section present");
1388 for key in [
1389 "last_observed_wal_pages",
1390 "truncate_attempts",
1391 "truncate_consecutive_failures",
1392 "checkpoint_skipped_ticks",
1393 "checkpoint_consecutive_skips",
1394 "checkpoint_last_skip_wal_pages",
1395 "checkpoint_pressure_elevated_ticks",
1396 "checkpoint_pressure_episodes_started",
1397 "checkpoint_pressure_episodes_recovered",
1398 "checkpoint_lifecycle_append_attempts",
1399 "checkpoint_lifecycle_append_failures",
1400 "checkpoint_lifecycle_enqueue_drops",
1401 "read_tx_max_age_evictions",
1402 ] {
1403 assert!(counters.get(key).is_some(), "counter {key} must be present");
1404 }
1405 assert_eq!(
1406 report.writer_contention.writer_acquisitions, 1,
1407 "the seed write checked the finite-wait pooled writer out once"
1408 );
1409 assert_eq!(report.writer_contention.pooled_writer_acquisitions, 1);
1410 assert_eq!(report.writer_contention.standalone_writer_acquisitions, 0);
1411 assert_eq!(report.writer_contention.writer_task_acquisitions, 0);
1412 assert_eq!(report.writer_contention.writer_acquisition_timeouts, 0);
1413 assert_eq!(
1414 report.graph_edge_integrity,
1415 Some(GraphEdgeIntegrity {
1416 duplicate_edge_id_groups: 0,
1417 graph_edges_rows: 0,
1418 graph_edges_seq_rows: 0,
1419 pre_v14_duplicate_edge_state_detected: false,
1420 })
1421 );
1422 assert!(report.graph_edge_integrity_error.is_none());
1423 assert!(report.writer_contention.audit_append_failures.is_none());
1424 assert!(
1425 report
1426 .writer_contention
1427 .audit_append_failures_unavailable_reason
1428 .is_some(),
1429 "a direct khive-db snapshot must not fabricate a runtime audit count"
1430 );
1431 }
1432
1433 #[test]
1434 fn diagnostics_composes_file_backed_standalone_acquisitions_without_counting_its_probe() {
1435 let dir = tempfile::tempdir().expect("tempdir");
1436 let (pool, _path) = seeded_pool(&dir);
1437
1438 drop(
1439 pool.open_standalone_writer()
1440 .expect("write-traffic standalone connection"),
1441 );
1442
1443 let report = collect_with_audit_append_failures(
1444 &pool,
1445 BuildIdentity::from_env("9.9.9", None),
1446 Duration::from_secs(30),
1447 0,
1448 );
1449 assert_eq!(report.writer_contention.writer_acquisitions, 2);
1450 assert_eq!(report.writer_contention.pooled_writer_acquisitions, 1);
1451 assert_eq!(report.writer_contention.standalone_writer_acquisitions, 1);
1452 assert_eq!(report.writer_contention.writer_task_acquisitions, 0);
1453 assert_eq!(report.writer_contention.writer_acquisition_timeouts, 0);
1454
1455 let second = collect_with_audit_append_failures(
1456 &pool,
1457 BuildIdentity::from_env("9.9.9", None),
1458 Duration::from_secs(30),
1459 0,
1460 );
1461 assert_eq!(
1462 second.writer_contention, report.writer_contention,
1463 "the diagnostics PASSIVE probe must not inflate write-traffic counters"
1464 );
1465 }
1466
1467 #[test]
1468 fn runtime_aware_collect_exposes_the_supplied_audit_failure_counter() {
1469 let pool = ConnectionPool::new(PoolConfig::default()).expect("in-memory pool");
1470
1471 let report = collect_with_audit_append_failures(
1472 &pool,
1473 BuildIdentity::from_env("9.9.9", None),
1474 Duration::from_secs(30),
1475 17,
1476 );
1477
1478 assert_eq!(report.writer_contention.audit_append_failures, Some(17));
1479 assert!(
1480 report
1481 .writer_contention
1482 .audit_append_failures_unavailable_reason
1483 .is_none(),
1484 "a supplied runtime counter must not carry an unavailable reason"
1485 );
1486 }
1487
1488 #[test]
1489 fn diagnostics_exposes_an_induced_writer_checkout_timeout() {
1490 let pool = ConnectionPool::new(PoolConfig {
1491 checkout_timeout: Duration::from_millis(1),
1492 ..PoolConfig::default()
1493 })
1494 .expect("in-memory pool");
1495
1496 let held = pool.writer().expect("first writer checkout succeeds");
1497 assert!(
1498 matches!(
1499 pool.writer(),
1500 Err(crate::SqliteError::WriterPoolCheckoutTimeout { .. })
1501 ),
1502 "holding the sole writer must exercise the typed timeout path"
1503 );
1504 drop(held);
1505
1506 let report = collect_with_audit_append_failures(
1507 &pool,
1508 BuildIdentity::from_env("9.9.9", None),
1509 Duration::from_secs(30),
1510 0,
1511 );
1512 assert_eq!(report.writer_contention.writer_acquisitions, 1);
1513 assert_eq!(report.writer_contention.pooled_writer_acquisitions, 1);
1514 assert_eq!(report.writer_contention.standalone_writer_acquisitions, 0);
1515 assert_eq!(report.writer_contention.writer_task_acquisitions, 0);
1516 assert_eq!(report.writer_contention.writer_acquisition_timeouts, 1);
1517 }
1518
1519 #[test]
1522 fn never_observed_sentinels_serialize_as_null() {
1523 let counters = CheckpointCounters {
1524 last_observed_wal_pages: None,
1525 truncate_attempts: 0,
1526 truncate_consecutive_failures: 0,
1527 checkpoint_skipped_ticks: 0,
1528 checkpoint_consecutive_skips: 0,
1529 checkpoint_last_skip_wal_pages: None,
1530 checkpoint_pressure_elevated_ticks: 0,
1531 checkpoint_pressure_episodes_started: 0,
1532 checkpoint_pressure_episodes_recovered: 0,
1533 checkpoint_lifecycle_append_attempts: 0,
1534 checkpoint_lifecycle_append_failures: 0,
1535 checkpoint_lifecycle_enqueue_drops: 0,
1536 read_tx_max_age_evictions: 0,
1537 };
1538 let json = serde_json::to_value(counters).expect("serializes");
1539 assert!(json["last_observed_wal_pages"].is_null());
1540 assert!(json["checkpoint_last_skip_wal_pages"].is_null());
1541 }
1542
1543 #[test]
1544 fn graph_edge_integrity_detects_the_pre_v14_duplicate_state() {
1545 let conn = Connection::open_in_memory().expect("in-memory sqlite");
1546 conn.execute_batch(
1547 "CREATE TABLE graph_edges (
1548 namespace TEXT NOT NULL,
1549 id TEXT NOT NULL,
1550 PRIMARY KEY (namespace, id)
1551 );
1552 CREATE TABLE graph_edges_seq (
1553 seq INTEGER PRIMARY KEY AUTOINCREMENT,
1554 edge_id TEXT NOT NULL UNIQUE
1555 );
1556 INSERT INTO graph_edges(namespace, id)
1557 VALUES ('alpha', 'shared-edge'), ('beta', 'shared-edge');
1558 INSERT INTO graph_edges_seq(edge_id) VALUES ('shared-edge');",
1559 )
1560 .expect("seed the state possible before the V14 uniqueness guard");
1561
1562 let integrity = graph_edge_integrity(&conn).expect("integrity query succeeds");
1563
1564 assert_eq!(integrity.duplicate_edge_id_groups, 1);
1565 assert_eq!(integrity.graph_edges_rows, 2);
1566 assert_eq!(integrity.graph_edges_seq_rows, 1);
1567 assert!(integrity.pre_v14_duplicate_edge_state_detected);
1568 }
1569
1570 #[test]
1571 fn graph_edge_integrity_does_not_mislabel_retained_delete_history_as_a_duplicate() {
1572 let conn = Connection::open_in_memory().expect("in-memory sqlite");
1573 conn.execute_batch(
1574 "CREATE TABLE graph_edges (
1575 namespace TEXT NOT NULL,
1576 id TEXT NOT NULL,
1577 PRIMARY KEY (namespace, id)
1578 );
1579 CREATE TABLE graph_edges_seq (
1580 seq INTEGER PRIMARY KEY AUTOINCREMENT,
1581 edge_id TEXT NOT NULL UNIQUE
1582 );
1583 INSERT INTO graph_edges(namespace, id) VALUES ('local', 'live-edge');
1584 INSERT INTO graph_edges_seq(edge_id)
1585 VALUES ('deleted-edge'), ('live-edge');",
1586 )
1587 .expect("seed a retained sequence row for a hard-deleted edge");
1588
1589 let integrity = graph_edge_integrity(&conn).expect("integrity query succeeds");
1590
1591 assert_eq!(integrity.duplicate_edge_id_groups, 0);
1592 assert_eq!(integrity.graph_edges_rows, 1);
1593 assert_eq!(integrity.graph_edges_seq_rows, 2);
1594 assert!(
1595 !integrity.pre_v14_duplicate_edge_state_detected,
1596 "ledger rows intentionally survive hard deletion; count mismatch alone is not the \
1597 pre-V14 duplicate state"
1598 );
1599 }
1600
1601 #[cfg(unix)]
1602 #[test]
1603 fn unmeasured_sidecar_cleanup_fields_are_absent_from_the_wire_payload() {
1604 let pin = wal_pin_attribution_from_census(crate::walpin::CensusResult {
1605 holders: std::collections::HashSet::new(),
1606 uninspectable_pids: Vec::new(),
1607 truncated: false,
1608 });
1609
1610 assert_eq!(pin.sidecar_listing_truncated, None);
1611 assert_eq!(pin.sidecar_entries_cleanup_would_reap, None);
1612 let json = serde_json::to_value(pin).expect("attribution serializes");
1613 assert!(
1614 json.get("sidecar_listing_truncated").is_none(),
1615 "a skipped enumeration must omit sidecar_listing_truncated, not fabricate false"
1616 );
1617 assert!(
1618 json.get("sidecar_entries_cleanup_would_reap").is_none(),
1619 "a skipped enumeration must omit sidecar_entries_cleanup_would_reap, not fabricate 0"
1620 );
1621 }
1622
1623 #[cfg(unix)]
1624 #[test]
1625 fn wal_pin_census_serializes_only_under_the_nested_carrier() {
1626 let pin = wal_pin_attribution_from_census(crate::walpin::CensusResult {
1627 holders: std::collections::HashSet::from([41, 7]),
1628 uninspectable_pids: vec![99],
1629 truncated: true,
1630 });
1631
1632 let json = serde_json::to_value(pin).expect("attribution serializes");
1633 assert_eq!(json["census"]["holder_pids"], serde_json::json!([7, 41]));
1634 assert_eq!(
1635 json["census"]["uninspectable_pids"],
1636 serde_json::json!([99])
1637 );
1638 assert_eq!(json["census"]["truncated"], true);
1639
1640 for duplicate in [
1641 "census_holder_pids",
1642 "census_uninspectable_pids",
1643 "census_truncated",
1644 "census_is_complete",
1645 ] {
1646 assert!(
1647 json.get(duplicate).is_none(),
1648 "wal_pin.{duplicate} must not duplicate wal_pin.census: {json}"
1649 );
1650 }
1651 }
1652
1653 #[test]
1657 fn probe_refuses_a_missing_configured_path_without_creating_it() {
1658 let dir = tempfile::tempdir().expect("tempdir");
1659 let (pool, path) = seeded_pool(&dir);
1660
1661 for suffix in ["", "-wal", "-shm"] {
1662 let mut p = path.as_os_str().to_os_string();
1663 p.push(suffix);
1664 let _ = std::fs::remove_file(PathBuf::from(p));
1665 }
1666 assert!(!path.exists(), "precondition: the database file is gone");
1667
1668 let report = collect(
1669 &pool,
1670 BuildIdentity::from_env("0.0.0", None),
1671 Duration::from_secs(30),
1672 );
1673
1674 assert!(
1675 report.checkpoint_probe.is_none(),
1676 "a missing database must not yield a probe result: {report:?}"
1677 );
1678 assert!(
1679 report.checkpoint_probe_error.is_some(),
1680 "a missing database must say why there is no probe: {report:?}"
1681 );
1682 assert!(
1683 !path.exists(),
1684 "a diagnostics request must never create the database it was asked about"
1685 );
1686 }
1687
1688 #[test]
1691 fn collect_degrades_gracefully_for_an_in_memory_backend() {
1692 let pool = ConnectionPool::new(PoolConfig::default()).expect("in-memory pool");
1693 let report = collect(
1694 &pool,
1695 BuildIdentity::from_env("0.0.0", None),
1696 Duration::from_secs(30),
1697 );
1698
1699 assert!(report.db_path.is_none());
1700 assert!(report.wal_file.is_none());
1701 assert!(report.checkpoint_probe.is_none());
1702 assert!(
1703 report.checkpoint_probe_error.is_some(),
1704 "an in-memory report must say WHY there is no probe"
1705 );
1706 assert!(!report.wal_pin.available);
1707 assert!(report.wal_pin.unavailable_reason.is_some());
1708 assert_eq!(report.wal_pin.status, WalPinAttributionStatus::Unavailable);
1709 assert!(matches!(
1710 report.wal_pin.census,
1711 WalPinCensus::Unavailable { .. }
1712 ));
1713 }
1714
1715 #[cfg(unix)]
1716 #[test]
1717 fn incomplete_holder_census_is_a_tagged_degraded_result() {
1718 let census = crate::walpin::CensusResult {
1719 holders: std::collections::HashSet::from([41, 7]),
1720 uninspectable_pids: vec![99, 99],
1721 truncated: true,
1722 };
1723
1724 let pin = wal_pin_attribution_from_census(census);
1725
1726 assert_eq!(pin.status, WalPinAttributionStatus::Degraded);
1727 assert!(!pin.available);
1728 assert!(!pin.census_is_complete);
1729 assert_eq!(pin.census_holder_pids, vec![7, 41]);
1730 assert_eq!(pin.census_uninspectable_pids, vec![99]);
1731 assert!(
1732 pin.unavailable_reason.as_deref().is_some_and(
1733 |reason| reason.contains("additional database holders cannot be ruled out")
1734 ),
1735 "the legacy reason must also fail loud for old consumers: {pin:?}"
1736 );
1737 match &pin.census {
1738 WalPinCensus::Incomplete {
1739 holder_pids,
1740 uninspectable_pids,
1741 truncated,
1742 reason,
1743 } => {
1744 assert_eq!(holder_pids, &vec![7, 41]);
1745 assert_eq!(uninspectable_pids, &vec![99]);
1746 assert!(*truncated);
1747 assert!(reason.contains("additional database holders cannot be ruled out"));
1748 }
1749 other => panic!("incomplete scan must serialize as incomplete, got {other:?}"),
1750 }
1751
1752 let json = serde_json::to_value(&pin).expect("serializes");
1753 assert_eq!(json["status"], "degraded");
1754 assert_eq!(json["census"]["status"], "incomplete");
1755 }
1756
1757 #[cfg(unix)]
1758 #[test]
1759 fn complete_holder_census_stays_explicit_while_attribution_is_degraded() {
1760 let census = crate::walpin::CensusResult {
1761 holders: std::collections::HashSet::from([7]),
1762 uninspectable_pids: Vec::new(),
1763 truncated: false,
1764 };
1765
1766 let pin = wal_pin_attribution_from_census(census);
1767
1768 assert_eq!(pin.status, WalPinAttributionStatus::Degraded);
1769 assert!(pin.census_is_complete);
1770 assert!(matches!(
1771 pin.census,
1772 WalPinCensus::Complete { ref holder_pids } if holder_pids == &vec![7]
1773 ));
1774 assert_eq!(
1775 pin.status_reasons.len(),
1776 1,
1777 "only missing sidecar reconciliation degrades a complete OS census"
1778 );
1779 }
1780
1781 #[cfg(unix)]
1784 #[test]
1785 fn wal_pin_attribution_reports_census_but_never_claims_full_attribution() {
1786 let dir = tempfile::tempdir().expect("tempdir");
1787 let (pool, path) = seeded_pool(&dir);
1788 let _ = &pool;
1789
1790 let pin = wal_pin_attribution(&path, Duration::from_secs(30));
1791
1792 assert!(
1793 !pin.fully_attributed,
1794 "sidecar reconciliation is not ported: this must never claim completeness"
1795 );
1796 assert!(
1797 pin.unavailable_reason.is_some(),
1798 "the gap must be explained, not silent: {pin:?}"
1799 );
1800 assert!(pin.sidecar_entries.is_empty());
1801 assert!(pin.reporting.is_empty());
1802 assert_eq!(pin.status, WalPinAttributionStatus::Degraded);
1803 assert!(matches!(
1804 pin.census,
1805 WalPinCensus::Complete { .. } | WalPinCensus::Incomplete { .. }
1806 ));
1807 }
1808}