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
This commit is contained in:
lzt404 committed 2026-09-06 01:28:58 +08:00
1 parent 45ecd37ce5
commit 9cf3878b28
19 files changed
+1050 -300

No files matched your search

+64
View File
@@ -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();
+11 -1
View File
@@ -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::<GuiConfigState>().snapshot() else {
continue;
};
+10 -6
View File
@@ -11,12 +11,16 @@ pub(crate) fn detect_core_platform() -> Result<CorePlatform, String> {
}
#[tauri::command]
pub(crate) fn get_core_status(
process_state: tauri::State<'_, CoreProcessState>,
gui_config_state: tauri::State<'_, GuiConfigState>,
) -> Result<CoreStatus, String> {
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<CoreStatus, String> {
tauri::async_runtime::spawn_blocking(move || {
let config = app.state::<GuiConfigState>().snapshot()?;
current_core_status(
Some(app.state::<CoreProcessState>().inner()),
Some(config.port),
)
})
.await
.map_err(|error| format!("内核状态后台任务失败: {error}"))?
}
pub(crate) fn emit_core_status(app: &tauri::AppHandle, status: &CoreStatus) {
+13 -10
View File
@@ -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}"))?;
+152 -45
View File
@@ -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::<GuiConfigState>().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<PathBuf>,
error: Option<String>,
}
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<notify::Event, notify::Error>| 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<notify::Event, notify::Error>| {
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<PathBuf>| {
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]);
}
}
+102 -82
View File
@@ -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<CoreStatus, String>,
) -> Result<CoreStatus, String> {
tauri::async_runtime::spawn_blocking(move || {
let _guard = CORE_OPERATION_LOCK
.try_lock()
.map_err(|_| "内核正在执行其他操作,请稍后重试".to_string())?;
let status = operation(
app.state::<CoreProcessState>().inner(),
app.state::<GuiConfigState>().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<Option<BundledCoreInfo>, 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<CoreInstallResult, String> {
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::<CoreDownloadState>();
let process_state = app.state::<CoreProcessState>();
let gui_config_state = app.state::<GuiConfigState>();
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<String>,
) -> Result<CoreInstallResult, String> {
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::<CoreDownloadState>();
let process_state = app.state::<CoreProcessState>();
let gui_config_state = app.state::<GuiConfigState>();
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<F>(
@@ -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<CoreStatus, String> {
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<CoreStatus, String> {
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<CoreStatus, String> {
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<CoreStatus, String> {
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<CoreStatus, String> {
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<CoreStatus, String> {
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))?;
+4
View File
@@ -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::<GuiConfigState>();
let process_state = core_app.state::<CoreProcessState>();
let Ok(config) = gui_config_state.snapshot() else {
+16 -12
View File
@@ -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, String> {
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<Result<reqwest::Client, String>> = 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(
+36
View File
@@ -0,0 +1,36 @@
use std::time::{Duration, Instant};
#[derive(Default)]
pub(crate) struct ProgressThrottle {
last: Option<Instant>,
}
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));
}
}
+5
View File
@@ -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
+348 -107
View File
@@ -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<T, F>(task: F) -> Result<T, String>
where
T: Send + 'static,
F: FnOnce() -> Result<T, String> + 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<UsageCollectorInner>,
}
@@ -545,9 +564,8 @@ fn initialize_usage_storage_at(root: &Path) -> Result<(), String> {
}
#[tauri::command]
pub(crate) fn repair_usage_cache_records() -> Result<UsageRepairResult, String> {
let root = usage_root_dir()?;
repair_usage_cache_records_at(&root)
pub(crate) async fn repair_usage_cache_records() -> Result<UsageRepairResult, String> {
run_usage_task(|| repair_usage_cache_records_at(&usage_root_dir()?)).await
}
fn repair_usage_cache_records_at(root: &Path) -> Result<UsageRepairResult, String> {
@@ -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<UsageSubscription> = 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::<UsageCollectorState>()
.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::<UsageCollectorState>()
.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::<CoreProcessState>();
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::<CoreProcessState>();
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<UsageOverview, String> {
let connection = open_usage_database()?;
load_usage_overview(&connection, &query)
pub(crate) async fn get_usage_overview(query: UsageQuery) -> Result<UsageOverview, String> {
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<ModelPrice> {
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<HashMap<String, ModelPrice>, 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, ModelPrice>, String> {
static PRICES: LazyLock<Result<HashMap<String, ModelPrice>, 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<UsagePricing, String> {
let connection = open_usage_database()?;
load_usage_pricing(&connection, &query)
pub(crate) async fn get_usage_pricing(query: UsageQuery) -> Result<UsagePricing, String> {
run_usage_task(move || load_usage_pricing(&open_usage_database()?, &query)).await
}
fn load_usage_pricing(connection: &Connection, query: &UsageQuery) -> Result<UsagePricing, String> {
@@ -3043,25 +3074,27 @@ fn load_usage_pricing(connection: &Connection, query: &UsageQuery) -> Result<Usa
}
#[tauri::command]
pub(crate) fn save_usage_model_price(mut price: ModelPrice) -> 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::<std::collections::HashSet<_>>();
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::<std::collections::HashSet<_>>();
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::<std::collections::BTreeSet<_>>();
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::<std::collections::BTreeSet<_>>();
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<UsageAnalysis, String> {
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<UsageAnalysis, String> {
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<String, UsageCategory>; 4] = Default::default();
let mut key_labels = HashMap::<String, (String, String)>::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<String, UsageCategory>) -> Vec<UsageCategory> {
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<UsageEventPage, String> {
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<UsageCategory>| {
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();
+20 -12
View File
@@ -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<CoreRuntimeContextValue | null>(null);
export function CoreRuntimeProvider({ children }: { children: ReactNode }) {
const [status, setStatus] = useState<CoreStatus | null>(null);
const [statusError, setStatusError] = useState('');
const refreshRequest = useRef<Promise<void> | 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<CoreStatus>('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<CoreStatus>('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;
+15 -4
View File
@@ -1338,6 +1338,13 @@ function ProviderHealthDialog({ row, onClose }: ProviderHealthDialogProps) {
const [search, setSearch] = useState('');
const [results, setResults] = useState<Record<string, ProviderModelHealthState>>({});
const [checkingAll, setCheckingAll] = useState(false);
const healthControllerRef = useRef<AbortController | null>(null);
useEffect(() => {
const controller = new AbortController();
healthControllerRef.current = controller;
return () => controller.abort();
}, []);
const healthOptions = useMemo<ProviderHealthCheckOptions>(() => ({
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);
}
};
+57 -16
View File
@@ -125,6 +125,7 @@ export function EasyModePage({
message: string;
} | null>(null);
const oauthPollTimer = useRef<number | null>(null);
const oauthGeneration = useRef(0);
const [selectedApiSection, setSelectedApiSection] = useState<ApiSection>("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);
+20 -3
View File
@@ -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<ReturnType<typeof createRefreshScheduler> | 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<UsageOverview>('get_usage_overview', { query }),
invoke<UsageAnalysis>('get_usage_analysis', { query }),
model || provider || source || apiKeyHash || result !== 'all'
? invoke<UsageAnalysis>('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(() => {
+6 -2
View File
@@ -273,25 +273,27 @@ export async function runProviderModelHealthChecks(
checkModel: (model: ModelOption, index: number) => Promise<ProviderModelHealthResult>,
onModelChecked?: (result: ProviderModelHealthResult, index: number) => void,
concurrency = PROVIDER_HEALTH_CONCURRENCY,
signal?: AbortSignal,
): Promise<ProviderModelHealthResult[]> {
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<ProviderModelHealthResult[]> {
return runProviderModelHealthChecks(
models,
(model) => checkProviderModelHealth(options, model.name),
onModelChecked,
concurrency,
signal,
);
}
+66
View File
@@ -0,0 +1,66 @@
export function createRefreshScheduler(minIntervalMs = 1_000) {
type Task = () => Promise<void>;
type Completion = { promise: Promise<void>; 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<typeof setTimeout> | 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<void> {
pending = task;
urgent ||= immediate;
if (!completion) {
let resolve!: () => void;
let reject!: (error: unknown) => void;
const promise = new Promise<void>((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;
},
};
}
+37
View File
@@ -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<void>((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(
[
+68
View File
@@ -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<void>((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);
});
});