From 90a4606c65e94e9039d969a208d533bffddb9c80 Mon Sep 17 00:00:00 2001 From: Barzabel Date: Wed, 14 Oct 2026 22:52:43 +0300 Subject: [PATCH] M9: a compaction cut short closes the file instead of confirming lost edits (#183) --- apps/figcraft-server/src/store.rs | 47 +++++++++++++++++++------ apps/figcraft-server/src/store_tests.rs | 44 +++++++++++++++++++++++ 2 files changed, 81 insertions(+), 10 deletions(-) diff --git a/apps/figcraft-server/src/store.rs b/apps/figcraft-server/src/store.rs index e5e56b8f..f6cf50c2 100644 --- a/apps/figcraft-server/src/store.rs +++ b/apps/figcraft-server/src/store.rs @@ -265,6 +265,9 @@ pub struct Store { next_sweep: u64, /// Compactions so far (tests count them). compactions: u64, + /// Open files closed by a compaction cut short (their snapshot replaced, their log not): + /// reported at the next flush, so nothing appended after it is confirmed. + cut_short: Vec, /// The free space last seen on the data directory's disk, and when. free: Option<(Instant, Option)>, _lock: File, @@ -360,6 +363,7 @@ impl Store { gone_codes: VecDeque::new(), next_sweep: 0, compactions: 0, + cut_short: Vec::new(), free: None, _lock: lock, }; @@ -1124,7 +1128,7 @@ impl Store { Log { generation, file: f, pending: Vec::new(), entries: kept, bytes, dirty: kept > 0 } } // A new snapshot and log of the next generation, so appends start clean. - _ => self.rewrite(file, generation + 1, &server)?, + _ => self.rewrite(file, generation + 1, &server).map_err(|e| e.why)?, }; self.entry(file)?.log = Some(log); Ok(server) @@ -1137,21 +1141,24 @@ impl Store { } /// Write `server`'s state as snapshot `generation` and start an empty log of it. - fn rewrite(&mut self, file: &str, generation: u64, server: &Server) -> Result { + fn rewrite(&mut self, file: &str, generation: u64, server: &Server) -> Result { let dir = self.dir(file); - let state = server.state().map_err(|e| e.to_string())?; - let bytes = disk::encode_snapshot(generation, &state)?; + let kept = |why: String| RewriteError { why, snapshot_replaced: false }; + let state = server.state().map_err(|e| kept(e.to_string()))?; + let bytes = disk::encode_snapshot(generation, &state).map_err(kept)?; // Never replace a readable snapshot with one that can't be opened again: the old snapshot // and log still hold everything. if bytes.len() as u64 > self.snapshot_cap() { - return Err(format!("the snapshot would be {} MiB, more than can be opened again", bytes.len() >> 20)); + return Err(kept(format!("the snapshot would be {} MiB, more than can be opened again", bytes.len() >> 20))); } if bytes.len() as u64 > self.cfg.max_file_bytes { log::warn!("file {file} is {} MiB, over the {} MiB cap (kept anyway)", bytes.len() >> 20, self.cfg.max_file_bytes >> 20); } - disk::write_atomic(&dir.join(SNAPSHOT), &bytes).map_err(|e| format!("couldn't write the snapshot: {e}"))?; - disk::write_atomic(&dir.join(LOG), &disk::log_header(generation)).map_err(|e| format!("couldn't start the log: {e}"))?; - let f = OpenOptions::new().append(true).open(dir.join(LOG)).map_err(|e| format!("couldn't open the log: {e}"))?; + disk::write_atomic(&dir.join(SNAPSHOT), &bytes).map_err(|e| kept(format!("couldn't write the snapshot: {e}")))?; + // From here the old log no longer goes with the snapshot on disk. + let replaced = |why: String| RewriteError { why, snapshot_replaced: true }; + disk::write_atomic(&dir.join(LOG), &disk::log_header(generation)).map_err(|e| replaced(format!("couldn't start the log: {e}")))?; + let f = OpenOptions::new().append(true).open(dir.join(LOG)).map_err(|e| replaced(format!("couldn't open the log: {e}")))?; if let Ok(e) = self.entry(file) { e.bytes = (bytes.len() + LOG_HEADER) as u64; e.meta.seq = server.seq(); @@ -1204,8 +1211,19 @@ impl Store { } log::debug!("compacted file {file}"); } + // The new snapshot holds everything, but a restart would drop whatever is appended to + // the old log now (its generation no longer matches): the file closes, the hub hears + // it at the next flush, and it opens again from the new snapshot. + Err(RewriteError { why, snapshot_replaced: true }) => { + self.note(format!("file {file}: compaction was cut short ({why}); it was closed")); + log::error!("file {file}: compaction was cut short: {why}"); + if let Some(e) = self.files.get_mut(file) { + e.log = None; + } + self.cut_short.push(file.to_string()); + } // The old snapshot and log still hold everything: carry on appending. - Err(e) => log::error!("file {file}: compaction failed: {e}"), + Err(RewriteError { why, .. }) => log::error!("file {file}: compaction failed: {why}"), } } @@ -1213,7 +1231,7 @@ impl Store { /// Those are closed here (their pending records dropped: nobody was told of them) and open /// again from what is on disk. fn flush_all(&mut self) -> Vec { - let mut failed = Vec::new(); + let mut failed = std::mem::take(&mut self.cut_short); for (id, e) in &mut self.files { if let Some(log) = e.log.as_mut() { match flush_log(id, log) { @@ -1229,6 +1247,13 @@ impl Store { } } +/// Why a snapshot and log couldn't be rewritten, and whether the snapshot on disk was replaced +/// before it failed (then the old log no longer goes with it). +struct RewriteError { + why: String, + snapshot_replaced: bool, +} + /// Write and sync a log's pending records: the bytes written, or `Err` when that failed (the log /// can't be trusted to hold them). fn flush_log(id: &str, log: &mut Log) -> Result { @@ -1316,6 +1341,8 @@ impl Backend for Store { if dirty { self.compact(file, server); } + // It closes anyway, everything in its snapshot: a compaction cut short needs no report. + self.cut_short.retain(|f| f != file); if let Some(e) = self.files.get_mut(file) { e.log = None; } diff --git a/apps/figcraft-server/src/store_tests.rs b/apps/figcraft-server/src/store_tests.rs index fda51d4f..f380568c 100644 --- a/apps/figcraft-server/src/store_tests.rs +++ b/apps/figcraft-server/src/store_tests.rs @@ -269,3 +269,47 @@ fn random_bytes_never_panic_the_readers() { let _ = disk::read_log(&bytes); } } + +/// The first page's name in `server`'s document. +fn page_name(server: &Server) -> String { + let d = server.replica().document(); + d.node(d.first_page().unwrap()).unwrap().name.clone() +} + +#[test] +fn edits_confirmed_after_a_compaction_cut_short_survive_a_restart() { + let dir = TempDir::new("cutshort"); + let c = StoreConfig { compact_after: 3, ..cfg(dir.path()) }; + let folder = dir.path().join(FILES); + let mut store = Store::open(dir.path(), c.clone()).unwrap(); + let (id, _, _) = publish(&mut store, "Busy"); + let mut server = store.open(&id).unwrap(); + let mut a = join(&mut server, Access::Edit); + store.changed(&id, &mut server); + // The next compaction writes its snapshot, then can't start its log. + let blocker = disk::temp_of(&folder.join(&id).join(LOG)); + fs::create_dir(&blocker).unwrap(); + // What the hub would confirm: every change whose flush didn't report the file. + let mut confirmed = String::new(); + let mut failed = false; + let names = ["P0", "P1", "P2", "After 1", "After 2"]; + for name in names { + rename(&mut server, &mut a, name); + store.changed(&id, &mut server); + if store.flush().contains(&id) { + failed = true; + break; + } + confirmed = name.to_string(); + } + assert!(failed, "the hub hears the file can't be kept; confirmed {confirmed:?}"); + // A crash, then the disk recovers. + drop(store); + fs::remove_dir(&blocker).unwrap(); + let mut store = Store::open(dir.path(), c).unwrap(); + let restored = store.open(&id).unwrap(); + // Every confirmed change is there (the snapshot may also hold the one that was cut short). + let at = |n: &str| names.iter().position(|x| *x == n); + let name = page_name(&restored); + assert!(at(&name) >= at(&confirmed), "restored {name:?}, confirmed {confirmed:?}"); +}