diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 9e00bea..5f151a8 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -51,3 +51,11 @@ | 发布 | 统一逐 ID 验收 runner;MSVC 签名安装更新包、SBOM/依赖扫描;两台独立设备 E-01–05 | Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写入口在 desktop 模式明确拒绝,不能据此宣称 AI 笔记工作流完成。Host `sync/extensions` 能力仍为 false。缺少实机与签名环境不构成其余普通工程尚未完成的理由。 + + +## 增量:可恢复的同步冲突解决 + +- Workspace schema 5 在修改文件前持久保存解决选择和三个操作 ID;保留本地、采用远端、另存副本均通过现有文件 journal 提交,再归档旧上传队列。启动下一轮 push/pull 时恢复未完成的解决事务。 +- 旧任务进入 conflict/archived 后拒绝上传或提交回执更新;未解决文件暂停发送,其他文件仍可推进。同路径删除后重建使用新的远端 file_id,原删除身份保留为 tombstone。 +- 两个独立 Rust Workspace/会话通过真实本地 Sync HTTP 服务验证三种选择最终收敛及副本独立身份。新增三项测试覆盖文件已提交但解决记录未提交时重启、删除对编辑、远端删除后重建与恢复旧身份。 +- Rust desktop 全目标 42 项通过,Clippy `-D warnings` 通过。测试服务使用隔离 SQLite/文件对象存储;尚不代表 PostgreSQL/MinIO 部署验收。不同文件身份占用同一路径仍返回明确的需改名错误,解决界面和 Host 会话接入继续实现中。 diff --git a/frontend/src-tauri/src/lib.rs b/frontend/src-tauri/src/lib.rs index 0ce43d4..0c631e8 100644 --- a/frontend/src-tauri/src/lib.rs +++ b/frontend/src-tauri/src/lib.rs @@ -11,6 +11,7 @@ pub mod session_lock; #[cfg(feature = "desktop")] pub mod sync_client; pub mod sync_inbox; +pub mod sync_resolution; pub mod sync_state; pub mod workspace; pub mod workspace_broker; diff --git a/frontend/src-tauri/src/sync_client.rs b/frontend/src-tauri/src/sync_client.rs index b01377e..2609188 100644 --- a/frontend/src-tauri/src/sync_client.rs +++ b/frontend/src-tauri/src/sync_client.rs @@ -236,6 +236,7 @@ impl SyncClient { } identifier(&binding.remote_vault)?; let job = workspace.access(|ws| { + ws.sync_resume_resolutions(&binding.id)?; ws.sync_capture(&binding.id)?; ws.sync_next(&binding.id) })?; @@ -272,6 +273,7 @@ impl SyncClient { } identifier(&binding.remote_vault)?; workspace.access(|ws| { + ws.sync_resume_resolutions(&binding.id)?; while ws.sync_apply_pending(&binding.id)? {} Ok(()) })?; diff --git a/frontend/src-tauri/src/sync_inbox.rs b/frontend/src-tauri/src/sync_inbox.rs index 7ee9136..2983384 100644 --- a/frontend/src-tauri/src/sync_inbox.rs +++ b/frontend/src-tauri/src/sync_inbox.rs @@ -224,7 +224,7 @@ impl Workspace { remote ], )?; - tx.execute("UPDATE sync_jobs SET state='conflict' WHERE binding=?1 AND file_id=?2 AND state!='acked'", params![binding,revision.file_id])?; + tx.execute("UPDATE sync_jobs SET state='conflict' WHERE binding=?1 AND file_id=?2 AND state NOT IN ('acked','archived')", params![binding,revision.file_id])?; tx.commit()?; self.sync_finish(binding, revision, "conflict") } @@ -239,7 +239,8 @@ impl Workspace { let own: Option = self.db.query_row("SELECT binding,operation_id,file_id,path,hash,size,operation,state,base_revision,upload_id FROM sync_jobs WHERE binding=?1 AND operation_id=?2", params![binding,revision.operation_id], |r| { Ok(Job { binding:r.get(0)?,operation_id:r.get(1)?,file_id:r.get(2)?,path:r.get(3)?,hash:r.get(4)?,size:r.get(5)?,operation:r.get(6)?,state:r.get(7)?,base_revision:r.get(8)?,upload_id:r.get(9)? }) }).optional()?; - if let Some(job) = own { + if let Some(job) = own.filter(|job| !matches!(job.state.as_str(), "archived" | "conflict")) + { self.sync_ack( &job, &serde_json::to_value(&revision) diff --git a/frontend/src-tauri/src/sync_resolution.rs b/frontend/src-tauri/src/sync_resolution.rs new file mode 100644 index 0000000..5b7a60b --- /dev/null +++ b/frontend/src-tauri/src/sync_resolution.rs @@ -0,0 +1,420 @@ +//! User decisions are durable before changing files; journal IDs make restart replay safe. +use crate::{ + sync_inbox::RemoteRevision, + workspace::{hash, Entry, HostError, Result, Workspace}, +}; +use rusqlite::{params, OptionalExtension}; +use std::fs; +use uuid::Uuid; + +impl Workspace { + pub fn sync_resolve( + &mut self, + binding: &str, + sequence: i64, + choice: &str, + destination: &str, + expected: &str, + ) -> Result<()> { + self.check_binding(binding)?; + if !matches!(choice, "local" | "remote" | "copy") { + return Err(HostError::new("SYNC_RESOLUTION_INVALID")); + } + let existing: Option<(String,String,String)> = self.db.query_row("SELECT choice,destination,expected FROM sync_resolutions WHERE binding=?1 AND sequence=?2",params![binding,sequence],|r| Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?; + if let Some(previous) = existing { + if previous + != ( + choice.to_owned(), + destination.to_owned(), + expected.to_owned(), + ) + { + 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 source = self.resolve(&path)?; + let current = if source.is_file() { + self.sync_store_bytes(&fs::read(source)?)? + } else { + String::new() + }; + if current != expected { + return Err(HostError::new("REVISION_CONFLICT")); + } + if !current.is_empty() && self.path_for_id(&revision.file_id).is_err() { + return Err(HostError::new("SYNC_CONFLICT_REQUIRES_RENAME")); + } + if choice == "copy" { + if current.is_empty() || self.resolve(destination)?.exists() { + return Err(HostError::new("PATH_CONFLICT")); + } + } else if !destination.is_empty() { + return Err(HostError::new("SYNC_RESOLUTION_INVALID")); + } + self.db.execute( + "INSERT INTO sync_resolutions VALUES (?1,?2,?3,?4,?5,?6,?7,?8,'pending')", + params![ + binding, + sequence, + choice, + destination, + expected, + Uuid::new_v4().to_string(), + Uuid::new_v4().to_string(), + Uuid::new_v4().to_string() + ], + )?; + } + self.sync_apply_resolution(binding, sequence) + } + pub fn sync_resume_resolutions(&mut self, binding: &str) -> Result<()> { + self.check_binding(binding)?; + let pending = { + let mut statement=self.db.prepare("SELECT sequence FROM sync_resolutions WHERE binding=?1 AND state='pending' ORDER BY sequence")?; + let values = statement + .query_map([binding], |r| r.get::<_, i64>(0))? + .collect::, _>>()?; + values + }; + for sequence in pending { + self.sync_apply_resolution(binding, sequence)?; + } + Ok(()) + } + 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)?)))?; + if state == "completed" { + return Ok(()); + } + let (stored_path, remote): (String, String) = self.db.query_row( + "SELECT local_path,remote FROM sync_conflicts WHERE binding=?1 AND sequence=?2", + 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 head: i64 = self.db.query_row( + "SELECT revision FROM sync_heads WHERE binding=?1 AND file_id=?2", + params![binding, revision.file_id], + |r| r.get(0), + )?; + if head != sequence { + return Err(HostError::new("SYNC_CONFLICT_CHANGED")); + } + let path = self.path_for_id(&revision.file_id).unwrap_or(stored_path); + if !self + .operation(&operation)? + .is_some_and(|value| value["state"] == "committed") + { + let source = self.resolve(&path)?; + let current = if source.is_file() { + hash(&fs::read(source)?) + } else { + String::new() + }; + if current != expected { + return Err(HostError::new("REVISION_CONFLICT")); + } + if choice == "copy" { + let content = fs::read(self.sync_spool(&expected)?)?; + self.write_operation(&destination, "", &content, "local", ©)?; + } + if choice == "local" { + if current.is_empty() { + self.sync_delete_intent(&revision, &operation)?; + } else { + let content = fs::read(self.sync_spool(&expected)?)?; + self.write_operation(&path, &expected, &content, "local", &operation)?; + } + } else if revision.operation == "delete" { + if !current.is_empty() { + self.mutate_with_origin("delete", &path, "", ¤t, &operation, "remote")?; + } + } else { + if path != revision.path && !current.is_empty() { + self.mutate_with_origin( + "rename", + &path, + &revision.path, + ¤t, + &rename, + "remote", + )?; + } + let content = fs::read( + self.sync_spool( + revision + .hash + .as_deref() + .ok_or_else(|| HostError::new("SYNC_RESPONSE_INVALID"))?, + )?, + )?; + self.write_with_identity( + &revision.path, + ¤t, + &content, + "remote", + &operation, + Some(&revision.file_id), + )?; + } + } + let tx = self.db.transaction()?; + tx.execute("UPDATE outbox SET state='archived' WHERE file_id=?1 AND state IN ('pending','queued') AND operation_id!=?2",params![revision.file_id,operation])?; + tx.execute("UPDATE sync_jobs SET state='archived' WHERE binding=?1 AND file_id=?2 AND state!='acked' AND operation_id!=?3",params![binding,revision.file_id,operation])?; + tx.execute( + "UPDATE sync_conflicts SET state='resolved' WHERE binding=?1 AND sequence=?2", + params![binding, sequence], + )?; + tx.execute( + "UPDATE sync_resolutions SET state='completed' WHERE binding=?1 AND sequence=?2", + params![binding, sequence], + )?; + tx.commit()?; + Ok(()) + } + fn sync_delete_intent(&mut self, revision: &RemoteRevision, operation: &str) -> Result<()> { + let tx = self.db.transaction()?; + tx.execute( + "UPDATE files SET deleted=1,revision=revision+1 WHERE id=?1", + [&revision.file_id], + )?; + let entry: Entry = tx.query_row( + "SELECT id,path,hash,revision FROM files WHERE id=?1", + [&revision.file_id], + |r| { + Ok(Entry { + file_id: r.get(0)?, + path: r.get(1)?, + hash: r.get(2)?, + revision: r.get(3)?, + deleted: true, + is_folder: false, + }) + }, + )?; + tx.execute( + "INSERT INTO outbox VALUES (?1,?2,?3,?4,'','delete',X'','pending')", + params![operation, entry.file_id, entry.revision, revision.path], + )?; + let encoded = + serde_json::to_string(&entry).map_err(|_| HostError::new("DATABASE_ERROR"))?; + tx.execute( + "INSERT INTO operations VALUES (?1,?2,'committed',?3)", + params![ + operation, + hash(format!("sync-delete:{}:{}", revision.file_id, revision.sequence).as_bytes()), + encoded + ], + )?; + tx.commit()?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + fn receive(ws: &mut Workspace, binding: &str, revision: &RemoteRevision) { + ws.sync_set_boundary(binding, revision.sequence).unwrap(); + ws.sync_stage(binding, revision).unwrap(); + ws.sync_apply_pending(binding).unwrap(); + } + #[test] + fn decisions_recover_after_file_commit_without_duplicate_outbox() { + for choice in ["local", "remote", "copy"] { + 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 file_id = Uuid::new_v4().to_string(); + let mut remote = RemoteRevision { + vault_id: "remote-vault".into(), + sequence: 1, + file_id, + base_revision: 0, + path: "a.md".into(), + operation: "put".into(), + hash: Some(ws.sync_store_bytes(b"one").unwrap()), + size: 3, + operation_id: Uuid::new_v4().to_string(), + }; + receive(&mut ws, &binding.id, &remote); + let local = ws.write("a.md", &hash(b"one"), b"local", "local").unwrap(); + ws.sync_capture(&binding.id).unwrap(); + let stale = ws.sync_next(&binding.id).unwrap().unwrap(); + remote.sequence = 2; + remote.base_revision = 1; + remote.hash = Some(ws.sync_store_bytes(b"two").unwrap()); + remote.operation_id = Uuid::new_v4().to_string(); + receive(&mut ws, &binding.id, &remote); + assert!(ws.sync_next(&binding.id).unwrap().is_none()); + assert_eq!( + ws.sync_commit_payload(&stale).unwrap_err().code, + "SYNC_OPERATION_SUPERSEDED" + ); + let operation = Uuid::new_v4().to_string(); + let copy = Uuid::new_v4().to_string(); + ws.db + .execute( + "INSERT INTO sync_resolutions VALUES (?1,2,?2,?3,?4,?5,?6,?7,'pending')", + params![ + binding.id, + choice, + if choice == "copy" { "copy.md" } else { "" }, + local.hash, + operation, + Uuid::new_v4().to_string(), + copy + ], + ) + .unwrap(); + // Simulate a crash after the filesystem journal commits but before the resolution transaction. + if choice == "copy" { + ws.write_operation("copy.md", "", b"local", "local", ©) + .unwrap(); + } + if choice == "local" { + ws.write_operation("a.md", &local.hash, b"local", "local", &operation) + .unwrap(); + } else { + ws.write_with_identity( + "a.md", + &local.hash, + b"two", + "remote", + &operation, + Some(&remote.file_id), + ) + .unwrap(); + } + drop(ws); + let mut ws = Workspace::open(root.path()).unwrap(); + ws.sync_resume_resolutions(&binding.id).unwrap(); + ws.sync_resume_resolutions(&binding.id).unwrap(); + assert!(ws.sync_conflicts(&binding.id).unwrap().is_empty()); + assert_eq!( + ws.read("a.md").unwrap().content, + if choice == "local" { "local" } else { "two" } + ); + assert_eq!( + ws.pending_count().unwrap(), + if choice == "remote" { 0 } else { 1 } + ); + if choice == "copy" { + assert_eq!(ws.read("copy.md").unwrap().content, "local"); + } + assert_eq!( + ws.sync_commit_payload(&stale).unwrap_err().code, + "SYNC_OPERATION_SUPERSEDED" + ); + } + } + #[test] + fn deletion_resolution_and_remote_recreation_preserve_identity() { + for choice in ["local", "remote", "copy"] { + 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 remote = RemoteRevision { + vault_id: "remote-vault".into(), + sequence: 1, + file_id: Uuid::new_v4().to_string(), + base_revision: 0, + path: "a.md".into(), + operation: "put".into(), + hash: Some(ws.sync_store_bytes(b"one").unwrap()), + size: 3, + operation_id: Uuid::new_v4().to_string(), + }; + receive(&mut ws, &binding.id, &remote); + let original = remote.file_id.clone(); + let local = ws.write("a.md", &hash(b"one"), b"local", "local").unwrap(); + remote.sequence = 2; + remote.base_revision = 1; + remote.operation = "delete".into(); + remote.hash = None; + remote.size = 0; + remote.operation_id = Uuid::new_v4().to_string(); + receive(&mut ws, &binding.id, &remote); + ws.sync_resolve( + &binding.id, + 2, + choice, + if choice == "copy" { "copy.md" } else { "" }, + &local.hash, + ) + .unwrap(); + if choice == "local" { + assert_eq!(ws.read("a.md").unwrap().content, "local"); + continue; + } + assert!(!root.path().join("a.md").exists()); + remote.sequence = 3; + remote.base_revision = 0; + remote.file_id = Uuid::new_v4().to_string(); + remote.operation = "put".into(); + remote.hash = Some(ws.sync_store_bytes(b"new").unwrap()); + remote.size = 3; + remote.operation_id = Uuid::new_v4().to_string(); + receive(&mut ws, &binding.id, &remote); + assert_eq!(ws.read("a.md").unwrap().entry.file_id, remote.file_id); + // A tombstoned identity can reappear at another free path. + remote.sequence = 4; + remote.base_revision = 2; + remote.file_id = original.clone(); + remote.path = "restored.md".into(); + remote.operation_id = Uuid::new_v4().to_string(); + receive(&mut ws, &binding.id, &remote); + assert_eq!(ws.read("restored.md").unwrap().entry.file_id, original); + assert_eq!(ws.read("a.md").unwrap().content, "new"); + } + } + #[test] + fn keep_local_deletion_queues_new_delete_on_remote_head() { + 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 remote = RemoteRevision { + vault_id: "remote-vault".into(), + sequence: 1, + file_id: Uuid::new_v4().to_string(), + base_revision: 0, + path: "a.md".into(), + operation: "put".into(), + hash: Some(ws.sync_store_bytes(b"one").unwrap()), + size: 3, + operation_id: Uuid::new_v4().to_string(), + }; + receive(&mut ws, &binding.id, &remote); + ws.mutate_operation( + "delete", + "a.md", + "", + &hash(b"one"), + &Uuid::new_v4().to_string(), + ) + .unwrap(); + remote.sequence = 2; + remote.base_revision = 1; + remote.hash = Some(ws.sync_store_bytes(b"two").unwrap()); + remote.operation_id = Uuid::new_v4().to_string(); + receive(&mut ws, &binding.id, &remote); + ws.sync_resolve(&binding.id, 2, "local", "", "").unwrap(); + ws.sync_capture(&binding.id).unwrap(); + let job = ws.sync_next(&binding.id).unwrap().unwrap(); + let payload = ws.sync_commit_payload(&job).unwrap(); + assert_eq!(payload["base_revision"], 2); + assert_eq!(payload["operation"], "delete"); + assert_eq!(ws.pending_count().unwrap(), 1); + } +} diff --git a/frontend/src-tauri/src/sync_state.rs b/frontend/src-tauri/src/sync_state.rs index de98448..84b6231 100644 --- a/frontend/src-tauri/src/sync_state.rs +++ b/frontend/src-tauri/src/sync_state.rs @@ -104,7 +104,6 @@ impl Workspace { { return Err(HostError::new("SYNC_HASH_INVALID")); } - self.resolve(&format!("attachments/{digest}"))?; // Enforce the platform's general path rules. let root = self.root.join(".ainote/sync-spool"); if root.exists() { let meta = fs::symlink_metadata(&root)?; @@ -163,17 +162,29 @@ impl Workspace { } pub fn sync_next(&self, binding: &str) -> Result> { self.check_binding(binding)?; - Ok(self.db.query_row("SELECT binding,operation_id,file_id,path,hash,size,operation,state,base_revision,upload_id FROM sync_jobs WHERE binding=?1 AND state NOT IN ('acked','archived') ORDER BY rowid LIMIT 1", [binding], |r| { + Ok(self.db.query_row("SELECT binding,operation_id,file_id,path,hash,size,operation,state,base_revision,upload_id FROM sync_jobs WHERE binding=?1 AND state NOT IN ('acked','archived','conflict') AND NOT EXISTS (SELECT 1 FROM sync_conflicts c WHERE c.binding=sync_jobs.binding AND c.file_id=sync_jobs.file_id AND c.state='open') ORDER BY rowid LIMIT 1", [binding], |r| { Ok(Job { binding:r.get(0)?,operation_id:r.get(1)?,file_id:r.get(2)?,path:r.get(3)?,hash:r.get(4)?,size:r.get(5)?,operation:r.get(6)?,state:r.get(7)?,base_revision:r.get(8)?,upload_id:r.get(9)? }) }).optional()?) } - pub fn sync_upload(&self, job: &Job, upload: Option<&str>) -> Result<()> { + 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", + params![job.binding, job.operation_id], + |r| r.get(0), + )?; + if matches!(state.as_str(), "archived" | "conflict") { + return Err(HostError::new("SYNC_OPERATION_SUPERSEDED")); + } + Ok(()) + } + pub fn sync_upload(&self, job: &Job, upload: Option<&str>) -> Result<()> { + self.check_job(job)?; self.db.execute("UPDATE sync_jobs SET state='uploading',upload_id=?3 WHERE binding=?1 AND operation_id=?2 AND base_revision IS NULL", params![job.binding,job.operation_id,upload])?; Ok(()) } pub fn sync_commit_payload(&self, job: &Job) -> Result { - self.check_binding(&job.binding)?; + self.check_job(job)?; // The base is frozen exactly once. A response loss reuses the byte-equivalent payload. self.db.execute("UPDATE sync_jobs SET state='committing',base_revision=COALESCE((SELECT revision FROM sync_heads WHERE binding=?1 AND file_id=?3),0) WHERE binding=?1 AND operation_id=?2 AND base_revision IS NULL", params![job.binding,job.operation_id,job.file_id])?; diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index f92ddcf..293a088 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 > 4 { + if version > 5 { return Err(HostError::new("SCHEMA_INCOMPATIBLE")); } - if (1..4).contains(&version) { + if (1..5).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()])?; @@ -156,7 +156,8 @@ impl Workspace { CREATE TABLE IF NOT EXISTS sync_heads (binding TEXT NOT NULL,file_id TEXT NOT NULL,revision INTEGER NOT NULL,path TEXT NOT NULL,hash TEXT NOT NULL,PRIMARY KEY(binding,file_id)); CREATE TABLE IF NOT EXISTS sync_windows (binding TEXT PRIMARY KEY,boundary INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS sync_inbox (binding TEXT NOT NULL,sequence INTEGER NOT NULL,revision TEXT NOT NULL,operation_id TEXT NOT NULL,rename_id TEXT NOT NULL,state TEXT NOT NULL,PRIMARY KEY(binding,sequence)); - CREATE TABLE IF NOT EXISTS sync_conflicts (binding TEXT NOT NULL,sequence INTEGER NOT NULL,file_id TEXT NOT NULL,local_path TEXT NOT NULL,local_hash TEXT NOT NULL,remote TEXT NOT NULL,state TEXT NOT NULL,PRIMARY KEY(binding,sequence));")?; + CREATE TABLE IF NOT EXISTS sync_conflicts (binding TEXT NOT NULL,sequence INTEGER NOT NULL,file_id TEXT NOT NULL,local_path TEXT NOT NULL,local_hash TEXT NOT NULL,remote 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,PRIMARY KEY(binding,sequence));")?; let has_origin: bool = db.query_row( "SELECT EXISTS(SELECT 1 FROM pragma_table_info('file_ops') WHERE name='origin')", [], @@ -168,7 +169,7 @@ impl Workspace { [], )?; } - db.execute_batch("PRAGMA user_version=4; COMMIT;")?; + db.execute_batch("PRAGMA user_version=5; COMMIT;")?; let vault_id: String = db .query_row("SELECT id FROM identity", [], |r| r.get(0)) .optional()? @@ -429,7 +430,41 @@ impl Workspace { if current != expected { return Err(HostError::new("REVISION_CONFLICT")); } - let previous = self.entry(path)?; + let mut previous = self.entry(path)?; + if let Some(id) = identity { + // Tombstone metadata can yield its old path to a new remote identity. + if let Some(retired) = previous + .as_ref() + .filter(|entry| entry.deleted && entry.file_id != id) + { + self.db.execute( + "UPDATE files SET path=?2 WHERE id=?1 AND deleted=1", + params![ + retired.file_id, + format!(".ainote/retired/{}", retired.file_id) + ], + )?; + previous = None; + } + if previous.is_none() { + let retired: Option<(String, bool)> = self + .db + .query_row("SELECT path,deleted FROM files WHERE id=?1", [id], |r| { + Ok((r.get(0)?, r.get(1)?)) + }) + .optional()?; + if let Some((_, deleted)) = retired { + if !deleted { + return Err(HostError::new("PATH_CONFLICT")); + } + self.db.execute( + "UPDATE files SET path=?2 WHERE id=?1 AND deleted=1", + params![id, path], + )?; + previous = self.entry(path)?; + } + } + } if identity.is_some_and(|id| previous.as_ref().is_some_and(|entry| entry.file_id != id)) { return Err(HostError::new("PATH_CONFLICT")); } diff --git a/frontend/src-tauri/tests/sync_push.rs b/frontend/src-tauri/tests/sync_push.rs index 0c6ba2d..5ddc86c 100644 --- a/frontend/src-tauri/tests/sync_push.rs +++ b/frontend/src-tauri/tests/sync_push.rs @@ -220,4 +220,87 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate .cursor, 21 ); + // All three explicit choices converge; the local copy gets an independent file ID. + for (iteration, choice) in ["local", "remote", "copy"].into_iter().enumerate() { + if iteration > 0 { + for (ws, content) in [(&workspace, "next-a"), (&workspace_b, "next-b")] { + let mut ws = ws.lock().unwrap(); + let current = ws.read("note.md").unwrap(); + ws.write("note.md", ¤t.entry.hash, content.as_bytes(), "local") + .unwrap(); + } + assert!(client.push_one(&workspace, &binding).await.unwrap()); + assert_eq!( + client_b.pull_page(&workspace_b, &binding_b).await.unwrap(), + 1 + ); + } + { + let mut ws = workspace_b.lock().unwrap(); + let conflict = ws.sync_conflicts(&binding_b.id).unwrap().remove(0); + let sequence = conflict["sequence"].as_i64().unwrap(); + let expected = ws.read("note.md").unwrap().entry.hash; + assert_eq!( + ws.sync_resolve( + &binding_b.id, + sequence, + choice, + if choice == "copy" { "copy.md" } else { "" }, + "wrong" + ) + .unwrap_err() + .code, + "REVISION_CONFLICT" + ); + ws.sync_resolve( + &binding_b.id, + sequence, + choice, + if choice == "copy" { "copy.md" } else { "" }, + &expected, + ) + .unwrap(); + assert!(ws.sync_conflicts(&binding_b.id).unwrap().is_empty()); + // Repeating a persisted decision is harmless. + ws.sync_resolve( + &binding_b.id, + sequence, + choice, + if choice == "copy" { "copy.md" } else { "" }, + &expected, + ) + .unwrap(); + } + while client_b.push_one(&workspace_b, &binding_b).await.unwrap() {} + client.pull_page(&workspace, &binding).await.unwrap(); + client_b.pull_page(&workspace_b, &binding_b).await.unwrap(); + let expected = if choice == "local" { + "offline-b" + } else { + "next-a" + }; + assert_eq!( + workspace.lock().unwrap().read("note.md").unwrap().content, + expected + ); + assert_eq!( + workspace_b.lock().unwrap().read("note.md").unwrap().content, + expected + ); + if choice == "copy" { + let copy = workspace.lock().unwrap().read("copy.md").unwrap(); + assert_eq!(copy.content, "next-b"); + assert_ne!(copy.entry.file_id, first.file_id); + assert_eq!( + copy.entry.file_id, + workspace_b + .lock() + .unwrap() + .read("copy.md") + .unwrap() + .entry + .file_id + ); + } + } }