verify/
verify.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::{
17    any::Any,
18    env,
19    ffi::OsStr,
20    io::{self, Read},
21    panic::{self, AssertUnwindSafe},
22    path::{Path, PathBuf},
23    process::{Command, ExitCode, Output, Stdio},
24    thread,
25    time::{Duration, Instant},
26};
27
28use nautilus_event_store::{
29    EventStoreError, IndexDrift, MarkerCountKind, MarkerFinding, MarkerRecordKind, MarkerVerifier,
30    MarkerVerifyReport, RedbMarkerBackend, Verifier, VerifyError, VerifyFinding, VerifyReport,
31};
32
33const EXIT_CLEAN: u8 = 0;
34const EXIT_CORRUPT: u8 = 1;
35const EXIT_ERROR: u8 = 2;
36const WORKER_POLL_INTERVAL: Duration = Duration::from_millis(100);
37const DEFAULT_WORKER_TIMEOUT: Duration = Duration::from_secs(30);
38const TIMEOUT_ENV: &str = "NAUTILUS_EVENT_STORE_VERIFY_TIMEOUT_SECS";
39#[cfg(debug_assertions)]
40const WORKER_SLEEP_ENV: &str = "NAUTILUS_EVENT_STORE_VERIFY_SLEEP_WORKER_MS";
41
42enum WorkerOutput {
43    Exited(Output),
44    TimedOut { output: Output, timeout: Duration },
45}
46
47enum MarkerScan {
48    Absent,
49    Present(MarkerVerifyReport),
50}
51
52fn main() -> ExitCode {
53    let mut args = env::args_os();
54    let program = args.next().unwrap_or_else(|| OsStr::new("verify").into());
55    let Some(first) = args.next() else {
56        print_usage(&program);
57        return ExitCode::from(EXIT_ERROR);
58    };
59
60    if first.as_os_str() == OsStr::new("--worker") {
61        let Some(path) = args.next() else {
62            print_usage(&program);
63            return ExitCode::from(EXIT_ERROR);
64        };
65
66        if args.next().is_some() {
67            print_usage(&program);
68            return ExitCode::from(EXIT_ERROR);
69        }
70
71        let path = PathBuf::from(path);
72        return verify_run_file_worker(path.as_path());
73    }
74
75    if args.next().is_some() {
76        print_usage(&program);
77        return ExitCode::from(EXIT_ERROR);
78    }
79
80    let path = PathBuf::from(first);
81    verify_run_file(path.as_path())
82}
83
84fn verify_run_file(path: &Path) -> ExitCode {
85    let output = match run_worker(path) {
86        Ok(output) => output,
87        Err(e) => {
88            eprintln!("error path={} error=\"worker: {e}\"", path.display());
89            return ExitCode::from(EXIT_ERROR);
90        }
91    };
92
93    match output {
94        WorkerOutput::Exited(output) => classify_worker_exit(path, &output),
95        WorkerOutput::TimedOut { output, timeout } => {
96            println!(
97                "corrupt path={} worker_status=\"timeout after {}s\" quarantine=not-performed",
98                path.display(),
99                timeout.as_secs(),
100            );
101            relay_output(&output);
102            ExitCode::from(EXIT_CORRUPT)
103        }
104    }
105}
106
107fn classify_worker_exit(path: &Path, output: &Output) -> ExitCode {
108    match output.status.code() {
109        Some(code)
110            if code == i32::from(EXIT_CLEAN)
111                || code == i32::from(EXIT_CORRUPT)
112                || code == i32::from(EXIT_ERROR) =>
113        {
114            relay_output(output);
115            ExitCode::from(u8::try_from(code).expect("known verifier exit code"))
116        }
117        _ => {
118            println!(
119                "corrupt path={} worker_status=\"{}\" quarantine=not-performed",
120                path.display(),
121                output.status,
122            );
123            relay_output(output);
124            ExitCode::from(EXIT_CORRUPT)
125        }
126    }
127}
128
129fn verify_run_file_worker(path: &Path) -> ExitCode {
130    abort_worker_when_requested();
131    sleep_worker_when_requested();
132
133    let previous_hook = panic::take_hook();
134    panic::set_hook(Box::new(|_| {}));
135    let result = panic::catch_unwind(AssertUnwindSafe(|| verify_run_file_inner(path)));
136    panic::set_hook(previous_hook);
137
138    match result {
139        Ok(code) => code,
140        Err(payload) => {
141            println!(
142                "corrupt path={} panic=\"{}\" quarantine=not-performed",
143                path.display(),
144                panic_message(payload.as_ref()),
145            );
146            ExitCode::from(EXIT_CORRUPT)
147        }
148    }
149}
150
151fn verify_run_file_inner(path: &Path) -> ExitCode {
152    match Verifier::open_redb_file(path).and_then(|verifier| verifier.verify()) {
153        Ok(report) => match scan_marker_sidecar(path, &report) {
154            Ok(markers) => print_report(&report, &markers),
155            Err((marker_path, err)) => {
156                if report.is_clean() {
157                    return print_marker_error(marker_path.as_path(), &err);
158                }
159
160                // Relay the entry findings before the marker error so a failing
161                // sidecar cannot mask them.
162                println!(
163                    "corrupt run_id={} status={:?} high_watermark={} entries_scanned={} findings={} markers=error quarantine=not-performed",
164                    report.run_id,
165                    report.status,
166                    report.high_watermark,
167                    report.entries_scanned,
168                    report.findings.len(),
169                );
170
171                for finding in &report.findings {
172                    print_finding(finding);
173                }
174
175                print_marker_error(marker_path.as_path(), &err);
176                ExitCode::from(EXIT_CORRUPT)
177            }
178        },
179        Err(e) => print_error(path, &e),
180    }
181}
182
183fn scan_marker_sidecar(
184    path: &Path,
185    entry_report: &VerifyReport,
186) -> Result<MarkerScan, (PathBuf, VerifyError)> {
187    let Some(marker_path) = marker_sidecar_path(path) else {
188        return Ok(MarkerScan::Absent);
189    };
190
191    if !marker_path.exists() {
192        return Ok(MarkerScan::Absent);
193    }
194
195    let backend = RedbMarkerBackend::open_read_only_file(&marker_path)
196        .map_err(|e| (marker_path.clone(), VerifyError::Backend(e)))?;
197    let report = MarkerVerifier::scan(&backend, entry_report.high_watermark)
198        .map_err(|e| (marker_path.clone(), VerifyError::Backend(e)))?;
199    Ok(MarkerScan::Present(report))
200}
201
202fn marker_sidecar_path(path: &Path) -> Option<PathBuf> {
203    let stem = path.file_stem()?;
204    let mut file_name = stem.to_os_string();
205    file_name.push(".markers.redb");
206    Some(path.with_file_name(file_name))
207}
208
209fn print_report(report: &VerifyReport, markers: &MarkerScan) -> ExitCode {
210    let marker_findings = marker_finding_count(markers);
211
212    if report.is_clean() && marker_findings == 0 {
213        println!(
214            "clean run_id={} status={:?} high_watermark={} entries_scanned={} {}",
215            report.run_id,
216            report.status,
217            report.high_watermark,
218            report.entries_scanned,
219            marker_summary(markers),
220        );
221        return ExitCode::from(EXIT_CLEAN);
222    }
223
224    println!(
225        "corrupt run_id={} status={:?} high_watermark={} entries_scanned={} findings={} marker_findings={} {} quarantine=not-performed",
226        report.run_id,
227        report.status,
228        report.high_watermark,
229        report.entries_scanned,
230        report.findings.len() + marker_findings,
231        marker_findings,
232        marker_summary(markers),
233    );
234
235    for finding in &report.findings {
236        print_finding(finding);
237    }
238
239    if let MarkerScan::Present(report) = markers {
240        for finding in &report.findings {
241            print_marker_finding(finding);
242        }
243    }
244
245    ExitCode::from(EXIT_CORRUPT)
246}
247
248fn marker_finding_count(markers: &MarkerScan) -> usize {
249    match markers {
250        MarkerScan::Absent => 0,
251        MarkerScan::Present(report) => report.findings.len(),
252    }
253}
254
255fn marker_summary(markers: &MarkerScan) -> String {
256    match markers {
257        MarkerScan::Absent => "markers=absent".to_string(),
258        MarkerScan::Present(report) if report.is_clean() => format!(
259            "markers=clean marker_run_id={} marker_status={:?} marker_snapshots_scanned={} marker_hifi_scanned={} marker_gaps_scanned={} marker_dict_entries_scanned={}",
260            report.run_id,
261            report.status,
262            report.snapshots_scanned,
263            report.hifi_scanned,
264            report.gaps_scanned,
265            report.dict_entries_scanned,
266        ),
267        MarkerScan::Present(report) => format!(
268            "markers=corrupt marker_run_id={} marker_status={:?} marker_snapshots_scanned={} marker_hifi_scanned={} marker_gaps_scanned={} marker_dict_entries_scanned={} marker_findings={}",
269            report.run_id,
270            report.status,
271            report.snapshots_scanned,
272            report.hifi_scanned,
273            report.gaps_scanned,
274            report.dict_entries_scanned,
275            report.findings.len(),
276        ),
277    }
278}
279
280fn print_error(path: &Path, err: &VerifyError) -> ExitCode {
281    if matches!(err, VerifyError::Backend(EventStoreError::Corrupted(_))) {
282        println!(
283            "corrupt path={} error=\"{}\" quarantine=not-performed",
284            path.display(),
285            err,
286        );
287        return ExitCode::from(EXIT_CORRUPT);
288    }
289
290    eprintln!("error path={} error=\"{}\"", path.display(), err);
291    ExitCode::from(EXIT_ERROR)
292}
293
294fn print_marker_error(path: &Path, err: &VerifyError) -> ExitCode {
295    if matches!(err, VerifyError::Backend(EventStoreError::Corrupted(_))) {
296        println!(
297            "corrupt path={} markers=error error=\"{}\" quarantine=not-performed",
298            path.display(),
299            err,
300        );
301        return ExitCode::from(EXIT_CORRUPT);
302    }
303
304    eprintln!(
305        "error path={} markers=error error=\"{}\"",
306        path.display(),
307        err
308    );
309    ExitCode::from(EXIT_ERROR)
310}
311
312fn print_finding(finding: &VerifyFinding) {
313    match finding {
314        VerifyFinding::HashMismatch { seq } => {
315            println!("- hash mismatch at seq {seq}");
316        }
317        VerifyFinding::Gap { range } => {
318            println!("- gap from seq {} to {}", range.from, range.to);
319        }
320        VerifyFinding::SeqMismatch {
321            table_key,
322            embedded_seq,
323        } => {
324            println!("- seq mismatch at table key {table_key}: embedded seq was {embedded_seq}");
325        }
326        VerifyFinding::Undecodable { seq, reason } => {
327            println!("- undecodable entry at seq {seq}: {reason}");
328        }
329        VerifyFinding::IndexDrift { kind, key, drift } => {
330            print_index_drift(*kind, key, *drift);
331        }
332        VerifyFinding::ManifestMismatch { kind, reason } => {
333            println!("- manifest mismatch {kind:?}: {reason}");
334        }
335        VerifyFinding::SnapshotAnchorInvalid { reason } => {
336            println!("- snapshot anchor invalid: {reason}");
337        }
338    }
339}
340
341fn print_index_drift(kind: nautilus_event_store::IndexKind, key: &str, drift: IndexDrift) {
342    match drift {
343        IndexDrift::DanglingTarget { stored_seq } => {
344            println!("- index drift {kind:?} key={key}: dangling target seq {stored_seq}");
345        }
346        IndexDrift::TargetCorrupted { stored_seq } => {
347            println!("- index drift {kind:?} key={key}: corrupted target seq {stored_seq}");
348        }
349    }
350}
351
352fn print_marker_finding(finding: &MarkerFinding) {
353    match finding {
354        MarkerFinding::ManifestCountMismatch {
355            kind,
356            manifest_count,
357            scanned_count,
358        } => {
359            println!(
360                "- marker manifest count mismatch {}: manifest={manifest_count} scanned={scanned_count}",
361                marker_count_name(*kind),
362            );
363        }
364        MarkerFinding::MarkerSeqGap {
365            from_marker_seq,
366            to_marker_seq,
367        } => {
368            println!("- marker seq gap from {from_marker_seq} to {to_marker_seq}");
369        }
370        MarkerFinding::MarkerSeqOverlap {
371            from_marker_seq,
372            to_marker_seq,
373        } => {
374            println!("- marker seq overlap from {from_marker_seq} to {to_marker_seq}");
375        }
376        MarkerFinding::InvalidMarkerGap {
377            from_marker_seq,
378            to_marker_seq,
379        } => {
380            println!("- invalid marker gap from {from_marker_seq} to {to_marker_seq}");
381        }
382        MarkerFinding::EventSeqRegressed {
383            marker_seq,
384            previous_event_seq_before,
385            event_seq_before,
386        } => {
387            println!(
388                "- marker event seq regressed marker_seq={marker_seq}: previous={previous_event_seq_before} current={event_seq_before}",
389            );
390        }
391        MarkerFinding::EventSeqExceedsHighWatermark {
392            marker_seq,
393            event_seq_before,
394            high_watermark,
395        } => {
396            println!(
397                "- marker event seq exceeds high watermark marker_seq={marker_seq}: event_seq_before={event_seq_before} high_watermark={high_watermark}",
398            );
399        }
400        MarkerFinding::CursorCountRegressed {
401            marker_seq,
402            slot,
403            previous_count,
404            count,
405        } => {
406            println!(
407                "- marker cursor count regressed marker_seq={marker_seq} slot={slot}: previous={previous_count} current={count}",
408            );
409        }
410        MarkerFinding::CursorTsInitRegressed {
411            marker_seq,
412            slot,
413            previous_ts_init_hi,
414            ts_init_hi,
415        } => {
416            println!(
417                "- marker cursor ts_init_hi regressed marker_seq={marker_seq} slot={slot}: previous={} current={}",
418                previous_ts_init_hi.as_u64(),
419                ts_init_hi.as_u64(),
420            );
421        }
422        MarkerFinding::HashMismatch {
423            record,
424            marker_seq,
425            slot,
426        } => {
427            print_marker_hash_mismatch(*record, *marker_seq, *slot);
428        }
429    }
430}
431
432fn print_marker_hash_mismatch(
433    record: MarkerRecordKind,
434    marker_seq: Option<u64>,
435    slot: Option<u32>,
436) {
437    match (marker_seq, slot) {
438        (Some(marker_seq), Some(slot)) => println!(
439            "- marker hash mismatch {} marker_seq={marker_seq} slot={slot}",
440            marker_record_name(record),
441        ),
442        (Some(marker_seq), None) => println!(
443            "- marker hash mismatch {} marker_seq={marker_seq}",
444            marker_record_name(record),
445        ),
446        (None, Some(slot)) => println!(
447            "- marker hash mismatch {} slot={slot}",
448            marker_record_name(record),
449        ),
450        (None, None) => println!("- marker hash mismatch {}", marker_record_name(record)),
451    }
452}
453
454fn marker_record_name(record: MarkerRecordKind) -> &'static str {
455    match record {
456        MarkerRecordKind::Snapshot => "snapshot",
457        MarkerRecordKind::HiFi => "hifi",
458        MarkerRecordKind::Gap => "gap",
459        MarkerRecordKind::Dict => "dict",
460    }
461}
462
463fn marker_count_name(kind: MarkerCountKind) -> &'static str {
464    match kind {
465        MarkerCountKind::Snapshot => "snapshot",
466        MarkerCountKind::HiFi => "hifi",
467        MarkerCountKind::Gap => "gap",
468        MarkerCountKind::Dict => "dict",
469    }
470}
471
472fn print_usage(program: &OsStr) {
473    eprintln!("usage: {} <run-file.redb>", program.to_string_lossy());
474}
475
476fn run_worker(path: &Path) -> io::Result<WorkerOutput> {
477    let current_exe = env::current_exe()?;
478    let mut child = Command::new(current_exe)
479        .arg("--worker")
480        .arg(path)
481        .stdout(Stdio::piped())
482        .stderr(Stdio::piped())
483        .spawn()?;
484    let timeout = worker_timeout();
485    let deadline = Instant::now() + timeout;
486
487    // A report larger than the OS pipe buffer would otherwise block mid-print
488    // and be killed as a timeout.
489    let stdout_drain = spawn_pipe_drain(child.stdout.take());
490    let stderr_drain = spawn_pipe_drain(child.stderr.take());
491
492    let mut timed_out = false;
493    let status = loop {
494        if let Some(status) = child.try_wait()? {
495            break status;
496        }
497
498        if Instant::now() >= deadline {
499            let _ = child.kill();
500            timed_out = true;
501            break child.wait()?;
502        }
503
504        thread::sleep(WORKER_POLL_INTERVAL);
505    };
506
507    let output = Output {
508        status,
509        stdout: stdout_drain.join().unwrap_or_default(),
510        stderr: stderr_drain.join().unwrap_or_default(),
511    };
512
513    if timed_out {
514        Ok(WorkerOutput::TimedOut { output, timeout })
515    } else {
516        Ok(WorkerOutput::Exited(output))
517    }
518}
519
520fn spawn_pipe_drain<R: Read + Send + 'static>(pipe: Option<R>) -> thread::JoinHandle<Vec<u8>> {
521    thread::spawn(move || {
522        let mut buf = Vec::new();
523        if let Some(mut pipe) = pipe {
524            let _ = pipe.read_to_end(&mut buf);
525        }
526        buf
527    })
528}
529
530fn relay_output(output: &Output) {
531    if !output.stdout.is_empty() {
532        print!("{}", String::from_utf8_lossy(&output.stdout));
533    }
534
535    if !output.stderr.is_empty() {
536        eprint!("{}", String::from_utf8_lossy(&output.stderr));
537    }
538}
539
540fn panic_message(payload: &(dyn Any + Send)) -> &str {
541    if let Some(message) = payload.downcast_ref::<&'static str>() {
542        message
543    } else if let Some(message) = payload.downcast_ref::<String>() {
544        message.as_str()
545    } else {
546        "unknown panic"
547    }
548}
549
550fn worker_timeout() -> Duration {
551    env::var(TIMEOUT_ENV)
552        .ok()
553        .and_then(|value| value.parse::<u64>().ok())
554        .map_or(DEFAULT_WORKER_TIMEOUT, Duration::from_secs)
555}
556
557fn abort_worker_when_requested() {
558    #[cfg(debug_assertions)]
559    if env::var_os("NAUTILUS_EVENT_STORE_VERIFY_ABORT_WORKER").is_some() {
560        std::process::abort();
561    }
562}
563
564fn sleep_worker_when_requested() {
565    #[cfg(debug_assertions)]
566    if let Some(raw) = env::var_os(WORKER_SLEEP_ENV)
567        && let Ok(ms) = raw.to_string_lossy().parse::<u64>()
568    {
569        thread::sleep(Duration::from_millis(ms));
570    }
571}
verify/verify.rs

verify/
verify.rs
