nodedb_lite/engine/array/
catalog.rs

1//! Array catalog persisted in the `Array` namespace.
2//!
3//! Key layout: `catalog:{name}` → zerompk-encoded `ArrayCatalogEntry`.
4//! The full list of array names is stored under `catalog_index` so cold-start
5//! restoration can enumerate entries without a namespace scan.
6
7use std::collections::HashMap;
8use std::sync::Arc;
9
10use nodedb_array::schema::ArraySchema;
11use nodedb_types::Namespace;
12use serde::{Deserialize, Serialize};
13
14use crate::error::LiteError;
15use crate::storage::engine::StorageEngine;
16
17const CATALOG_PREFIX: &str = "catalog:";
18const CATALOG_INDEX_KEY: &[u8] = b"catalog_index";
19
20/// Persisted metadata for one array.
21#[derive(
22    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
23)]
24pub struct ArrayCatalogEntry {
25    /// Array name (human-readable, also used as the storage key).
26    pub name: String,
27    /// Serialized schema (zerompk of `ArraySchema`).
28    pub schema_bytes: Vec<u8>,
29    /// FNV-1a hash of the schema used for open-guard mismatch detection.
30    pub schema_hash: u64,
31    /// Optional bitemporal audit retention in milliseconds.
32    pub audit_retain_ms: Option<i64>,
33    /// Minimum allowed audit retention floor.
34    pub minimum_audit_retain_ms: Option<u64>,
35}
36
37impl ArrayCatalogEntry {
38    pub fn schema(&self) -> Result<ArraySchema, LiteError> {
39        zerompk::from_msgpack(&self.schema_bytes).map_err(|e| LiteError::Serialization {
40            detail: format!("decode ArraySchema: {e}"),
41        })
42    }
43}
44
45/// In-memory + persisted catalog for all arrays.
46pub struct ArrayCatalog<S: StorageEngine> {
47    storage: Arc<S>,
48    /// Cached entries — the source of truth after open().
49    entries: HashMap<String, ArrayCatalogEntry>,
50}
51
52impl<S: StorageEngine> ArrayCatalog<S> {
53    fn catalog_key(name: &str) -> Vec<u8> {
54        let mut k = CATALOG_PREFIX.as_bytes().to_vec();
55        k.extend_from_slice(name.as_bytes());
56        k
57    }
58
59    /// Load catalog from storage.
60    pub async fn open(storage: Arc<S>) -> Result<Self, LiteError> {
61        let mut entries = HashMap::new();
62
63        let index_bytes = storage.get(Namespace::Array, CATALOG_INDEX_KEY).await?;
64        let names: Vec<String> = match index_bytes {
65            Some(b) => zerompk::from_msgpack(&b).map_err(|e| LiteError::Serialization {
66                detail: format!("decode catalog index: {e}"),
67            })?,
68            None => Vec::new(),
69        };
70
71        for name in names {
72            let key = Self::catalog_key(&name);
73            if let Some(bytes) = storage.get(Namespace::Array, &key).await? {
74                let entry: ArrayCatalogEntry =
75                    zerompk::from_msgpack(&bytes).map_err(|e| LiteError::Serialization {
76                        detail: format!("decode catalog entry '{name}': {e}"),
77                    })?;
78                entries.insert(name, entry);
79            }
80        }
81
82        Ok(Self { storage, entries })
83    }
84
85    /// Insert a new entry and persist atomically.
86    pub async fn insert(&mut self, entry: ArrayCatalogEntry) -> Result<(), LiteError> {
87        let name = entry.name.clone();
88        let entry_bytes =
89            zerompk::to_msgpack_vec(&entry).map_err(|e| LiteError::Serialization {
90                detail: format!("encode catalog entry: {e}"),
91            })?;
92
93        self.entries.insert(name.clone(), entry);
94
95        let names: Vec<&str> = self.entries.keys().map(|s| s.as_str()).collect();
96        let index_bytes =
97            zerompk::to_msgpack_vec(&names).map_err(|e| LiteError::Serialization {
98                detail: format!("encode catalog index: {e}"),
99            })?;
100
101        let key = Self::catalog_key(&name);
102        self.storage
103            .batch_write(&[
104                crate::storage::engine::WriteOp::Put {
105                    ns: Namespace::Array,
106                    key,
107                    value: entry_bytes,
108                },
109                crate::storage::engine::WriteOp::Put {
110                    ns: Namespace::Array,
111                    key: CATALOG_INDEX_KEY.to_vec(),
112                    value: index_bytes,
113                },
114            ])
115            .await
116    }
117
118    /// Remove an entry from the catalog and persist atomically.
119    pub async fn remove(&mut self, name: &str) -> Result<(), LiteError> {
120        self.entries.remove(name);
121
122        let names: Vec<&str> = self.entries.keys().map(|s| s.as_str()).collect();
123        let index_bytes =
124            zerompk::to_msgpack_vec(&names).map_err(|e| LiteError::Serialization {
125                detail: format!("encode catalog index: {e}"),
126            })?;
127
128        let key = Self::catalog_key(name);
129        self.storage
130            .batch_write(&[
131                crate::storage::engine::WriteOp::Delete {
132                    ns: Namespace::Array,
133                    key,
134                },
135                crate::storage::engine::WriteOp::Put {
136                    ns: Namespace::Array,
137                    key: CATALOG_INDEX_KEY.to_vec(),
138                    value: index_bytes,
139                },
140            ])
141            .await
142    }
143
144    pub fn get(&self, name: &str) -> Option<&ArrayCatalogEntry> {
145        self.entries.get(name)
146    }
147
148    pub fn names(&self) -> impl Iterator<Item = &str> {
149        self.entries.keys().map(|s| s.as_str())
150    }
151}
152
153/// Compute a stable FNV-1a hash for an `ArraySchema`.
154pub fn hash_schema(schema: &ArraySchema) -> Result<u64, LiteError> {
155    let bytes = zerompk::to_msgpack_vec(schema).map_err(|e| LiteError::Serialization {
156        detail: format!("hash_schema encode: {e}"),
157    })?;
158    let mut h: u64 = 14695981039346656037;
159    for b in bytes {
160        h ^= b as u64;
161        h = h.wrapping_mul(1099511628211);
162    }
163    Ok(h)
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169    use crate::storage::pagedb_storage::PagedbStorageMem;
170    use nodedb_array::schema::ArraySchemaBuilder;
171    use nodedb_array::schema::attr_spec::{AttrSpec, AttrType};
172    use nodedb_array::schema::dim_spec::{DimSpec, DimType};
173    use nodedb_array::types::domain::{Domain, DomainBound};
174
175    fn test_schema() -> ArraySchema {
176        ArraySchemaBuilder::new("t")
177            .dim(DimSpec::new(
178                "x",
179                DimType::Int64,
180                Domain::new(DomainBound::Int64(0), DomainBound::Int64(15)),
181            ))
182            .attr(AttrSpec::new("v", AttrType::Int64, true))
183            .tile_extents(vec![4])
184            .build()
185            .unwrap()
186    }
187
188    fn make_entry(schema: &ArraySchema) -> ArrayCatalogEntry {
189        let schema_bytes = zerompk::to_msgpack_vec(schema).unwrap();
190        let schema_hash = hash_schema(schema).unwrap();
191        ArrayCatalogEntry {
192            name: "t".into(),
193            schema_bytes,
194            schema_hash,
195            audit_retain_ms: None,
196            minimum_audit_retain_ms: None,
197        }
198    }
199
200    #[tokio::test]
201    async fn insert_and_get() {
202        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
203        let mut catalog = ArrayCatalog::open(Arc::clone(&storage)).await.unwrap();
204        let schema = test_schema();
205        let entry = make_entry(&schema);
206        catalog.insert(entry).await.unwrap();
207        assert!(catalog.get("t").is_some());
208    }
209
210    #[tokio::test]
211    async fn persists_across_reopen() {
212        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
213        {
214            let mut catalog = ArrayCatalog::open(Arc::clone(&storage)).await.unwrap();
215            catalog.insert(make_entry(&test_schema())).await.unwrap();
216        }
217        let catalog2 = ArrayCatalog::open(Arc::clone(&storage)).await.unwrap();
218        assert!(catalog2.get("t").is_some());
219    }
220
221    #[tokio::test]
222    async fn remove_entry() {
223        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
224        let mut catalog = ArrayCatalog::open(Arc::clone(&storage)).await.unwrap();
225        catalog.insert(make_entry(&test_schema())).await.unwrap();
226        catalog.remove("t").await.unwrap();
227        assert!(catalog.get("t").is_none());
228    }
229
230    #[test]
231    fn hash_schema_stable() {
232        let s = test_schema();
233        let h1 = hash_schema(&s).unwrap();
234        let h2 = hash_schema(&s).unwrap();
235        assert_eq!(h1, h2);
236    }
237}
nodedb_lite/engine/array/catalog.rs

nodedb_lite/engine/array/
catalog.rs
