zeppelin_embed/
verify.rs

1//! Read-only end-to-end store verification.
2//!
3//! [`verify_store`](crate::verify::verify_store) walks one store directory with the decoders the engine
4//! already uses to open it: the manifest frame, every referenced segment's
5//! header, region checksums, file trailer and region decoders, and the WAL
6//! prefix through the same replay that recovery runs. It reports every damaged
7//! artifact it finds instead of stopping at the first, and it never writes:
8//! the walk calls only the read half of [`Vfs`](crate::vfs::Vfs) (`open`, `read`,
9//! `open_for_map` and `list`).
10//!
11//! A finding is damage, never a notice. Recovery refuses a WAL that does not
12//! end cleanly, so a torn WAL tail is reported like any other WAL corruption.
13//! Unreferenced segment files and leftover temporary files are not findings,
14//! because a writable open removes them without losing data.
15//!
16//! The walk reads a point-in-time view of the directory. Verifying a store
17//! while another process writes to it can report a write that is in flight.
18
19use 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
36/// Canonical WAL filename inside a store directory.
37const WAL_FILE: &str = "wal.ze";
38
39/// What is damaged. Values are stable; new kinds are only appended.
40#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
41#[non_exhaustive]
42pub enum FindingKind {
43    /// `manifest.ze` is absent, but the WAL or segment files prove that a
44    /// committed snapshot existed and data it covered is now unreachable.
45    ManifestMissing,
46    /// The manifest frame, checksum, or payload failed to decode.
47    ManifestCorrupt,
48    /// The manifest covers WAL sequences that the WAL does not hold.
49    ManifestAheadOfWal,
50    /// A segment the manifest references does not exist.
51    SegmentMissing,
52    /// A segment header, length, identity, or file trailer failed validation.
53    SegmentCorrupt,
54    /// A segment header disagrees with the manifest's record of it.
55    SegmentMismatch,
56    /// A segment region's checksum does not match its bytes.
57    SegmentRegionCorrupt,
58    /// A checksum-valid region failed its decoder or cross-structure checks.
59    SegmentIndexInvalid,
60    /// The WAL is absent although the manifest covers WAL sequences.
61    WalMissing,
62    /// The WAL file header is truncated or invalid.
63    WalHeaderCorrupt,
64    /// A WAL record failed framing, checksum, or sequence validation.
65    WalRecordCorrupt,
66    /// A checksum-valid WAL record cannot be replayed into the store.
67    WalRecordInvalid,
68    /// A store file exists but could not be read.
69    Unreadable,
70    /// The pending purge intent `purge.ze` failed its frame or decoder.
71    PurgeIntentCorrupt,
72}
73
74impl FindingKind {
75    /// Stable lower-camel-case name used by the bindings.
76    #[must_use]
77    pub const fn as_str(self) -> &'static str {
78        match self {
79            Self::ManifestMissing => "manifestMissing",
80            Self::ManifestCorrupt => "manifestCorrupt",
81            Self::ManifestAheadOfWal => "manifestAheadOfWal",
82            Self::SegmentMissing => "segmentMissing",
83            Self::SegmentCorrupt => "segmentCorrupt",
84            Self::SegmentMismatch => "segmentMismatch",
85            Self::SegmentRegionCorrupt => "segmentRegionCorrupt",
86            Self::SegmentIndexInvalid => "segmentIndexInvalid",
87            Self::WalMissing => "walMissing",
88            Self::WalHeaderCorrupt => "walHeaderCorrupt",
89            Self::WalRecordCorrupt => "walRecordCorrupt",
90            Self::WalRecordInvalid => "walRecordInvalid",
91            Self::Unreadable => "unreadable",
92            Self::PurgeIntentCorrupt => "purgeIntentCorrupt",
93        }
94    }
95}
96
97/// One damaged artifact.
98#[derive(Clone, Debug, Eq, PartialEq)]
99pub struct Finding {
100    /// What is damaged.
101    pub kind: FindingKind,
102    /// File name relative to the store directory.
103    pub file: String,
104    /// Byte offset of the damage inside `file`, when the decoder knows it.
105    pub offset: Option<u64>,
106    /// Human-readable decoder detail.
107    pub detail: String,
108}
109
110/// Result of one verification walk.
111#[derive(Clone, Debug, Default, Eq, PartialEq)]
112pub struct VerifyReport {
113    /// Every damaged artifact, in walk order: manifest, segments, WAL.
114    pub findings: Vec<Finding>,
115    /// Generation of the decoded manifest, or zero without one.
116    pub generation: u64,
117    /// Segments the manifest references, whether or not they were readable.
118    pub segments_checked: u64,
119    /// WAL records that passed checksum and sequence validation.
120    pub wal_records_checked: u64,
121}
122
123impl VerifyReport {
124    /// True when no damage was found.
125    #[must_use]
126    pub fn is_clean(&self) -> bool {
127        self.findings.is_empty()
128    }
129}
130
131/// Why a walk could not start. Damage inside the store is never an error; it
132/// is a [`Finding`].
133#[derive(Debug)]
134pub enum VerifyError {
135    /// The store path does not exist.
136    NotFound {
137        /// Path supplied by the caller.
138        path: PathBuf,
139    },
140    /// The store path exists but is not a directory.
141    NotDirectory {
142        /// Path supplied by the caller.
143        path: PathBuf,
144    },
145    /// The directory could not be inspected or listed.
146    Io {
147        /// Path involved in the operation.
148        path: PathBuf,
149        /// Underlying platform error.
150        source: std::io::Error,
151    },
152    /// The default tokenizer could not be built for WAL replay.
153    Tokenizer(String),
154}
155
156impl std::fmt::Display for VerifyError {
157    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
158        match self {
159            Self::NotFound { path } => {
160                write!(formatter, "store path {} does not exist", path.display())
161            }
162            Self::NotDirectory { path } => {
163                write!(
164                    formatter,
165                    "store path {} is not a directory",
166                    path.display()
167                )
168            }
169            Self::Io { path, source } => {
170                write!(formatter, "store I/O {}: {source}", path.display())
171            }
172            Self::Tokenizer(detail) => write!(formatter, "verify tokenizer: {detail}"),
173        }
174    }
175}
176
177impl std::error::Error for VerifyError {
178    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
179        match self {
180            Self::Io { source, .. } => Some(source),
181            Self::NotFound { .. } | Self::NotDirectory { .. } | Self::Tokenizer(_) => None,
182        }
183    }
184}
185
186/// Verifies the store directory at `path` on the platform filesystem.
187pub fn verify_store(path: impl AsRef<Path>) -> Result<VerifyReport, VerifyError> {
188    verify_store_on_vfs(&StdVfs, path.as_ref())
189}
190
191/// Verifies the store directory at `path` through `vfs`, using only its read
192/// operations.
193pub fn verify_store_on_vfs(vfs: &dyn Vfs, path: &Path) -> Result<VerifyReport, VerifyError> {
194    match vfs.ensure_directory(path, false) {
195        Ok(true) => {}
196        Ok(false) => {
197            return Err(VerifyError::NotDirectory {
198                path: path.to_path_buf(),
199            });
200        }
201        Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
202            return Err(VerifyError::NotFound {
203                path: path.to_path_buf(),
204            });
205        }
206        Err(source) => {
207            return Err(VerifyError::Io {
208                path: path.to_path_buf(),
209                source,
210            });
211        }
212    }
213    let entries = vfs.list(path).map_err(|source| VerifyError::Io {
214        path: path.to_path_buf(),
215        source,
216    })?;
217    let has_segment_files = entries.iter().any(|entry| is_segment_file(entry));
218    let mut walk = Walk {
219        vfs,
220        directory: path,
221        report: VerifyReport::default(),
222    };
223    let manifest = walk.manifest();
224    if let Some(decoded) = manifest.decoded() {
225        walk.report.generation = decoded.generation;
226        walk.segments(decoded);
227    }
228    walk.wal(&manifest, has_segment_files)?;
229    walk.purge_intent();
230    Ok(walk.report)
231}
232
233struct Walk<'a> {
234    vfs: &'a dyn Vfs,
235    directory: &'a Path,
236    report: VerifyReport,
237}
238
239/// What the manifest walk established.
240enum ManifestState {
241    /// No `manifest.ze`: the store never committed a snapshot, or lost it.
242    Absent,
243    /// A manifest exists but is unreadable or corrupt; already reported.
244    Damaged,
245    /// A decoded manifest.
246    Decoded(Manifest),
247}
248
249impl ManifestState {
250    fn decoded(&self) -> Option<&Manifest> {
251        match self {
252            Self::Decoded(manifest) => Some(manifest),
253            Self::Absent | Self::Damaged => None,
254        }
255    }
256}
257
258impl Walk<'_> {
259    fn record(&mut self, kind: FindingKind, file: &str, offset: Option<u64>, detail: String) {
260        self.report.findings.push(Finding {
261            kind,
262            file: file.to_owned(),
263            offset,
264            detail,
265        });
266    }
267
268    fn manifest(&mut self) -> ManifestState {
269        let bytes = match self.vfs.read(&self.directory.join(MANIFEST_FILE)) {
270            Ok(bytes) => bytes,
271            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
272                return ManifestState::Absent;
273            }
274            Err(error) => {
275                self.record(
276                    FindingKind::Unreadable,
277                    MANIFEST_FILE,
278                    None,
279                    error.to_string(),
280                );
281                return ManifestState::Damaged;
282            }
283        };
284        match decode_manifest(MANIFEST_FILE, &bytes) {
285            Ok(manifest) => ManifestState::Decoded(manifest),
286            Err(error) => {
287                self.record(
288                    FindingKind::ManifestCorrupt,
289                    MANIFEST_FILE,
290                    None,
291                    error.to_string(),
292                );
293                ManifestState::Damaged
294            }
295        }
296    }
297
298    fn segments(&mut self, manifest: &Manifest) {
299        // Snapshot readers decode columns against the manifest schema, so an
300        // older segment reads an added nullable attribute as null and a
301        // conflicting column fails; verify decodes the same way.
302        let schema = Arc::new(manifest.schema.clone());
303        for expected in &manifest.segments {
304            self.report.segments_checked = self.report.segments_checked.saturating_add(1);
305            self.segment(expected, &schema);
306        }
307    }
308
309    /// A pending intent is a purge the next writable open completes, so only
310    /// an intent that fails to decode is damage.
311    fn purge_intent(&mut self) {
312        let path = self.directory.join(PURGE_INTENT_FILE);
313        match self.vfs.open(&path) {
314            Ok(_) => {}
315            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return,
316            Err(error) => {
317                self.record(
318                    FindingKind::Unreadable,
319                    PURGE_INTENT_FILE,
320                    None,
321                    error.to_string(),
322                );
323                return;
324            }
325        }
326        if let Err(error) = read_intent(self.vfs, self.directory) {
327            self.record(
328                FindingKind::PurgeIntentCorrupt,
329                PURGE_INTENT_FILE,
330                None,
331                error.to_string(),
332            );
333        }
334    }
335
336    fn segment(&mut self, expected: &SegmentMeta, schema: &Arc<Schema>) {
337        let file = expected.id.file_name();
338        let path = self.directory.join(&file);
339        let reader = match SegmentReader::open(self.vfs, &path, expected.id) {
340            Ok(reader) => reader.with_collection_schema(Arc::clone(schema)),
341            Err(SegmentError::Io { source, .. })
342                if source.kind() == std::io::ErrorKind::NotFound =>
343            {
344                self.record(
345                    FindingKind::SegmentMissing,
346                    &file,
347                    None,
348                    format!(
349                        "the manifest references segment {} with {} rows",
350                        expected.id, expected.row_count
351                    ),
352                );
353                return;
354            }
355            Err(SegmentError::Io { source, .. }) => {
356                self.record(FindingKind::Unreadable, &file, None, source.to_string());
357                return;
358            }
359            Err(error) => {
360                self.record(FindingKind::SegmentCorrupt, &file, None, error.to_string());
361                return;
362            }
363        };
364        if !reader.meta().same_segment_file(expected) {
365            self.record(
366                FindingKind::SegmentMismatch,
367                &file,
368                None,
369                format!(
370                    "manifest records {expected:?}, segment header records {:?}",
371                    reader.meta()
372                ),
373            );
374            return;
375        }
376        let mut regions_valid = true;
377        for entry in reader.directory() {
378            if let Err(error) = reader.region_by_id(entry.kind) {
379                regions_valid = false;
380                self.record(
381                    FindingKind::SegmentRegionCorrupt,
382                    &file,
383                    Some(entry.offset),
384                    format!("{}: {error}", region_name(entry.kind)),
385                );
386            }
387        }
388        if !regions_valid {
389            return;
390        }
391        if let Err(error) = reader.validate_all() {
392            self.record(FindingKind::SegmentCorrupt, &file, None, error.to_string());
393            return;
394        }
395        if let Err(error) = decode_regions(&reader) {
396            self.record(
397                FindingKind::SegmentIndexInvalid,
398                &file,
399                None,
400                error.to_string(),
401            );
402        }
403    }
404
405    fn wal(
406        &mut self,
407        manifest: &ManifestState,
408        has_segment_files: bool,
409    ) -> Result<(), VerifyError> {
410        let decoded = manifest.decoded();
411        let absent = matches!(manifest, ManifestState::Absent);
412        let absorbed_through = decoded.map_or(0, |manifest| manifest.log_seq);
413        let wal_path = self.directory.join(WAL_FILE);
414        match self.vfs.open(&wal_path) {
415            Ok(length) if length > 0 => {}
416            Err(error) if error.kind() != std::io::ErrorKind::NotFound => {
417                self.record(FindingKind::Unreadable, WAL_FILE, None, error.to_string());
418                return Ok(());
419            }
420            Ok(_) | Err(_) => {
421                if absorbed_through > 0 {
422                    self.record(
423                        FindingKind::WalMissing,
424                        WAL_FILE,
425                        None,
426                        format!("the manifest covers WAL sequences through {absorbed_through}"),
427                    );
428                } else if absent && has_segment_files {
429                    self.record(
430                        FindingKind::ManifestMissing,
431                        MANIFEST_FILE,
432                        None,
433                        "segment files exist but neither a manifest nor a WAL holds their rows"
434                            .to_owned(),
435                    );
436                }
437                return Ok(());
438            }
439        }
440        let reader = match WalReader::open(self.vfs, &wal_path) {
441            Ok(reader) => reader,
442            Err(WalReadError::Io(error)) => {
443                self.record(FindingKind::Unreadable, WAL_FILE, None, error.to_string());
444                return Ok(());
445            }
446            Err(error) => {
447                self.record(
448                    FindingKind::WalHeaderCorrupt,
449                    WAL_FILE,
450                    Some(0),
451                    error.to_string(),
452                );
453                return Ok(());
454            }
455        };
456        self.report.wal_records_checked = u64::try_from(reader.records().len()).unwrap_or(u64::MAX);
457        match reader.terminator() {
458            Some(ReplayTerminator::CleanEnd) | None => {}
459            Some(ReplayTerminator::InvalidHeader(error)) => {
460                self.record(
461                    FindingKind::WalHeaderCorrupt,
462                    WAL_FILE,
463                    Some(0),
464                    error.to_string(),
465                );
466                return Ok(());
467            }
468            Some(ReplayTerminator::CorruptAt { offset, reason }) => {
469                self.record(
470                    FindingKind::WalRecordCorrupt,
471                    WAL_FILE,
472                    Some(u64::try_from(offset).unwrap_or(u64::MAX)),
473                    format!("{reason:?}"),
474                );
475                return Ok(());
476            }
477        }
478        let durable_end = reader.durable_end();
479        let first_seq = reader
480            .records()
481            .first()
482            .map_or(durable_end.saturating_add(1), |record| record.seq.get());
483        drop(reader);
484        if durable_end < absorbed_through {
485            self.record(
486                FindingKind::ManifestAheadOfWal,
487                MANIFEST_FILE,
488                None,
489                format!(
490                    "the manifest covers WAL sequences through {absorbed_through}, \
491                     the WAL ends at {durable_end}"
492                ),
493            );
494            return Ok(());
495        }
496        if absent && first_seq > 1 {
497            self.record(
498                FindingKind::ManifestMissing,
499                MANIFEST_FILE,
500                None,
501                format!(
502                    "the WAL starts at sequence {first_seq}; earlier sequences were \
503                     absorbed by a snapshot that is gone"
504                ),
505            );
506            return Ok(());
507        }
508        if matches!(manifest, ManifestState::Damaged) {
509            // Replay needs the manifest's schema and absorbed boundary.
510            return Ok(());
511        }
512        let schema =
513            decoded.map_or_else(Schema::timestamp_only, |manifest| manifest.schema.clone());
514        let analyzer = Analyzer::new(TokenizerConfig::text_default())
515            .map_err(|error| VerifyError::Tokenizer(error.to_string()))?;
516        let accounting = Arc::new(Accounting::new(u64::MAX, u64::MAX));
517        let generation = decoded.map_or(0, |manifest| manifest.generation);
518        if let Err(error) = crate::ingest::ActiveState::recover(
519            self.vfs,
520            &wal_path,
521            generation,
522            absorbed_through,
523            &accounting,
524            &schema,
525            &analyzer,
526        ) {
527            if absent && has_segment_files {
528                self.record(
529                    FindingKind::ManifestMissing,
530                    MANIFEST_FILE,
531                    None,
532                    format!("the WAL cannot be replayed without the manifest's schema: {error}"),
533                );
534            } else {
535                self.record(
536                    FindingKind::WalRecordInvalid,
537                    WAL_FILE,
538                    None,
539                    error.to_string(),
540                );
541            }
542        }
543        Ok(())
544    }
545}
546
547/// Runs every typed decoder whose region is present. Each decoder also checks
548/// its row count against the segment header, which is the cross-structure
549/// index consistency the engine relies on at query time.
550fn decode_regions(reader: &SegmentReader) -> Result<(), SegmentError> {
551    reader.columns()?;
552    reader.alive()?;
553    reader.postings()?;
554    reader.stored_text()?;
555    reader.stored_metadata()?;
556    if reader
557        .directory()
558        .iter()
559        .any(|entry| entry.kind == RegionKind::GraphNodeBlocks.id())
560    {
561        reader.graph_node_blocks()?;
562    }
563    Ok(())
564}
565
566fn region_name(kind: u16) -> String {
567    RegionKind::from_id(kind).map_or_else(|| format!("region {kind}"), |kind| format!("{kind:?}"))
568}
569
570fn is_segment_file(path: &Path) -> bool {
571    path.file_name()
572        .and_then(|name| name.to_str())
573        .is_some_and(|name| name.starts_with("segment-") && name.ends_with(".zseg"))
574}
zeppelin_embed/verify.rs

zeppelin_embed/
verify.rs
