diff --git a/backend/app/services/desktop_projection.py b/backend/app/services/desktop_projection.py index 8f9a709..27e0e31 100644 --- a/backend/app/services/desktop_projection.py +++ b/backend/app/services/desktop_projection.py @@ -35,6 +35,11 @@ def _refresh(): removed = set(old) - {entry['file_id'] for entry in current} # Content is verified before starting the projection transaction. No model/network IO inside. with transaction(conn): + task_links = [] + for entry in current: + for alias in entry.get('aliases', []): + if alias in removed: + task_links.extend((entry['file_id'], row['task_id']) for row in conn.execute('SELECT task_id FROM tasks WHERE note_id=?', [alias])) for file_id in removed: for block_id in repository.delete_note(file_id, conn=conn): conn.execute('DELETE FROM vec_blocks WHERE block_id=?', [block_id]) @@ -47,6 +52,8 @@ def _refresh(): conn.execute('DELETE FROM vec_blocks WHERE block_id=?', [block_id]) conn.execute('UPDATE blocks SET embedding_local_only=? WHERE note_id=?', (int(parsed.embedding_local_only), parsed.note_id)) conn.execute('INSERT OR REPLACE INTO host_projection VALUES (?,?,?)', (parsed.note_id, document['hash'], parsed.file_path)) + for file_id, task_id in task_links: + conn.execute('UPDATE tasks SET note_id=? WHERE task_id=? AND note_id IS NULL', [file_id, task_id]) if removed or changed: repository.set_index_meta({'workspace_vectors_pending': '1'}, conn=conn) finally: diff --git a/backend/tests/test_desktop_projection.py b/backend/tests/test_desktop_projection.py index 42f1a9f..8650b29 100644 --- a/backend/tests/test_desktop_projection.py +++ b/backend/tests/test_desktop_projection.py @@ -74,3 +74,23 @@ def test_desktop_semantic_rebuild_preserves_host_file_id(tmp_path, monkeypatch): assert [record.note_id for record in repository.list_note_locations()] == ['stable-host-id'] finally: host_bridge.vault_id.reset(token) + + +def test_host_identity_adoption_keeps_existing_task_links(tmp_path, monkeypatch): + settings = replace(get_settings(), environment='desktop', data_dir=tmp_path, db_path=tmp_path/'global.sqlite3') + monkeypatch.setattr(db, 'get_settings', lambda: settings) + monkeypatch.setattr('app.config.get_settings', lambda: settings) + document = {'file_id': 'before-merge', 'path': 'same.md', 'hash': sha256(b'test').hexdigest(), 'content': 'test', 'created_at': 0, 'updated_at': 1} + monkeypatch.setattr(desktop_notes, 'call', lambda method, **params: {'items': [document], 'total': 1} if method == 'list' else document) + from app.services import task_service + token = host_bridge.vault_id.set(str(uuid4())) + try: + task = task_service.create_task(title='Preserve link', note_id='before-merge') + document['file_id'] = 'after-merge' + document['aliases'] = ['before-merge'] + asyncio.run(desktop_projection.refresh()) + assert task_service.get_task(task.task_id).note_id == 'after-merge' + assert repository.get_note_record('before-merge') is None + assert repository.get_note_record('after-merge') is not None + finally: + host_bridge.vault_id.reset(token) diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 00b69dc..63a8f3a 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -79,3 +79,14 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写 - 此测试尚未覆盖完整 S-02 的每个 pull 边界 kill 20 轮，也不代表四并发服务 RSS 或生产 MinIO 性能验收。外部 rename 的稳定身份识别和目录大规模扫描优化继续实施。 - 本增量 Rust desktop 全目标 46 项通过、1 个受父测试驱动的独立进程辅助入口标记 ignored（父测试实际运行并强杀该入口 10 次）；Clippy `-D warnings` 通过。 + + +## 增量：非空初始合并与文件身份关联 + +- 同名独立创建的文件支持保留本地、采用远端、另存副本；先通过 journal 留存旧身份内容，再采用远端稳定身份，旧队列封存。三种选择及中途重开各 20 轮通过，真实双客户端也完成了同名创建后的收敛。 +- schema 8 保存固定远端初始快照。设置页展示上传/下载/相同/冲突预览，每页 100 条；服务器、账户、远端或本地 Vault 变化会清除预览。确认时复核摘要，发生变化返回 SYNC_PREVIEW_CHANGED。初始同步只应用各文件最新可见版本，不重放历史旧路径；历史 tombstone 仅初始化远端基线，不删除本地文件。 +- 初始文件全部落盘或持久保留冲突后，cursor 才一次推进到快照 boundary；任何未完成初始快照都禁止 push。逐文件提交前后重启各 20 轮通过，真实第三个 Workspace 完成相同内容身份采纳、本地独有上传、远端独有下载和冲突解决。 +- Host 保存身份别名，旧文件引用仍可解析；Core 每个 Vault 的投影在身份采纳时同事务迁移本地任务链接，避免外键删除把任务 note_id 清空。新增后端关联测试通过。 +- 目前快照最多处理 100000 条历史 revision，超限明确拒绝；大库快照/扫描性能、任务和配置逻辑同步、完整分类故障矩阵及签名/沙箱等其他生产门槛继续实施。 + +- 本增量后端全量 900 项、Rust desktop 全目标 49 项通过（另 1 个父测试使用的进程辅助入口）；Sync 界面 3 项交互测试、两个 TypeScript 项目检查和 Clippy `-D warnings` 通过。 diff --git a/frontend/src-tauri/build.rs b/frontend/src-tauri/build.rs index 1589d5d..798a948 100644 --- a/frontend/src-tauri/build.rs +++ b/frontend/src-tauri/build.rs @@ -20,6 +20,7 @@ fn main() { "sync_vaults", "sync_create_vault", "sync_bind", + "sync_preview", "sync_unbind", "sync_pause", "sync_status", diff --git a/frontend/src-tauri/capabilities/main.json b/frontend/src-tauri/capabilities/main.json index 26f6e08..2b7cc1c 100644 --- a/frontend/src-tauri/capabilities/main.json +++ b/frontend/src-tauri/capabilities/main.json @@ -46,6 +46,7 @@ "allow-sync-status", "allow-sync-resolve", "allow-sync-logout", - "allow-sync-run" + "allow-sync-run", + "allow-sync-preview" ] } diff --git a/frontend/src-tauri/permissions/autogenerated/sync_preview.toml b/frontend/src-tauri/permissions/autogenerated/sync_preview.toml new file mode 100644 index 0000000..6d5ddfd --- /dev/null +++ b/frontend/src-tauri/permissions/autogenerated/sync_preview.toml @@ -0,0 +1,11 @@ +# Automatically generated - DO NOT EDIT! + +[[permission]] +identifier = "allow-sync-preview" +description = "Enables the sync_preview command without any pre-configured scope." +commands.allow = ["sync_preview"] + +[[permission]] +identifier = "deny-sync-preview" +description = "Denies the sync_preview command without any pre-configured scope." +commands.deny = ["sync_preview"] diff --git a/frontend/src-tauri/src/lib.rs b/frontend/src-tauri/src/lib.rs index da5259f..a4698b9 100644 --- a/frontend/src-tauri/src/lib.rs +++ b/frontend/src-tauri/src/lib.rs @@ -15,6 +15,7 @@ pub mod sync_auth; pub mod sync_client; pub mod sync_discovery; pub mod sync_inbox; +pub mod sync_initial; pub mod sync_resolution; pub mod sync_state; pub mod workspace; diff --git a/frontend/src-tauri/src/main.rs b/frontend/src-tauri/src/main.rs index e2ef7c1..b2a4e6b 100644 --- a/frontend/src-tauri/src/main.rs +++ b/frontend/src-tauri/src/main.rs @@ -835,6 +835,7 @@ fn main() { sync_vaults, sync_create_vault, sync_bind, + sync_preview, sync_unbind, sync_pause, sync_status, diff --git a/frontend/src-tauri/src/sync_client.rs b/frontend/src-tauri/src/sync_client.rs index 6506c2c..85beb46 100644 --- a/frontend/src-tauri/src/sync_client.rs +++ b/frontend/src-tauri/src/sync_client.rs @@ -222,6 +222,50 @@ impl SyncClient { } Ok(()) } + pub async fn snapshot(&self, remote_vault: &str) -> Result { + identifier(remote_vault)?; + let mut cursor = 0i64; + let mut boundary = None; + let mut heads = std::collections::BTreeMap::new(); + loop { + let mut path = + format!("sync/v1/vaults/{remote_vault}/changes?cursor={cursor}&limit=500"); + if let Some(end) = boundary { + path.push_str(&format!("&boundary={end}")); + } + let page = self.json(Method::GET, &path, None).await?; + let end = page["boundary"] + .as_i64() + .ok_or_else(|| SyncError::new("SYNC_RESPONSE_INVALID"))?; + if end < cursor || boundary.is_some_and(|old| old != end) || end > 100000 { + return Err(SyncError::new("SYNC_SNAPSHOT_LIMIT")); + } + boundary = Some(end); + let items = page["items"] + .as_array() + .filter(|v| v.len() <= 500) + .ok_or_else(|| SyncError::new("SYNC_RESPONSE_INVALID"))?; + if items.is_empty() && cursor != end { + return Err(SyncError::new("SYNC_RESPONSE_INVALID")); + } + for item in items { + let revision: crate::sync_inbox::RemoteRevision = + serde_json::from_value(item.clone()) + .map_err(|_| SyncError::new("SYNC_RESPONSE_INVALID"))?; + if revision.sequence != cursor + 1 || revision.sequence > end { + return Err(SyncError::new("SYNC_RESPONSE_INVALID")); + } + cursor = revision.sequence; + heads.insert(revision.file_id.clone(), revision); + } + if cursor == end { + return Ok(crate::sync_initial::Snapshot { + boundary: end, + items: heads.into_values().collect(), + }); + } + } + } pub async fn verify_empty(&self, remote_vault: &str) -> Result<()> { identifier(remote_vault)?; let page = self @@ -248,6 +292,12 @@ impl SyncClient { return Err(SyncError::new("SYNC_BINDING_CHANGED")); } identifier(&binding.remote_vault)?; + if workspace + .access(|ws| ws.sync_initial_pending(&binding.id))? + .is_some() + { + return Ok(false); + } let job = workspace.access(|ws| { ws.sync_resume_resolutions(&binding.id)?; ws.sync_capture(&binding.id)?; @@ -290,6 +340,20 @@ impl SyncClient { while ws.sync_apply_pending(&binding.id)? {} Ok(()) })?; + if let Some(initial) = workspace.access(|ws| ws.sync_initial_pending(&binding.id))? { + let count = initial.len().min(100); + for revision in initial.iter().take(count) { + if revision.operation == "put" { + self.download(workspace, binding, revision).await?; + } + workspace.access(|ws| { + ws.sync_stage(&binding.id, revision)?; + ws.sync_apply_pending(&binding.id)?; + Ok(()) + })?; + } + return Ok(count); + } let (cursor, boundary) = workspace.access(|ws| { ws.check_binding(&binding.id)?; Ok(( diff --git a/frontend/src-tauri/src/sync_commands.rs b/frontend/src-tauri/src/sync_commands.rs index d53ddfc..7f32bc0 100644 --- a/frontend/src-tauri/src/sync_commands.rs +++ b/frontend/src-tauri/src/sync_commands.rs @@ -119,6 +119,37 @@ pub struct Bind { account: String, remote_vault: String, mode: String, + #[serde(default)] + fingerprint: Option, +} +#[tauri::command] +pub async fn sync_preview( + host: State<'_, Host>, + request: Bind, +) -> Result { + let _guard = host.sync.gate.lock().await; + let client = sync_auth::client( + &host.credentials, + &request.endpoint, + &request.account, + false, + ) + .await + .map_err(|e| e.code)?; + let snapshot = sync_auth::guarded(&host.credentials, client.snapshot(&request.remote_vault)) + .await + .map_err(|e| e.code)?; + with_workspace(&host, |ws| { + if ws.vault_id != request.vault_id { + return Err(notesagent_host::workspace::HostError::new("VAULT_CHANGED")); + } + ws.sync_preview( + &request.endpoint, + &request.remote_vault, + &request.account, + &snapshot, + ) + }) } #[tauri::command] pub async fn sync_bind(host: State<'_, Host>, request: Bind) -> Result { @@ -131,6 +162,24 @@ pub async fn sync_bind(host: State<'_, Host>, request: Bind) -> Result end) @@ -177,18 +183,38 @@ impl Workspace { fn sync_finish(&mut self, binding: &str, revision: &RemoteRevision, state: &str) -> Result<()> { self.check_binding(binding)?; let tx = self.db.transaction()?; - let changed = tx.execute( - "UPDATE sync_bindings SET cursor=?2 WHERE id=?1 AND state='active' AND cursor=?3", - params![binding, revision.sequence, revision.sequence - 1], - )?; - if changed != 1 { - return Err(HostError::new("SYNC_CURSOR_INVALID")); + let initial: Option = tx + .query_row( + "SELECT boundary FROM sync_initial WHERE binding=?1", + [binding], + |r| r.get(0), + ) + .optional()?; + if initial.is_none() { + let changed = tx.execute( + "UPDATE sync_bindings SET cursor=?2 WHERE id=?1 AND state='active' AND cursor=?3", + params![binding, revision.sequence, revision.sequence - 1], + )?; + if changed != 1 { + return Err(HostError::new("SYNC_CURSOR_INVALID")); + } } tx.execute("INSERT INTO sync_heads VALUES (?1,?2,?3,?4,?5) ON CONFLICT(binding,file_id) DO UPDATE SET revision=excluded.revision,path=excluded.path,hash=excluded.hash WHERE sync_heads.revision = 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), + ) + .optional()?; + if known.is_some_and(|head| head >= revision.sequence) { + self.sync_finish(binding, &revision, "applied")?; + 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 local = self.resolve(path)?; diff --git a/frontend/src-tauri/src/sync_initial.rs b/frontend/src-tauri/src/sync_initial.rs new file mode 100644 index 0000000..ea73026 --- /dev/null +++ b/frontend/src-tauri/src/sync_initial.rs @@ -0,0 +1,372 @@ +//! Initial merge uses a confirmed fixed remote snapshot and never replays obsolete paths. +use crate::{ + sync_inbox::RemoteRevision, + sync_state::Binding, + workspace::{hash, HostError, Result, Workspace}, +}; +use rusqlite::{params, OptionalExtension}; +use serde::{Deserialize, Serialize}; +use std::{collections::BTreeMap, fs}; +use uuid::Uuid; +#[derive(Clone, Serialize, Deserialize)] +pub struct Snapshot { + pub boundary: i64, + pub items: Vec, +} +#[derive(Serialize)] +pub struct Preview { + pub fingerprint: String, + pub boundary: i64, + pub items: Vec, +} +#[derive(Serialize)] +pub struct PreviewItem { + pub path: String, + pub action: String, +} +#[derive(Serialize)] +struct Local { + path: String, + hash: String, + size: usize, +} +impl Workspace { + fn initial_local(&self) -> Result> { + self.sync_paths()? + .into_iter() + .map(|path| { + let source = self.resolve(&path)?; + if fs::metadata(&source)?.len() > 104857600 { + return Err(HostError::new("FILE_TOO_LARGE")); + } + let bytes = fs::read(source)?; + Ok(Local { + path, + hash: hash(&bytes), + size: bytes.len(), + }) + }) + .collect() + } + pub fn sync_preview( + &self, + endpoint: &str, + remote: &str, + account: &str, + snapshot: &Snapshot, + ) -> Result { + if self.sync_binding()?.is_some() { + return Err(HostError::new("SYNC_ALREADY_BOUND")); + } + let local = self.initial_local()?; + let binding = Binding { + id: String::new(), + endpoint: endpoint.into(), + remote_vault: remote.into(), + account: account.into(), + cursor: 0, + }; + let mut paths = BTreeMap::new(); + for item in &snapshot.items { + item.validate(&binding)?; + self.resolve(&item.path)?; + if item.sequence > snapshot.boundary { + return Err(HostError::new("SYNC_RESPONSE_INVALID")); + } + if item.operation == "put" { + paths.insert(item.path.clone(), "download"); + } + } + for item in &local { + let remote = snapshot + .items + .iter() + .find(|v| v.path == item.path && v.operation == "put"); + paths.insert( + item.path.clone(), + match remote { + Some(r) if r.operation == "put" && r.hash.as_deref() == Some(&item.hash) => { + "identical" + } + Some(_) => "conflict", + None => "upload", + }, + ); + } + let fingerprint = hash( + &serde_json::to_vec(&(&self.vault_id, endpoint, remote, account, &local, snapshot)) + .map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?, + ); + Ok(Preview { + fingerprint, + boundary: snapshot.boundary, + items: paths + .into_iter() + .map(|(path, action)| PreviewItem { + path, + action: action.into(), + }) + .collect(), + }) + } + pub fn sync_bind_initial( + &mut self, + endpoint: &str, + remote: &str, + account: &str, + snapshot: &Snapshot, + expected: &str, + ) -> Result { + if self + .sync_preview(endpoint, remote, account, snapshot)? + .fingerprint + != expected + { + return Err(HostError::new("SYNC_PREVIEW_CHANGED")); + } + let local = self.initial_local()?; + let mut prepared = Vec::new(); + for item in local { + let operation = Uuid::new_v4().to_string(); + let bytes = fs::read(self.resolve(&item.path)?)?; + if hash(&bytes) != item.hash { + return Err(HostError::new("SYNC_PREVIEW_CHANGED")); + } + self.store_payload(&operation, &bytes)?; + let old = self.entry(&item.path)?; + let remote = snapshot + .items + .iter() + .find(|r| r.path == item.path && r.operation == "put"); + let file_id = remote + .map(|r| r.file_id.clone()) + .or_else(|| old.as_ref().map(|v| v.file_id.clone())) + .unwrap_or_else(|| Uuid::new_v4().to_string()); + prepared.push((item, operation, old, file_id, remote)); + } + let id = Uuid::new_v4().to_string(); + let tx = self.db.transaction()?; + tx.execute( + "UPDATE outbox SET state='archived' WHERE state IN ('pending','queued')", + [], + )?; + tx.execute( + "INSERT INTO sync_bindings VALUES (?1,?2,?3,?4,'active',0)", + params![id, endpoint, remote, account], + )?; + tx.execute( + "INSERT INTO sync_initial VALUES (?1,?2)", + params![id, snapshot.boundary], + )?; + for remote in &snapshot.items { + if remote.operation == "put" { + tx.execute( + "INSERT INTO sync_initial_items VALUES (?1,?2,?3)", + params![ + id, + remote.sequence, + serde_json::to_string(remote) + .map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))? + ], + )?; + } else { + tx.execute( + "INSERT INTO sync_heads VALUES (?1,?2,?3,?4,'')", + params![id, remote.file_id, remote.sequence, remote.path], + )?; + } + } + for (item, operation, old, file_id, remote) in prepared { + if let Some(old) = old { + if old.file_id != file_id { + let occupied: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM files WHERE id=?1)", + [&file_id], + |r| r.get(0), + )?; + if occupied { + return Err(HostError::new("SYNC_IDENTITY_CONFLICT")); + } + tx.execute( + "UPDATE files SET id=?1 WHERE id=?2", + params![file_id, old.file_id], + )?; + tx.execute( + "UPDATE file_aliases SET file_id=?1 WHERE file_id=?2", + params![file_id, old.file_id], + )?; + tx.execute( + "INSERT OR REPLACE INTO file_aliases VALUES (?1,?2)", + params![old.file_id, file_id], + )?; + tx.execute("DELETE FROM sync_observed WHERE file_id=?1", [&old.file_id])?; + } + } + tx.execute("INSERT INTO files VALUES (?1,?2,?3,1,0) ON CONFLICT(path) DO UPDATE SET hash=excluded.hash,deleted=0",params![file_id,item.path,item.hash])?; + tx.execute("INSERT INTO sync_observed VALUES (?1,?2,?3,0) ON CONFLICT(file_id) DO UPDATE SET path=excluded.path,hash=excluded.hash,deleted=0",params![file_id,item.path,item.hash])?; + if let Some(remote) = remote.filter(|r| r.hash.as_deref() == Some(&item.hash)) { + tx.execute( + "INSERT INTO sync_heads VALUES (?1,?2,?3,?4,?5)", + params![id, file_id, remote.sequence, item.path, item.hash], + )?; + } else { + tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',X'','pending' FROM files WHERE id=?2",params![operation,file_id])?; + } + } + if snapshot.items.iter().all(|item| item.operation == "delete") { + tx.execute( + "UPDATE sync_bindings SET cursor=?2 WHERE id=?1", + params![id, snapshot.boundary], + )?; + tx.execute("DELETE FROM sync_initial WHERE binding=?1", [&id])?; + } + tx.commit()?; + self.sync_binding()? + .ok_or_else(|| HostError::new("DATABASE_ERROR")) + } + pub fn sync_initial_pending(&self, binding: &str) -> Result>> { + self.check_binding(binding)?; + let active: bool = self.db.query_row( + "SELECT EXISTS(SELECT 1 FROM sync_initial WHERE binding=?1)", + [binding], + |r| r.get(0), + )?; + if !active { + return Ok(None); + } + let mut statement=self.db.prepare("SELECT i.revision FROM sync_initial_items i WHERE binding=?1 AND NOT EXISTS(SELECT 1 FROM sync_inbox n WHERE n.binding=i.binding AND n.sequence=i.sequence AND n.state!='pending') ORDER BY sequence")?; + let rows = statement + .query_map([binding], |r| r.get::<_, String>(0))? + .collect::, _>>()?; + Ok(Some( + rows.iter() + .map(|v| { + serde_json::from_str(v).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID")) + }) + .collect::>()?, + )) + } + pub(crate) fn initial_revision(&self, binding: &str, sequence: i64) -> Result> { + Ok(self.db.query_row("SELECT revision FROM sync_initial_items WHERE binding=?1 AND sequence=?2 AND EXISTS(SELECT 1 FROM sync_initial WHERE binding=?1)",params![binding,sequence],|r|r.get(0)).optional()?) + } +} + +#[cfg(test)] +mod tests { + use super::*; + fn revision(sequence: i64, path: &str, content: &[u8]) -> RemoteRevision { + RemoteRevision { + vault_id: "remote".into(), + sequence, + file_id: Uuid::new_v4().to_string(), + base_revision: 0, + path: path.into(), + operation: "put".into(), + hash: Some(hash(content)), + size: content.len() as i64, + operation_id: Uuid::new_v4().to_string(), + } + } + #[test] + fn initial_snapshot_cursor_waits_for_all_files_and_recovers_twenty_rounds() { + for _ in 0..20 { + for committed in [false, true] { + let root = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(root.path()).unwrap(); + ws.write("same.md", "", b"same", "local").unwrap(); + ws.write("local.md", "", b"local", "local").unwrap(); + let same = revision(2, "same.md", b"same"); + let new = revision(4, "remote.md", b"remote"); + let mut deleted = revision(5, "old.md", b""); + deleted.operation = "delete".into(); + deleted.hash = None; + let snapshot = Snapshot { + boundary: 5, + items: vec![same.clone(), new.clone(), deleted], + }; + let preview = ws + .sync_preview("https://sync.example", "remote", "account", &snapshot) + .unwrap(); + let binding = ws + .sync_bind_initial( + "https://sync.example", + "remote", + "account", + &snapshot, + &preview.fingerprint, + ) + .unwrap(); + ws.sync_stage(&binding.id, &same).unwrap(); + ws.sync_apply_pending(&binding.id).unwrap(); + assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 0); + ws.sync_store_bytes(b"remote").unwrap(); + ws.sync_stage(&binding.id, &new).unwrap(); + if committed { + let operation: String = ws + .db + .query_row( + "SELECT operation_id FROM sync_inbox WHERE sequence=4", + [], + |r| r.get(0), + ) + .unwrap(); + ws.write_with_identity( + "remote.md", + "", + b"remote", + "remote", + &operation, + Some(&new.file_id), + ) + .unwrap(); + } + drop(ws); + let mut ws = Workspace::open(root.path()).unwrap(); + assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 0); + ws.sync_apply_pending(&binding.id).unwrap(); + assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 5); + assert!(ws.sync_initial_pending(&binding.id).unwrap().is_none()); + assert_eq!(ws.read("same.md").unwrap().entry.file_id, same.file_id); + assert_eq!(ws.read("remote.md").unwrap().content, "remote"); + assert_eq!(ws.pending_count().unwrap(), 1); + let deletes: i64 = ws + .db + .query_row( + "SELECT count(*) FROM outbox WHERE operation='delete'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(deletes, 0); + } + } + } + #[test] + fn stale_preview_is_rejected_without_binding_or_file_changes() { + let root = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(root.path()).unwrap(); + let snapshot = Snapshot { + boundary: 0, + items: vec![], + }; + let preview = ws + .sync_preview("https://sync.example", "remote", "account", &snapshot) + .unwrap(); + ws.write("new.md", "", b"new", "local").unwrap(); + assert_eq!( + ws.sync_bind_initial( + "https://sync.example", + "remote", + "account", + &snapshot, + &preview.fingerprint + ) + .err() + .unwrap() + .code, + "SYNC_PREVIEW_CHANGED" + ); + assert!(ws.sync_binding().unwrap().is_none()); + assert_eq!(ws.read("new.md").unwrap().content, "new"); + } +} diff --git a/frontend/src-tauri/src/sync_resolution.rs b/frontend/src-tauri/src/sync_resolution.rs index 89b051d..19f885f 100644 --- a/frontend/src-tauri/src/sync_resolution.rs +++ b/frontend/src-tauri/src/sync_resolution.rs @@ -45,9 +45,6 @@ impl Workspace { 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")); @@ -112,24 +109,41 @@ impl Workspace { .is_some_and(|value| value["state"] == "committed") { let source = self.resolve(&path)?; - let current = if source.is_file() { + let mut current = if source.is_file() { hash(&fs::read(source)?) } else { String::new() }; - if current != expected { + let retired = self.operation(&rename)?.is_some_and(|value| { + value["state"] == "committed" && value["result"]["deleted"] == true + }); + if current != if retired { "" } else { &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 self + .entry(&path)? + .is_some_and(|entry| !entry.deleted && entry.file_id != revision.file_id) + { + self.mutate_with_origin("delete", &path, "", ¤t, &rename, "remote")?; + current.clear(); + } if choice == "local" { - if current.is_empty() { + if expected.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)?; + self.write_with_identity( + &path, + ¤t, + &content, + "local", + &operation, + Some(&revision.file_id), + )?; } } else if revision.operation == "delete" { if !current.is_empty() { @@ -164,7 +178,24 @@ impl Workspace { )?; } } + let retired = self.operation(&rename)?.and_then(|value| { + (value["state"] == "committed" && value["result"]["deleted"] == true) + .then(|| value["result"]["file_id"].as_str().map(str::to_owned)) + .flatten() + }); 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], + )?; + 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])?; + } 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( @@ -422,4 +453,70 @@ mod tests { assert_eq!(payload["operation"], "delete"); assert_eq!(ws.pending_count().unwrap(), 1); } + #[test] + fn same_path_independent_identities_resolve_and_recover_twenty_rounds() { + for _ in 0..20 { + for choice in ["local", "remote", "copy"] { + for crash in [false, true] { + let root = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(root.path()).unwrap(); + let local = ws.write("a.md", "", b"local", "local").unwrap(); + let binding = ws + .sync_bind_empty("https://sync.example", "remote-vault", "account") + .unwrap(); + let 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"remote").unwrap()), + size: 6, + operation_id: Uuid::new_v4().to_string(), + }; + receive(&mut ws, &binding.id, &remote); + if crash { + 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(); + if choice == "copy" { + ws.write_operation("copy.md", "", b"local", "local", ©) + .unwrap(); + } + ws.mutate_with_origin("delete", "a.md", "", &local.hash, &rename, "remote") + .unwrap(); + drop(ws); + ws = Workspace::open(root.path()).unwrap(); + ws.sync_resume_resolutions(&binding.id).unwrap(); + } else { + ws.sync_resolve( + &binding.id, + 1, + choice, + if choice == "copy" { "copy.md" } else { "" }, + &local.hash, + ) + .unwrap(); + } + let actual = ws.read("a.md").unwrap(); + assert_eq!(actual.entry.file_id, remote.file_id); + assert_eq!( + actual.content, + if choice == "local" { "local" } else { "remote" } + ); + assert!(ws.sync_conflicts(&binding.id).unwrap().is_empty()); + assert_eq!(ws.path_for_id(&local.file_id).unwrap(), "a.md"); + ws.sync_capture(&binding.id).unwrap(); + let job = ws.sync_next(&binding.id).unwrap(); + if choice == "remote" { + assert!(job.is_none()); + } else { + assert_ne!(job.unwrap().file_id, local.file_id); + } + } + } + } + } } diff --git a/frontend/src-tauri/src/sync_state.rs b/frontend/src-tauri/src/sync_state.rs index 0a63c83..8f2d169 100644 --- a/frontend/src-tauri/src/sync_state.rs +++ b/frontend/src-tauri/src/sync_state.rs @@ -182,7 +182,7 @@ impl Workspace { } pub fn sync_next(&self, binding: &str) -> Result> { self.check_binding(binding)?; - let job=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| { + let job=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 OR c.local_path=sync_jobs.path) 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()?; if job diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index 4a7fd9f..498367f 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 > 7 { + if version > 8 { return Err(HostError::new("SCHEMA_INCOMPATIBLE")); } - if (1..7).contains(&version) { + if (1..8).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()])?; @@ -157,6 +157,9 @@ impl Workspace { 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 file_aliases (alias TEXT PRIMARY KEY,file_id TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS sync_initial (binding TEXT PRIMARY KEY,boundary INTEGER NOT NULL); + 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_preferences (binding TEXT PRIMARY KEY,paused INTEGER NOT NULL DEFAULT 0); @@ -175,7 +178,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=7; COMMIT;")?; + db.execute_batch("PRAGMA user_version=8; COMMIT;")?; let vault_id: String = db .query_row("SELECT id FROM identity", [], |r| r.get(0)) .optional()? @@ -331,10 +334,19 @@ impl Workspace { Ok(Document { entry, content }) } + pub fn aliases_for_id(&self, file_id: &str) -> Result> { + let mut statement = self + .db + .prepare("SELECT alias FROM file_aliases WHERE file_id=?1 ORDER BY alias")?; + let rows = statement + .query_map([file_id], |r| r.get(0))? + .collect::, _>>()?; + Ok(rows) + } pub fn path_for_id(&self, file_id: &str) -> Result { self.db .query_row( - "SELECT path FROM files WHERE id=?1 AND deleted=0", + "SELECT path FROM files WHERE deleted=0 AND (id=?1 OR id=(SELECT file_id FROM file_aliases WHERE alias=?1)) ORDER BY id=?1 DESC LIMIT 1", [file_id], |row| row.get(0), ) diff --git a/frontend/src-tauri/src/workspace_broker.rs b/frontend/src-tauri/src/workspace_broker.rs index 1579dd5..41ed305 100644 --- a/frontend/src-tauri/src/workspace_broker.rs +++ b/frontend/src-tauri/src/workspace_broker.rs @@ -64,9 +64,19 @@ pub fn dispatch(ws: &mut Workspace, request: &Value) -> Result { entries.retain(|e| !e.deleted && !e.is_folder); entries.sort_by(|a, b| a.path.cmp(&b.path)); let total = entries.len(); - Ok( - json!({"items":entries.into_iter().skip(p.offset).take(p.limit).collect::>(),"total":total}), - ) + let items = entries + .into_iter() + .skip(p.offset) + .take(p.limit) + .map(|entry| { + let aliases = ws.aliases_for_id(&entry.file_id).map_err(|e| e.code)?; + let mut value = serde_json::to_value(entry) + .map_err(|_| "HOST_SERIALIZE_FAILED".to_owned())?; + value["aliases"] = json!(aliases); + Ok(value) + }) + .collect::, String>>()?; + Ok(json!({"items":items,"total":total})) } "workspace.read" => { let p: Read = decode(params)?; diff --git a/frontend/src-tauri/tests/sync_push.rs b/frontend/src-tauri/tests/sync_push.rs index bd2c14a..ef4a894 100644 --- a/frontend/src-tauri/tests/sync_push.rs +++ b/frontend/src-tauri/tests/sync_push.rs @@ -304,6 +304,125 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate } } + workspace + .lock() + .unwrap() + .write("collision.md", "", b"created-a", "local") + .unwrap(); + workspace_b + .lock() + .unwrap() + .write("collision.md", "", b"created-b", "local") + .unwrap(); + client.push_one(&workspace, &binding).await.unwrap(); + client_b.pull_page(&workspace_b, &binding_b).await.unwrap(); + { + let mut ws = workspace_b.lock().unwrap(); + let conflict = ws.sync_conflicts(&binding_b.id).unwrap().remove(0); + ws.sync_resolve( + &binding_b.id, + conflict["sequence"].as_i64().unwrap(), + "local", + "", + conflict["current_hash"].as_str().unwrap(), + ) + .unwrap(); + } + 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(); + assert_eq!( + workspace + .lock() + .unwrap() + .read("collision.md") + .unwrap() + .content, + "created-b" + ); + assert_eq!( + workspace + .lock() + .unwrap() + .read("collision.md") + .unwrap() + .entry + .file_id, + workspace_b + .lock() + .unwrap() + .read("collision.md") + .unwrap() + .entry + .file_id + ); + + // Initial merge reviews a fixed snapshot and preserves conflicting local content. + let merge_root = tempfile::tempdir().unwrap(); + let merge_ws = Arc::new(Mutex::new(Workspace::open(merge_root.path()).unwrap())); + let snapshot = client.snapshot(remote).await.unwrap(); + let merge_binding = { + let mut ws = merge_ws.lock().unwrap(); + ws.write("note.md", "", b"next-a", "local").unwrap(); + ws.write("collision.md", "", b"merge-local", "local") + .unwrap(); + ws.write("local-only.md", "", b"only-local", "local") + .unwrap(); + let preview = ws + .sync_preview(&endpoint, remote, "rust-fixture", &snapshot) + .unwrap(); + assert!(preview + .items + .iter() + .any(|v| v.path == "note.md" && v.action == "identical")); + assert!(preview + .items + .iter() + .any(|v| v.path == "collision.md" && v.action == "conflict")); + ws.sync_bind_initial( + &endpoint, + remote, + "rust-fixture", + &snapshot, + &preview.fingerprint, + ) + .unwrap() + }; + assert!(!client_b.push_one(&merge_ws, &merge_binding).await.unwrap()); + assert_eq!( + client_b.pull_page(&merge_ws, &merge_binding).await.unwrap(), + snapshot.items.len() + ); + { + let mut ws = merge_ws.lock().unwrap(); + assert_eq!(ws.read("collision.md").unwrap().content, "merge-local"); + assert_eq!(ws.read("copy.md").unwrap().content, "next-b"); + assert_eq!( + ws.sync_binding().unwrap().unwrap().cursor, + snapshot.boundary + ); + let conflict = ws.sync_conflicts(&merge_binding.id).unwrap().remove(0); + ws.sync_resolve( + &merge_binding.id, + conflict["sequence"].as_i64().unwrap(), + "remote", + "", + conflict["current_hash"].as_str().unwrap(), + ) + .unwrap(); + } + assert!(client_b.push_one(&merge_ws, &merge_binding).await.unwrap()); + assert!(!client_b.push_one(&merge_ws, &merge_binding).await.unwrap()); + client.pull_page(&workspace, &binding).await.unwrap(); + assert_eq!( + workspace + .lock() + .unwrap() + .read("local-only.md") + .unwrap() + .content, + "only-local" + ); // Kill the actual client process after each durable 10 MiB server offset, // before its response reaches the client. The next process must query offset. use sha2::{Digest, Sha256}; diff --git a/frontend/src/features/settings/SyncSettings.spec.ts b/frontend/src/features/settings/SyncSettings.spec.ts index 35194cb..8097fe2 100644 --- a/frontend/src/features/settings/SyncSettings.spec.ts +++ b/frontend/src/features/settings/SyncSettings.spec.ts @@ -35,3 +35,29 @@ it('cancels destructive choices and binds accepted conflict decisions to their s expect(wrapper.get('input[type=url]').attributes('disabled')).toBeDefined() wrapper.unmount() }) + +it('requires a reviewed merge fingerprint and invalidates preview when the remote changes', async () => { + vi.mocked(hostInvoke).mockImplementation(async command => { + if (command === 'sync_status') return empty() + if (command === 'sync_login') return { endpoint: 'https://test.example/', account: 'test' } + if (command === 'sync_vaults') return { items: [{ id: 'one', name: 'One', sequence: 2 }, { id: 'two', name: 'Two', sequence: 3 }] } + if (command === 'sync_preview') return { fingerprint: 'reviewed-snapshot', boundary: 2, items: [{ path: 'same.md', action: 'conflict' }] } + return null + }) + const wrapper = mount(SyncSettings); await flushPromises() + await wrapper.get('input[type=url]').setValue('https://test.example/') + await wrapper.get('input[autocomplete=username]').setValue('test') + await wrapper.get('input[type=password]').setValue('fixture') + await wrapper.get('form').trigger('submit'); await flushPromises() + await wrapper.get('select').setValue('one') + await wrapper.findAll('button').find(button => button.text() === '预览合并')!.trigger('click'); await flushPromises() + expect(wrapper.text()).toContain('same.md') + await wrapper.get('select').setValue('two') + expect(wrapper.text()).not.toContain('same.md') + await wrapper.get('select').setValue('one') + await wrapper.findAll('button').find(button => button.text() === '预览合并')!.trigger('click'); await flushPromises() + confirm.mockResolvedValue(true) + await wrapper.findAll('button').find(button => button.text() === '确认合并并绑定')!.trigger('click'); await flushPromises() + expect(hostInvoke).toHaveBeenCalledWith('sync_bind', { request: { vault_id: 'local', endpoint: 'https://test.example/', account: 'test', remote_vault: 'one', mode: 'merge', fingerprint: 'reviewed-snapshot' } }) + wrapper.unmount() +}) diff --git a/frontend/src/features/settings/SyncSettings.vue b/frontend/src/features/settings/SyncSettings.vue index 320ea19..9fa5874 100644 --- a/frontend/src/features/settings/SyncSettings.vue +++ b/frontend/src/features/settings/SyncSettings.vue @@ -1,5 +1,5 @@