From 24938c9d7821d5c6deeacc68b407648a3137dee7 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Wed, 9 Sep 2026 11:33:06 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20=E5=AE=8C=E6=88=90=20S-03=20?= =?UTF-8?q?=E5=86=B2=E7=AA=81=E6=94=B6=E6=95=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/development/OpenNexus生产验收Runner.md | 2 +- frontend/src-tauri/src/sync_inbox.rs | 16 +- frontend/src-tauri/src/sync_resolution.rs | 289 +++++++++-- frontend/src-tauri/src/sync_scope.rs | 2 +- frontend/src-tauri/src/workspace.rs | 80 ++- frontend/src-tauri/tests/sync_conflicts.rs | 508 ++++++++++++++++++++ scripts/acceptance_cases/s03_sync_client.py | 136 ++++++ scripts/phase3_acceptance.py | 5 + 8 files changed, 988 insertions(+), 50 deletions(-) create mode 100644 frontend/src-tauri/tests/sync_conflicts.rs create mode 100644 scripts/acceptance_cases/s03_sync_client.py diff --git a/docs/development/OpenNexus生产验收Runner.md b/docs/development/OpenNexus生产验收Runner.md index 78c6dcd..2b4adf9 100644 --- a/docs/development/OpenNexus生产验收Runner.md +++ b/docs/development/OpenNexus生产验收Runner.md @@ -24,4 +24,4 @@ python scripts/phase3-production-acceptance.py ` 报告目录包含 `summary.json`、`case-manifest.json`、`junit.xml`、`cases/.json` 和脱敏的 `logs/.log`。摘要记录 commit、各锁文件 SHA-256、配置摘要与已提供安装产物摘要。日志将仓库、数据根、报告根、用户主目录和配置声明的秘密值替换为占位符，并限制为 10 MiB。报告目录必须为空，避免单例复跑覆盖原始证据。 -当前 runner 与失败闭合行为已实现，A-02/A-03 Sidecar、B-01/B-02 凭据、D-01 扩展包以及 S-01/S-02 Sync 客户端 driver 已登记；其余 23 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。 +当前 runner 与失败闭合行为已实现，A-02/A-03 Sidecar、B-01/B-02 凭据、D-01 扩展包以及 S-01/S-02/S-03 Sync 客户端 driver 已登记；其余 22 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。 diff --git a/frontend/src-tauri/src/sync_inbox.rs b/frontend/src-tauri/src/sync_inbox.rs index d2bf0e3..9a755a7 100644 --- a/frontend/src-tauri/src/sync_inbox.rs +++ b/frontend/src-tauri/src/sync_inbox.rs @@ -293,7 +293,8 @@ impl Workspace { return Ok(true); } let local_path = self.path_for_id(&revision.file_id).ok(); - let path = local_path.as_deref().unwrap_or(&revision.path); + let conflict_path = self.sync_conflict_local_path(&revision)?; + let path = conflict_path.as_str(); let local = self.resolve(path)?; let current = if local.is_file() { crate::payloads::hash_file(&local)? @@ -309,10 +310,14 @@ impl Workspace { |r| r.get(0), ) .optional()?; + let target_collision = local_path + .as_deref() + .is_some_and(|previous| previous != revision.path) + && self.resolve(&revision.path)?.exists(); let conflict = queued || (local_path.is_none() && local.exists()) || (local_path.is_some() && head.as_deref() != Some(current.as_str())) - || (path != revision.path && self.resolve(&revision.path)?.exists()); + || target_collision; if conflict { self.sync_preserve_conflict(binding, &revision, path)?; return Ok(true); @@ -366,11 +371,10 @@ impl Workspace { })? .collect::, _>>()?; rows.into_iter().map(|(sequence,file_id,local_path,local_hash,remote)| { - let remote: Value = serde_json::from_str(&remote).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?; - let current_path=self.path_for_id(&file_id).unwrap_or_else(|_|local_path.clone()); - let source=self.resolve(¤t_path)?; + let revision: RemoteRevision = serde_json::from_str(&remote).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?; + let source=self.resolve(&local_path)?; let current_hash=if source.is_file() {crate::payloads::hash_file(&source)?} else {String::new()}; - Ok(serde_json::json!({"sequence":sequence,"file_id":file_id,"local_path":local_path,"local_hash":local_hash,"current_path":current_path,"current_hash":current_hash,"remote":remote})) + Ok(serde_json::json!({"sequence":sequence,"file_id":file_id,"local_path":local_path,"local_hash":local_hash,"current_path":local_path,"current_hash":current_hash,"remote":revision})) }).collect() } } diff --git a/frontend/src-tauri/src/sync_resolution.rs b/frontend/src-tauri/src/sync_resolution.rs index 6d53502..e172856 100644 --- a/frontend/src-tauri/src/sync_resolution.rs +++ b/frontend/src-tauri/src/sync_resolution.rs @@ -9,6 +9,21 @@ use std::fs; use uuid::Uuid; impl Workspace { + pub(crate) fn sync_conflict_local_path(&self, revision: &RemoteRevision) -> Result { + let identity_path = self.path_for_id(&revision.file_id).ok(); + let destination_occupied = self + .entry(&revision.path)? + .is_some_and(|entry| !entry.deleted && entry.file_id != revision.file_id); + if destination_occupied + && identity_path + .as_deref() + .is_some_and(|path| path != revision.path) + { + return Ok(revision.path.clone()); + } + Ok(identity_path.unwrap_or_else(|| revision.path.clone())) + } + pub fn sync_resolve( &mut self, binding: &str, @@ -46,10 +61,8 @@ impl Workspace { return Err(HostError::new("SYNC_RESOLUTION_CHANGED")); } } else { - let (path,remote): (String,String) = self.db.query_row("SELECT local_path,remote FROM sync_conflicts WHERE binding=?1 AND sequence=?2 AND state='open'",params![binding,sequence],|r| Ok((r.get(0)?,r.get(1)?)))?; - let revision: RemoteRevision = serde_json::from_str(&remote) - .map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?; - let path = self.path_for_id(&revision.file_id).unwrap_or(path); + let stored_path: String = self.db.query_row("SELECT local_path FROM sync_conflicts WHERE binding=?1 AND sequence=?2 AND state='open'",params![binding,sequence],|r| r.get(0))?; + let path = stored_path; let source = self.resolve(&path)?; let current = if source.is_file() { self.sync_store_file(&source)? @@ -78,7 +91,7 @@ impl Workspace { return Err(HostError::new("SYNC_RESOLUTION_INVALID")); } self.db.execute( - "INSERT INTO sync_resolutions VALUES (?1,?2,?3,?4,?5,?6,?7,?8,'pending')", + "INSERT INTO sync_resolutions (binding,sequence,choice,destination,expected,operation_id,rename_id,copy_id,state,retire_id,restore_id) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,'pending',?9,?10)", params![ binding, sequence, @@ -87,6 +100,8 @@ impl Workspace { expected, Uuid::new_v4().to_string(), Uuid::new_v4().to_string(), + Uuid::new_v4().to_string(), + Uuid::new_v4().to_string(), Uuid::new_v4().to_string() ], )?; @@ -109,7 +124,22 @@ impl Workspace { } fn sync_apply_resolution(&mut self, binding: &str, sequence: i64) -> Result<()> { self.check_binding(binding)?; - let (choice,destination,expected,operation,rename,copy,state): (String,String,String,String,String,String,String) = self.db.query_row("SELECT choice,destination,expected,operation_id,rename_id,copy_id,state FROM sync_resolutions WHERE binding=?1 AND sequence=?2",params![binding,sequence],|r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?,r.get(6)?)))?; + let (choice,destination,expected,operation,rename,copy,state,mut retire,mut restore): (String,String,String,String,String,String,String,String,String) = self.db.query_row("SELECT choice,destination,expected,operation_id,rename_id,copy_id,state,retire_id,restore_id FROM sync_resolutions WHERE binding=?1 AND sequence=?2",params![binding,sequence],|r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?,r.get(6)?,r.get(7)?,r.get(8)?)))?; + if retire.is_empty() || restore.is_empty() { + retire = Uuid::new_v4().to_string(); + restore = Uuid::new_v4().to_string(); + self.db.execute( + "UPDATE sync_resolutions SET retire_id=CASE WHEN retire_id='' THEN ?3 ELSE retire_id END,restore_id=CASE WHEN restore_id='' THEN ?4 ELSE restore_id END WHERE binding=?1 AND sequence=?2", + params![binding, sequence, retire, restore], + )?; + let ids: (String, String) = self.db.query_row( + "SELECT retire_id,restore_id FROM sync_resolutions WHERE binding=?1 AND sequence=?2", + params![binding, sequence], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + retire = ids.0; + restore = ids.1; + } if state == "completed" { return Ok(()); } @@ -131,21 +161,26 @@ impl Workspace { if head != sequence { return Err(HostError::new("SYNC_CONFLICT_CHANGED")); } - let path = self.path_for_id(&revision.file_id).unwrap_or(stored_path); + // local_path is frozen when the conflict is recorded. Recomputing it from + // file identity after a partial resolution can select a different file. + let path = stored_path; if !self .operation(&operation)? .is_some_and(|value| value["state"] == "committed") { - let source = self.resolve(&path)?; - let mut current = if source.is_file() { - crate::payloads::hash_file(&source)? + let conflict_source = self.resolve(&path)?; + let current = if conflict_source.is_file() { + crate::payloads::hash_file(&conflict_source)? } else { String::new() }; - let retired = self.operation(&rename)?.is_some_and(|value| { + let target_retired = self.operation(&retire)?.is_some_and(|value| { value["state"] == "committed" && value["result"]["deleted"] == true }); - if current != if retired { "" } else { &expected } { + let source_renamed = self + .operation(&rename)? + .is_some_and(|value| value["state"] == "committed"); + if current != expected && !target_retired && !source_renamed { return Err(HostError::new("REVISION_CONFLICT")); } if choice == "copy" { @@ -159,21 +194,60 @@ impl Workspace { None, )?; } - if self - .entry(&path)? - .is_some_and(|entry| !entry.deleted && entry.file_id != revision.file_id) + let target_collision = target_retired + || (path == revision.path + && self + .entry(&path)? + .is_some_and(|entry| !entry.deleted && entry.file_id != revision.file_id)); + if target_collision && !target_retired { + self.mutate_with_origin("delete", &path, "", ¤t, &retire, "remote")?; + } + let source_path = self + .path_for_id(&revision.file_id) + .unwrap_or_else(|_| revision.path.clone()); + let source = self.resolve(&source_path)?; + let mut source_hash = if source.is_file() { + crate::payloads::hash_file(&source)? + } else { + String::new() + }; + if target_collision + && source_path != revision.path + && !source_hash.is_empty() + && !source_renamed { - self.mutate_with_origin("delete", &path, "", ¤t, &rename, "remote")?; - current.clear(); + // The deleted target still owns its unique database path until the + // final identity-aware write retires that tombstone. Retire the + // incoming identity at its old path, then resurrect it at the + // target with the frozen remote or chosen-local bytes. + self.mutate_with_origin( + "delete", + &source_path, + "", + &source_hash, + &rename, + "remote", + )?; + source_hash.clear(); } if choice == "local" { if expected.is_empty() { self.sync_delete_intent(&revision, &operation)?; } else { + let destination = if target_collision { + revision.path.as_str() + } else { + path.as_str() + }; + let destination_hash = if target_collision { + source_hash.as_str() + } else { + current.as_str() + }; let size = fs::metadata(self.sync_spool(&expected)?)?.len(); self.write_spooled_with_identity( - &path, - ¤t, + destination, + destination_hash, (&expected, size), "local", &operation, @@ -181,19 +255,31 @@ impl Workspace { )?; } } else if revision.operation == "delete" { - if !current.is_empty() { - self.mutate_with_origin("delete", &path, "", ¤t, &operation, "remote")?; + if !source_hash.is_empty() { + self.mutate_with_origin( + "delete", + &source_path, + "", + &source_hash, + &operation, + "remote", + )?; } } else { - if path != revision.path && !current.is_empty() { + if !target_collision + && source_path != revision.path + && !source_hash.is_empty() + && !source_renamed + { self.mutate_with_origin( "rename", - &path, + &source_path, &revision.path, - ¤t, + &source_hash, &rename, "remote", )?; + source_hash = crate::payloads::hash_file(&self.resolve(&revision.path)?)?; } let digest = revision .hash @@ -201,7 +287,7 @@ impl Workspace { .ok_or_else(|| HostError::new("SYNC_RESPONSE_INVALID"))?; self.write_spooled_with_identity( &revision.path, - ¤t, + &source_hash, (digest, revision.size as u64), "remote", &operation, @@ -209,21 +295,56 @@ impl Workspace { )?; } } - let retired = self.operation(&rename)?.and_then(|value| { + let retired = self.operation(&retire)?.and_then(|value| { (value["state"] == "committed" && value["result"]["deleted"] == true) .then(|| value["result"]["file_id"].as_str().map(str::to_owned)) .flatten() }); + let mut restored_retired = false; + if let Some(retired_id) = retired.as_deref().filter(|id| *id != revision.file_id) { + let remote_head: Option<(String, String)> = self + .db + .query_row( + "SELECT path,hash FROM sync_heads WHERE binding=?1 AND file_id=?2 AND hash!=''", + params![binding, retired_id], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional()?; + if let Some((head_path, head_hash)) = + remote_head.filter(|(head_path, _)| head_path != &revision.path) + { + if !self + .operation(&restore)? + .is_some_and(|value| value["state"] == "committed") + { + if self.resolve(&head_path)?.exists() { + return Err(HostError::new("REVISION_CONFLICT")); + } + let size = fs::metadata(self.sync_spool(&head_hash)?)?.len(); + self.write_spooled_with_identity( + &head_path, + "", + (&head_hash, size), + "remote", + &restore, + Some(retired_id), + )?; + } + restored_retired = true; + } + } let tx = self.db.transaction()?; if let Some(retired) = retired.filter(|id| id != &revision.file_id) { - tx.execute( - "UPDATE file_aliases SET file_id=?1 WHERE file_id=?2", - params![revision.file_id, retired], - )?; - tx.execute( - "INSERT OR REPLACE INTO file_aliases VALUES (?1,?2)", - params![retired, revision.file_id], - )?; + if !restored_retired { + tx.execute( + "UPDATE file_aliases SET file_id=?1 WHERE file_id=?2", + params![revision.file_id, retired], + )?; + tx.execute( + "INSERT OR REPLACE INTO file_aliases VALUES (?1,?2)", + params![retired, revision.file_id], + )?; + } tx.execute("UPDATE outbox SET state='archived' WHERE file_id=?1 AND state IN ('pending','queued')",[&retired])?; tx.execute("UPDATE sync_jobs SET state='archived' WHERE binding=?1 AND file_id=?2 AND state!='acked'",params![binding,retired])?; } @@ -504,7 +625,7 @@ mod tests { let copy = Uuid::new_v4().to_string(); ws.db .execute( - "INSERT INTO sync_resolutions VALUES (?1,2,?2,?3,?4,?5,?6,?7,'pending')", + "INSERT INTO sync_resolutions (binding,sequence,choice,destination,expected,operation_id,rename_id,copy_id,state,retire_id,restore_id) VALUES (?1,2,?2,?3,?4,?5,?6,?7,'pending',?8,?9)", params![ binding.id, choice, @@ -512,7 +633,9 @@ mod tests { local.hash, operation, Uuid::new_v4().to_string(), - copy + copy, + Uuid::new_v4().to_string(), + Uuid::new_v4().to_string() ], ) .unwrap(); @@ -686,12 +809,13 @@ mod tests { let operation = Uuid::new_v4().to_string(); let rename = Uuid::new_v4().to_string(); let copy = Uuid::new_v4().to_string(); - ws.db.execute("INSERT INTO sync_resolutions VALUES (?1,1,?2,?3,?4,?5,?6,?7,'pending')",params![binding.id,choice,if choice=="copy" {"copy.md"} else {""},local.hash,operation,rename,copy]).unwrap(); + let retire = Uuid::new_v4().to_string(); + ws.db.execute("INSERT INTO sync_resolutions (binding,sequence,choice,destination,expected,operation_id,rename_id,copy_id,state,retire_id,restore_id) VALUES (?1,1,?2,?3,?4,?5,?6,?7,'pending',?8,?9)",params![binding.id,choice,if choice=="copy" {"copy.md"} else {""},local.hash,operation,rename,copy,retire,Uuid::new_v4().to_string()]).unwrap(); if choice == "copy" { ws.write_operation("copy.md", "", b"local", "local", ©) .unwrap(); } - ws.mutate_with_origin("delete", "a.md", "", &local.hash, &rename, "remote") + ws.mutate_with_origin("delete", "a.md", "", &local.hash, &retire, "remote") .unwrap(); drop(ws); ws = Workspace::open(root.path()).unwrap(); @@ -725,4 +849,93 @@ mod tests { } } } + + #[test] + fn same_target_renames_preserve_the_occupant_for_all_choices_twenty_rounds() { + for round in 0..20 { + let choice = ["local", "remote", "copy"][round % 3]; + let root = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(root.path()).unwrap(); + let binding = ws + .sync_bind_download("https://sync.example", "remote-vault", "account") + .unwrap(); + let mut left = RemoteRevision { + vault_id: "remote-vault".into(), + sequence: 1, + file_id: Uuid::new_v4().to_string(), + base_revision: 0, + path: "left.md".into(), + operation: "put".into(), + hash: Some(ws.sync_store_bytes(b"left-content").unwrap()), + size: 12, + operation_id: Uuid::new_v4().to_string(), + }; + let right = RemoteRevision { + vault_id: "remote-vault".into(), + sequence: 2, + file_id: Uuid::new_v4().to_string(), + base_revision: 0, + path: "right.md".into(), + operation: "put".into(), + hash: Some(ws.sync_store_bytes(b"right-content").unwrap()), + size: 13, + operation_id: Uuid::new_v4().to_string(), + }; + receive(&mut ws, &binding.id, &left); + receive(&mut ws, &binding.id, &right); + let right_hash = right.hash.as_deref().unwrap(); + ws.rename("right.md", "target.md", right_hash).unwrap(); + + left.sequence = 3; + left.base_revision = 1; + left.path = "target.md".into(); + left.operation_id = Uuid::new_v4().to_string(); + receive(&mut ws, &binding.id, &left); + + let conflict = ws.sync_conflicts(&binding.id).unwrap().remove(0); + assert_eq!(conflict["current_path"], "target.md"); + assert_eq!(conflict["current_hash"], right_hash); + assert!(ws.sync_spool(right_hash).unwrap().is_file()); + assert!(ws + .sync_spool(left.hash.as_deref().unwrap()) + .unwrap() + .is_file()); + let copy = format!("copies/target-{round}.md"); + ws.sync_resolve( + &binding.id, + 3, + choice, + if choice == "copy" { © } else { "" }, + right_hash, + ) + .unwrap(); + drop(ws); + + let mut ws = Workspace::open(root.path()).unwrap(); + ws.sync_resume_resolutions(&binding.id).unwrap(); + assert!(ws.sync_conflicts(&binding.id).unwrap().is_empty()); + assert!(!root.path().join("left.md").exists()); + let target = ws.read("target.md").unwrap(); + assert_eq!(target.entry.file_id, left.file_id); + assert_eq!( + target.content, + if choice == "local" { + "right-content" + } else { + "left-content" + } + ); + let restored = ws.read("right.md").unwrap(); + assert_eq!(restored.entry.file_id, right.file_id); + assert_eq!(restored.content, "right-content"); + if choice == "copy" { + assert_eq!(ws.read(©).unwrap().content, "right-content"); + } + ws.sync_capture(&binding.id).unwrap(); + assert_eq!( + ws.pending_count().unwrap(), + if choice == "remote" { 0 } else { 1 } + ); + } + } } diff --git a/frontend/src-tauri/src/sync_scope.rs b/frontend/src-tauri/src/sync_scope.rs index d3b7368..fa12145 100644 --- a/frontend/src-tauri/src/sync_scope.rs +++ b/frontend/src-tauri/src/sync_scope.rs @@ -176,7 +176,7 @@ mod tests { ws.db .query_row("PRAGMA user_version", [], |r| r.get::<_, i64>(0)) .unwrap(), - 11 + 12 ); } #[test] diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index cd72664..dfbd0ca 100644 --- a/frontend/src-tauri/src/workspace.rs +++ b/frontend/src-tauri/src/workspace.rs @@ -136,10 +136,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 > 11 { + if version > 12 { return Err(HostError::new("SCHEMA_INCOMPATIBLE")); } - if (1..11).contains(&version) { + if (1..12).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()])?; @@ -167,7 +167,7 @@ impl Workspace { 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_optional_scope (id INTEGER PRIMARY KEY CHECK(id=1),persona INTEGER NOT NULL CHECK(persona IN (0,1)),layout INTEGER NOT NULL CHECK(layout IN (0,1))); - 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));")?; + 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,retire_id TEXT NOT NULL,restore_id TEXT NOT NULL,PRIMARY KEY(binding,sequence));")?; let has_origin: bool = db.query_row( "SELECT EXISTS(SELECT 1 FROM pragma_table_info('file_ops') WHERE name='origin')", [], @@ -179,10 +179,32 @@ impl Workspace { [], )?; } + let has_retire_id: bool = db.query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info('sync_resolutions') WHERE name='retire_id')", + [], + |r| r.get(0), + )?; + if !has_retire_id { + db.execute( + "ALTER TABLE sync_resolutions ADD COLUMN retire_id TEXT NOT NULL DEFAULT ''", + [], + )?; + } + let has_restore_id: bool = db.query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info('sync_resolutions') WHERE name='restore_id')", + [], + |r| r.get(0), + )?; + if !has_restore_id { + db.execute( + "ALTER TABLE sync_resolutions ADD COLUMN restore_id TEXT NOT NULL DEFAULT ''", + [], + )?; + } 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("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=11; 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=12; COMMIT;")?; let vault_id: String = db .query_row("SELECT id FROM identity", [], |r| r.get(0)) .optional()? @@ -1375,6 +1397,56 @@ mod tests { ); } + #[test] + fn schema_eleven_upgrade_adds_durable_conflict_operation_ids() { + let dir = tempfile::tempdir().unwrap(); + let ws = Workspace::open(dir.path()).unwrap(); + ws.db + .execute_batch( + "DROP TABLE sync_resolutions; + CREATE TABLE 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)); + PRAGMA user_version=11;", + ) + .unwrap(); + drop(ws); + + let ws = Workspace::open(dir.path()).unwrap(); + assert_eq!( + ws.db + .query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0)) + .unwrap(), + 12 + ); + for column in ["retire_id", "restore_id"] { + assert!(ws + .db + .query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info('sync_resolutions') WHERE name=?1)", + [column], + |row| row.get::<_, bool>(0), + ) + .unwrap()); + } + let backup = fs::read_dir(dir.path().join(".ainote")) + .unwrap() + .filter_map(|entry| entry.ok()) + .find(|entry| { + entry + .file_name() + .to_string_lossy() + .starts_with("host-schema11-") + }) + .unwrap(); + let old = Connection::open(backup.path()).unwrap(); + assert!(!old + .query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info('sync_resolutions') WHERE name='retire_id')", + [], + |row| row.get::<_, bool>(0), + ) + .unwrap()); + } + #[test] fn unsafe_paths_external_change_and_remote_origin() { let dir = tempfile::tempdir().unwrap(); diff --git a/frontend/src-tauri/tests/sync_conflicts.rs b/frontend/src-tauri/tests/sync_conflicts.rs new file mode 100644 index 0000000..24e515d --- /dev/null +++ b/frontend/src-tauri/tests/sync_conflicts.rs @@ -0,0 +1,508 @@ +#![cfg(feature = "desktop")] + +use notesagent_host::{ + sync_client::SyncClient, + sync_state::Binding, + workspace::{hash, Workspace}, +}; +use serde_json::{json, Value}; +use std::{ + io::{BufRead, BufReader}, + path::Path, + process::{Child, Command, Stdio}, + sync::{Arc, Mutex}, +}; +use zeroize::Zeroizing; + +struct Server(Child); + +impl Drop for Server { + fn drop(&mut self) { + self.0.stdin.take(); + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[derive(Clone, Copy)] +enum ConflictKind { + SameEdit, + EditDelete, + EditRename, + SameTargetRename, + HistoryRestore, +} + +impl ConflictKind { + fn name(self) -> &'static str { + match self { + Self::SameEdit => "same-edit", + Self::EditDelete => "edit-delete", + Self::EditRename => "edit-rename", + Self::SameTargetRename => "same-target-rename", + Self::HistoryRestore => "history-restore", + } + } +} + +async fn push_all(client: &SyncClient, workspace: &Arc>, binding: &Binding) { + while client.push_one(workspace, binding).await.unwrap() {} +} + +async fn pull_all(client: &SyncClient, workspace: &Arc>, binding: &Binding) { + while client.pull_page(workspace, binding).await.unwrap() > 0 {} +} + +struct SyncPair<'a> { + client_a: &'a SyncClient, + client_b: &'a SyncClient, + workspace_a: &'a Arc>, + workspace_b: &'a Arc>, + binding_a: &'a Binding, + binding_b: &'a Binding, +} + +impl SyncPair<'_> { + async fn resolve_copy_and_converge( + &self, + copy_path: &str, + local_content: &str, + main_path: &str, + remote_content: Option<&str>, + ) { + let conflict = self + .workspace_b + .lock() + .unwrap() + .sync_conflicts(&self.binding_b.id) + .unwrap() + .remove(0); + let local_hash = hash(local_content.as_bytes()); + assert_eq!(conflict["current_hash"], local_hash); + { + let mut ws = self.workspace_b.lock().unwrap(); + assert!(ws.sync_spool(&local_hash).unwrap().is_file()); + if let Some(remote_hash) = conflict["remote"]["hash"].as_str() { + assert!(ws.sync_spool(remote_hash).unwrap().is_file()); + } + ws.sync_resolve( + &self.binding_b.id, + conflict["sequence"].as_i64().unwrap(), + "copy", + copy_path, + &local_hash, + ) + .unwrap(); + } + push_all(self.client_b, self.workspace_b, self.binding_b).await; + pull_all(self.client_a, self.workspace_a, self.binding_a).await; + pull_all(self.client_b, self.workspace_b, self.binding_b).await; + + let copy_a = self.workspace_a.lock().unwrap().read(copy_path).unwrap(); + let copy_b = self.workspace_b.lock().unwrap().read(copy_path).unwrap(); + assert_eq!(copy_a.content, local_content); + assert_eq!(copy_a.entry.hash, copy_b.entry.hash); + assert_eq!(copy_a.entry.file_id, copy_b.entry.file_id); + for workspace in [self.workspace_a, self.workspace_b] { + let mut ws = workspace.lock().unwrap(); + if let Some(remote_content) = remote_content { + assert_eq!(ws.read(main_path).unwrap().content, remote_content); + } else { + assert!(!ws.root.join(main_path).exists()); + } + } + assert!(self + .workspace_a + .lock() + .unwrap() + .sync_conflicts(&self.binding_a.id) + .unwrap() + .is_empty()); + assert!(self + .workspace_b + .lock() + .unwrap() + .sync_conflicts(&self.binding_b.id) + .unwrap() + .is_empty()); + } +} + +#[tokio::test] +async fn s03_actual_service_converges_five_conflict_classes_twenty_rounds() { + let root = tempfile::tempdir().unwrap(); + std::fs::write(root.path().join(".opennexus-test"), b"fixture").unwrap(); + let service = Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../server sync") + .canonicalize() + .unwrap(); + let python = service.join(if cfg!(windows) { + ".venv/Scripts/python.exe" + } else { + ".venv/bin/python" + }); + let mut server = Server( + Command::new(python) + .args(["-m", "tests.host_fixture"]) + .arg(root.path()) + .current_dir(service) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .unwrap(), + ); + let mut line = String::new(); + BufReader::new(server.0.stdout.take().unwrap()) + .read_line(&mut line) + .unwrap(); + let ready: Value = serde_json::from_str(&line).unwrap(); + let endpoint = format!("http://127.0.0.1:{}", ready["port"]); + let public = SyncClient::new(&endpoint, Zeroizing::new(String::new()), true).unwrap(); + public.handshake().await.unwrap(); + let session_a = public + .login( + "rust-fixture", + Zeroizing::new("controlled-fixture-password".into()), + "S-03 A", + ) + .await + .unwrap(); + let session_b = public + .login( + "rust-fixture", + Zeroizing::new("controlled-fixture-password".into()), + "S-03 B", + ) + .await + .unwrap(); + let client_a = SyncClient::new( + &endpoint, + Zeroizing::new(session_a.access_token.clone()), + true, + ) + .unwrap(); + let client_b = SyncClient::new( + &endpoint, + Zeroizing::new(session_b.access_token.clone()), + true, + ) + .unwrap(); + let vault = client_a + .json( + reqwest::Method::POST, + "sync/v1/vaults", + Some(json!({"name":"S-03 matrix"})), + ) + .await + .unwrap(); + let remote = vault["vault_id"].as_str().unwrap(); + client_a.verify_empty(remote).await.unwrap(); + + let root_a = tempfile::tempdir().unwrap(); + let root_b = tempfile::tempdir().unwrap(); + let workspace_a = Arc::new(Mutex::new(Workspace::open(root_a.path()).unwrap())); + let workspace_b = Arc::new(Mutex::new(Workspace::open(root_b.path()).unwrap())); + workspace_a + .lock() + .unwrap() + .write("first-bind.md", "", b"local snapshot", "local") + .unwrap(); + let binding_a = workspace_a + .lock() + .unwrap() + .sync_bind_empty(&endpoint, remote, "rust-fixture") + .unwrap(); + let first_job = workspace_a + .lock() + .unwrap() + .sync_next(&binding_a.id) + .unwrap() + .unwrap(); + assert_eq!(first_job.operation, "put"); + let before = client_a + .json( + reqwest::Method::GET, + &format!("sync/v1/vaults/{remote}/changes"), + None, + ) + .await + .unwrap(); + assert_eq!(before["boundary"], 0); + assert!(before["items"].as_array().unwrap().is_empty()); + push_all(&client_a, &workspace_a, &binding_a).await; + let after = client_a + .json( + reqwest::Method::GET, + &format!("sync/v1/vaults/{remote}/changes"), + None, + ) + .await + .unwrap(); + assert_eq!( + after["items"] + .as_array() + .unwrap() + .iter() + .filter(|revision| revision["operation"] == "delete") + .count(), + 0 + ); + let binding_b = workspace_b + .lock() + .unwrap() + .sync_bind_download(&endpoint, remote, "rust-fixture") + .unwrap(); + pull_all(&client_b, &workspace_b, &binding_b).await; + let pair = SyncPair { + client_a: &client_a, + client_b: &client_b, + workspace_a: &workspace_a, + workspace_b: &workspace_b, + binding_a: &binding_a, + binding_b: &binding_b, + }; + + let kinds = [ + ConflictKind::SameEdit, + ConflictKind::EditDelete, + ConflictKind::EditRename, + ConflictKind::SameTargetRename, + ConflictKind::HistoryRestore, + ]; + for kind in kinds { + for round in 0..20 { + let stem = format!("matrix/{}-{round}", kind.name()); + let source = format!("{stem}-source.md"); + let target = match kind { + ConflictKind::EditRename | ConflictKind::SameTargetRename => { + format!("{stem}-target.md") + } + _ => source.clone(), + }; + let right = format!("{stem}-right.md"); + let (base, local_content, remote_content) = match kind { + ConflictKind::SameEdit => ( + "base".to_owned(), + format!("local-edit-{round}"), + Some(format!("remote-edit-{round}")), + ), + ConflictKind::EditDelete => { + ("base".to_owned(), format!("local-survivor-{round}"), None) + } + ConflictKind::EditRename => ( + "rename-base".to_owned(), + format!("edited-before-rename-{round}"), + Some("rename-base".to_owned()), + ), + ConflictKind::SameTargetRename => ( + "left-content".to_owned(), + "right-content".to_owned(), + Some("left-content".to_owned()), + ), + ConflictKind::HistoryRestore => ( + "historical".to_owned(), + format!("local-history-{round}"), + Some(format!("remote-history-{round}")), + ), + }; + workspace_a + .lock() + .unwrap() + .write(&source, "", base.as_bytes(), "local") + .unwrap(); + if matches!(kind, ConflictKind::SameTargetRename) { + workspace_a + .lock() + .unwrap() + .write(&right, "", local_content.as_bytes(), "local") + .unwrap(); + } + push_all(&client_a, &workspace_a, &binding_a).await; + pull_all(&client_b, &workspace_b, &binding_b).await; + + if matches!(kind, ConflictKind::HistoryRestore) { + { + let mut ws = workspace_a.lock().unwrap(); + let current = ws.read(&source).unwrap(); + ws.delete(&source, ¤t.entry.hash).unwrap(); + } + push_all(&client_a, &workspace_a, &binding_a).await; + pull_all(&client_b, &workspace_b, &binding_b).await; + } + + match kind { + ConflictKind::SameEdit => { + for (workspace, content) in [ + (&workspace_a, remote_content.as_deref().unwrap()), + (&workspace_b, local_content.as_str()), + ] { + let mut ws = workspace.lock().unwrap(); + let current = ws.read(&source).unwrap(); + ws.write(&source, ¤t.entry.hash, content.as_bytes(), "local") + .unwrap(); + } + } + ConflictKind::EditDelete => { + { + let mut ws = workspace_a.lock().unwrap(); + let current = ws.read(&source).unwrap(); + ws.delete(&source, ¤t.entry.hash).unwrap(); + } + { + let mut ws = workspace_b.lock().unwrap(); + let current = ws.read(&source).unwrap(); + ws.write( + &source, + ¤t.entry.hash, + local_content.as_bytes(), + "local", + ) + .unwrap(); + } + } + ConflictKind::EditRename => { + { + let mut ws = workspace_a.lock().unwrap(); + let current = ws.read(&source).unwrap(); + ws.rename(&source, &target, ¤t.entry.hash).unwrap(); + } + { + let mut ws = workspace_b.lock().unwrap(); + let current = ws.read(&source).unwrap(); + ws.write( + &source, + ¤t.entry.hash, + local_content.as_bytes(), + "local", + ) + .unwrap(); + } + } + ConflictKind::SameTargetRename => { + { + let mut ws = workspace_a.lock().unwrap(); + let current = ws.read(&source).unwrap(); + ws.rename(&source, &target, ¤t.entry.hash).unwrap(); + } + { + let mut ws = workspace_b.lock().unwrap(); + let current = ws.read(&right).unwrap(); + ws.rename(&right, &target, ¤t.entry.hash).unwrap(); + } + } + ConflictKind::HistoryRestore => { + workspace_a + .lock() + .unwrap() + .write( + &source, + "", + remote_content.as_deref().unwrap().as_bytes(), + "local", + ) + .unwrap(); + workspace_b + .lock() + .unwrap() + .write(&source, "", local_content.as_bytes(), "local") + .unwrap(); + } + } + push_all(&client_a, &workspace_a, &binding_a).await; + pull_all(&client_b, &workspace_b, &binding_b).await; + let copy = format!("copies/{}-{round}.md", kind.name()); + pair.resolve_copy_and_converge( + ©, + &local_content, + &target, + remote_content.as_deref(), + ) + .await; + if matches!( + kind, + ConflictKind::EditRename | ConflictKind::SameTargetRename + ) { + assert!(!workspace_a.lock().unwrap().root.join(&source).exists()); + assert!(!workspace_b.lock().unwrap().root.join(&source).exists()); + } + if matches!(kind, ConflictKind::SameTargetRename) { + let right_a = workspace_a.lock().unwrap().read(&right).unwrap(); + let right_b = workspace_b.lock().unwrap().read(&right).unwrap(); + assert_eq!(right_a.content, "right-content"); + assert_eq!(right_a.entry.hash, right_b.entry.hash); + assert_eq!(right_a.entry.file_id, right_b.entry.file_id); + } + } + } + + let rebound_vault = client_a + .json( + reqwest::Method::POST, + "sync/v1/vaults", + Some(json!({"name":"S-03 rebind"})), + ) + .await + .unwrap(); + let rebound_remote = rebound_vault["vault_id"].as_str().unwrap(); + let rebound_root = tempfile::tempdir().unwrap(); + let rebound_ws = Arc::new(Mutex::new(Workspace::open(rebound_root.path()).unwrap())); + rebound_ws + .lock() + .unwrap() + .write("rebound.md", "", b"current", "local") + .unwrap(); + let old_binding = rebound_ws + .lock() + .unwrap() + .sync_bind_empty(&endpoint, rebound_remote, "rust-fixture") + .unwrap(); + let old_operation = rebound_ws + .lock() + .unwrap() + .sync_next(&old_binding.id) + .unwrap() + .unwrap() + .operation_id; + rebound_ws + .lock() + .unwrap() + .sync_unbind(&old_binding.id) + .unwrap(); + let new_binding = rebound_ws + .lock() + .unwrap() + .sync_bind_empty(&endpoint, rebound_remote, "rust-fixture") + .unwrap(); + let new_operation = rebound_ws + .lock() + .unwrap() + .sync_next(&new_binding.id) + .unwrap() + .unwrap() + .operation_id; + assert_ne!(old_operation, new_operation); + assert_eq!( + client_a + .push_one(&rebound_ws, &old_binding) + .await + .unwrap_err() + .code, + "SYNC_BINDING_CHANGED" + ); + push_all(&client_a, &rebound_ws, &new_binding).await; + let rebound_changes = client_a + .json( + reqwest::Method::GET, + &format!("sync/v1/vaults/{rebound_remote}/changes"), + None, + ) + .await + .unwrap(); + let rebound_items = rebound_changes["items"].as_array().unwrap(); + assert_eq!(rebound_items.len(), 1); + assert_eq!(rebound_items[0]["operation_id"], new_operation); + assert!(rebound_items + .iter() + .all(|revision| revision["operation_id"] != old_operation)); +} diff --git a/scripts/acceptance_cases/s03_sync_client.py b/scripts/acceptance_cases/s03_sync_client.py new file mode 100644 index 0000000..aedbeb3 --- /dev/null +++ b/scripts/acceptance_cases/s03_sync_client.py @@ -0,0 +1,136 @@ +"""S-03 conflict matrix, first-bind, and rebind isolation acceptance driver.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import subprocess +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +MANIFEST = ROOT / "frontend" / "src-tauri" / "Cargo.toml" +SERVICE_TEST = "s03_actual_service_converges_five_conflict_classes_twenty_rounds" +TARGET_RENAME_TEST = ( + "sync_resolution::tests::same_target_renames_preserve_the_occupant_for_all_choices_twenty_rounds" +) + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def run_exact(command: list[str]) -> bool: + completed = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=False) + print(completed.stdout, end="") + print(completed.stderr, end="") + return completed.returncode == 0 and "1 passed; 0 failed" in completed.stdout + + +def result(case_id: str, status: str, reason: str, assertions: list[dict]) -> dict: + evidence = f"cargo exact tests {SERVICE_TEST} and {TARGET_RENAME_TEST}" + for assertion in assertions: + assertion.update({"status": status, "evidence": evidence}) + files = [] + for relative in ( + "frontend/src-tauri/src/workspace.rs", + "frontend/src-tauri/src/sync_inbox.rs", + "frontend/src-tauri/src/sync_resolution.rs", + "frontend/src-tauri/tests/sync_conflicts.rs", + "server sync/tests/host_fixture.py", + ): + files.append({"path": relative, "sha256": sha256(ROOT / relative)}) + return { + "schema": 1, + "case_id": case_id, + "status": status, + "reason": reason, + "assertions": assertions, + "metrics": {"peak_rss_bytes": None, "max_process_count": None, "denied_access_count": None}, + "files": files, + "revisions": [ + { + "scope": "actual two-client HTTP conflict matrix", + "classes": 5, + "rounds_per_class": 20, + "resolution": "copy", + }, + { + "scope": "same-target rename durable choices", + "rounds": 20, + "choices": ["local", "remote", "copy"], + }, + ], + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--config", required=True) + parser.add_argument("--output", required=True) + args = parser.parse_args() + case_id = os.environ.get("OPENNEXUS_ACCEPTANCE_CASE_ID", "") + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + assertions = [ + {"name": "same-edit conflicts preserve both contents and converge for twenty rounds"}, + {"name": "edit-delete conflicts preserve edited bytes and converge for twenty rounds"}, + {"name": "edit-rename conflicts preserve both contents and converge for twenty rounds"}, + {"name": "same-target renames restore the displaced server head and converge for twenty rounds"}, + {"name": "history restores preserve both versions and converge for twenty rounds"}, + {"name": "first binding to an empty Vault emits zero server delete revisions"}, + {"name": "unbind and rebind reject and never transmit the archived operation ID"}, + {"name": "all same-target choices remain complete after Workspace reopen"}, + ] + cargo = shutil.which(os.environ.get("CARGO", "cargo")) + passed = case_id == "S-03" and cargo is not None + if passed: + passed = run_exact( + [ + cargo, + "test", + "--manifest-path", + str(MANIFEST), + "--locked", + "--features", + "desktop", + "--test", + "sync_conflicts", + SERVICE_TEST, + "--", + "--exact", + "--nocapture", + ] + ) + passed = run_exact( + [ + cargo, + "test", + "--manifest-path", + str(MANIFEST), + "--locked", + "--lib", + TARGET_RENAME_TEST, + "--", + "--exact", + "--nocapture", + ] + ) and passed + payload = result( + case_id, + "PASSED" if passed else "FAILED", + "" if passed else "An exact S-03 conflict, convergence, or binding-isolation oracle failed.", + assertions, + ) + output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + return 0 if passed else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/phase3_acceptance.py b/scripts/phase3_acceptance.py index e30497e..add4e9f 100644 --- a/scripts/phase3_acceptance.py +++ b/scripts/phase3_acceptance.py @@ -67,6 +67,11 @@ CASE_DRIVERS: dict[str, dict[str, Any]] = { "timeout_seconds": 900, "required_metrics": (), }, + "S-03": { + "driver": "scripts/acceptance_cases/s03_sync_client.py", + "timeout_seconds": 900, + "required_metrics": (), + }, } ENV_NAME = re.compile(r"[A-Z][A-Z0-9_]{2,127}") RUN_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{2,63}")