hayson_engine/database/
verify.rs

1use std::fs;
2use std::io::{self, BufRead, BufReader};
3use std::path::{Path, PathBuf};
4
5use serde_json::Value;
6
7use crate::catalog::{parse_catalog, Catalog, TableEntry};
8use crate::storage::{parse_entry, partition_names_in, LOG_SUFFIX};
9
10use super::database::{Database, LockMode, OpenOptions};
11use super::layout::{Layout, LayoutOptions};
12
13/// How much a finding matters to someone about to open the directory.
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum Severity {
16    /// The directory cannot be opened as it is, or data in it cannot be read back.
17    Error,
18    /// Opening works; the engine repairs it on the next writable open, or it is only worth a look.
19    Warning,
20}
21
22/// One finding of [`verify`]. `code` is a stable kebab-case identifier callers can match on;
23/// `message` is for people.
24#[derive(Debug, Clone, PartialEq)]
25pub struct Problem {
26    pub severity: Severity,
27    pub code: &'static str,
28    pub table: Option<String>,
29    /// Path of the file concerned, relative to the database directory.
30    pub file: Option<String>,
31    /// Byte offset of the offending line in `file`.
32    pub offset: Option<u64>,
33    pub message: String,
34}
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub struct VerifyOptions {
38    /// Also open the database read-only (no lock) and report relation orphans. Skipped when the
39    /// structural checks already found an error, because the open would only repeat it.
40    pub relations: bool,
41}
42
43impl Default for VerifyOptions {
44    fn default() -> Self {
45        VerifyOptions { relations: true }
46    }
47}
48
49#[derive(Debug, Clone, PartialEq, Default)]
50pub struct VerifyReport {
51    pub problems: Vec<Problem>,
52    pub tables_checked: usize,
53    pub log_lines_checked: u64,
54}
55
56impl VerifyReport {
57    pub fn has_errors(&self) -> bool {
58        self.problems.iter().any(|p| p.severity == Severity::Error)
59    }
60
61    pub fn is_clean(&self) -> bool {
62        self.problems.is_empty()
63    }
64}
65
66/// Checks a database directory without taking its lock or changing a byte, and reports what is
67/// wrong with it instead of failing on it: every damaged, missing or unreadable file becomes a
68/// [`Problem`], never an `Err` and never a panic.
69///
70/// Looks at the metadata document, each declared table's log files line by line, logs the catalog
71/// does not declare, an unfinished unit-of-work journal and (optionally) declared relations.
72pub fn verify(dir: &Path, options: &VerifyOptions) -> VerifyReport {
73    let mut report = VerifyReport::default();
74    let mut checker = Checker {
75        root: dir.to_path_buf(),
76        report: &mut report,
77    };
78    checker.run(options);
79    report
80}
81
82struct Checker<'a> {
83    root: PathBuf,
84    report: &'a mut VerifyReport,
85}
86
87impl Checker<'_> {
88    fn push(&mut self, severity: Severity, code: &'static str, message: String) -> &mut Problem {
89        self.report.problems.push(Problem {
90            severity,
91            code,
92            table: None,
93            file: None,
94            offset: None,
95            message,
96        });
97        self.report.problems.last_mut().expect("just pushed")
98    }
99
100    fn relative(&self, path: &Path) -> String {
101        path.strip_prefix(&self.root).unwrap_or(path).display().to_string()
102    }
103
104    fn run(&mut self, options: &VerifyOptions) {
105        if !self.root.is_dir() {
106            let message = format!("\"{}\" is not a directory", self.root.display());
107            self.push(Severity::Error, "directory-missing", message);
108            return;
109        }
110        let layout = match Database::resolve_layout(&self.root, &LayoutOptions::default()) {
111            Ok(layout) => layout,
112            Err(error) => {
113                self.push(Severity::Error, "layout-unresolved", error.to_string());
114                return;
115            }
116        };
117        let Some(catalog) = self.read_catalog(&layout) else {
118            return;
119        };
120
121        for (name, entry) in &catalog.tables {
122            self.check_table(&layout, name, entry);
123        }
124        self.check_orphan_logs(&layout, &catalog);
125        self.check_journal(&layout);
126
127        if options.relations && !self.report.has_errors() {
128            self.check_relations();
129        }
130    }
131
132    fn read_catalog(&mut self, layout: &Layout) -> Option<Catalog> {
133        let path = layout.metadata_path();
134        let file = self.relative(path);
135        let raw = match fs::read_to_string(path) {
136            Ok(raw) => raw,
137            Err(error) if error.kind() == io::ErrorKind::NotFound => {
138                let holds_logs = !self.log_files_under(layout).is_empty();
139                let severity = if holds_logs { Severity::Error } else { Severity::Warning };
140                self.push(severity, "catalog-missing", format!("\"{file}\" does not exist"))
141                    .file = Some(file.clone());
142                return None;
143            }
144            Err(error) => {
145                self.push(Severity::Error, "catalog-unreadable", format!("\"{file}\": {error}"))
146                    .file = Some(file.clone());
147                return None;
148            }
149        };
150        let value: Value = match serde_json::from_str(&raw) {
151            Ok(value) => value,
152            Err(error) => {
153                self.push(
154                    Severity::Error,
155                    "catalog-invalid-json",
156                    format!("\"{file}\" is not valid JSON: {error}"),
157                )
158                .file = Some(file.clone());
159                return None;
160            }
161        };
162        match parse_catalog(&value) {
163            Ok(catalog) => Some(catalog),
164            Err(error) => {
165                self.push(Severity::Error, "catalog-invalid", format!("\"{file}\": {error}"))
166                    .file = Some(file.clone());
167                None
168            }
169        }
170    }
171
172    /// Every `*.jsonl` file directly in the tables directory or one level below it.
173    fn log_files_under(&self, layout: &Layout) -> Vec<PathBuf> {
174        let is_log = |path: &Path| path.is_file() && path.to_string_lossy().ends_with(LOG_SUFFIX);
175        let mut found = Vec::new();
176        let Ok(top) = fs::read_dir(layout.tables()) else {
177            return found;
178        };
179        for path in top.filter_map(Result::ok).map(|entry| entry.path()) {
180            if is_log(&path) {
181                found.push(path);
182            } else if path.is_dir() {
183                if let Ok(inner) = fs::read_dir(&path) {
184                    found.extend(
185                        inner
186                            .filter_map(Result::ok)
187                            .map(|entry| entry.path())
188                            .filter(|p| is_log(p)),
189                    );
190                }
191            }
192        }
193        found.sort();
194        found
195    }
196
197    fn logs_of(&self, layout: &Layout, name: &str, entry: &TableEntry) -> Vec<PathBuf> {
198        let flat = layout.log_path(name);
199        if entry.partition_by.is_some() {
200            let dir = layout.tables().join(name);
201            let parts = partition_names_in(&dir);
202            if !parts.is_empty() {
203                return parts
204                    .iter()
205                    .map(|part| dir.join(format!("{part}{LOG_SUFFIX}")))
206                    .collect();
207            }
208        }
209        if fs::symlink_metadata(&flat).is_ok() {
210            vec![flat]
211        } else {
212            Vec::new()
213        }
214    }
215
216    fn check_table(&mut self, layout: &Layout, name: &str, entry: &TableEntry) {
217        self.report.tables_checked += 1;
218        for path in self.logs_of(layout, name, entry) {
219            self.check_log(name, &path);
220        }
221    }
222
223    fn check_log(&mut self, table: &str, path: &Path) {
224        let file = self.relative(path);
225        let opened = match fs::File::open(path) {
226            Ok(opened) => opened,
227            Err(error) => {
228                let problem = self.push(Severity::Error, "log-unreadable", format!("\"{file}\": {error}"));
229                problem.table = Some(table.to_string());
230                problem.file = Some(file.clone());
231                return;
232            }
233        };
234        let mut reader = BufReader::new(opened);
235        let mut buffer = Vec::new();
236        let mut offset = 0u64;
237        loop {
238            buffer.clear();
239            let read = match reader.read_until(b'\n', &mut buffer) {
240                Ok(0) => break,
241                Ok(read) => read,
242                Err(error) => {
243                    let problem = self.push(
244                        Severity::Error,
245                        "log-unreadable",
246                        format!("\"{file}\" stopped being readable at byte {offset}: {error}"),
247                    );
248                    problem.table = Some(table.to_string());
249                    problem.file = Some(file.clone());
250                    problem.offset = Some(offset);
251                    return;
252                }
253            };
254            let terminated = buffer.last() == Some(&b'\n');
255            if !terminated {
256                let problem = self.push(
257                    Severity::Warning,
258                    "log-truncated-tail",
259                    format!("\"{file}\" ends with {read} bytes of an unfinished line at byte {offset}; a writable open cuts it off"),
260                );
261                problem.table = Some(table.to_string());
262                problem.file = Some(file.clone());
263                problem.offset = Some(offset);
264                return;
265            }
266            let content = &buffer[..buffer.len() - 1];
267            if !content.is_empty() {
268                self.report.log_lines_checked += 1;
269                let reason = match std::str::from_utf8(content) {
270                    Ok(line) => parse_entry(line, &file, offset).err().map(|error| error.reason),
271                    Err(error) => Some(format!("line is not valid UTF-8 ({error})")),
272                };
273                if let Some(reason) = reason {
274                    let problem = self.push(
275                        Severity::Error,
276                        "log-corrupt-line",
277                        format!("\"{file}\" at byte {offset}: {reason}"),
278                    );
279                    problem.table = Some(table.to_string());
280                    problem.file = Some(file.clone());
281                    problem.offset = Some(offset);
282                }
283            }
284            offset += read as u64;
285        }
286    }
287
288    fn check_orphan_logs(&mut self, layout: &Layout, catalog: &Catalog) {
289        for path in self.log_files_under(layout) {
290            let Some(table) = table_of_log(layout, &path) else {
291                continue;
292            };
293            if !catalog.tables.contains_key(&table) {
294                let file = self.relative(&path);
295                let problem = self.push(
296                    Severity::Warning,
297                    "log-undeclared",
298                    format!("\"{file}\" belongs to table \"{table}\", which the catalog does not declare"),
299                );
300                problem.table = Some(table);
301                problem.file = Some(file.clone());
302            }
303        }
304    }
305
306    fn check_journal(&mut self, layout: &Layout) {
307        let path = layout.journal_path();
308        if fs::symlink_metadata(&path).is_ok() {
309            let file = self.relative(&path);
310            self.push(
311                Severity::Warning,
312                "journal-unfinished",
313                format!(
314                    "\"{file}\" is a unit of work that never finished; the next writable open completes or discards it"
315                ),
316            )
317            .file = Some(file.clone());
318        }
319    }
320
321    fn check_relations(&mut self) {
322        let options = OpenOptions {
323            readonly: true,
324            lock: LockMode::Skip,
325            ..OpenOptions::default()
326        };
327        let database = match Database::open_with(&self.root, options) {
328            Ok(database) => database,
329            Err(error) => {
330                self.push(Severity::Error, "open-failed", error.to_string());
331                return;
332            }
333        };
334        match database.check_integrity() {
335            Ok(integrity) => {
336                for orphan in integrity.orphans {
337                    let problem = self.push(
338                        Severity::Warning,
339                        "relation-orphan",
340                        format!(
341                            "relation \"{}\": {}.{} = {} of record {} points at nothing",
342                            orphan.relation, orphan.table, orphan.field, orphan.value, orphan.primary_key
343                        ),
344                    );
345                    problem.table = Some(orphan.table);
346                }
347            }
348            Err(error) => {
349                self.push(Severity::Error, "integrity-check-failed", error.to_string());
350            }
351        }
352    }
353}
354
355/// The table a log file belongs to: `<table>.jsonl`, or `<table>/<part>.jsonl` one level down.
356fn table_of_log(layout: &Layout, path: &Path) -> Option<String> {
357    let relative = path.strip_prefix(layout.tables()).ok()?;
358    let mut parts = relative.components();
359    let first = parts.next()?.as_os_str().to_string_lossy().to_string();
360    if parts.next().is_some() {
361        Some(first)
362    } else {
363        first.strip_suffix(LOG_SUFFIX).map(str::to_string)
364    }
365}
hayson_engine/database/verify.rs

hayson_engine/database/
verify.rs
