1use std::collections::{BTreeMap, BTreeSet};
4use std::time::Duration;
5
6use crate::epoch::EpochId;
7use crate::fts::search::SearchCounters;
8use crate::fusion::FusionReport;
9use crate::graph::search::QueryQosClass;
10use crate::ingest::GraphSearchStats;
11use crate::planner::SegmentPlan;
12use crate::scan::ScanStats;
13use crate::wal::LogSeq;
14
15pub const QUERY_TIMING_ENABLED: bool = cfg!(feature = "query-timing");
18
19#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
22pub struct QueryTimings {
23 pub admission: Duration,
25 pub vector: Duration,
27 pub lexical_queue: Duration,
29 pub lexical: Duration,
31 pub fusion_cross_fill: Duration,
33}
34
35pub(crate) fn timing_start(
36 clock: &dyn crate::lifecycle::MonotonicClock,
37) -> Option<std::time::Instant> {
38 QUERY_TIMING_ENABLED.then(|| clock.now())
39}
40
41pub(crate) fn timing_elapsed(
42 clock: &dyn crate::lifecycle::MonotonicClock,
43 started: Option<std::time::Instant>,
44) -> Duration {
45 started.map_or(Duration::ZERO, |start| {
46 clock.now().saturating_duration_since(start)
47 })
48}
49
50#[derive(Clone, Debug, PartialEq)]
52#[non_exhaustive]
53pub struct QueryCounters {
54 pub scan: ScanStats,
56 pub graph: GraphSearchStats,
58 pub lexical: SearchCounters,
60 pub lexical_cache_hits: usize,
62 pub lexical_cache_builds: usize,
64}
65
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
68#[non_exhaustive]
69pub struct ObservedQos {
70 pub class: QueryQosClass,
72 pub relative_priority: i32,
74}
75
76impl ObservedQos {
77 #[cfg(any(target_os = "macos", target_os = "ios"))]
78 fn current() -> Self {
79 match crate::sys::darwin::observed_qos() {
80 Ok((class, relative_priority)) => Self {
81 class: match class {
82 crate::sys::darwin::QosClass::UserInteractive => QueryQosClass::UserInteractive,
83 crate::sys::darwin::QosClass::UserInitiated => QueryQosClass::UserInitiated,
84 crate::sys::darwin::QosClass::Default => QueryQosClass::Default,
85 crate::sys::darwin::QosClass::Utility => QueryQosClass::Utility,
86 crate::sys::darwin::QosClass::Background => QueryQosClass::Background,
87 crate::sys::darwin::QosClass::Unspecified => QueryQosClass::Unspecified,
88 },
89 relative_priority,
90 },
91 Err(_) => Self {
92 class: QueryQosClass::Unknown,
93 relative_priority: 0,
94 },
95 }
96 }
97
98 #[cfg(not(any(target_os = "macos", target_os = "ios")))]
99 const fn current() -> Self {
100 Self {
101 class: QueryQosClass::Unavailable,
102 relative_priority: 0,
103 }
104 }
105}
106
107#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
109#[non_exhaustive]
110pub struct ScanRescoreCounters {
111 pub coarse_rows: u64,
113 pub eligible_rows: u64,
115 pub candidates_rescored: u64,
117 pub coarse_bytes: u64,
120 pub rescore_bytes: u64,
122}
123
124#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
126#[non_exhaustive]
127pub struct MaterializationCounters {
128 pub row_lookups: u64,
130 pub text_copies: u64,
132 pub text_bytes: u64,
134}
135
136#[derive(Clone, Debug)]
138#[non_exhaustive]
139pub struct QueryDiagnostics {
140 pub snapshot_generation: u64,
142 pub indexed_through_seq: LogSeq,
144 pub plan: Vec<SegmentPlan>,
146 pub approximate: bool,
149 pub exact_rescore: bool,
151 pub scan_rescore: Option<ScanRescoreCounters>,
153 pub materialization: Option<MaterializationCounters>,
155 pub fusion: Option<FusionReport>,
157 pub hybrid: Option<HybridReport>,
159 pub hybrid_tier_resolution: Option<HybridTierResolution>,
162 pub requested_k: usize,
164 pub returned: usize,
166 pub budget_exhausted: bool,
168 pub counters: QueryCounters,
170 pub embedding_epoch: Option<EpochId>,
172 pub tokenizer_epoch: Option<crate::fts::tokenizer::TokenizerEpoch>,
174 pub observed_qos: ObservedQos,
176 pub elapsed: Duration,
178 pub timings: Option<QueryTimings>,
180}
181
182impl PartialEq for QueryDiagnostics {
186 fn eq(&self, other: &Self) -> bool {
187 let Self {
188 snapshot_generation,
189 indexed_through_seq,
190 plan,
191 approximate,
192 exact_rescore,
193 scan_rescore,
194 materialization,
195 fusion,
196 hybrid,
197 hybrid_tier_resolution,
198 requested_k,
199 returned,
200 budget_exhausted,
201 counters,
202 embedding_epoch,
203 tokenizer_epoch,
204 observed_qos,
205 elapsed: _,
206 timings: _,
207 } = self;
208
209 snapshot_generation == &other.snapshot_generation
210 && indexed_through_seq == &other.indexed_through_seq
211 && plan == &other.plan
212 && approximate == &other.approximate
213 && exact_rescore == &other.exact_rescore
214 && scan_rescore == &other.scan_rescore
215 && materialization == &other.materialization
216 && fusion == &other.fusion
217 && hybrid == &other.hybrid
218 && hybrid_tier_resolution == &other.hybrid_tier_resolution
219 && requested_k == &other.requested_k
220 && returned == &other.returned
221 && budget_exhausted == &other.budget_exhausted
222 && counters == &other.counters
223 && embedding_epoch == &other.embedding_epoch
224 && tokenizer_epoch == &other.tokenizer_epoch
225 && observed_qos == &other.observed_qos
226 }
227}
228
229#[derive(Clone, Copy, Debug, Eq, PartialEq)]
236pub struct HybridReport {
237 pub normalization_policy_version: u16,
239 pub lexical_full_materializations: usize,
243 pub provenance: crate::fusion::HybridProvenance,
245 pub window: usize,
247 pub vector_returned: usize,
249 pub lexical_returned: usize,
251 pub cross_filled_vector: usize,
253 pub cross_filled_lexical: usize,
255 pub total_cross_filled_vector: usize,
257 pub total_cross_filled_lexical: usize,
259 pub vector_candidates_produced: usize,
261 pub lexical_candidates_produced: usize,
263}
264
265#[derive(Clone, Copy, Debug, Eq, PartialEq)]
267pub enum HybridTierResolution {
268 Auto,
271 Exact,
274}
275
276pub(crate) struct VectorDiagnostics {
278 pub snapshot_generation: u64,
279 pub indexed_through_seq: LogSeq,
280 pub plan: Vec<SegmentPlan>,
281 pub approximate: bool,
282 pub exact_rescore: bool,
283 pub requested_k: usize,
284 pub returned: usize,
285 pub budget_exhausted: bool,
286 pub scan: ScanStats,
287 pub graph: GraphSearchStats,
288 pub epoch: Option<crate::epoch::EpochIdentity>,
289 pub elapsed: Duration,
290}
291
292pub(crate) struct LexicalDiagnostics {
293 pub snapshot_generation: u64,
294 pub indexed_through_seq: LogSeq,
295 pub requested_k: usize,
296 pub returned: usize,
297 pub counters: SearchCounters,
298 pub elapsed: Duration,
299}
300
301pub(crate) struct HybridDiagnostics {
302 pub snapshot_generation: u64,
303 pub indexed_through_seq: LogSeq,
304 pub plan: Vec<SegmentPlan>,
305 pub approximate: bool,
306 pub exact_rescore: bool,
307 pub requested_k: usize,
308 pub returned: usize,
309 pub scan: ScanStats,
310 pub graph: GraphSearchStats,
311 pub lexical: SearchCounters,
312 pub report: FusionReport,
313 pub hybrid: HybridReport,
314 pub tier_resolution: Option<HybridTierResolution>,
315 pub epoch: Option<crate::epoch::EpochIdentity>,
316 pub elapsed: Duration,
317}
318
319impl QueryDiagnostics {
320 pub(crate) fn vector(input: VectorDiagnostics) -> Self {
321 Self {
322 snapshot_generation: input.snapshot_generation,
323 indexed_through_seq: input.indexed_through_seq,
324 plan: input.plan,
325 approximate: input.approximate,
326 exact_rescore: input.exact_rescore,
327 scan_rescore: None,
328 materialization: None,
329 fusion: None,
330 hybrid: None,
331 hybrid_tier_resolution: None,
332 requested_k: input.requested_k,
333 returned: input.returned,
334 budget_exhausted: input.budget_exhausted,
335 counters: QueryCounters {
336 scan: input.scan,
337 graph: input.graph,
338 lexical: SearchCounters::default(),
339 lexical_cache_hits: 0,
340 lexical_cache_builds: 0,
341 },
342 embedding_epoch: input.epoch.map(|epoch| epoch.embedding),
343 tokenizer_epoch: None,
344 observed_qos: ObservedQos::current(),
345 elapsed: input.elapsed,
346 timings: None,
347 }
348 }
349
350 pub(crate) fn lexical(input: LexicalDiagnostics) -> Self {
351 Self {
352 snapshot_generation: input.snapshot_generation,
353 indexed_through_seq: input.indexed_through_seq,
354 plan: Vec::new(),
355 approximate: false,
356 exact_rescore: false,
357 scan_rescore: None,
358 materialization: None,
359 fusion: None,
360 hybrid: None,
361 hybrid_tier_resolution: None,
362 requested_k: input.requested_k,
363 returned: input.returned,
364 budget_exhausted: false,
365 counters: QueryCounters {
366 scan: ScanStats {
367 dims_touched: 0,
368 bytes_read: 0,
369 threads_used: 0,
370 worker_thread_ids: Vec::new(),
371 },
372 graph: GraphSearchStats::default(),
373 lexical: input.counters,
374 lexical_cache_hits: 0,
375 lexical_cache_builds: 0,
376 },
377 embedding_epoch: None,
378 tokenizer_epoch: Some(crate::fts::tokenizer::TokenizerConfig::text_default().epoch()),
379 observed_qos: ObservedQos::current(),
380 elapsed: input.elapsed,
381 timings: None,
382 }
383 }
384
385 pub(crate) fn hybrid(input: HybridDiagnostics) -> Self {
386 let budget_exhausted = input.report.budget_exhausted;
387 Self {
388 snapshot_generation: input.snapshot_generation,
389 indexed_through_seq: input.indexed_through_seq,
390 plan: input.plan,
391 approximate: input.approximate,
392 exact_rescore: input.exact_rescore,
393 scan_rescore: None,
394 materialization: None,
395 fusion: Some(input.report),
396 hybrid: Some(input.hybrid),
397 hybrid_tier_resolution: input.tier_resolution,
398 requested_k: input.requested_k,
399 returned: input.returned,
400 budget_exhausted,
401 counters: QueryCounters {
402 scan: input.scan,
403 graph: input.graph,
404 lexical: input.lexical,
405 lexical_cache_hits: 0,
406 lexical_cache_builds: 0,
407 },
408 embedding_epoch: input.epoch.map(|epoch| epoch.embedding),
409 tokenizer_epoch: Some(crate::fts::tokenizer::TokenizerConfig::text_default().epoch()),
410 observed_qos: ObservedQos::current(),
411 elapsed: input.elapsed,
412 timings: None,
413 }
414 }
415}
416
417#[derive(Clone, Debug, Eq, PartialEq)]
419#[non_exhaustive]
420pub struct SegmentHealth {
421 pub source: crate::ingest::RowSource,
423 pub tier: crate::tier::SegmentTier,
425 pub rows: u64,
427 pub tombstones: u64,
429 pub bytes: u64,
431 pub clustering_key_range: crate::segment::ClusteringKeyRange,
433}
434
435#[derive(Clone, Debug, Eq, PartialEq)]
437pub enum MaintenanceOutcome {
438 Complete,
440 BudgetExhausted,
442 Failed(String),
444}
445
446#[derive(Clone, Debug, Eq, PartialEq)]
448#[non_exhaustive]
449pub struct MaintenanceSummary {
450 pub graphs_built: u64,
452 pub bytes_consumed: u64,
454 pub checkpoints_resumed: u64,
456 pub outcome: MaintenanceOutcome,
458}
459
460#[derive(Clone, Copy, Debug, Eq, PartialEq)]
462pub enum HealthStatus {
463 Unchecked,
465 Healthy,
467 Degraded,
469 Unhealthy,
471}
472
473#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
475pub enum ArtifactRef {
476 Manifest,
478 Wal,
480 Segment {
482 id: crate::segment::SegmentId,
484 },
485 Region {
487 segment_id: crate::segment::SegmentId,
489 kind: u16,
491 },
492 Checkpoint {
494 name: String,
496 },
497 Active,
499}
500
501#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
503pub enum HealthFaultKind {
504 Io,
506 Format,
508 Semantic,
510 Query,
512 Accounting,
514}
515
516#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
518pub struct HealthFaultKey {
519 pub kind: HealthFaultKind,
521 pub artifact: ArtifactRef,
523}
524
525#[derive(Clone, Debug, Eq, PartialEq)]
527pub struct HealthFault {
528 pub key: HealthFaultKey,
530 pub detail: String,
532}
533
534#[derive(Clone, Debug, PartialEq)]
536#[non_exhaustive]
537pub struct Health {
538 pub status: HealthStatus,
540 pub unresolved_faults: BTreeMap<HealthFaultKey, HealthFault>,
542 pub stats: crate::lifecycle::Stats,
544 pub generation: u64,
546 pub segments: Vec<SegmentHealth>,
548 pub pending_docs: u64,
550 pub graph_coverage: f64,
553 pub retained_through: Option<i64>,
555 pub last_maintenance: Option<MaintenanceSummary>,
557}
558
559#[derive(Clone, Debug, Eq, PartialEq)]
561#[non_exhaustive]
562pub struct SelfCheckFailure {
563 pub stage: &'static str,
565 pub detail: String,
567}
568
569#[derive(Clone, Debug, PartialEq)]
571#[non_exhaustive]
572pub struct SelfCheckReport {
573 pub health_status: HealthStatus,
575 pub examined_artifacts: Vec<ArtifactRef>,
577 pub requested_samples: usize,
579 pub sampled_documents: usize,
581 pub seed: u64,
583 pub expected_hits: usize,
585 pub recalled_hits: usize,
587 pub recall: f64,
589 pub max_score_delta: f64,
591 pub failures: Vec<SelfCheckFailure>,
593}
594
595#[derive(Clone)]
596struct StoredSelfCheckDocument {
597 version: crate::ingest::DocumentVersion,
598 vector: Vec<f32>,
599}
600
601pub(crate) struct HealthState {
602 retained_through: Option<i64>,
603 last_maintenance: Option<MaintenanceSummary>,
604 status: HealthStatus,
605 unresolved_faults: BTreeMap<HealthFaultKey, HealthFault>,
606}
607
608impl Default for HealthState {
609 fn default() -> Self {
610 Self {
611 retained_through: None,
612 last_maintenance: None,
613 status: HealthStatus::Unchecked,
614 unresolved_faults: BTreeMap::new(),
615 }
616 }
617}
618
619impl crate::lifecycle::Store {
620 #[must_use]
623 pub fn self_check(&self, sample: usize, seed: u64) -> SelfCheckReport {
624 let (examined, artifact_faults) = self.audit_artifacts();
625 let mut report = self.self_check_inner(sample, seed, false);
626 let mut faults = artifact_faults;
627 if !report.failures.is_empty() || report.recall < 1.0 {
628 let detail = if report.failures.is_empty() {
629 format!("semantic sample recall was {}", report.recall)
630 } else {
631 report
632 .failures
633 .iter()
634 .map(|failure| format!("{}: {}", failure.stage, failure.detail))
635 .collect::<Vec<_>>()
636 .join("; ")
637 };
638 faults.push(HealthFault {
639 key: HealthFaultKey {
640 kind: HealthFaultKind::Query,
641 artifact: ArtifactRef::Active,
642 },
643 detail,
644 });
645 }
646 let (status, _) = self.update_health_ledger(&examined, faults);
647 report.health_status = status;
648 report.examined_artifacts = examined.into_iter().collect();
649 report
650 }
651
652 fn self_check_inner(&self, sample: usize, seed: u64, force_miss: bool) -> SelfCheckReport {
653 let documents = match self.self_check_documents() {
654 Ok(documents) => documents,
655 Err(failure) => {
656 return SelfCheckReport {
657 health_status: HealthStatus::Unchecked,
658 examined_artifacts: Vec::new(),
659 requested_samples: sample,
660 sampled_documents: 0,
661 seed,
662 expected_hits: 0,
663 recalled_hits: 0,
664 recall: 0.0,
665 max_score_delta: 0.0,
666 failures: vec![failure],
667 };
668 }
669 };
670 let mut ranked_samples = (0..documents.len())
671 .map(|index| {
672 let index_seed = u64::try_from(index).map_or(u64::MAX, |value| value);
673 (splitmix64(seed ^ index_seed), index)
674 })
675 .collect::<Vec<_>>();
676 ranked_samples.sort_unstable();
677 let mut sampled = ranked_samples
678 .into_iter()
679 .map(|(_, index)| index)
680 .collect::<Vec<_>>();
681 sampled.truncate(sample.min(sampled.len()));
682 let mut expected_hits = 0_usize;
683 let mut recalled_hits = 0_usize;
684 let mut max_score_delta = 0.0_f64;
685 let mut failures = Vec::new();
686 for sampled_index in &sampled {
687 let Some(query_document) = documents.get(*sampled_index) else {
688 failures.push(SelfCheckFailure {
689 stage: "sample",
690 detail: format!("sample index {sampled_index} left the gathered corpus"),
691 });
692 continue;
693 };
694 let k = documents.len().min(10);
695 let expected = brute_force_self_check(&documents, &query_document.vector, k);
696 expected_hits = expected_hits.saturating_add(expected.len());
697 let outcome = self.search(
698 crate::ingest::SearchRequest::new(&query_document.vector),
699 k,
700 crate::lifecycle::SearchOptions::default(),
701 crate::lifecycle::QueryControl::Cancel(crate::lifecycle::CancelToken::new()),
702 );
703 let mut actual = match outcome {
704 Ok(outcome) => outcome
705 .candidates
706 .iter()
707 .filter_map(|candidate| {
708 candidate
709 .document()
710 .map(|version| (version, candidate.score()))
711 })
712 .collect::<Vec<_>>(),
713 Err(error) => {
714 failures.push(SelfCheckFailure {
715 stage: "query",
716 detail: error.to_string(),
717 });
718 continue;
719 }
720 };
721 if force_miss && !actual.is_empty() {
722 actual.remove(0);
723 }
724 for (version, expected_score) in expected {
725 if let Some((_, actual_score)) = actual
726 .iter()
727 .find(|(actual_version, _)| *actual_version == version)
728 {
729 recalled_hits = recalled_hits.saturating_add(1);
730 max_score_delta = max_score_delta
731 .max((f64::from(*actual_score) - f64::from(expected_score)).abs());
732 }
733 }
734 }
735 let recall = if expected_hits == 0 {
736 1.0
737 } else {
738 recalled_hits as f64 / expected_hits as f64
739 };
740 SelfCheckReport {
741 health_status: HealthStatus::Unchecked,
742 examined_artifacts: Vec::new(),
743 requested_samples: sample,
744 sampled_documents: sampled.len(),
745 seed,
746 expected_hits,
747 recalled_hits,
748 recall,
749 max_score_delta,
750 failures,
751 }
752 }
753
754 fn self_check_documents(&self) -> Result<Vec<StoredSelfCheckDocument>, SelfCheckFailure> {
755 let state = self.state.lock().map_err(|_| SelfCheckFailure {
756 stage: "state",
757 detail: "store state lock was poisoned".to_owned(),
758 })?;
759 match *state {
760 crate::lifecycle::StoreState::Open => {}
761 crate::lifecycle::StoreState::Closing => {
762 return Err(SelfCheckFailure {
763 stage: "state",
764 detail: crate::lifecycle::StoreError::Closing.to_string(),
765 });
766 }
767 crate::lifecycle::StoreState::Closed => {
768 return Err(SelfCheckFailure {
769 stage: "state",
770 detail: crate::lifecycle::StoreError::Closed.to_string(),
771 });
772 }
773 }
774 let active = self.active.lock().map_err(|_| SelfCheckFailure {
775 stage: "active",
776 detail: "active segment lock was poisoned".to_owned(),
777 })?;
778 let active = active.as_ref().ok_or_else(|| SelfCheckFailure {
779 stage: "active",
780 detail: crate::lifecycle::StoreError::Closed.to_string(),
781 })?;
782 let mut documents = Vec::new();
783 let active_rows = active.segment.row_count();
784 if active_rows != 0 {
785 let active_vectors = active.segment.vectors();
786 let dims = active_vectors
787 .len()
788 .checked_div(active_rows)
789 .ok_or_else(|| SelfCheckFailure {
790 stage: "active",
791 detail: "active vector dimensions overflowed".to_owned(),
792 })?;
793 if dims.checked_mul(active_rows) != Some(active_vectors.len()) {
794 return Err(SelfCheckFailure {
795 stage: "active",
796 detail: "active vector rows are not rectangular".to_owned(),
797 });
798 }
799 let alive = active.segment.alive().map_err(|error| SelfCheckFailure {
800 stage: "active",
801 detail: error.to_string(),
802 })?;
803 for row in 0..active_rows {
804 let row_u32 = u32::try_from(row).map_err(|_| SelfCheckFailure {
805 stage: "active",
806 detail: "active row exceeds u32".to_owned(),
807 })?;
808 if !alive.is_alive(row_u32) {
809 continue;
810 }
811 let start = row.checked_mul(dims).ok_or_else(|| SelfCheckFailure {
812 stage: "active",
813 detail: "active vector offset overflowed".to_owned(),
814 })?;
815 let end = start.checked_add(dims).ok_or_else(|| SelfCheckFailure {
816 stage: "active",
817 detail: "active vector end overflowed".to_owned(),
818 })?;
819 let vector = active_vectors
820 .get(start..end)
821 .ok_or_else(|| SelfCheckFailure {
822 stage: "active",
823 detail: "active vector row is out of bounds".to_owned(),
824 })?;
825 let version = active
826 .segment
827 .document(row)
828 .ok_or_else(|| SelfCheckFailure {
829 stage: "active",
830 detail: format!("active row {row} has no document identity"),
831 })?;
832 documents.push(StoredSelfCheckDocument {
833 version,
834 vector: vector.to_vec(),
835 });
836 }
837 }
838 let snapshot = self
839 .snapshot
840 .read()
841 .map_err(|_| SelfCheckFailure {
842 stage: "snapshot",
843 detail: "published snapshot lock was poisoned".to_owned(),
844 })?
845 .as_ref()
846 .cloned()
847 .ok_or_else(|| SelfCheckFailure {
848 stage: "snapshot",
849 detail: crate::lifecycle::StoreError::Closed.to_string(),
850 })?;
851 for segment in snapshot.all_segments() {
852 let rows = segment.meta().row_count as usize;
853 let dims = segment.meta().dims as usize;
854 let vectors = segment.rescore_f32().map_err(|error| SelfCheckFailure {
855 stage: "sealed",
856 detail: error.to_string(),
857 })?;
858 if rows.checked_mul(dims) != Some(vectors.len()) {
859 return Err(SelfCheckFailure {
860 stage: "sealed",
861 detail: format!("segment {} f32 rows are not rectangular", segment.meta().id),
862 });
863 }
864 let alive = segment.alive().map_err(|error| SelfCheckFailure {
865 stage: "sealed",
866 detail: error.to_string(),
867 })?;
868 for row in 0..rows {
869 let row_u32 = u32::try_from(row).map_err(|_| SelfCheckFailure {
870 stage: "sealed",
871 detail: "sealed row exceeds u32".to_owned(),
872 })?;
873 if !alive.is_alive(row_u32) {
874 continue;
875 }
876 let start = row.checked_mul(dims).ok_or_else(|| SelfCheckFailure {
877 stage: "sealed",
878 detail: "sealed vector offset overflowed".to_owned(),
879 })?;
880 let end = start.checked_add(dims).ok_or_else(|| SelfCheckFailure {
881 stage: "sealed",
882 detail: "sealed vector end overflowed".to_owned(),
883 })?;
884 let vector = vectors.get(start..end).ok_or_else(|| SelfCheckFailure {
885 stage: "sealed",
886 detail: "sealed vector row is out of bounds".to_owned(),
887 })?;
888 let version = segment
889 .document_version(row)
890 .map_err(|error| SelfCheckFailure {
891 stage: "sealed",
892 detail: error.to_string(),
893 })?
894 .ok_or_else(|| SelfCheckFailure {
895 stage: "sealed",
896 detail: format!(
897 "segment {} row {row} has no document identity",
898 segment.meta().id
899 ),
900 })?;
901 documents.push(StoredSelfCheckDocument {
902 version,
903 vector: vector.to_vec(),
904 });
905 }
906 }
907 Ok(documents)
908 }
909
910 fn audit_artifacts(&self) -> (BTreeSet<ArtifactRef>, Vec<HealthFault>) {
911 use crate::manifest::io::{DurableLog, MANIFEST_FILE};
912 use crate::segment::layout::RegionKind;
913
914 let mut examined = BTreeSet::new();
915 let mut faults = Vec::new();
916 let mut record = |kind: HealthFaultKind, artifact: ArtifactRef, detail: String| {
917 faults.push(HealthFault {
918 key: HealthFaultKey { kind, artifact },
919 detail,
920 });
921 };
922
923 let wal_artifact = ArtifactRef::Wal;
924 examined.insert(wal_artifact.clone());
925 let wal_path = self.directory.join("wal.ze");
926 let wal = match crate::wal::WalReader::open(self.vfs.as_ref(), &wal_path) {
927 Ok(reader) => {
928 match reader.terminator() {
929 Some(crate::wal::replay::ReplayTerminator::CleanEnd) => {}
930 Some(crate::wal::replay::ReplayTerminator::InvalidHeader(
931 crate::wal::header::WalHeaderError::Missing,
932 )) if self.vfs.open(&wal_path).is_ok_and(|length| length == 0) => {}
933 Some(other) => record(
934 HealthFaultKind::Format,
935 wal_artifact.clone(),
936 format!("WAL replay stopped at {other:?}"),
937 ),
938 None => record(
939 HealthFaultKind::Io,
940 wal_artifact.clone(),
941 "WAL artifact is absent".to_owned(),
942 ),
943 }
944 Some(reader)
945 }
946 Err(error) => {
947 record(HealthFaultKind::Format, wal_artifact, error.to_string());
948 None
949 }
950 };
951
952 let manifest_artifact = ArtifactRef::Manifest;
953 examined.insert(manifest_artifact.clone());
954 let durable_end = wal.as_ref().map_or(u64::MAX, DurableLog::durable_end);
955 let manifest_path = self.directory.join(MANIFEST_FILE);
956 match self.vfs.open(&manifest_path) {
957 Ok(_) => {
958 if let Err(error) = crate::manifest::io::load_manifest(
959 self.vfs.as_ref(),
960 &manifest_path,
961 durable_end,
962 ) {
963 record(
964 HealthFaultKind::Format,
965 manifest_artifact,
966 error.to_string(),
967 );
968 }
969 }
970 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
971 Err(error) => record(HealthFaultKind::Io, manifest_artifact, error.to_string()),
972 }
973
974 let active_artifact = ArtifactRef::Active;
975 examined.insert(active_artifact.clone());
976 match self.active.lock() {
977 Ok(active) => match active.as_ref() {
978 Some(active) => {
979 if let Err(error) = active.segment.alive() {
980 record(
981 HealthFaultKind::Semantic,
982 active_artifact.clone(),
983 error.to_string(),
984 );
985 }
986 let rows = active.segment.row_count();
987 if rows != 0 && active.segment.vectors().len() % rows != 0 {
988 record(
989 HealthFaultKind::Semantic,
990 active_artifact.clone(),
991 "active vectors are not rectangular".to_owned(),
992 );
993 }
994 for row in 0..rows {
995 if active.segment.document(row).is_none() {
996 record(
997 HealthFaultKind::Semantic,
998 active_artifact.clone(),
999 format!("active row {row} has no document identity"),
1000 );
1001 break;
1002 }
1003 }
1004 }
1005 None => record(
1006 HealthFaultKind::Semantic,
1007 active_artifact,
1008 "active state is absent".to_owned(),
1009 ),
1010 },
1011 Err(_) => record(
1012 HealthFaultKind::Semantic,
1013 active_artifact,
1014 "active state lock is poisoned".to_owned(),
1015 ),
1016 }
1017
1018 match self.snapshot.read() {
1019 Ok(snapshot) => {
1020 if let Some(snapshot) = snapshot.as_ref() {
1021 for segment in snapshot.all_segments() {
1022 let expected = segment.meta();
1023 let segment_artifact = ArtifactRef::Segment { id: expected.id };
1024 examined.insert(segment_artifact.clone());
1025 let fresh = match crate::segment::reader::SegmentReader::open(
1026 self.vfs.as_ref(),
1027 &self.directory.join(expected.id.file_name()),
1028 expected.id,
1029 ) {
1030 Ok(fresh) => fresh,
1031 Err(error) => {
1032 record(
1033 HealthFaultKind::Format,
1034 segment_artifact,
1035 error.to_string(),
1036 );
1037 continue;
1038 }
1039 };
1040 let segment = &fresh;
1041 let mut region_failed = false;
1042 for entry in segment.directory() {
1043 let artifact = ArtifactRef::Region {
1044 segment_id: segment.meta().id,
1045 kind: entry.kind,
1046 };
1047 examined.insert(artifact.clone());
1048 if let Some(kind) = RegionKind::from_id(entry.kind)
1049 && let Err(error) = segment.region(kind)
1050 {
1051 region_failed = true;
1052 record(HealthFaultKind::Format, artifact, error.to_string());
1053 }
1054 }
1055 if !region_failed {
1056 if let Err(error) = segment.validate_all() {
1057 record(
1058 HealthFaultKind::Format,
1059 segment_artifact,
1060 error.to_string(),
1061 );
1062 continue;
1063 }
1064 if let Err(error) = segment.stored_text() {
1065 record(
1066 HealthFaultKind::Semantic,
1067 ArtifactRef::Region {
1068 segment_id: segment.meta().id,
1069 kind: RegionKind::StoredText.id(),
1070 },
1071 error.to_string(),
1072 );
1073 }
1074 for row in 0..segment.meta().row_count as usize {
1075 if let Err(error) = segment.document_version(row) {
1076 record(
1077 HealthFaultKind::Semantic,
1078 ArtifactRef::Region {
1079 segment_id: segment.meta().id,
1080 kind: RegionKind::DocumentVersions.id(),
1081 },
1082 error.to_string(),
1083 );
1084 break;
1085 }
1086 }
1087 }
1088 }
1089 }
1090 }
1091 Err(_) => record(
1092 HealthFaultKind::Semantic,
1093 ArtifactRef::Manifest,
1094 "published snapshot lock is poisoned".to_owned(),
1095 ),
1096 }
1097
1098 match self.vfs.list(&self.directory) {
1099 Ok(paths) => {
1100 if let Ok(state) = self.health_state.lock() {
1106 examined.extend(state.unresolved_faults.keys().filter_map(|key| {
1107 if matches!(key.artifact, ArtifactRef::Checkpoint { .. }) {
1108 Some(key.artifact.clone())
1109 } else {
1110 None
1111 }
1112 }));
1113 }
1114 for path in paths {
1115 let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
1116 continue;
1117 };
1118 let expected_magic: &[u8] =
1119 if name.starts_with(".tier-") && name.ends_with(".graph.checkpoint") {
1120 b"ZEVAMCP1"
1121 } else if name == ".consolidate.checkpoint" {
1122 b"ZECONCP1"
1123 } else {
1124 continue;
1125 };
1126 let artifact = ArtifactRef::Checkpoint {
1127 name: name.to_owned(),
1128 };
1129 examined.insert(artifact.clone());
1130 match self.vfs.read(&path) {
1131 Ok(bytes) => {
1132 let valid = bytes.len() >= 16
1133 && bytes.get(..8) == Some(expected_magic)
1134 && bytes
1135 .len()
1136 .checked_sub(8)
1137 .and_then(|start| {
1138 bytes.get(start..).and_then(|checksum| {
1139 checksum.try_into().ok().map(u64::from_le_bytes).map(
1140 |expected| {
1141 bytes.get(..start).is_some_and(|payload| {
1142 xxhash_rust::xxh3::xxh3_64(payload)
1143 == expected
1144 })
1145 },
1146 )
1147 })
1148 })
1149 .unwrap_or(false);
1150 if !valid {
1151 record(
1152 HealthFaultKind::Format,
1153 artifact,
1154 "maintenance checkpoint identity or checksum is invalid"
1155 .to_owned(),
1156 );
1157 }
1158 }
1159 Err(error) => record(HealthFaultKind::Io, artifact, error.to_string()),
1160 }
1161 }
1162 }
1163 Err(error) => record(
1164 HealthFaultKind::Io,
1165 ArtifactRef::Active,
1166 format!("list store artifacts: {error}"),
1167 ),
1168 }
1169 (examined, faults)
1170 }
1171
1172 fn update_health_ledger(
1173 &self,
1174 examined: &BTreeSet<ArtifactRef>,
1175 faults: Vec<HealthFault>,
1176 ) -> (HealthStatus, BTreeMap<HealthFaultKey, HealthFault>) {
1177 let Ok(mut state) = self.health_state.lock() else {
1178 return (HealthStatus::Unhealthy, BTreeMap::new());
1179 };
1180 state
1181 .unresolved_faults
1182 .retain(|key, _| !examined.contains(&key.artifact));
1183 for fault in faults {
1184 state.unresolved_faults.insert(fault.key.clone(), fault);
1185 }
1186 state.status = health_status(&state.unresolved_faults);
1187 (state.status, state.unresolved_faults.clone())
1188 }
1189
1190 pub(crate) fn record_retention(
1191 &self,
1192 retained_through: i64,
1193 ) -> Result<(), crate::lifecycle::StoreError> {
1194 let mut state = self.health_state.lock().map_err(|_| {
1195 crate::lifecycle::StoreError::Synchronization {
1196 component: "health state",
1197 }
1198 })?;
1199 state.retained_through = Some(retained_through);
1200 Ok(())
1201 }
1202
1203 pub(crate) fn record_maintenance(
1204 &self,
1205 report: &crate::tier::MaintenanceReport,
1206 ) -> Result<(), crate::lifecycle::StoreError> {
1207 let outcome = match &report.status {
1208 crate::tier::MaintenanceStatus::Complete => MaintenanceOutcome::Complete,
1209 crate::tier::MaintenanceStatus::BudgetExhausted => MaintenanceOutcome::BudgetExhausted,
1210 crate::tier::MaintenanceStatus::Failed(error) => {
1211 MaintenanceOutcome::Failed(error.to_string())
1212 }
1213 };
1214 let mut state = self.health_state.lock().map_err(|_| {
1215 crate::lifecycle::StoreError::Synchronization {
1216 component: "health state",
1217 }
1218 })?;
1219 state.last_maintenance = Some(MaintenanceSummary {
1220 graphs_built: report.graphs_built,
1221 bytes_consumed: report.bytes_consumed,
1222 checkpoints_resumed: report.checkpoints_resumed,
1223 outcome,
1224 });
1225 Ok(())
1226 }
1227
1228 pub fn health(&self) -> Result<Health, crate::lifecycle::StoreError> {
1230 let lifecycle_state = self
1231 .state
1232 .lock()
1233 .map_err(|_| crate::lifecycle::StoreError::Synchronization { component: "state" })?;
1234 match *lifecycle_state {
1235 crate::lifecycle::StoreState::Open => {}
1236 crate::lifecycle::StoreState::Closing => {
1237 return Err(crate::lifecycle::StoreError::Closing);
1238 }
1239 crate::lifecycle::StoreState::Closed => {
1240 return Err(crate::lifecycle::StoreError::Closed);
1241 }
1242 }
1243 let stats = self.stats_while_open()?;
1244 let active =
1245 self.active
1246 .lock()
1247 .map_err(|_| crate::lifecycle::StoreError::Synchronization {
1248 component: "active segment",
1249 })?;
1250 let active = active
1251 .as_ref()
1252 .ok_or(crate::lifecycle::StoreError::Closed)?;
1253 let generation = active.generation;
1254 let active_tombstones = active.segment.tombstone_count();
1255 let mut segments = Vec::new();
1256 if stats.active_row_count != 0 {
1257 segments.push(SegmentHealth {
1258 source: crate::ingest::RowSource::Active,
1259 tier: crate::tier::SegmentTier::ActiveScan,
1260 rows: stats.active_row_count,
1261 tombstones: active_tombstones,
1262 bytes: stats.active_segment_bytes,
1263 clustering_key_range: crate::segment::ClusteringKeyRange::Unstamped,
1264 });
1265 }
1266 let snapshot = self
1267 .snapshot
1268 .read()
1269 .map_err(|_| crate::lifecycle::StoreError::Synchronization {
1270 component: "published snapshot",
1271 })?
1272 .as_ref()
1273 .cloned()
1274 .ok_or(crate::lifecycle::StoreError::Closed)?;
1275 for segment in snapshot.all_segments() {
1276 let alive = segment
1277 .alive()
1278 .map_err(crate::lifecycle::StoreError::Segment)?;
1279 let tier =
1280 if segment.directory().iter().any(|entry| {
1281 entry.kind == crate::segment::layout::RegionKind::GraphNodeBlocks.id()
1282 }) {
1283 crate::tier::SegmentTier::SealedGraph
1284 } else {
1285 crate::tier::SegmentTier::SealedScan
1286 };
1287 segments.push(SegmentHealth {
1288 source: crate::ingest::RowSource::Sealed(segment.meta().id),
1289 tier,
1290 rows: u64::from(segment.meta().row_count),
1291 tombstones: alive.tombstone_count(),
1292 bytes: u64::try_from(segment.mapped_bytes()).map_err(|_| {
1293 crate::lifecycle::StoreError::Statistics {
1294 component: "segment health bytes",
1295 source: std::io::Error::other("mapped segment bytes exceed u64"),
1296 }
1297 })?,
1298 clustering_key_range: segment.meta().clustering_key_range,
1299 });
1300 }
1301 segments.sort_unstable_by_key(|segment| segment.source);
1302 let (alive_rows, graph_rows) =
1303 segments
1304 .iter()
1305 .fold((0_u64, 0_u64), |(alive_total, graph_total), segment| {
1306 let alive = segment.rows.saturating_sub(segment.tombstones);
1307 (
1308 alive_total.saturating_add(alive),
1309 graph_total.saturating_add(
1310 if segment.tier == crate::tier::SegmentTier::SealedGraph {
1311 alive
1312 } else {
1313 0
1314 },
1315 ),
1316 )
1317 });
1318 let graph_coverage = if alive_rows == 0 {
1319 1.0
1320 } else {
1321 graph_rows as f64 / alive_rows as f64
1322 };
1323 let health_state = self.health_state.lock().map_err(|_| {
1324 crate::lifecycle::StoreError::Synchronization {
1325 component: "health state",
1326 }
1327 })?;
1328 Ok(Health {
1329 status: health_state.status,
1330 unresolved_faults: health_state.unresolved_faults.clone(),
1331 stats,
1332 generation,
1333 segments,
1334 pending_docs: stats.active_row_count.saturating_sub(active_tombstones),
1335 graph_coverage,
1336 retained_through: health_state.retained_through,
1337 last_maintenance: health_state.last_maintenance.clone(),
1338 })
1339 }
1340}
1341
1342fn health_status(faults: &BTreeMap<HealthFaultKey, HealthFault>) -> HealthStatus {
1343 if faults.is_empty() {
1344 return HealthStatus::Healthy;
1345 }
1346 if faults.keys().any(|key| {
1347 matches!(
1348 key.kind,
1349 HealthFaultKind::Io | HealthFaultKind::Format | HealthFaultKind::Accounting
1350 )
1351 }) {
1352 HealthStatus::Unhealthy
1353 } else {
1354 HealthStatus::Degraded
1355 }
1356}
1357
1358fn splitmix64(mut value: u64) -> u64 {
1359 value = value.wrapping_add(0x9e37_79b9_7f4a_7c15);
1360 let mut mixed = value;
1361 mixed = (mixed ^ (mixed >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
1362 mixed = (mixed ^ (mixed >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
1363 mixed ^ (mixed >> 31)
1364}
1365
1366fn brute_force_self_check(
1367 documents: &[StoredSelfCheckDocument],
1368 query: &[f32],
1369 k: usize,
1370) -> Vec<(crate::ingest::DocumentVersion, f32)> {
1371 let mut scored = documents
1372 .iter()
1373 .filter(|document| document.vector.len() == query.len())
1374 .map(|document| {
1375 let distance = document
1376 .vector
1377 .iter()
1378 .zip(query)
1379 .map(|(left, right)| {
1380 let delta = f64::from(*left) - f64::from(*right);
1381 delta * delta
1382 })
1383 .sum::<f64>();
1384 (document.version, -(distance as f32))
1385 })
1386 .collect::<Vec<_>>();
1387 scored.sort_unstable_by(|left, right| {
1388 right
1389 .1
1390 .total_cmp(&left.1)
1391 .then_with(|| left.0.doc_id().cmp(&right.0.doc_id()))
1392 .then_with(|| left.0.revision().cmp(&right.0.revision()))
1393 });
1394 scored.truncate(k);
1395 scored
1396}
1397
1398#[cfg(test)]
1399#[allow(clippy::expect_used)]
1400mod tests {
1401 use tempfile::tempdir;
1402
1403 use crate::ingest::{DocId, DocumentVersion, IngestBatch, IngestDocument, Revision};
1404 use crate::lifecycle::{OpenOptions, Store};
1405
1406 fn populated_store() -> (tempfile::TempDir, Store) {
1407 let directory = tempdir().expect("self-check directory");
1408 let store = Store::open(directory.path(), OpenOptions::default()).expect("open store");
1409 let documents = (0..8)
1410 .map(|row| {
1411 IngestDocument::new(
1412 DocumentVersion::new(DocId::new(row + 1), Revision::new(1)),
1413 vec![row as f32, -(row as f32)],
1414 )
1415 })
1416 .collect();
1417 store
1418 .ingest(IngestBatch::new(documents))
1419 .expect("ingest self-check corpus");
1420 (directory, store)
1421 }
1422
1423 #[test]
1424 fn self_check_is_deterministic_per_seed() {
1425 let (_directory, store) = populated_store();
1426 assert_eq!(store.self_check(5, 0x18), store.self_check(5, 0x18));
1427 store.close().expect("close store");
1428 }
1429
1430 #[test]
1431 fn self_check_reports_a_planted_degradation() {
1432 let (_directory, store) = populated_store();
1433 let report = store.self_check_inner(5, 0x18, true);
1434 println!(
1435 "planted self-check degradation recall={} recalled={}/{}",
1436 report.recall, report.recalled_hits, report.expected_hits
1437 );
1438 assert!(report.failures.is_empty());
1439 assert!(report.recall < 1.0);
1440 store.close().expect("close store");
1441 }
1442
1443 #[test]
1444 fn health_status_distinguishes_empty_degraded_and_unhealthy_ledgers() {
1445 let mut faults = std::collections::BTreeMap::new();
1446 assert_eq!(super::health_status(&faults), super::HealthStatus::Healthy);
1447
1448 let degraded = super::HealthFault {
1449 key: super::HealthFaultKey {
1450 kind: super::HealthFaultKind::Query,
1451 artifact: super::ArtifactRef::Active,
1452 },
1453 detail: "semantic sample mismatch".to_owned(),
1454 };
1455 faults.insert(degraded.key.clone(), degraded);
1456 assert_eq!(super::health_status(&faults), super::HealthStatus::Degraded);
1457
1458 let unhealthy = super::HealthFault {
1459 key: super::HealthFaultKey {
1460 kind: super::HealthFaultKind::Format,
1461 artifact: super::ArtifactRef::Wal,
1462 },
1463 detail: "damaged WAL".to_owned(),
1464 };
1465 faults.insert(unhealthy.key.clone(), unhealthy);
1466 assert_eq!(
1467 super::health_status(&faults),
1468 super::HealthStatus::Unhealthy
1469 );
1470 }
1471}