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.
This commit is contained in:
Jason
2026-07-20 12:15:32 +08:00
parent 01fca69641
commit eb105eae76
8 changed files with 148 additions and 96 deletions
+88 -7
View File
@@ -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<String>,
}
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<tokio::sync::Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
}
fn merge_sync_step(
aggregate: &mut SessionSyncResult,
name: &str,
step: Result<SessionSyncResult, AppError>,
) {
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<SessionSyncResult, AppE
imported: 0,
skipped: 0,
files_scanned: 0,
suspected_duplicates: 0,
deferred_files: 0,
errors: vec![],
});
}
@@ -73,6 +137,8 @@ pub fn sync_claude_session_logs(db: &Database) -> Result<SessionSyncResult, AppE
imported: 0,
skipped: 0,
files_scanned: 0,
suspected_duplicates: 0,
deferred_files: 0,
errors: vec![],
};
@@ -513,12 +579,7 @@ fn insert_session_log_entry(
)
.map_err(|e| AppError::Database(format!("插入会话日志失败: {e}")))?;
// 仅在确实写入新行时通知前端,避免 INSERT OR IGNORE 跳过时产生空刷新
if inserted_rows > 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<Vec<DataSourceSummary>
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"}"#;
@@ -307,6 +307,8 @@ pub fn sync_codex_usage(db: &Database) -> Result<SessionSyncResult, AppError> {
skipped: 0,
files_scanned: files.len() as u32,
errors: vec![],
suspected_duplicates: 0,
deferred_files: 0,
};
if files.is_empty() {
@@ -46,6 +46,8 @@ pub fn sync_gemini_usage(db: &Database) -> Result<SessionSyncResult, AppError> {
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)
}
@@ -50,6 +50,8 @@ pub fn sync_opencode_usage(db: &Database) -> Result<SessionSyncResult, AppError>
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<SessionSyncResult, AppError>
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<SessionSyncResult, AppError>
imported: 0,
skipped: 0,
files_scanned: 1,
suspected_duplicates: 0,
deferred_files: 0,
errors: vec![],
};
let mut has_sync_errors = false;