From f08a338ece093c580fbbfcdd53c9a64e1029be23 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Tue, 8 Sep 2026 16:15:09 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=9A=82=E5=AD=98=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E5=8C=BA=E8=BD=BD=E8=8D=B7=E5=B9=B6=E6=8D=95=E8=8E=B7=E5=A4=96?= =?UTF-8?q?=E9=83=A8=20Sync=20=E5=8F=98=E6=9B=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../OpenNexus生产化实施进度-2026-09-08.md | 11 + frontend/src-tauri/src/lib.rs | 2 + frontend/src-tauri/src/payloads.rs | 87 ++++++ frontend/src-tauri/src/sync_commands.rs | 1 + frontend/src-tauri/src/sync_discovery.rs | 247 ++++++++++++++++++ frontend/src-tauri/src/sync_inbox.rs | 9 +- frontend/src-tauri/src/sync_resolution.rs | 4 + frontend/src-tauri/src/sync_state.rs | 55 ++-- frontend/src-tauri/src/workspace.rs | 45 +++- frontend/src-tauri/tests/sync_push.rs | 151 +++++++++++ server sync/tests/host_fixture.py | 12 + 11 files changed, 589 insertions(+), 35 deletions(-) create mode 100644 frontend/src-tauri/src/payloads.rs create mode 100644 frontend/src-tauri/src/sync_discovery.rs diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 99eefbb..00b69dc 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -68,3 +68,14 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写 - Host 提供每 5 秒串行同步，先完成固定拉取窗口再发送本地作业；SQLite schema 6 保存暂停偏好。网络失败使用有界退避并尊重 Retry-After；认证、权限、协议等错误停止自动重试并显示。绑定变化/暂停取消旧轮次。 - 真实本地 HTTP 服务验证 Stronghold 会话存储、强制刷新、关闭重开保险库后复用、退出后旧令牌 401。新增 UI 测试验证密码清空、测试 HTTP 显式传参、确认取消和冲突 CAS 参数。Rust 全目标 43 项与前端全量 95 文件/508 项通过，正式 TypeScript 两项目检查通过。前端测试仍输出既有 KaTeX quirks/localhost:3000 连接警告，没有失败项。 - 尚未完成：后台轮询真实桌面实机验收、初始非空合并预览、外部文件变化捕获、附件流式持久化及全故障矩阵、数据分类和不同身份同路径解决。不能据此标记所有 S/E 验收 ID 已通过。 + + +## 增量：外部编辑捕获与大附件崩溃续传 + +- Workspace schema 7 将新 write/rename/delete journal 和 outbox 的正文替换为不可变 payload 引用，先 fsync 文件再记录摘要/长度；兼容旧内联正文恢复与上传迁移。100 MiB 不再写入 SQLite 正文列。暂不执行历史 payload GC。 +- 独立 sync_observed 保存上次提交快照；文件树/编辑器读取更新 files 缓存不会吞掉外部变更。后台捕获外部新增、编辑、删除，采用快照校验且不向外部编辑器写回。重复扫描不新增 outbox；删除捕获支持重启恢复。 +- 当前文件分类允许 Markdown 和 attachments 下常用附件类型，排除隐藏目录、已知缓存/日志/模型目录和可执行包扩展名；普通 JSON/配置不会直接复制。旧/新待发作业均检查分类，未支持的远端类型拒绝应用并保留游标。默认任务/Skill/配置等逻辑记录的白名单尚待实现，不作为 S-08 完整证据。 +- 真实本地 HTTP 100 MiB 测试每 10 MiB 在服务端写入后、响应前强杀独立 Rust 客户端，共 10 次。恢复进程复用同一设备会话、查询服务端 offset，最终上传/下载长度、SHA-256、file_id 一致且远端落盘无回流；两个临时 Vault 的元数据文件均小于 5 MiB。最初测试错误地重新登录生成不同设备，按协议无法复用原上传，现已改为通过 stdin 传递同一受控测试会话。 +- 此测试尚未覆盖完整 S-02 的每个 pull 边界 kill 20 轮，也不代表四并发服务 RSS 或生产 MinIO 性能验收。外部 rename 的稳定身份识别和目录大规模扫描优化继续实施。 + +- 本增量 Rust desktop 全目标 46 项通过、1 个受父测试驱动的独立进程辅助入口标记 ignored（父测试实际运行并强杀该入口 10 次）；Clippy `-D warnings` 通过。 diff --git a/frontend/src-tauri/src/lib.rs b/frontend/src-tauri/src/lib.rs index 62fe7a1..da5259f 100644 --- a/frontend/src-tauri/src/lib.rs +++ b/frontend/src-tauri/src/lib.rs @@ -2,6 +2,7 @@ pub mod core; pub mod credentials; +mod payloads; pub mod recent; #[cfg(feature = "desktop")] pub mod request_lifecycle; @@ -12,6 +13,7 @@ pub mod session_lock; pub mod sync_auth; #[cfg(feature = "desktop")] pub mod sync_client; +pub mod sync_discovery; pub mod sync_inbox; pub mod sync_resolution; pub mod sync_state; diff --git a/frontend/src-tauri/src/payloads.rs b/frontend/src-tauri/src/payloads.rs new file mode 100644 index 0000000..7a71a22 --- /dev/null +++ b/frontend/src-tauri/src/payloads.rs @@ -0,0 +1,87 @@ +//! Immutable payloads are fsynced before any SQLite reference becomes visible. +use crate::workspace::{hash, HostError, Result, Workspace}; +use rusqlite::{params, OptionalExtension}; +use sha2::{Digest, Sha256}; +use std::{ + fs, + io::{Read, Write}, + path::Path, +}; +impl Workspace { + pub(crate) fn store_payload(&self, operation: &str, content: &[u8]) -> Result<()> { + let digest = hash(content); + let target = self.sync_spool(&digest)?; + if target.exists() { + verify(&target, &digest, content.len() as u64)?; + } else { + let mut temp = tempfile::NamedTempFile::new_in(target.parent().unwrap())?; + temp.write_all(content)?; + temp.as_file().sync_all()?; + temp.persist_noclobber(&target) + .map_err(|_| HostError::new("SYNC_SPOOL_FAILED"))?; + #[cfg(unix)] + fs::File::open(target.parent().unwrap())?.sync_all()?; + } + let old: Option<(String, i64)> = self + .db + .query_row( + "SELECT hash,size FROM payloads WHERE operation_id=?1", + [operation], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .optional()?; + if old.is_some_and(|v| v != (digest.clone(), content.len() as i64)) { + return Err(HostError::new("OPERATION_PAYLOAD_CONFLICT")); + } + self.db.execute( + "INSERT OR IGNORE INTO payloads VALUES (?1,?2,?3)", + params![operation, digest, content.len() as i64], + )?; + Ok(()) + } + pub(crate) fn payload(&self, operation: &str, legacy: &[u8]) -> Result> { + let Some((digest, size)) = self.payload_ref(operation)? else { + return Ok(legacy.to_vec()); + }; + let target = self.sync_spool(&digest)?; + verify(&target, &digest, size as u64)?; + Ok(fs::read(target)?) + } + pub(crate) fn payload_ref(&self, operation: &str) -> Result> { + Ok(self + .db + .query_row( + "SELECT hash,size FROM payloads WHERE operation_id=?1", + [operation], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .optional()?) + } +} +pub(crate) fn verify(path: &Path, digest: &str, size: u64) -> Result<()> { + let meta = fs::symlink_metadata(path)?; + if meta.file_type().is_symlink() || !meta.is_file() || meta.len() != size { + return Err(HostError::new("SYNC_SPOOL_CORRUPT")); + } + #[cfg(windows)] + { + use std::os::windows::fs::MetadataExt; + if meta.file_attributes() & 0x400 != 0 { + return Err(HostError::new("UNSAFE_PATH")); + } + } + let mut stream = fs::File::open(path)?; + let mut hasher = Sha256::new(); + let mut buffer = vec![0; 1024 * 1024]; + loop { + let count = stream.read(&mut buffer)?; + if count == 0 { + break; + } + hasher.update(&buffer[..count]); + } + if format!("{:x}", hasher.finalize()) != digest { + return Err(HostError::new("SYNC_SPOOL_CORRUPT")); + } + Ok(()) +} diff --git a/frontend/src-tauri/src/sync_commands.rs b/frontend/src-tauri/src/sync_commands.rs index 7f52080..d53ddfc 100644 --- a/frontend/src-tauri/src/sync_commands.rs +++ b/frontend/src-tauri/src/sync_commands.rs @@ -321,6 +321,7 @@ async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> { .await?; let work = async { client.handshake().await?; + host.workspace.access(|ws| ws.sync_discover(&binding.id))?; for _ in 0..10 { if client.pull_page(&host.workspace, binding).await? == 0 { break; diff --git a/frontend/src-tauri/src/sync_discovery.rs b/frontend/src-tauri/src/sync_discovery.rs new file mode 100644 index 0000000..68ab1f7 --- /dev/null +++ b/frontend/src-tauri/src/sync_discovery.rs @@ -0,0 +1,247 @@ +//! Reconcile externally edited files against committed snapshots, never against UI read caches. +use crate::workspace::{hash, HostError, Result, Workspace}; +use rusqlite::{params, OptionalExtension}; +use std::{collections::HashSet, fs, io::Read, path::Path}; +use uuid::Uuid; +/// File transport only. Logical records receive their own versioned whitelist separately. +pub fn allowed(path: &str) -> bool { + let parts: Vec<_> = path.split('/').collect(); + if parts.iter().any(|part| { + part.starts_with('.') + || matches!( + part.to_ascii_lowercase().as_str(), + "node_modules" | "target" | "__pycache__" | "cache" | "logs" | "models" + ) + }) { + return false; + } + let extension = Path::new(path) + .extension() + .and_then(|v| v.to_str()) + .unwrap_or("") + .to_ascii_lowercase(); + extension == "md" + || (parts + .first() + .is_some_and(|v| v.eq_ignore_ascii_case("attachments")) + && matches!( + extension.as_str(), + "png" + | "jpg" + | "jpeg" + | "gif" + | "webp" + | "svg" + | "pdf" + | "mp3" + | "wav" + | "m4a" + | "ogg" + | "flac" + | "mp4" + | "webm" + | "mov" + | "txt" + | "csv" + | "docx" + | "xlsx" + | "pptx" + | "bin" + )) +} +impl Workspace { + pub(crate) fn sync_paths(&self) -> Result> { + fn walk(ws: &Workspace, directory: &Path, paths: &mut Vec) -> Result<()> { + for item in fs::read_dir(directory)? { + let item = item?; + let name = item.file_name().to_string_lossy().into_owned(); + if name.starts_with('.') + || matches!( + name.to_ascii_lowercase().as_str(), + "node_modules" | "target" | "__pycache__" | "cache" | "logs" | "models" + ) + { + continue; + } + let path = item + .path() + .strip_prefix(&ws.root) + .map_err(|_| HostError::new("UNSAFE_PATH"))? + .to_string_lossy() + .replace('\\', "/"); + let resolved = ws.resolve(&path)?; + if item.file_type()?.is_dir() { + walk(ws, &resolved, paths)?; + } else if allowed(&path) && item.file_type()?.is_file() { + paths.push(path); + } + } + Ok(()) + } + let mut paths = Vec::new(); + walk(self, &self.root, &mut paths)?; + paths.sort(); + Ok(paths) + } + pub fn sync_discover(&mut self, binding: &str) -> Result { + self.check_binding(binding)?; + let paths = self.sync_paths()?; + let mut seen = HashSet::new(); + let mut changes = 0; + for path in paths { + let target = self.resolve(&path)?; + if fs::metadata(&target)?.len() > 104857600 { + return Err(HostError::new("FILE_TOO_LARGE")); + } + let mut bytes = Vec::new(); + fs::File::open(&target)? + .take(104857601) + .read_to_end(&mut bytes)?; + if bytes.len() > 104857600 { + return Err(HostError::new("FILE_TOO_LARGE")); + } + let digest = hash(&bytes); + let previous = self.entry(&path)?; + let observed: Option<(String, String, bool)> = if let Some(entry) = &previous { + self.db + .query_row( + "SELECT path,hash,deleted FROM sync_observed WHERE file_id=?1", + [&entry.file_id], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .optional()? + } else { + None + }; + if let Some(entry) = &previous { + seen.insert(entry.file_id.clone()); + } + if observed + .as_ref() + .is_some_and(|v| v == &(path.clone(), digest.clone(), false)) + { + continue; + } + // Confirm the snapshot without writing back over an external editor. + crate::payloads::verify(&target, &digest, bytes.len() as u64)?; + let operation = Uuid::new_v4().to_string(); + self.store_payload(&operation, &bytes)?; + let file_id = previous.map_or_else(|| Uuid::new_v4().to_string(), |e| e.file_id); + let tx = self.db.transaction()?; + tx.execute("INSERT INTO files VALUES (?1,?2,?3,1,0) ON CONFLICT(path) DO UPDATE SET hash=excluded.hash,revision=files.revision+1,deleted=0",params![file_id,path,digest])?; + tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',X'','pending' FROM files WHERE id=?2",params![operation,file_id])?; + 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,path,digest])?; + tx.commit()?; + seen.insert(file_id); + changes += 1; + } + let previous = { + let mut statement = self + .db + .prepare("SELECT file_id,path FROM sync_observed WHERE deleted=0")?; + let rows = statement + .query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))? + .collect::, _>>()?; + rows + }; + for (file_id, path) in previous { + if seen.contains(&file_id) || !allowed(&path) || self.resolve(&path)?.exists() { + continue; + } + let operation = Uuid::new_v4().to_string(); + let tx = self.db.transaction()?; + tx.execute( + "UPDATE files SET deleted=1,revision=revision+1 WHERE id=?1", + [&file_id], + )?; + tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,'','delete',X'','pending' FROM files WHERE id=?2",params![operation,file_id])?; + tx.execute( + "UPDATE sync_observed SET deleted=1 WHERE file_id=?1", + [&file_id], + )?; + tx.commit()?; + changes += 1; + } + Ok(changes) + } +} +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn external_edits_after_ui_reads_are_queued_once_and_deletions_survive_restart() { + let root = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(root.path()).unwrap(); + let binding = ws + .sync_bind_empty("https://sync.example", "remote", "account") + .unwrap(); + fs::write(root.path().join("a.md"), b"one").unwrap(); + ws.read("a.md").unwrap(); + ws.scan().unwrap(); + assert_eq!(ws.sync_discover(&binding.id).unwrap(), 1); + assert_eq!(ws.sync_discover(&binding.id).unwrap(), 0); + fs::write(root.path().join("a.md"), b"two").unwrap(); + ws.read("a.md").unwrap(); + assert_eq!(ws.sync_discover(&binding.id).unwrap(), 1); + fs::remove_file(root.path().join("a.md")).unwrap(); + drop(ws); + let mut ws = Workspace::open(root.path()).unwrap(); + assert_eq!(ws.sync_discover(&binding.id).unwrap(), 1); + assert_eq!(ws.sync_discover(&binding.id).unwrap(), 0); + assert_eq!(ws.pending_count().unwrap(), 3); + let inline: i64 = ws + .db + .query_row( + "SELECT COALESCE(SUM(length(content)),0) FROM outbox", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(inline, 0); + } + #[test] + fn file_classification_excludes_credentials_caches_and_executable_packages() { + for path in [ + ".env", + ".ainote/credentials.md", + ".git/readme.md", + "models/model.md", + "logs/trace.md", + "node_modules/package/readme.md", + "attachments/tool.exe", + "attachments/plugin.zip", + "settings.json", + ] { + assert!(!allowed(path), "{path}"); + } + for path in [ + "notes/a.md", + "attachments/movie.mp4", + "attachments/image.png", + "attachments/fixture.bin", + ] { + assert!(allowed(path), "{path}"); + } + } + #[test] + fn prohibited_workspace_outbox_is_retained_but_never_sent() { + let root = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(root.path()).unwrap(); + ws.write("provider-settings.json", "", b"fixture-secret", "local") + .unwrap(); + let binding = ws + .sync_bind_empty("https://sync.example", "remote", "account") + .unwrap(); + assert!(ws.sync_next(&binding.id).unwrap().is_none()); + assert_eq!(ws.pending_count().unwrap(), 0); + assert_eq!( + ws.read("provider-settings.json").unwrap().content, + "fixture-secret" + ); + let state: String = ws + .db + .query_row("SELECT state FROM outbox", [], |r| r.get(0)) + .unwrap(); + assert_eq!(state, "excluded"); + } +} diff --git a/frontend/src-tauri/src/sync_inbox.rs b/frontend/src-tauri/src/sync_inbox.rs index 1865c8b..1ec11a2 100644 --- a/frontend/src-tauri/src/sync_inbox.rs +++ b/frontend/src-tauri/src/sync_inbox.rs @@ -47,6 +47,9 @@ impl RemoteRevision { { return Err(HostError::new("SYNC_RESPONSE_INVALID")); } + if !crate::sync_discovery::allowed(&self.path) { + return Err(HostError::new("SYNC_CLASS_UNSUPPORTED")); + } Ok(()) } } @@ -61,11 +64,7 @@ impl Workspace { if self.sync_binding()?.is_some() { return Err(HostError::new("SYNC_ALREADY_BOUND")); } - if self - .scan()? - .iter() - .any(|entry| !entry.is_folder && !entry.deleted) - { + if !self.sync_paths()?.is_empty() { return Err(HostError::new("SYNC_RECONCILIATION_REQUIRED")); } let id = Uuid::new_v4().to_string(); diff --git a/frontend/src-tauri/src/sync_resolution.rs b/frontend/src-tauri/src/sync_resolution.rs index 3cd4df5..89b051d 100644 --- a/frontend/src-tauri/src/sync_resolution.rs +++ b/frontend/src-tauri/src/sync_resolution.rs @@ -184,6 +184,10 @@ impl Workspace { "UPDATE files SET deleted=1,revision=revision+1 WHERE id=?1", [&revision.file_id], )?; + tx.execute( + "UPDATE sync_observed SET deleted=1 WHERE file_id=?1", + [&revision.file_id], + )?; let entry: Entry = tx.query_row( "SELECT id,path,hash,revision FROM files WHERE id=?1", [&revision.file_id], diff --git a/frontend/src-tauri/src/sync_state.rs b/frontend/src-tauri/src/sync_state.rs index 647a2ad..0a63c83 100644 --- a/frontend/src-tauri/src/sync_state.rs +++ b/frontend/src-tauri/src/sync_state.rs @@ -1,8 +1,8 @@ //! Durable queue state. Network code never invents a remote base from a local revision. -use crate::workspace::{hash, HostError, Result, Workspace}; +use crate::workspace::{HostError, Result, Workspace}; use rusqlite::{params, OptionalExtension}; use serde::{Deserialize, Serialize}; -use std::{fs, io::Write, path::PathBuf}; +use std::{fs, path::PathBuf}; use uuid::Uuid; #[derive(Clone, Serialize, Deserialize)] @@ -147,28 +147,31 @@ impl Workspace { let Some((operation_id, file_id, path, digest, operation, content)) = pending else { break; }; - if operation == "put" { - if hash(&content) != digest { + if !crate::sync_discovery::allowed(&path) { + self.db.execute( + "UPDATE outbox SET state='excluded' WHERE operation_id=?1", + [&operation_id], + )?; + continue; + } + let size = if operation == "put" { + if self.payload_ref(&operation_id)?.is_none() { + self.store_payload(&operation_id, &content)?; + } + let (stored, size) = self + .payload_ref(&operation_id)? + .ok_or_else(|| HostError::new("SYNC_SPOOL_CORRUPT"))?; + if stored != digest { return Err(HostError::new("SYNC_SPOOL_CORRUPT")); } - let target = self.sync_spool(&digest)?; - if target.exists() { - if fs::symlink_metadata(&target)?.file_type().is_symlink() - || hash(&fs::read(&target)?) != digest - { - return Err(HostError::new("SYNC_SPOOL_CORRUPT")); - } - } else { - let mut temp = tempfile::NamedTempFile::new_in(target.parent().unwrap())?; - temp.write_all(&content)?; - temp.as_file().sync_all()?; - temp.persist_noclobber(target) - .map_err(|_| HostError::new("SYNC_SPOOL_FAILED"))?; - } - } + crate::payloads::verify(&self.sync_spool(&digest)?, &digest, size as u64)?; + size + } else { + 0 + }; let tx = self.db.transaction()?; tx.execute("INSERT OR IGNORE INTO sync_jobs VALUES (?1,?2,?3,?4,?5,?6,?7,'pending',NULL,NULL,NULL,NULL)", - params![binding,operation_id,file_id,path,digest,content.len() as i64,operation])?; + params![binding,operation_id,file_id,path,digest,size,operation])?; tx.execute( "UPDATE outbox SET state='queued',content=X'' WHERE operation_id=?1", [&operation_id], @@ -179,9 +182,16 @@ 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','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 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()?) + }).optional()?; + if job + .as_ref() + .is_some_and(|job| !crate::sync_discovery::allowed(&job.path)) + { + return Err(HostError::new("SYNC_CLASS_UNSUPPORTED")); + } + Ok(job) } fn check_job(&self, job: &Job) -> Result<()> { self.check_binding(&job.binding)?; @@ -258,6 +268,7 @@ impl Workspace { #[cfg(test)] mod tests { use super::*; + use crate::workspace::hash; #[test] fn queue_uses_remote_bases_and_keeps_retry_payload_across_restart() { let root = tempfile::tempdir().unwrap(); diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index b38eea9..4a7fd9f 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 > 6 { + if version > 7 { return Err(HostError::new("SCHEMA_INCOMPATIBLE")); } - if (1..6).contains(&version) { + if (1..7).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,8 @@ 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 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); 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( @@ -170,7 +172,10 @@ impl Workspace { [], )?; } - db.execute_batch("PRAGMA user_version=6; COMMIT;")?; + 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;")?; let vault_id: String = db .query_row("SELECT id FROM identity", [], |r| r.get(0)) .optional()? @@ -222,7 +227,7 @@ impl Workspace { Ok(path) } - fn entry(&self, path: &str) -> Result> { + pub(crate) fn entry(&self, path: &str) -> Result> { Ok(self .db .query_row( @@ -477,6 +482,7 @@ impl Workspace { }, |entry| entry.file_id, ); + self.store_payload(operation_id, content)?; let tx = self.db.transaction()?; tx.execute( "INSERT INTO operations VALUES (?1,?2,'pending',NULL)", @@ -484,7 +490,14 @@ impl Workspace { )?; tx.execute( "INSERT INTO journal VALUES (?1,?2,?3,?4,?5,?6,'pending')", - params![operation_id, file_id, path, expected, content, origin], + params![ + operation_id, + file_id, + path, + expected, + b"".as_slice(), + origin + ], )?; tx.commit()?; self.apply_journal(operation_id, &file_id, path, expected, content, origin)?; @@ -534,11 +547,14 @@ impl Workspace { #[cfg(unix)] File::open(parent)?.sync_all()?; } + // Upgrade legacy inline journal payloads before publishing an outbox reference. + self.store_payload(operation_id, content)?; // 文件成功但 DB 未提交时，重启凭 journal 补齐同一 operation_id，避免丢 outbox。 let tx = self.db.transaction()?; tx.execute("INSERT INTO files VALUES (?1,?2,?3,1,0) ON CONFLICT(path) DO UPDATE SET hash=excluded.hash,revision=files.revision+1,deleted=0", params![file_id,path,digest])?; + 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,path,digest])?; if origin == "local" { - tx.execute("INSERT OR IGNORE INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE path=?3", params![operation_id,content,path])?; + tx.execute("INSERT OR IGNORE INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE path=?3", params![operation_id,b"".as_slice(),path])?; } let entry = tx.query_row( "SELECT id,path,hash,revision,deleted FROM files WHERE path=?1", @@ -597,6 +613,7 @@ impl Workspace { result }; for (op, id, path, expected, content, origin) in pending { + let content = self.payload(&op, &content)?; match self.apply_journal(&op, &id, &path, &expected, &content, &origin) { Err(e) if e.code == "RECOVERY_CONFLICT" => {} result => result?, @@ -738,6 +755,7 @@ impl Workspace { } self.entry(path)? .ok_or_else(|| HostError::new("FILE_NOT_FOUND"))?; + self.store_payload(id, &content)?; let tx = self.db.transaction()?; tx.execute( "INSERT INTO operations VALUES (?1,?2,'pending',NULL)", @@ -745,7 +763,15 @@ impl Workspace { )?; tx.execute( "INSERT INTO file_ops VALUES (?1,?2,?3,?4,?5,?6,'pending',?7)", - params![id, kind, path, destination, expected, content, origin], + params![ + id, + kind, + path, + destination, + expected, + b"".as_slice(), + origin + ], )?; tx.commit()?; Ok(id.to_owned()) @@ -773,6 +799,8 @@ impl Workspace { )) }, )?; + let content = self.payload(id, &content)?; + self.store_payload(id, &content)?; let source = self.resolve(&path)?; let previous = self .entry(&path)? @@ -822,7 +850,7 @@ impl Workspace { params![destination, 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])?; + tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE id=?3", params![id,b"".as_slice(),previous.file_id])?; } } else { tx.execute( @@ -833,6 +861,7 @@ impl Workspace { 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("INSERT INTO sync_observed SELECT id,path,hash,deleted FROM files WHERE id=?1 ON CONFLICT(file_id) DO UPDATE SET path=excluded.path,hash=excluded.hash,deleted=excluded.deleted",[&previous.file_id])?; tx.execute("DELETE FROM file_ops WHERE id=?1", [id])?; let mut result = previous; result.revision += 1; diff --git a/frontend/src-tauri/tests/sync_push.rs b/frontend/src-tauri/tests/sync_push.rs index 3426140..bd2c14a 100644 --- a/frontend/src-tauri/tests/sync_push.rs +++ b/frontend/src-tauri/tests/sync_push.rs @@ -303,6 +303,140 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate ); } } + + // 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}; + use std::io::Write; + let large_remote = client + .json( + reqwest::Method::POST, + "sync/v1/vaults", + Some(json!({"name":"100 MiB resumable"})), + ) + .await + .unwrap(); + let large_remote = large_remote["vault_id"].as_str().unwrap(); + let large_root = tempfile::tempdir().unwrap(); + let mut large_ws = Workspace::open(large_root.path()).unwrap(); + let large_binding = large_ws + .sync_bind_empty(&endpoint, large_remote, "rust-fixture") + .unwrap(); + std::fs::create_dir(large_root.path().join("attachments")).unwrap(); + let mut attachment = + std::fs::File::create(large_root.path().join("attachments/large.bin")).unwrap(); + let block = vec![42u8; 1024 * 1024]; + let mut hasher = Sha256::new(); + for _ in 0..100 { + attachment.write_all(&block).unwrap(); + hasher.update(&block); + } + attachment.sync_all().unwrap(); + drop(attachment); + let expected_hash = format!("{:x}", hasher.finalize()); + assert_eq!(large_ws.sync_discover(&large_binding.id).unwrap(), 1); + large_ws.sync_capture(&large_binding.id).unwrap(); + let large_id = large_ws + .sync_next(&large_binding.id) + .unwrap() + .unwrap() + .file_id; + drop(large_ws); + std::fs::write(root.path().join("interrupt-upload"), b"controlled-fixture").unwrap(); + for boundary in 1..=10 { + let mut worker = Server( + Command::new(std::env::current_exe().unwrap()) + .args(["--ignored", "--exact", "resumable_upload_worker"]) + .env("OPENNEXUS_SYNC_WORKER_ROOT", large_root.path()) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(), + ); + worker + .0 + .stdin + .take() + .unwrap() + .write_all(session.access_token.as_bytes()) + .unwrap(); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(45); + let marker = root.path().join("upload-boundary"); + while !marker.exists() { + assert!( + worker.0.try_wait().unwrap().is_none(), + "upload worker exited before boundary {boundary}" + ); + assert!( + std::time::Instant::now() < deadline, + "upload boundary timeout" + ); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + assert_eq!( + std::fs::read_to_string(&marker) + .unwrap() + .parse::() + .unwrap(), + boundary * 10 * 1024 * 1024 + ); + worker.0.kill().unwrap(); + worker.0.wait().unwrap(); + std::fs::remove_file(marker).unwrap(); + } + std::fs::remove_file(root.path().join("interrupt-upload")).unwrap(); + let large_workspace = Arc::new(Mutex::new(Workspace::open(large_root.path()).unwrap())); + assert!(client + .push_one(&large_workspace, &large_binding) + .await + .unwrap()); + assert!(!client + .push_one(&large_workspace, &large_binding) + .await + .unwrap()); + let download_root = tempfile::tempdir().unwrap(); + let download = Arc::new(Mutex::new(Workspace::open(download_root.path()).unwrap())); + let download_binding = download + .lock() + .unwrap() + .sync_bind_download(&endpoint, large_remote, "rust-fixture") + .unwrap(); + assert_eq!( + client_b + .pull_page(&download, &download_binding) + .await + .unwrap(), + 1 + ); + let received = std::fs::read(download_root.path().join("attachments/large.bin")).unwrap(); + assert_eq!(received.len(), 104857600); + assert_eq!(format!("{:x}", Sha256::digest(&received)), expected_hash); + assert_eq!( + download.lock().unwrap().path_for_id(&large_id).unwrap(), + "attachments/large.bin" + ); + assert_eq!(download.lock().unwrap().pending_count().unwrap(), 0); + assert_eq!( + download + .lock() + .unwrap() + .sync_discover(&download_binding.id) + .unwrap(), + 0 + ); + // SQLite stores metadata, never the 100 MiB body. + for directory in [large_root.path(), download_root.path()] { + let managed = directory.join(".ainote"); + for item in std::fs::read_dir(managed).unwrap().flatten() { + if item.file_type().unwrap().is_file() { + assert!( + item.metadata().unwrap().len() < 5 * 1024 * 1024, + "large body leaked into metadata storage" + ); + } + } + } // Host sessions survive encrypted storage reopen and refresh on the actual service. use notesagent_host::{credentials::CredentialBroker, sync_auth}; let credential_root = tempfile::tempdir().unwrap(); @@ -359,3 +493,20 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate 401 ); } + +#[tokio::test] +#[ignore = "helper process driven and killed by the parent fault test"] +async fn resumable_upload_worker() { + let root = std::env::var("OPENNEXUS_SYNC_WORKER_ROOT").expect("controlled fixture root"); + let ws = Arc::new(Mutex::new(Workspace::open(Path::new(&root)).unwrap())); + let binding = ws.lock().unwrap().sync_binding().unwrap().unwrap(); + assert!(binding.endpoint.starts_with("http://127.0.0.1:")); + use std::io::Read; + let mut token = Zeroizing::new(String::new()); + std::io::stdin() + .take(4096) + .read_to_string(&mut token) + .unwrap(); + let client = SyncClient::new(&binding.endpoint, token, true).unwrap(); + client.push_one(&ws, &binding).await.unwrap(); +} diff --git a/server sync/tests/host_fixture.py b/server sync/tests/host_fixture.py index 5cdcf2e..6cc867e 100644 --- a/server sync/tests/host_fixture.py +++ b/server sync/tests/host_fixture.py @@ -18,6 +18,18 @@ def main(): database = Database('sqlite:///' + str(root / 'sync.sqlite3')) app = create_app(database, DiskObjects(root / 'objects'), root / 'staging') database.add_user('rust-fixture', 'controlled-fixture-password') + @app.middleware('http') + async def interrupt_upload(request, call_next): + response = await call_next(request) + if (root / 'interrupt-upload').exists() and request.method == 'PUT' and '/uploads/' in request.url.path and response.status_code == 200: + offset = int(request.query_params.get('offset', '0')) + int(request.headers.get('content-length', '0')) + if offset and offset % (10 * 1024 * 1024) == 0: + marker = root / 'upload-boundary' + marker.write_text(str(offset), encoding='ascii') + for _ in range(600): + if not marker.exists(): break + await asyncio.sleep(.05) + return response sock = socket.socket() sock.bind(('127.0.0.1', 0)) sock.listen(128)