From f39c957ab683f6861081fc4540572a8ee749e932 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 20 Jul 2026 12:21:13 +0700 Subject: [PATCH] =?UTF-8?q?refactor(storage):=20one=20typed-accessor=20ske?= =?UTF-8?q?leton=20=E2=80=94=20ValueKind/OwnedKind=20generics=20replace=20?= =?UTF-8?q?Database's=20per-type=20accessor=20matrix=20(W5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Database had grown a ~20-method typed-accessor matrix: get_X, get_or_create_X, and get_X_ref_if_alive for each container type (hash, list, set, sorted set, stream), every one a copy-paste of the same skeleton — expiry check -> cold promote/read-through -> compact- encoding upgrade -> variant projection — with only the enum arms differing. Skeleton fixes (the P0 cold-collection-visibility fix, the promote-before-fabricate fix) had to be applied N times and could silently miss a copy. The skeleton now exists ONCE per access shape as a generic on Database: get_ref_if_alive:: &self; hot probe -> expiry -> classify -> non-promoting cold read-through (Owned view) get_or_create:: &mut; expiry-drop -> cold promote -> insert-on-miss -> upgrade -> project mut get_promoted:: &mut; expiry-drop -> cold promote -> upgrade -> project shared Everything genuinely per-type — which enum variants map to which view, how a compact encoding upgrades (incl. the deliberate asymmetries: sorted sets upgrade the legacy BTreeMap form but NOT SortedSetListpack, whose upgrade lives in the zset command layer), what a fresh entry looks like — lives in the new storage::db_kind module as ValueKind/OwnedKind impls on zero-sized markers. Static dispatch only: every K is a concrete marker resolved at compile time, so the generated code is identical to the hand-written originals. The public per-type methods (get_hash, get_or_create_set, get_list_ref_if_alive, get_stream_if_alive, ...) remain as thin delegators — command-layer call sites are untouched. The generic get_or_create also replaces the per-type unwrap-after-insert with the defensive log-and-error path only the hash accessor previously had. Deleted: the four caller-less get_X_if_alive variants (hash/list/set/ sorted_set; compact encodings returned None — long superseded by the _ref_if_alive family). get_or_create_intset and the *_listpack threshold accessors keep their bespoke shapes (their Option-returning contracts don't fit the trait and have single call sites). Pure refactor — no behavior change: - Behavior pins added GREEN BEFORE the refactor and re-verified after: hot-WRONGTYPE through all four ref accessors; a hot HashWithTtl entry maps to HashRef::WithTtl carrying the CALLER's now_ms (expired field filters, live field reads through). Cold-Owned legs, promote-instead- of-fabricate, and wrongtype pins already existed and still pass. - Net -318 lines; db.rs 3431 -> 3113 lines. Gates: lib suite 4427 green (4425 + 2 new pins) on macOS + Linux VM release-fast, monoio + tokio,jemalloc feature sets; crash matrix + kill-9 offload + cold-visibility suites on Linux VM; clippy -D warnings on both feature sets; fmt. Stacked on refactor/w4-eviction-api (PR #399). author: Tin Dang --- CHANGELOG.md | 16 + src/command/geo/geo_cmd.rs | 2 +- src/command/sorted_set/sorted_set_read.rs | 2 +- src/storage/db.rs | 652 ++++++---------------- src/storage/db_kind.rs | 383 +++++++++++++ src/storage/mod.rs | 1 + 6 files changed, 568 insertions(+), 488 deletions(-) create mode 100644 src/storage/db_kind.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 473ace8c..62cac85d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 tolerance is now an exact equality). ### Changed +- **One typed-accessor skeleton in `Database` (W5).** The ~20-method + accessor matrix (`get_X` / `get_or_create_X` / `get_X_ref_if_alive` per + container type) copy-pasted the same skeleton — expiry check → cold + promote/read-through → compact-encoding upgrade → variant projection — + once per type. The skeleton now exists once per access shape + (`get_ref_if_alive::` / `get_or_create::` / `get_promoted::`); + everything genuinely per-type lives in the new `storage::db_kind` module + as `ValueKind`/`OwnedKind` marker impls (static dispatch — generated code + identical to the hand-written originals). The public per-type methods + remain as thin delegators, so command-layer call sites are unchanged. + The four caller-less `get_X_if_alive` variants (compact encodings + returned `None`; superseded by the `_ref_if_alive` family) are deleted. + Pure refactor — behavior pinned by two new tests (hot-WRONGTYPE through + the ref accessors; `HashWithTtl` → `HashRef::WithTtl` wiring incl. + caller-`now_ms` propagation) plus the existing cold-visibility pins. + - **One eviction entry point (W4).** The 13-name `try_evict_if_needed*` family (every combination of spill-sink / explicit-total / elastic-budget / plain-drop-reporting encoded as a function-name suffix) is replaced by a diff --git a/src/command/geo/geo_cmd.rs b/src/command/geo/geo_cmd.rs index 94ce74dc..e94447e3 100644 --- a/src/command/geo/geo_cmd.rs +++ b/src/command/geo/geo_cmd.rs @@ -760,7 +760,7 @@ fn geosearch_core( // Read-only twins for the shared-lock (dispatch_read) path // --------------------------------------------------------------------------- // -// GEO data is stored as a sorted set: all twins use `get_sorted_set_if_alive`. +// GEO data is stored as a sorted set: all twins use `get_sorted_set_ref_if_alive`. /// GEOPOS key member [member …] — read-only twin. pub fn geopos_readonly(db: &crate::storage::db::Database, args: &[Frame], now_ms: u64) -> Frame { diff --git a/src/command/sorted_set/sorted_set_read.rs b/src/command/sorted_set/sorted_set_read.rs index 2eae2104..7aab43b5 100644 --- a/src/command/sorted_set/sorted_set_read.rs +++ b/src/command/sorted_set/sorted_set_read.rs @@ -1396,7 +1396,7 @@ fn collect_source_sets( Ok(source_data) } -/// Read-only twin: collect source sets using `get_sorted_set_if_alive`. +/// Read-only twin: collect source sets using `get_sorted_set_ref_if_alive`. /// /// Compact (listpack) encodings return an empty map — callers see an absent /// set, which is correct because compact sets are upgraded to BPTree on first diff --git a/src/storage/db.rs b/src/storage/db.rs index af6a92a1..2fa324b3 100644 --- a/src/storage/db.rs +++ b/src/storage/db.rs @@ -5,8 +5,9 @@ pub use super::db_read::{HashRef, ListRef, SetRef, SortedSetRef, StreamRef}; use super::bptree::BPTree; use super::compact_key::CompactKey; -use super::compact_value::{CompactValue, RedisValueRef}; +use super::compact_value::RedisValueRef; use super::dashtable::{DashTable, InsertOrUpdate}; +use super::db_kind::{self, OwnedKind, ValueKind}; use super::entry::{CachedClock, Entry, RedisValue, current_secs, current_time_ms}; use super::intset::Intset; use super::stream::Stream as StreamData; @@ -1649,140 +1650,147 @@ impl Database { } } - /// Get or create a hash entry. Returns mutable ref to inner HashMap. - /// Returns Err(WRONGTYPE) if key exists with wrong type. - /// - /// New keys start with compact listpack encoding and are upgraded to - /// full HashMap on first mutable access (eager upgrade). - #[allow(clippy::unwrap_used)] // get_mut() after insert guarantees key present; as_redis_value_mut() on known type - pub fn get_or_create_hash(&mut self, key: &[u8]) -> Result<&mut HashMap, Frame> { - let now_ms = self.cached_now_ms; - // Expire check + // ── W5: generic typed accessors ───────────────────────────────────── + // + // The per-type accessor skeleton (expiry check → cold promote/ + // read-through → compact-encoding upgrade → variant projection) exists + // once per access shape below; everything genuinely per-type lives in + // `storage::db_kind` as a `ValueKind`/`OwnedKind` marker impl. The + // public `get_hash`/`get_or_create_set`/`get_list_ref_if_alive`/… + // methods are thin delegators, so command-layer call sites are + // unchanged and dispatch is fully static. + + /// Remove `key` if it is expired at `now_ms`, crediting its memory. + fn drop_if_expired(&mut self, key: &[u8], now_ms: u64) { if Self::check_expired(&self.data, key, now_ms) { if let Some(entry) = self.data.remove(key) { self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); } } + } + + /// Read-only typed access via the kind's `*Ref` view. + /// + /// Takes `&self` because this backs BOTH the exclusive-dispatch path + /// (`&mut Database`, reborrowed) AND the RwLock-shared-read dispatch + /// path (`&Database` only). The latter cannot promote a cold hit into + /// hot RAM, so a hot miss falls back to a non-promoting cold + /// read-through returning the view's `Owned` variant (P0 + /// cold-collection-visibility fix) — the fast path (key present hot) + /// still costs exactly one probe. Expiry is checked against the + /// caller's `now_ms` without removing the key or touching LRU. + pub fn get_ref_if_alive( + &self, + key: &[u8], + now_ms: u64, + ) -> Result>, Frame> { + if let Some(entry) = self.data.get(key) { + if entry.is_expired_at(now_ms) { + return Ok(None); + } + return match K::classify_hot(entry.value.as_redis_value(), now_ms) { + Ok(r) => Ok(Some(r)), + Err(db_kind::WrongType) => Err(Self::wrongtype_error()), + }; + } + match self.cold_read_only(key, now_ms) { + Some(v) => match K::classify_cold(v, now_ms) { + Ok(r) => Ok(Some(r)), + Err(db_kind::WrongType) => Err(Self::wrongtype_error()), + }, + None => Ok(None), + } + } + + /// Get-or-create typed access to the full (non-compact) encoding. + /// + /// Promotes a cold-spilled value back to hot RAM before fabricating an + /// empty container (P0 fix: a write on an evicted key must not silently + /// shadow the cold copy — see `Self::promote_cold_if_present`), then + /// upgrades the kind's compact encoding(s) in place. + pub fn get_or_create(&mut self, key: &[u8]) -> Result, Frame> { + let now_ms = self.cached_now_ms; + self.drop_if_expired(key, now_ms); if !self.data.contains_key(key) { - // P0 fix: a hash may have been spilled to the cold tier by - // eviction. Promote it back to hot RAM before fabricating an - // empty container, or HSET on an evicted hash would silently - // destroy the cold copy (see `Self::promote_cold_if_present`). self.promote_cold_if_present(key, now_ms); if !self.data.contains_key(key) { - let entry = Entry::new_hash(); + let entry = K::new_entry(); let k = CompactKey::from(key); self.used_memory += entry_overhead(key, &entry); self.data.insert(k, entry); } } - let entry = match self.data.get_mut(key) { - Some(e) => e, - None => { - // This should not happen — insert was just called above. - // Log and return an error instead of panicking. - tracing::error!( - "get_or_create_hash: get_mut returned None after insert for key len={}", - key.len() - ); - return Err(Frame::Error(bytes::Bytes::from_static( - b"ERR internal: hash lookup failed after insert", - ))); - } + let Some(entry) = self.data.get_mut(key) else { + // Should not happen — insert was just called above. Log and + // return an error instead of panicking. + tracing::error!( + "get_or_create: get_mut returned None after insert for key len={}", + key.len() + ); + return Err(Frame::Error(bytes::Bytes::from_static( + b"ERR internal: lookup failed after insert", + ))); }; - // Upgrade compact listpack to full HashMap if needed - let needs_upgrade = matches!( - entry.value.as_redis_value_mut(), - Some(RedisValue::HashListpack(_)) - ); - if needs_upgrade { - if let Some(RedisValue::HashListpack(lp)) = entry.value.as_redis_value_mut() { - let map = lp.to_hash_map(); - if let Some(v) = entry.value.as_redis_value_mut() { - *v = RedisValue::Hash(map); - } - } - } + K::upgrade(entry); match entry.value.as_redis_value_mut() { - Some(RedisValue::Hash(map)) => Ok(map), - // HashWithTtl: the fields sub-map is a plain HashMap. - // HSET/HMSET/HINCRBY must be able to mutate it directly; they never - // touch the ttls sidecar, which is managed by the HEXPIRE family. - Some(RedisValue::HashWithTtl { fields, .. }) => Ok(fields), - _ => Err(Self::wrongtype_error()), + Some(v) => match K::project_mut(v) { + Ok(m) => Ok(m), + Err(db_kind::WrongType) => Err(Self::wrongtype_error()), + }, + None => Err(Self::wrongtype_error()), } } - /// Get a hash entry (read-only). Returns None if key missing, Err if wrong type. - /// Upgrades compact encoding to full HashMap if found. - /// - /// Promotes a cold-spilled hash back to hot RAM on miss (P0 - /// cold-collection-visibility fix) — this accessor takes `&mut self`, so - /// unlike the enum-based `get_hash_ref_if_alive` it can promote directly - /// instead of decoding a throwaway copy on every call. - #[allow(clippy::unwrap_used)] // as_redis_value_mut() on known compact type during upgrade - pub fn get_hash(&mut self, key: &[u8]) -> Result>, Frame> { + /// Read typed access to the full (non-compact) encoding, promoting a + /// cold-spilled value back to hot RAM on miss (this accessor takes + /// `&mut self` — unlike the enum-based `get_ref_if_alive` it can + /// promote directly instead of decoding a throwaway copy per call) and + /// upgrading the kind's compact encoding(s) in place when present. + pub fn get_promoted( + &mut self, + key: &[u8], + ) -> Result>, Frame> { let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } + self.drop_if_expired(key, now_ms); if !self.data.contains_key(key) { self.promote_cold_if_present(key, now_ms); } - // Upgrade compact encoding if present if let Some(entry) = self.data.get_mut(key) { - if let Some(RedisValue::HashListpack(lp)) = entry.value.as_redis_value_mut() { - let map = lp.to_hash_map(); - *entry.value.as_redis_value_mut().unwrap() = RedisValue::Hash(map); - } + K::upgrade(entry); } match self.data.get(key) { None => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::Hash(map) => Ok(Some(map)), - // HashWithTtl: expose the fields sub-map for reads. Callers - // (HGET, HMGET, HKEYS, HVALS, HGETALL, etc.) see the HashMap - // directly; expired-field filtering happens in hash_field_state - // or via the active-expiry tick, not on every read. - RedisValueRef::HashWithTtl { fields, .. } => Ok(Some(fields)), - _ => Err(Self::wrongtype_error()), + Some(entry) => match K::project_ref(entry.value.as_redis_value()) { + Ok(r) => Ok(Some(r)), + Err(db_kind::WrongType) => Err(Self::wrongtype_error()), }, } } + /// Get or create a hash entry. Returns mutable ref to inner HashMap. + /// Returns Err(WRONGTYPE) if key exists with wrong type. + /// + /// New keys start with compact listpack encoding and are upgraded to + /// full HashMap on first mutable access (eager upgrade). + pub fn get_or_create_hash(&mut self, key: &[u8]) -> Result<&mut HashMap, Frame> { + self.get_or_create::(key) + } + + /// Get a hash entry (read-only). Returns None if key missing, Err if wrong type. + /// Upgrades compact encoding to full HashMap if found. + /// + /// Promotes a cold-spilled hash back to hot RAM on miss (P0 + /// cold-collection-visibility fix) — this accessor takes `&mut self`, so + /// unlike the enum-based `get_hash_ref_if_alive` it can promote directly + /// instead of decoding a throwaway copy on every call. + pub fn get_hash(&mut self, key: &[u8]) -> Result>, Frame> { + self.get_promoted::(key) + } + /// Get or create a list entry. Returns mutable ref to inner VecDeque. /// New keys start with full encoding. Upgrades compact listpack on access. - #[allow(clippy::unwrap_used)] // get_mut() after insert guarantees key present; as_redis_value_mut() on known type pub fn get_or_create_list(&mut self, key: &[u8]) -> Result<&mut VecDeque, Frame> { - let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } - if !self.data.contains_key(key) { - // P0 fix: promote a cold-spilled list before fabricating an empty - // one (see `Self::promote_cold_if_present`). - self.promote_cold_if_present(key, now_ms); - if !self.data.contains_key(key) { - let entry = Entry::new_list(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); - } - } - let entry = self.data.get_mut(key).unwrap(); - // Upgrade compact listpack to full VecDeque if needed - if let Some(RedisValue::ListListpack(lp)) = entry.value.as_redis_value_mut() { - let list = lp.to_vec_deque(); - *entry.value.as_redis_value_mut().unwrap() = RedisValue::List(list); - } - match entry.value.as_redis_value_mut() { - Some(RedisValue::List(list)) => Ok(list), - _ => Err(Self::wrongtype_error()), - } + self.get_or_create::(key) } /// Get a list entry (read-only). Returns None if key missing, Err if wrong type. @@ -1790,71 +1798,14 @@ impl Database { /// /// Promotes a cold-spilled list back to hot RAM on miss (P0 /// cold-collection-visibility fix) — this accessor takes `&mut self`. - #[allow(clippy::unwrap_used)] // as_redis_value_mut() on known compact type during upgrade pub fn get_list(&mut self, key: &[u8]) -> Result>, Frame> { - let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } - if !self.data.contains_key(key) { - self.promote_cold_if_present(key, now_ms); - } - // Upgrade compact encoding if present - if let Some(entry) = self.data.get_mut(key) { - if let Some(RedisValue::ListListpack(lp)) = entry.value.as_redis_value_mut() { - let list = lp.to_vec_deque(); - *entry.value.as_redis_value_mut().unwrap() = RedisValue::List(list); - } - } - match self.data.get(key) { - None => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::List(list) => Ok(Some(list)), - _ => Err(Self::wrongtype_error()), - }, - } + self.get_promoted::(key) } /// Get or create a set entry. Returns mutable ref to inner HashSet. /// New keys start with full encoding. Upgrades compact encodings on access. - #[allow(clippy::unwrap_used)] // get_mut() after insert guarantees key present; as_redis_value_mut() on known type pub fn get_or_create_set(&mut self, key: &[u8]) -> Result<&mut HashSet, Frame> { - let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } - if !self.data.contains_key(key) { - // P0 fix: promote a cold-spilled set before fabricating an empty - // one (see `Self::promote_cold_if_present`). - self.promote_cold_if_present(key, now_ms); - if !self.data.contains_key(key) { - let entry = Entry::new_set(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); - } - } - let entry = self.data.get_mut(key).unwrap(); - // Upgrade compact encodings to full HashSet - match entry.value.as_redis_value_mut() { - Some(RedisValue::SetListpack(lp)) => { - let set = lp.to_hash_set(); - *entry.value.as_redis_value_mut().unwrap() = RedisValue::Set(set); - } - Some(RedisValue::SetIntset(is)) => { - let set = is.to_hash_set(); - *entry.value.as_redis_value_mut().unwrap() = RedisValue::Set(set); - } - _ => {} - } - match entry.value.as_redis_value_mut() { - Some(RedisValue::Set(set)) => Ok(set), - _ => Err(Self::wrongtype_error()), - } + self.get_or_create::(key) } /// Get a set entry (read-only). Returns None if key missing, Err if wrong type. @@ -1862,38 +1813,8 @@ impl Database { /// /// Promotes a cold-spilled set back to hot RAM on miss (P0 /// cold-collection-visibility fix) — this accessor takes `&mut self`. - #[allow(clippy::unwrap_used)] // as_redis_value_mut() on known compact type during upgrade pub fn get_set(&mut self, key: &[u8]) -> Result>, Frame> { - let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } - if !self.data.contains_key(key) { - self.promote_cold_if_present(key, now_ms); - } - // Upgrade compact encodings if present - if let Some(entry) = self.data.get_mut(key) { - match entry.value.as_redis_value_mut() { - Some(RedisValue::SetListpack(lp)) => { - let set = lp.to_hash_set(); - *entry.value.as_redis_value_mut().unwrap() = RedisValue::Set(set); - } - Some(RedisValue::SetIntset(is)) => { - let set = is.to_hash_set(); - *entry.value.as_redis_value_mut().unwrap() = RedisValue::Set(set); - } - _ => {} - } - } - match self.data.get(key) { - None => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::Set(set) => Ok(Some(set)), - _ => Err(Self::wrongtype_error()), - }, - } + self.get_promoted::(key) } /// Get or create an intset entry. Creates a new SetIntset if the key doesn't exist. @@ -2070,97 +1991,22 @@ impl Database { /// /// New keys start with SortedSetBPTree encoding. Legacy SortedSet (BTreeMap) /// entries are upgraded to SortedSetBPTree on access for backward compatibility. - #[allow(clippy::unwrap_used)] // get_mut() after insert guarantees key present; legacy upgrade is infallible pub fn get_or_create_sorted_set( &mut self, key: &[u8], ) -> Result<(&mut HashMap, &mut BPTree), Frame> { - let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } - if !self.data.contains_key(key) { - // P0 fix: promote a cold-spilled sorted set before fabricating - // an empty one (see `Self::promote_cold_if_present`). - self.promote_cold_if_present(key, now_ms); - if !self.data.contains_key(key) { - let entry = Entry::new_sorted_set_bptree(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); - } - } - // Upgrade old SortedSet (BTreeMap) to SortedSetBPTree on access - let entry = self.data.get_mut(key).unwrap(); - if let Some(RedisValue::SortedSet { members, scores }) = entry.value.as_redis_value_mut() { - let mut tree = BPTree::new(); - let new_members = std::mem::take(members); - let old_scores = std::mem::take(scores); - for ((score, member), ()) in old_scores { - tree.insert(score, member); - } - entry.value = CompactValue::from_redis_value(RedisValue::SortedSetBPTree { - tree, - members: new_members, - }); - } - match entry.value.as_redis_value_mut() { - Some(RedisValue::SortedSetBPTree { members, tree }) => Ok((members, tree)), - _ => Err(Self::wrongtype_error()), - } + self.get_or_create::(key) } /// Get a sorted set entry (read-only). Returns None if key missing, Err if wrong type. /// /// Promotes a cold-spilled sorted set back to hot RAM on miss (P0 /// cold-collection-visibility fix) — this accessor takes `&mut self`. - #[allow(clippy::unwrap_used)] // get_mut() after confirmed existence; legacy BTreeMap upgrade is infallible pub fn get_sorted_set( &mut self, key: &[u8], ) -> Result, &BPTree)>, Frame> { - let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } - if !self.data.contains_key(key) { - self.promote_cold_if_present(key, now_ms); - } - // Upgrade old SortedSet (BTreeMap) to SortedSetBPTree on access - if let Some(entry) = self.data.get(key) { - if matches!( - entry.value.as_redis_value(), - RedisValueRef::SortedSet { .. } - ) { - let entry = self.data.get_mut(key).unwrap(); - if let Some(RedisValue::SortedSet { members, scores }) = - entry.value.as_redis_value_mut() - { - let mut tree = BPTree::new(); - let new_members = std::mem::take(members); - let old_scores = std::mem::take(scores); - for ((score, member), ()) in old_scores { - tree.insert(score, member); - } - entry.value = CompactValue::from_redis_value(RedisValue::SortedSetBPTree { - tree, - members: new_members, - }); - } - } - } - match self.data.get(key) { - None => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::SortedSetBPTree { tree, members } => Ok(Some((members, tree))), - RedisValueRef::SortedSet { .. } => unreachable!("should have been upgraded"), - _ => Err(Self::wrongtype_error()), - }, - } + self.get_promoted::(key) } /// Collect keys that have an expiration set. @@ -2282,82 +2128,6 @@ impl Database { } } - /// Read-only hash access. Returns None if key missing or expired, Err if wrong type. - /// Compact encodings return None (caller falls through to mutable upgrade path). - #[allow(dead_code)] - pub fn get_hash_if_alive( - &self, - key: &[u8], - now_ms: u64, - ) -> Result>, Frame> { - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::Hash(map) => Ok(Some(map)), - RedisValueRef::HashListpack(_) => Ok(None), - _ => Err(Self::wrongtype_error()), - }, - } - } - - /// Read-only list access. Returns None if key missing or expired, Err if wrong type. - /// Compact encodings return None (caller falls through to mutable upgrade path). - #[allow(dead_code)] - pub fn get_list_if_alive( - &self, - key: &[u8], - now_ms: u64, - ) -> Result>, Frame> { - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::List(list) => Ok(Some(list)), - RedisValueRef::ListListpack(_) => Ok(None), - _ => Err(Self::wrongtype_error()), - }, - } - } - - /// Read-only set access. Returns None if key missing or expired, Err if wrong type. - /// Compact encodings return None (caller falls through to mutable upgrade path). - #[allow(dead_code)] - pub fn get_set_if_alive( - &self, - key: &[u8], - now_ms: u64, - ) -> Result>, Frame> { - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::Set(set) => Ok(Some(set)), - RedisValueRef::SetListpack(_) | RedisValueRef::SetIntset(_) => Ok(None), - _ => Err(Self::wrongtype_error()), - }, - } - } - - /// Read-only sorted set access. Returns None if key missing or expired, Err if wrong type. - /// Compact encodings return None (caller falls through to mutable upgrade path). - #[allow(dead_code)] - pub fn get_sorted_set_if_alive( - &self, - key: &[u8], - now_ms: u64, - ) -> Result, &BPTree)>, Frame> { - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::SortedSetBPTree { tree, members } => Ok(Some((members, tree))), - RedisValueRef::SortedSet { .. } | RedisValueRef::SortedSetListpack(_) => Ok(None), - _ => Err(Self::wrongtype_error()), - }, - } - } - // ---- Enum-based readonly accessors that handle compact encodings ---- /// Read-only hash access via HashRef enum. @@ -2379,41 +2149,7 @@ impl Database { key: &[u8], now_ms: u64, ) -> Result>, Frame> { - if let Some(entry) = self.data.get(key) { - if entry.is_expired_at(now_ms) { - return Ok(None); - } - return match entry.value.as_redis_value() { - RedisValueRef::Hash(map) => Ok(Some(HashRef::Map(map))), - RedisValueRef::HashListpack(lp) => Ok(Some(HashRef::Listpack(lp))), - RedisValueRef::HashWithTtl { - fields, - ttls, - min_expiry_ms, - } => Ok(Some(HashRef::WithTtl { - fields, - ttls, - now_ms, - min_expiry_ms, - })), - _ => Err(Self::wrongtype_error()), - }; - } - match self.cold_read_only(key, now_ms) { - Some(RedisValue::Hash(map)) => Ok(Some(HashRef::Owned(map))), - Some(RedisValue::HashWithTtl { - fields, - ttls, - min_expiry_ms, - }) => Ok(Some(HashRef::OwnedWithTtl { - fields, - ttls, - now_ms, - min_expiry_ms, - })), - Some(_) => Err(Self::wrongtype_error()), - None => Ok(None), - } + self.get_ref_if_alive::(key, now_ms) } /// Read-only list access via ListRef enum. Handles both VecDeque and Listpack. @@ -2426,21 +2162,7 @@ impl Database { key: &[u8], now_ms: u64, ) -> Result>, Frame> { - if let Some(entry) = self.data.get(key) { - if entry.is_expired_at(now_ms) { - return Ok(None); - } - return match entry.value.as_redis_value() { - RedisValueRef::List(list) => Ok(Some(ListRef::Deque(list))), - RedisValueRef::ListListpack(lp) => Ok(Some(ListRef::Listpack(lp))), - _ => Err(Self::wrongtype_error()), - }; - } - match self.cold_read_only(key, now_ms) { - Some(RedisValue::List(list)) => Ok(Some(ListRef::Owned(list))), - Some(_) => Err(Self::wrongtype_error()), - None => Ok(None), - } + self.get_ref_if_alive::(key, now_ms) } /// Read-only set access via SetRef enum. Handles HashSet, Listpack, and Intset. @@ -2453,22 +2175,7 @@ impl Database { key: &[u8], now_ms: u64, ) -> Result>, Frame> { - if let Some(entry) = self.data.get(key) { - if entry.is_expired_at(now_ms) { - return Ok(None); - } - return match entry.value.as_redis_value() { - RedisValueRef::Set(set) => Ok(Some(SetRef::Hash(set))), - RedisValueRef::SetListpack(lp) => Ok(Some(SetRef::Listpack(lp))), - RedisValueRef::SetIntset(is) => Ok(Some(SetRef::Intset(is))), - _ => Err(Self::wrongtype_error()), - }; - } - match self.cold_read_only(key, now_ms) { - Some(RedisValue::Set(set)) => Ok(Some(SetRef::Owned(set))), - Some(_) => Err(Self::wrongtype_error()), - None => Ok(None), - } + self.get_ref_if_alive::(key, now_ms) } /// Read-only sorted set access via SortedSetRef enum. Handles BPTree, Listpack, and Legacy. @@ -2481,28 +2188,7 @@ impl Database { key: &[u8], now_ms: u64, ) -> Result>, Frame> { - if let Some(entry) = self.data.get(key) { - if entry.is_expired_at(now_ms) { - return Ok(None); - } - return match entry.value.as_redis_value() { - RedisValueRef::SortedSetBPTree { tree, members } => { - Ok(Some(SortedSetRef::BPTree { tree, members })) - } - RedisValueRef::SortedSetListpack(lp) => Ok(Some(SortedSetRef::Listpack(lp))), - RedisValueRef::SortedSet { members, scores } => { - Ok(Some(SortedSetRef::Legacy { members, scores })) - } - _ => Err(Self::wrongtype_error()), - }; - } - match self.cold_read_only(key, now_ms) { - Some(RedisValue::SortedSetBPTree { tree, members }) => { - Ok(Some(SortedSetRef::Owned { tree, members })) - } - Some(_) => Err(Self::wrongtype_error()), - None => Ok(None), - } + self.get_ref_if_alive::(key, now_ms) } // ---- Low-level helpers for blocking wakeup hooks ---- @@ -2589,30 +2275,8 @@ impl Database { } /// Get or create a stream at the given key. Returns WRONGTYPE if key holds another type. - #[allow(clippy::unwrap_used)] // get_mut() after insert guarantees key present pub fn get_or_create_stream(&mut self, key: &[u8]) -> Result<&mut StreamData, Frame> { - let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } - if !self.data.contains_key(key) { - // P0 fix: promote a cold-spilled stream before fabricating an - // empty one (see `Self::promote_cold_if_present`). - self.promote_cold_if_present(key, now_ms); - if !self.data.contains_key(key) { - let entry = Entry::new_stream(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); - } - } - let entry = self.data.get_mut(key).unwrap(); - match entry.value.as_redis_value_mut() { - Some(RedisValue::Stream(s)) => Ok(s.as_mut()), - _ => Err(Self::wrongtype_error()), - } + self.get_or_create::(key) } /// Read-only stream access for the shared-lock read path. @@ -2633,20 +2297,7 @@ impl Database { key: &[u8], now_ms: u64, ) -> Result>, Frame> { - if let Some(entry) = self.data.get(key) { - if entry.is_expired_at(now_ms) { - return Ok(None); - } - return match entry.value.as_redis_value() { - RedisValueRef::Stream(s) => Ok(Some(StreamRef::Borrowed(s))), - _ => Err(Self::wrongtype_error()), - }; - } - match self.cold_read_only(key, now_ms) { - Some(RedisValue::Stream(s)) => Ok(Some(StreamRef::Owned(s))), - Some(_) => Err(Self::wrongtype_error()), - None => Ok(None), - } + self.get_ref_if_alive::(key, now_ms) } /// Get a read-only reference to a stream. Returns Ok(None) if key doesn't exist. @@ -2655,22 +2306,7 @@ impl Database { /// Promotes a cold-spilled stream back to hot RAM on miss (P0 /// cold-collection-visibility fix) — this accessor takes `&mut self`. pub fn get_stream(&mut self, key: &[u8]) -> Result, Frame> { - let now_ms = self.cached_now_ms; - if Self::check_expired(&self.data, key, now_ms) { - if let Some(entry) = self.data.remove(key) { - self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); - } - } - if !self.data.contains_key(key) { - self.promote_cold_if_present(key, now_ms); - } - match self.data.get(key) { - None => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::Stream(s) => Ok(Some(s)), - _ => Err(Self::wrongtype_error()), - }, - } + self.get_promoted::(key) } /// Get a mutable reference to an existing stream. Returns Ok(None) if key doesn't exist. @@ -3249,6 +2885,50 @@ mod tests { ); } + /// W5 pin: the ref accessor's WRONGTYPE leg on a live hot key. + #[test] + fn test_ref_accessors_wrongtype_on_hot_string() { + let mut db = Database::new(); + db.set_string(Bytes::from_static(b"s"), Bytes::from_static(b"v")); + assert!(db.get_hash_ref_if_alive(b"s", 0).is_err()); + assert!(db.get_list_ref_if_alive(b"s", 0).is_err()); + assert!(db.get_set_ref_if_alive(b"s", 0).is_err()); + assert!(db.get_sorted_set_ref_if_alive(b"s", 0).is_err()); + } + + /// W5 pin: `get_hash_ref_if_alive` must map a hot `HashWithTtl` entry to + /// `HashRef::WithTtl` carrying the CALLER's `now_ms` — an expired field + /// filters out, a live one reads through. + #[test] + fn test_hash_ref_with_ttl_leg_filters_expired_fields() { + let mut db = Database::new(); + let mut fields = HashMap::new(); + fields.insert(Bytes::from_static(b"dead"), Bytes::from_static(b"x")); + fields.insert(Bytes::from_static(b"live"), Bytes::from_static(b"y")); + let mut ttls = HashMap::new(); + ttls.insert(Bytes::from_static(b"dead"), 1_000u64); + ttls.insert(Bytes::from_static(b"live"), 5_000u64); + let mut entry = Entry::new_hash(); + entry.value = crate::storage::compact_value::CompactValue::from_redis_value( + RedisValue::HashWithTtl { + fields, + ttls, + min_expiry_ms: 1_000, + }, + ); + db.set(Bytes::from_static(b"h"), entry); + + // now_ms between the two field deadlines: "dead" filtered, "live" seen. + let href = db.get_hash_ref_if_alive(b"h", 2_000).unwrap().unwrap(); + assert!(matches!(href, HashRef::WithTtl { .. })); + assert_eq!(href.get_field(b"dead"), None, "expired field must filter"); + assert_eq!( + href.get_field(b"live").unwrap(), + Bytes::from_static(b"y"), + "live field must read through" + ); + } + #[test] fn test_get_or_create_list_promotes_cold_instead_of_fabricating() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/storage/db_kind.rs b/src/storage/db_kind.rs new file mode 100644 index 00000000..80bb9468 --- /dev/null +++ b/src/storage/db_kind.rs @@ -0,0 +1,383 @@ +//! W5: per-type classification/projection traits backing `Database`'s +//! generic typed accessors. +//! +//! The ~20-method accessor matrix in `db.rs` (`get_X` / `get_or_create_X` / +//! `get_X_ref_if_alive` per container type) copy-pasted the same skeleton — +//! expiry check → cold promote/read-through → compact-encoding upgrade → +//! variant match — once per type. The skeleton now exists once per access +//! shape as a generic in `db.rs` (`get_ref_if_alive::`, +//! `get_or_create::`, `get_promoted::`); everything genuinely +//! per-type — which enum variants map to which view, how a compact encoding +//! upgrades, what a fresh entry looks like — lives here as a [`ValueKind`] / +//! [`OwnedKind`] impl on a zero-sized marker type. Static dispatch only: +//! every `K` is a concrete marker resolved at compile time, so the generated +//! code is identical to the hand-written originals. + +use bytes::Bytes; +use std::collections::{HashMap, HashSet, VecDeque}; + +use super::bptree::BPTree; +use super::compact_value::{CompactValue, RedisValueRef}; +use super::db_read::{HashRef, ListRef, SetRef, SortedSetRef, StreamRef}; +use super::entry::{Entry, RedisValue}; +use super::stream::Stream as StreamData; + +/// Classification failure: the value exists but holds a different type. +/// The generic accessors translate this into the `WRONGTYPE` error frame. +pub struct WrongType; + +/// A container type readable through a borrowed-or-owned view (`*Ref` enum) +/// without `&mut Database` — backs `Database::get_ref_if_alive::`. +pub trait ValueKind { + /// The view handed to command code (`HashRef`, `ListRef`, …). + type Ref<'a>; + + /// Classify a HOT entry's borrowed value into the view. + fn classify_hot(val: RedisValueRef<'_>, now_ms: u64) -> Result, WrongType>; + + /// Classify an OWNED value decoded fresh from the cold tier. The + /// returned view holds owned data, so it satisfies any caller lifetime. + fn classify_cold<'a>(val: RedisValue, now_ms: u64) -> Result, WrongType>; +} + +/// A container type accessible in its full (non-compact) encoding via +/// `&mut Database` — backs `Database::get_or_create::` and +/// `Database::get_promoted::`. +pub trait OwnedKind { + /// Mutable projection of the full encoding (e.g. `&mut HashMap<..>`, or + /// the `(&mut members, &mut tree)` pair for sorted sets). + type Mut<'a>; + /// Shared projection of the full encoding. + type Shared<'a>; + + /// A fresh empty entry for `get_or_create` to insert on miss. + fn new_entry() -> Entry; + + /// Upgrade this type's compact encoding(s) to the full one, in place. + /// Exactly the conversions the hand-written accessors performed — e.g. + /// sorted sets upgrade the legacy `BTreeMap` form but deliberately NOT + /// `SortedSetListpack` (its upgrade lives in the zset command layer). + fn upgrade(entry: &mut Entry); + + /// Project the (post-upgrade) full encoding out of the value, mutably. + fn project_mut(val: &mut RedisValue) -> Result, WrongType>; + + /// Project the (post-upgrade) full encoding out of the value, shared. + fn project_ref(val: RedisValueRef<'_>) -> Result, WrongType>; +} + +// ── Markers ───────────────────────────────────────────────────────────── + +pub struct HashKind; +pub struct ListKind; +pub struct SetKind; +pub struct SortedSetKind; +pub struct StreamKind; + +// ── Hash ──────────────────────────────────────────────────────────────── + +impl ValueKind for HashKind { + type Ref<'a> = HashRef<'a>; + + fn classify_hot(val: RedisValueRef<'_>, now_ms: u64) -> Result, WrongType> { + match val { + RedisValueRef::Hash(map) => Ok(HashRef::Map(map)), + RedisValueRef::HashListpack(lp) => Ok(HashRef::Listpack(lp)), + RedisValueRef::HashWithTtl { + fields, + ttls, + min_expiry_ms, + } => Ok(HashRef::WithTtl { + fields, + ttls, + now_ms, + min_expiry_ms, + }), + _ => Err(WrongType), + } + } + + fn classify_cold<'a>(val: RedisValue, now_ms: u64) -> Result, WrongType> { + match val { + RedisValue::Hash(map) => Ok(HashRef::Owned(map)), + RedisValue::HashWithTtl { + fields, + ttls, + min_expiry_ms, + } => Ok(HashRef::OwnedWithTtl { + fields, + ttls, + now_ms, + min_expiry_ms, + }), + _ => Err(WrongType), + } + } +} + +impl OwnedKind for HashKind { + type Mut<'a> = &'a mut HashMap; + type Shared<'a> = &'a HashMap; + + fn new_entry() -> Entry { + Entry::new_hash() + } + + fn upgrade(entry: &mut Entry) { + if let Some(v) = entry.value.as_redis_value_mut() { + if let RedisValue::HashListpack(lp) = v { + let map = lp.to_hash_map(); + *v = RedisValue::Hash(map); + } + } + } + + fn project_mut(val: &mut RedisValue) -> Result, WrongType> { + match val { + RedisValue::Hash(map) => Ok(map), + // HashWithTtl: the fields sub-map is a plain HashMap. HSET/ + // HMSET/HINCRBY mutate it directly; they never touch the ttls + // sidecar, which is managed by the HEXPIRE family. + RedisValue::HashWithTtl { fields, .. } => Ok(fields), + _ => Err(WrongType), + } + } + + fn project_ref(val: RedisValueRef<'_>) -> Result, WrongType> { + match val { + RedisValueRef::Hash(map) => Ok(map), + RedisValueRef::HashWithTtl { fields, .. } => Ok(fields), + _ => Err(WrongType), + } + } +} + +// ── List ──────────────────────────────────────────────────────────────── + +impl ValueKind for ListKind { + type Ref<'a> = ListRef<'a>; + + fn classify_hot(val: RedisValueRef<'_>, _now_ms: u64) -> Result, WrongType> { + match val { + RedisValueRef::List(list) => Ok(ListRef::Deque(list)), + RedisValueRef::ListListpack(lp) => Ok(ListRef::Listpack(lp)), + _ => Err(WrongType), + } + } + + fn classify_cold<'a>(val: RedisValue, _now_ms: u64) -> Result, WrongType> { + match val { + RedisValue::List(list) => Ok(ListRef::Owned(list)), + _ => Err(WrongType), + } + } +} + +impl OwnedKind for ListKind { + type Mut<'a> = &'a mut VecDeque; + type Shared<'a> = &'a VecDeque; + + fn new_entry() -> Entry { + Entry::new_list() + } + + fn upgrade(entry: &mut Entry) { + if let Some(v) = entry.value.as_redis_value_mut() { + if let RedisValue::ListListpack(lp) = v { + let list = lp.to_vec_deque(); + *v = RedisValue::List(list); + } + } + } + + fn project_mut(val: &mut RedisValue) -> Result, WrongType> { + match val { + RedisValue::List(list) => Ok(list), + _ => Err(WrongType), + } + } + + fn project_ref(val: RedisValueRef<'_>) -> Result, WrongType> { + match val { + RedisValueRef::List(list) => Ok(list), + _ => Err(WrongType), + } + } +} + +// ── Set ───────────────────────────────────────────────────────────────── + +impl ValueKind for SetKind { + type Ref<'a> = SetRef<'a>; + + fn classify_hot(val: RedisValueRef<'_>, _now_ms: u64) -> Result, WrongType> { + match val { + RedisValueRef::Set(set) => Ok(SetRef::Hash(set)), + RedisValueRef::SetListpack(lp) => Ok(SetRef::Listpack(lp)), + RedisValueRef::SetIntset(is) => Ok(SetRef::Intset(is)), + _ => Err(WrongType), + } + } + + fn classify_cold<'a>(val: RedisValue, _now_ms: u64) -> Result, WrongType> { + match val { + RedisValue::Set(set) => Ok(SetRef::Owned(set)), + _ => Err(WrongType), + } + } +} + +impl OwnedKind for SetKind { + type Mut<'a> = &'a mut HashSet; + type Shared<'a> = &'a HashSet; + + fn new_entry() -> Entry { + Entry::new_set() + } + + fn upgrade(entry: &mut Entry) { + if let Some(v) = entry.value.as_redis_value_mut() { + match v { + RedisValue::SetListpack(lp) => { + let set = lp.to_hash_set(); + *v = RedisValue::Set(set); + } + RedisValue::SetIntset(is) => { + let set = is.to_hash_set(); + *v = RedisValue::Set(set); + } + _ => {} + } + } + } + + fn project_mut(val: &mut RedisValue) -> Result, WrongType> { + match val { + RedisValue::Set(set) => Ok(set), + _ => Err(WrongType), + } + } + + fn project_ref(val: RedisValueRef<'_>) -> Result, WrongType> { + match val { + RedisValueRef::Set(set) => Ok(set), + _ => Err(WrongType), + } + } +} + +// ── Sorted set ────────────────────────────────────────────────────────── + +impl ValueKind for SortedSetKind { + type Ref<'a> = SortedSetRef<'a>; + + fn classify_hot(val: RedisValueRef<'_>, _now_ms: u64) -> Result, WrongType> { + match val { + RedisValueRef::SortedSetBPTree { tree, members } => { + Ok(SortedSetRef::BPTree { tree, members }) + } + RedisValueRef::SortedSetListpack(lp) => Ok(SortedSetRef::Listpack(lp)), + RedisValueRef::SortedSet { members, scores } => { + Ok(SortedSetRef::Legacy { members, scores }) + } + _ => Err(WrongType), + } + } + + fn classify_cold<'a>(val: RedisValue, _now_ms: u64) -> Result, WrongType> { + match val { + RedisValue::SortedSetBPTree { tree, members } => { + Ok(SortedSetRef::Owned { tree, members }) + } + _ => Err(WrongType), + } + } +} + +impl OwnedKind for SortedSetKind { + type Mut<'a> = (&'a mut HashMap, &'a mut BPTree); + type Shared<'a> = (&'a HashMap, &'a BPTree); + + fn new_entry() -> Entry { + Entry::new_sorted_set_bptree() + } + + /// Upgrade the legacy `SortedSet` (BTreeMap) form to `SortedSetBPTree`. + /// `SortedSetListpack` is deliberately NOT upgraded here — its upgrade + /// path lives in the zset command layer (same as the hand-written + /// accessor this replaces). + fn upgrade(entry: &mut Entry) { + if let Some(RedisValue::SortedSet { members, scores }) = entry.value.as_redis_value_mut() { + let mut tree = BPTree::new(); + let new_members = std::mem::take(members); + let old_scores = std::mem::take(scores); + for ((score, member), ()) in old_scores { + tree.insert(score, member); + } + entry.value = CompactValue::from_redis_value(RedisValue::SortedSetBPTree { + tree, + members: new_members, + }); + } + } + + fn project_mut(val: &mut RedisValue) -> Result, WrongType> { + match val { + RedisValue::SortedSetBPTree { members, tree } => Ok((members, tree)), + _ => Err(WrongType), + } + } + + fn project_ref(val: RedisValueRef<'_>) -> Result, WrongType> { + match val { + RedisValueRef::SortedSetBPTree { tree, members } => Ok((members, tree)), + _ => Err(WrongType), + } + } +} + +// ── Stream ────────────────────────────────────────────────────────────── + +impl ValueKind for StreamKind { + type Ref<'a> = StreamRef<'a>; + + fn classify_hot(val: RedisValueRef<'_>, _now_ms: u64) -> Result, WrongType> { + match val { + RedisValueRef::Stream(s) => Ok(StreamRef::Borrowed(s)), + _ => Err(WrongType), + } + } + + fn classify_cold<'a>(val: RedisValue, _now_ms: u64) -> Result, WrongType> { + match val { + RedisValue::Stream(s) => Ok(StreamRef::Owned(s)), + _ => Err(WrongType), + } + } +} + +impl OwnedKind for StreamKind { + type Mut<'a> = &'a mut StreamData; + type Shared<'a> = &'a StreamData; + + fn new_entry() -> Entry { + Entry::new_stream() + } + + /// Streams have no compact encoding — nothing to upgrade. + fn upgrade(_entry: &mut Entry) {} + + fn project_mut(val: &mut RedisValue) -> Result, WrongType> { + match val { + RedisValue::Stream(s) => Ok(s.as_mut()), + _ => Err(WrongType), + } + } + + fn project_ref(val: RedisValueRef<'_>) -> Result, WrongType> { + match val { + RedisValueRef::Stream(s) => Ok(s), + _ => Err(WrongType), + } + } +} diff --git a/src/storage/mod.rs b/src/storage/mod.rs index ef8e7588..914c6f13 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -5,6 +5,7 @@ pub mod compact_value; pub mod dashtable; pub mod db; pub mod db_hash_ttl; +pub mod db_kind; pub mod db_quota; pub mod db_read; pub mod engine;