diff --git a/crates/notebook/tests/live_batch.rs b/crates/notebook/tests/live_batch.rs index 2945cfe0a6cab863c6beb05943bbd297c8315f23..e03bc5b4596da76878314e9f8ca3470418ecb9e1 100644 --- a/crates/notebook/tests/live_batch.rs +++ b/crates/notebook/tests/live_batch.rs @@ -12,24 +12,70 @@ mod live; use live::*; #[test] -fn a_lost_batch_receipt_is_confirmed_without_repeating_its_edit() { - use notebook::{PendingEdit, Remote}; +fn edit_receipts_survive_guest_and_host_process_death() { + use notebook::live::share::{self, Guest, Host}; + use notebook::session::Notebook; + use notebook::{EditStatus, PendingEdit, Remote}; use onestore::{CommitError, CommitState, ExGuid, Stamp, Transaction}; - use std::{collections::BTreeMap, io}; + use std::{ + collections::BTreeMap, + io, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + time::{Duration, Instant}, + }; - struct Lost(HostedRemote, bool); - impl Remote for Lost { + const TEST: &str = "edit_receipts_survive_guest_and_host_process_death"; + const ROLE: &str = "SNOWBOUND_LIVE_RECOVERY_ROLE"; + const ROOT: &str = "SNOWBOUND_LIVE_RECOVERY_ROOT"; + const URL: &str = "SNOWBOUND_LIVE_RECOVERY_URL"; + const BOUNDARY: &str = "SNOWBOUND_LIVE_RECOVERY_BOUNDARY"; + const INSERTION: &str = " CRASH_RECOVERY"; + + fn body(image: &[u8]) -> (ExGuid, ExGuid, String) { + let store = onestore::Store::parse(image).unwrap(); + assert!(store.checksum_mismatches.is_empty()); + let index = onestore::RevisionIndex::parse(&store).unwrap(); + index.validate_current().unwrap(); + let document = onestore::document::Document::parse(&index).unwrap(); + let space = document.pages().unwrap()[0].0; + let (text, content) = document.spaces[&space] + .active() + .unwrap() + .nodes + .iter() + .find_map(|(id, node)| match &node.kind { + onestore::document::Kind::RichText { text, .. } => Some((*id, text.clone())), + _ => None, + }) + .unwrap(); + (space, text, content) + } + + fn stopped_at_boundary(root: &Path) -> ! { + std::fs::write(root.join("guest-ready"), b"").unwrap(); + loop { + std::thread::park(); + } + } + + struct Interrupted<'a> { + remote: HostedRemote, + boundary: &'a str, + root: &'a Path, + } + impl Remote for Interrupted<'_> { fn read(&mut self) -> io::Result> { - self.0.read() + self.remote.read() } fn stamp(&mut self) -> io::Result { - self.0.stamp() + self.remote.stamp() } fn publish(&mut self, transaction: &Transaction) -> Result<(), CommitError> { - self.0.publish(transaction) + self.remote.publish(transaction) } fn confirm(&mut self, stamp: &Stamp) -> Result<(), CommitError> { - self.0.confirm(stamp) + self.remote.confirm(stamp) } fn accepts_edits(&self) -> bool { true @@ -40,8 +86,11 @@ fn a_lost_batch_receipt_is_confirmed_without_repeating_its_edit() { edits: &[PendingEdit], revisions: &BTreeMap, ) -> Result<(), CommitError> { - self.0.publish_edits(transaction, edits, revisions)?; - if std::mem::take(&mut self.1) { + if self.boundary == "unsent" { + stopped_at_boundary(self.root); + } + self.remote.publish_edits(transaction, edits, revisions)?; + if self.boundary == "lost" { return Err(CommitError { state: CommitState::Unknown, error: io::ErrorKind::BrokenPipe.into(), @@ -50,53 +99,362 @@ fn a_lost_batch_receipt_is_confirmed_without_repeating_its_edit() { Ok(()) } } - let directory = tempfile::tempdir().unwrap(); - let folder = notebook(directory.path()); - let url = relay(Default::default()); - let sharing = Sharing::new("").unwrap(); - let host = host(&folder, &directory.path().join("host"), &sharing, &url); - let (guest, _) = guest("Alice", &code(&host), &url, &directory.path().join("alice")); - let file = folder.join("Garden.one"); - let image = std::fs::read(&file).unwrap(); - let (space, text, _) = server::text(&image); - let replica = Replica::open_or_create(directory.path().join("replica.sqlite"), None, || { - Ok(image.clone()) - }) - .unwrap(); - let id = replica - .apply( - "Alice", - Edit { - at: 134_000_000_000_000_000, - ops: vec![Op::Page { - space, - op: PageOp::Text { - text, - range: 13..13, - with: "X".into(), + + if let Ok(role) = std::env::var(ROLE) { + let root = PathBuf::from(std::env::var_os(ROOT).unwrap()); + let url = std::env::var(URL).unwrap(); + let boundary = std::env::var(BOUNDARY).unwrap(); + if role == "host" { + let saved = root.join("sharing.json"); + let sharing = if saved.exists() { + serde_json::from_slice(&std::fs::read(&saved).unwrap()).unwrap() + } else { + Sharing::new("").unwrap() + }; + let host = Host::start( + Notebook::open(root.join("Garden"), root.join("host")) + .unwrap() + .into_storage(), + hello("Ada"), + sharing, + "Garden", + None, + Some(&url), + || {}, + move |sharing| std::fs::write(&saved, serde_json::to_vec(sharing)?), + ) + .unwrap(); + std::fs::write(root.join("code"), code(&host)).unwrap(); + std::fs::write(root.join("host-ready"), b"").unwrap(); + loop { + std::thread::park(); + } + } + let credentials = root.join("guest.json"); + let (share, secret): ([u8; 16], [u8; 16]) = if role == "guest" { + let code = std::fs::read_to_string(root.join("code")).unwrap(); + let welcome = share::join(hello("Alice"), &code, "", None, Some(&url)).unwrap(); + let issued = (welcome.share, welcome.secret); + std::fs::write(&credentials, serde_json::to_vec(&issued).unwrap()).unwrap(); + issued + } else { + serde_json::from_slice(&std::fs::read(credentials).unwrap()).unwrap() + }; + let guest = Guest::start(hello("Alice"), share, secret, None, Some(&url), || {}).unwrap(); + until("the restarted guest never met its host", || { + guest.host().is_some() + }); + let database = root.join("replica.sqlite"); + let replica = if role == "guest" { + let image = std::fs::read(root.join("Garden/Garden.one")).unwrap(); + let (space, text, content) = body(&image); + let replica = Replica::create(&database, &image).unwrap(); + let end = u32::try_from(content.encode_utf16().count()).unwrap(); + let id = replica + .apply( + "Alice", + Edit { + at: 134_000_000_000_000_000, + ops: vec![Op::Page { + space, + op: PageOp::Text { + text, + range: end..end, + with: INSERTION.into(), + }, + }], }, - }], - }, - ) - .unwrap(); - let mut remote = Lost(HostedRemote::new(&guest, "Garden.one"), true); - assert!(matches!( - replica.sync_once(&mut remote), - Err(notebook::Error::Remote(CommitError { - state: CommitState::Unknown, - .. - })) - )); - replica.sync_once(&mut remote).unwrap(); - assert!(matches!( - replica.status(id).unwrap(), - Some(notebook::EditStatus::Published { .. }) - )); - assert_eq!( - server::text(&std::fs::read(file).unwrap()).2, - "Original textX" - ); - assert!(replica.pending().unwrap().is_empty()); + ) + .unwrap(); + std::fs::write(root.join("edit-id"), id.to_string()).unwrap(); + replica + } else { + Replica::open(&database).unwrap() + }; + let id = std::fs::read_to_string(root.join("edit-id")) + .unwrap() + .parse() + .unwrap(); + if role == "guest" { + if boundary == "before" { + stopped_at_boundary(&root); + } + let mut remote = Interrupted { + remote: HostedRemote::new(&guest, "Garden.one"), + boundary: &boundary, + root: &root, + }; + let result = replica.sync_once(&mut remote); + if boundary == "lost" { + assert!(matches!( + result, + Err(notebook::Error::Remote(CommitError { + state: CommitState::Unknown, + .. + })) + )); + assert!(matches!( + replica.status(id).unwrap(), + Some(EditStatus::AwaitingConfirmation { .. }) + )); + } else { + result.unwrap(); + assert!(matches!( + replica.status(id).unwrap(), + Some(EditStatus::Published { .. }) + )); + } + stopped_at_boundary(&root); + } + let mut remote = HostedRemote::new(&guest, "Garden.one"); + if boundary == "unsent" { + let attempted = replica.status(id).unwrap(); + assert!(matches!( + attempted, + Some(EditStatus::AwaitingConfirmation { .. }) + )); + let pending = replica.pending().unwrap(); + let snapshot = replica.snapshot().unwrap(); + let original = std::fs::read(root.join("Garden/Garden.one")).unwrap(); + for _ in 0..3 { + replica.sync_once(&mut remote).unwrap(); + assert_eq!(replica.status(id).unwrap(), attempted); + assert_eq!(replica.pending().unwrap(), pending); + assert_eq!(replica.snapshot().unwrap(), snapshot); + assert_eq!( + std::fs::read(root.join("Garden/Garden.one")).unwrap(), + original + ); + } + let archive = root.join("before-review.sqlite"); + replica + .release(id, &archive, notebook::Resolution::Mine) + .unwrap(); + let archive = notebook::Recovery::open(archive).unwrap(); + assert_eq!(archive.status(id).unwrap(), attempted); + assert_eq!(archive.pending().unwrap(), pending); + assert_eq!(archive.snapshot().unwrap(), snapshot); + } + for _ in 0..5 { + replica.sync_once(&mut remote).unwrap(); + if matches!( + replica.status(id).unwrap(), + Some(EditStatus::Published { .. }) + ) { + break; + } + } + assert!(matches!( + replica.status(id).unwrap(), + Some(EditStatus::Published { .. }) + )); + assert!(replica.pending().unwrap().is_empty()); + let settled = std::fs::read(root.join("Garden/Garden.one")).unwrap(); + for _ in 0..3 { + replica.sync_once(&mut remote).unwrap(); + assert_eq!( + std::fs::read(root.join("Garden/Garden.one")).unwrap(), + settled + ); + } + assert_eq!(replica.snapshot().unwrap(), settled); + drop(replica); + std::process::exit(0); + } + + struct Process { + child: Child, + log: PathBuf, + } + impl Process { + fn start(root: &Path, url: &str, role: &str, boundary: &str) -> Self { + let log = root.join(format!("{role}.log")); + let output = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&log) + .unwrap(); + let child = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", TEST, "--nocapture"]) + .env(ROLE, role) + .env(ROOT, root) + .env(URL, url) + .env(BOUNDARY, boundary) + .stdout(output.try_clone().unwrap()) + .stderr(Stdio::from(output)) + .spawn() + .unwrap(); + Self { child, log } + } + + fn ready(&mut self, marker: &Path) { + let deadline = Instant::now() + Duration::from_secs(30); + while !marker.exists() { + assert!( + self.child.try_wait().unwrap().is_none() && Instant::now() < deadline, + "{}: {}", + marker.display(), + std::fs::read_to_string(&self.log).unwrap() + ); + std::thread::sleep(Duration::from_millis(20)); + } + } + + fn kill(&mut self) { + self.child.kill().unwrap(); + assert!(!self.child.wait().unwrap().success()); + } + } + impl Drop for Process { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + } + + for boundary in ["before", "unsent", "lost", "confirmed"] { + let directory = tempfile::tempdir().unwrap(); + let root = directory.path(); + let folder = notebook(root); + let file = folder.join("Garden.one"); + let original = + include_bytes!("../../../corpus/conflict-page/native/conflict/notebook/synthetic.one"); + std::fs::write(&file, original).unwrap(); + let original_conflicts = server::conflicts(original); + assert!(!original_conflicts.is_empty()); + let (space, _, content) = body(original); + let sealed = std::fs::read(folder.join("Sealed.one")).unwrap(); + let url = relay(Default::default()); + let mut host = Process::start(root, &url, "host", boundary); + host.ready(&root.join("host-ready")); + let mut guest = Process::start(root, &url, "guest", boundary); + guest.ready(&root.join("guest-ready")); + guest.kill(); + host.kill(); + + let id: u64 = std::fs::read_to_string(root.join("edit-id")) + .unwrap() + .parse() + .unwrap(); + let database = root.join("replica.sqlite"); + let replica = Replica::open(&database).unwrap(); + assert_eq!( + body(&replica.snapshot().unwrap()).2, + format!("{content}{INSERTION}") + ); + if boundary == "confirmed" { + assert!(matches!( + replica.status(id).unwrap(), + Some(EditStatus::Published { .. }) + )); + assert!(replica.pending().unwrap().is_empty()); + } else if matches!(boundary, "lost" | "unsent") { + assert!(matches!( + replica.status(id).unwrap(), + Some(EditStatus::AwaitingConfirmation { .. }) + )); + } else { + assert_eq!(replica.status(id).unwrap(), Some(EditStatus::Pending)); + } + if boundary != "confirmed" { + assert_eq!(replica.pending().unwrap()[0].id, id); + assert_eq!(replica.pending().unwrap().len(), 1); + } + drop(replica); + let committed = std::fs::read(&file).unwrap(); + if matches!(boundary, "before" | "unsent") { + assert_eq!(committed, original); + } else { + assert_eq!(body(&committed).2, format!("{content}{INSERTION}")); + } + let sharing: Sharing = + serde_json::from_slice(&std::fs::read(root.join("sharing.json")).unwrap()).unwrap(); + let credentials = std::fs::read(root.join("guest.json")).unwrap(); + std::fs::remove_file(root.join("host-ready")).unwrap(); + let mut host = Process::start(root, &url, "host", boundary); + host.ready(&root.join("host-ready")); + let mut resumed = Process::start(root, &url, "resume", boundary); + let deadline = Instant::now() + Duration::from_secs(30); + let status = loop { + if let Some(status) = resumed.child.try_wait().unwrap() { + break status; + } + assert!(Instant::now() < deadline, "resumed guest timed out"); + std::thread::sleep(Duration::from_millis(20)); + }; + if !status.success() { + let log = std::fs::read_to_string(&resumed.log).unwrap(); + host.kill(); + let kept = directory.keep(); + panic!("{}: {log}", kept.display()); + } + host.kill(); + let result = std::fs::read(&file).unwrap(); + assert_eq!(body(&result).2, format!("{content}{INSERTION}")); + if matches!(boundary, "lost" | "confirmed") { + assert_eq!( + &result[1024..], + &committed[1024..], + "settlement must not append another revision" + ); + } + let revisions = |image: &[u8]| { + let store = onestore::Store::parse(image).unwrap(); + onestore::RevisionIndex::parse(&store).unwrap().spaces[&space] + .revisions + .len() + }; + assert_eq!(revisions(&result), revisions(original) + 1); + assert_eq!(server::conflicts(&result), original_conflicts); + assert_eq!(std::fs::read(folder.join("Sealed.one")).unwrap(), sealed); + let restarted: Sharing = + serde_json::from_slice(&std::fs::read(root.join("sharing.json")).unwrap()).unwrap(); + assert_eq!(restarted.share, sharing.share); + assert_eq!(restarted.secret, sharing.secret); + assert_eq!(restarted.members, sharing.members); + assert_eq!(std::fs::read(root.join("guest.json")).unwrap(), credentials); + let replica = Replica::open(&database).unwrap(); + assert!(matches!( + replica.status(id).unwrap(), + Some(EditStatus::Published { .. }) + )); + assert!(replica.pending().unwrap().is_empty()); + assert_eq!(replica.snapshot().unwrap(), result); + drop(replica); + if boundary == "unsent" { + let archive = notebook::Recovery::open(root.join("before-review.sqlite")).unwrap(); + assert!(matches!( + archive.status(id).unwrap(), + Some(EditStatus::AwaitingConfirmation { .. }) + )); + assert_eq!(archive.pending().unwrap()[0].id, id); + assert_eq!(archive.pending().unwrap().len(), 1); + assert_eq!( + body(&archive.snapshot().unwrap()).2, + format!("{content}{INSERTION}") + ); + let database = rusqlite::Connection::open(root.join("before-review.sqlite")).unwrap(); + assert_eq!( + database + .query_row("PRAGMA integrity_check", [], |row| row.get::<_, String>(0)) + .unwrap(), + "ok" + ); + println!( + "unsent: cold uncertainty retained; explicit Mine review archived the original queue before publication" + ); + } + let database = rusqlite::Connection::open(&database).unwrap(); + assert_eq!( + database + .query_row("PRAGMA integrity_check", [], |row| row.get::<_, String>(0)) + .unwrap(), + "ok" + ); + println!( + "{boundary}: original edit {id} published once, both processes restarted, SQLite integrity ok" + ); + } } #[test]