feat: protobuf schema registry, typed storage, and persistence

Cargo.toml

@@ -61,6 +61,9 @@ rustls = { version = "0.23", default-features = false, features = ["std", "tls12
 rustls-pemfile = "2"
 rustls-native-certs = "0.8"
 
+# dynamic protobuf messages
+prost-reflect = "0.16"
+
 # internal crates (version required for crates.io publishing)
 emberkv-core = { version = "0.4.2", path = "crates/ember-core" }
 ember-protocol = { version = "0.4.2", path = "crates/ember-protocol" }

crates/ember-core/Cargo.toml

@@ -11,6 +11,7 @@ readme = "README.md"
 
 [features]
 encryption = ["ember-persistence/encryption"]
+protobuf = ["prost-reflect", "ember-persistence/protobuf"]
 
 [lib]
 name = "ember_core"
@@ -25,6 +26,7 @@ tokio = { workspace = true }
 tracing = { workspace = true }
 rand = { workspace = true }
 ordered-float = { workspace = true }
+prost-reflect = { workspace = true, optional = true }
 dashmap = "6"
 parking_lot = "0.12"
 

crates/ember-core/src/engine.rs

@@ -24,6 +24,10 @@ pub struct EngineConfig {
     /// Optional persistence configuration. When set, each shard gets
     /// its own AOF and snapshot files under this directory.
     pub persistence: Option<ShardPersistenceConfig>,
+    /// Optional schema registry for protobuf value validation.
+    /// When set, enables PROTO.* commands.
+    #[cfg(feature = "protobuf")]
+    pub schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 }
 
 /// The sharded engine. Owns handles to all shard tasks and routes
@@ -34,6 +38,8 @@ pub struct EngineConfig {
 #[derive(Debug, Clone)]
 pub struct Engine {
     shards: Vec<ShardHandle>,
+    #[cfg(feature = "protobuf")]
+    schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 }
 
 impl Engine {
@@ -69,7 +75,11 @@ impl Engine {
             })
             .collect();
 
-        Self { shards }
+        Self {
+            shards,
+            #[cfg(feature = "protobuf")]
+            schema_registry: config.schema_registry,
+        }
     }
 
     /// Creates an engine with one shard per available CPU core.
@@ -88,6 +98,12 @@ impl Engine {
         Self::with_config(cores, config)
     }
 
+    /// Returns a reference to the schema registry, if protobuf is enabled.
+    #[cfg(feature = "protobuf")]
+    pub fn schema_registry(&self) -> Option<&crate::schema::SharedSchemaRegistry> {
+        self.schema_registry.as_ref()
+    }
+
     /// Returns the number of shards.
     pub fn shard_count(&self) -> usize {
         self.shards.len()

crates/ember-core/src/keyspace.rs

@@ -1842,6 +1842,92 @@ impl Keyspace {
         expired
     }
 
+    // -- protobuf operations --
+
+    /// Stores a protobuf value. No schema validation here — that's the
+    /// server's responsibility. Follows the same pattern as `set()`.
+    #[cfg(feature = "protobuf")]
+    pub fn proto_set(
+        &mut self,
+        key: String,
+        type_name: String,
+        data: Bytes,
+        expire: Option<Duration>,
+    ) -> SetResult {
+        let has_expiry = expire.is_some();
+        let new_value = Value::Proto { type_name, data };
+
+        let new_size = memory::entry_size(&key, &new_value);
+        let old_size = self
+            .entries
+            .get(&key)
+            .map(|e| memory::entry_size(&key, &e.value))
+            .unwrap_or(0);
+        let net_increase = new_size.saturating_sub(old_size);
+
+        if !self.enforce_memory_limit(net_increase) {
+            return SetResult::OutOfMemory;
+        }
+
+        if let Some(old_entry) = self.entries.get(&key) {
+            self.memory.replace(&key, &old_entry.value, &new_value);
+            let had_expiry = old_entry.expires_at_ms != 0;
+            match (had_expiry, has_expiry) {
+                (false, true) => self.expiry_count += 1,
+                (true, false) => self.expiry_count = self.expiry_count.saturating_sub(1),
+                _ => {}
+            }
+        } else {
+            self.memory.add(&key, &new_value);
+            if has_expiry {
+                self.expiry_count += 1;
+            }
+        }
+
+        self.entries.insert(key, Entry::new(new_value, expire));
+        SetResult::Ok
+    }
+
+    /// Retrieves a proto value, returning `(type_name, data)` or `None`.
+    ///
+    /// Returns `Err(WrongType)` if the key holds a different value type.
+    #[cfg(feature = "protobuf")]
+    pub fn proto_get(&mut self, key: &str) -> Result<Option<(String, Bytes)>, WrongType> {
+        if self.remove_if_expired(key) {
+            return Ok(None);
+        }
+        match self.entries.get_mut(key) {
+            Some(e) => {
+                if let Value::Proto { type_name, data } = &e.value {
+                    let result = (type_name.clone(), data.clone());
+                    e.touch();
+                    Ok(Some(result))
+                } else {
+                    Err(WrongType)
+                }
+            }
+            None => Ok(None),
+        }
+    }
+
+    /// Returns the protobuf message type name for a key, or `None` if
+    /// the key doesn't exist.
+    ///
+    /// Returns `Err(WrongType)` if the key holds a non-proto value.
+    #[cfg(feature = "protobuf")]
+    pub fn proto_type(&mut self, key: &str) -> Result<Option<String>, WrongType> {
+        if self.remove_if_expired(key) {
+            return Ok(None);
+        }
+        match self.entries.get(key) {
+            Some(e) => match &e.value {
+                Value::Proto { type_name, .. } => Ok(Some(type_name.clone())),
+                _ => Err(WrongType),
+            },
+            None => Ok(None),
+        }
+    }
+
     /// Sends a value to the background drop thread if one is configured
     /// and the value is large enough to justify the overhead.
     fn defer_drop(&self, value: Value) {

crates/ember-core/src/lib.rs

@@ -15,6 +15,9 @@ pub mod shard;
 pub mod time;
 pub mod types;
 
+#[cfg(feature = "protobuf")]
+pub mod schema;
+
 pub use concurrent::ConcurrentKeyspace;
 pub use engine::{Engine, EngineConfig};
 pub use error::ShardError;

crates/ember-core/src/memory.rs

@@ -160,6 +160,10 @@ pub fn is_large_value(value: &Value) -> bool {
         Value::SortedSet(ss) => ss.len() > LAZY_FREE_THRESHOLD,
         Value::Hash(m) => m.len() > LAZY_FREE_THRESHOLD,
         Value::Set(s) => s.len() > LAZY_FREE_THRESHOLD,
+        // Proto values use Bytes (ref-counted, O(1) drop) + a String.
+        // Neither is expensive to drop.
+        #[cfg(feature = "protobuf")]
+        Value::Proto { .. } => false,
     }
 }
 
@@ -219,6 +223,9 @@ pub fn value_size(value: &Value) -> usize {
             let member_bytes: usize = set.iter().map(|m| m.len() + HASHSET_MEMBER_OVERHEAD).sum();
             HASHSET_BASE_OVERHEAD + member_bytes
         }
+        // type_name String (24 bytes ptr+len+cap on heap) + data Bytes (24 bytes).
+        #[cfg(feature = "protobuf")]
+        Value::Proto { type_name, data } => type_name.len() + data.len() + 24,
     }
 }