1use 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
49pub 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 #[must_use]
68 pub fn new(backend: Box<dyn EventStore>) -> Self {
69 Self { backend }
70 }
71
72 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 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 #[must_use]
114 pub fn backend(&self) -> &dyn EventStore {
115 self.backend.as_ref()
116 }
117
118 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 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
274fn 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#[derive(Debug, Clone, PartialEq, Eq)]
364pub struct VerifyReport {
365 pub run_id: RunId,
367 pub status: RunStatus,
369 pub high_watermark: u64,
371 pub entries_scanned: u64,
373 pub findings: Vec<VerifyFinding>,
375}
376
377impl VerifyReport {
378 #[must_use]
380 pub fn is_clean(&self) -> bool {
381 self.findings.is_empty()
382 }
383}
384
385#[derive(Debug, Clone, PartialEq, Eq)]
387pub enum VerifyFinding {
388 HashMismatch {
390 seq: u64,
392 },
393 Gap {
395 range: GapRange,
397 },
398 SeqMismatch {
406 table_key: u64,
408 embedded_seq: u64,
410 },
411 IndexDrift {
414 kind: IndexKind,
416 key: String,
418 drift: IndexDrift,
420 },
421 ManifestMismatch {
424 kind: ManifestField,
426 reason: String,
428 },
429 SnapshotAnchorInvalid {
432 reason: String,
434 },
435}
436
437#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
439pub struct GapRange {
440 pub from: u64,
442 pub to: u64,
444}
445
446#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
453pub enum IndexDrift {
454 DanglingTarget {
456 stored_seq: u64,
458 },
459 TargetCorrupted {
461 stored_seq: u64,
463 },
464}
465
466#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
468pub enum ManifestField {
469 HighWatermark,
471 StartTsInit,
473 EndTsInit,
476}
477
478#[derive(Debug, thiserror::Error)]
484pub enum VerifyError {
485 #[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 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 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 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 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 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 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 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 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 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 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 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 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 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 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}