From c8a91f6eafc015fee81235da0de59424a424b857 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BE=90=E6=B6=9B?= Date: Sun, 5 Jul 2026 17:09:45 +0800 Subject: [PATCH] =?UTF-8?q?refactor(memory):=20=E5=B0=86=20store.rs=20?= =?UTF-8?q?=E6=8B=86=E5=88=86=E4=B8=BA=E6=A8=A1=E5=9D=97=E7=9B=AE=E5=BD=95?= =?UTF-8?q?=EF=BC=8C=E4=BB=85=E7=BB=93=E6=9E=84=E6=90=AC=E7=A7=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将 src/memory/store.rs 单体文件拆分为模块根 + store/ 目录结构 - InMemoryStore（struct + impl + Default + 6 个内联测试）整体提取到 src/memory/store/in_memory.rs - store.rs 保留 MemoryStore trait 与 EvictionPolicy/EvictionConfig - 外部消费者导入路径 crate::memory::store::MemoryStore 不变 --- src/memory/store.rs | 263 +-------------------------------- src/memory/store/in_memory.rs | 266 ++++++++++++++++++++++++++++++++++ 2 files changed, 270 insertions(+), 259 deletions(-) create mode 100644 src/memory/store/in_memory.rs diff --git a/src/memory/store.rs b/src/memory/store.rs index 2f65814..389d3e9 100644 --- a/src/memory/store.rs +++ b/src/memory/store.rs @@ -1,14 +1,14 @@ //! MemoryStore 抽象接口与默认实现。 -use std::collections::HashMap; -use std::sync::Mutex; - use async_trait::async_trait; -use time::OffsetDateTime; use crate::memory::error::MemoryError; use crate::memory::types::{MemoryFilter, MemoryItem}; +pub mod in_memory; + +pub use in_memory::InMemoryStore; + /// 底层记忆存储抽象接口。 /// /// 下游可实现此 trait 以对接持久化后端（JSON 文件、SQLite、Redis 等）。 @@ -58,258 +58,3 @@ impl Default for EvictionConfig { } } } - -/// 进程内默认实现 —— 基于 HashMap + Mutex，纯内存。 -pub struct InMemoryStore { - items: Mutex>, - eviction: EvictionConfig, - /// 自上次淘汰检查以来的写入次数。 - writes_since_check: Mutex, -} - -impl InMemoryStore { - /// 创建一个无淘汰策略的 InMemoryStore。 - pub fn new() -> Self { - Self { - items: Mutex::new(HashMap::new()), - eviction: EvictionConfig::default(), - writes_since_check: Mutex::new(0), - } - } - - /// 创建一个带淘汰配置的 InMemoryStore。 - pub fn with_eviction(eviction: EvictionConfig) -> Self { - Self { - items: Mutex::new(HashMap::new()), - eviction, - writes_since_check: Mutex::new(0), - } - } - - fn maybe_evict(&self) { - // 不使用 .lock().await 跨点，先取计数判断是否需要淘汰 - let should_check = { - let mut counter = self.writes_since_check.lock().unwrap(); - *counter += 1; - if *counter >= self.eviction.check_interval { - *counter = 0; - true - } else { - false - } - }; - if !should_check { - return; - } - - let policy = self.eviction.policy.clone(); - match policy { - EvictionPolicy::None => {} - EvictionPolicy::Ttl { ttl_secs } => { - let cutoff = OffsetDateTime::now_utc() - time::Duration::seconds(ttl_secs as i64); - let mut items = self.items.lock().unwrap(); - items.retain(|_, v| v.created_at > cutoff); - } - EvictionPolicy::Capacity { max_items } => { - let mut items = self.items.lock().unwrap(); - if items.len() > max_items { - let mut vec: Vec<_> = items.drain().collect(); - // O(n) 部分排序：保留 created_at 最大的 max_items 个 - vec.select_nth_unstable_by(max_items, |a, b| { - b.1.created_at.cmp(&a.1.created_at) - }); - vec.truncate(max_items); - *items = vec.into_iter().collect(); - } - } - } - } -} - -impl Default for InMemoryStore { - fn default() -> Self { - Self::new() - } -} - -#[async_trait] -impl MemoryStore for InMemoryStore { - async fn save(&self, item: MemoryItem) -> Result<(), MemoryError> { - { - let mut items = self.items.lock().unwrap(); - items.insert(item.id.clone(), item); - } - self.maybe_evict(); - Ok(()) - } - - async fn get(&self, id: &str) -> Result, MemoryError> { - let items = self.items.lock().unwrap(); - Ok(items.get(id).cloned()) - } - - async fn delete(&self, id: &str) -> Result<(), MemoryError> { - let mut items = self.items.lock().unwrap(); - items.remove(id); - Ok(()) - } - - async fn list(&self, filter: &MemoryFilter) -> Result, MemoryError> { - let items = self.items.lock().unwrap(); - let mut result: Vec = items - .values() - .filter(|v| match &filter.prefix { - Some(p) => v.id.starts_with(p), - None => true, - }) - .filter(|v| match filter.since { - Some(t) => v.created_at > t, - None => true, - }) - .cloned() - .collect(); - // 按 created_at 升序排列（最旧在前） - result.sort_by_key(|v| v.created_at); - // 应用 offset - if let Some(offset) = filter.offset { - if offset < result.len() { - result.drain(..offset); - } else { - result.clear(); - } - } - // 应用 limit - if let Some(limit) = filter.limit { - result.truncate(limit); - } - Ok(result) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use time::OffsetDateTime; - - fn make_item(id: &str) -> MemoryItem { - MemoryItem { - id: id.to_string(), - content: format!("content-{id}"), - metadata: serde_json::json!({}), - created_at: OffsetDateTime::now_utc(), - } - } - - #[tokio::test] - async fn save_get_delete_list() { - let store = InMemoryStore::new(); - store.save(make_item("a")).await.unwrap(); - store.save(make_item("b")).await.unwrap(); - - let got = store.get("a").await.unwrap(); - assert!(got.is_some()); - assert_eq!(got.unwrap().id, "a"); - - let list = store.list(&MemoryFilter::default()).await.unwrap(); - assert_eq!(list.len(), 2); - - store.delete("a").await.unwrap(); - assert!(store.get("a").await.unwrap().is_none()); - } - - #[tokio::test] - async fn save_is_upsert() { - let store = InMemoryStore::new(); - store.save(make_item("a")).await.unwrap(); - let mut item = make_item("a"); - item.content = "updated".to_string(); - store.save(item).await.unwrap(); - let got = store.get("a").await.unwrap().unwrap(); - assert_eq!(got.content, "updated"); - let list = store.list(&MemoryFilter::default()).await.unwrap(); - assert_eq!(list.len(), 1); - } - - #[tokio::test] - async fn list_with_prefix_and_limit() { - let store = InMemoryStore::new(); - store.save(make_item("foo_a")).await.unwrap(); - store.save(make_item("foo_b")).await.unwrap(); - store.save(make_item("bar_a")).await.unwrap(); - - let filter = MemoryFilter { - prefix: Some("foo_".to_string()), - ..Default::default() - }; - let list = store.list(&filter).await.unwrap(); - assert_eq!(list.len(), 2); - - let filter = MemoryFilter { - prefix: Some("foo_".to_string()), - limit: Some(1), - ..Default::default() - }; - let list = store.list(&filter).await.unwrap(); - assert_eq!(list.len(), 1); - } - - #[tokio::test] - async fn capacity_eviction() { - // 强制每次写入都检查 - let eviction = EvictionConfig { - policy: EvictionPolicy::Capacity { max_items: 2 }, - check_interval: 1, - }; - let store = InMemoryStore::with_eviction(eviction); - // 第一条和第二条共存 - store.save(make_item("a")).await.unwrap(); - store.save(make_item("b")).await.unwrap(); - // 第三条写入触发淘汰：a 或 b 之一被淘汰 - store.save(make_item("c")).await.unwrap(); - - let list = store.list(&MemoryFilter::default()).await.unwrap(); - assert_eq!(list.len(), 2); - // 留下的应该是 b 和 c（最新的两个） - let ids: Vec<&str> = list.iter().map(|v| v.id.as_str()).collect(); - assert!(ids.contains(&"b")); - assert!(ids.contains(&"c")); - } - - #[tokio::test] - async fn ttl_eviction() { - // TTL 设为 0 会立即过期，但我们想保留 "a" 等待 "b" 写入后被淘汰。 - // 改用小 TTL + 睡眠：先 save a，sleep，save b 时 a 已过期被淘汰。 - let eviction = EvictionConfig { - policy: EvictionPolicy::Ttl { ttl_secs: 1 }, - check_interval: 1, - }; - let store = InMemoryStore::with_eviction(eviction); - store.save(make_item("a")).await.unwrap(); - // 等待超过 1 秒 - std::thread::sleep(std::time::Duration::from_millis(1100)); - // 触发淘汰：a 已超过 ttl_secs=1，应被淘汰 - store.save(make_item("b")).await.unwrap(); - let list = store.list(&MemoryFilter::default()).await.unwrap(); - // 由于 ttl_secs=1，且 b 刚写入，可能刚好处于临界值。 - // 我们只断言 list 不包含 "a" 即可。 - let ids: Vec<&str> = list.iter().map(|v| v.id.as_str()).collect(); - assert!( - !ids.contains(&"a"), - "expected 'a' to be evicted, but found in {ids:?}" - ); - } - - #[tokio::test] - async fn none_policy_no_eviction() { - let eviction = EvictionConfig { - policy: EvictionPolicy::None, - check_interval: 1, - }; - let store = InMemoryStore::with_eviction(eviction); - for i in 0..100 { - store.save(make_item(&format!("item_{i}"))).await.unwrap(); - } - let list = store.list(&MemoryFilter::default()).await.unwrap(); - assert_eq!(list.len(), 100); - } -} diff --git a/src/memory/store/in_memory.rs b/src/memory/store/in_memory.rs new file mode 100644 index 0000000..3923dd7 --- /dev/null +++ b/src/memory/store/in_memory.rs @@ -0,0 +1,266 @@ +//! 进程内默认实现 —— 基于 HashMap + Mutex，纯内存。 + +use std::collections::HashMap; +use std::sync::Mutex; + +use async_trait::async_trait; +use time::OffsetDateTime; + +use crate::memory::error::MemoryError; +use crate::memory::store::{EvictionConfig, EvictionPolicy, MemoryStore}; +use crate::memory::types::{MemoryFilter, MemoryItem}; + +/// 进程内默认实现 —— 基于 HashMap + Mutex，纯内存。 +pub struct InMemoryStore { + items: Mutex>, + eviction: EvictionConfig, + /// 自上次淘汰检查以来的写入次数。 + writes_since_check: Mutex, +} + +impl InMemoryStore { + /// 创建一个无淘汰策略的 InMemoryStore。 + pub fn new() -> Self { + Self { + items: Mutex::new(HashMap::new()), + eviction: EvictionConfig::default(), + writes_since_check: Mutex::new(0), + } + } + + /// 创建一个带淘汰配置的 InMemoryStore。 + pub fn with_eviction(eviction: EvictionConfig) -> Self { + Self { + items: Mutex::new(HashMap::new()), + eviction, + writes_since_check: Mutex::new(0), + } + } + + fn maybe_evict(&self) { + // 不使用 .lock().await 跨点，先取计数判断是否需要淘汰 + let should_check = { + let mut counter = self.writes_since_check.lock().unwrap(); + *counter += 1; + if *counter >= self.eviction.check_interval { + *counter = 0; + true + } else { + false + } + }; + if !should_check { + return; + } + + let policy = self.eviction.policy.clone(); + match policy { + EvictionPolicy::None => {} + EvictionPolicy::Ttl { ttl_secs } => { + let cutoff = OffsetDateTime::now_utc() - time::Duration::seconds(ttl_secs as i64); + let mut items = self.items.lock().unwrap(); + items.retain(|_, v| v.created_at > cutoff); + } + EvictionPolicy::Capacity { max_items } => { + let mut items = self.items.lock().unwrap(); + if items.len() > max_items { + let mut vec: Vec<_> = items.drain().collect(); + // O(n) 部分排序：保留 created_at 最大的 max_items 个 + vec.select_nth_unstable_by(max_items, |a, b| { + b.1.created_at.cmp(&a.1.created_at) + }); + vec.truncate(max_items); + *items = vec.into_iter().collect(); + } + } + } + } +} + +impl Default for InMemoryStore { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl MemoryStore for InMemoryStore { + async fn save(&self, item: MemoryItem) -> Result<(), MemoryError> { + { + let mut items = self.items.lock().unwrap(); + items.insert(item.id.clone(), item); + } + self.maybe_evict(); + Ok(()) + } + + async fn get(&self, id: &str) -> Result, MemoryError> { + let items = self.items.lock().unwrap(); + Ok(items.get(id).cloned()) + } + + async fn delete(&self, id: &str) -> Result<(), MemoryError> { + let mut items = self.items.lock().unwrap(); + items.remove(id); + Ok(()) + } + + async fn list(&self, filter: &MemoryFilter) -> Result, MemoryError> { + let items = self.items.lock().unwrap(); + let mut result: Vec = items + .values() + .filter(|v| match &filter.prefix { + Some(p) => v.id.starts_with(p), + None => true, + }) + .filter(|v| match filter.since { + Some(t) => v.created_at > t, + None => true, + }) + .cloned() + .collect(); + // 按 created_at 升序排列（最旧在前） + result.sort_by_key(|v| v.created_at); + // 应用 offset + if let Some(offset) = filter.offset { + if offset < result.len() { + result.drain(..offset); + } else { + result.clear(); + } + } + // 应用 limit + if let Some(limit) = filter.limit { + result.truncate(limit); + } + Ok(result) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use time::OffsetDateTime; + + fn make_item(id: &str) -> MemoryItem { + MemoryItem { + id: id.to_string(), + content: format!("content-{id}"), + metadata: serde_json::json!({}), + created_at: OffsetDateTime::now_utc(), + } + } + + #[tokio::test] + async fn save_get_delete_list() { + let store = InMemoryStore::new(); + store.save(make_item("a")).await.unwrap(); + store.save(make_item("b")).await.unwrap(); + + let got = store.get("a").await.unwrap(); + assert!(got.is_some()); + assert_eq!(got.unwrap().id, "a"); + + let list = store.list(&MemoryFilter::default()).await.unwrap(); + assert_eq!(list.len(), 2); + + store.delete("a").await.unwrap(); + assert!(store.get("a").await.unwrap().is_none()); + } + + #[tokio::test] + async fn save_is_upsert() { + let store = InMemoryStore::new(); + store.save(make_item("a")).await.unwrap(); + let mut item = make_item("a"); + item.content = "updated".to_string(); + store.save(item).await.unwrap(); + let got = store.get("a").await.unwrap().unwrap(); + assert_eq!(got.content, "updated"); + let list = store.list(&MemoryFilter::default()).await.unwrap(); + assert_eq!(list.len(), 1); + } + + #[tokio::test] + async fn list_with_prefix_and_limit() { + let store = InMemoryStore::new(); + store.save(make_item("foo_a")).await.unwrap(); + store.save(make_item("foo_b")).await.unwrap(); + store.save(make_item("bar_a")).await.unwrap(); + + let filter = MemoryFilter { + prefix: Some("foo_".to_string()), + ..Default::default() + }; + let list = store.list(&filter).await.unwrap(); + assert_eq!(list.len(), 2); + + let filter = MemoryFilter { + prefix: Some("foo_".to_string()), + limit: Some(1), + ..Default::default() + }; + let list = store.list(&filter).await.unwrap(); + assert_eq!(list.len(), 1); + } + + #[tokio::test] + async fn capacity_eviction() { + // 强制每次写入都检查 + let eviction = EvictionConfig { + policy: EvictionPolicy::Capacity { max_items: 2 }, + check_interval: 1, + }; + let store = InMemoryStore::with_eviction(eviction); + // 第一条和第二条共存 + store.save(make_item("a")).await.unwrap(); + store.save(make_item("b")).await.unwrap(); + // 第三条写入触发淘汰：a 或 b 之一被淘汰 + store.save(make_item("c")).await.unwrap(); + + let list = store.list(&MemoryFilter::default()).await.unwrap(); + assert_eq!(list.len(), 2); + // 留下的应该是 b 和 c（最新的两个） + let ids: Vec<&str> = list.iter().map(|v| v.id.as_str()).collect(); + assert!(ids.contains(&"b")); + assert!(ids.contains(&"c")); + } + + #[tokio::test] + async fn ttl_eviction() { + // TTL 设为 0 会立即过期，但我们想保留 "a" 等待 "b" 写入后被淘汰。 + // 改用小 TTL + 睡眠：先 save a，sleep，save b 时 a 已过期被淘汰。 + let eviction = EvictionConfig { + policy: EvictionPolicy::Ttl { ttl_secs: 1 }, + check_interval: 1, + }; + let store = InMemoryStore::with_eviction(eviction); + store.save(make_item("a")).await.unwrap(); + // 等待超过 1 秒 + std::thread::sleep(std::time::Duration::from_millis(1100)); + // 触发淘汰：a 已超过 ttl_secs=1，应被淘汰 + store.save(make_item("b")).await.unwrap(); + let list = store.list(&MemoryFilter::default()).await.unwrap(); + // 由于 ttl_secs=1，且 b 刚写入，可能刚好处于临界值。 + // 我们只断言 list 不包含 "a" 即可。 + let ids: Vec<&str> = list.iter().map(|v| v.id.as_str()).collect(); + assert!( + !ids.contains(&"a"), + "expected 'a' to be evicted, but found in {ids:?}" + ); + } + + #[tokio::test] + async fn none_policy_no_eviction() { + let eviction = EvictionConfig { + policy: EvictionPolicy::None, + check_interval: 1, + }; + let store = InMemoryStore::with_eviction(eviction); + for i in 0..100 { + store.save(make_item(&format!("item_{i}"))).await.unwrap(); + } + let list = store.list(&MemoryFilter::default()).await.unwrap(); + assert_eq!(list.len(), 100); + } +}