From 5a6fa86d109d29f3b64455b7261a3e6b98fe675c Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Tue, 8 Sep 2026 20:10:26 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20=E6=81=A2=E5=A4=8D=E8=BF=87?= =?UTF-8?q?=E6=9C=9F=E4=BC=9A=E8=AF=9D=E5=B9=B6=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E5=B0=9D=E8=AF=95=E7=BB=93=E6=9E=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../OpenNexus生产化实施进度-2026-09-08.md | 12 +- frontend/src-tauri/src/sync_auth.rs | 21 ++++ frontend/src-tauri/src/sync_client.rs | 90 +++++++++++++-- frontend/src-tauri/src/sync_commands.rs | 28 ++--- frontend/src-tauri/src/sync_retry.rs | 105 ++++++++++++++++-- frontend/src-tauri/src/sync_state.rs | 2 +- frontend/src-tauri/src/workspace.rs | 7 +- frontend/src-tauri/tests/sync_push.rs | 85 ++++++++++++++ .../features/settings/SyncSettings.spec.ts | 10 ++ .../src/features/settings/SyncSettings.vue | 6 +- 10 files changed, 325 insertions(+), 41 deletions(-) diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 5198a41..a4f02d9 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -6,7 +6,7 @@ ## 当前状态摘要 -截至 `7b69537`，Host Sync capability 已开启，扩展 capability 仍关闭。笔记与 Task CRUD 已接入当前 Rust Vault；同步具备上传/拉取、冲突解决、非空初始合并、100 MiB 续传、主题和编辑器偏好、持久退避。Workspace 当前 schema 9。凭据备份/恢复及 WTS 撤销已实现，但实机锁屏验收仍缺。 +截至 `7b69537`，Host Sync capability 已开启，扩展 capability 仍关闭。笔记与 Task CRUD 已接入当前 Rust Vault；同步具备上传/拉取、冲突解决、非空初始合并、100 MiB 续传、主题和编辑器偏好、持久退避。Workspace 当前 schema 10（上传尝试记录增量后）。凭据备份/恢复及 WTS 撤销已实现，但实机锁屏验收仍缺。 最近完整测试：后端 900 项（Task 增量后）、前端 517 项、Rust desktop 52 项；后两者在重试持久化增量后复跑。以下各节为历史增量，出现的旧 capability、旧 schema 和旧测试数量描述对应当时状态，以本摘要及后续增量为准。OS 沙箱、Rust 扩展管理、剩余逻辑数据类别、完整故障/性能/发布验收仍未完成。 @@ -129,3 +129,13 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写 - 使用锁定 packaging 环境重新生成 PyInstaller onedir，清单包含 1523 个文件。构建日志 `.build/core-build-production.log`，产物 `.build/sidecar/dist/opennexus-core` 与 `.build/sidecar/manifest.json`。 - 新增显式打包测试 `cargo test --locked --features desktop --test core_bundle packaged_core_twenty -- --ignored`，调用实际可执行文件并在每次启动前校验完整文件清单。20 个独立临时数据目录、20 个不同会话代际、带鉴权和代际头的 HTTP 健康请求全部成功，逐次退出后端口关闭。测试耗时 95.64 秒，日志 `.build/core-packaged-cold-starts.log`。初版测试遗漏代际请求头导致拒绝，修正测试后通过。 - 该测试按要求先构建后显式运行，普通源码全目标测试会跳过此入口。Clippy `-D warnings` 通过。此证据来自当前 Windows 开发机，不能替代干净 VM、MSVC 签名安装包、两台真实设备或完整 A-01 验收。 + + +## 增量：401 恢复与每作业持久尝试记录 + +- 协调器对可重放的同步轮次收到 401 时仅强制刷新一次；刷新仍校验设备身份且受凭据锁撤销。刷新失败或第二次 401 返回停止错误，403 不刷新。普通非幂等业务创建没有套用重放包装。 +- 真实本地 HTTP 测试先在隔离数据库中提前过期 Access Token，确认读取经一次刷新恢复；删除测试会话后 Refresh 也被拒绝，不重新登录、不注册新设备。另注入连续 401 和 403，分别确认最多两次与一次操作调用。 +- Workspace schema 10 保存每个上传作业的累计尝试次数与 running/failed/interrupted/succeeded 结果；普通取消通过析构记录中断，进程强杀则由重开修复。文件已确认而结果未记录时，恢复以已持久的 ack 为准。错误只保存限定机器码；待处理作业最多展示 20 项恢复记录。 +- 20 次重开验证尝试累加且冻结的提交载荷不变；取消测试确认无需重启即显示 interrupted。真实 100 MiB 十次强杀后，重开查询准确显示 10 次尝试与 interrupted，最终继续上传、下载与摘要校验通过。测试过程使用独立临时 Vault 和服务账号。 +- Rust desktop 全目标 53 项通过，之后新增取消测试单独通过；更新后的真实 HTTP 集成再跑通过。两个 ignored 入口分别是已在前轮显式运行的打包测试和父测试实际驱动的强杀辅助进程。前端 SyncSettings 5 项、两项目 TypeScript 检查及最终 Clippy `-D warnings` 通过。日志 `.build/job-recovery-rust-tests.log`、`.build/job-recovery-http-tests.log`；本增量无 Python 变更。 +- 此增量不等于完整 S-02/S-03 故障矩阵，pull 各边界 20 次强杀、同目标 rename 与历史恢复等完整验收仍需推进。 diff --git a/frontend/src-tauri/src/sync_auth.rs b/frontend/src-tauri/src/sync_auth.rs index 4c2e2bf..b51672a 100644 --- a/frontend/src-tauri/src/sync_auth.rs +++ b/frontend/src-tauri/src/sync_auth.rs @@ -158,6 +158,27 @@ pub async fn client( saved.allow_test_http, ) } +/// The coordinator must serialize calls. Only use for read-only or durably idempotent work: +/// a 401 repeats the operation once with the same device after rotating its session. +pub async fn authenticated( + credentials: &Credentials, + endpoint: &str, + account: &str, + mut action: F, +) -> Result +where + F: FnMut(SyncClient) -> Fut, + Fut: Future>, +{ + let first = client(credentials, endpoint, account, false).await?; + match guarded(credentials, action(first)).await { + Err(error) if error.status == 401 => { + let refreshed = client(credentials, endpoint, account, true).await?; + guarded(credentials, action(refreshed)).await + } + result => result, + } +} pub async fn logout(credentials: &Credentials, endpoint: &str, account: &str) -> Result<()> { let client = client(credentials, endpoint, account, false).await?; // A failed server revocation is reported; the encrypted record remains available for retry. diff --git a/frontend/src-tauri/src/sync_client.rs b/frontend/src-tauri/src/sync_client.rs index 85beb46..13b0ec2 100644 --- a/frontend/src-tauri/src/sync_client.rs +++ b/frontend/src-tauri/src/sync_client.rs @@ -30,6 +30,20 @@ impl SyncError { } } type Result = std::result::Result; +struct Attempt<'a, W: WorkspaceAccess> { + workspace: &'a W, + job: &'a Job, + finished: bool, +} +impl Drop for Attempt<'_, W> { + fn drop(&mut self) { + if !self.finished { + let _ = self + .workspace + .access(|ws| ws.sync_attempt_interrupt(self.job)); + } + } +} pub trait WorkspaceAccess: Send + Sync { fn access( &self, @@ -309,19 +323,36 @@ impl SyncClient { if job.state == "conflict" { return Err(SyncError::new("REVISION_CONFLICT")); } - if job.operation == "put" && job.base_revision.is_none() { - self.upload(workspace, binding, &job).await?; + workspace.access(|ws| ws.sync_attempt_start(&job))?; + let mut attempt = Attempt { + workspace, + job: &job, + finished: false, + }; + let result = async { + if job.operation == "put" && job.base_revision.is_none() { + self.upload(workspace, binding, &job).await?; + } + let payload = workspace.access(|ws| ws.sync_commit_payload(&job))?; + let revision = self + .json( + Method::POST, + &format!("sync/v1/vaults/{}/revisions", binding.remote_vault), + Some(payload), + ) + .await?; + workspace.access(|ws| ws.sync_ack(&job, &revision))?; + Ok(true) } - let payload = workspace.access(|ws| ws.sync_commit_payload(&job))?; - let revision = self - .json( - Method::POST, - &format!("sync/v1/vaults/{}/revisions", binding.remote_vault), - Some(payload), + .await; + workspace.access(|ws| { + ws.sync_attempt_finish( + &job, + result.as_ref().err().map(|e: &SyncError| e.code.as_str()), ) - .await?; - workspace.access(|ws| ws.sync_ack(&job, &revision))?; - Ok(true) + })?; + attempt.finished = true; + result } pub async fn pull_page( &self, @@ -577,3 +608,40 @@ fn identifier(value: &str) -> Result<()> { } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + #[tokio::test] + async fn dropping_a_pending_attempt_marks_interruption_without_restart() { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(dir.path()).unwrap(); + ws.write("cancel.md", "", b"retained", "local").unwrap(); + let binding = ws + .sync_bind_empty("https://sync.example", "remote", "account") + .unwrap(); + let job = ws.sync_next(&binding.id).unwrap().unwrap(); + let workspace = Arc::new(Mutex::new(ws)); + let future = async { + workspace.access(|ws| ws.sync_attempt_start(&job)).unwrap(); + let _attempt = Attempt { + workspace: &workspace, + job: &job, + finished: false, + }; + std::future::pending::<()>().await; + }; + assert!(tokio::time::timeout(Duration::from_millis(10), future) + .await + .is_err()); + let rows = workspace + .access(|ws| ws.sync_attempts(&binding.id)) + .unwrap(); + assert_eq!(rows[0]["outcome"], "interrupted"); + assert_eq!(rows[0]["attempts"], 1); + assert_eq!( + workspace.access(|ws| ws.read("cancel.md")).unwrap().content, + "retained" + ); + } +} diff --git a/frontend/src-tauri/src/sync_commands.rs b/frontend/src-tauri/src/sync_commands.rs index 1343ba6..c4ca07b 100644 --- a/frontend/src-tauri/src/sync_commands.rs +++ b/frontend/src-tauri/src/sync_commands.rs @@ -268,6 +268,11 @@ pub fn sync_status(host: State<'_, Host>) -> Result { paused, conflicts, ws.pending_count()?, + binding + .as_ref() + .map(|b| ws.sync_attempts(&b.id)) + .transpose()? + .unwrap_or_default(), binding .as_ref() .map(|b| ws.sync_retry(&b.id)) @@ -275,7 +280,7 @@ pub fn sync_status(host: State<'_, Host>) -> Result { .unwrap_or_default(), )) })?; - let (vault_id, binding, paused, conflicts, pending, retry) = snapshot; + let (vault_id, binding, paused, conflicts, pending, attempts, retry) = snapshot; let credential_state = if let Some(b) = &binding { sync_auth::available(&host.credentials, &b.endpoint, &b.account) .map(|exists| { @@ -293,7 +298,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":retry.error,"retry_in":retry.retry_at.map(|_| retry.remaining(now())),"failures":retry.failures,"halted":retry.halted}), + 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,"attempts":attempts}), ) } #[tauri::command] @@ -387,15 +392,11 @@ fn now() -> i64 { async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> { let epoch = host.sync.epoch.load(Ordering::SeqCst); - let work = async { - let client = sync_auth::client( - &host.credentials, - &binding.endpoint, - &binding.account, - false, - ) - .await?; - let work = async { + let work = sync_auth::authenticated( + &host.credentials, + &binding.endpoint, + &binding.account, + |client| async move { client.handshake().await?; host.workspace.access(|ws| ws.sync_discover(&binding.id))?; for _ in 0..10 { @@ -418,9 +419,8 @@ async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> { } client.pull_page(&host.workspace, binding).await?; Ok(()) - }; - sync_auth::guarded(&host.credentials, work).await - }; + }, + ); tokio::pin!(work); let mut tick = tokio::time::interval(Duration::from_millis(50)); loop { diff --git a/frontend/src-tauri/src/sync_retry.rs b/frontend/src-tauri/src/sync_retry.rs index b6be3ef..b6b83fd 100644 --- a/frontend/src-tauri/src/sync_retry.rs +++ b/frontend/src-tauri/src/sync_retry.rs @@ -17,7 +17,57 @@ impl Retry { .unwrap_or(0) } } +fn machine_code(code: &str) -> &str { + 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" + } +} impl Workspace { + pub fn sync_attempt_start(&self, job: &crate::sync_state::Job) -> Result<()> { + self.check_job(job)?; + self.db.execute("INSERT INTO sync_attempts VALUES (?1,?2,1,'running',NULL) ON CONFLICT(binding,operation_id) DO UPDATE SET attempts=MIN(attempts+1,2147483647),outcome='running',error=NULL", params![job.binding, job.operation_id])?; + Ok(()) + } + pub fn sync_attempt_interrupt(&self, job: &crate::sync_state::Job) -> Result<()> { + self.check_job(job)?; + self.db.execute("UPDATE sync_attempts SET outcome=CASE WHEN EXISTS(SELECT 1 FROM sync_jobs j WHERE j.binding=?1 AND j.operation_id=?2 AND j.state='acked') THEN 'succeeded' ELSE 'interrupted' END WHERE binding=?1 AND operation_id=?2 AND outcome='running'", params![job.binding,job.operation_id])?; + Ok(()) + } + pub fn sync_attempt_finish( + &self, + job: &crate::sync_state::Job, + error: Option<&str>, + ) -> Result<()> { + self.check_job(job)?; + self.db.execute( + "UPDATE sync_attempts SET outcome=?3,error=?4 WHERE binding=?1 AND operation_id=?2", + params![ + job.binding, + job.operation_id, + if error.is_some() { + "failed" + } else { + "succeeded" + }, + error.map(machine_code) + ], + )?; + Ok(()) + } + pub fn sync_attempts(&self, binding: &str) -> Result> { + self.check_binding(binding)?; + let mut statement = self.db.prepare("SELECT a.operation_id,j.path,a.attempts,a.outcome,a.error,j.state FROM sync_attempts a JOIN sync_jobs j ON j.binding=a.binding AND j.operation_id=a.operation_id WHERE a.binding=?1 AND j.state NOT IN ('acked','archived') ORDER BY a.rowid LIMIT 20")?; + let rows = statement.query_map([binding], |row| Ok(serde_json::json!({"operation_id":row.get::<_,String>(0)?,"path":row.get::<_,String>(1)?,"attempts":row.get::<_,i64>(2)?,"outcome":row.get::<_,String>(3)?,"error":row.get::<_,Option>(4)?,"state":row.get::<_,String>(5)?})))?; + Ok(rows.collect::, _>>()?) + } + pub fn sync_retry(&self, binding: &str) -> Result { self.check_binding(binding)?; Ok(self @@ -56,16 +106,7 @@ impl Workspace { 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 code = machine_code(code); let failures = if code == "CREDENTIALS_LOCKED" { 0 } else { @@ -87,6 +128,50 @@ impl Workspace { mod tests { use super::*; #[test] + fn interrupted_job_attempts_survive_twenty_restarts_without_changing_commit_base() { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(dir.path()).unwrap(); + ws.write("attempt.md", "", b"body", "local").unwrap(); + let binding = ws + .sync_bind_empty("https://sync.example", "remote", "account") + .unwrap(); + let job = ws.sync_next(&binding.id).unwrap().unwrap(); + let payload = ws.sync_commit_payload(&job).unwrap(); + for count in 1..=20 { + ws.sync_attempt_start(&job).unwrap(); + drop(ws); + ws = Workspace::open(dir.path()).unwrap(); + let rows = ws.sync_attempts(&binding.id).unwrap(); + assert_eq!(rows[0]["attempts"], count); + assert_eq!(rows[0]["outcome"], "interrupted"); + assert_eq!(ws.sync_commit_payload(&job).unwrap(), payload); + } + ws.sync_attempt_finish(&job, Some("response with secret body")) + .unwrap(); + assert_eq!( + ws.sync_attempts(&binding.id).unwrap()[0]["error"], + "SYNC_REMOTE_ERROR" + ); + ws.sync_attempt_start(&job).unwrap(); + let mut revision = payload.clone(); + revision["hash"] = payload["content_hash"].clone(); + revision["vault_id"] = serde_json::json!("remote"); + revision["sequence"] = serde_json::json!(1); + ws.sync_ack(&job, &revision).unwrap(); + drop(ws); + ws = Workspace::open(dir.path()).unwrap(); + assert!(ws.sync_attempts(&binding.id).unwrap().is_empty()); + assert_eq!( + ws.db + .query_row("SELECT outcome FROM sync_attempts", [], |r| r + .get::<_, String>(0)) + .unwrap(), + "succeeded" + ); + ws.sync_unbind(&binding.id).unwrap(); + assert!(ws.sync_attempt_finish(&job, None).is_err()); + } + #[test] fn retry_history_survives_twenty_reopens_and_isolates_bindings() { let dir = tempfile::tempdir().unwrap(); let mut ws = Workspace::open(dir.path()).unwrap(); diff --git a/frontend/src-tauri/src/sync_state.rs b/frontend/src-tauri/src/sync_state.rs index f527d6d..685a050 100644 --- a/frontend/src-tauri/src/sync_state.rs +++ b/frontend/src-tauri/src/sync_state.rs @@ -197,7 +197,7 @@ impl Workspace { } Ok(job) } - fn check_job(&self, job: &Job) -> Result<()> { + pub(crate) fn check_job(&self, job: &Job) -> Result<()> { self.check_binding(&job.binding)?; let state: String = self.db.query_row( "SELECT state FROM sync_jobs WHERE binding=?1 AND operation_id=?2", diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index 3a74a4e..dcf574c 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 > 9 { + if version > 10 { return Err(HostError::new("SCHEMA_INCOMPATIBLE")); } - if (1..9).contains(&version) { + if (1..10).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_attempts (binding TEXT NOT NULL,operation_id TEXT NOT NULL,attempts INTEGER NOT NULL,outcome TEXT NOT NULL,error TEXT,PRIMARY KEY(binding,operation_id)); 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));")?; @@ -179,7 +180,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=9; COMMIT;")?; + db.execute_batch("UPDATE sync_attempts SET outcome=CASE WHEN EXISTS(SELECT 1 FROM sync_jobs j WHERE j.binding=sync_attempts.binding AND j.operation_id=sync_attempts.operation_id AND j.state='acked') THEN 'succeeded' ELSE 'interrupted' END WHERE outcome='running'; PRAGMA user_version=10; COMMIT;")?; let vault_id: String = db .query_row("SELECT id FROM identity", [], |r| r.get(0)) .optional()? diff --git a/frontend/src-tauri/tests/sync_push.rs b/frontend/src-tauri/tests/sync_push.rs index d80ff95..d4575a1 100644 --- a/frontend/src-tauri/tests/sync_push.rs +++ b/frontend/src-tauri/tests/sync_push.rs @@ -618,6 +618,14 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate } std::fs::remove_file(root.path().join("interrupt-upload")).unwrap(); let large_workspace = Arc::new(Mutex::new(Workspace::open(large_root.path()).unwrap())); + let interrupted = large_workspace + .lock() + .unwrap() + .sync_attempts(&large_binding.id) + .unwrap(); + assert_eq!(interrupted.len(), 1); + assert_eq!(interrupted[0]["attempts"], 10); + assert_eq!(interrupted[0]["outcome"], "interrupted"); assert!(client .push_one(&large_workspace, &large_binding) .await @@ -707,6 +715,59 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate .unwrap() .unlock(Zeroizing::new(b"fixture-stronghold-password".to_vec())) .unwrap(); + // Expire the access token early in this isolated fixture; the refresh token stays valid. + let database = rusqlite::Connection::open(root.path().join("sync.sqlite3")).unwrap(); + database + .execute("UPDATE sessions SET expires=0", []) + .unwrap(); + let attempts = std::sync::atomic::AtomicUsize::new(0); + let recovered = sync_auth::authenticated(&credentials, &canonical, "rust-fixture", |client| { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async move { + client + .json(reqwest::Method::GET, "sync/v1/vaults", None) + .await + } + }) + .await + .unwrap(); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2); + assert!(recovered["items"] + .as_array() + .unwrap() + .iter() + .any(|v| v["id"] == remote)); + // Even a repeated 401 must stop after one rotation, rather than refresh indefinitely. + attempts.store(0, std::sync::atomic::Ordering::SeqCst); + let denied = sync_auth::authenticated(&credentials, &canonical, "rust-fixture", |_client| { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async { + Err::<(), _>(notesagent_host::sync_client::SyncError { + code: "UNAUTHORIZED".into(), + status: 401, + retry_after: None, + }) + } + }) + .await + .unwrap_err(); + assert_eq!(denied.status, 401); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2); + attempts.store(0, std::sync::atomic::Ordering::SeqCst); + let denied = sync_auth::authenticated(&credentials, &canonical, "rust-fixture", |_client| { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async { + Err::<(), _>(notesagent_host::sync_client::SyncError { + code: "FORBIDDEN".into(), + status: 403, + retry_after: None, + }) + } + }) + .await + .unwrap_err(); + assert_eq!(denied.status, 403); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1); let restored = sync_auth::client(&credentials, &canonical, "rust-fixture", false) .await .unwrap(); @@ -723,6 +784,30 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate .status, 401 ); + sync_auth::login( + &credentials, + &endpoint, + "rust-fixture", + Zeroizing::new("controlled-fixture-password".into()), + "Revoked Host", + true, + ) + .await + .unwrap(); + database.execute("DELETE FROM sessions", []).unwrap(); + attempts.store(0, std::sync::atomic::Ordering::SeqCst); + let revoked = sync_auth::authenticated(&credentials, &canonical, "rust-fixture", |client| { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async move { + client + .json(reqwest::Method::GET, "sync/v1/vaults", None) + .await + } + }) + .await + .unwrap_err(); + assert_eq!(revoked.status, 401); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1); } #[tokio::test] diff --git a/frontend/src/features/settings/SyncSettings.spec.ts b/frontend/src/features/settings/SyncSettings.spec.ts index a957004..f33bc41 100644 --- a/frontend/src/features/settings/SyncSettings.spec.ts +++ b/frontend/src/features/settings/SyncSettings.spec.ts @@ -71,3 +71,13 @@ it('shows a persisted halt without suggesting an automatic retry countdown', asy expect(wrapper.text()).not.toContain('120s') wrapper.unmount() }) + +it('shows persisted per-job interruption counts without starting work from the view', async () => { + vi.mocked(hostInvoke).mockResolvedValue({ ...empty(), binding: { id: 'binding', endpoint: 'https://test.example/', account: 'test', remote_vault: 'remote', cursor: 7 }, attempts: [{ operation_id: 'op', path: 'attachments/large.pdf', attempts: 11, outcome: 'interrupted', error: null }] }) + const wrapper = mount(SyncSettings); await flushPromises() + expect(wrapper.text()).toContain('attachments/large.pdf') + expect(wrapper.text()).toContain('尝试次数 11') + expect(wrapper.text()).toContain('从已确认位置恢复') + expect(vi.mocked(hostInvoke).mock.calls.every(([command]) => command === 'sync_status')).toBe(true) + wrapper.unmount() +}) diff --git a/frontend/src/features/settings/SyncSettings.vue b/frontend/src/features/settings/SyncSettings.vue index 8c8da0d..2059d83 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; failures: number; halted: boolean } +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; attempts?: Array<{ operation_id: string; path: string; attempts: number; outcome: string; error: string | null }> } 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) @@ -129,6 +129,10 @@ onUnmounted(() => { mounted = false; clearInterval(timer); password.value = '' } +
+ {{ t('上传作业恢复记录（最多 20 项）', 'Upload recovery records (up to 20)') }} +

{{ job.path }} · {{ t('尝试次数', 'Attempts') }} {{ job.attempts }} · {{ job.outcome === 'interrupted' ? t('上次上传已中断，将从已确认位置恢复', 'Previous upload interrupted; resumes from the confirmed offset') : job.outcome === 'failed' ? t('上次尝试失败', 'Last attempt failed') : t('上传处理中', 'Upload in progress') }} · {{ job.error }}

+

{{ conflict.local_path }}

{{ t('远端版本', 'Remote revision') }} {{ conflict.sequence }} · {{ conflict.remote.operation }} · {{ conflict.remote.path }}