181aad91e6
Sync and the server - crates/sync: revision ledger (stale writers conflict, edit beats delete), client planner, three-way merge, conflict names, path and hash gates - apps/notescraft-sync-server: pairing, device tokens, change log with long poll, blobs, accounts with many vaults, roles, invites, audit log, quota, GC, readiness, request ids - server history (revisions, restore, retention), chunked resumable uploads, live editing endpoints (SSE rooms, presence, flush through the ledger) - engine: HttpRemote, sync_once, SyncWorker, sync.* commands, share.* and server.* commands, live session client, note.capture, vault catalog and vaults.* commands Editing together - crates/collab: operational transformation core (TextOp, transform, client state machine, room, presence) - ui-egui: Settings > Sync, sync pill, conflicts, Share dialog, live pill and remote carets, launcher Connect and Join flows App shell - vault launcher (desktop and phone layouts) wired into the app host (--launcher, File > Vaults...) - compact (phone) layout with a bottom tab bar Verified with cargo xtask ci (fmt, clippy -D warnings, workspace tests, layers, tree, i18n). Android and iOS shells are not built: the build machine has no NDK or Xcode. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
384 lines
14 KiB
Rust
384 lines
14 KiB
Rust
//! The sync driver against a server that lives in memory.
|
|
//!
|
|
//! Two vaults ("phone" and "laptop") talk to one fake server that keeps the
|
|
//! same records a real one does, so the rules of `notescraft_sync` decide
|
|
//! every outcome here.
|
|
|
|
use std::collections::HashMap;
|
|
use std::path::PathBuf;
|
|
use std::sync::{Arc, Mutex, MutexGuard};
|
|
|
|
use notescraft_engine::{Changes, FileRecord, HttpRemote, Pairing, PutReply, Remote, SyncError, SyncReport, SyncResult, SyncState, sync_once};
|
|
use notescraft_sync::{Base, Ledger, Outcome, Write, conflict_path, hash_hex};
|
|
use notescraft_vault::Vault;
|
|
|
|
/// The date text the conflict copies carry.
|
|
const STAMP: &str = "2026-10-20";
|
|
|
|
/// The server the two devices talk to.
|
|
struct Server {
|
|
inner: Mutex<Inner>,
|
|
}
|
|
|
|
/// What the server keeps.
|
|
struct Inner {
|
|
ledger: Ledger,
|
|
blobs: HashMap<String, Vec<u8>>,
|
|
changes_calls: usize,
|
|
writes: usize,
|
|
hostile: Option<FileRecord>,
|
|
}
|
|
|
|
impl Server {
|
|
fn new() -> Arc<Server> {
|
|
Arc::new(Server { inner: Mutex::new(Inner { ledger: Ledger::new(), blobs: HashMap::new(), changes_calls: 0, writes: 0, hostile: None }) })
|
|
}
|
|
|
|
fn lock(&self) -> MutexGuard<'_, Inner> {
|
|
self.inner.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
|
|
}
|
|
|
|
/// The newest revision of the server.
|
|
fn head(&self) -> u64 {
|
|
self.lock().ledger.head()
|
|
}
|
|
|
|
/// How many times the change log was read.
|
|
fn changes_calls(&self) -> usize {
|
|
self.lock().changes_calls
|
|
}
|
|
|
|
/// How many writes the server took.
|
|
fn writes(&self) -> usize {
|
|
self.lock().writes
|
|
}
|
|
|
|
/// Whether the server knows a path.
|
|
fn knows(&self, path: &str) -> bool {
|
|
self.lock().ledger.get(path).is_some()
|
|
}
|
|
|
|
/// Replaces the content of a path with something else.
|
|
fn corrupt(&self, path: &str, bytes: &[u8]) {
|
|
self.lock().blobs.insert(path.to_owned(), bytes.to_vec());
|
|
}
|
|
|
|
/// Makes the change log offer a path the client must refuse.
|
|
fn poison(&self, record: FileRecord) {
|
|
self.lock().hostile = Some(record);
|
|
}
|
|
}
|
|
|
|
/// The client the driver talks to.
|
|
struct FakeRemote {
|
|
server: Arc<Server>,
|
|
}
|
|
|
|
impl Remote for FakeRemote {
|
|
fn changes(&self, since: u64, limit: usize, _wait: u64) -> SyncResult<Changes> {
|
|
let mut inner = self.server.lock();
|
|
inner.changes_calls += 1;
|
|
let mut page = inner.ledger.changes_since(since, limit);
|
|
if let Some(hostile) = inner.hostile.clone() {
|
|
page.records.insert(0, hostile);
|
|
page.more = true;
|
|
}
|
|
Ok(page)
|
|
}
|
|
|
|
fn get(&self, path: &str) -> SyncResult<Vec<u8>> {
|
|
let inner = self.server.lock();
|
|
inner.blobs.get(path).cloned().ok_or_else(|| SyncError::Status { status: 404, body: path.to_owned() })
|
|
}
|
|
|
|
fn put(&self, path: &str, bytes: &[u8], base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
|
|
let mut inner = self.server.lock();
|
|
let outcome = inner
|
|
.ledger
|
|
.apply(Write { path: path.to_owned(), hash: Some(hash_hex(bytes)), size: bytes.len() as u64, mtime_ms, base_seq })
|
|
.map_err(|e| SyncError::Status { status: 400, body: e.to_string() })?;
|
|
inner.writes += 1;
|
|
match outcome {
|
|
Outcome::Applied(record) => {
|
|
inner.blobs.insert(path.to_owned(), bytes.to_vec());
|
|
Ok(PutReply::Applied(record))
|
|
}
|
|
Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)),
|
|
Outcome::Conflict(record) => Ok(PutReply::Conflict(record)),
|
|
}
|
|
}
|
|
|
|
fn delete(&self, path: &str, base_seq: u64) -> SyncResult<PutReply> {
|
|
let mut inner = self.server.lock();
|
|
let outcome = inner
|
|
.ledger
|
|
.apply(Write { path: path.to_owned(), hash: None, size: 0, mtime_ms: 0, base_seq })
|
|
.map_err(|e| SyncError::Status { status: 400, body: e.to_string() })?;
|
|
inner.writes += 1;
|
|
match outcome {
|
|
Outcome::Applied(record) => Ok(PutReply::Applied(record)),
|
|
Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)),
|
|
Outcome::Conflict(record) => Ok(PutReply::Conflict(record)),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A vault in a fresh folder.
|
|
fn temp_vault(name: &str) -> (Vault, PathBuf) {
|
|
let root = std::env::temp_dir().join(format!("nc-engine-sync-{name}-{}", std::process::id()));
|
|
let _ = std::fs::remove_dir_all(&root);
|
|
std::fs::create_dir_all(&root).expect("temp folder");
|
|
let vault = Vault::open(&root).expect("vault");
|
|
(vault, root)
|
|
}
|
|
|
|
/// A device with its own state.
|
|
struct Device {
|
|
vault: Vault,
|
|
state: SyncState,
|
|
remote: FakeRemote,
|
|
}
|
|
|
|
impl Device {
|
|
fn new(name: &str, server: &Arc<Server>) -> Self {
|
|
let (vault, _root) = temp_vault(name);
|
|
Self { vault, state: SyncState::default(), remote: FakeRemote { server: server.clone() } }
|
|
}
|
|
|
|
/// Syncs once and returns the report.
|
|
fn sync(&mut self, device: &str) -> SyncReport {
|
|
sync_once(&self.vault, &self.remote, &mut self.state, device, STAMP).expect("sync")
|
|
}
|
|
|
|
/// The text of a note.
|
|
fn read(&self, path: &str) -> String {
|
|
String::from_utf8(self.vault.read(path).expect("read")).expect("text")
|
|
}
|
|
|
|
/// Writes a note.
|
|
fn write(&self, path: &str, text: &str) {
|
|
self.vault.write(path, text.as_bytes()).expect("write");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn the_http_client_is_built_without_touching_the_network() {
|
|
let _remote = HttpRemote::new("https://sync.example/", "v-1234", "token");
|
|
let pairing = Pairing { device_id: "d-1".to_string(), vault_id: "v-1234".to_string(), token: "secret".to_string() };
|
|
assert_eq!(pairing.vault_id, "v-1234");
|
|
}
|
|
|
|
#[test]
|
|
fn the_first_sync_uploads_every_note_and_the_second_does_nothing() {
|
|
let server = Server::new();
|
|
let mut phone = Device::new("first-phone", &server);
|
|
phone.write("a.md", "# one\n");
|
|
phone.write("notes/b.md", "# two\n");
|
|
|
|
let report = phone.sync("phone");
|
|
assert_eq!(report.uploaded, vec!["a.md".to_string(), "notes/b.md".to_string()]);
|
|
assert_eq!(server.head(), 2);
|
|
assert_eq!(report.conflicts, Vec::<String>::new());
|
|
|
|
let writes = server.writes();
|
|
let calls = server.changes_calls();
|
|
let again = phone.sync("phone");
|
|
assert_eq!(again, SyncReport::default());
|
|
assert_eq!(server.head(), 2);
|
|
assert_eq!(server.writes(), writes, "the second run wrote to the server");
|
|
assert_eq!(server.changes_calls() - calls, 1, "the second run asked the change log once");
|
|
}
|
|
|
|
#[test]
|
|
fn a_second_device_ends_with_the_same_files_and_bases() {
|
|
let server = Server::new();
|
|
let mut phone = Device::new("same-phone", &server);
|
|
let mut laptop = Device::new("same-laptop", &server);
|
|
phone.write("a.md", "# one\n");
|
|
phone.sync("phone");
|
|
|
|
let report = laptop.sync("laptop");
|
|
assert_eq!(report.downloaded, vec!["a.md".to_string()]);
|
|
assert_eq!(laptop.read("a.md"), phone.read("a.md"));
|
|
assert_eq!(laptop.state.bases, phone.state.bases);
|
|
}
|
|
|
|
#[test]
|
|
fn two_devices_editing_different_lines_end_with_the_merged_text() {
|
|
let server = Server::new();
|
|
let mut phone = Device::new("merge-phone", &server);
|
|
let mut laptop = Device::new("merge-laptop", &server);
|
|
phone.write("a.md", "line1\nline2\nline3\n");
|
|
phone.sync("phone");
|
|
laptop.sync("laptop");
|
|
|
|
phone.write("a.md", "phone\nline2\nline3\n");
|
|
laptop.write("a.md", "line1\nline2\nlaptop\n");
|
|
phone.sync("phone");
|
|
let report = laptop.sync("laptop");
|
|
|
|
assert_eq!(report.merged, vec!["a.md".to_string()]);
|
|
assert_eq!(laptop.read("a.md"), "phone\nline2\nlaptop\n");
|
|
assert_eq!(report.conflicts, Vec::<String>::new());
|
|
|
|
// The phone picks the merged text up on its next pass.
|
|
phone.sync("phone");
|
|
assert_eq!(phone.read("a.md"), "phone\nline2\nlaptop\n");
|
|
}
|
|
|
|
#[test]
|
|
fn two_devices_editing_the_same_line_keep_both_versions() {
|
|
let server = Server::new();
|
|
let mut phone = Device::new("conflict-phone", &server);
|
|
let mut laptop = Device::new("conflict-laptop", &server);
|
|
phone.write("a.md", "line1\nline2\nline3\n");
|
|
phone.sync("phone");
|
|
laptop.sync("laptop");
|
|
|
|
phone.write("a.md", "line1\nphone\nline3\n");
|
|
laptop.write("a.md", "line1\nlaptop\nline3\n");
|
|
phone.sync("phone");
|
|
let report = laptop.sync("laptop");
|
|
|
|
let copy = conflict_path("a.md", "laptop", STAMP);
|
|
assert_eq!(report.conflicts, vec![copy.clone()]);
|
|
assert_eq!(laptop.read("a.md"), "line1\nphone\nline3\n", "the first synced version wins the path");
|
|
assert_eq!(laptop.read(©), "line1\nlaptop\nline3\n", "the other version is kept");
|
|
assert!(server.knows(©), "the copy was uploaded");
|
|
|
|
// The other device gets the copy too, and keeps its own version.
|
|
phone.sync("phone");
|
|
assert_eq!(phone.read("a.md"), "line1\nphone\nline3\n");
|
|
assert_eq!(phone.read(©), "line1\nlaptop\nline3\n");
|
|
}
|
|
|
|
#[test]
|
|
fn a_second_conflict_of_the_same_day_gets_its_own_copy() {
|
|
let server = Server::new();
|
|
let mut phone = Device::new("twice-phone", &server);
|
|
let mut laptop = Device::new("twice-laptop", &server);
|
|
phone.write("a.md", "line1\nline2\n");
|
|
phone.sync("phone");
|
|
laptop.sync("laptop");
|
|
|
|
phone.write("a.md", "line1\nphone\n");
|
|
laptop.write("a.md", "line1\nlaptop\n");
|
|
phone.sync("phone");
|
|
laptop.sync("laptop");
|
|
|
|
// Both sides change the file again on the same day.
|
|
phone.write("a.md", "line1\nphone two\n");
|
|
laptop.write("a.md", "line1\nlaptop two\n");
|
|
phone.sync("phone");
|
|
let report = laptop.sync("laptop");
|
|
|
|
let first = conflict_path("a.md", "laptop", STAMP);
|
|
let second = conflict_path("a.md", "laptop", &format!("{STAMP}-2"));
|
|
assert_eq!(report.conflicts, vec![second.clone()]);
|
|
assert_eq!(laptop.read(&first), "line1\nlaptop\n", "the first copy is still there");
|
|
assert_eq!(laptop.read(&second), "line1\nlaptop two\n", "the second version has its own copy");
|
|
}
|
|
|
|
#[test]
|
|
fn a_delete_reaches_the_other_device() {
|
|
let server = Server::new();
|
|
let mut phone = Device::new("delete-phone", &server);
|
|
let mut laptop = Device::new("delete-laptop", &server);
|
|
phone.write("a.md", "# one\n");
|
|
phone.sync("phone");
|
|
laptop.sync("laptop");
|
|
|
|
phone.vault.delete_to_trash("a.md").expect("delete");
|
|
let report = phone.sync("phone");
|
|
assert_eq!(report.deleted_remote, vec!["a.md".to_string()]);
|
|
|
|
let report = laptop.sync("laptop");
|
|
assert_eq!(report.deleted_local, vec!["a.md".to_string()]);
|
|
assert!(laptop.vault.read("a.md").is_err(), "the file is gone");
|
|
let trash = laptop.vault.root().join(".notescraft/trash");
|
|
let kept = std::fs::read_dir(&trash).expect("trash").count();
|
|
assert_eq!(kept, 1, "the file waits in the trash");
|
|
}
|
|
|
|
#[test]
|
|
fn an_edit_wins_over_a_delete() {
|
|
let server = Server::new();
|
|
let mut phone = Device::new("edit-phone", &server);
|
|
let mut laptop = Device::new("edit-laptop", &server);
|
|
phone.write("a.md", "# one\n");
|
|
phone.sync("phone");
|
|
laptop.sync("laptop");
|
|
|
|
laptop.vault.delete_to_trash("a.md").expect("delete");
|
|
laptop.sync("laptop");
|
|
phone.write("a.md", "# edited\n");
|
|
phone.sync("phone");
|
|
|
|
assert_eq!(phone.read("a.md"), "# edited\n");
|
|
laptop.sync("laptop");
|
|
assert_eq!(laptop.read("a.md"), "# edited\n", "the edit came back");
|
|
}
|
|
|
|
#[test]
|
|
fn the_folders_that_never_sync_are_never_uploaded() {
|
|
let server = Server::new();
|
|
let mut device = Device::new("reserved", &server);
|
|
let root = device.vault.root().to_path_buf();
|
|
for path in [".obsidian/app.json", ".notescraft/notes.bin", ".git/config"] {
|
|
let full = root.join(path);
|
|
std::fs::create_dir_all(full.parent().expect("parent")).expect("folder");
|
|
std::fs::write(&full, b"private").expect("write");
|
|
}
|
|
device.write("a.md", "# one\n");
|
|
|
|
let report = device.sync("phone");
|
|
assert_eq!(report.uploaded, vec!["a.md".to_string()]);
|
|
for path in [".obsidian/app.json", ".notescraft/notes.bin", ".git/config"] {
|
|
assert!(!server.knows(path), "{path} must never be uploaded");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_hostile_path_from_the_server_is_refused() {
|
|
let server = Server::new();
|
|
let mut device = Device::new("hostile", &server);
|
|
server.poison(FileRecord { path: "../evil.md".to_string(), hash: hash_hex(b"evil"), size: 4, mtime_ms: 0, seq: 1, deleted: false });
|
|
|
|
let error = sync_once(&device.vault, &device.remote, &mut device.state, "laptop", STAMP).expect_err("hostile");
|
|
assert!(matches!(error, SyncError::BadRemote(_)), "{error:?}");
|
|
assert!(!std::env::temp_dir().join("evil.md").exists(), "nothing was written outside the vault");
|
|
assert_eq!(server.writes(), 0, "nothing was uploaded");
|
|
}
|
|
|
|
#[test]
|
|
fn a_blob_that_does_not_match_its_hash_is_refused() {
|
|
let server = Server::new();
|
|
let mut phone = Device::new("blob-phone", &server);
|
|
let mut laptop = Device::new("blob-laptop", &server);
|
|
phone.write("a.md", "# one\n");
|
|
phone.sync("phone");
|
|
laptop.sync("laptop");
|
|
|
|
phone.write("a.md", "# two\n");
|
|
phone.sync("phone");
|
|
server.corrupt("a.md", b"tampered");
|
|
|
|
let error = sync_once(&laptop.vault, &laptop.remote, &mut laptop.state, "laptop", STAMP).expect_err("tampered");
|
|
assert!(matches!(error, SyncError::BadRemote(_)), "{error:?}");
|
|
assert_eq!(laptop.read("a.md"), "# one\n", "the good copy stayed");
|
|
}
|
|
|
|
#[test]
|
|
fn the_state_survives_a_save_and_a_broken_one_is_an_error() {
|
|
let (vault, _root) = temp_vault("state");
|
|
let mut state = SyncState { last_seq: 7, ..SyncState::default() };
|
|
state.bases.insert("a.md".to_string(), Base { hash: hash_hex(b"x"), seq: 3 });
|
|
state.save(&vault).expect("save");
|
|
|
|
let loaded = SyncState::load(&vault).expect("load");
|
|
assert_eq!(loaded, state);
|
|
|
|
std::fs::write(vault.root().join(".notescraft/sync/state.json"), b"{ not json").expect("write");
|
|
assert!(SyncState::load(&vault).is_err(), "a broken state is never a silent reset");
|
|
}
|