From 9e295b9a5d126d3ad52cc5ba2a04d486cb3227a6 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Tue, 8 Sep 2026 19:58:39 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20=E8=B7=A8=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E5=8C=BA=E9=87=8D=E5=90=AF=E4=BF=9D=E7=95=99=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E4=B8=8E=E6=9A=82=E5=81=9C=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../OpenNexus生产化实施进度-2026-09-08.md | 8 ++ frontend/src-tauri/src/lib.rs | 2 + frontend/src-tauri/src/sync_commands.rs | 115 ++++++++++------ frontend/src-tauri/src/sync_retry.rs | 130 ++++++++++++++++++ frontend/src-tauri/src/workspace.rs | 7 +- .../features/settings/SyncSettings.spec.ts | 9 ++ .../src/features/settings/SyncSettings.vue | 6 +- 7 files changed, 227 insertions(+), 50 deletions(-) create mode 100644 frontend/src-tauri/src/sync_retry.rs diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 05f6ef2..d0ca998 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -108,3 +108,11 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写 - 设置编辑先保存按 Vault 隔离的持久草稿，再通过 CAS 写入 Workspace。覆盖选择串行等待在途写入，选择远端期间的新编辑不被丢弃；存储删除失败时保留内存及磁盘草稿并报告失败。远端主题应用不产生同步回流。 - 初始上传和合并预览包含当前偏好；空库下载不提前创建记录。真实双客户端 HTTP 测试新增主题记录传播，嵌套未知字段和非法值由 Rust 测试验证。 - 本增量全量前端 97 文件 / 515 项通过；其后补充存储故障修复，两文件 7 项针对性测试通过。TypeScript 两项目检查通过，Rust desktop 全目标通过（数量见测试日志），Clippy `-D warnings` 通过。测试日志为 `.build/preferences-frontend-tests.log` 和 `.build/preferences-rust-tests.log`。生产签名、OS 沙箱、Skill/配置等未完成项仍未通过整体验收。 + + +## 增量：跨重启的同步退避和停止状态 + +- Workspace schema 9 持久保存每个绑定的错误码、失败次数、下次重试时间及 halted 状态。旧 schema 升级前保留独立 SQLite 备份。重启后自动同步沿用该状态；成功、同账户登录恢复或手动继续会清除当前绑定状态。 +- 取消/切换绑定不增加失败次数，旧绑定回调不能修改新绑定；保存的错误仅允许有长度限制的机器码，不保存服务端任意响应正文。Retry-After 和指数退避保持 1–3600 秒限制。界面对停止自动重试状态不再显示误导性的倒计时。 +- 连续关闭重开 20 次验证失败次数与时间保留，另验证认证停止、状态清除、取消、错误正文过滤和新旧绑定隔离。当前是绑定级轮次状态；每个上传作业的独立尝试历史、强制刷新后的 401 重试和完整服务故障矩阵仍需完成。 +- 全量前端 97 文件 / 517 项、Rust desktop 全目标 52 项通过（另 1 个父测试驱动并强杀的辅助入口）；TypeScript 两项目检查和 Clippy `-D warnings` 通过。日志 `.build/retry-frontend-tests.log`、`.build/retry-rust-tests.log`。Core 打包正在更新，不视为签名发布包。 diff --git a/frontend/src-tauri/src/lib.rs b/frontend/src-tauri/src/lib.rs index 348cf1c..4d924f6 100644 --- a/frontend/src-tauri/src/lib.rs +++ b/frontend/src-tauri/src/lib.rs @@ -22,3 +22,5 @@ pub mod sync_resolution; pub mod sync_state; pub mod workspace; pub mod workspace_broker; + +pub mod sync_retry; diff --git a/frontend/src-tauri/src/sync_commands.rs b/frontend/src-tauri/src/sync_commands.rs index 7f32bc0..1343ba6 100644 --- a/frontend/src-tauri/src/sync_commands.rs +++ b/frontend/src-tauri/src/sync_commands.rs @@ -13,7 +13,7 @@ use std::{ atomic::{AtomicU64, Ordering}, Mutex, }, - time::{Duration, Instant}, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use tauri::State; use zeroize::Zeroizing; @@ -26,10 +26,6 @@ pub struct Runtime { #[derive(Default)] struct Progress { running: bool, - error: Option, - failures: u32, - retry: Option, - halted: bool, } impl Runtime { pub fn cancel(&self) { @@ -63,6 +59,21 @@ pub async fn sync_login(host: State<'_, Host>, request: Login) -> Result, binding_id: String) -> Result<(), Stri #[tauri::command] pub fn sync_pause(host: State<'_, Host>, binding_id: String, paused: bool) -> Result<(), String> { host.sync.cancel(); - with_workspace(&host, |ws| ws.sync_pause(&binding_id, paused))?; + with_workspace(&host, |ws| { + ws.sync_pause(&binding_id, paused)?; + if !paused { + ws.sync_retry_clear(&binding_id)?; + } + Ok(()) + })?; host.sync .status .lock() @@ -247,13 +264,18 @@ pub fn sync_status(host: State<'_, Host>) -> Result { .unwrap_or_default(); Ok(( ws.vault_id.clone(), - binding, + binding.clone(), paused, conflicts, ws.pending_count()?, + binding + .as_ref() + .map(|b| ws.sync_retry(&b.id)) + .transpose()? + .unwrap_or_default(), )) })?; - let (vault_id, binding, paused, conflicts, pending) = snapshot; + let (vault_id, binding, paused, conflicts, pending, retry) = snapshot; let credential_state = if let Some(b) = &binding { sync_auth::available(&host.credentials, &b.endpoint, &b.account) .map(|exists| { @@ -271,7 +293,7 @@ pub fn sync_status(host: State<'_, Host>) -> Result { let statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?; let status = binding.as_ref().and_then(|b| statuses.get(&b.id)); Ok( - json!({"vault_id":vault_id,"binding":binding,"paused":paused,"pending":pending,"conflicts":conflicts,"credential_state":credential_state,"running":status.is_some_and(|s|s.running),"error":status.and_then(|s|s.error.as_ref()),"retry_in":status.and_then(|s|s.retry).map(|time|time.saturating_duration_since(Instant::now()).as_secs())}), + json!({"vault_id":vault_id,"binding":binding,"paused":paused,"pending":pending,"conflicts":conflicts,"credential_state":credential_state,"running":status.is_some_and(|s|s.running),"error":retry.error,"retry_in":retry.retry_at.map(|_| retry.remaining(now())),"failures":retry.failures,"halted":retry.halted}), ) } #[tauri::command] @@ -315,49 +337,54 @@ pub async fn run(host: &Host, manual: bool) -> Result<(), String> { if with_workspace(host, |ws| ws.sync_paused(&binding.id))? { return Err("SYNC_PAUSED".into()); } - { - let mut statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?; - let state = statuses.entry(binding.id.clone()).or_default(); - if !manual && (state.halted || state.retry.is_some_and(|v| v > Instant::now())) { - return Ok(()); - } - state.running = true; - state.error = None; + let retry = with_workspace(host, |ws| ws.sync_retry(&binding.id))?; + if !manual && (retry.halted || retry.remaining(now()) > 0) { + return Ok(()); } + host.sync + .status + .lock() + .map_err(|_| "HOST_BUSY")? + .entry(binding.id.clone()) + .or_default() + .running = true; let result = cycle(host, &binding).await; - let mut statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?; - let state = statuses.entry(binding.id.clone()).or_default(); - state.running = false; + host.sync + .status + .lock() + .map_err(|_| "HOST_BUSY")? + .entry(binding.id.clone()) + .or_default() + .running = false; match result { - Ok(()) => { - *state = Progress::default(); - Ok(()) - } + Ok(()) => with_workspace(host, |ws| ws.sync_retry_clear(&binding.id)), Err(error) => { - state.failures = if error.code == "CREDENTIALS_LOCKED" { - 0 - } else { - state.failures.saturating_add(1) - }; - state.error = Some(error.code.clone()); - state.halted = matches!(error.status, 401 | 403 | 413 | 426 | 507) - || matches!( - error.code.as_str(), - "PROTOCOL_INCOMPATIBLE" | "SYNC_LOGIN_REQUIRED" - ); - state.retry = Some( - Instant::now() - + Duration::from_secs( - error - .retry_after - .unwrap_or(2u64.saturating_pow(state.failures.min(8))) - .clamp(1, 3600), - ), - ); + if !matches!( + error.code.as_str(), + "SYNC_CANCELLED" | "SYNC_BINDING_CHANGED" | "VAULT_CHANGED" + ) { + with_workspace(host, |ws| { + ws.sync_retry_fail( + &binding.id, + &error.code, + error.status, + error.retry_after, + now(), + ) + })?; + } Err(error.code) } } } +fn now() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + .min(i64::MAX as u64) as i64 +} + async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> { let epoch = host.sync.epoch.load(Ordering::SeqCst); let work = async { diff --git a/frontend/src-tauri/src/sync_retry.rs b/frontend/src-tauri/src/sync_retry.rs new file mode 100644 index 0000000..b6be3ef --- /dev/null +++ b/frontend/src-tauri/src/sync_retry.rs @@ -0,0 +1,130 @@ +//! Binding-scoped retry decisions survive restart; wall time is bounded after clock changes. +use crate::workspace::{Result, Workspace}; +use rusqlite::{params, OptionalExtension}; +use serde::Serialize; + +#[derive(Default, Debug, Serialize)] +pub struct Retry { + pub error: Option, + pub failures: u32, + pub retry_at: Option, + pub halted: bool, +} +impl Retry { + pub fn remaining(&self, now: i64) -> u64 { + self.retry_at + .map(|at| at.saturating_sub(now).clamp(0, 3600) as u64) + .unwrap_or(0) + } +} +impl Workspace { + pub fn sync_retry(&self, binding: &str) -> Result { + self.check_binding(binding)?; + Ok(self + .db + .query_row( + "SELECT error,failures,retry_at,halted FROM sync_retry WHERE binding=?1", + [binding], + |row| { + Ok(Retry { + error: row.get(0)?, + failures: row.get(1)?, + retry_at: row.get(2)?, + halted: row.get(3)?, + }) + }, + ) + .optional()? + .unwrap_or_default()) + } + pub fn sync_retry_clear(&mut self, binding: &str) -> Result<()> { + self.check_binding(binding)?; + self.db + .execute("DELETE FROM sync_retry WHERE binding=?1", [binding])?; + Ok(()) + } + pub fn sync_retry_fail( + &mut self, + binding: &str, + code: &str, + status: u16, + retry_after: Option, + now: i64, + ) -> Result<()> { + let previous = self.sync_retry(binding)?; + if code == "SYNC_CANCELLED" { + return Ok(()); + } + // Only retain a bounded machine code, never an arbitrary remote response string. + let code = if !code.is_empty() + && code.len() <= 80 + && code + .bytes() + .all(|b| b.is_ascii_uppercase() || b.is_ascii_digit() || b == b'_') + { + code + } else { + "SYNC_REMOTE_ERROR" + }; + let failures = if code == "CREDENTIALS_LOCKED" { + 0 + } else { + previous.failures.saturating_add(1) + }; + let halted = matches!(status, 401 | 403 | 413 | 426 | 507) + || matches!(code, "PROTOCOL_INCOMPATIBLE" | "SYNC_LOGIN_REQUIRED"); + let retry_at = now.saturating_add( + retry_after + .unwrap_or(2u64.saturating_pow(failures.min(8))) + .clamp(1, 3600) as i64, + ); + self.db.execute("INSERT INTO sync_retry VALUES (?1,?2,?3,?4,?5) ON CONFLICT(binding) DO UPDATE SET error=excluded.error,failures=excluded.failures,retry_at=excluded.retry_at,halted=excluded.halted", params![binding,code,failures,retry_at,halted])?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn retry_history_survives_twenty_reopens_and_isolates_bindings() { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(dir.path()).unwrap(); + let b = ws + .sync_bind_empty("https://sync.example", "remote", "account") + .unwrap(); + for attempt in 1..=20 { + ws.sync_retry_fail(&b.id, "NETWORK_ERROR", 0, Some(120), 1000) + .unwrap(); + drop(ws); + ws = Workspace::open(dir.path()).unwrap(); + let state = ws.sync_retry(&b.id).unwrap(); + assert_eq!(state.failures, attempt); + assert_eq!(state.remaining(1010), 110); + assert_eq!(state.remaining(-10000), 3600); + assert_eq!(state.remaining(1200), 0); + } + ws.sync_retry_fail(&b.id, "SYNC_CANCELLED", 0, None, 1100) + .unwrap(); + assert_eq!(ws.sync_retry(&b.id).unwrap().failures, 20); + ws.sync_retry_fail(&b.id, "UNAUTHORIZED", 401, None, 1100) + .unwrap(); + drop(ws); + ws = Workspace::open(dir.path()).unwrap(); + assert!(ws.sync_retry(&b.id).unwrap().halted); + ws.sync_retry_clear(&b.id).unwrap(); + assert!(ws.sync_retry(&b.id).unwrap().error.is_none()); + ws.sync_retry_fail(&b.id, "secret response body", 500, None, 1100) + .unwrap(); + assert_eq!( + ws.sync_retry(&b.id).unwrap().error.as_deref(), + Some("SYNC_REMOTE_ERROR") + ); + ws.sync_unbind(&b.id).unwrap(); + let next = ws + .sync_bind_empty("https://sync.example", "other", "account") + .unwrap(); + assert!(ws.sync_retry(&next.id).unwrap().error.is_none()); + assert!(ws.sync_retry_clear(&b.id).is_err()); + } +} diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index 01bbac0..3a74a4e 100644 --- a/frontend/src-tauri/src/workspace.rs +++ b/frontend/src-tauri/src/workspace.rs @@ -135,10 +135,10 @@ impl Workspace { let db = Connection::open(db_path)?; db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?; let version: i64 = db.query_row("PRAGMA user_version", [], |r| r.get(0))?; - if version > 8 { + if version > 9 { return Err(HostError::new("SCHEMA_INCOMPATIBLE")); } - if (1..8).contains(&version) { + if (1..9).contains(&version) { // Independent, complete SQLite backup before the schema ownership change. let backup = managed.join(format!("host-schema{version}-{}.sqlite3", Uuid::new_v4())); db.execute("VACUUM INTO ?1", [backup.to_string_lossy().as_ref()])?; @@ -162,6 +162,7 @@ impl Workspace { CREATE TABLE IF NOT EXISTS sync_initial_items (binding TEXT NOT NULL,sequence INTEGER NOT NULL,revision TEXT NOT NULL,PRIMARY KEY(binding,sequence)); CREATE TABLE IF NOT EXISTS payloads (operation_id TEXT PRIMARY KEY,hash TEXT NOT NULL,size INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS sync_observed (file_id TEXT PRIMARY KEY,path TEXT NOT NULL,hash TEXT NOT NULL,deleted INTEGER NOT NULL); + CREATE TABLE IF NOT EXISTS sync_retry (binding TEXT PRIMARY KEY,error TEXT,failures INTEGER NOT NULL,retry_at INTEGER,halted INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS sync_preferences (binding TEXT PRIMARY KEY,paused INTEGER NOT NULL DEFAULT 0); CREATE TABLE IF NOT EXISTS sync_resolutions (binding TEXT NOT NULL,sequence INTEGER NOT NULL,choice TEXT NOT NULL,destination TEXT NOT NULL,expected TEXT NOT NULL,operation_id TEXT NOT NULL,rename_id TEXT NOT NULL,copy_id TEXT NOT NULL,state TEXT NOT NULL,PRIMARY KEY(binding,sequence));")?; let has_origin: bool = db.query_row( @@ -178,7 +179,7 @@ impl Workspace { if version < 7 { db.execute_batch("INSERT OR IGNORE INTO sync_observed SELECT f.id,COALESCE((SELECT o.path FROM outbox o WHERE o.file_id=f.id AND o.state IN ('pending','queued') ORDER BY rowid DESC LIMIT 1),(SELECT h.path FROM sync_heads h JOIN sync_bindings b ON h.binding=b.id WHERE h.file_id=f.id AND b.state='active'),f.path),COALESCE((SELECT o.hash FROM outbox o WHERE o.file_id=f.id AND o.state IN ('pending','queued') ORDER BY rowid DESC LIMIT 1),(SELECT h.hash FROM sync_heads h JOIN sync_bindings b ON h.binding=b.id WHERE h.file_id=f.id AND b.state='active'),f.hash),f.deleted FROM files f;")?; } - db.execute_batch("PRAGMA user_version=8; COMMIT;")?; + db.execute_batch("PRAGMA user_version=9; COMMIT;")?; let vault_id: String = db .query_row("SELECT id FROM identity", [], |r| r.get(0)) .optional()? diff --git a/frontend/src/features/settings/SyncSettings.spec.ts b/frontend/src/features/settings/SyncSettings.spec.ts index 8a9b6aa..a957004 100644 --- a/frontend/src/features/settings/SyncSettings.spec.ts +++ b/frontend/src/features/settings/SyncSettings.spec.ts @@ -62,3 +62,12 @@ it('requires a reviewed merge fingerprint and invalidates preview when the remot expect(hostInvoke).toHaveBeenCalledWith('sync_bind', { request: { vault_id: 'local', endpoint: 'https://test.example/', account: 'test', remote_vault: 'one', mode: 'merge', fingerprint: 'reviewed-snapshot' } }) wrapper.unmount() }) + +it('shows a persisted halt without suggesting an automatic retry countdown', async () => { + vi.mocked(hostInvoke).mockResolvedValue({ ...empty(), binding: { id: 'binding', endpoint: 'https://test.example/', account: 'test', remote_vault: 'remote', cursor: 7 }, halted: true, failures: 3, error: 'UNAUTHORIZED', retry_in: 120 }) + const wrapper = mount(SyncSettings); await flushPromises() + expect(wrapper.text()).toContain('自动同步已停止') + expect(wrapper.text()).toContain('UNAUTHORIZED') + expect(wrapper.text()).not.toContain('120s') + wrapper.unmount() +}) diff --git a/frontend/src/features/settings/SyncSettings.vue b/frontend/src/features/settings/SyncSettings.vue index 30fc5b6..8c8da0d 100644 --- a/frontend/src/features/settings/SyncSettings.vue +++ b/frontend/src/features/settings/SyncSettings.vue @@ -8,7 +8,7 @@ import { t } from '@/i18n' const { actionDialog, resolveAction, askConfirm } = useActionDialog() interface Binding { id: string; endpoint: string; account: string; remote_vault: string; cursor: number } interface Conflict { sequence: number; local_path: string; local_hash: string; current_hash?: string; current_path?: string; remote: { path: string; operation: string } } -interface Status { vault_id: string; binding: Binding | null; paused: boolean; pending: number; conflicts: Conflict[]; credential_state: string; running: boolean; error: string | null; retry_in: number | null } +interface Status { vault_id: string; binding: Binding | null; paused: boolean; pending: number; conflicts: Conflict[]; credential_state: string; running: boolean; error: string | null; retry_in: number | null; failures: number; halted: boolean } interface RemoteVault { id: string; name: string; sequence: number; used: number; quota: number } const status = ref(null) const endpoint = ref('https://'), account = ref(''), password = ref(''), device = ref('OpenNexus Desktop'), testHttp = ref(false) @@ -120,9 +120,9 @@ onUnmounted(() => { mounted = false; clearInterval(timer); password.value = '' }

{{ status.binding.endpoint }} · {{ status.binding.account }} · {{ status.binding.remote_vault }}

-

{{ status.paused ? t('已暂停', 'Paused') : status.running ? t('同步中', 'Syncing') : t('等待下一轮同步', 'Waiting for next sync') }} · {{ t('待上传', 'Pending') }} {{ status.pending }} · cursor {{ status.binding.cursor }}

+

{{ status.paused ? t('已暂停', 'Paused') : status.running ? t('同步中', 'Syncing') : status.halted ? t('自动同步已停止，请处理错误后重试', 'Automatic sync stopped; resolve the error and retry') : t('等待下一轮同步', 'Waiting for next sync') }} · {{ t('待上传', 'Pending') }} {{ status.pending }} · cursor {{ status.binding.cursor }}

{{ status.credential_state }}

-

{{ status.error }} · {{ status.retry_in }}s

+

{{ status.error }} · {{ status.retry_in }}s