From 166980b5cacc5eaea9581051898c52f845f078e1 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 13 Jul 2026 13:12:48 +0700 Subject: [PATCH] fix(persistence): adopt shared atomic_write_durable at remaining bare-write sites Task #49 (kernel M4 prep): eight durable-state writers still hand-rolled their own tmp-write/rename sequence instead of the shared K3 primitive (src/persistence/atomic.rs, PR #296). ACL SAVE (both acl::io::acl_save and the ACL SAVE command handler) had zero atomicity at all -- a bare fs::write + rename with no fsync whatsoever. CONFIG REWRITE, cluster nodes.conf (save_nodes_conf), and replication.state (save_replication_state) all did write+rename but skipped both the temp-file fsync and the parent-directory fsync, so a kill-9 immediately after rename() returns can still revert the directory entry to stale or 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 heap data file visible to a concurrent reader. The native RDB save paths used by BGSAVE (persistence::rdb::save / save_from_snapshot, persistence::redis_rdb::save) also lacked the parent-directory fsync step. All eight sites now route through atomic_write_durable (temp file -> sync_all -> rename -> parent-dir fsync), matching the pattern already proven at the graph CSR / text sidecar / vector Stack-B call sites from PR #296. Each conversion is paired with a "no leftover temp file after a successful write" regression test (8 new tests total). Out of scope per task brief: AOF appends, WAL segments (own framing/fsync mechanisms), and the legacy #[allow(dead_code)] rewrite_aof_sync RDB-preamble path (dead code, not a live BGSAVE path). Verified: cargo test --lib (default features + runtime-tokio,jemalloc), cargo fmt --check, cargo clippy -D warnings (both feature sets). author: Tin Dang --- CHANGELOG.md | 24 +++++++++++++ src/acl/io.rs | 41 +++++++++++++++++++--- src/cluster/migration.rs | 33 ++++++++++++++---- src/command/acl.rs | 21 ++++++------ src/command/config.rs | 14 ++++---- src/persistence/clog.rs | 33 ++++++++++++++++-- src/persistence/kv_page.rs | 66 +++++++++++++++++++++++++++++------- src/persistence/rdb.rs | 40 ++++++++++++++-------- src/persistence/redis_rdb.rs | 9 +++-- src/replication/state.rs | 30 ++++++++++++++-- 10 files changed, 247 insertions(+), 64 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index aebc824e..3d5de35d 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/src/acl/io.rs b/src/acl/io.rs index 1df9153c..1b2b3bd9 100644 --- a/src/acl/io.rs +++ b/src/acl/io.rs @@ -72,16 +72,25 @@ pub fn parse_acl_line(line: &str) -> Option { 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")]); + } } diff --git a/src/cluster/migration.rs b/src/cluster/migration.rs index 7332f1cd..2c78257b 100644 --- a/src/cluster/migration.rs +++ b/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 = 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() { diff --git a/src/command/acl.rs b/src/command/acl.rs index bd6ad8c1..6dd2a1c3 100644 --- a/src/command/acl.rs +++ b/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))), } } diff --git a/src/command/config.rs b/src/command/config.rs index 3acfbcf8..9173958b 100644 --- a/src/command/config.rs +++ b/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")) } diff --git a/src/persistence/clog.rs b/src/persistence/clog.rs index f89949be..e1232841 100644 --- a/src/persistence/clog.rs +++ b/src/persistence/clog.rs @@ -155,11 +155,17 @@ pub fn scan_clog_dir(clog_dir: &std::path::Path) -> std::io::Result 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")]); + } } diff --git a/src/persistence/kv_page.rs b/src/persistence/kv_page.rs index 4e41bcb3..fa0a949c 100644 --- a/src/persistence/kv_page.rs +++ b/src/persistence/kv_page.rs @@ -569,20 +569,23 @@ pub fn read_overflow_chain(file_data: &[u8], start_page_idx: usize) -> Option 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(()) } @@ -590,15 +593,16 @@ pub fn write_datafile_mixed( /// 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); diff --git a/src/persistence/rdb.rs b/src/persistence/rdb.rs index 128ea554..1cd13bc3 100644 --- a/src/persistence/rdb.rs +++ b/src/persistence/rdb.rs @@ -102,15 +102,12 @@ pub fn save_to_bytes(databases: &[Database]) -> Result, MoonError> { pub fn save(databases: &[Database], path: &Path) -> Result<(), MoonError> { let buf = save_to_bytes(databases)?; - // Atomic write: write to tmp, then rename - let tmp_path = path.with_extension("rdb.tmp"); - std::fs::write(&tmp_path, &buf).map_err(|e| RdbError::Io { - path: tmp_path.clone(), - source: e, - })?; - std::fs::rename(&tmp_path, path).map_err(|e| RdbError::Io { + // Atomic write: temp + fsync + rename + dir-fsync via the shared K3 + // primitive (task #49) -- kill-9 between write and rename can never + // surface a torn or empty RDB at `path`. + crate::persistence::atomic::atomic_write_durable(path, &buf).map_err(|e| RdbError::Io { path: path.to_path_buf(), - source: e, + source: e.into(), })?; Ok(()) @@ -157,14 +154,9 @@ pub fn save_from_snapshot( let checksum = hasher.finalize(); buf.write_all(&checksum.to_le_bytes())?; - let tmp_path = path.with_extension("rdb.tmp"); - std::fs::write(&tmp_path, &buf).map_err(|e| RdbError::Io { - path: tmp_path.clone(), - source: e, - })?; - std::fs::rename(&tmp_path, path).map_err(|e| RdbError::Io { + crate::persistence::atomic::atomic_write_durable(path, &buf).map_err(|e| RdbError::Io { path: path.to_path_buf(), - source: e, + source: e.into(), })?; Ok(()) @@ -1812,6 +1804,24 @@ mod tests { assert!(loaded[2].get(b"k2").is_some()); } + /// Task #49: `save` (and `save_from_snapshot`) must go through + /// `atomic_write_durable` instead of a bare tmp-write + rename. + /// Regression pin: no leftover `.rdb.tmp` after a successful save. + #[test] + fn test_save_leaves_no_leftover_temp_file() { + let (dir, path) = rdb_path(); + let mut dbs = vec![Database::new()]; + dbs[0].set_string(Bytes::from_static(b"k"), Bytes::from_static(b"v")); + + save(&dbs, &path).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("dump.rdb")]); + } + #[test] fn test_missing_file_returns_error() { let dir = tempdir().unwrap(); diff --git a/src/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs index 2673c6e2..0f9a8be8 100644 --- a/src/persistence/redis_rdb.rs +++ b/src/persistence/redis_rdb.rs @@ -602,9 +602,12 @@ pub fn save(databases: &[Database], path: &Path) -> anyhow::Result<()> { let mut buf = Vec::new(); write_rdb(databases, &mut buf); - let tmp_path = path.with_extension("rdb.tmp"); - std::fs::write(&tmp_path, &buf).context("Failed to write temporary Redis RDB file")?; - std::fs::rename(&tmp_path, path).context("Failed to rename temporary Redis RDB file")?; + // Temp + fsync + rename + dir-fsync via the shared K3 primitive (task + // #49) -- a bare tmp-write+rename (the prior code) can leave an EMPTY + // file at `path` on ext4/xfs if the crash lands between rename and the + // parent-directory fsync. + crate::persistence::atomic::atomic_write_durable(path, &buf) + .context("Failed to atomically write Redis RDB file")?; Ok(()) } diff --git a/src/replication/state.rs b/src/replication/state.rs index a2b447a5..280c9a86 100644 --- a/src/replication/state.rs +++ b/src/replication/state.rs @@ -331,10 +331,17 @@ pub fn save_replication_state( repl_id: &str, repl_id2: &str, ) -> std::io::Result<()> { - let tmp = dir.join("replication.state.tmp"); let dst = dir.join("replication.state"); - std::fs::write(&tmp, format!("{}\n{}\n", repl_id, repl_id2))?; - std::fs::rename(&tmp, &dst)?; + // Atomic via `atomic_write_durable` (task #49): temp + fsync + rename + // + dir-fsync. The prior code (comment claimed "same pattern as + // WalWriter::truncate_after_snapshot") only did write+rename with no + // fsync -- a kill-9 right after `rename()` returned could still revert + // replication.state on ext4/xfs, losing the master replication ID + // across a restart. + crate::persistence::atomic::atomic_write_durable( + &dst, + format!("{}\n{}\n", repl_id, repl_id2).as_bytes(), + )?; Ok(()) } @@ -510,6 +517,23 @@ mod tests { assert_eq!(loaded2, id2); } + /// Task #49: `save_replication_state` must go through + /// `atomic_write_durable`, not a bare write+rename. Regression pin: no + /// leftover `replication.state.tmp` after a successful save. + #[test] + fn test_save_replication_state_leaves_no_leftover_temp_file() { + let dir = tempfile::tempdir().unwrap(); + let id1 = generate_repl_id(); + let id2 = generate_repl_id(); + save_replication_state(dir.path(), &id1, &id2).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("replication.state")]); + } + #[test] fn test_load_generates_fresh_when_missing() { let dir = tempfile::tempdir().unwrap();