Marcos Paulo
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>
426 lines
18 KiB
Rust
426 lines
18 KiB
Rust
//! The live endpoints over real HTTP: rooms, operations, the event stream, presence
|
|
//! and the flush of a room's text into the vault.
|
|
//!
|
|
//! Time is injected, never waited for: the flush is called directly and the presence
|
|
//! is pruned with an explicit clock. The event stream is read by a thread that takes
|
|
//! the Server-Sent Events line by line.
|
|
|
|
mod common;
|
|
|
|
use std::io::BufRead;
|
|
use std::sync::mpsc::Receiver;
|
|
use std::time::Duration;
|
|
|
|
use common::Server;
|
|
use notescraft_collab::{Client, HISTORY_CAP, TextOp};
|
|
use notescraft_sync_server::{AppState, Store, Timings, flush, now_ms};
|
|
use serde_json::{Value, json};
|
|
|
|
/// The live windows the tests use: a short keepalive, no background sweep.
|
|
fn windows() -> Timings {
|
|
Timings { keepalive_ms: 200, idle_ms: 5_000, presence_ttl_ms: 30_000, flush_debounce_ms: 60_000, sweep_ms: 60_000, max_rooms: 8, max_peers: 64 }
|
|
}
|
|
|
|
/// A server over a fresh folder with the given live windows.
|
|
fn server(name: &str, windows: Timings) -> Server {
|
|
let dir = common::temp_dir(name);
|
|
let state = AppState::new(Store::open(&dir.join("sync.db")).expect("db"), dir.join("blobs"), common::ADMIN);
|
|
state.set_live_timings(windows);
|
|
common::start_with_state(state)
|
|
}
|
|
|
|
/// `POST .../join`: the snapshot a peer adopts.
|
|
fn join(server: &Server, vault: &str, token: &str, path: &str, client: &str, name: &str) -> (u16, Value) {
|
|
server.json("POST", &format!("/v1/vaults/{vault}/live/{path}/join"), Some(token), &[], Some(&json!({"client": client, "name": name})))
|
|
}
|
|
|
|
/// `POST .../ops`: one operation of a peer.
|
|
fn ops(server: &Server, vault: &str, token: &str, path: &str, client: &str, base_rev: u64, op: &TextOp) -> (u16, Value) {
|
|
server.json("POST", &format!("/v1/vaults/{vault}/live/{path}/ops"), Some(token), &[], Some(&json!({"client": client, "base_rev": base_rev, "op": op})))
|
|
}
|
|
|
|
/// The event stream of a peer, read line by line in a thread.
|
|
struct Stream {
|
|
lines: Receiver<String>,
|
|
}
|
|
|
|
impl Stream {
|
|
/// Opens the stream of a peer that already joined the room.
|
|
fn open(server: &Server, vault: &str, token: &str, path: &str, client: &str) -> Stream {
|
|
let url = format!("{}/v1/vaults/{vault}/live/{path}/events?client={client}", server.base);
|
|
let token = token.to_owned();
|
|
let agent = server.agent.clone();
|
|
let (tx, lines) = std::sync::mpsc::channel();
|
|
std::thread::spawn(move || {
|
|
let Ok(response) = agent.get(&url).header("Authorization", format!("Bearer {token}")).call() else { return };
|
|
let reader = std::io::BufReader::new(response.into_body().into_reader());
|
|
for line in reader.lines() {
|
|
let Ok(line) = line else { break };
|
|
if tx.send(line).is_err() {
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
Stream { lines }
|
|
}
|
|
|
|
/// The next raw line of the stream.
|
|
fn line(&self) -> Option<String> {
|
|
self.lines.recv_timeout(Duration::from_secs(10)).ok()
|
|
}
|
|
|
|
/// The next event `(name, data)`; comments are skipped.
|
|
fn next(&self) -> Option<(String, String)> {
|
|
let mut name = String::new();
|
|
let mut data = String::new();
|
|
loop {
|
|
let line = self.line()?;
|
|
if line.is_empty() {
|
|
if !name.is_empty() {
|
|
return Some((name, data));
|
|
}
|
|
continue;
|
|
}
|
|
if let Some(rest) = line.strip_prefix("event: ") {
|
|
rest.clone_into(&mut name);
|
|
} else if let Some(rest) = line.strip_prefix("data: ") {
|
|
rest.clone_into(&mut data);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The next event with the given name.
|
|
fn next_named(&self, wanted: &str) -> Option<(String, String)> {
|
|
loop {
|
|
let (name, data) = self.next()?;
|
|
if name == wanted {
|
|
return Some((name, data));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The operation an event carries.
|
|
fn event_op(data: &str) -> TextOp {
|
|
let value: Value = serde_json::from_str(data).expect("event data");
|
|
serde_json::from_value(value["op"].clone()).expect("operation")
|
|
}
|
|
|
|
/// The body the file endpoint serves.
|
|
fn file_body(server: &Server, vault: &str, token: &str, path: &str) -> (u16, String) {
|
|
let (status, body) = server.send("GET", &format!("/v1/vaults/{vault}/files/{path}"), Some(token), &[], None);
|
|
(status, String::from_utf8_lossy(&body).into_owned())
|
|
}
|
|
|
|
/// A path with the spaces and brackets a conflict copy carries, in URL form.
|
|
fn url_path(path: &str) -> String {
|
|
path.replace(' ', "%20").replace('(', "%28").replace(')', "%29")
|
|
}
|
|
|
|
#[test]
|
|
fn joining_answers_the_snapshot_and_the_revision() {
|
|
let server = server("live-join", windows());
|
|
let (vault, token, _) = server.owned_vault("live-join");
|
|
assert_eq!(server.put(&vault, &token, "note.md", "hello", 0).0, 201);
|
|
|
|
let (status, reply) = join(&server, &vault, &token, "note.md", "client-a", "Ana");
|
|
assert_eq!(status, 200);
|
|
assert_eq!(reply["text"], json!("hello"), "the room starts with the file's text");
|
|
assert_eq!(reply["rev"], json!(0), "a new room is at revision zero");
|
|
let peers = reply["peers"].as_array().expect("peers");
|
|
assert_eq!(peers.len(), 1);
|
|
assert_eq!(peers[0]["client"], json!("client-a"));
|
|
assert_eq!(peers[0]["name"], json!("Ana"));
|
|
|
|
// A second peer sees the same room.
|
|
let (status, reply) = join(&server, &vault, &token, "note.md", "client-b", "Bo");
|
|
assert_eq!(status, 200);
|
|
assert_eq!(reply["rev"], json!(0));
|
|
assert_eq!(reply["peers"].as_array().map(Vec::len), Some(2));
|
|
}
|
|
|
|
#[test]
|
|
fn two_clients_converge_on_the_room_text() {
|
|
let server = server("live-converge", windows());
|
|
let (vault, token, _) = server.owned_vault("live-converge");
|
|
assert_eq!(server.put(&vault, &token, "note.md", "hi", 0).0, 201);
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "b", "Bo").0, 200);
|
|
|
|
let stream_a = Stream::open(&server, &vault, &token, "note.md", "a");
|
|
let stream_b = Stream::open(&server, &vault, &token, "note.md", "b");
|
|
|
|
// Both edit the same base revision.
|
|
let mut client_a = Client::new(0);
|
|
let mut client_b = Client::new(0);
|
|
let edit_a = TextOp::new().retain(2).insert("A");
|
|
let edit_b = TextOp::new().retain(2).insert("B");
|
|
let submit_a = client_a.local_edit(edit_a.clone()).expect("edit").expect("submit");
|
|
let submit_b = client_b.local_edit(edit_b.clone()).expect("edit").expect("submit");
|
|
let mut text_a = edit_a.apply("hi").expect("apply");
|
|
let mut text_b = edit_b.apply("hi").expect("apply");
|
|
|
|
// The first operation is accepted as it is.
|
|
let (status, reply) = ops(&server, &vault, &token, "note.md", "a", submit_a.base_rev, &submit_a.op);
|
|
assert_eq!(status, 200);
|
|
assert_eq!(reply["rev"], json!(1));
|
|
// The second one is transformed against it.
|
|
let (status, reply) = ops(&server, &vault, &token, "note.md", "b", submit_b.base_rev, &submit_b.op);
|
|
assert_eq!(status, 200);
|
|
assert_eq!(reply["rev"], json!(2));
|
|
|
|
// A hears its own acknowledgement first, then the other operation.
|
|
let (name, data) = stream_a.next_named("ack").expect("ack");
|
|
assert_eq!(name, "ack");
|
|
assert_eq!(data, json!({"rev": 1}).to_string());
|
|
client_a.server_ack().expect("ack");
|
|
let (_name, data) = stream_a.next_named("op").expect("op");
|
|
let applied = client_a.remote_op(event_op(&data)).expect("remote");
|
|
text_a = applied.apply(&text_a).expect("apply");
|
|
|
|
// B hears the other operation first, then its own acknowledgement.
|
|
let (_name, data) = stream_b.next_named("op").expect("op");
|
|
let applied = client_b.remote_op(event_op(&data)).expect("remote");
|
|
text_b = applied.apply(&text_b).expect("apply");
|
|
client_b.server_ack().expect("ack");
|
|
|
|
// Both hold the text the room holds.
|
|
let (status, joined) = join(&server, &vault, &token, "note.md", "c", "Cy");
|
|
assert_eq!(status, 200);
|
|
let room = joined["text"].as_str().expect("text");
|
|
assert_eq!(text_a, room, "the first client converged");
|
|
assert_eq!(text_b, room, "the second client converged");
|
|
}
|
|
|
|
#[test]
|
|
fn a_viewer_may_join_but_not_edit() {
|
|
let server = server("live-viewer", windows());
|
|
let (vault, owner, _) = server.owned_vault("live-viewer");
|
|
let (_, viewer) = server.paired("live-viewer");
|
|
let (status, code) = server.invite(&vault, &owner, "viewer");
|
|
assert_eq!(status, 200);
|
|
assert_eq!(server.accept(&viewer, &code).0, 200);
|
|
|
|
// The viewer joins the room and sees it.
|
|
let (status, reply) = join(&server, &vault, &viewer, "note.md", "v", "Vic");
|
|
assert_eq!(status, 200);
|
|
assert_eq!(reply["text"], json!(""));
|
|
|
|
// Its operations are refused.
|
|
let op = TextOp::new().insert("x");
|
|
let (status, _reply) = ops(&server, &vault, &viewer, "note.md", "v", 0, &op);
|
|
assert_eq!(status, 403);
|
|
}
|
|
|
|
#[test]
|
|
fn a_base_revision_older_than_the_history_is_refused() {
|
|
let server = server("live-stale", windows());
|
|
let (vault, token, _) = server.owned_vault("live-stale");
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
|
|
// The room keeps a bounded history; past it a client has to rejoin.
|
|
let mut len = 0usize;
|
|
for rev in 0..=HISTORY_CAP as u64 {
|
|
let op = TextOp::new().retain(len).insert("x");
|
|
let (status, _reply) = ops(&server, &vault, &token, "note.md", "a", rev, &op);
|
|
assert_eq!(status, 200, "revision {rev}");
|
|
len += 1;
|
|
}
|
|
|
|
// The first revision is out of the kept history now.
|
|
let op = TextOp::new().retain(len).insert("x");
|
|
let (status, reply) = ops(&server, &vault, &token, "note.md", "a", 0, &op);
|
|
assert_eq!(status, 409);
|
|
assert_eq!(reply["error"], json!("stale"), "the client must rejoin");
|
|
}
|
|
|
|
#[test]
|
|
fn a_malformed_operation_is_a_bad_request() {
|
|
let server = server("live-malformed", windows());
|
|
let (vault, token, _) = server.owned_vault("live-malformed");
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
|
|
// Not an operation at all.
|
|
let (status, _reply) =
|
|
server.json("POST", &format!("/v1/vaults/{vault}/live/note.md/ops"), Some(&token), &[], Some(&json!({"client": "a", "base_rev": 0, "op": "nonsense"})));
|
|
assert_eq!(status, 400);
|
|
|
|
// Not JSON at all.
|
|
let (status, _body) = server.send("POST", &format!("/v1/vaults/{vault}/live/note.md/ops"), Some(&token), &[], Some(b"{not json"));
|
|
assert_eq!(status, 400);
|
|
}
|
|
|
|
#[test]
|
|
fn a_body_over_the_cap_is_refused() {
|
|
let server = server("live-oversize", windows());
|
|
let (vault, token, _) = server.owned_vault("live-oversize");
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
|
|
let big = "x".repeat(1024 * 1024 + 64);
|
|
let body = json!({"client": "a", "base_rev": 0, "op": TextOp::new().insert(&big)}).to_string();
|
|
let (status, _reply) = server.send("POST", &format!("/v1/vaults/{vault}/live/note.md/ops"), Some(&token), &[], Some(body.as_bytes()));
|
|
assert_eq!(status, 413);
|
|
}
|
|
|
|
#[test]
|
|
fn a_file_that_is_too_large_or_not_text_cannot_be_joined() {
|
|
let server = server("live-files", windows());
|
|
let (vault, token, _) = server.owned_vault("live-files");
|
|
|
|
// A file over the document cap.
|
|
let big = "x".repeat(2 * 1024 * 1024 + 1);
|
|
let (status, _body) = server.send("PUT", &format!("/v1/vaults/{vault}/files/big.md"), Some(&token), &[("X-Base-Seq", "0")], Some(big.as_bytes()));
|
|
assert_eq!(status, 201);
|
|
let (status, _reply) = join(&server, &vault, &token, "big.md", "a", "Ana");
|
|
assert_eq!(status, 413);
|
|
|
|
// A file that is not UTF-8.
|
|
let (status, _body) = server.send("PUT", &format!("/v1/vaults/{vault}/files/raw.md"), Some(&token), &[("X-Base-Seq", "0")], Some(&[0xff, 0xfe, 0x00]));
|
|
assert_eq!(status, 201);
|
|
let (status, _reply) = join(&server, &vault, &token, "raw.md", "a", "Ana");
|
|
assert_eq!(status, 415);
|
|
}
|
|
|
|
#[test]
|
|
fn presence_lists_the_peers_and_prunes_the_silent_ones() {
|
|
let server = server("live-presence", windows());
|
|
let (vault, token, _) = server.owned_vault("live-presence");
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "b", "Bo").0, 200);
|
|
|
|
// A heartbeat with a caret.
|
|
let (status, reply) = server.json(
|
|
"POST",
|
|
&format!("/v1/vaults/{vault}/live/note.md/presence"),
|
|
Some(&token),
|
|
&[],
|
|
Some(&json!({"client": "a", "selection": {"anchor": 1, "head": 1}})),
|
|
);
|
|
assert_eq!(status, 200);
|
|
let peers = reply["peers"].as_array().expect("peers");
|
|
assert_eq!(peers.len(), 2);
|
|
let first = peers.iter().find(|peer| peer["client"] == json!("a")).expect("peer a");
|
|
assert_eq!(first["selection"], json!({"anchor": 1, "head": 1}));
|
|
|
|
// Time passes: the silent peers are dropped.
|
|
let pruned = server.state.prune_live_presence(&vault, "note.md", now_ms() + 60_000);
|
|
assert_eq!(pruned.len(), 2, "both peers fell silent: {pruned:?}");
|
|
|
|
// A pruned peer is gone.
|
|
let (status, _reply) = server.json("POST", &format!("/v1/vaults/{vault}/live/note.md/presence"), Some(&token), &[], Some(&json!({"client": "a"})));
|
|
assert_eq!(status, 404);
|
|
}
|
|
|
|
#[test]
|
|
fn flushing_a_room_writes_its_text_to_the_vault() {
|
|
let server = server("live-flush", windows());
|
|
let (vault, token, _) = server.owned_vault("live-flush");
|
|
assert_eq!(server.put(&vault, &token, "note.md", "hi", 0).0, 201);
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
let edit = TextOp::new().retain(2).insert("!");
|
|
let (status, reply) = ops(&server, &vault, &token, "note.md", "a", 0, &edit);
|
|
assert_eq!(status, 200);
|
|
assert_eq!(reply["rev"], json!(1));
|
|
|
|
// The flush is a plain call: no waiting.
|
|
let report = flush(&server.state, &vault, "note.md", now_ms()).expect("flush");
|
|
assert!(report.flushed);
|
|
assert!(!report.conflict);
|
|
assert_eq!(report.path, "note.md");
|
|
|
|
// The file the normal endpoint serves is the room's text.
|
|
let (status, body) = file_body(&server, &vault, &token, "note.md");
|
|
assert_eq!(status, 200);
|
|
assert_eq!(body, "hi!");
|
|
}
|
|
|
|
#[test]
|
|
fn flushing_after_an_external_write_makes_a_conflict_copy() {
|
|
let server = server("live-conflict", windows());
|
|
let (vault, token, _) = server.owned_vault("live-conflict");
|
|
assert_eq!(server.put(&vault, &token, "note.md", "one", 0).0, 201);
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
let edit = TextOp::new().retain(3).insert("!");
|
|
assert_eq!(ops(&server, &vault, &token, "note.md", "a", 0, &edit).0, 200);
|
|
|
|
// Someone else writes the file while the room is open.
|
|
assert_eq!(server.put(&vault, &token, "note.md", "two", 1).0, 201);
|
|
|
|
let report = flush(&server.state, &vault, "note.md", now_ms()).expect("flush");
|
|
assert!(report.flushed);
|
|
assert!(report.conflict, "the ledger held a newer version");
|
|
assert_ne!(report.path, "note.md");
|
|
|
|
// The file itself is untouched.
|
|
let (status, body) = file_body(&server, &vault, &token, "note.md");
|
|
assert_eq!(status, 200);
|
|
assert_eq!(body, "two", "the external write stands");
|
|
|
|
// The room's text went to the conflict copy.
|
|
let (status, body) = file_body(&server, &vault, &token, &url_path(&report.path));
|
|
assert_eq!(status, 200);
|
|
assert_eq!(body, "one!");
|
|
}
|
|
|
|
#[test]
|
|
fn the_last_peer_leaving_writes_the_room_text() {
|
|
let server = server("live-leave", windows());
|
|
let (vault, token, _) = server.owned_vault("live-leave");
|
|
assert_eq!(server.put(&vault, &token, "note.md", "hi", 0).0, 201);
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
let edit = TextOp::new().retain(2).insert("!");
|
|
assert_eq!(ops(&server, &vault, &token, "note.md", "a", 0, &edit).0, 200);
|
|
|
|
let (status, _reply) = server.json("POST", &format!("/v1/vaults/{vault}/live/note.md/leave"), Some(&token), &[], Some(&json!({"client": "a"})));
|
|
assert_eq!(status, 204);
|
|
|
|
let (status, body) = file_body(&server, &vault, &token, "note.md");
|
|
assert_eq!(status, 200);
|
|
assert_eq!(body, "hi!", "the room's text reached the vault");
|
|
}
|
|
|
|
#[test]
|
|
fn the_room_and_peer_caps_are_enforced() {
|
|
let mut windows = windows();
|
|
windows.max_rooms = 2;
|
|
windows.max_peers = 2;
|
|
let server = server("live-caps", windows);
|
|
let (vault, token, _) = server.owned_vault("live-caps");
|
|
|
|
// Two rooms fit.
|
|
assert_eq!(join(&server, &vault, &token, "one.md", "a", "Ana").0, 200);
|
|
assert_eq!(join(&server, &vault, &token, "two.md", "b", "Bo").0, 200);
|
|
// The third room has no room left.
|
|
let (status, _reply) = join(&server, &vault, &token, "three.md", "c", "Cy");
|
|
assert_eq!(status, 503);
|
|
|
|
// The peer cap holds per room.
|
|
assert_eq!(join(&server, &vault, &token, "one.md", "d", "Di").0, 200);
|
|
let (status, _reply) = join(&server, &vault, &token, "one.md", "e", "Ed");
|
|
assert_eq!(status, 409);
|
|
}
|
|
|
|
#[test]
|
|
fn the_event_stream_keeps_alive() {
|
|
let server = server("live-keepalive", windows());
|
|
let (vault, token, _) = server.owned_vault("live-keepalive");
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
|
|
let stream = Stream::open(&server, &vault, &token, "note.md", "a");
|
|
let line = stream.line().expect("a line");
|
|
assert!(line.contains("keepalive"), "the stream comments itself alive: {line}");
|
|
}
|
|
|
|
#[test]
|
|
fn the_audit_log_names_the_live_events_without_the_text() {
|
|
let server = server("live-audit", windows());
|
|
let (vault, token, _) = server.owned_vault("live-audit");
|
|
assert_eq!(server.put(&vault, &token, "note.md", "secret text", 0).0, 201);
|
|
assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
|
|
|
|
let rows = server.audit();
|
|
let joined = rows.iter().find(|row| row["kind"] == json!("live_joined")).expect("a join row");
|
|
assert_eq!(joined["detail"], json!("note.md"));
|
|
assert!(!joined.to_string().contains("secret text"), "the note text is not logged");
|
|
}
|