nautilus_event_store/backend/
memory.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//! In-memory [`EventStore`] backend used by writer and reader unit tests and by the
17//! `cfg(madsim)` simulation backend.
18
19use indexmap::IndexMap;
20use nautilus_core::UnixNanos;
21
22use crate::{
23    backend::{AppendEntry, EventStore, IndexKey, IndexKind, ScanDirection},
24    entry::EventStoreEntry,
25    error::EventStoreError,
26    manifest::{RunManifest, RunStatus},
27    snapshot::{SnapshotAnchor, validate_new_anchor},
28};
29
30/// In-memory implementation of [`EventStore`].
31///
32/// Stores entries densely in a `Vec` keyed by `seq - 1` plus one [`IndexMap`] per
33/// [`IndexKind`] for the sidecar indices. Hash recomputation on read is structurally
34/// redundant (entries live in process memory) but kept for parity with persistent
35/// backends so callers see uniform behavior.
36///
37/// One backend instance owns at most one open run at a time. Sealing the open run leaves
38/// the manifest and entries readable until the next [`EventStore::open_run`] call replaces
39/// them with a fresh run. Reopening while a `Running` run still exists returns
40/// [`EventStoreError::CrashedPredecessor`] so callers exercise the same crash-recovery
41/// path persistent backends surface on reopen.
42#[derive(Debug, Default)]
43pub struct MemoryBackend {
44    state: Option<RunState>,
45}
46
47#[derive(Debug)]
48struct RunState {
49    manifest: RunManifest,
50    entries: Vec<EventStoreEntry>,
51    indices: Indices,
52    snapshot_anchor: Option<SnapshotAnchor>,
53    high_watermark: u64,
54    max_ts_init: UnixNanos,
55}
56
57#[derive(Debug, Default)]
58struct Indices {
59    client_order: IndexMap<String, u64>,
60    venue_order: IndexMap<String, u64>,
61}
62
63impl Indices {
64    fn map_for(&self, kind: IndexKind) -> &IndexMap<String, u64> {
65        match kind {
66            IndexKind::ClientOrderId => &self.client_order,
67            IndexKind::VenueOrderId => &self.venue_order,
68        }
69    }
70
71    fn map_for_mut(&mut self, kind: IndexKind) -> &mut IndexMap<String, u64> {
72        match kind {
73            IndexKind::ClientOrderId => &mut self.client_order,
74            IndexKind::VenueOrderId => &mut self.venue_order,
75        }
76    }
77}
78
79impl MemoryBackend {
80    /// Creates a new empty [`MemoryBackend`] with no run open.
81    #[must_use]
82    pub fn new() -> Self {
83        Self::default()
84    }
85
86    fn state(&self) -> Result<&RunState, EventStoreError> {
87        self.state
88            .as_ref()
89            .ok_or_else(|| EventStoreError::Backend("no run open".to_string()))
90    }
91
92    fn state_mut(&mut self) -> Result<&mut RunState, EventStoreError> {
93        self.state
94            .as_mut()
95            .ok_or_else(|| EventStoreError::Backend("no run open".to_string()))
96    }
97}
98
99impl EventStore for MemoryBackend {
100    fn open_run(&mut self, mut manifest: RunManifest) -> Result<(), EventStoreError> {
101        if let Some(state) = &self.state
102            && !state.manifest.is_sealed()
103        {
104            return Err(EventStoreError::CrashedPredecessor);
105        }
106
107        manifest.status = RunStatus::Running;
108        manifest.end_ts_init = None;
109        manifest.high_watermark = 0;
110
111        self.state = Some(RunState {
112            manifest,
113            entries: Vec::new(),
114            indices: Indices::default(),
115            snapshot_anchor: None,
116            high_watermark: 0,
117            max_ts_init: UnixNanos::default(),
118        });
119        Ok(())
120    }
121
122    fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
123        let state = self.state_mut()?;
124
125        if state.manifest.is_sealed() {
126            return Err(EventStoreError::Closed);
127        }
128
129        if entries.is_empty() {
130            return Ok(state.high_watermark);
131        }
132
133        for (expected, append) in (state.high_watermark + 1..).zip(entries.iter()) {
134            if append.entry.seq != expected {
135                // Batch is atomically rejected: report the durable high-watermark, not
136                // the within-batch validation cursor, so callers that resync from this
137                // value never skip entries that were never committed.
138                return Err(EventStoreError::OutOfOrder {
139                    high_watermark: state.high_watermark,
140                    seq: append.entry.seq,
141                });
142            }
143        }
144
145        for append in entries {
146            for IndexKey { kind, key } in &append.index_keys {
147                state
148                    .indices
149                    .map_for_mut(*kind)
150                    .entry(key.clone())
151                    .or_insert(append.entry.seq);
152            }
153
154            if append.entry.ts_init > state.max_ts_init {
155                state.max_ts_init = append.entry.ts_init;
156            }
157            state.high_watermark = append.entry.seq;
158            state.entries.push(append.entry.clone());
159        }
160
161        state.manifest.high_watermark = state.high_watermark;
162        Ok(state.high_watermark)
163    }
164
165    fn scan_range(
166        &self,
167        from: u64,
168        to: u64,
169        direction: ScanDirection,
170    ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
171        let state = self.state()?;
172
173        if from > to || from == 0 || state.entries.is_empty() {
174            return Ok(Vec::new());
175        }
176
177        let lo = usize::try_from(from)
178            .unwrap_or(usize::MAX)
179            .saturating_sub(1);
180        let hi = usize::try_from(to)
181            .unwrap_or(usize::MAX)
182            .min(state.entries.len());
183
184        if lo >= hi {
185            return Ok(Vec::new());
186        }
187
188        let slice = &state.entries[lo..hi];
189        for entry in slice {
190            if entry.recompute_hash() != entry.entry_hash {
191                return Err(EventStoreError::HashMismatch { seq: entry.seq });
192            }
193        }
194
195        let mut out: Vec<EventStoreEntry> = slice.to_vec();
196        if matches!(direction, ScanDirection::Reverse) {
197            out.reverse();
198        }
199        Ok(out)
200    }
201
202    fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
203        let state = self.state()?;
204
205        if seq == 0 || seq > state.high_watermark {
206            return Ok(None);
207        }
208
209        let idx = usize::try_from(seq - 1)
210            .map_err(|e| EventStoreError::Backend(format!("seq {seq} out of usize range: {e}")))?;
211        let entry = &state.entries[idx];
212        if entry.recompute_hash() != entry.entry_hash {
213            return Err(EventStoreError::HashMismatch { seq });
214        }
215        Ok(Some(entry.clone()))
216    }
217
218    fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
219        let state = self.state()?;
220        Ok(state.indices.map_for(kind).get(key).copied())
221    }
222
223    fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
224        let state = self.state()?;
225        Ok(state
226            .indices
227            .map_for(kind)
228            .iter()
229            .map(|(k, v)| (k.clone(), *v))
230            .collect())
231    }
232
233    fn record_snapshot_anchor(&mut self, anchor: SnapshotAnchor) -> Result<(), EventStoreError> {
234        let state = self.state_mut()?;
235
236        if state.manifest.is_sealed() {
237            return Err(EventStoreError::Closed);
238        }
239
240        validate_new_anchor(
241            &anchor,
242            state.high_watermark,
243            state.snapshot_anchor.as_ref(),
244        )?;
245        state.snapshot_anchor = Some(anchor);
246        Ok(())
247    }
248
249    fn latest_snapshot_anchor(&self) -> Result<Option<SnapshotAnchor>, EventStoreError> {
250        Ok(self.state()?.snapshot_anchor.clone())
251    }
252
253    fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
254        let state = self.state_mut()?;
255
256        // `RunStatus::Running` is not a terminal state; accepting it would leave the
257        // manifest unsealed (`is_sealed()` returns false) while still returning Ok,
258        // so subsequent `append_batch` calls would not see `Closed`.
259        if matches!(status, RunStatus::Running) {
260            return Err(EventStoreError::Backend(
261                "seal status must be a terminal state, was Running".to_string(),
262            ));
263        }
264
265        if state.manifest.is_sealed() {
266            return Err(EventStoreError::Closed);
267        }
268
269        state.manifest.status = status;
270        state.manifest.high_watermark = state.high_watermark;
271        if state.high_watermark > 0 {
272            state.manifest.end_ts_init = Some(state.max_ts_init);
273        }
274        Ok(())
275    }
276
277    fn manifest(&self) -> Result<RunManifest, EventStoreError> {
278        Ok(self.state()?.manifest.clone())
279    }
280
281    fn high_watermark(&self) -> Result<u64, EventStoreError> {
282        Ok(self.state()?.high_watermark)
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use bytes::Bytes;
289    use indexmap::IndexMap;
290    use nautilus_core::{UUID4, UnixNanos};
291    use rstest::{fixture, rstest};
292    use ustr::Ustr;
293
294    use super::*;
295    use crate::{
296        compute_entry_hash,
297        entry::{EventStoreEntry, Topic},
298        headers::Headers,
299        manifest::{RegisteredComponents, RunManifest, RunStatus},
300    };
301
302    fn manifest(run_id: &str) -> RunManifest {
303        RunManifest {
304            run_id: run_id.to_string(),
305            parent_run_id: None,
306            instance_id: "trader-001".to_string(),
307            binary_hash: "deadbeef".to_string(),
308            schema_version: 1,
309            crate_versions: "feedface".to_string(),
310            feature_flags: Vec::new(),
311            adapter_versions: IndexMap::new(),
312            config_hash: "cafebabe".to_string(),
313            registered_components: RegisteredComponents::default(),
314            seed: None,
315            start_ts_init: UnixNanos::from(0),
316            end_ts_init: None,
317            high_watermark: 0,
318            status: RunStatus::Running,
319        }
320    }
321
322    fn build_entry(seq: u64, headers: Headers, ts_init: u64) -> EventStoreEntry {
323        let topic: Topic = "exec.command.SubmitOrder".into();
324        let payload_type = Ustr::from("SubmitOrder");
325        let payload = Bytes::from_static(b"\x01\x02\x03\x04");
326        let ts_publish = UnixNanos::from(ts_init + 1);
327        let ts_init = UnixNanos::from(ts_init);
328        let hash = compute_entry_hash(
329            seq,
330            ts_init,
331            ts_publish,
332            topic.as_ref(),
333            payload_type.as_str(),
334            &payload,
335            &headers,
336        );
337
338        EventStoreEntry::new(
339            hash,
340            seq,
341            headers,
342            topic,
343            payload_type,
344            payload,
345            ts_init,
346            ts_publish,
347        )
348    }
349
350    fn append_with(seq: u64, ts_init: u64, index_keys: Vec<IndexKey>) -> AppendEntry {
351        AppendEntry::new(build_entry(seq, Headers::empty(), ts_init), index_keys)
352    }
353
354    #[fixture]
355    fn open_backend() -> MemoryBackend {
356        let mut backend = MemoryBackend::new();
357        backend
358            .open_run(manifest("1700000000-aaaa1111"))
359            .expect("open run");
360        backend
361    }
362
363    #[rstest]
364    fn manifest_errors_when_no_run_open() {
365        let backend = MemoryBackend::new();
366
367        match backend.manifest() {
368            Err(EventStoreError::Backend(msg)) => {
369                assert!(msg.contains("no run open"), "msg was: {msg}");
370            }
371            other => panic!("expected Backend, was {other:?}"),
372        }
373
374        match backend.high_watermark() {
375            Err(EventStoreError::Backend(msg)) => {
376                assert!(msg.contains("no run open"), "msg was: {msg}");
377            }
378            other => panic!("expected Backend, was {other:?}"),
379        }
380    }
381
382    #[rstest]
383    #[case::append_batch("append_batch")]
384    #[case::scan_range("scan_range")]
385    #[case::scan_seq("scan_seq")]
386    #[case::lookup("lookup")]
387    #[case::record_snapshot_anchor("record_snapshot_anchor")]
388    #[case::latest_snapshot_anchor("latest_snapshot_anchor")]
389    #[case::seal("seal")]
390    fn methods_error_when_no_run_open(#[case] op: &str) {
391        let mut backend = MemoryBackend::new();
392        let err = match op {
393            "append_batch" => backend.append_batch(&[]).unwrap_err(),
394            "scan_range" => backend
395                .scan_range(1, 1, ScanDirection::Forward)
396                .unwrap_err(),
397            "scan_seq" => backend.scan_seq(1).unwrap_err(),
398            "lookup" => backend.lookup(IndexKind::ClientOrderId, "k").unwrap_err(),
399            "record_snapshot_anchor" => backend
400                .record_snapshot_anchor(SnapshotAnchor::new(0, "blob", "hash"))
401                .unwrap_err(),
402            "latest_snapshot_anchor" => backend.latest_snapshot_anchor().unwrap_err(),
403            "seal" => backend.seal(RunStatus::Ended).unwrap_err(),
404            _ => unreachable!(),
405        };
406
407        match err {
408            EventStoreError::Backend(msg) => {
409                assert!(msg.contains("no run open"), "msg was: {msg}");
410            }
411            other => panic!("expected Backend, was {other:?}"),
412        }
413    }
414
415    #[rstest]
416    fn open_run_normalizes_status_and_zeroes_progress(open_backend: MemoryBackend) {
417        let m = open_backend.manifest().expect("manifest");
418
419        assert_eq!(m.status, RunStatus::Running);
420        assert_eq!(m.high_watermark, 0);
421        assert!(m.end_ts_init.is_none());
422        assert_eq!(open_backend.high_watermark().expect("hwm"), 0);
423    }
424
425    #[rstest]
426    fn append_advances_high_watermark(mut open_backend: MemoryBackend) {
427        let batch = vec![
428            append_with(1, 10, Vec::new()),
429            append_with(2, 11, Vec::new()),
430            append_with(3, 12, Vec::new()),
431        ];
432
433        let hwm = open_backend.append_batch(&batch).expect("append");
434
435        assert_eq!(hwm, 3);
436        assert_eq!(open_backend.high_watermark().expect("hwm"), 3);
437        assert_eq!(open_backend.manifest().expect("m").high_watermark, 3);
438    }
439
440    #[rstest]
441    fn append_rejects_first_seq_not_at_watermark_plus_one(mut open_backend: MemoryBackend) {
442        let batch = vec![append_with(2, 10, Vec::new())];
443
444        let err = open_backend.append_batch(&batch).expect_err("must reject");
445
446        assert!(matches!(
447            err,
448            EventStoreError::OutOfOrder {
449                high_watermark: 0,
450                seq: 2,
451            }
452        ));
453    }
454
455    #[rstest]
456    fn append_rejects_within_batch_seq_gap(mut open_backend: MemoryBackend) {
457        let batch = vec![
458            append_with(1, 10, Vec::new()),
459            append_with(3, 11, Vec::new()),
460        ];
461
462        let err = open_backend.append_batch(&batch).expect_err("must reject");
463
464        // Atomically rejected: durable hwm is still 0, not the within-batch cursor.
465        assert!(matches!(
466            err,
467            EventStoreError::OutOfOrder {
468                high_watermark: 0,
469                seq: 3,
470            }
471        ));
472        // Failed batch must not have partially landed.
473        assert_eq!(open_backend.high_watermark().expect("hwm"), 0);
474    }
475
476    #[rstest]
477    fn append_after_seal_returns_closed(mut open_backend: MemoryBackend) {
478        open_backend
479            .append_batch(&[append_with(1, 10, Vec::new())])
480            .expect("append");
481        open_backend.seal(RunStatus::Ended).expect("seal");
482
483        let err = open_backend
484            .append_batch(&[append_with(2, 11, Vec::new())])
485            .expect_err("must reject");
486
487        assert!(matches!(err, EventStoreError::Closed));
488    }
489
490    #[rstest]
491    fn empty_batch_is_a_noop(mut open_backend: MemoryBackend) {
492        let hwm = open_backend.append_batch(&[]).expect("append");
493
494        assert_eq!(hwm, 0);
495        assert_eq!(open_backend.high_watermark().expect("hwm"), 0);
496    }
497
498    #[rstest]
499    fn snapshot_anchor_is_none_until_recorded(open_backend: MemoryBackend) {
500        assert!(
501            open_backend
502                .latest_snapshot_anchor()
503                .expect("latest anchor")
504                .is_none()
505        );
506    }
507
508    #[rstest]
509    fn snapshot_anchor_round_trips(mut open_backend: MemoryBackend) {
510        open_backend
511            .append_batch(&[append_with(1, 10, Vec::new())])
512            .expect("append");
513        let anchor = SnapshotAnchor::new(1, "cache://snapshots/run-1/1", "blake3:abc");
514
515        open_backend
516            .record_snapshot_anchor(anchor.clone())
517            .expect("record anchor");
518
519        assert_eq!(
520            open_backend
521                .latest_snapshot_anchor()
522                .expect("latest anchor"),
523            Some(anchor),
524        );
525    }
526
527    #[rstest]
528    fn snapshot_anchor_rejects_watermark_past_durable_hwm(mut open_backend: MemoryBackend) {
529        let anchor = SnapshotAnchor::new(1, "cache://snapshots/run-1/1", "blake3:abc");
530        let err = open_backend
531            .record_snapshot_anchor(anchor)
532            .expect_err("must reject");
533
534        match err {
535            EventStoreError::Backend(msg) => {
536                assert!(
537                    msg.contains("exceeds durable high_watermark"),
538                    "msg was: {msg}",
539                );
540            }
541            other => panic!("expected Backend, was {other:?}"),
542        }
543    }
544
545    #[rstest]
546    fn snapshot_anchor_rejects_backward_move(mut open_backend: MemoryBackend) {
547        open_backend
548            .append_batch(&[
549                append_with(1, 10, Vec::new()),
550                append_with(2, 11, Vec::new()),
551            ])
552            .expect("append");
553        open_backend
554            .record_snapshot_anchor(SnapshotAnchor::new(2, "latest", "hash-latest"))
555            .expect("record latest");
556
557        let err = open_backend
558            .record_snapshot_anchor(SnapshotAnchor::new(1, "older", "hash-older"))
559            .expect_err("must reject older anchor");
560
561        match err {
562            EventStoreError::Backend(msg) => {
563                assert!(msg.contains("older than latest anchor"), "msg was: {msg}");
564            }
565            other => panic!("expected Backend, was {other:?}"),
566        }
567    }
568
569    #[rstest]
570    fn snapshot_anchor_after_seal_returns_closed(mut open_backend: MemoryBackend) {
571        open_backend
572            .append_batch(&[append_with(1, 10, Vec::new())])
573            .expect("append");
574        open_backend.seal(RunStatus::Ended).expect("seal");
575
576        let err = open_backend
577            .record_snapshot_anchor(SnapshotAnchor::new(1, "blob", "hash"))
578            .expect_err("must reject");
579
580        assert!(matches!(err, EventStoreError::Closed));
581    }
582
583    #[rstest]
584    fn scan_seq_returns_committed_entry(mut open_backend: MemoryBackend) {
585        open_backend
586            .append_batch(&[
587                append_with(1, 10, Vec::new()),
588                append_with(2, 11, Vec::new()),
589            ])
590            .expect("append");
591
592        let entry = open_backend.scan_seq(2).expect("scan").expect("present");
593
594        assert_eq!(entry.seq, 2);
595        assert_eq!(entry.ts_init, UnixNanos::from(11));
596    }
597
598    #[rstest]
599    fn scan_seq_returns_none_outside_watermark(mut open_backend: MemoryBackend) {
600        open_backend
601            .append_batch(&[append_with(1, 10, Vec::new())])
602            .expect("append");
603
604        assert!(open_backend.scan_seq(0).expect("scan").is_none());
605        assert!(open_backend.scan_seq(2).expect("scan").is_none());
606    }
607
608    #[rstest]
609    #[case::forward_full(1, 3, ScanDirection::Forward, vec![1, 2, 3])]
610    #[case::reverse_full(1, 3, ScanDirection::Reverse, vec![3, 2, 1])]
611    #[case::forward_window(2, 3, ScanDirection::Forward, vec![2, 3])]
612    #[case::reverse_window(2, 3, ScanDirection::Reverse, vec![3, 2])]
613    #[case::clipped_to_watermark(2, 99, ScanDirection::Forward, vec![2, 3])]
614    #[case::reverse_clipped(2, 99, ScanDirection::Reverse, vec![3, 2])]
615    #[case::empty_inverted(3, 1, ScanDirection::Forward, vec![])]
616    #[case::empty_zero(0, 0, ScanDirection::Forward, vec![])]
617    fn scan_range_yields_expected_seqs(
618        mut open_backend: MemoryBackend,
619        #[case] from: u64,
620        #[case] to: u64,
621        #[case] direction: ScanDirection,
622        #[case] expected: Vec<u64>,
623    ) {
624        open_backend
625            .append_batch(&[
626                append_with(1, 10, Vec::new()),
627                append_with(2, 11, Vec::new()),
628                append_with(3, 12, Vec::new()),
629            ])
630            .expect("append");
631
632        let seqs: Vec<u64> = open_backend
633            .scan_range(from, to, direction)
634            .expect("scan")
635            .into_iter()
636            .map(|e| e.seq)
637            .collect();
638
639        assert_eq!(seqs, expected);
640    }
641
642    #[rstest]
643    fn lookup_records_first_occurrence_per_kind(mut open_backend: MemoryBackend) {
644        let cl_ord = "O-1".to_string();
645        let venue = "V-1".to_string();
646        open_backend
647            .append_batch(&[
648                AppendEntry::new(
649                    build_entry(1, Headers::empty(), 10),
650                    vec![
651                        IndexKey::new(IndexKind::ClientOrderId, cl_ord.clone()),
652                        IndexKey::new(IndexKind::VenueOrderId, venue.clone()),
653                    ],
654                ),
655                AppendEntry::new(
656                    build_entry(2, Headers::empty(), 11),
657                    vec![
658                        // Same keys re-emitted: lookups must continue to point at seq=1.
659                        IndexKey::new(IndexKind::ClientOrderId, cl_ord.clone()),
660                        IndexKey::new(IndexKind::VenueOrderId, venue.clone()),
661                    ],
662                ),
663            ])
664            .expect("append");
665
666        assert_eq!(
667            open_backend
668                .lookup(IndexKind::ClientOrderId, &cl_ord)
669                .expect("lookup"),
670            Some(1),
671        );
672        assert_eq!(
673            open_backend
674                .lookup(IndexKind::VenueOrderId, &venue)
675                .expect("lookup"),
676            Some(1),
677        );
678        assert!(
679            open_backend
680                .lookup(IndexKind::ClientOrderId, "missing")
681                .expect("lookup")
682                .is_none(),
683        );
684    }
685
686    #[rstest]
687    fn within_entry_duplicate_keys_resolve_to_first_seq(mut open_backend: MemoryBackend) {
688        // First-write-wins applies within a single entry's index_keys vec as well
689        // as across entries: a duplicate key (within entry 1) and a later entry's
690        // re-emission (entry 2) both leave the lookup pointing at seq=1.
691        let key = "O-1".to_string();
692        open_backend
693            .append_batch(&[
694                AppendEntry::new(
695                    build_entry(1, Headers::empty(), 10),
696                    vec![
697                        IndexKey::new(IndexKind::ClientOrderId, key.clone()),
698                        IndexKey::new(IndexKind::ClientOrderId, key.clone()),
699                    ],
700                ),
701                AppendEntry::new(
702                    build_entry(2, Headers::empty(), 11),
703                    vec![IndexKey::new(IndexKind::ClientOrderId, key.clone())],
704                ),
705            ])
706            .expect("append");
707
708        assert_eq!(
709            open_backend
710                .lookup(IndexKind::ClientOrderId, &key)
711                .expect("lookup"),
712            Some(1),
713        );
714    }
715
716    #[rstest]
717    fn lookup_isolates_keys_by_kind(mut open_backend: MemoryBackend) {
718        // Same string under two different IndexKinds must not collide.
719        let key = "shared".to_string();
720        open_backend
721            .append_batch(&[AppendEntry::new(
722                build_entry(1, Headers::empty(), 10),
723                vec![IndexKey::new(IndexKind::ClientOrderId, key.clone())],
724            )])
725            .expect("append");
726
727        assert_eq!(
728            open_backend
729                .lookup(IndexKind::ClientOrderId, &key)
730                .expect("lookup"),
731            Some(1),
732        );
733        assert!(
734            open_backend
735                .lookup(IndexKind::VenueOrderId, &key)
736                .expect("lookup")
737                .is_none(),
738        );
739    }
740
741    #[rstest]
742    #[case::ended(RunStatus::Ended)]
743    #[case::crashed_recovered(RunStatus::CrashedRecovered)]
744    #[case::quarantined(RunStatus::Quarantined)]
745    fn seal_stamps_end_ts_and_blocks_re_seal(
746        mut open_backend: MemoryBackend,
747        #[case] status: RunStatus,
748    ) {
749        open_backend
750            .append_batch(&[
751                append_with(1, 10, Vec::new()),
752                append_with(2, 25, Vec::new()),
753                append_with(3, 17, Vec::new()),
754            ])
755            .expect("append");
756
757        open_backend.seal(status).expect("seal");
758
759        let m = open_backend.manifest().expect("manifest");
760        assert_eq!(m.status, status);
761        assert_eq!(m.high_watermark, 3);
762        // Highest ts_init across the run, not the last-arrived.
763        assert_eq!(m.end_ts_init, Some(UnixNanos::from(25)));
764
765        let err = open_backend.seal(RunStatus::Ended).expect_err("re-seal");
766        assert!(matches!(err, EventStoreError::Closed));
767    }
768
769    #[rstest]
770    fn seal_rejects_running_status(mut open_backend: MemoryBackend) {
771        // Running is not a terminal state. Rejecting it keeps the seal contract
772        // honest: a successful seal must make subsequent appends return Closed.
773        let err = open_backend
774            .seal(RunStatus::Running)
775            .expect_err("must reject");
776
777        match err {
778            EventStoreError::Backend(msg) => {
779                assert!(msg.contains("Running"), "msg was: {msg}");
780            }
781            other => panic!("expected Backend, was {other:?}"),
782        }
783        assert!(!open_backend.manifest().expect("manifest").is_sealed());
784        // The run is still writeable.
785        open_backend
786            .append_batch(&[append_with(1, 10, Vec::new())])
787            .expect("append");
788    }
789
790    #[rstest]
791    fn seal_with_no_entries_leaves_end_ts_unset(mut open_backend: MemoryBackend) {
792        open_backend.seal(RunStatus::Ended).expect("seal");
793
794        let m = open_backend.manifest().expect("manifest");
795        assert_eq!(m.status, RunStatus::Ended);
796        assert!(m.end_ts_init.is_none());
797        assert_eq!(m.high_watermark, 0);
798    }
799
800    #[rstest]
801    fn reopening_running_run_returns_crashed_predecessor() {
802        let mut backend = MemoryBackend::new();
803        backend.open_run(manifest("run-1")).expect("open 1");
804        backend
805            .append_batch(&[append_with(1, 10, Vec::new())])
806            .expect("append");
807
808        // Caller forgot to seal; the second open_run flags it for crash recovery.
809        let err = backend.open_run(manifest("run-2")).expect_err("must flag");
810        assert!(matches!(err, EventStoreError::CrashedPredecessor));
811
812        // The failed open must preserve the predecessor's entries so the verifier
813        // can scan them before the kernel decides CrashedRecovered vs Quarantined.
814        assert!(
815            backend.scan_seq(1).expect("scan").is_some(),
816            "predecessor entry must survive failed open_run",
817        );
818
819        // After sealing the predecessor, a fresh open succeeds.
820        backend.seal(RunStatus::CrashedRecovered).expect("seal");
821        backend.open_run(manifest("run-2")).expect("open 2");
822        assert_eq!(
823            backend.manifest().expect("manifest").run_id,
824            "run-2".to_string(),
825        );
826        assert_eq!(backend.high_watermark().expect("hwm"), 0);
827    }
828
829    #[rstest]
830    fn reopening_after_clean_seal_succeeds() {
831        let mut backend = MemoryBackend::new();
832        backend.open_run(manifest("run-1")).expect("open 1");
833        backend.seal(RunStatus::Ended).expect("seal");
834
835        backend.open_run(manifest("run-2")).expect("open 2");
836        assert_eq!(
837            backend.manifest().expect("manifest").run_id,
838            "run-2".to_string(),
839        );
840    }
841
842    #[rstest]
843    fn scan_recomputes_hash_and_quarantines_on_mismatch(mut open_backend: MemoryBackend) {
844        // Tampered entry: payload doesn't match the stored entry_hash. Scans must
845        // return HashMismatch rather than silently surfacing the corrupted row.
846        let mut tampered = build_entry(1, Headers::empty(), 10);
847        tampered.payload = Bytes::from_static(b"\xFF\xFF");
848        open_backend
849            .append_batch(&[AppendEntry::without_indices(tampered)])
850            .expect("append");
851
852        assert!(matches!(
853            open_backend.scan_seq(1),
854            Err(EventStoreError::HashMismatch { seq: 1 }),
855        ));
856        assert!(matches!(
857            open_backend.scan_range(1, 1, ScanDirection::Forward),
858            Err(EventStoreError::HashMismatch { seq: 1 }),
859        ));
860    }
861
862    #[rstest]
863    fn append_extracts_no_indices_when_keys_empty(mut open_backend: MemoryBackend) {
864        // Backend treats AppendEntry::index_keys as the sole authority. Headers on
865        // the entry are not auto-extracted; the writer/encoder is responsible.
866        let headers = Headers {
867            correlation_id: Some(UUID4::new()),
868            ..Headers::empty()
869        };
870        open_backend
871            .append_batch(&[AppendEntry::without_indices(build_entry(1, headers, 10))])
872            .expect("append");
873
874        assert!(
875            open_backend
876                .lookup(IndexKind::ClientOrderId, "any")
877                .expect("lookup")
878                .is_none(),
879        );
880    }
881
882    #[rstest]
883    fn iter_index_keys_enumerates_first_write_wins_pairs(mut open_backend: MemoryBackend) {
884        // Walks every (key, seq) pair the verifier needs to cross-check the
885        // sidecar indices: distinct kinds stay isolated, duplicate keys hold the
886        // first-seen seq, and unrelated kinds return empty without leaking pairs
887        // across kind boundaries.
888        open_backend
889            .append_batch(&[
890                AppendEntry::new(
891                    build_entry(1, Headers::empty(), 10),
892                    vec![
893                        IndexKey::new(IndexKind::ClientOrderId, "O-1".to_string()),
894                        IndexKey::new(IndexKind::VenueOrderId, "V-1".to_string()),
895                    ],
896                ),
897                AppendEntry::new(
898                    build_entry(2, Headers::empty(), 11),
899                    vec![
900                        // Re-emit O-1: first-write-wins must keep the seq=1 entry.
901                        IndexKey::new(IndexKind::ClientOrderId, "O-1".to_string()),
902                        IndexKey::new(IndexKind::ClientOrderId, "O-2".to_string()),
903                    ],
904                ),
905            ])
906            .expect("append");
907
908        let mut client = open_backend
909            .iter_index_keys(IndexKind::ClientOrderId)
910            .expect("iter");
911        client.sort();
912        assert_eq!(
913            client,
914            vec![("O-1".to_string(), 1u64), ("O-2".to_string(), 2u64)],
915        );
916
917        let venue = open_backend
918            .iter_index_keys(IndexKind::VenueOrderId)
919            .expect("iter");
920        assert_eq!(venue, vec![("V-1".to_string(), 1u64)]);
921    }
922
923    #[rstest]
924    fn iter_index_keys_errors_when_no_run_open() {
925        let backend = MemoryBackend::new();
926
927        match backend.iter_index_keys(IndexKind::ClientOrderId) {
928            Err(EventStoreError::Backend(msg)) => {
929                assert!(msg.contains("no run open"), "msg was: {msg}");
930            }
931            other => panic!("expected Backend, was {other:?}"),
932        }
933    }
934}
nautilus_event_store/backend/memory.rs

nautilus_event_store/backend/
memory.rs
