mirror of
https://github.com/aiclientproxy/proxycast.git
synced 2026-09-24 23:10:56 +08:00
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) <noreply@anthropic.com>
This commit is contained in:
coso
Generated
+1
@@ -7345,6 +7345,7 @@ dependencies = [
|
||||
"dirs 5.0.1",
|
||||
"futures",
|
||||
"hex",
|
||||
"indexmap 2.13.0",
|
||||
"once_cell",
|
||||
"parking_lot",
|
||||
"proptest",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<I>(&self, reasons: I)
|
||||
where
|
||||
I: IntoIterator<Item = CapabilityFilterExcludedReason>,
|
||||
{
|
||||
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()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Regex> = 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<Notify> },
|
||||
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<Notify>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct CompletedEntry {
|
||||
status: u16,
|
||||
body: String,
|
||||
completed_at: Instant,
|
||||
}
|
||||
|
||||
pub struct RequestDedupStore {
|
||||
config: RequestDedupConfig,
|
||||
inflight: Mutex<HashMap<String, InflightEntry>>,
|
||||
completed: Mutex<HashMap<String, CompletedEntry>>,
|
||||
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<Notify>,
|
||||
) -> Option<CompletedReplay> {
|
||||
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<String, Value>) -> 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<u16>,
|
||||
}
|
||||
|
||||
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<u16> {
|
||||
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<String, String>,
|
||||
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<IndexMap<String, CacheEntry>>,
|
||||
counters: Mutex<CacheCounters>,
|
||||
}
|
||||
|
||||
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<CachedHttpResponse> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user