authorclover caruso <git@paperclover.net> 2026-10-11 12:59:32-07:00
committerclover caruso <git@paperclover.net> 2026-10-11 13:36:39-07:00
logeda8a790491b2ff2ca460b8550a86b41dc5ec84b
treebeb940da2f584008444b87431a411ca30e48e371
parentd179e57e9d023e275f399908164444480b498def
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

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

1 files changed, 415 insertions(+), 57 deletions(-)

crates/notebook/tests/live_batch.rs+415-57
......@@ -12,24 +12,70 @@ mod live;
1212use live::*;
1313
1414#[test]
15fn a_lost_batch_receipt_is_confirmed_without_repeating_its_edit() {
16 use notebook::{PendingEdit, Remote};
15fn edit_receipts_survive_guest_and_host_process_death() {
16 use notebook::live::share::{self, Guest, Host};
17 use notebook::session::Notebook;
18 use notebook::{EditStatus, PendingEdit, Remote};
1719 use onestore::{CommitError, CommitState, ExGuid, Stamp, Transaction};
18 use std::{collections::BTreeMap, io};
20 use std::{
21 collections::BTreeMap,
22 io,
23 path::{Path, PathBuf},
24 process::{Child, Command, Stdio},
25 time::{Duration, Instant},
26 };
27
28 const TEST: &str = "edit_receipts_survive_guest_and_host_process_death";
29 const ROLE: &str = "SNOWBOUND_LIVE_RECOVERY_ROLE";
30 const ROOT: &str = "SNOWBOUND_LIVE_RECOVERY_ROOT";
31 const URL: &str = "SNOWBOUND_LIVE_RECOVERY_URL";
32 const BOUNDARY: &str = "SNOWBOUND_LIVE_RECOVERY_BOUNDARY";
33 const INSERTION: &str = " CRASH_RECOVERY";
34
35 fn body(image: &[u8]) -> (ExGuid, ExGuid, String) {
36 let store = onestore::Store::parse(image).unwrap();
37 assert!(store.checksum_mismatches.is_empty());
38 let index = onestore::RevisionIndex::parse(&store).unwrap();
39 index.validate_current().unwrap();
40 let document = onestore::document::Document::parse(&index).unwrap();
41 let space = document.pages().unwrap()[0].0;
42 let (text, content) = document.spaces[&space]
43 .active()
44 .unwrap()
45 .nodes
46 .iter()
47 .find_map(|(id, node)| match &node.kind {
48 onestore::document::Kind::RichText { text, .. } => Some((*id, text.clone())),
49 _ => None,
50 })
51 .unwrap();
52 (space, text, content)
53 }
54
55 fn stopped_at_boundary(root: &Path) -> ! {
56 std::fs::write(root.join("guest-ready"), b"").unwrap();
57 loop {
58 std::thread::park();
59 }
60 }
1961
20 struct Lost(HostedRemote, bool);
21 impl Remote for Lost {
62 struct Interrupted<'a> {
63 remote: HostedRemote,
64 boundary: &'a str,
65 root: &'a Path,
66 }
67 impl Remote for Interrupted<'_> {
2268 fn read(&mut self) -> io::Result<Vec<u8>> {
23 self.0.read()
69 self.remote.read()
2470 }
2571 fn stamp(&mut self) -> io::Result<Stamp> {
26 self.0.stamp()
72 self.remote.stamp()
2773 }
2874 fn publish(&mut self, transaction: &Transaction) -> Result<(), CommitError> {
29 self.0.publish(transaction)
75 self.remote.publish(transaction)
3076 }
3177 fn confirm(&mut self, stamp: &Stamp) -> Result<(), CommitError> {
32 self.0.confirm(stamp)
78 self.remote.confirm(stamp)
3379 }
3480 fn accepts_edits(&self) -> bool {
3581 true
......@@ -40,8 +86,11 @@ fn a_lost_batch_receipt_is_confirmed_without_repeating_its_edit() {
4086 edits: &[PendingEdit],
4187 revisions: &BTreeMap<ExGuid, ExGuid>,
4288 ) -> Result<(), CommitError> {
43 self.0.publish_edits(transaction, edits, revisions)?;
44 if std::mem::take(&mut self.1) {
89 if self.boundary == "unsent" {
90 stopped_at_boundary(self.root);
91 }
92 self.remote.publish_edits(transaction, edits, revisions)?;
93 if self.boundary == "lost" {
4594 return Err(CommitError {
4695 state: CommitState::Unknown,
4796 error: io::ErrorKind::BrokenPipe.into(),
......@@ -50,53 +99,362 @@ fn a_lost_batch_receipt_is_confirmed_without_repeating_its_edit() {
5099 Ok(())
51100 }
52101 }
53 let directory = tempfile::tempdir().unwrap();
54 let folder = notebook(directory.path());
55 let url = relay(Default::default());
56 let sharing = Sharing::new("").unwrap();
57 let host = host(&folder, &directory.path().join("host"), &sharing, &url);
58 let (guest, _) = guest("Alice", &code(&host), &url, &directory.path().join("alice"));
59 let file = folder.join("Garden.one");
60 let image = std::fs::read(&file).unwrap();
61 let (space, text, _) = server::text(&image);
62 let replica = Replica::open_or_create(directory.path().join("replica.sqlite"), None, || {
63 Ok(image.clone())
64 })
65 .unwrap();
66 let id = replica
67 .apply(
68 "Alice",
69 Edit {
70 at: 134_000_000_000_000_000,
71 ops: vec![Op::Page {
72 space,
73 op: PageOp::Text {
74 text,
75 range: 13..13,
76 with: "X".into(),
102
103 if let Ok(role) = std::env::var(ROLE) {
104 let root = PathBuf::from(std::env::var_os(ROOT).unwrap());
105 let url = std::env::var(URL).unwrap();
106 let boundary = std::env::var(BOUNDARY).unwrap();
107 if role == "host" {
108 let saved = root.join("sharing.json");
109 let sharing = if saved.exists() {
110 serde_json::from_slice(&std::fs::read(&saved).unwrap()).unwrap()
111 } else {
112 Sharing::new("").unwrap()
113 };
114 let host = Host::start(
115 Notebook::open(root.join("Garden"), root.join("host"))
116 .unwrap()
117 .into_storage(),
118 hello("Ada"),
119 sharing,
120 "Garden",
121 None,
122 Some(&url),
123 || {},
124 move |sharing| std::fs::write(&saved, serde_json::to_vec(sharing)?),
125 )
126 .unwrap();
127 std::fs::write(root.join("code"), code(&host)).unwrap();
128 std::fs::write(root.join("host-ready"), b"").unwrap();
129 loop {
130 std::thread::park();
131 }
132 }
133 let credentials = root.join("guest.json");
134 let (share, secret): ([u8; 16], [u8; 16]) = if role == "guest" {
135 let code = std::fs::read_to_string(root.join("code")).unwrap();
136 let welcome = share::join(hello("Alice"), &code, "", None, Some(&url)).unwrap();
137 let issued = (welcome.share, welcome.secret);
138 std::fs::write(&credentials, serde_json::to_vec(&issued).unwrap()).unwrap();
139 issued
140 } else {
141 serde_json::from_slice(&std::fs::read(credentials).unwrap()).unwrap()
142 };
143 let guest = Guest::start(hello("Alice"), share, secret, None, Some(&url), || {}).unwrap();
144 until("the restarted guest never met its host", || {
145 guest.host().is_some()
146 });
147 let database = root.join("replica.sqlite");
148 let replica = if role == "guest" {
149 let image = std::fs::read(root.join("Garden/Garden.one")).unwrap();
150 let (space, text, content) = body(&image);
151 let replica = Replica::create(&database, &image).unwrap();
152 let end = u32::try_from(content.encode_utf16().count()).unwrap();
153 let id = replica
154 .apply(
155 "Alice",
156 Edit {
157 at: 134_000_000_000_000_000,
158 ops: vec![Op::Page {
159 space,
160 op: PageOp::Text {
161 text,
162 range: end..end,
163 with: INSERTION.into(),
164 },
165 }],
77166 },
78 }],
79 },
80 )
81 .unwrap();
82 let mut remote = Lost(HostedRemote::new(&guest, "Garden.one"), true);
83 assert!(matches!(
84 replica.sync_once(&mut remote),
85 Err(notebook::Error::Remote(CommitError {
86 state: CommitState::Unknown,
87 ..
88 }))
89 ));
90 replica.sync_once(&mut remote).unwrap();
91 assert!(matches!(
92 replica.status(id).unwrap(),
93 Some(notebook::EditStatus::Published { .. })
94 ));
95 assert_eq!(
96 server::text(&std::fs::read(file).unwrap()).2,
97 "Original textX"
98 );
99 assert!(replica.pending().unwrap().is_empty());
167 )
168 .unwrap();
169 std::fs::write(root.join("edit-id"), id.to_string()).unwrap();
170 replica
171 } else {
172 Replica::open(&database).unwrap()
173 };
174 let id = std::fs::read_to_string(root.join("edit-id"))
175 .unwrap()
176 .parse()
177 .unwrap();
178 if role == "guest" {
179 if boundary == "before" {
180 stopped_at_boundary(&root);
181 }
182 let mut remote = Interrupted {
183 remote: HostedRemote::new(&guest, "Garden.one"),
184 boundary: &boundary,
185 root: &root,
186 };
187 let result = replica.sync_once(&mut remote);
188 if boundary == "lost" {
189 assert!(matches!(
190 result,
191 Err(notebook::Error::Remote(CommitError {
192 state: CommitState::Unknown,
193 ..
194 }))
195 ));
196 assert!(matches!(
197 replica.status(id).unwrap(),
198 Some(EditStatus::AwaitingConfirmation { .. })
199 ));
200 } else {
201 result.unwrap();
202 assert!(matches!(
203 replica.status(id).unwrap(),
204 Some(EditStatus::Published { .. })
205 ));
206 }
207 stopped_at_boundary(&root);
208 }
209 let mut remote = HostedRemote::new(&guest, "Garden.one");
210 if boundary == "unsent" {
211 let attempted = replica.status(id).unwrap();
212 assert!(matches!(
213 attempted,
214 Some(EditStatus::AwaitingConfirmation { .. })
215 ));
216 let pending = replica.pending().unwrap();
217 let snapshot = replica.snapshot().unwrap();
218 let original = std::fs::read(root.join("Garden/Garden.one")).unwrap();
219 for _ in 0..3 {
220 replica.sync_once(&mut remote).unwrap();
221 assert_eq!(replica.status(id).unwrap(), attempted);
222 assert_eq!(replica.pending().unwrap(), pending);
223 assert_eq!(replica.snapshot().unwrap(), snapshot);
224 assert_eq!(
225 std::fs::read(root.join("Garden/Garden.one")).unwrap(),
226 original
227 );
228 }
229 let archive = root.join("before-review.sqlite");
230 replica
231 .release(id, &archive, notebook::Resolution::Mine)
232 .unwrap();
233 let archive = notebook::Recovery::open(archive).unwrap();
234 assert_eq!(archive.status(id).unwrap(), attempted);
235 assert_eq!(archive.pending().unwrap(), pending);
236 assert_eq!(archive.snapshot().unwrap(), snapshot);
237 }
238 for _ in 0..5 {
239 replica.sync_once(&mut remote).unwrap();
240 if matches!(
241 replica.status(id).unwrap(),
242 Some(EditStatus::Published { .. })
243 ) {
244 break;
245 }
246 }
247 assert!(matches!(
248 replica.status(id).unwrap(),
249 Some(EditStatus::Published { .. })
250 ));
251 assert!(replica.pending().unwrap().is_empty());
252 let settled = std::fs::read(root.join("Garden/Garden.one")).unwrap();
253 for _ in 0..3 {
254 replica.sync_once(&mut remote).unwrap();
255 assert_eq!(
256 std::fs::read(root.join("Garden/Garden.one")).unwrap(),
257 settled
258 );
259 }
260 assert_eq!(replica.snapshot().unwrap(), settled);
261 drop(replica);
262 std::process::exit(0);
263 }
264
265 struct Process {
266 child: Child,
267 log: PathBuf,
268 }
269 impl Process {
270 fn start(root: &Path, url: &str, role: &str, boundary: &str) -> Self {
271 let log = root.join(format!("{role}.log"));
272 let output = std::fs::OpenOptions::new()
273 .create(true)
274 .append(true)
275 .open(&log)
276 .unwrap();
277 let child = Command::new(std::env::current_exe().unwrap())
278 .args(["--exact", TEST, "--nocapture"])
279 .env(ROLE, role)
280 .env(ROOT, root)
281 .env(URL, url)
282 .env(BOUNDARY, boundary)
283 .stdout(output.try_clone().unwrap())
284 .stderr(Stdio::from(output))
285 .spawn()
286 .unwrap();
287 Self { child, log }
288 }
289
290 fn ready(&mut self, marker: &Path) {
291 let deadline = Instant::now() + Duration::from_secs(30);
292 while !marker.exists() {
293 assert!(
294 self.child.try_wait().unwrap().is_none() && Instant::now() < deadline,
295 "{}: {}",
296 marker.display(),
297 std::fs::read_to_string(&self.log).unwrap()
298 );
299 std::thread::sleep(Duration::from_millis(20));
300 }
301 }
302
303 fn kill(&mut self) {
304 self.child.kill().unwrap();
305 assert!(!self.child.wait().unwrap().success());
306 }
307 }
308 impl Drop for Process {
309 fn drop(&mut self) {
310 let _ = self.child.kill();
311 let _ = self.child.wait();
312 }
313 }
314
315 for boundary in ["before", "unsent", "lost", "confirmed"] {
316 let directory = tempfile::tempdir().unwrap();
317 let root = directory.path();
318 let folder = notebook(root);
319 let file = folder.join("Garden.one");
320 let original =
321 include_bytes!("../../../corpus/conflict-page/native/conflict/notebook/synthetic.one");
322 std::fs::write(&file, original).unwrap();
323 let original_conflicts = server::conflicts(original);
324 assert!(!original_conflicts.is_empty());
325 let (space, _, content) = body(original);
326 let sealed = std::fs::read(folder.join("Sealed.one")).unwrap();
327 let url = relay(Default::default());
328 let mut host = Process::start(root, &url, "host", boundary);
329 host.ready(&root.join("host-ready"));
330 let mut guest = Process::start(root, &url, "guest", boundary);
331 guest.ready(&root.join("guest-ready"));
332 guest.kill();
333 host.kill();
334
335 let id: u64 = std::fs::read_to_string(root.join("edit-id"))
336 .unwrap()
337 .parse()
338 .unwrap();
339 let database = root.join("replica.sqlite");
340 let replica = Replica::open(&database).unwrap();
341 assert_eq!(
342 body(&replica.snapshot().unwrap()).2,
343 format!("{content}{INSERTION}")
344 );
345 if boundary == "confirmed" {
346 assert!(matches!(
347 replica.status(id).unwrap(),
348 Some(EditStatus::Published { .. })
349 ));
350 assert!(replica.pending().unwrap().is_empty());
351 } else if matches!(boundary, "lost" | "unsent") {
352 assert!(matches!(
353 replica.status(id).unwrap(),
354 Some(EditStatus::AwaitingConfirmation { .. })
355 ));
356 } else {
357 assert_eq!(replica.status(id).unwrap(), Some(EditStatus::Pending));
358 }
359 if boundary != "confirmed" {
360 assert_eq!(replica.pending().unwrap()[0].id, id);
361 assert_eq!(replica.pending().unwrap().len(), 1);
362 }
363 drop(replica);
364 let committed = std::fs::read(&file).unwrap();
365 if matches!(boundary, "before" | "unsent") {
366 assert_eq!(committed, original);
367 } else {
368 assert_eq!(body(&committed).2, format!("{content}{INSERTION}"));
369 }
370 let sharing: Sharing =
371 serde_json::from_slice(&std::fs::read(root.join("sharing.json")).unwrap()).unwrap();
372 let credentials = std::fs::read(root.join("guest.json")).unwrap();
373 std::fs::remove_file(root.join("host-ready")).unwrap();
374 let mut host = Process::start(root, &url, "host", boundary);
375 host.ready(&root.join("host-ready"));
376 let mut resumed = Process::start(root, &url, "resume", boundary);
377 let deadline = Instant::now() + Duration::from_secs(30);
378 let status = loop {
379 if let Some(status) = resumed.child.try_wait().unwrap() {
380 break status;
381 }
382 assert!(Instant::now() < deadline, "resumed guest timed out");
383 std::thread::sleep(Duration::from_millis(20));
384 };
385 if !status.success() {
386 let log = std::fs::read_to_string(&resumed.log).unwrap();
387 host.kill();
388 let kept = directory.keep();
389 panic!("{}: {log}", kept.display());
390 }
391 host.kill();
392 let result = std::fs::read(&file).unwrap();
393 assert_eq!(body(&result).2, format!("{content}{INSERTION}"));
394 if matches!(boundary, "lost" | "confirmed") {
395 assert_eq!(
396 &result[1024..],
397 &committed[1024..],
398 "settlement must not append another revision"
399 );
400 }
401 let revisions = |image: &[u8]| {
402 let store = onestore::Store::parse(image).unwrap();
403 onestore::RevisionIndex::parse(&store).unwrap().spaces[&space]
404 .revisions
405 .len()
406 };
407 assert_eq!(revisions(&result), revisions(original) + 1);
408 assert_eq!(server::conflicts(&result), original_conflicts);
409 assert_eq!(std::fs::read(folder.join("Sealed.one")).unwrap(), sealed);
410 let restarted: Sharing =
411 serde_json::from_slice(&std::fs::read(root.join("sharing.json")).unwrap()).unwrap();
412 assert_eq!(restarted.share, sharing.share);
413 assert_eq!(restarted.secret, sharing.secret);
414 assert_eq!(restarted.members, sharing.members);
415 assert_eq!(std::fs::read(root.join("guest.json")).unwrap(), credentials);
416 let replica = Replica::open(&database).unwrap();
417 assert!(matches!(
418 replica.status(id).unwrap(),
419 Some(EditStatus::Published { .. })
420 ));
421 assert!(replica.pending().unwrap().is_empty());
422 assert_eq!(replica.snapshot().unwrap(), result);
423 drop(replica);
424 if boundary == "unsent" {
425 let archive = notebook::Recovery::open(root.join("before-review.sqlite")).unwrap();
426 assert!(matches!(
427 archive.status(id).unwrap(),
428 Some(EditStatus::AwaitingConfirmation { .. })
429 ));
430 assert_eq!(archive.pending().unwrap()[0].id, id);
431 assert_eq!(archive.pending().unwrap().len(), 1);
432 assert_eq!(
433 body(&archive.snapshot().unwrap()).2,
434 format!("{content}{INSERTION}")
435 );
436 let database = rusqlite::Connection::open(root.join("before-review.sqlite")).unwrap();
437 assert_eq!(
438 database
439 .query_row("PRAGMA integrity_check", [], |row| row.get::<_, String>(0))
440 .unwrap(),
441 "ok"
442 );
443 println!(
444 "unsent: cold uncertainty retained; explicit Mine review archived the original queue before publication"
445 );
446 }
447 let database = rusqlite::Connection::open(&database).unwrap();
448 assert_eq!(
449 database
450 .query_row("PRAGMA integrity_check", [], |row| row.get::<_, String>(0))
451 .unwrap(),
452 "ok"
453 );
454 println!(
455 "{boundary}: original edit {id} published once, both processes restarted, SQLite integrity ok"
456 );
457 }
100458}
101459
102460#[test]