Kronecker/NotesAgentic
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 5 Wiki Activity

feat: 预览并恢复首次 Sync 合并且不丢失笔记链接

Browse Source
This commit is contained in:
admin
2026-09-08 16:34:00 +08:00
parent f08a338ece
commit ef722145c4
19 changed files with 885 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src-tauri/build.rs
+1
View File
@@ -20,6 +20,7 @@ fn main() {
"sync_vaults",
"sync_create_vault",
"sync_bind",
"sync_preview",
"sync_unbind",
"sync_pause",
"sync_status",
frontend/src-tauri/capabilities/main.json
+2 -1
View File
@@ -46,6 +46,7 @@
"allow-sync-status",
"allow-sync-resolve",
"allow-sync-logout",
"allow-sync-run"
"allow-sync-run",
"allow-sync-preview"
]
}
frontend/src-tauri/permissions/autogenerated/sync_preview.toml
+11
View File
@@ -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"]
frontend/src-tauri/src/lib.rs
+1
View File
@@ -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;
frontend/src-tauri/src/main.rs
+1
View File
@@ -835,6 +835,7 @@ fn main() {
sync_vaults,
sync_create_vault,
sync_bind,
sync_preview,
sync_unbind,
sync_pause,
sync_status,
frontend/src-tauri/src/sync_client.rs
+64
View File
@@ -222,6 +222,50 @@ impl SyncClient {
}
Ok(())
}
pub async fn snapshot(&self, remote_vault: &str) -> Result<crate::sync_initial::Snapshot> {
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((
frontend/src-tauri/src/sync_commands.rs
+49
View File
@@ -119,6 +119,37 @@ pub struct Bind {
account: String,
remote_vault: String,
mode: String,
#[serde(default)]
fingerprint: Option<String>,
}
#[tauri::command]
pub async fn sync_preview(
host: State<'_, Host>,
request: Bind,
) -> Result<notesagent_host::sync_initial::Preview, String> {
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<Binding, String> {
@@ -131,6 +162,24 @@ pub async fn sync_bind(host: State<'_, Host>, request: Bind) -> Result<Binding,
)
.await
.map_err(|e| e.code)?;
if request.mode == "merge" {
let snapshot =
sync_auth::guarded(&host.credentials, client.snapshot(&request.remote_vault))
.await
.map_err(|e| e.code)?;
return with_workspace(&host, |ws| {
if ws.vault_id != request.vault_id {
return Err(notesagent_host::workspace::HostError::new("VAULT_CHANGED"));
}
ws.sync_bind_initial(
&request.endpoint,
&request.remote_vault,
&request.account,
&snapshot,
request.fingerprint.as_deref().unwrap_or(""),
)
});
}
if request.mode == "upload" {
sync_auth::guarded(
&host.credentials,
frontend/src-tauri/src/sync_inbox.rs
+46 -8
View File
@@ -136,7 +136,13 @@ impl Workspace {
.ok_or_else(|| HostError::new("SYNC_BINDING_CHANGED"))?;
revision.validate(&active)?;
self.resolve(&revision.path)?;
if revision.sequence != active.cursor + 1
let encoded =
serde_json::to_string(revision).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?;
if let Some(initial) = self.initial_revision(binding, revision.sequence)? {
if initial != encoded {
return Err(HostError::new("SYNC_REVISION_CHANGED"));
}
} else if revision.sequence != active.cursor + 1
|| self
.sync_boundary(binding)?
.is_none_or(|end| revision.sequence > 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<i64> = 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<excluded.revision", params![binding,revision.file_id,revision.sequence,revision.path,revision.hash.as_deref().unwrap_or("")])?;
tx.execute(
"UPDATE sync_inbox SET state=?3 WHERE binding=?1 AND sequence=?2",
params![binding, revision.sequence, state],
)?;
if let Some(boundary) = initial {
let pending:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM sync_initial_items i WHERE i.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'))",[binding],|r|r.get(0))?;
if !pending {
tx.execute(
"UPDATE sync_bindings SET cursor=?2 WHERE id=?1",
params![binding, boundary],
)?;
tx.execute("DELETE FROM sync_initial WHERE binding=?1", [binding])?;
tx.execute("DELETE FROM sync_initial_items WHERE binding=?1", [binding])?;
}
}
tx.execute(
"DELETE FROM sync_windows WHERE binding=?1 AND boundary=?2",
params![binding, revision.sequence],
@@ -223,7 +249,7 @@ impl Workspace {
remote
],
)?;
tx.execute("UPDATE sync_jobs SET state='conflict' WHERE binding=?1 AND file_id=?2 AND state NOT IN ('acked','archived')", params![binding,revision.file_id])?;
tx.execute("UPDATE sync_jobs SET state='conflict' WHERE binding=?1 AND (file_id=?2 OR path=?3) AND state NOT IN ('acked','archived')", params![binding,revision.file_id,local_path])?;
tx.commit()?;
self.sync_finish(binding, revision, "conflict")
}
@@ -255,6 +281,18 @@ impl Workspace {
self.sync_finish(binding, &revision, "applied")?;
return Ok(true);
}
let known: Option<i64> = self
.db
.query_row(
"SELECT revision FROM sync_heads WHERE binding=?1 AND file_id=?2",
params![binding, revision.file_id],
|r| r.get(0),
)
.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)?;
frontend/src-tauri/src/sync_initial.rs
+372
View File
@@ -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<RemoteRevision>,
}
#[derive(Serialize)]
pub struct Preview {
pub fingerprint: String,
pub boundary: i64,
pub items: Vec<PreviewItem>,
}
#[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<Vec<Local>> {
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<Preview> {
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<Binding> {
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<Option<Vec<RemoteRevision>>> {
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::<std::result::Result<Vec<_>, _>>()?;
Ok(Some(
rows.iter()
.map(|v| {
serde_json::from_str(v).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))
})
.collect::<Result<_>>()?,
))
}
pub(crate) fn initial_revision(&self, binding: &str, sequence: i64) -> Result<Option<String>> {
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");
}
}
frontend/src-tauri/src/sync_resolution.rs
+104 -7
View File
@@ -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", &copy)?;
}
if self
.entry(&path)?
.is_some_and(|entry| !entry.deleted && entry.file_id != revision.file_id)
{
self.mutate_with_origin("delete", &path, "", &current, &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,
&current,
&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", &copy)
.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);
}
}
}
}
}
}
frontend/src-tauri/src/sync_state.rs
+1 -1
View File
@@ -182,7 +182,7 @@ impl Workspace {
}
pub fn sync_next(&self, binding: &str) -> Result<Option<Job>> {
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
frontend/src-tauri/src/workspace.rs
+16 -4
View File
@@ -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<Vec<String>> {
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::<std::result::Result<Vec<_>, _>>()?;
Ok(rows)
}
pub fn path_for_id(&self, file_id: &str) -> Result<String> {
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),
)
frontend/src-tauri/src/workspace_broker.rs
+13 -3
View File
@@ -64,9 +64,19 @@ pub fn dispatch(ws: &mut Workspace, request: &Value) -> Result<Value, String> {
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::<Vec<_>>(),"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::<Result<Vec<_>, String>>()?;
Ok(json!({"items":items,"total":total}))
}
"workspace.read" => {
let p: Read = decode(params)?;
frontend/src-tauri/tests/sync_push.rs
+119
View File
@@ -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};