diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e3c40ae..93326c7c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed +- **One value codec for RDB snapshots and KV disk-offload spill (W1).** The + ~150-line value-body encoder/decoder that existed twice — `rdb::write_entry`'s + value section and `kv_serde::serialize_collection` — kept bit-compatible only + by discipline, is now a single spec-documented module + (`src/storage/value_codec.rs`) that both call sites delegate to; the + `RedisValueRef → ValueType` mapping likewise exists once (the former inline + copies in `eviction.rs` and `kv_spill.rs` delegate). Wire format is + byte-identical (pinned by hand-built golden byte-spec tests + legacy + pre-trailer decode tests); existing RDB files and spill pages load unchanged. + +### Fixed +- **Spill decode allocation-DoS hardening.** The spill value decoder missed the + RDB count-validation fix and fed corrupt length fields straight into + `Vec::with_capacity`; the unified codec validates every count against + remaining input before allocating, on both planes. +- **Corrupt sorted-set listpack scores are fail-closed.** Both former encoder + copies silently wrote `0.0` for a listpack zset score that failed to parse; + the codec now refuses to encode (`ValueCodecError::CorruptScore`) — an RDB + save fails loudly and a spill aborts instead of persisting a corrupted score. + ### Documentation - **New "The Moon Journey" page (`docs/journey.md`) — the development story toward a production-efficient database, grounded in real evidence.** Traces diff --git a/src/persistence/rdb.rs b/src/persistence/rdb.rs index 1cd13bc3..d556684a 100644 --- a/src/persistence/rdb.rs +++ b/src/persistence/rdb.rs @@ -19,12 +19,14 @@ use crc32fast::Hasher; use ordered_float::OrderedFloat; use crate::error::{MoonError, RdbError}; +use crate::persistence::kv_page::ValueType; use crate::storage::bptree::BPTree; use crate::storage::compact_key::CompactKey; use crate::storage::compact_value::RedisValueRef; use crate::storage::db::Database; use crate::storage::entry::{Entry, RedisValue, current_secs, current_time_ms}; use crate::storage::stream::{Stream as StreamData, StreamId}; +use crate::storage::value_codec::{self, HashTtlTrailer}; // Format constants const RDB_MAGIC: &[u8] = b"MOON"; @@ -1028,20 +1030,16 @@ pub(crate) fn write_entry( entry: &Entry, base_ts: u32, ) -> Result<(), MoonError> { + let val_ref = entry.value.as_redis_value(); + // Type tag -- compact variants serialize as the same type as their full-size counterparts - let type_tag = match entry.value.as_redis_value() { - RedisValueRef::String(_) => TYPE_STRING, - RedisValueRef::Hash(_) - | RedisValueRef::HashListpack(_) - | RedisValueRef::HashWithTtl { .. } => TYPE_HASH, - RedisValueRef::List(_) | RedisValueRef::ListListpack(_) => TYPE_LIST, - RedisValueRef::Set(_) | RedisValueRef::SetListpack(_) | RedisValueRef::SetIntset(_) => { - TYPE_SET - } - RedisValueRef::SortedSet { .. } - | RedisValueRef::SortedSetBPTree { .. } - | RedisValueRef::SortedSetListpack(_) => TYPE_SORTED_SET, - RedisValueRef::Stream(_) => TYPE_STREAM, + let type_tag = match value_codec::value_type_of(&val_ref) { + ValueType::String => TYPE_STRING, + ValueType::Hash => TYPE_HASH, + ValueType::List => TYPE_LIST, + ValueType::Set => TYPE_SET, + ValueType::ZSet => TYPE_SORTED_SET, + ValueType::Stream => TYPE_STREAM, }; buf.write_all(&[type_tag])?; @@ -1056,148 +1054,19 @@ pub(crate) fn write_entry( }; buf.write_all(&ttl_ms.to_le_bytes())?; - // Value data -- compact variants expand to element-level format for persistence - match entry.value.as_redis_value() { + // Value data -- strings inline, collections via the shared value codec + // (compact variants expand to the canonical element-level format). + match val_ref { RedisValueRef::String(s) => { write_bytes(buf, s)?; } - RedisValueRef::Hash(map) => { - buf.write_all(&(map.len() as u32).to_le_bytes())?; - for (field, val) in map.iter() { - write_bytes(buf, field)?; - write_bytes(buf, val)?; - } - // v2 trailer: no per-field TTLs for plain Hash. Emit ttl_count=0. - buf.write_all(&0u32.to_le_bytes())?; - } - RedisValueRef::HashWithTtl { fields, ttls, .. } => { - buf.write_all(&(fields.len() as u32).to_le_bytes())?; - for (field, val) in fields.iter() { - write_bytes(buf, field)?; - write_bytes(buf, val)?; - } - // v2 trailer: per-field TTL sidecar. - buf.write_all(&(ttls.len() as u32).to_le_bytes())?; - for (field, ttl_ms) in ttls.iter() { - write_bytes(buf, field)?; - buf.write_all(&ttl_ms.to_le_bytes())?; - } - } - RedisValueRef::HashListpack(lp) => { - let map = lp.to_hash_map(); - buf.write_all(&(map.len() as u32).to_le_bytes())?; - for (field, val) in &map { - write_bytes(buf, field)?; - write_bytes(buf, val)?; - } - // v2 trailer: listpack-encoded hashes never carry TTLs. - buf.write_all(&0u32.to_le_bytes())?; - } - RedisValueRef::List(list) => { - buf.write_all(&(list.len() as u32).to_le_bytes())?; - for elem in list.iter() { - write_bytes(buf, elem)?; - } - } - RedisValueRef::ListListpack(lp) => { - let list = lp.to_vec_deque(); - buf.write_all(&(list.len() as u32).to_le_bytes())?; - for elem in &list { - write_bytes(buf, elem)?; - } - } - RedisValueRef::Set(set) => { - buf.write_all(&(set.len() as u32).to_le_bytes())?; - for member in set.iter() { - write_bytes(buf, member)?; - } - } - RedisValueRef::SetListpack(lp) => { - let set = lp.to_hash_set(); - buf.write_all(&(set.len() as u32).to_le_bytes())?; - for member in &set { - write_bytes(buf, member)?; - } - } - RedisValueRef::SetIntset(is) => { - let set = is.to_hash_set(); - buf.write_all(&(set.len() as u32).to_le_bytes())?; - for member in &set { - write_bytes(buf, member)?; - } - } - RedisValueRef::SortedSet { members, .. } - | RedisValueRef::SortedSetBPTree { members, .. } => { - buf.write_all(&(members.len() as u32).to_le_bytes())?; - for (member, score) in members.iter() { - write_bytes(buf, member)?; - buf.write_all(&score.to_le_bytes())?; - } - } - RedisValueRef::SortedSetListpack(lp) => { - // Listpack stores sorted set as [member, score, member, score, ...] - let mut count: u32 = 0; - let pairs: Vec<_> = lp.iter_pairs().collect(); - // Write count placeholder, then entries - let count_pos = buf.len(); - buf.write_all(&0u32.to_le_bytes())?; - for (member_entry, score_entry) in &pairs { - let member_bytes = member_entry.as_bytes(); - let score_bytes = score_entry.as_bytes(); - let score: f64 = std::str::from_utf8(&score_bytes) - .ok() - .and_then(|s| s.parse().ok()) - .unwrap_or(0.0); - write_bytes(buf, &member_bytes)?; - buf.write_all(&score.to_le_bytes())?; - count += 1; - } - // Patch count - let count_bytes = count.to_le_bytes(); - buf[count_pos..count_pos + 4].copy_from_slice(&count_bytes); - } - RedisValueRef::Stream(stream) => { - // Entry count + last_id - buf.write_all(&(stream.entries.len() as u64).to_le_bytes())?; - buf.write_all(&stream.last_id.ms.to_le_bytes())?; - buf.write_all(&stream.last_id.seq.to_le_bytes())?; - // Entries - for (id, fields) in &stream.entries { - buf.write_all(&id.ms.to_le_bytes())?; - buf.write_all(&id.seq.to_le_bytes())?; - buf.write_all(&(fields.len() as u32).to_le_bytes())?; - for (field, value) in fields { - write_bytes(buf, field)?; - write_bytes(buf, value)?; - } - } - // Consumer groups - buf.write_all(&(stream.groups.len() as u32).to_le_bytes())?; - for (group_name, group) in &stream.groups { - write_bytes(buf, group_name)?; - buf.write_all(&group.last_delivered_id.ms.to_le_bytes())?; - buf.write_all(&group.last_delivered_id.seq.to_le_bytes())?; - // PEL - buf.write_all(&(group.pel.len() as u32).to_le_bytes())?; - for (id, pe) in &group.pel { - buf.write_all(&id.ms.to_le_bytes())?; - buf.write_all(&id.seq.to_le_bytes())?; - write_bytes(buf, &pe.consumer)?; - buf.write_all(&pe.delivery_time.to_le_bytes())?; - buf.write_all(&pe.delivery_count.to_le_bytes())?; - } - // Consumers - buf.write_all(&(group.consumers.len() as u32).to_le_bytes())?; - for (cname, consumer) in &group.consumers { - write_bytes(buf, cname)?; - buf.write_all(&consumer.seen_time.to_le_bytes())?; - buf.write_all(&(consumer.pending.len() as u32).to_le_bytes())?; - for (id, _) in &consumer.pending { - buf.write_all(&id.ms.to_le_bytes())?; - buf.write_all(&id.seq.to_le_bytes())?; - } - } - } + ref other => { + value_codec::encode_value_body(other, buf).map_err(|e| RdbError::Corrupted { + detail: format!( + "encoding value for key {}: {e}", + String::from_utf8_lossy(key) + ), + })?; } } @@ -1220,212 +1089,33 @@ pub(crate) fn read_entry( // expires_at_ms: if ttl_ms > 0 it's already absolute unix millis let expires_at_ms = if ttl_ms > 0 { ttl_ms as u64 } else { 0 }; - // Value + // Value -- strings inline, collections via the shared value codec. let value = match type_tag { TYPE_STRING => { let data = read_bytes(cursor)?; RedisValue::String(data) } - TYPE_HASH => { - let count = read_u32(cursor)? as usize; - validate_count(cursor, count, 8, "hash")?; // min 4+4 bytes per field+value length - let mut map = HashMap::with_capacity(count); - for _ in 0..count { - let field = read_bytes(cursor)?; - let val = read_bytes(cursor)?; - map.insert(field, val); - } - if has_hash_ttl_trailer { - let ttl_count = read_u32(cursor)? as usize; - validate_count(cursor, ttl_count, 12, "hash_ttls")?; - if ttl_count > 0 { - let mut ttls = HashMap::with_capacity(ttl_count); - for _ in 0..ttl_count { - let field = read_bytes(cursor)?; - let mut ttl_buf = [0u8; 8]; - cursor.read_exact(&mut ttl_buf)?; - let ttl_ms = u64::from_le_bytes(ttl_buf); - ttls.insert(field, ttl_ms); - } - // min_expiry_ms is purely in-memory; recompute from the - // decoded ttls map (not stored in the RDB file). - let min_expiry_ms = ttls.values().copied().min().unwrap_or(u64::MAX); - RedisValue::HashWithTtl { - fields: map, - ttls, - min_expiry_ms, - } - } else { - RedisValue::Hash(map) - } + _ => { + let value_type = match type_tag { + TYPE_HASH => ValueType::Hash, + TYPE_LIST => ValueType::List, + TYPE_SET => ValueType::Set, + TYPE_SORTED_SET => ValueType::ZSet, + TYPE_STREAM => ValueType::Stream, + _ => return Err(RdbError::UnsupportedType { type_tag }.into()), + }; + // v1 files predate the hash-TTL trailer; v2 requires it. + let trailer = if has_hash_ttl_trailer { + HashTtlTrailer::Required } else { - RedisValue::Hash(map) - } - } - TYPE_LIST => { - let count = read_u32(cursor)? as usize; - validate_count(cursor, count, 4, "list")?; // min 4 bytes per element length - let mut list = VecDeque::with_capacity(count); - for _ in 0..count { - list.push_back(read_bytes(cursor)?); - } - RedisValue::List(list) - } - TYPE_SET => { - let count = read_u32(cursor)? as usize; - validate_count(cursor, count, 4, "set")?; // min 4 bytes per element length - let mut set = HashSet::with_capacity(count); - for _ in 0..count { - set.insert(read_bytes(cursor)?); - } - RedisValue::Set(set) - } - TYPE_SORTED_SET => { - let count = read_u32(cursor)? as usize; - validate_count(cursor, count, 12, "sorted_set")?; // min 4 (member len) + 8 (f64 score) - let mut members = HashMap::with_capacity(count); - let mut tree = BPTree::new(); - for _ in 0..count { - let member = read_bytes(cursor)?; - let mut score_buf = [0u8; 8]; - cursor.read_exact(&mut score_buf)?; - let score = f64::from_le_bytes(score_buf); - members.insert(member.clone(), score); - tree.insert(OrderedFloat(score), member); - } - RedisValue::SortedSetBPTree { tree, members } - } - TYPE_STREAM => { - let mut entry_count_buf = [0u8; 8]; - cursor.read_exact(&mut entry_count_buf)?; - let entry_count = u64::from_le_bytes(entry_count_buf) as usize; - - let mut last_id_ms_buf = [0u8; 8]; - let mut last_id_seq_buf = [0u8; 8]; - cursor.read_exact(&mut last_id_ms_buf)?; - cursor.read_exact(&mut last_id_seq_buf)?; - let last_id = StreamId { - ms: u64::from_le_bytes(last_id_ms_buf), - seq: u64::from_le_bytes(last_id_seq_buf), + HashTtlTrailer::Absent }; - - let mut stream = StreamData::new(); - stream.last_id = last_id; - - validate_count(cursor, entry_count, 20, "stream_entries")?; // min 16 (id) + 4 (field_count) - for _ in 0..entry_count { - let mut ms_buf = [0u8; 8]; - let mut seq_buf = [0u8; 8]; - cursor.read_exact(&mut ms_buf)?; - cursor.read_exact(&mut seq_buf)?; - let id = StreamId { - ms: u64::from_le_bytes(ms_buf), - seq: u64::from_le_bytes(seq_buf), - }; - let field_count = read_u32(cursor)? as usize; - validate_count(cursor, field_count, 8, "stream_fields")?; - let mut fields = Vec::with_capacity(field_count); - for _ in 0..field_count { - let field = read_bytes(cursor)?; - let value = read_bytes(cursor)?; - fields.push((field, value)); - } - stream.entries.insert(id, fields); - stream.length += 1; - } - - // Consumer groups - let group_count = read_u32(cursor)? as usize; - // min 4 (name len) + 16 (last_delivered_id) + 4 (pel_count) + 4 (consumer_count) - validate_count(cursor, group_count, 28, "stream_groups")?; - for _ in 0..group_count { - let group_name = read_bytes(cursor)?; - let mut gld_ms = [0u8; 8]; - let mut gld_seq = [0u8; 8]; - cursor.read_exact(&mut gld_ms)?; - cursor.read_exact(&mut gld_seq)?; - let last_delivered_id = StreamId { - ms: u64::from_le_bytes(gld_ms), - seq: u64::from_le_bytes(gld_seq), - }; - - let pel_count = read_u32(cursor)? as usize; - // min 16 (StreamId) + 4 (consumer name len) + 16 (delivery_time+delivery_count) - validate_count(cursor, pel_count, 36, "stream_pel")?; - let mut pel = BTreeMap::new(); - for _ in 0..pel_count { - let mut pid_ms = [0u8; 8]; - let mut pid_seq = [0u8; 8]; - cursor.read_exact(&mut pid_ms)?; - cursor.read_exact(&mut pid_seq)?; - let pid = StreamId { - ms: u64::from_le_bytes(pid_ms), - seq: u64::from_le_bytes(pid_seq), - }; - let consumer_name = read_bytes(cursor)?; - let mut dt_buf = [0u8; 8]; - let mut dc_buf = [0u8; 8]; - cursor.read_exact(&mut dt_buf)?; - cursor.read_exact(&mut dc_buf)?; - pel.insert( - pid, - crate::storage::stream::PendingEntry { - consumer: consumer_name, - delivery_time: u64::from_le_bytes(dt_buf), - delivery_count: u64::from_le_bytes(dc_buf), - }, - ); - } - - let consumer_count = read_u32(cursor)? as usize; - // min 4 (name len) + 8 (seen_time) + 4 (pending_count) - validate_count(cursor, consumer_count, 16, "stream_consumers")?; - let mut consumers = HashMap::new(); - for _ in 0..consumer_count { - let cname = read_bytes(cursor)?; - let mut st_buf = [0u8; 8]; - cursor.read_exact(&mut st_buf)?; - let seen_time = u64::from_le_bytes(st_buf); - let pending_count = read_u32(cursor)? as usize; - // min 16 (StreamId) - validate_count(cursor, pending_count, 16, "stream_pending")?; - let mut pending = BTreeMap::new(); - for _ in 0..pending_count { - let mut cid_ms = [0u8; 8]; - let mut cid_seq = [0u8; 8]; - cursor.read_exact(&mut cid_ms)?; - cursor.read_exact(&mut cid_seq)?; - pending.insert( - StreamId { - ms: u64::from_le_bytes(cid_ms), - seq: u64::from_le_bytes(cid_seq), - }, - (), - ); - } - consumers.insert( - cname.clone(), - crate::storage::stream::Consumer { - name: cname, - pending, - seen_time, - }, - ); - } - - stream.groups.insert( - group_name, - crate::storage::stream::ConsumerGroup { - last_delivered_id, - pel, - consumers, - }, - ); - } - - RedisValue::Stream(Box::new(stream)) + value_codec::decode_value_body(cursor, value_type, trailer).map_err(|e| { + MoonError::from(RdbError::Corrupted { + detail: e.to_string(), + }) + })? } - _ => return Err(RdbError::UnsupportedType { type_tag }.into()), }; // Use current_secs() as base_ts for loaded entries (matches Database::new()) diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs index e40e2405..bce27a76 100644 --- a/src/storage/eviction.rs +++ b/src/storage/eviction.rs @@ -892,20 +892,7 @@ fn evict_batch_durable_no_aof( let (value_type, value_bytes): (ValueType, &[u8]) = match val_ref { RedisValueRef::String(s) => (ValueType::String, s), ref other => { - let vt = match other { - RedisValueRef::Hash(_) - | RedisValueRef::HashListpack(_) - | RedisValueRef::HashWithTtl { .. } => ValueType::Hash, - RedisValueRef::List(_) | RedisValueRef::ListListpack(_) => ValueType::List, - RedisValueRef::Set(_) - | RedisValueRef::SetListpack(_) - | RedisValueRef::SetIntset(_) => ValueType::Set, - RedisValueRef::SortedSet { .. } - | RedisValueRef::SortedSetBPTree { .. } - | RedisValueRef::SortedSetListpack(_) => ValueType::ZSet, - RedisValueRef::Stream(_) => ValueType::Stream, - RedisValueRef::String(_) => unreachable!(), - }; + let vt = kv_spill::value_type_of(other); collection_buf = kv_serde::serialize_collection(other).unwrap_or_default(); (vt, collection_buf.as_slice()) } @@ -1035,20 +1022,7 @@ fn evict_one_async_spill( let (value_type, value_bytes): (ValueType, &[u8]) = match val_ref { RedisValueRef::String(s) => (ValueType::String, s), ref other => { - let vt = match other { - RedisValueRef::Hash(_) - | RedisValueRef::HashListpack(_) - | RedisValueRef::HashWithTtl { .. } => ValueType::Hash, - RedisValueRef::List(_) | RedisValueRef::ListListpack(_) => ValueType::List, - RedisValueRef::Set(_) - | RedisValueRef::SetListpack(_) - | RedisValueRef::SetIntset(_) => ValueType::Set, - RedisValueRef::SortedSet { .. } - | RedisValueRef::SortedSetBPTree { .. } - | RedisValueRef::SortedSetListpack(_) => ValueType::ZSet, - RedisValueRef::Stream(_) => ValueType::Stream, - RedisValueRef::String(_) => unreachable!(), - }; + let vt = kv_spill::value_type_of(other); collection_buf = kv_serde::serialize_collection(other).unwrap_or_default(); (vt, collection_buf.as_slice()) } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 08d2b9a5..ef8e7588 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -17,6 +17,7 @@ pub mod listpack; pub mod stream; pub mod tier; pub mod tiered; +pub mod value_codec; pub use db::Database; pub use entry::{Entry, RedisValue}; diff --git a/src/storage/tiered/kv_serde.rs b/src/storage/tiered/kv_serde.rs index aaf71668..e4c7c8d4 100644 --- a/src/storage/tiered/kv_serde.rs +++ b/src/storage/tiered/kv_serde.rs @@ -1,398 +1,61 @@ //! Collection serialization/deserialization for KV disk offload. //! -//! Converts between `RedisValueRef` / `RedisValue` and a compact binary format -//! for storage in KvLeafPage entries. The wire format mirrors rdb.rs but omits -//! the type tag prefix (stored separately in the KvLeafPage entry header). +//! Thin wrapper over [`crate::storage::value_codec`] — the wire format is the +//! same value-body format RDB uses, minus the type tag prefix (stored +//! separately in the KvLeafPage entry header). Before the W1 unification this +//! file was a ~330-line hand-maintained copy of `rdb::write_entry`'s value +//! section; the codec module is now the single implementation. -use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; -use std::io::{self, Cursor, Read, Write}; - -use bytes::Bytes; -use ordered_float::OrderedFloat; +use std::io::Cursor; use crate::persistence::kv_page::ValueType; -use crate::storage::bptree::BPTree; use crate::storage::compact_value::RedisValueRef; use crate::storage::entry::RedisValue; -use crate::storage::stream::{ - Consumer, ConsumerGroup, PendingEntry, Stream as StreamData, StreamId, -}; - -// ── Helpers (local, avoids coupling to rdb module internals) ── - -#[inline] -fn write_len_bytes(buf: &mut Vec, data: &[u8]) { - buf.extend_from_slice(&(data.len() as u32).to_le_bytes()); - buf.extend_from_slice(data); -} - -#[inline] -fn read_len_bytes(cursor: &mut Cursor<&[u8]>) -> io::Result { - let mut len_buf = [0u8; 4]; - cursor.read_exact(&mut len_buf)?; - let len = u32::from_le_bytes(len_buf) as usize; - let pos = cursor.position() as usize; - let data = cursor.get_ref(); - if pos + len > data.len() { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "truncated data", - )); - } - let result = Bytes::copy_from_slice(&data[pos..pos + len]); - cursor.set_position((pos + len) as u64); - Ok(result) -} - -#[inline] -fn read_u32_le(cursor: &mut Cursor<&[u8]>) -> io::Result { - let mut buf = [0u8; 4]; - cursor.read_exact(&mut buf)?; - Ok(u32::from_le_bytes(buf)) -} - -#[inline] -fn read_u64_le(cursor: &mut Cursor<&[u8]>) -> io::Result { - let mut buf = [0u8; 8]; - cursor.read_exact(&mut buf)?; - Ok(u64::from_le_bytes(buf)) -} - -#[inline] -fn read_f64_le(cursor: &mut Cursor<&[u8]>) -> io::Result { - let mut buf = [0u8; 8]; - cursor.read_exact(&mut buf)?; - Ok(f64::from_le_bytes(buf)) -} - -// ── Public API ── +use crate::storage::value_codec::{self, HashTtlTrailer}; /// Serialize a collection `RedisValueRef` into bytes for KvLeafPage storage. /// -/// Uses a binary format identical to rdb.rs `write_entry` value section -/// (u32-length-prefixed fields) but without the type tag prefix. -/// -/// Returns `None` for String type (strings go directly as value bytes). +/// Returns `None` for String type (strings go directly as value bytes) and +/// for values that fail to encode (in-memory corruption, e.g. an unparseable +/// sorted-set listpack score — logged loudly; the caller must treat this as +/// "cannot spill", never as an empty value). pub fn serialize_collection(value: &RedisValueRef<'_>) -> Option> { + if matches!(value, RedisValueRef::String(_)) { + return None; + } let mut buf = Vec::with_capacity(256); - match value { - RedisValueRef::String(_) => return None, - - RedisValueRef::Hash(map) => { - buf.write_all(&(map.len() as u32).to_le_bytes()).ok()?; - for (field, val) in map.iter() { - write_len_bytes(&mut buf, field); - write_len_bytes(&mut buf, val); - } - // Hash-field TTL trailer: ttl_count=0 for plain Hash. - // Same layout as the RDB v2 trailer (see src/persistence/rdb.rs). - buf.write_all(&0u32.to_le_bytes()).ok()?; - } - RedisValueRef::HashWithTtl { fields, ttls, .. } => { - buf.write_all(&(fields.len() as u32).to_le_bytes()).ok()?; - for (field, val) in fields.iter() { - write_len_bytes(&mut buf, field); - write_len_bytes(&mut buf, val); - } - // Hash-field TTL trailer: ttl_count + [field, ttl_ms u64]*. - buf.write_all(&(ttls.len() as u32).to_le_bytes()).ok()?; - for (field, ttl_ms) in ttls.iter() { - write_len_bytes(&mut buf, field); - buf.write_all(&ttl_ms.to_le_bytes()).ok()?; - } - } - RedisValueRef::HashListpack(lp) => { - let map = lp.to_hash_map(); - buf.write_all(&(map.len() as u32).to_le_bytes()).ok()?; - for (field, val) in &map { - write_len_bytes(&mut buf, field); - write_len_bytes(&mut buf, val); - } - // Listpack hashes can never carry per-field TTLs. - buf.write_all(&0u32.to_le_bytes()).ok()?; - } - RedisValueRef::List(list) => { - buf.write_all(&(list.len() as u32).to_le_bytes()).ok()?; - for elem in list.iter() { - write_len_bytes(&mut buf, elem); - } - } - RedisValueRef::ListListpack(lp) => { - let list = lp.to_vec_deque(); - buf.write_all(&(list.len() as u32).to_le_bytes()).ok()?; - for elem in &list { - write_len_bytes(&mut buf, elem); - } - } - RedisValueRef::Set(set) => { - buf.write_all(&(set.len() as u32).to_le_bytes()).ok()?; - for member in set.iter() { - write_len_bytes(&mut buf, member); - } - } - RedisValueRef::SetListpack(lp) => { - let set = lp.to_hash_set(); - buf.write_all(&(set.len() as u32).to_le_bytes()).ok()?; - for member in &set { - write_len_bytes(&mut buf, member); - } - } - RedisValueRef::SetIntset(is) => { - let set = is.to_hash_set(); - buf.write_all(&(set.len() as u32).to_le_bytes()).ok()?; - for member in &set { - write_len_bytes(&mut buf, member); - } - } - RedisValueRef::SortedSet { members, .. } - | RedisValueRef::SortedSetBPTree { members, .. } => { - buf.write_all(&(members.len() as u32).to_le_bytes()).ok()?; - for (member, score) in members.iter() { - write_len_bytes(&mut buf, member); - buf.write_all(&score.to_le_bytes()).ok()?; - } - } - RedisValueRef::SortedSetListpack(lp) => { - let pairs: Vec<_> = lp.iter_pairs().collect(); - let count_pos = buf.len(); - buf.write_all(&0u32.to_le_bytes()).ok()?; - let mut count: u32 = 0; - for (member_entry, score_entry) in &pairs { - let member_bytes = member_entry.as_bytes(); - let score_bytes = score_entry.as_bytes(); - let score: f64 = std::str::from_utf8(&score_bytes) - .ok() - .and_then(|s| s.parse().ok()) - .unwrap_or(0.0); - write_len_bytes(&mut buf, &member_bytes); - buf.write_all(&score.to_le_bytes()).ok()?; - count += 1; - } - buf[count_pos..count_pos + 4].copy_from_slice(&count.to_le_bytes()); - } - RedisValueRef::Stream(stream) => { - // Entry count + last_id - buf.write_all(&(stream.entries.len() as u64).to_le_bytes()) - .ok()?; - buf.write_all(&stream.last_id.ms.to_le_bytes()).ok()?; - buf.write_all(&stream.last_id.seq.to_le_bytes()).ok()?; - // Entries - for (id, fields) in &stream.entries { - buf.write_all(&id.ms.to_le_bytes()).ok()?; - buf.write_all(&id.seq.to_le_bytes()).ok()?; - buf.write_all(&(fields.len() as u32).to_le_bytes()).ok()?; - for (field, value) in fields { - write_len_bytes(&mut buf, field); - write_len_bytes(&mut buf, value); - } - } - // Consumer groups - buf.write_all(&(stream.groups.len() as u32).to_le_bytes()) - .ok()?; - for (group_name, group) in &stream.groups { - write_len_bytes(&mut buf, group_name); - buf.write_all(&group.last_delivered_id.ms.to_le_bytes()) - .ok()?; - buf.write_all(&group.last_delivered_id.seq.to_le_bytes()) - .ok()?; - // PEL - buf.write_all(&(group.pel.len() as u32).to_le_bytes()) - .ok()?; - for (id, pe) in &group.pel { - buf.write_all(&id.ms.to_le_bytes()).ok()?; - buf.write_all(&id.seq.to_le_bytes()).ok()?; - write_len_bytes(&mut buf, &pe.consumer); - buf.write_all(&pe.delivery_time.to_le_bytes()).ok()?; - buf.write_all(&pe.delivery_count.to_le_bytes()).ok()?; - } - // Consumers - buf.write_all(&(group.consumers.len() as u32).to_le_bytes()) - .ok()?; - for (cname, consumer) in &group.consumers { - write_len_bytes(&mut buf, cname); - buf.write_all(&consumer.seen_time.to_le_bytes()).ok()?; - buf.write_all(&(consumer.pending.len() as u32).to_le_bytes()) - .ok()?; - for (id, _) in &consumer.pending { - buf.write_all(&id.ms.to_le_bytes()).ok()?; - buf.write_all(&id.seq.to_le_bytes()).ok()?; - } - } - } + match value_codec::encode_value_body(value, &mut buf) { + Ok(()) => Some(buf), + Err(e) => { + tracing::error!( + error = %e, + "kv_serde: refusing to serialize corrupt value for spill" + ); + None } } - Some(buf) } /// Deserialize collection bytes back into a `RedisValue`. /// /// `value_type` determines which collection format to parse. -/// Returns `None` for String type or on parse failure. +/// Returns `None` for String type or on parse failure (logged). Legacy +/// pre-trailer hash blobs (old in-process spill) decode as plain `Hash` — +/// see [`HashTtlTrailer::Lenient`]. pub fn deserialize_collection(data: &[u8], value_type: ValueType) -> Option { if value_type == ValueType::String { return None; } let mut cursor = Cursor::new(data); - match value_type { - ValueType::String => None, - ValueType::Hash => { - let count = read_u32_le(&mut cursor).ok()? as usize; - let mut map = HashMap::with_capacity(count); - for _ in 0..count { - let field = read_len_bytes(&mut cursor).ok()?; - let val = read_len_bytes(&mut cursor).ok()?; - map.insert(field, val); - } - // Per-field TTL trailer (phase 200). ttl_count = 0 → plain Hash. - // Pre-trailer blobs (old in-process spill) lack the count and - // EOF here — treat as plain Hash for graceful migration. - match read_u32_le(&mut cursor) { - Ok(0) => Some(RedisValue::Hash(map)), - Ok(ttl_count) => { - let mut ttls = HashMap::with_capacity(ttl_count as usize); - for _ in 0..ttl_count as usize { - let field = read_len_bytes(&mut cursor).ok()?; - let ttl_ms = read_u64_le(&mut cursor).ok()?; - ttls.insert(field, ttl_ms); - } - // min_expiry_ms is purely in-memory; recompute after decode. - let min_expiry_ms = ttls.values().copied().min().unwrap_or(u64::MAX); - Some(RedisValue::HashWithTtl { - fields: map, - ttls, - min_expiry_ms, - }) - } - Err(_) => Some(RedisValue::Hash(map)), - } - } - ValueType::List => { - let count = read_u32_le(&mut cursor).ok()? as usize; - let mut list = VecDeque::with_capacity(count); - for _ in 0..count { - list.push_back(read_len_bytes(&mut cursor).ok()?); - } - Some(RedisValue::List(list)) - } - ValueType::Set => { - let count = read_u32_le(&mut cursor).ok()? as usize; - let mut set = HashSet::with_capacity(count); - for _ in 0..count { - set.insert(read_len_bytes(&mut cursor).ok()?); - } - Some(RedisValue::Set(set)) - } - ValueType::ZSet => { - let count = read_u32_le(&mut cursor).ok()? as usize; - let mut members = HashMap::with_capacity(count); - let mut tree = BPTree::new(); - for _ in 0..count { - let member = read_len_bytes(&mut cursor).ok()?; - let score = read_f64_le(&mut cursor).ok()?; - members.insert(member.clone(), score); - tree.insert(OrderedFloat(score), member); - } - Some(RedisValue::SortedSetBPTree { tree, members }) - } - ValueType::Stream => { - let entry_count = read_u64_le(&mut cursor).ok()? as usize; - let last_id_ms = read_u64_le(&mut cursor).ok()?; - let last_id_seq = read_u64_le(&mut cursor).ok()?; - let last_id = StreamId { - ms: last_id_ms, - seq: last_id_seq, - }; - - let mut stream = StreamData::new(); - stream.last_id = last_id; - - for _ in 0..entry_count { - let ms = read_u64_le(&mut cursor).ok()?; - let seq = read_u64_le(&mut cursor).ok()?; - let id = StreamId { ms, seq }; - let field_count = read_u32_le(&mut cursor).ok()? as usize; - let mut fields = Vec::with_capacity(field_count); - for _ in 0..field_count { - let field = read_len_bytes(&mut cursor).ok()?; - let value = read_len_bytes(&mut cursor).ok()?; - fields.push((field, value)); - } - stream.entries.insert(id, fields); - stream.length += 1; - } - - // Consumer groups - let group_count = read_u32_le(&mut cursor).ok()? as usize; - for _ in 0..group_count { - let group_name = read_len_bytes(&mut cursor).ok()?; - let gld_ms = read_u64_le(&mut cursor).ok()?; - let gld_seq = read_u64_le(&mut cursor).ok()?; - let last_delivered_id = StreamId { - ms: gld_ms, - seq: gld_seq, - }; - - let pel_count = read_u32_le(&mut cursor).ok()? as usize; - let mut pel = BTreeMap::new(); - for _ in 0..pel_count { - let pid_ms = read_u64_le(&mut cursor).ok()?; - let pid_seq = read_u64_le(&mut cursor).ok()?; - let pid = StreamId { - ms: pid_ms, - seq: pid_seq, - }; - let consumer_name = read_len_bytes(&mut cursor).ok()?; - let delivery_time = read_u64_le(&mut cursor).ok()?; - let delivery_count = read_u64_le(&mut cursor).ok()?; - pel.insert( - pid, - PendingEntry { - consumer: consumer_name, - delivery_time, - delivery_count, - }, - ); - } - - let consumer_count = read_u32_le(&mut cursor).ok()? as usize; - let mut consumers = HashMap::new(); - for _ in 0..consumer_count { - let cname = read_len_bytes(&mut cursor).ok()?; - let seen_time = read_u64_le(&mut cursor).ok()?; - let pending_count = read_u32_le(&mut cursor).ok()? as usize; - let mut pending = BTreeMap::new(); - for _ in 0..pending_count { - let cid_ms = read_u64_le(&mut cursor).ok()?; - let cid_seq = read_u64_le(&mut cursor).ok()?; - pending.insert( - StreamId { - ms: cid_ms, - seq: cid_seq, - }, - (), - ); - } - consumers.insert( - cname.clone(), - Consumer { - name: cname, - pending, - seen_time, - }, - ); - } - - stream.groups.insert( - group_name, - ConsumerGroup { - last_delivered_id, - pel, - consumers, - }, - ); - } - - Some(RedisValue::Stream(Box::new(stream))) + match value_codec::decode_value_body(&mut cursor, value_type, HashTtlTrailer::Lenient) { + Ok(v) => Some(v), + Err(e) => { + tracing::warn!( + error = %e, + ?value_type, + "kv_serde: corrupt spill value body" + ); + None } } } @@ -400,6 +63,11 @@ pub fn deserialize_collection(data: &[u8], value_type: ValueType) -> Option ValueType { - match val { - RedisValueRef::String(_) => ValueType::String, - RedisValueRef::Hash(_) - | RedisValueRef::HashListpack(_) - | RedisValueRef::HashWithTtl { .. } => ValueType::Hash, - RedisValueRef::List(_) | RedisValueRef::ListListpack(_) => ValueType::List, - RedisValueRef::Set(_) | RedisValueRef::SetListpack(_) | RedisValueRef::SetIntset(_) => { - ValueType::Set - } - RedisValueRef::SortedSet { .. } - | RedisValueRef::SortedSetBPTree { .. } - | RedisValueRef::SortedSetListpack(_) => ValueType::ZSet, - RedisValueRef::Stream(_) => ValueType::Stream, - } -} +/// Re-exported from the shared value codec (W1 unification) — used by every +/// spill entry point and by the sync eviction path to populate +/// `ColdLocation::value_type` (#364). +pub use crate::storage::value_codec::value_type_of; /// Spill a single evicted KV entry to a DataFile on disk. /// diff --git a/src/storage/value_codec.rs b/src/storage/value_codec.rs new file mode 100644 index 00000000..0bdc5863 --- /dev/null +++ b/src/storage/value_codec.rs @@ -0,0 +1,886 @@ +//! Single value-body codec shared by RDB snapshots and KV disk-offload spill. +//! +//! Historically this format existed twice — `rdb::write_entry`'s value section +//! and `kv_serde::serialize_collection` — kept bit-compatible by hand ("format +//! identical to rdb.rs" was a doc comment, not a compiler guarantee). Every +//! format evolution (e.g. the v2 hash-TTL trailer) had to be applied to both +//! copies. This module is now the only implementation; both call sites +//! delegate here. +//! +//! # Wire format (little-endian throughout) +//! +//! The *value body* excludes any type tag / key / entry-TTL framing — those +//! belong to the container (RDB entry header or `KvLeafPage` slot header). +//! +//! | Type | Layout | +//! |--------|-----------------------------------------------------------------| +//! | Hash | `count:u32` then `count` × (`len:u32 field` `len:u32 value`), then TTL trailer `ttl_count:u32` + `ttl_count` × (`len:u32 field` `ttl_ms:u64`) | +//! | List | `count:u32` then `count` × `len:u32 elem` | +//! | Set | `count:u32` then `count` × `len:u32 member` | +//! | ZSet | `count:u32` then `count` × (`len:u32 member` `score:f64`) | +//! | Stream | `entry_count:u64` `last_id.ms:u64` `last_id.seq:u64`, entries, then `group_count:u32` + groups (see code) | +//! +//! Strings have no collection body — the container stores their bytes +//! directly. [`encode_value_body`] returns [`ValueCodecError::StringHasNoBody`] +//! if handed one, so a caller can never silently persist an empty body for a +//! string. +//! +//! # Decode hardening +//! +//! Every count field is validated against the remaining input length *before* +//! allocation (`validate_count`), mirroring the RDB-DoS fix — previously the +//! spill decoder (`kv_serde`) was missed by that fix and fed attacker-length +//! counts straight into `Vec::with_capacity`. + +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; +use std::io::{Cursor, Read}; + +use bytes::Bytes; +use ordered_float::OrderedFloat; + +use crate::persistence::kv_page::ValueType; +use crate::storage::bptree::BPTree; +use crate::storage::compact_value::RedisValueRef; +use crate::storage::entry::RedisValue; +use crate::storage::stream::{ + Consumer, ConsumerGroup, PendingEntry, Stream as StreamData, StreamId, +}; + +/// How the hash per-field-TTL trailer is handled on decode. +/// +/// Writers always emit the trailer (v2 format). Readers differ by container: +/// RDB files are versioned (v1 = no trailer, v2 = trailer required), while +/// spill blobs are unversioned — pre-trailer blobs simply end after the +/// fields, so a missing trailer is forgiven. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HashTtlTrailer { + /// v1 RDB: no trailer bytes exist; do not attempt to read one. + Absent, + /// v2 RDB: trailer must be present; truncation is corruption. + Required, + /// Spill blobs: attempt the trailer, treat clean EOF at the trailer + /// count as a legacy pre-trailer blob (plain `Hash`). + Lenient, +} + +/// Error from [`encode_value_body`] / [`decode_value_body`]. +#[derive(Debug, thiserror::Error)] +pub enum ValueCodecError { + /// Strings carry no collection body; the container stores their bytes. + #[error("string values have no collection body")] + StringHasNoBody, + /// A sorted-set listpack held a score that does not parse as f64. + /// + /// This is fail-closed by design: the old duplicated encoders wrote + /// `0.0` for such scores (silent corruption of the persisted zset). + #[error("unparseable sorted-set listpack score (in-memory corruption?)")] + CorruptScore, + /// Truncated or self-inconsistent input. + #[error("corrupted value body: {detail}")] + Corrupted { detail: String }, +} + +impl ValueCodecError { + fn truncated(what: &str) -> Self { + ValueCodecError::Corrupted { + detail: format!("truncated reading {what}"), + } + } +} + +/// On-disk [`ValueType`] tag for a hot value. +/// +/// Exhaustive over `RedisValueRef` by design: adding a new value variant +/// without a `ValueType` mapping must be a compile error, never a silent +/// mis-typed spill. Single source of truth — the former copies in +/// `kv_spill` and (inline, twice) in `eviction.rs` delegate here. +pub fn value_type_of(val: &RedisValueRef) -> ValueType { + match val { + RedisValueRef::String(_) => ValueType::String, + RedisValueRef::Hash(_) + | RedisValueRef::HashListpack(_) + | RedisValueRef::HashWithTtl { .. } => ValueType::Hash, + RedisValueRef::List(_) | RedisValueRef::ListListpack(_) => ValueType::List, + RedisValueRef::Set(_) | RedisValueRef::SetListpack(_) | RedisValueRef::SetIntset(_) => { + ValueType::Set + } + RedisValueRef::SortedSet { .. } + | RedisValueRef::SortedSetBPTree { .. } + | RedisValueRef::SortedSetListpack(_) => ValueType::ZSet, + RedisValueRef::Stream(_) => ValueType::Stream, + } +} + +// ── Encode helpers ── + +#[inline] +fn put_len_bytes(buf: &mut Vec, data: &[u8]) { + buf.extend_from_slice(&(data.len() as u32).to_le_bytes()); + buf.extend_from_slice(data); +} + +#[inline] +fn put_u32(buf: &mut Vec, v: u32) { + buf.extend_from_slice(&v.to_le_bytes()); +} + +#[inline] +fn put_u64(buf: &mut Vec, v: u64) { + buf.extend_from_slice(&v.to_le_bytes()); +} + +#[inline] +fn put_f64(buf: &mut Vec, v: f64) { + buf.extend_from_slice(&v.to_le_bytes()); +} + +/// Encode the collection body of `value` into `buf` (appended). +/// +/// Compact variants (listpack / intset) expand to the element-level format — +/// the encoding is canonical per logical type, so a value round-trips to its +/// full-size representation regardless of its in-memory encoding. +pub fn encode_value_body( + value: &RedisValueRef<'_>, + buf: &mut Vec, +) -> Result<(), ValueCodecError> { + match value { + RedisValueRef::String(_) => return Err(ValueCodecError::StringHasNoBody), + + RedisValueRef::Hash(map) => { + put_u32(buf, map.len() as u32); + for (field, val) in map.iter() { + put_len_bytes(buf, field); + put_len_bytes(buf, val); + } + // v2 trailer: plain hashes carry no per-field TTLs. + put_u32(buf, 0); + } + RedisValueRef::HashWithTtl { fields, ttls, .. } => { + put_u32(buf, fields.len() as u32); + for (field, val) in fields.iter() { + put_len_bytes(buf, field); + put_len_bytes(buf, val); + } + // v2 trailer: per-field TTL sidecar. + put_u32(buf, ttls.len() as u32); + for (field, ttl_ms) in ttls.iter() { + put_len_bytes(buf, field); + put_u64(buf, *ttl_ms); + } + } + RedisValueRef::HashListpack(lp) => { + let map = lp.to_hash_map(); + put_u32(buf, map.len() as u32); + for (field, val) in &map { + put_len_bytes(buf, field); + put_len_bytes(buf, val); + } + // Listpack hashes can never carry per-field TTLs. + put_u32(buf, 0); + } + RedisValueRef::List(list) => { + put_u32(buf, list.len() as u32); + for elem in list.iter() { + put_len_bytes(buf, elem); + } + } + RedisValueRef::ListListpack(lp) => { + let list = lp.to_vec_deque(); + put_u32(buf, list.len() as u32); + for elem in &list { + put_len_bytes(buf, elem); + } + } + RedisValueRef::Set(set) => { + put_u32(buf, set.len() as u32); + for member in set.iter() { + put_len_bytes(buf, member); + } + } + RedisValueRef::SetListpack(lp) => { + let set = lp.to_hash_set(); + put_u32(buf, set.len() as u32); + for member in &set { + put_len_bytes(buf, member); + } + } + RedisValueRef::SetIntset(is) => { + let set = is.to_hash_set(); + put_u32(buf, set.len() as u32); + for member in &set { + put_len_bytes(buf, member); + } + } + RedisValueRef::SortedSet { members, .. } + | RedisValueRef::SortedSetBPTree { members, .. } => { + put_u32(buf, members.len() as u32); + for (member, score) in members.iter() { + put_len_bytes(buf, member); + put_f64(buf, *score); + } + } + RedisValueRef::SortedSetListpack(lp) => { + let pairs: Vec<_> = lp.iter_pairs().collect(); + put_u32(buf, pairs.len() as u32); + for (member_entry, score_entry) in &pairs { + let member_bytes = member_entry.as_bytes(); + let score_bytes = score_entry.as_bytes(); + // Fail-closed: an unparseable score means in-memory + // corruption — refuse to persist rather than write 0.0. + let score: f64 = std::str::from_utf8(&score_bytes) + .ok() + .and_then(|s| s.parse().ok()) + .ok_or(ValueCodecError::CorruptScore)?; + put_len_bytes(buf, &member_bytes); + put_f64(buf, score); + } + } + RedisValueRef::Stream(stream) => { + put_u64(buf, stream.entries.len() as u64); + put_u64(buf, stream.last_id.ms); + put_u64(buf, stream.last_id.seq); + for (id, fields) in &stream.entries { + put_u64(buf, id.ms); + put_u64(buf, id.seq); + put_u32(buf, fields.len() as u32); + for (field, value) in fields { + put_len_bytes(buf, field); + put_len_bytes(buf, value); + } + } + put_u32(buf, stream.groups.len() as u32); + for (group_name, group) in &stream.groups { + put_len_bytes(buf, group_name); + put_u64(buf, group.last_delivered_id.ms); + put_u64(buf, group.last_delivered_id.seq); + put_u32(buf, group.pel.len() as u32); + for (id, pe) in &group.pel { + put_u64(buf, id.ms); + put_u64(buf, id.seq); + put_len_bytes(buf, &pe.consumer); + put_u64(buf, pe.delivery_time); + put_u64(buf, pe.delivery_count); + } + put_u32(buf, group.consumers.len() as u32); + for (cname, consumer) in &group.consumers { + put_len_bytes(buf, cname); + put_u64(buf, consumer.seen_time); + put_u32(buf, consumer.pending.len() as u32); + for (id, _) in &consumer.pending { + put_u64(buf, id.ms); + put_u64(buf, id.seq); + } + } + } + } + } + Ok(()) +} + +// ── Decode helpers ── + +#[inline] +fn get_u32(cursor: &mut Cursor<&[u8]>, what: &str) -> Result { + let mut buf = [0u8; 4]; + cursor + .read_exact(&mut buf) + .map_err(|_| ValueCodecError::truncated(what))?; + Ok(u32::from_le_bytes(buf)) +} + +#[inline] +fn get_u64(cursor: &mut Cursor<&[u8]>, what: &str) -> Result { + let mut buf = [0u8; 8]; + cursor + .read_exact(&mut buf) + .map_err(|_| ValueCodecError::truncated(what))?; + Ok(u64::from_le_bytes(buf)) +} + +#[inline] +fn get_f64(cursor: &mut Cursor<&[u8]>, what: &str) -> Result { + let mut buf = [0u8; 8]; + cursor + .read_exact(&mut buf) + .map_err(|_| ValueCodecError::truncated(what))?; + Ok(f64::from_le_bytes(buf)) +} + +#[inline] +fn get_len_bytes(cursor: &mut Cursor<&[u8]>, what: &str) -> Result { + let len = get_u32(cursor, what)? as usize; + let pos = cursor.position() as usize; + let data = cursor.get_ref(); + let remaining = data.len().saturating_sub(pos); + if len > remaining { + return Err(ValueCodecError::Corrupted { + detail: format!("{what}: length {len} exceeds remaining {remaining}"), + }); + } + let result = Bytes::copy_from_slice(&data[pos..pos + len]); + cursor.set_position((pos + len) as u64); + Ok(result) +} + +/// Validate a collection count against remaining input before allocating. +/// +/// `min_bytes_per_item` is the minimum wire size of one item; a count that +/// could not possibly fit in the remaining bytes is rejected up front so a +/// corrupt length field can never drive `Vec::with_capacity` (allocation +/// DoS — the fix rdb.rs received that the spill decoder previously lacked). +fn validate_count( + cursor: &Cursor<&[u8]>, + count: usize, + min_bytes_per_item: usize, + kind: &str, +) -> Result<(), ValueCodecError> { + let remaining = cursor + .get_ref() + .len() + .saturating_sub(cursor.position() as usize); + if min_bytes_per_item > 0 && count > remaining / min_bytes_per_item { + return Err(ValueCodecError::Corrupted { + detail: format!("{kind} count {count} exceeds remaining data ({remaining} bytes)"), + }); + } + Ok(()) +} + +/// Decode a collection body of logical type `value_type` from `cursor`. +/// +/// The cursor is left positioned at the first byte after the body, so RDB +/// entry streams can continue reading subsequent entries. +/// +/// Hashes decode to `Hash` or `HashWithTtl` depending on the TTL trailer — +/// see [`HashTtlTrailer`] for per-container trailer semantics. Sorted sets +/// always decode to `SortedSetBPTree` (the canonical full-size form). +pub fn decode_value_body( + cursor: &mut Cursor<&[u8]>, + value_type: ValueType, + trailer: HashTtlTrailer, +) -> Result { + match value_type { + ValueType::String => Err(ValueCodecError::StringHasNoBody), + ValueType::Hash => { + let count = get_u32(cursor, "hash count")? as usize; + validate_count(cursor, count, 8, "hash")?; + let mut map = HashMap::with_capacity(count); + for _ in 0..count { + let field = get_len_bytes(cursor, "hash field")?; + let val = get_len_bytes(cursor, "hash value")?; + map.insert(field, val); + } + let ttl_count = match trailer { + HashTtlTrailer::Absent => 0, + HashTtlTrailer::Required => get_u32(cursor, "hash ttl trailer")? as usize, + // Legacy pre-trailer spill blobs end exactly here; forgive a + // clean EOF at the trailer count (and only there). + HashTtlTrailer::Lenient => match get_u32(cursor, "hash ttl trailer") { + Ok(n) => n as usize, + Err(_) => 0, + }, + }; + if ttl_count == 0 { + return Ok(RedisValue::Hash(map)); + } + validate_count(cursor, ttl_count, 12, "hash_ttls")?; + let mut ttls = HashMap::with_capacity(ttl_count); + for _ in 0..ttl_count { + let field = get_len_bytes(cursor, "hash ttl field")?; + let ttl_ms = get_u64(cursor, "hash ttl value")?; + ttls.insert(field, ttl_ms); + } + // min_expiry_ms is purely in-memory; recompute after decode. + let min_expiry_ms = ttls.values().copied().min().unwrap_or(u64::MAX); + Ok(RedisValue::HashWithTtl { + fields: map, + ttls, + min_expiry_ms, + }) + } + ValueType::List => { + let count = get_u32(cursor, "list count")? as usize; + validate_count(cursor, count, 4, "list")?; + let mut list = VecDeque::with_capacity(count); + for _ in 0..count { + list.push_back(get_len_bytes(cursor, "list elem")?); + } + Ok(RedisValue::List(list)) + } + ValueType::Set => { + let count = get_u32(cursor, "set count")? as usize; + validate_count(cursor, count, 4, "set")?; + let mut set = HashSet::with_capacity(count); + for _ in 0..count { + set.insert(get_len_bytes(cursor, "set member")?); + } + Ok(RedisValue::Set(set)) + } + ValueType::ZSet => { + let count = get_u32(cursor, "zset count")? as usize; + validate_count(cursor, count, 12, "sorted_set")?; + let mut members = HashMap::with_capacity(count); + let mut tree = BPTree::new(); + for _ in 0..count { + let member = get_len_bytes(cursor, "zset member")?; + let score = get_f64(cursor, "zset score")?; + members.insert(member.clone(), score); + tree.insert(OrderedFloat(score), member); + } + Ok(RedisValue::SortedSetBPTree { tree, members }) + } + ValueType::Stream => { + let entry_count = get_u64(cursor, "stream entry count")? as usize; + let last_id = StreamId { + ms: get_u64(cursor, "stream last_id.ms")?, + seq: get_u64(cursor, "stream last_id.seq")?, + }; + let mut stream = StreamData::new(); + stream.last_id = last_id; + + validate_count(cursor, entry_count, 20, "stream_entries")?; + for _ in 0..entry_count { + let id = StreamId { + ms: get_u64(cursor, "stream id.ms")?, + seq: get_u64(cursor, "stream id.seq")?, + }; + let field_count = get_u32(cursor, "stream field count")? as usize; + validate_count(cursor, field_count, 8, "stream_fields")?; + let mut fields = Vec::with_capacity(field_count); + for _ in 0..field_count { + let field = get_len_bytes(cursor, "stream field")?; + let value = get_len_bytes(cursor, "stream value")?; + fields.push((field, value)); + } + stream.entries.insert(id, fields); + stream.length += 1; + } + + let group_count = get_u32(cursor, "stream group count")? as usize; + validate_count(cursor, group_count, 28, "stream_groups")?; + for _ in 0..group_count { + let group_name = get_len_bytes(cursor, "group name")?; + let last_delivered_id = StreamId { + ms: get_u64(cursor, "group last_delivered.ms")?, + seq: get_u64(cursor, "group last_delivered.seq")?, + }; + + let pel_count = get_u32(cursor, "group pel count")? as usize; + validate_count(cursor, pel_count, 36, "stream_pel")?; + let mut pel = BTreeMap::new(); + for _ in 0..pel_count { + let pid = StreamId { + ms: get_u64(cursor, "pel id.ms")?, + seq: get_u64(cursor, "pel id.seq")?, + }; + let consumer_name = get_len_bytes(cursor, "pel consumer")?; + let delivery_time = get_u64(cursor, "pel delivery_time")?; + let delivery_count = get_u64(cursor, "pel delivery_count")?; + pel.insert( + pid, + PendingEntry { + consumer: consumer_name, + delivery_time, + delivery_count, + }, + ); + } + + let consumer_count = get_u32(cursor, "group consumer count")? as usize; + validate_count(cursor, consumer_count, 16, "stream_consumers")?; + let mut consumers = HashMap::new(); + for _ in 0..consumer_count { + let cname = get_len_bytes(cursor, "consumer name")?; + let seen_time = get_u64(cursor, "consumer seen_time")?; + let pending_count = get_u32(cursor, "consumer pending count")? as usize; + validate_count(cursor, pending_count, 16, "stream_pending")?; + let mut pending = BTreeMap::new(); + for _ in 0..pending_count { + pending.insert( + StreamId { + ms: get_u64(cursor, "pending id.ms")?, + seq: get_u64(cursor, "pending id.seq")?, + }, + (), + ); + } + consumers.insert( + cname.clone(), + Consumer { + name: cname, + pending, + seen_time, + }, + ); + } + + stream.groups.insert( + group_name, + ConsumerGroup { + last_delivered_id, + pel, + consumers, + }, + ); + } + + Ok(RedisValue::Stream(Box::new(stream))) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::storage::listpack::Listpack; + + fn decode(data: &[u8], vt: ValueType, trailer: HashTtlTrailer) -> RedisValue { + let mut cursor = Cursor::new(data); + decode_value_body(&mut cursor, vt, trailer).expect("decode") + } + + /// Independent byte-level spec builders. These reconstruct the wire + /// format by hand so the tests pin the FORMAT, not "whatever the encoder + /// currently does" — any accidental format drift in the encoder fails + /// these before it can corrupt cross-version compatibility. + fn spec_len_bytes(out: &mut Vec, b: &[u8]) { + out.extend_from_slice(&(b.len() as u32).to_le_bytes()); + out.extend_from_slice(b); + } + + #[test] + fn golden_hash_single_field_matches_spec() { + let mut map = std::collections::HashMap::new(); + map.insert(Bytes::from_static(b"f1"), Bytes::from_static(b"v1")); + let mut buf = Vec::new(); + encode_value_body(&RedisValueRef::Hash(&map), &mut buf).unwrap(); + + let mut expect = Vec::new(); + expect.extend_from_slice(&1u32.to_le_bytes()); + spec_len_bytes(&mut expect, b"f1"); + spec_len_bytes(&mut expect, b"v1"); + expect.extend_from_slice(&0u32.to_le_bytes()); // v2 trailer, no TTLs + assert_eq!(buf, expect); + + match decode(&buf, ValueType::Hash, HashTtlTrailer::Lenient) { + RedisValue::Hash(m) => { + assert_eq!(m.get(b"f1".as_ref()).unwrap(), &Bytes::from_static(b"v1")); + } + other => panic!("expected Hash, got {other:?}"), + } + } + + #[test] + fn golden_hash_with_ttl_matches_spec() { + let mut fields = std::collections::HashMap::new(); + fields.insert(Bytes::from_static(b"f1"), Bytes::from_static(b"v1")); + let mut ttls = std::collections::HashMap::new(); + ttls.insert(Bytes::from_static(b"f1"), 5000u64); + let mut buf = Vec::new(); + encode_value_body( + &RedisValueRef::HashWithTtl { + fields: &fields, + ttls: &ttls, + min_expiry_ms: 5000, + }, + &mut buf, + ) + .unwrap(); + + let mut expect = Vec::new(); + expect.extend_from_slice(&1u32.to_le_bytes()); + spec_len_bytes(&mut expect, b"f1"); + spec_len_bytes(&mut expect, b"v1"); + expect.extend_from_slice(&1u32.to_le_bytes()); + spec_len_bytes(&mut expect, b"f1"); + expect.extend_from_slice(&5000u64.to_le_bytes()); + assert_eq!(buf, expect); + + match decode(&buf, ValueType::Hash, HashTtlTrailer::Required) { + RedisValue::HashWithTtl { + fields: f, + ttls: t, + min_expiry_ms, + } => { + assert_eq!(f.len(), 1); + assert_eq!(t.get(b"f1".as_ref()), Some(&5000)); + assert_eq!(min_expiry_ms, 5000); + } + other => panic!("expected HashWithTtl, got {other:?}"), + } + } + + #[test] + fn golden_list_matches_spec() { + let list: std::collections::VecDeque = + [Bytes::from_static(b"a"), Bytes::from_static(b"bb")] + .into_iter() + .collect(); + let mut buf = Vec::new(); + encode_value_body(&RedisValueRef::List(&list), &mut buf).unwrap(); + + let mut expect = Vec::new(); + expect.extend_from_slice(&2u32.to_le_bytes()); + spec_len_bytes(&mut expect, b"a"); + spec_len_bytes(&mut expect, b"bb"); + assert_eq!(buf, expect); + + match decode(&buf, ValueType::List, HashTtlTrailer::Lenient) { + RedisValue::List(l) => assert_eq!(l.len(), 2), + other => panic!("expected List, got {other:?}"), + } + } + + #[test] + fn golden_set_single_member_matches_spec() { + let mut set = std::collections::HashSet::new(); + set.insert(Bytes::from_static(b"m")); + let mut buf = Vec::new(); + encode_value_body(&RedisValueRef::Set(&set), &mut buf).unwrap(); + + let mut expect = Vec::new(); + expect.extend_from_slice(&1u32.to_le_bytes()); + spec_len_bytes(&mut expect, b"m"); + assert_eq!(buf, expect); + } + + #[test] + fn golden_zset_single_member_matches_spec() { + let mut members = std::collections::HashMap::new(); + members.insert(Bytes::from_static(b"m"), 1.5f64); + let mut tree = BPTree::new(); + tree.insert(OrderedFloat(1.5), Bytes::from_static(b"m")); + let mut buf = Vec::new(); + encode_value_body( + &RedisValueRef::SortedSetBPTree { + tree: &tree, + members: &members, + }, + &mut buf, + ) + .unwrap(); + + let mut expect = Vec::new(); + expect.extend_from_slice(&1u32.to_le_bytes()); + spec_len_bytes(&mut expect, b"m"); + expect.extend_from_slice(&1.5f64.to_le_bytes()); + assert_eq!(buf, expect); + + match decode(&buf, ValueType::ZSet, HashTtlTrailer::Lenient) { + RedisValue::SortedSetBPTree { members: m, .. } => { + assert_eq!(m.get(b"m".as_ref()), Some(&1.5)); + } + other => panic!("expected SortedSetBPTree, got {other:?}"), + } + } + + #[test] + fn golden_stream_matches_spec() { + let mut stream = StreamData::new(); + let id = StreamId { ms: 1, seq: 1 }; + stream.last_id = id; + stream.entries.insert( + id, + vec![(Bytes::from_static(b"f"), Bytes::from_static(b"v"))], + ); + stream.length = 1; + let mut pel = BTreeMap::new(); + pel.insert( + id, + PendingEntry { + consumer: Bytes::from_static(b"c"), + delivery_time: 7, + delivery_count: 2, + }, + ); + let mut pending = BTreeMap::new(); + pending.insert(id, ()); + let mut consumers = HashMap::new(); + consumers.insert( + Bytes::from_static(b"c"), + Consumer { + name: Bytes::from_static(b"c"), + pending, + seen_time: 9, + }, + ); + stream.groups.insert( + Bytes::from_static(b"g"), + ConsumerGroup { + last_delivered_id: id, + pel, + consumers, + }, + ); + + let mut buf = Vec::new(); + encode_value_body(&RedisValueRef::Stream(&stream), &mut buf).unwrap(); + + let mut expect = Vec::new(); + expect.extend_from_slice(&1u64.to_le_bytes()); // entry_count + expect.extend_from_slice(&1u64.to_le_bytes()); // last_id.ms + expect.extend_from_slice(&1u64.to_le_bytes()); // last_id.seq + expect.extend_from_slice(&1u64.to_le_bytes()); // entry id.ms + expect.extend_from_slice(&1u64.to_le_bytes()); // entry id.seq + expect.extend_from_slice(&1u32.to_le_bytes()); // field_count + spec_len_bytes(&mut expect, b"f"); + spec_len_bytes(&mut expect, b"v"); + expect.extend_from_slice(&1u32.to_le_bytes()); // group_count + spec_len_bytes(&mut expect, b"g"); + expect.extend_from_slice(&1u64.to_le_bytes()); // last_delivered.ms + expect.extend_from_slice(&1u64.to_le_bytes()); // last_delivered.seq + expect.extend_from_slice(&1u32.to_le_bytes()); // pel_count + expect.extend_from_slice(&1u64.to_le_bytes()); // pel id.ms + expect.extend_from_slice(&1u64.to_le_bytes()); // pel id.seq + spec_len_bytes(&mut expect, b"c"); + expect.extend_from_slice(&7u64.to_le_bytes()); // delivery_time + expect.extend_from_slice(&2u64.to_le_bytes()); // delivery_count + expect.extend_from_slice(&1u32.to_le_bytes()); // consumer_count + spec_len_bytes(&mut expect, b"c"); + expect.extend_from_slice(&9u64.to_le_bytes()); // seen_time + expect.extend_from_slice(&1u32.to_le_bytes()); // pending_count + expect.extend_from_slice(&1u64.to_le_bytes()); // pending id.ms + expect.extend_from_slice(&1u64.to_le_bytes()); // pending id.seq + assert_eq!(buf, expect); + + match decode(&buf, ValueType::Stream, HashTtlTrailer::Lenient) { + RedisValue::Stream(s) => { + assert_eq!(s.entries.len(), 1); + let g = s.groups.get(b"g".as_ref()).unwrap(); + assert_eq!(g.pel.len(), 1); + assert_eq!(g.consumers.len(), 1); + } + other => panic!("expected Stream, got {other:?}"), + } + } + + #[test] + fn corrupt_listpack_score_is_fail_closed() { + // A sorted-set listpack holding a non-numeric score must refuse to + // encode (the old duplicated encoders silently wrote 0.0). + let mut lp = Listpack::new(); + lp.push_back(b"member"); + lp.push_back(b"not-a-number"); + let mut buf = Vec::new(); + let err = encode_value_body(&RedisValueRef::SortedSetListpack(&lp), &mut buf) + .expect_err("corrupt score must not encode"); + assert!(matches!(err, ValueCodecError::CorruptScore)); + } + + #[test] + fn string_has_no_body() { + let mut buf = Vec::new(); + assert!(matches!( + encode_value_body(&RedisValueRef::String(b"x"), &mut buf), + Err(ValueCodecError::StringHasNoBody) + )); + let mut cursor = Cursor::new(&b"anything"[..]); + assert!(matches!( + decode_value_body(&mut cursor, ValueType::String, HashTtlTrailer::Lenient), + Err(ValueCodecError::StringHasNoBody) + )); + } + + #[test] + fn legacy_pre_trailer_hash_blob_decodes_lenient_only() { + // Body that ends right after the fields (no trailer): old spill blobs. + let mut blob = Vec::new(); + blob.extend_from_slice(&1u32.to_le_bytes()); + spec_len_bytes(&mut blob, b"f"); + spec_len_bytes(&mut blob, b"v"); + + // Lenient (spill): decodes as plain Hash. + match decode(&blob, ValueType::Hash, HashTtlTrailer::Lenient) { + RedisValue::Hash(m) => assert_eq!(m.len(), 1), + other => panic!("expected Hash, got {other:?}"), + } + // Required (RDB v2): truncation is corruption. + let mut cursor = Cursor::new(blob.as_slice()); + assert!(decode_value_body(&mut cursor, ValueType::Hash, HashTtlTrailer::Required).is_err()); + // Absent (RDB v1): trailer bytes are not consumed at all. + let mut cursor = Cursor::new(blob.as_slice()); + assert!(matches!( + decode_value_body(&mut cursor, ValueType::Hash, HashTtlTrailer::Absent).unwrap(), + RedisValue::Hash(_) + )); + assert_eq!(cursor.position() as usize, blob.len()); + } + + #[test] + fn oversized_count_rejected_before_allocation() { + // count=u32::MAX with a 4-byte body: must error, not with_capacity(4B). + let mut blob = Vec::new(); + blob.extend_from_slice(&u32::MAX.to_le_bytes()); + blob.extend_from_slice(b"xxxx"); + for vt in [ + ValueType::Hash, + ValueType::List, + ValueType::Set, + ValueType::ZSet, + ] { + let mut cursor = Cursor::new(blob.as_slice()); + assert!( + decode_value_body(&mut cursor, vt, HashTtlTrailer::Lenient).is_err(), + "{vt:?} must reject oversized count" + ); + } + } + + #[test] + fn compact_variants_encode_canonical_format() { + // A listpack hash and a HashMap hash with identical single-field + // content must produce identical bytes. + let mut lp = Listpack::new(); + lp.push_back(b"f1"); + lp.push_back(b"v1"); + let mut lp_buf = Vec::new(); + encode_value_body(&RedisValueRef::HashListpack(&lp), &mut lp_buf).unwrap(); + + let mut map = std::collections::HashMap::new(); + map.insert(Bytes::from_static(b"f1"), Bytes::from_static(b"v1")); + let mut map_buf = Vec::new(); + encode_value_body(&RedisValueRef::Hash(&map), &mut map_buf).unwrap(); + + assert_eq!(lp_buf, map_buf); + } + + #[test] + fn round_trip_multi_element_collections() { + // Hash 3 fields + let mut map = std::collections::HashMap::new(); + for i in 0..3u8 { + map.insert( + Bytes::copy_from_slice(&[b'f', b'0' + i]), + Bytes::copy_from_slice(&[b'v', b'0' + i]), + ); + } + let mut buf = Vec::new(); + encode_value_body(&RedisValueRef::Hash(&map), &mut buf).unwrap(); + match decode(&buf, ValueType::Hash, HashTtlTrailer::Lenient) { + RedisValue::Hash(m) => assert_eq!(m, map), + other => panic!("expected Hash, got {other:?}"), + } + + // ZSet 3 members + let mut members = std::collections::HashMap::new(); + let mut tree = BPTree::new(); + for i in 0..3u8 { + let member = Bytes::copy_from_slice(&[b'm', b'0' + i]); + members.insert(member.clone(), i as f64 * 0.5); + tree.insert(OrderedFloat(i as f64 * 0.5), member); + } + let mut buf = Vec::new(); + encode_value_body( + &RedisValueRef::SortedSetBPTree { + tree: &tree, + members: &members, + }, + &mut buf, + ) + .unwrap(); + match decode(&buf, ValueType::ZSet, HashTtlTrailer::Lenient) { + RedisValue::SortedSetBPTree { members: m, .. } => assert_eq!(m, members), + other => panic!("expected SortedSetBPTree, got {other:?}"), + } + } +}