diff --git a/README.md b/README.md index 69cb65d7..181ec8d9 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ a low-latency, memory-efficient, distributed cache written in Rust. designed to - **key commands** — DEL, EXISTS, EXPIRE, TTL, PEXPIRE, PTTL, PERSIST, TYPE, SCAN, KEYS, RENAME - **server commands** — PING, ECHO, INFO, DBSIZE, FLUSHDB, BGSAVE, BGREWRITEAOF, AUTH, QUIT - **pub/sub** — SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, PUBLISH, plus PUBSUB introspection +- **protobuf storage** — schema-validated protobuf values with field-level access (compile with `--features protobuf`) - **authentication** — `--requirepass` for redis-compatible AUTH (legacy and username/password forms) - **tls support** — redis-compatible TLS on a separate port, with optional mTLS for client certificates - **protected mode** — rejects non-loopback connections when no password is set on public binds @@ -118,6 +119,31 @@ DBSIZE # => (integer) 6 ember-cli -p 6380 --tls --tls-insecure PING ``` +## protobuf storage + +ember can store schema-validated protobuf messages and access individual fields server-side. compile with `--features protobuf` to enable. + +```bash +# build with protobuf support +cargo build --release --features protobuf +``` + +**commands**: + +| command | description | +|---------|-------------| +| `PROTO.REGISTER name ` | register a compiled FileDescriptorSet | +| `PROTO.SET key type_name [EX s] [PX ms] [NX\|XX]` | store a validated protobuf value | +| `PROTO.GET key` | retrieve the full encoded message | +| `PROTO.TYPE key` | return the message type name | +| `PROTO.SCHEMAS` | list all registered schema names | +| `PROTO.DESCRIBE name` | list message types in a schema | +| `PROTO.GETFIELD key field_path` | read a single field (dot-separated nested paths) | +| `PROTO.SETFIELD key field_path value` | update a single scalar field | +| `PROTO.DELFIELD key field_path` | clear a field to its default value | + +field-level operations decode/mutate/re-encode on the server, so clients don't need protobuf libraries for simple reads and writes. nested paths use dot notation (e.g., `address.city`). complex types (repeated, map, nested messages) require `PROTO.GET`/`PROTO.SET` for full replacement. + ## configuration | flag | default | description | @@ -241,7 +267,7 @@ contributions welcome — see [CONTRIBUTING.md](CONTRIBUTING.md). | 4 | clustering (raft, gossip, slots, migration) | ✅ complete | | 5 | developer experience (observability, CLI, clients) | 🚧 in progress | -**current**: 85 commands, 906 tests, ~18k lines of code (excluding tests) +**current**: 94 commands, 967 tests, ~18k lines of code (excluding tests) ## security diff --git a/crates/ember-core/src/schema.rs b/crates/ember-core/src/schema.rs index f1891ef6..b4f34c6b 100644 --- a/crates/ember-core/src/schema.rs +++ b/crates/ember-core/src/schema.rs @@ -194,6 +194,61 @@ impl SchemaRegistry { value_to_frame(&value, &field_desc) } + /// Updates a single scalar field in an encoded protobuf message. + /// + /// Decodes the message, walks the dot-separated `field_path`, parses + /// `raw_value` according to the field's type descriptor, sets the field, + /// and re-encodes the message. Returns the new encoded bytes. + /// + /// Only supports scalar fields — repeated, map, and message fields + /// return an error directing clients to use `PROTO.SET`. + pub fn set_field( + &self, + type_name: &str, + data: &[u8], + field_path: &str, + raw_value: &str, + ) -> Result { + let descriptor = self.find_message(type_name)?; + let mut msg = DynamicMessage::decode(descriptor, data) + .map_err(|e| SchemaError::ValidationFailed(e.to_string()))?; + + let (parent, leaf_name, leaf_desc) = resolve_field_path_mut(&mut msg, field_path)?; + let parsed = parse_field_value(raw_value, &leaf_desc)?; + parent.set_field_by_name(&leaf_name, parsed); + + let mut buf = Vec::new(); + use prost_reflect::prost::Message; + msg.encode(&mut buf) + .map_err(|e| SchemaError::ValidationFailed(format!("re-encode failed: {e}")))?; + Ok(Bytes::from(buf)) + } + + /// Clears a single field in an encoded protobuf message, resetting it + /// to its default value. + /// + /// Decodes the message, walks the dot-separated `field_path`, clears the + /// leaf field, and re-encodes. Returns the new encoded bytes. + pub fn clear_field( + &self, + type_name: &str, + data: &[u8], + field_path: &str, + ) -> Result { + let descriptor = self.find_message(type_name)?; + let mut msg = DynamicMessage::decode(descriptor, data) + .map_err(|e| SchemaError::ValidationFailed(e.to_string()))?; + + let (parent, leaf_name, _leaf_desc) = resolve_field_path_mut(&mut msg, field_path)?; + parent.clear_field_by_name(&leaf_name); + + let mut buf = Vec::new(); + use prost_reflect::prost::Message; + msg.encode(&mut buf) + .map_err(|e| SchemaError::ValidationFailed(format!("re-encode failed: {e}")))?; + Ok(Bytes::from(buf)) + } + /// Looks up a message descriptor by full name across all schemas. fn find_message(&self, message_type: &str) -> Result { for schema in self.schemas.values() { @@ -254,7 +309,11 @@ fn resolve_field_path( } } - unreachable!("loop always returns at the leaf segment") + // the loop always returns at the leaf segment, but if the segments vec + // were somehow empty after validation, return a clear error instead of panicking + Err(SchemaError::FieldNotFound( + "failed to resolve field path".into(), + )) } /// Converts a `prost_reflect::Value` + its field descriptor into a RESP3 frame. @@ -278,7 +337,13 @@ fn value_to_frame( prost_reflect::Value::I32(n) => Ok(Frame::Integer(i64::from(*n))), prost_reflect::Value::I64(n) => Ok(Frame::Integer(*n)), prost_reflect::Value::U32(n) => Ok(Frame::Integer(i64::from(*n))), - prost_reflect::Value::U64(n) => Ok(Frame::Integer(*n as i64)), + prost_reflect::Value::U64(n) => { + // RESP3 integers are signed 64-bit; large u64 values would wrap + match i64::try_from(*n) { + Ok(i) => Ok(Frame::Integer(i)), + Err(_) => Ok(Frame::Bulk(Bytes::from(n.to_string()))), + } + } prost_reflect::Value::F32(n) => Ok(Frame::Bulk(Bytes::from(format!("{n}")))), prost_reflect::Value::F64(n) => Ok(Frame::Bulk(Bytes::from(format!("{n}")))), prost_reflect::Value::Bool(b) => Ok(Frame::Integer(if *b { 1 } else { 0 })), @@ -304,6 +369,167 @@ fn value_to_frame( } } +/// Walks a dot-separated field path through a mutable `DynamicMessage`, +/// returning the parent message (mutably), the leaf field name, and its +/// descriptor. Used by `set_field` and `clear_field`. +/// +/// For a simple path like `"name"`, returns `(msg, "name", desc)`. +/// For a nested path like `"address.city"`, drills into the `address` +/// message field and returns `(address_msg, "city", city_desc)`. +fn resolve_field_path_mut<'a>( + msg: &'a mut DynamicMessage, + path: &str, +) -> Result<(&'a mut DynamicMessage, String, FieldDescriptor), SchemaError> { + if path.is_empty() { + return Err(SchemaError::FieldNotFound("empty field path".into())); + } + + let segments: Vec<&str> = path.split('.').collect(); + for seg in &segments { + if seg.is_empty() { + return Err(SchemaError::FieldNotFound(format!( + "invalid field path '{path}': empty segment" + ))); + } + } + + // for a single segment, just verify the field exists and return + if segments.len() == 1 { + let field_desc = msg + .descriptor() + .get_field_by_name(segments[0]) + .ok_or_else(|| SchemaError::FieldNotFound(segments[0].to_string()))?; + return Ok((msg, segments[0].to_string(), field_desc)); + } + + // walk intermediate segments mutably, stopping before the leaf + let mut current = msg; + for segment in &segments[..segments.len() - 1] { + let field_desc = current + .descriptor() + .get_field_by_name(segment) + .ok_or_else(|| SchemaError::FieldNotFound(segment.to_string()))?; + + if !matches!(field_desc.kind(), Kind::Message(_)) { + return Err(SchemaError::FieldNotFound(format!( + "'{segment}' is not a message field, cannot traverse further" + ))); + } + + // ensure the nested message exists (get or init default) + if !current.has_field_by_name(segment) { + let Kind::Message(nested_desc) = field_desc.kind() else { + return Err(SchemaError::FieldNotFound(format!( + "'{segment}' is not a message field" + ))); + }; + current.set_field_by_name( + segment, + prost_reflect::Value::Message(DynamicMessage::new(nested_desc)), + ); + } + + // get mutable reference to the nested message + let val = current.get_field_by_name_mut(segment).ok_or_else(|| { + SchemaError::FieldNotFound(format!("failed to get mutable reference to '{segment}'")) + })?; + current = match val { + prost_reflect::Value::Message(ref mut nested) => nested, + _ => { + return Err(SchemaError::FieldNotFound(format!( + "'{segment}' is not a message field" + ))); + } + }; + } + + let leaf = segments + .last() + .ok_or_else(|| SchemaError::FieldNotFound("failed to resolve field path".into()))?; + let leaf_desc = current + .descriptor() + .get_field_by_name(leaf) + .ok_or_else(|| SchemaError::FieldNotFound(leaf.to_string()))?; + + Ok((current, leaf.to_string(), leaf_desc)) +} + +/// Parses a raw string value into a `prost_reflect::Value` based on the +/// field descriptor's type. Only supports scalar types. +fn parse_field_value( + raw: &str, + field_desc: &FieldDescriptor, +) -> Result { + if field_desc.is_list() || field_desc.is_map() { + return Err(SchemaError::ValidationFailed( + "use PROTO.SET for repeated/map fields".into(), + )); + } + + match field_desc.kind() { + Kind::String => Ok(prost_reflect::Value::String(raw.to_owned())), + Kind::Bytes => Ok(prost_reflect::Value::Bytes(Bytes::from(raw.to_owned()))), + Kind::Bool => match raw { + "true" | "1" => Ok(prost_reflect::Value::Bool(true)), + "false" | "0" => Ok(prost_reflect::Value::Bool(false)), + _ => Err(SchemaError::ValidationFailed(format!( + "invalid bool value: '{raw}' (expected true/false/1/0)" + ))), + }, + Kind::Int32 | Kind::Sint32 | Kind::Sfixed32 => { + let n: i32 = raw + .parse() + .map_err(|e| SchemaError::ValidationFailed(format!("invalid int32 value: {e}")))?; + Ok(prost_reflect::Value::I32(n)) + } + Kind::Int64 | Kind::Sint64 | Kind::Sfixed64 => { + let n: i64 = raw + .parse() + .map_err(|e| SchemaError::ValidationFailed(format!("invalid int64 value: {e}")))?; + Ok(prost_reflect::Value::I64(n)) + } + Kind::Uint32 | Kind::Fixed32 => { + let n: u32 = raw + .parse() + .map_err(|e| SchemaError::ValidationFailed(format!("invalid uint32 value: {e}")))?; + Ok(prost_reflect::Value::U32(n)) + } + Kind::Uint64 | Kind::Fixed64 => { + let n: u64 = raw + .parse() + .map_err(|e| SchemaError::ValidationFailed(format!("invalid uint64 value: {e}")))?; + Ok(prost_reflect::Value::U64(n)) + } + Kind::Float => { + let n: f32 = raw + .parse() + .map_err(|e| SchemaError::ValidationFailed(format!("invalid float value: {e}")))?; + Ok(prost_reflect::Value::F32(n)) + } + Kind::Double => { + let n: f64 = raw + .parse() + .map_err(|e| SchemaError::ValidationFailed(format!("invalid double value: {e}")))?; + Ok(prost_reflect::Value::F64(n)) + } + Kind::Enum(enum_desc) => { + // try name lookup first, then parse as number + if let Some(val) = enum_desc.get_value_by_name(raw) { + return Ok(prost_reflect::Value::EnumNumber(val.number())); + } + let n: i32 = raw.parse().map_err(|_| { + SchemaError::ValidationFailed(format!( + "invalid enum value: '{raw}' (not a valid name or number)" + )) + })?; + Ok(prost_reflect::Value::EnumNumber(n)) + } + Kind::Message(_) => Err(SchemaError::ValidationFailed( + "use PROTO.SET for nested message fields".into(), + )), + } +} + impl std::fmt::Debug for SchemaRegistry { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("SchemaRegistry") @@ -683,4 +909,172 @@ mod tests { let err = registry.get_field("test.User", &data, "").unwrap_err(); assert!(matches!(err, SchemaError::FieldNotFound(_))); } + + // --- set_field / clear_field tests --- + + /// Builds a descriptor with string, int32, and bool fields for mutation testing. + fn make_multi_field_descriptor() -> Bytes { + use prost_reflect::prost_types::{ + DescriptorProto, FieldDescriptorProto, FileDescriptorProto, FileDescriptorSet, + }; + + let fds = FileDescriptorSet { + file: vec![FileDescriptorProto { + name: Some("test.proto".into()), + package: Some("test".into()), + message_type: vec![DescriptorProto { + name: Some("Profile".into()), + field: vec![ + FieldDescriptorProto { + name: Some("name".into()), + number: Some(1), + r#type: Some(9), // TYPE_STRING + label: Some(1), + ..Default::default() + }, + FieldDescriptorProto { + name: Some("age".into()), + number: Some(2), + r#type: Some(5), // TYPE_INT32 + label: Some(1), + ..Default::default() + }, + FieldDescriptorProto { + name: Some("active".into()), + number: Some(3), + r#type: Some(8), // TYPE_BOOL + label: Some(1), + ..Default::default() + }, + ], + ..Default::default() + }], + ..Default::default() + }], + }; + let mut buf = Vec::new(); + use prost_reflect::prost::Message; + fds.encode(&mut buf).unwrap(); + Bytes::from(buf) + } + + /// Helper: encode a test.Profile message with initial values. + fn encode_profile(registry: &SchemaRegistry, name: &str, age: i32, active: bool) -> Vec { + let pool = ®istry.schemas["profiles"].pool; + let msg_desc = pool.get_message_by_name("test.Profile").unwrap(); + let mut msg = DynamicMessage::new(msg_desc); + msg.set_field_by_name("name", prost_reflect::Value::String(name.into())); + msg.set_field_by_name("age", prost_reflect::Value::I32(age)); + msg.set_field_by_name("active", prost_reflect::Value::Bool(active)); + let mut buf = Vec::new(); + use prost_reflect::prost::Message; + msg.encode(&mut buf).unwrap(); + buf + } + + #[test] + fn set_field_string() { + let desc = make_multi_field_descriptor(); + let mut registry = SchemaRegistry::new(); + registry.register("profiles".into(), desc).unwrap(); + + let data = encode_profile(®istry, "alice", 25, true); + let new_data = registry + .set_field("test.Profile", &data, "name", "bob") + .unwrap(); + + // verify the field was updated + let frame = registry + .get_field("test.Profile", &new_data, "name") + .unwrap(); + assert_eq!(frame, Frame::Bulk(Bytes::from("bob"))); + + // verify other fields are preserved + let frame = registry + .get_field("test.Profile", &new_data, "age") + .unwrap(); + assert_eq!(frame, Frame::Integer(25)); + } + + #[test] + fn set_field_int32() { + let desc = make_multi_field_descriptor(); + let mut registry = SchemaRegistry::new(); + registry.register("profiles".into(), desc).unwrap(); + + let data = encode_profile(®istry, "alice", 25, true); + let new_data = registry + .set_field("test.Profile", &data, "age", "30") + .unwrap(); + + let frame = registry + .get_field("test.Profile", &new_data, "age") + .unwrap(); + assert_eq!(frame, Frame::Integer(30)); + } + + #[test] + fn set_field_bool() { + let desc = make_multi_field_descriptor(); + let mut registry = SchemaRegistry::new(); + registry.register("profiles".into(), desc).unwrap(); + + let data = encode_profile(®istry, "alice", 25, true); + let new_data = registry + .set_field("test.Profile", &data, "active", "false") + .unwrap(); + + let frame = registry + .get_field("test.Profile", &new_data, "active") + .unwrap(); + assert_eq!(frame, Frame::Integer(0)); + } + + #[test] + fn set_field_invalid_int_value() { + let desc = make_multi_field_descriptor(); + let mut registry = SchemaRegistry::new(); + registry.register("profiles".into(), desc).unwrap(); + + let data = encode_profile(®istry, "alice", 25, true); + let err = registry + .set_field("test.Profile", &data, "age", "not_a_number") + .unwrap_err(); + assert!(matches!(err, SchemaError::ValidationFailed(_))); + } + + #[test] + fn set_field_nonexistent() { + let desc = make_multi_field_descriptor(); + let mut registry = SchemaRegistry::new(); + registry.register("profiles".into(), desc).unwrap(); + + let data = encode_profile(®istry, "alice", 25, true); + let err = registry + .set_field("test.Profile", &data, "nonexistent", "value") + .unwrap_err(); + assert!(matches!(err, SchemaError::FieldNotFound(_))); + } + + #[test] + fn clear_field_resets_to_default() { + let desc = make_multi_field_descriptor(); + let mut registry = SchemaRegistry::new(); + registry.register("profiles".into(), desc).unwrap(); + + let data = encode_profile(®istry, "alice", 25, true); + let new_data = registry.clear_field("test.Profile", &data, "name").unwrap(); + + // string default is empty + let frame = registry + .get_field("test.Profile", &new_data, "name") + .unwrap(); + assert_eq!(frame, Frame::Bulk(Bytes::from(""))); + + // other fields preserved + let frame = registry + .get_field("test.Profile", &new_data, "age") + .unwrap(); + assert_eq!(frame, Frame::Integer(25)); + } } diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index 1b873842..925ada21 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -355,6 +355,17 @@ pub enum Command { /// protobuf value, returning it as a native RESP3 type. ProtoGetField { key: String, field_path: String }, + /// PROTO.SETFIELD `key` `field_path` `value`. Updates a single scalar + /// field in a stored protobuf value. + ProtoSetField { + key: String, + field_path: String, + value: String, + }, + + /// PROTO.DELFIELD `key` `field_path`. Clears a field to its default value. + ProtoDelField { key: String, field_path: String }, + /// AUTH \[username\] password. Authenticate the connection. Auth { /// Username for ACL-style auth. None for legacy AUTH. @@ -488,6 +499,8 @@ impl Command { Command::ProtoSchemas => "proto.schemas", Command::ProtoDescribe { .. } => "proto.describe", Command::ProtoGetField { .. } => "proto.getfield", + Command::ProtoSetField { .. } => "proto.setfield", + Command::ProtoDelField { .. } => "proto.delfield", Command::Auth { .. } => "auth", Command::Quit => "quit", Command::Unknown(_) => "unknown", @@ -592,6 +605,8 @@ impl Command { "PROTO.SCHEMAS" => parse_proto_schemas(&frames[1..]), "PROTO.DESCRIBE" => parse_proto_describe(&frames[1..]), "PROTO.GETFIELD" => parse_proto_getfield(&frames[1..]), + "PROTO.SETFIELD" => parse_proto_setfield(&frames[1..]), + "PROTO.DELFIELD" => parse_proto_delfield(&frames[1..]), "AUTH" => parse_auth(&frames[1..]), "QUIT" => parse_quit(&frames[1..]), _ => Ok(Command::Unknown(name)), @@ -1870,6 +1885,29 @@ fn parse_proto_getfield(args: &[Frame]) -> Result { Ok(Command::ProtoGetField { key, field_path }) } +fn parse_proto_setfield(args: &[Frame]) -> Result { + if args.len() != 3 { + return Err(ProtocolError::WrongArity("PROTO.SETFIELD".into())); + } + let key = extract_string(&args[0])?; + let field_path = extract_string(&args[1])?; + let value = extract_string(&args[2])?; + Ok(Command::ProtoSetField { + key, + field_path, + value, + }) +} + +fn parse_proto_delfield(args: &[Frame]) -> Result { + if args.len() != 2 { + return Err(ProtocolError::WrongArity("PROTO.DELFIELD".into())); + } + let key = extract_string(&args[0])?; + let field_path = extract_string(&args[1])?; + Ok(Command::ProtoDelField { key, field_path }) +} + fn parse_auth(args: &[Frame]) -> Result { match args.len() { 1 => { @@ -4290,4 +4328,60 @@ mod tests { Command::from_frame(cmd(&["PROTO.GETFIELD", "key", "field", "extra"])).unwrap_err(); assert!(matches!(err, ProtocolError::WrongArity(_))); } + + // --- proto.setfield --- + + #[test] + fn proto_setfield_basic() { + assert_eq!( + Command::from_frame(cmd(&["PROTO.SETFIELD", "user:1", "name", "bob"])).unwrap(), + Command::ProtoSetField { + key: "user:1".into(), + field_path: "name".into(), + value: "bob".into(), + }, + ); + } + + #[test] + fn proto_setfield_wrong_arity() { + let err = Command::from_frame(cmd(&["PROTO.SETFIELD"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + + let err = Command::from_frame(cmd(&["PROTO.SETFIELD", "key"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + + let err = Command::from_frame(cmd(&["PROTO.SETFIELD", "key", "field"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + + let err = Command::from_frame(cmd(&["PROTO.SETFIELD", "key", "field", "value", "extra"])) + .unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + // --- proto.delfield --- + + #[test] + fn proto_delfield_basic() { + assert_eq!( + Command::from_frame(cmd(&["PROTO.DELFIELD", "user:1", "name"])).unwrap(), + Command::ProtoDelField { + key: "user:1".into(), + field_path: "name".into(), + }, + ); + } + + #[test] + fn proto_delfield_wrong_arity() { + let err = Command::from_frame(cmd(&["PROTO.DELFIELD"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + + let err = Command::from_frame(cmd(&["PROTO.DELFIELD", "key"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + + let err = + Command::from_frame(cmd(&["PROTO.DELFIELD", "key", "field", "extra"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } } diff --git a/crates/ember-server/src/concurrent_handler.rs b/crates/ember-server/src/concurrent_handler.rs index 84f67971..aef45c9d 100644 --- a/crates/ember-server/src/concurrent_handler.rs +++ b/crates/ember-server/src/concurrent_handler.rs @@ -475,6 +475,108 @@ async fn execute_concurrent( } } + #[cfg(feature = "protobuf")] + Command::ProtoSetField { + key, + field_path, + value, + } => { + let registry = match _engine.schema_registry() { + Some(r) => r, + None => return Frame::Error("ERR protobuf support is not enabled".into()), + }; + let req = ember_core::ShardRequest::ProtoGet { key: key.clone() }; + let (type_name, data) = match _engine.route(&key, req).await { + Ok(ember_core::ShardResponse::ProtoValue(Some(pair))) => pair, + Ok(ember_core::ShardResponse::ProtoValue(None)) => return Frame::Null, + Ok(ember_core::ShardResponse::WrongType) => { + return Frame::Error( + "WRONGTYPE Operation against a key holding the wrong kind of value".into(), + ) + } + Ok(other) => { + return Frame::Error(format!("ERR unexpected shard response: {other:?}")) + } + Err(e) => return Frame::Error(format!("ERR {e}")), + }; + let new_data = { + let reg = match registry.read() { + Ok(r) => r, + Err(_) => return Frame::Error("ERR schema registry lock poisoned".into()), + }; + match reg.set_field(&type_name, &data, &field_path, &value) { + Ok(d) => d, + Err(e) => return Frame::Error(format!("ERR {e}")), + } + }; + let req = ember_core::ShardRequest::ProtoSet { + key: key.clone(), + type_name, + data: new_data, + expire: None, + nx: false, + xx: true, + }; + 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::ProtoDelField { key, field_path } => { + let registry = match _engine.schema_registry() { + Some(r) => r, + None => return Frame::Error("ERR protobuf support is not enabled".into()), + }; + let req = ember_core::ShardRequest::ProtoGet { key: key.clone() }; + let (type_name, data) = match _engine.route(&key, req).await { + Ok(ember_core::ShardResponse::ProtoValue(Some(pair))) => pair, + Ok(ember_core::ShardResponse::ProtoValue(None)) => return Frame::Null, + Ok(ember_core::ShardResponse::WrongType) => { + return Frame::Error( + "WRONGTYPE Operation against a key holding the wrong kind of value".into(), + ) + } + Ok(other) => { + return Frame::Error(format!("ERR unexpected shard response: {other:?}")) + } + Err(e) => return Frame::Error(format!("ERR {e}")), + }; + let new_data = { + let reg = match registry.read() { + Ok(r) => r, + Err(_) => return Frame::Error("ERR schema registry lock poisoned".into()), + }; + match reg.clear_field(&type_name, &data, &field_path) { + Ok(d) => d, + Err(e) => return Frame::Error(format!("ERR {e}")), + } + }; + let req = ember_core::ShardRequest::ProtoSet { + key: key.clone(), + type_name, + data: new_data, + expire: None, + nx: false, + xx: true, + }; + match _engine.route(&key, req).await { + Ok(ember_core::ShardResponse::Ok) => Frame::Integer(1), + 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(not(feature = "protobuf"))] Command::ProtoRegister { .. } | Command::ProtoSet { .. } @@ -482,7 +584,9 @@ async fn execute_concurrent( | Command::ProtoType { .. } | Command::ProtoSchemas | Command::ProtoDescribe { .. } - | Command::ProtoGetField { .. } => { + | Command::ProtoGetField { .. } + | Command::ProtoSetField { .. } + | Command::ProtoDelField { .. } => { Frame::Error("ERR unknown command (protobuf support not compiled)".into()) } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index 95859822..50a8b717 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -558,7 +558,9 @@ async fn cluster_slot_check(ctx: &ServerContext, cmd: &Command) -> Option | Command::ProtoSet { ref key, .. } | Command::ProtoGet { ref key } | Command::ProtoType { ref key } - | Command::ProtoGetField { ref key, .. } => cluster.check_slot(key.as_bytes()).await, + | Command::ProtoGetField { ref key, .. } + | Command::ProtoSetField { ref key, .. } + | Command::ProtoDelField { ref key, .. } => cluster.check_slot(key.as_bytes()).await, // multi-key commands — crossslot validation + slot ownership Command::Del { ref keys } @@ -1800,6 +1802,102 @@ async fn execute( } } + #[cfg(feature = "protobuf")] + Command::ProtoSetField { + key, + field_path, + value, + } => { + let registry = match engine.schema_registry() { + Some(r) => r, + None => return Frame::Error("ERR protobuf support is not enabled".into()), + }; + // step 1: fetch current value + let req = ShardRequest::ProtoGet { key: key.clone() }; + let (type_name, data) = match engine.route(&key, req).await { + Ok(ShardResponse::ProtoValue(Some(pair))) => pair, + Ok(ShardResponse::ProtoValue(None)) => return Frame::Null, + Ok(ShardResponse::WrongType) => return wrongtype_error(), + Ok(other) => { + return Frame::Error(format!("ERR unexpected shard response: {other:?}")) + } + Err(e) => return Frame::Error(format!("ERR {e}")), + }; + // step 2: decode, mutate, re-encode + let new_data = { + let reg = match registry.read() { + Ok(r) => r, + Err(_) => return Frame::Error("ERR schema registry lock poisoned".into()), + }; + match reg.set_field(&type_name, &data, &field_path, &value) { + Ok(d) => d, + Err(e) => return Frame::Error(format!("ERR {e}")), + } + }; + // step 3: store back (XX = only if key still exists) + let req = ShardRequest::ProtoSet { + key: key.clone(), + type_name, + data: new_data, + expire: None, + nx: false, + xx: true, + }; + match engine.route(&key, req).await { + Ok(ShardResponse::Ok) => Frame::Simple("OK".into()), + Ok(ShardResponse::Value(None)) => Frame::Null, + Ok(ShardResponse::OutOfMemory) => oom_error(), + Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + + #[cfg(feature = "protobuf")] + Command::ProtoDelField { key, field_path } => { + let registry = match engine.schema_registry() { + Some(r) => r, + None => return Frame::Error("ERR protobuf support is not enabled".into()), + }; + // step 1: fetch current value + let req = ShardRequest::ProtoGet { key: key.clone() }; + let (type_name, data) = match engine.route(&key, req).await { + Ok(ShardResponse::ProtoValue(Some(pair))) => pair, + Ok(ShardResponse::ProtoValue(None)) => return Frame::Null, + Ok(ShardResponse::WrongType) => return wrongtype_error(), + Ok(other) => { + return Frame::Error(format!("ERR unexpected shard response: {other:?}")) + } + Err(e) => return Frame::Error(format!("ERR {e}")), + }; + // step 2: decode, clear field, re-encode + let new_data = { + let reg = match registry.read() { + Ok(r) => r, + Err(_) => return Frame::Error("ERR schema registry lock poisoned".into()), + }; + match reg.clear_field(&type_name, &data, &field_path) { + Ok(d) => d, + Err(e) => return Frame::Error(format!("ERR {e}")), + } + }; + // step 3: store back (XX = only if key still exists) + let req = ShardRequest::ProtoSet { + key: key.clone(), + type_name, + data: new_data, + expire: None, + nx: false, + xx: true, + }; + match engine.route(&key, req).await { + Ok(ShardResponse::Ok) => Frame::Integer(1), + Ok(ShardResponse::Value(None)) => Frame::Null, + Ok(ShardResponse::OutOfMemory) => oom_error(), + Ok(other) => Frame::Error(format!("ERR unexpected shard response: {other:?}")), + Err(e) => Frame::Error(format!("ERR {e}")), + } + } + // when protobuf feature is disabled, proto commands are unknown #[cfg(not(feature = "protobuf"))] Command::ProtoRegister { .. } @@ -1808,7 +1906,9 @@ async fn execute( | Command::ProtoType { .. } | Command::ProtoSchemas | Command::ProtoDescribe { .. } - | Command::ProtoGetField { .. } => { + | Command::ProtoGetField { .. } + | Command::ProtoSetField { .. } + | Command::ProtoDelField { .. } => { Frame::Error("ERR unknown command (protobuf support not compiled)".into()) } diff --git a/tests/integration/src/proto.rs b/tests/integration/src/proto.rs index 9fb2b274..f66653af 100644 --- a/tests/integration/src/proto.rs +++ b/tests/integration/src/proto.rs @@ -49,6 +49,62 @@ fn encode_message(descriptor_bytes: &[u8], type_name: &str, field: &str, value: buf } +/// Builds a FileDescriptorSet with string, int32, and bool fields. +fn make_multi_field_descriptor() -> Vec { + let fds = FileDescriptorSet { + file: vec![FileDescriptorProto { + name: Some("test.proto".into()), + package: Some("test".into()), + message_type: vec![DescriptorProto { + name: Some("Profile".into()), + field: vec![ + FieldDescriptorProto { + name: Some("name".into()), + number: Some(1), + r#type: Some(9), // TYPE_STRING + label: Some(1), + ..Default::default() + }, + FieldDescriptorProto { + name: Some("age".into()), + number: Some(2), + r#type: Some(5), // TYPE_INT32 + label: Some(1), + ..Default::default() + }, + FieldDescriptorProto { + name: Some("active".into()), + number: Some(3), + r#type: Some(8), // TYPE_BOOL + label: Some(1), + ..Default::default() + }, + ], + ..Default::default() + }], + ..Default::default() + }], + }; + let mut buf = Vec::new(); + fds.encode(&mut buf).expect("encode descriptor"); + buf +} + +/// Encodes a Profile message with name, age, and active fields. +fn encode_profile(descriptor_bytes: &[u8], name: &str, age: i32, active: bool) -> Vec { + let pool = DescriptorPool::decode(descriptor_bytes).expect("decode pool"); + let msg_desc = pool + .get_message_by_name("test.Profile") + .expect("find message"); + let mut msg = DynamicMessage::new(msg_desc); + msg.set_field_by_name("name", prost_reflect::Value::String(name.into())); + msg.set_field_by_name("age", prost_reflect::Value::I32(age)); + msg.set_field_by_name("active", prost_reflect::Value::Bool(active)); + let mut buf = Vec::new(); + msg.encode(&mut buf).expect("encode message"); + buf +} + fn start_proto_server(concurrent: bool) -> TestServer { TestServer::start_with(ServerOptions { protobuf: true, @@ -379,6 +435,179 @@ async fn getfield_default_value() { assert_eq!(resp, Frame::Bulk(Bytes::from(""))); } +// ---- PROTO.SETFIELD sharded tests ---- + +#[tokio::test] +async fn setfield_string() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + let resp = c.cmd(&["PROTO.SETFIELD", "p:1", "name", "bob"]).await; + assert!(matches!(resp, Frame::Simple(ref s) if s == "OK")); + + let resp = c.cmd(&["PROTO.GETFIELD", "p:1", "name"]).await; + assert_eq!(resp, Frame::Bulk(Bytes::from("bob"))); +} + +#[tokio::test] +async fn setfield_integer() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + let resp = c.cmd(&["PROTO.SETFIELD", "p:1", "age", "30"]).await; + assert!(matches!(resp, Frame::Simple(ref s) if s == "OK")); + + let resp = c.cmd(&["PROTO.GETFIELD", "p:1", "age"]).await; + assert_eq!(resp, Frame::Integer(30)); +} + +#[tokio::test] +async fn setfield_bool() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + let resp = c.cmd(&["PROTO.SETFIELD", "p:1", "active", "false"]).await; + assert!(matches!(resp, Frame::Simple(ref s) if s == "OK")); + + let resp = c.cmd(&["PROTO.GETFIELD", "p:1", "active"]).await; + assert_eq!(resp, Frame::Integer(0)); +} + +#[tokio::test] +async fn setfield_missing_key() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let resp = c + .cmd(&["PROTO.SETFIELD", "nonexistent", "name", "bob"]) + .await; + assert!(matches!(resp, Frame::Null)); +} + +#[tokio::test] +async fn setfield_wrong_type() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + c.ok(&["SET", "str:key", "hello"]).await; + + let resp = c.cmd(&["PROTO.SETFIELD", "str:key", "name", "bob"]).await; + assert!(matches!(resp, Frame::Error(ref s) if s.starts_with("WRONGTYPE"))); +} + +#[tokio::test] +async fn setfield_nonexistent_field() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + let resp = c + .cmd(&["PROTO.SETFIELD", "p:1", "nonexistent", "value"]) + .await; + assert!(matches!(resp, Frame::Error(_))); +} + +#[tokio::test] +async fn setfield_invalid_value() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + // "abc" is not a valid int32 + let resp = c.cmd(&["PROTO.SETFIELD", "p:1", "age", "abc"]).await; + assert!(matches!(resp, Frame::Error(_))); +} + +// ---- PROTO.DELFIELD sharded tests ---- + +#[tokio::test] +async fn delfield_clears_field() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + let resp = c.cmd(&["PROTO.DELFIELD", "p:1", "name"]).await; + assert_eq!(resp, Frame::Integer(1)); + + // field should be reset to default (empty string) + let resp = c.cmd(&["PROTO.GETFIELD", "p:1", "name"]).await; + assert_eq!(resp, Frame::Bulk(Bytes::from(""))); + + // other fields preserved + let resp = c.cmd(&["PROTO.GETFIELD", "p:1", "age"]).await; + assert_eq!(resp, Frame::Integer(25)); +} + +#[tokio::test] +async fn delfield_missing_key() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let resp = c.cmd(&["PROTO.DELFIELD", "nonexistent", "name"]).await; + assert!(matches!(resp, Frame::Null)); +} + +#[tokio::test] +async fn delfield_returns_integer() { + let server = start_proto_server(false); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + let resp = c.cmd(&["PROTO.DELFIELD", "p:1", "name"]).await; + assert_eq!(resp, Frame::Integer(1)); +} + // ---- concurrent mode tests ---- // These mirror the core sharded tests to verify the concurrent handler's // proto command routing through the engine fallback path. @@ -527,3 +756,41 @@ async fn concurrent_getfield_string() { let resp = c.cmd(&["PROTO.GETFIELD", "user:1", "name"]).await; assert_eq!(resp, Frame::Bulk(Bytes::from("alice"))); } + +#[tokio::test] +async fn concurrent_setfield_string() { + let server = start_proto_server(true); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + let resp = c.cmd(&["PROTO.SETFIELD", "p:1", "name", "bob"]).await; + assert!(matches!(resp, Frame::Simple(ref s) if s == "OK")); + + let resp = c.cmd(&["PROTO.GETFIELD", "p:1", "name"]).await; + assert_eq!(resp, Frame::Bulk(Bytes::from("bob"))); +} + +#[tokio::test] +async fn concurrent_delfield_clears_field() { + let server = start_proto_server(true); + let mut c = server.connect().await; + + let desc = make_multi_field_descriptor(); + c.cmd_raw(&[b"PROTO.REGISTER", b"profiles", &desc]).await; + + let data = encode_profile(&desc, "alice", 25, true); + c.cmd_raw(&[b"PROTO.SET", b"p:1", b"test.Profile", &data]) + .await; + + let resp = c.cmd(&["PROTO.DELFIELD", "p:1", "name"]).await; + assert_eq!(resp, Frame::Integer(1)); + + let resp = c.cmd(&["PROTO.GETFIELD", "p:1", "name"]).await; + assert_eq!(resp, Frame::Bulk(Bytes::from(""))); +}