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        manifests.sort_by_key(|m| m.start_ts_init);
300        Ok(manifests)
301    }
302
303    fn read_run_manifest(path: &Path) -> Result<RunManifest, EventStoreError> {
304        let manifest = match ReadOnlyDatabase::open(path) {
305            Ok(db) => {
306                Self::verify_listed_store_format(&db, path)?;
307                Self::read_manifest(&db)?
308            }
309            // Each durable commit deletes redb's allocator-state table and only a clean
310            // `Database::drop` rewrites it, so a hard-killed process leaves a file the
311            // read-only open refuses. A writable open repairs it for future opens.
312            Err(DatabaseError::RepairAborted) => {
313                log::warn!(
314                    "Run file {} was not shut down cleanly, repairing",
315                    path.display()
316                );
317                let db = Database::open(path).map_err(map_database_err)?;
318                Self::verify_listed_store_format(&db, path)?;
319                Self::read_manifest(&db)?
320            }
321            Err(e) => return Err(map_read_only_database_err(e)),
322        };
323        manifest.ok_or_else(|| missing_manifest(path))
324    }
325
326    fn verify_listed_store_format<D: ReadableDatabase + ?Sized>(
327        db: &D,
328        path: &Path,
329    ) -> Result<(), EventStoreError> {
330        match format::verify_store_format(db) {
331            Ok(()) => Ok(()),
332            Err(e) if format::is_missing_store_format(&e) && !Self::manifest_row_exists(db)? => {
333                Err(missing_manifest(path))
334            }
335            Err(e) => Err(e),
336        }
337    }
338
339    fn manifest_row_exists<D: ReadableDatabase + ?Sized>(db: &D) -> Result<bool, EventStoreError> {
340        let txn = db.begin_read().map_err(map_transaction_err)?;
341        let table = match txn.open_table(MANIFEST_TABLE) {
342            Ok(table) => table,
343            Err(TableError::TableDoesNotExist(_)) => return Ok(false),
344            Err(e) => return Err(map_table_err(e)),
345        };
346
347        table
348            .get(MANIFEST_KEY)
349            .map(|value| value.is_some())
350            .map_err(map_storage_err)
351    }
352
353    fn state(&self) -> Result<&RunState, EventStoreError> {
354        self.state
355            .as_ref()
356            .ok_or_else(|| EventStoreError::Backend("no run open".to_string()))
357    }
358
359    fn state_mut(&mut self) -> Result<&mut RunState, EventStoreError> {
360        self.state
361            .as_mut()
362            .ok_or_else(|| EventStoreError::Backend("no run open".to_string()))
363    }
364
365    fn initialize_fresh(db: &Database, manifest: &RunManifest) -> Result<(), EventStoreError> {
366        let txn = begin_immediate_write(db)?;
367        {
368            txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
369            txn.open_table(CLIENT_ORDER_INDEX).map_err(map_table_err)?;
370            txn.open_table(VENUE_ORDER_INDEX).map_err(map_table_err)?;
371            txn.open_table(SNAPSHOT_ANCHOR_TABLE)
372                .map_err(map_table_err)?;
373        }
374        format::write_store_format(&txn)?;
375        insert_run_manifest(&txn, manifest)?;
376        txn.commit().map_err(map_commit_err)?;
377        Ok(())
378    }
379
380    fn write_manifest(db: &Database, manifest: &RunManifest) -> Result<(), EventStoreError> {
381        let txn = begin_immediate_write(db)?;
382        insert_run_manifest(&txn, manifest)?;
383        txn.commit().map_err(map_commit_err)?;
384        Ok(())
385    }
386
387    fn read_manifest<D: ReadableDatabase + ?Sized>(
388        db: &D,
389    ) -> Result<Option<RunManifest>, EventStoreError> {
390        let txn = db.begin_read().map_err(map_transaction_err)?;
391        let table = txn.open_table(MANIFEST_TABLE).map_err(map_table_err)?;
392        let Some(value) = table.get(MANIFEST_KEY).map_err(map_storage_err)? else {
393            return Ok(None);
394        };
395        let bytes = value.value();
396        let manifest = codec::decode_from_slice::<RunManifest>(bytes)
397            .map_err(|e| EventStoreError::Corrupted(format!("decode manifest: {e}")))?;
398        Ok(Some(manifest))
399    }
400
401    fn read_snapshot_anchor<D: ReadableDatabase + ?Sized>(
402        db: &D,
403    ) -> Result<Option<SnapshotAnchor>, EventStoreError> {
404        let txn = db.begin_read().map_err(map_transaction_err)?;
405        let table = match txn.open_table(SNAPSHOT_ANCHOR_TABLE) {
406            Ok(table) => table,
407            Err(TableError::TableDoesNotExist(_)) => return Ok(None),
408            Err(e) => return Err(map_table_err(e)),
409        };
410        let Some(value) = table.get(SNAPSHOT_ANCHOR_KEY).map_err(map_storage_err)? else {
411            return Ok(None);
412        };
413        let bytes = value.value();
414        let anchor = codec::decode_from_slice::<SnapshotAnchor>(bytes)
415            .map_err(|e| EventStoreError::Corrupted(format!("decode snapshot anchor: {e}")))?;
416        Ok(Some(anchor))
417    }
418
419    fn compute_progress<D: ReadableDatabase + ?Sized>(
420        db: &D,
421    ) -> Result<(u64, UnixNanos), EventStoreError> {
422        let txn = db.begin_read().map_err(map_transaction_err)?;
423        let table = txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
424
425        let Some((last_key, _)) = table.last().map_err(map_storage_err)? else {
426            return Ok((0, UnixNanos::default()));
427        };
428        let high_watermark = last_key.value();
429
430        // Walk the entry table once to recover the maximum `ts_init`. Memory.rs tracks this
431        // across appends; on crash recovery we have nothing to fall back on, so we recompute
432        // it from the durable rows. An undecodable row must not make the run unopenable:
433        // max ts_init is best-effort, and the corruption itself surfaces on the scan paths,
434        // where the recovery sweep quarantines the run.
435        let mut max_ts = UnixNanos::default();
436        let iter = table.iter().map_err(map_storage_err)?;
437
438        for row in iter {
439            let (key, value) = row.map_err(map_storage_err)?;
440            let bytes = value.value();
441
442            match codec::decode_from_slice::<EventStoreEntry>(bytes) {
443                Ok(entry) => {
444                    if entry.ts_init > max_ts {
445                        max_ts = entry.ts_init;
446                    }
447                }
448                Err(e) => {
449                    log::error!("Undecodable entry at seq {} on load: {e}", key.value());
450                }
451            }
452        }
453
454        Ok((high_watermark, max_ts))
455    }
456}
457
458impl EventStore for RedbBackend {
459    fn open_run(&mut self, mut manifest: RunManifest) -> Result<(), EventStoreError> {
460        if let Some(state) = &self.state {
461            if matches!(state.db, RunDatabase::ReadOnly(_)) {
462                return Err(EventStoreError::Closed);
463            }
464
465            if !state.manifest.is_sealed() {
466                return Err(EventStoreError::CrashedPredecessor);
467            }
468        }
469
470        let dir = self.run_dir(&manifest.instance_id);
471        fs::create_dir_all(&dir).map_err(|e| {
472            let msg = format!("create dir {}: {e}", dir.display());
473
474            if is_disk_pressure(e.kind()) {
475                EventStoreError::Disk(msg)
476            } else {
477                EventStoreError::Backend(msg)
478            }
479        })?;
480        let path = self.run_path(&manifest.instance_id, &manifest.run_id);
481        let path_existed = path.exists();
482
483        let db = Database::create(&path).map_err(map_database_err)?;
484
485        if path_existed {
486            format::verify_store_format(&db)?;
487            let on_disk = Self::read_manifest(&db)?.ok_or_else(|| {
488                EventStoreError::Corrupted(format!(
489                    "missing manifest in existing run file at {}",
490                    path.display()
491                ))
492            })?;
493
494            if !matches!(on_disk.status, RunStatus::Running) {
495                return Err(EventStoreError::Backend(format!(
496                    "run file at {} already sealed, status was {:?}",
497                    path.display(),
498                    on_disk.status
499                )));
500            }
501
502            let (high_watermark, max_ts_init) = Self::compute_progress(&db)?;
503            let mut recovered = on_disk;
504            recovered.high_watermark = high_watermark;
505            self.state = Some(RunState {
506                db: RunDatabase::ReadWrite(db),
507                manifest: recovered,
508                high_watermark,
509                max_ts_init,
510                file_path: path,
511            });
512            return Err(EventStoreError::CrashedPredecessor);
513        }
514
515        manifest.status = RunStatus::Running;
516        manifest.end_ts_init = None;
517        manifest.high_watermark = 0;
518        Self::initialize_fresh(&db, &manifest)?;
519
520        self.state = Some(RunState {
521            db: RunDatabase::ReadWrite(db),
522            manifest,
523            high_watermark: 0,
524            max_ts_init: UnixNanos::default(),
525            file_path: path,
526        });
527        Ok(())
528    }
529
530    fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
531        let state = self.state_mut()?;
532
533        if state.manifest.is_sealed() {
534            return Err(EventStoreError::Closed);
535        }
536
537        if entries.is_empty() {
538            return Ok(state.high_watermark);
539        }
540
541        for (expected, append) in (state.high_watermark + 1..).zip(entries.iter()) {
542            if append.entry.seq != expected {
543                // Atomically rejected: surface the durable high-watermark, not the within-batch
544                // validation cursor, so callers that resync from this value never skip entries
545                // that were never committed.
546                return Err(EventStoreError::OutOfOrder {
547                    high_watermark: state.high_watermark,
548                    seq: append.entry.seq,
549                });
550            }
551        }
552
553        let encoded: Vec<Vec<u8>> = entries
554            .iter()
555            .map(|append| {
556                codec::encode_to_vec(&append.entry).map_err(|e| {
557                    EventStoreError::Backend(format!("encode entry seq={}: {e}", append.entry.seq))
558                })
559            })
560            .collect::<Result<_, _>>()?;
561
562        let db = state.db.read_write()?;
563        let txn = begin_immediate_write(db)?;
564        {
565            let mut entries_table = txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
566            let mut client_table = txn.open_table(CLIENT_ORDER_INDEX).map_err(map_table_err)?;
567            let mut venue_table = txn.open_table(VENUE_ORDER_INDEX).map_err(map_table_err)?;
568
569            for (append, bytes) in entries.iter().zip(encoded.iter()) {
570                entries_table
571                    .insert(append.entry.seq, bytes.as_slice())
572                    .map_err(map_storage_err)?;
573
574                for IndexKey { kind, key } in &append.index_keys {
575                    let table = match kind {
576                        IndexKind::ClientOrderId => &mut client_table,
577                        IndexKind::VenueOrderId => &mut venue_table,
578                    };
579                    let already = table.get(key.as_str()).map_err(map_storage_err)?.is_some();
580
581                    if !already {
582                        table
583                            .insert(key.as_str(), append.entry.seq)
584                            .map_err(map_storage_err)?;
585                    }
586                }
587            }
588        }
589        txn.commit().map_err(map_commit_err)?;
590
591        let mut max_ts = state.max_ts_init;
592        let mut new_hwm = state.high_watermark;
593
594        for append in entries {
595            if append.entry.ts_init > max_ts {
596                max_ts = append.entry.ts_init;
597            }
598            new_hwm = append.entry.seq;
599        }
600        state.high_watermark = new_hwm;
601        state.max_ts_init = max_ts;
602        state.manifest.high_watermark = new_hwm;
603
604        Ok(new_hwm)
605    }
606
607    fn scan_range(
608        &self,
609        from: u64,
610        to: u64,
611        direction: ScanDirection,
612    ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
613        let state = self.state()?;
614
615        if from > to || from == 0 || state.high_watermark == 0 {
616            return Ok(Vec::new());
617        }
618
619        let lo = from;
620        let hi = to.min(state.high_watermark);
621
622        if lo > hi {
623            return Ok(Vec::new());
624        }
625
626        let txn = state.db.begin_read()?;
627        let table = txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
628
629        // hi is capped to high_watermark above, so every seq in [lo, hi] is supposed to be
630        // present. redb iterates only existing keys, so a missing row inside this range
631        // means a committed sequence has been lost (corruption, external tampering); we
632        // surface Gap rather than silently shortening the result.
633        let mut out = Vec::new();
634        let mut expected = lo;
635        let iter = table.range(lo..=hi).map_err(map_storage_err)?;
636
637        for row in iter {
638            let (k, v) = row.map_err(map_storage_err)?;
639            let seq = k.value();
640
641            if seq != expected {
642                return Err(EventStoreError::Gap {
643                    prev: expected.saturating_sub(1),
644                    next: seq,
645                    missing: expected,
646                });
647            }
648            let bytes = v.value();
649            let entry = codec::decode_from_slice::<EventStoreEntry>(bytes)
650                .map_err(|e| EventStoreError::Corrupted(format!("decode entry seq={seq}: {e}")))?;
651
652            if entry.recompute_hash() != entry.entry_hash {
653                return Err(EventStoreError::HashMismatch { seq });
654            }
655            out.push(entry);
656            expected = seq + 1;
657        }
658
659        if expected <= hi {
660            return Err(EventStoreError::Gap {
661                prev: expected.saturating_sub(1),
662                next: hi + 1,
663                missing: expected,
664            });
665        }
666
667        if matches!(direction, ScanDirection::Reverse) {
668            out.reverse();
669        }
670        Ok(out)
671    }
672
673    fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
674        let state = self.state()?;
675
676        if seq == 0 || seq > state.high_watermark {
677            return Ok(None);
678        }
679
680        let txn = state.db.begin_read()?;
681        let table = txn.open_table(ENTRIES_TABLE).map_err(map_table_err)?;
682        let Some(value) = table.get(seq).map_err(map_storage_err)? else {
683            // seq is inside the watermark per the guard above, so the row must exist;
684            // its absence is a committed-but-missing entry.
685            return Err(EventStoreError::Gap {
686                prev: seq.saturating_sub(1),
687                next: seq + 1,
688                missing: seq,
689            });
690        };
691
692        let bytes = value.value();
693        let entry = codec::decode_from_slice::<EventStoreEntry>(bytes)
694            .map_err(|e| EventStoreError::Corrupted(format!("decode entry seq={seq}: {e}")))?;
695
696        if entry.recompute_hash() != entry.entry_hash {
697            return Err(EventStoreError::HashMismatch { seq });
698        }
699        Ok(Some(entry))
700    }
701
702    fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
703        let state = self.state()?;
704        let txn = state.db.begin_read()?;
705        let definition = match kind {
706            IndexKind::ClientOrderId => CLIENT_ORDER_INDEX,
707            IndexKind::VenueOrderId => VENUE_ORDER_INDEX,
708        };
709        let table = txn.open_table(definition).map_err(map_table_err)?;
710        let value = table.get(key).map_err(map_storage_err)?;
711        Ok(value.map(|v| v.value()))
712    }
713
714    fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
715        let state = self.state()?;
716        let txn = state.db.begin_read()?;
717        let definition = match kind {
718            IndexKind::ClientOrderId => CLIENT_ORDER_INDEX,
719            IndexKind::VenueOrderId => VENUE_ORDER_INDEX,
720        };
721        let table = txn.open_table(definition).map_err(map_table_err)?;
722        let iter = table.iter().map_err(map_storage_err)?;
723        let mut out = Vec::new();
724
725        for row in iter {
726            let (k, v) = row.map_err(map_storage_err)?;
727            out.push((k.value().to_string(), v.value()));
728        }
729        Ok(out)
730    }
731
732    fn record_snapshot_anchor(&mut self, anchor: SnapshotAnchor) -> Result<(), EventStoreError> {
733        let state = self.state_mut()?;
734
735        if state.manifest.is_sealed() {
736            return Err(EventStoreError::Closed);
737        }
738
739        let latest = Self::read_snapshot_anchor(state.db.readable())?;
740        validate_new_anchor(&anchor, state.high_watermark, latest.as_ref())?;
741
742        let bytes = codec::encode_to_vec(&anchor)
743            .map_err(|e| EventStoreError::Backend(format!("encode snapshot anchor: {e}")))?;
744        let db = state.db.read_write()?;
745        let txn = begin_immediate_write(db)?;
746        {
747            let mut table = txn
748                .open_table(SNAPSHOT_ANCHOR_TABLE)
749                .map_err(map_table_err)?;
750            table
751                .insert(SNAPSHOT_ANCHOR_KEY, bytes.as_slice())
752                .map_err(map_storage_err)?;
753        }
754        txn.commit().map_err(map_commit_err)?;
755        Ok(())
756    }
757
758    fn latest_snapshot_anchor(&self) -> Result<Option<SnapshotAnchor>, EventStoreError> {
759        Self::read_snapshot_anchor(self.state()?.db.readable())
760    }
761
762    fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
763        let state = self.state_mut()?;
764
765        // Running is not a terminal state; accepting it would leave `is_sealed()` returning
766        // false while the seal call returned Ok, so subsequent appends would not see Closed.
767        if matches!(status, RunStatus::Running) {
768            return Err(EventStoreError::Backend(
769                "seal status must be a terminal state, was Running".to_string(),
770            ));
771        }
772
773        if state.manifest.is_sealed() {
774            return Err(EventStoreError::Closed);
775        }
776
777        let mut updated = state.manifest.clone();
778        updated.status = status;
779        updated.high_watermark = state.high_watermark;
780
781        if state.high_watermark > 0 {
782            updated.end_ts_init = Some(state.max_ts_init);
783        }
784
785        Self::write_manifest(state.db.read_write()?, &updated)?;
786        state.manifest = updated;
787        Ok(())
788    }
789
790    fn manifest(&self) -> Result<RunManifest, EventStoreError> {
791        Ok(self.state()?.manifest.clone())
792    }
793
794    fn high_watermark(&self) -> Result<u64, EventStoreError> {
795        Ok(self.state()?.high_watermark)
796    }
797}
798
799fn missing_manifest(path: &Path) -> EventStoreError {
800    EventStoreError::Corrupted(format!(
801        "missing manifest in run file at {}",
802        path.display()
803    ))
804}
805
806fn begin_immediate_write(db: &Database) -> Result<WriteTransaction, EventStoreError> {
807    let mut txn = db.begin_write().map_err(map_transaction_err)?;
808    txn.set_durability(Durability::Immediate)
809        .map_err(|e| EventStoreError::Backend(format!("set durability: {e}")))?;
810    Ok(txn)
811}
812
813fn insert_run_manifest(
814    txn: &WriteTransaction,
815    manifest: &RunManifest,
816) -> Result<(), EventStoreError> {
817    let bytes = codec::encode_to_vec(manifest)
818        .map_err(|e| EventStoreError::Backend(format!("encode manifest: {e}")))?;
819    let mut table = txn.open_table(MANIFEST_TABLE).map_err(map_table_err)?;
820    table
821        .insert(MANIFEST_KEY, bytes.as_slice())
822        .map_err(map_storage_err)?;
823    Ok(())
824}
825
826fn map_storage_err(err: StorageError) -> EventStoreError {
827    match err {
828        StorageError::Io(io_err) if is_disk_pressure(io_err.kind()) => {
829            EventStoreError::Disk(io_err.to_string())
830        }
831        StorageError::Corrupted(msg) => EventStoreError::Corrupted(msg),
832        other => EventStoreError::Backend(other.to_string()),
833    }
834}
835
836// `EventStoreError::Disk` documents ENOSPC and `RLIMIT_FSIZE` as its targets. On the
837// stable toolchain `ENOSPC` surfaces as `StorageFull`, `RLIMIT_FSIZE`/`EFBIG` as
838// `FileTooLarge`, and `EDQUOT` as `QuotaExceeded`; the kernel halt path keys off
839// `Disk`, so all three must classify the same way.
840fn is_disk_pressure(kind: ErrorKind) -> bool {
841    matches!(
842        kind,
843        ErrorKind::FileTooLarge | ErrorKind::StorageFull | ErrorKind::QuotaExceeded
844    )
845}
846
847fn map_database_err(err: DatabaseError) -> EventStoreError {
848    match err {
849        DatabaseError::RepairAborted => EventStoreError::Corrupted(
850            "database requires repair and cannot be verified read-only".to_string(),
851        ),
852        DatabaseError::UpgradeRequired(version) => EventStoreError::Corrupted(format!(
853            "database file format version {version} requires manual upgrade",
854        )),
855        DatabaseError::Storage(storage) => map_storage_err(storage),
856        other => EventStoreError::Backend(other.to_string()),
857    }
858}
859
860fn map_read_only_database_err(err: DatabaseError) -> EventStoreError {
861    match err {
862        DatabaseError::Storage(StorageError::Io(io_err)) if is_corrupt_read(io_err.kind()) => {
863            EventStoreError::Corrupted(format!("read-only open failed: {io_err}"))
864        }
865        other => map_database_err(other),
866    }
867}
868
869fn is_corrupt_read(kind: ErrorKind) -> bool {
870    matches!(kind, ErrorKind::UnexpectedEof | ErrorKind::InvalidData)
871}
872
873fn map_table_err(err: TableError) -> EventStoreError {
874    // Mirror redb's own classification: schema-shape failures (missing table, type
875    // mismatch, definition drift) are structural corruption, not generic backend
876    // errors. Programmer-error variants (`TableAlreadyOpen`, `TableExists`) stay
877    // Backend so they surface as bugs rather than quarantine triggers.
878    match err {
879        TableError::Storage(storage) => map_storage_err(storage),
880        TableError::TableDoesNotExist(_)
881        | TableError::TableTypeMismatch { .. }
882        | TableError::TableIsMultimap(_)
883        | TableError::TableIsNotMultimap(_)
884        | TableError::TypeDefinitionChanged { .. } => EventStoreError::Corrupted(err.to_string()),
885        other => EventStoreError::Backend(other.to_string()),
886    }
887}
888
889fn map_commit_err(err: CommitError) -> EventStoreError {
890    match err {
891        CommitError::Storage(storage) => map_storage_err(storage),
892        other => EventStoreError::Backend(other.to_string()),
893    }
894}
895
896fn is_run_file(path: &Path) -> bool {
897    path.extension().and_then(|s| s.to_str()) == Some("redb")
898        && path
899            .file_name()
900            .and_then(|s| s.to_str())
901            .is_none_or(|name| !name.ends_with(".markers.redb"))
902}
903
904fn map_transaction_err(err: TransactionError) -> EventStoreError {
905    match err {
906        TransactionError::Storage(storage) => map_storage_err(storage),
907        other => EventStoreError::Backend(other.to_string()),
908    }
909}
910
911#[cfg(test)]
912mod tests {
913    use rstest::rstest;
914    use tempfile::TempDir;
915
916    use super::*;
917
918    fn raw_run_path(base: &Path, run_id: &str) -> PathBuf {
919        let dir = base.join("trader-001");
920        std::fs::create_dir_all(&dir).expect("mkdir");
921        dir.join(format!("{run_id}.redb"))
922    }
923
924    fn create_pre_codec_run_file(path: &Path) {
925        let entries: TableDefinition<u64, &[u8]> = TableDefinition::new("entries");
926        let manifest: TableDefinition<&str, &[u8]> = TableDefinition::new("manifest");
927        let db = Database::create(path).expect("create redb");
928        let txn = db.begin_write().expect("begin write");
929        {
930            txn.open_table(entries).expect("open entries");
931            let mut table = txn.open_table(manifest).expect("open manifest");
932            table
933                .insert("current", b"old-format".as_slice())
934                .expect("insert");
935        }
936        txn.commit().expect("commit");
937    }
938
939    #[rstest]
940    fn read_run_manifest_rejects_store_without_format_marker() {
941        let tmp = TempDir::new().expect("tempdir");
942        let path = raw_run_path(tmp.path(), "run-old-format");
943        create_pre_codec_run_file(&path);
944
945        let err = RedbBackend::read_run_manifest(&path).expect_err("must reject old format");
946
947        match err {
948            EventStoreError::Corrupted(msg) => {
949                assert!(msg.contains("regenerated"), "msg was: {msg}");
950            }
951            other => panic!("expected Corrupted, was {other:?}"),
952        }
953    }
954
955    #[rstest]
956    #[case::file_too_large(ErrorKind::FileTooLarge, true)]
957    #[case::storage_full(ErrorKind::StorageFull, true)]
958    #[case::quota_exceeded(ErrorKind::QuotaExceeded, true)]
959    #[case::other(ErrorKind::Other, false)]
960    #[case::not_found(ErrorKind::NotFound, false)]
961    #[case::permission_denied(ErrorKind::PermissionDenied, false)]
962    #[case::interrupted(ErrorKind::Interrupted, false)]
963    fn is_disk_pressure_matches_documented_kinds(#[case] kind: ErrorKind, #[case] expected: bool) {
964        assert_eq!(is_disk_pressure(kind), expected);
965    }
966
967    #[rstest]
968    fn map_storage_err_classifies_disk_pressure_as_disk() {
969        let io_err = std::io::Error::from(ErrorKind::StorageFull);
970        let mapped = map_storage_err(StorageError::Io(io_err));
971
972        match mapped {
973            EventStoreError::Disk(_) => {}
974            other => panic!("expected Disk, was {other:?}"),
975        }
976    }
977
978    #[rstest]
979    fn map_storage_err_classifies_corrupted_as_corrupted() {
980        let mapped = map_storage_err(StorageError::Corrupted("boom".to_string()));
981
982        match mapped {
983            EventStoreError::Corrupted(msg) => assert!(msg.contains("boom")),
984            other => panic!("expected Corrupted, was {other:?}"),
985        }
986    }
987
988    #[rstest]
989    fn map_storage_err_falls_back_to_backend_for_unrelated_io() {
990        let io_err = std::io::Error::from(ErrorKind::PermissionDenied);
991        let mapped = map_storage_err(StorageError::Io(io_err));
992
993        match mapped {
994            EventStoreError::Backend(_) => {}
995            other => panic!("expected Backend, was {other:?}"),
996        }
997    }
998}
nautilus_event_store/backend/redb.rs

nautilus_event_store/backend/
redb.rs
