Marcos Paulo
d9359ba93d
- engine: live.join, live.edit, live.poll, live.peers, live.selection, live.info and live.leave over the live session client (at most 8 rooms, no token in any answer, dropping a session with open rooms returns within 2 s); server.file_meta names the last editor of a file - engine: sync.pending lists the local changes the server does not have yet (new, edited, deleted; at most 500, local comparison only) - ui-egui: a "Waiting to sync" queue sheet opened from the sync sheet, and the "You are offline" banner on the compact Notes page - tests: two sessions converge through the real server using live.* commands; stub-server, queue and banner tests Verified with cargo xtask ci (fmt, clippy -D warnings, workspace tests, layers, tree, i18n). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
389 lines
16 KiB
Rust
389 lines
16 KiB
Rust
//! The "Waiting to sync" queue driven the way a user does: real frames at
|
|
//! 390 x 844 over a vault bound to a stub server (designs `03-sync-sheet`,
|
|
//! `07-offline`).
|
|
|
|
use egui::{Context, Event, Key, PointerButton, Pos2, RawInput};
|
|
use notescraft_engine::{Session, SyncPhase, Timings, bind_folder};
|
|
use notescraft_ui_egui::app::NotesApp;
|
|
use notescraft_ui_egui::panels::sync_queue::Queue;
|
|
use notescraft_ui_egui::shell::compact::CompactPage;
|
|
use std::io::{BufRead, BufReader, Read, Write};
|
|
use std::net::{TcpListener, TcpStream};
|
|
use std::path::Path;
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
|
use std::time::{Duration, Instant};
|
|
|
|
static NEXT: AtomicU64 = AtomicU64::new(0);
|
|
/// The window of a phone.
|
|
const PHONE: egui::Vec2 = egui::vec2(390.0, 844.0);
|
|
/// The token of the bound vault; it must never reach the screen.
|
|
const TOKEN: &str = "secret-token";
|
|
/// Longest the tests wait for the worker.
|
|
const WAIT: Duration = Duration::from_secs(5);
|
|
|
|
/// A stub server: the device list, an empty change log and one applied upload.
|
|
struct Stub {
|
|
base: String,
|
|
stop: Arc<AtomicBool>,
|
|
polls: Arc<AtomicU64>,
|
|
}
|
|
|
|
impl Stub {
|
|
fn spawn() -> Self {
|
|
let listener = TcpListener::bind("127.0.0.1:0").expect("bind the stub");
|
|
let port = listener.local_addr().expect("addr").port();
|
|
let record = r#"{"path":"a.md","hash":"x","size":1,"mtime_ms":0,"seq":1,"deleted":false}"#;
|
|
let empty = r#"{"head":0,"records":[],"more":false}"#;
|
|
let answer = |body: &str| format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len());
|
|
let stop = Arc::new(AtomicBool::new(false));
|
|
let polls = Arc::new(AtomicU64::new(0));
|
|
let (thread_stop, thread_polls) = (Arc::clone(&stop), Arc::clone(&polls));
|
|
// The listener is nonblocking so the stop flag works: the accept loop polls
|
|
// it, and after an accept the stream goes back to blocking reads.
|
|
listener.set_nonblocking(true).expect("nonblocking listener");
|
|
std::thread::spawn(move || {
|
|
while !thread_stop.load(Ordering::Relaxed) {
|
|
let accepted = match listener.accept() {
|
|
Ok((stream, _)) => Some(stream),
|
|
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => None,
|
|
Err(_) => break,
|
|
};
|
|
let Some(stream) = accepted else {
|
|
std::thread::sleep(Duration::from_millis(2));
|
|
continue;
|
|
};
|
|
stream.set_nonblocking(false).expect("the stream blocks its reads again");
|
|
let (mut reader, mut writer) = (BufReader::new(stream.try_clone().expect("clone")), stream);
|
|
while let Some((method, path, length)) = read_head(&mut reader) {
|
|
if length > 0 {
|
|
let mut body = vec![0_u8; length];
|
|
let _ = reader.read_exact(&mut body);
|
|
}
|
|
if method == "PUT" {
|
|
let created = format!("HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{record}", record.len());
|
|
let _ = writer.write_all(created.as_bytes());
|
|
} else if path.contains("/changes") {
|
|
thread_polls.fetch_add(1, Ordering::Relaxed);
|
|
let _ = writer.write_all(answer(empty).as_bytes());
|
|
} else {
|
|
let _ = writer.write_all("HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n".as_bytes());
|
|
}
|
|
let _ = writer.flush();
|
|
}
|
|
}
|
|
});
|
|
Self { base: format!("http://127.0.0.1:{port}"), stop, polls }
|
|
}
|
|
|
|
/// How often the change log was polled.
|
|
fn poll_count(&self) -> u64 {
|
|
self.polls.load(Ordering::Relaxed)
|
|
}
|
|
}
|
|
|
|
impl Drop for Stub {
|
|
fn drop(&mut self) {
|
|
self.stop.store(true, Ordering::Relaxed);
|
|
}
|
|
}
|
|
|
|
/// The request line and the content length of one request.
|
|
fn read_head(reader: &mut BufReader<TcpStream>) -> Option<(String, String, usize)> {
|
|
let mut line = String::new();
|
|
reader.read_line(&mut line).ok()?;
|
|
let mut parts = line.split_whitespace();
|
|
let method = parts.next()?.to_owned();
|
|
let path = parts.next()?.to_owned();
|
|
let mut length = 0_usize;
|
|
loop {
|
|
let mut header = String::new();
|
|
if reader.read_line(&mut header).ok()? == 0 {
|
|
break;
|
|
}
|
|
if header.trim_end().is_empty() {
|
|
break;
|
|
}
|
|
if let Some((name, value)) = header.trim_end().split_once(':')
|
|
&& name.trim().eq_ignore_ascii_case("content-length")
|
|
{
|
|
length = value.trim().parse().unwrap_or(0);
|
|
}
|
|
}
|
|
Some((method, path, length))
|
|
}
|
|
|
|
struct Driver {
|
|
app: NotesApp,
|
|
ctx: Context,
|
|
time: f64,
|
|
}
|
|
|
|
/// An app over a fresh vault bound to `stub` at phone size, animations off.
|
|
fn driven(stub_base: &str, files: &[(&str, &str)]) -> Driver {
|
|
let root = std::env::temp_dir().join(format!("notescraft-queue-{}-{}", std::process::id(), NEXT.fetch_add(1, Ordering::Relaxed)));
|
|
let _ = std::fs::remove_dir_all(&root);
|
|
std::fs::create_dir_all(&root).expect("root");
|
|
seed(&root, files);
|
|
bind_folder(&root, stub_base, "v-1", TOKEN, "d-1", Some("u-1")).expect("bind the vault");
|
|
let config = root.parent().unwrap_or(&root).join("notescraft-queue-config");
|
|
let mut session = Session::open(&root).expect("session");
|
|
session.use_settings_dir(config);
|
|
let mut app = NotesApp::new(session, notescraft_themes::builtin("catppuccin-mocha").expect("theme"));
|
|
app.state.animations = false;
|
|
Driver { app, ctx: Context::default(), time: 1.0 }
|
|
}
|
|
|
|
/// Writes `(path, content)` into `root`, creating the folders.
|
|
fn seed(root: &Path, files: &[(&str, &str)]) {
|
|
for (path, text) in files {
|
|
if let Some((folder, _)) = path.rsplit_once('/') {
|
|
std::fs::create_dir_all(root.join(folder)).expect("folder");
|
|
}
|
|
std::fs::write(root.join(path), text).expect("seed");
|
|
}
|
|
}
|
|
|
|
impl Driver {
|
|
fn frame(&mut self, events: Vec<Event>) -> Vec<Painted> {
|
|
self.time += 0.05;
|
|
let input = RawInput { screen_rect: Some(egui::Rect::from_min_size(Pos2::ZERO, PHONE)), time: Some(self.time), events, ..Default::default() };
|
|
let app = &mut self.app;
|
|
let mut out = self.ctx.run_ui(input, |ui| app.ui(ui.ctx()));
|
|
out.textures_delta.clear();
|
|
out.shapes
|
|
.iter()
|
|
.filter_map(|shape| match &shape.shape {
|
|
egui::Shape::Text(text) => Some(Painted { text: text.galley.text().to_owned(), pos: text.pos, galley: text.galley.clone() }),
|
|
_ => None,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn settle(&mut self) -> Vec<Painted> {
|
|
let mut last = Vec::new();
|
|
for _ in 0..24 {
|
|
last = self.frame(Vec::new());
|
|
}
|
|
last
|
|
}
|
|
|
|
/// The centre of the first painted `wanted` label, for tapping it.
|
|
fn center(&mut self, wanted: &str) -> Pos2 {
|
|
let painted = self.find(wanted);
|
|
painted.pos + painted.galley.size() / 2.0
|
|
}
|
|
|
|
/// The first painted text that is exactly `wanted`, or panics.
|
|
fn find(&mut self, wanted: &str) -> Painted {
|
|
let texts = self.settle();
|
|
let all: Vec<String> = texts.iter().map(|painted| painted.text.clone()).collect();
|
|
texts.into_iter().find(|painted| painted.text == wanted).unwrap_or_else(move || panic!("no {wanted:?} among {all:?}"))
|
|
}
|
|
|
|
/// A click through press and release on `at`.
|
|
fn click(&mut self, at: Pos2) {
|
|
self.frame(vec![Event::PointerMoved(at)]);
|
|
self.frame(vec![Event::PointerButton { pos: at, button: PointerButton::Primary, pressed: true, modifiers: egui::Modifiers::NONE }]);
|
|
self.frame(vec![Event::PointerButton { pos: at, button: PointerButton::Primary, pressed: false, modifiers: egui::Modifiers::NONE }]);
|
|
self.settle();
|
|
}
|
|
|
|
/// The escape key.
|
|
fn escape(&mut self) {
|
|
self.frame(vec![Event::Key { key: Key::Escape, physical_key: None, pressed: true, repeat: false, modifiers: egui::Modifiers::NONE }]);
|
|
self.settle();
|
|
}
|
|
|
|
/// The worker runs fast enough for the tests.
|
|
fn fast_worker(&mut self) {
|
|
if let Some(worker) = self.app.session.sync_worker() {
|
|
worker.set_timings(Timings {
|
|
debounce: Duration::from_millis(10),
|
|
poll: Duration::from_mins(15),
|
|
backoff: Duration::from_millis(20),
|
|
backoff_max: Duration::from_millis(60),
|
|
});
|
|
}
|
|
}
|
|
|
|
/// Waits, bounded, until the worker's phase and pending count match.
|
|
fn worker_reaches(&mut self, phase: SyncPhase, pending: usize) {
|
|
let deadline = Instant::now() + WAIT;
|
|
while Instant::now() < deadline {
|
|
if self.app.session.sync_worker().is_some_and(|worker| worker.status().phase == phase && worker.status().pending == pending) {
|
|
return;
|
|
}
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
}
|
|
panic!("the worker never reached {phase:?} with {pending} pending: {:?}", self.app.session.sync_worker().map(notescraft_engine::SyncWorker::status));
|
|
}
|
|
}
|
|
|
|
/// A painted text: what it says, where it starts and its galley.
|
|
struct Painted {
|
|
text: String,
|
|
pos: Pos2,
|
|
galley: Arc<egui::Galley>,
|
|
}
|
|
|
|
/// Waits until the worker phase is the wanted one, no count asked.
|
|
fn phase_of(d: &mut Driver) -> Option<SyncPhase> {
|
|
d.app.session.sync_worker().map(|worker| worker.status().phase)
|
|
}
|
|
|
|
/// Waits, bounded, for an idle worker that has finished a full cycle: a fresh
|
|
/// worker's status is Idle at once, so the answer's report is the proof.
|
|
fn wait_idle(d: &mut Driver) {
|
|
let deadline = Instant::now() + WAIT;
|
|
while Instant::now() < deadline {
|
|
if let Some(worker) = d.app.session.sync_worker() {
|
|
let status = worker.status();
|
|
if status.last_report.is_some() && status.phase == SyncPhase::Idle {
|
|
break;
|
|
}
|
|
}
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
}
|
|
}
|
|
|
|
/// Opens the queue over a vault whose worker is idle, without the sheet.
|
|
fn idle_queue(d: &mut Driver) {
|
|
wait_idle(d);
|
|
d.app.state.compact_page = CompactPage::Notes;
|
|
d.app.sync_queue = Some(Queue::open(&d.app.session));
|
|
d.settle();
|
|
}
|
|
/// The pill shows "Offline" and the sheet has the waiting row; the queue it
|
|
/// opens lists the file, its folder and the kinds.
|
|
#[test]
|
|
fn the_waiting_row_opens_the_queue_with_the_kinds() {
|
|
let stub = Stub::spawn();
|
|
let mut d = driven(&stub.base, &[("a.md", "A"), ("img/f.png", "bytes")]);
|
|
drop(stub); // the phone is offline; the local changes wait
|
|
d.settle();
|
|
d.fast_worker();
|
|
d.worker_reaches(SyncPhase::Offline, 2);
|
|
// The status bar (and its pill) shows when a note is open.
|
|
d.app.open_note("a.md");
|
|
d.settle();
|
|
let offline_pill = d.center("Offline");
|
|
d.click(offline_pill);
|
|
assert!(d.app.sync_sheet.is_some(), "the pill opened the sheet");
|
|
let _ = d.find("2 changes waiting");
|
|
|
|
let row = d.center("2 changes waiting");
|
|
d.click(row);
|
|
assert!(d.app.sync_sheet.is_none(), "the row closed the sheet");
|
|
assert!(d.app.sync_queue.is_some(), "the row opened the queue");
|
|
let texts = d.settle();
|
|
for wanted in ["Waiting to sync", "2 changes waiting", "a.md", "1 B", "Sync now"] {
|
|
assert!(
|
|
texts.iter().any(|painted| painted.text == wanted),
|
|
"no {wanted:?} among {:?}",
|
|
texts.iter().map(|painted| painted.text.as_str()).collect::<Vec<_>>()
|
|
);
|
|
}
|
|
assert!(texts.iter().any(|painted| painted.text == "img"), "the folder of img/f.png shows muted");
|
|
assert!(
|
|
texts.iter().filter(|painted| painted.text == "New").count() == 2,
|
|
"two New chips: {:?}",
|
|
texts.iter().map(|painted| painted.text.as_str()).collect::<Vec<_>>()
|
|
);
|
|
|
|
d.escape();
|
|
assert!(d.app.sync_queue.is_none(), "Escape closed the queue");
|
|
}
|
|
|
|
/// A recorded base and a deleted file make the red Deleted chip.
|
|
#[test]
|
|
fn the_queue_lists_a_deleted_file() {
|
|
let stub = Stub::spawn();
|
|
let mut d = driven(&stub.base, &[("a.md", "A"), ("img/f.png", "bytes")]);
|
|
drop(stub);
|
|
d.settle();
|
|
d.fast_worker();
|
|
d.worker_reaches(SyncPhase::Offline, 2);
|
|
// the worker is offline, so nothing touches the bases again.
|
|
let same = "a".repeat(64);
|
|
let bases = format!(r#"{{"last_seq":2,"bases":{{"deleted.md":{{"hash":"{same}","seq":2}}}}}}"#);
|
|
d.app.session.vault().write_private(".notescraft/sync/state.json", bases.as_bytes()).expect("record the bases");
|
|
d.app.state.compact_page = CompactPage::Notes;
|
|
d.app.sync_queue = Some(Queue::open(&d.app.session));
|
|
let texts = d.settle();
|
|
let _ = d.find("Deleted");
|
|
assert!(
|
|
texts.iter().any(|painted| painted.text == "New"),
|
|
"the unread files are New rows: {:?}",
|
|
texts.iter().map(|painted| painted.text.as_str()).collect::<Vec<_>>()
|
|
);
|
|
let _ = d.find("deleted.md");
|
|
}
|
|
|
|
/// Everything synced: the queue says so, with no rows and no chips.
|
|
#[test]
|
|
fn the_empty_state_shows_when_the_vault_is_synced() {
|
|
let stub = Stub::spawn();
|
|
let mut d = driven(&stub.base, &[("a.md", "A")]);
|
|
d.settle();
|
|
d.fast_worker();
|
|
idle_queue(&mut d);
|
|
let texts = d.settle();
|
|
let _ = d.find("Everything is up to date.");
|
|
assert!(!texts.iter().any(|painted| painted.text == "New"), "no chips in the empty state");
|
|
assert!(d.app.sync_queue.is_some(), "the queue stays open");
|
|
d.app.sync_queue = None;
|
|
}
|
|
|
|
/// Sync now in the queue reaches the worker: the server sees a second put.
|
|
#[test]
|
|
fn sync_now_in_the_queue_wakes_the_worker() {
|
|
let stub = Stub::spawn();
|
|
let mut d = driven(&stub.base, &[("a.md", "A")]);
|
|
d.settle();
|
|
d.fast_worker();
|
|
idle_queue(&mut d);
|
|
let before = stub.poll_count();
|
|
let button = d.center("Sync now");
|
|
d.click(button);
|
|
let deadline = Instant::now() + WAIT;
|
|
while Instant::now() < deadline && stub.poll_count() == before {
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
}
|
|
assert!(stub.poll_count() > before, "Sync now woke the worker: {} -> {}", before, stub.poll_count());
|
|
assert!(d.app.sync_queue.is_some(), "Sync now keeps the queue open");
|
|
d.app.sync_queue = None;
|
|
}
|
|
|
|
/// The offline banner sits on the compact Notes page while the phase is
|
|
/// Offline, and never over a note page.
|
|
#[test]
|
|
fn the_offline_banner_shows_only_while_offline() {
|
|
let stub = Stub::spawn();
|
|
let mut d = driven(&stub.base, &[("a.md", "A")]);
|
|
drop(stub);
|
|
d.settle();
|
|
d.fast_worker();
|
|
d.worker_reaches(SyncPhase::Offline, 1);
|
|
d.app.state.compact_page = CompactPage::Notes;
|
|
let texts = d.settle();
|
|
let _ = d.find("You are offline");
|
|
assert!(texts.iter().any(|painted| painted.text.contains("Edits are saved on this phone")), "the line under the headline");
|
|
|
|
// On the Note page (a note open) the banner has no room and no job.
|
|
d.app.state.compact_page = CompactPage::Note;
|
|
let note_page = d.settle();
|
|
assert!(!note_page.iter().any(|painted| painted.text == "You are offline"), "no banner over a note");
|
|
|
|
// A healthy vault: no banner on the Notes page either (the phase never left Idle).
|
|
let stub = Stub::spawn();
|
|
let mut d = driven(&stub.base, &[("a.md", "A")]);
|
|
d.settle();
|
|
d.fast_worker();
|
|
wait_idle(&mut d);
|
|
d.app.state.compact_page = CompactPage::Notes;
|
|
let after = d.settle();
|
|
assert_eq!(phase_of(&mut d), Some(SyncPhase::Idle), "the worker is idle");
|
|
assert!(!after.iter().any(|painted| painted.text == "You are offline"), "the banner never shows while idle");
|
|
}
|