nodedb_lite/nodedb/core/open/
recovery.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Opening a Lite database from a filesystem path.
4//!
5//! Corruption can surface at two points: while opening the backing `Db`, and
6//! later during identity load or CRDT/index restore, where it arrives typed as
7//! [`ErrorDetails::SegmentCorrupted`]. Both are the same fault to the caller,
8//! so both obey the same
9//! [`CorruptionPolicy`](crate::storage::corruption::CorruptionPolicy) — either
10//! the open fails and the store is left alone, or the caller has opted into
11//! discarding it and the whole open sequence is re-driven once against a fresh
12//! store.
13
14use std::path::Path;
15use std::sync::Arc;
16
17use nodedb_types::error::{ErrorDetails, NodeDbResult};
18
19use crate::config::LiteConfig;
20use crate::storage::encryption::Encryption;
21use crate::storage::pagedb_storage::PagedbStorageDefault;
22
23use crate::nodedb::core::types::NodeDbLite;
24
25impl NodeDbLite<PagedbStorageDefault> {
26    /// Open (or create) a Lite database at `path`.
27    ///
28    /// A store that cannot be read is reported and left untouched — the
29    /// [`FailClosed`](crate::storage::corruption::CorruptionPolicy::FailClosed) default. Use
30    /// [`open_at_path_with_config`](Self::open_at_path_with_config) with
31    /// `corruption_policy` set to choose otherwise.
32    pub async fn open_at_path(
33        path: impl AsRef<Path>,
34        encryption: Encryption,
35    ) -> NodeDbResult<Arc<Self>> {
36        Self::open_at_path_with_config(path, encryption, LiteConfig::default()).await
37    }
38
39    /// Like [`open_at_path`](Self::open_at_path), but with an explicit
40    /// [`LiteConfig`].
41    ///
42    /// `config.corruption_policy` governs both corruption points. Under
43    /// [`DiscardStoreAndRecreate`](crate::storage::corruption::CorruptionPolicy::DiscardStoreAndRecreate) a post-open corruption
44    /// renames the store aside and retries the open exactly once; the retry is
45    /// unguarded, so a second corruption surfaces to the caller rather than
46    /// looping.
47    pub async fn open_at_path_with_config(
48        path: impl AsRef<Path>,
49        encryption: Encryption,
50        config: LiteConfig,
51    ) -> NodeDbResult<Arc<Self>> {
52        let path = path.as_ref();
53        let policy = config.corruption_policy;
54        let storage =
55            PagedbStorageDefault::open_with_policy(path, encryption.clone(), policy).await?;
56        match Self::open_with_config(storage, config.clone()).await {
57            Ok(db) => Ok(db),
58            Err(e)
59                if policy.may_discard()
60                    && matches!(e.details(), ErrorDetails::SegmentCorrupted { .. }) =>
61            {
62                tracing::error!(
63                    path = %path.display(),
64                    error = %e,
65                    "post-open corruption detected after storage opened cleanly — the caller \
66                     opted into discarding the store, renaming it aside and retrying once"
67                );
68                let storage = PagedbStorageDefault::discard_and_recreate(path, &encryption).await?;
69                Self::open_with_config(storage, config).await
70            }
71            Err(e) => Err(e),
72        }
73    }
74}
75
76#[cfg(test)]
77mod tests {
78    use std::sync::atomic::{AtomicBool, Ordering};
79
80    use async_trait::async_trait;
81    use nodedb_types::Namespace;
82    use nodedb_types::error::ErrorDetails;
83
84    use crate::error::LiteError;
85    use crate::nodedb::core::types::NodeDbLite;
86    use crate::storage::engine::{CompactionOutcome, KvPair, StorageEngine, WriteOp};
87    use crate::storage::pagedb_storage::PagedbStorageMem;
88
89    /// Wraps an in-memory engine and injects a single corruption on the first
90    /// read of the Lite identity key (`meta:lite_id`). This proves the
91    /// corruption signal travels typed all the way from a post-open storage
92    /// read to `NodeDbError::details() == ErrorDetails::SegmentCorrupted`.
93    struct CorruptOnFirstIdentityRead {
94        inner: PagedbStorageMem,
95        tripped: AtomicBool,
96    }
97
98    #[async_trait]
99    impl StorageEngine for CorruptOnFirstIdentityRead {
100        async fn get(&self, ns: Namespace, key: &[u8]) -> Result<Option<Vec<u8>>, LiteError> {
101            if ns == Namespace::Meta
102                && key == b"meta:lite_id"
103                && !self.tripped.swap(true, Ordering::SeqCst)
104            {
105                return Err(LiteError::Corrupted {
106                    detail: "injected identity-read corruption".into(),
107                });
108            }
109            self.inner.get(ns, key).await
110        }
111
112        async fn put(&self, ns: Namespace, key: &[u8], value: &[u8]) -> Result<(), LiteError> {
113            self.inner.put(ns, key, value).await
114        }
115
116        async fn delete(&self, ns: Namespace, key: &[u8]) -> Result<(), LiteError> {
117            self.inner.delete(ns, key).await
118        }
119
120        async fn scan_prefix(
121            &self,
122            ns: Namespace,
123            prefix: &[u8],
124        ) -> Result<Vec<KvPair>, LiteError> {
125            self.inner.scan_prefix(ns, prefix).await
126        }
127
128        async fn batch_write(&self, ops: &[WriteOp]) -> Result<(), LiteError> {
129            self.inner.batch_write(ops).await
130        }
131
132        async fn count(&self, ns: Namespace) -> Result<u64, LiteError> {
133            self.inner.count(ns).await
134        }
135
136        async fn compact(&self) -> Result<CompactionOutcome, LiteError> {
137            self.inner.compact().await
138        }
139
140        async fn scan_range(
141            &self,
142            ns: Namespace,
143            start: &[u8],
144            limit: usize,
145        ) -> Result<Vec<KvPair>, LiteError> {
146            self.inner.scan_range(ns, start, limit).await
147        }
148
149        async fn scan_range_bounded(
150            &self,
151            ns: Namespace,
152            start: Option<&[u8]>,
153            end: Option<&[u8]>,
154            limit: Option<usize>,
155        ) -> Result<Vec<KvPair>, LiteError> {
156            self.inner.scan_range_bounded(ns, start, end, limit).await
157        }
158    }
159
160    #[tokio::test]
161    async fn post_open_corruption_surfaces_typed() {
162        let inner = PagedbStorageMem::open_in_memory()
163            .await
164            .expect("in-memory storage opens");
165        let wrapped = CorruptOnFirstIdentityRead {
166            inner,
167            tripped: AtomicBool::new(false),
168        };
169
170        // `NodeDbLite` isn't `Debug`, so match rather than `expect_err`.
171        let err = match NodeDbLite::open(wrapped).await {
172            Ok(_) => panic!("first identity read is corrupted, so open must fail"),
173            Err(e) => e,
174        };
175
176        assert!(
177            matches!(err.details(), ErrorDetails::SegmentCorrupted { .. }),
178            "corruption must propagate as SegmentCorrupted, got: {:?}",
179            err.details()
180        );
181    }
182}
nodedb_lite/nodedb/core/open/recovery.rs

nodedb_lite/nodedb/core/open/
recovery.rs
