//! The engine's live session against the real sync server: two peers type in the //! same room, their texts meet and the room reaches the vault through the flush. //! //! The time is never waited for: the windows of the room are long, the flush is //! called by hand and the clients are polled until they are quiet. mod common; use std::time::{Duration, Instant}; use common::Server; use notescraft_engine::{LiveConfig, LiveSession}; use notescraft_sync_server::{AppState, Store, Timings, flush, now_ms}; use serde_json::json; /// How many rounds both peers type in. const ROUNDS: usize = 5; /// How long a test waits for the clients to settle. const DEADLINE: Duration = Duration::from_secs(10); /// How many polls without an event mean that the clients are quiet. const QUIET_POLLS: usize = 40; /// The note both peers type in. const NOTE: &str = "notes/note.md"; /// The windows of the room: a short keepalive and long debounces, so the only /// write of the test is the flush at the end. fn windows() -> Timings { Timings { keepalive_ms: 200, idle_ms: 30_000, presence_ttl_ms: 60_000, flush_debounce_ms: 60_000, sweep_ms: 60_000, max_rooms: 8, max_peers: 64 } } /// A server over a fresh folder with the windows of the test. fn server(name: &str) -> 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) } /// The connection details of one peer of the room. fn config(server: &Server, vault: &str, token: &str, client: &str, name: &str) -> LiveConfig { LiveConfig { server: server.base.clone(), vault_id: vault.to_owned(), token: token.to_owned(), client_id: client.to_owned(), name: name.to_owned() } } /// A generator of positions that repeats itself, so a failing run repeats too. struct Lcg(u64); impl Lcg { /// A generator with the given seed. fn new(seed: u64) -> Self { Self(seed) } /// The next number of the sequence. fn next(&mut self) -> u64 { self.0 = self.0.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1_442_695_040_888_963_407); self.0 >> 33 } /// A position inside a text of `len` characters. fn position(&mut self, len: usize) -> usize { let span = u64::try_from(len).unwrap_or(u64::MAX).max(1); usize::try_from(self.next() % span).unwrap_or(0) } } /// The places a marker may go: the start of a line. A marker put there keeps /// its shape, whatever the other peers type later. fn line_starts(text: &str) -> Vec { let mut starts = vec![0]; for (index, character) in text.chars().enumerate() { if character == '\n' { starts.push(index + 1); } } starts } /// The text with `marker` put in at the given position. fn insert_at(text: &str, at: usize, marker: &str) -> String { let mut out = String::with_capacity(text.len() + marker.len()); for (index, character) in text.chars().enumerate() { if index == at { out.push_str(marker); } out.push(character); } if at >= text.chars().count() { out.push_str(marker); } out } /// Polls both clients until they are quiet and hold the same text. fn settle(a: &mut LiveSession, b: &mut LiveSession) -> bool { let deadline = Instant::now() + DEADLINE; let mut quiet = 0; while Instant::now() < deadline { let events = a.poll().len() + b.poll().len(); if events == 0 { quiet += 1; } else { quiet = 0; } if quiet >= QUIET_POLLS && a.text() == b.text() && a.rev() == b.rev() { return true; } std::thread::sleep(Duration::from_millis(10)); } false } /// Polls the client until it sees the peer with that id. fn sees_peer(session: &mut LiveSession, client: &str) -> bool { let deadline = Instant::now() + DEADLINE; while Instant::now() < deadline { session.poll(); if session.peers().iter().any(|peer| peer.client == client) { return true; } std::thread::sleep(Duration::from_millis(10)); } false } /// The text and the revision of the room, read by joining as a third peer. fn room_snapshot(server: &Server, vault: &str, token: &str, path: &str) -> (String, u64) { let (status, reply) = server.json("POST", &format!("/v1/vaults/{vault}/live/{path}/join"), Some(token), &[], Some(&json!({"client": "observer", "name": "Observer"}))); assert_eq!(status, 200, "{reply}"); let text = reply["text"].as_str().expect("the text of the room").to_owned(); let rev = reply["rev"].as_u64().expect("the revision of the room"); (text, rev) } #[test] fn two_peers_type_in_the_same_room() { let server = server("live-e2e"); let (vault, token, _user) = server.owned_vault("live-e2e"); let (status, _) = server.put(&vault, &token, NOTE, "one\ntwo\nthree\n", 0); assert_eq!(status, 201, "the note is in the vault"); let mut lcg = Lcg::new(0x5EED_1234); let (mut a, text_a) = LiveSession::join(config(&server, &vault, &token, "client-a", "Ana"), NOTE).expect("join a"); assert_eq!(text_a, "one\ntwo\nthree\n", "the room hands its text to the first client"); let (mut b, text_b) = LiveSession::join(config(&server, &vault, &token, "client-b", "Ben"), NOTE).expect("join b"); assert_eq!(text_b, "one\ntwo\nthree\n", "the room hands its text to the second client"); for round in 0..ROUNDS { let marker_a = format!("A{round} "); let marker_b = format!("B{round} "); let starts_a = line_starts(a.text()); let starts_b = line_starts(b.text()); let at_a = starts_a.get(lcg.position(starts_a.len())).copied().unwrap_or(0); let at_b = starts_b.get(lcg.position(starts_b.len())).copied().unwrap_or(0); let edited_a = insert_at(a.text(), at_a, &marker_a); let edited_b = insert_at(b.text(), at_b, &marker_b); a.local_edit(&edited_a).expect("the edit of the first client"); b.local_edit(&edited_b).expect("the edit of the second client"); assert!(settle(&mut a, &mut b), "the clients did not settle in round {round}"); } assert!(sees_peer(&mut a, "client-b"), "the first client sees the second one"); assert!(sees_peer(&mut b, "client-a"), "the second client sees the first one"); let (room_text, room_rev) = room_snapshot(&server, &vault, &token, NOTE); assert_eq!(a.text(), room_text, "the first client holds the text of the room"); assert_eq!(b.text(), room_text, "the second client holds the text of the room"); assert_eq!(a.rev(), room_rev, "the first client is at the revision of the room"); assert_eq!(b.rev(), room_rev, "the second client is at the revision of the room"); for round in 0..ROUNDS { let marker_a = format!("A{round} "); let marker_b = format!("B{round} "); assert!(room_text.contains(&marker_a), "the marker {marker_a} was lost"); assert!(room_text.contains(&marker_b), "the marker {marker_b} was lost"); } a.leave(); b.leave(); let report = flush(&server.state, &vault, NOTE, now_ms()).expect("the room is written"); assert!(report.flushed, "the room was written to the vault"); assert!(!report.conflict, "the room did not conflict with the vault"); let (status, body) = server.send("GET", &format!("/v1/vaults/{vault}/files/{NOTE}"), Some(&token), &[], None); assert_eq!(status, 200); assert_eq!(String::from_utf8_lossy(&body), room_text, "the file holds the text of the room"); }