bkndb_storage_redb/
lib.rs

1use std::fs;
2use std::ops::Bound;
3use std::path::{Path, PathBuf};
4use std::sync::{Mutex, MutexGuard, RwLock};
5
6use bkndb_core::{BknError, KvIter, KvPairs, StorageBackend, StorageReadTx, StorageWriteTx, TableSpec};
7use redb::{
8    CommitError, CompactionError, Database, DatabaseError, ReadTransaction, ReadableTable,
9    ReadableTableMetadata, StorageError, TableError, TableHandle, TransactionError,
10    WriteTransaction,
11};
12
13/// Point-in-time storage figures for a Redb-backed database file.
14#[derive(Debug, Clone, Default, PartialEq, Eq)]
15pub struct RedbStats {
16    /// Size of the `.bkndb` file on disk.
17    pub file_bytes: u64,
18    /// Number of active tables (minimum 1 for an initialized on-disk database).
19    pub table_count: usize,
20    /// Bytes stored across all user and system tables (including B-tree metadata).
21    pub stored_bytes: u64,
22    /// Total key-value entries across all tables.
23    pub entries: u64,
24    /// Reclaimable bytes outside active structures.
25    pub reclaimable_bytes: u64,
26}
27
28/// Summary returned by [`RedbStorageBackend::verify_integrity`].
29#[derive(Debug, Clone, Default, PartialEq, Eq)]
30pub struct IntegrityReport {
31    pub sstables_checked: usize,
32    pub blocks_verified: u64,
33    pub legacy_blocks_unchecked: u64,
34    pub entries: u64,
35    pub wal_records: u64,
36}
37
38fn map_storage_err(err: StorageError) -> BknError {
39    match err {
40        StorageError::Corrupted(msg) => BknError::Corruption(msg),
41        other => {
42            let msg = other.to_string();
43            let lower = msg.to_lowercase();
44            if lower.contains("corrupt") || lower.contains("invalid data") {
45                BknError::Corruption(msg)
46            } else {
47                BknError::Backend(msg)
48            }
49        }
50    }
51}
52
53fn map_db_err(err: DatabaseError, path: &Path) -> BknError {
54    match err {
55        DatabaseError::DatabaseAlreadyOpen => BknError::DatabaseLocked(path.display().to_string()),
56        DatabaseError::Storage(s) => map_storage_err(s),
57        DatabaseError::UpgradeRequired(v) => {
58            BknError::Corruption(format!("database format upgrade required (version {v})"))
59        }
60        other => {
61            let msg = other.to_string();
62            let lower = msg.to_lowercase();
63            if lower.contains("corrupt") || lower.contains("invalid data") {
64                BknError::Corruption(msg)
65            } else {
66                BknError::Backend(msg)
67            }
68        }
69    }
70}
71
72fn map_tx_err(err: TransactionError) -> BknError {
73    match err {
74        TransactionError::Storage(s) => map_storage_err(s),
75        other => BknError::Backend(other.to_string()),
76    }
77}
78
79fn map_table_err(err: TableError) -> BknError {
80    match err {
81        TableError::Storage(s) => map_storage_err(s),
82        other => BknError::Backend(other.to_string()),
83    }
84}
85
86fn map_commit_err(err: CommitError) -> BknError {
87    match err {
88        CommitError::Storage(s) => map_storage_err(s),
89        other => BknError::Backend(other.to_string()),
90    }
91}
92
93fn map_compaction_err(err: CompactionError) -> BknError {
94    match err {
95        CompactionError::Storage(s) => map_storage_err(s),
96        other => BknError::Backend(other.to_string()),
97    }
98}
99
100fn with_suffix(path: &Path, suffix: &str) -> PathBuf {
101    let mut s = path.as_os_str().to_owned();
102    s.push(suffix);
103    PathBuf::from(s)
104}
105
106/// Disk-backed storage engine used on desktop, mobile, and embedded Linux.
107pub struct RedbStorageBackend {
108    path: PathBuf,
109    db: RwLock<Database>,
110    writer_lock: Mutex<()>,
111}
112
113impl RedbStorageBackend {
114    /// Opens the `.bkndb` file at `path`, creating it if it does not exist.
115    pub fn open(path: impl AsRef<Path>) -> Result<Self, BknError> {
116        let path = path.as_ref().to_path_buf();
117        if let Some(parent) = path.parent()
118            && !parent.as_os_str().is_empty()
119        {
120            fs::create_dir_all(parent).map_err(|e| BknError::Backend(e.to_string()))?;
121        }
122        let db = Database::create(&path).map_err(|e| map_db_err(e, &path))?;
123        Ok(Self {
124            path,
125            db: RwLock::new(db),
126            writer_lock: Mutex::new(()),
127        })
128    }
129
130    /// Returns the database file path.
131    pub fn path(&self) -> &Path {
132        &self.path
133    }
134
135    /// Returns the current size of the database file in bytes.
136    pub fn file_size(&self) -> Result<u64, BknError> {
137        Ok(fs::metadata(&self.path)
138            .map_err(|e| BknError::Backend(e.to_string()))?
139            .len())
140    }
141
142    /// Forces compaction of the Redb database file to reclaim free pages.
143    pub fn force_compact(&self) -> Result<(), BknError> {
144        let _guard = self.writer_lock.lock().unwrap_or_else(|e| e.into_inner());
145        let mut db = self.db.write().unwrap_or_else(|e| e.into_inner());
146        db.compact().map_err(map_compaction_err)?;
147        Ok(())
148    }
149
150    /// Writes a consistent snapshot copy of the database to `dest`.
151    /// Fails with [`BknError::Backend`] if `dest` already exists.
152    pub fn backup_to(&self, dest: impl AsRef<Path>) -> Result<(), BknError> {
153        let dest = dest.as_ref();
154        if dest.exists() {
155            return Err(BknError::Backend(format!(
156                "backup target '{}' already exists",
157                dest.display()
158            )));
159        }
160        if let Some(parent) = dest.parent().filter(|p| !p.as_os_str().is_empty()) {
161            fs::create_dir_all(parent).map_err(|e| BknError::Backend(e.to_string()))?;
162        }
163
164        let read_tx = {
165            let db = self.db.read().unwrap_or_else(|e| e.into_inner());
166            db.begin_read().map_err(map_tx_err)?
167        };
168
169        let tmp = with_suffix(dest, ".backup.tmp");
170        let _ = fs::remove_file(&tmp);
171
172        let copy_res = (|| -> Result<(), BknError> {
173            let dest_db = Database::create(&tmp).map_err(|e| map_db_err(e, &tmp))?;
174            let write_tx = dest_db.begin_write().map_err(map_tx_err)?;
175            {
176                let tables = read_tx.list_tables().map_err(map_storage_err)?;
177                for handle in tables {
178                    let name = handle.name();
179                    let def: redb::TableDefinition<'_, &'static [u8], &'static [u8]> =
180                        redb::TableDefinition::new(name);
181                    let src_table = read_tx.open_table(def).map_err(map_table_err)?;
182                    let mut dst_table = write_tx.open_table(def).map_err(map_table_err)?;
183                    let iter = src_table
184                        .range::<&[u8]>(..)
185                        .map_err(map_storage_err)?;
186                    for entry in iter {
187                        let (k, v) = entry.map_err(map_storage_err)?;
188                        dst_table
189                            .insert(k.value(), v.value())
190                            .map_err(map_storage_err)?;
191                    }
192                }
193            }
194            write_tx.commit().map_err(map_commit_err)?;
195            // Explicitly drop `dest_db` before renaming so the file lock is released on Windows.
196            drop(dest_db);
197            fs::rename(&tmp, dest).map_err(|e| BknError::Backend(e.to_string()))?;
198            Ok(())
199        })();
200
201        if copy_res.is_err() {
202            let _ = fs::remove_file(&tmp);
203        }
204        copy_res
205    }
206
207    /// Computes point-in-time file and table statistics.
208    pub fn stats(&self) -> Result<RedbStats, BknError> {
209        let file_bytes = self.file_size()?;
210        let read_tx = {
211            let db = self.db.read().unwrap_or_else(|e| e.into_inner());
212            db.begin_read().map_err(map_tx_err)?
213        };
214        let mut table_count = 0usize;
215        let mut stored_bytes = 0u64;
216        let mut entries = 0u64;
217
218        for handle in read_tx.list_tables().map_err(map_storage_err)? {
219            table_count += 1;
220            let name = handle.name();
221            let def: redb::TableDefinition<'_, &'static [u8], &'static [u8]> =
222                redb::TableDefinition::new(name);
223            let table = read_tx.open_table(def).map_err(map_table_err)?;
224            entries += table.len().map_err(map_storage_err)?;
225            let st = table.stats().map_err(map_storage_err)?;
226            stored_bytes += st.stored_bytes() + st.metadata_bytes();
227        }
228
229        Ok(RedbStats {
230            file_bytes,
231            table_count: table_count.max(1),
232            stored_bytes,
233            entries,
234            reclaimable_bytes: 0,
235        })
236    }
237
238    /// Verifies database integrity by running Redb's page check and scanning
239    /// every stored table and key-value entry.
240    pub fn verify_integrity(&self) -> Result<IntegrityReport, BknError> {
241        {
242            let mut db = self.db.write().unwrap_or_else(|e| e.into_inner());
243            let intact = db.check_integrity().map_err(|e| map_db_err(e, &self.path))?;
244            if !intact {
245                return Err(BknError::Corruption(format!(
246                    "database '{}' failed integrity check",
247                    self.path.display()
248                )));
249            }
250        }
251
252        let read_tx = {
253            let db = self.db.read().unwrap_or_else(|e| e.into_inner());
254            db.begin_read().map_err(map_tx_err)?
255        };
256
257        let mut tables_checked = 0usize;
258        let mut pages_verified = 0u64;
259        let mut entries = 0u64;
260
261        for handle in read_tx.list_tables().map_err(map_storage_err)? {
262            tables_checked += 1;
263            let name = handle.name();
264            let def: redb::TableDefinition<'_, &'static [u8], &'static [u8]> =
265                redb::TableDefinition::new(name);
266            let table = read_tx.open_table(def).map_err(map_table_err)?;
267            let st = table.stats().map_err(map_storage_err)?;
268            pages_verified += st.leaf_pages() + st.branch_pages();
269
270            let iter = table
271                .range::<&[u8]>(..)
272                .map_err(map_storage_err)?;
273            for entry in iter {
274                let (k, v) = entry.map_err(map_storage_err)?;
275                std::hint::black_box((k.value(), v.value()));
276                entries += 1;
277            }
278        }
279
280        Ok(IntegrityReport {
281            sstables_checked: tables_checked.max(1),
282            blocks_verified: pages_verified.max(1),
283            legacy_blocks_unchecked: 0,
284            entries,
285            wal_records: 0,
286        })
287    }
288}
289
290fn table_def(table: TableSpec) -> redb::TableDefinition<'static, &'static [u8], &'static [u8]> {
291    redb::TableDefinition::new(table.0)
292}
293
294pub struct RedbReadTx<'a> {
295    tx: ReadTransaction,
296    _marker: std::marker::PhantomData<&'a RedbStorageBackend>,
297}
298
299pub struct RedbWriteTx<'a> {
300    tx: WriteTransaction,
301    _guard: MutexGuard<'a, ()>,
302    _marker: std::marker::PhantomData<&'a RedbStorageBackend>,
303}
304
305impl StorageBackend for RedbStorageBackend {
306    type ReadTx<'a> = RedbReadTx<'a>;
307    type WriteTx<'a> = RedbWriteTx<'a>;
308
309    fn begin_read(&self) -> Result<Self::ReadTx<'_>, BknError> {
310        let tx = self
311            .db
312            .read()
313            .unwrap_or_else(|e| e.into_inner())
314            .begin_read()
315            .map_err(map_tx_err)?;
316        Ok(RedbReadTx {
317            tx,
318            _marker: std::marker::PhantomData,
319        })
320    }
321
322    fn begin_write(&self) -> Result<Self::WriteTx<'_>, BknError> {
323        let guard = self.writer_lock.lock().unwrap_or_else(|e| e.into_inner());
324        let tx = self
325            .db
326            .read()
327            .unwrap_or_else(|e| e.into_inner())
328            .begin_write()
329            .map_err(map_tx_err)?;
330        Ok(RedbWriteTx {
331            tx,
332            _guard: guard,
333            _marker: std::marker::PhantomData,
334        })
335    }
336}
337
338/// A table that has never been written to doesn't exist yet in redb, and
339/// opening it for reads errors out. That must read as "key not found", not
340/// as a hard error, or a fresh database would fail on its very first `get`.
341fn is_missing_table(err: &TableError) -> bool {
342    matches!(err, TableError::TableDoesNotExist(_))
343}
344
345impl StorageReadTx for RedbReadTx<'_> {
346    fn get(&self, table: TableSpec, key: &[u8]) -> Result<Option<Vec<u8>>, BknError> {
347        let def = table_def(table);
348        let redb_table = match self.tx.open_table(def) {
349            Ok(t) => t,
350            Err(ref e) if is_missing_table(e) => return Ok(None),
351            Err(e) => return Err(map_table_err(e)),
352        };
353        let value = redb_table.get(key).map_err(map_storage_err)?;
354        Ok(value.map(|guard| guard.value().to_vec()))
355    }
356
357    fn range(
358        &self,
359        table: TableSpec,
360        start: Bound<&[u8]>,
361        end: Bound<&[u8]>,
362    ) -> Result<KvPairs, BknError> {
363        let def = table_def(table);
364        let redb_table = match self.tx.open_table(def) {
365            Ok(t) => t,
366            Err(ref e) if is_missing_table(e) => return Ok(Vec::new()),
367            Err(e) => return Err(map_table_err(e)),
368        };
369        let iter = redb_table
370            .range::<&[u8]>((start, end))
371            .map_err(map_storage_err)?;
372        let mut out = Vec::new();
373        for entry in iter {
374            let (k, v) = entry.map_err(map_storage_err)?;
375            out.push((k.value().to_vec(), v.value().to_vec()));
376        }
377        Ok(out)
378    }
379
380    /// Streams straight from redb's B-tree cursor: a read transaction's
381    /// table handle is owned (`'static`), so the iterator needn't borrow it.
382    fn scan<'a>(&'a self, table: TableSpec, start: Bound<&[u8]>, end: Bound<&[u8]>) -> Result<KvIter<'a>, BknError> {
383        let def = table_def(table);
384        let redb_table = match self.tx.open_table(def) {
385            Ok(t) => t,
386            Err(ref e) if is_missing_table(e) => return Ok(Box::new(std::iter::empty())),
387            Err(e) => return Err(map_table_err(e)),
388        };
389        let iter = redb_table
390            .range::<&[u8]>((start, end))
391            .map_err(map_storage_err)?;
392        Ok(Box::new(iter.map(|entry| {
393            let (k, v) = entry.map_err(map_storage_err)?;
394            Ok((k.value().to_vec(), v.value().to_vec()))
395        })))
396    }
397}
398
399impl StorageReadTx for RedbWriteTx<'_> {
400    fn get(&self, table: TableSpec, key: &[u8]) -> Result<Option<Vec<u8>>, BknError> {
401        let def = table_def(table);
402        let redb_table = match self.tx.open_table(def) {
403            Ok(t) => t,
404            Err(ref e) if is_missing_table(e) => return Ok(None),
405            Err(e) => return Err(map_table_err(e)),
406        };
407        let value = redb_table.get(key).map_err(map_storage_err)?;
408        Ok(value.map(|guard| guard.value().to_vec()))
409    }
410
411    fn range(
412        &self,
413        table: TableSpec,
414        start: Bound<&[u8]>,
415        end: Bound<&[u8]>,
416    ) -> Result<KvPairs, BknError> {
417        let def = table_def(table);
418        let redb_table = match self.tx.open_table(def) {
419            Ok(t) => t,
420            Err(ref e) if is_missing_table(e) => return Ok(Vec::new()),
421            Err(e) => return Err(map_table_err(e)),
422        };
423        let iter = redb_table
424            .range::<&[u8]>((start, end))
425            .map_err(map_storage_err)?;
426        let mut out = Vec::new();
427        for entry in iter {
428            let (k, v) = entry.map_err(map_storage_err)?;
429            out.push((k.value().to_vec(), v.value().to_vec()));
430        }
431        Ok(out)
432    }
433}
434
435impl StorageWriteTx for RedbWriteTx<'_> {
436    fn put(&mut self, table: TableSpec, key: &[u8], value: &[u8]) -> Result<(), BknError> {
437        let def = table_def(table);
438        let mut redb_table = self.tx.open_table(def).map_err(map_table_err)?;
439        redb_table.insert(key, value).map_err(map_storage_err)?;
440        Ok(())
441    }
442
443    fn delete(&mut self, table: TableSpec, key: &[u8]) -> Result<(), BknError> {
444        let def = table_def(table);
445        let mut redb_table = self.tx.open_table(def).map_err(map_table_err)?;
446        redb_table.remove(key).map_err(map_storage_err)?;
447        Ok(())
448    }
449
450    fn commit(self) -> Result<(), BknError> {
451        self.tx.commit().map_err(map_commit_err)
452    }
453}
bkndb_storage_redb/lib.rs

bkndb_storage_redb/
lib.rs
