Merge #340: M9: a compaction cut short closes the file instead of confirming lost edits (#183) (@Barzabel)

apps/figcraft-server/src/store.rs

@@ -266,6 +266,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<String>,
     /// The free space last seen on the data directory's disk, and when.
     free: Option<(Instant, Option<u64>)>,
     _lock: File,
@@ -361,6 +364,7 @@ impl Store {
             gone_codes: VecDeque::new(),
             next_sweep: 0,
             compactions: 0,
+            cut_short: Vec::new(),
             free: None,
             _lock: lock,
         };
@@ -1136,7 +1140,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)
@@ -1149,21 +1153,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<Log, String> {
+    fn rewrite(&mut self, file: &str, generation: u64, server: &Server) -> Result<Log, RewriteError> {
         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();
@@ -1216,8 +1223,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}"),
         }
     }
 
@@ -1225,7 +1243,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<String> {
-        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) {
@@ -1241,6 +1259,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<u64, ()> {
@@ -1340,6 +1365,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;
         }

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:?}");
+}