nodedb_lite/nodedb/core/open/
recovery.rs1use 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 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 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 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 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}