feat: 完成 OpenNexus 第三阶段核心功能与生产化基础 #45
@@ -625,3 +625,13 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写
|
|||||||
- 首轮全目标编译发现删除决策 fingerprint 仍需要生产 hash 导入,已恢复其非 test 范围,同时移除 sync_inbox 中不再使用的 fs 导入;该问题在全目标回归期间发现,未将仅 lib test 编译通过当作全部目标通过。
|
- 首轮全目标编译发现删除决策 fingerprint 仍需要生产 hash 导入,已恢复其非 test 范围,同时移除 sync_inbox 中不再使用的 fs 导入;该问题在全目标回归期间发现,未将仅 lib test 编译通过当作全部目标通过。
|
||||||
- 外部变更扫描、初始绑定对账和部分逻辑记录/API 仍有整块读取;未将这些功能测试当作 RSS 基准或完整强杀/断电矩阵。大附件端到端内存与完整生产化目标继续未完成。
|
- 外部变更扫描、初始绑定对账和部分逻辑记录/API 仍有整块读取;未将这些功能测试当作 RSS 基准或完整强杀/断电矩阵。大附件端到端内存与完整生产化目标继续未完成。
|
||||||
- 最终 desktop 全目标累计 135 通过、12 ignored(库 120、Host 8、其余集成 7),包括实际 Sync 服务与故障恢复集成;日志 `.build/sync-conflict-stream-full.log`。desktop 全目标 Clippy -D warnings 通过,日志 `.build/sync-conflict-stream-clippy.log`。本轮没有重新执行 ignored 沙箱长时验收,全部生产化条件仍未满足。
|
- 最终 desktop 全目标累计 135 通过、12 ignored(库 120、Host 8、其余集成 7),包括实际 Sync 服务与故障恢复集成;日志 `.build/sync-conflict-stream-full.log`。desktop 全目标 Clippy -D warnings 通过,日志 `.build/sync-conflict-stream-clippy.log`。本轮没有重新执行 ignored 沙箱长时验收,全部生产化条件仍未满足。
|
||||||
|
|
||||||
|
|
||||||
|
## 增量:扫描、首次对账与重新绑定的流式读取
|
||||||
|
|
||||||
|
- 共用 hash_file_info 返回实际读取的摘要与长度;sync_file_info 对普通文件有界流式读取,对逻辑记录仍最多读取 1 MiB 加 1 字节并执行现有 schema 校验。sync_discover 不再读入整块文件,发现变化后用 store_payload_file 保存经过复核的快照再提交观察记录与 outbox。
|
||||||
|
- 首次合并预览使用流式摘要/实际长度,确认后的本地准备使用流式 payload 快照,内容变化仍拒绝旧预览。Workspace.scan 的 Markdown 摘要也改为有界读取,避免 UI 扫描先分配整个大文件。
|
||||||
|
- 空远端绑定原先只调用 Markdown scan;重新绑定时已观察过的附件没有新 outbox,后续 discover 也可能因摘要未变而跳过。本轮改为枚举同步白名单路径(含允许的附件/记录),归档旧绑定队列后为当前文件准备流式快照与本地写入意图,再建立新绑定队列;不再仅以 Markdown 集合作为绑定数据集。
|
||||||
|
- 新增 100 MiB Markdown 与 100 MiB 附件测试,分别走首次合并与空远端绑定:仅改末字节即可使旧预览失效,绑定后重复发现不增加事件,UI scan 后的外部修改仍被 discover 识别一次,重开不重复;切换远端 Vault/账号重新绑定时两类文件均进入新队列且原 file_id 保持。测试通过,耗时 10.36 秒,日志 `.build/sync-discovery-stream-large.log`。先前同步回归 17 项通过,日志 `.build/sync-discovery-stream-tests.log`。
|
||||||
|
- 测试补充阶段修正了 Binding 不实现 Debug 时不能使用 unwrap_err 的测试写法,没有为调试输出扩大生产类型的可见字段。上述测试不是真实进程峰值内存测量、10000 文件基准或完整首次绑定故障矩阵;这些验收仍需继续。
|
||||||
|
- 最终 desktop 全目标累计 136 通过、12 ignored(库 121、Host 8、其余集成 7),包含真实 Sync 服务与故障恢复集成,日志 `.build/sync-discovery-stream-full.log`;desktop 全目标 Clippy -D warnings 通过,日志 `.build/sync-discovery-stream-clippy.log`。仍需独立进程峰值内存和完整基准/部署/发布验收,完整生产化目标未完成。
|
||||||
|
|||||||
@@ -165,6 +165,21 @@ pub(crate) fn validate_record(file: &mut fs::File, path: &str, size: u64) -> Res
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
pub(crate) fn hash_file(path: &Path) -> Result<String> {
|
pub(crate) fn hash_file(path: &Path) -> Result<String> {
|
||||||
|
hash_file_info(path).map(|(digest, _)| digest)
|
||||||
|
}
|
||||||
|
pub(crate) fn sync_file_info(source: &Path, path: &str) -> Result<(String, u64)> {
|
||||||
|
if crate::records::is_record(path) {
|
||||||
|
let mut bytes = Vec::new();
|
||||||
|
fs::File::open(source)?
|
||||||
|
.take(1024 * 1024 + 1)
|
||||||
|
.read_to_end(&mut bytes)?;
|
||||||
|
crate::records::validate(path, &bytes)?;
|
||||||
|
Ok((hash(&bytes), bytes.len() as u64))
|
||||||
|
} else {
|
||||||
|
hash_file_info(source)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
fn hash_file_info(path: &Path) -> Result<(String, u64)> {
|
||||||
let mut file = fs::File::open(path)?;
|
let mut file = fs::File::open(path)?;
|
||||||
let mut buffer = vec![0u8; VERIFY_BUFFER_BYTES];
|
let mut buffer = vec![0u8; VERIFY_BUFFER_BYTES];
|
||||||
let mut hasher = Sha256::new();
|
let mut hasher = Sha256::new();
|
||||||
@@ -180,7 +195,7 @@ pub(crate) fn hash_file(path: &Path) -> Result<String> {
|
|||||||
}
|
}
|
||||||
hasher.update(&buffer[..count]);
|
hasher.update(&buffer[..count]);
|
||||||
}
|
}
|
||||||
Ok(format!("{:x}", hasher.finalize()))
|
Ok((format!("{:x}", hasher.finalize()), total))
|
||||||
}
|
}
|
||||||
pub(crate) fn verify(path: &Path, digest: &str, size: u64) -> Result<()> {
|
pub(crate) fn verify(path: &Path, digest: &str, size: u64) -> Result<()> {
|
||||||
open_verified(path, digest, size).map(drop)
|
open_verified(path, digest, size).map(drop)
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
//! Reconcile externally edited files against committed snapshots, never against UI read caches.
|
//! Reconcile externally edited files against committed snapshots, never against UI read caches.
|
||||||
use crate::workspace::{hash, HostError, Result, Workspace};
|
use crate::workspace::{HostError, Result, Workspace};
|
||||||
use rusqlite::{params, OptionalExtension};
|
use rusqlite::{params, OptionalExtension};
|
||||||
use std::{collections::HashSet, fs, io::Read, path::Path};
|
use std::{collections::HashSet, fs, path::Path};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
/// File transport only. Logical records receive their own versioned whitelist separately.
|
/// File transport only. Logical records receive their own versioned whitelist separately.
|
||||||
pub fn allowed(path: &str) -> bool {
|
pub fn allowed(path: &str) -> bool {
|
||||||
@@ -93,20 +93,7 @@ impl Workspace {
|
|||||||
let mut changes = 0;
|
let mut changes = 0;
|
||||||
for path in paths {
|
for path in paths {
|
||||||
let target = self.resolve(&path)?;
|
let target = self.resolve(&path)?;
|
||||||
if fs::metadata(&target)?.len() > 104857600 {
|
let (digest, _) = crate::payloads::sync_file_info(&target, &path)?;
|
||||||
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"));
|
|
||||||
}
|
|
||||||
if crate::records::is_record(&path) {
|
|
||||||
crate::records::validate(&path, &bytes)?;
|
|
||||||
}
|
|
||||||
let digest = hash(&bytes);
|
|
||||||
let previous = self.entry(&path)?;
|
let previous = self.entry(&path)?;
|
||||||
let observed: Option<(String, String, bool)> = if let Some(entry) = &previous {
|
let observed: Option<(String, String, bool)> = if let Some(entry) = &previous {
|
||||||
self.db
|
self.db
|
||||||
@@ -129,9 +116,8 @@ impl Workspace {
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
// Confirm the snapshot without writing back over an external editor.
|
// 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();
|
let operation = Uuid::new_v4().to_string();
|
||||||
self.store_payload(&operation, &bytes)?;
|
self.store_payload_file(&operation, &target, &digest)?;
|
||||||
let file_id = previous.map_or_else(|| Uuid::new_v4().to_string(), |e| e.file_id);
|
let file_id = previous.map_or_else(|| Uuid::new_v4().to_string(), |e| e.file_id);
|
||||||
let tx = self.db.transaction()?;
|
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 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])?;
|
||||||
|
|||||||
@@ -6,7 +6,9 @@ use crate::{
|
|||||||
};
|
};
|
||||||
use rusqlite::{params, OptionalExtension};
|
use rusqlite::{params, OptionalExtension};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use std::{collections::BTreeMap, fs};
|
use std::collections::BTreeMap;
|
||||||
|
#[cfg(test)]
|
||||||
|
use std::fs;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
#[derive(Clone, Serialize, Deserialize)]
|
#[derive(Clone, Serialize, Deserialize)]
|
||||||
pub struct Snapshot {
|
pub struct Snapshot {
|
||||||
@@ -36,17 +38,11 @@ impl Workspace {
|
|||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|path| {
|
.map(|path| {
|
||||||
let source = self.resolve(&path)?;
|
let source = self.resolve(&path)?;
|
||||||
if fs::metadata(&source)?.len() > 104857600 {
|
let (hash, size) = crate::payloads::sync_file_info(&source, &path)?;
|
||||||
return Err(HostError::new("FILE_TOO_LARGE"));
|
|
||||||
}
|
|
||||||
let bytes = fs::read(source)?;
|
|
||||||
if crate::records::is_record(&path) {
|
|
||||||
crate::records::validate(&path, &bytes)?;
|
|
||||||
}
|
|
||||||
Ok(Local {
|
Ok(Local {
|
||||||
path,
|
path,
|
||||||
hash: hash(&bytes),
|
hash,
|
||||||
size: bytes.len(),
|
size: size as usize,
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
.collect()
|
.collect()
|
||||||
@@ -131,11 +127,14 @@ impl Workspace {
|
|||||||
let mut prepared = Vec::new();
|
let mut prepared = Vec::new();
|
||||||
for item in local {
|
for item in local {
|
||||||
let operation = Uuid::new_v4().to_string();
|
let operation = Uuid::new_v4().to_string();
|
||||||
let bytes = fs::read(self.resolve(&item.path)?)?;
|
self.store_payload_file(&operation, &self.resolve(&item.path)?, &item.hash)
|
||||||
if hash(&bytes) != item.hash {
|
.map_err(|error| {
|
||||||
return Err(HostError::new("SYNC_PREVIEW_CHANGED"));
|
if error.code == "REVISION_CONFLICT" {
|
||||||
}
|
HostError::new("SYNC_PREVIEW_CHANGED")
|
||||||
self.store_payload(&operation, &bytes)?;
|
} else {
|
||||||
|
error
|
||||||
|
}
|
||||||
|
})?;
|
||||||
let old = self.entry(&item.path)?;
|
let old = self.entry(&item.path)?;
|
||||||
let remote = snapshot
|
let remote = snapshot
|
||||||
.items
|
.items
|
||||||
@@ -271,6 +270,98 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
#[test]
|
#[test]
|
||||||
|
fn hundred_mib_discovery_preview_and_rebinding_preserve_all_current_files() {
|
||||||
|
use std::io::{Seek, SeekFrom, Write};
|
||||||
|
for initial in [false, true] {
|
||||||
|
let root = tempfile::tempdir().unwrap();
|
||||||
|
fs::create_dir(root.path().join("attachments")).unwrap();
|
||||||
|
let source = root.path().join("large.md");
|
||||||
|
let mut file = fs::File::create(&source).unwrap();
|
||||||
|
let block = vec![b'a'; 64 * 1024];
|
||||||
|
for _ in 0..1600 {
|
||||||
|
file.write_all(&block).unwrap();
|
||||||
|
}
|
||||||
|
file.sync_all().unwrap();
|
||||||
|
drop(file);
|
||||||
|
fs::copy(&source, root.path().join("attachments/large.bin")).unwrap();
|
||||||
|
let mut ws = Workspace::open(root.path()).unwrap();
|
||||||
|
let snapshot = Snapshot {
|
||||||
|
boundary: 0,
|
||||||
|
items: Vec::new(),
|
||||||
|
};
|
||||||
|
let binding = if initial {
|
||||||
|
let preview = ws
|
||||||
|
.sync_preview("https://sync.example", "remote", "account", &snapshot)
|
||||||
|
.unwrap();
|
||||||
|
let mut file = fs::OpenOptions::new().write(true).open(&source).unwrap();
|
||||||
|
file.seek(SeekFrom::End(-1)).unwrap();
|
||||||
|
file.write_all(b"b").unwrap();
|
||||||
|
file.sync_all().unwrap();
|
||||||
|
drop(file);
|
||||||
|
assert_eq!(
|
||||||
|
ws.sync_bind_initial(
|
||||||
|
"https://sync.example",
|
||||||
|
"remote",
|
||||||
|
"account",
|
||||||
|
&snapshot,
|
||||||
|
&preview.fingerprint
|
||||||
|
)
|
||||||
|
.err()
|
||||||
|
.unwrap()
|
||||||
|
.code,
|
||||||
|
"SYNC_PREVIEW_CHANGED"
|
||||||
|
);
|
||||||
|
let preview = ws
|
||||||
|
.sync_preview("https://sync.example", "remote", "account", &snapshot)
|
||||||
|
.unwrap();
|
||||||
|
ws.sync_bind_initial(
|
||||||
|
"https://sync.example",
|
||||||
|
"remote",
|
||||||
|
"account",
|
||||||
|
&snapshot,
|
||||||
|
&preview.fingerprint,
|
||||||
|
)
|
||||||
|
.unwrap()
|
||||||
|
} else {
|
||||||
|
ws.sync_bind_empty("https://sync.example", "remote", "account")
|
||||||
|
.unwrap()
|
||||||
|
};
|
||||||
|
assert_eq!(ws.sync_discover(&binding.id).unwrap(), 0);
|
||||||
|
ws.sync_capture(&binding.id).unwrap();
|
||||||
|
assert_eq!(ws.pending_count().unwrap(), 2);
|
||||||
|
let identity = ws.entry("large.md").unwrap().unwrap().file_id;
|
||||||
|
let mut file = fs::OpenOptions::new().write(true).open(&source).unwrap();
|
||||||
|
file.seek(SeekFrom::End(-1)).unwrap();
|
||||||
|
file.write_all(b"c").unwrap();
|
||||||
|
file.sync_all().unwrap();
|
||||||
|
drop(file);
|
||||||
|
ws.scan().unwrap();
|
||||||
|
assert_eq!(ws.sync_discover(&binding.id).unwrap(), 1);
|
||||||
|
assert_eq!(ws.entry("large.md").unwrap().unwrap().file_id, identity);
|
||||||
|
drop(ws);
|
||||||
|
ws = Workspace::open(root.path()).unwrap();
|
||||||
|
assert_eq!(ws.sync_discover(&binding.id).unwrap(), 0);
|
||||||
|
assert_eq!(ws.pending_count().unwrap(), 3);
|
||||||
|
ws.sync_unbind(&binding.id).unwrap();
|
||||||
|
let rebound = ws
|
||||||
|
.sync_bind_empty("https://sync.example", "another-vault", "another-account")
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(ws.pending_count().unwrap(), 2);
|
||||||
|
assert_eq!(
|
||||||
|
ws.db
|
||||||
|
.query_row(
|
||||||
|
"SELECT COUNT(*) FROM sync_jobs WHERE binding=?1",
|
||||||
|
[&rebound.id],
|
||||||
|
|r| r.get::<_, i64>(0)
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
2
|
||||||
|
);
|
||||||
|
assert_eq!(ws.sync_discover(&rebound.id).unwrap(), 0);
|
||||||
|
assert_eq!(ws.entry("large.md").unwrap().unwrap().file_id, identity);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
#[test]
|
||||||
fn initial_snapshot_cursor_waits_for_all_files_and_recovers_twenty_rounds() {
|
fn initial_snapshot_cursor_waits_for_all_files_and_recovers_twenty_rounds() {
|
||||||
for _ in 0..20 {
|
for _ in 0..20 {
|
||||||
for committed in [false, true] {
|
for committed in [false, true] {
|
||||||
|
|||||||
@@ -71,7 +71,7 @@ impl Workspace {
|
|||||||
.query_row("SELECT EXISTS(SELECT 1 FROM sync_bindings)", [], |r| {
|
.query_row("SELECT EXISTS(SELECT 1 FROM sync_bindings)", [], |r| {
|
||||||
r.get(0)
|
r.get(0)
|
||||||
})?;
|
})?;
|
||||||
let entries = self.scan()?;
|
let paths = self.sync_paths()?;
|
||||||
let id = Uuid::new_v4().to_string();
|
let id = Uuid::new_v4().to_string();
|
||||||
// Rebinding explicitly starts from the current snapshot, never an old account's queue.
|
// Rebinding explicitly starts from the current snapshot, never an old account's queue.
|
||||||
if had_binding {
|
if had_binding {
|
||||||
@@ -80,15 +80,30 @@ impl Workspace {
|
|||||||
[],
|
[],
|
||||||
)?;
|
)?;
|
||||||
}
|
}
|
||||||
for entry in entries.into_iter().filter(|e| !e.is_folder && !e.deleted) {
|
for path in paths {
|
||||||
let queued: bool = self.db.query_row(
|
let entry = self.entry(&path)?;
|
||||||
"SELECT EXISTS(SELECT 1 FROM outbox WHERE file_id=?1 AND state='pending')",
|
let queued = if let Some(entry) = &entry {
|
||||||
[&entry.file_id],
|
self.db.query_row(
|
||||||
|r| r.get(0),
|
"SELECT EXISTS(SELECT 1 FROM outbox WHERE file_id=?1 AND state='pending')",
|
||||||
)?;
|
[&entry.file_id],
|
||||||
|
|r| r.get::<_, bool>(0),
|
||||||
|
)?
|
||||||
|
} else {
|
||||||
|
false
|
||||||
|
};
|
||||||
if !queued {
|
if !queued {
|
||||||
let content = fs::read(self.resolve(&entry.path)?)?;
|
let source = self.resolve(&path)?;
|
||||||
self.write(&entry.path, &entry.hash, &content, "local")?;
|
let (digest, size) = crate::payloads::sync_file_info(&source, &path)?;
|
||||||
|
let operation = Uuid::new_v4().to_string();
|
||||||
|
self.store_payload_file(&operation, &source, &digest)?;
|
||||||
|
self.write_spooled_with_identity(
|
||||||
|
&path,
|
||||||
|
&digest,
|
||||||
|
(&digest, size),
|
||||||
|
"local",
|
||||||
|
&operation,
|
||||||
|
None,
|
||||||
|
)?;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
self.db.execute(
|
self.db.execute(
|
||||||
|
|||||||
@@ -298,8 +298,7 @@ impl Workspace {
|
|||||||
});
|
});
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
let content = fs::read(self.resolve(&path)?)?;
|
let digest = crate::payloads::hash_file(&self.resolve(&path)?)?;
|
||||||
let digest = hash(&content);
|
|
||||||
let previous = self.entry(&path)?;
|
let previous = self.entry(&path)?;
|
||||||
if previous
|
if previous
|
||||||
.as_ref()
|
.as_ref()
|
||||||
|
|||||||
Reference in New Issue
Block a user