Skip to main content

nautilus_event_store/verifier/
mod.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
16//! Off-trader verifier that proves a run file's integrity before the trader opens it.
17//!
18//! See `README.md` "Storage backend" and "Determinism contract" sections for the SPEC
19//! posture: redb 4.x does not framewise-checksum data pages, so a `zero-tail` corruption
20//! opens cleanly and panics on first read. The verifier therefore exercises every entry
21//! and every stored index pair, accumulating findings so a single run produces one
22//! actionable report rather than failing fast on the first hit. The supervisor runs the
23//! verifier in an isolated process so a bad file aborts the verifier, not trading.
24//!
25//! Scope:
26//!
27//! - Walk every `seq` over `[1, high_watermark]` and recompute [`crate::EntryHash`].
28//! - Detect gaps in the seq sequence (the SPEC's gap-detection idempotency primitive).
29//! - Validate that every `client_order_id` and `venue_order_id` stored target seq still
30//!   resolves to a clean entry; full payload-derived rebuild is deferred until the
31//!   wrapper-type encoders land.
32//! - Validate manifest invariants: `high_watermark` matches the durable last seq, the
33//!   recorded `start_ts_init` and `end_ts_init` bracket the entry stream, and a sealed
34//!   manifest's status is a terminal state.
35//!
36//! The library API stays narrow: a single [`Verifier`] type that owns a backend, a
37//! [`VerifyReport`] structured for downstream operator tooling, and a [`VerifyError`]
38//! reserved for failures that prevent the verifier from producing any report at all.
39
40use std::{collections::BTreeSet, fmt::Debug, path::Path};
41
42use crate::{
43    backend::{EventStore, IndexKind, RedbBackend},
44    entry::EventStoreEntry,
45    error::EventStoreError,
46    manifest::{RunId, RunManifest, RunStatus},
47};
48
49/// Verifier over a single open run.
50///
51/// Constructed either by passing an already-open backend ([`Verifier::new`]) or by
52/// opening a sealed redb file directly ([`Verifier::open_redb`] or
53/// [`Verifier::open_redb_file`]). The verifier never mutates the backend; it walks the
54/// entry table and the secondary indices, then emits a typed [`VerifyReport`].
55pub struct Verifier {
56    backend: Box<dyn EventStore>,
57}
58
59impl Debug for Verifier {
60    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
61        f.debug_struct(stringify!(Verifier)).finish_non_exhaustive()
62    }
63}
64
65impl Verifier {
66    /// Wraps an already-open backend for read-only verification.
67    #[must_use]
68    pub fn new(backend: Box<dyn EventStore>) -> Self {
69        Self { backend }
70    }
71
72    /// Opens a sealed redb run file at `<base_dir>/<instance_id>/<run_id>.redb` and
73    /// wraps it for verification.
74    ///
75    /// Mirrors [`crate::backend::RedbBackend::open_sealed`]: only sealed files are
76    /// accepted, since opening a still-`Running` file would race with a live writer
77    /// and break the off-trader-process posture.
78    ///
79    /// # Errors
80    ///
81    /// Returns [`VerifyError::Backend`] when the underlying backend rejects the open
82    /// (file missing, run still `Running`, header corruption).
83    pub fn open_redb(
84        base_dir: impl AsRef<Path>,
85        instance_id: &str,
86        run_id: &str,
87    ) -> Result<Self, VerifyError> {
88        let backend =
89            RedbBackend::open_sealed(base_dir.as_ref().to_path_buf(), instance_id, run_id)?;
90        Ok(Self {
91            backend: Box::new(backend),
92        })
93    }
94
95    /// Opens a sealed redb run file directly by path and wraps it for verification.
96    ///
97    /// The backend uses a read-only database handle for this path. The verifier
98    /// reports findings, but it never seals or quarantines the file; a supervisor or
99    /// operator process decides that policy from the returned [`VerifyReport`].
100    ///
101    /// # Errors
102    ///
103    /// Returns [`VerifyError::Backend`] when the underlying backend rejects the open
104    /// (file missing, run still `Running`, header corruption).
105    pub fn open_redb_file(path: impl AsRef<Path>) -> Result<Self, VerifyError> {
106        let backend = RedbBackend::open_sealed_file(path.as_ref().to_path_buf())?;
107        Ok(Self {
108            backend: Box::new(backend),
109        })
110    }
111
112    /// Returns a reference to the wrapped backend.
113    #[must_use]
114    pub fn backend(&self) -> &dyn EventStore {
115        self.backend.as_ref()
116    }
117
118    /// Performs a full integrity scan of the open run and returns the typed report.
119    ///
120    /// `verify` reads the manifest, walks every `seq` in `[1, high_watermark]`,
121    /// cross-checks the stored client- and venue-order-id indices, and validates manifest
122    /// invariants. Hash mismatches, gaps, index drift, and manifest mismatches surface as
123    /// [`VerifyFinding`]s on the returned report; only failures that prevent the verifier
124    /// from producing a report at all surface as [`VerifyError`].
125    ///
126    /// # Errors
127    ///
128    /// Returns [`VerifyError::Backend`] when the backend refuses a read-side
129    /// operation (no run open, disk pressure, manifest decode failure).
130    pub fn verify(&self) -> Result<VerifyReport, VerifyError> {
131        let manifest = self.backend.manifest()?;
132        let high_watermark = self.backend.high_watermark()?;
133
134        let mut findings = Vec::new();
135        let scan = self.scan_entries(high_watermark, &mut findings)?;
136
137        self.cross_check_indices(&scan, &mut findings)?;
138        check_snapshot_anchor(self.backend.as_ref(), high_watermark, &mut findings)?;
139        validate_manifest(&manifest, high_watermark, &scan, &mut findings);
140
141        Ok(VerifyReport {
142            run_id: manifest.run_id.clone(),
143            status: manifest.status,
144            high_watermark,
145            entries_scanned: scan.scanned,
146            findings,
147        })
148    }
149
150    fn scan_entries(
151        &self,
152        high_watermark: u64,
153        findings: &mut Vec<VerifyFinding>,
154    ) -> Result<EntryScan, VerifyError> {
155        let mut scanned: u64 = 0;
156        let mut min_ts: Option<u64> = None;
157        let mut max_ts: Option<u64> = None;
158        let mut clean_seqs: BTreeSet<u64> = BTreeSet::new();
159        let mut corrupted_seqs: BTreeSet<u64> = BTreeSet::new();
160        let mut gap_cursor: Option<u64> = None;
161
162        for seq in 1..=high_watermark {
163            match self.backend.scan_seq(seq) {
164                Ok(Some(entry)) => {
165                    flush_pending_gap(seq, &mut gap_cursor, findings);
166
167                    // The recomputed hash check inside scan_seq covers the entry
168                    // contents, but the entry's embedded seq is one of those
169                    // contents: a row whose value is moved or duplicated under a
170                    // different table key still hashes correctly. Cross-check the
171                    // table key against the embedded seq so the verifier catches
172                    // that class of corruption rather than reporting a clean run.
173                    if entry.seq != seq {
174                        findings.push(VerifyFinding::SeqMismatch {
175                            table_key: seq,
176                            embedded_seq: entry.seq,
177                        });
178                        corrupted_seqs.insert(seq);
179                        scanned += 1;
180                        continue;
181                    }
182                    record_entry(&entry, &mut min_ts, &mut max_ts);
183                    clean_seqs.insert(seq);
184                    scanned += 1;
185                }
186                Ok(None) | Err(EventStoreError::Gap { .. }) => {
187                    extend_pending_gap(seq, &mut gap_cursor);
188                }
189                Err(EventStoreError::HashMismatch { seq: bad }) => {
190                    flush_pending_gap(seq, &mut gap_cursor, findings);
191                    findings.push(VerifyFinding::HashMismatch { seq: bad });
192                    corrupted_seqs.insert(seq);
193                    scanned += 1;
194                }
195                Err(other) => return Err(VerifyError::Backend(other)),
196            }
197        }
198
199        flush_pending_gap(high_watermark + 1, &mut gap_cursor, findings);
200
201        Ok(EntryScan {
202            scanned,
203            min_ts,
204            max_ts,
205            clean_seqs,
206            corrupted_seqs,
207        })
208    }
209
210    fn cross_check_indices(
211        &self,
212        scan: &EntryScan,
213        findings: &mut Vec<VerifyFinding>,
214    ) -> Result<(), VerifyError> {
215        for kind in [IndexKind::ClientOrderId, IndexKind::VenueOrderId] {
216            for (key, stored_seq) in self.backend.iter_index_keys(kind)? {
217                let drift = classify_target(stored_seq, scan);
218                if let Some(drift) = drift {
219                    findings.push(VerifyFinding::IndexDrift { kind, key, drift });
220                }
221            }
222        }
223
224        Ok(())
225    }
226}
227
228#[derive(Debug)]
229struct EntryScan {
230    scanned: u64,
231    min_ts: Option<u64>,
232    max_ts: Option<u64>,
233    clean_seqs: BTreeSet<u64>,
234    corrupted_seqs: BTreeSet<u64>,
235}
236
237fn record_entry(entry: &EventStoreEntry, min_ts: &mut Option<u64>, max_ts: &mut Option<u64>) {
238    let ts = entry.ts_init.as_u64();
239    *min_ts = Some(min_ts.map_or(ts, |cur| cur.min(ts)));
240    *max_ts = Some(max_ts.map_or(ts, |cur| cur.max(ts)));
241}
242
243fn extend_pending_gap(seq: u64, gap_cursor: &mut Option<u64>) {
244    if gap_cursor.is_none() {
245        *gap_cursor = Some(seq);
246    }
247}
248
249fn flush_pending_gap(
250    next_seq: u64,
251    gap_cursor: &mut Option<u64>,
252    findings: &mut Vec<VerifyFinding>,
253) {
254    if let Some(start) = gap_cursor.take() {
255        findings.push(VerifyFinding::Gap {
256            range: GapRange {
257                from: start,
258                to: next_seq - 1,
259            },
260        });
261    }
262}
263
264fn classify_target(stored_seq: u64, scan: &EntryScan) -> Option<IndexDrift> {
265    if scan.clean_seqs.contains(&stored_seq) {
266        None
267    } else if scan.corrupted_seqs.contains(&stored_seq) {
268        Some(IndexDrift::TargetCorrupted { stored_seq })
269    } else {
270        Some(IndexDrift::DanglingTarget { stored_seq })
271    }
272}
273
274// The restore path reads the snapshot anchor before tail replay, so an anchor that
275// fails to decode or points past the durable watermark must not verify clean. Other
276// read failures (disk pressure, storage errors) propagate: suppressing them would
277// pass a run whose restore would fail reading the same anchor.
278fn check_snapshot_anchor(
279    backend: &dyn EventStore,
280    high_watermark: u64,
281    findings: &mut Vec<VerifyFinding>,
282) -> Result<(), VerifyError> {
283    match backend.latest_snapshot_anchor() {
284        Ok(Some(anchor)) if anchor.high_watermark > high_watermark => {
285            findings.push(VerifyFinding::SnapshotAnchorInvalid {
286                reason: format!(
287                    "snapshot anchor high_watermark {} exceeds durable high_watermark {high_watermark}",
288                    anchor.high_watermark,
289                ),
290            });
291        }
292        Ok(_) => {}
293        Err(EventStoreError::Corrupted(msg)) => {
294            findings.push(VerifyFinding::SnapshotAnchorInvalid {
295                reason: format!("snapshot anchor unreadable: {msg}"),
296            });
297        }
298        Err(other) => return Err(VerifyError::Backend(other)),
299    }
300    Ok(())
301}
302
303fn validate_manifest(
304    manifest: &RunManifest,
305    high_watermark: u64,
306    scan: &EntryScan,
307    findings: &mut Vec<VerifyFinding>,
308) {
309    if manifest.high_watermark != high_watermark {
310        findings.push(VerifyFinding::ManifestMismatch {
311            kind: ManifestField::HighWatermark,
312            reason: format!(
313                "manifest high_watermark {} disagrees with durable high_watermark {high_watermark}",
314                manifest.high_watermark,
315            ),
316        });
317    }
318
319    if let Some(min_ts) = scan.min_ts
320        && manifest.start_ts_init.as_u64() > min_ts
321    {
322        findings.push(VerifyFinding::ManifestMismatch {
323            kind: ManifestField::StartTsInit,
324            reason: format!(
325                "manifest start_ts_init {} sits above earliest entry ts_init {min_ts}",
326                manifest.start_ts_init.as_u64(),
327            ),
328        });
329    }
330
331    if manifest.is_sealed() {
332        match (manifest.end_ts_init.map(|t| t.as_u64()), scan.max_ts) {
333            (Some(stored), Some(observed)) if stored != observed => {
334                findings.push(VerifyFinding::ManifestMismatch {
335                    kind: ManifestField::EndTsInit,
336                    reason: format!(
337                        "manifest end_ts_init {stored} disagrees with last observed ts_init {observed}",
338                    ),
339                });
340            }
341            (None, Some(observed)) => findings.push(VerifyFinding::ManifestMismatch {
342                kind: ManifestField::EndTsInit,
343                reason: format!(
344                    "sealed manifest is missing end_ts_init while entries up to ts_init {observed} exist",
345                ),
346            }),
347            (Some(stored), None) => findings.push(VerifyFinding::ManifestMismatch {
348                kind: ManifestField::EndTsInit,
349                reason: format!(
350                    "sealed manifest carries end_ts_init {stored} despite empty entry table",
351                ),
352            }),
353            _ => {}
354        }
355    }
356}
357
358/// The structured report produced by [`Verifier::verify`].
359///
360/// Operators key on [`VerifyReport::is_clean`] for the binary verdict and walk
361/// [`VerifyReport::findings`] for the actionable items. The verifier never
362/// quarantines on its own: that is the supervisor's call given the report.
363#[derive(Debug, Clone, PartialEq, Eq)]
364pub struct VerifyReport {
365    /// The id of the verified run, copied from the manifest.
366    pub run_id: RunId,
367    /// The lifecycle status the run carried at verification time.
368    pub status: RunStatus,
369    /// The durable high-watermark the verifier walked up to.
370    pub high_watermark: u64,
371    /// The number of `seq` slots the verifier successfully read (clean or hash-mismatched).
372    pub entries_scanned: u64,
373    /// Every integrity finding the verifier accumulated.
374    pub findings: Vec<VerifyFinding>,
375}
376
377impl VerifyReport {
378    /// Returns `true` when the verifier accumulated no findings.
379    #[must_use]
380    pub fn is_clean(&self) -> bool {
381        self.findings.is_empty()
382    }
383}
384
385/// One actionable integrity finding from a verifier run.
386#[derive(Debug, Clone, PartialEq, Eq)]
387pub enum VerifyFinding {
388    /// The recomputed canonical hash of `seq` did not match the stored value.
389    HashMismatch {
390        /// The sequence number whose hash diverged.
391        seq: u64,
392    },
393    /// One or more contiguous `seq` slots inside the high-watermark are missing.
394    Gap {
395        /// The inclusive range of missing seqs.
396        range: GapRange,
397    },
398    /// The entry stored at table key `table_key` carries an `entry.seq` that
399    /// disagrees with the key.
400    ///
401    /// The canonical hash hashes `entry.seq` rather than the table key, so a row
402    /// whose bytes were moved or duplicated under a different key still passes the
403    /// hash check. The verifier surfaces the divergence so that class of
404    /// corruption never reads as clean.
405    SeqMismatch {
406        /// The redb table key (the slot the verifier was reading).
407        table_key: u64,
408        /// The seq embedded inside the decoded entry value.
409        embedded_seq: u64,
410    },
411    /// A stored sidecar index entry diverges from the projection rebuilt from the
412    /// entry table.
413    IndexDrift {
414        /// Which sidecar index the finding applies to.
415        kind: IndexKind,
416        /// The stringified key inside that index.
417        key: String,
418        /// The kind of drift observed.
419        drift: IndexDrift,
420    },
421    /// A manifest field disagrees with the entry table or violates a sealed-state
422    /// invariant.
423    ManifestMismatch {
424        /// Which manifest field the finding applies to.
425        kind: ManifestField,
426        /// Operator-readable explanation of the mismatch.
427        reason: String,
428    },
429    /// The recorded snapshot anchor cannot support a restore: it fails to decode or
430    /// points past the durable high-watermark.
431    SnapshotAnchorInvalid {
432        /// Operator-readable explanation of the failure.
433        reason: String,
434    },
435}
436
437/// An inclusive `[from, to]` range of missing seqs.
438#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
439pub struct GapRange {
440    /// First missing seq.
441    pub from: u64,
442    /// Last missing seq.
443    pub to: u64,
444}
445
446/// The kind of drift observed for a sidecar index key.
447///
448/// Today the verifier only reports target reachability for the `client_order_id` and
449/// `venue_order_id` indices because the rebuild is not yet payload-aware. Variants for
450/// rebuild-vs-stored mismatches (missing from stored, divergent seq, unknown key) will
451/// land when wrapper-type encoders provide a payload-derived projection.
452#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
453pub enum IndexDrift {
454    /// The stored index points at a seq that does not exist inside the high-watermark.
455    DanglingTarget {
456        /// The seq the stored index recorded.
457        stored_seq: u64,
458    },
459    /// The stored index points at a seq whose entry failed the hash check.
460    TargetCorrupted {
461        /// The seq the stored index recorded.
462        stored_seq: u64,
463    },
464}
465
466/// A manifest field flagged by [`VerifyFinding::ManifestMismatch`].
467#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
468pub enum ManifestField {
469    /// `manifest.high_watermark` does not match the durable last seq.
470    HighWatermark,
471    /// `manifest.start_ts_init` sits above the earliest observed `ts_init`.
472    StartTsInit,
473    /// `manifest.end_ts_init` does not bracket the entry stream as expected for a
474    /// sealed run.
475    EndTsInit,
476}
477
478/// Errors that prevent the verifier from producing a report at all.
479///
480/// Findings on a successful report cover the operator's actionable surface; this
481/// type captures the verifier's own failure modes (no run open, disk pressure on a
482/// read, manifest header damage that prevents loading the manifest).
483#[derive(Debug, thiserror::Error)]
484pub enum VerifyError {
485    /// A backend operation refused service before the verifier could produce a report.
486    #[error("backend access failed: {0}")]
487    Backend(#[from] EventStoreError),
488}
489
490#[cfg(test)]
491mod tests {
492    use bytes::Bytes;
493    use indexmap::IndexMap;
494    use nautilus_core::UnixNanos;
495    use rstest::{fixture, rstest};
496    use ustr::Ustr;
497
498    use super::*;
499    use crate::{
500        backend::{AppendEntry, IndexKey, MemoryBackend, ScanDirection},
501        compute_entry_hash,
502        entry::Topic,
503        headers::Headers,
504        manifest::{RegisteredComponents, RunManifest, RunStatus},
505    };
506
507    fn manifest(run_id: &str) -> RunManifest {
508        RunManifest {
509            run_id: run_id.to_string(),
510            parent_run_id: None,
511            instance_id: "trader-001".to_string(),
512            binary_hash: "deadbeef".to_string(),
513            schema_version: 1,
514            crate_versions: "feedface".to_string(),
515            feature_flags: Vec::new(),
516            adapter_versions: IndexMap::new(),
517            config_hash: "cafebabe".to_string(),
518            registered_components: RegisteredComponents::default(),
519            seed: None,
520            start_ts_init: UnixNanos::from(0),
521            end_ts_init: None,
522            high_watermark: 0,
523            status: RunStatus::Running,
524        }
525    }
526
527    fn build_entry(seq: u64, headers: Headers, ts_init: u64) -> EventStoreEntry {
528        let topic: Topic = "exec.command.SubmitOrder".into();
529        let payload_type = Ustr::from("SubmitOrder");
530        let payload = Bytes::from_static(b"\x01\x02\x03\x04");
531        let ts_publish = UnixNanos::from(ts_init + 1);
532        let ts_init = UnixNanos::from(ts_init);
533        let hash = compute_entry_hash(
534            seq,
535            ts_init,
536            ts_publish,
537            topic.as_ref(),
538            payload_type.as_str(),
539            &payload,
540            &headers,
541        );
542
543        EventStoreEntry::new(
544            hash,
545            seq,
546            headers,
547            topic,
548            payload_type,
549            payload,
550            ts_init,
551            ts_publish,
552        )
553    }
554
555    fn append_with(seq: u64, ts_init: u64, index_keys: Vec<IndexKey>) -> AppendEntry {
556        AppendEntry::new(build_entry(seq, Headers::empty(), ts_init), index_keys)
557    }
558
559    /// Test-only wrapper that delegates every call to an inner backend except the
560    /// manifest, which it returns verbatim, and (optionally) the high-watermark.
561    /// Lets unit tests drive manifest-mismatch and trailing-gap findings the
562    /// public `MemoryBackend` API would normalize away on seal.
563    struct ManifestOverrideBackend {
564        inner: MemoryBackend,
565        manifest_override: RunManifest,
566        high_watermark_override: Option<u64>,
567    }
568
569    impl ManifestOverrideBackend {
570        fn new(inner: MemoryBackend, manifest_override: RunManifest) -> Self {
571            Self {
572                inner,
573                manifest_override,
574                high_watermark_override: None,
575            }
576        }
577
578        fn with_high_watermark(mut self, hwm: u64) -> Self {
579            self.high_watermark_override = Some(hwm);
580            self
581        }
582    }
583
584    impl EventStore for ManifestOverrideBackend {
585        fn open_run(&mut self, m: RunManifest) -> Result<(), EventStoreError> {
586            self.inner.open_run(m)
587        }
588
589        fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
590            self.inner.append_batch(entries)
591        }
592
593        fn scan_range(
594            &self,
595            from: u64,
596            to: u64,
597            direction: ScanDirection,
598        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
599            self.inner.scan_range(from, to, direction)
600        }
601
602        fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
603            self.inner.scan_seq(seq)
604        }
605
606        fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
607            self.inner.lookup(kind, key)
608        }
609
610        fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
611            self.inner.iter_index_keys(kind)
612        }
613
614        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
615            self.inner.seal(status)
616        }
617
618        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
619            Ok(self.manifest_override.clone())
620        }
621
622        fn high_watermark(&self) -> Result<u64, EventStoreError> {
623            if let Some(hwm) = self.high_watermark_override {
624                return Ok(hwm);
625            }
626            self.inner.high_watermark()
627        }
628    }
629
630    #[fixture]
631    fn open_backend() -> MemoryBackend {
632        let mut backend = MemoryBackend::new();
633        backend
634            .open_run(manifest("1700000000-aaaa1111"))
635            .expect("open run");
636        backend
637    }
638
639    fn verifier_for(backend: MemoryBackend) -> Verifier {
640        Verifier::new(Box::new(backend))
641    }
642
643    #[rstest]
644    fn clean_run_reports_no_findings(mut open_backend: MemoryBackend) {
645        open_backend
646            .append_batch(&[
647                append_with(1, 10, Vec::new()),
648                append_with(2, 11, Vec::new()),
649                append_with(3, 12, Vec::new()),
650            ])
651            .expect("append");
652        open_backend.seal(RunStatus::Ended).expect("seal");
653
654        let report = verifier_for(open_backend).verify().expect("verify");
655
656        // Lock the canonical clean case to zero findings exactly: any spurious
657        // additional finding must fail this test rather than slip past is_clean()
658        // matchers in the more targeted suites below.
659        assert!(report.is_clean(), "findings was: {:?}", report.findings);
660        assert_eq!(report.findings.len(), 0);
661        assert_eq!(report.high_watermark, 3);
662        assert_eq!(report.entries_scanned, 3);
663        assert_eq!(report.status, RunStatus::Ended);
664    }
665
666    #[rstest]
667    fn empty_run_reports_no_findings(mut open_backend: MemoryBackend) {
668        open_backend.seal(RunStatus::Ended).expect("seal");
669
670        let report = verifier_for(open_backend).verify().expect("verify");
671
672        assert!(report.is_clean(), "findings was: {:?}", report.findings);
673        assert_eq!(report.entries_scanned, 0);
674    }
675
676    #[rstest]
677    fn hash_mismatch_surfaces_per_seq(mut open_backend: MemoryBackend) {
678        open_backend
679            .append_batch(&[append_with(1, 10, Vec::new())])
680            .expect("append");
681        let mut tampered = build_entry(2, Headers::empty(), 11);
682        tampered.payload = Bytes::from_static(b"\xFF");
683        open_backend
684            .append_batch(&[AppendEntry::without_indices(tampered)])
685            .expect("append");
686        open_backend
687            .append_batch(&[append_with(3, 12, Vec::new())])
688            .expect("append");
689
690        let report = verifier_for(open_backend).verify().expect("verify");
691
692        assert!(
693            report
694                .findings
695                .iter()
696                .any(|f| matches!(f, VerifyFinding::HashMismatch { seq: 2 })),
697            "findings was: {:?}",
698            report.findings,
699        );
700        assert_eq!(report.entries_scanned, 3);
701        assert_eq!(report.high_watermark, 3);
702    }
703
704    #[rstest]
705    fn multiple_hash_mismatches_all_surface(mut open_backend: MemoryBackend) {
706        // Confirms the verifier walks past hash mismatches instead of bailing on the
707        // first hit: seq=2 and seq=4 are both tampered, and both must appear in the
708        // report.
709        for seq in 1..=4u64 {
710            let mut entry = build_entry(seq, Headers::empty(), 10 + seq);
711            if seq == 2 || seq == 4 {
712                entry.payload = Bytes::from_static(b"\xFF");
713            }
714            open_backend
715                .append_batch(&[AppendEntry::without_indices(entry)])
716                .expect("append");
717        }
718
719        let report = verifier_for(open_backend).verify().expect("verify");
720
721        let mismatch_seqs: Vec<u64> = report
722            .findings
723            .iter()
724            .filter_map(|f| match f {
725                VerifyFinding::HashMismatch { seq } => Some(*seq),
726                _ => None,
727            })
728            .collect();
729        assert_eq!(mismatch_seqs, vec![2, 4]);
730    }
731
732    #[rstest]
733    fn client_order_id_index_clean_when_target_resolves(mut open_backend: MemoryBackend) {
734        open_backend
735            .append_batch(&[AppendEntry::new(
736                build_entry(1, Headers::empty(), 10),
737                vec![IndexKey::new(IndexKind::ClientOrderId, "O-1".to_string())],
738            )])
739            .expect("append");
740        open_backend.seal(RunStatus::Ended).expect("seal");
741
742        let report = verifier_for(open_backend).verify().expect("verify");
743
744        assert!(report.is_clean(), "findings was: {:?}", report.findings);
745    }
746
747    #[rstest]
748    #[case::client_order_id(IndexKind::ClientOrderId)]
749    #[case::venue_order_id(IndexKind::VenueOrderId)]
750    fn entity_index_target_corrupted_drift(
751        mut open_backend: MemoryBackend,
752        #[case] kind: IndexKind,
753    ) {
754        // Stored entity-index entry points at seq=1 whose stored hash no longer
755        // matches the recomputed hash. The verifier must surface TargetCorrupted
756        // for both ClientOrderId and VenueOrderId so a drop of either kind from
757        // the cross-check loop fails this test.
758        let mut tampered = build_entry(1, Headers::empty(), 10);
759        tampered.payload = Bytes::from_static(b"\xFF");
760        open_backend
761            .append_batch(&[AppendEntry::new(
762                tampered,
763                vec![IndexKey::new(kind, "K-1".to_string())],
764            )])
765            .expect("append");
766
767        let report = verifier_for(open_backend).verify().expect("verify");
768
769        assert!(
770            report.findings.iter().any(|f| matches!(
771                f,
772                VerifyFinding::IndexDrift {
773                    kind: drift_kind,
774                    drift: IndexDrift::TargetCorrupted { stored_seq: 1 },
775                    ..
776                } if *drift_kind == kind
777            )),
778            "findings was: {:?}",
779            report.findings,
780        );
781    }
782
783    fn find_manifest_mismatch(findings: &[VerifyFinding], target: ManifestField) -> &str {
784        findings
785            .iter()
786            .find_map(|f| match f {
787                VerifyFinding::ManifestMismatch { kind, reason } if *kind == target => {
788                    Some(reason.as_str())
789                }
790                _ => None,
791            })
792            .unwrap_or_else(|| {
793                panic!("expected ManifestMismatch({target:?}), findings was: {findings:?}")
794            })
795    }
796
797    #[rstest]
798    fn manifest_high_watermark_drift() {
799        // Real durable hwm is 1, but the manifest reports 99. Verifier must surface
800        // a HighWatermark mismatch whose reason carries both values so a swap of
801        // observed and stored sides would fail this test.
802        let mut inner = MemoryBackend::new();
803        inner.open_run(manifest("run-hwm")).expect("open run");
804        inner
805            .append_batch(&[append_with(1, 10, Vec::new())])
806            .expect("append");
807        inner.seal(RunStatus::Ended).expect("seal");
808
809        let mut stale = inner.manifest().expect("manifest");
810        stale.high_watermark = 99;
811        let backend = ManifestOverrideBackend::new(inner, stale);
812
813        let report = Verifier::new(Box::new(backend)).verify().expect("verify");
814        let reason = find_manifest_mismatch(&report.findings, ManifestField::HighWatermark);
815
816        assert!(reason.contains("99"), "reason was: {reason}");
817        assert!(reason.contains('1'), "reason was: {reason}");
818    }
819
820    #[rstest]
821    fn manifest_end_ts_init_drift_when_sealed() {
822        // Real durable max ts_init is 25, but the sealed manifest's end_ts_init
823        // claims 99. The reason must surface both values; without that assertion,
824        // a min/max swap inside record_entry would still pass.
825        let mut inner = MemoryBackend::new();
826        inner.open_run(manifest("run-end-ts")).expect("open run");
827        inner
828            .append_batch(&[
829                append_with(1, 10, Vec::new()),
830                append_with(2, 25, Vec::new()),
831            ])
832            .expect("append");
833        inner.seal(RunStatus::Ended).expect("seal");
834
835        let mut drifted = inner.manifest().expect("manifest");
836        drifted.end_ts_init = Some(UnixNanos::from(99));
837        let backend = ManifestOverrideBackend::new(inner, drifted);
838
839        let report = Verifier::new(Box::new(backend)).verify().expect("verify");
840        let reason = find_manifest_mismatch(&report.findings, ManifestField::EndTsInit);
841
842        assert!(reason.contains("99"), "reason was: {reason}");
843        assert!(reason.contains("25"), "reason was: {reason}");
844    }
845
846    #[rstest]
847    fn manifest_end_ts_init_missing_when_sealed_with_entries() {
848        // Sealed manifest forgot to record end_ts_init while the entry stream is
849        // non-empty: validate_manifest's (None, Some) arm must fire and the
850        // reason must carry the observed last ts_init.
851        let mut inner = MemoryBackend::new();
852        inner
853            .open_run(manifest("run-end-ts-missing"))
854            .expect("open run");
855        inner
856            .append_batch(&[append_with(1, 42, Vec::new())])
857            .expect("append");
858        inner.seal(RunStatus::Ended).expect("seal");
859
860        let mut drifted = inner.manifest().expect("manifest");
861        drifted.end_ts_init = None;
862        let backend = ManifestOverrideBackend::new(inner, drifted);
863
864        let report = Verifier::new(Box::new(backend)).verify().expect("verify");
865        let reason = find_manifest_mismatch(&report.findings, ManifestField::EndTsInit);
866
867        assert!(reason.contains("missing"), "reason was: {reason}");
868        assert!(reason.contains("42"), "reason was: {reason}");
869    }
870
871    #[rstest]
872    fn manifest_end_ts_init_set_on_sealed_empty_run() {
873        // Sealed manifest carries end_ts_init even though the entry table is
874        // empty: validate_manifest's (Some, None) arm must fire and the reason
875        // must carry the spurious stored value.
876        let mut inner = MemoryBackend::new();
877        inner
878            .open_run(manifest("run-end-ts-empty"))
879            .expect("open run");
880        inner.seal(RunStatus::Ended).expect("seal");
881
882        let mut drifted = inner.manifest().expect("manifest");
883        drifted.end_ts_init = Some(UnixNanos::from(77));
884        let backend = ManifestOverrideBackend::new(inner, drifted);
885
886        let report = Verifier::new(Box::new(backend)).verify().expect("verify");
887        let reason = find_manifest_mismatch(&report.findings, ManifestField::EndTsInit);
888
889        assert!(reason.contains("77"), "reason was: {reason}");
890        assert!(reason.contains("empty"), "reason was: {reason}");
891    }
892
893    #[rstest]
894    fn manifest_start_ts_init_drift() {
895        // Earliest entry ts_init is 10, but the manifest's start_ts_init is 50.
896        // Reason must carry both values so a flipped comparison or wrong-side
897        // formatting fails the test.
898        let mut inner = MemoryBackend::new();
899        inner.open_run(manifest("run-start-ts")).expect("open run");
900        inner
901            .append_batch(&[
902                append_with(1, 10, Vec::new()),
903                append_with(2, 25, Vec::new()),
904            ])
905            .expect("append");
906        inner.seal(RunStatus::Ended).expect("seal");
907
908        let mut drifted = inner.manifest().expect("manifest");
909        drifted.start_ts_init = UnixNanos::from(50);
910        let backend = ManifestOverrideBackend::new(inner, drifted);
911
912        let report = Verifier::new(Box::new(backend)).verify().expect("verify");
913        let reason = find_manifest_mismatch(&report.findings, ManifestField::StartTsInit);
914
915        assert!(reason.contains("50"), "reason was: {reason}");
916        assert!(reason.contains("10"), "reason was: {reason}");
917    }
918
919    #[rstest]
920    fn trailing_gap_surfaces_when_last_seqs_missing() {
921        // Inner backend holds seqs 1..=3, but both the manifest and the advertised
922        // high-watermark claim 5. The verifier must walk to seq=5, find seqs 4-5
923        // missing, and emit a single trailing GapRange{4,5}. Removing the
924        // `flush_pending_gap(high_watermark + 1, ...)` call after the loop would
925        // drop this finding entirely.
926        let mut inner = MemoryBackend::new();
927        inner
928            .open_run(manifest("run-trailing-gap"))
929            .expect("open run");
930        inner
931            .append_batch(&[
932                append_with(1, 10, Vec::new()),
933                append_with(2, 11, Vec::new()),
934                append_with(3, 12, Vec::new()),
935            ])
936            .expect("append");
937        inner.seal(RunStatus::Ended).expect("seal");
938
939        let mut drifted = inner.manifest().expect("manifest");
940        drifted.high_watermark = 5;
941        // Advertise hwm=5 on both sides so the HighWatermark mismatch path stays
942        // quiet and the test pins only the trailing-gap behavior.
943        let backend = ManifestOverrideBackend::new(inner, drifted).with_high_watermark(5);
944
945        let report = Verifier::new(Box::new(backend)).verify().expect("verify");
946
947        let gaps: Vec<GapRange> = report
948            .findings
949            .iter()
950            .filter_map(|f| match f {
951                VerifyFinding::Gap { range } => Some(*range),
952                _ => None,
953            })
954            .collect();
955        assert_eq!(gaps, vec![GapRange { from: 4, to: 5 }]);
956        assert_eq!(report.entries_scanned, 3);
957        assert_eq!(report.high_watermark, 5);
958    }
959
960    /// Test backend that rewrites a single `scan_seq` result so the value's
961    /// embedded seq disagrees with the requested table key. Lets the unit suite
962    /// exercise the redb-only "row moved under wrong key" corruption class
963    /// without setting up a real on-disk file.
964    struct SeqRewriteBackend {
965        inner: MemoryBackend,
966        target_key: u64,
967        substitute: EventStoreEntry,
968    }
969
970    impl EventStore for SeqRewriteBackend {
971        fn open_run(&mut self, m: RunManifest) -> Result<(), EventStoreError> {
972            self.inner.open_run(m)
973        }
974        fn append_batch(&mut self, e: &[AppendEntry]) -> Result<u64, EventStoreError> {
975            self.inner.append_batch(e)
976        }
977        fn scan_range(
978            &self,
979            from: u64,
980            to: u64,
981            direction: ScanDirection,
982        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
983            self.inner.scan_range(from, to, direction)
984        }
985        fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
986            if seq == self.target_key {
987                return Ok(Some(self.substitute.clone()));
988            }
989            self.inner.scan_seq(seq)
990        }
991        fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
992            self.inner.lookup(kind, key)
993        }
994        fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
995            self.inner.iter_index_keys(kind)
996        }
997        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
998            self.inner.seal(status)
999        }
1000        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
1001            self.inner.manifest()
1002        }
1003        fn high_watermark(&self) -> Result<u64, EventStoreError> {
1004            self.inner.high_watermark()
1005        }
1006    }
1007
1008    #[rstest]
1009    fn seq_mismatch_surfaces_when_row_value_disagrees_with_key() {
1010        // Row at table_key=2 holds the bytes of an entry whose embedded seq is 99.
1011        // The hash recomputes correctly (because the hash covers entry.seq=99),
1012        // so scan_seq returns Ok(Some(entry)) without raising HashMismatch. The
1013        // verifier must catch the key/embedded-seq divergence rather than mark
1014        // the slot clean.
1015        let mut inner = MemoryBackend::new();
1016        inner
1017            .open_run(manifest("run-seq-mismatch"))
1018            .expect("open run");
1019        inner
1020            .append_batch(&[
1021                append_with(1, 10, Vec::new()),
1022                append_with(2, 11, Vec::new()),
1023                append_with(3, 12, Vec::new()),
1024            ])
1025            .expect("append");
1026        inner.seal(RunStatus::Ended).expect("seal");
1027
1028        let substitute = build_entry(99, Headers::empty(), 11);
1029        let backend = SeqRewriteBackend {
1030            inner,
1031            target_key: 2,
1032            substitute,
1033        };
1034
1035        let report = Verifier::new(Box::new(backend)).verify().expect("verify");
1036
1037        assert!(
1038            report.findings.iter().any(|f| matches!(
1039                f,
1040                VerifyFinding::SeqMismatch {
1041                    table_key: 2,
1042                    embedded_seq: 99,
1043                }
1044            )),
1045            "findings was: {:?}",
1046            report.findings,
1047        );
1048    }
1049
1050    #[rstest]
1051    fn seq_mismatch_marks_target_corrupted_for_dependent_indices() {
1052        // Same row corruption as above, but the stored client_order_id index
1053        // points at the rewritten slot. The slot must be classified as corrupted
1054        // so the index drift surfaces TargetCorrupted rather than silently
1055        // accepting the lookup.
1056        let mut inner = MemoryBackend::new();
1057        inner
1058            .open_run(manifest("run-seq-mismatch-idx"))
1059            .expect("open run");
1060        inner
1061            .append_batch(&[
1062                append_with(1, 10, Vec::new()),
1063                AppendEntry::new(
1064                    build_entry(2, Headers::empty(), 11),
1065                    vec![IndexKey::new(IndexKind::ClientOrderId, "O-1".to_string())],
1066                ),
1067            ])
1068            .expect("append");
1069        inner.seal(RunStatus::Ended).expect("seal");
1070
1071        let substitute = build_entry(99, Headers::empty(), 11);
1072        let backend = SeqRewriteBackend {
1073            inner,
1074            target_key: 2,
1075            substitute,
1076        };
1077
1078        let report = Verifier::new(Box::new(backend)).verify().expect("verify");
1079
1080        assert!(
1081            report.findings.iter().any(|f| matches!(
1082                f,
1083                VerifyFinding::IndexDrift {
1084                    kind: IndexKind::ClientOrderId,
1085                    drift: IndexDrift::TargetCorrupted { stored_seq: 2 },
1086                    ..
1087                }
1088            )),
1089            "findings was: {:?}",
1090            report.findings,
1091        );
1092    }
1093
1094    #[rstest]
1095    fn verify_propagates_no_run_open_as_error() {
1096        let backend = MemoryBackend::new();
1097        let verifier = Verifier::new(Box::new(backend));
1098
1099        let err = verifier.verify().expect_err("must fail");
1100
1101        match err {
1102            VerifyError::Backend(EventStoreError::Backend(msg)) => {
1103                assert!(msg.contains("no run open"), "msg was: {msg}");
1104            }
1105            VerifyError::Backend(other) => {
1106                panic!("expected Backend(no run open), was {other:?}")
1107            }
1108        }
1109    }
1110}