1use std::path::{Path, PathBuf};
20use std::sync::Arc;
21
22use crate::fts::tokenizer::{Analyzer, TokenizerConfig};
23use crate::ingest::{PURGE_INTENT_FILE, read_intent};
24use crate::lifecycle::stats::Accounting;
25use crate::manifest::io::{DurableLog, MANIFEST_FILE};
26use crate::manifest::{Manifest, decode_manifest};
27use crate::meta::Schema;
28use crate::segment::SegmentError;
29use crate::segment::SegmentMeta;
30use crate::segment::layout::RegionKind;
31use crate::segment::reader::SegmentReader;
32use crate::vfs::{StdVfs, Vfs};
33use crate::wal::replay::ReplayTerminator;
34use crate::wal::{WalReadError, WalReader};
35
36const WAL_FILE: &str = "wal.ze";
38
39#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
41#[non_exhaustive]
42pub enum FindingKind {
43 ManifestMissing,
46 ManifestCorrupt,
48 ManifestAheadOfWal,
50 SegmentMissing,
52 SegmentCorrupt,
54 SegmentMismatch,
56 SegmentRegionCorrupt,
58 SegmentIndexInvalid,
60 WalMissing,
62 WalHeaderCorrupt,
64 WalRecordCorrupt,
66 WalRecordInvalid,
68 Unreadable,
70 PurgeIntentCorrupt,
72 #[cfg(feature = "graph-cypher")]
74 GraphObjectMissing,
75 #[cfg(feature = "graph-cypher")]
77 GraphObjectCorrupt,
78 #[cfg(feature = "graph-cypher")]
80 GraphInventoryInvalid,
81}
82
83impl FindingKind {
84 #[must_use]
86 pub const fn as_str(self) -> &'static str {
87 match self {
88 Self::ManifestMissing => "manifestMissing",
89 Self::ManifestCorrupt => "manifestCorrupt",
90 Self::ManifestAheadOfWal => "manifestAheadOfWal",
91 Self::SegmentMissing => "segmentMissing",
92 Self::SegmentCorrupt => "segmentCorrupt",
93 Self::SegmentMismatch => "segmentMismatch",
94 Self::SegmentRegionCorrupt => "segmentRegionCorrupt",
95 Self::SegmentIndexInvalid => "segmentIndexInvalid",
96 Self::WalMissing => "walMissing",
97 Self::WalHeaderCorrupt => "walHeaderCorrupt",
98 Self::WalRecordCorrupt => "walRecordCorrupt",
99 Self::WalRecordInvalid => "walRecordInvalid",
100 Self::Unreadable => "unreadable",
101 Self::PurgeIntentCorrupt => "purgeIntentCorrupt",
102 #[cfg(feature = "graph-cypher")]
103 Self::GraphObjectMissing => "graphObjectMissing",
104 #[cfg(feature = "graph-cypher")]
105 Self::GraphObjectCorrupt => "graphObjectCorrupt",
106 #[cfg(feature = "graph-cypher")]
107 Self::GraphInventoryInvalid => "graphInventoryInvalid",
108 }
109 }
110}
111
112#[derive(Clone, Debug, Eq, PartialEq)]
114pub struct Finding {
115 pub kind: FindingKind,
117 pub file: String,
119 pub offset: Option<u64>,
121 pub detail: String,
123}
124
125#[derive(Clone, Debug, Default, Eq, PartialEq)]
127pub struct VerifyReport {
128 pub findings: Vec<Finding>,
130 pub generation: u64,
132 pub segments_checked: u64,
134 pub wal_records_checked: u64,
136}
137
138impl VerifyReport {
139 #[must_use]
141 pub fn is_clean(&self) -> bool {
142 self.findings.is_empty()
143 }
144}
145
146#[derive(Debug)]
149pub enum VerifyError {
150 NotFound {
152 path: PathBuf,
154 },
155 NotDirectory {
157 path: PathBuf,
159 },
160 Io {
162 path: PathBuf,
164 source: std::io::Error,
166 },
167 Tokenizer(String),
169}
170
171impl std::fmt::Display for VerifyError {
172 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
173 match self {
174 Self::NotFound { path } => {
175 write!(formatter, "store path {} does not exist", path.display())
176 }
177 Self::NotDirectory { path } => {
178 write!(
179 formatter,
180 "store path {} is not a directory",
181 path.display()
182 )
183 }
184 Self::Io { path, source } => {
185 write!(formatter, "store I/O {}: {source}", path.display())
186 }
187 Self::Tokenizer(detail) => write!(formatter, "verify tokenizer: {detail}"),
188 }
189 }
190}
191
192impl std::error::Error for VerifyError {
193 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
194 match self {
195 Self::Io { source, .. } => Some(source),
196 Self::NotFound { .. } | Self::NotDirectory { .. } | Self::Tokenizer(_) => None,
197 }
198 }
199}
200
201pub fn verify_store(path: impl AsRef<Path>) -> Result<VerifyReport, VerifyError> {
203 verify_store_on_vfs(&StdVfs, path.as_ref())
204}
205
206pub fn verify_store_on_vfs(vfs: &dyn Vfs, path: &Path) -> Result<VerifyReport, VerifyError> {
209 match vfs.ensure_directory(path, false) {
210 Ok(true) => {}
211 Ok(false) => {
212 return Err(VerifyError::NotDirectory {
213 path: path.to_path_buf(),
214 });
215 }
216 Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
217 return Err(VerifyError::NotFound {
218 path: path.to_path_buf(),
219 });
220 }
221 Err(source) => {
222 return Err(VerifyError::Io {
223 path: path.to_path_buf(),
224 source,
225 });
226 }
227 }
228 let entries = vfs.list(path).map_err(|source| VerifyError::Io {
229 path: path.to_path_buf(),
230 source,
231 })?;
232 let has_segment_files = entries.iter().any(|entry| is_segment_file(entry));
233 let mut walk = Walk {
234 vfs,
235 directory: path,
236 report: VerifyReport::default(),
237 };
238 let manifest = walk.manifest();
239 if let Some(decoded) = manifest.decoded() {
240 walk.report.generation = decoded.generation;
241 walk.segments(decoded);
242 #[cfg(feature = "graph-cypher")]
243 if let Some(graph) = &decoded.graph {
244 match graph.state() {
245 Ok(state) => walk.graph_objects(&graph.objects, state.store),
246 Err(error) => walk.record(
247 FindingKind::GraphInventoryInvalid,
248 MANIFEST_FILE,
249 None,
250 error.to_string(),
251 ),
252 }
253 }
254 }
255 walk.wal(&manifest, has_segment_files)?;
256 walk.purge_intent();
257 Ok(walk.report)
258}
259
260struct Walk<'a> {
261 vfs: &'a dyn Vfs,
262 directory: &'a Path,
263 report: VerifyReport,
264}
265
266enum ManifestState {
268 Absent,
270 Damaged,
272 Decoded(Box<Manifest>),
274}
275
276impl ManifestState {
277 fn decoded(&self) -> Option<&Manifest> {
278 match self {
279 Self::Decoded(manifest) => Some(manifest),
280 Self::Absent | Self::Damaged => None,
281 }
282 }
283}
284
285impl Walk<'_> {
286 fn record(&mut self, kind: FindingKind, file: &str, offset: Option<u64>, detail: String) {
287 self.report.findings.push(Finding {
288 kind,
289 file: file.to_owned(),
290 offset,
291 detail,
292 });
293 }
294
295 fn manifest(&mut self) -> ManifestState {
296 let bytes = match self.vfs.read(&self.directory.join(MANIFEST_FILE)) {
297 Ok(bytes) => bytes,
298 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
299 return ManifestState::Absent;
300 }
301 Err(error) => {
302 self.record(
303 FindingKind::Unreadable,
304 MANIFEST_FILE,
305 None,
306 error.to_string(),
307 );
308 return ManifestState::Damaged;
309 }
310 };
311 match decode_manifest(MANIFEST_FILE, &bytes) {
312 Ok(manifest) => ManifestState::Decoded(Box::new(manifest)),
313 Err(error) => {
314 self.record(
315 FindingKind::ManifestCorrupt,
316 MANIFEST_FILE,
317 None,
318 error.to_string(),
319 );
320 ManifestState::Damaged
321 }
322 }
323 }
324
325 fn segments(&mut self, manifest: &Manifest) {
326 let schema = Arc::new(manifest.schema.clone());
330 for expected in &manifest.segments {
331 self.report.segments_checked = self.report.segments_checked.saturating_add(1);
332 self.segment(expected, &schema);
333 }
334 }
335
336 #[cfg(feature = "graph-cypher")]
337 fn graph_objects(
338 &mut self,
339 objects: &[crate::manifest::GraphObject],
340 store: crate::property_graph::StoreInstanceId,
341 ) {
342 use crate::property_graph::storage::{
343 allocation::artifact_path,
344 artifact::{self, ContainerKind},
345 };
346 for object in objects {
347 let path = artifact_path(self.directory, object.artifact);
348 let file = path
349 .file_name()
350 .map(|name| name.to_string_lossy().into_owned())
351 .unwrap_or_default();
352 let length = match self.vfs.open(&path) {
353 Ok(length) => length,
354 Err(error) => {
355 let kind = if error.kind() == std::io::ErrorKind::NotFound {
356 FindingKind::GraphObjectMissing
357 } else {
358 FindingKind::Unreadable
359 };
360 self.record(kind, &file, None, error.to_string());
361 continue;
362 }
363 };
364 if length != object.length
365 || length > crate::property_graph::storage::artifact::MAX_ARTIFACT_BYTES as u64
366 {
367 self.record(
368 FindingKind::GraphObjectCorrupt,
369 &file,
370 None,
371 "graph object length differs from its inventory or exceeds the format bound"
372 .into(),
373 );
374 continue;
375 }
376 let bytes = match self.vfs.read(&path) {
377 Ok(bytes) => bytes,
378 Err(error) => {
379 let kind = if error.kind() == std::io::ErrorKind::NotFound {
380 FindingKind::GraphObjectMissing
381 } else {
382 FindingKind::Unreadable
383 };
384 self.record(kind, &file, None, error.to_string());
385 continue;
386 }
387 };
388 let result = artifact::decode(
389 ContainerKind::Object,
390 Some((store, object.artifact)),
391 &bytes,
392 );
393 match result {
394 Ok(frame)
395 if bytes.len() as u64 == object.length
396 && frame.file_checksum() == object.checksum => {}
397 Ok(_) => self.record(
398 FindingKind::GraphObjectCorrupt,
399 &file,
400 None,
401 "graph object differs from its inventory length/checksum".into(),
402 ),
403 Err(error) => self.record(
404 FindingKind::GraphObjectCorrupt,
405 &file,
406 None,
407 error.to_string(),
408 ),
409 }
410 }
411 }
412
413 #[cfg(feature = "graph-cypher")]
414 fn graph_replay(
415 &mut self,
416 manifest: &Manifest,
417 commits: &[crate::ingest::RecoveredGraphCommit],
418 ) {
419 let Some(graph) = &manifest.graph else {
420 return;
421 };
422 if self.report.findings.iter().any(|finding| {
423 matches!(
424 finding.kind,
425 FindingKind::GraphObjectMissing
426 | FindingKind::GraphObjectCorrupt
427 | FindingKind::Unreadable
428 )
429 }) {
430 return;
431 }
432 if let Err(error) = crate::lifecycle::native_graph::recovery::verify_unified(
433 self.vfs,
434 self.directory,
435 graph,
436 manifest.generation,
437 commits,
438 ) {
439 if let Some(path) = &error.file {
443 let state = match graph.state() {
444 Ok(state) => state,
445 Err(_) => return,
446 };
447 match self.vfs.read(path) {
448 Err(source) => {
449 let kind = if source.kind() == std::io::ErrorKind::NotFound {
450 FindingKind::GraphObjectMissing
451 } else {
452 FindingKind::Unreadable
453 };
454 let file = path
455 .file_name()
456 .map(|name| name.to_string_lossy().into_owned())
457 .unwrap_or_default();
458 self.record(kind, &file, None, source.to_string());
459 return;
460 }
461 Ok(bytes) => {
462 let artifact = path
463 .file_name()
464 .and_then(|name| name.to_str())
465 .and_then(|name| name.strip_prefix("graph-"))
466 .and_then(|name| name.strip_suffix(".zgraph"))
467 .and_then(|name| u128::from_str_radix(name, 16).ok())
468 .and_then(|id| {
469 crate::property_graph::storage::artifact::ArtifactId::new(id).ok()
470 });
471 let Some(artifact) = artifact else {
472 self.record(
473 FindingKind::GraphInventoryInvalid,
474 MANIFEST_FILE,
475 None,
476 "invalid graph participant filename".into(),
477 );
478 return;
479 };
480 let result = crate::property_graph::storage::artifact::decode(
481 crate::property_graph::storage::artifact::ContainerKind::Object,
482 Some((state.store, artifact)),
483 &bytes,
484 );
485 if let Err(source) = result {
486 let file = path
487 .file_name()
488 .map(|name| name.to_string_lossy().into_owned())
489 .unwrap_or_default();
490 self.record(
491 FindingKind::GraphObjectCorrupt,
492 &file,
493 None,
494 source.to_string(),
495 );
496 return;
497 }
498 }
499 }
500 }
501 let (kind, file) = if commits.is_empty() {
502 (FindingKind::GraphInventoryInvalid, MANIFEST_FILE)
503 } else {
504 (FindingKind::WalRecordInvalid, WAL_FILE)
505 };
506 self.record(kind, file, None, error.source.to_string());
507 }
508 }
509
510 fn purge_intent(&mut self) {
513 let path = self.directory.join(PURGE_INTENT_FILE);
514 match self.vfs.open(&path) {
515 Ok(_) => {}
516 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return,
517 Err(error) => {
518 self.record(
519 FindingKind::Unreadable,
520 PURGE_INTENT_FILE,
521 None,
522 error.to_string(),
523 );
524 return;
525 }
526 }
527 if let Err(error) = read_intent(self.vfs, self.directory) {
528 self.record(
529 FindingKind::PurgeIntentCorrupt,
530 PURGE_INTENT_FILE,
531 None,
532 error.to_string(),
533 );
534 }
535 }
536
537 fn segment(&mut self, expected: &SegmentMeta, schema: &Arc<Schema>) {
538 let file = expected.id.file_name();
539 let path = self.directory.join(&file);
540 let reader = match SegmentReader::open(self.vfs, &path, expected.id) {
541 Ok(reader) => reader.with_collection_schema(Arc::clone(schema)),
542 Err(SegmentError::Io { source, .. })
543 if source.kind() == std::io::ErrorKind::NotFound =>
544 {
545 self.record(
546 FindingKind::SegmentMissing,
547 &file,
548 None,
549 format!(
550 "the manifest references segment {} with {} rows",
551 expected.id, expected.row_count
552 ),
553 );
554 return;
555 }
556 Err(SegmentError::Io { source, .. }) => {
557 self.record(FindingKind::Unreadable, &file, None, source.to_string());
558 return;
559 }
560 Err(error) => {
561 self.record(FindingKind::SegmentCorrupt, &file, None, error.to_string());
562 return;
563 }
564 };
565 if !reader.meta().same_segment_file(expected) {
566 self.record(
567 FindingKind::SegmentMismatch,
568 &file,
569 None,
570 format!(
571 "manifest records {expected:?}, segment header records {:?}",
572 reader.meta()
573 ),
574 );
575 return;
576 }
577 let mut regions_valid = true;
578 for entry in reader.directory() {
579 if let Err(error) = reader.region_by_id(entry.kind) {
580 regions_valid = false;
581 self.record(
582 FindingKind::SegmentRegionCorrupt,
583 &file,
584 Some(entry.offset),
585 format!("{}: {error}", region_name(entry.kind)),
586 );
587 }
588 }
589 if !regions_valid {
590 return;
591 }
592 if let Err(error) = reader.validate_all() {
593 self.record(FindingKind::SegmentCorrupt, &file, None, error.to_string());
594 return;
595 }
596 if let Err(error) = decode_regions(&reader) {
597 self.record(
598 FindingKind::SegmentIndexInvalid,
599 &file,
600 None,
601 error.to_string(),
602 );
603 }
604 }
605
606 fn wal(
607 &mut self,
608 manifest: &ManifestState,
609 has_segment_files: bool,
610 ) -> Result<(), VerifyError> {
611 let decoded = manifest.decoded();
612 let absent = matches!(manifest, ManifestState::Absent);
613 let absorbed_through = decoded.map_or(0, |manifest| manifest.log_seq);
614 #[cfg(feature = "graph-cypher")]
615 let covered_through = decoded
616 .and_then(|manifest| manifest.graph.as_ref())
617 .map_or(absorbed_through, |graph| {
618 absorbed_through.max(graph.graph_absorbed_through)
619 });
620 #[cfg(not(feature = "graph-cypher"))]
621 let covered_through = absorbed_through;
622 let wal_path = self.directory.join(WAL_FILE);
623 match self.vfs.open(&wal_path) {
624 Ok(length) if length > 0 => {}
625 Err(error) if error.kind() != std::io::ErrorKind::NotFound => {
626 self.record(FindingKind::Unreadable, WAL_FILE, None, error.to_string());
627 return Ok(());
628 }
629 Ok(_) | Err(_) => {
630 if covered_through > 0 {
631 self.record(
632 FindingKind::WalMissing,
633 WAL_FILE,
634 None,
635 format!("the manifest covers WAL sequences through {absorbed_through}"),
636 );
637 } else if absent && has_segment_files {
638 self.record(
639 FindingKind::ManifestMissing,
640 MANIFEST_FILE,
641 None,
642 "segment files exist but neither a manifest nor a WAL holds their rows"
643 .to_owned(),
644 );
645 }
646 #[cfg(feature = "graph-cypher")]
647 if covered_through == 0
648 && let Some(manifest) = decoded
649 {
650 self.graph_replay(manifest, &[]);
651 }
652 return Ok(());
653 }
654 }
655 let reader = match WalReader::open(self.vfs, &wal_path) {
656 Ok(reader) => reader,
657 Err(WalReadError::Io(error)) => {
658 self.record(FindingKind::Unreadable, WAL_FILE, None, error.to_string());
659 return Ok(());
660 }
661 Err(error) => {
662 self.record(
663 FindingKind::WalHeaderCorrupt,
664 WAL_FILE,
665 Some(0),
666 error.to_string(),
667 );
668 return Ok(());
669 }
670 };
671 self.report.wal_records_checked = u64::try_from(reader.records().len()).unwrap_or(u64::MAX);
672 match reader.terminator() {
673 Some(ReplayTerminator::CleanEnd) | None => {}
674 Some(ReplayTerminator::InvalidHeader(error)) => {
675 self.record(
676 FindingKind::WalHeaderCorrupt,
677 WAL_FILE,
678 Some(0),
679 error.to_string(),
680 );
681 return Ok(());
682 }
683 Some(ReplayTerminator::CorruptAt { offset, reason }) => {
684 self.record(
685 FindingKind::WalRecordCorrupt,
686 WAL_FILE,
687 Some(u64::try_from(offset).unwrap_or(u64::MAX)),
688 format!("{reason:?}"),
689 );
690 return Ok(());
691 }
692 }
693 let durable_end = reader.durable_end();
694 let first_seq = reader
695 .records()
696 .first()
697 .map_or(durable_end.saturating_add(1), |record| record.seq.get());
698 #[cfg(feature = "graph-cypher")]
699 let graph_objects = decoded
700 .filter(|manifest| manifest.graph.is_some())
701 .map(|manifest| {
702 crate::lifecycle::native_graph::recovery::snapshot_graph_objects(
703 self.vfs,
704 self.directory,
705 manifest,
706 reader.records(),
707 )
708 });
709 drop(reader);
710 if durable_end < covered_through {
711 self.record(
712 FindingKind::ManifestAheadOfWal,
713 MANIFEST_FILE,
714 None,
715 format!(
716 "the manifest covers WAL sequences through {covered_through}, \
717 the WAL ends at {durable_end}"
718 ),
719 );
720 return Ok(());
721 }
722 if absent && first_seq > 1 {
723 self.record(
724 FindingKind::ManifestMissing,
725 MANIFEST_FILE,
726 None,
727 format!(
728 "the WAL starts at sequence {first_seq}; earlier sequences were \
729 absorbed by a snapshot that is gone"
730 ),
731 );
732 return Ok(());
733 }
734 if matches!(manifest, ManifestState::Damaged) {
735 return Ok(());
737 }
738 #[cfg(feature = "graph-cypher")]
739 if let Some(census) = graph_objects {
740 match census {
741 Ok(mut objects) => {
742 if let Some(graph) = decoded.and_then(|manifest| manifest.graph.as_ref()) {
743 let known: std::collections::BTreeSet<_> =
744 graph.objects.iter().map(|object| object.artifact).collect();
745 objects.retain(|object| !known.contains(&object.artifact));
746 if let Ok(state) = graph.state() {
747 self.graph_objects(&objects, state.store);
748 }
749 }
750 }
751 Err(error) => {
752 self.record(
753 FindingKind::WalRecordInvalid,
754 WAL_FILE,
755 None,
756 error.to_string(),
757 );
758 return Ok(());
759 }
760 }
761 }
762 let schema =
763 decoded.map_or_else(Schema::timestamp_only, |manifest| manifest.schema.clone());
764 let analyzer = Analyzer::new(TokenizerConfig::text_default())
765 .map_err(|error| VerifyError::Tokenizer(error.to_string()))?;
766 let accounting = Arc::new(Accounting::new(u64::MAX, u64::MAX));
767 let generation = decoded.map_or(0, |manifest| manifest.generation);
768 let recovered = crate::ingest::ActiveState::recover(
769 self.vfs,
770 &wal_path,
771 generation,
772 absorbed_through,
773 {
774 #[cfg(feature = "graph-cypher")]
775 {
776 decoded
777 .and_then(|manifest| manifest.graph.as_ref())
778 .map_or(absorbed_through, |graph| graph.graph_absorbed_through)
779 }
780 #[cfg(not(feature = "graph-cypher"))]
781 {
782 absorbed_through
783 }
784 },
785 #[cfg(feature = "graph-cypher")]
786 decoded.and_then(|manifest| manifest.graph.as_ref()),
787 #[cfg(feature = "graph-cypher")]
788 decoded.is_some_and(|manifest| manifest.graph.is_some()),
789 &accounting,
790 &schema,
791 &analyzer,
792 None,
793 );
794 match recovered {
795 Ok(_recovered) =>
796 {
797 #[cfg(feature = "graph-cypher")]
798 if let Some(manifest) = decoded {
799 self.graph_replay(manifest, &_recovered.graph);
800 }
801 }
802 Err(error) => {
803 if absent && has_segment_files {
804 self.record(
805 FindingKind::ManifestMissing,
806 MANIFEST_FILE,
807 None,
808 format!(
809 "the WAL cannot be replayed without the manifest's schema: {error}"
810 ),
811 );
812 } else {
813 self.record(
814 FindingKind::WalRecordInvalid,
815 WAL_FILE,
816 None,
817 error.to_string(),
818 );
819 }
820 }
821 }
822 Ok(())
823 }
824}
825
826fn decode_regions(reader: &SegmentReader) -> Result<(), SegmentError> {
830 reader.columns()?;
831 reader.alive()?;
832 reader.postings()?;
833 reader.stored_text()?;
834 reader.stored_metadata()?;
835 if reader
836 .directory()
837 .iter()
838 .any(|entry| entry.kind == RegionKind::GraphNodeBlocks.id())
839 {
840 reader.graph_node_blocks()?;
841 }
842 Ok(())
843}
844
845fn region_name(kind: u16) -> String {
846 RegionKind::from_id(kind).map_or_else(|| format!("region {kind}"), |kind| format!("{kind:?}"))
847}
848
849fn is_segment_file(path: &Path) -> bool {
850 path.file_name()
851 .and_then(|name| name.to_str())
852 .is_some_and(|name| name.starts_with("segment-") && name.ends_with(".zseg"))
853}