From eb105eae76415b9d61cf48a28be021823840be86 Mon Sep 17 00:00:00 2001 From: Jason Date: Mon, 20 Jul 2026 12:15:32 +0800 Subject: [PATCH] perf(usage): coalesce session-sync notifications and serialize sync execution Replace per-insert usage notifications with a single notification per sync pass, guard all sync entry points (startup backfill, 60s loop, manual sync) behind a process-wide single-flight tokio mutex, and move blocking file/DB work onto spawn_blocking with MissedTickBehavior::Skip. Extend SessionSyncResult with deferredFiles/suspectedDuplicates observability fields (saturating aggregation) and add a test seam that counts notifications even without an injected AppHandle. --- src-tauri/src/commands/usage.rs | 54 ++--------- src-tauri/src/lib.rs | 67 +++++-------- src-tauri/src/services/session_usage.rs | 95 +++++++++++++++++-- src-tauri/src/services/session_usage_codex.rs | 2 + .../src/services/session_usage_gemini.rs | 5 +- .../src/services/session_usage_opencode.rs | 6 ++ src-tauri/src/usage_events.rs | 13 +++ src/types/usage.ts | 2 + 8 files changed, 148 insertions(+), 96 deletions(-) diff --git a/src-tauri/src/commands/usage.rs b/src-tauri/src/commands/usage.rs index d968bbdc5..e85e71ca4 100644 --- a/src-tauri/src/commands/usage.rs +++ b/src-tauri/src/commands/usage.rs @@ -275,52 +275,18 @@ pub fn delete_model_pricing(state: State<'_, AppState>, model_id: String) -> Res /// 手动触发会话日志同步 #[tauri::command] -pub fn sync_session_usage( +pub async fn sync_session_usage( state: State<'_, AppState>, ) -> Result { - // 同步 Claude 会话日志 - let mut result = crate::services::session_usage::sync_claude_session_logs(&state.db)?; - - // 同步 Codex 使用数据 - match crate::services::session_usage_codex::sync_codex_usage(&state.db) { - Ok(codex_result) => { - result.imported += codex_result.imported; - result.skipped += codex_result.skipped; - result.files_scanned += codex_result.files_scanned; - result.errors.extend(codex_result.errors); - } - Err(e) => { - result.errors.push(format!("Codex 同步失败: {e}")); - } - } - - // 同步 Gemini 使用数据 - match crate::services::session_usage_gemini::sync_gemini_usage(&state.db) { - Ok(gemini_result) => { - result.imported += gemini_result.imported; - result.skipped += gemini_result.skipped; - result.files_scanned += gemini_result.files_scanned; - result.errors.extend(gemini_result.errors); - } - Err(e) => { - result.errors.push(format!("Gemini 同步失败: {e}")); - } - } - - // 同步 OpenCode 使用数据 - match crate::services::session_usage_opencode::sync_opencode_usage(&state.db) { - Ok(opencode_result) => { - result.imported += opencode_result.imported; - result.skipped += opencode_result.skipped; - result.files_scanned += opencode_result.files_scanned; - result.errors.extend(opencode_result.errors); - } - Err(e) => { - result.errors.push(format!("OpenCode 同步失败: {e}")); - } - } - - Ok(result) + let db = state.db.clone(); + let _guard = crate::services::session_usage::session_sync_mutex() + .lock() + .await; + tauri::async_runtime::spawn_blocking(move || { + crate::services::session_usage::sync_all_unlocked(&db) + }) + .await + .map_err(|error| AppError::Message(format!("会话用量同步任务失败: {error}"))) } /// 获取数据来源分布 diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index d5b866522..05bcd75bc 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1214,59 +1214,42 @@ pub fn run() { tauri::async_runtime::spawn(async move { const SESSION_SYNC_INTERVAL_SECS: u64 = 60; - fn run_step(name: &str, result: Result) { - if let Err(e) = result { - log::warn!("{name} failed: {e}"); + async fn run_session_sync(db: std::sync::Arc, backfill: bool) { + let _guard = crate::services::session_usage::session_sync_mutex() + .lock() + .await; + let task = tauri::async_runtime::spawn_blocking(move || { + if backfill { + if let Err(error) = db.backfill_missing_usage_costs() { + log::warn!("Usage cost startup backfill failed: {error}"); + } + } + crate::services::session_usage::sync_all_unlocked(&db) + }); + match task.await { + Ok(result) if !result.errors.is_empty() => { + log::warn!( + "Session usage sync completed with {} error(s)", + result.errors.len() + ); + } + Ok(_) => {} + Err(error) => log::warn!("Session usage blocking task failed: {error}"), } } - let db = &db_for_session_sync; - - // 首次同步 - run_step( - "Usage cost startup backfill", - db.backfill_missing_usage_costs(), - ); - run_step( - "Session usage initial sync", - crate::services::session_usage::sync_claude_session_logs(db), - ); - run_step( - "Codex usage initial sync", - crate::services::session_usage_codex::sync_codex_usage(db), - ); - run_step( - "Gemini usage initial sync", - crate::services::session_usage_gemini::sync_gemini_usage(db), - ); - run_step( - "OpenCode usage initial sync", - crate::services::session_usage_opencode::sync_opencode_usage(db), - ); + // 首次同步（含费用回填） + run_session_sync(db_for_session_sync.clone(), true).await; // 定期同步 let mut interval = tokio::time::interval(std::time::Duration::from_secs( SESSION_SYNC_INTERVAL_SECS, )); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); interval.tick().await; // skip immediate first tick loop { interval.tick().await; - run_step( - "Session usage periodic sync", - crate::services::session_usage::sync_claude_session_logs(db), - ); - run_step( - "Codex usage periodic sync", - crate::services::session_usage_codex::sync_codex_usage(db), - ); - run_step( - "Gemini usage periodic sync", - crate::services::session_usage_gemini::sync_gemini_usage(db), - ); - run_step( - "OpenCode usage periodic sync", - crate::services::session_usage_opencode::sync_opencode_usage(db), - ); + run_session_sync(db_for_session_sync.clone(), false).await; } }); }); diff --git a/src-tauri/src/services/session_usage.rs b/src-tauri/src/services/session_usage.rs index 7d8756722..37e414964 100644 --- a/src-tauri/src/services/session_usage.rs +++ b/src-tauri/src/services/session_usage.rs @@ -22,18 +22,80 @@ use std::collections::HashMap; use std::fs; use std::io::{BufRead, BufReader}; use std::path::{Path, PathBuf}; +use std::sync::OnceLock; use std::time::SystemTime; /// 同步结果 -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct SessionSyncResult { pub imported: u32, pub skipped: u32, pub files_scanned: u32, + pub suspected_duplicates: u32, + pub deferred_files: u32, pub errors: Vec, } +impl SessionSyncResult { + pub fn merge(&mut self, other: SessionSyncResult) { + self.imported = self.imported.saturating_add(other.imported); + self.skipped = self.skipped.saturating_add(other.skipped); + self.files_scanned = self.files_scanned.saturating_add(other.files_scanned); + self.suspected_duplicates = self + .suspected_duplicates + .saturating_add(other.suspected_duplicates); + self.deferred_files = self.deferred_files.saturating_add(other.deferred_files); + self.errors.extend(other.errors); + } +} + +pub fn session_sync_mutex() -> &'static tokio::sync::Mutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| tokio::sync::Mutex::new(())) +} + +fn merge_sync_step( + aggregate: &mut SessionSyncResult, + name: &str, + step: Result, +) { + match step { + Ok(result) => aggregate.merge(result), + Err(error) => aggregate.errors.push(format!("{name} 同步失败: {error}")), + } +} + +/// 调用方必须持有 [`session_sync_mutex`]。此函数是同步内核，供后台任务、 +/// 手动同步和 Codex 重建共享，避免 tokio Mutex 重入。 +pub fn sync_all_unlocked(db: &Database) -> SessionSyncResult { + let mut result = SessionSyncResult::default(); + merge_sync_step(&mut result, "Claude", sync_claude_session_logs(db)); + merge_sync_step( + &mut result, + "Codex", + crate::services::session_usage_codex::sync_codex_usage(db), + ); + merge_sync_step( + &mut result, + "Gemini", + crate::services::session_usage_gemini::sync_gemini_usage(db), + ); + merge_sync_step( + &mut result, + "OpenCode", + crate::services::session_usage_opencode::sync_opencode_usage(db), + ); + notify_sync_result(&result); + result +} + +pub(crate) fn notify_sync_result(result: &SessionSyncResult) { + if result.imported > 0 { + crate::usage_events::notify_log_recorded(); + } +} + /// 数据来源分布 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -65,6 +127,8 @@ pub fn sync_claude_session_logs(db: &Database) -> Result Result 0 { - crate::usage_events::notify_log_recorded(); - } - - Ok(true) + Ok(inserted_rows > 0) } /// 从 model_pricing 表查找模型定价（支持模糊匹配） @@ -565,6 +626,26 @@ pub fn get_data_source_breakdown(db: &Database) -> Result mod tests { use super::*; + #[test] + fn sync_result_notification_is_coalesced_to_one_call() { + crate::usage_events::take_test_notify_count(); + notify_sync_result(&SessionSyncResult::default()); + let result = SessionSyncResult { + imported: 25, + ..SessionSyncResult::default() + }; + notify_sync_result(&result); + assert_eq!(crate::usage_events::take_test_notify_count(), 1); + } + + #[tokio::test] + async fn session_sync_mutex_serializes_callers() { + let first = session_sync_mutex().lock().await; + assert!(session_sync_mutex().try_lock().is_err()); + drop(first); + assert!(session_sync_mutex().try_lock().is_ok()); + } + #[test] fn test_parse_usage_from_jsonl_line() { let line = r#"{"type":"assistant","message":{"id":"msg_test123","model":"claude-opus-4-6","usage":{"input_tokens":3,"output_tokens":150,"cache_read_input_tokens":5000,"cache_creation_input_tokens":10000},"stop_reason":"end_turn"},"timestamp":"2026-04-05T12:00:00Z","sessionId":"session-abc"}"#; diff --git a/src-tauri/src/services/session_usage_codex.rs b/src-tauri/src/services/session_usage_codex.rs index 5757f9ded..92e5b6935 100644 --- a/src-tauri/src/services/session_usage_codex.rs +++ b/src-tauri/src/services/session_usage_codex.rs @@ -307,6 +307,8 @@ pub fn sync_codex_usage(db: &Database) -> Result { skipped: 0, files_scanned: files.len() as u32, errors: vec![], + suspected_duplicates: 0, + deferred_files: 0, }; if files.is_empty() { diff --git a/src-tauri/src/services/session_usage_gemini.rs b/src-tauri/src/services/session_usage_gemini.rs index c904726f2..b4d7df69c 100644 --- a/src-tauri/src/services/session_usage_gemini.rs +++ b/src-tauri/src/services/session_usage_gemini.rs @@ -46,6 +46,8 @@ pub fn sync_gemini_usage(db: &Database) -> Result { imported: 0, skipped: 0, files_scanned: files.len() as u32, + suspected_duplicates: 0, + deferred_files: 0, errors: vec![], }; @@ -351,9 +353,6 @@ fn insert_gemini_session_entry( // changes() > 0 表示新插入或已更新，== 0 表示值完全相同（无实际变更） let changed = conn.changes() > 0; - if changed { - crate::usage_events::notify_log_recorded(); - } Ok(changed) } diff --git a/src-tauri/src/services/session_usage_opencode.rs b/src-tauri/src/services/session_usage_opencode.rs index 83574bb62..92934e58f 100644 --- a/src-tauri/src/services/session_usage_opencode.rs +++ b/src-tauri/src/services/session_usage_opencode.rs @@ -50,6 +50,8 @@ pub fn sync_opencode_usage(db: &Database) -> Result imported: 0, skipped: 0, files_scanned: 0, + suspected_duplicates: 0, + deferred_files: 0, errors: vec![], }); } @@ -77,6 +79,8 @@ pub fn sync_opencode_usage(db: &Database) -> Result imported: 0, skipped: 0, files_scanned: 1, + suspected_duplicates: 0, + deferred_files: 0, errors: vec![], }); } @@ -90,6 +94,8 @@ pub fn sync_opencode_usage(db: &Database) -> Result imported: 0, skipped: 0, files_scanned: 1, + suspected_duplicates: 0, + deferred_files: 0, errors: vec![], }; let mut has_sync_errors = false; diff --git a/src-tauri/src/usage_events.rs b/src-tauri/src/usage_events.rs index d5487bfe2..e1f6aa3b3 100644 --- a/src-tauri/src/usage_events.rs +++ b/src-tauri/src/usage_events.rs @@ -44,6 +44,9 @@ pub fn init(handle: AppHandle) { /// 调用方**不**需要持有 AppHandle，可以从任意线程/任意写入路径调用。 /// 内部 200ms 防抖合并，绝不阻塞调用线程。 pub fn notify_log_recorded() { + #[cfg(test)] + TEST_NOTIFY_COUNT.with(|count| count.set(count.get().saturating_add(1))); + // AppHandle 未注入（典型出现在单元测试或 setup 之前）：直接放弃。 let Some(handle) = APP_HANDLE.get() else { return; @@ -66,3 +69,13 @@ pub fn notify_log_recorded() { } }); } + +#[cfg(test)] +thread_local! { + static TEST_NOTIFY_COUNT: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +#[cfg(test)] +pub(crate) fn take_test_notify_count() -> u32 { + TEST_NOTIFY_COUNT.with(|count| count.replace(0)) +} diff --git a/src/types/usage.ts b/src/types/usage.ts index b791a34df..840fdf89f 100644 --- a/src/types/usage.ts +++ b/src/types/usage.ts @@ -40,6 +40,8 @@ export interface SessionSyncResult { imported: number; skipped: number; filesScanned: number; + suspectedDuplicates: number; + deferredFiles: number; errors: string[]; }