Skip to content

tasteful refactoring — clarity, dedup, structure #94

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kacy merged 6 commits into from
Feb 12, 2026
Merged

tasteful refactoring — clarity, dedup, structure #94

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
73455d2
refactor: move capped_capacity to shared format module
kacy Feb 12, 2026
55fd45f
docs: group aof tag constants by data type
kacy Feb 12, 2026
6b707cf
refactor: extract read_string_list helper in aof deserialization
kacy Feb 12, 2026
9fd4b67
refactor: extract parse_snap_value for encrypted snapshot reading
kacy Feb 12, 2026
3d6de5f
refactor: extract incr/write error conversion helpers in shard dispatch
kacy Feb 12, 2026
ad0eaec
refactor: extract startup helpers from main
kacy Feb 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 29 additions & 55 deletions crates/ember-core/src/shard.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -656,6 +656,25 @@ fn describe_request(req: &ShardRequest) -> RequestKind {
}
}

/// Converts an `IncrError` result into a `ShardResponse::Integer`.
fn incr_result(result: Result<i64, IncrError>) -> ShardResponse {
match result {
Ok(val) => ShardResponse::Integer(val),
Err(IncrError::WrongType) => ShardResponse::WrongType,
Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory,
Err(e) => ShardResponse::Err(e.to_string()),
}
}

/// Converts a `WriteError` result into a `ShardResponse::Len`.
fn write_result_len(result: Result<usize, WriteError>) -> ShardResponse {
match result {
Ok(len) => ShardResponse::Len(len),
Err(WriteError::WrongType) => ShardResponse::WrongType,
Err(WriteError::OutOfMemory) => ShardResponse::OutOfMemory,
}
}

/// Executes a single request against the keyspace.
fn dispatch(
ks: &mut Keyspace,
Expand Down Expand Up @@ -687,31 +706,11 @@ fn dispatch(
SetResult::OutOfMemory => ShardResponse::OutOfMemory,
}
}
ShardRequest::Incr { key } => match ks.incr(key) {
Ok(val) => ShardResponse::Integer(val),
Err(IncrError::WrongType) => ShardResponse::WrongType,
Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory,
Err(e) => ShardResponse::Err(e.to_string()),
},
ShardRequest::Decr { key } => match ks.decr(key) {
Ok(val) => ShardResponse::Integer(val),
Err(IncrError::WrongType) => ShardResponse::WrongType,
Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory,
Err(e) => ShardResponse::Err(e.to_string()),
},
ShardRequest::IncrBy { key, delta } => match ks.incr_by(key, *delta) {
Ok(val) => ShardResponse::Integer(val),
Err(IncrError::WrongType) => ShardResponse::WrongType,
Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory,
Err(e) => ShardResponse::Err(e.to_string()),
},
ShardRequest::Incr { key } => incr_result(ks.incr(key)),
ShardRequest::Decr { key } => incr_result(ks.decr(key)),
ShardRequest::IncrBy { key, delta } => incr_result(ks.incr_by(key, *delta)),
ShardRequest::DecrBy { key, delta } => match delta.checked_neg() {
Some(neg) => match ks.incr_by(key, neg) {
Ok(val) => ShardResponse::Integer(val),
Err(IncrError::WrongType) => ShardResponse::WrongType,
Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory,
Err(e) => ShardResponse::Err(e.to_string()),
},
Some(neg) => incr_result(ks.incr_by(key, neg)),
None => ShardResponse::Err("ERR increment or decrement would overflow".into()),
},
ShardRequest::IncrByFloat { key, delta } => match ks.incr_by_float(key, *delta) {
Expand All @@ -720,11 +719,7 @@ fn dispatch(
Err(IncrFloatError::OutOfMemory) => ShardResponse::OutOfMemory,
Err(e) => ShardResponse::Err(e.to_string()),
},
ShardRequest::Append { key, value } => match ks.append(key, value) {
Ok(len) => ShardResponse::Len(len),
Err(WriteError::WrongType) => ShardResponse::WrongType,
Err(WriteError::OutOfMemory) => ShardResponse::OutOfMemory,
},
ShardRequest::Append { key, value } => write_result_len(ks.append(key, value)),
ShardRequest::Strlen { key } => match ks.strlen(key) {
Ok(len) => ShardResponse::Len(len),
Err(_) => ShardResponse::WrongType,
Expand All @@ -750,16 +745,8 @@ fn dispatch(
ShardRequest::Pexpire { key, milliseconds } => {
ShardResponse::Bool(ks.pexpire(key, *milliseconds))
}
ShardRequest::LPush { key, values } => match ks.lpush(key, values) {
Ok(len) => ShardResponse::Len(len),
Err(WriteError::WrongType) => ShardResponse::WrongType,
Err(WriteError::OutOfMemory) => ShardResponse::OutOfMemory,
},
ShardRequest::RPush { key, values } => match ks.rpush(key, values) {
Ok(len) => ShardResponse::Len(len),
Err(WriteError::WrongType) => ShardResponse::WrongType,
Err(WriteError::OutOfMemory) => ShardResponse::OutOfMemory,
},
ShardRequest::LPush { key, values } => write_result_len(ks.lpush(key, values)),
ShardRequest::RPush { key, values } => write_result_len(ks.rpush(key, values)),
ShardRequest::LPop { key } => match ks.lpop(key) {
Ok(val) => ShardResponse::Value(val.map(Value::String)),
Err(_) => ShardResponse::WrongType,
Expand Down Expand Up @@ -844,11 +831,7 @@ fn dispatch(
keys,
}
}
ShardRequest::HSet { key, fields } => match ks.hset(key, fields) {
Ok(count) => ShardResponse::Len(count),
Err(WriteError::WrongType) => ShardResponse::WrongType,
Err(WriteError::OutOfMemory) => ShardResponse::OutOfMemory,
},
ShardRequest::HSet { key, fields } => write_result_len(ks.hset(key, fields)),
ShardRequest::HGet { key, field } => match ks.hget(key, field) {
Ok(val) => ShardResponse::Value(val.map(Value::String)),
Err(_) => ShardResponse::WrongType,
Expand All @@ -872,12 +855,7 @@ fn dispatch(
Ok(len) => ShardResponse::Len(len),
Err(_) => ShardResponse::WrongType,
},
ShardRequest::HIncrBy { key, field, delta } => match ks.hincrby(key, field, *delta) {
Ok(val) => ShardResponse::Integer(val),
Err(IncrError::WrongType) => ShardResponse::WrongType,
Err(IncrError::OutOfMemory) => ShardResponse::OutOfMemory,
Err(e) => ShardResponse::Err(e.to_string()),
},
ShardRequest::HIncrBy { key, field, delta } => incr_result(ks.hincrby(key, field, *delta)),
ShardRequest::HKeys { key } => match ks.hkeys(key) {
Ok(keys) => ShardResponse::StringArray(keys),
Err(_) => ShardResponse::WrongType,
Expand All @@ -890,11 +868,7 @@ fn dispatch(
Ok(vals) => ShardResponse::OptionalArray(vals),
Err(_) => ShardResponse::WrongType,
},
ShardRequest::SAdd { key, members } => match ks.sadd(key, members) {
Ok(count) => ShardResponse::Len(count),
Err(WriteError::WrongType) => ShardResponse::WrongType,
Err(WriteError::OutOfMemory) => ShardResponse::OutOfMemory,
},
ShardRequest::SAdd { key, members } => write_result_len(ks.sadd(key, members)),
ShardRequest::SRem { key, members } => match ks.srem(key, members) {
Ok(count) => ShardResponse::Len(count),
Err(_) => ShardResponse::WrongType,
Expand Down
83 changes: 43 additions & 40 deletions crates/ember-persistence/src/aof.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,29 +38,55 @@ fn read_string(r: &mut impl io::Read, field: &str) -> Result<String, FormatError
})
}

/// Record tags for the AOF format.
/// Reads a count-prefixed list of strings: `[count: u32][string]*`.
/// Used by SADD, SREM, HDEL, and ZREM deserialization.
fn read_string_list(r: &mut impl io::Read, field: &str) -> Result<Vec<String>, FormatError> {
let count = format::read_u32(r)?;
let mut items = Vec::with_capacity(format::capped_capacity(count));
for _ in 0..count {
items.push(read_string(r, field)?);
}
Ok(items)
}

// -- record tags --
// values are stable and must not change (on-disk format).

// string
const TAG_SET: u8 = 1;
const TAG_DEL: u8 = 2;
const TAG_EXPIRE: u8 = 3;
const TAG_INCR: u8 = 12;
const TAG_DECR: u8 = 13;
const TAG_INCRBY: u8 = 19;
const TAG_DECRBY: u8 = 20;
const TAG_APPEND: u8 = 21;

// list
const TAG_LPUSH: u8 = 4;
const TAG_RPUSH: u8 = 5;
const TAG_LPOP: u8 = 6;
const TAG_RPOP: u8 = 7;

// sorted set
const TAG_ZADD: u8 = 8;
const TAG_ZREM: u8 = 9;
const TAG_PERSIST: u8 = 10;
const TAG_PEXPIRE: u8 = 11;
const TAG_INCR: u8 = 12;
const TAG_DECR: u8 = 13;

// hash
const TAG_HSET: u8 = 14;
const TAG_HDEL: u8 = 15;
const TAG_HINCRBY: u8 = 16;

// set
const TAG_SADD: u8 = 17;
const TAG_SREM: u8 = 18;
const TAG_INCRBY: u8 = 19;
const TAG_DECRBY: u8 = 20;
const TAG_APPEND: u8 = 21;

// key lifecycle
const TAG_DEL: u8 = 2;
const TAG_EXPIRE: u8 = 3;
const TAG_PERSIST: u8 = 10;
const TAG_PEXPIRE: u8 = 11;
const TAG_RENAME: u8 = 22;

// protobuf
#[cfg(feature = "protobuf")]
const TAG_PROTO_SET: u8 = 23;
#[cfg(feature = "protobuf")]
Expand Down Expand Up @@ -304,13 +330,6 @@ impl AofRecord {
Ok(buf)
}

/// Cap pre-allocation to avoid huge allocations from corrupt count fields.
/// The loop will still iterate `count` times — this just limits the
/// up-front reservation so a bogus u32 can't exhaust memory.
fn capped_capacity(count: u32) -> usize {
(count as usize).min(65_536)
}

/// Deserializes a record from a byte slice (tag + payload, no CRC).
fn from_bytes(data: &[u8]) -> Result<Self, FormatError> {
let mut cursor = io::Cursor::new(data);
Expand Down Expand Up @@ -338,7 +357,7 @@ impl AofRecord {
TAG_LPUSH | TAG_RPUSH => {
let key = read_string(&mut cursor, "key")?;
let count = format::read_u32(&mut cursor)?;
let mut values = Vec::with_capacity(Self::capped_capacity(count));
let mut values = Vec::with_capacity(format::capped_capacity(count));
for _ in 0..count {
values.push(Bytes::from(format::read_bytes(&mut cursor)?));
}
Expand All @@ -359,7 +378,7 @@ impl AofRecord {
TAG_ZADD => {
let key = read_string(&mut cursor, "key")?;
let count = format::read_u32(&mut cursor)?;
let mut members = Vec::with_capacity(Self::capped_capacity(count));
let mut members = Vec::with_capacity(format::capped_capacity(count));
for _ in 0..count {
let score = format::read_f64(&mut cursor)?;
let member = read_string(&mut cursor, "member")?;
Expand All @@ -369,11 +388,7 @@ impl AofRecord {
}
TAG_ZREM => {
let key = read_string(&mut cursor, "key")?;
let count = format::read_u32(&mut cursor)?;
let mut members = Vec::with_capacity(Self::capped_capacity(count));
for _ in 0..count {
members.push(read_string(&mut cursor, "member")?);
}
let members = read_string_list(&mut cursor, "member")?;
Ok(AofRecord::ZRem { key, members })
}
TAG_PERSIST => {
Expand All @@ -396,7 +411,7 @@ impl AofRecord {
TAG_HSET => {
let key = read_string(&mut cursor, "key")?;
let count = format::read_u32(&mut cursor)?;
let mut fields = Vec::with_capacity(Self::capped_capacity(count));
let mut fields = Vec::with_capacity(format::capped_capacity(count));
for _ in 0..count {
let field = read_string(&mut cursor, "field")?;
let value = Bytes::from(format::read_bytes(&mut cursor)?);
Expand All @@ -406,11 +421,7 @@ impl AofRecord {
}
TAG_HDEL => {
let key = read_string(&mut cursor, "key")?;
let count = format::read_u32(&mut cursor)?;
let mut fields = Vec::with_capacity(Self::capped_capacity(count));
for _ in 0..count {
fields.push(read_string(&mut cursor, "field")?);
}
let fields = read_string_list(&mut cursor, "field")?;
Ok(AofRecord::HDel { key, fields })
}
TAG_HINCRBY => {
Expand All @@ -421,20 +432,12 @@ impl AofRecord {
}
TAG_SADD => {
let key = read_string(&mut cursor, "key")?;
let count = format::read_u32(&mut cursor)?;
let mut members = Vec::with_capacity(Self::capped_capacity(count));
for _ in 0..count {
members.push(read_string(&mut cursor, "member")?);
}
let members = read_string_list(&mut cursor, "member")?;
Ok(AofRecord::SAdd { key, members })
}
TAG_SREM => {
let key = read_string(&mut cursor, "key")?;
let count = format::read_u32(&mut cursor)?;
let mut members = Vec::with_capacity(Self::capped_capacity(count));
for _ in 0..count {
members.push(read_string(&mut cursor, "member")?);
}
let members = read_string_list(&mut cursor, "member")?;
Ok(AofRecord::SRem { key, members })
}
TAG_INCRBY => {
Expand Down
7 changes: 7 additions & 0 deletions crates/ember-persistence/src/format.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,13 @@ pub fn verify_crc32(data: &[u8], expected: u32) -> Result<(), FormatError> {
verify_crc32_values(actual, expected)
}

/// Caps pre-allocation to avoid huge allocations from corrupt count fields.
/// The loop will still iterate `count` times — this just limits the
/// up-front reservation so a bogus u32 can't exhaust memory.
pub fn capped_capacity(count: u32) -> usize {
(count as usize).min(65_536)
}

/// Verifies that two CRC32 values match.
pub fn verify_crc32_values(computed: u32, stored: u32) -> Result<(), FormatError> {
if computed != stored {
Expand Down
Loading
Loading