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