修复长时间运行的内存与数据库开销问题

desktop-tauri/src-tauri/server/src/server/api/chat.rs

@@ -114,9 +114,15 @@ pub async fn chat_completions(
         .map(|items| items.len())
         .unwrap_or(0);
     let user_agent = pipeline::user_agent_of(&headers);
+    let body_len = body.len();
+    let dedupe_key = pipeline::sha256_hex(&body);
+    let raw_request = pipeline::capture_raw_body(&body);
 
     // ③ 调试落盘：保存最近一次入站请求体，供重放分析（覆盖写）
     write_debug_files(&body, method, path, &user_agent);
+    // 原始字节已全部转成后续步骤所需的值，解析后的 payload 已持有正文，
+    // 此时释放 Bytes 可把百 MB 级请求的峰值内存降下一份
+    drop(body);
 
     // ④ 模型路由走公共管道（默认模型回落 / 别名映射 / 目录校验，
     //    与另外两条协议入口逐条同源）。下游原始请求名在改写前取一次：
@@ -146,14 +152,13 @@ pub async fn chat_completions(
         &format!(
             "← {method} {path} model={} stream={stream} msgs={message_count} bytes={} ua={user_agent}",
             if requested_model.is_empty() { "(未指定)" } else { &requested_model },
-            body.len(),
+            body_len,
         ),
     );
 
     // ⑤ 转发：请求体原样交给转发层（内容处理在每家 provider 发送前按作用范围
     // 逐家判定，见 `core::upstream::payload` 的 `send_body`）；去重键取**客户端
     // 原始请求体**的哈希，与「某一家是否被处理」无关。
-    let dedupe_key = pipeline::sha256_hex(&body);
     let telemetry = Arc::new(RequestTelemetry::with_id());
     // 「进行中」行：模型解析已成功、转发即将开始 —— 先插一条 status=0 的明细，
     // 让请求日志在响应还没跑完时就能看到这条请求（收尾时由 `record` 的 UPDATE
@@ -192,10 +197,9 @@ pub async fn chat_completions(
         telemetry_id,
         started_at,
     ));
-    // 下游原始请求体在此刻抄一份（request_raw 表的请求侧，见 `capture_raw_body`）：
-    // body 还是客户端发来的原值；送进转发层后 payload 会被就地改写（默认模型注入）。
-    // **仅调试模式开启时采集** —— 关着时这里是零开销的 None（预览对话没有正文）
-    let raw_request = pipeline::capture_raw_body(&body);
+    // 下游原始请求体已在解析后、释放 Bytes 前按原样抄入 raw_request
+    // （request_raw 表的请求侧，见 `capture_raw_body`），转发层的就地改写
+    // 不会影响这份快照，调试关闭时它是零开销的 None
     // 自动续写的上下文（开关 `autoContinue` **默认关**）。关着时这里连一份 clone
     // 都不做 —— `plan()` 先问开关，再决定要不要复制请求体/请求头/白名单。
     // 判据、多轮拼接与用量合并口径都在 `core::upstream::continuation`。
@@ -490,4 +494,4 @@ pub fn spawn_catalog_refresh(state: &ServerState) {
     crate::spawn_task(async move {
         crate::server::core::providers::adapter::refresh_implemented(&store).await;
     });
-}
+}

desktop-tauri/src-tauri/server/src/server/api/debug_api.rs

@@ -97,6 +97,6 @@ fn debug_json() -> Value {
     json!({
         config::KEY_DEBUG_MODE: config::current().debug_mode(),
         "count": debug_traffic::count(),
-        "limit": debug_traffic::MAX_ENTRIES,
+        "limit": debug_traffic::max_entries(),
     })
 }

desktop-tauri/src-tauri/server/src/server/api/live_progress.rs

@@ -1,15 +1,28 @@
 use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
 
 use crate::server::request_stats::{RequestStats, RunningProgress};
 
+/// 在途状态的最小写库间隔：状态变化仍保留，最多合并这段时间内的快照
+const MIN_WRITE_INTERVAL: Duration = Duration::from_millis(1000);
+
 pub(super) fn sink(stats: Arc<RequestStats>, id: String) -> impl Fn(RunningProgress) + Send + Sync {
     let (sender, mut receiver) = tokio::sync::mpsc::channel::<()>(1);
-    let pending = Arc::new(Mutex::new(None::<RunningProgress>));
-    let worker = pending.clone();
-    let _ = crate::spawn_task(async move {
+    let latest = Arc::new(Mutex::new(None::<RunningProgress>));
+    let worker = latest.clone();
+    crate::spawn_task(async move {
+        let mut last_write = Instant::now() - MIN_WRITE_INTERVAL;
         while receiver.recv().await.is_some() {
+            // 先等窗口结束再取快照，把窗口内的多次状态变化合并成最后一次
+            let wait = MIN_WRITE_INTERVAL.saturating_sub(last_write.elapsed());
+            if !wait.is_zero() {
+                tokio::time::sleep(wait).await;
+            }
+            // 等待期间最后一条快照已覆盖前值；把尚未消费的唤醒合并在本次提交里
+            while receiver.try_recv().is_ok() {}
             let progress = worker.lock().unwrap_or_else(|error| error.into_inner()).take();
             if let Some(progress) = progress {
+                last_write = Instant::now();
                 let stats = stats.clone();
                 let id = id.clone();
                 if let Err(error) = tokio::task::spawn_blocking(move || stats.update_running(&id, &progress)).await {
@@ -18,9 +31,11 @@ pub(super) fn sink(stats: Arc<RequestStats>, id: String) -> impl Fn(RunningProgr
             }
         }
     });
+    let latest_for_sink = latest.clone();
     move |progress| {
-        // 只保留最后一次完整快照；终态写入后的 UPDATE 由 status=0 条件挡住。
-        *pending.lock().unwrap_or_else(|error| error.into_inner()) = Some(progress);
+        // 只保留最后一次完整快照；终态写入后的 UPDATE 由 status=0 条件挡住，
+        // 一秒内的连续变化由工作线程合并后只提交一次
+        *latest_for_sink.lock().unwrap_or_else(|error| error.into_inner()) = Some(progress);
         let _ = sender.try_send(());
     }
 }

desktop-tauri/src-tauri/server/src/server/api/mod.rs

@@ -126,7 +126,6 @@ pub mod onboarding;
 pub mod panel;
 pub mod pipeline;
 mod live_progress;
-mod record_persist;
 // 提供商插件包的管理面（列表 / 安装 / 卸载 / 启停）：插件是用户安装的，
 // 这一组端点就是「装插件」的唯一入口，见该文件模块头
 pub mod plugins;

desktop-tauri/src-tauri/server/src/server/api/pipeline.rs

@@ -36,7 +36,7 @@ use crate::server::core::upstream::usage::{self, RequestTelemetry, TelemetrySnap
 use crate::server::errors::GatewayError;
 use crate::server::logging;
 use crate::server::request_stats::{
-    AttemptDetail, NewRequestEntry, RequestStats, RetryEvent, RunningProgress, SensitiveHit,
+    AttemptDetail, NewRequestEntry, RequestStats, RetryEvent, SensitiveHit,
 };
 use crate::server::ServerState;
 
@@ -48,6 +48,8 @@ use raw::RawCapture;
 const DEBUG_DIR: &str = "debug";
 const DEBUG_REQUEST_FILE: &str = "last-request.json";
 const DEBUG_META_FILE: &str = "last-request.meta.txt";
+/// 调试快照文件的开关状态：0 = 未判定、1 = 当前开启、2 = 当前关闭
+static DEBUG_SNAPSHOT_STATE: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
 
 /// 非调试请求的错误摘要上限；调试请求保留完整错误。
 const ERROR_SUMMARY_CHARS: usize = 200;
@@ -495,32 +497,22 @@ pub fn live_row_sink(
     id: String,
     started_at: i64,
 ) -> impl Fn(&TelemetrySnapshot) + Send + Sync {
+    let live = super::live_progress::sink(stats.clone(), id.clone());
     move |snapshot| {
-        stats.update_running(
-            &id,
-            &RunningProgress {
-                provider: snapshot.provider.clone().unwrap_or_default(),
-                account_id: snapshot.account_id.clone(),
-                account_name: snapshot.account_name.clone(),
-                upstream_model: snapshot.upstream_model.clone(),
-                upstream_reasoning: snapshot.upstream_reasoning.clone(),
-                // 与收尾同口径：一次都没发出去（0）按 1 次算，理由见
-                // `TelemetrySnapshot::attempts` 的说明
-                attempts: snapshot.attempts.max(1),
-                first_response_ms: snapshot.first_response_at.map(|at| (at - started_at).max(0)),
-                // 阶段与阶段起点：起点为 None 只可能是「还停在初始阶段（连接中）、
-                // 一次都没切换过」——那个阶段的起点就是**请求开始时刻**（下面这个
-                // started_at，与插入在途行时写进 phase_started_at 的值同源），
-                // 所以用 started_at 兜底。不兜的话这一列会写进 NULL，
-                // 而状态列的第二行按它算「连接中已持续多久」会算出个负数或
-                // 干脆没有 —— 两种都不该出现在「已受理但还没发出去」的请求上
-                phase: snapshot.phase.as_str().to_string(),
-                phase_started_at: snapshot.phase_started_at.or(Some(started_at)),
-                attempt_details: stored_attempt_details(&snapshot.attempts_detail),
-                route: snapshot.route.clone(),
-                sensitive_hits: stored_sensitive_hits(&snapshot.sensitive_hits),
-            },
-        );
+        live(crate::server::request_stats::RunningProgress {
+            provider: snapshot.provider.clone().unwrap_or_default(),
+            account_id: snapshot.account_id.clone(),
+            account_name: snapshot.account_name.clone(),
+            upstream_model: snapshot.upstream_model.clone(),
+            upstream_reasoning: snapshot.upstream_reasoning.clone(),
+            attempts: snapshot.attempts.max(1),
+            first_response_ms: snapshot.first_response_at.map(|at| (at - started_at).max(0)),
+            phase: snapshot.phase.as_str().to_string(),
+            phase_started_at: snapshot.phase_started_at.or(Some(started_at)),
+            attempt_details: stored_attempt_details(&snapshot.attempts_detail),
+            route: snapshot.route.clone(),
+            sensitive_hits: stored_sensitive_hits(&snapshot.sensitive_hits),
+        });
     }
 }
 
@@ -662,11 +654,15 @@ pub fn record_entry(context: &RecordContext, fallback_error: Option<String>) {
     let limited_model = if entry.upstream_model.is_empty() { entry.model.clone() } else { entry.upstream_model.clone() };
     let limited_request = entry.id.clone();
     let sent = snapshot.attempts > 0;
-    context.stats.record(entry);
-    if sent {
-        limiter::counter::note(&limited_account, &limited_request, &limited_model,
-            &snapshot.quota_channel, limited_tokens, context.started_at);
-    }
+    let limiter_event = sent.then_some((
+        limited_account.as_str(),
+        limited_request.as_str(),
+        limited_model.as_str(),
+        snapshot.quota_channel.as_str(),
+        limited_tokens,
+        context.started_at,
+    ));
+    context.stats.record(entry, limiter_event);
     if !limited_account.is_empty() {
         limiter::note_request_tokens(&limited_account, limited_tokens, finished_at);
     }
@@ -1009,13 +1005,26 @@ pub fn error_response(
 /// 调试落盘：原始 body + 一行 meta（覆盖写；**失败不影响请求**）。
 ///
 /// 与 Node 版一致：目录不存在时创建，任何 IO 失败都静默吞掉 ——
-/// 调试落盘是排障辅助，不能因为它失败就让用户的请求失败。
+/// 调试落盘是排障辅助，不能因为它失败就让用户的请求失败，超过 8 MiB 的
+/// 请求只写元数据，避免每次请求都把百 MB 级正文复制到群晖磁盘
 pub fn write_debug_files(body: &[u8], method: &str, path: &str, user_agent: &str) {
     let dir = config::config_dir().join(DEBUG_DIR);
+    use std::sync::atomic::Ordering;
+    if !crate::server::core::debug_traffic::enabled() {
+        if DEBUG_SNAPSHOT_STATE.swap(2, Ordering::Relaxed) != 2 {
+            cleanup_debug_files();
+        }
+        return;
+    }
+    DEBUG_SNAPSHOT_STATE.store(1, Ordering::Relaxed);
     if std::fs::create_dir_all(&dir).is_err() {
         return;
     }
-    let _ = std::fs::write(dir.join(DEBUG_REQUEST_FILE), body);
+    if body.len() <= crate::server::core::debug_traffic::MAX_BODY_BYTES {
+        let _ = std::fs::write(dir.join(DEBUG_REQUEST_FILE), body);
+    } else {
+        let _ = std::fs::remove_file(dir.join(DEBUG_REQUEST_FILE));
+    }
     let meta = format!(
         "{} {method} {path} {}B ua={user_agent}",
         iso_timestamp(),
@@ -1024,6 +1033,20 @@ pub fn write_debug_files(body: &[u8], method: &str, path: &str, user_agent: &str
     let _ = std::fs::write(dir.join(DEBUG_META_FILE), meta);
 }
 
+/// 调试关闭时清理历史快照，避免关闭后仍长期占着上次的大请求正文
+pub fn cleanup_debug_files_if_disabled() {
+    if !crate::server::core::debug_traffic::enabled() {
+        cleanup_debug_files();
+        DEBUG_SNAPSHOT_STATE.store(2, std::sync::atomic::Ordering::Relaxed);
+    }
+}
+
+fn cleanup_debug_files() {
+    let dir = config::config_dir().join(DEBUG_DIR);
+    let _ = std::fs::remove_file(dir.join(DEBUG_REQUEST_FILE));
+    let _ = std::fs::remove_file(dir.join(DEBUG_META_FILE));
+}
+
 /// ISO-8601 UTC 时间戳（对应 Node 的 `new Date().toISOString()`）
 pub fn iso_timestamp() -> String {
     let millis = logging::now_ms();
@@ -1236,6 +1259,10 @@ mod tests {
             limit: Some(5),
             model: None,
             provider: None,
+            client: None,
+            api_key: None,
+            production_only: false,
+            completed_only: false,
             status: None,
             start: None,
             end: None,

desktop-tauri/src-tauri/server/src/server/api/pipeline/raw.rs

@@ -1,9 +1,14 @@
-//! 调试模式下的下游正文采集，不对日志副本设置单条大小上限。
+//! 调试模式下的下游正文采集，单条大小上限与上游调试报文保持一致。
+
+const MAX_RAW_BYTES: usize = crate::server::core::debug_traffic::MAX_BODY_BYTES;
 
 pub fn raw_body_text(bytes: &[u8]) -> Option<String> {
     if bytes.is_empty() {
         return None;
     }
+    if bytes.len() > MAX_RAW_BYTES {
+        return None;
+    }
     Some(String::from_utf8_lossy(bytes).into_owned())
 }
 
@@ -18,7 +23,8 @@ pub fn capture_raw_json<T: serde::Serialize>(value: &T) -> Option<String> {
     if !crate::server::core::debug_traffic::enabled() {
         return None;
     }
-    serde_json::to_string(value).ok()
+    let text = serde_json::to_string(value).ok()?;
+    (text.len() <= MAX_RAW_BYTES).then_some(text)
 }
 
 pub(super) struct RawCapture {
@@ -32,8 +38,12 @@ impl RawCapture {
     }
 
     pub(super) fn push(&mut self, chunk: &[u8]) {
-        if self.enabled {
+        if self.enabled && self.buf.len().saturating_add(chunk.len()) <= MAX_RAW_BYTES {
             self.buf.extend_from_slice(chunk);
+        } else if self.enabled {
+            // 超过上限后停止累积，避免超大响应流把整份正文复制进内存
+            self.enabled = false;
+            self.buf.clear();
         }
     }
 
@@ -49,3 +59,16 @@ impl RawCapture {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{RawCapture, MAX_RAW_BYTES};
+
+    #[test]
+    fn oversized_stream_stops_accumulating() {
+        let mut capture = RawCapture::new(true);
+        let chunk = vec![b'x'; MAX_RAW_BYTES + 1];
+        capture.push(&chunk);
+        assert!(capture.into_text().is_none());
+    }
+}