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`, borrowed: when the key is visible, `visit` sees
31 /// the stored value bytes in place (empty for a keys-only projection)
32 /// and the result is `true`; otherwise `visit` is not called. `current`
33 /// selects the latest-commit fast path. This is a backend's one
34 /// implementation of point-read visibility, limits and read accounting;
35 /// [`Self::read_value`] copies out of it. `visit` runs while the backend
36 /// holds its read (a redb page guard, the memory maps' read locks), so it
37 /// must not read the store again.
38 fn visit_value(
39 &self,
40 key: &[u8],
41 ci: u64,
42 current: bool,
43 limits: ReadLimits,
44 projection: ScanProjection,
45 visit: &mut dyn FnMut(&[u8]),
46 ) -> Result<bool, StorageError>;
47
48 /// Point read at `ci` into an owned copy ([`Self::visit_value`]).
49 fn read_value(
50 &self,
51 key: &[u8],
52 ci: u64,
53 current: bool,
54 limits: ReadLimits,
55 projection: ScanProjection,
56 ) -> Result<Option<Vec<u8>>, StorageError> {
57 let mut value = None;
58 self.visit_value(key, ci, current, limits, projection, &mut |raw| {
59 value = Some(raw.to_vec());
60 })?;
61 Ok(value)
62 }
63
64 /// Smallest key in `(lower, upper)`; `include_history` widens to retained
65 /// versions for a non-current snapshot.
66 fn next_key(
67 &self,
68 lower: &Bound<Vec<u8>>,
69 upper: &Bound<Vec<u8>>,
70 limits: ReadLimits,
71 include_history: bool,
72 ) -> Result<Option<Vec<u8>>, StorageError>;
73
74 /// Ordered current-value range with the requested projection.
75 fn current_range<'a>(
76 &'a self,
77 start: &[u8],
78 end: &[u8],
79 limits: ReadLimits,
80 projection: ScanProjection,
81 ) -> Result<Box<dyn Iterator<Item = Result<(Vec<u8>, Vec<u8>), StorageError>> + 'a>, StorageError>;
82
83 /// `ScanCursor`'s v2 native-iterator fast path (B7 follow-up): a genuine
84 /// backend range cursor over `(lower, upper)` matching `ScanCursor`'s own
85 /// `Bound` representation exactly, unlike [`Self::current_range`]'s
86 /// simplified included-start/excluded-end-or-unbounded convention. A
87 /// real cursor walks consecutively (one seek for the whole scan, not one
88 /// per row); callers must already know this view's `current` fast path
89 /// is valid — same gate [`Self::visit_value`]'s `current` flag needs,
90 /// since this reads the live current-value table.
91 ///
92 /// `Ok(None)` means this backend has no such cursor; the caller falls
93 /// back to [`Self::next_key`] + [`Self::visit_value`] per row. The
94 /// default is `Ok(None)` so adding this is additive — a backend opts in
95 /// explicitly.
96 fn native_range<'a>(
97 &'a self,
98 _lower: &Bound<Vec<u8>>,
99 _upper: &Bound<Vec<u8>>,
100 _limits: ReadLimits,
101 _projection: ScanProjection,
102 ) -> Result<
103 Option<Box<dyn Iterator<Item = Result<(Vec<u8>, Vec<u8>), StorageError>> + 'a>>,
104 StorageError,
105 > {
106 Ok(None)
107 }
108
109 /// Durable LSN this view can serve.
110 fn visible_ci(&self) -> Result<u64, StorageError>;
111
112 /// Oldest commit for which this view still retains history.
113 fn history_floor(&self) -> Result<u64, StorageError>;
114
115 /// Whether this view is a physically pinned snapshot, so the `current`
116 /// fast path (ignore `ci`, read the live table) stays valid after later
117 /// commits. Redb read transactions pin pages (true); the live in-memory
118 /// maps do not (false), so a view must resolve reads by `ci`.
119 fn snapshot_pinned(&self) -> bool {
120 false
121 }
122}
123
124/// One commit's writes for a backend: its base ops followed by the ops
125/// derived from them, applied at `ci` in the same transaction. The two are
126/// kept as borrowed slices so publication never concatenates them into a
127/// copy of the commit.
128#[derive(Debug, Clone, Copy)]
129pub(crate) struct CommitOps<'a> {
130 pub(crate) ci: u64,
131 pub(crate) base: &'a [Op],
132 pub(crate) derived: &'a [Op],
133}
134
135impl<'a> CommitOps<'a> {
136 /// Base ops then derived ops, in apply order.
137 pub(crate) fn ops(&self) -> impl Iterator<Item = &'a Op> + Clone {
138 let (base, derived) = (self.base, self.derived);
139 base.iter().chain(derived)
140 }
141}
142
143/// Backend-loaded state used to bootstrap a store.
144#[derive(Debug, Default)]
145pub struct LoadedState {
146 /// Current key/value data.
147 pub data: BTreeMap<Vec<u8>, Vec<u8>>,
148 /// Creation commit index per live key.
149 pub created: BTreeMap<Vec<u8>, u64>,
150 /// Retained superseded versions per key.
151 pub history: BTreeMap<Vec<u8>, Vec<VersionEntry>>,
152 /// Durable GC floor.
153 pub gc_watermark: u64,
154}
155
156/// [`StorageBackend::compact`]'s error when a read transaction is open: the
157/// caller retries with every lock released.
158pub(crate) const COMPACT_BUSY: &str = "compaction busy: read transactions open";
159
160/// Write/durability contract of the physical storage substrate.
161///
162/// Implementations must make `apply`/`apply_many` atomic and, when the
163/// configured durability is `Local`, durable before returning. `retire_before`
164/// reclaims retained versions; `compact` releases freed pages.
165///
166/// Visibility is `pub(crate)` in W3: the seam already isolates the write/read
167/// surface so a future async backend (`io_uring` / direct IO) or `RushDB`
168/// `PageStore` can be swapped in without touching LMPG/query/search. The trait
169/// becomes `pub` once an external backend implementation needs to name it.
170pub(crate) trait StorageBackend: Send + Sync {
171 /// Load the full current state (data, MVCC metadata, GC watermark).
172 fn load(&self) -> Result<LoadedState, String>;
173
174 /// Read only the durable-LSN watermark and GC floor: startup needs
175 /// these to restore the committed version WITHOUT materializing the full
176 /// dataset into the RAM mirror. Implementations should read only the two
177 /// metadata keys; the default falls back to a full `load`.
178 fn load_watermark(&self) -> Result<(u64, u64), String> {
179 let state = self.load()?;
180 let lsn = match state.data.get(crate::LAST_APPLIED_KEY) {
181 Some(value) => crate::wal::decode_last_applied(value)
182 .map_err(|error| format!("corrupt last-applied watermark: {error}"))?,
183 None => 0,
184 };
185 Ok((lsn, state.gc_watermark))
186 }
187
188 /// Apply one logical op-set at `commit_index`.
189 fn apply(&self, ops: &[Op], commit_index: u64) -> Result<(), String>;
190
191 /// Apply many commits (ascending commit-index order) in ONE write
192 /// transaction: a single durable acknowledgement covers the whole batch.
193 fn apply_many(&self, batches: &[CommitOps<'_>]) -> Result<(), String>;
194
195 /// Reclaim retained versions superseded before `watermark`.
196 fn retire_before(&self, watermark: u64) -> Result<(), String>;
197
198 /// Reclaim a bounded batch of history keys superseded before `watermark`,
199 /// resuming after `cursor`. Returns the next key to resume from, or `None`
200 /// when the walk is complete. Each call is one write transaction, so GC is
201 /// bounded and resumable instead of one monolithic txn.
202 fn retire_before_batch(
203 &self,
204 watermark: u64,
205 _cursor: Option<Vec<u8>>,
206 _max_keys: usize,
207 ) -> Result<Option<Vec<u8>>, String> {
208 // Default: fall back to the monolithic reclamation. Implementations
209 // that support bounded GC override this.
210 self.retire_before(watermark)?;
211 Ok(None)
212 }
213
214 /// Signal in-flight maintenance scans to stop promptly (shutdown). Backends
215 /// with no long scans treat this as a no-op.
216 fn begin_shutdown(&self) {}
217
218 /// Force queued writes durable (fsync) without changing logical state, and
219 /// durably record `durable_lsn` as the materialization checkpoint. Under
220 /// log-first the WAL is the durable truth and the backend is buffered, so a
221 /// checkpoint is what makes the materialization durable and lets WAL
222 /// segments at or below `durable_lsn` be truncated.
223 ///
224 /// Returns whether anything was written: a checkpoint with no buffered
225 /// write since the last one, at the same `durable_lsn`, writes nothing.
226 /// Backends with no buffered writes always return `false`.
227 fn checkpoint(&self, _durable_lsn: u64) -> Result<bool, String> {
228 Ok(false)
229 }
230
231 /// Run `run` while this backend commits nothing (readers continue), so
232 /// its file can be copied consistently. Backends without a file just run
233 /// it.
234 fn while_writes_paused(&self, run: &mut dyn FnMut()) -> Result<(), String> {
235 run();
236 Ok(())
237 }
238
239 /// Compact the backend, releasing freed pages. Returns [`COMPACT_BUSY`]
240 /// when a read transaction is open; it never waits for one.
241 fn compact(&self) -> Result<(), String>;
242
243 /// Durable byte accounting by key family. Default implementations return
244 /// an explicit error (unavailable); file-backed backends override it.
245 fn storage_stats(&self) -> Result<StorageStatistics, String> {
246 Err("backend does not provide storage statistics".into())
247 }
248
249 /// Open an owned read view (snapshot) of the current committed state.
250 fn read_view(&self) -> Result<Box<dyn StorageReadView>, StorageError>;
251
252 /// Cumulative time spent waiting for and holding the backend's database
253 /// lock. Backends without such a lock report zeros.
254 fn lock_times(&self) -> crate::BackendLockTimes {
255 crate::BackendLockTimes::default()
256 }
257 /// Writer holds by holder. Backends without a single writer
258 /// report none.
259 fn writer_holds(&self) -> Vec<(crate::WriterHolder, crate::WriterHolds)> {
260 Vec::new()
261 }
262}
263
264/// Open the redb backend as a [`StorageBackend`] handle.
265pub(crate) fn open_backend(path: &Path, cache: ByteReservation) -> Result<BackendHandle, String> {
266 Ok(Arc::new(crate::redb_backend::RedbBackend::open(
267 path, cache,
268 )?))
269}
270
271/// Convenience alias used by [`Store`] for its durable backend handle.
272pub(crate) type BackendHandle = Arc<dyn StorageBackend>;