nautilus_event_store/backend/
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 [`EventStore`] implementation.
17//!
18//! One redb file per run at `<base>/<instance_id>/<run_id>.redb`. Every commit uses
19//! [`Durability::Immediate`] so a crashed writer never leaves the in-flight tail visible
20//! after reopen, and the high-watermark only advances after a durable acknowledgement.
21
22use std::{
23    fmt::Debug,
24    fs,
25    io::ErrorKind,
26    path::{Path, PathBuf},
27};
28
29use nautilus_core::UnixNanos;
30use redb::{
31    CommitError, Database, DatabaseError, Durability, ReadOnlyDatabase, ReadTransaction,
32    ReadableDatabase, ReadableTable, StorageError, TableDefinition, TableError, TransactionError,
33    WriteTransaction,
34};
35
36use crate::{
37    backend::{AppendEntry, EventStore, IndexKey, IndexKind, ScanDirection},
38    codec,
39    entry::EventStoreEntry,
40    error::EventStoreError,
41    format,
42    manifest::{RunManifest, RunStatus},
43    snapshot::{SnapshotAnchor, validate_new_anchor},
44};
45
46const ENTRIES_TABLE: TableDefinition<u64, &[u8]> = TableDefinition::new("entries");
47const MANIFEST_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("manifest");
48const CLIENT_ORDER_INDEX: TableDefinition<&str, u64> = TableDefinition::new("client_order_id_idx");
49const VENUE_ORDER_INDEX: TableDefinition<&str, u64> = TableDefinition::new("venue_order_id_idx");
50const SNAPSHOT_ANCHOR_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("snapshot_anchor");
51
52const MANIFEST_KEY: &str = "current";
53const SNAPSHOT_ANCHOR_KEY: &str = "latest";
54
55/// On-disk [`EventStore`] backed by a per-run [`redb`] file.
56///
57/// One backend instance owns at most one open run at a time. Opening a fresh run creates
58/// `<base>/<instance_id>/<run_id>.redb` and writes the manifest with status
59/// [`RunStatus::Running`] before returning. Reopening a path whose manifest is still
60/// [`RunStatus::Running`] returns [`EventStoreError::CrashedPredecessor`]; the caller seals
61/// it as [`RunStatus::CrashedRecovered`] (or [`RunStatus::Quarantined`]) and then opens a new
62/// run, mirroring the in-memory backend's contract.
63#[derive(Debug)]
64pub struct RedbBackend {
65    base_dir: PathBuf,
66    state: Option<RunState>,
67}
68
69#[derive(Debug)]
70struct RunState {
71    db: RunDatabase,
72    manifest: RunManifest,
73    high_watermark: u64,
74    max_ts_init: UnixNanos,
75    file_path: PathBuf,
76}
77
78enum RunDatabase {
79    ReadWrite(Database),
80    ReadOnly(ReadOnlyDatabase),
81}
82
83impl Debug for RunDatabase {
84    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85        match self {
86            Self::ReadWrite(_) => f.write_str("RunDatabase::ReadWrite"),
87            Self::ReadOnly(_) => f.write_str("RunDatabase::ReadOnly"),
88        }
89    }
90}
91
92impl RunDatabase {
93    fn readable(&self) -> &dyn ReadableDatabase {
94        match self {
95            Self::ReadWrite(db) => db,
96            Self::ReadOnly(db) => db,
97        }
98    }
99
100    fn read_write(&self) -> Result<&Database, EventStoreError> {
101        match self {
102            Self::ReadWrite(db) => Ok(db),
103            Self::ReadOnly(_) => Err(EventStoreError::Closed),
104        }
105    }
106
107    fn begin_read(&self) -> Result<ReadTransaction, EventStoreError> {
108        self.readable().begin_read().map_err(map_transaction_err)
109    }
110}
111
112impl RedbBackend {
113    /// Creates a new [`RedbBackend`] rooted at `base_dir`.
114    ///
115    /// The backend creates `<base_dir>/<instance_id>/` lazily on the first
116    /// [`EventStore::open_run`] call.
117    #[must_use]
118    pub fn new(base_dir: impl Into<PathBuf>) -> Self {
119        Self {
120            base_dir: base_dir.into(),
121            state: None,
122        }
123    }
124
125    /// Returns the directory the backend writes run files to for `instance_id`.
126    #[must_use]
127    pub fn run_dir(&self, instance_id: &str) -> PathBuf {
128        self.base_dir.join(instance_id)
129    }
130
131    /// Returns the on-disk path the backend uses for `(instance_id, run_id)`.
132    #[must_use]
133    pub fn run_path(&self, instance_id: &str, run_id: &str) -> PathBuf {
134        self.run_dir(instance_id).join(format!("{run_id}.redb"))
135    }
136
137    /// Returns the path of the currently open run file.
138    ///
139    /// # Errors
140    ///
141    /// Returns [`EventStoreError::Backend`] when no run is open.
142    pub fn current_path(&self) -> Result<&Path, EventStoreError> {
143        Ok(self.state()?.file_path.as_path())
144    }
145
146    /// Opens the sealed run file at `<base>/<instance_id>/<run_id>.redb` for read-only replay.
147    ///
148    /// # Design
149    ///
150    /// The standard [`EventStore::open_run`] path rejects sealed files: that is the
151    /// crash-recovery guard, a successor must not silently reopen a predecessor's log
152    /// without going through seal. Event-store replay is the legitimate case for touching
153    /// a sealed file, so the reader uses this constructor instead.
154    ///
155    /// The shared [`EventStore`] trait is held intentionally narrow and is locked by
156    /// design; adding a sealed-open method to it would force the in-memory backend
157    /// (whose sealed runs stay readable in place without a reopen step) to carry a
158    /// useless second entry point, and would conflate the writer's open-or-recover
159    /// lifecycle with the reader's pure read-only path. The sealed-open path therefore
160    /// lives as a backend-specific constructor: each backend adds the entry points it
161    /// actually needs. The resulting [`RedbBackend`] still implements [`EventStore`],
162    /// so the reader composes over the locked trait without pulling in writer-only
163    /// methods. [`crate::backend::MemoryBackend`] has no equivalent constructor: a
164    /// sealed in-memory run keeps its state accessible to any reader holding the
165    /// backend instance, and the reader receives that instance directly.
166    ///
167    /// The returned backend holds a read-only database handle, rejects
168    /// [`EventStore::append_batch`] with [`EventStoreError::Closed`] (the manifest is
169    /// already sealed), and exposes every read path: [`EventStore::scan_range`],
170    /// [`EventStore::scan_seq`], [`EventStore::lookup`], and [`EventStore::manifest`].
171    ///
172    /// # Errors
173    ///
174    /// Returns [`EventStoreError::Backend`] when the run file does not exist or its
175    /// status is not a sealed terminal state (use [`EventStore::open_run`] for that
176    /// path); [`EventStoreError::Corrupted`] when the run file lacks a manifest or
177    /// fails to decode.
178    pub fn open_sealed(
179        base_dir: impl Into<PathBuf>,
180        instance_id: &str,
181        run_id: &str,
182    ) -> Result<Self, EventStoreError> {
183        let base = base_dir.into();
184        let path = base.join(instance_id).join(format!("{run_id}.redb"));
185        Self::open_sealed_path(base, path)
186    }
187
188    /// Opens a sealed redb run file directly by path for read-only replay or verification.
189    ///
190    /// # Errors
191    ///
192    /// Returns [`EventStoreError::Backend`] when the run file does not exist or its
193    /// status is not a sealed terminal state (use [`EventStore::open_run`] for that
194    /// path); [`EventStoreError::Corrupted`] when the run file lacks a manifest or
195    /// fails to decode.
196    pub fn open_sealed_file(path: impl Into<PathBuf>) -> Result<Self, EventStoreError> {
197        let path = path.into();
198        let base = path
199            .parent()
200            .and_then(Path::parent)
201            .map_or_else(PathBuf::new, Path::to_path_buf);
202        Self::open_sealed_path(base, path)
203    }
204
205    fn open_sealed_path(base: PathBuf, path: PathBuf) -> Result<Self, EventStoreError> {
206        if !path.exists() {
207            return Err(EventStoreError::Backend(format!(
208                "no run file at {}",
209                path.display()
210            )));
211        }
212
213        let db = ReadOnlyDatabase::open(&path).map_err(map_read_only_database_err)?;
214        format::verify_store_format(&db)?;
215        let manifest = Self::read_manifest(&db)?.ok_or_else(|| {
216            EventStoreError::Corrupted(format!(
217                "missing manifest in run file at {}",
218                path.display()
219            ))
220        })?;
221
222        if !manifest.is_sealed() {
223            return Err(EventStoreError::Backend(format!(
224                "run file at {} is not sealed, status was {:?}",
225                path.display(),
226                manifest.status,
227            )));
228        }
229        let (high_watermark, max_ts_init) = Self::compute_progress(&db)?;
230
231        Ok(Self {
232            base_dir: base,
233            state: Some(RunState {
234                db: RunDatabase::ReadOnly(db),
235                manifest,
236                high_watermark,
237                max_ts_init,
238                file_path: path,
239            }),
240        })
241    }
242
243    /// Lists the manifests of every run file under `<base_dir>/<instance_id>/*.redb`.
244    ///
245    /// Used by the reader for forensics navigation across runs without requiring an
246    /// active backend instance per run. The result is sorted by `start_ts_init` so
247    /// chronologically-newer runs appear last.
248    ///
249    /// Opens each run file with a read-only database handle. A run file whose process
250    /// died hard (kill, OOM, power loss) lacks redb's allocator-state table and refuses
251    /// the read-only open; the listing falls back to a writable open, which performs
252    /// redb's repair pass and leaves the file readable again. Files that still cannot
253    /// be opened or that lack a manifest are skipped with a logged error so one
254    /// current-format damaged file cannot block recovery or retention over the healthy
255    /// runs; such files never become recovery parents or reclaim candidates and are
256    /// left in place for manual inspection. Unsupported store formats are returned as
257    /// errors rather than skipped because they require operator action.
258    ///
259    /// # Errors
260    ///
261    /// Returns [`EventStoreError::Backend`] when the directory iterator fails, or
262    /// [`EventStoreError::Corrupted`] when a run file uses an unsupported store format.
263    pub fn list_runs(
264        base_dir: &Path,
265        instance_id: &str,
266    ) -> Result<Vec<RunManifest>, EventStoreError> {
267        let dir = base_dir.join(instance_id);
268        let entries = match fs::read_dir(&dir) {
269            Ok(it) => it,
270            Err(e) if e.kind() == ErrorKind::NotFound => return Ok(Vec::new()),
271            Err(e) => {
272                return Err(EventStoreError::Backend(format!(
273                    "read_dir {}: {e}",
274                    dir.display()
275                )));
276            }
277        };
278
279        let mut manifests = Vec::new();
280
281        for entry in entries {
282            let entry = entry.map_err(|e| {
283                EventStoreError::Backend(format!("read_dir entry in {}: {e}", dir.display()))
284            })?;
285            let path = entry.path();
286
287            if !is_run_file(&path) {
288                continue;
289            }
290
291            match Self::read_run_manifest(&path) {
292                Ok(manifest) => manifests.push(manifest),
293                Err(e) if format::is_unsupported_store_format(&e) => return Err(e),
294                Err(e) => {
295                    log::error!("Skipping unreadable run file {}: {e}", path.display());
296                }
297            }
298        }
299        // Break start-time ties on the run id so parent selection and retention stay
300        // deterministic: stable sort alone preserves the platform-dependent `read_dir`
301        // order.
302        manifests.sort_by(|a, b| {
303            a.start_ts_init
304                .cmp(&b.start_ts_init)
305                .then_with(|| a.run_id.cmp(&b.run_id))
306        });
307        Ok(manifests)
308    }
309
310    fn read_run_manifest(path: &Path) -> Result<RunManifest, EventStoreError> {
311        let manifest = match ReadOnlyDatabase::open(path) {
312            Ok(db) => {
313                Self::verify_listed_store_format(&db, path)?;
314                Self::read_manifest(&db)?
315            }
316            // Each durable commit deletes redb's allocator-state table and only a clean
317            // `Database::drop` rewrites it, so a hard-killed process leaves a file the
318            // read-only open refuses. A writable open repairs it for future opens.
319            Err(DatabaseError::RepairAborted) => {
320                log::warn!(
321                    "Run file {} was not shut down cleanly, repairing",
322                    path.display()
323                );
324                let db = Database::open(path).map_err(map_database_err)?;
325                Self::verify_listed_store_format(&db, path)?;
326                Self::read_manifest(&db)?
327            }
328            Err(e) => return Err(map_read_only_database_err(e)),
329        };
330        manifest.ok_or_else(|| missing_manifest(path))
331    }
332
333    fn verify_listed_store_format<D: ReadableDatabase + ?Sized>(
334        db: &D,
335        path: &Path,
336    ) -> Result<(), EventStoreError> {
337        match format::verify_store_format(db) {
338            Ok(()) => Ok(()),
339            Err(e) if format::is_missing_store_format(&e) && !Self::manifest_row_exists(db)? => {
340                Err(missing_manifest(path))
341            }
342            Err(e) => Err(e),
343        }
344    }
345
346    fn manifest_row_exists<D: ReadableDatabase + ?Sized>(db: &D) -> Result<bool, EventStoreError> {
347        let txn = db.begin_read().map_err(map_transaction_err)?;
348        let table = match txn.open_table(MANIFEST_TABLE) {
349            Ok(table) => table,
350            Err(TableError::TableDoesNotExist(_)) => return Ok(false),
351            Err(e) => return Err(map_table_err(e)),
352        };
353
354        table
355            .get(MANIFEST_KEY)
356            .map(|value| value.is_some())
357            .map_err(map_storage_err)
358    }
359
360    fn state(&self) -> Result<&RunState, EventStoreError> {
361        self.state
362            .as_ref()
363            .ok_or_else(|| EventStoreError::Backend("no run open".to_string()))
364    }
365
366    fn state_mut(&mut self) -> Result<&mut RunState, EventStoreError> {
367        self.state
368            .as_mut()
369            .ok_or_else(|| EventStoreError::Backend("no run open".to_string()))
370    }
371
372    fn initialize_fresh(db: &Database, manifest: &RunManifest) -> Result<(), EventStoreError> {
373        let txn = begin_immediate_write(db)?;
374        {
375            txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
376            txn.open_table(CLIENT_ORDER_INDEX).map_err(map_table_err)?;
377            txn.open_table(VENUE_ORDER_INDEX).map_err(map_table_err)?;
378            txn.open_table(SNAPSHOT_ANCHOR_TABLE)
379                .map_err(map_table_err)?;
380        }
381        format::write_store_format(&txn)?;
382        insert_run_manifest(&txn, manifest)?;
383        txn.commit().map_err(map_commit_err)?;
384        Ok(())
385    }
386
387    fn write_manifest(db: &Database, manifest: &RunManifest) -> Result<(), EventStoreError> {
388        let txn = begin_immediate_write(db)?;
389        insert_run_manifest(&txn, manifest)?;
390        txn.commit().map_err(map_commit_err)?;
391        Ok(())
392    }
393
394    fn read_manifest<D: ReadableDatabase + ?Sized>(
395        db: &D,
396    ) -> Result<Option<RunManifest>, EventStoreError> {
397        let txn = db.begin_read().map_err(map_transaction_err)?;
398        let table = txn.open_table(MANIFEST_TABLE).map_err(map_table_err)?;
399        let Some(value) = table.get(MANIFEST_KEY).map_err(map_storage_err)? else {
400            return Ok(None);
401        };
402        let bytes = value.value();
403        let manifest = codec::decode_from_slice::<RunManifest>(bytes)
404            .map_err(|e| EventStoreError::Corrupted(format!("decode manifest: {e}")))?;
405        Ok(Some(manifest))
406    }
407
408    fn read_snapshot_anchor<D: ReadableDatabase + ?Sized>(
409        db: &D,
410    ) -> Result<Option<SnapshotAnchor>, EventStoreError> {
411        let txn = db.begin_read().map_err(map_transaction_err)?;
412        let table = match txn.open_table(SNAPSHOT_ANCHOR_TABLE) {
413            Ok(table) => table,
414            Err(TableError::TableDoesNotExist(_)) => return Ok(None),
415            Err(e) => return Err(map_table_err(e)),
416        };
417        let Some(value) = table.get(SNAPSHOT_ANCHOR_KEY).map_err(map_storage_err)? else {
418            return Ok(None);
419        };
420        let bytes = value.value();
421        let anchor = codec::decode_from_slice::<SnapshotAnchor>(bytes)
422            .map_err(|e| EventStoreError::Corrupted(format!("decode snapshot anchor: {e}")))?;
423        Ok(Some(anchor))
424    }
425
426    fn compute_progress<D: ReadableDatabase + ?Sized>(
427        db: &D,
428    ) -> Result<(u64, UnixNanos), EventStoreError> {
429        let txn = db.begin_read().map_err(map_transaction_err)?;
430        let table = txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
431
432        let Some((last_key, _)) = table.last().map_err(map_storage_err)? else {
433            return Ok((0, UnixNanos::default()));
434        };
435        let high_watermark = last_key.value();
436
437        // Walk the entry table once to recover the maximum `ts_init`. Memory.rs tracks this
438        // across appends; on crash recovery we have nothing to fall back on, so we recompute
439        // it from the durable rows. An undecodable row must not make the run unopenable:
440        // max ts_init is best-effort, and the corruption itself surfaces on the scan paths,
441        // where the recovery sweep quarantines the run.
442        let mut max_ts = UnixNanos::default();
443        let iter = table.iter().map_err(map_storage_err)?;
444
445        for row in iter {
446            let (key, value) = row.map_err(map_storage_err)?;
447            let bytes = value.value();
448
449            match codec::decode_from_slice::<EventStoreEntry>(bytes) {
450                Ok(entry) => {
451                    if entry.ts_init > max_ts {
452                        max_ts = entry.ts_init;
453                    }
454                }
455                Err(e) => {
456                    log::error!("Undecodable entry at seq {} on load: {e}", key.value());
457                }
458            }
459        }
460
461        Ok((high_watermark, max_ts))
462    }
463}
464
465impl EventStore for RedbBackend {
466    fn open_run(&mut self, mut manifest: RunManifest) -> Result<(), EventStoreError> {
467        if let Some(state) = &self.state {
468            if matches!(state.db, RunDatabase::ReadOnly(_)) {
469                return Err(EventStoreError::Closed);
470            }
471
472            if !state.manifest.is_sealed() {
473                return Err(EventStoreError::CrashedPredecessor);
474            }
475        }
476
477        let dir = self.run_dir(&manifest.instance_id);
478        fs::create_dir_all(&dir).map_err(|e| {
479            let msg = format!("create dir {}: {e}", dir.display());
480
481            if is_disk_pressure(e.kind()) {
482                EventStoreError::Disk(msg)
483            } else {
484                EventStoreError::Backend(msg)
485            }
486        })?;
487        let path = self.run_path(&manifest.instance_id, &manifest.run_id);
488        let path_existed = path.exists();
489
490        let db = Database::create(&path).map_err(map_database_err)?;
491
492        if path_existed {
493            format::verify_store_format(&db)?;
494            let on_disk = Self::read_manifest(&db)?.ok_or_else(|| {
495                EventStoreError::Corrupted(format!(
496                    "missing manifest in existing run file at {}",
497                    path.display()
498                ))
499            })?;
500
501            if !matches!(on_disk.status, RunStatus::Running) {
502                return Err(EventStoreError::Backend(format!(
503                    "run file at {} already sealed, status was {:?}",
504                    path.display(),
505                    on_disk.status
506                )));
507            }
508
509            let (high_watermark, max_ts_init) = Self::compute_progress(&db)?;
510            let mut recovered = on_disk;
511            recovered.high_watermark = high_watermark;
512            self.state = Some(RunState {
513                db: RunDatabase::ReadWrite(db),
514                manifest: recovered,
515                high_watermark,
516                max_ts_init,
517                file_path: path,
518            });
519            return Err(EventStoreError::CrashedPredecessor);
520        }
521
522        manifest.status = RunStatus::Running;
523        manifest.end_ts_init = None;
524        manifest.high_watermark = 0;
525        Self::initialize_fresh(&db, &manifest)?;
526
527        self.state = Some(RunState {
528            db: RunDatabase::ReadWrite(db),
529            manifest,
530            high_watermark: 0,
531            max_ts_init: UnixNanos::default(),
532            file_path: path,
533        });
534        Ok(())
535    }
536
537    fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
538        let state = self.state_mut()?;
539
540        if state.manifest.is_sealed() {
541            return Err(EventStoreError::Closed);
542        }
543
544        if entries.is_empty() {
545            return Ok(state.high_watermark);
546        }
547
548        for (expected, append) in (state.high_watermark + 1..).zip(entries.iter()) {
549            if append.entry.seq != expected {
550                // Atomically rejected: surface the durable high-watermark, not the within-batch
551                // validation cursor, so callers that resync from this value never skip entries
552                // that were never committed.
553                return Err(EventStoreError::OutOfOrder {
554                    high_watermark: state.high_watermark,
555                    seq: append.entry.seq,
556                });
557            }
558        }
559
560        let encoded: Vec<Vec<u8>> = entries
561            .iter()
562            .map(|append| {
563                codec::encode_to_vec(&append.entry).map_err(|e| {
564                    EventStoreError::Backend(format!("encode entry seq={}: {e}", append.entry.seq))
565                })
566            })
567            .collect::<Result<_, _>>()?;
568
569        let db = state.db.read_write()?;
570        let txn = begin_immediate_write(db)?;
571        {
572            let mut entries_table = txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
573            let mut client_table = txn.open_table(CLIENT_ORDER_INDEX).map_err(map_table_err)?;
574            let mut venue_table = txn.open_table(VENUE_ORDER_INDEX).map_err(map_table_err)?;
575
576            for (append, bytes) in entries.iter().zip(encoded.iter()) {
577                entries_table
578                    .insert(append.entry.seq, bytes.as_slice())
579                    .map_err(map_storage_err)?;
580
581                for IndexKey { kind, key } in &append.index_keys {
582                    let table = match kind {
583                        IndexKind::ClientOrderId => &mut client_table,
584                        IndexKind::VenueOrderId => &mut venue_table,
585                    };
586                    let already = table.get(key.as_str()).map_err(map_storage_err)?.is_some();
587
588                    if !already {
589                        table
590                            .insert(key.as_str(), append.entry.seq)
591                            .map_err(map_storage_err)?;
592                    }
593                }
594            }
595        }
596        txn.commit().map_err(map_commit_err)?;
597
598        let mut max_ts = state.max_ts_init;
599        let mut new_hwm = state.high_watermark;
600
601        for append in entries {
602            if append.entry.ts_init > max_ts {
603                max_ts = append.entry.ts_init;
604            }
605            new_hwm = append.entry.seq;
606        }
607        state.high_watermark = new_hwm;
608        state.max_ts_init = max_ts;
609        state.manifest.high_watermark = new_hwm;
610
611        Ok(new_hwm)
612    }
613
614    fn scan_range(
615        &self,
616        from: u64,
617        to: u64,
618        direction: ScanDirection,
619    ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
620        let state = self.state()?;
621
622        if from > to || from == 0 || state.high_watermark == 0 {
623            return Ok(Vec::new());
624        }
625
626        let lo = from;
627        let hi = to.min(state.high_watermark);
628
629        if lo > hi {
630            return Ok(Vec::new());
631        }
632
633        let txn = state.db.begin_read()?;
634        let table = txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
635
636        // hi is capped to high_watermark above, so every seq in [lo, hi] is supposed to be
637        // present. redb iterates only existing keys, so a missing row inside this range
638        // means a committed sequence has been lost (corruption, external tampering); we
639        // surface Gap rather than silently shortening the result.
640        let mut out = Vec::new();
641        let mut expected = lo;
642        let iter = table.range(lo..=hi).map_err(map_storage_err)?;
643
644        for row in iter {
645            let (k, v) = row.map_err(map_storage_err)?;
646            let seq = k.value();
647
648            if seq != expected {
649                return Err(EventStoreError::Gap {
650                    prev: expected.saturating_sub(1),
651                    next: seq,
652                    missing: expected,
653                });
654            }
655            let bytes = v.value();
656            let entry = codec::decode_from_slice::<EventStoreEntry>(bytes)
657                .map_err(|e| EventStoreError::Corrupted(format!("decode entry seq={seq}: {e}")))?;
658
659            check_embedded_seq(seq, &entry)?;
660
661            if entry.recompute_hash() != entry.entry_hash {
662                return Err(EventStoreError::HashMismatch { seq });
663            }
664            out.push(entry);
665            expected = seq + 1;
666        }
667
668        if expected <= hi {
669            return Err(EventStoreError::Gap {
670                prev: expected.saturating_sub(1),
671                next: hi + 1,
672                missing: expected,
673            });
674        }
675
676        if matches!(direction, ScanDirection::Reverse) {
677            out.reverse();
678        }
679        Ok(out)
680    }
681
682    fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
683        let state = self.state()?;
684
685        if seq == 0 || seq > state.high_watermark {
686            return Ok(None);
687        }
688
689        let txn = state.db.begin_read()?;
690        let table = txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
691        let Some(value) = table.get(seq).map_err(map_storage_err)? else {
692            // seq is inside the watermark per the guard above, so the row must exist;
693            // its absence is a committed-but-missing entry.
694            return Err(EventStoreError::Gap {
695                prev: seq.saturating_sub(1),
696                next: seq + 1,
697                missing: seq,
698            });
699        };
700
701        let bytes = value.value();
702        let entry = codec::decode_from_slice::<EventStoreEntry>(bytes)
703            .map_err(|e| EventStoreError::Corrupted(format!("decode entry seq={seq}: {e}")))?;
704
705        check_embedded_seq(seq, &entry)?;
706
707        if entry.recompute_hash() != entry.entry_hash {
708            return Err(EventStoreError::HashMismatch { seq });
709        }
710        Ok(Some(entry))
711    }
712
713    fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
714        let state = self.state()?;
715        let txn = state.db.begin_read()?;
716        let definition = match kind {
717            IndexKind::ClientOrderId => CLIENT_ORDER_INDEX,
718            IndexKind::VenueOrderId => VENUE_ORDER_INDEX,
719        };
720        let table = txn.open_table(definition).map_err(map_table_err)?;
721        let value = table.get(key).map_err(map_storage_err)?;
722        Ok(value.map(|v| v.value()))
723    }
724
725    fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
726        let state = self.state()?;
727        let txn = state.db.begin_read()?;
728        let definition = match kind {
729            IndexKind::ClientOrderId => CLIENT_ORDER_INDEX,
730            IndexKind::VenueOrderId => VENUE_ORDER_INDEX,
731        };
732        let table = txn.open_table(definition).map_err(map_table_err)?;
733        let iter = table.iter().map_err(map_storage_err)?;
734        let mut out = Vec::new();
735
736        for row in iter {
737            let (k, v) = row.map_err(map_storage_err)?;
738            out.push((k.value().to_string(), v.value()));
739        }
740        Ok(out)
741    }
742
743    fn record_snapshot_anchor(&mut self, anchor: SnapshotAnchor) -> Result<(), EventStoreError> {
744        let state = self.state_mut()?;
745
746        if state.manifest.is_sealed() {
747            return Err(EventStoreError::Closed);
748        }
749
750        let latest = Self::read_snapshot_anchor(state.db.readable())?;
751        validate_new_anchor(&anchor, state.high_watermark, latest.as_ref())?;
752
753        let bytes = codec::encode_to_vec(&anchor)
754            .map_err(|e| EventStoreError::Backend(format!("encode snapshot anchor: {e}")))?;
755        let db = state.db.read_write()?;
756        let txn = begin_immediate_write(db)?;
757        {
758            let mut table = txn
759                .open_table(SNAPSHOT_ANCHOR_TABLE)
760                .map_err(map_table_err)?;
761            table
762                .insert(SNAPSHOT_ANCHOR_KEY, bytes.as_slice())
763                .map_err(map_storage_err)?;
764        }
765        txn.commit().map_err(map_commit_err)?;
766        Ok(())
767    }
768
769    fn latest_snapshot_anchor(&self) -> Result<Option<SnapshotAnchor>, EventStoreError> {
770        Self::read_snapshot_anchor(self.state()?.db.readable())
771    }
772
773    fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
774        let state = self.state_mut()?;
775
776        // Running is not a terminal state; accepting it would leave `is_sealed()` returning
777        // false while the seal call returned Ok, so subsequent appends would not see Closed.
778        if matches!(status, RunStatus::Running) {
779            return Err(EventStoreError::Backend(
780                "seal status must be a terminal state, was Running".to_string(),
781            ));
782        }
783
784        if state.manifest.is_sealed() {
785            return Err(EventStoreError::Closed);
786        }
787
788        let mut updated = state.manifest.clone();
789        updated.status = status;
790        updated.high_watermark = state.high_watermark;
791
792        if state.high_watermark > 0 {
793            updated.end_ts_init = Some(state.max_ts_init);
794        }
795
796        Self::write_manifest(state.db.read_write()?, &updated)?;
797        state.manifest = updated;
798        Ok(())
799    }
800
801    fn manifest(&self) -> Result<RunManifest, EventStoreError> {
802        Ok(self.state()?.manifest.clone())
803    }
804
805    fn high_watermark(&self) -> Result<u64, EventStoreError> {
806        Ok(self.state()?.high_watermark)
807    }
808}
809
810fn missing_manifest(path: &Path) -> EventStoreError {
811    EventStoreError::Corrupted(format!(
812        "missing manifest in run file at {}",
813        path.display()
814    ))
815}
816
817// The entry hash covers the embedded seq, not the table key, so a moved or
818// duplicated row still hashes correctly; both read paths refuse it here.
819fn check_embedded_seq(seq: u64, entry: &EventStoreEntry) -> Result<(), EventStoreError> {
820    if entry.seq != seq {
821        return Err(EventStoreError::SeqMismatch {
822            table_key: seq,
823            embedded_seq: entry.seq,
824        });
825    }
826    Ok(())
827}
828
829fn begin_immediate_write(db: &Database) -> Result<WriteTransaction, EventStoreError> {
830    let mut txn = db.begin_write().map_err(map_transaction_err)?;
831    txn.set_durability(Durability::Immediate)
832        .map_err(|e| EventStoreError::Backend(format!("set durability: {e}")))?;
833    Ok(txn)
834}
835
836fn insert_run_manifest(
837    txn: &WriteTransaction,
838    manifest: &RunManifest,
839) -> Result<(), EventStoreError> {
840    let bytes = codec::encode_to_vec(manifest)
841        .map_err(|e| EventStoreError::Backend(format!("encode manifest: {e}")))?;
842    let mut table = txn.open_table(MANIFEST_TABLE).map_err(map_table_err)?;
843    table
844        .insert(MANIFEST_KEY, bytes.as_slice())
845        .map_err(map_storage_err)?;
846    Ok(())
847}
848
849fn map_storage_err(err: StorageError) -> EventStoreError {
850    match err {
851        StorageError::Io(io_err) if is_disk_pressure(io_err.kind()) => {
852            EventStoreError::Disk(io_err.to_string())
853        }
854        StorageError::Corrupted(msg) => EventStoreError::Corrupted(msg),
855        other => EventStoreError::Backend(other.to_string()),
856    }
857}
858
859// `EventStoreError::Disk` documents ENOSPC and `RLIMIT_FSIZE` as its targets. On the
860// stable toolchain `ENOSPC` surfaces as `StorageFull`, `RLIMIT_FSIZE`/`EFBIG` as
861// `FileTooLarge`, and `EDQUOT` as `QuotaExceeded`; the kernel halt path keys off
862// `Disk`, so all three must classify the same way.
863fn is_disk_pressure(kind: ErrorKind) -> bool {
864    matches!(
865        kind,
866        ErrorKind::FileTooLarge | ErrorKind::StorageFull | ErrorKind::QuotaExceeded
867    )
868}
869
870fn map_database_err(err: DatabaseError) -> EventStoreError {
871    match err {
872        DatabaseError::RepairAborted => EventStoreError::Corrupted(
873            "database requires repair and cannot be verified read-only".to_string(),
874        ),
875        DatabaseError::UpgradeRequired(version) => EventStoreError::Corrupted(format!(
876            "database file format version {version} requires manual upgrade",
877        )),
878        DatabaseError::Storage(storage) => map_storage_err(storage),
879        other => EventStoreError::Backend(other.to_string()),
880    }
881}
882
883fn map_read_only_database_err(err: DatabaseError) -> EventStoreError {
884    match err {
885        DatabaseError::Storage(StorageError::Io(io_err)) if is_corrupt_read(io_err.kind()) => {
886            EventStoreError::Corrupted(format!("read-only open failed: {io_err}"))
887        }
888        other => map_database_err(other),
889    }
890}
891
892fn is_corrupt_read(kind: ErrorKind) -> bool {
893    matches!(kind, ErrorKind::UnexpectedEof | ErrorKind::InvalidData)
894}
895
896fn map_table_err(err: TableError) -> EventStoreError {
897    // Mirror redb's own classification: schema-shape failures (missing table, type
898    // mismatch, definition drift) are structural corruption, not generic backend
899    // errors. Programmer-error variants (`TableAlreadyOpen`, `TableExists`) stay
900    // Backend so they surface as bugs rather than quarantine triggers.
901    match err {
902        TableError::Storage(storage) => map_storage_err(storage),
903        TableError::TableDoesNotExist(_)
904        | TableError::TableTypeMismatch { .. }
905        | TableError::TableIsMultimap(_)
906        | TableError::TableIsNotMultimap(_)
907        | TableError::TypeDefinitionChanged { .. } => EventStoreError::Corrupted(err.to_string()),
908        other => EventStoreError::Backend(other.to_string()),
909    }
910}
911
912fn map_commit_err(err: CommitError) -> EventStoreError {
913    match err {
914        CommitError::Storage(storage) => map_storage_err(storage),
915        other => EventStoreError::Backend(other.to_string()),
916    }
917}
918
919fn is_run_file(path: &Path) -> bool {
920    path.extension().and_then(|s| s.to_str()) == Some("redb")
921        && path
922            .file_name()
923            .and_then(|s| s.to_str())
924            .is_none_or(|name| !name.ends_with(".markers.redb"))
925}
926
927fn map_transaction_err(err: TransactionError) -> EventStoreError {
928    match err {
929        TransactionError::Storage(storage) => map_storage_err(storage),
930        other => EventStoreError::Backend(other.to_string()),
931    }
932}
933
934#[cfg(test)]
935mod tests {
936    use rstest::rstest;
937    use tempfile::TempDir;
938
939    use super::*;
940
941    fn raw_run_path(base: &Path, run_id: &str) -> PathBuf {
942        let dir = base.join("trader-001");
943        std::fs::create_dir_all(&dir).expect("mkdir");
944        dir.join(format!("{run_id}.redb"))
945    }
946
947    fn create_pre_codec_run_file(path: &Path) {
948        let entries: TableDefinition<u64, &[u8]> = TableDefinition::new("entries");
949        let manifest: TableDefinition<&str, &[u8]> = TableDefinition::new("manifest");
950        let db = Database::create(path).expect("create redb");
951        let txn = db.begin_write().expect("begin write");
952        {
953            txn.open_table(entries).expect("open entries");
954            let mut table = txn.open_table(manifest).expect("open manifest");
955            table
956                .insert("current", b"old-format".as_slice())
957                .expect("insert");
958        }
959        txn.commit().expect("commit");
960    }
961
962    #[rstest]
963    fn read_run_manifest_rejects_store_without_format_marker() {
964        let tmp = TempDir::new().expect("tempdir");
965        let path = raw_run_path(tmp.path(), "run-old-format");
966        create_pre_codec_run_file(&path);
967
968        let err = RedbBackend::read_run_manifest(&path).expect_err("must reject old format");
969
970        match err {
971            EventStoreError::Corrupted(msg) => {
972                assert!(msg.contains("regenerated"), "msg was: {msg}");
973            }
974            other => panic!("expected Corrupted, was {other:?}"),
975        }
976    }
977
978    #[rstest]
979    #[case::file_too_large(ErrorKind::FileTooLarge, true)]
980    #[case::storage_full(ErrorKind::StorageFull, true)]
981    #[case::quota_exceeded(ErrorKind::QuotaExceeded, true)]
982    #[case::other(ErrorKind::Other, false)]
983    #[case::not_found(ErrorKind::NotFound, false)]
984    #[case::permission_denied(ErrorKind::PermissionDenied, false)]
985    #[case::interrupted(ErrorKind::Interrupted, false)]
986    fn is_disk_pressure_matches_documented_kinds(#[case] kind: ErrorKind, #[case] expected: bool) {
987        assert_eq!(is_disk_pressure(kind), expected);
988    }
989
990    #[rstest]
991    fn map_storage_err_classifies_disk_pressure_as_disk() {
992        let io_err = std::io::Error::from(ErrorKind::StorageFull);
993        let mapped = map_storage_err(StorageError::Io(io_err));
994
995        match mapped {
996            EventStoreError::Disk(_) => {}
997            other => panic!("expected Disk, was {other:?}"),
998        }
999    }
1000
1001    #[rstest]
1002    fn map_storage_err_classifies_corrupted_as_corrupted() {
1003        let mapped = map_storage_err(StorageError::Corrupted("boom".to_string()));
1004
1005        match mapped {
1006            EventStoreError::Corrupted(msg) => assert!(msg.contains("boom")),
1007            other => panic!("expected Corrupted, was {other:?}"),
1008        }
1009    }
1010
1011    #[rstest]
1012    fn map_storage_err_falls_back_to_backend_for_unrelated_io() {
1013        let io_err = std::io::Error::from(ErrorKind::PermissionDenied);
1014        let mapped = map_storage_err(StorageError::Io(io_err));
1015
1016        match mapped {
1017            EventStoreError::Backend(_) => {}
1018            other => panic!("expected Backend, was {other:?}"),
1019        }
1020    }
1021}
nautilus_event_store/backend/redb.rs

nautilus_event_store/backend/
redb.rs
