rushdb_storage/backend.rs
1//! Narrow storage-substrate seam (research D6, plan W3).
2//!
3//! `StorageBackend` captures the durability/write surface the [`Store`] needs
4//! from its physical substrate. The synchronous redb implementation
5//! (`RedbBackend`) satisfies it today; an async backend (`io_uring` / direct IO)
6//! or a future `RushDB` `PageStore` can replace it without touching LMPG, query,
7//! search, or the replication/transaction layers, because those layers depend
8//! only on this trait and on the concrete read-view type.
9//!
10//! Greenfield note: the read view (`RedbReadView`) remains concrete in V1 — it
11//! is the synchronous backend's owned snapshot. An async backend brings its own
12//! view type behind this same seam; the point of the trait is that the write /
13//! durability contract is already isolated.
14
15use crate::read_view::{ReadLimits, ScanProjection};
16use crate::resources::ByteReservation;
17use crate::{Op, StorageError, StorageStatistics, VersionEntry};
18use std::collections::BTreeMap;
19use std::ops::Bound;
20use std::path::Path;
21use std::sync::Arc;
22
23/// Object-safe read snapshot of one physical backend.
24///
25/// Both the redb view (`RedbReadView`) and the in-memory view
26/// (`MemoryReadView`) implement it, so [`crate::ReadView`] has one read path
27/// regardless of substrate. Multi-shard stores merge the per-shard
28/// views in `MergedViews`.
29pub(crate) trait StorageReadView: Send + Sync {
30 /// Point read at `ci`; `current` selects the latest-commit fast path.
31 fn read_value(
32 &self,
33 key: &[u8],
34 ci: u64,
35 current: bool,
36 limits: ReadLimits,
37 projection: ScanProjection,
38 ) -> Result<Option<Vec<u8>>, StorageError>;
39
40 /// Smallest key in `(lower, upper)`; `include_history` widens to retained
41 /// versions for a non-current snapshot.
42 fn next_key(
43 &self,
44 lower: &Bound<Vec<u8>>,
45 upper: &Bound<Vec<u8>>,
46 limits: ReadLimits,
47 include_history: bool,
48 ) -> Result<Option<Vec<u8>>, StorageError>;
49
50 /// Ordered current-value range with the requested projection.
51 fn current_range<'a>(
52 &'a self,
53 start: &[u8],
54 end: &[u8],
55 limits: ReadLimits,
56 projection: ScanProjection,
57 ) -> Result<Box<dyn Iterator<Item = Result<(Vec<u8>, Vec<u8>), StorageError>> + 'a>, StorageError>;
58
59 /// `ScanCursor`'s v2 native-iterator fast path (B7 follow-up): a genuine
60 /// backend range cursor over `(lower, upper)` matching `ScanCursor`'s own
61 /// `Bound` representation exactly, unlike [`Self::current_range`]'s
62 /// simplified included-start/excluded-end-or-unbounded convention. A
63 /// real cursor walks consecutively (one seek for the whole scan, not one
64 /// per row); callers must already know this view's `current` fast path
65 /// is valid — same gate [`Self::read_value`]'s `current` flag needs,
66 /// since this reads the live current-value table.
67 ///
68 /// `Ok(None)` means this backend has no such cursor; the caller falls
69 /// back to [`Self::next_key`] + [`Self::read_value`] per row. The
70 /// default is `Ok(None)` so adding this is additive — a backend opts in
71 /// explicitly.
72 fn native_range<'a>(
73 &'a self,
74 _lower: &Bound<Vec<u8>>,
75 _upper: &Bound<Vec<u8>>,
76 _limits: ReadLimits,
77 _projection: ScanProjection,
78 ) -> Result<
79 Option<Box<dyn Iterator<Item = Result<(Vec<u8>, Vec<u8>), StorageError>> + 'a>>,
80 StorageError,
81 > {
82 Ok(None)
83 }
84
85 /// Durable LSN this view can serve.
86 fn visible_ci(&self) -> Result<u64, StorageError>;
87
88 /// Oldest commit for which this view still retains history.
89 fn history_floor(&self) -> Result<u64, StorageError>;
90
91 /// Whether this view is a physically pinned snapshot, so the `current`
92 /// fast path (ignore `ci`, read the live table) stays valid after later
93 /// commits. Redb read transactions pin pages (true); the live in-memory
94 /// maps do not (false), so a view must resolve reads by `ci`.
95 fn snapshot_pinned(&self) -> bool {
96 false
97 }
98}
99
100/// Backend-loaded state used to bootstrap a store.
101#[derive(Debug, Default)]
102pub struct LoadedState {
103 /// Current key/value data.
104 pub data: BTreeMap<Vec<u8>, Vec<u8>>,
105 /// Creation commit index per live key.
106 pub created: BTreeMap<Vec<u8>, u64>,
107 /// Retained superseded versions per key.
108 pub history: BTreeMap<Vec<u8>, Vec<VersionEntry>>,
109 /// Durable GC floor.
110 pub gc_watermark: u64,
111}
112
113/// [`StorageBackend::compact`]'s error when a read transaction is open: the
114/// caller retries with every lock released.
115pub(crate) const COMPACT_BUSY: &str = "compaction busy: read transactions open";
116
117/// Write/durability contract of the physical storage substrate.
118///
119/// Implementations must make `apply`/`apply_many` atomic and, when the
120/// configured durability is `Local`, durable before returning. `retire_before`
121/// reclaims retained versions; `compact` releases freed pages.
122///
123/// Visibility is `pub(crate)` in W3: the seam already isolates the write/read
124/// surface so a future async backend (`io_uring` / direct IO) or `RushDB`
125/// `PageStore` can be swapped in without touching LMPG/query/search. The trait
126/// becomes `pub` once an external backend implementation needs to name it.
127pub(crate) trait StorageBackend: Send + Sync {
128 /// Load the full current state (data, MVCC metadata, GC watermark).
129 fn load(&self) -> Result<LoadedState, String>;
130
131 /// Read only the durable-LSN watermark and GC floor: startup needs
132 /// these to restore the committed version WITHOUT materializing the full
133 /// dataset into the RAM mirror. Implementations should read only the two
134 /// metadata keys; the default falls back to a full `load`.
135 fn load_watermark(&self) -> Result<(u64, u64), String> {
136 let state = self.load()?;
137 let lsn = match state.data.get(crate::LAST_APPLIED_KEY) {
138 Some(value) => crate::wal::decode_last_applied(value)
139 .map_err(|error| format!("corrupt last-applied watermark: {error}"))?,
140 None => 0,
141 };
142 Ok((lsn, state.gc_watermark))
143 }
144
145 /// Apply one logical op-set at `commit_index`.
146 fn apply(&self, ops: &[Op], commit_index: u64) -> Result<(), String>;
147
148 /// Apply many op-sets (ascending commit-index order) in ONE write
149 /// transaction: a single durable acknowledgement covers the whole batch.
150 fn apply_many(&self, batches: &[(u64, &[Op])]) -> Result<(), String>;
151
152 /// Reclaim retained versions superseded before `watermark`.
153 fn retire_before(&self, watermark: u64) -> Result<(), String>;
154
155 /// Reclaim a bounded batch of history keys superseded before `watermark`,
156 /// resuming after `cursor`. Returns the next key to resume from, or `None`
157 /// when the walk is complete. Each call is one write transaction, so GC is
158 /// bounded and resumable instead of one monolithic txn.
159 fn retire_before_batch(
160 &self,
161 watermark: u64,
162 _cursor: Option<Vec<u8>>,
163 _max_keys: usize,
164 ) -> Result<Option<Vec<u8>>, String> {
165 // Default: fall back to the monolithic reclamation. Implementations
166 // that support bounded GC override this.
167 self.retire_before(watermark)?;
168 Ok(None)
169 }
170
171 /// Signal in-flight maintenance scans to stop promptly (shutdown). Backends
172 /// with no long scans treat this as a no-op.
173 fn begin_shutdown(&self) {}
174
175 /// Force queued writes durable (fsync) without changing logical state, and
176 /// durably record `durable_lsn` as the materialization checkpoint. Under
177 /// log-first the WAL is the durable truth and the backend is buffered, so a
178 /// checkpoint is what makes the materialization durable and lets WAL
179 /// segments at or below `durable_lsn` be truncated. Backends with no
180 /// buffered writes treat this as a no-op.
181 fn checkpoint(&self, _durable_lsn: u64) -> Result<(), String> {
182 Ok(())
183 }
184
185 /// Run `run` while this backend commits nothing (readers continue), so
186 /// its file can be copied consistently. Backends without a file just run
187 /// it.
188 fn while_writes_paused(&self, run: &mut dyn FnMut()) -> Result<(), String> {
189 run();
190 Ok(())
191 }
192
193 /// Compact the backend, releasing freed pages. Returns [`COMPACT_BUSY`]
194 /// when a read transaction is open; it never waits for one.
195 fn compact(&self) -> Result<(), String>;
196
197 /// Durable byte accounting by key family. Default implementations return
198 /// an explicit error (unavailable); file-backed backends override it.
199 fn storage_stats(&self) -> Result<StorageStatistics, String> {
200 Err("backend does not provide storage statistics".into())
201 }
202
203 /// Open an owned read view (snapshot) of the current committed state.
204 fn read_view(&self) -> Result<Box<dyn StorageReadView>, StorageError>;
205
206 /// Cumulative time spent waiting for and holding the backend's database
207 /// lock. Backends without such a lock report zeros.
208 fn lock_times(&self) -> crate::BackendLockTimes {
209 crate::BackendLockTimes::default()
210 }
211 /// Writer holds by holder. Backends without a single writer
212 /// report none.
213 fn writer_holds(&self) -> Vec<(crate::WriterHolder, crate::WriterHolds)> {
214 Vec::new()
215 }
216}
217
218/// Open the redb backend as a [`StorageBackend`] handle.
219pub(crate) fn open_backend(path: &Path, cache: ByteReservation) -> Result<BackendHandle, String> {
220 Ok(Arc::new(crate::redb_backend::RedbBackend::open(
221 path, cache,
222 )?))
223}
224
225/// Convenience alias used by [`Store`] for its durable backend handle.
226pub(crate) type BackendHandle = Arc<dyn StorageBackend>;