storage_fjall/
lib.rs

1//! Fjall-backed persistence backend for the EML append-only library.
2//!
3//! Implements the library's generic `Storage` trait, so it serves any EML
4//! instantiation (eml, emt, …) rather than a specific one.
5
6use std::path::Path;
7
8use fjall::{Database, Keyspace, KeyspaceCreateOptions, PersistMode};
9use polydigest::{AlgorithmMetas, Storage};
10
11/// Reserved 9-byte key for log metadata in the `eml_metadata` keyspace.
12///
13/// All algorithm-epoch keys are exactly 8 bytes (alg_id as big-endian u64), so
14/// this 9-byte key never collides with any valid algorithm entry.
15const LOG_META_KEY: [u8; 9] = *b"_logmeta_";
16
17/// 10-byte key prefix for per-algorithm checkpoint roots in `eml_metadata`.
18///
19/// Format: `[b'r', b'o', alg_id_be_bytes[0..8]]` — does not collide with the
20/// 8-byte epoch keys or the 9-byte LOG_META_KEY.
21fn checkpoint_root_key(alg_id: u64) -> [u8; 10] {
22    let mut key = [0u8; 10];
23    key[0] = b'r';
24    key[1] = b'o';
25    key[2..10].copy_from_slice(&alg_id.to_be_bytes());
26    key
27}
28
29/// Error type for [`FjallStorage`] operations.
30#[derive(Debug, thiserror::Error)]
31pub enum FjallStorageError {
32    /// An error occurred in the underlying database engine.
33    #[error("Database error: {0}")]
34    Database(String),
35
36    /// An I/O error occurred.
37    #[error("I/O error: {0}")]
38    Io(#[from] std::io::Error),
39
40    /// Serialized epoch data is malformed or corrupted.
41    #[error("Epoch metadata corruption: {0}")]
42    MetadataCorruption(String),
43
44    /// The supplied keyspace-scoping prefix is invalid.
45    #[error("invalid keyspace prefix: {0:?}")]
46    InvalidPrefix(String),
47}
48
49/// A production-grade EML storage backend backed by a Fjall database.
50///
51/// Not `Clone` — enforces single-writer ownership so the tree-level
52/// read-count→write-at-count window cannot race across aliased handles.
53pub struct FjallStorage {
54    #[allow(dead_code)]
55    db: Database,
56    leaves: Keyspace,
57    nodes: Keyspace,
58    metadata: Keyspace,
59}
60
61impl FjallStorage {
62    /// Open or create a new Merkle log storage database at the specified directory path.
63    ///
64    /// # Errors
65    ///
66    /// Returns a [`FjallStorageError`] if the directory cannot be created or the database
67    /// fails to initialize.
68    pub fn open(path: &Path) -> Result<Self, FjallStorageError> {
69        let db = Database::builder(path)
70            .open()
71            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
72        Self::with_database(db)
73    }
74
75    /// Initialize storage keyspaces using an existing, shared Fjall database.
76    ///
77    /// Lets a caller that owns its own [`Database`] (e.g. to share one
78    /// physical database — one WAL, atomic cross-keyspace batches — with
79    /// other partitions it manages) hand it to this storage backend instead
80    /// of having [`Self::open`] create a dedicated one.
81    pub fn with_database(db: Database) -> Result<Self, FjallStorageError> {
82        let leaves = db
83            .keyspace("eml_leaves", KeyspaceCreateOptions::default)
84            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
85        let nodes = db
86            .keyspace("eml_nodes", KeyspaceCreateOptions::default)
87            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
88        let metadata = db
89            .keyspace("eml_metadata", KeyspaceCreateOptions::default)
90            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
91
92        Ok(Self {
93            db,
94            leaves,
95            nodes,
96            metadata,
97        })
98    }
99
100    /// Initialize storage keyspaces under a caller-supplied scope, using an
101    /// existing, shared Fjall database.
102    ///
103    /// [`Self::with_database`] opens fixed keyspace names
104    /// (`eml_leaves`/`eml_nodes`/`eml_metadata`), so two [`FjallStorage`]
105    /// instances sharing one physical [`Database`] would read and write the
106    /// same keyspaces. That's fine for content-addressed data, but EML's
107    /// leaf/node keys are positional (leaf index; alg_id+left+height), so
108    /// two unrelated logical logs' "leaf 0" would collide at the identical
109    /// key. This constructor opens `{prefix}_eml_leaves` /
110    /// `{prefix}_eml_nodes` / `{prefix}_eml_metadata` instead, so distinct
111    /// prefixes give each logical log (e.g. each Cyphr principal) its own
112    /// isolated triplet within the same database.
113    ///
114    /// # Errors
115    ///
116    /// Returns [`FjallStorageError::InvalidPrefix`] if `prefix` is empty,
117    /// contains characters outside fjall's documented keyspace-name charset
118    /// (alphanumeric, `_`, `-`, `.`, `#`, `$`), or would produce a keyspace
119    /// name exceeding fjall's 255-byte limit. Returns
120    /// [`FjallStorageError::Database`] if the underlying keyspaces fail to
121    /// open.
122    pub fn with_database_scoped(db: Database, prefix: &str) -> Result<Self, FjallStorageError> {
123        if prefix.is_empty()
124            || !prefix
125                .chars()
126                .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.' | '#' | '$'))
127        {
128            return Err(FjallStorageError::InvalidPrefix(prefix.to_string()));
129        }
130
131        let leaves_name = format!("{prefix}_eml_leaves");
132        let nodes_name = format!("{prefix}_eml_nodes");
133        let metadata_name = format!("{prefix}_eml_metadata");
134
135        // metadata_name is always the longest of the three suffixes, so
136        // bounding it bounds the other two.
137        if metadata_name.len() > 255 {
138            return Err(FjallStorageError::InvalidPrefix(prefix.to_string()));
139        }
140
141        let leaves = db
142            .keyspace(&leaves_name, KeyspaceCreateOptions::default)
143            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
144        let nodes = db
145            .keyspace(&nodes_name, KeyspaceCreateOptions::default)
146            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
147        let metadata = db
148            .keyspace(&metadata_name, KeyspaceCreateOptions::default)
149            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
150
151        Ok(Self {
152            db,
153            leaves,
154            nodes,
155            metadata,
156        })
157    }
158}
159
160impl Storage for FjallStorage {
161    type Error = FjallStorageError;
162
163    async fn store_leaf(&mut self, index: u64, data: &[u8]) -> Result<(), Self::Error> {
164        let key = index.to_be_bytes();
165        self.leaves
166            .insert(key, data)
167            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
168        Ok(())
169    }
170
171    async fn get_leaf(&self, index: u64) -> Result<Vec<u8>, Self::Error> {
172        let key = index.to_be_bytes();
173        let value = self
174            .leaves
175            .get(key)
176            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
177        match value {
178            Some(bytes) => Ok(bytes.to_vec()),
179            None => Err(FjallStorageError::Io(std::io::Error::new(
180                std::io::ErrorKind::NotFound,
181                format!("leaf index {index} not found"),
182            ))),
183        }
184    }
185
186    async fn len(&self) -> Result<u64, Self::Error> {
187        match self.leaves.iter().next_back() {
188            None => Ok(0),
189            Some(guard) => {
190                let key = guard
191                    .key()
192                    .map_err(|e| FjallStorageError::Database(e.to_string()))?;
193                let arr: [u8; 8] = key.as_ref().try_into().map_err(|_| {
194                    FjallStorageError::MetadataCorruption("leaf key has wrong length".to_string())
195                })?;
196                Ok(u64::from_be_bytes(arr) + 1)
197            },
198        }
199    }
200
201    async fn store_node(
202        &mut self,
203        alg_id: u64,
204        left: u64,
205        height: u32,
206        hash: &[u8],
207    ) -> Result<(), Self::Error> {
208        let mut key = [0u8; 20];
209        key[0..8].copy_from_slice(&alg_id.to_be_bytes());
210        key[8..16].copy_from_slice(&left.to_be_bytes());
211        key[16..20].copy_from_slice(&height.to_be_bytes());
212
213        self.nodes
214            .insert(key, hash)
215            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
216        Ok(())
217    }
218
219    async fn get_node(
220        &self,
221        alg_id: u64,
222        left: u64,
223        height: u32,
224    ) -> Result<Option<Vec<u8>>, Self::Error> {
225        let mut key = [0u8; 20];
226        key[0..8].copy_from_slice(&alg_id.to_be_bytes());
227        key[8..16].copy_from_slice(&left.to_be_bytes());
228        key[16..20].copy_from_slice(&height.to_be_bytes());
229
230        let value = self
231            .nodes
232            .get(key)
233            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
234        Ok(value.map(|bytes| bytes.to_vec()))
235    }
236
237    async fn store_algorithm_meta(
238        &mut self,
239        alg_id: u64,
240        epochs: &[(u64, u64)],
241    ) -> Result<(), Self::Error> {
242        let key = alg_id.to_be_bytes();
243        let mut bytes = Vec::with_capacity(epochs.len() * 16);
244        for &(start, end) in epochs {
245            bytes.extend_from_slice(&start.to_be_bytes());
246            bytes.extend_from_slice(&end.to_be_bytes());
247        }
248        self.metadata
249            .insert(key, bytes)
250            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
251        Ok(())
252    }
253
254    async fn load_algorithm_metas(&self) -> Result<AlgorithmMetas, Self::Error> {
255        let mut metas = Vec::new();
256        for item in self.metadata.iter() {
257            let (key_bytes, val_bytes) = item
258                .into_inner()
259                .map_err(|e| FjallStorageError::Database(e.to_string()))?;
260            // Skip reserved entries: log-metadata (9 bytes) and checkpoint roots (10 bytes).
261            let key_ref = key_bytes.as_ref();
262            if key_ref == LOG_META_KEY
263                || (key_ref.len() == 10 && key_ref[0] == b'r' && key_ref[1] == b'o')
264            {
265                continue;
266            }
267            let alg_id = {
268                let arr: [u8; 8] = key_ref.try_into().map_err(|_| {
269                    FjallStorageError::MetadataCorruption("invalid key length".to_string())
270                })?;
271                u64::from_be_bytes(arr)
272            };
273
274            if val_bytes.len() % 16 != 0 {
275                return Err(FjallStorageError::MetadataCorruption(
276                    "metadata value length is not a multiple of 16".to_string(),
277                ));
278            }
279
280            let mut epochs = Vec::with_capacity(val_bytes.len() / 16);
281            for chunk in val_bytes.chunks_exact(16) {
282                let start_bytes =
283                    chunk[0..8]
284                        .try_into()
285                        .map_err(|e: std::array::TryFromSliceError| {
286                            FjallStorageError::MetadataCorruption(e.to_string())
287                        })?;
288                let end_bytes =
289                    chunk[8..16]
290                        .try_into()
291                        .map_err(|e: std::array::TryFromSliceError| {
292                            FjallStorageError::MetadataCorruption(e.to_string())
293                        })?;
294                let start = u64::from_be_bytes(start_bytes);
295                let end = u64::from_be_bytes(end_bytes);
296                epochs.push((start, end));
297            }
298            metas.push((alg_id, epochs));
299        }
300        Ok(metas)
301    }
302
303    async fn load_log_meta(&self) -> Result<Option<(u64, u8)>, Self::Error> {
304        let value = self
305            .metadata
306            .get(LOG_META_KEY)
307            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
308        match value {
309            None => Ok(None),
310            Some(bytes) => {
311                if bytes.len() != 9 {
312                    return Err(FjallStorageError::MetadataCorruption(
313                        "log_meta value must be 9 bytes".to_string(),
314                    ));
315                }
316                let count = u64::from_be_bytes(bytes[0..8].try_into().unwrap());
317                let kind = bytes[8];
318                Ok(Some((count, kind)))
319            },
320        }
321    }
322
323    async fn load_checkpoint_roots(&self) -> Result<Vec<(u64, Vec<u8>)>, Self::Error> {
324        let mut roots = Vec::new();
325        for item in self.metadata.iter() {
326            let (key_bytes, val_bytes) = item
327                .into_inner()
328                .map_err(|e| FjallStorageError::Database(e.to_string()))?;
329            let key_ref = key_bytes.as_ref();
330            if key_ref.len() == 10 && key_ref[0] == b'r' && key_ref[1] == b'o' {
331                let arr: [u8; 8] = key_ref[2..10].try_into().map_err(|_| {
332                    FjallStorageError::MetadataCorruption(
333                        "checkpoint root key has wrong alg_id length".to_string(),
334                    )
335                })?;
336                let alg_id = u64::from_be_bytes(arr);
337                roots.push((alg_id, val_bytes.to_vec()));
338            }
339        }
340        Ok(roots)
341    }
342
343    async fn write_batch(
344        &mut self,
345        leaves: &[(u64, &[u8])],
346        nodes: &[(u64, u64, u32, &[u8])],
347        algorithm_metas: &[(u64, &[(u64, u64)])],
348        log_meta: Option<(u64, u8)>,
349        checkpoint_roots: &[(u64, &[u8])],
350    ) -> Result<(), Self::Error> {
351        let mut batch = self.db.batch();
352
353        for &(index, data) in leaves {
354            let key = index.to_be_bytes();
355            batch.insert(&self.leaves, key, data);
356        }
357
358        for &(alg_id, left, height, hash) in nodes {
359            let mut key = [0u8; 20];
360            key[0..8].copy_from_slice(&alg_id.to_be_bytes());
361            key[8..16].copy_from_slice(&left.to_be_bytes());
362            key[16..20].copy_from_slice(&height.to_be_bytes());
363            batch.insert(&self.nodes, key, hash);
364        }
365
366        for &(alg_id, epochs) in algorithm_metas {
367            let key = alg_id.to_be_bytes();
368            let mut bytes = Vec::with_capacity(epochs.len() * 16);
369            for &(start, end) in epochs {
370                bytes.extend_from_slice(&start.to_be_bytes());
371                bytes.extend_from_slice(&end.to_be_bytes());
372            }
373            batch.insert(&self.metadata, key, bytes);
374        }
375
376        if let Some((count, kind)) = log_meta {
377            let mut value = [0u8; 9];
378            value[0..8].copy_from_slice(&count.to_be_bytes());
379            value[8] = kind;
380            batch.insert(&self.metadata, LOG_META_KEY, value);
381        }
382
383        for &(alg_id, root) in checkpoint_roots {
384            let key = checkpoint_root_key(alg_id);
385            batch.insert(&self.metadata, key, root);
386        }
387
388        batch
389            .durability(Some(PersistMode::SyncAll))
390            .commit()
391            .map_err(|e| FjallStorageError::Database(e.to_string()))?;
392
393        Ok(())
394    }
395}
storage_fjall/lib.rs

storage_fjall/
lib.rs
