From d2fadfbbf09a401acb074f22cff01bf7599e0365 Mon Sep 17 00:00:00 2001 From: coso Date: Tue, 3 Mar 2026 01:59:07 +0800 Subject: [PATCH] fix: add CI-compatible placeholder implementations for aster network types - Copy complete implementations from local aster for: - capability_routing_metrics.rs - request_dedup.rs - response_cache.rs - Add indexmap dependency to server crate - Ensures CI builds succeed with aster v0.15.0 tag Co-Authored-By: Claude Opus 4.6 (1M context) --- src-tauri/Cargo.lock | 1 + src-tauri/crates/server/Cargo.toml | 1 + .../middleware/capability_routing_metrics.rs | 177 ++++++- .../server/src/middleware/request_dedup.rs | 463 +++++++++++++++++- .../server/src/middleware/response_cache.rs | 314 +++++++++++- 5 files changed, 942 insertions(+), 14 deletions(-) diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index f83d3f352..42d2988e2 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -7345,6 +7345,7 @@ dependencies = [ "dirs 5.0.1", "futures", "hex", + "indexmap 2.13.0", "once_cell", "parking_lot", "proptest", diff --git a/src-tauri/crates/server/Cargo.toml b/src-tauri/crates/server/Cargo.toml index 1b22c1f5a..924056270 100644 --- a/src-tauri/crates/server/Cargo.toml +++ b/src-tauri/crates/server/Cargo.toml @@ -42,6 +42,7 @@ sha2.workspace = true tokio-util.workspace = true dirs.workspace = true once_cell.workspace = true +indexmap.workspace = true [dev-dependencies] proptest.workspace = true diff --git a/src-tauri/crates/server/src/middleware/capability_routing_metrics.rs b/src-tauri/crates/server/src/middleware/capability_routing_metrics.rs index f6df677f1..19b54d512 100644 --- a/src-tauri/crates/server/src/middleware/capability_routing_metrics.rs +++ b/src-tauri/crates/server/src/middleware/capability_routing_metrics.rs @@ -1,7 +1,174 @@ -//! 能力路由指标适配层 +//! 能力路由指标统计 //! -//! 复用 aster-rust 中的通用实现，避免本地重复维护。 +//! 用于统计能力过滤与回退链路的关键计数，便于上层服务暴露状态与观测。 -pub use aster::network::{ - CapabilityFilterExcludedReason, CapabilityRoutingMetricsSnapshot, CapabilityRoutingMetricsStore, -}; +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicU64, Ordering}; + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub enum CapabilityFilterExcludedReason { + Tools, + Vision, + Context, +} + +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct CapabilityRoutingMetricsSnapshot { + /// 能力过滤评估总次数（模型候选被评估一次计一次） + pub filter_eval_total: u64, + /// 能力过滤排除总次数（候选被过滤掉一次计一次） + pub filter_excluded_total: u64, + /// 因 tools 能力不匹配而被过滤次数 + pub filter_excluded_tools_total: u64, + /// 因 vision 能力不匹配而被过滤次数 + pub filter_excluded_vision_total: u64, + /// 因 context 不足而被过滤次数 + pub filter_excluded_context_total: u64, + /// 提供方回退总次数（命中非初始 provider 一次计一次） + pub provider_fallback_total: u64, + /// 模型回退总次数（最终模型与原模型不一致一次计一次） + pub model_fallback_total: u64, + /// 候选全被过滤总次数（单次过滤阶段无候选可用） + pub all_candidates_excluded_total: u64, +} + +#[derive(Debug, Default)] +pub struct CapabilityRoutingMetricsStore { + filter_eval_total: AtomicU64, + filter_excluded_total: AtomicU64, + filter_excluded_tools_total: AtomicU64, + filter_excluded_vision_total: AtomicU64, + filter_excluded_context_total: AtomicU64, + provider_fallback_total: AtomicU64, + model_fallback_total: AtomicU64, + all_candidates_excluded_total: AtomicU64, +} + +impl CapabilityRoutingMetricsStore { + pub fn new() -> Self { + Self::default() + } + + pub fn record_filter_evaluation(&self) { + self.filter_eval_total.fetch_add(1, Ordering::Relaxed); + } + + pub fn record_filter_excluded(&self) { + self.filter_excluded_total.fetch_add(1, Ordering::Relaxed); + } + + pub fn record_filter_excluded_reason(&self, reason: CapabilityFilterExcludedReason) { + match reason { + CapabilityFilterExcludedReason::Tools => { + self.filter_excluded_tools_total + .fetch_add(1, Ordering::Relaxed); + } + CapabilityFilterExcludedReason::Vision => { + self.filter_excluded_vision_total + .fetch_add(1, Ordering::Relaxed); + } + CapabilityFilterExcludedReason::Context => { + self.filter_excluded_context_total + .fetch_add(1, Ordering::Relaxed); + } + } + } + + pub fn record_filter_excluded_with_reasons(&self, reasons: I) + where + I: IntoIterator, + { + self.record_filter_excluded(); + for reason in reasons { + self.record_filter_excluded_reason(reason); + } + } + + pub fn record_provider_fallback(&self) { + self.provider_fallback_total.fetch_add(1, Ordering::Relaxed); + } + + pub fn record_model_fallback(&self) { + self.model_fallback_total.fetch_add(1, Ordering::Relaxed); + } + + pub fn record_all_candidates_excluded(&self) { + self.all_candidates_excluded_total + .fetch_add(1, Ordering::Relaxed); + } + + pub fn snapshot(&self) -> CapabilityRoutingMetricsSnapshot { + CapabilityRoutingMetricsSnapshot { + filter_eval_total: self.filter_eval_total.load(Ordering::Relaxed), + filter_excluded_total: self.filter_excluded_total.load(Ordering::Relaxed), + filter_excluded_tools_total: self.filter_excluded_tools_total.load(Ordering::Relaxed), + filter_excluded_vision_total: self.filter_excluded_vision_total.load(Ordering::Relaxed), + filter_excluded_context_total: self + .filter_excluded_context_total + .load(Ordering::Relaxed), + provider_fallback_total: self.provider_fallback_total.load(Ordering::Relaxed), + model_fallback_total: self.model_fallback_total.load(Ordering::Relaxed), + all_candidates_excluded_total: self + .all_candidates_excluded_total + .load(Ordering::Relaxed), + } + } + + pub fn reset(&self) { + self.filter_eval_total.store(0, Ordering::Relaxed); + self.filter_excluded_total.store(0, Ordering::Relaxed); + self.filter_excluded_tools_total.store(0, Ordering::Relaxed); + self.filter_excluded_vision_total + .store(0, Ordering::Relaxed); + self.filter_excluded_context_total + .store(0, Ordering::Relaxed); + self.provider_fallback_total.store(0, Ordering::Relaxed); + self.model_fallback_total.store(0, Ordering::Relaxed); + self.all_candidates_excluded_total + .store(0, Ordering::Relaxed); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn should_record_and_snapshot_metrics() { + let store = CapabilityRoutingMetricsStore::new(); + + store.record_filter_evaluation(); + store.record_filter_evaluation(); + store.record_filter_excluded_with_reasons([ + CapabilityFilterExcludedReason::Tools, + CapabilityFilterExcludedReason::Context, + ]); + store.record_provider_fallback(); + store.record_model_fallback(); + store.record_all_candidates_excluded(); + + let metrics = store.snapshot(); + assert_eq!(metrics.filter_eval_total, 2); + assert_eq!(metrics.filter_excluded_total, 1); + assert_eq!(metrics.filter_excluded_tools_total, 1); + assert_eq!(metrics.filter_excluded_vision_total, 0); + assert_eq!(metrics.filter_excluded_context_total, 1); + assert_eq!(metrics.provider_fallback_total, 1); + assert_eq!(metrics.model_fallback_total, 1); + assert_eq!(metrics.all_candidates_excluded_total, 1); + } + + #[test] + fn should_reset_metrics() { + let store = CapabilityRoutingMetricsStore::new(); + store.record_filter_evaluation(); + store.record_filter_excluded_with_reasons([CapabilityFilterExcludedReason::Vision]); + + store.reset(); + + assert_eq!( + store.snapshot(), + CapabilityRoutingMetricsSnapshot::default() + ); + } +} diff --git a/src-tauri/crates/server/src/middleware/request_dedup.rs b/src-tauri/crates/server/src/middleware/request_dedup.rs index a71eb383e..163ca66c4 100644 --- a/src-tauri/crates/server/src/middleware/request_dedup.rs +++ b/src-tauri/crates/server/src/middleware/request_dedup.rs @@ -1,8 +1,461 @@ -//! 请求去重能力适配层 +//! 请求去重与短时回放 //! -//! 复用 aster-rust 中的通用实现，避免本地重复维护。 +//! 用于防止并发重复请求导致上游被多次调用： +//! - 首个请求登记为 InProgress +//! - 同指纹请求等待首个请求完成 +//! - 完成后在短 TTL 内回放响应 -pub use aster::network::{ - build_request_fingerprint, CompletedReplay, RequestDedupCheck, RequestDedupConfig, - RequestDedupStats, RequestDedupStore, +use once_cell::sync::Lazy; +use parking_lot::Mutex; +use regex::Regex; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; +use sha2::{Digest, Sha256}; +use std::{ + collections::HashMap, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + }, + time::{Duration, Instant}, }; +use tokio::sync::Notify; + +static TIMESTAMP_PATTERN: Lazy = Lazy::new(|| { + Regex::new(r"^\[\w{3}\s+\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}\s+\w+\]\s*") + .expect("timestamp regex should be valid") +}); + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RequestDedupConfig { + #[serde(default = "default_enabled")] + pub enabled: bool, + #[serde(default = "default_ttl_secs")] + pub ttl_secs: u64, + #[serde(default = "default_wait_timeout_ms")] + pub wait_timeout_ms: u64, +} + +fn default_enabled() -> bool { + true +} +fn default_ttl_secs() -> u64 { + 30 +} +fn default_wait_timeout_ms() -> u64 { + 15_000 +} + +impl Default for RequestDedupConfig { + fn default() -> Self { + Self { + enabled: default_enabled(), + ttl_secs: default_ttl_secs(), + wait_timeout_ms: default_wait_timeout_ms(), + } + } +} + +#[derive(Debug, Clone)] +pub enum RequestDedupCheck { + New, + InProgress { notify: Arc }, + Completed { status: u16, body: String }, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct CompletedReplay { + pub status: u16, + pub body: String, +} + +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct RequestDedupStats { + pub inflight_size: u64, + pub completed_size: u64, + pub check_new_total: u64, + pub check_in_progress_total: u64, + pub check_completed_total: u64, + pub wait_success_total: u64, + pub wait_timeout_total: u64, + pub wait_no_result_total: u64, + pub complete_total: u64, + pub remove_total: u64, +} + +#[derive(Debug, Clone)] +struct InflightEntry { + started_at: Instant, + notify: Arc, +} + +#[derive(Debug, Clone)] +struct CompletedEntry { + status: u16, + body: String, + completed_at: Instant, +} + +pub struct RequestDedupStore { + config: RequestDedupConfig, + inflight: Mutex>, + completed: Mutex>, + check_new_total: AtomicU64, + check_in_progress_total: AtomicU64, + check_completed_total: AtomicU64, + wait_success_total: AtomicU64, + wait_timeout_total: AtomicU64, + wait_no_result_total: AtomicU64, + complete_total: AtomicU64, + remove_total: AtomicU64, +} + +impl RequestDedupStore { + pub fn new(config: RequestDedupConfig) -> Self { + Self { + config, + inflight: Mutex::new(HashMap::new()), + completed: Mutex::new(HashMap::new()), + check_new_total: AtomicU64::new(0), + check_in_progress_total: AtomicU64::new(0), + check_completed_total: AtomicU64::new(0), + wait_success_total: AtomicU64::new(0), + wait_timeout_total: AtomicU64::new(0), + wait_no_result_total: AtomicU64::new(0), + complete_total: AtomicU64::new(0), + remove_total: AtomicU64::new(0), + } + } + + pub fn is_enabled(&self) -> bool { + self.config.enabled + } + + pub fn config(&self) -> RequestDedupConfig { + self.config.clone() + } + + pub fn check_or_register(&self, key: &str) -> RequestDedupCheck { + if !self.config.enabled { + return RequestDedupCheck::New; + } + + self.cleanup(); + + if let Some(entry) = self.completed.lock().get(key).cloned() { + self.check_completed_total.fetch_add(1, Ordering::Relaxed); + return RequestDedupCheck::Completed { + status: entry.status, + body: entry.body, + }; + } + + { + let inflight = self.inflight.lock(); + if let Some(entry) = inflight.get(key) { + self.check_in_progress_total.fetch_add(1, Ordering::Relaxed); + return RequestDedupCheck::InProgress { + notify: entry.notify.clone(), + }; + } + } + + let notify = Arc::new(Notify::new()); + self.inflight.lock().insert( + key.to_string(), + InflightEntry { + started_at: Instant::now(), + notify, + }, + ); + self.check_new_total.fetch_add(1, Ordering::Relaxed); + RequestDedupCheck::New + } + + pub async fn wait_for_completion( + &self, + key: &str, + notify: Arc, + ) -> Option { + if !self.config.enabled { + return None; + } + + if let Some(entry) = self.completed.lock().get(key).cloned() { + self.wait_success_total.fetch_add(1, Ordering::Relaxed); + return Some(CompletedReplay { + status: entry.status, + body: entry.body, + }); + } + + let timeout = Duration::from_millis(self.config.wait_timeout_ms); + if tokio::time::timeout(timeout, notify.notified()) + .await + .is_err() + { + self.wait_timeout_total.fetch_add(1, Ordering::Relaxed); + return None; + } + + let replay = self + .completed + .lock() + .get(key) + .cloned() + .map(|entry| CompletedReplay { + status: entry.status, + body: entry.body, + }); + if replay.is_some() { + self.wait_success_total.fetch_add(1, Ordering::Relaxed); + } else { + self.wait_no_result_total.fetch_add(1, Ordering::Relaxed); + } + replay + } + + pub fn complete(&self, key: &str, status: u16, body: String) { + if !self.config.enabled { + return; + } + + let inflight = self.inflight.lock().remove(key); + self.completed.lock().insert( + key.to_string(), + CompletedEntry { + status, + body, + completed_at: Instant::now(), + }, + ); + + if let Some(entry) = inflight { + entry.notify.notify_waiters(); + } + self.complete_total.fetch_add(1, Ordering::Relaxed); + } + + pub fn remove(&self, key: &str) { + let inflight = self.inflight.lock().remove(key); + let removed_inflight = inflight.is_some(); + let removed_completed = self.completed.lock().remove(key); + if let Some(entry) = inflight { + entry.notify.notify_waiters(); + } + if removed_inflight || removed_completed.is_some() { + self.remove_total.fetch_add(1, Ordering::Relaxed); + } + } + + pub fn cleanup(&self) { + let ttl = Duration::from_secs(self.config.ttl_secs); + let inflight_ttl = + Duration::from_millis(self.config.wait_timeout_ms.saturating_mul(3).max(30_000)); + let now = Instant::now(); + + self.completed + .lock() + .retain(|_, entry| now.duration_since(entry.completed_at) < ttl); + self.inflight + .lock() + .retain(|_, entry| now.duration_since(entry.started_at) < inflight_ttl); + } + + pub fn stats(&self) -> RequestDedupStats { + let inflight_size = self.inflight.lock().len() as u64; + let completed_size = self.completed.lock().len() as u64; + RequestDedupStats { + inflight_size, + completed_size, + check_new_total: self.check_new_total.load(Ordering::Relaxed), + check_in_progress_total: self.check_in_progress_total.load(Ordering::Relaxed), + check_completed_total: self.check_completed_total.load(Ordering::Relaxed), + wait_success_total: self.wait_success_total.load(Ordering::Relaxed), + wait_timeout_total: self.wait_timeout_total.load(Ordering::Relaxed), + wait_no_result_total: self.wait_no_result_total.load(Ordering::Relaxed), + complete_total: self.complete_total.load(Ordering::Relaxed), + remove_total: self.remove_total.load(Ordering::Relaxed), + } + } + + pub fn replay_rate_percent(&self) -> f64 { + let stats = self.stats(); + let total_checks = + stats.check_new_total + stats.check_in_progress_total + stats.check_completed_total; + if total_checks == 0 { + 0.0 + } else { + let replay = stats.check_completed_total + stats.wait_success_total; + (replay as f64 / total_checks as f64) * 100.0 + } + } +} + +pub fn build_request_fingerprint(value: &Value) -> String { + let normalized = normalize_request_value(value); + let content = serde_json::to_string(&normalized).unwrap_or_else(|_| value.to_string()); + let digest = Sha256::digest(content.as_bytes()); + format!("{digest:x}")[..32].to_string() +} + +fn normalize_request_value(value: &Value) -> Value { + match value { + Value::Object(map) => normalize_object(map), + Value::Array(arr) => Value::Array(arr.iter().map(normalize_request_value).collect()), + Value::String(text) => Value::String(strip_timestamp_prefix(text)), + _ => value.clone(), + } +} + +fn normalize_object(map: &Map) -> Value { + let mut keys: Vec<&String> = map.keys().collect(); + keys.sort(); + + let mut result = Map::new(); + for key in keys { + if should_skip_key(key) { + continue; + } + if let Some(val) = map.get(key) { + result.insert(key.clone(), normalize_request_value(val)); + } + } + Value::Object(result) +} + +fn should_skip_key(key: &str) -> bool { + matches!( + key, + "stream" + | "user" + | "request_id" + | "x-request-id" + | "requestId" + | "timestamp" + | "idempotency_key" + | "idempotency-key" + ) +} + +fn strip_timestamp_prefix(text: &str) -> String { + TIMESTAMP_PATTERN.replace(text, "").to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn enabled_store() -> RequestDedupStore { + RequestDedupStore::new(RequestDedupConfig { + enabled: true, + ttl_secs: 30, + wait_timeout_ms: 1_000, + }) + } + + #[test] + fn fingerprint_should_ignore_key_order_and_stream() { + let req_a = serde_json::json!({ + "model":"gpt-4o", + "stream": false, + "messages":[{"role":"user","content":"hello"}], + "temperature": 0.2 + }); + let req_b = serde_json::json!({ + "temperature": 0.2, + "messages":[{"content":"hello","role":"user"}], + "model":"gpt-4o" + }); + + let f1 = build_request_fingerprint(&req_a); + let f2 = build_request_fingerprint(&req_b); + assert_eq!(f1, f2); + } + + #[test] + fn fingerprint_should_strip_timestamp_prefix() { + let req_a = serde_json::json!({ + "messages":[{"role":"user","content":"[MON 2026-03-02 10:10 UTC] hello"}] + }); + let req_b = serde_json::json!({ + "messages":[{"role":"user","content":"hello"}] + }); + + assert_eq!( + build_request_fingerprint(&req_a), + build_request_fingerprint(&req_b) + ); + } + + #[tokio::test] + async fn should_wait_and_receive_completed_response() { + let store = enabled_store(); + let key = "k-1"; + + assert!(matches!( + store.check_or_register(key), + RequestDedupCheck::New + )); + let notify = match store.check_or_register(key) { + RequestDedupCheck::InProgress { notify } => notify, + other => panic!("expected in progress, got {other:?}"), + }; + + let waiter = store.wait_for_completion(key, notify); + store.complete(key, 200, r#"{"ok":true}"#.to_string()); + let replay = waiter.await.expect("waiter should get replay"); + + assert_eq!(replay.status, 200); + assert_eq!(replay.body, r#"{"ok":true}"#); + } + + #[test] + fn remove_should_clear_inflight_and_allow_new() { + let store = enabled_store(); + let key = "k-2"; + + assert!(matches!( + store.check_or_register(key), + RequestDedupCheck::New + )); + store.remove(key); + assert!(matches!( + store.check_or_register(key), + RequestDedupCheck::New + )); + } + + #[tokio::test] + async fn stats_should_track_check_wait_and_complete() { + let store = enabled_store(); + let key = "k-stats"; + + assert!(matches!( + store.check_or_register(key), + RequestDedupCheck::New + )); + let notify = match store.check_or_register(key) { + RequestDedupCheck::InProgress { notify } => notify, + other => panic!("expected in progress, got {other:?}"), + }; + + let wait = store.wait_for_completion(key, notify); + store.complete(key, 200, "ok".to_string()); + let replay = wait.await; + assert!(replay.is_some()); + + assert!(matches!( + store.check_or_register(key), + RequestDedupCheck::Completed { .. } + )); + + let stats = store.stats(); + assert_eq!(stats.check_new_total, 1); + assert_eq!(stats.check_in_progress_total, 1); + assert_eq!(stats.check_completed_total, 1); + assert_eq!(stats.wait_success_total, 1); + assert_eq!(stats.wait_timeout_total, 0); + assert_eq!(stats.complete_total, 1); + } +} diff --git a/src-tauri/crates/server/src/middleware/response_cache.rs b/src-tauri/crates/server/src/middleware/response_cache.rs index afeb558d9..a2085ebcd 100644 --- a/src-tauri/crates/server/src/middleware/response_cache.rs +++ b/src-tauri/crates/server/src/middleware/response_cache.rs @@ -1,7 +1,313 @@ -//! 响应缓存能力适配层 +//! 响应缓存（非流式） //! -//! 复用 aster-rust 中的通用实现，避免本地重复维护。 +//! 用于缓存短时间内的完全相同请求响应，降低上游成本与时延。 +//! 典型使用方式： +//! - 请求进入时：按规范化请求生成 key，先查缓存 +//! - 响应返回时：对可缓存状态码（默认仅 200）且体积可接受的响应写入缓存 -pub use aster::network::{ - CachedHttpResponse, ResponseCacheConfig, ResponseCacheStats, ResponseCacheStore, +use indexmap::IndexMap; +use parking_lot::Mutex; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashMap, + time::{Duration, Instant}, }; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResponseCacheConfig { + #[serde(default = "default_enabled")] + pub enabled: bool, + #[serde(default = "default_ttl_secs")] + pub ttl_secs: u64, + #[serde(default = "default_max_entries")] + pub max_entries: usize, + #[serde(default = "default_max_body_bytes")] + pub max_body_bytes: usize, + #[serde(default = "default_cacheable_status_codes")] + pub cacheable_status_codes: Vec, +} + +fn default_enabled() -> bool { + true +} +fn default_ttl_secs() -> u64 { + 600 +} +fn default_max_entries() -> usize { + 200 +} +fn default_max_body_bytes() -> usize { + 1_048_576 +} +fn default_cacheable_status_codes() -> Vec { + vec![200] +} + +impl Default for ResponseCacheConfig { + fn default() -> Self { + Self { + enabled: default_enabled(), + ttl_secs: default_ttl_secs(), + max_entries: default_max_entries(), + max_body_bytes: default_max_body_bytes(), + cacheable_status_codes: default_cacheable_status_codes(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CachedHttpResponse { + pub status: u16, + pub headers: HashMap, + pub body: String, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub struct ResponseCacheStats { + pub size: usize, + pub hits: u64, + pub misses: u64, + pub evictions: u64, +} + +#[derive(Debug, Clone)] +struct CacheEntry { + response: CachedHttpResponse, + cached_at: Instant, +} + +#[derive(Debug, Default, Clone)] +struct CacheCounters { + hits: u64, + misses: u64, + evictions: u64, +} + +pub struct ResponseCacheStore { + config: ResponseCacheConfig, + entries: Mutex>, + counters: Mutex, +} + +impl ResponseCacheStore { + pub fn new(config: ResponseCacheConfig) -> Self { + Self { + config, + entries: Mutex::new(IndexMap::new()), + counters: Mutex::new(CacheCounters::default()), + } + } + + pub fn is_enabled(&self) -> bool { + self.config.enabled + } + + pub fn config(&self) -> ResponseCacheConfig { + self.config.clone() + } + + pub fn should_cache_status(&self, status: u16) -> bool { + self.config.cacheable_status_codes.contains(&status) + } + + pub fn get(&self, key: &str) -> Option { + if !self.config.enabled { + return None; + } + + self.cleanup(); + + let mut entries = self.entries.lock(); + let entry = entries.shift_remove(key); + match entry { + None => { + self.counters.lock().misses += 1; + None + } + Some(entry) => { + let ttl = Duration::from_secs(self.config.ttl_secs); + if entry.cached_at.elapsed() > ttl { + self.counters.lock().misses += 1; + None + } else { + let response = entry.response.clone(); + entries.insert(key.to_string(), entry); + self.counters.lock().hits += 1; + Some(response) + } + } + } + } + + pub fn set(&self, key: &str, response: CachedHttpResponse) -> bool { + if !self.config.enabled { + return false; + } + + if !self.should_cache_status(response.status) { + return false; + } + + if response.body.len() > self.config.max_body_bytes { + return false; + } + + self.cleanup(); + + let mut entries = self.entries.lock(); + entries.shift_remove(key); + entries.insert( + key.to_string(), + CacheEntry { + response, + cached_at: Instant::now(), + }, + ); + + while entries.len() > self.config.max_entries { + if entries.shift_remove_index(0).is_some() { + self.counters.lock().evictions += 1; + } + } + + true + } + + pub fn clear(&self) { + self.entries.lock().clear(); + } + + pub fn cleanup(&self) { + let ttl = Duration::from_secs(self.config.ttl_secs); + self.entries + .lock() + .retain(|_, entry| entry.cached_at.elapsed() <= ttl); + } + + pub fn stats(&self) -> ResponseCacheStats { + let size = self.entries.lock().len(); + let counters = self.counters.lock().clone(); + ResponseCacheStats { + size, + hits: counters.hits, + misses: counters.misses, + evictions: counters.evictions, + } + } + + pub fn hit_rate_percent(&self) -> f64 { + let stats = self.stats(); + let total = stats.hits + stats.misses; + if total == 0 { + 0.0 + } else { + (stats.hits as f64 / total as f64) * 100.0 + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_response(body: &str) -> CachedHttpResponse { + CachedHttpResponse { + status: 200, + headers: HashMap::from([("content-type".to_string(), "application/json".to_string())]), + body: body.to_string(), + } + } + + #[test] + fn should_cache_success_response() { + let store = ResponseCacheStore::new(ResponseCacheConfig::default()); + assert!(store.set("k1", make_response(r#"{"ok":true}"#))); + let got = store.get("k1").expect("cache hit expected"); + assert_eq!(got.status, 200); + assert_eq!(got.body, r#"{"ok":true}"#); + } + + #[test] + fn should_not_cache_error_response() { + let store = ResponseCacheStore::new(ResponseCacheConfig::default()); + let inserted = store.set( + "k2", + CachedHttpResponse { + status: 500, + headers: HashMap::new(), + body: "boom".to_string(), + }, + ); + assert!(!inserted); + assert!(store.get("k2").is_none()); + } + + #[test] + fn should_only_cache_200_by_default() { + let store = ResponseCacheStore::new(ResponseCacheConfig::default()); + let inserted = store.set( + "k200", + CachedHttpResponse { + status: 201, + headers: HashMap::new(), + body: "created".to_string(), + }, + ); + assert!(!inserted); + assert!(store.get("k200").is_none()); + } + + #[test] + fn should_support_custom_cacheable_status_codes() { + let store = ResponseCacheStore::new(ResponseCacheConfig { + enabled: true, + ttl_secs: 600, + max_entries: 10, + max_body_bytes: 1024, + cacheable_status_codes: vec![200, 201, 204], + }); + let inserted = store.set( + "k201", + CachedHttpResponse { + status: 201, + headers: HashMap::new(), + body: "created".to_string(), + }, + ); + assert!(inserted); + assert!(store.get("k201").is_some()); + } + + #[test] + fn should_evict_oldest_when_capacity_exceeded() { + let store = ResponseCacheStore::new(ResponseCacheConfig { + enabled: true, + ttl_secs: 600, + max_entries: 2, + max_body_bytes: 1024, + cacheable_status_codes: vec![200], + }); + assert!(store.set("k1", make_response("1"))); + assert!(store.set("k2", make_response("2"))); + assert!(store.set("k3", make_response("3"))); + + assert!(store.get("k1").is_none()); + assert!(store.get("k2").is_some()); + assert!(store.get("k3").is_some()); + assert!(store.stats().evictions >= 1); + } + + #[test] + fn should_expire_entries_by_ttl() { + let store = ResponseCacheStore::new(ResponseCacheConfig { + enabled: true, + ttl_secs: 1, + max_entries: 10, + max_body_bytes: 1024, + cacheable_status_codes: vec![200], + }); + assert!(store.set("k4", make_response("x"))); + std::thread::sleep(Duration::from_millis(1100)); + assert!(store.get("k4").is_none()); + } +}