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

nautilus_event_store/backend/
memory.rs
