From 069c50da76d5efb22f40a66e1a2ae2225da72f18 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sat, 14 Feb 2026 17:20:05 -0500 Subject: [PATCH 1/2] refactor: add tag() method, combine aof serialization patterns added AofRecord::tag() that maps each variant to its on-disk tag byte. moved the write_u8(tag) call before the match, then combined variants with identical serialization shapes: - key-only: Del | LPop | RPop | Persist | Incr | Decr - key + byte list: LPush | RPush - key + string list: ZRem | SAdd | SRem - key + i64: IncrBy | DecrBy each group writes the tag once via self.tag(), then shares the payload serialization logic. unique variants (ZAdd, HSet, VAdd, etc.) keep their own arms with section comments. --- crates/ember-persistence/src/aof.rs | 174 ++++++++++++++-------------- 1 file changed, 88 insertions(+), 86 deletions(-) diff --git a/crates/ember-persistence/src/aof.rs b/crates/ember-persistence/src/aof.rs index fd3c89a2..99dc0698 100644 --- a/crates/ember-persistence/src/aof.rs +++ b/crates/ember-persistence/src/aof.rs @@ -191,6 +191,42 @@ pub enum AofRecord { } impl AofRecord { + /// Returns the on-disk tag byte for this record variant. + fn tag(&self) -> u8 { + match self { + AofRecord::Set { .. } => TAG_SET, + AofRecord::Del { .. } => TAG_DEL, + AofRecord::Expire { .. } => TAG_EXPIRE, + AofRecord::LPush { .. } => TAG_LPUSH, + AofRecord::RPush { .. } => TAG_RPUSH, + AofRecord::LPop { .. } => TAG_LPOP, + AofRecord::RPop { .. } => TAG_RPOP, + AofRecord::ZAdd { .. } => TAG_ZADD, + AofRecord::ZRem { .. } => TAG_ZREM, + AofRecord::Persist { .. } => TAG_PERSIST, + AofRecord::Pexpire { .. } => TAG_PEXPIRE, + AofRecord::Incr { .. } => TAG_INCR, + AofRecord::Decr { .. } => TAG_DECR, + AofRecord::HSet { .. } => TAG_HSET, + AofRecord::HDel { .. } => TAG_HDEL, + AofRecord::HIncrBy { .. } => TAG_HINCRBY, + AofRecord::SAdd { .. } => TAG_SADD, + AofRecord::SRem { .. } => TAG_SREM, + AofRecord::IncrBy { .. } => TAG_INCRBY, + AofRecord::DecrBy { .. } => TAG_DECRBY, + AofRecord::Append { .. } => TAG_APPEND, + AofRecord::Rename { .. } => TAG_RENAME, + #[cfg(feature = "vector")] + AofRecord::VAdd { .. } => TAG_VADD, + #[cfg(feature = "vector")] + AofRecord::VRem { .. } => TAG_VREM, + #[cfg(feature = "protobuf")] + AofRecord::ProtoSet { .. } => TAG_PROTO_SET, + #[cfg(feature = "protobuf")] + AofRecord::ProtoRegister { .. } => TAG_PROTO_REGISTER, + } + } + /// Estimates the serialized size of this record in bytes. /// /// Used as a capacity hint for `to_bytes()` to avoid intermediate @@ -291,52 +327,75 @@ impl AofRecord { /// Serializes this record into a byte vector (tag + payload, no CRC). fn to_bytes(&self) -> Result, FormatError> { let mut buf = Vec::with_capacity(self.estimated_size()); + format::write_u8(&mut buf, self.tag())?; + match self { + // key-only: tag + key + AofRecord::Del { key } + | AofRecord::LPop { key } + | AofRecord::RPop { key } + | AofRecord::Persist { key } + | AofRecord::Incr { key } + | AofRecord::Decr { key } => { + format::write_bytes(&mut buf, key.as_bytes())?; + } + + // key + bytes value + expire AofRecord::Set { key, value, expire_ms, } => { - format::write_u8(&mut buf, TAG_SET)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_bytes(&mut buf, value)?; format::write_i64(&mut buf, *expire_ms)?; } - AofRecord::Del { key } => { - format::write_u8(&mut buf, TAG_DEL)?; - format::write_bytes(&mut buf, key.as_bytes())?; - } + + // key + i64 AofRecord::Expire { key, seconds } => { - format::write_u8(&mut buf, TAG_EXPIRE)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_i64(&mut buf, *seconds as i64)?; } - AofRecord::LPush { key, values } => { - format::write_u8(&mut buf, TAG_LPUSH)?; + AofRecord::Pexpire { key, milliseconds } => { format::write_bytes(&mut buf, key.as_bytes())?; - format::write_len(&mut buf, values.len())?; - for v in values { - format::write_bytes(&mut buf, v)?; - } + format::write_i64(&mut buf, *milliseconds as i64)?; + } + AofRecord::IncrBy { key, delta } + | AofRecord::DecrBy { key, delta } => { + format::write_bytes(&mut buf, key.as_bytes())?; + format::write_i64(&mut buf, *delta)?; } - AofRecord::RPush { key, values } => { - format::write_u8(&mut buf, TAG_RPUSH)?; + + // key + byte list + AofRecord::LPush { key, values } + | AofRecord::RPush { key, values } => { format::write_bytes(&mut buf, key.as_bytes())?; format::write_len(&mut buf, values.len())?; for v in values { format::write_bytes(&mut buf, v)?; } } - AofRecord::LPop { key } => { - format::write_u8(&mut buf, TAG_LPOP)?; + + // key + string list + AofRecord::ZRem { key, members } + | AofRecord::SAdd { key, members } + | AofRecord::SRem { key, members } => { format::write_bytes(&mut buf, key.as_bytes())?; + format::write_len(&mut buf, members.len())?; + for member in members { + format::write_bytes(&mut buf, member.as_bytes())?; + } } - AofRecord::RPop { key } => { - format::write_u8(&mut buf, TAG_RPOP)?; + AofRecord::HDel { key, fields } => { format::write_bytes(&mut buf, key.as_bytes())?; + format::write_len(&mut buf, fields.len())?; + for field in fields { + format::write_bytes(&mut buf, field.as_bytes())?; + } } + + // key + scored members AofRecord::ZAdd { key, members } => { - format::write_u8(&mut buf, TAG_ZADD)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_len(&mut buf, members.len())?; for (score, member) in members { @@ -344,33 +403,9 @@ impl AofRecord { format::write_bytes(&mut buf, member.as_bytes())?; } } - AofRecord::ZRem { key, members } => { - format::write_u8(&mut buf, TAG_ZREM)?; - format::write_bytes(&mut buf, key.as_bytes())?; - format::write_len(&mut buf, members.len())?; - for member in members { - format::write_bytes(&mut buf, member.as_bytes())?; - } - } - AofRecord::Persist { key } => { - format::write_u8(&mut buf, TAG_PERSIST)?; - format::write_bytes(&mut buf, key.as_bytes())?; - } - AofRecord::Pexpire { key, milliseconds } => { - format::write_u8(&mut buf, TAG_PEXPIRE)?; - format::write_bytes(&mut buf, key.as_bytes())?; - format::write_i64(&mut buf, *milliseconds as i64)?; - } - AofRecord::Incr { key } => { - format::write_u8(&mut buf, TAG_INCR)?; - format::write_bytes(&mut buf, key.as_bytes())?; - } - AofRecord::Decr { key } => { - format::write_u8(&mut buf, TAG_DECR)?; - format::write_bytes(&mut buf, key.as_bytes())?; - } + + // key + field-value pairs AofRecord::HSet { key, fields } => { - format::write_u8(&mut buf, TAG_HSET)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_len(&mut buf, fields.len())?; for (field, value) in fields { @@ -378,56 +413,26 @@ impl AofRecord { format::write_bytes(&mut buf, value)?; } } - AofRecord::HDel { key, fields } => { - format::write_u8(&mut buf, TAG_HDEL)?; - format::write_bytes(&mut buf, key.as_bytes())?; - format::write_len(&mut buf, fields.len())?; - for field in fields { - format::write_bytes(&mut buf, field.as_bytes())?; - } - } + + // key + field + delta AofRecord::HIncrBy { key, field, delta } => { - format::write_u8(&mut buf, TAG_HINCRBY)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_bytes(&mut buf, field.as_bytes())?; format::write_i64(&mut buf, *delta)?; } - AofRecord::SAdd { key, members } => { - format::write_u8(&mut buf, TAG_SADD)?; - format::write_bytes(&mut buf, key.as_bytes())?; - format::write_len(&mut buf, members.len())?; - for member in members { - format::write_bytes(&mut buf, member.as_bytes())?; - } - } - AofRecord::SRem { key, members } => { - format::write_u8(&mut buf, TAG_SREM)?; - format::write_bytes(&mut buf, key.as_bytes())?; - format::write_len(&mut buf, members.len())?; - for member in members { - format::write_bytes(&mut buf, member.as_bytes())?; - } - } - AofRecord::IncrBy { key, delta } => { - format::write_u8(&mut buf, TAG_INCRBY)?; - format::write_bytes(&mut buf, key.as_bytes())?; - format::write_i64(&mut buf, *delta)?; - } - AofRecord::DecrBy { key, delta } => { - format::write_u8(&mut buf, TAG_DECRBY)?; - format::write_bytes(&mut buf, key.as_bytes())?; - format::write_i64(&mut buf, *delta)?; - } + + // key + bytes value (no expire) AofRecord::Append { key, value } => { - format::write_u8(&mut buf, TAG_APPEND)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_bytes(&mut buf, value)?; } + + // key + newkey AofRecord::Rename { key, newkey } => { - format::write_u8(&mut buf, TAG_RENAME)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_bytes(&mut buf, newkey.as_bytes())?; } + #[cfg(feature = "vector")] AofRecord::VAdd { key, @@ -438,7 +443,6 @@ impl AofRecord { connectivity, expansion_add, } => { - format::write_u8(&mut buf, TAG_VADD)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_bytes(&mut buf, element.as_bytes())?; format::write_len(&mut buf, vector.len())?; @@ -452,10 +456,10 @@ impl AofRecord { } #[cfg(feature = "vector")] AofRecord::VRem { key, element } => { - format::write_u8(&mut buf, TAG_VREM)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_bytes(&mut buf, element.as_bytes())?; } + #[cfg(feature = "protobuf")] AofRecord::ProtoSet { key, @@ -463,7 +467,6 @@ impl AofRecord { data, expire_ms, } => { - format::write_u8(&mut buf, TAG_PROTO_SET)?; format::write_bytes(&mut buf, key.as_bytes())?; format::write_bytes(&mut buf, type_name.as_bytes())?; format::write_bytes(&mut buf, data)?; @@ -471,7 +474,6 @@ impl AofRecord { } #[cfg(feature = "protobuf")] AofRecord::ProtoRegister { name, descriptor } => { - format::write_u8(&mut buf, TAG_PROTO_REGISTER)?; format::write_bytes(&mut buf, name.as_bytes())?; format::write_bytes(&mut buf, descriptor)?; } From 125e6edf1f2f89d7fa22de84922694e6b2ee85b2 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sat, 14 Feb 2026 17:20:12 -0500 Subject: [PATCH 2/2] refactor: extract parse_utf8 helper in snapshot deserialization the pattern String::from_utf8(bytes).map_err(|_| FormatError::Io(...)) was repeated 6 times across read_plaintext_entry. extracted a parse_utf8 helper that takes the bytes and a field name for the error message. also wired the existing read_snap_string (encryption path) to use it. --- crates/ember-persistence/src/snapshot.rs | 57 +++++++----------------- 1 file changed, 17 insertions(+), 40 deletions(-) diff --git a/crates/ember-persistence/src/snapshot.rs b/crates/ember-persistence/src/snapshot.rs index 2c3a2d2f..fca48bbe 100644 --- a/crates/ember-persistence/src/snapshot.rs +++ b/crates/ember-persistence/src/snapshot.rs @@ -41,10 +41,10 @@ const TYPE_VECTOR: u8 = 6; #[cfg(feature = "protobuf")] const TYPE_PROTO: u8 = 5; -/// Reads a UTF-8 string from a length-prefixed byte field. -#[cfg(feature = "encryption")] -fn read_snap_string(r: &mut impl io::Read, field: &str) -> Result { - let bytes = format::read_bytes(r)?; +/// Converts raw bytes to a UTF-8 string, returning a descriptive error +/// on invalid data. `field` names the field for the error message +/// (e.g. "key", "member", "hash field"). +fn parse_utf8(bytes: Vec, field: &str) -> Result { String::from_utf8(bytes).map_err(|_| { FormatError::Io(io::Error::new( io::ErrorKind::InvalidData, @@ -53,6 +53,13 @@ fn read_snap_string(r: &mut impl io::Read, field: &str) -> Result Result { + let bytes = format::read_bytes(r)?; + parse_utf8(bytes, field) +} + /// Parses a type-tagged SnapValue from a reader (v2+ format). /// /// Used by `read_encrypted_entry` to parse the `[type_tag][payload]` @@ -608,12 +615,7 @@ impl SnapshotReader { format::write_f64(&mut buf, score)?; let member_bytes = format::read_bytes(&mut self.reader)?; format::write_bytes(&mut buf, &member_bytes)?; - let member = String::from_utf8(member_bytes).map_err(|_| { - FormatError::Io(io::Error::new( - io::ErrorKind::InvalidData, - "member is not valid utf-8", - )) - })?; + let member = parse_utf8(member_bytes, "member")?; members.push((score, member)); } SnapValue::SortedSet(members) @@ -626,12 +628,7 @@ impl SnapshotReader { for _ in 0..count { let field_bytes = format::read_bytes(&mut self.reader)?; format::write_bytes(&mut buf, &field_bytes)?; - let field = String::from_utf8(field_bytes).map_err(|_| { - FormatError::Io(io::Error::new( - io::ErrorKind::InvalidData, - "hash field is not valid utf-8", - )) - })?; + let field = parse_utf8(field_bytes, "hash field")?; let value_bytes = format::read_bytes(&mut self.reader)?; format::write_bytes(&mut buf, &value_bytes)?; map.insert(field, Bytes::from(value_bytes)); @@ -646,12 +643,7 @@ impl SnapshotReader { for _ in 0..count { let member_bytes = format::read_bytes(&mut self.reader)?; format::write_bytes(&mut buf, &member_bytes)?; - let member = String::from_utf8(member_bytes).map_err(|_| { - FormatError::Io(io::Error::new( - io::ErrorKind::InvalidData, - "set member is not valid utf-8", - )) - })?; + let member = parse_utf8(member_bytes, "set member")?; set.insert(member); } SnapValue::Set(set) @@ -696,12 +688,7 @@ impl SnapshotReader { for _ in 0..count { let name_bytes = format::read_bytes(&mut self.reader)?; format::write_bytes(&mut buf, &name_bytes)?; - let name = String::from_utf8(name_bytes).map_err(|_| { - FormatError::Io(io::Error::new( - io::ErrorKind::InvalidData, - "vector element name is not valid utf-8", - )) - })?; + let name = parse_utf8(name_bytes, "vector element name")?; let mut vector = Vec::with_capacity(dim as usize); for _ in 0..dim { let v = format::read_f32(&mut self.reader)?; @@ -723,12 +710,7 @@ impl SnapshotReader { TYPE_PROTO => { let type_name_bytes = format::read_bytes(&mut self.reader)?; format::write_bytes(&mut buf, &type_name_bytes)?; - let type_name = String::from_utf8(type_name_bytes).map_err(|_| { - FormatError::Io(io::Error::new( - io::ErrorKind::InvalidData, - "proto type_name is not valid utf-8", - )) - })?; + let type_name = parse_utf8(type_name_bytes, "proto type_name")?; let data = format::read_bytes(&mut self.reader)?; format::write_bytes(&mut buf, &data)?; SnapValue::Proto { @@ -746,12 +728,7 @@ impl SnapshotReader { format::write_i64(&mut buf, expire_ms)?; self.hasher.update(&buf); - let key = String::from_utf8(key_bytes).map_err(|_| { - FormatError::Io(io::Error::new( - io::ErrorKind::InvalidData, - "key is not valid utf-8", - )) - })?; + let key = parse_utf8(key_bytes, "key")?; self.read_so_far += 1; Ok(Some(SnapEntry {