From 251d3bac7d822fd93a72a1ec54f74821ff6fa578 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Tue, 8 Sep 2026 15:38:10 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8C=81=E4=B9=85=E5=8C=96=20Sync=20?= =?UTF-8?q?=E6=94=B6=E4=BB=B6=E7=AE=B1=E5=B9=B6=E5=9C=A8=E6=8B=89=E5=8F=96?= =?UTF-8?q?=E6=97=B6=E4=BF=9D=E7=95=99=E7=A6=BB=E7=BA=BF=E5=86=B2=E7=AA=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/contracts/Sync-v1契约.md | 6 +- .../OpenNexus生产化实施进度-2026-09-08.md | 2 + frontend/src-tauri/src/lib.rs | 1 + frontend/src-tauri/src/sync_client.rs | 135 ++++++ frontend/src-tauri/src/sync_inbox.rs | 397 ++++++++++++++++++ frontend/src-tauri/src/workspace.rs | 97 ++++- frontend/src-tauri/tests/sync_push.rs | 84 ++++ 7 files changed, 705 insertions(+), 17 deletions(-) create mode 100644 frontend/src-tauri/src/sync_inbox.rs diff --git a/docs/contracts/Sync-v1契约.md b/docs/contracts/Sync-v1契约.md index d1cf361..89ee4f0 100644 --- a/docs/contracts/Sync-v1契约.md +++ b/docs/contracts/Sync-v1契约.md @@ -8,7 +8,11 @@ Workspace schema 3 新增持久绑定、上传作业与远端 heads。首次向 Rust HTTP 客户端已实现握手、登录、空远端复核、1 MiB 分块上传、查询 offset 续传、complete 和 Revision 提交。默认 HTTPS，测试 HTTP 必须显式启用；不跟随重定向，令牌不进入 URL，响应有大小限制。真实本地 HTTP Fixture 已验证 20 次编辑形成 20 个正确远端基线、同一提交重复 100 次无重复 revision。此 Fixture 使用 SQLite/磁盘对象，不替代生产 PostgreSQL/MinIO 验收。 -此增量尚未开放桌面 Sync capability：Stronghold 会话接入、拉取/inbox、冲突 UI、附件故障矩阵及数据分类仍在实施。 +此增量尚未开放桌面 Sync capability：Stronghold 会话接入、冲突解决与 UI、附件故障矩阵及数据分类仍在实施。 + +Workspace schema 4 增加持久 inbox、分页 boundary 和冲突记录，并为文件移动/删除日志记录 local/remote 来源。下载先验证长度与摘要，再登记 inbox；文件操作通过 Workspace journal 应用后才推进 cursor。进程在文件提交与 cursor 提交之间重启时使用操作回执去重。远端新增保留 file_id，远端移动/删除不回流 outbox。本机历史提交回放仅确认操作，不回退新编辑。本地待上传修改或摘要不符产生冲突，保存本地与远端内容引用后推进接收游标；同一文件保留最新待处理冲突，旧冲突记录仍保留。空本地库可以绑定已有远端并仅拉取。 + +真实本地 HTTP 集成已验证第二设备拉取 20 个历史 revision、稳定 file_id、零回流 outbox、本机旧历史不覆盖新编辑，以及离线同改时保留本地内容和远端冲突。重启边界测试验证文件提交前后 cursor 都不会提前推进。冲突解决事务、完整附件故障矩阵和真实双机 UI 验收尚未完成。 ## 身份与数据边界 diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 1e4f17f..9e00bea 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -6,6 +6,8 @@ ## 持续实施增量 +- Rust 同步新增 schema 4 inbox/boundary、稳定远端 file_id、remote 来源的移动/删除日志和冲突保留。真实 HTTP 双客户端测试覆盖历史拉取、零回流、本机提交回放保护和同改冲突；文件提交前后的 inbox 重启测试通过。冲突解决和用户界面正在继续实施，仍不开放完整 Sync capability。 + - Rust 同步上传队列与 HTTP 客户端已实现首批链路：持久绑定、spool、冻结远端基线、offset 查询续传及原子确认。20 次离线编辑/重启重试、解绑隔离测试通过；真实本地 Sync 服务验证 20 条 revision 与 100 次提交重放。Sync capability 仍保持 false，等待双向同步、会话与 UI 完成。 - 桌面全文/向量投影按 Vault 隔离，搜索前经 Host 对账文件摘要；向量重建从 Host 读取正文并保留 file_id。任务和笔记关联使用同一 Vault 的持久库。新增测试覆盖同路径双 Vault 隔离、变更/删除刷新、稳定 ID 和任务跨 Vault 不可见；真实 Core 测试增加全文搜索与删除后的检索验证。 diff --git a/frontend/src-tauri/src/lib.rs b/frontend/src-tauri/src/lib.rs index 4440c1d..0ce43d4 100644 --- a/frontend/src-tauri/src/lib.rs +++ b/frontend/src-tauri/src/lib.rs @@ -10,6 +10,7 @@ mod runtime_compat; pub mod session_lock; #[cfg(feature = "desktop")] pub mod sync_client; +pub mod sync_inbox; 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 b88dcef..b01377e 100644 --- a/frontend/src-tauri/src/sync_client.rs +++ b/frontend/src-tauri/src/sync_client.rs @@ -259,6 +259,141 @@ impl SyncClient { workspace.access(|ws| ws.sync_ack(&job, &revision))?; Ok(true) } + pub async fn pull_page( + &self, + workspace: &impl WorkspaceAccess, + binding: &Binding, + ) -> Result { + if Url::parse(&binding.endpoint) + .ok() + .is_none_or(|url| url != self.endpoint) + { + return Err(SyncError::new("SYNC_BINDING_CHANGED")); + } + identifier(&binding.remote_vault)?; + workspace.access(|ws| { + while ws.sync_apply_pending(&binding.id)? {} + Ok(()) + })?; + let (cursor, boundary) = workspace.access(|ws| { + ws.check_binding(&binding.id)?; + Ok(( + ws.sync_binding()? + .ok_or_else(|| crate::workspace::HostError::new("SYNC_BINDING_CHANGED"))? + .cursor, + ws.sync_boundary(&binding.id)?, + )) + })?; + let mut path = format!( + "sync/v1/vaults/{}/changes?cursor={cursor}&limit=100", + binding.remote_vault + ); + 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"))?; + let items = page["items"] + .as_array() + .filter(|items| items.len() <= 100) + .ok_or_else(|| SyncError::new("SYNC_RESPONSE_INVALID"))?; + if items.is_empty() { + if end != cursor { + return Err(SyncError::new("SYNC_RESPONSE_INVALID")); + } + return Ok(0); + } + workspace.access(|ws| ws.sync_set_boundary(&binding.id, end))?; + for (index, item) in items.iter().enumerate() { + let revision: crate::sync_inbox::RemoteRevision = serde_json::from_value(item.clone()) + .map_err(|_| SyncError::new("SYNC_RESPONSE_INVALID"))?; + revision.validate(binding)?; + if revision.sequence != cursor + index as i64 + 1 || revision.sequence > end { + return Err(SyncError::new("SYNC_RESPONSE_INVALID")); + } + 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(()) + })?; + } + Ok(items.len()) + } + async fn download( + &self, + workspace: &impl WorkspaceAccess, + binding: &Binding, + revision: &crate::sync_inbox::RemoteRevision, + ) -> Result<()> { + let digest = revision + .hash + .as_deref() + .ok_or_else(|| SyncError::new("SYNC_RESPONSE_INVALID"))?; + let target = workspace.access(|ws| { + ws.check_binding(&binding.id)?; + ws.sync_spool(digest) + })?; + if target.exists() { + let bytes = std::fs::read(&target)?; + if bytes.len() as i64 == revision.size && crate::workspace::hash(&bytes) == digest { + return Ok(()); + } + return Err(SyncError::new("SYNC_SPOOL_CORRUPT")); + } + let url = self + .endpoint + .join(&format!( + "sync/v1/vaults/{}/objects/{digest}", + binding.remote_vault + )) + .map_err(|_| SyncError::new("SYNC_PATH_INVALID"))?; + let mut response = self + .client + .get(url) + .bearer_auth(self.token.as_str()) + .send() + .await + .map_err(|_| SyncError::new("SYNC_NETWORK_ERROR"))?; + if !response.status().is_success() { + return Err(SyncError { + code: "SYNC_DOWNLOAD_FAILED".into(), + status: response.status().as_u16(), + retry_after: None, + }); + } + let mut file = tempfile::NamedTempFile::new_in( + target + .parent() + .ok_or_else(|| SyncError::new("SYNC_SPOOL_FAILED"))?, + )?; + let mut hasher = Sha256::new(); + let mut length = 0u64; + while let Some(chunk) = response + .chunk() + .await + .map_err(|_| SyncError::new("SYNC_NETWORK_ERROR"))? + { + length += chunk.len() as u64; + if length > revision.size as u64 { + return Err(SyncError::new("SYNC_OBJECT_CORRUPT")); + } + workspace.access(|ws| ws.check_binding(&binding.id))?; + std::io::Write::write_all(&mut file, &chunk)?; + hasher.update(&chunk); + } + if length != revision.size as u64 || format!("{:x}", hasher.finalize()) != digest { + return Err(SyncError::new("SYNC_OBJECT_CORRUPT")); + } + file.as_file().sync_all()?; + file.persist_noclobber(target) + .map_err(|_| SyncError::new("SYNC_SPOOL_FAILED"))?; + Ok(()) + } async fn upload( &self, workspace: &impl WorkspaceAccess, diff --git a/frontend/src-tauri/src/sync_inbox.rs b/frontend/src-tauri/src/sync_inbox.rs new file mode 100644 index 0000000..7ee9136 --- /dev/null +++ b/frontend/src-tauri/src/sync_inbox.rs @@ -0,0 +1,397 @@ +//! Persist received revisions before Workspace writes; cursor advancement follows application. +use crate::{ + sync_state::{Binding, Job}, + workspace::{hash, HostError, Result, Workspace}, +}; +use rusqlite::{params, OptionalExtension}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::{fs, io::Write}; +use uuid::Uuid; + +#[derive(Clone, Serialize, Deserialize)] +pub struct RemoteRevision { + pub vault_id: String, + pub sequence: i64, + pub file_id: String, + pub base_revision: i64, + pub path: String, + pub operation: String, + pub hash: Option, + pub size: i64, + pub operation_id: String, +} + +impl RemoteRevision { + pub fn validate(&self, binding: &Binding) -> Result<()> { + if self.vault_id != binding.remote_vault + || self.sequence <= 0 + || self.base_revision < 0 + || self.base_revision >= self.sequence + || !(0..=104857600).contains(&self.size) + || self.file_id.len() < 16 + || self.file_id.len() > 80 + || !self + .file_id + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-') + || !matches!(self.operation.as_str(), "put" | "delete") + || (self.operation == "delete" && (self.hash.is_some() || self.size != 0)) + || (self.operation == "put" + && self.hash.as_ref().is_none_or(|h| { + h.len() != 64 + || !h + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + })) + { + return Err(HostError::new("SYNC_RESPONSE_INVALID")); + } + Ok(()) + } +} + +impl Workspace { + pub fn sync_bind_download( + &mut self, + endpoint: &str, + remote_vault: &str, + account: &str, + ) -> Result { + if self.sync_binding()?.is_some() { + return Err(HostError::new("SYNC_ALREADY_BOUND")); + } + if self + .scan()? + .iter() + .any(|entry| !entry.is_folder && !entry.deleted) + { + return Err(HostError::new("SYNC_RECONCILIATION_REQUIRED")); + } + 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_vault, account], + )?; + tx.commit()?; + self.sync_binding()? + .ok_or_else(|| HostError::new("DATABASE_ERROR")) + } + pub fn sync_boundary(&self, binding: &str) -> Result> { + self.check_binding(binding)?; + Ok(self + .db + .query_row( + "SELECT boundary FROM sync_windows WHERE binding=?1", + [binding], + |r| r.get(0), + ) + .optional()?) + } + pub fn sync_set_boundary(&self, binding: &str, boundary: i64) -> Result<()> { + self.check_binding(binding)?; + let cursor = self + .sync_binding()? + .ok_or_else(|| HostError::new("SYNC_BINDING_CHANGED"))? + .cursor; + if boundary < cursor + || self + .sync_boundary(binding)? + .is_some_and(|old| old != boundary) + { + return Err(HostError::new("SYNC_RESPONSE_INVALID")); + } + self.db.execute( + "INSERT OR IGNORE INTO sync_windows VALUES (?1,?2)", + params![binding, boundary], + )?; + Ok(()) + } + pub fn sync_store_bytes(&self, bytes: &[u8]) -> Result { + let digest = hash(bytes); + let path = self.sync_spool(&digest)?; + if path.exists() { + if fs::symlink_metadata(&path)?.file_type().is_symlink() + || hash(&fs::read(&path)?) != digest + { + return Err(HostError::new("SYNC_SPOOL_CORRUPT")); + } + } else { + let mut temp = tempfile::NamedTempFile::new_in(path.parent().unwrap())?; + temp.write_all(bytes)?; + temp.as_file().sync_all()?; + temp.persist_noclobber(path) + .map_err(|_| HostError::new("SYNC_SPOOL_FAILED"))?; + } + Ok(digest) + } + pub fn sync_stage(&self, binding: &str, revision: &RemoteRevision) -> Result<()> { + self.check_binding(binding)?; + let active = self + .sync_binding()? + .ok_or_else(|| HostError::new("SYNC_BINDING_CHANGED"))?; + revision.validate(&active)?; + self.resolve(&revision.path)?; + if revision.sequence != active.cursor + 1 + || self + .sync_boundary(binding)? + .is_none_or(|end| revision.sequence > end) + { + return Err(HostError::new("SYNC_CURSOR_INVALID")); + } + if let Some(digest) = &revision.hash { + let bytes = fs::read(self.sync_spool(digest)?)?; + if bytes.len() as i64 != revision.size || hash(&bytes) != *digest { + return Err(HostError::new("SYNC_SPOOL_CORRUPT")); + } + } + let encoded = + serde_json::to_string(revision).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?; + let existing: Option = self + .db + .query_row( + "SELECT revision FROM sync_inbox WHERE binding=?1 AND sequence=?2", + params![binding, revision.sequence], + |r| r.get(0), + ) + .optional()?; + if existing.is_some_and(|value| value != encoded) { + return Err(HostError::new("SYNC_REVISION_CHANGED")); + } + self.db.execute( + "INSERT OR IGNORE INTO sync_inbox VALUES (?1,?2,?3,?4,?5,'pending')", + params![ + binding, + revision.sequence, + encoded, + Uuid::new_v4().to_string(), + Uuid::new_v4().to_string() + ], + )?; + Ok(()) + } + 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")); + } + 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 Result<()> { + let path = self.resolve(local_path)?; + let digest = if path.is_file() { + self.sync_store_bytes(&fs::read(path)?)? + } else { + String::new() + }; + let remote = + serde_json::to_string(revision).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?; + let tx = self.db.transaction()?; + tx.execute("UPDATE sync_conflicts SET state='superseded' WHERE binding=?1 AND file_id=?2 AND state='open' AND sequence Result { + self.check_binding(binding)?; + let pending: Option<(String,String,String)> = self.db.query_row("SELECT revision,operation_id,rename_id FROM sync_inbox WHERE binding=?1 AND state='pending' ORDER BY sequence LIMIT 1", [binding], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?; + let Some((encoded, operation_id, rename_id)) = pending else { + return Ok(false); + }; + let revision: RemoteRevision = + serde_json::from_str(&encoded).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?; + 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 { + self.sync_ack( + &job, + &serde_json::to_value(&revision) + .map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?, + )?; + self.sync_finish(binding, &revision, "applied")?; + return Ok(true); + } + if self + .operation(&operation_id)? + .is_some_and(|value| value["state"] == "committed") + { + 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)?; + let current = if local.is_file() { + hash(&fs::read(&local)?) + } else { + String::new() + }; + let queued: bool = self.db.query_row("SELECT EXISTS(SELECT 1 FROM outbox WHERE file_id=?1 AND state IN ('pending','queued'))", [&revision.file_id], |r| r.get(0))?; + let head: Option = self + .db + .query_row( + "SELECT hash FROM sync_heads WHERE binding=?1 AND file_id=?2", + params![binding, revision.file_id], + |r| r.get(0), + ) + .optional()?; + 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()); + if conflict { + self.sync_preserve_conflict(binding, &revision, path)?; + return Ok(true); + } + if revision.operation == "delete" { + if local_path.is_some() && local.exists() { + self.mutate_with_origin("delete", path, "", ¤t, &operation_id, "remote")?; + } + } else { + if let Some(previous) = local_path + .as_deref() + .filter(|previous| *previous != revision.path) + { + self.mutate_with_origin( + "rename", + previous, + &revision.path, + ¤t, + &rename_id, + "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_id, + Some(&revision.file_id), + )?; + } + self.sync_finish(binding, &revision, "applied")?; + Ok(true) + } + pub fn sync_conflicts(&self, binding: &str) -> Result> { + self.check_binding(binding)?; + let mut statement = self.db.prepare("SELECT sequence,file_id,local_path,local_hash,remote FROM sync_conflicts WHERE binding=?1 AND state='open' ORDER BY sequence")?; + let rows = statement + .query_map([binding], |r| { + Ok(( + r.get::<_, i64>(0)?, + r.get::<_, String>(1)?, + r.get::<_, String>(2)?, + r.get::<_, String>(3)?, + r.get::<_, String>(4)?, + )) + })? + .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"))?; + Ok(serde_json::json!({"sequence":sequence,"file_id":file_id,"local_path":local_path,"local_hash":local_hash,"remote":remote})) + }).collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn inbox_reopen_before_and_after_file_commit_never_advances_cursor_early() { + for committed in [false, true] { + 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 digest = ws.sync_store_bytes(b"remote-content").unwrap(); + let revision = RemoteRevision { + vault_id: "remote-vault".into(), + sequence: 1, + file_id: file_id.clone(), + base_revision: 0, + path: "nested/a.md".into(), + operation: "put".into(), + hash: Some(digest), + size: 14, + operation_id: Uuid::new_v4().to_string(), + }; + ws.sync_set_boundary(&binding.id, 1).unwrap(); + ws.sync_stage(&binding.id, &revision).unwrap(); + if committed { + let operation: String = ws + .db + .query_row("SELECT operation_id FROM sync_inbox", [], |r| r.get(0)) + .unwrap(); + ws.write_with_identity( + "nested/a.md", + "", + b"remote-content", + "remote", + &operation, + Some(&file_id), + ) + .unwrap(); + } + assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 0); + drop(ws); + let mut ws = Workspace::open(root.path()).unwrap(); + assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 0); + assert!(ws.sync_apply_pending(&binding.id).unwrap()); + assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 1); + let document = ws.read("nested/a.md").unwrap(); + assert_eq!(document.content, "remote-content"); + assert_eq!(document.entry.file_id, file_id); + assert_eq!(document.entry.revision, 1); + assert_eq!(ws.pending_count().unwrap(), 0); + assert!(!ws.sync_apply_pending(&binding.id).unwrap()); + } + } +} diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index 788136c..f92ddcf 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 > 3 { + if version > 4 { return Err(HostError::new("SCHEMA_INCOMPATIBLE")); } - if (1..3).contains(&version) { + if (1..4).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()])?; @@ -154,7 +154,21 @@ impl Workspace { CREATE UNIQUE INDEX IF NOT EXISTS sync_active ON sync_bindings(state) WHERE state='active'; CREATE TABLE IF NOT EXISTS sync_jobs (binding TEXT NOT NULL,operation_id TEXT NOT NULL,file_id TEXT NOT NULL,path TEXT NOT NULL,hash TEXT NOT NULL,size INTEGER NOT NULL,operation TEXT NOT NULL,state TEXT NOT NULL,base_revision INTEGER,upload_id TEXT,remote_revision INTEGER,error TEXT,PRIMARY KEY(binding,operation_id)); 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)); - PRAGMA user_version=3; COMMIT;")?; + 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));")?; + let has_origin: bool = db.query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info('file_ops') WHERE name='origin')", + [], + |r| r.get(0), + )?; + if !has_origin { + db.execute( + "ALTER TABLE file_ops ADD COLUMN origin TEXT NOT NULL DEFAULT 'local'", + [], + )?; + } + db.execute_batch("PRAGMA user_version=4; COMMIT;")?; let vault_id: String = db .query_row("SELECT id FROM identity", [], |r| r.get(0)) .optional()? @@ -359,6 +373,17 @@ impl Workspace { content: &[u8], origin: &str, operation_id: &str, + ) -> Result { + self.write_with_identity(path, expected, content, origin, operation_id, None) + } + pub(crate) fn write_with_identity( + &mut self, + path: &str, + expected: &str, + content: &[u8], + origin: &str, + operation_id: &str, + identity: Option<&str>, ) -> Result { if Uuid::parse_str(operation_id).is_err() { return Err(HostError::new("OPERATION_ID_INVALID")); @@ -404,9 +429,18 @@ impl Workspace { if current != expected { return Err(HostError::new("REVISION_CONFLICT")); } - let file_id = self - .entry(path)? - .map_or_else(|| Uuid::new_v4().to_string(), |e| e.file_id); + let 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")); + } + let file_id = previous.map_or_else( + || { + identity + .map(str::to_owned) + .unwrap_or_else(|| Uuid::new_v4().to_string()) + }, + |entry| entry.file_id, + ); let tx = self.db.transaction()?; tx.execute( "INSERT INTO operations VALUES (?1,?2,'pending',NULL)", @@ -577,6 +611,7 @@ impl Workspace { destination, expected, &Uuid::new_v4().to_string(), + "local", ) } @@ -588,7 +623,19 @@ impl Workspace { expected: &str, operation_id: &str, ) -> Result { - let id = self.prepare_file_op_with_id(kind, path, destination, expected, operation_id)?; + self.mutate_with_origin(kind, path, destination, expected, operation_id, "local") + } + pub(crate) fn mutate_with_origin( + &mut self, + kind: &str, + path: &str, + destination: &str, + expected: &str, + operation_id: &str, + origin: &str, + ) -> Result { + let id = + self.prepare_file_op_with_id(kind, path, destination, expected, operation_id, origin)?; if self .operation(&id)? .is_some_and(|v| v["state"] == "committed") @@ -609,12 +656,16 @@ impl Workspace { destination: &str, expected: &str, id: &str, + origin: &str, ) -> Result { - if !matches!(kind, "rename" | "delete") || Uuid::parse_str(id).is_err() { + if !matches!(origin, "local" | "remote") + || !matches!(kind, "rename" | "delete") + || Uuid::parse_str(id).is_err() + { return Err(HostError::new("INVALID_OPERATION")); } let fingerprint = hash( - &serde_json::to_vec(&(kind, path, destination, expected)) + &serde_json::to_vec(&(kind, path, destination, expected, origin)) .map_err(|_| HostError::new("INVALID_OPERATION"))?, ); let previous: Option = self @@ -657,24 +708,34 @@ impl Workspace { params![id, fingerprint], )?; tx.execute( - "INSERT INTO file_ops VALUES (?1,?2,?3,?4,?5,?6,'pending')", - params![id, kind, path, destination, expected, content], + "INSERT INTO file_ops VALUES (?1,?2,?3,?4,?5,?6,'pending',?7)", + params![id, kind, path, destination, expected, content, origin], )?; tx.commit()?; Ok(id.to_owned()) } fn apply_file_op(&mut self, id: &str) -> Result<()> { - let (kind, path, destination, expected, content): ( + let (kind, path, destination, expected, content, origin): ( String, String, String, String, Vec, + String, ) = self.db.query_row( - "SELECT kind,path,destination,hash,content FROM file_ops WHERE id=?1", + "SELECT kind,path,destination,hash,content,origin FROM file_ops WHERE id=?1", [id], - |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)), + |r| { + Ok(( + r.get(0)?, + r.get(1)?, + r.get(2)?, + r.get(3)?, + r.get(4)?, + r.get(5)?, + )) + }, )?; let source = self.resolve(&path)?; let previous = self @@ -724,13 +785,17 @@ impl Workspace { "UPDATE files SET path=?1,revision=revision+1 WHERE id=?2", params![destination, previous.file_id], )?; - tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE id=?3", params![id,content,previous.file_id])?; + if origin == "local" { + tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE id=?3", params![id,content,previous.file_id])?; + } } else { tx.execute( "UPDATE files SET deleted=1,revision=revision+1 WHERE id=?1", [&previous.file_id], )?; - tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,'','delete',X'','pending' FROM files WHERE id=?2", params![id,previous.file_id])?; + if origin == "local" { + tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,'','delete',X'','pending' FROM files WHERE id=?2", params![id,previous.file_id])?; + } } tx.execute("DELETE FROM file_ops WHERE id=?1", [id])?; let mut result = previous; diff --git a/frontend/src-tauri/tests/sync_push.rs b/frontend/src-tauri/tests/sync_push.rs index 20c12b4..0c6ba2d 100644 --- a/frontend/src-tauri/tests/sync_push.rs +++ b/frontend/src-tauri/tests/sync_push.rs @@ -136,4 +136,88 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate client.verify_empty(remote).await.unwrap_err().code, "SYNC_RECONCILIATION_REQUIRED" ); + let session_b = public + .login( + "rust-fixture", + Zeroizing::new("controlled-fixture-password".into()), + "Device B", + ) + .await + .unwrap(); + let client_b = SyncClient::new( + &endpoint, + Zeroizing::new(session_b.access_token.clone()), + true, + ) + .unwrap(); + let root_b = tempfile::tempdir().unwrap(); + let workspace_b = Arc::new(Mutex::new(Workspace::open(root_b.path()).unwrap())); + let binding_b = workspace_b + .lock() + .unwrap() + .sync_bind_download(&endpoint, remote, "rust-fixture") + .unwrap(); + assert_eq!( + client_b.pull_page(&workspace_b, &binding_b).await.unwrap(), + 20 + ); + assert_eq!( + workspace_b.lock().unwrap().read("note.md").unwrap().content, + "fixture-19" + ); + assert_eq!(workspace_b.lock().unwrap().pending_count().unwrap(), 0); + assert_eq!( + workspace_b + .lock() + .unwrap() + .read("note.md") + .unwrap() + .entry + .file_id, + first.file_id + ); + // Receiving one's historical commits never rolls back newer local edits. + { + let mut ws = workspace.lock().unwrap(); + let current = ws.read("note.md").unwrap(); + ws.write("note.md", ¤t.entry.hash, b"new-a", "local") + .unwrap(); + } + assert_eq!(client.pull_page(&workspace, &binding).await.unwrap(), 20); + assert_eq!( + workspace.lock().unwrap().read("note.md").unwrap().content, + "new-a" + ); + { + let mut ws = workspace_b.lock().unwrap(); + let current = ws.read("note.md").unwrap(); + ws.write("note.md", ¤t.entry.hash, b"offline-b", "local") + .unwrap(); + } + client.push_one(&workspace, &binding).await.unwrap(); + assert_eq!( + client_b.pull_page(&workspace_b, &binding_b).await.unwrap(), + 1 + ); + assert_eq!( + workspace_b.lock().unwrap().read("note.md").unwrap().content, + "offline-b" + ); + let conflicts = workspace_b + .lock() + .unwrap() + .sync_conflicts(&binding_b.id) + .unwrap(); + assert_eq!(conflicts.len(), 1); + assert_eq!(conflicts[0]["remote"]["sequence"], 21); + assert_eq!( + workspace_b + .lock() + .unwrap() + .sync_binding() + .unwrap() + .unwrap() + .cursor, + 21 + ); }