feat: concurrent mode proto support and integration tests

crates/ember-server/src/concurrent_handler.rs

@@ -16,6 +16,8 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+#[cfg(feature = "protobuf")]
+use bytes::Bytes;
 use bytes::BytesMut;
 use ember_core::{ConcurrentKeyspace, Engine, TtlResult};
 use ember_protocol::{parse_frame, Command, Frame, SetExpire};
@@ -290,6 +292,172 @@ async fn execute_concurrent(
             }
         },
 
+        // -- protobuf commands --
+        #[cfg(feature = "protobuf")]
+        Command::ProtoRegister { name, descriptor } => {
+            let registry = match _engine.schema_registry() {
+                Some(r) => r,
+                None => return Frame::Error("ERR protobuf support is not enabled".into()),
+            };
+            let result = {
+                let mut reg = match registry.write() {
+                    Ok(r) => r,
+                    Err(_) => return Frame::Error("ERR schema registry lock poisoned".into()),
+                };
+                reg.register(name.clone(), descriptor.clone())
+            };
+            match result {
+                Ok(types) => {
+                    let _ = _engine
+                        .broadcast(|| ember_core::ShardRequest::ProtoRegisterAof {
+                            name: name.clone(),
+                            descriptor: descriptor.clone(),
+                        })
+                        .await;
+                    Frame::Array(
+                        types
+                            .into_iter()
+                            .map(|t| Frame::Bulk(Bytes::from(t)))
+                            .collect(),
+                    )
+                }
+                Err(e) => Frame::Error(format!("ERR {e}")),
+            }
+        }
+
+        #[cfg(feature = "protobuf")]
+        Command::ProtoSet {
+            key,
+            type_name,
+            data,
+            expire,
+            nx,
+            xx,
+        } => {
+            let registry = match _engine.schema_registry() {
+                Some(r) => r,
+                None => return Frame::Error("ERR protobuf support is not enabled".into()),
+            };
+            {
+                let reg = match registry.read() {
+                    Ok(r) => r,
+                    Err(_) => return Frame::Error("ERR schema registry lock poisoned".into()),
+                };
+                if let Err(e) = reg.validate(&type_name, &data) {
+                    return Frame::Error(format!("ERR {e}"));
+                }
+            }
+            let duration = expire.map(|e| match e {
+                SetExpire::Ex(secs) => Duration::from_secs(secs),
+                SetExpire::Px(millis) => Duration::from_millis(millis),
+            });
+            let req = ember_core::ShardRequest::ProtoSet {
+                key: key.clone(),
+                type_name,
+                data,
+                expire: duration,
+                nx,
+                xx,
+            };
+            match _engine.route(&key, req).await {
+                Ok(ember_core::ShardResponse::Ok) => Frame::Simple("OK".into()),
+                Ok(ember_core::ShardResponse::Value(None)) => Frame::Null,
+                Ok(ember_core::ShardResponse::OutOfMemory) => {
+                    Frame::Error("OOM command not allowed when used memory > 'maxmemory'".into())
+                }
+                Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")),
+                Err(e) => Frame::Error(format!("ERR {e}")),
+            }
+        }
+
+        #[cfg(feature = "protobuf")]
+        Command::ProtoGet { key } => {
+            if _engine.schema_registry().is_none() {
+                return Frame::Error("ERR protobuf support is not enabled".into());
+            }
+            let req = ember_core::ShardRequest::ProtoGet { key: key.clone() };
+            match _engine.route(&key, req).await {
+                Ok(ember_core::ShardResponse::ProtoValue(Some((type_name, data)))) => {
+                    Frame::Array(vec![Frame::Bulk(Bytes::from(type_name)), Frame::Bulk(data)])
+                }
+                Ok(ember_core::ShardResponse::ProtoValue(None)) => Frame::Null,
+                Ok(ember_core::ShardResponse::WrongType) => Frame::Error(
+                    "WRONGTYPE Operation against a key holding the wrong kind of value".into(),
+                ),
+                Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")),
+                Err(e) => Frame::Error(format!("ERR {e}")),
+            }
+        }
+
+        #[cfg(feature = "protobuf")]
+        Command::ProtoType { key } => {
+            if _engine.schema_registry().is_none() {
+                return Frame::Error("ERR protobuf support is not enabled".into());
+            }
+            let req = ember_core::ShardRequest::ProtoType { key: key.clone() };
+            match _engine.route(&key, req).await {
+                Ok(ember_core::ShardResponse::ProtoTypeName(Some(name))) => {
+                    Frame::Bulk(Bytes::from(name))
+                }
+                Ok(ember_core::ShardResponse::ProtoTypeName(None)) => Frame::Null,
+                Ok(ember_core::ShardResponse::WrongType) => Frame::Error(
+                    "WRONGTYPE Operation against a key holding the wrong kind of value".into(),
+                ),
+                Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")),
+                Err(e) => Frame::Error(format!("ERR {e}")),
+            }
+        }
+
+        #[cfg(feature = "protobuf")]
+        Command::ProtoSchemas => {
+            let registry = match _engine.schema_registry() {
+                Some(r) => r,
+                None => return Frame::Error("ERR protobuf support is not enabled".into()),
+            };
+            let reg = match registry.read() {
+                Ok(r) => r,
+                Err(_) => return Frame::Error("ERR schema registry lock poisoned".into()),
+            };
+            let names = reg.schema_names();
+            Frame::Array(
+                names
+                    .into_iter()
+                    .map(|n| Frame::Bulk(Bytes::from(n)))
+                    .collect(),
+            )
+        }
+
+        #[cfg(feature = "protobuf")]
+        Command::ProtoDescribe { name } => {
+            let registry = match _engine.schema_registry() {
+                Some(r) => r,
+                None => return Frame::Error("ERR protobuf support is not enabled".into()),
+            };
+            let reg = match registry.read() {
+                Ok(r) => r,
+                Err(_) => return Frame::Error("ERR schema registry lock poisoned".into()),
+            };
+            match reg.describe(&name) {
+                Some(types) => Frame::Array(
+                    types
+                        .into_iter()
+                        .map(|t| Frame::Bulk(Bytes::from(t)))
+                        .collect(),
+                ),
+                None => Frame::Error(format!("ERR unknown schema '{name}'")),
+            }
+        }
+
+        #[cfg(not(feature = "protobuf"))]
+        Command::ProtoRegister { .. }
+        | Command::ProtoSet { .. }
+        | Command::ProtoGet { .. }
+        | Command::ProtoType { .. }
+        | Command::ProtoSchemas
+        | Command::ProtoDescribe { .. } => {
+            Frame::Error("ERR unknown command (protobuf support not compiled)".into())
+        }
+
         Command::Quit => Frame::Simple("OK".into()),
 
         // For unsupported commands, return an error

tests/integration/Cargo.toml

@@ -14,4 +14,5 @@ harness = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "io-util", "time"] }
 bytes = { workspace = true }
 ember-protocol = { workspace = true }
+prost-reflect = { workspace = true }
 tempfile = "3"

tests/integration/src/helpers.rs

@@ -31,6 +31,10 @@ pub struct ServerOptions {
     pub cluster_enabled: bool,
     /// Bootstrap as a single-node cluster owning all 16384 slots.
     pub cluster_bootstrap: bool,
+    /// Enable protobuf value storage.
+    pub protobuf: bool,
+    /// Use concurrent (DashMap) mode instead of sharded channels.
+    pub concurrent: bool,
 }
 
 impl TestServer {
@@ -58,6 +62,14 @@ impl TestServer {
             cmd.arg("--requirepass").arg(pass);
         }
 
+        if opts.protobuf {
+            cmd.arg("--protobuf");
+        }
+
+        if opts.concurrent {
+            cmd.arg("--concurrent");
+        }
+
         if opts.cluster_enabled {
             cmd.arg("--cluster-enabled");
             // use a small offset so the gossip port stays in valid u16 range.
@@ -144,6 +156,36 @@ impl TestClient {
         }
     }
 
+    /// Sends a command with raw byte arguments and returns the parsed response.
+    /// Useful for binary data like protobuf descriptors.
+    pub async fn cmd_raw(&mut self, args: &[&[u8]]) -> Frame {
+        let parts: Vec<Frame> = args
+            .iter()
+            .map(|a| Frame::Bulk(Bytes::copy_from_slice(a)))
+            .collect();
+        let frame = Frame::Array(parts);
+
+        let mut out = BytesMut::new();
+        frame.serialize(&mut out);
+        self.stream.write_all(&out).await.unwrap();
+
+        loop {
+            match parse_frame(&self.buf) {
+                Ok(Some((frame, consumed))) => {
+                    let _ = self.buf.split_to(consumed);
+                    return frame;
+                }
+                Ok(None) => {
+                    let n = self.stream.read_buf(&mut self.buf).await.unwrap();
+                    if n == 0 {
+                        panic!("server closed connection while waiting for response");
+                    }
+                }
+                Err(e) => panic!("protocol error: {e}"),
+            }
+        }
+    }
+
     /// Sends a command and returns the parsed response frame.
     pub async fn cmd(&mut self, args: &[&str]) -> Frame {
         // build RESP3 array

tests/integration/src/main.rs

@@ -6,4 +6,5 @@ mod cli;
 mod cluster;
 mod data_types;
 mod persistence;
+mod proto;
 mod pubsub;