nautilus_event_store/markers/
redb.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//! redb-backed marker backend for the data marker sidecar.
17
18use std::{
19    fmt::Debug,
20    fs,
21    io::ErrorKind,
22    path::{Path, PathBuf},
23};
24
25use redb::{
26    CommitError, Database, DatabaseError, Durability, ReadOnlyDatabase, ReadTransaction,
27    ReadableDatabase, ReadableTable, StorageError, TableDefinition, TableError, TransactionError,
28    WriteTransaction,
29};
30use serde::{Serialize, de::DeserializeOwned};
31
32use crate::{
33    codec,
34    error::EventStoreError,
35    format,
36    manifest::RunStatus,
37    markers::{
38        DataCursorSnapshot, HiFiMarker, MarkerBackend, MarkerGap, MarkerManifest,
39        StoredMarkerRecord, StreamDictEntry,
40    },
41};
42
43const CURSOR_SNAPSHOTS_TABLE: TableDefinition<u64, &[u8]> =
44    TableDefinition::new("cursor_snapshots");
45const HIFI_MARKERS_TABLE: TableDefinition<u64, &[u8]> = TableDefinition::new("hifi_markers");
46const MARKER_GAPS_TABLE: TableDefinition<u64, &[u8]> = TableDefinition::new("marker_gaps");
47const STREAM_DICT_TABLE: TableDefinition<u32, &[u8]> = TableDefinition::new("stream_dict");
48const MARKER_MANIFEST_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("marker_manifest");
49
50const MANIFEST_KEY: &str = "current";
51
52/// On-disk [`MarkerBackend`] backed by a per-run [`redb`] marker sidecar file.
53///
54/// The caller supplies the full file path, normally
55/// `<base>/<instance_id>/<run_id>.markers.redb`. The backend creates the parent directory on
56/// first open, stores every durable record with its precomputed 32-byte integrity hash, and
57/// commits each mutation with [`Durability::Immediate`]. Reopening an unsealed marker file seals
58/// it as [`RunStatus::CrashedRecovered`] and returns `Ok(())`, because marker files never block
59/// trader boot.
60#[derive(Debug)]
61pub struct RedbMarkerBackend {
62    file_path: PathBuf,
63    state: Option<RunState>,
64}
65
66#[derive(Debug)]
67struct RunState {
68    db: MarkerDatabase,
69    manifest: MarkerManifest,
70    file_path: PathBuf,
71}
72
73enum MarkerDatabase {
74    ReadWrite(Database),
75    ReadOnly(ReadOnlyDatabase),
76}
77
78impl Debug for MarkerDatabase {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        match self {
81            Self::ReadWrite(_) => f.write_str("MarkerDatabase::ReadWrite"),
82            Self::ReadOnly(_) => f.write_str("MarkerDatabase::ReadOnly"),
83        }
84    }
85}
86
87impl MarkerDatabase {
88    fn readable(&self) -> &dyn ReadableDatabase {
89        match self {
90            Self::ReadWrite(db) => db,
91            Self::ReadOnly(db) => db,
92        }
93    }
94
95    fn read_write(&self) -> Result<&Database, EventStoreError> {
96        match self {
97            Self::ReadWrite(db) => Ok(db),
98            Self::ReadOnly(_) => Err(EventStoreError::Closed),
99        }
100    }
101
102    fn begin_read(&self) -> Result<ReadTransaction, EventStoreError> {
103        self.readable().begin_read().map_err(map_transaction_err)
104    }
105}
106
107impl RedbMarkerBackend {
108    /// Creates a backend for the supplied marker sidecar file path.
109    #[must_use]
110    pub fn new(file_path: impl Into<PathBuf>) -> Self {
111        Self {
112            file_path: file_path.into(),
113            state: None,
114        }
115    }
116
117    /// Returns the path of the currently open marker file.
118    ///
119    /// # Errors
120    ///
121    /// Returns [`EventStoreError::Backend`] when no run is open.
122    pub fn current_path(&self) -> Result<&Path, EventStoreError> {
123        Ok(self.state()?.file_path.as_path())
124    }
125
126    /// Opens an existing marker sidecar file read-only for verification.
127    ///
128    /// Marker verification must not seal, recover, or otherwise mutate the sidecar it scans. This
129    /// constructor accepts any manifest status and leaves lifecycle policy to the caller's report.
130    ///
131    /// # Errors
132    ///
133    /// Returns [`EventStoreError::Backend`] when the marker file is missing, and
134    /// [`EventStoreError::Corrupted`] when the file lacks a marker manifest or fails to decode.
135    pub fn open_read_only_file(path: impl Into<PathBuf>) -> Result<Self, EventStoreError> {
136        let path = path.into();
137        if !path.exists() {
138            return Err(EventStoreError::Backend(format!(
139                "no marker run file at {}",
140                path.display()
141            )));
142        }
143
144        let db = ReadOnlyDatabase::open(&path).map_err(map_database_err)?;
145        format::verify_store_format(&db)?;
146        let manifest = Self::read_manifest(&db)?.ok_or_else(|| {
147            EventStoreError::Corrupted(format!(
148                "missing marker manifest in run file at {}",
149                path.display()
150            ))
151        })?;
152
153        Ok(Self {
154            file_path: path.clone(),
155            state: Some(RunState {
156                db: MarkerDatabase::ReadOnly(db),
157                manifest,
158                file_path: path,
159            }),
160        })
161    }
162
163    fn state(&self) -> Result<&RunState, EventStoreError> {
164        self.state
165            .as_ref()
166            .ok_or_else(|| EventStoreError::Backend("no marker run open".to_string()))
167    }
168
169    fn state_mut(&mut self) -> Result<&mut RunState, EventStoreError> {
170        self.state
171            .as_mut()
172            .ok_or_else(|| EventStoreError::Backend("no marker run open".to_string()))
173    }
174
175    fn writable_state(&mut self) -> Result<&mut RunState, EventStoreError> {
176        let state = self.state_mut()?;
177        if state.manifest.is_sealed() {
178            return Err(EventStoreError::Closed);
179        }
180        state.db.read_write()?;
181        Ok(state)
182    }
183
184    fn initialize_fresh(db: &Database, manifest: &MarkerManifest) -> Result<(), EventStoreError> {
185        let txn = begin_immediate_write(db)?;
186        {
187            txn.open_table(CURSOR_SNAPSHOTS_TABLE)
188                .map_err(map_table_err)?;
189            txn.open_table(HIFI_MARKERS_TABLE).map_err(map_table_err)?;
190            txn.open_table(MARKER_GAPS_TABLE).map_err(map_table_err)?;
191            txn.open_table(STREAM_DICT_TABLE).map_err(map_table_err)?;
192        }
193        format::write_store_format(&txn)?;
194        insert_marker_manifest(&txn, manifest)?;
195        txn.commit().map_err(map_commit_err)?;
196        Ok(())
197    }
198
199    fn write_manifest(db: &Database, manifest: &MarkerManifest) -> Result<(), EventStoreError> {
200        let txn = begin_immediate_write(db)?;
201        insert_marker_manifest(&txn, manifest)?;
202        txn.commit().map_err(map_commit_err)?;
203        Ok(())
204    }
205
206    fn read_manifest<D: ReadableDatabase + ?Sized>(
207        db: &D,
208    ) -> Result<Option<MarkerManifest>, EventStoreError> {
209        let txn = db.begin_read().map_err(map_transaction_err)?;
210        let table = txn
211            .open_table(MARKER_MANIFEST_TABLE)
212            .map_err(map_table_err)?;
213        let Some(value) = table.get(MANIFEST_KEY).map_err(map_storage_err)? else {
214            return Ok(None);
215        };
216        let bytes = value.value();
217        Ok(Some(decode_value("marker manifest", bytes)?))
218    }
219}
220
221impl MarkerBackend for RedbMarkerBackend {
222    fn open_run(&mut self, mut manifest: MarkerManifest) -> Result<(), EventStoreError> {
223        self.state = None;
224
225        if let Some(parent) = self
226            .file_path
227            .parent()
228            .filter(|parent| !parent.as_os_str().is_empty())
229        {
230            fs::create_dir_all(parent).map_err(|e| {
231                let msg = format!("create dir {}: {e}", parent.display());
232
233                if is_disk_pressure(e.kind()) {
234                    EventStoreError::Disk(msg)
235                } else {
236                    EventStoreError::Backend(msg)
237                }
238            })?;
239        }
240
241        let path = self.file_path.clone();
242        let path_existed = path.exists();
243        let db = Database::create(&path).map_err(map_database_err)?;
244
245        if path_existed {
246            format::verify_store_format(&db)?;
247            let on_disk = Self::read_manifest(&db)?.ok_or_else(|| {
248                EventStoreError::Corrupted(format!(
249                    "missing marker manifest in existing run file at {}",
250                    path.display()
251                ))
252            })?;
253
254            if on_disk.run_id != manifest.run_id {
255                return Err(EventStoreError::Backend(format!(
256                    "marker run file at {} belongs to run_id {}, not {}",
257                    path.display(),
258                    on_disk.run_id,
259                    manifest.run_id
260                )));
261            }
262
263            let mut opened = on_disk;
264            if matches!(opened.status, RunStatus::Running) {
265                opened.status = RunStatus::CrashedRecovered;
266                Self::write_manifest(&db, &opened)?;
267            }
268
269            self.state = Some(RunState {
270                db: MarkerDatabase::ReadWrite(db),
271                manifest: opened,
272                file_path: path,
273            });
274            return Ok(());
275        }
276
277        manifest.status = RunStatus::Running;
278        manifest.snapshot_count = 0;
279        manifest.hifi_count = 0;
280        manifest.gap_count = 0;
281        manifest.dict_count = 0;
282        Self::initialize_fresh(&db, &manifest)?;
283
284        self.state = Some(RunState {
285            db: MarkerDatabase::ReadWrite(db),
286            manifest,
287            file_path: path,
288        });
289        Ok(())
290    }
291
292    fn append_snapshot(
293        &mut self,
294        snapshot: &DataCursorSnapshot,
295        hash: [u8; 32],
296    ) -> Result<(), EventStoreError> {
297        let state = self.writable_state()?;
298        let stored = StoredMarkerRecord {
299            record: snapshot.clone(),
300            hash,
301        };
302        let record_bytes = encode_value("cursor snapshot", &stored)?;
303        let mut updated = state.manifest.clone();
304        updated.snapshot_count += 1;
305        let db = state.db.read_write()?;
306        append_u64_marker_record(
307            db,
308            CURSOR_SNAPSHOTS_TABLE,
309            snapshot.marker_seq,
310            record_bytes.as_slice(),
311            &updated,
312        )?;
313
314        state.manifest = updated;
315        Ok(())
316    }
317
318    fn append_hifi(&mut self, marker: &HiFiMarker, hash: [u8; 32]) -> Result<(), EventStoreError> {
319        let state = self.writable_state()?;
320        let stored = StoredMarkerRecord {
321            record: marker.clone(),
322            hash,
323        };
324        let record_bytes = encode_value("hifi marker", &stored)?;
325        let mut updated = state.manifest.clone();
326        updated.hifi_count += 1;
327        let db = state.db.read_write()?;
328        append_u64_marker_record(
329            db,
330            HIFI_MARKERS_TABLE,
331            marker.marker_seq,
332            record_bytes.as_slice(),
333            &updated,
334        )?;
335
336        state.manifest = updated;
337        Ok(())
338    }
339
340    fn append_gap(&mut self, gap: &MarkerGap, hash: [u8; 32]) -> Result<(), EventStoreError> {
341        let state = self.writable_state()?;
342        let stored = StoredMarkerRecord {
343            record: gap.clone(),
344            hash,
345        };
346        let record_bytes = encode_value("marker gap", &stored)?;
347        let mut updated = state.manifest.clone();
348        updated.gap_count += 1;
349        let db = state.db.read_write()?;
350        append_u64_marker_record(
351            db,
352            MARKER_GAPS_TABLE,
353            gap.from_marker_seq,
354            record_bytes.as_slice(),
355            &updated,
356        )?;
357
358        state.manifest = updated;
359        Ok(())
360    }
361
362    fn put_dict(&mut self, entry: &StreamDictEntry, hash: [u8; 32]) -> Result<(), EventStoreError> {
363        let state = self.writable_state()?;
364        let stored = StoredMarkerRecord {
365            record: entry.clone(),
366            hash,
367        };
368        let record_bytes = encode_value("stream dict entry", &stored)?;
369        let mut updated = state.manifest.clone();
370        updated.dict_count += 1;
371        let mut inserted = false;
372
373        let db = state.db.read_write()?;
374        let txn = begin_immediate_write(db)?;
375        {
376            let mut table = txn.open_table(STREAM_DICT_TABLE).map_err(map_table_err)?;
377            let already = table.get(entry.slot).map_err(map_storage_err)?.is_some();
378
379            if !already {
380                table
381                    .insert(entry.slot, record_bytes.as_slice())
382                    .map_err(map_storage_err)?;
383                inserted = true;
384            }
385        }
386
387        if inserted {
388            insert_marker_manifest(&txn, &updated)?;
389        }
390
391        txn.commit().map_err(map_commit_err)?;
392
393        if inserted {
394            state.manifest = updated;
395        }
396        Ok(())
397    }
398
399    fn scan_snapshots(&self) -> Result<Vec<DataCursorSnapshot>, EventStoreError> {
400        let out = self
401            .scan_snapshot_records()?
402            .unwrap_or_default()
403            .into_iter()
404            .map(|stored| stored.record)
405            .collect();
406        Ok(out)
407    }
408
409    fn scan_snapshot_records(
410        &self,
411    ) -> Result<Option<Vec<StoredMarkerRecord<DataCursorSnapshot>>>, EventStoreError> {
412        let state = self.state()?;
413        let txn = state.db.begin_read()?;
414        let table = txn
415            .open_table(CURSOR_SNAPSHOTS_TABLE)
416            .map_err(map_table_err)?;
417        let iter = table.iter().map_err(map_storage_err)?;
418        let mut out = Vec::new();
419
420        for row in iter {
421            let (_, value) = row.map_err(map_storage_err)?;
422            let stored: StoredMarkerRecord<DataCursorSnapshot> =
423                decode_value("cursor snapshot", value.value())?;
424            out.push(stored);
425        }
426        Ok(Some(out))
427    }
428
429    fn scan_hifi(&self) -> Result<Vec<HiFiMarker>, EventStoreError> {
430        let out = self
431            .scan_hifi_records()?
432            .unwrap_or_default()
433            .into_iter()
434            .map(|stored| stored.record)
435            .collect();
436        Ok(out)
437    }
438
439    fn scan_hifi_records(
440        &self,
441    ) -> Result<Option<Vec<StoredMarkerRecord<HiFiMarker>>>, EventStoreError> {
442        let state = self.state()?;
443        let txn = state.db.begin_read()?;
444        let table = txn.open_table(HIFI_MARKERS_TABLE).map_err(map_table_err)?;
445        let iter = table.iter().map_err(map_storage_err)?;
446        let mut out = Vec::new();
447
448        for row in iter {
449            let (_, value) = row.map_err(map_storage_err)?;
450            let stored: StoredMarkerRecord<HiFiMarker> =
451                decode_value("hifi marker", value.value())?;
452            out.push(stored);
453        }
454        Ok(Some(out))
455    }
456
457    fn scan_gaps(&self) -> Result<Vec<MarkerGap>, EventStoreError> {
458        let out = self
459            .scan_gap_records()?
460            .unwrap_or_default()
461            .into_iter()
462            .map(|stored| stored.record)
463            .collect();
464        Ok(out)
465    }
466
467    fn scan_gap_records(
468        &self,
469    ) -> Result<Option<Vec<StoredMarkerRecord<MarkerGap>>>, EventStoreError> {
470        let state = self.state()?;
471        let txn = state.db.begin_read()?;
472        let table = txn.open_table(MARKER_GAPS_TABLE).map_err(map_table_err)?;
473        let iter = table.iter().map_err(map_storage_err)?;
474        let mut out = Vec::new();
475
476        for row in iter {
477            let (_, value) = row.map_err(map_storage_err)?;
478            let stored: StoredMarkerRecord<MarkerGap> = decode_value("marker gap", value.value())?;
479            out.push(stored);
480        }
481        Ok(Some(out))
482    }
483
484    fn scan_dict(&self) -> Result<Vec<StreamDictEntry>, EventStoreError> {
485        let out = self
486            .scan_dict_records()?
487            .unwrap_or_default()
488            .into_iter()
489            .map(|stored| stored.record)
490            .collect();
491        Ok(out)
492    }
493
494    fn scan_dict_records(
495        &self,
496    ) -> Result<Option<Vec<StoredMarkerRecord<StreamDictEntry>>>, EventStoreError> {
497        let state = self.state()?;
498        let txn = state.db.begin_read()?;
499        let table = txn.open_table(STREAM_DICT_TABLE).map_err(map_table_err)?;
500        let iter = table.iter().map_err(map_storage_err)?;
501        let mut out = Vec::new();
502
503        for row in iter {
504            let (_, value) = row.map_err(map_storage_err)?;
505            let stored: StoredMarkerRecord<StreamDictEntry> =
506                decode_value("stream dict entry", value.value())?;
507            out.push(stored);
508        }
509        Ok(Some(out))
510    }
511
512    fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
513        let state = self.state_mut()?;
514
515        if matches!(status, RunStatus::Running) {
516            return Err(EventStoreError::Backend(
517                "seal status must be a terminal state, was Running".to_string(),
518            ));
519        }
520
521        if state.manifest.is_sealed() {
522            return Err(EventStoreError::Closed);
523        }
524
525        let mut updated = state.manifest.clone();
526        updated.status = status;
527        let db = state.db.read_write()?;
528        Self::write_manifest(db, &updated)?;
529        state.manifest = updated;
530        Ok(())
531    }
532
533    fn manifest(&self) -> Result<MarkerManifest, EventStoreError> {
534        Ok(self.state()?.manifest.clone())
535    }
536}
537
538fn append_u64_marker_record(
539    db: &Database,
540    table_def: TableDefinition<u64, &[u8]>,
541    key: u64,
542    record_bytes: &[u8],
543    manifest: &MarkerManifest,
544) -> Result<(), EventStoreError> {
545    let txn = begin_immediate_write(db)?;
546    {
547        let mut table = txn.open_table(table_def).map_err(map_table_err)?;
548        table.insert(key, record_bytes).map_err(map_storage_err)?;
549    }
550    insert_marker_manifest(&txn, manifest)?;
551    txn.commit().map_err(map_commit_err)?;
552    Ok(())
553}
554
555fn begin_immediate_write(db: &Database) -> Result<WriteTransaction, EventStoreError> {
556    let mut txn = db.begin_write().map_err(map_transaction_err)?;
557    txn.set_durability(Durability::Immediate)
558        .map_err(|e| EventStoreError::Backend(format!("set durability: {e}")))?;
559    Ok(txn)
560}
561
562fn insert_marker_manifest(
563    txn: &WriteTransaction,
564    manifest: &MarkerManifest,
565) -> Result<(), EventStoreError> {
566    let manifest_bytes = encode_value("marker manifest", manifest)?;
567    let mut table = txn
568        .open_table(MARKER_MANIFEST_TABLE)
569        .map_err(map_table_err)?;
570    table
571        .insert(MANIFEST_KEY, manifest_bytes.as_slice())
572        .map_err(map_storage_err)?;
573    Ok(())
574}
575
576fn encode_value<T: Serialize>(label: &str, value: &T) -> Result<Vec<u8>, EventStoreError> {
577    codec::encode_to_vec(value)
578        .map_err(|e| EventStoreError::Backend(format!("encode {label}: {e}")))
579}
580
581fn decode_value<T: DeserializeOwned>(label: &str, bytes: &[u8]) -> Result<T, EventStoreError> {
582    codec::decode_from_slice::<T>(bytes)
583        .map_err(|e| EventStoreError::Corrupted(format!("decode {label}: {e}")))
584}
585
586fn map_storage_err(err: StorageError) -> EventStoreError {
587    match err {
588        StorageError::Io(io_err) if is_disk_pressure(io_err.kind()) => {
589            EventStoreError::Disk(io_err.to_string())
590        }
591        StorageError::Corrupted(msg) => EventStoreError::Corrupted(msg),
592        other => EventStoreError::Backend(other.to_string()),
593    }
594}
595
596fn is_disk_pressure(kind: ErrorKind) -> bool {
597    matches!(
598        kind,
599        ErrorKind::FileTooLarge | ErrorKind::StorageFull | ErrorKind::QuotaExceeded
600    )
601}
602
603fn map_database_err(err: DatabaseError) -> EventStoreError {
604    match err {
605        DatabaseError::RepairAborted => EventStoreError::Corrupted(
606            "database requires repair and cannot be verified read-only".to_string(),
607        ),
608        DatabaseError::UpgradeRequired(version) => EventStoreError::Corrupted(format!(
609            "database file format version {version} requires manual upgrade",
610        )),
611        DatabaseError::Storage(storage) => map_storage_err(storage),
612        other => EventStoreError::Backend(other.to_string()),
613    }
614}
615
616fn map_table_err(err: TableError) -> EventStoreError {
617    match err {
618        TableError::Storage(storage) => map_storage_err(storage),
619        TableError::TableDoesNotExist(_)
620        | TableError::TableTypeMismatch { .. }
621        | TableError::TableIsMultimap(_)
622        | TableError::TableIsNotMultimap(_)
623        | TableError::TypeDefinitionChanged { .. } => EventStoreError::Corrupted(err.to_string()),
624        other => EventStoreError::Backend(other.to_string()),
625    }
626}
627
628fn map_commit_err(err: CommitError) -> EventStoreError {
629    match err {
630        CommitError::Storage(storage) => map_storage_err(storage),
631        other => EventStoreError::Backend(other.to_string()),
632    }
633}
634
635fn map_transaction_err(err: TransactionError) -> EventStoreError {
636    match err {
637        TransactionError::Storage(storage) => map_storage_err(storage),
638        other => EventStoreError::Backend(other.to_string()),
639    }
640}
641
642#[cfg(test)]
643mod tests {
644    use std::{
645        fmt::Debug,
646        path::{Path, PathBuf},
647    };
648
649    use nautilus_core::UnixNanos;
650    use rstest::{fixture, rstest};
651    use tempfile::TempDir;
652
653    use super::RedbMarkerBackend;
654    use crate::{
655        error::EventStoreError,
656        format,
657        manifest::RunStatus,
658        markers::{
659            DataClass, DataCursorSnapshot, HiFiMarker, MarkerBackend, MarkerGap, MarkerGapReason,
660            MarkerManifest, MarkerVerifier, StreamCursor, StreamDictEntry, compute_dict_hash,
661            compute_gap_hash, compute_hifi_hash, compute_marker_hash,
662        },
663    };
664
665    fn marker_path(base: &Path, instance_id: &str, run_id: &str) -> PathBuf {
666        base.join(instance_id)
667            .join(format!("{run_id}.markers.redb"))
668    }
669
670    fn manifest(run_id: &str) -> MarkerManifest {
671        MarkerManifest {
672            run_id: run_id.to_string(),
673            enabled_classes: vec![DataClass::Quote, DataClass::Trade],
674            high_fidelity: true,
675            snapshot_count: 0,
676            hifi_count: 0,
677            gap_count: 0,
678            dict_count: 0,
679            status: RunStatus::Running,
680        }
681    }
682
683    fn snapshot(marker_seq: u64, event_seq_before: u64) -> DataCursorSnapshot {
684        DataCursorSnapshot {
685            marker_seq,
686            event_seq_before,
687            ts_init: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
688            advanced: vec![StreamCursor {
689                slot: 0,
690                ts_init_hi: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
691                count: marker_seq,
692            }],
693        }
694    }
695
696    fn hifi(marker_seq: u64) -> HiFiMarker {
697        HiFiMarker {
698            marker_seq,
699            event_seq_before: 42,
700            slot: 0,
701            ts_event: UnixNanos::from(1_700_000_000_000_000_100 + marker_seq),
702            ts_init: UnixNanos::from(1_700_000_000_000_000_200 + marker_seq),
703            same_ts_ordinal: 0,
704            record_fingerprint: [7u8; 32],
705        }
706    }
707
708    fn dict(slot: u32, data_cls: DataClass, identifier: &str) -> StreamDictEntry {
709        StreamDictEntry {
710            slot,
711            data_cls,
712            identifier: identifier.to_string(),
713        }
714    }
715
716    fn gap(from_marker_seq: u64, to_marker_seq: u64) -> MarkerGap {
717        MarkerGap {
718            from_marker_seq,
719            to_marker_seq,
720            reason: MarkerGapReason::Overflow,
721        }
722    }
723
724    fn assert_closed<T: Debug>(result: Result<T, EventStoreError>) {
725        match result {
726            Err(EventStoreError::Closed) => {}
727            other => panic!("expected Closed error, was {other:?}"),
728        }
729    }
730
731    fn assert_backend_message<T: Debug>(result: Result<T, EventStoreError>, expected: &str) {
732        match result {
733            Err(EventStoreError::Backend(message)) => assert_eq!(message, expected),
734            other => panic!("expected Backend error, was {other:?}"),
735        }
736    }
737
738    #[fixture]
739    fn temp_dir() -> TempDir {
740        TempDir::new().expect("create temp dir")
741    }
742
743    fn create_pre_codec_marker_file(path: &Path) {
744        let manifest: redb::TableDefinition<&str, &[u8]> =
745            redb::TableDefinition::new("marker_manifest");
746        let db = redb::Database::create(path).expect("create redb");
747        let txn = db.begin_write().expect("begin write");
748        {
749            let mut table = txn.open_table(manifest).expect("open marker manifest");
750            table
751                .insert("current", b"old-format".as_slice())
752                .expect("insert");
753        }
754        txn.commit().expect("commit");
755    }
756
757    fn create_future_marker_file(path: &Path) {
758        let manifest: redb::TableDefinition<&str, &[u8]> =
759            redb::TableDefinition::new("marker_manifest");
760        let store_format: redb::TableDefinition<&str, u32> =
761            redb::TableDefinition::new("store_format");
762        let db = redb::Database::create(path).expect("create redb");
763        let txn = db.begin_write().expect("begin write");
764        {
765            let mut format_table = txn.open_table(store_format).expect("open store format");
766            format_table
767                .insert("codec", format::STORE_FORMAT_VERSION + 1)
768                .expect("insert format");
769            let mut manifest_table = txn.open_table(manifest).expect("open marker manifest");
770            manifest_table
771                .insert("current", b"future-format".as_slice())
772                .expect("insert manifest");
773        }
774        txn.commit().expect("commit");
775    }
776
777    fn assert_corrupted_regenerated<T: Debug>(result: Result<T, EventStoreError>) {
778        match result {
779            Err(EventStoreError::Corrupted(msg)) => {
780                assert!(msg.contains("regenerated"), "msg was: {msg}");
781            }
782            other => panic!("expected Corrupted regeneration error, was {other:?}"),
783        }
784    }
785
786    fn assert_corrupted<T: Debug>(result: Result<T, EventStoreError>) {
787        match result {
788            Err(EventStoreError::Corrupted(_)) => {}
789            other => panic!("expected Corrupted error, was {other:?}"),
790        }
791    }
792
793    #[rstest]
794    fn open_rejects_marker_store_without_format_marker(temp_dir: TempDir) {
795        let run_id = "1700000000-redb-old-format";
796        let path = marker_path(temp_dir.path(), "trader-001", run_id);
797        std::fs::create_dir_all(path.parent().expect("parent")).expect("mkdir");
798        create_pre_codec_marker_file(&path);
799
800        let mut backend = RedbMarkerBackend::new(&path);
801        assert_corrupted_regenerated(backend.open_run(manifest(run_id)));
802        assert_corrupted_regenerated(RedbMarkerBackend::open_read_only_file(&path));
803    }
804
805    #[rstest]
806    fn open_rejects_marker_store_with_future_format_marker(temp_dir: TempDir) {
807        let run_id = "1700000000-redb-future-format";
808        let path = marker_path(temp_dir.path(), "trader-001", run_id);
809        std::fs::create_dir_all(path.parent().expect("parent")).expect("mkdir");
810        create_future_marker_file(&path);
811
812        let mut backend = RedbMarkerBackend::new(&path);
813        assert_corrupted(backend.open_run(manifest(run_id)));
814        assert_corrupted(RedbMarkerBackend::open_read_only_file(&path));
815    }
816
817    #[rstest]
818    fn roundtrip_snapshots_dict_gaps_through_redb(temp_dir: TempDir) {
819        let run_id = "1700000000-redb-roundtrip";
820        let path = marker_path(temp_dir.path(), "trader-001", run_id);
821        let s2 = snapshot(2, 20);
822        let s1 = snapshot(1, 10);
823        let h2 = hifi(4);
824        let h1 = hifi(3);
825        let g2 = gap(20, 24);
826        let g1 = gap(5, 8);
827        let first_dict = dict(0, DataClass::Quote, "ETHUSDT.BINANCE");
828        let remap_dict = dict(0, DataClass::Trade, "BTCUSDT.BINANCE");
829        let second_dict = dict(1, DataClass::Trade, "BTCUSDT.BINANCE");
830
831        {
832            let mut backend = RedbMarkerBackend::new(&path);
833            backend.open_run(manifest(run_id)).expect("open run");
834            backend
835                .append_snapshot(&s2, compute_marker_hash(&s2))
836                .expect("append snapshot 2");
837            backend
838                .append_snapshot(&s1, compute_marker_hash(&s1))
839                .expect("append snapshot 1");
840            backend
841                .append_hifi(&h2, compute_hifi_hash(&h2))
842                .expect("append hifi 4");
843            backend
844                .append_hifi(&h1, compute_hifi_hash(&h1))
845                .expect("append hifi 3");
846            backend
847                .append_gap(&g2, compute_gap_hash(&g2))
848                .expect("append gap 2");
849            backend
850                .append_gap(&g1, compute_gap_hash(&g1))
851                .expect("append gap 1");
852            backend
853                .put_dict(&first_dict, compute_dict_hash(&first_dict))
854                .expect("put dict 0");
855            backend
856                .put_dict(&remap_dict, compute_dict_hash(&remap_dict))
857                .expect("re-put dict 0");
858            backend
859                .put_dict(&second_dict, compute_dict_hash(&second_dict))
860                .expect("put dict 1");
861        }
862
863        let mut reopened = RedbMarkerBackend::new(&path);
864        reopened.open_run(manifest(run_id)).expect("reopen run");
865
866        assert_eq!(
867            reopened.scan_snapshots().expect("scan snapshots"),
868            vec![s1, s2]
869        );
870        assert_eq!(reopened.scan_hifi().expect("scan hifi"), vec![h1, h2]);
871        assert_eq!(reopened.scan_gaps().expect("scan gaps"), vec![g1, g2]);
872        assert_eq!(
873            reopened.scan_dict().expect("scan dict"),
874            vec![first_dict, second_dict]
875        );
876    }
877
878    #[rstest]
879    fn open_running_then_reopen_reports_crashed(temp_dir: TempDir) {
880        let run_id = "1700000000-redb-crash";
881        let path = marker_path(temp_dir.path(), "trader-001", run_id);
882        let s1 = snapshot(1, 10);
883
884        {
885            let mut backend = RedbMarkerBackend::new(&path);
886            backend.open_run(manifest(run_id)).expect("open run");
887            backend
888                .append_snapshot(&s1, compute_marker_hash(&s1))
889                .expect("append snapshot");
890        }
891
892        let mut reopened = RedbMarkerBackend::new(&path);
893        reopened.open_run(manifest(run_id)).expect("reopen run");
894
895        assert_eq!(
896            reopened.manifest().expect("manifest").status,
897            RunStatus::CrashedRecovered
898        );
899        assert_eq!(reopened.scan_snapshots().expect("scan snapshots"), vec![s1]);
900    }
901
902    #[rstest]
903    fn read_only_marker_backend_rejects_writes_without_recovery_mutation(temp_dir: TempDir) {
904        let run_id = "1700000000-redb-read-only-running";
905        let path = marker_path(temp_dir.path(), "trader-001", run_id);
906        let s1 = snapshot(1, 10);
907
908        {
909            let mut backend = RedbMarkerBackend::new(&path);
910            backend.open_run(manifest(run_id)).expect("open run");
911            backend
912                .append_snapshot(&s1, compute_marker_hash(&s1))
913                .expect("append snapshot");
914        }
915
916        let mut read_only =
917            RedbMarkerBackend::open_read_only_file(&path).expect("open read-only marker file");
918        let s2 = snapshot(2, 11);
919
920        assert_eq!(
921            read_only.manifest().expect("manifest").status,
922            RunStatus::Running,
923        );
924        assert_closed(read_only.append_snapshot(&s2, compute_marker_hash(&s2)));
925        assert_closed(read_only.seal(RunStatus::Ended));
926        assert_eq!(
927            read_only.scan_snapshots().expect("scan snapshots"),
928            vec![s1]
929        );
930    }
931
932    #[rstest]
933    fn seal_persists_manifest_counts(temp_dir: TempDir) {
934        let run_id = "1700000000-redb-seal";
935        let path = marker_path(temp_dir.path(), "trader-001", run_id);
936        let s1 = snapshot(1, 10);
937        let h1 = hifi(2);
938        let g1 = gap(3, 5);
939        let d1 = dict(0, DataClass::Quote, "ETHUSDT.BINANCE");
940
941        {
942            let mut backend = RedbMarkerBackend::new(&path);
943            backend.open_run(manifest(run_id)).expect("open run");
944            backend
945                .append_snapshot(&s1, compute_marker_hash(&s1))
946                .expect("append snapshot");
947            backend
948                .append_hifi(&h1, compute_hifi_hash(&h1))
949                .expect("append hifi");
950            backend
951                .append_gap(&g1, compute_gap_hash(&g1))
952                .expect("append gap");
953            backend
954                .put_dict(&d1, compute_dict_hash(&d1))
955                .expect("put dict");
956            backend.seal(RunStatus::Ended).expect("seal");
957        }
958
959        let mut reopened = RedbMarkerBackend::new(&path);
960        reopened
961            .open_run(manifest(run_id))
962            .expect("reopen sealed run");
963        let persisted = reopened.manifest().expect("manifest");
964
965        assert_eq!(persisted.status, RunStatus::Ended);
966        assert_eq!(persisted.snapshot_count, 1);
967        assert_eq!(persisted.hifi_count, 1);
968        assert_eq!(persisted.gap_count, 1);
969        assert_eq!(persisted.dict_count, 1);
970    }
971
972    #[rstest]
973    fn unsealed_appends_persist_manifest_counts_for_verifier(temp_dir: TempDir) {
974        let run_id = "1700000000-redb-unsealed-counts";
975        let path = marker_path(temp_dir.path(), "trader-001", run_id);
976        let s1 = snapshot(1, 1);
977        let h1 = hifi(2);
978        let g1 = gap(3, 3);
979        let first_dict = dict(0, DataClass::Quote, "ETHUSDT.BINANCE");
980        let remap_dict = dict(0, DataClass::Trade, "BTCUSDT.BINANCE");
981        let second_dict = dict(1, DataClass::Trade, "BTCUSDT.BINANCE");
982
983        {
984            let mut backend = RedbMarkerBackend::new(&path);
985            backend.open_run(manifest(run_id)).expect("open run");
986            backend
987                .append_snapshot(&s1, compute_marker_hash(&s1))
988                .expect("append snapshot");
989            backend
990                .append_hifi(&h1, compute_hifi_hash(&h1))
991                .expect("append hifi");
992            backend
993                .append_gap(&g1, compute_gap_hash(&g1))
994                .expect("append gap");
995            backend
996                .put_dict(&first_dict, compute_dict_hash(&first_dict))
997                .expect("put dict 0");
998            backend
999                .put_dict(&remap_dict, compute_dict_hash(&remap_dict))
1000                .expect("re-put dict 0");
1001            backend
1002                .put_dict(&second_dict, compute_dict_hash(&second_dict))
1003                .expect("put dict 1");
1004        }
1005
1006        let read_only =
1007            RedbMarkerBackend::open_read_only_file(&path).expect("open read-only marker file");
1008        let persisted = read_only.manifest().expect("manifest");
1009        let report = MarkerVerifier::scan(&read_only, 42).expect("scan");
1010
1011        assert_eq!(persisted.status, RunStatus::Running);
1012        assert_eq!(persisted.snapshot_count, 1);
1013        assert_eq!(persisted.hifi_count, 1);
1014        assert_eq!(persisted.gap_count, 1);
1015        assert_eq!(persisted.dict_count, 2);
1016        assert_eq!(report.snapshots_scanned, 1);
1017        assert_eq!(report.hifi_scanned, 1);
1018        assert_eq!(report.gaps_scanned, 1);
1019        assert_eq!(report.dict_entries_scanned, 2);
1020        assert!(report.is_clean(), "findings was: {:?}", report.findings);
1021    }
1022
1023    #[rstest]
1024    fn sealed_backend_rejects_writes(temp_dir: TempDir) {
1025        let run_id = "1700000000-redb-sealed-writes";
1026        let path = marker_path(temp_dir.path(), "trader-001", run_id);
1027        let mut backend = RedbMarkerBackend::new(&path);
1028        backend.open_run(manifest(run_id)).expect("open run");
1029        backend.seal(RunStatus::Ended).expect("seal");
1030
1031        let s1 = snapshot(1, 10);
1032        let h1 = hifi(2);
1033        let g1 = gap(3, 5);
1034        let d1 = dict(0, DataClass::Quote, "ETHUSDT.BINANCE");
1035
1036        assert_closed(backend.append_snapshot(&s1, compute_marker_hash(&s1)));
1037        assert_closed(backend.append_hifi(&h1, compute_hifi_hash(&h1)));
1038        assert_closed(backend.append_gap(&g1, compute_gap_hash(&g1)));
1039        assert_closed(backend.put_dict(&d1, compute_dict_hash(&d1)));
1040
1041        let persisted = backend.manifest().expect("manifest");
1042        assert_eq!(persisted.status, RunStatus::Ended);
1043        assert_eq!(persisted.snapshot_count, 0);
1044        assert_eq!(persisted.hifi_count, 0);
1045        assert_eq!(persisted.gap_count, 0);
1046        assert_eq!(persisted.dict_count, 0);
1047    }
1048
1049    #[rstest]
1050    fn crash_recovered_backend_rejects_writes(temp_dir: TempDir) {
1051        let run_id = "1700000000-redb-crash-closed";
1052        let path = marker_path(temp_dir.path(), "trader-001", run_id);
1053        let s1 = snapshot(1, 10);
1054        let s2 = snapshot(2, 20);
1055
1056        {
1057            let mut backend = RedbMarkerBackend::new(&path);
1058            backend.open_run(manifest(run_id)).expect("open run");
1059            backend
1060                .append_snapshot(&s1, compute_marker_hash(&s1))
1061                .expect("append snapshot");
1062        }
1063
1064        let mut reopened = RedbMarkerBackend::new(&path);
1065        reopened.open_run(manifest(run_id)).expect("reopen run");
1066
1067        assert_eq!(
1068            reopened.manifest().expect("manifest").status,
1069            RunStatus::CrashedRecovered
1070        );
1071        assert_closed(reopened.append_snapshot(&s2, compute_marker_hash(&s2)));
1072        assert_eq!(reopened.scan_snapshots().expect("scan snapshots"), vec![s1]);
1073    }
1074
1075    #[rstest]
1076    fn seal_running_is_rejected_and_does_not_close(temp_dir: TempDir) {
1077        let run_id = "1700000000-redb-running-seal";
1078        let path = marker_path(temp_dir.path(), "trader-001", run_id);
1079        let s1 = snapshot(1, 10);
1080        let mut backend = RedbMarkerBackend::new(&path);
1081        backend.open_run(manifest(run_id)).expect("open run");
1082
1083        assert_backend_message(
1084            backend.seal(RunStatus::Running),
1085            "seal status must be a terminal state, was Running",
1086        );
1087        backend
1088            .append_snapshot(&s1, compute_marker_hash(&s1))
1089            .expect("append after rejected seal");
1090        backend.seal(RunStatus::Ended).expect("seal ended");
1091
1092        let persisted = backend.manifest().expect("manifest");
1093        assert_eq!(persisted.status, RunStatus::Ended);
1094        assert_eq!(persisted.snapshot_count, 1);
1095    }
1096
1097    #[rstest]
1098    fn unopened_backend_reports_backend_error(temp_dir: TempDir) {
1099        let path = marker_path(temp_dir.path(), "trader-001", "1700000000-redb-unopened");
1100        let mut backend = RedbMarkerBackend::new(&path);
1101        let s1 = snapshot(1, 10);
1102        let h1 = hifi(2);
1103        let g1 = gap(3, 5);
1104        let d1 = dict(0, DataClass::Quote, "ETHUSDT.BINANCE");
1105
1106        assert_backend_message(backend.current_path(), "no marker run open");
1107        assert_backend_message(backend.manifest(), "no marker run open");
1108        assert_backend_message(backend.scan_snapshots(), "no marker run open");
1109        assert_backend_message(backend.scan_hifi(), "no marker run open");
1110        assert_backend_message(backend.scan_gaps(), "no marker run open");
1111        assert_backend_message(backend.scan_dict(), "no marker run open");
1112        assert_backend_message(backend.seal(RunStatus::Ended), "no marker run open");
1113        assert_backend_message(
1114            backend.append_snapshot(&s1, compute_marker_hash(&s1)),
1115            "no marker run open",
1116        );
1117        assert_backend_message(
1118            backend.append_hifi(&h1, compute_hifi_hash(&h1)),
1119            "no marker run open",
1120        );
1121        assert_backend_message(
1122            backend.append_gap(&g1, compute_gap_hash(&g1)),
1123            "no marker run open",
1124        );
1125        assert_backend_message(
1126            backend.put_dict(&d1, compute_dict_hash(&d1)),
1127            "no marker run open",
1128        );
1129    }
1130}
nautilus_event_store/markers/redb.rs

nautilus_event_store/markers/
redb.rs
