writerslogic_warm/core/format/
verify.rs

1// Copyright 2024-2026 WritersLogic Contributors
2// SPDX-License-Identifier: Apache-2.0
3//! Offline verification of a store directory. Read-only: it never opens the engine unless
4//! `deep` is set, and then only on a private copy, because opening a store writes to it
5//! (end marker, `format.json`, `documents.text`, the lock file).
6
7use std::collections::BTreeSet;
8use std::fs::File;
9use std::path::{Path, PathBuf};
10
11use anyhow::{bail, Context, Result};
12use memmap2::Mmap;
13use serde_json::Value;
14
15use super::scan::{self, PayloadKind, Replay, SegmentEnd, TextHeader};
16use super::{
17    Report, Severity, TenantReport, CURRENT, SALT_LEN, STORAGE_FORMAT_CURRENT,
18    STORAGE_FORMAT_OLDEST, STORE_SCHEMA_OLDEST, STORE_SCHEMA_VERSION,
19};
20use crate::core::storage::{
21    list_segments, FormatManifest, FORMAT_MAGIC, LEGACY_SEGMENT_SIZE, SEGMENT_SIZE,
22};
23
24/// Directory of the arena inside a store.
25const ARENA_DIRECTORY: &str = "vectors_data.bin";
26
27/// What `verify_path` does beyond the default offline checks.
28#[derive(Debug, Clone, Default)]
29pub struct VerifyOptions {
30    /// Environment variable holding the passphrase of an encrypted store. Without it the
31    /// frames of an encrypted store are checked for framing and CRC32 only.
32    pub passphrase_env: Option<String>,
33    /// Also open a private copy of the store with the engine and compare what it loaded
34    /// with what the journal says. Needs the disk space of one copy.
35    pub deep: bool,
36}
37
38/// Verifies a store directory, or a server data directory (`tenants/<hex name>/` stores).
39/// Damage is reported as findings; only an unreadable path is an `Err`.
40pub fn verify_path(path: &Path, options: &VerifyOptions) -> Result<Report> {
41    let metadata =
42        std::fs::metadata(path).with_context(|| format!("reading {}", path.display()))?;
43    if !metadata.is_dir() {
44        bail!("{} is not a directory", path.display());
45    }
46    let mut report = Report::new(path);
47    let is_store = path.join("store.json").exists() || path.join(ARENA_DIRECTORY).exists();
48    let tenants = path.join("tenants");
49    if !is_store && tenants.is_dir() {
50        let mut names: Vec<_> = std::fs::read_dir(&tenants)?
51            .collect::<std::io::Result<Vec<_>>>()?
52            .into_iter()
53            .filter(|e| e.path().is_dir())
54            .map(|e| e.file_name().to_string_lossy().into_owned())
55            .collect();
56        names.sort();
57        for dir in names {
58            let name = decode_tenant(&dir);
59            if name.is_none() {
60                report.add(
61                    Severity::Info,
62                    "tenant-name",
63                    format!("tenants/{dir}"),
64                    None,
65                    "directory name is not the hex encoding of a UTF-8 tenant name",
66                );
67            }
68            let sub = verify_store(&tenants.join(&dir), options)?;
69            report.tenants.push(TenantReport {
70                name: name.unwrap_or(dir),
71                report: sub,
72            });
73        }
74        if report.tenants.is_empty() {
75            report.add(
76                Severity::Info,
77                "no-tenants",
78                "tenants",
79                None,
80                "no tenant stores",
81            );
82        }
83        report.checks.push("data-directory");
84        report.finish();
85        return Ok(report);
86    }
87    if !is_store {
88        report.add(
89            Severity::Error,
90            "not-a-store",
91            ".",
92            None,
93            "neither store.json, vectors_data.bin nor a tenants directory is present",
94        );
95        report.finish();
96        return Ok(report);
97    }
98    verify_store(path, options)
99}
100
101fn decode_tenant(dir: &str) -> Option<String> {
102    if dir.is_empty() || !dir.len().is_multiple_of(2) || !dir.bytes().all(|b| b.is_ascii_hexdigit())
103    {
104        return None;
105    }
106    let bytes: Option<Vec<u8>> = (0..dir.len() / 2)
107        .map(|i| u8::from_str_radix(&dir[2 * i..2 * i + 2], 16).ok())
108        .collect();
109    String::from_utf8(bytes?).ok()
110}
111
112/// What the verifier needs to know about the store after reading `store.json`.
113struct Schema {
114    dimensions: Option<u64>,
115    encrypted: bool,
116    meaning_enabled: bool,
117    raw: Value,
118}
119
120fn verify_store(store: &Path, options: &VerifyOptions) -> Result<Report> {
121    let mut report = Report::new(store);
122    let schema = check_schema(store, &mut report);
123    let arena = check_arena(store, &schema, options, &mut report)?;
124    check_text_segment(store, &schema, &mut report);
125    check_indices(store, &arena, &mut report);
126    check_ancillary(store, &schema, &mut report);
127    if options.deep && report.findings.iter().all(|f| f.severity < Severity::Error) {
128        if let Some(replay) = &arena.replay {
129            if let Err(error) = deep_check(store, &schema, replay, options, &mut report) {
130                report.add(
131                    Severity::Error,
132                    "deep-open",
133                    ".",
134                    None,
135                    format!("engine cross-check failed: {error:#}"),
136                );
137            }
138        } else {
139            report.add(
140                Severity::Info,
141                "deep-skipped",
142                ".",
143                None,
144                "payloads were not authenticated, so there is no journal state to cross-check",
145            );
146        }
147    }
148    report.finish();
149    Ok(report)
150}
151
152fn check_schema(store: &Path, report: &mut Report) -> Schema {
153    let mut schema = Schema {
154        dimensions: None,
155        encrypted: false,
156        meaning_enabled: false,
157        raw: Value::Null,
158    };
159    let path = store.join("store.json");
160    if !path.exists() {
161        if store.join(ARENA_DIRECTORY).exists() {
162            report.add(
163                Severity::Unsupported,
164                "pre-schema-store",
165                "store.json",
166                None,
167                "no store.json: written before stores recorded their embedding schema. The \
168                 engine refuses to open it; rebuild with `warm-admin reencode` from the source \
169                 documents",
170            );
171        } else {
172            report.add(
173                Severity::Error,
174                "missing-file",
175                "store.json",
176                None,
177                "store.json is missing",
178            );
179        }
180        return schema;
181    }
182    let bytes = match crate::core::durable_file::read_manifest(&path) {
183        Ok((bytes, verified)) => {
184            if verified {
185                report.checks.push("store.json checksum");
186            } else {
187                report.add(
188                    Severity::Info,
189                    "no-checksum-sidecar",
190                    "store.json",
191                    None,
192                    "no store.json.crc32 (written by a release before checksum sidecars)",
193                );
194            }
195            bytes
196        }
197        Err(error) => {
198            report.add(
199                Severity::Error,
200                "manifest-checksum",
201                "store.json",
202                None,
203                format!("{error:#}"),
204            );
205            let Ok(bytes) = std::fs::read(&path) else {
206                return schema;
207            };
208            bytes
209        }
210    };
211    let value: Value = match serde_json::from_slice(&bytes) {
212        Ok(value) => value,
213        Err(error) => {
214            report.add(
215                Severity::Error,
216                "schema-invalid",
217                "store.json",
218                None,
219                error.to_string(),
220            );
221            return schema;
222        }
223    };
224    let version = value.get("version").and_then(Value::as_u64);
225    report.versions.store_schema = version.map(|v| v as u32);
226    match version {
227        Some(v)
228            if (u64::from(STORE_SCHEMA_OLDEST)..=u64::from(STORE_SCHEMA_VERSION)).contains(&v) => {}
229        Some(v) if v > u64::from(STORE_SCHEMA_VERSION) => report.add(
230            Severity::Unsupported,
231            "schema-newer",
232            "store.json",
233            None,
234            format!(
235                "store.json version {v} is newer than the supported version {STORE_SCHEMA_VERSION}"
236            ),
237        ),
238        other => report.add(
239            Severity::Unsupported,
240            "schema-version",
241            "store.json",
242            None,
243            format!("store.json version {other:?} is not a version this build reads"),
244        ),
245    }
246    schema.dimensions = value.get("dimensions").and_then(Value::as_u64);
247    if schema.dimensions.is_none_or(|d| d == 0 || d > 1_000_000) {
248        report.add(
249            Severity::Error,
250            "schema-invalid",
251            "store.json",
252            None,
253            "dimensions is missing or outside 1..=1000000",
254        );
255    }
256    report.versions.dimensions = schema.dimensions;
257    schema.encrypted = value
258        .get("encrypted")
259        .and_then(Value::as_bool)
260        .unwrap_or(false);
261    schema.meaning_enabled = value
262        .get("meaning_enabled")
263        .and_then(Value::as_bool)
264        .unwrap_or(false);
265    report.versions.encrypted = Some(schema.encrypted);
266    if let Some(name) = value.get("text_encoder").and_then(Value::as_str) {
267        report.versions.text_encoder = Some(name.to_owned());
268        if name.parse::<crate::core::encoding::TextEncoder>().is_err() {
269            report.add(
270                Severity::Unsupported,
271                "unknown-encoder",
272                "store.json",
273                None,
274                format!("text encoder {name:?} is not known to this build"),
275            );
276        }
277    }
278    if let Some(name) = value.get("vector_encoder").and_then(Value::as_str) {
279        report.versions.vector_encoder = Some(name.to_owned());
280        if name != "signed-projection-v2" {
281            report.add(
282                Severity::Unsupported,
283                "unknown-encoder",
284                "store.json",
285                None,
286                format!("vector encoder {name:?} is not known to this build"),
287            );
288        }
289    }
290    if value.get("metric").and_then(Value::as_str) != Some("jaccard") {
291        report.add(
292            Severity::Unsupported,
293            "unknown-metric",
294            "store.json",
295            None,
296            "metric is not jaccard",
297        );
298    }
299    schema.raw = value;
300    schema
301}
302
303/// What the arena check learned that later checks need.
304struct ArenaState {
305    /// Name of the generation directory and journal length: the engine's checkpoint.
306    checkpoint: Option<(String, u64)>,
307    replay: Option<Replay>,
308}
309
310/// The directory holding the live segments of the arena at `arena` (the one `CURRENT` names).
311pub fn resolve_generation(arena: &Path) -> Result<PathBuf> {
312    crate::core::durable_file::arena_generation(arena)
313}
314
315#[cfg(feature = "security")]
316type Cipher = crate::core::security::EncryptionManager;
317#[cfg(not(feature = "security"))]
318type Cipher = std::convert::Infallible;
319
320fn open_cipher(
321    store: &Path,
322    schema: &Schema,
323    options: &VerifyOptions,
324    report: &mut Report,
325) -> Option<Cipher> {
326    if !schema.encrypted {
327        return None;
328    }
329    let salt = store.join("encryption.salt");
330    match std::fs::metadata(&salt) {
331        Ok(m) if m.len() == SALT_LEN as u64 => {}
332        Ok(m) => {
333            report.add(
334                Severity::Error,
335                "salt-invalid",
336                "encryption.salt",
337                None,
338                format!("expected {SALT_LEN} bytes, found {}", m.len()),
339            );
340            return None;
341        }
342        Err(_) => {
343            report.add(
344                Severity::Error,
345                "missing-file",
346                "encryption.salt",
347                None,
348                "the store is encrypted but encryption.salt is missing: the key cannot be derived",
349            );
350            return None;
351        }
352    }
353    let Some(variable) = &options.passphrase_env else {
354        report.add(
355            Severity::Info,
356            "payloads-not-authenticated",
357            ARENA_DIRECTORY,
358            None,
359            "encrypted store: frames were checked for framing and CRC32 only; pass \
360             --passphrase-env to authenticate and decode every payload",
361        );
362        return None;
363    };
364    #[cfg(feature = "security")]
365    {
366        let Ok(passphrase) = crate::compatibility::var(variable) else {
367            report.add(
368                Severity::Info,
369                "payloads-not-authenticated",
370                ARENA_DIRECTORY,
371                None,
372                format!("environment variable {variable} is not set"),
373            );
374            return None;
375        };
376        match Cipher::new(&passphrase, store) {
377            Ok(cipher) => Some(cipher),
378            Err(error) => {
379                report.add(
380                    Severity::Error,
381                    "salt-invalid",
382                    "encryption.salt",
383                    None,
384                    format!("{error:#}"),
385                );
386                None
387            }
388        }
389    }
390    #[cfg(not(feature = "security"))]
391    {
392        let _ = variable;
393        report.add(
394            Severity::Info,
395            "payloads-not-authenticated",
396            ARENA_DIRECTORY,
397            None,
398            "this build lacks the `security` feature, so encrypted payloads cannot be decrypted",
399        );
400        None
401    }
402}
403
404fn check_arena(
405    store: &Path,
406    schema: &Schema,
407    options: &VerifyOptions,
408    report: &mut Report,
409) -> Result<ArenaState> {
410    let mut state = ArenaState {
411        checkpoint: None,
412        replay: None,
413    };
414    let base = store.join(ARENA_DIRECTORY);
415    if !base.is_dir() {
416        if schema.raw != Value::Null {
417            report.add(
418                Severity::Error,
419                "missing-file",
420                ARENA_DIRECTORY,
421                None,
422                "arena directory is missing",
423            );
424        }
425        return Ok(state);
426    }
427    let generation = match crate::core::durable_file::arena_generation(&base) {
428        Ok(path) => path,
429        Err(error) => {
430            report.add(
431                Severity::Error,
432                "pointer-invalid",
433                "vectors_data.bin/CURRENT",
434                None,
435                format!("{error:#}"),
436            );
437            return Ok(state);
438        }
439    };
440    let generation_name = generation
441        .file_name()
442        .map(|n| n.to_string_lossy().into_owned())
443        .unwrap_or_default();
444    // Compaction staging directories and superseded generations are leftovers, not state.
445    for entry in std::fs::read_dir(&base)? {
446        let entry = entry?;
447        let name = entry.file_name().to_string_lossy().into_owned();
448        if name.starts_with("gen-") && entry.path() != generation {
449            report.add(
450                Severity::Info,
451                "orphan-generation",
452                format!("{ARENA_DIRECTORY}/{name}"),
453                None,
454                "a generation directory CURRENT does not name (an interrupted or superseded \
455                 compaction); safe to delete",
456            );
457        }
458    }
459
460    let manifest = check_manifest(&base, &generation, report);
461    let Some(manifest) = manifest else {
462        return Ok(state);
463    };
464    if report.versions.storage_format != Some(1) {
465        report.versions.storage_format = Some(manifest.version);
466    }
467    let segment_size = if manifest.version == 3 {
468        SEGMENT_SIZE
469    } else {
470        LEGACY_SEGMENT_SIZE
471    };
472    if manifest.segment_size != segment_size {
473        report.add(
474            Severity::Error,
475            "segment-size",
476            format!("{ARENA_DIRECTORY}/format.json"),
477            None,
478            format!(
479                "segment_size {} is not the {segment_size} that format {} requires",
480                manifest.segment_size, manifest.version
481            ),
482        );
483        return Ok(state);
484    }
485
486    let ids = list_segments(&generation)?;
487    if ids.is_empty() {
488        report.add(
489            Severity::Error,
490            "segment-gap",
491            ARENA_DIRECTORY,
492            None,
493            "the arena has no segments",
494        );
495        return Ok(state);
496    }
497    if ids.iter().enumerate().any(|(i, &id)| i != id) {
498        report.add(
499            Severity::Error,
500            "segment-gap",
501            ARENA_DIRECTORY,
502            None,
503            format!("segments are not numbered contiguously from 0: {ids:?}"),
504        );
505        return Ok(state);
506    }
507    report.segments = ids.len();
508
509    let cipher = open_cipher(store, schema, options, report);
510    let authenticated = !schema.encrypted || cipher.is_some();
511    let mut replay = Replay::default();
512    let mut transaction_formats = BTreeSet::new();
513    let mut payload_problems = 0usize;
514    let mut final_end = 0usize;
515    let mut zero_terminated = false;
516    for (n, &id) in ids.iter().enumerate() {
517        let is_final = n + 1 == ids.len();
518        let relative = format!("{ARENA_DIRECTORY}/seg_{id}.bin");
519        let path = generation.join(format!("seg_{id}.bin"));
520        let file = File::open(&path).with_context(|| format!("opening {}", path.display()))?;
521        let length = file.metadata()?.len();
522        check_segment_size(&relative, length, is_final, &manifest, report);
523        if length == 0 {
524            report.add(
525                Severity::Error,
526                "segment-size",
527                &relative,
528                None,
529                "segment file is empty",
530            );
531            continue;
532        }
533        // SAFETY: read-only mapping of a file this process does not modify; the verifier
534        // holds no reference into it past this loop iteration.
535        let map = unsafe { Mmap::map(&file)? };
536        let scan = scan::scan_segment(&map, is_final, |frame, payload| {
537            if payload_problems >= 20 {
538                return;
539            }
540            let offset = frame.offset as u64;
541            #[cfg(feature = "security")]
542            let plain: Vec<u8>;
543            let payload = if schema.encrypted {
544                #[cfg(feature = "security")]
545                match cipher.as_ref().map(|c| c.decrypt(payload)) {
546                    Some(Ok(bytes)) => {
547                        plain = bytes;
548                        &plain[..]
549                    }
550                    Some(Err(error)) => {
551                        payload_problems += 1;
552                        report.add(
553                            Severity::Error,
554                            "auth-failed",
555                            &relative,
556                            Some(offset),
557                            format!("{error:#} (wrong passphrase, or the frame was altered)"),
558                        );
559                        return;
560                    }
561                    None => return,
562                }
563                #[cfg(not(feature = "security"))]
564                {
565                    let _ = &cipher;
566                    return;
567                }
568            } else {
569                payload
570            };
571            match scan::classify_payload(payload) {
572                kind @ (PayloadKind::Transaction(_) | PayloadKind::BinaryTransaction(_)) => {
573                    transaction_formats.insert(
574                        if matches!(kind, PayloadKind::Transaction(_)) {
575                            "HMS-TXN-2"
576                        } else {
577                            "HMS-TXN-3"
578                        }
579                        .to_owned(),
580                    );
581                    if schema.encrypted && !authenticated {
582                        return;
583                    }
584                    let checked = scan::transaction_json(payload)
585                        .expect("a transaction payload")
586                        .and_then(|json| {
587                            scan::check_transaction_with_schema(
588                                &json,
589                                schema.dimensions,
590                                schema.raw["version"].as_u64().unwrap_or(0) as u32,
591                                &mut replay,
592                            )
593                        });
594                    match checked {
595                        Ok(_) => {}
596                        Err(message) => {
597                            payload_problems += 1;
598                            report.add(
599                                Severity::Error,
600                                "payload-structure",
601                                &relative,
602                                Some(offset),
603                                message,
604                            );
605                        }
606                    }
607                }
608                PayloadKind::OtherTransactionVersion(version) => {
609                    payload_problems += 1;
610                    let newer = version.parse::<u32>().ok().is_none_or(|v| v > 3);
611                    transaction_formats.insert(format!("HMS-TXN-{version}"));
612                    report.add(
613                        Severity::Unsupported,
614                        if newer {
615                            "transaction-newer"
616                        } else {
617                            "transaction-older"
618                        },
619                        &relative,
620                        Some(offset),
621                        format!(
622                            "journal frame format HMS-TXN-{version} is {} than the supported \
623                             HMS-TXN-2 and HMS-TXN-3",
624                            if newer { "newer" } else { "older" }
625                        ),
626                    );
627                }
628                PayloadKind::Unrecognized => {
629                    payload_problems += 1;
630                    report.add(
631                        Severity::Error,
632                        "payload-unrecognized",
633                        &relative,
634                        Some(offset),
635                        if schema.encrypted {
636                            "decrypted payload has no journal magic"
637                        } else {
638                            "payload has no journal magic (an encrypted frame in a store marked \
639                             unencrypted, or damage)"
640                        },
641                    );
642                }
643            }
644        });
645        report.frames += scan.frames.len();
646        match &scan.end {
647            SegmentEnd::Marker | SegmentEnd::Exhausted => {}
648            SegmentEnd::ZeroHeader => zero_terminated = true,
649            SegmentEnd::TornTail { reason } => report.add(
650                Severity::Warning,
651                "torn-tail",
652                &relative,
653                Some(scan.end_offset as u64),
654                format!(
655                    "the last append did not complete ({reason}); the engine truncates the log \
656                     to offset {} on open",
657                    scan.end_offset
658                ),
659            ),
660            SegmentEnd::Corrupt {
661                reason,
662                committed_at,
663            } => report.add(
664                Severity::Error,
665                "frame-corrupt",
666                &relative,
667                Some(scan.end_offset as u64),
668                match committed_at {
669                    Some(next) => format!(
670                        "frame fails its checks ({reason}) but committed data follows at \
671                         offset {next}: damage inside committed data"
672                    ),
673                    None => format!(
674                        "frame fails its checks ({reason}) inside a segment that was complete"
675                    ),
676                },
677            ),
678        }
679        if is_final {
680            final_end = scan.end_offset;
681        }
682    }
683    if zero_terminated {
684        report.add(
685            Severity::Info,
686            "legacy-zero-terminator",
687            ARENA_DIRECTORY,
688            None,
689            "a segment ends in an all-zero header, the form written before end markers; the \
690             engine rewrites it as an end marker on open",
691        );
692    }
693    report.versions.transaction_formats = transaction_formats.into_iter().collect();
694    report.checks.push("frames");
695    state.checkpoint = Some((
696        generation_name,
697        ((ids.len() - 1) as u64) * manifest.segment_size as u64 + final_end as u64,
698    ));
699    report.payloads_authenticated = authenticated;
700    if authenticated {
701        report.mutations = replay.mutations;
702        report.live_documents = Some(replay.documents.len());
703        report.live_vectors = Some(replay.vector_total());
704        report.memory_stamped_documents =
705            Some(replay.documents.values().filter(|d| d.stamped).count());
706        if payload_problems == 0 {
707            report.checks.push("payloads");
708        }
709        state.replay = Some(replay);
710    }
711    Ok(state)
712}
713
714fn check_manifest(base: &Path, generation: &Path, report: &mut Report) -> Option<FormatManifest> {
715    let top = base.join("format.json");
716    let path = if top.exists() {
717        top
718    } else {
719        generation.join("format.json")
720    };
721    let relative = format!("{ARENA_DIRECTORY}/format.json");
722    if !path.exists() {
723        if generation.join("seg_0.bin").exists() {
724            report.versions.storage_format = Some(1);
725            report.add(
726                Severity::Info,
727                "pre-manifest-store",
728                &relative,
729                None,
730                "no format.json: a format-1 store; the engine reads it as format 2 and writes \
731                 the manifest on open",
732            );
733            return Some(FormatManifest {
734                magic: FORMAT_MAGIC.to_owned(),
735                version: 2,
736                segment_size: LEGACY_SEGMENT_SIZE,
737            });
738        }
739        report.add(
740            Severity::Error,
741            "missing-file",
742            &relative,
743            None,
744            "format.json is missing",
745        );
746        return None;
747    }
748    let bytes = match crate::core::durable_file::read_manifest(&path) {
749        Ok((bytes, verified)) => {
750            if verified {
751                report.checks.push("format.json checksum");
752            } else {
753                report.add(
754                    Severity::Info,
755                    "no-checksum-sidecar",
756                    &relative,
757                    None,
758                    "no format.json.crc32 (written by a release before checksum sidecars)",
759                );
760            }
761            bytes
762        }
763        Err(error) => {
764            report.add(
765                Severity::Error,
766                "manifest-checksum",
767                &relative,
768                None,
769                format!("{error:#}"),
770            );
771            std::fs::read(&path).ok()?
772        }
773    };
774    let manifest: FormatManifest = match serde_json::from_slice(&bytes) {
775        Ok(manifest) => manifest,
776        Err(error) => {
777            report.add(
778                Severity::Error,
779                "manifest-invalid",
780                &relative,
781                None,
782                error.to_string(),
783            );
784            return None;
785        }
786    };
787    if manifest.magic != FORMAT_MAGIC {
788        report.add(
789            Severity::Unsupported,
790            "unknown-magic",
791            &relative,
792            None,
793            format!(
794                "manifest magic {:?} is not {FORMAT_MAGIC:?}",
795                manifest.magic
796            ),
797        );
798        return None;
799    }
800    if manifest.version > STORAGE_FORMAT_CURRENT {
801        report.versions.storage_format = Some(manifest.version);
802        report.add(
803            Severity::Unsupported,
804            "storage-format-newer",
805            &relative,
806            None,
807            format!(
808                "storage format {} is newer than the supported format {STORAGE_FORMAT_CURRENT}",
809                manifest.version
810            ),
811        );
812        return None;
813    }
814    if manifest.version < STORAGE_FORMAT_OLDEST {
815        report.versions.storage_format = Some(manifest.version);
816        report.add(
817            Severity::Unsupported,
818            "storage-format-unknown",
819            &relative,
820            None,
821            format!(
822                "storage format {} was never written by a release",
823                manifest.version
824            ),
825        );
826        return None;
827    }
828    Some(manifest)
829}
830
831fn check_segment_size(
832    relative: &str,
833    length: u64,
834    is_final: bool,
835    manifest: &FormatManifest,
836    report: &mut Report,
837) {
838    let declared = manifest.segment_size as u64;
839    if length > declared {
840        report.add(
841            Severity::Error,
842            "segment-size",
843            relative,
844            None,
845            format!("segment is {length} bytes, longer than the declared {declared}"),
846        );
847    } else if manifest.version >= 4 && is_final {
848        if !length.is_multiple_of(SEGMENT_SIZE as u64) || length < SEGMENT_SIZE as u64 {
849            report.add(
850                Severity::Warning,
851                "segment-size-unaligned",
852                relative,
853                None,
854                format!(
855                    "the active segment of a format-{} store is allocated in {SEGMENT_SIZE}-byte \
856                     steps; this one is {length} bytes (a copy that was trimmed? the engine \
857                     extends it with zeros on open)",
858                    manifest.version
859                ),
860            );
861        }
862    } else if length != declared {
863        let severity = if is_final {
864            Severity::Warning
865        } else {
866            Severity::Error
867        };
868        report.add(
869            severity,
870            "segment-size",
871            relative,
872            None,
873            format!(
874                "segment is {length} bytes, the declared size is {declared}{}",
875                if is_final {
876                    " (the engine extends the active segment with zeros on open)"
877                } else {
878                    ""
879                }
880            ),
881        );
882    }
883}
884
885fn check_text_segment(store: &Path, schema: &Schema, report: &mut Report) {
886    let path = store.join("documents.text");
887    let Ok(mut file) = File::open(&path) else {
888        return;
889    };
890    let mut header = [0u8; super::TEXT_SEGMENT_HEADER_LEN];
891    let read = std::io::Read::read(&mut file, &mut header).unwrap_or(0);
892    match scan::parse_text_header(&header[..read]) {
893        TextHeader::Versioned { version, flags } => {
894            report.versions.text_segment = Some(version);
895            let encrypted = flags & 1 == 1;
896            let note = if version > CURRENT.text_segment {
897                "newer than this build writes"
898            } else if encrypted != schema.encrypted {
899                "its encryption flag differs from the store's"
900            } else {
901                "a cache the engine rebuilds from the journal on every open"
902            };
903            report.add(
904                Severity::Info,
905                "text-segment-cache",
906                "documents.text",
907                None,
908                format!("version {version}, flags {flags:#04x}: {note}"),
909            );
910        }
911        TextHeader::Headerless => report.add(
912            Severity::Info,
913            "text-segment-cache",
914            "documents.text",
915            None,
916            "no WARMT header (a format-1 text segment); the engine truncates it on open",
917        ),
918    }
919}
920
921fn check_indices(store: &Path, arena: &ArenaState, report: &mut Report) {
922    let path = store.join("indices.json");
923    if !path.exists() {
924        return;
925    }
926    let bytes = match crate::core::durable_file::read_manifest(&path) {
927        Ok((bytes, verified)) => {
928            if verified {
929                report.checks.push("indices.json checksum");
930            }
931            bytes
932        }
933        Err(error) => {
934            report.add(
935                Severity::Error,
936                "manifest-checksum",
937                "indices.json",
938                None,
939                format!("{error:#}"),
940            );
941            return;
942        }
943    };
944    let meta = match scan::parse_indices(&bytes) {
945        Ok(meta) => meta,
946        Err(error) => {
947            report.add(
948                Severity::Warning,
949                "index-cache-invalid",
950                "indices.json",
951                None,
952                format!("{error}; the engine ignores it and rebuilds the indices"),
953            );
954            return;
955        }
956    };
957    let fresh = arena.checkpoint.as_ref() == Some(&meta.checkpoint);
958    report.add(
959        Severity::Info,
960        if fresh {
961            "index-cache-fresh"
962        } else {
963            "index-cache-stale"
964        },
965        "indices.json",
966        None,
967        if fresh {
968            "bound to the current journal end".to_owned()
969        } else {
970            format!(
971                "bound to {:?} at byte {}, not the current journal end; the engine rebuilds \
972                 the indices",
973                meta.checkpoint.0, meta.checkpoint.1
974            )
975        },
976    );
977    let multi = meta.shard_count > 1;
978    for (i, shard) in meta.shards.iter().enumerate() {
979        for (flag, crc, stem) in [
980            (shard.nsg, shard.nsg_crc, "nsg_index"),
981            (shard.ivf, shard.ivf_crc, "ivf_index"),
982        ] {
983            let name = if multi {
984                format!("{stem}_{i}.bin")
985            } else {
986                format!("{stem}.bin")
987            };
988            let file = store.join(&name);
989            if !flag || !file.exists() {
990                continue;
991            }
992            match (std::fs::read(&file), crc) {
993                (Ok(bytes), Some(expected)) if crc32fast::hash(&bytes) == expected => {
994                    report.checks.push("index cache checksum");
995                }
996                (Ok(_), _) => report.add(
997                    Severity::Warning,
998                    "index-cache-invalid",
999                    &name,
1000                    None,
1001                    "checksum differs from indices.json; the engine discards this cache",
1002                ),
1003                (Err(error), _) => report.add(
1004                    Severity::Warning,
1005                    "index-cache-invalid",
1006                    &name,
1007                    None,
1008                    error.to_string(),
1009                ),
1010            }
1011        }
1012    }
1013}
1014
1015fn check_ancillary(store: &Path, schema: &Schema, report: &mut Report) {
1016    const KNOWN: [&str; 17] = [
1017        "store.json",
1018        "store.json.crc32",
1019        "indices.json",
1020        "indices.json.crc32",
1021        "documents.text",
1022        "encryption.salt",
1023        "shard_meta.json",
1024        "audit.bin",
1025        "hms_signing.key",
1026        "hms_provenance.key",
1027        "provenance_log.jsonl",
1028        "provenance_log.head",
1029        "provenance_log.statuslist",
1030        "kel.json",
1031        "connection_graph.log",
1032        ".hms.lock",
1033        ARENA_DIRECTORY,
1034    ];
1035    if let Ok(bytes) = std::fs::read(store.join("audit.bin")) {
1036        let audit = scan::scan_audit(&bytes);
1037        report.checks.push("audit.bin");
1038        if audit.trailing_bytes != 0 {
1039            report.add(
1040                Severity::Warning,
1041                "audit-torn-tail",
1042                "audit.bin",
1043                Some((audit.entries * super::AUDIT_ENTRY_LEN) as u64),
1044                format!(
1045                    "{} trailing bytes after {} whole entries (an interrupted append)",
1046                    audit.trailing_bytes, audit.entries
1047                ),
1048            );
1049        }
1050        if audit.unknown_ops != 0 {
1051            report.add(
1052                Severity::Warning,
1053                "audit-unknown-op",
1054                "audit.bin",
1055                None,
1056                format!(
1057                    "{} entries have an unknown operation byte; readers skip them",
1058                    audit.unknown_ops
1059                ),
1060            );
1061        }
1062    }
1063    if let Ok(bytes) = std::fs::read(store.join("shard_meta.json")) {
1064        let count = serde_json::from_slice::<Value>(&bytes)
1065            .ok()
1066            .and_then(|v| v.get("shard_count").and_then(Value::as_u64));
1067        if count.is_none() {
1068            report.add(
1069                Severity::Warning,
1070                "shard-meta-invalid",
1071                "shard_meta.json",
1072                None,
1073                "shard_count is missing; the engine starts with a single shard",
1074            );
1075        }
1076    }
1077    if let Ok(metadata) = std::fs::metadata(store.join("hms_signing.key")) {
1078        if metadata.len() != 32 {
1079            report.add(
1080                Severity::Error,
1081                "key-invalid",
1082                "hms_signing.key",
1083                None,
1084                format!("expected 32 bytes, found {}", metadata.len()),
1085            );
1086        }
1087    }
1088    if !schema.encrypted && store.join("encryption.salt").exists() {
1089        report.add(
1090            Severity::Info,
1091            "salt-unused",
1092            "encryption.salt",
1093            None,
1094            "present but store.json says the store is not encrypted",
1095        );
1096    }
1097    if let Ok(entries) = std::fs::read_dir(store) {
1098        let mut unknown = Vec::new();
1099        for entry in entries.flatten() {
1100            let name = entry.file_name().to_string_lossy().into_owned();
1101            let cache = (name.starts_with("nsg_index") || name.starts_with("ivf_index"))
1102                && name.ends_with(".bin");
1103            if !cache && !KNOWN.contains(&name.as_str()) {
1104                unknown.push(name);
1105            }
1106        }
1107        unknown.sort();
1108        for name in unknown {
1109            report.add(
1110                Severity::Info,
1111                "unrecognized-file",
1112                name,
1113                None,
1114                "not a file this format version defines; ignored by the engine",
1115            );
1116        }
1117    }
1118}
1119
1120fn copy_store(source: &Path, destination: &Path) -> Result<()> {
1121    std::fs::create_dir_all(destination)?;
1122    for entry in std::fs::read_dir(source)? {
1123        let entry = entry?;
1124        let name = entry.file_name();
1125        if name == crate::core::store_lock::LOCK_FILE {
1126            continue;
1127        }
1128        let target: PathBuf = destination.join(&name);
1129        let kind = entry.file_type()?;
1130        if kind.is_dir() {
1131            copy_store(&entry.path(), &target)?;
1132        } else if kind.is_file() {
1133            std::fs::copy(entry.path(), &target)?;
1134        }
1135    }
1136    Ok(())
1137}
1138
1139/// Opens a private copy with the engine and compares what it loaded with the journal replay
1140/// computed here, independently of the engine.
1141fn deep_check(
1142    store: &Path,
1143    schema: &Schema,
1144    replay: &Replay,
1145    options: &VerifyOptions,
1146    report: &mut Report,
1147) -> Result<()> {
1148    let dimensions = u32::try_from(schema.dimensions.context("store.json lacks dimensions")?)?;
1149    let mut config = crate::core::WarmConfig::default();
1150    if let Some(name) = schema.raw["text_encoder"].as_str() {
1151        config.text_encoder = name.parse().map_err(|e| anyhow::anyhow!("{e}"))?;
1152    }
1153    config.meaning.enabled = schema.meaning_enabled;
1154    config.embedding_space = serde_json::from_value(schema.raw["embedding_space"].clone())
1155        .context("store.json embedding_space")?;
1156    if schema.encrypted {
1157        config.security.encryption_enabled = true;
1158        config.security.encryption_passphrase_env = options.passphrase_env.clone();
1159    }
1160    let scratch = tempfile::tempdir()?;
1161    let copy = scratch.path().join("store");
1162    copy_store(store, &copy)?;
1163    let core =
1164        crate::core::WarmCore::new(dimensions, Some(copy.display().to_string()), Some(config))?;
1165    let missing: Vec<&String> = replay
1166        .documents
1167        .keys()
1168        .filter(|id| !core.has_document(id))
1169        .collect();
1170    anyhow::ensure!(
1171        missing.is_empty(),
1172        "documents in the journal that the engine did not load: {missing:?}"
1173    );
1174    if !schema.meaning_enabled {
1175        anyhow::ensure!(
1176            core.vector_count() as usize == replay.vector_total(),
1177            "engine holds {} vectors, the journal replays to {}",
1178            core.vector_count(),
1179            replay.vector_total()
1180        );
1181        anyhow::ensure!(
1182            core.relation_count() == replay.relations.len(),
1183            "engine holds {} relations, the journal replays to {}",
1184            core.relation_count(),
1185            replay.relations.len()
1186        );
1187    }
1188    report.checks.push("engine open");
1189    // A journal checkpoint (`documents.ckpt`) holds the documents of the frames it replaced,
1190    // which the replay above never sees; the engine checked it whole when it opened.
1191    let checkpointed = std::fs::read_dir(store.join(ARENA_DIRECTORY))
1192        .into_iter()
1193        .flatten()
1194        .filter_map(Result::ok)
1195        .any(|entry| entry.path().join("documents.ckpt").is_file());
1196    if checkpointed && report.live_documents.is_some() {
1197        report.live_documents = Some(core.document_count());
1198        report.checks.push("checkpoint");
1199    }
1200    Ok(())
1201}
writerslogic_warm/core/format/verify.rs

writerslogic_warm/core/format/
verify.rs
