feat(webdav): add robust auto sync with failure feedback

(cherry picked from commit bb6760124a62a964b36902c004e173534910728f)
This commit is contained in:
saladday
2026-02-14 18:34:05 +08:00
parent 7b20c17ea4
commit ff29b939a8
17 changed files with 495 additions and 18 deletions
+100 -16
View File
@@ -1,9 +1,7 @@
#![allow(non_snake_case)]
use serde_json::{Value, json};
use std::future::Future;
use std::sync::OnceLock;
use tauri::State;
use serde_json::{json, Value};
use tauri::{AppHandle, Emitter, State};
use crate::commands::sync_support::{
attach_warning, post_sync_warning_from_result, run_post_import_sync,
@@ -13,8 +11,9 @@ use crate::services::webdav_sync as webdav_sync_service;
use crate::settings::{self, WebDavSyncSettings};
use crate::store::AppState;
fn persist_sync_error(settings: &mut WebDavSyncSettings, error: &AppError) {
fn persist_sync_error(settings: &mut WebDavSyncSettings, error: &AppError, source: &str) {
settings.status.last_error = Some(error.to_string());
settings.status.last_error_source = Some(source.to_string());
let _ = settings::update_webdav_sync_status(settings.status.clone());
}
@@ -57,20 +56,16 @@ fn resolve_password_for_request(
incoming
}
#[cfg(test)]
fn webdav_sync_mutex() -> &'static tokio::sync::Mutex<()> {
static LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
webdav_sync_service::sync_mutex()
}
async fn run_with_webdav_lock<T, Fut>(operation: Fut) -> Result<T, AppError>
where
Fut: Future<Output = Result<T, AppError>>,
Fut: std::future::Future<Output = Result<T, AppError>>,
{
let result = {
let _guard = webdav_sync_mutex().lock().await;
operation.await
};
result
webdav_sync_service::run_with_sync_lock(operation).await
}
fn map_sync_result<T, F>(result: Result<T, AppError>, on_error: F) -> Result<T, String>
@@ -86,6 +81,64 @@ where
}
}
fn should_run_auto_sync(settings: Option<&WebDavSyncSettings>) -> bool {
let Some(sync) = settings else {
return false;
};
sync.enabled && sync.auto_sync
}
fn emit_webdav_sync_status_updated(
app: &AppHandle,
source: &str,
status: &str,
error: Option<&str>,
) {
let payload = match error {
Some(message) => json!({
"source": source,
"status": status,
"error": message,
}),
None => json!({
"source": source,
"status": status,
}),
};
if let Err(err) = app.emit("webdav-sync-status-updated", payload) {
log::debug!("[WebDAV] failed to emit sync status update event: {err}");
}
}
pub(crate) async fn run_auto_sync_upload(
db: &crate::database::Database,
app: &AppHandle,
) -> Result<(), AppError> {
let mut settings = settings::get_webdav_sync_settings();
if !should_run_auto_sync(settings.as_ref()) {
return Ok(());
}
let mut sync_settings = match settings.take() {
Some(value) => value,
None => return Ok(()),
};
let result = run_with_webdav_lock(webdav_sync_service::upload(db, &mut sync_settings)).await;
match result {
Ok(_) => {
emit_webdav_sync_status_updated(app, "auto", "success", None);
Ok(())
}
Err(err) => {
persist_sync_error(&mut sync_settings, &err, "auto");
emit_webdav_sync_status_updated(app, "auto", "error", Some(&err.to_string()));
Err(err)
}
}
}
#[tauri::command]
pub async fn webdav_test_connection(
settings: WebDavSyncSettings,
@@ -112,7 +165,9 @@ pub async fn webdav_sync_upload(state: State<'_, AppState>) -> Result<Value, Str
let mut settings = require_enabled_webdav_settings()?;
let result = run_with_webdav_lock(webdav_sync_service::upload(&db, &mut settings)).await;
map_sync_result(result, |error| persist_sync_error(&mut settings, error))
map_sync_result(result, |error| {
persist_sync_error(&mut settings, error, "manual")
})
}
#[tauri::command]
@@ -120,10 +175,11 @@ pub async fn webdav_sync_download(state: State<'_, AppState>) -> Result<Value, S
let db = state.db.clone();
let db_for_sync = db.clone();
let mut settings = require_enabled_webdav_settings()?;
let _auto_sync_suppression = crate::services::webdav_auto_sync::AutoSyncSuppressionGuard::new();
let sync_result = run_with_webdav_lock(webdav_sync_service::download(&db, &mut settings)).await;
let mut result = map_sync_result(sync_result, |error| {
persist_sync_error(&mut settings, error)
persist_sync_error(&mut settings, error, "manual")
})?;
// Post-download sync is best-effort: snapshot restore has already succeeded.
@@ -179,8 +235,8 @@ mod tests {
use crate::error::AppError;
use crate::settings::{AppSettings, WebDavSyncSettings};
use serial_test::serial;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
#[tokio::test]
@@ -287,6 +343,7 @@ mod tests {
persist_sync_error(
&mut current,
&crate::error::AppError::Config("boom".to_string()),
"manual",
);
let after = crate::settings::get_webdav_sync_settings().expect("read webdav settings");
@@ -304,6 +361,7 @@ mod tests {
.contains("boom"),
"status error should be updated"
);
assert_eq!(after.status.last_error_source.as_deref(), Some("manual"));
}
#[test]
@@ -354,4 +412,30 @@ mod tests {
assert!(settings.enabled);
assert_eq!(settings.base_url, "https://dav.example.com/dav/");
}
#[test]
fn should_run_auto_sync_requires_enabled_and_auto_sync_flag() {
assert!(!super::should_run_auto_sync(None));
let disabled = WebDavSyncSettings {
enabled: false,
auto_sync: true,
..WebDavSyncSettings::default()
};
assert!(!super::should_run_auto_sync(Some(&disabled)));
let auto_sync_off = WebDavSyncSettings {
enabled: true,
auto_sync: false,
..WebDavSyncSettings::default()
};
assert!(!super::should_run_auto_sync(Some(&auto_sync_off)));
let enabled = WebDavSyncSettings {
enabled: true,
auto_sync: true,
..WebDavSyncSettings::default()
};
assert!(super::should_run_auto_sync(Some(&enabled)));
}
}
+14 -1
View File
@@ -37,7 +37,7 @@ pub use dao::OmoGlobalConfig;
use crate::config::get_app_config_dir;
use crate::error::AppError;
use rusqlite::Connection;
use rusqlite::{hooks::Action, Connection};
use serde::Serialize;
use std::sync::Mutex;
@@ -76,6 +76,17 @@ pub struct Database {
pub(crate) conn: Mutex<Connection>,
}
fn register_db_change_hook(conn: &Connection) {
conn.update_hook(Some(
|action: Action, _database: &str, table: &str, _row_id: i64| match action {
Action::SQLITE_INSERT | Action::SQLITE_UPDATE | Action::SQLITE_DELETE => {
crate::services::webdav_auto_sync::notify_db_changed(table);
}
_ => {}
},
));
}
impl Database {
/// 初始化数据库连接并创建表
///
@@ -93,6 +104,7 @@ impl Database {
// 启用外键约束
conn.execute("PRAGMA foreign_keys = ON;", [])
.map_err(|e| AppError::Database(e.to_string()))?;
register_db_change_hook(&conn);
let db = Self {
conn: Mutex::new(conn),
@@ -111,6 +123,7 @@ impl Database {
// 启用外键约束
conn.execute("PRAGMA foreign_keys = ON;", [])
.map_err(|e| AppError::Database(e.to_string()))?;
register_db_change_hook(&conn);
let db = Self {
conn: Mutex::new(conn),
+4
View File
@@ -702,6 +702,10 @@ pub fn run() {
}
let _tray = tray_builder.build(app)?;
crate::services::webdav_auto_sync::start_worker(
app_state.db.clone(),
app.handle().clone(),
);
// 将同一个实例注入到全局状态,避免重复创建导致的不一致
app.manage(app_state);
+1
View File
@@ -11,6 +11,7 @@ pub mod speedtest;
pub mod stream_check;
pub mod usage_stats;
pub mod webdav;
pub mod webdav_auto_sync;
pub mod webdav_sync;
pub use config::ConfigService;
+171
View File
@@ -0,0 +1,171 @@
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::{Receiver, Sender, channel};
const AUTO_SYNC_DEBOUNCE_MS: u64 = 1000;
pub(crate) const MAX_AUTO_SYNC_WAIT_MS: u64 = 10_000;
static DB_CHANGE_TX: OnceLock<Sender<String>> = OnceLock::new();
static AUTO_SYNC_SUPPRESS_DEPTH: AtomicUsize = AtomicUsize::new(0);
pub(crate) struct AutoSyncSuppressionGuard;
impl AutoSyncSuppressionGuard {
pub fn new() -> Self {
AUTO_SYNC_SUPPRESS_DEPTH.fetch_add(1, Ordering::SeqCst);
Self
}
}
impl Drop for AutoSyncSuppressionGuard {
fn drop(&mut self) {
let _ = AUTO_SYNC_SUPPRESS_DEPTH.fetch_update(
Ordering::SeqCst,
Ordering::SeqCst,
|value| Some(value.saturating_sub(1)),
);
}
}
pub(crate) fn is_auto_sync_suppressed() -> bool {
AUTO_SYNC_SUPPRESS_DEPTH.load(Ordering::SeqCst) > 0
}
pub fn should_trigger_for_table(table: &str) -> bool {
let normalized = table.trim().to_ascii_lowercase();
matches!(
normalized.as_str(),
"providers"
| "provider_endpoints"
| "mcp_servers"
| "prompts"
| "skills"
| "skill_repos"
| "settings"
| "proxy_config"
)
}
pub(crate) fn enqueue_change_signal(tx: &Sender<String>, table: &str) -> bool {
match tx.try_send(table.to_string()) {
Ok(()) => true,
Err(TrySendError::Full(_)) | Err(TrySendError::Closed(_)) => false,
}
}
pub(crate) fn auto_sync_wait_duration(started_at: Instant, now: Instant) -> Option<Duration> {
let max_wait = Duration::from_millis(MAX_AUTO_SYNC_WAIT_MS);
let debounce = Duration::from_millis(AUTO_SYNC_DEBOUNCE_MS);
let elapsed = now.saturating_duration_since(started_at);
if elapsed >= max_wait {
return None;
}
Some(debounce.min(max_wait - elapsed))
}
pub fn notify_db_changed(table: &str) {
if is_auto_sync_suppressed() {
return;
}
if !should_trigger_for_table(table) {
return;
}
let Some(tx) = DB_CHANGE_TX.get() else {
return;
};
let _ = enqueue_change_signal(tx, table);
}
pub fn start_worker(db: Arc<crate::database::Database>, app: tauri::AppHandle) {
if DB_CHANGE_TX.get().is_some() {
return;
}
// Buffer size 1 is enough: we only need "dirty" signals, not every event.
let (tx, rx) = channel::<String>(1);
if DB_CHANGE_TX.set(tx).is_err() {
return;
}
tauri::async_runtime::spawn(async move {
run_worker_loop(db, rx, app).await;
});
}
async fn run_worker_loop(
db: Arc<crate::database::Database>,
mut rx: Receiver<String>,
app: tauri::AppHandle,
) {
while let Some(first_table) = rx.recv().await {
let started_at = Instant::now();
let mut merged_count = 1usize;
loop {
let Some(wait_for) = auto_sync_wait_duration(started_at, Instant::now()) else {
break;
};
let timeout = tokio::time::timeout(wait_for, rx.recv()).await;
match timeout {
Ok(Some(_)) => merged_count += 1,
Ok(None) => return,
Err(_) => break,
}
}
log::debug!(
"[WebDAV][AutoSync] Triggered by table={first_table}, merged_changes={merged_count}"
);
if let Err(err) = crate::commands::run_auto_sync_upload(&db, &app).await {
log::warn!("[WebDAV][AutoSync] Upload failed: {err}");
}
}
}
#[cfg(test)]
mod tests {
use super::{
AutoSyncSuppressionGuard, MAX_AUTO_SYNC_WAIT_MS, auto_sync_wait_duration,
enqueue_change_signal, is_auto_sync_suppressed, should_trigger_for_table,
};
use std::time::{Duration, Instant};
use tokio::sync::mpsc::channel;
#[test]
fn should_trigger_sync_for_config_tables_only() {
assert!(should_trigger_for_table("providers"));
assert!(should_trigger_for_table("settings"));
assert!(!should_trigger_for_table("proxy_request_logs"));
assert!(!should_trigger_for_table("provider_health"));
}
#[test]
fn suppression_guard_enables_and_restores_state() {
assert!(!is_auto_sync_suppressed());
{
let _guard = AutoSyncSuppressionGuard::new();
assert!(is_auto_sync_suppressed());
}
assert!(!is_auto_sync_suppressed());
}
#[test]
fn max_wait_caps_flush_latency_for_continuous_events() {
let started = Instant::now();
let later = started + Duration::from_millis(MAX_AUTO_SYNC_WAIT_MS + 1);
assert!(auto_sync_wait_duration(started, later).is_none());
}
#[tokio::test]
async fn enqueue_change_signal_drops_when_channel_is_full() {
let (tx, _rx) = channel::<String>(1);
assert!(enqueue_change_signal(&tx, "providers"));
assert!(!enqueue_change_signal(&tx, "providers"));
}
}
+16
View File
@@ -5,7 +5,9 @@
use std::collections::BTreeMap;
use std::fs;
use std::future::Future;
use std::process::Command;
use std::sync::OnceLock;
use chrono::Utc;
use serde::{Deserialize, Serialize};
@@ -34,6 +36,19 @@ const REMOTE_SKILLS_ZIP: &str = "skills.zip";
const REMOTE_MANIFEST: &str = "manifest.json";
const MAX_DEVICE_NAME_LEN: usize = 64;
pub fn sync_mutex() -> &'static tokio::sync::Mutex<()> {
static LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
}
pub async fn run_with_sync_lock<T, Fut>(operation: Fut) -> Result<T, AppError>
where
Fut: Future<Output = Result<T, AppError>>,
{
let _guard = sync_mutex().lock().await;
operation.await
}
fn localized(key: &'static str, zh: impl Into<String>, en: impl Into<String>) -> AppError {
AppError::localized(key, zh, en)
}
@@ -214,6 +229,7 @@ fn persist_sync_success(
let status = WebDavSyncStatus {
last_sync_at: Some(Utc::now().timestamp()),
last_error: None,
last_error_source: None,
last_local_manifest_hash: Some(manifest_hash.clone()),
last_remote_manifest_hash: Some(manifest_hash),
last_remote_etag: etag,
+5
View File
@@ -72,6 +72,8 @@ pub struct WebDavSyncStatus {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_error: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_error_source: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_remote_etag: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_local_manifest_hash: Option<String>,
@@ -93,6 +95,8 @@ pub struct WebDavSyncSettings {
#[serde(default)]
pub enabled: bool,
#[serde(default)]
pub auto_sync: bool,
#[serde(default)]
pub base_url: String,
#[serde(default)]
pub username: String,
@@ -110,6 +114,7 @@ impl Default for WebDavSyncSettings {
fn default() -> Self {
Self {
enabled: false,
auto_sync: false,
base_url: String::new(),
username: String::new(),
password: String::new(),