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

nautilus_event_store/backend/
redb.rs
