From eda8a790491b2ff2ca460b8550a86b41dc5ec84b Mon Sep 17 00:00:00 2001 From: clover caruso Date: Sun, 11 Oct 2026 12:59:32 -0700 Subject: [PATCH] test: recover Live Share receipts after peer process death Restart real host and guest processes with saved credentials and the original SQLite queue at four publication boundaries. Preserve uncertain unsent edits until explicit reviewed recovery, and verify committed edits settle once without losing native conflict pages. Refs #122 Assisted-by: gpt-6.1-sol --- crates/notebook/tests/live_batch.rs | 472 ++++++++++++++++++++++++---- 1 file changed, 415 insertions(+), 57 deletions(-) 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] -- 2.54.0