nautilus_event_store/markers/
verifier.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//! Non-fatal integrity verifier for data marker sidecars.
17
18use std::cmp::Ordering;
19
20use ahash::AHashMap;
21use nautilus_core::UnixNanos;
22
23use crate::{
24    error::EventStoreError,
25    manifest::{RunId, RunStatus},
26    markers::{
27        DataCursorSnapshot, HiFiMarker, MarkerBackend, MarkerGap, MarkerManifest, StreamCursor,
28        StreamDictEntry, StreamSlot, compute_dict_hash, compute_gap_hash, compute_hifi_hash,
29        compute_marker_hash,
30    },
31};
32
33/// Verifier for a single marker sidecar backend.
34#[derive(Debug, Default)]
35pub struct MarkerVerifier;
36
37impl MarkerVerifier {
38    /// Scans the marker sidecar and returns all non-fatal findings.
39    ///
40    /// `entry_high_watermark` comes from the sibling entry run and bounds marker
41    /// `event_seq_before` values.
42    ///
43    /// # Errors
44    ///
45    /// Returns [`EventStoreError`] when the backend cannot scan a marker table or manifest.
46    pub fn scan(
47        backend: &dyn MarkerBackend,
48        entry_high_watermark: u64,
49    ) -> Result<MarkerVerifyReport, EventStoreError> {
50        let manifest = backend.manifest()?;
51        let mut findings = Vec::new();
52        let snapshots = read_snapshots(backend, &mut findings)?;
53        let hifi = read_hifi(backend, &mut findings)?;
54        let gaps = read_gaps(backend, &mut findings)?;
55        let dict = read_dict(backend, &mut findings)?;
56
57        check_manifest_counts(&manifest, &snapshots, &hifi, &gaps, &dict, &mut findings);
58        check_marker_sequence(&snapshots, &hifi, &gaps, &mut findings);
59        check_event_seq(&snapshots, &hifi, entry_high_watermark, &mut findings);
60        check_cursor_monotonicity(&snapshots, &mut findings);
61
62        Ok(MarkerVerifyReport {
63            run_id: manifest.run_id,
64            status: manifest.status,
65            snapshots_scanned: snapshots.len() as u64,
66            hifi_scanned: hifi.len() as u64,
67            gaps_scanned: gaps.len() as u64,
68            dict_entries_scanned: dict.len() as u64,
69            findings,
70        })
71    }
72}
73
74/// Structured report produced by [`MarkerVerifier`].
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub struct MarkerVerifyReport {
77    /// The id of the run this marker sidecar belongs to.
78    pub run_id: RunId,
79    /// The marker sidecar lifecycle status at verification time.
80    pub status: RunStatus,
81    /// Number of cursor snapshots scanned.
82    pub snapshots_scanned: u64,
83    /// Number of high-fidelity markers scanned.
84    pub hifi_scanned: u64,
85    /// Number of marker gaps scanned.
86    pub gaps_scanned: u64,
87    /// Number of stream dictionary entries scanned.
88    pub dict_entries_scanned: u64,
89    /// Every marker-sidecar integrity finding.
90    pub findings: Vec<MarkerFinding>,
91}
92
93impl MarkerVerifyReport {
94    /// Returns `true` when no marker findings were accumulated.
95    #[must_use]
96    pub fn is_clean(&self) -> bool {
97        self.findings.is_empty()
98    }
99}
100
101/// Durable marker table kind used in hash findings.
102#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
103pub enum MarkerRecordKind {
104    /// Cursor snapshot table.
105    Snapshot,
106    /// High-fidelity marker table.
107    HiFi,
108    /// Marker gap table.
109    Gap,
110    /// Stream dictionary table.
111    Dict,
112}
113
114/// Marker manifest count field used in count-mismatch findings.
115#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
116pub enum MarkerCountKind {
117    /// `snapshot_count`.
118    Snapshot,
119    /// `hifi_count`.
120    HiFi,
121    /// `gap_count`.
122    Gap,
123    /// `dict_count`.
124    Dict,
125}
126
127/// One non-fatal integrity finding from a marker verification scan.
128#[derive(Debug, Clone, PartialEq, Eq)]
129pub enum MarkerFinding {
130    /// A marker manifest count disagrees with the scanned table row count.
131    ManifestCountMismatch {
132        /// Which manifest count diverged.
133        kind: MarkerCountKind,
134        /// The count recorded in the marker manifest.
135        manifest_count: u64,
136        /// The count observed by scanning the marker table.
137        scanned_count: u64,
138    },
139    /// One or more marker sequence values were neither present nor covered by a gap.
140    MarkerSeqGap {
141        /// First missing marker sequence.
142        from_marker_seq: u64,
143        /// Last missing marker sequence.
144        to_marker_seq: u64,
145    },
146    /// A marker sequence value was covered more than once.
147    MarkerSeqOverlap {
148        /// First overlapped marker sequence.
149        from_marker_seq: u64,
150        /// Last overlapped marker sequence.
151        to_marker_seq: u64,
152    },
153    /// A stored gap has an invalid inclusive range.
154    InvalidMarkerGap {
155        /// The gap's first marker sequence.
156        from_marker_seq: u64,
157        /// The gap's last marker sequence.
158        to_marker_seq: u64,
159    },
160    /// `event_seq_before` decreased as `marker_seq` advanced.
161    EventSeqRegressed {
162        /// The marker sequence where the regression was observed.
163        marker_seq: u64,
164        /// The previous event sequence boundary.
165        previous_event_seq_before: u64,
166        /// The current event sequence boundary.
167        event_seq_before: u64,
168    },
169    /// `event_seq_before` exceeded the sibling entry run high-watermark.
170    EventSeqExceedsHighWatermark {
171        /// The marker sequence carrying the invalid event boundary.
172        marker_seq: u64,
173        /// The invalid event sequence boundary.
174        event_seq_before: u64,
175        /// The sibling entry run high-watermark.
176        high_watermark: u64,
177    },
178    /// A per-slot cursor count decreased between snapshots.
179    CursorCountRegressed {
180        /// The marker sequence carrying the regressed cursor.
181        marker_seq: u64,
182        /// The stream slot whose cursor regressed.
183        slot: StreamSlot,
184        /// The previous count for this slot.
185        previous_count: u64,
186        /// The current count for this slot.
187        count: u64,
188    },
189    /// A per-slot highest `ts_init` decreased between snapshots.
190    CursorTsInitRegressed {
191        /// The marker sequence carrying the regressed cursor.
192        marker_seq: u64,
193        /// The stream slot whose cursor regressed.
194        slot: StreamSlot,
195        /// The previous highest `ts_init` for this slot.
196        previous_ts_init_hi: UnixNanos,
197        /// The current highest `ts_init` for this slot.
198        ts_init_hi: UnixNanos,
199    },
200    /// A stored durable record hash did not match the recomputed canonical hash.
201    HashMismatch {
202        /// The table kind whose stored hash diverged.
203        record: MarkerRecordKind,
204        /// The marker sequence when the record kind carries one.
205        marker_seq: Option<u64>,
206        /// The stream slot when the record kind carries one.
207        slot: Option<StreamSlot>,
208    },
209}
210
211#[derive(Debug, Clone)]
212struct ScannedRecord<T> {
213    record: T,
214}
215
216#[derive(Debug, Clone, Copy)]
217struct SequencedMarker {
218    marker_seq: u64,
219    event_seq_before: u64,
220}
221
222fn read_snapshots(
223    backend: &dyn MarkerBackend,
224    findings: &mut Vec<MarkerFinding>,
225) -> Result<Vec<ScannedRecord<DataCursorSnapshot>>, EventStoreError> {
226    if let Some(stored) = backend.scan_snapshot_records()? {
227        let out = stored
228            .into_iter()
229            .map(|stored| {
230                check_hash(
231                    stored.hash,
232                    compute_marker_hash(&stored.record),
233                    MarkerRecordKind::Snapshot,
234                    Some(stored.record.marker_seq),
235                    None,
236                    findings,
237                );
238                ScannedRecord {
239                    record: stored.record,
240                }
241            })
242            .collect();
243        return Ok(out);
244    }
245
246    Ok(backend
247        .scan_snapshots()?
248        .into_iter()
249        .map(|record| ScannedRecord { record })
250        .collect())
251}
252
253fn read_hifi(
254    backend: &dyn MarkerBackend,
255    findings: &mut Vec<MarkerFinding>,
256) -> Result<Vec<ScannedRecord<HiFiMarker>>, EventStoreError> {
257    if let Some(stored) = backend.scan_hifi_records()? {
258        let out = stored
259            .into_iter()
260            .map(|stored| {
261                check_hash(
262                    stored.hash,
263                    compute_hifi_hash(&stored.record),
264                    MarkerRecordKind::HiFi,
265                    Some(stored.record.marker_seq),
266                    Some(stored.record.slot),
267                    findings,
268                );
269                ScannedRecord {
270                    record: stored.record,
271                }
272            })
273            .collect();
274        return Ok(out);
275    }
276
277    Ok(backend
278        .scan_hifi()?
279        .into_iter()
280        .map(|record| ScannedRecord { record })
281        .collect())
282}
283
284fn read_gaps(
285    backend: &dyn MarkerBackend,
286    findings: &mut Vec<MarkerFinding>,
287) -> Result<Vec<ScannedRecord<MarkerGap>>, EventStoreError> {
288    if let Some(stored) = backend.scan_gap_records()? {
289        let out = stored
290            .into_iter()
291            .map(|stored| {
292                check_hash(
293                    stored.hash,
294                    compute_gap_hash(&stored.record),
295                    MarkerRecordKind::Gap,
296                    None,
297                    None,
298                    findings,
299                );
300                ScannedRecord {
301                    record: stored.record,
302                }
303            })
304            .collect();
305        return Ok(out);
306    }
307
308    Ok(backend
309        .scan_gaps()?
310        .into_iter()
311        .map(|record| ScannedRecord { record })
312        .collect())
313}
314
315fn read_dict(
316    backend: &dyn MarkerBackend,
317    findings: &mut Vec<MarkerFinding>,
318) -> Result<Vec<ScannedRecord<StreamDictEntry>>, EventStoreError> {
319    if let Some(stored) = backend.scan_dict_records()? {
320        let out = stored
321            .into_iter()
322            .map(|stored| {
323                check_hash(
324                    stored.hash,
325                    compute_dict_hash(&stored.record),
326                    MarkerRecordKind::Dict,
327                    None,
328                    Some(stored.record.slot),
329                    findings,
330                );
331                ScannedRecord {
332                    record: stored.record,
333                }
334            })
335            .collect();
336        return Ok(out);
337    }
338
339    Ok(backend
340        .scan_dict()?
341        .into_iter()
342        .map(|record| ScannedRecord { record })
343        .collect())
344}
345
346fn check_hash(
347    stored_hash: [u8; 32],
348    computed_hash: [u8; 32],
349    record: MarkerRecordKind,
350    marker_seq: Option<u64>,
351    slot: Option<StreamSlot>,
352    findings: &mut Vec<MarkerFinding>,
353) {
354    if stored_hash != computed_hash {
355        findings.push(MarkerFinding::HashMismatch {
356            record,
357            marker_seq,
358            slot,
359        });
360    }
361}
362
363fn check_manifest_counts(
364    manifest: &MarkerManifest,
365    snapshots: &[ScannedRecord<DataCursorSnapshot>],
366    hifi: &[ScannedRecord<HiFiMarker>],
367    gaps: &[ScannedRecord<MarkerGap>],
368    dict: &[ScannedRecord<StreamDictEntry>],
369    findings: &mut Vec<MarkerFinding>,
370) {
371    check_manifest_count(
372        MarkerCountKind::Snapshot,
373        manifest.snapshot_count,
374        snapshots.len() as u64,
375        findings,
376    );
377    check_manifest_count(
378        MarkerCountKind::HiFi,
379        manifest.hifi_count,
380        hifi.len() as u64,
381        findings,
382    );
383    check_manifest_count(
384        MarkerCountKind::Gap,
385        manifest.gap_count,
386        gaps.len() as u64,
387        findings,
388    );
389    check_manifest_count(
390        MarkerCountKind::Dict,
391        manifest.dict_count,
392        dict.len() as u64,
393        findings,
394    );
395}
396
397fn check_manifest_count(
398    kind: MarkerCountKind,
399    manifest_count: u64,
400    scanned_count: u64,
401    findings: &mut Vec<MarkerFinding>,
402) {
403    if manifest_count != scanned_count {
404        findings.push(MarkerFinding::ManifestCountMismatch {
405            kind,
406            manifest_count,
407            scanned_count,
408        });
409    }
410}
411
412fn check_marker_sequence(
413    snapshots: &[ScannedRecord<DataCursorSnapshot>],
414    hifi: &[ScannedRecord<HiFiMarker>],
415    gaps: &[ScannedRecord<MarkerGap>],
416    findings: &mut Vec<MarkerFinding>,
417) {
418    let mut coverages = Vec::new();
419
420    for snapshot in snapshots {
421        coverages.push((snapshot.record.marker_seq, snapshot.record.marker_seq));
422    }
423
424    for marker in hifi {
425        coverages.push((marker.record.marker_seq, marker.record.marker_seq));
426    }
427
428    for gap in gaps {
429        if gap.record.from_marker_seq > gap.record.to_marker_seq {
430            findings.push(MarkerFinding::InvalidMarkerGap {
431                from_marker_seq: gap.record.from_marker_seq,
432                to_marker_seq: gap.record.to_marker_seq,
433            });
434            continue;
435        }
436        coverages.push((gap.record.from_marker_seq, gap.record.to_marker_seq));
437    }
438
439    coverages.sort_unstable_by_key(|(from, to)| (*from, *to));
440    let mut expected = 1_u64;
441
442    for (from, to) in coverages {
443        match from.cmp(&expected) {
444            Ordering::Greater => {
445                findings.push(MarkerFinding::MarkerSeqGap {
446                    from_marker_seq: expected,
447                    to_marker_seq: from - 1,
448                });
449                expected = to.saturating_add(1);
450            }
451            Ordering::Less => {
452                findings.push(MarkerFinding::MarkerSeqOverlap {
453                    from_marker_seq: from,
454                    to_marker_seq: to.min(expected - 1),
455                });
456
457                if to >= expected {
458                    expected = to.saturating_add(1);
459                }
460            }
461            Ordering::Equal => {
462                expected = to.saturating_add(1);
463            }
464        }
465    }
466}
467
468fn check_event_seq(
469    snapshots: &[ScannedRecord<DataCursorSnapshot>],
470    hifi: &[ScannedRecord<HiFiMarker>],
471    entry_high_watermark: u64,
472    findings: &mut Vec<MarkerFinding>,
473) {
474    let mut markers = Vec::with_capacity(snapshots.len() + hifi.len());
475
476    markers.extend(snapshots.iter().map(|snapshot| SequencedMarker {
477        marker_seq: snapshot.record.marker_seq,
478        event_seq_before: snapshot.record.event_seq_before,
479    }));
480    markers.extend(hifi.iter().map(|marker| SequencedMarker {
481        marker_seq: marker.record.marker_seq,
482        event_seq_before: marker.record.event_seq_before,
483    }));
484    markers.sort_unstable_by_key(|marker| marker.marker_seq);
485
486    let mut previous_event_seq_before = None;
487
488    for marker in markers {
489        if marker.event_seq_before > entry_high_watermark {
490            findings.push(MarkerFinding::EventSeqExceedsHighWatermark {
491                marker_seq: marker.marker_seq,
492                event_seq_before: marker.event_seq_before,
493                high_watermark: entry_high_watermark,
494            });
495        }
496
497        if let Some(previous) = previous_event_seq_before
498            && marker.event_seq_before < previous
499        {
500            findings.push(MarkerFinding::EventSeqRegressed {
501                marker_seq: marker.marker_seq,
502                previous_event_seq_before: previous,
503                event_seq_before: marker.event_seq_before,
504            });
505        }
506
507        previous_event_seq_before = Some(marker.event_seq_before);
508    }
509}
510
511fn check_cursor_monotonicity(
512    snapshots: &[ScannedRecord<DataCursorSnapshot>],
513    findings: &mut Vec<MarkerFinding>,
514) {
515    let mut ordered: Vec<&ScannedRecord<DataCursorSnapshot>> = snapshots.iter().collect();
516    ordered.sort_unstable_by_key(|snapshot| snapshot.record.marker_seq);
517    let mut cursors: AHashMap<StreamSlot, StreamCursor> = AHashMap::new();
518
519    for snapshot in ordered {
520        for cursor in &snapshot.record.advanced {
521            if let Some(previous) = cursors.get(&cursor.slot) {
522                if cursor.count < previous.count {
523                    findings.push(MarkerFinding::CursorCountRegressed {
524                        marker_seq: snapshot.record.marker_seq,
525                        slot: cursor.slot,
526                        previous_count: previous.count,
527                        count: cursor.count,
528                    });
529                }
530
531                if cursor.ts_init_hi < previous.ts_init_hi {
532                    findings.push(MarkerFinding::CursorTsInitRegressed {
533                        marker_seq: snapshot.record.marker_seq,
534                        slot: cursor.slot,
535                        previous_ts_init_hi: previous.ts_init_hi,
536                        ts_init_hi: cursor.ts_init_hi,
537                    });
538                }
539            }
540
541            cursors.insert(cursor.slot, cursor.clone());
542        }
543    }
544}
545
546#[cfg(test)]
547mod tests {
548    use rstest::rstest;
549
550    use super::*;
551    use crate::{
552        manifest::RunStatus,
553        markers::{
554            DataClass, DataCursorSnapshot, MarkerBackend, MarkerGap, MarkerGapReason,
555            MarkerManifest, MemoryMarkerBackend, StreamCursor, StreamDictEntry, compute_gap_hash,
556            compute_hifi_hash, compute_marker_hash,
557        },
558    };
559
560    fn manifest() -> MarkerManifest {
561        MarkerManifest {
562            run_id: "1700000000-marker-verifier".to_string(),
563            enabled_classes: vec![DataClass::Quote],
564            high_fidelity: false,
565            snapshot_count: 0,
566            hifi_count: 0,
567            gap_count: 0,
568            dict_count: 0,
569            status: RunStatus::Running,
570        }
571    }
572
573    fn snapshot(
574        marker_seq: u64,
575        event_seq_before: u64,
576        ts_init_hi: u64,
577        count: u64,
578    ) -> DataCursorSnapshot {
579        DataCursorSnapshot {
580            marker_seq,
581            event_seq_before,
582            ts_init: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
583            advanced: vec![StreamCursor {
584                slot: 0,
585                ts_init_hi: UnixNanos::from(ts_init_hi),
586                count,
587            }],
588        }
589    }
590
591    fn hifi(marker_seq: u64, event_seq_before: u64, slot: StreamSlot) -> HiFiMarker {
592        HiFiMarker {
593            marker_seq,
594            event_seq_before,
595            slot,
596            ts_event: UnixNanos::from(1_700_000_000_000_000_100 + marker_seq),
597            ts_init: UnixNanos::from(1_700_000_000_000_000_200 + marker_seq),
598            same_ts_ordinal: 0,
599            record_fingerprint: [7; 32],
600        }
601    }
602
603    fn dict(slot: StreamSlot) -> StreamDictEntry {
604        StreamDictEntry {
605            slot,
606            data_cls: DataClass::Quote,
607            identifier: "ETHUSDT.BINANCE".to_string(),
608        }
609    }
610
611    fn open_backend() -> MemoryMarkerBackend {
612        let mut backend = MemoryMarkerBackend::new();
613        backend.open_run(manifest()).expect("open run");
614        backend
615    }
616
617    #[derive(Debug)]
618    struct ManifestOverrideMarkerBackend {
619        inner: MemoryMarkerBackend,
620        manifest_override: MarkerManifest,
621    }
622
623    impl ManifestOverrideMarkerBackend {
624        fn new(inner: MemoryMarkerBackend, manifest_override: MarkerManifest) -> Self {
625            Self {
626                inner,
627                manifest_override,
628            }
629        }
630    }
631
632    impl MarkerBackend for ManifestOverrideMarkerBackend {
633        fn open_run(&mut self, manifest: MarkerManifest) -> Result<(), EventStoreError> {
634            self.inner.open_run(manifest)
635        }
636
637        fn append_snapshot(
638            &mut self,
639            snapshot: &DataCursorSnapshot,
640            hash: [u8; 32],
641        ) -> Result<(), EventStoreError> {
642            self.inner.append_snapshot(snapshot, hash)
643        }
644
645        fn append_hifi(
646            &mut self,
647            marker: &HiFiMarker,
648            hash: [u8; 32],
649        ) -> Result<(), EventStoreError> {
650            self.inner.append_hifi(marker, hash)
651        }
652
653        fn append_gap(&mut self, gap: &MarkerGap, hash: [u8; 32]) -> Result<(), EventStoreError> {
654            self.inner.append_gap(gap, hash)
655        }
656
657        fn put_dict(
658            &mut self,
659            entry: &StreamDictEntry,
660            hash: [u8; 32],
661        ) -> Result<(), EventStoreError> {
662            self.inner.put_dict(entry, hash)
663        }
664
665        fn scan_snapshots(&self) -> Result<Vec<DataCursorSnapshot>, EventStoreError> {
666            self.inner.scan_snapshots()
667        }
668
669        fn scan_snapshot_records(
670            &self,
671        ) -> Result<
672            Option<Vec<crate::markers::StoredMarkerRecord<DataCursorSnapshot>>>,
673            EventStoreError,
674        > {
675            self.inner.scan_snapshot_records()
676        }
677
678        fn scan_hifi(&self) -> Result<Vec<HiFiMarker>, EventStoreError> {
679            self.inner.scan_hifi()
680        }
681
682        fn scan_hifi_records(
683            &self,
684        ) -> Result<Option<Vec<crate::markers::StoredMarkerRecord<HiFiMarker>>>, EventStoreError>
685        {
686            self.inner.scan_hifi_records()
687        }
688
689        fn scan_gaps(&self) -> Result<Vec<MarkerGap>, EventStoreError> {
690            self.inner.scan_gaps()
691        }
692
693        fn scan_gap_records(
694            &self,
695        ) -> Result<Option<Vec<crate::markers::StoredMarkerRecord<MarkerGap>>>, EventStoreError>
696        {
697            self.inner.scan_gap_records()
698        }
699
700        fn scan_dict(&self) -> Result<Vec<StreamDictEntry>, EventStoreError> {
701            self.inner.scan_dict()
702        }
703
704        fn scan_dict_records(
705            &self,
706        ) -> Result<Option<Vec<crate::markers::StoredMarkerRecord<StreamDictEntry>>>, EventStoreError>
707        {
708            self.inner.scan_dict_records()
709        }
710
711        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
712            self.inner.seal(status)
713        }
714
715        fn manifest(&self) -> Result<MarkerManifest, EventStoreError> {
716            Ok(self.manifest_override.clone())
717        }
718    }
719
720    #[rstest]
721    fn contiguous_marker_seq_passes() {
722        let mut backend = open_backend();
723        let s1 = snapshot(1, 1, 100, 1);
724        let s2 = snapshot(2, 2, 200, 2);
725        backend
726            .append_snapshot(&s1, compute_marker_hash(&s1))
727            .expect("append s1");
728        backend
729            .append_snapshot(&s2, compute_marker_hash(&s2))
730            .expect("append s2");
731
732        let report = MarkerVerifier::scan(&backend, 2).expect("scan");
733
734        assert!(report.is_clean(), "findings was: {:?}", report.findings);
735    }
736
737    #[rstest]
738    fn hifi_markers_participate_in_marker_seq_coverage() {
739        let mut backend = open_backend();
740        let s1 = snapshot(1, 1, 100, 1);
741        let m2 = hifi(2, 1, 0);
742        let s3 = snapshot(3, 2, 200, 2);
743        backend
744            .append_snapshot(&s1, compute_marker_hash(&s1))
745            .expect("append s1");
746        backend
747            .append_hifi(&m2, compute_hifi_hash(&m2))
748            .expect("append hifi");
749        backend
750            .append_snapshot(&s3, compute_marker_hash(&s3))
751            .expect("append s3");
752
753        let report = MarkerVerifier::scan(&backend, 2).expect("scan");
754
755        assert!(report.is_clean(), "findings was: {:?}", report.findings);
756    }
757
758    #[rstest]
759    fn manifest_count_mismatch_is_corrupt() {
760        let mut backend = open_backend();
761        let s1 = snapshot(1, 1, 100, 1);
762        backend
763            .append_snapshot(&s1, compute_marker_hash(&s1))
764            .expect("append s1");
765        let mut manifest_override = backend.manifest().expect("manifest");
766        manifest_override.snapshot_count = 2;
767        let backend = ManifestOverrideMarkerBackend::new(backend, manifest_override);
768
769        let report = MarkerVerifier::scan(&backend, 1).expect("scan");
770
771        assert!(
772            report.findings.iter().any(|finding| matches!(
773                finding,
774                MarkerFinding::ManifestCountMismatch {
775                    kind: MarkerCountKind::Snapshot,
776                    manifest_count: 2,
777                    scanned_count: 1,
778                }
779            )),
780            "findings was: {:?}",
781            report.findings,
782        );
783    }
784
785    #[rstest]
786    fn missing_marker_seq_without_gap_is_corrupt() {
787        let mut backend = open_backend();
788        let s1 = snapshot(1, 1, 100, 1);
789        let s3 = snapshot(3, 2, 200, 2);
790        backend
791            .append_snapshot(&s1, compute_marker_hash(&s1))
792            .expect("append s1");
793        backend
794            .append_snapshot(&s3, compute_marker_hash(&s3))
795            .expect("append s3");
796
797        let report = MarkerVerifier::scan(&backend, 3).expect("scan");
798
799        assert!(
800            report.findings.iter().any(|finding| matches!(
801                finding,
802                MarkerFinding::MarkerSeqGap {
803                    from_marker_seq: 2,
804                    to_marker_seq: 2,
805                }
806            )),
807            "findings was: {:?}",
808            report.findings,
809        );
810    }
811
812    #[rstest]
813    fn invalid_marker_gap_is_corrupt() {
814        let mut backend = open_backend();
815        let gap = MarkerGap {
816            from_marker_seq: 4,
817            to_marker_seq: 2,
818            reason: MarkerGapReason::Overflow,
819        };
820        backend
821            .append_gap(&gap, compute_gap_hash(&gap))
822            .expect("append invalid gap");
823
824        let report = MarkerVerifier::scan(&backend, 0).expect("scan");
825
826        assert!(
827            report.findings.iter().any(|finding| matches!(
828                finding,
829                MarkerFinding::InvalidMarkerGap {
830                    from_marker_seq: 4,
831                    to_marker_seq: 2,
832                }
833            )),
834            "findings was: {:?}",
835            report.findings,
836        );
837    }
838
839    #[rstest]
840    fn overlapping_marker_coverage_is_corrupt() {
841        let mut backend = open_backend();
842        let s1 = snapshot(1, 1, 100, 1);
843        let gap = MarkerGap {
844            from_marker_seq: 1,
845            to_marker_seq: 1,
846            reason: MarkerGapReason::Overflow,
847        };
848        backend
849            .append_snapshot(&s1, compute_marker_hash(&s1))
850            .expect("append s1");
851        backend
852            .append_gap(&gap, compute_gap_hash(&gap))
853            .expect("append overlapping gap");
854
855        let report = MarkerVerifier::scan(&backend, 1).expect("scan");
856
857        assert!(
858            report.findings.iter().any(|finding| matches!(
859                finding,
860                MarkerFinding::MarkerSeqOverlap {
861                    from_marker_seq: 1,
862                    to_marker_seq: 1,
863                }
864            )),
865            "findings was: {:?}",
866            report.findings,
867        );
868    }
869
870    #[rstest]
871    fn non_monotonic_count_is_corrupt() {
872        let mut backend = open_backend();
873        let s1 = snapshot(1, 1, 100, 10);
874        let s2 = snapshot(2, 2, 200, 9);
875        backend
876            .append_snapshot(&s1, compute_marker_hash(&s1))
877            .expect("append s1");
878        backend
879            .append_snapshot(&s2, compute_marker_hash(&s2))
880            .expect("append s2");
881
882        let report = MarkerVerifier::scan(&backend, 2).expect("scan");
883
884        assert!(
885            report.findings.iter().any(|finding| matches!(
886                finding,
887                MarkerFinding::CursorCountRegressed {
888                    marker_seq: 2,
889                    slot: 0,
890                    previous_count: 10,
891                    count: 9,
892                }
893            )),
894            "findings was: {:?}",
895            report.findings,
896        );
897    }
898
899    #[rstest]
900    fn non_monotonic_ts_init_hi_is_corrupt() {
901        let mut backend = open_backend();
902        let s1 = snapshot(1, 1, 200, 1);
903        let s2 = snapshot(2, 2, 199, 2);
904        backend
905            .append_snapshot(&s1, compute_marker_hash(&s1))
906            .expect("append s1");
907        backend
908            .append_snapshot(&s2, compute_marker_hash(&s2))
909            .expect("append s2");
910
911        let report = MarkerVerifier::scan(&backend, 2).expect("scan");
912
913        assert!(
914            report.findings.iter().any(|finding| matches!(
915                finding,
916                MarkerFinding::CursorTsInitRegressed {
917                    marker_seq: 2,
918                    slot: 0,
919                    previous_ts_init_hi,
920                    ts_init_hi,
921                } if *previous_ts_init_hi == UnixNanos::from(200)
922                    && *ts_init_hi == UnixNanos::from(199)
923            )),
924            "findings was: {:?}",
925            report.findings,
926        );
927    }
928
929    #[rstest]
930    fn event_seq_before_regression_is_corrupt() {
931        let mut backend = open_backend();
932        let s1 = snapshot(1, 10, 100, 1);
933        let s2 = snapshot(2, 9, 200, 2);
934        backend
935            .append_snapshot(&s1, compute_marker_hash(&s1))
936            .expect("append s1");
937        backend
938            .append_snapshot(&s2, compute_marker_hash(&s2))
939            .expect("append s2");
940
941        let report = MarkerVerifier::scan(&backend, 10).expect("scan");
942
943        assert!(
944            report.findings.iter().any(|finding| matches!(
945                finding,
946                MarkerFinding::EventSeqRegressed {
947                    marker_seq: 2,
948                    previous_event_seq_before: 10,
949                    event_seq_before: 9,
950                }
951            )),
952            "findings was: {:?}",
953            report.findings,
954        );
955    }
956
957    #[rstest]
958    fn event_seq_before_exceeding_high_watermark_is_corrupt() {
959        let mut backend = open_backend();
960        let s1 = snapshot(1, 11, 100, 1);
961        backend
962            .append_snapshot(&s1, compute_marker_hash(&s1))
963            .expect("append s1");
964
965        let report = MarkerVerifier::scan(&backend, 10).expect("scan");
966
967        assert!(
968            report.findings.iter().any(|finding| matches!(
969                finding,
970                MarkerFinding::EventSeqExceedsHighWatermark {
971                    marker_seq: 1,
972                    event_seq_before: 11,
973                    high_watermark: 10,
974                }
975            )),
976            "findings was: {:?}",
977            report.findings,
978        );
979    }
980
981    #[rstest]
982    fn non_snapshot_record_hash_mismatches_are_corrupt() {
983        let mut backend = open_backend();
984        let s1 = snapshot(1, 1, 100, 1);
985        let m2 = hifi(2, 1, 0);
986        let g3 = MarkerGap {
987            from_marker_seq: 3,
988            to_marker_seq: 3,
989            reason: MarkerGapReason::Overflow,
990        };
991        let d0 = dict(0);
992        backend
993            .append_snapshot(&s1, compute_marker_hash(&s1))
994            .expect("append s1");
995        backend
996            .append_hifi(&m2, [0xBB; 32])
997            .expect("append bad hifi hash");
998        backend
999            .append_gap(&g3, [0xCC; 32])
1000            .expect("append bad gap hash");
1001        backend
1002            .put_dict(&d0, [0xDD; 32])
1003            .expect("put bad dict hash");
1004
1005        let report = MarkerVerifier::scan(&backend, 1).expect("scan");
1006
1007        assert!(
1008            report.findings.iter().any(|finding| matches!(
1009                finding,
1010                MarkerFinding::HashMismatch {
1011                    record: MarkerRecordKind::HiFi,
1012                    marker_seq: Some(2),
1013                    slot: Some(0),
1014                }
1015            )),
1016            "findings was: {:?}",
1017            report.findings,
1018        );
1019        assert!(
1020            report.findings.iter().any(|finding| matches!(
1021                finding,
1022                MarkerFinding::HashMismatch {
1023                    record: MarkerRecordKind::Gap,
1024                    marker_seq: None,
1025                    slot: None,
1026                }
1027            )),
1028            "findings was: {:?}",
1029            report.findings,
1030        );
1031        assert!(
1032            report.findings.iter().any(|finding| matches!(
1033                finding,
1034                MarkerFinding::HashMismatch {
1035                    record: MarkerRecordKind::Dict,
1036                    marker_seq: None,
1037                    slot: Some(0),
1038                }
1039            )),
1040            "findings was: {:?}",
1041            report.findings,
1042        );
1043    }
1044
1045    #[rstest]
1046    fn bad_record_hash_is_corrupt() {
1047        let mut backend = open_backend();
1048        let s1 = snapshot(1, 1, 100, 1);
1049        backend
1050            .append_snapshot(&s1, [0xAA; 32])
1051            .expect("append bad hash");
1052
1053        let report = MarkerVerifier::scan(&backend, 1).expect("scan");
1054
1055        assert!(
1056            report.findings.iter().any(|finding| matches!(
1057                finding,
1058                MarkerFinding::HashMismatch {
1059                    record: MarkerRecordKind::Snapshot,
1060                    marker_seq: Some(1),
1061                    slot: None,
1062                }
1063            )),
1064            "findings was: {:?}",
1065            report.findings,
1066        );
1067    }
1068
1069    #[rstest]
1070    fn marker_gap_covers_missing_marker_seq() {
1071        let mut backend = open_backend();
1072        let s1 = snapshot(1, 1, 100, 1);
1073        let s3 = snapshot(3, 2, 200, 2);
1074        let gap = MarkerGap {
1075            from_marker_seq: 2,
1076            to_marker_seq: 2,
1077            reason: MarkerGapReason::Overflow,
1078        };
1079        backend
1080            .append_snapshot(&s1, compute_marker_hash(&s1))
1081            .expect("append s1");
1082        backend
1083            .append_gap(&gap, compute_gap_hash(&gap))
1084            .expect("append gap");
1085        backend
1086            .append_snapshot(&s3, compute_marker_hash(&s3))
1087            .expect("append s3");
1088
1089        let report = MarkerVerifier::scan(&backend, 3).expect("scan");
1090
1091        assert!(report.is_clean(), "findings was: {:?}", report.findings);
1092    }
1093}
nautilus_event_store/markers/verifier.rs

nautilus_event_store/markers/
verifier.rs
