From 9cf3878b2862986e2d3d219531e4de41d0abb5d5 Mon Sep 17 00:00:00 2001 From: lzt404 <2596933790@qq.com> Date: Sun, 6 Sep 2026 01:09:02 +0800 Subject: [PATCH] perf: reduce background overhead and improve UI responsiveness - Move blocking usage queries and core lifecycle operations to background workers, limit concurrency, and prevent overlapping core operations - Coalesce statistics refreshes and consolidate category aggregations, reducing unfiltered analysis queries from 11 to 4 - Cache bundled model prices, reuse management API connections, and reduce redundant checks in the usage collector - Throttle download progress and guide positioning updates, and avoid redundant renders when core status is unchanged - Bound and coalesce configuration watcher events, and reuse a single model catalog refresh worker - Stop scheduling remaining health checks when the dialog closes and limit backend health-check concurrency - Replace overlapping OAuth polling with sequential requests, retry backoff, timeouts, and stale callback protection - Add regression tests and a statistics benchmark script, and move performance verification documentation to agent_docs --- scripts/benchmark-usage.ts | 64 ++++ src-tauri/src/agents/commands.rs | 12 +- src-tauri/src/app_settings.rs | 16 +- src-tauri/src/app_update.rs | 23 +- src-tauri/src/configuration_watcher.rs | 197 ++++++++--- src-tauri/src/core_runtime.rs | 184 +++++----- src-tauri/src/main.rs | 4 + src-tauri/src/management_api.rs | 28 +- src-tauri/src/progress.rs | 36 ++ src-tauri/src/provider_health.rs | 5 + src-tauri/src/usage.rs | 455 +++++++++++++++++++------ src/coreRuntime.tsx | 32 +- src/pages/ApiAccessPage.tsx | 19 +- src/pages/EasyModePage.tsx | 73 +++- src/pages/UsageRecordsPage.tsx | 23 +- src/services/providerHealthCheck.ts | 8 +- src/services/refreshScheduler.ts | 66 ++++ tests/providerHealthCheck.test.ts | 37 ++ tests/refreshScheduler.test.ts | 68 ++++ 19 files changed, 1050 insertions(+), 300 deletions(-) create mode 100644 scripts/benchmark-usage.ts create mode 100644 src-tauri/src/progress.rs create mode 100644 src/services/refreshScheduler.ts create mode 100644 tests/refreshScheduler.test.ts diff --git a/scripts/benchmark-usage.ts b/scripts/benchmark-usage.ts new file mode 100644 index 0000000..0d0bddf --- /dev/null +++ b/scripts/benchmark-usage.ts @@ -0,0 +1,64 @@ +import { Database } from 'bun:sqlite'; +import { readFileSync } from 'node:fs'; + +const source = readFileSync(new URL('../src-tauri/src/usage.rs', import.meta.url), 'utf8'); +function sqlForFunction(name: string) { + const start = source.indexOf('fn ' + name + '('); + if (start < 0) throw new Error('Missing function: ' + name); + const remaining = source.slice(start); + const next = remaining.indexOf('\nfn ', 1); + return [...remaining.slice(0, next < 0 ? undefined : next).matchAll(/r#"([\s\S]*?)"#/g)] + .map((match) => match[1]!); +} + +const database = new Database(':memory:'); +database.exec(sqlForFunction('initialize_usage_schema')[0]!); +database.exec('CREATE INDEX idx_usage_events_canceled_timestamp ON usage_events(canceled, timestamp_ms DESC)'); +const filter = ' WHERE timestamp_ms >= 0 AND timestamp_ms <= 1000000000000'; +const expand = (sql: string) => sql.replaceAll('{}', filter) + .replaceAll('{LONG_CONTEXT_INPUT_TOKEN_THRESHOLD}', '272000'); +const categorySql = sqlForFunction('load_simple_categories')[0]!; +const previousCategories = ['model', 'provider', 'source'].map((column) => ({ + statement: database.query(expand(categorySql.replaceAll('{column}', column))), args: ['unknown'], +})); +previousCategories.push({ + statement: database.query(expand(sqlForFunction('load_api_key_categories')[0]!)), args: [], +}); +const overview = [ + sqlForFunction('load_usage_overview')[0]!, + sqlForFunction('load_usage_cost_groups')[0]!, + sqlForFunction('load_usage_overview')[1]!, +].map((sql) => ({ statement: database.query(expand(sql)), args: [] as string[] })); +const previous = [...previousCategories, ...overview, ...previousCategories]; +const optimized = [ + { statement: database.query(expand(sqlForFunction('load_usage_analysis')[0]!)), args: [] }, + ...overview, +]; +const medianMs = (queries: typeof previous) => { + const samples = Array.from({ length: 4 }, () => { + const start = performance.now(); + for (const query of queries) query.statement.all(...query.args); + return performance.now() - start; + }).slice(1).sort((left, right) => left - right); + return Math.round(samples[1]!); +}; + +const insert = database.query('INSERT INTO usage_events(event_key,timestamp,timestamp_ms,local_hour,model,provider,source,api_key_hash,latency_ms,ttft_ms,input_tokens,output_tokens,total_tokens,created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 3000, 1000, 1000, 200, 1200, ?)'); +const timestamp = '2026-09-05T00:00:00Z'; +let inserted = 0; +console.log('Synthetic in-memory SQLite; repository schema/indexes/SQL; all rows match; SQL only, excluding Rust folding, IPC and rendering.'); +for (const size of [10_000, 100_000, 300_000]) { + database.transaction(() => { + while (inserted < size) { + inserted += 1; + insert.run('event-' + inserted, timestamp, inserted, '2026-09-05-' + inserted % 24, + 'model-' + inserted % 20, 'provider-' + inserted % 5, 'source-' + inserted % 100, + 'key-' + inserted % 30, timestamp); + } + })(); + console.log(JSON.stringify({ records: size, + before: { queries: previous.length, medianMs: medianMs(previous) }, + after: { queries: optimized.length, medianMs: medianMs(optimized) }, + })); +} +database.close(); diff --git a/src-tauri/src/agents/commands.rs b/src-tauri/src/agents/commands.rs index 365843d..e9da5f5 100644 --- a/src-tauri/src/agents/commands.rs +++ b/src-tauri/src/agents/commands.rs @@ -3,6 +3,7 @@ use super::*; const AGENT_STATUS_DETECTION_CONCURRENCY: usize = 4; static CODEX_CATALOG_SYNC_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); static CODEX_CATALOG_UPDATE_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +static CODEX_CATALOG_REFRESH: tokio::sync::Notify = tokio::sync::Notify::const_new(); #[derive(Clone, Copy)] enum AgentStatusDetectionTarget { @@ -877,6 +878,10 @@ pub(crate) async fn refresh_applied_codex_model_catalog( sync_prepared_codex_model_catalog(app, config, &prepared) } +pub(crate) fn request_codex_model_catalog_refresh() { + CODEX_CATALOG_REFRESH.notify_one(); +} + pub(crate) fn start_codex_model_catalog_sync(app: tauri::AppHandle) { tauri::async_runtime::spawn(async move { let Ok(home) = app.path().home_dir() else { @@ -885,7 +890,12 @@ pub(crate) fn start_codex_model_catalog_sync(app: tauri::AppHandle) { let mut interval = tokio::time::interval(Duration::from_secs(30)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { - interval.tick().await; + tokio::select! { + _ = interval.tick() => {}, + _ = CODEX_CATALOG_REFRESH.notified() => { + tokio::time::sleep(Duration::from_millis(500)).await; + }, + } let Ok(config) = app.state::().snapshot() else { continue; }; diff --git a/src-tauri/src/app_settings.rs b/src-tauri/src/app_settings.rs index 0ce9e36..5e5589b 100644 --- a/src-tauri/src/app_settings.rs +++ b/src-tauri/src/app_settings.rs @@ -11,12 +11,16 @@ pub(crate) fn detect_core_platform() -> Result { } #[tauri::command] -pub(crate) fn get_core_status( - process_state: tauri::State<'_, CoreProcessState>, - gui_config_state: tauri::State<'_, GuiConfigState>, -) -> Result { - let config = gui_config_state.snapshot()?; - current_core_status(Some(process_state.inner()), Some(config.port)) +pub(crate) async fn get_core_status(app: tauri::AppHandle) -> Result { + tauri::async_runtime::spawn_blocking(move || { + let config = app.state::().snapshot()?; + current_core_status( + Some(app.state::().inner()), + Some(config.port), + ) + }) + .await + .map_err(|error| format!("内核状态后台任务失败: {error}"))? } pub(crate) fn emit_core_status(app: &tauri::AppHandle, status: &CoreStatus) { diff --git a/src-tauri/src/app_update.rs b/src-tauri/src/app_update.rs index 52ebe3d..70ddf99 100644 --- a/src-tauri/src/app_update.rs +++ b/src-tauri/src/app_update.rs @@ -1138,6 +1138,7 @@ pub(crate) async fn download_portable_update_archive_url( let mut file = File::create(destination).map_err(|error| format!("创建应用更新临时文件失败: {error}"))?; let mut downloaded = 0_u64; + let mut progress = crate::progress::ProgressThrottle::default(); while let Some(chunk) = stream.next().await { if token.is_cancelled() { return Err("应用更新下载已取消".to_string()); @@ -1149,16 +1150,18 @@ pub(crate) async fn download_portable_update_archive_url( if downloaded > pending.asset.size_bytes { return Err("应用更新包超过清单声明大小".to_string()); } - update_app_task(app, |task| { - task.downloaded_bytes = downloaded; - task.total_bytes = Some(pending.asset.size_bytes); - task.percent = Some((downloaded as f64 / pending.asset.size_bytes as f64) * 100.0); - task.message = Some(format!( - "{} / {}", - format_byte_count(downloaded), - format_byte_count(pending.asset.size_bytes) - )); - }); + if progress.ready(Instant::now(), downloaded == pending.asset.size_bytes) { + update_app_task(app, |task| { + task.downloaded_bytes = downloaded; + task.total_bytes = Some(pending.asset.size_bytes); + task.percent = Some((downloaded as f64 / pending.asset.size_bytes as f64) * 100.0); + task.message = Some(format!( + "{} / {}", + format_byte_count(downloaded), + format_byte_count(pending.asset.size_bytes) + )); + }); + } } file.flush() .map_err(|error| format!("保存应用更新临时文件失败: {error}"))?; diff --git a/src-tauri/src/configuration_watcher.rs b/src-tauri/src/configuration_watcher.rs index 0462d37..911ca44 100644 --- a/src-tauri/src/configuration_watcher.rs +++ b/src-tauri/src/configuration_watcher.rs @@ -2,7 +2,7 @@ use super::{ agent_managed_paths, apply_gui_managed_settings, consume_software_write, core_config_settings_from_value, core_install_dir, gui_config_path, is_loopback_host, lock_core_config_file, normalized_config_path, path_to_string, - refresh_agent_config_status_cache, refresh_applied_codex_model_catalog, validate_gui_config, + refresh_agent_config_status_cache, request_codex_model_catalog_refresh, validate_gui_config, write_yaml_if_changed, AgentClient, AgentConfigStatusCache, ConfigFilesChangedPayload, CoreConfigSettings, GuiConfigFile, GuiConfigState, CONFIG_FILES_CHANGED_EVENT, CORE_CONFIG_FILE, @@ -11,8 +11,9 @@ use notify::{RecommendedWatcher, RecursiveMode, Watcher}; use std::{ fs, path::{Path, PathBuf}, + sync::{Arc, Mutex}, thread, - time::Duration, + time::{Duration, Instant}, }; use tauri::{Emitter, Manager}; @@ -204,20 +205,7 @@ fn handle_configuration_file_changes( } } - let catalog_app = app.clone(); - tauri::async_runtime::spawn(async move { - tokio::time::sleep(Duration::from_millis(500)).await; - let config = match catalog_app.state::().snapshot() { - Ok(config) => config, - Err(error) => { - eprintln!("读取配置以刷新 Codex 模型目录失败: {error}"); - return; - } - }; - if let Err(error) = refresh_applied_codex_model_catalog(&catalog_app, &config).await { - eprintln!("配置变化后刷新 Codex 模型目录失败: {error}"); - } - }); + request_codex_model_catalog_refresh(); } refresh_agents |= tracked_changes @@ -278,44 +266,107 @@ fn ensure_configuration_watch_directories( Ok(newly_watched) } +#[derive(Default)] +struct PendingConfigurationChanges { + paths: Vec, + error: Option, +} + +fn configuration_watch_path(path: &Path) -> PathBuf { + fs::canonicalize(path).unwrap_or_else(|_| { + nearest_existing_watch_directory(path) + .and_then(|parent| { + path.strip_prefix(&parent) + .ok() + .map(|suffix| normalized_config_path(&parent).join(suffix)) + }) + .unwrap_or_else(|| path.to_path_buf()) + }) +} + +impl PendingConfigurationChanges { + fn insert(&mut self, path: PathBuf) { + self.paths.retain(|existing| existing != &path); + self.paths.push(path); + } + + fn record(&mut self, event: notify::Event, tracked_paths: &[(PathBuf, PathBuf)]) -> bool { + if event.kind.is_access() { + return false; + } + if event.need_rescan() { + for (tracked, _) in tracked_paths { + self.insert(tracked.clone()); + } + return true; + } + let mut relevant = false; + for path in &event.paths { + let normalized = configuration_watch_path(path); + for (tracked, canonical) in tracked_paths { + if tracked.starts_with(path) || canonical.starts_with(&normalized) { + self.insert(tracked.clone()); + relevant = true; + } + } + } + relevant + } +} + pub(crate) fn start_configuration_file_watcher(app: tauri::AppHandle) -> Result<(), String> { let tracked_paths = tracked_configuration_paths(&app)?; - let (sender, receiver) = std::sync::mpsc::channel(); - let mut watcher: RecommendedWatcher = notify::recommended_watcher( - move |event: Result| match event { - Ok(event) if event.kind.is_access() => {} - event => { - let _ = sender.send(event); + let callback_paths: Vec<_> = tracked_paths + .iter() + .map(|path| (path.clone(), configuration_watch_path(path))) + .collect(); + let pending = Arc::new(Mutex::new(PendingConfigurationChanges::default())); + let callback_pending = pending.clone(); + let (sender, receiver) = std::sync::mpsc::sync_channel(1); + let mut watcher: RecommendedWatcher = + notify::recommended_watcher(move |event: Result| { + let Ok(mut pending) = callback_pending.lock() else { + return; + }; + let relevant = match event { + Ok(event) => pending.record(event, &callback_paths), + Err(error) => { + pending.error = Some(error.to_string()); + for (path, _) in &callback_paths { + pending.insert(path.clone()); + } + true + } + }; + if relevant { + let _ = sender.try_send(()); } - }, - ) - .map_err(|error| format!("创建配置文件监控器失败: {error}"))?; + }) + .map_err(|error| format!("创建配置文件监控器失败: {error}"))?; let mut watched_directories = Vec::new(); ensure_configuration_watch_directories(&mut watcher, &tracked_paths, &mut watched_directories)?; thread::spawn(move || { let mut watcher = watcher; loop { - let first = match receiver.recv() { - Ok(event) => event, + if receiver.recv().is_err() { + return; + } + let deadline = Instant::now() + Duration::from_secs(2); + while let Some(remaining) = deadline.checked_duration_since(Instant::now()) { + if receiver + .recv_timeout(remaining.min(Duration::from_millis(500))) + .is_err() + { + break; + } + } + let mut changes = match pending.lock() { + Ok(mut pending) => std::mem::take(&mut *pending), Err(_) => return, }; - let mut paths = Vec::new(); - let mut append_paths = |event_paths: Vec| { - for path in event_paths { - paths.retain(|existing| existing != &path); - paths.push(path); - } - }; - match first { - Ok(event) => append_paths(event.paths), - Err(error) => eprintln!("配置文件监控错误: {error}"), - } - while let Ok(event) = receiver.recv_timeout(Duration::from_millis(500)) { - match event { - Ok(event) => append_paths(event.paths), - Err(error) => eprintln!("配置文件监控错误: {error}"), - } + if let Some(error) = changes.error.take() { + eprintln!("配置文件监控错误: {error}"); } match ensure_configuration_watch_directories( &mut watcher, @@ -330,14 +381,70 @@ pub(crate) fn start_configuration_file_watcher(app: tauri::AppHandle) -> Result< new_directories.iter().any(|directory| directory == parent) }) { - append_paths(vec![tracked_path.clone()]); + changes.insert(tracked_path.clone()); } } } Err(error) => eprintln!("配置目录监控更新失败: {error}"), } - handle_configuration_file_changes(&app, paths, &tracked_paths); + handle_configuration_file_changes(&app, changes.paths, &tracked_paths); } }); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn missing_paths_match_canonical_events_after_the_directory_is_created() { + let root = std::env::temp_dir(); + let relative = Path::new("cpa-nonexistent-watch-unit-test/agent/config.toml"); + let tracked = root.join(relative); + let canonical = fs::canonicalize(&root).unwrap().join(relative); + let mut pending = PendingConfigurationChanges::default(); + let event = notify::Event::new(notify::EventKind::Create(notify::event::CreateKind::File)) + .add_path(canonical); + assert!(pending.record( + event, + &[(tracked.clone(), configuration_watch_path(&tracked))] + )); + assert_eq!(pending.paths, vec![tracked]); + } + + #[test] + fn pending_changes_ignore_noise_and_remain_bounded_in_last_write_order() { + let root = std::env::temp_dir().join("cpa-watcher-unit-test"); + let gui = root.join("gui.toml"); + let core = root.join("core.yaml"); + let tracked = vec![(gui.clone(), gui.clone()), (core.clone(), core.clone())]; + let mut pending = PendingConfigurationChanges::default(); + let event = |path: PathBuf| { + notify::Event::new(notify::EventKind::Modify(notify::event::ModifyKind::Any)) + .add_path(path) + }; + for index in 0..1000 { + assert!(!pending.record(event(root.join(format!("noise-{index}"))), &tracked)); + } + assert!(pending.paths.is_empty()); + for _ in 0..1000 { + pending.record(event(gui.clone()), &tracked); + pending.record(event(core.clone()), &tracked); + } + pending.record(event(gui.clone()), &tracked); + assert_eq!(pending.paths, vec![core, gui]); + } + + #[test] + fn parent_directory_events_keep_new_configuration_directories_discoverable() { + let directory = std::env::temp_dir().join("cpa-watch-missing-parent"); + let path = directory.join("agent/config.toml"); + let mut pending = PendingConfigurationChanges::default(); + let event = + notify::Event::new(notify::EventKind::Create(notify::event::CreateKind::Folder)) + .add_path(directory); + assert!(pending.record(event, &[(path.clone(), path.clone())])); + assert_eq!(pending.paths, vec![path]); + } +} diff --git a/src-tauri/src/core_runtime.rs b/src-tauri/src/core_runtime.rs index 041a222..984c53f 100644 --- a/src-tauri/src/core_runtime.rs +++ b/src-tauri/src/core_runtime.rs @@ -1,5 +1,26 @@ use super::*; +pub(crate) static CORE_OPERATION_LOCK: Mutex<()> = Mutex::new(()); + +async fn run_core_command( + app: tauri::AppHandle, + operation: fn(&CoreProcessState, &GuiConfigState) -> Result, +) -> Result { + tauri::async_runtime::spawn_blocking(move || { + let _guard = CORE_OPERATION_LOCK + .try_lock() + .map_err(|_| "内核正在执行其他操作，请稍后重试".to_string())?; + let status = operation( + app.state::().inner(), + app.state::().inner(), + )?; + emit_core_status(&app, &status); + Ok(status) + }) + .await + .map_err(|error| format!("内核后台任务失败: {error}"))? +} + #[tauri::command] pub(crate) async fn check_latest_core( app: tauri::AppHandle, @@ -38,28 +59,35 @@ pub(crate) fn detect_bundled_core() -> Result, String> { } #[tauri::command] -pub(crate) fn install_bundled_core( +pub(crate) async fn install_bundled_core( app: tauri::AppHandle, window: tauri::Window, - state: tauri::State<'_, CoreDownloadState>, - process_state: tauri::State<'_, CoreProcessState>, - gui_config_state: tauri::State<'_, GuiConfigState>, ) -> Result { - let (info, archive_path) = bundled_core_archive()? - .ok_or_else(|| "当前发行包没有匹配此系统架构的内置内核".to_string())?; - let token = CancellationToken::new(); - state.start(token, Some(info.version.clone()))?; - let result = install_core_with_runtime_restore( - &app, - process_state.inner(), - gui_config_state.inner(), - || install_bundled_core_inner(&window, state.inner(), &info, &archive_path), - ); - if result.is_err() { - let _ = cleanup_core_work_dirs(); - } - state.finish(&window, result.clone()); - result + tauri::async_runtime::spawn_blocking(move || { + let _guard = CORE_OPERATION_LOCK + .try_lock() + .map_err(|_| "内核正在执行其他操作，请稍后重试".to_string())?; + let state = app.state::(); + let process_state = app.state::(); + let gui_config_state = app.state::(); + let (info, archive_path) = bundled_core_archive()? + .ok_or_else(|| "当前发行包没有匹配此系统架构的内置内核".to_string())?; + let token = CancellationToken::new(); + state.start(token, Some(info.version.clone()))?; + let result = install_core_with_runtime_restore( + &app, + process_state.inner(), + gui_config_state.inner(), + || install_bundled_core_inner(&window, state.inner(), &info, &archive_path), + ); + if result.is_err() { + let _ = cleanup_core_work_dirs(); + } + state.finish(&window, result.clone()); + result + }) + .await + .map_err(|error| format!("离线内核安装后台任务失败: {error}"))? } pub(crate) fn core_needs_bundled_bootstrap(install_dir: &Path) -> bool { @@ -103,47 +131,53 @@ pub(crate) fn get_core_install_task(state: tauri::State<'_, CoreDownloadState>) pub(crate) async fn install_core_version( app: tauri::AppHandle, window: tauri::Window, - state: tauri::State<'_, CoreDownloadState>, - process_state: tauri::State<'_, CoreProcessState>, - gui_config_state: tauri::State<'_, GuiConfigState>, version: Option, ) -> Result { - let config = gui_config_state.snapshot()?; - let proxy_url = config.proxy_url.clone(); - let token = CancellationToken::new(); - state.start(token.clone(), version.clone())?; - let (was_running, install_result) = - match pause_core_for_install(&app, process_state.inner(), &config) { - Ok(was_running) => ( - was_running, - install_core_version_inner( - &app, - &window, - state.inner(), - gui_config_state.inner(), - token, - version, - &proxy_url, - config.selected_download_candidate(), - config.custom_download_mirrors.clone(), - ) - .await, - ), - Err(error) => (false, Err(error)), - }; - let result = restore_core_after_install( - &app, - process_state.inner(), - &config, - was_running, - install_result, - ); - if result.is_err() { - let _ = cleanup_core_work_dirs(); - } - state.finish(&window, result.clone()); + tauri::async_runtime::spawn_blocking(move || { + let _guard = CORE_OPERATION_LOCK + .try_lock() + .map_err(|_| "内核正在执行其他操作，请稍后重试".to_string())?; + let state = app.state::(); + let process_state = app.state::(); + let gui_config_state = app.state::(); + let config = gui_config_state.snapshot()?; + let proxy_url = config.proxy_url.clone(); + let token = CancellationToken::new(); + state.start(token.clone(), version.clone())?; + let (was_running, install_result) = + match pause_core_for_install(&app, process_state.inner(), &config) { + Ok(was_running) => ( + was_running, + tauri::async_runtime::block_on(install_core_version_inner( + &app, + &window, + state.inner(), + gui_config_state.inner(), + token, + version, + &proxy_url, + config.selected_download_candidate(), + config.custom_download_mirrors.clone(), + )), + ), + Err(error) => (false, Err(error)), + }; + let result = restore_core_after_install( + &app, + process_state.inner(), + &config, + was_running, + install_result, + ); + if result.is_err() { + let _ = cleanup_core_work_dirs(); + } + state.finish(&window, result.clone()); - result + result + }) + .await + .map_err(|error| format!("内核安装后台任务失败: {error}"))? } fn install_core_with_runtime_restore( @@ -213,14 +247,8 @@ fn emit_current_core_status(app: &tauri::AppHandle, process_state: &CoreProcessS } #[tauri::command] -pub(crate) fn start_core_process( - app: tauri::AppHandle, - process_state: tauri::State<'_, CoreProcessState>, - gui_config_state: tauri::State<'_, GuiConfigState>, -) -> Result { - let status = start_core_process_with_state(process_state.inner(), gui_config_state.inner())?; - emit_core_status(&app, &status); - Ok(status) +pub(crate) async fn start_core_process(app: tauri::AppHandle) -> Result { + run_core_command(app, start_core_process_with_state).await } pub(crate) fn start_core_process_with_state( @@ -237,14 +265,8 @@ pub(crate) fn start_core_process_with_state( } #[tauri::command] -pub(crate) fn stop_core_process( - app: tauri::AppHandle, - process_state: tauri::State<'_, CoreProcessState>, - gui_config_state: tauri::State<'_, GuiConfigState>, -) -> Result { - let status = stop_core_process_with_state(process_state.inner(), gui_config_state.inner())?; - emit_core_status(&app, &status); - Ok(status) +pub(crate) async fn stop_core_process(app: tauri::AppHandle) -> Result { + run_core_command(app, stop_core_process_with_state).await } pub(crate) fn stop_core_process_with_state( @@ -257,14 +279,8 @@ pub(crate) fn stop_core_process_with_state( } #[tauri::command] -pub(crate) fn restart_core_process( - app: tauri::AppHandle, - process_state: tauri::State<'_, CoreProcessState>, - gui_config_state: tauri::State<'_, GuiConfigState>, -) -> Result { - let status = restart_core_process_with_state(process_state.inner(), gui_config_state.inner())?; - emit_core_status(&app, &status); - Ok(status) +pub(crate) async fn restart_core_process(app: tauri::AppHandle) -> Result { + run_core_command(app, restart_core_process_with_state).await } pub(crate) fn restart_core_process_with_state( @@ -995,6 +1011,7 @@ pub(crate) async fn download_asset_inner( File::create(archive_path).map_err(|err| format!("创建内核压缩包失败: {err}"))?; let mut downloaded = 0_u64; let mut hasher = Sha256::new(); + let mut progress = crate::progress::ProgressThrottle::default(); while let Some(chunk) = tokio::select! { chunk = stream.next() => chunk, @@ -1007,9 +1024,12 @@ pub(crate) async fn download_asset_inner( .map_err(|err| format!("保存下载数据失败: {err}"))?; hasher.update(&chunk); downloaded += chunk.len() as u64; - state.progress(window, "下载中", downloaded, total, true); + if progress.ready(Instant::now(), total == Some(downloaded)) { + state.progress(window, "下载中", downloaded, total, true); + } } + state.progress(window, "下载中", downloaded, total, true); file.flush() .map_err(|err| format!("刷新内核压缩包失败: {err}"))?; ensure_not_cancelled(token, Some(archive_path))?; diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index 6ce09da..b8e5c85 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -12,6 +12,7 @@ mod core_runtime; mod instance_lock; mod management_api; mod oauth_browser; +mod progress; mod provider_health; #[cfg(any(target_os = "macos", target_os = "windows"))] mod tray; @@ -2406,6 +2407,9 @@ fn main() { let core_app = app.handle().clone(); tauri::async_runtime::spawn_blocking(move || { + let Ok(_guard) = CORE_OPERATION_LOCK.lock() else { + return; + }; let gui_config_state = core_app.state::(); let process_state = core_app.state::(); let Ok(config) = gui_config_state.snapshot() else { diff --git a/src-tauri/src/management_api.rs b/src-tauri/src/management_api.rs index 1636998..94a4833 100644 --- a/src-tauri/src/management_api.rs +++ b/src-tauri/src/management_api.rs @@ -12,6 +12,7 @@ use std::{ fs, path::{Path, PathBuf}, process::{Command, Stdio}, + sync::LazyLock, time::Duration, }; @@ -299,18 +300,21 @@ pub(crate) async fn submit_oauth_callback( } pub(crate) fn management_http_client() -> Result { - reqwest::Client::builder() - // GUI-to-Core management traffic must always connect directly. Upstream - // traffic still uses Core's proxy-url, and other GUI HTTP clients keep - // their independently configured proxy behavior. - .no_proxy() - .connect_timeout(Duration::from_secs(10)) - .timeout(Duration::from_secs(30)) - // Management requests target the configured local listener. This keeps - // self-signed/private-CA certificates usable without weakening upstream clients. - .danger_accept_invalid_certs(true) - .build() - .map_err(|err| format_management_request_error("创建管理 API 客户端失败", &err)) + static CLIENT: LazyLock> = LazyLock::new(|| { + reqwest::Client::builder() + // GUI-to-Core management traffic must always connect directly. Upstream + // traffic still uses Core's proxy-url, and other GUI HTTP clients keep + // their independently configured proxy behavior. + .no_proxy() + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(30)) + // Management requests target the configured local listener. This keeps + // self-signed/private-CA certificates usable without weakening upstream clients. + .danger_accept_invalid_certs(true) + .build() + .map_err(|err| format_management_request_error("创建管理 API 客户端失败", &err)) + }); + CLIENT.as_ref().cloned().map_err(Clone::clone) } pub(crate) fn format_management_request_error( diff --git a/src-tauri/src/progress.rs b/src-tauri/src/progress.rs new file mode 100644 index 0000000..9213437 --- /dev/null +++ b/src-tauri/src/progress.rs @@ -0,0 +1,36 @@ +use std::time::{Duration, Instant}; + +#[derive(Default)] +pub(crate) struct ProgressThrottle { + last: Option, +} + +impl ProgressThrottle { + pub(crate) fn ready(&mut self, now: Instant, finished: bool) -> bool { + if finished + || self + .last + .is_none_or(|last| now.duration_since(last) >= Duration::from_millis(100)) + { + self.last = Some(now); + true + } else { + false + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn throttles_chunks_but_keeps_first_and_final_progress() { + let mut throttle = ProgressThrottle::default(); + let now = Instant::now(); + assert!(throttle.ready(now, false)); + assert!(!throttle.ready(now + Duration::from_millis(10), false)); + assert!(throttle.ready(now + Duration::from_millis(100), false)); + assert!(throttle.ready(now + Duration::from_millis(101), true)); + } +} diff --git a/src-tauri/src/provider_health.rs b/src-tauri/src/provider_health.rs index c8d4561..d7666e1 100644 --- a/src-tauri/src/provider_health.rs +++ b/src-tauri/src/provider_health.rs @@ -10,6 +10,7 @@ use std::{ }; const MAX_PROVIDER_HEALTH_STREAM_BYTES: usize = 256 * 1024; +static PROVIDER_HEALTH_SLOTS: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(4); #[derive(Deserialize)] #[serde(rename_all = "camelCase")] @@ -304,6 +305,10 @@ pub(crate) async fn provider_health_probe( return Err("健康检测请求体过大".to_string()); } + let _permit = PROVIDER_HEALTH_SLOTS + .acquire() + .await + .map_err(|error| error.to_string())?; let timeout = Duration::from_millis(request.timeout_ms.unwrap_or(15_000).clamp(1_000, 120_000)); let config = gui_config_state.snapshot()?; let proxy_url = url diff --git a/src-tauri/src/usage.rs b/src-tauri/src/usage.rs index 97a0e91..f1421e2 100644 --- a/src-tauri/src/usage.rs +++ b/src-tauri/src/usage.rs @@ -21,7 +21,7 @@ use std::{ collections::{HashMap, HashSet}, fs, path::{Path, PathBuf}, - sync::Mutex, + sync::{LazyLock, Mutex}, time::{Duration, SystemTime, UNIX_EPOCH}, }; use tauri::{Emitter, Manager}; @@ -53,6 +53,25 @@ const BUNDLED_MODEL_PRICE_CATALOG: &str = include_str!("../resources/model_price const MODEL_PRICE_SYNC_URL: &str = "https://raw.githubusercontent.com/router-for-me/EasyCLIProxyAPI/main/src-tauri/resources/model_prices.json"; +static USAGE_QUERY_SLOTS: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(2); + +async fn run_usage_task(task: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Result + Send + 'static, +{ + let permit = USAGE_QUERY_SLOTS + .acquire() + .await + .map_err(|error| error.to_string())?; + tauri::async_runtime::spawn_blocking(move || { + let _permit = permit; + task() + }) + .await + .map_err(|error| format!("使用记录后台任务失败: {error}"))? +} + pub(crate) struct UsageCollectorState { inner: Mutex, } @@ -545,9 +564,8 @@ fn initialize_usage_storage_at(root: &Path) -> Result<(), String> { } #[tauri::command] -pub(crate) fn repair_usage_cache_records() -> Result { - let root = usage_root_dir()?; - repair_usage_cache_records_at(&root) +pub(crate) async fn repair_usage_cache_records() -> Result { + run_usage_task(|| repair_usage_cache_records_at(&usage_root_dir()?)).await } fn repair_usage_cache_records_at(root: &Path) -> Result { @@ -1380,8 +1398,8 @@ pub(crate) fn start_usage_collector(app: tauri::AppHandle) { let Some(token) = state.start() else { return; }; - tauri::async_runtime::spawn(async move { - usage_collector_loop(app, token).await; + tauri::async_runtime::spawn_blocking(move || { + tauri::async_runtime::block_on(usage_collector_loop(app, token)); }); } @@ -1402,6 +1420,9 @@ async fn usage_collector_loop(app: tauri::AppHandle, token: CancellationToken) { let mut subscription: Option = None; let mut subscribe_retry_at = tokio::time::Instant::now(); let mut next_inbox_cleanup_at = tokio::time::Instant::now() + Duration::from_secs(60 * 60); + let mut next_inbox_recovery_at = tokio::time::Instant::now(); + let mut next_core_check_at = tokio::time::Instant::now(); + let mut core_running = false; loop { if token.is_cancelled() { return; @@ -1423,35 +1444,42 @@ async fn usage_collector_loop(app: tauri::AppHandle, token: CancellationToken) { } next_inbox_cleanup_at = tokio::time::Instant::now() + Duration::from_secs(60 * 60); } - let recovered = open_usage_database_at(&root) - .and_then(|mut connection| process_usage_inbox(&mut connection, &config)); - match recovered { - Ok(saved) if saved > 0 => { - let collected_at = Local::now().to_rfc3339(); - app.state::() - .increment_total_records(saved); - set_collector_status( - &app, - "collecting", - &format!("已恢复 {saved} 条待处理记录"), - Some(collected_at.clone()), - ); - let _ = app.emit(USAGE_UPDATED_EVENT, collected_at); - retry_seconds = 1; - continue; - } - Ok(_) => {} - Err(error) => { - set_collector_error(&app, error); - wait_or_cancel(&token, retry_seconds).await; - retry_seconds = (retry_seconds * 2).min(10); - continue; + if tokio::time::Instant::now() >= next_inbox_recovery_at { + let recovered = open_usage_database_at(&root) + .and_then(|mut connection| process_usage_inbox(&mut connection, &config)); + match recovered { + Ok(saved) if saved > 0 => { + let collected_at = Local::now().to_rfc3339(); + app.state::() + .increment_total_records(saved); + set_collector_status( + &app, + "collecting", + &format!("已恢复 {saved} 条待处理记录"), + Some(collected_at.clone()), + ); + let _ = app.emit(USAGE_UPDATED_EVENT, collected_at); + retry_seconds = 1; + continue; + } + Ok(_) => { + next_inbox_recovery_at = tokio::time::Instant::now() + Duration::from_secs(2); + } + Err(error) => { + set_collector_error(&app, error); + wait_or_cancel(&token, retry_seconds).await; + retry_seconds = (retry_seconds * 2).min(10); + continue; + } } } - let process_state = app.state::(); - let core_running = current_core_status(Some(process_state.inner()), Some(config.port)) - .map(|status| status.running) - .unwrap_or(false); + if tokio::time::Instant::now() >= next_core_check_at { + let process_state = app.state::(); + core_running = current_core_status(Some(process_state.inner()), Some(config.port)) + .map(|status| status.running) + .unwrap_or(false); + next_core_check_at = tokio::time::Instant::now() + Duration::from_secs(2); + } if !core_running { subscription = None; set_collector_status(&app, "waiting-core", "等待内核启动", None); @@ -2382,9 +2410,8 @@ pub(crate) fn get_usage_collector_status( } #[tauri::command] -pub(crate) fn get_usage_overview(query: UsageQuery) -> Result { - let connection = open_usage_database()?; - load_usage_overview(&connection, &query) +pub(crate) async fn get_usage_overview(query: UsageQuery) -> Result { + run_usage_task(move || load_usage_overview(&open_usage_database()?, &query)).await } fn load_usage_overview( @@ -2697,12 +2724,17 @@ fn cost_for_token_segment( } fn official_model_price(model: &str) -> Option { - let prices = bundled_model_prices().ok()?; - find_model_price(&prices, model).cloned() + find_model_price(cached_bundled_model_prices().ok()?, model).cloned() } fn bundled_model_prices() -> Result, String> { - parse_model_price_catalog(BUNDLED_MODEL_PRICE_CATALOG, "builtin", 0) + cached_bundled_model_prices().cloned() +} + +fn cached_bundled_model_prices() -> Result<&'static HashMap, String> { + static PRICES: LazyLock, String>> = + LazyLock::new(|| parse_model_price_catalog(BUNDLED_MODEL_PRICE_CATALOG, "builtin", 0)); + PRICES.as_ref().map_err(Clone::clone) } fn parse_model_price_catalog( @@ -2966,9 +2998,8 @@ fn upsert_model_price(connection: &Connection, price: &ModelPrice) -> Result<(), } #[tauri::command] -pub(crate) fn get_usage_pricing(query: UsageQuery) -> Result { - let connection = open_usage_database()?; - load_usage_pricing(&connection, &query) +pub(crate) async fn get_usage_pricing(query: UsageQuery) -> Result { + run_usage_task(move || load_usage_pricing(&open_usage_database()?, &query)).await } fn load_usage_pricing(connection: &Connection, query: &UsageQuery) -> Result { @@ -3043,25 +3074,27 @@ fn load_usage_pricing(connection: &Connection, query: &UsageQuery) -> Result Result<(), String> { - let connection = open_usage_database()?; +pub(crate) async fn save_usage_model_price(mut price: ModelPrice) -> Result<(), String> { price.model = price.model.trim().to_string(); price.source = "manual".to_string(); price.source_model_id.clear(); price.updated_at_ms = Local::now().timestamp_millis(); - upsert_model_price(&connection, &price) + run_usage_task(move || upsert_model_price(&open_usage_database()?, &price)).await } #[tauri::command] -pub(crate) fn delete_usage_model_price(model: String) -> Result<(), String> { - let connection = open_usage_database()?; - connection - .execute( - "DELETE FROM model_prices WHERE model = ?1 COLLATE NOCASE", - params![model.trim()], - ) - .map_err(|error| format!("删除模型价格失败: {error}"))?; - Ok(()) +pub(crate) async fn delete_usage_model_price(model: String) -> Result<(), String> { + run_usage_task(move || { + let connection = open_usage_database()?; + connection + .execute( + "DELETE FROM model_prices WHERE model = ?1 COLLATE NOCASE", + params![model.trim()], + ) + .map_err(|error| format!("删除模型价格失败: {error}"))?; + Ok(()) + }) + .await } #[tauri::command] @@ -3097,53 +3130,56 @@ pub(crate) async fn sync_usage_model_prices( None => (bundled_model_prices()?, true), }; - let mut connection = open_usage_database()?; - let current_prices = load_model_prices(&connection)?; - let manual_models = current_prices - .values() - .filter(|price| price.source == "manual") - .map(|price| price.model.to_ascii_lowercase()) - .collect::>(); - let mut result = ModelPriceSyncResult { - used_builtin, - ..ModelPriceSyncResult::default() - }; - if !used_builtin { - let transaction = connection - .transaction() - .map_err(|error| format!("开始更新模型价格失败: {error}"))?; - transaction - .execute("DELETE FROM model_prices WHERE source = 'github'", []) - .map_err(|error| format!("清理旧模型价格失败: {error}"))?; - for price in remote_prices.values() { - if manual_models.contains(&price.model.to_ascii_lowercase()) { - result.skipped += 1; - continue; + run_usage_task(move || { + let mut connection = open_usage_database()?; + let current_prices = load_model_prices(&connection)?; + let manual_models = current_prices + .values() + .filter(|price| price.source == "manual") + .map(|price| price.model.to_ascii_lowercase()) + .collect::>(); + let mut result = ModelPriceSyncResult { + used_builtin, + ..ModelPriceSyncResult::default() + }; + if !used_builtin { + let transaction = connection + .transaction() + .map_err(|error| format!("开始更新模型价格失败: {error}"))?; + transaction + .execute("DELETE FROM model_prices WHERE source = 'github'", []) + .map_err(|error| format!("清理旧模型价格失败: {error}"))?; + for price in remote_prices.values() { + if manual_models.contains(&price.model.to_ascii_lowercase()) { + result.skipped += 1; + continue; + } + upsert_model_price(&transaction, price)?; + result.imported += 1; } - upsert_model_price(&transaction, price)?; - result.imported += 1; + transaction + .commit() + .map_err(|error| format!("提交模型价格更新失败: {error}"))?; + } else { + connection + .execute("DELETE FROM model_prices WHERE source = 'github'", []) + .map_err(|error| format!("恢复软件内置模型价格失败: {error}"))?; } - transaction - .commit() - .map_err(|error| format!("提交模型价格更新失败: {error}"))?; - } else { - connection - .execute("DELETE FROM model_prices WHERE source = 'github'", []) - .map_err(|error| format!("恢复软件内置模型价格失败: {error}"))?; - } - let filter = build_usage_filter(&query); - let models = load_usage_cost_groups(&connection, &filter)? - .into_iter() - .map(|group| group.model) - .collect::>(); - let effective_prices = load_model_prices(&connection)?; - for model in models { - if resolve_model_price(&model, "", &effective_prices).is_none() { - result.unmatched.push(model); + let filter = build_usage_filter(&query); + let models = load_usage_cost_groups(&connection, &filter)? + .into_iter() + .map(|group| group.model) + .collect::>(); + let effective_prices = load_model_prices(&connection)?; + for model in models { + if resolve_model_price(&model, "", &effective_prices).is_none() { + result.unmatched.push(model); + } } - } - Ok(result) + Ok(result) + }) + .await } fn canonical_model_tail(value: &str) -> String { @@ -3163,13 +3199,12 @@ fn normalized_model_tail(value: &str) -> String { } #[tauri::command] -pub(crate) fn get_usage_analysis( +pub(crate) async fn get_usage_analysis( query: UsageQuery, gui_config_state: tauri::State<'_, GuiConfigState>, ) -> Result { - let connection = open_usage_database()?; let config = gui_config_state.snapshot()?; - load_usage_analysis(&connection, &query, &config) + run_usage_task(move || load_usage_analysis(&open_usage_database()?, &query, &config)).await } fn load_usage_analysis( @@ -3177,14 +3212,99 @@ fn load_usage_analysis( query: &UsageQuery, config: &GuiConfigFile, ) -> Result { + let filter = build_usage_filter(query); + let sql = format!( + r#" + SELECT + COALESCE(NULLIF(TRIM(model), ''), 'unknown'), + COALESCE(NULLIF(TRIM(provider), ''), '未知 Provider'), + COALESCE(NULLIF(TRIM(source), ''), '未知来源'), + COALESCE(NULLIF(TRIM(api_key_hash), ''), '未记录密钥'), + MAX(TRIM(api_key_remark)), MAX(TRIM(api_key_display)), + COUNT(*), + COALESCE(SUM(CASE WHEN failed != 0 AND canceled = 0 THEN 1 ELSE 0 END), 0), + COALESCE(SUM(total_tokens), 0) + FROM usage_events{} + GROUP BY 1, 2, 3, 4 + "#, + filter.clause + ); + let mut statement = connection + .prepare(&sql) + .map_err(|error| format!("准备使用分析查询失败: {error}"))?; + let mut rows = statement + .query(params_from_iter(filter.params.iter())) + .map_err(|error| format!("查询使用分析失败: {error}"))?; + let mut categories: [HashMap; 4] = Default::default(); + let mut key_labels = HashMap::::new(); + let read_result = (|| -> rusqlite::Result<()> { + while let Some(row) = rows.next()? { + let requests = from_sql_i64(row.get(6)?); + let failures = from_sql_i64(row.get(7)?); + let tokens = from_sql_i64(row.get(8)?); + for (index, group) in categories.iter_mut().enumerate() { + let key: String = row.get(index)?; + if index == 3 { + let remark: String = row.get(4)?; + let display: String = row.get(5)?; + let labels = key_labels.entry(key.clone()).or_default(); + if remark > labels.0 { + labels.0 = remark; + } + if display > labels.1 { + labels.1 = display; + } + } + let category = group.entry(key.clone()).or_insert_with(|| UsageCategory { + label: key.clone(), + key, + ..UsageCategory::default() + }); + category.requests = category.requests.saturating_add(requests); + category.failures = category.failures.saturating_add(failures); + category.tokens = category.tokens.saturating_add(tokens); + } + } + Ok(()) + })(); + read_result.map_err(|error| format!("读取使用分析失败: {error}"))?; + let [models, providers, sources, api_keys] = categories.map(sorted_usage_categories); + let sources = sources + .into_iter() + .map(|mut category| { + category.label = usage_source_display(config, "", &category.key); + category + }) + .collect(); + let api_keys = api_keys + .into_iter() + .map(|mut category| { + let (remark, display) = key_labels.remove(&category.key).unwrap_or_default(); + category.label = api_key_category_label(remark, display); + category + }) + .collect(); Ok(UsageAnalysis { - models: load_simple_categories(connection, query, "model", "unknown")?, - providers: load_simple_categories(connection, query, "provider", "未知 Provider")?, - sources: load_source_categories(connection, query, config)?, - api_keys: load_api_key_categories(connection, query)?, + models, + providers, + sources, + api_keys, }) } +fn sorted_usage_categories(categories: HashMap) -> Vec { + let mut categories: Vec<_> = categories.into_values().collect(); + categories.sort_by(|left, right| { + right + .tokens + .cmp(&left.tokens) + .then_with(|| right.requests.cmp(&left.requests)) + .then_with(|| left.key.cmp(&right.key)) + }); + categories +} + +#[cfg(test)] fn load_source_categories( connection: &Connection, query: &UsageQuery, @@ -3197,6 +3317,7 @@ fn load_source_categories( Ok(categories) } +#[cfg(test)] fn load_simple_categories( connection: &Connection, query: &UsageQuery, @@ -3240,6 +3361,7 @@ fn load_simple_categories( Ok(categories) } +#[cfg(test)] fn load_api_key_categories( connection: &Connection, query: &UsageQuery, @@ -3284,13 +3406,12 @@ fn load_api_key_categories( } #[tauri::command] -pub(crate) fn get_usage_events( +pub(crate) async fn get_usage_events( query: UsageQuery, gui_config_state: tauri::State<'_, GuiConfigState>, ) -> Result { - let connection = open_usage_database()?; let config = gui_config_state.snapshot()?; - load_usage_events(&connection, &query, &config) + run_usage_task(move || load_usage_events(&open_usage_database()?, &query, &config)).await } fn load_usage_events( @@ -3559,6 +3680,126 @@ fn usage_source_display(config: &GuiConfigFile, provider: &str, source: &str) -> mod tests { use super::*; + #[tokio::test] + async fn background_usage_jobs_limit_database_concurrency() { + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let mut jobs = Vec::new(); + for index in 0..6 { + let active = active.clone(); + let peak = peak.clone(); + jobs.push(tokio::spawn(run_usage_task(move || { + let count = active.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(count, Ordering::SeqCst); + std::thread::sleep(Duration::from_millis(10)); + active.fetch_sub(1, Ordering::SeqCst); + Ok(index) + }))); + } + for (index, job) in jobs.into_iter().enumerate() { + assert_eq!(job.await.unwrap().unwrap(), index); + } + assert!(peak.load(Ordering::SeqCst) <= 2); + assert_eq!(active.load(Ordering::SeqCst), 0); + } + + #[test] + fn single_scan_analysis_matches_independent_categories_for_mixed_data_and_filters() { + let mut connection = Connection::open_in_memory().unwrap(); + initialize_usage_schema(&connection).unwrap(); + let records: Vec<_> = (0..600) + .map(|index| { + let mut record = sample_record( + &format!("analysis-{index}"), + if index % 2 == 0 { + "2026-08-27T00:00:00Z" + } else { + "2026-08-28T00:00:00Z" + }, + [" Model-A ", "model-a", "", "unknown", "Model-B"][index % 5], + ); + record.provider = ["openai", "claude", " "][index % 3].to_string(); + record.source = + ["source-a", " ", "source-b", "sk-secret-source"][index % 4].to_string(); + record.api_key_hash = ["hash-a", "hash-b", " "][index % 3].to_string(); + record.api_key_remark = ["", " alpha ", "zeta", "测试备注"][index % 4].to_string(); + record.api_key_display = ["", "ab••••", "xy••••"][index % 3].to_string(); + record.failed = index % 3 == 0; + record.canceled = index % 6 == 0; + record.tokens.total_tokens = (index % 17) as u64; + record + }) + .collect(); + insert_usage_records(&mut connection, &records).unwrap(); + let config = GuiConfigFile::default(); + let queries = [ + UsageQuery::default(), + UsageQuery { + model: Some("MODEL-A".into()), + ..UsageQuery::default() + }, + UsageQuery { + provider: Some("claude".into()), + ..UsageQuery::default() + }, + UsageQuery { + source: Some("source-a".into()), + ..UsageQuery::default() + }, + UsageQuery { + api_key_hash: Some("hash-a".into()), + ..UsageQuery::default() + }, + UsageQuery { + failed: Some(true), + canceled: Some(false), + ..UsageQuery::default() + }, + UsageQuery { + canceled: Some(true), + ..UsageQuery::default() + }, + UsageQuery { + start: Some("2026-08-28T00:00:00Z".into()), + ..UsageQuery::default() + }, + UsageQuery { + model: Some("not-present".into()), + ..UsageQuery::default() + }, + ]; + let canonical = |mut categories: Vec| { + categories.sort_by(|left, right| left.key.cmp(&right.key)); + serde_json::to_value(categories).unwrap() + }; + for query in queries { + let actual = load_usage_analysis(&connection, &query, &config).unwrap(); + assert_eq!( + canonical(actual.models), + canonical(load_simple_categories(&connection, &query, "model", "unknown").unwrap()) + ); + assert_eq!( + canonical(actual.providers), + canonical( + load_simple_categories(&connection, &query, "provider", "未知 Provider") + .unwrap() + ) + ); + assert_eq!( + canonical(actual.sources), + canonical(load_source_categories(&connection, &query, &config).unwrap()) + ); + assert_eq!( + canonical(actual.api_keys), + canonical(load_api_key_categories(&connection, &query).unwrap()) + ); + } + } + #[test] fn collector_total_records_are_cached_and_incremented() { let state = UsageCollectorState::default(); diff --git a/src/coreRuntime.tsx b/src/coreRuntime.tsx index 7d9b3f0..5f54c33 100644 --- a/src/coreRuntime.tsx +++ b/src/coreRuntime.tsx @@ -1,4 +1,4 @@ -import { createContext, useCallback, useContext, useEffect, useMemo, useState, type ReactNode } from 'react'; +import { createContext, useCallback, useContext, useEffect, useMemo, useRef, useState, type ReactNode } from 'react'; import { invoke } from '@tauri-apps/api/core'; import { listen } from '@tauri-apps/api/event'; @@ -26,24 +26,32 @@ const CoreRuntimeContext = createContext(null); export function CoreRuntimeProvider({ children }: { children: ReactNode }) { const [status, setStatus] = useState(null); const [statusError, setStatusError] = useState(''); + const refreshRequest = useRef | null>(null); + const statusRevision = useRef(0); const publishStatus = useCallback((nextStatus: CoreStatus | null) => { - setStatus(nextStatus); + ++statusRevision.current; + setStatus((previous) => JSON.stringify(previous) === JSON.stringify(nextStatus) ? previous : nextStatus); if (nextStatus) { setStatusError(''); } }, []); - const refreshStatus = useCallback(async () => { - try { - const nextStatus = await invoke('get_core_status'); - setStatus(nextStatus); - setStatusError(''); - } catch (error) { - setStatus(null); - setStatusError(String(error)); - } - }, []); + const refreshStatus = useCallback(() => { + if (refreshRequest.current) return refreshRequest.current; + const revision = statusRevision.current; + refreshRequest.current = invoke('get_core_status') + .then((nextStatus) => { + if (revision === statusRevision.current) publishStatus(nextStatus); + }) + .catch((error) => { + if (revision !== statusRevision.current) return; + setStatus(null); + setStatusError(String(error)); + }) + .finally(() => { refreshRequest.current = null; }); + return refreshRequest.current; + }, [publishStatus]); useEffect(() => { let disposed = false; diff --git a/src/pages/ApiAccessPage.tsx b/src/pages/ApiAccessPage.tsx index 6bd114b..6d75fc0 100644 --- a/src/pages/ApiAccessPage.tsx +++ b/src/pages/ApiAccessPage.tsx @@ -1338,6 +1338,13 @@ function ProviderHealthDialog({ row, onClose }: ProviderHealthDialogProps) { const [search, setSearch] = useState(''); const [results, setResults] = useState>({}); const [checkingAll, setCheckingAll] = useState(false); + const healthControllerRef = useRef(null); + + useEffect(() => { + const controller = new AbortController(); + healthControllerRef.current = controller; + return () => controller.abort(); + }, []); const healthOptions = useMemo(() => ({ provider: providerModelType(row.section), @@ -1393,22 +1400,26 @@ function ProviderHealthDialog({ row, onClose }: ProviderHealthDialogProps) { }; const checkOneModel = async (model: ModelOption) => { + const signal = healthControllerRef.current?.signal; + if (!signal || signal.aborted) return; const key = model.name.toLowerCase(); setResults((current) => ({ ...current, [key]: { status: 'checking' } })); - saveResult(await checkProviderModelHealth(healthOptions, model.name)); + const result = await checkProviderModelHealth(healthOptions, model.name); + if (!signal.aborted) saveResult(result); }; const checkAllModels = async () => { - if (models.length === 0) return; + const signal = healthControllerRef.current?.signal; + if (models.length === 0 || checkingAll || !signal || signal.aborted) return; setCheckingAll(true); setResults(Object.fromEntries(models.map((model) => [ model.name.toLowerCase(), { status: 'checking' } satisfies ProviderModelHealthState, ]))); try { - await checkProviderModelsHealth(healthOptions, models, saveResult); + await checkProviderModelsHealth(healthOptions, models, saveResult, undefined, signal); } finally { - setCheckingAll(false); + if (!signal.aborted) setCheckingAll(false); } }; diff --git a/src/pages/EasyModePage.tsx b/src/pages/EasyModePage.tsx index d3f9262..55d9f84 100644 --- a/src/pages/EasyModePage.tsx +++ b/src/pages/EasyModePage.tsx @@ -125,6 +125,7 @@ export function EasyModePage({ message: string; } | null>(null); const oauthPollTimer = useRef(null); + const oauthGeneration = useRef(0); const [selectedApiSection, setSelectedApiSection] = useState("openai-compatibility"); const [apiBaseUrl, setApiBaseUrl] = useState(""); @@ -200,7 +201,8 @@ export function EasyModePage({ useEffect(() => { void refreshSourceStatus(); return () => { - if (oauthPollTimer.current) window.clearInterval(oauthPollTimer.current); + ++oauthGeneration.current; + if (oauthPollTimer.current !== null) window.clearTimeout(oauthPollTimer.current); }; }, [refreshSourceStatus]); @@ -248,7 +250,9 @@ export function EasyModePage({ setGuideOAuthProvider(provider); setGuideOAuthCompleted(false); } - if (oauthPollTimer.current) window.clearInterval(oauthPollTimer.current); + const generation = ++oauthGeneration.current; + if (oauthPollTimer.current !== null) window.clearTimeout(oauthPollTimer.current); + oauthPollTimer.current = null; setOauthLoggingIn(provider); setOauthNotice(null); @@ -263,6 +267,8 @@ export function EasyModePage({ browser: "default", }); + if (generation !== oauthGeneration.current) return; + if (!result.state) { setOauthNotice({ tone: "error", @@ -273,15 +279,24 @@ export function EasyModePage({ } const stateKey = result.state; - oauthPollTimer.current = window.setInterval(async () => { + const deadline = Date.now() + 10 * 60_000; + let failures = 0; + const poll = async () => { + if (generation !== oauthGeneration.current) return; + if (Date.now() >= deadline) { + setOauthLoggingIn(null); + setOauthNotice({ tone: "error", message: "授权等待超时，请重新发起登录" }); + return; + } try { const pollRes = await invoke<{ status: string; error?: string }>( "get_oauth_status", { state: stateKey }, ); + if (generation !== oauthGeneration.current) return; + failures = 0; const status = (pollRes.status || "").toLowerCase(); if (status === "ok") { - if (oauthPollTimer.current) window.clearInterval(oauthPollTimer.current); setOauthLoggingIn(null); setOauthNotice({ tone: "success", @@ -290,16 +305,29 @@ export function EasyModePage({ setGuideOAuthCompleted(true); void refreshSourceStatus(); } else if (status === "error") { - if (oauthPollTimer.current) window.clearInterval(oauthPollTimer.current); setOauthLoggingIn(null); setOauthNotice({ tone: "error", message: pollRes.error ? `授权失败: ${pollRes.error}` : "授权失败，请重试", }); } - } catch {} - }, 1500); + if (status === "ok" || status === "error") return; + } catch (error) { + if (generation !== oauthGeneration.current) return; + failures += 1; + if (failures >= 3) { + setOauthLoggingIn(null); + setOauthNotice({ tone: "error", message: String(error) }); + return; + } + } + if (generation === oauthGeneration.current) { + oauthPollTimer.current = window.setTimeout(poll, Math.min(1500 * 2 ** failures, 10_000)); + } + }; + oauthPollTimer.current = window.setTimeout(poll, 1500); } catch (err) { + if (generation !== oauthGeneration.current) return; setOauthLoggingIn(null); setOauthNotice({ tone: "error", @@ -451,28 +479,41 @@ export function EasyModePage({ const el = document.getElementById(currentTargetId); if (el) { const rect = el.getBoundingClientRect(); - setSpotlightRect({ + const next = { top: Math.max(0, rect.top), left: Math.max(0, rect.left), width: rect.width, height: rect.height, - }); - el.scrollIntoView({ behavior: "smooth", block: "nearest" }); + }; + setSpotlightRect((previous) => previous && previous.top === next.top && previous.left === next.left + && previous.width === next.width && previous.height === next.height ? previous : next); } else { setSpotlightRect(null); } }, [guideActive, currentTargetId]); useEffect(() => { - const timer = setTimeout(updateSpotlightPosition, 100); - window.addEventListener("resize", updateSpotlightPosition); - window.addEventListener("scroll", updateSpotlightPosition, true); + let frame: number | null = null; + const schedulePosition = () => { + if (frame !== null) return; + frame = window.requestAnimationFrame(() => { + frame = null; + updateSpotlightPosition(); + }); + }; + const timer = setTimeout(() => { + if (currentTargetId) document.getElementById(currentTargetId)?.scrollIntoView({ behavior: "smooth", block: "nearest" }); + schedulePosition(); + }, 100); + window.addEventListener("resize", schedulePosition); + window.addEventListener("scroll", schedulePosition, { capture: true, passive: true }); return () => { clearTimeout(timer); - window.removeEventListener("resize", updateSpotlightPosition); - window.removeEventListener("scroll", updateSpotlightPosition, true); + if (frame !== null) window.cancelAnimationFrame(frame); + window.removeEventListener("resize", schedulePosition); + window.removeEventListener("scroll", schedulePosition, true); }; - }, [updateSpotlightPosition]); + }, [currentTargetId, updateSpotlightPosition]); useEffect(() => { setGuideCardPosition(null); diff --git a/src/pages/UsageRecordsPage.tsx b/src/pages/UsageRecordsPage.tsx index fe2aa0a..ce4ae06 100644 --- a/src/pages/UsageRecordsPage.tsx +++ b/src/pages/UsageRecordsPage.tsx @@ -29,6 +29,7 @@ import { getCurrentLocale, useI18n } from '../i18n'; import type { MessageKey } from '../i18n/resources'; import { formatCacheReadRate, formatGenerationSpeed } from '../services/usageMetrics'; import { formatUsageNumber } from '../services/usageNumber'; +import { createRefreshScheduler } from '../services/refreshScheduler'; type UsageTab = 'overview' | 'analysis' | 'events' | 'pricing' | 'data-management'; type UsageRange = '4h' | '24h' | 'today' | '7d' | '30d' | 'all' | 'custom'; @@ -292,6 +293,8 @@ export function UsageRecordsPage() { const [loading, setLoading] = useState(true); const [error, setError] = useState(''); const requestIdRef = useRef(0); + const schedulerRef = useRef | null>(null); + if (!schedulerRef.current) schedulerRef.current = createRefreshScheduler(); useEffect(() => { try { @@ -325,7 +328,7 @@ export function UsageRecordsPage() { }; }, [apiKeyHash, customEnd, customStart, model, provider, range, result, source]); - const loadData = useCallback( + const executeLoadData = useCallback( async (quiet = false) => { const requestId = ++requestIdRef.current; const { timeQuery, query } = buildQueries(); @@ -348,7 +351,9 @@ export function UsageRecordsPage() { statusRequest, optionsRequest, invoke('get_usage_overview', { query }), - invoke('get_usage_analysis', { query }), + model || provider || source || apiKeyHash || result !== 'all' + ? invoke('get_usage_analysis', { query }) + : optionsRequest, ]); if (requestId !== requestIdRef.current) return; setStatus(nextStatus); @@ -390,11 +395,23 @@ export function UsageRecordsPage() { if (requestId === requestIdRef.current) setLoading(false); } }, - [activeTab, buildQueries, page, pageSize] + [activeTab, buildQueries, page, pageSize, model, provider, source, apiKeyHash, result] + ); + + const loadData = useCallback( + (quiet = false) => { + if (!quiet) setLoading(true); + return schedulerRef.current!.schedule(() => executeLoadData(quiet), !quiet); + }, + [executeLoadData], ); useEffect(() => { void loadData(); + return () => { + ++requestIdRef.current; + schedulerRef.current?.cancelPending(); + }; }, [loadData]); useEffect(() => { diff --git a/src/services/providerHealthCheck.ts b/src/services/providerHealthCheck.ts index 3c9ad5a..1d48d07 100644 --- a/src/services/providerHealthCheck.ts +++ b/src/services/providerHealthCheck.ts @@ -273,25 +273,27 @@ export async function runProviderModelHealthChecks( checkModel: (model: ModelOption, index: number) => Promise, onModelChecked?: (result: ProviderModelHealthResult, index: number) => void, concurrency = PROVIDER_HEALTH_CONCURRENCY, + signal?: AbortSignal, ): Promise { const results: ProviderModelHealthResult[] = new Array(models.length); let nextIndex = 0; const workerCount = Math.min(models.length, Math.max(1, Math.floor(concurrency))); const runWorker = async () => { - while (nextIndex < models.length) { + while (nextIndex < models.length && !signal?.aborted) { const index = nextIndex; nextIndex += 1; const model = models[index]; if (!model) continue; const result = await checkModel(model, index); + if (signal?.aborted) break; results[index] = result; onModelChecked?.(result, index); } }; await Promise.all(Array.from({ length: workerCount }, runWorker)); - return results; + return results.filter((result) => result !== undefined); } export async function checkProviderModelsHealth( @@ -299,11 +301,13 @@ export async function checkProviderModelsHealth( models: ModelOption[], onModelChecked?: (result: ProviderModelHealthResult, index: number) => void, concurrency = PROVIDER_HEALTH_CONCURRENCY, + signal?: AbortSignal, ): Promise { return runProviderModelHealthChecks( models, (model) => checkProviderModelHealth(options, model.name), onModelChecked, concurrency, + signal, ); } diff --git a/src/services/refreshScheduler.ts b/src/services/refreshScheduler.ts new file mode 100644 index 0000000..d5dd239 --- /dev/null +++ b/src/services/refreshScheduler.ts @@ -0,0 +1,66 @@ +export function createRefreshScheduler(minIntervalMs = 1_000) { + type Task = () => Promise; + type Completion = { promise: Promise; resolve: () => void; reject: (error: unknown) => void }; + let running = false; + let completedAt = -Infinity; + let pending: Task | null = null; + let urgent = false; + let completion: Completion | null = null; + let timer: ReturnType | null = null; + + const pump = () => { + if (running || !pending) return; + if (timer !== null) clearTimeout(timer); + timer = null; + const delay = urgent ? 0 : Math.max(0, minIntervalMs - (performance.now() - completedAt)); + if (delay > 0) { + timer = setTimeout(pump, delay); + return; + } + const task = pending; + const currentCompletion = completion!; + pending = null; + completion = null; + urgent = false; + running = true; + void (async () => { + try { + await task(); + currentCompletion.resolve(); + } catch (error) { + currentCompletion.reject(error); + } finally { + running = false; + completedAt = performance.now(); + pump(); + } + })(); + }; + + return { + schedule(task: Task, immediate = false): Promise { + pending = task; + urgent ||= immediate; + if (!completion) { + let resolve!: () => void; + let reject!: (error: unknown) => void; + const promise = new Promise((onResolved, onRejected) => { + resolve = onResolved; + reject = onRejected; + }); + completion = { promise, resolve, reject }; + } + const result = completion.promise; + pump(); + return result; + }, + cancelPending() { + if (timer !== null) clearTimeout(timer); + timer = null; + pending = null; + urgent = false; + completion?.resolve(); + completion = null; + }, + }; +} diff --git a/tests/providerHealthCheck.test.ts b/tests/providerHealthCheck.test.ts index 45742e8..a05b9e3 100644 --- a/tests/providerHealthCheck.test.ts +++ b/tests/providerHealthCheck.test.ts @@ -8,6 +8,43 @@ import { import { modelEndpointCandidates } from '../src/services/modelService'; describe('API 接入健康检测', () => { + it('取消后不派发剩余模型，也不更新已关闭的界面', async () => { + const controller = new AbortController(); + const requested: string[] = []; + const delivered: string[] = []; + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + const checking = runProviderModelHealthChecks( + Array.from({ length: 100 }, (_, index) => ({ name: String(index) })), + async (model) => { + requested.push(model.name); + await gate; + return { model: model.name, status: 'healthy', success: true }; + }, + (result) => delivered.push(result.model), + 4, + controller.signal, + ); + expect(requested).toHaveLength(4); + controller.abort(); + release(); + expect(await checking).toEqual([]); + expect(requested).toHaveLength(4); + expect(delivered).toEqual([]); + }); + + it('已取消的批次不会发起请求', async () => { + const controller = new AbortController(); + controller.abort(); + let requested = false; + const results = await runProviderModelHealthChecks([{ name: 'unused' }], async (model) => { + requested = true; + return { model: model.name, status: 'healthy', success: true }; + }, undefined, 4, controller.signal); + expect(results).toEqual([]); + expect(requested).toBe(false); + }); + it('健康检测只保留已勾选模型，并使用发现结果补充模型信息', () => { const models = mergeProviderHealthModels( [ diff --git a/tests/refreshScheduler.test.ts b/tests/refreshScheduler.test.ts new file mode 100644 index 0000000..1f79ebc --- /dev/null +++ b/tests/refreshScheduler.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from 'bun:test'; +import { createRefreshScheduler } from '../src/services/refreshScheduler'; + +const deferred = () => { + let resolve!: () => void; + const promise = new Promise((done) => { resolve = done; }); + return { promise, resolve }; +}; + +describe('refresh scheduler', () => { + it('runs one request at a time and retains only the latest trailing request', async () => { + const scheduler = createRefreshScheduler(0); + const firstGate = deferred(); + const calls: number[] = []; + const first = scheduler.schedule(async () => { + calls.push(1); + await firstGate.promise; + }); + const replaced = scheduler.schedule(async () => { calls.push(2); }); + let latest = replaced; + for (let index = 3; index <= 1000; index += 1) { + latest = scheduler.schedule(async () => { calls.push(index); }); + expect(latest).toBe(replaced); + } + expect(calls).toEqual([1]); + firstGate.resolve(); + await Promise.all([first, replaced, latest]); + expect(calls).toEqual([1, 1000]); + }); + + it('waits between background refreshes but lets an explicit refresh bypass the delay', async () => { + const scheduler = createRefreshScheduler(30); + await scheduler.schedule(async () => {}); + const calls: string[] = []; + const queued = scheduler.schedule(async () => { calls.push('background'); }); + expect(calls).toEqual([]); + await scheduler.schedule(async () => { calls.push('manual'); }, true); + await queued; + expect(calls).toEqual(['manual']); + const started = performance.now(); + await scheduler.schedule(async () => { calls.push('delayed'); }); + expect(performance.now() - started).toBeGreaterThanOrEqual(25); + }); + + it('cancels queued work during navigation and accepts a fresh request', async () => { + const scheduler = createRefreshScheduler(0); + const gate = deferred(); + const active = scheduler.schedule(() => gate.promise); + let obsoleteRan = false; + const obsolete = scheduler.schedule(async () => { obsoleteRan = true; }); + scheduler.cancelPending(); + await obsolete; + let freshRan = false; + const fresh = scheduler.schedule(async () => { freshRan = true; }, true); + gate.resolve(); + await Promise.all([active, fresh]); + expect(obsoleteRan).toBe(false); + expect(freshRan).toBe(true); + }); + + it('continues after a failed refresh', async () => { + const scheduler = createRefreshScheduler(0); + await expect(scheduler.schedule(async () => { throw new Error('query failed'); })).rejects.toThrow('query failed'); + let ran = false; + await scheduler.schedule(async () => { ran = true; }); + expect(ran).toBe(true); + }); +});