diff --git a/crates/mobile/src/library.rs b/crates/mobile/src/library.rs index 40a615c8fdad0f71dd154efb6fdb8fae606ca456..87f71f2bd4a3bcb4addae182105e91c051f91fac 100644 --- a/crates/mobile/src/library.rs +++ b/crates/mobile/src/library.rs @@ -1409,13 +1409,11 @@ impl Library { { ops.push(Op::Section(SectionOp::Delete(vec![discard]))); } - Ok(section.apply( - author, - Edit { - at: filetime(), - ops, - }, - )?) + let edit = Edit { + at: filetime(), + ops, + }; + Ok(section.replica().apply(author, edit)?) })?; let added = self.with_section(from, |section| { let mut ops = vec![Op::Section(SectionOp::Delete(vec![space]))]; @@ -1428,13 +1426,13 @@ impl Library { } _ => None, }; - section.apply( - author, - Edit { - at: filetime(), - ops, - }, - )?; + let edit = Edit { + at: filetime(), + ops, + }; + section + .replica() + .apply_unchanged(author, edit, space, page)?; Ok(added) })?; let id = |space: Option| space.map(|space| space.to_string()); diff --git a/crates/mobile/src/tests.rs b/crates/mobile/src/tests.rs index a2a6e4dc6530747c85da03fcd4099d225d16f3d0..4d588e0206e2d299259613fdf120c1c1402c5b35 100644 --- a/crates/mobile/src/tests.rs +++ b/crates/mobile/src/tests.rs @@ -1043,13 +1043,34 @@ fn a_share_notebook_syncs_only_its_readable_sections() { /// Paths the test coordinator was asked for, with whether to write. static COORDINATED: Mutex> = Mutex::new(Vec::new()); +/// A file whose next coordinated access the test coordinator holds, as another app may: who +/// hears that it is held, and what lets it go. +type Held = ( + String, + std::sync::mpsc::Sender<()>, + std::sync::mpsc::Receiver<()>, +); +static HELD: Mutex> = Mutex::new(None); + extern "C" fn coordinator( path: *const c_char, write: bool, body: extern "C" fn(*mut c_void), context: *mut c_void, ) { - COORDINATED.lock().unwrap().push((string(path), write)); + let path = string(path); + COORDINATED.lock().unwrap().push((path.clone(), write)); + let held = { + let mut held = HELD.lock().unwrap(); + match &*held { + Some((file, ..)) if *file == path => held.take(), + _ => None, + } + }; + if let Some((_, entered, released)) = held { + let _ = entered.send(()); + let _ = released.recv(); + } body(context); } @@ -1758,6 +1779,111 @@ fn a_page_moves_to_another_section_and_back_where_it_was() { assert_eq!(stored.page(back).unwrap().title, title); } +/// A page edited while it moves stays, with the edit, beside the copy that arrived; a page added +/// meanwhile lets it go. The files hold both once the notebook closes. +#[test] +fn a_page_edited_while_it_moves_stays_with_its_edit() { + library::sb_set_coordinator(coordinator); + for edited in [true, false] { + let (_directory, root, library, features) = notebook_open("Features.one"); + assert!(eventually(|| { + let status = library.sync_status(); + !status.is_empty() + && status + .iter() + .all(|sync| sync.state == SyncState::UpToDate as u8) + })); + let space = space(&features, "Task controls"); + // Empty.one is held for the move, which opens it, until the edit below is made. + let (entered, holding) = std::sync::mpsc::channel(); + let (release, released) = std::sync::mpsc::channel(); + let empty = root.canonicalize().unwrap().join("Empty.one"); + *HELD.lock().unwrap() = Some((empty.to_string_lossy().into_owned(), entered, released)); + let moving = Arc::clone(&library); + let date = ["Thursday, October 1, 2026", "9:00 AM"]; + let move_page = std::thread::spawn(move || { + moving + .move_page( + "Features.one", + space, + "Empty.one", + None, + 1, + None, + "Clover Test", + date, + ) + .map(|_| ()) + .map_err(|error| error.to_string()) + }); + holding.recv_timeout(Duration::from_secs(10)).unwrap(); + if edited { + let mut canvas = canvas(&features, "Task controls"); + focus(&mut canvas, "Disabled standalone task"); + canvas.insert("Edited ".into()).unwrap(); + features + .shared + .apply(canvas.edit().unwrap().unwrap()) + .unwrap(); + } else { + features.new_page(None, date[0], date[1]).unwrap(); + } + features.shared.section.written().unwrap(); + drop(release); + let moved = move_page.join().unwrap(); + assert_eq!(moved.is_ok(), !edited, "{moved:?}"); + drop(features); + library.close(Duration::from_secs(20)).unwrap(); + let arena = onestore::Arena::default(); + let stored = onestore::read_file(root.join("Features.one")).unwrap(); + let stayed = onestore::Section::open(&arena, stored).unwrap().page(space); + assert_eq!(stayed.is_ok(), edited); + if edited { + assert!(canvas::search::page_text(&stayed.unwrap()).contains("Edited ")); + } + let arena = onestore::Arena::default(); + let stored = onestore::read_file(root.join("Empty.one")).unwrap(); + let mut empty = onestore::Section::open(&arena, stored).unwrap(); + let copies: Vec = (empty.pages().unwrap().into_iter()) + .filter(|(_, title, _)| title == "Task controls") + .map(|(copy, ..)| canvas::search::page_text(&empty.page(copy).unwrap())) + .collect(); + assert_eq!(copies.len(), 1); + assert!(!copies[0].contains("Edited ")); + } +} + +/// A page the section it moves to refuses stays where it was, and the move fails. +#[test] +fn a_page_the_other_section_refuses_stays() { + let (_directory, _root, library, features) = notebook_open("Features.one"); + let before = levels(&features); + let (space, _) = before[1]; + let page = features.shared.section.page(space).unwrap(); + let empty = open(&library, "Empty.one"); + let listed = levels(&empty); + // A section's first page has level 1: Empty.one, holding none, refuses this one. + assert!( + library + .move_page( + "Features.one", + space, + "Empty.one", + None, + 2, + None, + "Clover Test", + ["Thursday, October 1, 2026", "9:00 AM"], + ) + .is_err() + ); + empty.shared.section.written().unwrap(); + features.shared.section.written().unwrap(); + assert_eq!(levels(&empty), listed); + assert_eq!(levels(&features), before); + assert_eq!(features.shared.section.page(space).unwrap(), page); +} + #[test] fn a_sections_last_page_moving_leaves_a_page_that_undo_takes_out() { let (_directory, _root, library, features) = notebook_open("Features.one"); diff --git a/crates/notebook/README.md b/crates/notebook/README.md index 2ce910c1fa02384de9a5546e4a183dfe6c75eabf..e331f76c7bc6ac9a390946986965b88a2325261c 100644 --- a/crates/notebook/README.md +++ b/crates/notebook/README.md @@ -38,6 +38,11 @@ assert_eq!(reopened.pending()?.last().map(|edit| edit.id), Some(id)); # } ``` +`apply_unchanged(author, edit, space, page)` applies the edit only while page `space` +still holds `page`, checked on the section thread in order with every other edit, and +refuses it as `Rejected` otherwise: a move across sections copies a page there, then retires +it here this way, so an edit made since the copy keeps the page. + Page reads never wait for the sync thread: a rebase or reread of the queue runs on a new thread while the current one keeps answering `page` and `pages` from the section as it was, then hands its requests over; edits sent meanwhile apply to the rebuilt section. diff --git a/crates/notebook/src/lib.rs b/crates/notebook/src/lib.rs index 3b3c96854967a9d6053de89da4ce4897ca175d8e..693a16903a35610ddeaa711221f7ab0f9dfaec89 100644 --- a/crates/notebook/src/lib.rs +++ b/crates/notebook/src/lib.rs @@ -308,6 +308,39 @@ impl Replica { edit, reply, waiting: None, + unchanged: None, + }) + }) + .await + } + + /// `apply`, provided page `space` still holds `page`; refused as `Rejected` otherwise. + pub fn apply_unchanged( + &self, + author: &str, + edit: Edit, + space: ExGuid, + page: Page, + ) -> Result { + task::blocking(self.apply_unchanged_async(author, edit, space, page))? + } + + /// `apply_async`, provided page `space` still holds `page`, checked where edits apply in + /// order: a move retires the page it copied this way, keeping any edit made since. + pub async fn apply_unchanged_async( + &self, + author: &str, + edit: Edit, + space: ExGuid, + page: Page, + ) -> Result { + self.ask_async(|reply| { + working::Request::Apply(working::Accepted { + author: author.to_owned(), + edit, + reply, + waiting: None, + unchanged: Some((space, page)), }) }) .await @@ -327,6 +360,7 @@ impl Replica { edit, reply, waiting: Some(waiting), + unchanged: None, })) } diff --git a/crates/notebook/src/session.rs b/crates/notebook/src/session.rs index 3e14e308092b9b8750e95d84a51221370fcff793..b132e46cbf3c5ea8ac89a8b28510549daa0f3662 100644 --- a/crates/notebook/src/session.rs +++ b/crates/notebook/src/session.rs @@ -2565,7 +2565,8 @@ impl Section { } /// Removes pages or conflict pages permanently, queued like the user's own edits (a - /// move across sections is `import_page` there, then this here). + /// move across sections imports a page there, then retires it here with + /// `Replica::apply_unchanged`). pub fn delete_pages(&self, pages: &[ExGuid]) -> Result { crate::task::blocking(self.delete_pages_async(pages))? } diff --git a/crates/notebook/src/working.rs b/crates/notebook/src/working.rs index b9a0a33d50eabd6b1859f0cec327585ba7931b9b..cf8ef019fab2910a41e8e13616798b403fb48300 100644 --- a/crates/notebook/src/working.rs +++ b/crates/notebook/src/working.rs @@ -481,6 +481,8 @@ pub(crate) struct Accepted { pub reply: Reply, /// Set where nobody blocks on the reply: a failed write keeps the edit and reports here. pub waiting: Option, + /// A page as a move copied it, which the edit applies only while it still holds. + pub unchanged: Option<(ExGuid, Page)>, } struct Working<'a> { @@ -700,6 +702,15 @@ impl<'a> Working<'a> { /// Applies an edit, keeping it to be written with the burst. A refused edit is answered /// here; true when it failed part way and the section must be reread. fn accept(&mut self, accepted: Accepted) -> bool { + if let Some((space, page)) = &accepted.unchanged + && !self.section.page(*space).is_ok_and(|now| now == *page) + { + let changed = OpError::StructureChanged( + "The page changed while it moved. Both sections now have a copy.", + ); + (accepted.reply)(Err(changed.into())); + return false; + } match self.section.apply(&accepted.author, &accepted.edit) { Ok(()) => { self.touched diff --git a/crates/notebook/tests/sync_pages.rs b/crates/notebook/tests/sync_pages.rs index 9415c4ff637b29a58c34a04830a5dd5fcdb20034..2a3cfe0c4f954fbedbc933a15264b5f3bc09455d 100644 --- a/crates/notebook/tests/sync_pages.rs +++ b/crates/notebook/tests/sync_pages.rs @@ -743,3 +743,97 @@ fn page_moves_and_a_removed_page_merge_as_onenote_merges_them() { ); } } + +/// A page a move copied leaves only as it was copied: an edit made here since, published or +/// not, or another writer's that a rebase took, keeps it; edits to other pages, published or +/// rebased, do not. Either way the queue reopens as it was left and publishes it. +#[test] +fn a_moved_page_leaves_only_as_it_was_copied() { + // The first two pages holding text. + let texts = texts(SOURCE); + let mut written = order(SOURCE).into_iter().filter_map(|(space, _)| { + let (&(_, text), _) = texts.iter().find(|((sid, _), _)| *sid == space)?; + Some((space, text)) + }); + let (moving, moving_text) = written.next().unwrap(); + let (other, other_text) = written.next().unwrap(); + fn here(cache: &Replica, (space, text): (ExGuid, ExGuid)) { + let op = onestore::op::PageOp::Text { + text, + range: 0..0, + with: "Here ".into(), + }; + let at = 133_000_000_000_000_001; + let ops = vec![Op::Page { space, op }]; + cache + .apply("Typist", onestore::op::Edit { at, ops }) + .unwrap(); + } + fn there(server: &mut Server, (space, text): (ExGuid, ExGuid)) { + server.visible = typed(&server.visible, space, text, 0..0, "There "); + server.durable.clone_from(&server.visible); + } + type Between = fn(&Replica, &mut Server, [(ExGuid, ExGuid); 2]); + let schedules: [(&str, bool, Between); 5] = [ + ("unrelated", true, |cache, server, [_, other]| { + here(cache, other); + cache.sync_once(server).unwrap(); + there(server, other); + cache.sync_once(server).unwrap(); + }), + ("appended", false, |cache, _, [moving, _]| { + here(cache, moving) + }), + ("published", false, |cache, server, [moving, _]| { + here(cache, moving); + cache.sync_once(server).unwrap(); + }), + ("rebased", false, |cache, server, [moving, _]| { + there(server, moving); + cache.sync_once(server).unwrap(); + }), + ( + "rebased and appended", + false, + |cache, server, [moving, other]| { + there(server, other); + here(cache, moving); + cache.sync_once(server).unwrap(); + }, + ), + ]; + for (schedule, leaves, between) in schedules { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("moving.sqlite"); + let cache = Replica::create(&path, SOURCE).unwrap(); + let mut server = Server::new(SOURCE); + let copied = cache.page(moving).unwrap(); + between( + &cache, + &mut server, + [(moving, moving_text), (other, other_text)], + ); + let kept = cache.page(moving).unwrap(); + let delete = onestore::op::Edit { + at: 133_000_000_000_000_002, + ops: vec![Op::Section(SectionOp::Delete(vec![moving]))], + }; + let retired = cache.apply_unchanged("Mover", delete, moving, copied); + assert_eq!(retired.is_ok(), leaves, "{schedule}: {retired:?}"); + assert!( + leaves || matches!(retired, Err(notebook::Error::Rejected(_))), + "{schedule}" + ); + drop(cache); + let cache = Replica::open(&path).unwrap(); + let listed = |image: &[u8]| order(image).iter().any(|(space, _)| *space == moving); + assert_eq!( + cache.page(moving).ok(), + (!leaves).then_some(kept), + "{schedule}" + ); + cache.sync_once(&mut server).unwrap(); + assert_eq!(listed(&server.durable), !leaves, "{schedule}"); + assert!(cache.pending().unwrap().is_empty(), "{schedule}"); + } +} diff --git a/crates/snowbound/src/undo.rs b/crates/snowbound/src/undo.rs index b98f10b2ce8e0fdae59cdb2926d31fea5a78e2ac..3dea7aa09b7f082e6227ebdc4d9a473e869cad19 100644 --- a/crates/snowbound/src/undo.rs +++ b/crates/snowbound/src/undo.rs @@ -567,7 +567,8 @@ impl Site { let (Some(from), Some(path)) = (at(page), self.section_path(target)) else { return Ok(None); }; - let import = notebook::session::moved(&self.replica.page(page)?, &self.author)?; + let sent = self.replica.page(page)?; + let import = notebook::session::moved(&sent, &self.author)?; let arrived = imported(&import); let before = listed[from + 1..].first().map(|(space, ..)| *space); let level = listed[from].2; @@ -585,7 +586,13 @@ impl Site { ) .await?; section.close_async().await?; - self.apply(ops).await?; + let edit = Edit { + at: crate::filetime(), + ops, + }; + self.replica + .apply_unchanged_async(&self.author, edit, page, sent) + .await?; Applied { undo: Change::Return { page: arrived, @@ -629,15 +636,13 @@ impl Site { ]; let gone = self.take_added(added, &listed, &mut ops); self.apply(ops).await?; + let delete = Edit { + at: crate::filetime(), + ops: vec![Op::Section(SectionOp::Delete(vec![page]))], + }; section .replica() - .apply_async( - &self.author, - Edit { - at: crate::filetime(), - ops: vec![Op::Section(SectionOp::Delete(vec![page]))], - }, - ) + .apply_unchanged_async(&self.author, delete, page, sent) .await?; section.close_async().await?; Applied { @@ -1682,6 +1687,85 @@ mod tests { assert_eq!(fixture.titles(), titled(&["First", "Third"])); } + /// A page edited while it is on its way to another section stays, with the edit, beside + /// the copy that arrived there; an edit to another page meanwhile lets it go. Both hold + /// once each section is opened again. + #[test] + fn a_page_edited_while_it_is_sent_stays_with_its_edit() { + struct Unreached; + impl notebook::Remote for Unreached { + fn read(&mut self) -> std::io::Result> { + unreachable!() + } + fn stamp(&mut self) -> std::io::Result { + unreachable!() + } + fn publish(&mut self, _: &onestore::Transaction) -> Result<(), onestore::CommitError> { + unreachable!() + } + fn confirm(&mut self, _: &onestore::Stamp) -> Result<(), onestore::CommitError> { + unreachable!() + } + } + + for edited in ["Second", "Third"] { + let mut fixture = + Fixture::new(&format!("send-{edited}"), &["First", "Second", "Third"]); + let target = fixture.library.section_identity("B.one").unwrap(); + let page = fixture.space("Second"); + // B's sync holds in connecting until released, so the move, closing B, waits + // there once the copy has landed. + let mut b = fixture.library.open("B.one", || {}).unwrap(); + let arrived = Arc::clone(b.replica()); + let (entered, holding) = std::sync::mpsc::channel(); + let (release, released) = std::sync::mpsc::channel::<()>(); + b.relocate_with(fixture.folder.join("Notebook/B.one"), move |_| { + let _ = entered.send(()); + let _ = released.recv(); + Err::(std::io::ErrorKind::TimedOut.into()) + }) + .unwrap(); + holding.recv().unwrap(); + fixture.library.keep("B.one", b); + let mut site = fixture.site(); + let shown = fixture.pages()[0].0; + let sending = std::thread::spawn(move || { + pollster::block_on(site.change(Change::Send { page, target }, shown)) + .map(|applied| applied.is_some()) + .map_err(|error| error.to_string()) + }); + while !(arrived.pages().unwrap().iter()).any(|(_, title, _)| title == "Second") { + std::thread::sleep(std::time::Duration::from_millis(5)); + } + fixture.retitle(fixture.space(edited), &format!("{edited} edited")); + drop(release); + drop(arrived); + let sent = sending.join().unwrap(); + + let stayed = match edited { + "Second" => { + assert!(sent.is_err(), "{sent:?}"); + titled(&["First", "Second edited", "Third"]) + } + _ => { + assert_eq!(sent, Ok(true)); + titled(&["First", "Third edited"]) + } + }; + fixture.section.take().unwrap().close().unwrap(); + let titles = |path: &str| { + let section = fixture.library.open(path, || {}).unwrap(); + let titles: Vec<(String, u32)> = (section.pages().unwrap().into_iter()) + .map(|(_, title, level)| (title, level)) + .collect(); + section.close().unwrap(); + titles + }; + assert_eq!(titles("New Section 1.one"), stayed); + assert_eq!(titles("B.one"), titled(&["", "Second"])); + } + } + /// A title given in the page list goes back while the page still has it. #[test] fn a_title_goes_back_while_the_page_has_it() {