nautilus_event_store/backend/mod.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//! Backend abstraction for the event store.
17//!
18//! The [`EventStore`] trait keeps backends interchangeable: in-memory under simulation,
19//! `redb` in production, and any future swap (custom WAL, segmented log) without touching
20//! consumers. Backend-specific types (`redb::Database`, `redb::Error`) never appear in this
21//! trait surface.
22
23pub mod memory;
24pub mod redb;
25
26pub use memory::MemoryBackend;
27pub use redb::RedbBackend;
28
29use crate::{
30 entry::EventStoreEntry,
31 error::EventStoreError,
32 manifest::{RunManifest, RunStatus},
33 snapshot::SnapshotAnchor,
34};
35
36/// The kind of secondary index the reader can look up.
37///
38/// Indices are rebuildable projections from the canonical `seq -> entry` table, not
39/// authoritative storage. The verifier rebuilds them and cross-checks against the stored
40/// rows.
41#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
42pub enum IndexKind {
43 /// `client_order_id -> seq`. Looks up the first entry that mentions a client order id.
44 ClientOrderId,
45 /// `venue_order_id -> seq`. Looks up the first entry that mentions a venue order id.
46 VenueOrderId,
47}
48
49/// Direction for a sequence-keyed scan.
50#[derive(Copy, Clone, Debug, PartialEq, Eq)]
51pub enum ScanDirection {
52 /// Forward scan: ascending `seq`.
53 Forward,
54 /// Reverse scan: descending `seq`.
55 Reverse,
56}
57
58/// A single sidecar index entry recorded atomically with an event store entry.
59///
60/// The writer (or its encoder) produces one [`IndexKey`] per `(IndexKind, key)` pair the
61/// entry should be locatable under. The backend records the first occurrence of each pair;
62/// subsequent occurrences for the same `(kind, key)` are no-ops, so [`EventStore::lookup`]
63/// always returns the earliest `seq` that mentioned the key.
64///
65/// Keys are stringified at the encoder boundary: `client_order_id` and `venue_order_id` are
66/// already strings on the wire types.
67#[derive(Clone, Debug, PartialEq, Eq, Hash)]
68pub struct IndexKey {
69 /// The secondary index this key belongs to.
70 pub kind: IndexKind,
71 /// The stringified key. Owned because the backend retains it.
72 pub key: String,
73}
74
75impl IndexKey {
76 /// Creates a new [`IndexKey`].
77 #[must_use]
78 pub const fn new(kind: IndexKind, key: String) -> Self {
79 Self { kind, key }
80 }
81}
82
83/// One entry plus its sidecar index keys, as accepted by [`EventStore::append_batch`].
84///
85/// Keeping the indices alongside the entry makes the commit atomic: the backend records
86/// the entry and its index keys in a single transaction, so a reader can never observe a
87/// committed `seq` whose secondary indices are missing.
88#[derive(Clone, Debug, PartialEq, Eq)]
89pub struct AppendEntry {
90 /// The captured entry. The writer has already assigned `seq`, `ts_publish`, and the
91 /// canonical `entry_hash` before constructing this value.
92 pub entry: EventStoreEntry,
93 /// Sidecar index keys produced for this entry. May be empty.
94 pub index_keys: Vec<IndexKey>,
95}
96
97impl AppendEntry {
98 /// Creates a new [`AppendEntry`].
99 #[must_use]
100 pub const fn new(entry: EventStoreEntry, index_keys: Vec<IndexKey>) -> Self {
101 Self { entry, index_keys }
102 }
103
104 /// Creates a new [`AppendEntry`] with no sidecar index keys.
105 #[must_use]
106 pub const fn without_indices(entry: EventStoreEntry) -> Self {
107 Self {
108 entry,
109 index_keys: Vec::new(),
110 }
111 }
112}
113
114/// The single-node embedded event store.
115///
116/// One backend instance owns one open run. Writes funnel through `append_batch`; reads are
117/// `seq`-keyed scans or single-key lookups across the secondary indices.
118///
119/// Backends are responsible for:
120///
121/// - Per-run organization on disk (one redb file per run, one in-memory log per run).
122/// - Durable commit semantics (`Durability::Immediate` for redb).
123/// - High-watermark advance after commit acknowledgement.
124/// - Mapping backend-specific errors onto [`EventStoreError`].
125///
126/// This trait does not own batching or thread-management policy: the writer
127/// (`crates/event_store/src/writer/`) is the dedicated thread that batches entries and
128/// invokes `append_batch` against the backend.
129pub trait EventStore: Send {
130 /// Opens an existing run or creates a new one with the supplied manifest.
131 ///
132 /// On reopening a run whose status is [`RunStatus::Running`] without a `RunEnded`
133 /// entry, the backend returns [`EventStoreError::CrashedPredecessor`] so the kernel
134 /// can seal it before opening a new run.
135 ///
136 /// # Errors
137 ///
138 /// Returns an error if the backend fails, the header region is damaged, or disk pressure
139 /// prevents creation.
140 fn open_run(&mut self, manifest: RunManifest) -> Result<(), EventStoreError>;
141
142 /// Appends a batch of `(entry, index_keys)` pairs in a single backend transaction.
143 ///
144 /// The writer assigns `seq`, `ts_publish`, and the canonical `entry_hash`, plus any
145 /// sidecar [`IndexKey`]s the encoder produced, before constructing each
146 /// [`AppendEntry`]. The backend rejects batches whose first `seq` is not exactly
147 /// `high_watermark + 1`, and whose subsequent seqs are not contiguous (each successor
148 /// is `prev + 1`).
149 ///
150 /// On successful commit, the backend advances its high-watermark and returns the new
151 /// value.
152 ///
153 /// # Errors
154 ///
155 /// Returns an error if the run is sealed, if the first seq is not exactly `high_watermark + 1`
156 /// or a within-batch seq is not contiguous, if the backing storage refuses the write, or if the
157 /// backend fails.
158 fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError>;
159
160 /// Scans entries by `seq` over the inclusive range `[from, to]`.
161 ///
162 /// Backends may stream rows lazily; the trait's vector return is a simple
163 /// implementation default for the in-memory backend and the verifier hot path. The
164 /// reader (`crates/event_store/src/reader/`) wraps this with iterator semantics.
165 ///
166 /// # Errors
167 ///
168 /// Returns an error if a row's recomputed hash diverges from the stored value, if the scan
169 /// observes a missing `seq`, or if the backend fails.
170 fn scan_range(
171 &self,
172 from: u64,
173 to: u64,
174 direction: ScanDirection,
175 ) -> Result<Vec<EventStoreEntry>, EventStoreError>;
176
177 /// Reads a single entry by `seq`.
178 ///
179 /// # Errors
180 ///
181 /// Returns an error if [`EventStore::scan_range`] fails.
182 fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError>;
183
184 /// Looks up the first `seq` recorded under the given index key.
185 ///
186 /// Returns `None` when the key has not been observed.
187 ///
188 /// # Errors
189 ///
190 /// Returns an error if the backend fails.
191 fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError>;
192
193 /// Enumerates every `(key, seq)` pair stored under the given secondary index.
194 ///
195 /// Used by the verifier to cross-check the stored sidecar indices against the
196 /// projection rebuilt from the entry table. The returned vector's order is
197 /// backend-defined; callers that need a stable comparison sort it themselves.
198 ///
199 /// # Errors
200 ///
201 /// Returns an error if the backend fails.
202 fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError>;
203
204 /// Records the latest cache snapshot anchor for the open run.
205 ///
206 /// The anchor's high-watermark must be less than or equal to the durable
207 /// high-watermark and must not move backward relative to the latest recorded
208 /// anchor. Backends persist the anchor independently from the snapshot blob; the
209 /// cache owns the blob and content-hash calculation.
210 ///
211 /// # Errors
212 ///
213 /// Returns an error if no run is open, when the backend does not support snapshot anchors, or
214 /// when the anchor is invalid for the current high-watermark, or if the run is sealed.
215 fn record_snapshot_anchor(&mut self, _anchor: SnapshotAnchor) -> Result<(), EventStoreError> {
216 Err(EventStoreError::Backend(
217 "snapshot anchors are not supported by this backend".to_string(),
218 ))
219 }
220
221 /// Returns the latest recorded snapshot anchor for the open run.
222 ///
223 /// Returns `Ok(None)` when no snapshot anchor has been recorded yet. The default
224 /// implementation also returns `Ok(None)`: a backend without anchor support
225 /// truthfully has no anchor to report, and consumers (the verifier, retention
226 /// planning) must be able to distinguish "no anchor" from a real read failure.
227 ///
228 /// # Errors
229 ///
230 /// Returns an error if no run is open or a stored anchor fails to decode.
231 fn latest_snapshot_anchor(&self) -> Result<Option<SnapshotAnchor>, EventStoreError> {
232 Ok(None)
233 }
234
235 /// Seals the open run with the given final status and persists the manifest update.
236 ///
237 /// Subsequent calls to [`EventStore::append_batch`] return [`EventStoreError::Closed`].
238 ///
239 /// # Errors
240 ///
241 /// Returns an error if the backend fails or disk pressure prevents the seal commit.
242 fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError>;
243
244 /// Returns the current manifest snapshot.
245 ///
246 /// # Errors
247 ///
248 /// Returns an error if no run is open.
249 fn manifest(&self) -> Result<RunManifest, EventStoreError>;
250
251 /// Returns the largest `seq` durably acknowledged for the open run.
252 ///
253 /// Returns `0` when no entries have been committed yet (the writer assigns `seq`
254 /// starting at `1`).
255 ///
256 /// # Errors
257 ///
258 /// Returns an error if no run is open.
259 fn high_watermark(&self) -> Result<u64, EventStoreError>;
260}